From ebec92e928d40d6baccfdab81212dfd59ffe58bc Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Tue, 23 Jan 2018 14:10:34 +0100 Subject: [PATCH] swarm: an attempt to debug syncer tests --- swarm/network/stream/delivery.go | 28 ++++++++++---- swarm/network/stream/delivery_test.go | 11 +++++- swarm/network/stream/messages.go | 3 ++ swarm/network/stream/syncer.go | 4 +- swarm/network/stream/syncer_test.go | 50 +++++++++++++++++++------ swarm/network/stream/testing/testing.go | 5 +-- swarm/storage/localstore.go | 1 - 7 files changed, 76 insertions(+), 26 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 4355997fd0..606be1703a 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -33,11 +33,14 @@ const ( ) type Delivery struct { - db *storage.DBAPI - overlay network.Overlay - receiveC chan *ChunkDeliveryMsg - getPeer func(discover.NodeID) *Peer - quit chan struct{} + db *storage.DBAPI + overlay network.Overlay + receiveC chan *ChunkDeliveryMsg + getPeer func(discover.NodeID) *Peer + quit chan struct{} + counterIn int + counterDone int + counterHash int } func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery { @@ -159,6 +162,7 @@ type ChunkDeliveryMsg struct { } func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error { + d.counterIn++ d.receiveC <- req return nil } @@ -182,10 +186,11 @@ R: chunk.SData = req.SData d.db.Put(chunk) log.Warn("reecived delivery", "hash", chunk.Key) - chunk.WaitToStore() - log.Warn("received delivery stored", "hash", chunk.Key) close(chunk.ReqC) + chunk.WaitToStore() + //log.Warn("received delivery stored", "hash", chunk.Key) log.Warn("received delivery requesters notified", "hash", chunk.Key) + d.counterDone++ } } @@ -220,3 +225,12 @@ func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ... } return errors.New("no peer found") } + +func (d *Delivery) PrintCounters(id discover.NodeID) { + if d.counterHash != d.counterDone { + log.Error(fmt.Sprintf("delivery %s: HASH and DONE not the same", id)) + } + log.Error(fmt.Sprintf("delivery %s chunks hash: %d", id, d.counterHash)) + log.Error(fmt.Sprintf("delivery %s chunks in: %d", id, d.counterIn)) + log.Error(fmt.Sprintf("delivery %s chunks done: %d", id, d.counterDone)) +} diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 1ca89ea5e6..904830535d 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -449,7 +449,10 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck }, } startedAt := time.Now() - result, err := sim.Run(conf) + timeout := 300 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + result, err := sim.Run(ctx, conf) finishedAt := time.Now() if err != nil { t.Fatalf("Setting up simulation failed: %v", err) @@ -586,7 +589,11 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // run the simulation in the background errc := make(chan error) go func() { - _, err := sim.Run(conf) + timeout := 300 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + _, err := sim.Run(ctx, conf) errc <- err }() diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 1cf9aeab43..24a63391d8 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -121,6 +121,9 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { wg := sync.WaitGroup{} for i := 0; i < len(hashes); i += HashSize { hash := hashes[i : i+HashSize] + + p.streamer.delivery.counterHash++ + if wait := s.NeedData(hash); wait != nil { want.Set(i/HashSize, true) wg.Add(1) diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index eff42b4cd0..9df6ea78d5 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -75,7 +75,9 @@ func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) { // GetSection retrieves the actual chunk from localstore func (s *SwarmSyncerServer) GetData(key []byte) []byte { chunk, err := s.db.Get(storage.Key(key)) - if err != nil { + if err == storage.ErrFetching { + <-chunk.ReqC + } else if err != nil { return nil } return chunk.SData diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index d57be133cb..57ec1dd59b 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -34,7 +34,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -const dataChunkCount = 500 +const dataChunkCount = 1000 func TestSyncerSimulation(t *testing.T) { // testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1) @@ -67,6 +67,16 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck if err != nil { t.Fatal(err.Error()) } + + defer func() { + for _, id := range sim.IDs { + deliveries[id].PrintCounters(id) + } + // for id, delivery := range deliveries { + // delivery.PrintCounters(id) + // } + }() + stores = make(map[discover.NodeID]storage.ChunkStore) deliveries = make(map[discover.NodeID]*Delivery) for i, id := range sim.IDs { @@ -91,14 +101,16 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } // collect hashes in po 1 from all nodes - var hashes []storage.Key + hashes := make([][]storage.Key, nodes) dbs := make([]*storage.DBAPI, nodes) for i := 0; i < nodes; i++ { dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore)) } + totalHashes := 0 for i := 1; i < nodes; i++ { dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool { - hashes = append(hashes, key) + hashes[i] = append(hashes[i], key) + totalHashes++ return true }) } @@ -149,15 +161,28 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } var found int - total := len(hashes) - for _, key := range hashes { - _, err := dbs[0].Get(key) - if err == nil { - found++ + for i, n := range hashes { + for _, key := range n { + chunk, err := dbs[0].Get(key) + if err == storage.ErrFetching { + <-chunk.ReqC + found++ + } else if err == nil { + found++ + } + + log.Error("staring dbs check", "key", key) + for j := i; j > 0; j-- { + _, err := dbs[j].Get(key) + if err != nil { + log.Error("get from node", "node", sim.IDs[j], "nodeID", j, "key", key.Hex(), "err", err) + break + } + } } } - log.Error("sync check", "bin", po, "found", found, "total", total) - pass := found == total + log.Error("sync check", "bin", po, "found", found, "total", totalHashes) + pass := found == totalHashes if !pass { return false, nil } @@ -175,7 +200,10 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck }, } startedAt := time.Now() - result, err := sim.Run(conf) + timeout := 4 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + result, err := sim.Run(ctx, conf) finishedAt := time.Now() if err != nil { t.Fatalf("Setting up simulation failed: %v", err) diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index e92b009f86..e1c50919e7 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -170,7 +170,7 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { return s, teardown, nil } -func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) { +func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.StepResult, error) { // bring up nodes, launch the servive nodes := conf.NodeCount conns := conf.ConnLevel @@ -204,9 +204,6 @@ func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) { // create an only locally retrieving dpa for the pivot node to test // if retriee requests have arrived - timeout := 300 * time.Second - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() result := simulations.NewSimulation(s.Net).Run(ctx, conf.Step) return result, nil } diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index ac6d642950..066a27028a 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -106,7 +106,6 @@ func (self *LocalStore) Put(chunk *Chunk) { // ChunkStores are remote and can have long latency func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { chunk, err = self.memStore.Get(key) - if err == nil { if chunk.ReqC != nil { select {