chunker improvements

- fix want > 0 in join branching node iteration
- add chunkChanCapacity
- add EmbeddedReader / notReadyReader hack to have a reader that replaces itself on demand PHEW!
- add temporary debug info to chunkIO
- add quitC to test loop for Join

bzz/chunkIO.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"io"
+	"sync"
 	"time"
 )
 
@@ -112,6 +113,7 @@ func (s *ChunkReader) Read(p []byte) (n int, err error) {
 }
 
 func (s *ChunkReader) ReadAt(p []byte, off int64) (n int, err error) {
+	dpaLogger.DebugDetailf("chunk reader: slicelen: %v off: %v base %v limit %v", len(p), off, s.base, s.limit)
 	if off < 0 || off >= s.limit-s.base {
 		return 0, io.EOF
 	}
@@ -190,8 +192,10 @@ func (self *LazyChunkReader) Size() (n int64) {
 }
 
 func (self *LazyChunkReader) Read(b []byte) (read int, err error) {
+	dpaLogger.DebugDetailf("%v bytes to read %v - %v ", len(b), self.off, self.Size())
 	read, err = self.ReadAt(b, self.off)
 	self.off += int64(read)
+	dpaLogger.DebugDetailf("%v bytes to read %v - %v: %v %v", len(b), self.off, self.Size(), read, err)
 	return
 }
 
@@ -214,13 +218,15 @@ func (s *LazyChunkReader) Seek(offset int64, whence int) (int64, error) {
 }
 
 func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
-	if off < 0 || off > self.size {
+	dpaLogger.DebugDetailf("%v - %v", off, self.size)
+	if off < 0 || off >= self.size {
 		err = io.EOF
 		return
 	}
 	want := len(b)
 	got := int(self.size - off)
 	if want > got {
+		dpaLogger.DebugDetailf("%v > %v", want, got)
 		err = io.EOF
 	}
 	var index int
@@ -231,17 +237,20 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 	var limit int
 	var reader LazySectionReader
 
-	for i := index; i < len(self.sections) || want == 0; i++ {
+	for i := index; i < len(self.sections) || want > 0; i++ {
 		if reader = self.sections[i]; reader == nil {
 			reader = self.readerFs[i]() // this hooks into the go routines and does a wg.Done once the needed chunks are fetched and processed
 			self.sections[i] = reader   // memoize this reader
 		}
-		if want > int(self.size-off) {
-			limit = int(self.size)
+		if want > int(reader.Size()-off) {
+			limit = int(reader.Size())
 		} else {
 			limit = int(off) + want
 		}
+		dpaLogger.DebugDetailf("%v - %v (%v) want %v, got %v, read %v", off, limit, reader.Size(), want, got, read)
+		reader.Seek(0, 0)
 		if got, err = reader.ReadAt(b[off:limit], off); err != nil {
+			dpaLogger.DebugDetailf("oh oh oh oh")
 			return
 		}
 		read += got
@@ -250,6 +259,36 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 	return
 }
 
+type EmbeddedReader struct {
+	LazySectionReader
+	lock sync.Mutex
+}
+
+type NotReadyReader struct {
+	initF func()
+	r     LazySectionReader
+}
+
+func (self *NotReadyReader) Size() (n int64) {
+	self.initF()
+	return self.r.Size()
+}
+
+func (self *NotReadyReader) Read(b []byte) (read int, err error) {
+	self.initF()
+	return self.r.Read(b)
+}
+
+func (self *NotReadyReader) ReadAt(b []byte, off int64) (read int, err error) {
+	self.initF()
+	return self.r.ReadAt(b, off)
+}
+
+func (self *NotReadyReader) Seek(offset int64, whence int) (int64, error) {
+	self.initF()
+	return self.r.Seek(offset, whence)
+}
+
 // just a wrapper to make a SectionReader conform to LazySectionReader
 // with dummy implementation
 type lazyReader struct {

bzz/chunker.go

@@ -25,8 +25,9 @@ import (
 )
 
 const (
-	hasherfunc crypto.Hash = crypto.SHA256 // http://golang.org/pkg/hash/#Hash
-	branches   int64       = 4
+	hasherfunc        crypto.Hash = crypto.SHA256 // http://golang.org/pkg/hash/#Hash
+	branches          int64       = 4
+	chunkChanCapacity             = 10
 )
 
 var (
@@ -115,10 +116,13 @@ type Chunk struct {
 
 func (self *Chunk) String() string {
 	var size int64
+	var slice []byte
 	if self.Data != nil {
 		size = self.Data.Size()
+		slice = make([]byte, size)
+		self.Data.Read(slice)
 	}
-	return fmt.Sprintf("Key: [%x..] TreeSize: %v Chunksize: %v", self.Key[:4], self.Size, size)
+	return fmt.Sprintf("Key: [%x..] TreeSize: %v Chunksize: %v Data: %x\n", self.Key[:4], self.Size, size, slice)
 }
 
 // The treeChunkers own Hash hashes together
@@ -133,7 +137,7 @@ func (self *TreeChunker) Hash(size int64, input SectionReader) []byte {
 
 func (self *TreeChunker) Split(key Key, data SectionReader) (chunkC chan *Chunk, errC chan error) {
 	wg := &sync.WaitGroup{}
-	chunkC = make(chan *Chunk)
+	chunkC = make(chan *Chunk, chunkChanCapacity)
 	errC = make(chan error)
 	rerrC := make(chan error)
 	timeout := time.After(splitTimeout)
@@ -195,9 +199,6 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR
 
 	switch {
 	case depth == 0:
-		if size > self.chunkSize {
-			// panic("ouch")
-		}
 		// leaf nodes -> content chunks
 		hash = self.Hash(size, data)
 		dpaLogger.Debugf("content chunk: max subtree size: %v, data size: %v", treeSize, size)
@@ -261,7 +262,7 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR
 func (self *TreeChunker) Join(key Key) (data LazySectionReader, chunkC chan *Chunk, errC chan error) {
 	// initialise return parameters
 	errC = make(chan error)
-	chunkC = make(chan *Chunk)
+	chunkC = make(chan *Chunk, chunkChanCapacity)
 	// timer to time out the operation (needed within so as to avoid process leakage)
 	timeout := time.After(joinTimeout)
 	wg := &sync.WaitGroup{}
@@ -270,7 +271,25 @@ func (self *TreeChunker) Join(key Key) (data LazySectionReader, chunkC chan *Chu
 	quitC := make(chan bool)
 
 	// launch lazy recursive call on root chunk
-	data = self.join(0, 0, key, chunkC, rerrC, wg, quitC)()
+	notReadyReader := &NotReadyReader{}
+	reader := &EmbeddedReader{
+		LazySectionReader: &lazyReader{notReadyReader},
+	}
+	fun := self.join(0, 0, key, chunkC, rerrC, wg, quitC)
+	data = reader
+
+	// processing is triggered by reads on the LazySectionReader
+	wg.Add(1)
+	notReadyReader.r = reader
+	notReadyReader.initF = func() {
+		reader.lock.Lock()
+		if fun != nil {
+			reader.LazySectionReader = fun() // replace the reader while caller is waiting
+			wg.Done()                        // just to have one in waitgroup until reader funcs are called
+			fun = nil                        // to be only called once
+		}
+		reader.lock.Unlock()
+	}
 
 	// waits for all the processes to finish and signals by closing internal rerrc
 	go func() {
@@ -336,21 +355,22 @@ func (self *TreeChunker) join(depth int, treeSize int64, key Key, chunkC chan *C
 		var childKey Key
 		var readerF func() (r LazySectionReader)
 		var readerFs [](func() (r LazySectionReader))
-		dpaLogger.DebugDetailf("tree %v", chunk.Size)
 		branches := int64((chunk.Size-1)/treeSize) + 1
+		dpaLogger.DebugDetailf("tree node - size %v, chunk size: %v, subtreeSize %v, branches %v", chunk.Size, chunk.Data.Size(), treeSize, branches)
 
 		// iterate through the chunk containing the keys of children
 		// create lazy init functions that give back readers
 		for i < branches {
 			if chunk.Size-pos < treeSize {
 				secSize = chunk.Size - pos
-				dpaLogger.DebugDetailf("last section %v", secSize)
+				dpaLogger.DebugDetailf("tree node section %v: size %v", i, secSize)
 			} else {
 				secSize = treeSize
 			}
 			// create partial Chunk in order to send a retrieval request
 			childKey = make([]byte, self.hashSize) // preallocate hashSize long slice for key
 			// read the Hash of the subtree from the relevant section of the Chunk into the allocated byte slice in subtree.Key
+			chunk.Data.Seek(0, 0)
 			if _, err := chunk.Data.ReadAt(childKey, i*self.hashSize); err != nil {
 				dpaLogger.DebugDetailf("Read error: %v", err)
 				errC <- err

bzz/chunker_test.go

@@ -109,7 +109,7 @@ func (self *chunkerTester) Split(chunker *TreeChunker, l int) (key Key, input []
 	return
 }
 
-func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) LazySectionReader {
+func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) (LazySectionReader, chan bool) {
 	// reset but not the chunks
 	self.errors = nil
 	self.timeout = false
@@ -124,6 +124,9 @@ func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) Laz
 		for {
 			t.Logf("waiting to mock Chunk Store")
 			select {
+			case <-quitC:
+				break LOOP
+
 			case <-timeout:
 				self.timeout = true
 				break LOOP
@@ -159,17 +162,15 @@ func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) Laz
 				}
 			}
 		}
-		close(quitC)
 	}()
-	<-quitC // waiting for it to finish
-	return reader
+	return reader, quitC
 }
 
 func testRandomData(chunker *TreeChunker, tester *chunkerTester, n int, chunks int, t *testing.T) {
 	key, input := tester.Split(chunker, n)
 	tester.checkChunks(t, chunks)
 	t.Logf("chunks: %v", tester.chunks)
-	reader := tester.Join(t, chunker, key)
+	reader, quitC := tester.Join(t, chunker, key)
 	output := make([]byte, reader.Size())
 	_, err := reader.Read(output)
 	if err != nil {
@@ -179,6 +180,7 @@ func testRandomData(chunker *TreeChunker, tester *chunkerTester, n int, chunks i
 	if bytes.Compare(output, input) != 0 {
 		t.Errorf("input and output mismatch\n IN: %x\nOUT: %x\n", input, output)
 	}
+	close(quitC)
 }
 
 func TestRandomData(t *testing.T) {
@@ -194,4 +196,5 @@ func TestRandomData(t *testing.T) {
 	// testRandomData(chunker, tester, 179, 5, t)
 	// testRandomData(chunker, tester, 253, 7, t)
 	t.Logf("chunks %v", tester.chunks)
+	time.Sleep(2 * time.Second)
 }