85 lines
1.7 KiB
Go
85 lines
1.7 KiB
Go
|
package jobs
|
||
|
|
|
||
|
|
import (
|
||
|
|
"errors"
|
||
|
|
"sync/atomic"
|
||
|
|
"testing"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
)
|
||
|
|
|
||
|
|
func TestClampSyncWorkers(t *testing.T) {
|
||
|
|
t.Parallel()
|
||
|
|
cases := []struct {
|
||
|
|
in, want int
|
||
|
|
}{
|
||
|
|
{0, 1},
|
||
|
|
{-2, 1},
|
||
|
|
{1, 1},
|
||
|
|
{MaxSyncWorkers, MaxSyncWorkers},
|
||
|
|
{MaxSyncWorkers + 3, MaxSyncWorkers},
|
||
|
|
}
|
||
|
|
for _, tc := range cases {
|
||
|
|
if got := ClampSyncWorkers(tc.in); got != tc.want {
|
||
|
|
t.Fatalf("ClampSyncWorkers(%d)=%d want %d", tc.in, got, tc.want)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestSyncSlotsBoundsConcurrent(t *testing.T) {
|
||
|
|
t.Parallel()
|
||
|
|
slots := NewSyncSlots(2)
|
||
|
|
|
||
|
|
var inflight atomic.Int32
|
||
|
|
var maxInflight atomic.Int32
|
||
|
|
var claimed atomic.Int32
|
||
|
|
|
||
|
|
claim := func() error {
|
||
|
|
if claimed.Add(1) > 4 {
|
||
|
|
return pgx.ErrNoRows
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
for i := 0; i < 8; i++ {
|
||
|
|
_, err := slots.TryStart(claim, func() {
|
||
|
|
n := inflight.Add(1)
|
||
|
|
for {
|
||
|
|
cur := maxInflight.Load()
|
||
|
|
if n <= cur || maxInflight.CompareAndSwap(cur, n) {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
}
|
||
|
|
time.Sleep(30 * time.Millisecond)
|
||
|
|
inflight.Add(-1)
|
||
|
|
})
|
||
|
|
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
t.Fatalf("claim: %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
slots.Wait()
|
||
|
|
if maxInflight.Load() > 2 {
|
||
|
|
t.Fatalf("max inflight=%d want <=2", maxInflight.Load())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestSyncSlotsRejectsWhenFull(t *testing.T) {
|
||
|
|
t.Parallel()
|
||
|
|
slots := NewSyncSlots(1)
|
||
|
|
block := make(chan struct{})
|
||
|
|
started, err := slots.TryStart(func() error { return nil }, func() { <-block })
|
||
|
|
if err != nil || !started {
|
||
|
|
t.Fatalf("first start: started=%v err=%v", started, err)
|
||
|
|
}
|
||
|
|
started, err = slots.TryStart(func() error {
|
||
|
|
t.Fatal("should not claim when full")
|
||
|
|
return nil
|
||
|
|
}, func() {})
|
||
|
|
if err != nil || started {
|
||
|
|
t.Fatalf("second start: started=%v err=%v", started, err)
|
||
|
|
}
|
||
|
|
close(block)
|
||
|
|
slots.Wait()
|
||
|
|
}
|