Avoid multiple blocking calls by gathering UCX frames

[pentschev]

This change provides a slight performance boost, up to ~10%.

Before

$ python dask_cuda/benchmarks/local_cudf_merge.py -d 0,3 --runs 5 -c 20_000_000 -p ucx --runs 51
Merge benchmark
-------------------------------
backend        | dask
merge type     | gpu
rows-per-chunk | 20000000
base-chunks    | 2
other-chunks   | 2
broadcast      | default
protocol       | ucx
device(s)      | 0,3
rmm-pool       | True
frac-match     | 0.3
tcp            | True
ib             | True
nvlink         | True
message api    | TAG
data-processed | 1.19 GiB
===============================
Wall-clock     | Throughput
-------------------------------
238.46 ms      | 5.00 GiB/s
178.76 ms      | 6.67 GiB/s
180.46 ms      | 6.61 GiB/s
...
===============================
avg-throughput | 7.36 GiB/s
mdn-throughput | 7.88 GiB/s
(w1,w2)        | 25% 50% 75% (total nbytes)
-------------------------------
(00,03)        | 15.88 GiB/s 22.90 GiB/s 24.32 GiB/s (44.59 GiB)
(03,00)        | 18.03 GiB/s 23.06 GiB/s 25.06 GiB/s (43.70 GiB)

After

$ python dask_cuda/benchmarks/local_cudf_merge.py -d 0,3 --runs 5 -c 20_000_000 -p ucx --runs 51
Merge benchmark
-------------------------------
backend        | dask
merge type     | gpu
rows-per-chunk | 20000000
base-chunks    | 2
other-chunks   | 2
broadcast      | default
protocol       | ucx
device(s)      | 0,3
rmm-pool       | True
frac-match     | 0.3
tcp            | True
ib             | True
nvlink         | True
message api    | TAG
data-processed | 1.19 GiB
===============================
Wall-clock     | Throughput
-------------------------------
247.85 ms      | 4.81 GiB/s
193.90 ms      | 6.15 GiB/s
165.19 ms      | 7.22 GiB/s
...
===============================
avg-throughput | 7.98 GiB/s
mdn-throughput | 8.42 GiB/s
(w1,w2)        | 25% 50% 75% (total nbytes)
-------------------------------
(00,03)        | 19.44 GiB/s 24.10 GiB/s 26.87 GiB/s (39.00 GiB)
(03,00)        | 19.69 GiB/s 23.14 GiB/s 26.34 GiB/s (38.56 GiB)

[quasiben]

This is a really great improvement! Thanks @pentschev

[pentschev]

Thanks @quasiben !