From fa602ad9240851cf098f4c0c55b9556c7c057220 Mon Sep 17 00:00:00 2001 From: zelig Date: Sat, 20 Jan 2018 04:05:33 +0100 Subject: [PATCH] swarm/network/stream: - NewStreamerService now uses netStore - processReceivedChunks fix negation error; chunks found and no Req should be ignored - processReceivedChunks fix logic: no error but ReqC doesnt block - rename mustReadAll to ReadAll - use common.Hash in RPC hash - on streamer api field - streamer API ReadAll function takes hexencoded string, byte slice aint cut it - streamer service Start/Stop start and stop api.dpa. fixes nil chunk channel issue --- swarm/network/stream/common_test.go | 4 +-- swarm/network/stream/delivery.go | 14 +++------- swarm/network/stream/delivery_test.go | 38 ++++++++++++++------------- swarm/network/stream/stream.go | 30 ++++++++------------- swarm/network/stream/syncer_test.go | 4 +-- swarm/storage/chunker.go | 4 +-- swarm/storage/netstore.go | 5 ---- 7 files changed, 39 insertions(+), 60 deletions(-) diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 10946b0664..0387abab09 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -65,8 +65,8 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) { db := storage.NewDBAPI(store.(*storage.LocalStore)) delivery := NewDelivery(kad, db) deliveries[id] = delivery - //netStore := storage.NewNetStore(store.(*storage.LocalStore), func(*storage.Chunk) error { return errors.New("not retrieved yet") }) - r := NewRegistry(addr, delivery, store) + netStore := storage.NewNetStore(store.(*storage.LocalStore), func(*storage.Chunk) error { return nil }) + r := NewRegistry(addr, delivery, netStore) RegisterSwarmSyncerServer(r, db) RegisterSwarmSyncerClient(r, db) go func() { diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index d242dd01f3..9d485eb7ec 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -18,10 +18,8 @@ package stream import ( "errors" - "fmt" "time" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/storage" @@ -151,27 +149,21 @@ type ChunkDeliveryMsg struct { } func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error { - chunk, err := d.db.Get(req.Key) - if err != nil { - return err - } - d.receiveC <- req - - log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, d)) return nil } func (d *Delivery) processReceivedChunks() { for req := range d.receiveC { + // this should be has locally chunk, err := d.db.Get(req.Key) - if err != nil { + if err == nil && chunk.ReqC == nil { continue } - chunk.SData = req.SData select { case <-chunk.ReqC: default: + chunk.SData = req.SData d.db.Put(chunk) close(chunk.ReqC) } diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 364e7ddfaa..5a8ed5ec8e 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -25,6 +25,7 @@ import ( "testing" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/simulations" @@ -335,26 +336,19 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams()) rrdpa.Start() fileHash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size)) - defer rrdpa.Stop() - if err != nil { - t.Fatal(err.Error()) - } // wait until all chunks stored - // TODO: is wait() necessary? wait() - // each node Subscribes to each other's swarmChunkServerStreamName - // need to wait till an aynchronous process registers the peers in streamer.peers - // that is used by Subscribe - // time.Sleep(1 * time.Second) - // err := streamer.Subscribe(p.ID(), swarmChunkServerStreamName, nil, 0, 0, Top, true) + rrdpa.Stop() if err != nil { t.Fatal(err.Error()) } - waitPeerErrC = make(chan error) - action := func(context.Context) error { - + // each node Subscribes to each other's swarmChunkServerStreamName + // need to wait till an aynchronous process registers the peers in streamer.peers + // that is used by Subscribe + // using a global err channel to share betweem action and node service + waitPeerErrC = make(chan error) i := 0 for err := range waitPeerErrC { if err != nil { @@ -366,6 +360,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) } } + // each node subscribes to the upstream swarm chunk server stream + // which responds to chunk retrieve requests all but the last node in the chain does not for i := 0; i < len(sim.IDs)-1; i++ { id := sim.IDs[i] node := sim.Net.GetNode(id) @@ -376,6 +372,9 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) if err != nil { return fmt.Errorf("error getting node client: %s", err) } + // rpc call to streamer API subscribing to chunk Server to their + // unique upstream except for the last node in the chain + // Note in this test we only test one direction sid := sim.IDs[i+1] if err := client.Call(nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false); err != nil { return fmt.Errorf("error subscribing: %s", err) @@ -384,16 +383,18 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) // create a retriever dpa for the pivot node delivery := deliveries[sim.IDs[0]] - dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) }) + retrieveFunc := func(chunk *storage.Chunk) error { + return delivery.RequestFromPeers(chunk.Key[:], skipCheck) + } + dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), retrieveFunc) dpa := storage.NewDPA(dpacs, storage.NewChunkerParams()) dpa.Start() go func() { defer dpa.Stop() - log.Debug(fmt.Sprintf("retrieve %v", fileHash)) // start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks // we must wait for the peer connections to have started before requesting - n, err := mustReadAll(dpa, fileHash) + n, err := readAll(dpa, fileHash) log.Debug(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err) }() return nil @@ -417,8 +418,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) var total int64 ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() - err = client.CallContext(ctx, &total, "stream_readAll", fileHash) - // total, err := mustReadAll(dpa, fileHash) + // call RPC method to streamer API readAll method to check local availability + err = client.CallContext(ctx, &total, "stream_readAll", common.BytesToHash(fileHash)) log.Debug(fmt.Sprintf("check if %08x is available locally: number of bytes read %v/%v (error: %v)", fileHash, total, size, err)) if err != nil || total != int64(size) { return false, nil @@ -439,6 +440,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) conf.Step = &simulations.Step{ Action: action, Trigger: trigger, + // we are only testing the pivot node (net.Nodes[0]) Expect: &simulations.Expectation{ Nodes: sim.IDs[0:1], Check: check, diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 7cd54fede9..8e3e3ea223 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -22,6 +22,7 @@ import ( "math" "sync" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rpc" @@ -44,6 +45,7 @@ const ( // Registry registry for outgoing and incoming streamer constructors type Registry struct { + api *API addr *network.BzzAddr clientMu sync.RWMutex serverMu sync.RWMutex @@ -65,6 +67,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkS peers: make(map[discover.NodeID]*Peer), delivery: delivery, } + streamer.api = NewAPI(streamer, streamer.store) delivery.getPeer = streamer.getPeer streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, t []byte) (Server, error) { return NewSwarmChunkServer(delivery.db), nil @@ -271,7 +274,7 @@ type Client interface { BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) } -// NextBatch adjusts the indexes by inspecting the intervals +// nextBatch adjusts the indexes by inspecting the intervals func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) { var intervals []uint64 if c.live { @@ -302,7 +305,7 @@ func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) { return nextFrom, nextTo } -// Spec is the spec of the streamer protocol. +// Spec is the spec of the streamer protocol var Spec = &protocols.Spec{ Name: "stream", Version: 1, @@ -336,17 +339,19 @@ func (r *Registry) APIs() []rpc.API { { Namespace: "stream", Version: "0.1", - Service: NewAPI(r, r.store), + Service: r.api, Public: true, }, } } func (r *Registry) Start(server *p2p.Server) error { + r.api.dpa.Start() return nil } func (r *Registry) Stop() error { + r.api.dpa.Stop() return nil } @@ -363,7 +368,7 @@ func NewAPI(r *Registry, store storage.ChunkStore) *API { } } -func mustReadAll(dpa *storage.DPA, hash []byte) (int64, error) { +func readAll(dpa *storage.DPA, hash []byte) (int64, error) { r := dpa.Retrieve(hash) buf := make([]byte, 1024) var n int @@ -379,21 +384,8 @@ func mustReadAll(dpa *storage.DPA, hash []byte) (int64, error) { return total, nil } -func (api *API) ReadAll(hash []byte) (int64, error) { - r := api.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 - //return mustReadAll(api.dpa, hash) +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 { diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 0b90658ec4..9dde5bda72 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -35,7 +35,9 @@ import ( func TestSyncerSimulation(t *testing.T) { testSyncBetweenNodes(t, 2, 1, 81000, true, 1) + testSyncBetweenNodes(t, 2, 1, 81000, false, 1) testSyncBetweenNodes(t, 3, 1, 81000, true, 1) + testSyncBetweenNodes(t, 3, 1, 81000, false, 1) } func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool, po uint8) { @@ -59,7 +61,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool, } stores = make(map[discover.NodeID]storage.ChunkStore) deliveries = make(map[discover.NodeID]*Delivery) - log.Warn("Stores", "len", len(sim.Stores)) for i, id := range sim.IDs { stores[id] = sim.Stores[i] } @@ -136,7 +137,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool, var found, total int for i := 1; i < nodes; i++ { - dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool { _, err := dbs[0].Get(key) if err == nil { diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index 2ea81403bf..9ba6f5c1e0 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -13,7 +13,6 @@ // // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . - package storage import ( @@ -463,8 +462,7 @@ func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk { case <-chunk.C: // bells are ringing, data have been delivered } if len(chunk.SData) == 0 { - return nil // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - + return nil } return chunk } diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go index f01ffe4a69..334cf3635a 100644 --- a/swarm/storage/netstore.go +++ b/swarm/storage/netstore.go @@ -18,10 +18,7 @@ package storage import ( "encoding/binary" - "fmt" "time" - - "github.com/ethereum/go-ethereum/log" ) // NetStore implements the ChunkStore interface, @@ -43,7 +40,6 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) { var created bool chunk, created = self.localStore.GetOrCreateRequest(key) if chunk.ReqC == nil { - log.Trace(fmt.Sprintf("DPA.Get: %v found locally, %d bytes", key.Log(), len(chunk.SData))) return } @@ -57,7 +53,6 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) { select { case <-t.C: - log.Trace(fmt.Sprintf("DPA.Get: %v request time out ", key.Log())) return nil, notFound case <-chunk.ReqC: }