Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2015 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : // ksync_tx_queue.h
       7             : //
       8             : // Implmentation of a shared transmit queue between agent and KSync.
       9             : //
      10             : // KSync i/o operations are done on KSync Netlink socket. Even if the socket is
      11             : // set to non-blocking mode, the KSync socket i/o call will block till VRouter
      12             : // completes message processing. So, netlink i/o is not done in agent context.
      13             : // All messages are enqueued to a transmit and messages and i/o is done thru
      14             : // the transmit queue. This ensures agent doesnt block due to ksync processing
      15             : //
      16             : // There are two implementations of transmit queue
      17             : //
      18             : // WorkQueue based
      19             : // ---------------
      20             : // WorkQueue based implementation is used only for UT cases. WorkQueue based
      21             : // implementation is not used in production enviroment due to performance
      22             : // reasons. When we have slow producer and fast consumer, WorkQueue will
      23             : // result in spawning of a "task" for every message. The task spawning happens
      24             : // in producer context. In case of flow processing, agent is slow producer and
      25             : // vrouter is fast consumer. As a result, using WorkQueue will result in many
      26             : // task spawns thereby introducing latencies.
      27             : //
      28             : // We retain WorkQueue based implementations since UT code depends on
      29             : // WaitForIdle() APIs to continue with validations. Using Event-FD based
      30             : // implementations cannot rely on WaitForIdle()
      31             : //
      32             : //
      33             : // Event-FD based
      34             : // --------------
      35             : // This implementation is based on event_fd. The producer will add the entry
      36             : // into a tbb::concurrent_queue and notify the consumer on an event_fd. The
      37             : // consumer will block with "read" on event_fd. On getting an event, it will
      38             : // drain the queue of all messages in a tight loop. Consumer blocks on "read"
      39             : // when there is no data in the queue. This is an efficient implementation of
      40             : // queue between agent and ksync
      41             : //
      42             : #ifndef controller_src_ksync_ksync_tx_queue_h
      43             : #define controller_src_ksync_ksync_tx_queue_h
      44             : 
      45             : #include <sys/eventfd.h>
      46             : #include <pthread.h>
      47             : #include <algorithm>
      48             : #include <atomic>
      49             : #include <vector>
      50             : #include <set>
      51             : 
      52             : #include <tbb/concurrent_queue.h>
      53             : class KSyncSock;
      54             : class IoContext;
      55             : 
      56             : class KSyncTxQueue {
      57             : public:
      58             :     typedef tbb::concurrent_queue<IoContext *> Queue;
      59             : 
      60             :     KSyncTxQueue(KSyncSock *sock);
      61             :     ~KSyncTxQueue();
      62             : 
      63             :     void Init(bool use_work_queue, const std::string &cpu_pin_policy);
      64             :     void Shutdown();
      65             :     bool Run();
      66             : 
      67           6 :     size_t enqueues() const { return enqueues_; }
      68          10 :     size_t dequeues() const { return dequeues_; }
      69           1 :     uint32_t write_events() const { return write_events_; }
      70           6 :     uint32_t read_events() const { return read_events_; }
      71           7 :     size_t queue_len() const { return queue_len_; }
      72           5 :     uint64_t busy_time() const { return busy_time_; }
      73           5 :     uint32_t max_queue_len() const { return max_queue_len_; }
      74           7 :     void set_measure_busy_time(bool val) const { measure_busy_time_ = val; }
      75           2 :     void ClearStats() const {
      76           2 :         max_queue_len_ = 0;
      77           2 :         enqueues_ = 0;
      78           2 :         dequeues_ = 0;
      79           2 :         busy_time_ = 0;
      80           2 :         read_events_ = 0;
      81           2 :     }
      82             : 
      83       11249 :     bool Enqueue(IoContext *io_context) {
      84       11249 :         return EnqueueInternal(io_context);
      85             :     }
      86             : 
      87             : private:
      88             :     bool EnqueueInternal(IoContext *io_context);
      89             : 
      90             :     WorkQueue<IoContext *> *work_queue_;
      91             :     int event_fd_;
      92             :     // CPU pinning policy for netlink task
      93             :     std::string cpu_pin_policy_;
      94             :     KSyncSock *sock_;
      95             :     Queue queue_;
      96             :     std::atomic<bool> shutdown_;
      97             :     pthread_t event_thread_;
      98             :     std::atomic<size_t> queue_len_;
      99             :     mutable size_t max_queue_len_;
     100             : 
     101             :     mutable size_t enqueues_;
     102             :     mutable size_t dequeues_;
     103             :     mutable size_t write_events_;
     104             :     mutable size_t read_events_;
     105             :     mutable uint64_t busy_time_;
     106             :     mutable bool measure_busy_time_;
     107             : 
     108             :     DISALLOW_COPY_AND_ASSIGN(KSyncTxQueue);
     109             : };
     110             : 
     111             : #endif  // controller_src_ksync_ksync_tx_queue_h