TEST: CI: flaky ray test: ubuntu: ray.exceptions.RaySystemError: System error: Broken pipe

[mvashishtha]

Got Can't remove dependencies of tasks that are not queued. in pytest -n 2 modin/pandas/test/dataframe/test_reduce.py for this run.

Interesting lines:
[2022-10-12 21:47:21,914 C 17538 17538] (raylet) dependency_manager.cc:208: Check failed: task_entry != queued_task_requests_.end() Can't remove dependencies of tasks that are not queued.

Much later:

E ray.exceptions.RaySystemError: System error: Broken pipe

Stack trace that repeats

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
modin/pandas/test/utils.py:871: in create_test_dfs
    post_fn, [pd.DataFrame(*args, **kwargs), pandas.DataFrame(*args, **kwargs)]
modin/logging/logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin/_compat/pandas_api/latest/dataframe.py:52: in __init__
    self._init(
modin/logging/logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin/pandas/dataframe.py:189: in _init
    self._query_compiler = from_pandas(pandas_df)._query_compiler
modin/pandas/utils.py:83: in from_pandas
    return DataFrame(query_compiler=FactoryDispatcher.from_pandas(df))
modin/core/execution/dispatching/factories/dispatcher.py:160: in from_pandas
    return cls.__factory._from_pandas(df)
modin/core/execution/dispatching/factories/factories.py:167: in _from_pandas
    return cls.io_cls.from_pandas(df)
modin/core/io/io.py:75: in from_pandas
    return cls.query_compiler_cls.from_pandas(df, cls.frame_cls)
modin/logging/logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin/core/storage_formats/pandas/query_compiler.py:286: in from_pandas
    return cls(data_cls.from_pandas(df))
modin/logging/logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin/core/dataframe/pandas/dataframe/dataframe.py:2967: in from_pandas
    new_frame, new_lengths, new_widths = cls._partition_mgr_cls.from_pandas(
modin/logging/logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin/core/dataframe/pandas/partitioning/partition_manager.py:58: in wait
    result = func(cls, *args, **kwargs)
modin/core/dataframe/pandas/partitioning/partition_manager.py:741: in from_pandas
    parts = [
modin/core/dataframe/pandas/partitioning/partition_manager.py:742: in <listcomp>
    [
modin/core/dataframe/pandas/partitioning/partition_manager.py:745: in <listcomp>
    put_func(df.iloc[i : i + row_chunksize, j : j + col_chunksize]),
modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py:279: in put
    ray.put(obj), len(obj.index), len(obj.columns)
/usr/share/miniconda3/envs/modin/lib/python3.8/site-packages/ray/_private/client_mode_hook.py:105: in wrapper
    return func(*args, **kwargs)
/usr/share/miniconda3/envs/modin/lib/python3.8/site-packages/ray/_private/worker.py:2340: in put
    object_ref = worker.put_object(value, owner_address=serialize_owner_address)
/usr/share/miniconda3/envs/modin/lib/python3.8/site-packages/ray/_private/worker.py:620: in put_object
    self.core_worker.put_serialized_object_and_increment_local_ref(
python/ray/_raylet.pyx:1364: in ray._raylet.CoreWorker.put_serialized_object_and_increment_local_ref
    ???
python/ray/_raylet.pyx:1254: in ray._raylet.CoreWorker._create_put_buffer
    ???