swarm/storage/localstore: improve subscriptions

swarm/storage/localstore/localstore.go

@@ -65,12 +65,16 @@ type DB struct {
 	retrievalAccessIndex shed.Index
 	// push syncing index
 	pushIndex shed.Index
-	// provides push syncing subscriptions
-	pushFeed *feed
+	// push syncing subscriptions triggers
+	pushTriggers   []chan struct{}
+	pushTriggersMu sync.RWMutex
+
 	// pull syncing index
 	pullIndex shed.Index
-	// provides pull syncing subscriptions
-	pullFeed *feed
+	// pull syncing subscriptions triggers per bin
+	pullTriggers   map[uint8][]chan struct{}
+	pullTriggersMu sync.RWMutex
+
 	// garbage collection index
 	gcIndex shed.Index
 	// index that stores hashes that are not
@@ -254,8 +258,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)
+	// create a pull syncing triggers used by SubscribePull function
+	db.pullTriggers = make(map[uint8][]chan struct{})
 	// 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) {
@@ -279,8 +283,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)
+	// create a push syncing triggers used by SubscribePush function
+	db.pushTriggers = make([]chan struct{}, 0)
 	// 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) {
@@ -360,9 +364,6 @@ 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/mode_put.go

@@ -152,10 +152,10 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
 		db.incGCSize(gcSizeChange)
 	}
 	if triggerPullFeed {
-		db.pullFeed.trigger([]byte{db.po(item.Address)})
+		db.triggerPullSubscriptions(db.po(item.Address))
 	}
 	if triggerPushFeed {
-		db.pushFeed.trigger(nil)
+		db.triggerPushSubscriptions()
 	}
 	return nil
 }

swarm/storage/localstore/mode_set.go

@@ -199,7 +199,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		db.incGCSize(gcSizeChange)
 	}
 	if triggerPullFeed {
-		db.pullFeed.trigger([]byte{db.po(item.Address)})
+		db.triggerPullSubscriptions(db.po(item.Address))
 	}
 	return nil
 }

swarm/storage/localstore/subscription_pull.go

