Merge "Fix corner case in RibOut tail dequeue scheduling logic" into …

…R2.0
Juniper · Jun 25, 2015 · 049c4c5 · 049c4c5
2 parents b1bde6b + 7218839
commit 049c4c5
Show file tree

Hide file tree

Showing 6 changed files with 26 additions and 12 deletions.
diff --git a/src/bgp/bgp_peer_membership.cc b/src/bgp/bgp_peer_membership.cc
@@ -208,9 +208,14 @@ void IPeerRib::RegisterRibOut(RibExportPolicy policy) {
     ribout_->RegisterListener();
     ribout_->Register(ipeer_);
 
+    int index = ribout_->GetPeerIndex(ipeer_);
     RibOutUpdates *updates = ribout_->updates();
-    updates->QueueJoin(RibOutUpdates::QUPDATE, ribout_->GetPeerIndex(ipeer_));
-    updates->QueueJoin(RibOutUpdates::QBULK, ribout_->GetPeerIndex(ipeer_));
+    SchedulingGroup *group = ribout_->GetSchedulingGroup();
+    assert(group);
+    if (updates->QueueJoin(RibOutUpdates::QUPDATE, index))
+        group->RibOutActive(ribout_, RibOutUpdates::QUPDATE);
+    if (updates->QueueJoin(RibOutUpdates::QBULK, index))
+        group->RibOutActive(ribout_, RibOutUpdates::QBULK);
 
     SetRibOutRegistered(true);
 }
@@ -224,9 +229,10 @@ void IPeerRib::RegisterRibOut(RibExportPolicy policy) {
 // of the RibOut itself.
 //
 void IPeerRib::UnregisterRibOut() {
+    int index = ribout_->GetPeerIndex(ipeer_);
     RibOutUpdates *updates = ribout_->updates();
-    updates->QueueLeave(RibOutUpdates::QUPDATE, ribout_->GetPeerIndex(ipeer_));
-    updates->QueueLeave(RibOutUpdates::QBULK, ribout_->GetPeerIndex(ipeer_));
+    updates->QueueLeave(RibOutUpdates::QUPDATE, index);
+    updates->QueueLeave(RibOutUpdates::QBULK, index);
 
     ribout_->Unregister(ipeer_);
     ribout_ = NULL;

diff --git a/src/bgp/bgp_ribout_updates.cc b/src/bgp/bgp_ribout_updates.cc
@@ -571,9 +571,9 @@ bool RibOutUpdates::Empty() const {
     return true;
 }
 
-void RibOutUpdates::QueueJoin(int queue_id, int bit) {
+bool RibOutUpdates::QueueJoin(int queue_id, int bit) {
     UpdateQueue *queue = queue_vec_[queue_id];
-    queue->Join(bit);
+    return queue->Join(bit);
 }
 
 void RibOutUpdates::QueueLeave(int queue_id, int bit) {

diff --git a/src/bgp/bgp_ribout_updates.h b/src/bgp/bgp_ribout_updates.h
@@ -51,7 +51,7 @@ class RibOutUpdates {
                              const RibPeerSet &mready, RibPeerSet *blocked);
 
     // Enqueue a marker at the head of the queue with this bit set.
-    void QueueJoin(int queue_id, int bit);
+    bool QueueJoin(int queue_id, int bit);
     void QueueLeave(int queue_id, int bit);
 
     bool Empty() const;

diff --git a/src/bgp/bgp_update_queue.cc b/src/bgp/bgp_update_queue.cc
@@ -248,11 +248,18 @@ UpdateMarker *UpdateQueue::GetMarker(int bit) {
 // Since it's a new peer, it starts out at the tail marker.  Also add the
 // peer's bit index and the tail marker pair to the MarkerMap.
 //
-void UpdateQueue::Join(int bit) {
+// Return true if the tail marker is not the last entry in the queue. The
+// caller should trigger a tail dequeue for the RibOut if so to take care
+// of the possibility that all peers in the tail marker are blocked. Note
+// that a tail dequeue may already be pending in the SchedulingGroup, but
+// an extra one is harmless.
+//
+bool UpdateQueue::Join(int bit) {
     tbb::mutex::scoped_lock lock(mutex_);
     UpdateMarker *marker = &tail_marker_;
     marker->members.set(bit);
     markers_.insert(std::make_pair(bit, marker));
+    return (&tail_marker_ != &queue_.back());
 }
 
 //

diff --git a/src/bgp/bgp_update_queue.h b/src/bgp/bgp_update_queue.h
@@ -118,7 +118,7 @@ class UpdateQueue {
                      const RibPeerSet &bitset);
     UpdateMarker *GetMarker(int bit);
 
-    void Join(int bit);
+    bool Join(int bit);
     void Leave(int bit);
 
     bool CheckInvariants() const;

diff --git a/src/bgp/scheduling_group.cc b/src/bgp/scheduling_group.cc
@@ -773,7 +773,7 @@ void SchedulingGroup::GetPeerList(PeerList *plist) const {
 // active.
 //
 void SchedulingGroup::RibOutActive(RibOut *ribout, int queue_id) {
-    CHECK_CONCURRENCY("db::DBTable", "bgp::SendTask");
+    CHECK_CONCURRENCY("db::DBTable", "bgp::SendTask", "bgp::PeerMembership");
 
     WorkRibOutEnqueue(ribout, queue_id);
 }
@@ -831,7 +831,8 @@ auto_ptr<SchedulingGroup::WorkBase> SchedulingGroup::WorkDequeue() {
 // task if required.
 //
 void SchedulingGroup::WorkEnqueue(WorkBase *wentry) {
-    CHECK_CONCURRENCY("db::DBTable", "bgp::SendTask", "bgp::SendReadyTask");
+    CHECK_CONCURRENCY("db::DBTable", "bgp::SendTask", "bgp::SendReadyTask",
+        "bgp::PeerMembership");
 
     tbb::mutex::scoped_lock lock(mutex_);
     work_queue_.push_back(wentry);
@@ -857,7 +858,7 @@ void SchedulingGroup::WorkPeerEnqueue(IPeerUpdate *peer) {
 // Enqueue a WorkRibOut to the work queue.
 //
 void SchedulingGroup::WorkRibOutEnqueue(RibOut *ribout, int queue_id) {
-    CHECK_CONCURRENCY("db::DBTable", "bgp::SendTask");
+    CHECK_CONCURRENCY("db::DBTable", "bgp::SendTask", "bgp::PeerMembership");
 
     WorkBase *wentry = new WorkRibOut(ribout, queue_id);
     WorkEnqueue(wentry);