2.37

thoudsands → thousands

README.md

@@ -71,7 +71,7 @@ Table of Contents
 * Download the latest binary from https://github.com/chrislusf/seaweedfs/releases and unzip a single binary file `weed` or `weed.exe`
 * Run `weed server -dir=/some/data/dir -s3` to start one master, one volume server, one filer, and one S3 gateway.
 
-Also, to increase capacity, just add more volume servers by running `weed volume -dir="/some/data/dir2" -mserver="<master_host>:9333" -port=8081` locally, or on a different machine, or on thoudsands of machines. That is it!
+Also, to increase capacity, just add more volume servers by running `weed volume -dir="/some/data/dir2" -mserver="<master_host>:9333" -port=8081` locally, or on a different machine, or on thousands of machines. That is it!
 
 ## Introduction ##
 

docker/compose/local-minio-gateway-compose.yml

@@ -0,0 +1,50 @@
+version: '2'
+
+services:
+  master:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 9333:9333
+      - 19333:19333
+    command: "master -ip=master -volumeSizeLimitMB=1024"
+  volume:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8080:8080
+      - 18080:18080
+    command: "volume -mserver=master:9333 -port=8080 -ip=volume -max=0 -preStopSeconds=1"
+    depends_on:
+      - master
+  s3:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8888:8888
+      - 18888:18888
+      - 8333:8333
+    command: '-v 1 filer -master="master:9333" -s3 -s3.config=/etc/seaweedfs/s3.json -s3.port=8333'
+    volumes:
+      - ./s3.json:/etc/seaweedfs/s3.json
+    depends_on:
+      - master
+      - volume
+  minio-gateway-s3:
+    image: minio/minio
+    ports:
+      - 9000:9000
+    command: 'minio gateway s3 http://s3:8333'
+    restart: on-failure
+    environment:
+      MINIO_ACCESS_KEY: "some_access_key1"
+      MINIO_SECRET_KEY: "some_secret_key1"
+    depends_on:
+      - s3
+  minio-warp:
+    image: minio/warp
+    command: 'mixed --duration=5m --obj.size=3mb --autoterm'
+    restart: on-failure
+    environment:
+      WARP_HOST: "minio-gateway-s3:9000"
+      WARP_ACCESS_KEY: "some_access_key1"
+      WARP_SECRET_KEY: "some_secret_key1"
+    depends_on:
+      - minio-gateway-s3

docker/compose/local-registry-compose.yml

@@ -15,24 +15,18 @@ services:
     command: "volume -mserver=master:9333 -port=8080 -ip=volume -max=0 -preStopSeconds=1"
     depends_on:
       - master
-  filer:
+  s3:
     image: chrislusf/seaweedfs:local
     ports:
       - 8888:8888
       - 18888:18888
-    command: 'filer -master="master:9333"'
-    depends_on:
-      - master
-      - volume
-  s3:
-    image: chrislusf/seaweedfs:local
-    ports:
       - 8333:8333
-    command: '-v 9 s3 -filer="filer:8888"'
+    command: '-v 9 filer -master="master:9333" -s3 -s3.config=/etc/seaweedfs/s3.json -s3.port=8333'
+    volumes:
+      - ./s3.json:/etc/seaweedfs/s3.json
     depends_on:
       - master
       - volume
-      - filer
   minio:
     image: minio/minio
     ports:

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/Shopify/sarama v1.23.1
 	github.com/aws/aws-sdk-go v1.34.30
 	github.com/buraksezer/consistent v0.0.0-20191006190839-693edf70fd72
+	github.com/bwmarrin/snowflake v0.3.0
 	github.com/cespare/xxhash v1.1.0
 	github.com/chrislusf/raft v1.0.4
 	github.com/coreos/go-semver v0.3.0 // indirect

go.sum

@@ -141,6 +141,8 @@ github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4Yn
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
 github.com/buraksezer/consistent v0.0.0-20191006190839-693edf70fd72 h1:fUmDBbSvv1uOzo/t8WaxZMVb7BxJ8JECo5lGoR9c5bA=
 github.com/buraksezer/consistent v0.0.0-20191006190839-693edf70fd72/go.mod h1:OEE5igu/CDjGegM1Jn6ZMo7R6LlV/JChAkjfQQIRLpg=
+github.com/bwmarrin/snowflake v0.3.0 h1:xm67bEhkKh6ij1790JB83OujPR5CzNe8QuQqAgISZN0=
+github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/E9WsDpxqwE=
 github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ=
 github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/census-instrumentation/opencensus-proto v0.2.0 h1:LzQXZOgg4CQfE6bFvXGM30YZL1WW/M337pXml+GrcZ4=

k8s/seaweedfs/Chart.yaml

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

k8s/seaweedfs/values.yaml

@@ -4,7 +4,7 @@ global:
   registry: ""
   repository: ""
   imageName: chrislusf/seaweedfs
-  # imageTag: "2.35" - started using {.Chart.appVersion}
+  # imageTag: "2.37" - started using {.Chart.appVersion}
   imagePullPolicy: IfNotPresent
   imagePullSecrets: imagepullsecret
   restartPolicy: Always

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.2</version>
+    <version>1.6.4</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -17,7 +17,7 @@
         <protobuf.version>3.9.1</protobuf.version>
         <!-- follow https://github.com/grpc/grpc-java -->
         <grpc.version>1.23.0</grpc.version>
-        <guava.version>28.0-jre</guava.version>
+        <guava.version>30.0-jre</guava.version>
     </properties>
 
     <dependencies>

other/java/client/pom.xml.deploy

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

other/java/client/pom_debug.xml

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

other/java/client/src/main/java/seaweedfs/client/ByteBufferPool.java

@@ -3,6 +3,7 @@ package seaweedfs.client;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.nio.Buffer;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
@@ -34,7 +35,7 @@ public class ByteBufferPool {
     }
 
     public static synchronized void release(ByteBuffer obj) {
-        obj.clear();
+        ((Buffer)obj).clear();
         bufferList.add(0, obj);
     }
 

other/java/client/src/main/java/seaweedfs/client/FilerClient.java

@@ -94,7 +94,7 @@ public class FilerClient extends FilerGrpcClient {
             return true;
         }
         File pathFile = new File(path);
-        String parent = pathFile.getParent();
+        String parent = pathFile.getParent().replace('\\','/');
         String name = pathFile.getName();
 
         mkdirs(parent, mode, uid, gid, userName, groupNames);
