Merge branch 'master' into filerstore-tikv

4 files changed, 74 insertions, 70 deletions

diff --git a/weed/filer/filer_remote_storage.go b/weed/filer/filer_remote_storage.go
index dd8d2303c..65704e652 100644
--- a/weed/filer/filer_remote_storage.go
+++ b/weed/filer/filer_remote_storage.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
 	"github.com/chrislusf/seaweedfs/weed/remote_storage"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/golang/protobuf/proto"
@@ -21,13 +22,13 @@ const REMOTE_STORAGE_MOUNT_FILE = "mount.mapping"
 
 type FilerRemoteStorage struct {
 	rules             ptrie.Trie
-	storageNameToConf map[string]*filer_pb.RemoteConf
+	storageNameToConf map[string]*remote_pb.RemoteConf
 }
 
 func NewFilerRemoteStorage() (rs *FilerRemoteStorage) {
 	rs = &FilerRemoteStorage{
 		rules:             ptrie.New(),
-		storageNameToConf: make(map[string]*filer_pb.RemoteConf),
+		storageNameToConf: make(map[string]*remote_pb.RemoteConf),
 	}
 	return rs
 }
@@ -56,7 +57,7 @@ func (rs *FilerRemoteStorage) LoadRemoteStorageConfigurationsAndMapping(filer *F
 		if !strings.HasSuffix(entry.Name(), REMOTE_STORAGE_CONF_SUFFIX) {
 			return nil
 		}
-		conf := &filer_pb.RemoteConf{}
+		conf := &remote_pb.RemoteConf{}
 		if err := proto.Unmarshal(entry.Content, conf); err != nil {
 			return fmt.Errorf("unmarshal %s/%s: %v", DirectoryEtcRemote, entry.Name(), err)
 		}
@@ -66,7 +67,7 @@ func (rs *FilerRemoteStorage) LoadRemoteStorageConfigurationsAndMapping(filer *F
 }
 
 func (rs *FilerRemoteStorage) loadRemoteStorageMountMapping(data []byte) (err error) {
-	mappings := &filer_pb.RemoteStorageMapping{}
+	mappings := &remote_pb.RemoteStorageMapping{}
 	if err := proto.Unmarshal(data, mappings); err != nil {
 		return fmt.Errorf("unmarshal %s/%s: %v", DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, err)
 	}
@@ -76,23 +77,23 @@ func (rs *FilerRemoteStorage) loadRemoteStorageMountMapping(data []byte) (err er
 	return nil
 }
 
-func (rs *FilerRemoteStorage) mapDirectoryToRemoteStorage(dir util.FullPath, loc *filer_pb.RemoteStorageLocation) {
+func (rs *FilerRemoteStorage) mapDirectoryToRemoteStorage(dir util.FullPath, loc *remote_pb.RemoteStorageLocation) {
 	rs.rules.Put([]byte(dir+"/"), loc)
 }
 
-func (rs *FilerRemoteStorage) FindMountDirectory(p util.FullPath) (mountDir util.FullPath, remoteLocation *filer_pb.RemoteStorageLocation) {
+func (rs *FilerRemoteStorage) FindMountDirectory(p util.FullPath) (mountDir util.FullPath, remoteLocation *remote_pb.RemoteStorageLocation) {
 	rs.rules.MatchPrefix([]byte(p), func(key []byte, value interface{}) bool {
 		mountDir = util.FullPath(string(key[:len(key)-1]))
-		remoteLocation = value.(*filer_pb.RemoteStorageLocation)
+		remoteLocation = value.(*remote_pb.RemoteStorageLocation)
 		return true
 	})
 	return
 }
 
-func (rs *FilerRemoteStorage) FindRemoteStorageClient(p util.FullPath) (client remote_storage.RemoteStorageClient, remoteConf *filer_pb.RemoteConf, found bool) {
-	var storageLocation *filer_pb.RemoteStorageLocation
+func (rs *FilerRemoteStorage) FindRemoteStorageClient(p util.FullPath) (client remote_storage.RemoteStorageClient, remoteConf *remote_pb.RemoteConf, found bool) {
+	var storageLocation *remote_pb.RemoteStorageLocation
 	rs.rules.MatchPrefix([]byte(p), func(key []byte, value interface{}) bool {
-		storageLocation = value.(*filer_pb.RemoteStorageLocation)
+		storageLocation = value.(*remote_pb.RemoteStorageLocation)
 		return true
 	})
 
@@ -104,7 +105,7 @@ func (rs *FilerRemoteStorage) FindRemoteStorageClient(p util.FullPath) (client r
 	return rs.GetRemoteStorageClient(storageLocation.Name)
 }
 
-func (rs *FilerRemoteStorage) GetRemoteStorageClient(storageName string) (client remote_storage.RemoteStorageClient, remoteConf *filer_pb.RemoteConf, found bool) {
+func (rs *FilerRemoteStorage) GetRemoteStorageClient(storageName string) (client remote_storage.RemoteStorageClient, remoteConf *remote_pb.RemoteConf, found bool) {
 	remoteConf, found = rs.storageNameToConf[storageName]
 	if !found {
 		return
@@ -118,9 +119,9 @@ func (rs *FilerRemoteStorage) GetRemoteStorageClient(storageName string) (client
 	return
 }
 
-func UnmarshalRemoteStorageMappings(oldContent []byte) (mappings *filer_pb.RemoteStorageMapping, err error) {
-	mappings = &filer_pb.RemoteStorageMapping{
-		Mappings: make(map[string]*filer_pb.RemoteStorageLocation),
+func UnmarshalRemoteStorageMappings(oldContent []byte) (mappings *remote_pb.RemoteStorageMapping, err error) {
+	mappings = &remote_pb.RemoteStorageMapping{
+		Mappings: make(map[string]*remote_pb.RemoteStorageLocation),
 	}
 	if len(oldContent) > 0 {
 		if err = proto.Unmarshal(oldContent, mappings); err != nil {
@@ -130,7 +131,7 @@ func UnmarshalRemoteStorageMappings(oldContent []byte) (mappings *filer_pb.Remot
 	return
 }
 
-func AddRemoteStorageMapping(oldContent []byte, dir string, storageLocation *filer_pb.RemoteStorageLocation) (newContent []byte, err error) {
+func AddRemoteStorageMapping(oldContent []byte, dir string, storageLocation *remote_pb.RemoteStorageLocation) (newContent []byte, err error) {
 	mappings, unmarshalErr := UnmarshalRemoteStorageMappings(oldContent)
 	if unmarshalErr != nil {
 		// skip
@@ -162,7 +163,7 @@ func RemoveRemoteStorageMapping(oldContent []byte, dir string) (newContent []byt
 	return
 }
 
-func ReadMountMappings(grpcDialOption grpc.DialOption, filerAddress string) (mappings *filer_pb.RemoteStorageMapping, readErr error) {
+func ReadMountMappings(grpcDialOption grpc.DialOption, filerAddress string) (mappings *remote_pb.RemoteStorageMapping, readErr error) {
 	var oldContent []byte
 	if readErr = pb.WithFilerClient(filerAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 		oldContent, readErr = ReadInsideFiler(client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE)
@@ -179,7 +180,7 @@ func ReadMountMappings(grpcDialOption grpc.DialOption, filerAddress string) (map
 	return
 }
 
-func ReadRemoteStorageConf(grpcDialOption grpc.DialOption, filerAddress string, storageName string) (conf *filer_pb.RemoteConf, readErr error) {
+func ReadRemoteStorageConf(grpcDialOption grpc.DialOption, filerAddress string, storageName string) (conf *remote_pb.RemoteConf, readErr error) {
 	var oldContent []byte
 	if readErr = pb.WithFilerClient(filerAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 		oldContent, readErr = ReadInsideFiler(client, DirectoryEtcRemote, storageName+REMOTE_STORAGE_CONF_SUFFIX)
@@ -189,7 +190,7 @@ func ReadRemoteStorageConf(grpcDialOption grpc.DialOption, filerAddress string,
 	}
 
 	// unmarshal storage configuration
-	conf = &filer_pb.RemoteConf{}
+	conf = &remote_pb.RemoteConf{}
 	if unMarshalErr := proto.Unmarshal(oldContent, conf); unMarshalErr != nil {
 		readErr = fmt.Errorf("unmarshal %s/%s: %v", DirectoryEtcRemote, storageName+REMOTE_STORAGE_CONF_SUFFIX, unMarshalErr)
 		return
@@ -198,7 +199,7 @@ func ReadRemoteStorageConf(grpcDialOption grpc.DialOption, filerAddress string,
 	return
 }
 
-func DetectMountInfo(grpcDialOption grpc.DialOption, filerAddress string, dir string) (*filer_pb.RemoteStorageMapping, string, *filer_pb.RemoteStorageLocation, *filer_pb.RemoteConf, error) {
+func DetectMountInfo(grpcDialOption grpc.DialOption, filerAddress string, dir string) (*remote_pb.RemoteStorageMapping, string, *remote_pb.RemoteStorageLocation, *remote_pb.RemoteConf, error) {
 
 	mappings, listErr := ReadMountMappings(grpcDialOption, filerAddress)
 	if listErr != nil {
@@ -209,7 +210,7 @@ func DetectMountInfo(grpcDialOption grpc.DialOption, filerAddress string, dir st
 	}
 
 	var localMountedDir string
-	var remoteStorageMountedLocation *filer_pb.RemoteStorageLocation
+	var remoteStorageMountedLocation *remote_pb.RemoteStorageLocation
 	for k, loc := range mappings.Mappings {
 		if strings.HasPrefix(dir, k) {
 			localMountedDir, remoteStorageMountedLocation = k, loc
diff --git a/weed/filer/filer_remote_storage_test.go b/weed/filer/filer_remote_storage_test.go
index 35ffc7538..9f4d7af2f 100644
--- a/weed/filer/filer_remote_storage_test.go
+++ b/weed/filer/filer_remote_storage_test.go
@@ -1,20 +1,20 @@
 package filer
 
 import (
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
 	"github.com/stretchr/testify/assert"
 	"testing"
 )
 
 func TestFilerRemoteStorage_FindRemoteStorageClient(t *testing.T) {
-	conf := &filer_pb.RemoteConf{
+	conf := &remote_pb.RemoteConf{
 		Name: "s7",
 		Type: "s3",
 	}
 	rs := NewFilerRemoteStorage()
 	rs.storageNameToConf[conf.Name] = conf
 
-	rs.mapDirectoryToRemoteStorage("/a/b/c", &filer_pb.RemoteStorageLocation{
+	rs.mapDirectoryToRemoteStorage("/a/b/c", &remote_pb.RemoteStorageLocation{
 		Name:   "s7",
 		Bucket: "some",
 		Path:   "/dir",
diff --git a/weed/filer/read_remote.go b/weed/filer/read_remote.go
index eee536ff6..3406246a9 100644
--- a/weed/filer/read_remote.go
+++ b/weed/filer/read_remote.go
@@ -2,8 +2,8 @@ package filer
 
 import (
 	"context"
-	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
@@ -11,21 +11,8 @@ func (entry *Entry) IsInRemoteOnly() bool {
 	return len(entry.Chunks) == 0 && entry.Remote != nil && entry.Remote.RemoteSize > 0
 }
 
-func (f *Filer) ReadRemote(entry *Entry, offset int64, size int64) (data []byte, err error) {
-	client, _, found := f.RemoteStorage.GetRemoteStorageClient(entry.Remote.StorageName)
-	if !found {
-		return nil, fmt.Errorf("remote storage %v not found", entry.Remote.StorageName)
-	}
-
-	mountDir, remoteLoation := f.RemoteStorage.FindMountDirectory(entry.FullPath)
-
-	sourceLoc := MapFullPathToRemoteStorageLocation(mountDir, remoteLoation, entry.FullPath)
-
-	return client.ReadFile(sourceLoc, offset, size)
-}
-
-func MapFullPathToRemoteStorageLocation(localMountedDir util.FullPath, remoteMountedLocation *filer_pb.RemoteStorageLocation, fp util.FullPath) *filer_pb.RemoteStorageLocation {
-	remoteLocation := &filer_pb.RemoteStorageLocation{
+func MapFullPathToRemoteStorageLocation(localMountedDir util.FullPath, remoteMountedLocation *remote_pb.RemoteStorageLocation, fp util.FullPath) *remote_pb.RemoteStorageLocation {
+	remoteLocation := &remote_pb.RemoteStorageLocation{
 		Name:   remoteMountedLocation.Name,
 		Bucket: remoteMountedLocation.Bucket,
 		Path:   remoteMountedLocation.Path,
@@ -34,11 +21,11 @@ func MapFullPathToRemoteStorageLocation(localMountedDir util.FullPath, remoteMou
 	return remoteLocation
 }
 
-func MapRemoteStorageLocationPathToFullPath(localMountedDir util.FullPath, remoteMountedLocation *filer_pb.RemoteStorageLocation, remoteLocationPath string)(fp util.FullPath)  {
+func MapRemoteStorageLocationPathToFullPath(localMountedDir util.FullPath, remoteMountedLocation *remote_pb.RemoteStorageLocation, remoteLocationPath string)(fp util.FullPath)  {
 	return localMountedDir.Child(remoteLocationPath[len(remoteMountedLocation.Path):])
 }
 
-func DownloadToLocal(filerClient filer_pb.FilerClient, remoteConf *filer_pb.RemoteConf, remoteLocation *filer_pb.RemoteStorageLocation, parent util.FullPath, entry *filer_pb.Entry) error {
+func DownloadToLocal(filerClient filer_pb.FilerClient, remoteConf *remote_pb.RemoteConf, remoteLocation *remote_pb.RemoteStorageLocation, parent util.FullPath, entry *filer_pb.Entry) error {
 	return filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 		_, err := client.DownloadToLocal(context.Background(), &filer_pb.DownloadToLocalRequest{
 			Directory: string(parent),
diff --git a/weed/filer/stream.go b/weed/filer/stream.go
index 05621a5c3..ce0264cd3 100644
--- a/weed/filer/stream.go
+++ b/weed/filer/stream.go
@@ -7,6 +7,7 @@ import (
 	"math"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -131,8 +132,8 @@ type ChunkStreamReader struct {
 	logicOffset        int64
 	buffer             []byte
 	bufferOffset       int64
-	bufferPos          int
-	nextChunkViewIndex int
+	bufferLock         sync.Mutex
+	chunk              string
 	lookupFileId       wdclient.LookupFileIdFunctionType
 }
 
@@ -175,27 +176,29 @@ func NewChunkStreamReader(filerClient filer_pb.FilerClient, chunks []*filer_pb.F
 }
 
 func (c *ChunkStreamReader) ReadAt(p []byte, off int64) (n int, err error) {
+	c.bufferLock.Lock()
+	defer c.bufferLock.Unlock()
 	if err = c.prepareBufferFor(off); err != nil {
 		return
 	}
 	c.logicOffset = off
-	return c.Read(p)
+	return c.doRead(p)
 }
 
 func (c *ChunkStreamReader) Read(p []byte) (n int, err error) {
+	c.bufferLock.Lock()
+	defer c.bufferLock.Unlock()
+	return c.doRead(p)
+}
+
+func (c *ChunkStreamReader) doRead(p []byte) (n int, err error) {
+	// fmt.Printf("do read [%d,%d) at %s[%d,%d)\n", c.logicOffset, c.logicOffset+int64(len(p)), c.chunk, c.bufferOffset, c.bufferOffset+int64(len(c.buffer)))
 	for n < len(p) {
-		if c.isBufferEmpty() {
-			if c.nextChunkViewIndex >= len(c.chunkViews) {
-				return n, io.EOF
-			}
-			chunkView := c.chunkViews[c.nextChunkViewIndex]
-			if err = c.fetchChunkToBuffer(chunkView); err != nil {
-				return
-			}
-			c.nextChunkViewIndex++
+		// println("read", c.logicOffset)
+		if err = c.prepareBufferFor(c.logicOffset); err != nil {
+			return
 		}
-		t := copy(p[n:], c.buffer[c.bufferPos:])
-		c.bufferPos += t
+		t := copy(p[n:], c.buffer[c.logicOffset-c.bufferOffset:])
 		n += t
 		c.logicOffset += int64(t)
 	}
@@ -203,10 +206,12 @@ func (c *ChunkStreamReader) Read(p []byte) (n int, err error) {
 }
 
 func (c *ChunkStreamReader) isBufferEmpty() bool {
-	return len(c.buffer) <= c.bufferPos
+	return len(c.buffer) <= int(c.logicOffset - c.bufferOffset)
 }
 
 func (c *ChunkStreamReader) Seek(offset int64, whence int) (int64, error) {
+	c.bufferLock.Lock()
+	defer c.bufferLock.Unlock()
 
 	var err error
 	switch whence {
@@ -226,48 +231,59 @@ func (c *ChunkStreamReader) Seek(offset int64, whence int) (int64, error) {
 
 }
 
+func insideChunk(offset int64, chunk *ChunkView) bool {
+	return chunk.LogicOffset <= offset && offset < chunk.LogicOffset+int64(chunk.Size)
+}
+
 func (c *ChunkStreamReader) prepareBufferFor(offset int64) (err error) {
 	// stay in the same chunk
-	if !c.isBufferEmpty() {
-		if c.bufferOffset <= offset && offset < c.bufferOffset+int64(len(c.buffer)) {
-			c.bufferPos = int(offset - c.bufferOffset)
-			return nil
-		}
+	if c.bufferOffset <= offset && offset < c.bufferOffset+int64(len(c.buffer)) {
+		return nil
 	}
 
+	// fmt.Printf("fetch for offset %d\n", offset)
+
 	// need to seek to a different chunk
 	currentChunkIndex := sort.Search(len(c.chunkViews), func(i int) bool {
 		return offset < c.chunkViews[i].LogicOffset
 	})
 	if currentChunkIndex == len(c.chunkViews) {
 		// not found
-		if c.chunkViews[0].LogicOffset <= offset {
+		if insideChunk(offset, c.chunkViews[0]) {
+			// fmt.Printf("select0 chunk %d %s\n", currentChunkIndex, c.chunkViews[currentChunkIndex].FileId)
 			currentChunkIndex = 0
-		} else if c.chunkViews[len(c.chunkViews)-1].LogicOffset <= offset {
-			currentChunkIndex = len(c.chunkViews) -1
+		} else if insideChunk(offset, c.chunkViews[len(c.chunkViews)-1]) {
+			currentChunkIndex = len(c.chunkViews) - 1
+			// fmt.Printf("select last chunk %d %s\n", currentChunkIndex, c.chunkViews[currentChunkIndex].FileId)
 		} else {
 			return io.EOF
 		}
 	} else if currentChunkIndex > 0 {
-		if c.chunkViews[currentChunkIndex-1].LogicOffset <= offset {
+		if insideChunk(offset, c.chunkViews[currentChunkIndex]) {
+			// good hit
+		} else if insideChunk(offset, c.chunkViews[currentChunkIndex-1]){
 			currentChunkIndex -= 1
+			// fmt.Printf("select -1 chunk %d %s\n", currentChunkIndex, c.chunkViews[currentChunkIndex].FileId)
 		} else {
+			// glog.Fatalf("unexpected1 offset %d", offset)
 			return fmt.Errorf("unexpected1 offset %d", offset)
 		}
 	} else {
+		// glog.Fatalf("unexpected2 offset %d", offset)
 		return fmt.Errorf("unexpected2 offset %d", offset)
 	}
 
 	// positioning within the new chunk
 	chunk := c.chunkViews[currentChunkIndex]
-	if chunk.LogicOffset <= offset && offset < chunk.LogicOffset+int64(chunk.Size) {
+	if insideChunk(offset, chunk) {
 		if c.isBufferEmpty() || c.bufferOffset != chunk.LogicOffset {
 			if err = c.fetchChunkToBuffer(chunk); err != nil {
 				return
 			}
-			c.nextChunkViewIndex = currentChunkIndex + 1
 		}
-		c.bufferPos = int(offset - c.bufferOffset)
+	} else {
+		// glog.Fatalf("unexpected3 offset %d in %s [%d,%d)", offset, chunk.FileId, chunk.LogicOffset, chunk.LogicOffset+int64(chunk.Size))
+		return fmt.Errorf("unexpected3 offset %d in %s [%d,%d)", offset, chunk.FileId, chunk.LogicOffset, chunk.LogicOffset+int64(chunk.Size))
 	}
 	return
 }
@@ -298,10 +314,10 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
 		return err
 	}
 	c.buffer = buffer.Bytes()
-	c.bufferPos = 0
 	c.bufferOffset = chunkView.LogicOffset
+	c.chunk = chunkView.FileId
 
-	// glog.V(0).Infof("read %s [%d,%d)", chunkView.FileId, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size))
+	// glog.V(0).Infof("fetched %s [%d,%d)", chunkView.FileId, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size))
 
 	return nil
 }