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             : #include <atomic>
       7             : #include <string>
       8             : 
       9             : #include "base/os.h"
      10             : 
      11             : #if defined(__linux__)
      12             : #include <asm/types.h>
      13             : #include <linux/netlink.h>
      14             : #include <linux/rtnetlink.h>
      15             : #include <linux/genetlink.h>
      16             : #include <linux/sockios.h>
      17             : #endif
      18             : 
      19             : #include <sys/socket.h>
      20             : 
      21             : #include <boost/bind/bind.hpp>
      22             : 
      23             : #include <base/logging.h>
      24             : #include <db/db.h>
      25             : #include <db/db_entry.h>
      26             : #include <db/db_table.h>
      27             : #include <db/db_table_partition.h>
      28             : 
      29             : #include "ksync_index.h"
      30             : #include "ksync_entry.h"
      31             : #include "ksync_object.h"
      32             : #include "ksync_sock.h"
      33             : #include "ksync_sock_user.h"
      34             : #include "ksync_types.h"
      35             : 
      36             : #include "nl_util.h"
      37             : #include "udp_util.h"
      38             : #include "vr_genetlink.h"
      39             : #include "vr_types.h"
      40             : 
      41             : using namespace boost::asio;
      42             : using namespace boost::placeholders;
      43             : 
      44             : /* Note SO_RCVBUFFORCE is supported only for linux version 2.6.14 and above */
      45             : typedef boost::asio::detail::socket_option::integer<SOL_SOCKET,
      46             :         SO_RCVBUFFORCE> ReceiveBuffForceSize;
      47             : 
      48             : int KSyncSock::vnsw_netlink_family_id_;
      49             : AgentSandeshContext *KSyncSock::agent_sandesh_ctx_[kRxWorkQueueCount];
      50             : std::unique_ptr<KSyncSock> KSyncSock::sock_;
      51             : pid_t KSyncSock::pid_;
      52             : std::atomic<bool> KSyncSock::shutdown_;
      53             : 
      54             : // Name of task used in KSync Response work-queues
      55             : const char* IoContext::io_wq_names[IoContext::MAX_WORK_QUEUES] =
      56             :                                                 {
      57             :                                                     "Agent::Uve",
      58             :                                                     "Agent::KSync"
      59             :                                                 };
      60             : /////////////////////////////////////////////////////////////////////////////
      61             : // Netlink utilities
      62             : /////////////////////////////////////////////////////////////////////////////
      63           0 : uint32_t GetNetlinkSeqno(char *data) {
      64           0 :     struct nlmsghdr *nlh = (struct nlmsghdr *)data;
      65           0 :     return nlh->nlmsg_seq;
      66             : }
      67             : 
      68           0 : bool NetlinkMsgDone(char *data) {
      69           0 :     struct nlmsghdr *nlh = (struct nlmsghdr *)data;
      70           0 :     return ((nlh->nlmsg_flags & NLM_F_MULTI) != 0);
      71             : }
      72             : 
      73             : // Common validation for netlink messages
      74       11314 : bool ValidateNetlink(char *data) {
      75       11314 :     struct nlmsghdr *nlh = (struct nlmsghdr *)data;
      76       11314 :     if (nlh->nlmsg_type == NLMSG_ERROR) {
      77           0 :         LOG(ERROR, "Netlink error for seqno " << nlh->nlmsg_seq << " len "
      78             :             << nlh->nlmsg_len);
      79           0 :         assert(0);
      80             :         return false;
      81             :     }
      82             : 
      83       11314 :     if (nlh->nlmsg_len > KSyncSock::kBufLen) {
      84           0 :         LOG(ERROR, "Length of " << nlh->nlmsg_len << " is more than expected "
      85             :             "length of " << KSyncSock::kBufLen);
      86           0 :         assert(0);
      87             :         return false;
      88             :     }
      89             : 
      90       11314 :     if (nlh->nlmsg_type == NLMSG_DONE) {
      91          63 :         return true;
      92             :     }
      93             : 
      94             :     // Sanity checks for generic-netlink message
      95       11251 :     if (nlh->nlmsg_type != KSyncSock::GetNetlinkFamilyId()) {
      96           0 :         LOG(ERROR, "Netlink unknown message type : " << nlh->nlmsg_type);
      97           0 :         assert(0);
      98             :         return false;
      99             :     }
     100             : 
     101       11251 :     struct genlmsghdr *genlh = (struct genlmsghdr *) (data + NLMSG_HDRLEN);
     102       11251 :     if (genlh->cmd != SANDESH_REQUEST) {
     103           0 :         LOG(ERROR, "Unknown generic netlink cmd : " << genlh->cmd);
     104           0 :         assert(0);
     105             :         return false;
     106             :     }
     107             : 
     108       11251 :     struct nlattr * attr = (struct nlattr *)(data + NLMSG_HDRLEN
     109             :                                              + GENL_HDRLEN);
     110       11251 :     if (attr->nla_type != NL_ATTR_VR_MESSAGE_PROTOCOL) {
     111           0 :         LOG(ERROR, "Unknown generic netlink TLV type : " << attr->nla_type);
     112           0 :         assert(0);
     113             :         return false;
     114             :     }
     115             : 
     116       11251 :     return true;
     117             : }
     118             : 
     119       11314 : void GetNetlinkPayload(char *data, char **buf, uint32_t *buf_len) {
     120       11314 :     struct nlmsghdr *nlh = (struct nlmsghdr *)data;
     121       11314 :     int len = 0;
     122       11314 :     if (nlh->nlmsg_type == NLMSG_DONE) {
     123          63 :         len = NLMSG_HDRLEN;
     124             :     } else {
     125       11251 :         len = NLMSG_HDRLEN + GENL_HDRLEN + NLA_HDRLEN;
     126             :     }
     127             : 
     128       11314 :     *buf = data + len;
     129       11314 :     *buf_len = nlh->nlmsg_len - len;
     130       11314 : }
     131             : 
     132           2 : void InitNetlink(nl_client *client) {
     133           2 :     nl_init_generic_client_req(client, KSyncSock::GetNetlinkFamilyId());
     134             :     unsigned char *nl_buf;
     135             :     uint32_t nl_buf_len;
     136           2 :     assert(nl_build_header(client, &nl_buf, &nl_buf_len) >= 0);
     137           2 : }
     138             : 
     139           0 : void ResetNetlink(nl_client *client) {
     140             :     unsigned char *nl_buf;
     141             :     uint32_t nl_buf_len;
     142           0 :     client->cl_buf_offset = 0;
     143           0 :     nl_build_header(client, &nl_buf, &nl_buf_len);
     144           0 : }
     145             : 
     146           0 : void UpdateNetlink(nl_client *client, uint32_t len, uint32_t seq_no) {
     147           0 :     nl_update_header(client, len);
     148           0 :     struct nlmsghdr *nlh = (struct nlmsghdr *)client->cl_buf;
     149           0 :     nlh->nlmsg_pid = KSyncSock::GetPid();
     150           0 :     nlh->nlmsg_seq = seq_no;
     151           0 : }
     152             : 
     153       11314 : void DecodeSandeshMessages(char *buf, uint32_t buf_len, SandeshContext *sandesh_context,
     154             :                            uint32_t alignment) {
     155       22945 :     while (buf_len > (alignment - 1)) {
     156             :         int error;
     157       11631 :         int decode_len = Sandesh::ReceiveBinaryMsgOne((uint8_t *)buf, buf_len,
     158             :                                                       &error, sandesh_context);
     159       11631 :         if (decode_len < 0) {
     160           0 :             LOG(DEBUG, "Incorrect decode len " << decode_len);
     161           0 :             break;
     162             :         }
     163       11631 :         buf += decode_len;
     164       11631 :         buf_len -= decode_len;
     165             :     }
     166       11314 : }
     167             : 
     168             : /////////////////////////////////////////////////////////////////////////////
     169             : // KSyncSock routines
     170             : /////////////////////////////////////////////////////////////////////////////
     171           2 : KSyncSock::KSyncSock() :
     172           2 :     nl_client_(NULL), wait_tree_(), send_queue_(this),
     173           2 :     max_bulk_msg_count_(kMaxBulkMsgCount),
     174           2 :     bulk_seq_no_(kInvalidBulkSeqNo), bulk_buf_size_(0), bulk_msg_count_(0),
     175           2 :     rx_buff_(NULL), read_inline_(true), bulk_msg_context_(NULL),
     176           2 :     use_wait_tree_(true), process_data_inline_(false),
     177          10 :     ksync_bulk_sandesh_context_(), uve_bulk_sandesh_context_(),
     178           2 :     tx_count_(0), ack_count_(0), err_count_(0), 
     179           2 :     rx_process_queue_(TaskScheduler::GetInstance()->GetTaskId("Agent::KSync"), 0,
     180           2 :                     boost::bind(&KSyncSock::ProcessRxData, this, _1)) {
     181           2 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
     182             : 
     183             :     uint32_t uve_task_id =
     184           2 :         scheduler->GetTaskId(IoContext::io_wq_names[IoContext::IOC_UVE]);
     185             :     uint32_t ksync_task_id =
     186           2 :         scheduler->GetTaskId(IoContext::io_wq_names[IoContext::IOC_KSYNC]);
     187           6 :     for(uint32_t i = 0; i < kRxWorkQueueCount; i++) {
     188           4 :         ksync_rx_queue[i] = AllocQueue(ksync_bulk_sandesh_context_,
     189             :                                        ksync_task_id, i, "KSync Receive Queue");
     190           4 :         uve_rx_queue[i] = AllocQueue(uve_bulk_sandesh_context_,
     191             :                                      uve_task_id, i, "KSync UVE Receive Queue");
     192             :     }
     193             : 
     194           2 :     nl_client_ = (nl_client *)malloc(sizeof(nl_client));
     195           2 :     memset(nl_client_, 0, sizeof(nl_client));
     196           2 :     rx_buff_ = NULL;
     197           2 :     seqno_ = 0;
     198           2 :     uve_seqno_ = 0;
     199             : 
     200           2 :     memset(bulk_mctx_arr_, 0, sizeof(bulk_mctx_arr_));
     201           2 :     bmca_prod_ = bmca_cons_ = 0;
     202           2 : }
     203             : 
     204           0 : KSyncSock::~KSyncSock() {
     205           0 :     assert(wait_tree_.size() == 0);
     206             : 
     207           0 :     if (rx_buff_) {
     208           0 :         delete [] rx_buff_;
     209           0 :         rx_buff_ = NULL;
     210             :     }
     211             : 
     212           0 :     for(int i = 0; i < kRxWorkQueueCount; i++) {
     213           0 :         ksync_rx_queue[i]->Shutdown();
     214           0 :         delete ksync_rx_queue[i];
     215             : 
     216           0 :         uve_rx_queue[i]->Shutdown();
     217           0 :         delete uve_rx_queue[i];
     218             :     }
     219             : 
     220           0 :     if (nl_client_->cl_buf) {
     221           0 :         free(nl_client_->cl_buf);
     222             :     }
     223           0 :     free(nl_client_);
     224           0 : }
     225             : 
     226           2 : void KSyncSock::Shutdown() {
     227           2 :     shutdown_ = true;
     228           2 :     sock_->send_queue_.Shutdown();
     229           2 :     sock_.release();
     230           2 : }
     231             : 
     232           2 : void KSyncSock::Init(bool use_work_queue, const std::string &cpu_pin_policy) {
     233           2 :     sock_->send_queue_.Init(use_work_queue, cpu_pin_policy);
     234           2 :     pid_ = getpid();
     235           2 :     shutdown_ = false;
     236           2 : }
     237             : 
     238           8 : KSyncSock::KSyncReceiveQueue *KSyncSock::AllocQueue
     239             : (KSyncBulkSandeshContext ctxt[], uint32_t task_id, uint32_t instance,
     240             :  const char *name) {
     241             :     KSyncReceiveQueue *queue;
     242          16 :     queue = new KSyncReceiveQueue
     243             :         (task_id, instance, boost::bind(&KSyncSock::ProcessKernelData, this,
     244          16 :                                         &ctxt[instance], _1));
     245             :     char tmp[128];
     246           8 :     sprintf(tmp, "%s-%d", name, instance);
     247           8 :     queue->set_name(tmp);
     248           8 :     return queue;
     249             : }
     250             : 
     251           2 : void KSyncSock::SetMeasureQueueDelay(bool val) {
     252           2 :     sock_->send_queue_.set_measure_busy_time(val);
     253           6 :     for (int i = 0; i < kRxWorkQueueCount; i++) {
     254           4 :         ksync_rx_queue[i]->set_measure_busy_time(val);
     255             :     }
     256           2 : }
     257             : 
     258           2 : void KSyncSock::Start(bool read_inline) {
     259           2 :     sock_->read_inline_ = read_inline;
     260           2 :     if (sock_->read_inline_) {
     261           2 :         return;
     262             :     }
     263           0 :     sock_->rx_buff_ = new char[kBufLen];
     264           0 :     sock_->AsyncReceive(boost::asio::buffer(sock_->rx_buff_, kBufLen),
     265             :                         boost::bind(&KSyncSock::ReadHandler, sock_.get(),
     266             :                                     placeholders::error,
     267             :                                     placeholders::bytes_transferred));
     268             : }
     269             : 
     270           2 : void KSyncSock::SetSockTableEntry(KSyncSock *sock) {
     271           2 :     assert(sock_.get() == NULL);
     272           2 :     sock_.reset(sock);
     273           2 : }
     274             : 
     275           2 : void KSyncSock::SetNetlinkFamilyId(int id) {
     276           2 :     vnsw_netlink_family_id_ = id;
     277           2 :     InitNetlink(sock_->nl_client_);
     278           2 : }
     279             : 
     280           0 : uint32_t KSyncSock::WaitTreeSize() const {
     281           0 :     return wait_tree_.size();
     282             : }
     283             : 
     284           0 : void KSyncSock::SetSeqno(uint32_t seq) {
     285           0 :     seqno_ = seq;
     286           0 :     uve_seqno_ = seq;
     287           0 : }
     288             : 
     289       11249 : uint32_t KSyncSock::AllocSeqNo(IoContext::Type type, uint32_t instance) {
     290             :     uint32_t seq;
     291       11249 :     if (type == IoContext::IOC_UVE) {
     292           0 :         seq = uve_seqno_.fetch_add(1);
     293           0 :         seq = (seq * kRxWorkQueueCount + (instance % kRxWorkQueueCount)) << 1;
     294             :     } else {
     295       11249 :         seq = seqno_.fetch_add(1);
     296       11249 :         seq = (seq * kRxWorkQueueCount + (instance % kRxWorkQueueCount)) << 1;
     297       11249 :         seq |= KSYNC_DEFAULT_Q_ID_SEQ;
     298             :     }
     299       11249 :     if (seq == kInvalidBulkSeqNo) {
     300           0 :         return AllocSeqNo(type, instance);
     301             :     }
     302       11249 :     return seq;
     303             : }
     304             : 
     305           0 : uint32_t KSyncSock::AllocSeqNo(IoContext::Type type) {
     306           0 :     return AllocSeqNo(type, 0);
     307             : }
     308             : 
     309       11312 : KSyncSock::KSyncReceiveQueue *KSyncSock::GetReceiveQueue(IoContext::Type type,
     310             :                                                          uint32_t instance) {
     311       11312 :     if (type == IoContext::IOC_UVE) {
     312           0 :         return uve_rx_queue[instance % kRxWorkQueueCount];
     313             :     } else {
     314       11312 :         return ksync_rx_queue[instance % kRxWorkQueueCount];
     315             :     }
     316             : }
     317             : 
     318           0 : KSyncSock::KSyncReceiveQueue *KSyncSock::GetReceiveQueue(uint32_t seqno) {
     319             :     IoContext::Type type;
     320           0 :     if (seqno & KSYNC_DEFAULT_Q_ID_SEQ)
     321           0 :         type = IoContext::IOC_KSYNC;
     322             :     else
     323           0 :         type = IoContext::IOC_UVE;
     324             : 
     325           0 :     uint32_t instance = (seqno >> 1) % kRxWorkQueueCount;
     326           0 :     return GetReceiveQueue(type, instance);
     327             : }
     328       11249 : void KSyncSock::EnqueueRxProcessData(KSyncEntry *entry, KSyncEntry::KSyncEvent event) {
     329       11249 :     rx_process_queue_.Enqueue(KSyncRxQueueData(entry, event));
     330       11249 : }
     331       11249 : bool KSyncSock::ProcessRxData(KSyncRxQueueData data) {
     332       11249 :     assert(data.event_ != KSyncEntry::INVALID);
     333       11249 :     KSyncObject *object = data.entry_->GetObject();
     334       11249 :     object->NetlinkAck(data.entry_, data.event_);
     335       11249 :     return true;
     336             : }
     337           0 : KSyncBulkSandeshContext *KSyncSock::GetBulkSandeshContext(uint32_t seqno) {
     338             : 
     339           0 :     uint32_t instance = (seqno >> 1) % kRxWorkQueueCount;
     340           0 :     if (seqno & KSYNC_DEFAULT_Q_ID_SEQ)
     341           0 :         return &ksync_bulk_sandesh_context_[instance];
     342             :     else
     343           0 :         return &uve_bulk_sandesh_context_[instance];
     344             : }
     345             : 
     346           0 : KSyncSock *KSyncSock::Get(DBTablePartBase *partition) {
     347           0 :     return sock_.get();
     348             : }
     349             : 
     350       11258 : KSyncSock *KSyncSock::Get(int idx) {
     351       11258 :     assert(idx == 0);
     352       11258 :     return sock_.get();
     353             : }
     354             : 
     355       11312 : bool KSyncSock::ValidateAndEnqueue(char *data, KSyncBulkMsgContext *context) {
     356       11312 :     Validate(data);
     357             : 
     358             :     KSyncReceiveQueue *queue;
     359       11312 :     if (context) {
     360       11312 :         queue = GetReceiveQueue(context->io_context_type(),
     361             :                                 context->work_queue_index());
     362             :     } else {
     363           0 :         queue = GetReceiveQueue(GetSeqno(data));
     364             :     }
     365       11312 :     queue->Enqueue(KSyncRxData(data, context));
     366       11312 :     return true;
     367             : }
     368             : 
     369             : // Read handler registered with boost::asio. Demux done based on seqno_
     370           0 : void KSyncSock::ReadHandler(const boost::system::error_code& error,
     371             :                             size_t bytes_transferred) {
     372           0 :     if (error) {
     373           0 :         LOG(ERROR, "Error reading from Ksync sock. Error : " <<
     374             :             boost::system::system_error(error).what());
     375           0 :         if (shutdown_ == false) {
     376           0 :             assert(0);
     377             :         }
     378           0 :         return;
     379             :     }
     380             : 
     381           0 :     ValidateAndEnqueue(rx_buff_, NULL);
     382             : 
     383           0 :     rx_buff_ = new char[kBufLen];
     384           0 :     AsyncReceive(boost::asio::buffer(rx_buff_, kBufLen),
     385             :                  boost::bind(&KSyncSock::ReadHandler, this,
     386             :                              placeholders::error,
     387             :                              placeholders::bytes_transferred));
     388             : }
     389             : 
     390             : // Process kernel data - executes in the task specified by IoContext
     391             : // Currently only Agent::KSync and Agent::Uve are possibilities
     392       11312 : bool KSyncSock::ProcessKernelData(KSyncBulkSandeshContext *bulk_sandesh_context,
     393             :                                   const KSyncRxData &data) {
     394       11312 :     KSyncBulkMsgContext *bulk_message_context = data.bulk_msg_context_;
     395       11312 :     WaitTree::iterator it;
     396       11312 :     if (data.bulk_msg_context_ == NULL) {
     397           0 :         uint32_t seqno = GetSeqno(data.buff_);
     398             :         {
     399           0 :             std::scoped_lock lock(mutex_);
     400           0 :             it = wait_tree_.find(seqno);
     401           0 :         }
     402           0 :         if (it == wait_tree_.end()) {
     403           0 :             LOG(ERROR, "KSync error in finding for sequence number : "
     404             :                 << seqno);
     405           0 :             assert(0);
     406             :         }
     407           0 :         bulk_message_context = &(it->second);
     408             :     }
     409             : 
     410       11312 :     bulk_sandesh_context->set_bulk_message_context(bulk_message_context);
     411       11312 :     BulkDecoder(data.buff_, bulk_sandesh_context);
     412             :     // Remove the IoContext only on last netlink message
     413       11312 :     if (IsMoreData(data.buff_) == false) {
     414       11112 :         if (data.bulk_msg_context_ != NULL) {
     415       11112 :             delete data.bulk_msg_context_;
     416             :         } else {
     417           0 :             std::scoped_lock lock(mutex_);
     418           0 :             wait_tree_.erase(it);
     419           0 :         }
     420             :     }
     421       11312 :     delete[] data.buff_;
     422       11312 :     return true;
     423             : }
     424             : 
     425           2 : bool KSyncSock::BlockingRecv() {
     426             :     char data[kBufLen];
     427           2 :     bool ret = false;
     428             : 
     429             :     do {
     430           2 :         Receive(boost::asio::buffer(data, kBufLen));
     431           2 :         AgentSandeshContext *ctxt = KSyncSock::GetAgentSandeshContext(0);
     432           2 :         ctxt->SetErrno(0);
     433             :         // BlockingRecv used only during Init and doesnt support bulk messages
     434             :         // Use non-bulk version of decoder
     435           2 :         Decoder(data, ctxt);
     436           2 :         if (ctxt->GetErrno() != 0 && ctxt->GetErrno() != EEXIST) {
     437           0 :             KSYNC_ERROR(VRouterError, "VRouter operation failed. Error <",
     438             :                         ctxt->GetErrno(), ":",
     439             :                         KSyncEntry::VrouterErrorToString(ctxt->GetErrno()),
     440             :                         ">. Object <", "N/A", ">. State <", "N/A",
     441             :                         ">. Message number :", 0);
     442           0 :             ret = true;
     443             :         }
     444           2 :     } while (IsMoreData(data));
     445             : 
     446           2 :     return ret;
     447             : }
     448             : 
     449             : // BlockingSend does not support bulk messages.
     450           2 : size_t KSyncSock::BlockingSend(char *msg, int msg_len) {
     451           2 :     KSyncBufferList iovec;
     452           2 :     iovec.push_back(buffer(msg, msg_len));
     453           2 :     bulk_buf_size_ = msg_len;
     454           4 :     return SendTo(&iovec, 0);
     455           2 : }
     456             : 
     457           0 : void KSyncSock::GenericSend(IoContext *ioc) {
     458           0 :     send_queue_.Enqueue(ioc);
     459           0 : }
     460             : 
     461       11249 : void KSyncSock::SendAsync(KSyncEntry *entry, int msg_len, char *msg,
     462             :                           KSyncEntry::KSyncEvent event) {
     463       11249 :     KSyncIoContext *ioc = new KSyncIoContext(this, entry, msg_len, msg, event);
     464             :     // Pre-allocate buffers to minimize processing in KSyncTxQueue context
     465       11249 :     if (read_inline_ && entry->pre_alloc_rx_buffer()) {
     466         378 :         ioc->rx_buffer1_ = new char [kBufLen];
     467         378 :         ioc->rx_buffer2_ = new char [kBufLen];
     468             :     } else {
     469       10871 :         ioc->rx_buffer1_ = ioc->rx_buffer2_ = NULL;
     470             :     }
     471       11249 :     send_queue_.Enqueue(ioc);
     472       11249 : }
     473             : 
     474             : // Write handler registered with boost::asio
     475           0 : void KSyncSock::WriteHandler(const boost::system::error_code& error,
     476             :                              size_t bytes_transferred) {
     477           0 :     if (error) {
     478           0 :         LOG(ERROR, "Ksync sock write error : " <<
     479             :             boost::system::system_error(error).what());
     480           0 :         if (shutdown_ == false) {
     481           0 :             assert(0);
     482             :         }
     483             :     }
     484           0 : }
     485             : 
     486             : // End of messages in the work-queue. Send messages pending in bulk context
     487       11111 : void KSyncSock::OnEmptyQueue(bool done) {
     488       11111 :     if (bulk_seq_no_ == kInvalidBulkSeqNo)
     489           0 :         return;
     490             : 
     491       11111 :     KSyncBulkMsgContext *bulk_message_context = NULL;
     492       11111 :     if (use_wait_tree_) {
     493       11111 :         if (read_inline_ == false) {
     494           0 :             std::scoped_lock lock(mutex_);
     495           0 :             WaitTree::iterator it = wait_tree_.find(bulk_seq_no_);
     496           0 :             assert(it != wait_tree_.end());
     497           0 :             bulk_message_context = &it->second;
     498           0 :         } else {
     499       11111 :             bulk_message_context = bulk_msg_context_;
     500             :         }
     501             :     } else {
     502           0 :         bulk_message_context = bulk_mctx_arr_[bmca_prod_];
     503             :     }
     504             : 
     505       11111 :     SendBulkMessage(bulk_message_context, bulk_seq_no_);
     506             : }
     507             : 
     508             : // Send messages accumilated in bulk context
     509       11112 : int KSyncSock::SendBulkMessage(KSyncBulkMsgContext *bulk_message_context,
     510             :                                uint32_t seqno) {
     511       11112 :     KSyncBufferList iovec;
     512             :     // Get all buffers to send into single io-vector
     513       11112 :     bulk_message_context->Data(&iovec);
     514       11112 :     tx_count_++;
     515             : 
     516       11112 :     if (!read_inline_) {
     517           0 :         if (!use_wait_tree_) {
     518           0 :             bmca_prod_++;
     519           0 :             if (bmca_prod_ >= KSYNC_BMC_ARR_SIZE) {
     520           0 :                 bmca_prod_ = 0;
     521             :             }
     522             :         }
     523             : 
     524           0 :         AsyncSendTo(&iovec, seqno,
     525             :                     boost::bind(&KSyncSock::WriteHandler, this,
     526             :                                 placeholders::error,
     527             :                                 placeholders::bytes_transferred));
     528             :     } else {
     529       11112 :         SendTo(&iovec, seqno);
     530       11112 :         bool more_data = false;
     531       11312 :         do {
     532       11312 :             char *rxbuf = bulk_message_context->GetReceiveBuffer();
     533       11312 :             Receive(boost::asio::buffer(rxbuf, kBufLen));
     534       11312 :             more_data = IsMoreData(rxbuf);
     535       11312 :             if (!process_data_inline_) {
     536       11312 :                 ValidateAndEnqueue(rxbuf, bulk_message_context);
     537             :             } else {
     538           0 :                 ProcessDataInline(rxbuf);
     539             :             }
     540             :         } while(more_data);
     541             :     }
     542             : 
     543       11112 :     bulk_msg_context_ = NULL;
     544       11112 :     bulk_seq_no_ = kInvalidBulkSeqNo;
     545       11112 :     return true;
     546       11112 : }
     547             : 
     548             : // Get the bulk-context for sequence-number
     549       11250 : KSyncBulkMsgContext *KSyncSock::LocateBulkContext
     550             : (uint32_t seqno, IoContext::Type io_context_type,
     551             :  uint32_t work_queue_index) {
     552       11250 :     if (read_inline_) {
     553       11250 :         if (bulk_seq_no_ == kInvalidBulkSeqNo) {
     554       11112 :             assert(bulk_msg_context_ == NULL);
     555       11112 :             bulk_seq_no_ = seqno;
     556       11112 :             bulk_buf_size_ = 0;
     557       11112 :             bulk_msg_count_ = 0;
     558       11112 :             bulk_msg_context_ = new KSyncBulkMsgContext(io_context_type,
     559       11112 :                                                         work_queue_index);
     560             :         }
     561       11250 :         return bulk_msg_context_;
     562             :     }
     563             : 
     564           0 :     if (use_wait_tree_) {
     565           0 :         std::scoped_lock lock(mutex_);
     566           0 :         if (bulk_seq_no_ == kInvalidBulkSeqNo) {
     567           0 :             bulk_seq_no_ = seqno;
     568           0 :             bulk_buf_size_ = 0;
     569           0 :             bulk_msg_count_ = 0;
     570             : 
     571           0 :             wait_tree_.insert(WaitTreePair(seqno,
     572           0 :                                        KSyncBulkMsgContext(io_context_type,
     573             :                                                            work_queue_index)));
     574             :         }
     575             : 
     576           0 :         WaitTree::iterator it = wait_tree_.find(bulk_seq_no_);
     577           0 :         assert(it != wait_tree_.end());
     578           0 :         return &it->second;
     579           0 :     } else {
     580           0 :         if (bulk_seq_no_ == kInvalidBulkSeqNo) {
     581           0 :             bulk_seq_no_ = seqno;
     582           0 :             bulk_buf_size_ = 0;
     583           0 :             bulk_msg_count_ = 0;
     584             : 
     585           0 :             bulk_mctx_arr_[bmca_prod_] = new KSyncBulkMsgContext(io_context_type,
     586           0 :                                                             work_queue_index);
     587           0 :             bulk_mctx_arr_[bmca_prod_]->set_seqno(seqno);
     588             :         }
     589             : 
     590           0 :         return bulk_mctx_arr_[bmca_prod_];
     591             :     }
     592             : 
     593             :     return NULL;
     594             : }
     595             : 
     596             : // Try adding an io-context to bulk context. Returns
     597             : //  - true  : if message can be added to bulk context
     598             : //  - false : if message cannot be added to bulk context
     599       11250 : bool KSyncSock::TryAddToBulk(KSyncBulkMsgContext *bulk_message_context,
     600             :                              IoContext *ioc) {
     601       11250 :     if (bulk_msg_count_ >= max_bulk_msg_count_)
     602           1 :         return false;
     603             : 
     604       11249 :     if (bulk_message_context->io_context_type() != ioc->type())
     605           0 :         return false;
     606             : 
     607       11249 :     if (use_wait_tree_) {
     608       11249 :         if (bulk_message_context->work_queue_index() != ioc->index())
     609           0 :             return false;
     610             :     }
     611             : 
     612       11249 :     bulk_buf_size_ += ioc->GetMsgLen();
     613       11249 :     bulk_msg_count_++;
     614             : 
     615       11249 :     bulk_message_context->Insert(ioc);
     616       11249 :     if (ioc->rx_buffer1()) {
     617         378 :         bulk_message_context->AddReceiveBuffer(ioc->rx_buffer1());
     618         378 :         ioc->reset_rx_buffer1();
     619             : 
     620             :     }
     621       11249 :     if (ioc->rx_buffer2()) {
     622         378 :         bulk_message_context->AddReceiveBuffer(ioc->rx_buffer2());
     623         378 :         ioc->reset_rx_buffer2();
     624             :     }
     625       11249 :     return true;
     626             : }
     627             : 
     628       11249 : bool KSyncSock::SendAsyncImpl(IoContext *ioc) {
     629             :     KSyncBulkMsgContext *bulk_message_context =
     630       11249 :         LocateBulkContext(ioc->GetSeqno(), ioc->type(), ioc->index());
     631             :     // Try adding message to bulk-message list
     632       11249 :     if (TryAddToBulk(bulk_message_context, ioc)) {
     633             :         // Message added to bulk-list. Nothing more to do
     634       11248 :         return true;
     635             :     }
     636             : 
     637             :     // Message cannot be added to bulk-list. Send the current list
     638           1 :     SendBulkMessage(bulk_message_context, bulk_seq_no_);
     639             : 
     640             :     // Allocate a new context and add message to it
     641           1 :     bulk_message_context = LocateBulkContext(ioc->GetSeqno(), ioc->type(),
     642             :                                              ioc->index());
     643           1 :     assert(TryAddToBulk(bulk_message_context, ioc));
     644           1 :     return true;
     645             : }
     646             : 
     647             : 
     648             : /////////////////////////////////////////////////////////////////////////////
     649             : // KSyncSockNetlink routines
     650             : /////////////////////////////////////////////////////////////////////////////
     651           0 : KSyncSockNetlink::KSyncSockNetlink(boost::asio::io_context &ios, int protocol)
     652           0 :     : sock_(ios, protocol) {
     653           0 :     ReceiveBuffForceSize set_rcv_buf;
     654           0 :     set_rcv_buf = KSYNC_SOCK_RECV_BUFF_SIZE;
     655           0 :     boost::system::error_code ec;
     656           0 :     sock_.set_option(set_rcv_buf, ec);
     657           0 :     if (ec.value() != 0) {
     658           0 :         LOG(ERROR, "Error Changing netlink receive sock buffer size to " <<
     659             :                 set_rcv_buf.value() << " error = " <<
     660             :                 boost::system::system_error(ec).what());
     661             :     }
     662           0 :     boost::asio::socket_base::receive_buffer_size rcv_buf_size;
     663           0 :     boost::system::error_code ec1;
     664           0 :     sock_.get_option(rcv_buf_size, ec);
     665           0 :     LOG(INFO, "Current receive sock buffer size is " << rcv_buf_size.value());
     666           0 : }
     667             : 
     668           0 : KSyncSockNetlink::~KSyncSockNetlink() {
     669           0 : }
     670             : 
     671           0 : void KSyncSockNetlink::Init(io_service &ios, int protocol, bool use_work_queue,
     672             :                             const std::string &cpu_pin_policy) {
     673           0 :     KSyncSock::SetSockTableEntry(new KSyncSockNetlink(ios, protocol));
     674           0 :     KSyncSock::Init(use_work_queue, cpu_pin_policy);
     675           0 : }
     676             : 
     677           0 : uint32_t KSyncSockNetlink::GetSeqno(char *data) {
     678           0 :     return GetNetlinkSeqno(data);
     679             : }
     680             : 
     681           0 : bool KSyncSockNetlink::IsMoreData(char *data) {
     682           0 :     return NetlinkMsgDone(data);
     683             : }
     684             : 
     685           0 : bool KSyncSockNetlink::Validate(char *data) {
     686           0 :     return ValidateNetlink(data);
     687             : }
     688             : 
     689             : //netlink socket class for interacting with kernel
     690           0 : void KSyncSockNetlink::AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
     691             :                                    HandlerCb cb) {
     692           0 :     ResetNetlink(nl_client_);
     693           0 :     KSyncBufferList::iterator it = iovec->begin();
     694           0 :     iovec->insert(it, buffer((char *)nl_client_->cl_buf,
     695           0 :                              nl_client_->cl_buf_offset));
     696           0 :     UpdateNetlink(nl_client_, bulk_buf_size_, seq_no);
     697             : 
     698           0 :     boost::asio::netlink::raw::endpoint ep;
     699           0 :     sock_.async_send_to(*iovec, ep, cb);
     700           0 : }
     701             : 
     702           0 : size_t KSyncSockNetlink::SendTo(KSyncBufferList *iovec, uint32_t seq_no) {
     703           0 :     ResetNetlink(nl_client_);
     704           0 :     KSyncBufferList::iterator it = iovec->begin();
     705           0 :     iovec->insert(it, buffer((char *)nl_client_->cl_buf,
     706           0 :                              nl_client_->cl_buf_offset));
     707           0 :     UpdateNetlink(nl_client_, bulk_buf_size_, seq_no);
     708             : 
     709           0 :     boost::asio::netlink::raw::endpoint ep;
     710           0 :     return sock_.send_to(*iovec, ep);
     711             : }
     712             : 
     713             : // Static method to decode non-bulk message
     714           2 : void KSyncSockNetlink::NetlinkDecoder(char *data, SandeshContext *ctxt) {
     715           2 :     assert(ValidateNetlink(data));
     716           2 :     char *buf = NULL;
     717           2 :     uint32_t buf_len = 0;
     718           2 :     GetNetlinkPayload(data, &buf, &buf_len);
     719           2 :     DecodeSandeshMessages(buf, buf_len, ctxt, NLA_ALIGNTO);
     720           2 : }
     721             : 
     722           0 : bool KSyncSockNetlink::Decoder(char *data, AgentSandeshContext *context) {
     723           0 :     NetlinkDecoder(data, context);
     724           0 :     return true;
     725             : }
     726             : 
     727             : // Static method used in ksync_sock_user only
     728       11312 : void KSyncSockNetlink::NetlinkBulkDecoder(char *data, SandeshContext *ctxt,
     729             :                                           bool more) {
     730       11312 :     assert(ValidateNetlink(data));
     731       11312 :     char *buf = NULL;
     732       11312 :     uint32_t buf_len = 0;
     733       11312 :     GetNetlinkPayload(data, &buf, &buf_len);
     734             :     KSyncBulkSandeshContext *bulk_sandesh_context =
     735       11312 :         dynamic_cast<KSyncBulkSandeshContext *>(ctxt);
     736       11312 :     bulk_sandesh_context->Decoder(buf, buf_len, NLA_ALIGNTO, more);
     737       11312 : }
     738             : 
     739           0 : bool KSyncSockNetlink::BulkDecoder(char *data,
     740             :                                    KSyncBulkSandeshContext *bulk_sandesh_context) {
     741             :     // Get sandesh buffer and buffer-length
     742           0 :     uint32_t buf_len = 0;
     743           0 :     char *buf = NULL;
     744           0 :     GetNetlinkPayload(data, &buf, &buf_len);
     745           0 :     return bulk_sandesh_context->Decoder(buf, buf_len, NLA_ALIGNTO, IsMoreData(data));
     746             : }
     747             : 
     748           0 : void KSyncSockNetlink::AsyncReceive(mutable_buffers_1 buf, HandlerCb cb) {
     749           0 :     sock_.async_receive(buf, cb);
     750           0 : }
     751             : 
     752           0 : void KSyncSockNetlink::Receive(mutable_buffers_1 buf) {
     753           0 :     sock_.receive(buf);
     754           0 :     struct nlmsghdr *nlh = buffer_cast<struct nlmsghdr *>(buf);
     755           0 :     if (nlh->nlmsg_type == NLMSG_ERROR) {
     756           0 :         LOG(ERROR, "Netlink error for seqno " << nlh->nlmsg_seq
     757             :                 << " len " << nlh->nlmsg_len);
     758           0 :         assert(0);
     759             :     }
     760           0 : }
     761             : 
     762             : /////////////////////////////////////////////////////////////////////////////
     763             : // KSyncSockUdp routines
     764             : /////////////////////////////////////////////////////////////////////////////
     765             : //Udp socket class for interacting with kernel
     766           0 : KSyncSockUdp::KSyncSockUdp(boost::asio::io_context &ios, int port) :
     767           0 :     sock_(ios, boost::asio::ip::udp::endpoint(boost::asio::ip::udp::v4(), 0)),
     768           0 :     server_ep_(boost::asio::ip::address::from_string("127.0.0.1"), port) {
     769           0 : }
     770             : 
     771           0 : void KSyncSockUdp::Init(io_service &ios, int port,
     772             :                         const std::string &cpu_pin_policy) {
     773           0 :     KSyncSock::SetSockTableEntry(new KSyncSockUdp(ios, port));
     774           0 :     KSyncSock::Init(false, cpu_pin_policy);
     775           0 : }
     776             : 
     777           0 : uint32_t KSyncSockUdp::GetSeqno(char *data) {
     778           0 :     struct uvr_msg_hdr *hdr = (struct uvr_msg_hdr *)data;
     779           0 :     return hdr->seq_no;
     780             : }
     781             : 
     782           0 : bool KSyncSockUdp::IsMoreData(char *data) {
     783           0 :     struct uvr_msg_hdr *hdr = (struct uvr_msg_hdr *)data;
     784           0 :     return ((hdr->flags & UVR_MORE) == UVR_MORE);
     785             : }
     786             : 
     787             : // We dont expect any non-bulk operation on UDP
     788           0 : bool KSyncSockUdp::Decoder(char *data, AgentSandeshContext *context) {
     789           0 :     assert(0);
     790             :     return false;
     791             : }
     792             : 
     793           0 : bool KSyncSockUdp::BulkDecoder(char *data,
     794             :                                KSyncBulkSandeshContext *bulk_sandesh_context) {
     795           0 :     struct uvr_msg_hdr *hdr = (struct uvr_msg_hdr *)data;
     796           0 :     uint32_t buf_len = hdr->msg_len;
     797           0 :     char *buf = data + sizeof(struct uvr_msg_hdr);
     798           0 :     return bulk_sandesh_context->Decoder(buf, buf_len, 1, IsMoreData(data));
     799             : }
     800             : 
     801           0 : void KSyncSockUdp::AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
     802             :                                HandlerCb cb) {
     803             :     struct uvr_msg_hdr hdr;
     804           0 :     hdr.seq_no = seq_no;
     805           0 :     hdr.flags = 0;
     806           0 :     hdr.msg_len = bulk_buf_size_;
     807             : 
     808           0 :     KSyncBufferList::iterator it = iovec->begin();
     809           0 :     iovec->insert(it, buffer((char *)(&hdr), sizeof(hdr)));
     810             : 
     811           0 :     sock_.async_send_to(*iovec, server_ep_, cb);
     812           0 : }
     813             : 
     814           0 : size_t KSyncSockUdp::SendTo(KSyncBufferList *iovec, uint32_t seq_no) {
     815             :     struct uvr_msg_hdr hdr;
     816           0 :     hdr.seq_no = seq_no;
     817           0 :     hdr.flags = 0;
     818           0 :     hdr.msg_len = bulk_buf_size_;
     819             : 
     820           0 :     KSyncBufferList::iterator it = iovec->begin();
     821           0 :     iovec->insert(it, buffer((char *)(&hdr), sizeof(hdr)));
     822             : 
     823           0 :     return sock_.send_to(*iovec, server_ep_, MSG_DONTWAIT);
     824             : }
     825             : 
     826           0 : bool KSyncSockUdp::Validate(char *data) {
     827           0 :     return true;
     828             : }
     829             : 
     830           0 : void KSyncSockUdp::AsyncReceive(mutable_buffers_1 buf, HandlerCb cb) {
     831           0 :     boost::asio::ip::udp::endpoint ep;
     832           0 :     sock_.async_receive_from(buf, ep, cb);
     833           0 : }
     834             : 
     835           0 : void KSyncSockUdp::Receive(mutable_buffers_1 buf) {
     836           0 :     boost::asio::ip::udp::endpoint ep;
     837           0 :     sock_.receive_from(buf, ep);
     838           0 : }
     839             : 
     840           0 : void KSyncSock::ProcessDataInline(char *data) {
     841           0 :     KSyncBulkMsgContext *bulk_message_context = NULL;
     842             :     KSyncBulkSandeshContext *bulk_sandesh_context;
     843           0 :     uint32_t seqno = GetSeqno(data);
     844             : 
     845           0 :     assert(!use_wait_tree_);
     846           0 :     Validate(data);
     847             : 
     848           0 :     bulk_sandesh_context = GetBulkSandeshContext(seqno);
     849           0 :     bulk_message_context = bulk_mctx_arr_[bmca_cons_];
     850           0 :     assert(bulk_message_context->seqno() == seqno);
     851             : 
     852           0 :     bulk_sandesh_context->set_bulk_message_context(bulk_message_context);
     853           0 :     BulkDecoder(data, bulk_sandesh_context);
     854             : 
     855             :     // Remove the IoContext only on last netlink message
     856           0 :     if (IsMoreData(data) == false) {
     857           0 :         delete bulk_message_context;
     858           0 :         bmca_cons_++;
     859           0 :         if (bmca_cons_ >= KSYNC_BMC_ARR_SIZE) {
     860           0 :             bmca_cons_ = 0;
     861             :         }
     862             :     }
     863             : 
     864           0 :     return;
     865             : }
     866             : 
     867             : /////////////////////////////////////////////////////////////////////////////
     868             : // KSyncIoContext routines
     869             : /////////////////////////////////////////////////////////////////////////////
     870       11249 : KSyncIoContext::KSyncIoContext(KSyncSock *sock, KSyncEntry *sync_entry,
     871             :                                int msg_len, char *msg,
     872       11249 :                                KSyncEntry::KSyncEvent event) :
     873             :     IoContext(msg, msg_len, 0,
     874       11249 :               sock->GetAgentSandeshContext(sync_entry->GetTableIndex()),
     875       11249 :               IoContext::IOC_KSYNC, sync_entry->GetTableIndex()),
     876       11249 :     entry_(sync_entry), event_(event), sock_(sock) {
     877       11249 :     SetSeqno(sock->AllocSeqNo(type(), index()));
     878       11249 : }
     879             : 
     880       11249 : void KSyncIoContext::Handler() {
     881       11249 :     sock_->EnqueueRxProcessData(entry_, event_);
     882       11249 : }
     883             : 
     884           0 : void KSyncIoContext::ErrorHandler(int err) {
     885           0 :     entry_->ErrorHandler(err, GetSeqno(), event_);
     886           0 : }
     887             : 
     888             : /////////////////////////////////////////////////////////////////////////////
     889             : // Routines for KSyncBulkSandeshContext
     890             : /////////////////////////////////////////////////////////////////////////////
     891           8 : KSyncBulkSandeshContext::KSyncBulkSandeshContext() :
     892           8 :     AgentSandeshContext(), bulk_msg_context_(NULL)  { }
     893             : 
     894           0 : KSyncBulkSandeshContext::~KSyncBulkSandeshContext() {
     895           0 : }
     896             : 
     897             : // Sandesh responses for old context are done. Check for any errors
     898       11249 : void KSyncBulkSandeshContext::IoContextDone() {
     899       11249 :     IoContext *io_context = &(*bulk_msg_context_->io_context_list_it_);
     900       11249 :     AgentSandeshContext *sandesh_context = io_context->GetSandeshContext();
     901             : 
     902       11249 :     sandesh_context->set_ksync_io_ctx(NULL);
     903       11249 :     if (sandesh_context->GetErrno() != 0 &&
     904           0 :         sandesh_context->GetErrno() != EEXIST) {
     905           0 :         io_context->ErrorHandler(sandesh_context->GetErrno());
     906             :     }
     907       11249 :     io_context->Handler();
     908       11249 : }
     909             : 
     910       11249 : void KSyncBulkSandeshContext::IoContextStart() {
     911       11249 :     bulk_msg_context_->vr_response_count_++;
     912       11249 :     IoContext &io_context = *bulk_msg_context_->io_context_list_it_;
     913       11249 :     AgentSandeshContext *sandesh_context = io_context.GetSandeshContext();
     914             :     sandesh_context->set_ksync_io_ctx
     915       11249 :         (static_cast<KSyncIoContext *>(&io_context));
     916       11249 : }
     917             : 
     918             : // Process the sandesh messages
     919             : // There can be more then one sandesh messages in the netlink buffer.
     920             : // Iterate and process all of them
     921       11312 : bool KSyncBulkSandeshContext::Decoder(char *data, uint32_t len,
     922             :                                       uint32_t alignment, bool more) {
     923       11312 :     DecodeSandeshMessages(data, len, this, alignment);
     924       22624 :     assert(bulk_msg_context_->io_context_list_it_ !=
     925             :            bulk_msg_context_->io_context_list_.end());
     926       11312 :     if (more == true)
     927         200 :         return false;
     928             : 
     929       11112 :     IoContextDone();
     930             : 
     931             :     // No more netlink messages. Validate that iterator points to last element
     932             :     // in IoContextList
     933       11112 :     bulk_msg_context_->io_context_list_it_++;
     934       22224 :     assert(bulk_msg_context_->io_context_list_it_ ==
     935             :            bulk_msg_context_->io_context_list_.end());
     936       11112 :     return true;
     937             : }
     938             : 
     939       11249 : void KSyncBulkSandeshContext::SetErrno(int err) {
     940       11249 :     AgentSandeshContext *context = GetSandeshContext();
     941       11249 :     context->SetErrno(err);
     942       11249 : }
     943             : 
     944       11627 : AgentSandeshContext *KSyncBulkSandeshContext::GetSandeshContext() {
     945       11627 :     assert(bulk_msg_context_->vr_response_count_);
     946       23254 :     return bulk_msg_context_->io_context_list_it_->GetSandeshContext();
     947             : }
     948             : 
     949           0 : void KSyncBulkSandeshContext::IfMsgHandler(vr_interface_req *req) {
     950           0 :     AgentSandeshContext *context = GetSandeshContext();
     951           0 :     context->IfMsgHandler(req);
     952           0 : }
     953             : 
     954           0 : void KSyncBulkSandeshContext::NHMsgHandler(vr_nexthop_req *req) {
     955           0 :     AgentSandeshContext *context = GetSandeshContext();
     956           0 :     context->NHMsgHandler(req);
     957           0 : }
     958             : 
     959           0 : void KSyncBulkSandeshContext::RouteMsgHandler(vr_route_req *req) {
     960           0 :     AgentSandeshContext *context = GetSandeshContext();
     961           0 :     context->RouteMsgHandler(req);
     962           0 : }
     963             : 
     964           0 : void KSyncBulkSandeshContext::MplsMsgHandler(vr_mpls_req *req) {
     965           0 :     AgentSandeshContext *context = GetSandeshContext();
     966           0 :     context->MplsMsgHandler(req);
     967           0 : }
     968             : 
     969           0 : void KSyncBulkSandeshContext::QosConfigMsgHandler(vr_qos_map_req *req) {
     970           0 :     AgentSandeshContext *context = GetSandeshContext();
     971           0 :     context->QosConfigMsgHandler(req);
     972           0 : }
     973             : 
     974           0 : void KSyncBulkSandeshContext::ForwardingClassMsgHandler(vr_fc_map_req *req) {
     975           0 :     AgentSandeshContext *context = GetSandeshContext();
     976           0 :     context->ForwardingClassMsgHandler(req);
     977           0 : }
     978             : 
     979             : // vr_response message is treated as delimiter in a bulk-context. So, move to
     980             : // next io-context within bulk-message context.
     981       11249 : int KSyncBulkSandeshContext::VrResponseMsgHandler(vr_response *resp) {
     982       11249 :     AgentSandeshContext *sandesh_context = NULL;
     983             :     // If this is first vr_reponse received, move io-context to first entry in
     984             :     // bulk context
     985       11249 :     if (bulk_msg_context_->vr_response_count_ == 0) {
     986       11112 :         bulk_msg_context_->io_context_list_it_ =
     987       11112 :             bulk_msg_context_->io_context_list_.begin();
     988             :         sandesh_context =
     989       22224 :             bulk_msg_context_->io_context_list_it_->GetSandeshContext();
     990       11112 :         IoContextStart();
     991             :     } else {
     992             :         // Sandesh responses for old io-context are done.
     993             :         // Check for any errors and trigger state-machine for old io-context
     994         137 :         IoContextDone();
     995             :         // Move to the next io-context
     996         137 :         bulk_msg_context_->io_context_list_it_++;
     997         274 :         assert(bulk_msg_context_->io_context_list_it_ !=
     998             :                bulk_msg_context_->io_context_list_.end());
     999             :         sandesh_context =
    1000         274 :             bulk_msg_context_->io_context_list_it_->GetSandeshContext();
    1001         137 :         IoContextStart();
    1002             :     }
    1003       11249 :     return sandesh_context->VrResponseMsgHandler(resp);
    1004             : }
    1005             : 
    1006           0 : void KSyncBulkSandeshContext::MirrorMsgHandler(vr_mirror_req *req) {
    1007           0 :     AgentSandeshContext *context = GetSandeshContext();
    1008           0 :     context->MirrorMsgHandler(req);
    1009           0 : }
    1010             : 
    1011           0 : void KSyncBulkSandeshContext::FlowMsgHandler(vr_flow_req *req) {
    1012           0 :     AgentSandeshContext *context = GetSandeshContext();
    1013           0 :     context->FlowMsgHandler(req);
    1014           0 : }
    1015             : 
    1016         378 : void KSyncBulkSandeshContext::FlowResponseHandler(vr_flow_response *req) {
    1017         378 :     AgentSandeshContext *context = GetSandeshContext();
    1018         378 :     context->FlowResponseHandler(req);
    1019         378 : }
    1020             : 
    1021           0 : void KSyncBulkSandeshContext::VrfAssignMsgHandler(vr_vrf_assign_req *req) {
    1022           0 :     AgentSandeshContext *context = GetSandeshContext();
    1023           0 :     context->VrfAssignMsgHandler(req);
    1024           0 : }
    1025             : 
    1026           0 : void KSyncBulkSandeshContext::VrfMsgHandler(vr_vrf_req *req) {
    1027           0 :     AgentSandeshContext *context = GetSandeshContext();
    1028           0 :     context->VrfMsgHandler(req);
    1029           0 : }
    1030             : 
    1031           0 : void KSyncBulkSandeshContext::VrfStatsMsgHandler(vr_vrf_stats_req *req) {
    1032           0 :     AgentSandeshContext *context = GetSandeshContext();
    1033           0 :     context->VrfStatsMsgHandler(req);
    1034           0 : }
    1035             : 
    1036           0 : void KSyncBulkSandeshContext::DropStatsMsgHandler(vr_drop_stats_req *req) {
    1037           0 :     AgentSandeshContext *context = GetSandeshContext();
    1038           0 :     context->DropStatsMsgHandler(req);
    1039           0 : }
    1040             : 
    1041           0 : void KSyncBulkSandeshContext::VxLanMsgHandler(vr_vxlan_req *req) {
    1042           0 :     AgentSandeshContext *context = GetSandeshContext();
    1043           0 :     context->VxLanMsgHandler(req);
    1044           0 : }
    1045             : 
    1046           0 : void KSyncBulkSandeshContext::VrouterOpsMsgHandler(vrouter_ops *req) {
    1047           0 :     AgentSandeshContext *context = GetSandeshContext();
    1048           0 :     context->VrouterOpsMsgHandler(req);
    1049           0 : }
    1050             : 
    1051             : /////////////////////////////////////////////////////////////////////////////
    1052             : // KSyncBulkMsgContext routines
    1053             : /////////////////////////////////////////////////////////////////////////////
    1054       11112 : KSyncBulkMsgContext::KSyncBulkMsgContext(IoContext::Type type,
    1055       11112 :                                          uint32_t index) :
    1056       11112 :     io_context_list_(), io_context_type_(type), work_queue_index_(index),
    1057       11112 :     rx_buffer_index_(0), vr_response_count_(0), io_context_list_it_() {
    1058       11112 : }
    1059             : 
    1060           0 : KSyncBulkMsgContext::KSyncBulkMsgContext(const KSyncBulkMsgContext &rhs) :
    1061           0 :     io_context_list_(), io_context_type_(rhs.io_context_type_),
    1062           0 :     work_queue_index_(rhs.work_queue_index_),
    1063           0 :     rx_buffer_index_(0), vr_response_count_(0), io_context_list_it_() {
    1064           0 :     assert(rhs.vr_response_count_ == 0);
    1065           0 :     assert(rhs.rx_buffer_index_ == 0);
    1066           0 :     assert(rhs.io_context_list_.size() == 0);
    1067           0 : }
    1068             : 
    1069             : struct IoContextDisposer {
    1070       11249 :     void operator() (IoContext *io_context) { delete io_context; }
    1071             : };
    1072             : 
    1073       11112 : KSyncBulkMsgContext::~KSyncBulkMsgContext() {
    1074       11112 :     assert(vr_response_count_ == io_context_list_.size());
    1075       11112 :     io_context_list_.clear_and_dispose(IoContextDisposer());
    1076       11443 :     for (uint32_t i = 0; i < rx_buffer_index_; i++) {
    1077         331 :         delete[] rx_buffers_[i];
    1078             :     }
    1079       11112 : }
    1080             : 
    1081       11312 : char *KSyncBulkMsgContext::GetReceiveBuffer() {
    1082       11312 :     if (rx_buffer_index_ == 0)
    1083       10887 :         return new char[KSyncSock::kBufLen];
    1084             : 
    1085         425 :     return rx_buffers_[--rx_buffer_index_];
    1086             : }
    1087             : 
    1088         756 : void KSyncBulkMsgContext::AddReceiveBuffer(char *buff) {
    1089         756 :     assert(rx_buffer_index_ < kMaxRxBufferCount);
    1090         756 :     rx_buffers_[rx_buffer_index_++] = buff;
    1091         756 : }
    1092             : 
    1093       11249 : void KSyncBulkMsgContext::Insert(IoContext *ioc) {
    1094       11249 :     io_context_list_.push_back(*ioc);
    1095       11249 :     return;
    1096             : }
    1097             : 
    1098       11112 : void KSyncBulkMsgContext::Data(KSyncBufferList *iovec) {
    1099       11112 :     IoContextList::iterator it = io_context_list_.begin();
    1100       44722 :     while (it != io_context_list_.end()) {
    1101       33747 :         iovec->push_back(buffer(it->GetMsg(), it->GetMsgLen()));
    1102       11249 :         it++;
    1103             :     }
    1104       11112 : }