diff options
author | ethersphere <thesw@rm.eth> | 2018-06-20 20:06:27 +0800 |
---|---|---|
committer | ethersphere <thesw@rm.eth> | 2018-06-22 03:10:31 +0800 |
commit | e187711c6545487d4cac3701f0f506bb536234e2 (patch) | |
tree | d2f6150f70b84b36e49a449082aeda267b4b9046 /swarm/network/stream/stream.go | |
parent | 574378edb50c907b532946a1d4654dbd6701b20a (diff) | |
download | go-tangerine-e187711c6545487d4cac3701f0f506bb536234e2.tar go-tangerine-e187711c6545487d4cac3701f0f506bb536234e2.tar.gz go-tangerine-e187711c6545487d4cac3701f0f506bb536234e2.tar.bz2 go-tangerine-e187711c6545487d4cac3701f0f506bb536234e2.tar.lz go-tangerine-e187711c6545487d4cac3701f0f506bb536234e2.tar.xz go-tangerine-e187711c6545487d4cac3701f0f506bb536234e2.tar.zst go-tangerine-e187711c6545487d4cac3701f0f506bb536234e2.zip |
swarm: network rewrite merge
Diffstat (limited to 'swarm/network/stream/stream.go')
-rw-r--r-- | swarm/network/stream/stream.go | 739 |
1 files changed, 739 insertions, 0 deletions
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go new file mode 100644 index 000000000..9b4658c51 --- /dev/null +++ b/swarm/network/stream/stream.go @@ -0,0 +1,739 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// 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 <http://www.gnu.org/licenses/>. + +package stream + +import ( + "context" + "fmt" + "math" + "sync" + "time" + + "github.com/ethereum/go-ethereum/metrics" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm/log" + "github.com/ethereum/go-ethereum/swarm/network" + "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" + "github.com/ethereum/go-ethereum/swarm/pot" + "github.com/ethereum/go-ethereum/swarm/state" + "github.com/ethereum/go-ethereum/swarm/storage" +) + +const ( + Low uint8 = iota + Mid + High + Top + PriorityQueue // number of queues + PriorityQueueCap = 32 // queue capacity + HashSize = 32 +) + +// Registry registry for outgoing and incoming streamer constructors +type Registry struct { + api *API + addr *network.BzzAddr + skipCheck bool + clientMu sync.RWMutex + serverMu sync.RWMutex + peersMu sync.RWMutex + serverFuncs map[string]func(*Peer, string, bool) (Server, error) + clientFuncs map[string]func(*Peer, string, bool) (Client, error) + peers map[discover.NodeID]*Peer + delivery *Delivery + intervalsStore state.Store + doRetrieve bool +} + +// RegistryOptions holds optional values for NewRegistry constructor. +type RegistryOptions struct { + SkipCheck bool + DoSync bool + DoRetrieve bool + SyncUpdateDelay time.Duration +} + +// NewRegistry is Streamer constructor +func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, intervalsStore state.Store, options *RegistryOptions) *Registry { + if options == nil { + options = &RegistryOptions{} + } + if options.SyncUpdateDelay <= 0 { + options.SyncUpdateDelay = 15 * time.Second + } + streamer := &Registry{ + addr: addr, + skipCheck: options.SkipCheck, + serverFuncs: make(map[string]func(*Peer, string, bool) (Server, error)), + clientFuncs: make(map[string]func(*Peer, string, bool) (Client, error)), + peers: make(map[discover.NodeID]*Peer), + delivery: delivery, + intervalsStore: intervalsStore, + doRetrieve: options.DoRetrieve, + } + streamer.api = NewAPI(streamer) + delivery.getPeer = streamer.getPeer + streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) { + return NewSwarmChunkServer(delivery.db), nil + }) + streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t string, live bool) (Client, error) { + return NewSwarmSyncerClient(p, delivery.db, false, NewStream(swarmChunkServerStreamName, t, live)) + }) + RegisterSwarmSyncerServer(streamer, db) + RegisterSwarmSyncerClient(streamer, db) + + if options.DoSync { + // latestIntC function ensures that + // - receiving from the in chan is not blocked by processing inside the for loop + // - the latest int value is delivered to the loop after the processing is done + // In context of NeighbourhoodDepthC: + // after the syncing is done updating inside the loop, we do not need to update on the intermediate + // depth changes, only to the latest one + latestIntC := func(in <-chan int) <-chan int { + out := make(chan int, 1) + + go func() { + defer close(out) + + for i := range in { + select { + case <-out: + default: + } + out <- i + } + }() + + return out + } + + go func() { + // wait for kademlia table to be healthy + time.Sleep(options.SyncUpdateDelay) + + kad := streamer.delivery.overlay.(*network.Kademlia) + depthC := latestIntC(kad.NeighbourhoodDepthC()) + addressBookSizeC := latestIntC(kad.AddrCountC()) + + // initial requests for syncing subscription to peers + streamer.updateSyncing() + + for depth := range depthC { + log.Debug("Kademlia neighbourhood depth change", "depth", depth) + + // Prevent too early sync subscriptions by waiting until there are no + // new peers connecting. Sync streams updating will be done after no + // peers are connected for at least SyncUpdateDelay period. + timer := time.NewTimer(options.SyncUpdateDelay) + // Hard limit to sync update delay, preventing long delays + // on a very dynamic network + maxTimer := time.NewTimer(3 * time.Minute) + loop: + for { + select { + case <-maxTimer.C: + // force syncing update when a hard timeout is reached + log.Trace("Sync subscriptions update on hard timeout") + // request for syncing subscription to new peers + streamer.updateSyncing() + break loop + case <-timer.C: + // start syncing as no new peers has been added to kademlia + // for some time + log.Trace("Sync subscriptions update") + // request for syncing subscription to new peers + streamer.updateSyncing() + break loop + case size := <-addressBookSizeC: + log.Trace("Kademlia address book size changed on depth change", "size", size) + // new peers has been added to kademlia, + // reset the timer to prevent early sync subscriptions + if !timer.Stop() { + <-timer.C + } + timer.Reset(options.SyncUpdateDelay) + } + } + timer.Stop() + maxTimer.Stop() + } + }() + } + + return streamer +} + +// RegisterClient registers an incoming streamer constructor +func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, string, bool) (Client, error)) { + r.clientMu.Lock() + defer r.clientMu.Unlock() + + r.clientFuncs[stream] = f +} + +// RegisterServer registers an outgoing streamer constructor +func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, string, bool) (Server, error)) { + r.serverMu.Lock() + defer r.serverMu.Unlock() + + r.serverFuncs[stream] = f +} + +// GetClient accessor for incoming streamer constructors +func (r *Registry) GetClientFunc(stream string) (func(*Peer, string, bool) (Client, error), error) { + r.clientMu.RLock() + defer r.clientMu.RUnlock() + + f := r.clientFuncs[stream] + if f == nil { + return nil, fmt.Errorf("stream %v not registered", stream) + } + return f, nil +} + +// GetServer accessor for incoming streamer constructors +func (r *Registry) GetServerFunc(stream string) (func(*Peer, string, bool) (Server, error), error) { + r.serverMu.RLock() + defer r.serverMu.RUnlock() + + f := r.serverFuncs[stream] + if f == nil { + return nil, fmt.Errorf("stream %v not registered", stream) + } + return f, nil +} + +func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Range, prio uint8) error { + // check if the stream is registered + if _, err := r.GetServerFunc(s.Name); err != nil { + return err + } + + peer := r.getPeer(peerId) + if peer == nil { + return fmt.Errorf("peer not found %v", peerId) + } + + if _, err := peer.getServer(s); err != nil { + if e, ok := err.(*notFoundError); ok && e.t == "server" { + // request subscription only if the server for this stream is not created + log.Debug("RequestSubscription ", "peer", peerId, "stream", s, "history", h) + return peer.Send(&RequestSubscriptionMsg{ + Stream: s, + History: h, + Priority: prio, + }) + } + return err + } + log.Trace("RequestSubscription: already subscribed", "peer", peerId, "stream", s, "history", h) + return nil +} + +// Subscribe initiates the streamer +func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error { + // check if the stream is registered + if _, err := r.GetClientFunc(s.Name); err != nil { + return err + } + + peer := r.getPeer(peerId) + if peer == nil { + return fmt.Errorf("peer not found %v", peerId) + } + + var to uint64 + if !s.Live && h != nil { + to = h.To + } + + err := peer.setClientParams(s, newClientParams(priority, to)) + if err != nil { + return err + } + + if s.Live && h != nil { + if err := peer.setClientParams( + getHistoryStream(s), + newClientParams(getHistoryPriority(priority), h.To), + ); err != nil { + return err + } + } + + msg := &SubscribeMsg{ + Stream: s, + History: h, + Priority: priority, + } + log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h) + + return peer.SendPriority(msg, priority) +} + +func (r *Registry) Unsubscribe(peerId discover.NodeID, s Stream) error { + peer := r.getPeer(peerId) + if peer == nil { + return fmt.Errorf("peer not found %v", peerId) + } + + msg := &UnsubscribeMsg{ + Stream: s, + } + log.Debug("Unsubscribe ", "peer", peerId, "stream", s) + + if err := peer.Send(msg); err != nil { + return err + } + return peer.removeClient(s) +} + +// Quit sends the QuitMsg to the peer to remove the +// stream peer client and terminate the streaming. +func (r *Registry) Quit(peerId discover.NodeID, s Stream) error { + peer := r.getPeer(peerId) + if peer == nil { + log.Debug("stream quit: peer not found", "peer", peerId, "stream", s) + // if the peer is not found, abort the request + return nil + } + + msg := &QuitMsg{ + Stream: s, + } + log.Debug("Quit ", "peer", peerId, "stream", s) + + return peer.Send(msg) +} + +func (r *Registry) Retrieve(chunk *storage.Chunk) error { + return r.delivery.RequestFromPeers(chunk.Addr[:], r.skipCheck) +} + +func (r *Registry) NodeInfo() interface{} { + return nil +} + +func (r *Registry) PeerInfo(id discover.NodeID) interface{} { + return nil +} + +func (r *Registry) Close() error { + return r.intervalsStore.Close() +} + +func (r *Registry) getPeer(peerId discover.NodeID) *Peer { + r.peersMu.RLock() + defer r.peersMu.RUnlock() + + return r.peers[peerId] +} + +func (r *Registry) setPeer(peer *Peer) { + r.peersMu.Lock() + r.peers[peer.ID()] = peer + metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(r.peers))) + r.peersMu.Unlock() +} + +func (r *Registry) deletePeer(peer *Peer) { + r.peersMu.Lock() + delete(r.peers, peer.ID()) + metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(r.peers))) + r.peersMu.Unlock() +} + +func (r *Registry) peersCount() (c int) { + r.peersMu.Lock() + c = len(r.peers) + r.peersMu.Unlock() + return +} + +// Run protocol run function +func (r *Registry) Run(p *network.BzzPeer) error { + sp := NewPeer(p.Peer, r) + r.setPeer(sp) + defer r.deletePeer(sp) + defer close(sp.quit) + defer sp.close() + + if r.doRetrieve { + err := r.Subscribe(p.ID(), NewStream(swarmChunkServerStreamName, "", false), nil, Top) + if err != nil { + return err + } + } + + return sp.Run(sp.HandleMsg) +} + +// updateSyncing subscribes to SYNC streams by iterating over the +// kademlia connections and bins. If there are existing SYNC streams +// and they are no longer required after iteration, request to Quit +// them will be send to appropriate peers. +func (r *Registry) updateSyncing() { + // if overlay in not Kademlia, panic + kad := r.delivery.overlay.(*network.Kademlia) + + // map of all SYNC streams for all peers + // used at the and of the function to remove servers + // that are not needed anymore + subs := make(map[discover.NodeID]map[Stream]struct{}) + r.peersMu.RLock() + for id, peer := range r.peers { + peer.serverMu.RLock() + for stream := range peer.servers { + if stream.Name == "SYNC" { + if _, ok := subs[id]; !ok { + subs[id] = make(map[Stream]struct{}) + } + subs[id][stream] = struct{}{} + } + } + peer.serverMu.RUnlock() + } + r.peersMu.RUnlock() + + // request subscriptions for all nodes and bins + kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(conn network.OverlayConn, bin int) bool { + p := conn.(network.Peer) + log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin)) + + // bin is always less then 256 and it is safe to convert it to type uint8 + stream := NewStream("SYNC", FormatSyncBinKey(uint8(bin)), true) + if streams, ok := subs[p.ID()]; ok { + // delete live and history streams from the map, so that it won't be removed with a Quit request + delete(streams, stream) + delete(streams, getHistoryStream(stream)) + } + err := r.RequestSubscription(p.ID(), stream, NewRange(0, 0), High) + if err != nil { + log.Debug("Request subscription", "err", err, "peer", p.ID(), "stream", stream) + return false + } + return true + }) + + // remove SYNC servers that do not need to be subscribed + for id, streams := range subs { + if len(streams) == 0 { + continue + } + peer := r.getPeer(id) + if peer == nil { + continue + } + for stream := range streams { + log.Debug("Remove sync server", "peer", id, "stream", stream) + err := r.Quit(peer.ID(), stream) + if err != nil && err != p2p.ErrShuttingDown { + log.Error("quit", "err", err, "peer", peer.ID(), "stream", stream) + } + } + } +} + +func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { + peer := protocols.NewPeer(p, rw, Spec) + bzzPeer := network.NewBzzTestPeer(peer, r.addr) + r.delivery.overlay.On(bzzPeer) + defer r.delivery.overlay.Off(bzzPeer) + return r.Run(bzzPeer) +} + +// HandleMsg is the message handler that delegates incoming messages +func (p *Peer) HandleMsg(msg interface{}) error { + switch msg := msg.(type) { + + case *SubscribeMsg: + return p.handleSubscribeMsg(msg) + + case *SubscribeErrorMsg: + return p.handleSubscribeErrorMsg(msg) + + case *UnsubscribeMsg: + return p.handleUnsubscribeMsg(msg) + + case *OfferedHashesMsg: + return p.handleOfferedHashesMsg(msg) + + case *TakeoverProofMsg: + return p.handleTakeoverProofMsg(msg) + + case *WantedHashesMsg: + return p.handleWantedHashesMsg(msg) + + case *ChunkDeliveryMsg: + return p.streamer.delivery.handleChunkDeliveryMsg(p, msg) + + case *RetrieveRequestMsg: + return p.streamer.delivery.handleRetrieveRequestMsg(p, msg) + + case *RequestSubscriptionMsg: + return p.handleRequestSubscription(msg) + + case *QuitMsg: + return p.handleQuitMsg(msg) + + default: + return fmt.Errorf("unknown message type: %T", msg) + } +} + +type server struct { + Server + stream Stream + priority uint8 + currentBatch []byte +} + +// Server interface for outgoing peer Streamer +type Server interface { + SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) + GetData([]byte) ([]byte, error) + Close() +} + +type client struct { + Client + stream Stream + priority uint8 + sessionAt uint64 + to uint64 + next chan error + quit chan struct{} + + intervalsKey string + intervalsStore state.Store +} + +func peerStreamIntervalsKey(p *Peer, s Stream) string { + return p.ID().String() + s.String() +} + +func (c client) AddInterval(start, end uint64) (err error) { + i := &intervals.Intervals{} + err = c.intervalsStore.Get(c.intervalsKey, i) + if err != nil { + return err + } + i.Add(start, end) + return c.intervalsStore.Put(c.intervalsKey, i) +} + +func (c client) NextInterval() (start, end uint64, err error) { + i := &intervals.Intervals{} + err = c.intervalsStore.Get(c.intervalsKey, i) + if err != nil { + return 0, 0, err + } + start, end = i.Next() + return start, end, nil +} + +// Client interface for incoming peer Streamer +type Client interface { + NeedData([]byte) func() + BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) + Close() +} + +func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) { + if c.to > 0 && from >= c.to { + return 0, 0 + } + if c.stream.Live { + return from, 0 + } else if from >= c.sessionAt { + if c.to > 0 { + return from, c.to + } + return from, math.MaxUint64 + } + nextFrom, nextTo, err := c.NextInterval() + if err != nil { + log.Error("next intervals", "stream", c.stream) + return + } + if nextTo > c.to { + nextTo = c.to + } + if nextTo == 0 { + nextTo = c.sessionAt + } + return +} + +func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error { + if tf := c.BatchDone(req.Stream, req.From, hashes, req.Root); tf != nil { + tp, err := tf() + if err != nil { + return err + } + if err := p.SendPriority(tp, c.priority); err != nil { + return err + } + if c.to > 0 && tp.Takeover.End >= c.to { + return p.streamer.Unsubscribe(p.Peer.ID(), req.Stream) + } + return nil + } + // TODO: make a test case for testing if the interval is added when the batch is done + if err := c.AddInterval(req.From, req.To); err != nil { + return err + } + return nil +} + +func (c *client) close() { + select { + case <-c.quit: + default: + close(c.quit) + } + c.Close() +} + +// clientParams store parameters for the new client +// between a subscription and initial offered hashes request handling. +type clientParams struct { + priority uint8 + to uint64 + // signal when the client is created + clientCreatedC chan struct{} +} + +func newClientParams(priority uint8, to uint64) *clientParams { + return &clientParams{ + priority: priority, + to: to, + clientCreatedC: make(chan struct{}), + } +} + +func (c *clientParams) waitClient(ctx context.Context) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-c.clientCreatedC: + return nil + } +} + +func (c *clientParams) clientCreated() { + close(c.clientCreatedC) +} + +// Spec is the spec of the streamer protocol +var Spec = &protocols.Spec{ + Name: "stream", + Version: 4, + MaxMsgSize: 10 * 1024 * 1024, + Messages: []interface{}{ + UnsubscribeMsg{}, + OfferedHashesMsg{}, + WantedHashesMsg{}, + TakeoverProofMsg{}, + SubscribeMsg{}, + RetrieveRequestMsg{}, + ChunkDeliveryMsg{}, + SubscribeErrorMsg{}, + RequestSubscriptionMsg{}, + QuitMsg{}, + }, +} + +func (r *Registry) Protocols() []p2p.Protocol { + return []p2p.Protocol{ + { + Name: Spec.Name, + Version: Spec.Version, + Length: Spec.Length(), + Run: r.runProtocol, + // NodeInfo: , + // PeerInfo: , + }, + } +} + +func (r *Registry) APIs() []rpc.API { + return []rpc.API{ + { + Namespace: "stream", + Version: "3.0", + Service: r.api, + Public: true, + }, + } +} + +func (r *Registry) Start(server *p2p.Server) error { + log.Info("Streamer started") + return nil +} + +func (r *Registry) Stop() error { + return nil +} + +type Range struct { + From, To uint64 +} + +func NewRange(from, to uint64) *Range { + return &Range{ + From: from, + To: to, + } +} + +func (r *Range) String() string { + return fmt.Sprintf("%v-%v", r.From, r.To) +} + +func getHistoryPriority(priority uint8) uint8 { + if priority == 0 { + return 0 + } + return priority - 1 +} + +func getHistoryStream(s Stream) Stream { + return NewStream(s.Name, s.Key, false) +} + +type API struct { + streamer *Registry +} + +func NewAPI(r *Registry) *API { + return &API{ + streamer: r, + } +} + +func (api *API) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error { + return api.streamer.Subscribe(peerId, s, history, priority) +} + +func (api *API) UnsubscribeStream(peerId discover.NodeID, s Stream) error { + return api.streamer.Unsubscribe(peerId, s) +} |