package work
import (
"context"
"testing"
)
// A unit that turns out to need something other than its lane — an
// ingest round that becomes a rebuild — leaves it for good, and
// ordinary units run in the slot it gave back.
func TestALeavingUnitFreesItsSlotForGood(t *testing.T) {
l := NewLane("test", 1, 0)
left := make(chan struct{})
proceed := make(chan struct{})
done := make(chan error, 1)
go func() {
done <- l.DoNamedLeavable(context.Background(), "ingest", "a", func(leave func()) error {
leave()
leave() // harmless twice
close(left)
<-proceed
return nil
})
}()
<-left
if st := l.Stats(); st.InFlight != 0 || st.Waiting != 0 {
t.Errorf("FAIL: a unit that left still counts against the lane: %+v", st)
}
if items := l.Items(); len(items) != 0 {
t.Errorf("FAIL: a unit that left is still listed: %+v", items)
}
if err := l.Do(context.Background(), func() error { return nil }); err != nil {
t.Fatalf("FAIL: another unit was refused after the first left: %v", err)
}
t.Log("ok: another unit runs in the slot a leaving unit gave back")
close(proceed)
if err := <-done; err != nil {
t.Fatal(err)
}
if st := l.Stats(); st.InFlight != 0 || st.Waiting != 0 || st.Shed != 0 {
t.Errorf("FAIL: the lane is unbalanced after a leave: %+v", st)
} else {
t.Log("ok: the lane is balanced after a leave")
}
}
// A unit whose context ends while it queues never ran, and leaves the
// lane balanced.
func TestACancelledLeavableUnitLeavesTheLaneBalanced(t *testing.T) {
l := NewLane("test", 1, 1)
hold, held := make(chan struct{}), make(chan struct{})
go l.Do(context.Background(), func() error { close(held); <-hold; return nil })
<-held
ctx, cancel := context.WithCancel(context.Background())
cancel()
ran := false
err := l.DoNamedLeavable(ctx, "ingest", "b", func(func()) error { ran = true; return nil })
if err == nil || ran {
t.Errorf("FAIL: a cancelled unit ran (%v) or returned %v", ran, err)
}
close(hold)
if st := l.Stats(); st.Waiting > 1 {
t.Errorf("FAIL: the cancelled unit still holds a queue position: %+v", st)
}
}
// A second leave, after another unit has taken the slot the first gave
// back, must not free that other unit's slot.
func TestASecondLeaveDoesNotReleaseAnotherUnitsSlot(t *testing.T) {
l := NewLane("test", 1, 0)
holdB, bIn := make(chan struct{}), make(chan struct{})
bDone := make(chan error, 1)
err := l.DoNamedLeavable(context.Background(), "ingest", "a", func(leave func()) error {
leave()
go func() {
bDone <- l.Do(context.Background(), func() error { close(bIn); <-holdB; return nil })
}()
<-bIn
leave()
if st := l.Stats(); st.InFlight != 1 {
t.Errorf("FAIL: after a second leave the lane runs %d units, want the other unit's 1", st.InFlight)
}
if err := l.Do(context.Background(), func() error { return nil }); err == nil {
t.Error("FAIL: a third unit got in: the second leave freed the other unit's slot")
} else {
t.Log("ok: a second leave leaves the other unit's slot alone")
}
return nil
})
if err != nil {
t.Fatal(err)
}
close(holdB)
if err := <-bDone; err != nil {
t.Fatal(err)
}
}