diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index bbdf710483..3f66a73c9e 100644 --- a/swarm/storage/dpa.go +++ b/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 } diff --git a/swarm/storage/dpa_test.go b/swarm/storage/dpa_test.go index 1126f05a52..c1601807a3 100644 --- a/swarm/storage/dpa_test.go +++ b/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, diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index a09834fdd7..6fdbf469d4 100644 --- a/swarm/storage/ldbstore.go +++ b/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) + } } } diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index e3d1cada0a..908a10c704 100644 --- a/swarm/storage/localstore.go +++ b/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 diff --git a/swarm/storage/memstore_lrucache.go b/swarm/storage/memstore_lrucache.go index 130491a19b..cf560886e0 100644 --- a/swarm/storage/memstore_lrucache.go +++ b/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) { - if capacity == 0 { +func NewMemStore(_ *LDBStore, cacheCapacity uint, requestsCapacity uint) (m *MemStore) { + 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, + 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) } } diff --git a/swarm/storage/memstore_lrucache_test.go b/swarm/storage/memstore_lrucache_test.go index 01b368a5bb..0771b51e6c 100644 --- a/swarm/storage/memstore_lrucache_test.go +++ b/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++ { diff --git a/swarm/storage/memstore_test.go b/swarm/storage/memstore_test.go index 1da951a64d..74ce657c73 100644 --- a/swarm/storage/memstore_test.go +++ b/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) { diff --git a/swarm/storage/resource.go b/swarm/storage/resource.go index 6d00702688..304232142f 100644 --- a/swarm/storage/resource.go +++ b/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)