@@ -115,11 +115,11 @@ public class FilerClient extends FilerGrpcClient {
     public boolean mv(String oldPath, String newPath) {
 
         File oldPathFile = new File(oldPath);
-        String oldParent = oldPathFile.getParent();
+        String oldParent = oldPathFile.getParent().replace('\\','/');
         String oldName = oldPathFile.getName();
 
         File newPathFile = new File(newPath);
-        String newParent = newPathFile.getParent();
+        String newParent = newPathFile.getParent().replace('\\','/');
         String newName = newPathFile.getName();
 
         return atomicRenameEntry(oldParent, oldName, newParent, newName);
@@ -129,7 +129,7 @@ public class FilerClient extends FilerGrpcClient {
     public boolean rm(String path, boolean isRecursive, boolean ignoreRecusiveError) {
 
         File pathFile = new File(path);
-        String parent = pathFile.getParent();
+        String parent = pathFile.getParent().replace('\\','/');
         String name = pathFile.getName();
 
         return deleteEntry(
@@ -148,7 +148,7 @@ public class FilerClient extends FilerGrpcClient {
     public boolean touch(String path, int mode, int uid, int gid, String userName, String[] groupNames) {
 
         File pathFile = new File(path);
-        String parent = pathFile.getParent();
+        String parent = pathFile.getParent().replace('\\','/');
         String name = pathFile.getName();
 
         FilerProto.Entry entry = lookupEntry(parent, name);

other/java/examples/pom.xml

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

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

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

other/java/hdfs2/pom.xml

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

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

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

other/java/hdfs3/pom.xml

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

test/random_access/pom.xml

@@ -8,7 +8,7 @@
     <version>1.0-SNAPSHOT</version>
 
     <properties>
-        <guava.version>28.0-jre</guava.version>
+        <guava.version>30.0-jre</guava.version>
     </properties>
 
     <dependencies>

weed/command/filer.go

@@ -47,6 +47,7 @@ type FilerOptions struct {
 	metricsHttpPort         *int
 	saveToFilerLimit        *int
 	defaultLevelDbDirectory *string
+	concurrentUploadLimitMB *int
 }
 
 func init() {
@@ -59,7 +60,7 @@ func init() {
 	f.publicPort = cmdFiler.Flag.Int("port.readonly", 0, "readonly port opened to public")
 	f.defaultReplicaPlacement = cmdFiler.Flag.String("defaultReplicaPlacement", "", "default replication type. If not specified, use master setting.")
 	f.disableDirListing = cmdFiler.Flag.Bool("disableDirListing", false, "turn off directory listing")
-	f.maxMB = cmdFiler.Flag.Int("maxMB", 32, "split files larger than the limit")
+	f.maxMB = cmdFiler.Flag.Int("maxMB", 4, "split files larger than the limit")
 	f.dirListingLimit = cmdFiler.Flag.Int("dirListLimit", 100000, "limit sub dir listing size")
 	f.dataCenter = cmdFiler.Flag.String("dataCenter", "", "prefer to read and write to volumes in this data center")
 	f.rack = cmdFiler.Flag.String("rack", "", "prefer to write to volumes in this rack")
@@ -69,6 +70,7 @@ func init() {
 	f.metricsHttpPort = cmdFiler.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
 	f.saveToFilerLimit = cmdFiler.Flag.Int("saveToFilerLimit", 0, "files smaller than this limit will be saved in filer store")
 	f.defaultLevelDbDirectory = cmdFiler.Flag.String("defaultStoreDir", ".", "if filer.toml is empty, use an embedded filer store in the directory")
+	f.concurrentUploadLimitMB = cmdFiler.Flag.Int("concurrentUploadLimitMB", 128, "limit total concurrent upload size")
 
 	// start s3 on filer
 	filerStartS3 = cmdFiler.Flag.Bool("s3", false, "whether to start S3 gateway")
@@ -159,21 +161,22 @@ func (fo *FilerOptions) startFiler() {
 	}
 
 	fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{
-		Masters:            strings.Split(*fo.masters, ","),
-		Collection:         *fo.collection,
-		DefaultReplication: *fo.defaultReplicaPlacement,
-		DisableDirListing:  *fo.disableDirListing,
-		MaxMB:              *fo.maxMB,
-		DirListingLimit:    *fo.dirListingLimit,
-		DataCenter:         *fo.dataCenter,
-		Rack:               *fo.rack,
-		DefaultLevelDbDir:  defaultLevelDbDirectory,
-		DisableHttp:        *fo.disableHttp,
-		Host:               *fo.ip,
-		Port:               uint32(*fo.port),
-		Cipher:             *fo.cipher,
-		SaveToFilerLimit:   *fo.saveToFilerLimit,
-		Filers:             peers,
+		Masters:               strings.Split(*fo.masters, ","),
+		Collection:            *fo.collection,
+		DefaultReplication:    *fo.defaultReplicaPlacement,
+		DisableDirListing:     *fo.disableDirListing,
+		MaxMB:                 *fo.maxMB,
+		DirListingLimit:       *fo.dirListingLimit,
+		DataCenter:            *fo.dataCenter,
+		Rack:                  *fo.rack,
+		DefaultLevelDbDir:     defaultLevelDbDirectory,
+		DisableHttp:           *fo.disableHttp,
+		Host:                  *fo.ip,
+		Port:                  uint32(*fo.port),
+		Cipher:                *fo.cipher,
+		SaveToFilerLimit:      int64(*fo.saveToFilerLimit),
+		Filers:                peers,
+		ConcurrentUploadLimit: int64(*fo.concurrentUploadLimitMB) * 1024 * 1024,
 	})
 	if nfs_err != nil {
 		glog.Fatalf("Filer startup error: %v", nfs_err)

weed/command/filer_copy.go

@@ -56,7 +56,7 @@ func init() {
 	copy.collection = cmdCopy.Flag.String("collection", "", "optional collection name")
 	copy.ttl = cmdCopy.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y")
 	copy.diskType = cmdCopy.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
-	copy.maxMB = cmdCopy.Flag.Int("maxMB", 32, "split files larger than the limit")
+	copy.maxMB = cmdCopy.Flag.Int("maxMB", 4, "split files larger than the limit")
 	copy.concurrenctFiles = cmdCopy.Flag.Int("c", 8, "concurrent file copy goroutines")
 	copy.concurrenctChunks = cmdCopy.Flag.Int("concurrentChunks", 8, "concurrent chunk copy goroutines for each file")
 }

weed/command/master.go

@@ -6,7 +6,6 @@ import (
 	"google.golang.org/grpc/reflection"
 	"net/http"
 	"os"
-	"runtime"
 	"sort"
 	"strconv"
 	"strings"
@@ -48,7 +47,7 @@ type MasterOptions struct {
 func init() {
 	cmdMaster.Run = runMaster // break init cycle
 	m.port = cmdMaster.Flag.Int("port", 9333, "http listen port")
-	m.ip = cmdMaster.Flag.String("ip", util.DetectedHostAddress(), "master <ip>|<server> address")
+	m.ip = cmdMaster.Flag.String("ip", util.DetectedHostAddress(), "master <ip>|<server> address, also used as identifier")
 	m.ipBind = cmdMaster.Flag.String("ip.bind", "", "ip address to bind to")
 	m.metaFolder = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store meta data")
 	m.peers = cmdMaster.Flag.String("peers", "", "all master nodes in comma separated ip:port list, example: 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095")
@@ -86,7 +85,6 @@ func runMaster(cmd *Command, args []string) bool {
 	util.LoadConfiguration("security", false)
 	util.LoadConfiguration("master", false)
 
-	runtime.GOMAXPROCS(runtime.NumCPU())
 	grace.SetupProfiling(*masterCpuProfile, *masterMemProfile)
 
 	parent, _ := util.FullPath(*m.metaFolder).DirAndName()

weed/command/mount.go

@@ -45,7 +45,7 @@ func init() {
 	mountOptions.diskType = cmdMount.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
 	mountOptions.ttlSec = cmdMount.Flag.Int("ttl", 0, "file ttl in seconds")
 	mountOptions.chunkSizeLimitMB = cmdMount.Flag.Int("chunkSizeLimitMB", 2, "local write buffer size, also chunk large files")
-	mountOptions.concurrentWriters = cmdMount.Flag.Int("concurrentWriters", 128, "limit concurrent goroutine writers if not 0")
+	mountOptions.concurrentWriters = cmdMount.Flag.Int("concurrentWriters", 32, "limit concurrent goroutine writers if not 0")
 	mountOptions.cacheDir = cmdMount.Flag.String("cacheDir", os.TempDir(), "local cache directory for file chunks and meta data")
 	mountOptions.cacheSizeMB = cmdMount.Flag.Int64("cacheCapacityMB", 1000, "local file chunk cache capacity in MB (0 will disable cache)")
 	mountOptions.dataCenter = cmdMount.Flag.String("dataCenter", "", "prefer to write to the data center")

weed/command/scaffold.go

@@ -103,9 +103,9 @@ dir = "./filerrdb"					# directory to store rocksdb files
 
 [mysql]  # or memsql, tidb
 # CREATE TABLE IF NOT EXISTS filemeta (
-#   dirhash     BIGINT         COMMENT 'first 64 bits of MD5 hash value of directory field',
-#   name        VARCHAR(1000)  COMMENT 'directory or file name',
-#   directory   TEXT           COMMENT 'full path to parent directory',
+#   dirhash     BIGINT               COMMENT 'first 64 bits of MD5 hash value of directory field',
+#   name        VARCHAR(1000) BINARY COMMENT 'directory or file name',
+#   directory   TEXT                 COMMENT 'full path to parent directory',
 #   meta        LONGBLOB,
 #   PRIMARY KEY (dirhash, name)
 # ) DEFAULT CHARSET=utf8;
@@ -120,13 +120,16 @@ connection_max_idle = 2
 connection_max_open = 100
 connection_max_lifetime_seconds = 0
 interpolateParams = false
+# if insert/upsert failing, you can disable upsert or update query syntax to match your RDBMS syntax:
+enableUpsert = true
+upsertQuery = """INSERT INTO ` + "`%s`" + ` (dirhash,name,directory,meta) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE meta = VALUES(meta)"""
 
 [mysql2]  # or memsql, tidb
 enabled = false
 createTable = """
   CREATE TABLE IF NOT EXISTS ` + "`%s`" + ` (
     dirhash BIGINT,
-    name VARCHAR(1000),
+    name VARCHAR(1000) BINARY,
     directory TEXT,
     meta LONGBLOB,
     PRIMARY KEY (dirhash, name)
@@ -141,6 +144,9 @@ connection_max_idle = 2
 connection_max_open = 100
 connection_max_lifetime_seconds = 0
 interpolateParams = false
+# if insert/upsert failing, you can disable upsert or update query syntax to match your RDBMS syntax:
+enableUpsert = true
+upsertQuery = """INSERT INTO ` + "`%s`" + ` (dirhash,name,directory,meta) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE meta = VALUES(meta)"""
 
 [postgres] # or cockroachdb, YugabyteDB
 # CREATE TABLE IF NOT EXISTS filemeta (
@@ -161,6 +167,9 @@ sslmode = "disable"
 connection_max_idle = 100
 connection_max_open = 100
 connection_max_lifetime_seconds = 0
+# if insert/upsert failing, you can disable upsert or update query syntax to match your RDBMS syntax:
+enableUpsert = true
+upsertQuery = """INSERT INTO "%[1]s" (dirhash,name,directory,meta) VALUES($1,$2,$3,$4) ON CONFLICT (dirhash,name) DO UPDATE SET meta = EXCLUDED.meta WHERE "%[1]s".meta != EXCLUDED.meta"""
 
 [postgres2]
 enabled = false
@@ -183,6 +192,9 @@ sslmode = "disable"
 connection_max_idle = 100
 connection_max_open = 100
 connection_max_lifetime_seconds = 0
+# if insert/upsert failing, you can disable upsert or update query syntax to match your RDBMS syntax:
+enableUpsert = true
+upsertQuery = """INSERT INTO "%[1]s" (dirhash,name,directory,meta) VALUES($1,$2,$3,$4) ON CONFLICT (dirhash,name) DO UPDATE SET meta = EXCLUDED.meta WHERE "%[1]s".meta != EXCLUDED.meta"""
 
 [cassandra]
 # CREATE TABLE filemeta (
@@ -506,7 +518,7 @@ default = "localhost:8888"    # used by maintenance scripts if the scripts needs
 
 
 [master.sequencer]
-type = "raft"     # Choose [raft|etcd] type for storing the file id sequence
+type = "raft"     # Choose [raft|etcd|snowflake] type for storing the file id sequence
 # when sequencer.type = etcd, set listen client urls of etcd cluster that store file id sequence
 # example : http://127.0.0.1:2379,http://127.0.0.1:2389
 sequencer_etcd_urls = "http://127.0.0.1:2379"

weed/command/server.go

@@ -2,9 +2,8 @@ package command
 
 import (
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/util/grace"
 	"os"
-	"runtime"
-	"runtime/pprof"
 	"strings"
 	"time"
 
@@ -16,6 +15,7 @@ import (
 
 type ServerOptions struct {
 	cpuprofile *string
+	memprofile *string
 	v          VolumeServerOptions
 }
 
@@ -49,7 +49,7 @@ var cmdServer = &Command{
 }
 
 var (
-	serverIp                  = cmdServer.Flag.String("ip", util.DetectedHostAddress(), "ip or server name")
+	serverIp                  = cmdServer.Flag.String("ip", util.DetectedHostAddress(), "ip or server name, also used as identifier")
 	serverBindIp              = cmdServer.Flag.String("ip.bind", "", "ip address to bind to")
 	serverTimeout             = cmdServer.Flag.Int("idleTimeout", 30, "connection idle seconds")
 	serverDataCenter          = cmdServer.Flag.String("dataCenter", "", "current volume server's data center name")
@@ -76,6 +76,7 @@ var (
 
 func init() {
 	serverOptions.cpuprofile = cmdServer.Flag.String("cpuprofile", "", "cpu profile output file")
+	serverOptions.memprofile = cmdServer.Flag.String("memprofile", "", "memory profile output file")
 
 	masterOptions.port = cmdServer.Flag.Int("master.port", 9333, "master server http listen port")
 	masterOptions.metaFolder = cmdServer.Flag.String("master.dir", "", "data directory to store meta data, default to same as -dir specified")
@@ -93,11 +94,12 @@ func init() {
 	filerOptions.publicPort = cmdServer.Flag.Int("filer.port.public", 0, "filer server public http listen port")
 	filerOptions.defaultReplicaPlacement = cmdServer.Flag.String("filer.defaultReplicaPlacement", "", "default replication type. If not specified, use master setting.")
 	filerOptions.disableDirListing = cmdServer.Flag.Bool("filer.disableDirListing", false, "turn off directory listing")
-	filerOptions.maxMB = cmdServer.Flag.Int("filer.maxMB", 32, "split files larger than the limit")
+	filerOptions.maxMB = cmdServer.Flag.Int("filer.maxMB", 4, "split files larger than the limit")
 	filerOptions.dirListingLimit = cmdServer.Flag.Int("filer.dirListLimit", 1000, "limit sub dir listing size")
 	filerOptions.cipher = cmdServer.Flag.Bool("filer.encryptVolumeData", false, "encrypt data on volume servers")
 	filerOptions.peers = cmdServer.Flag.String("filer.peers", "", "all filers sharing the same filer store in comma separated ip:port list")
 	filerOptions.saveToFilerLimit = cmdServer.Flag.Int("filer.saveToFilerLimit", 0, "Small files smaller than this limit can be cached in filer store.")
+	filerOptions.concurrentUploadLimitMB = cmdServer.Flag.Int("filer.concurrentUploadLimitMB", 64, "limit total concurrent upload size")
 
 	serverOptions.v.port = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port")
 	serverOptions.v.publicPort = cmdServer.Flag.Int("volume.port.public", 0, "volume server public port")
@@ -107,6 +109,7 @@ func init() {
 	serverOptions.v.readRedirect = cmdServer.Flag.Bool("volume.read.redirect", true, "Redirect moved or non-local volumes.")
 	serverOptions.v.compactionMBPerSecond = cmdServer.Flag.Int("volume.compactionMBps", 0, "limit compaction speed in mega bytes per second")
 	serverOptions.v.fileSizeLimitMB = cmdServer.Flag.Int("volume.fileSizeLimitMB", 256, "limit file size to avoid out of memory")
+	serverOptions.v.concurrentUploadLimitMB = cmdServer.Flag.Int("volume.concurrentUploadLimitMB", 64, "limit total concurrent upload size")
 	serverOptions.v.publicUrl = cmdServer.Flag.String("volume.publicUrl", "", "publicly accessible address")
 	serverOptions.v.preStopSeconds = cmdServer.Flag.Int("volume.preStopSeconds", 10, "number of seconds between stop send heartbeats and stop volume server")
 	serverOptions.v.pprof = cmdServer.Flag.Bool("volume.pprof", false, "enable pprof http handlers. precludes --memprofile and --cpuprofile")
@@ -138,14 +141,7 @@ func runServer(cmd *Command, args []string) bool {
 	util.LoadConfiguration("security", false)
 	util.LoadConfiguration("master", false)
 
-	if *serverOptions.cpuprofile != "" {
-		f, err := os.Create(*serverOptions.cpuprofile)
-		if err != nil {
-			glog.Fatal(err)
-		}
-		pprof.StartCPUProfile(f)
-		defer pprof.StopCPUProfile()
-	}
+	grace.SetupProfiling(*serverOptions.cpuprofile, *serverOptions.memprofile)
 
 	if *isStartingS3 {
 		*isStartingFiler = true
@@ -192,7 +188,6 @@ func runServer(cmd *Command, args []string) bool {
 	webdavOptions.filer = &filerAddress
 	msgBrokerOptions.filer = &filerAddress
 
-	runtime.GOMAXPROCS(runtime.NumCPU())
 	go stats_collect.StartMetricsServer(*serverMetricsHttpPort)
 
 	folders := strings.Split(*volumeDataFolders, ",")

weed/command/upload.go

@@ -43,7 +43,7 @@ func init() {
 	upload.dataCenter = cmdUpload.Flag.String("dataCenter", "", "optional data center name")
 	upload.diskType = cmdUpload.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
 	upload.ttl = cmdUpload.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y")
-	upload.maxMB = cmdUpload.Flag.Int("maxMB", 32, "split files larger than the limit")
+	upload.maxMB = cmdUpload.Flag.Int("maxMB", 4, "split files larger than the limit")
 	upload.usePublicUrl = cmdUpload.Flag.Bool("usePublicUrl", false, "upload to public url from volume server")
 }
 

weed/command/volume.go

@@ -6,7 +6,6 @@ import (
 	"net/http"
 	httppprof "net/http/pprof"
 	"os"
-	"runtime"
 	"runtime/pprof"
 	"strconv"
 	"strings"
@@ -36,31 +35,32 @@ var (
 )
 
 type VolumeServerOptions struct {
-	port                  *int
-	publicPort            *int
-	folders               []string
-	folderMaxLimits       []int
-	idxFolder             *string
-	ip                    *string
-	publicUrl             *string
-	bindIp                *string
-	masters               *string
-	idleConnectionTimeout *int
-	dataCenter            *string
-	rack                  *string
-	whiteList             []string
-	indexType             *string
-	diskType              *string
-	fixJpgOrientation     *bool
-	readRedirect          *bool
-	cpuProfile            *string
-	memProfile            *string
-	compactionMBPerSecond *int
-	fileSizeLimitMB       *int
-	minFreeSpacePercents  []float32
-	pprof                 *bool
-	preStopSeconds        *int
-	metricsHttpPort       *int
+	port                    *int
+	publicPort              *int
+	folders                 []string
+	folderMaxLimits         []int
+	idxFolder               *string
+	ip                      *string
+	publicUrl               *string
+	bindIp                  *string
+	masters                 *string
+	idleConnectionTimeout   *int
+	dataCenter              *string
+	rack                    *string
+	whiteList               []string
+	indexType               *string
+	diskType                *string
+	fixJpgOrientation       *bool
+	readRedirect            *bool
+	cpuProfile              *string
+	memProfile              *string
+	compactionMBPerSecond   *int
+	fileSizeLimitMB         *int
+	concurrentUploadLimitMB *int
+	minFreeSpacePercents    []float32
+	pprof                   *bool
+	preStopSeconds          *int
+	metricsHttpPort         *int
 	// pulseSeconds          *int
 	enableTcp *bool
 }
@@ -69,7 +69,7 @@ func init() {
 	cmdVolume.Run = runVolume // break init cycle
 	v.port = cmdVolume.Flag.Int("port", 8080, "http listen port")
 	v.publicPort = cmdVolume.Flag.Int("port.public", 0, "port opened to public")
-	v.ip = cmdVolume.Flag.String("ip", util.DetectedHostAddress(), "ip or server name")
+	v.ip = cmdVolume.Flag.String("ip", util.DetectedHostAddress(), "ip or server name, also used as identifier")
 	v.publicUrl = cmdVolume.Flag.String("publicUrl", "", "Publicly accessible address")
 	v.bindIp = cmdVolume.Flag.String("ip.bind", "", "ip address to bind to")
 	v.masters = cmdVolume.Flag.String("mserver", "localhost:9333", "comma-separated master servers")
@@ -86,6 +86,7 @@ func init() {
 	v.memProfile = cmdVolume.Flag.String("memprofile", "", "memory profile output file")
 	v.compactionMBPerSecond = cmdVolume.Flag.Int("compactionMBps", 0, "limit background compaction or copying speed in mega bytes per second")
 	v.fileSizeLimitMB = cmdVolume.Flag.Int("fileSizeLimitMB", 256, "limit file size to avoid out of memory")
+	v.concurrentUploadLimitMB = cmdVolume.Flag.Int("concurrentUploadLimitMB", 128, "limit total concurrent upload size")
 	v.pprof = cmdVolume.Flag.Bool("pprof", false, "enable pprof http handlers. precludes --memprofile and --cpuprofile")
 	v.metricsHttpPort = cmdVolume.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
 	v.idxFolder = cmdVolume.Flag.String("dir.idx", "", "directory to store .idx files")
@@ -111,8 +112,6 @@ func runVolume(cmd *Command, args []string) bool {
 
 	util.LoadConfiguration("security", false)
 
-	runtime.GOMAXPROCS(runtime.NumCPU())
-
 	// If --pprof is set we assume the caller wants to be able to collect
 	// cpu and memory profiles via go tool pprof
 	if !*v.pprof {
@@ -240,6 +239,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 		*v.fixJpgOrientation, *v.readRedirect,
 		*v.compactionMBPerSecond,
 		*v.fileSizeLimitMB,
+		int64(*v.concurrentUploadLimitMB)*1024*1024,
 	)
 	// starting grpc server
 	grpcS := v.startGrpcService(volumeServer)

weed/filer/abstract_sql/abstract_sql_store.go

@@ -13,15 +13,15 @@ import (
 )
 
 type SqlGenerator interface {
-	GetSqlInsert(bucket string) string
-	GetSqlUpdate(bucket string) string
-	GetSqlFind(bucket string) string
-	GetSqlDelete(bucket string) string
-	GetSqlDeleteFolderChildren(bucket string) string
-	GetSqlListExclusive(bucket string) string
-	GetSqlListInclusive(bucket string) string
-	GetSqlCreateTable(bucket string) string
-	GetSqlDropTable(bucket string) string
+	GetSqlInsert(tableName string) string
+	GetSqlUpdate(tableName string) string
+	GetSqlFind(tableName string) string
+	GetSqlDelete(tableName string) string
+	GetSqlDeleteFolderChildren(tableName string) string
+	GetSqlListExclusive(tableName string) string
+	GetSqlListInclusive(tableName string) string
+	GetSqlCreateTable(tableName string) string
+	GetSqlDropTable(tableName string) string
 }
 
 type AbstractSqlStore struct {
@@ -36,6 +36,8 @@ func (store *AbstractSqlStore) OnBucketCreation(bucket string) {
 	store.dbsLock.Lock()
 	defer store.dbsLock.Unlock()
 
+	store.CreateTable(context.Background(), bucket)
+
 	if store.dbs == nil {
 		return
 	}
@@ -45,6 +47,8 @@ func (store *AbstractSqlStore) OnBucketDeletion(bucket string) {
 	store.dbsLock.Lock()
 	defer store.dbsLock.Unlock()
 
+	store.deleteTable(context.Background(), bucket)
+
 	if store.dbs == nil {
 		return
 	}

weed/filer/filechunk_manifest.go

@@ -91,26 +91,26 @@ func fetchChunk(lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string,
 		glog.Errorf("operation LookupFileId %s failed, err: %v", fileId, err)
 		return nil, err
 	}
-	return retriedFetchChunkData(urlStrings, cipherKey, isGzipped, true, false, 0, 0)
+	return retriedFetchChunkData(urlStrings, cipherKey, isGzipped, true, 0, 0)
 }
 
-func retriedFetchChunkData(urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, isCheck bool, offset int64, size int) ([]byte, error) {
+func retriedFetchChunkData(urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, size int) ([]byte, error) {
 
 	var err error
-	var buffer bytes.Buffer
 	var shouldRetry bool
+	receivedData := make([]byte, 0, size)
 
 	for waitTime := time.Second; waitTime < util.RetryWaitTime; waitTime += waitTime / 2 {
 		for _, urlString := range urlStrings {
+			receivedData = receivedData[:0]
 			shouldRetry, err = util.ReadUrlAsStream(urlString+"?readDeleted=true", cipherKey, isGzipped, isFullChunk, offset, size, func(data []byte) {
-				buffer.Write(data)
+				receivedData = append(receivedData, data...)
 			})
 			if !shouldRetry {
 				break
 			}
 			if err != nil {
 				glog.V(0).Infof("read %s failed, err: %v", urlString, err)
-				buffer.Reset()
 			} else {
 				break
 			}
@@ -123,7 +123,8 @@ func retriedFetchChunkData(urlStrings []string, cipherKey []byte, isGzipped bool
 		}
 	}
 
-	return buffer.Bytes(), err
+	return receivedData, err
+
 }
 
 func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, inputChunks []*filer_pb.FileChunk) (chunks []*filer_pb.FileChunk, err error) {

weed/filer/filer.go

@@ -241,12 +241,12 @@ func (f *Filer) UpdateEntry(ctx context.Context, oldEntry, entry *Entry) (err er
 	if oldEntry != nil {
 		entry.Attr.Crtime = oldEntry.Attr.Crtime
 		if oldEntry.IsDirectory() && !entry.IsDirectory() {
-			glog.Errorf("existing %s is a directory", entry.FullPath)
-			return fmt.Errorf("existing %s is a directory", entry.FullPath)
+			glog.Errorf("existing %s is a directory", oldEntry.FullPath)
+			return fmt.Errorf("existing %s is a directory", oldEntry.FullPath)
 		}
 		if !oldEntry.IsDirectory() && entry.IsDirectory() {
-			glog.Errorf("existing %s is a file", entry.FullPath)
-			return fmt.Errorf("existing %s is a file", entry.FullPath)
+			glog.Errorf("existing %s is a file", oldEntry.FullPath)
+			return fmt.Errorf("existing %s is a file", oldEntry.FullPath)
 		}
 	}
 	return f.Store.UpdateEntry(ctx, entry)

weed/filer/mysql/mysql_sql_gen.go

@@ -2,6 +2,7 @@ package mysql
 
 import (
 	"fmt"
+
 	"github.com/chrislusf/seaweedfs/weed/filer/abstract_sql"
 	_ "github.com/go-sql-driver/mysql"
 )
@@ -9,44 +10,49 @@ import (
 type SqlGenMysql struct {
 	CreateTableSqlTemplate string
 	DropTableSqlTemplate   string
+	UpsertQueryTemplate    string
 }
 
 var (
 	_ = abstract_sql.SqlGenerator(&SqlGenMysql{})
 )
 
-func (gen *SqlGenMysql) GetSqlInsert(bucket string) string {
-	return fmt.Sprintf("INSERT INTO `%s` (dirhash,name,directory,meta) VALUES(?,?,?,?)", bucket)
+func (gen *SqlGenMysql) GetSqlInsert(tableName string) string {
+	if gen.UpsertQueryTemplate != "" {
+		return fmt.Sprintf(gen.UpsertQueryTemplate, tableName)
+	} else {
+		return fmt.Sprintf("INSERT INTO `%s` (dirhash,name,directory,meta) VALUES(?,?,?,?)", tableName)
+	}
 }
 
-func (gen *SqlGenMysql) GetSqlUpdate(bucket string) string {
-	return fmt.Sprintf("UPDATE `%s` SET meta=? WHERE dirhash=? AND name=? AND directory=?", bucket)
+func (gen *SqlGenMysql) GetSqlUpdate(tableName string) string {
+	return fmt.Sprintf("UPDATE `%s` SET meta=? WHERE dirhash=? AND name=? AND directory=?", tableName)
 }
 
-func (gen *SqlGenMysql) GetSqlFind(bucket string) string {
-	return fmt.Sprintf("SELECT meta FROM `%s` WHERE dirhash=? AND name=? AND directory=?", bucket)
+func (gen *SqlGenMysql) GetSqlFind(tableName string) string {
+	return fmt.Sprintf("SELECT meta FROM `%s` WHERE dirhash=? AND name=? AND directory=?", tableName)
 }
 
-func (gen *SqlGenMysql) GetSqlDelete(bucket string) string {
-	return fmt.Sprintf("DELETE FROM `%s` WHERE dirhash=? AND name=? AND directory=?", bucket)
+func (gen *SqlGenMysql) GetSqlDelete(tableName string) string {
+	return fmt.Sprintf("DELETE FROM `%s` WHERE dirhash=? AND name=? AND directory=?", tableName)
 }
 
-func (gen *SqlGenMysql) GetSqlDeleteFolderChildren(bucket string) string {
-	return fmt.Sprintf("DELETE FROM `%s` WHERE dirhash=? AND directory=?", bucket)
+func (gen *SqlGenMysql) GetSqlDeleteFolderChildren(tableName string) string {
+	return fmt.Sprintf("DELETE FROM `%s` WHERE dirhash=? AND directory=?", tableName)
 }
 
-func (gen *SqlGenMysql) GetSqlListExclusive(bucket string) string {
-	return fmt.Sprintf("SELECT NAME, meta FROM `%s` WHERE dirhash=? AND name>? AND directory=? AND name like ? ORDER BY NAME ASC LIMIT ?", bucket)
+func (gen *SqlGenMysql) GetSqlListExclusive(tableName string) string {
+	return fmt.Sprintf("SELECT NAME, meta FROM `%s` WHERE dirhash=? AND name>? AND directory=? AND name like ? ORDER BY NAME ASC LIMIT ?", tableName)
 }
 
-func (gen *SqlGenMysql) GetSqlListInclusive(bucket string) string {
-	return fmt.Sprintf("SELECT NAME, meta FROM `%s` WHERE dirhash=? AND name>=? AND directory=? AND name like ? ORDER BY NAME ASC LIMIT ?", bucket)
+func (gen *SqlGenMysql) GetSqlListInclusive(tableName string) string {
+	return fmt.Sprintf("SELECT NAME, meta FROM `%s` WHERE dirhash=? AND name>=? AND directory=? AND name like ? ORDER BY NAME ASC LIMIT ?", tableName)
 }
 
-func (gen *SqlGenMysql) GetSqlCreateTable(bucket string) string {
-	return fmt.Sprintf(gen.CreateTableSqlTemplate, bucket)
+func (gen *SqlGenMysql) GetSqlCreateTable(tableName string) string {
+	return fmt.Sprintf(gen.CreateTableSqlTemplate, tableName)
 }
 
-func (gen *SqlGenMysql) GetSqlDropTable(bucket string) string {
-	return fmt.Sprintf(gen.DropTableSqlTemplate, bucket)
+func (gen *SqlGenMysql) GetSqlDropTable(tableName string) string {
+	return fmt.Sprintf(gen.DropTableSqlTemplate, tableName)
 }

weed/filer/mysql/mysql_store.go

@@ -3,9 +3,10 @@ package mysql
 import (
 	"database/sql"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/filer"
 	"time"
 
+	"github.com/chrislusf/seaweedfs/weed/filer"
+
 	"github.com/chrislusf/seaweedfs/weed/filer/abstract_sql"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	_ "github.com/go-sql-driver/mysql"
@@ -29,6 +30,8 @@ func (store *MysqlStore) GetName() string {
 
 func (store *MysqlStore) Initialize(configuration util.Configuration, prefix string) (err error) {
 	return store.initialize(
+		configuration.GetString(prefix+"upsertQuery"),
+		configuration.GetBool(prefix+"enableUpsert"),
 		configuration.GetString(prefix+"username"),
 		configuration.GetString(prefix+"password"),
 		configuration.GetString(prefix+"hostname"),
@@ -41,13 +44,17 @@ func (store *MysqlStore) Initialize(configuration util.Configuration, prefix str
 	)
 }
 
-func (store *MysqlStore) initialize(user, password, hostname string, port int, database string, maxIdle, maxOpen,
+func (store *MysqlStore) initialize(upsertQuery string, enableUpsert bool, user, password, hostname string, port int, database string, maxIdle, maxOpen,
 	maxLifetimeSeconds int, interpolateParams bool) (err error) {
 
 	store.SupportBucketTable = false
+	if !enableUpsert {
+		upsertQuery = ""
+	}
 	store.SqlGenerator = &SqlGenMysql{
 		CreateTableSqlTemplate: "",
 		DropTableSqlTemplate:   "drop table `%s`",
+		UpsertQueryTemplate:    upsertQuery,
 	}
 
 	sqlUrl := fmt.Sprintf(CONNECTION_URL_PATTERN, user, password, hostname, port, database)

weed/filer/mysql2/mysql2_store.go

@@ -32,6 +32,8 @@ func (store *MysqlStore2) GetName() string {
 func (store *MysqlStore2) Initialize(configuration util.Configuration, prefix string) (err error) {
 	return store.initialize(
 		configuration.GetString(prefix+"createTable"),
+		configuration.GetString(prefix+"upsertQuery"),
+		configuration.GetBool(prefix+"enableUpsert"),
 		configuration.GetString(prefix+"username"),
 		configuration.GetString(prefix+"password"),
 		configuration.GetString(prefix+"hostname"),
@@ -44,13 +46,17 @@ func (store *MysqlStore2) Initialize(configuration util.Configuration, prefix st
 	)
 }
 
-func (store *MysqlStore2) initialize(createTable, user, password, hostname string, port int, database string, maxIdle, maxOpen,
+func (store *MysqlStore2) initialize(createTable, upsertQuery string, enableUpsert bool, user, password, hostname string, port int, database string, maxIdle, maxOpen,
 	maxLifetimeSeconds int, interpolateParams bool) (err error) {
 
 	store.SupportBucketTable = true
+	if !enableUpsert {
+		upsertQuery = ""
+	}
 	store.SqlGenerator = &mysql.SqlGenMysql{
 		CreateTableSqlTemplate: createTable,
 		DropTableSqlTemplate:   "drop table `%s`",
+		UpsertQueryTemplate:    upsertQuery,
 	}
 
 	sqlUrl := fmt.Sprintf(CONNECTION_URL_PATTERN, user, password, hostname, port, database)

weed/filer/postgres/postgres_sql_gen.go

@@ -10,44 +10,49 @@ import (
 type SqlGenPostgres struct {
 	CreateTableSqlTemplate string
 	DropTableSqlTemplate   string
+	UpsertQueryTemplate    string
 }
 
 var (
 	_ = abstract_sql.SqlGenerator(&SqlGenPostgres{})
 )
 
-func (gen *SqlGenPostgres) GetSqlInsert(bucket string) string {
-	return fmt.Sprintf(`INSERT INTO "%s" (dirhash,name,directory,meta) VALUES($1,$2,$3,$4)`, bucket)
+func (gen *SqlGenPostgres) GetSqlInsert(tableName string) string {
+	if gen.UpsertQueryTemplate != "" {
+		return fmt.Sprintf(gen.UpsertQueryTemplate, tableName)
+	} else {
+		return fmt.Sprintf(`INSERT INTO "%s" (dirhash,name,directory,meta) VALUES($1,$2,$3,$4)`, tableName)
+	}
 }
 
-func (gen *SqlGenPostgres) GetSqlUpdate(bucket string) string {
-	return fmt.Sprintf(`UPDATE "%s" SET meta=$1 WHERE dirhash=$2 AND name=$3 AND directory=$4`, bucket)
+func (gen *SqlGenPostgres) GetSqlUpdate(tableName string) string {
+	return fmt.Sprintf(`UPDATE "%s" SET meta=$1 WHERE dirhash=$2 AND name=$3 AND directory=$4`, tableName)
 }
 
-func (gen *SqlGenPostgres) GetSqlFind(bucket string) string {
-	return fmt.Sprintf(`SELECT meta FROM "%s" WHERE dirhash=$1 AND name=$2 AND directory=$3`, bucket)
+func (gen *SqlGenPostgres) GetSqlFind(tableName string) string {
+	return fmt.Sprintf(`SELECT meta FROM "%s" WHERE dirhash=$1 AND name=$2 AND directory=$3`, tableName)
 }
 
-func (gen *SqlGenPostgres) GetSqlDelete(bucket string) string {
-	return fmt.Sprintf(`DELETE FROM "%s" WHERE dirhash=$1 AND name=$2 AND directory=$3`, bucket)
+func (gen *SqlGenPostgres) GetSqlDelete(tableName string) string {
+	return fmt.Sprintf(`DELETE FROM "%s" WHERE dirhash=$1 AND name=$2 AND directory=$3`, tableName)
 }
 
-func (gen *SqlGenPostgres) GetSqlDeleteFolderChildren(bucket string) string {
-	return fmt.Sprintf(`DELETE FROM "%s" WHERE dirhash=$1 AND directory=$2`, bucket)
+func (gen *SqlGenPostgres) GetSqlDeleteFolderChildren(tableName string) string {
+	return fmt.Sprintf(`DELETE FROM "%s" WHERE dirhash=$1 AND directory=$2`, tableName)
 }
 
-func (gen *SqlGenPostgres) GetSqlListExclusive(bucket string) string {
-	return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND name>$2 AND directory=$3 AND name like $4 ORDER BY NAME ASC LIMIT $5`, bucket)
+func (gen *SqlGenPostgres) GetSqlListExclusive(tableName string) string {
+	return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND name>$2 AND directory=$3 AND name like $4 ORDER BY NAME ASC LIMIT $5`, tableName)
 }
 
-func (gen *SqlGenPostgres) GetSqlListInclusive(bucket string) string {
-	return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND name>=$2 AND directory=$3 AND name like $4 ORDER BY NAME ASC LIMIT $5`, bucket)
+func (gen *SqlGenPostgres) GetSqlListInclusive(tableName string) string {
+	return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND name>=$2 AND directory=$3 AND name like $4 ORDER BY NAME ASC LIMIT $5`, tableName)
 }
 
-func (gen *SqlGenPostgres) GetSqlCreateTable(bucket string) string {
-	return fmt.Sprintf(gen.CreateTableSqlTemplate, bucket)
+func (gen *SqlGenPostgres) GetSqlCreateTable(tableName string) string {
+	return fmt.Sprintf(gen.CreateTableSqlTemplate, tableName)
 }
 
-func (gen *SqlGenPostgres) GetSqlDropTable(bucket string) string {
-	return fmt.Sprintf(gen.DropTableSqlTemplate, bucket)
+func (gen *SqlGenPostgres) GetSqlDropTable(tableName string) string {
+	return fmt.Sprintf(gen.DropTableSqlTemplate, tableName)
 }

weed/filer/postgres/postgres_store.go

@@ -29,6 +29,8 @@ func (store *PostgresStore) GetName() string {
 
 func (store *PostgresStore) Initialize(configuration util.Configuration, prefix string) (err error) {
 	return store.initialize(
+		configuration.GetString(prefix+"upsertQuery"),
+		configuration.GetBool(prefix+"enableUpsert"),
 		configuration.GetString(prefix+"username"),
 		configuration.GetString(prefix+"password"),
 		configuration.GetString(prefix+"hostname"),
@@ -42,12 +44,16 @@ func (store *PostgresStore) Initialize(configuration util.Configuration, prefix
 	)
 }
 
-func (store *PostgresStore) initialize(user, password, hostname string, port int, database, schema, sslmode string, maxIdle, maxOpen, maxLifetimeSeconds int) (err error) {
+func (store *PostgresStore) initialize(upsertQuery string, enableUpsert bool, user, password, hostname string, port int, database, schema, sslmode string, maxIdle, maxOpen, maxLifetimeSeconds int) (err error) {
 
 	store.SupportBucketTable = false
+	if !enableUpsert {
+		upsertQuery = ""
+	}
 	store.SqlGenerator = &SqlGenPostgres{
 		CreateTableSqlTemplate: "",
 		DropTableSqlTemplate:   `drop table "%s"`,
+		UpsertQueryTemplate:    upsertQuery,
 	}
 
 	sqlUrl := fmt.Sprintf(CONNECTION_URL_PATTERN, hostname, port, sslmode)

weed/filer/postgres2/postgres2_store.go

@@ -32,6 +32,8 @@ func (store *PostgresStore2) GetName() string {
 func (store *PostgresStore2) Initialize(configuration util.Configuration, prefix string) (err error) {
 	return store.initialize(
 		configuration.GetString(prefix+"createTable"),
+		configuration.GetString(prefix+"upsertQuery"),
+		configuration.GetBool(prefix+"enableUpsert"),
 		configuration.GetString(prefix+"username"),
 		configuration.GetString(prefix+"password"),
 		configuration.GetString(prefix+"hostname"),
@@ -45,12 +47,16 @@ func (store *PostgresStore2) Initialize(configuration util.Configuration, prefix
 	)
 }
 
-func (store *PostgresStore2) initialize(createTable, user, password, hostname string, port int, database, schema, sslmode string, maxIdle, maxOpen, maxLifetimeSeconds int) (err error) {
+func (store *PostgresStore2) initialize(createTable, upsertQuery string, enableUpsert bool, user, password, hostname string, port int, database, schema, sslmode string, maxIdle, maxOpen, maxLifetimeSeconds int) (err error) {
 
 	store.SupportBucketTable = true
+	if !enableUpsert {
+		upsertQuery = ""
+	}
 	store.SqlGenerator = &postgres.SqlGenPostgres{
 		CreateTableSqlTemplate: createTable,
 		DropTableSqlTemplate:   `drop table "%s"`,
+		UpsertQueryTemplate:    upsertQuery,
 	}
 
 	sqlUrl := fmt.Sprintf(CONNECTION_URL_PATTERN, hostname, port, sslmode)

weed/filer/stream.go

@@ -48,15 +48,7 @@ func StreamContent(masterClient wdclient.HasLookupFileIdFunction, w io.Writer, c
 	for _, chunkView := range chunkViews {
 
 		urlStrings := fileId2Url[chunkView.FileId]
-		data, err := retriedFetchChunkData(
-			urlStrings,
-			chunkView.CipherKey,
-			chunkView.IsGzipped,
-			chunkView.IsFullChunk(),
-			false,
-			chunkView.Offset,
-			int(chunkView.Size),
-		)
+		data, err := retriedFetchChunkData(urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size))
 		if err != nil {
 			glog.Errorf("read chunk: %v", err)
 			return fmt.Errorf("read chunk: %v", err)
@@ -78,14 +70,7 @@ func CheckAllChunkViews(chunkViews []*ChunkView, fileId2Url *map[string][]string
 		urlStrings := (*fileId2Url)[chunkView.FileId]
 		glog.V(9).Infof("Check chunk: %+v\n url: %v", chunkView, urlStrings)
 		gErr.Go(func() error {
-			_, err := retriedFetchChunkData(
-				urlStrings,
-				chunkView.CipherKey,
-				chunkView.IsGzipped,
-				chunkView.IsFullChunk(),
-				true,
-				chunkView.Offset,
-				int(chunkView.Size))
+			_, err := retriedFetchChunkData(urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size))
 			return err
 		})
 	}
@@ -110,7 +95,7 @@ func ReadAll(masterClient *wdclient.MasterClient, chunks []*filer_pb.FileChunk)
 			return nil, err
 		}
 
-		data, err := retriedFetchChunkData(urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), false, chunkView.Offset, int(chunkView.Size))
+		data, err := retriedFetchChunkData(urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size))
 		if err != nil {
 			return nil, err
 		}

weed/filesys/file.go

@@ -151,7 +151,7 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 			}
 			entry.Chunks = chunks
 			file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), chunks)
-			file.reader = nil
+			file.setReader(nil)
 		}
 		entry.Attributes.FileSize = req.Size
 		file.dirtyMetadata = true
@@ -261,6 +261,8 @@ func (file *File) Forget() {
 	t := util.NewFullPath(file.dir.FullPath(), file.Name)
 	glog.V(4).Infof("Forget file %s", t)
 	file.wfs.fsNodeCache.DeleteFsNode(t)
+	file.wfs.ReleaseHandle(t, 0)
+	file.setReader(nil)
 }
 
 func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
@@ -280,9 +282,6 @@ func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, er
 		return entry, err
 	}
 	if entry != nil {
-		if entry.Attributes == nil {
-			entry.Attributes = &filer_pb.FuseAttributes{}
-		}
 		file.setEntry(entry)
 	} else {
 		glog.Warningf("maybeLoadEntry not found entry %s/%s: %v", file.dir.FullPath(), file.Name, err)
@@ -330,19 +329,29 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 		file.entryViewCache = filer.MergeIntoVisibles(file.entryViewCache, chunk)
 	}
 
-	file.reader = nil
+	file.setReader(nil)
 
 	glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(entry.Chunks), len(chunks))
 
 	entry.Chunks = append(entry.Chunks, newChunks...)
 }
 
+func (file *File) setReader(reader io.ReaderAt) {
+	r := file.reader
+	if r != nil {
+		if closer, ok := r.(io.Closer); ok {
+			closer.Close()
+		}
+	}
+	file.reader = reader
+}
+
 func (file *File) setEntry(entry *filer_pb.Entry) {
 	file.entryLock.Lock()
 	defer file.entryLock.Unlock()
 	file.entry = entry
 	file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), entry.Chunks)
-	file.reader = nil
+	file.setReader(nil)
 }
 
 func (file *File) clearEntry() {
@@ -350,7 +359,7 @@ func (file *File) clearEntry() {
 	defer file.entryLock.Unlock()
 	file.entry = nil
 	file.entryViewCache = nil
-	file.reader = nil
+	file.setReader(nil)
 }
 
 func (file *File) saveEntry(entry *filer_pb.Entry) error {

weed/filesys/filehandle.go

@@ -23,7 +23,7 @@ type FileHandle struct {
 	dirtyPages  *ContinuousDirtyPages
 	contentType string
 	handle      uint64
-	sync.RWMutex
+	sync.Mutex
 
 	f         *File
 	RequestId fuse.RequestID // unique ID for request
@@ -59,8 +59,8 @@ var _ = fs.HandleReleaser(&FileHandle{})
 func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
 
 	glog.V(4).Infof("%s read fh %d: [%d,%d) size %d resp.Data cap=%d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, cap(resp.Data))
-	fh.RLock()
-	defer fh.RUnlock()
+	fh.Lock()
+	defer fh.Unlock()
 
 	if req.Size <= 0 {
 		return nil
@@ -130,7 +130,7 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 		if chunkResolveErr != nil {
 			return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
 		}
-		fh.f.reader = nil
+		fh.f.setReader(nil)
 	}
 
 	reader := fh.f.reader
@@ -138,7 +138,7 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 		chunkViews := filer.ViewFromVisibleIntervals(fh.f.entryViewCache, 0, math.MaxInt64)
 		reader = filer.NewChunkReaderAtFromClient(fh.f.wfs.LookupFn(), chunkViews, fh.f.wfs.chunkCache, fileSize)
 	}
-	fh.f.reader = reader
+	fh.f.setReader(reader)
 
 	totalRead, err := reader.ReadAt(buff, offset)
 
@@ -207,12 +207,7 @@ func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) err
 		fh.f.isOpen--
 
 		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
-		if closer, ok := fh.f.reader.(io.Closer); ok {
-			if closer != nil {
-				closer.Close()
-			}
-		}
-		fh.f.reader = nil
+		fh.f.setReader(nil)
 	}
 
 	return nil

weed/operation/buffer_pool.go

@@ -0,0 +1,24 @@
+package operation
+
+import (
+	"github.com/valyala/bytebufferpool"
+	"sync/atomic"
+)
+
+var bufferCounter int64
+
+func GetBuffer() *bytebufferpool.ByteBuffer {
+	defer func() {
+		atomic.AddInt64(&bufferCounter, 1)
+		// println("+", bufferCounter)
+	}()
+	return bytebufferpool.Get()
+}
+
+func PutBuffer(buf *bytebufferpool.ByteBuffer) {
+	defer func() {
+		atomic.AddInt64(&bufferCounter, -1)
+		// println("-", bufferCounter)
+	}()
+	bytebufferpool.Put(buf)
+}

weed/operation/upload_content.go

@@ -19,7 +19,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/valyala/bytebufferpool"
 )
 
 type UploadResult struct {
@@ -190,8 +189,8 @@ func doUploadData(uploadUrl string, filename string, cipher bool, data []byte, i
 }
 
 func upload_content(uploadUrl string, fillBufferFunction func(w io.Writer) error, filename string, isGzipped bool, originalDataSize int, mtype string, pairMap map[string]string, jwt security.EncodedJwt) (*UploadResult, error) {
-	buf := bytebufferpool.Get()
-	defer bytebufferpool.Put(buf)
+	buf := GetBuffer()
+	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(filename)))

weed/sequence/snowflake_sequencer.go

@@ -0,0 +1,46 @@
+package sequence
+
+import (
+	"fmt"
+	"hash/fnv"
+
+	"github.com/bwmarrin/snowflake"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+)
+
+// a simple snowflake Sequencer
+type SnowflakeSequencer struct {
+	node *snowflake.Node
+}
+
+func NewSnowflakeSequencer(nodeid string) (*SnowflakeSequencer, error) {
+	nodeid_hash := hash(nodeid) & 0x3ff
+	glog.V(0).Infof("use snowflake seq id generator, nodeid:%s hex_of_nodeid: %x", nodeid, nodeid_hash)
+	node, err := snowflake.NewNode(int64(nodeid_hash))
+	if err != nil {
+		fmt.Println(err)
+		return nil, err
+	}
+
+	sequencer := &SnowflakeSequencer{node: node}
+	return sequencer, nil
+}
+
+func hash(s string) uint32 {
+	h := fnv.New32a()
+	h.Write([]byte(s))
+	return h.Sum32()
+}
+
+func (m *SnowflakeSequencer) NextFileId(count uint64) uint64 {
+	return uint64(m.node.Generate().Int64())
+}
+
+// ignore setmax as we are snowflake
+func (m *SnowflakeSequencer) SetMax(seenValue uint64) {
+}
+
+// return a new id as no Peek is stored
+func (m *SnowflakeSequencer) Peek() uint64 {
+	return uint64(m.node.Generate().Int64())
+}

weed/server/filer_grpc_server_rename.go

@@ -33,8 +33,7 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 		return nil, fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
 	}
 
-	var events MoveEvents
-	moveErr := fs.moveEntry(ctx, oldParent, oldEntry, newParent, req.NewName, &events)
+	moveErr := fs.moveEntry(ctx, oldParent, oldEntry, newParent, req.NewName)
 	if moveErr != nil {
 		fs.filer.RollbackTransaction(ctx)
 		return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
@@ -48,11 +47,11 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 	return &filer_pb.AtomicRenameEntryResponse{}, nil
 }
 
-func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, events *MoveEvents) error {
+func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
 
-	if err := fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName, events, func() error {
+	if err := fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName, func() error {
 		if entry.IsDirectory() {
-			if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName, events); err != nil {
+			if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName); err != nil {
 				return err
 			}
 		}
@@ -64,7 +63,7 @@ func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, e
 	return nil
 }
 
-func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, events *MoveEvents) error {
+func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
 
 	currentDirPath := oldParent.Child(entry.Name())
 	newDirPath := newParent.Child(newName)
@@ -85,7 +84,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.
 		for _, item := range entries {
 			lastFileName = item.Name()
 			// println("processing", lastFileName)
-			err := fs.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name(), events)
+			err := fs.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name())
 			if err != nil {
 				return err
 			}
@@ -97,8 +96,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.
 	return nil
 }
 
-func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, events *MoveEvents,
-	moveFolderSubEntries func() error) error {
+func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error) error {
 
 	oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
 
@@ -122,8 +120,6 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
 		return createErr
 	}
 
-	events.newEntries = append(events.newEntries, newEntry)
-
 	if moveFolderSubEntries != nil {
 		if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
 			return moveChildrenErr
@@ -136,13 +132,6 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
 		return deleteErr
 	}
 
-	events.oldEntries = append(events.oldEntries, entry)
-
 	return nil
 
 }
-
-type MoveEvents struct {
-	oldEntries []*filer.Entry
-	newEntries []*filer.Entry
-}

weed/server/filer_server.go

@@ -45,22 +45,23 @@ import (
 )
 
 type FilerOption struct {
-	Masters            []string
-	Collection         string
-	DefaultReplication string
-	DisableDirListing  bool
-	MaxMB              int
-	DirListingLimit    int
-	DataCenter         string
-	Rack               string
-	DefaultLevelDbDir  string
-	DisableHttp        bool
-	Host               string
-	Port               uint32
-	recursiveDelete    bool
-	Cipher             bool
-	SaveToFilerLimit   int
-	Filers             []string
+	Masters               []string
+	Collection            string
+	DefaultReplication    string
+	DisableDirListing     bool
+	MaxMB                 int
+	DirListingLimit       int
+	DataCenter            string
+	Rack                  string
+	DefaultLevelDbDir     string
+	DisableHttp           bool
+	Host                  string
+	Port                  uint32
+	recursiveDelete       bool
+	Cipher                bool
+	SaveToFilerLimit      int64
+	Filers                []string
+	ConcurrentUploadLimit int64
 }
 
 type FilerServer struct {
@@ -79,14 +80,18 @@ type FilerServer struct {
 
 	brokers     map[string]map[string]bool
 	brokersLock sync.Mutex
+
+	inFlightDataSize      int64
+	inFlightDataLimitCond *sync.Cond
 }
 
 func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) {
 
 	fs = &FilerServer{
-		option:         option,
-		grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.filer"),
-		brokers:        make(map[string]map[string]bool),
+		option:                option,
+		grpcDialOption:        security.LoadClientTLS(util.GetViper(), "grpc.filer"),
+		brokers:               make(map[string]map[string]bool),
+		inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
 	}
 	fs.listenersCond = sync.NewCond(&fs.listenersLock)
 

weed/server/filer_server_handlers.go

@@ -4,6 +4,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"net/http"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/stats"
@@ -47,18 +48,34 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 			fs.DeleteHandler(w, r)
 		}
 		stats.FilerRequestHistogram.WithLabelValues("delete").Observe(time.Since(start).Seconds())
-	case "PUT":
-		stats.FilerRequestCounter.WithLabelValues("put").Inc()
-		if _, ok := r.URL.Query()["tagging"]; ok {
-			fs.PutTaggingHandler(w, r)
-		} else {
-			fs.PostHandler(w, r)
+	case "POST", "PUT":
+
+		// wait until in flight data is less than the limit
+		contentLength := getContentLength(r)
+		fs.inFlightDataLimitCond.L.Lock()
+		for atomic.LoadInt64(&fs.inFlightDataSize) > fs.option.ConcurrentUploadLimit {
+			fs.inFlightDataLimitCond.Wait()
+		}
+		atomic.AddInt64(&fs.inFlightDataSize, contentLength)
+		fs.inFlightDataLimitCond.L.Unlock()
+		defer func() {
+			atomic.AddInt64(&fs.inFlightDataSize, -contentLength)
+			fs.inFlightDataLimitCond.Signal()
+		}()
+
+		if r.Method == "PUT" {
+			stats.FilerRequestCounter.WithLabelValues("put").Inc()
+			if _, ok := r.URL.Query()["tagging"]; ok {
+				fs.PutTaggingHandler(w, r)
+			} else {
+				fs.PostHandler(w, r, contentLength)
+			}
+			stats.FilerRequestHistogram.WithLabelValues("put").Observe(time.Since(start).Seconds())
+		} else { // method == "POST"
+			stats.FilerRequestCounter.WithLabelValues("post").Inc()
+			fs.PostHandler(w, r, contentLength)
+			stats.FilerRequestHistogram.WithLabelValues("post").Observe(time.Since(start).Seconds())
 		}
-		stats.FilerRequestHistogram.WithLabelValues("put").Observe(time.Since(start).Seconds())
-	case "POST":
-		stats.FilerRequestCounter.WithLabelValues("post").Inc()
-		fs.PostHandler(w, r)
-		stats.FilerRequestHistogram.WithLabelValues("post").Observe(time.Since(start).Seconds())
 	case "OPTIONS":
 		stats.FilerRequestCounter.WithLabelValues("options").Inc()
 		OptionsHandler(w, r, false)

weed/server/filer_server_handlers_write.go

@@ -52,7 +52,7 @@ func (fs *FilerServer) assignNewFileInfo(so *operation.StorageOption) (fileId, u
 	return
 }
 
-func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
+func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request, contentLength int64) {
 
 	ctx := context.Background()
 
@@ -66,7 +66,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
 		query.Get("rack"),
 	)
 
-	fs.autoChunk(ctx, w, r, so)
+	fs.autoChunk(ctx, w, r, contentLength, so)
 	util.CloseRequest(r)
 
 }

weed/server/filer_server_handlers_write_autochunk.go

@@ -2,11 +2,8 @@ package weed_server
 
 import (
 	"context"
-	"crypto/md5"
 	"fmt"
-	"hash"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"os"
 	"path"
@@ -19,13 +16,12 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
-	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, so *operation.StorageOption) {
+func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, contentLength int64, so *operation.StorageOption) {
 
 	// autoChunking can be set at the command-line level or as a query param. Query param overrides command-line
 	query := r.URL.Query()
@@ -51,14 +47,16 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
 		if r.Header.Get("Content-Type") == "" && strings.HasSuffix(r.URL.Path, "/") {
 			reply, err = fs.mkdir(ctx, w, r)
 		} else {
-			reply, md5bytes, err = fs.doPostAutoChunk(ctx, w, r, chunkSize, so)
+			reply, md5bytes, err = fs.doPostAutoChunk(ctx, w, r, chunkSize, contentLength, so)
 		}
 	} else {
-		reply, md5bytes, err = fs.doPutAutoChunk(ctx, w, r, chunkSize, so)
+		reply, md5bytes, err = fs.doPutAutoChunk(ctx, w, r, chunkSize, contentLength, so)
 	}
 	if err != nil {
 		if strings.HasPrefix(err.Error(), "read input:") {
 			writeJsonError(w, r, 499, err)
+		} else if strings.HasSuffix(err.Error(), "is a file") {
+			writeJsonError(w, r, http.StatusConflict, err)
 		} else {
 			writeJsonError(w, r, http.StatusInternalServerError, err)
 		}
@@ -70,7 +68,7 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
 	}
 }
 
-func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, chunkSize int32, so *operation.StorageOption) (filerResult *FilerPostResult, md5bytes []byte, replyerr error) {
+func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, chunkSize int32, contentLength int64, so *operation.StorageOption) (filerResult *FilerPostResult, md5bytes []byte, replyerr error) {
 
 	multipartReader, multipartReaderErr := r.MultipartReader()
 	if multipartReaderErr != nil {
@@ -91,7 +89,7 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite
 		contentType = ""
 	}
 
-	fileChunks, md5Hash, chunkOffset, err, smallContent := fs.uploadReaderToChunks(w, r, part1, chunkSize, fileName, contentType, so)
+	fileChunks, md5Hash, chunkOffset, err, smallContent := fs.uploadReaderToChunks(w, r, part1, chunkSize, fileName, contentType, contentLength, so)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -102,7 +100,7 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite
 	return
 }
 
-func (fs *FilerServer) doPutAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, chunkSize int32, so *operation.StorageOption) (filerResult *FilerPostResult, md5bytes []byte, replyerr error) {
+func (fs *FilerServer) doPutAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, chunkSize int32, contentLength int64, so *operation.StorageOption) (filerResult *FilerPostResult, md5bytes []byte, replyerr error) {
 
 	fileName := path.Base(r.URL.Path)
 	contentType := r.Header.Get("Content-Type")
@@ -110,7 +108,7 @@ func (fs *FilerServer) doPutAutoChunk(ctx context.Context, w http.ResponseWriter
 		contentType = ""
 	}
 
-	fileChunks, md5Hash, chunkOffset, err, smallContent := fs.uploadReaderToChunks(w, r, r.Body, chunkSize, fileName, contentType, so)
+	fileChunks, md5Hash, chunkOffset, err, smallContent := fs.uploadReaderToChunks(w, r, r.Body, chunkSize, fileName, contentType, contentLength, so)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -229,92 +227,6 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 	return filerResult, replyerr
 }
 
-func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, so *operation.StorageOption) ([]*filer_pb.FileChunk, hash.Hash, int64, error, []byte) {
-	var fileChunks []*filer_pb.FileChunk
-
-	md5Hash := md5.New()
-	var partReader = ioutil.NopCloser(io.TeeReader(reader, md5Hash))
-
-	chunkOffset := int64(0)
-	var smallContent []byte
-
-	for {
-		limitedReader := io.LimitReader(partReader, int64(chunkSize))
-
-		data, err := ioutil.ReadAll(limitedReader)
-		if err != nil {
-			return nil, nil, 0, err, nil
-		}
-		if chunkOffset == 0 && !isAppend(r) {
-			if len(data) < fs.option.SaveToFilerLimit || strings.HasPrefix(r.URL.Path, filer.DirectoryEtcRoot) && len(data) < 4*1024 {
-				smallContent = data
-				chunkOffset += int64(len(data))
-				break
-			}
-		}
-		dataReader := util.NewBytesReader(data)
-
-		// retry to assign a different file id
-		var fileId, urlLocation string
-		var auth security.EncodedJwt
-		var assignErr, uploadErr error
-		var uploadResult *operation.UploadResult
-		for i := 0; i < 3; i++ {
-			// assign one file id for one chunk
-			fileId, urlLocation, auth, assignErr = fs.assignNewFileInfo(so)
-			if assignErr != nil {
-				return nil, nil, 0, assignErr, nil
-			}
-
-			// upload the chunk to the volume server
-			uploadResult, uploadErr, _ = fs.doUpload(urlLocation, w, r, dataReader, fileName, contentType, nil, auth)
-			if uploadErr != nil {
-				time.Sleep(251 * time.Millisecond)
-				continue
-			}
-			break
-		}
-		if uploadErr != nil {
-			return nil, nil, 0, uploadErr, nil
-		}
-
-		// if last chunk exhausted the reader exactly at the border
-		if uploadResult.Size == 0 {
-			break
-		}
-
-		// Save to chunk manifest structure
-		fileChunks = append(fileChunks, uploadResult.ToPbFileChunk(fileId, chunkOffset))
-
-		glog.V(4).Infof("uploaded %s chunk %d to %s [%d,%d)", fileName, len(fileChunks), fileId, chunkOffset, chunkOffset+int64(uploadResult.Size))
-
-		// reset variables for the next chunk
-		chunkOffset = chunkOffset + int64(uploadResult.Size)
-
-		// if last chunk was not at full chunk size, but already exhausted the reader
-		if int64(uploadResult.Size) < int64(chunkSize) {
-			break
-		}
-	}
-
-	return fileChunks, md5Hash, chunkOffset, nil, smallContent
-}
-
-func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *http.Request, limitedReader io.Reader, fileName string, contentType string, pairMap map[string]string, auth security.EncodedJwt) (*operation.UploadResult, error, []byte) {
-
-	stats.FilerRequestCounter.WithLabelValues("chunkUpload").Inc()
-	start := time.Now()
-	defer func() {
-		stats.FilerRequestHistogram.WithLabelValues("chunkUpload").Observe(time.Since(start).Seconds())
-	}()
-
-	uploadResult, err, data := operation.Upload(urlLocation, fileName, fs.option.Cipher, limitedReader, false, contentType, pairMap, auth)
-	if uploadResult != nil && uploadResult.RetryCount > 0 {
-		stats.FilerRequestCounter.WithLabelValues("chunkUploadRetry").Add(float64(uploadResult.RetryCount))
-	}
-	return uploadResult, err, data
-}
-
 func (fs *FilerServer) saveAsChunk(so *operation.StorageOption) filer.SaveDataAsChunkFunctionType {
 
 	return func(reader io.Reader, name string, offset int64) (*filer_pb.FileChunk, string, string, error) {

weed/server/filer_server_handlers_write_upload.go

@@ -0,0 +1,164 @@
+package weed_server
+
+import (
+	"crypto/md5"
+	"hash"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"runtime"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/operation"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/security"
+	"github.com/chrislusf/seaweedfs/weed/stats"
+	"github.com/chrislusf/seaweedfs/weed/util"
+)
+
+var (
+	limitedUploadProcessor = util.NewLimitedOutOfOrderProcessor(int32(runtime.NumCPU()))
+)
+
+func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, contentLength int64, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, dataSize int64, err error, smallContent []byte) {
+
+	md5Hash = md5.New()
+	var partReader = ioutil.NopCloser(io.TeeReader(reader, md5Hash))
+
+	// save small content directly
+	if !isAppend(r) && ((0 < contentLength && contentLength < fs.option.SaveToFilerLimit) || strings.HasPrefix(r.URL.Path, filer.DirectoryEtcRoot) && contentLength < 4*1024) {
+		smallContent, err = ioutil.ReadAll(partReader)
+		dataSize = int64(len(smallContent))
+		return
+	}
+
+	resultsChan := make(chan *ChunkCreationResult, 2)
+
+	var waitForAllData sync.WaitGroup
+	waitForAllData.Add(1)
+	go func() {
+		// process upload results
+		defer waitForAllData.Done()
+		for result := range resultsChan {
+			if result.err != nil {
+				err = result.err
+				continue
+			}
+
+			// Save to chunk manifest structure
+			fileChunks = append(fileChunks, result.chunk)
+		}
+	}()
+
+	var lock sync.Mutex
+	readOffset := int64(0)
+	var wg sync.WaitGroup
+	var readErr error
+
+	for readErr == nil {
+
+		wg.Add(1)
+		request := func() {
+			defer wg.Done()
+
+			var localOffset int64
+			var data []byte
+			// read from the input
+			lock.Lock()
+			localOffset = readOffset
+			limitedReader := io.LimitReader(partReader, int64(chunkSize))
+			data, readErr = ioutil.ReadAll(limitedReader)
+			readOffset += int64(len(data))
+			lock.Unlock()
+			// handle read errors
+			if readErr != nil {
+				if readErr != io.EOF {
+					resultsChan <- &ChunkCreationResult{
+						err: readErr,
+					}
+				}
+				return
+			}
+			if len(data) == 0 {
+				readErr = io.EOF
+				return
+			}
+
+			// upload
+			dataReader := util.NewBytesReader(data)
+			fileId, uploadResult, uploadErr := fs.doCreateChunk(w, r, so, dataReader, fileName, contentType)
+			if uploadErr != nil {
+				resultsChan <- &ChunkCreationResult{
+					err: uploadErr,
+				}
+				return
+			}
+
+			glog.V(4).Infof("uploaded %s to %s [%d,%d)", fileName, fileId, localOffset, localOffset+int64(uploadResult.Size))
+
+			// send back uploaded file chunk
+			resultsChan <- &ChunkCreationResult{
+				chunk: uploadResult.ToPbFileChunk(fileId, localOffset),
+			}
+
+		}
+		limitedUploadProcessor.Execute(request)
+	}
+
+	go func() {
+		wg.Wait()
+		close(resultsChan)
+	}()
+
+	waitForAllData.Wait()
+
+	return fileChunks, md5Hash, readOffset, err, nil
+}
+
+type ChunkCreationResult struct {
+	chunk *filer_pb.FileChunk
+	err   error
+}
+
+func (fs *FilerServer) doCreateChunk(w http.ResponseWriter, r *http.Request, so *operation.StorageOption, dataReader *util.BytesReader, fileName string, contentType string) (string, *operation.UploadResult, error) {
+	// retry to assign a different file id
+	var fileId, urlLocation string
+	var auth security.EncodedJwt
+	var assignErr, uploadErr error
+	var uploadResult *operation.UploadResult
+	for i := 0; i < 3; i++ {
+		// assign one file id for one chunk
+		fileId, urlLocation, auth, assignErr = fs.assignNewFileInfo(so)
+		if assignErr != nil {
+			return "", nil, assignErr
+		}
+
+		// upload the chunk to the volume server
+		uploadResult, uploadErr, _ = fs.doUpload(urlLocation, w, r, dataReader, fileName, contentType, nil, auth)
+		if uploadErr != nil {
+			time.Sleep(251 * time.Millisecond)
+			continue
+		}
+		break
+	}
+	return fileId, uploadResult, uploadErr
+}
+
+func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *http.Request, limitedReader io.Reader, fileName string, contentType string, pairMap map[string]string, auth security.EncodedJwt) (*operation.UploadResult, error, []byte) {
+
+	stats.FilerRequestCounter.WithLabelValues("chunkUpload").Inc()
+	start := time.Now()
+	defer func() {
+		stats.FilerRequestHistogram.WithLabelValues("chunkUpload").Observe(time.Since(start).Seconds())
+	}()
+
+	uploadResult, err, data := operation.Upload(urlLocation, fileName, fs.option.Cipher, limitedReader, false, contentType, pairMap, auth)
+	if uploadResult != nil && uploadResult.RetryCount > 0 {
+		stats.FilerRequestCounter.WithLabelValues("chunkUploadRetry").Add(float64(uploadResult.RetryCount))
+	}
+	return uploadResult, err, data
+}

weed/server/master_grpc_server.go

@@ -80,10 +80,14 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 		dn.AdjustMaxVolumeCounts(heartbeat.MaxVolumeCounts)
 
 		glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
+		var dataCenter string
+		if dc := dn.GetDataCenter(); dc != nil {
+			dataCenter = string(dc.Id())
+		}
 		message := &master_pb.VolumeLocation{
 			Url:        dn.Url(),
 			PublicUrl:  dn.PublicUrl,
-			DataCenter: string(dn.GetDataCenter().Id()),
+			DataCenter: dataCenter,
 		}
 		if len(heartbeat.NewVolumes) > 0 || len(heartbeat.DeletedVolumes) > 0 {
 			// process delta volume ids if exists for fast volume id updates

weed/server/master_server.go

@@ -277,6 +277,13 @@ func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer
 			glog.Error(err)
 			seq = nil
 		}
+	case "snowflake":
+		var err error
+		seq, err = sequence.NewSnowflakeSequencer(fmt.Sprintf("%s:%d", option.Host, option.Port))
+		if err != nil {
+			glog.Error(err)
+			seq = nil
+		}
 	default:
 		seq = sequence.NewMemorySequencer()
 	}

weed/server/volume_server.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
 	"net/http"
+	"sync"
 
 	"google.golang.org/grpc"
 
@@ -34,6 +35,10 @@ type VolumeServer struct {
 	fileSizeLimitBytes      int64
 	isHeartbeating          bool
 	stopChan                chan bool
+
+	inFlightDataSize      int64
+	inFlightDataLimitCond *sync.Cond
+	concurrentUploadLimit int64
 }
 
 func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
@@ -48,6 +53,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	readRedirect bool,
 	compactionMBPerSecond int,
 	fileSizeLimitMB int,
+	concurrentUploadLimit int64,
 ) *VolumeServer {
 
 	v := util.GetViper()
@@ -72,6 +78,8 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 		fileSizeLimitBytes:      int64(fileSizeLimitMB) * 1024 * 1024,
 		isHeartbeating:          true,
 		stopChan:                make(chan bool),
+		inFlightDataLimitCond:   sync.NewCond(new(sync.Mutex)),
+		concurrentUploadLimit:   concurrentUploadLimit,
 	}
 	vs.SeedMasterNodes = masterNodes
 

weed/server/volume_server_handlers.go

@@ -2,7 +2,9 @@ package weed_server
 
 import (
 	"net/http"
+	"strconv"
 	"strings"
+	"sync/atomic"
 
 	"github.com/chrislusf/seaweedfs/weed/util"
 
@@ -40,8 +42,24 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
 		stats.DeleteRequest()
 		vs.guard.WhiteList(vs.DeleteHandler)(w, r)
 	case "PUT", "POST":
+
+		// wait until in flight data is less than the limit
+		contentLength := getContentLength(r)
+		vs.inFlightDataLimitCond.L.Lock()
+		for atomic.LoadInt64(&vs.inFlightDataSize) > vs.concurrentUploadLimit {
+			vs.inFlightDataLimitCond.Wait()
+		}
+		atomic.AddInt64(&vs.inFlightDataSize, contentLength)
+		vs.inFlightDataLimitCond.L.Unlock()
+		defer func() {
+			atomic.AddInt64(&vs.inFlightDataSize, -contentLength)
+			vs.inFlightDataLimitCond.Signal()
+		}()
+
+		// processs uploads
 		stats.WriteRequest()
 		vs.guard.WhiteList(vs.PostHandler)(w, r)
+
 	case "OPTIONS":
 		stats.ReadRequest()
 		w.Header().Add("Access-Control-Allow-Methods", "PUT, POST, GET, DELETE, OPTIONS")
@@ -49,6 +67,18 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
 	}
 }
 
+func getContentLength(r *http.Request) int64 {
+	contentLength := r.Header.Get("Content-Length")
+	if contentLength != "" {
+		length, err := strconv.ParseInt(contentLength, 10, 64)
+		if err != nil {
+			return 0
+		}
+		return length
+	}
+	return 0
+}
+
 func (vs *VolumeServer) publicReadOnlyHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
 	if r.Header.Get("Origin") != "" {

weed/shell/command_volume_check_disk.go

@@ -46,8 +46,10 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 	}
 
 	fsckCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	slowMode := fsckCommand.Bool("slow", false, "slow mode checks all replicas even file counts are the same")
 	verbose := fsckCommand.Bool("v", false, "verbose mode")
 	applyChanges := fsckCommand.Bool("force", false, "apply the fix")
+	nonRepairThreshold := fsckCommand.Float64("nonRepairThreshold", 0.3, "repair when missing keys is not more than this limit")
 	if err = fsckCommand.Parse(args); err != nil {
 		return nil
 	}
@@ -75,9 +77,11 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 		})
 		for len(replicas) >= 2 {
 			a, b := replicas[0], replicas[1]
-			if fileCount(a) == fileCount(b) {
-				replicas = replicas[1:]
-				continue
+			if !*slowMode {
+				if fileCount(a) == fileCount(b) {
+					replicas = replicas[1:]
+					continue
+				}
 			}
 			if a.info.ReadOnly || b.info.ReadOnly {
 				fmt.Fprintf(writer, "skipping readonly volume %d on %s and %s\n", a.info.Id, a.location.dataNode.Id, b.location.dataNode.Id)
@@ -98,10 +102,10 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 			}
 
 			// find and make up the differnces
-			if err := c.doVolumeCheckDisk(aDB, bDB, a, b, *verbose, writer, *applyChanges); err != nil {
+			if err := c.doVolumeCheckDisk(aDB, bDB, a, b, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
 				return err
 			}
-			if err := c.doVolumeCheckDisk(bDB, aDB, b, a, *verbose, writer, *applyChanges); err != nil {
+			if err := c.doVolumeCheckDisk(bDB, aDB, b, a, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
 				return err
 			}
 			replicas = replicas[1:]
@@ -111,7 +115,7 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 	return nil
 }
 
-func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool) error {
+func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64) error {
 
 	// find missing keys
 	// hash join, can be more efficient
@@ -125,7 +129,18 @@ func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_m
 		return nil
 	})
 
-	fmt.Fprintf(writer, "%s has %d entries, %s missed %d entries\n", source.location.dataNode.Id, counter, target.location.dataNode.Id, len(missingNeedles))
+	fmt.Fprintf(writer, "volume %d %s has %d entries, %s missed %d entries\n", source.info.Id, source.location.dataNode.Id, counter, target.location.dataNode.Id, len(missingNeedles))
+
+	if counter == 0 || len(missingNeedles) == 0 {
+		return nil
+	}
+
+	missingNeedlesFraction := float64(len(missingNeedles)) / float64(counter)
+	if missingNeedlesFraction > nonRepairThreshold {
+		return fmt.Errorf(
+			"failed to start repair volume %d, percentage of missing keys is greater than the threshold: %.2f > %.2f",
+			source.info.Id, missingNeedlesFraction, nonRepairThreshold)
+	}
 
 	for _, needleValue := range missingNeedles {
 
@@ -148,7 +163,6 @@ func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_m
 
 	}
 
-
 	return nil
 }
 
@@ -158,8 +172,8 @@ func (c *commandVolumeCheckDisk) readSourceNeedleBlob(sourceVolumeServer string,
 		resp, err := client.ReadNeedleBlob(context.Background(), &volume_server_pb.ReadNeedleBlobRequest{
 			VolumeId: volumeId,
 			NeedleId: uint64(needleValue.Key),
-			Offset: needleValue.Offset.ToActualOffset(),
-			Size: int32(needleValue.Size),
+			Offset:   needleValue.Offset.ToActualOffset(),
+			Size:     int32(needleValue.Size),
 		})
 		if err != nil {
 			return err
@@ -174,9 +188,9 @@ func (c *commandVolumeCheckDisk) writeNeedleBlobToTarget(targetVolumeServer stri
 
 	return operation.WithVolumeServerClient(targetVolumeServer, c.env.option.GrpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
 		_, err := client.WriteNeedleBlob(context.Background(), &volume_server_pb.WriteNeedleBlobRequest{
-			VolumeId: volumeId,
-			NeedleId: uint64(needleValue.Key),
-			Size: int32(needleValue.Size),
+			VolumeId:   volumeId,
+			NeedleId:   uint64(needleValue.Key),
+			Size:       int32(needleValue.Size),
 			NeedleBlob: needleBlob,
 		})
 		return err

weed/storage/needle/needle_read_write.go

@@ -173,7 +173,7 @@ func WriteNeedleBlob(w backend.BackendStorageFile, dataSlice []byte, size Size,
 
 	if version == Version3 {
 		tsOffset := NeedleHeaderSize + size + NeedleChecksumSize
-		util.Uint64toBytes(dataSlice[tsOffset : tsOffset+TimestampSize], appendAtNs)
+		util.Uint64toBytes(dataSlice[tsOffset:tsOffset+TimestampSize], appendAtNs)
 	}
 
 	if err == nil {

weed/util/chunk_cache/chunk_cache_in_memory.go

@@ -32,5 +32,7 @@ func (c *ChunkCacheInMemory) GetChunk(fileId string) []byte {
 }
 
 func (c *ChunkCacheInMemory) SetChunk(fileId string, data []byte) {
-	c.cache.Set(fileId, data, time.Hour)
+	localCopy := make([]byte, len(data))
+	copy(localCopy, data)
+	c.cache.Set(fileId, localCopy, time.Hour)
 }

weed/util/constants.go

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

weed/util/limiter.go

@@ -1,5 +1,12 @@
 package util
 
+import (
+	"math/rand"
+	"reflect"
+	"sync"
+	"sync/atomic"
+)
+
 // initial version comes from https://github.com/korovkin/limiter/blob/master/limiter.go
 
 // LimitedConcurrentExecutor object
@@ -38,3 +45,70 @@ func (c *LimitedConcurrentExecutor) Execute(job func()) {
 		job()
 	}()
 }
+
+// a different implementation, but somehow more "conservative"
+type OperationRequest func()
+
+type LimitedOutOfOrderProcessor struct {
+	processorSlots     uint32
+	processors         []chan OperationRequest
+	processorLimit     int32
+	processorLimitCond *sync.Cond
+	currentProcessor   int32
+}
+
+func NewLimitedOutOfOrderProcessor(limit int32) (c *LimitedOutOfOrderProcessor) {
+
+	processorSlots := uint32(32)
+	c = &LimitedOutOfOrderProcessor{
+		processorSlots:     processorSlots,
+		processors:         make([]chan OperationRequest, processorSlots),
+		processorLimit:     limit,
+		processorLimitCond: sync.NewCond(new(sync.Mutex)),
+	}
+
+	for i := 0; i < int(processorSlots); i++ {
+		c.processors[i] = make(chan OperationRequest)
+	}
+
+	cases := make([]reflect.SelectCase, processorSlots)
+	for i, ch := range c.processors {
+		cases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)}
+	}
+
+	go func() {
+		for {
+			_, value, ok := reflect.Select(cases)
+			if !ok {
+				continue
+			}
+
+			request := value.Interface().(OperationRequest)
+
+			if c.processorLimit > 0 {
+				c.processorLimitCond.L.Lock()
+				for atomic.LoadInt32(&c.currentProcessor) > c.processorLimit {
+					c.processorLimitCond.Wait()
+				}
+				atomic.AddInt32(&c.currentProcessor, 1)
+				c.processorLimitCond.L.Unlock()
+			}
+
+			go func() {
+				if c.processorLimit > 0 {
+					defer atomic.AddInt32(&c.currentProcessor, -1)
+					defer c.processorLimitCond.Signal()
+				}
+				request()
+			}()
+
+		}
+	}()
+
+	return c
+}
+
+func (c *LimitedOutOfOrderProcessor) Execute(request OperationRequest) {
+	index := rand.Uint32() % c.processorSlots
+	c.processors[index] <- request
+}