Checkpoint. Adding data structures to support weighted round robin
diff --git a/model/fw/per-fib-limits.cc b/model/fw/per-fib-limits.cc
index 43355e3..4536783 100644
--- a/model/fw/per-fib-limits.cc
+++ b/model/fw/per-fib-limits.cc
@@ -90,6 +90,13 @@
NS_LOG_FUNCTION (this << pitEntry->GetPrefix ());
// totally override all (if any) parent processing
+ if (pitEntry->GetFwTag<PitQueueTag> () != boost::shared_ptr<PitQueueTag> ())
+ {
+ pitEntry->UpdateLifetime (Seconds (0.10));
+ NS_LOG_DEBUG ("Packet is still in queue and is waiting for its processing");
+ return true; // already in the queue
+ }
+
if (header->GetInterestLifetime () < Seconds (0.1))
{
NS_LOG_DEBUG( "What the fuck? Why interest lifetime is so short? [" << header->GetInterestLifetime ().ToDouble (Time::S) << "s]");
@@ -128,8 +135,17 @@
// std::cerr << (pitEntry->GetExpireTime () - Simulator::Now ()).ToDouble (Time::S) * 1000 << "ms" << std::endl;
pitEntry->OffsetLifetime (Seconds (-pitEntry->GetInterest ()->GetInterestLifetime ().ToDouble (Time::S)));
pitEntry->UpdateLifetime (Seconds (0.10));
+
+ // const ndnSIM::LoadStatsFace &stats = GetStatsTree ()[header->GetName ()].incoming ().find (inFace)->second;
+ const ndnSIM::LoadStatsFace &stats = GetStatsTree ()["/"].incoming ().find (inFace)->second;
+ double weight = std::min (1.0, stats.GetSatisfiedRatio ().get<0> ());
+ if (weight < 0)
+ {
+ // if stats is unknown, gracefully accept interest with normal priority
+ weight = 1.0;
+ }
- bool enqueued = m_pitQueues[outFace].Enqueue (inFace, pitEntry);
+ bool enqueued = m_pitQueues[outFace].Enqueue (inFace, pitEntry, weight);
// if (Simulator::GetContext () == 6)
// {
@@ -153,11 +169,21 @@
if (pitEntry->GetOutgoing ().size () == 0)
{
- Ptr<Packet> pkt = Create<Packet> ();
Ptr<InterestHeader> nackHeader = Create<InterestHeader> (*pitEntry->GetInterest ());
- nackHeader->SetNack (99);
+
+ NS_ASSERT (pitEntry->GetFwTag<PitQueueTag> () != boost::shared_ptr<PitQueueTag> ());
+ if (pitEntry->GetFwTag<PitQueueTag> ()->IsLastOneInQueues ())
+ {
+ nackHeader->SetNack (100);
+ }
+ else
+ {
+ nackHeader->SetNack (101);
+ }
+
+ Ptr<Packet> pkt = Create<Packet> ();
pkt->AddHeader (*nackHeader);
-
+
for (pit::Entry::in_container::iterator face = pitEntry->GetIncoming ().begin ();
face != pitEntry->GetIncoming ().end ();
face ++)
@@ -250,10 +276,12 @@
Ptr<pit::Entry> pitEntry)
{
// NS_LOG_FUNCTION (this << pitEntry->GetPrefix ());
+ PitQueue::Remove (pitEntry);
+
- Ptr<Packet> pkt = Create<Packet> ();
Ptr<InterestHeader> nackHeader = Create<InterestHeader> (*pitEntry->GetInterest ());
- nackHeader->SetNack (99);
+ nackHeader->SetNack (100);
+ Ptr<Packet> pkt = Create<Packet> ();
pkt->AddHeader (*nackHeader);
for (pit::Entry::in_container::iterator face = pitEntry->GetIncoming ().begin ();
@@ -262,9 +290,9 @@
{
face->m_face->Send (pkt->Copy ());
}
-
- PitQueue::Remove (pitEntry);
+
+
for (pit::Entry::out_container::iterator face = pitEntry->GetOutgoing ().begin ();
face != pitEntry->GetOutgoing ().end ();
face ++)
diff --git a/utils/ndn-pit-queue.cc b/utils/ndn-pit-queue.cc
index 34410ba..3423775 100644
--- a/utils/ndn-pit-queue.cc
+++ b/utils/ndn-pit-queue.cc
@@ -34,6 +34,8 @@
namespace ns3 {
namespace ndn {
+const double MIN_WEIGHT = 0.01;
+
PitQueue::PitQueue ()
// : m_maxQueueSize (20)
: m_lastQueue (m_queues.end ())
@@ -55,25 +57,29 @@
bool
PitQueue::Enqueue (Ptr<Face> inFace,
- Ptr<pit::Entry> pitEntry)
+ Ptr<pit::Entry> pitEntry,
+ double updatedWeight)
{
+ if (updatedWeight < MIN_WEIGHT) updatedWeight = MIN_WEIGHT;
+
PerInFaceQueue::iterator queue = m_queues.find (inFace);
if (queue == m_queues.end ())
{
- pair<PerInFaceQueue::iterator, bool> itemPair = m_queues.insert (make_pair (inFace, boost::make_shared<Queue> ()));
+ pair<PerInFaceQueue::iterator, bool> itemPair =
+ m_queues.insert (make_pair (inFace, boost::make_shared< WeightedQueue > (make_tuple (Queue (), updatedWeight, 0))));
m_lastQueue = m_queues.end (); // for some reason end() iterator is invalidated when new item is inserted
NS_ASSERT (itemPair.second == true);
queue = itemPair.first;
}
- if ((inFace->GetLimits ().GetMaxLimit () == 0 && queue->second->size () > 100) ||
- (inFace->GetLimits ().GetMaxLimit () != 0 && queue->second->size () >= 0.5 * inFace->GetLimits ().GetMaxLimit ()))
+ if ((inFace->GetLimits ().GetMaxLimit () == 0 && queue->second->get<0> ().size () > 100) ||
+ (inFace->GetLimits ().GetMaxLimit () != 0 && queue->second->get<0> ().size () >= 0.5 * inFace->GetLimits ().GetMaxLimit ()))
{
return false;
}
- Queue::iterator itemIterator = queue->second->insert (queue->second->end (), pitEntry);
+ Queue::iterator itemIterator = queue->second->get<0> ().insert (queue->second->get<0> ().end (), pitEntry);
shared_ptr<fw::PitQueueTag> tag = pitEntry->GetFwTag<fw::PitQueueTag> ();
if (tag == shared_ptr<fw::PitQueueTag> ())
@@ -82,6 +88,8 @@
pitEntry->AddFwTag (tag);
}
tag->InsertQueue (queue->second, itemIterator);
+
+ UpdateWeightedRounds ();
return true;
}
@@ -91,36 +99,45 @@
{
PerInFaceQueue::iterator queue = m_lastQueue;
- while (queue != m_queues.end () && queue->second->size () == 0) // advance iterator
+ // if (queue != m_queues.end () &&
+ // true)
+ // {
+ // queue ++; // actually implement round robin...
+ // }
+
+ while (queue != m_queues.end () && queue->second->get<0> ().size () == 0) // advance iterator
{
queue ++;
+ m_serviceCounter = 0;
}
if (queue == m_queues.end ())
{
queue = m_queues.begin (); // circle to the beginning
+ m_serviceCounter = 0;
}
- while (queue != m_queues.end () && queue->second->size () == 0) // advance iterator
+ while (queue != m_queues.end () && queue->second->get<0> ().size () == 0) // advance iterator
{
queue ++;
+ m_serviceCounter = 0;
}
if (queue == m_queues.end ()) // e.g., begin () == end ()
return 0;
- NS_ASSERT_MSG (queue->second->size () != 0, "Logic error");
+ NS_ASSERT_MSG (queue->second->get<0> ().size () != 0, "Logic error");
- Ptr<pit::Entry> entry = *queue->second->begin ();
+ Ptr<pit::Entry> entry = *queue->second->get<0> ().begin ();
shared_ptr<fw::PitQueueTag> tag = entry->GetFwTag<fw::PitQueueTag> ();
NS_ASSERT (tag != shared_ptr<fw::PitQueueTag> ());
#ifdef NS3_LOG_ENABLE
- size_t queueSize = queue->second->size ();
+ size_t queueSize = queue->second->get<0> ().size ();
#endif
tag->RemoveFromQueue (queue->second);
#ifdef NS3_LOG_ENABLE
- NS_ASSERT_MSG (queue->second->size () == queueSize-1, "Queue size should be reduced by one");
+ NS_ASSERT_MSG (queue->second->get<0> ().size () == queueSize-1, "Queue size should be reduced by one");
#endif
m_lastQueue = queue;
@@ -139,8 +156,8 @@
if (queue == m_queues.end ())
return;
- for (Queue::iterator pitEntry = queue->second->begin ();
- pitEntry != queue->second->end ();
+ for (Queue::iterator pitEntry = queue->second->get<0> ().begin ();
+ pitEntry != queue->second->get<0> ().end ();
pitEntry ++)
{
shared_ptr<fw::PitQueueTag> tag = (*pitEntry)->GetFwTag<fw::PitQueueTag> ();
@@ -149,7 +166,7 @@
tag->RemoveFromQueuesExcept (queue->second);
}
- NS_ASSERT_MSG (queue->second->size () == 0, "Queue size should be 0 by now");
+ NS_ASSERT_MSG (queue->second->get<0> ().size () == 0, "Queue size should be 0 by now");
m_queues.erase (queue);
}
@@ -172,17 +189,53 @@
queue != m_queues.end ();
queue ++)
{
- isEmpty &= (queue->second->size () == 0);
+ isEmpty &= (queue->second->get<0> ().size () == 0);
}
return isEmpty;
}
-void
-fw::PitQueueTag::InsertQueue (boost::shared_ptr<PitQueue::Queue> queue, PitQueue::Queue::iterator iterator)
+bool
+PitQueue::IsEmpty (Ptr<Face> inFace) const
{
+ PerInFaceQueue::const_iterator queue = m_queues.find (inFace);
+ if (queue == m_queues.end ())
+ return true;
+
+ return queue->second->get<0> ().size () == 0;
+}
+
+
+void
+PitQueue::UpdateWeightedRounds ()
+{
+ double minWeight = 100.0;
+ for (PerInFaceQueue::const_iterator queue = m_queues.begin ();
+ queue != m_queues.end ();
+ queue ++)
+ {
+ if (queue->second->get<1> () < minWeight)
+ minWeight = queue->second->get<1> ();
+ }
+
+ for (PerInFaceQueue::const_iterator queue = m_queues.begin ();
+ queue != m_queues.end ();
+ queue ++)
+ {
+ queue->second->get<2> () = static_cast<uint32_t>((queue->second->get<1> () + 0.5) / minWeight);
+ }
+}
+
+
+////////////////////////////////////////////////////////////////////////////////
+
+void
+fw::PitQueueTag::InsertQueue (boost::shared_ptr<PitQueue::WeightedQueue> queue, PitQueue::Queue::iterator iterator)
+{
+ // std::cerr << "size before: " << m_items.size () << " item: " << (*iterator)->GetPrefix ();
pair<MapOfItems::iterator, bool> item = m_items.insert (make_pair (queue, iterator));
- NS_ASSERT (item.second == true);
+ // std::cerr << " and after: " << m_items.size () << std::endl;
+ NS_ASSERT_MSG (item.second == true, "Should be a new tag for PIT entry, but something is wrong");
}
void
@@ -192,24 +245,24 @@
item != m_items.end ();
item ++)
{
- item->first->erase (item->second);
+ item->first->get<0> ().erase (item->second);
}
m_items.clear ();
}
void
-fw::PitQueueTag::RemoveFromQueue (boost::shared_ptr<PitQueue::Queue> queue)
+fw::PitQueueTag::RemoveFromQueue (boost::shared_ptr<PitQueue::WeightedQueue> queue)
{
MapOfItems::iterator item = m_items.find (queue);
if (item == m_items.end ())
return;
- item->first->erase (item->second);
+ item->first->get<0> ().erase (item->second);
m_items.erase (item);
}
void
-fw::PitQueueTag::RemoveFromQueuesExcept (boost::shared_ptr<PitQueue::Queue> queue)
+fw::PitQueueTag::RemoveFromQueuesExcept (boost::shared_ptr<PitQueue::WeightedQueue> queue)
{
for (MapOfItems::iterator item = m_items.begin ();
item != m_items.end (); )
@@ -220,7 +273,7 @@
continue;
}
- item->first->erase (item->second);
+ item->first->get<0> ().erase (item->second);
MapOfItems::iterator itemToDelete = item;
item ++;
@@ -228,6 +281,21 @@
}
}
+bool
+fw::PitQueueTag::IsLastOneInQueues () const
+{
+ bool lastOne = true;
+
+ for (MapOfItems::const_iterator item = m_items.begin ();
+ item != m_items.end ();
+ item ++)
+ {
+ lastOne &= (item->first->get<0> ().size () == 1);
+ }
+
+ return lastOne;
+}
+
} // namespace ndn
} // namespace ns3
diff --git a/utils/ndn-pit-queue.h b/utils/ndn-pit-queue.h
index ebac04a..d7e20ec 100644
--- a/utils/ndn-pit-queue.h
+++ b/utils/ndn-pit-queue.h
@@ -69,11 +69,13 @@
* @brief Enqueue PIT entry for delayed processing
* @param inFace incoming face to which queue PIT entry should enqueued
* @param pitEntry smart pointer to PIT entry
+ * @param updatedWeight the current value for the inFace weight (stats-based weight)
* return true if successfully enqueued, false if limit is reached or some other reason
*/
bool
Enqueue (Ptr<Face> inFace,
- Ptr<pit::Entry> pitEntry);
+ Ptr<pit::Entry> pitEntry,
+ double updatedWeight);
/**
* @brief Get next PIT entry
@@ -99,19 +101,33 @@
Remove (Ptr<pit::Entry> entry);
/**
- * @brief Check if queue is empty
+ * @brief Check if all per-in-face queues is empty
*/
bool
IsEmpty () const;
+
+ /**
+ * @brief Check if the specified per-in-face queue is empty
+ */
+ bool
+ IsEmpty (Ptr<Face> inFace) const;
+
+private:
+ void
+ UpdateWeightedRounds ();
public:
typedef std::list< Ptr<pit::Entry> > Queue;
- typedef std::map< Ptr<Face>, boost::shared_ptr<Queue> > PerInFaceQueue;
+ typedef boost::tuple< Queue, double, uint32_t > WeightedQueue;
+ typedef std::map< Ptr<Face>, boost::shared_ptr<WeightedQueue> > PerInFaceQueue;
private:
// uint32_t m_maxQueueSize;
PerInFaceQueue::iterator m_lastQueue; // last queue from which interest was taken
PerInFaceQueue m_queues;
+
+ uint32_t m_serviceCounter;
+
};
namespace fw {
@@ -125,7 +141,7 @@
{
public:
// map based on addresses, should be good enough
- typedef std::map< boost::shared_ptr<PitQueue::Queue>, PitQueue::Queue::iterator > MapOfItems;
+ typedef std::map< boost::shared_ptr<PitQueue::WeightedQueue>, PitQueue::Queue::iterator > MapOfItems;
public:
/**
@@ -140,7 +156,7 @@
* @brief iterator queue's iterator
*/
void
- InsertQueue (boost::shared_ptr<PitQueue::Queue> item, PitQueue::Queue::iterator iterator);
+ InsertQueue (boost::shared_ptr<PitQueue::WeightedQueue> item, PitQueue::Queue::iterator iterator);
/**
* @brief Remove PIT entry from all queues
@@ -153,14 +169,20 @@
* @param queue Queue from which PIT entry should be removed
*/
void
- RemoveFromQueue (boost::shared_ptr<PitQueue::Queue> queue);
+ RemoveFromQueue (boost::shared_ptr<PitQueue::WeightedQueue> queue);
/**
* @brief Remove reference to PIT from queues except the queue in parameter
* @param queue Queue from which PIT entry should not be removed (to preserve iterator)
*/
void
- RemoveFromQueuesExcept (boost::shared_ptr<PitQueue::Queue> queue);
+ RemoveFromQueuesExcept (boost::shared_ptr<PitQueue::WeightedQueue> queue);
+
+ /**
+ * @brief Check if removal of the entry will empty all the queues
+ */
+ bool
+ IsLastOneInQueues () const;
private:
MapOfItems m_items;