Merge branch 'master' of https://github.com/chrislusf/seaweedfs

add gocql timeout setting

go.mod

@@ -100,7 +100,7 @@ require (
 	github.com/prometheus/procfs v0.6.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
-	github.com/seaweedfs/fuse v1.2.1
+	github.com/seaweedfs/fuse v1.2.2
 	github.com/seaweedfs/goexif v1.0.2
 	github.com/sirupsen/logrus v1.6.0 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
@@ -136,7 +136,7 @@ require (
 	golang.org/x/image v0.0.0-20200119044424-58c23975cae1
 	golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d
 	golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f // indirect
-	golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e
+	golang.org/x/sys v0.0.0-20220111092808-5a964db01320
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.6-0.20210726203631-07bc1bf47fb2
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect

go.sum

@@ -732,6 +732,8 @@ github.com/seaweedfs/fuse v1.2.0 h1:m4Ar3I0TY/Tzyb80lfJiXUKFcMBHqA5e/cZwxQtilZo=
 github.com/seaweedfs/fuse v1.2.0/go.mod h1:iwbDQv5BZACY54r6AO/6xsLNuMaYcBKSkLTZVfmK594=
 github.com/seaweedfs/fuse v1.2.1 h1:920YH3+PnIqpNLo2lsZDTUJHUPLXnxgQGEXoVb/yMno=
 github.com/seaweedfs/fuse v1.2.1/go.mod h1:iwbDQv5BZACY54r6AO/6xsLNuMaYcBKSkLTZVfmK594=
+github.com/seaweedfs/fuse v1.2.2 h1:01l8OjIdyATRNqVc/gDPgFobuC8ubQF3hRKOPColROw=
+github.com/seaweedfs/fuse v1.2.2/go.mod h1:iwbDQv5BZACY54r6AO/6xsLNuMaYcBKSkLTZVfmK594=
 github.com/seaweedfs/goexif v1.0.2 h1:p+rTXYdQ2mgxd+1JaTrQ9N8DvYuw9UH9xgYmJ+Bb29E=
 github.com/seaweedfs/goexif v1.0.2/go.mod h1:MrKs5LK0HXdffrdCZrW3OIMegL2xXpC6ThLyXMyjdrk=
 github.com/secsy/goftp v0.0.0-20190720192957-f31499d7c79a h1:C6IhVTxNkhlb0tlCB6JfHOUv1f0xHPK7V8X4HlJZEJw=
@@ -1114,6 +1116,8 @@ golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365 h1:6wSTsvPddg9gc/mVEEyk9oOAo
 golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220111092808-5a964db01320 h1:0jf+tOCoZ3LyutmCOWpVni1chK4VfFLhRsDK7MhqGRY=
+golang.org/x/sys v0.0.0-20220111092808-5a964db01320/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

k8s/helm_charts2/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "2.84"
-version: "2.84"
+appVersion: "2.86"
+version: "2.86"

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>2.84</version>
+    <version>2.85</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -14,7 +14,7 @@
     </parent>
 
     <properties>
-        <protobuf.version>3.9.1</protobuf.version>
+        <protobuf.version>3.16.1</protobuf.version>
         <!-- follow https://github.com/grpc/grpc-java -->
         <grpc.version>1.23.0</grpc.version>
         <guava.version>30.0-jre</guava.version>

other/java/client/pom.xml.deploy

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>2.84</version>
+    <version>2.85</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>2.84</version>
+    <version>2.85</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/examples/pom.xml

@@ -11,13 +11,13 @@
         <dependency>
             <groupId>com.github.chrislusf</groupId>
             <artifactId>seaweedfs-client</artifactId>
-            <version>2.84</version>
+            <version>2.85</version>
             <scope>compile</scope>
         </dependency>
         <dependency>
             <groupId>com.github.chrislusf</groupId>
             <artifactId>seaweedfs-hadoop2-client</artifactId>
-            <version>2.84</version>
+            <version>2.85</version>
             <scope>compile</scope>
         </dependency>
         <dependency>

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

@@ -301,7 +301,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>2.84</seaweedfs.client.version>
+    <seaweedfs.client.version>2.85</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>2.84</seaweedfs.client.version>
+        <seaweedfs.client.version>2.85</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>2.84</seaweedfs.client.version>
+    <seaweedfs.client.version>2.85</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>2.84</seaweedfs.client.version>
+        <seaweedfs.client.version>2.85</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 

weed/Makefile

@@ -19,15 +19,15 @@ debug_shell:
 
 debug_mount:
 	go build -gcflags="all=-N -l"
-	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- -v=4 mount -dir=~/tmp/mm -cacheCapacityMB=0 -filer.path=/
+	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- -v=4 mount -dir=~/tmp/mm -cacheCapacityMB=0 -filer.path=/ -umask=000
 
 debug_server:
 	go build -gcflags="all=-N -l"
-	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- server -dir=/Volumes/mobile_disk/99 -filer -volume.port=8343 -s3 -volume.max=0 -master.volumeSizeLimitMB=1024 -volume.preStopSeconds=1
+	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- server -dir=~/tmp/99 -filer -volume.port=8343 -s3 -volume.max=0 -master.volumeSizeLimitMB=1024 -volume.preStopSeconds=1
 
 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
+	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- volume -dir=~/tmp/100 -port 8564 -max=30 -preStopSeconds=2
 
 debug_webdav:
 	go build -gcflags="all=-N -l"
@@ -52,4 +52,3 @@ debug_filer_remote_sync_buckets:
 debug_master_follower:
 	go build -gcflags="all=-N -l"
 	dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec weed -- -v=4 master.follower
-

weed/command/filer_copy.go

@@ -482,7 +482,8 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File,
 				})
 			})
 			if err != nil {
-				fmt.Printf("Failed to assign from %v: %v\n", worker.options.masters, err)
+				uploadError = fmt.Errorf("Failed to assign from %v: %v\n", worker.options.masters, err)
+				return
 			}
 
 			targetUrl := "http://" + assignResult.Location.Url + "/" + assignResult.FileId

weed/command/filer_sync.go

@@ -169,8 +169,11 @@ func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption,
 		return persistEventFn(resp)
 	}
 
+	var lastLogTsNs = time.Now().Nanosecond()
 	processEventFnWithOffset := pb.AddOffsetFunc(processEventFn, 3*time.Second, func(counter int64, lastTsNs int64) error {
-		glog.V(0).Infof("sync %s to %s progressed to %v %0.2f/sec", sourceFiler, targetFiler, time.Unix(0, lastTsNs), float64(counter)/float64(3))
+		now := time.Now().Nanosecond()
+		glog.V(0).Infof("sync %s to %s progressed to %v %0.2f/sec", sourceFiler, targetFiler, time.Unix(0, lastTsNs), float64(counter)/(float64(now-lastLogTsNs)/1e9))
+		lastLogTsNs = now
 		return setOffset(grpcDialOption, targetFiler, SyncKeyPrefix, sourceFilerSignature, lastTsNs)
 	})
 

weed/command/fix.go

