From ce2a7e23e61b05837de8613ba98bc7175e703d6c Mon Sep 17 00:00:00 2001 From: zelig Date: Fri, 15 Jul 2016 14:46:45 +0200 Subject: [PATCH] swarm/storage: complete pyramid splitter * add size calculation, chunk passing and waitgroup control * benchmark against tree chunker - conclusions remain * remove logging from chunker * fix joiner and extend tests and benchmarks --- swarm/storage/chunker.go | 23 +++--- swarm/storage/chunker_test.go | 133 +++++++++++++++++++--------------- swarm/storage/common_test.go | 28 ++++++- swarm/storage/dpa.go | 76 +++++++++---------- swarm/storage/pyramid.go | 66 ++++++++++++----- 5 files changed, 198 insertions(+), 128 deletions(-) diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index 490f5b432e..abe576f56f 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -7,9 +7,8 @@ import ( "hash" "io" "sync" - - "github.com/ethereum/go-ethereum/logger" - "github.com/ethereum/go-ethereum/logger/glog" + // "github.com/ethereum/go-ethereum/logger" + // "github.com/ethereum/go-ethereum/logger/glog" ) /* @@ -321,7 +320,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { if err != nil { return 0, err } - glog.V(logger.Detail).Infof("readAt: len(b): %v, off: %v, size: %v ", len(b), off, size) + // glog.V(logger.Detail).Infof("readAt: len(b): %v, off: %v, size: %v ", len(b), off, size) 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) @@ -350,9 +349,9 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { 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) + // glog.V(logger.Detail).Infof("end: len(b): %v, off: %v, size: %v ", len(b), off, size) if off+int64(len(b)) >= size { - glog.V(logger.Detail).Infof(" len(b): %v EOF", len(b)) + // glog.V(logger.Detail).Infof(" len(b): %v EOF", len(b)) return len(b), io.EOF } // glog.V(logger.Detail).Infof("[BZZ] ReadAt returning at %d: %v", read, err) @@ -362,7 +361,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { 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() // 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("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) @@ -376,7 +375,11 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr // leaf chunk found if depth == 0 { - glog.V(logger.Detail).Infof("[BZZ] depth: %v, len(b): %v, off: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, len(b), off, eoff, chunk.Size, treeSize) + // 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)) + if extra > 0 { + eoff -= extra + } copy(b, chunk.SData[8+off:8+eoff]) return // simply give back the chunks reader for content chunks } @@ -387,7 +390,7 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr wg := &sync.WaitGroup{} defer wg.Wait() - glog.V(logger.Detail).Infof("[BZZ] start %v,end %v", start, end) + // glog.V(logger.Detail).Infof("[BZZ] start %v,end %v", start, end) for i := start; i < end; i++ { soff := i * treeSize @@ -457,7 +460,7 @@ func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk { // Read keeps a cursor so cannot be called simulateously, see ReadAt func (self *LazyChunkReader) Read(b []byte) (read int, err error) { read, err = self.ReadAt(b, self.off) - glog.V(logger.Detail).Infof("[BZZ] read: %v, off: %v, error: %v", read, self.off, err) + // glog.V(logger.Detail).Infof("[BZZ] read: %v, off: %v, error: %v", read, self.off, err) self.off += int64(read) return diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go index cdf549ff45..33c3209926 100644 --- a/swarm/storage/chunker_test.go +++ b/swarm/storage/chunker_test.go @@ -2,32 +2,27 @@ package storage import ( "bytes" + "encoding/binary" "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{}) + Logf(string, ...interface{}) } type chunkerTester struct { - chunks []*Chunk + inputs map[uint64][]byte + chunks map[string]*Chunk t test } @@ -40,7 +35,12 @@ 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) { // reset - self.chunks = nil + self.chunks = make(map[string]*Chunk) + + if self.inputs == nil { + self.inputs = make(map[uint64][]byte) + } + quitC := make(chan bool) timeout := time.After(600 * time.Second) if chunkC != nil { @@ -57,7 +57,8 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c return } // 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 if chunk.wg != nil { chunk.wg.Done() } @@ -73,22 +74,26 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c if swg != nil { // glog.V(logger.Info).Infof("Waiting for storage to finish") swg.Wait() - // glog.V(logger.Info).Infof("St orage finished") + // glog.V(logger.Info).Infof("Storage finished") } close(chunkC) } if chunkC != nil { + // glog.V(logger.Info).Infof("waiting for splitter finished") <-quitC + // glog.V(logger.Info).Infof("Splitter finished") } return } -func (self *chunkerTester) Join(chunker *TreeChunker, 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 + // glog.V(logger.Info).Infof("Splitter finished") reader := chunker.Join(key, chunkC) timeout := time.After(600 * time.Second) + // glog.V(logger.Info).Infof("Splitter finished") i := 0 go func() { for { @@ -101,55 +106,59 @@ func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int, chunkC cha close(quitC) return } - i++ + // glog.V(logger.Info).Infof("chunk %v: %v", i, chunk.Key.String()) // this just mocks the behaviour of a chunk store retrieval - var found bool - for _, ch := range self.chunks { - if bytes.Equal(chunk.Key, ch.Key) { - found = true - chunk.SData = ch.SData - break - } - } - if !found { - self.t.Fatalf("not found ") + stored, success := self.chunks[chunk.Key.String()] + // glog.V(logger.Info).Infof("chunk %v, success: %v", chunk.Key.String(), success) + if !success { + self.t.Fatalf("not found") + return } + // glog.V(logger.Info).Infof("chunk %v: %v", i, chunk.Key.String()) + chunk.SData = stored.SData + chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) close(chunk.C) + i++ } } }() return reader } -func testRandomData(n int, chunks int, t *testing.T) { - chunker := NewTreeChunker(&ChunkerParams{ - Branches: 128, - Hash: "SHA3", - }) - tester := &chunkerTester{t: t} - data, input := testDataReaderAndSlice(n) +func testRandomData(splitter Splitter, n int, tester *chunkerTester) { + if tester.inputs == nil { + tester.inputs = make(map[uint64][]byte) + } + input, found := tester.inputs[uint64(n)] + var data io.Reader + if !found { + data, input = testDataReaderAndSlice(n) + tester.inputs[uint64(n)] = input + } else { + data = limitReader(bytes.NewReader(input), n) + } chunkC := make(chan *Chunk, 1000) swg := &sync.WaitGroup{} - splitter := chunker key := tester.Split(splitter, data, int64(n), chunkC, swg) + tester.t.Logf(" Key = %v\n", key) - // t.Logf(" Key = %v\n", key) - - // tester.checkChunks(t, chunks) chunkC = make(chan *Chunk, 1000) quitC := make(chan bool) + chunker := NewTreeChunker(NewChunkerParams()) reader := tester.Join(chunker, key, 0, chunkC, quitC) output := make([]byte, n) + // glog.V(logger.Info).Infof(" Key = %v\n", key) r, err := reader.Read(output) + // glog.V(logger.Info).Infof(" read = %v %v\n", r, err) if r != n || err != io.EOF { - 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) } if input != nil { if !bytes.Equal(output, input) { - t.Fatalf("input and output mismatch\n IN: %v\nOUT: %v\n", input, output) + tester.t.Fatalf("input and output mismatch\n IN: %v\nOUT: %v\n", input, output) } } close(chunkC) @@ -157,10 +166,17 @@ func testRandomData(n int, chunks int, t *testing.T) { } func TestRandomData(t *testing.T) { - testRandomData(60, 1, t) - testRandomData(83, 3, t) - testRandomData(179, 5, t) - testRandomData(253, 7, t) + // sizes := []int{123456} + sizes := []int{1, 60, 83, 179, 253, 1024, 4095, 4096, 4097, 123456} + tester := &chunkerTester{t: t} + chunker := NewTreeChunker(NewChunkerParams()) + for _, s := range sizes { + testRandomData(chunker, s, tester) + } + pyramid := NewPyramidChunker(NewChunkerParams()) + for _, s := range sizes { + testRandomData(pyramid, s, tester) + } } func readAll(reader LazySectionReader, result []byte) { @@ -186,11 +202,9 @@ func benchReadAll(reader LazySectionReader) { } func benchmarkJoin(n int, t *testing.B) { + t.ReportAllocs() for i := 0; i < t.N; i++ { - chunker := NewTreeChunker(&ChunkerParams{ - Branches: 128, - Hash: "SHA3", - }) + chunker := NewTreeChunker(NewChunkerParams()) tester := &chunkerTester{t: t} data := testDataReader(n) @@ -198,24 +212,24 @@ func benchmarkJoin(n int, t *testing.B) { swg := &sync.WaitGroup{} key := tester.Split(chunker, data, int64(n), chunkC, swg) - t.StartTimer() + // t.StartTimer() chunkC = make(chan *Chunk, 1000) quitC := make(chan bool) reader := tester.Join(chunker, key, i, chunkC, quitC) - t.StopTimer() benchReadAll(reader) close(chunkC) <-quitC + // t.StopTimer() } + stats := new(runtime.MemStats) + runtime.ReadMemStats(stats) + fmt.Println(stats.Sys) } func benchmarkSplitTree(n int, t *testing.B) { t.ReportAllocs() for i := 0; i < t.N; i++ { - chunker := NewTreeChunker(&ChunkerParams{ - Branches: 128, - Hash: "SHA3", - }) + chunker := NewTreeChunker(NewChunkerParams()) tester := &chunkerTester{t: t} data := testDataReader(n) // glog.V(logger.Info).Infof("splitting data of length %v", n) @@ -229,10 +243,7 @@ func benchmarkSplitTree(n int, t *testing.B) { func benchmarkSplitPyramid(n int, t *testing.B) { t.ReportAllocs() for i := 0; i < t.N; i++ { - splitter := NewPyramidChunker(&ChunkerParams{ - Branches: 128, - Hash: "SHA3", - }) + splitter := NewPyramidChunker(NewChunkerParams()) tester := &chunkerTester{t: t} data := testDataReader(n) // glog.V(logger.Info).Infof("splitting data of length %v", n) @@ -243,11 +254,13 @@ func benchmarkSplitPyramid(n int, t *testing.B) { fmt.Println(stats.Sys) } -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) } +func BenchmarkJoin_2(t *testing.B) { benchmarkJoin(100, t) } +func BenchmarkJoin_3(t *testing.B) { benchmarkJoin(1000, t) } +func BenchmarkJoin_4(t *testing.B) { benchmarkJoin(10000, t) } +func BenchmarkJoin_5(t *testing.B) { benchmarkJoin(100000, t) } +func BenchmarkJoin_6(t *testing.B) { benchmarkJoin(1000000, t) } +func BenchmarkJoin_7(t *testing.B) { benchmarkJoin(10000000, t) } +func BenchmarkJoin_8(t *testing.B) { benchmarkJoin(100000000, t) } func BenchmarkSplitTree_2(t *testing.B) { benchmarkSplitTree(100, t) } func BenchmarkSplitTree_2h(t *testing.B) { benchmarkSplitTree(500, t) } diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index 55fcbfd409..3a9ddb216a 100644 --- a/swarm/storage/common_test.go +++ b/swarm/storage/common_test.go @@ -11,8 +11,32 @@ import ( "github.com/ethereum/go-ethereum/logger/glog" ) +type limitedReader struct { + r io.Reader + off int64 + size int64 +} + +func limitReader(r io.Reader, size int) *limitedReader { + return &limitedReader{r, 0, int64(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) { - return io.LimitReader(rand.Reader, int64(l)) + return limitReader(rand.Reader, l) } func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) { @@ -20,7 +44,7 @@ func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) { if _, err := rand.Read(slice); err != nil { panic("rand error") } - r = bytes.NewReader(slice) + r = limitReader(bytes.NewReader(slice), l) return } diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index 2a23a06ae0..4d16943c77 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -23,12 +23,12 @@ implementation for storage or retrieval. */ const ( - storeChanCapacity = 100 - retrieveChanCapacity = 100 + storeChanCapacity = 1000 + retrieveChanCapacity = 1000 singletonSwarmDbCapacity = 50000 singletonSwarmCacheCapacity = 500 - maxStoreProcesses = 100 - maxRetrieveProcesses = 100 + maxStoreProcesses = 100 + maxRetrieveProcesses = 100 ) var ( @@ -44,7 +44,7 @@ type DPA struct { lock sync.Mutex running bool wg *sync.WaitGroup - quitC chan bool + quitC chan bool } // for testing locally @@ -112,58 +112,58 @@ func (self *DPA) Stop() { // retrieveLoop dispatches the parallel chunk retrieval requests received on the // retrieve channel to its ChunkStore (NetStore or LocalStore) func (self *DPA) retrieveLoop() { - for i:=0; i< maxRetrieveProcesses; i++ { + for i := 0; i < maxRetrieveProcesses; i++ { go self.retrieveWorker() } - glog.V(logger.Detail).Infof("[BZZ] dpa: retrieve loop spawning %v workers", maxRetrieveProcesses) + glog.V(logger.Detail).Infof("[BZZ] dpa: retrieve loop spawning %v workers", maxRetrieveProcesses) } func (self *DPA) retrieveWorker() { - for chunk := range self.retrieveC { - glog.V(logger.Detail).Infof("[BZZ] dpa: retrieve loop : chunk %v", chunk.Key.Log()) - storedChunk, err := self.Get(chunk.Key) - if err == notFound { - glog.V(logger.Detail).Infof("[BZZ] chunk %v not found", chunk.Key.Log()) - } else if err != nil { - glog.V(logger.Detail).Infof("[BZZ] error retrieving chunk %v: %v", chunk.Key.Log(), err) - } else { - chunk.SData = storedChunk.SData - chunk.Size = storedChunk.Size - } - close(chunk.C) + for chunk := range self.retrieveC { + glog.V(logger.Detail).Infof("[BZZ] dpa: retrieve loop : chunk %v", chunk.Key.Log()) + storedChunk, err := self.Get(chunk.Key) + if err == notFound { + glog.V(logger.Detail).Infof("[BZZ] chunk %v not found", chunk.Key.Log()) + } else if err != nil { + glog.V(logger.Detail).Infof("[BZZ] error retrieving chunk %v: %v", chunk.Key.Log(), err) + } else { + chunk.SData = storedChunk.SData + chunk.Size = storedChunk.Size + } + close(chunk.C) - select { - case <-self.quitC: - return - default: -} + select { + case <-self.quitC: + return + default: } } +} // storeLoop dispatches the parallel chunk store request processors // received on the store channel to its ChunkStore (NetStore or LocalStore) func (self *DPA) storeLoop() { - for i:=0; i< maxStoreProcesses; i++ { + for i := 0; i < maxStoreProcesses; i++ { go self.storeWorker() } - glog.V(logger.Detail).Infof("[BZZ] dpa: store spawning %v workers", maxStoreProcesses) + glog.V(logger.Detail).Infof("[BZZ] dpa: store spawning %v workers", maxStoreProcesses) } func (self *DPA) storeWorker() { - for chunk := range self.storeC { - self.Put(chunk) - if chunk.wg != nil { - glog.V(logger.Detail).Infof("[BZZ] dpa: store processor %v", chunk.Key.Log()) - chunk.wg.Done() + for chunk := range self.storeC { + self.Put(chunk) + if chunk.wg != nil { + glog.V(logger.Detail).Infof("[BZZ] dpa: store processor %v", chunk.Key.Log()) + chunk.wg.Done() - } - select { - case <-self.quitC: - return - default: -} -} + } + select { + case <-self.quitC: + return + default: + } + } } // DpaChunkStore implements the ChunkStore interface, diff --git a/swarm/storage/pyramid.go b/swarm/storage/pyramid.go index 507dc7b768..b93d932969 100644 --- a/swarm/storage/pyramid.go +++ b/swarm/storage/pyramid.go @@ -1,13 +1,14 @@ package storage import ( + "encoding/binary" + "fmt" "io" "math" + "strings" "sync" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/logger" - "github.com/ethereum/go-ethereum/logger/glog" ) const ( @@ -22,12 +23,22 @@ type Tree struct { type Node struct { Pending int64 + Size uint64 Children []common.Hash Last bool } +func (self *Node) String() string { + var children []string + for _, node := range self.Children { + children = append(children, node.Hex()) + } + return fmt.Sprintf("pending: %v, size: %v, last :%v, children: %v", self.Pending, self.Size, self.Last, strings.Join(children, ", ")) +} + type Task struct { - Index int64 // Index of the chunk being processed + Index int64 // Index of the chunk being processed + Size uint64 Data []byte // Binary blob of the chunk Last bool } @@ -54,7 +65,7 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk 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) + // glog.V(logger.Detail).Infof("chunks: %v, depth: %v", chunks, depth) results := Tree{ Chunks: chunks, @@ -69,22 +80,24 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk abortC := make(chan bool) for i := 0; i < processors; i++ { pend.Add(1) - go self.processor(pend, tasks, &results) + go self.processor(pend, swg, tasks, chunkC, &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 + n, err := data.Read(buffer[8:]) + last := err == io.ErrUnexpectedEOF || err == io.EOF + // glog.V(logger.Detail).Infof("n: %v, index: %v, depth: %v", n, index, depth) if err != nil && !last { - glog.V(logger.Info).Infof("error: %v", err) - + // glog.V(logger.Info).Infof("error: %v", err) close(abortC) + break } + binary.LittleEndian.PutUint64(buffer[:8], uint64(n)) 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 tasks <- &Task{Index: int64(index), Size: uint64(n), Data: buffer[:n+8], Last: last}: case <-abortC: return nil, err } @@ -102,7 +115,7 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk return key, nil } -func (self *PyramidChunker) processor(pend *sync.WaitGroup, tasks chan *Task, results *Tree) { +func (self *PyramidChunker) processor(pend, swg *sync.WaitGroup, tasks chan *Task, chunkC chan *Chunk, results *Tree) { defer pend.Done() // glog.V(logger.Info).Infof("processor started") @@ -111,16 +124,22 @@ func (self *PyramidChunker) processor(pend *sync.WaitGroup, tasks chan *Task, re 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) - + size := task.Size + data := task.Data 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 { + size = node.Size + data = make([]byte, hasher.Size()*len(node.Children)+8) + binary.LittleEndian.PutUint64(data[:8], size) + + hasher.Write(data[:8]) + for i, hash := range node.Children { + copy(data[i*hasher.Size()+8:], hash[:]) hasher.Write(hash[:]) } } @@ -134,22 +153,33 @@ func (self *PyramidChunker) processor(pend *sync.WaitGroup, tasks chan *Task, re 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} + node = &Node{pending, 0, make([]common.Hash, pending), last} 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-- + // glog.V(logger.Info).Infof("pending now: %v", 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) + node.Size += size left := node.Pending - + // glog.V(logger.Info).Infof("left pending now: %v, node size: %v", left, node.Size) + if chunkC != nil { + if swg != nil { + swg.Add(1) + } + select { + case chunkC <- &Chunk{Key: hash, SData: data, wg: swg}: + // case <- self.quitC + } + } 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)