From 365bd3b6d2bd417c676bcbe6c9b82d133ae2ea4f Mon Sep 17 00:00:00 2001 From: Balint Gabor Date: Tue, 16 Jan 2018 18:37:32 +0100 Subject: [PATCH] swarm/network: Draft test for syncer --- swarm/network/bitvector/bitvector.go | 4 +- swarm/network/request_test.go | 31 ++-- swarm/network/streamer.go | 8 +- swarm/network/syncer.go | 48 +++--- swarm/network/syncer_test.go | 220 +++++++++++++++++++++++++++ swarm/swarm.go | 4 +- 6 files changed, 275 insertions(+), 40 deletions(-) create mode 100644 swarm/network/syncer_test.go diff --git a/swarm/network/bitvector/bitvector.go b/swarm/network/bitvector/bitvector.go index 93e55a9d09..256c9fd5f3 100644 --- a/swarm/network/bitvector/bitvector.go +++ b/swarm/network/bitvector/bitvector.go @@ -1,6 +1,8 @@ package bitvector -import "errors" +import ( + "errors" +) var errInvalidLength = errors.New("invalid length") diff --git a/swarm/network/request_test.go b/swarm/network/request_test.go index 36f6db7ed3..cd014a4509 100644 --- a/swarm/network/request_test.go +++ b/swarm/network/request_test.go @@ -305,12 +305,9 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { } -// serviceName is used with the exec adapter so the exec'd binary knows which -// service to execute -const serviceName = "delivery" - var services = adapters.Services{ - serviceName: newDeliveryService, + "delivery": newDeliveryService, + "syncer": newSyncerService, } var ( @@ -498,6 +495,7 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter 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 @@ -549,7 +547,7 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter } } - result, err := runSimulation(nodes, conns, action, trigger, check, adapter) + result, err := runSimulation(nodes, conns, "delivery", action, trigger, check, adapter) if err != nil { return nil, fmt.Errorf("Setting up simulation failed: %v", err) } @@ -560,7 +558,7 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter } } -func runSimulation(nodes, conns int, 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) { +func runSimulation(nodes, conns int, serviceName string, 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) { // create network net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{ ID: "0", @@ -654,18 +652,21 @@ func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) { nodeCount++ log.Warn("new service created") - return &testDeliveryService{ + self := &testStreamerService{ addr: addr, streamer: streamer, - }, nil + } + self.run = self.runDelivery + return self, nil } -type testDeliveryService struct { +type testStreamerService struct { addr *BzzAddr streamer *Streamer + run func(p *p2p.Peer, rw p2p.MsgReadWriter) error } -func (tds *testDeliveryService) Protocols() []p2p.Protocol { +func (tds *testStreamerService) Protocols() []p2p.Protocol { log.Warn("Protocols function", "run", tds.run) return []p2p.Protocol{ { @@ -679,19 +680,19 @@ func (tds *testDeliveryService) Protocols() []p2p.Protocol { } } -func (b *testDeliveryService) APIs() []rpc.API { +func (b *testStreamerService) APIs() []rpc.API { return []rpc.API{} } -func (b *testDeliveryService) Start(server *p2p.Server) error { +func (b *testStreamerService) Start(server *p2p.Server) error { return nil } -func (b *testDeliveryService) Stop() error { +func (b *testStreamerService) Stop() error { return nil } -func (b *testDeliveryService) run(p *p2p.Peer, rw p2p.MsgReadWriter) error { +func (b *testStreamerService) runDelivery(p *p2p.Peer, rw p2p.MsgReadWriter) error { bzzPeer := &bzzPeer{ Peer: protocols.NewPeer(p, rw, StreamerSpec), localAddr: b.addr, diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index ea64734fd1..3c367f7b6f 100644 --- a/swarm/network/streamer.go +++ b/swarm/network/streamer.go @@ -22,6 +22,7 @@ import ( "fmt" "sync" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/protocols" bv "github.com/ethereum/go-ethereum/swarm/network/bitvector" @@ -264,7 +265,7 @@ func (self *StreamerPeer) getOutgoingStreamer(s string) (*outgoingStreamer, erro defer self.outgoingLock.RUnlock() streamer := self.outgoing[s] if streamer == nil { - return nil, fmt.Errorf("stream '%v' not provided", s) + return nil, fmt.Errorf("outgoing stream '%v' not provided to peer %v", s, self.ID()) } return streamer, nil } @@ -274,7 +275,7 @@ func (self *StreamerPeer) getIncomingStreamer(s string) (*incomingStreamer, erro defer self.incomingLock.RUnlock() streamer := self.incoming[s] if streamer == nil { - return nil, fmt.Errorf("stream '%v' not provided", s) + return nil, fmt.Errorf("incoming stream '%v' not provided to peer %v", s, self.ID()) } return streamer, nil } @@ -348,6 +349,7 @@ func (self *incomingStreamer) nextBatch(from uint64) (nextFrom uint64, nextTo ui // Subscribe initiates the streamer func (self *Streamer) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error { + log.Warn("!!!!!! Subscribe ", "peer", peerId) f, err := self.GetIncomingStreamer(s) if err != nil { return err @@ -362,7 +364,7 @@ func (self *Streamer) Subscribe(peerId discover.NodeID, s string, t []byte, from if err != nil { return err } - err = peer.setIncomingStreamer(s, is, priority, live) + err = peer.setIncomingStreamer(s+string(t), is, priority, live) if err != nil { return err } diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index c9fe1d3555..db2e551826 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -21,6 +21,8 @@ import ( "errors" "fmt" "io" + "math" + "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" @@ -58,6 +60,7 @@ func (self *DbAccess) iterator(from uint64, to uint64, po uint8, f func(storage. // to obtain the chunks from key or request db entry only func (self *DbAccess) getOrCreateRequest(key storage.Key) (*storage.Chunk, bool) { + log.Warn("getOrCreateRequest", "self", self) return self.loc.GetOrCreateRequest(key) } @@ -95,9 +98,11 @@ func NewOutgoingSwarmSyncer(live bool, po uint8, db *DbAccess) (*OutgoingSwarmSy const maxPO = 32 -func RegisterOutgoingSyncers(streamer *Streamer, db *DbAccess) { +func RegisterOutgoingSyncer(streamer *Streamer, db *DbAccess) { streamer.RegisterOutgoingStreamer("SYNC", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { syncType, po := parseSyncLabel(t) + // TODO: make this work for HISTORY too + syncType = "LIVE" switch syncType { case "LIVE": return NewOutgoingSwarmSyncer(true, po, db) @@ -128,17 +133,29 @@ func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64, if from == 0 { from = self.start } - err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool { - batch = append(batch, key[:]...) - i++ - to = idx - return i < batchSize - }) - if err != nil { - return nil, 0, 0, nil, err + if to <= from { + to = math.MaxUint64 } + log.Warn("!!!!!!!!!!!!! setNextBatch", "from", from, "to", to, "currentStoreCount", self.db.currentBucketStorageIndex(1)) + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for range ticker.C { + err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool { + batch = append(batch, key[:]...) + i++ + to = idx + return i < batchSize + }) + if err != nil { + return nil, 0, 0, nil, err + } + if len(batch) > 0 { + break + } + } + log.Debug("Swarm batch", "po", self.po, "len", i, "from", from, "to", to) - return batch, from, to, nil, nil + return batch, from, to + 1, nil, nil } // IncomingSwarmSyncer @@ -212,16 +229,9 @@ func StartSyncing(s *Streamer, peerId discover.NodeID, po uint8, nn bool) { } } -func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) { +func RegisterIncomingSyncer(streamer *Streamer, db *DbAccess) { streamer.RegisterIncomingStreamer("SYNC", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { - syncType, _ := parseSyncLabel(t) - switch syncType { - case "LIVE": - return NewIncomingSwarmSyncer(p, nil, nil) - case "HISTORY": - return NewIncomingSwarmSyncer(p, nil, nil) - } - return nil, fmt.Errorf("unknown sync type %q", syncType) + return NewIncomingSwarmSyncer(p, db, nil) }) // stream = fmt.Sprintf("SYNC-%02d-delete", po) // streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { diff --git a/swarm/network/syncer_test.go b/swarm/network/syncer_test.go new file mode 100644 index 0000000000..5962f79583 --- /dev/null +++ b/swarm/network/syncer_test.go @@ -0,0 +1,220 @@ +// Copyright 2016 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package network + +import ( + "context" + crand "crypto/rand" + "fmt" + "io" + "math" + "net" + "testing" + "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/storage" +) + +var nodeAddrById map[discover.NodeID]*BzzAddr + +func TestSyncerSimulation(t *testing.T) { + testSimulation(t, testSyncBetweenNodes(2, 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) { + nodeAddrById = make(map[discover.NodeID]*BzzAddr) + 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]) but simulation needs + // all nodes to pass the check so we trigger each and the check function + // will trivially return true + for i := 1; i < nodes; i++ { + triggerC <- net.Nodes[i].ID() + } + 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(localStores[1:]...), storage.NewChunkerParams()) + rrdpa.Start() + // create a retriever dpa for the pivot node + dpacs := storage.NewDpaChunkStore(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 + _, _, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) + if err != nil { + return err + } + // // wait until all chunks stored + // 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) { + dbAccesses := make([]*DbAccess, nodes) + + for i := 0; i < nodes; i++ { + dbAccesses[i] = NewDbAccess(localStores[i].(*storage.LocalStore)) + } + return func(ctx context.Context, id discover.NodeID) (bool, error) { + var found, total int + dbAccesses[1].iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool { + _, err := dbAccesses[0].get(key) + if err == nil { + found++ + } + total++ + return true + }) + + // + // if id != net.Nodes[0].ID() { + // return true, nil + // } + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + return found == total, nil + // // try to locally retrieve the file to check if retrieve requests have been successful + // log.Warn(fmt.Sprintf("try to locally retrieve %v", fileHash)) + // total, err := mustReadAll(dpa, fileHash) + // if err != nil || total != size { + // log.Warn(fmt.Sprintf("number of bytes read %v/%v (error: %v)", total, size, err)) + // 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, "syncer", 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 newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID + addr := NewAddrFromNodeID(id) + kad := NewKademlia(addr.Over(), NewKadParams()) + localStore := localStores[nodeCount] + dbAccess := NewDbAccess(localStore.(*storage.LocalStore)) + streamer := NewStreamer(NewDelivery(kad, dbAccess)) + log.Warn("!!!!!!!! Registering syncers") + RegisterIncomingSyncer(streamer, dbAccess) + RegisterOutgoingSyncer(streamer, dbAccess) + addrBytes := addr.Address() + if 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 + addrBytes[0] = 0x0 + } else { + addrBytes[0] = 0xF0 + } + addr = &BzzAddr{ + OAddr: addrBytes, + UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()), + } + nodeAddrById[id] = addr + + //else { + // RegisterOutgoingSyncer(streamer, dbAccess) + // } + nodeCount++ + + log.Warn("new service created") + self := &testStreamerService{ + addr: addr, + streamer: streamer, + } + self.run = self.runSyncer + return self, nil +} + +func (b *testStreamerService) runSyncer(p *p2p.Peer, rw p2p.MsgReadWriter) error { + bzzPeer := &bzzPeer{ + Peer: protocols.NewPeer(p, rw, StreamerSpec), + localAddr: b.addr, + BzzAddr: nodeAddrById[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(), "SYNC", []byte{uint8(1)}, 0, 0, Top, true) + if err != nil { + log.Warn("error in subscribe", "err", err) + } + }() + return b.streamer.Run(bzzPeer) +} diff --git a/swarm/swarm.go b/swarm/swarm.go index 11f5cf25f2..b0a1bc04d3 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -128,8 +128,8 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e dbAccess := network.NewDbAccess(self.lstore) self.streamer = network.NewStreamer(to, dbAccess) - network.RegisterOutgoingSyncers(self.streamer, dbAccess) - network.RegisterIncomingSyncers(self.streamer, dbAccess) + network.RegisterOutgoingSyncer(self.streamer, dbAccess) + network.RegisterIncomingSyncer(self.streamer, dbAccess) self.bzz = network.NewBzz(bzzconfig, to, nil, self.streamer)