int EventCenter::create_file_event(int fd, int mask, EventCallbackRef ctxt)
{
ceph_assert(in_thread());
+ ceph_assert(fd >= 0);
int r = 0;
if (fd >= nevent) {
int new_size = nevent << 2;
active(false), pending(false)
{
if (!cct->_conf->ms_async_rdma_cm) {
+ // Create the per-connection notify eventfd before the queue pair. Under fd
+ // pressure eventfd() can fail (-1); doing it first means a failure leaves no
+ // queue pair to leak. The two failure states are kept distinct on purpose so
+ // the accept/connect paths can return an honest errno: eventfd failure ->
+ // fd() < 0 && qp == nullptr; queue pair failure -> fd() >= 0 && qp == nullptr.
+ notify_fd = eventfd(0, EFD_CLOEXEC|EFD_NONBLOCK);
+ if (notify_fd < 0) {
+ lderr(cct) << __func__ << " failed to create eventfd: " << cpp_strerror(errno) << dendl;
+ return;
+ }
qp = ib->create_queue_pair(cct, dispatcher->get_tx_cq(), dispatcher->get_rx_cq(), IBV_QPT_RC, NULL);
if (!qp) {
lderr(cct) << __func__ << " queue pair create failed" << dendl;
return;
}
local_qpn = qp->get_local_qp_number();
- notify_fd = eventfd(0, EFD_CLOEXEC|EFD_NONBLOCK);
dispatcher->register_qp(qp, this);
dispatcher->perf_logger->inc(l_msgr_rdma_created_queue_pair);
dispatcher->perf_logger->inc(l_msgr_rdma_active_queue_pair);
ldout(cct, 20) << __func__ << " destruct." << dendl;
cleanup();
worker->remove_pending_conn(this);
- dispatcher->schedule_qp_destroy(local_qpn);
+ if (qp) // null on construction-failure paths; local_qpn 0 is unregistered
+ dispatcher->schedule_qp_destroy(local_qpn);
for (unsigned i=0; i < wc.size(); ++i) {
dispatcher->post_chunk_to_pool(reinterpret_cast<Chunk*>(wc[i].wr_id));
: RDMAConnectedSocketImpl(cct, ib, rdma_dispatcher, w), cm_con_handler(new C_handle_cm_connection(this))
{
status = IDLE;
- notify_fd = eventfd(0, EFD_CLOEXEC|EFD_NONBLOCK);
+ // Take ownership of the CM id/channel *before* creating the notify eventfd, so
+ // that if eventfd() fails (e.g. under fd exhaustion) the destructor still frees
+ // them (it only does so for status >= RDMA_ID_CREATED). Every failure path
+ // calls close_notify() so a later delete cannot block on close_condition.
if (info) {
is_server = true;
cm_id = info->cm_id;
cm_channel = info->cm_channel;
status = RDMA_ID_CREATED;
peer_qpn = info->qp_num;
+ } else {
+ is_server = false;
+ cm_channel = rdma_create_event_channel();
+ if (!cm_channel) {
+ lderr(cct) << __func__ << " failed to create cm event channel: " << cpp_strerror(errno) << dendl;
+ close_notify();
+ return;
+ }
+ if (rdma_create_id(cm_channel, &cm_id, NULL, RDMA_PS_TCP)) {
+ lderr(cct) << __func__ << " failed to create cm id: " << cpp_strerror(errno) << dendl;
+ rdma_destroy_event_channel(cm_channel);
+ cm_channel = nullptr;
+ close_notify();
+ return;
+ }
+ status = RDMA_ID_CREATED;
+ ldout(cct, 20) << __func__ << " successfully created cm id: " << cm_id << dendl;
+ }
+
+ notify_fd = eventfd(0, EFD_CLOEXEC|EFD_NONBLOCK);
+ if (notify_fd < 0) {
+ lderr(cct) << __func__ << " failed to create eventfd: " << cpp_strerror(errno) << dendl;
+ close_notify();
+ return;
+ }
+
+ if (info) {
if (alloc_resource()) {
close_notify();
return;
status = RESOURCE_ALLOCATED;
qp->get_local_cm_meta().peer_qpn = peer_qpn;
qp->get_peer_cm_meta().local_qpn = peer_qpn;
- } else {
- is_server = false;
- cm_channel = rdma_create_event_channel();
- rdma_create_id(cm_channel, &cm_id, NULL, RDMA_PS_TCP);
- status = RDMA_ID_CREATED;
- ldout(cct, 20) << __func__ << " successfully created cm id: " << cm_id << dendl;
}
}
rdma_get_cm_event(cm_channel, &cm_event);
ldout(cct, 20) << __func__ << " event name: " << rdma_event_str(cm_event->event) << dendl;
+ // rdma_get_cm_event() handed us a new cm id for the incoming connection; on any
+ // error path before it is adopted into a socket (whose destructor would free it)
+ // we must rdma_destroy_id() it ourselves -- acking the event does not. The event
+ // must be acked *before* destroying the id: rdma_destroy_id() blocks until the
+ // connect-request event for that id has been acknowledged.
struct rdma_cm_id *event_cm_id = cm_event->id;
struct rdma_event_channel *event_channel = rdma_create_event_channel();
+ if (!event_channel) {
+ lderr(cct) << __func__ << " failed to create event channel: " << cpp_strerror(errno) << dendl;
+ rdma_ack_cm_event(cm_event);
+ rdma_destroy_id(event_cm_id);
+ return -EMFILE;
+ }
if (net.set_nonblock(event_channel->fd) < 0) {
lderr(cct) << __func__ << " failed to switch event channel to non-block, close event channel " << dendl;
rdma_destroy_event_channel(event_channel);
rdma_ack_cm_event(cm_event);
+ rdma_destroy_id(event_cm_id);
return -errno;
}
struct rdma_conn_param *remote_conn_param = &cm_event->param.conn;
struct rdma_conn_param local_conn_param;
+ RDMAWorker *rw = dynamic_cast<RDMAWorker*>(w);
RDMACMInfo info(event_cm_id, event_channel, remote_conn_param->qp_num);
RDMAIWARPConnectedSocketImpl* server =
- new RDMAIWARPConnectedSocketImpl(cct, ib, dispatcher, dynamic_cast<RDMAWorker*>(w), &info);
+ new RDMAIWARPConnectedSocketImpl(cct, ib, dispatcher, rw, &info);
+
+ // Construction can fail under fd pressure (eventfd) or on queue pair creation;
+ // such a socket must not reach the messenger (its fd() is -1). Free it on its
+ // owning worker thread (close() marks it closed so ~IWARP won't block and frees
+ // the adopted cm id/channel) and return an honest errno (fd-first).
+ if (server->fd() < 0 || !server->get_qp()) {
+ int err = server->fd() < 0 ? -EMFILE : -EIO;
+ lderr(cct) << __func__ << " failed to create connected socket: "
+ << cpp_strerror(err) << dendl;
+ rdma_ack_cm_event(cm_event);
+ rw->center.submit_to(rw->center.get_id(), [server]() { server->close(); delete server; }, false);
+ return err;
+ }
// FIPS zeroization audit 20191115: this memset is not security related.
memset(&local_conn_param, 0, sizeof(local_conn_param));
local_conn_param.qp_num = server->get_local_qpn();
if (rdma_accept(event_cm_id, &local_conn_param)) {
+ lderr(cct) << __func__ << " rdma_accept failed: " << cpp_strerror(errno) << dendl;
+ rdma_ack_cm_event(cm_event);
+ rw->center.submit_to(rw->center.get_id(), [server]() { server->close(); delete server; }, false);
return -EAGAIN;
}
server->activate();
out->set_sockaddr((sockaddr*)&ss);
net.set_priority(sd, opt.priority, out->get_family());
+ RDMAWorker *rw = dynamic_cast<RDMAWorker*>(w);
RDMAConnectedSocketImpl* server;
//Worker* w = dispatcher->get_stack()->get_worker();
- server = new RDMAConnectedSocketImpl(cct, ib, dispatcher, dynamic_cast<RDMAWorker*>(w));
- if (!server->get_qp()) {
- lderr(cct) << __func__ << " server->qp is null" << dendl;
- // cann't use delete server here, destructor will fail.
- server->cleanup();
+ server = new RDMAConnectedSocketImpl(cct, ib, dispatcher, rw);
+ // Construction can fail under fd pressure (eventfd) or on queue pair creation.
+ // Such a socket has fd() < 0 and/or a null qp; handing it back on the success
+ // path would trip ceph_assert(socket.fd() >= 0) in AsyncConnection::accept.
+ // Return an honest errno (fd-first, so eventfd exhaustion maps to -EMFILE rather
+ // than -EIO) and free the half-built socket on its owning worker thread, where
+ // ~RDMAConnectedSocketImpl's center.in_thread() assertion holds.
+ if (server->fd() < 0 || !server->get_qp()) {
+ int err = server->fd() < 0 ? -EMFILE : -EIO;
+ lderr(cct) << __func__ << " failed to create connected socket: "
+ << cpp_strerror(err) << dendl;
+ rw->center.submit_to(rw->center.get_id(), [server]() { delete server; }, false);
::close(sd);
- return -1;
+ return err;
}
server->set_accept_fd(sd);
ldout(cct, 20) << __func__ << " accepted a new QP, tcp_fd: " << sd << dendl;
int RDMAWorker::listen(entity_addr_t &sa, unsigned addr_slot,
const SocketOptions &opt,ServerSocket *sock)
{
+ // The rdma_cm path and the iwarp socket classes are coupled: only the iwarp
+ // classes implement rdma_cm, and they always use it. Any other pairing leaves a
+ // socket half-built (the base ctor skips queue pair / eventfd creation when
+ // ms_async_rdma_cm=true, and a non-cm iwarp socket has no usable cm channel), so
+ // require the two to be enabled together and fail fast with a clear message.
+ if (cct->_conf->ms_async_rdma_cm != (cct->_conf->ms_async_rdma_type == "iwarp")) {
+ lderr(cct) << __func__ << " ms_async_rdma_cm and ms_async_rdma_type=iwarp must be "
+ "enabled together" << dendl;
+ return -EOPNOTSUPP;
+ }
+
ib->init();
dispatcher->polling_start();
int RDMAWorker::connect(const entity_addr_t &addr, const SocketOptions &opts, ConnectedSocket *socket)
{
+ const bool iwarp = cct->_conf->ms_async_rdma_type == "iwarp";
+ // The rdma_cm (connection manager) path and the iwarp socket classes are coupled:
+ // only the iwarp classes implement rdma_cm, and they always use it. Any other
+ // pairing leaves a socket half-built -- with ms_async_rdma_cm=true and a non-iwarp
+ // type the base ctor skips queue pair / eventfd creation (fd() == -1 on every
+ // connection); with an iwarp type and ms_async_rdma_cm=false the cm channel is
+ // never set up. Require the two together and fail fast.
+ if (cct->_conf->ms_async_rdma_cm != iwarp) {
+ lderr(cct) << __func__ << " ms_async_rdma_cm and ms_async_rdma_type=iwarp must be "
+ "enabled together" << dendl;
+ return -EOPNOTSUPP;
+ }
+
ib->init();
dispatcher->polling_start();
RDMAConnectedSocketImpl* p;
- if (cct->_conf->ms_async_rdma_type == "iwarp") {
+ if (iwarp) {
p = new RDMAIWARPConnectedSocketImpl(cct, ib, dispatcher, this);
} else {
p = new RDMAConnectedSocketImpl(cct, ib, dispatcher, this);
}
+
+ // Reject a socket whose construction failed before wiring it up, otherwise it
+ // reaches the messenger with fd() == -1. fd-first so eventfd exhaustion maps to
+ // -EMFILE; the non-iwarp qp check then catches a queue pair failure that left a
+ // valid eventfd (an iwarp client legitimately has a null qp until its route is
+ // resolved, so it is excluded here).
+ if (p->fd() < 0) {
+ lderr(cct) << __func__ << " failed to create eventfd (fd limit reached?)" << dendl;
+ delete p;
+ return -EMFILE;
+ }
+ if (!iwarp && !p->get_qp()) {
+ lderr(cct) << __func__ << " failed to create queue pair" << dendl;
+ delete p;
+ return -EIO;
+ }
+
int r = p->try_connect(addr, opts);
if (r < 0) {
protected:
CephContext *cct;
- Infiniband::QueuePair *qp;
+ Infiniband::QueuePair *qp = nullptr;
uint32_t peer_qpn = 0;
uint32_t local_qpn = 0;
int connected;
projects / ceph.git / commitdiff