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             : 
      11             : using namespace boost::asio;
      12             : using namespace boost::placeholders;
      13             : 
      14             : /////////////////////////////////////////////////////////////////////////////
      15             : // KSyncSockTcp routines
      16             : /////////////////////////////////////////////////////////////////////////////
      17             : //TCP socket class for interacting with vrouter
      18           0 : KSyncSockTcp::KSyncSockTcp(EventManager *evm,
      19           0 :     boost::asio::ip::address ip_address, int port) : TcpServer(evm), evm_(evm),
      20           0 :     session_(NULL), server_ep_(ip_address, port), connect_complete_(false) {
      21             : 
      22           0 :     reset_use_wait_tree();
      23           0 :     set_process_data_inline();
      24           0 :     if (rx_buff_ == NULL) {
      25           0 :         rx_buff_ = new char[kBufLen];
      26             :     }
      27           0 :     rx_buff_rem_ = new char[kBufLen];
      28           0 :     remain_ = 0;
      29             : 
      30           0 :     session_ = CreateSession();
      31           0 :     Connect(session_, server_ep_);
      32           0 : }
      33             : 
      34           0 : void KSyncSockTcp::Init(EventManager *evm, boost::asio::ip::address ip_addr,
      35             :                         int port, const std::string &cpu_pin_policy) {
      36           0 :     KSyncSock::SetSockTableEntry(new KSyncSockTcp(evm, ip_addr, port));
      37           0 :     SetNetlinkFamilyId(10);
      38           0 :     KSyncSock::Init(false, cpu_pin_policy);
      39           0 : }
      40             : 
      41           0 : TcpSession* KSyncSockTcp::AllocSession(Socket *socket) {
      42           0 :     TcpSession *session = new KSyncSockTcpSession(this, socket, false);
      43           0 :     session->set_observer(boost::bind(&KSyncSockTcp::OnSessionEvent,
      44             :                                       this, _1, _2));
      45           0 :     return session;
      46             : }
      47             : 
      48           0 : uint32_t KSyncSockTcp::GetSeqno(char *data) {
      49           0 :     return GetNetlinkSeqno(data);
      50             : }
      51             : 
      52           0 : bool KSyncSockTcp::IsMoreData(char *data) {
      53           0 :     return NetlinkMsgDone(data);
      54             : }
      55             : 
      56           0 : size_t KSyncSockTcp::SendTo(KSyncBufferList *iovec, uint32_t seq_no) {
      57           0 :     size_t len = 0, ret;
      58             :     struct msghdr msg;
      59           0 :     struct iovec iov[max_bulk_msg_count_*2];
      60             :     int i, fd;
      61             : 
      62           0 :     memset(&msg, 0, sizeof(msg));
      63           0 :     msg.msg_iov = iov;
      64             : 
      65           0 :     ResetNetlink(nl_client_);
      66           0 :     int offset = nl_client_->cl_buf_offset;
      67           0 :     UpdateNetlink(nl_client_, bulk_buf_size_, seq_no);
      68             : 
      69           0 :     KSyncBufferList::iterator it = iovec->begin();
      70           0 :     iovec->insert(it, buffer((char *)nl_client_->cl_buf, offset));
      71             : 
      72           0 :     int count = iovec->size();
      73           0 :     for(i = 0; i < count; i++) {
      74           0 :         mutable_buffers_1 buf = iovec->at(i);
      75           0 :         size_t buf_size = boost::asio::buffer_size(buf);
      76           0 :         void* cbuf = boost::asio::buffer_cast<void*>(buf);
      77           0 :         len += buf_size;
      78           0 :         iov[i].iov_base = cbuf;
      79           0 :         iov[i].iov_len = buf_size;
      80             :     }
      81             : 
      82           0 :     msg.msg_iovlen = i;
      83           0 :     fd = tcp_socket_->native_handle();
      84           0 :     ret = sendmsg(fd, &msg, 0);
      85           0 :     if (ret != len) {
      86           0 :         LOG(ERROR, "sendmsg failure " << ret << "len " << len);
      87             :     }
      88           0 :     return len;
      89           0 : }
      90             : 
      91           0 : void KSyncSockTcp::AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
      92             :                                HandlerCb cb) {
      93           0 :     SendTo(iovec, seq_no);
      94           0 :     return;
      95             : }
      96             : 
      97           0 : bool KSyncSockTcp::Validate(char *data) {
      98           0 :     return ValidateNetlink(data);
      99             : }
     100             : 
     101           0 : bool KSyncSockTcp::Decoder(char *data, AgentSandeshContext *context) {
     102           0 :     KSyncSockNetlink::NetlinkDecoder(data, context);
     103           0 :     return true;
     104             : }
     105             : 
     106           0 : bool KSyncSockTcp::BulkDecoder(char *data,
     107             :                                KSyncBulkSandeshContext *bulk_sandesh_context) {
     108             :     // Get sandesh buffer and buffer-length
     109           0 :     uint32_t buf_len = 0;
     110           0 :     char *buf = NULL;
     111           0 :     GetNetlinkPayload(data, &buf, &buf_len);
     112           0 :     return bulk_sandesh_context->Decoder(buf, buf_len, NLA_ALIGNTO, IsMoreData(data));
     113             : }
     114             : 
     115           0 : void KSyncSockTcp::AsyncReceive(mutable_buffers_1 buf, HandlerCb cb) {
     116             :     //Data would be read from ksync tcp session
     117             :     //hence no socket operation would be required
     118           0 : }
     119             : 
     120           0 : void KSyncSockTcp::Receive(mutable_buffers_1 buf) {
     121           0 :     uint32_t bytes_read = 0;
     122           0 :     boost::system::error_code ec;
     123           0 :     const struct nlmsghdr *nlh = NULL;
     124             : 
     125             :     //Create a buffer to read netlink header first
     126             :     mutable_buffers_1 netlink_header(buffer_cast<void *>(buf),
     127           0 :                                      sizeof(struct nlmsghdr));
     128             : 
     129           0 :     bool blocking_socket = session_->socket()->non_blocking();
     130           0 :     session_->socket()->non_blocking(false, ec);
     131           0 :     while (bytes_read < sizeof(struct nlmsghdr)) {
     132             :         mutable_buffers_1 buffer =
     133           0 :             static_cast<mutable_buffers_1>(netlink_header + bytes_read);
     134           0 :         bytes_read += session_->socket()->receive(buffer, 0, ec);
     135           0 :         if (ec.failed()) {
     136           0 :             assert(0);
     137             :         }
     138             :         //Data read is lesser than netlink header
     139             :         //continue reading
     140           0 :         if (bytes_read == sizeof(struct nlmsghdr)) {
     141           0 :             nlh =  buffer_cast<struct nlmsghdr *>(buf);
     142             :         }
     143             :     }
     144             : 
     145           0 :     if (nlh->nlmsg_type == NLMSG_ERROR) {
     146           0 :         LOG(ERROR, "Netlink error for seqno " << nlh->nlmsg_seq
     147             :                 << " len " << nlh->nlmsg_len);
     148           0 :         assert(0);
     149             :     }
     150             : 
     151           0 :     bytes_read = 0;
     152           0 :     uint32_t payload_size = nlh->nlmsg_len - sizeof(struct nlmsghdr);
     153             :     //Read data
     154           0 :     mutable_buffers_1 data(buffer_cast<void *>(buf + sizeof(struct nlmsghdr)),
     155           0 :                            payload_size);
     156             : 
     157           0 :     while (bytes_read < payload_size) {
     158             :         mutable_buffers_1 buffer =
     159           0 :             static_cast<mutable_buffers_1>(data + bytes_read);
     160           0 :         bytes_read += session_->socket()->receive(buffer, 0, ec);
     161           0 :         if (ec.failed()) {
     162           0 :             assert(0);
     163             :         }
     164             :     }
     165           0 :     session_->socket()->non_blocking(blocking_socket, ec);
     166           0 : }
     167             : 
     168           0 : bool KSyncSockTcp::ReceiveMsg(const u_int8_t *msg, size_t size) {
     169           0 :     AgentSandeshContext *ctxt = KSyncSock::GetAgentSandeshContext(0);
     170           0 :     ctxt->SetErrno(0);
     171           0 :     ProcessDataInline((char *) msg);
     172           0 :     return true;
     173             : }
     174             : 
     175           0 : bool KSyncSockTcp::Run() {
     176           0 :     AgentSandeshContext *ctxt = KSyncSock::GetAgentSandeshContext(0);
     177           0 :     int fd = tcp_socket_->native_handle();
     178             : 
     179             :     while (1) {
     180           0 :         char *bufp = rx_buff_;
     181           0 :         struct nlmsghdr *nlh = NULL;
     182             :         struct nlmsghdr tnlh;
     183           0 :         int offset = 0;
     184           0 :         int bytes_transferred = 0;
     185             : 
     186           0 :         bytes_transferred = recv(fd, rx_buff_, kBufLen, 0);
     187           0 :         if (bytes_transferred <= 0) {
     188           0 :              LOG(ERROR, "Connection to dpdk-vrouter lost.");
     189           0 :              sleep(10);
     190           0 :              exit(EXIT_FAILURE);
     191             :         }
     192             : 
     193           0 :         if (remain_ != 0) {
     194           0 :             if (remain_ < sizeof(struct nlmsghdr)) {
     195           0 :                 memcpy((char *)&tnlh, rx_buff_rem_, remain_);
     196           0 :                 memcpy(((char *)&tnlh) + remain_, rx_buff_,
     197           0 :                         (sizeof(struct nlmsghdr) - remain_));
     198           0 :                 nlh =  &tnlh;
     199             :             } else {
     200           0 :                 nlh =  (struct nlmsghdr *)rx_buff_rem_;
     201             :             }
     202             : 
     203           0 :             if (remain_ > nlh->nlmsg_len)
     204           0 :                 assert(0);
     205             : 
     206           0 :             memcpy(rx_buff_rem_+remain_, rx_buff_, nlh->nlmsg_len - remain_);
     207           0 :             bufp += (nlh->nlmsg_len - remain_);
     208           0 :             ctxt->SetErrno(0);
     209           0 :             ProcessDataInline(rx_buff_rem_);
     210           0 :             offset = nlh->nlmsg_len - remain_;
     211             :         }
     212             : 
     213           0 :         while (offset < bytes_transferred) {
     214           0 :             if ((unsigned int)(bytes_transferred - offset) >= (sizeof(struct nlmsghdr))) {
     215           0 :                 nlh =  (struct nlmsghdr *)(rx_buff_ + offset);
     216           0 :                 if ((unsigned int)(bytes_transferred - offset) >= nlh->nlmsg_len) {
     217           0 :                     ctxt->SetErrno(0);
     218           0 :                     ProcessDataInline(rx_buff_ + offset);
     219           0 :                     offset += nlh->nlmsg_len;
     220             :                 } else {
     221           0 :                     break;
     222             :                 }
     223             :             } else {
     224           0 :                 break;
     225             :             }
     226             :         }
     227             : 
     228           0 :         remain_ = bytes_transferred - offset;
     229           0 :         if (remain_) {
     230           0 :             memcpy(rx_buff_rem_, rx_buff_ + offset, bytes_transferred - offset);
     231             :         }
     232           0 :     }
     233             : 
     234             :     return true;
     235             : }
     236             : 
     237             : class KSyncSockTcpReadTask : public Task {
     238             : public:
     239           0 :     KSyncSockTcpReadTask(TaskScheduler *scheduler, KSyncSockTcp *sock) :
     240           0 :         Task(scheduler->GetTaskId("Ksync::KSyncSockTcpRead"), 0), sock_(sock) {
     241           0 :     }
     242           0 :     ~KSyncSockTcpReadTask() {
     243           0 :     }
     244             : 
     245           0 :     bool Run() {
     246           0 :         sock_->Run();
     247           0 :         return true;
     248             :     }
     249           0 :     std::string Description() const { return "KSyncSockTcpRead"; }
     250             : private:
     251             :     KSyncSockTcp *sock_;
     252             : 
     253             : };
     254             : 
     255           0 : void KSyncSockTcp::AsyncReadStart() {
     256             :     static int started = 0;
     257           0 :     boost::system::error_code ec;
     258             : 
     259           0 :     if (!started) {
     260           0 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
     261           0 :         KSyncSockTcpReadTask *task = new KSyncSockTcpReadTask(scheduler, this);
     262           0 :         tcp_socket_->non_blocking(false, ec);
     263           0 :         scheduler->Enqueue(task);
     264           0 :         started = 1;
     265             :     }
     266           0 : }
     267             : 
     268           0 : void KSyncSockTcp::OnSessionEvent(TcpSession *session,
     269             :                                   TcpSession::Event event) {
     270           0 :     switch (event) {
     271           0 :     case TcpSession::CONNECT_FAILED:
     272             :         //Retry
     273           0 :         Connect(session_, server_ep_);
     274           0 :         break;
     275           0 :     case TcpSession::CLOSE:
     276           0 :         LOG(ERROR, "Connection to dpdk-vrouter lost.");
     277           0 :         sleep(10);
     278           0 :         exit(EXIT_FAILURE);
     279             :         break;
     280           0 :     case TcpSession::CONNECT_COMPLETE:
     281           0 :         tcp_socket_ = session_->socket();
     282           0 :         connect_complete_ = true;
     283           0 :         session_->SetTcpNoDelay();
     284           0 :         session_->SetTcpSendBufSize(max_bulk_buf_size_*16);
     285           0 :         session_->SetTcpRecvBufSize(max_bulk_buf_size_*16);
     286           0 :     default:
     287           0 :         break;
     288             :     }
     289           0 : }
     290             : 
     291             : /////////////////////////////////////////////////////////////////////////////
     292             : // KSyncSockTcpSession routines
     293             : /////////////////////////////////////////////////////////////////////////////
     294           0 : KSyncSockTcpSession::KSyncSockTcpSession(TcpServer *server, Socket *sock,
     295           0 :     bool async_ready) : TcpSession(server, sock, async_ready) {
     296           0 :     KSyncSockTcp *tcp_ptr = static_cast<KSyncSockTcp *>(server);
     297           0 :     reader_ = new KSyncSockTcpSessionReader(this,
     298           0 :                        boost::bind(&KSyncSockTcp::ReceiveMsg, tcp_ptr, _1, _2));
     299           0 : }
     300             : 
     301           0 : KSyncSockTcpSession::~KSyncSockTcpSession() {
     302           0 :     if (reader_) {
     303           0 :         delete reader_;
     304             :     }
     305           0 : }
     306             : 
     307           0 : void KSyncSockTcpSession::OnRead(Buffer buffer) {
     308           0 :     reader_->OnRead(buffer);
     309           0 : }
     310             : 
     311           0 : KSyncSockTcpSessionReader::KSyncSockTcpSessionReader(
     312           0 :     TcpSession *session, ReceiveCallback callback) :
     313           0 :     TcpMessageReader(session, callback) {
     314           0 : }
     315             : 
     316           0 : int KSyncSockTcpSessionReader::MsgLength(Buffer buffer, int offset) {
     317           0 :     size_t size = TcpSession::BufferSize(buffer);
     318           0 :     int remain = size - offset;
     319           0 :     if (remain < GetHeaderLenSize()) {
     320           0 :         return -1;
     321             :     }
     322             : 
     323             :     //Byte ordering?
     324             :     const struct nlmsghdr *nlh =
     325           0 :         (const struct nlmsghdr *)(TcpSession::BufferData(buffer) + offset);
     326           0 :     return nlh->nlmsg_len;
     327             : }