UCT/EFA/SRD: Add pending and control request and response #10604

tvegas1 · 2025-04-04T13:51:10Z

What?

Add pending management and control message exchange to remotely add AH, before posting RMA operations.

How?

Tracking remote AH state on iface or MD could cause issues as there could be multiple MD (protection domain) opened on the same remote HCA, and while locally iface or MD could remain, the remote entities could have been released and reopened with same QP id, making it difficult to identify if sending another control request is needed.

Steps

All posts are anyways heavily reordered so:

Only one pending queue for an endpoint
We put RMA on pending if confirmation has not been received, as they cannot be posted
We put AM on pending if our pending is not empty (we could probably simply post but this could cause prio issues wrt RMA)
We allow the sending of AM if pending is empty, even when still expecting confirmation of remote AH add
We can only start dequeuing pending after receiving confirmation of remote AH add, since we do not know which type of request is in the queue

dmitry-monakhov · 2025-04-05T08:02:48Z

Hi, how do you test it? I've tried ucx_iperf on p5e.48xlarge instance

env UCX_TLS=self,srd UCX_NET_DEVICES='rdmap113s0:1' /opt/ucx/bin/ucx_perftest -t stream_bw localhost
[1743838571.902753] [dmonakhov-dev:2912039:0]        perftest.c:800  UCX  WARN  CPU affinity is not set (bound to 192 cpus). Performance may be impacted.
+--------------+--------------+------------------------------+---------------------+-----------------------+
|              |              |       overhead (usec)        |   bandwidth (MB/s)  |  message rate (msg/s) |
+--------------+--------------+----------+---------+---------+----------+----------+-----------+-----------+
|    Stage     | # iterations | 50.0%ile | average | overall |  average |  overall |  average  |  overall  |
+--------------+--------------+----------+---------+---------+----------+----------+-----------+-----------+
Final:               1000000      0.610     0.771     0.771        9.90       9.90     1297377     1297377
[1743838582.024622] [dmonakhov-dev:2912039:0]           flush.c:58   UCX  ERROR req 0x627bc1fc5280: error during flush: Unsupported operation
[1743838582.024646] [dmonakhov-dev:2912039:0]         libperf.c:951  UCX  WARN  failed to close ep 0x79cae0f3e000 on thread 0: Unsupported operation

I've tried naive approach, simply ignore unsupported flush, it seems to work.

