From 17a5afba53e7a7433b5681998b1421359c253462 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Mon, 19 Feb 2018 23:00:37 +0100 Subject: [PATCH] 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 }