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

swarm/storage:

Browse source 
* optimise splitter
* simplify reader
* benchmarks, tests improved
* simplify IO, remove chunkreader.go and io.SectionReader clone
* simplify error/timeout handling
* port pyramid splitter by karalabe, adapt to Splitter/DPA interface, rework params etc
This commit is contained in:
zelig 2016-06-06 23:39:44 +01:00
parent 6bff0b3f7f
commit 74cc29789a
15 changed files with 608 additions and 525 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
swarm/api/api.go
View file

@ -47,8 +47,8 @@ func (self *Api) Retrieve(key storage.Key) storage.SectionReader {
return self.dpa.Retrieve(key)
}
func (self *Api) Store(data storage.SectionReader, wg *sync.WaitGroup) (key storage.Key, err error) {
return self.dpa.Store(data, wg)
func (self *Api) Store(data io.Reader, size int64, wg *sync.WaitGroup) (key storage.Key, err error) {
return self.dpa.Store(data, size, wg)
}
type ErrResolve error
@ -105,15 +105,15 @@ func (self *Api) parseAndResolve(uri string, nameresolver bool) (contentHash sto
// Put provides singleton manifest creation on top of dpa store
func (self *Api) Put(content, contentType string) (string, error) {
sr := io.NewSectionReader(strings.NewReader(content), 0, int64(len(content)))
r := strings.NewReader(content)
wg := &sync.WaitGroup{}
key, err := self.dpa.Store(sr, wg)
key, err := self.dpa.Store(r, int64(len(content)), wg)
if err != nil {
return "", err
}
manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType)
sr = io.NewSectionReader(strings.NewReader(manifest), 0, int64(len(manifest)))
key, err = self.dpa.Store(sr, wg)
r = strings.NewReader(manifest)
key, err = self.dpa.Store(r, int64(len(manifest)), wg)
if err != nil {
return "", err
}

2
swarm/api/config_test.go
View file

@ -20,8 +20,6 @@ var (
"Radius": 0,
"Branches": 128,
"Hash": "SHA256",
"JoinTimeout": 120,
"SplitTimeout": 120,
"CallInterval": 3000000000,
"KadDbPath": "` + filepath.Join("TMPDIR", "0d2f62485607cf38d9d795d93682a517661e513e", "bzz-peers.json") + `",
"MaxProx": 8,

5
swarm/api/filesystem.go
View file

@ -96,17 +96,16 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
f, err := os.Open(entry.Path)
if err == nil {
stat, _ := f.Stat()
sr := io.NewSectionReader(f, 0, stat.Size())
wg := &sync.WaitGroup{}
var hash storage.Key
hash, err = self.api.dpa.Store(sr, wg)
hash, err = self.api.dpa.Store(f, stat.Size(), wg)
if hash != nil {
list[i].Hash = hash.String()
}
wg.Wait()
if err == nil {
first512 := make([]byte, 512)
fread, _ := sr.ReadAt(first512, 0)
fread, _ := f.ReadAt(first512, 0)
if fread > 0 {
mimeType := http.DetectContentType(first512[:fread])
if filepath.Ext(entry.Path) == ".css" {

5
swarm/api/http/server.go
View file

@ -96,10 +96,7 @@ func handler(w http.ResponseWriter, r *http.Request, a *api.Api) {
switch {
case r.Method == "POST" || r.Method == "PUT":
key, err := a.Store(io.NewSectionReader(&sequentialReader{
reader: r.Body,
ahead: make(map[int64]chan bool),
}, 0, r.ContentLength), nil)
key, err := a.Store(r.Body, r.ContentLength, nil)
if err == nil {
glog.V(logger.Debug).Infof("[BZZ] Swarm: Content for %v stored", key.Log())
} else {

5
swarm/api/manifest.go
View file

@ -4,7 +4,6 @@ import (
"bytes"
"encoding/json"
"fmt"
"io"
"sync"
"github.com/ethereum/go-ethereum/common"
@ -198,9 +197,9 @@ func (self *manifestTrie) recalcAndStore() error {
return err
}
sr := io.NewSectionReader(bytes.NewReader(manifest), 0, int64(len(manifest)))
sr := bytes.NewReader(manifest)
wg := &sync.WaitGroup{}
key, err2 := self.dpa.Store(sr, wg)
key, err2 := self.dpa.Store(sr, int64(len(manifest)), wg)
wg.Wait()
self.hash = key
return err2

11
swarm/cmd/bzzhash/bzzhash.go
View file

@ -3,7 +3,6 @@ package main
import (
"fmt"
"io"
"os"
"runtime"
@ -24,15 +23,11 @@ func main() {
}
stat, _ := f.Stat()
sr := io.NewSectionReader(f, 0, stat.Size())
chunker := storage.NewTreeChunker(storage.NewChunkerParams())
hash := make([]byte, chunker.KeySize())
errC := chunker.Split(hash, sr, nil, nil)
err, ok := <-errC
key, err := chunker.Split(f, stat.Size(), nil, nil, nil)
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
}
if !ok {
fmt.Printf("%064x\n", hash)
} else {
fmt.Printf("%v\n", key)
}
}

2
swarm/cmd/swarm/swarm.sh
View file

@ -336,6 +336,8 @@ case $cmd in
up $*;;
"down" )
down $*;;
"download" )
download $*;;
"init" )
init $*;;
"start" )

311
swarm/storage/chunker.go
View file

@ -2,10 +2,11 @@ package storage
import (
"encoding/binary"
"errors"
"fmt"
"hash"
"io"
"sync"
"time"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
@ -37,11 +38,9 @@ data_{i} := size(subtree_{i}) || key_{j} || key_{j+1} .... || key_{j+n-1}
*/
const (
// defaultHash = "SHA3" // http://golang.org/pkg/hash/#Hash
// defaultHash = "SHA3" // http://golang.org/pkg/hash/#Hash
defaultHash = "SHA256" // http://golang.org/pkg/hash/#Hash
defaultBranches int64 = 128
joinTimeout = 120 // second
splitTimeout = 120 // second
// hashSize int64 = hasherfunc.New().Size() // hasher knows about its own length in bytes
// chunksize int64 = branches * hashSize // chunk is defined as this
)
@ -55,130 +54,107 @@ The hashing itself does use extra copies and allocation though, since it does ne
*/
type ChunkerParams struct {
Branches int64
Hash string
JoinTimeout time.Duration
SplitTimeout time.Duration
Branches int64
Hash string
}
func NewChunkerParams() *ChunkerParams {
return &ChunkerParams{
Branches: defaultBranches,
Hash: defaultHash,
JoinTimeout: joinTimeout,
SplitTimeout: splitTimeout,
Branches: defaultBranches,
Hash: defaultHash,
}
}
type TreeChunker struct {
branches int64
hashFunc Hasher
joinTimeout time.Duration
splitTimeout time.Duration
branches int64
hashFunc Hasher
// calculated
hashSize int64 // self.hashFunc.New().Size()
chunkSize int64 // hashSize* branches
hashSize int64 // self.hashFunc.New().Size()
chunkSize int64 // hashSize* branches
workerCount int
}
func NewTreeChunker(params *ChunkerParams) (self *TreeChunker) {
self = &TreeChunker{}
self.hashFunc = MakeHashFunc(params.Hash)
self.branches = params.Branches
self.joinTimeout = params.JoinTimeout * time.Second
self.splitTimeout = params.SplitTimeout * time.Second
self.hashSize = int64(self.hashFunc().Size())
self.chunkSize = self.hashSize * self.branches
self.workerCount = 1
return
}
func (self *TreeChunker) KeySize() int64 {
return self.hashSize
}
// func (self *TreeChunker) KeySize() int64 {
// return self.hashSize
// }
// String() for pretty printing
func (self *Chunk) String() string {
return fmt.Sprintf("Key: %v TreeSize: %v Chunksize: %v", self.Key.Log(), self.Size, len(self.SData))
}
// 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(input []byte) []byte {
hasher := self.hashFunc()
hasher.Write(input)
return hasher.Sum(nil)
type hashJob struct {
key Key
chunk []byte
size int64
parentWg *sync.WaitGroup
}
func (self *TreeChunker) Split(key Key, data SectionReader, chunkC chan *Chunk, swg *sync.WaitGroup) (errC chan error) {
if swg != nil {
swg.Add(1)
defer swg.Done()
}
func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, swg, wwg *sync.WaitGroup) (Key, error) {
if self.chunkSize <= 0 {
panic("chunker must be initialised")
}
if int64(len(key)) != self.hashSize {
panic(fmt.Sprintf("root key buffer must be allocated byte slice of length %d", self.hashSize))
jobC := make(chan *hashJob, 2*processors)
wg := &sync.WaitGroup{}
errC := make(chan error)
// wwg = workers waitgroup keeps track of hashworkers spawned by this split call
if wwg != nil {
wwg.Add(1)
}
go self.hashWorker(jobC, chunkC, errC, swg, wwg)
depth := 0
treeSize := self.chunkSize
// 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++
}
wg := &sync.WaitGroup{}
errC = make(chan error)
rerrC := make(chan error)
timeout := time.After(self.splitTimeout)
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
wg.Add(1)
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++
}
// glog.V(logger.Detail).Infof("[BZZ] 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, swg)
}()
//launch actual recursive function passing the workgroup
go self.split(depth, treeSize/self.branches, key, data, size, jobC, chunkC, errC, wg, swg, wwg)
// closes internal error channel if all subprocesses in the workgroup finished
go func() {
// waiting for all threads to finish
wg.Wait()
close(rerrC)
}()
// waiting for request to end with wg finishing, error, or timeout
go func() {
select {
case err := <-rerrC:
if err != nil {
errC <- err
} // otherwise splitting is complete
case <-timeout:
errC <- fmt.Errorf("split time out")
// if storage waitgroup is non-nil, we wait for storage to finish too
if swg != nil {
// glog.V(logger.Detail).Infof("Waiting for storage to finish")
swg.Wait()
}
close(errC)
}()
return
select {
case err := <-errC:
if err != nil {
return nil, err
}
//
}
return key, nil
}
func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionReader, chunkC chan *Chunk, errc chan error, parentWg *sync.WaitGroup, swg *sync.WaitGroup) {
defer parentWg.Done()
size := data.Size()
var newChunk *Chunk
var hash Key
// glog.V(logger.Detail).Infof("[BZZ] depth: %v, max subtree size: %v, data size: %v", depth, treeSize, size)
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) {
for depth > 0 && size < treeSize {
treeSize /= self.branches
@ -187,71 +163,110 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR
if depth == 0 {
// leaf nodes -> content chunks
chunkData := make([]byte, data.Size()+8)
chunkData := make([]byte, size+8)
binary.LittleEndian.PutUint64(chunkData[0:8], uint64(size))
data.ReadAt(chunkData[8:], 0)
hash = self.Hash(chunkData)
// glog.V(logger.Detail).Infof("[BZZ] content chunk: max subtree size: %v, data size: %v", treeSize, size)
newChunk = &Chunk{
Key: hash,
SData: chunkData,
Size: size,
data.Read(chunkData[8:])
select {
case jobC <- &hashJob{key, chunkData, size, parentWg}:
case <-errC:
}
} else {
// intermediate chunk containing child nodes hashes
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)
// glog.V(logger.Detail).Infof("[BZZ] read %v", size)
return
}
// intermediate chunk containing child nodes hashes
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 pos, i int64
var chunk []byte = make([]byte, branchCnt*self.hashSize+8)
var pos, i int64
binary.LittleEndian.PutUint64(chunk[0:8], uint64(size))
binary.LittleEndian.PutUint64(chunk[0:8], uint64(size))
childrenWg := &sync.WaitGroup{}
var secSize int64
for i < branchCnt {
// the last item can have shorter data
if size-pos < treeSize {
secSize = size - pos
} else {
secSize = treeSize
childrenWg := &sync.WaitGroup{}
var secSize int64
for i < branchCnt {
// the last item can have shorter data
if size-pos < treeSize {
secSize = size - pos
} else {
secSize = treeSize
}
// the hash of that data
subTreeKey := chunk[8+i*self.hashSize : 8+(i+1)*self.hashSize]
childrenWg.Add(1)
self.split(depth-1, treeSize/self.branches, subTreeKey, data, secSize, jobC, chunkC, errC, childrenWg, swg, wwg)
i++
pos += treeSize
}
// wait for all the children to complete calculating their hashes and copying them onto sections of the chunk
// parentWg.Add(1)
// go func() {
childrenWg.Wait()
if len(jobC) > self.workerCount && self.workerCount < processors {
if wwg != nil {
wwg.Add(1)
}
self.workerCount++
go self.hashWorker(jobC, chunkC, errC, swg, wwg)
}
select {
case jobC <- &hashJob{key, chunk, size, parentWg}:
case <-errC:
}
}
func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC chan error, swg, wwg *sync.WaitGroup) {
hasher := self.hashFunc()
if wwg != nil {
defer wwg.Done()
}
for {
select {
case job, ok := <-jobC:
if !ok {
return
}
// take the section of the data encoded in the subTree
subTreeData := NewChunkReader(data, pos, secSize)
// the hash of that data
subTreeKey := chunk[8+i*self.hashSize : 8+(i+1)*self.hashSize]
childrenWg.Add(1)
go self.split(depth-1, treeSize/self.branches, subTreeKey, subTreeData, chunkC, errc, childrenWg, swg)
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 chunks
hash = self.Hash(chunk)
newChunk = &Chunk{
Key: hash,
SData: chunk,
Size: size,
wg: swg,
// now we got the hashes in the chunk, then hash the chunks
hasher.Reset()
self.hashChunk(hasher, job, chunkC, swg)
// glog.V(logger.Detail).Infof("[BZZ] hash chunk (%v)", job.size)
case <-errC:
return
}
}
}
// 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) hashChunk(hasher hash.Hash, job *hashJob, chunkC chan *Chunk, swg *sync.WaitGroup) {
hasher.Write(job.chunk)
h := hasher.Sum(nil)
newChunk := &Chunk{
Key: h,
SData: job.chunk,
Size: job.size,
wg: swg,
}
// report hash of this chunk one level up (keys corresponds to the proper subslice of the parent chunk)
copy(job.key, h)
// send off new chunk to storage
if chunkC != nil {
if swg != nil {
swg.Add(1)
}
}
// send off new chunk to storage
job.parentWg.Done()
if chunkC != nil {
chunkC <- newChunk
}
// report hash of this chunk one level up (keys corresponds to the proper subslice of the parent chunk)x
copy(key, hash)
}
// implements the Joiner interface
func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) SectionReader {
return &LazyChunkReader{
@ -263,7 +278,7 @@ func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) SectionReader {
}
}
// LazyChunkReader implements LazySectionReader
// LazyChunkReader implements Lazy.SectionReader
type LazyChunkReader struct {
key Key // root key
chunkC chan *Chunk // chunk channel to send retrieve requests on
@ -282,7 +297,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
C: make(chan bool), // close channel to signal data delivery
}
self.chunkC <- chunk // submit retrieval request, someone should be listening on the other side (or we will time out globally)
glog.V(logger.Detail).Infof("[BZZ] readAt: reading %v into %d bytes at offset %d.", chunk.Key.Log(), len(b), off)
// glog.V(logger.Detail).Infof("[BZZ] readAt: reading %v into %d bytes at offset %d.", chunk.Key.Log(), len(b), off)
// waiting for the chunk retrieval
select {
@ -409,3 +424,35 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
} //for
wg.Wait()
}
func (self *LazyChunkReader) Size() (n int64) {
self.ReadAt(nil, 0)
return self.size
}
func (self *LazyChunkReader) Read(b []byte) (read int, err error) {
read, err = self.ReadAt(b, self.off)
self.off += int64(read)
return
}
var errWhence = errors.New("Seek: invalid whence")
var errOffset = errors.New("Seek: invalid offset")
func (s *LazyChunkReader) Seek(offset int64, whence int) (int64, error) {
switch whence {
default:
return 0, errWhence
case 0:
offset += 0
case 1:
offset += s.off
case 2:
offset += s.size
}
if offset < 0 {
return 0, errOffset
}
s.off = offset
return offset, nil
}

264
swarm/storage/chunker_test.go
View file

@ -2,20 +2,33 @@ package storage
import (
"bytes"
// "fmt"
"fmt"
"io"
"runtime"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
)
func init() {
glog.SetV(logger.Info)
glog.SetToStderr(true)
}
/*
Tests TreeChunker by splitting and joining a random byte slice
*/
type test interface {
Fatalf(string, ...interface{})
}
type chunkerTester struct {
errors []error
chunks []*Chunk
timeout bool
chunks []*Chunk
t test
}
func (self *chunkerTester) checkChunks(t *testing.T, want int) {
@ -25,77 +38,70 @@ func (self *chunkerTester) checkChunks(t *testing.T, want int) {
}
}
func (self *chunkerTester) Split(chunker *TreeChunker, l int) (key Key, input []byte) {
func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, chunkC chan *Chunk, swg *sync.WaitGroup) (key Key) {
// reset
self.errors = nil
self.chunks = nil
self.timeout = false
data, slice := testDataReader(l)
input = slice
key = make([]byte, 32)
chunkC := make(chan *Chunk, 1000)
errC := chunker.Split(key, data, chunkC, nil)
quitC := make(chan bool)
timeout := time.After(600 * time.Second)
if chunkC != nil {
go func() {
for {
select {
case <-timeout:
self.t.Fatalf("Join timeout error")
go func() {
LOOP:
for {
select {
case <-timeout:
self.timeout = true
break LOOP
case chunk := <-chunkC:
if chunk != nil {
case chunk, ok := <-chunkC:
if !ok {
// glog.V(logger.Info).Infof("chunkC closed quitting")
close(quitC)
return
}
// glog.V(logger.Info).Infof("chunk %v received", len(self.chunks))
self.chunks = append(self.chunks, chunk)
} else {
break LOOP
}
case err, ok := <-errC:
if err != nil {
self.errors = append(self.errors, err)
}
// fmt.Printf("err %v", err)
if !ok {
close(chunkC)
errC = nil
if chunk.wg != nil {
chunk.wg.Done()
}
}
}
}()
}
key, err := chunker.Split(data, size, chunkC, swg, nil)
if err != nil {
self.t.Fatalf("Split error: %v", err)
}
if chunkC != nil {
if swg != nil {
// glog.V(logger.Info).Infof("Waiting for storage to finish")
swg.Wait()
// glog.V(logger.Info).Infof("St orage finished")
}
close(quitC)
}()
<-quitC // waiting for it to finish
close(chunkC)
}
if chunkC != nil {
<-quitC
}
return
}
func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int) SectionReader {
func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int, chunkC chan *Chunk, quitC chan bool) SectionReader {
// reset but not the chunks
self.errors = nil
self.timeout = false
chunkC := make(chan *Chunk, 1000)
reader := chunker.Join(key, chunkC)
quitC := make(chan bool)
timeout := time.After(600 * time.Second)
i := 0
go func() {
LOOP:
for {
select {
case <-quitC:
break LOOP
case <-timeout:
self.timeout = true
break LOOP
self.t.Fatalf("Join timeout error")
case chunk := <-chunkC:
case chunk, ok := <-chunkC:
if !ok {
close(quitC)
return
}
i++
// dpaLogger.DebugDetailf("TESTER: chunk request %x", chunk.Key[:4])
// this just mocks the behaviour of a chunk store retrieval
var found bool
for _, ch := range self.chunks {
@ -106,53 +112,55 @@ func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int) SectionRea
}
}
if !found {
// fmt.Printf("TESTER: chunk unknown for %x", chunk.Key[:4])
self.t.Fatalf("not found ")
}
close(chunk.C)
// dpaLogger.DebugDetailf("TESTER: chunk request served %x", chunk.Key[:4])
}
}
}()
return reader
}
func testRandomData(chunker *TreeChunker, tester *chunkerTester, n int, chunks int, t *testing.T) {
key, input := tester.Split(chunker, n)
func testRandomData(n int, chunks int, t *testing.T) {
chunker := NewTreeChunker(&ChunkerParams{
Branches: 128,
Hash: "SHA3",
})
tester := &chunkerTester{t: t}
data, input := testDataReaderAndSlice(n)
t.Logf(" Key = %x\n", key)
chunkC := make(chan *Chunk, 1000)
swg := &sync.WaitGroup{}
tester.checkChunks(t, chunks)
time.Sleep(100 * time.Millisecond)
splitter := chunker
key := tester.Split(splitter, data, int64(n), chunkC, swg)
reader := tester.Join(chunker, key, 0)
// t.Logf(" Key = %v\n", key)
// tester.checkChunks(t, chunks)
chunkC = make(chan *Chunk, 1000)
quitC := make(chan bool)
reader := tester.Join(chunker, key, 0, chunkC, quitC)
output := make([]byte, n)
r, err := reader.Read(output)
if r != n || err != io.EOF {
t.Errorf("read error read: %v n = %v err = %v\n", r, n, err)
t.Fatalf("read error read: %v n = %v err = %v\n", r, n, err)
}
// t.Logf(" IN: %x\nOUT: %x\n", input, output)
if !bytes.Equal(output, input) {
t.Errorf("input and output mismatch\n IN: %x\nOUT: %x\n", input, output)
if input != nil {
if !bytes.Equal(output, input) {
t.Fatalf("input and output mismatch\n IN: %v\nOUT: %v\n", input, output)
}
}
close(chunkC)
<-quitC
}
func TestRandomData(t *testing.T) {
chunker, tester := chunkerAndTester()
testRandomData(chunker, tester, 60, 1, t)
testRandomData(chunker, tester, 179, 5, t)
testRandomData(chunker, tester, 253, 7, t)
// t.Logf("chunks %v", tester.chunks)
}
func chunkerAndTester() (chunker *TreeChunker, tester *chunkerTester) {
chunker = NewTreeChunker(&ChunkerParams{
Branches: 2,
Hash: "SHA256",
SplitTimeout: 10,
JoinTimeout: 10,
})
tester = &chunkerTester{}
return
testRandomData(60, 1, t)
testRandomData(83, 3, t)
testRandomData(179, 5, t)
testRandomData(253, 7, t)
}
func readAll(reader SectionReader, result []byte) {
@ -177,38 +185,90 @@ func benchReadAll(reader SectionReader) {
}
}
func benchmarkJoinRandomData(n int, chunks int, t *testing.B) {
t.StopTimer()
func benchmarkJoin(n int, t *testing.B) {
for i := 0; i < t.N; i++ {
// fmt.Printf("round %v\n", i)
chunker, tester := chunkerAndTester()
key, _ := tester.Split(chunker, n)
// fmt.Printf("split done %v, joining...\n", i)
chunker := NewTreeChunker(&ChunkerParams{
Branches: 128,
Hash: "SHA3",
})
tester := &chunkerTester{t: t}
data := testDataReader(n)
chunkC := make(chan *Chunk, 1000)
swg := &sync.WaitGroup{}
key := tester.Split(chunker, data, int64(n), chunkC, swg)
t.StartTimer()
reader := tester.Join(chunker, key, i)
// fmt.Printf("join done %v, reading...\n", i)
chunkC = make(chan *Chunk, 1000)
quitC := make(chan bool)
reader := tester.Join(chunker, key, i, chunkC, quitC)
t.StopTimer()
benchReadAll(reader)
close(chunkC)
<-quitC
}
}
func benchmarkSplitRandomData(n int, chunks int, t *testing.B) {
func benchmarkSplitTree(n int, t *testing.B) {
t.ReportAllocs()
for i := 0; i < t.N; i++ {
chunker, tester := chunkerAndTester()
tester.Split(chunker, n)
chunker := NewTreeChunker(&ChunkerParams{
Branches: 128,
Hash: "SHA3",
})
tester := &chunkerTester{t: t}
data := testDataReader(n)
// glog.V(logger.Info).Infof("splitting data of length %v", n)
tester.Split(chunker, data, int64(n), nil, nil)
}
stats := new(runtime.MemStats)
runtime.ReadMemStats(stats)
fmt.Println(stats.Sys)
}
func BenchmarkJoinRandomData_100_2(t *testing.B) { benchmarkJoinRandomData(100, 3, t) }
func BenchmarkJoinRandomData_1000_2(t *testing.B) { benchmarkJoinRandomData(1000, 3, t) }
func BenchmarkJoinRandomData_10000_2(t *testing.B) { benchmarkJoinRandomData(10000, 3, t) }
func BenchmarkJoinRandomData_100000_2(t *testing.B) { benchmarkJoinRandomData(100000, 3, t) }
func BenchmarkJoinRandomData_1000000_2(t *testing.B) { benchmarkJoinRandomData(1000000, 3, t) }
func benchmarkSplitPyramid(n int, t *testing.B) {
t.ReportAllocs()
for i := 0; i < t.N; i++ {
splitter := NewPyramidChunker(&ChunkerParams{
Branches: 128,
Hash: "SHA3",
})
tester := &chunkerTester{t: t}
data := testDataReader(n)
// glog.V(logger.Info).Infof("splitting data of length %v", n)
tester.Split(splitter, data, int64(n), nil, nil)
}
stats := new(runtime.MemStats)
runtime.ReadMemStats(stats)
fmt.Println(stats.Sys)
}
func BenchmarkSplitRandomData_100_2(t *testing.B) { benchmarkSplitRandomData(100, 3, t) }
func BenchmarkSplitRandomData_1000_2(t *testing.B) { benchmarkSplitRandomData(1000, 3, t) }
func BenchmarkSplitRandomData_10000_2(t *testing.B) { benchmarkSplitRandomData(10000, 3, t) }
func BenchmarkSplitRandomData_100000_2(t *testing.B) { benchmarkSplitRandomData(100000, 3, t) }
func BenchmarkSplitRandomData_1000000_2(t *testing.B) { benchmarkSplitRandomData(1000000, 3, t) }
func BenchmarkSplitRandomData_10000000_2(t *testing.B) { benchmarkSplitRandomData(10000000, 3, t) }
func BenchmarkJoin_100_2(t *testing.B) { benchmarkJoin(100, t) }
func BenchmarkJoin_1000_2(t *testing.B) { benchmarkJoin(1000, t) }
func BenchmarkJoin_10000_2(t *testing.B) { benchmarkJoin(10000, t) }
func BenchmarkJoin_100000_2(t *testing.B) { benchmarkJoin(100000, t) }
func BenchmarkJoin_1000000_2(t *testing.B) { benchmarkJoin(1000000, t) }
// go test -bench ./bzz -cpuprofile cpu.out -memprofile mem.out
func BenchmarkSplitTree_2(t *testing.B) { benchmarkSplitTree(100, t) }
func BenchmarkSplitTree_2h(t *testing.B) { benchmarkSplitTree(500, t) }
func BenchmarkSplitTree_3(t *testing.B) { benchmarkSplitTree(1000, t) }
func BenchmarkSplitTree_3h(t *testing.B) { benchmarkSplitTree(5000, t) }
func BenchmarkSplitTree_4(t *testing.B) { benchmarkSplitTree(10000, t) }
func BenchmarkSplitTree_4h(t *testing.B) { benchmarkSplitTree(50000, t) }
func BenchmarkSplitTree_5(t *testing.B) { benchmarkSplitTree(100000, t) }
func BenchmarkSplitTree_6(t *testing.B) { benchmarkSplitTree(1000000, t) }
func BenchmarkSplitTree_7(t *testing.B) { benchmarkSplitTree(10000000, t) }
func BenchmarkSplitTree_8(t *testing.B) { benchmarkSplitTree(100000000, t) }
func BenchmarkSplitPyramid_2(t *testing.B) { benchmarkSplitPyramid(100, t) }
func BenchmarkSplitPyramid_2h(t *testing.B) { benchmarkSplitPyramid(500, t) }
func BenchmarkSplitPyramid_3(t *testing.B) { benchmarkSplitPyramid(1000, t) }
func BenchmarkSplitPyramid_3h(t *testing.B) { benchmarkSplitPyramid(5000, t) }
func BenchmarkSplitPyramid_4(t *testing.B) { benchmarkSplitPyramid(10000, t) }
func BenchmarkSplitPyramid_4h(t *testing.B) { benchmarkSplitPyramid(50000, t) }
func BenchmarkSplitPyramid_5(t *testing.B) { benchmarkSplitPyramid(100000, t) }
func BenchmarkSplitPyramid_6(t *testing.B) { benchmarkSplitPyramid(1000000, t) }
func BenchmarkSplitPyramid_7(t *testing.B) { benchmarkSplitPyramid(10000000, t) }
func BenchmarkSplitPyramid_8(t *testing.B) { benchmarkSplitPyramid(100000000, t) }
// godep go test -bench ./swarm/storage -cpuprofile cpu.out -memprofile mem.out

194
swarm/storage/chunkreader.go
View file

@ -1,194 +0,0 @@
package storage
import (
"bytes"
"errors"
"io"
)
type Bounded interface {
Size() int64
}
type Sliced interface {
Slice(int64, int64) (b []byte, err error)
}
// Size, Seek, Read, ReadAt
type SectionReader interface {
Bounded
io.Seeker
io.Reader
io.ReaderAt
}
// ChunkReader implements SectionReader on a section
// of an underlying ReaderAt.
type ChunkReader struct {
r io.ReaderAt
base int64
off int64
limit int64
}
// 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}
}
// ByteSliceReader just extends byte.Reader to make base slice accessible
type ByteSliceReader struct {
*bytes.Reader
base []byte
}
func NewByteSliceReader(b []byte) *ByteSliceReader {
return &ByteSliceReader{
base: b,
Reader: bytes.NewReader(b),
}
}
// ByteSliceReader implements the Sliced interface
func (self *ByteSliceReader) Slice(from, to int64) (b []byte, err error) {
if from < 0 || to >= int64(self.Len()) {
err = io.EOF
} else {
b = self.base[from:to]
}
return
}
// NewChunkReaderFromBytes is a convenience shortcut to get a SectionReader over a byte slice
func NewChunkReaderFromBytes(b []byte) *ChunkReader {
return NewChunkReader(NewByteSliceReader(b), 0, int64(len(b)))
}
/*
The following is adapted from io.SectionReader
*/
func (s *ChunkReader) 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 *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 *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
}
n, err = s.r.ReadAt(p, off)
return
}
// added methods to that ChunkReader implements the Sliced interface
func (s *ChunkReader) Slice(from, to int64) (b []byte, err error) {
if from < 0 || to >= s.Size() {
err = io.EOF
} else {
if sl, ok := s.r.(Sliced); ok {
b, err = sl.Slice(s.base+from, s.base+to)
} else {
err = errors.New("not sliceable base")
}
}
return
}
// added method so that ChunkReader implements the io.WriterTo interface
// WriteTo method is used by io.Copy
// This is so that we avoid one extra step of allocation (if the underlying initial Reader implements Sliced
func (r *ChunkReader) WriteTo(w io.Writer) (n int64, err error) {
var b []byte
var m int
// if b, _ := r.Slice(r.off-r.base, r.limit-r.base); 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
}
// w
return
}
func (self *LazyChunkReader) Size() (n int64) {
self.ReadAt(nil, 0)
return self.size
}
func (self *LazyChunkReader) Read(b []byte) (read int, err error) {
read, err = self.ReadAt(b, self.off)
self.off += int64(read)
return
}
func (s *LazyChunkReader) Seek(offset int64, whence int) (int64, error) {
switch whence {
default:
return 0, errWhence
case 0:
offset += 0
case 1:
offset += s.off
case 2:
offset += s.size
}
if offset < 0 {
return 0, errOffset
}
s.off = offset
return offset, nil
}

64
swarm/storage/common_test.go
View file

@ -1,6 +1,7 @@
package storage
import (
"bytes"
"crypto/rand"
"io"
"sync"
@ -10,61 +11,40 @@ import (
"github.com/ethereum/go-ethereum/logger/glog"
)
func testDataReader(l int) (r *ChunkReader, slice []byte) {
func testDataReader(l int) (r io.Reader) {
return io.LimitReader(rand.Reader, int64(l))
}
func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) {
slice = make([]byte, l)
if _, err := rand.Read(slice); err != nil {
panic("rand error")
}
r = NewChunkReaderFromBytes(slice)
return
}
func randomChunks(l int64, branches int64, chunkC chan *Chunk) (key Key, errC chan error) {
chunker := NewTreeChunker(&ChunkerParams{
Branches: branches,
Hash: defaultHash,
SplitTimeout: splitTimeout,
})
key = make([]byte, 32)
b := make([]byte, l)
_, err := rand.Read(b)
if err != nil {
panic("no rand")
}
wg := &sync.WaitGroup{}
errC = chunker.Split(key, NewChunkReaderFromBytes(b), chunkC, wg)
wg.Wait()
r = bytes.NewReader(slice)
return
}
func testStore(m ChunkStore, l int64, branches int64, t *testing.T) {
chunkC := make(chan *Chunk)
key, errC := randomChunks(l, branches, chunkC)
SPLIT:
for {
select {
case chunk := <-chunkC:
go func() {
for chunk := range chunkC {
m.Put(chunk)
case err, ok := <-errC:
if err != nil {
t.Errorf("Chunker error: %v", err)
return
}
if !ok {
break SPLIT
if chunk.wg != nil {
chunk.wg.Done()
}
}
}
}()
chunker := NewTreeChunker(&ChunkerParams{
Branches: branches,
Hash: defaultHash,
SplitTimeout: splitTimeout,
Branches: branches,
Hash: defaultHash,
})
swg := &sync.WaitGroup{}
key, err := chunker.Split(rand.Reader, l, chunkC, swg, nil)
swg.Wait()
close(chunkC)
chunkC = make(chan *Chunk)
var r SectionReader
r = chunker.Join(key, chunkC)
r := chunker.Join(key, chunkC)
quit := make(chan bool)
@ -83,12 +63,14 @@ SPLIT:
close(chunk.C)
}(ch)
}
close(quit)
}()
b := make([]byte, l)
n, err := r.ReadAt(b, 0)
if err != io.EOF {
t.Errorf("read error (%v/%v) %v", n, l, err)
close(quit)
t.Fatalf("read error (%v/%v) %v", n, l, err)
}
close(chunkC)
<-quit
}

24
swarm/storage/dpa.go
View file

@ -2,6 +2,7 @@ package storage
import (
"errors"
"io"
"sync"
"time"
@ -78,27 +79,8 @@ func (self *DPA) Retrieve(key Key) SectionReader {
// Public API. Main entry point for document storage directly. Used by the
// FS-aware API and httpaccess
func (self *DPA) Store(data SectionReader, wg *sync.WaitGroup) (key Key, err error) {
key = make([]byte, self.Chunker.KeySize())
errC := self.Chunker.Split(key, data, self.storeC, wg)
SPLIT:
for {
select {
case err, ok := <-errC:
if err != nil {
glog.V(logger.Error).Infof("[BZZ] chunker split error: %v", err)
}
if !ok {
break SPLIT
}
case <-self.quitC:
break SPLIT
}
}
return
func (self *DPA) Store(data io.Reader, size int64, wg *sync.WaitGroup) (key Key, err error) {
return self.Chunker.Split(data, size, self.storeC, nil, wg)
}
func (self *DPA) Start() {

8
swarm/storage/dpa_test.go
View file

@ -29,9 +29,9 @@ func TestDPArandom(t *testing.T) {
ChunkStore: localStore,
}
dpa.Start()
reader, slice := testDataReader(testDataSize)
reader, slice := testDataReaderAndSlice(testDataSize)
wg := &sync.WaitGroup{}
key, err := dpa.Store(reader, wg)
key, err := dpa.Store(reader, testDataSize, wg)
if err != nil {
t.Errorf("Store error: %v", err)
}
@ -85,9 +85,9 @@ func TestDPA_capacity(t *testing.T) {
ChunkStore: localStore,
}
dpa.Start()
reader, slice := testDataReader(testDataSize)
reader, slice := testDataReaderAndSlice(testDataSize)
wg := &sync.WaitGroup{}
key, err := dpa.Store(reader, wg)
key, err := dpa.Store(reader, testDataSize, wg)
if err != nil {
t.Errorf("Store error: %v", err)
}

165
swarm/storage/pyramid.go Normal file
View file

@ -0,0 +1,165 @@
package storage
import (
"io"
"math"
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
)
const (
processors = 8
)
type Tree struct {
Chunks int64
Levels []map[int64]*Node
Lock sync.RWMutex
}
type Node struct {
Pending int64
Children []common.Hash
Last bool
}
type Task struct {
Index int64 // Index of the chunk being processed
Data []byte // Binary blob of the chunk
Last bool
}
type PyramidChunker struct {
hashFunc Hasher
chunkSize int64
hashSize int64
branches int64
workerCount int
}
func NewPyramidChunker(params *ChunkerParams) (self *PyramidChunker) {
self = &PyramidChunker{}
self.hashFunc = MakeHashFunc(params.Hash)
self.branches = params.Branches
self.hashSize = int64(self.hashFunc().Size())
self.chunkSize = self.hashSize * self.branches
self.workerCount = 1
return
}
func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, swg, wwg *sync.WaitGroup) (Key, error) {
chunks := (size + self.chunkSize - 1) / self.chunkSize
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{
Chunks: chunks,
Levels: make([]map[int64]*Node, depth),
}
for i := 0; i < depth; i++ {
results.Levels[i] = make(map[int64]*Node)
}
// Create a pool of workers to crunch through the file
tasks := make(chan *Task, 2*processors)
pend := new(sync.WaitGroup)
abortC := make(chan bool)
for i := 0; i < processors; i++ {
pend.Add(1)
go self.processor(pend, tasks, &results)
}
// Feed the chunks into the task pool
for index := 0; ; index++ {
buffer := make([]byte, self.chunkSize+8)
n, err := io.ReadFull(data, buffer)
last := err == io.ErrUnexpectedEOF
if err != nil && !last {
glog.V(logger.Info).Infof("error: %v", err)
close(abortC)
}
pend.Add(1)
// glog.V(logger.Info).Infof("-> task %v (%v)", index, n)
select {
case tasks <- &Task{Index: int64(index), Data: buffer[:n+8], Last: last}:
case <-abortC:
return nil, err
}
if last {
// glog.V(logger.Info).Infof("last task %v (%v)", index, n)
break
}
}
// Wait for the workers and return
close(tasks)
pend.Wait()
// glog.V(logger.Info).Infof("len: %v", results.Levels[0][0])
key := results.Levels[0][0].Children[0][:]
return key, nil
}
func (self *PyramidChunker) processor(pend *sync.WaitGroup, tasks chan *Task, results *Tree) {
defer pend.Done()
// glog.V(logger.Info).Infof("processor started")
// Start processing leaf chunks ad infinitum
hasher := self.hashFunc()
for task := range tasks {
depth, pow := len(results.Levels)-1, self.branches
// glog.V(logger.Info).Infof("task: %v, last: %v", task.Index, task.Last)
var node *Node
for depth >= 0 {
// New chunk received, reset the hasher and start processing
hasher.Reset()
if node == nil { // Leaf node, hash the data chunk
hasher.Write(task.Data)
} else { // Internal node, hash the children
for _, hash := range node.Children {
hasher.Write(hash[:])
}
}
hash := hasher.Sum(nil)
last := task.Last || (node != nil) && node.Last
// Insert the subresult into the memoization tree
results.Lock.Lock()
if node = results.Levels[depth][task.Index/pow]; node == nil {
// Figure out the pending tasks
pending := self.branches
if task.Index/pow == results.Chunks/pow {
pending = (results.Chunks + pow/self.branches - 1) / (pow / self.branches) % self.branches
}
node = &Node{pending, make([]common.Hash, pending), last}
results.Levels[depth][task.Index/pow] = node
}
node.Pending--
i := task.Index / (pow / self.branches) % self.branches
if last {
node.Pending -= self.branches - i
node.Children = node.Children[:i+1]
node.Last = true
}
copy(node.Children[i][:], hash)
left := node.Pending
if depth+1 < len(results.Levels) {
delete(results.Levels[depth+1], task.Index/(pow/self.branches))
}
results.Lock.Unlock()
// If there's more work to be done, leave for others
// glog.V(logger.Info).Infof("left %v", left)
if left > 0 {
break
}
// We're the last ones in this batch, merge the children together
depth--
pow *= self.branches
}
pend.Done()
}
}

61
swarm/storage/types.go
View file

@ -5,6 +5,7 @@ import (
"crypto"
"fmt"
"hash"
"io"
"sync"
"github.com/ethereum/go-ethereum/common"
@ -17,6 +18,41 @@ type Peer interface{}
type Key []byte
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
}
func IsZeroKey(key Key) bool {
return len(key) == 0 || bytes.Equal(key, ZeroKey)
}
@ -127,7 +163,7 @@ After getting notified that all the data has been split (the error channel is cl
When calling Join with a root key, the caller gets returned a seekable lazy reader. The caller again provides a channel 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). As the seekable reader is used, the chunker then puts these together the relevant parts on demand.
*/
type Chunker interface {
type Splitter 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 storage channel, which the caller provides.
@ -135,7 +171,10 @@ type Chunker interface {
The caller gets returned an error channel, if an error is encountered during splitting, it is fed to errC error channel.
A closed error signals process completion at which point the key can be considered final if there were no errors.
*/
Split(key Key, data SectionReader, chunkC chan *Chunk, wg *sync.WaitGroup) chan error
Split(io.Reader, int64, chan *Chunk, *sync.WaitGroup, *sync.WaitGroup) (Key, error)
}
type Joiner interface {
/*
Join reconstructs original content based on a root key.
When joining, the caller gets returned a Lazy SectionReader, which is
@ -149,7 +188,19 @@ type Chunker interface {
is because it is left to the DPA to decide which sources are trusted.
*/
Join(key Key, chunkC chan *Chunk) SectionReader
// returns the key length
KeySize() int64
}
type Chunker interface {
Joiner
Splitter
// returns the key length
// KeySize() int64
}
// Size, Seek, Read, ReadAt
type SectionReader interface {
Size() int64
io.Seeker
io.Reader
io.ReaderAt
}