forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 01:13:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm/network, swarm/storage: Further refactor fixes

Browse source
This commit is contained in:
Balint Gabor 2018-01-18 18:47:43 +01:00
parent 35609bec2e
commit f0f62218a3
5 changed files with 72 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
swarm/network/stream/common_test.go
View file

@ -96,13 +96,13 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
delivery := NewDelivery(to, db)
streamer := NewRegistry(delivery)
run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
BzzPeer := &BzzPeer{
bzzPeer := &network.BzzPeer{
Peer: protocols.NewPeer(p, rw, Spec),
localAddr: addr,
BzzAddr: network.NewAddrFromNodeID(p.ID()),
}
to.On(BzzPeer)
return streamer.Run(BzzPeer)
to.On(bzzPeer)
return streamer.Run(bzzPeer)
}
protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, run)

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

@ -33,6 +33,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@ -323,10 +324,10 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter
action := func(net *simulations.Network) func(context.Context) error {
// here we distribute chunks of a random file into localstores of nodes 1 to nodes
rrdpa := storage.NewDPA(newRoundRobinStore(localStores[1:]...), storage.NewChunkerParams())
rrdpa := storage.NewDPA(newRoundRobinStore(testing.LocalStores[1:]...), storage.NewChunkerParams())
rrdpa.Start()
// create a retriever dpa for the pivot node
dpacs := storage.NewNetStore(localStores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
dpacs := storage.NewNetStore(testing.LocalStores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
dpa := storage.NewDPA(dpacs, storage.NewChunkerParams())
dpa.Start()
return func(context.Context) error {
@ -404,41 +405,38 @@ func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) {
id := ctx.Config.ID
addr := NewAddrFromNodeID(id)
kad := NewKademlia(addr.Over(), NewKadParams())
localStore := localStores[nodeCount]
localStore := testing.LocalStores[testing.NodeCount]
db := NewDBAPI(localStore.(*storage.LocalStore))
streamer := NewStreamerRegistry(NewDelivery(kad, db))
if nodeCount == 0 {
if testing.NodeCount == 0 {
// the delivery service for the pivot node is assigned globally
// so that the simulation action call can use it for the
// swarm enabled dpa
delivery = streamer.delivery
}
self := &testStreamerService{
addr: addr,
streamer: streamer,
}
self.run = self.runDelivery
nodeCount++
return self, nil
testing.NodeCount++
return testing.NewTestStreamerService(Spec, makeRunFunc(addr, streamer)), nil
}
func (b *testStreamerService) runDelivery(p *p2p.Peer, rw p2p.MsgReadWriter) error {
BzzPeer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, StreamerSpec),
localAddr: b.addr,
BzzAddr: NewAddrFromNodeID(p.ID()),
}
b.streamer.delivery.overlay.On(BzzPeer)
defer b.streamer.delivery.overlay.Off(BzzPeer)
go func() {
// each node Subscribes to each other's retrieveRequestStream
// need to wait till an aynchronous process registers the peers in streamer.peers
// that is used by Subscribe
time.Sleep(1 * time.Second)
err := b.streamer.Subscribe(p.ID(), retrieveRequestStream, nil, 0, 0, Top, true)
if err != nil {
log.Warn("error in subscribe", "err", err)
func makeRunFunc(addr network.Addr, streamer *Registry) (func(p *p2p.Peer, rw p2p.MsgReadWriter), error) {
return func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
bzzPeer := &network.BzzPeer{
Peer: protocols.NewPeer(p, rw, Spec),
localAddr: addr,
BzzAddr: NewAddrFromNodeID(p.ID()),
}
}()
return b.streamer.Run(BzzPeer)
streamer.delivery.overlay.On(bzzPeer)
defer streamer.delivery.overlay.Off(bzzPeer)
go func() {
// each node Subscribes to each other's retrieveRequestStream
// need to wait till an aynchronous process registers the peers in streamer.peers
// that is used by Subscribe
time.Sleep(1 * time.Second)
err := streamer.Subscribe(p.ID(), retrieveRequestStream, nil, 0, 0, Top, true)
if err != nil {
log.Warn("error in subscribe", "err", err)
}
}()
return streamer.Run(bzzPeer)
}
}

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

@ -58,7 +58,7 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
action := func(net *simulations.Network) func(context.Context) error {
// here we distribute chunks of a random file into localstores of nodes 1 to nodes
rrdpa := storage.NewDPA(newRoundRobinStore(localStores[1:]...), storage.NewChunkerParams())
rrdpa := storage.NewDPA(newRoundRobinStore(testing.LocalStores[1:]...), storage.NewChunkerParams())
rrdpa.Start()
// create a retriever dpa for the pivot node
return func(context.Context) error {
@ -78,7 +78,7 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
dbs := make([]*storage.DBAPI, nodes)
for i := 0; i < nodes; i++ {
dbs[i] = NewDbAccess(localStores[i].(*storage.LocalStore))
dbs[i] = NewDbAccess(testing.LocalStores[i].(*storage.LocalStore))
}
return func(ctx context.Context, id discover.NodeID) (bool, error) {
if id != net.Nodes[0].ID() {
@ -128,42 +128,38 @@ func newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) {
// for the test we make all peers share 8 bits so that syncing full bins make sense
addr.OAddr[0] = byte(0)
kad := NewKademlia(addr.Over(), NewKadParams())
localStore := localStores[nodeCount]
localStore := testing.LocalStores[testing.NodeCount]
db := NewDbAccess(localStore.(*storage.LocalStore))
streamer := NewRegistry(NewDelivery(kad, db))
RegisterIncomingSyncer(streamer, db)
RegisterOutgoingSyncer(streamer, db)
self := &testStreamerService{
index: nodeCount,
addr: addr,
streamer: streamer,
}
self.run = self.runSyncer
nodeCount++
return self, nil
testing.NodeCount++
return testing.NewTestStreamerService(Spec, makeRunFunc(addr, streamer)), nil
}
func (b *testStreamerService) runSyncer(p *p2p.Peer, rw p2p.MsgReadWriter) error {
addr := network.NewAddrFromNodeID(p.ID())
addr.OAddr[0] = byte(0)
BzzPeer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, Spec),
localAddr: b.addr,
BzzAddr: addr,
}
b.streamer.delivery.overlay.On(BzzPeer)
defer b.streamer.delivery.overlay.Off(BzzPeer)
// if len(addr) > b.index+1 && bytes.Equal(addrs[b.index+1], addr) {
go func() {
// each node Subscribes to each other's retrieveRequestStream
// need to wait till an aynchronous process registers the peers in streamer.peers
// that is used by Subscribe
time.Sleep(1 * time.Second)
if err := b.streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
log.Warn("error in subscribe", "err", err)
func makeRunFunc(localAddr network.Addr, streamer *Registry) (func(p *p2p.Peer, rw p2p.MsgReadWriter), error) {
return func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
remoteAddr := network.NewAddrFromNodeID(p.ID())
remoteAddr.OAddr[0] = byte(0)
bzzPeer := &network.BzzPeer{
Peer: protocols.NewPeer(p, rw, Spec),
localAddr: localAddr,
BzzAddr: remoteAddr,
}
}()
// }
return b.streamer.Run(BzzPeer)
streamer.delivery.overlay.On(bzzPeer)
defer streamer.delivery.overlay.Off(bzzPeer)
// if len(addr) > b.index+1 && bytes.Equal(testing.Addrs[b.index+1], addr) {
go func() {
// each node Subscribes to each other's retrieveRequestStream
// need to wait till an aynchronous process registers the peers in streamer.peers
// that is used by Subscribe
time.Sleep(1 * time.Second)
if err := streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
log.Warn("error in subscribe", "err", err)
}
}()
// }
return streamer.Run(bzzPeer)
}
}

25
swarm/network/stream/testing/testing.go
View file

@ -31,11 +31,11 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@ -223,23 +223,26 @@ func (rrs *roundRobinStore) Close() {
}
type TestStreamerService struct {
index int
addr *network.BzzAddr
streamer *stream.Registry
run func(s *TestStreamerService, p *p2p.Peer, rw p2p.MsgReadWriter) error
// index int
// addr *network.BzzAddr
// // streamer *stream.Registry
run func(p *p2p.Peer, rw p2p.MsgReadWriter) error
spec *protocols.Spec
}
func NewTestStreamerService(run func(s *TestStreamerService, p *p2p.Peer, rw p2p.MsgReadWriter) error) TestStreamerService {
t := &TestStreamerService{}
t.run = run
func NewTestStreamerService(spec *protocols.Spec, run func(p *p2p.Peer, rw p2p.MsgReadWriter) error) *TestStreamerService {
return &TestStreamerService{
run: run,
spec: spec,
}
}
func (tds *TestStreamerService) Protocols() []p2p.Protocol {
return []p2p.Protocol{
{
Name: stream.Spec.Name,
Version: stream.Spec.Version,
Length: stream.Spec.Length(),
Name: tds.spec.Name,
Version: tds.spec.Version,
Length: tds.spec.Length(),
Run: tds.run,
// NodeInfo: ,
// PeerInfo: ,

0
swarm/storage/dbaccess.go → swarm/storage/dbapi.go
View file