forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-21 20:26:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

reorganise architecture, rewrite chunker

Browse source
This commit is contained in:
zelig 2015-01-12 23:46:03 +00:00
parent 1f7ec3539f
commit a9eb8915f2
12 changed files with 1254 additions and 631 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

377
bzz/blockhash.go
View file

@ -1,377 +0,0 @@
/*
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 bzz
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
}
}

97
bzz/blockhash_test.go
View file

@ -1,97 +0,0 @@
// test bench for the package blockhash
package bzz
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])
}
}
}
}

303
bzz/chunkIO.go Normal file
View file

@ -0,0 +1,303 @@
package bzz
import (
"bytes"
"errors"
"io"
)
type Bounded interface {
Size() int64
}
type Resizeable interface {
Bounded
Resize(int64) error
}
type Sliced interface {
Slice(int64, int64) []byte
}
// Size, Seek, Read, ReadAt and WriteTo
type SectionReader interface {
Bounded
Sliced
io.Seeker
io.Reader
io.ReaderAt
io.WriterTo
}
// Size, Seek, Write, WriteAt and ReaderFrom
type SectionWriter interface {
Bounded
Sliced
io.Seeker
io.Writer
io.WriterAt
io.ReaderFrom
}
// ChunkReader implements SectionReader on a section
// of an underlying ReaderAt.
type ChunkReader struct {
r io.ReaderAt
base int64
off int64
limit int64
}
// ChunkWriter implements SectionWriter on a section
// of an underlying WriterAt.
type ChunkWriter struct {
w io.WriterAt
base int64
off int64
limit int64
}
type SectionReadWriter struct {
Bounded
Sliced
io.Seeker
io.Reader
io.ReaderAt
io.WriterTo
io.Writer
io.WriterAt
io.ReaderFrom
}
// NewChunkReader returns a ChunkReader that reads from r
// starting at offset off and stops with EOF after n bytes.
func NewChunkReader(r io.ReaderAt, off int64, n int64) *ChunkReader {
return &ChunkReader{r: r, base: off, off: off, limit: off + n}
}
func NewChunkReaderFromBytes(b []byte) *ChunkReader {
return NewChunkReader(bytes.NewReader(b), 0, int64(len(b)))
}
// NewChunkWriter returns a ChunkWriter that writes to w
// starting at offset off and stops with EOF if write would go past off+n
func NewChunkWriter(w io.WriterAt, off int64, n int64) *ChunkWriter {
return &ChunkWriter{w: w, base: off, off: off, limit: off + n}
}
func NewChunkWriterFromBytes(b []byte) *ChunkWriter {
return NewChunkWriter(NewByteSliceWriter(b), 0, int64(len(b)))
}
// the write equivalent of bytes.NewReader(b)
func NewByteSliceWriter(b []byte) *ByteSliceWriter {
return &ByteSliceWriter{b: b, off: 0, limit: int64(len(b))}
}
type ByteSliceWriter struct {
b []byte
off int64
limit int64
}
func (self *ByteSliceWriter) Slice(from, to int64) (slice []byte) {
if from >= 0 && to <= self.limit {
slice = self.b[from:to]
}
return
}
func (self *ByteSliceWriter) WriteAt(b []byte, off int64) (n int, err error) {
if off < 0 || off >= self.limit {
return 0, io.ErrShortWrite
}
if n = int(self.limit - off); len(b) > n {
err = io.ErrShortWrite
} else {
n = len(b)
}
copy(self.b[off:], b)
return
}
func (self *ByteSliceWriter) Size() (size int64) {
return self.limit
}
var errUnableToResize = errors.New("unable to resize")
func (self *ByteSliceWriter) Resize(size int64) (err error) {
if self.Size() != 0 {
err = errUnableToResize
} else {
self.b = make([]byte, size)
self.limit = size
}
return
}
// Size returns the size of the section in bytes.
func (s *ChunkReader) Size() int64 { return s.limit - s.base }
func (s *ChunkWriter) Size() int64 { return s.limit - s.base }
var errWhence = errors.New("Seek: invalid whence")
var errOffset = errors.New("Seek: invalid offset")
func (s *ChunkReader) Seek(offset int64, whence int) (int64, error) {
switch whence {
default:
return 0, errWhence
case 0:
offset += s.base
case 1:
offset += s.off
case 2:
offset += s.limit
}
if offset < s.base {
return 0, errOffset
}
s.off = offset
return offset - s.base, nil
}
func (s *ChunkWriter) Seek(offset int64, whence int) (int64, error) {
switch whence {
default:
return 0, errWhence
case 0:
offset += s.base
case 1:
offset += s.off
case 2:
offset += s.limit
}
if offset < s.base {
return 0, errOffset
}
s.off = offset
return offset - s.base, nil
}
func (self *ChunkWriter) Resize(size int64) (err error) {
err = errUnableToResize
if self.Size() >= size {
err = nil
} else {
if self.Size() == 0 {
if ws, ok := self.w.(Resizeable); ok {
err = ws.Resize(size)
if err == nil {
self.limit = size
}
}
}
}
return
}
func (s *ChunkReader) Read(p []byte) (n int, err error) {
if s.off >= s.limit {
return 0, io.EOF
}
if max := s.limit - s.off; int64(len(p)) > max {
p = p[0:max]
}
n, err = s.r.ReadAt(p, s.off)
s.off += int64(n)
return
}
func (s *ChunkWriter) Write(p []byte) (n int, err error) {
if s.off >= s.limit {
return 0, io.EOF
}
if max := s.limit - s.off; int64(len(p)) > max {
p = p[0:max]
}
n, err = s.w.WriteAt(p, s.off)
s.off += int64(n)
return
}
func (s *ChunkReader) Slice(from, to int64) []byte {
if sl, ok := s.r.(Sliced); ok {
return sl.Slice(from, to)
}
return nil
}
func (s *ChunkWriter) Slice(from, to int64) (b []byte) {
if sl, ok := s.w.(Sliced); ok {
b = sl.Slice(from, to)
}
return
}
//
func (s *ChunkReader) ReadAt(p []byte, off int64) (n int, err error) {
if off < 0 || off >= s.limit-s.base {
return 0, io.EOF
}
off += s.base
if max := s.limit - off; int64(len(p)) > max {
p = p[0:max]
n, err = s.r.ReadAt(p, off)
if err == nil {
err = io.EOF
}
return n, err
}
return s.r.ReadAt(p, off)
}
//
func (s *ChunkWriter) WriteAt(p []byte, off int64) (n int, err error) {
if off < 0 || off >= s.limit-s.base {
return 0, io.EOF
}
off += s.base
if max := s.limit - off; int64(len(p)) > max {
p = p[0:max]
n, err = s.w.WriteAt(p, off)
return n, err
}
return s.w.WriteAt(p, off)
}
func (s *ChunkWriter) ReadFrom(r io.Reader) (n int64, err error) {
var m int
// if byte slice is available
if slice := s.Slice(s.off, s.limit); slice != nil {
m, err = r.Read(slice)
} else {
b := make([]byte, s.limit-s.off)
_, err = r.Read(b)
m, err = s.Write(b)
}
n = int64(m)
return
}
func (r *ChunkReader) WriteTo(w io.Writer) (n int64, err error) {
var b []byte
var m int
if b := r.Slice(r.off, r.limit); b == nil {
// if slices not available we do it with extra allocation
b = make([]byte, r.limit-r.off)
m, err = r.Read(b)
if err != nil {
return
}
}
m, err = w.Write(b)
if m > len(b) {
panic("bytes.Reader.WriteTo: invalid Write count")
}
r.off = r.base + int64(m)
n = int64(m)
if m != len(b) && err == nil {
err = io.ErrShortWrite
}
return
}

401
bzz/chunker.go Normal file
View file

@ -0,0 +1,401 @@
/*
The distributed storage implemented in this package requires fix sized chunks of content
Chunker is the interface to a component that is responsible for disassembling and assembling larger data.
TreeChunker implements a Chunker based on a tree structure defined as follows:
1 if size is no more than chunksize, it is stored in a single chunk
key = sha256(int64(size) + data)
2 if size is more than chunksize*HashCount^l, but no more than chunksize*
HashCount^(l+1), the data vector is split into slices of chunksize*
HashCount^l length (except the last one).
key = sha256(int64(size) + key(slice0) + key(slice1) + ...)
*/
package bzz
import (
"crypto"
"encoding/binary"
"fmt"
"sync"
"time"
)
const (
hasherfunc crypto.Hash = crypto.SHA256 // http://golang.org/pkg/hash/#Hash
branches int64 = 4
)
var (
// hashSize int64 = hasherfunc.New().Size() // hasher knows about its own length in bytes
// chunksize int64 = branches * hashSize // chunk is defined as this
joinTimeout = 120 * time.Second
splitTimeout = 120 * time.Second
)
type Key []byte
/*
Chunker is the interface to a component that is responsible for disassembling and assembling larger data and indended to be the dependency of a DPA storage system with fixed maximum chunksize.
It relies on the underlying chunking model.
When calling Split, the caller gets returned a channel (chan *Chunk) on which it receives chunks to store. The DPA delegates to storage layers (implementing ChunkStore interface). NewChunkstore(DB) is a convenience wrapper with which all DBs (conforming to DB interface) can serve as ChunkStores. See chunkStore.go
After getting notified that all the data has been split (the error channel and chunk channel are closed), the caller can safely read or save the root key. Optionally it times out if not all chunks get stored or not the entire stream of data has been processed. By inspecting the errc channel the caller can check if any explicit errors (typically IO read/write failures) occured during splitting.
When calling Join with a root key, the data can be nil ponter in which case it will be initialized as a byte slice based reader corresponding the size of the entire subtree encoded in the chunk. The caller gets returned a channel and an error channel. The chunk channel is the one on which the caller receives placeholder chunks with missing data. The DPA is supposed to forward this to the chunk stores and notify the chunker if the data has been delivered (i.e. retrieved from memory cache, disk-persisted db or cloud based swarm delivery. The chunker then puts these together and notifies the DPA if data has been assembled by a closed error channel. Once the DPA finds the data has been joined, it is free to deliver it back to swarm in full (if the original request was via the bzz protocol) or save and serve if it it was a local client request.
*/
type Chunker interface {
/*
When splitting, data is given as a SectionReader, and the key is a hashSize long byte slice (Key), the root hash of the entire content will fill this once processing finishes.
New chunks to store are coming to caller via the chunk channel (first return parameter)
If an error is encountered during splitting, it is fed to errC error channel (second return parameter)
A closed error and chunk channel signal process completion at which point the key can be considered final if there were no errors.
*/
Split(key Key, data SectionReader) (chan *Chunk, chan error)
/*
Join reconstructs original content based on a root key
When joining, data is given as a SectionWriter preallocation for which is taken care of by the caller.
If the size of the SectionWriter is found 0 Chunker will resize it once the entire data size is known from the root chunk. If resize is not supported by the writer, an error is given.
Any other size value is checked and if it does not fit the actual datasize, an error will be reported.
New chunks to retrieve are coming to caller via the Chunk channel (first return parameter)
If an error is encountered during joining, it is fed to errC error channel (second return parameter)
A closed error and chunk channel signal process completion at which point the data can be considered final if there were no errors.
*/
Join(key Key, data SectionWriter) (chan *Chunk, chan error)
}
/*
Tree chunker is a concrete implementation of data chunking.
This chunker works in a simple way, it builds a tree out of the document so that each node either represents a chunk of real data or a chunk of data representing an branching non-leaf node of the tree. In particular each such non-leaf chunk will represent is a concatenation of the hash of its respective children. This scheme simultaneously guarantees data integrity as well as self addressing. Abstract nodes are transparent since their represented size component is strictly greater than their maximum data size, since they encode a subtree.
If all is well it is possible to implement this by simply composing readers and writers so that no extra allocation or buffering is necessary for the data splitting. This means that in principle there can be direct IO between : memory, file system, network socket (bzz peers storage request is read from the socket ). In practice there may be need for several stages of internal buffering.
Unfortunately the hashing itself does use extra copies and allocatetion though since it does need it.
*/
type TreeChunker struct {
Branches int64
HashFunc crypto.Hash
JoinTimeout time.Duration
SplitTimeout time.Duration
// calculated
hashSize int64 // self.HashFunc.New().Size()
chunkSize int64 // hashSize* Branches
}
func (self *TreeChunker) Init() {
if self.HashFunc == 0 {
self.HashFunc = hasherfunc
}
if self.Branches == 0 {
self.Branches = branches
}
if self.JoinTimeout == 0 {
self.JoinTimeout = joinTimeout
}
if self.SplitTimeout == 0 {
self.SplitTimeout = splitTimeout
}
self.hashSize = int64(self.HashFunc.New().Size())
self.chunkSize = self.hashSize * self.Branches
dpaLogger.Debugf("Chunker initialised: branches: %v, hashsize: %v, chunksize: %v, join timeout: %v , split timeout: %v", self.Branches, self.hashSize, self.chunkSize, self.JoinTimeout, self.SplitTimeout)
}
type Chunk struct {
Data SectionReader // nil if request, to be supplied by dpa
Size int64 // size of the data covered by the subtree encoded in this chunk
// not the size of data, which is Data.Size() see SectionReader
// 0 if request, to be supplied by dpa
Key Key // always
C chan bool // to signal data delivery by the dpa
wg sync.WaitGroup
}
func (self *Chunk) String() string {
var size int64
if self.Data != nil {
size = self.Data.Size()
}
return fmt.Sprintf("Key: [%x..] TreeSize: %v Chunksize: %v", self.Key[:4], self.Size, size)
}
// The treeChunkers own Hash hashes together
// - the size (of the subtree encoded in the Chunk)
// - the Chunk, ie. the contents read from the input reader
func (self *TreeChunker) Hash(size int64, input SectionReader) []byte {
hasher := self.HashFunc.New()
binary.Write(hasher, binary.LittleEndian, size)
input.WriteTo(hasher) // SectionReader implements io.WriterTo
return hasher.Sum(nil)
}
func (self *TreeChunker) Split(key Key, data SectionReader) (chunkC chan *Chunk, errC chan error) {
wg := &sync.WaitGroup{}
chunkC = make(chan *Chunk)
errC = make(chan error)
rerrC := make(chan error)
timeout := time.After(splitTimeout)
if key == nil {
dpaLogger.Debugf("please allocate byte slice for root key")
return
}
wg.Add(1)
dpaLogger.Debugf("add one")
go func() {
depth := 0
treeSize := self.chunkSize
size := data.Size()
// takes lowest depth such that chunksize*HashCount^(depth+1) > size
// power series, will find the order of magnitude of the data size in base hashCount or numbers of levels of branching in the resulting tree.
for ; treeSize < size; treeSize *= self.Branches {
depth++
}
dpaLogger.Debugf("split request received for data (%v bytes, depth: %v)", size, depth)
//launch actual recursive function passing the workgroup
self.split(depth, treeSize, key, data, chunkC, rerrC, wg)
}()
// closes internal error channel if all subprocesses in the workgroup finished
go func() {
dpaLogger.Debugf("waiting for splitter to finish")
wg.Wait()
dpaLogger.Debugf("splitter finished. closing rerrC")
close(rerrC)
}()
// waiting for request to end with wg finishing, error, or timeout
go func() {
dpaLogger.Debugf("waiting for rerrC to close")
select {
case err := <-rerrC:
dpaLogger.Debugf("action on rerrC")
if err != nil {
dpaLogger.Debugf("error on rerrC")
errC <- err
} // otherwise splitting is complete
case <-timeout:
errC <- fmt.Errorf("split time out")
}
close(chunkC)
close(errC)
}()
return
}
func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionReader, chunkC chan *Chunk, errc chan error, parentWg *sync.WaitGroup) {
defer parentWg.Done()
size := data.Size()
var newChunk *Chunk
var hash Key
dpaLogger.Debugf("depth: %v, max subtree size: %v, data size: %v", depth, treeSize, size)
switch {
case depth == 0:
if size > treeSize {
panic("ouch")
}
// leaf nodes -> content chunks
hash = self.Hash(size, data)
dpaLogger.Debugf("content chunk: max subtree size: %v, data size: %v", treeSize, size)
newChunk = &Chunk{
Key: hash,
Data: data,
Size: size,
}
case size < treeSize:
// last item on this level (== size % self.Branches ^ (depth + 1) )
self.split(depth-1, int64(treeSize/self.Branches), key, data, chunkC, errc, parentWg)
return
default:
treeSize /= self.Branches
// intermediate chunk containing child nodes hashes
branches := int64(size/treeSize) + 1
dpaLogger.Debugf("intermediate node: setting branches: %v, depth: %v, max subtree size: %v, data size: %v", branches, depth, treeSize, size)
var chunk []byte = make([]byte, branches*self.hashSize)
var pos, i int64
childrenWg := &sync.WaitGroup{}
for i < branches {
// the last item can have shorter data
if size-pos < treeSize*self.Branches {
treeSize = size - pos
}
// take the section of the data corresponding encoded in the subTree
subTreeData := NewChunkReader(data, pos, treeSize)
// the hash of that data
subTreeKey := chunk[i*self.hashSize : (i+1)*self.hashSize]
childrenWg.Add(1)
go self.split(depth-1, treeSize, subTreeKey, subTreeData, chunkC, errc, childrenWg)
i++
pos += treeSize
}
// wait for all the children to complete calculating their hashes and copying them onto sections of the chunk
childrenWg.Wait()
// now we got the hashes in the chunk, then hash the chunk
chunkReader := NewChunkReaderFromBytes(chunk) // bytes.Reader almost implements SectionReader
hash = self.Hash(treeSize, chunkReader)
newChunk = &Chunk{
Key: hash,
Data: chunkReader,
Size: treeSize,
}
}
// send off new chunk to storage
dpaLogger.Debugf("sending chunk on chunk channel")
chunkC <- newChunk
dpaLogger.Debugf("sent chunk on chunk channel")
// report hash of this chunk one level up (keys corresponds to the proper subslice of the parent chunk)
dpaLogger.Debugf("copying parent key ")
copy(key, hash)
dpaLogger.Debugf("copied parent key ")
}
func (self *TreeChunker) Join(key Key, data SectionWriter) (chunkC chan *Chunk, errC chan error) {
// initialise return parameters
errC = make(chan error)
chunkC = make(chan *Chunk)
// timer to time out the operation (needed within so as to avoid process leakage)
timeout := time.After(joinTimeout)
wg := &sync.WaitGroup{}
// initialise internal error channel
rerrC := make(chan error)
quitC := make(chan bool)
wg.Add(1)
go func() {
// create the 'chunk' for root chunk of the data tree
chunk := &Chunk{
Key: key,
C: make(chan bool, 1),
}
// request data
dpaLogger.Debugf("request root chunk for key %x", key[:4])
chunkC <- chunk
// wait for reponse, if no root, we cannot go on
select {
case <-chunk.C: // bells ringing data delivered
dpaLogger.Debugf("request root chunk data has come, size %v", chunk.Size)
case <-timeout:
err := fmt.Errorf("split time out waiting for root")
rerrC <- err
wg.Done()
return
}
if data.Size() < chunk.Size {
dpaLogger.Debugf("trying to resize writer to size %v for join data", chunk.Size)
var err error
if resizeable, ok := data.(Resizeable); ok {
err = resizeable.Resize(chunk.Size)
} else {
err = fmt.Errorf("writer does not support resizing and has insufficient size %v (need %v)", data.Size(), chunk.Size)
}
if err != nil {
dpaLogger.Debugf("%v", err)
rerrC <- err
wg.Done()
return
}
}
// calculate depth and max treeSize
var depth int
var treeSize int64 = self.chunkSize
for ; treeSize < chunk.Size; treeSize *= self.Branches {
depth++
}
// launch recursive call on root chunk
self.join(depth, treeSize, chunk, data, chunkC, rerrC, wg, quitC)
}()
// waits for all the processes to finish and signals by closing internal rerrc
go func() {
wg.Wait()
close(rerrC)
}()
go func() {
select {
case err := <-rerrC:
if err != nil {
errC <- err
} // otherwise channel is closed, data joining complete
case <-timeout:
errC <- fmt.Errorf("join time out")
close(quitC)
}
// this will indicate to the caller that processing is finished (with or without error)
close(errC)
close(chunkC)
}()
return
}
func (self *TreeChunker) join(depth int, treeSize int64, chunk *Chunk, data SectionWriter, chunkC chan *Chunk, errC chan error, wg *sync.WaitGroup, quitC chan bool) {
defer wg.Done()
select {
case <-quitC:
case <-chunk.C: // bells are ringing, data have been delivered
dpaLogger.Debugf("received chunk data: %v", chunk)
switch {
case chunk.Size <= treeSize && depth == 0:
dpaLogger.Debugf("reading into data")
// we received a chunk for a leaf node representing actual content
if _, err := data.ReadFrom(chunk.Data); err != nil {
errC <- err
}
return
case chunk.Size < treeSize:
// this must be a last item on its level
self.join(depth-1, treeSize/self.Branches, chunk, data, chunkC, errC, wg, quitC)
return
default:
// intermediate chunk, chunk containing hashes of child nodes
var pos, i int64
for pos < chunk.Size {
// create partial Chunk in order to send a retrieval request
subtree := &Chunk{
Key: make([]byte, self.hashSize), // preallocate hashSize long slice for key
C: make(chan bool, 1), // close channel to signal data delivery
}
// read the Hash of the subtree from the relevant section of the Chunk into the allocated byte slice in subtree.Key
chunk.Data.ReadAt(subtree.Key, i*self.hashSize)
// call recursively on the subtree
subTreeData := NewChunkWriter(data, pos, treeSize)
wg.Add(1)
go self.join(depth-1, treeSize/self.Branches, subtree, subTreeData, chunkC, errC, wg, quitC)
// submit request
chunkC <- subtree
i++
pos += subtree.Size
}
}
}
}

