roysc/plugeth
1
0
Fork 0
forked from cerc-io/plugeth
Code Pull Requests Activity

cmd/swarm, swarm: LocalStore storage integration

Browse Source
This commit is contained in:
Janoš Guljaš
2019-05-10 12:26:26 +02:00
committed by Anton Evangelatov
parent c94d582aa7
commit 996755c4a8
62 changed files with 3093 additions and 3675 deletions
Download Patch File Download Diff File Expand all files Collapse all files
swarm/network/stream/common_test.go
+55 -38
View File
@@ -30,16 +30,19 @@ import (
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
mockmem "github.com/ethereum/go-ethereum/swarm/storage/mock/mem"
"github.com/ethereum/go-ethereum/swarm/storage/localstore"
"github.com/ethereum/go-ethereum/swarm/storage/mock"
"github.com/ethereum/go-ethereum/swarm/testutil"
colorable "github.com/mattn/go-colorable"
)
@@ -51,7 +54,6 @@ var (
useMockStore = flag.Bool("mockstore", false, "disabled mock store (default: enabled)")
longrunning = flag.Bool("longrunning", false, "do run long-running tests")
bucketKeyDB = simulation.BucketKey("db")
bucketKeyStore = simulation.BucketKey("store")
bucketKeyFileStore = simulation.BucketKey("filestore")
bucketKeyNetStore = simulation.BucketKey("netstore")
@@ -113,16 +115,15 @@ func newNetStoreAndDeliveryWithRequestFunc(ctx *adapters.ServiceContext, bucket
func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map, addr *network.BzzAddr) (*storage.NetStore, *Delivery, func(), error) {
n := ctx.Config.Node()
store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
if *useMockStore {
store, datadir, err = createMockStore(mockmem.NewGlobalStore(), n.ID(), addr)
}
localStore, localStoreCleanup, err := newTestLocalStore(n.ID(), addr, nil)
if err != nil {
return nil, nil, nil, err
}
localStore := store.(*storage.LocalStore)
netStore, err := storage.NewNetStore(localStore, nil)
if err != nil {
localStore.Close()
localStoreCleanup()
return nil, nil, nil, err
}
@@ -131,8 +132,7 @@ func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map,
kad := network.NewKademlia(addr.Over(), network.NewKadParams())
delivery := NewDelivery(kad, netStore)
bucket.Store(bucketKeyStore, store)
bucket.Store(bucketKeyDB, netStore)
bucket.Store(bucketKeyStore, localStore)
bucket.Store(bucketKeyDelivery, delivery)
bucket.Store(bucketKeyFileStore, fileStore)
// for the kademlia object, we use the global key from the simulation package,
@@ -141,13 +141,13 @@ func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map,
cleanup := func() {
netStore.Close()
os.RemoveAll(datadir)
localStoreCleanup()
}
return netStore, delivery, cleanup, nil
}
func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTester, *Registry, *localstore.DB, func(), error) {
// setup
addr := network.RandomAddr() // tested peers peer address
to := network.NewKademlia(addr.OAddr, network.NewKadParams())
@@ -161,11 +161,7 @@ func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTeste
os.RemoveAll(datadir)
}
params := storage.NewDefaultLocalStoreParams()
params.Init(datadir)
params.BaseKey = addr.Over()
localStore, err := storage.NewTestLocalStoreForAddr(params)
localStore, err := localstore.New(datadir, addr.Over(), nil)
if err != nil {
removeDataDir()
return nil, nil, nil, nil, err
@@ -173,15 +169,19 @@ func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTeste
netStore, err := storage.NewNetStore(localStore, nil)
if err != nil {
localStore.Close()
removeDataDir()
return nil, nil, nil, nil, err
}
delivery := NewDelivery(to, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
streamer := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), registryOptions, nil)
intervalsStore := state.NewInmemoryStore()
streamer := NewRegistry(addr.ID(), delivery, netStore, intervalsStore, registryOptions, nil)
teardown := func() {
streamer.Close()
intervalsStore.Close()
netStore.Close()
removeDataDir()
}
prvkey, err := crypto.GenerateKey()
@@ -228,24 +228,37 @@ func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore {
}
// not used in this context, only to fulfill ChunkStore interface
func (rrs *roundRobinStore) Has(ctx context.Context, addr storage.Address) bool {
panic("RoundRobinStor doesn't support HasChunk")
func (rrs *roundRobinStore) Has(_ context.Context, _ storage.Address) (bool, error) {
return false, errors.New("roundRobinStore doesn't support Has")
}
func (rrs *roundRobinStore) Get(ctx context.Context, addr storage.Address) (storage.Chunk, error) {
return nil, errors.New("get not well defined on round robin store")
func (rrs *roundRobinStore) Get(_ context.Context, _ chunk.ModeGet, _ storage.Address) (storage.Chunk, error) {
return nil, errors.New("roundRobinStore doesn't support Get")
}
func (rrs *roundRobinStore) Put(ctx context.Context, chunk storage.Chunk) error {
func (rrs *roundRobinStore) Put(ctx context.Context, mode chunk.ModePut, ch storage.Chunk) (bool, error) {
i := atomic.AddUint32(&rrs.index, 1)
idx := int(i) % len(rrs.stores)
return rrs.stores[idx].Put(ctx, chunk)
return rrs.stores[idx].Put(ctx, mode, ch)
}
func (rrs *roundRobinStore) Close() {
func (rrs *roundRobinStore) Set(ctx context.Context, mode chunk.ModeSet, addr chunk.Address) (err error) {
return errors.New("roundRobinStore doesn't support Set")
}
func (rrs *roundRobinStore) LastPullSubscriptionBinID(bin uint8) (id uint64, err error) {
return 0, errors.New("roundRobinStore doesn't support LastPullSubscriptionBinID")
}
func (rrs *roundRobinStore) SubscribePull(ctx context.Context, bin uint8, since, until uint64) (c <-chan chunk.Descriptor, stop func()) {
return nil, nil
}
func (rrs *roundRobinStore) Close() error {
for _, store := range rrs.stores {
store.Close()
}
return nil
}
func readAll(fileStore *storage.FileStore, hash []byte) (int64, error) {
@@ -311,24 +324,28 @@ func generateRandomFile() (string, error) {
return string(b), nil
}
//create a local store for the given node
func createTestLocalStorageForID(id enode.ID, addr *network.BzzAddr) (storage.ChunkStore, string, error) {
var datadir string
var err error
datadir, err = ioutil.TempDir("", fmt.Sprintf("syncer-test-%s", id.TerminalString()))
func newTestLocalStore(id enode.ID, addr *network.BzzAddr, globalStore mock.GlobalStorer) (localStore *localstore.DB, cleanup func(), err error) {
dir, err := ioutil.TempDir("", "swarm-stream-")
if err != nil {
return nil, "", err
return nil, nil, err
}
var store storage.ChunkStore
params := storage.NewDefaultLocalStoreParams()
params.ChunkDbPath = datadir
params.BaseKey = addr.Over()
store, err = storage.NewTestLocalStoreForAddr(params)
cleanup = func() {
os.RemoveAll(dir)
}
var mockStore *mock.NodeStore
if globalStore != nil {
mockStore = globalStore.NewNodeStore(common.BytesToAddress(id.Bytes()))
}
localStore, err = localstore.New(dir, addr.Over(), &localstore.Options{
MockStore: mockStore,
})
if err != nil {
os.RemoveAll(datadir)
return nil, "", err
cleanup()
return nil, nil, err
}
return store, datadir, nil
return localStore, cleanup, nil
}
// watchDisconnections receives simulation peer events in a new goroutine and sets atomic value
swarm/network/stream/delivery.go
+41 -19
View File
@@ -23,6 +23,7 @@ import (
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/spancontext"
@@ -47,12 +48,12 @@ var (
)
type Delivery struct {
chunkStore storage.SyncChunkStore
chunkStore chunk.FetchStore
kad *network.Kademlia
getPeer func(enode.ID) *Peer
}
func NewDelivery(kad *network.Kademlia, chunkStore storage.SyncChunkStore) *Delivery {
func NewDelivery(kad *network.Kademlia, chunkStore chunk.FetchStore) *Delivery {
return &Delivery{
chunkStore: chunkStore,
kad: kad,
@@ -122,13 +123,13 @@ func (s *SwarmChunkServer) Close() {
close(s.quit)
}
// GetData retrives chunk data from db store
// GetData retrieves chunk data from db store
func (s *SwarmChunkServer) GetData(ctx context.Context, key []byte) ([]byte, error) {
chunk, err := s.chunkStore.Get(ctx, storage.Address(key))
ch, err := s.chunkStore.Get(ctx, chunk.ModeGetRequest, storage.Address(key))
if err != nil {
return nil, err
}
return chunk.Data(), nil
return ch.Data(), nil
}
// RetrieveRequestMsg is the protocol msg for chunk retrieve requests
@@ -171,7 +172,7 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
go func() {
defer osp.Finish()
chunk, err := d.chunkStore.Get(ctx, req.Addr)
ch, err := d.chunkStore.Get(ctx, chunk.ModeGetRequest, req.Addr)
if err != nil {
retrieveChunkFail.Inc(1)
log.Debug("ChunkStore.Get can not retrieve chunk", "peer", sp.ID().String(), "addr", req.Addr, "hopcount", req.HopCount, "err", err)
@@ -181,7 +182,7 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
syncing := false
osp.LogFields(olog.Bool("skipCheck", true))
err = sp.Deliver(ctx, chunk, s.priority, syncing)
err = sp.Deliver(ctx, ch, s.priority, syncing)
if err != nil {
log.Warn("ERROR in handleRetrieveRequestMsg", "err", err)
}
@@ -190,7 +191,7 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
}
osp.LogFields(olog.Bool("skipCheck", false))
select {
case streamer.deliveryC <- chunk.Address()[:]:
case streamer.deliveryC <- ch.Address()[:]:
case <-streamer.quit:
}
@@ -216,7 +217,7 @@ type ChunkDeliveryMsgRetrieval ChunkDeliveryMsg
type ChunkDeliveryMsgSyncing ChunkDeliveryMsg
// chunk delivery msg is response to retrieverequest msg
func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *ChunkDeliveryMsg) error {
func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req interface{}) error {
var osp opentracing.Span
ctx, osp = spancontext.StartSpan(
ctx,
@@ -224,11 +225,32 @@ func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *Ch
processReceivedChunksCount.Inc(1)
// retrieve the span for the originating retrieverequest
spanId := fmt.Sprintf("stream.send.request.%v.%v", sp.ID(), req.Addr)
span := tracing.ShiftSpanByKey(spanId)
var msg *ChunkDeliveryMsg
var mode chunk.ModePut
switch r := req.(type) {
case *ChunkDeliveryMsgRetrieval:
msg = (*ChunkDeliveryMsg)(r)
peerPO := chunk.Proximity(sp.ID().Bytes(), msg.Addr)
po := chunk.Proximity(d.kad.BaseAddr(), msg.Addr)
depth := d.kad.NeighbourhoodDepth()
// chunks within the area of responsibility should always sync
// https://github.com/ethersphere/go-ethereum/pull/1282#discussion_r269406125
if po >= depth || peerPO < po {
mode = chunk.ModePutSync
} else {
// do not sync if peer that is sending us a chunk is closer to the chunk then we are
mode = chunk.ModePutRequest
}
case *ChunkDeliveryMsgSyncing:
msg = (*ChunkDeliveryMsg)(r)
mode = chunk.ModePutSync
}
log.Trace("handle.chunk.delivery", "ref", req.Addr, "from peer", sp.ID())
// retrieve the span for the originating retrieverequest
spanID := fmt.Sprintf("stream.send.request.%v.%v", sp.ID(), msg.Addr)
span := tracing.ShiftSpanByKey(spanID)
log.Trace("handle.chunk.delivery", "ref", msg.Addr, "from peer", sp.ID())
go func() {
defer osp.Finish()
@@ -238,18 +260,18 @@ func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *Ch
defer span.Finish()
}
req.peer = sp
log.Trace("handle.chunk.delivery", "put", req.Addr)
err := d.chunkStore.Put(ctx, storage.NewChunk(req.Addr, req.SData))
msg.peer = sp
log.Trace("handle.chunk.delivery", "put", msg.Addr)
_, err := d.chunkStore.Put(ctx, mode, storage.NewChunk(msg.Addr, msg.SData))
if err != nil {
if err == storage.ErrChunkInvalid {
// we removed this log because it spams the logs
// TODO: Enable this log line
// log.Warn("invalid chunk delivered", "peer", sp.ID(), "chunk", req.Addr, )
req.peer.Drop(err)
// log.Warn("invalid chunk delivered", "peer", sp.ID(), "chunk", msg.Addr, )
msg.peer.Drop(err)
}
}
log.Trace("handle.chunk.delivery", "done put", req.Addr, "err", err)
log.Trace("handle.chunk.delivery", "done put", msg.Addr, "err", err)
}()
return nil
}
swarm/network/stream/delivery_test.go
+8 -7
View File
@@ -31,6 +31,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network"
pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue"
@@ -189,8 +190,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
})
hash := storage.Address(hash0[:])
chunk := storage.NewChunk(hash, hash)
err = localStore.Put(context.TODO(), chunk)
ch := storage.NewChunk(hash, hash)
_, err = localStore.Put(context.TODO(), chunk.ModePutUpload, ch)
if err != nil {
t.Fatalf("Expected no err got %v", err)
}
@@ -241,8 +242,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
}
hash = storage.Address(hash1[:])
chunk = storage.NewChunk(hash, hash1[:])
err = localStore.Put(context.TODO(), chunk)
ch = storage.NewChunk(hash, hash1[:])
_, err = localStore.Put(context.TODO(), chunk.ModePutUpload, ch)
if err != nil {
t.Fatalf("Expected no err got %v", err)
}
@@ -420,14 +421,14 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
defer cancel()
// wait for the chunk to get stored
storedChunk, err := localStore.Get(ctx, chunkKey)
storedChunk, err := localStore.Get(ctx, chunk.ModeGetRequest, chunkKey)
for err != nil {
select {
case <-ctx.Done():
t.Fatalf("Chunk is not in localstore after timeout, err: %v", err)
default:
}
storedChunk, err = localStore.Get(ctx, chunkKey)
storedChunk, err = localStore.Get(ctx, chunk.ModeGetRequest, chunkKey)
time.Sleep(50 * time.Millisecond)
}
@@ -700,7 +701,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
errs := make(chan error)
for _, hash := range hashes {
go func(h storage.Address) {
_, err := netStore.Get(ctx, h)
_, err := netStore.Get(ctx, chunk.ModeGetRequest, h)
log.Warn("test check netstore get", "hash", h, "err", err)
errs <- err
}(hash)
swarm/network/stream/intervals_test.go
+3 -2
View File
@@ -29,6 +29,7 @@ import (
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
@@ -287,11 +288,11 @@ func enableNotifications(r *Registry, peerID enode.ID, s Stream) error {
type testExternalClient struct {
hashes chan []byte
store storage.SyncChunkStore
store chunk.FetchStore
enableNotificationsC chan struct{}
}
func newTestExternalClient(store storage.SyncChunkStore) *testExternalClient {
func newTestExternalClient(store chunk.FetchStore) *testExternalClient {
return &testExternalClient{
hashes: make(chan []byte),
store: store,
swarm/network/stream/snapshot_retrieval_test.go
+3 -2
View File
@@ -25,6 +25,7 @@ import (
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/state"
@@ -278,8 +279,8 @@ func runRetrievalTest(t *testing.T, chunkCount int, nodeCount int) error {
if !ok {
return fmt.Errorf("No localstore")
}
lstore := item.(*storage.LocalStore)
conf.hashes, err = uploadFileToSingleNodeStore(node.ID(), chunkCount, lstore)
store := item.(chunk.Store)
conf.hashes, err = uploadFileToSingleNodeStore(node.ID(), chunkCount, store)
if err != nil {
return err
}
swarm/network/stream/snapshot_sync_test.go
+15 -14
View File
@@ -31,6 +31,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/pot"
@@ -190,10 +191,10 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
node := sim.Net.GetRandomUpNode()
item, ok := sim.NodeItem(node.ID(), bucketKeyStore)
if !ok {
return fmt.Errorf("No localstore")
return errors.New("no store in simulation bucket")
}
lstore := item.(*storage.LocalStore)
hashes, err := uploadFileToSingleNodeStore(node.ID(), chunkCount, lstore)
store := item.(chunk.Store)
hashes, err := uploadFileToSingleNodeStore(node.ID(), chunkCount, store)
if err != nil {
return err
}
@@ -221,25 +222,25 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
localChunks := conf.idToChunksMap[id]
for _, ch := range localChunks {
//get the real chunk by the index in the index array
chunk := conf.hashes[ch]
log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
ch := conf.hashes[ch]
log.Trace("node has chunk", "address", ch)
//check if the expected chunk is indeed in the localstore
var err error
if *useMockStore {
//use the globalStore if the mockStore should be used; in that case,
//the complete localStore stack is bypassed for getting the chunk
_, err = globalStore.Get(common.BytesToAddress(id.Bytes()), chunk)
_, err = globalStore.Get(common.BytesToAddress(id.Bytes()), ch)
} else {
//use the actual localstore
item, ok := sim.NodeItem(id, bucketKeyStore)
if !ok {
return fmt.Errorf("Error accessing localstore")
return errors.New("no store in simulation bucket")
}
lstore := item.(*storage.LocalStore)
_, err = lstore.Get(ctx, chunk)
store := item.(chunk.Store)
_, err = store.Get(ctx, chunk.ModeGetLookup, ch)
}
if err != nil {
log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
log.Debug("chunk not found", "address", ch.Hex(), "node", id)
// Do not get crazy with logging the warn message
time.Sleep(500 * time.Millisecond)
continue REPEAT
@@ -247,10 +248,10 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
evt := &simulations.Event{
Type: EventTypeChunkArrived,
Node: sim.Net.GetNode(id),
Data: chunk.String(),
Data: ch.String(),
}
sim.Net.Events().Send(evt)
log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id))
log.Trace("chunk found", "address", ch.Hex(), "node", id)
}
}
return nil
@@ -296,9 +297,9 @@ func mapKeysToNodes(conf *synctestConfig) {
}
//upload a file(chunks) to a single local node store
func uploadFileToSingleNodeStore(id enode.ID, chunkCount int, lstore *storage.LocalStore) ([]storage.Address, error) {
func uploadFileToSingleNodeStore(id enode.ID, chunkCount int, store chunk.Store) ([]storage.Address, error) {
log.Debug(fmt.Sprintf("Uploading to node id: %s", id))
fileStore := storage.NewFileStore(lstore, storage.NewFileStoreParams())
fileStore := storage.NewFileStore(store, storage.NewFileStoreParams())
size := chunkSize
var rootAddrs []storage.Address
for i := 0; i < chunkCount; i++ {
swarm/network/stream/stream.go
+4 -9
View File
@@ -30,11 +30,11 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
)
const (
@@ -108,7 +108,7 @@ type RegistryOptions struct {
}
// NewRegistry is Streamer constructor
func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.SyncChunkStore, intervalsStore state.Store, options *RegistryOptions, balance protocols.Balance) *Registry {
func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore chunk.FetchStore, intervalsStore state.Store, options *RegistryOptions, balance protocols.Balance) *Registry {
if options == nil {
options = &RegistryOptions{}
}
@@ -627,13 +627,8 @@ func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error {
case *WantedHashesMsg:
return p.handleWantedHashesMsg(ctx, msg)
case *ChunkDeliveryMsgRetrieval:
// handling chunk delivery is the same for retrieval and syncing, so let's cast the msg
return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, ((*ChunkDeliveryMsg)(msg)))
case *ChunkDeliveryMsgSyncing:
// handling chunk delivery is the same for retrieval and syncing, so let's cast the msg
return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, ((*ChunkDeliveryMsg)(msg)))
case *ChunkDeliveryMsgRetrieval, *ChunkDeliveryMsgSyncing:
return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, msg)
case *RetrieveRequestMsg:
return p.streamer.delivery.handleRetrieveRequestMsg(ctx, p, msg)
swarm/network/stream/syncer.go
+75 -49
View File
@@ -21,8 +21,7 @@ import (
"strconv"
"time"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@@ -36,12 +35,12 @@ const (
// * (live/non-live historical) chunk syncing per proximity bin
type SwarmSyncerServer struct {
po uint8
store storage.SyncChunkStore
store chunk.FetchStore
quit chan struct{}
}
// NewSwarmSyncerServer is constructor for SwarmSyncerServer
func NewSwarmSyncerServer(po uint8, syncChunkStore storage.SyncChunkStore) (*SwarmSyncerServer, error) {
func NewSwarmSyncerServer(po uint8, syncChunkStore chunk.FetchStore) (*SwarmSyncerServer, error) {
return &SwarmSyncerServer{
po: po,
store: syncChunkStore,
@@ -49,7 +48,7 @@ func NewSwarmSyncerServer(po uint8, syncChunkStore storage.SyncChunkStore) (*Swa
}, nil
}
func RegisterSwarmSyncerServer(streamer *Registry, syncChunkStore storage.SyncChunkStore) {
func RegisterSwarmSyncerServer(streamer *Registry, syncChunkStore chunk.FetchStore) {
streamer.RegisterServerFunc("SYNC", func(_ *Peer, t string, _ bool) (Server, error) {
po, err := ParseSyncBinKey(t)
if err != nil {
@@ -69,76 +68,103 @@ func (s *SwarmSyncerServer) Close() {
// GetData retrieves the actual chunk from netstore
func (s *SwarmSyncerServer) GetData(ctx context.Context, key []byte) ([]byte, error) {
chunk, err := s.store.Get(ctx, storage.Address(key))
ch, err := s.store.Get(ctx, chunk.ModeGetSync, storage.Address(key))
if err != nil {
return nil, err
}
return chunk.Data(), nil
return ch.Data(), nil
}
// SessionIndex returns current storage bin (po) index.
func (s *SwarmSyncerServer) SessionIndex() (uint64, error) {
return s.store.BinIndex(s.po), nil
return s.store.LastPullSubscriptionBinID(s.po)
}
// GetBatch retrieves the next batch of hashes from the dbstore
// SetNextBatch retrieves the next batch of hashes from the localstore.
// It expects a range of bin IDs, both ends inclusive in syncing, and returns
// concatenated byte slice of chunk addresses and bin IDs of the first and
// the last one in that slice. The batch may have up to BatchSize number of
// chunk addresses. If at least one chunk is added to the batch and no new chunks
// are added in batchTimeout period, the batch will be returned. This function
// will block until new chunks are received from localstore pull subscription.
func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
var batch []byte
i := 0
descriptors, stop := s.store.SubscribePull(context.Background(), s.po, from, to)
defer stop()
var ticker *time.Ticker
const batchTimeout = 2 * time.Second
var (
batch []byte
batchSize int
batchStartID *uint64
batchEndID uint64
timer *time.Timer
timerC <-chan time.Time
)
defer func() {
if ticker != nil {
ticker.Stop()
if timer != nil {
timer.Stop()
}
}()
var wait bool
for {
if wait {
if ticker == nil {
ticker = time.NewTicker(1000 * time.Millisecond)
}
select {
case <-ticker.C:
case <-s.quit:
return nil, 0, 0, nil, nil
}
}
metrics.GetOrRegisterCounter("syncer.setnextbatch.iterator", nil).Inc(1)
err := s.store.Iterator(from, to, s.po, func(key storage.Address, idx uint64) bool {
select {
case <-s.quit:
return false
default:
for iterate := true; iterate; {
select {
case d, ok := <-descriptors:
if !ok {
iterate = false
break
}
batch = append(batch, key[:]...)
i++
to = idx
return i < BatchSize
})
if err != nil {
return nil, 0, 0, nil, err
batch = append(batch, d.Address[:]...)
// This is the most naive approach to label the chunk as synced
// allowing it to be garbage collected. A proper way requires
// validating that the chunk is successfully stored by the peer.
err := s.store.Set(context.Background(), chunk.ModeSetSync, d.Address)
if err != nil {
return nil, 0, 0, nil, err
}
batchSize++
if batchStartID == nil {
// set batch start id only if
// this is the first iteration
batchStartID = &d.BinID
}
batchEndID = d.BinID
if batchSize >= BatchSize {
iterate = false
}
if timer == nil {
timer = time.NewTimer(batchTimeout)
} else {
if !timer.Stop() {
<-timer.C
}
timer.Reset(batchTimeout)
}
timerC = timer.C
case <-timerC:
// return batch if new chunks are not
// received after some time
iterate = false
case <-s.quit:
iterate = false
}
if len(batch) > 0 {
break
}
wait = true
}
log.Trace("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.store.BinIndex(s.po))
return batch, from, to, nil, nil
if batchStartID == nil {
// if batch start id is not set, return 0
batchStartID = new(uint64)
}
return batch, *batchStartID, batchEndID, nil, nil
}
// SwarmSyncerClient
type SwarmSyncerClient struct {
store storage.SyncChunkStore
store chunk.FetchStore
peer *Peer
stream Stream
}
// NewSwarmSyncerClient is a contructor for provable data exchange syncer
func NewSwarmSyncerClient(p *Peer, store storage.SyncChunkStore, stream Stream) (*SwarmSyncerClient, error) {
func NewSwarmSyncerClient(p *Peer, store chunk.FetchStore, stream Stream) (*SwarmSyncerClient, error) {
return &SwarmSyncerClient{
store: store,
peer: p,
@@ -184,7 +210,7 @@ func NewSwarmSyncerClient(p *Peer, store storage.SyncChunkStore, stream Stream)
// RegisterSwarmSyncerClient registers the client constructor function for
// to handle incoming sync streams
func RegisterSwarmSyncerClient(streamer *Registry, store storage.SyncChunkStore) {
func RegisterSwarmSyncerClient(streamer *Registry, store chunk.FetchStore) {
streamer.RegisterClientFunc("SYNC", func(p *Peer, t string, live bool) (Client, error) {
return NewSwarmSyncerClient(p, store, NewStream("SYNC", t, live))
})
swarm/network/stream/syncer_test.go
+28 -33
View File
@@ -21,22 +21,20 @@ import (
"errors"
"fmt"
"io/ioutil"
"math"
"os"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/storage/mock"
"github.com/ethereum/go-ethereum/swarm/testutil"
)
@@ -55,24 +53,6 @@ func TestSyncerSimulation(t *testing.T) {
}
}
func createMockStore(globalStore mock.GlobalStorer, id enode.ID, addr *network.BzzAddr) (lstore storage.ChunkStore, datadir string, err error) {
address := common.BytesToAddress(id.Bytes())
mockStore := globalStore.NewNodeStore(address)
params := storage.NewDefaultLocalStoreParams()
datadir, err = ioutil.TempDir("", "localMockStore-"+id.TerminalString())
if err != nil {
return nil, "", err
}
params.Init(datadir)
params.BaseKey = addr.Over()
lstore, err = storage.NewLocalStore(params, mockStore)
if err != nil {
return nil, "", err
}
return lstore, datadir, nil
}
func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, po uint8) {
sim := simulation.New(map[string]simulation.ServiceFunc{
@@ -181,17 +161,32 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
if i < nodes-1 {
hashCounts[i] = hashCounts[i+1]
}
item, ok := sim.NodeItem(nodeIDs[i], bucketKeyDB)
item, ok := sim.NodeItem(nodeIDs[i], bucketKeyStore)
if !ok {
return fmt.Errorf("No DB")
}
netStore := item.(*storage.NetStore)
netStore.Iterator(0, math.MaxUint64, po, func(addr storage.Address, index uint64) bool {
hashes[i] = append(hashes[i], addr)
totalHashes++
hashCounts[i]++
return true
})
store := item.(chunk.Store)
until, err := store.LastPullSubscriptionBinID(po)
if err != nil {
return err
}
if until > 0 {
c, _ := store.SubscribePull(ctx, po, 0, until)
for iterate := true; iterate; {
select {
case cd, ok := <-c:
if !ok {
iterate = false
break
}
hashes[i] = append(hashes[i], cd.Address)
totalHashes++
hashCounts[i]++
case <-ctx.Done():
return ctx.Err()
}
}
}
}
var total, found int
for _, node := range nodeIDs {
@@ -200,12 +195,12 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
for j := i; j < nodes; j++ {
total += len(hashes[j])
for _, key := range hashes[j] {
item, ok := sim.NodeItem(nodeIDs[j], bucketKeyDB)
item, ok := sim.NodeItem(nodeIDs[j], bucketKeyStore)
if !ok {
return fmt.Errorf("No DB")
}
db := item.(*storage.NetStore)
_, err := db.Get(ctx, key)
db := item.(chunk.Store)
_, err := db.Get(ctx, chunk.ModeGetRequest, key)
if err == nil {
found++
}
@@ -216,7 +211,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
if total == found && total > 0 {
return nil
}
return fmt.Errorf("Total not equallying found: total is %d", total)
return fmt.Errorf("Total not equallying found %v: total is %d", found, total)
})
if result.Error != nil {