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             : #ifndef SRC_BGP_BGP_UPDATE_SENDER_H_
       7             : #define SRC_BGP_BGP_UPDATE_SENDER_H_
       8             : 
       9             : #include <boost/ptr_container/ptr_list.hpp>
      10             : 
      11             : #include <vector>
      12             : 
      13             : #include "base/bitset.h"
      14             : #include "base/index_map.h"
      15             : #include "base/queue_task.h"
      16             : 
      17             : class BgpServer;
      18             : class BgpUpdateSender;
      19             : class IPeerUpdate;
      20             : class RibOut;
      21             : class RibPeerSet;
      22             : 
      23             : //
      24             : // This class maintains state to generate updates for a DB partition for all
      25             : // RibOuts and IPeerUpdates in a BgpServer.  All BgpSenderPartitions work in
      26             : // parallel and maintain their own view of whether an IPeerUpdate is blocked,
      27             : // in sync etc.
      28             : //
      29             : // A BgpSenderPartition maintains two indexed maps for internal bookkeeping.
      30             : //
      31             : // o PeerStateMap allocates an index per IPeerUpdate to allow direct access to
      32             : //   the corresponding PeerState using the index.  The PeerState nested class
      33             : //   is described elsewhere.
      34             : // o RibStateMap allocates a bit index per RibOut and allows direct access to
      35             : //   the corresponding RibState using the index. The RibState nested class is
      36             : //   described elsewhere.
      37             : //
      38             : // BgpSenderPartition contains a WorkQueue of WorkBase entries to represent
      39             : // pending work. A WorkBase can either be a WorkRibOut, which corresponds to
      40             : // to a tail dequeue for a (RibOut, QueueId) or a WorkPeer which corresponds
      41             : // to a peer dequeue.
      42             : //
      43             : // A mutex is used to control access to the WorkQueue between producers that
      44             : // need to enqueue WorkBase entries and the Worker which dequeues the entries
      45             : // and processes them. The producers are the BgpExport class which creates a
      46             : // WorkRibOut entry after adding a RouteUpdate to an empty UpdateQueue, and
      47             : // IPeerUpdate class which creates a WorkPeer entry when it becomes unblocked.
      48             : //
      49             : class BgpSenderPartition {
      50             : public:
      51             :     BgpSenderPartition(BgpUpdateSender *sender, int index);
      52             :     ~BgpSenderPartition();
      53             : 
      54             :     void Add(RibOut *ribout, IPeerUpdate *peer);
      55             :     void Remove(RibOut *ribout, IPeerUpdate *peer);
      56             : 
      57             :     void RibOutActive(RibOut *ribout, int queue_id);
      58             : 
      59             :     void PeerSendReady(IPeerUpdate *peer);
      60             :     bool PeerIsSendReady(IPeerUpdate *peer) const;
      61             :     bool PeerIsRegistered(IPeerUpdate *peer) const;
      62             :     bool PeerInSync(IPeerUpdate *peer) const;
      63             : 
      64             :     bool CheckInvariants() const;
      65             : 
      66             :     int task_id() const;
      67      218668 :     int index() const { return index_; }
      68             : 
      69             :     // For unit testing.
      70             :     void set_disabled(bool disabled);
      71             : 
      72             : private:
      73             :     friend class BgpUpdateSenderTest;
      74             :     friend class RibOutUpdatesTest;
      75             : 
      76             :     struct WorkBase {
      77             :         enum Type {
      78             :             WPeer,
      79             :             WRibOut
      80             :         };
      81      612483 :         explicit WorkBase(Type type)
      82      612483 :             : type(type), valid(true) {
      83      612483 :         }
      84             :         Type type;
      85             :         bool valid;
      86             :     };
      87             : 
      88             :     struct WorkRibOut : public WorkBase {
      89      612041 :         WorkRibOut(RibOut *ribout, int queue_id)
      90      612041 :             : WorkBase(WRibOut), ribout(ribout), queue_id(queue_id) {
      91      612100 :         }
      92             :         RibOut *ribout;
      93             :         int queue_id;
      94             :     };
      95             : 
      96             :     struct WorkPeer : public WorkBase {
      97         390 :         explicit WorkPeer(IPeerUpdate *peer)
      98         390 :             : WorkBase(WPeer), peer(peer) {
      99         390 :         }
     100             :         IPeerUpdate *peer;
     101             :     };
     102             : 
     103             :     class PeerState;
     104             :     struct PeerRibState;
     105             :     class RibState;
     106             :     class Worker;
     107             : 
     108             :     typedef boost::ptr_list<WorkBase> WorkQueue;
     109             :     typedef IndexMap<IPeerUpdate *, PeerState> PeerStateMap;
     110             :     typedef IndexMap<RibOut *, RibState> RibStateMap;
     111             : 
     112             :     void MaybeStartWorker();
     113             :     std::unique_ptr<WorkBase> WorkDequeue();
     114             :     void WorkEnqueue(WorkBase *wentry);
     115             :     void WorkPeerEnqueue(IPeerUpdate *peer);
     116             :     void WorkPeerInvalidate(IPeerUpdate *peer);
     117             :     void WorkRibOutEnqueue(RibOut *ribout, int queue_id);
     118             :     void WorkRibOutInvalidate(RibOut *ribout);
     119             : 
     120             :     void UpdateRibOut(RibOut *ribout, int queue_id);
     121             :     void UpdatePeer(IPeerUpdate *peer);
     122             : 
     123             :     bool UpdatePeerQueue(IPeerUpdate *peer, PeerState *ps, int queue_id);
     124             : 
     125             :     void BuildSyncBitSet(const RibOut *ribout, RibState *rs, RibPeerSet *msync);
     126             : 
     127             :     void SetQueueActive(const RibOut *ribout, RibState *rs, int queue_id,
     128             :                         const RibPeerSet &munsync);
     129             :     void SetQueueActive(RibOut *ribout, int queue_id, IPeerUpdate *peer);
     130             :     bool IsQueueActive(RibOut *ribout, int queue_id, IPeerUpdate *peer);
     131             :     void SetSendBlocked(RibOut *ribout, int queue_id,
     132             :                         const RibPeerSet &blocked);
     133             :     void SetSendBlocked(const RibOut *ribout, RibState *rs, int queue_id,
     134             :                         const RibPeerSet &blocked);
     135             :     void SetQueueSync(PeerState *ps, int queue_id);
     136             : 
     137             :     BgpUpdateSender *sender_;
     138             :     int index_;
     139             :     bool running_;
     140             :     bool disabled_;
     141             :     WorkQueue work_queue_;
     142             :     Worker *worker_task_;
     143             :     PeerStateMap peer_state_imap_;
     144             :     RibStateMap rib_state_imap_;
     145             : 
     146             :     DISALLOW_COPY_AND_ASSIGN(BgpSenderPartition);
     147             : };
     148             : 
     149             : //
     150             : // This is a wrapper that hides the existence of multiple BgpSenderPartitions
     151             : // from client classes.  It relays APIs to appropriate/all BgpSenderPartition
     152             : // instance(s).
     153             : //
     154             : // The send ready WorkQueue is needed to process send ready notifications for
     155             : // IPeerUpdates in the context of bgp::SendReadyTask. This ensures that there
     156             : // are no concurrency issues in case the IPeerUpdate gets unblocked while we
     157             : // are still processing the previous WorkBase which caused it to get blocked
     158             : // in the first place.
     159             : //
     160             : class BgpUpdateSender {
     161             : public:
     162             :     explicit BgpUpdateSender(BgpServer *server);
     163             :     ~BgpUpdateSender();
     164             : 
     165             :     void Join(RibOut *ribout, IPeerUpdate *peer);
     166             :     void Leave(RibOut *ribout, IPeerUpdate *peer);
     167             : 
     168             :     void RibOutActive(int index, RibOut *ribout, int queue_id);
     169             :     void PeerSendReady(IPeerUpdate *peer);
     170             :     bool PeerIsRegistered(IPeerUpdate *peer) const;
     171             :     bool PeerInSync(IPeerUpdate *peer) const;
     172             : 
     173      218525 :     int task_id() const { return task_id_; }
     174             :     bool CheckInvariants() const;
     175             : 
     176             :     // For unit testing.
     177             :     void DisableProcessing();
     178             :     void EnableProcessing();
     179             : 
     180             : private:
     181             :     friend class BgpTestPeer;
     182             :     friend class BgpUpdateSenderTest;
     183             :     friend class RibOutUpdatesTest;
     184             : 
     185             :     bool SendReadyCallback(IPeerUpdate *peer);
     186          99 :     BgpSenderPartition *partition(int index) { return partitions_[index]; }
     187             : 
     188             :     BgpServer *server_;
     189             :     int task_id_;
     190             :     std::vector<BgpSenderPartition *> partitions_;
     191             :     WorkQueue<IPeerUpdate *> send_ready_queue_;
     192             : 
     193             :     DISALLOW_COPY_AND_ASSIGN(BgpUpdateSender);
     194             : };
     195             : 
     196             : #endif  // SRC_BGP_BGP_UPDATE_SENDER_H_