point-to-point: Add preliminary support for flow control

The (unique) transmission queue is stopped when enqueuing a packet fails, so
that upper layers do not send other packets down. When a packet transmission
is completed, if the queue is empty then wake the upper layers. If the queue
was stopped, there is now room for another packet and hence wake the upper
layers as well.

src/point-to-point/model/point-to-point-net-device.cc

@@ -206,6 +206,16 @@ PointToPointNetDevice::ProcessHeader (Ptr<Packet> p, uint16_t& param)
   return true;
 }
 
+void
+PointToPointNetDevice::DoInitialize (void)
+{
+  NS_LOG_FUNCTION (this);
+  // A NetDeviceQueueInterface object must have been aggregated to this device
+  // by the traffic control layer
+  m_queueInterface = GetObject<NetDeviceQueueInterface> ();
+  NetDevice::DoInitialize ();
+}
+
 void
 PointToPointNetDevice::DoDispose ()
 {
@@ -280,18 +290,27 @@ PointToPointNetDevice::TransmitComplete (void)
   m_phyTxEndTrace (m_currentPkt);
   m_currentPkt = 0;
 
+  NS_ASSERT (m_queueInterface);
+  Ptr<NetDeviceQueue> txq = m_queueInterface->GetTxQueue (0);
+
   Ptr<QueueItem> item = m_queue->Dequeue ();
   if (item == 0)
     {
-      //
-      // No packet was on the queue, so we just exit.
-      //
+      NS_LOG_LOGIC ("No pending packets in device queue after tx complete");
+      txq->Wake ();
       return;
     }
 
   //
-  // Got another packet off of the queue, so start the transmit process agin.
+  // Got another packet off of the queue, so start the transmit process again.
+  // If the queue was stopped, start it again. Note that we cannot wake the upper
+  // layers because otherwise a packet is sent to the device while the machine
+  // state is busy, thus causing the assert in TransmitStart to fail.
   //
+  if (txq->IsStopped ())
+    {
+      txq->Start ();
+    }
   Ptr<Packet> p = item->GetPacket ();
   m_snifferTrace (p);
   m_promiscSnifferTrace (p);
@@ -511,6 +530,11 @@ PointToPointNetDevice::Send (
   const Address &dest, 
   uint16_t protocolNumber)
 {
+  NS_ASSERT (m_queueInterface);
+  Ptr<NetDeviceQueue> txq = m_queueInterface->GetTxQueue (0);
+
+  NS_ASSERT_MSG (!txq->IsStopped (), "Send should not be called when the device is stopped");
+
   NS_LOG_FUNCTION (this << packet << dest << protocolNumber);
   NS_LOG_LOGIC ("p=" << packet << ", dest=" << &dest);
   NS_LOG_LOGIC ("UID is " << packet->GetUid ());
@@ -551,8 +575,10 @@ PointToPointNetDevice::Send (
       return true;
     }
 
-  // Enqueue may fail (overflow)
+  // Enqueue may fail (overflow). Stop the tx queue, so that the upper layers
+  // do not send packets until there is room in the queue again.
   m_macTxDropTrace (packet);
+  txq->Stop ();
   return false;
 }
 

src/point-to-point/model/point-to-point-net-device.h

@@ -198,6 +198,8 @@ protected:
    */
   void DoMpiReceive (Ptr<Packet> p);
 
+  virtual void DoInitialize (void);
+
 private:
 
   /**
@@ -435,6 +437,7 @@ private:
   TracedCallback<Ptr<const Packet> > m_promiscSnifferTrace;
 
   Ptr<Node> m_node;         //!< Node owning this NetDevice
+  Ptr<NetDeviceQueueInterface> m_queueInterface;   //!< NetDevice queue interface
   Mac48Address m_address;   //!< Mac48Address of this NetDevice
   NetDevice::ReceiveCallback m_rxCallback;   //!< Receive callback
   NetDevice::PromiscReceiveCallback m_promiscCallback;  //!< Receive callback

src/point-to-point/test/point-to-point-test.cc

@@ -86,6 +86,11 @@ PointToPointTest::DoRun (void)
   a->AddDevice (devA);
   b->AddDevice (devB);
 
+  Ptr<NetDeviceQueueInterface> ifaceA = CreateObject<NetDeviceQueueInterface> ();
+  devA->AggregateObject (ifaceA);
+  Ptr<NetDeviceQueueInterface> ifaceB = CreateObject<NetDeviceQueueInterface> ();
+  devB->AggregateObject (ifaceB);
+
   Simulator::Schedule (Seconds (1.0), &PointToPointTest::SendOnePacket, this, devA);
 
   Simulator::Run ();