From 7d5ca60930bfb05f0e004c70cc9e25003c0d8737 Mon Sep 17 00:00:00 2001 From: Brad Fitzpatrick Date: Thu, 8 Oct 2026 16:47:00 +0000 Subject: [PATCH] gocached: shard put-queue cleanup intents and temp files by blob On ci-gocached-aws-1's S3 Files NFS mount, the kernel serializes creates, unlinks, and renames within a directory, holding the directory's lock for each operation's server round trip of about 5ms. A benchmark measured it directly: with 64 goroutines creating and unlinking files in one directory, 62 were parked on the directory lock and about one was in an NFS RPC, for 96 creates+unlinks a second at any concurrency, while the same load over 256 directories reached 3,300 a second with 47 RPCs in flight. That one-directory ceiling is the put queue's: every copy into the main blob directory created and later unlinked a cleanup intent in the single .cleanup directory, and created its temp file in the main directory's root before renaming it out. The change in #8 stops writing intents for the most common case, blobs already present. For the blobs that are really copied, spread the remaining mutations over the 256 shards the main directory already uses: - Intents go in .cleanup//, xx being the first two hex digits of the blob's name, rather than directly in .cleanup. - The temp file is created in the blob's own shard directory instead of the main directory's root, so its rename into place stays within that one directory and locks nothing else. Both directories are created on demand: a create is tried first and the directory made only if it fails with ENOENT, as renameCreatingDir did for the rename, which no longer needs that since the temp file now lives in the target's directory. The sweeper lists .cleanup and then each shard subdirectory in it, sweeping intents in both the new layout and, for upgrades, the old flat one. That's 257 directory listings a minute instead of one, each locking only its own subdirectory, and only briefly. Nothing else lists the main directory's shard directories (usage stats come from SQLite, and GETs and eviction use exact names), so a temp file appearing in one while a copy is in flight changes nothing. Rolling back to an older version after this one has run would leave intents in shard subdirectories unswept, and the old sweeper would log each subdirectory as a malformed intent once a minute. Updates tailscale/corp#47471 Signed-off-by: Brad Fitzpatrick Change-Id: Icadb72b048c11f6e58c3332903a50438c561a283 --- gocache/gocached/putqueue.go | 95 +++++++++++--- gocache/gocached/putqueue_test.go | 211 +++++++++++++++++++++++++----- 2 files changed, 253 insertions(+), 53 deletions(-) diff --git a/gocache/gocached/putqueue.go b/gocache/gocached/putqueue.go index 62060c1..c791f61 100644 --- a/gocache/gocached/putqueue.go +++ b/gocache/gocached/putqueue.go @@ -37,6 +37,14 @@ const putQueueDirName = ".put-queue" // spool) and the sweeper later deletes the unaccounted-for blob. The // fixed-width hex time makes lexical directory order chronological, so the // sweeper can stop at the first intent that isn't due yet. +// +// Intents live in subdirectories named for the blob's shard (the first two +// hex digits of its name, like the main blob directory's), not directly in +// this directory. On a network filesystem the kernel serializes creates and +// unlinks within a directory, each holding the directory's lock for a +// server round trip, so one shared directory caps every mover combined at +// a few hundred intent operations a second. Intents written by older +// versions directly in this directory are still swept. const cleanupDirName = ".cleanup" const ( @@ -547,9 +555,12 @@ func (q *putQueue) mainPath(p *pendingPut) string { func (q *putQueue) writeCleanupIntent(p *pendingPut) error { if p.intentPath == "" { due := q.srv.now().Add(cleanupIntentDelay).Unix() - p.intentPath = filepath.Join(q.cleanupDir, fmt.Sprintf("%08x-%s", due, p.blobName())) + name := p.blobName() + p.intentPath = filepath.Join(q.cleanupDir, name[:2], fmt.Sprintf("%08x-%s", due, name)) } - f, err := os.OpenFile(p.intentPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) + f, err := createCreatingDir(filepath.Dir(p.intentPath), func() (*os.File, error) { + return os.OpenFile(p.intentPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) + }) if err != nil { return fmt.Errorf("writing cleanup intent: %w", err) } @@ -604,7 +615,48 @@ func (q *putQueue) sweepCleanupIntents() error { return err } now := q.srv.now().Unix() + // Intents written by older versions sit directly in cleanupDir. + if err := q.sweepIntentDir(q.cleanupDir, ents, now); err != nil { + return err + } for _, ent := range ents { + if !ent.IsDir() { + continue + } + name := ent.Name() + if !isShardName(name) { + q.srv.logf("put-queue: ignoring unexpected directory %q in cleanup dir", name) + continue + } + dir := filepath.Join(q.cleanupDir, name) + sub, err := os.ReadDir(dir) + if err != nil { + if os.IsNotExist(err) { + continue + } + return err + } + if err := q.sweepIntentDir(dir, sub, now); err != nil { + return err + } + } + return nil +} + +// isShardName reports whether name is a blob shard directory name: the +// first two lowercase hex digits of a blob's name. +func isShardName(name string) bool { + return len(name) == 2 && strings.Trim(name, "0123456789abcdef") == "" +} + +// sweepIntentDir processes the due cleanup intents among ents, the +// contents of dir, as described at [putQueue.sweepCleanupIntents]. +// Subdirectories are skipped. +func (q *putQueue) sweepIntentDir(dir string, ents []os.DirEntry, now int64) error { + for _, ent := range ents { + if ent.IsDir() { + continue + } name := ent.Name() due, blobName, ok := parseCleanupIntent(name) if !ok { @@ -638,7 +690,7 @@ func (q *putQueue) sweepCleanupIntents() error { default: return err } - if err := os.Remove(filepath.Join(q.cleanupDir, name)); err != nil && !os.IsNotExist(err) { + if err := os.Remove(filepath.Join(dir, name)); err != nil && !os.IsNotExist(err) { q.srv.logf("put-queue: removing cleanup intent %v: %v", name, err) } } @@ -675,7 +727,15 @@ func (q *putQueue) copyToMain(p *pendingPut) error { } defer src.Close() - tf, err := os.CreateTemp(srv.dir, fmt.Sprintf("upload-%d-*", srv.now().Unix())) + // Create the temp file in the target's own shard directory, not the + // main directory's root, so that its create and its rename into place + // lock only that shard directory: on a network filesystem the kernel + // serializes a directory's creates, unlinks, and renames, each held + // for a server round trip. + shardDir := filepath.Dir(target) + tf, err := createCreatingDir(shardDir, func() (*os.File, error) { + return os.CreateTemp(shardDir, fmt.Sprintf("upload-%d-*", srv.now().Unix())) + }) if err != nil { return err } @@ -685,28 +745,29 @@ func (q *putQueue) copyToMain(p *pendingPut) error { os.Remove(tmpName) return err } - if err := renameCreatingDir(tmpName, target); err != nil { + if err := os.Rename(tmpName, target); err != nil { os.Remove(tmpName) return err } return nil } -// renameCreatingDir renames src to dst, creating dst's parent directory if -// it doesn't exist. It attempts the rename first so that the common case, -// a shard directory that has existed since the first blob landed in it, -// costs one filesystem operation instead of an MkdirAll round trip plus -// the rename, and it recovers if the directory is removed out from under -// the server rather than trusting any remembered state. -func renameCreatingDir(src, dst string) error { - err := os.Rename(src, dst) +// createCreatingDir calls create, which creates a file in dir, and if that +// fails because dir doesn't exist, creates dir and calls create again. It +// tries create first so that the common case, a directory that has existed +// since the first file landed in it, costs one filesystem operation instead +// of an MkdirAll round trip plus the create, and it recovers if the +// directory is removed out from under the server rather than trusting any +// remembered state. +func createCreatingDir(dir string, create func() (*os.File, error)) (*os.File, error) { + f, err := create() if !errors.Is(err, fs.ErrNotExist) { - return err + return f, err } - if err := os.MkdirAll(filepath.Dir(dst), 0750); err != nil { - return err + if err := os.MkdirAll(dir, 0750); err != nil { + return nil, err } - return os.Rename(src, dst) + return create() } // flusherLoop batches entries from flushCh into single SQLite transactions. diff --git a/gocache/gocached/putqueue_test.go b/gocache/gocached/putqueue_test.go index 07ab131..411c5d1 100644 --- a/gocache/gocached/putqueue_test.go +++ b/gocache/gocached/putqueue_test.go @@ -139,45 +139,45 @@ func TestPutQueueReserveCountCap(t *testing.T) { } } -func TestRenameCreatingDir(t *testing.T) { - dir := t.TempDir() - dst := filepath.Join(dir, "ab", "blob") - write := func(name string) string { +func TestCreateCreatingDir(t *testing.T) { + dir := filepath.Join(t.TempDir(), "ab") + path := filepath.Join(dir, "file") + calls := 0 + create := func() (*os.File, error) { + calls++ + return os.Create(path) + } + check := func(wantCalls int) { t.Helper() - src := filepath.Join(dir, name) - if err := os.WriteFile(src, []byte(name), 0644); err != nil { - t.Fatal(err) + f, err := createCreatingDir(dir, create) + if err != nil { + t.Fatalf("createCreatingDir: %v", err) } - return src - } - - // The parent doesn't exist yet: created on demand. - if err := renameCreatingDir(write("one"), dst); err != nil { - t.Fatalf("rename into missing dir: %v", err) - } - if got, err := os.ReadFile(dst); err != nil || string(got) != "one" { - t.Fatalf("dst = %q, %v; want %q", got, err, "one") - } - - // The parent exists: the plain rename succeeds. - if err := renameCreatingDir(write("two"), dst); err != nil { - t.Fatalf("rename into existing dir: %v", err) + f.Close() + if calls != wantCalls { + t.Errorf("create called %d times; want %d", calls, wantCalls) + } + calls = 0 } - // The parent was removed out from under us: recreated, not cached. - if err := os.RemoveAll(filepath.Dir(dst)); err != nil { + // The directory doesn't exist yet: created on demand, then retried. + check(2) + // The directory exists: one call. + check(1) + // The directory was removed out from under us: recreated, not cached. + if err := os.RemoveAll(dir); err != nil { t.Fatal(err) } - if err := renameCreatingDir(write("three"), dst); err != nil { - t.Fatalf("rename after dir removal: %v", err) - } - if got, err := os.ReadFile(dst); err != nil || string(got) != "three" { - t.Fatalf("dst = %q, %v; want %q", got, err, "three") - } + check(2) - // A missing source is a real error, not something to paper over. - if err := renameCreatingDir(filepath.Join(dir, "nope"), dst); !errors.Is(err, fs.ErrNotExist) { - t.Fatalf("rename of missing src = %v, want not-exist", err) + // Other errors are returned as is, without creating anything. + errBoom := errors.New("boom") + other := filepath.Join(t.TempDir(), "other") + if _, err := createCreatingDir(other, func() (*os.File, error) { return nil, errBoom }); err != errBoom { + t.Errorf("err = %v; want %v", err, errBoom) + } + if _, err := os.Stat(other); !os.IsNotExist(err) { + t.Errorf("directory created after a non-ENOENT error: %v", err) } } @@ -410,8 +410,8 @@ func TestPutQueueSkipIntentWhenPresent(t *testing.T) { if p2.intentPath != "" { t.Errorf("intent %s written for an already-present blob", p2.intentPath) } - if ents, err := os.ReadDir(q.cleanupDir); err != nil || len(ents) != 0 { - t.Errorf("cleanup dir has %d entries, err=%v; want none", len(ents), err) + if got := intentFiles(t, q); len(got) != 0 { + t.Errorf("intents written: %q; want none", got) } st.wantMetric(&st.srv.m.PutQueueCopySkips, 1) if fi2, err := os.Stat(q.mainPath(p2)); err != nil || !os.SameFile(fi1, fi2) { @@ -500,6 +500,28 @@ func TestPutQueueDrainHotInstall(t *testing.T) { } } +// intentFiles returns the cleanup intent files under q's cleanup directory, +// in shard subdirectories or (from older versions) directly in it, as paths +// relative to it. +func intentFiles(t *testing.T, q *putQueue) []string { + t.Helper() + var ret []string + err := filepath.WalkDir(q.cleanupDir, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if !d.IsDir() { + rel, _ := filepath.Rel(q.cleanupDir, path) + ret = append(ret, rel) + } + return nil + }) + if err != nil { + t.Fatal(err) + } + return ret +} + func TestPutQueueCleanupIntents(t *testing.T) { st := newServerTester(t) q := st.srv.putq @@ -529,8 +551,8 @@ func TestPutQueueCleanupIntents(t *testing.T) { committed := makePending(t, q, 0, "aa01", bytes.Repeat([]byte("committed"), 1000)) q.enqueue(committed) st.drain() - if ents, err := os.ReadDir(q.cleanupDir); err != nil || len(ents) != 0 { - t.Fatalf("cleanup dir has %d entries after drain, err=%v; want empty", len(ents), err) + if got := intentFiles(t, q); len(got) != 0 { + t.Fatalf("intents left after drain: %q", got) } if !exists(blobPath(committed.blobName())) { t.Fatal("committed blob missing after drain") @@ -614,6 +636,123 @@ func TestPutQueueCleanupIntents(t *testing.T) { } } +// TestPutQueueShardedDirs checks that a copy creates its intent in the +// blob's cleanup shard subdirectory and its temp file in the blob's own +// shard directory, never in the shared top-level directories, creating +// either directory if missing, and that the sweeper handles intents in +// shard subdirectories and in the old flat layout alike. +func TestPutQueueShardedDirs(t *testing.T) { + st := newServerTester(t) + q := st.srv.putq + now := st.srv.now().Unix() + + p := makePending(t, q, 0, "ff01", bytes.Repeat([]byte("sharded"), 1000)) + q.enqueue(p) + name := p.blobName() + shardDir := filepath.Join(st.srv.dir, name[:2]) + if err := os.RemoveAll(shardDir); err != nil { + t.Fatal(err) + } + + // The shard directories don't exist yet; both are created on demand. + if err := q.writeCleanupIntent(p); err != nil { + t.Fatal(err) + } + if want := filepath.Join(q.cleanupDir, name[:2], filepath.Base(p.intentPath)); p.intentPath != want { + t.Errorf("intent path = %s; want %s", p.intentPath, want) + } + if _, err := os.Stat(p.intentPath); err != nil { + t.Errorf("intent: %v", err) + } + if err := q.copyToMain(p); err != nil { + t.Fatal(err) + } + ents, err := os.ReadDir(shardDir) + if err != nil { + t.Fatal(err) + } + if len(ents) != 1 || ents[0].Name() != name { + t.Errorf("shard dir entries = %v; want just %s", ents, name) + } + + // With the main directory's root read-only, a copy into an existing + // shard directory still works, so the temp file isn't created in the + // root. (Root ignores permissions, so this can't be checked as root.) + if os.Geteuid() != 0 { + p2 := makePending(t, q, 0, "ff02", bytes.Repeat([]byte("read-only root"), 1000)) + if err := os.MkdirAll(filepath.Dir(q.mainPath(p2)), 0750); err != nil { + t.Fatal(err) + } + if err := os.Chmod(st.srv.dir, 0555); err != nil { + t.Fatal(err) + } + err := q.copyToMain(p2) + os.Chmod(st.srv.dir, 0755) + if err != nil { + t.Errorf("copy with a read-only main directory root: %v", err) + } + } + + // The sweeper handles both layouts. + orphanSum := sha256.Sum256([]byte("sharded orphan")) + orphanName := hex.EncodeToString(orphanSum[:]) + ".lz4" + orphanBlob := filepath.Join(st.srv.dir, orphanName[:2], orphanName) + if err := os.MkdirAll(filepath.Dir(orphanBlob), 0750); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(orphanBlob, []byte("orphan"), 0644); err != nil { + t.Fatal(err) + } + writeAt := func(dir string, due int64, blob string) string { + t.Helper() + if err := os.MkdirAll(dir, 0750); err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, fmt.Sprintf("%08x-%s", due, blob)) + if err := os.WriteFile(path, nil, 0644); err != nil { + t.Fatal(err) + } + return path + } + sharded := writeAt(filepath.Join(q.cleanupDir, orphanName[:2]), now-1, orphanName) + flat := writeAt(q.cleanupDir, now-1, orphanName) + future := writeAt(filepath.Join(q.cleanupDir, orphanName[:2]), now+1000, orphanName) + strange := filepath.Join(q.cleanupDir, "not-a-shard") + if err := os.Mkdir(strange, 0750); err != nil { + t.Fatal(err) + } + + st.drain() // commits p, deleting its intent + if err := q.sweepCleanupIntents(); err != nil { + t.Fatal(err) + } + for _, path := range []string{sharded, flat, orphanBlob} { + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("%s not swept: %v", path, err) + } + } + for _, path := range []string{future, strange, q.mainPath(p)} { + if _, err := os.Stat(path); err != nil { + t.Errorf("%s: %v; want kept", path, err) + } + } + if got, want := intentFiles(t, q), []string{filepath.Join(orphanName[:2], filepath.Base(future))}; !slices.Equal(got, want) { + t.Errorf("intents left = %q; want %q", got, want) + } + st.wantMetric(&st.srv.m.PutQueueOrphansSwept, 1) +} + +func TestIsShardName(t *testing.T) { + for name, want := range map[string]bool{ + "00": true, "bf": true, "ff": true, "9a": true, + "": false, "0": false, "abc": false, "BF": false, "g0": false, ".c": false, + } { + if got := isShardName(name); got != want { + t.Errorf("isShardName(%q) = %v; want %v", name, got, want) + } + } +} + func TestParseCleanupIntent(t *testing.T) { sha := strings.Repeat("ab", 32) tests := []struct {