Merge pull request #224 from tnextday/feature/chunked-file-support

Feature/chunked file support

4 files changed, 322 insertions, 14 deletions

diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go
new file mode 100644
index 000000000..70564cbd2
--- /dev/null
+++ b/go/operation/chunked_file.go
@@ -0,0 +1,213 @@
+package operation
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"sort"
+
+	"sync"
+
+	"github.com/chrislusf/seaweedfs/go/glog"
+	"github.com/chrislusf/seaweedfs/go/util"
+)
+
+var (
+	// when the remote server does not allow range requests (Accept-Ranges was not set)
+	ErrRangeRequestsNotSupported = errors.New("Range requests are not supported by the remote server")
+	// ErrInvalidRange is returned by Read when trying to read past the end of the file
+	ErrInvalidRange = errors.New("Invalid range")
+)
+
+type ChunkInfo struct {
+	Fid    string `json:"fid"`
+	Offset int64  `json:"offset"`
+	Size   int64  `json:"size"`
+}
+
+type ChunkList []*ChunkInfo
+
+type ChunkManifest struct {
+	Name   string    `json:"name,omitempty"`
+	Mime   string    `json:"mime,omitempty"`
+	Size   int64     `json:"size,omitempty"`
+	Chunks ChunkList `json:"chunks,omitempty"`
+}
+
+// seekable chunked file reader
+type ChunkedFileReader struct {
+	Manifest *ChunkManifest
+	Master   string
+	pos      int64
+	pr       *io.PipeReader
+	pw       *io.PipeWriter
+	mutex    sync.Mutex
+}
+
+func (s ChunkList) Len() int           { return len(s) }
+func (s ChunkList) Less(i, j int) bool { return s[i].Offset < s[j].Offset }
+func (s ChunkList) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
+
+func LoadChunkManifest(buffer []byte, isGzipped bool) (*ChunkManifest, error) {
+	if isGzipped {
+		var err error
+		if buffer, err = UnGzipData(buffer); err != nil {
+			return nil, err
+		}
+	}
+	cm := ChunkManifest{}
+	if e := json.Unmarshal(buffer, &cm); e != nil {
+		return nil, e
+	}
+	sort.Sort(cm.Chunks)
+	return &cm, nil
+}
+
+func (cm *ChunkManifest) Marshal() ([]byte, error) {
+	return json.Marshal(cm)
+}
+
+func (cm *ChunkManifest) DeleteChunks(master string) error {
+	deleteError := 0
+	for _, ci := range cm.Chunks {
+		if e := DeleteFile(master, ci.Fid, ""); e != nil {
+			deleteError++
+			glog.V(0).Infof("Delete %s error: %s, master: %s", ci.Fid, e.Error(), master)
+		}
+	}
+	if deleteError > 0 {
+		return errors.New("Not all chunks deleted.")
+	}
+	return nil
+}
+
+func readChunkNeedle(fileUrl string, w io.Writer, offset int64) (written int64, e error) {
+	req, err := http.NewRequest("GET", fileUrl, nil)
+	if err != nil {
+		return written, err
+	}
+	if offset > 0 {
+		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
+	}
+
+	resp, err := util.Do(req)
+	if err != nil {
+		return written, err
+	}
+	defer resp.Body.Close()
+
+	switch resp.StatusCode {
+	case http.StatusRequestedRangeNotSatisfiable:
+		return written, ErrInvalidRange
+	case http.StatusOK:
+		if offset > 0 {
+			return written, ErrRangeRequestsNotSupported
+		}
+	case http.StatusPartialContent:
+		break
+	default:
+		return written, fmt.Errorf("Read chunk needle error: [%d] %s", resp.StatusCode, fileUrl)
+
+	}
+	return io.Copy(w, resp.Body)
+}
+
+func (cf *ChunkedFileReader) Seek(offset int64, whence int) (int64, error) {
+	var err error
+	switch whence {
+	case 0:
+	case 1:
+		offset += cf.pos
+	case 2:
+		offset = cf.Manifest.Size - offset
+	}
+	if offset > cf.Manifest.Size {
+		err = ErrInvalidRange
+	}
+	if cf.pos != offset {
+		cf.Close()
+	}
+	cf.pos = offset
+	return cf.pos, err
+}
+
+func (cf *ChunkedFileReader) WriteTo(w io.Writer) (n int64, err error) {
+	cm := cf.Manifest
+	chunkIndex := -1
+	chunkStartOffset := int64(0)
+	for i, ci := range cm.Chunks {
+		if cf.pos >= ci.Offset && cf.pos < ci.Offset+ci.Size {
+			chunkIndex = i
+			chunkStartOffset = cf.pos - ci.Offset
+			break
+		}
+	}
+	if chunkIndex < 0 {
+		return n, ErrInvalidRange
+	}
+	for ; chunkIndex < cm.Chunks.Len(); chunkIndex++ {
+		ci := cm.Chunks[chunkIndex]
+		// if we need read date from local volume server first?
+		fileUrl, lookupError := LookupFileId(cf.Master, ci.Fid)
+		if lookupError != nil {
+			return n, lookupError
+		}
+		if wn, e := readChunkNeedle(fileUrl, w, chunkStartOffset); e != nil {
+			return n, e
+		} else {
+			n += wn
+			cf.pos += wn
+		}
+
+		chunkStartOffset = 0
+	}
+	return n, nil
+}
+
+func (cf *ChunkedFileReader) ReadAt(p []byte, off int64) (n int, err error) {
+	cf.Seek(off, 0)
+	return cf.Read(p)
+}
+
+func (cf *ChunkedFileReader) Read(p []byte) (int, error) {
+	return cf.getPipeReader().Read(p)
+}
+
+func (cf *ChunkedFileReader) Close() (e error) {
+	cf.mutex.Lock()
+	defer cf.mutex.Unlock()
+	return cf.closePipe()
+}
+
+func (cf *ChunkedFileReader) closePipe() (e error) {
+	if cf.pr != nil {
+		if err := cf.pr.Close(); err != nil {
+			e = err
+		}
+	}
+	cf.pr = nil
+	if cf.pw != nil {
+		if err := cf.pw.Close(); err != nil {
+			e = err
+		}
+	}
+	cf.pw = nil
+	return e
+}
+
+func (cf *ChunkedFileReader) getPipeReader() io.Reader {
+	cf.mutex.Lock()
+	defer cf.mutex.Unlock()
+	if cf.pr != nil && cf.pw != nil {
+		return cf.pr
+	}
+	cf.closePipe()
+	cf.pr, cf.pw = io.Pipe()
+	go func(pw *io.PipeWriter) {
+		_, e := cf.WriteTo(pw)
+		pw.CloseWithError(e)
+	}(cf.pw)
+	return cf.pr
+}
diff --git a/go/operation/compress.go b/go/operation/compress.go
new file mode 100644
index 000000000..b1105ba4b
--- /dev/null
+++ b/go/operation/compress.go
@@ -0,0 +1,59 @@
+package operation
+
+import (
+	"bytes"
+	"compress/flate"
+	"compress/gzip"
+	"io/ioutil"
+	"strings"
+
+	"github.com/chrislusf/seaweedfs/go/glog"
+)
+
+/*
+* Default more not to gzip since gzip can be done on client side.
+ */
+func IsGzippable(ext, mtype string) bool {
+	if strings.HasPrefix(mtype, "text/") {
+		return true
+	}
+	switch ext {
+	case ".zip", ".rar", ".gz", ".bz2", ".xz":
+		return false
+	case ".pdf", ".txt", ".html", ".htm", ".css", ".js", ".json":
+		return true
+	}
+	if strings.HasPrefix(mtype, "application/") {
+		if strings.HasSuffix(mtype, "xml") {
+			return true
+		}
+		if strings.HasSuffix(mtype, "script") {
+			return true
+		}
+	}
+	return false
+}
+
+func GzipData(input []byte) ([]byte, error) {
+	buf := new(bytes.Buffer)
+	w, _ := gzip.NewWriterLevel(buf, flate.BestCompression)
+	if _, err := w.Write(input); err != nil {
+		glog.V(2).Infoln("error compressing data:", err)
+		return nil, err
+	}
+	if err := w.Close(); err != nil {
+		glog.V(2).Infoln("error closing compressed data:", err)
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+func UnGzipData(input []byte) ([]byte, error) {
+	buf := bytes.NewBuffer(input)
+	r, _ := gzip.NewReader(buf)
+	defer r.Close()
+	output, err := ioutil.ReadAll(r)
+	if err != nil {
+		glog.V(2).Infoln("error uncompressing data:", err)
+	}
+	return output, err
+}
diff --git a/go/operation/delete_content.go b/go/operation/delete_content.go
index ac3dfa6b2..32ad69b17 100644
--- a/go/operation/delete_content.go
+++ b/go/operation/delete_content.go
@@ -7,14 +7,17 @@ import (
 	"strings"
 	"sync"
 
+	"net/http"
+
 	"github.com/chrislusf/seaweedfs/go/security"
 	"github.com/chrislusf/seaweedfs/go/util"
 )
 
 type DeleteResult struct {
-	Fid   string `json:"fid"`
-	Size  int    `json:"size"`
-	Error string `json:"error,omitempty"`
+	Fid    string `json:"fid"`
+	Size   int    `json:"size"`
+	Status int    `json:"status"`
+	Error  string `json:"error,omitempty"`
 }
 
 func DeleteFile(master string, fileId string, jwt security.EncodedJwt) error {
@@ -45,7 +48,11 @@ func DeleteFiles(master string, fileIds []string) (*DeleteFilesResult, error) {
 	for _, fileId := range fileIds {
 		vid, _, err := ParseFileId(fileId)
 		if err != nil {
-			ret.Results = append(ret.Results, DeleteResult{Fid: vid, Error: err.Error()})
+			ret.Results = append(ret.Results, DeleteResult{
+				Fid:    vid,
+				Status: http.StatusBadRequest,
+				Error:  err.Error()},
+			)
 			continue
 		}
 		if _, ok := vid_to_fileIds[vid]; !ok {
@@ -76,6 +83,7 @@ func DeleteFiles(master string, fileIds []string) (*DeleteFilesResult, error) {
 	}
 
 	var wg sync.WaitGroup
+
 	for server, fidList := range server_to_fileIds {
 		wg.Add(1)
 		go func(server string, fidList []string) {
diff --git a/go/operation/submit.go b/go/operation/submit.go
index fec5d3801..d996d63f0 100644
--- a/go/operation/submit.go
+++ b/go/operation/submit.go
@@ -9,6 +9,8 @@ import (
 	"strconv"
 	"strings"
 
+	"net/url"
+
 	"github.com/chrislusf/seaweedfs/go/glog"
 	"github.com/chrislusf/seaweedfs/go/security"
 )
@@ -114,25 +116,44 @@ func (fi FilePart) Upload(maxMB int, master string, secret security.Secret) (ret
 	if closer, ok := fi.Reader.(io.Closer); ok {
 		defer closer.Close()
 	}
+	baseName := path.Base(fi.FileName)
 	if maxMB > 0 && fi.FileSize > int64(maxMB*1024*1024) {
 		chunkSize := int64(maxMB * 1024 * 1024)
 		chunks := fi.FileSize/chunkSize + 1
-		var fids []string
+		cm := ChunkManifest{
+			Name:   baseName,
+			Size:   fi.FileSize,
+			Mime:   fi.MimeType,
+			Chunks: make([]*ChunkInfo, 0, chunks),
+		}
+
 		for i := int64(0); i < chunks; i++ {
 			id, count, e := upload_one_chunk(
-				fi.FileName+"-"+strconv.FormatInt(i+1, 10),
+				baseName+"-"+strconv.FormatInt(i+1, 10),
 				io.LimitReader(fi.Reader, chunkSize),
 				master, fi.Replication, fi.Collection, fi.Ttl,
 				jwt)
 			if e != nil {
+				// delete all uploaded chunks
+				cm.DeleteChunks(master)
 				return 0, e
 			}
-			fids = append(fids, id)
+			cm.Chunks = append(cm.Chunks,
+				&ChunkInfo{
+					Offset: i * chunkSize,
+					Size:   int64(count),
+					Fid:    id,
+				},
+			)
 			retSize += count
 		}
-		err = upload_file_id_list(fileUrl, fi.FileName+"-list", fids, jwt)
+		err = upload_chunked_file_manifest(fileUrl, &cm, jwt)
+		if err != nil {
+			// delete all uploaded chunks
+			cm.DeleteChunks(master)
+		}
 	} else {
-		ret, e := Upload(fileUrl, fi.FileName, fi.Reader, fi.IsGzipped, fi.MimeType, jwt)
+		ret, e := Upload(fileUrl, baseName, fi.Reader, fi.IsGzipped, fi.MimeType, jwt)
 		if e != nil {
 			return 0, e
 		}
@@ -158,10 +179,17 @@ func upload_one_chunk(filename string, reader io.Reader, master,
 	return fid, uploadResult.Size, nil
 }
 
-func upload_file_id_list(fileUrl, filename string, fids []string, jwt security.EncodedJwt) error {
-	var buf bytes.Buffer
-	buf.WriteString(strings.Join(fids, "\n"))
-	glog.V(4).Info("Uploading final list ", filename, " to ", fileUrl, "...")
-	_, e := Upload(fileUrl, filename, &buf, false, "text/plain", jwt)
+func upload_chunked_file_manifest(fileUrl string, manifest *ChunkManifest, jwt security.EncodedJwt) error {
+	buf, e := manifest.Marshal()
+	if e != nil {
+		return e
+	}
+	bufReader := bytes.NewReader(buf)
+	glog.V(4).Info("Uploading chunks manifest ", manifest.Name, " to ", fileUrl, "...")
+	u, _ := url.Parse(fileUrl)
+	q := u.Query()
+	q.Set("cm", "1")
+	u.RawQuery = q.Encode()
+	_, e = Upload(u.String(), manifest.Name, bufReader, false, "application/json", jwt)
 	return e
 }