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 #include "include/compat.h"
  21 #include "include/sock_compat.h"
  22
  23 #define dout_subsys ceph_subsys_ms
  24 #undef dout_prefix
  25 #define dout_prefix *_dout << " RDMAServerSocketImpl "
  26
  27 RDMAServerSocketImpl::RDMAServerSocketImpl(
  28   CephContext *cct, Infiniband* i, RDMADispatcher *s, RDMAWorker *w,
  29   entity_addr_t& a, unsigned slot)
  30   : ServerSocketImpl(a.get_type(), slot),
  31     cct(cct), net(cct), server_setup_socket(-1), infiniband(i),
  32     dispatcher(s), worker(w), sa(a)
  33 {
  34 }
  35
  36 int RDMAServerSocketImpl::listen(entity_addr_t &sa, const SocketOptions &opt)
  37 {
  38   int rc = 0;
  39   server_setup_socket = net.create_socket(sa.get_family(), true);
  40   if (server_setup_socket < 0) {
  41     rc = -errno;
  42     lderr(cct) << __func__ << " failed to create server socket: "
  43                << cpp_strerror(errno) << dendl;
  44     return rc;
  45   }
  46
  47   rc = net.set_nonblock(server_setup_socket);
  48   if (rc < 0) {
  49     goto err;
  50   }
  51
  52   rc = net.set_socket_options(server_setup_socket, opt.nodelay, opt.rcbuf_size);
  53   if (rc < 0) {
  54     goto err;
  55   }
  56
  57   rc = ::bind(server_setup_socket, sa.get_sockaddr(), sa.get_sockaddr_len());
  58   if (rc < 0) {
  59     rc = -errno;
  60     ldout(cct, 10) << __func__ << " unable to bind to " << sa.get_sockaddr()
  61                    << " on port " << sa.get_port() << ": " << cpp_strerror(errno) << dendl;
  62     goto err;
  63   }
  64
  65   rc = ::listen(server_setup_socket, cct->_conf->ms_tcp_listen_backlog);
  66   if (rc < 0) {
  67     rc = -errno;
  68     lderr(cct) << __func__ << " unable to listen on " << sa << ": " << cpp_strerror(errno) << dendl;
  69     goto err;
  70   }
  71
  72   ldout(cct, 20) << __func__ << " bind to " << sa.get_sockaddr() << " on port " << sa.get_port()  << dendl;
  73   return 0;
  74
  75 err:
  76   ::close(server_setup_socket);
  77   server_setup_socket = -1;
  78   return rc;
  79 }
  80
  81 int RDMAServerSocketImpl::accept(ConnectedSocket *sock, const SocketOptions &opt, entity_addr_t *out, Worker *w)
  82 {
  83   ldout(cct, 15) << __func__ << dendl;
  84
  85   ceph_assert(sock);
  86
  87   sockaddr_storage ss;
  88   socklen_t slen = sizeof(ss);
  89   int sd = accept_cloexec(server_setup_socket, (sockaddr*)&ss, &slen);
  90   if (sd < 0) {
  91     return -errno;
  92   }
  93
  94   int r = net.set_nonblock(sd);
  95   if (r < 0) {
  96     ::close(sd);
  97     return -errno;
  98   }
  99
 100   r = net.set_socket_options(sd, opt.nodelay, opt.rcbuf_size);
 101   if (r < 0) {
 102     ::close(sd);
 103     return -errno;
 104   }
 105
 106   ceph_assert(NULL != out); //out should not be NULL in accept connection
 107
 108   out->set_type(addr_type);
 109   out->set_sockaddr((sockaddr*)&ss);
 110   net.set_priority(sd, opt.priority, out->get_family());
 111
 112   RDMAConnectedSocketImpl* server;
 113   //Worker* w = dispatcher->get_stack()->get_worker();
 114   server = new RDMAConnectedSocketImpl(cct, infiniband, dispatcher, dynamic_cast<RDMAWorker*>(w));
 115   server->set_accept_fd(sd);
 116   ldout(cct, 20) << __func__ << " accepted a new QP, tcp_fd: " << sd << dendl;
 117   std::unique_ptr<RDMAConnectedSocketImpl> csi(server);
 118   *sock = ConnectedSocket(std::move(csi));
 119
 120   return 0;
 121 }
 122
 123 void RDMAServerSocketImpl::abort_accept()
 124 {
 125   if (server_setup_socket >= 0)
 126     ::close(server_setup_socket);
 127 }