From 6807d4d2ed35b69c695eeab0591b280488b998d0 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Tue, 6 Feb 2018 10:50:11 +0100 Subject: [PATCH 1/4] 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 From afe2f02efc55b29a361659125ed6d24bce037e9d Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Tue, 13 Feb 2018 17:21:46 +0100 Subject: [PATCH 2/4] swarm/netowrk/stream: create client on first offered hashes msg Store client init params in a dedicated Peer map and construct client only on the first offered hashes message. Update tests and fix issues with shared variables in testClient --- swarm/network/stream/delivery_test.go | 5 +- swarm/network/stream/messages.go | 31 ++-- swarm/network/stream/peer.go | 170 +++++++++++++++------ swarm/network/stream/stream.go | 44 +++--- swarm/network/stream/streamer_test.go | 210 +++++++++++++++++++------- 5 files changed, 322 insertions(+), 138 deletions(-) diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index b51e48dde9..6b4aff89b0 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -174,9 +174,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { Hashes: hash, From: 0, // TODO: why is this 32??? - To: 32, - Stream: NewStream(swarmChunkServerStreamName, nil, false), - Initial: true, + To: 32, + Stream: NewStream(swarmChunkServerStreamName, nil, false), }, Peer: peerID, }, diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index dbc3fc24d3..1948234625 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -23,7 +23,6 @@ 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" ) @@ -58,8 +57,8 @@ func (s Stream) String() string { // SubcribeMsg is the protocol msg for requesting a stream(section) type SubscribeMsg struct { Stream Stream - History *Range - Priority uint8 // delivered on priority channel + History *Range `rlp:"nil"` + Priority uint8 // delivered on priority channel } func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { @@ -97,7 +96,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { } go func() { - if err := p.SendOfferedHashes(os, from, to, true); err != nil { + if err := p.SendOfferedHashes(os, from, to); err != nil { p.Drop(err) } }() @@ -108,18 +107,13 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { 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) + + os, err := p.setServer(getHistoryStream(req.Stream), s, getHistoryPriority(req.Priority)) if err != nil { return err } go func() { - if err := p.SendOfferedHashes(os, req.History.From, req.History.To, true); err != nil { + if err := p.SendOfferedHashes(os, req.History.From, req.History.To); err != nil { p.Drop(err) } }() @@ -151,8 +145,7 @@ type OfferedHashesMsg struct { Stream Stream // name of Stream From, To uint64 // peer and db-specific entry count Hashes []byte // stream of hashes (128) - Initial bool - *HandoverProof // HandoverProof + *HandoverProof // HandoverProof } // String pretty prints OfferedHashesMsg @@ -163,7 +156,7 @@ func (m OfferedHashesMsg) String() string { // handleOfferedHashesMsg protocol msg handler calls the incoming streamer interface // Filter method func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { - c, err := p.getClient(req.Stream) + c, _, err := p.getOrSetClient(req.Stream, req.From, req.To) if err != nil { return err } @@ -207,12 +200,6 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { // except 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 := c.nextBatch(req.To) log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To) @@ -271,7 +258,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { 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, false); err != nil { + if err := p.SendOfferedHashes(s, req.From, req.To); err != nil { p.Drop(err) } }() diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 35375f5c34..65dbefea5d 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -33,31 +33,38 @@ import ( var sendTimeout = 5 * time.Second var ( - errServerNotFound = errors.New("server not found") - errClientNotFound = errors.New("client not found") + errServerNotFound = errors.New("server not found") + errClientNotFound = errors.New("client not found") + errClientParamsNotFound = errors.New("client params not found") ) // Peer is the Peer extention for the streaming protocol type Peer struct { *protocols.Peer - streamer *Registry - pq *pq.PriorityQueue - serverMu sync.RWMutex - clientMu sync.RWMutex - servers map[string]*server - clients map[string]*client - quit chan struct{} + streamer *Registry + pq *pq.PriorityQueue + serverMu sync.RWMutex + clientMu sync.RWMutex + clientParamsMu sync.RWMutex + servers map[string]*server + clients map[string]*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 + quit chan struct{} } // NewPeer is the constructor for Peer func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer { p := &Peer{ - Peer: peer, - pq: pq.New(int(PriorityQueue), PriorityQueueCap), - streamer: streamer, - servers: make(map[string]*server), - clients: make(map[string]*client), - quit: make(chan struct{}), + 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), + quit: make(chan struct{}), } ctx, cancel := context.WithCancel(context.Background()) go p.pq.Run(ctx, func(i interface{}) { p.Send(i) }) @@ -85,7 +92,7 @@ func (p *Peer) SendPriority(msg interface{}, priority uint8) error { } // SendOfferedHashes sends OfferedHashesMsg protocol msg -func (p *Peer) SendOfferedHashes(s *server, f, t uint64, initial bool) error { +func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { hashes, from, to, proof, err := s.SetNextBatch(f, t) if err != nil { return err @@ -106,7 +113,6 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64, initial bool) error { From: from, To: to, Stream: s.stream, - Initial: initial, } 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) @@ -123,17 +129,6 @@ func (p *Peer) getServer(s Stream) (*server, error) { return server, nil } -func (p *Peer) getClient(s Stream) (*client, error) { - p.clientMu.RLock() - defer p.clientMu.RUnlock() - - 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 Stream, o Server, priority uint8) (*server, error) { p.serverMu.Lock() defer p.serverMu.Unlock() @@ -165,7 +160,18 @@ func (p *Peer) removeServer(s Stream) error { return nil } -func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore intervals.Store) error { +func (p *Peer) getClient(s Stream) (*client, error) { + p.clientMu.RLock() + defer p.clientMu.RUnlock() + + 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) setClient(s Stream, from, to uint64) error { p.clientMu.Lock() defer p.clientMu.Unlock() @@ -174,18 +180,45 @@ func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore inte return fmt.Errorf("client %v already registered", sk) } + _, err := p.setClientNolock(s, from, to) + return err +} + +func (p *Peer) setClientNolock(s Stream, from, to uint64) (c *client, err error) { + f, err := p.streamer.GetClientFunc(s.Name) + if err != nil { + return nil, err + } + + is, err := f(p, s.Key, s.Live) + if err != nil { + return nil, err + } + + cp, err := p.getClientParams(s) + if err != nil { + return nil, err + } + defer func() { + if err == nil { + if err := p.removeClientParams(s); err != nil { + log.Error("stream set client: remove client params", "stream", s, "peer", p, "err", err) + } + } + }() + 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) + historyIntervals, err := p.streamer.intervalsStore.Get(historyKey) switch err { case nil: - liveIntervals, err := intervalsStore.Get(intervalsKey) + liveIntervals, err := p.streamer.intervalsStore.Get(intervalsKey) switch err { case nil: historyIntervals.Merge(liveIntervals) - if err := intervalsStore.Put(historyKey, historyIntervals); err != nil { + if err := p.streamer.intervalsStore.Put(historyKey, historyIntervals); err != nil { log.Error("stream set client: put history intervals", "stream", s, "peer", p, "err", err) } case intervals.ErrNotFound: @@ -196,25 +229,40 @@ func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore inte 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 - } + } + + if err := p.streamer.intervalsStore.Put(intervalsKey, intervals.NewIntervals(from)); err != nil { + return nil, err } next := make(chan error, 1) - p.clients[sk] = &client{ - Client: i, + c = &client{ + Client: is, stream: s, - priority: priority, + priority: cp.priority, next: next, - intervalsStore: intervalsStore, + intervalsStore: p.streamer.intervalsStore, intervalsKey: intervalsKey, } + p.clients[s.String()] = c next <- nil // this is to allow wantedKeysMsg before first batch arrives - return nil + return c, nil +} + +func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created bool, err error) { + p.clientMu.RLock() + defer p.clientMu.RUnlock() + + c = p.clients[s.String()] + if c != nil { + return c, false, nil + } + + c, err = p.setClientNolock(s, from, to) + if err != nil { + return nil, false, err + } + return c, true, nil } func (p *Peer) removeClient(s Stream) error { @@ -229,6 +277,42 @@ func (p *Peer) removeClient(s Stream) error { return nil } +func (p *Peer) getClientParams(s Stream) (*clientParams, error) { + p.clientParamsMu.RLock() + defer p.clientParamsMu.RUnlock() + + params := p.clientParams[s.String()] + if params == nil { + return nil, fmt.Errorf("client params '%v' not provided to peer %v", s, p.ID()) + } + return params, nil +} + +func (p *Peer) setClientParams(s Stream, params *clientParams) error { + p.clientParamsMu.Lock() + defer p.clientParamsMu.Unlock() + + sk := s.String() + if p.clientParams[sk] != nil { + return fmt.Errorf("client params %v already set", sk) + } + p.clientParams[sk] = params + return nil +} + +func (p *Peer) removeClientParams(s Stream) error { + p.clientParamsMu.Lock() + defer p.clientParamsMu.Unlock() + + sk := s.String() + _, ok := p.clientParams[sk] + if !ok { + return errClientParamsNotFound + } + delete(p.clientParams, sk) + return nil +} + func (p *Peer) close() { for _, s := range p.servers { s.Close() diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 17838e7a5c..85c9a79355 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -125,8 +125,8 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Serv // Subscribe initiates the streamer func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error { - f, err := r.GetClientFunc(s.Name) - if err != nil { + // check if the stream is registered + if _, err := r.GetClientFunc(s.Name); err != nil { return err } @@ -135,27 +135,18 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priorit return fmt.Errorf("peer not found %v", peerId) } - is, err := f(peer, s.Key, s.Live) - if err != nil { - return err - } - err = peer.setClient(s, is, priority, r.intervalsStore) + err := peer.setClientParams(s, &clientParams{priority: priority}) 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 { + if err := peer.setClientParams( + getHistoryStream(s), + &clientParams{ + priority: getHistoryPriority(priority), + }, + ); err != nil { return err } } @@ -367,6 +358,12 @@ func (c *client) close() { c.Close() } +// clientParams store parameters for the new client +// between a subscription and initial offered hashes request handling. +type clientParams struct { + priority uint8 +} + // Spec is the spec of the streamer protocol var Spec = &protocols.Spec{ Name: "stream", @@ -422,6 +419,17 @@ type Range struct { From, To uint64 } +func getHistoryPriority(priority uint8) uint8 { + if priority == 0 { + return 0 + } + return priority - 1 +} + +func getHistoryStream(s Stream) Stream { + return NewStream(s.Name, s.Key, false) +} + type API struct { streamer *Registry dpa *storage.DPA diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index da27f5907c..3b31017c0a 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -40,41 +40,57 @@ func TestStreamerSubscribe(t *testing.T) { } var ( - hash0 = sha3.Sum256([]byte{0}) - hash1 = sha3.Sum256([]byte{1}) - hash2 = sha3.Sum256([]byte{2}) - hashesTmp = append(hash0[:], hash1[:]...) - hashes = append(hashesTmp, hash2[:]...) - receivedHashes map[string][]byte = make(map[string][]byte) - wait0 = make(chan bool) - wait2 = make(chan bool) - batchDone = make(chan bool) + hash0 = sha3.Sum256([]byte{0}) + hash1 = sha3.Sum256([]byte{1}) + hash2 = sha3.Sum256([]byte{2}) + hashesTmp = append(hash0[:], hash1[:]...) + hashes = append(hashesTmp, hash2[:]...) ) type testClient struct { - t []byte + t []byte + wait0 chan bool + wait2 chan bool + batchDone chan bool + receivedHashes map[string][]byte +} + +func newTestClient(t []byte) *testClient { + return &testClient{ + t: t, + wait0: make(chan bool), + wait2: make(chan bool), + batchDone: make(chan bool), + receivedHashes: make(map[string][]byte), + } } type testServer struct { t []byte } +func newTestServer(t []byte) *testServer { + return &testServer{ + t: t, + } +} + func (self *testClient) NeedData(hash []byte) func() { - receivedHashes[string(hash)] = hash + self.receivedHashes[string(hash)] = hash if bytes.Equal(hash, hash0[:]) { return func() { - <-wait0 + <-self.wait0 } } else if bytes.Equal(hash, hash2[:]) { return func() { - <-wait2 + <-self.wait2 } } return nil } func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { - close(batchDone) + close(self.batchDone) return nil } @@ -99,9 +115,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { } streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) { - return &testClient{ - t: t, - }, nil + return newTestClient(t), nil }) peerID := tester.IDs[0] @@ -112,24 +126,56 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { t.Fatalf("Expected no error, got %v", err) } - err = tester.TestExchanges(p2ptest.Exchange{ - Label: "Subscribe message", - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 4, - Msg: &SubscribeMsg{ - Stream: stream, - History: &Range{ - From: 5, - To: 8, + err = tester.TestExchanges( + p2ptest.Exchange{ + Label: "Subscribe message", + Expects: []p2ptest.Expect{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: stream, + History: &Range{ + From: 5, + To: 8, + }, + Priority: Top, }, - Priority: Top, + Peer: peerID, }, - Peer: peerID, }, }, - }) - + // trigger OfferedHashesMsg to actually create the client + p2ptest.Exchange{ + Label: "OfferedHashes message", + Triggers: []p2ptest.Trigger{ + { + Code: 1, + Msg: &OfferedHashesMsg{ + HandoverProof: &HandoverProof{ + Handover: &Handover{}, + }, + Hashes: hashes, + From: 5, + To: 8, + Stream: stream, + }, + Peer: peerID, + }, + }, + Expects: []p2ptest.Expect{ + { + Code: 2, + Msg: &WantedHashesMsg{ + Stream: stream, + Want: []byte{5}, + From: 8, + To: 0, + }, + Peer: peerID, + }, + }, + }, + ) if err != nil { t.Fatal(err) } @@ -167,9 +213,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { stream := NewStream("foo", nil, false) streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { - return &testServer{ - t: t, - }, nil + return newTestServer(t), nil }) peerID := tester.IDs[0] @@ -198,10 +242,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { HandoverProof: &HandoverProof{ Handover: &Handover{}, }, - Hashes: make([]byte, HashSize), - From: 6, - To: 9, - Initial: true, + Hashes: make([]byte, HashSize), + From: 6, + To: 9, }, Peer: peerID, }, @@ -230,6 +273,72 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { } } +func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(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 newTestServer(t), nil + }) + + peerID := tester.IDs[0] + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "Subscribe message", + Triggers: []p2ptest.Trigger{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: stream, + Priority: Top, + }, + Peer: peerID, + }, + }, + Expects: []p2ptest.Expect{ + { + Code: 1, + Msg: &OfferedHashesMsg{ + Stream: stream, + HandoverProof: &HandoverProof{ + Handover: &Handover{}, + }, + Hashes: make([]byte, HashSize), + From: 1, + To: 1, + }, + Peer: peerID, + }, + }, + }) + + if err != nil { + t.Fatal(err) + } + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "unsubscribe message", + Triggers: []p2ptest.Trigger{ + { + Code: 0, + Msg: &UnsubscribeMsg{ + Stream: stream, + }, + Peer: peerID, + }, + }, + }) + + if err != nil { + t.Fatal(err) + } +} + func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(t) defer teardown() @@ -238,9 +347,7 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { } streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { - return &testServer{ - t: t, - }, nil + return newTestServer(t), nil }) stream := NewStream("bar", nil, true) @@ -325,7 +432,6 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { // Hashes: make([]byte, HashSize), // From: 6, // To: 9, -// Initial: true, // }, // Peer: peerID, // }, @@ -339,7 +445,6 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { // From: 1, // To: 1, // Hashes: make([]byte, HashSize), -// Initial: true, // }, // Peer: peerID, // }, @@ -360,10 +465,11 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { stream := NewStream("foo", nil, true) + var tc *testClient + streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) { - return &testClient{ - t: t, - }, nil + tc = newTestClient(t) + return tc, nil }) peerID := tester.IDs[0] @@ -424,28 +530,28 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { t.Fatal(err) } - if len(receivedHashes) != 3 { - t.Fatalf("Expected number of received hashes %v, got %v", 3, len(receivedHashes)) + if len(tc.receivedHashes) != 3 { + t.Fatalf("Expected number of received hashes %v, got %v", 3, len(tc.receivedHashes)) } - close(wait0) + close(tc.wait0) timeout := time.NewTimer(100 * time.Millisecond) defer timeout.Stop() select { - case <-batchDone: + case <-tc.batchDone: t.Fatal("batch done early") case <-timeout.C: } - close(wait2) + close(tc.wait2) timeout2 := time.NewTimer(10000 * time.Millisecond) defer timeout2.Stop() select { - case <-batchDone: + case <-tc.batchDone: case <-timeout2.C: t.Fatal("timeout waiting batchdone call") } From 9eab59c0c8fc946c604d02fe3ae728c2efd74684 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Wed, 14 Feb 2018 14:23:43 +0100 Subject: [PATCH 3/4] swarm/network/stream: leveldb intervals store --- swarm/network/stream/common_test.go | 8 +- swarm/network/stream/intervals/dbstore.go | 78 +++++++++++++++++++ .../network/stream/intervals/dbstore_test.go | 40 ++++++++++ swarm/network/stream/intervals/intervals.go | 52 +++++++++++++ swarm/network/stream/intervals/store.go | 7 +- swarm/network/stream/intervals/store_test.go | 5 +- swarm/network/stream/peer.go | 1 + swarm/network/stream/stream.go | 14 +++- 8 files changed, 200 insertions(+), 5 deletions(-) create mode 100644 swarm/network/stream/intervals/dbstore.go create mode 100644 swarm/network/stream/intervals/dbstore_test.go diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 46d6314624..102faa8287 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -88,18 +88,22 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora if err != nil { return nil, nil, nil, func() {}, err } - teardown := func() { + removeDataDir := func() { os.RemoveAll(datadir) } localStore, err := storage.NewTestLocalStoreForAddr(datadir, addr.Over()) if err != nil { - return nil, nil, nil, teardown, err + return nil, nil, nil, removeDataDir, err } db := storage.NewDBAPI(localStore) delivery := NewDelivery(to, db) streamer := NewRegistry(addr, delivery, localStore, intervals.NewMemStore(), defaultSkipCheck) + teardown := func() { + streamer.Close() + removeDataDir() + } protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol) err = waitForPeers(streamer, 1*time.Second, 1) diff --git a/swarm/network/stream/intervals/dbstore.go b/swarm/network/stream/intervals/dbstore.go new file mode 100644 index 0000000000..1849c09433 --- /dev/null +++ b/swarm/network/stream/intervals/dbstore.go @@ -0,0 +1,78 @@ +// 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 ( + "github.com/syndtr/goleveldb/leveldb" +) + +// DBStore uses LevelDB to store intervals. +type DBStore struct { + db *leveldb.DB +} + +// NewDBStore creates a new instance of DBStore. +func NewDBStore(path string) (s *DBStore, err error) { + db, err := leveldb.OpenFile(path, nil) + if err != nil { + return nil, err + } + return &DBStore{ + db: db, + }, nil +} + +// Get retrieves Intervals for a specific key. If there is no Intervals +// ErrNotFound is returned. +func (s *DBStore) Get(key string) (i *Intervals, err error) { + k := []byte(key) + has, err := s.db.Has(k, nil) + if err != nil { + return nil, ErrNotFound + } + if !has { + return nil, ErrNotFound + } + data, err := s.db.Get(k, nil) + if err == leveldb.ErrNotFound { + err = ErrNotFound + } + i = &Intervals{} + if err = i.UnmarshalBinary(data); err != nil { + return nil, err + } + return i, err +} + +// Put stores Intervals for a specific key. +func (s *DBStore) Put(key string, i *Intervals) (err error) { + data, err := i.MarshalBinary() + if err != nil { + return err + } + return s.db.Put([]byte(key), data, nil) +} + +// Delete removes Intervals stored under a specific key. +func (s *DBStore) Delete(key string) (err error) { + return s.db.Delete([]byte(key), nil) +} + +// Close releases the resources used by the underlying LevelDB. +func (s *DBStore) Close() error { + return s.db.Close() +} diff --git a/swarm/network/stream/intervals/dbstore_test.go b/swarm/network/stream/intervals/dbstore_test.go new file mode 100644 index 0000000000..75a7ddbfb4 --- /dev/null +++ b/swarm/network/stream/intervals/dbstore_test.go @@ -0,0 +1,40 @@ +// 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 ( + "io/ioutil" + "os" + "testing" +) + +// TestDBStore tests basic functionality of DBStore. +func TestDBStore(t *testing.T) { + dir, err := ioutil.TempDir("", "intervals_test_db_store") + if err != nil { + panic(err) + } + defer os.RemoveAll(dir) + + store, err := NewDBStore(dir) + if err != nil { + t.Fatal(err) + } + defer store.Close() + + testStore(t, store) +} diff --git a/swarm/network/stream/intervals/intervals.go b/swarm/network/stream/intervals/intervals.go index 4a53e1a9e7..5fd820da87 100644 --- a/swarm/network/stream/intervals/intervals.go +++ b/swarm/network/stream/intervals/intervals.go @@ -17,7 +17,9 @@ package intervals import ( + "bytes" "fmt" + "strconv" "sync" ) @@ -152,3 +154,53 @@ func (i *Intervals) Last() (end uint64) { func (i *Intervals) String() string { return fmt.Sprint(i.ranges) } + +// MarshalBinary encodes Intervals parameters into a semicolon separated list. +// The first element in the list is base36-encoded start value. The following +// elements are two base36-encoded value ranges separated by comma. +func (i *Intervals) MarshalBinary() (data []byte, err error) { + d := make([][]byte, len(i.ranges)+1) + d[0] = []byte(strconv.FormatUint(i.start, 36)) + for j := range i.ranges { + r := i.ranges[j] + d[j+1] = []byte(strconv.FormatUint(r[0], 36) + "," + strconv.FormatUint(r[1], 36)) + } + return bytes.Join(d, []byte(";")), nil +} + +// UnmarshalBinary decodes data according to the Intervals.MarshalBinary format. +func (i *Intervals) UnmarshalBinary(data []byte) (err error) { + d := bytes.Split(data, []byte(";")) + l := len(d) + if l == 0 { + return nil + } + if l >= 1 { + i.start, err = strconv.ParseUint(string(d[0]), 36, 64) + if err != nil { + return err + } + } + if l == 1 { + return nil + } + + i.ranges = make([][2]uint64, 0, l-1) + for j := 1; j < l; j++ { + r := bytes.SplitN(d[j], []byte(","), 2) + if len(r) < 2 { + return fmt.Errorf("range %d has less then 2 elements", j) + } + start, err := strconv.ParseUint(string(r[0]), 36, 64) + if err != nil { + return fmt.Errorf("parsing the first element in range %d: %v", j, err) + } + end, err := strconv.ParseUint(string(r[1]), 36, 64) + if err != nil { + return fmt.Errorf("parsing the second element in range %d: %v", j, err) + } + i.ranges = append(i.ranges, [2]uint64{start, end}) + } + + return nil +} diff --git a/swarm/network/stream/intervals/store.go b/swarm/network/stream/intervals/store.go index 87b06c7ed4..9b56f8b585 100644 --- a/swarm/network/stream/intervals/store.go +++ b/swarm/network/stream/intervals/store.go @@ -14,7 +14,6 @@ // 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 ( @@ -33,6 +32,7 @@ type Store interface { Get(key string) (i *Intervals, err error) Put(key string, i *Intervals) (err error) Delete(key string) (err error) + Close() error } // MemStore is the reference implementation of Store interface that is supposed @@ -82,3 +82,8 @@ func (s *MemStore) Delete(key string) (err error) { delete(s.db, key) return nil } + +// Close doesnot do anything. +func (s *MemStore) Close() error { + return nil +} diff --git a/swarm/network/stream/intervals/store_test.go b/swarm/network/stream/intervals/store_test.go index 9a30b5d2e0..0b7344345f 100644 --- a/swarm/network/stream/intervals/store_test.go +++ b/swarm/network/stream/intervals/store_test.go @@ -20,8 +20,11 @@ import "testing" // TestMemStore tests basic functionality of MemStore. func TestMemStore(t *testing.T) { - s := NewMemStore() + testStore(t, NewMemStore()) +} +// testStore is a helper function to test various Store implementations. +func testStore(t *testing.T, s Store) { key1 := "key1" i1 := NewIntervals(0) i1.Add(10, 20) diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 65dbefea5d..c9b51c43ba 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -240,6 +240,7 @@ func (p *Peer) setClientNolock(s Stream, from, to uint64) (c *client, err error) Client: is, stream: s, priority: cp.priority, + to: to, next: next, intervalsStore: p.streamer.intervalsStore, intervalsKey: intervalsKey, diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 85c9a79355..90bf23c641 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -190,6 +190,11 @@ func (r *Registry) PeerInfo(id discover.NodeID) interface{} { return nil } +func (r *Registry) Close() error { + r.store.Close() + return r.intervalsStore.Close() +} + func (r *Registry) getPeer(peerId discover.NodeID) *Peer { r.peersMu.RLock() defer r.peersMu.RUnlock() @@ -286,6 +291,7 @@ type client struct { stream Stream priority uint8 sessionAt uint64 + to uint64 next chan error intervalsKey string @@ -348,7 +354,13 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error if err := c.AddInterval(tp.Takeover.Start, tp.Takeover.End); err != nil { return err } - return p.SendPriority(tp, c.priority) + if err := p.SendPriority(tp, c.priority); err != nil { + return err + } + if c.to > 0 && tp.Takeover.End >= c.to { + return p.streamer.Unsubscribe(p.Peer.ID(), req.Stream) + } + return nil } return nil } From e6ec8007080bdb76208be2ef12ebe8fe194b53a4 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Mon, 19 Feb 2018 23:00:37 +0100 Subject: [PATCH 4/4] swarm/network/stream: start of TestIntervals implementation --- swarm/network/stream/common_test.go | 150 +++++++++++++++++++- swarm/network/stream/intervals_test.go | 188 +++++++++++++++++++++++++ swarm/network/stream/stream.go | 22 --- swarm/network/stream/streamer_test.go | 20 +-- 4 files changed, 346 insertions(+), 34 deletions(-) create mode 100644 swarm/network/stream/intervals_test.go diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 102faa8287..b00c3f452f 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -17,18 +17,24 @@ package stream import ( + "context" "errors" "flag" + "fmt" + "io" "io/ioutil" "os" "sync/atomic" "testing" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" + "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" "github.com/ethereum/go-ethereum/swarm/storage" @@ -46,7 +52,8 @@ var ( ) var services = adapters.Services{ - "streamer": NewStreamerService, + "streamer": NewStreamerService, + "intervalsStreamer": newIntervalsStreamerService, } func init() { @@ -75,7 +82,7 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) { go func() { waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id)) }() - return r, nil + return &TestRegistry{Registry: r}, nil } func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) { @@ -155,3 +162,142 @@ func (rrs *roundRobinStore) Close() { store.Close() } } + +type TestRegistry struct { + *Registry +} + +func (r *TestRegistry) APIs() []rpc.API { + a := r.Registry.APIs() + a = append(a, rpc.API{ + Namespace: "stream", + Version: "0.1", + Service: r, + Public: true, + }) + return a +} + +func readAll(dpa *storage.DPA, hash []byte) (int64, error) { + r := dpa.Retrieve(hash) + buf := make([]byte, 1024) + var n int + var total int64 + var err error + for (total == 0 || n > 0) && err == nil { + n, err = r.ReadAt(buf, total) + total += int64(n) + } + if err != nil && err != io.EOF { + return total, err + } + return total, nil +} + +func (r *TestRegistry) ReadAll(hash common.Hash) (int64, error) { + return readAll(r.api.dpa, hash[:]) +} + +type TestExternalRegistry struct { + *Registry + hashesChan chan []byte +} + +func (r *TestExternalRegistry) APIs() []rpc.API { + a := r.Registry.APIs() + a = append(a, rpc.API{ + Namespace: "stream", + Version: "0.1", + Service: r, + Public: true, + }) + return a +} + +func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.NodeID, s Stream) (*rpc.Subscription, error) { + + peer := r.getPeer(peerId) + + client, err := peer.getClient(s) + if err != nil { + return nil, err + } + + c := client.Client.(*testExternalClient) + + notifier, supported := rpc.NotifierFromContext(ctx) + if !supported { + return nil, fmt.Errorf("Subscribe not supported") + } + + sub := notifier.CreateSubscription() + + go func() { + for { + select { + case h := <-c.hashes: + if err := notifier.Notify(sub.ID, h); err != nil { + log.Warn(fmt.Sprintf("rpc sub notifier notify stream %s: %v", s, err)) + } + case err := <-sub.Err(): + log.Warn(fmt.Sprintf("caught subscription error in stream %s: %v", s, err)) + case <-notifier.Closed(): + log.Warn(fmt.Sprintf("rpc sub notifier closed")) + } + } + }() + + return sub, nil +} + +// TODO: merge functionalities of testExternalClient and testExternalServer +// with testClient and testServer. + +type testExternalClient struct { + t []byte + // wait0 chan bool + // batchDone chan bool + hashes chan []byte +} + +func newTestExternalClient(t []byte, hashesChan chan []byte) *testExternalClient { + return &testExternalClient{ + t: t, + // wait0: make(chan bool), + // batchDone: make(chan bool), + hashes: hashesChan, + } +} + +func (self *testExternalClient) NeedData(hash []byte) func() { + self.hashes <- hash + return func() {} +} + +func (self *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { + // close(self.batchDone) + return nil +} + +func (self *testExternalClient) Close() {} + +type testExternalServer struct { + t []byte +} + +func newTestExternalServer(t []byte) *testExternalServer { + return &testExternalServer{ + t: t, + } +} + +func (self *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { + return make([]byte, HashSize), from + 1, to + 1, nil, nil +} + +func (self *testExternalServer) GetData([]byte) ([]byte, error) { + return nil, nil +} + +func (self *testExternalServer) Close() { +} diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go new file mode 100644 index 0000000000..b26d04efe3 --- /dev/null +++ b/swarm/network/stream/intervals_test.go @@ -0,0 +1,188 @@ +// 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 stream + +import ( + "context" + crand "crypto/rand" + "encoding/binary" + "fmt" + "io" + "testing" + "time" + + "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/simulations" + "github.com/ethereum/go-ethereum/p2p/simulations/adapters" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm/network" + "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" + streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing" + "github.com/ethereum/go-ethereum/swarm/storage" +) + +var externalStreamName = "externalStream" + +func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID + addr := toAddr(id) + kad := network.NewKademlia(addr.Over(), network.NewKadParams()) + store := stores[id].(*storage.LocalStore) + db := storage.NewDBAPI(store) + delivery := NewDelivery(kad, db) + deliveries[id] = delivery + netStore := storage.NewNetStore(store, nil) + hashesChan := make(chan []byte) // this chanel is only for one client, in need for more clients, create a map + r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck) + + r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Client, error) { + return newTestExternalClient(t, hashesChan), nil + }) + r.RegisterServerFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Server, error) { + return newTestExternalServer(t), nil + }) + + go func() { + waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id)) + }() + return &TestExternalRegistry{r, hashesChan}, nil +} + +func XTestIntervals(t *testing.T) { + nodes := 2 + chunkCount := dataChunkCount + skipCheck := false + + defaultSkipCheck = skipCheck + toAddr = network.NewAddrFromNodeID + conf := &streamTesting.RunConfig{ + Adapter: *adapter, + NodeCount: nodes, + ConnLevel: 1, + ToAddr: toAddr, + Services: services, + } + + sim, teardown, err := streamTesting.NewSimulation(conf) + defer teardown() + if err != nil { + t.Fatal(err) + } + + peerCount = func(id discover.NodeID) int { + return 1 + } + + dpa := storage.NewDPA(sim.Stores[0], storage.NewChunkerParams()) + dpa.Start() + size := chunkCount * chunkSize + _, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) + wait() + defer dpa.Stop() + if err != nil { + t.Fatal(err) + } + + errc := make(chan error, 1) + waitPeerErrC = make(chan error) + quitC := make(chan struct{}) + + action := func(ctx context.Context) error { + i := 0 + for err := range waitPeerErrC { + if err != nil { + return fmt.Errorf("error waiting for peers: %s", err) + } + i++ + if i == nodes { + break + } + } + + liveHashesChan := make(chan []byte) + historyHashesChan := make(chan []byte) + id := sim.IDs[1] + err := sim.CallClient(id, func(client *rpc.Client) error { + err := streamTesting.WatchDisconnections(id, client, errc, quitC) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + sid := sim.IDs[0] + err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, nil, true), &Range{From: 0, To: 5}, Top) + + if err != nil { + return err + } + // live stream + _, err = client.Subscribe(ctx, "stream_getHashes", liveHashesChan, sid, NewStream(externalStreamName, nil, true)) + if err != nil { + return err + } + // history stream + _, err = client.Subscribe(ctx, "stream_getHashes", historyHashesChan, sid, NewStream(externalStreamName, nil, false)) + return err + }) + if err != nil { + return err + } + + go func() { + for i := uint64(0); i < 5; i++ { + h := binary.BigEndian.Uint64(<-historyHashesChan) + if h != i { + errc <- fmt.Errorf("") + } + } + }() + return nil + } + check := func(ctx context.Context, id discover.NodeID) (bool, error) { + select { + case err := <-errc: + return false, err + case <-ctx.Done(): + return false, ctx.Err() + default: + } + return true, nil + } + + conf.Step = &simulations.Step{ + Action: action, + Trigger: streamTesting.Trigger(10*time.Millisecond, quitC, sim.IDs[0]), + Expect: &simulations.Expectation{ + Nodes: sim.IDs[0:1], + Check: check, + }, + } + startedAt := time.Now() + timeout := 300 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + result, err := sim.Run(ctx, conf) + finishedAt := time.Now() + if err != nil { + t.Fatalf("Setting up simulation failed: %v", err) + } + if result.Error != nil { + t.Fatalf("Simulation failed: %s", result.Error) + } + streamTesting.CheckResult(t, result, startedAt, finishedAt) +} diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 90bf23c641..b981a0ef8b 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -18,11 +18,9 @@ package stream import ( "fmt" - "io" "math" "sync" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rpc" @@ -455,26 +453,6 @@ func NewAPI(r *Registry, store storage.ChunkStore) *API { } } -func readAll(dpa *storage.DPA, hash []byte) (int64, error) { - r := dpa.Retrieve(hash) - buf := make([]byte, 1024) - var n int - var total int64 - var err error - for (total == 0 || n > 0) && err == nil { - n, err = r.ReadAt(buf, total) - total += int64(n) - } - if err != nil && err != io.EOF { - return total, err - } - return total, nil -} - -func (api *API) ReadAll(hash common.Hash) (int64, error) { - return readAll(api.dpa, hash[:]) -} - func (api *API) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error { return api.streamer.Subscribe(peerId, s, history, priority) } diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index cf00de3771..1edd1fc03b 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -65,16 +65,6 @@ func newTestClient(t []byte) *testClient { } } -type testServer struct { - t []byte -} - -func newTestServer(t []byte) *testServer { - return &testServer{ - t: t, - } -} - func (self *testClient) NeedData(hash []byte) func() { self.receivedHashes[string(hash)] = hash if bytes.Equal(hash, hash0[:]) { @@ -96,6 +86,16 @@ func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*Takeo func (self *testClient) Close() {} +type testServer struct { + t []byte +} + +func newTestServer(t []byte) *testServer { + return &testServer{ + t: t, + } +} + func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { return make([]byte, HashSize), from + 1, to + 1, nil, nil }