From 8802b9ce7fe8f9764bd50dfe072b37ebc9e84cb8 Mon Sep 17 00:00:00 2001
From: Anton Evangelatov <anton.evangelatov@gmail.com>
Date: Mon, 6 May 2019 10:58:42 +0200
Subject: swarm-smoke: add syncDelay flag

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

swarm/network: synchronise peer.close() (#1369)
---
 swarm/network/stream/messages.go | 16 ++++++++++++++++
 swarm/network/stream/peer.go     |  5 +++++
 2 files changed, 21 insertions(+)

(limited to 'swarm')

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
-- 
cgit v1.2.3