diff --git a/bzz/chunkIO.go b/bzz/chunkIO.go index e557ca6ac2..95d78d6470 100644 --- a/bzz/chunkIO.go +++ b/bzz/chunkIO.go @@ -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") diff --git a/bzz/chunker.go b/bzz/chunker.go index a571df209b..6235f264da 100644 --- a/bzz/chunker.go +++ b/bzz/chunker.go @@ -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 } diff --git a/bzz/chunker_test.go b/bzz/chunker_test.go index 62155ffe6b..84b50c72f3 100644 --- a/bzz/chunker_test.go +++ b/bzz/chunker_test.go @@ -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) diff --git a/bzz/dpa.go b/bzz/dpa.go index 376f22fdd8..d486244305 100644 --- a/bzz/dpa.go +++ b/bzz/dpa.go @@ -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 { diff --git a/bzz/memstore.go b/bzz/memstore.go index 8793d08c87..852cbe3134 100644 --- a/bzz/memstore.go +++ b/bzz/memstore.go @@ -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) - -}