forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 17:33:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

storage/network: refactor retrieve requests

Browse source
This commit is contained in:
zelig 2018-01-13 17:02:54 +01:00
parent 20d4a867fd
commit b469552371
4 changed files with 238 additions and 246 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

106
swarm/network/request_test.go
View file

@ -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,
},
},
})

160
swarm/network/requests.go
View file

@ -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")
}

134
swarm/network/streamer.go
View file

@ -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)

84
swarm/network/streamer_test.go
View file

@ -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)