2018-06-21 11:56:28 +02:00
|
|
|
/** Node type: infiniband
|
|
|
|
*
|
|
|
|
* @author Dennis Potter <dennis@dennispotter.eu>
|
2020-01-20 17:17:00 +01:00
|
|
|
* @copyright 2014-2020, Institute for Automation of Complex Power Systems, EONERC
|
2018-06-21 11:56:28 +02:00
|
|
|
* @license GNU General Public License (version 3)
|
|
|
|
*
|
|
|
|
* VILLASnode
|
|
|
|
*
|
|
|
|
* This program is free software: you can redistribute it and/or modify
|
|
|
|
* it under the terms of the GNU General Public License as published by
|
|
|
|
* the Free Software Foundation, either version 3 of the License, or
|
|
|
|
* any later version.
|
|
|
|
*
|
|
|
|
* This program is distributed in the hope that it will be useful,
|
|
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
|
|
* GNU General Public License for more details.
|
|
|
|
*
|
|
|
|
* You should have received a copy of the GNU General Public License
|
|
|
|
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
*********************************************************************************/
|
|
|
|
|
2019-06-18 18:51:52 +01:00
|
|
|
#include <cstring>
|
|
|
|
#include <cmath>
|
|
|
|
#include <cinttypes>
|
2018-10-20 14:22:03 +02:00
|
|
|
#include <netdb.h>
|
2018-06-21 11:56:28 +02:00
|
|
|
|
2018-08-23 17:31:01 +02:00
|
|
|
#include <villas/node/config.h>
|
2019-04-23 00:12:31 +02:00
|
|
|
#include <villas/nodes/infiniband.hpp>
|
2018-06-21 11:56:28 +02:00
|
|
|
#include <villas/plugin.h>
|
2019-04-23 13:15:00 +02:00
|
|
|
#include <villas/utils.hpp>
|
2018-06-21 11:56:28 +02:00
|
|
|
#include <villas/format_type.h>
|
2018-06-24 13:02:04 +02:00
|
|
|
#include <villas/memory.h>
|
2018-07-04 15:26:22 +02:00
|
|
|
#include <villas/memory/ib.h>
|
2018-07-21 12:52:25 +02:00
|
|
|
#include <villas/timing.h>
|
2018-06-24 13:02:04 +02:00
|
|
|
|
2019-06-04 16:55:38 +02:00
|
|
|
using namespace villas::utils;
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
static int ib_disconnect(struct vnode *n)
|
2018-06-28 12:46:16 +02:00
|
|
|
{
|
2018-07-03 17:39:06 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
Node gives back samples to framework at disconnect
The node blocks a certain amount of samples to use in its queues.
Before this commit, the only moment to release them to the framwork was
during ib_read()/ib_write().
But, there were a couple of problems. In the following I will take
ib_read() as example, but ib_write() will be analogous.
The first problem was:
1. If a QP disconnect, all Work Requests get invalidated and will be
"flushed" to a Completion Queue.
A possible solution would be, to save them in an intermediate buffer.
We could then "exchange" these samples with the framework as soon as the node
connects again and ib_read() is called again. So, we would get valid
samples from the framwork, post them, and give the "invalidated" samples back.
But, there is a second problem:
2. We cannot assume that ib_read() is ever called again after
ib_disconnect(). This is for example the case if the disconnect is
triggered by ib_stop() and not by an external node that disconnects.
This would result in a memory leak, since the samples would never be
returned to the framework, although the node is stopped.
Because of this second problem, I decided to return all samples with
sample_put() in the disconnect function. An additional benefit is that
this is more convenient than another buffer to temporarily safe the
invalidated samples.
2018-07-16 13:41:53 +02:00
|
|
|
struct ibv_wc wc[MAX(ib->recv_cq_size, ib->send_cq_size)];
|
|
|
|
int wcs;
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 1, "Starting to clean up");
|
2018-06-30 18:20:30 +02:00
|
|
|
|
2018-07-07 15:34:07 +02:00
|
|
|
rdma_disconnect(ib->ctx.id);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* If there is anything in the Completion Queue, it should be given back to the framework Receive Queue. */
|
2018-07-20 23:40:43 +02:00
|
|
|
while (ib->conn.available_recv_wrs) {
|
|
|
|
wcs = ibv_poll_cq(ib->ctx.recv_cq, ib->recv_cq_size, wc);
|
|
|
|
|
Node gives back samples to framework at disconnect
The node blocks a certain amount of samples to use in its queues.
Before this commit, the only moment to release them to the framwork was
during ib_read()/ib_write().
But, there were a couple of problems. In the following I will take
ib_read() as example, but ib_write() will be analogous.
The first problem was:
1. If a QP disconnect, all Work Requests get invalidated and will be
"flushed" to a Completion Queue.
A possible solution would be, to save them in an intermediate buffer.
We could then "exchange" these samples with the framework as soon as the node
connects again and ib_read() is called again. So, we would get valid
samples from the framwork, post them, and give the "invalidated" samples back.
But, there is a second problem:
2. We cannot assume that ib_read() is ever called again after
ib_disconnect(). This is for example the case if the disconnect is
triggered by ib_stop() and not by an external node that disconnects.
This would result in a memory leak, since the samples would never be
returned to the framework, although the node is stopped.
Because of this second problem, I decided to return all samples with
sample_put() in the disconnect function. An additional benefit is that
this is more convenient than another buffer to temporarily safe the
invalidated samples.
2018-07-16 13:41:53 +02:00
|
|
|
ib->conn.available_recv_wrs -= wcs;
|
|
|
|
|
|
|
|
for (int j = 0; j < wcs; j++)
|
2019-04-05 03:50:59 +02:00
|
|
|
sample_decref((struct sample *) (intptr_t) (wc[j].wr_id));
|
Node gives back samples to framework at disconnect
The node blocks a certain amount of samples to use in its queues.
Before this commit, the only moment to release them to the framwork was
during ib_read()/ib_write().
But, there were a couple of problems. In the following I will take
ib_read() as example, but ib_write() will be analogous.
The first problem was:
1. If a QP disconnect, all Work Requests get invalidated and will be
"flushed" to a Completion Queue.
A possible solution would be, to save them in an intermediate buffer.
We could then "exchange" these samples with the framework as soon as the node
connects again and ib_read() is called again. So, we would get valid
samples from the framwork, post them, and give the "invalidated" samples back.
But, there is a second problem:
2. We cannot assume that ib_read() is ever called again after
ib_disconnect(). This is for example the case if the disconnect is
triggered by ib_stop() and not by an external node that disconnects.
This would result in a memory leak, since the samples would never be
returned to the framework, although the node is stopped.
Because of this second problem, I decided to return all samples with
sample_put() in the disconnect function. An additional benefit is that
this is more convenient than another buffer to temporarily safe the
invalidated samples.
2018-07-16 13:41:53 +02:00
|
|
|
}
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Send Queue */
|
Node gives back samples to framework at disconnect
The node blocks a certain amount of samples to use in its queues.
Before this commit, the only moment to release them to the framwork was
during ib_read()/ib_write().
But, there were a couple of problems. In the following I will take
ib_read() as example, but ib_write() will be analogous.
The first problem was:
1. If a QP disconnect, all Work Requests get invalidated and will be
"flushed" to a Completion Queue.
A possible solution would be, to save them in an intermediate buffer.
We could then "exchange" these samples with the framework as soon as the node
connects again and ib_read() is called again. So, we would get valid
samples from the framwork, post them, and give the "invalidated" samples back.
But, there is a second problem:
2. We cannot assume that ib_read() is ever called again after
ib_disconnect(). This is for example the case if the disconnect is
triggered by ib_stop() and not by an external node that disconnects.
This would result in a memory leak, since the samples would never be
returned to the framework, although the node is stopped.
Because of this second problem, I decided to return all samples with
sample_put() in the disconnect function. An additional benefit is that
this is more convenient than another buffer to temporarily safe the
invalidated samples.
2018-07-16 13:41:53 +02:00
|
|
|
while ((wcs = ibv_poll_cq(ib->ctx.send_cq, ib->send_cq_size, wc)))
|
|
|
|
for (int j = 0; j < wcs; j++)
|
|
|
|
if (wc[j].wr_id > 0)
|
2019-04-05 03:50:59 +02:00
|
|
|
sample_decref((struct sample *) (intptr_t) (wc[j].wr_id));
|
Node gives back samples to framework at disconnect
The node blocks a certain amount of samples to use in its queues.
Before this commit, the only moment to release them to the framwork was
during ib_read()/ib_write().
But, there were a couple of problems. In the following I will take
ib_read() as example, but ib_write() will be analogous.
The first problem was:
1. If a QP disconnect, all Work Requests get invalidated and will be
"flushed" to a Completion Queue.
A possible solution would be, to save them in an intermediate buffer.
We could then "exchange" these samples with the framework as soon as the node
connects again and ib_read() is called again. So, we would get valid
samples from the framwork, post them, and give the "invalidated" samples back.
But, there is a second problem:
2. We cannot assume that ib_read() is ever called again after
ib_disconnect(). This is for example the case if the disconnect is
triggered by ib_stop() and not by an external node that disconnects.
This would result in a memory leak, since the samples would never be
returned to the framework, although the node is stopped.
Because of this second problem, I decided to return all samples with
sample_put() in the disconnect function. An additional benefit is that
this is more convenient than another buffer to temporarily safe the
invalidated samples.
2018-07-16 13:41:53 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Destroy QP */
|
2018-07-03 17:39:06 +02:00
|
|
|
rdma_destroy_qp(ib->ctx.id);
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 3, "Destroyed QP");
|
2018-06-28 12:46:16 +02:00
|
|
|
|
2018-07-08 15:00:47 +02:00
|
|
|
return ib->stopThreads;
|
2018-06-28 12:46:16 +02:00
|
|
|
}
|
2018-06-27 17:01:47 +02:00
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
static void ib_build_ibv(struct vnode *n)
|
2018-06-23 14:53:37 +02:00
|
|
|
{
|
2018-07-03 18:01:49 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
2018-07-08 15:32:28 +02:00
|
|
|
int ret;
|
2018-07-05 13:57:25 +02:00
|
|
|
|
2018-07-08 15:32:28 +02:00
|
|
|
debug(LOG_IB | 1, "Starting to build IBV components");
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Create completion queues (No completion channel!) */
|
2019-04-22 23:45:38 +02:00
|
|
|
ib->ctx.recv_cq = ibv_create_cq(ib->ctx.id->verbs, ib->recv_cq_size, nullptr, nullptr, 0);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (!ib->ctx.recv_cq)
|
2018-07-05 13:57:25 +02:00
|
|
|
error("Could not create receive completion queue in node %s", node_name(n));
|
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Created receive Completion Queue");
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-04-22 23:45:38 +02:00
|
|
|
ib->ctx.send_cq = ibv_create_cq(ib->ctx.id->verbs, ib->send_cq_size, nullptr, nullptr, 0);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (!ib->ctx.send_cq)
|
2018-07-05 13:57:25 +02:00
|
|
|
error("Could not create send completion queue in node %s", node_name(n));
|
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Created send Completion Queue");
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Prepare remaining Queue Pair (QP) attributes */
|
2018-07-03 18:01:49 +02:00
|
|
|
ib->qp_init.send_cq = ib->ctx.send_cq;
|
|
|
|
ib->qp_init.recv_cq = ib->ctx.recv_cq;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Create the actual QP */
|
2018-07-03 18:01:49 +02:00
|
|
|
ret = rdma_create_qp(ib->ctx.id, ib->ctx.pd, &ib->qp_init);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-05 13:57:25 +02:00
|
|
|
error("Failed to create Queue Pair in node %s", node_name(n));
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 3, "Created Queue Pair with %i receive and %i send elements",
|
2018-07-03 18:01:49 +02:00
|
|
|
ib->qp_init.cap.max_recv_wr, ib->qp_init.cap.max_send_wr);
|
|
|
|
|
2018-07-15 13:51:18 +02:00
|
|
|
if (ib->conn.send_inline)
|
2018-07-14 15:20:24 +02:00
|
|
|
info("Maximum inline size is set to %i byte", ib->qp_init.cap.max_inline_data);
|
2018-06-23 14:53:37 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
static int ib_addr_resolved(struct vnode *n)
|
2018-06-23 14:53:37 +02:00
|
|
|
{
|
2018-07-03 17:39:06 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
int ret;
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 1, "Successfully resolved address");
|
2018-06-23 19:05:33 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Build all components from IB Verbs */
|
2018-07-03 17:39:06 +02:00
|
|
|
ib_build_ibv(n);
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Resolve address */
|
2018-07-03 17:39:06 +02:00
|
|
|
ret = rdma_resolve_route(ib->ctx.id, ib->conn.timeout);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-05 13:57:25 +02:00
|
|
|
error("Failed to resolve route in node %s", node_name(n));
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2018-07-03 17:39:06 +02:00
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
static int ib_route_resolved(struct vnode *n)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-03 17:39:06 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
int ret;
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2018-07-03 17:39:06 +02:00
|
|
|
struct rdma_conn_param cm_params;
|
|
|
|
memset(&cm_params, 0, sizeof(cm_params));
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Send connection request */
|
2018-07-03 17:39:06 +02:00
|
|
|
ret = rdma_connect(ib->ctx.id, &cm_params);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-05 13:57:25 +02:00
|
|
|
error("Failed to connect in node %s", node_name(n));
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 1, "Called rdma_connect");
|
2018-06-30 18:20:30 +02:00
|
|
|
|
2018-07-03 17:39:06 +02:00
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
static int ib_connect_request(struct vnode *n, struct rdma_cm_id *id)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-03 17:39:06 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
int ret;
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 1, "Received a connection request!");
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2018-07-03 17:39:06 +02:00
|
|
|
ib->ctx.id = id;
|
|
|
|
ib_build_ibv(n);
|
2018-06-30 18:20:30 +02:00
|
|
|
|
2018-07-03 17:39:06 +02:00
|
|
|
struct rdma_conn_param cm_params;
|
|
|
|
memset(&cm_params, 0, sizeof(cm_params));
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Accept connection request */
|
2018-07-03 17:39:06 +02:00
|
|
|
ret = rdma_accept(ib->ctx.id, &cm_params);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-05 13:57:25 +02:00
|
|
|
error("Failed to connect in node %s", node_name(n));
|
2018-06-23 14:53:37 +02:00
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
info("Successfully accepted connection request in node %s", node_name(n));
|
2018-06-30 18:20:30 +02:00
|
|
|
|
2018-07-03 17:39:06 +02:00
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int ib_reverse(struct vnode *n)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-16 10:34:50 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
|
|
|
|
SWAP(ib->conn.src_addr, ib->conn.dst_addr);
|
|
|
|
|
2018-07-03 18:01:49 +02:00
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int ib_parse(struct vnode *n, json_t *cfg)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-03 18:01:49 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
|
|
|
|
int ret;
|
2019-04-22 23:45:38 +02:00
|
|
|
char *local = nullptr, *remote = nullptr, *lasts;
|
2018-08-07 16:45:05 +02:00
|
|
|
const char *transport_mode = "RC";
|
2018-07-03 18:01:49 +02:00
|
|
|
int timeout = 1000;
|
2018-07-15 13:51:18 +02:00
|
|
|
int recv_cq_size = 128;
|
|
|
|
int send_cq_size = 128;
|
2018-07-03 18:01:49 +02:00
|
|
|
int max_send_wr = 128;
|
|
|
|
int max_recv_wr = 128;
|
2018-07-13 13:50:30 +02:00
|
|
|
int max_inline_data = 0;
|
2018-07-15 13:51:18 +02:00
|
|
|
int send_inline = 1;
|
|
|
|
int vectorize_in = 1;
|
|
|
|
int vectorize_out = 1;
|
2018-07-25 16:22:34 +02:00
|
|
|
int buffer_subtraction = 16;
|
2018-08-02 10:41:37 +02:00
|
|
|
int use_fallback = 1;
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Parse JSON files and copy to local variables */
|
2019-04-22 23:45:38 +02:00
|
|
|
json_t *json_in = nullptr;
|
|
|
|
json_t *json_out = nullptr;
|
2018-07-03 18:01:49 +02:00
|
|
|
json_error_t err;
|
2018-07-15 13:51:18 +02:00
|
|
|
|
2019-03-31 22:42:58 +02:00
|
|
|
ret = json_unpack_ex(cfg, &err, 0, "{ s?: o, s?: o, s?: s }",
|
2018-07-15 13:51:18 +02:00
|
|
|
"in", &json_in,
|
|
|
|
"out", &json_out,
|
2018-08-07 16:45:05 +02:00
|
|
|
"rdma_transport_mode", &transport_mode
|
2018-07-03 18:01:49 +02:00
|
|
|
);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-15 13:51:18 +02:00
|
|
|
jerror(&err, "Failed to parse in/out json blocks");
|
|
|
|
|
|
|
|
|
|
|
|
if (json_in) {
|
2018-10-14 01:54:54 +02:00
|
|
|
ret = json_unpack_ex(json_in, &err, 0, "{ s?: s, s?: i, s?: i, s?: i, s?: i}",
|
2018-07-15 13:51:18 +02:00
|
|
|
"address", &local,
|
|
|
|
"cq_size", &recv_cq_size,
|
|
|
|
"max_wrs", &max_recv_wr,
|
|
|
|
"vectorize", &vectorize_in,
|
|
|
|
"buffer_subtraction", &buffer_subtraction
|
|
|
|
);
|
|
|
|
if (ret)
|
|
|
|
jerror(&err, "Failed to parse input configuration of node %s", node_name(n));
|
|
|
|
}
|
|
|
|
|
|
|
|
if (json_out) {
|
2018-10-21 12:35:21 +02:00
|
|
|
ret = json_unpack_ex(json_out, &err, 0, "{ s?: s, s?: i, s?: i, s?: i, s?: i, s?: b, s?: i, s?: b, s?: i}",
|
2018-07-15 13:51:18 +02:00
|
|
|
"address", &remote,
|
|
|
|
"resolution_timeout", &timeout,
|
|
|
|
"cq_size", &send_cq_size,
|
|
|
|
"max_wrs", &max_send_wr,
|
|
|
|
"max_inline_data", &max_inline_data,
|
|
|
|
"send_inline", &send_inline,
|
2018-08-02 10:41:37 +02:00
|
|
|
"vectorize", &vectorize_out,
|
2018-10-21 12:35:21 +02:00
|
|
|
"use_fallback", &use_fallback,
|
2018-11-02 12:46:12 +01:00
|
|
|
"periodic_signaling", &ib->periodic_signaling
|
2018-07-15 13:51:18 +02:00
|
|
|
);
|
|
|
|
if (ret)
|
|
|
|
jerror(&err, "Failed to parse output configuration of node %s", node_name(n));
|
|
|
|
|
2018-07-25 16:22:34 +02:00
|
|
|
if (remote) {
|
|
|
|
ib->is_source = 1;
|
2018-07-15 13:51:18 +02:00
|
|
|
|
2018-07-25 16:22:34 +02:00
|
|
|
debug(LOG_IB | 3, "Node %s is up as source and target", node_name(n));
|
|
|
|
}
|
2018-07-15 13:51:18 +02:00
|
|
|
}
|
|
|
|
else {
|
|
|
|
ib->is_source = 0;
|
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Node %s is up as target", node_name(n));
|
|
|
|
}
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set fallback mode */
|
2018-08-02 10:41:37 +02:00
|
|
|
ib->conn.use_fallback = use_fallback;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set vectorize mode. Do not print, since framework will print this information */
|
2018-07-15 13:51:18 +02:00
|
|
|
n->in.vectorize = vectorize_in;
|
|
|
|
n->out.vectorize = vectorize_out;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set buffer subtraction */
|
2018-07-15 13:51:18 +02:00
|
|
|
ib->conn.buffer_subtraction = buffer_subtraction;
|
|
|
|
|
|
|
|
debug(LOG_IB | 4, "Set buffer subtraction to %i in node %s", buffer_subtraction, node_name(n));
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Translate IP:PORT to a struct addrinfo */
|
2019-04-05 03:50:59 +02:00
|
|
|
char *ip_adr = strtok_r(local, ":", &lasts);
|
2019-04-22 23:45:38 +02:00
|
|
|
char *port = strtok_r(nullptr, ":", &lasts);
|
2018-07-04 19:04:08 +02:00
|
|
|
|
2019-04-22 23:45:38 +02:00
|
|
|
ret = getaddrinfo(ip_adr, port, nullptr, &ib->conn.src_addr);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-03 18:01:49 +02:00
|
|
|
error("Failed to resolve local address '%s' of node %s: %s",
|
|
|
|
local, node_name(n), gai_strerror(ret));
|
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 4, "Translated %s:%s to a struct addrinfo in node %s", ip_adr, port, node_name(n));
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Translate port space */
|
2018-08-07 16:45:05 +02:00
|
|
|
if (strcmp(transport_mode, "RC") == 0) {
|
2018-07-19 20:42:20 +02:00
|
|
|
ib->conn.port_space = RDMA_PS_TCP;
|
|
|
|
ib->qp_init.qp_type = IBV_QPT_RC;
|
|
|
|
}
|
2018-08-07 16:45:05 +02:00
|
|
|
else if (strcmp(transport_mode, "UC") == 0) {
|
|
|
|
#ifdef RDMA_CMA_H_CUSTOM
|
|
|
|
ib->conn.port_space = RDMA_PS_IB;
|
|
|
|
ib->qp_init.qp_type = IBV_QPT_UC;
|
|
|
|
#else
|
2018-09-09 12:37:46 +02:00
|
|
|
error("Unreliable Connected (UC) mode is only available with an adapted version of librdma. "
|
|
|
|
"Please read the Infiniband node type Documentation for more information on UC!");
|
2018-08-07 16:45:05 +02:00
|
|
|
#endif
|
|
|
|
}
|
|
|
|
else if (strcmp(transport_mode, "UD") == 0) {
|
2018-07-19 20:42:20 +02:00
|
|
|
ib->conn.port_space = RDMA_PS_UDP;
|
2018-07-20 23:34:52 +02:00
|
|
|
ib->qp_init.qp_type = IBV_QPT_UD;
|
2018-07-19 20:42:20 +02:00
|
|
|
}
|
2018-07-04 19:04:08 +02:00
|
|
|
else
|
2018-08-07 16:45:05 +02:00
|
|
|
error("transport_mode = %s is not a valid transport mode in node %s!",
|
|
|
|
transport_mode, node_name(n));
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2018-08-07 16:45:05 +02:00
|
|
|
debug(LOG_IB | 4, "Set transport mode to %s in node %s", transport_mode, node_name(n));
|
2018-07-05 13:57:25 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set timeout */
|
2018-07-03 18:01:49 +02:00
|
|
|
ib->conn.timeout = timeout;
|
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 4, "Set timeout to %i in node %s", timeout, node_name(n));
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set completion queue size */
|
2018-07-15 13:51:18 +02:00
|
|
|
ib->recv_cq_size = recv_cq_size;
|
|
|
|
ib->send_cq_size = send_cq_size;
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2018-07-15 13:51:18 +02:00
|
|
|
debug(LOG_IB | 4, "Set Completion Queue size to %i & %i (in & out) in node %s",
|
|
|
|
recv_cq_size, send_cq_size, node_name(n));
|
2018-07-05 13:57:25 +02:00
|
|
|
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Translate inline mode */
|
2018-07-15 13:51:18 +02:00
|
|
|
ib->conn.send_inline = send_inline;
|
2018-07-13 13:50:30 +02:00
|
|
|
|
2018-07-15 13:51:18 +02:00
|
|
|
debug(LOG_IB | 4, "Set send_inline to %i in node %s", send_inline, node_name(n));
|
2018-07-13 13:50:30 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set max. send and receive Work Requests */
|
2018-07-03 18:01:49 +02:00
|
|
|
ib->qp_init.cap.max_send_wr = max_send_wr;
|
|
|
|
ib->qp_init.cap.max_recv_wr = max_recv_wr;
|
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 4, "Set max_send_wr and max_recv_wr in node %s to %i and %i, respectively",
|
|
|
|
node_name(n), max_send_wr, max_recv_wr);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set available receive Work Requests to 0 */
|
2018-07-03 18:01:49 +02:00
|
|
|
ib->conn.available_recv_wrs = 0;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set remaining QP attributes */
|
2018-07-21 12:52:25 +02:00
|
|
|
ib->qp_init.cap.max_send_sge = 4;
|
2018-07-23 16:52:38 +02:00
|
|
|
ib->qp_init.cap.max_recv_sge = (ib->conn.port_space == RDMA_PS_UDP) ? 5 : 4;
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set number of bytes to be send inline */
|
2018-07-13 13:50:30 +02:00
|
|
|
ib->qp_init.cap.max_inline_data = max_inline_data;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* If node will send data, set remote address */
|
2018-07-15 13:51:18 +02:00
|
|
|
if (ib->is_source) {
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Translate address info */
|
2019-02-18 01:09:33 +01:00
|
|
|
char *ip_adr = strtok_r(remote, ":", &lasts);
|
2019-04-22 23:45:38 +02:00
|
|
|
char *port = strtok_r(nullptr, ":", &lasts);
|
2018-07-04 19:04:08 +02:00
|
|
|
|
2019-04-22 23:45:38 +02:00
|
|
|
ret = getaddrinfo(ip_adr, port, nullptr, &ib->conn.dst_addr);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-03 18:01:49 +02:00
|
|
|
error("Failed to resolve remote address '%s' of node %s: %s",
|
|
|
|
remote, node_name(n), gai_strerror(ret));
|
|
|
|
|
2018-07-07 14:36:23 +02:00
|
|
|
debug(LOG_IB | 4, "Translated %s:%s to a struct addrinfo", ip_adr, port);
|
2018-07-03 18:01:49 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int ib_check(struct vnode *n)
|
2018-07-05 15:30:33 +02:00
|
|
|
{
|
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Check if read substraction makes sense */
|
2018-07-15 13:51:18 +02:00
|
|
|
if (ib->conn.buffer_subtraction < 2 * n->in.vectorize)
|
|
|
|
error("The buffer substraction value must be bigger than 2 * in.vectorize");
|
|
|
|
|
|
|
|
if (ib->conn.buffer_subtraction >= ib->qp_init.cap.max_recv_wr - n->in.vectorize)
|
2018-07-25 16:22:34 +02:00
|
|
|
error("The buffer substraction value cannot be bigger than in.max_wrs - in.vectorize");
|
2018-07-15 13:51:18 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Check if the set value is a power of 2, and warn the user if this is not the case */
|
2019-04-22 23:43:46 +02:00
|
|
|
unsigned max_send_pow = (int) pow(2, ceil(log2(ib->qp_init.cap.max_send_wr)));
|
|
|
|
unsigned max_recv_pow = (int) pow(2, ceil(log2(ib->qp_init.cap.max_recv_wr)));
|
2018-07-05 15:30:33 +02:00
|
|
|
|
2018-07-13 12:21:59 +02:00
|
|
|
if (ib->qp_init.cap.max_send_wr != max_send_pow) {
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Max nr. of send WRs (%i) is not a power of 2! It will be changed to a power of 2: %i",
|
2018-07-05 15:30:33 +02:00
|
|
|
ib->qp_init.cap.max_send_wr, max_send_pow);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Change it now, because otherwise errors are possible in ib_start(). */
|
2018-07-13 12:21:59 +02:00
|
|
|
ib->qp_init.cap.max_send_wr = max_send_pow;
|
|
|
|
}
|
|
|
|
|
|
|
|
if (ib->qp_init.cap.max_recv_wr != max_recv_pow) {
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Max nr. of recv WRs (%i) is not a power of 2! It will be changed to a power of 2: %i",
|
2018-07-05 15:30:33 +02:00
|
|
|
ib->qp_init.cap.max_recv_wr, max_recv_pow);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Change it now, because otherwise errors are possible in ib_start(). */
|
2018-07-13 12:21:59 +02:00
|
|
|
ib->qp_init.cap.max_recv_wr = max_recv_pow;
|
|
|
|
}
|
2018-07-05 15:30:33 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Check maximum size of max_recv_wr and max_send_wr */
|
2018-07-05 15:30:33 +02:00
|
|
|
if (ib->qp_init.cap.max_send_wr > 8192)
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Max number of send WRs (%i) is bigger than send queue!", ib->qp_init.cap.max_send_wr);
|
2018-07-05 15:30:33 +02:00
|
|
|
|
|
|
|
if (ib->qp_init.cap.max_recv_wr > 8192)
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Max number of receive WRs (%i) is bigger than send queue!", ib->qp_init.cap.max_recv_wr);
|
2018-07-05 15:30:33 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set periodic signaling
|
|
|
|
* This is done here, so that it uses the checked max_send_wr value */
|
2018-11-02 12:46:12 +01:00
|
|
|
if (ib->periodic_signaling == 0)
|
|
|
|
ib->periodic_signaling = ib->qp_init.cap.max_send_wr / 2;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Warn user if he changed the default inline value */
|
2018-07-13 13:50:30 +02:00
|
|
|
if (ib->qp_init.cap.max_inline_data != 0)
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("You changed the default value of max_inline_data. This might influence the maximum number "
|
2018-08-02 10:41:37 +02:00
|
|
|
"of outstanding Work Requests in the Queue Pair and can be a reason for the Queue Pair creation to fail");
|
2018-07-14 15:20:24 +02:00
|
|
|
|
2018-07-05 15:30:33 +02:00
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
char * ib_print(struct vnode *n)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-03 18:01:49 +02:00
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int ib_destroy(struct vnode *n)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-03 18:01:49 +02:00
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
static void ib_create_bind_id(struct vnode *n)
|
2018-07-15 16:37:52 +02:00
|
|
|
{
|
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
int ret;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Create rdma_cm_id
|
|
|
|
*
|
2018-08-07 16:45:05 +02:00
|
|
|
* The unreliable connected mode is officially not supported by the rdma_cm library. Only the Reliable
|
|
|
|
* Connected mode (RDMA_PS_TCP) and the Unreliable Datagram mode (RDMA_PS_UDP). Although it is not officially
|
|
|
|
* supported, it is possible to use it with a few small adaptions to the sourcecode. To enable the
|
|
|
|
* support for UC connections follow the steps below:
|
|
|
|
*
|
|
|
|
* 1. git clone https://github.com/linux-rdma/rdma-core
|
|
|
|
* 2. cd rdma-core
|
|
|
|
* 2. Edit librdmacm/cma.c and remove the keyword 'static' in front of:
|
|
|
|
*
|
|
|
|
* static int rdma_create_id2(struct rdma_event_channel *channel,
|
|
|
|
* struct rdma_cm_id **id, void *context,
|
|
|
|
* enum rdma_port_space ps, enum ibv_qp_type qp_type)
|
|
|
|
*
|
|
|
|
* 3. Edit librdmacm/rdma_cma.h and add the following two entries to the file:
|
|
|
|
*
|
|
|
|
* #define RDMA_CMA_H_CUSTOM
|
|
|
|
*
|
|
|
|
* int rdma_create_id2(struct rdma_event_channel *channel,
|
|
|
|
* struct rdma_cm_id **id, void *context,
|
|
|
|
* enum rdma_port_space ps, enum ibv_qp_type qp_type);
|
|
|
|
*
|
2018-08-07 17:36:29 +02:00
|
|
|
* 4. Edit librdmacm/librdmacm.map and add a new line with:
|
|
|
|
*
|
|
|
|
* rdma_create_id2
|
|
|
|
*
|
|
|
|
* 5. ./build.sh
|
|
|
|
* 6. cd build && sudo make install
|
2018-08-07 16:45:05 +02:00
|
|
|
*
|
|
|
|
*/
|
|
|
|
#ifdef RDMA_CMA_H_CUSTOM
|
2019-04-22 23:45:38 +02:00
|
|
|
ret = rdma_create_id2(ib->ctx.ec, &ib->ctx.id, nullptr, ib->conn.port_space, ib->qp_init.qp_type);
|
2018-08-07 16:45:05 +02:00
|
|
|
#else
|
2019-04-22 23:45:38 +02:00
|
|
|
ret = rdma_create_id(ib->ctx.ec, &ib->ctx.id, nullptr, ib->conn.port_space);
|
2018-08-07 16:45:05 +02:00
|
|
|
#endif
|
2018-07-15 16:37:52 +02:00
|
|
|
if (ret)
|
|
|
|
error("Failed to create rdma_cm_id of node %s: %s", node_name(n), gai_strerror(ret));
|
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Created rdma_cm_id");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Bind rdma_cm_id to the HCA */
|
2018-07-15 16:37:52 +02:00
|
|
|
ret = rdma_bind_addr(ib->ctx.id, ib->conn.src_addr->ai_addr);
|
|
|
|
if (ret)
|
|
|
|
error("Failed to bind to local device of node %s: %s",
|
|
|
|
node_name(n), gai_strerror(ret));
|
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Bound rdma_cm_id to Infiniband device");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* The ID will be overwritten for the target. If the event type is
|
|
|
|
* RDMA_CM_EVENT_CONNECT_REQUEST, >then this references a new id for
|
|
|
|
* that communication.
|
|
|
|
*/
|
2018-07-15 16:37:52 +02:00
|
|
|
ib->ctx.listen_id = ib->ctx.id;
|
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
static void ib_continue_as_listen(struct vnode *n, struct rdma_cm_event *event)
|
2018-07-15 16:37:52 +02:00
|
|
|
{
|
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
int ret;
|
|
|
|
|
2018-08-02 10:41:37 +02:00
|
|
|
if (ib->conn.use_fallback)
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Trying to continue as listening node");
|
2018-08-02 10:41:37 +02:00
|
|
|
else
|
|
|
|
error("Cannot establish a connection with remote host! If you want that %s tries to "
|
|
|
|
"continue as listening node in such cases, set use_fallback = true in the configuration",
|
|
|
|
node_name(n));
|
|
|
|
|
2019-06-23 16:13:23 +02:00
|
|
|
n->state = State::STARTED;
|
2018-07-15 16:37:52 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Acknowledge event */
|
2018-07-15 16:37:52 +02:00
|
|
|
rdma_ack_cm_event(event);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Destroy ID */
|
2018-07-15 16:37:52 +02:00
|
|
|
rdma_destroy_id(ib->ctx.listen_id);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Create rdma_cm_id and bind to device */
|
2018-07-15 16:37:52 +02:00
|
|
|
ib_create_bind_id(n);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Listen to id for events */
|
2018-07-15 16:37:52 +02:00
|
|
|
ret = rdma_listen(ib->ctx.listen_id, 10);
|
|
|
|
if (ret)
|
|
|
|
error("Failed to listen to rdma_cm_id on node %s", node_name(n));
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Node is not a source (and will not send data */
|
2018-07-15 16:37:52 +02:00
|
|
|
ib->is_source = 0;
|
|
|
|
|
|
|
|
info("Node %s is set to listening mode", node_name(n));
|
|
|
|
}
|
|
|
|
|
2018-07-05 18:26:32 +02:00
|
|
|
void * ib_rdma_cm_event_thread(void *n)
|
|
|
|
{
|
2020-08-25 21:00:52 +02:00
|
|
|
struct vnode *node = (struct vnode *) n;
|
2018-07-07 12:49:22 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) node->_vd;
|
2018-07-05 18:26:32 +02:00
|
|
|
struct rdma_cm_event *event;
|
2018-07-07 12:49:22 +02:00
|
|
|
int ret = 0;
|
|
|
|
|
2018-07-07 13:08:08 +02:00
|
|
|
debug(LOG_IB | 1, "Started rdma_cm_event thread of node %s", node_name(node));
|
2018-07-05 18:26:32 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Wait until node is completely started */
|
2019-06-23 16:13:23 +02:00
|
|
|
while (node->state != State::STARTED);
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Monitor event channel */
|
2018-07-07 12:49:22 +02:00
|
|
|
while (rdma_get_cm_event(ib->ctx.ec, &event) == 0) {
|
2018-07-17 11:10:05 +02:00
|
|
|
debug(LOG_IB | 2, "Received communication event: %s", rdma_event_str(event->event));
|
2018-07-05 18:26:32 +02:00
|
|
|
|
2018-07-07 13:08:08 +02:00
|
|
|
switch(event->event) {
|
2018-07-07 12:49:22 +02:00
|
|
|
case RDMA_CM_EVENT_ADDR_RESOLVED:
|
2019-04-22 23:43:46 +02:00
|
|
|
ret = ib_addr_resolved(node);
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
2018-07-05 18:26:32 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
case RDMA_CM_EVENT_ADDR_ERROR:
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Address resolution (rdma_resolve_addr) failed!");
|
2018-07-15 16:37:52 +02:00
|
|
|
|
2019-04-22 23:43:46 +02:00
|
|
|
ib_continue_as_listen(node, event);
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
|
|
|
|
|
|
|
case RDMA_CM_EVENT_ROUTE_RESOLVED:
|
2019-04-22 23:43:46 +02:00
|
|
|
ret = ib_route_resolved(node);
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
2018-07-05 13:57:25 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
case RDMA_CM_EVENT_ROUTE_ERROR:
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Route resolution (rdma_resovle_route) failed!");
|
2018-07-15 16:37:52 +02:00
|
|
|
|
2019-04-22 23:43:46 +02:00
|
|
|
ib_continue_as_listen(node, event);
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
|
|
|
|
2018-07-19 18:32:06 +02:00
|
|
|
case RDMA_CM_EVENT_UNREACHABLE:
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Remote server unreachable!");
|
2018-07-19 18:32:06 +02:00
|
|
|
|
2019-04-22 23:43:46 +02:00
|
|
|
ib_continue_as_listen(node, event);
|
2018-07-19 18:32:06 +02:00
|
|
|
break;
|
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
case RDMA_CM_EVENT_CONNECT_REQUEST:
|
2019-04-22 23:43:46 +02:00
|
|
|
ret = ib_connect_request(node, event->id);
|
2018-07-19 20:33:41 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* A target UDP node will never really connect. In order to receive data,
|
|
|
|
* we set it to connected after it answered the connection request
|
|
|
|
* with rdma_connect.
|
|
|
|
*/
|
2018-08-13 14:50:49 +02:00
|
|
|
if (ib->conn.port_space == RDMA_PS_UDP && !ib->is_source)
|
2019-06-23 16:13:23 +02:00
|
|
|
node->state = State::CONNECTED;
|
2018-08-13 14:50:49 +02:00
|
|
|
else
|
2019-06-23 16:13:23 +02:00
|
|
|
node->state = State::PENDING_CONNECT;
|
2018-08-04 17:34:52 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
|
|
|
|
|
|
|
case RDMA_CM_EVENT_CONNECT_ERROR:
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("An error has occurred trying to establish a connection!");
|
2018-07-15 16:37:52 +02:00
|
|
|
|
2019-04-22 23:43:46 +02:00
|
|
|
ib_continue_as_listen(node, event);
|
2018-07-07 12:49:22 +02:00
|
|
|
|
|
|
|
break;
|
|
|
|
|
|
|
|
case RDMA_CM_EVENT_REJECTED:
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Connection request or response was rejected by the remote end point!");
|
2018-07-15 16:37:52 +02:00
|
|
|
|
2019-04-22 23:43:46 +02:00
|
|
|
ib_continue_as_listen(node, event);
|
2018-07-07 12:49:22 +02:00
|
|
|
|
|
|
|
break;
|
2018-07-15 16:37:52 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
case RDMA_CM_EVENT_ESTABLISHED:
|
2019-03-26 07:09:55 +01:00
|
|
|
/* If the connection is unreliable connectionless, set appropriate variables */
|
2018-07-25 18:51:28 +02:00
|
|
|
if (ib->conn.port_space == RDMA_PS_UDP) {
|
|
|
|
ib->conn.ud.ud = event->param.ud;
|
|
|
|
ib->conn.ud.ah = ibv_create_ah(ib->ctx.pd, &ib->conn.ud.ud.ah_attr);
|
|
|
|
}
|
2018-07-19 18:32:06 +02:00
|
|
|
|
2019-06-23 16:13:23 +02:00
|
|
|
node->state = State::CONNECTED;
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2019-04-22 23:43:46 +02:00
|
|
|
info("Connection established in node %s", node_name(node));
|
2018-07-21 12:14:25 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
|
|
|
|
|
|
|
case RDMA_CM_EVENT_DISCONNECTED:
|
2019-06-23 16:13:23 +02:00
|
|
|
node->state = State::STARTED;
|
Node gives back samples to framework at disconnect
The node blocks a certain amount of samples to use in its queues.
Before this commit, the only moment to release them to the framwork was
during ib_read()/ib_write().
But, there were a couple of problems. In the following I will take
ib_read() as example, but ib_write() will be analogous.
The first problem was:
1. If a QP disconnect, all Work Requests get invalidated and will be
"flushed" to a Completion Queue.
A possible solution would be, to save them in an intermediate buffer.
We could then "exchange" these samples with the framework as soon as the node
connects again and ib_read() is called again. So, we would get valid
samples from the framwork, post them, and give the "invalidated" samples back.
But, there is a second problem:
2. We cannot assume that ib_read() is ever called again after
ib_disconnect(). This is for example the case if the disconnect is
triggered by ib_stop() and not by an external node that disconnects.
This would result in a memory leak, since the samples would never be
returned to the framework, although the node is stopped.
Because of this second problem, I decided to return all samples with
sample_put() in the disconnect function. An additional benefit is that
this is more convenient than another buffer to temporarily safe the
invalidated samples.
2018-07-16 13:41:53 +02:00
|
|
|
|
2019-04-22 23:43:46 +02:00
|
|
|
ret = ib_disconnect(node);
|
2018-07-07 14:36:23 +02:00
|
|
|
|
2018-07-14 16:46:23 +02:00
|
|
|
if (!ret)
|
|
|
|
info("Host disconnected. Ready to accept new connections.");
|
2018-07-07 15:34:07 +02:00
|
|
|
|
2018-07-07 14:36:23 +02:00
|
|
|
break;
|
|
|
|
|
|
|
|
case RDMA_CM_EVENT_TIMEWAIT_EXIT:
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
|
|
|
|
|
|
|
default:
|
2018-07-07 13:08:08 +02:00
|
|
|
error("Unknown event occurred: %u", event->event);
|
2018-07-03 18:01:49 +02:00
|
|
|
}
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2018-07-07 13:08:08 +02:00
|
|
|
rdma_ack_cm_event(event);
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2018-07-08 15:00:47 +02:00
|
|
|
if (ret)
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
2018-07-03 18:01:49 +02:00
|
|
|
}
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2019-04-22 23:45:38 +02:00
|
|
|
return nullptr;
|
2018-06-28 12:46:16 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int ib_start(struct vnode *n)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-03 18:01:49 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
int ret;
|
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 1, "Started ib_start");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Create event channel */
|
2018-07-07 12:49:22 +02:00
|
|
|
ib->ctx.ec = rdma_create_event_channel();
|
|
|
|
if (!ib->ctx.ec)
|
2018-07-03 18:01:49 +02:00
|
|
|
error("Failed to create event channel in node %s!", node_name(n));
|
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 3, "Created event channel");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Create rdma_cm_id and bind to device */
|
2018-07-15 16:37:52 +02:00
|
|
|
ib_create_bind_id(n);
|
2018-07-07 15:34:07 +02:00
|
|
|
|
2018-07-21 12:07:43 +02:00
|
|
|
debug(LOG_IB | 3, "Initialized Work Completion Buffer");
|
2018-07-07 15:34:07 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Resolve address or listen to rdma_cm_id */
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ib->is_source) {
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Resolve address */
|
2019-04-22 23:45:38 +02:00
|
|
|
ret = rdma_resolve_addr(ib->ctx.id, nullptr, ib->conn.dst_addr->ai_addr, ib->conn.timeout);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-03 18:01:49 +02:00
|
|
|
error("Failed to resolve remote address after %ims of node %s: %s",
|
|
|
|
ib->conn.timeout, node_name(n), gai_strerror(ret));
|
|
|
|
}
|
2018-07-04 19:04:08 +02:00
|
|
|
else {
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Listen on rdma_cm_id for events */
|
2018-07-03 18:01:49 +02:00
|
|
|
ret = rdma_listen(ib->ctx.listen_id, 10);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-03 18:01:49 +02:00
|
|
|
error("Failed to listen to rdma_cm_id on node %s", node_name(n));
|
2018-07-05 13:57:25 +02:00
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Started to listen to rdma_cm_id");
|
2018-07-03 18:01:49 +02:00
|
|
|
}
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Allocate protection domain */
|
2018-07-07 12:49:22 +02:00
|
|
|
ib->ctx.pd = ibv_alloc_pd(ib->ctx.id->verbs);
|
|
|
|
if (!ib->ctx.pd)
|
|
|
|
error("Could not allocate protection domain in node %s", node_name(n));
|
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Allocated Protection Domain");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Allocate space for 40 Byte GHR. We don't use this. */
|
2018-07-19 20:33:41 +02:00
|
|
|
if (ib->conn.port_space == RDMA_PS_UDP) {
|
2020-01-21 16:26:51 +01:00
|
|
|
ib->conn.ud.grh_ptr = new char[GRH_SIZE];
|
2020-07-04 16:22:10 +02:00
|
|
|
if (!ib->conn.ud.grh_ptr)
|
|
|
|
throw MemoryAllocationError();
|
|
|
|
|
2018-08-01 18:26:42 +02:00
|
|
|
ib->conn.ud.grh_mr = ibv_reg_mr(ib->ctx.pd, ib->conn.ud.grh_ptr, GRH_SIZE, IBV_ACCESS_LOCAL_WRITE);
|
2018-07-19 20:33:41 +02:00
|
|
|
}
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Several events should occur on the event channel, to make
|
|
|
|
* sure the nodes are succesfully connected.
|
|
|
|
*/
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 1, "Starting to monitor events on rdma_cm_id");
|
2018-07-03 18:01:49 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Create thread to monitor rdma_cm_event channel */
|
2019-04-22 23:45:38 +02:00
|
|
|
ret = pthread_create(&ib->conn.rdma_cm_event_thread, nullptr, ib_rdma_cm_event_thread, n);
|
2018-07-04 19:04:08 +02:00
|
|
|
if (ret)
|
2018-07-07 12:49:22 +02:00
|
|
|
error("Failed to create thread to monitor rdma_cm events in node %s: %s",
|
2018-07-03 18:01:49 +02:00
|
|
|
node_name(n), gai_strerror(ret));
|
|
|
|
|
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int ib_stop(struct vnode *n)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-03 17:39:06 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
|
|
|
int ret;
|
2018-06-28 12:46:16 +02:00
|
|
|
|
2018-07-07 14:36:23 +02:00
|
|
|
debug(LOG_IB | 1, "Called ib_stop");
|
|
|
|
|
|
|
|
ib->stopThreads = 1;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Call RDMA disconnect function
|
|
|
|
* Will flush all outstanding WRs to the Completion Queue and
|
|
|
|
* will call RDMA_CM_EVENT_DISCONNECTED if that is done.
|
|
|
|
*/
|
2019-06-23 16:13:23 +02:00
|
|
|
if (n->state == State::CONNECTED && ib->conn.port_space != RDMA_PS_UDP) {
|
2018-07-07 15:34:07 +02:00
|
|
|
ret = rdma_disconnect(ib->ctx.id);
|
2018-07-04 19:04:08 +02:00
|
|
|
|
2018-07-14 16:46:23 +02:00
|
|
|
if (ret)
|
|
|
|
error("Error while calling rdma_disconnect in node %s: %s",
|
|
|
|
node_name(n), gai_strerror(ret));
|
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Called rdma_disconnect");
|
|
|
|
}
|
|
|
|
else {
|
2018-09-09 12:37:46 +02:00
|
|
|
pthread_cancel(ib->conn.rdma_cm_event_thread);
|
2018-07-14 16:46:23 +02:00
|
|
|
|
2018-09-09 12:37:46 +02:00
|
|
|
debug(LOG_IB | 3, "Called pthread_cancel() on communication management thread.");
|
2018-07-14 16:46:23 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
info("Disconnecting... Waiting for threads to join.");
|
2018-07-07 14:36:23 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Wait for event thread to join */
|
2019-04-22 23:45:38 +02:00
|
|
|
ret = pthread_join(ib->conn.rdma_cm_event_thread, nullptr);
|
2018-07-07 14:36:23 +02:00
|
|
|
if (ret)
|
|
|
|
error("Error while joining rdma_cm_event_thread in node %s: %i", node_name(n), ret);
|
|
|
|
|
|
|
|
debug(LOG_IB | 3, "Joined rdma_cm_event_thread");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Destroy RDMA CM ID */
|
2018-07-07 14:36:23 +02:00
|
|
|
rdma_destroy_id(ib->ctx.id);
|
|
|
|
debug(LOG_IB | 3, "Destroyed rdma_cm_id");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Dealloc Protection Domain */
|
2018-07-07 14:36:23 +02:00
|
|
|
ibv_dealloc_pd(ib->ctx.pd);
|
|
|
|
debug(LOG_IB | 3, "Destroyed protection domain");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Destroy event channel */
|
2018-07-07 14:36:23 +02:00
|
|
|
rdma_destroy_event_channel(ib->ctx.ec);
|
|
|
|
debug(LOG_IB | 3, "Destroyed event channel");
|
2018-06-30 18:20:30 +02:00
|
|
|
|
2018-07-07 14:36:23 +02:00
|
|
|
info("Successfully stopped %s", node_name(n));
|
2018-06-28 12:46:16 +02:00
|
|
|
|
2018-07-03 17:39:06 +02:00
|
|
|
return 0;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int ib_read(struct vnode *n, struct sample *smps[], unsigned cnt, unsigned *release)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-01 12:56:03 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
2018-07-12 17:49:17 +02:00
|
|
|
struct ibv_wc wc[cnt];
|
2019-04-22 23:45:38 +02:00
|
|
|
struct ibv_recv_wr wr[cnt], *bad_wr = nullptr;
|
2018-07-21 12:52:25 +02:00
|
|
|
struct ibv_sge sge[cnt][ib->qp_init.cap.max_recv_sge];
|
2018-07-04 19:04:08 +02:00
|
|
|
struct ibv_mr *mr;
|
2018-07-21 12:52:25 +02:00
|
|
|
struct timespec ts_receive;
|
2018-07-12 17:49:17 +02:00
|
|
|
int ret = 0, wcs = 0, read_values = 0, max_wr_post;
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 15, "ib_read is called");
|
|
|
|
|
2019-06-23 16:13:23 +02:00
|
|
|
if (n->state == State::CONNECTED || n->state == State::PENDING_CONNECT) {
|
2018-07-03 17:39:06 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
max_wr_post = cnt;
|
2018-07-03 17:39:06 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Poll Completion Queue
|
|
|
|
* If we've already posted enough receive WRs, try to pull cnt
|
|
|
|
*/
|
2018-07-15 13:51:18 +02:00
|
|
|
if (ib->conn.available_recv_wrs >= (ib->qp_init.cap.max_recv_wr - ib->conn.buffer_subtraction) ) {
|
2018-07-26 15:46:41 +02:00
|
|
|
for (int i = 0;; i++) {
|
2018-08-01 18:26:42 +02:00
|
|
|
if (i % CHK_PER_ITER == CHK_PER_ITER - 1) pthread_testcancel();
|
2018-07-26 15:46:41 +02:00
|
|
|
|
2019-06-23 16:13:23 +02:00
|
|
|
/* If IB node disconnects or if it is still in State::PENDING_CONNECT, ib_read
|
2019-03-26 07:09:55 +01:00
|
|
|
* should return immediately if this condition holds
|
|
|
|
*/
|
2019-06-23 16:13:23 +02:00
|
|
|
if (n->state != State::CONNECTED) return 0;
|
Node gives back samples to framework at disconnect
The node blocks a certain amount of samples to use in its queues.
Before this commit, the only moment to release them to the framwork was
during ib_read()/ib_write().
But, there were a couple of problems. In the following I will take
ib_read() as example, but ib_write() will be analogous.
The first problem was:
1. If a QP disconnect, all Work Requests get invalidated and will be
"flushed" to a Completion Queue.
A possible solution would be, to save them in an intermediate buffer.
We could then "exchange" these samples with the framework as soon as the node
connects again and ib_read() is called again. So, we would get valid
samples from the framwork, post them, and give the "invalidated" samples back.
But, there is a second problem:
2. We cannot assume that ib_read() is ever called again after
ib_disconnect(). This is for example the case if the disconnect is
triggered by ib_stop() and not by an external node that disconnects.
This would result in a memory leak, since the samples would never be
returned to the framework, although the node is stopped.
Because of this second problem, I decided to return all samples with
sample_put() in the disconnect function. An additional benefit is that
this is more convenient than another buffer to temporarily safe the
invalidated samples.
2018-07-16 13:41:53 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
wcs = ibv_poll_cq(ib->ctx.recv_cq, cnt, wc);
|
|
|
|
if (wcs) {
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Get time directly after something arrived in Completion Queue */
|
2018-07-21 12:52:25 +02:00
|
|
|
ts_receive = time_now();
|
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
debug(LOG_IB | 10, "Received %i Work Completions", wcs);
|
2018-07-03 17:39:06 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
read_values = wcs; /* Value to return */
|
|
|
|
max_wr_post = wcs; /* Make space free in smps[] */
|
2018-07-08 15:00:47 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* All samples (wcs * received + unposted) should be released. Let
|
|
|
|
* *release be equal to allocated.
|
|
|
|
*
|
|
|
|
* This is set in the framework, before this function was called.
|
|
|
|
*/
|
2018-07-12 17:49:17 +02:00
|
|
|
}
|
|
|
|
else {
|
|
|
|
ib->conn.available_recv_wrs += max_wr_post;
|
2019-03-26 07:09:55 +01:00
|
|
|
*release = 0; /* While we fill the receive queue, we always use all samples */
|
2018-07-07 12:49:22 +02:00
|
|
|
}
|
2018-07-05 13:57:25 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Get Memory Region */
|
2018-08-13 14:07:29 +02:00
|
|
|
mr = memory_ib_get_mr(pool_buffer(sample_pool(smps[0])));
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
for (int i = 0; i < max_wr_post; i++) {
|
2018-07-20 22:55:33 +02:00
|
|
|
int j = 0;
|
2018-07-19 20:33:41 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Prepare receive Scatter/Gather element */
|
2018-07-21 12:52:25 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* First 40 byte of UD data are GRH and unused in our case */
|
2018-07-19 20:33:41 +02:00
|
|
|
if (ib->conn.port_space == RDMA_PS_UDP) {
|
2018-07-25 18:51:28 +02:00
|
|
|
sge[i][j].addr = (uint64_t) ib->conn.ud.grh_ptr;
|
2018-08-01 18:26:42 +02:00
|
|
|
sge[i][j].length = GRH_SIZE;
|
2018-07-25 18:51:28 +02:00
|
|
|
sge[i][j].lkey = ib->conn.ud.grh_mr->lkey;
|
2018-07-20 22:55:33 +02:00
|
|
|
|
|
|
|
j++;
|
2018-07-19 20:33:41 +02:00
|
|
|
}
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Sequence */
|
2018-07-21 12:52:25 +02:00
|
|
|
sge[i][j].addr = (uint64_t) &smps[i]->sequence;
|
|
|
|
sge[i][j].length = sizeof(smps[i]->sequence);
|
|
|
|
sge[i][j].lkey = mr->lkey;
|
|
|
|
|
|
|
|
j++;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Timespec origin */
|
2018-07-21 12:52:25 +02:00
|
|
|
sge[i][j].addr = (uint64_t) &smps[i]->ts.origin;
|
|
|
|
sge[i][j].length = sizeof(smps[i]->ts.origin);
|
|
|
|
sge[i][j].lkey = mr->lkey;
|
|
|
|
|
|
|
|
j++;
|
|
|
|
|
2018-07-20 22:55:33 +02:00
|
|
|
sge[i][j].addr = (uint64_t) &smps[i]->data;
|
2018-08-02 10:43:49 +02:00
|
|
|
sge[i][j].length = SAMPLE_DATA_LENGTH(DEFAULT_SAMPLE_LENGTH);
|
2018-07-20 22:55:33 +02:00
|
|
|
sge[i][j].lkey = mr->lkey;
|
|
|
|
|
|
|
|
j++;
|
2018-07-12 17:49:17 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Prepare a receive Work Request */
|
2018-07-12 17:49:17 +02:00
|
|
|
wr[i].wr_id = (uintptr_t) smps[i];
|
|
|
|
wr[i].next = &wr[i+1];
|
2018-07-19 20:33:41 +02:00
|
|
|
wr[i].sg_list = sge[i];
|
2018-07-20 22:55:33 +02:00
|
|
|
wr[i].num_sge = j;
|
2018-07-12 17:49:17 +02:00
|
|
|
}
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2019-04-22 23:45:38 +02:00
|
|
|
wr[max_wr_post-1].next = nullptr;
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
debug(LOG_IB | 5, "Prepared %i new receive Work Requests", max_wr_post);
|
|
|
|
debug(LOG_IB | 5, "%i receive Work Requests in Receive Queue", ib->conn.available_recv_wrs);
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Post list of Work Requests */
|
2018-07-12 17:49:17 +02:00
|
|
|
ret = ibv_post_recv(ib->ctx.id->qp, &wr[0], &bad_wr);
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
if (ret)
|
2019-06-18 18:51:52 +01:00
|
|
|
error("Was unable to post receive WR in node %s: %i, bad WR ID: 0x%" PRIu64,
|
2018-07-12 17:49:17 +02:00
|
|
|
node_name(n), ret, bad_wr->wr_id);
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
debug(LOG_IB | 10, "Succesfully posted receive Work Requests");
|
2018-07-08 14:05:48 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Doesn't start if wcs == 0 */
|
2018-07-12 17:49:17 +02:00
|
|
|
for (int j = 0; j < wcs; j++) {
|
2018-08-04 17:34:52 +02:00
|
|
|
if ( !( (wc[j].opcode & IBV_WC_RECV) && wc[j].status == IBV_WC_SUCCESS) ) {
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Drop all values, we don't know where the error occured */
|
2018-08-04 17:34:52 +02:00
|
|
|
read_values = 0;
|
|
|
|
}
|
2018-07-08 14:05:48 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
if (wc[j].status == IBV_WC_WR_FLUSH_ERR)
|
|
|
|
debug(LOG_IB | 5, "Received IBV_WC_WR_FLUSH_ERR (ib_read). Ignore it.");
|
|
|
|
else if (wc[j].status != IBV_WC_SUCCESS)
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Work Completion status was not IBV_WC_SUCCES in node %s: %i",
|
2018-07-12 17:49:17 +02:00
|
|
|
node_name(n), wc[j].status);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* 32 byte of meta data is always transferred. We should substract it.
|
|
|
|
* Furthermore, in case of an unreliable connection, a 40 byte
|
|
|
|
* global routing header is transferred. This should be substracted as well.
|
|
|
|
*/
|
2018-08-01 18:26:42 +02:00
|
|
|
int correction = (ib->conn.port_space == RDMA_PS_UDP) ? META_GRH_SIZE : META_SIZE;
|
2018-07-19 20:33:41 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
smps[j] = (struct sample *) (wc[j].wr_id);
|
2018-08-04 17:34:52 +02:00
|
|
|
|
2018-10-11 14:18:30 +02:00
|
|
|
smps[j]->length = SAMPLE_NUMBER_OF_VALUES(wc[j].byte_len - correction);
|
2018-07-21 12:52:25 +02:00
|
|
|
smps[j]->ts.received = ts_receive;
|
2019-06-23 16:13:23 +02:00
|
|
|
smps[j]->flags = (int) SampleFlags::HAS_TS_ORIGIN | (int) SampleFlags::HAS_TS_RECEIVED | (int) SampleFlags::HAS_SEQUENCE;
|
2020-07-06 15:07:05 +02:00
|
|
|
smps[j]->signals = &n->in.signals;
|
2018-07-12 17:49:17 +02:00
|
|
|
}
|
2018-10-11 14:18:30 +02:00
|
|
|
|
2018-07-08 14:05:48 +02:00
|
|
|
}
|
2018-07-12 17:49:17 +02:00
|
|
|
return read_values;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int ib_write(struct vnode *n, struct sample *smps[], unsigned cnt, unsigned *release)
|
2018-06-21 11:56:28 +02:00
|
|
|
{
|
2018-07-01 12:56:03 +02:00
|
|
|
struct infiniband *ib = (struct infiniband *) n->_vd;
|
2019-04-22 23:45:38 +02:00
|
|
|
struct ibv_send_wr wr[cnt], *bad_wr = nullptr;
|
2018-07-21 12:52:25 +02:00
|
|
|
struct ibv_sge sge[cnt][ib->qp_init.cap.max_recv_sge];
|
2018-07-12 17:49:17 +02:00
|
|
|
struct ibv_wc wc[cnt];
|
2018-07-04 19:04:08 +02:00
|
|
|
struct ibv_mr *mr;
|
2018-07-12 17:49:17 +02:00
|
|
|
|
2018-07-01 12:56:03 +02:00
|
|
|
int ret;
|
2019-04-22 23:43:46 +02:00
|
|
|
unsigned sent = 0; /* Used for first loop: prepare work requests to post to send queue */
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2018-07-05 13:57:25 +02:00
|
|
|
debug(LOG_IB | 10, "ib_write is called");
|
|
|
|
|
2019-06-23 16:13:23 +02:00
|
|
|
if (n->state == State::CONNECTED) {
|
2018-07-23 16:52:38 +02:00
|
|
|
*release = 0;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* First, write */
|
2018-07-05 13:57:25 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Get Memory Region */
|
2018-08-13 14:07:29 +02:00
|
|
|
mr = memory_ib_get_mr(pool_buffer(sample_pool(smps[0])));
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
for (sent = 0; sent < cnt; sent++) {
|
2018-07-21 12:52:25 +02:00
|
|
|
int j = 0;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set Scatter/Gather element to data of sample */
|
|
|
|
|
|
|
|
/* Sequence */
|
2018-07-21 12:52:25 +02:00
|
|
|
sge[sent][j].addr = (uint64_t) &smps[sent]->sequence;
|
|
|
|
sge[sent][j].length = sizeof(smps[sent]->sequence);
|
|
|
|
sge[sent][j].lkey = mr->lkey;
|
|
|
|
|
|
|
|
j++;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Timespec origin */
|
2018-07-21 12:52:25 +02:00
|
|
|
sge[sent][j].addr = (uint64_t) &smps[sent]->ts.origin;
|
|
|
|
sge[sent][j].length = sizeof(smps[sent]->ts.origin);
|
|
|
|
sge[sent][j].lkey = mr->lkey;
|
|
|
|
|
|
|
|
j++;
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Actual Payload */
|
2018-07-21 12:52:25 +02:00
|
|
|
sge[sent][j].addr = (uint64_t) &smps[sent]->data;
|
2018-10-11 14:18:30 +02:00
|
|
|
sge[sent][j].length = SAMPLE_DATA_LENGTH(smps[sent]->length);
|
2018-07-21 12:52:25 +02:00
|
|
|
sge[sent][j].lkey = mr->lkey;
|
|
|
|
|
|
|
|
j++;
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Check if connection is connected or unconnected and set appropriate values */
|
2018-07-19 18:47:27 +02:00
|
|
|
if (ib->conn.port_space == RDMA_PS_UDP) {
|
2018-07-25 18:51:28 +02:00
|
|
|
wr[sent].wr.ud.ah = ib->conn.ud.ah;
|
|
|
|
wr[sent].wr.ud.remote_qkey = ib->conn.ud.ud.qkey;
|
|
|
|
wr[sent].wr.ud.remote_qpn = ib->conn.ud.ud.qp_num;
|
2018-07-19 18:32:06 +02:00
|
|
|
}
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Check if data can be send inline
|
|
|
|
* 32 byte meta data is always send.
|
|
|
|
* Once every max_send_wr iterations a signal must be generated. Since we would need
|
|
|
|
* an additional buffer if we were sending inlines with IBV_SEND_SIGNALED, we prefer
|
|
|
|
* to send one samples every max_send_wr NOT inline (which thus generates a signal).
|
|
|
|
*/
|
2018-10-21 12:35:21 +02:00
|
|
|
int send_inline = ((sge[sent][j-1].length + META_SIZE) < ib->qp_init.cap.max_inline_data)
|
2018-11-02 12:46:12 +01:00
|
|
|
&& ((++ib->signaling_counter % ib->periodic_signaling) != 0) ?
|
2018-10-21 12:35:21 +02:00
|
|
|
ib->conn.send_inline : 0;
|
2018-07-13 13:50:30 +02:00
|
|
|
|
|
|
|
debug(LOG_IB | 10, "Sample will be send inline [0/1]: %i", send_inline);
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Set Send Work Request */
|
2018-10-20 17:05:52 +02:00
|
|
|
wr[sent].wr_id = (uintptr_t) smps[sent];
|
2018-07-21 12:52:25 +02:00
|
|
|
wr[sent].sg_list = sge[sent];
|
|
|
|
wr[sent].num_sge = j;
|
2018-07-14 15:20:24 +02:00
|
|
|
wr[sent].next = &wr[sent+1];
|
2018-07-07 12:49:22 +02:00
|
|
|
|
2018-10-20 17:05:52 +02:00
|
|
|
wr[sent].send_flags = send_inline ? IBV_SEND_INLINE : IBV_SEND_SIGNALED;
|
2018-08-01 18:26:42 +02:00
|
|
|
wr[sent].opcode = IBV_WR_SEND;
|
2018-07-05 13:57:25 +02:00
|
|
|
}
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2018-07-14 15:20:24 +02:00
|
|
|
debug(LOG_IB | 10, "Prepared %i send Work Requests", cnt);
|
2019-04-22 23:45:38 +02:00
|
|
|
wr[cnt-1].next = nullptr;
|
2018-07-14 15:20:24 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Send linked list of Work Requests */
|
2018-07-07 12:49:22 +02:00
|
|
|
ret = ibv_post_send(ib->ctx.id->qp, wr, &bad_wr);
|
2018-07-12 17:49:17 +02:00
|
|
|
debug(LOG_IB | 4, "Posted send Work Requests");
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Reorder list. Place inline and unposted samples to the top
|
|
|
|
* m will always be equal or smaller than *release
|
|
|
|
*/
|
2019-04-22 23:43:46 +02:00
|
|
|
for (unsigned m = 0; m < cnt; m++) {
|
2019-03-26 07:09:55 +01:00
|
|
|
/* We can't use wr_id as identifier, since it is 0 for inline
|
|
|
|
* elements
|
|
|
|
*/
|
2018-07-12 17:49:17 +02:00
|
|
|
if (ret && (wr[m].sg_list == bad_wr->sg_list)) {
|
2019-03-26 07:09:55 +01:00
|
|
|
/* The remaining work requests will be bad. Ripple through list
|
|
|
|
* and prepare them to be released
|
|
|
|
*/
|
2019-06-18 18:51:52 +01:00
|
|
|
debug(LOG_IB | 4, "Bad WR occured with ID: 0x%" PRIu64 " and S/G address: 0x%px: %i",
|
2018-07-12 17:49:17 +02:00
|
|
|
bad_wr->wr_id, bad_wr->sg_list, ret);
|
|
|
|
|
|
|
|
while (1) {
|
|
|
|
smps[*release] = smps[m];
|
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
(*release)++; /* Increment number of samples to be released */
|
|
|
|
sent--; /* Decrement the number of succesfully posted elements */
|
2018-07-12 17:49:17 +02:00
|
|
|
|
|
|
|
if (++m == cnt) break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
else if (wr[m].send_flags & IBV_SEND_INLINE) {
|
|
|
|
smps[*release] = smps[m];
|
|
|
|
|
|
|
|
(*release)++;
|
|
|
|
}
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2018-07-07 12:49:22 +02:00
|
|
|
}
|
2018-07-01 12:56:03 +02:00
|
|
|
|
2018-07-21 12:07:43 +02:00
|
|
|
debug(LOG_IB | 4, "%i samples will be released (before WC)", *release);
|
2018-07-08 15:00:47 +02:00
|
|
|
|
2019-03-26 07:09:55 +01:00
|
|
|
/* Try to grab as many CQEs from CQ as there is space in *smps[] */
|
2018-10-20 17:05:52 +02:00
|
|
|
ret = ibv_poll_cq(ib->ctx.send_cq, cnt - *release, wc);
|
2018-07-08 15:00:47 +02:00
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
for (int i = 0; i < ret; i++) {
|
|
|
|
if (wc[i].status != IBV_WC_SUCCESS && wc[i].status != IBV_WC_WR_FLUSH_ERR)
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Work Completion status was not IBV_WC_SUCCES in node %s: %i",
|
2018-07-12 17:49:17 +02:00
|
|
|
node_name(n), wc[i].status);
|
2018-07-08 15:00:47 +02:00
|
|
|
|
2018-10-20 17:05:52 +02:00
|
|
|
smps[*release] = (struct sample *) (wc[i].wr_id);
|
|
|
|
(*release)++;
|
2018-07-08 15:00:47 +02:00
|
|
|
}
|
|
|
|
|
2018-07-21 12:07:43 +02:00
|
|
|
debug(LOG_IB | 4, "%i samples will be released (after WC)", *release);
|
2018-07-01 12:56:03 +02:00
|
|
|
}
|
|
|
|
|
2018-07-12 17:49:17 +02:00
|
|
|
return sent;
|
2018-06-21 11:56:28 +02:00
|
|
|
}
|
|
|
|
|
2019-04-23 00:36:06 +02:00
|
|
|
static struct plugin p;
|
|
|
|
|
|
|
|
__attribute__((constructor(110)))
|
|
|
|
static void register_plugin() {
|
|
|
|
p.name = "infiniband";
|
|
|
|
p.description = "Infiniband interface (libibverbs, librdmacm)";
|
2019-06-23 16:13:23 +02:00
|
|
|
p.type = PluginType::NODE;
|
|
|
|
p.node.instances.state = State::DESTROYED;
|
2019-04-23 00:36:06 +02:00
|
|
|
p.node.vectorize = 0;
|
|
|
|
p.node.size = sizeof(struct infiniband);
|
|
|
|
p.node.pool_size = 8192;
|
|
|
|
p.node.destroy = ib_destroy;
|
|
|
|
p.node.parse = ib_parse;
|
|
|
|
p.node.check = ib_check;
|
|
|
|
p.node.print = ib_print;
|
|
|
|
p.node.start = ib_start;
|
|
|
|
p.node.stop = ib_stop;
|
|
|
|
p.node.read = ib_read;
|
|
|
|
p.node.write = ib_write;
|
|
|
|
p.node.reverse = ib_reverse;
|
|
|
|
p.node.memory_type = memory_ib;
|
|
|
|
|
2020-09-10 11:11:42 +02:00
|
|
|
int ret = vlist_init(&p.node.instances);
|
|
|
|
if (!ret)
|
|
|
|
vlist_init_and_push(&plugins, &p);
|
2019-04-23 00:36:06 +02:00
|
|
|
}
|
2018-06-21 11:56:28 +02:00
|
|
|
|
2019-04-23 00:36:06 +02:00
|
|
|
__attribute__((destructor(110)))
|
|
|
|
static void deregister_plugin() {
|
2020-06-16 02:35:34 +02:00
|
|
|
vlist_remove_all(&plugins, &p);
|
2019-04-23 13:15:00 +02:00
|
|
|
}
|