From e2ec5801ed5321d3f9b174bcd4567724c2a16eca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jano=C5=A1=20Gulja=C5=A1?= Date: Wed, 21 Mar 2018 18:35:28 +0100 Subject: [PATCH] cmd/swarm, swarm/network: dynamic sync streams update - NewRegistry creates a goroutine that updates sync streams based on changes in kademlia. It calls updateSyncing method to create new and quit unintended sync streams. To early subscriptions are prevented by waiting until no new peers are added for a configured period of time SyncUpdateDelay. - Implements two new methods on Kademlia: - NeighbourhoodDepthC which returns a channel that returns kademlia neighbourhood depth on each change - AddressBookSizeC which returns address book size when it changes This methods are required for making a decision when to request sync stream subscriptions update in a dynamic network. - To avoid the parsing of stream strings to distinguish the SYNC ones, Registry now uses Stream structure as the key for clients, servers and clientParams maps. This required that the Stream must be comparable and Key field is changed from []byte to string. The consequence is that key is changed to string in all function signatures that accepted it. Also, new functions FormatSyncBinKey and ParseSyncBinKey are introduced to explicitly handle the parsing and formating of bin numbers that are used as Key values in Stream. - Implements RegistryOptions that holds optional values for stream Registry initialization. The number of arguments for NewRegistry is a bit high with adding another option SyncUpdateDelay, and optional and required ones are separated with RegistryOptions. - Range now has constructor NewRange for better initialization and also has the String method for better logging. - Registry now has the Quit method that sends QuitMsg to the peer to terminate the stream client on its side. This is required to terminate not needed sync streams when kademlia table changes by adding new peers. - Changes the RequestSubscription to skip the request if the stream is already registered. This ensures that no request subscriptions are made if the server already exists. - New SyncUpdateDelay cli flag is added to control the delay of syncing update. Default is set to 15s. --- .travis.yml | 11 -- cmd/swarm/config.go | 12 ++ cmd/swarm/main.go | 6 + swarm/api/config.go | 3 + swarm/network/kademlia.go | 62 +++++-- swarm/network/light/lightnode.go | 190 -------------------- swarm/network/stream/common_test.go | 22 +-- swarm/network/stream/delivery.go | 2 +- swarm/network/stream/delivery_test.go | 33 ++-- swarm/network/stream/intervals_test.go | 26 +-- swarm/network/stream/messages.go | 17 +- swarm/network/stream/peer.go | 61 +++---- swarm/network/stream/stream.go | 233 +++++++++++++++++++++---- swarm/network/stream/streamer_test.go | 205 +++++++++++++++++----- swarm/network/stream/syncer.go | 32 +++- swarm/network/stream/syncer_test.go | 2 +- swarm/swarm.go | 6 +- 17 files changed, 542 insertions(+), 381 deletions(-) delete mode 100644 swarm/network/light/lightnode.go diff --git a/.travis.yml b/.travis.yml index b3757ff7d9..cade11700f 100644 --- a/.travis.yml +++ b/.travis.yml @@ -3,17 +3,6 @@ go_import_path: github.com/ethereum/go-ethereum sudo: false matrix: include: - - os: linux - dist: trusty - sudo: required - go: 1.8.x - script: - - sudo modprobe fuse - - sudo chmod 666 /dev/fuse - - sudo chown root:$USER /etc/fuse.conf - - go run build/ci.go install - - go run build/ci.go test -coverage - - os: linux dist: trusty sudo: required diff --git a/cmd/swarm/config.go b/cmd/swarm/config.go index 2bafe97228..f29c18d7c0 100644 --- a/cmd/swarm/config.go +++ b/cmd/swarm/config.go @@ -24,6 +24,7 @@ import ( "reflect" "strconv" "strings" + "time" "unicode" cli "gopkg.in/urfave/cli.v1" @@ -66,6 +67,7 @@ const ( SWARM_ENV_SWAP_ENABLE = "SWARM_SWAP_ENABLE" SWARM_ENV_SWAP_API = "SWARM_SWAP_API" SWARM_ENV_SYNC_ENABLE = "SWARM_SYNC_ENABLE" + SWARM_ENV_SYNC_UPDATE_DELAY = "SWARM_ENV_SYNC_UPDATE_DELAY" SWARM_ENV_ENS_API = "SWARM_ENS_API" SWARM_ENV_ENS_ADDR = "SWARM_ENS_ADDR" SWARM_ENV_CORS = "SWARM_CORS" @@ -200,6 +202,10 @@ func cmdLineOverride(currentConfig *bzzapi.Config, ctx *cli.Context) *bzzapi.Con currentConfig.SyncEnabled = true } + if d := ctx.GlobalDuration(SwarmSyncUpdateDelay.Name); d > 0 { + currentConfig.SyncUpdateDelay = d + } + currentConfig.SwapApi = ctx.GlobalString(SwarmSwapAPIFlag.Name) if currentConfig.SwapEnabled && currentConfig.SwapApi == "" { utils.Fatalf(SWARM_ERR_SWAP_SET_NO_API) @@ -293,6 +299,12 @@ func envVarsOverride(currentConfig *bzzapi.Config) (config *bzzapi.Config) { } } + if v := os.Getenv(SWARM_ENV_SYNC_UPDATE_DELAY); v != "" { + if d, err := time.ParseDuration(v); err != nil { + currentConfig.SyncUpdateDelay = d + } + } + if swapapi := os.Getenv(SWARM_ENV_SWAP_API); swapapi != "" { currentConfig.SwapApi = swapapi } diff --git a/cmd/swarm/main.go b/cmd/swarm/main.go index ea59c2ea42..057b032ce0 100644 --- a/cmd/swarm/main.go +++ b/cmd/swarm/main.go @@ -106,6 +106,11 @@ var ( Usage: "Swarm Syncing enabled (default true)", EnvVar: SWARM_ENV_SYNC_ENABLE, } + SwarmSyncUpdateDelay = cli.DurationFlag{ + Name: "sync-update-delay", + Usage: "Duration for sync subscriptions update after no new peers are added (default 15s)", + EnvVar: SWARM_ENV_SYNC_UPDATE_DELAY, + } EnsAPIFlag = cli.StringSliceFlag{ Name: "ens-api", Usage: "ENS API endpoint for a TLD and with contract address, can be repeated, format [tld:][contract-addr@]url", @@ -356,6 +361,7 @@ Remove corrupt entries from a local chunk database. SwarmSwapEnabledFlag, SwarmSwapAPIFlag, SwarmSyncEnabledFlag, + SwarmSyncUpdateDelay, SwarmListenAddrFlag, SwarmPortFlag, SwarmAccountFlag, diff --git a/swarm/api/config.go b/swarm/api/config.go index 6bb4d435d7..de58f1d4c8 100644 --- a/swarm/api/config.go +++ b/swarm/api/config.go @@ -21,6 +21,7 @@ import ( "fmt" "os" "path/filepath" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/contracts/ens" @@ -57,6 +58,7 @@ type Config struct { NetworkId uint64 SwapEnabled bool SyncEnabled bool + SyncUpdateDelay time.Duration PssEnabled bool ResourceEnabled bool SwapApi string @@ -83,6 +85,7 @@ func NewConfig() (self *Config) { NetworkId: network.NetworkID, SwapEnabled: false, SyncEnabled: true, + SyncUpdateDelay: 15 * time.Second, PssEnabled: true, ResourceEnabled: true, SwapApi: "", diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index 4bb1184b75..e1888ca85c 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -88,6 +88,9 @@ type Kademlia struct { addrs *pot.Pot // pots container for known peer addresses conns *pot.Pot // pots container for live peer connections depth uint8 // stores the last current depth of saturation + nDepth int // stores the last neighbourhood depth + nDepthC chan int // returned by DepthC function to signal neighbourhood depth change + addrCountC chan int // returned by AddrCountC function to signal peer count change } // NewKademlia creates a Kademlia table for base address addr @@ -198,6 +201,10 @@ func (k *Kademlia) Register(peers []OverlayAddr) error { } size++ } + // send new address count value only if there are new addresses + if k.addrCountC != nil && size-known > 0 { + k.addrCountC <- k.addrs.Size() + } // log.Trace(fmt.Sprintf("%x registered %v peers, %v known, total: %v", k.BaseAddr()[:4], size, known, k.addrs.Size())) return nil } @@ -296,6 +303,10 @@ func (k *Kademlia) On(p OverlayConn) (uint8, bool) { k.addrs, _, _, _ = pot.Swap(k.addrs, p, pof, func(v pot.Val) pot.Val { return e }) + // send new address count value only if the peer is inserted + if k.addrCountC != nil { + k.addrCountC <- k.addrs.Size() + } } log.Trace(k.string()) // calculate if depth of saturation changed @@ -305,9 +316,38 @@ func (k *Kademlia) On(p OverlayConn) (uint8, bool) { changed = true k.depth = depth } + if k.nDepthC != nil { + nDepth := k.neighbourhoodDepth() + if nDepth != k.nDepth { + k.nDepth = nDepth + k.nDepthC <- nDepth + } + } return k.depth, changed } +// NeighbourhoodDepthC returns the channel that sends a new kademlia +// neighbourhood depth on each change. +// Not receiving from the returned channel will block On function +// when the neighbourhood depth is changed. +func (k *Kademlia) NeighbourhoodDepthC() <-chan int { + if k.nDepthC == nil { + k.nDepthC = make(chan int) + } + return k.nDepthC +} + +// AddrCountC returns the channel that sends a new +// address count value on each change. +// Not receiving from the returned channel will block Register function +// when address count value changes. +func (k *Kademlia) AddrCountC() <-chan int { + if k.addrCountC == nil { + k.addrCountC = make(chan int) + } + return k.addrCountC +} + // Off removes a peer from among live peers func (k *Kademlia) Off(p OverlayConn) { k.lock.Lock() @@ -326,6 +366,10 @@ func (k *Kademlia) Off(p OverlayConn) { // v cannot be nil, but no need to check return nil }) + // send new address count value only if the peer is deleted + if k.addrCountC != nil { + k.addrCountC <- k.addrs.Size() + } } } @@ -333,23 +377,17 @@ func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(con k.lock.RLock() defer k.lock.RUnlock() - var i int var startPo int var endPo int - kadDepth := int(k.depth) + kadDepth := k.neighbourhoodDepth() k.conns.EachBin(base, pof, o, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool { + if startPo > 0 && endPo != k.MaxProxDisplay { + startPo = endPo + 1 + } if po < kadDepth { endPo = po - if i > 0 { - startPo = endPo + 1 - } - } else if endPo < kadDepth || endPo == 0 { - if po == 0 && kadDepth == 0 { - startPo = endPo - } else { - startPo = endPo + 1 - } + } else { endPo = k.MaxProxDisplay } @@ -358,10 +396,8 @@ func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(con return eachBinFunc(val.(*entry).conn(), bin) }) } - i++ return true }) - } // EachConn is an iterator with args (base, po, f) applies f to each live peer diff --git a/swarm/network/light/lightnode.go b/swarm/network/light/lightnode.go deleted file mode 100644 index ad62a88425..0000000000 --- a/swarm/network/light/lightnode.go +++ /dev/null @@ -1,190 +0,0 @@ -// Copyright 2018 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 light - -import ( - "errors" - - "github.com/ethereum/go-ethereum/swarm/network/stream" - "github.com/ethereum/go-ethereum/swarm/storage" -) - -// RemoteReader implements IncomingStreamer -type RemoteSectionReader struct { - db *storage.DBAPI - 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 *storage.DBAPI) *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() { - select { - case <-chunk.ReqC: - case <-r.quit: - } - } -} - -func (r *RemoteSectionReader) BatchDone(s stream.Stream, from uint64, hashes []byte, root []byte) func() (*stream.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 += stream.HashSize { - hash := r.currentHashes[i : i+stream.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 - } - - for { - select { - case <-r.quit: - return n, errors.New("aborted") - case hashes := <-r.hashes: - var i int - for ; !end && i < len(hashes); i += stream.HashSize { - hash := hashes[i : i+stream.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:] - } - } -} - -func (r *RemoteSectionReader) Close() {} - -// RemoteSectionServer implements OutgoingStreamer -type RemoteSectionServer struct { - // quit chan struct{} - root []byte - db *storage.DBAPI - r *storage.LazyChunkReader -} - -// NewRemoteReader is the constructor for RemoteReader -func NewRemoteSectionServer(db *storage.DBAPI, r *storage.LazyChunkReader) *RemoteSectionServer { - return &RemoteSectionServer{ - db: db, - r: r, - } -} - -// GetData retrieves the actual chunk from localstore -func (s *RemoteSectionServer) GetData(key []byte) ([]byte, error) { - chunk, err := s.db.Get(storage.Key(key)) - if err != nil { - return nil, err - } - return chunk.SData, nil -} - -// GetBatch retrieves the next batch of hashes from the dbstore -func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *stream.HandoverProof, error) { - if to > from+stream.BatchSize { - to = from + stream.BatchSize - } - batch := make([]byte, (to-from)*stream.HashSize) - s.r.ReadAt(batch, int64(from)) - return batch, from, to, nil, nil -} - -func (s *RemoteSectionServer) Close() {} - -// RegisterRemoteSectionReader registers RemoteSectionReader on light downstream node -func RegisterRemoteSectionReader(s *stream.Registry, db *storage.DBAPI) { - s.RegisterClientFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte, live bool) (stream.Client, error) { - return NewRemoteSectionReader(t, db), nil - }) -} - -// RegisterRemoteSectionServer registers RemoteSectionServer outgoing streamer on -// upstream light server node -func RegisterRemoteSectionServer(s *stream.Registry, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) { - s.RegisterServerFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte, live bool) (stream.Server, error) { - r := rf(t) - return NewRemoteSectionServer(db, r), nil - }) -} - -// RegisterRemoteDownloader registers RemoteDownloader incoming streamer -// on downstream light node -// func RegisterRemoteDownloader(s *Streamer, db *storage.DBAPI) { -// s.RegisterIncomingStreamer("REMOTE_DOWNLOADER", func(p *stream.Peer, t []byte) (IncomingStreamer, error) { -// return NewRemoteDownloader(t, db), nil -// }) -// } -// -// // RegisterRemoteDownloadServer registers RemoteDownloadServer outgoing streamer on -// // upstream light server node -// func RegisterRemoteDownloadServer(s *Streamer, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) { -// s.RegisterOutgoingStreamer("REMOTE_DOWNLOADER", func(p *stream.Peer, t []byte) (OutgoingStreamer, error) { -// r := rf(t) -// return NewRemoteDownloadServer(db, r), nil -// }) -// } diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 9c84d9e270..998b6adf85 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -78,7 +78,11 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) { db := storage.NewDBAPI(store) delivery := NewDelivery(kad, db) deliveries[id] = delivery - r := NewRegistry(addr, delivery, db, state.NewMemStore(), defaultSkipCheck, false, false) + r := NewRegistry(addr, delivery, db, state.NewMemStore(), &RegistryOptions{ + SkipCheck: defaultSkipCheck, + }) + RegisterSwarmSyncerServer(r, db) + RegisterSwarmSyncerClient(r, db) go func() { waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id)) }() @@ -107,7 +111,9 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora db := storage.NewDBAPI(localStore) delivery := NewDelivery(to, db) - streamer := NewRegistry(addr, delivery, db, state.NewMemStore(), defaultSkipCheck, false, false) + streamer := NewRegistry(addr, delivery, db, state.NewMemStore(), &RegistryOptions{ + SkipCheck: defaultSkipCheck, + }) teardown := func() { streamer.Close() removeDataDir() @@ -289,19 +295,13 @@ func (r *TestExternalRegistry) EnableNotifications(peerId discover.NodeID, s Str // with testClient and testServer. type testExternalClient struct { - t []byte - // wait0 chan bool - // batchDone chan bool hashes chan []byte db *storage.DBAPI enableNotificationsC chan struct{} } -func newTestExternalClient(t []byte, db *storage.DBAPI) *testExternalClient { +func newTestExternalClient(db *storage.DBAPI) *testExternalClient { return &testExternalClient{ - t: t, - // wait0: make(chan bool), - // batchDone: make(chan bool), hashes: make(chan []byte), db: db, enableNotificationsC: make(chan struct{}), @@ -328,14 +328,14 @@ func (c *testExternalClient) Close() {} const testExternalServerBatchSize = 10 type testExternalServer struct { - t []byte + t string keyFunc func(key []byte, index uint64) sessionAt uint64 maxKeys uint64 streamer *TestExternalRegistry } -func newTestExternalServer(t []byte, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer { +func newTestExternalServer(t string, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer { if keyFunc == nil { keyFunc = binary.BigEndian.PutUint64 } diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 545b72c1e4..33449780a4 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -128,7 +128,7 @@ type RetrieveRequestMsg struct { func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) error { log.Debug("received request", "peer", sp.ID(), "hash", req.Key) - s, err := sp.getServer(NewStream(swarmChunkServerStreamName, nil, false)) + s, err := sp.getServer(NewStream(swarmChunkServerStreamName, "", false)) if err != nil { return err } diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 690ae707dd..16273d15af 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -87,11 +87,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { peer := streamer.getPeer(peerID) peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: NewStream(swarmChunkServerStreamName, nil, false), - History: &Range{ - From: 0, - To: 0, - }, + Stream: NewStream(swarmChunkServerStreamName, "", false), + History: NewRange(0, 0), Priority: Top, }) @@ -139,11 +136,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { peer := streamer.getPeer(peerID) peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: NewStream(swarmChunkServerStreamName, nil, false), - History: &Range{ - From: 0, - To: 0, - }, + Stream: NewStream(swarmChunkServerStreamName, "", false), + History: NewRange(0, 0), Priority: Top, }) @@ -175,7 +169,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { From: 0, // TODO: why is this 32??? To: 32, - Stream: NewStream(swarmChunkServerStreamName, nil, false), + Stream: NewStream(swarmChunkServerStreamName, "", false), }, Peer: peerID, }, @@ -228,7 +222,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) { + streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) { return &testClient{ t: t, }, nil @@ -236,8 +230,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { peerID := tester.IDs[0] - stream := NewStream("foo", nil, true) - err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top) + stream := NewStream("foo", "", true) + err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -261,11 +255,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { { Code: 4, Msg: &SubscribeMsg{ - Stream: stream, - History: &Range{ - From: 5, - To: 8, - }, + Stream: stream, + History: NewRange(5, 8), Priority: Top, }, Peer: peerID, @@ -392,7 +383,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck ctx, cancel := context.WithTimeout(ctx, 1*time.Second) defer cancel() sid := sim.IDs[j+1] - return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top) + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, "", false), NewRange(0, 0), Top) }) if err != nil { return err @@ -566,7 +557,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip ctx, cancel := context.WithTimeout(ctx, 1*time.Second) defer cancel() sid := sim.IDs[j+1] // the upstream peer's id - return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top) + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, "", false), NewRange(0, 0), Top) }) if err != nil { break diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index 25cdfe917f..4d05b866e0 100644 --- a/swarm/network/stream/intervals_test.go +++ b/swarm/network/stream/intervals_test.go @@ -50,12 +50,14 @@ func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, er db := storage.NewDBAPI(store) delivery := NewDelivery(kad, db) deliveries[id] = delivery - r := NewRegistry(addr, delivery, db, state.NewMemStore(), defaultSkipCheck, false, false) - - r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Client, error) { - return newTestExternalClient(t, db), nil + r := NewRegistry(addr, delivery, db, state.NewMemStore(), &RegistryOptions{ + SkipCheck: defaultSkipCheck, }) - r.RegisterServerFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Server, error) { + + r.RegisterClientFunc(externalStreamName, func(p *Peer, t string, live bool) (Client, error) { + return newTestExternalClient(db), nil + }) + r.RegisterServerFunc(externalStreamName, func(p *Peer, t string, live bool) (Server, error) { return newTestExternalServer(t, externalStreamSessionAt, externalStreamMaxKeys, nil), nil }) @@ -67,8 +69,8 @@ func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, er func TestIntervals(t *testing.T) { testIntervals(t, true, nil) - testIntervals(t, false, &Range{From: 9, To: 26}) - testIntervals(t, true, &Range{From: 9, To: 26}) + testIntervals(t, false, NewRange(9, 26)) + testIntervals(t, true, NewRange(9, 26)) } func testIntervals(t *testing.T, live bool, history *Range) { @@ -143,7 +145,7 @@ func testIntervals(t *testing.T, live bool, history *Range) { ctx, cancel := context.WithTimeout(ctx, 100*time.Second) defer cancel() - err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, nil, live), history, Top) + err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, "", live), history, Top) if err != nil { return err } @@ -164,7 +166,7 @@ func testIntervals(t *testing.T, live bool, history *Range) { // live stream liveHashesChan := make(chan []byte) - liveSubscription, err := client.Subscribe(ctx, "stream", liveHashesChan, "getHashes", sid, NewStream(externalStreamName, nil, true)) + liveSubscription, err := client.Subscribe(ctx, "stream", liveHashesChan, "getHashes", sid, NewStream(externalStreamName, "", true)) if err != nil { return } @@ -173,7 +175,7 @@ func testIntervals(t *testing.T, live bool, history *Range) { i := externalStreamSessionAt // we have subscribed, enable notifications - err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, nil, true)) + err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, "", true)) if err != nil { return } @@ -211,7 +213,7 @@ func testIntervals(t *testing.T, live bool, history *Range) { // history stream historyHashesChan := make(chan []byte) - historySubscription, err := client.Subscribe(ctx, "stream", historyHashesChan, "getHashes", sid, NewStream(externalStreamName, nil, false)) + historySubscription, err := client.Subscribe(ctx, "stream", historyHashesChan, "getHashes", sid, NewStream(externalStreamName, "", false)) if err != nil { return } @@ -227,7 +229,7 @@ func testIntervals(t *testing.T, live bool, history *Range) { } // we have subscribed, enable notifications - err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, nil, false)) + err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, "", false)) if err != nil { return } diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 3b3de367f5..8fabe2c3bc 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -31,13 +31,13 @@ type Stream struct { // Name is used for Client and Server functions identification. Name string // Key is the name of specific stream data. - Key []byte + Key string // Live defines whether the stream delivers only new data // for the specific stream. Live bool } -func NewStream(name string, key []byte, live bool) Stream { +func NewStream(name string, key string, live bool) Stream { return Stream{ Name: name, Key: key, @@ -51,7 +51,7 @@ func (s Stream) String() string { if s.Live { t = "l" } - return fmt.Sprintf("%s|%x|%s", s.Name, s.Key, t) + return fmt.Sprintf("%s|%s|%s", s.Name, s.Key, t) } // SubcribeMsg is the protocol msg for requesting a stream(section) @@ -148,8 +148,15 @@ type UnsubscribeMsg struct { } func (p *Peer) handleUnsubscribeMsg(req *UnsubscribeMsg) error { - p.removeServer(req.Stream) - return nil + return p.removeServer(req.Stream) +} + +type QuitMsg struct { + Stream Stream +} + +func (p *Peer) handleQuitMsg(req *QuitMsg) error { + return p.removeClient(req.Stream) } // OfferedHashesMsg is the protocol msg for offering to hand over a diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 0bc079b1cc..f715fc0a17 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -52,12 +52,12 @@ type Peer struct { pq *pq.PriorityQueue serverMu sync.RWMutex clientMu sync.RWMutex // protects both clients and clientParams - servers map[string]*server - clients map[string]*client + servers map[Stream]*server + clients map[Stream]*client // clientParams map keeps required client arguments // that are set on Registry.Subscribe and used // on creating a new client in offered hashes handler. - clientParams map[string]*clientParams + clientParams map[Stream]*clientParams quit chan struct{} } @@ -67,9 +67,9 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer { Peer: peer, pq: pq.New(int(PriorityQueue), PriorityQueueCap), streamer: streamer, - servers: make(map[string]*server), - clients: make(map[string]*client), - clientParams: make(map[string]*clientParams), + servers: make(map[Stream]*server), + clients: make(map[Stream]*client), + clientParams: make(map[Stream]*clientParams), quit: make(chan struct{}), } ctx, cancel := context.WithCancel(context.Background()) @@ -128,9 +128,9 @@ func (p *Peer) getServer(s Stream) (*server, error) { p.serverMu.RLock() defer p.serverMu.RUnlock() - server := p.servers[s.String()] + server := p.servers[s] if server == nil { - return nil, fmt.Errorf("server '%v' not provided to peer %v", s, p.ID()) + return nil, newNotFoundError("server", s) } return server, nil } @@ -139,16 +139,15 @@ func (p *Peer) setServer(s Stream, o Server, priority uint8) (*server, error) { p.serverMu.Lock() defer p.serverMu.Unlock() - sk := s.String() - if p.servers[sk] != nil { - return nil, fmt.Errorf("server %v already registered", sk) + if p.servers[s] != nil { + return nil, fmt.Errorf("server %s already registered", s) } os := &server{ Server: o, stream: s, priority: priority, } - p.servers[sk] = os + p.servers[s] = os return os, nil } @@ -156,28 +155,26 @@ func (p *Peer) removeServer(s Stream) error { p.serverMu.Lock() defer p.serverMu.Unlock() - sk := s.String() - server, ok := p.servers[sk] + server, ok := p.servers[s] if !ok { return newNotFoundError("server", s) } server.Close() - delete(p.servers, sk) + delete(p.servers, s) return nil } func (p *Peer) getClient(ctx context.Context, s Stream) (c *client, err error) { var params *clientParams - sk := s.String() func() { p.clientMu.RLock() defer p.clientMu.RUnlock() - c = p.clients[sk] + c = p.clients[s] if c != nil { return } - params = p.clientParams[sk] + params = p.clientParams[s] }() if c != nil { return c, nil @@ -193,7 +190,7 @@ func (p *Peer) getClient(ctx context.Context, s Stream) (c *client, err error) { p.clientMu.RLock() defer p.clientMu.RUnlock() - c = p.clients[sk] + c = p.clients[s] if c != nil { return c, nil } @@ -201,12 +198,10 @@ func (p *Peer) getClient(ctx context.Context, s Stream) (c *client, err error) { } func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created bool, err error) { - sk := s.String() - p.clientMu.Lock() defer p.clientMu.Unlock() - c = p.clients[sk] + c = p.clients[s] if c != nil { return c, false, nil } @@ -273,7 +268,7 @@ func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created boo intervalsStore: p.streamer.intervalsStore, intervalsKey: intervalsKey, } - p.clients[sk] = c + p.clients[s] = c cp.clientCreated() // unblock all possible getClient calls that are waiting next <- nil // this is to allow wantedKeysMsg before first batch arrives return c, true, nil @@ -283,7 +278,7 @@ func (p *Peer) removeClient(s Stream) error { p.clientMu.Lock() defer p.clientMu.Unlock() - client, ok := p.clients[s.String()] + client, ok := p.clients[s] if !ok { return newNotFoundError("client", s) } @@ -295,19 +290,18 @@ func (p *Peer) setClientParams(s Stream, params *clientParams) error { p.clientMu.Lock() defer p.clientMu.Unlock() - sk := s.String() - if p.clients[sk] != nil { - return fmt.Errorf("client %v already exists", sk) + if p.clients[s] != nil { + return fmt.Errorf("client %s already exists", s) } - if p.clientParams[sk] != nil { - return fmt.Errorf("client params %v already set", sk) + if p.clientParams[s] != nil { + return fmt.Errorf("client params %s already set", s) } - p.clientParams[sk] = params + p.clientParams[s] = params return nil } func (p *Peer) getClientParams(s Stream) (*clientParams, error) { - params := p.clientParams[s.String()] + params := p.clientParams[s] if params == nil { return nil, fmt.Errorf("client params '%v' not provided to peer %v", s, p.ID()) } @@ -315,12 +309,11 @@ func (p *Peer) getClientParams(s Stream) (*clientParams, error) { } func (p *Peer) removeClientParams(s Stream) error { - sk := s.String() - _, ok := p.clientParams[sk] + _, ok := p.clientParams[s] if !ok { return newNotFoundError("client params", s) } - delete(p.clientParams, sk) + delete(p.clientParams, s) return nil } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 1386abdcb2..d5e5927195 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -53,43 +53,126 @@ type Registry struct { clientMu sync.RWMutex serverMu sync.RWMutex peersMu sync.RWMutex - serverFuncs map[string]func(*Peer, []byte, bool) (Server, error) - clientFuncs map[string]func(*Peer, []byte, bool) (Client, error) + 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, skipCheck, doSync, doRetrieve bool) *Registry { +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: skipCheck, - serverFuncs: make(map[string]func(*Peer, []byte, bool) (Server, error)), - clientFuncs: make(map[string]func(*Peer, []byte, bool) (Client, error)), + 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: doRetrieve, + doRetrieve: options.DoRetrieve, } streamer.api = NewAPI(streamer) delivery.getPeer = streamer.getPeer - streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ []byte, _ bool) (Server, error) { + streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) { return NewSwarmChunkServer(delivery.db), nil }) - streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ []byte, _ bool) (Client, error) { + streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ string, _ bool) (Client, error) { return NewSwarmSyncerClient(p, delivery.db, nil) }) RegisterSwarmSyncerServer(streamer, db) RegisterSwarmSyncerClient(streamer, db) - if doSync { - go func() { - // this is a temporary workaround to wait for kademlia table to be healthy - time.Sleep(30 * time.Second) + 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) - streamer.startSyncing() + 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) + + // initial requests for syncing subscription to peers + streamer.updateSyncing() + + kad := streamer.delivery.overlay.(*network.Kademlia) + depthC := latestIntC(kad.NeighbourhoodDepthC()) + addressBookSizeC := latestIntC(kad.AddrCountC()) + + 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() + } }() } @@ -97,7 +180,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i } // RegisterClient registers an incoming streamer constructor -func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte, bool) (Client, error)) { +func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, string, bool) (Client, error)) { r.clientMu.Lock() defer r.clientMu.Unlock() @@ -105,7 +188,7 @@ func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte, bool) } // RegisterServer registers an outgoing streamer constructor -func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte, bool) (Server, error)) { +func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, string, bool) (Server, error)) { r.serverMu.Lock() defer r.serverMu.Unlock() @@ -113,7 +196,7 @@ func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte, bool) } // GetClient accessor for incoming streamer constructors -func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte, bool) (Client, error), error) { +func (r *Registry) GetClientFunc(stream string) (func(*Peer, string, bool) (Client, error), error) { r.clientMu.RLock() defer r.clientMu.RUnlock() @@ -125,7 +208,7 @@ func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte, bool) (Clie } // GetServer accessor for incoming streamer constructors -func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Server, error), error) { +func (r *Registry) GetServerFunc(stream string) (func(*Peer, string, bool) (Server, error), error) { r.serverMu.RLock() defer r.serverMu.RUnlock() @@ -138,7 +221,7 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Serv func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Range, prio uint8) error { // check if the stream is registered - if _, err := r.GetClientFunc(s.Name); err != nil { + if _, err := r.GetServerFunc(s.Name); err != nil { return err } @@ -147,13 +230,20 @@ func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Rang return fmt.Errorf("peer not found %v", peerId) } - msg := &RequestSubscriptionMsg{ - Stream: s, - History: h, - Priority: prio, + 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.Debug("RequestSubscription ", "peer", peerId, "stream", s, "history", h) - return peer.Send(msg) + log.Trace("RequestSubscription: already subscribed", "peer", peerId, "stream", s, "history", h) + return nil } // Subscribe initiates the streamer @@ -214,6 +304,24 @@ func (r *Registry) Unsubscribe(peerId discover.NodeID, s Stream) error { 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.Key[:], r.skipCheck) } @@ -265,7 +373,7 @@ func (r *Registry) Run(p *network.BzzPeer) error { defer sp.close() if r.doRetrieve { - err := r.Subscribe(p.ID(), NewStream(swarmChunkServerStreamName, nil, false), nil, Top) + err := r.Subscribe(p.ID(), NewStream(swarmChunkServerStreamName, "", false), nil, Top) if err != nil { return err } @@ -274,22 +382,70 @@ func (r *Registry) Run(p *network.BzzPeer) error { return sp.Run(sp.HandleMsg) } -func (r *Registry) startSyncing() { - // panic freely +// 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)) - stream := NewStream("SYNC", []byte{uint8(bin)}, true) - err := r.RequestSubscription(p.ID(), stream, &Range{}, Top) + // 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), Top) if err != nil { - log.Error("request subscription", "err", err, "peer", p.ID(), "stream", stream) + log.Error("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 { + log.Error("quit", "err", err, "peer", peer.ID(), "stream", stream) + } + } + } } func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { @@ -331,6 +487,9 @@ func (p *Peer) HandleMsg(msg interface{}) error { case *RequestSubscriptionMsg: return p.handleRequestSubscription(msg) + case *QuitMsg: + return p.handleQuitMsg(msg) + default: return fmt.Errorf("unknown message type: %T", msg) } @@ -490,6 +649,7 @@ var Spec = &protocols.Spec{ ChunkDeliveryMsg{}, SubscribeErrorMsg{}, RequestSubscriptionMsg{}, + QuitMsg{}, }, } @@ -530,6 +690,17 @@ 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 diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index 44175b254f..dae4dbfdc4 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -32,8 +32,8 @@ func TestStreamerSubscribe(t *testing.T) { t.Fatal(err) } - stream := NewStream("foo", nil, true) - err = streamer.Subscribe(tester.IDs[0], stream, &Range{From: 0, To: 0}, Top) + stream := NewStream("foo", "", true) + err = streamer.Subscribe(tester.IDs[0], stream, NewRange(0, 0), Top) if err == nil || err.Error() != "stream foo not registered" { t.Fatalf("Expected error %v, got %v", "stream foo not registered", err) } @@ -48,14 +48,14 @@ var ( ) type testClient struct { - t []byte + t string wait0 chan bool wait2 chan bool batchDone chan bool receivedHashes map[string][]byte } -func newTestClient(t []byte) *testClient { +func newTestClient(t string) *testClient { return &testClient{ t: t, wait0: make(chan bool), @@ -87,10 +87,10 @@ func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*Takeo func (self *testClient) Close() {} type testServer struct { - t []byte + t string } -func newTestServer(t []byte) *testServer { +func newTestServer(t string) *testServer { return &testServer{ t: t, } @@ -114,14 +114,14 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) { + streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) { return newTestClient(t), nil }) peerID := tester.IDs[0] - stream := NewStream("foo", nil, true) - err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top) + stream := NewStream("foo", "", true) + err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -133,11 +133,8 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { { Code: 4, Msg: &SubscribeMsg{ - Stream: stream, - History: &Range{ - From: 5, - To: 8, - }, + Stream: stream, + History: NewRange(5, 8), Priority: Top, }, Peer: peerID, @@ -210,9 +207,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { t.Fatal(err) } - stream := NewStream("foo", nil, false) + stream := NewStream("foo", "", false) - streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { + streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t), nil }) @@ -224,11 +221,8 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { { Code: 4, Msg: &SubscribeMsg{ - Stream: stream, - History: &Range{ - From: 5, - To: 8, - }, + Stream: stream, + History: NewRange(5, 8), Priority: Top, }, Peer: peerID, @@ -280,9 +274,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) { t.Fatal(err) } - stream := NewStream("foo", nil, true) + stream := NewStream("foo", "", true) - streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { + streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t), nil }) @@ -346,11 +340,11 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { + streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t), nil }) - stream := NewStream("bar", nil, true) + stream := NewStream("bar", "", true) peerID := tester.IDs[0] @@ -360,11 +354,8 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { { Code: 4, Msg: &SubscribeMsg{ - Stream: stream, - History: &Range{ - From: 5, - To: 8, - }, + Stream: stream, + History: NewRange(5, 8), Priority: Top, }, Peer: peerID, @@ -393,9 +384,9 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) { t.Fatal(err) } - stream := NewStream("foo", nil, true) + stream := NewStream("foo", "", true) - streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { + streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return &testServer{ t: t, }, nil @@ -409,11 +400,8 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) { { Code: 4, Msg: &SubscribeMsg{ - Stream: stream, - History: &Range{ - From: 5, - To: 8, - }, + Stream: stream, + History: NewRange(5, 8), Priority: Top, }, Peer: peerID, @@ -423,7 +411,7 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) { { Code: 1, Msg: &OfferedHashesMsg{ - Stream: NewStream("foo", nil, false), + Stream: NewStream("foo", "", false), HandoverProof: &HandoverProof{ Handover: &Handover{}, }, @@ -461,18 +449,18 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { t.Fatal(err) } - stream := NewStream("foo", nil, true) + stream := NewStream("foo", "", true) var tc *testClient - streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) { + streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) { tc = newTestClient(t) return tc, nil }) peerID := tester.IDs[0] - err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top) + err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -483,11 +471,8 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { { Code: 4, Msg: &SubscribeMsg{ - Stream: stream, - History: &Range{ - From: 5, - To: 8, - }, + Stream: stream, + History: NewRange(5, 8), Priority: Top, }, Peer: peerID, @@ -555,3 +540,131 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { } } + +func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) { + tester, streamer, _, teardown, err := newStreamerTester(t) + defer teardown() + if err != nil { + t.Fatal(err) + } + + streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { + return newTestServer(t), nil + }) + + peerID := tester.IDs[0] + + stream := NewStream("foo", "", true) + err = streamer.RequestSubscription(peerID, stream, NewRange(5, 8), Top) + if err != nil { + t.Fatalf("Expected no error, got %v", err) + } + + err = tester.TestExchanges( + p2ptest.Exchange{ + Label: "RequestSubscription message", + Expects: []p2ptest.Expect{ + { + Code: 8, + Msg: &RequestSubscriptionMsg{ + Stream: stream, + History: NewRange(5, 8), + Priority: Top, + }, + Peer: peerID, + }, + }, + }, + p2ptest.Exchange{ + Label: "Subscribe message", + Triggers: []p2ptest.Trigger{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: stream, + History: NewRange(5, 8), + Priority: Top, + }, + Peer: peerID, + }, + }, + Expects: []p2ptest.Expect{ + { + Code: 1, + Msg: &OfferedHashesMsg{ + Stream: NewStream("foo", "", false), + HandoverProof: &HandoverProof{ + Handover: &Handover{}, + }, + Hashes: make([]byte, HashSize), + From: 6, + To: 9, + }, + Peer: peerID, + }, + { + Code: 1, + Msg: &OfferedHashesMsg{ + Stream: stream, + HandoverProof: &HandoverProof{ + Handover: &Handover{}, + }, + From: 1, + To: 1, + Hashes: make([]byte, HashSize), + }, + Peer: peerID, + }, + }, + }, + ) + if err != nil { + t.Fatal(err) + } + + err = streamer.Quit(peerID, stream) + if err != nil { + t.Fatalf("Expected no error, got %v", err) + } + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "Quit message", + Expects: []p2ptest.Expect{ + { + Code: 9, + Msg: &QuitMsg{ + Stream: stream, + }, + Peer: peerID, + }, + }, + }) + + if err != nil { + t.Fatal(err) + } + + historyStream := getHistoryStream(stream) + + err = streamer.Quit(peerID, historyStream) + if err != nil { + t.Fatalf("Expected no error, got %v", err) + } + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "Quit message", + Expects: []p2ptest.Expect{ + { + Code: 9, + Msg: &QuitMsg{ + Stream: historyStream, + }, + Peer: peerID, + }, + }, + }) + + if err != nil { + t.Fatal(err) + } +} diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index ded9163b89..74690d5a2b 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -22,6 +22,7 @@ import ( "fmt" "io" "math" + "strconv" "time" "github.com/ethereum/go-ethereum/log" @@ -64,8 +65,11 @@ func NewSwarmSyncerServer(live bool, po uint8, db *storage.DBAPI) (*SwarmSyncerS const maxPO = 32 func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) { - streamer.RegisterServerFunc("SYNC", func(p *Peer, t []byte, live bool) (Server, error) { - po := t[0] + streamer.RegisterServerFunc("SYNC", func(p *Peer, t string, live bool) (Server, error) { + po, err := ParseSyncBinKey(t) + if err != nil { + return nil, err + } return NewSwarmSyncerServer(live, po, db) }) // streamer.RegisterServerFunc(stream, func(p *Peer) (Server, error) { @@ -99,7 +103,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 if to <= from || from >= s.sessionAt { to = math.MaxUint64 } - ticker := time.NewTicker(10 * time.Millisecond) + ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() for { select { @@ -187,7 +191,7 @@ func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, chunker storage.Chunker) ( // RegisterSwarmSyncerClient registers the client constructor function for // to handle incoming sync streams func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) { - streamer.RegisterClientFunc("SYNC", func(p *Peer, t []byte, love bool) (Client, error) { + streamer.RegisterClientFunc("SYNC", func(p *Peer, _ string, love bool) (Client, error) { return NewSwarmSyncerClient(p, db, nil) }) } @@ -253,3 +257,23 @@ func (s *SwarmSyncerClient) TakeoverProof(stream Stream, from uint64, hashes []b } func (s *SwarmSyncerClient) Close() {} + +// base for parsing and formating sync bin key +// it must be 2 <= base <= 36 +const syncBinKeyBase = 36 + +// FormatSyncBinKey returns a string representation of +// Kademlia bin number to be used as key for SYNC stream. +func FormatSyncBinKey(bin uint8) string { + return strconv.FormatUint(uint64(bin), syncBinKeyBase) +} + +// ParseSyncBinKey parses the string representation +// and returns the Kademlia bin number. +func ParseSyncBinKey(s string) (uint8, error) { + bin, err := strconv.ParseUint(s, syncBinKeyBase, 8) + if err != nil { + return 0, err + } + return uint8(bin), nil +} diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 938c33d98c..9c8e7f9125 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -161,7 +161,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck defer cancel() // start syncing, i.e., subscribe to upstream peers po 1 bin sid := sim.IDs[j+1] - return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", []byte{1}, false), &Range{From: 0, To: 0}, Top) + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", FormatSyncBinKey(1), false), NewRange(0, 0), Top) }) if err != nil { return err diff --git a/swarm/swarm.go b/swarm/swarm.go index 09d5a2bb79..491c755e36 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -153,7 +153,11 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. if err != nil { return } - self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, false, true, true) + self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, &stream.RegistryOptions{ + DoSync: true, + DoRetrieve: true, + SyncUpdateDelay: config.SyncUpdateDelay, + }) self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run)