ceph/src/msg/async/rdma/RDMAServerSocketImpl.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2016 XSKY <haomai@xsky.com>
   7  *
   8  * Author: Haomai Wang <haomaiwang@gmail.com>
   9  *
  10  * This is free software; you can redistribute it and/or
  11  * modify it under the terms of the GNU Lesser General Public
  12  * License version 2.1, as published by the Free Software
  13  * Foundation.  See file COPYING.
  14  *
  15  */
  16
  17 #include "msg/async/net_handler.h"
  18 #include "RDMAStack.h"
  19
  20 #define dout_subsys ceph_subsys_ms
  21 #undef dout_prefix
  22 #define dout_prefix *_dout << " RDMAServerSocketImpl "
  23
  24 RDMAServerSocketImpl::RDMAServerSocketImpl(CephContext *cct, Infiniband* i, RDMADispatcher *s, RDMAWorker *w, entity_addr_t& a)
  25   : cct(cct), net(cct), server_setup_socket(-1), infiniband(i), dispatcher(s), worker(w), sa(a)
  26 {
  27 }
  28
  29 int RDMAServerSocketImpl::listen(entity_addr_t &sa, const SocketOptions &opt)
  30 {
  31   int rc = 0;
  32   server_setup_socket = net.create_socket(sa.get_family(), true);
  33   if (server_setup_socket < 0) {
  34     rc = -errno;
  35     lderr(cct) << __func__ << " failed to create server socket: "
  36                << cpp_strerror(errno) << dendl;
  37     return rc;
  38   }
  39
  40   rc = net.set_nonblock(server_setup_socket);
  41   if (rc < 0) {
  42     goto err;
  43   }
  44
  45   rc = net.set_socket_options(server_setup_socket, opt.nodelay, opt.rcbuf_size);
  46   if (rc < 0) {
  47     goto err;
  48   }
  49   net.set_close_on_exec(server_setup_socket);
  50
  51   rc = ::bind(server_setup_socket, sa.get_sockaddr(), sa.get_sockaddr_len());
  52   if (rc < 0) {
  53     rc = -errno;
  54     ldout(cct, 10) << __func__ << " unable to bind to " << sa.get_sockaddr()
  55                    << " on port " << sa.get_port() << ": " << cpp_strerror(errno) << dendl;
  56     goto err;
  57   }
  58
  59   rc = ::listen(server_setup_socket, 128);
  60   if (rc < 0) {
  61     rc = -errno;
  62     lderr(cct) << __func__ << " unable to listen on " << sa << ": " << cpp_strerror(errno) << dendl;
  63     goto err;
  64   }
  65
  66   ldout(cct, 20) << __func__ << " bind to " << sa.get_sockaddr() << " on port " << sa.get_port()  << dendl;
  67   return 0;
  68
  69 err:
  70   ::close(server_setup_socket);
  71   server_setup_socket = -1;
  72   return -errno;
  73 }
  74
  75 int RDMAServerSocketImpl::accept(ConnectedSocket *sock, const SocketOptions &opt, entity_addr_t *out, Worker *w)
  76 {
  77   ldout(cct, 15) << __func__ << dendl;
  78
  79   assert(sock);
  80   sockaddr_storage ss;
  81   socklen_t slen = sizeof(ss);
  82   int sd = ::accept(server_setup_socket, (sockaddr*)&ss, &slen);
  83   if (sd < 0) {
  84     return -errno;
  85   }
  86
  87   net.set_close_on_exec(sd);
  88   int r = net.set_nonblock(sd);
  89   if (r < 0) {
  90     ::close(sd);
  91     return -errno;
  92   }
  93
  94   r = net.set_socket_options(sd, opt.nodelay, opt.rcbuf_size);
  95   if (r < 0) {
  96     ::close(sd);
  97     return -errno;
  98   }
  99
 100   assert(NULL != out); //out should not be NULL in accept connection
 101
 102   out->set_sockaddr((sockaddr*)&ss);
 103   net.set_priority(sd, opt.priority, out->get_family());
 104
 105   RDMAConnectedSocketImpl* server;
 106   //Worker* w = dispatcher->get_stack()->get_worker();
 107   server = new RDMAConnectedSocketImpl(cct, infiniband, dispatcher, dynamic_cast<RDMAWorker*>(w));
 108   server->set_accept_fd(sd);
 109   ldout(cct, 20) << __func__ << " accepted a new QP, tcp_fd: " << sd << dendl;
 110   std::unique_ptr<RDMAConnectedSocketImpl> csi(server);
 111   *sock = ConnectedSocket(std::move(csi));
 112
 113   return 0;
 114 }
 115
 116 void RDMAServerSocketImpl::abort_accept()
 117 {
 118   if (server_setup_socket >= 0)
 119     ::close(server_setup_socket);
 120 }