rewritten Chunker Join with LazySectionReader, much cleaner IO API, lazy on-demand processing,driven by reader

bzz/chunkIO.go

@@ -4,39 +4,29 @@ import (
 	"bytes"
 	"errors"
 	"io"
+	"time"
 )
 
 type Bounded interface {
 	Size() int64
 }
 
-type Resizeable interface {
-	Bounded
-	Resize(int64) error
-}
-
 type Sliced interface {
-	Slice(int64, int64) []byte
+	Slice(int64, int64) (b []byte, err error)
 }
 
-// Size, Seek, Read, ReadAt and WriteTo
+// Size, Seek, Read, ReadAt
 type SectionReader interface {
 	Bounded
-	Sliced
 	io.Seeker
 	io.Reader
 	io.ReaderAt
-	io.WriterTo
 }
 
-// Size, Seek, Write, WriteAt and ReaderFrom
+type LazySectionReader interface {
-type SectionWriter interface {
+	SectionReader
-	Bounded
+	GetTimeout() time.Time
-	Sliced
+	SetTimeout(time.Time) error
-	io.Seeker
-	io.Writer
-	io.WriterAt
-	io.ReaderFrom
 }
 
 // ChunkReader implements SectionReader on a section
@@ -48,98 +38,45 @@ type ChunkReader struct {
 	limit int64
 }
 
-// ChunkWriter implements SectionWriter on a section
-// of an underlying WriterAt.
-type ChunkWriter struct {
-	w     io.WriterAt
-	base  int64
-	off   int64
-	limit int64
-}
-
-type SectionReadWriter struct {
-	Bounded
-	Sliced
-	io.Seeker
-	io.Reader
-	io.ReaderAt
-	io.WriterTo
-	io.Writer
-	io.WriterAt
-	io.ReaderFrom
-}
-
 // NewChunkReader returns a ChunkReader that reads from r
 // starting at offset off and stops with EOF after n bytes.
 func NewChunkReader(r io.ReaderAt, off int64, n int64) *ChunkReader {
 	return &ChunkReader{r: r, base: off, off: off, limit: off + n}
 }
 
+// ByteSliceReader just extends byte.Reader to make base slice accessible
+type ByteSliceReader struct {
+	*bytes.Reader
+	base []byte
+}
+
+func NewByteSliceReader(b []byte) *ByteSliceReader {
+	return &ByteSliceReader{
+		base:   b,
+		Reader: bytes.NewReader(b),
+	}
+}
+
+// ByteSliceReader implements the Sliced interface
+func (self *ByteSliceReader) Slice(from, to int64) (b []byte, err error) {
+	if from < 0 || to >= int64(self.Len()) {
+		err = io.EOF
+	} else {
+		b = self.base[from:to]
+	}
+	return
+}
+
+// NewChunkReaderFromBytes is a convenience shortcut to get a SectionReader over a byte slice
 func NewChunkReaderFromBytes(b []byte) *ChunkReader {
-	return NewChunkReader(bytes.NewReader(b), 0, int64(len(b)))
+	return NewChunkReader(NewByteSliceReader(b), 0, int64(len(b)))
 }
 
-// NewChunkWriter returns a ChunkWriter that writes to w
+/*
-// starting at offset off and stops with EOF if write would go past off+n
+The following is adapted from io.SectionReader
-func NewChunkWriter(w io.WriterAt, off int64, n int64) *ChunkWriter {
+*/
-	return &ChunkWriter{w: w, base: off, off: off, limit: off + n}
-}
 
