|
|
|
@@ -13,11 +13,20 @@ import (
|
|
|
|
|
"time"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// poolUnwind bounds how long a goroutine is given to leave a pool
|
|
|
|
|
// after its context is cancelled. Only a failing run ever waits this
|
|
|
|
|
// long: a pool that ignored its cancellation parks forever, and this
|
|
|
|
|
// is what turns that into a failed assertion instead of a suite that
|
|
|
|
|
// hangs until the test binary's own timeout.
|
|
|
|
|
// This file gathers the tests for scan cancellation and worker-pool
|
|
|
|
|
// unwinding. Everything it exercises lives in scan.go, so by the repo's
|
|
|
|
|
// convention of one test file per source file it would belong in
|
|
|
|
|
// scan_test.go. It is kept separate on purpose: cancellation behaviour
|
|
|
|
|
// cuts across both the walk pool and the hash pool as a single concern,
|
|
|
|
|
// and scan_test.go is already 900+ lines. That is the deliberate
|
|
|
|
|
// exception the convention otherwise expects to be stated.
|
|
|
|
|
|
|
|
|
|
// poolUnwind bounds how long a goroutine is given to leave a pool after
|
|
|
|
|
// its context is cancelled. The tests that use it turn a pool that
|
|
|
|
|
// ignored its cancellation — and so parks forever — into a failed
|
|
|
|
|
// assertion within this bound instead of a hang. Not every test in this
|
|
|
|
|
// file has that property: a few catch their regression only as the test
|
|
|
|
|
// binary's own timeout, and each of those says so.
|
|
|
|
|
const poolUnwind = 2 * time.Second
|
|
|
|
|
|
|
|
|
|
// walkClock is a context whose cancellation is driven by the scan's
|
|
|
|
@@ -28,9 +37,12 @@ const poolUnwind = 2 * time.Second
|
|
|
|
|
//
|
|
|
|
|
// The accounting behind the n chosen by each test: every blocking
|
|
|
|
|
// channel operation in the walk selects on Done, so the walk spends
|
|
|
|
|
// one consultation per file event plus a couple per directory, while
|
|
|
|
|
// the index load that runs ahead of it spends a small fixed number
|
|
|
|
|
// (three) whatever the record count.
|
|
|
|
|
// one consultation per file event plus a couple per directory. The
|
|
|
|
|
// index load that runs ahead of it also consults Done, but a bounded
|
|
|
|
|
// number of times that does not grow with the record count. The tests
|
|
|
|
|
// depend on that property, not on the bound's exact value: each picks
|
|
|
|
|
// an n comfortably above it and well short of the walk's total, so the
|
|
|
|
|
// cancellation lands deep inside the walk whatever the record count.
|
|
|
|
|
type walkClock struct {
|
|
|
|
|
n int64
|
|
|
|
|
seen atomic.Int64
|
|
|
|
@@ -86,11 +98,15 @@ func (c *walkClock) Value(_ any) any {
|
|
|
|
|
// directory still queued and only the handful already in flight can
|
|
|
|
|
// emit anything more.
|
|
|
|
|
const (
|
|
|
|
|
walkCancelDirs = 100
|
|
|
|
|
walkCancelFilesPerDir = 20
|
|
|
|
|
walkCancelFiles = walkCancelDirs * walkCancelFilesPerDir
|
|
|
|
|
walkCancelWorkers = 4
|
|
|
|
|
walkCancelInFlightDirs = walkCancelWorkers * walkCancelFilesPerDir
|
|
|
|
|
walkCancelDirs = 100
|
|
|
|
|
walkCancelFilesPerDir = 20
|
|
|
|
|
walkCancelFiles = walkCancelDirs * walkCancelFilesPerDir
|
|
|
|
|
walkCancelWorkers = 4
|
|
|
|
|
// The files carried by the walkCancelWorkers directories already in
|
|
|
|
|
// flight when the scan is cancelled: each such directory can still
|
|
|
|
|
// emit its walkCancelFilesPerDir files. This is a file count, not a
|
|
|
|
|
// directory count.
|
|
|
|
|
walkCancelInFlightFiles = walkCancelWorkers * walkCancelFilesPerDir
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// walkCancelAtDone is the consultation on which the fixture's context
|
|
|
|
@@ -154,6 +170,11 @@ func assertRecordsIntact(t *testing.T, db *sql.DB, before []string) {
|
|
|
|
|
// phase and fail there instead, with a wrapped error rather than the
|
|
|
|
|
// bare cancellation.
|
|
|
|
|
//
|
|
|
|
|
// The syncScan call here is not bounded by poolUnwind: a regression
|
|
|
|
|
// that left a worker pool parked would hang it, and that regression is
|
|
|
|
|
// caught only by the test binary's own timeout, not by a quick
|
|
|
|
|
// assertion.
|
|
|
|
|
//
|
|
|
|
|
//nolint:paralleltest // counts goroutines: must not run beside others
|
|
|
|
|
func TestSyncScanCancelledMidWalkKeepsRecords(t *testing.T) {
|
|
|
|
|
dir := buildWalkCancelTree(t)
|
|
|
|
@@ -211,10 +232,11 @@ func assertWalkGuardAborted(t *testing.T, st scanStats, err error) {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// The workers drop every directory still queued once the scan is
|
|
|
|
|
// cancelled, so only the directories already in flight can add to
|
|
|
|
|
// the census after the fact. A census beyond that bound would mean
|
|
|
|
|
// the cancellation was not observed where it should have been.
|
|
|
|
|
limit := walkCancelAtDone + walkCancelInFlightDirs
|
|
|
|
|
// cancelled, so only the files in the directories already in flight
|
|
|
|
|
// can add to the census after the fact. A census beyond that bound
|
|
|
|
|
// would mean the cancellation was not observed where it should have
|
|
|
|
|
// been.
|
|
|
|
|
limit := walkCancelAtDone + walkCancelInFlightFiles
|
|
|
|
|
if st.unchanged > limit {
|
|
|
|
|
t.Errorf("census covers %d files, want at most %d: the walk kept "+
|
|
|
|
|
"taking directories off the queue after cancellation",
|
|
|
|
@@ -412,7 +434,11 @@ func TestDispatchDirsClosesJobsWhenCancelled(t *testing.T) {
|
|
|
|
|
|
|
|
|
|
// TestFeedHashJobsClosesJobsWhenCancelled checks that the hash feeder
|
|
|
|
|
// abandons the runs it has not queued yet and still closes the job
|
|
|
|
|
// channel, which is what lets the workers' range terminate.
|
|
|
|
|
// channel, which is what lets the workers' range terminate. The
|
|
|
|
|
// receive on jobs below is not bounded: a feeder that returned without
|
|
|
|
|
// closing jobs would leave that receive with no sender and no close, so
|
|
|
|
|
// this regression is caught by the test binary's timeout rather than by
|
|
|
|
|
// a bounded assertion.
|
|
|
|
|
func TestFeedHashJobsClosesJobsWhenCancelled(t *testing.T) {
|
|
|
|
|
t.Parallel()
|
|
|
|
|
|
|
|
|
@@ -440,6 +466,16 @@ func TestFeedHashJobsClosesJobsWhenCancelled(t *testing.T) {
|
|
|
|
|
// nobody wants the hashes of — while still letting the range run out
|
|
|
|
|
// so the pool tears down. The queued run names a file that does not
|
|
|
|
|
// exist, so a worker that hashed it anyway would produce a result.
|
|
|
|
|
//
|
|
|
|
|
// hashWorker has a second cancellation exit: the send of a completed
|
|
|
|
|
// result on the results channel. That branch is reachable from the
|
|
|
|
|
// production scan path, not dead code — pool.stop() cancels the context
|
|
|
|
|
// before it starts draining results, so a worker parked on that send
|
|
|
|
|
// leaves through this case, freed by the drain rather than by an empty
|
|
|
|
|
// jobs channel. It is exercised by TestScanHashWriteFailureUnwindsPool,
|
|
|
|
|
// which strands every worker on a full results channel until stop()
|
|
|
|
|
// unwinds the pool. With both branches covered, every cancellation
|
|
|
|
|
// branch of the walk and hash pools has a test.
|
|
|
|
|
func TestHashWorkerDropsQueuedRuns(t *testing.T) {
|
|
|
|
|
t.Parallel()
|
|
|
|
|
|
|
|
|
@@ -472,7 +508,10 @@ func TestHashWorkerDropsQueuedRuns(t *testing.T) {
|
|
|
|
|
// TestHashPhaseCancelledReturnsContextError checks the result loop's
|
|
|
|
|
// own exit: with the pool cancelled, no result will ever arrive, and
|
|
|
|
|
// the loop must leave through the cancellation rather than wait for a
|
|
|
|
|
// receive that cannot happen.
|
|
|
|
|
// receive that cannot happen. This call is not bounded by poolUnwind: a
|
|
|
|
|
// loop that dropped its cancellation case would block on that receive,
|
|
|
|
|
// so the regression surfaces as the test binary's timeout rather than
|
|
|
|
|
// as a bounded assertion.
|
|
|
|
|
func TestHashPhaseCancelledReturnsContextError(t *testing.T) {
|
|
|
|
|
t.Parallel()
|
|
|
|
|
|
|
|
|
|