forky: improve shard offset acquisition

storage/fcds/fcds.go

@@ -22,9 +22,11 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"sort"
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethersphere/swarm/log"
 
 	"github.com/ethersphere/swarm/chunk"
@@ -36,8 +38,9 @@ import (
 type Storer interface {
 	Get(addr chunk.Address) (ch chunk.Chunk, err error)
 	Has(addr chunk.Address) (yes bool, err error)
-	Put(ch chunk.Chunk) (err error)
+	Put(ch chunk.Chunk) (shard uint8, err error)
 	Delete(addr chunk.Address) (err error)
+	ShardSize() (slots []ShardSize, err error)
 	Count() (count int, err error)
 	Iterate(func(ch chunk.Chunk) (stop bool, err error)) (err error)
 	Close() (err error)
@@ -46,7 +49,7 @@ type Storer interface {
 var _ Storer = new(Store)
 
 // Number of files that store chunk data.
-const shardCount = 32
+var ShardCount = uint8(32)
 
 // ErrStoreClosed is returned if store is already closed.
 var ErrStoreClosed = errors.New("closed store")
@@ -56,9 +59,6 @@ var ErrStoreClosed = errors.New("closed store")
 type Store struct {
 	shards       []shard        // relations with shard id and a shard file and their mutexes
 	meta         MetaStore      // stores chunk offsets
-	free         []bool         // which shards have free offsets
-	freeMu       sync.RWMutex   // protects free field
-	freeCache    *offsetCache   // optional cache of free offset values
 	wg           sync.WaitGroup // blocks Close until all other method calls are done
 	maxChunkSize int            // maximal chunk data size
 	quit         chan struct{}  // quit disables all operations after Close is called
@@ -68,24 +68,11 @@ type Store struct {
 // Option is an optional argument passed to New.
 type Option func(*Store)
 
-// WithCache is an optional argument to New constructor that enables
-// in memory cache of free chunk data positions in files
-func WithCache(yes bool) Option {
-	return func(s *Store) {
-		if yes {
-			s.freeCache = newOffsetCache(shardCount)
-		} else {
-			s.freeCache = nil
-		}
-	}
-}
-
 // New constructs a new Store with files at path, with specified max chunk size.
 func New(path string, maxChunkSize int, metaStore MetaStore, opts ...Option) (s *Store, err error) {
 	s = &Store{
-		shards:       make([]shard, shardCount),
+		shards:       make([]shard, ShardCount),
 		meta:         metaStore,
-		free:         make([]bool, shardCount),
 		maxChunkSize: maxChunkSize,
 		quit:         make(chan struct{}),
 	}
@@ -95,7 +82,7 @@ func New(path string, maxChunkSize int, metaStore MetaStore, opts ...Option) (s
 	if err := os.MkdirAll(path, 0777); err != nil {
 		return nil, err
 	}
-	for i := byte(0); i < shardCount; i++ {
+	for i := byte(0); i < ShardCount; i++ {
 		s.shards[i].f, err = os.OpenFile(filepath.Join(path, fmt.Sprintf("chunks-%v.db", i)), os.O_CREATE|os.O_RDWR, 0666)
 		if err != nil {
 			return nil, err
@@ -105,6 +92,21 @@ func New(path string, maxChunkSize int, metaStore MetaStore, opts ...Option) (s
 	return s, nil
 }
 
+func (s *Store) ShardSize() (slots []ShardSize, err error) {
+	slots = make([]ShardSize, len(s.shards))
+	for i, sh := range s.shards {
+		sh.mu.Lock()
+		fs, err := sh.f.Stat()
+		sh.mu.Unlock()
+		if err != nil {
+			return nil, err
+		}
+		slots[i] = ShardSize{Shard: uint8(i), Size: fs.Size()}
+	}
+
+	return slots, nil
+}
+
 // Get returns a chunk with data.
 func (s *Store) Get(addr chunk.Address) (ch chunk.Chunk, err error) {
 	if err := s.protect(); err != nil {
@@ -117,18 +119,23 @@ func (s *Store) Get(addr chunk.Address) (ch chunk.Chunk, err error) {
 		return nil, err
 	}
 
-	sh := s.shards[getShard(addr)]
+	sh := s.shards[m.Shard]
 	sh.mu.Lock()
 	defer sh.mu.Unlock()
 
 	data := make([]byte, m.Size)
 	n, err := sh.f.ReadAt(data, m.Offset)
 	if err != nil && err != io.EOF {
+		metrics.GetOrRegisterCounter("fcds/get/error", nil).Inc(1)
+
 		return nil, err
 	}
 	if n != int(m.Size) {
 		return nil, fmt.Errorf("incomplete chunk data, read %v of %v", n, m.Size)
 	}
+
+	metrics.GetOrRegisterCounter("fcds/get/ok", nil).Inc(1)
+
 	return chunk.NewChunk(addr, data), nil
 }
 
@@ -139,97 +146,111 @@ func (s *Store) Has(addr chunk.Address) (yes bool, err error) {
 	}
 	defer s.unprotect()
 
-	return s.meta.Has(addr)
+	_, err = s.getMeta(addr)
+	if err != nil {
+		if err == chunk.ErrChunkNotFound {
+			metrics.GetOrRegisterCounter("fcds/has/no", nil).Inc(1)
+			return false, nil
+		}
+		metrics.GetOrRegisterCounter("fcds/has/err", nil).Inc(1)
+		return false, err
+	}
+	metrics.GetOrRegisterCounter("fcds/has/ok", nil).Inc(1)
+
+	return true, nil
 }
 
 // Put stores chunk data.
-func (s *Store) Put(ch chunk.Chunk) (err error) {
+// Returns the shard number into which the chunk was added.
+func (s *Store) Put(ch chunk.Chunk) (uint8, error) {
 	if err := s.protect(); err != nil {
-		return err
+		return 0, err
 	}
 	defer s.unprotect()
-
+	m, err := s.getMeta(ch.Address())
+	if err == nil {
+		return m.Shard, nil
+	}
 	addr := ch.Address()
 	data := ch.Data()
 
 	size := len(data)
 	if size > s.maxChunkSize {
-		return fmt.Errorf("chunk data size %v exceeds %v bytes", size, s.maxChunkSize)
+		return 0, fmt.Errorf("chunk data size %v exceeds %v bytes", size, s.maxChunkSize)
 	}
 
 	section := make([]byte, s.maxChunkSize)
 	copy(section, data)
 
-	shard := getShard(addr)
-	sh := s.shards[shard]
-
-	has, err := s.meta.Has(addr)
+	shardId, offset, reclaimed, cancel, err := s.getOffset()
 	if err != nil {
-		return err
-	}
-	if has {
-		return nil
+		return 0, err
 	}
 
+	sh := s.shards[shardId]
 	sh.mu.Lock()
 	defer sh.mu.Unlock()
 
-	offset, reclaimed, err := s.getOffset(shard)
-	if err != nil {
-		return err
+	if reclaimed {
+		metrics.GetOrRegisterCounter("fcds/put/reclaimed", nil).Inc(1)
 	}
 
 	if offset < 0 {
+		metrics.GetOrRegisterCounter("fcds/put/append", nil).Inc(1)
 		// no free offsets found,
 		// append the chunk data by
 		// seeking to the end of the file
 		offset, err = sh.f.Seek(0, io.SeekEnd)
 	} else {
+		metrics.GetOrRegisterCounter("fcds/put/offset", nil).Inc(1)
 		// seek to the offset position
 		// to replace the chunk data at that position
 		_, err = sh.f.Seek(offset, io.SeekStart)
 	}
 	if err != nil {
-		return err
+		cancel()
+		return 0, err
 	}
 
 	if _, err = sh.f.Write(section); err != nil {
-		return err
-	}
-	if reclaimed && s.freeCache != nil {
-		s.freeCache.remove(shard, offset)
+		cancel()
+		return 0, err
 	}
-	return s.meta.Set(addr, shard, reclaimed, &Meta{
+
+	err = s.meta.Set(addr, shardId, reclaimed, &Meta{
 		Size:   uint16(size),
 		Offset: offset,
+		Shard:  shardId,
 	})
+	if err != nil {
+		cancel()
+	}
+
+	return shardId, err
 }
 
-// getOffset returns an offset where chunk data can be written to
+// getOffset returns an offset on a shard where chunk data can be written to
 // and a flag if the offset is reclaimed from a previously removed chunk.
 // If offset is less then 0, no free offsets are available.
-func (s *Store) getOffset(shard uint8) (offset int64, reclaimed bool, err error) {
-	if !s.shardHasFreeOffsets(shard) {
-		return -1, false, nil
+func (s *Store) getOffset() (shard uint8, offset int64, reclaimed bool, cancel func(), err error) {
+	cancel = func() {}
+	shard, offset, cancel = s.meta.FreeOffset()
+	if offset >= 0 {
+		return shard, offset, true, cancel, nil
 	}
 
-	offset = -1
-	if s.freeCache != nil {
-		offset = s.freeCache.get(shard)
+	// each element Val is the shard size in bytes
+	shardSizes, err := s.ShardSize()
+	if err != nil {
+		return 0, 0, false, cancel, err
 	}
 
-	if offset < 0 {
-		offset, err = s.meta.FreeOffset(shard)
-		if err != nil {
-			return 0, false, err
-		}
-	}
-	if offset < 0 {
-		s.markShardWithFreeOffsets(shard, false)
-		return -1, false, nil
-	}
+	// sorting them will make the first element the largest shard and the last
+	// element the smallest shard; pick the smallest
+	sort.Sort(bySize(shardSizes))
+
+	return shardSizes[len(shardSizes)-1].Shard, -1, false, cancel, nil
 
-	return offset, true, nil
 }
 
 // Delete makes the chunk unavailable.
@@ -239,17 +260,23 @@ func (s *Store) Delete(addr chunk.Address) (err error) {
 	}
 	defer s.unprotect()
 
-	shard := getShard(addr)
-	s.markShardWithFreeOffsets(shard, true)
+	m, err := s.getMeta(addr)
+	if err != nil {
+		return err
+	}
 
-	if s.freeCache != nil {
-		m, err := s.getMeta(addr)
-		if err != nil {
-			return err
-		}
-		s.freeCache.set(shard, m.Offset)
+	mu := s.shards[m.Shard].mu
+	mu.Lock()
+	defer mu.Unlock()
+
+	err = s.meta.Remove(addr, m.Shard)
+	if err != nil {
+		metrics.GetOrRegisterCounter("fcds/delete/fail", nil).Inc(1)
+		return err
 	}
-	return s.meta.Remove(addr, shard)
+
+	metrics.GetOrRegisterCounter("fcds/delete/ok", nil).Inc(1)
+	return nil
 }
 
 // Count returns a number of stored chunks.
@@ -263,7 +290,6 @@ func (s *Store) Iterate(fn func(chunk.Chunk) (stop bool, err error)) (err error)
 		return err
 	}
 	defer s.unprotect()
-
 	for _, sh := range s.shards {
 		sh.mu.Lock()
 	}
@@ -275,7 +301,7 @@ func (s *Store) Iterate(fn func(chunk.Chunk) (stop bool, err error)) (err error)
 
 	return s.meta.Iterate(func(addr chunk.Address, m *Meta) (stop bool, err error) {
 		data := make([]byte, m.Size)
-		_, err = s.shards[getShard(addr)].f.ReadAt(data, m.Offset)
+		_, err = s.shards[m.Shard].f.ReadAt(data, m.Offset)
 		if err != nil {
 			return true, err
 		}
@@ -338,24 +364,6 @@ func (s *Store) getMeta(addr chunk.Address) (m *Meta, err error) {
 	return s.meta.Get(addr)
 }
 
-func (s *Store) markShardWithFreeOffsets(shard uint8, has bool) {
-	s.freeMu.Lock()
-	s.free[shard] = has
-	s.freeMu.Unlock()
-}
-
-func (s *Store) shardHasFreeOffsets(shard uint8) (has bool) {
-	s.freeMu.RLock()
-	has = s.free[shard]
-	s.freeMu.RUnlock()
-	return has
-}
-
-// getShard returns a shard number for the chunk address.
-func getShard(addr chunk.Address) (shard uint8) {
-	return addr[len(addr)-1] % shardCount
-}
-
 type shard struct {
 	f  *os.File
 	mu *sync.Mutex