[jira] [Updated] (AIRFLOW-4401) multiprocessing.Queue.empty() is unreliable

["Jarek Potiuk (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/AIRFLOW-4401?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Jarek Potiuk updated AIRFLOW-4401:
----------------------------------
    Summary: multiprocessing.Queue.empty() is unreliable  (was: multiprocessing.Queue is unreliable)

> multiprocessing.Queue.empty() is unreliable
> -------------------------------------------
>
>                 Key: AIRFLOW-4401
>                 URL: https://issues.apache.org/jira/browse/AIRFLOW-4401
>             Project: Apache Airflow
>          Issue Type: Bug
>            Reporter: Jarek Potiuk
>            Priority: Major
>
> After discussing with [~ash] and [~BasPH] potential reasons for flakiness of LocalExecutor tests, I took a deeper dive into the problem and what I found raised the remaining hair on top of my head. 
> We had a number of flaky tests in the local executor that resulted in result_queue not being empty where it should have been emptied a moment before. More details and discussion can be found in [https://github.com/apache/airflow/pull/5159] 
> The problem turned out to be ... unreliability of multiprocessing.Queue empty() implementation. Itt turned out that multiprocessing.Queue.empty() implementation is not fully synchronized and it might return True even if put() operation has been already completed in another thread. What's more - empty() might return True even if qsize() of the queue returns > 0 (!) 
> It's a bit mind-boggling but it is "as intended' as documented in [https://bugs.python.org/issue23582]  (resolved as "not a bug" !!!!) 
> A few people have stumbled upon this problem. For example [https://github.com/vterron/lemon/commit/9ca6b4b1212228dbd4f69b88aaf88b12952d7d6f] and [https://github.com/keras-team/autokeras/issues/368] 
> Also we seemed to experienced that in Airflow before. In jobs.py year ago (31.07.2016) - we can see the comment below (but we used multiprocessing.queue empty() nevertheless):
> {code:java}
> # Not using multiprocessing.Queue() since it's no longer a separate
> # process and due to some unusual behavior. (empty() incorrectly
> # returns true?){code}
> The solution available in [https://bugs.python.org/issue23582]  using qsize() was not usable because qsize() does not work on MacOS (throws NotImplementedError).
> The working solution is to implement a reliable queue (SynchronizedQueue) based on [https://github.com/vterron/lemon/commit/9ca6b4b1212228dbd4f69b88aaf88b12952d7d6f] (butwith a twist that __init__ of class deriving from Queue has to be changed for python 3.4+ as described in [https://stackoverflow.com/questions/24941359/ctx-parameter-in-multiprocessing-queue].
> Luckily we are now Python3.5+
> We should replace all usages of multiprocessing.Queue where empty() is used with the SynchronizedQueue. And make sure we do not use multiprocessing.Queue in similar way in the future.
>  
>  
>  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)