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