Properly cleanup processes and queues for MPRS and Fix `pause` for prefetch by ejguan · Pull Request #1075 · meta-pytorch/data

ejguan · 2023-03-10T22:45:32Z

Fixes issue about MPRS.finalize when dataloader2.shutdown() is called

Changes

DataLoader2 should always clean up datapipe_iter at shutdown
Guard MPRS to finalize once
Fix the problem of ConnectionError when DataLoader early exits
- This is caused by queue is joined when main/worker/dispatching process exits. No more request/response can be passed across processes.
  - Consumer process shouldn't join the req_queue at exit to make sure producer process can still access the remaining request. And, consumer will close req_queue after clean up to prevent any further request sent to queue.
  - Produce process shouldn't join the res_queue at exit to make sure consumer process can still access response. And, producer will close res_queue after clean up to prevent any further response sent to queue.
    - Main (Consumer) <-> Worker (Producer)
    - Worker (Consumer) -> Dispatching (Producer)
Fix pause API for DataLoader2
- Invoke pause lazily until the limit+1 iteration is reached to align with python's iterator behavior.
- Make prefetch.pause blocking unless there might be potential racing issue. Main thread is paused but prefetch worker is still trying to fetch data from iter.
Add tests to validate

facebook-github-bot · 2023-03-10T22:46:05Z

@ejguan has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

ejguan · 2023-03-10T22:46:19Z

TBH, I hope we can cherrypick this PR to release branch, but probably not feasible as the final RC has been cut.

facebook-github-bot · 2023-03-13T14:53:33Z

@ejguan has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

ejguan · 2023-03-13T16:42:26Z

Converting back to draft as this PR seems breaking some tests

facebook-github-bot · 2023-03-15T21:56:16Z

@ejguan has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

torchdata/dataloader2/dataloader2.py

torchdata/datapipes/iter/util/prefetcher.py

facebook-github-bot · 2023-03-16T14:08:08Z

@ejguan has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

NivekT

LGTM!

NivekT · 2023-03-16T15:14:44Z

torchdata/dataloader2/reading_service.py

                res_queue.cancel_join_thread()
+            process.daemon = True
+            process.start()
+            self._dispatch_process = (process, req_queues, res_queues)


non-blocking note: Maybe we should refactor out the dispatching code at some point... It is getting too much

Yeah. We should spin off it and worker launch to a separate file

NivekT · 2023-03-16T15:18:37Z

torchdata/dataloader2/reading_service.py

                process_name=f"worker process {worker_id}",
                call_on_process_init=call_on_process_init,
            )
+            req_queue.cancel_join_thread()


IIUC, this allows join_thread to be non-blocking?

Emmm, not exactly. This would make sure req_queue not being joined by gc when the main process wants to exit. Then, it would guarantee that worker process can still access the TerminationRequest from req_queue rather than any corrupted data from a closed queue

torchdata/dataloader2/reading_service.py

NivekT · 2023-03-16T15:29:45Z

torchdata/dataloader2/communication/iter.py

                        response = self.res_buffers[idx].popleft()
                    else:
-                        response = self.datapipes[idx].protocol.get_response_next(block=True)
+                        while not self._terminated:


Question: so we are guaranteeing termination message will be sent?

This is guarantee if there is a prefetch attached after this DataPipe the worker thread would be able to know it's terminated. If we keep the blocking request here, the worker thread will be stuck at the blocking call even though we have already terminated this DataPipe -> hanging at the end of thread.join() in prefetch.

NivekT · 2023-03-16T15:30:42Z

test/dataloader2/test_mprs.py

@@ -29,12 +32,60 @@ def _add_one(x: int) -> int:
 dp_parametrize = parametrize("dp", test_dps)




Did none of the existing pause tests break? Amazing

Lots of tests are broken before I make pause blocking for Prefetcher. So, I have to make the change.

facebook-github-bot · 2023-03-16T16:09:33Z

@ejguan has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

facebook-github-bot · 2023-03-16T16:57:39Z

@ejguan has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

facebook-github-bot · 2023-03-16T20:02:54Z

@ejguan has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

facebook-github-bot · 2023-03-17T00:31:54Z

This pull request was exported from Phabricator. Differential Revision: D43994511

