feat: Better parallelism - #408
Conversation
Rather than waiting for entire batches to complete, we re-query whenever there's available parallelism, until we find no more blobs. Note that this *only* works when new blobs become available while we're waiting to send some into `addBlobs`. If blobs become available between `addBlobs` consuming all the ones we had (beginning their uploads) and all of those uploads completing, we won't catch them until the next round one level up. In that case, we still waste parallel capacity. Therefore, this is still an incomplete solution, but moves in the right direction.
05d8b18 to
fc0cd47
Compare
fc0cd47 to
95a5177
Compare
alanshaw
left a comment
alanshaw
left a comment
There was a problem hiding this comment.
Happy for this to merge but please consider the feedback. I would most like to have a type for the non-fatal task return value - I suggested a "report" that could contain warning messages, but I'm not attached to that naming.
| err = fmt.Errorf("failed to add shard %s: %w", shard, err) | ||
| // [gtypes.BlobUploadError]s are non-fatal. | ||
| var errBlobUpload gtypes.BlobUploadError | ||
| if errors.As(err, &errBlobUpload) { |
There was a problem hiding this comment.
Could just use errors.Is? You're not using errBlobUpload at all.
There was a problem hiding this comment.
Minor, but personally I'd flip this round to return early for the fatal case.
There was a problem hiding this comment.
Could just use
errors.Is? You're not usingerrBlobUploadat all.
errors.Is is for instances, errors.As is for types, unless I'm misunderstanding.
| // error that should cause immediate cancellation of all other tasks. | ||
| type IDTask struct { | ||
| ID id.ID | ||
| Run func(context.Context) (error, error) |
There was a problem hiding this comment.
Perhaps use a type for the non-fatal error e.g.
type Report struct {
Warnings []error
}It's just weird IMO to see a function that returns two errors.
| } | ||
|
|
||
| if err := a.updateBlob(ctx, blob); err != nil { | ||
| if err := a.updateBlob(context.WithoutCancel(ctx), blob); err != nil { |
There was a problem hiding this comment.
Because we're recording something that happened externally. We need to make sure we do that. We can't make the DB update truly transactional with the external side effect, but we can at least make sure we don't give up on recording what happened because the context canceled. Otherwise we might do things like upload the same shard twice. (And given updateBlob() is a simple DB update, the potential cost is low.)
| // [id.ID]. The task's Run function returns two errors: a non-fatal error that | ||
| // can be collected and reported after all tasks have completed, and a fatal | ||
| // error that should cause immediate cancellation of all other tasks. | ||
| type IDTask struct { |
There was a problem hiding this comment.
Do (should?) we have non-ID tasks? Is this just a Task?
There was a problem hiding this comment.
Should this live in errors.go?
There was a problem hiding this comment.
My intent here was separation of concerns: the Worker code doesn't handle deduping tasks and leaves it entirely to the worker definition which uses Worker. In particular, that means the non-parallel workers don't have to make up a bogus "ID" to satisfy an interface, they just don't do any deduping. I'm certainly open to thoughts here, though; I debated it a bit myself.
| // can be collected and reported after all tasks have completed, and a fatal | ||
| // error that should cause immediate cancellation of all other tasks. | ||
| type IDTask struct { | ||
| ID id.ID |
There was a problem hiding this comment.
Suggestion: add a non-unique Desc string which could be used in error messages.
|
@alanshaw @hannahhoward How's this look? Part of the intention with this |
2 participants
Previously, the
Workerwould always run in batches. When signaled, it would find new work to do, then perform all of that work, then wait for a new signal. Some workers would perform their work in parallel, but only within the batch, meaning that one long-lived task in the batch would waste the rest of the parallelism, and no new tasks would be pulled in until the entire batch completed.Now,
Workerunderstands tasks and parallelism natively. Whenever it has free parallelism, it looks up pending tasks. Then it queues those as it works through them. Notably, eachWorkerdefinition is responsible for tracking what's in flight and not re-queuing it. That is, since we don't store that we've placed a task in the queue, we'll keep finding it in the DB until the task is complete, and we need to filter that out. But it's pretty straightforward to manage that by DB ID. We keep that in memory rather than in the DB, because if the process is interrupted it naturally results in a consistent state: the queue is emptied, and so is the set of what we consider "in-flight".My gut says that there's further we can push this, but this is a significant improvement as it is. I've tested it with
doupload. #409 demonstrates it with the (shell version of the) Smelt-baseddoupload.Closes #267