diff --git a/swarm/api/api.go b/swarm/api/api.go index 883f226cf2..cc76b36ae5 100644 --- a/swarm/api/api.go +++ b/swarm/api/api.go @@ -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 } diff --git a/swarm/api/config_test.go b/swarm/api/config_test.go index c772d6aa88..48fbfacb20 100644 --- a/swarm/api/config_test.go +++ b/swarm/api/config_test.go @@ -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, diff --git a/swarm/api/filesystem.go b/swarm/api/filesystem.go index 002bb95273..34f2a8f0c8 100644 --- a/swarm/api/filesystem.go +++ b/swarm/api/filesystem.go @@ -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" { diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index 6ac2b113e6..945fc4f6e0 100644 --- a/swarm/api/http/server.go +++ b/swarm/api/http/server.go @@ -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 { diff --git a/swarm/api/manifest.go b/swarm/api/manifest.go index 0337382ff0..ab74a35bb0 100644 --- a/swarm/api/manifest.go +++ b/swarm/api/manifest.go @@ -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 diff --git a/swarm/cmd/bzzhash/bzzhash.go b/swarm/cmd/bzzhash/bzzhash.go index d4df5fb50b..37df224bfd 100644 --- a/swarm/cmd/bzzhash/bzzhash.go +++ b/swarm/cmd/bzzhash/bzzhash.go @@ -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) } } diff --git a/swarm/cmd/swarm/swarm.sh b/swarm/cmd/swarm/swarm.sh index 8546e1b8d8..8e3e2bb9c1 100644 --- a/swarm/cmd/swarm/swarm.sh +++ b/swarm/cmd/swarm/swarm.sh @@ -336,6 +336,8 @@ case $cmd in up $*;; "down" ) down $*;; + "download" ) + download $*;; "init" ) init $*;; "start" ) diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index c19bb622ec..156b64d7e0 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -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 +} diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go index 7dd301fff5..fbcf210588 100644 --- a/swarm/storage/chunker_test.go +++ b/swarm/storage/chunker_test.go @@ -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 diff --git a/swarm/storage/chunkreader.go b/swarm/storage/chunkreader.go deleted file mode 100644 index b147c85bf8..0000000000 --- a/swarm/storage/chunkreader.go +++ /dev/null @@ -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 -} diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index 40dc35fc69..b3fdc027b7 100644 --- a/swarm/storage/common_test.go +++ b/swarm/storage/common_test.go @@ -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 } diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index 922ff3bec2..ba8e0e2c56 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -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() { diff --git a/swarm/storage/dpa_test.go b/swarm/storage/dpa_test.go index a4400783fb..4c50a7214f 100644 --- a/swarm/storage/dpa_test.go +++ b/swarm/storage/dpa_test.go @@ -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) } diff --git a/swarm/storage/pyramid.go b/swarm/storage/pyramid.go new file mode 100644 index 0000000000..507dc7b768 --- /dev/null +++ b/swarm/storage/pyramid.go @@ -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() + } +} diff --git a/swarm/storage/types.go b/swarm/storage/types.go index 124a56a085..0a40fb4614 100644 --- a/swarm/storage/types.go +++ b/swarm/storage/types.go @@ -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<