Dask-cuda: AttributeError: 'LocalCUDACluster' object has no attribute 'status'

[jakirkham]

Seeing this CI failure recently

09:57:53 FAILED dask_cuda/tests/test_explicit_comms.py::test_dataframe_merge_empty_partitions
09:57:53 = 1 failed, 1065 passed, 4 skipped, 3 xfailed, 3 xpassed, 1120 warnings in 627.56s (0:10:27) =
09:57:53 Exception ignored in: <function Cluster.__del__ at 0x7fa649caf200>
09:57:53 Traceback (most recent call last):
09:57:53   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/deploy/cluster.py", line 112, in __del__
09:57:53     if self.status != Status.closed:
09:57:53 AttributeError: 'LocalCUDACluster' object has no attribute 'status'
09:57:53 Exception ignored in: <function Cluster.__del__ at 0x7fa649caf200>
09:57:53 Traceback (most recent call last):
09:57:53   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/deploy/cluster.py", line 112, in __del__
09:57:53     if self.status != Status.closed:
09:57:53 AttributeError: 'LocalCUDACluster' object has no attribute 'status'

https://gpuci.gpuopenanalytics.com/job/rapidsai/job/gpuci-v0.20/job/dask-cuda/job/prb/job/dask-cuda-gpu-test/CUDA=11.0,GPU_LABEL=gpu-a100,OS=ubuntu16.04,PYTHON=3.7/29/console

[pentschev]

The test that is reported failing may have a better hint at what's happening:

18:50:58 dask_cuda/tests/test_explicit_comms.py::test_dataframe_merge_empty_partitions Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:50:58 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:07 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:07 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:07 distributed.worker - WARNING - Compute Failed
18:51:07 Function:  _run_coroutine_on_worker
18:51:07 args:      (268138213725285481833464634942100322041, <function local_df_merge at 0x7fe9c65963b0>, ({0, 1}, [{0: 1}, {1: 1}], [[], [   key  payload2
18:51:07 0    1         0
18:51:07 1    0         1]], ['key'], ['key']))
18:51:07 kwargs:    {}
18:51:07 Exception: AttributeError("'NoneType' object has no attribute 'merge'")
18:51:07 

18:51:07 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:07 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:07 distributed.worker - WARNING - Compute Failed
18:51:07 Function:  _run_coroutine_on_worker
18:51:07 args:      (268138213725285481833464634942100322041, <function local_df_merge at 0x7f881da88dd0>, ({0, 1}, [{0: 1}, {1: 1}], [[   key  payload1
18:51:07 0    0         0
18:51:07 1    1         1], []], ['key'], ['key']))
18:51:07 kwargs:    {}
18:51:07 Exception: TypeError("Can only merge Series or DataFrame objects, a <class 'NoneType'> was passed")
18:51:07 

18:51:07 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:07 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:07 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:07 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:08 Process SpawnProcess-15:
18:51:08 Traceback (most recent call last):
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
18:51:08     self.run()
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/multiprocessing/process.py", line 99, in run
18:51:08     self._target(*self._args, **self._kwargs)
18:51:08   File "/workspace/dask_cuda/tests/test_explicit_comms.py", line 128, in _test_dataframe_merge_empty_partitions
18:51:08     ddf3 = explicit_comms_merge(ddf1, ddf2, on=["key"]).set_index("key")
18:51:08   File "/workspace/dask_cuda/explicit_comms/dataframe/merge.py", line 333, in merge
18:51:08     extra_args=(left_on, right_on),
18:51:08   File "/workspace/dask_cuda/explicit_comms/dataframe/merge.py", line 276, in submit_dataframe_operation
18:51:08     return dd.from_delayed(ret)
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/dask/dataframe/io/io.py", line 592, in from_delayed
18:51:08     meta = delayed(make_meta)(dfs[0]).compute()
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/dask/base.py", line 285, in compute
18:51:08     (result,) = compute(self, traverse=False, **kwargs)
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/dask/base.py", line 567, in compute
18:51:08     results = schedule(dsk, keys, **kwargs)
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/client.py", line 2666, in get
18:51:08     results = self.gather(packed, asynchronous=asynchronous, direct=direct)
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/client.py", line 1981, in gather
18:51:08     asynchronous=asynchronous,
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/client.py", line 844, in sync
18:51:08     self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/utils.py", line 353, in sync
18:51:08     raise exc.with_traceback(tb)
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/utils.py", line 336, in f
18:51:08     result[0] = yield future
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/tornado/gen.py", line 762, in run
18:51:08     value = future.result()
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/client.py", line 1840, in _gather
18:51:08     raise exception.with_traceback(traceback)
18:51:08   File "/workspace/dask_cuda/explicit_comms/comms.py", line 103, in _run_coroutine_on_worker
18:51:08     return executor.submit(_run).result()
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/concurrent/futures/_base.py", line 435, in result
18:51:08     return self.__get_result()
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
18:51:08     raise self._exception
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/concurrent/futures/thread.py", line 57, in run
18:51:08     result = self.fn(*self.args, **self.kwargs)
18:51:08   File "/workspace/dask_cuda/explicit_comms/comms.py", line 100, in _run
18:51:08     return future.result()
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/concurrent/futures/_base.py", line 435, in result
18:51:08     return self.__get_result()
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
18:51:08     raise self._exception
18:51:08   File "/workspace/dask_cuda/explicit_comms/dataframe/merge.py", line 180, in local_df_merge
18:51:08     return df1.merge(df2, left_on=left_on, right_on=right_on)
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/pandas/core/frame.py", line 8208, in merge
18:51:08     validate=validate,
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/pandas/core/reshape/merge.py", line 87, in merge
18:51:08     validate=validate,
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/pandas/core/reshape/merge.py", line 599, in __init__
18:51:08     _right = _validate_operand(right)
18:51:08   File "/opt/conda/envs/rapids/lib/python3.7/site-packages/pandas/core/reshape/merge.py", line 2149, in _validate_operand
18:51:08     f"Can only merge Series or DataFrame objects, a {type(obj)} was passed"
18:51:08 TypeError: Can only merge Series or DataFrame objects, a <class 'NoneType'> was passed
18:51:09 Coverage.py warning: --include is ignored because --source is set (include-ignored)
18:51:09 FAILED

May this be related to some recent change in Dask or Distributed, @madsbk @rjzamora ?

[pentschev]

I confirmed that https://github.com/dask/distributed/pull/4638 is what causes this. As John pointed out, this is likely due to a dependency on the number of workers in https://github.com/rapidsai/dask-cuda/blob/branch-0.20/dask_cuda/tests/test_explicit_comms.py#L116, given that the Distributed PR doesn't touch an Dataframe code. Naively playing a bit with parameters in that tests didn't resolve the issues. To unblock CI I'm xfailing that test in https://github.com/rapidsai/dask-cuda/pull/577 .

@madsbk could you take a look at that test and see if you can fix it when you have a chance?

[madsbk]

@madsbk could you take a look at that test and see if you can fix it when you have a chance?

The plan is to re-implement merge using the same approach as in shuffle, which support arbitrary partitioning of the data frames thus xfailing the test is fine for now.