diff options
Diffstat (limited to 'weed')
| -rw-r--r-- | weed/filer2/filechunks.go | 10 | ||||
| -rw-r--r-- | weed/filesys/filehandle.go | 3 | ||||
| -rw-r--r-- | weed/replication/sink/s3sink/s3_write.go | 2 | ||||
| -rw-r--r-- | weed/util/http_util.go | 21 |
4 files changed, 30 insertions, 6 deletions
diff --git a/weed/filer2/filechunks.go b/weed/filer2/filechunks.go index 8f9746324..39e43cf3c 100644 --- a/weed/filer2/filechunks.go +++ b/weed/filer2/filechunks.go @@ -70,6 +70,7 @@ type ChunkView struct { Offset int64 Size uint64 LogicOffset int64 + IsFullChunk bool } func ViewFromChunks(chunks []*filer_pb.FileChunk, offset int64, size int) (views []*ChunkView) { @@ -80,11 +81,13 @@ func ViewFromChunks(chunks []*filer_pb.FileChunk, offset int64, size int) (views for _, chunk := range visibles { if chunk.start <= offset && offset < chunk.stop && offset < stop { + isFullChunk := chunk.isFullChunk && chunk.start == offset && chunk.stop <= stop views = append(views, &ChunkView{ FileId: chunk.fileId, Offset: offset - chunk.start, // offset is the data starting location in this file id Size: uint64(min(chunk.stop, stop) - offset), LogicOffset: offset, + IsFullChunk: isFullChunk, }) offset = min(chunk.stop, stop) } @@ -116,6 +119,7 @@ func mergeIntoVisibles(visibles, newVisibles []*visibleInterval, chunk *filer_pb chunk.Offset+int64(chunk.Size), chunk.FileId, chunk.Mtime, + true, ) length := len(visibles) @@ -135,6 +139,7 @@ func mergeIntoVisibles(visibles, newVisibles []*visibleInterval, chunk *filer_pb chunk.Offset, v.fileId, v.modifiedTime, + false, )) } chunkStop := chunk.Offset + int64(chunk.Size) @@ -144,6 +149,7 @@ func mergeIntoVisibles(visibles, newVisibles []*visibleInterval, chunk *filer_pb v.stop, v.fileId, v.modifiedTime, + false, )) } if chunkStop <= v.start || v.stop <= chunk.Offset { @@ -195,14 +201,16 @@ type visibleInterval struct { stop int64 modifiedTime int64 fileId string + isFullChunk bool } -func newVisibleInterval(start, stop int64, fileId string, modifiedTime int64) *visibleInterval { +func newVisibleInterval(start, stop int64, fileId string, modifiedTime int64, isFullChunk bool) *visibleInterval { return &visibleInterval{ start: start, stop: stop, fileId: fileId, modifiedTime: modifiedTime, + isFullChunk: isFullChunk, } } diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go index b4c2920bb..e74228f3c 100644 --- a/weed/filesys/filehandle.go +++ b/weed/filesys/filehandle.go @@ -117,7 +117,8 @@ func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fus fmt.Sprintf("http://%s/%s", locations.Locations[0].Url, chunkView.FileId), chunkView.Offset, int(chunkView.Size), - buff[chunkView.LogicOffset-req.Offset:chunkView.LogicOffset-req.Offset+int64(chunkView.Size)]) + buff[chunkView.LogicOffset-req.Offset:chunkView.LogicOffset-req.Offset+int64(chunkView.Size)], + !chunkView.IsFullChunk) if err != nil { diff --git a/weed/replication/sink/s3sink/s3_write.go b/weed/replication/sink/s3sink/s3_write.go index 8ff722e67..5c4be7aee 100644 --- a/weed/replication/sink/s3sink/s3_write.go +++ b/weed/replication/sink/s3sink/s3_write.go @@ -161,6 +161,6 @@ func (s3sink *S3Sink) buildReadSeeker(chunk *filer2.ChunkView) (io.ReadSeeker, e return nil, err } buf := make([]byte, chunk.Size) - util.ReadUrl(fileUrl, chunk.Offset, int(chunk.Size), buf) + util.ReadUrl(fileUrl, chunk.Offset, int(chunk.Size), buf, true) return bytes.NewReader(buf), nil } diff --git a/weed/util/http_util.go b/weed/util/http_util.go index 7ae5713bb..77a7a5fa3 100644 --- a/weed/util/http_util.go +++ b/weed/util/http_util.go @@ -2,6 +2,7 @@ package util import ( "bytes" + "compress/gzip" "encoding/json" "errors" "fmt" @@ -184,24 +185,38 @@ func NormalizeUrl(url string) string { return "http://" + url } -func ReadUrl(fileUrl string, offset int64, size int, buf []byte) (n int64, e error) { +func ReadUrl(fileUrl string, offset int64, size int, buf []byte, isReadRange bool) (n int64, e error) { req, _ := http.NewRequest("GET", fileUrl, nil) - req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size))) + if isReadRange { + req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size))) + } else { + req.Header.Set("Accept-Encoding", "gzip") + } r, err := client.Do(req) if err != nil { return 0, err } + defer r.Body.Close() if r.StatusCode >= 400 { return 0, fmt.Errorf("%s: %s", fileUrl, r.Status) } + var reader io.ReadCloser + switch r.Header.Get("Content-Encoding") { + case "gzip": + reader, err = gzip.NewReader(r.Body) + defer reader.Close() + default: + reader = r.Body + } + var i, m int for { - m, err = r.Body.Read(buf[i:]) + m, err = reader.Read(buf[i:]) if m == 0 { return } |