Repository: modin-project/modin
Pull request: FIX-#3551: fix hanging behaviour for to_csv
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>

<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/CONTRIBUTING.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->
If an exception occurs in one of the tasks, we must complete the rest, which are waiting for messages from `Queue` (communication channel). When using `get` function, this is done automatically.

Another reason for the hang could be the implementation of communication between tasks using one queue. Situations could arise when the same task put and took a message from the queue many times in a row, preventing other tasks from taking the message. To avoid this situation, I propose to use its own communication channel for each task (a separate events).

- [ ] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/contributing.html)
- [ ] passes `flake8 modin`
- [ ] passes `black --check modin`
- [ ] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #3551 <!-- issue must be created for each patch -->
- [ ] tests added and passing
- [ ] module layout described at `docs/developer/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->

**Performance**:

Version| Master| PR
-|-|-
Pandas| 40,5 sec| 40,5 sec
Modin - 4 cores| 12,4 sec| 12,7 sec
Modin - 8 cores| 7,0 sec| 7,1 sec
Modin - 16 cores| 4,6 sec| 4,5 sec


<details><summary>Benchmark</summary>

```python
from time import time
import os
import numpy as np
import modin.pandas as pd
#import pandas as pd

df = pd.DataFrame(np.random.randint(10**6, size=(375 * 10**4, 32)))

filename = "test_to_csv.csv"
for i in range(5):
    start = time()
    df.to_csv(filename)
    print(f"TO_CSV TIME: {time()-start}")
    os.remove(filename)
```

</details>

Problem statement:
Modin stuck at to_csv operation 
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: macOS Big sur, also tested on WSL with windows 10
- **Modin version** (`modin.__version__`): 0.10.2
- **Python version**: Python 3.8.11
- **Code we can use to reproduce**:

### Describe the problem
I have a post-processing function to arrange forecast results from gluonts to a dataframe.
The function below works well and computes a dataframe. The function output is then saved as a csv file where raylet is getting stuck. (Not even defaulting to pandas)

### Source code / logs
I first initialize ray with a certain number of CPU count. I'm testing Modin with 1 CPU (8 cores)
```
import os
os.environ["MODIN_ENGINE"] = "ray"  # Modin will use Ray
import ray
ray.init(num_cpus=6)
```

Here the post-processing function
```
def transform_outputs(
    kpi_name: str,
    prediction_list: List[gluonts.model.forecast.SampleForecast],
    confidence_intervals: List[float] = [0.5, 0.9]
):
    
    list_dfs = []
    for forecast in tqdm(prediction_list):
        ns, horz = forecast.samples.shape
        dates = pd.date_range(forecast.start_date, freq=forecast.freq, periods=horz)
        forecast_serie = {}
        forecast_serie[f"{kpi_name}_predict"] = forecast.quantile(0.5)
        df_forecast = pd.DataFrame(forecast_serie)
        
        intervals_dfs = []
        for interval in confidence_intervals:
            lower_p = (1.0 - interval) / 2
            upper_p = (1.0 + interval) / 2

            percentage = str(round(interval*100))
            series={}
            series[f"{kpi_name}_lower_predict_{percentage}%"] = forecast.quantile(q=lower_p)
            series[f"{kpi_name}_upper_predict_{percentage}%"] = forecast.quantile(q=upper_p)
            intervals_dfs.append(pd.DataFrame(series))

        df_intervals = pd.concat(intervals_dfs, axis=1)
        df = pd.concat([df_forecast, df_intervals], axis=1)
        df.index = dates
        df['info'] = forecast.item_id
        list_dfs.append(df)

    return pd.concat(list_dfs, axis=0)

df_forecast_group = transform_outputs(ts_name=ts_name, prediction_list=list_forecasts, confidence_intervals=[0.5, 0.8, 0.9])
logger.info('finnished post processing and start saving')
df_forecast_group.to_csv(f"{output_filepath}/{file_name.replace('.csv', '_forecast.csv')}",
                             index=True,
                             date_format='%Y-%m-%d'
                             )
