Line data    Source code

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