From edc87fcce89c55d078a5e6d0aaf08d49521ad0bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Julio=20L=C3=B3pez?= Date: Wed, 11 Mar 2020 21:47:05 -0700 Subject: [PATCH] Refactor content stats (#340) * Remove unused fields from content.Stats * Refactor content.Stats --- cli/command_snapshot_create.go | 2 +- repo/content/content_manager.go | 10 -- repo/content/content_manager_lock_free.go | 28 +++--- repo/content/stats.go | 115 +++++++++++++++++++--- snapshot/snapshotfs/upload.go | 3 +- 5 files changed, 113 insertions(+), 45 deletions(-) diff --git a/cli/command_snapshot_create.go b/cli/command_snapshot_create.go index 279a9c317..fb8ab415f 100644 --- a/cli/command_snapshot_create.go +++ b/cli/command_snapshot_create.go @@ -103,7 +103,7 @@ func snapshotSingleSource(ctx context.Context, rep *repo.Repository, u *snapshot t0 := time.Now() - rep.Content.ResetStats() + rep.Content.Stats.Reset() localEntry, err := getLocalFSEntry(ctx, sourceInfo.Path) if err != nil { diff --git a/repo/content/content_manager.go b/repo/content/content_manager.go index 2e5745347..e8e217f8b 100644 --- a/repo/content/content_manager.go +++ b/repo/content/content_manager.go @@ -208,16 +208,6 @@ func (bm *Manager) addToPackUnlocked(ctx context.Context, contentID ID, data []b return nil } -// Stats returns statistics about content manager operations. -func (bm *Manager) Stats() Stats { - return bm.stats -} - -// ResetStats resets statistics to zero values. -func (bm *Manager) ResetStats() { - bm.stats = Stats{} -} - // DisableIndexFlush increments the counter preventing automatic index flushes. func (bm *Manager) DisableIndexFlush(ctx context.Context) { bm.lock() diff --git a/repo/content/content_manager_lock_free.go b/repo/content/content_manager_lock_free.go index 2787fc15c..ac85118c3 100644 --- a/repo/content/content_manager_lock_free.go +++ b/repo/content/content_manager_lock_free.go @@ -10,7 +10,6 @@ import ( "math/rand" "strings" "sync" - "sync/atomic" "time" "github.com/pkg/errors" @@ -23,7 +22,7 @@ import ( // lockFreeManager contains parts of Manager state that can be accessed without locking type lockFreeManager struct { // this one is not lock-free - stats Stats + Stats Stats listCache *listCache st blob.Storage @@ -220,8 +219,7 @@ func (bm *lockFreeManager) getContentDataUnlocked(ctx context.Context, bi *Info) return nil, err } - atomic.AddInt32(&bm.stats.ReadContents, 1) - atomic.AddInt64(&bm.stats.ReadBytes, int64(len(payload))) + bm.Stats.readContent(len(payload)) iv, err := getPackedContentIV(bi.ID) if err != nil { @@ -242,7 +240,7 @@ func (bm *lockFreeManager) decryptAndVerify(encrypted, iv []byte) ([]byte, error return nil, errors.Wrap(err, "decrypt") } - atomic.AddInt64(&bm.stats.DecryptedBytes, int64(len(decrypted))) + bm.Stats.decrypted(len(decrypted)) if bm.encryptor.IsAuthenticated() { // already verified @@ -341,11 +339,10 @@ func (bm *lockFreeManager) getIndexBlobInternal(ctx context.Context, blobID blob return nil, err } - atomic.AddInt32(&bm.stats.ReadContents, 1) - atomic.AddInt64(&bm.stats.ReadBytes, int64(len(payload))) + bm.Stats.readContent(len(payload)) payload, err = bm.encryptor.Decrypt(payload, iv) - atomic.AddInt64(&bm.stats.DecryptedBytes, int64(len(payload))) + bm.Stats.decrypted(len(payload)) if err != nil { return nil, err @@ -373,8 +370,7 @@ func getIndexBlobIV(s blob.ID) ([]byte, error) { } func (bm *lockFreeManager) writePackFileNotLocked(ctx context.Context, packFile blob.ID, data []byte) error { - atomic.AddInt32(&bm.stats.WrittenContents, 1) - atomic.AddInt64(&bm.stats.WrittenBytes, int64(len(data))) + bm.Stats.wroteContent(len(data)) bm.listCache.deleteListCache() return bm.st.PutBlob(ctx, packFile, data) @@ -385,15 +381,14 @@ func (bm *lockFreeManager) encryptAndWriteContentNotLocked(ctx context.Context, blobID := prefix + blob.ID(hex.EncodeToString(hash)) // Encrypt the content in-place. - atomic.AddInt64(&bm.stats.EncryptedBytes, int64(len(data))) + bm.Stats.encrypted(len(data)) data2, err := bm.encryptor.Encrypt(data, hash) if err != nil { return "", err } - atomic.AddInt32(&bm.stats.WrittenContents, 1) - atomic.AddInt64(&bm.stats.WrittenBytes, int64(len(data2))) + bm.Stats.wroteContent(len(data2)) bm.listCache.deleteListCache() if err := bm.st.PutBlob(ctx, blobID, data2); err != nil { @@ -406,8 +401,7 @@ func (bm *lockFreeManager) encryptAndWriteContentNotLocked(ctx context.Context, func (bm *lockFreeManager) hashData(data []byte) []byte { // Hash the content and compute encryption key. contentID := bm.hasher(data) - atomic.AddInt32(&bm.stats.HashedContents, 1) - atomic.AddInt64(&bm.stats.HashedBytes, int64(len(data))) + bm.Stats.hashedContent(len(data)) return contentID } @@ -421,11 +415,11 @@ func (bm *lockFreeManager) verifyChecksum(data, contentID []byte) error { expected = expected[len(expected)-aes.BlockSize:] if !bytes.HasSuffix(contentID, expected) { - atomic.AddInt32(&bm.stats.InvalidContents, 1) + bm.Stats.foundInvalidContent() return errors.Errorf("invalid checksum for blob %x, expected %x", contentID, expected) } - atomic.AddInt32(&bm.stats.ValidContents, 1) + bm.Stats.foundValidContent() return nil } diff --git a/repo/content/stats.go b/repo/content/stats.go index 14001dd3d..0374ae3bb 100644 --- a/repo/content/stats.go +++ b/repo/content/stats.go @@ -1,25 +1,108 @@ package content +import ( + "sync/atomic" +) + // Stats exposes statistics about content operation. type Stats struct { - // Keep int64 fields first to ensure they get aligned to at least 64-bit boundaries - // which is required for atomic access on ARM and x86-32. - ReadBytes int64 `json:"readBytes,omitempty"` - WrittenBytes int64 `json:"writtenBytes,omitempty"` - DecryptedBytes int64 `json:"decryptedBytes,omitempty"` - EncryptedBytes int64 `json:"encryptedBytes,omitempty"` - HashedBytes int64 `json:"hashedBytes,omitempty"` + // Keep int64 fields first to ensure they get aligned to at least 64-bit + // boundaries, which is required for atomic access on ARM and x86-32. + readBytes int64 + writtenBytes int64 + decryptedBytes int64 + encryptedBytes int64 + hashedBytes int64 - ReadContents int32 `json:"readContents,omitempty"` - WrittenContents int32 `json:"writtenContents,omitempty"` - CheckedContents int32 `json:"checkedContents,omitempty"` - HashedContents int32 `json:"hashedContents,omitempty"` - InvalidContents int32 `json:"invalidContents,omitempty"` - PresentContents int32 `json:"presentContents,omitempty"` - ValidContents int32 `json:"validContents,omitempty"` + readContents uint32 + writtenContents uint32 + hashedContents uint32 + invalidContents uint32 + validContents uint32 } -// Reset clears all repository statistics. +// Reset clears all content statistics. func (s *Stats) Reset() { - *s = Stats{} + // while not atomic, it ensures values are propagated + atomic.StoreInt64(&s.readBytes, 0) + atomic.StoreInt64(&s.writtenBytes, 0) + atomic.StoreInt64(&s.decryptedBytes, 0) + atomic.StoreInt64(&s.encryptedBytes, 0) + atomic.StoreInt64(&s.hashedBytes, 0) + atomic.StoreUint32(&s.readContents, 0) + atomic.StoreUint32(&s.writtenContents, 0) + atomic.StoreUint32(&s.hashedContents, 0) + atomic.StoreUint32(&s.invalidContents, 0) + atomic.StoreUint32(&s.validContents, 0) +} + +// ReadContent returns the approximate read content count and their total size in bytes +func (s *Stats) ReadContent() (count uint32, bytes int64) { + return readCountSum(&s.readContents, &s.readBytes) +} + +// WrittenContent returns the approximate written content count and their total size in bytes +func (s *Stats) WrittenContent() (count uint32, bytes int64) { + return readCountSum(&s.writtenContents, &s.writtenBytes) +} + +// HashedContent returns the approximate hashed content count and their total size in bytes +func (s *Stats) HashedContent() (count uint32, bytes int64) { + return readCountSum(&s.hashedContents, &s.hashedBytes) +} + +// DecryptedBytes returns the approximate total number of decrypted bytes +func (s *Stats) DecryptedBytes() int64 { + return atomic.LoadInt64(&s.decryptedBytes) +} + +// EncryptedBytes returns the approximate total number of decrypted bytes +func (s *Stats) EncryptedBytes() int64 { + return atomic.LoadInt64(&s.encryptedBytes) +} + +// InvalidContents returns the approximate count of invalid contents found +func (s *Stats) InvalidContents() uint32 { + return atomic.LoadUint32(&s.invalidContents) +} + +// ValidContents returns the approximate count of valid contents found +func (s *Stats) ValidContents() uint32 { + return atomic.LoadUint32(&s.validContents) +} + +func (s *Stats) decrypted(size int) int64 { // nolint:unparam + return atomic.AddInt64(&s.decryptedBytes, int64(size)) +} + +func (s *Stats) encrypted(size int) int64 { + return atomic.AddInt64(&s.encryptedBytes, int64(size)) +} + +func (s *Stats) readContent(size int) (count uint32, sum int64) { + return updateCountSum(&s.readContents, &s.readBytes, size) +} + +func (s *Stats) wroteContent(size int) (count uint32, sum int64) { + return updateCountSum(&s.writtenContents, &s.writtenBytes, size) +} + +func (s *Stats) hashedContent(size int) (count uint32, sum int64) { + return updateCountSum(&s.hashedContents, &s.hashedBytes, size) +} + +func (s *Stats) foundValidContent() uint32 { + return atomic.AddUint32(&s.validContents, 1) +} + +func (s *Stats) foundInvalidContent() uint32 { + return atomic.AddUint32(&s.invalidContents, 1) +} + +func updateCountSum(count *uint32, sum *int64, delta int) (updatedCount uint32, updatedSum int64) { + return atomic.AddUint32(count, 1), atomic.AddInt64(sum, int64(delta)) +} + +func readCountSum(count *uint32, sum *int64) (c uint32, s int64) { + return atomic.LoadUint32(count), atomic.LoadInt64(sum) } diff --git a/snapshot/snapshotfs/upload.go b/snapshot/snapshotfs/upload.go index cab5b1fd8..c7e94b392 100644 --- a/snapshot/snapshotfs/upload.go +++ b/snapshot/snapshotfs/upload.go @@ -63,7 +63,8 @@ func (u *Uploader) cancelReason() string { return "canceled" } - if mub := u.MaxUploadBytes; mub > 0 && u.repo.Content.Stats().WrittenBytes > mub { + _, wb := u.repo.Content.Stats.WrittenContent() + if mub := u.MaxUploadBytes; mub > 0 && wb > mub { return "limit reached" }