ashb commented on a change in pull request #5200: [AIRFLOW-4401] Use managers
for Queue synchronization
URL: https://github.com/apache/airflow/pull/5200#discussion_r279339138
##########
File path: airflow/contrib/executors/kubernetes_executor.py
##########
@@ -373,7 +372,8 @@ def __init__(self, kube_config, task_queue, result_queue,
kube_client, worker_uu
self.kube_client = kube_client
self.launcher = PodLauncher(kube_client=self.kube_client)
self.worker_configuration =
WorkerConfiguration(kube_config=self.kube_config)
- self.watcher_queue = multiprocessing.Queue()
+ self._manager = multiprocessing.Manager()
+ self.watcher_queue = self._manager.Queue()
Review comment:
Instead of using this here could we instead use a normal Queue and use
get_nowait instead? Since the only use appears to be:
```python
while not self.watcher_queue.empty():
self.process_watcher_task()
```
could we instead do:
```python
try:
while True:
task = self.watcher_queue.get_nowait()
self.process_watcher_task(task)
except queue.Empty:
pass
```
(and change process_watcher_task fn to accept the event rather than having
it pull it off the queue itself).
i.e. don't ever call `.empty()` on this queue.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
