Refactor content stats (#340)

* Remove unused fields from content.Stats
* Refactor content.Stats

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 {

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()

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
 }

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)
 }

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"
 	}