From 2ea9cf58f23926a3318cca8282b3ddbdae8d4e5c Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Fri, 5 Jan 2018 17:07:07 +0100 Subject: [PATCH] swarm: request streamer implementation --- swarm/network/protocol.go | 4 +- swarm/network/requests.go | 206 +++++++++++++++++------------------ swarm/network/streamer.go | 211 +++++++++++++++++++++++++++++------- swarm/network/syncer.go | 39 ++++--- swarm/storage/dpa.go | 57 ++++------ swarm/storage/forwarder.go | 16 --- swarm/storage/localstore.go | 23 ++++ swarm/storage/memstore.go | 4 +- swarm/storage/netstore.go | 157 +++++++++++++-------------- swarm/storage/types.go | 39 ++----- swarm/swarm.go | 21 ++-- 11 files changed, 442 insertions(+), 335 deletions(-) delete mode 100644 swarm/storage/forwarder.go diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 1f25464b11..9374d844c8 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -115,9 +115,9 @@ type Bzz struct { // * bzz config // * overlay driver // * peer store -func NewBzz(config *BzzConfig, kad Overlay, store StateStore) *Bzz { +func NewBzz(config *BzzConfig, kad Overlay, store StateStore, streamer *Streamer) *Bzz { return &Bzz{ - Streamer: NewStreamer(), + Streamer: streamer, Hive: NewHive(config.HiveParams, kad, store), localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr}, handshakes: make(map[discover.NodeID]*HandshakeMsg), diff --git a/swarm/network/requests.go b/swarm/network/requests.go index ef6ae18450..f5cea0ea44 100644 --- a/swarm/network/requests.go +++ b/swarm/network/requests.go @@ -16,122 +16,122 @@ package network -import ( - "fmt" +// import ( +// "fmt" - "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/swarm/storage" -) +// "github.com/ethereum/go-ethereum/log" +// "github.com/ethereum/go-ethereum/swarm/storage" +// ) -/* - Retrieve Request and store Request handling -*/ +// /* +// Retrieve Request and store Request handling +// */ -// Handler for storage/retrieval related protocol requests -type RequestHandler struct { - netStore *storage.NetStore -} +// // Handler for storage/retrieval related protocol requests +// type RequestHandler struct { +// netStore *storage.NetStore +// } -// NewEwquestHandler creates a new RequestHandler -// netStore to -func NewRequestHandler(netStore *storage.NetStore) *RequestHandler { - return &RequestHandler{ - netStore: netStore, // entrypoint internal - } -} +// // NewEwquestHandler creates a new RequestHandler +// // netStore to +// func NewRequestHandler(netStore *storage.NetStore) *RequestHandler { +// return &RequestHandler{ +// netStore: netStore, // entrypoint internal +// } +// } -/* -Retrieve request +// /* +// Retrieve request -MaxSize specifies the maximum size that the peer will accept. This is useful in -particular if we allow storage and delivery of multichunk payload representing -the entire or partial subtree unfolding from the requested root key. -So when only interested in limited part of a stream (infinite trees) or only -testing chunk availability etc etc, we can indicate it by limiting the size here. +// MaxSize specifies the maximum size that the peer will accept. This is useful in +// particular if we allow storage and delivery of multichunk payload representing +// the entire or partial subtree unfolding from the requested root key. +// So when only interested in limited part of a stream (infinite trees) or only +// testing chunk availability etc etc, we can indicate it by limiting the size here. -Request ID can be newly generated or kept from the request originator. +// Request ID can be newly generated or kept from the request originator. -*/ -type retrieveRequestMsg struct { - Key storage.Key // target Key address of chunk to be retrieved - Id uint64 // request id, request is a lookup if missing or zero - MaxSize uint64 // maximum size of delivery accepted - from *StreamerPeer // -} +// */ +// type retrieveRequestMsg struct { +// Key storage.Key // target Key address of chunk to be retrieved +// Id uint64 // request id, request is a lookup if missing or zero +// MaxSize uint64 // maximum size of delivery accepted +// from *StreamerPeer // +// } -func (self retrieveRequestMsg) String() string { - var from string - if self.from == nil { - from = "ourselves" - } else { - from = fmt.Sprintf("%x", self.from.Over()) - } - var target []byte - if len(self.Key) > 3 { - target = self.Key[:4] - } - return fmt.Sprintf("Requester: %v, Key: %x; ID: %v, MaxSize: %v", from, target, self.Id, self.MaxSize) -} +// func (self retrieveRequestMsg) String() string { +// var from string +// if self.from == nil { +// from = "ourselves" +// } else { +// from = fmt.Sprintf("%x", self.from.Over()) +// } +// var target []byte +// if len(self.Key) > 3 { +// target = self.Key[:4] +// } +// return fmt.Sprintf("Requester: %v, Key: %x; ID: %v, MaxSize: %v", from, target, self.Id, self.MaxSize) +// } // entrypoint for retrieve requests coming from the bzz wire protocol // checks swap balance - return if peer has no credit -func (self *StreamerPeer) handleRetrieveRequestMsg(msg interface{}) error { - req := msg.(*retrieveRequestMsg) - req.from = self - // TODO: - // swap - record credit for 1 request - // note that only charge actual reqsearches +// func (self *StreamerPeer) handleRetrieveRequestMsg(msg interface{}) error { +// req := msg.(*retrieveRequestMsg) +// req.from = self +// // TODO: +// // swap - record credit for 1 request +// // note that only charge actual reqsearches - // call storage.NetStore#Get which - // blocks until local retrieval finished - // launches cloud retrieval - chunk, _ := self.netStore.Get(req.Key) - rs := chunk.Req - if rs != nil { - rs = storage.NewRequestStatus(req.Key) - addRequester(rs, req) - chunk.Req = rs - } +// // call storage.NetStore#Get which +// // blocks until local retrieval finished +// // launches cloud retrieval +// chunk, _ := self.netStore.Get(req.Key) +// rs := chunk.Req +// if rs != nil { +// rs = storage.NewRequestStatus(req.Key) +// addRequester(rs, req) +// chunk.Req = rs +// } - // check if we can immediately deliver - if chunk.SData != nil { - if req.MaxSize == 0 || int64(req.MaxSize) >= chunk.Size { - err := self.Deliver(chunk, Top) - if err != nil { - log.Trace(fmt.Sprintf("%v - content found, delivery error: %v", req.Key.Log(), err)) - return nil - } - log.Trace(fmt.Sprintf("%v - content found, delivering...", req.Key.Log())) - } else { - log.Trace(fmt.Sprintf("%v - content found, not wanted", req.Key.Log())) - } - } else { - log.Trace(fmt.Sprintf("content not found locally, retrieve via bzz", req.Key.Log())) - } - return nil -} +// // check if we can immediately deliver +// if chunk.SData != nil { +// if req.MaxSize == 0 || int64(req.MaxSize) >= chunk.Size { +// err := self.Deliver(chunk, Top) +// if err != nil { +// log.Trace(fmt.Sprintf("%v - content found, delivery error: %v", req.Key.Log(), err)) +// return nil +// } +// log.Trace(fmt.Sprintf("%v - content found, delivering...", req.Key.Log())) +// } else { +// log.Trace(fmt.Sprintf("%v - content found, not wanted", req.Key.Log())) +// } +// } else { +// log.Trace(fmt.Sprintf("content not found locally, retrieve via bzz", req.Key.Log())) +// } +// return nil +// } -/* -adds a new peer to an existing open request -only add if less than requesterCount peers forwarded the same request id so far -note this is done irrespective of status (searching or found) -*/ -func addRequester(rs *storage.RequestStatus, req *retrieveRequestMsg) { - log.Trace(fmt.Sprintf("Depo.addRequester: key %v - add peer to req.Id %v", req.Key.Log(), req.Id)) - list := rs.Requesters[req.Id] - rs.Requesters[req.Id] = append(list, req) -} +// /* +// adds a new peer to an existing open request +// only add if less than requesterCount peers forwarded the same request id so far +// note this is done irrespective of status (searching or found) +// */ +// func addRequester(rs *storage.RequestStatus, req *retrieveRequestMsg) { +// log.Trace(fmt.Sprintf("Depo.addRequester: key %v - add peer to req.Id %v", req.Key.Log(), req.Id)) +// list := rs.Requesters[req.Id] +// rs.Requesters[req.Id] = append(list, req) +// } -func (self storeRequestMsg) String() string { - var from string - if self.from == nil { - from = "self" - } else { - from = fmt.Sprintf("%x", self.from.Over()) - } - end := len(self.SData) - if len(self.SData) > 10 { - end = 10 - } - return fmt.Sprintf("from: %v, ID: %v, SData %x", from, self.Id, self.SData[:end]) -} +// func (self storeRequestMsg) String() string { +// var from string +// if self.from == nil { +// from = "self" +// } else { +// from = fmt.Sprintf("%x", self.from.Over()) +// } +// end := len(self.SData) +// if len(self.SData) > 10 { +// end = 10 +// } +// return fmt.Sprintf("from: %v, ID: %v, SData %x", from, self.Id, self.SData[:end]) +// } diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index 3c7314ceef..19f948b4fa 100644 --- a/swarm/network/streamer.go +++ b/swarm/network/streamer.go @@ -21,6 +21,7 @@ import ( "errors" "fmt" "sync" + "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" @@ -128,13 +129,20 @@ type Streamer struct { outgoingLock sync.RWMutex outgoing map[Stream]func(*StreamerPeer) (OutgoingStreamer, error) incoming map[Stream]func(*StreamerPeer) (IncomingStreamer, error) + + dbAccess *DbAccess + overlay Overlay + receiveC chan *ChunkDeliveryMsg } // NewStreamer is Streamer constructor -func NewStreamer() *Streamer { +func NewStreamer(overlay Overlay, dbAccess *DbAccess) *Streamer { return &Streamer{ outgoing: make(map[Stream]func(*StreamerPeer) (OutgoingStreamer, error)), incoming: make(map[Stream]func(*StreamerPeer) (IncomingStreamer, error)), + dbAccess: dbAccess, + overlay: overlay, + receiveC: make(chan *ChunkDeliveryMsg, 10), } } @@ -183,15 +191,15 @@ func (self *Streamer) PeerInfo(id discover.NodeID) interface{} { } // OutgoingStreamer interface for outgoing peer Streamer -type OutgoingStreamerBackend interface { +type OutgoingStreamer interface { CurrentBatch() []byte - SetNextBatch(uint64, uint64) ([]byte, uint64, uint64, *HandoverProof, error) + SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) GetData([]byte) []byte Priority() int } // IncomingStreamer interface for incoming peer Streamer -type IncomingStreamerBackend interface { +type IncomingStreamer interface { NextBatch(uint64) (uint64, uint64) NeedData([]byte) func() Priority() int @@ -200,9 +208,10 @@ type IncomingStreamerBackend interface { // StreamerPeer is the Peer extention for the streaming protocol type StreamerPeer struct { Peer - streamer *Streamer - pq *pq.PriorityQueue - netStore storage.ChunkStore + streamer *Streamer + pq *pq.PriorityQueue + //netStore storage.ChunkStore + dbAccess *DbAccess outgoingLock sync.RWMutex incomingLock sync.RWMutex outgoing map[Stream]OutgoingStreamer @@ -210,15 +219,15 @@ type StreamerPeer struct { quit chan struct{} } -type IncomingStreamer struct { - priority uint8 - peer *StreamerPeer -} +// type IncomingStreamer struct { +// priority uint8 +// peer *StreamerPeer +// } -type OutgoingStreamer 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 { @@ -238,6 +247,142 @@ func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer { return self } +type RetrieveRequestMsg struct { + Key storage.Key +} + +// RetrieveRequestStreamer implements OutgoingStreamer +type RetrieveRequestStreamer struct { + deliveryC chan *storage.Chunk + batchC chan []byte + dbAccess *DbAccess + currentBatch []byte + currentLen uint64 +} + +func RegisterRequestStreamer(streamer *Streamer, dbAccess *DbAccess) { + streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer) (OutgoingStreamer, error) { + return NewRetrieveRequestStreamer(dbAccess), nil + }) + streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer) (IncomingStreamer, error) { + return NewIncomingSwarmSyncer(Top, nil, p, dbAccess, nil) + }) +} + +func NewRetrieveRequestStreamer(dbAccess *DbAccess) *RetrieveRequestStreamer { + s := &RetrieveRequestStreamer{ + deliveryC: make(chan *storage.Chunk), + batchC: make(chan []byte), + dbAccess: dbAccess, + } + go s.processDeliveries() + return s +} + +func (s *RetrieveRequestStreamer) processDeliveries() { + var hashes []byte + for { + select { + case delivery := <-s.deliveryC: + hashes = append(hashes, delivery.Key[:]...) + case s.batchC <- hashes: + hashes = nil + } + } +} + +func (s *RetrieveRequestStreamer) CurrentBatch() []byte { + return s.currentBatch +} + +func (s *RetrieveRequestStreamer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) { + hashes = <-s.batchC + s.currentBatch = hashes + from = s.currentLen + s.currentLen += uint64(len(hashes)) + to = s.currentLen + return +} + +func (s *RetrieveRequestStreamer) GetData(key []byte) []byte { + chunk, _ := s.dbAccess.get(storage.Key(key)) + return chunk.SData +} + +func (s *RetrieveRequestStreamer) Priority() int { + return Top +} + +const retrieveRequestStream = Stream("RETRIEVE_REQUEST") + +func (self *StreamerPeer) handleRetrieveRequestMsg(req *RetrieveRequestMsg) error { + chunk, created := self.dbAccess.getOrCreateRequest(req.Key) + s, err := self.getOutgoingStreamer(retrieveRequestStream) + if err != nil { + return err + } + streamer := s.(*RetrieveRequestStreamer) + if chunk.ReqC != nil { + if created { + if err := self.streamer.Retrieve(chunk); err != nil { + return err + } + } + 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 +} + +func (self *Streamer) Retrieve(chunk *storage.Chunk) error { + // TODO: using the overlay find the closes peer to send the retrieve + // request to. + // self.Overlay.EachConn(to, 256, func(op network.OverlayConn, po int, isproxbin bool) bool {}) + return nil +} + +func (self *StreamerPeer) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error { + chunk, err := self.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 Stream) (OutgoingStreamer, error) { self.outgoingLock.RLock() defer self.outgoingLock.RUnlock() @@ -278,10 +423,6 @@ 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) @@ -303,8 +444,7 @@ func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error { return nil } -func (self *StreamerPeer) handleSubscribeMsg(msg interface{}) error { - req := msg.(*SubscribeMsg) +func (self *StreamerPeer) handleSubscribeMsg(req *SubscribeMsg) error { f, err := self.streamer.GetOutgoingStreamer(req.Stream) if err != nil { return err @@ -322,8 +462,7 @@ func (self *StreamerPeer) handleSubscribeMsg(msg interface{}) error { // handleUnsyncedKeysMsg protocol msg handler calls the incoming streamer interface // Filter method -func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error { - req := msg.(*UnsyncedKeysMsg) +func (self *StreamerPeer) handleUnsyncedKeysMsg(req *UnsyncedKeysMsg) error { s, err := self.getIncomingStreamer(req.Stream) if err != nil { return err @@ -357,7 +496,7 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error { if from == to { return nil } - msg = &WantedKeysMsg{ + msg := &WantedKeysMsg{ Stream: req.Stream, Want: want.Bytes(), From: from, @@ -370,8 +509,7 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error { // handleWantedKeysMsg protocol msg handler // * sends the next batch of unsynced keys // * sends the actual data chunks as per WantedKeysMsg -func (self *StreamerPeer) handleWantedKeysMsg(msg interface{}) error { - req := msg.(*WantedKeysMsg) +func (self *StreamerPeer) handleWantedKeysMsg(req *WantedKeysMsg) error { s, err := self.getOutgoingStreamer(req.Stream) if err != nil { return err @@ -401,8 +539,7 @@ func (self *StreamerPeer) handleWantedKeysMsg(msg interface{}) error { return nil } -func (self *StreamerPeer) handleTakeoverProofMsg(msg interface{}) error { - req := msg.(*TakeoverProofMsg) +func (self *StreamerPeer) handleTakeoverProofMsg(req *TakeoverProofMsg) error { _, err := self.getOutgoingStreamer(req.Stream) if err != nil { return err @@ -411,21 +548,9 @@ 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{ + msg := &ChunkDeliveryMsg{ Key: chunk.Key, SData: chunk.SData, } @@ -470,6 +595,10 @@ var StreamerSpec = &protocols.Spec{ func (s *Streamer) Run(p *bzzPeer) error { sp := NewStreamerPeer(p, s) // load saved intervals + sp.handleSubscribeMsg(&SubscribeMsg{ + Stream: retrieveRequestStream, + Priority: uint8(Top), + }) defer close(sp.quit) return sp.Run(sp.HandleMsg) } diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index fe212ff988..ed87e57870 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -55,6 +55,16 @@ func (self *DbAccess) iterator(from uint64, to uint64, po uint8, f func(storage. return self.db.SyncIterator(from, to, po, f) } +// to obtain the chunks from key or request db entry only +func (self *DbAccess) getOrCreateRequest(key storage.Key) (*storage.Chunk, bool) { + return self.loc.GetOrCreateRequest(key) +} + +// to obtain the chunks from key or request db entry only +func (self *DbAccess) put(chunk *storage.Chunk) { + self.loc.Put(chunk) +} + // OutgoingSwarmSyncer implements an OutgoingStreamer for history syncing on bins // offered streams: // * live request delivery with or without checkback @@ -133,7 +143,6 @@ func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64, // IncomingSwarmSyncer type IncomingSwarmSyncer struct { - po uint8 priority int sessionAt uint64 nextC chan struct{} @@ -142,23 +151,19 @@ type IncomingSwarmSyncer struct { sessionReader storage.LazySectionReader retrieveC chan *storage.Chunk storeC chan *storage.Chunk - store storage.ChunkStore + dbAccess *DbAccess chunker storage.Chunker currentRoot storage.Key + requestFunc func(chunk *storage.Chunk) 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) { +func NewIncomingSwarmSyncer(priority int, intervals []uint64, p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) { self := &IncomingSwarmSyncer{ - po: po, priority: priority, intervals: intervals, - store: store, + dbAccess: dbAccess, chunker: chunker, } return self, nil @@ -194,12 +199,12 @@ func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) { for po := uint8(0); po < maxPO; po++ { stream := Stream(fmt.Sprintf("SYNC-%02d-live", po)) streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) { - return NewIncomingSwarmSyncer(po, High, nil, p, nil, nil) + return NewIncomingSwarmSyncer(High, nil, p, nil, nil) }) stream = Stream(fmt.Sprintf("SYNC-%02d-history", po)) streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) { //intervals := loadIntervals(p, po, false) - return NewIncomingSwarmSyncer(po, Mid, nil, p, nil, nil) + return NewIncomingSwarmSyncer(Mid, nil, p, nil, nil) }) // stream = fmt.Sprintf("SYNC-%02d-delete", po) // streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { @@ -210,13 +215,11 @@ func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) { } // NeedData -func (self *IncomingSwarmSyncer) NeedData(key []byte) func() { - chunk, err := self.store.Get(key) - if err == nil { - if chunk.SData == nil { - // send a request instead - return nil - } +func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) { + chunk, created := self.dbAccess.getOrCreateRequest(key) + // TODO: we may want to request from this peer anyway even if the request exists + if chunk.ReqC == nil || !created { + return nil } // create request and wait until the chunk data arrives and is stored return chunk.WaitToStore diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index 354ff32c15..7ef8a80a99 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -179,53 +179,44 @@ func (self *DPA) storeWorker() { // access by calling network is blocking with a timeout type dpaChunkStore struct { - n int - localStore ChunkStore - netStore ChunkStore + localStore *LocalStore + retrieve func(chunk *Chunk) error } -func NewDpaChunkStore(localStore, netStore ChunkStore) *dpaChunkStore { - return &dpaChunkStore{0, localStore, netStore} +func NewDpaChunkStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *dpaChunkStore { + return &dpaChunkStore{localStore, retrieve} } // Get is the entrypoint for local retrieve requests // waits for response or times out func (self *dpaChunkStore) Get(key Key) (chunk *Chunk, err error) { - chunk, err = self.netStore.Get(key) - if chunk.SData != nil { + var created bool + chunk, created = self.localStore.GetOrCreateRequest(key) + if chunk.ReqC == nil { log.Trace(fmt.Sprintf("DPA.Get: %v found locally, %d bytes", key.Log(), len(chunk.SData))) return } - // TODO: use self.timer time.Timer and reset with defer disableTimer - timer := time.After(searchTimeout) - select { - case <-timer: - log.Trace(fmt.Sprintf("DPA.Get: %v request time out ", key.Log())) - err = notFound - case <-chunk.Req.C: - log.Trace(fmt.Sprintf("DPA.Get: %v retrieved, %d bytes (%p)", key.Log(), len(chunk.SData), chunk)) + + if created { + if err := self.retrieve(chunk); err != nil { + return nil, err + } } - return + t := time.NewTicker(searchTimeout) + defer t.Stop() + + select { + case <-t.C: + log.Trace(fmt.Sprintf("DPA.Get: %v request time out ", key.Log())) + return nil, notFound + case <-chunk.ReqC: + } + return chunk, nil } // Put is the entrypoint for local store requests coming from storeLoop -func (self *dpaChunkStore) Put(entry *Chunk) { - chunk, err := self.localStore.Get(entry.Key) - if err != nil { - log.Trace(fmt.Sprintf("DPA.Put: %v new chunk. call netStore.Put", entry.Key.Log())) - chunk = entry - } else if chunk.SData == nil { - log.Trace(fmt.Sprintf("DPA.Put: %v request entry found", entry.Key.Log())) - chunk.SData = entry.SData - chunk.Size = entry.Size - } else { - log.Trace(fmt.Sprintf("DPA.Put: %v chunk already known", entry.Key.Log())) - return - } - // from this point on the storage logic is the same with network storage requests - log.Trace(fmt.Sprintf("DPA.Put %v: %v", self.n, chunk.Key.Log())) - self.n++ - self.netStore.Put(chunk) +func (self *dpaChunkStore) Put(chunk *Chunk) { + self.localStore.Put(chunk) } // Close chunk store diff --git a/swarm/storage/forwarder.go b/swarm/storage/forwarder.go deleted file mode 100644 index 0d1acfab57..0000000000 --- a/swarm/storage/forwarder.go +++ /dev/null @@ -1,16 +0,0 @@ -package storage - -// implements CloudStore -// noop placeholder for netstore functionality - -type Forwarder struct { -} - -func (self *Forwarder) Store(chunk *Chunk) { -} - -func (self *Forwarder) Retrieve(chunk *Chunk) { -} - -func (self *Forwarder) Deliver(chunk *Chunk) { -} diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index c70271a09a..843505c2cb 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -18,6 +18,9 @@ package storage import ( "encoding/binary" + "fmt" + + "github.com/ethereum/go-ethereum/log" ) // LocalStore is a combination of inmemory db over a disk persisted db @@ -66,6 +69,26 @@ func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { return } +// retrieve logic common for local and network chunk retrieval requests +func (self *LocalStore) GetOrCreateRequest(key Key) (chunk *Chunk, created bool) { + var err error + chunk, err = self.Get(key) + if err == nil { + if chunk.ReqC == nil { + log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v found locally", key)) + } else { + log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v hit on an existing request", key)) + // no need to launch again + } + return chunk, false + } + // no data and no request status + log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v not found locally. open new request", key)) + chunk = NewChunk(key, make(chan bool)) + self.memStore.Put(chunk) + return chunk, true +} + // Close local store func (self *LocalStore) Close() { self.DbStore.Close() diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go index fed1ae0094..e8e393baa9 100644 --- a/swarm/storage/memstore.go +++ b/swarm/storage/memstore.go @@ -168,8 +168,8 @@ func (s *MemStore) Put(entry *Chunk) { entry.Size = node.entry.Size entry.SData = node.entry.SData } - if entry.Req == nil { - entry.Req = node.entry.Req + if entry.ReqC == nil { + entry.ReqC = node.entry.ReqC } entry.C = node.entry.C node.entry = entry diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go index 2afba75811..4a7caf7c2e 100644 --- a/swarm/storage/netstore.go +++ b/swarm/storage/netstore.go @@ -17,38 +17,43 @@ package storage import ( - "fmt" "path/filepath" "time" - - "github.com/ethereum/go-ethereum/log" ) -/* -NetStore is a cloud storage access abstaction layer for swarm -it contains the shared logic of network served chunk store/retrieval requests -both local (coming from DPA api) and remote (coming from peers via bzz protocol) -it implements the ChunkStore interface and embeds LocalStore +// import ( +// "fmt" +// "path/filepath" +// "time" -It is called by the bzz protocol instances via Depo (the store/retrieve request handler) -a protocol instance is running on each peer, so this is heavily parallelised. -NetStore falls back to a backend (CloudStorage interface) -implemented by bzz/network/forwarder. forwarder or IPFS or IPΞS -*/ -type NetStore struct { - hashfunc SwarmHasher - localStore *LocalStore - cloud CloudStore -} +// "github.com/ethereum/go-ethereum/log" +// ) -// backend engine for cloud store -// It can be aggregate dispatching to several parallel implementations: -// bzz/network/forwarder. forwarder or IPFS or IPΞS -type CloudStore interface { - Store(*Chunk) - Deliver(*Chunk) - Retrieve(*Chunk) -} +// /* +// NetStore is a cloud storage access abstaction layer for swarm +// it contains the shared logic of network served chunk store/retrieval requests +// both local (coming from DPA api) and remote (coming from peers via bzz protocol) +// it implements the ChunkStore interface and embeds LocalStore + +// It is called by the bzz protocol instances via Depo (the store/retrieve request handler) +// a protocol instance is running on each peer, so this is heavily parallelised. +// NetStore falls back to a backend (CloudStorage interface) +// implemented by bzz/network/forwarder. forwarder or IPFS or IPΞS +// */ +// type NetStore struct { +// hashfunc SwarmHasher +// localStore *LocalStore +// cloud CloudStore +// } + +// // backend engine for cloud store +// // It can be aggregate dispatching to several parallel implementations: +// // bzz/network/forwarder. forwarder or IPFS or IPΞS +// type CloudStore interface { +// Store(*Chunk) +// Deliver(*Chunk) +// Retrieve(*Chunk) +// } type StoreParams struct { ChunkDbPath string @@ -72,70 +77,56 @@ func (self *StoreParams) Init(path string) { self.ChunkDbPath = filepath.Join(path, "chunks") } -// netstore contructor, takes path argument that is used to initialise dbStore, -// the persistent (disk) storage component of LocalStore -// the second argument is the hive, the connection/logistics manager for the node -func NewNetStore(hash SwarmHasher, lstore *LocalStore, cloud CloudStore, params *StoreParams) *NetStore { - return &NetStore{ - hashfunc: hash, - localStore: lstore, - cloud: cloud, - } -} +// // netstore contructor, takes path argument that is used to initialise dbStore, +// // the persistent (disk) storage component of LocalStore +// // the second argument is the hive, the connection/logistics manager for the node +// func NewNetStore(hash SwarmHasher, lstore *LocalStore, cloud CloudStore, params *StoreParams) *NetStore { +// return &NetStore{ +// hashfunc: hash, +// localStore: lstore, +// cloud: cloud, +// } +// } -const ( - // maximum number of peers that a retrieved message is delivered to - requesterCount = 3 -) +// const ( +// // maximum number of peers that a retrieved message is delivered to +// requesterCount = 3 +// ) var ( // timeout interval before retrieval is timed out searchTimeout = 3 * time.Second ) -// store logic common to local and network chunk store requests -// ~ unsafe put in localdb no check if exists no extra copy no hash validation -// the chunk is forced to propagate (Cloud.Store) even if locally found! -// caller needs to make sure if that is wanted -func (self *NetStore) Put(entry *Chunk) { - self.localStore.Put(entry) +// // store logic common to local and network chunk store requests +// // ~ unsafe put in localdb no check if exists no extra copy no hash validation +// // the chunk is forced to propagate (Cloud.Store) even if locally found! +// // caller needs to make sure if that is wanted +// func (self *NetStore) Put(entry *Chunk) { +// self.localStore.Put(entry) - // handle deliveries - if entry.Req != nil { - log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v hit existing request...delivering", entry.Key.Log())) - // closing C signals to other routines (local requests) - // that the chunk is has been retrieved - close(entry.Req.C) - // deliver the chunk to requesters upstream - go self.cloud.Deliver(entry) - } else { - log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v stored locally", entry.Key.Log())) - // handle propagating store requests - // go self.cloud.Store(entry) - go self.cloud.Store(entry) - } -} +// // handle deliveries +// if entry.ReqC != nil { +// log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v hit existing request...delivering", entry.Key.Log())) +// // closing C signals to other routines (local requests) +// // that the chunk is has been retrieved +// close(entry.ReqC) +// // deliver the chunk to requesters upstream +// go self.cloud.Deliver(entry) +// } else { +// log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v stored locally", entry.Key.Log())) +// // handle propagating store requests +// // go self.cloud.Store(entry) +// go self.cloud.Store(entry) +// } +// } -// retrieve logic common for local and network chunk retrieval requests -func (self *NetStore) Get(key Key) (*Chunk, error) { - var err error - chunk, err := self.localStore.Get(key) - if err == nil { - if chunk.Req == nil { - log.Trace(fmt.Sprintf("NetStore.Get: %v found locally", key)) - } else { - log.Trace(fmt.Sprintf("NetStore.Get: %v hit on an existing request", key)) - // no need to launch again - } - return chunk, err - } - // no data and no request status - log.Trace(fmt.Sprintf("NetStore.Get: %v not found locally. open new request", key)) - chunk = NewChunk(key, NewRequestStatus(key)) - self.localStore.memStore.Put(chunk) - go self.cloud.Retrieve(chunk) - return chunk, nil -} +// // retrieve logic common for local and network chunk retrieval requests +// func (self *NetStore) Get(key Key) (*Chunk, error) { +// chunk, _ := self.localStore.GetOrCreateRequest(key) +// go self.cloud.Retrieve(chunk) +// return chunk, nil +// } -// Close netstore -func (self *NetStore) Close() {} +// // Close netstore +// func (self *NetStore) Close() {} diff --git a/swarm/storage/types.go b/swarm/storage/types.go index 833a1b4262..73a4f758d6 100644 --- a/swarm/storage/types.go +++ b/swarm/storage/types.go @@ -166,44 +166,23 @@ func (c KeyCollection) Swap(i, j int) { c[i], c[j] = c[j], c[i] } -// each chunk when first requested opens a record associated with the request -// next time a request for the same chunk arrives, this record is updated -// this request status keeps track of the request ID-s as well as the requesting -// peers and has a channel that is closed when the chunk is retrieved. Multiple -// local callers can wait on this channel (or combined with a timeout, block with a -// select). -type RequestStatus struct { - Key Key - Source Peer - C chan bool - Requesters map[uint64][]interface{} -} - -func NewRequestStatus(key Key) *RequestStatus { - return &RequestStatus{ - Key: key, - Requesters: make(map[uint64][]interface{}), - C: make(chan bool), - } -} - // Chunk also serves as a request object passed to ChunkStores // in case it is a retrieval request, Data is nil and Size is 0 // Note that Size is not the size of the data chunk, which is Data.Size() // but the size of the subtree encoded in the chunk // 0 if request, to be supplied by the dpa type Chunk struct { - Key Key // always - SData []byte // nil if request, to be supplied by dpa - Size int64 // size of the data covered by the subtree encoded in this chunk - Source Peer // peer - C chan bool // to signal data delivery by the dpa - Req *RequestStatus // request Status needed by netStore - dbStored chan bool // never remove a chunk from memStore before it is written to dbStore + Key Key // always + SData []byte // nil if request, to be supplied by dpa + Size int64 // size of the data covered by the subtree encoded in this chunk + //Source Peer // peer + C chan bool // to signal data delivery by the dpa + ReqC chan bool // to signal the request done + dbStored chan bool // never remove a chunk from memStore before it is written to dbStore } -func NewChunk(key Key, rs *RequestStatus) *Chunk { - return &Chunk{Key: key, Req: rs, dbStored: make(chan bool)} +func NewChunk(key Key, reqC chan bool) *Chunk { + return &Chunk{Key: key, ReqC: reqC, dbStored: make(chan bool)} } func (c *Chunk) WaitToStore() { diff --git a/swarm/swarm.go b/swarm/swarm.go index b2349c0fc4..11f5cf25f2 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -49,10 +49,11 @@ type Swarm struct { api *api.Api // high level api layer (fs/manifest) dns api.Resolver // DNS registrar //dbAccess *network.DbAccess // access to local chunk db iterator and storage counter - storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends - dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support + //storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends + dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support //depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage - cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud) + streamer *network.Streamer + //cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud) bzz *network.Bzz // the logistic manager backend chequebook.Backend // simple blockchain Backend privateKey *ecdsa.PrivateKey @@ -114,8 +115,8 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e config.HiveParams.Discovery = true // setup cloud storage internal access layer - self.cloud = &storage.Forwarder{} - self.storage = storage.NewNetStore(hash, self.lstore, self.cloud, config.StoreParams) + //self.cloud = &storage.Forwarder{} + //self.storage = storage.NewNetStore(hash, self.lstore, self.cloud, config.StoreParams) log.Debug(fmt.Sprintf("-> swarm net store shared access layer to Swarm Chunk Store")) nodeid := discover.PubkeyID(crypto.ToECDSAPub(common.FromHex(config.PublicKey))) addr := network.NewAddrFromNodeID(nodeid) @@ -124,10 +125,16 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e UnderlayAddr: addr.UAddr, HiveParams: config.HiveParams, } - self.bzz = network.NewBzz(bzzconfig, to, nil) + + dbAccess := network.NewDbAccess(self.lstore) + self.streamer = network.NewStreamer(to, dbAccess) + network.RegisterOutgoingSyncers(self.streamer, dbAccess) + network.RegisterIncomingSyncers(self.streamer, dbAccess) + + self.bzz = network.NewBzz(bzzconfig, to, nil, self.streamer) // set up DPA, the cloud storage local access layer - dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.storage) + dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.streamer.Retrieve) log.Debug(fmt.Sprintf("-> Local Access to Swarm")) // Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage self.dpa = storage.NewDPA(dpaChunkStore, self.config.ChunkerParams)