From ebec92e928d40d6baccfdab81212dfd59ffe58bc Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Tue, 23 Jan 2018 14:10:34 +0100 Subject: [PATCH 01/15] swarm: an attempt to debug syncer tests --- swarm/network/stream/delivery.go | 28 ++++++++++---- swarm/network/stream/delivery_test.go | 11 +++++- swarm/network/stream/messages.go | 3 ++ swarm/network/stream/syncer.go | 4 +- swarm/network/stream/syncer_test.go | 50 +++++++++++++++++++------ swarm/network/stream/testing/testing.go | 5 +-- swarm/storage/localstore.go | 1 - 7 files changed, 76 insertions(+), 26 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 4355997fd0..606be1703a 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -33,11 +33,14 @@ const ( ) type Delivery struct { - db *storage.DBAPI - overlay network.Overlay - receiveC chan *ChunkDeliveryMsg - getPeer func(discover.NodeID) *Peer - quit chan struct{} + db *storage.DBAPI + overlay network.Overlay + receiveC chan *ChunkDeliveryMsg + getPeer func(discover.NodeID) *Peer + quit chan struct{} + counterIn int + counterDone int + counterHash int } func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery { @@ -159,6 +162,7 @@ type ChunkDeliveryMsg struct { } func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error { + d.counterIn++ d.receiveC <- req return nil } @@ -182,10 +186,11 @@ R: chunk.SData = req.SData d.db.Put(chunk) log.Warn("reecived delivery", "hash", chunk.Key) - chunk.WaitToStore() - log.Warn("received delivery stored", "hash", chunk.Key) close(chunk.ReqC) + chunk.WaitToStore() + //log.Warn("received delivery stored", "hash", chunk.Key) log.Warn("received delivery requesters notified", "hash", chunk.Key) + d.counterDone++ } } @@ -220,3 +225,12 @@ func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ... } return errors.New("no peer found") } + +func (d *Delivery) PrintCounters(id discover.NodeID) { + if d.counterHash != d.counterDone { + log.Error(fmt.Sprintf("delivery %s: HASH and DONE not the same", id)) + } + log.Error(fmt.Sprintf("delivery %s chunks hash: %d", id, d.counterHash)) + log.Error(fmt.Sprintf("delivery %s chunks in: %d", id, d.counterIn)) + log.Error(fmt.Sprintf("delivery %s chunks done: %d", id, d.counterDone)) +} diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 1ca89ea5e6..904830535d 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -449,7 +449,10 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck }, } startedAt := time.Now() - result, err := sim.Run(conf) + 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) @@ -586,7 +589,11 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // run the simulation in the background errc := make(chan error) go func() { - _, err := sim.Run(conf) + timeout := 300 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + _, err := sim.Run(ctx, conf) errc <- err }() diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 1cf9aeab43..24a63391d8 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -121,6 +121,9 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { wg := sync.WaitGroup{} for i := 0; i < len(hashes); i += HashSize { hash := hashes[i : i+HashSize] + + p.streamer.delivery.counterHash++ + if wait := s.NeedData(hash); wait != nil { want.Set(i/HashSize, true) wg.Add(1) diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index eff42b4cd0..9df6ea78d5 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -75,7 +75,9 @@ func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) { // GetSection retrieves the actual chunk from localstore func (s *SwarmSyncerServer) GetData(key []byte) []byte { chunk, err := s.db.Get(storage.Key(key)) - if err != nil { + if err == storage.ErrFetching { + <-chunk.ReqC + } else if err != nil { return nil } return chunk.SData diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index d57be133cb..57ec1dd59b 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -34,7 +34,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -const dataChunkCount = 500 +const dataChunkCount = 1000 func TestSyncerSimulation(t *testing.T) { // testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1) @@ -67,6 +67,16 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck if err != nil { t.Fatal(err.Error()) } + + defer func() { + for _, id := range sim.IDs { + deliveries[id].PrintCounters(id) + } + // for id, delivery := range deliveries { + // delivery.PrintCounters(id) + // } + }() + stores = make(map[discover.NodeID]storage.ChunkStore) deliveries = make(map[discover.NodeID]*Delivery) for i, id := range sim.IDs { @@ -91,14 +101,16 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } // collect hashes in po 1 from all nodes - var hashes []storage.Key + hashes := make([][]storage.Key, nodes) dbs := make([]*storage.DBAPI, nodes) for i := 0; i < nodes; i++ { dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore)) } + totalHashes := 0 for i := 1; i < nodes; i++ { dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool { - hashes = append(hashes, key) + hashes[i] = append(hashes[i], key) + totalHashes++ return true }) } @@ -149,15 +161,28 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } var found int - total := len(hashes) - for _, key := range hashes { - _, err := dbs[0].Get(key) - if err == nil { - found++ + for i, n := range hashes { + for _, key := range n { + chunk, err := dbs[0].Get(key) + if err == storage.ErrFetching { + <-chunk.ReqC + found++ + } else if err == nil { + found++ + } + + log.Error("staring dbs check", "key", key) + for j := i; j > 0; j-- { + _, err := dbs[j].Get(key) + if err != nil { + log.Error("get from node", "node", sim.IDs[j], "nodeID", j, "key", key.Hex(), "err", err) + break + } + } } } - log.Error("sync check", "bin", po, "found", found, "total", total) - pass := found == total + log.Error("sync check", "bin", po, "found", found, "total", totalHashes) + pass := found == totalHashes if !pass { return false, nil } @@ -175,7 +200,10 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck }, } startedAt := time.Now() - result, err := sim.Run(conf) + timeout := 4 * 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) diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index e92b009f86..e1c50919e7 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -170,7 +170,7 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { return s, teardown, nil } -func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) { +func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.StepResult, error) { // bring up nodes, launch the servive nodes := conf.NodeCount conns := conf.ConnLevel @@ -204,9 +204,6 @@ func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) { // create an only locally retrieving dpa for the pivot node to test // if retriee requests have arrived - timeout := 300 * time.Second - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() result := simulations.NewSimulation(s.Net).Run(ctx, conf.Step) return result, nil } diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index ac6d642950..066a27028a 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -106,7 +106,6 @@ func (self *LocalStore) Put(chunk *Chunk) { // ChunkStores are remote and can have long latency func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { chunk, err = self.memStore.Get(key) - if err == nil { if chunk.ReqC != nil { select { From 97e4497c49bb0c40ae9949bc1d3decffc8115f75 Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 23 Jan 2018 15:06:37 +0100 Subject: [PATCH 02/15] debug --- swarm/network/stream/syncer_test.go | 50 +++++++++++++++++------------ 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 57ec1dd59b..72f9b60568 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -107,10 +107,15 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore)) } totalHashes := 0 - for i := 1; i < nodes; i++ { + hashCounts := make([]int, nodes) + for i := nodes - 1; i >= 0; i-- { + if i < nodes-1 { + hashCounts[i] = hashCounts[i+1] + } dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool { hashes[i] = append(hashes[i], key) totalHashes++ + hashCounts[i]++ return true }) } @@ -160,29 +165,34 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck default: } - var found int - for i, n := range hashes { - for _, key := range n { - chunk, err := dbs[0].Get(key) - if err == storage.ErrFetching { - <-chunk.ReqC - found++ - } else if err == nil { - found++ - } - - log.Error("staring dbs check", "key", key) - for j := i; j > 0; j-- { - _, err := dbs[j].Get(key) - if err != nil { - log.Error("get from node", "node", sim.IDs[j], "nodeID", j, "key", key.Hex(), "err", err) - break + var pass bool + var i int + log.Error("staring dbs check") + for i = nodes - 1; i >= 0; i-- { + nodeHashCount := hashCounts[i] + nodeHashFound := 0 + for j := i; j < nodes; j++ { + nodeHashes := hashes[j] + for _, key := range nodeHashes { + chunk, err := dbs[i].Get(key) + if err == storage.ErrFetching { + <-chunk.ReqC + nodeHashFound++ + } else if err == nil { + nodeHashFound++ + } else { + log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) } } } + log.Error("sync check", "node", sim.IDs[i], "index", i, "bin", po, "found", nodeHashFound, "total", nodeHashCount) + pass = nodeHashFound == nodeHashCount + if !pass { + break + } } - log.Error("sync check", "bin", po, "found", found, "total", totalHashes) - pass := found == totalHashes + // log.Error("sync check", "bin", po, "found", found, "total", totalHashes) + // pass := found == totalHashes if !pass { return false, nil } From 172eb9e14d16669d71386307d246cf8d6bfcd680 Mon Sep 17 00:00:00 2001 From: Balint Gabor Date: Tue, 23 Jan 2018 17:50:32 +0100 Subject: [PATCH 03/15] swarm/storage, swarm/network: Fix delivery tests --- swarm/network/stream/common_test.go | 7 ++++--- swarm/storage/localstore.go | 1 + swarm/storage/netstore.go | 27 +++++++++++++++++++-------- 3 files changed, 24 insertions(+), 11 deletions(-) diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 7997e977e9..1deb6ffba9 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -63,11 +63,12 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) { id := ctx.Config.ID addr := toAddr(id) kad := network.NewKademlia(addr.Over(), network.NewKadParams()) - store := stores[id] - db := storage.NewDBAPI(store.(*storage.LocalStore)) + store := stores[id].(*storage.LocalStore) + db := storage.NewDBAPI(store) delivery := NewDelivery(kad, db) deliveries[id] = delivery - r := NewRegistry(addr, delivery, store, defaultSkipCheck) + netStore := storage.NewNetStore(store, nil) + r := NewRegistry(addr, delivery, netStore, defaultSkipCheck) RegisterSwarmSyncerServer(r, db) RegisterSwarmSyncerClient(r, db) go func() { diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 066a27028a..898e7f18ea 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -94,6 +94,7 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error) // LocalStore is itself a chunk store // unsafe, in that the data is not integrity checked func (self *LocalStore) Put(chunk *Chunk) { + chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) self.memStore.Put(chunk) go func() { self.DbStore.Put(chunk) diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go index 265baa5337..bbf702fdc8 100644 --- a/swarm/storage/netstore.go +++ b/swarm/storage/netstore.go @@ -36,17 +36,28 @@ func NewNetStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *Net // Get is the entrypoint for local retrieve requests // waits for response or times out func (self *NetStore) Get(key Key) (chunk *Chunk, err error) { - var created bool - chunk, created = self.localStore.GetOrCreateRequest(key) - if chunk.ReqC == nil { - return chunk, nil - } - - if created { - if err := self.retrieve(chunk); err != nil { + if self.retrieve == nil { + chunk, err = self.localStore.Get(key) + if err == nil { + return chunk, nil + } + if err != ErrFetching { return nil, err } + } else { + var created bool + chunk, created = self.localStore.GetOrCreateRequest(key) + if chunk.ReqC == nil { + return chunk, nil + } + + if created { + if err := self.retrieve(chunk); err != nil { + return nil, err + } + } } + t := time.NewTicker(searchTimeout) defer t.Stop() From eae4473a81a70a03f512f2c8c88b535ea7c91e6c Mon Sep 17 00:00:00 2001 From: zelig Date: Wed, 24 Jan 2018 10:11:09 +0100 Subject: [PATCH 04/15] more debug --- swarm/network/stream/delivery.go | 24 +++++++++++++++--------- swarm/network/stream/peer.go | 8 ++++++-- swarm/network/stream/stream.go | 6 +++--- swarm/network/stream/syncer.go | 1 - swarm/network/stream/syncer_test.go | 6 ++++-- swarm/storage/dbstore.go | 16 ++++++++-------- swarm/storage/localstore.go | 6 +++--- 7 files changed, 39 insertions(+), 28 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 606be1703a..cb62e5d149 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -161,7 +161,7 @@ type ChunkDeliveryMsg struct { SData []byte // the stored chunk Data (incl size) } -func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error { +func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error { d.counterIn++ d.receiveC <- req return nil @@ -172,7 +172,9 @@ R: for req := range d.receiveC { // this should be has locally chunk, err := d.db.Get(req.Key) + log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) if err == nil { + log.Error("found existing?", "hash", chunk.Key.Hex()) continue R } if err != storage.ErrFetching { @@ -180,17 +182,21 @@ R: } select { case <-chunk.ReqC: + log.Error("someone else delivered?", "hash", chunk.Key.Hex()) continue R default: } - chunk.SData = req.SData - d.db.Put(chunk) - log.Warn("reecived delivery", "hash", chunk.Key) - close(chunk.ReqC) - chunk.WaitToStore() - //log.Warn("received delivery stored", "hash", chunk.Key) - log.Warn("received delivery requesters notified", "hash", chunk.Key) - d.counterDone++ + go func() { + chunk.SData = req.SData + log.Error("received delivery", "hash", chunk.Key.Hex()) + d.db.Put(chunk) + log.Error("put to db", "hash", chunk.Key.Hex()) + chunk.WaitToStore() + close(chunk.ReqC) + //log.Warn("received delivery stored", "hash", chunk.Key) + log.Error("requesters notified", "hash", chunk.Key.Hex()) + d.counterDone++ + }() } } diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 708a3fbcc7..ed2366f5c6 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -28,7 +28,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -var sendTimeout = 5 * time.Second +var sendTimeout = 1 * time.Second // Peer is the Peer extention for the streaming protocol type Peer struct { @@ -97,7 +97,11 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { Stream: s.stream, Key: s.key, } - log.Warn("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to) + log.Error("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to) + for i := 0; i < len(hashes); i += HashSize { + hash := hashes[i : i+HashSize] + log.Error("Swarm syncer offer hash", "peer", p.ID(), "stream", s.stream, "hash", storage.Key(hash).Hex(), "len", len(hashes), "from", from, "to", to) + } return p.SendPriority(msg, s.priority) } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 87a56483c6..a85745a539 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -38,8 +38,8 @@ const ( Mid High Top - PriorityQueue // number of queues - PriorityQueueCap = 32 // queue capacity + PriorityQueue // number of queues + PriorityQueueCap = 3 // queue capacity HashSize = 32 ) @@ -227,7 +227,7 @@ func (p *Peer) HandleMsg(msg interface{}) error { return p.handleWantedHashesMsg(msg) case *ChunkDeliveryMsg: - return p.streamer.delivery.handleChunkDeliveryMsg(msg) + return p.streamer.delivery.handleChunkDeliveryMsg(p, msg) case *RetrieveRequestMsg: return p.streamer.delivery.handleRetrieveRequestMsg(p, msg) diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index 9df6ea78d5..d85233e6df 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -186,7 +186,6 @@ func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) { chunk, _ := s.db.GetOrCreateRequest(key) // TODO: we may want to request from this peer anyway even if the request exists if chunk.ReqC == nil { - log.Error("oops this is found") return nil } // create request and wait until the chunk data arrives and is stored diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 72f9b60568..038a721e52 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -22,6 +22,7 @@ import ( "fmt" "io" "math" + "runtime/debug" "testing" "time" @@ -203,7 +204,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck conf.Step = &simulations.Step{ Action: action, - Trigger: streamTesting.PivotTrigger(100*time.Millisecond, checkC, sim.IDs[0]), + Trigger: streamTesting.PivotTrigger(500*time.Millisecond, checkC, sim.IDs[0]), Expect: &simulations.Expectation{ Nodes: sim.IDs[0:1], Check: check, @@ -220,6 +221,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } if result.Error != nil { t.Fatalf("Simulation failed: %s", result.Error) + streamTesting.CheckResult(t, result, startedAt, finishedAt) + debug.PrintStack() } - streamTesting.CheckResult(t, result, startedAt, finishedAt) } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index d8dace4758..e95cfe58d8 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -538,6 +538,7 @@ func (s *DbStore) CurrentStorageIndex() uint64 { } func (s *DbStore) Put(chunk *Chunk) { + log.Error("DbStore.Put", "hash", chunk.Key.Hex()) s.lock.Lock() defer s.lock.Unlock() @@ -549,17 +550,23 @@ func (s *DbStore) Put(chunk *Chunk) { idata, err := s.db.Get(ikey) if err != nil { s.doPut(chunk, ikey, &index, po) + batchC := s.batchC + go func() { + <-batchC + close(chunk.dbStored) + }() + log.Error("DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) } else { log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access")) decodeIndex(idata, &index) close(chunk.dbStored) + log.Error("DbStore.Put already found", "hash", chunk.Key.Hex()) } index.Access = s.accessCnt s.accessCnt++ idata = encodeIndex(&index) s.batch.Put(ikey, idata) select { - case <-s.quit: case s.batchesC <- struct{}{}: default: } @@ -579,13 +586,6 @@ func (s *DbStore) doPut(chunk *Chunk, ikey []byte, index *dpaDBIndex, po uint8) cntKey[1] = po s.batch.Put(cntKey, U64ToBytes(s.bucketCnt[po])) - batchC := s.batchC - go func() { - <-batchC - close(chunk.dbStored) - }() - - log.Trace(fmt.Sprintf("DbStore.Put: %v. db storage counter: %v ", chunk.Key.Log(), s.dataIdx)) } func (s *DbStore) writeBatches() { diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 898e7f18ea..f7cc4092d6 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -96,9 +96,9 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error) func (self *LocalStore) Put(chunk *Chunk) { chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) self.memStore.Put(chunk) - go func() { - self.DbStore.Put(chunk) - }() + log.Error("put to memstore", "hash", chunk.Key.Hex()) + self.DbStore.Put(chunk) + log.Error("put to dbstore", "hash", chunk.Key.Hex()) } // Get(chunk *Chunk) looks up a chunk in the local stores From 82e24a488d7e3d27f6ff6bf9a7a8aa25d52c5bbb Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Wed, 24 Jan 2018 13:50:38 +0100 Subject: [PATCH 05/15] More logging with fmt package --- swarm/network/stream/delivery.go | 13 +++++++------ swarm/network/stream/syncer_test.go | 7 +++---- swarm/storage/dbstore.go | 29 ++++++++++++++++++++++++++--- swarm/storage/localstore.go | 6 ++++-- 4 files changed, 40 insertions(+), 15 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index cb62e5d149..98d0a1631d 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -19,6 +19,7 @@ package stream import ( "errors" "fmt" + "os" "time" "github.com/ethereum/go-ethereum/log" @@ -172,9 +173,9 @@ R: for req := range d.receiveC { // this should be has locally chunk, err := d.db.Get(req.Key) - log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) + fmt.Fprintln(os.Stderr, "pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) if err == nil { - log.Error("found existing?", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, "found existing?", "hash", chunk.Key.Hex()) continue R } if err != storage.ErrFetching { @@ -182,19 +183,19 @@ R: } select { case <-chunk.ReqC: - log.Error("someone else delivered?", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, "someone else delivered?", "hash", chunk.Key.Hex()) continue R default: } go func() { chunk.SData = req.SData - log.Error("received delivery", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, "received delivery", "hash", chunk.Key.Hex()) d.db.Put(chunk) - log.Error("put to db", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, "put to db", "hash", chunk.Key.Hex()) chunk.WaitToStore() close(chunk.ReqC) //log.Warn("received delivery stored", "hash", chunk.Key) - log.Error("requesters notified", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, "requesters notified", "hash", chunk.Key.Hex()) d.counterDone++ }() } diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 038a721e52..dcfa81a3a5 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -22,7 +22,7 @@ import ( "fmt" "io" "math" - "runtime/debug" + "os" "testing" "time" @@ -182,7 +182,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } else if err == nil { nodeHashFound++ } else { - log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) + fmt.Fprintln(os.Stderr, time.Now(), "not found", "index", i, "origin", j, "key", key.Hex(), "err", err) } } } @@ -211,7 +211,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck }, } startedAt := time.Now() - timeout := 4 * time.Second + timeout := 30 * time.Second ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() result, err := sim.Run(ctx, conf) @@ -222,6 +222,5 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck if result.Error != nil { t.Fatalf("Simulation failed: %s", result.Error) streamTesting.CheckResult(t, result, startedAt, finishedAt) - debug.PrintStack() } } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index e95cfe58d8..f23a6bb701 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -30,7 +30,9 @@ import ( "fmt" "io" "io/ioutil" + "os" "sync" + "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rlp" @@ -538,8 +540,22 @@ func (s *DbStore) CurrentStorageIndex() uint64 { } func (s *DbStore) Put(chunk *Chunk) { - log.Error("DbStore.Put", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put", "hash", chunk.Key.Hex()) + done := make(chan struct{}) + defer close(done) + go func() { + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex()) + select { + case <-time.After(1 * time.Second): + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put WAITING", "hash", chunk.Key.Hex()) + case <-done: + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put EXITED", "hash", chunk.Key.Hex()) + } + }() + + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquiring", "hash", chunk.Key.Hex()) s.lock.Lock() + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquired", "hash", chunk.Key.Hex()) defer s.lock.Unlock() ikey := getIndexKey(chunk.Key) @@ -548,19 +564,26 @@ func (s *DbStore) Put(chunk *Chunk) { po := s.po(chunk.Key) idata, err := s.db.Get(ikey) + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get", "hash", chunk.Key.Hex(), "err", err) if err != nil { s.doPut(chunk, ikey, &index, po) batchC := s.batchC go func() { + defer func() { + if err := recover(); err != nil { + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err) + } + }() + <-batchC close(chunk.dbStored) }() - log.Error("DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) } else { log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access")) decodeIndex(idata, &index) close(chunk.dbStored) - log.Error("DbStore.Put already found", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put already found", "hash", chunk.Key.Hex()) } index.Access = s.accessCnt s.accessCnt++ diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index f7cc4092d6..00dc10d17b 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -19,7 +19,9 @@ package storage import ( "encoding/binary" "fmt" + "os" "path/filepath" + "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/swarm/storage/mock" @@ -96,9 +98,9 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error) func (self *LocalStore) Put(chunk *Chunk) { chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) self.memStore.Put(chunk) - log.Error("put to memstore", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, time.Now(), "put to memstore", "hash", chunk.Key.Hex()) self.DbStore.Put(chunk) - log.Error("put to dbstore", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, time.Now(), "put to dbstore", "hash", chunk.Key.Hex()) } // Get(chunk *Chunk) looks up a chunk in the local stores From 6d7cc72c915ae1c02ec24049d1f7868c9c3a192b Mon Sep 17 00:00:00 2001 From: Balint Gabor Date: Wed, 24 Jan 2018 15:27:25 +0100 Subject: [PATCH 06/15] swarm/storage: investigation of dbstore.Put deadlock --- swarm/storage/dbstore.go | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index f23a6bb701..25de2dcb70 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -553,38 +553,39 @@ func (s *DbStore) Put(chunk *Chunk) { } }() + ikey := getIndexKey(chunk.Key) + var index dpaDBIndex + + po := s.po(chunk.Key) + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get is being called...", "hash", chunk.Key.Hex()) + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquiring", "hash", chunk.Key.Hex()) s.lock.Lock() fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquired", "hash", chunk.Key.Hex()) defer s.lock.Unlock() - ikey := getIndexKey(chunk.Key) - var index dpaDBIndex - - po := s.po(chunk.Key) - idata, err := s.db.Get(ikey) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get", "hash", chunk.Key.Hex(), "err", err) + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err) if err != nil { s.doPut(chunk, ikey, &index, po) - batchC := s.batchC go func() { defer func() { if err := recover(); err != nil { fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err) } }() - - <-batchC - close(chunk.dbStored) }() fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) } else { log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access")) decodeIndex(idata, &index) - close(chunk.dbStored) fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put already found", "hash", chunk.Key.Hex()) } + batchC := s.batchC + go func() { + <-batchC + close(chunk.dbStored) + }() index.Access = s.accessCnt s.accessCnt++ idata = encodeIndex(&index) @@ -593,6 +594,7 @@ func (s *DbStore) Put(chunk *Chunk) { case s.batchesC <- struct{}{}: default: } + fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err) } // force putting into db, does not check access index From c55b99418b04fa2d4eb88f4daeca1a945216afed Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Wed, 24 Jan 2018 17:56:36 +0100 Subject: [PATCH 07/15] Fix dbstore iterator bug and add even more logging --- p2p/protocols/protocol.go | 2 + swarm/network/stream/delivery.go | 26 +++++++------ swarm/network/stream/delivery_test.go | 4 +- swarm/network/stream/messages.go | 14 +++---- swarm/network/stream/stream.go | 2 +- swarm/network/stream/streamer_test.go | 4 +- swarm/network/stream/syncer.go | 8 ++-- swarm/network/stream/syncer_test.go | 5 +-- swarm/network/stream/testing/testing.go | 12 ++++-- swarm/storage/dbstore.go | 51 +++++++++++++------------ swarm/storage/localstore.go | 6 +-- 11 files changed, 71 insertions(+), 63 deletions(-) diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 7b04069edf..48fc5e9fcc 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -34,6 +34,7 @@ import ( "reflect" "sync" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" ) @@ -210,6 +211,7 @@ func (p *Peer) Run(handler func(msg interface{}) error) error { // if they are useful for other protocols // overwrite Disconnect for testing, so that protocol readloop quits func (p *Peer) Drop(err error) { + log.Error("p2p protocol DROP", "err", err) p.Disconnect(p2p.DiscSubprotocolError) } diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 98d0a1631d..d397cd7b62 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -19,7 +19,6 @@ package stream import ( "errors" "fmt" - "os" "time" "github.com/ethereum/go-ethereum/log" @@ -100,9 +99,14 @@ func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64 } // GetData retrives chunk data from db store -func (s *SwarmChunkServer) GetData(key []byte) []byte { - chunk, _ := s.db.Get(storage.Key(key)) - return chunk.SData +func (s *SwarmChunkServer) GetData(key []byte) ([]byte, error) { + chunk, err := s.db.Get(storage.Key(key)) + if err == storage.ErrFetching { + <-chunk.ReqC + } else if err != nil { + return nil, err + } + return chunk.SData, nil } // RetrieveRequestMsg is the protocol msg for chunk retrieve requests @@ -141,7 +145,7 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e if req.SkipCheck { err := sp.Deliver(chunk, s.priority) if err != nil { - sp.Drop(err) + sp.Drop(fmt.Errorf("handleRetrieveRequestMsg: %v", err)) } } streamer.deliveryC <- chunk.Key[:] @@ -173,9 +177,9 @@ R: for req := range d.receiveC { // this should be has locally chunk, err := d.db.Get(req.Key) - fmt.Fprintln(os.Stderr, "pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) + log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) if err == nil { - fmt.Fprintln(os.Stderr, "found existing?", "hash", chunk.Key.Hex()) + log.Error("found existing?", "hash", chunk.Key.Hex()) continue R } if err != storage.ErrFetching { @@ -183,19 +187,19 @@ R: } select { case <-chunk.ReqC: - fmt.Fprintln(os.Stderr, "someone else delivered?", "hash", chunk.Key.Hex()) + log.Error("someone else delivered?", "hash", chunk.Key.Hex()) continue R default: } go func() { chunk.SData = req.SData - fmt.Fprintln(os.Stderr, "received delivery", "hash", chunk.Key.Hex()) + log.Error("received delivery", "hash", chunk.Key.Hex()) d.db.Put(chunk) - fmt.Fprintln(os.Stderr, "put to db", "hash", chunk.Key.Hex()) + log.Error("put to db", "hash", chunk.Key.Hex()) chunk.WaitToStore() close(chunk.ReqC) //log.Warn("received delivery stored", "hash", chunk.Key) - fmt.Fprintln(os.Stderr, "requesters notified", "hash", chunk.Key.Hex()) + log.Error("requesters notified", "hash", chunk.Key.Hex()) d.counterDone++ }() } diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 904830535d..24b35ae122 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -380,7 +380,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck // which responds to chunk retrieve requests all but the last node in the chain does not var j int err := sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, errc, quitC) + err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) if err != nil { return err } @@ -547,7 +547,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // which responds to chunk retrieve requests var j int simErrC <- sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, simErrC, quitC) + err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), simErrC, quitC) if err != nil { return err } diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 24a63391d8..a11c97332d 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -83,7 +83,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error { log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To) go func() { if err := p.SendOfferedHashes(os, req.From, req.To); err != nil { - p.Drop(err) + p.Drop(fmt.Errorf("handleSubscribeMsg SendOfferedHashes: %v", err)) } }() return nil @@ -176,7 +176,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { return case err := <-s.next: if err != nil { - p.Drop(err) + p.Drop(fmt.Errorf("handleOfferedHashesMsg next: %v", err)) return } case <-s.quit: @@ -185,7 +185,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { 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) if err != nil { - p.Drop(err) + p.Drop(fmt.Errorf("handleOfferedHashesMsg set priority: %v", err)) } }() return nil @@ -218,7 +218,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { // launch in go routine since GetBatch blocks until new hashes arrive go func() { if err := p.SendOfferedHashes(s, req.From, req.To); err != nil { - p.Drop(err) + p.Drop(fmt.Errorf("handleWantedHashesMsg SendOfferedHashes: %v", err)) } }() // go p.SendOfferedHashes(s, req.From, req.To) @@ -230,9 +230,9 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { for i := 0; i < l; i++ { if want.Get(i) { hash := hashes[i*HashSize : (i+1)*HashSize] - data := s.GetData(hash) - if data == nil { - return errors.New("not found") + data, err := s.GetData(hash) + if err != nil { + return fmt.Errorf("handleWantedHashesMsg get data %x: %v", hash, err) } chunk := storage.NewChunk(hash, nil) chunk.SData = data diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index a85745a539..ac6d027d2d 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -256,7 +256,7 @@ type server struct { // Server interface for outgoing peer Streamer type Server interface { SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) - GetData([]byte) []byte + GetData([]byte) ([]byte, error) } type client struct { diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index a905f4c963..fec23a06d3 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -81,8 +81,8 @@ func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, ui return make([]byte, HashSize), from + 1, to + 1, nil, nil } -func (self *testServer) GetData([]byte) []byte { - return nil +func (self *testServer) GetData([]byte) ([]byte, error) { + return nil, nil } func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) { diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index d85233e6df..af1bbb0b2e 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -73,14 +73,14 @@ func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) { } // GetSection retrieves the actual chunk from localstore -func (s *SwarmSyncerServer) GetData(key []byte) []byte { +func (s *SwarmSyncerServer) GetData(key []byte) ([]byte, error) { chunk, err := s.db.Get(storage.Key(key)) if err == storage.ErrFetching { <-chunk.ReqC } else if err != nil { - return nil + return nil, err } - return chunk.SData + return chunk.SData, nil } // GetBatch retrieves the next batch of hashes from the dbstore @@ -111,7 +111,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 } log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po)) - return batch, from, to + 1, nil, nil + return batch, from, to, nil, nil } // SwarmSyncerClient diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index dcfa81a3a5..263d230183 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -22,7 +22,6 @@ import ( "fmt" "io" "math" - "os" "testing" "time" @@ -142,7 +141,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck // each node Subscribes to each other's swarmChunkServerStreamName j := 0 return sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, errc, quitC) + err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) if err != nil { return err } @@ -182,7 +181,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } else if err == nil { nodeHashFound++ } else { - fmt.Fprintln(os.Stderr, time.Now(), "not found", "index", i, "origin", j, "key", key.Hex(), "err", err) + log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) } } } diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index e1c50919e7..9585de369b 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -154,9 +154,9 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { // set nodes number of Stores available stores, storeTeardown, err := SetStores(addrs...) teardown = func() { - storeTeardown() - adapterTeardown() net.Shutdown() + adapterTeardown() + storeTeardown() } if err != nil { return nil, teardown, err @@ -208,7 +208,7 @@ func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.Ste return result, nil } -func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) error { +func WatchDisconnections(id discover.NodeID, client *rpc.Client, expectedConnCount int, errc chan error, quitC chan struct{}) error { events := make(chan *p2p.PeerEvent) sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { @@ -218,10 +218,14 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error defer sub.Unsubscribe() select { case <-quitC: - return + if expectedConnCount <= 0 { + return + } case e := <-events: + expectedConnCount-- errc <- fmt.Errorf("peerEvent for node %v: %v", id, e) case err := <-sub.Err(): + expectedConnCount = 0 if err != nil { errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err) } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 25de2dcb70..41b0b67a24 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -30,7 +30,6 @@ import ( "fmt" "io" "io/ioutil" - "os" "sync" "time" @@ -540,16 +539,16 @@ func (s *DbStore) CurrentStorageIndex() uint64 { } func (s *DbStore) Put(chunk *Chunk) { - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put", "hash", chunk.Key.Hex()) + log.Error("DbStore.Put", "hash", chunk.Key.Hex()) done := make(chan struct{}) defer close(done) go func() { - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex()) + log.Error("DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex()) select { case <-time.After(1 * time.Second): - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put WAITING", "hash", chunk.Key.Hex()) + log.Error("DbStore.Put WAITING", "hash", chunk.Key.Hex()) case <-done: - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put EXITED", "hash", chunk.Key.Hex()) + log.Error("DbStore.Put EXITED", "hash", chunk.Key.Hex()) } }() @@ -557,35 +556,35 @@ func (s *DbStore) Put(chunk *Chunk) { var index dpaDBIndex po := s.po(chunk.Key) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get is being called...", "hash", chunk.Key.Hex()) + log.Error("DbStore.db.Get is being called...", "hash", chunk.Key.Hex()) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquiring", "hash", chunk.Key.Hex()) + log.Error("DbStore.LOCK acquiring", "hash", chunk.Key.Hex()) s.lock.Lock() - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquired", "hash", chunk.Key.Hex()) + log.Error("DbStore.LOCK acquired", "hash", chunk.Key.Hex()) defer s.lock.Unlock() idata, err := s.db.Get(ikey) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err) + log.Error("DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err) if err != nil { s.doPut(chunk, ikey, &index, po) + batchC := s.batchC go func() { defer func() { if err := recover(); err != nil { - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err) + log.Error("DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err) } }() + + <-batchC + close(chunk.dbStored) }() - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) + log.Error("DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) } else { log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access")) decodeIndex(idata, &index) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put already found", "hash", chunk.Key.Hex()) - } - batchC := s.batchC - go func() { - <-batchC close(chunk.dbStored) - }() + log.Error("DbStore.Put already found", "hash", chunk.Key.Hex()) + } index.Access = s.accessCnt s.accessCnt++ idata = encodeIndex(&index) @@ -594,7 +593,7 @@ func (s *DbStore) Put(chunk *Chunk) { case s.batchesC <- struct{}{}: default: } - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err) + log.Error("DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err) } // force putting into db, does not check access index @@ -716,9 +715,9 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) { hash := hasher.Sum(nil) if !bytes.Equal(hash, key) { - log.Trace(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:])) + log.Error(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:])) s.delete(indx.Idx, getIndexKey(key), s.po(key)) - log.Warn("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'") + log.Error("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'") } } @@ -784,15 +783,18 @@ func (s *DbStore) Close() { // initialises a sync iterator from a syncToken (passed in with the handshake) func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, uint64) bool) error { - s.lock.Lock() - defer s.lock.Unlock() + // probably, the lock is not needed + // s.lock.Lock() + // defer s.lock.Unlock() + untilkey := getDataKey(until, po) it := s.db.NewIterator() seek := getDataKey(since, po) it.Seek(seek) defer it.Release() - for it.Valid() { + + for it.Next() { dbkey := it.Key() if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 { break @@ -803,9 +805,8 @@ func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) { break } - it.Next() } - return nil + return it.Error() } func databaseExists(path string) bool { diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 00dc10d17b..f7cc4092d6 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -19,9 +19,7 @@ package storage import ( "encoding/binary" "fmt" - "os" "path/filepath" - "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/swarm/storage/mock" @@ -98,9 +96,9 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error) func (self *LocalStore) Put(chunk *Chunk) { chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) self.memStore.Put(chunk) - fmt.Fprintln(os.Stderr, time.Now(), "put to memstore", "hash", chunk.Key.Hex()) + log.Error("put to memstore", "hash", chunk.Key.Hex()) self.DbStore.Put(chunk) - fmt.Fprintln(os.Stderr, time.Now(), "put to dbstore", "hash", chunk.Key.Hex()) + log.Error("put to dbstore", "hash", chunk.Key.Hex()) } // Get(chunk *Chunk) looks up a chunk in the local stores From 288a5b09c9f0dbc511e39d4482c8270978198f57 Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 25 Jan 2018 01:58:33 +0100 Subject: [PATCH 08/15] swarm/network/stream, swarm/storage: simplify testing code, add more debug - add Close to server - fixes closed leveldb issue - Trigger func simplified - ClientCall simplified - syncer simulation check now call on each node - syncer simulation move defer cancel context before teardown - added timeout and logging process deliveries - improve debug log and comments --- swarm/network/stream/delivery.go | 63 +++++++++---- swarm/network/stream/delivery_test.go | 76 +++++++-------- swarm/network/stream/messages.go | 2 - swarm/network/stream/peer.go | 10 ++ swarm/network/stream/stream.go | 8 +- swarm/network/stream/streamer_test.go | 3 + swarm/network/stream/syncer.go | 14 ++- swarm/network/stream/syncer_test.go | 117 +++++++++++++----------- swarm/network/stream/testing/testing.go | 32 +++---- swarm/storage/dbstore.go | 15 +-- 10 files changed, 197 insertions(+), 143 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index d397cd7b62..5a5263f6a0 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -60,6 +60,7 @@ type SwarmChunkServer struct { batchC chan []byte db *storage.DBAPI currentLen uint64 + quit chan struct{} } // NewSwarmChunkServer is SwarmChunkServer constructor @@ -68,6 +69,7 @@ func NewSwarmChunkServer(db *storage.DBAPI) *SwarmChunkServer { deliveryC: make(chan []byte, deliveryCap), batchC: make(chan []byte), db: db, + quit: make(chan struct{}), } go s.processDeliveries() return s @@ -79,6 +81,8 @@ func (s *SwarmChunkServer) processDeliveries() { var batchC chan []byte for { select { + case <-s.quit: + return case hash := <-s.deliveryC: hashes = append(hashes, hash...) batchC = s.batchC @@ -98,6 +102,11 @@ func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64 return } +// Close needs to be called on a stream server +func (s *SwarmChunkServer) Close() { + close(s.quit) +} + // GetData retrives chunk data from db store func (s *SwarmChunkServer) GetData(key []byte) ([]byte, error) { chunk, err := s.db.Get(storage.Key(key)) @@ -168,30 +177,40 @@ type ChunkDeliveryMsg struct { func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error { d.counterIn++ + log.Error("push to receiveC", "hash", storage.Key(req.Key).Hex()) d.receiveC <- req return nil } func (d *Delivery) processReceivedChunks() { -R: + done := make(chan struct{}) + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + // R: for req := range d.receiveC { - // this should be has locally - chunk, err := d.db.Get(req.Key) - log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) - if err == nil { - log.Error("found existing?", "hash", chunk.Key.Hex()) - continue R - } - if err != storage.ErrFetching { - panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) - } - select { - case <-chunk.ReqC: - log.Error("someone else delivered?", "hash", chunk.Key.Hex()) - continue R - default: - } - go func() { + log.Error("pop from receiveC", "hash", storage.Key(req.Key).Hex()) + timer.Reset(1 * time.Second) + go func(req *ChunkDeliveryMsg) { + defer func() { done <- struct{}{} }() + // this should be has locally + chunk, err := d.db.Get(req.Key) + log.Error("after db.Get", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) + if err == nil { + log.Error("found existing?", "hash", chunk.Key.Hex()) + // continue R + return + } + if err != storage.ErrFetching { + panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) + } + select { + case <-chunk.ReqC: + log.Error("someone else delivered?", "hash", chunk.Key.Hex()) + // continue R + return + default: + } + // go func() { chunk.SData = req.SData log.Error("received delivery", "hash", chunk.Key.Hex()) d.db.Put(chunk) @@ -201,7 +220,13 @@ R: //log.Warn("received delivery stored", "hash", chunk.Key) log.Error("requesters notified", "hash", chunk.Key.Hex()) d.counterDone++ - }() + // }() + }(req) + select { + case <-timer.C: + log.Error("!!!unable to process", "hash", req.Key.Hex()) + case <-done: + } } } diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 24b35ae122..05bac692e0 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -378,22 +378,23 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck // 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 - var j int - err := sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) + for j := 0; j < nodes-1; j++ { + id := sim.IDs[j] + err := sim.CallClient(id, func(client *rpc.Client) error { + err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + j++ + sid := sim.IDs[j] + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) + }) if err != nil { return err } - ctx, cancel := context.WithTimeout(ctx, 1*time.Second) - defer cancel() - j++ - sid := sim.IDs[j] - return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) - }, sim.IDs[0:nodes-1]...) - if err != nil { - return err } - // create a retriever dpa for the pivot node delivery := deliveries[sim.IDs[0]] retrieveFunc := func(chunk *storage.Chunk) error { @@ -426,22 +427,21 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck default: } var total int64 - err := sim.CallClient(func(client *rpc.Client) error { + err := sim.CallClient(id, func(client *rpc.Client) error { ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() return client.CallContext(ctx, &total, "stream_readAll", common.BytesToHash(fileHash)) - }, id) + }) log.Info(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 } - close(quitC) return true, nil } conf.Step = &simulations.Step{ Action: action, - Trigger: streamTesting.PivotTrigger(10*time.Millisecond, checkC, sim.IDs[0]), + Trigger: streamTesting.Trigger(10*time.Millisecond, quitC, sim.IDs[0]), // we are only testing the pivot node (net.Nodes[0]) Expect: &simulations.Expectation{ Nodes: sim.IDs[0:1], @@ -490,7 +490,12 @@ func BenchmarkDeliveryFromNodesWithCheck(b *testing.B) { } func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skipCheck bool) { + defaultSkipCheck = skipCheck toAddr = network.NewAddrFromNodeID + timeout := 300 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + conf := &streamTesting.RunConfig{ Adapter: *adapter, NodeCount: nodes, @@ -498,12 +503,12 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip ToAddr: toAddr, Services: services, } - defaultSkipCheck = skipCheck sim, teardown, err := streamTesting.NewSimulation(conf) defer teardown() if err != nil { b.Fatal(err.Error()) } + stores = make(map[discover.NodeID]storage.ChunkStore) deliveries = make(map[discover.NodeID]*Delivery) for i, id := range sim.IDs { @@ -515,17 +520,19 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip } return 2 } + // wait channel for all nodes all peer connections to set up + waitPeerErrC = make(chan error) + // create a dpa for the last node in the chain which we are gonna write to remoteDpa := storage.NewDPA(sim.Stores[nodes-1], storage.NewChunkerParams()) remoteDpa.Start() defer remoteDpa.Stop() - // wait channel for all nodes all peer connections to set up - waitPeerErrC = make(chan error) // channel to signal simulation initialisation with action call complete // or node disconnections simErrC := make(chan error) quitC := make(chan struct{}) + defer close(quitC) action := func(ctx context.Context) error { // each node Subscribes to each other's swarmChunkServerStreamName @@ -545,18 +552,19 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // each node except the last one subscribes to the upstream swarm chunk server stream // which responds to chunk retrieve requests - var j int - simErrC <- sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), simErrC, quitC) - if err != nil { - return err - } - ctx, cancel := context.WithTimeout(ctx, 1*time.Second) - defer cancel() - j++ - sid := sim.IDs[j] // the upstream peer's id - return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) - }, sim.IDs[0:nodes-1]...) + for j := 0; j < nodes-1; j++ { + id := sim.IDs[j] + simErrC <- sim.CallClient(id, func(client *rpc.Client) error { + err := streamTesting.WatchDisconnections(id, client, peerCount(id), simErrC, quitC) + if err != nil { + return err + } + 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) + }) + } // signal to the benchmark that setup is complete return err } @@ -589,10 +597,6 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // run the simulation in the background errc := make(chan error) go func() { - timeout := 300 * time.Second - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - _, err := sim.Run(ctx, conf) errc <- err }() @@ -608,6 +612,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip select { case err = <-simErrC: case <-quitC: + return } trigger <- sim.IDs[0] checkC <- err @@ -669,7 +674,6 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip } } // benchmark over, trigger the check function to conclude the simulation - close(quitC) err = <-errc if err != nil { b.Fatalf("expected no error. got %v", err) diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index a11c97332d..aa8f5f75c4 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -179,8 +179,6 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { p.Drop(fmt.Errorf("handleOfferedHashesMsg next: %v", err)) return } - case <-s.quit: - 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) diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index ed2366f5c6..cd7cdc2c9a 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -83,6 +83,10 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { if err != nil { return err } + // true only when quiting + if len(hashes) == 0 { + return nil + } if proof == nil { proof = &HandoverProof{ Handover: &Handover{}, @@ -171,3 +175,9 @@ func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bo next <- nil // this is to allow wantedKeysMsg before first batch arrives 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 ac6d027d2d..30cd71548f 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -198,7 +198,8 @@ func (r *Registry) run(p *protocols.Peer) error { sp := NewPeer(p, r) r.setPeer(sp) defer r.deletePeer(sp) - defer close(sp.quit) + // defer close(sp.quit + defer sp.close() return sp.Run(sp.HandleMsg) } @@ -257,6 +258,7 @@ type server struct { type Server interface { SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) GetData([]byte) ([]byte, error) + Close() } type client struct { @@ -266,8 +268,8 @@ type client struct { live bool stream string key []byte - quit chan struct{} - next chan error + // quit chan struct{} + next chan error } // Client interface for incoming peer Streamer diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index fec23a06d3..71c0b4bda9 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -85,6 +85,9 @@ func (self *testServer) GetData([]byte) ([]byte, error) { return nil, nil } +func (self *testServer) Close() { +} + func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(t) defer teardown() diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index af1bbb0b2e..9620a625da 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -42,6 +42,7 @@ type SwarmSyncerServer struct { db *storage.DBAPI sessionAt uint64 start uint64 + quit chan struct{} } // NewSwarmSyncerServer is contructor for SwarmSyncerServer @@ -56,6 +57,7 @@ func NewSwarmSyncerServer(live bool, po uint8, db *storage.DBAPI) (*SwarmSyncerS db: db, sessionAt: sessionAt, start: start, + quit: make(chan struct{}), }, nil } @@ -72,6 +74,11 @@ func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) { // }) } +// Close needs to be called on a stream server +func (s *SwarmSyncerServer) Close() { + close(s.quit) +} + // GetSection retrieves the actual chunk from localstore func (s *SwarmSyncerServer) GetData(key []byte) ([]byte, error) { chunk, err := s.db.Get(storage.Key(key)) @@ -95,7 +102,12 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 } ticker := time.NewTicker(10 * time.Millisecond) defer ticker.Stop() - for range ticker.C { + for { + select { + case <-ticker.C: + case <-s.quit: + return nil, 0, 0, nil, nil + } err := s.db.Iterator(from, to, s.po, func(key storage.Key, idx uint64) bool { batch = append(batch, key[:]...) i++ diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 263d230183..1bcd5fe1aa 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -44,6 +44,7 @@ func TestSyncerSimulation(t *testing.T) { // testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1) // // testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1) testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 1) + // testSyncBetweenNodes(t, 32, 1, dataChunkCount, true, 1) // testSyncBetweenNodes(t, 16, 1, dataChunkCount, false, 1) } @@ -61,34 +62,49 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck ToAddr: toAddr, Services: services, } + // create context for simulation run + timeout := 30 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + // defer cancel should come before defer simulation teardown + defer cancel() + // create simulation network with the config sim, teardown, err := streamTesting.NewSimulation(conf) defer teardown() if err != nil { t.Fatal(err.Error()) } + // DEBUG: defer func() { for _, id := range sim.IDs { deliveries[id].PrintCounters(id) } - // for id, delivery := range deliveries { - // delivery.PrintCounters(id) - // } }() + // HACK: these are global variables in the test so that they are available for + // the service constructor function + // TODO: will this work with exec/docker adapter? + // localstore of nodes made available for action and check calls stores = make(map[discover.NodeID]storage.ChunkStore) - deliveries = make(map[discover.NodeID]*Delivery) + nodeIndex := make(map[discover.NodeID]int) for i, id := range sim.IDs { + nodeIndex[id] = i stores[id] = sim.Stores[i] } + deliveries = make(map[discover.NodeID]*Delivery) + // peerCount function gives the number of peer connections for a nodeID + // this is needed for the service run function to wait until + // each protocol instance runs and the streamer peers are available peerCount = func(id discover.NodeID) int { if sim.IDs[0] == id || sim.IDs[nodes-1] == id { return 1 } return 2 } - // here we distribute chunks of a random file into Stores of nodes 1 to nodes + waitPeerErrC = make(chan error) + + // here we distribute chunks of a random file into stores 1...nodes rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams()) rrdpa.Start() size := chunkCount * chunkSize @@ -100,12 +116,14 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck t.Fatal(err.Error()) } - // collect hashes in po 1 from all nodes - hashes := make([][]storage.Key, nodes) + // create DBAPI-s for all nodes dbs := make([]*storage.DBAPI, nodes) for i := 0; i < nodes; i++ { dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore)) } + + // collect hashes in po 1 bin for each node + hashes := make([][]storage.Key, nodes) totalHashes := 0 hashCounts := make([]int, nodes) for i := nodes - 1; i >= 0; i-- { @@ -120,9 +138,12 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck }) } + // errc is error channel for simulation errc := make(chan error, 1) - waitPeerErrC = make(chan error) quitC := make(chan struct{}) + defer close(quitC) + + // action is subscribe action := func(ctx context.Context) error { // need to wait till an aynchronous process registers the peers in streamer.peers // that is used by Subscribe @@ -139,24 +160,29 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } } // each node Subscribes to each other's swarmChunkServerStreamName - j := 0 - return sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) + for j := 0; j < nodes-1; j++ { + id := sim.IDs[j] + err := sim.CallClient(id, func(client *rpc.Client) error { + // report disconnect events to the error channel cos peers should not disconnect + err := streamTesting.WatchDisconnections(id, client, peerCount(id), errc, quitC) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + 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) + }) if err != nil { return err } - ctx, cancel := context.WithTimeout(ctx, 1*time.Second) - defer cancel() - j++ - return client.CallContext(ctx, nil, "stream_subscribeStream", sim.IDs[j], "SYNC", []byte{1}, 0, 0, Top, false) - }, sim.IDs[0:nodes-1]...) + } + return nil } // this makes sure check is not called before the previous call finishes - checkC := make(chan struct{}) check := func(ctx context.Context, id discover.NodeID) (bool, error) { - defer func() { checkC <- struct{}{} }() - select { case err := <-errc: return false, err @@ -165,54 +191,37 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck default: } - var pass bool - var i int - log.Error("staring dbs check") - for i = nodes - 1; i >= 0; i-- { - nodeHashCount := hashCounts[i] - nodeHashFound := 0 - for j := i; j < nodes; j++ { - nodeHashes := hashes[j] - for _, key := range nodeHashes { - chunk, err := dbs[i].Get(key) - if err == storage.ErrFetching { - <-chunk.ReqC - nodeHashFound++ - } else if err == nil { - nodeHashFound++ - } else { - log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) - } + log.Error("starting dbs check", "node", id) + i := nodeIndex[id] + var total, found int + for j := i; j < nodes; j++ { + total += len(hashes[j]) + for _, key := range hashes[j] { + chunk, err := dbs[i].Get(key) + if err == storage.ErrFetching { + <-chunk.ReqC + } else if err != nil { + log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) + continue } - } - log.Error("sync check", "node", sim.IDs[i], "index", i, "bin", po, "found", nodeHashFound, "total", nodeHashCount) - pass = nodeHashFound == nodeHashCount - if !pass { - break + // needed for leveldb not to be closed? + // chunk.WaitToStore() + found++ } } - // log.Error("sync check", "bin", po, "found", found, "total", totalHashes) - // pass := found == totalHashes - if !pass { - return false, nil - } - close(quitC) - return true, nil - + log.Error("sync check", "node", id, "index", i, "bin", po, "found", found, "total", total) + return total == found, nil } conf.Step = &simulations.Step{ Action: action, - Trigger: streamTesting.PivotTrigger(500*time.Millisecond, checkC, sim.IDs[0]), + Trigger: streamTesting.Trigger(500*time.Millisecond, quitC, sim.IDs[0:nodes-1]...), Expect: &simulations.Expectation{ Nodes: sim.IDs[0:1], Check: check, }, } startedAt := time.Now() - timeout := 30 * time.Second - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() result, err := sim.Run(ctx, conf) finishedAt := time.Now() if err != nil { diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index 9585de369b..8ef750df1b 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -234,7 +234,7 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, expectedConnCou return nil } -func PivotTrigger(d time.Duration, checkC chan struct{}, ids ...discover.NodeID) chan discover.NodeID { +func Trigger(d time.Duration, quitC chan struct{}, ids ...discover.NodeID) chan discover.NodeID { trigger := make(chan discover.NodeID) go func() { ticker := time.NewTicker(d) @@ -242,28 +242,24 @@ func PivotTrigger(d time.Duration, checkC chan struct{}, ids ...discover.NodeID) // we are only testing the pivot node (net.Nodes[0]) for range ticker.C { for _, id := range ids { - trigger <- id + select { + case trigger <- id: + case <-quitC: + } } - <-checkC } }() return trigger } -func (sim *Simulation) CallClient(f func(*rpc.Client) error, ids ...discover.NodeID) error { - for _, id := range ids { - node := sim.Net.GetNode(id) - if node == nil { - return fmt.Errorf("unknown node: %s", id) - } - client, err := node.Client() - if err != nil { - return fmt.Errorf("error getting node client: %s", err) - } - err = f(client) - if err != nil { - return err - } +func (sim *Simulation) CallClient(id discover.NodeID, f func(*rpc.Client) error) error { + node := sim.Net.GetNode(id) + if node == nil { + return fmt.Errorf("unknown node: %s", id) } - return nil + client, err := node.Client() + if err != nil { + return fmt.Errorf("error getting node client: %s", err) + } + return f(client) } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 41b0b67a24..74890b5ddf 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -781,27 +781,22 @@ func (s *DbStore) Close() { s.db.Close() } -// initialises a sync iterator from a syncToken (passed in with the handshake) +// SyncIterator(start, stop, po, f) calls f on each hash of a bin po from start to stop func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, uint64) bool) error { - // probably, the lock is not needed - // s.lock.Lock() - // defer s.lock.Unlock() - + sincekey := getDataKey(since, po) untilkey := getDataKey(until, po) - it := s.db.NewIterator() - seek := getDataKey(since, po) - it.Seek(seek) defer it.Release() + it.Seek(sincekey) for it.Next() { dbkey := it.Key() if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 { break } - key := make([]byte, 32) - copy(key, it.Value()[:32]) + val := it.Value() + copy(key, val[:32]) if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) { break } From 6bf67decd242135bbd7b8ab79650a464a058cfc8 Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 25 Jan 2018 12:18:39 +0100 Subject: [PATCH 09/15] debug --- swarm/network/stream/delivery.go | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 5a5263f6a0..cb3392550e 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -173,10 +173,12 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e type ChunkDeliveryMsg struct { Key storage.Key SData []byte // the stored chunk Data (incl size) + peer *Peer } func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error { d.counterIn++ + req.peer = sp log.Error("push to receiveC", "hash", storage.Key(req.Key).Hex()) d.receiveC <- req return nil @@ -188,15 +190,16 @@ func (d *Delivery) processReceivedChunks() { defer timer.Stop() // R: for req := range d.receiveC { - log.Error("pop from receiveC", "hash", storage.Key(req.Key).Hex()) + log.Error("pop from receiveC", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex()) timer.Reset(1 * time.Second) go func(req *ChunkDeliveryMsg) { defer func() { done <- struct{}{} }() // this should be has locally + log.Error("before db.Get", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex()) chunk, err := d.db.Get(req.Key) - log.Error("after db.Get", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) + log.Error("after db.Get", "peer", req.peer.ID(), "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) if err == nil { - log.Error("found existing?", "hash", chunk.Key.Hex()) + log.Error("found existing?", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) // continue R return } @@ -212,20 +215,21 @@ func (d *Delivery) processReceivedChunks() { } // go func() { chunk.SData = req.SData - log.Error("received delivery", "hash", chunk.Key.Hex()) + log.Error("received delivery", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) d.db.Put(chunk) - log.Error("put to db", "hash", chunk.Key.Hex()) + log.Error("put to db", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) chunk.WaitToStore() close(chunk.ReqC) //log.Warn("received delivery stored", "hash", chunk.Key) - log.Error("requesters notified", "hash", chunk.Key.Hex()) + log.Error("requesters notified", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) d.counterDone++ // }() }(req) select { case <-timer.C: - log.Error("!!!unable to process", "hash", req.Key.Hex()) + log.Error("!!!unable to process delivery", "peer", req.peer.ID(), "hash", req.Key.Hex()) case <-done: + log.Error("done processing delivery", "peer", req.peer.ID(), "hash", req.Key.Hex()) } } } From f43390f5af74b30fa07dd10611de866812e1de30 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Thu, 25 Jan 2018 14:47:16 +0100 Subject: [PATCH 10/15] A workaround for LocalStore put corrupting the chunk key --- swarm/network/stream/delivery.go | 4 ++++ swarm/storage/dbstore.go | 23 +++++++++++++++++++---- swarm/storage/localstore.go | 19 ++++++++++++++----- swarm/storage/memstore.go | 5 +++++ 4 files changed, 42 insertions(+), 9 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index cb3392550e..de05cd5293 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -17,6 +17,7 @@ package stream import ( + "bytes" "errors" "fmt" "time" @@ -197,6 +198,9 @@ func (d *Delivery) processReceivedChunks() { // this should be has locally log.Error("before db.Get", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex()) chunk, err := d.db.Get(req.Key) + if !bytes.Equal(chunk.Key, req.Key) { + panic(fmt.Errorf("processReceivedChunks: chunk key %s != req key %s (peer %s)", chunk.Key.Hex(), storage.Key(req.Key).Hex(), req.peer.ID())) + } log.Error("after db.Get", "peer", req.peer.ID(), "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) if err == nil { log.Error("found existing?", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 74890b5ddf..fecb199e1b 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -27,6 +27,7 @@ import ( "bytes" "encoding/binary" "encoding/hex" + "encoding/json" "fmt" "io" "io/ioutil" @@ -34,7 +35,6 @@ import ( "time" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/swarm/storage/mock" "github.com/syndtr/goleveldb/leveldb" "github.com/syndtr/goleveldb/leveldb/opt" @@ -221,7 +221,12 @@ func getDataKey(idx uint64, po uint8) []byte { } func encodeIndex(index *dpaDBIndex) []byte { - data, _ := rlp.EncodeToBytes(index) + //data, _ := rlp.EncodeToBytes(index) + + data, err := json.Marshal(index) + if err != nil { + panic(err) + } return data } @@ -230,8 +235,10 @@ func encodeData(chunk *Chunk) []byte { } func decodeIndex(data []byte, index *dpaDBIndex) error { - dec := rlp.NewStream(bytes.NewReader(data), 0) - return dec.Decode(index) + // dec := rlp.NewStream(bytes.NewReader(data), 0) + // return dec.Decode(index) + return json.Unmarshal(data, index) + } func decodeData(data []byte, chunk *Chunk) { @@ -542,6 +549,7 @@ func (s *DbStore) Put(chunk *Chunk) { log.Error("DbStore.Put", "hash", chunk.Key.Hex()) done := make(chan struct{}) defer close(done) + key := Key(append(make([]byte, 0), chunk.Key...)) go func() { log.Error("DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex()) select { @@ -549,6 +557,9 @@ func (s *DbStore) Put(chunk *Chunk) { log.Error("DbStore.Put WAITING", "hash", chunk.Key.Hex()) case <-done: log.Error("DbStore.Put EXITED", "hash", chunk.Key.Hex()) + if !bytes.Equal(chunk.Key, key) { + panic(fmt.Errorf("DbStore.Get: chunk key %s != req key %s", chunk.Key.Hex(), key.Hex())) + } } }() @@ -724,6 +735,10 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) { chunk = NewChunk(key, nil) decodeData(data, chunk) + if !bytes.Equal(chunk.Key, key) { + panic(fmt.Errorf("DbStore.Get: chunk key %s != req key %s", chunk.Key.Hex(), key.Hex())) + } + } else { err = ErrNotFound } diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index f7cc4092d6..fe6646402a 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -17,6 +17,7 @@ package storage import ( + "bytes" "encoding/binary" "fmt" "path/filepath" @@ -95,10 +96,18 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error) // unsafe, in that the data is not integrity checked func (self *LocalStore) Put(chunk *Chunk) { chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - self.memStore.Put(chunk) - log.Error("put to memstore", "hash", chunk.Key.Hex()) - self.DbStore.Put(chunk) - log.Error("put to dbstore", "hash", chunk.Key.Hex()) + c := &Chunk{ + Key: Key(append([]byte{}, chunk.Key...)), + SData: append([]byte{}, chunk.SData...), + dbStored: chunk.dbStored, + } + self.memStore.Put(c) + log.Error("put to memstore", "hash", c.Key.Hex()) + self.DbStore.Put(c) + log.Error("put to dbstore", "hash", c.Key.Hex()) + if !bytes.Equal(chunk.Key, c.Key) { + panic(fmt.Errorf("LocalStore.Put: chunk %s != c %s", chunk.Key.Hex(), c.Key.Hex())) + } } // Get(chunk *Chunk) looks up a chunk in the local stores @@ -122,7 +131,7 @@ func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { return } chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - self.memStore.Put(chunk) + //self.memStore.Put(chunk) return } diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go index 31e2baf454..e96e225a4d 100644 --- a/swarm/storage/memstore.go +++ b/swarm/storage/memstore.go @@ -19,6 +19,8 @@ package storage import ( + "bytes" + "fmt" "sync" ) @@ -329,6 +331,9 @@ func (m *MemStore) Get(key Key) (*Chunk, error) { if !ok { return nil, ErrNotFound } + if !bytes.Equal(c.Key, key) { + panic(fmt.Errorf("MemStore.Get: chunk key %s != req key %s", c.Key.Hex(), key.Hex())) + } return c, nil } From b5a174a9e76bc27d42e559a17602dbbbc93747a6 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Thu, 25 Jan 2018 18:06:20 +0100 Subject: [PATCH 11/15] Fix LazyChunkReader slice bounds out of range error --- swarm/storage/localstore.go | 1 + 1 file changed, 1 insertion(+) diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index fe6646402a..40103503be 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -99,6 +99,7 @@ func (self *LocalStore) Put(chunk *Chunk) { c := &Chunk{ Key: Key(append([]byte{}, chunk.Key...)), SData: append([]byte{}, chunk.SData...), + Size: chunk.Size, dbStored: chunk.dbStored, } self.memStore.Put(c) From 0fb5df8ac2bfb3967245e9768d3e0f416ba11f15 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Fri, 26 Jan 2018 10:04:38 +0100 Subject: [PATCH 12/15] Use the old MemStore implementation --- swarm/storage/memstore.go | 603 +++++++++++++++++++------------------- 1 file changed, 302 insertions(+), 301 deletions(-) diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go index e96e225a4d..822eb28b4c 100644 --- a/swarm/storage/memstore.go +++ b/swarm/storage/memstore.go @@ -19,9 +19,10 @@ package storage import ( - "bytes" "fmt" "sync" + + "github.com/ethereum/go-ethereum/log" ) const ( @@ -31,321 +32,321 @@ const ( defaultCacheCapacity = 5000 ) -// type MemStore struct { -// memtree *memTree -// entryCnt, capacity uint // stored entries -// accessCnt uint64 // access counter; oldest is thrown away when full -// dbAccessCnt uint64 -// dbStore *DbStore -// lock sync.Mutex -// } -// -// /* -// a hash prefix subtree containing subtrees or one storage entry (but never both) -// -// - access[0] stores the smallest (oldest) access count value in this subtree -// - if it contains more subtrees and its subtree count is at least 4, access[1:2] -// stores the smallest access count in the first and second halves of subtrees -// (so that access[0] = min(access[1], access[2]) -// - likewise, if subtree count is at least 8, -// access[1] = min(access[3], access[4]) -// access[2] = min(access[5], access[6]) -// (access[] is a binary tree inside the multi-bit leveled hash tree) -// */ -// -// func NewMemStore(d *DbStore, capacity uint) (m *MemStore) { -// m = &MemStore{} -// m.memtree = newMemTree(memTreeFLW, nil, 0) -// m.dbStore = d -// m.setCapacity(capacity) -// return -// } -// -// type memTree struct { -// subtree []*memTree -// parent *memTree -// parentIdx uint -// -// bits uint // log2(subtree count) -// width uint // subtree count -// -// entry *Chunk // if subtrees are present, entry should be nil -// lastDBaccess uint64 -// access []uint64 -// } -// -// func newMemTree(b uint, parent *memTree, pidx uint) (node *memTree) { -// node = new(memTree) -// node.bits = b -// node.width = 1 << b -// node.subtree = make([]*memTree, node.width) -// node.access = make([]uint64, node.width-1) -// node.parent = parent -// node.parentIdx = pidx -// if parent != nil { -// parent.subtree[pidx] = node -// } -// -// return node -// } -// -// func (node *memTree) updateAccess(a uint64) { -// aidx := uint(0) -// var aa uint64 -// oa := node.access[0] -// for node.access[aidx] == oa { -// node.access[aidx] = a -// if aidx > 0 { -// aa = node.access[((aidx-1)^1)+1] -// aidx = (aidx - 1) >> 1 -// } else { -// pidx := node.parentIdx -// node = node.parent -// if node == nil { -// return -// } -// nn := node.subtree[pidx^1] -// if nn != nil { -// aa = nn.access[0] -// } else { -// aa = 0 -// } -// aidx = (node.width + pidx - 2) >> 1 -// } -// -// if (aa != 0) && (aa < a) { -// a = aa -// } -// } -// } -// -// func (s *MemStore) setCapacity(c uint) { -// s.lock.Lock() -// defer s.lock.Unlock() -// -// for c < s.entryCnt { -// s.removeOldest() -// } -// s.capacity = c -// } -// -// // entry (not its copy) is going to be in MemStore -// func (s *MemStore) Put(entry *Chunk) { -// if s.capacity == 0 { -// return -// } -// -// s.lock.Lock() -// defer s.lock.Unlock() -// -// if s.entryCnt >= s.capacity { -// s.removeOldest() -// } -// -// s.accessCnt++ -// -// node := s.memtree -// bitpos := uint(0) -// for node.entry == nil { -// l := entry.Key.bits(bitpos, node.bits) -// st := node.subtree[l] -// if st == nil { -// st = newMemTree(memTreeLW, node, l) -// bitpos += node.bits -// node = st -// break -// } -// bitpos += node.bits -// node = st -// } -// -// if node.entry != nil { -// -// if node.entry.Key.isEqual(entry.Key) { -// node.updateAccess(s.accessCnt) -// if entry.SData == nil { -// entry.Size = node.entry.Size -// entry.SData = node.entry.SData -// } -// if entry.ReqC == nil { -// entry.ReqC = node.entry.ReqC -// } -// entry.C = node.entry.C -// node.entry = entry -// return -// } -// -// for node.entry != nil { -// -// l := node.entry.Key.bits(bitpos, node.bits) -// st := node.subtree[l] -// if st == nil { -// st = newMemTree(memTreeLW, node, l) -// } -// st.entry = node.entry -// node.entry = nil -// st.updateAccess(node.access[0]) -// -// l = entry.Key.bits(bitpos, node.bits) -// st = node.subtree[l] -// if st == nil { -// st = newMemTree(memTreeLW, node, l) -// } -// bitpos += node.bits -// node = st -// -// } -// } -// -// node.entry = entry -// node.lastDBaccess = s.dbAccessCnt -// node.updateAccess(s.accessCnt) -// s.entryCnt++ -// } -// -// func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) { -// s.lock.Lock() -// defer s.lock.Unlock() -// -// node := s.memtree -// bitpos := uint(0) -// for node.entry == nil { -// l := hash.bits(bitpos, node.bits) -// st := node.subtree[l] -// if st == nil { -// return nil, ErrNotFound -// } -// bitpos += node.bits -// node = st -// } -// -// if node.entry.Key.isEqual(hash) { -// s.accessCnt++ -// node.updateAccess(s.accessCnt) -// chunk = node.entry -// if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt { -// s.dbAccessCnt++ -// node.lastDBaccess = s.dbAccessCnt -// if s.dbStore != nil { -// s.dbStore.updateAccessCnt(hash) -// } -// } -// } else { -// err = ErrNotFound -// } -// -// return -// } -// -// func (s *MemStore) removeOldest() { -// node := s.memtree -// log.Warn("purge memstore") -// for node.entry == nil { -// -// aidx := uint(0) -// av := node.access[aidx] -// -// for aidx < node.width/2-1 { -// if av == node.access[aidx*2+1] { -// node.access[aidx] = node.access[aidx*2+2] -// aidx = aidx*2 + 1 -// } else if av == node.access[aidx*2+2] { -// node.access[aidx] = node.access[aidx*2+1] -// aidx = aidx*2 + 2 -// } else { -// panic(nil) -// } -// } -// pidx := aidx*2 + 2 - node.width -// if (node.subtree[pidx] != nil) && (av == node.subtree[pidx].access[0]) { -// if node.subtree[pidx+1] != nil { -// node.access[aidx] = node.subtree[pidx+1].access[0] -// } else { -// node.access[aidx] = 0 -// } -// } else if (node.subtree[pidx+1] != nil) && (av == node.subtree[pidx+1].access[0]) { -// if node.subtree[pidx] != nil { -// node.access[aidx] = node.subtree[pidx].access[0] -// } else { -// node.access[aidx] = 0 -// } -// pidx++ -// } else { -// panic(nil) -// } -// -// //fmt.Println(pidx) -// node = node.subtree[pidx] -// -// } -// -// log.Trace(fmt.Sprintf("Memstore Clean: Waiting for chunk %v to be saved", node.entry.Key.Log())) -// <-node.entry.dbStored -// log.Trace(fmt.Sprintf("Memstore Clean: Chunk %v saved to DBStore. Ready to clear from mem.", node.entry.Key.Log())) -// -// if node.entry.ReqC == nil { -// node.entry = nil -// s.entryCnt-- -// } else { -// return -// } -// -// node.access[0] = 0 -// -// //--- -// -// aidx := uint(0) -// for { -// aa := node.access[aidx] -// if aidx > 0 { -// aidx = (aidx - 1) >> 1 -// } else { -// pidx := node.parentIdx -// node = node.parent -// if node == nil { -// return -// } -// aidx = (node.width + pidx - 2) >> 1 -// } -// if (aa != 0) && ((aa < node.access[aidx]) || (node.access[aidx] == 0)) { -// node.access[aidx] = aa -// } -// } -// } - type MemStore struct { - m map[string]*Chunk - mu sync.RWMutex + memtree *memTree + entryCnt, capacity uint // stored entries + accessCnt uint64 // access counter; oldest is thrown away when full + dbAccessCnt uint64 + dbStore *DbStore + lock sync.Mutex } +/* +a hash prefix subtree containing subtrees or one storage entry (but never both) + +- access[0] stores the smallest (oldest) access count value in this subtree +- if it contains more subtrees and its subtree count is at least 4, access[1:2] + stores the smallest access count in the first and second halves of subtrees + (so that access[0] = min(access[1], access[2]) +- likewise, if subtree count is at least 8, + access[1] = min(access[3], access[4]) + access[2] = min(access[5], access[6]) + (access[] is a binary tree inside the multi-bit leveled hash tree) +*/ + func NewMemStore(d *DbStore, capacity uint) (m *MemStore) { - return &MemStore{ - m: make(map[string]*Chunk), + m = &MemStore{} + m.memtree = newMemTree(memTreeFLW, nil, 0) + m.dbStore = d + m.setCapacity(capacity) + return +} + +type memTree struct { + subtree []*memTree + parent *memTree + parentIdx uint + + bits uint // log2(subtree count) + width uint // subtree count + + entry *Chunk // if subtrees are present, entry should be nil + lastDBaccess uint64 + access []uint64 +} + +func newMemTree(b uint, parent *memTree, pidx uint) (node *memTree) { + node = new(memTree) + node.bits = b + node.width = 1 << b + node.subtree = make([]*memTree, node.width) + node.access = make([]uint64, node.width-1) + node.parent = parent + node.parentIdx = pidx + if parent != nil { + parent.subtree[pidx] = node + } + + return node +} + +func (node *memTree) updateAccess(a uint64) { + aidx := uint(0) + var aa uint64 + oa := node.access[0] + for node.access[aidx] == oa { + node.access[aidx] = a + if aidx > 0 { + aa = node.access[((aidx-1)^1)+1] + aidx = (aidx - 1) >> 1 + } else { + pidx := node.parentIdx + node = node.parent + if node == nil { + return + } + nn := node.subtree[pidx^1] + if nn != nil { + aa = nn.access[0] + } else { + aa = 0 + } + aidx = (node.width + pidx - 2) >> 1 + } + + if (aa != 0) && (aa < a) { + a = aa + } } } -func (m *MemStore) Get(key Key) (*Chunk, error) { - m.mu.RLock() - defer m.mu.RUnlock() - c, ok := m.m[string(key[:])] - if !ok { - return nil, ErrNotFound +func (s *MemStore) setCapacity(c uint) { + s.lock.Lock() + defer s.lock.Unlock() + + for c < s.entryCnt { + s.removeOldest() } - if !bytes.Equal(c.Key, key) { - panic(fmt.Errorf("MemStore.Get: chunk key %s != req key %s", c.Key.Hex(), key.Hex())) + s.capacity = c +} + +// entry (not its copy) is going to be in MemStore +func (s *MemStore) Put(entry *Chunk) { + if s.capacity == 0 { + return } - return c, nil + + s.lock.Lock() + defer s.lock.Unlock() + + if s.entryCnt >= s.capacity { + s.removeOldest() + } + + s.accessCnt++ + + node := s.memtree + bitpos := uint(0) + for node.entry == nil { + l := entry.Key.bits(bitpos, node.bits) + st := node.subtree[l] + if st == nil { + st = newMemTree(memTreeLW, node, l) + bitpos += node.bits + node = st + break + } + bitpos += node.bits + node = st + } + + if node.entry != nil { + + if node.entry.Key.isEqual(entry.Key) { + node.updateAccess(s.accessCnt) + if entry.SData == nil { + entry.Size = node.entry.Size + entry.SData = node.entry.SData + } + if entry.ReqC == nil { + entry.ReqC = node.entry.ReqC + } + entry.C = node.entry.C + node.entry = entry + return + } + + for node.entry != nil { + + l := node.entry.Key.bits(bitpos, node.bits) + st := node.subtree[l] + if st == nil { + st = newMemTree(memTreeLW, node, l) + } + st.entry = node.entry + node.entry = nil + st.updateAccess(node.access[0]) + + l = entry.Key.bits(bitpos, node.bits) + st = node.subtree[l] + if st == nil { + st = newMemTree(memTreeLW, node, l) + } + bitpos += node.bits + node = st + + } + } + + node.entry = entry + node.lastDBaccess = s.dbAccessCnt + node.updateAccess(s.accessCnt) + s.entryCnt++ } -func (m *MemStore) Put(c *Chunk) { - m.mu.Lock() - defer m.mu.Unlock() - m.m[string(c.Key[:])] = c +func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) { + s.lock.Lock() + defer s.lock.Unlock() + + node := s.memtree + bitpos := uint(0) + for node.entry == nil { + l := hash.bits(bitpos, node.bits) + st := node.subtree[l] + if st == nil { + return nil, ErrNotFound + } + bitpos += node.bits + node = st + } + + if node.entry.Key.isEqual(hash) { + s.accessCnt++ + node.updateAccess(s.accessCnt) + chunk = node.entry + if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt { + s.dbAccessCnt++ + node.lastDBaccess = s.dbAccessCnt + if s.dbStore != nil { + s.dbStore.updateAccessCnt(hash) + } + } + } else { + err = ErrNotFound + } + + return } -func (m *MemStore) setCapacity(n int) { +func (s *MemStore) removeOldest() { + node := s.memtree + log.Warn("purge memstore") + for node.entry == nil { + aidx := uint(0) + av := node.access[aidx] + + for aidx < node.width/2-1 { + if av == node.access[aidx*2+1] { + node.access[aidx] = node.access[aidx*2+2] + aidx = aidx*2 + 1 + } else if av == node.access[aidx*2+2] { + node.access[aidx] = node.access[aidx*2+1] + aidx = aidx*2 + 2 + } else { + panic(nil) + } + } + pidx := aidx*2 + 2 - node.width + if (node.subtree[pidx] != nil) && (av == node.subtree[pidx].access[0]) { + if node.subtree[pidx+1] != nil { + node.access[aidx] = node.subtree[pidx+1].access[0] + } else { + node.access[aidx] = 0 + } + } else if (node.subtree[pidx+1] != nil) && (av == node.subtree[pidx+1].access[0]) { + if node.subtree[pidx] != nil { + node.access[aidx] = node.subtree[pidx].access[0] + } else { + node.access[aidx] = 0 + } + pidx++ + } else { + panic(nil) + } + + //fmt.Println(pidx) + node = node.subtree[pidx] + + } + + log.Trace(fmt.Sprintf("Memstore Clean: Waiting for chunk %v to be saved", node.entry.Key.Log())) + <-node.entry.dbStored + log.Trace(fmt.Sprintf("Memstore Clean: Chunk %v saved to DBStore. Ready to clear from mem.", node.entry.Key.Log())) + + if node.entry.ReqC == nil { + node.entry = nil + s.entryCnt-- + } else { + return + } + + node.access[0] = 0 + + //--- + + aidx := uint(0) + for { + aa := node.access[aidx] + if aidx > 0 { + aidx = (aidx - 1) >> 1 + } else { + pidx := node.parentIdx + node = node.parent + if node == nil { + return + } + aidx = (node.width + pidx - 2) >> 1 + } + if (aa != 0) && ((aa < node.access[aidx]) || (node.access[aidx] == 0)) { + node.access[aidx] = aa + } + } } +// type MemStore struct { +// m map[string]*Chunk +// mu sync.RWMutex +// } + +// func NewMemStore(d *DbStore, capacity uint) (m *MemStore) { +// return &MemStore{ +// m: make(map[string]*Chunk), +// } +// } + +// func (m *MemStore) Get(key Key) (*Chunk, error) { +// m.mu.RLock() +// defer m.mu.RUnlock() +// c, ok := m.m[string(key[:])] +// if !ok { +// return nil, ErrNotFound +// } +// if !bytes.Equal(c.Key, key) { +// panic(fmt.Errorf("MemStore.Get: chunk key %s != req key %s", c.Key.Hex(), key.Hex())) +// } +// return c, nil +// } + +// func (m *MemStore) Put(c *Chunk) { +// m.mu.Lock() +// defer m.mu.Unlock() +// m.m[string(c.Key[:])] = c +// } + +// func (m *MemStore) setCapacity(n int) { + +// } + // Close memstore func (s *MemStore) Close() {} From 1f731cd01688bb66a80bf92d1d64dee1555b40ed Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Fri, 26 Jan 2018 12:19:10 +0100 Subject: [PATCH 13/15] swarm/network/stream: Fix TestDeliveryFromNodes --- swarm/network/stream/delivery_test.go | 7 ++----- swarm/network/stream/syncer_test.go | 11 +++-------- 2 files changed, 5 insertions(+), 13 deletions(-) diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 05bac692e0..e55e78aee9 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -381,14 +381,13 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck for j := 0; j < nodes-1; j++ { id := sim.IDs[j] err := sim.CallClient(id, func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) + err := streamTesting.WatchDisconnections(id, client, peerCount(id), errc, quitC) if err != nil { return err } ctx, cancel := context.WithTimeout(ctx, 1*time.Second) defer cancel() - j++ - sid := sim.IDs[j] + sid := sim.IDs[j+1] return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) }) if err != nil { @@ -416,9 +415,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck }() return nil } - checkC := make(chan struct{}) check := func(ctx context.Context, id discover.NodeID) (bool, error) { - defer func() { checkC <- struct{}{} }() select { case err := <-errc: return false, err diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 1bcd5fe1aa..72f4a53b4b 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -37,15 +37,10 @@ import ( const dataChunkCount = 1000 func TestSyncerSimulation(t *testing.T) { - // testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1) - // testSyncBetweenNodes(t, 2, 1, dataChunkCount, false, 1) - // testSyncBetweenNodes(t, 4, 1, dataChunkCount, true, 1) - // // testSyncBetweenNodes(t, 4, 1, dataChunkCount, false, 1) - // testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1) - // // testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1) + testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1) + testSyncBetweenNodes(t, 4, 1, dataChunkCount, true, 1) + testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1) testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 1) - // testSyncBetweenNodes(t, 32, 1, dataChunkCount, true, 1) - // testSyncBetweenNodes(t, 16, 1, dataChunkCount, false, 1) } func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck bool, po uint8) { From 1f7ee0d2d76c850f8b88263780b22f7d7b164e50 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Fri, 26 Jan 2018 17:31:38 +0100 Subject: [PATCH 14/15] swarm/network/stream: test improvements --- swarm/network/stream/delivery.go | 7 ++- swarm/network/stream/delivery_test.go | 72 ++++++++++++++----------- swarm/network/stream/stream.go | 2 +- swarm/network/stream/syncer_test.go | 4 +- swarm/network/stream/testing/testing.go | 25 +++++---- 5 files changed, 61 insertions(+), 49 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index de05cd5293..2f7dde3721 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -96,7 +96,12 @@ func (s *SwarmChunkServer) processDeliveries() { // SetNextBatch func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) { - hashes = <-s.batchC + select { + case hashes = <-s.batchC: + case <-s.quit: + return + } + from = s.currentLen s.currentLen += uint64(len(hashes)) to = s.currentLen diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index e55e78aee9..183ea2b9e9 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -381,7 +381,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck for j := 0; j < nodes-1; j++ { id := sim.IDs[j] err := sim.CallClient(id, func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(id, client, peerCount(id), errc, quitC) + err := streamTesting.WatchDisconnections(id, client, errc, quitC) if err != nil { return err } @@ -489,6 +489,7 @@ func BenchmarkDeliveryFromNodesWithCheck(b *testing.B) { func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skipCheck bool) { defaultSkipCheck = skipCheck toAddr = network.NewAddrFromNodeID + timeout := 300 * time.Second ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() @@ -527,9 +528,10 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // channel to signal simulation initialisation with action call complete // or node disconnections - simErrC := make(chan error) + disconnectC := make(chan error) quitC := make(chan struct{}) - defer close(quitC) + + initC := make(chan error) action := func(ctx context.Context) error { // each node Subscribes to each other's swarmChunkServerStreamName @@ -546,13 +548,13 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip break } } - + var err error // each node except the last one subscribes to the upstream swarm chunk server stream // which responds to chunk retrieve requests for j := 0; j < nodes-1; j++ { id := sim.IDs[j] - simErrC <- sim.CallClient(id, func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(id, client, peerCount(id), simErrC, quitC) + err = sim.CallClient(id, func(client *rpc.Client) error { + err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC) if err != nil { return err } @@ -561,23 +563,17 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip sid := sim.IDs[j+1] // the upstream peer's id return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) }) + if err != nil { + break + } } - // signal to the benchmark that setup is complete - return err + initC <- err + return nil } // the check function is only triggered when the benchmark finishes - checkC := make(chan error) trigger := make(chan discover.NodeID) check := func(ctx context.Context, id discover.NodeID) (_ bool, err error) { - select { - case <-ctx.Done(): - err = ctx.Err() - case err = <-checkC: - } - if err != nil { - return false, err - } return true, nil } @@ -595,26 +591,15 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip errc := make(chan error) go func() { _, err := sim.Run(ctx, conf) + close(quitC) errc <- err }() // wait for simulation action to complete stream subscriptions - err = <-simErrC + err = <-initC if err != nil { b.Fatalf("simulation failed to initialise. expected no error. got %v", err) } - go func() { - for { - var err error - select { - case err = <-simErrC: - case <-quitC: - return - } - trigger <- sim.IDs[0] - checkC <- err - } - }() // create a retriever dpa for the pivot node // by now deliveries are set for each node by the streamer service @@ -627,6 +612,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // benchmark loop b.ResetTimer() b.StopTimer() +Loop: for i := 0; i < b.N; i++ { // uploading chunkCount random chunks to the last node hashes := make([]storage.Key, chunkCount) @@ -666,12 +652,34 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip } } b.StopTimer() + + select { + case err = <-disconnectC: + if err != nil { + break Loop + } + default: + } + if misses > 0 { - simErrC <- fmt.Errorf("%v chunk not found out of %v", misses, total) + err = fmt.Errorf("%v chunk not found out of %v", misses, total) + break Loop } } + + select { + case <-quitC: + case trigger <- sim.IDs[0]: + } + if err == nil { + err = <-errc + } else { + if e := <-errc; e != nil { + b.Errorf("sim.Run function error: %v", e) + } + } + // benchmark over, trigger the check function to conclude the simulation - err = <-errc if err != nil { b.Fatalf("expected no error. got %v", err) } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 30cd71548f..e29881df79 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -198,7 +198,7 @@ func (r *Registry) run(p *protocols.Peer) error { sp := NewPeer(p, r) r.setPeer(sp) defer r.deletePeer(sp) - // defer close(sp.quit + defer close(sp.quit) defer sp.close() return sp.Run(sp.HandleMsg) } diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 72f4a53b4b..e80cbb77d1 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -159,7 +159,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck id := sim.IDs[j] err := sim.CallClient(id, func(client *rpc.Client) error { // report disconnect events to the error channel cos peers should not disconnect - err := streamTesting.WatchDisconnections(id, client, peerCount(id), errc, quitC) + err := streamTesting.WatchDisconnections(id, client, errc, quitC) if err != nil { return err } @@ -224,6 +224,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } if result.Error != nil { t.Fatalf("Simulation failed: %s", result.Error) - streamTesting.CheckResult(t, result, startedAt, finishedAt) } + streamTesting.CheckResult(t, result, startedAt, finishedAt) } diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index 8ef750df1b..e788e13dd8 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -208,26 +208,23 @@ func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.Ste return result, nil } -func WatchDisconnections(id discover.NodeID, client *rpc.Client, expectedConnCount int, errc chan error, quitC chan struct{}) error { +func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) error { events := make(chan *p2p.PeerEvent) sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { return fmt.Errorf("error getting peer events for node %v: %s", id, err) } go func() { - defer sub.Unsubscribe() - select { - case <-quitC: - if expectedConnCount <= 0 { + for { + select { + case <-quitC: return - } - case e := <-events: - expectedConnCount-- - errc <- fmt.Errorf("peerEvent for node %v: %v", id, e) - case err := <-sub.Err(): - expectedConnCount = 0 - if err != nil { - errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err) + case e := <-events: + errc <- fmt.Errorf("peerEvent for node %v: %v", id, e) + case err := <-sub.Err(): + if err != nil { + errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err) + } } } }() @@ -237,6 +234,7 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, expectedConnCou func Trigger(d time.Duration, quitC chan struct{}, ids ...discover.NodeID) chan discover.NodeID { trigger := make(chan discover.NodeID) go func() { + defer close(trigger) ticker := time.NewTicker(d) defer ticker.Stop() // we are only testing the pivot node (net.Nodes[0]) @@ -245,6 +243,7 @@ func Trigger(d time.Duration, quitC chan struct{}, ids ...discover.NodeID) chan select { case trigger <- id: case <-quitC: + return } } } From bd69bcb0ce10a4f14ab3a440ed334ab953d19c18 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jano=C5=A1=20Gulja=C5=A1?= Date: Wed, 31 Jan 2018 15:42:51 +0100 Subject: [PATCH 15/15] p2p/protocols, swarm/network/stream, swarm/storage: clean debug changes Branch swarm-network-rewrite-syncer-test contains a number of changes related to swarm/network/stream package debugging. This change removes this changes and sets changed variables to the ones in swarm-network-rewrite-syncer branch. --- p2p/protocols/protocol.go | 2 - swarm/network/stream/delivery.go | 101 +++++++++------------------- swarm/network/stream/messages.go | 15 ++--- swarm/network/stream/peer.go | 8 +-- swarm/network/stream/stream.go | 7 +- swarm/network/stream/syncer_test.go | 13 +--- swarm/storage/dbstore.go | 56 ++------------- swarm/storage/localstore.go | 8 +-- swarm/swarm.go | 2 +- 9 files changed, 50 insertions(+), 162 deletions(-) diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 48fc5e9fcc..7b04069edf 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -34,7 +34,6 @@ import ( "reflect" "sync" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" ) @@ -211,7 +210,6 @@ func (p *Peer) Run(handler func(msg interface{}) error) error { // if they are useful for other protocols // overwrite Disconnect for testing, so that protocol readloop quits func (p *Peer) Drop(err error) { - log.Error("p2p protocol DROP", "err", err) p.Disconnect(p2p.DiscSubprotocolError) } diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 2f7dde3721..30bdd80fd8 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -34,14 +34,11 @@ const ( ) type Delivery struct { - db *storage.DBAPI - overlay network.Overlay - receiveC chan *ChunkDeliveryMsg - getPeer func(discover.NodeID) *Peer - quit chan struct{} - counterIn int - counterDone int - counterHash int + db *storage.DBAPI + overlay network.Overlay + receiveC chan *ChunkDeliveryMsg + getPeer func(discover.NodeID) *Peer + quit chan struct{} } func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery { @@ -160,7 +157,7 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e if req.SkipCheck { err := sp.Deliver(chunk, s.priority) if err != nil { - sp.Drop(fmt.Errorf("handleRetrieveRequestMsg: %v", err)) + sp.Drop(err) } } streamer.deliveryC <- chunk.Key[:] @@ -179,67 +176,39 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e type ChunkDeliveryMsg struct { Key storage.Key SData []byte // the stored chunk Data (incl size) - peer *Peer + peer *Peer // set in handleChunkDeliveryMsg } func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error { - d.counterIn++ req.peer = sp - log.Error("push to receiveC", "hash", storage.Key(req.Key).Hex()) d.receiveC <- req return nil } func (d *Delivery) processReceivedChunks() { - done := make(chan struct{}) - timer := time.NewTimer(2 * time.Second) - defer timer.Stop() - // R: +R: for req := range d.receiveC { - log.Error("pop from receiveC", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex()) - timer.Reset(1 * time.Second) - go func(req *ChunkDeliveryMsg) { - defer func() { done <- struct{}{} }() - // this should be has locally - log.Error("before db.Get", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex()) - chunk, err := d.db.Get(req.Key) - if !bytes.Equal(chunk.Key, req.Key) { - panic(fmt.Errorf("processReceivedChunks: chunk key %s != req key %s (peer %s)", chunk.Key.Hex(), storage.Key(req.Key).Hex(), req.peer.ID())) - } - log.Error("after db.Get", "peer", req.peer.ID(), "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) - if err == nil { - log.Error("found existing?", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) - // continue R - return - } - if err != storage.ErrFetching { - panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) - } - select { - case <-chunk.ReqC: - log.Error("someone else delivered?", "hash", chunk.Key.Hex()) - // continue R - return - default: - } - // go func() { - chunk.SData = req.SData - log.Error("received delivery", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) - d.db.Put(chunk) - log.Error("put to db", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) - chunk.WaitToStore() - close(chunk.ReqC) - //log.Warn("received delivery stored", "hash", chunk.Key) - log.Error("requesters notified", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) - d.counterDone++ - // }() - }(req) - select { - case <-timer.C: - log.Error("!!!unable to process delivery", "peer", req.peer.ID(), "hash", req.Key.Hex()) - case <-done: - log.Error("done processing delivery", "peer", req.peer.ID(), "hash", req.Key.Hex()) + // this should be has locally + chunk, err := d.db.Get(req.Key) + if !bytes.Equal(chunk.Key, req.Key) { + panic(fmt.Errorf("processReceivedChunks: chunk key %s != req key %s (peer %s)", chunk.Key.Hex(), storage.Key(req.Key).Hex(), req.peer.ID())) } + if err == nil { + continue R + } + if err != storage.ErrFetching { + panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) + } + select { + case <-chunk.ReqC: + log.Error("someone else delivered?", "hash", chunk.Key.Hex()) + continue R + default: + } + chunk.SData = req.SData + d.db.Put(chunk) + chunk.WaitToStore() + close(chunk.ReqC) } } @@ -247,18 +216,17 @@ func (d *Delivery) processReceivedChunks() { func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ...discover.NodeID) error { var success bool var err error - log.Warn("request", "hash", hash) d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool { spId := p.(*network.BzzPeer).ID() for _, p := range peersToSkip { if p == spId { - log.Warn("skip peer", "peer", spId) + log.Trace("Delivery.RequestFromPeers: skip peer", "peer", spId) return true } } sp := d.getPeer(spId) if sp == nil { - log.Warn("peer not found", "id", spId) + log.Warn("Delivery.RequestFromPeers: peer not found", "id", spId) return true } // TODO: skip light nodes that do not accept retrieve requests @@ -274,12 +242,3 @@ func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ... } return errors.New("no peer found") } - -func (d *Delivery) PrintCounters(id discover.NodeID) { - if d.counterHash != d.counterDone { - log.Error(fmt.Sprintf("delivery %s: HASH and DONE not the same", id)) - } - log.Error(fmt.Sprintf("delivery %s chunks hash: %d", id, d.counterHash)) - log.Error(fmt.Sprintf("delivery %s chunks in: %d", id, d.counterIn)) - log.Error(fmt.Sprintf("delivery %s chunks done: %d", id, d.counterDone)) -} diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index aa8f5f75c4..b4aadc75f1 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -17,7 +17,6 @@ package stream import ( - "errors" "fmt" "sync" "time" @@ -83,7 +82,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error { log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To) go func() { if err := p.SendOfferedHashes(os, req.From, req.To); err != nil { - p.Drop(fmt.Errorf("handleSubscribeMsg SendOfferedHashes: %v", err)) + p.Drop(err) } }() return nil @@ -122,8 +121,6 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { for i := 0; i < len(hashes); i += HashSize { hash := hashes[i : i+HashSize] - p.streamer.delivery.counterHash++ - if wait := s.NeedData(hash); wait != nil { want.Set(i/HashSize, true) wg.Add(1) @@ -171,19 +168,19 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { } go func() { select { - case <-time.After(1 * time.Second): - p.Drop(errors.New("timeout waiting for batch to be delivered")) + case <-time.After(30 * time.Second): + p.Drop(err) return case err := <-s.next: if err != nil { - p.Drop(fmt.Errorf("handleOfferedHashesMsg next: %v", err)) + 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) if err != nil { - p.Drop(fmt.Errorf("handleOfferedHashesMsg set priority: %v", err)) + p.Drop(err) } }() return nil @@ -216,7 +213,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { // launch in go routine since GetBatch blocks until new hashes arrive go func() { if err := p.SendOfferedHashes(s, req.From, req.To); err != nil { - p.Drop(fmt.Errorf("handleWantedHashesMsg SendOfferedHashes: %v", err)) + p.Drop(err) } }() // go p.SendOfferedHashes(s, req.From, req.To) diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index cd7cdc2c9a..75bec53818 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -28,7 +28,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -var sendTimeout = 1 * time.Second +var sendTimeout = 5 * time.Second // Peer is the Peer extention for the streaming protocol type Peer struct { @@ -101,11 +101,7 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { Stream: s.stream, Key: s.key, } - log.Error("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to) - for i := 0; i < len(hashes); i += HashSize { - hash := hashes[i : i+HashSize] - log.Error("Swarm syncer offer hash", "peer", p.ID(), "stream", s.stream, "hash", storage.Key(hash).Hex(), "len", len(hashes), "from", from, "to", to) - } + log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to) return p.SendPriority(msg, s.priority) } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index e29881df79..d6514808be 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -38,8 +38,8 @@ const ( Mid High Top - PriorityQueue // number of queues - PriorityQueueCap = 3 // queue capacity + PriorityQueue // number of queues + PriorityQueueCap = 32 // queue capacity HashSize = 32 ) @@ -268,8 +268,7 @@ type client struct { live bool stream string key []byte - // quit chan struct{} - next chan error + next chan error } // Client interface for incoming peer Streamer diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index e80cbb77d1..58d780c36f 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -34,7 +34,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -const dataChunkCount = 1000 +const dataChunkCount = 500 func TestSyncerSimulation(t *testing.T) { testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1) @@ -70,13 +70,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck t.Fatal(err.Error()) } - // DEBUG: - defer func() { - for _, id := range sim.IDs { - deliveries[id].PrintCounters(id) - } - }() - // HACK: these are global variables in the test so that they are available for // the service constructor function // TODO: will this work with exec/docker adapter? @@ -186,7 +179,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck default: } - log.Error("starting dbs check", "node", id) i := nodeIndex[id] var total, found int for j := i; j < nodes; j++ { @@ -196,7 +188,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck if err == storage.ErrFetching { <-chunk.ReqC } else if err != nil { - log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) continue } // needed for leveldb not to be closed? @@ -204,7 +195,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck found++ } } - log.Error("sync check", "node", id, "index", i, "bin", po, "found", found, "total", total) + log.Debug("sync check", "node", id, "index", i, "bin", po, "found", found, "total", total) return total == found, nil } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index fecb199e1b..17a7534646 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -27,14 +27,13 @@ import ( "bytes" "encoding/binary" "encoding/hex" - "encoding/json" "fmt" "io" "io/ioutil" "sync" - "time" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/swarm/storage/mock" "github.com/syndtr/goleveldb/leveldb" "github.com/syndtr/goleveldb/leveldb/opt" @@ -82,7 +81,6 @@ type DbStore struct { po func(Key) uint8 batchC chan bool - quit chan struct{} batchesC chan struct{} batch *leveldb.Batch lock sync.RWMutex @@ -106,7 +104,6 @@ func NewDbStore(path string, hash SwarmHasher, capacity uint64, po func(Key) uin s.hashfunc = hash s.batchC = make(chan bool) - s.quit = make(chan struct{}) s.batchesC = make(chan struct{}, 1) go s.writeBatches() s.batch = new(leveldb.Batch) @@ -221,12 +218,7 @@ func getDataKey(idx uint64, po uint8) []byte { } func encodeIndex(index *dpaDBIndex) []byte { - //data, _ := rlp.EncodeToBytes(index) - - data, err := json.Marshal(index) - if err != nil { - panic(err) - } + data, _ := rlp.EncodeToBytes(index) return data } @@ -235,9 +227,8 @@ func encodeData(chunk *Chunk) []byte { } func decodeIndex(data []byte, index *dpaDBIndex) error { - // dec := rlp.NewStream(bytes.NewReader(data), 0) - // return dec.Decode(index) - return json.Unmarshal(data, index) + dec := rlp.NewStream(bytes.NewReader(data), 0) + return dec.Decode(index) } @@ -546,55 +537,25 @@ func (s *DbStore) CurrentStorageIndex() uint64 { } func (s *DbStore) Put(chunk *Chunk) { - log.Error("DbStore.Put", "hash", chunk.Key.Hex()) - done := make(chan struct{}) - defer close(done) - key := Key(append(make([]byte, 0), chunk.Key...)) - go func() { - log.Error("DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex()) - select { - case <-time.After(1 * time.Second): - log.Error("DbStore.Put WAITING", "hash", chunk.Key.Hex()) - case <-done: - log.Error("DbStore.Put EXITED", "hash", chunk.Key.Hex()) - if !bytes.Equal(chunk.Key, key) { - panic(fmt.Errorf("DbStore.Get: chunk key %s != req key %s", chunk.Key.Hex(), key.Hex())) - } - } - }() - ikey := getIndexKey(chunk.Key) var index dpaDBIndex po := s.po(chunk.Key) - log.Error("DbStore.db.Get is being called...", "hash", chunk.Key.Hex()) - - log.Error("DbStore.LOCK acquiring", "hash", chunk.Key.Hex()) s.lock.Lock() - log.Error("DbStore.LOCK acquired", "hash", chunk.Key.Hex()) defer s.lock.Unlock() idata, err := s.db.Get(ikey) - log.Error("DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err) if err != nil { s.doPut(chunk, ikey, &index, po) batchC := s.batchC go func() { - defer func() { - if err := recover(); err != nil { - log.Error("DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err) - } - }() - <-batchC close(chunk.dbStored) }() - log.Error("DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) } else { log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access")) decodeIndex(idata, &index) close(chunk.dbStored) - log.Error("DbStore.Put already found", "hash", chunk.Key.Hex()) } index.Access = s.accessCnt s.accessCnt++ @@ -604,7 +565,6 @@ func (s *DbStore) Put(chunk *Chunk) { case s.batchesC <- struct{}{}: default: } - log.Error("DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err) } // force putting into db, does not check access index @@ -734,11 +694,6 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) { chunk = NewChunk(key, nil) decodeData(data, chunk) - - if !bytes.Equal(chunk.Key, key) { - panic(fmt.Errorf("DbStore.Get: chunk key %s != req key %s", chunk.Key.Hex(), key.Hex())) - } - } else { err = ErrNotFound } @@ -802,9 +757,8 @@ func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, untilkey := getDataKey(until, po) it := s.db.NewIterator() defer it.Release() - it.Seek(sincekey) - for it.Next() { + for ok := it.Seek(sincekey); ok; ok = it.Next() { dbkey := it.Key() if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 { break diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 40103503be..0f77488b64 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -17,7 +17,6 @@ package storage import ( - "bytes" "encoding/binary" "fmt" "path/filepath" @@ -103,12 +102,7 @@ func (self *LocalStore) Put(chunk *Chunk) { dbStored: chunk.dbStored, } self.memStore.Put(c) - log.Error("put to memstore", "hash", c.Key.Hex()) self.DbStore.Put(c) - log.Error("put to dbstore", "hash", c.Key.Hex()) - if !bytes.Equal(chunk.Key, c.Key) { - panic(fmt.Errorf("LocalStore.Put: chunk %s != c %s", chunk.Key.Hex(), c.Key.Hex())) - } } // Get(chunk *Chunk) looks up a chunk in the local stores @@ -132,7 +126,7 @@ func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { return } chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - //self.memStore.Put(chunk) + self.memStore.Put(chunk) return } diff --git a/swarm/swarm.go b/swarm/swarm.go index b97390e369..d566918fe7 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -132,7 +132,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e db := storage.NewDBAPI(self.lstore) delivery := stream.NewDelivery(to, db) - self.streamer = stream.NewRegistry(addr, delivery) + self.streamer = stream.NewRegistry(addr, delivery, self.lstore, false) stream.RegisterSwarmSyncerServer(self.streamer, db) stream.RegisterSwarmSyncerClient(self.streamer, db)