Merge pull request #336 from ethersphere/more_tracing

deadlock on retrieve chunk + more tracing for internal swarm components

log/format.go

@@ -15,7 +15,7 @@ import (
 
 const (
 	timeFormat     = "2006-01-02T15:04:05-0700"
-	termTimeFormat = "01-02|15:04:05"
+	termTimeFormat = "01-02|15:04:05.999999"
 	floatFormat    = 'f'
 	termMsgJust    = 40
 )

swarm/storage/chunker.go

@@ -23,6 +23,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 )
 
@@ -332,6 +333,7 @@ func (self *TreeChunker) Join(key Key, chunkC chan *Chunk, depth int) LazySectio
 
 // Size is meant to be called on the LazySectionReader
 func (self *LazyChunkReader) Size(quitC chan bool) (n int64, err error) {
+	log.Debug("lazychunkreader.size", "key", self.key)
 	if self.chunk != nil {
 		return self.chunk.Size, nil
 	}
@@ -459,15 +461,18 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
 // block until they time out or arrive
 // abort if quitC is readable
 func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk {
+	log.Debug("retrieve", "key", key)
 	chunk := NewChunk(key, nil)
 	chunk.C = make(chan bool)
 	// submit chunk for retrieval
+	log.Debug("submit chunk for retrieval", "key", key)
 	select {
 	case chunkC <- chunk: // submit retrieval request, someone should be listening on the other side (or we will time out globally)
 	case <-quitC:
 		return nil
 	}
 	// waiting for the chunk retrieval
+	log.Debug("waiting for the chunk retrieval", "key", key)
 	select { // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
 
 	case <-quitC:
@@ -478,11 +483,13 @@ func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk {
 	if len(chunk.SData) == 0 {
 		return nil
 	}
+	log.Debug("chunk retrieved", "key", key)
 	return chunk
 }
 
 // Read keeps a cursor so cannot be called simulateously, see ReadAt
 func (self *LazyChunkReader) Read(b []byte) (read int, err error) {
+	log.Debug("lazychunkreader.read", "key", self.key)
 	read, err = self.ReadAt(b, self.off)
 
 	self.off += int64(read)
@@ -494,6 +501,7 @@ var errWhence = errors.New("Seek: invalid whence")
 var errOffset = errors.New("Seek: invalid offset")
 
 func (s *LazyChunkReader) Seek(offset int64, whence int) (int64, error) {
+	log.Debug("lazychunkreader.seek", "key", s.key, "offset", offset)
 	switch whence {
 	default:
 		return 0, errWhence

swarm/storage/ldbstore.go

@@ -546,6 +546,7 @@ func (s *LDBStore) CurrentStorageIndex() uint64 {
 }
 
 func (s *LDBStore) Put(chunk *Chunk) {
+	log.Trace("ldbstore.put", "key", chunk.Key)
 	ikey := getIndexKey(chunk.Key)
 	var index dpaDBIndex
 
@@ -553,6 +554,7 @@ func (s *LDBStore) Put(chunk *Chunk) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
+	log.Trace("ldbstore.put: s.db.Get", "key", chunk.Key)
 	idata, err := s.db.Get(ikey)
 	if err != nil {
 		s.doPut(chunk, ikey, &index, po)
@@ -562,7 +564,7 @@ func (s *LDBStore) Put(chunk *Chunk) {
 			close(chunk.dbStored)
 		}()
 	} else {
-		log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access"))
+		log.Trace("ldbstore.put: chunk already exists, only update access", "key", chunk.Key)
 		decodeIndex(idata, &index)
 		close(chunk.dbStored)
 	}
@@ -578,6 +580,7 @@ func (s *LDBStore) Put(chunk *Chunk) {
 
 // force putting into db, does not check access index
 func (s *LDBStore) doPut(chunk *Chunk, ikey []byte, index *dpaDBIndex, po uint8) {
+	log.Trace("ldbstore.doPut", "key", chunk.Key)
 	data := s.encodeDataFunc(chunk)
 	s.batch.Put(getDataKey(s.dataIdx, po), data)
 	index.Idx = s.dataIdx
@@ -659,6 +662,7 @@ func (s *LDBStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
 }
 
 func (s *LDBStore) Get(key Key) (chunk *Chunk, err error) {
+	log.Trace("ldbstore.get", "key", key)
 	s.lock.Lock()
 	defer s.lock.Unlock()
 	return s.get(key)
@@ -671,6 +675,7 @@ func (s *LDBStore) get(key Key) (chunk *Chunk, err error) {
 		var data []byte
 		if s.getDataFunc != nil {
 			// if getDataFunc is defined, use it to retrieve the chunk data
+			log.Trace("ldbstore.get retrieve with getDataFunc", "key", key)
 			data, err = s.getDataFunc(key)
 			if err != nil {
 				return
@@ -680,9 +685,9 @@ func (s *LDBStore) get(key Key) (chunk *Chunk, err error) {
 			proximity := s.po(key)
 			datakey := getDataKey(indx.Idx, proximity)
 			data, err = s.db.Get(datakey)
-			log.Trace(fmt.Sprintf("DBStore: Chunk %v indexkey %v datakey %x proximity %d", key.Log(), indx.Idx, datakey, proximity))
+			log.Trace("ldbstore.get retrieve", "key", key, "indexkey", indx.Idx, "datakey", datakey, "proximity", proximity)
 			if err != nil {
-				log.Trace(fmt.Sprintf("DBStore: Chunk %v found but could not be accessed: %v", key.Log(), err))
+				log.Trace("ldbstore.get chunk found but could not be accessed", "key", key, "err", err)
 				s.delete(indx.Idx, getIndexKey(key), s.po(key))
 				return
 			}
@@ -695,13 +700,14 @@ func (s *LDBStore) get(key Key) (chunk *Chunk, err error) {
 			hash := hasher.Sum(nil)
 
 			if !bytes.Equal(hash, key) {
-				log.Error(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:]))
+				log.Error("apparent key/hash mismatch", "hash", hash, "key", key[:])
 				s.delete(indx.Idx, getIndexKey(key), s.po(key))
-				log.Error("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'")
+				log.Error("invalid chunk in database.")
 			}
 		}
 
 		chunk = NewChunk(key, nil)
+		close(chunk.dbStored)
 		decodeData(data, chunk)
 	} else {
 		err = ErrChunkNotFound

swarm/storage/localstore.go

@@ -85,6 +85,7 @@ func (self *LocalStore) CacheCounter() uint64 {
 // LocalStore is itself a chunk store
 // unsafe, in that the data is not integrity checked
 func (self *LocalStore) Put(chunk *Chunk) {
+	log.Trace("localstore.put", "key", chunk.Key)
 	self.mu.Lock()
 	defer self.mu.Unlock()
 

swarm/storage/memstore.go

@@ -144,6 +144,7 @@ func (s *MemStore) Counter() uint {
 
 // entry (not its copy) is going to be in MemStore
 func (s *MemStore) Put(entry *Chunk) {
+	log.Trace("memstore.put", "key", entry.Key)
 	if s.capacity == 0 {
 		return
 	}
@@ -219,6 +220,7 @@ func (s *MemStore) Put(entry *Chunk) {
 }
 
 func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) {
+	log.Trace("memstore.get", "key", hash)
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -228,6 +230,7 @@ func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) {
 		l := hash.bits(bitpos, node.bits)
 		st := node.subtree[l]
 		if st == nil {
+			log.Trace("memstore.get ErrChunkNotFound", "key", hash)
 			return nil, ErrChunkNotFound
 		}
 		bitpos += node.bits
@@ -249,6 +252,7 @@ func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) {
 		err = ErrChunkNotFound
 	}
 
+	log.Trace("memstore.get return", "key", hash, "chunk", chunk, "err", err)
 	return
 }
 
@@ -296,11 +300,11 @@ func (s *MemStore) removeOldest() {
 
 	}
 
+	if node.entry.ReqC == nil {
 		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 {
 		memstoreRemoveCounter.Inc(1)
 		node.entry = nil
 		s.entryCnt--

swarm/storage/pyramid.go

@@ -266,8 +266,6 @@ func (self *PyramidChunker) processor(id int64, jobC chan *chunkJob, chunkC chan
 }
 
 func (self *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJob, chunkC chan *Chunk, storageWG *sync.WaitGroup) {
-	log.Debug("pyramid.chunker: processChunk()", "id", id)
-
 	hasher.ResetWithLength(job.chunk[:8]) // 8 bytes of length
 	hasher.Write(job.chunk[8:])           // minus 8 []byte length
 	h := hasher.Sum(nil)