diff --git a/bzz/chunker.go b/bzz/chunker.go index dc11f3d884..c66b75f80a 100644 --- a/bzz/chunker.go +++ b/bzz/chunker.go @@ -306,7 +306,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { dpaLogger.Debugf("chunk data received for %x", chunk.Key[:4]) fmt.Printf("chunk data received for %x\n", chunk.Key[:4]) } - if chunk.Data == nil { + if len(chunk.Data) == 0 { return 0, notFound } @@ -347,10 +347,11 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr treeSize /= self.chunker.Branches depth-- } - dpaLogger.Debugf("-> depth: %v, loff: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, off, eoff, chunk.Size, treeSize) - fmt.Printf("-> depth: %v, loff: %v, eoff: %v, chunk.Size: %v, treeSize: %v\n", depth, off, eoff, chunk.Size, treeSize) + fmt.Printf("-> depth: %v, loff: %v, eoff: %v, len(chunk.Data): %v, chunk.Size: %v, treeSize: %v\n", depth, off, eoff, len(chunk.Data), chunk.Size, treeSize) 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) + copy(b, chunk.Data[off:eoff]) return // simply give back the chunks reader for content chunks } @@ -375,32 +376,33 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr wg.Add(1) go func(j int64) { - dpaLogger.Debugf("subtree index: %v", j) childKey := chunk.Data[j*self.chunker.hashSize : (j+1)*self.chunker.hashSize] + dpaLogger.Debugf("subtree index: %v -> %x", j, childKey[:4]) - chunk := &Chunk{ + ch := &Chunk{ Key: childKey, - C: make(chan bool, 1), // close channel to signal data delivery + C: make(chan bool), // close channel to signal data delivery } - dpaLogger.Debugf("chunk data sent for %x (key interval in chunk %v-%v)", chunk.Key[:4], j*self.chunker.hashSize, (j+1)*self.chunker.hashSize) - self.chunkC <- chunk // submit retrieval request, someone should be listening on the other side (or we will time out globally) + dpaLogger.Debugf("chunk data sent for %x (key interval in chunk %v-%v)", ch.Key[:4], j*self.chunker.hashSize, (j+1)*self.chunker.hashSize) + fmt.Printf("chunk data sent for %x (key interval in chunk %v-%v)\n", ch.Key[:4], j*self.chunker.hashSize, (j+1)*self.chunker.hashSize) + self.chunkC <- ch // submit retrieval request, someone should be listening on the other side (or we will time out globally) // waiting for the chunk retrieval select { case <-self.quitC: // this is how we control process leakage (quitC is closed once join is finished (after timeout)) return - case <-chunk.C: // bells are ringing, data have been delivered + case <-ch.C: // bells are ringing, data have been delivered dpaLogger.Debugf("chunk data received") } if soff < off { soff = off } - if chunk.Data == nil { + if len(ch.Data) == 0 { self.errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize) return } - self.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/self.chunker.Branches, chunk, &wg) + self.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/self.chunker.Branches, ch, &wg) }(i) } //for wg.Wait() diff --git a/bzz/common_test.go b/bzz/common_test.go index 17f3c86c57..0f577cea83 100644 --- a/bzz/common_test.go +++ b/bzz/common_test.go @@ -61,8 +61,8 @@ SPLIT: quit := make(chan bool) go func() { - for chunk := range chunkC { - go func() { + for ch := range chunkC { + go func(chunk *Chunk) { storedChunk, err := m.Get(chunk.Key) if err == notFound { dpaLogger.DebugDetailf("chunk '%x' not found", chunk.Key) @@ -72,8 +72,9 @@ SPLIT: chunk.Data = storedChunk.Data chunk.Size = storedChunk.Size } + dpaLogger.DebugDetailf("chunk '%x' not found", chunk.Key[:4]) close(chunk.C) - }() + }(ch) } }() diff --git a/bzz/dbstore_test.go b/bzz/dbstore_test.go index d1d8939259..79ee4315f8 100644 --- a/bzz/dbstore_test.go +++ b/bzz/dbstore_test.go @@ -22,6 +22,12 @@ func testDbStore(l int64, branches int64, t *testing.T) { testStore(m, l, branches, t) } +func TestDbStore128_0x1000000(t *testing.T) { + // defer test.Testlog(t).Detach() + test.LogInit() + testDbStore(0x1000000, 128, t) +} + func TestDbStore128_10000(t *testing.T) { // defer test.Testlog(t).Detach() test.LogInit() diff --git a/bzz/dpa.go b/bzz/dpa.go index 30f4c2f24d..b6a68e71a6 100644 --- a/bzz/dpa.go +++ b/bzz/dpa.go @@ -124,9 +124,9 @@ func (self *DPA) retrieveLoop() { go func() { RETRIEVE: - for chunk := range self.retrieveC { + for ch := range self.retrieveC { - go func() { + go func(chunk *Chunk) { storedChunk, err := self.ChunkStore.Get(chunk.Key) if err == notFound { dpaLogger.DebugDetailf("chunk '%x' not found", chunk.Key) @@ -137,7 +137,7 @@ func (self *DPA) retrieveLoop() { chunk.Size = storedChunk.Size } close(chunk.C) - }() + }(ch) select { case <-self.quitC: break RETRIEVE @@ -151,8 +151,11 @@ func (self *DPA) storeLoop() { self.storeC = make(chan *Chunk) go func() { STORE: - for chunk := range self.storeC { - self.ChunkStore.Put(chunk) + for ch := range self.storeC { + // go func(chunk *Chunk) { + self.ChunkStore.Put(ch) + // self.ChunkStore.Put(chunk) + // }(ch) select { case <-self.quitC: break STORE diff --git a/bzz/dpa_test.go b/bzz/dpa_test.go index f455d4ad07..978dd385c2 100644 --- a/bzz/dpa_test.go +++ b/bzz/dpa_test.go @@ -14,6 +14,7 @@ func TestDPA(t *testing.T) { test.LogInit() os.RemoveAll("/tmp/bzz") dbStore, err := newDbStore("/tmp/bzz") + // dbStore.setCapacity(50000) if err != nil { t.Errorf("DB error: %v", err) }