diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go
index dc0563bb44..5854fa9140 100644
--- a/swarm/storage/localstore/localstore.go
+++ b/swarm/storage/localstore/localstore.go
@@ -63,9 +63,14 @@ type DB struct {
// retrieval indexes
retrievalDataIndex shed.Index
retrievalAccessIndex shed.Index
- // sync indexes
+ // push syncing index
pushIndex shed.Index
+ // provides push syncing subscriptions
+ pushFeed *feed
+ // pull syncing index
pullIndex shed.Index
+ // provides pull syncing subscriptions
+ pullFeed *feed
// garbage collection index
gcIndex shed.Index
// index that stores hashes that are not
@@ -249,6 +254,8 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
if err != nil {
return nil, err
}
+ // create a pull syncing feed used by SubscribePull function
+ db.pullFeed = newFeed(db.pullIndex, db.retrievalDataIndex)
// push index contains as yet unsynced chunks
db.pushIndex, err = db.shed.NewIndex("StoredTimestamp|Hash->nil", shed.IndexFuncs{
EncodeKey: func(fields shed.Item) (key []byte, err error) {
@@ -272,6 +279,8 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
if err != nil {
return nil, err
}
+ // create a push syncing feed used by SubscribePush function
+ db.pushFeed = newFeed(db.pushIndex, db.retrievalDataIndex)
// gc index for removable chunk ordered by ascending last access time
db.gcIndex, err = db.shed.NewIndex("AccessTimestamp|StoredTimestamp|Hash->nil", shed.IndexFuncs{
EncodeKey: func(fields shed.Item) (key []byte, err error) {
@@ -351,6 +360,9 @@ func (db *DB) Close() (err error) {
if err := db.writeGCSize(atomic.LoadInt64(&db.gcSize)); err != nil {
log.Error("localstore: write gc size", "err", err)
}
+ // stop all subscriptions
+ db.pullFeed.close()
+ db.pushFeed.close()
return db.shed.Close()
}
diff --git a/swarm/storage/localstore/localstore_test.go b/swarm/storage/localstore/localstore_test.go
index cd2812a138..4be6df537e 100644
--- a/swarm/storage/localstore/localstore_test.go
+++ b/swarm/storage/localstore/localstore_test.go
@@ -67,6 +67,8 @@ func TestDB(t *testing.T) {
// setting a custom testHookUpdateGC function with a sleep
// and a count current and maximal number of goroutines.
func TestDB_updateGCSem(t *testing.T) {
+ t.Parallel()
+
defer func(m int) { maxParallelUpdateGC = m }(maxParallelUpdateGC)
maxParallelUpdateGC = 3
diff --git a/swarm/storage/localstore/mode_put.go b/swarm/storage/localstore/mode_put.go
index 85e8b17e89..71d62adf3a 100644
--- a/swarm/storage/localstore/mode_put.go
+++ b/swarm/storage/localstore/mode_put.go
@@ -72,6 +72,12 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
batch := new(leveldb.Batch)
+ // variables that provide information for operations
+ // to be done after write batch function successfully executes
+ var gcSizeChange int64 // number to add or subtract from gcSize
+ var triggerPullFeed bool // signal pull feed subscriptions to iterate
+ var triggerPushFeed bool // signal push feed subscriptions to iterate
+
switch mode {
case ModePutRequest:
// put to indexes: retrieve, gc; it does not enter the syncpool
@@ -99,7 +105,7 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
if item.AccessTimestamp != 0 {
// delete current entry from the gc index
db.gcIndex.DeleteInBatch(batch, item)
- db.incGCSize(-1)
+ gcSizeChange--
}
if item.StoreTimestamp == 0 {
item.StoreTimestamp = now()
@@ -111,7 +117,7 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
// add new entry to gc index
db.gcIndex.PutInBatch(batch, item)
db.gcUncountedHashesIndex.PutInBatch(batch, item)
- db.incGCSize(1)
+ gcSizeChange++
db.retrievalDataIndex.PutInBatch(batch, item)
db.retrievalAccessIndex.PutInBatch(batch, item)
@@ -122,7 +128,9 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
item.StoreTimestamp = now()
db.retrievalDataIndex.PutInBatch(batch, item)
db.pullIndex.PutInBatch(batch, item)
+ triggerPullFeed = true
db.pushIndex.PutInBatch(batch, item)
+ triggerPushFeed = true
case ModePutSync:
// put to indexes: retrieve, pull
@@ -130,10 +138,24 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
item.StoreTimestamp = now()
db.retrievalDataIndex.PutInBatch(batch, item)
db.pullIndex.PutInBatch(batch, item)
+ triggerPullFeed = true
default:
return ErrInvalidMode
}
- return db.shed.WriteBatch(batch)
+ err = db.shed.WriteBatch(batch)
+ if err != nil {
+ return err
+ }
+ if gcSizeChange != 0 {
+ db.incGCSize(gcSizeChange)
+ }
+ if triggerPullFeed {
+ db.pullFeed.trigger([]byte{db.po(item.Address)})
+ }
+ if triggerPushFeed {
+ db.pushFeed.trigger(nil)
+ }
+ return nil
}
diff --git a/swarm/storage/localstore/mode_set.go b/swarm/storage/localstore/mode_set.go
index 80c7fd3892..20a15c4098 100644
--- a/swarm/storage/localstore/mode_set.go
+++ b/swarm/storage/localstore/mode_set.go
@@ -71,6 +71,11 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
batch := new(leveldb.Batch)
+ // variables that provide information for operations
+ // to be done after write batch function successfully executes
+ var gcSizeChange int64 // number to add or subtract from gcSize
+ var triggerPullFeed bool // signal pull feed subscriptions to iterate
+
item := addressToItem(addr)
switch mode {
@@ -97,7 +102,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
case nil:
item.AccessTimestamp = i.AccessTimestamp
db.gcIndex.DeleteInBatch(batch, item)
- db.incGCSize(-1)
+ gcSizeChange--
case leveldb.ErrNotFound:
// the chunk is not accessed before
default:
@@ -106,9 +111,10 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
item.AccessTimestamp = now()
db.retrievalAccessIndex.PutInBatch(batch, item)
db.pullIndex.PutInBatch(batch, item)
+ triggerPullFeed = true
db.gcIndex.PutInBatch(batch, item)
db.gcUncountedHashesIndex.PutInBatch(batch, item)
- db.incGCSize(1)
+ gcSizeChange++
case ModeSetSync:
// delete from push, insert to gc
@@ -135,7 +141,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
case nil:
item.AccessTimestamp = i.AccessTimestamp
db.gcIndex.DeleteInBatch(batch, item)
- db.incGCSize(-1)
+ gcSizeChange--
case leveldb.ErrNotFound:
// the chunk is not accessed before
default:
@@ -146,7 +152,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
db.pushIndex.DeleteInBatch(batch, item)
db.gcIndex.PutInBatch(batch, item)
db.gcUncountedHashesIndex.PutInBatch(batch, item)
- db.incGCSize(1)
+ gcSizeChange++
case modeSetRemove:
// delete from retrieve, pull, gc
@@ -178,12 +184,22 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
// as delete is not reporting if the key/value pair
// is deleted or not
if _, err := db.gcIndex.Get(item); err == nil {
- db.incGCSize(-1)
+ gcSizeChange = -1
}
default:
return ErrInvalidMode
}
- return db.shed.WriteBatch(batch)
+ err = db.shed.WriteBatch(batch)
+ if err != nil {
+ return err
+ }
+ if gcSizeChange != 0 {
+ db.incGCSize(gcSizeChange)
+ }
+ if triggerPullFeed {
+ db.pullFeed.trigger([]byte{db.po(item.Address)})
+ }
+ return nil
}
diff --git a/swarm/storage/localstore/subscriptions.go b/swarm/storage/localstore/subscriptions.go
new file mode 100644
index 0000000000..a60260d989
--- /dev/null
+++ b/swarm/storage/localstore/subscriptions.go
@@ -0,0 +1,261 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package localstore
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "sync"
+
+ "github.com/ethereum/go-ethereum/swarm/shed"
+ "github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+var ErrSubscriptionFeedClosed = errors.New("subscription feed closed")
+
+// SubscribePull returns a Subscription for pull syncing index.
+// Pull syncing index can be only subscribed to a particular
+// proximity order bin.
+func (db *DB) SubscribePull(ctx context.Context, bin uint8) (s *Subscription, err error) {
+ return db.pullFeed.subscribe(ctx, []byte{bin})
+}
+
+// SubscribePush returns a Subscription for push syncing index.
+func (db *DB) SubscribePush(ctx context.Context) (s *Subscription, err error) {
+ return db.pushFeed.subscribe(ctx, nil)
+}
+
+// Subscription provides stream of Chunks in a particular order
+// through the Chunks channel. That channel will not be closed
+// when the last Chunk is read, but will block until the new Chunk
+// is added to database index. Subscription should be used for
+// getting Chunks and waiting for new ones. It provides methods
+// to control and get information about subscription state.
+type Subscription struct {
+ // Chunks is the read-only channel that provides stream of chunks.
+ // This is the subscription main purpose.
+ Chunks <-chan storage.Chunk
+
+ // subscribe to set of keys only with this prefix
+ prefix []byte
+ // signals subscription to gracefully stop
+ stopChan chan struct{}
+ // protects stopChan form multiple closing
+ stopOnce sync.Once
+ // provides information if subscription is done
+ doneChan chan struct{}
+ // trigger signals a new index iteration
+ // when index receives new items
+ trigger chan struct{}
+ // an error from the subscription, if any
+ err error
+ // protects err field
+ mu sync.RWMutex
+}
+
+// Done returns a read-only channel that will be closed
+// when the subscription is stopped or encountered an error.
+func (s *Subscription) Done() <-chan struct{} {
+ return s.doneChan
+}
+
+// Err returns an error that subscription encountered.
+// It should be usually called after the Done is read from.
+// It is safe to call this function multiple times.
+func (s *Subscription) Err() (err error) {
+ s.mu.RLock()
+ err = s.err
+ s.mu.RUnlock()
+ return err
+}
+
+// Stop terminates the subscription without any error.
+// It is safe to call this function multiple times.
+func (s *Subscription) Stop() {
+ s.stopOnce.Do(func() {
+ close(s.stopChan)
+ })
+}
+
+// feed is a collection of Chunks subscriptions of order given
+// by sort index and Chunk data provided by data index.
+// It provides methods to create, trigger and remove subscriptions.
+// It is the internal core component for push and pull
+// index subscriptions.
+type feed struct {
+ // index on which keys the order of Chunks will be
+ // provided by subscriptions
+ sortIndex shed.Index
+ // index that contains chunk data
+ dataIndex shed.Index
+ // collection fo subscriptions on this feed
+ subscriptions []*Subscription
+ // protects subscriptions slice
+ mu sync.Mutex
+ // closed when subscription is closed
+ closeChan chan struct{}
+ // protects closeChan form multiple closing
+ closeOnce sync.Once
+}
+
+// newFeed creates a new feed with from sort and data indexes.
+// Sort index provides ordering of Chunks and data index
+// provides Chunk data.
+func newFeed(sortIndex, dataIndex shed.Index) (f *feed) {
+ return &feed{
+ sortIndex: sortIndex,
+ dataIndex: dataIndex,
+ subscriptions: make([]*Subscription, 0),
+ closeChan: make(chan struct{}),
+ }
+}
+
+// subscribe creates a new subscription on the feed.
+// It creates a new goroutine which will iterate over existing sort index keys
+// and creates new iterators when trigger method is called.
+func (f *feed) subscribe(ctx context.Context, prefix []byte) (s *Subscription, err error) {
+ // prevent new subscription after the feed is closed
+ select {
+ case <-f.closeChan:
+ return nil, ErrSubscriptionFeedClosed
+ default:
+ }
+ chunks := make(chan storage.Chunk)
+ s = &Subscription{
+ Chunks: chunks,
+ prefix: prefix,
+ stopChan: make(chan struct{}),
+ doneChan: make(chan struct{}),
+ trigger: make(chan struct{}, 1),
+ }
+ f.mu.Lock()
+ f.subscriptions = append(f.subscriptions, s)
+ f.mu.Unlock()
+
+ // send signal for the initial iteration
+ s.trigger <- struct{}{}
+
+ go func() {
+ // this error will be set in deferred unsubscribe
+ // function call and set as Subscription.err value
+ var err error
+ defer func() {
+ f.unsubscribe(s, err)
+ }()
+
+ // startFrom is the Item from which the next iteration
+ // should start. The first iteration starts from the first Item.
+ var startFrom *shed.Item
+ for {
+ select {
+ case <-s.trigger:
+ // iterate until:
+ // - last index Item is reached
+ // - subscription stop is called
+ // - context is done
+ err = f.sortIndex.Iterate(func(item shed.Item) (stop bool, err error) {
+ // get chunk data
+ dataItem, err := f.dataIndex.Get(item)
+ if err != nil {
+ return true, err
+ }
+
+ select {
+ case chunks <- storage.NewChunk(dataItem.Address, dataItem.Data):
+ // set next iteration start item
+ // when its chunk is successfully sent to channel
+ startFrom = &item
+ return false, nil
+ case <-s.stopChan:
+ // gracefully stop the iteration
+ return true, nil
+ case <-ctx.Done():
+ return true, ctx.Err()
+ }
+ }, &shed.IterateOptions{
+ StartFrom: startFrom,
+ // startFrom was sent as the last Chunk in the previous
+ // iterator call, skip it in this one
+ SkipStartFromItem: true,
+ Prefix: prefix,
+ })
+ if err != nil {
+ return
+ }
+ case <-s.stopChan:
+ // gracefully stop the iteration
+ return
+ case <-ctx.Done():
+ if err == nil {
+ err = ctx.Err()
+ }
+ return
+ }
+ }
+ }()
+
+ return s, nil
+}
+
+// unsubscribe removes a subscription from the feed.
+// This function is called when subscription goroutine terminates
+// to cleanup feed subscriptions and set error on subscription.
+func (f *feed) unsubscribe(s *Subscription, err error) {
+ s.mu.Lock()
+ s.err = err
+ s.mu.Unlock()
+
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ for i, sub := range f.subscriptions {
+ if sub == s {
+ f.subscriptions = append(f.subscriptions[:i], f.subscriptions[i+1:]...)
+ }
+ }
+
+ // signal that the subscription is done
+ close(s.doneChan)
+}
+
+// close stops all subscriptions and prevents any new subscriptions
+// to be made by closing the closeChan.
+func (f *feed) close() {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ for _, s := range f.subscriptions {
+ s.Stop()
+ }
+ f.closeOnce.Do(func() {
+ close(f.closeChan)
+ })
+}
+
+// trigger signals all subscriptions with tprovided prefix
+// that they should continue iterating over index keys
+// where they stopped in the last iteration. This method
+// should be called when new data is put to the index.
+func (f *feed) trigger(prefix []byte) {
+ for _, s := range f.subscriptions {
+ if bytes.Equal(prefix, s.prefix) {
+ select {
+ case s.trigger <- struct{}{}:
+ default:
+ }
+ }
+ }
+}
diff --git a/swarm/storage/localstore/subscriptions_test.go b/swarm/storage/localstore/subscriptions_test.go
new file mode 100644
index 0000000000..066715033f
--- /dev/null
+++ b/swarm/storage/localstore/subscriptions_test.go
@@ -0,0 +1,115 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package localstore
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// TestSubscribePush uploads some chunks before and after
+// push syncing subscription is created and validates if
+// all chunks are received in the right order.
+func TestSubscribePush(t *testing.T) {
+ t.Parallel()
+
+ db, cleanupFunc := newTestDB(t, nil)
+ defer cleanupFunc()
+
+ uploader := db.NewPutter(ModePutUpload)
+
+ chunks := make([]storage.Chunk, 0)
+
+ uploadRandomChunks := func(count int) {
+ for i := 0; i < count; i++ {
+ chunk := generateRandomChunk()
+
+ err := uploader.Put(chunk)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ chunks = append(chunks, chunk)
+ }
+ }
+
+ // prepopulate database with some chunks
+ // before the subscription
+ uploadRandomChunks(10)
+
+ // set a timeout on subscription
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ // collect all errors from validating chunks, even nil ones
+ // to validate the number of chunks received by the subscription
+ errChan := make(chan error)
+
+ sub, err := db.SubscribePush(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer sub.Stop()
+
+ // receive and validate chunks from the subscription
+ go func() {
+ var i int // chunk index
+ for {
+ select {
+ case got := <-sub.Chunks:
+ want := chunks[i]
+ var err error
+ if !bytes.Equal(got.Data(), want.Data()) {
+ err = fmt.Errorf("got chunk %v data %x, want %x", i, got.Data(), want.Data())
+ }
+ if !bytes.Equal(got.Address(), want.Address()) {
+ err = fmt.Errorf("got chunk %v address %s, want %s", i, got.Address().Hex(), want.Address().Hex())
+ }
+ i++
+ // send one and only one error per received chunk
+ errChan <- err
+ case <-ctx.Done():
+ return
+ }
+ }
+ }()
+
+ // upload some chunks just after subscribe
+ uploadRandomChunks(5)
+
+ time.Sleep(500 * time.Millisecond)
+
+ // upload some chunks after some short time
+ uploadRandomChunks(3)
+
+ totalChunks := len(chunks)
+ for i := 0; i < totalChunks; i++ {
+ select {
+ case err := <-errChan:
+ if err != nil {
+ t.Error(err)
+ }
+ case <-ctx.Done():
+ t.Error(ctx.Err())
+ }
+ }
+}