swarm-smoke: add syncDelay flag

swarm/network: add want delay timer to syncing (#1367)

swarm/network: synchronise peer.close() (#1369)

2 files changed, 21 insertions, 0 deletions

diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go
index 821cdaa9a..b43fdeee2 100644
--- a/swarm/network/stream/messages.go
+++ b/swarm/network/stream/messages.go
@@ -223,6 +223,9 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg
 		return fmt.Errorf("error initiaising bitvector of length %v: %v", lenHashes/HashSize, err)
 	}
 
+	var wantDelaySet bool
+	var wantDelay time.Time
+
 	ctr := 0
 	errC := make(chan error)
 	ctx, cancel := context.WithTimeout(ctx, syncBatchTimeout)
@@ -234,6 +237,13 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg
 		if wait := c.NeedData(ctx, hash); wait != nil {
 			ctr++
 			want.Set(i/HashSize, true)
+
+			// measure how long it takes before we mark chunks for retrieval, and actually send the request
+			if !wantDelaySet {
+				wantDelaySet = true
+				wantDelay = time.Now()
+			}
+
 			// create request and wait until the chunk data arrives and is stored
 			go func(w func(context.Context) error) {
 				select {
@@ -304,6 +314,12 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg
 			return
 		}
 		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
+
+		// record want delay
+		if wantDelaySet {
+			metrics.GetOrRegisterResettingTimer("handleoffered.wantdelay", nil).UpdateSince(wantDelay)
+		}
+
 		err := p.SendPriority(ctx, msg, c.priority)
 		if err != nil {
 			log.Warn("SendPriority error", "err", err)
diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go
index 17ce0d798..28fd06e4d 100644
--- a/swarm/network/stream/peer.go
+++ b/swarm/network/stream/peer.go
@@ -415,9 +415,14 @@ func (p *Peer) removeClientParams(s Stream) error {
 }
 
 func (p *Peer) close() {
+	p.serverMu.Lock()
+	defer p.serverMu.Unlock()
+
 	for _, s := range p.servers {
 		s.Close()
 	}
+
+	p.servers = nil
 }
 
 // runUpdateSyncing is a long running function that creates the initial