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

memstore done with tests

Browse source 
- Chunk struct to dpa
- Data -> Reader, storage uses []byte (dpa should fill in, see memstore_test)
- use self.SplitTimeout
- assert self.Branches <= 1 || self.chunkSize <= 0 coming from no Init()
- make ChunkStore Get blocking, so interface changes to Get() (*Chunk, error)
- ChunkStore Put cannot error by contract
- clean up memstore, remove Init, introduce constructor
This commit is contained in:
zelig 2015-02-03 19:38:12 +01:00
parent aab16c49bb
commit 0785a9b461
5 changed files with 66 additions and 110 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
bzz/chunkIO.go
View file

@ -1,4 +1,3 @@
//
package bzz
import (
@ -78,7 +77,9 @@ func NewChunkReaderFromBytes(b []byte) *ChunkReader {
The following is adapted from io.SectionReader
*/
func (s *ChunkReader) Size() int64 { return s.limit - s.base }
func (s *ChunkReader) Size() int64 {
return s.limit - s.base
}
var errWhence = errors.New("Seek: invalid whence")
var errOffset = errors.New("Seek: invalid offset")

58
bzz/chunker.go
View file

@ -107,7 +107,7 @@ func (self *TreeChunker) Init() {
}
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)
// 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)
}
@ -115,29 +115,16 @@ func (self *TreeChunker) HashSize() int64 {
return self.hashSize
}
// Chunk serves also serves as a request object passed to ChunkStores
// in case it is a retrieval request, Data is nil and Size is 0
// Note that Size is not the size of the data chunk, which is Data.Size() see SectionReader
// but the size of the subtree encoded in the chunk
// 0 if request, to be supplied by the dpa
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
Key Key // always
C chan bool // to signal data delivery by the dpa
update bool //
}
// String() for pretty printing
func (self *Chunk) String() string {
var size int64
var slice []byte
var n int
var err error
if self.Data != nil {
if self.Reader != nil {
size = 32 // we are printing 32 bytes of the data
slice = make([]byte, size)
n, err = self.Data.ReadAt(slice, 0)
n, err = self.Reader.ReadAt(slice, 0)
if err != nil && err != io.EOF {
slice = []byte(fmt.Sprintf("ERROR: %v", err))
}
@ -159,9 +146,9 @@ func (self *TreeChunker) Split(key Key, data SectionReader, chunkC chan *Chunk)
wg := &sync.WaitGroup{}
errC = make(chan error)
rerrC := make(chan error)
timeout := time.After(splitTimeout)
timeout := time.After(self.SplitTimeout)
if key == nil {
//dpaLogger.Debugf("please allocate byte slice for root key")
// dpaLogger.Debugf("please allocate byte slice for root key")
return
}
wg.Add(1)
@ -173,11 +160,15 @@ func (self *TreeChunker) Split(key Key, data SectionReader, chunkC chan *Chunk)
// 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.
if self.Branches <= 1 || self.chunkSize <= 0 {
panic("chunker must be initialised")
}
for ; treeSize < size; treeSize *= self.Branches {
depth++
}
//dpaLogger.Debugf("split request received for data (%v bytes, depth: %v)", size, 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/self.Branches, key, data, chunkC, rerrC, wg)
@ -213,7 +204,7 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR
size := data.Size()
var newChunk *Chunk
var hash Key
//dpaLogger.Debugf("depth: %v, max subtree size: %v, data size: %v", depth, treeSize, size)
// dpaLogger.Debugf("depth: %v, max subtree size: %v, data size: %v", depth, treeSize, size)
for depth > 0 && size < treeSize {
treeSize /= self.Branches
@ -223,16 +214,16 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR
if depth == 0 {
// leaf nodes -> content chunks
hash = self.Hash(size, data)
//dpaLogger.Debugf("content chunk: max subtree size: %v, data size: %v", treeSize, size)
// dpaLogger.Debugf("content chunk: max subtree size: %v, data size: %v", treeSize, size)
newChunk = &Chunk{
Key: hash,
Data: data,
Size: size,
Key: hash,
Reader: data,
Size: size,
}
} else {
// intermediate chunk containing child nodes hashes
branches := int64((size-1)/treeSize) + 1
//dpaLogger.Debugf("intermediate node: setting branches: %v, depth: %v, max subtree size: %v, data size: %v", branches, depth, treeSize, size)
// 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
@ -263,9 +254,9 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR
chunkReader := NewChunkReaderFromBytes(chunk) // bytes.Reader almost implements SectionReader
hash = self.Hash(size, chunkReader)
newChunk = &Chunk{
Key: hash,
Data: chunkReader,
Size: size,
Key: hash,
Reader: chunkReader,
Size: size,
}
}
// send off new chunk to storage
@ -281,7 +272,7 @@ func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) (data LazySectionRead
// initialise return parameters
errC = make(chan error)
// timer to time out the operation (needed within so as to avoid process leakage)
timeout := time.After(joinTimeout)
timeout := time.After(self.JoinTimeout)
wg := &sync.WaitGroup{}
// initialise internal error channel
rerrC := make(chan error)
@ -357,7 +348,8 @@ func (self *TreeChunker) join(depth int, treeSize int64, key Key, chunkC chan *C
}
if depth == 0 {
return LazyReader(chunk.Data) // simply give back the chunks reader for content chunks
r = LazyReader(chunk.Reader)
return // simply give back the chunks reader for content chunks
}
// find appropriate block level
@ -372,7 +364,7 @@ func (self *TreeChunker) join(depth int, treeSize int64, key Key, chunkC chan *C
var readerF func() (r LazySectionReader)
var readerFs [](func() (r LazySectionReader))
branches := int64((chunk.Size-1)/treeSize) + 1
//dpaLogger.DebugDetailf("tree node - size %v, chunk size: %v, subtreeSize %v, branches %v", chunk.Size, chunk.Data.Size(), treeSize, branches)
// dpaLogger.DebugDetailf("tree node - size %v, chunk size: %v, subtreeSize %v, branches %v", chunk.Size, chunk.Reader.Size(), treeSize, branches)
// iterate through the chunk containing the keys of children
// create lazy init functions that give back readers
@ -380,8 +372,8 @@ func (self *TreeChunker) join(depth int, treeSize int64, key Key, chunkC chan *C
// create partial Chunk in order to send a retrieval request
childKey = make([]byte, self.hashSize) // preallocate hashSize long slice for key
// read the Hash of the subtree from the relevant section of the Chunk into the allocated byte slice in subtree.Key
if _, err := chunk.Data.ReadAt(childKey, i*self.hashSize); err != nil {
//dpaLogger.DebugDetailf("Read error: %v", err)
if _, err := chunk.Reader.ReadAt(childKey, i*self.hashSize); err != nil {
// dpaLogger.DebugDetailf("Read error: %v", err)
errC <- err
break
}

71
bzz/chunker_test.go
View file

@ -2,76 +2,23 @@ package bzz
import (
"bytes"
"crypto/rand"
"fmt"
"math/rand"
"testing"
"time"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/bzz/test"
)
/*
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
}
type benchLogger struct{ b *testing.B }
func benchlog(b *testing.B) benchLogger {
logger.Reset()
l := benchLogger{b}
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 (benchLogger) GetLogLevel() logger.LogLevel { return logger.Silence }
// func (benchLogger) GetLogLevel() logger.LogLevel { return logger.DebugLevel }
func (benchLogger) SetLogLevel(logger.LogLevel) {}
func (l benchLogger) LogPrint(level logger.LogLevel, msg string) {
l.b.Logf("%s", msg)
}
func (benchLogger) 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)
slice = make([]byte, l)
if _, err := rand.Read(slice); err != nil {
panic("rand error")
}
r = NewChunkReaderFromBytes(slice)
return
}
@ -163,7 +110,7 @@ func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int) (LazySecti
for _, ch := range self.chunks {
if bytes.Compare(chunk.Key, ch.Key) == 0 {
found = true
chunk.Data = ch.Data
chunk.Reader = ch.Reader
chunk.Size = ch.Size
close(chunk.C)
break
@ -206,7 +153,7 @@ func testRandomData(chunker *TreeChunker, tester *chunkerTester, n int, chunks i
}
func TestRandomData(t *testing.T) {
defer testlog(t).detach()
defer test.Testlog(t).Detach()
chunker := &TreeChunker{
Branches: 2,
SplitTimeout: 10 * time.Second,
@ -252,7 +199,7 @@ func benchmarkJoinRandomData(n int, chunks int, t *testing.B) {
}
func benchmarkSplitRandomData(n int, chunks int, t *testing.B) {
defer benchlog(t).detach()
defer test.Benchlog(t).Detach()
for i := 0; i < t.N; i++ {
chunker, tester := chunkerAndTester()
tester.Split(chunker, n)

25
bzz/dpa.go
View file

@ -41,9 +41,23 @@ type DPA struct {
quitC chan bool
}
// Chunk serves also serves as a request object passed to ChunkStores
// in case it is a retrieval request, Data is nil and Size is 0
// Note that Size is not the size of the data chunk, which is Data.Size() see SectionReader
// but the size of the subtree encoded in the chunk
// 0 if request, to be supplied by the dpa
type Chunk struct {
Reader SectionReader // nil if request, to be supplied by dpa
Data []byte // nil if request, to be supplied by dpa
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 {
Put(*Chunk) error
Get(*Chunk) error
Put(*Chunk) // effectively there is no error even if there is no error
Get() (*Chunk, error)
}
func (self *DPA) Retrieve(key Key) (data LazySectionReader, err error) {
@ -138,7 +152,7 @@ func (self *DPA) retrieveLoop() {
for chunk := range self.retrieveC {
go func() {
for _, store := range self.Stores {
if err := store.Get(chunk); err != nil { // no waiting/blocking here
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 {
@ -163,9 +177,8 @@ func (self *DPA) storeLoop() {
for chunk := range self.storeC {
go func() {
for _, store := range self.Stores {
if err := store.Put(chunk); err != nil { // no waiting/blocking here
dpaLogger.DebugDetailf("%v storing chunk %x: %v", store, chunk.Key, err)
} // no waiting/blocking here
store.Put(chunk)
// no waiting/blocking here
}
}()
select {

17
bzz/memstore.go
View file

@ -33,6 +33,12 @@ 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) {
m = &memStore{}
m.memtree = newMemTree(memTreeFLW, nil, 0)
return
}
func (x Key) Size() uint {
return uint(len(x))
}
@ -132,6 +138,7 @@ func (node *memTree) updateAccess(a uint64) {
}
func (s *memStore) Put(entry *Chunk) (err error) {
if s.entryCnt >= maxEntries {
s.removeOldest()
}
@ -192,13 +199,14 @@ func (s *memStore) Put(entry *Chunk) (err error) {
func (s *memStore) Get(chunk *Chunk) (err error) {
hash := chunk.Key
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
return notFound
}
bitpos += node.bits
node = st
@ -217,6 +225,7 @@ func (s *memStore) Get(chunk *Chunk) (err error) {
} else {
err = notFound
}
return
}
@ -288,9 +297,3 @@ func (s *memStore) removeOldest() {
}
}
func (s *memStore) Init() {
s.memtree = newMemTree(memTreeFLW, nil, 0)
}