From 8deb2d1900e479a479893971db7a718db46f5906 Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 16 Jan 2018 12:40:31 +0100 Subject: [PATCH] swarm/network: TestDeliveryFromNodes passes for 2,3 nodes --- swarm/network/request_test.go | 234 ++++++++++++++++++--------------- swarm/network/requests.go | 13 +- swarm/network/streamer.go | 11 +- swarm/network/streamer_test.go | 5 +- 4 files changed, 145 insertions(+), 118 deletions(-) diff --git a/swarm/network/request_test.go b/swarm/network/request_test.go index 7aa892dc13..36f6db7ed3 100644 --- a/swarm/network/request_test.go +++ b/swarm/network/request_test.go @@ -310,7 +310,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { const serviceName = "delivery" var services = adapters.Services{ - serviceName: newService, + serviceName: newDeliveryService, } var ( @@ -405,7 +405,10 @@ func testSimulation(t *testing.T, simf func(adapters.NodeAdapter) (*simulations. } func TestDeliveryFromNodes(t *testing.T) { - testSimulation(t, testDeliveryFromNodes) + 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)) } var ( @@ -459,94 +462,102 @@ func mustReadAll(dpa *storage.DPA, hash storage.Key) (int, error) { return total, nil } -func testDeliveryFromNodes(adapter adapters.NodeAdapter) (*simulations.StepResult, error) { - nodes := 2 - conns := 1 - size := 8100 - skipCheck := true - - trigger := func(net *simulations.Network) chan discover.NodeID { - triggerC := make(chan discover.NodeID) - ticker := time.NewTicker(500 * time.Millisecond) - go func() { - defer ticker.Stop() - 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 { - rrdpa := storage.NewDPA(newRoundRobinStore(localStores[1:]...), storage.NewChunkerParams()) - rrdpa.Start() - 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() - hash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) - if err != nil { - return err - } - wait() - fileHash = hash +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 dpa.Stop() - log.Debug(fmt.Sprintf("retrieve %v", fileHash)) - time.Sleep(2 * time.Second) - n, err := mustReadAll(dpa, fileHash) - log.Debug(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err) + 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 nil + return triggerC } - } - 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() { + 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 + hash, wait, 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) { + 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 + 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 } - select { - case <-ctx.Done(): - return false, ctx.Err() - default: - } - 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, action, trigger, check, adapter) - if err != nil { - return nil, fmt.Errorf("Setting up simulation failed: %v", err) + result, err := runSimulation(nodes, conns, 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 } - if result.Error != nil { - return nil, fmt.Errorf("Simulation failed: %s", result.Error) - } - return result, err } 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) { @@ -556,6 +567,7 @@ func runSimulation(nodes, conns int, action func(*simulations.Network) func(cont DefaultService: serviceName, }) defer net.Shutdown() + // set nodes number of localstores globally available teardown, err := setLocalStores(nodes) defer teardown() if err != nil { @@ -563,6 +575,7 @@ func runSimulation(nodes, conns int, action func(*simulations.Network) func(cont } ids := make([]discover.NodeID, nodes) nodeCount = 0 + // start nodes for i := 0; i < nodes; i++ { node, err := net.NewNode() if err != nil { @@ -574,8 +587,7 @@ func runSimulation(nodes, conns int, action func(*simulations.Network) func(cont ids[i] = node.ID() } - // run a simulation which connects the 10 nodes in a ring and waits - // for full peer discovery + // run a simulation which connects the 10 nodes in a chain var addrs [][]byte wg := sync.WaitGroup{} log.Warn("runSimulation 1") @@ -606,10 +618,11 @@ func runSimulation(nodes, conns int, action func(*simulations.Network) func(cont log.Debug(fmt.Sprintf("nodes: %v", len(addrs))) - // 64 nodes ~ 1min - // 128 nodes ~ + // 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() @@ -624,7 +637,8 @@ func runSimulation(nodes, conns int, action func(*simulations.Network) func(cont return result, nil } -func newService(ctx *adapters.ServiceContext) (node.Service, error) { +// newDeliveryService +func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) { id := ctx.Config.ID addr := NewAddrFromNodeID(id) kad := NewKademlia(addr.Over(), NewKadParams()) @@ -632,34 +646,23 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) { dbAccess := NewDbAccess(localStore.(*storage.LocalStore)) streamer := NewStreamer(NewDelivery(kad, dbAccess)) 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 } nodeCount++ - run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - bzzPeer := &bzzPeer{ - Peer: protocols.NewPeer(p, rw, StreamerSpec), - localAddr: addr, - BzzAddr: NewAddrFromNodeID(p.ID()), - } - log.Warn("Run function kad On ", "local", id, "remote", p.ID()) - kad.On(bzzPeer) - go func() { - 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) - } + log.Warn("new service created") return &testDeliveryService{ - run: run, + addr: addr, + streamer: streamer, }, nil } type testDeliveryService struct { - run func(p *p2p.Peer, rw p2p.MsgReadWriter) error + addr *BzzAddr + streamer *Streamer } func (tds *testDeliveryService) Protocols() []p2p.Protocol { @@ -687,3 +690,24 @@ func (b *testDeliveryService) Start(server *p2p.Server) error { func (b *testDeliveryService) Stop() error { return nil } + +func (b *testDeliveryService) run(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) + } + }() + return b.streamer.Run(bzzPeer) +} diff --git a/swarm/network/requests.go b/swarm/network/requests.go index d2cf9fa344..7cc0a47946 100644 --- a/swarm/network/requests.go +++ b/swarm/network/requests.go @@ -49,7 +49,7 @@ func NewDelivery(overlay Overlay, dbAccess *DbAccess) *Delivery { // RetrieveRequestStreamer implements OutgoingStreamer type RetrieveRequestStreamer struct { - deliveryC chan *storage.Chunk + deliveryC chan []byte batchC chan []byte dbAccess *DbAccess currentLen uint64 @@ -58,7 +58,7 @@ type RetrieveRequestStreamer struct { // NewRetrieveRequestStreamer is RetrieveRequestStreamer constructor func NewRetrieveRequestStreamer(dbAccess *DbAccess) *RetrieveRequestStreamer { s := &RetrieveRequestStreamer{ - deliveryC: make(chan *storage.Chunk), + deliveryC: make(chan []byte), batchC: make(chan []byte), dbAccess: dbAccess, } @@ -72,11 +72,12 @@ func (s *RetrieveRequestStreamer) processDeliveries() { var batchC chan []byte for { select { - case delivery := <-s.deliveryC: - hashes = append(hashes, delivery.Key[:]...) + case hash := <-s.deliveryC: + hashes = append(hashes, hash...) batchC = s.batchC case batchC <- hashes: hashes = nil + batchC = nil } } } @@ -131,7 +132,7 @@ func (self *Delivery) handleRetrieveRequestMsg(sp *StreamerPeer, req *RetrieveRe sp.Deliver(chunk, s.priority) return } - streamer.deliveryC <- chunk + streamer.deliveryC <- chunk.Key[:] }() return nil } @@ -139,7 +140,7 @@ func (self *Delivery) handleRetrieveRequestMsg(sp *StreamerPeer, req *RetrieveRe if req.SkipCheck { return sp.Deliver(chunk, s.priority) } - streamer.deliveryC <- chunk + streamer.deliveryC <- chunk.Key[:] return nil } diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index 61351bf392..ea64734fd1 100644 --- a/swarm/network/streamer.go +++ b/swarm/network/streamer.go @@ -406,7 +406,7 @@ func (self *StreamerPeer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { hashes := req.Hashes want, err := bv.New(len(hashes) / HashSize) if err != nil { - return err + return fmt.Errorf("error initiaising bitvector of length %v: %v", len(hashes)/HashSize, err) } wg := sync.WaitGroup{} for i := 0; i < len(hashes); i += HashSize { @@ -472,7 +472,7 @@ func (self *StreamerPeer) handleWantedHashesMsg(req *WantedHashesMsg) error { l := len(hashes) / HashSize want, err := bv.NewFromBytes(req.Want, l) if err != nil { - return err + return fmt.Errorf("error initiaising bitvector of length %v: %v", l, err) } for i := 0; i < l; i++ { if want.Get(i) { @@ -514,12 +514,17 @@ func (self *StreamerPeer) SendPriority(msg interface{}, priority uint8) error { return self.pq.Push(nil, msg, int(priority)) } -// OfferedHashes sends OfferedHashesMsg protocol msg +// SendOfferedHashes sends OfferedHashesMsg protocol msg func (self *StreamerPeer) SendOfferedHashes(s *outgoingStreamer, f, t uint64) error { hashes, from, to, proof, err := s.SetNextBatch(f, t) if err != nil { return err } + if proof == nil { + proof = &HandoverProof{ + Handover: &Handover{}, + } + } s.currentBatch = hashes msg := &OfferedHashesMsg{ HandoverProof: proof, diff --git a/swarm/network/streamer_test.go b/swarm/network/streamer_test.go index 9d44934385..86ba965071 100644 --- a/swarm/network/streamer_test.go +++ b/swarm/network/streamer_test.go @@ -130,10 +130,7 @@ func (self *testIncomingStreamer) BatchDone(string, uint64, []byte, []byte) func } func (self *testOutgoingStreamer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { - proof := &HandoverProof{ - Handover: &Handover{}, - } - return make([]byte, HashSize), from + 1, to + 1, proof, nil + return make([]byte, HashSize), from + 1, to + 1, nil, nil } func (self *testOutgoingStreamer) GetData([]byte) []byte {