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