swarm/storage/localstore: implement feed subscriptions

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

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
 

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
 }

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
 }

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 <http://www.gnu.org/licenses/>.
+
+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:
+			}
+		}
+	}
+}

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 <http://www.gnu.org/licenses/>.
+
+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())
+		}
+	}
+}