forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/storage/localstore: subscription of addresses instead chunks

Browse source
This commit is contained in:
Janos Guljas 2018-12-21 15:08:33 +01:00
parent 534009fa49
commit 5edd22d075
3 changed files with 76 additions and 98 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
swarm/storage/localstore/localstore.go
View file

@ -255,7 +255,7 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
return nil, err
}
// create a pull syncing feed used by SubscribePull function
db.pullFeed = newFeed(db.pullIndex, db.retrievalDataIndex)
db.pullFeed = newFeed(db.pullIndex)
// 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) {
@ -280,7 +280,7 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
return nil, err
}
// create a push syncing feed used by SubscribePush function
db.pushFeed = newFeed(db.pushIndex, db.retrievalDataIndex)
db.pushFeed = newFeed(db.pushIndex)
// 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) {

44
swarm/storage/localstore/subscriptions.go
View file

@ -40,16 +40,16 @@ 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
// 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 Chunks and waiting for new ones. It provides methods
// getting Addresses 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
// 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
@ -92,17 +92,15 @@ func (s *Subscription) Stop() {
})
}
// feed is a collection of Chunks subscriptions of order given
// by sort index and Chunk data provided by data index.
// 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 Chunks will be
// index on which keys the order of Addresses 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
@ -114,12 +112,10 @@ type feed struct {
}
// 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) {
// Sort index provides ordering of Addresses.
func newFeed(sortIndex shed.Index) (f *feed) {
return &feed{
sortIndex: sortIndex,
dataIndex: dataIndex,
subscriptions: make([]*Subscription, 0),
closeChan: make(chan struct{}),
}
@ -135,9 +131,9 @@ func (f *feed) subscribe(ctx context.Context, prefix []byte) (s *Subscription, e
return nil, ErrSubscriptionFeedClosed
default:
}
chunks := make(chan storage.Chunk)
addrs := make(chan storage.Address)
s = &Subscription{
Chunks: chunks,
Addrs: addrs,
prefix: prefix,
stopChan: make(chan struct{}),
doneChan: make(chan struct{}),
@ -169,14 +165,8 @@ func (f *feed) subscribe(ctx context.Context, prefix []byte) (s *Subscription, e
// - 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):
case addrs <- storage.Address(item.Address):
// set next iteration start item
// when its chunk is successfully sent to channel
startFrom = &item
@ -189,7 +179,7 @@ func (f *feed) subscribe(ctx context.Context, prefix []byte) (s *Subscription, e
}
}, &shed.IterateOptions{
StartFrom: startFrom,
// startFrom was sent as the last Chunk in the previous
// startFrom was sent as the last Address in the previous
// iterator call, skip it in this one
SkipStartFromItem: true,
Prefix: prefix,
@ -245,7 +235,7 @@ func (f *feed) close() {
})
}
// trigger signals all subscriptions with tprovided prefix
// 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.

126
swarm/storage/localstore/subscriptions_test.go
View file

@ -28,7 +28,7 @@ import (
// TestSubscribePush uploads some chunks before and after
// push syncing subscription is created and validates if
// all chunks are received in the right order.
// all addresses are received in the right order.
func TestSubscribePush(t *testing.T) {
t.Parallel()
@ -37,7 +37,7 @@ func TestSubscribePush(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
chunks := make([]storage.Chunk, 0)
addrs := make([]storage.Address, 0)
uploadRandomChunks := func(count int) {
for i := 0; i < count; i++ {
@ -48,7 +48,7 @@ func TestSubscribePush(t *testing.T) {
t.Fatal(err)
}
chunks = append(chunks, chunk)
addrs = append(addrs, chunk.Address())
}
}
@ -60,8 +60,8 @@ func TestSubscribePush(t *testing.T) {
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
// 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)
@ -70,22 +70,19 @@ func TestSubscribePush(t *testing.T) {
}
defer sub.Stop()
// receive and validate chunks from the subscription
// receive and validate addresses from the subscription
go func() {
var i int // chunk index
var i int // address index
for {
select {
case got := <-sub.Chunks:
want := chunks[i]
case got := <-sub.Addrs:
want := addrs[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())
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 chunk
// send one and only one error per received address
errChan <- err
case <-ctx.Done():
return
@ -101,7 +98,7 @@ func TestSubscribePush(t *testing.T) {
// upload some chunks after some short time
uploadRandomChunks(3)
totalChunks := len(chunks)
totalChunks := len(addrs)
for i := 0; i < totalChunks; i++ {
select {
case err := <-errChan:
@ -116,7 +113,7 @@ func TestSubscribePush(t *testing.T) {
// TestSubscribePush_multiple uploads chunks before and after
// multiple push syncing subscriptions are created and
// validates if all chunks are received in the right order.
// validates if all addresses are received in the right order.
func TestSubscribePush_multiple(t *testing.T) {
t.Parallel()
@ -125,7 +122,7 @@ func TestSubscribePush_multiple(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
chunks := make([]storage.Chunk, 0)
addrs := make([]storage.Address, 0)
uploadRandomChunks := func(count int) {
for i := 0; i < count; i++ {
@ -136,7 +133,7 @@ func TestSubscribePush_multiple(t *testing.T) {
t.Fatal(err)
}
chunks = append(chunks, chunk)
addrs = append(addrs, chunk.Address())
}
}
@ -148,14 +145,14 @@ func TestSubscribePush_multiple(t *testing.T) {
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
// 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 chunk error to errChan
// 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 {
@ -163,22 +160,19 @@ func TestSubscribePush_multiple(t *testing.T) {
}
defer sub.Stop()
// receive and validate chunks from the subscription
// receive and validate addresses from the subscription
go func(j int) {
var i int // chunk index
var i int // address index
for {
select {
case got := <-sub.Chunks:
want := chunks[i]
case got := <-sub.Addrs:
want := addrs[i]
var err error
if !bytes.Equal(got.Data(), want.Data()) {
err = fmt.Errorf("got chunk %v on subscription %v data %x, want %x", i, j, got.Data(), want.Data())
}
if !bytes.Equal(got.Address(), want.Address()) {
err = fmt.Errorf("got chunk %v on subscription %v address %s, want %s", i, j, got.Address().Hex(), want.Address().Hex())
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 chunk
// send one and only one error per received address
errChan <- err
case <-ctx.Done():
return
@ -195,8 +189,8 @@ func TestSubscribePush_multiple(t *testing.T) {
// upload some chunks after some short time
uploadRandomChunks(3)
// number of chunks received by all subscriptions
totalChunks := len(chunks) * subsCount
// number of addresses received by all subscriptions
totalChunks := len(addrs) * subsCount
for i := 0; i < totalChunks; i++ {
select {
case err := <-errChan:
@ -211,7 +205,7 @@ func TestSubscribePush_multiple(t *testing.T) {
// TestSubscribePull uploads some chunks before and after
// pull syncing subscription is created and validates if
// all chunks are received in the right order
// all addresses are received in the right order
// for expected proximity order bins.
func TestSubscribePull(t *testing.T) {
t.Parallel()
@ -221,7 +215,7 @@ func TestSubscribePull(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
chunks := make(map[uint8][]storage.Chunk)
addrs := make(map[uint8][]storage.Address)
var uploadedChunksCount int
uploadRandomChunks := func(count int) {
@ -234,11 +228,11 @@ func TestSubscribePull(t *testing.T) {
}
bin := db.po(chunk.Address())
if _, ok := chunks[bin]; !ok {
chunks[bin] = make([]storage.Chunk, 0)
if _, ok := addrs[bin]; !ok {
addrs[bin] = make([]storage.Address, 0)
}
chunks[bin] = append(chunks[bin], chunk)
addrs[bin] = append(addrs[bin], chunk.Address())
uploadedChunksCount++
}
}
@ -251,8 +245,8 @@ func TestSubscribePull(t *testing.T) {
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
// 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++ {
@ -262,22 +256,19 @@ func TestSubscribePull(t *testing.T) {
}
defer sub.Stop()
// receive and validate chunks from the subscription
// receive and validate addresses from the subscription
go func(bin uint8) {
var i int // chunk index
var i int // address index
for {
select {
case got := <-sub.Chunks:
want := chunks[bin][i]
case got := <-sub.Addrs:
want := addrs[bin][i]
var err error
if !bytes.Equal(got.Data(), want.Data()) {
err = fmt.Errorf("got chunk %v in bin %v data %x, want %x", i, bin, got.Data(), want.Data())
}
if !bytes.Equal(got.Address(), want.Address()) {
err = fmt.Errorf("got chunk %v in bin %v address %s, want %s", i, bin, got.Address().Hex(), want.Address().Hex())
if !bytes.Equal(got, want) {
err = fmt.Errorf("got chunk address %v in bin %v %s, want %s", i, bin, got, want)
}
i++
// send one and only one error per received chunk
// send one and only one error per received address
errChan <- err
case <-ctx.Done():
return
@ -308,7 +299,7 @@ func TestSubscribePull(t *testing.T) {
// TestSubscribePull_multiple uploads chunks before and after
// multiple pull syncing subscriptions are created and
// validates if all chunks are received in the right order
// validates if all addresses are received in the right order
// for expected proximity order bins.
func TestSubscribePull_multiple(t *testing.T) {
t.Parallel()
@ -318,7 +309,7 @@ func TestSubscribePull_multiple(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
chunks := make(map[uint8][]storage.Chunk)
addrs := make(map[uint8][]storage.Address)
var uploadedChunksCount int
uploadRandomChunks := func(count int) {
@ -331,11 +322,11 @@ func TestSubscribePull_multiple(t *testing.T) {
}
bin := db.po(chunk.Address())
if _, ok := chunks[bin]; !ok {
chunks[bin] = make([]storage.Chunk, 0)
if _, ok := addrs[bin]; !ok {
addrs[bin] = make([]storage.Address, 0)
}
chunks[bin] = append(chunks[bin], chunk)
addrs[bin] = append(addrs[bin], chunk.Address())
uploadedChunksCount++
}
}
@ -348,14 +339,14 @@ func TestSubscribePull_multiple(t *testing.T) {
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
// 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 chunk error to errChan
// 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)
@ -364,22 +355,19 @@ func TestSubscribePull_multiple(t *testing.T) {
}
defer sub.Stop()
// receive and validate chunks from the subscription
// receive and validate addresses from the subscription
go func(bin uint8, j int) {
var i int // chunk index
var i int // address index
for {
select {
case got := <-sub.Chunks:
want := chunks[bin][i]
case got := <-sub.Addrs:
want := addrs[bin][i]
var err error
if !bytes.Equal(got.Data(), want.Data()) {
err = fmt.Errorf("got chunk %v in bin %v on subscription %v data %x, want %x", i, bin, j, got.Data(), want.Data())
}
if !bytes.Equal(got.Address(), want.Address()) {
err = fmt.Errorf("got chunk %v in bin %v on subscription %v address %s, want %s", i, bin, j, got.Address().Hex(), want.Address().Hex())
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)
}
i++
// send one and only one error per received chunk
// send one and only one error per received address
errChan <- err
case <-ctx.Done():
return