From e32cdc4244dc49698c4d43543e9c6140c461d96a Mon Sep 17 00:00:00 2001 From: zelig Date: Fri, 19 Dec 2014 17:50:28 +0000 Subject: [PATCH] initial commit for swarm/bzz - to include blockhash pkg https://github.com/nagydani/go-dpa --- bzz/blockhash.go | 377 ++++++++++++++++++++++++++++++++++++++++++ bzz/blockhash_test.go | 97 +++++++++++ bzz/database.go | 89 ++++++++++ bzz/dbstore.go | 117 +++++++++++++ bzz/memstore.go | 302 +++++++++++++++++++++++++++++++++ 5 files changed, 982 insertions(+) create mode 100644 bzz/blockhash.go create mode 100644 bzz/blockhash_test.go create mode 100644 bzz/database.go create mode 100644 bzz/dbstore.go create mode 100644 bzz/memstore.go diff --git a/bzz/blockhash.go b/bzz/blockhash.go new file mode 100644 index 0000000000..ac3aa28800 --- /dev/null +++ b/bzz/blockhash.go @@ -0,0 +1,377 @@ +/* +The blockhash package implements a hash tree based fixed block size distributed +data storage +The block hash of a byte array is defined as follows: + +- if size is no more than BlockSize, it is stored in a single block + blockhash = sha256(int64(size) + data) + +- if size is more than BlockSize*BlockHashCount^l, but no more than BlockSize* + BlockHashCount^(l+1), the data vector is split into slices of BlockSize* + BlockHashCount^l length (except the last one). + blockhash = sha256(int64(size) + blockhash(slice0) + blockhash(slice1) + ...) +*/ + +package blockhash + +import ( + "bytes" + "crypto/sha256" + "encoding/binary" + "fmt" + "io" +) + +const HashSize = 32 +const BlockSize = 4096 +const BlockHashCount = BlockSize / HashSize + +type HashType []byte + +/* +The layered (memory, disk, distributed) storage model provides two channels, one +for storing and one for retrieving blocks. The layers are chained so that every +layer can store blocks and try to retrieve them if the previous layer did not +succeed. +*/ + +type dpaStorage struct { + store_chn chan *dpaStoreReq + retrieve_chn chan *dpaRetrieveReq + chain *dpaStorage +} + +type dpaReaderAt struct { + hash HashType + store *dpaStorage + size int64 +} + +type dpaNode struct { + data []byte + size int64 // denotes the size of data represented by the whole subtree +} + +type dpaStoreReq struct { + dpaNode + hash HashType +} + +type dpaRetrieveRes struct { + dpaNode + req_id int +} + +type dpaRetrieveReq struct { + hash HashType + req_id int + result_chn chan *dpaRetrieveRes +} + +func (h HashType) bits(i, j uint) uint { + + ii := i >> 3 + jj := i & 7 + if ii >= HashSize { + return 0 + } + + if jj+j <= 8 { + return uint((h[ii] >> jj) & ((1 << j) - 1)) + } + + res := uint(h[ii] >> jj) + jj = 8 - jj + j -= jj + for j != 0 { + ii++ + if j < 8 { + res += uint(h[ii]&((1< BlockHashCount { + stc = (stc-1)/BlockHashCount + 1 + SubtreeSize *= BlockHashCount + } + SubtreeCount := int(stc) + + block = make([]byte, SubtreeCount*HashSize) + + hdone := make(chan bool, SubtreeCount) + + ptr := int64(0) + hptr := 0 + for i := 0; i < SubtreeCount; i++ { + ptr2 := ptr + SubtreeSize + if ptr2 > size { + ptr2 = size + } + go goGetDPAhash(io.NewSectionReader(reader, ptr, ptr2-ptr), store, HashType(block[hptr:hptr+HashSize]), hdone) + ptr = ptr2 + hptr += HashSize + } + + for i := 0; i < SubtreeCount; i++ { + if !<-hdone { + return nil + } + } + } + + hashfn := sha256.New() + //binary.LittleEndian.PutUint16(b, uint16(i)) + //fmt.Printf("%d\n", size) + binary.Write(hashfn, binary.LittleEndian, int64(size)) + hashfn.Write(block) + hash := hashfn.Sum(nil) + + if store != nil { + req := new(dpaStoreReq) + req.data = block + req.size = int64(size) + req.hash = hash + store.store_chn <- req + } + + return hash + +} + +// recursive function to retrieve a section of a subtree +// len(data) == stop-start + +func getDPAblock(res *dpaRetrieveRes, data []byte, start int64, stop int64, bsize int64, retrv chan<- *dpaRetrieveReq, done chan<- bool) bool { + + for bsize >= res.size { + if bsize == BlockSize { + bsize = 0 + } else { + bsize /= BlockHashCount + } + } + + if bsize < BlockSize { + if res.size < stop { + if done != nil { + done <- false + } + return false + } + copy(data[:], res.data[start:stop]) + if done != nil { + done <- true + } + return true + } + + bstart := int(start / bsize) + bstop := int((stop + bsize - 1) / bsize) + + if len(res.data) < bstop*HashSize { + if done != nil { + done <- false + } + return false + } + + chn := make(chan *dpaRetrieveRes, bstop-bstart) + sdone := make(chan bool, bstop-bstart) + + for i := bstart; i < bstop; i++ { + + hash := HashType(res.data[i*HashSize : (i+1)*HashSize]) + req := new(dpaRetrieveReq) + req.hash = hash + req.req_id = i + req.result_chn = chn + retrv <- req + + } + + for j := bstart; j < bstop; j++ { + + res := <-chn + + i := int64(res.req_id) + a := i * bsize + aa := a + b := a + bsize + + if a < start { + a = start + } + if b > stop { + b = stop + } + + if res.size < b-aa { + if done != nil { + done <- false + } + return false + } + + if bsize == BlockSize { + getDPAblock(res, data[a-start:b-start], a-aa, b-aa, 0, retrv, sdone) + } else { + go getDPAblock(res, data[a-start:b-start], a-aa, b-aa, bsize/BlockHashCount, retrv, sdone) + } + } + + dd := true + for j := bstart; j < bstop; j++ { + if !<-sdone { + dd = false + break + } + } + + if done != nil { + done <- dd + } + return dd + +} + +func (r *dpaReaderAt) ReadAt(p []byte, off int64) (n int, err error) { + + chn := make(chan *dpaRetrieveRes) + + req := new(dpaRetrieveReq) + req.hash = r.hash + req.req_id = 0 + req.result_chn = chn + + r.store.retrieve_chn <- req + res := <-chn + + if res.size == 0 { + return 0, fmt.Errorf("Block hash %064x not found", r.hash) + } + + r.size = res.size + if len(p) == 0 { + return 0, nil + } + + bsize := int64(0) + if res.size > BlockSize { + bsize = int64(BlockSize) + for bsize*BlockHashCount < res.size { + bsize *= BlockHashCount + } + } + + err = error(nil) + + eoff := off + int64(len(p)) + if eoff > res.size { + eoff = res.size + err = io.EOF + } + + if !getDPAblock(res, p, off, eoff, bsize, r.store.retrieve_chn, nil) { + return 0, fmt.Errorf("Can't load section [%d:%d] of block hash %064x", off, eoff, r.hash) + } + + return int(eoff - off), err + +} + +func GetDPAreader(hash HashType, st *dpaStorage) *io.SectionReader { + + rd := new(dpaReaderAt) + rd.hash = hash + rd.store = st + rd.size = -1 + + rd.ReadAt(nil, 0) + + if rd.size >= 0 { + return io.NewSectionReader(rd, 0, rd.size) + } else { + return nil + } + +} + +// retrieve a data vector of a given block hash from the given storage + +func GetDPAdata(hash HashType, st *dpaStorage) []byte { + + sr := GetDPAreader(hash, st) + if sr == nil { + return nil + } + + size := sr.Size() + + data := make([]byte, int(size)) + br, _ := sr.Read(data) + if int64(br) == size { + return data + } else { + return nil + } + +} diff --git a/bzz/blockhash_test.go b/bzz/blockhash_test.go new file mode 100644 index 0000000000..35556d52f9 --- /dev/null +++ b/bzz/blockhash_test.go @@ -0,0 +1,97 @@ +// test bench for the package blockhash + +package blockhash + +import ( + // "fmt" + "math" + "math/rand" + "testing" +) + +func maketest(l int) []byte { + + r := rand.New(rand.NewSource(int64(l))) + + test := make([]byte, l) + for i := 0; i < l; i++ { + test[i] = byte(r.Intn(256)) + } + + return test +} + +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]) + } + } + } + +} diff --git a/bzz/database.go b/bzz/database.go new file mode 100644 index 0000000000..655bf9a495 --- /dev/null +++ b/bzz/database.go @@ -0,0 +1,89 @@ +package blockhash + +import ( + "fmt" + "path" + + "github.com/ethereum/go-ethereum/compression/rle" + "github.com/ethereum/go-ethereum/ethutil" + "github.com/syndtr/goleveldb/leveldb" + "github.com/syndtr/goleveldb/leveldb/iterator" +) + +type LDBDatabase struct { + db *leveldb.DB + comp bool +} + +func NewLDBDatabase(name 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} + + return database, nil +} + +func (self *LDBDatabase) Put(key []byte, value []byte) { + if self.comp { + value = rle.Compress(value) + } + + err := self.db.Put(key, value, nil) + if err != nil { + fmt.Println("Error put", err) + } +} + +func (self *LDBDatabase) Get(key []byte) ([]byte, error) { + dat, err := self.db.Get(key, nil) + if err != nil { + return nil, err + } + + if self.comp { + return rle.Decompress(dat) + } + + return dat, nil +} + +func (self *LDBDatabase) Delete(key []byte) error { + return self.db.Delete(key, nil) +} + +func (self *LDBDatabase) LastKnownTD() []byte { + data, _ := self.Get([]byte("LTD")) + + if len(data) == 0 { + data = []byte{0x0} + } + + return data +} + +func (self *LDBDatabase) NewIterator() iterator.Iterator { + return self.db.NewIterator(nil, nil) +} + +func (self *LDBDatabase) Close() { + // Close the leveldb database + self.db.Close() +} + +func (self *LDBDatabase) Print() { + iter := self.db.NewIterator(nil, nil) + for iter.Next() { + key := iter.Key() + value := iter.Value() + + fmt.Printf("%x(%d): ", key, len(key)) + node := ethutil.NewValueFromBytes(value) + fmt.Printf("%v\n", node) + } +} diff --git a/bzz/dbstore.go b/bzz/dbstore.go new file mode 100644 index 0000000000..fe69816064 --- /dev/null +++ b/bzz/dbstore.go @@ -0,0 +1,117 @@ +// disk storage layer for the package blockhash +// inefficient work-in-progress version + +package blockhash + +import ( + // "crypto/sha256" + // "encoding/binary" + "bytes" + "fmt" + // "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/ethutil" + "github.com/ethereum/go-ethereum/rlp" + "github.com/syndtr/goleveldb/leveldb" + "path" +) + +type dpaDBStorage struct { + dpaStorage + // db *ethdb.LDBDatabase + db *LDBDatabase +} + +func (s *dpaDBStorage) add(entry *dpaStoreReq) { + + data := ethutil.Encode([]interface{}{entry.data, entry.size}) + + s.db.Put(entry.hash, data) + +} + +func (s *dpaDBStorage) find(hash HashType) (entry dpaNode) { + + fmt.Printf("mao") + + data, err := s.db.Get(hash) + if err != nil { + panic("hi") + } + fmt.Printf("mao") + + dec := rlp.NewListStream(bytes.NewReader(data), uint64(len(data))) + dec.Decode(&entry) + + return + +} + +func (s *dpaDBStorage) process_store(req *dpaStoreReq) { + + s.add(req) + + if s.chain != nil { + s.chain.store_chn <- req + } + +} + +func (s *dpaDBStorage) process_retrieve(req *dpaRetrieveReq) { + + fmt.Printf("mao") + entry := s.find(req.hash) + + fmt.Printf("%v", entry.size) + 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) { + + // dbPath := path.Join(ethutil.Config.ExecPath, "bzz") + dbPath := path.Join(".", "bzz") + + // Open the db + db, err := leveldb.OpenFile(dbPath, nil) + if err != nil { + return + } + + // s.db = ðdb.LDBDatabase{db: db, comp: false} + s.db = &LDBDatabase{db: db, comp: false} + +} + +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) + } + } + +} diff --git a/bzz/memstore.go b/bzz/memstore.go new file mode 100644 index 0000000000..c03d45c3cd --- /dev/null +++ b/bzz/memstore.go @@ -0,0 +1,302 @@ +// memory storage layer for the package blockhash + +package blockhash + +const MaxEntries = 500 // max number of stored (cached) blocks +const MemTreeLW = 2 // log2(subtree count) of the subtrees +const MemTreeFLW = 14 // log2(subtree count) of the root layer + +type dpaMemStorage struct { + dpaStorage + memtree *dpaMemTree + entry_cnt uint // stored entries + access_cnt uint64 // access counter; oldest is thrown away when full +} + +/* +a hash prefix subtree containing subtrees or one storage entry (but never both) + +- access[0] stores the smallest (oldest) access count value in this subtree +- if it contains more subtrees and its subtree count is at least 4, access[1:2] + stores the smallest access count in the first and second halves of subtrees + (so that access[0] = min(access[1], access[2]) +- likewise, if subtree count is at least 8, + access[1] = min(access[3], access[4]) + access[2] = min(access[5], access[6]) + (access[] is a binary tree inside the multi-bit leveled hash tree) +*/ + +type dpaMemTree struct { + subtree []*dpaMemTree + parent *dpaMemTree + parent_idx uint + + bits uint // log2(subtree count) + width uint // subtree count + + entry *dpaStoreReq // if subtrees are present, entry should be nil + access []uint64 +} + +func newTreeNode(b uint, parent *dpaMemTree, pidx uint) (node *dpaMemTree) { + + node = new(dpaMemTree) + node.bits = b + node.width = 1 << uint(b) + node.subtree = make([]*dpaMemTree, node.width) + node.access = make([]uint64, node.width-1) + node.parent = parent + node.parent_idx = pidx + if parent != nil { + parent.subtree[pidx] = node + } + + return node + +} + +func (node *dpaMemTree) update_access(a uint64) { + + aidx := uint(0) + var aa uint64 + oa := node.access[0] + for node.access[aidx] == oa { + node.access[aidx] = a + if aidx > 0 { + aa = node.access[((aidx-1)^1)+1] + aidx = (aidx - 1) >> 1 + } else { + pidx := node.parent_idx + node = node.parent + if node == nil { + return + } + nn := node.subtree[pidx^1] + if nn != nil { + aa = nn.access[0] + } else { + aa = 0 + } + aidx = (node.width + pidx - 2) >> 1 + } + + if (aa != 0) && (aa < a) { + a = aa + } + } + +} + +func (s *dpaMemStorage) add(entry *dpaStoreReq) { + + s.access_cnt++ + + node := s.memtree + bitpos := uint(0) + for node.entry == nil { + l := entry.hash.bits(bitpos, node.bits) + st := node.subtree[l] + if st == nil { + st = newTreeNode(MemTreeLW, node, l) + bitpos += node.bits + node = st + break + } + bitpos += node.bits + node = st + } + + if node.entry != nil { + + if node.entry.hash.isEqual(entry.hash) { + node.update_access(s.access_cnt) + return + } + + for node.entry != nil { + + l := node.entry.hash.bits(bitpos, node.bits) + st := node.subtree[l] + if st == nil { + st = newTreeNode(MemTreeLW, node, l) + } + st.entry = node.entry + node.entry = nil + st.update_access(node.access[0]) + + l = entry.hash.bits(bitpos, node.bits) + st = node.subtree[l] + if st == nil { + st = newTreeNode(MemTreeLW, node, l) + } + bitpos += node.bits + node = st + + } + } + + node.entry = entry + node.update_access(s.access_cnt) + s.entry_cnt++ + +} + +func (s *dpaMemStorage) find(hash HashType) (entry *dpaStoreReq) { + + node := s.memtree + bitpos := uint(0) + for node.entry == nil { + l := hash.bits(bitpos, node.bits) + st := node.subtree[l] + if st == nil { + return nil + } + bitpos += node.bits + node = st + } + + if node.entry.hash.isEqual(hash) { + s.access_cnt++ + node.update_access(s.access_cnt) + return node.entry + } else { + return nil + } +} + +func (s *dpaMemStorage) remove_oldest() { + + node := s.memtree + + for node.entry == nil { + + aidx := uint(0) + av := node.access[aidx] + + for aidx < node.width/2-1 { + if av == node.access[aidx*2+1] { + node.access[aidx] = node.access[aidx*2+2] + aidx = aidx*2 + 1 + } else if av == node.access[aidx*2+2] { + node.access[aidx] = node.access[aidx*2+1] + aidx = aidx*2 + 2 + } else { + panic(nil) + } + } + pidx := aidx*2 + 2 - node.width + if (node.subtree[pidx] != nil) && (av == node.subtree[pidx].access[0]) { + if node.subtree[pidx+1] != nil { + node.access[aidx] = node.subtree[pidx+1].access[0] + } else { + node.access[aidx] = 0 + } + } else if (node.subtree[pidx+1] != nil) && (av == node.subtree[pidx+1].access[0]) { + if node.subtree[pidx] != nil { + node.access[aidx] = node.subtree[pidx].access[0] + } else { + node.access[aidx] = 0 + } + pidx++ + } else { + panic(nil) + } + + //fmt.Println(pidx) + node = node.subtree[pidx] + + } + + node.entry = nil + s.entry_cnt-- + node.access[0] = 0 + + //--- + + aidx := uint(0) + for { + aa := node.access[aidx] + if aidx > 0 { + aidx = (aidx - 1) >> 1 + } else { + pidx := node.parent_idx + node = node.parent + if node == nil { + return + } + aidx = (node.width + pidx - 2) >> 1 + } + if (aa != 0) && ((aa < node.access[aidx]) || (node.access[aidx] == 0)) { + node.access[aidx] = aa + } + } + +} + +// process store channel requests + +func (s *dpaMemStorage) process_store(req *dpaStoreReq) { + + if s.entry_cnt >= MaxEntries { + s.remove_oldest() + } + s.add(req) + + if s.chain != nil { + s.chain.store_chn <- req + } + +} + +// process retrieve channel requests + +func (s *dpaMemStorage) process_retrieve(req *dpaRetrieveReq) { + + entry := s.find(req.hash) + if entry == nil { + if s.chain != nil { + s.chain.retrieve_chn <- req + return + } + } + + res := new(dpaRetrieveRes) + if entry != nil { + res.dpaNode = entry.dpaNode + } + res.req_id = req.req_id + req.result_chn <- res + +} + +func (s *dpaMemStorage) Init(ch *dpaStorage) { + + s.dpaStorage.Init() + s.memtree = newTreeNode(MemTreeFLW, nil, 0) + s.chain = ch + +} + +// storage main goroutine; always processes store messages first + +func (s *dpaMemStorage) 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) + } + } + +}