@@ -0,0 +1,185 @@
+// Copyright 2019 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/log"
+	"github.com/ethereum/go-ethereum/swarm/shed"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// SubscribePull returns a channel that provides chunk addresses and stored times from pull syncing index.
+// Pull syncing index can be only subscribed to a particular proximity order bin. If since
+// is not nil, the iteration will start from the first item stored after that timestamp. If until is not nil,
+// only chunks stored up to this timestamp will be send to the channel, and the returned channel will be
+// closed. The since-until interval is open on the left and closed on the right (since,until]. Returned stop
+// function will terminate current and further iterations without errors, and also close the returned channel.
+// Make sure that you check the second returned parameter from the channel to stop iteration when its value
+// is false.
+func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkInfo) (c <-chan ChunkInfo, stop func()) {
+	chunkInfos := make(chan ChunkInfo)
+	trigger := make(chan struct{}, 1)
+
+	db.pullTriggersMu.Lock()
+	if _, ok := db.pullTriggers[bin]; !ok {
+		db.pullTriggers[bin] = make([]chan struct{}, 0)
+	}
+	db.pullTriggers[bin] = append(db.pullTriggers[bin], trigger)
+	db.pullTriggersMu.Unlock()
+
+	// send signal for the initial iteration
+	trigger <- struct{}{}
+
+	stopChan := make(chan struct{})
+	var stopChanOnce sync.Once
+
+	// used to provide information from the iterator to
+	// stop subscription when until chunk info is reached
+	var errStopSubscription = errors.New("stop subscription")
+
+	go func() {
+		// close the returned chunkInfo channel at the end to
+		// signal that the subscription is done
+		defer close(chunkInfos)
+		// sinceItem is the Item from which the next iteration
+		// should start. The first iteration starts from the first Item.
+		var sinceItem *shed.Item
+		if since != nil {
+			sinceItem = &shed.Item{
+				Address:        since.Address,
+				StoreTimestamp: since.StoreTimestamp,
+			}
+		}
+		for {
+			select {
+			case <-trigger:
+				// iterate until:
+				// - last index Item is reached
+				// - subscription stop is called
+				// - context is done
+				err := db.pullIndex.Iterate(func(item shed.Item) (stop bool, err error) {
+					select {
+					case chunkInfos <- ChunkInfo{
+						Address:        item.Address,
+						StoreTimestamp: item.StoreTimestamp,
+					}:
+						// until chunk info is sent
+						// break the iteration
+						if until != nil &&
+							(item.StoreTimestamp >= until.StoreTimestamp ||
+								bytes.Equal(item.Address, until.Address)) {
+							return true, errStopSubscription
+						}
+						// set next iteration start item
+						// when its chunk is successfully sent to channel
+						sinceItem = &item
+						return false, nil
+					case <-stopChan:
+						// gracefully stop the iteration
+						// on stop
+						return true, nil
+					case <-db.close:
+						// gracefully stop the iteration
+						// on database close
+						return true, nil
+					case <-ctx.Done():
+						return true, ctx.Err()
+					}
+				}, &shed.IterateOptions{
+					StartFrom: sinceItem,
+					// sinceItem was sent as the last Address in the previous
+					// iterator call, skip it in this one
+					SkipStartFromItem: true,
+					Prefix:            []byte{bin},
+				})
+				if err != nil {
+					if err == errStopSubscription {
+						// stop subscription without any errors
+						// if until is reached
+						return
+					}
+					log.Error("localstore pull subscription iteration", "err", err)
+					return
+				}
+			case <-stopChan:
+				// terminate the subscription
+				// on stop
+				return
+			case <-db.close:
+				// terminate the subscription
+				// on database close
+				return
+			case <-ctx.Done():
+				err := ctx.Err()
+				if err != nil {
+					log.Error("localstore pull subscription", "err", err)
+				}
+				return
+			}
+		}
+	}()
+
+	stop = func() {
+		stopChanOnce.Do(func() {
+			close(stopChan)
+		})
+
+		db.pullTriggersMu.Lock()
+		defer db.pullTriggersMu.Unlock()
+
+		for i, t := range db.pullTriggers[bin] {
+			if t == trigger {
+				db.pullTriggers[bin] = append(db.pullTriggers[bin][:i], db.pullTriggers[bin][i+1:]...)
+				break
+			}
+		}
+	}
+
+	return chunkInfos, stop
+}
+
+// ChunkInfo holds information required for Pull syncing. This struct
+// is provided by subscribing to pull index.
+type ChunkInfo struct {
+	Address        storage.Address
+	StoreTimestamp int64
+}
+
+// triggerPullSubscriptions is used internally for starting iterations
+// on Pull subscriptions for a particular bin. When new item with address
+// that is in particular bin for DB's baseKey is added to pull index
+// this function should be called.
+func (db *DB) triggerPullSubscriptions(bin uint8) {
+	db.pullTriggersMu.RLock()
+	triggers, ok := db.pullTriggers[bin]
+	db.pullTriggersMu.RUnlock()
+	if !ok {
+		return
+	}
+
+	for _, t := range triggers {
+		select {
+		case t <- struct{}{}:
+		default:
+		}
+	}
+}

swarm/storage/localstore/subscription_pull_test.go

@@ -1,4 +1,4 @@
-// Copyright 2018 The go-ethereum Authors
+// Copyright 2019 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
@@ -20,189 +20,13 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"sync/atomic"
 	"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 addresses are received in the right order.
