diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go
index f04da404d0..6266a3959f 100644
--- a/swarm/storage/localstore/localstore.go
+++ b/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()
}
diff --git a/swarm/storage/localstore/mode_put.go b/swarm/storage/localstore/mode_put.go
index 71d62adf3a..d1a0d54d6b 100644
--- a/swarm/storage/localstore/mode_put.go
+++ b/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
}
diff --git a/swarm/storage/localstore/mode_set.go b/swarm/storage/localstore/mode_set.go
index 20a15c4098..a522f4447c 100644
--- a/swarm/storage/localstore/mode_set.go
+++ b/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
}
diff --git a/swarm/storage/localstore/subscription_pull.go b/swarm/storage/localstore/subscription_pull.go
new file mode 100644
index 0000000000..e795cf9168
--- /dev/null
+++ b/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 .
+
+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:
+ }
+ }
+}
diff --git a/swarm/storage/localstore/subscriptions_test.go b/swarm/storage/localstore/subscription_pull_test.go
similarity index 62%
rename from swarm/storage/localstore/subscriptions_test.go
rename to swarm/storage/localstore/subscription_pull_test.go
index d358656499..0affa3a218 100644
--- a/swarm/storage/localstore/subscriptions_test.go
+++ b/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())
}
}
}
diff --git a/swarm/storage/localstore/subscription_push.go b/swarm/storage/localstore/subscription_push.go
new file mode 100644
index 0000000000..b13f293998
--- /dev/null
+++ b/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 .
+
+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:
+ }
+ }
+}
diff --git a/swarm/storage/localstore/subscription_push_test.go b/swarm/storage/localstore/subscription_push_test.go
new file mode 100644
index 0000000000..184aa0f65d
--- /dev/null
+++ b/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 .
+
+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())
+ }
+ }
+}
diff --git a/swarm/storage/localstore/subscriptions.go b/swarm/storage/localstore/subscriptions.go
deleted file mode 100644
index 3e21cced58..0000000000
--- a/swarm/storage/localstore/subscriptions.go
+++ /dev/null
@@ -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 .
-
-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:
- }
- }
- }
-}