Merge "don't access ksync objects in ksyncsockUdsRead context"

Juniper · Jun 25, 2018 · a476ce8 · a476ce8
2 parents 8f2ba16 + a621719
commit a476ce8
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 9 deletions.
diff --git a/src/ksync/ksync_entry.h b/src/ksync/ksync_entry.h
@@ -56,7 +56,8 @@ class KSyncEntry {
         DEL_ADD_REQ,
         DEL_ACK,
         RE_EVAL,
-        INT_PTR_REL
+        INT_PTR_REL,
+        INVALID
     };
 
     std::string StateString() const;

diff --git a/src/ksync/ksync_object.cc b/src/ksync/ksync_object.cc
@@ -659,8 +659,10 @@ std::string KSyncEntry::EventString(KSyncEvent event) const {
     case INT_PTR_REL:
         str << "Reference release";
         break;
+    case INVALID:
+        str << "Invalid";
+        break;
     }
-
     str << '(' << event << ')';
     return str.str();
 }

diff --git a/src/ksync/ksync_sock.cc b/src/ksync/ksync_sock.cc
@@ -173,7 +173,9 @@ KSyncSock::KSyncSock() :
     rx_buff_(NULL), read_inline_(true), bulk_msg_context_(NULL),
     use_wait_tree_(true), process_data_inline_(false),
     ksync_bulk_sandesh_context_(), uve_bulk_sandesh_context_(),
-    tx_count_(0), ack_count_(0), err_count_(0) {
+    tx_count_(0), ack_count_(0), err_count_(0), 
+    rx_process_queue_(TaskScheduler::GetInstance()->GetTaskId("Agent::KSync"), 0,
+                    boost::bind(&KSyncSock::ProcessRxData, this, _1)) {
     TaskScheduler *scheduler = TaskScheduler::GetInstance();
 
     uint32_t uve_task_id =
@@ -321,7 +323,15 @@ KSyncSock::KSyncReceiveQueue *KSyncSock::GetReceiveQueue(uint32_t seqno) {
     uint32_t instance = (seqno >> 1) % kRxWorkQueueCount;
     return GetReceiveQueue(type, instance);
 }
-
+void KSyncSock::EnqueueRxProcessData(KSyncEntry *entry, KSyncEntry::KSyncEvent event) {
+    rx_process_queue_.Enqueue(KSyncRxQueueData(entry, event));
+}
+bool KSyncSock::ProcessRxData(KSyncRxQueueData data) {
+    assert(data.event_ != KSyncEntry::INVALID);
+    KSyncObject *object = data.entry_->GetObject();
+    object->NetlinkAck(data.entry_, data.event_);
+    return true;
+}
 KSyncBulkSandeshContext *KSyncSock::GetBulkSandeshContext(uint32_t seqno) {
 
     uint32_t instance = (seqno >> 1) % kRxWorkQueueCount;
@@ -913,14 +923,12 @@ KSyncIoContext::KSyncIoContext(KSyncSock *sock, KSyncEntry *sync_entry,
     IoContext(msg, msg_len, 0,
               sock->GetAgentSandeshContext(sync_entry->GetTableIndex()),
               IoContext::IOC_KSYNC, sync_entry->GetTableIndex()),
-    entry_(sync_entry), event_(event) {
+    entry_(sync_entry), event_(event), sock_(sock) {
     SetSeqno(sock->AllocSeqNo(type(), index()));
 }
 
 void KSyncIoContext::Handler() {
-    if (KSyncObject *obj = entry_->GetObject()) {
-        obj->NetlinkAck(entry_, event_);
-    }
+    sock_->EnqueueRxProcessData(entry_, event_);
 }
 
 void KSyncIoContext::ErrorHandler(int err) {

diff --git a/src/ksync/ksync_sock.h b/src/ksync/ksync_sock.h
@@ -182,6 +182,7 @@ class  KSyncIoContext : public IoContext {
     KSyncEntry *entry_;
     KSyncEntry::KSyncEvent event_;
     AgentSandeshContext *agent_sandesh_ctx_;
+    KSyncSock *sock_;
 };
 
 /*
@@ -342,6 +343,16 @@ class KSyncSock {
         }
     };
     typedef WorkQueue<KSyncRxData> KSyncReceiveQueue;
+    // structure for ksyncrprocess Rx queue
+    struct KSyncRxQueueData {
+         KSyncEntry             *entry_;
+         KSyncEntry::KSyncEvent event_;
+         KSyncRxQueueData():entry_(NULL),event_(KSyncEntry::INVALID) {}
+         KSyncRxQueueData(KSyncEntry *entry, KSyncEntry::KSyncEvent event) :
+                 entry_(entry), event_(event) {
+         }
+     };
+    typedef WorkQueue<KSyncRxQueueData> KSyncRxWorkQueue;
 
     KSyncSock();
     virtual ~KSyncSock();
@@ -404,6 +415,8 @@ class KSyncSock {
     void SetMeasureQueueDelay(bool val);
     void reset_use_wait_tree() { use_wait_tree_ = false; }
     void set_process_data_inline() { process_data_inline_ = true; }
+    // API to enqueue ksync events to rx process work queue
+    void EnqueueRxProcessData(KSyncEntry *entry, KSyncEntry::KSyncEvent event);
 protected:
     static void Init(bool use_work_queue, const std::string &cpu_pin_policy);
     static void SetSockTableEntry(KSyncSock *sock);
@@ -459,6 +472,7 @@ class KSyncSock {
 
     bool ProcessKernelData(KSyncBulkSandeshContext *ksync_context,
                            const KSyncRxData &data);
+    bool ProcessRxData(KSyncRxQueueData data);
     bool SendAsyncImpl(IoContext *ioc);
     bool SendAsyncStart() {
         tbb::mutex::scoped_lock lock(mutex_);
@@ -482,7 +496,11 @@ class KSyncSock {
     int tx_count_;
     int ack_count_;
     int err_count_;
-
+
+    // IO context can defer ksync event processing 
+    // by defering them to this work queue, this queue gets 
+    // processed in Agent::KSync context
+    KSyncRxWorkQueue  rx_process_queue_;
     static std::auto_ptr<KSyncSock> sock_;
     static pid_t pid_;
     static int vnsw_netlink_family_id_;