Skip to content
quarry Artifact [a3c86a52a8]
Login

Artifact [a3c86a52a8]

Artifact a3c86a52a80c9f82ef6dc84b89d289f0b2d65d0e5dddf71d766d6bc21ad722d6:


package work

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// The limit is the whole point: without it the file page forked a
// subprocess per request and throughput went backwards under load.
func TestLaneNeverExceedsItsLimit(t *testing.T) {
	l := NewLane("test", 3, 100)
	var now, peak atomic.Int64
	var wg sync.WaitGroup
	for i := 0; i < 40; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			l.Do(context.Background(), func() error {
				n := now.Add(1)
				for {
					p := peak.Load()
					if n <= p || peak.CompareAndSwap(p, n) {
						break
					}
				}
				time.Sleep(2 * time.Millisecond)
				now.Add(-1)
				return nil
			})
		}()
	}
	wg.Wait()
	if got := peak.Load(); got > 3 {
		t.Errorf("ran %d concurrently, limit is 3", got)
	}
	if got := peak.Load(); got < 2 {
		t.Errorf("peak concurrency %d — the lane serialized instead of admitting its limit", got)
	}
}

// Past the cap a lane must REFUSE, not accept work it cannot serve.
// An accepted request that times out consumed the capacity that would
// have served someone else.
func TestLaneShedsPastItsQueue(t *testing.T) {
	l := NewLane("test", 1, 0) // one at a time, no waiting room

	release := make(chan struct{})
	started := make(chan struct{})
	go l.Do(context.Background(), func() error {
		close(started)
		<-release
		return nil
	})
	<-started

	err := l.Do(context.Background(), func() error {
		t.Error("ran a function the lane should have refused")
		return nil
	})
	if !errors.Is(err, ErrBusy) {
		t.Fatalf("err = %v, want ErrBusy", err)
	}
	if s := l.Stats(); s.Shed != 1 {
		t.Errorf("Shed = %d, want 1 — the refusal must be counted, it is the scaling signal", s.Shed)
	}
	close(release)
}

// A caller who gives up while queued must not run later against a
// context nobody is waiting on.
func TestLaneHonoursCancellation(t *testing.T) {
	l := NewLane("test", 1, 10)
	release := make(chan struct{})
	started := make(chan struct{})
	go l.Do(context.Background(), func() error {
		close(started)
		<-release
		return nil
	})
	<-started

	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	var ran bool
	err := l.Do(ctx, func() error { ran = true; return nil })
	if !errors.Is(err, context.Canceled) {
		t.Errorf("err = %v, want context.Canceled", err)
	}
	if ran {
		t.Error("ran the function for a caller who had already given up")
	}
	close(release)
}

// The error from fn must reach the caller unchanged; a lane is a gate,
// not a wrapper.
func TestLanePassesErrorsThrough(t *testing.T) {
	l := NewLane("test", 2, 2)
	want := errors.New("boom")
	if got := l.Do(context.Background(), func() error { return want }); !errors.Is(got, want) {
		t.Errorf("err = %v, want %v", got, want)
	}
}

// A nil lane runs the work — so a caller with no lane configured is not
// a special case at every call site.
func TestNilLaneRuns(t *testing.T) {
	var l *Lane
	ran := false
	if err := l.Do(context.Background(), func() error { ran = true; return nil }); err != nil {
		t.Fatal(err)
	}
	if !ran {
		t.Error("nil lane did not run the work")
	}
}