Memory store conforms to ChunkStore interface, added test, dbstore temporarily removed.

bzz/dbstore.go

@@ -1,54 +1,415 @@
-// db-backed storage layer for chunks
+// disk storage layer for the package blockhash
+// inefficient work-in-progress version
 
 package bzz
 
 import (
+	//	"crypto/sha256"
 	"bytes"
-	// "fmt"
+	"encoding/binary"
+	"fmt"
 	"github.com/ethereum/go-ethereum/ethdb"
-	"github.com/ethereum/go-ethereum/ethutil"
+	//	"github.com/ethereum/go-ethereum/ethutil"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/syndtr/goleveldb/leveldb"
-	"path"
+	//	"path"
 )
 
+const dbMaxEntries = 5000 // max number of stored (cached) blocks
+
+const gcArraySize = 500
+const gcArrayFreeRatio = 10
+
+// key prefixes for leveldb storage
+const kpIndex = 0
+const kpData = 1
+
+var keyAccessCnt = []byte{2}
+var keyEntryCnt = []byte{3}
+var keyDataIdx = []byte{4}
+var keyGCPos = []byte{5}
+
+type gcItem struct {
+	idx    uint64
+	value  uint64
+	idxKey []byte
+}
+
 type dpaDBStorage struct {
+	dpaStorage
 	db *ethdb.LDBDatabase
+
+	// this should be stored in db, accessed transactionally
+	entryCnt, accessCnt, dataIdx uint64
+
+	gcPos, gcStartPos []byte
+	gcArray           []*gcItem
 }
 
