Line data    Source code

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