Introspect changes to debug flow queue performances

- Added knobs to show "busy-time" for a queue
- Added flow-misc-event queue. This queue is used in subsequent comits

Change-Id: I879ef9e0afe0d13c59fef9b42895fcb0b88de022
Partial-Bug: #1584210
Partial-Bug: #1584224

src/base/queue_task.h

@@ -24,6 +24,7 @@
 #include <tbb/spin_rw_mutex.h>
 
 #include <base/task.h>
+#include <base/time_util.h>
 
 // WaterMarkInfo
 typedef boost::function<void (size_t)> WaterMarkCallback;
@@ -80,6 +81,10 @@ class QueueTaskRunner : public Task {
             return queue_->RunnerDone();
         }
 
+        uint64_t start = 0;
+        if (queue_->measure_busy_time_)
+            start = ClockMonotonicUsec();
+
         QueueEntryT entry = QueueEntryT();
         size_t count = 0;
         while (queue_->Dequeue(&entry)) {
@@ -88,10 +93,15 @@ class QueueTaskRunner : public Task {
                 break;
             }
             if (++count == queue_->max_iterations_) {
+                if (start)
+                    queue_->add_busy_time(ClockMonotonicUsec() - start);
                 return queue_->RunnerDone();
             }
         }
 
+        if (start)
+            queue_->add_busy_time(ClockMonotonicUsec() - start);
+
         // Running is done if queue_ is empty
         // While notification is being run, its possible that more entries
         // are added into queue_
