diff --git a/gocache/gocached/gocached.go b/gocache/gocached/gocached.go index aa38189..68e47cb 100644 --- a/gocache/gocached/gocached.go +++ b/gocache/gocached/gocached.go @@ -1180,6 +1180,7 @@ type Server struct { PutQueueBlockedSpoolBytes expvar.Int `type:"counter" name:"put_queue_blocked_spool_bytes" help:"PUT requests that waited on the spooled lane's byte cap; raise the spool capacity if the disk has room, or look at the movers if pending never drains"` PutQueueBlockedSpoolCount expvar.Int `type:"counter" name:"put_queue_blocked_spool_count" help:"PUT requests that waited on the spooled lane's count cap; many small blobs are piling up faster than the movers copy them"` PutQueueCopyErrs expvar.Int `type:"counter" name:"put_queue_copy_errs" help:"failed attempts to copy a spooled blob into the main blob directory; retried before the PUT is dropped"` + PutQueueCopySkips expvar.Int `type:"counter" name:"put_queue_copy_skips" help:"spooled blobs already present in the main blob directory with the expected size, so neither copied nor given a cleanup intent; a subset of gocached_put_queue_copy_duration_seconds_count"` PutQueueDropped expvar.Int `type:"counter" name:"put_queue_dropped" help:"pending PUTs abandoned after repeated copy or flush failures; the client saw success but the object was lost"` PutQueueFlushes expvar.Int `type:"counter" name:"put_queue_flushes" help:"metadata batch transactions committed by the put-queue flusher"` PutQueueFlushedItems expvar.Int `type:"counter" name:"put_queue_flushed_items" help:"pending PUTs whose metadata was committed by the put-queue flusher"` diff --git a/gocache/gocached/putqueue.go b/gocache/gocached/putqueue.go index 78de358..62060c1 100644 --- a/gocache/gocached/putqueue.go +++ b/gocache/gocached/putqueue.go @@ -492,16 +492,13 @@ func (q *putQueue) moveToFlush(ctx context.Context, p *pendingPut) { q.drop(p) } -// copyTimed performs one attempt at p's intent write and blob copy into the -// main blob directory and records the attempt's wall time. +// copyTimed performs one attempt at installing p's blob in the main blob +// directory and records the attempt's wall time. func (q *putQueue) copyTimed(p *pendingPut) error { q.activeCopies.Add(1) defer q.activeCopies.Add(-1) start := time.Now() - err := q.writeCleanupIntent(p) - if err == nil { - err = q.copyToMain(p) - } + err := q.installInMain(p) if q.srv.putCopyDuration != nil { result := "ok" if err != nil { @@ -512,6 +509,36 @@ func (q *putQueue) copyTimed(p *pendingPut) error { return err } +// installInMain makes p's blob present in the main blob directory. +// +// A blob already there with the expected size is left alone, since blobs +// are content-addressed, and it gets no cleanup intent: this PUT creates no +// file that the sweeper might have to reap. An intent wouldn't protect the +// existing blob anyway. The sweeper keeps any blob that SQLite accounts +// for or that a pending PUT references, which covers p until its metadata +// commits. In production most spooled blobs are already present, and on a +// network filesystem the intent's create and later unlink are the costly +// part: they are mutations of the one cleanup directory, and the kernel +// serializes those. +// +// Otherwise it writes p's cleanup intent and then copies the blob. +func (q *putQueue) installInMain(p *pendingPut) error { + if fi, err := os.Stat(q.mainPath(p)); err == nil && fi.Size() == p.storedSize { + q.srv.m.PutQueueCopySkips.Add(1) + return nil + } + if err := q.writeCleanupIntent(p); err != nil { + return err + } + return q.copyToMain(p) +} + +// mainPath returns the path of p's blob in the main blob directory. +func (q *putQueue) mainPath(p *pendingPut) string { + name := p.blobName() + return filepath.Join(q.srv.dir, name[:2], name) +} + // writeCleanupIntent creates p's cleanup intent record: a durable note that // p's blob is about to exist in the main blob directory without SQLite // accounting, and should be deleted at the encoded time if it's still @@ -633,18 +660,14 @@ func (q *putQueue) pendingSHA(sha string) bool { return false } -// copyToMain installs p's spooled blob into the main blob directory, leaving +// copyToMain copies p's spooled blob into the main blob directory, leaving // the spool file in place: it remains the bytes served for GETs of the // pending entry, and finishCommitted may later rename it into the hot tier. -// A blob already present with the expected size is left alone; blobs are -// content-addressed, so it holds the same bytes. +// Callers use installInMain, which skips the copy if the blob is already +// present. func (q *putQueue) copyToMain(p *pendingPut) error { srv := q.srv - name := p.blobName() - target := filepath.Join(srv.dir, name[:2], name) - if fi, err := os.Stat(target); err == nil && fi.Size() == p.storedSize { - return nil - } + target := q.mainPath(p) src, err := os.Open(p.queueFile) if err != nil { @@ -872,11 +895,7 @@ empty: toFlush := batch[:0] for _, p := range batch { if p.queueFile != "" { - err := q.writeCleanupIntent(p) - if err == nil { - err = q.copyToMain(p) - } - if err != nil { + if err := q.installInMain(p); err != nil { srv.logf("put-queue: copying blob %v to main dir: %v", p.sha256hex, err) srv.m.PutQueueCopyErrs.Add(1) q.drop(p) diff --git a/gocache/gocached/putqueue_test.go b/gocache/gocached/putqueue_test.go index d99b3b4..07ab131 100644 --- a/gocache/gocached/putqueue_test.go +++ b/gocache/gocached/putqueue_test.go @@ -384,6 +384,68 @@ func TestPutQueueDrain(t *testing.T) { st.wantMetric(&st.srv.m.PutQueueFlushes, 0) } +// TestPutQueueSkipIntentWhenPresent verifies that a spooled blob already in +// the main blob directory is neither copied nor given a cleanup intent, and +// still commits, while a new blob gets its intent as usual. +func TestPutQueueSkipIntentWhenPresent(t *testing.T) { + st := newServerTester(t) + q := st.srv.putq + content := bytes.Repeat([]byte("present"), 1000) + + p1 := makePending(t, q, 0, "ee01", content) + q.enqueue(p1) + st.drain() + st.wantMetric(&st.srv.m.PutQueueCopySkips, 0) + fi1, err := os.Stat(q.mainPath(p1)) + if err != nil { + t.Fatal(err) + } + + // A different action with the same output: its blob is already there. + p2 := makePending(t, q, 0, "ee02", content) + q.enqueue(p2) + if err := q.installInMain(p2); err != nil { + t.Fatal(err) + } + 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) + } + st.wantMetric(&st.srv.m.PutQueueCopySkips, 1) + if fi2, err := os.Stat(q.mainPath(p2)); err != nil || !os.SameFile(fi1, fi2) { + t.Errorf("present blob was replaced (err=%v)", err) + } + + // A new blob still gets an intent before its copy. + p3 := makePending(t, q, 0, "ee03", bytes.Repeat([]byte("new"), 1000)) + q.enqueue(p3) + if err := q.installInMain(p3); err != nil { + t.Fatal(err) + } + if p3.intentPath == "" { + t.Fatal("no intent written for a new blob") + } + if _, err := os.Stat(p3.intentPath); err != nil { + t.Errorf("intent for new blob: %v", err) + } + st.wantMetric(&st.srv.m.PutQueueCopySkips, 0) + + // Both commit. + st.drain() + var n int + if err := st.srv.db.QueryRow("SELECT COUNT(*) FROM Actions").Scan(&n); err != nil { + t.Fatal(err) + } + if n != 3 { + t.Errorf("Actions rows = %d, want 3", n) + } + if _, err := os.Stat(q.mainPath(p3)); err != nil { + t.Errorf("new blob not installed: %v", err) + } +} + func TestPutQueueDrainFlushDup(t *testing.T) { st := newServerTester(t) q := st.srv.putq