diff --git a/quota_hardening_test.go b/quota_hardening_test.go index 7194809..04f5510 100644 --- a/quota_hardening_test.go +++ b/quota_hardening_test.go @@ -7,6 +7,7 @@ import ( "net" "net/http" "net/http/httptest" + "os" "strings" "sync" "sync/atomic" @@ -240,15 +241,21 @@ func TestXHTTPSessionsIgnoreLegacyGlobalCapAndReleaseCounters(t *testing.T) { } } -func TestNegativeXHTTPSessionLimitMeansUnlimited(t *testing.T) { - if got := (&nativeInbound{}).xhttpMaxActiveSessions(); got != 0 { - t.Fatalf("unlimited XHTTP session limit normalized to %d", got) +func TestXHTTPSessionSafetyWindowIsAboveProductionScale(t *testing.T) { + if got := (&nativeInbound{}).xhttpMaxActiveSessions(); got != fixedNativeMaxXHTTPSessions { + t.Fatalf("XHTTP session safety window = %d, want %d", got, fixedNativeMaxXHTTPSessions) + } + if got := nativeXHTTPSessionLimit(); got < 8_000 { + t.Fatalf("XHTTP session safety window = %d, want room for at least 8K users", got) } } -func TestNativeHTTP2StreamsIgnoreLegacyRequestCeiling(t *testing.T) { - if got := nativeHTTP2MaxConcurrentStreams(); got != ^uint32(0) { - t.Fatalf("HTTP/2 stream setting = %d, want unlimited uint32 range", got) +func TestNativeHTTP2StreamsUseFiniteTransportBackpressure(t *testing.T) { + if got := nativeHTTP2MaxConcurrentStreams(); got != fixedNativeHTTP2ConcurrentStreams { + t.Fatalf("HTTP/2 stream setting = %d, want %d", got, fixedNativeHTTP2ConcurrentStreams) + } + if got := nativeHTTP2MaxConcurrentStreams(); got < 1024 { + t.Fatalf("HTTP/2 stream setting = %d, too small for XHTTP packet bursts", got) } if got := nativeMuxMaxSessionLimit(); got != 64 { t.Fatalf("per-transport Mux session guard = %d, want 64", got) @@ -265,6 +272,20 @@ func TestXHTTPHandlerDoesNotApplyWebRequestCeiling(t *testing.T) { } } +func TestXHTTPHandlerSafetySlotIsReleased(t *testing.T) { + before := nativeXHTTPRequests.Load() + ib := &nativeInbound{transport: "xhttp", path: "/"} + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + ib.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("empty XHTTP request status = %d, want 400", rec.Code) + } + if got := nativeXHTTPRequests.Load(); got != before { + t.Fatalf("XHTTP handler counter after return = %d, want %d", got, before) + } +} + func TestPersistedXHTTPAdmissionTuningIsAlwaysUnlimited(t *testing.T) { got := normalizeNativeXrayTuning(&XrayNativeTuning{ MuxGlobalSessions: 8192, @@ -411,6 +432,80 @@ func TestXHTTPUploadQueueBackpressuresInsteadOfRejectingBurst(t *testing.T) { } } +func TestXHTTPGlobalMemoryBackpressureWakesWaitersFIFO(t *testing.T) { + before := nativeXHTTPBufferedBytes.Load() + fillBytes := nativeXHTTPMaxBufferedGlobalBytes - before + filler, ok := acquireNativeXHTTPMemory(fillBytes) + if !ok { + t.Fatal("failed to fill XHTTP memory budget for waiter test") + } + defer filler.release() + + type result struct { + lease *nativeXHTTPMemoryLease + err error + } + startWaiter := func() <-chan result { + done := make(chan result, 1) + go func() { + lease, err := acquireNativeXHTTPMemoryContext(context.Background(), nativeXHTTPMinPacketAccountingBytes) + done <- result{lease: lease, err: err} + }() + return done + } + waitForWaiters := func(want int) { + t.Helper() + deadline := time.Now().Add(time.Second) + for { + nativeXHTTPMemoryWait.Lock() + got := nativeXHTTPMemoryWait.queued + nativeXHTTPMemoryWait.Unlock() + if got == want { + return + } + if time.Now().After(deadline) { + t.Fatalf("memory waiters = %d, want %d", got, want) + } + time.Sleep(time.Millisecond) + } + } + + firstDone := startWaiter() + waitForWaiters(1) + secondDone := startWaiter() + waitForWaiters(2) + + filler.shrink(fillBytes - nativeXHTTPMinPacketAccountingBytes) + first := <-firstDone + if first.err != nil || first.lease == nil { + t.Fatalf("first memory waiter = (%v, %v)", first.lease, first.err) + } + select { + case second := <-secondDone: + if second.lease != nil { + second.lease.release() + } + t.Fatalf("second waiter woke before FIFO capacity was released: %v", second.err) + case <-time.After(25 * time.Millisecond): + } + + first.lease.release() + select { + case second := <-secondDone: + if second.err != nil || second.lease == nil { + t.Fatalf("second memory waiter = (%v, %v)", second.lease, second.err) + } + second.lease.release() + case <-time.After(time.Second): + t.Fatal("second memory waiter did not wake after first released") + } + + filler.release() + if got := nativeXHTTPBufferedBytes.Load(); got != before { + t.Fatalf("FIFO waiter test leaked %d buffered bytes (baseline %d)", got, before) + } +} + func TestXHTTPReassemblyHasNoPacketRequestCountCeiling(t *testing.T) { before := nativeXHTTPBufferedBytes.Load() q := newNativeXHTTPUploadQueue(1, 4*nativeXHTTPMinPacketAccountingBytes) @@ -678,3 +773,98 @@ func TestNativeXHTTPQueueCloseClosesQueuedStreamReader(t *testing.T) { t.Fatal("queued stream reader was not closed during queue shutdown") } } + +func TestNativeXHTTPQueueSkipsEmptyPacketsWithoutZeroProgressRead(t *testing.T) { + before := nativeXHTTPBufferedBytes.Load() + q := newNativeXHTTPUploadQueue(2, 2*nativeXHTTPMinPacketAccountingBytes) + defer q.close() + + for seq, payload := range [][]byte{nil, []byte("x")} { + accounted := nativeXHTTPAccountedPacketBytes(int64(len(payload))) + lease, ok := acquireNativeXHTTPMemory(accounted) + if !ok { + t.Fatal("failed to reserve packet memory") + } + if err := q.push(context.Background(), nativeXHTTPPacket{Payload: payload, Seq: uint64(seq)}, lease); err != nil { + lease.release() + t.Fatalf("queue packet %d: %v", seq, err) + } + lease.release() + } + + buf := make([]byte, 1) + n, err := q.Read(buf) + if err != nil || n != 1 || string(buf[:n]) != "x" { + t.Fatalf("queue read after empty packet = (%d, %v, %q), want (1, nil, x)", n, err, buf[:n]) + } + q.close() + if got := nativeXHTTPBufferedBytes.Load(); got != before { + t.Fatalf("empty-packet test leaked %d buffered bytes (baseline %d)", got, before) + } +} + +type deadlineUnblockingResponseWriter struct { + header http.Header + writeStart chan struct{} + unblock chan struct{} + startOnce sync.Once + unblockOnce sync.Once +} + +func newDeadlineUnblockingResponseWriter() *deadlineUnblockingResponseWriter { + return &deadlineUnblockingResponseWriter{ + header: make(http.Header), + writeStart: make(chan struct{}), + unblock: make(chan struct{}), + } +} + +func (w *deadlineUnblockingResponseWriter) Header() http.Header { return w.header } +func (w *deadlineUnblockingResponseWriter) WriteHeader(int) {} +func (w *deadlineUnblockingResponseWriter) Flush() {} +func (w *deadlineUnblockingResponseWriter) Write([]byte) (int, error) { + w.startOnce.Do(func() { close(w.writeStart) }) + <-w.unblock + return 0, os.ErrDeadlineExceeded +} +func (w *deadlineUnblockingResponseWriter) SetWriteDeadline(deadline time.Time) error { + if !deadline.IsZero() && !deadline.After(time.Now().Add(10*time.Millisecond)) { + w.unblockOnce.Do(func() { close(w.unblock) }) + } + return nil +} + +func TestNativeXHTTPResponseCloseInterruptsStalledWrite(t *testing.T) { + underlying := newDeadlineUnblockingResponseWriter() + writer := newNativeXHTTPResponseWriter(underlying) + writeDone := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("blocked")) + writeDone <- err + }() + + select { + case <-underlying.writeStart: + case <-time.After(time.Second): + t.Fatal("response write did not start") + } + + closeDone := make(chan struct{}) + go func() { + writer.close() + close(closeDone) + }() + select { + case <-closeDone: + case <-time.After(time.Second): + t.Fatal("response close blocked behind stalled write") + } + select { + case err := <-writeDone: + if !errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("stalled write error = %v, want deadline exceeded", err) + } + case <-time.After(time.Second): + t.Fatal("stalled response write was not interrupted") + } +} diff --git a/xray_native_safety.go b/xray_native_safety.go index 27f0536..fa3f20c 100644 --- a/xray_native_safety.go +++ b/xray_native_safety.go @@ -36,8 +36,10 @@ func init() { var ( nativeTransportConnections atomic.Int64 nativeXHTTPSessions atomic.Int64 + nativeXHTTPRequests atomic.Int64 nativeClientConnsRejected atomic.Int64 nativePreAuthRejected atomic.Int64 + nativeXHTTPRejected atomic.Int64 nativeTransportAccepting atomic.Bool nativeTransportRegistry = struct { @@ -47,8 +49,8 @@ var ( ) // acquireNativeCounter tracks a counted resource and returns an exactly-once -// release function. Native transport/XHTTP admission calls it with limit=0 -// because VPN traffic must not be rejected by a global website-style ceiling. +// release function. Limits here are simultaneous resource-safety windows, not +// traffic-volume or request-rate ceilings. func acquireNativeCounter(active *atomic.Int64, limit int) (func(), bool) { for { current := active.Load() @@ -96,11 +98,15 @@ func logNativePreAuthRejection(format string, args ...interface{}) { } func acquireNativeTransportConnection() (func(), bool) { - return acquireNativeCounter(&nativeTransportConnections, 0) + return acquireNativeCounter(&nativeTransportConnections, nativeTransportConnectionLimit()) } func acquireNativeXHTTPSession() (func(), bool) { - return acquireNativeCounter(&nativeXHTTPSessions, 0) + return acquireNativeCounter(&nativeXHTTPSessions, nativeXHTTPSessionLimit()) +} + +func acquireNativeXHTTPRequest() (func(), bool) { + return acquireNativeCounter(&nativeXHTTPRequests, nativeXHTTPRequestLimit()) } func configureNativeTransportSocket(c net.Conn) { @@ -192,8 +198,9 @@ func registerTrackedNativeTransportConn(c net.Conn, release func()) (net.Conn, b return counted, true } -// waitWrapTrackedNativeTransportConn is used by raw native accept loops. Global -// admission is unlimited; the loop remains only to coordinate listener shutdown. +// waitWrapTrackedNativeTransportConn is used by raw native accept loops. Waiting +// here, before another connection is admitted to the protocol handler, applies +// socket/kernel backpressure instead of creating an unbounded goroutine backlog. func waitWrapTrackedNativeTransportConn(c net.Conn) (net.Conn, bool) { if c == nil { return nil, false @@ -231,8 +238,8 @@ func closeAllNativeTransportConnections() { } // nativeTrackingListener registers every accepted XHTTP socket so a live -// stop/reload can close it. It counts sockets for diagnostics but never rejects -// or delays one because of a global application limit. +// stop/reload can close it. It reserves capacity before Accept so overload stays +// in the kernel accept queue rather than allocating more Go handlers. type nativeTrackingListener struct { net.Listener } diff --git a/xray_native_tuning.go b/xray_native_tuning.go index 2798e43..1e71fd2 100644 --- a/xray_native_tuning.go +++ b/xray_native_tuning.go @@ -37,6 +37,18 @@ const ( // buffers. Operators may request more, up to the hard cap enforced there. defaultNativeXHTTPBufferedPosts = 64 + // These are simultaneous resource-safety windows, not request-rate limits. + // They are deliberately far above the expected 6-8K connected-user load, but + // finite so a reconnect storm, broken CDN, or hostile client cannot retain an + // unbounded number of sockets, HTTP handlers, sessions, and goroutine stacks. + // Transport Accept waits at capacity (kernel backpressure); XHTTP overloads + // receive 503 rather than the web-rate-limit semantics of 429. + fixedNativeMaxTransportConnections = 65536 + fixedNativeMaxXHTTPRequests = 65536 + fixedNativeMaxXHTTPSessions = 65536 + fixedNativeHTTP2ConcurrentStreams = 4096 + fixedNativeXHTTPWriteTimeoutMS = 60 * 1000 + // Backstop reaper for connected XHTTP VPN sessions. The stream-down GET's // request context is the primary lifetime owner, but behind a CDN that context // frequently never fires when a client silently drops (mobile networks, CDN @@ -111,10 +123,13 @@ func nativeMuxUDPReadBufferSize() int { return fixedNativeMuxUDPReadBuffer } func nativeMuxUDPWriteBufferSize() int { return fixedNativeMuxUDPWriteBuffer } func nativeXHTTPBufferedPostLimit() int { return defaultNativeXHTTPBufferedPosts } func nativeHTTP2MaxConcurrentStreams() uint32 { - // x/net/http2 otherwise installs its own finite default when this is zero. - // Advertise the protocol's full uint32 range so ordinary packet-up bursts can - // never be refused by a website-style concurrent-stream setting. - return ^uint32(0) + return fixedNativeHTTP2ConcurrentStreams +} +func nativeTransportConnectionLimit() int { return fixedNativeMaxTransportConnections } +func nativeXHTTPRequestLimit() int { return fixedNativeMaxXHTTPRequests } +func nativeXHTTPSessionLimit() int { return fixedNativeMaxXHTTPSessions } +func nativeXHTTPWriteTimeout() time.Duration { + return fixedNativeXHTTPWriteTimeoutMS * time.Millisecond } func nativeMuxUDPIdleTimeout() time.Duration { return fixedNativeMuxUDPIdleMS * time.Millisecond diff --git a/xray_xhttp.go b/xray_xhttp.go index a9126b0..324429d 100644 --- a/xray_xhttp.go +++ b/xray_xhttp.go @@ -41,10 +41,18 @@ var ( errNativeXHTTPUploadBufferFull = errors.New("xhttp upload buffer limit reached") nativeXHTTPMemoryWait = struct { sync.Mutex - changed chan struct{} - }{changed: make(chan struct{})} + waiters []*nativeXHTTPMemoryWaiter + head int + queued int + }{} ) +type nativeXHTTPMemoryWaiter struct { + bytes int64 + ready chan struct{} + granted bool +} + const ( xhttpPlacementPath = "path" xhttpPlacementQuery = "query" @@ -328,6 +336,15 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } + releaseRequest, ok := acquireNativeXHTTPRequest() + if !ok { + logNativeLimitRejection("simultaneous XHTTP handlers", &nativeXHTTPRejected, nativeXHTTPRequestLimit()) + w.Header().Set("Retry-After", "1") + http.Error(w, "xhttp transport temporarily busy", http.StatusServiceUnavailable) + return + } + defer releaseRequest() + sessionID, seqStr := ib.extractXHTTPMeta(r, base) if len(sessionID) > nativeXHTTPMaxSessionIDBytes || len(seqStr) > nativeXHTTPMaxSequenceBytes { logNativePreAuthRejection("native xray: xhttp reject inbound=%q reason=metadata-size remote=%s", ib.tag, r.RemoteAddr) @@ -560,7 +577,13 @@ func (ib *nativeInbound) upsertXHTTPSession(w http.ResponseWriter, id string) *n s.touch() return s } - releaseSlot, _ := acquireNativeXHTTPSession() + releaseSlot, ok := acquireNativeXHTTPSession() + if !ok { + logNativeLimitRejection("simultaneous XHTTP sessions", &nativeXHTTPRejected, nativeXHTTPSessionLimit()) + w.Header().Set("Retry-After", "1") + http.Error(w, "xhttp session capacity temporarily busy", http.StatusServiceUnavailable) + return nil + } s := &nativeXHTTPSession{ id: id, queue: newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts, nativeXHTTPMaxBufferedSessionBytes), @@ -576,7 +599,7 @@ func (ib *nativeInbound) upsertXHTTPSession(w http.ResponseWriter, id string) *n } func (ib *nativeInbound) xhttpMaxActiveSessions() int { - return 0 + return nativeXHTTPSessionLimit() } func (ib *nativeInbound) reapUnconnectedXHTTPSession(id string, s *nativeXHTTPSession) { @@ -1048,12 +1071,19 @@ func (c *nativeXHTTPConn) SetReadDeadline(t time.Time) error { return nil } -func (c *nativeXHTTPConn) SetWriteDeadline(time.Time) error { return nil } +func (c *nativeXHTTPConn) SetWriteDeadline(t time.Time) error { + if dw, ok := c.writer.(interface{ SetWriteDeadline(time.Time) error }); ok { + return dw.SetWriteDeadline(t) + } + return nil +} type nativeXHTTPResponseWriter struct { - mu sync.Mutex - w http.ResponseWriter - closed bool + writeMu sync.Mutex + stateMu sync.Mutex + w http.ResponseWriter + closed bool + deadline time.Time } func newNativeXHTTPResponseWriter(w http.ResponseWriter) *nativeXHTTPResponseWriter { @@ -1061,22 +1091,53 @@ func newNativeXHTTPResponseWriter(w http.ResponseWriter) *nativeXHTTPResponseWri } func (w *nativeXHTTPResponseWriter) Write(p []byte) (int, error) { - w.mu.Lock() - defer w.mu.Unlock() - if w.closed { + w.writeMu.Lock() + defer w.writeMu.Unlock() + + w.stateMu.Lock() + closed := w.closed + deadline := w.deadline + w.stateMu.Unlock() + if closed { return 0, io.ErrClosedPipe } + + safetyDeadline := time.Now().Add(nativeXHTTPWriteTimeout()) + if deadline.IsZero() || deadline.After(safetyDeadline) { + deadline = safetyDeadline + } + controller := http.NewResponseController(w.w) + if err := controller.SetWriteDeadline(deadline); err != nil && !errors.Is(err, http.ErrNotSupported) { + return 0, err + } n, err := w.w.Write(p) if err == nil { - flushHTTP(w.w) + if flushErr := controller.Flush(); flushErr != nil && !errors.Is(flushErr, http.ErrNotSupported) { + err = flushErr + } } return n, err } func (w *nativeXHTTPResponseWriter) close() { - w.mu.Lock() + // Do not wait for writeMu: Close is commonly called by the request-context + // watcher specifically because a CDN write is stalled. Mark the writer closed + // and force the active net/http write deadline to expire so Write returns. + w.stateMu.Lock() w.closed = true - w.mu.Unlock() + w.stateMu.Unlock() + _ = http.NewResponseController(w.w).SetWriteDeadline(time.Now()) +} + +func (w *nativeXHTTPResponseWriter) SetWriteDeadline(t time.Time) error { + w.stateMu.Lock() + w.deadline = t + w.stateMu.Unlock() + err := http.NewResponseController(w.w).SetWriteDeadline(t) + if errors.Is(err, http.ErrNotSupported) { + return nil + } + return err } // nativeXHTTPMemoryLease reserves from a process-wide byte budget before a @@ -1091,16 +1152,15 @@ func acquireNativeXHTTPMemory(n int64) (*nativeXHTTPMemoryLease, bool) { if n <= 0 { return &nativeXHTTPMemoryLease{}, true } - for { - current := nativeXHTTPBufferedBytes.Load() - if current > nativeXHTTPMaxBufferedGlobalBytes-n { - logNativeLimitRejection("XHTTP buffered upload bytes", &nativeXHTTPBufferRejected, nativeXHTTPMaxBufferedGlobalBytes) - return nil, false - } - if nativeXHTTPBufferedBytes.CompareAndSwap(current, current+n) { - return &nativeXHTTPMemoryLease{bytes: n}, true - } + nativeXHTTPMemoryWait.Lock() + defer nativeXHTTPMemoryWait.Unlock() + current := nativeXHTTPBufferedBytes.Load() + if nativeXHTTPMemoryWait.queued != 0 || current > nativeXHTTPMaxBufferedGlobalBytes-n { + logNativeLimitRejection("XHTTP buffered upload bytes", &nativeXHTTPBufferRejected, nativeXHTTPMaxBufferedGlobalBytes) + return nil, false } + nativeXHTTPBufferedBytes.Store(current + n) + return &nativeXHTTPMemoryLease{bytes: n}, true } // acquireNativeXHTTPMemoryContext applies process-wide memory backpressure. @@ -1113,27 +1173,42 @@ func acquireNativeXHTTPMemoryContext(ctx context.Context, n int64) (*nativeXHTTP if n > nativeXHTTPMaxBufferedGlobalBytes { return nil, errNativeXHTTPUploadBufferFull } - for { - current := nativeXHTTPBufferedBytes.Load() - if current <= nativeXHTTPMaxBufferedGlobalBytes-n && nativeXHTTPBufferedBytes.CompareAndSwap(current, current+n) { - return &nativeXHTTPMemoryLease{bytes: n}, nil - } - - nativeXHTTPMemoryWait.Lock() - // Recheck while holding the generation lock so a release cannot happen - // between the failed check and subscribing to the notification channel. - current = nativeXHTTPBufferedBytes.Load() - if current <= nativeXHTTPMaxBufferedGlobalBytes-n { - nativeXHTTPMemoryWait.Unlock() - continue - } - changed := nativeXHTTPMemoryWait.changed + waiter := &nativeXHTTPMemoryWaiter{bytes: n, ready: make(chan struct{})} + nativeXHTTPMemoryWait.Lock() + current := nativeXHTTPBufferedBytes.Load() + if nativeXHTTPMemoryWait.queued == 0 && current <= nativeXHTTPMaxBufferedGlobalBytes-n { + nativeXHTTPBufferedBytes.Store(current + n) nativeXHTTPMemoryWait.Unlock() - select { - case <-changed: - case <-ctx.Done(): - return nil, ctx.Err() + return &nativeXHTTPMemoryLease{bytes: n}, nil + } + nativeXHTTPMemoryWait.waiters = append(nativeXHTTPMemoryWait.waiters, waiter) + nativeXHTTPMemoryWait.queued++ + nativeXHTTPMemoryWait.Unlock() + + select { + case <-waiter.ready: + return &nativeXHTTPMemoryLease{bytes: n}, nil + case <-ctx.Done(): + nativeXHTTPMemoryWait.Lock() + if waiter.granted { + current := nativeXHTTPBufferedBytes.Load() - n + if current < 0 { + current = 0 + } + nativeXHTTPBufferedBytes.Store(current) + } else { + for i := nativeXHTTPMemoryWait.head; i < len(nativeXHTTPMemoryWait.waiters); i++ { + candidate := nativeXHTTPMemoryWait.waiters[i] + if candidate == waiter { + nativeXHTTPMemoryWait.waiters[i] = nil + nativeXHTTPMemoryWait.queued-- + break + } + } } + grantNativeXHTTPMemoryWaitersLocked() + nativeXHTTPMemoryWait.Unlock() + return nil, ctx.Err() } } @@ -1141,20 +1216,69 @@ func releaseNativeXHTTPMemory(n int64) { if n <= 0 { return } - for { - current := nativeXHTTPBufferedBytes.Load() - next := current - n - if next < 0 { - next = 0 + nativeXHTTPMemoryWait.Lock() + current := nativeXHTTPBufferedBytes.Load() + next := current - n + if next < 0 { + next = 0 + } + nativeXHTTPBufferedBytes.Store(next) + grantNativeXHTTPMemoryWaitersLocked() + nativeXHTTPMemoryWait.Unlock() +} + +// grantNativeXHTTPMemoryWaitersLocked wakes only the FIFO waiters whose exact +// reservations now fit. The former broadcast channel woke every blocked HTTP +// handler after every tiny release, creating a thundering herd and sustained +// multi-core CPU usage while the 128 MB budget was full. +func grantNativeXHTTPMemoryWaitersLocked() { + for nativeXHTTPMemoryWait.queued > 0 { + for nativeXHTTPMemoryWait.head < len(nativeXHTTPMemoryWait.waiters) && + nativeXHTTPMemoryWait.waiters[nativeXHTTPMemoryWait.head] == nil { + nativeXHTTPMemoryWait.head++ } - if nativeXHTTPBufferedBytes.CompareAndSwap(current, next) { - nativeXHTTPMemoryWait.Lock() - close(nativeXHTTPMemoryWait.changed) - nativeXHTTPMemoryWait.changed = make(chan struct{}) - nativeXHTTPMemoryWait.Unlock() + if nativeXHTTPMemoryWait.head >= len(nativeXHTTPMemoryWait.waiters) { + nativeXHTTPMemoryWait.waiters = nil + nativeXHTTPMemoryWait.head = 0 + nativeXHTTPMemoryWait.queued = 0 return } + + waiter := nativeXHTTPMemoryWait.waiters[nativeXHTTPMemoryWait.head] + current := nativeXHTTPBufferedBytes.Load() + if current > nativeXHTTPMaxBufferedGlobalBytes-waiter.bytes { + compactNativeXHTTPMemoryWaitersLocked() + return + } + nativeXHTTPMemoryWait.waiters[nativeXHTTPMemoryWait.head] = nil + nativeXHTTPMemoryWait.head++ + nativeXHTTPMemoryWait.queued-- + nativeXHTTPBufferedBytes.Store(current + waiter.bytes) + waiter.granted = true + close(waiter.ready) } + compactNativeXHTTPMemoryWaitersLocked() +} + +func compactNativeXHTTPMemoryWaitersLocked() { + head := nativeXHTTPMemoryWait.head + if head == 0 { + return + } + if nativeXHTTPMemoryWait.queued == 0 { + nativeXHTTPMemoryWait.waiters = nil + nativeXHTTPMemoryWait.head = 0 + return + } + if head < 1024 && head*2 < len(nativeXHTTPMemoryWait.waiters) { + return + } + remaining := copy(nativeXHTTPMemoryWait.waiters, nativeXHTTPMemoryWait.waiters[head:]) + for i := remaining; i < len(nativeXHTTPMemoryWait.waiters); i++ { + nativeXHTTPMemoryWait.waiters[i] = nil + } + nativeXHTTPMemoryWait.waiters = nativeXHTTPMemoryWait.waiters[:remaining] + nativeXHTTPMemoryWait.head = 0 } func (l *nativeXHTTPMemoryLease) shrink(n int64) { @@ -1427,71 +1551,74 @@ func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) { return reader.Read(b) } - select { - case <-q.closed: - return 0, io.EOF - default: - } - - if len(q.heap) == 0 { - p, err := q.recv() - if err != nil { - return 0, err - } - if p.Reader != nil { - if !q.setReader(p.Reader) { - _ = p.Reader.Close() - return 0, io.EOF - } - return p.Reader.Read(b) - } + for { select { case <-q.closed: - q.releasePayloadMemory(p.accountedBytes) return 0, io.EOF default: } - heap.Push(&q.heap, p) - } - for len(q.heap) > 0 { - packet := heap.Pop(&q.heap).(nativeXHTTPPacket) - - if packet.Seq == q.nextSeq { - n := copy(b, packet.Payload) - if n < len(packet.Payload) { - q.releasePayloadMemory(int64(n)) - packet.accountedBytes -= int64(n) - packet.Payload = packet.Payload[n:] - heap.Push(&q.heap, packet) - } else { - q.releasePayloadMemory(packet.accountedBytes) - q.nextSeq = packet.Seq + 1 - } - return n, nil - } - - if packet.Seq > q.nextSeq { - // Do not apply a packet/request count ceiling. The per-session and global - // accounted-byte budgets backpressure producers, including empty packets. - heap.Push(&q.heap, packet) + if len(q.heap) == 0 { p, err := q.recv() if err != nil { return 0, err } if p.Reader != nil { - _ = p.Reader.Close() - return 0, errors.New("xhttp mixed stream-up and packet-up upload") + if !q.setReader(p.Reader) { + _ = p.Reader.Close() + return 0, io.EOF + } + return p.Reader.Read(b) + } + select { + case <-q.closed: + q.releasePayloadMemory(p.accountedBytes) + return 0, io.EOF + default: } heap.Push(&q.heap, p) - continue } - // A duplicate/late packet is discarded; release the bytes it owned. - q.releasePayloadMemory(packet.accountedBytes) - } + for len(q.heap) > 0 { + packet := heap.Pop(&q.heap).(nativeXHTTPPacket) - return 0, nil + if packet.Seq == q.nextSeq { + if len(packet.Payload) == 0 { + q.releasePayloadMemory(packet.accountedBytes) + q.nextSeq = packet.Seq + 1 + continue + } + n := copy(b, packet.Payload) + if n < len(packet.Payload) { + q.releasePayloadMemory(int64(n)) + packet.accountedBytes -= int64(n) + packet.Payload = packet.Payload[n:] + heap.Push(&q.heap, packet) + } else { + q.releasePayloadMemory(packet.accountedBytes) + q.nextSeq = packet.Seq + 1 + } + return n, nil + } + + if packet.Seq > q.nextSeq { + heap.Push(&q.heap, packet) + p, err := q.recv() + if err != nil { + return 0, err + } + if p.Reader != nil { + _ = p.Reader.Close() + return 0, errors.New("xhttp mixed stream-up and packet-up upload") + } + heap.Push(&q.heap, p) + continue + } + + // A duplicate/late packet is discarded; release the bytes it owned. + q.releasePayloadMemory(packet.accountedBytes) + } + } } func (q *nativeXHTTPUploadQueue) loadReader() io.ReadCloser {