-func (s *dpaDBStorage) Put(entry *Chunk) error {
-
-	data := ethutil.Encode([]interface{}{entry.Data, entry.Size})
-
-	s.db.Put(entry.Key, data)
-
-	return nil
+type dpaDBIndex struct {
+	Idx    uint64
+	Access uint64
 }
 
-func (s *dpaDBStorage) Get(hash Key) (chunk *Chunk, err error) {
+func bytesToU64(data []byte) uint64 {
 
-	data, err := s.db.Get(hash)
-	if err != nil {
-		panic("hi")
+	if len(data) < 8 {
+		return 0
+	}
+	return binary.LittleEndian.Uint64(data)
+
+}
+
+func u64ToBytes(val uint64) []byte {
+
+	data := make([]byte, 8)
+	binary.LittleEndian.PutUint64(data, val)
+	return data
+
+}
+
+func getIndexGCValue(index *dpaDBIndex) uint64 {
+
+	return index.Access
+
+}
+
+func (s *dpaDBStorage) updateIndexAccess(index *dpaDBIndex) {
+
+	index.Access = s.accessCnt
+
+}
+
+func getIndexKey(hash HashType) []byte {
+
+	key := make([]byte, HashSize+1)
+	key[0] = 0
+	// db keys derived from hash:
+	// two halves swapped for uniformly distributed prefix
+	copy(key[1:HashSize/2+1], hash[HashSize/2:HashSize])
+	copy(key[HashSize/2+1:HashSize+1], hash[0:HashSize/2])
+
+	return key
+}
+
+func getDataKey(idx uint64) []byte {
+
+	key := make([]byte, 9)
+	key[0] = 1
+	binary.BigEndian.PutUint64(key[1:9], idx)
+
+	return key
+}
+
+func encodeIndex(index *dpaDBIndex) []byte {
+
+	data, _ := rlp.EncodeToBytes(index)
+	return data
+
+}
+
+func encodeData(entry *dpaNode) []byte {
+
+	var rlpEntry struct {
+		Data []byte
+		Size uint64
 	}
 
-	dec := rlp.NewListStream(bytes.NewReader(data), uint64(len(data)))
-	dec.Decode(&chunk)
+	rlpEntry.Data = entry.data
+	rlpEntry.Size = uint64(entry.size)
+
+	data, _ := rlp.EncodeToBytes(rlpEntry)
+	return data
+
+}
+
+func decodeIndex(data []byte, index *dpaDBIndex) {
+
+	dec := rlp.NewStream(bytes.NewReader(data))
+	dec.Decode(index)
+
+}
+
+func decodeData(data []byte, entry *dpaNode) {
+
+	var rlpEntry struct {
+		Data []byte
+		Size uint64
+	}
+
+	dec := rlp.NewStream(bytes.NewReader(data))
+	dec.Decode(&rlpEntry)
+	entry.data = rlpEntry.Data
+	entry.size = int64(rlpEntry.Size)
+}
+
+func gcListPartition(list []*gcItem, left int, right int, pivotIndex int) int {
+	pivotValue := list[pivotIndex].value
+	dd := list[pivotIndex]
+	list[pivotIndex] = list[right]
+	list[right] = dd
+	storeIndex := left
+	for i := left; i < right; i++ {
+		if list[i].value < pivotValue {
+			dd = list[storeIndex]
+			list[storeIndex] = list[i]
+			list[i] = dd
+			storeIndex++
+		}
+	}
+	dd = list[storeIndex]
+	list[storeIndex] = list[right]
+	list[right] = dd
+	return storeIndex
+}
+
+func gcListSelect(list []*gcItem, left int, right int, n int) int {
+	if left == right {
+		return left
+	}
+	pivotIndex := (left + right) / 2
+	pivotIndex = gcListPartition(list, left, right, pivotIndex)
+	if n == pivotIndex {
+		return n
+	} else {
+		if n < pivotIndex {
+			return gcListSelect(list, left, pivotIndex-1, n)
+		} else {
+			return gcListSelect(list, pivotIndex+1, right, n)
+		}
+	}
+}
+
+func (s *dpaDBStorage) collectGarbage() {
+
+	it := s.db.NewIterator()
+	it.Seek(s.gcPos)
+	if it.Valid() {
+		s.gcPos = it.Key()
+	} else {
+		s.gcPos = nil
+	}
+	gcnt := 0
+
+	for gcnt < gcArraySize {
+
+		if (s.gcPos == nil) || (s.gcPos[0] != kpIndex) {
+			it.Seek(s.gcStartPos)
+			if it.Valid() {
+				s.gcPos = it.Key()
+			} else {
+				s.gcPos = nil
+			}
+		}
+
+		if (s.gcPos == nil) || (s.gcPos[0] != kpIndex) {
+			break
+		}
+
+		gci := new(gcItem)
+		gci.idxKey = s.gcPos
+		var index dpaDBIndex
+		decodeIndex(it.Value(), &index)
+		gci.idx = index.Idx
+		// the smaller, the more likely to be gc'd
+		gci.value = getIndexGCValue(&index)
+		s.gcArray[gcnt] = gci
+		gcnt++
+		it.Next()
+		if it.Valid() {
+			s.gcPos = it.Key()
+		} else {
+			s.gcPos = nil
+		}
+	}
+
+	cutidx := gcListSelect(s.gcArray, 0, gcnt-1, gcnt/gcArrayFreeRatio)
+	cutval := s.gcArray[cutidx].value
+
+	//fmt.Print(s.entryCnt, " ")
+
+	// actual gc
+	for i := 0; i < gcnt; i++ {
+		if s.gcArray[i].value < cutval {
+			batch := new(leveldb.Batch)
+			batch.Delete(s.gcArray[i].idxKey)
+			batch.Delete(getDataKey(s.gcArray[i].idx))
+			s.entryCnt--
+			batch.Put(keyEntryCnt, u64ToBytes(s.entryCnt))
+			s.db.Write(batch)
+		}
+	}
+
+	//fmt.Println(s.entryCnt)
+
+	s.db.Put(keyGCPos, s.gcPos)
+
+}
+
+func (s *dpaDBStorage) add(entry *dpaStoreReq) {
+
+	ikey := getIndexKey(entry.hash)
+	var index dpaDBIndex
+
+	if s.tryAccessIdx(ikey, &index) {
+		return // already exists, only update access
+	}
+
+	data := encodeData(&entry.dpaNode)
+	//data := ethutil.Encode([]interface{}{entry})
+
+	if s.entryCnt >= dbMaxEntries {
+		s.collectGarbage()
+	}
+
+	batch := new(leveldb.Batch)
+
+	s.entryCnt++
+	batch.Put(keyEntryCnt, u64ToBytes(s.entryCnt))
+	s.dataIdx++
+	batch.Put(keyDataIdx, u64ToBytes(s.dataIdx))
+	s.accessCnt++
+	batch.Put(keyAccessCnt, u64ToBytes(s.accessCnt))
+
+	batch.Put(getDataKey(s.dataIdx), data)
+
+	index.Idx = s.dataIdx
+	s.updateIndexAccess(&index)
+
+	idata := encodeIndex(&index)
+	batch.Put(ikey, idata)
+
+	s.db.Write(batch)
+
+}
+
+// try to find index; if found, update access cnt and return true
+func (s *dpaDBStorage) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
+
+	idata, err := s.db.Get(ikey)
+	if err != nil {
+		return false
+	}
+	decodeIndex(idata, index)
+
+	batch := new(leveldb.Batch)
+
+	s.accessCnt++
+	batch.Put(keyAccessCnt, u64ToBytes(s.accessCnt))
+	s.updateIndexAccess(index)
+	idata = encodeIndex(index)
+	batch.Put(ikey, idata)
+
+	s.db.Write(batch)
+
+	return true
+}
+
+func (s *dpaDBStorage) find(hash HashType) (entry dpaNode) {
+
+	key := getIndexKey(hash)
+	var index dpaDBIndex
+
+	if s.tryAccessIdx(key, &index) {
+		data, _ := s.db.Get(getDataKey(index.Idx))
+		decodeData(data, &entry)
+	}
 
 	return
 
 }
 
-func (s *dpaDBStorage) Init() {
+func (s *dpaDBStorage) process_store(req *dpaStoreReq) {
 
-	dbPath := path.Join(".", "bzz")
+	s.add(req)
+
+	if s.chain != nil {
+		s.chain.store_chn <- req
+	}
+
+}
+
+func (s *dpaDBStorage) process_retrieve(req *dpaRetrieveReq) {
+
+	if req.result_chn == nil {
+
+		key := getIndexKey(req.hash)
+		var index dpaDBIndex
+		s.tryAccessIdx(key, &index) // result_chn == nil, only update access cnt
 
-	// Open the db
-	db, err := leveldb.OpenFile(dbPath, nil)
-	if err != nil {
 		return
 	}
 
-	s.db = &ethdb.LDBDatabase{DB: db, Comp: false}
+	entry := s.find(req.hash)
+
+	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)
+	s.entryCnt = bytesToU64(data)
+	data, _ = s.db.Get(keyAccessCnt)
+	s.accessCnt = bytesToU64(data)
+	data, _ = s.db.Get(keyDataIdx)
+	s.dataIdx = bytesToU64(data)
+	s.gcPos, _ = s.db.Get(keyGCPos)
+	if s.gcPos == nil {
+		s.gcPos = s.gcStartPos
+	}
+
+	//	fmt.Println(s.entryCnt)
+	//	fmt.Println(s.accessCnt)
+	//	fmt.Println(s.dataIdx)
+
+}
+
+func (s *dpaDBStorage) Run() {
+
+	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/memstore.go

@@ -137,7 +137,7 @@ func (node *memTree) updateAccess(a uint64) {
 
 }
 
-func (s *memStore) Put(entry *Chunk) (err error) {
+func (s *memStore) Put(entry *Chunk) {
 
 	if s.entryCnt >= maxEntries {
 		s.removeOldest()
@@ -197,8 +197,7 @@ func (s *memStore) Put(entry *Chunk) (err error) {
 	return
 }
 
-func (s *memStore) Get(chunk *Chunk) (err error) {
-	hash := chunk.Key
+func (s *memStore) Get(hash Key) (chunk *Chunk, err error) {
 
 	node := s.memtree
 	bitpos := uint(0)
@@ -206,7 +205,7 @@ func (s *memStore) Get(chunk *Chunk) (err error) {
 		l := hash.bits(bitpos, node.bits)
 		st := node.subtree[l]
 		if st == nil {
-			return notFound
+			return nil, notFound
 		}
 		bitpos += node.bits
 		node = st
@@ -215,13 +214,16 @@ func (s *memStore) Get(chunk *Chunk) (err error) {
 	if node.entry.Key.isEqual(hash) {
 		s.accessCnt++
 		node.updateAccess(s.accessCnt)
+		chunk = &Chunk{
+			Key:  hash,
+			Data: node.entry.Data,
+			Size: node.entry.Size,
+		}
 		if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt {
 			s.dbAccessCnt++
 			node.lastDBaccess = s.dbAccessCnt
 			chunk.update = true
 		}
-		chunk.Data = node.entry.Data
-		chunk.Size = node.entry.Size
 	} else {
 		err = notFound
 	}

bzz/memstore_test.go

@@ -0,0 +1,123 @@
+package bzz
+
+import (
+	"crypto/rand"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/bzz/test"
+)
+
+func randomChunks(l int64, branches int64, chunkC chan *Chunk) (key Key, errC chan error) {
+	chunker := &TreeChunker{
+		Branches: branches,
+	}
+	chunker.Init()
+	key = make([]byte, 32)
+	b := make([]byte, l)
+	_, err := rand.Read(b)
+	if err != nil {
+		panic("no rand")
+	}
+	errC = chunker.Split(key, NewChunkReaderFromBytes(b), chunkC)
+	return
+}
+
+func testMemStore(l int64, branches int64, t *testing.T) {
+	m := newMemStore()
+	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 TestMemStore128_10000(t *testing.T) {
+	// defer test.Testlog(t).Detach()
+	test.LogInit()
+	testMemStore(10000, 128, t)
+}
+
+func TestMemStore128_1000(t *testing.T) {
+	// defer test.Testlog(t).Detach()
+	test.LogInit()
+	testMemStore(1000, 128, t)
+}
+
+func TestMemStore128_100(t *testing.T) {
+	// defer test.Testlog(t).Detach()
+	test.LogInit()
+	testMemStore(100, 128, t)
+}
+
+func TestMemStore2_100(t *testing.T) {
+	// defer test.Testlog(t).Detach()
+	test.LogInit()
+	testMemStore(100, 2, t)
+}