207
bzz/chunker_test.go Normal file
View file

@ -0,0 +1,207 @@
package bzz
import (
"bytes"
"math/rand"
"testing"
"time"
"github.com/ethereum/go-ethereum/logger"
)
/*
Tests TreeChunker by splitting and joining a random byte slice
*/
type testLogger struct{ t *testing.T }
func testlog(t *testing.T) testLogger {
logger.Reset()
l := testLogger{t}
logger.AddLogSystem(l)
return l
}
func (testLogger) GetLogLevel() logger.LogLevel { return logger.DebugDetailLevel }
func (testLogger) SetLogLevel(logger.LogLevel) {}
func (l testLogger) LogPrint(level logger.LogLevel, msg string) {
l.t.Logf("%s", msg)
}
func (testLogger) detach() {
logger.Flush()
logger.Reset()
}
func randomByteSlice(l int) (b []byte) {
r := rand.New(rand.NewSource(int64(l)))
b = make([]byte, l)
for i := 0; i < l; i++ {
b[i] = byte(r.Intn(256))
}
return
}
func testDataReader(l int) (r *ChunkReader, slice []byte) {
slice = randomByteSlice(l)
r = NewChunkReaderFromBytes(slice)
return
}
type chunkerTester struct {
errors []error
chunks []*Chunk
timeout bool
}
func (self *chunkerTester) checkChunks(t *testing.T, want int) {
l := len(self.chunks)
if l != want {
t.Errorf("expected %v chunks, got %v", want, l)
}
}
func (self *chunkerTester) Split(chunker *TreeChunker, l int) (key Key, input []byte) {
// reset
self.errors = nil
self.chunks = nil
self.timeout = false
data, slice := testDataReader(l)
input = slice
key = make([]byte, 32)
chunkC, errC := chunker.Split(key, data)
quitC := make(chan bool)
timeout := time.After(60 * time.Second)
go func() {
LOOP:
for {
select {
case <-timeout:
self.timeout = true
break LOOP
case chunk, ok := <-chunkC:
if chunk != nil {
self.chunks = append(self.chunks, chunk)
}
if !ok { // game over but need to continue to see errc still
chunkC = nil // make it block so no infinite loop
}
case err, ok := <-errC:
if err != nil {
self.errors = append(self.errors, err)
}
if !ok {
break LOOP
}
}
}
close(quitC)
}()
<-quitC // waiting for it to finish
return
}
func (self *chunkerTester) Join(t *testing.T, chunker *TreeChunker, key Key) (data []byte) {
// reset but not the chunks
self.errors = nil
self.timeout = false
w := NewChunkWriterFromBytes(nil)
chunkC, errC := chunker.Join(key, w)
quitC := make(chan bool)
timeout := time.After(60 * time.Second)
go func() {
LOOP:
for {
t.Logf("waiting to mock Chunk Store")
select {
case <-timeout:
self.timeout = true
break LOOP
case chunk, ok := <-chunkC:
if chunk != nil {
t.Logf("got request %v", chunk)
// this just mocks the behaviour of a chunk store retrieval
var found bool
for _, ch := range self.chunks {
if bytes.Compare(chunk.Key, ch.Key) == 0 {
found = true
t.Logf("found data %v", ch)
// ch.Data.Seek(0, 0) // the reader has to be reset
chunk.Data = ch.Data
chunk.Size = ch.Size
t.Logf("updated chunk %v", chunk)
close(chunk.C)
break
}
}
if !found {
t.Errorf("chunk request unknown for %x", chunk.Key[:4])
}
}
if !ok { // game over but need to continue to see errc still
chunkC = nil // make it block so no infinite loop
}
case err, ok := <-errC:
if err != nil {
self.errors = append(self.errors, err)
}
if !ok {
break LOOP
}
}
}
close(quitC)
}()
<-quitC // waiting for it to finish
w.Seek(0, 0)
t.Logf("reader size %v", w.Size())
return w.Slice(0, w.Size())
}
func TestChunker0(t *testing.T) {
defer testlog(t).detach()
chunker := &TreeChunker{
Branches: 2,
SplitTimeout: 100 * time.Millisecond,
}
chunker.Init()
tester := &chunkerTester{}
key, input := tester.Split(chunker, 32)
tester.checkChunks(t, 1)
t.Logf("chunks: %v", tester.chunks)
output := tester.Join(t, chunker, key)
if bytes.Compare(output, input) != 0 {
t.Errorf("input and output mismatch\n IN: %x\nOUT: %x\n", input, output)
}
}
func TestChunker1(t *testing.T) {
defer testlog(t).detach()
chunker := &TreeChunker{
Branches: 2,
SplitTimeout: 100 * time.Millisecond,
}
chunker.Init()
tester := &chunkerTester{}
key, input := tester.Split(chunker, 70)
tester.checkChunks(t, 2)
t.Logf("chunks: %v", tester.chunks)
output := tester.Join(t, chunker, key)
if bytes.Compare(output, input) != 0 {
t.Errorf("input and output mismatch\n IN: %x\nOUT: %x\n", input, output)
}
}

