diff --git a/bzz/chunkIO.go b/bzz/chunkIO.go index f4580a8957..cac93b5f43 100644 --- a/bzz/chunkIO.go +++ b/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 { diff --git a/bzz/chunker.go b/bzz/chunker.go index 2fded20c32..da7da494eb 100644 --- a/bzz/chunker.go +++ b/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 diff --git a/bzz/chunker_test.go b/bzz/chunker_test.go index e48c6a2c28..31a6dfefa4 100644 --- a/bzz/chunker_test.go +++ b/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) }