-func TestSubscribePush(t *testing.T) {
-	t.Parallel()
-
-	db, cleanupFunc := newTestDB(t, nil)
-	defer cleanupFunc()
-
-	uploader := db.NewPutter(ModePutUpload)
-
-	addrs := make([]storage.Address, 0)
-
-	uploadRandomChunks := func(count int) {
-		for i := 0; i < count; i++ {
-			chunk := generateRandomChunk()
-
-			err := uploader.Put(chunk)
-			if err != nil {
-				t.Fatal(err)
-			}
-
-			addrs = append(addrs, chunk.Address())
-		}
-	}
-
-	// 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 addresses, even nil ones
-	// to validate the number of addresses 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 addresses from the subscription
-	go func() {
-		var i int // address index
-		for {
-			select {
-			case got := <-sub.Addrs:
-				want := addrs[i]
-				var err error
-				if !bytes.Equal(got, want) {
-					err = fmt.Errorf("got chunk %v address %s, want %s", i, got, want)
-				}
-				i++
-				// send one and only one error per received address
-				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(addrs)
-	for i := 0; i < totalChunks; i++ {
-		select {
-		case err := <-errChan:
-			if err != nil {
-				t.Error(err)
-			}
-		case <-ctx.Done():
-			t.Error(ctx.Err())
-		}
-	}
-}
-
-// TestSubscribePush_multiple uploads chunks before and after
-// multiple push syncing subscriptions are created and
-// validates if all addresses are received in the right order.
-func TestSubscribePush_multiple(t *testing.T) {
-	t.Parallel()
-
-	db, cleanupFunc := newTestDB(t, nil)
-	defer cleanupFunc()
-
-	uploader := db.NewPutter(ModePutUpload)
-
-	addrs := make([]storage.Address, 0)
-
-	uploadRandomChunks := func(count int) {
-		for i := 0; i < count; i++ {
-			chunk := generateRandomChunk()
-
-			err := uploader.Put(chunk)
-			if err != nil {
-				t.Fatal(err)
-			}
-
-			addrs = append(addrs, chunk.Address())
-		}
-	}
-
-	// 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 addresses, even nil ones
-	// to validate the number of addresses received by the subscription
-	errChan := make(chan error)
-
-	subsCount := 10
-
-	// start a number of subscriptions
-	// that all of them will write every addresses error to errChan
-	for j := 0; j < subsCount; j++ {
-		sub, err := db.SubscribePush(ctx)
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer sub.Stop()
-
-		// receive and validate addresses from the subscription
-		go func(j int) {
-			var i int // address index
-			for {
-				select {
-				case got := <-sub.Addrs:
-					want := addrs[i]
-					var err error
-					if !bytes.Equal(got, want) {
-						err = fmt.Errorf("got chunk %v address on subscription %v %s, want %s", i, j, got, want)
-					}
-					i++
-					// send one and only one error per received address
-					errChan <- err
-				case <-ctx.Done():
-					return
-				}
-			}
-		}(j)
-	}
-
-	// upload some chunks just after subscribe
-	uploadRandomChunks(5)
-
-	time.Sleep(500 * time.Millisecond)
-
-	// upload some chunks after some short time
-	uploadRandomChunks(3)
-
-	// number of addresses received by all subscriptions
-	totalChunks := len(addrs) * subsCount
-	for i := 0; i < totalChunks; i++ {
-		select {
-		case err := <-errChan:
-			if err != nil {
-				t.Error(err)
-			}
-		case <-ctx.Done():
-			t.Error(ctx.Err())
-		}
-	}
-}
-
 // TestSubscribePull uploads some chunks before and after
 // pull syncing subscription is created and validates if
 // all addresses are received in the right order
@@ -216,7 +40,7 @@ func TestSubscribePull(t *testing.T) {
 	uploader := db.NewPutter(ModePutUpload)
 
 	addrs := make(map[uint8][]storage.Address)
-	var uploadedChunksCount int
+	var wantedChunksCount int
 
 	uploadRandomChunks := func(count int) {
 		for i := 0; i < count; i++ {
@@ -233,7 +57,7 @@ func TestSubscribePull(t *testing.T) {
 			}
 
 			addrs[bin] = append(addrs[bin], chunk.Address())
-			uploadedChunksCount++
+			wantedChunksCount++
 		}
 	}
 
@@ -250,22 +74,22 @@ func TestSubscribePull(t *testing.T) {
 	errChan := make(chan error)
 
 	for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
-		sub, err := db.SubscribePull(ctx, bin)
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer sub.Stop()
+		ch, stop := db.SubscribePull(ctx, bin, nil, nil)
+		defer stop()
 
 		// receive and validate addresses from the subscription
 		go func(bin uint8) {
 			var i int // address index
 			for {
 				select {
-				case got := <-sub.Addrs:
+				case got, ok := <-ch:
+					if !ok {
+						return
+					}
 					want := addrs[bin][i]
 					var err error
-					if !bytes.Equal(got, want) {
-						err = fmt.Errorf("got chunk address %v in bin %v %s, want %s", i, bin, got, want)
+					if !bytes.Equal(got.Address, want) {
+						err = fmt.Errorf("got chunk address %v in bin %v %s, want %s", i, bin, got.Address.Hex(), want)
 					}
 					i++
 					// send one and only one error per received address
@@ -285,14 +109,14 @@ func TestSubscribePull(t *testing.T) {
 	// upload some chunks after some short time
 	uploadRandomChunks(3)
 
-	for i := 0; i < uploadedChunksCount; i++ {
+	for i := 0; i < wantedChunksCount; i++ {
 		select {
 		case err := <-errChan:
 			if err != nil {
 				t.Error(err)
 			}
 		case <-ctx.Done():
-			t.Error(ctx.Err())
+			t.Fatal(ctx.Err())
 		}
 	}
 }
@@ -310,7 +134,7 @@ func TestSubscribePull_multiple(t *testing.T) {
 	uploader := db.NewPutter(ModePutUpload)
 
 	addrs := make(map[uint8][]storage.Address)
-	var uploadedChunksCount int
+	var wantedChunksCount int
 
 	uploadRandomChunks := func(count int) {
 		for i := 0; i < count; i++ {
@@ -327,7 +151,7 @@ func TestSubscribePull_multiple(t *testing.T) {
 			}
 
 			addrs[bin] = append(addrs[bin], chunk.Address())
-			uploadedChunksCount++
+			wantedChunksCount++
 		}
 	}
 
@@ -349,22 +173,22 @@ func TestSubscribePull_multiple(t *testing.T) {
 	// that all of them will write every address error to errChan
 	for j := 0; j < subsCount; j++ {
 		for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
-			sub, err := db.SubscribePull(ctx, bin)
-			if err != nil {
-				t.Fatal(err)
-			}
-			defer sub.Stop()
+			ch, stop := db.SubscribePull(ctx, bin, nil, nil)
+			defer stop()
 
 			// receive and validate addresses from the subscription
 			go func(bin uint8, j int) {
 				var i int // address index
 				for {
 					select {
-					case got := <-sub.Addrs:
+					case got, ok := <-ch:
+						if !ok {
+							return
+						}
 						want := addrs[bin][i]
 						var err error
-						if !bytes.Equal(got, want) {
-							err = fmt.Errorf("got chunk address %v in bin %v on subscription %v %s, want %s", i, bin, j, got, want)
+						if !bytes.Equal(got.Address, want) {
+							err = fmt.Errorf("got chunk address %v in bin %v on subscription %v %s, want %s", i, bin, j, got.Address.Hex(), want)
 						}
 						i++
 						// send one and only one error per received address
@@ -385,7 +209,7 @@ func TestSubscribePull_multiple(t *testing.T) {
 	// upload some chunks after some short time
 	uploadRandomChunks(3)
 
-	totalChunks := uploadedChunksCount * subsCount
+	totalChunks := wantedChunksCount * subsCount
 
 	for i := 0; i < totalChunks; i++ {
 		select {
@@ -394,7 +218,221 @@ func TestSubscribePull_multiple(t *testing.T) {
 				t.Error(err)
 			}
 		case <-ctx.Done():
-			t.Error(ctx.Err())
+			t.Fatal(ctx.Err())
+		}
+	}
+}
+
+// TestSubscribePull_since uploads chunks before and after
+// pull syncing subscriptions are created with a since argument
+// and validates if all expected addresses are received in the
+// right order for expected proximity order bins.
+func TestSubscribePull_since(t *testing.T) {
+	db, cleanupFunc := newTestDB(t, nil)
+	defer cleanupFunc()
+
+	uploader := db.NewPutter(ModePutUpload)
+
+	addrs := make(map[uint8][]storage.Address)
+	var wantedChunksCount int
+
+	lastTimestamp := time.Now().UTC().UnixNano()
+	defer setNow(func() (t int64) {
+		return atomic.AddInt64(&lastTimestamp, 1)
+	})()
+
+	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkInfo) {
+		last = make(map[uint8]ChunkInfo)
+		for i := 0; i < count; i++ {
+			chunk := generateRandomChunk()
+
+			err := uploader.Put(chunk)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			bin := db.po(chunk.Address())
+			if _, ok := addrs[bin]; !ok {
+				addrs[bin] = make([]storage.Address, 0)
+			}
+
+			if wanted {
+				addrs[bin] = append(addrs[bin], chunk.Address())
+				wantedChunksCount++
+			}
+
+			last[bin] = ChunkInfo{
+				Address:        chunk.Address(),
+				StoreTimestamp: atomic.LoadInt64(&lastTimestamp),
+			}
+		}
+		return last
+	}
+
+	// prepopulate database with some chunks
+	// before the subscription
+	last := uploadRandomChunks(30, false)
+
+	uploadRandomChunks(25, true)
+
+	// set a timeout on subscription
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	// collect all errors from validating addresses, even nil ones
+	// to validate the number of addresses received by the subscription
+	errChan := make(chan error)
+
+	for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
+		var since *ChunkInfo
+		if c, ok := last[bin]; ok {
+			since = &c
+		}
+		ch, stop := db.SubscribePull(ctx, bin, since, nil)
+		defer stop()
+
+		// receive and validate addresses from the subscription
+		go func(bin uint8) {
+			var i int // address index
+			for {
+				select {
+				case got, ok := <-ch:
+					if !ok {
+						return
+					}
+					want := addrs[bin][i]
+					var err error
+					if !bytes.Equal(got.Address, want) {
+						err = fmt.Errorf("got chunk address %v in bin %v %s, want %s", i, bin, got.Address.Hex(), want)
+					}
+					i++
+					// send one and only one error per received address
+					errChan <- err
+				case <-ctx.Done():
+					return
+				}
+			}
+		}(bin)
+	}
+
+	// upload some chunks just after subscribe
+	uploadRandomChunks(15, true)
+
+	for i := 0; i < wantedChunksCount; i++ {
+		select {
+		case err := <-errChan:
+			if err != nil {
+				t.Error(err)
+			}
+		case <-ctx.Done():
+			t.Fatal(ctx.Err())
+		}
+	}
+}
+
+// TestSubscribePull_until uploads chunks before and after
+// pull syncing subscriptions are created with an until argument
+// and validates if all expected addresses are received in the
+// right order for expected proximity order bins.
+func TestSubscribePull_until(t *testing.T) {
+	db, cleanupFunc := newTestDB(t, nil)
+	defer cleanupFunc()
+
+	uploader := db.NewPutter(ModePutUpload)
+
+	addrs := make(map[uint8][]storage.Address)
+	var wantedChunksCount int
+
+	lastTimestamp := time.Now().UTC().UnixNano()
+	defer setNow(func() (t int64) {
+		return atomic.AddInt64(&lastTimestamp, 1)
+	})()
+
+	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkInfo) {
+		last = make(map[uint8]ChunkInfo)
+		for i := 0; i < count; i++ {
+			chunk := generateRandomChunk()
+
+			err := uploader.Put(chunk)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			bin := db.po(chunk.Address())
+			if _, ok := addrs[bin]; !ok {
+				addrs[bin] = make([]storage.Address, 0)
+			}
+
+			if wanted {
+				addrs[bin] = append(addrs[bin], chunk.Address())
+				wantedChunksCount++
+			}
+
+			last[bin] = ChunkInfo{
+				Address:        chunk.Address(),
+				StoreTimestamp: atomic.LoadInt64(&lastTimestamp),
+			}
+		}
+		return last
+	}
+
+	// prepopulate database with some chunks
+	// before the subscription
+	last := uploadRandomChunks(30, true)
+
+	uploadRandomChunks(25, false)
+
+	// set a timeout on subscription
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	// collect all errors from validating addresses, even nil ones
+	// to validate the number of addresses received by the subscription
+	errChan := make(chan error)
+
+	for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
+		until, ok := last[bin]
+		if !ok {
+			continue
+		}
+		ch, stop := db.SubscribePull(ctx, bin, nil, &until)
+		defer stop()
+
+		// receive and validate addresses from the subscription
+		go func(bin uint8) {
+			var i int // address index
+			for {
+				select {
+				case got, ok := <-ch:
+					if !ok {
+						return
+					}
+					want := addrs[bin][i]
+					var err error
+					if !bytes.Equal(got.Address, want) {
+						err = fmt.Errorf("got chunk address %v in bin %v %s, want %s", i, bin, got.Address.Hex(), want)
+					}
+					i++
+					// send one and only one error per received address
+					errChan <- err
+				case <-ctx.Done():
+					return
+				}
+			}
+		}(bin)
+	}
+
+	// upload some chunks just after subscribe
+	uploadRandomChunks(15, false)
+
+	for i := 0; i < wantedChunksCount; i++ {
+		select {
+		case err := <-errChan:
+			if err != nil {
+				t.Error(err)
+			}
+		case <-ctx.Done():
+			t.Fatal(ctx.Err())
 		}
 	}
 }

swarm/storage/localstore/subscription_push.go

@@ -0,0 +1,145 @@
+// Copyright 2019 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 (
+	"context"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/swarm/shed"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// SubscribePush returns a channel that provides storage chunks with ordering from push syncing index.
+// Returned stop function will terminate current and further iterations, and also it will close
+// the returned channel without any errors. Make sure that you check the second returned parameter
+// from the channel to stop iteration when its value is false.
+func (db *DB) SubscribePush(ctx context.Context) (c <-chan storage.Chunk, stop func()) {
+	chunks := make(chan storage.Chunk)
+	trigger := make(chan struct{}, 1)
+
+	db.pushTriggersMu.Lock()
+	db.pushTriggers = append(db.pushTriggers, trigger)
+	db.pushTriggersMu.Unlock()
+
+	// send signal for the initial iteration
+	trigger <- struct{}{}
+
+	stopChan := make(chan struct{})
+	var stopChanOnce sync.Once
+
+	go func() {
+		// close the returned chunkInfo channel at the end to
+		// signal that the subscription is done
+		defer close(chunks)
+		// sinceItem is the Item from which the next iteration
+		// should start. The first iteration starts from the first Item.
+		var sinceItem *shed.Item
+		for {
+			select {
+			case <-trigger:
+				// iterate until:
+				// - last index Item is reached
+				// - subscription stop is called
+				// - context is done
+				err := db.pushIndex.Iterate(func(item shed.Item) (stop bool, err error) {
+					// get chunk data
+					dataItem, err := db.retrievalDataIndex.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
+						sinceItem = &item
+						return false, nil
+					case <-stopChan:
+						// gracefully stop the iteration
+						// on stop
+						return true, nil
+					case <-db.close:
+						// gracefully stop the iteration
+						// on database close
+						return true, nil
+					case <-ctx.Done():
+						return true, ctx.Err()
+					}
+				}, &shed.IterateOptions{
+					StartFrom: sinceItem,
+					// sinceItem was sent as the last Address in the previous
+					// iterator call, skip it in this one
+					SkipStartFromItem: true,
+				})
+				if err != nil {
+					log.Error("localstore push subscription iteration", "err", err)
+					return
+				}
+			case <-stopChan:
+				// terminate the subscription
+				// on stop
+				return
+			case <-db.close:
+				// terminate the subscription
+				// on database close
+				return
+			case <-ctx.Done():
+				err := ctx.Err()
+				if err != nil {
+					log.Error("localstore push subscription", "err", err)
+				}
+				return
+			}
+		}
+	}()
+
+	stop = func() {
+		stopChanOnce.Do(func() {
+			close(stopChan)
+		})
+
+		db.pushTriggersMu.Lock()
+		defer db.pushTriggersMu.Unlock()
+
+		for i, t := range db.pushTriggers {
+			if t == trigger {
+				db.pushTriggers = append(db.pushTriggers[:i], db.pushTriggers[i+1:]...)
+				break
+			}
+		}
+	}
+
+	return chunks, stop
+}
+
+// triggerPushSubscriptions is used internally for starting iterations
+// on Push subscriptions. Whenever new item is added to the push index,
+// this function should be called.
+func (db *DB) triggerPushSubscriptions() {
+	db.pushTriggersMu.RLock()
+	triggers := db.pushTriggers
+	db.pushTriggersMu.RUnlock()
+
+	for _, t := range triggers {
+		select {
+		case t <- struct{}{}:
+		default:
+		}
+	}
+}

swarm/storage/localstore/subscription_push_test.go

@@ -0,0 +1,207 @@
+// Copyright 2019 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 addresses 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 addresses, even nil ones
+	// to validate the number of addresses received by the subscription
+	errChan := make(chan error)
+
+	ch, stop := db.SubscribePush(ctx)
+	defer stop()
+
+	// receive and validate addresses from the subscription
+	go func() {
+		var i int // address index
+		for {
+			select {
+			case got, ok := <-ch:
+				if !ok {
+					return
+				}
+				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 address
+				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.Fatal(ctx.Err())
+		}
+	}
+}
+
+// TestSubscribePush_multiple uploads chunks before and after
+// multiple push syncing subscriptions are created and
+// validates if all addresses are received in the right order.
+func TestSubscribePush_multiple(t *testing.T) {
+	t.Parallel()
+
+	db, cleanupFunc := newTestDB(t, nil)
+	defer cleanupFunc()
+
+	uploader := db.NewPutter(ModePutUpload)
+
+	addrs := make([]storage.Address, 0)
+
+	uploadRandomChunks := func(count int) {
+		for i := 0; i < count; i++ {
+			chunk := generateRandomChunk()
+
+			err := uploader.Put(chunk)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			addrs = append(addrs, chunk.Address())
+		}
+	}
+
+	// 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 addresses, even nil ones
+	// to validate the number of addresses received by the subscription
+	errChan := make(chan error)
+
+	subsCount := 10
+
+	// start a number of subscriptions
+	// that all of them will write every addresses error to errChan
+	for j := 0; j < subsCount; j++ {
+		ch, stop := db.SubscribePush(ctx)
+		defer stop()
+
+		// receive and validate addresses from the subscription
+		go func(j int) {
+			var i int // address index
+			for {
+				select {
+				case got, ok := <-ch:
+					if !ok {
+						return
+					}
+					want := addrs[i]
+					var err error
+					if !bytes.Equal(got.Address(), want) {
+						err = fmt.Errorf("got chunk %v address on subscription %v %s, want %s", i, j, got, want)
+					}
+					i++
+					// send one and only one error per received address
+					errChan <- err
+				case <-ctx.Done():
+					return
+				}
+			}
+		}(j)
+	}
+
+	// upload some chunks just after subscribe
+	uploadRandomChunks(5)
+
+	time.Sleep(500 * time.Millisecond)
+
+	// upload some chunks after some short time
+	uploadRandomChunks(3)
+
+	// number of addresses received by all subscriptions
+	totalChunks := len(addrs) * subsCount
+	for i := 0; i < totalChunks; i++ {
+		select {
+		case err := <-errChan:
+			if err != nil {
+				t.Error(err)
+			}
+		case <-ctx.Done():
+			t.Fatal(ctx.Err())
+		}
+	}
+}

swarm/storage/localstore/subscriptions.go

@@ -1,251 +0,0 @@
-// 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 Addresses in a particular order
-// through the Addrs channel. That channel will not be closed
-// when the last Address is read, but will block until the new Address
-// is added to database index. Subscription should be used for
-// getting Addresses and waiting for new ones. It provides methods
-// to control and get information about subscription state.
-type Subscription struct {
-	// Addrs is the read-only channel that provides stream of
-	// chunks addresses. This is the subscription main purpose.
-	Addrs <-chan storage.Address
-
-	// 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 Address subscriptions of order given
-// by the sort 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 Addresses will be
-	// provided by subscriptions
-	sortIndex 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 Addresses.
-func newFeed(sortIndex shed.Index) (f *feed) {
-	return &feed{
-		sortIndex:     sortIndex,
-		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:
-	}
-	addrs := make(chan storage.Address)
-	s = &Subscription{
-		Addrs:    addrs,
-		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) {
-					select {
-					case addrs <- storage.Address(item.Address):
-						// 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 Address 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 provided 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:
-			}
-		}
-	}
-}