package work
import (
"context"
"testing"
"time"
)
// A queued caller is already visible. The whole reason to list a lane's
// contents is that it is saturated, which means most of what a reader
// wants to see has not been admitted yet — a list that only showed the
// running units would be empty exactly when it mattered.
func TestAQueuedUnitIsVisibleBeforeItRuns(t *testing.T) {
l := NewLane("ingest", 1, 4)
release := make(chan struct{})
go l.DoNamed(context.Background(), "ingest", "pkgsrc", func() error {
<-release
return nil
})
// Wait for the first to be running, then queue a second.
waitFor(t, func() bool {
it := l.Items()
return len(it) == 1 && it[0].Running
}, "the first unit to start running")
go l.DoNamed(context.Background(), "first ingest", "tcl", func() error { return nil })
waitFor(t, func() bool { return len(l.Items()) == 2 }, "the queued unit to appear")
it := l.Items()
if !it[0].Running || it[0].Subject != "pkgsrc" {
t.Errorf("FAIL: first item = %+v, want pkgsrc running", it[0])
}
if it[1].Running || it[1].Subject != "tcl" || it[1].Verb != "first ingest" {
t.Errorf("FAIL: second item = %+v, want tcl queued", it[1])
}
close(release)
waitFor(t, func() bool { return len(l.Items()) == 0 }, "the lane to empty")
}
// Running before queued, oldest first within each. An operator scanning
// a saturated lane is looking for the unit that has been there longest;
// a list in admission order buries it.
func TestItemsPutRunningFirstThenOldest(t *testing.T) {
l := NewLane("bulk", 0, 0) // unbounded: everything runs
l.itemMu.Lock()
now := time.Now()
l.items = map[int64]*item{
1: {verb: "a", subject: "young-run", since: now.Add(-1 * time.Second), running: true},
2: {verb: "a", subject: "old-run", since: now.Add(-9 * time.Second), running: true},
3: {verb: "a", subject: "old-wait", since: now.Add(-5 * time.Second)},
}
l.itemMu.Unlock()
var order []string
for _, it := range l.Items() {
order = append(order, it.Subject)
}
want := []string{"old-run", "young-run", "old-wait"}
for i := range want {
if i >= len(order) || order[i] != want[i] {
t.Fatalf("FAIL: order = %q, want %q", order, want)
}
}
}
// A unit the lane shed never appears. Being told busy is not being held,
// and a list that carried refusals would grow without bound.
func TestAShedUnitIsNotHeld(t *testing.T) {
l := NewLane("ingest", 1, 0)
release := make(chan struct{})
go l.DoNamed(context.Background(), "ingest", "held", func() error {
<-release
return nil
})
waitFor(t, func() bool { return len(l.Items()) == 1 }, "the first unit to start")
if err := l.DoNamed(context.Background(), "ingest", "shed", func() error { return nil }); err == nil {
t.Fatal("FAIL: a full lane admitted a second caller")
}
for _, it := range l.Items() {
if it.Subject == "shed" {
t.Errorf("FAIL: a shed unit is listed as held: %+v", it)
}
}
close(release)
}
func waitFor(t *testing.T, ok func() bool, what string) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if ok() {
return
}
time.Sleep(time.Millisecond)
}
t.Fatalf("FAIL: timed out waiting for %s", what)
}