Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

TL/MLX5: complete the multi mcast group design #1075

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

TL/MLX5: complete the multi mcast group design #1075

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 26 additions & 19 deletions src/components/tl/mlx5/mcast/tl_mlx5_mcast.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ typedef struct ucc_tl_mlx5_mcast_coll_comm_init_spec {
int post_recv_thresh;
int scq_moderation;
int wsize;
int mcast_group_count;
int max_push_send;
int max_eager;
int cuda_mem_enabled;
Expand Down Expand Up @@ -176,8 +177,8 @@ typedef struct ucc_tl_mlx5_mcast_coll_context {

typedef struct ucc_tl_mlx5_mcast_join_info_t {
ucc_status_t status;
uint16_t dlid;
union ibv_gid dgid;
uint16_t dlid[MAX_GROUP_COUNT];
union ibv_gid dgid[MAX_GROUP_COUNT];
} ucc_tl_mlx5_mcast_join_info_t;

typedef struct ucc_tl_mlx5_mcast_context {
Expand Down Expand Up @@ -468,32 +469,38 @@ static inline ucc_status_t ucc_tl_mlx5_mcast_post_recv_buffers(ucc_tl_mlx5_mcast
struct ibv_recv_wr *rwr = comm->call_rwr;
struct ibv_sge *sge = comm->call_rsgs;
struct pp_packet *pp = NULL;
int count = comm->params.rx_depth - comm->pending_recv;
int i;
int j;
int count;
int count_per_qp;

count = comm->params.rx_depth - comm->pending_recv;
if (count <= comm->params.post_recv_thresh) {
return UCC_OK;
}

for (i = 0; i < count - 1; i++) {
if (NULL == (pp = ucc_tl_mlx5_mcast_buf_get_free(comm))) {
break;
count_per_qp = count / comm->mcast_group_count;
for (j = 0; j < comm->mcast_group_count; j++) {
for (i = 0; i < count_per_qp; i++) {
if (NULL == (pp = ucc_tl_mlx5_mcast_buf_get_free(comm))) {
break;
}
rwr[i].wr_id = ((uint64_t) pp);
rwr[i].next = &rwr[i+1];
sge[2*i + 1].addr = pp->buf;
assert((uint64_t)comm->pp <= rwr[i].wr_id
&& ((uint64_t)comm->pp + comm->buf_n * sizeof(struct pp_packet)) > rwr[i].wr_id);
}

rwr[i].wr_id = ((uint64_t) pp);
rwr[i].next = &rwr[i+1];
sge[2*i + 1].addr = pp->buf;

ucc_assert((uint64_t)comm->pp <= rwr[i].wr_id
&& ((uint64_t)comm->pp + comm->buf_n * sizeof(struct pp_packet)) > rwr[i].wr_id);
}
if (i != 0) {
rwr[i-1].next = NULL;
if (ibv_post_recv(comm->mcast.groups[0].qp, &rwr[0], &bad_wr)) {
tl_error(comm->lib, "failed to prepost recvs: errno %d", errno);
return UCC_ERR_NO_RESOURCE;
if (i > 0) {
rwr[i-1].next = NULL;
if (ibv_post_recv(comm->mcast.groups[j].qp, &rwr[0], &bad_wr)) {
tl_error(comm->lib, "Failed to prepost recvs: errno %d qp index %d buffer count %d",
errno, j, i);
return UCC_ERR_NO_RESOURCE;
}
comm->pending_recv += i;
}
comm->pending_recv += i;
}

return UCC_OK;
Expand Down
65 changes: 26 additions & 39 deletions src/components/tl/mlx5/mcast/tl_mlx5_mcast_helper.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,7 @@ ucc_status_t ucc_tl_mlx5_probe_ip_over_ib(char* ib_dev, struct

ucc_status_t ucc_tl_mlx5_mcast_join_mcast_post(ucc_tl_mlx5_mcast_coll_context_t *ctx,
struct sockaddr_in6 *net_addr,
struct mcast_group *group,
int is_root)
{
char buf[40];
Expand All @@ -209,19 +210,18 @@ ucc_status_t ucc_tl_mlx5_mcast_join_mcast_post(ucc_tl_mlx5_mcast_coll_context_t
return UCC_ERR_NO_RESOURCE;
}

tl_debug(ctx->lib, "joining addr: %s is_root %d", buf, is_root);
tl_debug(ctx->lib, "joining addr: %s is_root %d group %p", buf, is_root, group);

if (rdma_join_multicast(ctx->id, (struct sockaddr*)net_addr, NULL)) {
if (rdma_join_multicast(ctx->id, (struct sockaddr*)net_addr, (void *)group)) {
tl_warn(ctx->lib, "rdma_join_multicast failed errno %d", errno);
return UCC_ERR_NO_RESOURCE;
}

return UCC_OK;
}

ucc_status_t ucc_tl_mlx5_mcast_join_mcast_test(ucc_tl_mlx5_mcast_coll_context_t *ctx,
struct rdma_cm_event **event,
int is_root)
ucc_status_t ucc_tl_mlx5_mcast_join_mcast_get_event(ucc_tl_mlx5_mcast_coll_context_t *ctx,
struct rdma_cm_event **event)
{
char buf[40];
const char *dst;
Expand All @@ -232,14 +232,15 @@ ucc_status_t ucc_tl_mlx5_mcast_join_mcast_test(ucc_tl_mlx5_mcast_coll_context_t
errno, strerror(errno));
return UCC_ERR_NO_RESOURCE;
} else {
/* need to retry again */
return UCC_INPROGRESS;
}
}

if (RDMA_CM_EVENT_MULTICAST_JOIN != (*event)->event) {
tl_warn(ctx->lib, "failed to join multicast, is_root %d. unexpected event was"
tl_warn(ctx->lib, "failed to join multicast, unexpected event was"
" received: event=%d, str=%s, status=%d",
is_root, (*event)->event, rdma_event_str((*event)->event),
(*event)->event, rdma_event_str((*event)->event),
(*event)->status);
if (rdma_ack_cm_event(*event) < 0) {
tl_warn(ctx->lib, "rdma_ack_cm_event failed");
Expand All @@ -253,32 +254,13 @@ ucc_status_t ucc_tl_mlx5_mcast_join_mcast_test(ucc_tl_mlx5_mcast_coll_context_t
return UCC_ERR_NO_RESOURCE;
}

tl_debug(ctx->lib, "is_root %d: joined dgid: %s, mlid 0x%x, sl %d", is_root, buf,
tl_debug(ctx->lib, "joined dgid: %s, mlid 0x%x, sl %d", buf,
(*event)->param.ud.ah_attr.dlid, (*event)->param.ud.ah_attr.sl);

return UCC_OK;

}

ucc_status_t ucc_tl_mlx5_setup_mcast_group_join_post(ucc_tl_mlx5_mcast_coll_comm_t *comm)
{
ucc_status_t status;
struct sockaddr_in6 net_addr = {0,};

if (comm->rank == 0) {
net_addr.sin6_family = AF_INET6;
net_addr.sin6_flowinfo = comm->comm_id;

status = ucc_tl_mlx5_mcast_join_mcast_post(comm->ctx, &net_addr, 1);
if (status < 0) {
tl_warn(comm->lib, "rank 0 is unable to join mcast group");
return status;
}
}

return UCC_OK;
}

ucc_status_t ucc_tl_mlx5_mcast_init_qps(ucc_tl_mlx5_mcast_coll_context_t *ctx,
ucc_tl_mlx5_mcast_coll_comm_t *comm)
{
Expand Down Expand Up @@ -571,23 +553,28 @@ ucc_status_t ucc_tl_mlx5_mcast_modify_rc_qps(ucc_tl_mlx5_mcast_coll_context_t *c
return UCC_OK;
}

ucc_status_t ucc_tl_mlx5_fini_mcast_group(ucc_tl_mlx5_mcast_coll_context_t *ctx,
ucc_tl_mlx5_mcast_coll_comm_t *comm)
ucc_status_t ucc_tl_mlx5_leave_mcast_group(ucc_tl_mlx5_mcast_coll_context_t *ctx,
ucc_tl_mlx5_mcast_coll_comm_t *comm)
{
char buf[40];
const char *dst;
int i;

dst = inet_ntop(AF_INET6, &comm->mcast.groups[0].mcast_addr, buf, 40);
if (NULL == dst) {
tl_error(comm->lib, "inet_ntop failed");
return UCC_ERR_NO_RESOURCE;
}
for (i = 0; i < comm->mcast_group_count; i++) {
if (comm->mcast.groups[i].mcast_addr.sin6_flowinfo != 0) {
dst = inet_ntop(AF_INET6, &comm->mcast.groups[i].mcast_addr, buf, 40);
if (NULL == dst) {
tl_error(comm->lib, "inet_ntop failed");
return UCC_ERR_NO_RESOURCE;
}

tl_debug(ctx->lib, "mcast leave: ctx %p, comm %p, dgid: %s", ctx, comm, buf);
tl_debug(ctx->lib, "mcast leave: ctx %p, comm %p, dgid: %s group %d", ctx, comm, buf, i);

if (rdma_leave_multicast(ctx->id, (struct sockaddr*)&comm->mcast.groups[0].mcast_addr)) {
tl_error(comm->lib, "mcast rmda_leave_multicast failed");
return UCC_ERR_NO_RESOURCE;
if (rdma_leave_multicast(ctx->id, (struct sockaddr*)&comm->mcast.groups[i].mcast_addr)) {
tl_error(comm->lib, "mcast rmda_leave_multicast failed for group %d", i);
return UCC_ERR_NO_RESOURCE;
}
}
}

return UCC_OK;
Expand Down Expand Up @@ -639,7 +626,7 @@ ucc_status_t ucc_tl_mlx5_clean_mcast_comm(ucc_tl_mlx5_mcast_coll_comm_t *comm)
}
}

status = ucc_tl_mlx5_fini_mcast_group(comm->ctx, comm);
status = ucc_tl_mlx5_leave_mcast_group(comm->ctx, comm);
if (status) {
tl_error(comm->lib, "couldn't leave mcast group");
return status;
Expand Down
14 changes: 10 additions & 4 deletions src/components/tl/mlx5/mcast/tl_mlx5_mcast_helper.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ static inline ucc_status_t ucc_tl_mlx5_mcast_send(ucc_tl_mlx5_mcast_coll_comm_t
int rc;
int length;
ucc_status_t status;
int mcast_group_index;
ucc_memory_type_t mem_type = comm->cuda_mem_enabled ? UCC_MEMORY_TYPE_CUDA
: UCC_MEMORY_TYPE_HOST;

Expand Down Expand Up @@ -89,6 +90,8 @@ static inline ucc_status_t ucc_tl_mlx5_mcast_send(ucc_tl_mlx5_mcast_coll_comm_t

ssg[0].length = length;
ssg[0].lkey = req->mr->lkey;
mcast_group_index = i % comm->mcast_group_count;
swr[0].wr.ud.ah = comm->mcast.groups[mcast_group_index].ah;
swr[0].wr_id = MCAST_BCASTSEND_WR;
swr[0].imm_data = htonl(pp->psn);
swr[0].send_flags = (length <= comm->max_inline) ? IBV_SEND_INLINE : 0;
Expand All @@ -108,7 +111,7 @@ static inline ucc_status_t ucc_tl_mlx5_mcast_send(ucc_tl_mlx5_mcast_coll_comm_t
tl_trace(comm->lib, "post_send, psn %d, length %d, zcopy %d, signaled %d",
pp->psn, pp->length, zcopy, swr[0].send_flags & IBV_SEND_SIGNALED);

if (0 != (rc = ibv_post_send(comm->mcast.groups[0].qp, &swr[0], &bad_wr))) {
if (0 != (rc = ibv_post_send(comm->mcast.groups[mcast_group_index].qp, &swr[0], &bad_wr))) {
tl_error(comm->lib, "post send failed: ret %d, start_psn %d, to_send %d, "
"to_recv %d, length %d, psn %d, inline %d",
rc, req->start_psn, req->to_send, req->to_recv,
Expand Down Expand Up @@ -577,10 +580,13 @@ ucc_status_t ucc_tl_mlx5_clean_mcast_comm(ucc_tl_mlx5_mcast_coll_comm_t *comm);

ucc_status_t ucc_tl_mlx5_mcast_join_mcast_post(ucc_tl_mlx5_mcast_coll_context_t *ctx,
struct sockaddr_in6 *net_addr,
struct mcast_group *group,
int is_root);

ucc_status_t ucc_tl_mlx5_mcast_join_mcast_test(ucc_tl_mlx5_mcast_coll_context_t *ctx,
struct rdma_cm_event **event,
int is_root);
ucc_status_t ucc_tl_mlx5_mcast_join_mcast_get_event(ucc_tl_mlx5_mcast_coll_context_t *ctx,
struct rdma_cm_event **event);

ucc_status_t ucc_tl_mlx5_leave_mcast_group(ucc_tl_mlx5_mcast_coll_context_t *ctx,
ucc_tl_mlx5_mcast_coll_comm_t *comm);

#endif /* TL_MLX5_MCAST_HELPER_H_ */
Loading
Loading