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