Repository: modin-project/modin
Pull request: FIX-#4449 Drain the call queue before waiting on result in benchmark mode.
<!--
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] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
- [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 #4449 
- [ ] tests added and 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 -->
- [x] added (Issue Number: PR title (PR Number)) and github username to release notes for next major release <!-- e.g. DOCS-#4077: Add release notes template to docs folder (#4078) -->


Problem statement:
Benchmark Mode blocks on computation so all computation occurs on one node
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**:
- **Modin version** (`modin.__version__`):
- **Python version**:
- **Code we can use to reproduce**:

<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->

### Describe the problem
<!-- Describe the problem clearly here. -->
Since Benchmark Mode calls `wait` on each partition sequentially, computation that can be done in parallel ends up happening sequentially on a single node. We need to instead collect all the OID's of all partitions, and wait on all of them together - that way each of the tasks is kicked off in parallel, and computation happens in parallel.
### Source code / logs
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->

Hints from issue discussion:
I just want to clarify that the bug applies only when we have a function like a transpose or a reindex that adds a remote function to each partition's call queue, but doesn't trigger the function's execution. In that case, benchmark mode will [trigger](https://github.com/modin-project/modin/blob/c17dde71ae8114723a13279e36bf5532dcba328a/modin/core/dataframe/pandas/partitioning/partition_manager.py#L66) all of the executions serially instead of in parallel.

Commit message: Add a release note.

Signed-off-by: mvashishtha <mahesh@ponder.io>

Patch to review:
diff --git a/docs/release_notes/release_notes-0.15.0.rst b/docs/release_notes/release_notes-0.15.0.rst
index 52d41b5b8f1..827e14267bd 100644
--- a/docs/release_notes/release_notes-0.15.0.rst
+++ b/docs/release_notes/release_notes-0.15.0.rst
@@ -18,6 +18,7 @@ Key Features and Updates
   * FIX-#4394: Fix issue with multiindex metadata desync (#4395)
   * FIX-#4425: Add parameters to groupby pct_change (#4429)
   * FIX-#4461: Fix S3 CSV data path (#4462)
+  * FIX-#4449: Drain the call queue before waiting on result in benchmark mode (#4472)
 * Performance enhancements
   * FEAT-#4320: Add connectorx as an alternative engine for read_sql (#4346)
 * Benchmarking enhancements
diff --git a/modin/core/dataframe/pandas/partitioning/partition_manager.py b/modin/core/dataframe/pandas/partitioning/partition_manager.py
index 834f86478cf..31daa8b9d37 100644
--- a/modin/core/dataframe/pandas/partitioning/partition_manager.py
+++ b/modin/core/dataframe/pandas/partitioning/partition_manager.py
@@ -60,6 +60,13 @@ def wait(*args, **kwargs):
                 partitions = result[0]
             else:
                 partitions = result
+            # When partitions have a deferred call queue, calling
+            # parittion.wait() on each partition serially will serially kick
+            # off each deferred computation and wait for each partition to
+            # finish before kicking off the next one. Instead, we want to
+            # serially kick off all the deferred computations so that they can
+            # all run aynschronously, then serially wait on all the results.
+            [part.drain_call_queue() for row in partitions for part in row]
             # need to go through all the values of the map iterator
             # since `wait` does not return anything, we need to explicitly add
             # the return `True` value from the lambda


Code review comments:
diff_hunk: @@ -60,6 +60,13 @@ def wait(*args, **kwargs):
                 partitions = result[0]
             else:
                 partitions = result
+            # When partitions have a deferred call queue, calling
+            # parittion.wait() on each partition serially will serially kick
path: modin/core/dataframe/pandas/partitioning/partition_manager.py
text: @user1:
```suggestion
            # partition.wait() on each partition serially will serially kick
```