diff --git a/backend/library/library.go b/backend/library/library.go index 4087933..e161c0f 100644 --- a/backend/library/library.go +++ b/backend/library/library.go @@ -289,8 +289,13 @@ func (l *Library) scanInternal( l.mu.Unlock() }() + // The configured mode, not a hardcoded "auto". `ScanConcurrency` + // has been a validated config field with three values and one + // caller passing a constant, so choosing `ssd` or `hdd` by hand + // did nothing at all. + diskProfile := system.ProfileForPath(libraryPath) workerCount := resolveScanWorkerCount( - ScanConcurrencyAuto, + l.conf.ScanConcurrency, libraryPath, ) @@ -300,6 +305,10 @@ func (l *Library) scanInternal( "libraryName", libraryName, "libraryPath", libraryPath, "workers", workerCount, + "mode", l.conf.ScanConcurrency, + "device", diskProfile.Device, + "rotational", diskProfile.Rotational, + "queueDepth", diskProfile.QueueDepth, ) // Helper to build a ScanProgress with library identification. @@ -818,7 +827,7 @@ func (l *Library) scanInternal( g := new(errgroup.Group) g.SetLimit(workerCount) - for work := range workChan { + for work := range readaheadWork(scanCtx, workChan, diskProfile) { g.Go(func() error { if err := l.waitIfPaused(scanCtx); err != nil { return err @@ -1285,9 +1294,101 @@ func surveyAudioFiles( return count, maxModTime } -// hddWorkerCount is the maximum number of concurrent extraction -// workers when the library resides on a spinning disk. -const hddWorkerCount = 2 +// How many extraction workers a spinning disk gets, and why it is two +// numbers rather than one. +// +// Extraction is not CPU work — every parser here reads headers and +// returns — so on a spinning disk the whole cost is seek latency, and +// the only question worth asking is how many reads should be in flight +// at once. That has two different right answers and the drive says +// which: +// +// - A drive with command queueing (NCQ: /sys/block//device/ +// queue_depth reports 31 or 32 on any SATA disk with it enabled) +// reorders outstanding reads into the order its head passes over +// them. Handing it several at once is most of why a parallel scan +// beats a serial one at all, and four is where the returns flatten: +// the drive needs a few requests to have anything to reorder, and +// past that it is queueing requests it was already going to +// service in that order. +// - A drive without it — queue_depth 1, which is what a USB bridge +// or a pre-2004 disk reports — services one command at a time in +// the order given. Every extra worker there is one more seek +// competing for one head, and the scan gets *slower* the harder it +// is pushed. Two is kept rather than one because the readahead +// hints (see readaheadWork) do the overlapping that concurrency +// was standing in for, and one worker cannot hide a stall. +// +// This used to be a flat 2 for anything rotational, which is a +// pre-NCQ assumption: it left a modern spinning disk with a quarter of +// the queue depth it can use. +const ( + hddWorkerCountQueued = 4 + hddWorkerCountSerial = 2 +) + +// Readahead tuning. +const ( + // readaheadDepth is how many files ahead of the workers the + // prefetcher runs. It is the channel's buffer, so it is also the + // number of `WILLNEED` hints outstanding at once — comfortably more + // than a queueing drive's 32-command window is worth filling with + // one library, and small enough that a cancelled scan is not + // holding a long tail of queued reads. + readaheadDepth = 16 + + // readaheadBytes is how much of each file to pull in. Everything + // the scanner reads lives at the head: ID3v2 and FLAC's + // STREAMINFO/VORBIS_COMMENT/PICTURE blocks, and the first MPEG + // frame with its Xing header. 512 KB covers a tag carrying + // embedded cover art, which is the large case — and reading a + // little too much sequentially costs a spinning disk almost + // nothing next to the seek that got there. + readaheadBytes = 512 << 10 +) + +// readaheadWork forwards scan work while asking the kernel to fetch +// each file's header before a worker reaches it. +// +// The buffered channel *is* the lookahead: this goroutine runs ahead +// of the workers until the buffer fills, hinting every file as it goes, +// so by the time a worker takes an item the read it needs has been in +// flight for `readaheadDepth` files' worth of parsing. That is the +// only thing that helps a spinning disk here, because the per-file work +// is already header-only — every parser in `backend/metadata` reads a +// few hundred bytes and returns, so the scan is not waiting on CPU or +// on bytes, it is waiting on the head to arrive. +// +// It runs on rotational disks only. An SSD has no seek to hide and +// already has one worker per core; issuing hints there is pure syscall +// overhead against an OS readahead that is already ahead of us. +func readaheadWork( + ctx context.Context, + in <-chan scanWork, + profile system.DiskProfile, +) <-chan scanWork { + if !profile.Rotational { + return in + } + + out := make(chan scanWork, readaheadDepth) + + go func() { + defer close(out) + + for work := range in { + hintReadahead(work.absolutePath, readaheadBytes) + + select { + case out <- work: + case <-ctx.Done(): + return + } + } + }() + + return out +} // resolveScanWorkerCount returns the number of concurrent // extraction workers based on the configured concurrency mode @@ -1296,20 +1397,45 @@ func resolveScanWorkerCount( mode ScanConcurrency, libraryPath string, ) int { + return workersForProfile( + mode, + system.ProfileForPath(libraryPath), + goruntime.NumCPU(), + ) +} + +// workersForProfile is the policy on its own, so it can be tested +// against drives this machine does not have. +// +// `hdd` and `ssd` override what the device says rather than being a +// separate branch: the mode is the user overruling detection, and +// detection is right about the queue depth either way — a user who +// picks `hdd` on a queueing drive still wants that drive's queue used. +func workersForProfile( + mode ScanConcurrency, + profile system.DiskProfile, + cpus int, +) int { + spinning := profile.Rotational + switch mode { case ScanConcurrencySSD: - return goruntime.NumCPU() + spinning = false case ScanConcurrencyHDD: - return min(hddWorkerCount, goruntime.NumCPU()) - default: // auto - if system.IsRotationalDisk(libraryPath) { - return min( - hddWorkerCount, goruntime.NumCPU(), - ) - } - - return goruntime.NumCPU() + spinning = true + case ScanConcurrencyAuto: } + + if !spinning { + return cpus + } + + workers := hddWorkerCountSerial + if profile.Queues() { + workers = hddWorkerCountQueued + } + + return min(workers, cpus) } // scanWork represents a file to be processed by a worker. diff --git a/backend/library/readahead_linux.go b/backend/library/readahead_linux.go new file mode 100644 index 0000000..59f86eb --- /dev/null +++ b/backend/library/readahead_linux.go @@ -0,0 +1,38 @@ +//go:build linux + +package library + +import ( + "os" + + "golang.org/x/sys/unix" +) + +// hintReadahead asks the kernel to start fetching the head of a file +// that is about to be read. +// +// `POSIX_FADV_WILLNEED` returns immediately and queues the read, which +// is the whole point: on a spinning disk the first access to a file +// costs a seek of several milliseconds, and that latency can only be +// hidden by having the next seek already in flight while the current +// file is being parsed. A drive with command queueing can then service +// the queued reads in head order rather than in the order they were +// asked for. +// +// Errors are dropped on purpose. This is a hint: a file that has since +// been deleted, a filesystem that does not implement fadvise, or a +// permission the walk saw and this open does not, all mean "no +// prefetch", never "fail the scan". The read that follows is what +// reports a genuine problem. +func hintReadahead(path string, bytes int64) { + f, err := os.Open(path) + if err != nil { + return + } + + defer func() { _ = f.Close() }() + + _ = unix.Fadvise( + int(f.Fd()), 0, bytes, unix.FADV_WILLNEED, + ) +} diff --git a/backend/library/readahead_other.go b/backend/library/readahead_other.go new file mode 100644 index 0000000..e862e63 --- /dev/null +++ b/backend/library/readahead_other.go @@ -0,0 +1,13 @@ +//go:build !linux + +package library + +// hintReadahead is a no-op off Linux. +// +// macOS has `F_RDADVISE` and Windows has `FILE_FLAG_SEQUENTIAL_SCAN`, +// and neither is wired up here for the reason the scan concurrency +// heuristic is not either: this package cannot tell a spinning disk +// from an SSD on those platforms (see system.ProfileForPath), so it +// would be prefetching without knowing whether prefetching is what the +// device wants. +func hintReadahead(_ string, _ int64) {} diff --git a/backend/library/scan_workers_test.go b/backend/library/scan_workers_test.go new file mode 100644 index 0000000..5720c2c --- /dev/null +++ b/backend/library/scan_workers_test.go @@ -0,0 +1,122 @@ +package library + +import ( + "context" + "testing" + + "yellowjacket/backend/system" +) + +// How many workers a scan gets is decided by two facts about the +// device, and the second one is new: a spinning disk that can queue +// commands wants several reads in flight, and one that cannot wants +// almost none. Before this it was a flat 2 for anything rotational, +// which is a pre-NCQ assumption — a modern SATA disk reports a queue +// depth of 32 and was being given a quarter of what it can use. +func TestWorkersForProfile(t *testing.T) { + t.Parallel() + + const cpus = 16 + + ssd := system.DiskProfile{Device: "sda", QueueDepth: 32} + hddQueued := system.DiskProfile{ + Device: "sdb", Rotational: true, QueueDepth: 32, + } + hddSerial := system.DiskProfile{ + Device: "sdc", Rotational: true, QueueDepth: 1, + } + // Neither NVMe nor a device-mapper volume publishes queue_depth. + // An unknown depth must not be read as "cannot queue", or every + // such device would be scanned as if it were a 2003 drive. + unknown := system.DiskProfile{Device: "dm-0", Rotational: true} + + tests := []struct { + name string + mode ScanConcurrency + profile system.DiskProfile + want int + }{ + {"ssd auto", ScanConcurrencyAuto, ssd, cpus}, + {"queueing hdd auto", ScanConcurrencyAuto, hddQueued, hddWorkerCountQueued}, + {"serial hdd auto", ScanConcurrencyAuto, hddSerial, hddWorkerCountSerial}, + {"unknown depth queues", ScanConcurrencyAuto, unknown, hddWorkerCountQueued}, + + // The mode overrules detection about the *disk*, never about + // its queue: forcing hdd on a queueing drive still uses it. + {"forced hdd on an ssd", ScanConcurrencyHDD, ssd, hddWorkerCountQueued}, + {"forced ssd on an hdd", ScanConcurrencySSD, hddQueued, cpus}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + if got := workersForProfile(tt.mode, tt.profile, cpus); got != tt.want { + t.Errorf( + "workersForProfile(%q, %+v) = %d, want %d", + tt.mode, tt.profile, got, tt.want, + ) + } + }) + } +} + +// A machine with fewer cores than the policy asks for gets its cores. +func TestWorkersNeverExceedTheCPUCount(t *testing.T) { + t.Parallel() + + hdd := system.DiskProfile{Rotational: true, QueueDepth: 32} + + if got := workersForProfile(ScanConcurrencyAuto, hdd, 1); got != 1 { + t.Errorf("single-core hdd = %d workers, want 1", got) + } +} + +// The prefetch stage must forward every item and nothing else: it is a +// pass-through with a side effect, and a scan that drops a file because +// of a *hint* would be a spectacular way to lose part of a library. +func TestReadaheadForwardsEveryFile(t *testing.T) { + t.Parallel() + + in := make(chan scanWork, 4) + for _, p := range []string{"/a", "/b", "/c", "/d"} { + in <- scanWork{absolutePath: p} + } + + close(in) + + var got []string + for w := range readaheadWork( + context.Background(), + in, + system.DiskProfile{Rotational: true, QueueDepth: 32}, + ) { + got = append(got, w.absolutePath) + } + + want := []string{"/a", "/b", "/c", "/d"} + if len(got) != len(want) { + t.Fatalf("forwarded %v, want %v", got, want) + } + + for i := range want { + if got[i] != want[i] { + t.Errorf("item %d = %q, want %q", i, got[i], want[i]) + } + } +} + +// On an SSD the stage is not inserted at all — the channel comes back +// unchanged, so a scan there pays nothing for a feature it cannot use. +func TestReadaheadIsSkippedOnSolidState(t *testing.T) { + t.Parallel() + + in := make(chan scanWork) + out := readaheadWork( + context.Background(), in, system.DiskProfile{QueueDepth: 32}, + ) + + if out != (<-chan scanWork)(in) { + t.Error("an ssd must get the original channel, unwrapped") + } +}