Fixdownloadbug (#363)

Fixdownloadbug

Author: gongweibao

go/filemanager/pfsmodules/cp.go

@@ -149,7 +149,7 @@ func RunCp(cmd *CpCmd) error {
 }
 
 if err != nil {
-ColorError("err:%s\n", err)
+ColorError("proc %s err:%s\n", arg, err)
 return err
 }
 

go/filemanager/pfsmodules/download.go

@@ -31,47 +31,48 @@ func downloadFile(src string, srcFileSize int64, dst string, verbose bool, chunk
 
 for {
 start := time.Now()
-m, errm := r.GetChunkMeta(offset, chunkSize)
-if errm != nil && errm != io.EOF {
-return errm
+sm, errs := r.GetChunkMeta(offset, chunkSize)
+if errs != nil && errs != io.EOF {
+return errs
 }
-log.V(2).Infoln("remote chunk info:", m)
+log.V(2).Infoln("remote chunk info:", sm)
 
-c, errc := w.ReadChunk(offset, chunkSize)
-if errc != nil && errc != io.EOF {
-return errc
+wm, errw := w.GetChunkMeta(offset, chunkSize)
+if errw != nil && errw != io.EOF {
+return errw
 }
-offset += m.Len
-log.V(2).Infoln("local chunk info:" + c.String())
+log.V(2).Infoln("local chunk info:", wm)
 
-if m.Checksum == c.Checksum {
+if sm.Checksum == wm.Checksum {
 if verbose {
 used := time.Since(start).Nanoseconds() / time.Millisecond.Nanoseconds()
-ColorInfoOverWrite("%s download %d%% %dKB/s", src, offset*100/srcFileSize, m.Len/used)
+ColorInfoOverWrite("%s download %d%% %dKB/s", src, offset*100/srcFileSize, sm.Len/used)
 }
-log.V(2).Infof("remote chunk is same as local chunk:%s\n\n", c.String())
-if errc == io.EOF || errm == io.EOF {
+offset += sm.Len
+log.V(2).Infoln("remote chunk is same as local chunk:", sm)
+if errs == io.EOF || errw == io.EOF {
 break
 }
 continue
 }
 
-c, err := r.ReadChunk(offset, m.Len)
+c, err := r.ReadChunk(offset, sm.Len)
 if err != nil && err != io.EOF {
 return err
 }
 
 if err := w.WriteChunk(c); err != nil {
 return err
 }
+offset += sm.Len
 
 if verbose {
 used := time.Since(start).Nanoseconds() / time.Millisecond.Nanoseconds()
-ColorInfoOverWrite("%s download %d%% %dKB/s", src, offset*100/srcFileSize, m.Len/used)
+ColorInfoOverWrite("%s download %d%% %dKB/s", src, offset*100/srcFileSize, sm.Len/used)
 }
 
 log.V(2).Infof("downlod chunk:%s ok\n\n", c.String())
-if errc == io.EOF || errm == io.EOF {
+if errs == io.EOF || errw == io.EOF {
 break
 }
 }
@@ -128,7 +129,7 @@ func download(src, dst string, verbose bool, chunkSize int64) error {
 ColorError("Download %s to %s error info:%s\n", realSrc, realDst, err)
 return err
 }
-ColorInfo("Downloaded %s\n", realSrc)
+ColorInfoOverWrite("Downloaded %s\n", realSrc)
 }
 
 return nil

go/filemanager/pfsmodules/localfile.go

@@ -125,11 +125,15 @@ func (f *FileHandle) CopyN(w io.Writer, offset, len int64) error {
 n, err := io.CopyN(w, f.F, len)
 log.V(2).Infof("CopyN expect %d real %d\n", len, n)
 
-if err != nil && err != io.EOF {
+if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 return err
 }
 f.Offset += int64(n)
 
+if err == io.ErrUnexpectedEOF {
+err = io.EOF
+}
+
 return err
 }
 

go/filemanager/pfsmodules/remotefile.go

@@ -73,18 +73,26 @@ func getChunkData(m ChunkParam) (*Chunk, error) {
 break
 }
 
-if part.FormName() == "chunk" {
-m1, err := ParseChunkParam(part.FileName())
-if err != nil {
-return nil, errors.New(err.Error())
-}
-
-c = NewChunk(m1.Size)
-c.Len = m1.Size
-c.Offset = m1.Offset
-if _, err := part.Read(c.Data); err != nil && err != io.EOF {
-return nil, err
-}
+if part.FormName() != "chunk" {
+continue
+}
+
+log.V(2).Infof("received post chunk param:%s\n", part.FileName())
+m1, err := ParseChunkParam(part.FileName())
+if err != nil {
+return nil, errors.New(err.Error())
+}
+
+c = NewChunk(m1.Size)
+c.Len = m1.Size
+c.Offset = m1.Offset
+n, err := io.ReadFull(part, c.Data)
+if err != nil {
+return c, err
+}
+
+if int64(n) != m1.Size {
+log.V(2).Infof("download chunk data error expected %d real %d", m1.Size, n)
 }
 }
 

go/filemanager/pfsmodules/stat.go

@@ -30,7 +30,6 @@ func (p *StatCmd) ToURLParam() url.Values {
 parameters.Add("path", p.Path)
 
 return parameters
-
 }
 
 // ToJSON encodes memebers to json string.

go/filemanager/pfsserver/handler.go

@@ -329,64 +329,60 @@ func checkChunkAuthorization(path string, header http.Header, cmdName string) (s
 return user, resp
 }
 
-// GetChunkHandler processes GET Chunk request.
-func GetChunkHandler(w http.ResponseWriter, r *http.Request) {
-log.V(1).Infof("begin proc GetChunkHandler")
-
-p, err := pfsmod.ParseChunkParam(r.URL.RawQuery)
-if err != nil {
-writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, response{})
-return
-}
-
-user, resp := checkChunkAuthorization(p.Path, r.Header, "GetChunk")
-if resp.Err != "" {
-writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, resp)
-return
-}
-
+func getChunk(w http.ResponseWriter, r *http.Request, param *pfsmod.ChunkParam) error {
 writer := multipart.NewWriter(w)
 writer.SetBoundary(pfsmod.DefaultMultiPartBoundary)
 
-fileName := p.ToURLParam().Encode()
+fileName := param.ToURLParam().Encode()
 part, err := writer.CreateFormFile("chunk", fileName)
 if err != nil {
-log.Error(err)
-writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, response{})
-return
+return err
 }
-
-resp = response{}
+defer writer.Close()
 
 fr := pfsmod.FileHandle{}
-if err := fr.Open(p.Path, os.O_RDONLY, 0); err != nil {
-resp.Err = err.Error()
-log.Error(err)
-writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, resp)
-return
+if err := fr.Open(param.Path, os.O_RDONLY, 0); err != nil {
+return err
 }
 defer fr.Close()
 
-log.Infof("user:%s download %s\n", user, p.String())
-
-if err = fr.CopyN(part, p.Offset, p.Size); err != nil {
+err = fr.CopyN(part, param.Offset, param.Size)
+if err != nil {
 if err != io.EOF {
-resp.Err = err.Error()
-log.Error(err)
-writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, resp)
-return
+return err
 }
 
-resp.Err = pfsmod.StatusFileEOF
-writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, resp)
+return errors.New(pfsmod.StatusFileEOF)
 }
+return nil
+}
 
-err = writer.Close()
+// GetChunkHandler processes GET Chunk request.
+func GetChunkHandler(w http.ResponseWriter, r *http.Request) {
+log.V(1).Infof("begin proc GetChunkHandler")
+
+p, err := pfsmod.ParseChunkParam(r.URL.RawQuery)
 if err != nil {
-log.Error(err)
+writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, response{})
 return
 }
 
+user, resp := checkChunkAuthorization(p.Path, r.Header, "GetChunk")
+if resp.Err != "" {
+writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, resp)
+return
+}
+log.Infof("user:%s download %s\n", user, p.String())
+
+err = getChunk(w, r, p)
+if err != nil {
+if err.Error() != pfsmod.StatusFileEOF {
+log.Errorln("cmd %s error info:%s", p.String(), err)
+}
+resp.Err = err.Error()
+}
+
+writeJSONResponse(w, r.URL.RawQuery, http.StatusOK, resp)
 log.V(1).Info("end proc GetChunkHandler")
 return
 }

go/filemanager/pfsserver/handler_test.go

@@ -0,0 +1,64 @@
+package pfsserver
+
+import (
+"crypto/md5"
+"encoding/hex"
+"fmt"
+"io"
+"mime/multipart"
+"net/http/httptest"
+"reflect"
+"testing"
+
+pfsmod "github.com/PaddlePaddle/cloud/go/filemanager/pfsmodules"
+)
+
+func TestGetChunk(t *testing.T) {
+param := pfsmod.ChunkParam{}
+param.Path = "./testdata/test_lt_chunk.dat"
+param.Offset = 0
+param.Size = int64(4096)
+
+req := httptest.NewRequest("GET", "http://example.com/foo", nil)
+w := httptest.NewRecorder()
+err := getChunk(w, req, &param)
+if err == nil ||
+err.Error() != pfsmod.StatusFileEOF {
+t.Error(err)
+}
+
+fmt.Println(w)
+
+partReader := multipart.NewReader(w.Body, pfsmod.DefaultMultiPartBoundary)
+for {
+part, err := partReader.NextPart()
+if err == io.EOF {
+break
+}
+
+if part.FormName() != "chunk" {
+continue
+}
+
+m1, err := pfsmod.ParseChunkParam(part.FileName())
+if err != nil {
+t.Error(err)
+}
+
+if reflect.DeepEqual(m1, param) {
+t.Error("m1 != param, m1:%v", m1)
+}
+
+data := make([]byte, m1.Size)
+n, err := io.ReadFull(part, data)
+if err != io.ErrUnexpectedEOF {
+t.Error(err)
+}
+
+sum := md5.Sum(data[:n])
+checksum := hex.EncodeToString(sum[:])
+if checksum != "77c58f04583c86f78c51df158e3f35e8" {
+t.Error("checksum error")
+}
+}
+}

go/filemanager/pfsserver/testdata/test_lt_chunk.dat

@@ -0,0 +1,7 @@
+1
+2
+3
+4
+5
+6
+7