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

Merge branch 'swarm' of git://github.com/ethereum/go-ethereum into swarm

Browse source
This commit is contained in:
Daniel A. Nagy 2016-10-26 09:12:17 +02:00
commit b01dfb24fa
4 changed files with 160 additions and 105 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

80
swarm/storage/chunker.go
View file

@ -1,3 +1,19 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package storage package storage
import ( import (
@ -7,8 +23,6 @@ import (
"hash" "hash"
"io" "io"
"sync" "sync"
// "github.com/ethereum/go-ethereum/logger"
// "github.com/ethereum/go-ethereum/logger/glog"
) )
/* /*
@ -108,12 +122,13 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s
jobC := make(chan *hashJob, 2*processors) jobC := make(chan *hashJob, 2*processors)
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
errC := make(chan error) errC := make(chan error)
quitC := make(chan bool)
// wwg = workers waitgroup keeps track of hashworkers spawned by this split call // wwg = workers waitgroup keeps track of hashworkers spawned by this split call
if wwg != nil { if wwg != nil {
wwg.Add(1) wwg.Add(1)
} }
go self.hashWorker(jobC, chunkC, errC, swg, wwg) go self.hashWorker(jobC, chunkC, errC, quitC, swg, wwg)
depth := 0 depth := 0
treeSize := self.chunkSize treeSize := self.chunkSize
@ -125,11 +140,10 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s
} }
key := make([]byte, self.hashFunc().Size()) key := make([]byte, self.hashFunc().Size())
// glog.V(logger.Detail).Infof("[BZZ] split request received for data (%v bytes, depth: %v)", size, depth)
// this waitgroup member is released after the root hash is calculated // this waitgroup member is released after the root hash is calculated
wg.Add(1) wg.Add(1)
//launch actual recursive function passing the waitgroups //launch actual recursive function passing the waitgroups
go self.split(depth, treeSize/self.branches, key, data, size, jobC, chunkC, errC, wg, swg, wwg) go self.split(depth, treeSize/self.branches, key, data, size, jobC, chunkC, errC, quitC, wg, swg, wwg)
// closes internal error channel if all subprocesses in the workgroup finished // closes internal error channel if all subprocesses in the workgroup finished
go func() { go func() {
@ -137,7 +151,6 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s
wg.Wait() wg.Wait()
// if storage waitgroup is non-nil, we wait for storage to finish too // if storage waitgroup is non-nil, we wait for storage to finish too
if swg != nil { if swg != nil {
// glog.V(logger.Detail).Infof("Waiting for storage to finish")
swg.Wait() swg.Wait()
} }
close(errC) close(errC)
@ -146,14 +159,15 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s
select { select {
case err := <-errC: case err := <-errC:
if err != nil { if err != nil {
close(quitC)
return nil, err return nil, err
} }
// //TODO: add a timeout
} }
return key, nil return key, nil
} }
func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reader, size int64, jobC chan *hashJob, chunkC chan *Chunk, errC chan error, parentWg, swg, wwg *sync.WaitGroup) { func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reader, size int64, jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, parentWg, swg, wwg *sync.WaitGroup) {
for depth > 0 && size < treeSize { for depth > 0 && size < treeSize {
treeSize /= self.branches treeSize /= self.branches
@ -164,17 +178,24 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade
// leaf nodes -> content chunks // leaf nodes -> content chunks
chunkData := make([]byte, size+8) chunkData := make([]byte, size+8)
binary.LittleEndian.PutUint64(chunkData[0:8], uint64(size)) binary.LittleEndian.PutUint64(chunkData[0:8], uint64(size))
data.Read(chunkData[8:]) var readBytes int64
select { for readBytes < size {
case jobC <- &hashJob{key, chunkData, size, parentWg}: n, err := data.Read(chunkData[8+readBytes:])
case <-errC: readBytes += int64(n)
} if err != nil && !(err == io.EOF && readBytes == size) {
// glog.V(logger.Detail).Infof("[BZZ] read %v", size) errC <- err
return return
} }
}
select {
case jobC <- &hashJob{key, chunkData, size, parentWg}:
case <-quitC:
}
return
}
// dept > 0
// intermediate chunk containing child nodes hashes // intermediate chunk containing child nodes hashes
branchCnt := int64((size + treeSize - 1) / treeSize) branchCnt := int64((size + treeSize - 1) / treeSize)
// glog.V(logger.Detail).Infof("[BZZ] intermediate node: setting branches: %v, depth: %v, max subtree size: %v, data size: %v", branches, depth, treeSize, size)
var chunk []byte = make([]byte, branchCnt*self.hashSize+8) var chunk []byte = make([]byte, branchCnt*self.hashSize+8)
var pos, i int64 var pos, i int64
@ -194,7 +215,7 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade
subTreeKey := chunk[8+i*self.hashSize : 8+(i+1)*self.hashSize] subTreeKey := chunk[8+i*self.hashSize : 8+(i+1)*self.hashSize]
childrenWg.Add(1) childrenWg.Add(1)
self.split(depth-1, treeSize/self.branches, subTreeKey, data, secSize, jobC, chunkC, errC, childrenWg, swg, wwg) self.split(depth-1, treeSize/self.branches, subTreeKey, data, secSize, jobC, chunkC, errC, quitC, childrenWg, swg, wwg)
i++ i++
pos += treeSize pos += treeSize
@ -208,15 +229,15 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade
wwg.Add(1) wwg.Add(1)
} }
self.workerCount++ self.workerCount++
go self.hashWorker(jobC, chunkC, errC, swg, wwg) go self.hashWorker(jobC, chunkC, errC, quitC, swg, wwg)
} }
select { select {
case jobC <- &hashJob{key, chunk, size, parentWg}: case jobC <- &hashJob{key, chunk, size, parentWg}:
case <-errC: case <-quitC:
} }
} }
func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC chan error, swg, wwg *sync.WaitGroup) { func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, swg, wwg *sync.WaitGroup) {
hasher := self.hashFunc() hasher := self.hashFunc()
if wwg != nil { if wwg != nil {
defer wwg.Done() defer wwg.Done()
@ -231,8 +252,7 @@ func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC
// now we got the hashes in the chunk, then hash the chunks // now we got the hashes in the chunk, then hash the chunks
hasher.Reset() hasher.Reset()
self.hashChunk(hasher, job, chunkC, swg) self.hashChunk(hasher, job, chunkC, swg)
// glog.V(logger.Detail).Infof("[BZZ] hash chunk (%v)", job.size) case <-quitC:
case <-errC:
return return
} }
} }
@ -260,6 +280,7 @@ func (self *TreeChunker) hashChunk(hasher hash.Hash, job *hashJob, chunkC chan *
} }
} }
job.parentWg.Done() job.parentWg.Done()
if chunkC != nil { if chunkC != nil {
chunkC <- newChunk chunkC <- newChunk
} }
@ -312,7 +333,6 @@ func (self *LazyChunkReader) Size(quitC chan bool) (n int64, err error) {
func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
// this is correct, a swarm doc cannot be zero length, so no EOF is expected // this is correct, a swarm doc cannot be zero length, so no EOF is expected
if len(b) == 0 { if len(b) == 0 {
// glog.V(logger.Detail).Infof("[BZZ] Size query for %v", chunk.Key.Log())
return 0, nil return 0, nil
} }
quitC := make(chan bool) quitC := make(chan bool)
@ -320,13 +340,10 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
if err != nil { if err != nil {
return 0, err return 0, err
} }
// glog.V(logger.Detail).Infof("readAt: len(b): %v, off: %v, size: %v ", len(b), off, size)
errC := make(chan error) errC := make(chan error)
// glog.V(logger.Detail).Infof("[BZZ] readAt: reading %v into %d bytes at offset %d.", self.chunk.Key.Log(), len(b), off)
// } // }
// glog.V(logger.Detail).Infof("-> want: %v, off: %v size: %v ", want, off, self.size)
var treeSize int64 var treeSize int64
var depth int var depth int
// calculate depth and max treeSize // calculate depth and max treeSize
@ -348,22 +365,15 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
return 0, err return 0, err
} }
// glog.V(logger.Detail).Infof("[BZZ] ReadAt received %v", err)
// glog.V(logger.Detail).Infof("end: len(b): %v, off: %v, size: %v ", len(b), off, size)
if off+int64(len(b)) >= size { if off+int64(len(b)) >= size {
// glog.V(logger.Detail).Infof(" len(b): %v EOF", len(b))
return len(b), io.EOF return len(b), io.EOF
} }
// glog.V(logger.Detail).Infof("[BZZ] ReadAt returning at %d: %v", read, err)
return len(b), nil return len(b), nil
} }
func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) { func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) {
defer parentWg.Done() defer parentWg.Done()
// return NewDPA(&LocalStore{}) // return NewDPA(&LocalStore{})
// glog.V(logger.Detail).Infof("inh len(b): %v, off: %v eoff: %v ", len(b), off, eoff)
// glog.V(logger.Detail).Infof("[BZZ] depth: %v, loff: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, off, eoff, chunk.Size, treeSize)
// chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
@ -375,7 +385,6 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
// leaf chunk found // leaf chunk found
if depth == 0 { if depth == 0 {
// glog.V(logger.Detail).Infof("[BZZ] depth: %v, len(b): %v, off: %v, eoff: %v, chunk.Size: %v %v, treeSize: %v", depth, len(b), off, eoff, chunk.Size, len(chunk.SData), treeSize)
extra := 8 + eoff - int64(len(chunk.SData)) extra := 8 + eoff - int64(len(chunk.SData))
if extra > 0 { if extra > 0 {
eoff -= extra eoff -= extra
@ -390,7 +399,6 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
defer wg.Wait() defer wg.Wait()
// glog.V(logger.Detail).Infof("[BZZ] start %v,end %v", start, end)
for i := start; i < end; i++ { for i := start; i < end; i++ {
soff := i * treeSize soff := i * treeSize
@ -409,7 +417,6 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
wg.Add(1) wg.Add(1)
go func(j int64) { go func(j int64) {
childKey := chunk.SData[8+j*self.hashSize : 8+(j+1)*self.hashSize] childKey := chunk.SData[8+j*self.hashSize : 8+(j+1)*self.hashSize]
// glog.V(logger.Detail).Infof("[BZZ] subtree ind.ex: %v -> %v", j, childKey.Log())
chunk := retrieve(childKey, self.chunkC, quitC) chunk := retrieve(childKey, self.chunkC, quitC)
if chunk == nil { if chunk == nil {
select { select {
@ -434,7 +441,6 @@ func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk {
Key: key, Key: key,
C: make(chan bool), // close channel to signal data delivery C: make(chan bool), // close channel to signal data delivery
} }
// glog.V(logger.Detail).Infof("[BZZ] chunk data sent for %v (key interval in chunk %v-%v)", ch.Key.Log(), j*self.chunker.hashSize, (j+1)*self.chunker.hashSize)
// submit chunk for retrieval // submit chunk for retrieval
select { select {
case chunkC <- chunk: // submit retrieval request, someone should be listening on the other side (or we will time out globally) case chunkC <- chunk: // submit retrieval request, someone should be listening on the other side (or we will time out globally)
@ -448,7 +454,6 @@ func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk {
// this is how we control process leakage (quitC is closed once join is finished (after timeout)) // this is how we control process leakage (quitC is closed once join is finished (after timeout))
return nil return nil
case <-chunk.C: // bells are ringing, data have been delivered case <-chunk.C: // bells are ringing, data have been delivered
// glog.V(logger.Detail).Infof("[BZZ] chunk data received")
} }
if len(chunk.SData) == 0 { if len(chunk.SData) == 0 {
return nil // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) return nil // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
@ -460,7 +465,6 @@ func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk {
// Read keeps a cursor so cannot be called simulateously, see ReadAt // Read keeps a cursor so cannot be called simulateously, see ReadAt
func (self *LazyChunkReader) Read(b []byte) (read int, err error) { func (self *LazyChunkReader) Read(b []byte) (read int, err error) {
read, err = self.ReadAt(b, self.off) read, err = self.ReadAt(b, self.off)
// glog.V(logger.Detail).Infof("[BZZ] read: %v, off: %v, error: %v", read, self.off, err)
self.off += int64(read) self.off += int64(read)
return return

93
swarm/storage/chunker_test.go
View file

@ -1,7 +1,24 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package storage package storage
import ( import (
"bytes" "bytes"
"crypto/rand"
"encoding/binary" "encoding/binary"
"fmt" "fmt"
"io" "io"
@ -11,6 +28,7 @@ import (
"time" "time"
) )
/* /*
Tests TreeChunker by splitting and joining a random byte slice Tests TreeChunker by splitting and joining a random byte slice
*/ */
@ -33,7 +51,7 @@ func (self *chunkerTester) checkChunks(t *testing.T, want int) {
} }
} }
func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, chunkC chan *Chunk, swg *sync.WaitGroup) (key Key) { func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, chunkC chan *Chunk, swg *sync.WaitGroup, expectedError error) (key Key) {
// reset // reset
self.chunks = make(map[string]*Chunk) self.chunks = make(map[string]*Chunk)
@ -49,14 +67,9 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c
select { select {
case <-timeout: case <-timeout:
self.t.Fatalf("Join timeout error") self.t.Fatalf("Join timeout error")
case <-quitC:
case chunk, ok := <-chunkC:
if !ok {
// glog.V(logger.Info).Infof("chunkC closed quitting")
close(quitC)
return return
} case chunk := <-chunkC:
// glog.V(logger.Info).Infof("chunk %v received", len(self.chunks))
// self.chunks = append(self.chunks, chunk) // self.chunks = append(self.chunks, chunk)
self.chunks[chunk.Key.String()] = chunk self.chunks[chunk.Key.String()] = chunk
if chunk.wg != nil { if chunk.wg != nil {
@ -67,21 +80,16 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c
}() }()
} }
key, err := chunker.Split(data, size, chunkC, swg, nil) key, err := chunker.Split(data, size, chunkC, swg, nil)
if err != nil { if err != nil && expectedError == nil {
self.t.Fatalf("Split error: %v", err) self.t.Fatalf("Split error: %v", err)
} else if expectedError != nil && (err == nil || err.Error() != expectedError.Error()) {
self.t.Fatalf("Not receiving the correct error! Expected %v, received %v", expectedError, err)
} }
if chunkC != nil { if chunkC != nil {
if swg != nil { if swg != nil {
// glog.V(logger.Info).Infof("Waiting for storage to finish")
swg.Wait() swg.Wait()
// glog.V(logger.Info).Infof("Storage finished")
} }
close(chunkC) close(quitC)
}
if chunkC != nil {
// glog.V(logger.Info).Infof("waiting for splitter finished")
<-quitC
// glog.V(logger.Info).Infof("Splitter finished")
} }
return return
} }
@ -89,11 +97,9 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c
func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Chunk, quitC chan bool) LazySectionReader { func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Chunk, quitC chan bool) LazySectionReader {
// reset but not the chunks // reset but not the chunks
// glog.V(logger.Info).Infof("Splitter finished")
reader := chunker.Join(key, chunkC) reader := chunker.Join(key, chunkC)
timeout := time.After(600 * time.Second) timeout := time.After(600 * time.Second)
// glog.V(logger.Info).Infof("Splitter finished")
i := 0 i := 0
go func() { go func() {
for { for {
@ -106,15 +112,12 @@ func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Ch
close(quitC) close(quitC)
return return
} }
// glog.V(logger.Info).Infof("chunk %v: %v", i, chunk.Key.String())
// this just mocks the behaviour of a chunk store retrieval // this just mocks the behaviour of a chunk store retrieval
stored, success := self.chunks[chunk.Key.String()] stored, success := self.chunks[chunk.Key.String()]
// glog.V(logger.Info).Infof("chunk %v, success: %v", chunk.Key.String(), success)
if !success { if !success {
self.t.Fatalf("not found") self.t.Fatalf("not found")
return return
} }
// glog.V(logger.Info).Infof("chunk %v: %v", i, chunk.Key.String())
chunk.SData = stored.SData chunk.SData = stored.SData
chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
close(chunk.C) close(chunk.C)
@ -125,6 +128,26 @@ func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Ch
return reader return reader
} }
func testRandomBrokenData(splitter Splitter, n int, tester *chunkerTester) {
data := io.LimitReader(rand.Reader, int64(n))
brokendata := brokenLimitReader(data, n, n/2)
buf := make([]byte, n)
_, err := brokendata.Read(buf)
if err == nil || err.Error() != "Broken reader" {
tester.t.Fatalf("Broken reader is not broken, hence broken. Returns: %v", err)
}
data = io.LimitReader(rand.Reader, int64(n))
brokendata = brokenLimitReader(data, n, n/2)
chunkC := make(chan *Chunk, 1000)
swg := &sync.WaitGroup{}
key := tester.Split(splitter, brokendata, int64(n), chunkC, swg, fmt.Errorf("Broken reader"))
tester.t.Logf(" Key = %v\n", key)
}
func testRandomData(splitter Splitter, n int, tester *chunkerTester) { func testRandomData(splitter Splitter, n int, tester *chunkerTester) {
if tester.inputs == nil { if tester.inputs == nil {
tester.inputs = make(map[uint64][]byte) tester.inputs = make(map[uint64][]byte)
@ -135,13 +158,13 @@ func testRandomData(splitter Splitter, n int, tester *chunkerTester) {
data, input = testDataReaderAndSlice(n) data, input = testDataReaderAndSlice(n)
tester.inputs[uint64(n)] = input tester.inputs[uint64(n)] = input
} else { } else {
data = limitReader(bytes.NewReader(input), n) data = io.LimitReader(bytes.NewReader(input), int64(n))
} }
chunkC := make(chan *Chunk, 1000) chunkC := make(chan *Chunk, 1000)
swg := &sync.WaitGroup{} swg := &sync.WaitGroup{}
key := tester.Split(splitter, data, int64(n), chunkC, swg) key := tester.Split(splitter, data, int64(n), chunkC, swg, nil)
tester.t.Logf(" Key = %v\n", key) tester.t.Logf(" Key = %v\n", key)
chunkC = make(chan *Chunk, 1000) chunkC = make(chan *Chunk, 1000)
@ -150,9 +173,7 @@ func testRandomData(splitter Splitter, n int, tester *chunkerTester) {
chunker := NewTreeChunker(NewChunkerParams()) chunker := NewTreeChunker(NewChunkerParams())
reader := tester.Join(chunker, key, 0, chunkC, quitC) reader := tester.Join(chunker, key, 0, chunkC, quitC)
output := make([]byte, n) output := make([]byte, n)
// glog.V(logger.Info).Infof(" Key = %v\n", key)
r, err := reader.Read(output) r, err := reader.Read(output)
// glog.V(logger.Info).Infof(" read = %v %v\n", r, err)
if r != n || err != io.EOF { if r != n || err != io.EOF {
tester.t.Fatalf("read error read: %v n = %v err = %v\n", r, n, err) tester.t.Fatalf("read error read: %v n = %v err = %v\n", r, n, err)
} }
@ -167,7 +188,7 @@ func testRandomData(splitter Splitter, n int, tester *chunkerTester) {
func TestRandomData(t *testing.T) { func TestRandomData(t *testing.T) {
// sizes := []int{123456} // sizes := []int{123456}
sizes := []int{1, 60, 83, 179, 253, 1024, 4095, 4096, 4097, 123456} sizes := []int{1, 60, 83, 179, 253, 1024, 4095, 4096, 4097, 8191, 8192, 8193, 123456, 2345678}
tester := &chunkerTester{t: t} tester := &chunkerTester{t: t}
chunker := NewTreeChunker(NewChunkerParams()) chunker := NewTreeChunker(NewChunkerParams())
for _, s := range sizes { for _, s := range sizes {
@ -179,6 +200,16 @@ func TestRandomData(t *testing.T) {
} }
} }
func TestRandomBrokenData(t *testing.T) {
sizes := []int{1, 60, 83, 179, 253, 1024, 4095, 4096, 4097, 8191, 8192, 8193, 123456, 2345678}
tester := &chunkerTester{t: t}
chunker := NewTreeChunker(NewChunkerParams())
for _, s := range sizes {
testRandomBrokenData(chunker, s, tester)
t.Logf("done size: %v", s)
}
}
func readAll(reader LazySectionReader, result []byte) { func readAll(reader LazySectionReader, result []byte) {
size := int64(len(result)) size := int64(len(result))
@ -211,7 +242,7 @@ func benchmarkJoin(n int, t *testing.B) {
chunkC := make(chan *Chunk, 1000) chunkC := make(chan *Chunk, 1000)
swg := &sync.WaitGroup{} swg := &sync.WaitGroup{}
key := tester.Split(chunker, data, int64(n), chunkC, swg) key := tester.Split(chunker, data, int64(n), chunkC, swg, nil)
// t.StartTimer() // t.StartTimer()
chunkC = make(chan *Chunk, 1000) chunkC = make(chan *Chunk, 1000)
quitC := make(chan bool) quitC := make(chan bool)
@ -232,8 +263,7 @@ func benchmarkSplitTree(n int, t *testing.B) {
chunker := NewTreeChunker(NewChunkerParams()) chunker := NewTreeChunker(NewChunkerParams())
tester := &chunkerTester{t: t} tester := &chunkerTester{t: t}
data := testDataReader(n) data := testDataReader(n)
// glog.V(logger.Info).Infof("splitting data of length %v", n) tester.Split(chunker, data, int64(n), nil, nil, nil)
tester.Split(chunker, data, int64(n), nil, nil)
} }
stats := new(runtime.MemStats) stats := new(runtime.MemStats)
runtime.ReadMemStats(stats) runtime.ReadMemStats(stats)
@ -246,8 +276,7 @@ func benchmarkSplitPyramid(n int, t *testing.B) {
splitter := NewPyramidChunker(NewChunkerParams()) splitter := NewPyramidChunker(NewChunkerParams())
tester := &chunkerTester{t: t} tester := &chunkerTester{t: t}
data := testDataReader(n) data := testDataReader(n)
// glog.V(logger.Info).Infof("splitting data of length %v", n) tester.Split(splitter, data, int64(n), nil, nil, nil)
tester.Split(splitter, data, int64(n), nil, nil)
} }
stats := new(runtime.MemStats) stats := new(runtime.MemStats)
runtime.ReadMemStats(stats) runtime.ReadMemStats(stats)

64
swarm/storage/common_test.go
View file

@ -1,8 +1,25 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package storage package storage
import ( import (
"bytes" "bytes"
"crypto/rand" "crypto/rand"
"fmt"
"io" "io"
"sync" "sync"
"testing" "testing"
@ -11,32 +28,31 @@ import (
"github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/logger/glog"
) )
type limitedReader struct { type brokenLimitedReader struct {
r io.Reader lr io.Reader
off int64 errAt int
size int64 off int
size int
} }
func limitReader(r io.Reader, size int) *limitedReader { func brokenLimitReader(data io.Reader, size int, errAt int) *brokenLimitedReader {
return &limitedReader{r, 0, int64(size)} return &brokenLimitedReader{
lr: data,
errAt: errAt,
size: size,
} }
func (self *limitedReader) Read(buf []byte) (int, error) {
limit := int64(len(buf))
left := self.size - self.off
if limit >= left {
limit = left
}
n, err := self.r.Read(buf[:limit])
if err == nil && limit == left {
err = io.EOF
}
self.off += int64(n)
return n, err
} }
func testDataReader(l int) (r io.Reader) { func testDataReader(l int) (r io.Reader) {
return limitReader(rand.Reader, l) return io.LimitReader(rand.Reader, int64(l))
}
func (self *brokenLimitedReader) Read(buf []byte) (int, error) {
if self.off+len(buf) > self.errAt {
return 0, fmt.Errorf("Broken reader")
}
self.off += len(buf)
return self.lr.Read(buf)
} }
func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) { func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) {
@ -44,7 +60,7 @@ func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) {
if _, err := rand.Read(slice); err != nil { if _, err := rand.Read(slice); err != nil {
panic("rand error") panic("rand error")
} }
r = limitReader(bytes.NewReader(slice), l) r = io.LimitReader(bytes.NewReader(slice), int64(l))
return return
} }
@ -76,14 +92,14 @@ func testStore(m ChunkStore, l int64, branches int64, t *testing.T) {
go func(chunk *Chunk) { go func(chunk *Chunk) {
storedChunk, err := m.Get(chunk.Key) storedChunk, err := m.Get(chunk.Key)
if err == notFound { if err == notFound {
glog.V(logger.Detail).Infof("[BZZ] chunk '%v' not found", chunk.Key.Log()) glog.V(logger.Detail).Infof("chunk '%v' not found", chunk.Key.Log())
} else if err != nil { } else if err != nil {
glog.V(logger.Detail).Infof("[BZZ] error retrieving chunk %v: %v", chunk.Key.Log(), err) glog.V(logger.Detail).Infof("error retrieving chunk %v: %v", chunk.Key.Log(), err)
} else { } else {
chunk.SData = storedChunk.SData chunk.SData = storedChunk.SData
chunk.Size = storedChunk.Size chunk.Size = storedChunk.Size
} }
glog.V(logger.Detail).Infof("[BZZ] chunk '%v' not found", chunk.Key.Log()) glog.V(logger.Detail).Infof("chunk '%v' not found", chunk.Key.Log())
close(chunk.C) close(chunk.C)
}(ch) }(ch)
} }

32
swarm/storage/pyramid.go
View file

@ -1,3 +1,19 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package storage package storage
import ( import (
@ -65,7 +81,6 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk
chunks := (size + self.chunkSize - 1) / self.chunkSize chunks := (size + self.chunkSize - 1) / self.chunkSize
depth := int(math.Ceil(math.Log(float64(chunks))/math.Log(float64(self.branches)))) + 1 depth := int(math.Ceil(math.Log(float64(chunks))/math.Log(float64(self.branches)))) + 1
// glog.V(logger.Detail).Infof("chunks: %v, depth: %v", chunks, depth)
results := Tree{ results := Tree{
Chunks: chunks, Chunks: chunks,
@ -83,26 +98,24 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk
go self.processor(pend, swg, tasks, chunkC, &results) go self.processor(pend, swg, tasks, chunkC, &results)
} }
// Feed the chunks into the task pool // Feed the chunks into the task pool
read := 0
for index := 0; ; index++ { for index := 0; ; index++ {
buffer := make([]byte, self.chunkSize+8) buffer := make([]byte, self.chunkSize+8)
n, err := data.Read(buffer[8:]) n, err := data.Read(buffer[8:])
last := err == io.ErrUnexpectedEOF || err == io.EOF read += n
// glog.V(logger.Detail).Infof("n: %v, index: %v, depth: %v", n, index, depth) last := int64(read) == size || err == io.ErrUnexpectedEOF || err == io.EOF
if err != nil && !last { if err != nil && !last {
// glog.V(logger.Info).Infof("error: %v", err)
close(abortC) close(abortC)
break break
} }
binary.LittleEndian.PutUint64(buffer[:8], uint64(n)) binary.LittleEndian.PutUint64(buffer[:8], uint64(n))
pend.Add(1) pend.Add(1)
// glog.V(logger.Info).Infof("-> task %v (%v)", index, n)
select { select {
case tasks <- &Task{Index: int64(index), Size: uint64(n), Data: buffer[:n+8], Last: last}: case tasks <- &Task{Index: int64(index), Size: uint64(n), Data: buffer[:n+8], Last: last}:
case <-abortC: case <-abortC:
return nil, err return nil, err
} }
if last { if last {
// glog.V(logger.Info).Infof("last task %v (%v)", index, n)
break break
} }
} }
@ -110,7 +123,6 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk
close(tasks) close(tasks)
pend.Wait() pend.Wait()
// glog.V(logger.Info).Infof("len: %v", results.Levels[0][0])
key := results.Levels[0][0].Children[0][:] key := results.Levels[0][0].Children[0][:]
return key, nil return key, nil
} }
@ -118,12 +130,10 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk
func (self *PyramidChunker) processor(pend, swg *sync.WaitGroup, tasks chan *Task, chunkC chan *Chunk, results *Tree) { func (self *PyramidChunker) processor(pend, swg *sync.WaitGroup, tasks chan *Task, chunkC chan *Chunk, results *Tree) {
defer pend.Done() defer pend.Done()
// glog.V(logger.Info).Infof("processor started")
// Start processing leaf chunks ad infinitum // Start processing leaf chunks ad infinitum
hasher := self.hashFunc() hasher := self.hashFunc()
for task := range tasks { for task := range tasks {
depth, pow := len(results.Levels)-1, self.branches depth, pow := len(results.Levels)-1, self.branches
// glog.V(logger.Info).Infof("task: %v, last: %v", task.Index, task.Last)
size := task.Size size := task.Size
data := task.Data data := task.Data
var node *Node var node *Node
@ -155,10 +165,8 @@ func (self *PyramidChunker) processor(pend, swg *sync.WaitGroup, tasks chan *Tas
} }
node = &Node{pending, 0, make([]common.Hash, pending), last} node = &Node{pending, 0, make([]common.Hash, pending), last}
results.Levels[depth][task.Index/pow] = node results.Levels[depth][task.Index/pow] = node
// glog.V(logger.Info).Infof("create node %v, %v (%v children, all pending)", depth, task.Index/pow, pending)
} }
node.Pending-- node.Pending--
// glog.V(logger.Info).Infof("pending now: %v", node.Pending)
i := task.Index / (pow / self.branches) % self.branches i := task.Index / (pow / self.branches) % self.branches
if last { if last {
node.Last = true node.Last = true
@ -166,7 +174,6 @@ func (self *PyramidChunker) processor(pend, swg *sync.WaitGroup, tasks chan *Tas
copy(node.Children[i][:], hash) copy(node.Children[i][:], hash)
node.Size += size node.Size += size
left := node.Pending left := node.Pending
// glog.V(logger.Info).Infof("left pending now: %v, node size: %v", left, node.Size)
if chunkC != nil { if chunkC != nil {
if swg != nil { if swg != nil {
swg.Add(1) swg.Add(1)
@ -182,7 +189,6 @@ func (self *PyramidChunker) processor(pend, swg *sync.WaitGroup, tasks chan *Tas
results.Lock.Unlock() results.Lock.Unlock()
// If there's more work to be done, leave for others // If there's more work to be done, leave for others
// glog.V(logger.Info).Infof("left %v", left)
if left > 0 { if left > 0 {
break break
} }