mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
filer: set file size, streaming chunk file uploading
fix https://github.com/chrislusf/seaweedfs/issues/1193
This commit is contained in:
parent
40ae533fa3
commit
382ff218d3
|
@ -32,7 +32,7 @@ var (
|
||||||
|
|
||||||
type FilerPostResult struct {
|
type FilerPostResult struct {
|
||||||
Name string `json:"name,omitempty"`
|
Name string `json:"name,omitempty"`
|
||||||
Size uint32 `json:"size,omitempty"`
|
Size int64 `json:"size,omitempty"`
|
||||||
Error string `json:"error,omitempty"`
|
Error string `json:"error,omitempty"`
|
||||||
Fid string `json:"fid,omitempty"`
|
Fid string `json:"fid,omitempty"`
|
||||||
Url string `json:"url,omitempty"`
|
Url string `json:"url,omitempty"`
|
||||||
|
@ -130,7 +130,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
|
||||||
// send back post result
|
// send back post result
|
||||||
reply := FilerPostResult{
|
reply := FilerPostResult{
|
||||||
Name: ret.Name,
|
Name: ret.Name,
|
||||||
Size: ret.Size,
|
Size: int64(ret.Size),
|
||||||
Error: ret.Error,
|
Error: ret.Error,
|
||||||
Fid: fileId,
|
Fid: fileId,
|
||||||
Url: urlLocation,
|
Url: urlLocation,
|
||||||
|
|
|
@ -1,10 +1,8 @@
|
||||||
package weed_server
|
package weed_server
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"context"
|
"context"
|
||||||
"io"
|
"io"
|
||||||
"io/ioutil"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"path"
|
"path"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
@ -92,66 +90,47 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
|
||||||
|
|
||||||
var fileChunks []*filer_pb.FileChunk
|
var fileChunks []*filer_pb.FileChunk
|
||||||
|
|
||||||
totalBytesRead := int64(0)
|
|
||||||
tmpBufferSize := int32(1024 * 1024)
|
|
||||||
tmpBuffer := bytes.NewBuffer(make([]byte, 0, tmpBufferSize))
|
|
||||||
chunkBuf := make([]byte, chunkSize+tmpBufferSize, chunkSize+tmpBufferSize) // chunk size plus a little overflow
|
|
||||||
chunkBufOffset := int32(0)
|
|
||||||
chunkOffset := int64(0)
|
chunkOffset := int64(0)
|
||||||
writtenChunks := 0
|
|
||||||
|
|
||||||
filerResult = &FilerPostResult{
|
for chunkOffset < contentLength {
|
||||||
Name: fileName,
|
limitedReader := io.LimitReader(part1, int64(chunkSize))
|
||||||
}
|
|
||||||
|
|
||||||
for totalBytesRead < contentLength {
|
// assign one file id for one chunk
|
||||||
tmpBuffer.Reset()
|
fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(w, r, replication, collection, dataCenter)
|
||||||
bytesRead, readErr := io.CopyN(tmpBuffer, part1, int64(tmpBufferSize))
|
if assignErr != nil {
|
||||||
readFully := readErr != nil && readErr == io.EOF
|
return nil, assignErr
|
||||||
tmpBuf := tmpBuffer.Bytes()
|
|
||||||
bytesToCopy := tmpBuf[0:int(bytesRead)]
|
|
||||||
|
|
||||||
copy(chunkBuf[chunkBufOffset:chunkBufOffset+int32(bytesRead)], bytesToCopy)
|
|
||||||
chunkBufOffset = chunkBufOffset + int32(bytesRead)
|
|
||||||
|
|
||||||
if chunkBufOffset >= chunkSize || readFully || (chunkBufOffset > 0 && bytesRead == 0) {
|
|
||||||
writtenChunks = writtenChunks + 1
|
|
||||||
fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(w, r, replication, collection, dataCenter)
|
|
||||||
if assignErr != nil {
|
|
||||||
return nil, assignErr
|
|
||||||
}
|
|
||||||
|
|
||||||
// upload the chunk to the volume server
|
|
||||||
chunkName := fileName + "_chunk_" + strconv.FormatInt(int64(len(fileChunks)+1), 10)
|
|
||||||
uploadErr := fs.doUpload(urlLocation, w, r, chunkBuf[0:chunkBufOffset], chunkName, "", fileId, auth)
|
|
||||||
if uploadErr != nil {
|
|
||||||
return nil, uploadErr
|
|
||||||
}
|
|
||||||
|
|
||||||
// Save to chunk manifest structure
|
|
||||||
fileChunks = append(fileChunks,
|
|
||||||
&filer_pb.FileChunk{
|
|
||||||
FileId: fileId,
|
|
||||||
Offset: chunkOffset,
|
|
||||||
Size: uint64(chunkBufOffset),
|
|
||||||
Mtime: time.Now().UnixNano(),
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
// reset variables for the next chunk
|
|
||||||
chunkBufOffset = 0
|
|
||||||
chunkOffset = totalBytesRead + int64(bytesRead)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
totalBytesRead = totalBytesRead + int64(bytesRead)
|
// upload the chunk to the volume server
|
||||||
|
chunkName := fileName + "_chunk_" + strconv.FormatInt(int64(len(fileChunks)+1), 10)
|
||||||
|
uploadedSize, uploadErr := fs.doUpload(urlLocation, w, r, limitedReader, chunkName, "", fileId, auth)
|
||||||
|
if uploadErr != nil {
|
||||||
|
return nil, uploadErr
|
||||||
|
}
|
||||||
|
|
||||||
if bytesRead == 0 || readFully {
|
// if last chunk exhausted the reader exactly at the border
|
||||||
|
if uploadedSize == 0 {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
if readErr != nil {
|
// Save to chunk manifest structure
|
||||||
return nil, readErr
|
fileChunks = append(fileChunks,
|
||||||
|
&filer_pb.FileChunk{
|
||||||
|
FileId: fileId,
|
||||||
|
Offset: chunkOffset,
|
||||||
|
Size: uint64(uploadedSize),
|
||||||
|
Mtime: time.Now().UnixNano(),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
glog.V(4).Infof("uploaded %s chunk %d to %s [%d,%d) of %d", fileName, len(fileChunks), fileId, chunkOffset, chunkOffset+int64(uploadedSize), contentLength)
|
||||||
|
|
||||||
|
// if last chunk was not at full chunk size, but already exhausted the reader
|
||||||
|
if uploadedSize < int64(chunkSize) {
|
||||||
|
break
|
||||||
}
|
}
|
||||||
|
// reset variables for the next chunk
|
||||||
|
chunkOffset = chunkOffset + int64(uploadedSize)
|
||||||
}
|
}
|
||||||
|
|
||||||
path := r.URL.Path
|
path := r.URL.Path
|
||||||
|
@ -176,6 +155,12 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
|
||||||
},
|
},
|
||||||
Chunks: fileChunks,
|
Chunks: fileChunks,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
filerResult = &FilerPostResult{
|
||||||
|
Name: fileName,
|
||||||
|
Size: chunkOffset,
|
||||||
|
}
|
||||||
|
|
||||||
if dbErr := fs.filer.CreateEntry(ctx, entry, false); dbErr != nil {
|
if dbErr := fs.filer.CreateEntry(ctx, entry, false); dbErr != nil {
|
||||||
fs.filer.DeleteChunks(entry.Chunks)
|
fs.filer.DeleteChunks(entry.Chunks)
|
||||||
replyerr = dbErr
|
replyerr = dbErr
|
||||||
|
@ -188,7 +173,7 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
|
||||||
}
|
}
|
||||||
|
|
||||||
func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *http.Request,
|
func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *http.Request,
|
||||||
chunkBuf []byte, fileName string, contentType string, fileId string, auth security.EncodedJwt) (err error) {
|
limitedReader io.Reader, fileName string, contentType string, fileId string, auth security.EncodedJwt) (size int64, err error) {
|
||||||
|
|
||||||
stats.FilerRequestCounter.WithLabelValues("postAutoChunkUpload").Inc()
|
stats.FilerRequestCounter.WithLabelValues("postAutoChunkUpload").Inc()
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
|
@ -196,13 +181,9 @@ func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *ht
|
||||||
stats.FilerRequestHistogram.WithLabelValues("postAutoChunkUpload").Observe(time.Since(start).Seconds())
|
stats.FilerRequestHistogram.WithLabelValues("postAutoChunkUpload").Observe(time.Since(start).Seconds())
|
||||||
}()
|
}()
|
||||||
|
|
||||||
ioReader := ioutil.NopCloser(bytes.NewBuffer(chunkBuf))
|
uploadResult, uploadError := operation.Upload(urlLocation, fileName, limitedReader, false, contentType, nil, auth)
|
||||||
uploadResult, uploadError := operation.Upload(urlLocation, fileName, ioReader, false, contentType, nil, auth)
|
|
||||||
if uploadResult != nil {
|
|
||||||
glog.V(0).Infoln("Chunk upload result. Name:", uploadResult.Name, "Fid:", fileId, "Size:", uploadResult.Size)
|
|
||||||
}
|
|
||||||
if uploadError != nil {
|
if uploadError != nil {
|
||||||
err = uploadError
|
return 0, uploadError
|
||||||
}
|
}
|
||||||
return
|
return int64(uploadResult.Size), nil
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue