[WIP] Fix download and upload large file bugs

go/filemanager/pfsmodules/chunk.go

@@ -3,23 +3,24 @@ package pfsmodules
 import (
 	"errors"
 	"fmt"
-	"io"
 	"net/url"
-	"os"
 	"strconv"
-
-	log "github.com/golang/glog"
 )
 
-// Chunk respresents a chunk info.
-type Chunk struct {
+// ChunkParam respresents a chunk info.
+type ChunkParam struct {
 	Path   string
 	Offset int64
 	Size   int64
 }
 
+// ToString packs info of ChunkParam.
+func (p *ChunkParam) ToString() string {
+	return fmt.Sprintf("Path:%s Offset:%d Size:%d", p.Path, p.Offset, p.Size)
+}
+
 // ToURLParam encodes variables to url encoding parameters.
-func (p *Chunk) ToURLParam() url.Values {
+func (p *ChunkParam) ToURLParam() url.Values {
 	parameters := url.Values{}
 	parameters.Add("path", p.Path)
 
@@ -32,11 +33,11 @@ func (p *Chunk) ToURLParam() url.Values {
 	return parameters
 }
 
-// ParseChunk get a Chunk struct from path.
+// ParseChunkParam get a Chunk struct from path.
 // path example:
 // 	  path=/pfs/datacenter1/1.txt&offset=4096&chunksize=4096
-func ParseChunk(path string) (*Chunk, error) {
-	cmd := Chunk{}
+func ParseChunkParam(path string) (*ChunkParam, error) {
+	cmd := ChunkParam{}
 
 	m, err := url.ParseQuery(path)
 	if err != nil ||
@@ -61,38 +62,22 @@ func ParseChunk(path string) (*Chunk, error) {
 	return &cmd, nil
 }
 
-// LoadChunkData loads a specified chunk to io.Writer.
-func (p *Chunk) LoadChunkData(w io.Writer) error {
-	f, err := os.Open(p.Path)
-	if err != nil {
-		return err
-	}
-	defer Close(f)
-
-	_, err = f.Seek(p.Offset, 0)
-	if err != nil {
-		return err
-	}
-
-	loaded, err := io.CopyN(w, f, p.Size)
-	log.V(2).Infof("loaded:%d\n", loaded)
-	return err
+// Chunk is struct.
+type Chunk struct {
+	Offset   int64
+	Len      int64
+	Checksum string
+	Data     []byte
 }
 
-// SaveChunkData save data from io.Reader.
-func (p *Chunk) SaveChunkData(r io.Reader) error {
-	f, err := os.OpenFile(p.Path, os.O_WRONLY, 0600)
-	if err != nil {
-		return err
-	}
-	defer Close(f)
-
-	_, err = f.Seek(p.Offset, 0)
-	if err != nil {
-		return err
-	}
+// ToString packs info of Chunk
+func (c *Chunk) ToString() string {
+	return fmt.Sprintf("Offset:%d Len:%d Checksum:%s", c.Offset, c.Len, c.Checksum)
+}
 
-	writen, err := io.CopyN(f, r, p.Size)
-	log.V(2).Infof("chunksize:%d writen:%d\n", p.Size, writen)
-	return err
+// NewChunk make a Chunk struct.
+func NewChunk(capcity int64) *Chunk {
+	c := Chunk{}
+	c.Data = make([]byte, capcity)
+	return &c
 }

go/filemanager/pfsmodules/chunkmeta.go

@@ -1,17 +1,15 @@
 package pfsmodules
 
 import (
-	"crypto/md5"
-	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"net/url"
 	"os"
-	"sort"
 	"strconv"
+
+	"github.com/PaddlePaddle/cloud/go/utils/restclient"
 )
 
 const (
@@ -30,10 +28,16 @@ type ChunkMeta struct {
 	Len      int64  `json:"len"`
 }
 
+// ToString  pack a info tring of ChunkMeta.
+func (m *ChunkMeta) ToString() string {
+	return fmt.Sprintf("Offset:%d Checksum:%s Len:%d", m.Offset, m.Checksum, m.Len)
+}
+
 // ChunkMetaCmd is a command.
 type ChunkMetaCmd struct {
 	Method    string `json:"method"`
 	FilePath  string `json:"path"`
+	Offset    int64  `json:"offset"`
 	ChunkSize int64  `json:"chunksize"`
 }
 
@@ -46,6 +50,9 @@ func (p *ChunkMetaCmd) ToURLParam() url.Values {
 	str := fmt.Sprint(p.ChunkSize)
 	parameters.Add("chunksize", str)
 
+	str = fmt.Sprint(p.Offset)
+	parameters.Add("offset", str)
+
 	return parameters
 }
 
@@ -56,7 +63,23 @@ func (p *ChunkMetaCmd) ToJSON() ([]byte, error) {
 
 // Run is a functions which run ChunkMetaCmd.
 func (p *ChunkMetaCmd) Run() (interface{}, error) {
-	return GetChunkMeta(p.FilePath, p.ChunkSize)
+	f := FileHandle{}
+	if err := f.Open(p.FilePath, os.O_RDONLY, 0); err != nil {
+		return nil, err
+	}
+
+	defer f.Close()
+
+	c, err := f.ReadChunk(p.Offset, p.ChunkSize)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ChunkMeta{
+		Offset:   p.Offset,
+		Checksum: c.Checksum,
+		Len:      c.Len,
+	}, nil
 }
 
 func (p *ChunkMetaCmd) checkChunkSize() error {
@@ -87,11 +110,13 @@ func NewChunkMetaCmdFromURLParam(r *http.Request) (*ChunkMetaCmd, error) {
 	method := r.URL.Query().Get("method")
 	path := r.URL.Query().Get("path")
 	chunkStr := r.URL.Query().Get("chunksize")
+	offsetStr := r.URL.Query().Get("offset")
 
 	if len(method) == 0 ||
 		method != ChunkMetaCmdName ||
 		len(path) == 0 ||
-		len(chunkStr) == 0 {
+		len(chunkStr) == 0 ||
+		len(offsetStr) == 0 {
 		return nil, errors.New(http.StatusText(http.StatusBadRequest))
 	}
 
@@ -100,105 +125,46 @@ func NewChunkMetaCmdFromURLParam(r *http.Request) (*ChunkMetaCmd, error) {
 		return nil, errors.New(StatusBadChunkSize)
 	}
 
+	offset, err := strconv.ParseInt(offsetStr, 10, 64)
+	if err != nil {
+		return nil, errors.New(StatusBadChunkSize)
+	}
+
 	return &ChunkMetaCmd{
 		Method:    method,
 		FilePath:  path,
 		ChunkSize: chunkSize,
+		Offset:    offset,
 	}, nil
 }
-
-type metaSlice []ChunkMeta
-
-func (a metaSlice) Len() int           { return len(a) }
-func (a metaSlice) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
-func (a metaSlice) Less(i, j int) bool { return a[i].Offset < a[j].Offset }
-
-// GetDiffChunkMeta gets difference between srcMeta and dstMeta.
-func GetDiffChunkMeta(srcMeta []ChunkMeta, dstMeta []ChunkMeta) ([]ChunkMeta, error) {
-	if len(dstMeta) == 0 || len(srcMeta) == 0 {
-		return srcMeta, nil
-	}
-
-	if !sort.IsSorted(metaSlice(srcMeta)) {
-		sort.Sort(metaSlice(srcMeta))
-	}
-
-	if !sort.IsSorted(metaSlice(dstMeta)) {
-		sort.Sort(metaSlice(dstMeta))
-	}
-
-	dstIdx := 0
-	srcIdx := 0
-	diff := make([]ChunkMeta, 0, len(srcMeta))
-
-	for {
-		if srcMeta[srcIdx].Offset < dstMeta[dstIdx].Offset {
-			diff = append(diff, srcMeta[srcIdx])
-			srcIdx++
-		} else if srcMeta[srcIdx].Offset > dstMeta[dstIdx].Offset {
-			dstIdx++
-		} else {
-			if srcMeta[srcIdx].Checksum != dstMeta[dstIdx].Checksum {
-				diff = append(diff, srcMeta[srcIdx])
-			}
-
-			dstIdx++
-			srcIdx++
-		}
-
-		if dstIdx >= len(dstMeta) {
-			break
-		}
-
-		if srcIdx >= len(srcMeta) {
-			break
-		}
-	}
-
-	if srcIdx < len(srcMeta) {
-		diff = append(diff, srcMeta[srcIdx:]...)
+func remoteChunkMeta(path string, offset int64,
+	chunkSize int64) (*ChunkMeta, error) {
+	cmd := ChunkMetaCmd{
+		Method:    ChunkMetaCmdName,
+		FilePath:  path,
+		ChunkSize: chunkSize,
+		Offset:    offset,
 	}
 
-	return diff, nil
-}
-
-// GetChunkMeta gets chunk metas from path of file.
-func GetChunkMeta(path string, len int64) ([]ChunkMeta, error) {
-	f, err := os.Open(path)
+	t := fmt.Sprintf("%s/api/v1/pfs/chunks", Config.ActiveConfig.Endpoint)
+	ret, err := restclient.GetCall(t, cmd.ToURLParam())
 	if err != nil {
 		return nil, err
 	}
-	defer Close(f)
 
-	if len > defaultMaxChunkSize || len < defaultMinChunkSize {
-		return nil, errors.New(StatusBadChunkSize)
+	type chunkMetaResponse struct {
+		Err     string    `json:"err"`
+		Results ChunkMeta `json:"results"`
 	}
 
-	var metas []ChunkMeta
-
-	data := make([]byte, len)
-	offset := int64(0)
-
-	for {
-		n, err := f.Read(data)
-		if err != nil && err != io.EOF {
-			return nil, err
-		}
-
-		if err == io.EOF {
-			break
-		}
-
-		m := ChunkMeta{}
-		m.Offset = offset
-		sum := md5.Sum(data[:n])
-		m.Checksum = hex.EncodeToString(sum[:])
-		m.Len = int64(n)
-
-		metas = append(metas, m)
+	resp := chunkMetaResponse{}
+	if err := json.Unmarshal(ret, &resp); err != nil {
+		return nil, err
+	}
 
-		offset += int64(n)
+	if len(resp.Err) == 0 {
+		return &resp.Results, nil
 	}
 
-	return metas, nil
+	return &resp.Results, errors.New(resp.Err)
 }

go/filemanager/pfsmodules/chunkmeta_test.go

@@ -1,62 +1 @@
 package pfsmodules
-
-import (
-	"crypto/md5"
-	"fmt"
-	"testing"
-)
-
-func TestGetDiffMeta(t *testing.T) {
-	var src []ChunkMeta
-	var dst []ChunkMeta
-
-	var data []ChunkMeta
-
-	for i := 0; i < 4; i++ {
-		data = append(data, ChunkMeta{
-			Offset:   int64(i),
-			Checksum: fmt.Sprintf("%x", md5.Sum([]byte("1"))),
-			Len:      1})
-	}
-
-	src = data[0:3]
-	// check when len(dst) is 0
-	diff, _ := GetDiffChunkMeta(src, dst)
-	if len(diff) != 3 {
-		t.Error(len(diff))
-	}
-
-	for i := range diff {
-		if diff[i] != src[i] {
-			t.Error(i)
-		}
-	}
-
-	// check when dst is same as src
-	dst = src
-	diff, _ = GetDiffChunkMeta(src, dst)
-	if len(diff) != 0 {
-		t.Error(len(diff))
-	}
-
-	// check when dst is small than src
-	dst = dst[:0]
-	dst = append(dst, data[0])
-	dst = append(dst, data[2])
-
-	diff, _ = GetDiffChunkMeta(src, dst)
-	if len(diff) != 1 {
-		t.Error(len(diff))
-	}
-
-	if diff[0] != data[1] {
-		t.Error(0)
-	}
-
-	// check when dst is large then src
-	dst = data
-	diff, _ = GetDiffChunkMeta(src, dst)
-	if len(diff) != 0 {
-		t.Error(len(diff))
-	}
-}