introduce dbStored lock to make sure chunk not removed from memory store before written to disk

bzz/chunker.go

@@ -318,16 +318,16 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 		dpaLogger.Debugf("quit")
 		return
 	case <-chunk.C: // bells are ringing, data have been delivered
-		dpaLogger.Debugf("chunk data received for %x", chunk.Key[:4])
+		// dpaLogger.Debugf("chunk data received for %x", chunk.Key[:4])
 	}
 	if len(chunk.SData) == 0 {
-		dpaLogger.Debugf("No payload in %x.", chunk.Key)
+		// dpaLogger.Debugf("No payload in %x", chunk.Key)
 		return 0, notFound
 	}
 	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
 	self.size = chunk.Size
 	if b == nil {
-		dpaLogger.Debugf("Size query for %x.", chunk.Key[:4])
+		// dpaLogger.Debugf("Size query for %x", chunk.Key[:4])
 		return
 	}
 	want := int64(len(b))
@@ -350,14 +350,14 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 	}()
 	select {
 	case err = <-self.errC:
-		dpaLogger.Debugf("ReadAt received %v.", err)
+		dpaLogger.Debugf("ReadAt received %v", err)
 		read = len(b)
 		if off+int64(read) == self.size {
 			err = io.EOF
 		}
-		dpaLogger.Debugf("ReadAt returning with %d, %v.", read, err)
+		dpaLogger.Debugf("ReadAt returning with %d, %v", read, err)
 	case <-self.quitC:
-		dpaLogger.Debugf("ReadAt aborted with %d, %v.", read, err)
+		dpaLogger.Debugf("ReadAt aborted with %d, %v", read, err)
 	}
 	return
 }
@@ -365,7 +365,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup) {
 	defer parentWg.Done()
 
-	dpaLogger.Debugf("depth: %v, loff: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, off, eoff, chunk.Size, treeSize)
+	// dpaLogger.Debugf("depth: %v, loff: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, off, eoff, chunk.Size, treeSize)
 
 	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
 
@@ -376,7 +376,7 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
 	}
 
 	if depth == 0 {
-		dpaLogger.Debugf("len(b): %v, off: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, len(b), off, eoff, chunk.Size, treeSize)
+		// dpaLogger.Debugf("depth: %v, len(b): %v, off: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, len(b), off, eoff, chunk.Size, treeSize)
 		if int64(len(b)) != eoff-off {
 			//fmt.Printf("len(b) = %v  off = %v  eoff = %v", len(b), off, eoff)
 			panic("len(b) does not match")

bzz/dbstore.go

@@ -259,6 +259,7 @@ func (s *dbStore) Put(chunk *Chunk) {
 	var index dpaDBIndex
 
 	if s.tryAccessIdx(ikey, &index) {
+		chunk.dbStored.Unlock()
 		return // already exists, only update access
 	}
 
@@ -287,9 +288,7 @@ func (s *dbStore) Put(chunk *Chunk) {
 	batch.Put(ikey, idata)
 
 	s.db.Write(batch)
-	if chunk.dbStored != nil {
-		close(chunk.dbStored)
-	}
+	chunk.dbStored.Unlock()
 }
 
 // try to find index; if found, update access cnt and return true

bzz/dpa.go

@@ -60,7 +60,7 @@ type Chunk struct {
 	C        chan bool      // to signal data delivery by the dpa
 	req      *requestStatus //
 	wg       *sync.WaitGroup
-	dbStored chan bool
+	dbStored sync.Mutex
 	source   *peer
 }
 

bzz/localstore.go

@@ -9,7 +9,7 @@ type localStore struct {
 // localStore is itself a chunk store , to stores a chunk only
 // its integrity is checked ?
 func (self *localStore) Put(chunk *Chunk) {
-	chunk.dbStored = make(chan bool)
+	chunk.dbStored.Lock()
 	self.memStore.Put(chunk)
 	if chunk.wg != nil {
 		chunk.wg.Add(1)

bzz/memstore.go

@@ -314,9 +314,9 @@ func (s *memStore) removeOldest() {
 
 	}
 
-	if node.entry.dbStored != nil {
-		<-node.entry.dbStored
-	}
+	node.entry.dbStored.Lock()
+	node.entry.dbStored.Unlock()
+
 	if node.entry.SData != nil {
 		node.entry = nil
 		s.entryCnt--

bzz/protocol.go

@@ -136,6 +136,10 @@ type retrieveRequestMsgData struct {
 	peer peer // protocol registers the requester
 }
 
+func (self *retrieveRequestMsgData) String() string {
+	return fmt.Sprintf("From: %v, Key: %x; ID: %v, MaxSize: %v, MaxPeers: %v", self.peer.Addr(), self.Key[:4], self.Id, self.MaxSize, self.MaxPeers)
+}
+
 type peerAddr struct {
 	IP   net.IP
 	Port uint16