From c2bedb54fe13b272320bb458db1e1f86fef0deb7 Mon Sep 17 00:00:00 2001 From: Balint Gabor Date: Thu, 4 Jan 2018 16:51:11 +0100 Subject: [PATCH] Some draft stuff for Janos --- swarm/network/requests.go | 27 ----------------------- swarm/network/streamer.go | 46 +++++++++++++++++++++++++++++++++++++-- swarm/network/syncer.go | 4 ++++ 3 files changed, 48 insertions(+), 29 deletions(-) diff --git a/swarm/network/requests.go b/swarm/network/requests.go index 27e0ea2853..ef6ae18450 100644 --- a/swarm/network/requests.go +++ b/swarm/network/requests.go @@ -122,18 +122,6 @@ func addRequester(rs *storage.RequestStatus, req *retrieveRequestMsg) { rs.Requesters[req.Id] = append(list, req) } -/* - 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 storeRequestMsg 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 -} - func (self storeRequestMsg) String() string { var from string if self.from == nil { @@ -147,18 +135,3 @@ func (self storeRequestMsg) String() string { } return fmt.Sprintf("from: %v, ID: %v, SData %x", from, self.Id, self.SData[:end]) } - -// the entrypoint for store requests coming from the bzz wire protocol -// if key found locally, return. otherwise -// remote is untrusted, so hash is verified and chunk passed on to NetStore -func (self *RequestHandler) handleStoreRequestMsg(msg interface{}, p Peer) error { - req := msg.(*storeRequestMsg) - req.from = p - // TODO: chunk validation - chunk := storage.NewChunk(req.Key, nil) - chunk.SData = req.SData - chunk.Source = p - self.netStore.Put(chunk) - log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p)) - return nil -} diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index 0207f3a0fc..3c7314ceef 100644 --- a/swarm/network/streamer.go +++ b/swarm/network/streamer.go @@ -22,6 +22,7 @@ import ( "fmt" "sync" + "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,6 +92,18 @@ type UnsyncedKeysMsg struct { *HandoverProof // 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 UnsyncedKeysMsg func (self UnsyncedKeysMsg) String() string { return fmt.Sprintf("Stream '%v' [%v-%v] (%v)", self.Stream, self.From, self.To, len(self.Hashes)/HashSize) @@ -170,7 +183,7 @@ func (self *Streamer) PeerInfo(id discover.NodeID) interface{} { } // OutgoingStreamer interface for outgoing peer Streamer -type OutgoingStreamer interface { +type OutgoingStreamerBackend interface { CurrentBatch() []byte SetNextBatch(uint64, uint64) ([]byte, uint64, uint64, *HandoverProof, error) GetData([]byte) []byte @@ -178,7 +191,7 @@ type OutgoingStreamer interface { } // IncomingStreamer interface for incoming peer Streamer -type IncomingStreamer interface { +type IncomingStreamerBackend interface { NextBatch(uint64) (uint64, uint64) NeedData([]byte) func() Priority() int @@ -197,6 +210,16 @@ type StreamerPeer struct { quit chan struct{} } +type IncomingStreamer struct { + priority uint8 + peer *StreamerPeer +} + +type OutgoingStreamer struct { + priority uint8 + peer *StreamerPeer +} + // NewStreamerPeer is the constructor for StreamerPeer func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer { self := &StreamerPeer{ @@ -255,6 +278,10 @@ func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer) erro return nil } +func (self *OutgoingStreamer) Subscribe(s Stream) OutgoingStreamerBackend { + +} + // Subscribe initiates the streamer func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error { f, err := self.streamer.GetIncomingStreamer(s) @@ -384,6 +411,18 @@ func (self *StreamerPeer) handleTakeoverProofMsg(msg interface{}) error { return nil } +func (self *StreamerPeer) handleChunkDeliveryMsg(msg interface{}) error { + req := msg.(*chunkDeliveryMsg) + req.from = self + // TODO: chunk validation + chunk := storage.NewChunk(req.Key, nil) + chunk.SData = req.SData + chunk.Source = p + self.netStore.Put(chunk) + log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p)) + return nil +} + // Deliver sends a storeRequestMsg protocol message to the peer func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority int) error { msg := &storeRequestMsg{ @@ -451,6 +490,9 @@ func (self *StreamerPeer) HandleMsg(msg interface{}) error { case *WantedKeysMsg: return self.handleWantedKeysMsg(msg) + case *ChunkDeliveryMsg: + return self.handleChunkDeliveryMsg(msg) + default: return fmt.Errorf("unknown message type: %T", msg) } diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index 1d85116c99..fe212ff988 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -148,6 +148,10 @@ type IncomingSwarmSyncer struct { end, start uint64 } +type { + Subscribe() +} + // NewIncomingSwarmSyncer is a contructor for provable data exchange syncer func NewIncomingSwarmSyncer(po uint8, priority int, intervals []uint64, p Peer, store storage.ChunkStore, chunker storage.Chunker) (*IncomingSwarmSyncer, error) { self := &IncomingSwarmSyncer{