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 : }
|
