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)