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_update_sender.h"
7 :
8 : #include <boost/bind/bind.hpp>
9 : #include <boost/foreach.hpp>
10 :
11 : #include <map>
12 : #include <string>
13 : #include <atomic>
14 :
15 : #include "base/task_annotations.h"
16 : #include "bgp/ipeer.h"
17 : #include "bgp/bgp_ribout.h"
18 : #include "bgp/bgp_ribout_updates.h"
19 : #include "db/db.h"
20 :
21 : using std::unique_ptr;
22 : using std::make_pair;
23 : using std::map;
24 : using std::string;
25 : using std::vector;
26 : using namespace boost::placeholders;
27 :
28 : //
29 : // This struct represents RibOut specific state for a PeerState. There's one
30 : // instance of this for each RibOut that an IPeerUpdate has joined.
31 : //
32 : // The PeerRibState contains a bit mask to keep track of the QueueIds that are
33 : // currently active for the RibOut for the IPeerUpdate.
34 : //
35 : struct BgpSenderPartition::PeerRibState {
36 367821 : PeerRibState() : qactive(0) { }
37 : uint8_t qactive;
38 : };
39 :
40 : //
41 : // This nested class represents IPeerUpdate related state that's specific to
42 : // the BgpSenderPartition.
43 : //
44 : // A PeerState contains a Map of the index for a RibState to a PeerRibState.
45 : // Each entry in the map logically represents the state of the peer for the
46 : // ribout.
47 : //
48 : // The Map is used in conjunction with the RibStateMap in BgpSenderPartition
49 : // to implement regular and circular iterator classes that are used to walk
50 : // through all the RibState entries for a peer.
51 : //
52 : // A PeerState maintains the in_sync and send_ready state for the IPeerUpdate.
53 : //
54 : // The PeerState is considered to be send_ready when the underlying socket is
55 : // is writable. Note that the send_ready state in the PeerState may be out of
56 : // date with the actual socket state because the socket could have got blocked
57 : // when writing from another partition. Hence IPeerUpdate::send_ready() is the
58 : // more authoritative source.
59 : //
60 : // The PeerState is considered to be in_sync if it's send_ready and the marker
61 : // IPeerUpdate the peer has merged with the tail marker for all QueueIds in
62 : // all RiBOuts for which the IPeerUpdate is subscribed.
63 : //
64 : // The PeerState keeps count of the number of active RibOuts for each QueueId.
65 : // A (RibOut, QueueId) pair is considered to be active if the PeerState isn't
66 : // send_ready and there's RouteUpdates for the pair.
67 : //
68 : class BgpSenderPartition::PeerState {
69 : public:
70 : typedef map<size_t, PeerRibState> Map;
71 :
72 : class iterator : public boost::iterator_facade<
73 : iterator, RibOut, boost::forward_traversal_tag> {
74 : public:
75 33584 : explicit iterator(const RibStateMap &indexmap, Map *map, size_t index)
76 33584 : : indexmap_(indexmap), map_(map), index_(index) {
77 33584 : }
78 : size_t index() const { return index_; }
79 11436 : RibState *rib_state() { return indexmap_.At(index_); }
80 : const PeerRibState &peer_rib_state() const { return (*map_)[index_]; }
81 :
82 : private:
83 : friend class boost::iterator_core_access;
84 11436 : void increment() {
85 11436 : Map::const_iterator loc = map_->upper_bound(index_);
86 11436 : if (loc == map_->end()) {
87 11074 : index_ = -1;
88 : } else {
89 362 : index_ = loc->first;
90 : }
91 11436 : }
92 22510 : bool equal(const iterator &rhs) const {
93 22510 : return index_ == rhs.index_;
94 : }
95 : RibOut &dereference() const;
96 :
97 : const RibStateMap &indexmap_;
98 : Map *map_;
99 : size_t index_;
100 : };
101 :
102 : class circular_iterator : public boost::iterator_facade<
103 : circular_iterator, RibOut, boost::forward_traversal_tag> {
104 : public:
105 1203 : explicit circular_iterator(const RibStateMap &indexmap, Map *map,
106 : int start, bool is_valid)
107 1203 : : indexmap_(indexmap), map_(map), index_(-1), match_(true) {
108 1203 : if (map_->empty()) {
109 0 : return;
110 : }
111 1203 : Map::const_iterator loc = map_->lower_bound(start);
112 1203 : if (loc == map_->end()) {
113 1060 : loc = map_->begin();
114 : }
115 1203 : index_ = loc->first;
116 1203 : if (is_valid) match_ = false;
117 : }
118 37 : int index() const { return index_; }
119 447 : RibState *rib_state() { return indexmap_.At(index_); }
120 490 : const PeerRibState &peer_rib_state() const { return (*map_)[index_]; }
121 :
122 : private:
123 : friend class boost::iterator_core_access;
124 453 : void increment() {
125 453 : match_ = true;
126 453 : assert(!map_->empty());
127 453 : Map::const_iterator loc = map_->upper_bound(index_);
128 453 : if (loc == map_->end()) {
129 338 : loc = map_->begin();
130 : }
131 453 : index_ = loc->first;
132 453 : }
133 828 : bool equal(const circular_iterator &rhs) const {
134 828 : return ((match_ == rhs.match_) && (index_ == rhs.index_));
135 : }
136 : RibOut &dereference() const;
137 :
138 : const RibStateMap &indexmap_;
139 : Map *map_;
140 : int index_;
141 : bool match_;
142 : };
143 :
144 81397 : explicit PeerState(IPeerUpdate *peer)
145 81397 : : key_(peer), index_(-1),
146 81397 : qactive_cnt_(RibOutUpdates::QCOUNT),
147 81397 : in_sync_(true), rib_iterator_(BitSet::npos) {
148 81397 : send_ready_ = true;
149 244191 : for (int i = 0; i < RibOutUpdates::QCOUNT; i++) {
150 162794 : qactive_cnt_[i] = 0;
151 : }
152 81397 : }
153 :
154 : void Add(RibState *rs);
155 :
156 : void Remove(RibState *rs);
157 :
158 576 : bool IsMember(size_t index) const {
159 576 : return rib_bitset_.test(index);
160 : }
161 :
162 11074 : iterator begin(const RibStateMap &indexmap) {
163 11074 : Map::const_iterator it = rib_set_.begin();
164 11074 : size_t index = (it != rib_set_.end() ? it->first : -1);
165 11074 : return iterator(indexmap, &rib_set_, index);
166 : }
167 22510 : iterator end(const RibStateMap &indexmap) {
168 22510 : return iterator(indexmap, &rib_set_, -1);
169 : }
170 :
171 375 : circular_iterator circular_begin(const RibStateMap &indexmap) {
172 375 : return circular_iterator(indexmap, &rib_set_, rib_iterator_, true);
173 : }
174 828 : circular_iterator circular_end(const RibStateMap &indexmap) {
175 828 : return circular_iterator(indexmap, &rib_set_, rib_iterator_, false);
176 : }
177 :
178 37 : void SetIteratorStart(size_t start) { rib_iterator_ = start; }
179 :
180 24407 : void SetQueueActive(size_t rib_index, int queue_id) {
181 24407 : CHECK_CONCURRENCY("bgp::SendUpdate");
182 24413 : Map::iterator loc = rib_set_.find(rib_index);
183 24399 : assert(loc != rib_set_.end());
184 24398 : if (!BitIsSet(loc->second.qactive, queue_id)) {
185 21752 : SetBit(loc->second.qactive, queue_id);
186 21750 : qactive_cnt_[queue_id]++;
187 : }
188 24399 : }
189 :
190 736052 : void SetQueueInactive(size_t rib_index, int queue_id) {
191 736052 : CHECK_CONCURRENCY("bgp::SendUpdate", "bgp::PeerMembership");
192 736052 : Map::iterator loc = rib_set_.find(rib_index);
193 736052 : assert(loc != rib_set_.end());
194 736052 : if (BitIsSet(loc->second.qactive, queue_id)) {
195 21786 : ClearBit(loc->second.qactive, queue_id);
196 21786 : qactive_cnt_[queue_id]--;
197 : }
198 736052 : }
199 :
200 0 : bool IsQueueActive(size_t rib_index, int queue_id) {
201 0 : CHECK_CONCURRENCY("bgp::SendUpdate");
202 0 : Map::iterator loc = rib_set_.find(rib_index);
203 0 : assert(loc != rib_set_.end());
204 0 : return BitIsSet(loc->second.qactive, queue_id);
205 : }
206 :
207 10703 : int QueueCount(int queue_id) { return qactive_cnt_[queue_id]; }
208 :
209 3782584 : IPeerUpdate *peer() const { return key_; }
210 81397 : void set_index(size_t index) { index_ = index; }
211 817459 : size_t index() const { return index_; }
212 :
213 1937988 : bool in_sync() const { return in_sync_; }
214 4188 : void clear_sync() { in_sync_ = false; }
215 : void SetSync();
216 :
217 19232 : bool send_ready() const { return send_ready_; }
218 4578 : void set_send_ready(bool toggle) { send_ready_ = toggle; }
219 :
220 367821 : bool empty() const { return rib_set_.empty(); }
221 :
222 420 : bool CheckInvariants() const {
223 1260 : for (int i = 0; i < RibOutUpdates::QCOUNT; i++) {
224 840 : CHECK_INVARIANT(qactive_cnt_[i] <= (int) rib_set_.size());
225 : }
226 420 : return true;
227 : }
228 :
229 : private:
230 : IPeerUpdate *key_;
231 : size_t index_; // assigned from PeerStateMap
232 : Map rib_set_; // list of RibOuts advertised by the peer.
233 : BitSet rib_bitset_; // bitset of RibOuts advertised by the peer
234 : vector<int> qactive_cnt_;
235 : bool in_sync_; // whether the peer may dequeue tail markers.
236 : std::atomic<bool> send_ready_; // whether the peer may send updates.
237 : size_t rib_iterator_; // index of last processed rib.
238 :
239 : DISALLOW_COPY_AND_ASSIGN(PeerState);
240 : };
241 :
242 : //
243 : // This nested class represents RibOut related state that's specific to the
244 : // BgpSenderPartition.
245 : //
246 : // A RibState contains a BitSet of all the peers that are advertising the
247 : // RibOut associated with the RibState.
248 : //
249 : // The BitSet is used in conjunction with PeerStateMap in BgpSenderPartition
250 : // to implement an iterator that is used to walk through all the PeerState
251 : // entries for the ribout.
252 : //
253 : class BgpSenderPartition::RibState {
254 : public:
255 : class iterator : public boost::iterator_facade<
256 : iterator, PeerState, boost::forward_traversal_tag> {
257 : public:
258 3155932 : explicit iterator(const PeerStateMap &indexmap,
259 : const BitSet &set, size_t index)
260 3155932 : : indexmap_(indexmap), set_(set), index_(index) {
261 3155932 : }
262 : size_t index() const { return index_; }
263 :
264 : private:
265 : friend class boost::iterator_core_access;
266 1931469 : void increment() {
267 1931469 : index_ = set_.find_next(index_);
268 1931040 : }
269 2543432 : bool equal(const iterator &rhs) const {
270 2543432 : return index_ == rhs.index_;
271 : }
272 1931543 : PeerState &dereference() const {
273 1931543 : return *indexmap_.At(index_);
274 : }
275 : const PeerStateMap &indexmap_;
276 : const BitSet &set_;
277 : size_t index_;
278 : };
279 :
280 135724 : explicit RibState(RibOut *ribout)
281 135724 : : key_(ribout), index_(-1), in_sync_(RibOutUpdates::QCOUNT, true) {
282 135724 : }
283 :
284 : void Add(PeerState *ps);
285 : void Remove(PeerState *ps);
286 :
287 : bool QueueSync(int queue_id);
288 : void SetQueueSync(int queue_id);
289 : void SetQueueUnsync(int queue_id);
290 :
291 520 : RibOut *ribout() { return key_; }
292 :
293 612652 : iterator begin(const PeerStateMap &indexmap) {
294 612652 : return iterator(indexmap, peer_set_, peer_set_.find_first());
295 : }
296 :
297 2543487 : iterator end(const PeerStateMap &indexmap) {
298 2543487 : return iterator(indexmap, peer_set_, BitSet::npos);
299 : }
300 :
301 135724 : void set_index(size_t index) { index_ = index; }
302 2367610 : size_t index() const { return index_; }
303 :
304 367821 : bool empty() const { return peer_set_.none(); }
305 :
306 576 : const BitSet &peer_set() const { return peer_set_; }
307 :
308 : private:
309 : RibOut *key_;
310 : size_t index_;
311 : BitSet peer_set_;
312 : vector<bool> in_sync_;
313 :
314 : DISALLOW_COPY_AND_ASSIGN(RibState);
315 : };
316 :
317 :
318 367821 : void BgpSenderPartition::RibState::Add(BgpSenderPartition::PeerState *ps) {
319 367821 : CHECK_CONCURRENCY("bgp::PeerMembership");
320 367821 : peer_set_.set(ps->index());
321 367821 : }
322 :
323 367821 : void BgpSenderPartition::RibState::Remove(BgpSenderPartition::PeerState *ps) {
324 367821 : CHECK_CONCURRENCY("bgp::PeerMembership");
325 367821 : peer_set_.reset(ps->index());
326 367821 : }
327 :
328 10860 : bool BgpSenderPartition::RibState::QueueSync(int queue_id) {
329 10860 : CHECK_CONCURRENCY("bgp::SendUpdate");
330 10860 : return (in_sync_[queue_id]);
331 : }
332 :
333 73 : void BgpSenderPartition::RibState::SetQueueSync(int queue_id) {
334 73 : CHECK_CONCURRENCY("bgp::SendUpdate");
335 73 : in_sync_[queue_id] = true;
336 73 : }
337 :
338 6451 : void BgpSenderPartition::RibState::SetQueueUnsync(int queue_id) {
339 6451 : CHECK_CONCURRENCY("bgp::SendUpdate");
340 6450 : in_sync_[queue_id] = false;
341 6454 : }
342 :
343 367821 : void BgpSenderPartition::PeerState::Add(RibState *rs) {
344 367821 : CHECK_CONCURRENCY("bgp::PeerMembership");
345 367821 : PeerRibState init;
346 367821 : rib_set_.insert(make_pair(rs->index(), init));
347 367821 : rib_bitset_.set(rs->index());
348 367821 : }
349 :
350 367821 : void BgpSenderPartition::PeerState::Remove(RibState *rs) {
351 367821 : CHECK_CONCURRENCY("bgp::PeerMembership");
352 1103463 : for (int queue_id = 0; queue_id < RibOutUpdates::QCOUNT; queue_id++) {
353 735642 : SetQueueInactive(rs->index(), queue_id);
354 : }
355 367821 : rib_set_.erase(rs->index());
356 367821 : rib_bitset_.reset(rs->index());
357 367821 : }
358 :
359 5327 : void BgpSenderPartition::PeerState::SetSync() {
360 5327 : CHECK_CONCURRENCY("bgp::SendUpdate");
361 10757 : for (Map::iterator it = rib_set_.begin(); it != rib_set_.end(); ++it) {
362 5430 : assert(it->second.qactive == 0);
363 : }
364 15981 : for (int i = 0; i < RibOutUpdates::QCOUNT; i++) {
365 10654 : assert(qactive_cnt_[i] == 0);
366 : }
367 5327 : in_sync_ = true;
368 5327 : }
369 :
370 73 : RibOut &BgpSenderPartition::PeerState::iterator::dereference() const {
371 73 : return *indexmap_.At(index_)->ribout();
372 : }
373 :
374 447 : RibOut &BgpSenderPartition::PeerState::circular_iterator::dereference() const {
375 447 : return *indexmap_.At(index_)->ribout();
376 : }
377 :
378 : class BgpSenderPartition::Worker : public Task {
379 : public:
380 218543 : explicit Worker(BgpSenderPartition *partition)
381 218543 : : Task(partition->task_id(), partition->index()),
382 218543 : partition_(partition) {
383 218532 : }
384 :
385 218634 : virtual bool Run() {
386 218634 : CHECK_CONCURRENCY("bgp::SendUpdate");
387 :
388 : while (true) {
389 831266 : unique_ptr<WorkBase> wentry = partition_->WorkDequeue();
390 831101 : if (!wentry.get())
391 218598 : break;
392 612503 : if (!wentry->valid)
393 1 : continue;
394 612521 : switch (wentry->type) {
395 612129 : case WorkBase::WRibOut: {
396 612129 : WorkRibOut *workrib = static_cast<WorkRibOut *>(wentry.get());
397 612102 : partition_->UpdateRibOut(workrib->ribout, workrib->queue_id);
398 612218 : break;
399 : }
400 337 : case WorkBase::WPeer: {
401 337 : WorkPeer *workpeer = static_cast<WorkPeer *>(wentry.get());
402 390 : partition_->UpdatePeer(workpeer->peer);
403 390 : break;
404 : }
405 : }
406 1443895 : }
407 :
408 218601 : return true;
409 : }
410 0 : string Description() const { return "BgpSenderPartition::Worker"; }
411 :
412 : private:
413 : BgpSenderPartition *partition_;
414 : };
415 :
416 38483 : BgpSenderPartition::BgpSenderPartition(BgpUpdateSender *sender, int index)
417 38483 : : sender_(sender),
418 38483 : index_(index),
419 38483 : running_(false),
420 38483 : disabled_(false),
421 38483 : worker_task_(NULL) {
422 38483 : }
423 :
424 38483 : BgpSenderPartition::~BgpSenderPartition() {
425 38483 : if (worker_task_) {
426 0 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
427 0 : scheduler->Cancel(worker_task_);
428 : }
429 38483 : assert(peer_state_imap_.empty());
430 38483 : assert(rib_state_imap_.empty());
431 38483 : }
432 :
433 218525 : int BgpSenderPartition::task_id() const {
434 218525 : return sender_->task_id();
435 : }
436 :
437 : //
438 : // Add the (RibOut, IPeerUpdate) combo to the BgpSenderPartition.
439 : // Find or create the corresponding RibState and PeerState and sets up the
440 : // cross-linkage.
441 : //
442 367821 : void BgpSenderPartition::Add(RibOut *ribout, IPeerUpdate *peer) {
443 367821 : CHECK_CONCURRENCY("bgp::PeerMembership");
444 :
445 367821 : RibState *rs = rib_state_imap_.Locate(ribout);
446 367821 : PeerState *ps = peer_state_imap_.Locate(peer);
447 367821 : rs->Add(ps);
448 367821 : ps->Add(rs);
449 367821 : }
450 :
451 : //
452 : // Remove the (RibOut, IPeerUpdate) combo from the BgpSenderPartition.
453 : // Decouple cross linkage between the corresponding RibState and PeerState and
454 : // get rid of the RibState and/or PeerState if they are no longer needed.
455 : //
456 367821 : void BgpSenderPartition::Remove(RibOut *ribout, IPeerUpdate *peer) {
457 367821 : CHECK_CONCURRENCY("bgp::PeerMembership");
458 :
459 367821 : RibState *rs = rib_state_imap_.Find(ribout);
460 367821 : PeerState *ps = peer_state_imap_.Find(peer);
461 367821 : assert(rs != NULL);
462 367821 : assert(ps != NULL);
463 367821 : rs->Remove(ps);
464 367821 : ps->Remove(rs);
465 367821 : if (rs->empty()) {
466 135724 : WorkRibOutInvalidate(ribout);
467 135724 : rib_state_imap_.Remove(ribout, rs->index());
468 : }
469 367821 : if (ps->empty()) {
470 81397 : WorkPeerInvalidate(peer);
471 81397 : peer_state_imap_.Remove(peer, ps->index());
472 : }
473 367821 : }
474 :
475 : //
476 : // Create and enqueue new WorkRibOut entry since the RibOut is now
477 : // active.
478 : //
479 611982 : void BgpSenderPartition::RibOutActive(RibOut *ribout, int queue_id) {
480 611982 : CHECK_CONCURRENCY("db::DBTable", "bgp::SendUpdate", "bgp::PeerMembership");
481 :
482 612094 : WorkRibOutEnqueue(ribout, queue_id);
483 612124 : }
484 :
485 : //
486 : // Mark an IPeerUpdate to be send ready.
487 : //
488 5147 : void BgpSenderPartition::PeerSendReady(IPeerUpdate *peer) {
489 5147 : CHECK_CONCURRENCY("bgp::SendReadyTask");
490 :
491 : // The IPeerUpdate may not be registered if it has not reached Established
492 : // state or it may already have been unregistered by the time we get around
493 : // to processing the notification.
494 5147 : PeerState *ps = peer_state_imap_.Find(peer);
495 5147 : if (!ps)
496 0 : return;
497 :
498 : // Nothing to do if the IPeerUpdate's already in that state.
499 5147 : if (ps->send_ready())
500 4757 : return;
501 :
502 : // Create and enqueue new WorkPeer entry.
503 390 : ps->set_send_ready(true);
504 390 : WorkPeerEnqueue(peer);
505 : }
506 :
507 : //
508 : // Return true if the IPeer is send ready.
509 : //
510 2905 : bool BgpSenderPartition::PeerIsSendReady(IPeerUpdate *peer) const {
511 2905 : CHECK_CONCURRENCY("bgp::PeerMembership", "bgp::ShowCommand");
512 :
513 2905 : PeerState *ps = peer_state_imap_.Find(peer);
514 2905 : return ps->send_ready();
515 : }
516 :
517 : //
518 : // Return true if the IPeer is registered.
519 : //
520 9329 : bool BgpSenderPartition::PeerIsRegistered(IPeerUpdate *peer) const {
521 9329 : CHECK_CONCURRENCY("bgp::PeerMembership", "bgp::ShowCommand");
522 :
523 9329 : return (peer_state_imap_.Find(peer) != NULL);
524 : }
525 :
526 : //
527 : // Return true if the IPeer is in sync.
528 : //
529 7122 : bool BgpSenderPartition::PeerInSync(IPeerUpdate *peer) const {
530 7122 : CHECK_CONCURRENCY("bgp::PeerMembership", "bgp::ShowCommand");
531 :
532 7122 : PeerState *ps = peer_state_imap_.Find(peer);
533 7122 : return (ps ? ps->in_sync() : false);
534 : }
535 :
536 : //
537 : // Create a Worker if warranted and enqueue it to the TaskScheduler.
538 : // Assumes that the caller holds the BgpSenderPartition mutex.
539 : //
540 612291 : void BgpSenderPartition::MaybeStartWorker() {
541 612291 : if (!running_ && !disabled_) {
542 218529 : worker_task_ = new Worker(this);
543 218531 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
544 218534 : scheduler->Enqueue(worker_task_);
545 218640 : running_ = true;
546 : }
547 612402 : }
548 :
549 : //
550 : // Dequeue the first WorkBase item from the work queue and return an
551 : // unique_ptr to it. Clear out Worker related state if the work queue
552 : // is empty.
553 : //
554 831235 : unique_ptr<BgpSenderPartition::WorkBase> BgpSenderPartition::WorkDequeue() {
555 831235 : CHECK_CONCURRENCY("bgp::SendUpdate");
556 :
557 831161 : unique_ptr<WorkBase> wentry;
558 831161 : if (work_queue_.empty()) {
559 218599 : worker_task_ = NULL;
560 218599 : running_ = false;
561 : } else {
562 612555 : wentry.reset(work_queue_.pop_front().release());
563 : }
564 831121 : return wentry;
565 0 : }
566 :
567 : //
568 : // Enqueue a WorkBase entry into the the work queue and start a new Worker
569 : // task if required.
570 : //
571 612474 : void BgpSenderPartition::WorkEnqueue(WorkBase *wentry) {
572 612474 : CHECK_CONCURRENCY("db::DBTable", "bgp::SendUpdate", "bgp::SendReadyTask",
573 : "bgp::PeerMembership");
574 :
575 612510 : work_queue_.push_back(wentry);
576 612285 : MaybeStartWorker();
577 612518 : }
578 :
579 : //
580 : // Disable or enable the worker.
581 : // For unit testing.
582 : //
583 34 : void BgpSenderPartition::set_disabled(bool disabled) {
584 34 : disabled_ = disabled;
585 34 : MaybeStartWorker();
586 34 : }
587 :
588 : //
589 : // Enqueue a WorkPeer to the work queue.
590 : //
591 390 : void BgpSenderPartition::WorkPeerEnqueue(IPeerUpdate *peer) {
592 390 : CHECK_CONCURRENCY("bgp::SendReadyTask");
593 :
594 390 : WorkBase *wentry = new WorkPeer(peer);
595 390 : WorkEnqueue(wentry);
596 390 : }
597 :
598 : //
599 : // Invalidate all WorkBases for the given IPeerUpdate.
600 : // Used when a IPeerUpdate is removed.
601 : //
602 81397 : void BgpSenderPartition::WorkPeerInvalidate(IPeerUpdate *peer) {
603 81397 : CHECK_CONCURRENCY("bgp::PeerMembership");
604 :
605 81397 : for (WorkQueue::iterator it = work_queue_.begin();
606 81397 : it != work_queue_.end(); ++it) {
607 0 : WorkBase *wentry = it.operator->();
608 0 : if (wentry->type != WorkBase::WPeer)
609 0 : continue;
610 0 : WorkPeer *wpeer = static_cast<WorkPeer *>(wentry);
611 0 : if (wpeer->peer != peer)
612 0 : continue;
613 0 : wpeer->valid = false;
614 : }
615 81397 : }
616 :
617 : //
618 : // Enqueue a WorkRibOut to the work queue.
619 : //
620 612066 : void BgpSenderPartition::WorkRibOutEnqueue(RibOut *ribout, int queue_id) {
621 612066 : CHECK_CONCURRENCY("db::DBTable", "bgp::SendUpdate", "bgp::PeerMembership");
622 :
623 612088 : WorkBase *wentry = new WorkRibOut(ribout, queue_id);
624 612099 : WorkEnqueue(wentry);
625 612128 : }
626 :
627 : //
628 : // Invalidate all WorkBases for the given RibOut.
629 : // Used when a RibOut is removed.
630 : //
631 135724 : void BgpSenderPartition::WorkRibOutInvalidate(RibOut *ribout) {
632 135724 : CHECK_CONCURRENCY("bgp::PeerMembership");
633 :
634 135724 : for (WorkQueue::iterator it = work_queue_.begin();
635 135957 : it != work_queue_.end(); ++it) {
636 233 : WorkBase *wentry = it.operator->();
637 233 : if (wentry->type != WorkBase::WRibOut)
638 0 : continue;
639 233 : WorkRibOut *wribout = static_cast<WorkRibOut *>(wentry);
640 233 : if (wribout->ribout != ribout)
641 232 : continue;
642 1 : wribout->valid = false;
643 : }
644 135724 : }
645 :
646 : //
647 : // Build the RibPeerSet of IPeers for the RibOut that are in sync. Note that
648 : // we need to use bit indices that are specific to the RibOut, not the ones
649 : // from the BgpSenderPartition.
650 : //
651 612431 : void BgpSenderPartition::BuildSyncBitSet(const RibOut *ribout, RibState *rs,
652 : RibPeerSet *msync) {
653 612431 : CHECK_CONCURRENCY("bgp::SendUpdate");
654 :
655 612528 : for (RibState::iterator it = rs->begin(peer_state_imap_);
656 2543417 : it != rs->end(peer_state_imap_); ++it) {
657 1931042 : PeerState *ps = it.operator->();
658 :
659 : // If the PeerState is in sync but the IPeerUpdate is not send ready
660 : // then update the sync and send ready state in the PeerState. Note
661 : // that the RibOut queue for the PeerState will get marked active via
662 : // the call the SetQueueActive from UpdateRibOut.
663 1930894 : if (ps->in_sync()) {
664 1891700 : if (ps->peer()->send_ready()) {
665 1891630 : int rix = ribout->GetPeerIndex(ps->peer());
666 1891247 : msync->set(rix);
667 : } else {
668 53 : ps->clear_sync();
669 53 : ps->set_send_ready(false);
670 : }
671 : }
672 : }
673 612400 : }
674 :
675 : //
676 : // Take the RibPeerSet of blocked IPeers and update the relevant PeerStates.
677 : // Note that bit indices in the RibPeerSet and are specific to the RibOut.
678 : //
679 613060 : void BgpSenderPartition::SetSendBlocked(const RibOut *ribout, RibState *rs,
680 : int queue_id, const RibPeerSet &blocked) {
681 613060 : CHECK_CONCURRENCY("bgp::SendUpdate");
682 :
683 617133 : for (size_t bit = blocked.find_first(); bit != RibPeerSet::npos;
684 4138 : bit = blocked.find_next(bit)) {
685 4138 : IPeerUpdate *peer = ribout->GetPeer(bit);
686 4138 : PeerState *ps = peer_state_imap_.Find(peer);
687 4135 : ps->SetQueueActive(rs->index(), queue_id);
688 4135 : ps->clear_sync();
689 4135 : ps->set_send_ready(false);
690 : }
691 612989 : }
692 :
693 : //
694 : // For unit testing only.
695 : // Take the RibPeerSet of blocked IPeers and update the relevant PeerStates.
696 : //
697 8 : void BgpSenderPartition::SetSendBlocked(RibOut *ribout,
698 : int queue_id, const RibPeerSet &blocked) {
699 8 : CHECK_CONCURRENCY("bgp::SendUpdate");
700 :
701 8 : RibState *rs = rib_state_imap_.Find(ribout);
702 8 : assert(rs);
703 8 : SetSendBlocked(ribout, rs, queue_id, blocked);
704 8 : }
705 :
706 : //
707 : // Concurrency: called from bgp send task.
708 : //
709 : // Take the RibPeerSet of unsync IPeers and update the relevant PeerStates.
710 : // Note that bit indices in the RibPeerSet and are specific to the RibOut.
711 : //
712 612536 : void BgpSenderPartition::SetQueueActive(const RibOut *ribout, RibState *rs,
713 : int queue_id, const RibPeerSet &munsync) {
714 612536 : CHECK_CONCURRENCY("bgp::SendUpdate");
715 :
716 632822 : for (size_t bit = munsync.find_first(); bit != RibPeerSet::npos;
717 20276 : bit = munsync.find_next(bit)) {
718 20284 : IPeerUpdate *peer = ribout->GetPeer(bit);
719 20282 : PeerState *ps = peer_state_imap_.Find(peer);
720 20275 : ps->SetQueueActive(rs->index(), queue_id);
721 : }
722 612534 : }
723 :
724 : //
725 : // Concurrency: called from bgp send task.
726 : //
727 : // Mark the PeerRibState corresponding to the given IPeerUpdate and RibOut
728 : // as active.
729 : //
730 : // Used by unit test code.
731 : //
732 0 : void BgpSenderPartition::SetQueueActive(RibOut *ribout, int queue_id,
733 : IPeerUpdate *peer) {
734 0 : CHECK_CONCURRENCY("bgp::SendUpdate");
735 :
736 0 : PeerState *ps = peer_state_imap_.Find(peer);
737 0 : RibState *rs = rib_state_imap_.Find(ribout);
738 0 : ps->SetQueueActive(rs->index(), queue_id);
739 0 : }
740 :
741 : //
742 : // Concurrency: called from bgp send task.
743 : //
744 : // Check if the queue corresponding to IPeerUpdate, Ribout and queue id is
745 : // active.
746 : //
747 : // Used by unit test code.
748 : //
749 0 : bool BgpSenderPartition::IsQueueActive(RibOut *ribout, int queue_id,
750 : IPeerUpdate *peer) {
751 0 : CHECK_CONCURRENCY("bgp::SendUpdate");
752 :
753 0 : PeerState *ps = peer_state_imap_.Find(peer);
754 0 : RibState *rs = rib_state_imap_.Find(ribout);
755 0 : return ps->IsQueueActive(rs->index(), queue_id);
756 : }
757 :
758 : //
759 : // Concurrency: called from bgp send task.
760 : //
761 : // Mark all the RibStates for the given peer and queue id as being in sync
762 : // and trigger a tail dequeue.
763 : //
764 10654 : void BgpSenderPartition::SetQueueSync(PeerState *ps, int queue_id) {
765 10654 : CHECK_CONCURRENCY("bgp::SendUpdate");
766 :
767 10654 : for (PeerState::iterator it = ps->begin(rib_state_imap_);
768 21514 : it != ps->end(rib_state_imap_); ++it) {
769 10860 : RibState *rs = it.rib_state();
770 10860 : if (!rs->QueueSync(queue_id)) {
771 73 : RibOut *ribout = it.operator->();
772 73 : RibOutActive(ribout, queue_id);
773 73 : rs->SetQueueSync(queue_id);
774 : }
775 : }
776 10654 : }
777 :
778 : //
779 : // Drain the queue until there are no more updates or all the members become
780 : // blocked.
781 : //
782 612515 : void BgpSenderPartition::UpdateRibOut(RibOut *ribout, int queue_id) {
783 612515 : CHECK_CONCURRENCY("bgp::SendUpdate");
784 :
785 612577 : RibOutUpdates *updates = ribout->updates(index_);
786 612584 : RibState *rs = rib_state_imap_.Find(ribout);
787 612439 : RibPeerSet msync;
788 :
789 : // Convert group in-sync list to rib specific bitset.
790 612457 : BuildSyncBitSet(ribout, rs, &msync);
791 :
792 : // Drain the queue till we can do no more.
793 612404 : RibPeerSet blocked, munsync;
794 612409 : bool done = updates->TailDequeue(queue_id, msync, &blocked, &munsync);
795 612542 : assert(msync.Contains(blocked));
796 :
797 : // Mark peers as send blocked.
798 612628 : SetSendBlocked(ribout, rs, queue_id, blocked);
799 :
800 : // Set the queue to be active for any unsync peers. If we don't do this,
801 : // we will forget to mark the (RibOut,QueueId) as active for these peers
802 : // since the blocked RibPeerSet does not contain peers that are already
803 : // out of sync. Note that the unsync peers would have been split from
804 : // the tail marker in TailDequeue.
805 612557 : SetQueueActive(ribout, rs, queue_id, munsync);
806 :
807 : // If all peers are blocked, mark the queue as unsync in the RibState. We
808 : // will trigger tail dequeue for the (RibOut,QueueId) when any peer that
809 : // is interested in the RibOut becomes in sync.
810 612542 : if (!done)
811 6451 : rs->SetQueueUnsync(queue_id);
812 612545 : }
813 :
814 : //
815 : // Go through all RibOuts for the IPeerUpdate and drain the given queue till it
816 : // is up-to date or it becomes blocked. If it's blocked, select the next RibOut
817 : // to be processed when the IPeerUpdate becomes send ready.
818 : //
819 : // Return false if the IPeerUpdate got blocked.
820 : //
821 375 : bool BgpSenderPartition::UpdatePeerQueue(IPeerUpdate *peer, PeerState *ps,
822 : int queue_id) {
823 375 : CHECK_CONCURRENCY("bgp::SendUpdate");
824 :
825 375 : for (PeerState::circular_iterator it = ps->circular_begin(rib_state_imap_);
826 828 : it != ps->circular_end(rib_state_imap_); ++it) {
827 : // Skip if this queue is not active in the PeerRibState.
828 490 : if (!BitIsSet(it.peer_rib_state().qactive, queue_id))
829 43 : continue;
830 :
831 : // Drain the queue till we can do no more.
832 447 : RibOut *ribout = it.operator->();
833 447 : RibOutUpdates *updates = ribout->updates(index_);
834 447 : RibPeerSet blocked;
835 447 : bool done = updates->PeerDequeue(queue_id, peer, &blocked);
836 :
837 : // Process blocked mask.
838 447 : RibState *rs = it.rib_state();
839 447 : SetSendBlocked(ribout, rs, queue_id, blocked);
840 :
841 : // If the peer is still send_ready, mark the queue as inactive for
842 : // the peer. Need to check send_ready because the return value of
843 : // PeerDequeue only tells that *some* peer was merged with the tail
844 : // marker.
845 : // If the peer got blocked, remember where to start next time and
846 : // stop processing. We don't want to continue processing for other
847 : // merged peers if the lead peer is blocked. Processing for other
848 : // peers will continue when their own WorkPeer items are processed.
849 447 : if (ps->send_ready()) {
850 410 : assert(done);
851 410 : ps->SetQueueInactive(rs->index(), queue_id);
852 : } else {
853 37 : ps->SetIteratorStart(it.index());
854 37 : return false;
855 : }
856 447 : }
857 :
858 338 : return true;
859 : }
860 :
861 : //
862 : // Drain the queue of all updates for this IPeerUpdate, until it is up-to date
863 : // or it becomes blocked.
864 : //
865 5369 : void BgpSenderPartition::UpdatePeer(IPeerUpdate *peer) {
866 5369 : CHECK_CONCURRENCY("bgp::SendUpdate");
867 :
868 : // Bail if the PeerState is not send ready.
869 5369 : PeerState *ps = peer_state_imap_.Find(peer);
870 5369 : if (!ps->send_ready()) {
871 5 : return;
872 : }
873 :
874 : // Update the PeerState and bail if the IPeerUpdate is not send ready.
875 : // This happens if the IPeerUpdate gets blocked while processing some
876 : // other partition.
877 5364 : if (!peer->send_ready()) {
878 0 : ps->set_send_ready(false);
879 0 : return;
880 : }
881 :
882 : // Go through all queues and drain them if there's anything on them.
883 16030 : for (int queue_id = RibOutUpdates::QCOUNT - 1; queue_id >= 0; --queue_id) {
884 10703 : if (ps->QueueCount(queue_id) == 0) {
885 10328 : continue;
886 : }
887 375 : if (!UpdatePeerQueue(peer, ps, queue_id)) {
888 37 : assert(!ps->send_ready());
889 37 : return;
890 : }
891 : }
892 :
893 : // Checking the return value of UpdatePeerQueue above is not sufficient as
894 : // that only tells us that *some* peer(s) got merged with the tail marker.
895 : // Need to make sure that the IPeerUpdate that we are processing is still
896 : // send ready.
897 5327 : if (!ps->send_ready()) {
898 0 : return;
899 : }
900 :
901 : // Mark the peer as being in sync across all tables.
902 5327 : ps->SetSync();
903 :
904 : // Mark all RibStates for the peer as being in sync. This triggers a tail
905 : // dequeue for the corresponding (RibOut, QueueId) if necessary. This in
906 : // turn ensures that we do not get stuck in the case where all peers get
907 : // blocked and then get back in sync.
908 15981 : for (int queue_id = RibOutUpdates::QCOUNT - 1; queue_id >= 0; --queue_id) {
909 10654 : SetQueueSync(ps, queue_id);
910 : }
911 : }
912 :
913 : //
914 : // Check invariants for the BgpSenderPartition.
915 : //
916 249 : bool BgpSenderPartition::CheckInvariants() const {
917 249 : int grp_peer_count = 0;
918 249 : int peer_grp_count = 0;
919 393 : for (size_t i = 0; i < rib_state_imap_.size(); i++) {
920 144 : if (!rib_state_imap_.bits().test(i)) {
921 0 : continue;
922 : }
923 144 : RibState *rs = rib_state_imap_.At(i);
924 144 : assert(rs != NULL);
925 144 : CHECK_INVARIANT(rs->index() == i);
926 144 : for (RibState::iterator it = rs->begin(peer_state_imap_);
927 720 : it != rs->end(peer_state_imap_); ++it) {
928 576 : PeerState *ps = it.operator->();
929 576 : CHECK_INVARIANT(ps->IsMember(i));
930 576 : grp_peer_count++;
931 : }
932 : }
933 669 : for (size_t i = 0; i < peer_state_imap_.size(); i++) {
934 420 : if (!peer_state_imap_.bits().test(i)) {
935 0 : continue;
936 : }
937 420 : PeerState *ps = peer_state_imap_.At(i);
938 420 : assert(ps != NULL);
939 420 : CHECK_INVARIANT(ps->index() == i);
940 420 : if (!ps->CheckInvariants()) {
941 0 : return false;
942 : }
943 420 : for (PeerState::iterator it = ps->begin(rib_state_imap_);
944 996 : it != ps->end(rib_state_imap_); ++it) {
945 576 : RibState *rs = it.rib_state();
946 576 : CHECK_INVARIANT(rs->peer_set().test(i));
947 576 : peer_grp_count++;
948 : }
949 : }
950 :
951 249 : CHECK_INVARIANT(grp_peer_count == peer_grp_count);
952 249 : return true;
953 : }
954 :
955 : //
956 : // Constructor for BgpUpdateSender.
957 : // Initialize send ready WorkQueue and allocate BgpSenderPartitions.
958 : //
959 9752 : BgpUpdateSender::BgpUpdateSender(BgpServer *server)
960 9752 : : server_(server),
961 9752 : task_id_(TaskScheduler::GetInstance()->GetTaskId("bgp::SendUpdate")),
962 9752 : send_ready_queue_(
963 : TaskScheduler::GetInstance()->GetTaskId("bgp::SendReadyTask"), 0,
964 9752 : boost::bind(&BgpUpdateSender::SendReadyCallback, this, _1)) {
965 48235 : for (int idx = 0; idx < DB::PartitionCount(); ++idx) {
966 38483 : partitions_.push_back(new BgpSenderPartition(this, idx));
967 : }
968 9752 : }
969 :
970 : //
971 : // Destructor for BgpUpdateSender.
972 : // Shutdown the WorkQueue and delete all BgpSenderPartitions.
973 : //
974 9752 : BgpUpdateSender::~BgpUpdateSender() {
975 9752 : send_ready_queue_.Shutdown(false);
976 9752 : STLDeleteValues(&partitions_);
977 9752 : }
978 :
979 : //
980 : // Handle the join of an IPeerUpdate to a RibOut.
981 : //
982 93135 : void BgpUpdateSender::Join(RibOut *ribout, IPeerUpdate *peer) {
983 93135 : CHECK_CONCURRENCY("bgp::PeerMembership");
984 :
985 828777 : BOOST_FOREACH(BgpSenderPartition *partition, partitions_) {
986 367821 : partition->Add(ribout, peer);
987 : }
988 93135 : }
989 :
990 : //
991 : // Handle the leave of an IPeerUpdate from a RibOut.
992 : //
993 93135 : void BgpUpdateSender::Leave(RibOut *ribout, IPeerUpdate *peer) {
994 93135 : CHECK_CONCURRENCY("bgp::PeerMembership");
995 :
996 828777 : BOOST_FOREACH(BgpSenderPartition *partition, partitions_) {
997 367821 : partition->Remove(ribout, peer);
998 : }
999 93135 : }
1000 :
1001 : //
1002 : // Inform the specified BgpSenderPartition that it needs to schedule a tail
1003 : // dequeue for the given RibOut queue.
1004 : //
1005 611875 : void BgpUpdateSender::RibOutActive(int index, RibOut *ribout, int queue_id) {
1006 611875 : CHECK_CONCURRENCY("db::DBTable", "bgp::PeerMembership");
1007 :
1008 611956 : partitions_[index]->RibOutActive(ribout, queue_id);
1009 612051 : }
1010 :
1011 : //
1012 : // Concurrency: called from arbitrary task.
1013 : //
1014 : // Enqueue the IPeerUpdate to the send ready processing work queue.
1015 : // The callback is invoked in the context of bgp::SendReadyTask.
1016 : //
1017 14 : void BgpUpdateSender::PeerSendReady(IPeerUpdate *peer) {
1018 14 : send_ready_queue_.Enqueue(peer);
1019 14 : }
1020 :
1021 : //
1022 : // Return true if the IPeer is registered.
1023 : //
1024 543 : bool BgpUpdateSender::PeerIsRegistered(IPeerUpdate *peer) const {
1025 655 : BOOST_FOREACH(BgpSenderPartition *partition, partitions_) {
1026 585 : if (partition->PeerIsRegistered(peer))
1027 529 : return true;
1028 : }
1029 14 : return false;
1030 : }
1031 :
1032 : //
1033 : // Return true if the IPeer is in sync.
1034 : //
1035 529 : bool BgpUpdateSender::PeerInSync(IPeerUpdate *peer) const {
1036 4761 : BOOST_FOREACH(BgpSenderPartition *partition, partitions_) {
1037 2116 : if (!partition->PeerInSync(peer))
1038 0 : return false;
1039 : }
1040 529 : return true;
1041 : }
1042 :
1043 : //
1044 : // Callback to handle send ready notification for IPeerUpdate. Processing it
1045 : // in the context of bgp::SendeReadyTask ensures that there are no concurrency
1046 : // issues w.r.t. the BgpSenderPartition working on the IPeerUpdate while we are
1047 : // processing the notification.
1048 : //
1049 14 : bool BgpUpdateSender::SendReadyCallback(IPeerUpdate *peer) {
1050 14 : CHECK_CONCURRENCY("bgp::SendReadyTask");
1051 :
1052 114 : BOOST_FOREACH(BgpSenderPartition *partition, partitions_) {
1053 50 : partition->PeerSendReady(peer);
1054 : }
1055 14 : return true;
1056 : }
1057 :
1058 : //
1059 : // Check invariants for the BgpUpdateSender.
1060 : //
1061 59 : bool BgpUpdateSender::CheckInvariants() const {
1062 531 : BOOST_FOREACH(BgpSenderPartition *partition, partitions_) {
1063 236 : if (!partition->CheckInvariants())
1064 0 : return false;
1065 : }
1066 59 : return true;
1067 : }
1068 :
1069 : //
1070 : // Disable all BgpSenderPartitions.
1071 : //
1072 17 : void BgpUpdateSender::DisableProcessing() {
1073 51 : BOOST_FOREACH(BgpSenderPartition *partition, partitions_) {
1074 17 : partition->set_disabled(true);
1075 : }
1076 17 : }
1077 :
1078 : //
1079 : // Enable all BgpSenderPartitions.
1080 : //
1081 17 : void BgpUpdateSender::EnableProcessing() {
1082 51 : BOOST_FOREACH(BgpSenderPartition *partition, partitions_) {
1083 17 : partition->set_disabled(false);
1084 : }
1085 17 : }
|