…efetch (meta-pytorch#1075) Summary: Fixes issue about `MPRS.finalize` when `dataloader2.shutdown()` is called ### Changes - DataLoader2 should always clean up `datapipe_iter` at shutdown - Guard `MPRS` to finalize once - Fix the problem of `ConnectionError` when DataLoader early exits - This is caused by `queue` is joined when main/worker/dispatching process exits. No more request/response can be passed across processes. - Consumer process shouldn't join the `req_queue` at exit to make sure producer process can still access the remaining request. And, consumer will close `req_queue` after clean up to prevent any further request sent to queue. - Produce process shouldn't join the `res_queue` at exit to make sure consumer process can still access response. And, producer will close `res_queue` after clean up to prevent any further response sent to queue. - Main (Consumer) <-> Worker (Producer) - Worker (Consumer) -> Dispatching (Producer) - Fix `pause` API for DataLoader2 - Invoke `pause` lazily until the `limit+1` iteration is reached to align with python's iterator behavior. - Make `prefetch.pause` blocking unless there might be potential racing issue. Main thread is paused but prefetch worker is still trying to fetch data from `iter`. - Add tests to validate Pull Request resolved: meta-pytorch#1075 Reviewed By: NivekT Differential Revision: D43994511 Pulled By: ejguan fbshipit-source-id: 51dbecd5f68f6e34987235ba8f00461d3b0f4431

…efetch (meta-pytorch#1075) Summary: Fixes issue about `MPRS.finalize` when `dataloader2.shutdown()` is called ### Changes - DataLoader2 should always clean up `datapipe_iter` at shutdown - Guard `MPRS` to finalize once - Fix the problem of `ConnectionError` when DataLoader early exits - This is caused by `queue` is joined when main/worker/dispatching process exits. No more request/response can be passed across processes. - Consumer process shouldn't join the `req_queue` at exit to make sure producer process can still access the remaining request. And, consumer will close `req_queue` after clean up to prevent any further request sent to queue. - Produce process shouldn't join the `res_queue` at exit to make sure consumer process can still access response. And, producer will close `res_queue` after clean up to prevent any further response sent to queue. - Main (Consumer) <-> Worker (Producer) - Worker (Consumer) -> Dispatching (Producer) - Fix `pause` API for DataLoader2 - Invoke `pause` lazily until the `limit+1` iteration is reached to align with python's iterator behavior. - Make `prefetch.pause` blocking unless there might be potential racing issue. Main thread is paused but prefetch worker is still trying to fetch data from `iter`. - Add tests to validate Pull Request resolved: meta-pytorch#1075 Reviewed By: NivekT Differential Revision: D43994511 Pulled By: ejguan fbshipit-source-id: db077e09f62b96de2fff71635c46dc42c839c9ec

facebook-github-bot · 2023-03-17T01:09:02Z

This pull request was exported from Phabricator. Differential Revision: D43994511

facebook-github-bot · 2023-03-17T06:59:47Z

@ejguan merged this pull request in 8eecb77.

facebook-github-bot · 2023-03-17T14:11:54Z

This pull request has been reverted by 7d97475.

…efetch (meta-pytorch#1075) Summary: Fixes issue about `MPRS.finalize` when `dataloader2.shutdown()` is called ### Changes - DataLoader2 should always clean up `datapipe_iter` at shutdown - Guard `MPRS` to finalize once - Fix the problem of `ConnectionError` when DataLoader early exits - This is caused by `queue` is joined when main/worker/dispatching process exits. No more request/response can be passed across processes. - Consumer process shouldn't join the `req_queue` at exit to make sure producer process can still access the remaining request. And, consumer will close `req_queue` after clean up to prevent any further request sent to queue. - Produce process shouldn't join the `res_queue` at exit to make sure consumer process can still access response. And, producer will close `res_queue` after clean up to prevent any further response sent to queue. - Main (Consumer) <-> Worker (Producer) - Worker (Consumer) -> Dispatching (Producer) - Fix `pause` API for DataLoader2 - Invoke `pause` lazily until the `limit+1` iteration is reached to align with python's iterator behavior. - Make `prefetch.pause` blocking unless there might be potential racing issue. Main thread is paused but prefetch worker is still trying to fetch data from `iter`. - Add tests to validate Pull Request resolved: meta-pytorch#1075 Differential Revision: D44168655 Pulled By: ejguan fbshipit-source-id: 4fa8c34438414cad7f46527355eb0d51866e6b15

…efetch (#1096) Summary: Pull Request resolved: #1096 Fixes issue about `MPRS.finalize` when `dataloader2.shutdown()` is called ### Changes - DataLoader2 should always clean up `datapipe_iter` at shutdown - Guard `MPRS` to finalize once - Fix the problem of `ConnectionError` when DataLoader early exits - This is caused by `queue` is joined when main/worker/dispatching process exits. No more request/response can be passed across processes. - Consumer process shouldn't join the `req_queue` at exit to make sure producer process can still access the remaining request. And, consumer will close `req_queue` after clean up to prevent any further request sent to queue. - Produce process shouldn't join the `res_queue` at exit to make sure consumer process can still access response. And, producer will close `res_queue` after clean up to prevent any further response sent to queue. - Main (Consumer) <-> Worker (Producer) - Worker (Consumer) -> Dispatching (Producer) - Fix `pause` API for DataLoader2 - Invoke `pause` lazily until the `limit+1` iteration is reached to align with python's iterator behavior. - Make `prefetch.pause` blocking unless there might be potential racing issue. Main thread is paused but prefetch worker is still trying to fetch data from `iter`. - Add tests to validate Pull Request resolved: #1075 Reviewed By: NivekT Differential Revision: D44168655 Pulled By: ejguan fbshipit-source-id: fdfee5c27b512b5c0d5308e53a81b1cb2db70a43

facebook-github-bot added the CLA Signed This label is managed by the Facebook bot. Authors need to sign the CLA before a PR can be reviewed. label Mar 10, 2023

ejguan requested a review from NivekT March 10, 2023 22:47

ejguan marked this pull request as draft March 13, 2023 16:42

ejguan removed the request for review from NivekT March 13, 2023 16:42

ejguan force-pushed the exit_mprs branch from 3081c0d to 237d7bc Compare March 15, 2023 21:55

ejguan changed the title ~~Properly cleanup processes and queues for MPRS~~ Properly cleanup processes and queues for MPRS and Fix pause for prefetch Mar 15, 2023

ejguan marked this pull request as ready for review March 15, 2023 22:17

ejguan requested a review from NivekT March 15, 2023 22:17

ejguan commented Mar 15, 2023

View reviewed changes

torchdata/dataloader2/dataloader2.py Show resolved Hide resolved

torchdata/datapipes/iter/util/prefetcher.py Show resolved Hide resolved

ejguan force-pushed the exit_mprs branch from 237d7bc to efc2a1e Compare March 16, 2023 14:02

NivekT approved these changes Mar 16, 2023

View reviewed changes

ejguan force-pushed the exit_mprs branch from 142c330 to cd7e400 Compare March 16, 2023 16:52

ejguan force-pushed the exit_mprs branch from 498c906 to e6e4586 Compare March 17, 2023 00:31

ejguan force-pushed the exit_mprs branch from e6e4586 to 1df138a Compare March 17, 2023 01:08

facebook-github-bot closed this in 8eecb77 Mar 17, 2023

facebook-github-bot added the Merged label Mar 17, 2023

facebook-github-bot added the Reverted label Mar 17, 2023

ejguan mentioned this pull request Mar 17, 2023

Properly cleanup processes and queues for MPRS and Fix pause for prefetch (#1075) #1096

Closed

		@@ -29,12 +32,60 @@ def _add_one(x: int) -> int:
		dp_parametrize = parametrize("dp", test_dps)

Conversation

ejguan commented Mar 10, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes

Uh oh!

facebook-github-bot commented Mar 10, 2023

Uh oh!

ejguan commented Mar 10, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

facebook-github-bot commented Mar 13, 2023

Uh oh!

ejguan commented Mar 13, 2023

Uh oh!

facebook-github-bot commented Mar 15, 2023

Uh oh!

Uh oh!

Uh oh!

facebook-github-bot commented Mar 16, 2023

Uh oh!

NivekT left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

facebook-github-bot commented Mar 16, 2023

Uh oh!

facebook-github-bot commented Mar 16, 2023

Uh oh!

facebook-github-bot commented Mar 16, 2023

Uh oh!

facebook-github-bot commented Mar 17, 2023

Uh oh!

facebook-github-bot commented Mar 17, 2023

Uh oh!

facebook-github-bot commented Mar 17, 2023

Uh oh!

facebook-github-bot commented Mar 17, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ejguan commented Mar 10, 2023 •

edited

Loading

ejguan commented Mar 10, 2023 •

edited

Loading