Fix concurrency issues in tool execution #2730
Conversation
gspencergoog
force-pushed
the
fix_concurrency
branch
from
ce7032c to
cd58b5e
Compare
gspencergoog
changed the title
gspencergoog
force-pushed
the
fix_concurrency
branch
from
cd58b5e to
f55aa5f
Compare
gspencergoog
force-pushed
the
fix_concurrency
branch
from
f55aa5f to
a930149
Compare
gspencergoog
force-pushed
the
fix_concurrency
branch
from
6b5dde4 to
8625d68
Compare
srawlins
left a comment
srawlins
left a comment
There was a problem hiding this comment.
Great implementation. Just a few comments.
srawlins
left a comment
srawlins
left a comment
There was a problem hiding this comment.
I love this, but I'd love a review from @jcollins-g as well before we merge.
jcollins-g
left a comment
jcollins-g
left a comment
There was a problem hiding this comment.
This looks really good, a definite improvement on my original implementation. Function typedefs for the win, too!
| class MultiFutureTracker<T> { | ||
| /// Approximate maximum number of simultaneous active Futures. | ||
| final int parallel; | ||
| typedef TaskQueueClosure<T> = Future<T> Function(); |
There was a problem hiding this comment.
Oh, this makes things so much easier to read.
|
|
||
| final Queue<_TaskQueueItem<T>> _pendingTasks = Queue<_TaskQueueItem<T>>(); | ||
| final Set<_TaskQueueItem<T>> _activeTasks = <_TaskQueueItem<T>>{}; | ||
| final Set<Completer<void>> _completeListeners = <Completer<void>>{}; |
There was a problem hiding this comment.
yes, this structure is more easily verified than the queue-length original version for waiting on completion. nice.
lib/src/io_utils.dart
Outdated
| Future<T> add(TaskQueueClosure<T> task) { | ||
| final completer = Completer<T>(); | ||
| _pendingTasks.add(_TaskQueueItem<T>(task, completer)); | ||
| _processTasks(); |
There was a problem hiding this comment.
IIUC this seems like it will spawn a lot of _processTasks() that could potentially do "duplicate" work in the case where the _processTask() from the onComplete has it covered. This is probably fine though as the duplicates are written to collapse and do nothing. The old version probably got into trouble doing things cleverly to avoid this...
There was a problem hiding this comment.
Yes, possibly, but it would also make sure that when a new task is added, it starts running right away.
There was a problem hiding this comment.
Sure. I was thinking of the case though when the maximum concurrency is reached and _processTasks() here doesn't have any useful effect. It seems like it should be OK just to inline _processTasks() here and eliminate an unawaited future that sometimes doesn't do anything. Your choice, and of course, only if it works just as well.
There was a problem hiding this comment.
Oh, good point, that makes sense. OK, I inlined it.
3 participants
Description
This replaces the task queue called
MultiFutureTrackerwith aTaskQueue, which fixes the concurrency issue where the old class didn't really wait for tasks to complete before starting a new one.This uses a completer to provide a new future to wait for each task, uses a
Queueto keep the tasks, and returns aFuture<T>with the result of the task instead of waiting on aFuture<void>.There is also a future to wait on that will complete when the queue is empty (replacing
wait) calledtasksComplete.Related Issues
Tests