clean up DPA, takes Chunker, ChunkStore

bzz/dpa.go

@@ -12,10 +12,19 @@ import (
 /*
 DPA provides the client API entrypoints Store and Retrieve to store and retrieve
 It can store anything that has a byte slice representation, so files or serialised objects etc.
+
 Storage: DPA calls the Chunker to segment the input datastream of any size to a merkle hashed tree of chunks. The key of the root block is returned to the client.
+
 Retrieval: given the key of the root block, the DPA retrieves the block chunks and reconstructs the original data and passes it back as a lazy reader. A lazy reader is a reader with on-demand delayed processing, i.e. the chunks needed to reconstruct a large file are only fetched and processed if that particular part of the document is actually read.
 
-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.
+As the chunker produces chunks, DPA dispatches them to the chunk store for storage or retrieval.
+
+The ChunkStore interface is implemented by :
+
+- memStore: a memory cache
+- dbStore: local disk/db store
+- localStore: a combination (sequence of) memStoe and dbStoe
+- netStore: dht storage
 */
 
 const (
@@ -30,10 +39,10 @@ var (
 var dpaLogger = ethlogger.NewLogger("BZZ")
 
 type DPA struct {
-	Chunker   Chunker
+	Chunker    Chunker
-	Stores    []ChunkStore
+	ChunkStore ChunkStore
-	storeC    chan *Chunk
+	storeC     chan *Chunk
-	retrieveC chan *Chunk
+	retrieveC  chan *Chunk
 
 	lock    sync.Mutex
 	running bool
@@ -61,7 +70,7 @@ type ChunkStore interface {
 }
 
 func (self *DPA) Retrieve(key Key) (data LazySectionReader, err error) {
-	dpaLogger.Debugf("bzz honey retrieve")
+
 	reader, errC := self.Chunker.Join(key, self.retrieveC)
 	data = reader
 	// we can add subscriptions etc. or timeout here
@@ -87,8 +96,6 @@ func (self *DPA) Retrieve(key Key) (data LazySectionReader, err error) {
 
 func (self *DPA) Store(data SectionReader) (key Key, err error) {
 
-	dpaLogger.Debugf("bzz honey store")
-
 	errC := self.Chunker.Split(key, data, self.storeC)
 
 	go func() {
@@ -136,31 +143,25 @@ func (self *DPA) retrieveLoop() {
 	self.retrieveC = make(chan *Chunk, retrieveChanCapacity)
 
 	go func() {
-	LOOP:
+	RETRIEVE:
 		for chunk := range self.retrieveC {
 			go func() {
-				for i, store := range self.Stores {
+				storedChunk, err := self.ChunkStore.Get(chunk.Key)
-					storedChunk, err := store.Get(chunk.Key)
+				if err == notFound {
-					if err == notFound {
+					dpaLogger.DebugDetailf("chunk %x not found", chunk.Key)
-						dpaLogger.DebugDetailf("%v retrieving chunk %x: NOT FOUND", store, chunk.Key)
+					return
-						return
-					}
-					if err != nil {
-						dpaLogger.DebugDetailf("%v retrieving chunk %x: %v", store, chunk.Key, err)
-						return
-					}
-					chunk.Reader = NewChunkReaderFromBytes(storedChunk.Data)
-					chunk.Size = storedChunk.Size
-					close(chunk.C)
-					// if not in cache, cache it in memstore
-					if i > 0 {
-						self.Stores[0].Put(chunk)
-					}
 				}
+				if err != nil {
+					dpaLogger.DebugDetailf("error retrieving chunk %x: %v", chunk.Key, err)
+					return
+				}
+				chunk.Reader = NewChunkReaderFromBytes(storedChunk.Data)
+				chunk.Size = storedChunk.Size
+				close(chunk.C)
 			}()
 			select {
 			case <-self.quitC:
-				break LOOP
+				break RETRIEVE
 			default:
 			}
 		}
@@ -170,17 +171,12 @@ func (self *DPA) retrieveLoop() {
 func (self *DPA) storeLoop() {
 	self.storeC = make(chan *Chunk)
 	go func() {
-	LOOP:
+	STORE:
 		for chunk := range self.storeC {
-			go func() {
+			go self.ChunkStore.Put(chunk)
-				for _, store := range self.Stores {
-					store.Put(chunk)
-					// no waiting/blocking here
-				}
-			}()
 			select {
 			case <-self.quitC:
-				break LOOP
+				break STORE
 			default:
 			}
 		}

bzz/dpa_test.go

@@ -2,84 +2,4 @@
 
 package bzz
 
-import (
+import ()
-//	"fmt"
-// "math"
-// "math/rand"
-// "testing"
-)
-
-// func cmptest(a, b []byte) bool {
-
-// 	if len(a) != len(b) {
-// 		return false
-// 	}
-// 	for i := range a {
-// 		if a[i] != b[i] {
-// 			return false
-// 		}
-// 	}
-// 	return true
-// }
-
-// const testcnt = 10
-
-// func testlen(i int) int {
-
-// 	/*	if i == 79 {
-// 		return 16777217
-// 	}*/
-
-// 	return int(0.5 + math.Exp2(3.0+float64(i)/5))
-// }
-
-// func TestBlockHashStorage(t *testing.T) {
-// 	t.Logf("Creating DBStorage...")
-
-// 	dbstore := new(dpaDBStorage)
-// 	dbstore.Init(nil)
-// 	go dbstore.Run()
-
-// 	t.Logf("Creating MemStorage...")
-
-// 	memstore := new(dpaMemStorage)
-// 	memstore.Init(&dbstore.dpaStorage)
-// 	go memstore.Run()
-
-// 	t.Logf("Storing test vectors...")
-
-// 	test := make([][]byte, testcnt)
-// 	hash := make([]HashType, testcnt)
-// 	for i := 0; i < testcnt; i++ {
-// 		test[i] = maketest(testlen(i))
-// 		//t.Logf("Test[%d] = %x", i, test[i])
-// 		hash[i] = GetDPAroot(test[i], &memstore.dpaStorage)
-// 		//t.Logf("Hash[%d] = %x", i, hash[i])
-// 	}
-
-// 	t.Logf("Retrieving test vectors...")
-
-// 	rnd := rand.New(rand.NewSource(0))
-
-// 	for i := 0; i < testcnt; i++ {
-
-// 		tt := GetDPAdata(hash[i], &memstore.dpaStorage) // get the whole vector with byte array wrapper
-
-// 		sr := GetDPAreader(hash[i], &memstore.dpaStorage)
-// 		size := int(sr.Size())
-// 		pos := rnd.Intn(size - 1)
-// 		slen := rnd.Intn(size-1-pos) + 1
-// 		sr.Seek(int64(pos), 0)
-// 		br, _ := sr.Read(tt[pos : pos+slen]) // re-read a random section
-
-// 		if (br == slen) && cmptest(test[i], tt) {
-// 			t.Logf("Test case %d passed (test vector length %d)", i, len(tt))
-// 		} else {
-// 			t.Errorf("Test case %d failed", i)
-// 			if size < 20 {
-// 				t.Errorf("pos = %d  slen = %d  br = %d  vector = %x instead of %x", pos, slen, br, tt, test[i])
-// 			}
-// 		}
-// 	}
-
-// }