initial commit for swarm/bzz - to include blockhash pkg https://github.com/nagydani/go-dpa

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<<j)-1)) << jj
+			return res
+		}
+		res += uint(h[ii]) << jj
+		jj += 8
+		j -= 8
+	}
+	return res
+}
+
+func (h HashType) isEqual(h2 HashType) bool {
+
+	for i := range h {
+		if h[i] != h2[i] {
+			return false
+		}
+	}
+	return true
+
+}
+
+func (s *dpaStorage) Init() {
+
+	s.store_chn = make(chan *dpaStoreReq, 1000)
+	s.retrieve_chn = make(chan *dpaRetrieveReq, 1000)
+
+}
+
+// get the root hash of any data vector and store the blocks of the tree if store != nil
+
+func GetDPAroot(data []byte, store *dpaStorage) HashType {
+
+	return GetDPAhash(io.NewSectionReader(bytes.NewReader(data), 0, int64(len(data))), store)
+
+}
+
+func goGetDPAhash(reader *io.SectionReader, store *dpaStorage, hash HashType, done chan<- bool) {
+
+	hh := GetDPAhash(reader, store)
+	if hh == nil {
+		done <- false
+	} else {
+		copy(hash[:], hh[:])
+		done <- true
+	}
+
+}
+
+func GetDPAhash(reader *io.SectionReader, store *dpaStorage) HashType {
+
+	size := reader.Size()
+	var block []byte
+
+	if size <= BlockSize {
+		block = make([]byte, size)
+		br, _ := reader.Read(block)
+		if br < int(size) {
+			return nil
+		}
+	} else {
+		stc := (size + BlockSize - 1) / BlockSize
+		SubtreeSize := int64(BlockSize)
+
+		for stc > 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
+	}
+
+}

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])
+			}
+		}
+	}
+
+}

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)
+	}
+}

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 = &ethdb.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)
+		}
+	}
+
+}

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)
+		}
+	}
+
+}