From 8e6ef48a28e367d4c77cd43e8bd4f8374b120b88 Mon Sep 17 00:00:00 2001 From: zelig Date: Wed, 4 Feb 2015 12:34:40 +0100 Subject: [PATCH] dbStore + test - memStore now take dbStore and calls updateAccessCnt directly - remove update field from Chunk --- bzz/dbstore.go | 132 +++++++++++++++---------------------------- bzz/dbstore_test.go | 114 +++++++++++++++++++++++++++++++++++++ bzz/dpa.go | 5 -- bzz/memstore.go | 8 ++- bzz/memstore_test.go | 2 +- ethdb/database.go | 7 +-- 6 files changed, 167 insertions(+), 101 deletions(-) create mode 100644 bzz/dbstore_test.go diff --git a/bzz/dbstore.go b/bzz/dbstore.go index 1fd6453481..ad33099097 100644 --- a/bzz/dbstore.go +++ b/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 { - dpaStorage +type dbStore struct { 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.Size = uint64(entry.size) + rlpEntry.Data = chunk.Data + 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) - entry.data = rlpEntry.Data - entry.size = int64(rlpEntry.Size) + err := dec.Decode(&rlpEntry) + if err != nil { + 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) { - data, _ := s.db.Get(getDataKey(index.Idx)) - decodeData(data, &entry) + if s.tryAccessIdx(getIndexKey(key), &index) { + var data []byte + 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) - - if s.chain != nil { - s.chain.store_chn <- req - } + var index dpaDBIndex + s.tryAccessIdx(getIndexKey(key), &index) // result_chn == nil, only update access cnt } -func (s *dpaDBStorage) process_retrieve(req *dpaRetrieveReq) { +func newDbStore(path string) (s *dbStore, err error) { - if req.result_chn == nil { - - key := getIndexKey(req.hash) - var index dpaDBIndex - s.tryAccessIdx(key, &index) // result_chn == nil, only update access cnt + s = new(dbStore) + s.db, err = ethdb.NewLDBDatabase(path) + if err != nil { return } - 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.gcStartPos = make([]byte, 1) + s.gcStartPos[0] = kpIndex 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() { - - 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) - } - } - +func (s *dbStore) close() { + s.db.Close() } diff --git a/bzz/dbstore_test.go b/bzz/dbstore_test.go new file mode 100644 index 0000000000..fcd34c9f85 --- /dev/null +++ b/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) +} diff --git a/bzz/dpa.go b/bzz/dpa.go index d486244305..537595b8c0 100644 --- a/bzz/dpa.go +++ b/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 - } } } }() diff --git a/bzz/memstore.go b/bzz/memstore.go index 0a12bb4dc7..482418a7d5 100644 --- a/bzz/memstore.go +++ b/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 diff --git a/bzz/memstore_test.go b/bzz/memstore_test.go index c4cf2fb8d5..85c0425625 100644 --- a/bzz/memstore_test.go +++ b/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) diff --git a/ethdb/database.go b/ethdb/database.go index 2b5e265ddd..b56d3a42c6 100644 --- a/ethdb/database.go +++ b/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) { - dbPath := path.Join(ethutil.Config.ExecPath, name) - +func NewLDBDatabase(dbPath string) (*LDBDatabase, error) { // 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 }