From 6807d4d2ed35b69c695eeab0591b280488b998d0 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Tue, 6 Feb 2018 10:50:11 +0100 Subject: [PATCH] swarm/network/stream: implement intervals Implement intervals for stream Client. Change the Subscribe message handling to create both live and history streams if required. --- swarm/network/stream/common_test.go | 5 +- swarm/network/stream/delivery.go | 2 +- swarm/network/stream/delivery_test.go | 42 +- swarm/network/stream/intervals/intervals.go | 154 +++++++ .../stream/intervals/intervals_test.go | 395 ++++++++++++++++++ swarm/network/stream/intervals/store.go | 84 ++++ swarm/network/stream/intervals/store_test.go | 69 +++ swarm/network/stream/messages.go | 129 ++++-- swarm/network/stream/peer.go | 84 ++-- swarm/network/stream/stream.go | 189 +++++---- swarm/network/stream/streamer_test.go | 154 +++++-- swarm/network/stream/syncer.go | 22 +- swarm/network/stream/syncer_test.go | 2 +- 13 files changed, 1114 insertions(+), 217 deletions(-) create mode 100644 swarm/network/stream/intervals/intervals.go create mode 100644 swarm/network/stream/intervals/intervals_test.go create mode 100644 swarm/network/stream/intervals/store.go create mode 100644 swarm/network/stream/intervals/store_test.go diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 1deb6ffba9..46d6314624 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/simulations/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" "github.com/ethereum/go-ethereum/swarm/network" + "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -68,7 +69,7 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) { delivery := NewDelivery(kad, db) deliveries[id] = delivery netStore := storage.NewNetStore(store, nil) - r := NewRegistry(addr, delivery, netStore, defaultSkipCheck) + r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck) RegisterSwarmSyncerServer(r, db) RegisterSwarmSyncerClient(r, db) go func() { @@ -98,7 +99,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora db := storage.NewDBAPI(localStore) delivery := NewDelivery(to, db) - streamer := NewRegistry(addr, delivery, localStore, defaultSkipCheck) + streamer := NewRegistry(addr, delivery, localStore, intervals.NewMemStore(), defaultSkipCheck) protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol) err = waitForPeers(streamer, 1*time.Second, 1) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 3ef991158e..53101b4329 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -129,7 +129,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(swarmChunkServerStreamName) + s, err := sp.getServer(NewStream(swarmChunkServerStreamName, nil, false)) if err != nil { return err } diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 183ea2b9e9..b51e48dde9 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -87,10 +87,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { peer := streamer.getPeer(peerID) peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: swarmChunkServerStreamName, - Key: nil, - From: 0, - To: 0, + Stream: NewStream(swarmChunkServerStreamName, nil, false), + History: &Range{ + From: 0, + To: 0, + }, Priority: Top, }) @@ -138,10 +139,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { peer := streamer.getPeer(peerID) peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: swarmChunkServerStreamName, - Key: nil, - From: 0, - To: 0, + Stream: NewStream(swarmChunkServerStreamName, nil, false), + History: &Range{ + From: 0, + To: 0, + }, Priority: Top, }) @@ -172,9 +174,9 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { Hashes: hash, From: 0, // TODO: why is this 32??? - To: 32, - Key: []byte{}, - Stream: swarmChunkServerStreamName, + To: 32, + Stream: NewStream(swarmChunkServerStreamName, nil, false), + Initial: true, }, Peer: peerID, }, @@ -227,7 +229,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) { + streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) { return &testClient{ t: t, }, nil @@ -235,7 +237,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { peerID := tester.IDs[0] - err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true) + stream := NewStream("foo", nil, true) + err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -259,10 +262,11 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { p2ptest.Expect{ Code: 4, Msg: &SubscribeMsg{ - Stream: "foo", - Key: []byte{}, - From: 5, - To: 8, + Stream: stream, + History: &Range{ + From: 5, + To: 8, + }, Priority: Top, }, Peer: peerID, @@ -388,7 +392,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, swarmChunkServerStreamName, nil, 0, 0, Top, false) + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top) }) if err != nil { return err @@ -561,7 +565,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, swarmChunkServerStreamName, nil, 0, 0, Top, false) + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top) }) if err != nil { break diff --git a/swarm/network/stream/intervals/intervals.go b/swarm/network/stream/intervals/intervals.go new file mode 100644 index 0000000000..4a53e1a9e7 --- /dev/null +++ b/swarm/network/stream/intervals/intervals.go @@ -0,0 +1,154 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package intervals + +import ( + "fmt" + "sync" +) + +// Intervals store a list of intervals. Its purpose is to provide +// methods to add new intervals and retrieve missing intervals that +// need to be added. +// It may be used in synchronization of streaming data to persist +// retrieved data ranges between sessions. +type Intervals struct { + start uint64 + ranges [][2]uint64 + mu sync.RWMutex +} + +// New creates a new instance of Intervals. +// Start argument limits the lower bound of intervals. +// No range bellow start bound will be added by Add method or +// returned by Next method. This limit may be used for +// tracking "live" synchronization, where the sync session +// starts from a specific value, and if "live" sync intervals +// need to be merged with historical ones, it can be safely done. +func NewIntervals(start uint64) *Intervals { + return &Intervals{ + start: start, + } +} + +// Add adds a new range to intervals. Range start and end are values +// are both inclusive. +func (i *Intervals) Add(start, end uint64) { + i.mu.Lock() + defer i.mu.Unlock() + + i.add(start, end) +} + +func (i *Intervals) add(start, end uint64) { + if start < i.start { + start = i.start + } + if end < i.start { + return + } + minStartJ := -1 + maxEndJ := -1 + j := 0 + for ; j < len(i.ranges); j++ { + if minStartJ < 0 { + if (start <= i.ranges[j][0] && end+1 >= i.ranges[j][0]) || (start <= i.ranges[j][1]+1 && end+1 >= i.ranges[j][1]) { + if i.ranges[j][0] < start { + start = i.ranges[j][0] + } + minStartJ = j + } + } + if (start <= i.ranges[j][1] && end+1 >= i.ranges[j][1]) || (start <= i.ranges[j][0] && end+1 >= i.ranges[j][0]) { + if i.ranges[j][1] > end { + end = i.ranges[j][1] + } + maxEndJ = j + } + if end+1 <= i.ranges[j][0] { + break + } + } + if minStartJ < 0 && maxEndJ < 0 { + i.ranges = append(i.ranges[:j], append([][2]uint64{{start, end}}, i.ranges[j:]...)...) + return + } + if minStartJ >= 0 { + i.ranges[minStartJ][0] = start + } + if maxEndJ >= 0 { + i.ranges[maxEndJ][1] = end + } + if minStartJ >= 0 && maxEndJ >= 0 && minStartJ != maxEndJ { + i.ranges[maxEndJ][0] = start + i.ranges = append(i.ranges[:minStartJ], i.ranges[maxEndJ:]...) + } +} + +// Merge adds all the intervals from the the m Interval to current one. +func (i *Intervals) Merge(m *Intervals) { + m.mu.RLock() + defer m.mu.RUnlock() + i.mu.Lock() + defer i.mu.Unlock() + + for _, r := range m.ranges { + i.add(r[0], r[1]) + } +} + +// Next returns the first range interval that is not fulfilled. Returned +// start and end values are both inclusive, meaning that the whole range +// including start and end need to be added in order to full the gap +// in intervals. +// Returned value for end is 0 if the next interval is after the whole +// range that is stored in Intervals. Zero end value represents no limit +// on the next interval length. +func (i *Intervals) Next() (start, end uint64) { + i.mu.RLock() + defer i.mu.RUnlock() + + l := len(i.ranges) + if l == 0 { + return i.start, 0 + } + if i.ranges[0][0] != i.start { + return i.start, i.ranges[0][0] - 1 + } + if l == 1 { + return i.ranges[0][1] + 1, 0 + } + return i.ranges[0][1] + 1, i.ranges[1][0] - 1 +} + +// Last returns the value that is at the end of the last interval. +func (i *Intervals) Last() (end uint64) { + i.mu.RLock() + defer i.mu.RUnlock() + + l := len(i.ranges) + if l == 0 { + return 0 + } + return i.ranges[l-1][1] +} + +// String returns a descriptive representation of range intervals +// in [] notation, as a list of two element vectors. +func (i *Intervals) String() string { + return fmt.Sprint(i.ranges) +} diff --git a/swarm/network/stream/intervals/intervals_test.go b/swarm/network/stream/intervals/intervals_test.go new file mode 100644 index 0000000000..b5212f0d91 --- /dev/null +++ b/swarm/network/stream/intervals/intervals_test.go @@ -0,0 +1,395 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package intervals + +import "testing" + +// Test tests Interval methods Add, Next and Last for various +// initial state. +func Test(t *testing.T) { + for i, tc := range []struct { + startLimit uint64 + initial [][2]uint64 + start uint64 + end uint64 + expected string + nextStart uint64 + nextEnd uint64 + last uint64 + }{ + { + initial: nil, + start: 0, + end: 0, + expected: "[[0 0]]", + nextStart: 1, + nextEnd: 0, + last: 0, + }, + { + initial: nil, + start: 0, + end: 10, + expected: "[[0 10]]", + nextStart: 11, + nextEnd: 0, + last: 10, + }, + { + initial: nil, + start: 5, + end: 15, + expected: "[[5 15]]", + nextStart: 0, + nextEnd: 4, + last: 15, + }, + { + initial: [][2]uint64{{0, 0}}, + start: 0, + end: 0, + expected: "[[0 0]]", + nextStart: 1, + nextEnd: 0, + last: 0, + }, + { + initial: [][2]uint64{{0, 0}}, + start: 5, + end: 15, + expected: "[[0 0] [5 15]]", + nextStart: 1, + nextEnd: 4, + last: 15, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 5, + end: 15, + expected: "[[5 15]]", + nextStart: 0, + nextEnd: 4, + last: 15, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 5, + end: 20, + expected: "[[5 20]]", + nextStart: 0, + nextEnd: 4, + last: 20, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 10, + end: 20, + expected: "[[5 20]]", + nextStart: 0, + nextEnd: 4, + last: 20, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 0, + end: 20, + expected: "[[0 20]]", + nextStart: 21, + nextEnd: 0, + last: 20, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 2, + end: 10, + expected: "[[2 15]]", + nextStart: 0, + nextEnd: 1, + last: 15, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 2, + end: 4, + expected: "[[2 15]]", + nextStart: 0, + nextEnd: 1, + last: 15, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 2, + end: 5, + expected: "[[2 15]]", + nextStart: 0, + nextEnd: 1, + last: 15, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 2, + end: 3, + expected: "[[2 3] [5 15]]", + nextStart: 0, + nextEnd: 1, + last: 15, + }, + { + initial: [][2]uint64{{5, 15}}, + start: 2, + end: 4, + expected: "[[2 15]]", + nextStart: 0, + nextEnd: 1, + last: 15, + }, + { + initial: [][2]uint64{{0, 1}, {5, 15}}, + start: 2, + end: 4, + expected: "[[0 15]]", + nextStart: 16, + nextEnd: 0, + last: 15, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}}, + start: 2, + end: 10, + expected: "[[0 10] [15 20]]", + nextStart: 11, + nextEnd: 14, + last: 20, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}}, + start: 8, + end: 18, + expected: "[[0 5] [8 20]]", + nextStart: 6, + nextEnd: 7, + last: 20, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}}, + start: 2, + end: 17, + expected: "[[0 20]]", + nextStart: 21, + nextEnd: 0, + last: 20, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}}, + start: 2, + end: 25, + expected: "[[0 25]]", + nextStart: 26, + nextEnd: 0, + last: 25, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}}, + start: 5, + end: 14, + expected: "[[0 20]]", + nextStart: 21, + nextEnd: 0, + last: 20, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}}, + start: 6, + end: 14, + expected: "[[0 20]]", + nextStart: 21, + nextEnd: 0, + last: 20, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}, {30, 40}}, + start: 6, + end: 29, + expected: "[[0 40]]", + nextStart: 41, + nextEnd: 0, + last: 40, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}}, + start: 3, + end: 55, + expected: "[[0 60]]", + nextStart: 61, + nextEnd: 0, + last: 60, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}}, + start: 21, + end: 49, + expected: "[[0 5] [15 60]]", + nextStart: 6, + nextEnd: 14, + last: 60, + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}}, + start: 0, + end: 100, + expected: "[[0 100]]", + nextStart: 101, + nextEnd: 0, + last: 100, + }, + { + startLimit: 100, + initial: nil, + start: 0, + end: 0, + expected: "[]", + nextStart: 100, + nextEnd: 0, + last: 0, + }, + { + startLimit: 100, + initial: nil, + start: 20, + end: 30, + expected: "[]", + nextStart: 100, + nextEnd: 0, + last: 0, + }, + { + startLimit: 100, + initial: nil, + start: 50, + end: 100, + expected: "[[100 100]]", + nextStart: 101, + nextEnd: 0, + last: 100, + }, + { + startLimit: 100, + initial: nil, + start: 50, + end: 110, + expected: "[[100 110]]", + nextStart: 111, + nextEnd: 0, + last: 110, + }, + { + startLimit: 100, + initial: nil, + start: 120, + end: 130, + expected: "[[120 130]]", + nextStart: 100, + nextEnd: 119, + last: 130, + }, + { + startLimit: 100, + initial: nil, + start: 120, + end: 130, + expected: "[[120 130]]", + nextStart: 100, + nextEnd: 119, + last: 130, + }, + } { + intervals := NewIntervals(tc.startLimit) + intervals.ranges = tc.initial + intervals.Add(tc.start, tc.end) + got := intervals.String() + if got != tc.expected { + t.Errorf("interval #%d: expected %s, got %s", i, tc.expected, got) + } + nextStart, nextEnd := intervals.Next() + if nextStart != tc.nextStart { + t.Errorf("interval #%d, expected next start %d, got %d", i, tc.nextStart, nextStart) + } + if nextEnd != tc.nextEnd { + t.Errorf("interval #%d, expected next end %d, got %d", i, tc.nextEnd, nextEnd) + } + last := intervals.Last() + if last != tc.last { + t.Errorf("interval #%d, expected last %d, got %d", i, tc.last, last) + } + } +} + +func TestMerge(t *testing.T) { + for i, tc := range []struct { + initial [][2]uint64 + merge [][2]uint64 + expected string + }{ + { + initial: nil, + merge: nil, + expected: "[]", + }, + { + initial: [][2]uint64{{10, 20}}, + merge: nil, + expected: "[[10 20]]", + }, + { + initial: nil, + merge: [][2]uint64{{15, 25}}, + expected: "[[15 25]]", + }, + { + initial: [][2]uint64{{0, 100}}, + merge: [][2]uint64{{150, 250}}, + expected: "[[0 100] [150 250]]", + }, + { + initial: [][2]uint64{{0, 100}}, + merge: [][2]uint64{{101, 250}}, + expected: "[[0 250]]", + }, + { + initial: [][2]uint64{{0, 10}, {30, 40}}, + merge: [][2]uint64{{20, 25}, {41, 50}}, + expected: "[[0 10] [20 25] [30 50]]", + }, + { + initial: [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}}, + merge: [][2]uint64{{6, 25}}, + expected: "[[0 25] [30 40] [50 60]]", + }, + } { + intervals := NewIntervals(0) + intervals.ranges = tc.initial + m := NewIntervals(0) + m.ranges = tc.merge + + intervals.Merge(m) + + got := intervals.String() + if got != tc.expected { + t.Errorf("interval #%d: expected %s, got %s", i, tc.expected, got) + } + } +} diff --git a/swarm/network/stream/intervals/store.go b/swarm/network/stream/intervals/store.go new file mode 100644 index 0000000000..87b06c7ed4 --- /dev/null +++ b/swarm/network/stream/intervals/store.go @@ -0,0 +1,84 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package intervals TODO: implement LevelDB based Store. +package intervals + +import ( + "errors" + "sync" +) + +// ErrNotFound is returned by the Store implementation when the Interval +// for a specific key does not exist. +var ErrNotFound = errors.New("not found") + +// Store defines methods required to get and retrieve Intervals for different keys. +// It is meant to be used for intervals persistance for different streams in the +// stream package. +type Store interface { + Get(key string) (i *Intervals, err error) + Put(key string, i *Intervals) (err error) + Delete(key string) (err error) +} + +// MemStore is the reference implementation of Store interface that is supposed +// to be used in tests. +type MemStore struct { + db map[string]*Intervals + mu sync.RWMutex +} + +// NewMemStore returns a new instance of MemStore. +func NewMemStore() *MemStore { + return &MemStore{ + db: make(map[string]*Intervals), + } +} + +// Get retrieves Intervals for a specific key. If there is no Intervals +// ErrNotFound is returned. +func (s *MemStore) Get(key string) (i *Intervals, err error) { + s.mu.RLock() + defer s.mu.RUnlock() + + i, ok := s.db[key] + if !ok { + return nil, ErrNotFound + } + return i, nil +} + +// Put stores Intervals for a specific key. +func (s *MemStore) Put(key string, i *Intervals) (err error) { + s.mu.Lock() + defer s.mu.Unlock() + + s.db[key] = i + return nil +} + +// Delete removes Intervals stored under a specific key. +func (s *MemStore) Delete(key string) (err error) { + s.mu.Lock() + defer s.mu.Unlock() + + if _, ok := s.db[key]; !ok { + return ErrNotFound + } + delete(s.db, key) + return nil +} diff --git a/swarm/network/stream/intervals/store_test.go b/swarm/network/stream/intervals/store_test.go new file mode 100644 index 0000000000..9a30b5d2e0 --- /dev/null +++ b/swarm/network/stream/intervals/store_test.go @@ -0,0 +1,69 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package intervals + +import "testing" + +// TestMemStore tests basic functionality of MemStore. +func TestMemStore(t *testing.T) { + s := NewMemStore() + + key1 := "key1" + i1 := NewIntervals(0) + i1.Add(10, 20) + if err := s.Put(key1, i1); err != nil { + t.Fatal(err) + } + g, err := s.Get(key1) + if err != nil { + t.Fatal(err) + } + if g.String() != i1.String() { + t.Errorf("expected interval %s, got %s", i1, g) + } + + key2 := "key2" + i2 := NewIntervals(0) + i2.Add(10, 20) + if err := s.Put(key2, i2); err != nil { + t.Fatal(err) + } + g, err = s.Get(key2) + if err != nil { + t.Fatal(err) + } + if g.String() != i2.String() { + t.Errorf("expected interval %s, got %s", i2, g) + } + + if err := s.Delete(key1); err != nil { + t.Fatal(err) + } + if _, err := s.Get(key1); err != ErrNotFound { + t.Errorf("expected error %v, got %s", ErrNotFound, err) + } + if _, err := s.Get(key2); err != nil { + t.Errorf("expected error %v, got %s", nil, err) + } + + if err := s.Delete(key2); err != nil { + t.Fatal(err) + } + if _, err := s.Get(key2); err != ErrNotFound { + t.Errorf("expected error %v, got %s", ErrNotFound, err) + } +} diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 22592d288c..dbc3fc24d3 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -23,14 +23,42 @@ import ( "github.com/ethereum/go-ethereum/log" bv "github.com/ethereum/go-ethereum/swarm/network/bitvector" + "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" "github.com/ethereum/go-ethereum/swarm/storage" ) +// Stream defines a unique stream identifier. +type Stream struct { + // Name is used for Client and Server functions identification. + Name string + // Key is the name of specific stream data. + Key []byte + // Live defines whether the stream delivers only new data + // for the specific stream. + Live bool +} + +func NewStream(name string, key []byte, live bool) Stream { + return Stream{ + Name: name, + Key: key, + Live: live, + } +} + +// String return a stream id based on all Stream fields. +func (s Stream) String() string { + t := "h" + if s.Live { + t = "l" + } + return fmt.Sprintf("%s|%x|%s", s.Name, s.Key, t) +} + // SubcribeMsg is the protocol msg for requesting a stream(section) type SubscribeMsg struct { - Stream string - Key []byte - From, To uint64 + Stream Stream + History *Range Priority uint8 // delivered on priority channel } @@ -45,24 +73,58 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { } }() - f, err := p.streamer.GetServerFunc(req.Stream) + log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "history", req.History) + + f, err := p.streamer.GetServerFunc(req.Stream.Name) if err != nil { return err } - s, err := f(p, req.Key) + + s, err := f(p, req.Stream.Key, req.Stream.Live) if err != nil { return err } - os, err := p.setServer(req.Stream, req.Key, s, req.Priority) + os, err := p.setServer(req.Stream, s, req.Priority) if err != nil { return err } - log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To) + + var from uint64 + var to uint64 + if !req.Stream.Live && req.History != nil { + from = req.History.From + to = req.History.To + } + go func() { - if err := p.SendOfferedHashes(os, req.From, req.To); err != nil { + if err := p.SendOfferedHashes(os, from, to, true); err != nil { p.Drop(err) } }() + + if req.Stream.Live && req.History != nil { + // subscribe to the history stream as well + s, err := f(p, req.Stream.Key, false) + if err != nil { + return err + } + historyStream := NewStream(req.Stream.Name, req.Stream.Key, false) + priority := req.Priority + if priority > 0 { + // decrement history stream priority + priority-- + } + os, err := p.setServer(historyStream, s, priority) + if err != nil { + return err + } + go func() { + if err := p.SendOfferedHashes(os, req.History.From, req.History.To, true); err != nil { + p.Drop(err) + } + }() + } + return nil } @@ -75,23 +137,22 @@ func (p *Peer) handleSubscribeErrorMsg(req *SubscribeErrorMsg) (err error) { } type UnsubscribeMsg struct { - Stream string - Key []byte + Stream Stream } func (p *Peer) handleUnsubscribeMsg(req *UnsubscribeMsg) error { - p.removeServer(req.Stream, req.Key) + p.removeServer(req.Stream) return nil } // OfferedHashesMsg is the protocol msg for offering to hand over a // stream section type OfferedHashesMsg struct { - Stream string // name of Stream - Key []byte // subtype or key + Stream Stream // name of Stream From, To uint64 // peer and db-specific entry count Hashes []byte // stream of hashes (128) - *HandoverProof // HandoverProof + Initial bool + *HandoverProof // HandoverProof } // String pretty prints OfferedHashesMsg @@ -102,9 +163,7 @@ func (m OfferedHashesMsg) String() string { // handleOfferedHashesMsg protocol msg handler calls the incoming streamer interface // Filter method func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { - sk := req.Stream - sk += keyToString(req.Key) - s, err := p.getClient(sk) + c, err := p.getClient(req.Stream) if err != nil { return err } @@ -117,7 +176,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { for i := 0; i < len(hashes); i += HashSize { hash := hashes[i : i+HashSize] - if wait := s.NeedData(hash); wait != nil { + if wait := c.NeedData(hash); wait != nil { want.Set(i/HashSize, true) wg.Add(1) // create request and wait until the chunk data arrives and is stored @@ -142,22 +201,27 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { // }() go func() { wg.Wait() - s.next <- s.batchDone(p, req, hashes) + c.next <- c.batchDone(p, req, hashes) }() // only send wantedKeysMsg if all missing chunks of the previous batch arrived // except - if s.live { - s.sessionAt = req.From + if c.stream.Live { + c.sessionAt = req.From + if req.Initial { + // create initial intervals for live stream starting from the first From value + if err := c.intervalsStore.Put(peerStreamIntervalsKey(p, req.Stream), intervals.NewIntervals(req.From)); err != nil { + return err + } + } } - from, to := s.nextBatch(req.To) - log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To) + from, to := c.nextBatch(req.To) + log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To) if from == to { return nil } msg := &WantedHashesMsg{ Stream: req.Stream, - Key: req.Key, Want: want.Bytes(), From: from, To: to, @@ -167,14 +231,14 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { case <-time.After(30 * time.Second): p.Drop(err) return - case err := <-s.next: + case err := <-c.next: if err != nil { p.Drop(err) return } } - log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To) - err := p.SendPriority(msg, s.priority) + log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To) + err := p.SendPriority(msg, c.priority) if err != nil { p.Drop(err) } @@ -185,8 +249,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { // WantedHashesMsg is the protocol msg data for signaling which hashes // offered in OfferedHashesMsg downstream peer actually wants sent over type WantedHashesMsg struct { - Stream string // name of stream - Key []byte // subtype or key + Stream Stream Want []byte // bitvector indicating which keys of the batch needed From, To uint64 // next interval offset - empty if not to be continued } @@ -200,15 +263,15 @@ func (m WantedHashesMsg) String() string { // * sends the next batch of unsynced keys // * sends the actual data chunks as per WantedHashesMsg func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { - log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To) - s, err := p.getServer(req.Stream + keyToString(req.Key)) + log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To) + s, err := p.getServer(req.Stream) if err != nil { return err } hashes := s.currentBatch // launch in go routine since GetBatch blocks until new hashes arrive go func() { - if err := p.SendOfferedHashes(s, req.From, req.To); err != nil { + if err := p.SendOfferedHashes(s, req.From, req.To, false); err != nil { p.Drop(err) } }() @@ -237,7 +300,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { // Handover represents a statement that the upstream peer hands over the stream section type Handover struct { - Stream string // name of stream + Stream Stream // name of stream Start, End uint64 // index of hashes Root []byte // Root hash for indexed segment inclusion proofs } diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 12810789d9..35375f5c34 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -26,6 +26,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/protocols" pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue" + "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -84,7 +85,7 @@ func (p *Peer) SendPriority(msg interface{}, priority uint8) error { } // SendOfferedHashes sends OfferedHashesMsg protocol msg -func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { +func (p *Peer) SendOfferedHashes(s *server, f, t uint64, initial bool) error { hashes, from, to, proof, err := s.SetNextBatch(f, t) if err != nil { return err @@ -105,57 +106,56 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { From: from, To: to, Stream: s.stream, - Key: s.key, + Initial: initial, } - log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to) + log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to) return p.SendPriority(msg, s.priority) } -func (p *Peer) getServer(s string) (*server, error) { +func (p *Peer) getServer(s Stream) (*server, error) { p.serverMu.RLock() defer p.serverMu.RUnlock() - server := p.servers[s] + server := p.servers[s.String()] if server == nil { return nil, fmt.Errorf("server '%v' not provided to peer %v", s, p.ID()) } return server, nil } -func (p *Peer) getClient(s string) (*client, error) { +func (p *Peer) getClient(s Stream) (*client, error) { p.clientMu.RLock() defer p.clientMu.RUnlock() - client := p.clients[s] + client := p.clients[s.String()] if client == nil { return nil, fmt.Errorf("client '%v' not provided to peer %v", s, p.ID()) } return client, nil } -func (p *Peer) setServer(s string, key []byte, o Server, priority uint8) (*server, error) { +func (p *Peer) setServer(s Stream, o Server, priority uint8) (*server, error) { p.serverMu.Lock() defer p.serverMu.Unlock() - sk := s + keyToString(key) + sk := s.String() if p.servers[sk] != nil { return nil, fmt.Errorf("server %v already registered", sk) } os := &server{ Server: o, - priority: priority, stream: s, - key: key, + priority: priority, } p.servers[sk] = os return os, nil } -func (p *Peer) removeServer(s string, key []byte) error { +func (p *Peer) removeServer(s Stream) error { p.serverMu.Lock() defer p.serverMu.Unlock() - sk := s + keyToString(key) + sk := s.String() server, ok := p.servers[sk] if !ok { return errServerNotFound @@ -165,39 +165,63 @@ func (p *Peer) removeServer(s string, key []byte) error { return nil } -func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bool) error { +func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore intervals.Store) error { p.clientMu.Lock() defer p.clientMu.Unlock() - sk := s + keyToString(key) + sk := s.String() if p.clients[sk] != nil { return fmt.Errorf("client %v already registered", sk) } + + intervalsKey := peerStreamIntervalsKey(p, s) + if s.Live { + // try to find previous history and live intervals and merge live into history + historyKey := peerStreamIntervalsKey(p, NewStream(s.Name, s.Key, false)) + historyIntervals, err := intervalsStore.Get(historyKey) + switch err { + case nil: + liveIntervals, err := intervalsStore.Get(intervalsKey) + switch err { + case nil: + historyIntervals.Merge(liveIntervals) + if err := intervalsStore.Put(historyKey, historyIntervals); err != nil { + log.Error("stream set client: put history intervals", "stream", s, "peer", p, "err", err) + } + case intervals.ErrNotFound: + default: + log.Error("stream set client: get live intervals", "stream", s, "peer", p, "err", err) + } + case intervals.ErrNotFound: + default: + log.Error("stream set client: get history intervals", "stream", s, "peer", p, "err", err) + } + } else { + // create intervals for history stream + // live stream can create intervals when the first sessionAt is known + if err := intervalsStore.Put(intervalsKey, intervals.NewIntervals(0)); err != nil { + return err + } + } + next := make(chan error, 1) - // var intervals *Intervals - // if !live { - // key := s + p.ID().String() - // intervals = NewIntervals(key, p.streamer) - // } p.clients[sk] = &client{ - Client: i, - // intervals: intervals, - live: live, - priority: priority, - next: next, - stream: s, - key: key, + Client: i, + stream: s, + priority: priority, + next: next, + intervalsStore: intervalsStore, + intervalsKey: intervalsKey, } next <- nil // this is to allow wantedKeysMsg before first batch arrives return nil } -func (p *Peer) removeClient(s string, key []byte) error { +func (p *Peer) removeClient(s Stream) error { p.clientMu.Lock() defer p.clientMu.Unlock() - sk := s + keyToString(key) - client, ok := p.clients[sk] + client, ok := p.clients[s.String()] if !ok { return errClientNotFound } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 5770f679c3..17838e7a5c 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/swarm/network" + "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -45,43 +46,45 @@ const ( // Registry registry for outgoing and incoming streamer constructors type Registry struct { - api *API - addr *network.BzzAddr - skipCheck bool - clientMu sync.RWMutex - serverMu sync.RWMutex - peersMu sync.RWMutex - serverFuncs map[string]func(*Peer, []byte) (Server, error) - clientFuncs map[string]func(*Peer, []byte) (Client, error) - peers map[discover.NodeID]*Peer - delivery *Delivery - store storage.ChunkStore + api *API + addr *network.BzzAddr + skipCheck bool + 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) + peers map[discover.NodeID]*Peer + delivery *Delivery + store storage.ChunkStore + intervalsStore intervals.Store } // NewRegistry is Streamer constructor -func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, skipCheck bool) *Registry { +func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, intervalsStore intervals.Store, skipCheck bool) *Registry { streamer := &Registry{ - addr: addr, - skipCheck: skipCheck, - store: store, - serverFuncs: make(map[string]func(*Peer, []byte) (Server, error)), - clientFuncs: make(map[string]func(*Peer, []byte) (Client, error)), - peers: make(map[discover.NodeID]*Peer), - delivery: delivery, + addr: addr, + skipCheck: skipCheck, + store: store, + serverFuncs: make(map[string]func(*Peer, []byte, bool) (Server, error)), + clientFuncs: make(map[string]func(*Peer, []byte, bool) (Client, error)), + peers: make(map[discover.NodeID]*Peer), + delivery: delivery, + intervalsStore: intervalsStore, } streamer.api = NewAPI(streamer, streamer.store) delivery.getPeer = streamer.getPeer - streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, t []byte) (Server, error) { + streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ []byte, _ bool) (Server, error) { return NewSwarmChunkServer(delivery.db), nil }) - streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t []byte) (Client, error) { + streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ []byte, _ bool) (Client, error) { return NewSwarmSyncerClient(p, delivery.db, nil) }) return streamer } // RegisterClient registers an incoming streamer constructor -func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte) (Client, error)) { +func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte, bool) (Client, error)) { r.clientMu.Lock() defer r.clientMu.Unlock() @@ -89,7 +92,7 @@ func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte) (Clie } // RegisterServer registers an outgoing streamer constructor -func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte) (Server, error)) { +func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte, bool) (Server, error)) { r.serverMu.Lock() defer r.serverMu.Unlock() @@ -97,7 +100,7 @@ func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte) (Serv } // GetClient accessor for incoming streamer constructors -func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte) (Client, error), error) { +func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte, bool) (Client, error), error) { r.clientMu.RLock() defer r.clientMu.RUnlock() @@ -109,7 +112,7 @@ func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte) (Client, er } // GetServer accessor for incoming streamer constructors -func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte) (Server, error), error) { +func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Server, error), error) { r.serverMu.RLock() defer r.serverMu.RUnlock() @@ -121,8 +124,8 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte) (Server, er } // Subscribe initiates the streamer -func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error { - f, err := r.GetClientFunc(s) +func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error { + f, err := r.GetClientFunc(s.Name) if err != nil { return err } @@ -132,29 +135,42 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, t return fmt.Errorf("peer not found %v", peerId) } - is, err := f(peer, t) + is, err := f(peer, s.Key, s.Live) if err != nil { return err } - err = peer.setClient(s, t, is, priority, live) + err = peer.setClient(s, is, priority, r.intervalsStore) if err != nil { return err } + if s.Live && h != nil { + is, err := f(peer, s.Key, false) + if err != nil { + return err + } + p := priority + if p > 0 { + p-- + } + historyStream := NewStream(s.Name, s.Key, false) + err = peer.setClient(historyStream, is, p, r.intervalsStore) + if err != nil { + return err + } + } + msg := &SubscribeMsg{ - Stream: s, - Key: t, - // Live: live, - From: from, - To: to, + Stream: s, + History: h, Priority: priority, } - log.Debug("Subscribe ", "peer", peerId, "stream", s, "key", t, "from", from, "to", to) + log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h) return peer.SendPriority(msg, priority) } -func (r *Registry) Unsubscribe(peerId discover.NodeID, s string, t []byte) error { +func (r *Registry) Unsubscribe(peerId discover.NodeID, s Stream) error { peer := r.getPeer(peerId) if peer == nil { return fmt.Errorf("peer not found %v", peerId) @@ -162,14 +178,13 @@ func (r *Registry) Unsubscribe(peerId discover.NodeID, s string, t []byte) error msg := &UnsubscribeMsg{ Stream: s, - Key: t, } - log.Debug("Unsubscribe ", "peer", peerId, "stream", s, "key", t) + log.Debug("Unsubscribe ", "peer", peerId, "stream", s) if err := peer.Send(msg); err != nil { return err } - return peer.removeClient(s, t) + return peer.removeClient(s) } func (r *Registry) Retrieve(chunk *storage.Chunk) error { @@ -261,20 +276,11 @@ func (p *Peer) HandleMsg(msg interface{}) error { } } -func keyToString(key []byte) string { - l := len(key) - if l == 0 { - return "" - } - return fmt.Sprintf("%s-%d", string(key[:l-1]), uint8(key[l-1])) -} - type server struct { Server + stream Stream priority uint8 currentBatch []byte - stream string - key []byte } // Server interface for outgoing peer Streamer @@ -286,50 +292,59 @@ type Server interface { type client struct { Client + stream Stream priority uint8 sessionAt uint64 - live bool - stream string - key []byte next chan error + + intervalsKey string + intervalsStore intervals.Store +} + +func peerStreamIntervalsKey(p *Peer, s Stream) string { + return p.ID().String() + s.String() +} + +func (c client) AddInterval(start, end uint64) (err error) { + i, err := c.intervalsStore.Get(c.intervalsKey) + if err != nil { + return err + } + i.Add(start, end) + return c.intervalsStore.Put(c.intervalsKey, i) +} + +func (c client) NextInterval() (start, end uint64, err error) { + i, err := c.intervalsStore.Get(c.intervalsKey) + if err != nil { + return 0, 0, err + } + start, end = i.Next() + return start, end, nil } // Client interface for incoming peer Streamer type Client interface { NeedData([]byte) func() - BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) + BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) Close() } -// nextBatch adjusts the indexes by inspecting the intervals func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) { - var intervals []uint64 - if c.live { - if len(intervals) == 0 { - intervals = []uint64{c.sessionAt, from} - } else { - intervals[1] = from - } - nextFrom = from - } else if from >= c.sessionAt { // history sync complete - intervals = nil - nextFrom = from - nextTo = math.MaxUint64 - } else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals - intervals = append(intervals[:1], intervals[3:]...) - nextFrom = intervals[1] - if len(intervals) > 2 { - nextTo = intervals[2] - } else { - nextTo = c.sessionAt - } - } else { - nextFrom = from - intervals[1] = from + if c.stream.Live { + return from, 0 + } else if from >= c.sessionAt { + return from, math.MaxUint64 + } + nextFrom, nextTo, err := c.NextInterval() + if err != nil { + log.Error("next intervals", "stream", c.stream) + return + } + if nextTo == 0 { nextTo = c.sessionAt } - // b.intervals.set(intervals) - return nextFrom, nextTo + return } func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error { @@ -338,6 +353,10 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error if err != nil { return err } + // TODO: make a test case for testing if the interval is added when the batch is done + if err := c.AddInterval(tp.Takeover.Start, tp.Takeover.End); err != nil { + return err + } return p.SendPriority(tp, c.priority) } return nil @@ -399,6 +418,10 @@ func (r *Registry) Stop() error { return nil } +type Range struct { + From, To uint64 +} + type API struct { streamer *Registry dpa *storage.DPA @@ -432,10 +455,10 @@ func (api *API) ReadAll(hash common.Hash) (int64, error) { return readAll(api.dpa, hash[:]) } -func (api *API) SubscribeStream(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error { - return api.streamer.Subscribe(peerId, s, t, from, to, priority, live) +func (api *API) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error { + return api.streamer.Subscribe(peerId, s, history, priority) } -func (api *API) UnsubscribeStream(peerId discover.NodeID, s string, t []byte) error { - return api.streamer.Unsubscribe(peerId, s, t) +func (api *API) UnsubscribeStream(peerId discover.NodeID, s Stream) error { + return api.streamer.Unsubscribe(peerId, s) } diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index 951e008a53..da27f5907c 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -32,7 +32,8 @@ func TestStreamerSubscribe(t *testing.T) { t.Fatal(err) } - err = streamer.Subscribe(tester.IDs[0], "foo", nil, 0, 0, Top, true) + stream := NewStream("foo", nil, true) + err = streamer.Subscribe(tester.IDs[0], stream, &Range{From: 0, To: 0}, Top) if err == nil || err.Error() != "stream foo not registered" { t.Fatalf("Expected error %v, got %v", "stream foo not registered", err) } @@ -72,7 +73,7 @@ func (self *testClient) NeedData(hash []byte) func() { return nil } -func (self *testClient) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) { +func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { close(batchDone) return nil } @@ -97,7 +98,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) { + streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) { return &testClient{ t: t, }, nil @@ -105,7 +106,8 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { peerID := tester.IDs[0] - err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true) + stream := NewStream("foo", nil, true) + err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -116,10 +118,11 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { p2ptest.Expect{ Code: 4, Msg: &SubscribeMsg{ - Stream: "foo", - Key: []byte{}, - From: 5, - To: 8, + Stream: stream, + History: &Range{ + From: 5, + To: 8, + }, Priority: Top, }, Peer: peerID, @@ -131,7 +134,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { t.Fatal(err) } - err = streamer.Unsubscribe(peerID, "foo", []byte{}) + err = streamer.Unsubscribe(peerID, stream) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -142,8 +145,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { p2ptest.Expect{ Code: 0, Msg: &UnsubscribeMsg{ - Stream: "foo", - Key: []byte{}, + Stream: stream, }, Peer: peerID, }, @@ -162,7 +164,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) { + stream := NewStream("foo", nil, false) + + streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { return &testServer{ t: t, }, nil @@ -176,10 +180,11 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { p2ptest.Trigger{ Code: 4, Msg: &SubscribeMsg{ - Stream: "foo", - Key: []byte{}, - From: 5, - To: 8, + Stream: stream, + History: &Range{ + From: 5, + To: 8, + }, Priority: Top, }, Peer: peerID, @@ -189,14 +194,14 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { p2ptest.Expect{ Code: 1, Msg: &OfferedHashesMsg{ - Stream: "foo", - Key: []byte{}, + Stream: stream, HandoverProof: &HandoverProof{ Handover: &Handover{}, }, - Hashes: make([]byte, HashSize), - From: 6, - To: 9, + Hashes: make([]byte, HashSize), + From: 6, + To: 9, + Initial: true, }, Peer: peerID, }, @@ -213,8 +218,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { p2ptest.Trigger{ Code: 0, Msg: &UnsubscribeMsg{ - Stream: "foo", - Key: []byte{}, + Stream: stream, }, Peer: peerID, }, @@ -233,12 +237,14 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) { + streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { return &testServer{ t: t, }, nil }) + stream := NewStream("bar", nil, true) + peerID := tester.IDs[0] err = tester.TestExchanges(p2ptest.Exchange{ @@ -247,10 +253,11 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { p2ptest.Trigger{ Code: 4, Msg: &SubscribeMsg{ - Stream: "bar", - Key: []byte{}, - From: 5, - To: 8, + Stream: stream, + History: &Range{ + From: 5, + To: 8, + }, Priority: Top, }, Peer: peerID, @@ -272,6 +279,78 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { } } +// TODO: fix: tests with TestExchanges are inconsistent because Expects check +// ordering is not guarrantied but fails if the order is wrong. +// func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) { +// tester, streamer, _, teardown, err := newStreamerTester(t) +// defer teardown() +// if err != nil { +// t.Fatal(err) +// } + +// stream := NewStream("foo", nil, true) + +// streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { +// return &testServer{ +// t: t, +// }, nil +// }) + +// peerID := tester.IDs[0] + +// err = tester.TestExchanges(p2ptest.Exchange{ +// Label: "Subscribe message", +// Triggers: []p2ptest.Trigger{ +// { +// Code: 4, +// Msg: &SubscribeMsg{ +// Stream: stream, +// History: &Range{ +// From: 5, +// To: 8, +// }, +// Priority: Top, +// }, +// Peer: peerID, +// }, +// }, +// Expects: []p2ptest.Expect{ +// { +// Code: 1, +// Msg: &OfferedHashesMsg{ +// Stream: NewStream("foo", nil, false), +// HandoverProof: &HandoverProof{ +// Handover: &Handover{}, +// }, +// Hashes: make([]byte, HashSize), +// From: 6, +// To: 9, +// Initial: true, +// }, +// Peer: peerID, +// }, +// { +// Code: 1, +// Msg: &OfferedHashesMsg{ +// Stream: stream, +// HandoverProof: &HandoverProof{ +// Handover: &Handover{}, +// }, +// From: 1, +// To: 1, +// Hashes: make([]byte, HashSize), +// Initial: true, +// }, +// Peer: peerID, +// }, +// }, +// }) + +// if err != nil { +// t.Fatal(err) +// } +// } + func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(t) defer teardown() @@ -279,7 +358,9 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { t.Fatal(err) } - streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) { + stream := NewStream("foo", nil, true) + + streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) { return &testClient{ t: t, }, nil @@ -287,7 +368,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { peerID := tester.IDs[0] - err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true) + err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } @@ -298,10 +379,11 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { p2ptest.Expect{ Code: 4, Msg: &SubscribeMsg{ - Stream: "foo", - Key: []byte{}, - From: 5, - To: 8, + Stream: stream, + History: &Range{ + From: 5, + To: 8, + }, Priority: Top, }, Peer: peerID, @@ -320,7 +402,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { Hashes: hashes, From: 5, To: 8, - Stream: "foo", + Stream: stream, }, Peer: peerID, }, @@ -329,7 +411,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { p2ptest.Expect{ Code: 2, Msg: &WantedHashesMsg{ - Stream: "foo", + Stream: stream, Want: []byte{5}, From: 8, To: 0, diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index 6d8473afc9..252b7432ba 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -64,10 +64,9 @@ 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) (Server, error) { + streamer.RegisterServerFunc("SYNC", func(p *Peer, t []byte, live bool) (Server, error) { po := uint8(t[0]) - // TODO: make this work for HISTORY too - return NewSwarmSyncerServer(false, po, db) + return NewSwarmSyncerServer(live, po, db) }) // streamer.RegisterServerFunc(stream, func(p *Peer) (Server, error) { // return NewOutgoingProvableSwarmSyncer(po, db) @@ -188,7 +187,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) (Client, error) { + streamer.RegisterClientFunc("SYNC", func(p *Peer, t []byte, love bool) (Client, error) { return NewSwarmSyncerClient(p, db, nil) }) } @@ -207,14 +206,14 @@ func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) { } // BatchDone -func (s *SwarmSyncerClient) BatchDone(streamName string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) { +func (s *SwarmSyncerClient) BatchDone(stream Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) { if s.chunker != nil { - return func() (*TakeoverProof, error) { return s.TakeoverProof(streamName, from, hashes, root) } + return func() (*TakeoverProof, error) { return s.TakeoverProof(stream, from, hashes, root) } } return nil } -func (s *SwarmSyncerClient) TakeoverProof(streamName string, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) { +func (s *SwarmSyncerClient) TakeoverProof(stream Stream, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) { // for provable syncer currentRoot is non-zero length if s.chunker != nil { if from > s.sessionAt { // for live syncing currentRoot is always updated @@ -241,11 +240,10 @@ func (s *SwarmSyncerClient) TakeoverProof(streamName string, from uint64, hashes } s.end += uint64(len(hashes)) / HashSize takeover := &Takeover{ - Stream: streamName, - // Key: s.Key, - Start: s.start, - End: s.end, - Root: root, + Stream: stream, + Start: s.start, + End: s.end, + Root: root, } // serialise and sign return &TakeoverProof{ diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 58d780c36f..c8f4a0deca 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -160,7 +160,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, "SYNC", []byte{1}, 0, 0, Top, false) + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", []byte{1}, false), &Range{From: 0, To: 0}, Top) }) if err != nil { return err