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 {