```

Hints from issue discussion:
The .to_csv() method works when I force the dataframe to default to pandas by using the private method `DataFrame._to_pandas()`
Hello @mbrhd,

Can you provide a dataframe on which the `to_csv` operation stuck so that I can reproduce the problem for myself?

Commit message: FIX-#3551: address review comments

Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>

Patch to review:
diff --git a/modin/core/execution/ray/common/utils.py b/modin/core/execution/ray/common/utils.py
index eded55e99ee..9737c4e1624 100644
--- a/modin/core/execution/ray/common/utils.py
+++ b/modin/core/execution/ray/common/utils.py
@@ -17,6 +17,9 @@
 import sys
 import psutil
 import warnings
+import asyncio
+
+import ray
 
 from modin.config import (
     StorageFormat,
@@ -107,8 +110,6 @@ def initialize_ray(
         What password to use when connecting to Redis.
         If not specified, ``modin.config.RayRedisPassword`` is used.
     """
-    import ray
-
     if not ray.is_initialized() or override_is_cluster:
         cluster = override_is_cluster or IsRayCluster.get()
         redis_address = override_redis_address or RayRedisAddress.get()
@@ -196,3 +197,54 @@ def initialize_ray(
         NPartitions._put(num_gpus)
     else:
         NPartitions._put(num_cpus)
+
+
+@ray.remote
+class SignalActor:  # pragma: no cover
+    """
+    Help synchronize across tasks and actors on cluster.
+
+    For details see: https://docs.ray.io/en/latest/advanced.html?highlight=signalactor#multi-node-synchronization-using-an-actor
+
+    Parameters
+    ----------
+    event_count : int
+        Number of events required for synchronization.
+    """
+
+    def __init__(self, event_count: int):
+        self.events = [asyncio.Event() for _ in range(event_count)]
+
+    def send(self, event_idx: int):
+        """
+        Indicate that event with `event_idx` has occured.
+
+        Parameters
+        ----------
+        event_idx : int
+        """
+        self.events[event_idx].set()
+
+    async def wait(self, event_idx: int):
+        """
+        Wait until event with `event_idx` has occured.
+
+        Parameters
+        ----------
+        event_idx : int
+        """
+        await self.events[event_idx].wait()
+
+    def is_set(self, event_idx: int) -> bool:
+        """
+        Check that event with `event_idx` had occured or not.
+
+        Parameters
+        ----------
+        event_idx : int
+
+        Returns
+        -------
+        bool
+        """
+        return self.events[event_idx].is_set()
diff --git a/modin/core/execution/ray/generic/io/io.py b/modin/core/execution/ray/generic/io/io.py
index 700f9f53a0b..5aa0d1740a5 100644
--- a/modin/core/execution/ray/generic/io/io.py
+++ b/modin/core/execution/ray/generic/io/io.py
@@ -18,7 +18,7 @@
 import pandas
 
 from modin.core.io import BaseIO
-from ray.util.queue import Queue
+from modin.core.execution.ray.common.utils import SignalActor
 from ray import get
 
 
@@ -118,9 +118,7 @@ def to_csv(cls, qc, **kwargs):
         if not cls._to_csv_check_support(kwargs):
             return BaseIO.to_csv(qc, **kwargs)
 
-        # The partition id will be added to the queue, for which the moment
-        # of writing to the file has come
-        queue = Queue(maxsize=1)
+        signals = SignalActor.remote(len(qc._modin_frame._partitions) + 1)
 
         def func(df, **kw):
             """
@@ -134,40 +132,35 @@ def func(df, **kw):
                 Arguments to pass to ``pandas.to_csv(**kw)`` plus an extra argument
                 `partition_idx` serving as chunk index to maintain rows order.
             """
-            if kw["partition_idx"] != 0:
+            partition_idx = kw["partition_idx"]
+            # the copy is made to not implicitly change the input parameters;
+            # to write to an intermediate buffer, we need to change `path_or_buf` in kwargs
+            csv_kwargs = kwargs.copy()
+            if partition_idx != 0:
                 # we need to create a new file only for first recording
                 # all the rest should be recorded in appending mode
-                if "w" in kwargs["mode"]:
-                    kwargs["mode"] = kwargs["mode"].replace("w", "a")
+                if "w" in csv_kwargs["mode"]:
+                    csv_kwargs["mode"] = csv_kwargs["mode"].replace("w", "a")
                 # It is enough to write the header for the first partition
-                kwargs["header"] = False
+                csv_kwargs["header"] = False
 
             # for parallelization purposes, each partition is written to an intermediate buffer
-            path_or_buf = kwargs["path_or_buf"]
-            is_binary = "b" in kwargs["mode"]
-            if is_binary:
-                kwargs["path_or_buf"] = io.BytesIO()
-            else:
-                kwargs["path_or_buf"] = io.StringIO()
-            df.to_csv(**kwargs)
-            content = kwargs["path_or_buf"].getvalue()
-            kwargs["path_or_buf"].close()
-
-            # each process waits for its turn to write to a file;
-            # in case of violation of the order of receiving messages from the queue,
-            # the message is placed back
-            while True:
-                get_value = queue.get(block=True)
-                if get_value == kw["partition_idx"]:
-                    break
-                queue.put(get_value)
+            path_or_buf = csv_kwargs["path_or_buf"]
+            is_binary = "b" in csv_kwargs["mode"]
+            csv_kwargs["path_or_buf"] = io.BytesIO() if is_binary else io.StringIO()
+            df.to_csv(**csv_kwargs)
+            content = csv_kwargs["path_or_buf"].getvalue()
+            csv_kwargs["path_or_buf"].close()
+
+            # each process waits for its turn to write to a file
+            get(signals.wait.remote(partition_idx))
 
             # preparing to write data from the buffer to a file
             with pandas.io.common.get_handle(
                 path_or_buf,
                 # in case when using URL in implicit text mode
                 # pandas try to open `path_or_buf` in binary mode
-                kwargs["mode"] if is_binary else kwargs["mode"] + "t",
+                csv_kwargs["mode"] if is_binary else csv_kwargs["mode"] + "t",
                 encoding=kwargs["encoding"],
                 errors=kwargs["errors"],
                 compression=kwargs["compression"],
@@ -177,13 +170,12 @@ def func(df, **kw):
                 handles.handle.write(content)
 
             # signal that the next process can start writing to the file
-            queue.put(get_value + 1)
-
+            get(signals.send.remote(partition_idx + 1))
             # used for synchronization purposes
             return pandas.DataFrame()
 
         # signaling that the partition with id==0 can be written to the file
-        queue.put(0)
+        get(signals.send.remote(0))
         result = qc._modin_frame._partition_mgr_cls.map_axis_partitions(
             axis=1,
             partitions=qc._modin_frame._partitions,
@@ -193,7 +185,6 @@ def func(df, **kw):
             enumerate_partitions=True,
             max_retries=0,
         )
-
         # pending completion
         get([partition.oid for partition in result.flatten()])
 
@@ -268,4 +259,4 @@ def func(df, **kw):
             lengths=None,
             enumerate_partitions=True,
         )
-        get([partition.oid for partition in result.flatten()])
+        get([part.oid for row in result for part in row])
diff --git a/modin/pandas/test/test_io.py b/modin/pandas/test/test_io.py
index 92821392991..7a81d4d54cd 100644
--- a/modin/pandas/test/test_io.py
+++ b/modin/pandas/test/test_io.py
@@ -59,6 +59,15 @@
     import modin.pandas as pd
 else:
     import modin.experimental.pandas as pd
+
+try:
+    import ray
+
+    EXCEPTIONS = (ray.exceptions.WorkerCrashedError,)
+except ImportError:
+    EXCEPTIONS = ()
+
+
 from modin.config import NPartitions
 
 NPartitions.put(4)
@@ -143,7 +152,16 @@ def eval_to_file(modin_obj, pandas_obj, fn, extension, **fn_kwargs):
     unique_filename_pandas = get_unique_filename(extension=extension)
 
     try:
-        getattr(modin_obj, fn)(unique_filename_modin, **fn_kwargs)
+        # parameter `max_retries=0` is set for `to_csv` function on Ray engine,
+        # in order to increase the stability of tests, we repeat the call of
+        # the entire function manually
+        for _ in range(3):
+            try:
+                getattr(modin_obj, fn)(unique_filename_modin, **fn_kwargs)
+            except EXCEPTIONS:
+                continue
+            break
+
         getattr(pandas_obj, fn)(unique_filename_pandas, **fn_kwargs)
 
         assert assert_files_eq(unique_filename_modin, unique_filename_pandas)


Code review comments:
diff_hunk: @@ -148,51 +132,50 @@ def func(df, **kw):
                 Arguments to pass to ``pandas.to_csv(**kw)`` plus an extra argument
                 `partition_idx` serving as chunk index to maintain rows order.
             """
-            idx = kw["partition_idx"]
-            _kwargs = kwargs.copy()
-            if idx != 0:
+            partition_idx = kw["partition_idx"]
+            # the copy is made to not implicitly change the input parameters;
+            # to write to an intermediate buffer, we need to change `path_or_buf` in kwargs
+            csv_kwargs = kwargs.copy()
+            if partition_idx != 0:
                 # we need to create a new file only for first recording
                 # all the rest should be recorded in appending mode
-                if "w" in _kwargs["mode"]:
-                    _kwargs["mode"] = _kwargs["mode"].replace("w", "a")
+                if "w" in csv_kwargs["mode"]:
+                    csv_kwargs["mode"] = csv_kwargs["mode"].replace("w", "a")
                 # It is enough to write the header for the first partition
-                _kwargs["header"] = False
+                csv_kwargs["header"] = False
 
             # for parallelization purposes, each partition is written to an intermediate buffer
-            path_or_buf = _kwargs["path_or_buf"]
-            is_binary = "b" in _kwargs["mode"]
-            if is_binary:
-                _kwargs["path_or_buf"] = io.BytesIO()
-            else:
-                _kwargs["path_or_buf"] = io.StringIO()
-            df.to_csv(**_kwargs)
-            content = _kwargs["path_or_buf"].getvalue()
-            _kwargs["path_or_buf"].close()
+            path_or_buf = csv_kwargs["path_or_buf"]
+            is_binary = "b" in csv_kwargs["mode"]
+            csv_kwargs["path_or_buf"] = io.BytesIO() if is_binary else io.StringIO()
+            df.to_csv(**csv_kwargs)
+            content = csv_kwargs["path_or_buf"].getvalue()
+            csv_kwargs["path_or_buf"].close()
 
             # each process waits for its turn to write to a file
-            ray.get(signals.wait.remote(idx))
+            get(signals.wait.remote(partition_idx))
 
             # preparing to write data from the buffer to a file
             with pandas.io.common.get_handle(
                 path_or_buf,
                 # in case when using URL in implicit text mode
                 # pandas try to open `path_or_buf` in binary mode
-                _kwargs["mode"] if is_binary else _kwargs["mode"] + "t",
-                encoding=_kwargs["encoding"],
-                errors=_kwargs["errors"],
-                compression=_kwargs["compression"],
-                storage_options=_kwargs["storage_options"],
+                csv_kwargs["mode"] if is_binary else csv_kwargs["mode"] + "t",
+                encoding=kwargs["encoding"],
+                errors=kwargs["errors"],
+                compression=kwargs["compression"],
+                storage_options=kwargs["storage_options"],
                 is_text=False,
             ) as handles:
                 handles.handle.write(content)
 
             # signal that the next process can start writing to the file
-            ray.get(signals.send.remote(idx + 1))
+            get(signals.send.remote(partition_idx + 1))
path: modin/core/execution/ray/generic/io/io.py
text: @user1:
I think `ray.get()` is clearer, `get()` by itself feels too generic

diff_hunk: @@ -152,10 +152,15 @@ def eval_to_file(modin_obj, pandas_obj, fn, extension, **fn_kwargs):
     unique_filename_pandas = get_unique_filename(extension=extension)
 
     try:
-        try:
-            getattr(modin_obj, fn)(unique_filename_modin, **fn_kwargs)
-        except EXCEPTIONS:
-            getattr(modin_obj, fn)(unique_filename_modin, **fn_kwargs)
+        # parameter `max_retries=0` is set for `to_csv` function on Ray engine,
+        # in order to increase the stability of tests, we repeat the call of
+        # the entire function manually
+        for _ in range(3):
+            try:
+                getattr(modin_obj, fn)(unique_filename_modin, **fn_kwargs)
+            except EXCEPTIONS:
+                continue
+            break
path: modin/pandas/test/test_io.py
text: @user1:
please add `else:` clause to the `for` loop and raise an exception in it if all three attempts to create the file failed