From 9cebffc628fe2f3f886172c78be53fd3cae2ddfd Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Fri, 27 Aug 2021 04:22:18 -0700 Subject: [PATCH] Fix endurance test (#1254) --- cli/command_content_verify.go | 2 ++ cli/command_snapshot_verify.go | 4 +++ internal/gather/gather_write_buffer_chunk.go | 2 +- repo/content/committed_read_manager.go | 10 ------ repo/content/content_manager.go | 18 +++++++++- repo/maintenance/blob_gc.go | 33 +++++++++++++---- repo/maintenance/maintenance_run.go | 38 ++++++++++++-------- repo/repository.go | 7 ++++ 8 files changed, 81 insertions(+), 33 deletions(-) diff --git a/cli/command_content_verify.go b/cli/command_content_verify.go index 349592e5b..014ddcfa9 100644 --- a/cli/command_content_verify.go +++ b/cli/command_content_verify.go @@ -94,6 +94,8 @@ func (c *commandContentVerify) run(ctx context.Context, rep repo.DirectRepositor log(ctx).Infof("Verifying all contents...") + rep.DisableIndexRefresh() + throttle := new(timetrack.Throttle) est := timetrack.Start() diff --git a/cli/command_snapshot_verify.go b/cli/command_snapshot_verify.go index a6f4fc8c0..80f585da1 100644 --- a/cli/command_snapshot_verify.go +++ b/cli/command_snapshot_verify.go @@ -202,6 +202,10 @@ func (c *commandSnapshotVerify) run(ctx context.Context, rep repo.Repository) er log(ctx).Errorf("DEPRECATED: --all-sources flag has no effect and is the default when no sources are provided.") } + if dr, ok := rep.(repo.DirectRepositoryWriter); ok { + dr.DisableIndexRefresh() + } + v := &verifier{ rep: rep, tt: timetrack.Start(), diff --git a/internal/gather/gather_write_buffer_chunk.go b/internal/gather/gather_write_buffer_chunk.go index eccfe393d..276e1d123 100644 --- a/internal/gather/gather_write_buffer_chunk.go +++ b/internal/gather/gather_write_buffer_chunk.go @@ -78,7 +78,7 @@ func (a *chunkAllocator) dumpStats(ctx context.Context, prefix string) { a.mu.Lock() defer a.mu.Unlock() - log(ctx).Infof("%v (%v) - allocated %v chunks freed %v alive %v max %v free list high water mark: %v", + log(ctx).Debugf("%v (%v) - allocated %v chunks freed %v alive %v max %v free list high water mark: %v", prefix, units.Base2Bytes(int64(a.chunkSize)), a.allocated, a.freed, a.allocated-a.freed, a.allocHighWaterMark, a.freeListHighWaterMark) diff --git a/repo/content/committed_read_manager.go b/repo/content/committed_read_manager.go index 3344f41f5..3fbcc5c2a 100644 --- a/repo/content/committed_read_manager.go +++ b/repo/content/committed_read_manager.go @@ -494,16 +494,6 @@ func (sm *SharedManager) shouldRefreshIndexes() bool { return sm.timeNow().After(sm.refreshIndexesAfter) } -func (sm *SharedManager) maybeRefreshIndexes(ctx context.Context) error { - if sm.shouldRefreshIndexes() { - if err := sm.Refresh(ctx); err != nil { - return errors.Wrap(err, "error refreshing indexes") - } - } - - return nil -} - // NewSharedManager returns SharedManager that is used by SessionWriteManagers on top of a repository. func NewSharedManager(ctx context.Context, st blob.Storage, f *FormattingOptions, caching *CachingOptions, opts *ManagerOptions) (*SharedManager, error) { opts = opts.CloneOrDefault() diff --git a/repo/content/content_manager.go b/repo/content/content_manager.go index 431e1ca99..70b6b2de5 100644 --- a/repo/content/content_manager.go +++ b/repo/content/content_manager.go @@ -76,7 +76,8 @@ type IndexBlobInfo struct { // WriteManager builds content-addressable storage with encryption, deduplication and packaging on top of BLOB store. type WriteManager struct { - revision int64 // changes on each local write + revision int64 // changes on each local write + disableIndexRefresh int32 mu *sync.RWMutex cond *sync.Cond @@ -162,6 +163,16 @@ func (bm *WriteManager) DeleteContent(ctx context.Context, contentID ID) error { return bm.deletePreexistingContent(ctx, bi) } +func (bm *WriteManager) maybeRefreshIndexes(ctx context.Context) error { + if atomic.LoadInt32(&bm.disableIndexRefresh) == 0 && bm.shouldRefreshIndexes() { + if err := bm.Refresh(ctx); err != nil { + return errors.Wrap(err, "error refreshing indexes") + } + } + + return nil +} + // Intentionally passing bi by value. func (bm *WriteManager) deletePreexistingContent(ctx context.Context, ci Info) error { if ci.GetDeleted() { @@ -773,6 +784,11 @@ func (bm *WriteManager) ContentInfo(ctx context.Context, contentID ID) (Info, er return bi, err } +// DisableIndexRefresh disables index refresh for the remainder of this session. +func (bm *WriteManager) DisableIndexRefresh() { + atomic.StoreInt32(&bm.disableIndexRefresh, 1) +} + func (bm *WriteManager) lock() { bm.mu.Lock() } diff --git a/repo/maintenance/blob_gc.go b/repo/maintenance/blob_gc.go index 1286d530e..915708616 100644 --- a/repo/maintenance/blob_gc.go +++ b/repo/maintenance/blob_gc.go @@ -2,6 +2,7 @@ import ( "context" + "time" "github.com/pkg/errors" "golang.org/x/sync/errgroup" @@ -15,13 +16,14 @@ // DeleteUnreferencedBlobsOptions provides option for blob garbage collection algorithm. type DeleteUnreferencedBlobsOptions struct { - Parallel int - Prefix blob.ID - DryRun bool + Parallel int + Prefix blob.ID + DryRun bool + NotAfterTime time.Time } -// DeleteUnreferencedBlobs deletes old blobs that are no longer referenced by index entries. -// nolint:gocyclo +// DeleteUnreferencedBlobs deletes o was created after maintenance startederenced by index entries. +// nolint:gocyclo,funlen func DeleteUnreferencedBlobs(ctx context.Context, rep repo.DirectRepositoryWriter, opt DeleteUnreferencedBlobsOptions, safety SafetyParameters) (int, error) { if opt.Parallel == 0 { opt.Parallel = 16 @@ -69,17 +71,34 @@ func DeleteUnreferencedBlobs(ctx context.Context, rep repo.DirectRepositoryWrite return 0, errors.Wrap(err, "unable to load active sessions") } + cutoffTime := opt.NotAfterTime + if cutoffTime.IsZero() { + cutoffTime = rep.Time() + } + + // move the cutoff time a bit forward, because on Windows clock does not reliably move forward so we may end + // up not deleting some blobs - this only really affects tests, since BlobDeleteMinAge provides real + // protection here. + const cutoffTimeSlack = 1 * time.Second + + cutoffTime = cutoffTime.Add(cutoffTimeSlack) + // iterate all pack blobs + session blobs and keep ones that are too young or // belong to alive sessions. if err := rep.ContentManager().IterateUnreferencedBlobs(ctx, prefixes, opt.Parallel, func(bm blob.Metadata) error { - if age := rep.Time().Sub(bm.Timestamp); age < safety.BlobDeleteMinAge { + if bm.Timestamp.After(cutoffTime) { + log(ctx).Debugf(" preserving %v because it was created after maintenance started", bm.BlobID) + return nil + } + + if age := cutoffTime.Sub(bm.Timestamp); age < safety.BlobDeleteMinAge { log(ctx).Debugf(" preserving %v because it's too new (age: %v<%v)", bm.BlobID, age, safety.BlobDeleteMinAge) return nil } sid := content.SessionIDFromBlobID(bm.BlobID) if s, ok := activeSessions[sid]; ok { - if age := rep.Time().Sub(s.CheckpointTime); age < safety.SessionExpirationAge { + if age := cutoffTime.Sub(s.CheckpointTime); age < safety.SessionExpirationAge { log(ctx).Debugf(" preserving %v because it's part of an active session (%v)", bm.BlobID, sid) return nil } diff --git a/repo/maintenance/maintenance_run.go b/repo/maintenance/maintenance_run.go index ce8dd8d46..92eaf5fcc 100644 --- a/repo/maintenance/maintenance_run.go +++ b/repo/maintenance/maintenance_run.go @@ -11,7 +11,6 @@ "github.com/kopia/kopia/internal/clock" "github.com/kopia/kopia/repo" - "github.com/kopia/kopia/repo/blob" "github.com/kopia/kopia/repo/content" "github.com/kopia/kopia/repo/logging" ) @@ -124,6 +123,9 @@ type RunParameters struct { Mode Mode Params *Params + + // timestamp of the last update of maintenance schedule blob + MaintenanceStartTime time.Time } // NotOwnedError is returned when maintenance cannot run because it is owned by another user. @@ -139,6 +141,8 @@ func (e NotOwnedError) Error() string { // lock can be acquired. Lock is passed to the function, which ensures that every call to Run() // is within the exclusive context. func RunExclusive(ctx context.Context, rep repo.DirectRepositoryWriter, mode Mode, force bool, cb func(runParams RunParameters) error) error { + rep.DisableIndexRefresh() + p, err := GetParams(ctx, rep) if err != nil { return errors.Wrap(err, "unable to get maintenance params") @@ -160,7 +164,7 @@ func RunExclusive(ctx context.Context, rep repo.DirectRepositoryWriter, mode Mod return nil } - runParams := RunParameters{rep, mode, p} + runParams := RunParameters{rep, mode, p, time.Time{}} // update schedule so that we don't run the maintenance again immediately if // this process crashes. @@ -168,7 +172,14 @@ func RunExclusive(ctx context.Context, rep repo.DirectRepositoryWriter, mode Mod return errors.Wrap(err, "error updating maintenance schedule") } - if err = ensureNoClockSkew(ctx, runParams.rep.BlobReader(), runParams.rep.Time()); err != nil { + bm, err := runParams.rep.BlobReader().GetMetadata(ctx, maintenanceScheduleBlobID) + if err != nil { + return errors.Wrap(err, "error getting maintenance blob time") + } + + runParams.MaintenanceStartTime = bm.Timestamp + + if err = ensureNoClockSkew(runParams); err != nil { return errors.Wrap(err, "error checking for clock skew") } @@ -200,21 +211,17 @@ func RunExclusive(ctx context.Context, rep repo.DirectRepositoryWriter, mode Mod return cb(runParams) } -func ensureNoClockSkew(ctx context.Context, st blob.Reader, now time.Time) error { - bm, err := st.GetMetadata(ctx, maintenanceScheduleBlobID) - if err != nil { - return errors.Wrap(err, "error getting maintenance blob time") - } +func ensureNoClockSkew(rp RunParameters) error { + localTime := rp.rep.Time() + repoTime := rp.MaintenanceStartTime - repoTime := bm.Timestamp - - clockSkew := bm.Timestamp.Sub(now) + clockSkew := repoTime.Sub(localTime) if clockSkew < 0 { clockSkew = -clockSkew } if clockSkew > maxClockSkew { - return errors.Errorf("Clock skew detected: local clock is out of sync with repository timestamp by more than allowed %v (local: %v repository: %v). Refusing to run maintenance.", maxClockSkew, now, repoTime) + return errors.Errorf("Clock skew detected: local clock is out of sync with repository timestamp by more than allowed %v (local: %v repository: %v). Refusing to run maintenance.", maxClockSkew, localTime, repoTime) } return nil @@ -357,7 +364,9 @@ func runTaskRewriteContentsFull(ctx context.Context, runParams RunParameters, s func runTaskDeleteOrphanedBlobsFull(ctx context.Context, runParams RunParameters, s *Schedule, safety SafetyParameters) error { return ReportRun(ctx, runParams.rep, TaskDeleteOrphanedBlobsFull, s, func() error { - _, err := DeleteUnreferencedBlobs(ctx, runParams.rep, DeleteUnreferencedBlobsOptions{}, safety) + _, err := DeleteUnreferencedBlobs(ctx, runParams.rep, DeleteUnreferencedBlobsOptions{ + NotAfterTime: runParams.MaintenanceStartTime, + }, safety) return err }) } @@ -365,7 +374,8 @@ func runTaskDeleteOrphanedBlobsFull(ctx context.Context, runParams RunParameters func runTaskDeleteOrphanedBlobsQuick(ctx context.Context, runParams RunParameters, s *Schedule, safety SafetyParameters) error { return ReportRun(ctx, runParams.rep, TaskDeleteOrphanedBlobsQuick, s, func() error { _, err := DeleteUnreferencedBlobs(ctx, runParams.rep, DeleteUnreferencedBlobsOptions{ - Prefix: content.PackBlobIDPrefixSpecial, + NotAfterTime: runParams.MaintenanceStartTime, + Prefix: content.PackBlobIDPrefixSpecial, }, safety) return err }) diff --git a/repo/repository.go b/repo/repository.go index 8221de242..aa6865f0e 100644 --- a/repo/repository.go +++ b/repo/repository.go @@ -64,6 +64,8 @@ type DirectRepository interface { ConfigFilename() string DeriveKey(purpose []byte, keyLength int) []byte Token(password string) (string, error) + + DisableIndexRefresh() } // DirectRepositoryWriter provides low-level write access to the repository. @@ -143,6 +145,11 @@ func (r *directRepository) NewObjectWriter(ctx context.Context, opt object.Write return r.omgr.NewWriter(ctx, opt) } +// DisableIndexRefresh disables index refresh for the duration of the write session. +func (r *directRepository) DisableIndexRefresh() { + r.cmgr.DisableIndexRefresh() +} + // OpenObject opens the reader for a given object, returns object.ErrNotFound. func (r *directRepository) OpenObject(ctx context.Context, id object.ID) (object.Reader, error) { // nolint:wrapcheck