-func NewChunkWriterFromBytes(b []byte) *ChunkWriter {
-	return NewChunkWriter(NewByteSliceWriter(b), 0, int64(len(b)))
-}
-
-// the write equivalent of bytes.NewReader(b)
-func NewByteSliceWriter(b []byte) *ByteSliceWriter {
-	return &ByteSliceWriter{b: b, off: 0, limit: int64(len(b))}
-}
-
-type ByteSliceWriter struct {
-	b     []byte
-	off   int64
-	limit int64
-}
-
-func (self *ByteSliceWriter) Slice(from, to int64) (slice []byte) {
-	dpaLogger.DebugDetailf("bottom line %v:%v  (%v-%v)", from, to, self.off, self.limit)
-	if from >= 0 && to <= self.limit {
-		slice = self.b[from:to]
-	}
-	return
-}
-
-func (self *ByteSliceWriter) WriteAt(b []byte, off int64) (n int, err error) {
-	if off < 0 || off >= self.limit {
-		return 0, io.ErrShortWrite
-	}
-	if n = int(self.limit - off); len(b) > n {
-		err = io.ErrShortWrite
-	} else {
-		n = len(b)
-	}
-	copy(self.b[off:], b)
-	return
-}
-
-func (self *ByteSliceWriter) Size() (size int64) {
-	return self.limit
-}
-
-var errUnableToResize = errors.New("unable to resize")
-
-func (self *ByteSliceWriter) Resize(size int64) (err error) {
-	if self.Size() != 0 {
-		err = errUnableToResize
-	} else {
-		self.b = make([]byte, size)
-		self.limit = size
-	}
-	return
-}
-
-// Size returns the size of the section in bytes.
 func (s *ChunkReader) Size() int64 { return s.limit - s.base }
-func (s *ChunkWriter) Size() int64 { return s.limit - s.base }
 
 var errWhence = errors.New("Seek: invalid whence")
 var errOffset = errors.New("Seek: invalid offset")
@@ -162,41 +99,6 @@ func (s *ChunkReader) Seek(offset int64, whence int) (int64, error) {
 	return offset - s.base, nil
 }
 
