hesoyam/Xray-core
1
0
Fork 0
mirror of https://github.com/XTLS/Xray-core.git synced 2026-11-06 10:12:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

XDRIVE transport: A bunch of optimizations for Google Drive

Browse Source
This commit is contained in:
RPRX
2026-11-06 04:08:29 +00:00
committed by GitHub
parent 36756ad672
commit 47a10c0012
9 changed files with 1092 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
transport/internet/xdrive/client.go
+23 -4
View File
@@ -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()
transport/internet/xdrive/client_test.go
+100
View File
@@ -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)
}
}
transport/internet/xdrive/drive.go
+88 -13
View File
@@ -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 {
transport/internet/xdrive/drive_live_test.go
+132
View File
@@ -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)
}
transport/internet/xdrive/params.go
+1 -1
View File
@@ -13,7 +13,7 @@ const (
defaultConcurrency = 8
maxSegmentBytes = 16 * 1024 * 1024
maxConcurrency = 64
maxConcurrency = 256
)
type params struct {
transport/internet/xdrive/slots.go
+225 -20
View File
@@ -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
}
transport/internet/xdrive/slots_test.go
+369 -1
View File
@@ -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)
}
}
}
transport/internet/xdrive/xdrive.go
+147 -17
View File
@@ -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 {
transport/internet/xdrive/xdrive_test.go
+7
View File
@@ -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()