ENH: Fix cuda storage transfer deadlock on multiple GPUs

[luweizheng]

Fix storage transfer deadlock of CUDA storage on multiple GPUs. I do this work based on #488

The current implementation of the transfer function leads to a deadlock when executing Xorbits on multiple GPUs. The issue arises from the StorageHandlerActor.fetch_batch function, which invokes SenderManagerActor.send_batch_data and subsequently calls StorageHandlerActor.request_quota_with_spill. Due to the locking mechanism within the StorageHandlerActor method call, a deadlock arises.

NOTEs:

• fix dead lock on GPU
• fix GPU buffer size 0 issue
• test on TPC-H SF10 and it works. Much faster than Dask-cuDF.
• known issue: test_transfer_gpu.py of ucx channel has bugs while socket channel works.
• known issue: too many actor calls in this implement. May lead to performance down.

Check code requirements

• tests added / passed (if needed)
• Ensure all linting tests pass

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 82.46%. Comparing base (5bb0211) to head (b8ddb25).
Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #788      +/-   ##
==========================================
- Coverage   82.49%   82.46%   -0.03%     
==========================================
  Files        1071     1071              
  Lines       80119    80167      +48     
  Branches    12202    12207       +5     
==========================================
+ Hits        66094    66110      +16     
- Misses      12478    12496      +18     
- Partials     1547     1561      +14     

Flag	Coverage Δ
unittests	82.36% <ø> (-0.04%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[hucorz]

The original transfer process is illustrated below.Each StorageHandler corresponds to a Receiver. During a transfer, one handler requests the receiver to create a writer, and the receiver then asks a handler in the same pool to create the writer. However, in the case of GPUs, there is only one handler, leading to a deadlock.

The solution is to move part of the receiver's logic to the handler. When opening a writer, the process checks whether the current handler and receiver are in the same pool. If they are, the handler opens the writer and passes it to the receiver. Otherwise, the receiver requests another handler in the same pool to open the writer. The updated process is shown below.

[hucorz]

LGTM