87
bzz/dbstore.go
View file

@ -1,14 +1,11 @@
// disk storage layer for the package blockhash
// inefficient work-in-progress version
// db-backed storage layer for chunks
package bzz
import (
// "crypto/sha256"
// "encoding/binary"
"bytes"
"fmt"
// "github.com/ethereum/go-ethereum/ethdb"
// "fmt"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/ethutil"
"github.com/ethereum/go-ethereum/rlp"
"github.com/syndtr/goleveldb/leveldb"
@ -16,71 +13,34 @@ import (
)
type dpaDBStorage struct {
dpaStorage
// db *ethdb.LDBDatabase
db *LDBDatabase
db *ethdb.LDBDatabase
}
func (s *dpaDBStorage) add(entry *dpaStoreReq) {
func (s *dpaDBStorage) Put(entry *Chunk) error {
data := ethutil.Encode([]interface{}{entry.data, entry.size})
data := ethutil.Encode([]interface{}{entry.Data, entry.Size})
s.db.Put(entry.hash, data)
s.db.Put(entry.Key, data)
return nil
}
func (s *dpaDBStorage) find(hash HashType) (entry dpaNode) {
fmt.Printf("mao")
func (s *dpaDBStorage) Get(hash Key) (chunk *Chunk, err error) {
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)
dec.Decode(&chunk)
return
}
func (s *dpaDBStorage) process_store(req *dpaStoreReq) {
func (s *dpaDBStorage) Init() {
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
@ -89,29 +49,6 @@ func (s *dpaDBStorage) Init(ch *dpaStorage) {
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)
}
}
s.db = &ethdb.LDBDatabase{DB: db, Comp: false}
}

13
bzz/dhtstore.go Normal file
View file

@ -0,0 +1,13 @@
package bzz
/*
DHT implements the chunk store that directly communicates with the bzz protocol on the one hand and the kademlia node table on the other.
It accumulates requests from peers, keeping a request pool and does forwarding for incoming requests and handles expiry/timeout.
*/
// it implements the ChunkStore interface as well as the PeerPool interface for bzz
type DHTStore struct {
// note that it should be initialised with the same Cademlia instance that runs under the base protocol
// cad: *p2p.Cademlia
}

147
bzz/dpa.go Normal file
View file

@ -0,0 +1,147 @@
package bzz
import (
"sync"
// "time"
ethlogger "github.com/ethereum/go-ethereum/logger"
// "github.com/ethereum/go-ethereum/rlp"
)
/*
DPA provides the client API entrypoints Store and Retrieve to store and retrieve
It can store anything that has a byte slice representation, so files or serialised objects etc.
Storage: DPA calls the Chunker to segment the input datastream of any size to a merkle hashed tree of blocks. The key of the root block is returned to the client.
Retrieval: given the key of the root block, the DPA retrieves the block chunks and reconstructs the original data.
As the chunker produces chunks, DPA dispatches them to the chunk stores for storage or retrieval. The chunk stores are typically sequenced as memory cache, local disk/db store, cloud/distributed/dht storage. Storage requests will reach to all 3 components while retrieval requests stop after the first successful retrieval.
*/
var dpaLogger = ethlogger.NewLogger("BZZ")
type DPA struct {
Chunker Chunker
Stores []ChunkStore
wg sync.WaitGroup
quitC chan bool
}
type ChunkStore interface {
Put(*Chunk) error
Get(*Chunk) error
}
/*
convenience methods to help convert various typical data inputs to the canonical input to DPA storage: SectionReader
BytesToReader(data []byte) (SectionReader, error)
*/
// func BytesToReader(data []byte) (SectionReader, error) {
// return NewChunkReaderFromBytes(data), nil
// }
// func AnythingToReader(data interface{}) (SectionReader, error) {
// return NewChunkReaderFromBytes(rlp.Encode(data)), nil
// }
func (self *DPA) Retrieve(key Key, data SectionReadWriter) (err error) {
joinC := make(chan bool)
reqsC := make(chan bool)
var requests int
self.wg.Add(1)
chunkWg := sync.WaitGroup{}
chunkWg.Add(1)
go func() {
chunkWg.Wait() // wait for all chunk retrieval requests to process
close(reqsC) // signal to channel
}()
// r is potentially nil pointer, by default chunker allocates
// a SectionReadWriter based on a byte slice equal to the stored object image's bytes
dpaLogger.Debugf("bzz honey retrieve")
go func() {
var ok bool
chunkC, errC := self.Chunker.Join(key, data)
LOOP:
for {
select {
case chunk, ok := <-chunkC:
if chunk != nil { // game over
chunk.wg = chunkWg
chunkWg.Add(1) // need to call Done by any storage that first retrieves the data
for _, store := range self.Stores {
requests++
if err = store.Get(chunk); err != nil { // no waiting/blocking here
dpaLogger.DebugDetailf("%v retrieved chunk %x", store, chunk.Key)
break // the inner loop
}
}
}
if !ok { // game over but need to continue to see errc still
chunkC = nil // make it block so no infinite loop
chunkWg.Done()
}
case err, ok = <-errC:
dpaLogger.Warnf("%v", err)
if !ok {
break LOOP
}
dpaLogger.DebugDetailf("%v", err)
case <-reqsC:
dpaLogger.DebugDetailf("processed all %v chunk retrieval requests for root key %x", requests, key)
case <-self.quitC:
break LOOP
}
}
close(joinC)
self.wg.Done()
}()
<-joinC
return
}
func (self *DPA) Store(data SectionReader) (key Key, err error) {
dpaLogger.Debugf("bzz honey store")
chunkC, errC := self.Chunker.Split(key, data)
LOOP:
for {
select {
case chunk, ok := <-chunkC:
if chunk != nil {
for _, store := range self.Stores {
store.Put(chunk) // no waiting/blocking here
}
}
if !ok { // game over but need to continue to see errc still
chunkC = nil // make it block so no infinite loop
}
case err, ok := <-errC:
dpaLogger.Warnf("%v", err)
if !ok {
break LOOP
}
dpaLogger.DebugDetailf("%v", err)
case <-self.quitC:
break LOOP
}
}
return
}

85
bzz/dpa_test.go Normal file
View file

@ -0,0 +1,85 @@
// test bench for the package blockhash
package bzz
import (
// "fmt"
// "math"
// "math/rand"
// "testing"
)
// 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])
// }
// }
// }
// }

137
bzz/memstore.go
View file

@ -2,12 +2,17 @@
package bzz
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
import (
"bytes"
)
const (
maxEntries = 500 // max number of stored (cached) blocks
memTreeLW = 2 // log2(subtree count) of the subtrees
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
@ -26,6 +31,42 @@ 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 (x Key) Size() uint {
return uint(len(x))
}
func (x Key) isEqual(y Key) bool {
return bytes.Compare(x, y) == 0
}
func (h Key) bits(i, j uint) uint {
ii := i >> 3
jj := i & 7
if ii >= h.Size() {
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
}
type dpaMemTree struct {
subtree []*dpaMemTree
parent *dpaMemTree
@ -34,7 +75,7 @@ type dpaMemTree struct {
bits uint // log2(subtree count)
width uint // subtree count
entry *dpaStoreReq // if subtrees are present, entry should be nil
entry *Chunk // if subtrees are present, entry should be nil
access []uint64
}
@ -87,17 +128,17 @@ func (node *dpaMemTree) update_access(a uint64) {
}
func (s *dpaMemStorage) add(entry *dpaStoreReq) {
func (s *dpaMemStorage) add(entry *Chunk) {
s.access_cnt++
node := s.memtree
bitpos := uint(0)
for node.entry == nil {
l := entry.hash.bits(bitpos, node.bits)
l := entry.Key.bits(bitpos, node.bits)
st := node.subtree[l]
if st == nil {
st = newTreeNode(MemTreeLW, node, l)
st = newTreeNode(memTreeLW, node, l)
bitpos += node.bits
node = st
break
@ -108,26 +149,26 @@ func (s *dpaMemStorage) add(entry *dpaStoreReq) {
if node.entry != nil {
if node.entry.hash.isEqual(entry.hash) {
if node.entry.Key.isEqual(entry.Key) {
node.update_access(s.access_cnt)
return
}
for node.entry != nil {
l := node.entry.hash.bits(bitpos, node.bits)
l := node.entry.Key.bits(bitpos, node.bits)
st := node.subtree[l]
if st == nil {
st = newTreeNode(MemTreeLW, node, l)
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)
l = entry.Key.bits(bitpos, node.bits)
st = node.subtree[l]
if st == nil {
st = newTreeNode(MemTreeLW, node, l)
st = newTreeNode(memTreeLW, node, l)
}
bitpos += node.bits
node = st
@ -141,7 +182,7 @@ func (s *dpaMemStorage) add(entry *dpaStoreReq) {
}
func (s *dpaMemStorage) find(hash HashType) (entry *dpaStoreReq) {
func (s *dpaMemStorage) find(hash Key) (entry *Chunk) {
node := s.memtree
bitpos := uint(0)
@ -155,7 +196,7 @@ func (s *dpaMemStorage) find(hash HashType) (entry *dpaStoreReq) {
node = st
}
if node.entry.hash.isEqual(hash) {
if node.entry.Key.isEqual(hash) {
s.access_cnt++
node.update_access(s.access_cnt)
return node.entry
@ -233,70 +274,26 @@ func (s *dpaMemStorage) remove_oldest() {
}
// process store channel requests
func (s *dpaMemStorage) process_store(req *dpaStoreReq) {
if s.entry_cnt >= MaxEntries {
func (s *dpaMemStorage) Put(req *Chunk) error {
if s.entry_cnt >= maxEntries {
s.remove_oldest()
}
s.add(req)
if s.chain != nil {
s.chain.store_chn <- req
}
return nil
}
// process retrieve channel requests
func (s *dpaMemStorage) process_retrieve(req *dpaRetrieveReq) {
func (s *dpaMemStorage) Get(req *Chunk) {
entry := s.find(req.hash)
entry := s.find(req.Key)
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)
}
}
}
func (s *dpaMemStorage) Init() {
s.memtree = newTreeNode(memTreeFLW, nil, 0)
}

7
bzz/protocol.go Normal file
View file

@ -0,0 +1,7 @@
package bzz
/*
BZZ implements the bzz wire protocol of swarm
routing decoded storage and retrieval requests to DPA
and registering peers with the DHT
*/

24
ethdb/database.go
View file

@ -11,8 +11,8 @@ import (
)
type LDBDatabase struct {
db *leveldb.DB
comp bool
DB *leveldb.DB
Comp bool
}
func NewLDBDatabase(name string) (*LDBDatabase, error) {
@ -24,29 +24,29 @@ func NewLDBDatabase(name string) (*LDBDatabase, error) {
return nil, err
}
database := &LDBDatabase{db: db, comp: true}
database := &LDBDatabase{DB: db, Comp: true}
return database, nil
}
func (self *LDBDatabase) Put(key []byte, value []byte) {
if self.comp {
if self.Comp {
value = rle.Compress(value)
}
err := self.db.Put(key, value, nil)
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)
dat, err := self.DB.Get(key, nil)
if err != nil {
return nil, err
}
if self.comp {
if self.Comp {
return rle.Decompress(dat)
}
@ -54,7 +54,7 @@ func (self *LDBDatabase) Get(key []byte) ([]byte, error) {
}
func (self *LDBDatabase) Delete(key []byte) error {
return self.db.Delete(key, nil)
return self.DB.Delete(key, nil)
}
func (self *LDBDatabase) LastKnownTD() []byte {
@ -68,20 +68,20 @@ func (self *LDBDatabase) LastKnownTD() []byte {
}
func (self *LDBDatabase) NewIterator() iterator.Iterator {
return self.db.NewIterator(nil, nil)
return self.DB.NewIterator(nil, nil)
}
func (self *LDBDatabase) Write(batch *leveldb.Batch) error {
return self.db.Write(batch, nil)
return self.DB.Write(batch, nil)
}
func (self *LDBDatabase) Close() {
// Close the leveldb database
self.db.Close()
self.DB.Close()
}
func (self *LDBDatabase) Print() {
iter := self.db.NewIterator(nil, nil)
iter := self.DB.NewIterator(nil, nil)
for iter.Next() {
key := iter.Key()
value := iter.Value()