Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2018 Juniper Networks, Inc. All rights reserved.
       3             :  */
       4             : 
       5             : #include "ksync_sock.h"
       6             : 
       7             : #include <boost/asio.hpp>
       8             : #include <boost/bind/bind.hpp>
       9             : #include <errno.h>
      10             : 
      11             : class AgentParam;
      12             : 
      13             : using namespace boost::asio;
      14             : using namespace boost::placeholders;
      15             : 
      16             : /////////////////////////////////////////////////////////////////////////////
      17             : // KSyncSockUds routines
      18             : /////////////////////////////////////////////////////////////////////////////
      19             : //Unix domain socket class for interacting with dpdk vrouter
      20             : 
      21             : class KSyncSockUdsReadTask : public Task {
      22             : public:
      23           0 :     KSyncSockUdsReadTask(TaskScheduler *scheduler, KSyncSockUds *queue) :
      24           0 :         Task(scheduler->GetTaskId("Ksync::KSyncSockUdsRead"), 0), queue_(queue) {
      25           0 :     }
      26           0 :     ~KSyncSockUdsReadTask() {
      27           0 :     }
      28             : 
      29           0 :     bool Run() {
      30           0 :         queue_->Run();
      31           0 :         return true;
      32             :     }
      33           0 :     std::string Description() const { return "KSyncSockUdsRead"; }
      34             : private:
      35             :     KSyncSockUds *queue_;
      36             : 
      37             : };
      38             : 
      39             : 
      40             : 
      41             : string KSyncSockUds::sockpath_= KSYNC_AGENT_VROUTER_SOCK_PATH;
      42             : 
      43           0 : KSyncSockUds::KSyncSockUds(boost::asio::io_context &ios) :
      44           0 :     sock_(ios),
      45           0 :     server_ep_(sockpath_),
      46           0 :     rx_buff_(NULL),
      47           0 :     rx_buff_q_(NULL),
      48           0 :     remain_(0),
      49           0 :     socket_(0),
      50           0 :     connected_(false) {
      51           0 :     boost::system::error_code ec;
      52           0 :     reset_use_wait_tree();
      53           0 :     set_process_data_inline();
      54           0 : retry:;
      55           0 :     sock_.connect(server_ep_, ec);
      56           0 :     if (ec) {
      57           0 :         sleep(1);
      58           0 :         goto retry;
      59             :     }
      60           0 :     socket_ = sock_.native_handle();
      61           0 :     connected_ = true;
      62           0 :     rx_buff_   = new char[10*kBufLen];
      63           0 :     rx_buff_q_ = new char[10*kBufLen];
      64           0 : }
      65             : 
      66           0 : bool KSyncSockUds::Run() {
      67           0 :     AgentSandeshContext *ctxt = KSyncSock::GetAgentSandeshContext(0);
      68           0 :     char *ret_buff = new char[1024*kBufLen];
      69           0 :     boost::system::error_code ec;
      70             : 
      71             :     // Read data from the socket and append it to the existing
      72             :     // unprocessed data in the local buffer.
      73             :     while (1) {
      74           0 :         char *bufp = rx_buff_;
      75           0 :         struct nlmsghdr *nlh = NULL;
      76             :         struct nlmsghdr tnlh;
      77           0 :         size_t offset = 0;
      78             :         int ret_val;
      79           0 :         size_t bytes_transferred = 0;
      80           0 :         bytes_transferred = ret_val = recv(socket_, rx_buff_, 10*kBufLen, 0);
      81           0 :         if (ret_val == 0) {
      82             :             // connection reset by peer
      83             :             // close socket and exit
      84           0 :             sock_.close(ec);
      85           0 :             LOG(INFO, " dpdk vrouter is down, exiting.. errno:" << errno);
      86           0 :             exit(0);
      87             :         }
      88           0 :         if (ret_val < 0) {
      89           0 :             if (errno != EAGAIN) {
      90           0 :                 sock_.close(ec);
      91           0 :                 connected_ = false;
      92           0 : retry:;
      93           0 :                 sock_.connect(server_ep_, ec);
      94           0 :                 if (ec) {
      95           0 :                     sleep(1);
      96           0 :                     goto retry;
      97             :                 }
      98           0 :                 socket_ = sock_.native_handle();
      99           0 :                 connected_ = true;
     100             :             }
     101           0 :             continue;
     102             :         }
     103             : 
     104           0 :         if (remain_ != 0) {
     105           0 :             if (remain_ < sizeof(struct nlmsghdr)) {
     106           0 :                 memcpy((char *)&tnlh, rx_buff_q_, remain_);
     107           0 :                 memcpy(((char *)&tnlh) + remain_, rx_buff_,
     108           0 :                         (sizeof(struct nlmsghdr) - remain_));
     109           0 :                 nlh =  &tnlh;
     110             :             } else {
     111           0 :                 nlh =  (struct nlmsghdr *)rx_buff_q_;
     112             :             }
     113           0 :             if (remain_ > nlh->nlmsg_len)
     114           0 :                 assert(0);
     115           0 :             memcpy(ret_buff, rx_buff_q_, remain_);
     116           0 :             memcpy(ret_buff+remain_, rx_buff_, nlh->nlmsg_len - remain_);
     117           0 :             bufp += (nlh->nlmsg_len - remain_);
     118           0 :             ctxt->SetErrno(0);
     119           0 :             ProcessDataInline(ret_buff);
     120           0 :             offset = nlh->nlmsg_len - remain_;
     121             :         }
     122           0 :         while (offset < bytes_transferred) {
     123           0 :             if ((bytes_transferred - offset) > (sizeof(struct nlmsghdr))) {
     124           0 :                 nlh =  (struct nlmsghdr *)(rx_buff_ + offset);
     125           0 :                 if ((bytes_transferred - offset) > nlh->nlmsg_len) {
     126           0 :                     memcpy(ret_buff, rx_buff_ + offset, nlh->nlmsg_len);
     127           0 :                     ctxt->SetErrno(0);
     128           0 :                     ProcessDataInline(ret_buff);
     129           0 :                     offset += nlh->nlmsg_len;
     130             :                 } else {
     131           0 :                     break;
     132             :                 }
     133             :             } else {
     134           0 :                 break;
     135             :             }
     136             :         }
     137           0 :         memcpy(rx_buff_q_, rx_buff_ + offset, bytes_transferred - offset);
     138           0 :         remain_ = bytes_transferred - offset;
     139           0 :     }
     140             :     return true;
     141             : }
     142             : 
     143           0 : void KSyncSockUds::Init(io_service &ios, const std::string &cpu_pin_policy,
     144             :     const std::string &sockpathvr) {
     145           0 :     KSyncSock::SetSockTableEntry(new KSyncSockUds(ios));
     146           0 :     SetNetlinkFamilyId(10);
     147           0 :     KSyncSock::Init(false, cpu_pin_policy);
     148           0 :     sockpath_ = sockpathvr;
     149           0 : }
     150             : 
     151           0 : uint32_t KSyncSockUds::GetSeqno(char *data) {
     152           0 :     return GetNetlinkSeqno(data);
     153             : }
     154             : 
     155           0 : bool KSyncSockUds::IsMoreData(char *data) {
     156           0 :     return NetlinkMsgDone(data);
     157             : }
     158             : 
     159           0 : bool KSyncSockUds::Decoder(char *data, AgentSandeshContext *context) {
     160           0 :     KSyncSockNetlink::NetlinkDecoder(data, context);
     161           0 :     return true;
     162             : }
     163             : 
     164           0 : bool KSyncSockUds::BulkDecoder(char *data,
     165             :                                KSyncBulkSandeshContext *bulk_sandesh_context) {
     166             :     // Get sandesh buffer and buffer-length
     167           0 :     uint32_t buf_len = 0;
     168           0 :     char *buf = NULL;
     169           0 :     GetNetlinkPayload(data, &buf, &buf_len);
     170           0 :     return bulk_sandesh_context->Decoder(buf, buf_len, NLA_ALIGNTO, IsMoreData(data));
     171             : }
     172             : 
     173           0 : void KSyncSockUds::AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
     174             :                                HandlerCb cb) {
     175           0 :     if (connected_ == true)
     176           0 :         SendTo(iovec, seq_no);
     177           0 : }
     178             : 
     179           0 : size_t KSyncSockUds::SendTo(KSyncBufferList *iovec, uint32_t seq_no) {
     180           0 :     size_t len = 0, ret;
     181             :     struct msghdr msg;
     182           0 :     struct iovec iov[max_bulk_msg_count_*2];
     183             :     int i;
     184             : 
     185           0 :     memset(&msg, 0, sizeof(msg));
     186           0 :     msg.msg_iov = iov;
     187             : 
     188           0 :     ResetNetlink(nl_client_);
     189           0 :     int offset = nl_client_->cl_buf_offset;
     190           0 :     UpdateNetlink(nl_client_, bulk_buf_size_, seq_no);
     191             : 
     192           0 :     KSyncBufferList::iterator it = iovec->begin();
     193           0 :     iovec->insert(it, buffer((char *)nl_client_->cl_buf, offset));
     194             : 
     195           0 :     int count = iovec->size();
     196           0 :     for(i = 0; i < count; i++) {
     197           0 :         mutable_buffers_1 buf = iovec->at(i);
     198           0 :         size_t buf_size = boost::asio::buffer_size(buf);
     199           0 :         void* cbuf = boost::asio::buffer_cast<void*>(buf);
     200           0 :         len += buf_size;
     201           0 :         iov[i].iov_base = cbuf;
     202           0 :         iov[i].iov_len = buf_size;
     203             :     }
     204             : 
     205           0 :     msg.msg_iovlen = i;
     206           0 :     ret = sendmsg(socket_, &msg, 0);
     207           0 :     if (ret != len) {
     208           0 :         LOG(ERROR, "sendmsg failure " << ret << "len " << len);
     209             :     }
     210           0 :     return len;
     211           0 : }
     212             : 
     213           0 : bool KSyncSockUds::Validate(char *data) {
     214           0 :     return true;
     215             : }
     216             : 
     217           0 : void KSyncSockUds::AsyncReceive(mutable_buffers_1 buf, HandlerCb cb) {
     218             :     static int started = 0;
     219           0 :     if (!started) {
     220           0 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
     221             :         //Receive is handled in a separate Run() thread
     222           0 :         KSyncSockUdsReadTask *task = new KSyncSockUdsReadTask(scheduler, this);
     223           0 :         scheduler->Enqueue(task);
     224           0 :         started = 1;
     225             :     }
     226           0 : }
     227             : 
     228           0 : void KSyncSockUds::Receive(mutable_buffers_1 buf) {
     229           0 :     boost::system::error_code ec;
     230           0 :     uint32_t bytes_read = 0;
     231           0 :     const struct nlmsghdr *nlh = NULL;
     232             : 
     233           0 :     char *netlink_header(buffer_cast<char *>(buf));
     234             : 
     235           0 :     while (bytes_read < sizeof(struct nlmsghdr)) {
     236           0 :         char *buffer = netlink_header + bytes_read;
     237           0 :         bytes_read += recv(socket_, buffer, sizeof(struct nlmsghdr) - bytes_read, 0);
     238             :         //Data read is lesser than netlink header
     239             :         //continue reading
     240           0 :         if (bytes_read == sizeof(struct nlmsghdr)) {
     241           0 :             nlh =  buffer_cast<struct nlmsghdr *>(buf);
     242             :         }
     243             :     }
     244             : 
     245           0 :     if (nlh->nlmsg_type == NLMSG_ERROR) {
     246           0 :         LOG(ERROR, "Netlink error for seqno " << nlh->nlmsg_seq
     247             :                 << " len " << nlh->nlmsg_len);
     248           0 :         assert(0);
     249             :     }
     250             : 
     251           0 :     bytes_read = 0;
     252           0 :     uint32_t payload_size = nlh->nlmsg_len - sizeof(struct nlmsghdr);
     253           0 :     char *data(buffer_cast<char *>(buf + sizeof(struct nlmsghdr)));
     254             : 
     255           0 :     while (bytes_read < payload_size) {
     256           0 :         char *buffer = data + bytes_read;
     257           0 :         bytes_read += recv(socket_, buffer, payload_size - bytes_read, 0);
     258             :     }
     259           0 : }