swarm/storage: use two LRU caches for chunks containing data and requests

swarm/storage/dpa.go

@@ -72,7 +72,7 @@ func NewLocalDPA(datadir string, basekey []byte) (*DPA, error) {
 	}
 
 	return NewDPA(&LocalStore{
-		memStore: NewMemStore(dbStore, singletonSwarmCacheCapacity),
+		memStore: NewMemStore(dbStore, singletonSwarmCacheCapacity, singletonSwarmDbCapacity),
 		DbStore:  dbStore,
 	}, NewDPAParams()), nil
 }

swarm/storage/dpa_test.go

@@ -39,7 +39,7 @@ func testDpaRandom(toEncrypt bool, t *testing.T) {
 	defer tdb.close()
 	db := tdb.LDBStore
 	db.setCapacity(50000)
-	memStore := NewMemStore(db, defaultCacheCapacity)
+	memStore := NewMemStore(db, defaultCacheCapacity, defaultCacheCapacity)
 	localStore := &LocalStore{
 		memStore: memStore,
 		DbStore:  db,
@@ -68,7 +68,7 @@ func testDpaRandom(toEncrypt bool, t *testing.T) {
 	}
 	ioutil.WriteFile("/tmp/slice.bzz.16M", slice, 0666)
 	ioutil.WriteFile("/tmp/result.bzz.16M", resultSlice, 0666)
-	localStore.memStore = NewMemStore(db, defaultCacheCapacity)
+	localStore.memStore = NewMemStore(db, defaultCacheCapacity, defaultCacheCapacity)
 	resultReader = dpa.Retrieve(key)
 	for i := range resultSlice {
 		resultSlice[i] = 0
@@ -97,7 +97,7 @@ func testDPA_capacity(toEncrypt bool, t *testing.T) {
 	}
 	defer tdb.close()
 	db := tdb.LDBStore
-	memStore := NewMemStore(db, 0)
+	memStore := NewMemStore(db, 0, 0)
 	localStore := &LocalStore{
 		memStore: memStore,
 		DbStore:  db,

swarm/storage/ldbstore.go

@@ -618,6 +618,10 @@ func (s *LDBStore) writeBatches() {
 		if err != nil {
 			log.Error(fmt.Sprintf("DbStore: spawn batch write (%d chunks): %v", b.Len(), err))
 		}
+		if e >= s.capacity {
+			log.Trace(fmt.Sprintf("DbStore: collecting garbage...(%d chunks)", e))
+			s.collectGarbage(gcArrayFreeRatio)
+		}
 		close(c)
 	}
 	log.Trace(fmt.Sprintf("DbStore: quit batch write loop"))
@@ -758,6 +762,9 @@ func (s *LDBStore) setCapacity(c uint64) {
 		if ratio > 1 {
 			ratio = 1
 		}
+		for s.entryCnt > c {
+			s.collectGarbage(ratio)
+		}
 	}
 }
 

swarm/storage/localstore.go

@@ -60,7 +60,7 @@ func NewLocalStore(hash SwarmHasher, params *StoreParams, basekey []byte, mockSt
 		return nil, err
 	}
 	return &LocalStore{
-		memStore: NewMemStore(dbStore, params.CacheCapacity),
+		memStore: NewMemStore(dbStore, params.CacheCapacity, singletonSwarmDbCapacity),
 		DbStore:  dbStore,
 	}, nil
 }
@@ -72,7 +72,7 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error)
 		return nil, err
 	}
 	localStore := &LocalStore{
-		memStore: NewMemStore(dbStore, singletonSwarmDbCapacity),
+		memStore: NewMemStore(dbStore, singletonSwarmDbCapacity, singletonSwarmDbCapacity),
 		DbStore:  dbStore,
 	}
 	return localStore, nil

swarm/storage/memstore_lrucache.go

@@ -21,6 +21,7 @@ package storage
 import (
 	"bytes"
 	"fmt"
+	"sync"
 
 	lru "github.com/hashicorp/golang-lru"
 )
