Repository: modin-project/modin
Pull request: FEAT-#7249: Add `reload_modin` feature
<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_NOTE:_**  If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #7249  <!-- issue must be created for each patch -->
- [x] tests passing
- [x] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Problem statement:
how to take down ray and put up again in local mode
My program has memory risk, and part of it seems to come from memory leak (idling ray workers holding a big chunk of memory). I have a for loop to independently run chunks of csv file on a series of tasks, I wish to kill ray after each iteration to release memory, and let Modin to put it up again with fresh ray workers. However, my code is the following:
```py
import pandas

for df_ in pandas.read_csv('xxx.csv', chunk=5000):
    df_.to_csv(xxx)
    run_my_tasks(xxx) # Modin will initialize ray in first iteration
    ray.shutdown()
```

however, I got below error:
```
File "/home/.../lib/python3.9/site-packages/modin/core/execution/ray/common/deferred_execution.py", line 309, in _deconstruct_chain
    output[out_pos] = out_pos
IndexError: list assignment index out of range
```

Hints from issue discussion:
Hi @SiRumCz, thanks for posting this issue. I guess there might be an issue with multiple Ray initialization in Modin codebase. We would have to look into this deeper. Meanwhile, can you explicitly put `ray.init()` before `run_my_tasks(xxx)` to see if it works?
@YarShev Thanks for your response. Yes, I have tried that method, and unfortunately I got:
`ValueError: An application is trying to access a Ray object whose owner is unknown(00ffffffffffffffffffffffffffffffffffffff0100000002e1f505). Please make sure that all Ray objects you are trying to access are part of the current Ray session. Note that object IDs generated randomly (ObjectID.from_random()) or out-of-band (ObjectID.from_binary(...)) cannot be passed as a task argument because Ray does not know which task created them. If this was not how your object ID was generated, please file an issue at https://github.com/ray-project/ray/issues/`
@SiRumCz, could you try to execute `ray.init()` and `importlib.reload(pd)` before `run_my_tasks(xxx)`, where `pd` is `import modin.pandas as pd`?

@SiRumCz, I opened https://github.com/modin-project/modin/pull/7280, which adds `reload_modin` function. Tested on the following example and it passed to me.

```python
import modin.pandas as pd
from modin.utils import reload_modin
import ray

ray.init(num_cpus=16)  # can be commented out, works

df = pd.read_csv("example.csv")
df = df.abs()
print(df)

ray.shutdown()
reload_modin()
ray.init(num_cpus=16)  # can be commented out, works

df = pd.read_csv("example.csv")
df = df.abs()
print(df)
```

Commit message: Fix mypy

Signed-off-by: Igoshev, Iaroslav <iaroslav.igoshev@intel.com>

Patch to review:
diff --git a/modin/utils.py b/modin/utils.py
index 8305bc75da3..f977ce2cdf0 100644
--- a/modin/utils.py
+++ b/modin/utils.py
@@ -878,3 +878,17 @@ def __init__(self, func: Any):
 
     def __get__(self, instance: Any, owner: Any) -> Any:  # noqa: GL08
         return self.fget(owner)
+
+
+def reload_modin() -> None:
+    """
+    Reload all previously imported Modin modules.
+
+    The call to this function is required
+    if an execution engine has been shut down and
+    is going to be started up once again.
+    """
+    modules = sys.modules.copy()
+    for key, module in modules.items():
+        if key.startswith("modin"):
+            importlib.reload(module)


Code review comments:
diff_hunk: @@ -878,3 +878,17 @@ def __init__(self, func: Any):
 
     def __get__(self, instance: Any, owner: Any) -> Any:  # noqa: GL08
         return self.fget(owner)
+
+
+def reload_modin() -> None:
+    """
+    Reload all previously imported Modin modules.
+
+    The call to this function is required
+    if an execution engine has been shut down and
+    is going to be started up once again.
+    """
+    modules = sys.modules.copy()
+    for key, module in modules.items():
+        if key.startswith("modin"):
path: modin/utils.py
text: @author:
```suggestion
    for name, module in modules.items():
        if name.startswith("modin"):
```