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             : #include "io/udp_server.h"
       7             : 
       8             : #include <boost/bind/bind.hpp>
       9             : 
      10             : #include "base/logging.h"
      11             : #include "base/address_util.h"
      12             : #include "io/io_log.h"
      13             : #include "io/io_utils.h"
      14             : 
      15             : using boost::asio::buffer_cast;
      16             : using boost::asio::mutable_buffer;
      17             : using boost::asio::mutable_buffers_1;
      18             : using boost::asio::const_buffer;
      19             : using boost::asio::ip::udp;
      20             : using namespace boost::placeholders;
      21             : 
      22             : int UdpServer::reader_task_id_ = -1;
      23             : 
      24             : class UdpServer::Reader : public Task {
      25             : public:
      26           2 :     Reader(UdpServerPtr server, const udp::endpoint &remote_endpoint,
      27             :         const const_buffer &buffer)
      28           6 :         : Task(server->reader_task_id(),
      29           4 :                server->reader_task_instance(remote_endpoint)),
      30           2 :         server_(server),
      31           2 :         remote_endpoint_(remote_endpoint),
      32           2 :         buffer_(buffer) {
      33           2 :     }
      34             : 
      35           2 :     virtual bool Run() {
      36           2 :         std::scoped_lock lock(server_->state_guard_);
      37           2 :         if (server_->state_ == OK) {
      38           2 :             server_->OnRead(buffer_, remote_endpoint_);
      39           2 :             server_->DeallocateBuffer(buffer_);
      40             :         }
      41           2 :         return true;
      42           2 :     }
      43           0 :     std::string Description() const { return "UdpServer::Reader"; }
      44             : 
      45             : private:
      46             :     UdpServerPtr server_;
      47             :     udp::endpoint remote_endpoint_;
      48             :     const_buffer buffer_;
      49             : };
      50             : 
      51           3 : UdpServer::UdpServer(boost::asio::io_context *io_service, int buffer_size):
      52           3 :     socket_(*io_service),
      53           3 :     buffer_size_(buffer_size),
      54           3 :     state_(Uninitialized),
      55           3 :     evm_(NULL) {
      56           3 :     if (reader_task_id_ == -1) {
      57           0 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
      58           0 :         reader_task_id_ = scheduler->GetTaskId("io::udp::ReaderTask");
      59             :     }
      60           3 :     refcount_ = 0;
      61           3 :     UdpServerManager::AddServer(this);
      62           3 : }
      63             : 
      64           3 : UdpServer::UdpServer(EventManager *evm, int buffer_size):
      65           3 :     socket_(*(evm->io_service())),
      66           3 :     buffer_size_(buffer_size),
      67           3 :     state_(Uninitialized),
      68           6 :     evm_(evm) {
      69           3 :     if (reader_task_id_ == -1) {
      70           2 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
      71           2 :         reader_task_id_ = scheduler->GetTaskId("io::udp::ReaderTask");
      72             :     }
      73           3 :     refcount_ = 0;
      74           3 :     UdpServerManager::AddServer(this);
      75           3 : }
      76             : 
      77           2 : int UdpServer::reader_task_instance(const udp::endpoint &rep) const {
      78           2 :     return Task::kTaskInstanceAny;
      79             : }
      80             : 
      81           6 : void UdpServer::SetName(udp::endpoint ep) {
      82           6 :     std::ostringstream s;
      83           6 :     boost::system::error_code ec;
      84           6 :     s << "Udpsocket@" << ep;
      85           6 :     name_ = s.str();
      86           6 : }
      87             : 
      88           7 : UdpServer::~UdpServer() {
      89             :     {
      90           6 :         std::scoped_lock lock(state_guard_);
      91           6 :         assert(state_ == Uninitialized || state_ == SocketOpenFailed ||
      92             :             state_ == SocketBindFailed);
      93           6 :     }
      94             :     {
      95           6 :         std::scoped_lock lock(pbuf_guard_);
      96           6 :         assert(pbuf_.empty());
      97           6 :     }
      98           7 : }
      99             : 
     100           6 : void UdpServer::Shutdown() {
     101           6 :     std::scoped_lock lock(state_guard_);
     102             :     {
     103           6 :         std::scoped_lock lock_pbuf(pbuf_guard_);
     104          10 :         while (!pbuf_.empty()) {
     105           4 :             delete[] pbuf_.back();
     106           4 :             pbuf_.pop_back();
     107             :         }
     108           6 :     }
     109           6 :     if (socket_.is_open()) {
     110           6 :         boost::system::error_code ec;
     111           6 :         socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
     112           6 :         if (ec) {
     113           6 :             UDP_SERVER_LOG_ERROR(this, UDP_DIR_NA,
     114             :                 "ERROR shutdown UDP socket: " << ec);
     115             :         }
     116           6 :         socket_.close(ec);
     117           6 :         if (ec) {
     118           0 :             UDP_SERVER_LOG_ERROR(this, UDP_DIR_NA,
     119             :                 "ERROR closing UDP socket: " << ec);
     120             :         }
     121             :     }
     122           6 :     state_ = Uninitialized;
     123           6 : }
     124             : 
     125           1 : bool UdpServer::Initialize(const std::string &ipaddress, unsigned short port) {
     126           1 :     boost::system::error_code error;
     127           1 :     boost::asio::ip::address ip = AddressFromString(ipaddress, &error);
     128           1 :     if (!error) {
     129           1 :         udp::endpoint local_endpoint = udp::endpoint(ip, port);
     130           1 :         return Initialize(local_endpoint);
     131             :     } else {
     132           0 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_NA, "IP address conversion: "
     133             :             << ipaddress << ": " << error);
     134           0 :         return false;
     135             :     }
     136             : }
     137             : 
     138           6 : bool UdpServer::Initialize(unsigned short port) {
     139           6 :     udp::endpoint local_endpoint = udp::endpoint(udp::v4(), port);
     140           6 :     return Initialize(local_endpoint);
     141             : }
     142             : 
     143           7 : bool UdpServer::Initialize(udp::endpoint local_endpoint) {
     144           7 :     if (GetServerState() != Uninitialized) {
     145           1 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_NA,
     146             :             "Initialize UDP server in WRONG state: " << state_);
     147           1 :         return false;
     148             :     }
     149           6 :     boost::system::error_code error;
     150           6 :     socket_.open(udp::v4(), error);
     151           6 :     if (error) {
     152           0 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_NA, "UDP socket open FAILED: " <<
     153             :             error.message());
     154           0 :         state_ = SocketOpenFailed;
     155           0 :         return false;
     156             :     }
     157           6 :     socket_.bind(local_endpoint, error);
     158           6 :     if (error) {
     159           0 :         boost::system::error_code ec;
     160           0 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_NA, "UDP socket bind FAILED: "
     161             :             << error.message() << ":" << socket_.local_endpoint(ec));
     162           0 :         state_ = SocketBindFailed;
     163           0 :         socket_.close(ec);
     164           0 :         return false;
     165             :     }
     166           6 :     SetName(local_endpoint);
     167           6 :     state_ = OK;
     168           6 :     return true;
     169             : }
     170             : 
     171          12 : mutable_buffer UdpServer::AllocateBuffer(std::size_t s) {
     172          12 :     uint8_t *p = new uint8_t[s];
     173             :     {
     174          12 :         std::scoped_lock lock(pbuf_guard_);
     175          12 :         pbuf_.push_back(p);
     176          12 :     }
     177          12 :     return mutable_buffer(p, s);
     178             : }
     179             : 
     180           9 : mutable_buffer UdpServer::AllocateBuffer() {
     181           9 :     return AllocateBuffer(buffer_size_);
     182             : }
     183             : 
     184           8 : void UdpServer::DeallocateBuffer(const const_buffer &buffer) {
     185           8 :     const uint8_t *p = buffer_cast<const uint8_t *>(buffer);
     186             :     {
     187           8 :         std::scoped_lock lock(pbuf_guard_);
     188           8 :         std::vector<uint8_t *>::iterator f = std::find(pbuf_.begin(),
     189             :             pbuf_.end(), p);
     190           8 :         if (f != pbuf_.end())
     191           8 :             pbuf_.erase(f);
     192           8 :     }
     193           8 :     delete[] p;
     194           8 : }
     195             : 
     196           4 : void UdpServer::StartSend(udp::endpoint ep, std::size_t bytes_to_send,
     197             :         const_buffer buffer) {
     198           4 :     if (state_ == OK) {
     199           3 :         socket_.async_send_to(boost::asio::buffer(buffer), ep,
     200           6 :             boost::bind(&UdpServer::HandleSendInternal, UdpServerPtr(this),
     201             :             buffer, ep,
     202             :             boost::asio::placeholders::bytes_transferred,
     203             :             boost::asio::placeholders::error));
     204             :     } else {
     205           1 :         stats_.write_errors++;
     206           1 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_NA,
     207             :             "StartSend UDP server in WRONG state: " << state_);
     208           1 :         DeallocateBuffer(buffer);
     209             :     }
     210           4 : }
     211             : 
     212           3 : void UdpServer::HandleSendInternal(const const_buffer send_buffer,
     213             :         udp::endpoint remote_endpoint, std::size_t bytes_transferred,
     214             :         const boost::system::error_code& error) {
     215           3 :     std::scoped_lock lock(state_guard_);
     216           3 :     if (state_ != OK) {
     217           0 :         stats_.write_errors++;
     218           0 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_OUT,
     219             :             "Send UDP server in WRONG state: " << state_);
     220           0 :         return;
     221             :     }
     222           3 :     if (error) {
     223           0 :         stats_.write_errors++;
     224           0 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_OUT,
     225             :             "Send to " << remote_endpoint << " FAILED due to error: " <<
     226             :             error.value() << " : " << error.message());
     227           0 :         DeallocateBuffer(send_buffer);
     228           0 :         return;
     229             :     }
     230             :     // Update write statistics.
     231           3 :     stats_.write_calls++;
     232           3 :     stats_.write_bytes += bytes_transferred;
     233             :     // Call the handler
     234           3 :     HandleSend(send_buffer, remote_endpoint, bytes_transferred, error);
     235           3 : }
     236             : 
     237           9 : void UdpServer::StartReceive() {
     238           9 :     if (state_ == OK) {
     239           8 :         mutable_buffer b(AllocateBuffer());
     240           8 :         const_buffer buffer(buffer_cast<const uint8_t*>(b), buffer_size(b));
     241           8 :         socket_.async_receive_from(mutable_buffers_1(b),
     242          16 :             remote_endpoint_, boost::bind(&UdpServer::HandleReceiveInternal,
     243          16 :             UdpServerPtr(this), buffer,
     244             :             boost::asio::placeholders::bytes_transferred,
     245             :             boost::asio::placeholders::error));
     246             :     } else {
     247           1 :         stats_.read_errors++;
     248           1 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_NA,
     249             :             "StartReceive UDP server in WRONG state: " << state_);
     250             :     }
     251           9 : }
     252             : 
     253           5 : void UdpServer::HandleReceiveInternal(const_buffer recv_buffer,
     254             :     std::size_t bytes_transferred, const boost::system::error_code& error) {
     255           5 :     std::scoped_lock lock(state_guard_);
     256           5 :     if (state_ != OK) {
     257           1 :         stats_.read_errors++;
     258           1 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_IN,
     259             :             "Receive UDP server in WRONG state: " << state_);
     260           1 :         return;
     261             :     }
     262           4 :     if (error) {
     263           0 :         stats_.read_errors++;
     264           0 :         UDP_SERVER_LOG_ERROR(this, UDP_DIR_IN,
     265             :             "Read FAILED due to error: " << error.value() << " : " <<
     266             :             error.message());
     267           0 :         DeallocateBuffer(recv_buffer);
     268             :     } else {
     269             :         // Update read statistics.
     270           4 :         stats_.read_calls++;
     271           4 :         stats_.read_bytes += bytes_transferred;
     272             :         // Call the handler
     273           4 :         HandleReceive(recv_buffer, remote_endpoint_, bytes_transferred, error);
     274             :     }
     275           4 :     StartReceive();
     276           5 : }
     277             : 
     278           2 : void UdpServer::HandleReceive(const const_buffer &recv_buffer,
     279             :     udp::endpoint remote_endpoint, std::size_t bytes_transferred,
     280             :     const boost::system::error_code& error) {
     281           2 :     const_buffer rdbuf(buffer_cast<const uint8_t *>(recv_buffer),
     282           2 :                        bytes_transferred);
     283           2 :     Reader *task = new Reader(UdpServerPtr(this), remote_endpoint,
     284           2 :                               rdbuf);
     285             :     // Starting a new task for the session
     286           2 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
     287           2 :     scheduler->Enqueue(task);
     288           2 : }
     289             : 
     290           0 : void UdpServer::OnRead(const const_buffer &recv_buffer,
     291             :     const udp::endpoint &remote_endpoint) {
     292           0 :     UDP_SERVER_LOG_ERROR(this, UDP_DIR_IN, "Receive UDP: " <<
     293             :         "Default implementation of OnRead does NOT process received message");
     294           0 : }
     295             : 
     296           0 : void UdpServer::HandleSend(boost::asio::const_buffer send_buffer,
     297             :     udp::endpoint remote_endpoint, std::size_t bytes_transferred,
     298             :     const boost::system::error_code& error) {
     299           0 :     DeallocateBuffer(send_buffer);
     300           0 : }
     301             : 
     302           4 : udp::endpoint UdpServer::GetLocalEndpoint(boost::system::error_code *error)
     303             :         const {
     304           4 :     return socket_.local_endpoint(*error);
     305             : }
     306             : 
     307           0 : std::string UdpServer::GetLocalEndpointAddress() const {
     308           0 :     boost::system::error_code error;
     309           0 :     udp::endpoint ep = GetLocalEndpoint(&error);
     310           0 :     if (error.value())
     311           0 :         return "";
     312           0 :     return ep.address().to_string();
     313             : }
     314             : 
     315           1 : int UdpServer::GetLocalEndpointPort() const {
     316           1 :     boost::system::error_code error;
     317           1 :     udp::endpoint ep = GetLocalEndpoint(&error);
     318           1 :     if (error.value())
     319           1 :         return -1;
     320           0 :     return ep.port();
     321             : }
     322             : 
     323           1 : void UdpServer::GetRxSocketStats(SocketIOStats *socket_stats) const {
     324           1 :     stats_.GetRxStats(socket_stats);
     325           1 : }
     326             : 
     327           0 : void UdpServer::GetTxSocketStats(SocketIOStats *socket_stats) const {
     328           0 :     stats_.GetTxStats(socket_stats);
     329           0 : }
     330             : 
     331             : //
     332             : // UdpServerManager class routines
     333             : //
     334             : ServerManager<UdpServer, UdpServerPtr> UdpServerManager::impl_;
     335             : 
     336           6 : void UdpServerManager::AddServer(UdpServer *server) {
     337           6 :     impl_.AddServer(server);
     338           6 : }
     339             : 
     340           6 : void UdpServerManager::DeleteServer(UdpServer *server) {
     341           6 :     impl_.DeleteServer(server);
     342           6 : }