@@ -154,7 +164,9 @@ class WorkQueue {
         shutdown_scheduled_(false),
         delete_entries_on_shutdown_(true),
         task_starts_(0),
-        max_queue_len_(0) {
+        max_queue_len_(0),
+        busy_time_(0),
+        measure_busy_time_(false) {
         count_ = 0;
         hwater_index_ = -1;
         lwater_index_ = -1;
@@ -424,6 +436,17 @@ class WorkQueue {
 
     uint32_t task_starts() const { return task_starts_; }
     uint32_t max_queue_len() const { return max_queue_len_; }
+    bool measure_busy_time() const { return measure_busy_time_; }
+    void set_measure_busy_time(bool val) { measure_busy_time_ = val; }
+    uint64_t busy_time() const { return busy_time_; }
+    void add_busy_time(uint64_t t) { busy_time_ += t; }
+    void ClearStats() const {
+        max_queue_len_ = 0;
+        enqueues_ = 0;
+        dequeues_ = 0;
+        busy_time_ = 0;
+        task_starts_ = 0;
+    }
 private:
     // Returns true if pop is successful.
     bool DequeueInternal(QueueEntryT *entry) {
@@ -625,8 +648,8 @@ class WorkQueue {
     size_t on_entry_defer_count_;
     tbb::atomic<bool> disabled_;
     bool deleted_;
-    size_t enqueues_;
-    size_t dequeues_;
+    mutable size_t enqueues_;
+    mutable size_t dequeues_;
     size_t drops_;
     size_t max_iterations_;
     size_t size_;
@@ -642,8 +665,10 @@ class WorkQueue {
     mutable tbb::mutex water_mutex_;
     tbb::atomic<bool> hwater_mark_set_;
     tbb::atomic<bool> lwater_mark_set_;
-    uint32_t task_starts_;
-    uint32_t max_queue_len_;
+    mutable uint32_t task_starts_;
+    mutable uint32_t max_queue_len_;
+    mutable uint64_t busy_time_;
+    bool measure_busy_time_;
 
     friend class QueueTaskTest;
     friend class QueueTaskShutdownTest;

src/ksync/ksync_sock.cc

@@ -243,6 +243,13 @@ void KSyncSock::Init(bool use_work_queue) {
     shutdown_ = false;
 }
 
+void KSyncSock::SetMeasureQueueDelay(bool val) {
+    sock_->send_queue_.set_measure_busy_time(val);
+    for (int i = 0; i < IoContext::MAX_WORK_QUEUES; i++) {
+        receive_work_queue[i]->set_measure_busy_time(val);
+    }
+}
+
 void KSyncSock::Start(bool read_inline) {
     sock_->read_inline_ = read_inline;
     if (sock_->read_inline_) {

src/ksync/ksync_sock.h

@@ -276,6 +276,7 @@ class KSyncSock {
 
     uint32_t WaitTreeSize() const;
     void SetSeqno(uint32_t seq);
+    void SetMeasureQueueDelay(bool val);
 protected:
     static void Init(bool use_work_queue);
     static void SetSockTableEntry(KSyncSock *sock);

src/ksync/ksync_tx_queue.cc

@@ -38,7 +38,9 @@ KSyncTxQueue::KSyncTxQueue(KSyncSock *sock) :
     sock_(sock),
     enqueues_(0),
     dequeues_(0),
-    write_events_(0) {
+    write_events_(0),
+    read_events_(0),
+    busy_time_(0) {
     queue_len_ = 0;
     shutdown_ = false;
 }
@@ -116,6 +118,7 @@ bool KSyncTxQueue::Run() {
     while (1) {
         uint64_t u = 0;
         ssize_t num = 0;
+
         while (1) {
             num = read(event_fd_, &u, sizeof(u));
             if (num >= (int)sizeof(u)) {
@@ -127,6 +130,11 @@ bool KSyncTxQueue::Run() {
                 assert(0);
             }
         }
+        read_events_++;
+
+        uint64_t t1 = 0;
+        if (measure_busy_time_)
+            t1 = ClockMonotonicUsec();
         IoContext *io_context = NULL;
         while (queue_.try_pop(io_context)) {
             dequeues_++;
@@ -137,6 +145,9 @@ bool KSyncTxQueue::Run() {
         if (shutdown_) {
             break;
         }
+
+        if (t1)
+            busy_time_ += (ClockMonotonicUsec() - t1);
     }
     return true;
 }

src/ksync/ksync_tx_queue.h

@@ -65,8 +65,18 @@ class KSyncTxQueue {
     size_t enqueues() const { return enqueues_; }
     size_t dequeues() const { return dequeues_; }
     uint32_t write_events() const { return write_events_; }
+    uint32_t read_events() const { return read_events_; }
     size_t queue_len() const { return queue_len_; }
+    uint64_t busy_time() const { return busy_time_; }
     uint32_t max_queue_len() const { return max_queue_len_; }
+    void set_measure_busy_time(bool val) const { measure_busy_time_ = val; }
+    void ClearStats() const {
+        max_queue_len_ = 0;
+        enqueues_ = 0;
+        dequeues_ = 0;
+        busy_time_ = 0;
+        read_events_ = 0;
+    }
 
     bool Enqueue(IoContext *io_context) {
         return EnqueueInternal(io_context);
@@ -82,11 +92,14 @@ class KSyncTxQueue {
     tbb::atomic<bool> shutdown_;
     pthread_t event_thread_;
     tbb::atomic<size_t> queue_len_;
-    size_t max_queue_len_;
+    mutable size_t max_queue_len_;
 
-    size_t enqueues_;
-    size_t dequeues_;
-    size_t write_events_;
+    mutable size_t enqueues_;
+    mutable size_t dequeues_;
+    mutable size_t write_events_;
+    mutable size_t read_events_;
+    mutable uint64_t busy_time_;
+    mutable bool measure_busy_time_;
 
     DISALLOW_COPY_AND_ASSIGN(KSyncTxQueue);
 };

src/vnsw/agent/cmn/agent.cc

@@ -855,3 +855,6 @@ void Agent::TaskTrace(const char *file_name, uint32_t line_no,
                         description, delay, task->Description());
 }
 
+bool Agent::MeasureQueueDelay() {
+    return params_->measure_queue_delay();
+}

src/vnsw/agent/cmn/agent.h

@@ -994,6 +994,7 @@ class Agent {
         flow_stats_req_handler_ = req;
     }
 
+    bool MeasureQueueDelay();
     void TaskTrace(const char *file_name, uint32_t line_no, const Task *task,
                    const char *description, uint32_t delay);
 

src/vnsw/agent/contrail-vrouter-agent.conf

@@ -33,6 +33,9 @@
 # Number of tx-buffers on pkt0 interface
 # pkt0_tx_buffers=1000
 #
+# Measure delays in different queues
+# measure_queue_delay=0
+#
 # Local log file name
 log_file=/var/log/contrail/contrail-vrouter-agent.log
 

src/vnsw/agent/contrail/contrail_agent_init.cc

@@ -135,6 +135,7 @@ void ContrailAgentInit::InitDone() {
     PortIpcHandler pih(agent(), PortIpcHandler::kPortsDir,
                        !agent_param()->vrouter_on_host_dpdk());
     pih.ReloadAllPorts();
+    flow_stats_manager_->InitDone();
 }
 
 void ContrailAgentInit::ModulesShutdown() {

src/vnsw/agent/init/agent_param.cc

@@ -490,6 +490,10 @@ void AgentParam::ParseDefaultSection() {
                                     "DEFAULT.pkt0_tx_buffers")) {
         pkt0_tx_buffer_count_ = Agent::kPkt0TxBufferCount;
     }
+    if (!GetValueFromTree<bool>(measure_queue_delay_,
+                                "DEFAULT.measure_queue_delay")) {
+        measure_queue_delay_ = false;
+    }
 }
 
 void AgentParam::ParseTaskSection() {
@@ -753,6 +757,8 @@ void AgentParam::ParseDefaultSectionArguments
                           "DEFAULT.mirror_client_port");
     GetOptValue<uint32_t>(var_map, pkt0_tx_buffer_count_,
                           "DEFAULT.pkt0_tx_buffers");
+    GetOptValue<bool>(var_map, measure_queue_delay_,
+                      "DEFAULT.measure_queue_delay");
 }
 
 void AgentParam::ParseTaskSectionArguments
@@ -1302,6 +1308,7 @@ AgentParam::AgentParam(bool enable_flow_options,
         enable_service_options_(enable_service_options),
         agent_mode_(agent_mode), vhost_(),
         pkt0_tx_buffer_count_(Agent::kPkt0TxBufferCount),
+        measure_queue_delay_(false),
         agent_name_(), eth_port_(),
         eth_port_no_arp_(false), eth_port_encap_type_(),
         xmpp_instance_count_(),

src/vnsw/agent/init/agent_param.h

@@ -274,6 +274,7 @@ class AgentParam  {
     // pkt0 tx buffer
     uint32_t pkt0_tx_buffer_count() const { return pkt0_tx_buffer_count_; }
     void set_pkt0_tx_buffer_count(uint32_t val) { pkt0_tx_buffer_count_ = val; }
+    bool measure_queue_delay() const { return measure_queue_delay_; }
 protected:
     void set_hypervisor_mode(HypervisorMode m) { hypervisor_mode_ = m; }
     virtual void InitFromSystem();
@@ -399,6 +400,7 @@ class AgentParam  {
     PortInfo vhost_;
     // Number of tx-buffers on pkt0 device
     uint32_t pkt0_tx_buffer_count_;
+    bool measure_queue_delay_;
 
     std::string agent_name_;
     std::string eth_port_;