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: Better syncing and retrieval option definition

Browse source
This commit is contained in:
Fabio Barone 2018-10-26 07:27:31 -05:00
parent baee850471
commit 433ceb00d8
9 changed files with 120 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

76
swarm/network/stream/delivery_test.go
View file

@ -39,7 +39,11 @@ import (
)
func TestStreamerRetrieveRequest(t *testing.T) {
tester, streamer, _, teardown, err := newStreamerTester(t, nil)
regOpts := &RegistryOptions{
Retrieval: RetrievalClientOnly,
Syncing: SyncingDisabled,
}
tester, streamer, _, teardown, err := newStreamerTester(t, regOpts)
defer teardown()
if err != nil {
t.Fatal(err)
@ -55,9 +59,20 @@ func TestStreamerRetrieveRequest(t *testing.T) {
)
streamer.delivery.RequestFromPeers(ctx, req)
stream := NewStream(swarmChunkServerStreamName, "", true)
err = tester.TestExchanges(p2ptest.Exchange{
Label: "RetrieveRequestMsg",
Expects: []p2ptest.Expect{
{
Code: 4,
Msg: &SubscribeMsg{
Stream: stream,
History: nil,
Priority: Top,
},
Peer: node.ID(),
},
{
Code: 5,
Msg: &RetrieveRequestMsg{
@ -76,7 +91,8 @@ func TestStreamerRetrieveRequest(t *testing.T) {
func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
tester, streamer, _, teardown, err := newStreamerTester(t, &RegistryOptions{
DoServeRetrieve: true,
Retrieval: RetrievalEnabled,
Syncing: SyncingDisabled,
})
defer teardown()
if err != nil {
@ -89,13 +105,26 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
peer := streamer.getPeer(node.ID())
stream := NewStream(swarmChunkServerStreamName, "", true)
peer.handleSubscribeMsg(context.TODO(), &SubscribeMsg{
Stream: NewStream(swarmChunkServerStreamName, "", true),
Stream: stream,
History: nil,
Priority: Top,
})
err = tester.TestExchanges(p2ptest.Exchange{
Expects: []p2ptest.Expect{
{
Code: 4,
Msg: &SubscribeMsg{
Stream: stream,
History: nil,
Priority: Top,
},
Peer: node.ID(),
},
},
}, p2ptest.Exchange{
Label: "RetrieveRequestMsg",
Triggers: []p2ptest.Trigger{
{
@ -120,7 +149,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
},
})
expectedError := `exchange #0 "RetrieveRequestMsg": timed out`
expectedError := `exchange #1 "RetrieveRequestMsg": timed out`
if err == nil || err.Error() != expectedError {
t.Fatalf("Expected error %v, got %v", expectedError, err)
}
@ -130,7 +159,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
// offered hashes or delivery if skipHash is set to true
func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
tester, streamer, localStore, teardown, err := newStreamerTester(t, &RegistryOptions{
DoServeRetrieve: true,
Retrieval: RetrievalEnabled,
Syncing: SyncingDisabled,
})
defer teardown()
if err != nil {
@ -138,6 +168,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
}
node := tester.Nodes[0]
peer := streamer.getPeer(node.ID())
stream := NewStream(swarmChunkServerStreamName, "", true)
@ -156,6 +187,18 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
}
err = tester.TestExchanges(p2ptest.Exchange{
Expects: []p2ptest.Expect{
{
Code: 4,
Msg: &SubscribeMsg{
Stream: stream,
History: nil,
Priority: Top,
},
Peer: node.ID(),
},
},
}, p2ptest.Exchange{
Label: "RetrieveRequestMsg",
Triggers: []p2ptest.Trigger{
{
@ -226,7 +269,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
tester, streamer, localStore, teardown, err := newStreamerTester(t, &RegistryOptions{
DoServeRetrieve: true,
Retrieval: RetrievalDisabled,
Syncing: SyncingDisabled,
})
defer teardown()
if err != nil {
@ -343,7 +387,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
SkipCheck: skipCheck,
DoServeRetrieve: true,
Syncing: SyncingDisabled,
Retrieval: RetrievalEnabled,
})
bucket.Store(bucketKeyRegistry, r)
@ -408,20 +453,6 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
return err
}
//each of the nodes (except pivot node) subscribes to the stream of the next node
for j, node := range nodeIDs[0 : nodes-1] {
sid := nodeIDs[j+1]
item, ok := sim.NodeItem(node, bucketKeyRegistry)
if !ok {
return fmt.Errorf("No registry")
}
registry := item.(*Registry)
err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", true), nil, Top)
if err != nil {
return err
}
}
//get the pivot node's filestore
item, ok := sim.NodeItem(*sim.PivotNodeID(), bucketKeyFileStore)
if !ok {
@ -530,7 +561,8 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
SkipCheck: skipCheck,
DoSync: true,
Syncing: SyncingDisabled,
Retrieval: RetrievalDisabled,
SyncUpdateDelay: 0,
})

2
swarm/network/stream/intervals_test.go
View file

@ -83,6 +83,8 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
Retrieval: RetrievalDisabled,
Syncing: SyncingRegisterOnly,
SkipCheck: skipCheck,
})
bucket.Store(bucketKeyRegistry, r)

12
swarm/network/stream/lightnode_test.go
View file

@ -25,7 +25,8 @@ import (
// when it is serving Retrieve requests.
func TestLigthnodeRetrieveRequestWithRetrieve(t *testing.T) {
registryOptions := &RegistryOptions{
DoServeRetrieve: true,
Retrieval: RetrievalClientOnly,
Syncing: SyncingDisabled,
}
tester, _, _, teardown, err := newStreamerTester(t, registryOptions)
defer teardown()
@ -63,7 +64,8 @@ func TestLigthnodeRetrieveRequestWithRetrieve(t *testing.T) {
// requests are disabled
func TestLigthnodeRetrieveRequestWithoutRetrieve(t *testing.T) {
registryOptions := &RegistryOptions{
DoServeRetrieve: false,
Retrieval: RetrievalDisabled,
Syncing: SyncingDisabled,
}
tester, _, _, teardown, err := newStreamerTester(t, registryOptions)
defer teardown()
@ -106,7 +108,8 @@ func TestLigthnodeRetrieveRequestWithoutRetrieve(t *testing.T) {
// when syncing is enabled.
func TestLigthnodeRequestSubscriptionWithSync(t *testing.T) {
registryOptions := &RegistryOptions{
DoSync: true,
Retrieval: RetrievalDisabled,
Syncing: SyncingRegisterOnly,
}
tester, _, _, teardown, err := newStreamerTester(t, registryOptions)
defer teardown()
@ -150,7 +153,8 @@ func TestLigthnodeRequestSubscriptionWithSync(t *testing.T) {
// when syncing is disabled.
func TestLigthnodeRequestSubscriptionWithoutSync(t *testing.T) {
registryOptions := &RegistryOptions{
DoSync: false,
Retrieval: RetrievalDisabled,
Syncing: SyncingDisabled,
}
tester, _, _, teardown, err := newStreamerTester(t, registryOptions)
defer teardown()

5
swarm/network/stream/snapshot_retrieval_test.go
View file

@ -127,10 +127,9 @@ func retrievalStreamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s no
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
DoSync: true,
Retrieval: RetrievalEnabled,
Syncing: SyncingAutoSubscribe,
SyncUpdateDelay: 3 * time.Second,
DoRetrieve: true,
DoServeRetrieve: true,
})
fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())

8
swarm/network/stream/snapshot_sync_test.go
View file

@ -165,8 +165,8 @@ func streamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Servic
netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
DoSync: true,
DoServeRetrieve: true,
Retrieval: RetrievalDisabled,
Syncing: SyncingAutoSubscribe,
SyncUpdateDelay: 3 * time.Second,
})
@ -360,8 +360,8 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
DoServeRetrieve: true,
DoSync: true,
Retrieval: RetrievalDisabled,
Syncing: SyncingRegisterOnly,
})
bucket.Store(bucketKeyRegistry, r)

38
swarm/network/stream/stream.go
View file

@ -47,6 +47,23 @@ const (
HashSize = 32
)
//Enumerate options for syncing and retrieval
type SyncingOption int
type RetrievalOption int
//Syncing options
const (
SyncingDisabled SyncingOption = iota //Disable Syncing
SyncingRegisterOnly //Register the client but not subscribe
SyncingAutoSubscribe //Subscribe automatically
)
const (
RetrievalDisabled RetrievalOption = iota //Retrieval disabled
RetrievalClientOnly //Clients retrieve only (Light nodes)
RetrievalEnabled //Enabled
)
// Registry registry for outgoing and incoming streamer constructors
type Registry struct {
addr enode.ID
@ -60,16 +77,15 @@ type Registry struct {
peers map[enode.ID]*Peer
delivery *Delivery
intervalsStore state.Store
doRetrieve bool
autoRetrieval bool
maxPeerServers int
}
// RegistryOptions holds optional values for NewRegistry constructor.
type RegistryOptions struct {
SkipCheck bool
DoSync bool // Sets if the server syncs with peers. Default is true, set to false by lightnode or nosync flags.
DoRetrieve bool // Sets if the server issues Retrieve requests. Default is true.
DoServeRetrieve bool // Sets if the server serves Retrieve requests. Default is true, set to false by lightnode flag.
Syncing SyncingOption
Retrieval RetrievalOption
SyncUpdateDelay time.Duration
MaxPeerServers int // The limit of servers for each peer in registry
}
@ -82,6 +98,8 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
if options.SyncUpdateDelay <= 0 {
options.SyncUpdateDelay = 15 * time.Second
}
retrieval := options.Retrieval != RetrievalDisabled
streamer := &Registry{
addr: localID,
skipCheck: options.SkipCheck,
@ -90,13 +108,13 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
peers: make(map[enode.ID]*Peer),
delivery: delivery,
intervalsStore: intervalsStore,
doRetrieve: options.DoRetrieve,
autoRetrieval: retrieval,
maxPeerServers: options.MaxPeerServers,
}
streamer.api = NewAPI(streamer)
delivery.getPeer = streamer.getPeer
if options.DoServeRetrieve {
if options.Retrieval == RetrievalEnabled {
streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, live bool) (Server, error) {
if !live {
return nil, errors.New("only live retrieval requests supported")
@ -105,16 +123,18 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
})
}
if options.Retrieval != RetrievalDisabled {
streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t string, live bool) (Client, error) {
return NewSwarmSyncerClient(p, syncChunkStore, NewStream(swarmChunkServerStreamName, t, live))
})
}
if options.DoSync {
if options.Syncing != SyncingDisabled {
RegisterSwarmSyncerServer(streamer, syncChunkStore)
RegisterSwarmSyncerClient(streamer, syncChunkStore)
}
if options.DoSync {
if options.Syncing == SyncingAutoSubscribe {
// latestIntC function ensures that
// - receiving from the in chan is not blocked by processing inside the for loop
// - the latest int value is delivered to the loop after the processing is done
@ -385,7 +405,7 @@ func (r *Registry) Run(p *network.BzzPeer) error {
defer close(sp.quit)
defer sp.close()
if r.doRetrieve {
if r.autoRetrieval && !p.LightNode {
err := r.Subscribe(p.ID(), NewStream(swarmChunkServerStreamName, "", true), nil, Top)
if err != nil {
return err

2
swarm/network/stream/streamer_test.go
View file

@ -765,6 +765,8 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
func TestMaxPeerServersWithUnsubscribe(t *testing.T) {
var maxPeerServers = 6
tester, streamer, _, teardown, err := newStreamerTester(t, &RegistryOptions{
Retrieval: RetrievalDisabled,
Syncing: SyncingDisabled,
MaxPeerServers: maxPeerServers,
})
defer teardown()

2
swarm/network/stream/syncer_test.go
View file

@ -114,6 +114,8 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
bucket.Store(bucketKeyDelivery, delivery)
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
Retrieval: RetrievalDisabled,
Syncing: SyncingAutoSubscribe,
SkipCheck: skipCheck,
})

20
swarm/swarm.go
View file

@ -175,18 +175,24 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
if err := nodeID.UnmarshalText([]byte(config.NodeID)); err != nil {
return nil, err
}
syncing := stream.SyncingAutoSubscribe
if !config.SyncEnabled || config.LightNodeEnabled {
syncing = stream.SyncingDisabled
}
retrieval := stream.RetrievalEnabled
if config.LightNodeEnabled {
retrieval = stream.RetrievalClientOnly
}
registryOptions := &stream.RegistryOptions{
SkipCheck: config.DeliverySkipCheck,
DoSync: config.SyncEnabled,
DoRetrieve: true,
DoServeRetrieve: true,
Syncing: syncing,
Retrieval: retrieval,
SyncUpdateDelay: config.SyncUpdateDelay,
MaxPeerServers: config.MaxStreamPeerServers,
}
if config.LightNodeEnabled {
registryOptions.DoSync = false
registryOptions.DoRetrieve = false
}
self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, registryOptions)
// Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage