From afe2f02efc55b29a361659125ed6d24bce037e9d Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Tue, 13 Feb 2018 17:21:46 +0100 Subject: [PATCH] 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 | 5 +- swarm/network/stream/messages.go | 31 ++-- swarm/network/stream/peer.go | 170 +++++++++++++++------ swarm/network/stream/stream.go | 44 +++--- swarm/network/stream/streamer_test.go | 210 +++++++++++++++++++------- 5 files changed, 322 insertions(+), 138 deletions(-) diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index b51e48dde9..6b4aff89b0 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -174,9 +174,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { Hashes: hash, From: 0, // TODO: why is this 32??? - To: 32, - Stream: NewStream(swarmChunkServerStreamName, nil, false), - Initial: true, + To: 32, + Stream: NewStream(swarmChunkServerStreamName, nil, false), }, Peer: peerID, }, diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index dbc3fc24d3..1948234625 100644 --- a/swarm/network/stream/messages.go +++ b/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,8 +57,8 @@ func (s Stream) String() string { // SubcribeMsg is the protocol msg for requesting a stream(section) type SubscribeMsg struct { Stream Stream - History *Range - Priority uint8 // delivered on priority channel + History *Range `rlp:"nil"` + Priority uint8 // delivered on priority channel } func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { @@ -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 - if priority > 0 { - // decrement history stream priority - priority-- - } - os, err := p.setServer(historyStream, s, priority) + + os, err := p.setServer(getHistoryStream(req.Stream), s, getHistoryPriority(req.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,8 +145,7 @@ 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 + *HandoverProof // HandoverProof } // String pretty prints OfferedHashesMsg @@ -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) } }() diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 35375f5c34..65dbefea5d 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -33,31 +33,38 @@ import ( var sendTimeout = 5 * time.Second var ( - errServerNotFound = errors.New("server not found") - errClientNotFound = errors.New("client not found") + 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 type Peer struct { *protocols.Peer - streamer *Registry - pq *pq.PriorityQueue - serverMu sync.RWMutex - clientMu sync.RWMutex - servers map[string]*server - clients map[string]*client - quit chan struct{} + streamer *Registry + 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{} } // NewPeer is the constructor for Peer func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer { p := &Peer{ - Peer: peer, - pq: pq.New(int(PriorityQueue), PriorityQueueCap), - streamer: streamer, - servers: make(map[string]*server), - clients: make(map[string]*client), - quit: make(chan struct{}), + Peer: peer, + pq: pq.New(int(PriorityQueue), PriorityQueueCap), + 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()) go p.pq.Run(ctx, func(i interface{}) { p.Send(i) }) @@ -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{ - Client: i, + c = &client{ + 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() diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 17838e7a5c..85c9a79355 100644 --- a/swarm/network/stream/stream.go +++ b/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) - if err != nil { + // check if the stream is registered + 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) - if err != nil { - return err - } - err = peer.setClient(s, is, priority, r.intervalsStore) + err := peer.setClientParams(s, &clientParams{priority: priority}) if err != nil { return err } if s.Live && h != nil { - is, err := f(peer, s.Key, false) - if err != nil { - return err - } - p := priority - if p > 0 { - p-- - } - historyStream := NewStream(s.Name, s.Key, false) - err = peer.setClient(historyStream, is, p, r.intervalsStore) - if err != nil { + if err := peer.setClientParams( + getHistoryStream(s), + &clientParams{ + priority: getHistoryPriority(priority), + }, + ); 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 diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index da27f5907c..3b31017c0a 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -40,41 +40,57 @@ func TestStreamerSubscribe(t *testing.T) { } var ( - hash0 = sha3.Sum256([]byte{0}) - hash1 = sha3.Sum256([]byte{1}) - 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) + hash0 = sha3.Sum256([]byte{0}) + hash1 = sha3.Sum256([]byte{1}) + hash2 = sha3.Sum256([]byte{2}) + hashesTmp = append(hash0[:], hash1[:]...) + hashes = append(hashesTmp, hash2[:]...) ) type testClient struct { - t []byte + 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{ - t: t, - }, nil + return newTestClient(t), nil }) peerID := tester.IDs[0] @@ -112,24 +126,56 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { t.Fatalf("Expected no error, got %v", err) } - err = tester.TestExchanges(p2ptest.Exchange{ - Label: "Subscribe message", - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 4, - Msg: &SubscribeMsg{ - Stream: stream, - History: &Range{ - From: 5, - To: 8, + err = tester.TestExchanges( + p2ptest.Exchange{ + Label: "Subscribe message", + Expects: []p2ptest.Expect{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: stream, + History: &Range{ + From: 5, + To: 8, + }, + Priority: Top, }, - Priority: Top, + Peer: peerID, }, - 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{ - t: t, - }, nil + return newTestServer(t), nil }) peerID := tester.IDs[0] @@ -198,10 +242,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { HandoverProof: &HandoverProof{ Handover: &Handover{}, }, - Hashes: make([]byte, HashSize), - From: 6, - To: 9, - Initial: true, + Hashes: make([]byte, HashSize), + From: 6, + To: 9, }, 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{ - t: t, - }, nil + return newTestServer(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{ - t: t, - }, nil + tc = newTestClient(t) + return tc, nil }) peerID := tester.IDs[0] @@ -424,28 +530,28 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { t.Fatal(err) } - if len(receivedHashes) != 3 { - t.Fatalf("Expected number of received hashes %v, got %v", 3, len(receivedHashes)) + if len(tc.receivedHashes) != 3 { + 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") }