From c14c85f03bc06c51e759e5d8dea563dd38300f12 Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 14 May 2015 13:16:22 +0200 Subject: [PATCH] introduce dbStored lock to make sure chunk not removed from memory store before written to disk --- bzz/chunker.go | 16 ++++++++-------- bzz/dbstore.go | 5 ++--- bzz/dpa.go | 2 +- bzz/localstore.go | 2 +- bzz/memstore.go | 6 +++--- bzz/protocol.go | 4 ++++ 6 files changed, 19 insertions(+), 16 deletions(-) diff --git a/bzz/chunker.go b/bzz/chunker.go index 32015bf29e..b63a0d43b3 100644 --- a/bzz/chunker.go +++ b/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") diff --git a/bzz/dbstore.go b/bzz/dbstore.go index 78fcb6ec0b..8ca2f427ef 100644 --- a/bzz/dbstore.go +++ b/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 diff --git a/bzz/dpa.go b/bzz/dpa.go index 401585b087..3bd6813a1c 100644 --- a/bzz/dpa.go +++ b/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 } diff --git a/bzz/localstore.go b/bzz/localstore.go index 7f6a4eabd8..243eceda8b 100644 --- a/bzz/localstore.go +++ b/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) diff --git a/bzz/memstore.go b/bzz/memstore.go index 6204a067db..73786856cd 100644 --- a/bzz/memstore.go +++ b/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-- diff --git a/bzz/protocol.go b/bzz/protocol.go index ac75bb644b..b3adb373db 100644 --- a/bzz/protocol.go +++ b/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