--- a/src/ucp/rma/flush.c
+++ b/src/ucp/rma/flush.c
@@ -127,7 +127,10 @@ static void ucp_ep_flush_progress(ucp_request_t *req)
                       ucs_status_string(status));
         if (status == UCS_OK) {
             ucp_ep_flush_request_update_uct_comp(req, -1, UCS_BIT(lane));
-        } else if (status == UCS_INPROGRESS) {
+        } else if (status == UCS_ERR_UNSUPPORTED) {
+         ucs_error("Ignore unsupported flush \n");
+         ucp_ep_flush_request_update_uct_comp(req, -1, UCS_BIT(lane));
+       } else if (status == UCS_INPROGRESS) {
             ucp_ep_flush_request_update_uct_comp(req, 0, UCS_BIT(lane));
         } else if (status == UCS_ERR_NO_RESOURCE) {
             if (req->send.lane != UCP_NULL_LANE) {

iyastreb · 2025-04-08T12:46:09Z

test/gtest/uct/ib/test_srd.cc

+{
+    for (auto i = 0; i < 3; i++) {
+        m_req[i].func = [](uct_pending_req_t*) {
+            m_count++;


if m_count is accessed only from this lambda, then you can make it local static variable

iyastreb · 2025-04-08T12:49:38Z

test/gtest/uct/ib/test_srd.cc

+
+    ASSERT_UCS_OK(uct_ep_pending_add(m_e1->ep(0), &m_req[0], 0));
+    ASSERT_UCS_OK(uct_ep_pending_add(m_e1->ep(0), &m_req[1], 0));
+    uct_ep_pending_purge(


this code is repeated 3 times, maybe make a helper function that purges and ASSERT_EQ

void pending_purge(ep, count)

fixed (was being added in next pr)

iyastreb · 2025-04-08T12:54:44Z

src/uct/ib/efa/srd/srd_def.h

+typedef enum uct_srd_ctl_id {
+    UCT_SRD_CTL_ID_REQ  = UCT_AM_ID_MAX,
+    UCT_SRD_CTL_ID_RESP = UCT_SRD_CTL_ID_REQ + 1,


cosmetic:

Suggested change

typedef enum uct_srd_ctl_id {

UCT_SRD_CTL_ID_REQ = UCT_AM_ID_MAX,

UCT_SRD_CTL_ID_RESP = UCT_SRD_CTL_ID_REQ + 1,

typedef enum {

UCT_SRD_CTL_ID_REQ = UCT_AM_ID_MAX,

UCT_SRD_CTL_ID_RESP,

iyastreb · 2025-04-08T12:55:45Z

src/uct/ib/efa/srd/srd_log.c

+    if ((hdr->id == UCT_SRD_CTL_ID_REQ) || (hdr->id == UCT_SRD_CTL_ID_RESP)) {
+        ctl = (uct_srd_ctl_hdr_t*)hdr;
+        snprintf(p, endp - p, " %s qpn %d ep_uuid %" PRIx64 " ",
+                 (ctl->id == UCT_SRD_CTL_ID_REQ)  ? "CTL_REQ" :


maybe it makes sense to implement function "_to_string" for this enum?

iyastreb · 2025-04-08T12:57:59Z

src/uct/ib/efa/srd/srd_log.c

+                 (ctl->id == UCT_SRD_CTL_ID_RESP) ? "CTL_RESP" :
+                                                    "UNKNOWN",
+                 uct_ib_unpack_uint24(ctl->qpn), ctl->ep_uuid);
+        p += strlen(p);


maybe increment by value returned from snprintf?
We can also assert on that value (>0 and <max)

fixed, kept strlen() as in non-debug (no assert) we need to add effective size

iyastreb · 2025-04-08T13:01:00Z

src/uct/ib/efa/srd/srd_log.c

    char *p, *endp;
    int am_id;

    p    = buffer;
    endp = buffer + max;

+    if ((hdr->id == UCT_SRD_CTL_ID_REQ) || (hdr->id == UCT_SRD_CTL_ID_RESP)) {
+        ctl = (uct_srd_ctl_hdr_t*)hdr;


I would propose to split this block into a separate function e.g. _dump_header, because it's really independent functionality.

fixed, but function is already small

iyastreb · 2025-04-08T13:02:58Z

src/uct/ib/efa/srd/srd_iface.c

+        goto out;
+    }
+
+    ucs_assertv(ctl->id == UCT_SRD_CTL_ID_REQ, "iface=%p id=%u", iface,


we can also use "_to_string" here

iyastreb · 2025-04-08T13:03:27Z

src/uct/ib/efa/srd/srd_iface.c

+    ucs_assertv(ctl->id == UCT_SRD_CTL_ID_REQ, "iface=%p id=%u", iface,
+                ctl->id);
+    ucs_assertv((sizeof(*ctl) + iface->super.addr_size) == length,
+                "req size mismatched expected=%zu received=%u",


Suggested change

"req size mismatched expected=%zu received=%u",

"req size mismatch expected=%zu received=%u",

iyastreb · 2025-04-08T13:07:18Z

src/uct/ib/efa/srd/srd_ep.c

+    ep->pending++;
+    uct_pending_req_arb_group_push(&ep->pending_group, req);
+    ucs_arbiter_group_schedule(&iface->tx.pending_q, &ep->pending_group);
+    ucs_trace_data("ep=%p: added pending req=%p psn=%u", ep, req, ep->psn);


print iface like in other places?

Suggested change

ucs_trace_data("ep=%p: added pending req=%p psn=%u", ep, req, ep->psn);

ucs_trace_data("iface=%p ep=%p: added pending req=%p psn=%u", ep, req, ep->psn);

iyastreb · 2025-04-08T13:09:19Z

src/uct/ib/efa/srd/srd_ep.c

+    uct_srd_iface_t *iface   = ucs_derived_of(tl_ep->iface, uct_srd_iface_t);
+    uct_purge_cb_args_t args = {cb, cb_arg};
+
+    ucs_trace_func("ep=%p", ep);


Seems like it's missing iface and some description, e.g. iface=%p ep=%p: purge

it's trace function, added some arguments

brminich · 2025-04-08T12:57:37Z

src/uct/ib/efa/srd/srd_def.h

+    ucs_list_link_t   list;     /* Entry in iface tx pending control list */
+    struct ibv_ah     *ah;
+    int               dest_qpn;
+    uct_srd_ctl_hdr_t hdr[];


minor: imo, you could also remove it, like with hdr

removed all hdr[]

brminich · 2025-04-08T13:00:56Z

src/uct/ib/efa/srd/srd_ep.h

+    uint64_t            ep_uuid;       /* Random EP identifier */
+    uint32_t            dest_qpn;      /* Remote QP */
+    uint32_t            inflight;      /* Entries outstanding list */
+    int32_t             pending;       /* Count requests in pending queue */


can it be negative?
seems like we do not really need this counter

src/uct/ib/efa/srd/srd_ep.h

src/uct/ib/efa/srd/srd_ep.c

brminich · 2025-04-08T13:07:39Z

src/uct/ib/efa/srd/srd_iface.h

@@ -60,6 +63,7 @@ typedef struct uct_srd_iface {

    struct {
        int32_t                      available;
+        int                          in_pending; /* true if invoked from arbiter dispatch */


minor: can be for assert only

brminich · 2025-04-08T13:11:20Z

src/uct/ib/efa/srd/srd_ep.c

+    uct_pending_req_t *req       = ucs_container_of(elem, uct_pending_req_t,
+                                                    priv);
+
+    /* Purge all of them, as requests were only added with pending_add */


what does this comment mean?

removed (some transports add their own pending_req_t internally, srd currently does not)

brminich · 2025-04-08T13:15:33Z

src/uct/ib/efa/srd/srd_ep.c

@@ -183,6 +284,10 @@ static UCS_F_ALWAYS_INLINE ucs_status_t uct_srd_ep_am_short_prepare(
    uct_srd_am_short_hdr_t *am = &iface->tx.am_inl_hdr;
    uct_srd_send_op_t *send_op;

+    if (uct_srd_ep_skip_pending(ep, iface)) {


why do not we check for resources instead?
Specifically for uct_srd_iface_can_tx(iface) && ep->ah_added

src/uct/ib/efa/srd/srd_ep.c

brminich · 2025-04-08T13:18:06Z

src/uct/ib/efa/srd/srd_iface.c

+    ctl_op = ucs_malloc(sizeof(*ctl_op) + sizeof(*ctl_op->hdr) +
+                            iface->super.addr_size,
+                        "uct_srd_ctl_op_t");


indented back line with iface->super.addr_size

brminich · 2025-04-08T13:22:28Z

src/uct/ib/efa/srd/srd_iface.c

@@ -535,8 +779,15 @@ uct_srd_iface_poll_rx(uct_srd_iface_t *iface)
    UCT_IB_IFACE_VERBS_FOREACH_RXWQE(&iface->super, i, packet, wc, num_wcs) {
        uct_ib_log_recv_completion(&iface->super, &wc[i], packet,
                                   wc[i].byte_len, uct_srd_dump_packet);
-        uct_srd_iface_process_rx(iface, (uct_srd_hdr_t*)packet, wc[i].byte_len,
-                                 (uct_srd_recv_desc_t*)wc[i].wr_id);
+        if ((*(uint8_t*)packet < UCT_AM_ID_MAX)) {


i'd add likely here

brminich · 2025-04-08T13:26:01Z

test/gtest/uct/ib/test_srd.cc

+        for (auto count = 10; count > 0; count--) {
+            short_progress_loop();
+        }


can we progress until ah_added flag on ep is set?

iyastreb · 2025-04-08T13:16:56Z

src/uct/ib/efa/srd/srd_iface.c

-        uct_srd_iface_process_rx(iface, (uct_srd_hdr_t*)packet, wc[i].byte_len,
-                                 (uct_srd_recv_desc_t*)wc[i].wr_id);
+        if ((*(uint8_t*)packet < UCT_AM_ID_MAX)) {
+            uct_srd_iface_process_rx(iface, (uct_srd_hdr_t*)packet,


I think this cast and one below are redundant

iyastreb · 2025-04-08T13:20:46Z

src/uct/ib/efa/srd/srd_iface.c

+    ucs_mpool_put(desc);
+    return;
+
+err:


IMO it's better to get rid of this goto and just print error in 2 places, but up to you

trying something else

iyastreb · 2025-04-08T13:26:48Z

src/uct/ib/efa/srd/srd_iface.c

+    khiter_t iter;
+    ucs_status_t status;
+
+    iter = kh_get(uct_srd_ep_hash, &iface->ep_hash, ep->ep_uuid);


what if we put unconditionally and then check the return value:

already exists/ error - return error, without goto

then handle sucsess case

brminich · 2025-04-10T11:28:35Z

src/uct/ib/efa/srd/srd_ep.c

+    hdr = (uct_srd_hdr_t *)(desc + 1);
+    length = pack_cb(hdr + 1, arg);


minor: align by =

i see it already aligned

brminich · 2025-04-10T11:29:48Z

src/uct/ib/efa/srd/srd_iface.c

    ctl_op->ah           = ah;
    ctl_op->dest_qpn     = dest_qpn;


pls remove extra whitespaces now

i see it already aligned

src/uct/ib/efa/srd/srd_ep.c

brminich · 2025-04-10T11:41:55Z

src/uct/ib/efa/srd/srd_ep.c

+
+    ucs_trace_data("iface=%p ep=%p progressing pending request %p", iface, ep,
+                   req);
+    iface->tx.in_pending = 1;


seems it can be done when asserts enabled only

brminich · 2025-04-10T11:45:50Z

src/uct/ib/efa/srd/srd_ep.c

+    if (!uct_srd_ep_can_tx(ep, iface)) {
+        UCS_STATS_UPDATE_COUNTER(ep->super.stats, UCT_EP_STAT_NO_RES, 1);
+        return UCS_ERR_NO_RESOURCE;
+    }
+
+    uct_srd_iface_check_pending(iface, &ep->pending_group);


can we combine it to one macro, as it is used in other places

not sure as there is not a single macro used everywhere. maybe we could add to uct_srd_ep_can_tx()

src/uct/ib/efa/srd/srd_iface.c

src/uct/ib/base/ib_log.c

src/uct/ib/efa/srd/srd_def.h

src/uct/ib/efa/srd/srd_ep.c

src/uct/ib/efa/srd/srd_iface.c

src/uct/ib/efa/srd/srd_iface.h

src/uct/ib/efa/srd/srd_ep.c

src/uct/ib/efa/srd/srd_iface.c

tvegas1 force-pushed the srd_pending_ctl branch from f0c2b44 to 3779e3a Compare April 4, 2025 15:09

iyastreb reviewed Apr 8, 2025

View reviewed changes

brminich reviewed Apr 8, 2025

View reviewed changes

iyastreb reviewed Apr 8, 2025

View reviewed changes

brminich reviewed Apr 10, 2025

View reviewed changes

src/uct/ib/efa/srd/srd_iface.c Show resolved Hide resolved

yosefe reviewed Apr 10, 2025

View reviewed changes

brminich previously approved these changes Apr 15, 2025

View reviewed changes

tvegas1 dismissed brminich’s stale review via 413b26e April 15, 2025 12:20

yosefe reviewed Apr 15, 2025

View reviewed changes

src/uct/ib/efa/srd/srd_ep.c Outdated Show resolved Hide resolved

src/uct/ib/efa/srd/srd_iface.c Outdated Show resolved Hide resolved

src/uct/ib/efa/srd/srd_iface.c Outdated Show resolved Hide resolved

tvegas1 force-pushed the srd_pending_ctl branch from babc1b1 to bd41d7f Compare April 15, 2025 13:29

yosefe approved these changes Apr 15, 2025

View reviewed changes

UCT/EFA/SRD: Add pending and control request and response

58e4a0d

tvegas1 force-pushed the srd_pending_ctl branch from bd41d7f to 58e4a0d Compare April 15, 2025 16:43

yosefe enabled auto-merge April 15, 2025 17:57

yosefe merged commit 8c53ab5 into openucx:master Apr 16, 2025
151 checks passed

	"req size mismatched expected=%zu received=%u",
	"req size mismatch expected=%zu received=%u",

	ucs_trace_data("ep=%p: added pending req=%p psn=%u", ep, req, ep->psn);
	ucs_trace_data("iface=%p ep=%p: added pending req=%p psn=%u", ep, req, ep->psn);

		hdr = (uct_srd_hdr_t *)(desc + 1);
		length = pack_cb(hdr + 1, arg);

UCT/EFA/SRD: Add pending and control request and response #10604

UCT/EFA/SRD: Add pending and control request and response #10604

Conversation

tvegas1 commented Apr 4, 2025

What?

How?

Steps

dmitry-monakhov commented Apr 5, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment