filer, volume: add concurrent upload size limit to avoid OOM

add some back pressure when writes are slow

7 files changed, 137 insertions, 69 deletions

diff --git a/weed/command/filer.go b/weed/command/filer.go
index 68edb8e69..82994b971 100644
--- a/weed/command/filer.go
+++ b/weed/command/filer.go
@@ -47,6 +47,7 @@ type FilerOptions struct {
 	metricsHttpPort         *int
 	saveToFilerLimit        *int
 	defaultLevelDbDirectory *string
+	concurrentUploadLimitMB *int
 }
 
 func init() {
@@ -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:      *fo.saveToFilerLimit,
+		Filers:                peers,
+		ConcurrentUploadLimit: int64(*fo.concurrentUploadLimitMB) * 1024 * 1024,
 	})
 	if nfs_err != nil {
 		glog.Fatalf("Filer startup error: %v", nfs_err)
diff --git a/weed/command/server.go b/weed/command/server.go
index c0ac5ee4c..eef2945fc 100644
--- a/weed/command/server.go
+++ b/weed/command/server.go
@@ -99,6 +99,7 @@ func init() {
 	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")
@@ -108,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")
diff --git a/weed/command/volume.go b/weed/command/volume.go
index b75af0f71..8014feaa6 100644
--- a/weed/command/volume.go
+++ b/weed/command/volume.go
@@ -35,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
 }
@@ -85,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")
@@ -237,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)
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 9e0770afa..a876b6d83 100644
--- a/weed/server/filer_server.go
+++ b/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      int
+	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)
 
diff --git a/weed/server/filer_server_handlers.go b/weed/server/filer_server_handlers.go
index 3bc0c5d0d..0df00aa5c 100644
--- a/weed/server/filer_server_handlers.go
+++ b/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 {
+	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)
+			}
+			stats.FilerRequestHistogram.WithLabelValues("put").Observe(time.Since(start).Seconds())
+		} else { // method == "POST"
+			stats.FilerRequestCounter.WithLabelValues("post").Inc()
 			fs.PostHandler(w, r)
+			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)
diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go
index e496b1ce2..e11d607a4 100644
--- a/weed/server/volume_server.go
+++ b/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
 
diff --git a/weed/server/volume_server_handlers.go b/weed/server/volume_server_handlers.go
index 7852c950a..4527add44 100644
--- a/weed/server/volume_server_handlers.go
+++ b/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") != "" {