From 47a10c0012e13f82401e0dd7cbec18a48b43a02a Mon Sep 17 00:00:00 2001 From: RPRX <63339210+RPRX@users.noreply.github.com> Date: Fri, 6 Nov 2026 04:08:29 +0000 Subject: [PATCH] XDRIVE transport: A bunch of optimizations for Google Drive --- transport/internet/xdrive/client.go | 27 +- transport/internet/xdrive/client_test.go | 100 +++++ transport/internet/xdrive/drive.go | 101 ++++- transport/internet/xdrive/drive_live_test.go | 132 +++++++ transport/internet/xdrive/params.go | 2 +- transport/internet/xdrive/slots.go | 245 +++++++++++- transport/internet/xdrive/slots_test.go | 370 ++++++++++++++++++- transport/internet/xdrive/xdrive.go | 164 +++++++- transport/internet/xdrive/xdrive_test.go | 7 + 9 files changed, 1092 insertions(+), 56 deletions(-) diff --git a/transport/internet/xdrive/client.go b/transport/internet/xdrive/client.go index d1ccf3faa..8bb4f2f15 100644 --- a/transport/internet/xdrive/client.go +++ b/transport/internet/xdrive/client.go @@ -438,9 +438,11 @@ func (b *doneBody) Close() error { type sessionTransport struct { pool *connPool - mu sync.Mutex - client *splithttp.XmuxClient - closed bool + mu sync.Mutex + client *splithttp.XmuxClient + closed bool + slowAt, retiredAt time.Time + movedAt time.Time } func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) { @@ -450,7 +452,7 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) { return nil, errSessionClosed } client := t.client - if client == nil || client.LeftRequests.Add(-1) < 0 || !client.UnreusableAt.IsZero() && time.Now().After(client.UnreusableAt) { + if client == nil || client.XmuxConn.IsClosed() || client.LeftRequests.Add(-1) < 0 || !client.UnreusableAt.IsZero() && time.Now().After(client.UnreusableAt) { t.pool.mu.Lock() t.client = t.pool.xmux.GetXmuxClient(r.Context()) t.pool.mu.Unlock() @@ -458,6 +460,7 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) { t.client.LeftRequests.Add(-1) if client != nil { client.DoneRunning() + t.movedAt = time.Now() } client = t.client } @@ -471,6 +474,22 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) { return conn.roundTrip(r) } +func (t *sessionTransport) stalled(began time.Time) { + t.mu.Lock() + defer t.mu.Unlock() + if began.Before(t.movedAt) { + return + } + now := time.Now() + slow := now.Sub(t.slowAt) < 3*time.Second + t.slowAt = now + if !slow || t.client == nil || now.Sub(t.retiredAt) < 2*time.Second { + return + } + t.retiredAt = now + t.client.XmuxConn.(*poolConn).Close() +} + func (t *sessionTransport) Close() { t.mu.Lock() defer t.mu.Unlock() diff --git a/transport/internet/xdrive/client_test.go b/transport/internet/xdrive/client_test.go index 58b3f2511..574e20687 100644 --- a/transport/internet/xdrive/client_test.go +++ b/transport/internet/xdrive/client_test.go @@ -987,3 +987,103 @@ func TestDriveXmuxOverH3(t *testing.T) { } waitFor(t, "the first connection of XMUX to close", func() bool { return server.closed.Load() == 1 }) } + +type stallConn struct{} + +func (stallConn) RoundTrip(r *http.Request) (*http.Response, error) { + <-r.Context().Done() + return nil, r.Context().Err() +} + +type okConn struct{} + +func (okConn) RoundTrip(r *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"id":"file-1"}`)), + }, nil +} + +func TestSessionRetiresStalledConn(t *testing.T) { + made := atomic.Int32{} + transport := &sessionTransport{pool: newConnPool(nil, func(connCtx context.Context) *http.Client { + if made.Add(1) == 1 { + return &http.Client{Transport: stallConn{}} + } + return &http.Client{Transport: okConn{}} + })} + storage := &driveStorage{ + driveShared: &driveShared{ + token: "token", + tokenExpiry: time.Now().Add(time.Hour), + backoff: time.Millisecond, + }, + client: &http.Client{Transport: transport}, + inflight: make(chan struct{}, 1), + reserved: make(chan struct{}, 1), + } + + ctx := context.WithValue(context.Background(), attemptKey{}, 100*time.Millisecond) + code, body, err := storage.do(ctx, http.MethodPost, "http://drive.invalid/files", "", nil) + if err != nil { + t.Fatalf("the first request over a stalled connection: %v", err) + } + if code != http.StatusOK || string(body) != `{"id":"file-1"}` { + t.Fatalf("the stalled connection gave %d %q", code, body) + } + if n := made.Load(); n != 2 { + t.Fatalf("a stalled connection took %d connections, want it retired and one made after it", n) + } + if _, _, err := storage.do(ctx, http.MethodPost, "http://drive.invalid/files", "", nil); err != nil { + t.Fatalf("the request after the retirement: %v", err) + } + if n := made.Load(); n != 2 { + t.Fatalf("%d connections were made, want the healthy one kept", n) + } +} + +func TestSessionIgnoresStallOfRetiredConn(t *testing.T) { + var made atomic.Int32 + transport := &sessionTransport{pool: newConnPool(nil, func(connCtx context.Context) *http.Client { + made.Add(1) + return &http.Client{Transport: okConn{}} + })} + send := func() { + req, err := http.NewRequest(http.MethodGet, "http://drive.invalid/files", nil) + if err != nil { + t.Fatal(err) + } + resp, err := transport.RoundTrip(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + } + send() + old := time.Now() + time.Sleep(2 * time.Millisecond) + transport.stalled(old) + transport.stalled(old) + send() + if n := made.Load(); n != 2 { + t.Fatalf("a stalled connection took %d connections, want it retired and one made after it", n) + } + transport.mu.Lock() + transport.slowAt, transport.retiredAt = time.Time{}, time.Time{} + transport.mu.Unlock() + transport.stalled(old) + transport.stalled(old) + send() + if n := made.Load(); n != 2 { + t.Fatalf("the stall of a retired connection took the healthy one too, %d connections", n) + } + time.Sleep(2 * time.Millisecond) + fresh := time.Now() + transport.stalled(fresh) + transport.stalled(fresh) + send() + if n := made.Load(); n != 3 { + t.Fatalf("a stall of the current connection made %d connections, want it retired", n) + } +} diff --git a/transport/internet/xdrive/drive.go b/transport/internet/xdrive/drive.go index 658b14a5a..7efe7ac29 100644 --- a/transport/internet/xdrive/drive.go +++ b/transport/internet/xdrive/drive.go @@ -21,11 +21,12 @@ import ( const ( flatSeparator = "~" - driveBoundary = "xdrive-boundary" - drivePageSize = 1000 - driveMaxAttempts = 8 - driveMaxInflight = 32 - driveInlineLimit = 12000 + driveBoundary = "xdrive-boundary" + drivePageSize = 1000 + driveMaxAttempts = 8 + driveMaxInflight = 64 + driveReservedInflight = 8 + driveInlineLimit = 12000 // driveTimeout ends a request that makes no progress for that long driveTimeout = 60 * time.Second ) @@ -36,13 +37,21 @@ var ( driveFilesURL = "https://www.googleapis.com/drive/v3/files" driveUploadURL = "https://www.googleapis.com/upload/drive/v3/files?uploadType=multipart&fields=id,name" driveInitialBackoff = 200 * time.Millisecond + driveCooldown = 2 * time.Second + driveSlotRead = 3 * time.Second + driveSlotWrite = 5 * time.Second ) +type laneKey struct{} + +type attemptKey struct{} + type driveStorage struct { *driveShared // a session has a client and inflight of its own, see session client *http.Client inflight chan struct{} + reserved chan struct{} release func() } @@ -65,6 +74,9 @@ type driveShared struct { idMu sync.Mutex ids map[string]string + + coolMu sync.Mutex + coolUntil time.Time } func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config) (*driveStorage, error) { @@ -100,6 +112,7 @@ func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config }, client: client, inflight: make(chan struct{}, driveMaxInflight), + reserved: make(chan struct{}, driveReservedInflight), }, nil } @@ -111,6 +124,7 @@ func (s *driveStorage) session() Storage { driveShared: s.driveShared, client: &http.Client{Transport: transport}, inflight: make(chan struct{}, driveMaxInflight), + reserved: make(chan struct{}, driveReservedInflight), release: transport.Close, } } @@ -219,6 +233,17 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin backoff := s.backoff var lastErr error + s.coolMu.Lock() + until := s.coolUntil + s.coolMu.Unlock() + if d := time.Until(until); d > 0 { + select { + case <-ctx.Done(): + return 0, nil, ctx.Err() + case <-time.After(d): + } + } + for attempt := 0; attempt < driveMaxAttempts; attempt++ { if attempt > 0 { select { @@ -238,8 +263,12 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin continue } + pool := s.inflight + if ctx.Value(laneKey{}) != nil { + pool = s.reserved + } select { - case s.inflight <- struct{}{}: + case pool <- struct{}{}: case <-ctx.Done(): return 0, nil, ctx.Err() } @@ -248,8 +277,14 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin if body != nil { reader = bytes.NewReader(body) } - req, err := http.NewRequestWithContext(ctx, method, target, reader) + attemptCtx, cancel := ctx, context.CancelFunc(func() {}) + if d, ok := ctx.Value(attemptKey{}).(time.Duration); ok { + attemptCtx, cancel = context.WithTimeout(ctx, d) + } + req, err := http.NewRequestWithContext(attemptCtx, method, target, reader) if err != nil { + cancel() + <-pool return 0, nil, errors.New("failed to build a Drive request").Base(err) } req.Header.Set("Authorization", "Bearer "+token) @@ -257,9 +292,17 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin req.Header.Set("Content-Type", contentType) } + began := time.Now() resp, err := s.client.Do(req) - <-s.inflight + <-pool if err != nil { + timedOut := attemptCtx.Err() == context.DeadlineExceeded && ctx.Err() == nil + cancel() + if timedOut { + if t, ok := s.client.Transport.(*sessionTransport); ok { + t.stalled(began) + } + } if ctx.Err() != nil { return 0, nil, ctx.Err() } @@ -270,6 +313,7 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin } payload, err := io.ReadAll(resp.Body) resp.Body.Close() + cancel() if err != nil { if ctx.Err() != nil { return 0, nil, ctx.Err() @@ -286,13 +330,19 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin if resp.StatusCode == http.StatusForbidden && rateLimited(payload) { lastErr = errors.New("Drive is rate limiting: ", string(payload)) errors.LogWarning(ctx, "rate limited by Drive, attempt ", - attempt+1, " of ", driveMaxAttempts) + attempt+1, " of ", driveMaxAttempts, ", ", method, " ", len(body), + " bytes, body read ", strings.Contains(target, "alt=media"), + ", lane ", ctx.Value(laneKey{}) != nil, ": ", string(payload[:min(len(payload), 300)])) + s.cool() continue } if retryableStatus(resp.StatusCode) { lastErr = errors.New("Drive answered ", resp.StatusCode, ": ", string(payload)) errors.LogWarning(ctx, "retrying after Drive answered ", resp.StatusCode, ", attempt ", attempt+1, " of ", driveMaxAttempts) + if resp.StatusCode == http.StatusTooManyRequests { + s.cool() + } continue } return resp.StatusCode, payload, nil @@ -301,6 +351,12 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin return 0, nil, lastErr } +func (s *driveStorage) cool() { + s.coolMu.Lock() + s.coolUntil = time.Now().Add(jitter(driveCooldown)) + s.coolMu.Unlock() +} + func (s *driveStorage) invalidateToken() { s.tokenMu.Lock() s.token = "" @@ -542,12 +598,27 @@ func (s *driveStorage) Delete(ctx context.Context, name string) error { if err != nil { return err } + sem := make(chan struct{}, 8) + var wg sync.WaitGroup + var mu sync.Mutex + var first error for childName, file := range children { - if err := s.deleteID(ctx, childName, file.id); err != nil { - return err - } + wg.Add(1) + sem <- struct{}{} + go func(name, id string) { + defer wg.Done() + defer func() { <-sem }() + if err := s.deleteID(ctx, name, id); err != nil { + mu.Lock() + if first == nil { + first = err + } + mu.Unlock() + } + }(childName, file.id) } - return nil + wg.Wait() + return first } func (s *driveStorage) List(ctx context.Context, prefix string) ([]Entry, error) { @@ -604,6 +675,9 @@ func (s *driveStorage) makeSlot(ctx context.Context, name string) (string, error // setSlot implements slotStore, a description and a body go in one request func (s *driveStorage) setSlot(ctx context.Context, id, description string, body []byte) error { + if body == nil { + ctx = context.WithValue(ctx, attemptKey{}, driveSlotWrite) + } metadata, err := json.Marshal(map[string]string{"description": description}) if err != nil { return errors.New("failed to build the slot metadata").Base(err) @@ -631,6 +705,7 @@ func (s *driveStorage) setSlot(ctx context.Context, id, description string, body // readDescription implements slotStore func (s *driveStorage) readDescription(ctx context.Context, id string) (string, error) { + ctx = context.WithValue(ctx, attemptKey{}, driveSlotRead) status, payload, err := s.do(ctx, http.MethodGet, s.filesURL+"/"+url.PathEscape(id)+"?fields=description&supportsAllDrives=true", "", nil) switch { diff --git a/transport/internet/xdrive/drive_live_test.go b/transport/internet/xdrive/drive_live_test.go index 7808f8093..c629b6b4c 100644 --- a/transport/internet/xdrive/drive_live_test.go +++ b/transport/internet/xdrive/drive_live_test.go @@ -7,8 +7,13 @@ import ( "encoding/json" "fmt" "io" + "net/http" + "net/http/httptrace" + "net/url" "os" + "sort" "strconv" + "strings" "sync" "testing" "time" @@ -343,3 +348,130 @@ func TestLiveDriveListLag(t *testing.T) { } t.Logf("worst listing lag: %v", worst.Round(time.Millisecond)) } + +func TestLiveDriveLatencyFloor(t *testing.T) { + config := liveDriveConfig(t) + root, err := newDriveStorage(&internet.MemoryStreamConfig{}, config) + if err != nil { + t.Fatal(err) + } + defer root.Close() + store := root.session().(*driveStorage) + defer store.Close() + + ctx := context.Background() + const probe = "latency-floor/probe" + id, err := store.makeSlot(ctx, probe) + if err != nil { + t.Fatal(err) + } + defer store.deleteID(ctx, flatten(probe), id) + + small := newSlotFrame(1, slotInline, randomPayload(700)).encode() + large := newSlotFrame(1, slotInline, randomPayload(slotInlineLimit)).encode() + file := store.filesURL + "/" + url.PathEscape(id) + + type sample struct { + total, ttfb time.Duration + reused bool + } + measure := func(label string, op func(context.Context) error) []sample { + var out []sample + for i := 0; i < 28; i++ { + var wrote, first time.Time + var reused bool + trace := &httptrace.ClientTrace{ + GotConn: func(info httptrace.GotConnInfo) { reused = info.Reused }, + WroteRequest: func(httptrace.WroteRequestInfo) { wrote = time.Now() }, + GotFirstResponseByte: func() { first = time.Now() }, + } + start := time.Now() + err := op(httptrace.WithClientTrace(ctx, trace)) + total := time.Since(start) + if err != nil { + t.Logf("latency-floor: %s sample failed: %v", label, err) + } else if i >= 3 { + var ttfb time.Duration + if !wrote.IsZero() && !first.IsZero() { + ttfb = first.Sub(wrote) + } + out = append(out, sample{total, ttfb, reused}) + } + time.Sleep(250 * time.Millisecond) + } + return out + } + report := func(label string, s []sample) time.Duration { + if len(s) == 0 { + t.Logf("latency-floor: %-30s no samples", label) + return 0 + } + totals := make([]time.Duration, len(s)) + ttfbs := make([]time.Duration, len(s)) + reused := 0 + for i, x := range s { + totals[i], ttfbs[i] = x.total, x.ttfb + if x.reused { + reused++ + } + } + sort.Slice(totals, func(a, b int) bool { return totals[a] < totals[b] }) + sort.Slice(ttfbs, func(a, b int) bool { return ttfbs[a] < ttfbs[b] }) + q := func(v []time.Duration, p float64) time.Duration { return v[min(int(p*float64(len(v))), len(v)-1)] } + t.Logf("latency-floor: %-30s n=%d min=%v p50=%v p95=%v ttfb_p50=%v reused=%d/%d", + label, len(s), totals[0], q(totals, 0.5), q(totals, 0.95), q(ttfbs, 0.5), reused, len(s)) + return q(ttfbs, 0.5) + } + patch := func(target string, doc map[string]any) func(context.Context) error { + body, _ := json.Marshal(doc) + return func(ctx context.Context) error { + status, reply, err := store.do(ctx, http.MethodPatch, target, "application/json; charset=UTF-8", body) + if err == nil && status != http.StatusOK { + err = fmt.Errorf("status %d: %s", status, reply) + } + return err + } + } + get := func(target string) func(context.Context) error { + return func(ctx context.Context) error { + status, reply, err := store.do(ctx, http.MethodGet, target, "", nil) + if err == nil && status != http.StatusOK { + err = fmt.Errorf("status %d: %s", status, reply) + } + return err + } + } + + proto := "" + netFloor := report("net floor (401, no auth)", measure("net", func(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://www.googleapis.com/drive/v3/about?fields=user", nil) + if err != nil { + return err + } + resp, err := store.client.Do(req) + if err != nil { + return err + } + proto = resp.Proto + io.Copy(io.Discard, resp.Body) + return resp.Body.Close() + })) + t.Logf("latency-floor: negotiated protocol %s", proto) + readDesc := report("GET description (prod)", measure("get", func(ctx context.Context) error { + _, err := store.readDescription(ctx, id) + return err + })) + report("GET fields=id", measure("getid", get(file+"?fields=id&supportsAllDrives=true"))) + report("GET description, no allDrives", measure("getnoall", get(file+"?fields=description"))) + write1k := report("PATCH inline 1 KB (prod)", measure("p1k", func(ctx context.Context) error { + return store.setSlot(ctx, id, small, nil) + })) + write14k := report("PATCH inline 14 KB (prod)", measure("p14k", func(ctx context.Context) error { + return store.setSlot(ctx, id, large, nil) + })) + report("PATCH 1 KB, no allDrives", measure("pnoall", patch(file+"?fields=id", map[string]any{"description": small}))) + report("PATCH name 200 B", measure("pname", patch(file+"?fields=id&supportsAllDrives=true", map[string]any{"name": strings.Repeat("n", 200)}))) + report("PATCH appProperties 100 B", measure("pprop", patch(file+"?fields=id&supportsAllDrives=true", map[string]any{"appProperties": map[string]string{"k": strings.Repeat("a", 100)}}))) + t.Logf("latency-floor: derived: write backend ~ %v, read backend ~ %v, 14 KB payload cost ~ %v", + write1k-netFloor, readDesc-netFloor, write14k-write1k) +} diff --git a/transport/internet/xdrive/params.go b/transport/internet/xdrive/params.go index 01f80a31a..b667b4042 100644 --- a/transport/internet/xdrive/params.go +++ b/transport/internet/xdrive/params.go @@ -13,7 +13,7 @@ const ( defaultConcurrency = 8 maxSegmentBytes = 16 * 1024 * 1024 - maxConcurrency = 64 + maxConcurrency = 256 ) type params struct { diff --git a/transport/internet/xdrive/slots.go b/transport/internet/xdrive/slots.go index 8417acb99..a5b14c16c 100644 --- a/transport/internet/xdrive/slots.go +++ b/transport/internet/xdrive/slots.go @@ -10,6 +10,7 @@ import ( "io" "strconv" "sync" + "sync/atomic" "time" "github.com/xtls/xray-core/common/errors" @@ -27,7 +28,6 @@ const ( // slotInlineLimit is how large a segment is at most to go in the description // of its slot, which takes one request less than a body slotInlineLimit = 14 << 10 - slotDepthLimit = 8 // inlineRampSegments segments of one size in a row double it, from // slotInlineLimit up to segmentBytes inlineRampSegments = 8 @@ -41,19 +41,26 @@ const ( var ( // the head is asked for between slotHuntFloor and slotHuntCeiling, a segment // behind it at about the pace that segments come - slotHuntFloor = 50 * time.Millisecond - slotHuntCeiling = 200 * time.Millisecond + slotHuntFloor = 10 * time.Millisecond + slotHuntCeiling = 25 * time.Millisecond slotTailMin = 150 * time.Millisecond slotTailMax = 800 * time.Millisecond - slotFetchAhead = 6 + slotFetchAhead = 12 - slotAckRetry = 100 * time.Millisecond - slotWatermarkTTL = 250 * time.Millisecond + slotAckRetry = 30 * time.Millisecond + slotWatermarkTTL = 100 * time.Millisecond slotStallTimeout = 60 * time.Second slotCloseGrace = 10 * time.Second slotGoneTimeout = 60 * time.Second + slotBodyHedge = 2 * time.Second + + slotReadAhead int64 = 3 + slotReadAheadPoll = 40 * time.Millisecond + slotReadAheadBurst = 500 * time.Millisecond ) +var slotWriteInflight = 12 + // slotStore is a storage of objects that are written and read again by id, // each with a description and a body type slotStore interface { @@ -143,7 +150,7 @@ func decodeRing(b []byte) (ring, bool) { // makeRing makes the slots and watermarks of a session, all at once func makeRing(ctx context.Context, store slotStore, session string, p params) (ring, error) { - depth := min(max(p.concurrency, 2), slotDepthLimit) + depth := max(p.concurrency, 2) r := ring{ Up: slotTable{Slots: make([]string, depth)}, Down: slotTable{Slots: make([]string, depth)}, @@ -177,7 +184,7 @@ func makeRing(ctx context.Context, store slotStore, session string, p params) (r } func readWatermark(ctx context.Context, store slotStore, id string) (int64, error) { - d, err := store.readDescription(ctx, id) + d, err := store.readDescription(context.WithValue(ctx, laneKey{}, true), id) if err != nil || d == "" { return -1, err } @@ -216,7 +223,7 @@ func newSlotWriter(ctx context.Context, store slotStore, table slotTable, p para store: store, table: table, params: p, - slots: make(chan struct{}, len(table.Slots)), + slots: make(chan struct{}, min(len(table.Slots), slotWriteInflight)), size: min(slotInlineLimit, p.segmentBytes), acked: -1, progress: time.Now(), @@ -271,6 +278,10 @@ func (w *slotWriter) readyToFlush() bool { grew := len(w.buf) > w.lastSize w.lastSize = len(w.buf) + if len(w.slots) > 0 && w.held < maxCoalescedTicks && len(w.buf) < w.size/2 { + w.held++ + return false + } if grew && w.held < maxCoalescedTicks { w.held++ return false @@ -296,11 +307,19 @@ func (w *slotWriter) flushLocked() error { return w.err } + if w.size < w.segmentBytes && len(w.buf) >= w.size && len(w.slots)*4 > cap(w.slots)*3 { + w.size = w.segmentBytes + w.full = 0 + } + n := min(len(w.buf), w.size) if len(w.buf) > 4*slotInlineLimit { // a burst goes at the full size at once n = min(len(w.buf), w.segmentBytes) } + if n > slotInlineLimit && len(w.buf) <= 2*cap(w.slots)*slotInlineLimit { + n = slotInlineLimit + } if n == w.size && w.size < w.segmentBytes { if w.full++; w.full == inlineRampSegments { w.full = 0 @@ -332,15 +351,21 @@ func (w *slotWriter) upload(seq int64, chunk []byte) { if len(chunk) <= slotInlineLimit { mode = slotInline } + began := time.Now() if err := w.await(seq); err != nil { w.fail(errors.New("failed to store segment ", seq).Base(err)) return } + awaited := time.Since(began) + putAt := time.Now() if err := w.write(seq, mode, chunk); err != nil { w.fail(errors.New("failed to store segment ", seq).Base(err)) // the slot is free, the reader learns there what happened w.write(seq, slotErr, nil) + return } + errors.LogDebug(w.ctx, "xdrive write ", slotTag(w.table.Watermark), " seq=", seq, " bytes=", len(chunk), + " mode=", mode, " await=", awaited, " put=", time.Since(putAt)) } func (w *slotWriter) write(seq int64, mode byte, data []byte) error { @@ -393,7 +418,7 @@ func (w *slotWriter) await(seq int64) error { w.ackMu.Lock() still := time.Since(w.progress) w.ackMu.Unlock() - if time.Since(start) > slotStallTimeout || closed && still > slotCloseGrace { + if min(time.Since(start), still) > slotStallTimeout || closed && still > slotCloseGrace { return errors.New("the reader took nothing from slot ", w.slotID(seq), " for ", still).Base(err) } @@ -461,6 +486,7 @@ type slotReader struct { ch chan []byte wake chan struct{} acks chan struct{} + grow chan struct{} mu sync.Mutex head int64 @@ -472,7 +498,11 @@ type slotReader struct { gap time.Duration activeAt time.Time active chan struct{} + frontier int64 err error + reads atomic.Int64 + tailMin time.Duration + tailMax time.Duration } func newSlotReader(ctx context.Context, store slotStore, table slotTable, p params) *slotReader { @@ -484,9 +514,12 @@ func newSlotReader(ctx context.Context, store slotStore, table slotTable, p para ch: make(chan []byte, p.concurrency), wake: make(chan struct{}, 1), acks: make(chan struct{}, 1), + grow: make(chan struct{}, 1), hitAt: time.Now(), activeAt: time.Now(), active: make(chan struct{}), + tailMin: slotTailMin, + tailMax: slotTailMax, } go r.run() go r.ackLoop() @@ -516,9 +549,20 @@ type slotFetch struct { ready chan struct{} // promoted is closed when the segment becomes the head promoted chan struct{} + near chan struct{} body []byte end bool err error + born time.Time + readyAt time.Time +} + +func (f *slotFetch) enter() { + select { + case <-f.near: + default: + close(f.near) + } } func (r *slotReader) run() { @@ -529,17 +573,20 @@ func (r *slotReader) run() { fetches := make(map[int64]*slotFetch) for { r.mu.Lock() - head, started := r.head, r.started + head, started, frontier := r.head, r.started, r.frontier r.mu.Unlock() window := 1 if started { - window = min(slotFetchAhead, len(r.table.Slots)) + window = min(max(int(frontier-head)+slotFetchAhead, slotFetchAhead), len(r.table.Slots)) } for i := range window { seq := head + int64(i) if fetches[seq] == nil { - fetches[seq] = &slotFetch{ready: make(chan struct{}), promoted: make(chan struct{})} + fetches[seq] = &slotFetch{ready: make(chan struct{}), promoted: make(chan struct{}), near: make(chan struct{}), born: time.Now()} + if int64(i) <= slotReadAhead { + fetches[seq].enter() + } go r.fetch(ctx, seq, i, fetches[seq]) } } @@ -547,6 +594,8 @@ func (r *slotReader) run() { f := fetches[head] select { case <-f.ready: + case <-r.grow: + continue case <-ctx.Done(): return } @@ -564,8 +613,14 @@ func (r *slotReader) run() { r.head++ r.started = true r.mu.Unlock() + errors.LogDebug(r.ctx, "xdrive read ", slotTag(r.table.Watermark), " seq=", head, " fetch=", + f.readyAt.Sub(f.born), " blocked=", time.Since(f.readyAt), " bytes=", len(f.body), + " reads=", r.reads.Load()) if next := fetches[head+1]; next != nil { close(next.promoted) + if nf := fetches[head+1+slotReadAhead]; nf != nil { + nf.enter() + } } select { case r.acks <- struct{}{}: @@ -586,11 +641,35 @@ func (r *slotReader) isHead(seq int64) bool { return r.head == seq } +func (r *slotReader) bursting() bool { + r.mu.Lock() + defer r.mu.Unlock() + return time.Since(r.hitAt) <= slotReadAheadBurst +} + +func (r *slotReader) reach(seq int64) { + r.mu.Lock() + grown := seq > r.frontier + if grown { + r.frontier = seq + } + r.mu.Unlock() + if grown { + select { + case r.grow <- struct{}{}: + default: + } + } +} + // fetch asks for the slot of seq until it holds seq: the head often, a segment -// behind it at the pace that segments come, as it would wait for the head anyway +// behind it soon after it was asked for, then at the pace that segments come func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFetch) { - defer close(f.ready) - if offset > 0 && !r.hold(ctx, time.Duration(offset)*r.pace(), f.promoted, nil) { + defer func() { + f.readyAt = time.Now() + close(f.ready) + }() + if offset > 0 && !r.hold(ctx, r.tailMin, f.promoted, f.near) { return } @@ -598,7 +677,17 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe backoff := slotHuntFloor var goneSince, warnedAt time.Time for { - description, err := r.store.readDescription(ctx, id) + var description string + var err error + r.mu.Lock() + hot := time.Since(r.activeAt) <= r.eagerWindow + r.mu.Unlock() + if hot && r.isHead(seq) { + description, err = r.readHead(ctx, id, seq) + } else { + r.reads.Add(1) + description, err = r.store.readDescription(ctx, id) + } frame, ok := decodeSlot(description) switch { case err == nil && ok && frame.seq == seq: @@ -606,6 +695,7 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe r.mu.Lock() r.touch() r.mu.Unlock() + r.reach(seq) switch frame.mode { case slotEnd: f.end = true @@ -618,7 +708,7 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe r.hit() return case slotMedia: - if body, err := r.store.readBody(ctx, id); err == nil && frame.holds(body) { + if body, err := r.readBodyHedged(ctx, id); err == nil && frame.holds(body) { f.body = body r.hit() return @@ -654,7 +744,15 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe // only the head is asked for until something comes wait, a, b, backoff = 0, f.promoted, active, slotHuntFloor default: - wait, a, backoff = min(max(r.pace(), slotTailMin), slotTailMax), f.promoted, slotHuntFloor + wait, a, b, backoff = min(max(r.pace(), r.tailMin), r.tailMax), f.promoted, f.near, slotHuntFloor + select { + case <-f.near: + b = nil + if r.bursting() { + wait = slotReadAheadPoll + } + default: + } } if !r.hold(ctx, wait, a, b) { return @@ -662,6 +760,106 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe } } +func (r *slotReader) readHead(ctx context.Context, id string, seq int64) (string, error) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + ctx = context.WithValue(ctx, laneKey{}, true) + + type result struct { + description string + err error + } + results := make(chan result, 4) + outstanding := 0 + launch := func() { + outstanding++ + go func() { + r.reads.Add(1) + description, err := r.store.readDescription(ctx, id) + results <- result{description, err} + }() + } + launch() + ticker := time.NewTicker(slotHuntCeiling) + defer ticker.Stop() + var last result + seen := false + for { + select { + case res := <-results: + outstanding-- + if res.err != nil { + return res.description, res.err + } + if frame, ok := decodeSlot(res.description); ok && frame.seq == seq { + return res.description, nil + } + last, seen = res, true + case <-ticker.C: + r.mu.Lock() + idle := time.Since(r.activeAt) > r.eagerWindow + r.mu.Unlock() + switch { + case idle && seen: + return last.description, nil + case !idle && outstanding < 3: + launch() + } + case <-r.wake: + if outstanding < 3 { + launch() + } + case <-ctx.Done(): + return "", ctx.Err() + } + } +} + +func (r *slotReader) readBodyHedged(ctx context.Context, id string) ([]byte, error) { + if slotBodyHedge <= 0 { + return r.store.readBody(ctx, id) + } + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + type result struct { + body []byte + err error + } + results := make(chan result, 2) + launch := func() { + go func() { + body, err := r.store.readBody(ctx, id) + results <- result{body, err} + }() + } + launch() + outstanding := 1 + timer := time.NewTimer(slotBodyHedge) + defer timer.Stop() + var last result + for { + select { + case res := <-results: + outstanding-- + if res.err == nil { + return res.body, nil + } + last = res + if outstanding == 0 { + return nil, last.err + } + case <-timer.C: + if outstanding == 1 { + launch() + outstanding++ + } + case <-ctx.Done(): + return nil, ctx.Err() + } + } +} + // pace is about how long segments take to come one after the other func (r *slotReader) pace() time.Duration { r.mu.Lock() @@ -726,7 +924,7 @@ func (r *slotReader) ackLoop() { if want <= acked { break } - if err := r.store.setSlot(r.ctx, r.table.Watermark, strconv.FormatInt(want, 10), nil); err == nil { + if err := r.store.setSlot(context.WithValue(r.ctx, laneKey{}, true), r.table.Watermark, strconv.FormatInt(want, 10), nil); err == nil { acked, wait = want, slotAckRetry continue } @@ -756,3 +954,10 @@ func (r *slotReader) Err() error { } return io.EOF } + +func slotTag(id string) string { + if len(id) > 6 { + return id[len(id)-6:] + } + return id +} diff --git a/transport/internet/xdrive/slots_test.go b/transport/internet/xdrive/slots_test.go index a9a835db1..594f85543 100644 --- a/transport/internet/xdrive/slots_test.go +++ b/transport/internet/xdrive/slots_test.go @@ -10,6 +10,7 @@ import ( "os" "strconv" "sync" + "sync/atomic" "testing" "time" ) @@ -304,6 +305,373 @@ func TestSlotStaleBody(t *testing.T) { client, server := slotPair(t, &staleSlots{memSlots: newMemSlots(), served: make(map[string]bool)}, 2, 64<<10) payload := randomPayload(200 << 10) errs := make(chan error, 1) - writeAll(client, payload, true, errs) + go func() { + _, err := client.Write(payload) + if err == nil { + err = client.Close() + } + errs <- err + }() readAll(t, server, payload) } + +func TestSlotBurstSplitsInline(t *testing.T) { + store := newMemSlots() + client, server := slotPair(t, store, 8, 512<<10) + payload := randomPayload(90 << 10) + errs := make(chan error, 1) + writeAll(client, payload, true, errs) + readAll(t, server, payload) + store.mu.Lock() + bodies := len(store.bodies) + store.mu.Unlock() + if bodies != 0 { + t.Fatalf("a burst of 90 KiB went in %d bodies, want none", bodies) + } +} + +type stuckBodySlots struct { + *memSlots + calls atomic.Int32 +} + +func (s *stuckBodySlots) readBody(ctx context.Context, id string) ([]byte, error) { + if s.calls.Add(1) == 1 { + <-ctx.Done() + return nil, ctx.Err() + } + return s.memSlots.readBody(ctx, id) +} + +func TestSlotBodyHedge(t *testing.T) { + old := slotBodyHedge + slotBodyHedge = 50 * time.Millisecond + defer func() { slotBodyHedge = old }() + + store := &stuckBodySlots{memSlots: newMemSlots()} + client, server := slotPair(t, store, 2, 64<<10) + payload := randomPayload(200 << 10) + errs := make(chan error, 1) + go func() { + _, err := client.Write(payload) + if err == nil { + err = client.Close() + } + errs <- err + }() + readAll(t, server, payload) + if store.calls.Load() < 2 { + t.Fatalf("the stuck body read was not hedged, %d reads", store.calls.Load()) + } +} + +func TestTakeSpareSkipsStale(t *testing.T) { + cfg := &Config{} + p := params{sessionTTL: 5 * time.Minute} + st1, err := newLocalStorage(t.TempDir()) + if err != nil { + t.Fatal(err) + } + st2, err := newLocalStorage(t.TempDir()) + if err != nil { + t.Fatal(err) + } + box := spareBoxFor(cfg) + box.ch <- spareRing{storage: st1, session: "old", at: time.Now().Add(-time.Hour)} + box.ch <- spareRing{storage: st2, session: "new", at: time.Now()} + got, ok := takeSpare(cfg, p) + if !ok || got.session != "new" { + t.Fatalf("takeSpare gave %q, %v, want the fresh one", got.session, ok) + } + if _, ok := takeSpare(cfg, p); ok { + t.Fatal("takeSpare returned a second spare") + } +} + +func TestDropSpareDeletesSession(t *testing.T) { + st, err := newLocalStorage(t.TempDir()) + if err != nil { + t.Fatal(err) + } + name := sessionPrefix("gone") + "/" + uplinkDir + "/s0" + if err := st.Put(context.Background(), name, []byte("x")); err != nil { + t.Fatal(err) + } + dropSpare(spareRing{storage: st, session: "gone"}) + if got, _ := st.List(context.Background(), sessionPrefix("gone")); len(got) != 0 { + t.Fatalf("%d entries left of a dropped spare", len(got)) + } +} + +type stuckHeadSlots struct { + *memSlots + calls atomic.Int32 +} + +func (s *stuckHeadSlots) readDescription(ctx context.Context, id string) (string, error) { + if s.calls.Add(1) == 1 { + <-ctx.Done() + return "", ctx.Err() + } + return s.memSlots.readDescription(ctx, id) +} + +func TestSlotHeadReadOverlaps(t *testing.T) { + client, server := slotPair(t, &stuckHeadSlots{memSlots: newMemSlots()}, 4, 64<<10) + for _, c := range [][2]*Conn{{client, server}, {server, client}} { + c[0].Write([]byte("ping")) + if _, err := io.ReadFull(c[1], make([]byte, 4)); err != nil { + t.Fatal(err) + } + } +} + +func TestSlotBurstReadsTogether(t *testing.T) { + client, server := slotPair(t, newMemSlots(), 16, 512<<10) + payload := randomPayload(8 * slotInlineLimit) + go client.Write(payload) + start := time.Now() + got := make([]byte, len(payload)) + if _, err := io.ReadFull(server, got); err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("the burst came back changed") + } + if took := time.Since(start); took > time.Second { + t.Fatalf("a burst of 8 segments took %v to read", took) + } +} + +type lagSlots struct{ *memSlots } + +func (s lagSlots) readDescription(ctx context.Context, id string) (string, error) { + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-time.After(200 * time.Millisecond): + } + return s.memSlots.readDescription(ctx, id) +} + +func (s lagSlots) setSlot(ctx context.Context, id, description string, body []byte) error { + if _, ok := decodeSlot(description); ok { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(600 * time.Millisecond): + } + } + return s.memSlots.setSlot(ctx, id, description, body) +} + +func TestSlotReadAheadBurst(t *testing.T) { + oldMin, oldMax := slotTailMin, slotTailMax + slotTailMin, slotTailMax = 150*time.Millisecond, 800*time.Millisecond + t.Cleanup(func() { slotTailMin, slotTailMax = oldMin, oldMax }) + + client, server := slotPair(t, lagSlots{newMemSlots()}, 16, 512<<10) + payload := randomPayload(8 * slotInlineLimit) + go client.Write(payload) + start := time.Now() + got := make([]byte, len(payload)) + if _, err := io.ReadFull(server, got); err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("the burst came back changed") + } + if took := time.Since(start); took > 1700*time.Millisecond { + t.Fatalf("a burst of 8 segments took %v to read", took) + } + client.Close() + server.Close() +} + +func TestSlotLargeBurstInline(t *testing.T) { + store := newMemSlots() + client, server := slotPair(t, store, 16, 512<<10) + payload := randomPayload(10 * slotInlineLimit) + go client.Write(payload) + got := make([]byte, len(payload)) + if _, err := io.ReadFull(server, got); err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("the burst came back changed") + } + store.mu.Lock() + bodies := len(store.bodies) + store.mu.Unlock() + if bodies != 0 { + t.Fatalf("a burst of 140 KiB went in %d bodies, want none", bodies) + } +} + +func TestSlotDepthFollowsConcurrency(t *testing.T) { + for _, c := range []struct{ set, want int }{{100, 100}, {maxConcurrency, maxConcurrency}, {1 << 20, maxConcurrency}} { + p := paramsFromConfig(&Config{Concurrency: uint32(c.set)}) + r, err := makeRing(context.Background(), newMemSlots(), "session", p) + if err != nil { + t.Fatal(err) + } + if len(r.Up.Slots) != c.want || len(r.Down.Slots) != c.want { + t.Fatalf("concurrency %d made %d and %d slots, want %d", c.set, len(r.Up.Slots), len(r.Down.Slots), c.want) + } + } +} + +func TestSlotDeepRingRoundTrip(t *testing.T) { + client, server := slotPair(t, newMemSlots(), 64, 64<<10) + payload := randomPayload(2 << 20) + errs := make(chan error, 1) + writeAll(client, payload, true, errs) + readAll(t, server, payload) + if err := <-errs; err != nil { + t.Fatal(err) + } +} + +type pollSlots struct { + *memSlots + pmu sync.Mutex + polled map[string]bool +} + +func (s *pollSlots) readDescription(ctx context.Context, id string) (string, error) { + s.pmu.Lock() + s.polled[id] = true + s.pmu.Unlock() + return s.memSlots.readDescription(ctx, id) +} + +func TestSlotReaderReachesPastWindow(t *testing.T) { + store := &pollSlots{memSlots: newMemSlots(), polled: make(map[string]bool)} + p := paramsFromConfig(&Config{FlushIntervalMs: 1, PollIntervalMs: 1, MaxPollIntervalMs: 20, EagerWindowMs: 50, Concurrency: 64}) + ring, err := makeRing(context.Background(), store, "session", p) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + put := func(seq int64) { + slot := ring.Down.Slots[seq%int64(len(ring.Down.Slots))] + if err := store.setSlot(ctx, slot, newSlotFrame(seq, slotInline, []byte{byte(seq)}).encode(), nil); err != nil { + t.Fatal(err) + } + } + reader := newSlotReader(ctx, store, ring.Down, p) + next := func() []byte { + select { + case chunk := <-reader.chunks(): + return chunk + case <-time.After(10 * time.Second): + t.Fatal("no segment came") + return nil + } + } + put(0) + if got := next(); !bytes.Equal(got, []byte{0}) { + t.Fatalf("segment 0 came as %v", got) + } + for seq := int64(2); seq <= 40; seq++ { + put(seq) + } + last := ring.Down.Slots[40] + deadline := time.Now().Add(10 * time.Second) + for { + reader.Wake() + store.pmu.Lock() + polled := store.polled[last] + store.pmu.Unlock() + if polled { + break + } + if time.Now().After(deadline) { + t.Fatal("segment 40 was never asked for while segment 1 was missing") + } + time.Sleep(10 * time.Millisecond) + } + put(1) + for seq := int64(1); seq <= 40; seq++ { + if got := next(); !bytes.Equal(got, []byte{byte(seq)}) { + t.Fatalf("segment %d came as %v", seq, got) + } + } +} + +type countSlots struct { + *memSlots + cur atomic.Int32 + peak atomic.Int32 +} + +func (s *countSlots) setSlot(ctx context.Context, id, description string, body []byte) error { + if _, ok := decodeSlot(description); !ok { + return s.memSlots.setSlot(ctx, id, description, body) + } + n := s.cur.Add(1) + for { + p := s.peak.Load() + if n <= p || s.peak.CompareAndSwap(p, n) { + break + } + } + time.Sleep(20 * time.Millisecond) + s.cur.Add(-1) + return s.memSlots.setSlot(ctx, id, description, body) +} + +func TestSlotWriteInflightCap(t *testing.T) { + store := &countSlots{memSlots: newMemSlots()} + client, server := slotPair(t, store, 48, 512<<10) + payload := randomPayload(60 * slotInlineLimit) + errs := make(chan error, 1) + go func() { + for off := 0; off < len(payload); off += slotInlineLimit { + if _, err := client.Write(payload[off : off+slotInlineLimit]); err != nil { + errs <- err + return + } + } + errs <- nil + }() + got := make([]byte, len(payload)) + if _, err := io.ReadFull(server, got); err != nil { + t.Fatal(err) + } + if err := <-errs; err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("the payload came back changed") + } + if p := int(store.peak.Load()); p > slotWriteInflight { + t.Fatalf("%d segments were written at once, want at most %d", p, slotWriteInflight) + } +} + +func TestSlotBurstInlineWaves(t *testing.T) { + for _, c := range []struct { + pieces int + media bool + }{{20, false}, {30, true}} { + store := newMemSlots() + client, server := slotPair(t, store, 16, 512<<10) + payload := randomPayload(c.pieces * slotInlineLimit) + go client.Write(payload) + got := make([]byte, len(payload)) + if _, err := io.ReadFull(server, got); err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("the burst came back changed") + } + store.mu.Lock() + bodies := len(store.bodies) + store.mu.Unlock() + if (bodies > 0) != c.media { + t.Fatalf("a burst of %d pieces went in %d bodies, media wanted %v", c.pieces, bodies, c.media) + } + } +} diff --git a/transport/internet/xdrive/xdrive.go b/transport/internet/xdrive/xdrive.go index c326d506c..c02811aef 100644 --- a/transport/internet/xdrive/xdrive.go +++ b/transport/internet/xdrive/xdrive.go @@ -9,6 +9,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/xtls/xray-core/common" @@ -26,6 +27,11 @@ const ( downlinkDir = "s2c" ) +const ( + spareCount = 2 + spareIdleStop = 10 * time.Minute +) + func init() { common.Must(internet.RegisterProtocolConfigCreator(protocolName, func() interface{} { return new(Config) @@ -78,30 +84,152 @@ func streamConfig(streamSettings *internet.MemoryStreamConfig) (*Config, error) return config, nil } +type spareRing struct { + storage Storage + session string + ring ring + at time.Time +} + +type spareBox struct { + ch chan spareRing + busy atomic.Bool + last atomic.Int64 + stop atomic.Bool +} + +var spareBoxes sync.Map + +func spareBoxFor(config *Config) *spareBox { + if b, ok := spareBoxes.Load(config); ok { + return b.(*spareBox) + } + b, _ := spareBoxes.LoadOrStore(config, &spareBox{ch: make(chan spareRing, spareCount)}) + return b.(*spareBox) +} + +func dropSpare(s spareRing) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + s.storage.Delete(ctx, sessionPrefix(s.session)) + s.storage.Close() +} + +func takeSpare(config *Config, p params) (spareRing, bool) { + box := spareBoxFor(config) + for { + select { + case s := <-box.ch: + if time.Since(s.at) < p.sessionTTL*2/3 { + return s, true + } + go dropSpare(s) + default: + return spareRing{}, false + } + } +} + +func warmSpares(streamSettings *internet.MemoryStreamConfig, config *Config, p params) { + spareBoxFor(config).last.Store(time.Now().UnixNano()) + fillSpares(streamSettings, config, p) +} + +func fillSpares(streamSettings *internet.MemoryStreamConfig, config *Config, p params) { + box := spareBoxFor(config) + if !box.busy.CompareAndSwap(false, true) { + return + } + go func() { + defer box.busy.Store(false) + for n := len(box.ch); n > 0; n-- { + select { + case s := <-box.ch: + if time.Since(s.at) < p.sessionTTL/3 { + box.ch <- s + } else { + dropSpare(s) + } + default: + } + } + for len(box.ch) < cap(box.ch) { + for range 40 { + if box.stop.Load() { + return + } + time.Sleep(50 * time.Millisecond) + } + if box.stop.Load() { + return + } + storage, err := newStorage(streamSettings) + if err != nil { + return + } + storage = sessionStorage(storage) + slots, ok := storage.(slotStore) + if !ok { + storage.Close() + return + } + session, err := newSessionID() + if err != nil { + storage.Close() + return + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + r, err := makeRing(ctx, slots, session, p) + cancel() + if err != nil { + dropSpare(spareRing{storage: storage, session: session}) + return + } + select { + case box.ch <- spareRing{storage: storage, session: session, ring: r, at: time.Now()}: + default: + storage.Close() + return + } + } + time.AfterFunc(p.sessionTTL/4, func() { + if !box.stop.Load() && time.Since(time.Unix(0, box.last.Load())) < spareIdleStop { + fillSpares(streamSettings, config, p) + } + }) + }() +} + func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.MemoryStreamConfig) (stat.Connection, error) { config, err := streamConfig(streamSettings) if err != nil { return nil, err } - storage, err := newStorage(streamSettings) - if err != nil { - return nil, err - } - storage = sessionStorage(storage) - - session, err := newSessionID() - if err != nil { - storage.Close() - return nil, err - } - p := paramsFromConfig(config) + + spare, warm := takeSpare(config, p) + storage, session, r := spare.storage, spare.session, spare.ring + if !warm { + storage, err = newStorage(streamSettings) + if err != nil { + return nil, err + } + storage = sessionStorage(storage) + + session, err = newSessionID() + if err != nil { + storage.Close() + return nil, err + } + } + var announce []byte - var r ring slots, ok := storage.(slotStore) if ok { - if r, err = makeRing(ctx, slots, session, p); err == nil { + if warm { + announce, _ = json.Marshal(r) + } else if r, err = makeRing(ctx, slots, session, p); err == nil { announce, _ = json.Marshal(r) } else { ok = false @@ -115,6 +243,7 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me } if ok { + warmSpares(streamSettings, config, p) errors.LogInfo(ctx, "opened session ", session, " in ", len(r.Up.Slots), " slots") return newSlotConn(context.Background(), slots, r.Up, r.Down, p, func() { storage.Close() @@ -320,12 +449,13 @@ func (l *Listener) collect() error { } l.mu.Unlock() + var first error for _, session := range expired { - if err := l.storage.Delete(l.ctx, sessionPrefix(session)); err != nil { - return err + if err := l.storage.Delete(l.ctx, sessionPrefix(session)); err != nil && first == nil { + first = err } } - return nil + return first } func (l *Listener) Addr() net.Addr { diff --git a/transport/internet/xdrive/xdrive_test.go b/transport/internet/xdrive/xdrive_test.go index aed8e7873..947e43a19 100644 --- a/transport/internet/xdrive/xdrive_test.go +++ b/transport/internet/xdrive/xdrive_test.go @@ -69,6 +69,13 @@ func pairWith(t *testing.T, streamSettings *internet.MemoryStreamConfig) (client } return client, server, func() { + if config, err := streamConfig(streamSettings); err == nil { + box := spareBoxFor(config) + box.stop.Store(true) + for box.busy.Load() { + time.Sleep(10 * time.Millisecond) + } + } client.Close() server.Close() listener.Close()