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             : #ifndef SRC_IO_UDP_SERVER_H_
       7             : #define SRC_IO_UDP_SERVER_H_
       8             : 
       9             : #include <atomic>
      10             : #include <string>
      11             : #include <vector>
      12             : #include <mutex>
      13             : 
      14             : #include <boost/asio.hpp>
      15             : #include <boost/intrusive_ptr.hpp>
      16             : #include "io/event_manager.h"
      17             : #include "io/server_manager.h"
      18             : #include "io/io_utils.h"
      19             : 
      20             : class SocketIOStats;
      21             : 
      22             : class UdpServer {
      23             : public:
      24             :     typedef boost::asio::ip::udp::endpoint Endpoint;
      25             :     typedef boost::asio::ip::udp::socket Socket;
      26             : 
      27             :     enum ServerState {
      28             :         OK = 42,
      29             :         Uninitialized,
      30             :         SocketOpenFailed,
      31             :         SocketBindFailed,
      32             :     };
      33             :     static const int kDefaultBufferSize = 4 * 1024;
      34             : 
      35             :     explicit UdpServer(EventManager *evm, int buffer_size = kDefaultBufferSize);
      36             :     explicit UdpServer(boost::asio::io_context *io_service,
      37             :                        int buffer_size = kDefaultBufferSize);
      38             :     virtual ~UdpServer();
      39             : 
      40             :     virtual bool Initialize(unsigned short port);
      41             :     virtual bool Initialize(const std::string &ipaddress, unsigned short port);
      42             :     virtual bool Initialize(boost::asio::ip::udp::endpoint local_endpoint);
      43             :     virtual void Shutdown();
      44             : 
      45             :     // tx-rx
      46             :     // Assumes mutex is locked or called from the main thread
      47             :     void StartSend(boost::asio::ip::udp::endpoint ep, std::size_t bytes_to_send,
      48             :             boost::asio::const_buffer buffer);
      49             :     // Assumes mutex is locked or called from the main thread
      50             :     void StartReceive();
      51             :     // state
      52          21 :     ServerState GetServerState() const { return state_; }
      53             :     boost::asio::ip::udp::endpoint GetLocalEndpoint(
      54             :         boost::system::error_code *error) const;
      55             :     std::string GetLocalEndpointAddress() const;
      56             :     int GetLocalEndpointPort() const;
      57             : 
      58             :     // buffers
      59             :     // Assumes mutex is locked or called from the main thread
      60             :     boost::asio::mutable_buffer AllocateBuffer();
      61             :     boost::asio::mutable_buffer AllocateBuffer(std::size_t s);
      62             :     // Assumes mutex is locked or called from the main thread
      63             :     void DeallocateBuffer(const boost::asio::const_buffer &buffer);
      64             : 
      65             :     // statistics
      66             :     const io::SocketStats &GetSocketStats() const { return stats_; }
      67             :     void GetRxSocketStats(SocketIOStats *socket_stats) const;
      68             :     void GetTxSocketStats(SocketIOStats *socket_stats) const;
      69             : 
      70             : protected:
      71             :     EventManager *event_manager() { return evm_; }
      72          14 :     virtual bool DisableSandeshLogMessages() { return false; }
      73          28 :     virtual std::string ToString() { return name_; }
      74             : 
      75             :     // Assumes mutex is locked, deallocates the buffer
      76             :     virtual void HandleReceive(
      77             :             const boost::asio::const_buffer &recv_buffer,
      78             :             boost::asio::ip::udp::endpoint remote_endpoint,
      79             :             std::size_t bytes_transferred,
      80             :             const boost::system::error_code& error);
      81             : 
      82             :     virtual void OnRead(const boost::asio::const_buffer &recv_buffer,
      83             :         const boost::asio::ip::udp::endpoint &remote_endpoint);
      84             : 
      85           2 :     virtual int reader_task_id() const {
      86           2 :         return reader_task_id_;
      87             :     }
      88             :     // This function returns the instance to run ReaderTask.
      89             :     // Returning Task::kTaskInstanceAny would allow multiple reader tasks to
      90             :     // run in parallel.
      91             :     // Derived class may override implementation if it expects that all the
      92             :     // tasks of the reader to run in specific instance
      93             :     // Note: Two tasks of same task ID and task instance can't run
      94             :     //       at in parallel
      95             :     // E.g. User can create task per remote endpoint to ensure that
      96             :     //      there is one ReaderTask per remote endpoint
      97             :     virtual int reader_task_instance(
      98             :         const boost::asio::ip::udp::endpoint &remote_endpoint) const;
      99             : 
     100             :     // Assumes mutex is locked, deallocates the buffer
     101             :     virtual void HandleSend(boost::asio::const_buffer send_buffer,
     102             :             boost::asio::ip::udp::endpoint remote_endpoint,
     103             :             std::size_t bytes_transferred,
     104             :             const boost::system::error_code& error);
     105             : 
     106             : private:
     107             :     class Reader;
     108             :     friend void intrusive_ptr_add_ref(UdpServer *server);
     109             :     friend void intrusive_ptr_release(UdpServer *server);
     110             :     void SetName(boost::asio::ip::udp::endpoint ep);
     111             : 
     112             :     // Locks the mutex
     113             :     void HandleReceiveInternal(
     114             :             boost::asio::const_buffer recv_buffer,
     115             :             std::size_t bytes_transferred,
     116             :             const boost::system::error_code& error);
     117             : 
     118             :     // Locks the mutex
     119             :     void HandleSendInternal(boost::asio::const_buffer send_buffer,
     120             :             boost::asio::ip::udp::endpoint remote_endpoint,
     121             :             std::size_t bytes_transferred,
     122             :             const boost::system::error_code& error);
     123             : 
     124             :     static int reader_task_id_;
     125             :     boost::asio::ip::udp::socket socket_;
     126             :     int buffer_size_;
     127             :     ServerState state_;
     128             :     EventManager *evm_;
     129             :     std::string name_;
     130             :     boost::asio::ip::udp::endpoint remote_endpoint_;
     131             :     std::mutex state_guard_;
     132             :     std::mutex pbuf_guard_;
     133             :     std::vector<uint8_t *> pbuf_;
     134             :     std::atomic<int> refcount_;
     135             :     io::SocketStats stats_;
     136             : 
     137             :     DISALLOW_COPY_AND_ASSIGN(UdpServer);
     138             : };
     139             : 
     140             : typedef boost::intrusive_ptr<UdpServer> UdpServerPtr;
     141             : 
     142         107 : inline void intrusive_ptr_add_ref(UdpServer *server) {
     143         107 :     server->refcount_.fetch_add(1);
     144         107 : }
     145             : 
     146         107 : inline void intrusive_ptr_release(UdpServer *server) {
     147         107 :     int prev = server->refcount_.fetch_sub(1);
     148         107 :     if (prev == 1) {
     149           6 :         delete server;
     150             :     }
     151         107 : }
     152             : 
     153             : class UdpServerManager {
     154             : public:
     155             :     static void AddServer(UdpServer *server);
     156             :     static void DeleteServer(UdpServer *server);
     157             : 
     158             : private:
     159             :     static ServerManager<UdpServer, UdpServerPtr> impl_;
     160             : };
     161             : 
     162             : #endif  // SRC_IO_UDP_SERVER_H_