diff --git a/swarm/network/request_test.go b/swarm/network/request_test.go index 8b848611e3..8e032c6a3b 100644 --- a/swarm/network/request_test.go +++ b/swarm/network/request_test.go @@ -18,12 +18,42 @@ package network import ( "testing" - "time" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" "github.com/ethereum/go-ethereum/swarm/storage" ) +func TestStreamerRetrieveRequest(t *testing.T) { + // TODO: we only need streamer + tester, streamer, _, teardown, err := newStreamerTester(t) + defer teardown() + if err != nil { + t.Fatal(err) + } + + peerID := tester.IDs[0] + + streamer.delivery.RequestFromPeers(hash0[:], true) + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "RetrieveRequestMsg", + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 5, + Msg: &RetrieveRequestMsg{ + Key: hash0[:], + SkipCheck: true, + }, + Peer: peerID, + }, + }, + }) + + if err != nil { + t.Fatalf("Expected no error, got %v", err) + } +} + func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { // TODO: we only need streamer tester, streamer, _, teardown, err := newStreamerTester(t) @@ -32,16 +62,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { t.Fatal(err) } - err = waitForPeers(streamer, 1*time.Second) - if err != nil { - t.Fatal("timeout: peer is not created") - } - - peerId := tester.IDs[0] + peerID := tester.IDs[0] chunk := storage.NewChunk(storage.Key(hash0[:]), nil) - peer := streamer.getPeer(peerId) + peer := streamer.getPeer(peerID) peer.handleSubscribeMsg(&SubscribeMsg{ Stream: retrieveRequestStream, @@ -59,7 +84,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { Msg: &RetrieveRequestMsg{ Key: chunk.Key[:], }, - Peer: peerId, + Peer: peerID, }, }, Expects: []p2ptest.Expect{ @@ -71,7 +96,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { From: 0, To: 0, }, - Peer: peerId, + Peer: peerID, }, }, }) @@ -82,6 +107,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { } } +// upstream request server receives a retrieve Request and responds with +// offered hashes or delivery if skipHash is set to true func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { // TODO: we only need streamer tester, streamer, localStore, teardown, err := newStreamerTester(t) @@ -90,16 +117,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { t.Fatal(err) } - err = waitForPeers(streamer, 1*time.Second) - if err != nil { - t.Fatal("timeout: peer is not created") - } - - peerId := tester.IDs[0] - - chunk := storage.NewChunk(storage.Key(hash0[:]), nil) - - peer := streamer.getPeer(peerId) + peerID := tester.IDs[0] + peer := streamer.getPeer(peerID) peer.handleSubscribeMsg(&SubscribeMsg{ Stream: retrieveRequestStream, @@ -109,8 +128,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { Priority: Top, }) - chunk.SData = hash0[:] + hash := storage.Key(hash0[:]) + chunk := storage.NewChunk(hash, nil) + chunk.SData = hash localStore.Put(chunk) + chunk.WaitToStore() err = tester.TestExchanges(p2ptest.Exchange{ Label: "RetrieveRequestMsg", @@ -118,9 +140,9 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { p2ptest.Trigger{ Code: 5, Msg: &RetrieveRequestMsg{ - Key: chunk.Key[:], + Key: hash, }, - Peer: peerId, + Peer: peerID, }, }, Expects: []p2ptest.Expect{ @@ -128,14 +150,48 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { Code: 1, Msg: &OfferedHashesMsg{ HandoverProof: nil, - Hashes: chunk.Key[:], + Hashes: hash, From: 0, // TODO: why is this 32??? To: 32, Key: []byte{}, Stream: retrieveRequestStream, }, - Peer: peerId, + Peer: peerID, + }, + }, + }) + + if err != nil { + t.Fatal(err) + } + + hash = storage.Key(hash1[:]) + chunk = storage.NewChunk(hash, nil) + chunk.SData = hash1[:] + localStore.Put(chunk) + chunk.WaitToStore() + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "RetrieveRequestMsg", + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 5, + Msg: &RetrieveRequestMsg{ + Key: hash, + SkipCheck: true, + }, + Peer: peerID, + }, + }, + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 6, + Msg: &ChunkDeliveryMsg{ + Key: hash, + SData: hash, + }, + Peer: peerID, }, }, }) diff --git a/swarm/network/requests.go b/swarm/network/requests.go index ce68afd35d..e269e89d49 100644 --- a/swarm/network/requests.go +++ b/swarm/network/requests.go @@ -16,54 +16,48 @@ package network -import "github.com/ethereum/go-ethereum/swarm/storage" +import ( + "errors" + "fmt" + "time" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/swarm/storage" +) const retrieveRequestStream = "RETRIEVE_REQUEST" -// Intervals is a stream specific history of downloaded intervals -// for historical streams -type Intervals struct { - streamer *Streamer - key string +type Delivery struct { + dbAccess *DbAccess + overlay Overlay + receiveC chan *ChunkDeliveryMsg + getPeer func(discover.NodeID) *StreamerPeer + quit chan struct{} } -// func (s *Intervals) load() error { -// return s.streamer.load(s.key) -// } -// -// func (s *Intervals) save() error { -// return s.streamer.save(s.key) -// } -// -// func (s *Intervals) get() []uint64 { -// return s.streamer.get(s.key) -// } -// -// func (s *Intervals) set(v []uint64) { -// s.streamer.set(s.key, v) -// } -// -// func NewIntervals(key string, s *Streamer) *Intervals { -// return &Intervals{ -// streamer: s, -// key: key, -// } -// } +func NewDelivery(overlay Overlay, dbAccess *DbAccess) *Delivery { + return &Delivery{ + dbAccess: dbAccess, + overlay: overlay, + receiveC: make(chan *ChunkDeliveryMsg, 10), + } +} // RetrieveRequestStreamer implements OutgoingStreamer type RetrieveRequestStreamer struct { deliveryC chan *storage.Chunk batchC chan []byte - db *DbAccess + dbAccess *DbAccess currentLen uint64 } // NewRetrieveRequestStreamer is RetrieveRequestStreamer constructor -func NewRetrieveRequestStreamer(db *DbAccess) *RetrieveRequestStreamer { +func NewRetrieveRequestStreamer(dbAccess *DbAccess) *RetrieveRequestStreamer { s := &RetrieveRequestStreamer{ deliveryC: make(chan *storage.Chunk), batchC: make(chan []byte), - db: db, + dbAccess: dbAccess, } go s.processDeliveries() return s @@ -95,6 +89,108 @@ func (s *RetrieveRequestStreamer) SetNextBatch(_, _ uint64) (hashes []byte, from // GetData retrives chunk data from db store func (s *RetrieveRequestStreamer) GetData(key []byte) []byte { - chunk, _ := s.db.get(storage.Key(key)) + chunk, _ := s.dbAccess.get(storage.Key(key)) return chunk.SData } + +// RetrieveRequestMsg is the protocol msg for chunk retrieve requests +type RetrieveRequestMsg struct { + Key storage.Key + SkipCheck bool +} + +func (self *Delivery) handleRetrieveRequestMsg(sp *StreamerPeer, req *RetrieveRequestMsg) error { + s, err := sp.getOutgoingStreamer(retrieveRequestStream) + if err != nil { + return err + } + streamer := s.OutgoingStreamer.(*RetrieveRequestStreamer) + chunk, created := self.dbAccess.getOrCreateRequest(req.Key) + if chunk.ReqC != nil { + if created { + if err := self.RequestFromPeers(chunk.Key[:], false, sp.ID()); err != nil { + return nil + } + } + go func() { + t := time.NewTimer(3 * time.Minute) + defer t.Stop() + + select { + case <-chunk.ReqC: + case <-self.quit: + return + case <-t.C: + return + } + + if req.SkipCheck { + sp.Deliver(chunk, s.priority) + return + } + streamer.deliveryC <- chunk + }() + return nil + } + // TODO: call the retrieve function of the outgoing syncer + if req.SkipCheck { + sp.Deliver(chunk, s.priority) + return nil + } + streamer.deliveryC <- chunk + return nil +} + +type ChunkDeliveryMsg struct { + Key storage.Key + SData []byte // the stored chunk Data (incl size) +} + +func (self *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error { + chunk, err := self.dbAccess.get(req.Key) + if err != nil { + return err + } + + self.receiveC <- req + + log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, self)) + return nil +} + +func (self *Delivery) processReceivedChunks() { + for req := range self.receiveC { + chunk, err := self.dbAccess.get(req.Key) + if err != nil { + continue + } + chunk.SData = req.SData + self.dbAccess.put(chunk) + close(chunk.ReqC) + } +} + +// RequestFromPeers sends a chunk retrieve request to +func (self *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ...discover.NodeID) error { + var success bool + self.overlay.EachConn(hash, 255, func(p OverlayConn, po int, nn bool) bool { + spId := p.(Peer).ID() + for _, p := range peersToSkip { + if p == spId { + return true + } + } + sp := self.getPeer(spId) + // TODO: skip light nodes that do not accept retrieve requests + sp.SendPriority(&RetrieveRequestMsg{ + Key: hash, + SkipCheck: skipCheck, + }, Top) + success = true + return false + }) + if success { + return nil + } + return errors.New("no peer found") +} diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index 24dbc3b379..2d50a31390 100644 --- a/swarm/network/streamer.go +++ b/swarm/network/streamer.go @@ -21,9 +21,7 @@ import ( "errors" "fmt" "sync" - "time" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/protocols" bv "github.com/ethereum/go-ethereum/swarm/network/bitvector" @@ -91,18 +89,6 @@ type OfferedHashesMsg struct { *HandoverProof `rlp:"nil"` // HandoverProof } -/* - store requests are put in netstore so they are stored and then - forwarded to the peers in their kademlia proximity bin by the syncer -*/ -type ChunkDeliveryMsg struct { - Key storage.Key - SData []byte // the stored chunk Data (incl size) - // optional - Id uint64 // request ID. if delivery, the ID is retrieve request ID - from Peer // [not serialised] protocol registers the requester -} - // String pretty prints OfferedHashesMsg func (self OfferedHashesMsg) String() string { return fmt.Sprintf("Stream '%v' [%v-%v] (%v)", self.Stream, self.From, self.To, len(self.Hashes)/HashSize) @@ -129,28 +115,24 @@ type Streamer struct { peersLock sync.RWMutex outgoing map[string]func(*StreamerPeer, []byte) (OutgoingStreamer, error) incoming map[string]func(*StreamerPeer, []byte) (IncomingStreamer, error) - - dbAccess *DbAccess - overlay Overlay - receiveC chan *ChunkDeliveryMsg - peers map[discover.NodeID]*StreamerPeer + peers map[discover.NodeID]*StreamerPeer + delivery *Delivery } // NewStreamer is Streamer constructor -func NewStreamer(overlay Overlay, dbAccess *DbAccess) *Streamer { +func NewStreamer(delivery *Delivery) *Streamer { streamer := &Streamer{ outgoing: make(map[string]func(*StreamerPeer, []byte) (OutgoingStreamer, error)), incoming: make(map[string]func(*StreamerPeer, []byte) (IncomingStreamer, error)), - dbAccess: dbAccess, - overlay: overlay, - receiveC: make(chan *ChunkDeliveryMsg, 10), peers: make(map[discover.NodeID]*StreamerPeer), + delivery: delivery, } + delivery.getPeer = streamer.getPeer streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer, t []byte) (OutgoingStreamer, error) { - return NewRetrieveRequestStreamer(dbAccess), nil + return NewRetrieveRequestStreamer(delivery.dbAccess), nil }) streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { - return NewIncomingSwarmSyncer(p, dbAccess, nil) + return NewIncomingSwarmSyncer(p, delivery.dbAccess, nil) }) return streamer } @@ -215,7 +197,6 @@ type OutgoingStreamer interface { type incomingStreamer struct { IncomingStreamer priority uint8 - intervals *Intervals sessionAt uint64 live bool quit chan struct{} @@ -260,82 +241,13 @@ func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer { return self } -// RetrieveRequestMsg is the protocol msg for chunk retrieve requests -type RetrieveRequestMsg struct { - Key storage.Key -} - -func (self *StreamerPeer) handleRetrieveRequestMsg(req *RetrieveRequestMsg) error { - chunk, created := self.streamer.dbAccess.getOrCreateRequest(req.Key) - s, err := self.getOutgoingStreamer(retrieveRequestStream) - if err != nil { - return err - } - streamer := s.OutgoingStreamer.(*RetrieveRequestStreamer) - if chunk.ReqC != nil { - if created { - if err := self.streamer.Retrieve(chunk, self.ID()); err != nil { - return nil - } - } - go func() { - t := time.NewTicker(3 * time.Minute) - defer t.Stop() - - select { - case <-chunk.ReqC: - case <-self.quit: - return - case <-t.C: - return - } - - streamer.deliveryC <- chunk - }() - return nil - } - // TODO: call the retrieve function of the outgoing syncer - streamer.deliveryC <- chunk - return nil -} - -// Retrieve sends a chunk retrieve request to -func (self *Streamer) Retrieve(chunk *storage.Chunk, peersToSkip ...discover.NodeID) error { - var success bool - self.overlay.EachConn(chunk.Key[:], 255, func(p OverlayConn, po int, nn bool) bool { - spId := p.(Peer).ID() - for _, p := range peersToSkip { - if p == spId { - return true - } - } - sp := self.getPeer(spId) - // TODO: skip light nodes that do not accept retrieve requests - sp.SendPriority(&RetrieveRequestMsg{ - Key: chunk.Key[:], - }, Top) - success = true - return false - }) - if success { - return nil - } - return errors.New("no peer found") -} - func (self *Streamer) getPeer(peerId discover.NodeID) *StreamerPeer { - if self.peers == nil { - return nil - } self.peersLock.RLock() defer self.peersLock.RUnlock() return self.peers[peerId] } func (self *Streamer) setPeer(peer *StreamerPeer) { - if self.peers == nil { - self.peers = make(map[discover.NodeID]*StreamerPeer) - } self.peersLock.Lock() self.peers[peer.ID()] = peer self.peersLock.Unlock() @@ -347,33 +259,6 @@ func (self *Streamer) deletePeer(peer *StreamerPeer) { self.peersLock.Unlock() } -func (self *StreamerPeer) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error { - chunk, err := self.streamer.dbAccess.get(req.Key) - if err != nil { - return err - } - - self.streamer.receiveC <- req - - log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, self)) - return nil -} - -func (self *Streamer) processReceivedChunks() { - for { - select { - case req := <-self.receiveC: - chunk, err := self.dbAccess.get(req.Key) - if err != nil { - continue - } - chunk.SData = req.SData - self.dbAccess.put(chunk) - close(chunk.ReqC) - } - } -} - func (self *StreamerPeer) getOutgoingStreamer(s string) (*outgoingStreamer, error) { self.outgoingLock.RLock() defer self.outgoingLock.RUnlock() @@ -660,6 +545,7 @@ var StreamerSpec = &protocols.Spec{ TakeoverProofMsg{}, SubscribeMsg{}, RetrieveRequestMsg{}, + ChunkDeliveryMsg{}, }, } @@ -692,10 +578,10 @@ func (self *StreamerPeer) HandleMsg(msg interface{}) error { return self.handleWantedHashesMsg(msg) case *ChunkDeliveryMsg: - return self.handleChunkDeliveryMsg(msg) + return self.streamer.delivery.handleChunkDeliveryMsg(msg) case *RetrieveRequestMsg: - return self.handleRetrieveRequestMsg(msg) + return self.streamer.delivery.handleRetrieveRequestMsg(self, msg) default: return fmt.Errorf("unknown message type: %T", msg) diff --git a/swarm/network/streamer_test.go b/swarm/network/streamer_test.go index a28a0bf260..0d58d552c2 100644 --- a/swarm/network/streamer_test.go +++ b/swarm/network/streamer_test.go @@ -57,8 +57,8 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, *stora } dbAccess := NewDbAccess(localStore) - streamer := NewStreamer(to, dbAccess) - + delivery := NewDelivery(to, dbAccess) + streamer := NewStreamer(delivery) run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { bzzPeer := &bzzPeer{ Peer: protocols.NewPeer(p, rw, StreamerSpec), @@ -69,6 +69,12 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, *stora return streamer.Run(bzzPeer) } protocolTester := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), 1, run) + + err = waitForPeers(streamer, 1*time.Second) + if err != nil { + return nil, nil, nil, nil, errors.New("timeout: peer is not created") + } + return protocolTester, streamer, localStore, teardown, nil } @@ -151,14 +157,9 @@ func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) { }, nil }) - err = waitForPeers(streamer, 1*time.Second) - if err != nil { - t.Fatal("timeout: peer is not created") - } + peerID := tester.IDs[0] - peerId := tester.IDs[0] - - err = streamer.Subscribe(peerId, "foo", []byte{}, 5, 8, Top, true) + err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -175,7 +176,7 @@ func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) { To: 8, Priority: Top, }, - Peer: peerId, + Peer: peerID, }, }, }) @@ -199,12 +200,7 @@ func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) { }, nil }) - err = waitForPeers(streamer, 1*time.Second) - if err != nil { - t.Fatal("timeout: peer is not created") - } - - peerId := tester.IDs[0] + peerID := tester.IDs[0] err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", @@ -218,7 +214,7 @@ func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) { To: 8, Priority: Top, }, - Peer: peerId, + Peer: peerID, }, }, Expects: []p2ptest.Expect{ @@ -231,7 +227,7 @@ func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) { From: 6, To: 9, }, - Peer: peerId, + Peer: peerID, }, }, }) @@ -256,14 +252,9 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { }, nil }) - err = waitForPeers(streamer, 1*time.Second) - if err != nil { - t.Fatal("timeout: peer is not created") - } + peerID := tester.IDs[0] - peerId := tester.IDs[0] - - err = streamer.Subscribe(peerId, "foo", []byte{}, 5, 8, Top, true) + err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -280,7 +271,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { To: 8, Priority: Top, }, - Peer: peerId, + Peer: peerID, }, }, }, @@ -298,7 +289,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { To: 8, Stream: "foo", }, - Peer: peerId, + Peer: peerID, }, }, Expects: []p2ptest.Expect{ @@ -310,7 +301,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { From: 8, To: 0, }, - Peer: peerId, + Peer: peerID, }, }, }) @@ -346,43 +337,6 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { } -func TestRetrieveRequest(t *testing.T) { - // TODO: we only need streamer - tester, streamer, _, teardown, err := newStreamerTester(t) - defer teardown() - if err != nil { - t.Fatal(err) - } - - err = waitForPeers(streamer, 1*time.Second) - if err != nil { - t.Fatal("timeout: peer is not created") - } - - peerId := tester.IDs[0] - - chunk := storage.NewChunk(storage.Key(hash0[:]), nil) - - streamer.Retrieve(chunk) - - err = tester.TestExchanges(p2ptest.Exchange{ - Label: "RetrieveRequestMsg", - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 5, - Msg: &RetrieveRequestMsg{ - Key: chunk.Key[:], - }, - Peer: peerId, - }, - }, - }) - - if err != nil { - t.Fatalf("Expected no error, got %v", err) - } -} - func waitForPeers(streamer *Streamer, timeout time.Duration) error { ticker := time.NewTicker(10 * time.Millisecond) timeoutTimer := time.NewTimer(timeout)