Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2016 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : /*
       7             :  * The primary method implemented here is Send(), to transmit a
       8             :  * message over the Unix socket. It uses boost::asio::async_write to
       9             :  * send one message at a time over the socket, that is transmitted
      10             :  * asynchronously. The user can repeatedly call Send(). All those
      11             :  * buffers are tail-queued. Upon write_complete callback, the next
      12             :  * message from the front of the queue is sent.
      13             :  */
      14             : #include "io/usock_server.h"
      15             : 
      16             : using boost::asio::buffer_cast;
      17             : using boost::asio::buffer;
      18             : using boost::asio::mutable_buffer;
      19             : 
      20           2 : UnixDomainSocketSession::~UnixDomainSocketSession() {
      21           2 :     if (observer_) {
      22           0 :         observer_(this, CLOSE);
      23             :     }
      24             : 
      25             :     /* Free up any remaining buffers in the queue. */
      26           2 :     for (BufferQueue::iterator iter = buffer_queue_.begin();
      27           2 :          iter != buffer_queue_.end(); ++iter) {
      28           0 :         DeleteBuffer(*iter);
      29             :     }
      30           2 :     buffer_queue_.clear();
      31           2 : }
      32             : 
      33           1 : void UnixDomainSocketSession::Start() {
      34           1 :     if (observer_) {
      35           0 :         observer_(this, READY);
      36             :     }
      37             : 
      38           1 :     socket_.async_read_some(boost::asio::buffer(data_),
      39           2 :                             boost::bind(&UnixDomainSocketSession::
      40           2 :                                         HandleRead, shared_from_this(),
      41             :                                         boost::asio::placeholders::error,
      42             :                                         boost::asio::placeholders::
      43             :                                         bytes_transferred));
      44           1 : }
      45             : 
      46           1 : void UnixDomainSocketSession::Send(const uint8_t * data, int data_len) {
      47           1 :     if (!data || !data_len) {
      48           0 :         return;
      49             :     }
      50           1 :     bool write_now = buffer_queue_.empty();
      51           1 :     AppendBuffer(data, data_len);
      52           1 :     if (write_now) {
      53           1 :         WriteToSocket();
      54             :     }
      55             : }
      56             : 
      57           2 : void UnixDomainSocketSession::WriteToSocket() {
      58           2 :     if (buffer_queue_.empty()) {
      59           1 :         return;
      60             :     }
      61             : 
      62           1 :     boost::asio::mutable_buffer head = buffer_queue_.front();
      63           1 :     boost::asio::async_write(socket_,
      64           1 :                              buffer(buffer_cast <const uint8_t *>(head),
      65             :                                     boost::asio::buffer_size(head)),
      66           2 :                              boost::bind(&UnixDomainSocketSession::
      67           2 :                                          HandleWrite, shared_from_this(),
      68             :                                          boost::asio::placeholders::error));
      69             : }
      70             : 
      71           1 : void UnixDomainSocketSession::AppendBuffer(const uint8_t *src, int bytes) {
      72           1 :     u_int8_t *data = new u_int8_t[bytes];
      73           1 :     memcpy(data, src, bytes);
      74             :     boost::asio::mutable_buffer buffer =
      75           1 :         boost::asio::mutable_buffer(data, bytes);
      76           1 :     buffer_queue_.push_back(buffer);
      77           1 : }
      78             : 
      79           1 : void UnixDomainSocketSession::DeleteBuffer(boost::asio::mutable_buffer buffer) {
      80           1 :     const uint8_t *data = buffer_cast <const uint8_t *>(buffer);
      81           1 :     delete []data;
      82           1 :     return;
      83             : }
      84             : 
      85           1 : void UnixDomainSocketSession::HandleRead(const boost::system::error_code &error,
      86             :                                          size_t bytes_transferred) {
      87           1 :     if (error) {
      88           1 :         return;
      89             :     }
      90           0 :     if (observer_) {
      91           0 :         observer_(this, READY);
      92             :     }
      93             : }
      94             : 
      95           1 : void UnixDomainSocketSession::HandleWrite(
      96             :         const boost::system::error_code &error) {
      97             :     /*
      98             :      * async_write() is atomic in that it returns success once the entire message
      99             :      * is sent. If there is an error, it's okay to return from here so that the
     100             :      * session gets closed.
     101             :      */
     102           1 :     if (error) {
     103           0 :         return;
     104             :     }
     105             : 
     106             :     /*
     107             :      * We are done with the buffer at the head of the queue. Delete it.
     108             :      */
     109           1 :     DeleteBuffer(buffer_queue_.front());
     110           1 :     buffer_queue_.pop_front();
     111             : 
     112             :     /*
     113             :      * Write the next message, if there.
     114             :      */
     115           1 :     WriteToSocket();
     116             : 
     117             :     /*
     118             :      * Engage on the socket to keep it alive.
     119             :      */
     120           1 :     socket_.async_read_some(boost::asio::buffer(data_),
     121           2 :                             boost::bind(&UnixDomainSocketSession::
     122           2 :                                         HandleRead, shared_from_this(),
     123             :                                         boost::asio::placeholders::error,
     124             :                                         boost::asio::placeholders::
     125             :                                         bytes_transferred));
     126             : }
     127             : 
     128           1 : UnixDomainSocketServer::UnixDomainSocketServer(
     129           1 :         boost::asio::io_context *io, const std::string &file)
     130           1 :   : io_service_(io),
     131           1 :     acceptor_(*io, boost::asio::local::stream_protocol::endpoint(file)),
     132           1 :     session_idspace_(0) {
     133           1 :     SessionPtr new_session(new UnixDomainSocketSession(io_service_));
     134           1 :     acceptor_.async_accept(new_session->socket(),
     135           2 :                            boost::bind(&UnixDomainSocketServer::
     136             :                                        HandleAccept, this, new_session,
     137             :                                        boost::asio::placeholders::error));
     138           1 : }
     139             : 
     140             : void
     141           1 : UnixDomainSocketServer::HandleAccept(SessionPtr session,
     142             :                                      const boost::system::error_code &error) {
     143           1 :     UnixDomainSocketSession *socket_session = session.get();
     144             : 
     145           1 :     if (error) {
     146           0 :         if (observer_) {
     147           0 :             observer_(this, socket_session, DELETE_SESSION);
     148             :         }
     149           0 :         return;
     150             :     }
     151             : 
     152           1 :     socket_session->set_session_id(++session_idspace_);
     153           1 :     if (observer_) {
     154           1 :         observer_(this, socket_session, NEW_SESSION);
     155           1 :         session->Start();
     156             :     }
     157             : 
     158           1 :     SessionPtr new_session(new UnixDomainSocketSession(io_service_));
     159           1 :     acceptor_.async_accept(new_session->socket(),
     160           2 :                            boost::bind(&UnixDomainSocketServer::
     161             :                                        HandleAccept, this, new_session,
     162             :                                        boost::asio::placeholders::error));
     163           1 : }