-func (s *ChunkWriter) Seek(offset int64, whence int) (int64, error) {
-	switch whence {
-	default:
-		return 0, errWhence
-	case 0:
-		offset += s.base
-	case 1:
-		offset += s.off
-	case 2:
-		offset += s.limit
-	}
-	if offset < s.base {
-		return 0, errOffset
-	}
-	s.off = offset
-	return offset - s.base, nil
-}
-
-func (self *ChunkWriter) Resize(size int64) (err error) {
-	err = errUnableToResize
-	if self.Size() >= size {
-		err = nil
-	} else {
-		if self.Size() == 0 {
-			if ws, ok := self.w.(Resizeable); ok {
-				err = ws.Resize(size)
-				if err == nil {
-					self.limit = size
-				}
-			}
-		}
-	}
-	return
-}
-
 func (s *ChunkReader) Read(p []byte) (n int, err error) {
 	if s.off >= s.limit {
 		return 0, io.EOF
@@ -209,37 +111,6 @@ func (s *ChunkReader) Read(p []byte) (n int, err error) {
 	return
 }
 
-func (s *ChunkWriter) Write(p []byte) (n int, err error) {
-	if s.off >= s.limit {
-		return 0, io.EOF
-	}
-	if max := s.limit - s.off; int64(len(p)) > max {
-		p = p[0:max]
-	}
-	n, err = s.w.WriteAt(p, s.off)
-	s.off += int64(n)
-	return
-}
-
-func (s *ChunkReader) Slice(from, to int64) []byte {
-	dpaLogger.DebugDetailf("%v %v %v", s.base, s.off, s.limit)
-	if sl, ok := s.r.(Sliced); ok {
-		dpaLogger.DebugDetailf("%v-%v", s.base+from, s.base+to)
-		return sl.Slice(s.base+from, s.base+to)
-	}
-	return nil
-}
-
-func (s *ChunkWriter) Slice(from, to int64) (b []byte) {
-	dpaLogger.DebugDetailf("base: %v %v %v", s.base, s.off, s.limit)
-	if sl, ok := s.w.(Sliced); ok {
-		dpaLogger.DebugDetailf("%v-%v", s.base+from, s.base+to)
-		b = sl.Slice(s.base+from, s.base+to)
-	}
-	return
-}
-
-//
 func (s *ChunkReader) ReadAt(p []byte, off int64) (n int, err error) {
 	if off < 0 || off >= s.limit-s.base {
 		return 0, io.EOF
@@ -256,43 +127,27 @@ func (s *ChunkReader) ReadAt(p []byte, off int64) (n int, err error) {
 	return s.r.ReadAt(p, off)
 }
 
-//
+// added methods to that ChunkReader implements the Sliced interface
-func (s *ChunkWriter) WriteAt(p []byte, off int64) (n int, err error) {
+func (s *ChunkReader) Slice(from, to int64) (b []byte, err error) {
-	if off < 0 || off >= s.limit-s.base {
+	if from < 0 || to >= s.Size() {
-		return 0, io.EOF
+		err = io.EOF
-	}
-	off += s.base
-	if max := s.limit - off; int64(len(p)) > max {
-		p = p[0:max]
-		n, err = s.w.WriteAt(p, off)
-		return n, err
-	}
-	return s.w.WriteAt(p, off)
-}
-
-func (s *ChunkWriter) ReadFrom(r io.Reader) (n int64, err error) {
-	var m int
-	// if byte slice is available
-	if slice := s.Slice(s.off-s.base, s.limit-s.base); slice != nil {
-		dpaLogger.DebugDetailf("readfrom %v + %v-%v", s.base, s.off-s.base, s.limit-s.base)
-		m, err = r.Read(slice)
-		if err != nil {
-			dpaLogger.Debugf("%v (m%v)", err, m)
-		}
-		dpaLogger.DebugDetailf("read slice %x", slice)
 	} else {
-		b := make([]byte, s.limit-s.off)
+		if sl, ok := s.r.(Sliced); ok {
-		_, err = r.Read(b)
+			b, err = sl.Slice(s.base+from, s.base+to)
-		m, err = s.Write(b)
+		} else {
+			err = errors.New("not sliceable base")
+		}
 	}
-	n = int64(m)
 	return
 }
 
+// added method so that ChunkReader implements the io.WriterTo interface
+// WriteTo method is used by io.Copy
+// This is so that we avoid one extra step of allocation (if the underlying initial Reader implements Sliced
 func (r *ChunkReader) WriteTo(w io.Writer) (n int64, err error) {
 	var b []byte
 	var m int
-	if b := r.Slice(r.off, r.limit); b == nil {
+	if b, _ := r.Slice(r.off-r.base, r.limit-r.base); b == nil {
 		// if slices not available we do it with extra allocation
 		b = make([]byte, r.limit-r.off)
 		m, err = r.Read(b)
@@ -312,3 +167,105 @@ func (r *ChunkReader) WriteTo(w io.Writer) (n int64, err error) {
 	// w
 	return
 }
+
+// LazyChunkReader implements LazySectionReader
+type LazyChunkReader struct {
+	sections []LazySectionReader
+	readerFs [](func() LazySectionReader)
+	size     int64
+	treeSize int64
+	off      int64
+}
+
+func (self *LazyChunkReader) GetTimeout() (t time.Time) {
+	return
+}
+
+func (self *LazyChunkReader) SetTimeout(t time.Time) error {
+	return nil
+}
+
+func (self *LazyChunkReader) Size() (n int64) {
+	return self.size
+}
+
+func (self *LazyChunkReader) Read(b []byte) (read int, err error) {
+	read, err = self.ReadAt(b, self.off)
+	self.off += int64(read)
+	return
+}
+
+func (s *LazyChunkReader) Seek(offset int64, whence int) (int64, error) {
+	switch whence {
+	default:
+		return 0, errWhence
+	case 0:
+		offset += 0
+	case 1:
+		offset += s.off
+	case 2:
+		offset += s.size
+	}
+	if offset < 0 {
+		return 0, errOffset
+	}
+	s.off = offset
+	return offset, nil
+}
+
+func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
+	if off < 0 || off > self.size {
+		err = io.EOF
+		return
+	}
+	want := len(b)
+	got := int(self.size - off)
+	if want > got {
+		err = io.EOF
+	}
+	var index int
+	if off > 0 {
+		index = int((off - 1) / self.treeSize)
+	}
+	off -= int64(index) * self.treeSize
+	var limit int
+	var reader LazySectionReader
+
+	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)
+		} else {
+			limit = int(off) + want
+		}
+		if got, err = reader.ReadAt(b[off:limit], off); err != nil {
+			return
+		}
+		read += got
+		want -= got
+	}
+	return
+}
+
+// just a wrapper to make a SectionReader conform to LazySectionReader
+// with dummy implementation
+type lazyReader struct {
+	SectionReader
+}
+
+func LazyReader(r SectionReader) (l LazySectionReader) {
+	return &lazyReader{
+		SectionReader: r,
+	}
+}
+
+func (self *lazyReader) GetTimeout() (t time.Time) {
+	return
+}
+
+func (self *lazyReader) SetTimeout(t time.Time) error {
+	return nil
+}

bzz/chunker.go

@@ -19,6 +19,7 @@ import (
 	"crypto"
 	"encoding/binary"
 	"fmt"
+	"io"
 	"sync"
 	"time"
 )
@@ -56,21 +57,21 @@ type Chunker interface {
 	Split(key Key, data SectionReader) (chan *Chunk, chan error)
 	/*
 		Join reconstructs original content based on a root key
-		When joining, data is given as a SectionWriter preallocation for which is taken care of by the caller.
+		When joining, the caller gets returned a LazySectionReader , a chunk channel and an error channel.
-		If the size of the SectionWriter is found 0 Chunker will resize it once the entire data size is known from the root chunk. If resize is not supported by the writer, an error is given.
+		New chunks to retrieve are coming to caller via the Chunk channel.
-		Any other size value is checked and if it does not fit the actual datasize, an error will be reported.
+		If an error is encountered during joining, it is fed to errC error channel.
-		New chunks to retrieve are coming to caller via the Chunk channel (first return parameter)
-		If an error is encountered during joining, it is fed to errC error channel (second return parameter)
 		A closed error and chunk channel signal process completion at which point the data can be considered final if there were no errors.
+		The LazySectionReader provides on-demand fetching of content including chunks.
+		Lifecycle of the reader can be modified with SetTimeout()
 	*/
-	Join(key Key, data SectionWriter) (chan *Chunk, chan error)
+	Join(key Key) (LazySectionReader, chan *Chunk, chan error)
 }
 
 /*
 Tree chunker is a concrete implementation of data chunking.
 This chunker works in a simple way, it builds a tree out of the document so that each node either represents a chunk of real data or a chunk of data representing an branching non-leaf node of the tree. In particular each such non-leaf chunk will represent is a concatenation of the hash of its respective children. This scheme simultaneously guarantees data integrity as well as self addressing. Abstract nodes are transparent since their represented size component is strictly greater than their maximum data size, since they encode a subtree.
 
-If all is well it is possible to implement this by simply composing readers and writers so that no extra allocation or buffering is necessary for the data splitting. This means that in principle there can be direct IO between : memory, file system, network socket (bzz peers storage request is read from the socket ). In practice there may be need for several stages of internal buffering.
+If all is well it is possible to implement this by simply composing readers so that no extra allocation or buffering is necessary for the data splitting and joining. This means that in principle there can be direct IO between : memory, file system, network socket (bzz peers storage request is read from the socket ). In practice there may be need for several stages of internal buffering.
 Unfortunately the hashing itself does use extra copies and allocatetion though since it does need it.
 */
 type TreeChunker struct {
@@ -126,7 +127,7 @@ func (self *Chunk) String() string {
 func (self *TreeChunker) Hash(size int64, input SectionReader) []byte {
 	hasher := self.HashFunc.New()
 	binary.Write(hasher, binary.LittleEndian, size)
-	input.WriteTo(hasher) // SectionReader implements io.WriterTo
+	io.Copy(hasher, input) // it uses WriteTo if available, ChunkReader implements io.WriterTo
 	return hasher.Sum(nil)
 }
 
@@ -257,7 +258,7 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR
 
 }
 
-func (self *TreeChunker) Join(key Key, data SectionWriter) (chunkC chan *Chunk, errC chan error) {
+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)
@@ -268,53 +269,8 @@ func (self *TreeChunker) Join(key Key, data SectionWriter) (chunkC chan *Chunk,
 	rerrC := make(chan error)
 	quitC := make(chan bool)
 
-	wg.Add(1)
+	// launch lazy recursive call on root chunk
-	go func() {
+	data = self.join(0, 0, key, chunkC, rerrC, wg, quitC)()
-		// create the 'chunk' for root chunk of the data tree
-		chunk := &Chunk{
-			Key: key,
-			C:   make(chan bool, 1),
-		}
-		// request data
-		dpaLogger.Debugf("request root chunk for key %x", key[:4])
-		chunkC <- chunk
-		// wait for reponse, if no root, we cannot go on
-
-		select {
-		case <-chunk.C: // bells ringing data delivered
-			dpaLogger.Debugf("request root chunk data has come, size %v", chunk.Size)
-		case <-timeout:
-			err := fmt.Errorf("join time out waiting for root")
-			rerrC <- err
-			return
-		}
-
-		if data.Size() < chunk.Size {
-			dpaLogger.Debugf("trying to resize writer to size %v for join data", chunk.Size)
-
-			var err error
-			if resizeable, ok := data.(Resizeable); ok {
-				err = resizeable.Resize(chunk.Size)
-			} else {
-				err = fmt.Errorf("writer does not support resizing and has insufficient size %v (need %v)", data.Size(), chunk.Size)
-			}
-			if err != nil {
-				dpaLogger.Debugf("%v", err)
-				rerrC <- err
-				wg.Done()
-				return
-			}
-		}
-		// calculate depth and max treeSize
-		var depth int
-		var treeSize int64 = self.chunkSize
-
-		for ; treeSize < chunk.Size; treeSize *= self.Branches {
-			depth++
-		}
-		// launch recursive call on root chunk
-		self.join(depth, treeSize/self.Branches, chunk, data, chunkC, rerrC, wg, quitC)
-	}()
 
 	// waits for all the processes to finish and signals by closing internal rerrc
 	go func() {
@@ -340,57 +296,80 @@ func (self *TreeChunker) Join(key Key, data SectionWriter) (chunkC chan *Chunk,
 	return
 }
 
-func (self *TreeChunker) join(depth int, treeSize int64, chunk *Chunk, data SectionWriter, chunkC chan *Chunk, errC chan error, wg *sync.WaitGroup, quitC chan bool) {
+func (self *TreeChunker) join(depth int, treeSize int64, key Key, chunkC chan *Chunk, errC chan error, wg *sync.WaitGroup, quitC chan bool) (readerF func() LazySectionReader) {
-
+	wg.Add(1)
-	defer wg.Done()
+	readerF = func() (r LazySectionReader) {
-	select {
+		defer wg.Done()
-	case <-quitC:
+		chunk := &Chunk{
-	case <-chunk.C: // bells are ringing, data have been delivered
+			Key: key,
-		switch {
+			C:   make(chan bool, 1), // close channel to signal data delivery
-		case depth == 0:
-			if _, err := data.ReadFrom(chunk.Data); err != nil {
-				errC <- err
-			}
-
-		case chunk.Size < treeSize:
-			// this must be a last item on its level
-			wg.Add(1)
-			self.join(depth-1, treeSize/self.Branches, chunk, data, chunkC, errC, wg, quitC)
-
-		default:
-			// intermediate chunk, chunk containing hashes of child nodes
-			var pos, i int64
-			var secSize int64
-			dpaLogger.DebugDetailf("tree %v", chunk.Size)
-			branches := int64((chunk.Size-1)/treeSize) + 1
-			for i < branches {
-				if chunk.Size-pos < treeSize {
-					secSize = chunk.Size - pos
-					dpaLogger.DebugDetailf("last section %v", secSize)
-				} else {
-					secSize = treeSize
-				}
-				// create partial Chunk in order to send a retrieval request
-				subtree := &Chunk{
-					Key: make([]byte, self.hashSize), // preallocate hashSize long slice for key
-					C:   make(chan bool, 1),          // close channel to signal data delivery
-				}
-				// read the Hash of the subtree from the relevant section of the Chunk into the allocated byte slice in subtree.Key
-				if _, err := chunk.Data.ReadAt(subtree.Key, i*self.hashSize); err != nil {
-					dpaLogger.DebugDetailf("Read error: %v", err)
-					errC <- err
-					break
-				}
-				// call recursively on the subtree
-				subTreeData := NewChunkWriter(data, pos, secSize)
-				wg.Add(1)
-				go self.join(depth-1, treeSize/self.Branches, subtree, subTreeData, chunkC, errC, wg, quitC)
-				// submit request
-				chunkC <- subtree
-				i++
-				pos += treeSize
-			}
-
 		}
+		chunkC <- chunk // submit retrieval request, someone should be listening on the other side (or we will time out globally)
+
+		// waiting for the chunk retrieval
+		select {
+		case <-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
+		}
+
+		// calculate depth and max treeSize
+		var depth int
+		var treeSize int64 = self.hashSize
+		for ; treeSize*self.Branches < chunk.Size; treeSize *= self.Branches {
+			depth++
+		}
+
+		if depth == 0 {
+			return LazyReader(chunk.Data) // simply give back the chunks reader for content chunks
+		}
+
+		// find appropriate block level
+		for chunk.Size < treeSize {
+			treeSize /= self.Branches
+			depth--
+		}
+		// boooo
+		// intermediate chunk, chunk containing hashes of child nodes
+		var pos, i, secSize int64
+		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
+
+		// 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)
+			} 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
+			if _, err := chunk.Data.ReadAt(childKey, i*self.hashSize); err != nil {
+				dpaLogger.DebugDetailf("Read error: %v", err)
+				errC <- err
+				break
+			}
+			// call lazy reader function recursively on the subtree
+			readerF = self.join(depth-1, treeSize/self.Branches, childKey, chunkC, errC, wg, quitC)
+			readerFs = append(readerFs, readerF)
+			i++
+			pos += treeSize
+		}
+		// new reader created on demand:
+		r = &LazyChunkReader{
+			readerFs: readerFs,
+			sections: make([]LazySectionReader, branches),
+			size:     chunk.Size,
+			treeSize: treeSize,
+		}
+		return
 	}
+	return
 }

bzz/chunker_test.go

@@ -109,13 +109,13 @@ func (self *chunkerTester) Split(chunker *TreeChunker, l int) (key Key, input []
 	return
 }
 
-func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) (data []byte) {
+func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) LazySectionReader {
 	// reset but not the chunks
 	self.errors = nil
 	self.timeout = false
 
-	w := NewChunkWriterFromBytes(nil)
+	reader, chunkC, errC := chunker.Join(key)
-	chunkC, errC := chunker.Join(key, w)
+
 	quitC := make(chan bool)
 	timeout := time.After(60 * time.Second)
 
@@ -162,15 +162,19 @@ func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) (da
 		close(quitC)
 	}()
 	<-quitC // waiting for it to finish
-	w.Seek(0, 0)
+	return reader
-	return w.Slice(0, w.Size())
 }
 
 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)
-	output := tester.Join(t, chunker, key)
+	reader := tester.Join(t, chunker, key)
+	output := make([]byte, reader.Size())
+	_, err := reader.Read(output)
+	if err != nil {
+		t.Errorf("read error %v\n", err)
+	}
 	t.Logf(" IN: %x\nOUT: %x\n", input, output)
 	if bytes.Compare(output, input) != 0 {
 		t.Errorf("input and output mismatch\n IN: %x\nOUT: %x\n", input, output)

bzz/dpa.go

@@ -44,7 +44,7 @@ BytesToReader(data []byte) (SectionReader, error)
 // 	return NewChunkReaderFromBytes(rlp.Encode(data)), nil
 // }
 
-func (self *DPA) Retrieve(key Key, data SectionReadWriter) (err error) {
+func (self *DPA) Retrieve(key Key) (data LazySectionReader, err error) {
 	joinC := make(chan bool)
 	reqsC := make(chan bool)
 	var requests int
@@ -61,11 +61,12 @@ func (self *DPA) Retrieve(key Key, data SectionReadWriter) (err error) {
 	// a SectionReadWriter based on a byte slice equal to the stored object image's bytes
 
 	dpaLogger.Debugf("bzz honey retrieve")
+	reader, chunkC, errC := self.Chunker.Join(key)
+	data = reader
 
 	go func() {
 
 		var ok bool
-		chunkC, errC := self.Chunker.Join(key, data)
 
 	LOOP:
 		for {