swarm: an attempt to debug syncer tests

swarm/network/stream/delivery.go

@@ -38,6 +38,9 @@ type Delivery struct {
 	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))
+}

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
 	}()
 

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)

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

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 i, n := range hashes {
-		for _, key := range hashes {
+			for _, key := range n {
-			_, err := dbs[0].Get(key)
+				chunk, err := dbs[0].Get(key)
-			if err == nil {
+				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)

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
 }

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 {