-
Notifications
You must be signed in to change notification settings - Fork 2.1k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Addressing issue #419 #510
Merged
Merged
Changes from all commits
Commits
Show all changes
14 commits
Select commit
Hold shift + click to select a range
cae5ac1
issue #419 and fixing some things to be camel cased
c296171
Removed not needed comment of import
xibz f969c23
Fixed a bug where return was placed to late
xibz 1ef68c0
Checking content range if it is nil
xibz f6c89e0
Moved function call to variable
xibz 789a926
Added a content-range test when total is *
xibz 30f605b
Added a comment in an unclear spot
xibz 86c6606
Update to fix linting and some coding standards
xibz 904ce59
Merging master
xibz f5c7f5e
made more readible by using http constant
xibz 8755953
More elaborate comment
xibz a327e5b
Made a name more explicit on not being thread safe
xibz cbcc00a
Changing back the name
xibz dd0b428
Made it so states is passed in allowing for more states to be passed,…
xibz File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -3,11 +3,12 @@ package s3manager | |
import ( | ||
"fmt" | ||
"io" | ||
"net/http" | ||
"strconv" | ||
"strings" | ||
"sync" | ||
"time" | ||
|
||
"github.com/aws/aws-sdk-go/aws/awserr" | ||
"github.com/aws/aws-sdk-go/aws/awsutil" | ||
"github.com/aws/aws-sdk-go/aws/client" | ||
"github.com/aws/aws-sdk-go/aws/request" | ||
|
@@ -154,39 +155,48 @@ func (d *downloader) init() { | |
func (d *downloader) download() (n int64, err error) { | ||
d.init() | ||
|
||
// Spin up workers | ||
ch := make(chan dlchunk, d.ctx.Concurrency) | ||
for i := 0; i < d.ctx.Concurrency; i++ { | ||
d.wg.Add(1) | ||
go d.downloadPart(ch) | ||
} | ||
// Spin off first worker to check additional header information | ||
d.getChunk() | ||
|
||
if total := d.getTotalBytes(); total >= 0 { | ||
// Spin up workers | ||
ch := make(chan dlchunk, d.ctx.Concurrency) | ||
|
||
for i := 0; i < d.ctx.Concurrency; i++ { | ||
d.wg.Add(1) | ||
go d.downloadPart(ch) | ||
} | ||
|
||
// Assign work | ||
for d.geterr() == nil { | ||
if d.pos != 0 { | ||
// This is not the first chunk, let's wait until we know the total | ||
// size of the payload so we can see if we have read the entire | ||
// object. | ||
total := d.getTotalBytes() | ||
|
||
if total < 0 { | ||
// Total has not yet been set, so sleep and loop around while | ||
// waiting for our first worker to resolve this value. | ||
time.Sleep(10 * time.Millisecond) | ||
continue | ||
} else if d.pos >= total { | ||
// Assign work | ||
for d.getErr() == nil { | ||
if d.pos >= total { | ||
break // We're finished queueing chunks | ||
} | ||
|
||
// Queue the next range of bytes to read. | ||
ch <- dlchunk{w: d.w, start: d.pos, size: d.ctx.PartSize} | ||
d.pos += d.ctx.PartSize | ||
} | ||
|
||
// Queue the next range of bytes to read. | ||
ch <- dlchunk{w: d.w, start: d.pos, size: d.ctx.PartSize} | ||
d.pos += d.ctx.PartSize | ||
} | ||
// Wait for completion | ||
close(ch) | ||
d.wg.Wait() | ||
} else { | ||
// Checking if we read anything new | ||
for d.err == nil { | ||
d.getChunk() | ||
} | ||
|
||
// Wait for completion | ||
close(ch) | ||
d.wg.Wait() | ||
// We expect a 416 error letting us know we are done downloading the | ||
// total bytes. Since we do not know the content's length, this will | ||
// keep grabbing chunks of data until the range of bytes specified in | ||
// the request is out of range of the content. Once, this happens, a | ||
// 416 should occur. | ||
e, ok := d.err.(awserr.RequestFailure) | ||
if ok && e.StatusCode() == http.StatusRequestedRangeNotSatisfiable { | ||
d.err = nil | ||
} | ||
} | ||
|
||
// Return error | ||
return d.written, d.err | ||
|
@@ -199,39 +209,51 @@ func (d *downloader) download() (n int64, err error) { | |
// of bytes to be read so that the worker manager knows when it is finished. | ||
func (d *downloader) downloadPart(ch chan dlchunk) { | ||
defer d.wg.Done() | ||
|
||
for { | ||
chunk, ok := <-ch | ||
if !ok { | ||
break | ||
} | ||
d.downloadChunk(chunk) | ||
} | ||
} | ||
|
||
// getChunk grabs a chunk of data from the body. | ||
// Not thread safe. Should only used when grabbing data on a single thread. | ||
func (d *downloader) getChunk() { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Might make sense to move this logic into the |
||
chunk := dlchunk{w: d.w, start: d.pos, size: d.ctx.PartSize} | ||
d.pos += d.ctx.PartSize | ||
d.downloadChunk(chunk) | ||
} | ||
|
||
if d.geterr() == nil { | ||
// Get the next byte range of data | ||
in := &s3.GetObjectInput{} | ||
awsutil.Copy(in, d.in) | ||
rng := fmt.Sprintf("bytes=%d-%d", | ||
chunk.start, chunk.start+chunk.size-1) | ||
in.Range = &rng | ||
// downloadChunk downloads the chunk froom s3 | ||
func (d *downloader) downloadChunk(chunk dlchunk) { | ||
if d.getErr() != nil { | ||
return | ||
} | ||
// Get the next byte range of data | ||
in := &s3.GetObjectInput{} | ||
awsutil.Copy(in, d.in) | ||
rng := fmt.Sprintf("bytes=%d-%d", | ||
chunk.start, chunk.start+chunk.size-1) | ||
in.Range = &rng | ||
|
||
req, resp := d.ctx.S3.GetObjectRequest(in) | ||
req.Handlers.Build.PushBack(request.MakeAddToUserAgentFreeFormHandler("S3Manager")) | ||
err := req.Send() | ||
req, resp := d.ctx.S3.GetObjectRequest(in) | ||
req.Handlers.Build.PushBack(request.MakeAddToUserAgentFreeFormHandler("S3Manager")) | ||
err := req.Send() | ||
|
||
if err != nil { | ||
d.seterr(err) | ||
} else { | ||
d.setTotalBytes(resp) // Set total if not yet set. | ||
if err != nil { | ||
d.setErr(err) | ||
} else { | ||
d.setTotalBytes(resp) // Set total if not yet set. | ||
|
||
n, err := io.Copy(&chunk, resp.Body) | ||
resp.Body.Close() | ||
n, err := io.Copy(&chunk, resp.Body) | ||
resp.Body.Close() | ||
|
||
if err != nil { | ||
d.seterr(err) | ||
} | ||
d.incrwritten(n) | ||
} | ||
if err != nil { | ||
d.setErr(err) | ||
} | ||
d.incrWritten(n) | ||
} | ||
} | ||
|
||
|
@@ -259,37 +281,48 @@ func (d *downloader) setTotalBytes(resp *s3.GetObjectOutput) { | |
if resp.ContentRange == nil { | ||
// ContentRange is nil when the full file contents is provied, and | ||
// is not chunked. Use ContentLength instead. | ||
d.totalBytes = *resp.ContentLength | ||
return | ||
} | ||
if resp.ContentLength != nil { | ||
d.totalBytes = *resp.ContentLength | ||
return | ||
} | ||
} else { | ||
parts := strings.Split(*resp.ContentRange, "/") | ||
|
||
total := int64(-1) | ||
var err error | ||
// Checking for whether or not a numbered total exists | ||
// If one does not exist, we will assume the total to be -1, undefined, | ||
// and sequentially download each chunk until hitting a 416 error | ||
totalStr := parts[len(parts)-1] | ||
if totalStr != "*" { | ||
total, err = strconv.ParseInt(totalStr, 10, 64) | ||
if err != nil { | ||
d.err = err | ||
return | ||
} | ||
} | ||
|
||
parts := strings.Split(*resp.ContentRange, "/") | ||
total, err := strconv.ParseInt(parts[len(parts)-1], 10, 64) | ||
if err != nil { | ||
d.err = err | ||
return | ||
d.totalBytes = total | ||
} | ||
|
||
d.totalBytes = total | ||
} | ||
|
||
func (d *downloader) incrwritten(n int64) { | ||
func (d *downloader) incrWritten(n int64) { | ||
d.m.Lock() | ||
defer d.m.Unlock() | ||
|
||
d.written += n | ||
} | ||
|
||
// geterr is a thread-safe getter for the error object | ||
func (d *downloader) geterr() error { | ||
// getErr is a thread-safe getter for the error object | ||
func (d *downloader) getErr() error { | ||
d.m.Lock() | ||
defer d.m.Unlock() | ||
|
||
return d.err | ||
} | ||
|
||
// seterr is a thread-safe setter for the error object | ||
func (d *downloader) seterr(e error) { | ||
// setErr is a thread-safe setter for the error object | ||
func (d *downloader) setErr(e error) { | ||
d.m.Lock() | ||
defer d.m.Unlock() | ||
|
||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
could you expand this comment stating why 416 might be received?