diff --git a/weed/command/filer.go b/weed/command/filer.go index f91bffaa81..cd4ac62454 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -86,6 +86,7 @@ type FilerOptions struct { defaultLevelDbDirectory *string concurrentUploadLimitMB *int concurrentFileUploadLimit *int + inflightUploadDataTimeout *time.Duration debug *bool debugPort *int localSocket *string @@ -144,6 +145,7 @@ func init() { 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", 0, "limit total concurrent upload size, 0 means unlimited") f.concurrentFileUploadLimit = cmdFiler.Flag.Int("concurrentFileUploadLimit", 0, "limit number of concurrent file uploads, 0 means unlimited") + f.inflightUploadDataTimeout = cmdFiler.Flag.Duration("inflightUploadDataTimeout", 60*time.Second, "how long an upload waits for a slot under the concurrent upload limits before it is rejected with 429, 0 means no limit") f.debug = cmdFiler.Flag.Bool("debug", false, "serves runtime profiling data, e.g., http://localhost:/debug/pprof/goroutine?debug=2") f.debugPort = cmdFiler.Flag.Int("debug.port", 6060, "http port for debugging") f.localSocket = cmdFiler.Flag.String("localSocket", "", "default to /tmp/seaweedfs-filer-.sock") @@ -183,6 +185,7 @@ func init() { filerS3Options.idleTimeout = cmdFiler.Flag.Int("s3.idleTimeout", 120, "connection idle seconds") filerS3Options.concurrentUploadLimitMB = cmdFiler.Flag.Int("s3.concurrentUploadLimitMB", 0, "limit total concurrent upload size for S3, 0 means unlimited") filerS3Options.concurrentFileUploadLimit = cmdFiler.Flag.Int("s3.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads for S3, 0 means unlimited") + filerS3Options.inflightUploadDataTimeout = cmdFiler.Flag.Duration("s3.inflightUploadDataTimeout", 60*time.Second, "how long an upload waits for a slot under the concurrent upload limits before it is rejected with SlowDown, 0 means no limit") filerS3Options.enableIam = cmdFiler.Flag.Bool("s3.iam", true, "enable embedded IAM API on the same S3 port") filerS3Options.cipher = cmdFiler.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads; on whenever -encryptVolumeData is") filerS3Options.iamReadOnly = cmdFiler.Flag.Bool("s3.iam.readOnly", true, "disable IAM write operations on this server") @@ -438,6 +441,7 @@ func (fo *FilerOptions) startFiler() { SaveToFilerLimit: int64(*fo.saveToFilerLimit), ConcurrentUploadLimit: int64(*fo.concurrentUploadLimitMB) * 1024 * 1024, ConcurrentFileUploadLimit: int64(*fo.concurrentFileUploadLimit), + InflightUploadDataTimeout: *fo.inflightUploadDataTimeout, ShowUIDirectoryDelete: *fo.showUIDirectoryDelete, DownloadMaxBytesPs: int64(*fo.downloadMaxMBps) * 1024 * 1024, DiskType: *fo.diskType, diff --git a/weed/command/mini.go b/weed/command/mini.go index ffbdf4f5da..4b1fb6e584 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -483,6 +483,7 @@ func initMiniFilerFlags() { miniFilerOptions.saveToFilerLimit = cmdMini.Flag.Int("filer.saveToFilerLimit", 0, "files smaller than this limit will be saved in filer store") miniFilerOptions.concurrentUploadLimitMB = cmdMini.Flag.Int("filer.concurrentUploadLimitMB", 0, "limit total concurrent upload size") miniFilerOptions.concurrentFileUploadLimit = cmdMini.Flag.Int("filer.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads") + miniFilerOptions.inflightUploadDataTimeout = cmdMini.Flag.Duration("filer.inflightUploadDataTimeout", 60*time.Second, "how long an upload waits for a slot under the concurrent upload limits before it is rejected with 429, 0 means no limit") miniFilerOptions.localSocket = cmdMini.Flag.String("filer.localSocket", "", "default to /tmp/seaweedfs-filer-.sock") miniFilerOptions.showUIDirectoryDelete = cmdMini.Flag.Bool("filer.ui.deleteDir", true, "enable filer UI show delete directory button") miniFilerOptions.downloadMaxMBps = cmdMini.Flag.Int("filer.downloadMaxMBps", 0, "download max speed for each download request, in MB per second") @@ -553,6 +554,7 @@ func initMiniS3Flags() { miniS3Options.idleTimeout = cmdMini.Flag.Int("s3.idleTimeout", 120, "connection idle seconds") miniS3Options.concurrentUploadLimitMB = cmdMini.Flag.Int("s3.concurrentUploadLimitMB", 0, "limit total concurrent upload size") miniS3Options.concurrentFileUploadLimit = cmdMini.Flag.Int("s3.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads") + miniS3Options.inflightUploadDataTimeout = cmdMini.Flag.Duration("s3.inflightUploadDataTimeout", 60*time.Second, "how long an upload waits for a slot under the concurrent upload limits before it is rejected with SlowDown, 0 means no limit") miniS3Options.enableIam = cmdMini.Flag.Bool("s3.iam", true, "enable embedded IAM API on the same port") miniS3Options.iamReadOnly = miniS3IamReadOnly miniS3Options.dataCenter = cmdMini.Flag.String("s3.dataCenter", "", "prefer to read and write to volumes in this data center") diff --git a/weed/command/s3.go b/weed/command/s3.go index 8b65653a5d..c49f2d5004 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -74,6 +74,7 @@ type S3Options struct { idleTimeout *int concurrentUploadLimitMB *int concurrentFileUploadLimit *int + inflightUploadDataTimeout *time.Duration enableIam *bool iamReadOnly *bool debug *bool @@ -134,6 +135,7 @@ func init() { s3StandaloneOptions.idleTimeout = cmdS3.Flag.Int("idleTimeout", 120, "connection idle seconds") s3StandaloneOptions.concurrentUploadLimitMB = cmdS3.Flag.Int("concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited") s3StandaloneOptions.concurrentFileUploadLimit = cmdS3.Flag.Int("concurrentFileUploadLimit", 0, "limit number of concurrent file uploads, 0 means unlimited") + s3StandaloneOptions.inflightUploadDataTimeout = cmdS3.Flag.Duration("inflightUploadDataTimeout", 60*time.Second, "how long an upload waits for a slot under the concurrent upload limits before it is rejected with SlowDown, 0 means no limit") s3StandaloneOptions.enableIam = cmdS3.Flag.Bool("iam", true, "enable embedded IAM API on the same port") s3StandaloneOptions.iamReadOnly = cmdS3.Flag.Bool("iam.readOnly", true, "disable IAM write operations on this server") s3StandaloneOptions.debug = cmdS3.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port") @@ -402,6 +404,7 @@ func (s3opt *S3Options) startS3Server() bool { IamConfig: iamConfigPath, // Advanced IAM config (optional) ConcurrentUploadLimit: int64(*s3opt.concurrentUploadLimitMB) * 1024 * 1024, ConcurrentFileUploadLimit: int64(*s3opt.concurrentFileUploadLimit), + InflightUploadDataTimeout: *s3opt.inflightUploadDataTimeout, EnableIam: *s3opt.enableIam, // Embedded IAM API (enabled by default) IamReadOnly: *s3opt.iamReadOnly, Cipher: *s3opt.cipher || filerCipher, // encrypt data on volume servers diff --git a/weed/command/server.go b/weed/command/server.go index 6dd4cb3b53..78553d8ca8 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -131,6 +131,7 @@ func init() { 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", 0, "limit total concurrent upload size, 0 means unlimited") filerOptions.concurrentFileUploadLimit = cmdServer.Flag.Int("filer.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads, 0 means unlimited") + filerOptions.inflightUploadDataTimeout = cmdServer.Flag.Duration("filer.inflightUploadDataTimeout", 60*time.Second, "how long an upload waits for a slot under the concurrent upload limits before it is rejected with 429, 0 means no limit") filerOptions.localSocket = cmdServer.Flag.String("filer.localSocket", "", "default to /tmp/seaweedfs-filer-.sock") filerOptions.showUIDirectoryDelete = cmdServer.Flag.Bool("filer.ui.deleteDir", true, "enable filer UI show delete directory button") filerOptions.downloadMaxMBps = cmdServer.Flag.Int("filer.downloadMaxMBps", 0, "download max speed for each download request, in MB per second") @@ -199,6 +200,7 @@ func init() { s3Options.idleTimeout = cmdServer.Flag.Int("s3.idleTimeout", 120, "connection idle seconds") s3Options.concurrentUploadLimitMB = cmdServer.Flag.Int("s3.concurrentUploadLimitMB", 0, "limit total concurrent upload size for S3, 0 means unlimited") s3Options.concurrentFileUploadLimit = cmdServer.Flag.Int("s3.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads for S3, 0 means unlimited") + s3Options.inflightUploadDataTimeout = cmdServer.Flag.Duration("s3.inflightUploadDataTimeout", 60*time.Second, "how long an upload waits for a slot under the concurrent upload limits before it is rejected with SlowDown, 0 means no limit") s3Options.enableIam = cmdServer.Flag.Bool("s3.iam", true, "enable embedded IAM API on the same S3 port") s3Options.iamReadOnly = cmdServer.Flag.Bool("s3.iam.readOnly", true, "disable IAM write operations on this server") s3Options.cipher = cmdServer.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads; on whenever -filer.encryptVolumeData is") diff --git a/weed/s3api/s3api_circuit_breaker.go b/weed/s3api/s3api_circuit_breaker.go index 5bc9de126c..ecdf736831 100644 --- a/weed/s3api/s3api_circuit_breaker.go +++ b/weed/s3api/s3api_circuit_breaker.go @@ -17,6 +17,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/util" ) type CircuitBreaker struct { @@ -111,40 +112,19 @@ func (cb *CircuitBreaker) Limit(f func(w http.ResponseWriter, r *http.Request), contentLength = 0 } - // Wait until in flight data is less than the limit - cb.s3a.inFlightDataLimitCond.L.Lock() - inFlightDataSize := atomic.LoadInt64(&cb.s3a.inFlightDataSize) - inFlightUploads := atomic.LoadInt64(&cb.s3a.inFlightUploads) - - // Wait if either data size limit or file count limit is exceeded - for (cb.s3a.option.ConcurrentUploadLimit != 0 && inFlightDataSize > cb.s3a.option.ConcurrentUploadLimit) || - (cb.s3a.option.ConcurrentFileUploadLimit != 0 && inFlightUploads >= cb.s3a.option.ConcurrentFileUploadLimit) { - if cb.s3a.option.ConcurrentUploadLimit != 0 && inFlightDataSize > cb.s3a.option.ConcurrentUploadLimit { - glog.V(4).Infof("wait because inflight data %d > %d", inFlightDataSize, cb.s3a.option.ConcurrentUploadLimit) - } - if cb.s3a.option.ConcurrentFileUploadLimit != 0 && inFlightUploads >= cb.s3a.option.ConcurrentFileUploadLimit { - glog.V(4).Infof("wait because inflight uploads %d >= %d", inFlightUploads, cb.s3a.option.ConcurrentFileUploadLimit) - } - cb.s3a.inFlightDataLimitCond.Wait() - inFlightDataSize = atomic.LoadInt64(&cb.s3a.inFlightDataSize) - inFlightUploads = atomic.LoadInt64(&cb.s3a.inFlightUploads) + if !cb.waitForUploadSlot(w, r, contentLength) { + return } - cb.s3a.inFlightDataLimitCond.L.Unlock() - - // Increment counters - newUploads := atomic.AddInt64(&cb.s3a.inFlightUploads, 1) - newSize := atomic.AddInt64(&cb.s3a.inFlightDataSize, contentLength) - // Update metrics - stats.S3InFlightUploadCountGauge.Set(float64(newUploads)) - stats.S3InFlightUploadBytesGauge.Set(float64(newSize)) defer func() { + cb.s3a.inFlightDataLimitCond.L.Lock() + defer cb.s3a.inFlightDataLimitCond.L.Unlock() // Decrement counters newUploads := atomic.AddInt64(&cb.s3a.inFlightUploads, -1) newSize := atomic.AddInt64(&cb.s3a.inFlightDataSize, -contentLength) // Update metrics stats.S3InFlightUploadCountGauge.Set(float64(newUploads)) stats.S3InFlightUploadBytesGauge.Set(float64(newSize)) - cb.s3a.inFlightDataLimitCond.Signal() + cb.s3a.inFlightDataLimitCond.Broadcast() }() } @@ -186,6 +166,53 @@ func (cb *CircuitBreaker) Limit(f func(w http.ResponseWriter, r *http.Request), }, Action(action) } +func (cb *CircuitBreaker) waitForUploadSlot(w http.ResponseWriter, r *http.Request, contentLength int64) bool { + s3a := cb.s3a + s3a.inFlightDataLimitCond.L.Lock() + err := util.WaitUntil(r.Context(), s3a.inFlightDataLimitCond, s3a.option.InflightUploadDataTimeout, func() bool { + inFlightDataSize := atomic.LoadInt64(&s3a.inFlightDataSize) + if s3a.option.ConcurrentUploadLimit != 0 && inFlightDataSize > s3a.option.ConcurrentUploadLimit { + glog.V(4).Infof("wait because inflight data %d > %d", inFlightDataSize, s3a.option.ConcurrentUploadLimit) + return false + } + inFlightUploads := atomic.LoadInt64(&s3a.inFlightUploads) + if s3a.option.ConcurrentFileUploadLimit != 0 && inFlightUploads >= s3a.option.ConcurrentFileUploadLimit { + glog.V(4).Infof("wait because inflight uploads %d >= %d", inFlightUploads, s3a.option.ConcurrentFileUploadLimit) + return false + } + return true + }) + if err == nil { + // Increment counters + newUploads := atomic.AddInt64(&s3a.inFlightUploads, 1) + newSize := atomic.AddInt64(&s3a.inFlightDataSize, contentLength) + // Update metrics + stats.S3InFlightUploadCountGauge.Set(float64(newUploads)) + stats.S3InFlightUploadBytesGauge.Set(float64(newSize)) + } + s3a.inFlightDataLimitCond.L.Unlock() + + if err == nil { + return true + } + // Reject without reading the body: draining it would send an + // Expect: 100-continue client the go-ahead, and closing an HTTP/1 + // connection stops net/http from reading it before the reply. + if r.ProtoMajor == 1 { + w.Header().Set("Connection", "close") + } + rejected := r.WithContext(r.Context()) + rejected.Body = http.NoBody + if r.Context().Err() != nil { + glog.V(1).Infof("upload from %s cancelled while waiting for an upload slot: %v", r.RemoteAddr, err) + s3err.WriteErrorResponse(w, rejected, s3err.ErrClientDisconnected) + return false + } + glog.V(1).Infof("reject upload from %s: no upload slot within %v", r.RemoteAddr, s3a.option.InflightUploadDataTimeout) + s3err.WriteErrorResponse(w, rejected, s3err.ErrTooManyRequest) + return false +} + func (cb *CircuitBreaker) limit(r *http.Request, bucket string, action string) (rollback []func(), errCode s3err.ErrorCode) { //bucket simultaneous request count diff --git a/weed/s3api/s3api_circuit_breaker_test.go b/weed/s3api/s3api_circuit_breaker_test.go index cf5e7817d2..0749d5e9d0 100644 --- a/weed/s3api/s3api_circuit_breaker_test.go +++ b/weed/s3api/s3api_circuit_breaker_test.go @@ -1,11 +1,17 @@ package s3api import ( + "bufio" + "context" + "io" + "net" "net/http" "net/http/httptest" + "strings" "sync" "sync/atomic" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/s3_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -203,3 +209,172 @@ func TestLimitInterceptor(t *testing.T) { } }) } + +func newUploadLimitedCircuitBreaker(option *S3ApiServerOption) *CircuitBreaker { + return &CircuitBreaker{ + counters: make(map[string]*int64), + limitations: make(map[string]int64), + s3a: &S3ApiServer{option: option, inFlightDataLimitCond: sync.NewCond(new(sync.Mutex))}, + } +} + +func putRequest(ctx context.Context, size int) *http.Request { + return httptest.NewRequest(http.MethodPut, "/bucket/object", strings.NewReader(strings.Repeat("x", size))).WithContext(ctx) +} + +type readRecorder struct { + r io.Reader + read bool +} + +func (b *readRecorder) Read(p []byte) (int, error) { + b.read = true + return b.r.Read(p) +} + +func (b *readRecorder) Close() error { return nil } + +func holdUploadSlot(t *testing.T, cb *CircuitBreaker, size int) (release func()) { + entered, done := make(chan struct{}), make(chan struct{}) + finished := make(chan struct{}) + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { + close(entered) + <-done + }, s3_constants.ACTION_WRITE) + go func() { + defer close(finished) + h(httptest.NewRecorder(), putRequest(context.Background(), size)) + }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("first upload was not admitted") + } + return func() { + close(done) + <-finished + } +} + +func TestLimitUploadSlotWait(t *testing.T) { + t.Run("wait for a slot times out with SlowDown", func(t *testing.T) { + cb := newUploadLimitedCircuitBreaker(&S3ApiServerOption{ConcurrentFileUploadLimit: 1, InflightUploadDataTimeout: 100 * time.Millisecond}) + release := holdUploadSlot(t, cb, 10) + defer release() + + ran := false + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { ran = true }, s3_constants.ACTION_WRITE) + rec := httptest.NewRecorder() + body := &readRecorder{r: strings.NewReader("0123456789")} + req := putRequest(context.Background(), 10) + req.Body = body + start := time.Now() + h(rec, req) + if ran { + t.Fatal("upload ran without a free slot") + } + if body.read { + t.Fatal("rejected upload body was read") + } + if rec.Code != http.StatusServiceUnavailable || !strings.Contains(rec.Body.String(), "SlowDown") { + t.Fatalf("got %d %q, want 503 SlowDown", rec.Code, rec.Body.String()) + } + if elapsed := time.Since(start); elapsed < 100*time.Millisecond { + t.Fatalf("rejected after %v, before the wait timeout", elapsed) + } + if n := atomic.LoadInt64(&cb.s3a.inFlightUploads); n != 1 { + t.Fatalf("in-flight uploads %d, want 1", n) + } + }) + + t.Run("cancelled request stops waiting", func(t *testing.T) { + cb := newUploadLimitedCircuitBreaker(&S3ApiServerOption{ConcurrentFileUploadLimit: 1}) + release := holdUploadSlot(t, cb, 10) + defer release() + + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(50*time.Millisecond, cancel) + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { t.Error("cancelled upload ran") }, s3_constants.ACTION_WRITE) + done := make(chan int) + go func() { + rec := httptest.NewRecorder() + h(rec, putRequest(ctx, 10)) + done <- rec.Code + }() + select { + case code := <-done: + if code != 499 { + t.Fatalf("got %d, want 499", code) + } + case <-time.After(5 * time.Second): + t.Fatal("cancelled upload kept waiting for a slot") + } + }) + + t.Run("freed bytes admit every waiter that fits", func(t *testing.T) { + cb := newUploadLimitedCircuitBreaker(&S3ApiServerOption{ConcurrentUploadLimit: 100, InflightUploadDataTimeout: time.Minute}) + release := holdUploadSlot(t, cb, 1000) + + var admitted sync.WaitGroup + admitted.Add(2) + proceed := make(chan struct{}) + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { + admitted.Done() + <-proceed + }, s3_constants.ACTION_WRITE) + var finished sync.WaitGroup + for i := 0; i < 2; i++ { + finished.Add(1) + go func() { + defer finished.Done() + h(httptest.NewRecorder(), putRequest(context.Background(), 10)) + }() + } + time.Sleep(50 * time.Millisecond) + release() + + bothIn := make(chan struct{}) + go func() { + admitted.Wait() + close(bothIn) + }() + select { + case <-bothIn: + case <-time.After(5 * time.Second): + t.Error("only some waiters were admitted after the large upload finished") + } + close(proceed) + finished.Wait() + }) +} + +func stalledPut(t *testing.T, h http.HandlerFunc) *http.Response { + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + conn, err := net.Dial("tcp", srv.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + conn.SetDeadline(time.Now().Add(3 * time.Second)) + if _, err := io.WriteString(conn, "PUT /bucket/object HTTP/1.1\r\nHost: s3\r\nContent-Length: 10\r\n\r\n"); err != nil { + t.Fatal(err) + } + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("no reply while the body is unsent: %v", err) + } + return resp +} + +func TestLimitUploadSlotRejectsStalledBody(t *testing.T) { + cb := newUploadLimitedCircuitBreaker(&S3ApiServerOption{ConcurrentFileUploadLimit: 1, InflightUploadDataTimeout: 100 * time.Millisecond}) + release := holdUploadSlot(t, cb, 10) + defer release() + + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { t.Error("upload ran without a free slot") }, s3_constants.ACTION_WRITE) + resp := stalledPut(t, h) + if resp.StatusCode != http.StatusServiceUnavailable || !resp.Close { + t.Fatalf("got %d close=%v, want 503 with Connection: close", resp.StatusCode, resp.Close) + } +} diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 26d6006986..f054e32efc 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -64,6 +64,7 @@ type S3ApiServerOption struct { IamConfig string // Advanced IAM configuration file path ConcurrentUploadLimit int64 ConcurrentFileUploadLimit int64 + InflightUploadDataTimeout time.Duration EnableIam bool // Enable embedded IAM API on the same port IamReadOnly bool // Disable IAM write operations on this server Cipher bool // encrypt data on volume servers diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index bfbbcf1bd2..c520509195 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -81,6 +81,7 @@ type FilerOption struct { SaveToFilerLimit int64 ConcurrentUploadLimit int64 ConcurrentFileUploadLimit int64 + InflightUploadDataTimeout time.Duration ShowUIDirectoryDelete bool DownloadMaxBytesPs int64 DiskType string diff --git a/weed/server/filer_server_handlers.go b/weed/server/filer_server_handlers.go index 34e6fe3ad7..3db3f13f29 100644 --- a/weed/server/filer_server_handlers.go +++ b/weed/server/filer_server_handlers.go @@ -3,6 +3,7 @@ package weed_server import ( "context" "errors" + "fmt" "net/http" "path" "strconv" @@ -15,6 +16,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/util" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" "github.com/seaweedfs/seaweedfs/weed/util/version" ) @@ -96,38 +98,19 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) { case http.MethodPost, http.MethodPut: // wait until in flight data is less than the limit contentLength := getContentLength(r) - fs.inFlightDataLimitCond.L.Lock() - inFlightDataSize := atomic.LoadInt64(&fs.inFlightDataSize) - inFlightUploads := atomic.LoadInt64(&fs.inFlightUploads) - - // Wait if either data size limit or file count limit is exceeded - for (fs.option.ConcurrentUploadLimit != 0 && inFlightDataSize > fs.option.ConcurrentUploadLimit) || (fs.option.ConcurrentFileUploadLimit != 0 && inFlightUploads >= fs.option.ConcurrentFileUploadLimit) { - if fs.option.ConcurrentUploadLimit != 0 && inFlightDataSize > fs.option.ConcurrentUploadLimit { - glog.V(4).Infof("wait because inflight data %d > %d", inFlightDataSize, fs.option.ConcurrentUploadLimit) - } - if fs.option.ConcurrentFileUploadLimit != 0 && inFlightUploads >= fs.option.ConcurrentFileUploadLimit { - glog.V(4).Infof("wait because inflight uploads %d >= %d", inFlightUploads, fs.option.ConcurrentFileUploadLimit) - } - fs.inFlightDataLimitCond.Wait() - inFlightDataSize = atomic.LoadInt64(&fs.inFlightDataSize) - inFlightUploads = atomic.LoadInt64(&fs.inFlightUploads) + if !fs.waitForUploadSlot(w, r, contentLength) { + return } - fs.inFlightDataLimitCond.L.Unlock() - - // Increment counters - newUploads := atomic.AddInt64(&fs.inFlightUploads, 1) - newSize := atomic.AddInt64(&fs.inFlightDataSize, contentLength) - // Update metrics - stats.FilerInFlightUploadCountGauge.Set(float64(newUploads)) - stats.FilerInFlightUploadBytesGauge.Set(float64(newSize)) defer func() { + fs.inFlightDataLimitCond.L.Lock() + defer fs.inFlightDataLimitCond.L.Unlock() // Decrement counters newUploads := atomic.AddInt64(&fs.inFlightUploads, -1) newSize := atomic.AddInt64(&fs.inFlightDataSize, -contentLength) // Update metrics stats.FilerInFlightUploadCountGauge.Set(float64(newUploads)) stats.FilerInFlightUploadBytesGauge.Set(float64(newSize)) - fs.inFlightDataLimitCond.Signal() + fs.inFlightDataLimitCond.Broadcast() }() if r.Method == http.MethodPut { @@ -145,6 +128,50 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) { } } +func (fs *FilerServer) waitForUploadSlot(w http.ResponseWriter, r *http.Request, contentLength int64) bool { + fs.inFlightDataLimitCond.L.Lock() + err := util.WaitUntil(r.Context(), fs.inFlightDataLimitCond, fs.option.InflightUploadDataTimeout, func() bool { + inFlightDataSize := atomic.LoadInt64(&fs.inFlightDataSize) + if fs.option.ConcurrentUploadLimit != 0 && inFlightDataSize > fs.option.ConcurrentUploadLimit { + glog.V(4).Infof("wait because inflight data %d > %d", inFlightDataSize, fs.option.ConcurrentUploadLimit) + return false + } + inFlightUploads := atomic.LoadInt64(&fs.inFlightUploads) + if fs.option.ConcurrentFileUploadLimit != 0 && inFlightUploads >= fs.option.ConcurrentFileUploadLimit { + glog.V(4).Infof("wait because inflight uploads %d >= %d", inFlightUploads, fs.option.ConcurrentFileUploadLimit) + return false + } + return true + }) + if err == nil { + // Increment counters + newUploads := atomic.AddInt64(&fs.inFlightUploads, 1) + newSize := atomic.AddInt64(&fs.inFlightDataSize, contentLength) + // Update metrics + stats.FilerInFlightUploadCountGauge.Set(float64(newUploads)) + stats.FilerInFlightUploadBytesGauge.Set(float64(newSize)) + } + fs.inFlightDataLimitCond.L.Unlock() + + if err == nil { + return true + } + // Closing an HTTP/1 connection stops net/http from reading the unread + // body before the reply. + if r.ProtoMajor == 1 { + w.Header().Set("Connection", "close") + } + if r.Context().Err() != nil { + glog.V(1).Infof("upload from %s cancelled while waiting for an upload slot: %v", r.RemoteAddr, err) + writeJsonError(w, r, util.HttpStatusCancelled, r.Context().Err()) + return false + } + err = fmt.Errorf("reject because no upload slot freed within %v", fs.option.InflightUploadDataTimeout) + glog.V(1).Infof("too many requests from %s: %v", r.RemoteAddr, err) + writeJsonError(w, r, http.StatusTooManyRequests, err) + return false +} + func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Request) { start := time.Now() diff --git a/weed/server/filer_server_handlers_upload_slot_test.go b/weed/server/filer_server_handlers_upload_slot_test.go new file mode 100644 index 0000000000..a81f82aa39 --- /dev/null +++ b/weed/server/filer_server_handlers_upload_slot_test.go @@ -0,0 +1,99 @@ +package weed_server + +import ( + "bufio" + "context" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func filerPutRequest(ctx context.Context) *http.Request { + return httptest.NewRequest(http.MethodPut, "/dir/file", strings.NewReader("data")).WithContext(ctx) +} + +func TestFilerWaitForUploadSlot(t *testing.T) { + newFiler := func(timeout time.Duration) *FilerServer { + fs := &FilerServer{ + option: &FilerOption{ConcurrentFileUploadLimit: 1, InflightUploadDataTimeout: timeout}, + inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)), + } + if !fs.waitForUploadSlot(httptest.NewRecorder(), filerPutRequest(context.Background()), 4) { + t.Fatal("first upload was not admitted") + } + return fs + } + + t.Run("times out with 429", func(t *testing.T) { + fs := newFiler(100 * time.Millisecond) + rec := httptest.NewRecorder() + start := time.Now() + if fs.waitForUploadSlot(rec, filerPutRequest(context.Background()), 4) { + t.Fatal("upload admitted without a free slot") + } + if rec.Code != http.StatusTooManyRequests { + t.Fatalf("got %d, want 429", rec.Code) + } + if elapsed := time.Since(start); elapsed < 100*time.Millisecond { + t.Fatalf("rejected after %v, before the wait timeout", elapsed) + } + if fs.inFlightUploads != 1 || fs.inFlightDataSize != 4 { + t.Fatalf("in-flight uploads %d bytes %d, want 1 and 4", fs.inFlightUploads, fs.inFlightDataSize) + } + }) + + t.Run("stalled body still gets the 429", func(t *testing.T) { + fs := newFiler(100 * time.Millisecond) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if fs.waitForUploadSlot(w, r, r.ContentLength) { + t.Error("upload admitted without a free slot") + } + })) + defer srv.Close() + conn, err := net.Dial("tcp", srv.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(3 * time.Second)) + if _, err := io.WriteString(conn, "PUT /dir/file HTTP/1.1\r\nHost: filer\r\nContent-Length: 10\r\n\r\n"); err != nil { + t.Fatal(err) + } + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("no reply while the body is unsent: %v", err) + } + if resp.StatusCode != http.StatusTooManyRequests || !resp.Close { + t.Fatalf("got %d close=%v, want 429 with Connection: close", resp.StatusCode, resp.Close) + } + }) + + t.Run("cancelled request stops waiting", func(t *testing.T) { + fs := newFiler(0) + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(50*time.Millisecond, cancel) + done := make(chan int) + go func() { + rec := httptest.NewRecorder() + if fs.waitForUploadSlot(rec, filerPutRequest(ctx), 4) { + t.Error("cancelled upload admitted") + } + done <- rec.Code + }() + select { + case code := <-done: + if code != util.HttpStatusCancelled { + t.Fatalf("got %d, want %d", code, util.HttpStatusCancelled) + } + case <-time.After(5 * time.Second): + t.Fatal("cancelled upload kept waiting for a slot") + } + }) +} diff --git a/weed/util/cond_wait.go b/weed/util/cond_wait.go index 43b45de7c2..815ceb54af 100644 --- a/weed/util/cond_wait.go +++ b/weed/util/cond_wait.go @@ -30,3 +30,29 @@ func WaitWithTimeout(ctx context.Context, cond *sync.Cond, timer *time.Timer) in return HttpStatusCancelled } } + +// WaitUntil waits on cond, whose lock the caller holds, until ready returns +// true, ctx is done, or timeout elapses; a zero timeout waits without a bound. +func WaitUntil(ctx context.Context, cond *sync.Cond, timeout time.Duration, ready func() bool) error { + if ready() { + return nil + } + if timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, timeout) + defer cancel() + } + stop := context.AfterFunc(ctx, func() { + cond.L.Lock() + cond.Broadcast() + cond.L.Unlock() + }) + defer stop() + for !ready() { + if err := ctx.Err(); err != nil { + return err + } + cond.Wait() + } + return nil +} diff --git a/weed/util/cond_wait_test.go b/weed/util/cond_wait_test.go new file mode 100644 index 0000000000..ceca3c9b14 --- /dev/null +++ b/weed/util/cond_wait_test.go @@ -0,0 +1,53 @@ +package util + +import ( + "context" + "errors" + "sync" + "testing" + "time" +) + +func TestWaitUntilReturnsWhenReady(t *testing.T) { + cond := sync.NewCond(new(sync.Mutex)) + ready := false + go func() { + time.Sleep(20 * time.Millisecond) + cond.L.Lock() + ready = true + cond.L.Unlock() + cond.Broadcast() + }() + cond.L.Lock() + err := WaitUntil(context.Background(), cond, time.Minute, func() bool { return ready }) + cond.L.Unlock() + if err != nil { + t.Fatalf("WaitUntil: %v", err) + } +} + +func TestWaitUntilTimesOutWithoutWakeup(t *testing.T) { + cond := sync.NewCond(new(sync.Mutex)) + start := time.Now() + cond.L.Lock() + err := WaitUntil(context.Background(), cond, 50*time.Millisecond, func() bool { return false }) + cond.L.Unlock() + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("got %v, want deadline exceeded", err) + } + if elapsed := time.Since(start); elapsed < 50*time.Millisecond || elapsed > 5*time.Second { + t.Fatalf("returned after %v", elapsed) + } +} + +func TestWaitUntilStopsOnCancel(t *testing.T) { + cond := sync.NewCond(new(sync.Mutex)) + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(20*time.Millisecond, cancel) + cond.L.Lock() + err := WaitUntil(ctx, cond, 0, func() bool { return false }) + cond.L.Unlock() + if !errors.Is(err, context.Canceled) { + t.Fatalf("got %v, want canceled", err) + } +}