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