[vyasr]

This PR fixes a bug where if a consumer checked if intra-node exchange was supported before any producer arrived to fill the queue, the exchange queue would immediately and incontrovertibly disable intra-node exchange. This PR fixes that by adding a callback mechanism that implements a handshake for the producer and consumer once both are in place. Since an arbitrary number of callbacks could be registered with the communicator, the communicator executes them in bounded chunks to avoid starving the rest of the system of resources.

To support testing these changes, this PR also implements per-source runtime metrics for transport decisions and payloads. These metrics are used to ensure in various testing scenarios that the expected intra-node transport is in fact occurring when it should be.

[netlify]

✅ Deploy Preview for meta-velox canceled.

Name	Link
🔨 Latest commit	2acc113
🔍 Latest deploy log	https://app.netlify.com/projects/meta-velox/deploys/6ab1d2983d12b7000848cef8

[vyasr]

@karthikeyann @dan13bauer @devavret I'd appreciate reviews from as many of you who have time here.

[github-actions]

Selective Build Plan

Linux release with adapters is running a full build (changes touch velox/experimental/ or velox/external/). See the CI workflows README for what this means.

---

Selective build plan

[devavret]

Worth confirming if this is a real issue. Feel free to mark resolved if you think this is a false positive.

[devavret]

I let my agent look for ordering issues and it found this:

UcxOutputQueueManager.cpp:240 terminates the old queue only after releasing taskLifecycleMutex_. A replacement initializeTask() can therefore clear cancellation and publish its new queue first. The old queue’s pending getData callback then wakes its old exchange server; that server publishes an end marker and calls deleteResults() using only (taskId, destination) (UcxExchangeServer.cpp:308). It can consequently terminate or delete data from the new task’s queue. Keep queue termination within the retirement boundary, or attach a generation/incarnation ID to queue callbacks and registry keys. The new lifecycle test covers cancellation ordering, but not this pending-callback handoff.

[vyasr]

I agree that this is a real race. Fixed in 17df2fb

[dan13bauer]

Thanks @vyasr for this rather substantive PR. Clearly, the existing code does not enable the intra-node path when the consumer starts before the producer's queue exists.
The complexity is quite significant, involving deferred callbacks, another communicator queue, extending batching/fairness logic in the communicator, lifecycle synchronization, new lock/lifetime interactions.
I might be on the overly cautious side but do the benefits outweigh this complexity? What is the performance benefit, e.g. for TPC-H or TPC-DS?

• How frequently does the consumer-before-producer orderign occur?
• What is the end-to-end performance improvement?
• How does this change with the number of workers on the same system?

[vyasr]

Thanks @vyasr for this rather substantive PR. Clearly, the existing code does not enable the intra-node path when the consumer starts before the producer's queue exists. The complexity is quite significant, involving deferred callbacks, another communicator queue, extending batching/fairness logic in the communicator, lifecycle synchronization, new lock/lifetime interactions. I might be on the overly cautious side but do the benefits outweigh this complexity? What is the performance benefit, e.g. for TPC-H or TPC-DS?

* How frequently does the consumer-before-producer orderign occur?

* What is the end-to-end performance improvement?

* How does this change with the number of workers on the same system?

Thanks for the comments @dan13bauer! These are fair questions. Since I've only recently been ramping up on the Velox-cudf work, I have not run benchmarks nearly as frequently or on as diverse of hardware as some of the other devs, so I'll ask them to chime in. @devavret @karthikeyann @GregoryKimball @kjmph could you comment more here? I'm happy to collect some more data but my understanding is that all of you have already been encountering these problems fairly frequently and could comment today.

[devavret]

@vyasr @dan13bauer This issue was found a long time ago and I don't have the exact performance numbers at hand but on the hardware I was running (DGX 8xB200 nvlinked), it seemed to happen quite often and the performance drop was about 30%. This is from hazy memory though but I also recall it happening on an NVL72.

[dan13bauer]

I've run a simple test on our 8xA100 platform: TPC-H, sf-1000, with and without this PR, otherwise everything the same. This is not conclusive, I'm putting it here to have a data point. Also, the configuration is suboptimal, it is just meant as a relative comparison.

Arithmetic mean of runs 2–5, excluding the first lukewarm run. Times are in seconds. Positive percentages mean with_intranode_patched is faster.

Query	with_intranode_patched avg. (s)	ucx-plain avg. (s)	Patched vs plain
Q1	2.3808	2.4398	+2.42%
Q2	2.2243	2.2120	−0.55%
Q3	6.7475	6.2910	−7.26%
Q4	1.8578	1.9963	+6.94%
Q5	4.3208	4.1708	−3.60%
Q6	1.5745	1.5550	−1.25%
Q7	3.3973	3.1273	−8.63%
Q8	4.1763	3.8998	−7.09%
Q9	33.0340	38.0750	+13.24%
Q10	6.3075	7.0693	+10.78%
Q11	1.2145	1.2145	Tie
Q12	3.8935	3.6668	−6.18%
Q13	3.8920	3.9565	+1.63%
Q14	2.3268	2.3628	+1.52%
Q15	3.5005	3.5070	+0.19%
Q16	2.2350	2.6820	+16.67%
Q17	22.0890	21.4890	−2.79%
Q18	50.4415	50.9903	+1.08%
Q19	3.6445	3.6220	−0.62%
Q20	19.4235	21.1453	+8.14%
Q21	22.0195	18.7070	−17.71%
Q22	2.0913	1.7843	−17.21%
Total	202.7920	205.9633	+1.54%

The patched version's total is 3.1713 seconds lower. It wins 10 queries, ucx-plain wins 11, and Q11 is tied.

[vyasr]

openucx/ucx#11319 should give us the majority of the benefits of this PR. Prior to finding that PR I independently found that the main reason for the slow transfers we're observing is that UCXX error handling is turned on by default in Velox, and as a result we are unable to go to the local copy path. Enabling that properly in UCX is a better fix. The other smaller factor I encountered is that for small messages, the rendezvous scheme that UCX chooses may still result in PCIe transfers unless we explicitly disable that by forcing the rendezvous scheme with UCX_RNDV_SCHEME=get_zcopy. While I was able too write microbenchmarks where that difference was significant, in my larger scale TPC-H benchmarks enabling the small copies to avoid PCIe was within the noise threshold of the benchmarks, so the benefit of changing that is likely to be small for the benchmarks we are working with right now.

[vyasr]

openucx/ucx#11319 is now merged, so I'll need to redo benchmarks here and assess the value of this improvement.