[AIRFLOW-4401] Use managers for Queue synchronization #5200
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
6 tasks
potiuk
force-pushed
the
fix_queues_with_managers
branch
3 times, most recently
from
c293256
to
6b09440
Compare
potiuk
changed the title
potiuk
force-pushed
the
fix_queues_with_managers
branch
from
6b09440
to
a53341d
Compare
Codecov Report
@@ Coverage Diff @@
## master #5200 +/- ##
==========================================
+ Coverage 78.28% 78.54% +0.25%
==========================================
Files 469 469
Lines 29912 29952 +40
==========================================
+ Hits 23418 23527 +109
+ Misses 6494 6425 -69
Continue to review full report at Codecov.
|
potiuk
commented
Apr 29, 2019
tests/test_jobs.py
Outdated
| @@ -206,7 +206,7 @@ def test_trigger_controller_dag(self): | |||
|
|
|||
| scheduler = SchedulerJob() | |||
| queue = Mock() | |||
| scheduler._process_task_instances(target_dag, queue=queue) | |||
| scheduler._process_task_instances(target_dag, task_instances_list=queue) | |||
There was a problem hiding this comment.
I will rename all the queues later when we discuss which solution to choose.
ashb
reviewed
Apr 29, 2019
ashb
reviewed
Apr 29, 2019
potiuk
force-pushed
the
fix_queues_with_managers
branch
from
a53341d
to
ae5ad4d
Compare
BasPH
reviewed
Apr 29, 2019
potiuk
force-pushed
the
fix_queues_with_managers
branch
6 times, most recently
from
b1d8793
to
6ae1ac6
Compare
potiuk
commented
Apr 30, 2019
| @@ -769,3 +793,7 @@ def _change_state(self, key, state, pod_id): | |||
| def end(self): | |||
| self.log.info('Shutting down Kubernetes executor') | |||
| self.task_queue.join() | |||
There was a problem hiding this comment.
I think we had problem here before. task_queue.join() was used but task_done() never called., so I think we could have some hangs when trying to stop gracefully KubernetesScheduler.
|
@BasPH @Fokko @ashb -> I made some amendments now as I realised that Manager().Queue() is always a JoinableQueue (unlike multiprocessing.Queue). I updated the pattern everywhere to follow this general pattern below (also described it in the commit message). Since the managed queue is anyhow proxied to the shared one, get_nowait() will always return the message if it already has been put to the queue by another process. BTW. the get_nowait() message in case of managed queue is not really "instant" as in case of standard multiprocessing.Queue - it will poll the shared process, but that's exactly why it is good for us :). It also seems that in KubernetesExecutor there was a mistake. See the comment above (#5200 (comment)) - if we've heard about inability to gracefully shutdown the KubernetesExecutor - that could be it. |
potiuk
force-pushed
the
fix_queues_with_managers
branch
3 times, most recently
from
471df47
to
384af0f
Compare
milton0825
reviewed
Apr 30, 2019
| self.process_watcher_task(task) | ||
| finally: | ||
| self.watcher_queue.task_done() | ||
| except queue.Empty: |
There was a problem hiding this comment.
Suggested change
| except queue.Empty: | |
| except Empty: |
milton0825
approved these changes
Apr 30, 2019
It is a known problem https://bugs.python.org/issue23582 that multiprocessing.Queue empty() method is not reliable - sometimes it might return True even if another process already put something in the queue. This resulted in some of the tasks not picked up when sync() methods were called (in AirflowKubernetesScheduler, LocalExecutor, DagFileProcessor). This was less of a problem if the method was called in sync() - as the remaining jobs/files could be processed in next pass but it was a problem in tests and when graceful shutdown was executed (some tasks could be still unprocessed while the shutdown occured). We switched to Managers() managed queues to handle that - the queue in this case is run in a separate subprocess and each process using it uses a proxy to access this shared queue. Additionally all Queues() returned by managers are Joinable Queues so we should run task_done() after all processing and we can now perform join() in termination/end code to wait until all tasks are actually processed, not only retrieved from the queue. That increases gracefulness of shutdown. All the cases impacted follow the same general pattern now: while True: try: res = queue.get_nowait() try: .... do some processing finally: queue.task_done() except Empty: break In all these cases overhead for inter-processing locking is negligible comparing to the action executed (Parsing DAG, executing job) so it appears it should be safe for concurrency as well.
potiuk
force-pushed
the
fix_queues_with_managers
branch
from
384af0f
to
aa16653
Compare
|
One last build and I think we are good to go :) |
potiuk
added a commit
that referenced
this pull request
Apr 30, 2019
It is a known problem https://bugs.python.org/issue23582 that multiprocessing.Queue empty() method is not reliable - sometimes it might return True even if another process already put something in the queue. This resulted in some of the tasks not picked up when sync() methods were called (in AirflowKubernetesScheduler, LocalExecutor, DagFileProcessor). This was less of a problem if the method was called in sync() - as the remaining jobs/files could be processed in next pass but it was a problem in tests and when graceful shutdown was executed (some tasks could be still unprocessed while the shutdown occured). We switched to Managers() managed queues to handle that - the queue in this case is run in a separate subprocess and each process using it uses a proxy to access this shared queue. Additionally all Queues() returned by managers are Joinable Queues so we should run task_done() after all processing and we can now perform join() in termination/end code to wait until all tasks are actually processed, not only retrieved from the queue. That increases gracefulness of shutdown. All the cases impacted follow the same general pattern now: while True: try: res = queue.get_nowait() try: .... do some processing finally: queue.task_done() except Empty: break In all these cases overhead for inter-processing locking is negligible comparing to the action executed (Parsing DAG, executing job) so it appears it should be safe for concurrency as well. (cherry picked from commit 6952b19)
|
That |
potiuk
added a commit
that referenced
this pull request
Apr 30, 2019
potiuk
added a commit
that referenced
this pull request
Apr 30, 2019
It is a known problem https://bugs.python.org/issue23582 that multiprocessing.Queue empty() method is not reliable - sometimes it might return True even if another process already put something in the queue. This resulted in some of the tasks not picked up when sync() methods were called (in AirflowKubernetesScheduler, LocalExecutor, DagFileProcessor). This was less of a problem if the method was called in sync() - as the remaining jobs/files could be processed in next pass but it was a problem in tests and when graceful shutdown was executed (some tasks could be still unprocessed while the shutdown occured). We switched to Managers() managed queues to handle that - the queue in this case is run in a separate subprocess and each process using it uses a proxy to access this shared queue. Additionally all Queues() returned by managers are Joinable Queues so we should run task_done() after all processing and we can now perform join() in termination/end code to wait until all tasks are actually processed, not only retrieved from the queue. That increases gracefulness of shutdown. All the cases impacted follow the same general pattern now: while True: try: res = queue.get_nowait() try: .... do some processing finally: queue.task_done() except Empty: break In all these cases overhead for inter-processing locking is negligible comparing to the action executed (Parsing DAG, executing job) so it appears it should be safe for concurrency as well. (cherry picked from commit 6952b19)
6 tasks
andriisoldatenko
pushed a commit
to andriisoldatenko/airflow
that referenced
this pull request
Jul 26, 2019
It is a known problem https://bugs.python.org/issue23582 that multiprocessing.Queue empty() method is not reliable - sometimes it might return True even if another process already put something in the queue. This resulted in some of the tasks not picked up when sync() methods were called (in AirflowKubernetesScheduler, LocalExecutor, DagFileProcessor). This was less of a problem if the method was called in sync() - as the remaining jobs/files could be processed in next pass but it was a problem in tests and when graceful shutdown was executed (some tasks could be still unprocessed while the shutdown occured). We switched to Managers() managed queues to handle that - the queue in this case is run in a separate subprocess and each process using it uses a proxy to access this shared queue. Additionally all Queues() returned by managers are Joinable Queues so we should run task_done() after all processing and we can now perform join() in termination/end code to wait until all tasks are actually processed, not only retrieved from the queue. That increases gracefulness of shutdown. All the cases impacted follow the same general pattern now: while True: try: res = queue.get_nowait() try: .... do some processing finally: queue.task_done() except Empty: break In all these cases overhead for inter-processing locking is negligible comparing to the action executed (Parsing DAG, executing job) so it appears it should be safe for concurrency as well.
wmorris75
pushed a commit
to modmed/incubator-airflow
that referenced
this pull request
Jul 29, 2019
It is a known problem https://bugs.python.org/issue23582 that multiprocessing.Queue empty() method is not reliable - sometimes it might return True even if another process already put something in the queue. This resulted in some of the tasks not picked up when sync() methods were called (in AirflowKubernetesScheduler, LocalExecutor, DagFileProcessor). This was less of a problem if the method was called in sync() - as the remaining jobs/files could be processed in next pass but it was a problem in tests and when graceful shutdown was executed (some tasks could be still unprocessed while the shutdown occured). We switched to Managers() managed queues to handle that - the queue in this case is run in a separate subprocess and each process using it uses a proxy to access this shared queue. Additionally all Queues() returned by managers are Joinable Queues so we should run task_done() after all processing and we can now perform join() in termination/end code to wait until all tasks are actually processed, not only retrieved from the queue. That increases gracefulness of shutdown. All the cases impacted follow the same general pattern now: while True: try: res = queue.get_nowait() try: .... do some processing finally: queue.task_done() except Empty: break In all these cases overhead for inter-processing locking is negligible comparing to the action executed (Parsing DAG, executing job) so it appears it should be safe for concurrency as well.
dharamsk
pushed a commit
to postmates/airflow
that referenced
this pull request
Aug 8, 2019
It is a known problem https://bugs.python.org/issue23582 that multiprocessing.Queue empty() method is not reliable - sometimes it might return True even if another process already put something in the queue. This resulted in some of the tasks not picked up when sync() methods were called (in AirflowKubernetesScheduler, LocalExecutor, DagFileProcessor). This was less of a problem if the method was called in sync() - as the remaining jobs/files could be processed in next pass but it was a problem in tests and when graceful shutdown was executed (some tasks could be still unprocessed while the shutdown occured). We switched to Managers() managed queues to handle that - the queue in this case is run in a separate subprocess and each process using it uses a proxy to access this shared queue. Additionally all Queues() returned by managers are Joinable Queues so we should run task_done() after all processing and we can now perform join() in termination/end code to wait until all tasks are actually processed, not only retrieved from the queue. That increases gracefulness of shutdown. All the cases impacted follow the same general pattern now: while True: try: res = queue.get_nowait() try: .... do some processing finally: queue.task_done() except Empty: break In all these cases overhead for inter-processing locking is negligible comparing to the action executed (Parsing DAG, executing job) so it appears it should be safe for concurrency as well. (cherry picked from commit 6952b19)
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Make sure you have checked all steps below.
Jira
Description
It is a known problem https://bugs.python.org/issue23582 that
multiprocessing.Queue empty() method is not reliable - sometimes it might
return True even if another process already put something in the queue.
This resulted in some of the tasks not picked up when sync() methods
were called (in AirflowKubernetesScheduler, LocalExecutor,
DagFileProcessor). This was less of a problem if the method was called in sync()
in tests and when graceful shutdown was executed (some tasks could be still
unprocessed while the shutdown occured).
We switched to Managers() managed queues to handle that - the queue in this case
is run in a separate subprocess and each process using it uses a proxy to access
this shared queue.
All the cases impacted follow the same pattern now:
while not queue.empty():
res = queue.get()
....
This loop runs always in single (main) process so it is safe to run it this way -
there is no risk that some other process will retrieve the data from the queue in
between empty() and get().
In all these cases overhead for inter-processing locking is negligible
comparing to the action executed (Parsing DAG, executing job)
so it appears it should be safe to merge the change.
Tests
Tests were there (but flaky)
Commits
Documentation
Code Quality
flake8