From fcfca3fee373943cec5e6c9ddf6791352c0b97e2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 11 Nov 2026 10:17:18 -0800 Subject: [PATCH] s3, filer: bound the wait for an upload slot (#12559) * util: add WaitUntil, a cond wait bounded by a context and a timeout The caller holds the cond lock. Cancellation and the timeout wake the waiter through a broadcast taken under the same lock, so no wakeup is lost. Co-Authored-By: Claude Opus 5.5 * s3: bound the wait for an upload slot A PUT that arrived while the concurrent upload limits were exhausted waited on the condition variable with no bound. The handler reads nothing from the body while it waits, so a client that gave up and reset its connection went unnoticed: the socket and goroutine stayed pinned until a slot freed, and a stalled volume server grew the gateway's fd count by one per aborted PUT. The wait now ends on InflightUploadDataTimeout with 503 SlowDown, which S3 SDKs retry, and on request cancellation. An admitted upload takes its slot under the lock, and a finished upload broadcasts, so freed bytes admit every waiter that fits instead of one. Co-Authored-By: Claude Opus 5.5 * filer: bound the wait for an upload slot The filer's PUT/POST upload limits queued requests the same unbounded way. Reuse the bounded wait: reject with 429 after InflightUploadDataTimeout, as the volume server does, and stop on request cancellation. Co-Authored-By: Claude Opus 5.5 * command: add -inflightUploadDataTimeout for the S3 gateway and the filer weed s3 -inflightUploadDataTimeout, weed filer -inflightUploadDataTimeout and -s3.inflightUploadDataTimeout, weed server and weed mini -filer.inflightUploadDataTimeout and -s3.inflightUploadDataTimeout. The default of 60s matches the volume server's flag of the same name; 0 keeps the old unbounded wait. Co-Authored-By: Claude Opus 5.5 * s3: reject a queued upload without reading its body WriteErrorResponse drains the request body before answering. Behind the stats recorder the drain has no read deadline, and reading the body sends an Expect: 100-continue client the go-ahead to upload what is about to be discarded. Answer the rejection with an empty-body copy of the request so the drain is skipped; net/http closes the connection when a large unread body remains. Co-Authored-By: Claude Opus 5.5 * s3, filer: release an upload slot under the cond lock A waiter checks the counters under the cond lock and then waits. A release that decremented and broadcast without the lock could land between the two, so the waiter slept with a slot free until the next release or its timeout. Co-Authored-By: Claude Opus 5.5 * s3, filer: close the HTTP/1 connection when rejecting a queued upload net/http reads up to 256 KiB of an unread request body before sending a keep-alive reply, so a client that stalls its body held back the 503 or 429 until the connection idle timeout. Connection: close skips that read. HTTP/2 is left alone: there it would shut down the whole connection. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/command/filer.go | 4 + weed/command/mini.go | 2 + weed/command/s3.go | 3 + weed/command/server.go | 2 + weed/s3api/s3api_circuit_breaker.go | 79 +++++--- weed/s3api/s3api_circuit_breaker_test.go | 175 ++++++++++++++++++ weed/s3api/s3api_server.go | 1 + weed/server/filer_server.go | 1 + weed/server/filer_server_handlers.go | 75 +++++--- .../filer_server_handlers_upload_slot_test.go | 99 ++++++++++ weed/util/cond_wait.go | 26 +++ weed/util/cond_wait_test.go | 53 ++++++ 12 files changed, 470 insertions(+), 50 deletions(-) create mode 100644 weed/server/filer_server_handlers_upload_slot_test.go create mode 100644 weed/util/cond_wait_test.go 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) + } +}