dbStore + test

- memStore now take dbStore and calls updateAccessCnt directly
- remove update field from Chunk

bzz/dbstore.go

@@ -7,7 +7,6 @@ import (
 	//	"crypto/sha256"
 	"bytes"
 	"encoding/binary"
-	"fmt"
 	"github.com/ethereum/go-ethereum/ethdb"
 	//	"github.com/ethereum/go-ethereum/ethutil"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -35,8 +34,7 @@ type gcItem struct {
 	idxKey []byte
 }
 
-type dpaDBStorage struct {
+type dbStore struct {
-	dpaStorage
 	db *ethdb.LDBDatabase
 
 	// this should be stored in db, accessed transactionally
@@ -74,14 +72,15 @@ func getIndexGCValue(index *dpaDBIndex) uint64 {
 
 }
 
-func (s *dpaDBStorage) updateIndexAccess(index *dpaDBIndex) {
+func (s *dbStore) updateIndexAccess(index *dpaDBIndex) {
 
 	index.Access = s.accessCnt
 
 }
 
-func getIndexKey(hash HashType) []byte {
+func getIndexKey(hash Key) []byte {
 
+	HashSize := len(hash)
 	key := make([]byte, HashSize+1)
 	key[0] = 0
 	// db keys derived from hash:
@@ -108,15 +107,15 @@ func encodeIndex(index *dpaDBIndex) []byte {
 
 }
 
-func encodeData(entry *dpaNode) []byte {
+func encodeData(chunk *Chunk) []byte {
 
 	var rlpEntry struct {
 		Data []byte
 		Size uint64
 	}
 
-	rlpEntry.Data = entry.data
+	rlpEntry.Data = chunk.Data
-	rlpEntry.Size = uint64(entry.size)
+	rlpEntry.Size = uint64(chunk.Size)
 
 	data, _ := rlp.EncodeToBytes(rlpEntry)
 	return data
@@ -130,7 +129,7 @@ func decodeIndex(data []byte, index *dpaDBIndex) {
 
 }
 
-func decodeData(data []byte, entry *dpaNode) {
+func decodeData(data []byte, chunk *Chunk) {
 
 	var rlpEntry struct {
 		Data []byte
@@ -138,9 +137,12 @@ func decodeData(data []byte, entry *dpaNode) {
 	}
 
 	dec := rlp.NewStream(bytes.NewReader(data))
-	dec.Decode(&rlpEntry)
+	err := dec.Decode(&rlpEntry)
-	entry.data = rlpEntry.Data
+	if err != nil {
-	entry.size = int64(rlpEntry.Size)
+		panic(err.Error())
+	}
+	chunk.Data = rlpEntry.Data
+	chunk.Size = int64(rlpEntry.Size)
 }
 
 func gcListPartition(list []*gcItem, left int, right int, pivotIndex int) int {
@@ -180,7 +182,7 @@ func gcListSelect(list []*gcItem, left int, right int, n int) int {
 	}
 }
 
-func (s *dpaDBStorage) collectGarbage() {
+func (s *dbStore) collectGarbage() {
 
 	it := s.db.NewIterator()
 	it.Seek(s.gcPos)
@@ -246,16 +248,16 @@ func (s *dpaDBStorage) collectGarbage() {
 
 }
 
-func (s *dpaDBStorage) add(entry *dpaStoreReq) {
+func (s *dbStore) Put(chunk *Chunk) {
 
-	ikey := getIndexKey(entry.hash)
+	ikey := getIndexKey(chunk.Key)
 	var index dpaDBIndex
 
 	if s.tryAccessIdx(ikey, &index) {
 		return // already exists, only update access
 	}
 
-	data := encodeData(&entry.dpaNode)
+	data := encodeData(chunk)
 	//data := ethutil.Encode([]interface{}{entry})
 
 	if s.entryCnt >= dbMaxEntries {
@@ -284,7 +286,7 @@ func (s *dpaDBStorage) add(entry *dpaStoreReq) {
 }
 
 // try to find index; if found, update access cnt and return true
-func (s *dpaDBStorage) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
+func (s *dbStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
 
 	idata, err := s.db.Get(ikey)
 	if err != nil {
@@ -305,74 +307,46 @@ func (s *dpaDBStorage) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
 	return true
 }
 
-func (s *dpaDBStorage) find(hash HashType) (entry dpaNode) {
+func (s *dbStore) Get(key Key) (chunk *Chunk, err error) {
 
-	key := getIndexKey(hash)
 	var index dpaDBIndex
 
-	if s.tryAccessIdx(key, &index) {
+	if s.tryAccessIdx(getIndexKey(key), &index) {
-		data, _ := s.db.Get(getDataKey(index.Idx))
+		var data []byte
-		decodeData(data, &entry)
+		data, err = s.db.Get(getDataKey(index.Idx))
+		if err != nil {
+			return
+		}
+		chunk = &Chunk{
+			Key: key,
+		}
+		decodeData(data, chunk)
+	} else {
+		err = notFound
 	}
 
 	return
 
 }
 
-func (s *dpaDBStorage) process_store(req *dpaStoreReq) {
+func (s *dbStore) updateAccessCnt(key Key) {
 
-	s.add(req)
+	var index dpaDBIndex
-
+	s.tryAccessIdx(getIndexKey(key), &index) // result_chn == nil, only update access cnt
-	if s.chain != nil {
-		s.chain.store_chn <- req
-	}
 
 }
 
-func (s *dpaDBStorage) process_retrieve(req *dpaRetrieveReq) {
+func newDbStore(path string) (s *dbStore, err error) {
 
-	if req.result_chn == nil {
+	s = new(dbStore)
-
-		key := getIndexKey(req.hash)
-		var index dpaDBIndex
-		s.tryAccessIdx(key, &index) // result_chn == nil, only update access cnt
 
+	s.db, err = ethdb.NewLDBDatabase(path)
+	if err != nil {
 		return
 	}
 
-	entry := s.find(req.hash)
+	s.gcStartPos = make([]byte, 1)
-
+	s.gcStartPos[0] = kpIndex
-	if entry.data == nil {
-		if s.chain != nil {
-			s.chain.retrieve_chn <- req
-			return
-		}
-	}
-
-	res := new(dpaRetrieveRes)
-	if entry.data != nil {
-		res.dpaNode = entry
-	}
-	res.req_id = req.req_id
-	req.result_chn <- res
-
-}
-
-func (s *dpaDBStorage) Init(ch *dpaStorage) {
-
-	s.dpaStorage.Init()
-
-	var err error
-	s.db, err = ethdb.NewLDBDatabase("/tmp/bzz")
-	if err != nil {
-		fmt.Println("/tmp/bzz error:")
-		fmt.Println(err)
-	}
-	if s.db == nil {
-		fmt.Println("LDBDatabase is nil")
-	}
-
-	s.gcStartPos = make([]byte, HashSize+1)
 	s.gcArray = make([]*gcItem, gcArraySize)
 
 	data, _ := s.db.Get(keyEntryCnt)
@@ -385,31 +359,13 @@ func (s *dpaDBStorage) Init(ch *dpaStorage) {
 	if s.gcPos == nil {
 		s.gcPos = s.gcStartPos
 	}
-
+	return
 	//	fmt.Println(s.entryCnt)
 	//	fmt.Println(s.accessCnt)
 	//	fmt.Println(s.dataIdx)
 
 }
 
-func (s *dpaDBStorage) Run() {
+func (s *dbStore) close() {
-
+	s.db.Close()
-	for {
-		bb := true
-		for bb {
-			select {
-			case store := <-s.store_chn:
-				s.process_store(store)
-			default:
-				bb = false
-			}
-		}
-		select {
-		case store := <-s.store_chn:
-			s.process_store(store)
-		case retrv := <-s.retrieve_chn:
-			s.process_retrieve(retrv)
-		}
-	}
-
 }

bzz/dbstore_test.go

@@ -0,0 +1,114 @@
+package bzz
+
+import (
+	"os"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/bzz/test"
+)
+
+func testDbStore(l int64, branches int64, t *testing.T) {
+
+	os.RemoveAll("/tmp/bzz")
+	m, err := newDbStore("/tmp/bzz")
+	if err != nil {
+		panic("no dbStore")
+	}
+	defer m.close()
+	chunkC := make(chan *Chunk)
+	key, errC := randomChunks(l, branches, chunkC)
+
+SPLIT:
+	for {
+		select {
+		case chunk := <-chunkC:
+			chunk.Data = make([]byte, chunk.Reader.Size())
+			chunk.Reader.ReadAt(chunk.Data, 0)
+			m.Put(chunk)
+
+		case err, ok := <-errC:
+			if err != nil {
+				t.Errorf("Chunker error: %v", err)
+				return
+			}
+			if !ok {
+				t.Logf("quitting SPLIT loop\n")
+				break SPLIT
+			}
+		}
+	}
+
+	chunker := &TreeChunker{
+		Branches: branches,
+	}
+	chunker.Init()
+	chunkC = make(chan *Chunk)
+	var r LazySectionReader
+	r, errC = chunker.Join(key, chunkC)
+
+	quit := make(chan bool)
+
+	go func() {
+	JOIN:
+		for {
+			select {
+			case chunk := <-chunkC:
+				go func() {
+					storedChunk, err := m.Get(chunk.Key)
+					if err == notFound {
+						t.Errorf("Chunk not found: %v", err)
+						return
+					}
+					if err != nil {
+						t.Errorf("GET error: %v", err)
+						return
+					}
+					chunk.Reader = NewChunkReaderFromBytes(storedChunk.Data)
+					chunk.Size = storedChunk.Size
+					close(chunk.C)
+				}()
+			case err, ok := <-errC:
+				if err != nil {
+					t.Errorf("Chunker error: %v", err)
+					return
+				}
+				if !ok {
+					break JOIN
+				}
+			case <-quit:
+				break JOIN
+			}
+		}
+	}()
+
+	b := make([]byte, l)
+	n, err := r.ReadAt(b, 0)
+	if err != nil {
+		t.Errorf("read error (%v/%v) %v", n, l, err)
+		close(quit)
+	}
+}
+
+func TestDbStore128_10000(t *testing.T) {
+	// defer test.Testlog(t).Detach()
+	test.LogInit()
+	testDbStore(10000, 128, t)
+}
+
+func TestDbStore128_1000(t *testing.T) {
+	// defer test.Testlog(t).Detach()
+	test.LogInit()
+	testDbStore(1000, 128, t)
+}
+
+func TestDbStore128_100(t *testing.T) {
+	// defer test.Testlog(t).Detach()
+	test.LogInit()
+	testDbStore(100, 128, t)
+}
+
+func TestDbStore2_100(t *testing.T) {
+	// defer test.Testlog(t).Detach()
+	test.LogInit()
+	testDbStore(100, 2, t)
+}

bzz/dpa.go

@@ -52,7 +52,6 @@ type Chunk struct {
 	Size   int64         // size of the data covered by the subtree encoded in this chunk
 	Key    Key           // always
 	C      chan bool     // to signal data delivery by the dpa
-	update bool          //
 }
 
 type ChunkStore interface {
@@ -154,10 +153,6 @@ func (self *DPA) retrieveLoop() {
 				for _, store := range self.Stores {
 					if _, err := store.Get(); err != nil { // no waiting/blocking here
 						dpaLogger.DebugDetailf("%v retrieving chunk %x: %v", store, chunk.Key, err)
-					} else {
-						if !chunk.update {
-							break
-						}
 					}
 				}
 			}()

bzz/memstore.go

@@ -18,6 +18,7 @@ type memStore struct {
 	entryCnt    uint   // stored entries
 	accessCnt   uint64 // access counter; oldest is thrown away when full
 	dbAccessCnt uint64
+	dbStore     *dbStore
 }
 
 /*
@@ -33,9 +34,10 @@ a hash prefix subtree containing subtrees or one storage entry (but never both)
   (access[] is a binary tree inside the multi-bit leveled hash tree)
 */
 
-func newMemStore() (m *memStore) {
+func newMemStore(d *dbStore) (m *memStore) {
 	m = &memStore{}
 	m.memtree = newMemTree(memTreeFLW, nil, 0)
+	m.dbStore = d
 	return
 }
 
@@ -222,7 +224,9 @@ func (s *memStore) Get(hash Key) (chunk *Chunk, err error) {
 		if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt {
 			s.dbAccessCnt++
 			node.lastDBaccess = s.dbAccessCnt
-			chunk.update = true
+			if s.dbStore != nil {
+				s.dbStore.updateAccessCnt(hash)
+			}
 		}
 	} else {
 		err = notFound

bzz/memstore_test.go

@@ -23,7 +23,7 @@ func randomChunks(l int64, branches int64, chunkC chan *Chunk) (key Key, errC ch
 }
 
 func testMemStore(l int64, branches int64, t *testing.T) {
-	m := newMemStore()
+	m := newMemStore(nil)
 	chunkC := make(chan *Chunk)
 	key, errC := randomChunks(l, branches, chunkC)
 

ethdb/database.go

@@ -2,7 +2,6 @@ package ethdb
 
 import (
 	"fmt"
-	"path"
 
 	"github.com/ethereum/go-ethereum/compression/rle"
 	"github.com/ethereum/go-ethereum/ethutil"
@@ -15,16 +14,14 @@ type LDBDatabase struct {
 	Comp bool
 }
 
-func NewLDBDatabase(name string) (*LDBDatabase, error) {
+func NewLDBDatabase(dbPath string) (*LDBDatabase, error) {
-	dbPath := path.Join(ethutil.Config.ExecPath, name)
-
 	// Open the db
 	db, err := leveldb.OpenFile(dbPath, nil)
 	if err != nil {
 		return nil, err
 	}
 
-	database := &LDBDatabase{DB: db, Comp: true}
+	database := &LDBDatabase{DB: db, Comp: false}
 
 	return database, nil
 }