2.05

fix https://github.com/chrislusf/seaweedfs/issues/1344

k8s/seaweedfs/Chart.yaml

@@ -1,4 +1,4 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-version: 2.04
+version: 2.05

k8s/seaweedfs/templates/filer-service.yaml

@@ -17,6 +17,12 @@ spec:
     port: {{ .Values.filer.grpcPort }}
     targetPort: {{ .Values.filer.grpcPort }}
     protocol: TCP
+{{- if .Values.filer.metricsPort }}
+  - name: "swfs-filer-metrics"
+    port: {{ .Values.filer.metricsPort }}
+    targetPort: {{ .Values.filer.metricsPort }}
+    protocol: TCP
+{{- end }}
   selector:
     app: {{ template "seaweedfs.name" . }}
     component: filer

k8s/seaweedfs/templates/filer-servicemonitor.yaml

@@ -0,0 +1,18 @@
+{{- if .Values.filer.metricsPort }}
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: {{ template "seaweedfs.name" . }}-filer
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ template "seaweedfs.name" . }}
+    component: filer
+spec:
+  endpoints:
+    - interval: 30s
+      port: swfs-filer-metrics
+      scrapeTimeout: 5s
+  selector:
+    app: {{ template "seaweedfs.name" . }}
+    component: filer
+{{- end }}

k8s/seaweedfs/templates/filer-statefulset.yaml

@@ -99,6 +99,9 @@ spec:
               {{- end }}
               filer \
               -port={{ .Values.filer.port }} \
+              {{- if .Values.filer.metricsPort }}
+              -metricsPort {{ .Values.filer.metricsPort }} \
+              {{- end }}}
               {{- if .Values.filer.redirectOnRead }}
               -redirectOnRead \
               {{- end }}

k8s/seaweedfs/templates/s3-deployment.yaml

@@ -71,6 +71,9 @@ spec:
               {{- end }}
               s3 \
               -port={{ .Values.s3.port }} \
+              {{- if .Values.s3.metricsPort }}
+              -metricsPort {{ .Values.s3.metricsPort }} \
+              {{- end }}}
               {{- if .Values.global.enableSecurity }}
               -cert.file=/usr/local/share/ca-certificates/client/tls.crt \
               -key.file=/usr/local/share/ca-certificates/client/tls.key \

k8s/seaweedfs/templates/s3-service.yaml

@@ -12,6 +12,12 @@ spec:
     port: {{ .Values.s3.port }}
     targetPort: {{ .Values.s3.port }}
     protocol: TCP
+{{- if .Values.s3.metricsPort }}
+  - name: "swfs-s3-metrics"
+    port: {{ .Values.filer.s3 }}
+    targetPort: {{ .Values.s3.metricsPort }}
+    protocol: TCP
+{{- end }}}
   selector:
     app: {{ template "seaweedfs.name" . }}
     component: s3

k8s/seaweedfs/templates/s3-servicemonitor.yaml

@@ -0,0 +1,18 @@
+{{- if .Values.s3.metricsPort }}
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: {{ template "seaweedfs.name" . }}-s3
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ template "seaweedfs.name" . }}
+    component: s3
+spec:
+  endpoints:
+    - interval: 30s
+      port: swfs-s3-metrics
+      scrapeTimeout: 5s
+  selector:
+    app: {{ template "seaweedfs.name" . }}
+    component: s3
+{{- end }}}

k8s/seaweedfs/templates/volume-service.yaml

@@ -17,6 +17,12 @@ spec:
     port: {{ .Values.volume.grpcPort }}
     targetPort: {{ .Values.volume.grpcPort }}
     protocol: TCP
-  selector:
+{{- if .Values.volume.metricsPort }}
+  - name: "swfs-volume-metrics"
+    port: {{ .Values.volume.metricsPort }}
+    targetPort: {{ .Values.volume.metricsPort }}
+    protocol: TCP
+{{- end }}}
+selector:
     app: {{ template "seaweedfs.name" . }}
     component: volume

k8s/seaweedfs/templates/volume-servicemonitor.yaml

@@ -0,0 +1,18 @@
+{{- if .Values.volume.metricsPort }}
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: {{ template "seaweedfs.name" . }}-volume
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ template "seaweedfs.name" . }}
+    component: volume
+spec:
+  endpoints:
+    - interval: 30s
+      port: swfs-volume-metrics
+      scrapeTimeout: 5s
+  selector:
+    app: {{ template "seaweedfs.name" . }}
+    component: volume
+{{- end }}}

k8s/seaweedfs/templates/volume-statefulset.yaml

@@ -76,6 +76,9 @@ spec:
                 {{- end }}
                 volume \
                 -port={{ .Values.volume.port }} \
+                {{- if .Values.volume.metricsPort }}
+                -metricsPort {{ .Values.volume.metricsPort }} \
+                {{- end }}}
                 -dir={{ .Values.volume.dir }} \
                 -max={{ .Values.volume.maxVolumes }} \
                 {{- if .Values.volume.rack }}

k8s/seaweedfs/values.yaml

@@ -4,7 +4,7 @@ global:
   registry: ""
   repository: ""
   imageName: chrislusf/seaweedfs
-  imageTag: "2.04"
+  imageTag: "2.05"
   imagePullPolicy: IfNotPresent
   imagePullSecrets: imagepullsecret
   restartPolicy: Always
@@ -121,6 +121,7 @@ volume:
   restartPolicy: null
   port: 8080
   grpcPort: 18080
+  metricsPort: 9327
   ipBind: "0.0.0.0"
   replicas: 1
   loggingOverrideLevel: null
@@ -209,6 +210,7 @@ filer:
   replicas: 1
   port: 8888
   grpcPort: 18888
+  metricsPort: 9327
   loggingOverrideLevel: null
   #  replication type is XYZ:
   # X number of replica in other data centers
@@ -313,6 +315,7 @@ s3:
   restartPolicy: null
   replicas: 1
   port: 8333
+  metricsPort: 9327
   loggingOverrideLevel: null
 
   # Suffix of the host name, {bucket}.{domainName}

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.9</version>
+    <version>1.5.0</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -65,7 +65,7 @@
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
-            <version>4.12</version>
+            <version>4.13.1</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

other/java/client/pom.xml.deploy

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.9</version>
+    <version>1.5.0</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom_debug.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.9</version>
+    <version>1.5.0</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -65,7 +65,7 @@
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
-            <version>4.12</version>
+            <version>4.13.1</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

