diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 53776e5ba6..9afa69c3a9 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -253,6 +253,14 @@ type BzzPeer struct { lastActive time.Time // time is updated whenever mutexes are releasing } +func NewBzzTestPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer { + return &BzzPeer{ + Peer: p, + localAddr: addr, + BzzAddr: NewAddrFromNodeID(p.ID()), + } +} + // Off returns the overlay peer record for offline persistance func (p *BzzPeer) Off() OverlayAddr { return p.BzzAddr diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 4b9bc86d8a..f0468be935 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -19,15 +19,14 @@ package stream import ( "errors" "flag" - "io" "io/ioutil" "os" + "sync/atomic" "testing" "time" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" "github.com/ethereum/go-ethereum/swarm/network" @@ -40,8 +39,7 @@ var ( ) var services = adapters.Services{ - "delivery": newDeliveryService, - "syncer": newSyncerService, + "streamer": NewStreamerService, } func init() { @@ -53,24 +51,16 @@ func init() { log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) } -var ( - delivery *Delivery - fileHash storage.Key -) - -func mustReadAll(dpa *storage.DPA, hash storage.Key) (int, error) { - r := dpa.Retrieve(fileHash) - buf := make([]byte, 1024) - var n, total int - var err error - for (total == 0 || n > 0) && err == nil { - n, err = r.ReadAt(buf, int64(total)) - total += n - } - if err != nil && err != io.EOF { - return total, err - } - return total, nil +// newService +func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID + addr := toAddr(id) + kad := network.NewKademlia(addr.Over(), network.NewKadParams()) + store := stores[id] + db := storage.NewDBAPI(store.(*storage.LocalStore)) + delivery := NewDelivery(kad, db) + deliveries[id] = delivery + return NewRegistry(addr, delivery, store), nil } func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) { @@ -94,17 +84,8 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora db := storage.NewDBAPI(localStore) delivery := NewDelivery(to, db) - streamer := NewRegistry(delivery) - run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - bzzPeer := &network.BzzPeer{ - Peer: protocols.NewPeer(p, rw, Spec), - localAddr: addr, - BzzAddr: network.NewAddrFromNodeID(p.ID()), - } - to.On(bzzPeer) - return streamer.Run(bzzPeer) - } - protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, run) + streamer := NewRegistry(addr, delivery, localStore) + protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol) err = waitForPeers(streamer, 1*time.Second) if err != nil { @@ -128,3 +109,30 @@ func waitForPeers(streamer *Registry, timeout time.Duration) error { } } } + +type roundRobinStore struct { + index uint32 + stores []storage.ChunkStore +} + +func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore { + return &roundRobinStore{ + stores: stores, + } +} + +func (rrs *roundRobinStore) Get(key storage.Key) (*storage.Chunk, error) { + return nil, errors.New("get not well defined on round robin store") +} + +func (rrs *roundRobinStore) Put(chunk *storage.Chunk) { + i := atomic.AddUint32(&rrs.index, 1) + idx := int(i) % len(rrs.stores) + rrs.stores[idx].Put(chunk) +} + +func (rrs *roundRobinStore) Close() { + for _, store := range rrs.stores { + store.Close() + } +} diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index f19cbb9a4d..a37651a148 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -26,17 +26,20 @@ import ( "time" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/node" - "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" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" "github.com/ethereum/go-ethereum/swarm/network" + streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing" "github.com/ethereum/go-ethereum/swarm/storage" ) +var ( + deliveries map[discover.NodeID]*Delivery + stores map[discover.NodeID]storage.ChunkStore + toAddr func(discover.NodeID) *network.BzzAddr +) + func TestStreamerRetrieveRequest(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(t) defer teardown() @@ -81,7 +84,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { peer := streamer.getPeer(peerID) peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: retrieveRequestStream, + Stream: swarmChunkServerStreamName, Key: nil, From: 0, To: 0, @@ -132,7 +135,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { peer := streamer.getPeer(peerID) peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: retrieveRequestStream, + Stream: swarmChunkServerStreamName, Key: nil, From: 0, To: 0, @@ -168,7 +171,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { // TODO: why is this 32??? To: 32, Key: []byte{}, - Stream: retrieveRequestStream, + Stream: swarmChunkServerStreamName, }, Peer: peerID, }, @@ -221,8 +224,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterIncomingStreamer("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { - return &testIncomingStreamer{ + streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) { + return &testClient{ t: t, }, nil }) @@ -301,142 +304,127 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { } func TestDeliveryFromNodes(t *testing.T) { - testSimulation(t, testDeliveryFromNodes(2, 1, 8100, true)) - testSimulation(t, testDeliveryFromNodes(2, 1, 8100, false)) - testSimulation(t, testDeliveryFromNodes(3, 1, 8100, true)) - testSimulation(t, testDeliveryFromNodes(3, 1, 8100, false)) + testDeliveryFromNodes(t, 2, 1, 8100, true) + testDeliveryFromNodes(t, 2, 1, 8100, false) + testDeliveryFromNodes(t, 3, 1, 8100, true) + testDeliveryFromNodes(t, 3, 1, 8100, false) } -func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) { - return func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) { - trigger := func(net *simulations.Network) chan discover.NodeID { - triggerC := make(chan discover.NodeID) - ticker := time.NewTicker(500 * time.Millisecond) - go func() { - defer ticker.Stop() - // we are only testing the pivot node (net.Nodes[0]) - for range ticker.C { - triggerC <- net.Nodes[0].ID() - } - }() - return triggerC - } - - 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(testing.LocalStores[1:]...), storage.NewChunkerParams()) - rrdpa.Start() - // create a retriever dpa for the pivot node - 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 { - defer rrdpa.Stop() - // upload an actual random file of size size - hash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) - if err != nil { - return err - } - // wait until all chunks stored - // TODO: is wait() necessary? - wait() - // assign the fileHash to a global so that it is available for the check function - fileHash = hash - go func() { - defer dpa.Stop() - log.Debug(fmt.Sprintf("retrieve %v", fileHash)) - // start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks - // we must wait for the peer connections to have started before requesting - time.Sleep(2 * time.Second) - n, err := mustReadAll(dpa, fileHash) - log.Debug(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err) - }() - return nil - } - } - - check := func(net *simulations.Network, dpa *storage.DPA) func(ctx context.Context, id discover.NodeID) (bool, error) { - return func(ctx context.Context, id discover.NodeID) (bool, error) { - if id != net.Nodes[0].ID() { - return true, nil - } - select { - case <-ctx.Done(): - return false, ctx.Err() - default: - } - // try to locally retrieve the file to check if retrieve requests have been successful - total, err := mustReadAll(dpa, fileHash) - log.Debug(fmt.Sprintf("check if %08x is available locally: number of bytes read %v/%v (error: %v)", fileHash, total, size, err)) - if err != nil || total != size { - return false, nil - } - return true, nil - // node := net.GetNode(id) - // if node == nil { - // return false, fmt.Errorf("unknown node: %s", id) - // } - // client, err := node.Client() - // if err != nil { - // return false, fmt.Errorf("error getting node client: %s", err) - // } - // var response int - // if err := client.Call(&response, "test_haslocal", hash); err != nil { - // return false, fmt.Errorf("error getting bzz_has response: %s", err) - // } - // log.Debug(fmt.Sprintf("node has: %v\n%v", id, response)) - // return response == 0, nil - } - } - - result, err := runSimulation(nodes, conns, "delivery", NewAddrFromNodeID, action, trigger, check, adapter) - if err != nil { - return nil, fmt.Errorf("Setting up simulation failed: %v", err) - } - if result.Error != nil { - return nil, fmt.Errorf("Simulation failed: %s", result.Error) - } - return result, err +func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) { + toAddr = network.NewAddrFromNodeID + conf := &streamTesting.RunConfig{ + Adapter: *adapter, + NodeCount: nodes, + ConnLevel: conns, + ToAddr: toAddr, + Services: services, } -} -// newDeliveryService -func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) { - id := ctx.Config.ID - addr := NewAddrFromNodeID(id) - kad := NewKademlia(addr.Over(), NewKadParams()) - localStore := testing.LocalStores[testing.NodeCount] - db := NewDBAPI(localStore.(*storage.LocalStore)) - streamer := NewStreamerRegistry(NewDelivery(kad, db)) - 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 + sim, teardown, err := streamTesting.NewSimulation(conf) + defer teardown() + if err != nil { + t.Fatal(err.Error()) + } + stores = make(map[discover.NodeID]storage.ChunkStore) + deliveries = make(map[discover.NodeID]*Delivery) + for i, id := range sim.IDs { + stores[id] = sim.Stores[i] } - testing.NodeCount++ - return testing.NewTestStreamerService(Spec, makeRunFunc(addr, streamer)), nil -} -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()), - } - streamer.delivery.overlay.On(bzzPeer) - defer streamer.delivery.overlay.Off(bzzPeer) + // here we distribute chunks of a random file into Stores of nodes 1 to nodes + rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams()) + rrdpa.Start() + fileHash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) + defer rrdpa.Stop() + if err != nil { + t.Fatal(err.Error()) + } + // wait until all chunks stored + // TODO: is wait() necessary? + wait() + // each node Subscribes to each other's swarmChunkServerStreamName + // 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(), swarmChunkServerStreamName, nil, 0, 0, Top, true) + if err != nil { + t.Fatal(err.Error()) + } + // create a retriever dpa for the pivot node + delivery := deliveries[sim.IDs[0]] + dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) }) + dpa := storage.NewDPA(dpacs, storage.NewChunkerParams()) + dpa.Start() + action := func(context.Context) error { + dpa := storage.NewDPA(sim.Stores[0], storage.NewChunkerParams()) + dpa.Start() + // defer dpa.Stop() + 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) - } + defer dpa.Stop() + log.Debug(fmt.Sprintf("retrieve %v", fileHash)) + // start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks + // we must wait for the peer connections to have started before requesting + time.Sleep(2 * time.Second) + n, err := mustReadAll(dpa, fileHash) + log.Debug(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err) }() - return streamer.Run(bzzPeer) + return nil } + + check := func(ctx context.Context, id discover.NodeID) (bool, error) { + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + // try to locally retrieve the file to check if retrieve requests have been successful + node := sim.Net.GetNode(id) + if node == nil { + return false, fmt.Errorf("unknown node: %s", id) + } + client, err := node.Client() + if err != nil { + return false, fmt.Errorf("error getting node client: %s", err) + } + var total int64 + if err := client.Call(&total, "stream_readAll", fileHash); err != nil { + return false, fmt.Errorf("error reading all: %s (read %v)", err, total) + } + // total, err := mustReadAll(dpa, fileHash) + log.Debug(fmt.Sprintf("check if %08x is available locally: number of bytes read %v/%v (error: %v)", fileHash, total, size, err)) + if err != nil || total != int64(size) { + return false, nil + } + return true, nil + } + + trigger := make(chan discover.NodeID) + ticker := time.NewTicker(500 * time.Millisecond) + go func() { + defer ticker.Stop() + // we are only testing the pivot node (net.Nodes[0]) + for range ticker.C { + trigger <- sim.Net.Nodes[0].ID() + } + }() + + conf.Step = &simulations.Step{ + Action: action, + Trigger: trigger, + Expect: &simulations.Expectation{ + Nodes: sim.IDs[0:1], + Check: check, + }, + } + startedAt := time.Now() + result, err := sim.Run(conf) + finishedAt := time.Now() + if err != nil { + t.Fatalf("Setting up simulation failed: %v", err) + } + if result.Error != nil { + t.Fatalf("Simulation failed: %s", result.Error) + } + streamTesting.CheckResult(t, result, startedAt, finishedAt) } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 469ed9f126..7565785bee 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -18,14 +18,17 @@ package stream import ( "fmt" + "io" "math" "sync" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -41,6 +44,7 @@ const ( // Registry registry for outgoing and incoming streamer constructors type Registry struct { + addr *network.BzzAddr clientMu sync.RWMutex serverMu sync.RWMutex peersMu sync.RWMutex @@ -48,11 +52,14 @@ type Registry struct { clientFuncs map[string]func(*Peer, []byte) (Client, error) peers map[discover.NodeID]*Peer delivery *Delivery + store storage.ChunkStore } // NewRegistry is Streamer constructor -func NewRegistry(delivery *Delivery) *Registry { +func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore) *Registry { streamer := &Registry{ + addr: addr, + store: store, serverFuncs: make(map[string]func(*Peer, []byte) (Server, error)), clientFuncs: make(map[string]func(*Peer, []byte) (Client, error)), peers: make(map[discover.NodeID]*Peer), @@ -175,17 +182,22 @@ func (r *Registry) deletePeer(peer *Peer) { } // Run protocol run function -func (r *Registry) Run(p *protocols.Peer) error { +func (r *Registry) run(p *protocols.Peer) error { sp := NewPeer(p, r) - // load saved intervals - r.setPeer(sp) - defer r.deletePeer(sp) defer close(sp.quit) return sp.Run(sp.HandleMsg) } +func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { + peer := protocols.NewPeer(p, rw, Spec) + bzzPeer := network.NewBzzTestPeer(peer, r.addr) + r.delivery.overlay.On(bzzPeer) + defer r.delivery.overlay.Off(bzzPeer) + return r.run(peer) +} + // HandleMsg is the message handler that delegates incoming messages func (p *Peer) HandleMsg(msg interface{}) error { switch msg := msg.(type) { @@ -305,12 +317,65 @@ func (r *Registry) Protocols() []p2p.Protocol { Name: Spec.Name, Version: Spec.Version, Length: Spec.Length(), - Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - peer := protocols.NewPeer(p, rw, Spec) - return r.Run(peer) - }, - NodeInfo: r.NodeInfo, - PeerInfo: r.PeerInfo, + Run: r.runProtocol, + // NodeInfo: , + // PeerInfo: , }, } } + +func (r *Registry) APIs() []rpc.API { + return []rpc.API{ + { + Namespace: "stream", + Version: "0.1", + Service: NewAPI(r, r.store), + Public: true, + }, + } +} + +func (r *Registry) Start(server *p2p.Server) error { + return nil +} + +func (r *Registry) Stop() error { + return nil +} + +type API struct { + streamer *Registry + dpa *storage.DPA +} + +func NewAPI(r *Registry, store storage.ChunkStore) *API { + dpa := storage.NewDPA(store, storage.NewChunkerParams()) + return &API{ + streamer: r, + dpa: dpa, + } +} + +func mustReadAll(dpa *storage.DPA, hash []byte) (int64, error) { + r := dpa.Retrieve(hash) + buf := make([]byte, 1024) + var n int + var total int64 + var err error + for (total == 0 || n > 0) && err == nil { + n, err = r.ReadAt(buf, total) + total += int64(n) + } + if err != nil && err != io.EOF { + return total, err + } + return total, nil +} + +func (api *API) ReadAll(hash []byte) (int64, error) { + return mustReadAll(api.dpa, hash) +} + +func (api *API) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error { + return api.streamer.Subscribe(peerId, s, t, from, to, priority, live) +} diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index 250573ade4..a905f4c963 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -50,15 +50,15 @@ var ( batchDone = make(chan bool) ) -type testIncomingStreamer struct { +type testClient struct { t []byte } -type testOutgoingStreamer struct { +type testServer struct { t []byte } -func (self *testIncomingStreamer) NeedData(hash []byte) func() { +func (self *testClient) NeedData(hash []byte) func() { receivedHashes[string(hash)] = hash if bytes.Equal(hash, hash0[:]) { return func() { @@ -72,16 +72,16 @@ func (self *testIncomingStreamer) NeedData(hash []byte) func() { return nil } -func (self *testIncomingStreamer) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) { +func (self *testClient) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) { close(batchDone) return nil } -func (self *testOutgoingStreamer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { +func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { return make([]byte, HashSize), from + 1, to + 1, nil, nil } -func (self *testOutgoingStreamer) GetData([]byte) []byte { +func (self *testServer) GetData([]byte) []byte { return nil } @@ -92,8 +92,8 @@ func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterClientFunc("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { - return &testIncomingStreamer{ + streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) { + return &testClient{ t: t, }, nil }) @@ -134,8 +134,8 @@ func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterServerFunc("foo", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { - return &testOutgoingStreamer{ + streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) { + return &testServer{ t: t, }, nil }) @@ -188,8 +188,8 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterClientFunc("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { - return &testIncomingStreamer{ + streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) { + return &testClient{ t: t, }, nil }) diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index a6e0ab6b1a..cbe2b380b1 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -26,140 +26,144 @@ import ( "time" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/node" - "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/swarm/network" + streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing" "github.com/ethereum/go-ethereum/swarm/storage" ) func TestSyncerSimulation(t *testing.T) { - testSimulation(t, testSyncBetweenNodes(2, 1, 81000, true, 1)) - testSimulation(t, testSyncBetweenNodes(3, 1, 81000, true, 1)) + testSyncBetweenNodes(t, 2, 1, 81000, true, 1) + testSyncBetweenNodes(t, 3, 1, 81000, true, 1) } -func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) { - return func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) { - trigger := func(net *simulations.Network) chan discover.NodeID { - triggerC := make(chan discover.NodeID) - ticker := time.NewTicker(500 * time.Millisecond) - go func() { - defer ticker.Stop() - // we are only testing the pivot node (net.Nodes[0]) - for range ticker.C { - triggerC <- net.Nodes[0].ID() - } - }() - return triggerC - } - - 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(testing.LocalStores[1:]...), storage.NewChunkerParams()) - rrdpa.Start() - // create a retriever dpa for the pivot node - return func(context.Context) error { - defer rrdpa.Stop() - // upload an actual random file of size size - _, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) - if err != nil { - return err - } - // wait until all chunks stored - wait() - return nil - } - } - - check := func(net *simulations.Network, dpa *storage.DPA) func(ctx context.Context, id discover.NodeID) (bool, error) { - dbs := make([]*storage.DBAPI, nodes) - - for i := 0; i < nodes; i++ { - dbs[i] = NewDbAccess(testing.LocalStores[i].(*storage.LocalStore)) - } - return func(ctx context.Context, id discover.NodeID) (bool, error) { - if id != net.Nodes[0].ID() { - return true, nil - } - select { - case <-ctx.Done(): - return false, ctx.Err() - default: - } - - var found, total int - for i := 1; i < nodes; i++ { - dbs[i].iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool { - _, err := dbs[0].get(key) - if err == nil { - found++ - } - total++ - return true - }) - } - log.Debug("sync check", "bin", po, "found", found, "total", total) - return found == total, nil - } - } - toAddr := func(id discover.NodeID) *BzzAddr { - addr := network.NewAddrFromNodeID(id) - addr.OAddr[0] = byte(0) - return addr - } - - result, err := runSimulation(nodes, conns, "syncer", toAddr, action, trigger, check, adapter) - if err != nil { - return nil, fmt.Errorf("Setting up simulation failed: %v", err) - } - if result.Error != nil { - return nil, fmt.Errorf("Simulation failed: %s", result.Error) - } - return result, err +func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool, po uint8) { + toAddr = func(id discover.NodeID) *network.BzzAddr { + addr := network.NewAddrFromNodeID(id) + addr.OAddr[0] = byte(0) + return addr } -} - -func newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) { - id := ctx.Config.ID - addr := network.NewAddrFromNodeID(id) - // 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 := testing.LocalStores[testing.NodeCount] - db := NewDbAccess(localStore.(*storage.LocalStore)) - streamer := NewRegistry(NewDelivery(kad, db)) - RegisterIncomingSyncer(streamer, db) - RegisterOutgoingSyncer(streamer, db) - - testing.NodeCount++ - return testing.NewTestStreamerService(Spec, makeRunFunc(addr, streamer)), nil -} - -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, - } - 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) + conf := &streamTesting.RunConfig{ + Adapter: *adapter, + NodeCount: nodes, + ConnLevel: conns, + ToAddr: toAddr, + Services: services, } + + sim, teardown, err := streamTesting.NewSimulation(conf) + defer teardown() + if err != nil { + t.Fatal(err.Error()) + } + stores = make(map[discover.NodeID]storage.ChunkStore) + deliveries = make(map[discover.NodeID]*Delivery) + log.Warn("Stores", "len", len(sim.Stores)) + for i, id := range sim.IDs { + stores[id] = sim.Stores[i] + } + + // here we distribute chunks of a random file into Stores of nodes 1 to nodes + rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams()) + rrdpa.Start() + _, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) + defer rrdpa.Stop() + if err != nil { + t.Fatal(err.Error()) + } + // wait until all chunks stored + // TODO: is wait() necessary? + wait() + // each node Subscribes to each other's swarmChunkServerStreamName + // 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(), swarmChunkServerStreamName, nil, 0, 0, Top, true) + if err != nil { + t.Fatal(err.Error()) + } + // create a retriever dpa for the pivot node + action := func(context.Context) error { + for i := 0; i < len(sim.IDs)-1; i++ { + id := sim.IDs[i] + // if err := streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil { + // log.Warn("error in subscribe", "err", err) + // } + node := sim.Net.GetNode(id) + if node == nil { + return fmt.Errorf("unknown node: %s", id) + } + client, err := node.Client() + if err != nil { + return fmt.Errorf("error getting node client: %s", err) + } + var n int64 + sid := sim.IDs[i+1] + if err := client.Call(&n, "stream_subscribe", sid, "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil { + return fmt.Errorf("error subscribing: %s", err) + } + } + return nil + } + + dbs := make([]*storage.DBAPI, nodes) + for i := 0; i < nodes; i++ { + dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore)) + } + + check := func(ctx context.Context, id discover.NodeID) (bool, error) { + if id != sim.Net.Nodes[0].ID() { + return true, nil + } + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + + var found, total int + for i := 1; i < nodes; i++ { + + dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool { + _, err := dbs[0].Get(key) + if err == nil { + found++ + } + total++ + return true + }) + } + log.Debug("sync check", "bin", po, "found", found, "total", total) + return found == total, nil + } + + trigger := make(chan discover.NodeID) + ticker := time.NewTicker(500 * time.Millisecond) + go func() { + defer ticker.Stop() + // we are only testing the pivot node (net.Nodes[0]) + for range ticker.C { + trigger <- sim.Net.Nodes[0].ID() + } + }() + + conf.Step = &simulations.Step{ + Action: action, + Trigger: trigger, + Expect: &simulations.Expectation{ + Nodes: sim.IDs[0:1], + Check: check, + }, + } + startedAt := time.Now() + result, err := sim.Run(conf) + finishedAt := time.Now() + if err != nil { + t.Fatalf("Setting up simulation failed: %v", err) + } + if result.Error != nil { + t.Fatalf("Simulation failed: %s", result.Error) + } + streamTesting.CheckResult(t, result, startedAt, finishedAt) } diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index d3a78ecde2..b427efd532 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -24,84 +24,76 @@ import ( "math/rand" "os" "sync" - "sync/atomic" "testing" "time" "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/storage" ) -var ( - LocalStores []storage.ChunkStore - Addrs []network.Addr - NodeCount int -) +type Simulation struct { + Net *simulations.Network + Stores []storage.ChunkStore + Addrs []network.Addr + IDs []discover.NodeID +} -func setLocalStores(addrs ...network.Addr) (func(), error) { +func SetStores(addrs ...network.Addr) ([]storage.ChunkStore, func(), error) { var datadirs []string - LocalStores = make([]storage.ChunkStore, len(addrs)) + stores := make([]storage.ChunkStore, len(addrs)) var err error for i, addr := range addrs { - // TODO: remove temp datadir after test var datadir string datadir, err = ioutil.TempDir("", "streamer") if err != nil { break } - var localStore *storage.LocalStore - localStore, err = storage.NewTestLocalStoreForAddr(datadir, addr.Over()) + var store storage.ChunkStore + store, err = storage.NewTestLocalStoreForAddr(datadir, addr.Over()) if err != nil { break } datadirs = append(datadirs, datadir) - LocalStores[i] = localStore + stores[i] = store } teardown := func() { for _, datadir := range datadirs { os.RemoveAll(datadir) } } - return teardown, err + return stores, teardown, err } -func testSimulation(t *testing.T, services adapters.Services, adapter string, simf func(adapters.NodeAdapter) (*simulations.StepResult, error)) { - var err error - var result *simulations.StepResult - startedAt := time.Now() - - switch adapter { +func NewAdapter(adapterType string, services adapters.Services) (adapter adapters.NodeAdapter, teardown func(), err error) { + teardown = func() {} + switch adapterType { case "sim": - t.Logf("simadapter") - result, err = simf(adapters.NewSimAdapter(services)) + adapter = adapters.NewSimAdapter(services) case "socket": - result, err = simf(adapters.NewSocketAdapter(services)) + adapter = adapters.NewSocketAdapter(services) case "exec": baseDir, err0 := ioutil.TempDir("", "swarm-test") if err0 != nil { - t.Fatal(err0) + return nil, teardown, err0 } - defer os.RemoveAll(baseDir) - result, err = simf(adapters.NewExecAdapter(baseDir)) + teardown = func() { os.RemoveAll(baseDir) } + adapter = adapters.NewExecAdapter(baseDir) case "docker": - adapter, err0 := adapters.NewDockerAdapter() - if err0 != nil { - t.Fatal(err0) + adapter, err = adapters.NewDockerAdapter() + if err != nil { + return nil, teardown, err } - result, err = simf(adapter) default: - t.Fatal("adapter needs to be one of sim, socket, exec, docker") - } - if err != nil { - t.Fatal(err) + return nil, teardown, errors.New("adapter needs to be one of sim, socket, exec, docker") } + return adapter, teardown, nil +} + +func CheckResult(t *testing.T, result *simulations.StepResult, startedAt, finishedAt time.Time) { t.Logf("Simulation with %d nodes passed in %s", len(result.Passes), result.FinishedAt.Sub(result.StartedAt)) var min, max time.Duration var sum int @@ -116,148 +108,101 @@ func testSimulation(t *testing.T, services adapters.Services, adapter string, si sum += int(duration.Nanoseconds()) } t.Logf("Min: %s, Max: %s, Average: %s", min, max, time.Duration(sum/len(result.Passes))*time.Nanosecond) - finishedAt := time.Now() - t.Logf("Setup: %s, shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt)) + t.Logf("Setup: %s, Shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt)) } -func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.NodeID) *network.BzzAddr, action func(*simulations.Network) func(context.Context) error, trigger func(*simulations.Network) chan discover.NodeID, check func(*simulations.Network, *storage.DPA) func(context.Context, discover.NodeID) (bool, error), adapter adapters.NodeAdapter) (*simulations.StepResult, error) { +type RunConfig struct { + Adapter string + Step *simulations.Step + NodeCount int + ConnLevel int + ToAddr func(discover.NodeID) *network.BzzAddr + Services adapters.Services +} + +func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { // create network + nodes := conf.NodeCount + adapter, adapterTeardown, err := NewAdapter(conf.Adapter, conf.Services) + if err != nil { + return nil, adapterTeardown, err + } net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{ ID: "0", - DefaultService: serviceName, + DefaultService: "streamer", }) - defer net.Shutdown() + teardown := func() { + adapterTeardown() + net.Shutdown() + } ids := make([]discover.NodeID, nodes) - NodeCount = 0 - Addrs = make([]network.Addr, nodes) + addrs := make([]network.Addr, nodes) // start nodes for i := 0; i < nodes; i++ { node, err := net.NewNode() if err != nil { - return nil, fmt.Errorf("error creating node: %s", err) + return nil, teardown, fmt.Errorf("error creating node: %s", err) } ids[i] = node.ID() - Addrs[i] = toAddr(ids[i]) + addrs[i] = conf.ToAddr(ids[i]) + } + // set nodes number of Stores available + stores, storeTeardown, err := SetStores(addrs...) + teardown = func() { + storeTeardown() + adapterTeardown() + net.Shutdown() } - // set nodes number of localstores globally available - teardown, err := setLocalStores(Addrs...) - defer teardown() if err != nil { - return nil, err + return nil, teardown, err } + s := &Simulation{ + Net: net, + Stores: stores, + IDs: ids, + Addrs: addrs, + } + return s, teardown, nil +} +func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) { + // bring up nodes, launch the servive + nodes := conf.NodeCount + conns := conf.ConnLevel for i := 0; i < nodes; i++ { - if err := net.Start(ids[i]); err != nil { - return nil, fmt.Errorf("error starting node %s: %s", ids[i].TerminalString(), err) + if err := s.Net.Start(s.IDs[i]); err != nil { + return nil, fmt.Errorf("error starting node %s: %s", s.IDs[i].TerminalString(), err) } } - // run a simulation which connects the 10 nodes in a chain wg := sync.WaitGroup{} - for i := range ids { + for i := range s.IDs { // collect the overlay addresses, to for j := 0; j < conns; j++ { var k int if j == 0 { k = i - 1 } else { - k = rand.Intn(len(ids)) + k = rand.Intn(len(s.IDs)) } if i > 0 { wg.Add(1) go func(i, k int) { defer wg.Done() - net.Connect(ids[i], ids[k]) + s.Net.Connect(s.IDs[i], s.IDs[k]) }(i, k) } } } wg.Wait() - log.Debug(fmt.Sprintf("nodes: %v", len(Addrs))) + log.Debug(fmt.Sprintf("nodes: %v", len(s.Addrs))) // create an only locally retrieving dpa for the pivot node to test // if retriee requests have arrived - dpa := storage.NewDPA(LocalStores[0], storage.NewChunkerParams()) - dpa.Start() - defer dpa.Stop() timeout := 300 * time.Second ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{ - Action: action(net), - Trigger: trigger(net), - Expect: &simulations.Expectation{ - Nodes: ids[0:1], - Check: check(net, dpa), - }, - }) + result := simulations.NewSimulation(s.Net).Run(ctx, conf.Step) return result, nil } - -type roundRobinStore struct { - index uint32 - stores []storage.ChunkStore -} - -func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore { - return &roundRobinStore{ - stores: stores, - } -} - -func (rrs *roundRobinStore) Get(key storage.Key) (*storage.Chunk, error) { - return nil, errors.New("get not well defined on round robin store") -} - -func (rrs *roundRobinStore) Put(chunk *storage.Chunk) { - i := atomic.AddUint32(&rrs.index, 1) - idx := int(i) % len(rrs.stores) - rrs.stores[idx].Put(chunk) -} - -func (rrs *roundRobinStore) Close() { - for _, store := range rrs.stores { - store.Close() - } -} - -type TestStreamerService struct { - // index int - // addr *network.BzzAddr - // // streamer *stream.Registry - run func(p *p2p.Peer, rw p2p.MsgReadWriter) error - spec *protocols.Spec -} - -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: tds.spec.Name, - Version: tds.spec.Version, - Length: tds.spec.Length(), - Run: tds.run, - // NodeInfo: , - // PeerInfo: , - }, - } -} - -func (b *TestStreamerService) APIs() []rpc.API { - return []rpc.API{} -} - -func (b *TestStreamerService) Start(server *p2p.Server) error { - return nil -} - -func (b *TestStreamerService) Stop() error { - return nil -} diff --git a/swarm/swarm.go b/swarm/swarm.go index 6280d0fce7..b97390e369 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -132,7 +132,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e db := storage.NewDBAPI(self.lstore) delivery := stream.NewDelivery(to, db) - self.streamer = stream.NewRegistry(delivery) + self.streamer = stream.NewRegistry(addr, delivery) stream.RegisterSwarmSyncerServer(self.streamer, db) stream.RegisterSwarmSyncerClient(self.streamer, db) @@ -289,7 +289,7 @@ func (self *Swarm) RegisterPssProtocol(spec *protocols.Spec, targetprotocol *p2p } // implements node.Service -// Apis returns the RPC Api descriptors the Swarm implementation offers +// APIs returns the RPC Api descriptors the Swarm implementation offers func (self *Swarm) APIs() []rpc.API { apis := []rpc.API{