tyler.durden/compose
1
0
Fork 0
mirror of https://github.com/docker/compose.git synced 2025-07-23 13:45:00 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix race condition

Browse Source 
If processing of all objects finishes before the queue is drained,
parallel_execute_iter() returns prematurely.

Signed-off-by: Aanand Prasad <aanand.prasad@gmail.com>
This commit is contained in:
Aanand Prasad 2016-04-11 17:05:52 +01:00 committed by Daniel Nephin
parent ebae76bee8
commit 2160c787e3
1 changed files with 9 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
compose/parallel.py
View File

@ -17,6 +17,8 @@ from compose.utils import get_output_stream
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
STOP = object()
def parallel_execute(objects, func, get_name, msg, get_deps=None): def parallel_execute(objects, func, get_name, msg, get_deps=None):
"""Runs func on objects in parallel while ensuring that func is """Runs func on objects in parallel while ensuring that func is
@ -108,7 +110,7 @@ def parallel_execute_iter(objects, func, get_deps):
results = Queue() results = Queue()
state = State(objects) state = State(objects)
while not state.is_done(): while True:
feed_queue(objects, func, get_deps, results, state) feed_queue(objects, func, get_deps, results, state)
try: try:
@ -119,6 +121,9 @@ def parallel_execute_iter(objects, func, get_deps):
except thread.error: except thread.error:
raise ShutdownException() raise ShutdownException()
if event is STOP:
break
obj, _, exception = event obj, _, exception = event
if exception is None: if exception is None:
log.debug('Finished processing: {}'.format(obj)) log.debug('Finished processing: {}'.format(obj))
@ -170,6 +175,9 @@ def feed_queue(objects, func, get_deps, results, state):
t.start() t.start()
state.started.add(obj) state.started.add(obj)
if state.is_done():
results.put(STOP)
class UpstreamError(Exception): class UpstreamError(Exception):
pass pass