other/java/client/src/main/proto/filer.proto

@@ -37,6 +37,9 @@ service SeaweedFiler {
     rpc LookupVolume (LookupVolumeRequest) returns (LookupVolumeResponse) {
     }
 
+    rpc CollectionList (CollectionListRequest) returns (CollectionListResponse) {
+    }
+
     rpc DeleteCollection (DeleteCollectionRequest) returns (DeleteCollectionResponse) {
     }
 
@@ -244,6 +247,16 @@ message LookupVolumeResponse {
     map<string, Locations> locations_map = 1;
 }
 
+message Collection {
+    string name = 1;
+}
+message CollectionListRequest {
+    bool include_normal_volumes = 1;
+    bool include_ec_volumes = 2;
+}
+message CollectionListResponse {
+    repeated Collection collections = 1;
+}
 message DeleteCollectionRequest {
     string collection = 1;
 }

other/java/hdfs2/dependency-reduced-pom.xml

@@ -301,7 +301,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.4.9</seaweedfs.client.version>
+    <seaweedfs.client.version>1.5.0</seaweedfs.client.version>
     <hadoop.version>2.9.2</hadoop.version>
   </properties>
 </project>

other/java/hdfs2/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.4.9</seaweedfs.client.version>
+        <seaweedfs.client.version>1.5.0</seaweedfs.client.version>
         <hadoop.version>2.9.2</hadoop.version>
     </properties>
 

other/java/hdfs3/dependency-reduced-pom.xml

@@ -309,7 +309,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.4.9</seaweedfs.client.version>
+    <seaweedfs.client.version>1.5.0</seaweedfs.client.version>
     <hadoop.version>3.1.1</hadoop.version>
   </properties>
 </project>

other/java/hdfs3/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.4.9</seaweedfs.client.version>
+        <seaweedfs.client.version>1.5.0</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 

other/java/s3copier/pom.xml

@@ -28,7 +28,7 @@
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
-      <version>3.8.1</version>
+      <version>4.13.1</version>
       <scope>test</scope>
     </dependency>
   </dependencies>

test/random_access/pom.xml

@@ -25,7 +25,7 @@
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
-            <version>4.12</version>
+            <version>4.13.1</version>
             <scope>test</scope>
         </dependency>
         <dependency>

unmaintained/repeated_vacuum/repeated_vacuum.go

@@ -32,7 +32,7 @@ func main() {
 	go func() {
 		for {
 			println("vacuum threshold", *garbageThreshold)
-			_, err := util.Get(fmt.Sprintf("http://%s/vol/vacuum?garbageThreshold=%f", *master, *garbageThreshold))
+			_, _, err := util.Get(fmt.Sprintf("http://%s/vol/vacuum?garbageThreshold=%f", *master, *garbageThreshold))
 			if err != nil {
 				log.Fatalf("vacuum: %v", err)
 			}

weed/Makefile

@@ -16,7 +16,7 @@ debug_shell:
 
 debug_mount:
 	go build -gcflags="all=-N -l"
-	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- mount -dir=~/tmp/mm
+	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- mount -dir=~/tmp/mm -cacheCapacityMB=0 -filer.path=/buckets
 
 debug_server:
 	go build -gcflags="all=-N -l"
@@ -25,3 +25,7 @@ debug_server:
 debug_volume:
 	go build -gcflags="all=-N -l"
 	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- volume -dir=/Volumes/mobile_disk/100 -port 8564 -max=30 -preStopSeconds=2
+
+debug_webdav:
+	go build -gcflags="all=-N -l"
+	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- -v=4 webdav

weed/command/benchmark.go

@@ -290,7 +290,7 @@ func readFiles(fileIdLineChan chan string, s *stat) {
 		}
 		var bytes []byte
 		for _, url := range urls {
-			bytes, err = util.Get(url)
+			bytes, _, err = util.Get(url)
 			if err == nil {
 				break
 			}

weed/command/master.go

@@ -145,7 +145,7 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
 	go func() {
 		time.Sleep(1500 * time.Millisecond)
 		if ms.Topo.RaftServer.Leader() == "" && ms.Topo.RaftServer.IsLogEmpty() && isTheFirstOne(myMasterAddress, peers) {
-			if ms.MasterClient.FindLeader(myMasterAddress) == "" {
+			if ms.MasterClient.FindLeaderFromOtherPeers(myMasterAddress) == "" {
 				raftServer.DoJoinCommand()
 			}
 		}

weed/filer/abstract_sql/abstract_sql_store_kv.go

@@ -3,6 +3,7 @@ package abstract_sql
 import (
 	"context"
 	"database/sql"
+	"encoding/base64"
 	"fmt"
 	"strings"
 
@@ -80,8 +81,8 @@ func genDirAndName(key []byte) (dirStr string, dirHash int64, name string) {
 	}
 
 	dirHash = int64(util.BytesToUint64(key[:8]))
-	dirStr = string(key[:8])
-	name = string(key[8:])
+	dirStr = base64.StdEncoding.EncodeToString(key[:8])
+	name = base64.StdEncoding.EncodeToString(key[8:])
 
 	return
 }

weed/filer/cassandra/cassandra_store_kv.go

@@ -2,6 +2,7 @@ package cassandra
 
 import (
 	"context"
+	"encoding/base64"
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/gocql/gocql"
@@ -54,8 +55,8 @@ func genDirAndName(key []byte) (dir string, name string) {
 		key = append(key, 0)
 	}
 
-	dir = string(key[:8])
-	name = string(key[8:])
+	dir = base64.StdEncoding.EncodeToString(key[:8])
+	name = base64.StdEncoding.EncodeToString(key[8:])
 
 	return
 }

weed/filer/filechunk_manifest.go

@@ -97,12 +97,16 @@ func retriedFetchChunkData(urlStrings []string, cipherKey []byte, isGzipped bool
 
 	var err error
 	var buffer bytes.Buffer
+	var shouldRetry bool
 
 	for waitTime := time.Second; waitTime < ReadWaitTime; waitTime += waitTime / 2 {
 		for _, urlString := range urlStrings {
-			err = util.ReadUrlAsStream(urlString, cipherKey, isGzipped, isFullChunk, offset, size, func(data []byte) {
+			shouldRetry, err = util.ReadUrlAsStream(urlString, cipherKey, isGzipped, isFullChunk, offset, size, func(data []byte) {
 				buffer.Write(data)
 			})
+			if !shouldRetry {
+				break
+			}
 			if err != nil {
 				glog.V(0).Infof("read %s failed, err: %v", urlString, err)
 				buffer.Reset()
@@ -110,8 +114,8 @@ func retriedFetchChunkData(urlStrings []string, cipherKey []byte, isGzipped bool
 				break
 			}
 		}
-		if err != nil {
-			glog.V(0).Infof("sleep for %v before retrying reading", waitTime)
+		if err != nil && shouldRetry {
+			glog.V(0).Infof("retry reading in %v", waitTime)
 			time.Sleep(waitTime)
 		} else {
 			break

weed/filer/reader_at.go

@@ -156,7 +156,7 @@ func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, err error) {
 		n += delta
 	}
 
-	if err == nil && offset+int64(len(p)) > c.fileSize {
+	if err == nil && offset+int64(len(p)) >= c.fileSize {
 		err = io.EOF
 	}
 	// fmt.Printf("~~~ filled %d, err: %v\n\n", n, err)

weed/filer/reader_at_test.go

@@ -66,9 +66,9 @@ func TestReaderAt(t *testing.T) {
 		chunkCache:   &mockChunkCache{},
 	}
 
-	testReadAt(t, readerAt, 0, 10, 10, nil)
+	testReadAt(t, readerAt, 0, 10, 10, io.EOF)
 	testReadAt(t, readerAt, 0, 12, 10, io.EOF)
-	testReadAt(t, readerAt, 2, 8, 8, nil)
+	testReadAt(t, readerAt, 2, 8, 8, io.EOF)
 	testReadAt(t, readerAt, 3, 6, 6, nil)
 
 }
@@ -116,7 +116,7 @@ func TestReaderAt0(t *testing.T) {
 		chunkCache:   &mockChunkCache{},
 	}
 
-	testReadAt(t, readerAt, 0, 10, 10, nil)
+	testReadAt(t, readerAt, 0, 10, 10, io.EOF)
 	testReadAt(t, readerAt, 3, 16, 7, io.EOF)
 	testReadAt(t, readerAt, 3, 5, 5, nil)
 
@@ -144,7 +144,7 @@ func TestReaderAt1(t *testing.T) {
 		chunkCache:   &mockChunkCache{},
 	}
 
-	testReadAt(t, readerAt, 0, 20, 20, nil)
+	testReadAt(t, readerAt, 0, 20, 20, io.EOF)
 	testReadAt(t, readerAt, 1, 7, 7, nil)
 	testReadAt(t, readerAt, 0, 1, 1, nil)
 	testReadAt(t, readerAt, 18, 4, 2, io.EOF)

weed/filer/stream.go

@@ -2,6 +2,7 @@ package filer
 
 import (
 	"bytes"
+	"fmt"
 	"io"
 	"math"
 	"strings"
@@ -35,10 +36,14 @@ func StreamContent(masterClient *wdclient.MasterClient, w io.Writer, chunks []*f
 
 		data, err := retriedFetchChunkData(urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size))
 		if err != nil {
-			return err
+			glog.Errorf("read chunk: %v", err)
+			return fmt.Errorf("read chunk: %v", err)
+		}
+		_, err = w.Write(data)
+		if err != nil {
+			glog.Errorf("write chunk: %v", err)
+			return fmt.Errorf("write chunk: %v", err)
 		}
-		w.Write(data)
-
 	}
 
 	return nil
@@ -174,10 +179,14 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
 		return err
 	}
 	var buffer bytes.Buffer
+	var shouldRetry bool
 	for _, urlString := range urlStrings {
-		err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
+		shouldRetry, err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
 			buffer.Write(data)
 		})
+		if !shouldRetry {
+			break
+		}
 		if err != nil {
 			glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
 			buffer.Reset()

weed/filesys/dir.go

@@ -234,7 +234,11 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
 
 	fullFilePath := util.NewFullPath(dir.FullPath(), req.Name)
 	dirPath := util.FullPath(dir.FullPath())
-	meta_cache.EnsureVisited(dir.wfs.metaCache, dir.wfs, util.FullPath(dirPath))
+	visitErr := meta_cache.EnsureVisited(dir.wfs.metaCache, dir.wfs, dirPath)
+	if visitErr != nil {
+		glog.Errorf("dir Lookup %s: %v", dirPath, visitErr)
+		return nil, fuse.EIO
+	}
 	cachedEntry, cacheErr := dir.wfs.metaCache.FindEntry(context.Background(), fullFilePath)
 	if cacheErr == filer_pb.ErrNotFound {
 		return nil, fuse.ENOENT
@@ -296,7 +300,10 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 	}
 
 	dirPath := util.FullPath(dir.FullPath())
-	meta_cache.EnsureVisited(dir.wfs.metaCache, dir.wfs, dirPath)
+	if err = meta_cache.EnsureVisited(dir.wfs.metaCache, dir.wfs, dirPath); err != nil {
+		glog.Errorf("dir ReadDirAll %s: %v", dirPath, err)
+		return nil, fuse.EIO
+	}
 	listedEntries, listErr := dir.wfs.metaCache.ListDirectoryEntries(context.Background(), util.FullPath(dir.FullPath()), "", false, int(math.MaxInt32))
 	if listErr != nil {
 		glog.Errorf("list meta cache: %v", listErr)

weed/filesys/dirty_page.go

@@ -2,154 +2,111 @@ package filesys
 
 import (
 	"bytes"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"io"
 	"sync"
 	"time"
-
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 )
 
 type ContinuousDirtyPages struct {
-	intervals   *ContinuousIntervals
-	f           *File
-	lock        sync.Mutex
-	collection  string
-	replication string
+	intervals              *ContinuousIntervals
+	f                      *File
+	writeWaitGroup         sync.WaitGroup
+	chunkSaveErrChan       chan error
+	chunkSaveErrChanClosed bool
+	lock                   sync.Mutex
+	collection             string
+	replication            string
 }
 
 func newDirtyPages(file *File) *ContinuousDirtyPages {
 	return &ContinuousDirtyPages{
-		intervals: &ContinuousIntervals{},
-		f:         file,
+		intervals:        &ContinuousIntervals{},
+		f:                file,
+		chunkSaveErrChan: make(chan error, 8),
 	}
 }
 
-var counter = int32(0)
-
-func (pages *ContinuousDirtyPages) AddPage(offset int64, data []byte) (chunks []*filer_pb.FileChunk, err error) {
+func (pages *ContinuousDirtyPages) AddPage(offset int64, data []byte) {
 
 	glog.V(4).Infof("%s AddPage [%d,%d) of %d bytes", pages.f.fullpath(), offset, offset+int64(len(data)), pages.f.entry.Attributes.FileSize)
 
 	if len(data) > int(pages.f.wfs.option.ChunkSizeLimit) {
 		// this is more than what buffer can hold.
-		return pages.flushAndSave(offset, data)
+		pages.flushAndSave(offset, data)
 	}
 
 	pages.intervals.AddInterval(data, offset)
 
-	var chunk *filer_pb.FileChunk
-	var hasSavedData bool
-
 	if pages.intervals.TotalSize() > pages.f.wfs.option.ChunkSizeLimit {
-		chunk, hasSavedData, err = pages.saveExistingLargestPageToStorage()
-		if hasSavedData {
-			chunks = append(chunks, chunk)
-		}
+		pages.saveExistingLargestPageToStorage()
 	}
 
 	return
 }
 
-func (pages *ContinuousDirtyPages) flushAndSave(offset int64, data []byte) (chunks []*filer_pb.FileChunk, err error) {
-
-	var chunk *filer_pb.FileChunk
-	var newChunks []*filer_pb.FileChunk
+func (pages *ContinuousDirtyPages) flushAndSave(offset int64, data []byte) {
 
 	// flush existing
-	if newChunks, err = pages.saveExistingPagesToStorage(); err == nil {
-		if newChunks != nil {
-			chunks = append(chunks, newChunks...)
-		}
-	} else {
-		return
-	}
+	pages.saveExistingPagesToStorage()
 
 	// flush the new page
-	if chunk, err = pages.saveToStorage(bytes.NewReader(data), offset, int64(len(data))); err == nil {
-		if chunk != nil {
-			glog.V(4).Infof("%s/%s flush big request [%d,%d) to %s", pages.f.dir.FullPath(), pages.f.Name, chunk.Offset, chunk.Offset+int64(chunk.Size), chunk.FileId)
-			chunks = append(chunks, chunk)
-		}
-	} else {
-		glog.V(0).Infof("%s/%s failed to flush2 [%d,%d): %v", pages.f.dir.FullPath(), pages.f.Name, chunk.Offset, chunk.Offset+int64(chunk.Size), err)
-		return
-	}
+	pages.saveToStorage(bytes.NewReader(data), offset, int64(len(data)))
 
 	return
 }
 
-func (pages *ContinuousDirtyPages) saveExistingPagesToStorage() (chunks []*filer_pb.FileChunk, err error) {
-
-	var hasSavedData bool
-	var chunk *filer_pb.FileChunk
-
-	for {
-
-		chunk, hasSavedData, err = pages.saveExistingLargestPageToStorage()
-		if !hasSavedData {
-			return chunks, err
-		}
-
-		if err == nil {
-			if chunk != nil {
-				chunks = append(chunks, chunk)
-			}
-		} else {
-			return
-		}
+func (pages *ContinuousDirtyPages) saveExistingPagesToStorage() {
+	for pages.saveExistingLargestPageToStorage() {
 	}
-
 }
 
-func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage() (chunk *filer_pb.FileChunk, hasSavedData bool, err error) {
+func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage() (hasSavedData bool) {
 
 	maxList := pages.intervals.RemoveLargestIntervalLinkedList()
 	if maxList == nil {
-		return nil, false, nil
+		return false
 	}
 
 	fileSize := int64(pages.f.entry.Attributes.FileSize)
-	for {
-		chunkSize := min(maxList.Size(), fileSize-maxList.Offset())
-		if chunkSize == 0 {
-			return
-		}
-		chunk, err = pages.saveToStorage(maxList.ToReader(), maxList.Offset(), chunkSize)
-		if err == nil {
-			if chunk != nil {
-				hasSavedData = true
-			}
-			glog.V(4).Infof("saveToStorage %s %s [%d,%d) of %d bytes", pages.f.fullpath(), chunk.GetFileIdString(), maxList.Offset(), maxList.Offset()+chunkSize, fileSize)
-			return
-		} else {
-			glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.f.fullpath(), maxList.Offset(), maxList.Offset()+chunkSize, err)
-			time.Sleep(5 * time.Second)
-		}
+
+	chunkSize := min(maxList.Size(), fileSize-maxList.Offset())
+	if chunkSize == 0 {
+		return false
 	}
 
+	pages.saveToStorage(maxList.ToReader(), maxList.Offset(), chunkSize)
+
+	return true
 }
 
-func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64, size int64) (*filer_pb.FileChunk, error) {
+func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64, size int64) {
 
-	dir, _ := pages.f.fullpath().DirAndName()
-
-	reader = io.LimitReader(reader, size)
-	chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(dir)(reader, pages.f.Name, offset)
-	if err != nil {
-		return nil, err
+	if pages.chunkSaveErrChanClosed {
+		pages.chunkSaveErrChan = make(chan error, 8)
+		pages.chunkSaveErrChanClosed = false
 	}
-	pages.collection, pages.replication = collection, replication
 
-	return chunk, nil
+	mtime := time.Now().UnixNano()
+	pages.writeWaitGroup.Add(1)
+	go func() {
+		defer pages.writeWaitGroup.Done()
 
-}
+		dir, _ := pages.f.fullpath().DirAndName()
 
-func maxUint64(x, y uint64) uint64 {
-	if x > y {
-		return x
-	}
-	return y
+		reader = io.LimitReader(reader, size)
+		chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(dir)(reader, pages.f.Name, offset)
+		if err != nil {
+			glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.f.fullpath(), offset, offset+size, err)
+			pages.chunkSaveErrChan <- err
+			return
+		}
+		chunk.Mtime = mtime
+		pages.collection, pages.replication = collection, replication
+		pages.f.addChunks([]*filer_pb.FileChunk{chunk})
+		pages.chunkSaveErrChan <- nil
+	}()
 }
 
 func max(x, y int64) int64 {

weed/filesys/filehandle.go

@@ -148,11 +148,7 @@ func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *f
 	fh.f.entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(fh.f.entry.Attributes.FileSize)))
 	glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data))
 
-	chunks, err := fh.dirtyPages.AddPage(req.Offset, data)
-	if err != nil {
-		glog.Errorf("%v write fh %d: [%d,%d): %v", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(len(data)), err)
-		return fuse.EIO
-	}
+	fh.dirtyPages.AddPage(req.Offset, data)
 
 	resp.Size = len(data)
 
@@ -162,12 +158,7 @@ func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *f
 		fh.f.dirtyMetadata = true
 	}
 
-	if len(chunks) > 0 {
-
-		fh.f.addChunks(chunks)
-
-		fh.f.dirtyMetadata = true
-	}
+	fh.f.dirtyMetadata = true
 
 	return nil
 }
@@ -188,10 +179,16 @@ func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) err
 	}
 
 	if fh.f.isOpen == 0 {
-		fh.doFlush(ctx, req.Header)
+		if err := fh.doFlush(ctx, req.Header); err != nil {
+			glog.Errorf("Release doFlush %s: %v", fh.f.Name, err)
+		}
 		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
 	}
 
+	// stop the goroutine
+	fh.dirtyPages.chunkSaveErrChanClosed = true
+	close(fh.dirtyPages.chunkSaveErrChan)
+
 	return nil
 }
 
@@ -204,20 +201,24 @@ func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
 }
 
 func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
-	// fflush works at fh level
+	// flush works at fh level
 	// send the data to the OS
 	glog.V(4).Infof("doFlush %s fh %d", fh.f.fullpath(), fh.handle)
 
-	chunks, err := fh.dirtyPages.saveExistingPagesToStorage()
+	fh.dirtyPages.saveExistingPagesToStorage()
+
+	var err error
+	go func() {
+		for t := range fh.dirtyPages.chunkSaveErrChan {
+			if t != nil {
+				err = t
+			}
+		}
+	}()
+	fh.dirtyPages.writeWaitGroup.Wait()
+
 	if err != nil {
-		glog.Errorf("flush %s: %v", fh.f.fullpath(), err)
-		return fuse.EIO
-	}
-
-	if len(chunks) > 0 {
-
-		fh.f.addChunks(chunks)
-		fh.f.dirtyMetadata = true
+		return err
 	}
 
 	if !fh.f.dirtyMetadata {

weed/filesys/meta_cache/meta_cache.go

@@ -2,6 +2,7 @@ package meta_cache
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"sync"
 
@@ -22,10 +23,10 @@ type MetaCache struct {
 	uidGidMapper    *UidGidMapper
 }
 
-func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper) *MetaCache {
+func NewMetaCache(dbFolder string, baseDir util.FullPath, uidGidMapper *UidGidMapper) *MetaCache {
 	return &MetaCache{
 		localStore:      openMetaStore(dbFolder),
-		visitedBoundary: bounded_tree.NewBoundedTree(),
+		visitedBoundary: bounded_tree.NewBoundedTree(baseDir),
 		uidGidMapper:    uidGidMapper,
 	}
 }
@@ -116,6 +117,10 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full
 	mc.RLock()
 	defer mc.RUnlock()
 
+	if !mc.visitedBoundary.HasVisited(dirPath) {
+		return nil, fmt.Errorf("unsynchronized dir: %v", dirPath)
+	}
+
 	entries, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit)
 	if err != nil {
 		return nil, err

weed/filesys/meta_cache/meta_cache_init.go

@@ -3,6 +3,8 @@ package meta_cache
 import (
 	"context"
 	"fmt"
+	"strings"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -10,25 +12,34 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath) {
+func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath) error {
 
-	mc.visitedBoundary.EnsureVisited(dirPath, func(path util.FullPath) (childDirectories []string, err error) {
+	return mc.visitedBoundary.EnsureVisited(dirPath, func(path util.FullPath) (childDirectories []string, err error) {
 
 		glog.V(4).Infof("ReadDirAllEntries %s ...", path)
 
-		err = filer_pb.ReadDirAllEntries(client, dirPath, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
-			entry := filer.FromPbEntry(string(dirPath), pbEntry)
-			if err := mc.doInsertEntry(context.Background(), entry); err != nil {
-				glog.V(0).Infof("read %s: %v", entry.FullPath, err)
-				return err
+		for waitTime := time.Second; waitTime < filer.ReadWaitTime; waitTime += waitTime / 2 {
+			err = filer_pb.ReadDirAllEntries(client, dirPath, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
+				entry := filer.FromPbEntry(string(dirPath), pbEntry)
+				if err := mc.doInsertEntry(context.Background(), entry); err != nil {
+					glog.V(0).Infof("read %s: %v", entry.FullPath, err)
+					return err
+				}
+				if entry.IsDirectory() {
+					childDirectories = append(childDirectories, entry.Name())
+				}
+				return nil
+			})
+			if err == nil {
+				break
 			}
-			if entry.IsDirectory() {
-				childDirectories = append(childDirectories, entry.Name())
+			if strings.Contains(err.Error(), "transport: ") {
+				glog.V(0).Infof("ReadDirAllEntries %s: %v. Retry in %v", path, err, waitTime)
+				time.Sleep(waitTime)
+				continue
 			}
-			return nil
-		})
-		if err != nil {
 			err = fmt.Errorf("list %s: %v", dirPath, err)
+			break
 		}
 		return
 	})

weed/filesys/wfs.go

@@ -92,7 +92,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		wfs.chunkCache = chunk_cache.NewTieredChunkCache(256, cacheDir, option.CacheSizeMB, 1024*1024)
 	}
 
-	wfs.metaCache = meta_cache.NewMetaCache(path.Join(cacheDir, "meta"), option.UidGidMapper)
+	wfs.metaCache = meta_cache.NewMetaCache(path.Join(cacheDir, "meta"), util.FullPath(option.FilerMountRootPath), option.UidGidMapper)
 	startTime := time.Now()
 	go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano())
 	grace.OnInterrupt(func() {

weed/pb/filer.proto

@@ -37,6 +37,9 @@ service SeaweedFiler {
     rpc LookupVolume (LookupVolumeRequest) returns (LookupVolumeResponse) {
     }
 
+    rpc CollectionList (CollectionListRequest) returns (CollectionListResponse) {
+    }
+
     rpc DeleteCollection (DeleteCollectionRequest) returns (DeleteCollectionResponse) {
     }
 
@@ -244,6 +247,16 @@ message LookupVolumeResponse {
     map<string, Locations> locations_map = 1;
 }
 
+message Collection {
+    string name = 1;
+}
+message CollectionListRequest {
+    bool include_normal_volumes = 1;
+    bool include_ec_volumes = 2;
+}
+message CollectionListResponse {
+    repeated Collection collections = 1;
+}
 message DeleteCollectionRequest {
     string collection = 1;
 }

weed/replication/repl_util/replication_utli.go

@@ -19,7 +19,7 @@ func CopyFromChunkViews(chunkViews []*filer.ChunkView, filerSource *source.Filer
 		var writeErr error
 
 		for _, fileUrl := range fileUrls {
-			err = util.ReadUrlAsStream(fileUrl+"?readDeleted=true", nil, false, chunk.IsFullChunk(), chunk.Offset, int(chunk.Size), func(data []byte) {
+			_, err = util.ReadUrlAsStream(fileUrl+"?readDeleted=true", nil, false, chunk.IsFullChunk(), chunk.Offset, int(chunk.Size), func(data []byte) {
 				writeErr = writeFunc(data)
 			})
 			if err != nil {

weed/s3api/s3api_bucket_handlers.go

@@ -58,8 +58,36 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 
 	bucket, _ := getBucketAndObject(r)
 
+	// avoid duplicated buckets
+	errCode := s3err.ErrNone
+	if err := s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+		if resp, err := client.CollectionList(context.Background(), &filer_pb.CollectionListRequest{
+			IncludeEcVolumes:     true,
+			IncludeNormalVolumes: true,
+		}); err != nil {
+			glog.Errorf("list collection: %v", err)
+			return fmt.Errorf("list collections: %v", err)
+		}else {
+			for _, c := range resp.Collections {
+				if bucket == c.Name {
+					errCode = s3err.ErrBucketAlreadyExists
+					break
+				}
+			}
+		}
+		return nil
+	}); err != nil {
+		writeErrorResponse(w, s3err.ErrInternalError, r.URL)
+		return
+	}
+	if errCode != s3err.ErrNone {
+		writeErrorResponse(w, errCode, r.URL)
+		return
+	}
+
 	// create the folder for bucket, but lazily create actual collection
 	if err := s3a.mkdir(s3a.option.BucketsPath, bucket, nil); err != nil {
+		glog.Errorf("PutBucketHandler mkdir: %v", err)
 		writeErrorResponse(w, s3err.ErrInternalError, r.URL)
 		return
 	}

weed/server/common.go

@@ -274,7 +274,7 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
 		ra := ranges[0]
 		w.Header().Set("Content-Length", strconv.FormatInt(ra.length, 10))
 		w.Header().Set("Content-Range", ra.contentRange(totalSize))
-		// w.WriteHeader(http.StatusPartialContent)
+		w.WriteHeader(http.StatusPartialContent)
 
 		err = writeFn(w, ra.start, ra.length)
 		if err != nil {
@@ -315,7 +315,7 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
 	if w.Header().Get("Content-Encoding") == "" {
 		w.Header().Set("Content-Length", strconv.FormatInt(sendSize, 10))
 	}
-	// w.WriteHeader(http.StatusPartialContent)
+	w.WriteHeader(http.StatusPartialContent)
 	if _, err := io.CopyN(w, sendContent, sendSize); err != nil {
 		http.Error(w, "Internal Error", http.StatusInternalServerError)
 		return

weed/server/filer_grpc_server.go

@@ -383,6 +383,28 @@ func (fs *FilerServer) AssignVolume(ctx context.Context, req *filer_pb.AssignVol
 	}, nil
 }
 
+func (fs *FilerServer) CollectionList(ctx context.Context, req *filer_pb.CollectionListRequest) (resp *filer_pb.CollectionListResponse, err error) {
+
+	glog.V(4).Infof("CollectionList %v", req)
+	resp = &filer_pb.CollectionListResponse{}
+
+	err = fs.filer.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
+		masterResp, err := client.CollectionList(context.Background(), &master_pb.CollectionListRequest{
+			IncludeNormalVolumes: req.IncludeNormalVolumes,
+			IncludeEcVolumes:     req.IncludeEcVolumes,
+		})
+		if err != nil {
+			return err
+		}
+		for _, c := range masterResp.Collections {
+			resp.Collections = append(resp.Collections, &filer_pb.Collection{Name: c.Name})
+		}
+		return nil
+	})
+
+	return
+}
+
 func (fs *FilerServer) DeleteCollection(ctx context.Context, req *filer_pb.DeleteCollectionRequest) (resp *filer_pb.DeleteCollectionResponse, err error) {
 
 	glog.V(4).Infof("DeleteCollection %v", req)

weed/server/webdav_server.go

@@ -489,11 +489,7 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
 	glog.V(3).Infof("WebDavFileSystem.Read %v: [%d,%d)", f.name, f.off, f.off+int64(readSize))
 	f.off += int64(readSize)
 
-	if err == io.EOF {
-		err = nil
-	}
-
-	if err != nil {
+	if err != nil && err != io.EOF {
 		glog.Errorf("file read %s: %v", f.name, err)
 	}
 

weed/shell/command_volume_server_evacuate.go

@@ -9,6 +9,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/storage/super_block"
 	"io"
+	"os"
 	"sort"
 )
 
@@ -151,6 +152,11 @@ func moveAwayOneEcVolume(commandEnv *CommandEnv, ecShardInfo *master_pb.VolumeEc
 
 		for i := 0; i < len(otherNodes); i++ {
 			emptyNode := otherNodes[i]
+			collectionPrefix := ""
+			if ecShardInfo.Collection != "" {
+				collectionPrefix = ecShardInfo.Collection + "_"
+			}
+			fmt.Fprintf(os.Stdout, "moving ec volume %s%d.%d %s => %s\n", collectionPrefix, ecShardInfo.Id, shardId, thisNode.info.Id, emptyNode.info.Id)
 			err = moveMountedShardToEcNode(commandEnv, thisNode, ecShardInfo.Collection, needle.VolumeId(ecShardInfo.Id), shardId, emptyNode, applyChange)
 			if err != nil {
 				return

weed/stats/metrics.go

@@ -77,6 +77,14 @@ var (
 			Help:      "Number of volumes or shards.",
 		}, []string{"collection", "type"})
 
+	VolumeServerReadOnlyVolumeGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "SeaweedFS",
+			Subsystem: "volumeServer",
+			Name:      "read_only_volumes",
+			Help:      "Number of read only volumes.",
+		}, []string{"collection", "type"})
+
 	VolumeServerMaxVolumeCounter = prometheus.NewGauge(
 		prometheus.GaugeOpts{
 			Namespace: "SeaweedFS",
@@ -122,6 +130,7 @@ func init() {
 	Gather.MustRegister(VolumeServerRequestHistogram)
 	Gather.MustRegister(VolumeServerVolumeCounter)
 	Gather.MustRegister(VolumeServerMaxVolumeCounter)
+	Gather.MustRegister(VolumeServerReadOnlyVolumeGauge)
 	Gather.MustRegister(VolumeServerDiskSizeGauge)
 
 	Gather.MustRegister(S3RequestCounter)

weed/storage/store.go

@@ -200,6 +200,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	maxVolumeCount := 0
 	var maxFileKey NeedleId
 	collectionVolumeSize := make(map[string]uint64)
+	collectionVolumeReadOnlyCount := make(map[string]uint8)
 	for _, location := range s.Locations {
 		var deleteVids []needle.VolumeId
 		maxVolumeCount = maxVolumeCount + location.MaxVolumeCount
@@ -219,6 +220,9 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 			}
 			fileSize, _, _ := v.FileStat()
 			collectionVolumeSize[v.Collection] += fileSize
+			if v.IsReadOnly() {
+				collectionVolumeReadOnlyCount[v.Collection] += 1
+			}
 		}
 		location.volumesLock.RUnlock()
 
@@ -243,6 +247,10 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		stats.VolumeServerDiskSizeGauge.WithLabelValues(col, "normal").Set(float64(size))
 	}
 
+	for col, count := range collectionVolumeReadOnlyCount {
+		stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues(col, "normal").Set(float64(count))
+	}
+
 	return &master_pb.Heartbeat{
 		Ip:             s.Ip,
 		Port:           uint32(s.Port),

weed/util/bounded_tree/bounded_tree.go

@@ -16,13 +16,15 @@ type Node struct {
 type BoundedTree struct {
 	root *Node
 	sync.RWMutex
+	baseDir util.FullPath
 }
 
-func NewBoundedTree() *BoundedTree {
+func NewBoundedTree(baseDir util.FullPath) *BoundedTree {
 	return &BoundedTree{
 		root: &Node{
 			Name: "/",
 		},
+		baseDir: baseDir,
 	}
 }
 
@@ -32,21 +34,29 @@ type VisitNodeFunc func(path util.FullPath) (childDirectories []string, err erro
 // No action if the directory has been visited before or does not exist.
 // A leaf node, which has no children, represents a directory not visited.
 // A non-leaf node or a non-existing node represents a directory already visited, or does not need to visit.
-func (t *BoundedTree) EnsureVisited(p util.FullPath, visitFn VisitNodeFunc) {
+func (t *BoundedTree) EnsureVisited(p util.FullPath, visitFn VisitNodeFunc) (visitErr error) {
 	t.Lock()
 	defer t.Unlock()
 
 	if t.root == nil {
 		return
 	}
+	if t.baseDir != "/" {
+		p = p[len(t.baseDir):]
+	}
 	components := p.Split()
 	// fmt.Printf("components %v %d\n", components, len(components))
-	if canDelete := t.ensureVisited(t.root, util.FullPath("/"), components, 0, visitFn); canDelete {
+	canDelete, err := t.ensureVisited(t.root, t.baseDir, components, 0, visitFn)
+	if err != nil {
+		return err
+	}
+	if canDelete {
 		t.root = nil
 	}
+	return nil
 }
 
-func (t *BoundedTree) ensureVisited(n *Node, currentPath util.FullPath, components []string, i int, visitFn VisitNodeFunc) (canDeleteNode bool) {
+func (t *BoundedTree) ensureVisited(n *Node, currentPath util.FullPath, components []string, i int, visitFn VisitNodeFunc) (canDeleteNode bool, visitErr error) {
 
 	// println("ensureVisited", currentPath, i)
 
@@ -60,15 +70,20 @@ func (t *BoundedTree) ensureVisited(n *Node, currentPath util.FullPath, componen
 	} else {
 		// fmt.Printf("ensure %v\n", currentPath)
 
-		children, err := visitFn(currentPath)
+		filerPath := currentPath
+		if t.baseDir != "/" {
+			filerPath = t.baseDir + filerPath
+		}
+
+		children, err := visitFn(filerPath)
 		if err != nil {
 			glog.V(0).Infof("failed to visit %s: %v", currentPath, err)
-			return
+			return false, err
 		}
 
 		if len(children) == 0 {
 			// fmt.Printf("  canDelete %v without children\n", currentPath)
-			return true
+			return true, nil
 		}
 
 		n.Children = make(map[string]*Node)
@@ -93,19 +108,22 @@ func (t *BoundedTree) ensureVisited(n *Node, currentPath util.FullPath, componen
 	}
 
 	// fmt.Printf("  ensureVisited %v %v\n", currentPath, toVisitNode.Name)
-
-	if canDelete := t.ensureVisited(toVisitNode, currentPath.Child(components[i]), components, i+1, visitFn); canDelete {
+	canDelete, childVisitErr := t.ensureVisited(toVisitNode, currentPath.Child(components[i]), components, i+1, visitFn)
+	if childVisitErr != nil {
+		return false, childVisitErr
+	}
+	if canDelete {
 
 		// fmt.Printf("  delete %v %v\n", currentPath, components[i])
 		delete(n.Children, components[i])
 
 		if len(n.Children) == 0 {
 			// fmt.Printf("  canDelete %v\n", currentPath)
-			return true
+			return true, nil
 		}
 	}
 
-	return false
+	return false, nil
 
 }
 

weed/util/bounded_tree/bounded_tree_test.go

@@ -52,7 +52,7 @@ func TestBoundedTree(t *testing.T) {
 	// g
 	// h
 
-	tree := NewBoundedTree()
+	tree := NewBoundedTree(util.FullPath("/"))
 
 	tree.EnsureVisited(util.FullPath("/a/b/c"), visitFn)
 
@@ -100,7 +100,7 @@ func TestEmptyBoundedTree(t *testing.T) {
 	// g
 	// h
 
-	tree := NewBoundedTree()
+	tree := NewBoundedTree(util.FullPath("/"))
 
 	visitFn := func(path util.FullPath) (childDirectories []string, err error) {
 		fmt.Printf("  visit %v ...\n", path)

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 04)
+	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 05)
 	COMMIT  = ""
 )
 

weed/util/http_util.go

@@ -67,14 +67,14 @@ func Post(url string, values url.Values) ([]byte, error) {
 
 //	github.com/chrislusf/seaweedfs/unmaintained/repeated_vacuum/repeated_vacuum.go
 //	may need increasing http.Client.Timeout
-func Get(url string) ([]byte, error) {
+func Get(url string) ([]byte, bool, error) {
 
 	request, err := http.NewRequest("GET", url, nil)
 	request.Header.Add("Accept-Encoding", "gzip")
 
 	response, err := client.Do(request)
 	if err != nil {
-		return nil, err
+		return nil, true, err
 	}
 	defer response.Body.Close()
 
@@ -89,12 +89,13 @@ func Get(url string) ([]byte, error) {
 
 	b, err := ioutil.ReadAll(reader)
 	if response.StatusCode >= 400 {
-		return nil, fmt.Errorf("%s: %s", url, response.Status)
+		retryable := response.StatusCode >= 500
+		return nil, retryable, fmt.Errorf("%s: %s", url, response.Status)
 	}
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
-	return b, nil
+	return b, false, nil
 }
 
 func Head(url string) (http.Header, error) {
@@ -207,7 +208,7 @@ func ReadUrl(fileUrl string, cipherKey []byte, isContentCompressed bool, isFullC
 
 	if cipherKey != nil {
 		var n int
-		err := readEncryptedUrl(fileUrl, cipherKey, isContentCompressed, isFullChunk, offset, size, func(data []byte) {
+		_, err := readEncryptedUrl(fileUrl, cipherKey, isContentCompressed, isFullChunk, offset, size, func(data []byte) {
 			n = copy(buf, data)
 		})
 		return int64(n), err
@@ -272,7 +273,7 @@ func ReadUrl(fileUrl string, cipherKey []byte, isContentCompressed bool, isFullC
 	return n, err
 }
 
-func ReadUrlAsStream(fileUrl string, cipherKey []byte, isContentGzipped bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) error {
+func ReadUrlAsStream(fileUrl string, cipherKey []byte, isContentGzipped bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) (retryable bool, err error) {
 
 	if cipherKey != nil {
 		return readEncryptedUrl(fileUrl, cipherKey, isContentGzipped, isFullChunk, offset, size, fn)
@@ -280,7 +281,7 @@ func ReadUrlAsStream(fileUrl string, cipherKey []byte, isContentGzipped bool, is
 
 	req, err := http.NewRequest("GET", fileUrl, nil)
 	if err != nil {
-		return err
+		return false, err
 	}
 
 	if isFullChunk {
@@ -291,11 +292,12 @@ func ReadUrlAsStream(fileUrl string, cipherKey []byte, isContentGzipped bool, is
 
 	r, err := client.Do(req)
 	if err != nil {
-		return err
+		return true, err
 	}
 	defer CloseResponse(r)
 	if r.StatusCode >= 400 {
-		return fmt.Errorf("%s: %s", fileUrl, r.Status)
+		retryable = r.StatusCode >= 500
+		return retryable, fmt.Errorf("%s: %s", fileUrl, r.Status)
 	}
 
 	var reader io.ReadCloser
@@ -317,23 +319,23 @@ func ReadUrlAsStream(fileUrl string, cipherKey []byte, isContentGzipped bool, is
 		m, err = reader.Read(buf)
 		fn(buf[:m])
 		if err == io.EOF {
-			return nil
+			return false, nil
 		}
 		if err != nil {
-			return err
+			return false, err
 		}
 	}
 
 }
 
-func readEncryptedUrl(fileUrl string, cipherKey []byte, isContentCompressed bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) error {
-	encryptedData, err := Get(fileUrl)
+func readEncryptedUrl(fileUrl string, cipherKey []byte, isContentCompressed bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) (bool, error) {
+	encryptedData, retryable, err := Get(fileUrl)
 	if err != nil {
-		return fmt.Errorf("fetch %s: %v", fileUrl, err)
+		return retryable, fmt.Errorf("fetch %s: %v", fileUrl, err)
 	}
 	decryptedData, err := Decrypt(encryptedData, CipherKey(cipherKey))
 	if err != nil {
-		return fmt.Errorf("decrypt %s: %v", fileUrl, err)
+		return false, fmt.Errorf("decrypt %s: %v", fileUrl, err)
 	}
 	if isContentCompressed {
 		decryptedData, err = DecompressData(decryptedData)
@@ -342,14 +344,14 @@ func readEncryptedUrl(fileUrl string, cipherKey []byte, isContentCompressed bool
 		}
 	}
 	if len(decryptedData) < int(offset)+size {
-		return fmt.Errorf("read decrypted %s size %d [%d, %d)", fileUrl, len(decryptedData), offset, int(offset)+size)
+		return false, fmt.Errorf("read decrypted %s size %d [%d, %d)", fileUrl, len(decryptedData), offset, int(offset)+size)
 	}
 	if isFullChunk {
 		fn(decryptedData)
 	} else {
 		fn(decryptedData[int(offset) : int(offset)+size])
 	}
-	return nil
+	return false, nil
 }
 
 func ReadUrlAsReaderCloser(fileUrl string, rangeHeader string) (io.ReadCloser, error) {

weed/util/net_timeout.go

@@ -54,7 +54,8 @@ func (c *Conn) Read(b []byte) (count int, e error) {
 
 func (c *Conn) Write(b []byte) (count int, e error) {
 	if c.WriteTimeout != 0 {
-		err := c.Conn.SetWriteDeadline(time.Now().Add(c.WriteTimeout))
+		// minimum 4KB/s
+		err := c.Conn.SetWriteDeadline(time.Now().Add(c.WriteTimeout * time.Duration(len(b)/40000+1)))
 		if err != nil {
 			return 0, err
 		}

weed/wdclient/masterclient.go

@@ -52,7 +52,7 @@ func (mc *MasterClient) KeepConnectedToMaster() {
 	}
 }
 
-func (mc *MasterClient) FindLeader(myMasterAddress string) (leader string) {
+func (mc *MasterClient) FindLeaderFromOtherPeers(myMasterAddress string) (leader string) {
 	for _, master := range mc.masters {
 		if master == myMasterAddress {
 			continue