swarm/netowrk/stream: create client on first offered hashes msg

Store client init params in a dedicated Peer map and construct
client only on the first offered hashes message.

Update tests and fix issues with shared variables in testClient

swarm/network/stream/delivery_test.go

@@ -176,7 +176,6 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 					// TODO: why is this 32???
 					To:     32,
 					Stream: NewStream(swarmChunkServerStreamName, nil, false),
-					Initial: true,
 				},
 				Peer: peerID,
 			},

swarm/network/stream/messages.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/ethereum/go-ethereum/log"
 	bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
-	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -58,7 +57,7 @@ func (s Stream) String() string {
 // SubcribeMsg is the protocol msg for requesting a stream(section)
 type SubscribeMsg struct {
 	Stream   Stream
-	History  *Range
+	History  *Range `rlp:"nil"`
 	Priority uint8  // delivered on priority channel
 }
 
@@ -97,7 +96,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 	}
 
 	go func() {
-		if err := p.SendOfferedHashes(os, from, to, true); err != nil {
+		if err := p.SendOfferedHashes(os, from, to); err != nil {
 			p.Drop(err)
 		}
 	}()
@@ -108,18 +107,13 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 		if err != nil {
 			return err
 		}
-		historyStream := NewStream(req.Stream.Name, req.Stream.Key, false)
+
-		priority := req.Priority
+		os, err := p.setServer(getHistoryStream(req.Stream), s, getHistoryPriority(req.Priority))
-		if priority > 0 {
-			// decrement history stream priority
-			priority--
-		}
-		os, err := p.setServer(historyStream, s, priority)
 		if err != nil {
 			return err
 		}
 		go func() {
-			if err := p.SendOfferedHashes(os, req.History.From, req.History.To, true); err != nil {
+			if err := p.SendOfferedHashes(os, req.History.From, req.History.To); err != nil {
 				p.Drop(err)
 			}
 		}()
@@ -151,7 +145,6 @@ type OfferedHashesMsg struct {
 	Stream         Stream // name of Stream
 	From, To       uint64 // peer and db-specific entry count
 	Hashes         []byte // stream of hashes (128)
-	Initial        bool
 	*HandoverProof        // HandoverProof
 }
 
@@ -163,7 +156,7 @@ func (m OfferedHashesMsg) String() string {
 // handleOfferedHashesMsg protocol msg handler calls the incoming streamer interface
 // Filter method
 func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
-	c, err := p.getClient(req.Stream)
+	c, _, err := p.getOrSetClient(req.Stream, req.From, req.To)
 	if err != nil {
 		return err
 	}
@@ -207,12 +200,6 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	// except
 	if c.stream.Live {
 		c.sessionAt = req.From
-		if req.Initial {
-			// create initial intervals for live stream starting from the first From value
-			if err := c.intervalsStore.Put(peerStreamIntervalsKey(p, req.Stream), intervals.NewIntervals(req.From)); err != nil {
-				return err
-			}
-		}
 	}
 	from, to := c.nextBatch(req.To)
 	log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To)
@@ -271,7 +258,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
 	hashes := s.currentBatch
 	// launch in go routine since GetBatch blocks until new hashes arrive
 	go func() {
-		if err := p.SendOfferedHashes(s, req.From, req.To, false); err != nil {
+		if err := p.SendOfferedHashes(s, req.From, req.To); err != nil {
 			p.Drop(err)
 		}
 	}()

swarm/network/stream/peer.go

@@ -35,6 +35,7 @@ var sendTimeout = 5 * time.Second
 var (
 	errServerNotFound       = errors.New("server not found")
 	errClientNotFound       = errors.New("client not found")
+	errClientParamsNotFound = errors.New("client params not found")
 )
 
 // Peer is the Peer extention for the streaming protocol
@@ -44,8 +45,13 @@ type Peer struct {
 	pq             *pq.PriorityQueue
 	serverMu       sync.RWMutex
 	clientMu       sync.RWMutex
+	clientParamsMu sync.RWMutex
 	servers        map[string]*server
 	clients        map[string]*client
+	// clientParams map keeps required client arguments
+	// that are set on Registry.Subscribe and used
+	// on creating a new client in offered hashes handler.
+	clientParams map[string]*clientParams
 	quit         chan struct{}
 }
 
@@ -57,6 +63,7 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
 		streamer:     streamer,
 		servers:      make(map[string]*server),
 		clients:      make(map[string]*client),
+		clientParams: make(map[string]*clientParams),
 		quit:         make(chan struct{}),
 	}
 	ctx, cancel := context.WithCancel(context.Background())
@@ -85,7 +92,7 @@ func (p *Peer) SendPriority(msg interface{}, priority uint8) error {
 }
 
 // SendOfferedHashes sends OfferedHashesMsg protocol msg
-func (p *Peer) SendOfferedHashes(s *server, f, t uint64, initial bool) error {
+func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
 	hashes, from, to, proof, err := s.SetNextBatch(f, t)
 	if err != nil {
 		return err
@@ -106,7 +113,6 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64, initial bool) error {
 		From:          from,
 		To:            to,
 		Stream:        s.stream,
-		Initial:       initial,
 	}
 	log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to)
 	return p.SendPriority(msg, s.priority)
@@ -123,17 +129,6 @@ func (p *Peer) getServer(s Stream) (*server, error) {
 	return server, nil
 }
 
-func (p *Peer) getClient(s Stream) (*client, error) {
-	p.clientMu.RLock()
-	defer p.clientMu.RUnlock()
-
-	client := p.clients[s.String()]
-	if client == nil {
-		return nil, fmt.Errorf("client '%v' not provided to peer %v", s, p.ID())
-	}
-	return client, nil
-}
-
 func (p *Peer) setServer(s Stream, o Server, priority uint8) (*server, error) {
 	p.serverMu.Lock()
 	defer p.serverMu.Unlock()
@@ -165,7 +160,18 @@ func (p *Peer) removeServer(s Stream) error {
 	return nil
 }
 
-func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore intervals.Store) error {
+func (p *Peer) getClient(s Stream) (*client, error) {
+	p.clientMu.RLock()
+	defer p.clientMu.RUnlock()
+
+	client := p.clients[s.String()]
+	if client == nil {
+		return nil, fmt.Errorf("client '%v' not provided to peer %v", s, p.ID())
+	}
+	return client, nil
+}
+
+func (p *Peer) setClient(s Stream, from, to uint64) error {
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
@@ -174,18 +180,45 @@ func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore inte
 		return fmt.Errorf("client %v already registered", sk)
 	}
 
+	_, err := p.setClientNolock(s, from, to)
+	return err
+}
+
+func (p *Peer) setClientNolock(s Stream, from, to uint64) (c *client, err error) {
+	f, err := p.streamer.GetClientFunc(s.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	is, err := f(p, s.Key, s.Live)
+	if err != nil {
+		return nil, err
+	}
+
+	cp, err := p.getClientParams(s)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		if err == nil {
+			if err := p.removeClientParams(s); err != nil {
+				log.Error("stream set client: remove client params", "stream", s, "peer", p, "err", err)
+			}
+		}
+	}()
+
 	intervalsKey := peerStreamIntervalsKey(p, s)
 	if s.Live {
 		// try to find previous history and live intervals and merge live into history
 		historyKey := peerStreamIntervalsKey(p, NewStream(s.Name, s.Key, false))
-		historyIntervals, err := intervalsStore.Get(historyKey)
+		historyIntervals, err := p.streamer.intervalsStore.Get(historyKey)
 		switch err {
 		case nil:
-			liveIntervals, err := intervalsStore.Get(intervalsKey)
+			liveIntervals, err := p.streamer.intervalsStore.Get(intervalsKey)
 			switch err {
 			case nil:
 				historyIntervals.Merge(liveIntervals)
-				if err := intervalsStore.Put(historyKey, historyIntervals); err != nil {
+				if err := p.streamer.intervalsStore.Put(historyKey, historyIntervals); err != nil {
 					log.Error("stream set client: put history intervals", "stream", s, "peer", p, "err", err)
 				}
 			case intervals.ErrNotFound:
@@ -196,25 +229,40 @@ func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore inte
 		default:
 			log.Error("stream set client: get history intervals", "stream", s, "peer", p, "err", err)
 		}
-	} else {
-		// create intervals for history stream
-		// live stream can create intervals when the first sessionAt is known
-		if err := intervalsStore.Put(intervalsKey, intervals.NewIntervals(0)); err != nil {
-			return err
 	}
+
+	if err := p.streamer.intervalsStore.Put(intervalsKey, intervals.NewIntervals(from)); err != nil {
+		return nil, err
 	}
 
 	next := make(chan error, 1)
-	p.clients[sk] = &client{
+	c = &client{
-		Client:         i,
+		Client:         is,
 		stream:         s,
-		priority:       priority,
+		priority:       cp.priority,
 		next:           next,
-		intervalsStore: intervalsStore,
+		intervalsStore: p.streamer.intervalsStore,
 		intervalsKey:   intervalsKey,
 	}
+	p.clients[s.String()] = c
 	next <- nil // this is to allow wantedKeysMsg before first batch arrives
-	return nil
+	return c, nil
+}
+
+func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created bool, err error) {
+	p.clientMu.RLock()
+	defer p.clientMu.RUnlock()
+
+	c = p.clients[s.String()]
+	if c != nil {
+		return c, false, nil
+	}
+
+	c, err = p.setClientNolock(s, from, to)
+	if err != nil {
+		return nil, false, err
+	}
+	return c, true, nil
 }
 
 func (p *Peer) removeClient(s Stream) error {
@@ -229,6 +277,42 @@ func (p *Peer) removeClient(s Stream) error {
 	return nil
 }
 
+func (p *Peer) getClientParams(s Stream) (*clientParams, error) {
+	p.clientParamsMu.RLock()
+	defer p.clientParamsMu.RUnlock()
+
+	params := p.clientParams[s.String()]
+	if params == nil {
+		return nil, fmt.Errorf("client params '%v' not provided to peer %v", s, p.ID())
+	}
+	return params, nil
+}
+
+func (p *Peer) setClientParams(s Stream, params *clientParams) error {
+	p.clientParamsMu.Lock()
+	defer p.clientParamsMu.Unlock()
+
+	sk := s.String()
+	if p.clientParams[sk] != nil {
+		return fmt.Errorf("client params %v already set", sk)
+	}
+	p.clientParams[sk] = params
+	return nil
+}
+
+func (p *Peer) removeClientParams(s Stream) error {
+	p.clientParamsMu.Lock()
+	defer p.clientParamsMu.Unlock()
+
+	sk := s.String()
+	_, ok := p.clientParams[sk]
+	if !ok {
+		return errClientParamsNotFound
+	}
+	delete(p.clientParams, sk)
+	return nil
+}
+
 func (p *Peer) close() {
 	for _, s := range p.servers {
 		s.Close()

swarm/network/stream/stream.go

@@ -125,8 +125,8 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Serv
 
 // Subscribe initiates the streamer
 func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error {
-	f, err := r.GetClientFunc(s.Name)
+	// check if the stream is registered
-	if err != nil {
+	if _, err := r.GetClientFunc(s.Name); err != nil {
 		return err
 	}
 
@@ -135,27 +135,18 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priorit
 		return fmt.Errorf("peer not found %v", peerId)
 	}
 
-	is, err := f(peer, s.Key, s.Live)
+	err := peer.setClientParams(s, &clientParams{priority: priority})
-	if err != nil {
-		return err
-	}
-	err = peer.setClient(s, is, priority, r.intervalsStore)
 	if err != nil {
 		return err
 	}
 
 	if s.Live && h != nil {
-		is, err := f(peer, s.Key, false)
+		if err := peer.setClientParams(
-		if err != nil {
+			getHistoryStream(s),
-			return err
+			&clientParams{
-		}
+				priority: getHistoryPriority(priority),
-		p := priority
+			},
-		if p > 0 {
+		); err != nil {
-			p--
-		}
-		historyStream := NewStream(s.Name, s.Key, false)
-		err = peer.setClient(historyStream, is, p, r.intervalsStore)
-		if err != nil {
 			return err
 		}
 	}
@@ -367,6 +358,12 @@ func (c *client) close() {
 	c.Close()
 }
 
+// clientParams store parameters for the new client
+// between a subscription and initial offered hashes request handling.
+type clientParams struct {
+	priority uint8
+}
+
 // Spec is the spec of the streamer protocol
 var Spec = &protocols.Spec{
 	Name:       "stream",
@@ -422,6 +419,17 @@ type Range struct {
 	From, To uint64
 }
 
+func getHistoryPriority(priority uint8) uint8 {
+	if priority == 0 {
+		return 0
+	}
+	return priority - 1
+}
+
+func getHistoryStream(s Stream) Stream {
+	return NewStream(s.Name, s.Key, false)
+}
+
 type API struct {
 	streamer *Registry
 	dpa      *storage.DPA

swarm/network/stream/streamer_test.go

@@ -45,36 +45,52 @@ var (
 	hash2     = sha3.Sum256([]byte{2})
 	hashesTmp = append(hash0[:], hash1[:]...)
 	hashes    = append(hashesTmp, hash2[:]...)
-	receivedHashes map[string][]byte = make(map[string][]byte)
-	wait0                            = make(chan bool)
-	wait2                            = make(chan bool)
-	batchDone                        = make(chan bool)
 )
 
 type testClient struct {
 	t              []byte
+	wait0          chan bool
+	wait2          chan bool
+	batchDone      chan bool
+	receivedHashes map[string][]byte
+}
+
+func newTestClient(t []byte) *testClient {
+	return &testClient{
+		t:              t,
+		wait0:          make(chan bool),
+		wait2:          make(chan bool),
+		batchDone:      make(chan bool),
+		receivedHashes: make(map[string][]byte),
+	}
 }
 
 type testServer struct {
 	t []byte
 }
 
+func newTestServer(t []byte) *testServer {
+	return &testServer{
+		t: t,
+	}
+}
+
 func (self *testClient) NeedData(hash []byte) func() {
-	receivedHashes[string(hash)] = hash
+	self.receivedHashes[string(hash)] = hash
 	if bytes.Equal(hash, hash0[:]) {
 		return func() {
-			<-wait0
+			<-self.wait0
 		}
 	} else if bytes.Equal(hash, hash2[:]) {
 		return func() {
-			<-wait2
+			<-self.wait2
 		}
 	}
 	return nil
 }
 
 func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {
-	close(batchDone)
+	close(self.batchDone)
 	return nil
 }
 
@@ -99,9 +115,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 	}
 
 	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
-		return &testClient{
+		return newTestClient(t), nil
-			t: t,
-		}, nil
 	})
 
 	peerID := tester.IDs[0]
@@ -112,10 +126,11 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 		t.Fatalf("Expected no error, got %v", err)
 	}
 
-	err = tester.TestExchanges(p2ptest.Exchange{
+	err = tester.TestExchanges(
+		p2ptest.Exchange{
 			Label: "Subscribe message",
 			Expects: []p2ptest.Expect{
-			p2ptest.Expect{
+				{
 					Code: 4,
 					Msg: &SubscribeMsg{
 						Stream: stream,
@@ -128,8 +143,39 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 					Peer: peerID,
 				},
 			},
-	})
+		},
-
+		// trigger OfferedHashesMsg to actually create the client
+		p2ptest.Exchange{
+			Label: "OfferedHashes message",
+			Triggers: []p2ptest.Trigger{
+				{
+					Code: 1,
+					Msg: &OfferedHashesMsg{
+						HandoverProof: &HandoverProof{
+							Handover: &Handover{},
+						},
+						Hashes: hashes,
+						From:   5,
+						To:     8,
+						Stream: stream,
+					},
+					Peer: peerID,
+				},
+			},
+			Expects: []p2ptest.Expect{
+				{
+					Code: 2,
+					Msg: &WantedHashesMsg{
+						Stream: stream,
+						Want:   []byte{5},
+						From:   8,
+						To:     0,
+					},
+					Peer: peerID,
+				},
+			},
+		},
+	)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -167,9 +213,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 	stream := NewStream("foo", nil, false)
 
 	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
-		return &testServer{
+		return newTestServer(t), nil
-			t: t,
-		}, nil
 	})
 
 	peerID := tester.IDs[0]
@@ -201,7 +245,6 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 					Hashes: make([]byte, HashSize),
 					From:   6,
 					To:     9,
-					Initial: true,
 				},
 				Peer: peerID,
 			},
@@ -230,6 +273,72 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 	}
 }
 
+func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
+	tester, streamer, _, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stream := NewStream("foo", nil, true)
+
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+		return newTestServer(t), nil
+	})
+
+	peerID := tester.IDs[0]
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "Subscribe message",
+		Triggers: []p2ptest.Trigger{
+			{
+				Code: 4,
+				Msg: &SubscribeMsg{
+					Stream:   stream,
+					Priority: Top,
+				},
+				Peer: peerID,
+			},
+		},
+		Expects: []p2ptest.Expect{
+			{
+				Code: 1,
+				Msg: &OfferedHashesMsg{
+					Stream: stream,
+					HandoverProof: &HandoverProof{
+						Handover: &Handover{},
+					},
+					Hashes: make([]byte, HashSize),
+					From:   1,
+					To:     1,
+				},
+				Peer: peerID,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "unsubscribe message",
+		Triggers: []p2ptest.Trigger{
+			{
+				Code: 0,
+				Msg: &UnsubscribeMsg{
+					Stream: stream,
+				},
+				Peer: peerID,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 	tester, streamer, _, teardown, err := newStreamerTester(t)
 	defer teardown()
@@ -238,9 +347,7 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 	}
 
 	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
-		return &testServer{
+		return newTestServer(t), nil
-			t: t,
-		}, nil
 	})
 
 	stream := NewStream("bar", nil, true)
@@ -325,7 +432,6 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 // 					Hashes:  make([]byte, HashSize),
 // 					From:    6,
 // 					To:      9,
-// 					Initial: true,
 // 				},
 // 				Peer: peerID,
 // 			},
@@ -339,7 +445,6 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 // 					From:    1,
 // 					To:      1,
 // 					Hashes:  make([]byte, HashSize),
-// 					Initial: true,
 // 				},
 // 				Peer: peerID,
 // 			},
@@ -360,10 +465,11 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 
 	stream := NewStream("foo", nil, true)
 
+	var tc *testClient
+
 	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
-		return &testClient{
+		tc = newTestClient(t)
-			t: t,
+		return tc, nil
-		}, nil
 	})
 
 	peerID := tester.IDs[0]
@@ -424,28 +530,28 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if len(receivedHashes) != 3 {
+	if len(tc.receivedHashes) != 3 {
-		t.Fatalf("Expected number of received hashes %v, got %v", 3, len(receivedHashes))
+		t.Fatalf("Expected number of received hashes %v, got %v", 3, len(tc.receivedHashes))
 	}
 
-	close(wait0)
+	close(tc.wait0)
 
 	timeout := time.NewTimer(100 * time.Millisecond)
 	defer timeout.Stop()
 
 	select {
-	case <-batchDone:
+	case <-tc.batchDone:
 		t.Fatal("batch done early")
 	case <-timeout.C:
 	}
 
-	close(wait2)
+	close(tc.wait2)
 
 	timeout2 := time.NewTimer(10000 * time.Millisecond)
 	defer timeout2.Stop()
 
 	select {
-	case <-batchDone:
+	case <-tc.batchDone:
 	case <-timeout2.C:
 		t.Fatal("timeout waiting batchdone call")
 	}