LCOV - code coverage report
Current view: top level - bgp - bgp_ribout_updates.cc (source / functions) Hit Total Coverage
Test: OpenSDN C/C++ coverage (all TARGET_SET jobs) Lines: 284 290 97.9 %
Date: 2026-11-09 02:10:49 Functions: 24 24 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #include "bgp/bgp_ribout_updates.h"
       7             : 
       8             : #include <string>
       9             : 
      10             : #include "sandesh/sandesh_trace.h"
      11             : #include "base/task_annotations.h"
      12             : #include "bgp/bgp_log.h"
      13             : #include "bgp/bgp_peer_types.h"
      14             : #include "bgp/bgp_ribout.h"
      15             : #include "bgp/bgp_route.h"
      16             : #include "bgp/bgp_update_queue.h"
      17             : #include "bgp/bgp_update_monitor.h"
      18             : #include "bgp/bgp_update_sender.h"
      19             : #include "bgp/message_builder.h"
      20             : 
      21             : using std::unique_ptr;
      22             : using std::vector;
      23             : 
      24             : vector<Message *> RibOutUpdates::bgp_messages_;
      25             : vector<Message *> RibOutUpdates::xmpp_messages_;
      26             : 
      27             : //
      28             : // Create a new RibOutUpdates.  Also create the necessary UpdateQueue and
      29             : // add them to the vector.
      30             : //
      31      136016 : RibOutUpdates::RibOutUpdates(RibOut *ribout, int index)
      32      136016 :     : ribout_(ribout),
      33      136016 :       index_(index) {
      34      408048 :     for (int i = 0; i < QCOUNT; i++) {
      35      272032 :         UpdateQueue *queue = new UpdateQueue(ribout, i);
      36      272032 :         queue_vec_.push_back(queue);
      37             :     }
      38      136016 :     monitor_.reset(new RibUpdateMonitor(ribout, &queue_vec_));
      39      136016 :     memset(&stats_, 0, sizeof(stats_));
      40      136016 : }
      41             : 
      42             : //
      43             : // Destructor.  Get rid of all the UpdateQueues.
      44             : //
      45      271732 : RibOutUpdates::~RibOutUpdates() {
      46      136016 :     STLDeleteValues(&queue_vec_);
      47      271732 : }
      48             : 
      49             : //
      50             : // Initialize static vectors of bgp/xmpp message pointers to NULL.
      51             : //
      52         158 : void RibOutUpdates::Initialize() {
      53         158 :     bgp_messages_.resize(DB::PartitionCount(), NULL);
      54         158 :     xmpp_messages_.resize(DB::PartitionCount(), NULL);
      55         158 : }
      56             : 
      57             : //
      58             : // Free any memory allocated for bgp/xmpp messages.
      59             : //
      60          62 : void RibOutUpdates::Terminate() {
      61          62 :     STLDeleteValues(&bgp_messages_);
      62          62 :     STLDeleteValues(&xmpp_messages_);
      63          62 : }
      64             : 
      65             : //
      66             : // Create if needed, and return the bgp/xmpp message for this RibOutUpdates.
      67             : // Note that we use static vectors of bgp/xmpp messages, one per partition,
      68             : // so that we don't need to allocate and free messages repeatedly.
      69             : //
      70      612727 : Message *RibOutUpdates::GetMessage() const {
      71      612727 :     if (ribout_->IsEncodingBgp()) {
      72      156614 :         if (!bgp_messages_[index_]) {
      73             :             MessageBuilder *builder =
      74         154 :                 MessageBuilder::GetInstance(RibExportPolicy::BGP);
      75         154 :             Message *message = builder->Create();
      76         154 :             bgp_messages_[index_] = message;
      77             :         }
      78      156614 :         return bgp_messages_[index_];
      79             :     }
      80      456108 :     if (ribout_->IsEncodingXmpp()) {
      81      456149 :         if (!xmpp_messages_[index_]) {
      82             :             MessageBuilder *builder =
      83         144 :                 MessageBuilder::GetInstance(RibExportPolicy::XMPP);
      84         144 :             Message *message = builder->Create();
      85         144 :             xmpp_messages_[index_] = message;
      86             :         }
      87      456149 :         return xmpp_messages_[index_];
      88             :     }
      89           0 :     return NULL;
      90             : }
      91             : 
      92             : //
      93             : // Concurrency: Called in the context of the routing table partition task.
      94             : //
      95             : // Enqueue the RouteUpdate corresponding to the DBEntryBase into the queue.
      96             : // This is called in the context of the routing table partition task. All
      97             : // the concurrency issues are handled by going through the monitor.
      98             : //
      99             : // If the UpdateQueue corresponding to the RouteUpdate previously had no
     100             : // updates after the tail marker, we kick the BgpUpdateSender to perform
     101             : // a tail dequeue for the RibOut.
     102             : //
     103      900006 : void RibOutUpdates::Enqueue(DBEntryBase *db_entry, RouteUpdate *rt_update) {
     104      900006 :     CHECK_CONCURRENCY("db::DBTable");
     105             : 
     106      900088 :     bool need_tail_dequeue = monitor_->EnqueueUpdate(db_entry, rt_update);
     107      900226 :     if (need_tail_dequeue) {
     108      532327 :         ribout_->sender()->RibOutActive(index_, ribout_, rt_update->queue_id());
     109             :     }
     110      900328 : }
     111             : 
     112             : //
     113             : // Concurrency: Called in the context of the bgp::SendUpdate task.
     114             : //
     115             : // Common dequeue routine invoked by tail dequeue and peer dequeue. It builds
     116             : // and sends updates for each UpdateInfo element in the list hanging off the
     117             : // RouteUpdate. For each update that it builds, it also includes prefixes for
     118             : // other UpdateInfo elements that share the same attributes, provided that
     119             : // the associated RouteUpdate was enqueued after the original one.
     120             : 
     121             : // Each update is targeted at the peers in the RibPeerSet of the UpdateMarker
     122             : // passed in to us.  This set of peers is subsequently culled based on the
     123             : // RibPeerSet in each UpdateInfo.  IOW, the update is sent only to the set
     124             : // of peers in the intersection of the UpdateMarker and the UpdateInfo. Note
     125             : // that the UpdateMarker could specify a single peer if we are called from
     126             : // peer dequeue.
     127             : //
     128             : // Return false if all the peers in the marker get blocked.  In any case, the
     129             : // blocked parameter is populated with the set of peers that are send blocked.
     130             : //
     131      623333 : bool RibOutUpdates::DequeueCommon(UpdateQueue *queue, UpdateMarker *marker,
     132             :         RouteUpdate *rt_update, RibPeerSet *blocked) {
     133      623333 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     134             : 
     135             :     // Pass a hint to the Message telling it whether it needs to cache the
     136             :     // formatted version of each route. This is used only for xmpp messages.
     137             :     // Heuristic is to cache if there's markers other than the tail marker.
     138             :     // The reasoning is that the cached version can be used later when the
     139             :     // markers in question are being processed.  Put another way - there's
     140             :     // no need to cache if route is not going to be advertised to any peers
     141             :     // other than the ones in the given UpdateMarker.
     142      623468 :     bool cache_routes = queue->marker_count() != 0;
     143             : 
     144             :     // Go through all UpdateInfo elements for the RouteUpdate.
     145      623458 :     int queue_id = rt_update->queue_id();
     146      623459 :     RibPeerSet rt_blocked;
     147      623461 :     for (UpdateInfoSList::List::iterator iter = rt_update->Updates()->begin();
     148     2502926 :          iter != rt_update->Updates()->end();) {
     149             :         // Get the UpdateInfo and move the iterator to next one before doing
     150             :         // any processing, since we may delete the UpdateInfo further down.
     151      627942 :         UpdateInfo *uinfo = iter.operator->();
     152             :         ++iter;
     153             : 
     154             :         // Skip if there's no overlap between the UpdateMarker and the targets
     155             :         // for the UpdateInfo.  The intersection is the set of peers to which
     156             :         // the message we are about to build will be sent.
     157      627942 :         RibPeerSet msgset;
     158      627965 :         msgset.BuildIntersection(uinfo->target, marker->members);
     159      627868 :         if (msgset.empty()) {
     160       15133 :             continue;
     161             :         }
     162             : 
     163             :         // Generate the update, merge additional updates into that message and
     164             :         // send it message to the target RibPeerSet.
     165             :         //
     166             :         // In the rare case that the first route and it's attributes don't fit
     167             :         // into the message, clear the target bits in the UpdateInfo to ensure
     168             :         // that the UpdateQueue doesn't get wedged. However, don't update the
     169             :         // history bits in the RouteUpdate since the message did not get sent.
     170             :         //
     171             :         // The Create routine has the responsibility of logging an error and
     172             :         // incrementing any counters.
     173      612724 :         RibPeerSet msg_blocked;
     174      612719 :         stats_[queue_id].messages_built_count_++;
     175      612719 :         Message *message = GetMessage();
     176      612702 :         assert(message);
     177     1838220 :         bool msg_built = message->Start(
     178      612702 :             ribout_, cache_routes, &uinfo->roattr, rt_update->route());
     179      612824 :         if (msg_built) {
     180      612723 :             UpdatePack(queue_id, message, uinfo, msgset);
     181      612727 :             message->Finish();
     182      612716 :             UpdateSend(queue_id, message, msgset, &msg_blocked);
     183             :         }
     184             : 
     185             :         // Reset bits in the UpdateInfo.  Note that this has already been done
     186             :         // via UpdatePack for all the other UpdateInfo elements that we packed
     187             :         // into this message.
     188      612969 :         bool empty = ClearAdvertisedBits(rt_update, uinfo, msgset, msg_built);
     189      612927 :         if (empty) {
     190      612048 :             rt_update->RemoveUpdateInfo(uinfo);
     191             :         }
     192             : 
     193             :         // Update RibPeerSet of peers that got blocked while processing this
     194             :         // RouteUpdate. Since there's no overlap of peers between UpdateInfos
     195             :         // for the same RouteUpdate, we can update the markers for all blocked
     196             :         // peers in one shot i.e. outside this loop.
     197      612930 :         rt_blocked |= msg_blocked;
     198      628032 :     }
     199             : 
     200             :     // Update the markers for any peers that got blocked while processing this
     201             :     // RouteUpdate. If all peers in the UpdateMarker got blocked, we shouldn't
     202             :     // build any more update messages.  Return false to let the callers know
     203             :     // that this has happened.
     204      623459 :     if (rt_blocked.empty()) {
     205      621743 :         return true;
     206             :     } else {
     207        1810 :         *blocked |= rt_blocked;
     208        1814 :         return !UpdateMarkersOnBlocked(marker, rt_update, &rt_blocked);
     209             :     }
     210      623557 : }
     211             : 
     212             : //
     213             : // Concurrency: Called in the context of the bgp::SendUpdate task.
     214             : //
     215             : // Dequeue and build updates for the in-sync peers in the RibPeerSet of the
     216             : // tail marker for the given queue id.
     217             : //
     218             : // Return false if all the peers in the marker get blocked.  In any case, the
     219             : // blocked parameter is populated with the set of peers that are send blocked
     220             : // and the unsync parameter is populated with the set of peers from the tail
     221             : // marker that are not in the msync set passed in to the method.
     222             : //
     223      612245 : bool RibOutUpdates::TailDequeue(int queue_id, const RibPeerSet &msync,
     224             :         RibPeerSet *blocked, RibPeerSet *unsync) {
     225      612245 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     226             : 
     227      612369 :     stats_[queue_id].tail_dequeue_count_++;
     228      612369 :     UpdateQueue *queue = queue_vec_[queue_id];
     229      612362 :     UpdateMarker *start_marker = queue->tail_marker();
     230      612365 :     RouteUpdatePtr update = monitor_->GetNextUpdate(queue_id, start_marker);
     231             : 
     232      612389 :     if (update.get() == NULL) {
     233        1458 :         return true;
     234             :     }
     235             : 
     236             :     // Intersect marker membership and in-sync peers to come up with the
     237             :     // unsync peers. If all the peers are unsync return right away. The
     238             :     // BgpUpdateSender will take care of triggering a TailDequeue again
     239             :     // when at least one peer becomes in-sync.
     240      610930 :     unsync->BuildComplement(start_marker->members, msync);
     241      610867 :     if (*unsync == start_marker->members) {
     242        5278 :         return false;
     243             :     }
     244             : 
     245             :     // Split the unsync peers from the tail marker. Note that this updates
     246             :     // the RibPeerSet in the tail marker.
     247      605579 :     if (!unsync->empty()) {
     248        2594 :         stats_[queue_id].marker_split_count_++;
     249        2594 :         queue->MarkerSplit(start_marker, *unsync);
     250             :     }
     251             : 
     252             :     // Update send loop. Select next update to send, format a message.
     253             :     // Add other updates with the same attributes and replicate the
     254             :     // packet.
     255      605574 :     RibPeerSet members = start_marker->members;
     256      605577 :     RouteUpdatePtr next_update;
     257     1227475 :     for (; update.get() != NULL; update = next_update) {
     258      622945 :         if (!DequeueCommon(queue, start_marker, update.get(), blocked)) {
     259             :             // Be sure to get rid of the RouteUpdate if it's empty.
     260        1107 :             if (update->empty()) {
     261        1026 :                 ClearUpdate(&update);
     262             :             }
     263             : 
     264        1107 :             return false;
     265             :         }
     266             : 
     267             :         // Iterate to the next update before we potentially delete the
     268             :         // current one. If there are no more updates in the queue, the
     269             :         // marker will get moved so that it's after the current update.
     270      622000 :         next_update = monitor_->GetNextUpdate(queue_id, update.get());
     271             : 
     272             :         // Be sure to get rid of the RouteUpdate if it's empty.
     273      621998 :         if (update->empty()) {
     274      606324 :             ClearUpdate(&update);
     275             :         }
     276             :     }
     277             : 
     278             :     // Request peers to flush accumulated update messages.
     279             :     // Return false if all peers got blocked.
     280      604510 :     UpdateFlush(members, blocked);
     281      604618 :     return (members != *blocked);
     282      612471 : }
     283             : 
     284             : //
     285             : // Concurrency: Called in the context of the bgp::SendUpdate task.
     286             : //
     287             : // Dequeue and build updates for all the peers that share the same marker as
     288             : // the specified peer.  This routine has some extra intelligence beyond the
     289             : // TailDequeue. As it encounters update markers, it merges in any send ready
     290             : // peers from those with the marker being processed for dequeue. This is done
     291             : // to reduce the number of times we build an update message containing the
     292             : // the same information.
     293             : //
     294             : // Return false if all the peers in the marker get blocked.  In any case, the
     295             : // blocked parameter is populated with the set of peers that are send blocked.
     296             : //
     297         300 : bool RibOutUpdates::PeerDequeue(int queue_id, IPeerUpdate *peer,
     298             :         RibPeerSet *blocked) {
     299         300 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     300             : 
     301         300 :     stats_[queue_id].peer_dequeue_count_++;
     302         300 :     UpdateQueue *queue = queue_vec_[queue_id];
     303         300 :     int peer_idx = ribout_->GetPeerIndex(peer);
     304         300 :     UpdateMarker *start_marker = queue->GetMarker(peer_idx);
     305             : 
     306             :     // We're done if this is the same as the tail marker.  Updates will be
     307             :     // built subsequently via TailDequeue.
     308         300 :     assert(start_marker);
     309         300 :     if (start_marker == queue->tail_marker()) {
     310         210 :         return true;
     311             :     }
     312             : 
     313             :     // We're done if the lead peer is not send ready. This can happen if
     314             :     // the peer got blocked when processing updates in another partition.
     315          90 :     RibPeerSet mready;
     316          90 :     ribout_->BuildSendReadyBitSet(start_marker->members, &mready);
     317          90 :     if (!mready.test(peer_idx)) {
     318           0 :         blocked->set(peer_idx);
     319           0 :         return false;
     320             :     }
     321             : 
     322             :     // Split out any peers from the marker that are not send ready. Note that
     323             :     // this updates the RibPeerSet in the marker.
     324          90 :     RibPeerSet notready;
     325          90 :     notready.BuildComplement(start_marker->members, mready);
     326          90 :     if (!notready.empty()) {
     327           3 :         stats_[queue_id].marker_split_count_++;
     328           3 :         queue->MarkerSplit(start_marker, notready);
     329             :     }
     330             : 
     331             :     // Get the encapsulator for the first RouteUpdate.  Even if there's no
     332             :     // RouteUpdate, we should find another marker or the tail marker.
     333             :     UpdateEntry *upentry;
     334             :     RouteUpdatePtr update =
     335          90 :         monitor_->GetNextEntry(queue_id, start_marker, &upentry);
     336          90 :     assert(upentry);
     337             : 
     338             :     // Update loop.  Keep going till we reach the tail marker or till all the
     339             :     // peers get blocked.
     340          90 :     RibPeerSet members = start_marker->members;
     341          90 :     RouteUpdatePtr next_update;
     342             :     UpdateEntry *next_upentry;
     343         507 :     for (; upentry != NULL; upentry = next_upentry, update = next_update) {
     344         507 :         UpdateMarker *marker = NULL;
     345         507 :         if (upentry->IsMarker()) {
     346             :             // The queue entry is a marker.  We're done if we've reached the
     347             :             // tail marker.  Updates will be built later via TailDequeue.
     348          88 :             marker = static_cast<UpdateMarker *>(upentry);
     349          88 :             if (marker == queue->tail_marker()) {
     350          75 :                 stats_[queue_id].marker_merge_count_++;
     351          75 :                 queue->MarkerMerge(queue->tail_marker(), start_marker,
     352          75 :                         start_marker->members);
     353          75 :                 break;
     354             :             }
     355             :         } else {
     356             :             // The queue entry is a RouteUpdate. Go ahead and build an update
     357             :             // message.  Bail if all the peers in the marker get blocked.
     358         419 :             if (!DequeueCommon(queue, start_marker, update.get(), blocked)) {
     359             :                 // Be sure to get rid of the RouteUpdate if it's empty.
     360          15 :                 if (update->empty()) {
     361          14 :                     ClearUpdate(&update);
     362             :                 }
     363          15 :                 break;
     364             :             }
     365             :         }
     366             : 
     367             :         // Iterate to the next element before we potentially delete the
     368             :         // current one.
     369             :         next_update =
     370         417 :             monitor_->GetNextEntry(queue_id, upentry, &next_upentry);
     371             : 
     372         417 :         if (upentry->IsMarker()) {
     373             :             // As the entry is a marker, merge send-ready peers from it
     374             :             // with the marker that is being processed for dequeue.  Note
     375             :             // that this updates the RibPeerSet in the marker.
     376          13 :             RibPeerSet mmove;
     377          13 :             ribout_->BuildSendReadyBitSet(marker->members, &mmove);
     378          13 :             if  (!mmove.empty()) {
     379           9 :                 stats_[queue_id].marker_merge_count_++;
     380           9 :                 queue->MarkerMerge(start_marker, marker, mmove);
     381           9 :                 members |= mmove;
     382             :             }
     383         417 :         } else if (update->empty()) {
     384             :             // Be sure to get rid of the RouteUpdate since it's empty.
     385         359 :             ClearUpdate(&update);
     386             :         }
     387             :     }
     388             : 
     389             :     // Request peers to flush accumulated update messages.
     390             :     // Return false if all peers got blocked.
     391          90 :     UpdateFlush(members, blocked);
     392          90 :     return (members != *blocked);
     393          90 : }
     394             : 
     395             : //
     396             : // Concurrency: Called in the context of the bgp::SendUpdate task.
     397             : //
     398             : // Go through all the UpdateInfo elements that have the same attribute as
     399             : // the start parameter and pack the corresponding prefixes into the Message.
     400             : // The attributes and the prefix associated with start are already in the
     401             : // Message when this method is invoked.
     402             : //
     403             : // The set of peers for which this update is being built is represented by
     404             : // the msgset parameter.  As the msgset has already been determined by the
     405             : // caller, we should only add prefixes that need to go to all the peers in
     406             : // the msgset.
     407             : //
     408      612687 : void RibOutUpdates::UpdatePack(int queue_id, Message *message,
     409             :         UpdateInfo *start_uinfo, const RibPeerSet &msgset) {
     410      612687 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     411             : 
     412             :     UpdateInfo *uinfo, *next_uinfo;
     413      612676 :     RouteUpdatePtr next_update;
     414             : 
     415             :     // Walk through all the UpdateInfo elements with the same attribute in
     416             :     // enqueue order.
     417             :     RouteUpdatePtr update =
     418      612674 :         monitor_->GetAttrNext(queue_id, start_uinfo, &uinfo);
     419      927832 :     for (; update.get() != NULL; update = next_update, uinfo = next_uinfo) {
     420             :         // Iterate to the next element before we potentially delete the
     421             :         // current one.
     422      315331 :         next_update = monitor_->GetAttrNext(queue_id, uinfo, &next_uinfo);
     423             : 
     424             :         // Skip if the msgset RibPeerSet is not a subset of the target in
     425             :         // UpdateInfo.
     426      315267 :         if (!uinfo->target.Contains(msgset))
     427        3681 :             continue;
     428             : 
     429             :         // Go ahead and add the route to the message.  Terminate the loop
     430             :         // if the message doesn't have room for the route.  The route will
     431             :         // get included in another update message.
     432      311588 :         bool success = message->AddRoute(update->route(), &uinfo->roattr);
     433      311614 :         if (!success) {
     434         214 :             break;
     435             :         }
     436             : 
     437             :         // First clear the advertised bits as represented by msgset from
     438             :         // the target RibPeerSet in the UpdateInfo. If the target is now
     439             :         // empty, remove the UpdateInfo from the list container in the
     440             :         // underlying RouteUpdate.
     441             :         //
     442             :         // If the RouteUpdate itself is now empty i.e. there are no more
     443             :         // UpdateInfo elements associated with it, we can get rid of it.
     444      311400 :         bool empty = ClearAdvertisedBits(update.get(), uinfo, msgset, true);
     445      311351 :         if (empty && update->RemoveUpdateInfo(uinfo)) {
     446      310608 :             ClearUpdate(&update);
     447             :         }
     448             :     }
     449      612670 : }
     450             : 
     451             : //
     452             : // Concurrency: Called in the context of the bgp::SendUpdate task.
     453             : //
     454             : // Go through all the peers in the specified RibPeerSet and send the given
     455             : // message to each of them.  Update the blocked RibPeerSet with peers that
     456             : // become blocked after sending the message.
     457             : //
     458      612708 : void RibOutUpdates::UpdateSend(int queue_id, Message *message,
     459             :         const RibPeerSet &dst, RibPeerSet *blocked) {
     460      612708 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     461             : 
     462      612681 :     RibOut::PeerIterator iter(ribout_, dst);
     463     2351188 :     while (iter.HasNext()) {
     464     1738256 :         int ix_current = iter.index();
     465     1738217 :         IPeerUpdate *peer = iter.Next();
     466     1737731 :         size_t msgsize = 0;
     467     1737731 :         const string *msg_str = NULL;
     468     1737731 :         string temp;
     469     1737721 :         const uint8_t *data = message->GetData(peer, &msgsize, &msg_str, &temp);
     470     1737945 :         if (Sandesh::LoggingLevel() >= Sandesh::LoggingUtLevel()) {
     471     2107716 :             BGP_LOG_PEER(Message, peer, Sandesh::LoggingUtLevel(),
     472             :                 BGP_LOG_FLAG_SYSLOG, BGP_PEER_DIR_OUT,
     473             :                 "Update size " << msgsize <<
     474             :                 " reach " << message->num_reach_routes() <<
     475             :                 " unreach " << message->num_unreach_routes());
     476             :         }
     477     1738396 :         stats_[queue_id].messages_sent_count_++;
     478     1738396 :         stats_[queue_id].reach_count_ += message->num_reach_routes();
     479     1738313 :         stats_[queue_id].unreach_count_ += message->num_unreach_routes();
     480     1738325 :         bool more = peer->SendUpdate(data, msgsize, msg_str);
     481     1738372 :         if (!more) {
     482        3980 :             blocked->set(ix_current);
     483             :         }
     484     1738471 :         IPeer *ipeer = dynamic_cast<IPeer *>(peer);
     485     1738471 :         if (!ipeer) {
     486       10472 :             continue;
     487             :         }
     488     1727999 :         IPeerDebugStats *stats = ipeer->peer_stats();
     489     1727996 :         if (stats) {
     490     1727997 :             stats->UpdateTxReachRoute(message->num_reach_routes());
     491     1728071 :             stats->UpdateTxUnreachRoute(message->num_unreach_routes());
     492             :         }
     493     1738556 :     }
     494      612761 : }
     495             : 
     496             : //
     497             : // Concurrency: Called in the context of the bgp::SendUpdate task.
     498             : //
     499             : // Go through all the peers in the specified RibPeerSet and ask them to flush
     500             : // i.e. send immediately, any accumulated updates.  Update blocked RibPeerSet
     501             : // with peers that become blocked after flushing.
     502             : //
     503             : // Skip if the RibOut is XMPP.
     504             : //
     505      604649 : void RibOutUpdates::UpdateFlush(const RibPeerSet &dst, RibPeerSet *blocked) {
     506      604649 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     507             : 
     508      604670 :     if (ribout_->IsEncodingXmpp())
     509      464129 :         return;
     510             : 
     511      140534 :     RibOut::PeerIterator iter(ribout_, dst);
     512      296588 :     while (iter.HasNext()) {
     513      155991 :         int ix_current = iter.index();
     514      155990 :         IPeerUpdate *peer = iter.Next();
     515      155994 :         bool more = peer->FlushUpdate();
     516      156024 :         if (!more) {
     517          14 :             blocked->set(ix_current);
     518             :         }
     519             :     }
     520             : }
     521             : 
     522             : //
     523             : // Take the AdvertisedInfo history in the RouteUpdate and move it to a new
     524             : // RouteState. Go through the monitor to associate the new RouteState as the
     525             : // listener state for the Route.
     526             : //
     527      584056 : void RibOutUpdates::StoreHistory(RouteUpdate *rt_update) {
     528      584056 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     529             : 
     530      584028 :     BgpRoute *route = rt_update->route();
     531      584026 :     RouteState *rstate = new RouteState();
     532      584008 :     rt_update->MoveHistory(rstate);
     533      584031 :     monitor_->SetEntryState(route, rstate);
     534      584152 : }
     535             : 
     536             : //
     537             : // Go through the monitor to clear the listener state for the underlying Route
     538             : // of the RouteUpdate.
     539             : //
     540      334027 : void RibOutUpdates::ClearState(RouteUpdate *rt_update) {
     541      334027 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     542             : 
     543      333992 :     BgpRoute *route = rt_update->route();
     544      333989 :     monitor_->ClearEntryState(route);
     545      334129 : }
     546             : 
     547             : //
     548             : // Called when the RouteUpdate encapsulated by the RouteUpdatePtr has no more
     549             : // UpdateInfo elements. Releases ownership of the RouteUpdate and deletes the
     550             : // RouteUpdate, as well as any associated UpdateList if appropriate.
     551             : //
     552      918287 : void RibOutUpdates::ClearUpdate(RouteUpdatePtr *update) {
     553      918287 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     554             : 
     555             :     // Dequeue the route update.
     556      918165 :     RouteUpdate *rt_update = update->get();
     557      918163 :     monitor_->DequeueUpdate(rt_update);
     558             : 
     559      918273 :     if (rt_update->OnUpdateList()) {
     560             :         // Remove the route update from the update list and check if the list
     561             :         // can now be downgraded back to a route update.
     562         110 :         UpdateList *uplist = rt_update->GetUpdateList(ribout_);
     563         110 :         uplist->RemoveUpdate(rt_update);
     564         110 :         RouteUpdate *last_rt_update = uplist->MakeRouteUpdate();
     565             : 
     566             :         // If we were able to downgrade, set the DBEntry to point to the last
     567             :         // remaining route update and get rid of the current route update and
     568             :         // the update list.  Otherwise, just get rid of the route update.
     569         110 :         if (last_rt_update) {
     570         110 :             monitor_->SetEntryState(rt_update->route(), last_rt_update);
     571         110 :             update->release();
     572         110 :             delete rt_update;
     573         110 :             delete uplist;
     574             :         } else {
     575           0 :             update->release();
     576           0 :             delete rt_update;
     577             :         }
     578             :     } else {
     579             :         // Store the history from the route update or clear the state for the
     580             :         // DBEntry depending on whether we advertised the route.  In either
     581             :         // case, get rid of the route update.
     582      918150 :         if (rt_update->IsAdvertised()) {
     583      584078 :             StoreHistory(rt_update);
     584             :         } else {
     585      333997 :             ClearState(rt_update);
     586             :         }
     587      918262 :         update->release();
     588      918225 :         delete rt_update;
     589             :     }
     590      918235 : }
     591             : 
     592             : //
     593             : // Clear the advertised bits specified by isect from the target RibPeerSet in
     594             : // the UpdateInfo.  If the target is now empty, remove the UpdateInfo from the
     595             : // set container in the UpdateQueue.  Note that the UpdateInfo will still be
     596             : // on the SList in the RouteUpdate.
     597             : //
     598             : // Return true if the UpdateInfo was removed from the set container.
     599             : //
     600      924376 : bool RibOutUpdates::ClearAdvertisedBits(RouteUpdate *rt_update,
     601             :         UpdateInfo *uinfo, const RibPeerSet &isect, bool update_history) {
     602      924376 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     603             : 
     604      924175 :     if (update_history) {
     605      924063 :         rt_update->UpdateHistory(ribout_, &uinfo->roattr, isect);
     606             :     }
     607      924160 :     uinfo->target.Reset(isect);
     608      924094 :     bool empty = uinfo->target.empty();
     609      924078 :     if (empty) {
     610      922898 :         UpdateQueue *queue = queue_vec_[rt_update->queue_id()];
     611      922889 :         queue->AttrDequeue(uinfo);
     612             :     }
     613      924228 :     return empty;
     614             : }
     615             : 
     616             : //
     617             : // Concurrency: Called in the context of the bgp::SendUpdate task.
     618             : //
     619             : // Update the markers for all the peers in the blocked RibPeerSet.  In the
     620             : // general case we clear the blocked RibPeerSet from the UpdateMarker and
     621             : // create a new UpdateMarker for the blocked peers.
     622             : //
     623             : // Return true in the special case where all peers in the UpdateMarker have
     624             : // become blocked.
     625             : //
     626        1814 : bool RibOutUpdates::UpdateMarkersOnBlocked(UpdateMarker *marker,
     627             :         RouteUpdate *rt_update,
     628             :         const RibPeerSet *blocked) {
     629        1814 :     CHECK_CONCURRENCY("bgp::SendUpdate");
     630             : 
     631        1813 :     assert(!blocked->empty());
     632        1813 :     int queue_id = rt_update->queue_id();
     633        1813 :     UpdateQueue *queue = queue_vec_[queue_id];
     634             : 
     635             :     // If all the peers in the UpdateMarker are blocked, we simply move the
     636             :     // marker after the RouteUpdate.
     637        1813 :     if (marker->members == *blocked) {
     638        1121 :         stats_[queue_id].marker_move_count_++;
     639        1121 :         queue->MoveMarker(marker, rt_update);
     640        1122 :         return true;
     641             :     }
     642             : 
     643             :     // Reset bits in the specified UpdateMarker, create a new one for the
     644             :     // blocked peers and insert the new one after the RouteUpdate.
     645         692 :     marker->members.Reset(*blocked);
     646         692 :     assert(!marker->members.empty());
     647         692 :     UpdateMarker *new_marker = new UpdateMarker();
     648         692 :     new_marker->members = *blocked;
     649         692 :     stats_[queue_id].marker_split_count_++;
     650         692 :     queue->AddMarker(new_marker, rt_update);
     651             : 
     652         692 :     return false;
     653             : }
     654             : 
     655           8 : bool RibOutUpdates::Empty() const {
     656          24 :     for (int i = 0; i < RibOutUpdates::QCOUNT; ++i) {
     657          16 :         UpdateQueue *queue = queue_vec_[i];
     658          16 :         if (!queue->empty()) {
     659           0 :             return false;
     660             :         }
     661             :     }
     662           8 :     return true;
     663             : }
     664             : 
     665       15230 : size_t RibOutUpdates::queue_size(int queue_id) const {
     666       15230 :     const UpdateQueue *queue = queue_vec_[queue_id];
     667       15230 :     return queue->size();
     668             : }
     669             : 
     670       15230 : size_t RibOutUpdates::queue_marker_count(int queue_id) const {
     671       15230 :     const UpdateQueue *queue = queue_vec_[queue_id];
     672       15230 :     return queue->marker_count();
     673             : }
     674             : 
     675      735642 : bool RibOutUpdates::QueueJoin(int queue_id, int bit) {
     676      735642 :     UpdateQueue *queue = queue_vec_[queue_id];
     677      735642 :     return queue->Join(bit);
     678             : }
     679             : 
     680      735642 : void RibOutUpdates::QueueLeave(int queue_id, int bit) {
     681      735642 :     UpdateQueue *queue = queue_vec_[queue_id];
     682      735642 :     queue->Leave(bit);
     683      735642 : }
     684             : 
     685             : //
     686             : // Add statistics information to the provided Stats structure.
     687             : //
     688          16 : void RibOutUpdates::AddStatisticsInfo(int queue_id, Stats *stats) const {
     689          16 :     stats->messages_built_count_ += stats_[queue_id].messages_built_count_;
     690          16 :     stats->messages_sent_count_  += stats_[queue_id].messages_sent_count_;
     691          16 :     stats->reach_count_          += stats_[queue_id].reach_count_;
     692          16 :     stats->unreach_count_        += stats_[queue_id].unreach_count_;
     693          16 :     stats->tail_dequeue_count_   += stats_[queue_id].tail_dequeue_count_;
     694          16 :     stats->peer_dequeue_count_   += stats_[queue_id].peer_dequeue_count_;
     695          16 :     stats->marker_split_count_   += stats_[queue_id].marker_split_count_;
     696          16 :     stats->marker_merge_count_   += stats_[queue_id].marker_merge_count_;
     697          16 :     stats->marker_move_count_    += stats_[queue_id].marker_move_count_;
     698          16 : }

Generated by: LCOV version 1.14