diff --git a/internal/buf/pool_test.go b/internal/buf/pool_test.go index 5ad1ba733..b681440d9 100644 --- a/internal/buf/pool_test.go +++ b/internal/buf/pool_test.go @@ -22,14 +22,22 @@ func TestPool(t *testing.T) { runtime.ReadMemStats(&ms1) + repeat := 1000000 + numGoRoutines := 30 + + if runtime.GOARCH != "amd64" { + repeat = 10000 + numGoRoutines = 10 + } + // 30 gorouties, each allocating and releasing memory 1 M times - for i := 0; i < 30; i++ { + for i := 0; i < numGoRoutines; i++ { wg.Add(1) go func() { defer wg.Done() - for j := 0; j < 1000000; j++ { + for j := 0; j < repeat; j++ { const allocSize = 100000 b := a.Allocate(allocSize) diff --git a/repo/object/object_manager_test.go b/repo/object/object_manager_test.go index 0baca9cd8..05c3ff951 100644 --- a/repo/object/object_manager_test.go +++ b/repo/object/object_manager_test.go @@ -17,7 +17,9 @@ import ( "testing" "github.com/pkg/errors" + "golang.org/x/sync/errgroup" + "github.com/kopia/kopia/internal/clock" "github.com/kopia/kopia/internal/testlogging" "github.com/kopia/kopia/repo/blob" "github.com/kopia/kopia/repo/compression" @@ -187,11 +189,82 @@ func TestCheckpointing(t *testing.T) { t.Errorf("unexpected checkpoint2: %v err: %v", checkpoint2, err) } + result2, err := writer.Checkpoint() + verifyNoError(t, err) + + if result2 != result { + t.Errorf("invalid checkpoint after result: %v vs %v", result2, result) + } + verifyFull(ctx, t, om, checkpoint3, allZeroes) verifyFull(ctx, t, om, checkpoint4, make([]byte, 2<<20)) verifyFull(ctx, t, om, result, make([]byte, 2<<20+50)) } +func TestObjectWriterRaceBetweenCheckpointAndResult(t *testing.T) { + rand.Seed(clock.Now().UnixNano()) + + ctx := testlogging.Context(t) + data := map[content.ID][]byte{} + fcm := &fakeContentManager{ + data: data, + } + + om, err := NewObjectManager(testlogging.Context(t), fcm, Format{ + Splitter: "FIXED-1M", + }, ManagerOptions{}) + if err != nil { + t.Fatalf("can't create object manager: %v", err) + } + + allZeroes := make([]byte, 1<<20-5) + repeat := 100 + + if runtime.GOARCH == "arm" { + repeat = 10 + } + + for i := 0; i < repeat; i++ { + w := om.NewWriter(ctx, WriterOptions{ + AsyncWrites: 1, + }) + + w.Write(allZeroes) + w.Write(allZeroes) + w.Write(allZeroes) + + var eg errgroup.Group + + eg.Go(func() error { + _, rerr := w.Result() + + return rerr + }) + + eg.Go(func() error { + cpID, cperr := w.Checkpoint() + if cperr == nil && cpID != "" { + ids, verr := om.VerifyObject(ctx, cpID) + if verr != nil { + return errors.Wrapf(err, "Checkpoint() returned invalid object %v", cpID) + } + + for _, id := range ids { + if id == "" { + return errors.Errorf("checkpoint returned empty id") + } + } + } + + return nil + }) + + if err := eg.Wait(); err != nil { + t.Fatal(err) + } + } +} + func verifyFull(ctx context.Context, t *testing.T, om *Manager, oid ID, want []byte) { t.Helper() diff --git a/repo/object/object_writer.go b/repo/object/object_writer.go index 313bc6f25..8974984ff 100644 --- a/repo/object/object_writer.go +++ b/repo/object/object_writer.go @@ -83,7 +83,8 @@ type objectWriter struct { splitter splitter.Splitter - writeMutex sync.Mutex + // provides mutual exclusion of all public APIs (Write, Result, Checkpoint) + mu sync.Mutex asyncWritesSemaphore chan struct{} // async writes semaphore or nil asyncWritesWG sync.WaitGroup @@ -111,8 +112,8 @@ func (w *objectWriter) Close() error { } func (w *objectWriter) Write(data []byte) (n int, err error) { - w.writeMutex.Lock() - defer w.writeMutex.Unlock() + w.mu.Lock() + defer w.mu.Unlock() dataLen := len(data) w.totalLength += int64(dataLen) @@ -240,18 +241,10 @@ func maybeCompressedContentBytes(comp compression.Compressor, output *bytes.Buff return input, false, nil } -func (w *objectWriter) drainWrites() { - w.writeMutex.Lock() - - // wait for any in-flight asynchronous writes to finish - w.asyncWritesWG.Wait() -} - -func (w *objectWriter) undrainWrites() { - w.writeMutex.Unlock() -} - func (w *objectWriter) Result() (ID, error) { + w.mu.Lock() + defer w.mu.Unlock() + // no need to hold a lock on w.indirectIndexGrowMutex, since growing index only happens synchronously // and never in parallel with calling Result() if w.buffer.Len() > 0 || len(w.indirectIndex) == 0 { @@ -260,14 +253,21 @@ func (w *objectWriter) Result() (ID, error) { } } - return w.Checkpoint() + return w.checkpointLocked() } // Checkpoint returns object ID which represents portion of the object that has already been written. // The result may be an empty object ID if nothing has been flushed yet. func (w *objectWriter) Checkpoint() (ID, error) { - w.drainWrites() - defer w.undrainWrites() + w.mu.Lock() + defer w.mu.Unlock() + + return w.checkpointLocked() +} + +func (w *objectWriter) checkpointLocked() (ID, error) { + // wait for any in-flight asynchronous writes to finish + w.asyncWritesWG.Wait() if w.contentWriteError != nil { return "", w.contentWriteError diff --git a/snapshot/snapshotfs/checkpoint_registry.go b/snapshot/snapshotfs/checkpoint_registry.go index 2164545ac..f66998117 100644 --- a/snapshot/snapshotfs/checkpoint_registry.go +++ b/snapshot/snapshotfs/checkpoint_registry.go @@ -51,6 +51,11 @@ func (r *checkpointRegistry) runCheckpoints(checkpointBuilder *dirManifestBuilde return errors.Wrapf(err, "error checkpointing %v", n) } + if de == nil { + // no checkpoint. + continue + } + if de.Type != snapshot.EntryTypeDirectory { de.Name = ".checkpointed." + de.Name + "." + uuid.New().String() } diff --git a/snapshot/snapshotfs/checkpoint_registry_test.go b/snapshot/snapshotfs/checkpoint_registry_test.go index f3e79af81..b70d206f9 100644 --- a/snapshot/snapshotfs/checkpoint_registry_test.go +++ b/snapshot/snapshotfs/checkpoint_registry_test.go @@ -18,6 +18,7 @@ func TestCheckpointRegistry(t *testing.T) { f1 := d.AddFile("f1", []byte{1, 2, 3}, os.FileMode(0o755)) f2 := d.AddFile("f2", []byte{2, 3, 4}, os.FileMode(0o755)) f3 := d.AddFile("f3", []byte{2, 3, 4}, os.FileMode(0o755)) + f4 := d.AddFile("f3", []byte{2, 3, 4}, os.FileMode(0o755)) cp.addCheckpointCallback(dir1, func() (*snapshot.DirEntry, error) { return &snapshot.DirEntry{ @@ -44,6 +45,10 @@ func TestCheckpointRegistry(t *testing.T) { }, nil }) + cp.addCheckpointCallback(f4, func() (*snapshot.DirEntry, error) { + return nil, nil + }) + // remove callback before it has a chance of firing cp.removeCheckpointCallback(f3) cp.removeCheckpointCallback(f3) diff --git a/snapshot/snapshotfs/upload.go b/snapshot/snapshotfs/upload.go index 96ff21382..840a6eaa1 100644 --- a/snapshot/snapshotfs/upload.go +++ b/snapshot/snapshotfs/upload.go @@ -127,6 +127,10 @@ func (u *Uploader) uploadFileInternal(ctx context.Context, parentCheckpointRegis return nil, err } + if checkpointID == "" { + return nil, nil + } + return newDirEntry(f, checkpointID) })