From eae4473a81a70a03f512f2c8c88b535ea7c91e6c Mon Sep 17 00:00:00 2001 From: zelig Date: Wed, 24 Jan 2018 10:11:09 +0100 Subject: [PATCH] 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