@@ -31,11 +32,13 @@ const (
 
 type MemStore struct {
 	cache    *lru.Cache
+	requests *lru.Cache
+	mu       sync.Mutex
 	disabled bool
 }
 
-func NewMemStore(_ *LDBStore, capacity uint) (m *MemStore) {
+func NewMemStore(_ *LDBStore, cacheCapacity uint, requestsCapacity uint) (m *MemStore) {
-	if capacity == 0 {
+	if cacheCapacity == 0 {
 		return &MemStore{
 			disabled: true,
 		}
@@ -45,13 +48,19 @@ func NewMemStore(_ *LDBStore, capacity uint) (m *MemStore) {
 		v := value.(*Chunk)
 		<-v.dbStoredC
 	}
-	c, err := lru.NewWithEvict(int(capacity), onEvicted)
+	c, err := lru.NewWithEvict(int(cacheCapacity), onEvicted)
+	if err != nil {
+		panic(err)
+	}
+
+	r, err := lru.New(int(requestsCapacity))
 	if err != nil {
 		panic(err)
 	}
 
 	return &MemStore{
 		cache:    c,
+		requests: r,
 	}
 }
 
@@ -60,6 +69,16 @@ func (m *MemStore) Get(key Key) (*Chunk, error) {
 		return nil, ErrChunkNotFound
 	}
 
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	r, ok := m.requests.Get(string(key))
+	// it is a request
+	if ok {
+		return r.(*Chunk), nil
+	}
+
+	// it is not a request
 	c, ok := m.cache.Get(string(key))
 	if !ok {
 		return nil, ErrChunkNotFound
@@ -75,14 +94,37 @@ func (m *MemStore) Put(c *Chunk) {
 	if m.disabled {
 		return
 	}
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	// it is a request
+	if c.ReqC != nil {
+		select {
+		case <-c.ReqC:
+			ok := c.GetErrored()
+			if !ok {
+				m.requests.Remove(string(c.Key))
+				return
+			}
 			m.cache.Add(string(c.Key), c)
+			m.requests.Remove(string(c.Key))
+		default:
+			m.requests.Add(string(c.Key), c)
+		}
+		return
+	}
+
+	// it is not a request
+	m.cache.Add(string(c.Key), c)
+	m.requests.Remove(string(c.Key))
 }
 
 func (m *MemStore) setCapacity(n int) {
 	if n <= 0 {
 		m.disabled = true
 	} else {
-		m = NewMemStore(nil, uint(n))
+		m = NewMemStore(nil, uint(n), singletonSwarmDbCapacity)
 	}
 }
 

swarm/storage/memstore_lrucache_test.go

@@ -52,10 +52,12 @@ func newLDBStore(t *testing.T) (*LDBStore, func()) {
 
 func TestMemStoreAndLDBStore(t *testing.T) {
 	ldb, cleanup := newLDBStore(t)
-	ldb.setCapacity(50000)
+	ldb.setCapacity(4000)
 	defer cleanup()
 
-	memStore := NewMemStore(ldb, defaultCacheCapacity)
+	cacheCap := 200
+	requestsCap := 10000
+	memStore := NewMemStore(ldb, uint(cacheCap), uint(requestsCap))
 
 	tests := []struct {
 		n         int    // number of chunks to push to memStore
@@ -73,15 +75,20 @@ func TestMemStoreAndLDBStore(t *testing.T) {
 			request:   false,
 		},
 		{
-			n:         60001,
+			n:         501,
 			chunkSize: 4096,
 			request:   false,
 		},
 		{
-			n:         60001,
+			n:         15001,
 			chunkSize: 4096,
-			request:   true,
+			request:   false,
 		},
+		//{
+		//n:         60001,
+		//chunkSize: 4096,
+		//request:   true,
+		//},
 	}
 
 	for i, tt := range tests {
@@ -102,6 +109,14 @@ func TestMemStoreAndLDBStore(t *testing.T) {
 		for i := 0; i < tt.n; i++ {
 			go ldb.Put(chunks[i])
 			memStore.Put(chunks[i])
+
+			if got := memStore.cache.Len(); got > cacheCap {
+				t.Fatalf("expected to get cache capacity less than %v, but got %v", cacheCap, got)
+			}
+
+			if got := memStore.requests.Len(); got > requestsCap {
+				t.Fatalf("expected to get requests capacity less than %v, but got %v", requestsCap, got)
+			}
 		}
 
 		for i := 0; i < tt.n; i++ {

swarm/storage/memstore_test.go

@@ -19,7 +19,7 @@ package storage
 import "testing"
 
 func newTestMemStore() *MemStore {
-	return NewMemStore(nil, defaultCacheCapacity)
+	return NewMemStore(nil, defaultCacheCapacity, singletonSwarmDbCapacity)
 }
 
 func testMemStoreRandom(n int, processors int, chunksize int, t *testing.T) {

swarm/storage/resource.go

@@ -903,7 +903,7 @@ func NewTestResourceHandler(datadir string, ethClient headerGetter, validator Re
 		return nil, err
 	}
 	localStore := &LocalStore{
-		memStore: NewMemStore(dbStore, singletonSwarmDbCapacity),
+		memStore: NewMemStore(dbStore, singletonSwarmCacheCapacity, singletonSwarmDbCapacity),
 		DbStore:  dbStore,
 	}
 	resourceChunkStore := NewResourceChunkStore(localStore, nil)