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

@@ -7,9 +7,8 @@ import (
 	"hash"
 	"io"
 	"sync"
-
+	// "github.com/ethereum/go-ethereum/logger"
-	"github.com/ethereum/go-ethereum/logger"
+	// "github.com/ethereum/go-ethereum/logger/glog"
-	"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

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
+				stored, success := self.chunks[chunk.Key.String()]
-				for _, ch := range self.chunks {
+				// glog.V(logger.Info).Infof("chunk %v, success: %v", chunk.Key.String(), success)
-					if bytes.Equal(chunk.Key, ch.Key) {
+				if !success {
-						found = true
+					self.t.Fatalf("not found")
-						chunk.SData = ch.SData
+					return
-						break
-					}
-				}
-				if !found {
-					self.t.Fatalf("not found	")
 				}
+				// 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) {
+func testRandomData(splitter Splitter, n int, tester *chunkerTester) {
-	chunker := NewTreeChunker(&ChunkerParams{
+	if tester.inputs == nil {
-		Branches: 128,
+		tester.inputs = make(map[uint64][]byte)
-		Hash:     "SHA3",
+	}
-	})
+	input, found := tester.inputs[uint64(n)]
-	tester := &chunkerTester{t: t}
+	var data io.Reader
-	data, input := testDataReaderAndSlice(n)
+	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)
+	// sizes := []int{123456}
-	testRandomData(83, 3, t)
+	sizes := []int{1, 60, 83, 179, 253, 1024, 4095, 4096, 4097, 123456}
-	testRandomData(179, 5, t)
+	tester := &chunkerTester{t: t}
-	testRandomData(253, 7, 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{
+		chunker := NewTreeChunker(NewChunkerParams())
-			Branches: 128,
-			Hash:     "SHA3",
-		})
 		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{
+		chunker := NewTreeChunker(NewChunkerParams())
-			Branches: 128,
-			Hash:     "SHA3",
-		})
 		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{
+		splitter := NewPyramidChunker(NewChunkerParams())
-			Branches: 128,
-			Hash:     "SHA3",
-		})
 		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_2(t *testing.B) { benchmarkJoin(100, t) }
-func BenchmarkJoin_1000_2(t *testing.B)    { benchmarkJoin(1000, t) }
+func BenchmarkJoin_3(t *testing.B) { benchmarkJoin(1000, t) }
-func BenchmarkJoin_10000_2(t *testing.B)   { benchmarkJoin(10000, t) }
+func BenchmarkJoin_4(t *testing.B) { benchmarkJoin(10000, t) }
-func BenchmarkJoin_100000_2(t *testing.B)  { benchmarkJoin(100000, t) }
+func BenchmarkJoin_5(t *testing.B) { benchmarkJoin(100000, t) }
-func BenchmarkJoin_1000000_2(t *testing.B) { benchmarkJoin(1000000, 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) }

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
 }
 

swarm/storage/dpa.go

@@ -23,12 +23,12 @@ implementation for storage or retrieval.
 */
 
 const (
-	storeChanCapacity           = 100
+	storeChanCapacity           = 1000
-	retrieveChanCapacity        = 100
+	retrieveChanCapacity        = 1000
 	singletonSwarmDbCapacity    = 50000
 	singletonSwarmCacheCapacity = 500
-	maxStoreProcesses = 100
+	maxStoreProcesses           = 100
-	maxRetrieveProcesses = 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 {
+	for chunk := range self.retrieveC {
-				glog.V(logger.Detail).Infof("[BZZ] dpa: retrieve loop : chunk %v", chunk.Key.Log())
+		glog.V(logger.Detail).Infof("[BZZ] dpa: retrieve loop : chunk %v", chunk.Key.Log())
-				storedChunk, err := self.Get(chunk.Key)
+		storedChunk, err := self.Get(chunk.Key)
-				if err == notFound {
+		if err == notFound {
-					glog.V(logger.Detail).Infof("[BZZ] chunk %v not found", chunk.Key.Log())
+			glog.V(logger.Detail).Infof("[BZZ] chunk %v not found", chunk.Key.Log())
-				} else if err != nil {
+		} else if err != nil {
-					glog.V(logger.Detail).Infof("[BZZ] error retrieving chunk %v: %v", chunk.Key.Log(), err)
+			glog.V(logger.Detail).Infof("[BZZ] error retrieving chunk %v: %v", chunk.Key.Log(), err)
-				} else {
+		} else {
-					chunk.SData = storedChunk.SData
+			chunk.SData = storedChunk.SData
-					chunk.Size = storedChunk.Size
+			chunk.Size = storedChunk.Size
-				}
+		}
-				close(chunk.C)
+		close(chunk.C)
 
-				select {
+		select {
-			case <-self.quitC:
+		case <-self.quitC:
-				return
+			return
-			default:
+		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 {
+	for chunk := range self.storeC {
-				self.Put(chunk)
+		self.Put(chunk)
-				if chunk.wg != nil {
+		if chunk.wg != nil {
-					glog.V(logger.Detail).Infof("[BZZ] dpa: store processor %v", chunk.Key.Log())
+			glog.V(logger.Detail).Infof("[BZZ] dpa: store processor %v", chunk.Key.Log())
-					chunk.wg.Done()
+			chunk.wg.Done()
 
-				}
+		}
-				select {
+		select {
-			case <-self.quitC:
+		case <-self.quitC:
-				return
+			return
-			default:
+		default:
-}
+		}
-}
+	}
 }
 
 // DpaChunkStore implements the ChunkStore interface,

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)
+		n, err := data.Read(buffer[8:])
-		last := err == io.ErrUnexpectedEOF
+		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)