Repository navigation
UCP/RNDV: Throttle rndv fragment requests - #11062
ofirfarjun7 merged 76 commits into
Conversation
2be3040 to
b175673
Compare
amastbaum
added
Ready for Review
and removed
WIP-DNM
| ucp_proto_rndv_mtype_fc_increment(ucp_request_t *req) | ||
| { | ||
| ucp_worker_h worker = req->send.ep->worker; | ||
|
|
||
| if (worker->context->config.ext.rndv_ppln_worker_fc_enable) { | ||
| worker->rndv_ppln_fc.active_frags++; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Decrement active_frags counter and reschedule pending request if any. | ||
| */ | ||
| static UCS_F_ALWAYS_INLINE void | ||
| ucp_proto_rndv_mtype_fc_decrement(ucp_request_t *req) | ||
| { | ||
| ucp_worker_h worker = req->send.ep->worker; | ||
| ucp_context_h context = worker->context; | ||
|
|
||
| if (!context->config.ext.rndv_ppln_worker_fc_enable) { | ||
| return; | ||
| } | ||
|
|
||
| ucs_assert(worker->rndv_ppln_fc.active_frags > 0); | ||
| worker->rndv_ppln_fc.active_frags--; | ||
|
|
||
| if (!ucs_queue_is_empty(&worker->rndv_ppln_fc.pending_q)) { | ||
| ucp_request_t *pending_req; | ||
| ucs_queue_elem_t *elem; | ||
|
|
||
| elem = ucs_queue_pull(&worker->rndv_ppln_fc.pending_q); | ||
| pending_req = ucs_container_of(elem, ucp_request_t, | ||
| send.rndv.ppln.queue_elem); | ||
| ucs_callbackq_add_oneshot(&worker->uct->progress_q, pending_req, | ||
| ucp_proto_rndv_mtype_fc_reschedule_cb, | ||
| pending_req); | ||
| } | ||
| } |
There was a problem hiding this comment.
maybe it can be named better, or I should divide it differently
| {"RNDV_MTYPE_WORKER_MAX_FRAGS", "1024", | ||
| "Maximum number of concurrent mtype fragments per worker\n" | ||
| "(only applies when RNDV_MTYPE_WORKER_FC_ENABLE=y)", | ||
| ucs_offsetof(ucp_context_config_t, rndv_mtype_worker_max_frags), UCS_CONFIG_TYPE_ULUNITS}, |
There was a problem hiding this comment.
@shasson5 I am wondering if it's better to define RNDV_MTYPE_WORKER_MAX_FRAG_MEM instead, because the user shouldn't know how the amount of available memory is translated to the number of fragments.
The thing is that if we translate it in the code, it should be defined for each memory type (mem_type fragment size).
| worker->counters.ep_closures = 0; | ||
| worker->counters.ep_failures = 0; | ||
|
|
||
| /* Initialize RNDV mtype flow control */ |
There was a problem hiding this comment.
Do it conditionally, only if fc enabled?
| rpriv = req->send.proto_config->priv; | ||
|
|
||
| if (!(req->flags & UCP_REQUEST_FLAG_PROTO_INITIALIZED)) { | ||
| /* GET priority: 80% of total fragments */ |
There was a problem hiding this comment.
What is the meaning of 80%? Need to document/comment.
Will it always be a hard-coded magic value?
| * @return UCS_OK if not throttled, UCS_ERR_NO_RESOURCE if throttled and queued. | ||
| */ | ||
| static UCS_F_ALWAYS_INLINE ucs_status_t | ||
| ucp_proto_rndv_mtype_fc_check(ucp_request_t *req, size_t max_frags, |
There was a problem hiding this comment.
| rpriv = req->send.proto_config->priv; | ||
|
|
||
| /* RTR priority: 60% of total fragments */ | ||
| max_frags = rpriv->fc_max_frags / 5 * 3; |
| unsigned num_counters; | ||
| unsigned class_id; | ||
| const char* counter_names[]; | ||
| const char* counter_names[14]; /* Need to maintain the number of |
|
|
||
| class test_ucp_am_nbx_rndv_mtype_fc : public test_ucp_am_nbx_rndv_ppln { | ||
| protected: | ||
| void check_stats_ge(entity &e, uint64_t cntr, uint64_t min_value) |
There was a problem hiding this comment.
constify wherever possible
|
|
||
| void verify_clean_fc_state() | ||
| { | ||
| check_active_frags_zero(sender()); |
There was a problem hiding this comment.
smth like
void verify_clean_fc_state()
{
for (auto* ep : { sender(), receiver() })
{
check_active_frags_zero(ep);
check_pending_queues_empty(ep);
}
}
| check_stats_ge(receiver(), UCP_WORKER_STAT_RNDV_RTR_MTYPE, 1); | ||
|
|
||
| /* No throttling should have occurred (FC is disabled) */ | ||
| auto sender_throttled = UCS_STATS_GET_COUNTER(sender().worker()->stats, |
There was a problem hiding this comment.
This pattern and similar are repeated multiple times across this file.
|
@rakhmets can you pls review as well? |
422d461 to
057df5d
Compare
| "even if invalidation workflow isn't supported", | ||
| ucs_offsetof(ucp_context_config_t, rndv_errh_ppln_enable), UCS_CONFIG_TYPE_BOOL}, | ||
|
|
||
| {"RNDV_MTYPE_WORKER_MAX_MEM", "inf", |
There was a problem hiding this comment.
do we want to enable it by default?
| [UCP_WORKER_STAT_RNDV_RTR] = "rndv_rtr", | ||
| [UCP_WORKER_STAT_RNDV_RTR_MTYPE] = "rndv_rtr_mtype", | ||
| [UCP_WORKER_STAT_RNDV_RKEY_PTR] = "rndv_rkey_ptr", | ||
| [UCP_WORKER_STAT_RNDV_MTYPE_FC_THROTTLED] = "rndv_mtype_fc_throttled", |
There was a problem hiding this comment.
maybe smth like
UCP_WORKER_STAT_RNDV_FC_OK
UCP_WORKER_STAT_RNDV_FC_THROTTLED
| worker->rndv_mtype_fc.enabled = ucp_context_rndv_mtype_fc_enabled(context); | ||
| if (worker->rndv_mtype_fc.enabled) { |
There was a problem hiding this comment.
do we really need the variable worker->rndv_mtype_fc.enabled if already have it on the context config?
and we could initialize the queues unconditionally
There was a problem hiding this comment.
I thought I could save loading context fields (and derefs) and also some comparisons.
There was a problem hiding this comment.
maybe we could initialize the limit of frags to inf if it's unlimited - in the default case i think we want to enable the limit anyway
| ucs_queue_head_t put_pending_q; /* Throttled PUT (RNDV_SEND) requests */ | ||
| ucs_queue_head_t get_pending_q; /* Throttled GET requests */ | ||
| ucs_queue_head_t rtr_pending_q; /* Throttled RTR requests */ |
There was a problem hiding this comment.
maybe 2 queues could be enough? hi-prio for get/put and low-prio for rtr
There was a problem hiding this comment.
Probably will do the job too.
There was a problem hiding this comment.
- 2 queues are enough
- I also see that setting different memory caps for different operations is not needed, and that prioritizing PUT/GET is enough for preventing the deadlock.
| } | ||
|
|
||
| /* Dequeue with priority: PUT > GET > RTR */ | ||
| if (!ucs_queue_is_empty(&worker->rndv_mtype_fc.put_pending_q)) { |
There was a problem hiding this comment.
maybe give each queue an index (if we still need 3 queues):
UCP_WORKER_RNDV_FC_OP_PUT
UCP_WORKER_RNDV_FC_OP_GET
UCP_WORKER_RNDV_FC_OP_RTR
UCP_WORKER_RNDV_FC_OP_LAST
and keep a variable in the worker of the highest prio queue that has elements (LAST if FC is off), this way can find the relevant queue with less checks
| ucs_mpool_put_inline(req->send.rndv.mdesc); | ||
| } | ||
|
|
||
| ucp_proto_rndv_mtype_fc_decrement(req); |
There was a problem hiding this comment.
shall we move it inside the prev if condition?
|
|
||
| max_frags = UCP_PROTO_RNDV_MTYPE_FC_PUT_LIMIT(rpriv->bulk.fc_max_frags); | ||
| pending_q = &req->send.ep->worker->rndv_mtype_fc.put_pending_q; | ||
| if (ucp_proto_rndv_mtype_fc_throttle(req, max_frags, pending_q) == |
There was a problem hiding this comment.
seems ucp_proto_rndv_mtype_fc_throttle is always called right before ucp_proto_rndv_mtype_request_init, maybe move it inside ucp_proto_rndv_mtype_request_init?
1919e03 to
e0cabfd
Compare
e0cabfd to
cba5c7d
Compare
|
Test failure seems related as the hang in cuda memory perftest |
@yosefe Apparently it wasn't related to this PR. |
|
@yosefe can this PR be targeted for the upcoming v1.22 release ? I think there are already RCs but I don't see this being listed (perhaps I missed it) |
| } | ||
|
|
||
|
|
||
| UCP_INSTANTIATE_TEST_CASE_GPU_AWARE(test_ucp_am_nbx_rndv_mtype_fc); |
There was a problem hiding this comment.
can you please add similar tests for get/rndv and put/rndv?
There was a problem hiding this comment.
I am keeping it for the next PR that will include further fixes, as this one should be merged these days
|
🤖 Starting review — findings will be posted here when done. |
|
🤖 CI Triage Agent — TL;DR: The Coverity ( Full analysisSummary: Coverity static analysis gate failed: Root cause: The build and Implicated commit: File: Suggested fix: Value-initialize the struct in the constructor's member-initializer list: rndv_frag_mpools::rndv_frag_mpools(ucp_worker_h worker,
ucs_memory_type_t mem_type) :
m_worker(worker), m_fail_ops() /* zero-initializes all function pointers */
{(
|
|
🤖 Starting review — findings will be posted here when done. |
|
🤖 Starting review — findings will be posted here when done. |
|
🤖 CI Triage Agent — TL;DR: The new gtest Full analysisSummary: Root cause: Test-logic bug, not a library bug. The test is configured with The test then:
The first loop iteration hits This only reproduces on nodes with ≥2 TL system devices; everywhere else the test exits early via Implicated commit: File: Suggested fix: Don't re-request from the pool whose fragment is already held. Skip it in the loop: for (ucs_sys_device_t sys_dev : sys_devs) {
for (unsigned pool : {UCP_WORKER_RNDV_FRAG_POOL_SHARED,
UCP_WORKER_RNDV_FRAG_POOL_RESERVED}) {
/* The only fragment of this pool is held by 'first' */
if ((sys_dev == UCS_SYS_DEVICE_ID_UNKNOWN) &&
(pool == UCP_WORKER_RNDV_FRAG_POOL_SHARED)) {
continue;
}
ASSERT_UCS_OK(ucp_rndv_mpool_get(worker, UCS_MEMORY_TYPE_HOST,
sys_dev, pool, &mdesc));
ucs_mpool_put(mdesc);
}
}The hash still grows: the pre-existing Related: PR #11062 ("UCP/RNDV: Throttle rndv fragment requests"); same series as |
247506c to
58b3523
Compare
|
🤖 Starting review — findings will be posted here when done. |
|
🤖 Starting review — findings will be posted here when done. |
|
🤖 CI Triage Agent — TL;DR: The Coverity Full analysisSummary: Root cause: Implicated commit: File: Suggested fix: Value-initialize the member in the constructor's init list so all function pointers are zeroed: rndv_frag_mpools::rndv_frag_mpools(ucp_worker_h worker,
ucs_memory_type_t mem_type) :
m_worker(worker), m_fail_ops() /* or m_fail_ops{} — C++11 is enabled */
{(Equivalent alternatives: Related: PR #11062 (the failing PR itself); same checker previously hit in #11868 |
|
🤖 Starting review — findings will be posted here when done. |
|
🤖 Starting review — findings will be posted here when done. |
|
In
minor: in the placeholder insertion loop in Residual coverage gap: no test exercises the fall-through case in finding 2 (reserved pool failing with Residual coverage gap: the Residual coverage gap: all request-level throttle/reschedule coverage needs both |
What?
Added a worker-level throttling mechanism for rndv fragment requests (both pipeline and standalone mtype requests).
Why?
Without throttling, rendezvous protocols can spawn an unbounded number of concurrent fragment requests, each allocating staging memory from GPU memory pools. Under heavy workloads, this leads to GPU memory exhaustion (OOM), as tracked in https://nvbugspro.nvidia.com/bug/5359018. The throttling mechanism bounds the number of in-flight fragments per worker, preventing resource exhaustion while allowing work to progress as fragments complete.
How?
The throttling is implemented using a worker-level flow control state (
worker->rndv_ppln_fc) consisting of:active_fragscounter tracking currently in-flight fragment operationspending_qqueue holding throttled requests waiting for resourcesWhen a new fragment request would exceed
UCX_RNDV_PPLN_WORKER_MAX_FRAGS, the request is queued inpending_q. When fragments complete,active_fragsis decremented and one queued request is rescheduled viaucs_callbackq_add_oneshot()to the main progress loop.The mechanism applies to:
rndv_ppln.cmtypeprotocols inrndv_rtr.c,rndv_get.c, andrndv_put.c.