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