From 03655f7b7886b6a6cae2911b885440f4dd7ab315 Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 9 Jan 2018 15:05:21 +0100 Subject: [PATCH] swarm/storage, swarm/network: light mode request streamers --- swarm/network/lightnode.go | 191 +++++++++++++++++++++++++++++++ swarm/network/streamer.go | 206 +++++++++++++++++++++------------- swarm/network/syncer.go | 197 +++++++++++++++++++++----------- swarm/storage/chunker.go | 19 ++-- swarm/storage/chunker_test.go | 20 ++-- swarm/storage/dpa.go | 2 +- swarm/storage/types.go | 2 +- 7 files changed, 476 insertions(+), 161 deletions(-) create mode 100644 swarm/network/lightnode.go diff --git a/swarm/network/lightnode.go b/swarm/network/lightnode.go new file mode 100644 index 0000000000..ca1463b4d4 --- /dev/null +++ b/swarm/network/lightnode.go @@ -0,0 +1,191 @@ +// Copyright 2017 The go-ethereum Authors +// This file is part of the go-ethereum library.d +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package network + +import ( + "errors" + + "github.com/ethereum/go-ethereum/swarm/storage" +) + +// RemoteReader implements IncomingStreamer +type RemoteSectionReader struct { + db *DbAccess + start uint64 + end uint64 + hashes chan []byte + currentHashes []byte + currentData []byte + quit chan struct{} + root []byte +} + +// NewRemoteReader is the constructor for RemoteReader +func NewRemoteSectionReader(root []byte, db *DbAccess) *RemoteSectionReader { + return &RemoteSectionReader{ + db: db, + root: root, + hashes: make(chan []byte), + quit: make(chan struct{}), + } +} + +func (r *RemoteSectionReader) NeedData(key []byte) func() { + chunk, created := r.db.getOrCreateRequest(storage.Key(key)) + // TODO: we may want to request from this peer anyway even if the request exists + if chunk.ReqC == nil || !created { + return nil + } + return func() {} +} + +func (r *RemoteSectionReader) NextBatch(from uint64) (nextFrom uint64, nextTo uint64) { + return from, r.end +} + +func (r *RemoteSectionReader) BatchDone(s Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) { + r.hashes <- hashes + return nil +} + +func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) { + l := int64(len(b)) + m := int64(len(r.currentData)) + if m > l { + m = l + } + copy(b, r.currentData[:m]) + if m == l { + r.currentData = r.currentData[m:] + return l, nil + } + var end bool + for i := 0; !end && i < len(r.currentHashes); i += HashSize { + hash := r.currentHashes[i : i+HashSize] + chunk, err := r.db.get(hash) + if err != nil { + return n, err + } + m := chunk.Size + if n+m > l { + m = l - n + end = true + } + copy(b[n:], chunk.SData[:m]) + n += int64(m) + } + + for { + select { + case <-r.quit: + return n, errors.New("aborted") + case hashes := <-r.hashes: + var i int + for ; !end && i < len(hashes); i += HashSize { + hash := hashes[i : i+HashSize] + chunk, err := r.db.get(hash) + if err != nil { + return n, err + } + m := chunk.Size + if n+m > l { + m = l - n + end = true + + } + copy(b[n:], chunk.SData[:m]) + n += m + } + hashes = hashes[i:] + } + } + return n, nil +} + +// RemoteSectionServer implements OutgoingStreamer +type RemoteSectionServer struct { + // quit chan struct{} + currentBatch []byte + root []byte + db *DbAccess + r *storage.LazyChunkReader +} + +// NewRemoteReader is the constructor for RemoteReader +func NewRemoteSectionServer(db *DbAccess, r *storage.LazyChunkReader) *RemoteSectionServer { + return &RemoteSectionServer{ + db: db, + r: r, + } +} + +// GetData retrieves the actual chunk from localstore +func (s *RemoteSectionServer) GetData(key []byte) []byte { + chunk, err := s.db.get(storage.Key(key)) + if err != nil { + return nil + } + return chunk.SData +} + +// GetBatch retrieves the next batch of hashes from the dbstore +func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { + if to > from+batchSize { + to = from + batchSize + } + batch := make([]byte, (to-from)*HashSize) + s.r.ReadAt(batch, int64(from)) + s.currentBatch = batch + return batch, from, to, nil, nil +} + +// RegisterRemoteSectionReader registers RemoteSectionReader on light downstream node +func RegisterRemoteSectionReader(s *Streamer, db *DbAccess) { + name := Stream("REMOTE_SECTION") + s.RegisterIncomingStreamer(name, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { + return NewRemoteSectionReader(t, db), nil + }) +} + +// RegisterRemoteSectionServer registers RemoteSectionServer outgoing streamer on +// upstream light server node +func RegisterRemoteSectionServer(s *Streamer, db *DbAccess, rf func([]byte) *storage.LazyChunkReader) { + name := Stream("REMOTE_SECTION") + s.RegisterOutgoingStreamer(name, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { + r := rf(t) + return NewRemoteSectionServer(db, r), nil + }) +} + +// RegisterRemoteDownloader registers RemoteDownloader incoming streamer +// on downstream light node +func RegisterRemoteDownloader(s *Streamer, db *DbAccess) { + name := Stream("REMOTE_DOWNLOADER") + s.RegisterIncomingStreamer(name, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { + return NewRemoteDownloader(t, db), nil + }) +} + +// RegisterRemoteDownloadServer registers RemoteDownloadServer outgoing streamer on +// upstream light server node +func RegisterRemoteDownloadServer(s *Streamer, db *DbAccess, rf func([]byte) *storage.LazyChunkReader) { + name := Stream("REMOTE_DOWNLOADER") + s.RegisterOutgoingStreamer(name, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { + r := rf(t) + return NewRemoteDownloadServer(db, r), nil + }) +} diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index 19f948b4fa..bb6ace62b7 100644 --- a/swarm/network/streamer.go +++ b/swarm/network/streamer.go @@ -34,7 +34,7 @@ import ( const ( HashSize = 32 - Low int = iota + Low uint8 = iota Mid High Top @@ -80,6 +80,7 @@ func (self TakeoverProofMsg) String() string { // SubcribeMsg is the protocol msg for requesting a stream(section) type SubscribeMsg struct { Stream Stream + Key []byte From, To uint64 Priority uint8 // delivered on priority channel } @@ -88,6 +89,7 @@ type SubscribeMsg struct { // stream section type UnsyncedKeysMsg struct { Stream Stream // name of Stream + Key []byte // subtype or key From, To uint64 // peer and db-specific entry count Hashes []byte // stream of hashes (128) *HandoverProof // HandoverProof @@ -114,6 +116,7 @@ func (self UnsyncedKeysMsg) String() string { // offered in UnsyncedKeysMsg downstream peer actually wants sent over type WantedKeysMsg struct { Stream Stream // name of stream + Key []byte // subtype or key Want []byte // bitvector indicating which keys of the batch needed From, To uint64 // next interval offset - empty if not to be continued } @@ -127,8 +130,8 @@ func (self WantedKeysMsg) String() string { type Streamer struct { incomingLock sync.RWMutex outgoingLock sync.RWMutex - outgoing map[Stream]func(*StreamerPeer) (OutgoingStreamer, error) - incoming map[Stream]func(*StreamerPeer) (IncomingStreamer, error) + outgoing map[Stream]func(*StreamerPeer, []byte) (OutgoingStreamer, error) + incoming map[Stream]func(*StreamerPeer, []byte) (IncomingStreamer, error) dbAccess *DbAccess overlay Overlay @@ -138,8 +141,8 @@ type Streamer struct { // NewStreamer is Streamer constructor 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)), + outgoing: make(map[Stream]func(*StreamerPeer, []byte) (OutgoingStreamer, error)), + incoming: make(map[Stream]func(*StreamerPeer, []byte) (IncomingStreamer, error)), dbAccess: dbAccess, overlay: overlay, receiveC: make(chan *ChunkDeliveryMsg, 10), @@ -147,21 +150,21 @@ func NewStreamer(overlay Overlay, dbAccess *DbAccess) *Streamer { } // RegisterIncomingStreamer registers an incoming streamer constructor -func (self *Streamer) RegisterIncomingStreamer(stream Stream, f func(*StreamerPeer) (IncomingStreamer, error)) { +func (self *Streamer) RegisterIncomingStreamer(stream Stream, f func(*StreamerPeer, []byte) (IncomingStreamer, error)) { self.incomingLock.Lock() defer self.incomingLock.Unlock() self.incoming[stream] = f } // RegisterOutgoingStreamer registers an outgoing streamer constructor -func (self *Streamer) RegisterOutgoingStreamer(stream Stream, f func(*StreamerPeer) (OutgoingStreamer, error)) { +func (self *Streamer) RegisterOutgoingStreamer(stream Stream, f func(*StreamerPeer, []byte) (OutgoingStreamer, error)) { self.outgoingLock.Lock() defer self.outgoingLock.Unlock() self.outgoing[stream] = f } // GetIncomingStreamer accessor for incoming streamer constructors -func (self *Streamer) GetIncomingStreamer(stream Stream) (func(*StreamerPeer) (IncomingStreamer, error), error) { +func (self *Streamer) GetIncomingStreamer(stream Stream) (func(*StreamerPeer, []byte) (IncomingStreamer, error), error) { self.incomingLock.RLock() defer self.incomingLock.RUnlock() f := self.incoming[stream] @@ -172,7 +175,7 @@ func (self *Streamer) GetIncomingStreamer(stream Stream) (func(*StreamerPeer) (I } // GetOutgoingStreamer accessor for incoming streamer constructors -func (self *Streamer) GetOutgoingStreamer(stream Stream) (func(*StreamerPeer) (OutgoingStreamer, error), error) { +func (self *Streamer) GetOutgoingStreamer(stream Stream) (func(*StreamerPeer, []byte) (OutgoingStreamer, error), error) { self.outgoingLock.RLock() defer self.outgoingLock.RUnlock() f := self.outgoing[stream] @@ -190,19 +193,30 @@ func (self *Streamer) PeerInfo(id discover.NodeID) interface{} { return nil } +type outgoingStreamer struct { + OutgoingStreamer + priority uint8 + currentBatch []byte +} + // OutgoingStreamer interface for outgoing peer Streamer type OutgoingStreamer interface { - CurrentBatch() []byte SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) GetData([]byte) []byte - Priority() int +} + +type incomingStreamer struct { + IncomingStreamer + priority uint8 + quit chan struct{} + next chan struct{} } // IncomingStreamer interface for incoming peer Streamer type IncomingStreamer interface { NextBatch(uint64) (uint64, uint64) NeedData([]byte) func() - Priority() int + BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) } // StreamerPeer is the Peer extention for the streaming protocol @@ -214,8 +228,8 @@ type StreamerPeer struct { dbAccess *DbAccess outgoingLock sync.RWMutex incomingLock sync.RWMutex - outgoing map[Stream]OutgoingStreamer - incoming map[Stream]IncomingStreamer + outgoing map[Stream]*outgoingStreamer + incoming map[Stream]*incomingStreamer quit chan struct{} } @@ -232,10 +246,10 @@ type StreamerPeer struct { // NewStreamerPeer is the constructor for StreamerPeer func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer { self := &StreamerPeer{ - pq: pq.New(PriorityQueue, PriorityQueueCap), + pq: pq.New(int(PriorityQueue), PriorityQueueCap), streamer: streamer, - outgoing: make(map[Stream]OutgoingStreamer), - incoming: make(map[Stream]IncomingStreamer), + outgoing: make(map[Stream]*outgoingStreamer), + incoming: make(map[Stream]*incomingStreamer), quit: make(chan struct{}), } ctx, cancel := context.WithCancel(context.Background()) @@ -247,38 +261,41 @@ func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer { return self } +// RetrieveRequestMsg is the protocol msg for chunk retrieve requests 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 + deliveryC chan *storage.Chunk + batchC chan []byte + db *DbAccess + currentLen uint64 } -func RegisterRequestStreamer(streamer *Streamer, dbAccess *DbAccess) { - streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer) (OutgoingStreamer, error) { - return NewRetrieveRequestStreamer(dbAccess), nil +// RegisterRequestStreamer registers outgoing and incoming streamers for request handling +func RegisterRequestStreamer(streamer *Streamer, db *DbAccess) { + streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer, t []byte) (OutgoingStreamer, error) { + return NewRetrieveRequestStreamer(db), nil }) - streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer) (IncomingStreamer, error) { - return NewIncomingSwarmSyncer(Top, nil, p, dbAccess, nil) + streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { + return NewIncomingSwarmSyncer(nil, p, db, nil) }) } -func NewRetrieveRequestStreamer(dbAccess *DbAccess) *RetrieveRequestStreamer { +// NewRetrieveRequestStreamer is RetrieveRequestStreamer constructor +func NewRetrieveRequestStreamer(db *DbAccess) *RetrieveRequestStreamer { s := &RetrieveRequestStreamer{ deliveryC: make(chan *storage.Chunk), batchC: make(chan []byte), - dbAccess: dbAccess, + db: db, } go s.processDeliveries() return s } +// processDeliveries handles delivered chunk hashes func (s *RetrieveRequestStreamer) processDeliveries() { var hashes []byte for { @@ -291,28 +308,21 @@ func (s *RetrieveRequestStreamer) processDeliveries() { } } -func (s *RetrieveRequestStreamer) CurrentBatch() []byte { - return s.currentBatch -} - +// SetNextBatch 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 } +// GetData retrives chunk data from db store func (s *RetrieveRequestStreamer) GetData(key []byte) []byte { - chunk, _ := s.dbAccess.get(storage.Key(key)) + chunk, _ := s.db.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 { @@ -321,7 +331,7 @@ func (self *StreamerPeer) handleRetrieveRequestMsg(req *RetrieveRequestMsg) erro if err != nil { return err } - streamer := s.(*RetrieveRequestStreamer) + streamer := s.OutgoingStreamer.(*RetrieveRequestStreamer) if chunk.ReqC != nil { if created { if err := self.streamer.Retrieve(chunk); err != nil { @@ -349,10 +359,16 @@ func (self *StreamerPeer) handleRetrieveRequestMsg(req *RetrieveRequestMsg) erro return nil } +// Retrieve sends a chunk retrieve request to 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 {}) + self.overlay.EachConn(chunk.Key[:], 255, func(p OverlayConn, po int, nn bool) bool { + sp := p.(*StreamerPeer) + // TODO: skip light nodes that do not accept retrieve requests + sp.SendPriority(&RetrieveRequestMsg{ + Key: chunk.Key[:], + }, Top) + return false + }) return nil } @@ -383,7 +399,7 @@ func (self *Streamer) processReceivedChunks() { } } -func (self *StreamerPeer) getOutgoingStreamer(s Stream) (OutgoingStreamer, error) { +func (self *StreamerPeer) getOutgoingStreamer(s Stream) (*outgoingStreamer, error) { self.outgoingLock.RLock() defer self.outgoingLock.RUnlock() streamer := self.outgoing[s] @@ -393,7 +409,7 @@ func (self *StreamerPeer) getOutgoingStreamer(s Stream) (OutgoingStreamer, error return streamer, nil } -func (self *StreamerPeer) getIncomingStreamer(s Stream) (IncomingStreamer, error) { +func (self *StreamerPeer) getIncomingStreamer(s Stream) (*incomingStreamer, error) { self.incomingLock.RLock() defer self.incomingLock.RUnlock() streamer := self.incoming[s] @@ -403,44 +419,59 @@ func (self *StreamerPeer) getIncomingStreamer(s Stream) (IncomingStreamer, error return streamer, nil } -func (self *StreamerPeer) setOutgoingStreamer(s Stream, o OutgoingStreamer) error { +func (self *StreamerPeer) setOutgoingStreamer(s Stream, o OutgoingStreamer, priority uint8) (*outgoingStreamer, error) { self.outgoingLock.Lock() defer self.outgoingLock.Unlock() if self.outgoing[s] != nil { - return fmt.Errorf("stream %v already registered", s) + return nil, fmt.Errorf("stream %v already registered", s) } - self.outgoing[s] = o - return nil + os := &outgoingStreamer{ + OutgoingStreamer: o, + priority: priority, + } + self.outgoing[s] = os + return os, nil } -func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer) error { +func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer, priority uint8) error { self.incomingLock.Lock() defer self.incomingLock.Unlock() if self.incoming[s] != nil { return fmt.Errorf("stream %v already registered", s) } - self.incoming[s] = i + next := make(chan struct{}, 1) + self.incoming[s] = &incomingStreamer{ + IncomingStreamer: i, + priority: priority, + next: next, + } + next <- struct{}{} // this is to allow wantedKeysMsg before first batch arrives return nil } // Subscribe initiates the streamer -func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error { +func (self *StreamerPeer) Subscribe(s Stream, t []byte, from, to uint64, priority uint8) error { f, err := self.streamer.GetIncomingStreamer(s) if err != nil { return err } - is, err := f(self) + is, err := f(self, t) if err != nil { return err } - self.setIncomingStreamer(s, is) + err = self.setIncomingStreamer(s, is, priority) + if err != nil { + return err + } + msg := &SubscribeMsg{ Stream: s, + Key: t, From: from, To: to, - Priority: uint8(is.Priority()), + Priority: priority, } - self.SendPriority(msg, is.Priority()) + self.SendPriority(msg, priority) return nil } @@ -449,14 +480,16 @@ func (self *StreamerPeer) handleSubscribeMsg(req *SubscribeMsg) error { if err != nil { return err } - s, err := f(self) + s, err := f(self, req.Key) if err != nil { return err } - if err := self.setOutgoingStreamer(req.Stream, s); err != nil { + key := string(req.Stream) + string(req.Key) + os, err := self.setOutgoingStreamer(Stream(key), s, req.Priority) + if err != nil { return nil } - self.SendUnsyncedKeys(s, req.From, req.To, int(req.Priority)) + go self.SendUnsyncedKeys(os, req.From, req.To) return nil } @@ -485,11 +518,17 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(req *UnsyncedKeysMsg) error { }(wait) } } - // go func() { - // wg.Wait() - // msg := s.TakeoverProof(req.Stream, req.From, req.Hashes, req.Root) - // self.Send(msg, s.Priority()) - // }() + go func() { + wg.Wait() + if tf := s.BatchDone(req.Stream, req.From, hashes, req.Root); tf != nil { + tp, err := tf() + if err != nil { + return + } + self.SendPriority(tp, s.priority) + } + s.next <- struct{}{} + }() // only send wantedKeysMsg if all missing chunks of the previous batch arrived // except from, to := s.NextBatch(req.To) @@ -502,7 +541,14 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(req *UnsyncedKeysMsg) error { From: from, To: to, } - self.SendPriority(msg, s.Priority()) + go func() { + select { + case <-s.next: + case <-s.quit: + return + } + self.SendPriority(msg, s.priority) + }() return nil } @@ -514,9 +560,9 @@ func (self *StreamerPeer) handleWantedKeysMsg(req *WantedKeysMsg) error { if err != nil { return err } - hashes := s.CurrentBatch() + hashes := s.currentBatch // launch in go routine since GetBatch blocks until new hashes arrive - go self.SendUnsyncedKeys(s, req.From, req.To, s.Priority()) + go self.SendUnsyncedKeys(s, req.From, req.To) l := len(hashes) / HashSize want, err := bv.NewFromBytes(req.Want, l) if err != nil { @@ -531,7 +577,7 @@ func (self *StreamerPeer) handleWantedKeysMsg(req *WantedKeysMsg) error { } chunk := storage.NewChunk(hash, nil) chunk.SData = data - if err := self.Deliver(chunk, s.Priority()); err != nil { + if err := self.Deliver(chunk, s.priority); err != nil { return err } } @@ -549,32 +595,33 @@ func (self *StreamerPeer) handleTakeoverProofMsg(req *TakeoverProofMsg) error { } // Deliver sends a storeRequestMsg protocol message to the peer -func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority int) error { +func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority uint8) error { msg := &ChunkDeliveryMsg{ Key: chunk.Key, SData: chunk.SData, } - return self.pq.Push(nil, msg, priority) + return self.pq.Push(nil, msg, int(priority)) } // Deliver sends a storeRequestMsg protocol message to the peer -func (self *StreamerPeer) SendPriority(msg interface{}, priority int) error { - return self.pq.Push(nil, msg, priority) +func (self *StreamerPeer) SendPriority(msg interface{}, priority uint8) error { + return self.pq.Push(nil, msg, int(priority)) } // UnsyncedKeys sends UnsyncedKeysMsg protocol msg -func (self *StreamerPeer) SendUnsyncedKeys(s OutgoingStreamer, f, t uint64, priority int) error { +func (self *StreamerPeer) SendUnsyncedKeys(s *outgoingStreamer, f, t uint64) error { hashes, from, to, proof, err := s.SetNextBatch(f, t) if err != nil { return err } + s.currentBatch = hashes msg := &UnsyncedKeysMsg{ HandoverProof: proof, Hashes: hashes, From: from, To: to, } - return self.SendPriority(msg, s.Priority()) + return self.SendPriority(msg, s.priority) } // StreamerSpec is the spec of the streamer protocol. @@ -595,10 +642,13 @@ 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), - }) + // autosubscribe to request handler to serve request only for non-light nodes + // sp.handleSubscribeMsg(&SubscribeMsg{ + // Stream: retrieveRequestStream, + // Priority: uint8(Top), + // }) + // subscribe to request handling ; only with non-light nodes + sp.Subscribe(retrieveRequestStream, nil, 0, 0, Top) defer close(sp.quit) return sp.Run(sp.HandleMsg) } diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index ed87e57870..d0da14f78d 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -70,19 +70,24 @@ func (self *DbAccess) put(chunk *storage.Chunk) { // * live request delivery with or without checkback // * (live/non-live historical) chunk syncing per proximity bin type OutgoingSwarmSyncer struct { - po uint8 - db *DbAccess - sessionAt uint64 - currentBatch []byte - priority int + po uint8 + db *DbAccess + sessionAt uint64 + start uint64 } // NewOutgoingSwarmSyncer is contructor for OutgoingSwarmSyncer -func NewOutgoingSwarmSyncer(po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error) { +func NewOutgoingSwarmSyncer(live bool, po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error) { + sessionAt := db.currentBucketStorageIndex(po) + var start uint64 + if live { + start = sessionAt + } self := &OutgoingSwarmSyncer{ po: po, db: db, - sessionAt: db.currentBucketStorageIndex(po), + sessionAt: sessionAt, + start: start, } return self, nil } @@ -90,20 +95,22 @@ func NewOutgoingSwarmSyncer(po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error const maxPO = 32 func RegisterOutgoingSyncers(streamer *Streamer, db *DbAccess) { - for po := uint8(0); po < maxPO; po++ { - stream := Stream(fmt.Sprintf("SYNC-%02d-live", po)) - streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { - return NewOutgoingSwarmSyncer(po, db) - }) - stream = Stream(fmt.Sprintf("SYNC-%02d-history", po)) - streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { - return NewOutgoingSwarmSyncer(po, db) - }) - // stream = Stream(fmt.Sprintf("SYNC-%02d-delete", po)) - // streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { - // return NewOutgoingProvableSwarmSyncer(po, db) - // }) - } + stream := Stream("SYNC") + streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { + syncType, po := parseSyncLabel(t) + switch syncType { + case "LIVE": + return NewOutgoingSwarmSyncer(true, po, db) + case "HISTORY": + return NewOutgoingSwarmSyncer(false, po, db) + default: + return nil, errors.New("invalid sync type") + } + }) + // stream = Stream(fmt.Sprintf("SYNC-%02d-delete", po)) + // streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { + // return NewOutgoingProvableSwarmSyncer(po, db) + // }) } // GetSection retrieves the actual chunk from localstore @@ -115,18 +122,13 @@ func (self *OutgoingSwarmSyncer) GetData(key []byte) []byte { return chunk.SData } -func (self *OutgoingSwarmSyncer) CurrentBatch() []byte { - return self.currentBatch -} - -func (self *OutgoingSwarmSyncer) Priority() int { - return self.priority -} - // GetBatch retrieves the next batch of hashes from the dbstore func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { var batch []byte i := 0 + if from == 0 { + from = self.start + } err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool { batch = append(batch, key[:]...) i++ @@ -136,17 +138,15 @@ func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64, if err != nil { return nil, 0, 0, nil, err } - self.currentBatch = batch log.Debug("Swarm batch", "po", self.po, "len", i, "from", from, "to", to) return batch, from, to, nil, nil } // IncomingSwarmSyncer type IncomingSwarmSyncer struct { - priority int sessionAt uint64 nextC chan struct{} - intervals []uint64 + intervals *Intervals sessionRoot storage.Key sessionReader storage.LazySectionReader retrieveC chan *storage.Chunk @@ -159,9 +159,8 @@ type IncomingSwarmSyncer struct { } // NewIncomingSwarmSyncer is a contructor for provable data exchange syncer -func NewIncomingSwarmSyncer(priority int, intervals []uint64, p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) { +func NewIncomingSwarmSyncer(intervals *Intervals, p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) { self := &IncomingSwarmSyncer{ - priority: priority, intervals: intervals, dbAccess: dbAccess, chunker: chunker, @@ -169,10 +168,6 @@ func NewIncomingSwarmSyncer(priority int, intervals []uint64, p Peer, dbAccess * return self, nil } -func (s *IncomingSwarmSyncer) Priority() int { - return s.priority -} - // // NewIncomingProvableSwarmSyncer is a contructor for provable data exchange syncer // func NewIncomingProvableSwarmSyncer(po int, priority int, index uint64, sessionAt uint64, intervals []uint64, sessionRoot storage.Key, chunker *storage.PyramidChunker, store storage.ChunkStore, p Peer) *IncomingSwarmSyncer { // retrieveC := make(storage.Chunk, chunksCap) @@ -195,30 +190,91 @@ func (s *IncomingSwarmSyncer) Priority() int { // return self // } -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(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(Mid, nil, p, nil, nil) - }) - // stream = fmt.Sprintf("SYNC-%02d-delete", po) - // streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { - // intervals := loadIntervals(p, po, true) - // return NewIncomingSwarmSyncer(po, Mid, sessionAt, intervals, p) - // }) +type Syncer struct { + intervals map[string][]uint64 +} + +func NewSyncer() *Syncer { + return &Syncer{} +} + +type Intervals struct { + syncer *Syncer + key []byte +} + +func (s *Intervals) load() error { + return s.syncer.load(s.key) +} + +func (s *Intervals) save() error { + return s.syncer.save(s.key) +} + +func (s *Intervals) get() []uint64 { + return s.syncer.get(s.key) +} + +func (s *Intervals) set(v []uint64) { + s.syncer.set(s.key, v) +} + +func (s *Syncer) NewIntervals(key []byte) *Intervals { + return &Intervals{ + syncer: s, + key: key, } } +func newSyncLabel(typ string, po uint8) []byte { + t := []byte(typ) + t = append(t, byte(po)) + return t +} + +func parseSyncLabel(t []byte) (string, uint8) { + l := len(t) - 1 + return sstring(t[:l]), uint8(t[l]) +} + +// StartSyncing is called on the StreamerPeer to start the syncing process +// the idea is that it is called only after kademlia is close to healthy +func StartSyncing(s *StreamerPeer, po uint8, nn bool) { + lastPO := po + if nn { + lastPO = maxPO + } + for i := po; i <= lastPO; i++ { + s.Subscribe(Stream("SYNC"), newSyncLabel("LIVE", po), 0, 0, High) + s.Subscribe(Stream("SYNC"), newSyncLabel("HISTORY", po), 0, 0, Mid) + } +} + +func RegisterIncomingSyncers(streamer *Streamer, syncer *Syncer, db *DbAccess) { + stream := Stream("SYNC") + streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { + syncType, po := parseSyncLabel(t) + switch syncType { + case "LIVE": + return NewIncomingSwarmSyncer(nil, p, nil, nil) + case "HISTORY": + intervals := syncer.NewIntervals(t) + return NewIncomingSwarmSyncer(intervals, p, nil, nil) + } + return nil, fmt.Errorf("unknown sync type %q", syncType) + }) + // stream = fmt.Sprintf("SYNC-%02d-delete", po) + // streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { + // intervals := loadIntervals(p, po, true) + // return NewIncomingSwarmSyncer(po, Mid, sessionAt, intervals, p) + // }) +} + // NeedData func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) { - chunk, created := self.dbAccess.getOrCreateRequest(key) + chunk, _ := self.dbAccess.getOrCreateRequest(key) // TODO: we may want to request from this peer anyway even if the request exists - if chunk.ReqC == nil || !created { + if chunk.ReqC == nil { return nil } // create request and wait until the chunk data arrives and is stored @@ -227,28 +283,37 @@ func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) { // NextBatch adjusts the indexes by inspecting the intervals func (self *IncomingSwarmSyncer) NextBatch(from uint64) (nextFrom uint64, nextTo uint64) { - if self.intervals[0] >= self.sessionAt { // live syncing + intervals := self.intervals.get() + if intervals[0] >= self.sessionAt { // live syncing nextFrom = from - self.intervals[1] = from + intervals[1] = from } else if from >= self.sessionAt { // history sync complete - self.intervals = nil - } else if len(self.intervals) > 2 && from >= self.intervals[2] { // filled a gap in the intervals - self.intervals = append(self.intervals[:1], self.intervals[3:]...) - nextFrom = self.intervals[1] - if len(self.intervals) > 2 { - nextTo = self.intervals[2] + intervals = nil + } else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals + intervals = append(intervals[:1], intervals[3:]...) + nextFrom = intervals[1] + if len(intervals) > 2 { + nextTo = intervals[2] } else { nextTo = self.sessionAt } } else { nextFrom = from - self.intervals[1] = from + intervals[1] = from nextTo = self.sessionAt } + self.intervals.set(intervals) return nextFrom, nextTo } -// +// BatchDone +func (self *IncomingSwarmSyncer) BatchDone(s Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) { + if self.chunker != nil { + return func() (*TakeoverProof, error) { return self.TakeoverProof(s, from, hashes, root) } + } + return nil +} + func (self *IncomingSwarmSyncer) TakeoverProof(s Stream, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) { // for provable syncer currentRoot is non-zero length if self.chunker != nil { diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index be142f227b..918c0fc45b 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -302,16 +302,18 @@ type LazyChunkReader struct { chunkSize int64 // inherit from chunker branches int64 // inherit from chunker hashSize int64 // inherit from chunker + depth int } // implements the Joiner interface -func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) LazySectionReader { +func (self *TreeChunker) Join(key Key, chunkC chan *Chunk, depth int) LazySectionReader { return &LazyChunkReader{ key: key, chunkC: chunkC, chunkSize: self.chunkSize, branches: self.branches, hashSize: self.hashSize, + depth: depth, } } @@ -358,8 +360,13 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { depth++ } wg := sync.WaitGroup{} + length := int64(len(b)) + for d := 0; d < self.depth; d++ { + off *= self.chunkSize + length *= self.chunkSize + } wg.Add(1) - go self.join(b, off, off+int64(len(b)), depth, treeSize/self.branches, self.chunk, &wg, errC, quitC) + go self.join(b, off, off+length, depth, treeSize/self.branches, self.chunk, &wg, errC, quitC) go func() { wg.Wait() close(errC) @@ -379,18 +386,14 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) { defer parentWg.Done() - // return NewDPA(&LocalStore{}) - - // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - // find appropriate block level - for chunk.Size < treeSize && depth > 0 { + for chunk.Size < treeSize && depth > self.depth { treeSize /= self.branches depth-- } // leaf chunk found - if depth == 0 { + if depth == self.depth { extra := 8 + eoff - int64(len(chunk.SData)) if extra > 0 { eoff -= extra diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go index 5cb1125bf5..a0f82245d3 100644 --- a/swarm/storage/chunker_test.go +++ b/swarm/storage/chunker_test.go @@ -163,7 +163,7 @@ func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Ch } }() - reader := chunker.Join(key, chunkC) + reader := chunker.Join(key, chunkC, 0) return reader } @@ -489,7 +489,8 @@ func BenchmarkJoin_4(t *testing.B) { benchmarkJoin(10000, t) } func BenchmarkJoin_5(t *testing.B) { benchmarkJoin(100000, t) } func BenchmarkJoin_6(t *testing.B) { benchmarkJoin(1000000, t) } func BenchmarkJoin_7(t *testing.B) { benchmarkJoin(10000000, t) } -func BenchmarkJoin_8(t *testing.B) { benchmarkJoin(100000000, t) } + +// func BenchmarkJoin_8(t *testing.B) { benchmarkJoin(100000000, t) } func BenchmarkSplitTreeSHA3_2(t *testing.B) { benchmarkSplitTreeSHA3(100, t) } func BenchmarkSplitTreeSHA3_2h(t *testing.B) { benchmarkSplitTreeSHA3(500, t) } @@ -500,7 +501,8 @@ func BenchmarkSplitTreeSHA3_4h(t *testing.B) { benchmarkSplitTreeSHA3(50000, t) func BenchmarkSplitTreeSHA3_5(t *testing.B) { benchmarkSplitTreeSHA3(100000, t) } func BenchmarkSplitTreeSHA3_6(t *testing.B) { benchmarkSplitTreeSHA3(1000000, t) } func BenchmarkSplitTreeSHA3_7(t *testing.B) { benchmarkSplitTreeSHA3(10000000, t) } -func BenchmarkSplitTreeSHA3_8(t *testing.B) { benchmarkSplitTreeSHA3(100000000, t) } + +// func BenchmarkSplitTreeSHA3_8(t *testing.B) { benchmarkSplitTreeSHA3(100000000, t) } func BenchmarkSplitTreeBMT_2(t *testing.B) { benchmarkSplitTreeBMT(100, t) } func BenchmarkSplitTreeBMT_2h(t *testing.B) { benchmarkSplitTreeBMT(500, t) } @@ -511,7 +513,8 @@ func BenchmarkSplitTreeBMT_4h(t *testing.B) { benchmarkSplitTreeBMT(50000, t) } func BenchmarkSplitTreeBMT_5(t *testing.B) { benchmarkSplitTreeBMT(100000, t) } func BenchmarkSplitTreeBMT_6(t *testing.B) { benchmarkSplitTreeBMT(1000000, t) } func BenchmarkSplitTreeBMT_7(t *testing.B) { benchmarkSplitTreeBMT(10000000, t) } -func BenchmarkSplitTreeBMT_8(t *testing.B) { benchmarkSplitTreeBMT(100000000, t) } + +// func BenchmarkSplitTreeBMT_8(t *testing.B) { benchmarkSplitTreeBMT(100000000, t) } func BenchmarkSplitPyramidSHA3_2(t *testing.B) { benchmarkSplitPyramidSHA3(100, t) } func BenchmarkSplitPyramidSHA3_2h(t *testing.B) { benchmarkSplitPyramidSHA3(500, t) } @@ -522,7 +525,8 @@ func BenchmarkSplitPyramidSHA3_4h(t *testing.B) { benchmarkSplitPyramidSHA3(5000 func BenchmarkSplitPyramidSHA3_5(t *testing.B) { benchmarkSplitPyramidSHA3(100000, t) } func BenchmarkSplitPyramidSHA3_6(t *testing.B) { benchmarkSplitPyramidSHA3(1000000, t) } func BenchmarkSplitPyramidSHA3_7(t *testing.B) { benchmarkSplitPyramidSHA3(10000000, t) } -func BenchmarkSplitPyramidSHA3_8(t *testing.B) { benchmarkSplitPyramidSHA3(100000000, t) } + +// func BenchmarkSplitPyramidSHA3_8(t *testing.B) { benchmarkSplitPyramidSHA3(100000000, t) } func BenchmarkSplitPyramidBMT_2(t *testing.B) { benchmarkSplitPyramidBMT(100, t) } func BenchmarkSplitPyramidBMT_2h(t *testing.B) { benchmarkSplitPyramidBMT(500, t) } @@ -533,7 +537,8 @@ func BenchmarkSplitPyramidBMT_4h(t *testing.B) { benchmarkSplitPyramidBMT(50000, func BenchmarkSplitPyramidBMT_5(t *testing.B) { benchmarkSplitPyramidBMT(100000, t) } func BenchmarkSplitPyramidBMT_6(t *testing.B) { benchmarkSplitPyramidBMT(1000000, t) } func BenchmarkSplitPyramidBMT_7(t *testing.B) { benchmarkSplitPyramidBMT(10000000, t) } -func BenchmarkSplitPyramidBMT_8(t *testing.B) { benchmarkSplitPyramidBMT(100000000, t) } + +// func BenchmarkSplitPyramidBMT_8(t *testing.B) { benchmarkSplitPyramidBMT(100000000, t) } func BenchmarkAppendPyramid_2(t *testing.B) { benchmarkAppendPyramid(100, 1000, t) } func BenchmarkAppendPyramid_2h(t *testing.B) { benchmarkAppendPyramid(500, 1000, t) } @@ -543,7 +548,8 @@ func BenchmarkAppendPyramid_4h(t *testing.B) { benchmarkAppendPyramid(50000, 100 func BenchmarkAppendPyramid_5(t *testing.B) { benchmarkAppendPyramid(1000000, 1000, t) } func BenchmarkAppendPyramid_6(t *testing.B) { benchmarkAppendPyramid(1000000, 1000, t) } func BenchmarkAppendPyramid_7(t *testing.B) { benchmarkAppendPyramid(10000000, 1000, t) } -func BenchmarkAppendPyramid_8(t *testing.B) { benchmarkAppendPyramid(100000000, 1000, t) } + +// func BenchmarkAppendPyramid_8(t *testing.B) { benchmarkAppendPyramid(100000000, 1000, t) } // go test -timeout 20m -cpu 4 -bench=./swarm/storage -run no // If you dont add the timeout argument above .. the benchmark will timeout and dump diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index 7ef8a80a99..5f0a95470e 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -92,7 +92,7 @@ func NewDPA(store ChunkStore, params *ChunkerParams) *DPA { // Chunk retrieval blocks on netStore requests with a timeout so reader will // report error if retrieval of chunks within requested range time out. func (self *DPA) Retrieve(key Key) LazySectionReader { - return self.Chunker.Join(key, self.retrieveC) + return self.Chunker.Join(key, self.retrieveC, 0) } // Public API. Main entry point for document storage directly. Used by the diff --git a/swarm/storage/types.go b/swarm/storage/types.go index 73a4f758d6..956b8ddd8b 100644 --- a/swarm/storage/types.go +++ b/swarm/storage/types.go @@ -273,7 +273,7 @@ type Joiner interface { The chunks are not meant to be validated by the chunker when joining. This is because it is left to the DPA to decide which sources are trusted. */ - Join(key Key, chunkC chan *Chunk) LazySectionReader + Join(key Key, chunkC chan *Chunk, depth int) LazySectionReader } type Chunker interface {