@@ -1,9 +1,12 @@
 package command
 
 import (
+	"fmt"
+	"io/fs"
 	"os"
 	"path"
 	"strconv"
+	"strings"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/storage"
@@ -19,17 +22,15 @@ func init() {
 }
 
 var cmdFix = &Command{
-	UsageLine: "fix -dir=/tmp -volumeId=234",
-	Short:     "run weed tool fix on index file if corrupted",
-	Long: `Fix runs the SeaweedFS fix command to re-create the index .idx file.
-
+	UsageLine: "fix [-volumeId=234] [-collection=bigData] /tmp",
+	Short:     "run weed tool fix on files or whole folders to recreate index file(s) if corrupted",
+	Long: `Fix runs the SeaweedFS fix command on dat files or whole folders to re-create the index .idx file.
   `,
 }
 
 var (
-	fixVolumePath       = cmdFix.Flag.String("dir", ".", "data directory to store files")
-	fixVolumeCollection = cmdFix.Flag.String("collection", "", "the volume collection name")
-	fixVolumeId         = cmdFix.Flag.Int("volumeId", -1, "a volume id. The volume should already exist in the dir. The volume index file should not exist.")
+	fixVolumeCollection = cmdFix.Flag.String("collection", "", "an optional volume collection name, if specified only it will be processed")
+	fixVolumeId         = cmdFix.Flag.Int64("volumeId", 0, "an optional volume id, if not 0 (default) only it will be processed")
 )
 
 type VolumeFileScanner4Fix struct {
@@ -59,26 +60,68 @@ func (scanner *VolumeFileScanner4Fix) VisitNeedle(n *needle.Needle, offset int64
 }
 
 func runFix(cmd *Command, args []string) bool {
+	for _, arg := range args {
+		basePath, f := path.Split(util.ResolvePath(arg))
 
-	if *fixVolumeId == -1 {
-		return false
-	}
+		files := []fs.DirEntry{}
+		if f == "" {
+			fileInfo, err := os.ReadDir(basePath)
+			if err != nil {
+				fmt.Println(err)
+				return false
+			}
+			files = fileInfo
+		} else {
+			fileInfo, err := os.Stat(basePath + f)
+			if err != nil {
+				fmt.Println(err)
+				return false
+			}
+			files = []fs.DirEntry{fs.FileInfoToDirEntry(fileInfo)}
+		}
 
-	baseFileName := strconv.Itoa(*fixVolumeId)
-	if *fixVolumeCollection != "" {
-		baseFileName = *fixVolumeCollection + "_" + baseFileName
+		for _, file := range files {
+			if !strings.HasSuffix(file.Name(), ".dat") {
+				continue
+			}
+			if *fixVolumeCollection != "" {
+				if !strings.HasPrefix(file.Name(), *fixVolumeCollection+"_") {
+					continue
+				}
+			}
+			baseFileName := file.Name()[:len(file.Name())-4]
+			collection, volumeIdStr := "", baseFileName
+			if sepIndex := strings.LastIndex(baseFileName, "_"); sepIndex > 0 {
+				collection = baseFileName[:sepIndex]
+				volumeIdStr = baseFileName[sepIndex+1:]
+			}
+			volumeId, parseErr := strconv.ParseInt(volumeIdStr, 10, 64)
+			if parseErr != nil {
+				fmt.Printf("Failed to parse volume id from %s: %v\n", baseFileName, parseErr)
+				return false
+			}
+			if *fixVolumeId != 0 && *fixVolumeId != volumeId {
+				continue
+			}
+			doFixOneVolume(basePath, baseFileName, collection, volumeId)
+		}
 	}
-	indexFileName := path.Join(util.ResolvePath(*fixVolumePath), baseFileName+".idx")
+	return true
+}
+
+func doFixOneVolume(basepath string, baseFileName string, collection string, volumeId int64) {
+
+	indexFileName := path.Join(basepath, baseFileName+".idx")
 
 	nm := needle_map.NewMemDb()
 	defer nm.Close()
 
-	vid := needle.VolumeId(*fixVolumeId)
+	vid := needle.VolumeId(volumeId)
 	scanner := &VolumeFileScanner4Fix{
 		nm: nm,
 	}
 
-	if err := storage.ScanVolumeFile(util.ResolvePath(*fixVolumePath), *fixVolumeCollection, vid, storage.NeedleMapInMemory, scanner); err != nil {
+	if err := storage.ScanVolumeFile(basepath, collection, vid, storage.NeedleMapInMemory, scanner); err != nil {
 		glog.Fatalf("scan .dat File: %v", err)
 		os.Remove(indexFileName)
 	}
@@ -87,6 +130,4 @@ func runFix(cmd *Command, args []string) bool {
 		glog.Fatalf("save to .idx File: %v", err)
 		os.Remove(indexFileName)
 	}
-
-	return true
 }

weed/command/mount.go

@@ -26,6 +26,8 @@ type MountOptions struct {
 	uidMap             *string
 	gidMap             *string
 	readOnly           *bool
+	debug              *bool
+	debugPort          *int
 }
 
 var (
@@ -57,6 +59,8 @@ func init() {
 	mountOptions.uidMap = cmdMount.Flag.String("map.uid", "", "map local uid to uid on filer, comma-separated <local_uid>:<filer_uid>")
 	mountOptions.gidMap = cmdMount.Flag.String("map.gid", "", "map local gid to gid on filer, comma-separated <local_gid>:<filer_gid>")
 	mountOptions.readOnly = cmdMount.Flag.Bool("readOnly", false, "read only")
+	mountOptions.debug = cmdMount.Flag.Bool("debug", false, "serves runtime profiling data, e.g., http://localhost:<debug.port>/debug/pprof/goroutine?debug=2")
+	mountOptions.debugPort = cmdMount.Flag.Int("debug.port", 6061, "http port for debugging")
 
 	mountCpuProfile = cmdMount.Flag.String("cpuprofile", "", "cpu profile output file")
 	mountMemProfile = cmdMount.Flag.String("memprofile", "", "memory profile output file")

weed/command/mount_std.go

@@ -6,6 +6,7 @@ package command
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"os"
 	"os/user"
 	"path"
@@ -34,6 +35,10 @@ import (
 
 func runMount(cmd *Command, args []string) bool {
 
+	if *mountOptions.debug {
+		go http.ListenAndServe(fmt.Sprintf(":%d", *mountOptions.debugPort), nil)
+	}
+
 	grace.SetupProfiling(*mountCpuProfile, *mountMemProfile)
 	if *mountReadRetryTime < time.Second {
 		*mountReadRetryTime = time.Second
@@ -171,6 +176,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		fuse.NoAppleXattr(),
 		fuse.ExclCreate(),
 		fuse.DaemonTimeout("3600"),
+		fuse.AllowDev(),
 		fuse.AllowSUID(),
 		fuse.DefaultPermissions(),
 		fuse.MaxReadahead(1024 * 512),

weed/command/scaffold/filer.toml

@@ -153,6 +153,8 @@ password = ""
 superLargeDirectories = []
 # Name of the datacenter local to this filer, used as host selection fallback.
 localDC = ""
+# Gocql connection timeout, default: 600ms
+connection_timeout_millisecond = 600
 
 [hbase]
 enabled = false

weed/command/scaffold/master.toml

@@ -19,9 +19,6 @@ scripts = """
 """
 sleep_minutes = 17          # sleep minutes between each script execution
 
-[master.filer]
-default = "localhost:8888"    # used by maintenance scripts if the scripts needs to use fs related commands
-
 
 [master.sequencer]
 type = "raft"     # Choose [raft|snowflake] type for storing the file id sequence

weed/command/server.go

@@ -28,6 +28,7 @@ var (
 	masterOptions    MasterOptions
 	filerOptions     FilerOptions
 	s3Options        S3Options
+	iamOptions       IamOptions
 	webdavOptions    WebDavOption
 	msgBrokerOptions MessageBrokerOptions
 )
@@ -61,7 +62,7 @@ var (
 	serverWhiteListOption     = cmdServer.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
 	serverDisableHttp         = cmdServer.Flag.Bool("disableHttp", false, "disable http requests, only gRPC operations are allowed.")
 	volumeDataFolders         = cmdServer.Flag.String("dir", os.TempDir(), "directories to store data files. dir[,dir]...")
-	volumeMaxDataVolumeCounts = cmdServer.Flag.String("volume.max", "8", "maximum numbers of volumes, count[,count]... If set to zero, the limit will be auto configured.")
+	volumeMaxDataVolumeCounts = cmdServer.Flag.String("volume.max", "8", "maximum numbers of volumes, count[,count]... If set to zero, the limit will be auto configured as free disk space divided by volume size.")
 	volumeMinFreeSpacePercent = cmdServer.Flag.String("volume.minFreeSpacePercent", "1", "minimum free disk space (default to 1%). Low disk space will mark all volumes as ReadOnly (deprecated, use minFreeSpace instead).")
 	volumeMinFreeSpace        = cmdServer.Flag.String("volume.minFreeSpace", "", "min free disk space (value<=100 as percentage like 1, other as human readable bytes, like 10GiB). Low disk space will mark all volumes as ReadOnly.")
 	serverMetricsHttpPort     = cmdServer.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
@@ -71,6 +72,7 @@ var (
 	isStartingVolumeServer = cmdServer.Flag.Bool("volume", true, "whether to start volume server")
 	isStartingFiler        = cmdServer.Flag.Bool("filer", false, "whether to start filer")
 	isStartingS3           = cmdServer.Flag.Bool("s3", false, "whether to start S3 gateway")
+	isStartingIam          = cmdServer.Flag.Bool("iam", false, "whether to start IAM service")
 	isStartingWebDav       = cmdServer.Flag.Bool("webdav", false, "whether to start WebDAV gateway")
 	isStartingMsgBroker    = cmdServer.Flag.Bool("msgBroker", false, "whether to start message broker")
 
@@ -134,6 +136,8 @@ func init() {
 	s3Options.auditLogConfig = cmdServer.Flag.String("s3.auditLogConfig", "", "path to the audit log config file")
 	s3Options.allowEmptyFolder = cmdServer.Flag.Bool("s3.allowEmptyFolder", true, "allow empty folders")
 
+	iamOptions.port = cmdServer.Flag.Int("iam.port", 8111, "iam server http listen port")
+
 	webdavOptions.port = cmdServer.Flag.Int("webdav.port", 7333, "webdav server http listen port")
 	webdavOptions.collection = cmdServer.Flag.String("webdav.collection", "", "collection to create the files")
 	webdavOptions.replication = cmdServer.Flag.String("webdav.replication", "", "replication to create the files")
@@ -161,6 +165,9 @@ func runServer(cmd *Command, args []string) bool {
 	if *isStartingS3 {
 		*isStartingFiler = true
 	}
+	if *isStartingIam {
+		*isStartingFiler = true
+	}
 	if *isStartingWebDav {
 		*isStartingFiler = true
 	}
@@ -201,6 +208,7 @@ func runServer(cmd *Command, args []string) bool {
 
 	filerAddress := string(pb.NewServerAddress(*serverIp, *filerOptions.port, *filerOptions.portGrpc))
 	s3Options.filer = &filerAddress
+	iamOptions.filer = &filerAddress
 	webdavOptions.filer = &filerAddress
 	msgBrokerOptions.filer = &filerAddress
 
@@ -227,25 +235,27 @@ func runServer(cmd *Command, args []string) bool {
 	if *isStartingFiler {
 		go func() {
 			time.Sleep(1 * time.Second)
-
 			filerOptions.startFiler()
-
 		}()
 	}
 
 	if *isStartingS3 {
 		go func() {
 			time.Sleep(2 * time.Second)
-
 			s3Options.startS3Server()
+		}()
+	}
 
+	if *isStartingIam {
+		go func() {
+			time.Sleep(2 * time.Second)
+			iamOptions.startIamServer()
 		}()
 	}
 
 	if *isStartingWebDav {
 		go func() {
 			time.Sleep(2 * time.Second)
-
 			webdavOptions.startWebDav()
 
 		}()

weed/command/volume.go

@@ -107,7 +107,7 @@ var cmdVolume = &Command{
 
 var (
 	volumeFolders         = cmdVolume.Flag.String("dir", os.TempDir(), "directories to store data files. dir[,dir]...")
-	maxVolumeCounts       = cmdVolume.Flag.String("max", "8", "maximum numbers of volumes, count[,count]... If set to zero, the limit will be auto configured.")
+	maxVolumeCounts       = cmdVolume.Flag.String("max", "8", "maximum numbers of volumes, count[,count]... If set to zero, the limit will be auto configured as free disk space divided by volume size.")
 	volumeWhiteListOption = cmdVolume.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
 	minFreeSpacePercent   = cmdVolume.Flag.String("minFreeSpacePercent", "1", "minimum free disk space (default to 1%). Low disk space will mark all volumes as ReadOnly (deprecated, use minFreeSpace instead).")
 	minFreeSpace          = cmdVolume.Flag.String("minFreeSpace", "", "min free disk space (value<=100 as percentage like 1, other as human readable bytes, like 10GiB). Low disk space will mark all volumes as ReadOnly.")

weed/filer/cassandra/cassandra_store.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/gocql/gocql"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -33,6 +34,7 @@ func (store *CassandraStore) Initialize(configuration util.Configuration, prefix
 		configuration.GetString(prefix+"password"),
 		configuration.GetStringSlice(prefix+"superLargeDirectories"),
 		configuration.GetString(prefix+"localDC"),
+		configuration.GetInt(prefix+"connection_timeout_millisecond"),
 	)
 }
 
@@ -41,12 +43,14 @@ func (store *CassandraStore) isSuperLargeDirectory(dir string) (dirHash string,
 	return
 }
 
-func (store *CassandraStore) initialize(keyspace string, hosts []string, username string, password string, superLargeDirectories []string, localDC string) (err error) {
+func (store *CassandraStore) initialize(keyspace string, hosts []string, username string, password string, superLargeDirectories []string, localDC string, timeout int) (err error) {
 	store.cluster = gocql.NewCluster(hosts...)
 	if username != "" && password != "" {
 		store.cluster.Authenticator = gocql.PasswordAuthenticator{Username: username, Password: password}
 	}
 	store.cluster.Keyspace = keyspace
+	store.cluster.Timeout = time.Duration(timeout) * time.Millisecond
+	glog.V(0).Infof("timeout = %d", timeout)
 	fallback := gocql.RoundRobinHostPolicy()
 	if localDC != "" {
 		fallback = gocql.DCAwareRoundRobinPolicy(localDC)

weed/filer/filerstore.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"io"
 )
 
 var (
@@ -45,3 +46,7 @@ type BucketAware interface {
 	OnBucketDeletion(bucket string)
 	CanDropWholeBucket() bool
 }
+
+type Debuggable interface {
+	Debug(writer io.Writer)
+}

weed/filer/filerstore_wrapper.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/viant/ptrie"
+	"io"
 	"math"
 	"strings"
 	"time"
@@ -15,6 +16,7 @@ import (
 
 var (
 	_ = VirtualFilerStore(&FilerStoreWrapper{})
+	_ = Debuggable(&FilerStoreWrapper{})
 )
 
 type VirtualFilerStore interface {
@@ -333,3 +335,9 @@ func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []by
 func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
 	return fsw.getDefaultStore().KvDelete(ctx, key)
 }
+
+func (fsw *FilerStoreWrapper) Debug(writer io.Writer) {
+	if debuggable, ok := fsw.getDefaultStore().(Debuggable); ok {
+		debuggable.Debug(writer)
+	}
+}

weed/filer/leveldb/leveldb_store.go

@@ -9,6 +9,7 @@ import (
 	"github.com/syndtr/goleveldb/leveldb/filter"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
+	"io"
 	"os"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
@@ -21,6 +22,10 @@ const (
 	DIR_FILE_SEPARATOR = byte(0x00)
 )
 
+var (
+	_ = filer.Debuggable(&LevelDBStore{})
+)
+
 func init() {
 	filer.Stores = append(filer.Stores, &LevelDBStore{})
 }
@@ -242,3 +247,13 @@ func getNameFromKey(key []byte) string {
 func (store *LevelDBStore) Shutdown() {
 	store.db.Close()
 }
+
+func (store *LevelDBStore) Debug(writer io.Writer) {
+	iter := store.db.NewIterator(&leveldb_util.Range{}, nil)
+	for iter.Next() {
+		key := iter.Key()
+		fullName := bytes.Replace(key, []byte{DIR_FILE_SEPARATOR}, []byte{' '}, 1)
+		fmt.Fprintf(writer, "%v\n", string(fullName))
+	}
+	iter.Release()
+}

weed/filer/permission.go

@@ -1,22 +0,0 @@
-package filer
-
-func hasWritePermission(dir *Entry, entry *Entry) bool {
-
-	if dir == nil {
-		return false
-	}
-
-	if dir.Uid == entry.Uid && dir.Mode&0200 > 0 {
-		return true
-	}
-
-	if dir.Gid == entry.Gid && dir.Mode&0020 > 0 {
-		return true
-	}
-
-	if dir.Mode&0002 > 0 {
-		return true
-	}
-
-	return false
-}

weed/filer/reader_at.go

@@ -129,7 +129,7 @@ func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, err error) {
 		}
 		if startOffset < chunk.LogicOffset {
 			gap := int(chunk.LogicOffset - startOffset)
-			glog.V(4).Infof("zero [%d,%d)", startOffset, startOffset+int64(gap))
+			glog.V(4).Infof("zero [%d,%d)", startOffset, chunk.LogicOffset)
 			n += int(min(int64(gap), remaining))
 			startOffset, remaining = chunk.LogicOffset, remaining-int64(gap)
 			if remaining <= 0 {

weed/filer/stream.go

@@ -80,11 +80,23 @@ func StreamContent(masterClient wdclient.HasLookupFileIdFunction, writer io.Writ
 		fileId2Url[chunkView.FileId] = urlStrings
 	}
 
+	remaining := size
 	for _, chunkView := range chunkViews {
-
+		if offset < chunkView.LogicOffset {
+			gap := chunkView.LogicOffset - offset
+			remaining -= gap
+			glog.V(4).Infof("zero [%d,%d)", offset, chunkView.LogicOffset)
+			err := writeZero(writer, gap)
+			if err != nil {
+				return fmt.Errorf("write zero [%d,%d)", offset, chunkView.LogicOffset)
+			}
+			offset = chunkView.LogicOffset
+		}
 		urlStrings := fileId2Url[chunkView.FileId]
 		start := time.Now()
 		err := retriedStreamFetchChunkData(writer, urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size))
+		offset += int64(chunkView.Size)
+		remaining -= int64(chunkView.Size)
 		stats.FilerRequestHistogram.WithLabelValues("chunkDownload").Observe(time.Since(start).Seconds())
 		if err != nil {
 			stats.FilerRequestCounter.WithLabelValues("chunkDownloadError").Inc()
@@ -92,6 +104,11 @@ func StreamContent(masterClient wdclient.HasLookupFileIdFunction, writer io.Writ
 		}
 		stats.FilerRequestCounter.WithLabelValues("chunkDownload").Inc()
 	}
+	glog.V(4).Infof("zero [%d,%d)", offset, offset+remaining)
+	err := writeZero(writer, remaining)
+	if err != nil {
+		return fmt.Errorf("write zero [%d,%d)", offset, offset+remaining)
+	}
 
 	return nil
 
@@ -99,6 +116,23 @@ func StreamContent(masterClient wdclient.HasLookupFileIdFunction, writer io.Writ
 
 // ----------------  ReadAllReader ----------------------------------
 
+func writeZero(w io.Writer, size int64) (err error) {
+	zeroPadding := make([]byte, 1024)
+	var written int
+	for size > 0 {
+		if size > 1024 {
+			written, err = w.Write(zeroPadding)
+		} else {
+			written, err = w.Write(zeroPadding[:size])
+		}
+		size -= int64(written)
+		if err != nil {
+			return
+		}
+	}
+	return
+}
+
 func ReadAll(masterClient *wdclient.MasterClient, chunks []*filer_pb.FileChunk) ([]byte, error) {
 
 	buffer := bytes.Buffer{}

weed/filesys/dir.go

@@ -66,7 +66,7 @@ func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
 	attr.Mode = os.FileMode(entry.Attributes.FileMode) | os.ModeDir
 	attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
 	attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
-	attr.Ctime = time.Unix(entry.Attributes.Crtime, 0)
+	attr.Ctime = time.Unix(entry.Attributes.Mtime, 0)
 	attr.Atime = time.Unix(entry.Attributes.Mtime, 0)
 	attr.Gid = entry.Attributes.Gid
 	attr.Uid = entry.Attributes.Uid
@@ -100,10 +100,10 @@ func (dir *Dir) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
 	return nil
 }
 
-func (dir *Dir) newFile(name string) fs.Node {
+func (dir *Dir) newFile(name string, fileMode os.FileMode) fs.Node {
 
 	fileFullPath := util.NewFullPath(dir.FullPath(), name)
-	fileId := fileFullPath.AsInode()
+	fileId := fileFullPath.AsInode(fileMode)
 	dir.wfs.handlesLock.Lock()
 	existingHandle, found := dir.wfs.handles[fileId]
 	dir.wfs.handlesLock.Unlock()
@@ -122,13 +122,17 @@ func (dir *Dir) newFile(name string) fs.Node {
 
 func (dir *Dir) newDirectory(fullpath util.FullPath) fs.Node {
 
-	return &Dir{name: fullpath.Name(), wfs: dir.wfs, parent: dir, id: fullpath.AsInode()}
+	return &Dir{name: fullpath.Name(), wfs: dir.wfs, parent: dir, id: fullpath.AsInode(os.ModeDir)}
 
 }
 
 func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 	resp *fuse.CreateResponse) (fs.Node, fs.Handle, error) {
 
+	if err := checkName(req.Name); err != nil {
+		return nil, nil, err
+	}
+
 	exclusive := req.Flags&fuse.OpenExclusive != 0
 	isDirectory := req.Mode&os.ModeDir > 0
 
@@ -144,7 +148,7 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 		return node, node, nil
 	}
 
-	node = dir.newFile(req.Name)
+	node = dir.newFile(req.Name, req.Mode)
 	file := node.(*File)
 	file.entry = &filer_pb.Entry{
 		Name:        req.Name,
@@ -168,13 +172,19 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 
 func (dir *Dir) Mknod(ctx context.Context, req *fuse.MknodRequest) (fs.Node, error) {
 
+	if err := checkName(req.Name); err != nil {
+		return nil, err
+	}
+
+	glog.V(3).Infof("dir %s Mknod %+v", dir.FullPath(), req)
+
 	_, err := dir.doCreateEntry(req.Name, req.Mode, req.Uid, req.Gid, false)
 
 	if err != nil {
 		return nil, err
 	}
 	var node fs.Node
-	node = dir.newFile(req.Name)
+	node = dir.newFile(req.Name, req.Mode)
 	return node, nil
 }
 
@@ -226,6 +236,10 @@ func (dir *Dir) doCreateEntry(name string, mode os.FileMode, uid, gid uint32, ex
 
 func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
 
+	if err := checkName(req.Name); err != nil {
+		return nil, err
+	}
+
 	glog.V(4).Infof("mkdir %s: %s", dir.FullPath(), req.Name)
 
 	newEntry := &filer_pb.Entry{
@@ -280,6 +294,10 @@ func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, err
 
 func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.LookupResponse) (node fs.Node, err error) {
 
+	if err := checkName(req.Name); err != nil {
+		return nil, err
+	}
+
 	dirPath := util.FullPath(dir.FullPath())
 	// glog.V(4).Infof("dir Lookup %s: %s by %s", dirPath, req.Name, req.Header.String())
 
@@ -310,11 +328,10 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
 		if localEntry.IsDirectory() {
 			node = dir.newDirectory(fullFilePath)
 		} else {
-			node = dir.newFile(req.Name)
+			node = dir.newFile(req.Name, localEntry.Attr.Mode)
 		}
 
 		// resp.EntryValid = time.Second
-		resp.Attr.Inode = fullFilePath.AsInode()
 		resp.Attr.Valid = time.Second
 		resp.Attr.Size = localEntry.FileSize
 		resp.Attr.Mtime = localEntry.Attr.Mtime
@@ -340,10 +357,10 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 
 	processEachEntryFn := func(entry *filer.Entry, isLast bool) {
 		if entry.IsDirectory() {
-			dirent := fuse.Dirent{Name: entry.Name(), Type: fuse.DT_Dir, Inode: dirPath.Child(entry.Name()).AsInode()}
+			dirent := fuse.Dirent{Name: entry.Name(), Type: fuse.DT_Dir, Inode: dirPath.Child(entry.Name()).AsInode(os.ModeDir)}
 			ret = append(ret, dirent)
 		} else {
-			dirent := fuse.Dirent{Name: entry.Name(), Type: findFileType(uint16(entry.Attr.Mode)), Inode: dirPath.Child(entry.Name()).AsInode()}
+			dirent := fuse.Dirent{Name: entry.Name(), Type: findFileType(uint16(entry.Attr.Mode)), Inode: dirPath.Child(entry.Name()).AsInode(entry.Attr.Mode)}
 			ret = append(ret, dirent)
 		}
 	}
@@ -363,7 +380,7 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 
 	// create proper . and .. directories
 	ret = append(ret, fuse.Dirent{
-		Inode: dirPath.AsInode(),
+		Inode: dirPath.AsInode(os.ModeDir),
 		Name:  ".",
 		Type:  fuse.DT_Dir,
 	})
@@ -373,7 +390,7 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 	if string(dirPath) == dir.wfs.option.FilerMountRootPath {
 		inode = dir.wfs.option.MountParentInode
 	} else {
-		inode = util.FullPath(dir.parent.FullPath()).AsInode()
+		inode = util.FullPath(dir.parent.FullPath()).AsInode(os.ModeDir)
 	}
 
 	ret = append(ret, fuse.Dirent{
@@ -407,15 +424,14 @@ func findFileType(mode uint16) fuse.DirentType {
 
 func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
 
-	if !req.Dir {
-		return dir.removeOneFile(req)
+	parentHasPermission := false
+	parentEntry, err := dir.maybeLoadEntry()
+	if err != nil {
+		return err
+	}
+	if err := checkPermission(parentEntry, req.Uid, req.Gid, true); err == nil {
+		parentHasPermission = true
 	}
-
-	return dir.removeFolder(req)
-
-}
-
-func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 
 	dirFullPath := dir.FullPath()
 	filePath := util.NewFullPath(dirFullPath, req.Name)
@@ -423,11 +439,29 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 	if err != nil {
 		return err
 	}
+	if !parentHasPermission {
+		if err := checkPermission(entry, req.Uid, req.Gid, true); err != nil {
+			return err
+		}
+	}
+
+	if !req.Dir {
+		return dir.removeOneFile(entry, req)
+	}
+
+	return dir.removeFolder(entry, req)
+
+}
+
+func (dir *Dir) removeOneFile(entry *filer_pb.Entry, req *fuse.RemoveRequest) error {
+
+	dirFullPath := dir.FullPath()
+	filePath := util.NewFullPath(dirFullPath, req.Name)
 
 	// first, ensure the filer store can correctly delete
 	glog.V(3).Infof("remove file: %v", req)
 	isDeleteData := entry != nil && entry.HardLinkCounter <= 1
-	err = filer_pb.Remove(dir.wfs, dirFullPath, req.Name, isDeleteData, false, false, false, []int32{dir.wfs.signature})
+	err := filer_pb.Remove(dir.wfs, dirFullPath, req.Name, isDeleteData, false, false, false, []int32{dir.wfs.signature})
 	if err != nil {
 		glog.V(3).Infof("not found remove file %s: %v", filePath, err)
 		return fuse.ENOENT
@@ -442,7 +476,7 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 	// remove current file handle if any
 	dir.wfs.handlesLock.Lock()
 	defer dir.wfs.handlesLock.Unlock()
-	inodeId := filePath.AsInode()
+	inodeId := filePath.AsInode(0)
 	if fh, ok := dir.wfs.handles[inodeId]; ok {
 		delete(dir.wfs.handles, inodeId)
 		fh.isDeleted = true
@@ -452,15 +486,16 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 
 }
 
-func (dir *Dir) removeFolder(req *fuse.RemoveRequest) error {
+func (dir *Dir) removeFolder(entry *filer_pb.Entry, req *fuse.RemoveRequest) error {
 
 	dirFullPath := dir.FullPath()
+
 	glog.V(3).Infof("remove directory entry: %v", req)
-	ignoreRecursiveErr := true // ignore recursion error since the OS should manage it
-	err := filer_pb.Remove(dir.wfs, dirFullPath, req.Name, true, true, ignoreRecursiveErr, false, []int32{dir.wfs.signature})
+	ignoreRecursiveErr := false // ignore recursion error since the OS should manage it
+	err := filer_pb.Remove(dir.wfs, dirFullPath, req.Name, true, false, ignoreRecursiveErr, false, []int32{dir.wfs.signature})
 	if err != nil {
 		glog.V(0).Infof("remove %s/%s: %v", dirFullPath, req.Name, err)
-		if strings.Contains(err.Error(), "non-empty") {
+		if strings.Contains(err.Error(), filer.MsgFailDelNonEmptyFolder) {
 			return fuse.EEXIST
 		}
 		return fuse.ENOENT
@@ -475,7 +510,7 @@ func (dir *Dir) removeFolder(req *fuse.RemoveRequest) error {
 
 func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
 
-	glog.V(4).Infof("%v dir setattr %+v", dir.FullPath(), req)
+	glog.V(4).Infof("%v dir setattr %+v mode=%d", dir.FullPath(), req, req.Mode)
 
 	entry, err := dir.maybeLoadEntry()
 	if err != nil {
@@ -498,6 +533,8 @@ func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fus
 		entry.Attributes.Mtime = req.Mtime.Unix()
 	}
 
+	entry.Attributes.Mtime = time.Now().Unix()
+
 	return dir.saveEntry(entry)
 
 }

weed/filesys/dir_link.go

@@ -24,6 +24,10 @@ const (
 
 func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (fs.Node, error) {
 
+	if err := checkName(req.NewName); err != nil {
+		return nil, err
+	}
+
 	oldFile, ok := old.(*File)
 	if !ok {
 		glog.Errorf("old node is not a file: %+v", old)
@@ -53,6 +57,7 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 	}
 
 	// CreateLink 1.2 : update new file to hardlink mode
+	oldEntry.Attributes.Mtime = time.Now().Unix()
 	request := &filer_pb.CreateEntryRequest{
 		Directory: dir.FullPath(),
 		Entry: &filer_pb.Entry{
@@ -93,7 +98,7 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 	}
 
 	// create new file node
-	newNode := dir.newFile(req.NewName)
+	newNode := dir.newFile(req.NewName, 0)
 	newFile := newNode.(*File)
 
 	return newFile, err
@@ -102,6 +107,10 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 
 func (dir *Dir) Symlink(ctx context.Context, req *fuse.SymlinkRequest) (fs.Node, error) {
 
+	if err := checkName(req.NewName); err != nil {
+		return nil, err
+	}
+
 	glog.V(4).Infof("Symlink: %v/%v to %v", dir.FullPath(), req.NewName, req.Target)
 
 	request := &filer_pb.CreateEntryRequest{
@@ -136,7 +145,7 @@ func (dir *Dir) Symlink(ctx context.Context, req *fuse.SymlinkRequest) (fs.Node,
 		return nil
 	})
 
-	symlink := dir.newFile(req.NewName)
+	symlink := dir.newFile(req.NewName, os.ModeSymlink)
 
 	return symlink, err
 

weed/filesys/dir_rename.go

@@ -3,17 +3,25 @@ package filesys
 import (
 	"context"
 	"github.com/chrislusf/seaweedfs/weed/filer"
-	"github.com/seaweedfs/fuse"
-	"github.com/seaweedfs/fuse/fs"
-	"io"
-
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/seaweedfs/fuse"
+	"github.com/seaweedfs/fuse/fs"
+	"io"
+	"os"
+	"strings"
 )
 
 func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirectory fs.Node) error {
 
+	if err := checkName(req.NewName); err != nil {
+		return err
+	}
+	if err := checkName(req.OldName); err != nil {
+		return err
+	}
+
 	newDir := newDirectory.(*Dir)
 
 	newPath := util.NewFullPath(newDir.FullPath(), req.NewName)
@@ -37,7 +45,7 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 		stream, err := client.StreamRenameEntry(ctx, request)
 		if err != nil {
 			glog.Errorf("dir AtomicRenameEntry %s => %s : %v", oldPath, newPath, err)
-			return fuse.EXDEV
+			return fuse.EIO
 		}
 
 		for {
@@ -46,11 +54,19 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 				if recvErr == io.EOF {
 					break
 				} else {
+					glog.V(0).Infof("dir Rename %s => %s receive: %v", oldPath, newPath, recvErr)
+					if strings.Contains(recvErr.Error(), "not empty") {
+						return fuse.EEXIST
+					}
+					if strings.Contains(recvErr.Error(), "not directory") {
+						return fuse.ENOTDIR
+					}
 					return recvErr
 				}
 			}
 
 			if err = dir.handleRenameResponse(ctx, resp); err != nil {
+				glog.V(0).Infof("dir Rename %s => %s : %v", oldPath, newPath, err)
 				return err
 			}
 
@@ -59,12 +75,8 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 		return nil
 
 	})
-	if err != nil {
-		glog.V(0).Infof("dir Rename %s => %s : %v", oldPath, newPath, err)
-		return fuse.EIO
-	}
 
-	return nil
+	return err
 }
 
 func (dir *Dir) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamRenameEntryResponse) error {
@@ -80,11 +92,13 @@ func (dir *Dir) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamR
 		oldParent, newParent := util.FullPath(resp.Directory), util.FullPath(resp.EventNotification.NewParentPath)
 		oldName, newName := resp.EventNotification.OldEntry.Name, resp.EventNotification.NewEntry.Name
 
+		entryFileMode := newEntry.Attr.Mode
+
 		oldPath := oldParent.Child(oldName)
 		newPath := newParent.Child(newName)
-		oldFsNode := NodeWithId(oldPath.AsInode())
-		newFsNode := NodeWithId(newPath.AsInode())
-		newDirNode, found := dir.wfs.Server.FindInternalNode(NodeWithId(newParent.AsInode()))
+		oldFsNode := NodeWithId(oldPath.AsInode(entryFileMode))
+		newFsNode := NodeWithId(newPath.AsInode(entryFileMode))
+		newDirNode, found := dir.wfs.Server.FindInternalNode(NodeWithId(newParent.AsInode(os.ModeDir)))
 		var newDir *Dir
 		if found {
 			newDir = newDirNode.(*Dir)
@@ -109,14 +123,19 @@ func (dir *Dir) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamR
 		})
 
 		// change file handle
-		inodeId := oldPath.AsInode()
-		dir.wfs.handlesLock.Lock()
-		if existingHandle, found := dir.wfs.handles[inodeId]; found && existingHandle != nil {
-			glog.V(4).Infof("opened file handle %s => %s", oldPath, newPath)
-			delete(dir.wfs.handles, inodeId)
-			dir.wfs.handles[newPath.AsInode()] = existingHandle
+		if !newEntry.IsDirectory() {
+			inodeId := oldPath.AsInode(entryFileMode)
+			dir.wfs.handlesLock.Lock()
+			if existingHandle, found := dir.wfs.handles[inodeId]; found && existingHandle != nil {
+				glog.V(4).Infof("opened file handle %s => %s", oldPath, newPath)
+				delete(dir.wfs.handles, inodeId)
+				existingHandle.handle = newPath.AsInode(entryFileMode)
+				existingHandle.f.entry.Name = newName
+				existingHandle.f.id = newPath.AsInode(entryFileMode)
+				dir.wfs.handles[newPath.AsInode(entryFileMode)] = existingHandle
+			}
+			dir.wfs.handlesLock.Unlock()
 		}
-		dir.wfs.handlesLock.Unlock()
 
 	} else if resp.EventNotification.OldEntry != nil {
 		// without new entry, only old entry name exists. This is the second step to delete old entry

weed/filesys/dirty_pages_continuous.go

@@ -1,138 +0,0 @@
-package filesys
-
-import (
-	"bytes"
-	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/filesys/page_writer"
-	"io"
-	"sync"
-	"time"
-
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-)
-
-type ContinuousDirtyPages struct {
-	intervals      *page_writer.ContinuousIntervals
-	f              *File
-	writeWaitGroup sync.WaitGroup
-	chunkAddLock   sync.Mutex
-	lastErr        error
-	collection     string
-	replication    string
-}
-
-func newContinuousDirtyPages(file *File) *ContinuousDirtyPages {
-	dirtyPages := &ContinuousDirtyPages{
-		intervals: &page_writer.ContinuousIntervals{},
-		f:         file,
-	}
-	return dirtyPages
-}
-
-func (pages *ContinuousDirtyPages) AddPage(offset int64, data []byte) {
-
-	glog.V(4).Infof("%s AddPage [%d,%d)", pages.f.fullpath(), offset, offset+int64(len(data)))
-
-	if len(data) > int(pages.f.wfs.option.ChunkSizeLimit) {
-		// this is more than what buffer can hold.
-		pages.flushAndSave(offset, data)
-	}
-
-	pages.intervals.AddInterval(data, offset)
-
-	if pages.intervals.TotalSize() >= pages.f.wfs.option.ChunkSizeLimit {
-		pages.saveExistingLargestPageToStorage()
-	}
-
-	return
-}
-
-func (pages *ContinuousDirtyPages) flushAndSave(offset int64, data []byte) {
-
-	// flush existing
-	pages.saveExistingPagesToStorage()
-
-	// flush the new page
-	pages.saveToStorage(bytes.NewReader(data), offset, int64(len(data)))
-
-	return
-}
-
-func (pages *ContinuousDirtyPages) FlushData() error {
-
-	pages.saveExistingPagesToStorage()
-	pages.writeWaitGroup.Wait()
-	if pages.lastErr != nil {
-		return fmt.Errorf("flush data: %v", pages.lastErr)
-	}
-	return nil
-}
-
-func (pages *ContinuousDirtyPages) saveExistingPagesToStorage() {
-	for pages.saveExistingLargestPageToStorage() {
-	}
-}
-
-func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage() (hasSavedData bool) {
-
-	maxList := pages.intervals.RemoveLargestIntervalLinkedList()
-	if maxList == nil {
-		return false
-	}
-
-	entry := pages.f.getEntry()
-	if entry == nil {
-		return false
-	}
-
-	fileSize := int64(entry.Attributes.FileSize)
-
-	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) {
-
-	mtime := time.Now().UnixNano()
-	pages.writeWaitGroup.Add(1)
-	writer := func() {
-		defer pages.writeWaitGroup.Done()
-
-		reader = io.LimitReader(reader, size)
-		chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(pages.f.fullpath())(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.lastErr = err
-			return
-		}
-		chunk.Mtime = mtime
-		pages.collection, pages.replication = collection, replication
-		pages.chunkAddLock.Lock()
-		defer pages.chunkAddLock.Unlock()
-		pages.f.addChunks([]*filer_pb.FileChunk{chunk})
-		glog.V(3).Infof("%s saveToStorage [%d,%d)", pages.f.fullpath(), offset, offset+size)
-	}
-
-	if pages.f.wfs.concurrentWriters != nil {
-		pages.f.wfs.concurrentWriters.Execute(writer)
-	} else {
-		go writer()
-	}
-}
-
-func (pages *ContinuousDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
-	return pages.intervals.ReadDataAt(data, startOffset)
-}
-
-func (pages *ContinuousDirtyPages) GetStorageOptions() (collection, replication string) {
-	return pages.collection, pages.replication
-}
-func (pages ContinuousDirtyPages) Destroy() {
-}

weed/filesys/dirty_pages_mem_chunk.go

@@ -0,0 +1,101 @@
+package filesys
+
+import (
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/filesys/page_writer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"io"
+	"sync"
+	"time"
+)
+
+type MemoryChunkPages struct {
+	fh             *FileHandle
+	writeWaitGroup sync.WaitGroup
+	chunkAddLock   sync.Mutex
+	lastErr        error
+	collection     string
+	replication    string
+	uploadPipeline *page_writer.UploadPipeline
+	hasWrites      bool
+}
+
+var (
+	_ = page_writer.DirtyPages(&MemoryChunkPages{})
+)
+
+func newMemoryChunkPages(fh *FileHandle, chunkSize int64) *MemoryChunkPages {
+
+	dirtyPages := &MemoryChunkPages{
+		fh: fh,
+	}
+
+	dirtyPages.uploadPipeline = page_writer.NewUploadPipeline(fh.f.fullpath(),
+		fh.f.wfs.concurrentWriters, chunkSize, dirtyPages.saveChunkedFileIntevalToStorage)
+
+	return dirtyPages
+}
+
+func (pages *MemoryChunkPages) AddPage(offset int64, data []byte) {
+	pages.hasWrites = true
+
+	glog.V(4).Infof("%v memory AddPage [%d, %d)", pages.fh.f.fullpath(), offset, offset+int64(len(data)))
+	pages.uploadPipeline.SaveDataAt(data, offset)
+
+	return
+}
+
+func (pages *MemoryChunkPages) FlushData() error {
+	if !pages.hasWrites {
+		return nil
+	}
+	pages.uploadPipeline.FlushAll()
+	if pages.lastErr != nil {
+		return fmt.Errorf("flush data: %v", pages.lastErr)
+	}
+	return nil
+}
+
+func (pages *MemoryChunkPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
+	if !pages.hasWrites {
+		return
+	}
+	return pages.uploadPipeline.MaybeReadDataAt(data, startOffset)
+}
+
+func (pages *MemoryChunkPages) GetStorageOptions() (collection, replication string) {
+	return pages.collection, pages.replication
+}
+
+func (pages *MemoryChunkPages) saveChunkedFileIntevalToStorage(reader io.Reader, offset int64, size int64, cleanupFn func()) {
+
+	mtime := time.Now().UnixNano()
+	defer cleanupFn()
+
+	chunk, collection, replication, err := pages.fh.f.wfs.saveDataAsChunk(pages.fh.f.fullpath())(reader, pages.fh.f.Name, offset)
+	if err != nil {
+		glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.fh.f.fullpath(), offset, offset+size, err)
+		pages.lastErr = err
+		return
+	}
+	chunk.Mtime = mtime
+	pages.collection, pages.replication = collection, replication
+	pages.chunkAddLock.Lock()
+	pages.fh.f.addChunks([]*filer_pb.FileChunk{chunk})
+	pages.fh.entryViewCache = nil
+	glog.V(3).Infof("%s saveToStorage %s [%d,%d)", pages.fh.f.fullpath(), chunk.FileId, offset, offset+size)
+	pages.chunkAddLock.Unlock()
+
+}
+
+func (pages MemoryChunkPages) Destroy() {
+	pages.uploadPipeline.Shutdown()
+}
+
+func (pages *MemoryChunkPages) LockForRead(startOffset, stopOffset int64) {
+	pages.uploadPipeline.LockForRead(startOffset, stopOffset)
+}
+func (pages *MemoryChunkPages) UnlockForRead(startOffset, stopOffset int64) {
+	pages.uploadPipeline.UnlockForRead(startOffset, stopOffset)
+}

weed/filesys/dirty_pages_stream.go

@@ -1,106 +0,0 @@
-package filesys
-
-import (
-	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/filesys/page_writer"
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-	"io"
-	"sync"
-	"time"
-)
-
-type StreamDirtyPages struct {
-	f              *File
-	writeWaitGroup sync.WaitGroup
-	pageAddLock    sync.Mutex
-	chunkAddLock   sync.Mutex
-	lastErr        error
-	collection     string
-	replication    string
-	chunkedStream  *page_writer.ChunkedStreamWriter
-}
-
-func newStreamDirtyPages(file *File, chunkSize int64) *StreamDirtyPages {
-
-	dirtyPages := &StreamDirtyPages{
-		f:             file,
-		chunkedStream: page_writer.NewChunkedStreamWriter(chunkSize),
-	}
-
-	dirtyPages.chunkedStream.SetSaveToStorageFunction(dirtyPages.saveChunkedFileIntevalToStorage)
-
-	return dirtyPages
-}
-
-func (pages *StreamDirtyPages) AddPage(offset int64, data []byte) {
-
-	pages.pageAddLock.Lock()
-	defer pages.pageAddLock.Unlock()
-
-	glog.V(4).Infof("%v stream AddPage [%d, %d)", pages.f.fullpath(), offset, offset+int64(len(data)))
-	if _, err := pages.chunkedStream.WriteAt(data, offset); err != nil {
-		pages.lastErr = err
-	}
-
-	return
-}
-
-func (pages *StreamDirtyPages) FlushData() error {
-	pages.saveChunkedFileToStorage()
-	pages.writeWaitGroup.Wait()
-	if pages.lastErr != nil {
-		return fmt.Errorf("flush data: %v", pages.lastErr)
-	}
-	pages.chunkedStream.Reset()
-	return nil
-}
-
-func (pages *StreamDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
-	return pages.chunkedStream.ReadDataAt(data, startOffset)
-}
-
-func (pages *StreamDirtyPages) GetStorageOptions() (collection, replication string) {
-	return pages.collection, pages.replication
-}
-
-func (pages *StreamDirtyPages) saveChunkedFileToStorage() {
-
-	pages.chunkedStream.FlushAll()
-
-}
-
-func (pages *StreamDirtyPages) saveChunkedFileIntevalToStorage(reader io.Reader, offset int64, size int64, cleanupFn func()) {
-
-	mtime := time.Now().UnixNano()
-	pages.writeWaitGroup.Add(1)
-	writer := func() {
-		defer pages.writeWaitGroup.Done()
-		defer cleanupFn()
-
-		chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(pages.f.fullpath())(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.lastErr = err
-			return
-		}
-		chunk.Mtime = mtime
-		pages.collection, pages.replication = collection, replication
-		pages.chunkAddLock.Lock()
-		pages.f.addChunks([]*filer_pb.FileChunk{chunk})
-		glog.V(3).Infof("%s saveToStorage %s [%d,%d)", pages.f.fullpath(), chunk.FileId, offset, offset+size)
-		pages.chunkAddLock.Unlock()
-
-	}
-
-	if pages.f.wfs.concurrentWriters != nil {
-		pages.f.wfs.concurrentWriters.Execute(writer)
-	} else {
-		go writer()
-	}
-
-}
-
-func (pages StreamDirtyPages) Destroy() {
-	pages.chunkedStream.Reset()
-}

weed/filesys/dirty_pages_temp_file.go

@@ -104,3 +104,9 @@ func (pages *TempFileDirtyPages) saveChunkedFileIntevalToStorage(reader io.Reade
 func (pages TempFileDirtyPages) Destroy() {
 	pages.chunkedFile.Reset()
 }
+
+func (pages *TempFileDirtyPages) LockForRead(startOffset, stopOffset int64) {
+}
+
+func (pages *TempFileDirtyPages) UnlockForRead(startOffset, stopOffset int64) {
+}

weed/filesys/file.go

@@ -68,6 +68,7 @@ func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
 		glog.V(4).Infof("file Attr %s, open:%v, size: %d", file.fullpath(), file.isOpen, attr.Size)
 	}
 	attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
+	attr.Ctime = time.Unix(entry.Attributes.Mtime, 0)
 	attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
 	attr.Gid = entry.Attributes.Gid
 	attr.Uid = entry.Attributes.Uid
@@ -110,7 +111,7 @@ func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.Op
 
 func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
 
-	glog.V(4).Infof("%v file setattr %+v", file.fullpath(), req)
+	glog.V(4).Infof("%v file setattr %+v mode=%d", file.fullpath(), req, req.Mode)
 
 	entry, err := file.maybeLoadEntry(ctx)
 	if err != nil {
@@ -139,29 +140,42 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 					}
 				}
 			}
+			// set the new chunks and reset entry cache
 			entry.Chunks = chunks
+			file.wfs.handlesLock.Lock()
+			existingHandle, found := file.wfs.handles[file.Id()]
+			file.wfs.handlesLock.Unlock()
+			if found {
+				existingHandle.entryViewCache = nil
+			}
+
 		}
+		entry.Attributes.Mtime = time.Now().Unix()
 		entry.Attributes.FileSize = req.Size
 		file.dirtyMetadata = true
 	}
 
 	if req.Valid.Mode() && entry.Attributes.FileMode != uint32(req.Mode) {
 		entry.Attributes.FileMode = uint32(req.Mode)
+		entry.Attributes.Mtime = time.Now().Unix()
 		file.dirtyMetadata = true
 	}
 
 	if req.Valid.Uid() && entry.Attributes.Uid != req.Uid {
 		entry.Attributes.Uid = req.Uid
+		entry.Attributes.Mtime = time.Now().Unix()
 		file.dirtyMetadata = true
 	}
 
 	if req.Valid.Gid() && entry.Attributes.Gid != req.Gid {
 		entry.Attributes.Gid = req.Gid
+		entry.Attributes.Mtime = time.Now().Unix()
 		file.dirtyMetadata = true
 	}
 
 	if req.Valid.Crtime() {
 		entry.Attributes.Crtime = req.Crtime.Unix()
+		entry.Attributes.Mtime = time.Now().Unix()
 		file.dirtyMetadata = true
 	}
 
@@ -248,17 +262,19 @@ func (file *File) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, res
 }
 
 func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
-	// fsync works at OS level
+
 	// write the file chunks to the filerGrpcAddress
 	glog.V(4).Infof("%s/%s fsync file %+v", file.dir.FullPath(), file.Name, req)
 
-	return nil
+	return file.wfs.Fsync(file, req.Header)
+
 }
 
 func (file *File) Forget() {
 	t := util.NewFullPath(file.dir.FullPath(), file.Name)
 	glog.V(4).Infof("Forget file %s", t)
-	file.wfs.ReleaseHandle(t, fuse.HandleID(t.AsInode()))
+	file.wfs.ReleaseHandle(t, fuse.HandleID(t.AsInode(file.entry.FileMode())))
+
 }
 
 func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {

weed/filesys/filehandle.go

@@ -26,25 +26,22 @@ type FileHandle struct {
 	contentType    string
 	handle         uint64
 	sync.Mutex
-	sync.WaitGroup
 
 	f         *File
-	RequestId fuse.RequestID // unique ID for request
-	NodeId    fuse.NodeID    // file or directory the request is about
-	Uid       uint32         // user ID of process making request
-	Gid       uint32         // group ID of process making request
-	writeOnly bool
+	NodeId    fuse.NodeID // file or directory the request is about
+	Uid       uint32      // user ID of process making request
+	Gid       uint32      // group ID of process making request
 	isDeleted bool
 }
 
 func newFileHandle(file *File, uid, gid uint32) *FileHandle {
 	fh := &FileHandle{
-		f: file,
-		// dirtyPages: newContinuousDirtyPages(file, writeOnly),
-		dirtyPages: newPageWriter(file, file.wfs.option.ChunkSizeLimit),
-		Uid:        uid,
-		Gid:        gid,
+		f:   file,
+		Uid: uid,
+		Gid: gid,
 	}
+	// dirtyPages: newContinuousDirtyPages(file, writeOnly),
+	fh.dirtyPages = newPageWriter(fh, file.wfs.option.ChunkSizeLimit)
 	entry := fh.f.getEntry()
 	if entry != nil {
 		entry.Attributes.FileSize = filer.FileSize(entry)
@@ -63,9 +60,6 @@ var _ = fs.HandleReleaser(&FileHandle{})
 
 func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
 
-	fh.Add(1)
-	defer fh.Done()
-
 	fh.Lock()
 	defer fh.Unlock()
 
@@ -81,6 +75,8 @@ func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fus
 		buff = make([]byte, req.Size)
 	}
 
+	fh.lockForRead(req.Offset, len(buff))
+	defer fh.unlockForRead(req.Offset, len(buff))
 	totalRead, err := fh.readFromChunks(buff, req.Offset)
 	if err == nil || err == io.EOF {
 		maxStop := fh.readFromDirtyPages(buff, req.Offset)
@@ -107,6 +103,13 @@ func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fus
 	return err
 }
 
+func (fh *FileHandle) lockForRead(startOffset int64, size int) {
+	fh.dirtyPages.LockForRead(startOffset, startOffset+int64(size))
+}
+func (fh *FileHandle) unlockForRead(startOffset int64, size int) {
+	fh.dirtyPages.UnlockForRead(startOffset, startOffset+int64(size))
+}
+
 func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64) (maxStop int64) {
 	maxStop = fh.dirtyPages.ReadDirtyDataAt(buff, startOffset)
 	return
@@ -155,6 +158,10 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 	reader := fh.reader
 	if reader == nil {
 		chunkViews := filer.ViewFromVisibleIntervals(fh.entryViewCache, 0, math.MaxInt64)
+		glog.V(4).Infof("file handle read %s [%d,%d) from %+v", fileFullPath, offset, offset+int64(len(buff)), chunkViews)
+		for _, chunkView := range chunkViews {
+			glog.V(4).Infof("  read %s [%d,%d) from chunk %+v", fileFullPath, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size), chunkView.FileId)
+		}
 		reader = filer.NewChunkReaderAtFromClient(fh.f.wfs.LookupFn(), chunkViews, fh.f.wfs.chunkCache, fileSize)
 	}
 	fh.reader = reader
@@ -165,7 +172,7 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 		glog.Errorf("file handle read %s: %v", fileFullPath, err)
 	}
 
-	// glog.V(4).Infof("file handle read %s [%d,%d] %d : %v", fileFullPath, offset, offset+int64(totalRead), totalRead, err)
+	glog.V(4).Infof("file handle read %s [%d,%d] %d : %v", fileFullPath, offset, offset+int64(totalRead), totalRead, err)
 
 	return int64(totalRead), err
 }
@@ -173,9 +180,6 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 // Write to the file handle
 func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
 
-	fh.Add(1)
-	defer fh.Done()
-
 	fh.dirtyPages.writerPattern.MonitorWriteAt(req.Offset, len(req.Data))
 
 	fh.Lock()
@@ -217,8 +221,6 @@ func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) err
 
 	glog.V(4).Infof("Release %v fh %d open=%d", fh.f.fullpath(), fh.handle, fh.f.isOpen)
 
-	fh.Wait()
-
 	fh.f.wfs.handlesLock.Lock()
 	fh.f.isOpen--
 	fh.f.wfs.handlesLock.Unlock()
@@ -250,9 +252,6 @@ func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
 		return nil
 	}
 
-	fh.Add(1)
-	defer fh.Done()
-
 	fh.Lock()
 	defer fh.Unlock()
 

weed/filesys/meta_cache/meta_cache.go

@@ -5,6 +5,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/filer/leveldb"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/chrislusf/seaweedfs/weed/util/bounded_tree"
 	"os"
@@ -18,16 +19,16 @@ type MetaCache struct {
 	// sync.RWMutex
 	visitedBoundary *bounded_tree.BoundedTree
 	uidGidMapper    *UidGidMapper
-	invalidateFunc  func(util.FullPath)
+	invalidateFunc  func(fullpath util.FullPath, entry *filer_pb.Entry)
 }
 
-func NewMetaCache(dbFolder string, baseDir util.FullPath, uidGidMapper *UidGidMapper, invalidateFunc func(util.FullPath)) *MetaCache {
+func NewMetaCache(dbFolder string, baseDir util.FullPath, uidGidMapper *UidGidMapper, invalidateFunc func(util.FullPath, *filer_pb.Entry)) *MetaCache {
 	return &MetaCache{
 		localStore:      openMetaStore(dbFolder),
 		visitedBoundary: bounded_tree.NewBoundedTree(baseDir),
 		uidGidMapper:    uidGidMapper,
-		invalidateFunc: func(fullpath util.FullPath) {
-			invalidateFunc(fullpath)
+		invalidateFunc: func(fullpath util.FullPath, entry *filer_pb.Entry) {
+			invalidateFunc(fullpath, entry)
 		},
 	}
 }
@@ -144,3 +145,10 @@ func (mc *MetaCache) Shutdown() {
 func (mc *MetaCache) mapIdFromFilerToLocal(entry *filer.Entry) {
 	entry.Attr.Uid, entry.Attr.Gid = mc.uidGidMapper.FilerToLocal(entry.Attr.Uid, entry.Attr.Gid)
 }
+
+func (mc *MetaCache) Debug() {
+	if debuggable, ok := mc.localStore.(filer.Debuggable); ok {
+		println("start debugging")
+		debuggable.Debug(os.Stderr)
+	}
+}

weed/filesys/meta_cache/meta_cache_subscribe.go

@@ -40,16 +40,16 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
 		if err == nil {
 			if message.OldEntry != nil && message.NewEntry != nil {
 				oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
-				mc.invalidateFunc(oldKey)
+				mc.invalidateFunc(oldKey, message.OldEntry)
 				if message.OldEntry.Name != message.NewEntry.Name {
 					newKey := util.NewFullPath(dir, message.NewEntry.Name)
-					mc.invalidateFunc(newKey)
+					mc.invalidateFunc(newKey, message.NewEntry)
 				}
 			} else if message.OldEntry == nil && message.NewEntry != nil {
 				// no need to invaalidate
 			} else if message.OldEntry != nil && message.NewEntry == nil {
 				oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
-				mc.invalidateFunc(oldKey)
+				mc.invalidateFunc(oldKey, message.OldEntry)
 			}
 		}
 

weed/filesys/page_writer.go

@@ -6,36 +6,33 @@ import (
 )
 
 type PageWriter struct {
-	f             *File
+	fh            *FileHandle
 	collection    string
 	replication   string
 	chunkSize     int64
 	writerPattern *WriterPattern
 
 	randomWriter page_writer.DirtyPages
-	streamWriter page_writer.DirtyPages
 }
 
 var (
 	_ = page_writer.DirtyPages(&PageWriter{})
 )
 
-func newPageWriter(file *File, chunkSize int64) *PageWriter {
+func newPageWriter(fh *FileHandle, chunkSize int64) *PageWriter {
 	pw := &PageWriter{
-		f:             file,
+		fh:            fh,
 		chunkSize:     chunkSize,
 		writerPattern: NewWriterPattern(chunkSize),
-		randomWriter:  newTempFileDirtyPages(file, chunkSize),
-		streamWriter:  newStreamDirtyPages(file, chunkSize),
-		//streamWriter: newContinuousDirtyPages(file),
-		//streamWriter:  nil,
+		randomWriter:  newMemoryChunkPages(fh, chunkSize),
+		// randomWriter: newTempFileDirtyPages(fh.f, chunkSize),
 	}
 	return pw
 }
 
 func (pw *PageWriter) AddPage(offset int64, data []byte) {
 
-	glog.V(4).Infof("%v AddPage [%d, %d) streaming:%v", pw.f.fullpath(), offset, offset+int64(len(data)), pw.writerPattern.IsStreamingMode())
+	glog.V(4).Infof("%v AddPage [%d, %d) streaming:%v", pw.fh.f.fullpath(), offset, offset+int64(len(data)), pw.writerPattern.IsStreamingMode())
 
 	chunkIndex := offset / pw.chunkSize
 	for i := chunkIndex; len(data) > 0; i++ {
@@ -47,38 +44,22 @@ func (pw *PageWriter) AddPage(offset int64, data []byte) {
 }
 
 func (pw *PageWriter) addToOneChunk(chunkIndex, offset int64, data []byte) {
-	if chunkIndex > 0 {
-		if pw.writerPattern.IsStreamingMode() && pw.streamWriter != nil {
-			pw.streamWriter.AddPage(offset, data)
-			return
-		}
-	}
 	pw.randomWriter.AddPage(offset, data)
 }
 
 func (pw *PageWriter) FlushData() error {
 	pw.writerPattern.Reset()
-	if pw.streamWriter != nil {
-		if err := pw.streamWriter.FlushData(); err != nil {
-			return err
-		}
-	}
 	return pw.randomWriter.FlushData()
 }
 
 func (pw *PageWriter) ReadDirtyDataAt(data []byte, offset int64) (maxStop int64) {
-	glog.V(4).Infof("ReadDirtyDataAt %v [%d, %d)", pw.f.fullpath(), offset, offset+int64(len(data)))
+	glog.V(4).Infof("ReadDirtyDataAt %v [%d, %d)", pw.fh.f.fullpath(), offset, offset+int64(len(data)))
 
 	chunkIndex := offset / pw.chunkSize
 	for i := chunkIndex; len(data) > 0; i++ {
 		readSize := min(int64(len(data)), (i+1)*pw.chunkSize-offset)
 
-		if pw.streamWriter != nil {
-			m1 := pw.streamWriter.ReadDirtyDataAt(data[:readSize], offset)
-			maxStop = max(maxStop, m1)
-		}
-		m2 := pw.randomWriter.ReadDirtyDataAt(data[:readSize], offset)
-		maxStop = max(maxStop, m2)
+		maxStop = pw.randomWriter.ReadDirtyDataAt(data[:readSize], offset)
 
 		offset += readSize
 		data = data[readSize:]
@@ -88,16 +69,18 @@ func (pw *PageWriter) ReadDirtyDataAt(data []byte, offset int64) (maxStop int64)
 }
 
 func (pw *PageWriter) GetStorageOptions() (collection, replication string) {
-	if pw.writerPattern.IsStreamingMode() && pw.streamWriter != nil {
-		return pw.streamWriter.GetStorageOptions()
-	}
 	return pw.randomWriter.GetStorageOptions()
 }
 
+func (pw *PageWriter) LockForRead(startOffset, stopOffset int64) {
+	pw.randomWriter.LockForRead(startOffset, stopOffset)
+}
+
+func (pw *PageWriter) UnlockForRead(startOffset, stopOffset int64) {
+	pw.randomWriter.UnlockForRead(startOffset, stopOffset)
+}
+
 func (pw *PageWriter) Destroy() {
-	if pw.streamWriter != nil {
-		pw.streamWriter.Destroy()
-	}
 	pw.randomWriter.Destroy()
 }
 

weed/filesys/page_writer/chunked_file_writer.go

@@ -7,7 +7,6 @@ import (
 	"sync"
 )
 
-type LogicChunkIndex int
 type ActualChunkIndex int
 
 // ChunkedFileWriter assumes the write requests will come in within chunks

weed/filesys/page_writer/chunked_stream_writer.go

@@ -1,119 +1,12 @@
 package page_writer
 
 import (
-	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/chrislusf/seaweedfs/weed/util/mem"
 	"io"
-	"sync"
-	"sync/atomic"
 )
 
 type SaveToStorageFunc func(reader io.Reader, offset int64, size int64, cleanupFn func())
 
-// ChunkedStreamWriter assumes the write requests will come in within chunks and in streaming mode
-type ChunkedStreamWriter struct {
-	activeChunks     map[LogicChunkIndex]*MemChunk
-	activeChunksLock sync.Mutex
-	ChunkSize        int64
-	saveToStorageFn  SaveToStorageFunc
-	sync.Mutex
-}
-
 type MemChunk struct {
 	buf   []byte
 	usage *ChunkWrittenIntervalList
 }
-
-var _ = io.WriterAt(&ChunkedStreamWriter{})
-
-func NewChunkedStreamWriter(chunkSize int64) *ChunkedStreamWriter {
-	return &ChunkedStreamWriter{
-		ChunkSize:    chunkSize,
-		activeChunks: make(map[LogicChunkIndex]*MemChunk),
-	}
-}
-
-func (cw *ChunkedStreamWriter) SetSaveToStorageFunction(saveToStorageFn SaveToStorageFunc) {
-	cw.saveToStorageFn = saveToStorageFn
-}
-
-func (cw *ChunkedStreamWriter) WriteAt(p []byte, off int64) (n int, err error) {
-	cw.Lock()
-	defer cw.Unlock()
-
-	logicChunkIndex := LogicChunkIndex(off / cw.ChunkSize)
-	offsetRemainder := off % cw.ChunkSize
-
-	memChunk, found := cw.activeChunks[logicChunkIndex]
-	if !found {
-		memChunk = &MemChunk{
-			buf:   mem.Allocate(int(cw.ChunkSize)),
-			usage: newChunkWrittenIntervalList(),
-		}
-		cw.activeChunks[logicChunkIndex] = memChunk
-	}
-	n = copy(memChunk.buf[offsetRemainder:], p)
-	memChunk.usage.MarkWritten(offsetRemainder, offsetRemainder+int64(n))
-	if memChunk.usage.IsComplete(cw.ChunkSize) {
-		if cw.saveToStorageFn != nil {
-			cw.saveOneChunk(memChunk, logicChunkIndex)
-			delete(cw.activeChunks, logicChunkIndex)
-		}
-	}
-
-	return
-}
-
-func (cw *ChunkedStreamWriter) ReadDataAt(p []byte, off int64) (maxStop int64) {
-	cw.Lock()
-	defer cw.Unlock()
-
-	logicChunkIndex := LogicChunkIndex(off / cw.ChunkSize)
-	memChunkBaseOffset := int64(logicChunkIndex) * cw.ChunkSize
-	memChunk, found := cw.activeChunks[logicChunkIndex]
-	if !found {
-		return
-	}
-
-	for t := memChunk.usage.head.next; t != memChunk.usage.tail; t = t.next {
-		logicStart := max(off, int64(logicChunkIndex)*cw.ChunkSize+t.StartOffset)
-		logicStop := min(off+int64(len(p)), memChunkBaseOffset+t.stopOffset)
-		if logicStart < logicStop {
-			copy(p[logicStart-off:logicStop-off], memChunk.buf[logicStart-memChunkBaseOffset:logicStop-memChunkBaseOffset])
-			maxStop = max(maxStop, logicStop)
-		}
-	}
-	return
-}
-
-func (cw *ChunkedStreamWriter) FlushAll() {
-	cw.Lock()
-	defer cw.Unlock()
-	for logicChunkIndex, memChunk := range cw.activeChunks {
-		if cw.saveToStorageFn != nil {
-			cw.saveOneChunk(memChunk, logicChunkIndex)
-			delete(cw.activeChunks, logicChunkIndex)
-		}
-	}
-}
-
-func (cw *ChunkedStreamWriter) saveOneChunk(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) {
-	var referenceCounter = int32(memChunk.usage.size())
-	for t := memChunk.usage.head.next; t != memChunk.usage.tail; t = t.next {
-		reader := util.NewBytesReader(memChunk.buf[t.StartOffset:t.stopOffset])
-		cw.saveToStorageFn(reader, int64(logicChunkIndex)*cw.ChunkSize+t.StartOffset, t.Size(), func() {
-			atomic.AddInt32(&referenceCounter, -1)
-			if atomic.LoadInt32(&referenceCounter) == 0 {
-				mem.Free(memChunk.buf)
-			}
-		})
-	}
-}
-
-// Reset releases used resources
-func (cw *ChunkedStreamWriter) Reset() {
-	for t, memChunk := range cw.activeChunks {
-		mem.Free(memChunk.buf)
-		delete(cw.activeChunks, t)
-	}
-}

weed/filesys/page_writer/chunked_stream_writer_test.go

@@ -1,33 +0,0 @@
-package page_writer
-
-import (
-	"github.com/stretchr/testify/assert"
-	"os"
-	"testing"
-)
-
-func TestWriteChunkedStream(t *testing.T) {
-	x := NewChunkedStreamWriter(20)
-	defer x.Reset()
-	y := NewChunkedFileWriter(os.TempDir(), 12)
-	defer y.Reset()
-
-	batchSize := 4
-	buf := make([]byte, batchSize)
-	for i := 0; i < 256; i++ {
-		for x := 0; x < batchSize; x++ {
-			buf[x] = byte(i)
-		}
-		x.WriteAt(buf, int64(i*batchSize))
-		y.WriteAt(buf, int64((255-i)*batchSize))
-	}
-
-	a := make([]byte, 1)
-	b := make([]byte, 1)
-	for i := 0; i < 256*batchSize; i++ {
-		x.ReadDataAt(a, int64(i))
-		y.ReadDataAt(b, int64(256*batchSize-1-i))
-		assert.Equal(t, a[0], b[0], "same read")
-	}
-
-}

weed/filesys/page_writer/dirty_page_interval.go

@@ -1,222 +0,0 @@
-package page_writer
-
-import (
-	"io"
-
-	"github.com/chrislusf/seaweedfs/weed/util"
-)
-
-type IntervalNode struct {
-	Data   []byte
-	Offset int64
-	Size   int64
-	Next   *IntervalNode
-}
-
-type IntervalLinkedList struct {
-	Head *IntervalNode
-	Tail *IntervalNode
-}
-
-type ContinuousIntervals struct {
-	lists []*IntervalLinkedList
-}
-
-func (list *IntervalLinkedList) Offset() int64 {
-	return list.Head.Offset
-}
-func (list *IntervalLinkedList) Size() int64 {
-	return list.Tail.Offset + list.Tail.Size - list.Head.Offset
-}
-func (list *IntervalLinkedList) addNodeToTail(node *IntervalNode) {
-	// glog.V(4).Infof("add to tail [%d,%d) + [%d,%d) => [%d,%d)", list.Head.Offset, list.Tail.Offset+list.Tail.Size, node.Offset, node.Offset+node.Size, list.Head.Offset, node.Offset+node.Size)
-	list.Tail.Next = node
-	list.Tail = node
-}
-func (list *IntervalLinkedList) addNodeToHead(node *IntervalNode) {
-	// glog.V(4).Infof("add to head [%d,%d) + [%d,%d) => [%d,%d)", node.Offset, node.Offset+node.Size, list.Head.Offset, list.Tail.Offset+list.Tail.Size, node.Offset, list.Tail.Offset+list.Tail.Size)
-	node.Next = list.Head
-	list.Head = node
-}
-
-func (list *IntervalLinkedList) ReadData(buf []byte, start, stop int64) {
-	t := list.Head
-	for {
-
-		nodeStart, nodeStop := max(start, t.Offset), min(stop, t.Offset+t.Size)
-		if nodeStart < nodeStop {
-			// glog.V(0).Infof("copying start=%d stop=%d t=[%d,%d) t.data=%d => bufSize=%d nodeStart=%d, nodeStop=%d", start, stop, t.Offset, t.Offset+t.Size, len(t.Data), len(buf), nodeStart, nodeStop)
-			copy(buf[nodeStart-start:], t.Data[nodeStart-t.Offset:nodeStop-t.Offset])
-		}
-
-		if t.Next == nil {
-			break
-		}
-		t = t.Next
-	}
-}
-
-func (c *ContinuousIntervals) TotalSize() (total int64) {
-	for _, list := range c.lists {
-		total += list.Size()
-	}
-	return
-}
-
-func subList(list *IntervalLinkedList, start, stop int64) *IntervalLinkedList {
-	var nodes []*IntervalNode
-	for t := list.Head; t != nil; t = t.Next {
-		nodeStart, nodeStop := max(start, t.Offset), min(stop, t.Offset+t.Size)
-		if nodeStart >= nodeStop {
-			// skip non overlapping IntervalNode
-			continue
-		}
-		nodes = append(nodes, &IntervalNode{
-			Data:   t.Data[nodeStart-t.Offset : nodeStop-t.Offset],
-			Offset: nodeStart,
-			Size:   nodeStop - nodeStart,
-			Next:   nil,
-		})
-	}
-	for i := 1; i < len(nodes); i++ {
-		nodes[i-1].Next = nodes[i]
-	}
-	return &IntervalLinkedList{
-		Head: nodes[0],
-		Tail: nodes[len(nodes)-1],
-	}
-}
-
-func (c *ContinuousIntervals) AddInterval(data []byte, offset int64) {
-
-	interval := &IntervalNode{Data: data, Offset: offset, Size: int64(len(data))}
-
-	// append to the tail and return
-	if len(c.lists) == 1 {
-		lastSpan := c.lists[0]
-		if lastSpan.Tail.Offset+lastSpan.Tail.Size == offset {
-			lastSpan.addNodeToTail(interval)
-			return
-		}
-	}
-
-	var newLists []*IntervalLinkedList
-	for _, list := range c.lists {
-		// if list is to the left of new interval, add to the new list
-		if list.Tail.Offset+list.Tail.Size <= interval.Offset {
-			newLists = append(newLists, list)
-		}
-		// if list is to the right of new interval, add to the new list
-		if interval.Offset+interval.Size <= list.Head.Offset {
-			newLists = append(newLists, list)
-		}
-		// if new interval overwrite the right part of the list
-		if list.Head.Offset < interval.Offset && interval.Offset < list.Tail.Offset+list.Tail.Size {
-			// create a new list of the left part of existing list
-			newLists = append(newLists, subList(list, list.Offset(), interval.Offset))
-		}
-		// if new interval overwrite the left part of the list
-		if list.Head.Offset < interval.Offset+interval.Size && interval.Offset+interval.Size < list.Tail.Offset+list.Tail.Size {
-			// create a new list of the right part of existing list
-			newLists = append(newLists, subList(list, interval.Offset+interval.Size, list.Tail.Offset+list.Tail.Size))
-		}
-		// skip anything that is fully overwritten by the new interval
-	}
-
-	c.lists = newLists
-	// add the new interval to the lists, connecting neighbor lists
-	var prevList, nextList *IntervalLinkedList
-
-	for _, list := range c.lists {
-		if list.Head.Offset == interval.Offset+interval.Size {
-			nextList = list
-			break
-		}
-	}
-
-	for _, list := range c.lists {
-		if list.Head.Offset+list.Size() == offset {
-			list.addNodeToTail(interval)
-			prevList = list
-			break
-		}
-	}
-
-	if prevList != nil && nextList != nil {
-		// glog.V(4).Infof("connecting [%d,%d) + [%d,%d) => [%d,%d)", prevList.Head.Offset, prevList.Tail.Offset+prevList.Tail.Size, nextList.Head.Offset, nextList.Tail.Offset+nextList.Tail.Size, prevList.Head.Offset, nextList.Tail.Offset+nextList.Tail.Size)
-		prevList.Tail.Next = nextList.Head
-		prevList.Tail = nextList.Tail
-		c.removeList(nextList)
-	} else if nextList != nil {
-		// add to head was not done when checking
-		nextList.addNodeToHead(interval)
-	}
-	if prevList == nil && nextList == nil {
-		c.lists = append(c.lists, &IntervalLinkedList{
-			Head: interval,
-			Tail: interval,
-		})
-	}
-
-	return
-}
-
-func (c *ContinuousIntervals) RemoveLargestIntervalLinkedList() *IntervalLinkedList {
-	var maxSize int64
-	maxIndex := -1
-	for k, list := range c.lists {
-		if maxSize <= list.Size() {
-			maxSize = list.Size()
-			maxIndex = k
-		}
-	}
-	if maxSize <= 0 {
-		return nil
-	}
-
-	t := c.lists[maxIndex]
-	c.lists = append(c.lists[0:maxIndex], c.lists[maxIndex+1:]...)
-	return t
-
-}
-
-func (c *ContinuousIntervals) removeList(target *IntervalLinkedList) {
-	index := -1
-	for k, list := range c.lists {
-		if list.Offset() == target.Offset() {
-			index = k
-		}
-	}
-	if index < 0 {
-		return
-	}
-
-	c.lists = append(c.lists[0:index], c.lists[index+1:]...)
-
-}
-
-func (c *ContinuousIntervals) ReadDataAt(data []byte, startOffset int64) (maxStop int64) {
-	for _, list := range c.lists {
-		start := max(startOffset, list.Offset())
-		stop := min(startOffset+int64(len(data)), list.Offset()+list.Size())
-		if start < stop {
-			list.ReadData(data[start-startOffset:], start, stop)
-			maxStop = max(maxStop, stop)
-		}
-	}
-	return
-}
-
-func (l *IntervalLinkedList) ToReader() io.Reader {
-	var readers []io.Reader
-	t := l.Head
-	readers = append(readers, util.NewBytesReader(t.Data))
-	for t.Next != nil {
-		t = t.Next
-		readers = append(readers, util.NewBytesReader(t.Data))
-	}
-	if len(readers) == 1 {
-		return readers[0]
-	}
-	return io.MultiReader(readers...)
-}

weed/filesys/page_writer/dirty_page_interval_test.go

@@ -1,113 +0,0 @@
-package page_writer
-
-import (
-	"bytes"
-	"math/rand"
-	"testing"
-)
-
-func TestContinuousIntervals_AddIntervalAppend(t *testing.T) {
-
-	c := &ContinuousIntervals{}
-
-	// 25, 25, 25
-	c.AddInterval(getBytes(25, 3), 0)
-	//  _,  _, 23, 23, 23, 23
-	c.AddInterval(getBytes(23, 4), 2)
-
-	expectedData(t, c, 0, 25, 25, 23, 23, 23, 23)
-
-}
-
-func TestContinuousIntervals_AddIntervalInnerOverwrite(t *testing.T) {
-
-	c := &ContinuousIntervals{}
-
-	// 25, 25, 25, 25, 25
-	c.AddInterval(getBytes(25, 5), 0)
-	//  _,  _, 23, 23
-	c.AddInterval(getBytes(23, 2), 2)
-
-	expectedData(t, c, 0, 25, 25, 23, 23, 25)
-
-}
-
-func TestContinuousIntervals_AddIntervalFullOverwrite(t *testing.T) {
-
-	c := &ContinuousIntervals{}
-
-	// 1,
-	c.AddInterval(getBytes(1, 1), 0)
-	//  _,  2,
-	c.AddInterval(getBytes(2, 1), 1)
-	//  _,  _,  3,  3, 3
-	c.AddInterval(getBytes(3, 3), 2)
-	//  _,  _,  _,  4, 4, 4
-	c.AddInterval(getBytes(4, 3), 3)
-
-	expectedData(t, c, 0, 1, 2, 3, 4, 4, 4)
-
-}
-
-func TestContinuousIntervals_RealCase1(t *testing.T) {
-
-	c := &ContinuousIntervals{}
-
-	// 25,
-	c.AddInterval(getBytes(25, 1), 0)
-	//  _,  _,  _,  _, 23, 23
-	c.AddInterval(getBytes(23, 2), 4)
-	//  _,  _,  _, 24, 24, 24, 24
-	c.AddInterval(getBytes(24, 4), 3)
-
-	//  _,  22, 22
-	c.AddInterval(getBytes(22, 2), 1)
-
-	expectedData(t, c, 0, 25, 22, 22, 24, 24, 24, 24)
-
-}
-
-func TestRandomWrites(t *testing.T) {
-
-	c := &ContinuousIntervals{}
-
-	data := make([]byte, 1024)
-
-	for i := 0; i < 1024; i++ {
-
-		start, stop := rand.Intn(len(data)), rand.Intn(len(data))
-		if start > stop {
-			start, stop = stop, start
-		}
-
-		rand.Read(data[start : stop+1])
-
-		c.AddInterval(data[start:stop+1], int64(start))
-
-		expectedData(t, c, 0, data...)
-
-	}
-
-}
-
-func expectedData(t *testing.T, c *ContinuousIntervals, offset int, data ...byte) {
-	start, stop := int64(offset), int64(offset+len(data))
-	for _, list := range c.lists {
-		nodeStart, nodeStop := max(start, list.Head.Offset), min(stop, list.Head.Offset+list.Size())
-		if nodeStart < nodeStop {
-			buf := make([]byte, nodeStop-nodeStart)
-			list.ReadData(buf, nodeStart, nodeStop)
-			if bytes.Compare(buf, data[nodeStart-start:nodeStop-start]) != 0 {
-				t.Errorf("expected %v actual %v", data[nodeStart-start:nodeStop-start], buf)
-			}
-		}
-	}
-}
-
-func getBytes(content byte, length int) []byte {
-	data := make([]byte, length)
-	for i := 0; i < length; i++ {
-		data[i] = content
-	}
-	return data
-}

weed/filesys/page_writer/dirty_pages.go

@@ -6,6 +6,8 @@ type DirtyPages interface {
 	ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64)
 	GetStorageOptions() (collection, replication string)
 	Destroy()
+	LockForRead(startOffset, stopOffset int64)
+	UnlockForRead(startOffset, stopOffset int64)
 }
 
 func max(x, y int64) int64 {

weed/filesys/page_writer/upload_pipeline.go

@@ -0,0 +1,256 @@
+package page_writer
+
+import (
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/util/mem"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+type LogicChunkIndex int
+
+type UploadPipeline struct {
+	filepath             util.FullPath
+	ChunkSize            int64
+	writers              *util.LimitedConcurrentExecutor
+	writableChunks       map[LogicChunkIndex]*MemChunk
+	writableChunksLock   sync.Mutex
+	sealedChunks         map[LogicChunkIndex]*SealedChunk
+	sealedChunksLock     sync.Mutex
+	activeWriterCond     *sync.Cond
+	activeWriterCount    int32
+	activeReadChunks     map[LogicChunkIndex]int
+	activeReadChunksLock sync.Mutex
+	saveToStorageFn      SaveToStorageFunc
+}
+
+type SealedChunk struct {
+	chunk            *MemChunk
+	referenceCounter int // track uploading or reading processes
+}
+
+func (sc *SealedChunk) FreeReference(messageOnFree string) {
+	sc.referenceCounter--
+	if sc.referenceCounter == 0 {
+		glog.V(4).Infof("Free sealed chunk: %s", messageOnFree)
+		mem.Free(sc.chunk.buf)
+	}
+}
+
+func NewUploadPipeline(filepath util.FullPath, writers *util.LimitedConcurrentExecutor, chunkSize int64, saveToStorageFn SaveToStorageFunc) *UploadPipeline {
+	return &UploadPipeline{
+		ChunkSize:        chunkSize,
+		writableChunks:   make(map[LogicChunkIndex]*MemChunk),
+		sealedChunks:     make(map[LogicChunkIndex]*SealedChunk),
+		writers:          writers,
+		activeWriterCond: sync.NewCond(&sync.Mutex{}),
+		saveToStorageFn:  saveToStorageFn,
+		filepath:         filepath,
+		activeReadChunks: make(map[LogicChunkIndex]int),
+	}
+}
+
+func (cw *UploadPipeline) SaveDataAt(p []byte, off int64) (n int) {
+	cw.writableChunksLock.Lock()
+	defer cw.writableChunksLock.Unlock()
+
+	logicChunkIndex := LogicChunkIndex(off / cw.ChunkSize)
+	offsetRemainder := off % cw.ChunkSize
+
+	memChunk, found := cw.writableChunks[logicChunkIndex]
+	if !found {
+		memChunk = &MemChunk{
+			buf:   mem.Allocate(int(cw.ChunkSize)),
+			usage: newChunkWrittenIntervalList(),
+		}
+		cw.writableChunks[logicChunkIndex] = memChunk
+	}
+	n = copy(memChunk.buf[offsetRemainder:], p)
+	memChunk.usage.MarkWritten(offsetRemainder, offsetRemainder+int64(n))
+	cw.maybeMoveToSealed(memChunk, logicChunkIndex)
+
+	return
+}
+
+func (cw *UploadPipeline) MaybeReadDataAt(p []byte, off int64) (maxStop int64) {
+	logicChunkIndex := LogicChunkIndex(off / cw.ChunkSize)
+
+	// read from sealed chunks first
+	cw.sealedChunksLock.Lock()
+	sealedChunk, found := cw.sealedChunks[logicChunkIndex]
+	if found {
+		sealedChunk.referenceCounter++
+	}
+	cw.sealedChunksLock.Unlock()
+	if found {
+		maxStop = readMemChunk(sealedChunk.chunk, p, off, logicChunkIndex, cw.ChunkSize)
+		glog.V(4).Infof("%s read sealed memchunk [%d,%d)", cw.filepath, off, maxStop)
+		sealedChunk.FreeReference(fmt.Sprintf("%s finish reading chunk %d", cw.filepath, logicChunkIndex))
+	}
+
+	// read from writable chunks last
+	cw.writableChunksLock.Lock()
+	defer cw.writableChunksLock.Unlock()
+	writableChunk, found := cw.writableChunks[logicChunkIndex]
+	if !found {
+		return
+	}
+	writableMaxStop := readMemChunk(writableChunk, p, off, logicChunkIndex, cw.ChunkSize)
+	glog.V(4).Infof("%s read writable memchunk [%d,%d)", cw.filepath, off, writableMaxStop)
+	maxStop = max(maxStop, writableMaxStop)
+
+	return
+}
+
+func (cw *UploadPipeline) FlushAll() {
+	cw.writableChunksLock.Lock()
+	defer cw.writableChunksLock.Unlock()
+
+	for logicChunkIndex, memChunk := range cw.writableChunks {
+		cw.moveToSealed(memChunk, logicChunkIndex)
+	}
+
+	cw.waitForCurrentWritersToComplete()
+}
+
+func (cw *UploadPipeline) LockForRead(startOffset, stopOffset int64) {
+	startLogicChunkIndex := LogicChunkIndex(startOffset / cw.ChunkSize)
+	stopLogicChunkIndex := LogicChunkIndex(stopOffset / cw.ChunkSize)
+	if stopOffset%cw.ChunkSize > 0 {
+		stopLogicChunkIndex += 1
+	}
+	cw.activeReadChunksLock.Lock()
+	defer cw.activeReadChunksLock.Unlock()
+	for i := startLogicChunkIndex; i < stopLogicChunkIndex; i++ {
+		if count, found := cw.activeReadChunks[i]; found {
+			cw.activeReadChunks[i] = count + 1
+		} else {
+			cw.activeReadChunks[i] = 1
+		}
+	}
+}
+
+func (cw *UploadPipeline) UnlockForRead(startOffset, stopOffset int64) {
+	startLogicChunkIndex := LogicChunkIndex(startOffset / cw.ChunkSize)
+	stopLogicChunkIndex := LogicChunkIndex(stopOffset / cw.ChunkSize)
+	if stopOffset%cw.ChunkSize > 0 {
+		stopLogicChunkIndex += 1
+	}
+	cw.activeReadChunksLock.Lock()
+	defer cw.activeReadChunksLock.Unlock()
+	for i := startLogicChunkIndex; i < stopLogicChunkIndex; i++ {
+		if count, found := cw.activeReadChunks[i]; found {
+			if count == 1 {
+				delete(cw.activeReadChunks, i)
+			} else {
+				cw.activeReadChunks[i] = count - 1
+			}
+		}
+	}
+}
+
+func (cw *UploadPipeline) IsLocked(logicChunkIndex LogicChunkIndex) bool {
+	cw.activeReadChunksLock.Lock()
+	defer cw.activeReadChunksLock.Unlock()
+	if count, found := cw.activeReadChunks[logicChunkIndex]; found {
+		return count > 0
+	}
+	return false
+}
+
+func (cw *UploadPipeline) waitForCurrentWritersToComplete() {
+	cw.activeWriterCond.L.Lock()
+	t := int32(100)
+	for {
+		t = atomic.LoadInt32(&cw.activeWriterCount)
+		if t <= 0 {
+			break
+		}
+		glog.V(4).Infof("activeWriterCond is %d", t)
+		cw.activeWriterCond.Wait()
+	}
+	cw.activeWriterCond.L.Unlock()
+}
+
+func (cw *UploadPipeline) maybeMoveToSealed(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) {
+	if memChunk.usage.IsComplete(cw.ChunkSize) {
+		cw.moveToSealed(memChunk, logicChunkIndex)
+	}
+}
+
+func (cw *UploadPipeline) moveToSealed(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) {
+	atomic.AddInt32(&cw.activeWriterCount, 1)
+	glog.V(4).Infof("%s activeWriterCount %d ++> %d", cw.filepath, cw.activeWriterCount-1, cw.activeWriterCount)
+
+	cw.sealedChunksLock.Lock()
+
+	if oldMemChunk, found := cw.sealedChunks[logicChunkIndex]; found {
+		oldMemChunk.FreeReference(fmt.Sprintf("%s replace chunk %d", cw.filepath, logicChunkIndex))
+	}
+	sealedChunk := &SealedChunk{
+		chunk:            memChunk,
+		referenceCounter: 1, // default 1 is for uploading process
+	}
+	cw.sealedChunks[logicChunkIndex] = sealedChunk
+	delete(cw.writableChunks, logicChunkIndex)
+
+	cw.sealedChunksLock.Unlock()
+
+	cw.writers.Execute(func() {
+		// first add to the file chunks
+		cw.saveOneChunk(sealedChunk.chunk, logicChunkIndex)
+
+		// notify waiting process
+		atomic.AddInt32(&cw.activeWriterCount, -1)
+		glog.V(4).Infof("%s activeWriterCount %d --> %d", cw.filepath, cw.activeWriterCount+1, cw.activeWriterCount)
+		// Lock and Unlock are not required,
+		// but it may signal multiple times during one wakeup,
+		// and the waiting goroutine may miss some of them!
+		cw.activeWriterCond.L.Lock()
+		cw.activeWriterCond.Broadcast()
+		cw.activeWriterCond.L.Unlock()
+
+		// wait for readers
+		for cw.IsLocked(logicChunkIndex) {
+			time.Sleep(59 * time.Millisecond)
+		}
+
+		// then remove from sealed chunks
+		cw.sealedChunksLock.Lock()
+		defer cw.sealedChunksLock.Unlock()
+		delete(cw.sealedChunks, logicChunkIndex)
+		sealedChunk.FreeReference(fmt.Sprintf("%s finished uploading chunk %d", cw.filepath, logicChunkIndex))
+
+	})
+}
+
+func (cw *UploadPipeline) saveOneChunk(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) {
+	if cw.saveToStorageFn == nil {
+		return
+	}
+	for t := memChunk.usage.head.next; t != memChunk.usage.tail; t = t.next {
+		reader := util.NewBytesReader(memChunk.buf[t.StartOffset:t.stopOffset])
+		cw.saveToStorageFn(reader, int64(logicChunkIndex)*cw.ChunkSize+t.StartOffset, t.Size(), func() {
+		})
+	}
+}
+
+func readMemChunk(memChunk *MemChunk, p []byte, off int64, logicChunkIndex LogicChunkIndex, chunkSize int64) (maxStop int64) {
+	memChunkBaseOffset := int64(logicChunkIndex) * chunkSize
+	for t := memChunk.usage.head.next; t != memChunk.usage.tail; t = t.next {
+		logicStart := max(off, int64(logicChunkIndex)*chunkSize+t.StartOffset)
+		logicStop := min(off+int64(len(p)), memChunkBaseOffset+t.stopOffset)
+		if logicStart < logicStop {
+			copy(p[logicStart-off:logicStop-off], memChunk.buf[logicStart-memChunkBaseOffset:logicStop-memChunkBaseOffset])
+			maxStop = max(maxStop, logicStop)
+		}
+	}
+	return
+}
+
+func (p2 *UploadPipeline) Shutdown() {
+
+}

weed/filesys/page_writer/upload_pipeline_test.go

@@ -0,0 +1,47 @@
+package page_writer
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"testing"
+)
+
+func TestUploadPipeline(t *testing.T) {
+
+	uploadPipeline := NewUploadPipeline("", nil, 2*1024*1024, nil)
+
+	writeRange(uploadPipeline, 0, 131072)
+	writeRange(uploadPipeline, 131072, 262144)
+	writeRange(uploadPipeline, 262144, 1025536)
+
+	confirmRange(t, uploadPipeline, 0, 1025536)
+
+	writeRange(uploadPipeline, 1025536, 1296896)
+
+	confirmRange(t, uploadPipeline, 1025536, 1296896)
+
+	writeRange(uploadPipeline, 1296896, 2162688)
+
+	confirmRange(t, uploadPipeline, 1296896, 2162688)
+
+	confirmRange(t, uploadPipeline, 1296896, 2162688)
+}
+
+// startOff and stopOff must be divided by 4
+func writeRange(uploadPipeline *UploadPipeline, startOff, stopOff int64) {
+	p := make([]byte, 4)
+	for i := startOff / 4; i < stopOff/4; i += 4 {
+		util.Uint32toBytes(p, uint32(i))
+		uploadPipeline.SaveDataAt(p, i)
+	}
+}
+
+func confirmRange(t *testing.T, uploadPipeline *UploadPipeline, startOff, stopOff int64) {
+	p := make([]byte, 4)
+	for i := startOff; i < stopOff/4; i += 4 {
+		uploadPipeline.MaybeReadDataAt(p, i)
+		x := util.BytesToUint32(p)
+		if x != uint32(i) {
+			t.Errorf("expecting %d found %d at offset [%d,%d)", i, x, i, i+4)
+		}
+	}
+}

weed/filesys/permission.go

@@ -0,0 +1,63 @@
+package filesys
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/fuse"
+)
+
+func checkPermission(entry *filer_pb.Entry, uid, gid uint32, isWrite bool) error {
+	if uid == 0 || gid == 0 {
+		return nil
+	}
+	if entry == nil {
+		return nil
+	}
+	if entry.Attributes == nil {
+		return nil
+	}
+	attr := entry.Attributes
+	if attr.Uid == uid {
+		if isWrite {
+			if attr.FileMode&0200 > 0 {
+				return nil
+			} else {
+				return fuse.EPERM
+			}
+		} else {
+			if attr.FileMode&0400 > 0 {
+				return nil
+			} else {
+				return fuse.EPERM
+			}
+		}
+	} else if attr.Gid == gid {
+		if isWrite {
+			if attr.FileMode&0020 > 0 {
+				return nil
+			} else {
+				return fuse.EPERM
+			}
+		} else {
+			if attr.FileMode&0040 > 0 {
+				return nil
+			} else {
+				return fuse.EPERM
+			}
+		}
+	} else {
+		if isWrite {
+			if attr.FileMode&0002 > 0 {
+				return nil
+			} else {
+				return fuse.EPERM
+			}
+		} else {
+			if attr.FileMode&0004 > 0 {
+				return nil
+			} else {
+				return fuse.EPERM
+			}
+		}
+	}
+
+}

weed/filesys/wfs.go

@@ -109,15 +109,15 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		wfs.chunkCache = chunk_cache.NewTieredChunkCache(256, option.getUniqueCacheDir(), option.CacheSizeMB, 1024*1024)
 	}
 
-	wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.getUniqueCacheDir(), "meta"), util.FullPath(option.FilerMountRootPath), option.UidGidMapper, func(filePath util.FullPath) {
+	wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.getUniqueCacheDir(), "meta"), util.FullPath(option.FilerMountRootPath), option.UidGidMapper, func(filePath util.FullPath, entry *filer_pb.Entry) {
 
-		fsNode := NodeWithId(filePath.AsInode())
+		fsNode := NodeWithId(filePath.AsInode(entry.FileMode()))
 		if err := wfs.Server.InvalidateNodeData(fsNode); err != nil {
 			glog.V(4).Infof("InvalidateNodeData %s : %v", filePath, err)
 		}
 
 		dir, name := filePath.DirAndName()
-		parent := NodeWithId(util.FullPath(dir).AsInode())
+		parent := NodeWithId(util.FullPath(dir).AsInode(os.ModeDir))
 		if dir == option.FilerMountRootPath {
 			parent = NodeWithId(1)
 		}
@@ -179,6 +179,28 @@ func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHand
 	return
 }
 
+func (wfs *WFS) Fsync(file *File, header fuse.Header) error {
+
+	inodeId := file.Id()
+
+	wfs.handlesLock.Lock()
+	existingHandle, found := wfs.handles[inodeId]
+	wfs.handlesLock.Unlock()
+
+	if found && existingHandle != nil {
+
+		existingHandle.Lock()
+		defer existingHandle.Unlock()
+
+		if existingHandle.f.isOpen > 0 {
+			return existingHandle.doFlush(context.Background(), header)
+		}
+
+	}
+
+	return nil
+}
+
 func (wfs *WFS) ReleaseHandle(fullpath util.FullPath, handleId fuse.HandleID) {
 	wfs.handlesLock.Lock()
 	defer wfs.handlesLock.Unlock()

weed/filesys/xattr.go

@@ -3,6 +3,7 @@ package filesys
 import (
 	"context"
 	"strings"
+	"syscall"
 
 	"github.com/seaweedfs/fuse"
 
@@ -143,3 +144,10 @@ func (wfs *WFS) maybeLoadEntry(dir, name string) (entry *filer_pb.Entry, err err
 	}
 	return cachedEntry.ToProtoEntry(), cacheErr
 }
+
+func checkName(name string) error {
+	if len(name) >= 256 {
+		return syscall.ENAMETOOLONG
+	}
+	return nil
+}

weed/operation/upload_content.go

@@ -4,6 +4,10 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/security"
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"io"
 	"mime"
 	"mime/multipart"
@@ -12,11 +16,6 @@ import (
 	"path/filepath"
 	"strings"
 	"time"
-
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-	"github.com/chrislusf/seaweedfs/weed/security"
-	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 type UploadOption struct {
@@ -71,7 +70,7 @@ func init() {
 	}}
 }
 
-var fileNameEscaper = strings.NewReplacer(`\`, `\\`, `"`, `\"`)
+var fileNameEscaper = strings.NewReplacer(`\`, `\\`, `"`, `\"`, "\n", "")
 
 // Upload sends a POST request to a volume server to upload the content with adjustable compression level
 func UploadData(data []byte, option *UploadOption) (uploadResult *UploadResult, err error) {
@@ -217,7 +216,8 @@ func upload_content(fillBufferFunction func(w io.Writer) error, originalDataSize
 	defer PutBuffer(buf)
 	body_writer := multipart.NewWriter(buf)
 	h := make(textproto.MIMEHeader)
-	h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="file"; filename="%s"`, fileNameEscaper.Replace(option.Filename)))
+	filename := fileNameEscaper.Replace(option.Filename)
+	h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="file"; filename="%s"`, filename))
 	h.Set("Idempotency-Key", option.UploadUrl)
 	if option.MimeType == "" {
 		option.MimeType = mime.TypeByExtension(strings.ToLower(filepath.Ext(option.Filename)))

weed/pb/filer_pb/filer_pb_helper.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"strings"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -16,6 +17,13 @@ func (entry *Entry) IsInRemoteOnly() bool {
 	return len(entry.Chunks) == 0 && entry.RemoteEntry != nil && entry.RemoteEntry.RemoteSize > 0
 }
 
+func (entry *Entry) FileMode() (fileMode os.FileMode) {
+	if entry != nil && entry.Attributes != nil {
+		fileMode = os.FileMode(entry.Attributes.FileMode)
+	}
+	return
+}
+
 func ToFileIdObject(fileIdStr string) (*FileId, error) {
 	t, err := needle.ParseFileIdFromString(fileIdStr)
 	if err != nil {

weed/pb/filer_pb_tail.go

@@ -84,11 +84,11 @@ func AddOffsetFunc(processEventFn ProcessMetadataFunc, offsetInterval time.Durat
 		}
 		counter++
 		if lastWriteTime.Add(offsetInterval).Before(time.Now()) {
-			counter = 0
 			lastWriteTime = time.Now()
 			if err := offsetFunc(counter, resp.TsNs); err != nil {
 				return err
 			}
+			counter = 0
 		}
 		return nil
 	}

weed/s3api/auth_credentials.go

@@ -247,9 +247,9 @@ func (iam *IdentityAccessManagement) authRequest(r *http.Request, action Action)
 
 	glog.V(3).Infof("user name: %v actions: %v, action: %v", identity.Name, identity.Actions, action)
 
-	bucket, _ := xhttp.GetBucketAndObject(r)
+	bucket, object := xhttp.GetBucketAndObject(r)
 
-	if !identity.canDo(action, bucket) {
+	if !identity.canDo(action, bucket, object) {
 		return identity, s3err.ErrAccessDenied
 	}
 
@@ -307,7 +307,7 @@ func (iam *IdentityAccessManagement) authUser(r *http.Request) (*Identity, s3err
 	return identity, s3err.ErrNone
 }
 
-func (identity *Identity) canDo(action Action, bucket string) bool {
+func (identity *Identity) canDo(action Action, bucket string, objectKey string) bool {
 	if identity.isAdmin() {
 		return true
 	}
@@ -319,15 +319,17 @@ func (identity *Identity) canDo(action Action, bucket string) bool {
 	if bucket == "" {
 		return false
 	}
+	target := string(action) + ":" + bucket + objectKey
+	adminTarget := s3_constants.ACTION_ADMIN + ":" + bucket + objectKey
 	limitedByBucket := string(action) + ":" + bucket
 	adminLimitedByBucket := s3_constants.ACTION_ADMIN + ":" + bucket
 	for _, a := range identity.Actions {
 		act := string(a)
 		if strings.HasSuffix(act, "*") {
-			if strings.HasPrefix(limitedByBucket, act[:len(act)-1]) {
+			if strings.HasPrefix(target, act[:len(act)-1]) {
 				return true
 			}
-			if strings.HasPrefix(adminLimitedByBucket, act[:len(act)-1]) {
+			if strings.HasPrefix(adminTarget, act[:len(act)-1]) {
 				return true
 			}
 		} else {

weed/s3api/auth_credentials_test.go

@@ -2,6 +2,7 @@ package s3api
 
 import (
 	. "github.com/chrislusf/seaweedfs/weed/s3api/s3_constants"
+	"github.com/stretchr/testify/assert"
 	"testing"
 
 	"github.com/golang/protobuf/jsonpb"
@@ -67,3 +68,61 @@ func TestIdentityListFileFormat(t *testing.T) {
 	println(text)
 
 }
+
+func TestCanDo(t *testing.T) {
+	ident1 := &Identity{
+		Name: "anything",
+		Actions: []Action{
+			"Write:bucket1/a/b/c/*",
+			"Write:bucket1/a/b/other",
+		},
+	}
+	// object specific
+	assert.Equal(t, true, ident1.canDo(ACTION_WRITE, "bucket1", "/a/b/c/d.txt"))
+	assert.Equal(t, false, ident1.canDo(ACTION_WRITE, "bucket1", "/a/b/other/some"), "action without *")
+
+	// bucket specific
+	ident2 := &Identity{
+		Name: "anything",
+		Actions: []Action{
+			"Read:bucket1",
+			"Write:bucket1/*",
+		},
+	}
+	assert.Equal(t, true, ident2.canDo(ACTION_READ, "bucket1", "/a/b/c/d.txt"))
+	assert.Equal(t, true, ident2.canDo(ACTION_WRITE, "bucket1", "/a/b/c/d.txt"))
+	assert.Equal(t, false, ident2.canDo(ACTION_LIST, "bucket1", "/a/b/c/d.txt"))
+
+	// across buckets
+	ident3 := &Identity{
+		Name: "anything",
+		Actions: []Action{
+			"Read",
+			"Write",
+		},
+	}
+	assert.Equal(t, true, ident3.canDo(ACTION_READ, "bucket1", "/a/b/c/d.txt"))
+	assert.Equal(t, true, ident3.canDo(ACTION_WRITE, "bucket1", "/a/b/c/d.txt"))
+	assert.Equal(t, false, ident3.canDo(ACTION_LIST, "bucket1", "/a/b/other/some"))
+
+	// partial buckets
+	ident4 := &Identity{
+		Name: "anything",
+		Actions: []Action{
+			"Read:special_*",
+		},
+	}
+	assert.Equal(t, true, ident4.canDo(ACTION_READ, "special_bucket", "/a/b/c/d.txt"))
+	assert.Equal(t, false, ident4.canDo(ACTION_READ, "bucket1", "/a/b/c/d.txt"))
+
+	// admin buckets
+	ident5 := &Identity{
+		Name: "anything",
+		Actions: []Action{
+			"Admin:special_*",
+		},
+	}
+	assert.Equal(t, true, ident5.canDo(ACTION_READ, "special_bucket", "/a/b/c/d.txt"))
+	assert.Equal(t, true, ident5.canDo(ACTION_WRITE, "special_bucket", "/a/b/c/d.txt"))
+
+}

weed/s3api/chunked_reader_v4.go

@@ -25,6 +25,7 @@ import (
 	"encoding/hex"
 	"errors"
 	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
+	"github.com/chrislusf/seaweedfs/weed/s3api/s3_constants"
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 	"hash"
 	"io"
@@ -91,8 +92,8 @@ func (iam *IdentityAccessManagement) calculateSeedSignature(r *http.Request) (cr
 		return nil, "", "", time.Time{}, s3err.ErrInvalidAccessKeyID
 	}
 
-	bucket, _ := xhttp.GetBucketAndObject(r)
-	if !identity.canDo("Write", bucket) {
+	bucket, object := xhttp.GetBucketAndObject(r)
+	if !identity.canDo(s3_constants.ACTION_WRITE, bucket, object) {
 		errCode = s3err.ErrAccessDenied
 		return
 	}

weed/s3api/s3api_bucket_handlers.go

@@ -55,7 +55,7 @@ func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Reques
 	var buckets []*s3.Bucket
 	for _, entry := range entries {
 		if entry.IsDirectory {
-			if identity != nil && !identity.canDo(s3_constants.ACTION_LIST, entry.Name) {
+			if identity != nil && !identity.canDo(s3_constants.ACTION_LIST, entry.Name, "") {
 				continue
 			}
 			buckets = append(buckets, &s3.Bucket{

weed/s3api/s3api_object_copy_handlers.go

@@ -38,6 +38,10 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
 			return
 		}
+		if entry.IsDirectory {
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
+			return
+		}
 		entry.Extended = weed_server.SaveAmzMetaData(r, entry.Extended, isReplace(r))
 		err = s3a.touch(dir, name, entry)
 		if err != nil {

weed/server/filer_grpc_server_rename.go

@@ -72,6 +72,21 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest,
 		return fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
 	}
 
+	if oldEntry.IsDirectory() {
+		// follow https://pubs.opengroup.org/onlinepubs/000095399/functions/rename.html
+		targetDir := newParent.Child(req.NewName)
+		newEntry, err := fs.filer.FindEntry(ctx, targetDir)
+		if err == nil {
+			if !newEntry.IsDirectory() {
+				fs.filer.RollbackTransaction(ctx)
+				return fmt.Errorf("%s is not directory", targetDir)
+			}
+			if entries, _, _ := fs.filer.ListDirectoryEntries(context.Background(), targetDir, "", false, 1, "", "", ""); len(entries) > 0 {
+				return fmt.Errorf("%s is not empty", targetDir)
+			}
+		}
+	}
+
 	moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures)
 	if moveErr != nil {
 		fs.filer.RollbackTransaction(ctx)

weed/server/master_grpc_server_cluster.go

@@ -2,7 +2,10 @@ package weed_server
 
 import (
 	"context"
+	"github.com/chrislusf/seaweedfs/weed/cluster"
+	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+	"math/rand"
 )
 
 func (ms *MasterServer) ListClusterNodes(ctx context.Context, req *master_pb.ListClusterNodesRequest) (*master_pb.ListClusterNodesResponse, error) {
@@ -12,10 +15,26 @@ func (ms *MasterServer) ListClusterNodes(ctx context.Context, req *master_pb.Lis
 
 	for _, node := range clusterNodes {
 		resp.ClusterNodes = append(resp.ClusterNodes, &master_pb.ListClusterNodesResponse_ClusterNode{
-			Address: string(node.Address),
-			Version: node.Version,
+			Address:  string(node.Address),
+			Version:  node.Version,
 			IsLeader: ms.Cluster.IsOneLeader(node.Address),
 		})
 	}
 	return resp, nil
 }
+
+func (ms *MasterServer) GetOneFiler() pb.ServerAddress {
+
+	clusterNodes := ms.Cluster.ListClusterNode(cluster.FilerType)
+
+	var filers []pb.ServerAddress
+	for _, node := range clusterNodes {
+		if ms.Cluster.IsOneLeader(node.Address) {
+			filers = append(filers, node.Address)
+		}
+	}
+	if len(filers) > 0 {
+		return filers[rand.Intn(len(filers))]
+	}
+	return "localhost:8888"
+}

weed/server/master_server.go

@@ -208,7 +208,6 @@ func (ms *MasterServer) proxyToLeader(f http.HandlerFunc) http.HandlerFunc {
 }
 
 func (ms *MasterServer) startAdminScripts() {
-	var err error
 
 	v := util.GetViper()
 	adminScripts := v.GetString("master.maintenance.scripts")
@@ -220,9 +219,6 @@ func (ms *MasterServer) startAdminScripts() {
 	v.SetDefault("master.maintenance.sleep_minutes", 17)
 	sleepMinutes := v.GetInt("master.maintenance.sleep_minutes")
 
-	v.SetDefault("master.filer.default", "localhost:8888")
-	filerHostPort := v.GetString("master.filer.default")
-
 	scriptLines := strings.Split(adminScripts, "\n")
 	if !strings.Contains(adminScripts, "lock") {
 		scriptLines = append(append([]string{}, "lock"), scriptLines...)
@@ -235,14 +231,9 @@ func (ms *MasterServer) startAdminScripts() {
 	shellOptions.GrpcDialOption = security.LoadClientTLS(v, "grpc.master")
 	shellOptions.Masters = &masterAddress
 
-	shellOptions.FilerAddress = pb.ServerAddress(filerHostPort)
 	shellOptions.Directory = "/"
-	if err != nil {
-		glog.V(0).Infof("failed to parse master.filer.default = %s : %v\n", filerHostPort, err)
-		return
-	}
 
-	commandEnv := shell.NewCommandEnv(shellOptions)
+	commandEnv := shell.NewCommandEnv(&shellOptions)
 
 	reg, _ := regexp.Compile(`'.*?'|".*?"|\S+`)
 
@@ -254,6 +245,10 @@ func (ms *MasterServer) startAdminScripts() {
 		for {
 			time.Sleep(time.Duration(sleepMinutes) * time.Minute)
 			if ms.Topo.IsLeader() {
+				shellOptions.FilerAddress = ms.GetOneFiler()
+				if shellOptions.FilerAddress == "" {
+					continue
+				}
 				for _, line := range scriptLines {
 					for _, c := range strings.Split(line, ";") {
 						processEachCmd(reg, c, commandEnv)

weed/shell/command_volume_check_disk.go

@@ -27,7 +27,7 @@ func (c *commandVolumeCheckDisk) Name() string {
 }
 
 func (c *commandVolumeCheckDisk) Help() string {
-	return `check all replicated volumes to find and fix inconsistencies
+	return `check all replicated volumes to find and fix inconsistencies. It is optional and resource intensive.
 
 	How it works:
 	
@@ -68,9 +68,6 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 	fileCount := func(replica *VolumeReplica) uint64 {
 		return replica.info.FileCount - replica.info.DeleteCount
 	}
-	aDB, bDB := needle_map.NewMemDb(), needle_map.NewMemDb()
-	defer aDB.Close()
-	defer bDB.Close()
 
 	for _, replicas := range volumeReplicas {
 		sort.Slice(replicas, func(i, j int) bool {
@@ -90,7 +87,7 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 				continue
 			}
 
-			if err := c.syncTwoReplicas(aDB, bDB, a, verbose, writer, b, err, applyChanges, nonRepairThreshold); err != nil {
+			if err := c.syncTwoReplicas(a, b, *applyChanges, *nonRepairThreshold, *verbose, writer); err != nil {
 				fmt.Fprintf(writer, "sync volume %d on %s and %s: %v\n", a.info.Id, a.location.dataNode.Id, b.location.dataNode.Id, err)
 			}
 			replicas = replicas[1:]
@@ -100,33 +97,41 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 	return nil
 }
 
-func (c *commandVolumeCheckDisk) syncTwoReplicas(aDB *needle_map.MemDb, bDB *needle_map.MemDb, a *VolumeReplica, verbose *bool, writer io.Writer, b *VolumeReplica, err error, applyChanges *bool, nonRepairThreshold *float64) error {
+func (c *commandVolumeCheckDisk) syncTwoReplicas(a *VolumeReplica, b *VolumeReplica, applyChanges bool, nonRepairThreshold float64, verbose bool, writer io.Writer) (err error) {
 	aHasChanges, bHasChanges := true, true
 	for aHasChanges || bHasChanges {
-		// reset index db
-		aDB.Close()
-		bDB.Close()
-		aDB, bDB = needle_map.NewMemDb(), needle_map.NewMemDb()
-
-		// read index db
-		if err := c.readIndexDatabase(aDB, a.info.Collection, a.info.Id, pb.NewServerAddressFromDataNode(a.location.dataNode), *verbose, writer); err != nil {
-			return fmt.Errorf("readIndexDatabase %s volume %d: %v", a.location.dataNode, a.info.Id, err)
-		}
-		if err := c.readIndexDatabase(bDB, b.info.Collection, b.info.Id, pb.NewServerAddressFromDataNode(b.location.dataNode), *verbose, writer); err != nil {
-			return fmt.Errorf("readIndexDatabase %s volume %d: %v", b.location.dataNode, b.info.Id, err)
-		}
-
-		// find and make up the differences
-		if aHasChanges, err = c.doVolumeCheckDisk(bDB, aDB, b, a, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
-			return fmt.Errorf("doVolumeCheckDisk source:%s target:%s volume %d: %v", b.location.dataNode, a.location.dataNode, b.info.Id, err)
-		}
-		if bHasChanges, err = c.doVolumeCheckDisk(aDB, bDB, a, b, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
-			return fmt.Errorf("doVolumeCheckDisk source:%s target:%s volume %d: %v", a.location.dataNode, b.location.dataNode, a.info.Id, err)
+		if aHasChanges, bHasChanges, err = c.checkBoth(a, b, applyChanges, nonRepairThreshold, verbose, writer); err != nil {
+			return err
 		}
 	}
 	return nil
 }
 
+func (c *commandVolumeCheckDisk) checkBoth(a *VolumeReplica, b *VolumeReplica, applyChanges bool, nonRepairThreshold float64, verbose bool, writer io.Writer) (aHasChanges bool, bHasChanges bool, err error) {
+	aDB, bDB := needle_map.NewMemDb(), needle_map.NewMemDb()
+	defer func() {
+		aDB.Close()
+		bDB.Close()
+	}()
+
+	// read index db
+	if err = c.readIndexDatabase(aDB, a.info.Collection, a.info.Id, pb.NewServerAddressFromDataNode(a.location.dataNode), verbose, writer); err != nil {
+		return true, true, fmt.Errorf("readIndexDatabase %s volume %d: %v", a.location.dataNode, a.info.Id, err)
+	}
+	if err := c.readIndexDatabase(bDB, b.info.Collection, b.info.Id, pb.NewServerAddressFromDataNode(b.location.dataNode), verbose, writer); err != nil {
+		return true, true, fmt.Errorf("readIndexDatabase %s volume %d: %v", b.location.dataNode, b.info.Id, err)
+	}
+
+	// find and make up the differences
+	if aHasChanges, err = c.doVolumeCheckDisk(bDB, aDB, b, a, verbose, writer, applyChanges, nonRepairThreshold); err != nil {
+		return true, true, fmt.Errorf("doVolumeCheckDisk source:%s target:%s volume %d: %v", b.location.dataNode, a.location.dataNode, b.info.Id, err)
+	}
+	if bHasChanges, err = c.doVolumeCheckDisk(aDB, bDB, a, b, verbose, writer, applyChanges, nonRepairThreshold); err != nil {
+		return true, true, fmt.Errorf("doVolumeCheckDisk source:%s target:%s volume %d: %v", a.location.dataNode, b.location.dataNode, a.info.Id, err)
+	}
+	return
+}
+
 func (c *commandVolumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64) (hasChanges bool, err error) {
 
 	// find missing keys

weed/shell/command_volume_fsck.go

@@ -215,6 +215,10 @@ func (c *commandVolumeFsck) findExtraChunksInVolumeServers(volumeIdToVInfo map[u
 				fmt.Fprintf(writer, "Skip purging for Erasure Coded volume %d.\n", volumeId)
 				continue
 			}
+			if vinfo.isReadOnly {
+				fmt.Fprintf(writer, "Skip purging for read only volume %d.\n", volumeId)
+				continue
+			}
 			if inUseCount == 0 {
 				if err := deleteVolume(c.env.option.GrpcDialOption, needle.VolumeId(volumeId), vinfo.server); err != nil {
 					return fmt.Errorf("delete volume %d: %v", volumeId, err)
@@ -440,6 +444,7 @@ type VInfo struct {
 	server     pb.ServerAddress
 	collection string
 	isEcVolume bool
+	isReadOnly bool
 }
 
 func (c *commandVolumeFsck) collectVolumeIds(commandEnv *CommandEnv, verbose bool, writer io.Writer) (volumeIdToServer map[uint32]VInfo, err error) {
@@ -462,6 +467,7 @@ func (c *commandVolumeFsck) collectVolumeIds(commandEnv *CommandEnv, verbose boo
 					server:     pb.NewServerAddressFromDataNode(t),
 					collection: vi.Collection,
 					isEcVolume: false,
+					isReadOnly: vi.ReadOnly,
 				}
 			}
 			for _, ecShardInfo := range diskInfo.EcShardInfos {
@@ -469,6 +475,7 @@ func (c *commandVolumeFsck) collectVolumeIds(commandEnv *CommandEnv, verbose boo
 					server:     pb.NewServerAddressFromDataNode(t),
 					collection: ecShardInfo.Collection,
 					isEcVolume: true,
+					isReadOnly: true,
 				}
 			}
 		}

weed/shell/commands.go

@@ -29,7 +29,7 @@ type ShellOptions struct {
 type CommandEnv struct {
 	env          map[string]string
 	MasterClient *wdclient.MasterClient
-	option       ShellOptions
+	option       *ShellOptions
 	locker       *exclusive_locks.ExclusiveLocker
 }
 
@@ -43,7 +43,7 @@ var (
 	Commands = []command{}
 )
 
-func NewCommandEnv(options ShellOptions) *CommandEnv {
+func NewCommandEnv(options *ShellOptions) *CommandEnv {
 	ce := &CommandEnv{
 		env:          make(map[string]string),
 		MasterClient: wdclient.NewMasterClient(options.GrpcDialOption, pb.AdminShellClient, "", "", pb.ServerAddresses(*options.Masters).ToAddresses()),

weed/shell/shell_liner.go

@@ -46,7 +46,7 @@ func RunShell(options ShellOptions) {
 
 	reg, _ := regexp.Compile(`'.*?'|".*?"|\S+`)
 
-	commandEnv := NewCommandEnv(options)
+	commandEnv := NewCommandEnv(&options)
 
 	go commandEnv.MasterClient.KeepConnectedToMaster()
 	commandEnv.MasterClient.WaitUntilConnected()

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.84)
+	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.86)
 	VERSION        = sizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""
 )

weed/util/fullpath.go

@@ -1,6 +1,7 @@
 package util
 
 import (
+	"os"
 	"path/filepath"
 	"strings"
 )
@@ -41,8 +42,31 @@ func (fp FullPath) Child(name string) FullPath {
 	return FullPath(dir + "/" + noPrefix)
 }
 
-func (fp FullPath) AsInode() uint64 {
-	return uint64(HashStringToLong(string(fp)))
+// AsInode an in-memory only inode representation
+func (fp FullPath) AsInode(fileMode os.FileMode) uint64 {
+	inode := uint64(HashStringToLong(string(fp)))
+	inode = inode - inode%16
+	if fileMode == 0 {
+	} else if fileMode&os.ModeDir > 0 {
+		inode += 1
+	} else if fileMode&os.ModeSymlink > 0 {
+		inode += 2
+	} else if fileMode&os.ModeDevice > 0 {
+		if fileMode&os.ModeCharDevice > 0 {
+			inode += 6
+		} else {
+			inode += 3
+		}
+	} else if fileMode&os.ModeNamedPipe > 0 {
+		inode += 4
+	} else if fileMode&os.ModeSocket > 0 {
+		inode += 5
+	} else if fileMode&os.ModeCharDevice > 0 {
+		inode += 6
+	} else if fileMode&os.ModeIrregular > 0 {
+		inode += 7
+	}
+	return inode
 }
 
 // split, but skipping the root