fix: Address unbounded memory growth in upload_stream when source outpaces upload - #3407
fix: Address unbounded memory growth in upload_stream when source outpaces upload#3407richardwang1124 wants to merge 8 commits into
Conversation
|
Detected 1 possible performance regressions:
|
jterapin
left a comment
jterapin
left a comment
There was a problem hiding this comment.
Thanks for picking this up! this is a real gap and a good catch. I've left a few questions on the current implementation inline. We're also missing a CHANGELOG entry, so we'll want to add one before this merges.
jterapin
left a comment
jterapin
left a comment
There was a problem hiding this comment.
The overall approach makes sense. I would like to see a fuller PR description for historical purposes (aside from the issue link).
| def upload_stream(options = {}, &block) | ||
| upload_opts = options.merge(bucket: bucket_name, key: key) | ||
| executor = DefaultExecutor.new(max_threads: upload_opts.delete(:thread_count)) | ||
| thread_count = upload_opts.delete(:thread_count) || DefaultExecutor::DEFAULT_MAX_THREADS |
There was a problem hiding this comment.
Could we expose the queue size as its own option instead of deriving it from thread_count? Right now max_queue: thread_count couples two separate concerns: how many uploads run in parallel, and how far ahead we're allowed to buffer.
Or this could be an additional thing to have down the road. Your choice.
| true | ||
| rescue ClosedQueueError | ||
| # shutdown or kill happened while parked on a full queue | ||
| raise 'Executor has been shutdown and is no longer accepting tasks' |
There was a problem hiding this comment.
We should define an actual error class for this. Something like, RejectedExecutionError which is what concurrent-ruby has.
| end | ||
| # Count only successfully queued parts; a failed post never runs the | ||
| # block, so it never pushes :done and must not be waited on below. | ||
| upload_attempts += 1 |
There was a problem hiding this comment.
Could rename this var into queued_parts to be consistent and accurate
| Unreleased Changes | ||
| ------------------ | ||
|
|
||
| * Issue - Bound memory usage in `upload_stream` when the source produces data faster than parts can be uploaded. |
There was a problem hiding this comment.
| * Issue - Bound memory usage in `upload_stream` when the source produces data faster than parts can be uploaded. | |
| * Issue - Fix unbounded memory growth in `upload_stream` when the data source outpaces the upload (#3407). |
| completed_part = { etag: resp.etag, part_number: p[:part_number] } | ||
| apply_part_checksum(resp, completed_part) | ||
| completed.push(completed_part) | ||
| begin |
There was a problem hiding this comment.
I believe we will have to apply similar changes to FileDownloader but you could this in a fast-follow PR.
| break if job == :shutdown | ||
|
|
||
| args, block = job | ||
| block.call(*args) |
There was a problem hiding this comment.
Now that the queue is bounded, I think this needs a guard around block.call. If a task raises outside StandardError (OOM, a signal), the worker dies here, and with a full queue the producer stays parked on push with nothing to drain it and no way to spawn a replacement, so the upload freezes silently.
Any thoughts on how to counter this?
3 participants
Fixes unbounded memory growth in
upload_streamwhen the source produces data faster than parts can be uploaded. Previously every part read off the pipe was buffered with no limit. NowDefaultExecutorsupports a bounded queue that applies backpressure andupload_streamsets a limit tothread_count. Also switches in memory read path fromcopy_streamto single right-sized read.Fixes #3393.
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
To make sure we include your contribution in the release notes, please make sure to add description entry for your changes in the "unreleased changes" section of the
CHANGELOG.mdfile (at corresponding gem). For the description entry, please make sure it lives in one line and starts withFeatureorIssuein the correct format.For generated code changes, please checkout below instructions first:
https://github.com/aws/aws-sdk-ruby/blob/version-3/CONTRIBUTING.md
Thank you for your contribution!