chunkC is now an input argument since DPA typically has one common for storage and retrieval. improved documentation, adapted DPA

bzz/chunker.go

@@ -27,7 +27,6 @@ import (
 const (
 	hasherfunc crypto.Hash = crypto.SHA256 // http://golang.org/pkg/hash/#Hash
 	branches   int64       = 4
-	chunkChanCapacity             = 10
 )
 
 var (
@@ -51,21 +50,21 @@ When calling Join with a root key, the data can be nil ponter in which case it w
 type Chunker interface {
 	/*
 	 	When splitting, data is given as a SectionReader, and the key is a hashSize long byte slice (Key), the root hash of the entire content will fill this once processing finishes.
-	 	New chunks to store are coming to caller via the chunk channel (first return parameter)
+	 	New chunks to store are coming to caller via the chunk storage channel, which the caller provides.
-	 	If an error is encountered during splitting, it is fed to errC error channel (second return parameter)
+	 	The caller gets returned an error channel, if an error is encountered during splitting, it is fed to errC error channel.
-	   A closed error and chunk channel signal process completion at which point the key can be considered final if there were no errors.
+	   A closed error signals process completion at which point the key can be considered final if there were no errors.
 	*/
-	Split(key Key, data SectionReader) (chan *Chunk, chan error)
+	Split(key Key, data SectionReader, chunkC chan *Chunk) chan error
 	/*
-		Join reconstructs original content based on a root key
+		Join reconstructs original content based on a root key.
-		When joining, the caller gets returned a LazySectionReader , a chunk channel and an error channel.
+		When joining, the caller gets returned a LazySectionReader and an error channel.
-		New chunks to retrieve are coming to caller via the Chunk channel.
+		New chunks to retrieve are coming to caller via the Chunk channel, which the caller provides.
 		If an error is encountered during joining, it is fed to errC error channel.
-		A closed error and chunk channel signal process completion at which point the data can be considered final if there were no errors.
+		A closed error signals process completion at which point the data can be considered final and fully reconstructed 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) (LazySectionReader, chan *Chunk, chan error)
+	Join(key Key, chunkC chan *Chunk) (LazySectionReader, chan error)
 }
 
 /*
@@ -73,8 +72,9 @@ 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 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.
+Unfortunately the hashing itself does use extra copies and allocation though since it does need it.
 */
+
 type TreeChunker struct {
 	Branches     int64
 	HashFunc     crypto.Hash
@@ -135,9 +135,8 @@ func (self *TreeChunker) Hash(size int64, input SectionReader) []byte {
 	return hasher.Sum(nil)
 }
 
-func (self *TreeChunker) Split(key Key, data SectionReader) (chunkC chan *Chunk, errC chan error) {
+func (self *TreeChunker) Split(key Key, data SectionReader, chunkC chan *Chunk) (errC chan error) {
 	wg := &sync.WaitGroup{}
-	chunkC = make(chan *Chunk, chunkChanCapacity)
 	errC = make(chan error)
 	rerrC := make(chan error)
 	timeout := time.After(splitTimeout)
@@ -181,7 +180,6 @@ func (self *TreeChunker) Split(key Key, data SectionReader) (chunkC chan *Chunk,
 		case <-timeout:
 			errC <- fmt.Errorf("split time out")
 		}
-		close(chunkC)
 		close(errC)
 	}()
 
@@ -259,10 +257,9 @@ 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) {
+func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) (data LazySectionReader, errC chan error) {
 	// initialise return parameters
 	errC = make(chan error)
-	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{}

bzz/chunker_test.go

@@ -74,7 +74,8 @@ func (self *chunkerTester) Split(chunker *TreeChunker, l int) (key Key, input []
 	data, slice := testDataReader(l)
 	input = slice
 	key = make([]byte, 32)
-	chunkC, errC := chunker.Split(key, data)
+	chunkC := make(chan *Chunk)
+	errC := chunker.Split(key, data, chunkC)
 	quitC := make(chan bool)
 	timeout := time.After(60 * time.Second)
 
@@ -113,8 +114,9 @@ func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) (La
 	// reset but not the chunks
 	self.errors = nil
 	self.timeout = false
+	chunkC := make(chan *Chunk)
 
-	reader, chunkC, errC := chunker.Join(key)
+	reader, errC := chunker.Join(key, chunkC)
 
 	quitC := make(chan bool)
 	timeout := time.After(60 * time.Second)

bzz/dpa.go

@@ -17,6 +17,11 @@ Retrieval: given the key of the root block, the DPA retrieves the block chunks a
 As the chunker produces chunks, DPA dispatches them to the chunk stores for storage or retrieval. The chunk stores are typically sequenced as memory cache, local disk/db store, cloud/distributed/dht storage. Storage requests will reach to all 3 components while retrieval requests stop after the first successful retrieval.
 */
 
+const (
+	storeChanCapacity    = 100
+	retrieveChanCapacity = 100
+)
+
 var dpaLogger = ethlogger.NewLogger("BZZ")
 
 type DPA struct {
@@ -24,6 +29,8 @@ type DPA struct {
 	Stores    []ChunkStore
 	wg        sync.WaitGroup
 	quitC     chan bool
+	storeC    chan *Chunk
+	retrieveC chan *Chunk
 }
 
 type ChunkStore interface {
@@ -44,105 +51,91 @@ BytesToReader(data []byte) (SectionReader, error)
 // 	return NewChunkReaderFromBytes(rlp.Encode(data)), nil
 // }
 
-func (self *DPA) Retrieve(key Key) (data LazySectionReader, err error) {
+func (self *DPA) retrieveLoop() {
-	joinC := make(chan bool)
+	self.retrieveC = make(chan *Chunk, retrieveChanCapacity)
-	reqsC := make(chan bool)
-	var requests int
 
-	self.wg.Add(1)
-	chunkWg := sync.WaitGroup{}
-
-	chunkWg.Add(1)
 	go func() {
-		chunkWg.Wait() // wait for all chunk retrieval requests to process
+	LOOP:
-		close(reqsC)   // signal to channel
+		for chunk := range self.retrieveC {
+			for _, store := range self.Stores {
+				if err := store.Get(chunk); err != nil { // no waiting/blocking here
+					dpaLogger.DebugDetailf("%v retrieving chunk %x: %v", store, chunk.Key, err)
+				}
+			}
+			select {
+			case <-self.quitC:
+				break LOOP
+			default:
+			}
+		}
 	}()
-	// r is potentially nil pointer, by default chunker allocates
+}
-	// a SectionReadWriter based on a byte slice equal to the stored object image's bytes
 
+func (self *DPA) Retrieve(key Key) (data LazySectionReader, err error) {
 	dpaLogger.Debugf("bzz honey retrieve")
-	reader, chunkC, errC := self.Chunker.Join(key)
+	reader, errC := self.Chunker.Join(key, self.retrieveC)
 	data = reader
-
+	// we can add subscriptions etc. or timeout here
 	go func() {
-
-		var ok bool
-
 	LOOP:
 		for {
 			select {
-
+			case err, ok := <-errC:
-			case chunk, ok := <-chunkC:
+				if err != nil {
-				if chunk != nil { // game over
-					chunk.wg = chunkWg
-					chunkWg.Add(1) // need to call Done by any storage that first retrieves the data
-					for _, store := range self.Stores {
-						requests++
-						if err = store.Get(chunk); err != nil { // no waiting/blocking here
-							dpaLogger.DebugDetailf("%v retrieved chunk %x", store, chunk.Key)
-							break // the inner loop
-						}
-					}
-				}
-				if !ok { // game over but need to continue to see errc still
-					chunkC = nil // make it block so no infinite loop
-					chunkWg.Done()
-				}
-
-			case err, ok = <-errC:
 					dpaLogger.Warnf("%v", err)
+				}
 				if !ok {
 					break LOOP
 				}
-				dpaLogger.DebugDetailf("%v", err)
-
-			case <-reqsC:
-				dpaLogger.DebugDetailf("processed all %v chunk retrieval requests for root key %x", requests, key)
-
 			case <-self.quitC:
-				break LOOP
+				return
 			}
 		}
-		close(joinC)
-		self.wg.Done()
 	}()
 
-	<-joinC
-
 	return
 }
 
+func (self *DPA) storeLoop() {
+	self.storeC = make(chan *Chunk)
+	go func() {
+	LOOP:
+		for chunk := range self.storeC {
+			for _, store := range self.Stores {
+				if err := store.Put(chunk); err != nil { // no waiting/blocking here
+					dpaLogger.DebugDetailf("%v storing chunk %x: %v", store, chunk.Key, err)
+				} // no waiting/blocking here
+			}
+			select {
+			case <-self.quitC:
+				break LOOP
+			default:
+			}
+		}
+	}()
+}
+
 func (self *DPA) Store(data SectionReader) (key Key, err error) {
 
 	dpaLogger.Debugf("bzz honey store")
 
-	chunkC, errC := self.Chunker.Split(key, data)
+	errC := self.Chunker.Split(key, data, self.storeC)
 
-LOOP:
+	go func() {
+	LOOP:
 		for {
 			select {
-
-		case chunk, ok := <-chunkC:
-			if chunk != nil {
-				for _, store := range self.Stores {
-					store.Put(chunk) // no waiting/blocking here
-				}
-			}
-			if !ok { // game over but need to continue to see errc still
-				chunkC = nil // make it block so no infinite loop
-			}
-
 			case err, ok := <-errC:
 				dpaLogger.Warnf("%v", err)
 				if !ok {
 					break LOOP
 				}
-			dpaLogger.DebugDetailf("%v", err)
 
 			case <-self.quitC:
 				break LOOP
 			}
 		}
+	}()
 	return
 
 }