Merge "Flow mgmt optimization" into R3.0

src/vnsw/agent/pkt/flow_entry.cc

@@ -358,7 +358,8 @@ FlowEventLog::~FlowEventLog() {
 FlowEntry::FlowEntry(FlowTable *flow_table) :
     flow_table_(flow_table), flags_(0),
     tunnel_type_(TunnelType::INVALID),
-    fip_vmi_(AgentKey::ADD_DEL_CHANGE, nil_uuid(), "") {
+    fip_vmi_(AgentKey::ADD_DEL_CHANGE, nil_uuid(), ""),
+    flow_mgmt_request_(NULL), flow_mgmt_info_() {
     // ksync entry is set to NULL only on constructor and on flow delete
     // it should not have any other explicit set to NULL
     ksync_entry_ = NULL;
@@ -399,6 +400,8 @@ void FlowEntry::Reset() {
     pending_actions_.Reset();
     flow_retry_attempts_ = 0;
     is_flow_on_unresolved_list = false;
+    assert(flow_mgmt_request_ == NULL);
+    assert(flow_mgmt_info_.get() == NULL);
 }
 
 void FlowEntry::Reset(const FlowKey &k) {

src/vnsw/agent/pkt/flow_entry.h

@@ -38,6 +38,9 @@ class FlowEntry;
 struct FlowExportInfo;
 class FlowStatsCollector;
 class FlowToken;
+class FlowMgmtRequest;
+class FlowEntryInfo;
+typedef std::auto_ptr<FlowEntryInfo> FlowMgmtEntryInfoPtr;
 
 ////////////////////////////////////////////////////////////////////////////
 // This is helper struct to carry parameters of reverse-flow. When flow is
@@ -625,6 +628,16 @@ class FlowEntry {
     void ResetRetryCount(){ flow_retry_attempts_ = 0; }
     bool IsOnUnresolvedList(){ return is_flow_on_unresolved_list;}
     void SetUnResolvedList(bool added){ is_flow_on_unresolved_list = added;}
+
+    FlowMgmtRequest *flow_mgmt_request() const { return flow_mgmt_request_; }
+    void set_flow_mgmt_request(FlowMgmtRequest *req) {
+        flow_mgmt_request_ = req;
+    }
+
+    FlowEntryInfo *flow_mgmt_info() const { return flow_mgmt_info_.get(); }
+    void set_flow_mgmt_info(FlowEntryInfo *info) {
+        flow_mgmt_info_.reset(info);
+    }
 private:
     friend class FlowTable;
     friend class FlowEntryFreeList;
@@ -691,6 +704,15 @@ class FlowEntry {
     static SecurityGroupList default_sg_list_;
     uint8_t flow_retry_attempts_;
     bool is_flow_on_unresolved_list;
+    // flow_mgmt_request used for compressing events to flow-mgmt queue.
+    // flow_mgmt_request_ is set when flow is enqueued to flow-mgmt queue. No
+    // subsequent enqueues are done till this field is set. The request can be
+    // updated with new values to reflect latest state
+    FlowMgmtRequest *flow_mgmt_request_;
+
+    // Field used by flow-mgmt module. Its stored here to optimize flow-mgmt
+    // and avoid lookups
+    FlowMgmtEntryInfoPtr flow_mgmt_info_;
     // IMPORTANT: Remember to update Reset() routine if new fields are added
     // IMPORTANT: Remember to update Copy() routine if new fields are added
 };

src/vnsw/agent/pkt/flow_mgmt.cc

@@ -112,18 +112,30 @@ void FlowMgmtManager::SetAclFlowSandeshData(const AclDBEntry *acl,
 // Utility methods to enqueue events into work-queue
 /////////////////////////////////////////////////////////////////////////////
 void FlowMgmtManager::AddEvent(FlowEntry *flow) {
-    FlowEntryPtr flow_ptr(flow);
-    FlowMgmtRequestPtr req(new FlowMgmtRequest(FlowMgmtRequest::ADD_FLOW,
-                                               flow_ptr));
-    request_queue_.Enqueue(req);
+    // Check if there is a flow-mgmt request already pending
+    // Flow mgmt takes care of current state of flow. So, there is no need to
+    // enqueue duplicate requests
+    FlowMgmtRequest *req = flow->flow_mgmt_request();
+    if (req == NULL) {
+        req = new FlowMgmtRequest(FlowMgmtRequest::UPDATE_FLOW, flow);
+        flow->set_flow_mgmt_request(req);
+        request_queue_.Enqueue(FlowMgmtRequestPtr(req));
+    }
 }
 
 void FlowMgmtManager::DeleteEvent(FlowEntry *flow,
                                   const RevFlowDepParams &params) {
-    FlowEntryPtr flow_ptr(flow);
-    FlowMgmtRequestPtr req(new FlowMgmtRequest(FlowMgmtRequest::DELETE_FLOW,
-                                               flow_ptr, params));
-    request_queue_.Enqueue(req);
+    // Check if there is a flow-mgmt request already pending
+    // Flow mgmt takes care of current state of flow. So, there is no need to
+    // enqueue duplicate requests
+    FlowMgmtRequest *req = flow->flow_mgmt_request();
+    if (req == NULL) {
+        req = new FlowMgmtRequest(FlowMgmtRequest::UPDATE_FLOW, flow);
+        flow->set_flow_mgmt_request(req);
+        request_queue_.Enqueue(FlowMgmtRequestPtr(req));
+    }
+
+    req->set_params(params);
 }
 
 void FlowMgmtManager::FlowStatsUpdateEvent(FlowEntry *flow, uint32_t bytes,
@@ -137,9 +149,8 @@ void FlowMgmtManager::FlowStatsUpdateEvent(FlowEntry *flow, uint32_t bytes,
     if (agent_->tsn_enabled()) {
         return;
     }
-    FlowEntryPtr flow_ptr(flow);
     FlowMgmtRequestPtr req(new FlowMgmtRequest
-                           (FlowMgmtRequest::UPDATE_FLOW_STATS, flow_ptr,
+                           (FlowMgmtRequest::UPDATE_FLOW_STATS, flow,
                             bytes, packets, oflow_bytes));
     request_queue_.Enqueue(req);
 }
@@ -380,17 +391,31 @@ FlowMgmtManager::BgpAsAServiceRequestHandler(FlowMgmtRequest *req) {
 
 bool FlowMgmtManager::RequestHandler(FlowMgmtRequestPtr req) {
     switch (req->event()) {
-    case FlowMgmtRequest::ADD_FLOW: {
-        log_queue_.Enqueue(req);
-        //Handle the Add request for flow-mgmt
-        AddFlow(req->flow());
-        break;
-    }
+    case FlowMgmtRequest::UPDATE_FLOW: {
+        FlowEntry *flow = req->flow().get();
+        {
+            // Before processing event, set the request pointer in flow to
+            // NULL. This ensures flow-entry enqueues new request from now
+            // onwards
+            tbb::mutex::scoped_lock mutex(flow->mutex());
+            flow->set_flow_mgmt_request(NULL);
+        }
 
-    case FlowMgmtRequest::DELETE_FLOW: {
-        log_queue_.Enqueue(req);
-        //Handle the Delete request for flow-mgmt
-        DeleteFlow(req->flow(), req->params());
+        // Update flow-mgmt information based on flow-state
+        if (req->flow()->deleted() == false) {
+            FlowMgmtRequestPtr log_req(new FlowMgmtRequest
+                                       (FlowMgmtRequest::ADD_FLOW,
+                                        req->flow().get()));
+            log_queue_.Enqueue(log_req);
+            AddFlow(req->flow());
+
+        } else {
+            FlowMgmtRequestPtr log_req(new FlowMgmtRequest
+                                       (FlowMgmtRequest::DELETE_FLOW,
+                                        req->flow().get(), req->params()));
+            log_queue_.Enqueue(log_req);
+            DeleteFlow(req->flow(), req->params());
+        }
         break;
     }
 
@@ -533,7 +558,6 @@ void FlowMgmtManager::EnqueueUveDeleteEvent(const FlowEntry *flow) const {
 }
 
 void FlowMgmtManager::AddFlow(FlowEntryPtr &flow) {
-    // Trace the flow add/change
     FlowMgmtKeyTree new_tree;
     MakeFlowMgmtKeyTree(flow.get(), &new_tree);
 
@@ -644,33 +668,28 @@ void FlowMgmtManager::VnFlowCounters(const VnEntry *vn, uint32_t *ingress_flow_c
                                       egress_flow_count);
 }
 
-FlowMgmtManager::FlowEntryInfo *
+FlowEntryInfo *
 FlowMgmtManager::FindFlowEntryInfo(const FlowEntryPtr &flow) {
-    FlowEntryTree::iterator it = flow_tree_.find(flow);
-    if (it == flow_tree_.end()) {
-        return NULL;
-    }
-
-    return &it->second;
+    return flow->flow_mgmt_info();
 }
 
-FlowMgmtManager::FlowEntryInfo *
+FlowEntryInfo *
 FlowMgmtManager::LocateFlowEntryInfo(FlowEntryPtr &flow) {
     FlowEntryInfo *info = FindFlowEntryInfo(flow);
     if (info != NULL)
         return info;
-
-    flow_tree_.insert(make_pair(flow, FlowEntryInfo()));
-    return FindFlowEntryInfo(flow);
+    info = new FlowEntryInfo(flow.get());
+    flow->set_flow_mgmt_info(info);
+    return info;
 }
 
 void FlowMgmtManager::DeleteFlowEntryInfo(FlowEntryPtr &flow) {
-    FlowEntryTree::iterator it = flow_tree_.find(flow);
-    if (it == flow_tree_.end())
+    FlowEntryInfo *info = flow->flow_mgmt_info();
+    if (info == NULL)
         return;
 
-    assert(it->second.tree_.size() == 0);
-    flow_tree_.erase(it);
+    assert(info->tree_.size() == 0);
+    flow->set_flow_mgmt_info(NULL);
     return;
 }
 

src/vnsw/agent/pkt/flow_mgmt.h

@@ -976,22 +976,31 @@ class BgpAsAServiceFlowMgmtTree : public FlowMgmtTree {
     DISALLOW_COPY_AND_ASSIGN(BgpAsAServiceFlowMgmtTree);
 };
 
+////////////////////////////////////////////////////////////////////////////
+// Per flow information stored in flow-mgmt module. Holds a reference to
+// flow so that flow active till flow-mgmt processing is done
+////////////////////////////////////////////////////////////////////////////
+class FlowEntryInfo {
+public:
+    FlowEntryInfo(FlowEntry *flow) :
+        flow_(flow), tree_(), count_(0), ingress_(false), local_flow_(false) {
+    }
+    virtual ~FlowEntryInfo() { assert(tree_.size() == 0); }
+private:
+    friend class FlowMgmtManager;
+    FlowEntryPtr flow_;
+    FlowMgmtKeyTree tree_;
+    uint32_t count_; // Number of times tree modified
+    bool ingress_;
+    bool local_flow_;
+    DISALLOW_COPY_AND_ASSIGN(FlowEntryInfo);
+};
 
 class FlowMgmtManager {
 public:
     static const std::string kFlowMgmtTask;
     typedef boost::shared_ptr<FlowMgmtRequest> FlowMgmtRequestPtr;
     typedef WorkQueue<FlowMgmtRequestPtr> FlowMgmtQueue;
-    struct FlowEntryInfo {
-        FlowMgmtKeyTree tree_;
-        uint32_t count_; // Number of times tree modified
-        bool ingress_;   // Ingress flow?
-        bool local_flow_;
-
-        FlowEntryInfo() : tree_(), count_(0), ingress_(false),
-            local_flow_(false) { }
-        virtual ~FlowEntryInfo() { assert(tree_.size() == 0); }
-    };
 
     // Comparator for FlowEntryPtr
     struct FlowEntryRefCmp {

src/vnsw/agent/pkt/flow_mgmt_request.h

@@ -16,6 +16,7 @@ class FlowMgmtRequest {
         INVALID,
         ADD_FLOW,
         DELETE_FLOW,
+        UPDATE_FLOW,
         ADD_DBENTRY,
         CHANGE_DBENTRY,
         DELETE_DBENTRY,
@@ -26,20 +27,20 @@ class FlowMgmtRequest {
 
     };
 
-    FlowMgmtRequest(Event event, FlowEntryPtr &flow) :
+    FlowMgmtRequest(Event event, FlowEntry *flow) :
         event_(event), flow_(flow), db_entry_(NULL), vrf_id_(0), gen_id_(),
         bytes_(), packets_(), oflow_bytes_(), params_() {
             if (event == RETRY_DELETE_VRF)
                 assert(vrf_id_);
     }
 
-    FlowMgmtRequest(Event event, FlowEntryPtr &flow,
+    FlowMgmtRequest(Event event, FlowEntry *flow,
                     const RevFlowDepParams &params) :
         event_(event), flow_(flow), db_entry_(NULL), vrf_id_(0), gen_id_(),
         bytes_(), packets_(), oflow_bytes_(), params_(params) {
     }
 
-    FlowMgmtRequest(Event event, FlowEntryPtr &flow, uint32_t bytes,
+    FlowMgmtRequest(Event event, FlowEntry *flow, uint32_t bytes,
                     uint32_t packets, uint32_t oflow_bytes) :
         event_(event), flow_(flow), db_entry_(NULL), vrf_id_(0), gen_id_(),
         bytes_(bytes), packets_(packets), oflow_bytes_(oflow_bytes), params_() {
@@ -105,6 +106,9 @@ class FlowMgmtRequest {
     uint32_t packets() const { return packets_;}
     uint32_t oflow_bytes() const { return oflow_bytes_;}
     const RevFlowDepParams& params() const { return params_; }
+    void set_params(const RevFlowDepParams &params) {
+        params_ = params;
+    }
 
 private:
     Event event_;

src/vnsw/agent/pkt/flow_table.cc

@@ -913,5 +913,6 @@ void FlowEntryFreeList::Free(FlowEntry *flow) {
     total_free_++;
     flow->Reset();
     free_list_.push_back(*flow);
+    assert(flow->flow_mgmt_info() == NULL);
     // TODO : Free entry if beyound threshold
 }