diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go index e699d1740e..5354ec9ffc 100644 --- a/swarm/storage/chunker_test.go +++ b/swarm/storage/chunker_test.go @@ -199,7 +199,7 @@ func testRandomData(splitter Splitter, n int, tester *chunkerTester) Key { input, found := tester.inputs[uint64(n)] var data io.Reader if !found { - data, input = testDataReaderAndSlice(n) + data, input = generateRandomData(n) tester.inputs[uint64(n)] = input } else { data = io.LimitReader(bytes.NewReader(input), int64(n)) @@ -234,66 +234,6 @@ func testRandomData(splitter Splitter, n int, tester *chunkerTester) Key { return key } -func testRandomDataAppend(splitter Splitter, n, m 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 = io.LimitReader(bytes.NewReader(input), int64(n)) - } - - chunkC := make(chan *Chunk, 1000) - - key, wait, err := tester.Split(splitter, data, int64(n), chunkC, nil) - if err != nil { - tester.t.Fatalf(err.Error()) - } - wait() - tester.t.Logf(" Key = %v\n", key) - - //create a append data stream - appendInput, found := tester.inputs[uint64(m)] - var appendData io.Reader - if !found { - appendData, appendInput = testDataReaderAndSlice(m) - tester.inputs[uint64(m)] = appendInput - } else { - appendData = io.LimitReader(bytes.NewReader(appendInput), int64(m)) - } - - chunkC = make(chan *Chunk, 1000) - - newKey, wait, err := tester.Append(splitter, key, appendData, chunkC, nil) - if err != nil { - tester.t.Fatalf(err.Error()) - } - wait() - tester.t.Logf(" NewKey = %v\n", newKey) - - chunkC = make(chan *Chunk, 1000) - quitC := make(chan bool) - - chunker := NewTreeChunker(NewChunkerParams()) - reader := tester.Join(chunker, newKey, 0, chunkC, quitC) - newOutput := make([]byte, n+m) - r, err := reader.Read(newOutput) - if r != (n + m) { - tester.t.Fatalf("read error read: %v n = %v err = %v\n", r, n, err) - } - - newInput := append(input, appendInput...) - if !bytes.Equal(newOutput, newInput) { - tester.t.Fatalf("input and output mismatch\n IN: %v\nOUT: %v\n", newInput, newOutput) - } - - close(chunkC) -} - func TestSha3ForCorrectness(t *testing.T) { tester := &chunkerTester{t: t} @@ -323,19 +263,73 @@ func TestSha3ForCorrectness(t *testing.T) { } -// func TestDataAppend(t *testing.T) { -// // sizes := []int{1, 1, 1, 4095, 4096, 4097, 1, 1, 1, 123456, 2345678, 2345678} -// sizes := []int{1} -// // appendSizes := []int{4095, 4096, 4097, 1, 1, 1, 8191, 8192, 8193, 9000, 3000, 5000} -// appendSizes := []int{4095} -// -// tester := &chunkerTester{t: t} -// chunker := NewPyramidChunker(NewChunkerParams()) -// for i, s := range sizes { -// testRandomDataAppend(chunker, s, appendSizes[i], tester) -// -// } -// } +func TestDataAppend(t *testing.T) { + sizes := []int{1, 1, 1, 4095, 4096, 4097, 1, 1, 1, 123456, 2345678, 2345678} + appendSizes := []int{4095, 4096, 4097, 1, 1, 1, 8191, 8192, 8193, 9000, 3000, 5000} + + tester := &chunkerTester{t: t} + for i := range sizes { + n := sizes[i] + m := appendSizes[i] + + if tester.inputs == nil { + tester.inputs = make(map[uint64][]byte) + } + input, found := tester.inputs[uint64(n)] + var data io.Reader + if !found { + data, input = generateRandomData(n) + tester.inputs[uint64(n)] = input + } else { + data = io.LimitReader(bytes.NewReader(input), int64(n)) + } + + chunkC := make(chan *Chunk, 1000) + + chunker := NewPyramidChunker(NewChunkerParams()) + key, wait, err := tester.Split(chunker, data, int64(n), chunkC, nil) + if err != nil { + tester.t.Fatalf(err.Error()) + } + wait() + + //create a append data stream + appendInput, found := tester.inputs[uint64(m)] + var appendData io.Reader + if !found { + appendData, appendInput = generateRandomData(m) + tester.inputs[uint64(m)] = appendInput + } else { + appendData = io.LimitReader(bytes.NewReader(appendInput), int64(m)) + } + + chunkC = make(chan *Chunk, 1000) + + newKey, wait, err := tester.Append(chunker, key, appendData, chunkC, nil) + if err != nil { + tester.t.Fatalf(err.Error()) + } + wait() + + chunkC = make(chan *Chunk, 1000) + quitC := make(chan bool) + + treeChunker := NewTreeChunker(NewChunkerParams()) + reader := tester.Join(treeChunker, newKey, 0, chunkC, quitC) + newOutput := make([]byte, n+m) + r, err := reader.Read(newOutput) + if r != (n + m) { + tester.t.Fatalf("read error read: %v n = %v m = %v err = %v\n", r, n, m, err) + } + + newInput := append(input, appendInput...) + if !bytes.Equal(newOutput, newInput) { + tester.t.Fatalf("input and output mismatch\n IN: %v\nOUT: %v\n", newInput, newOutput) + } + + close(chunkC) + } +} func TestRandomData(t *testing.T) { sizes := []int{1, 60, 83, 179, 253, 1024, 4095, 4096, 4097, 8191, 8192, 8193, 12287, 12288, 12289, 123456, 2345678} diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index 21f8e5bbc1..52a964114e 100644 --- a/swarm/storage/common_test.go +++ b/swarm/storage/common_test.go @@ -24,13 +24,13 @@ import ( "fmt" "hash" "io" - "os" "sync" "testing" "time" "github.com/ethereum/go-ethereum/crypto/sha3" "github.com/ethereum/go-ethereum/log" + colorable "github.com/mattn/go-colorable" ) var ( @@ -39,7 +39,8 @@ var ( func init() { flag.Parse() - log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) + log.PrintOrigins(true) + log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true)))) } type brokenLimitedReader struct { @@ -170,7 +171,7 @@ func (r *brokenLimitedReader) Read(buf []byte) (int, error) { return r.lr.Read(buf) } -func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) { +func generateRandomData(l int) (r io.Reader, slice []byte) { slice = make([]byte, l) if _, err := rand.Read(slice); err != nil { panic("rand error") diff --git a/swarm/storage/dpa_test.go b/swarm/storage/dpa_test.go index 20b07216a3..51b276d2e8 100644 --- a/swarm/storage/dpa_test.go +++ b/swarm/storage/dpa_test.go @@ -48,7 +48,7 @@ func TestDPArandom(t *testing.T) { defer dpa.Stop() defer os.RemoveAll("/tmp/bzz") - reader, slice := testDataReaderAndSlice(testDataSize) + reader, slice := generateRandomData(testDataSize) key, wait, err := dpa.Store(reader, testDataSize) if err != nil { t.Errorf("Store error: %v", err) @@ -103,7 +103,7 @@ func TestDPA_capacity(t *testing.T) { ChunkStore: localStore, } dpa.Start() - reader, slice := testDataReaderAndSlice(testDataSize) + reader, slice := generateRandomData(testDataSize) key, wait, err := dpa.Store(reader, testDataSize) if err != nil { t.Errorf("Store error: %v", err) diff --git a/swarm/storage/pyramid.go b/swarm/storage/pyramid.go index a34b73993f..d3f50f293e 100644 --- a/swarm/storage/pyramid.go +++ b/swarm/storage/pyramid.go @@ -20,8 +20,11 @@ import ( "encoding/binary" "errors" "io" + "io/ioutil" "sync" "time" + + "github.com/ethereum/go-ethereum/log" ) /* @@ -166,6 +169,7 @@ func (self *PyramidChunker) decrementWorkerCount() { } func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk) (k Key, wait func(), err error) { + log.Trace("pyramid.chunker: Split()") jobC := make(chan *chunkJob, 2*ChunkProcessors) wg := &sync.WaitGroup{} storageWG := &sync.WaitGroup{} @@ -204,6 +208,7 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk } func (self *PyramidChunker) Append(key Key, data io.Reader, chunkC chan *Chunk) (k Key, wait func(), err error) { + log.Trace("pyramid.chunker: Append()") quitC := make(chan bool) rootKey := make([]byte, self.hashSize) chunkLevel := make([][]*TreeEntry, self.branches) @@ -262,6 +267,8 @@ func (self *PyramidChunker) processor(id int64, jobC chan *chunkJob, chunkC chan } func (self *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJob, chunkC chan *Chunk, storageWG *sync.WaitGroup) { + log.Trace("pyramid.chunker: processChunk()", "id", id) + hasher.ResetWithLength(job.chunk[:8]) // 8 bytes of length hasher.Write(job.chunk[8:]) // minus 8 []byte length h := hasher.Sum(nil) @@ -287,11 +294,13 @@ func (self *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJ } func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC chan *Chunk, quitC chan bool) error { + log.Trace("pyramid.chunker: loadTree()") // Get the root chunk to get the total size chunk := retrieve(key, chunkC, quitC) if chunk == nil { return errLoadingTreeRootChunk } + log.Trace("pyramid.chunker: root chunk", "chunk.Size", chunk.Size, "self.chunkSize", self.chunkSize) //if data size is less than a chunk... add a parent with update as pending if chunk.Size <= self.chunkSize { @@ -315,6 +324,7 @@ func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC for ; treeSize < chunk.Size; treeSize *= self.branches { depth++ } + log.Trace("pyramid.chunker", "depth", depth) // Add the root chunk entry branchCount := int64(len(chunk.SData)-8) / self.hashSize @@ -367,20 +377,19 @@ func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC } func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEntry, data io.Reader, rootKey []byte, quitC chan bool, wg *sync.WaitGroup, jobC chan *chunkJob, chunkC chan *Chunk, errC chan error, storageWG *sync.WaitGroup) { + log.Trace("pyramid.chunker: prepareChunks", "isAppend", isAppend) defer wg.Done() chunkWG := &sync.WaitGroup{} - totalDataSize := 0 self.incrementWorkerCount() go self.processor(self.workerCount, jobC, chunkC, errC, quitC, storageWG) parent := NewTreeEntry(self) - var unFinishedChunk *Chunk + var unfinishedChunk *Chunk if isAppend && len(chunkLevel[0]) != 0 { - lastIndex := len(chunkLevel[0]) - 1 ent := chunkLevel[0][lastIndex] @@ -398,42 +407,43 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt lastBranch := parent.branchCount - 1 lastKey := parent.chunk[8+lastBranch*self.hashSize : 8+(lastBranch+1)*self.hashSize] - unFinishedChunk = retrieve(lastKey, chunkC, quitC) - if unFinishedChunk.Size < self.chunkSize { - - parent.subtreeSize = parent.subtreeSize - uint64(unFinishedChunk.Size) + unfinishedChunk = retrieve(lastKey, chunkC, quitC) + if unfinishedChunk.Size < self.chunkSize { + parent.subtreeSize = parent.subtreeSize - uint64(unfinishedChunk.Size) parent.branchCount = parent.branchCount - 1 } else { - unFinishedChunk = nil + unfinishedChunk = nil } } } for index := 0; ; index++ { - - var n int var err error chunkData := make([]byte, self.chunkSize+8) - maxBuf := len(chunkData) - readBytes := 8 - if unFinishedChunk != nil { - copy(chunkData, unFinishedChunk.SData) - readBytes += int(unFinishedChunk.Size) - } - for readBytes < maxBuf { - n0, err0 := data.Read(chunkData[readBytes:]) - readBytes += n0 - n += n0 - if err0 != nil { - if err0 != io.EOF || (n0 == 0 && maxBuf == readBytes) || n == 0 || n0 != 0 { - err = err0 - } - break - } - } - unFinishedChunk = nil - totalDataSize += n + var readBytes int + + if unfinishedChunk != nil { + copy(chunkData, unfinishedChunk.SData) + readBytes += int(unfinishedChunk.Size) + unfinishedChunk = nil + log.Trace("pyramid.chunker: found unfinished chunk", "readBytes", readBytes) + } + + var res []byte + res, err = ioutil.ReadAll(io.LimitReader(data, int64(len(chunkData)-(8+readBytes)))) + + // hack for ioutil.ReadAll: + // a successful call to ioutil.ReadAll returns err == nil, not err == EOF, whereas we + // want to propagate the io.EOF error + if len(res) == 0 && err == nil { + err = io.EOF + } + copy(chunkData[8+readBytes:], res) + + readBytes += len(res) + log.Trace("pyramid.chunker: copied all data", "readBytes", readBytes) + if err != nil { if err == io.EOF || err == io.ErrUnexpectedEOF { if parent.branchCount == 1 { @@ -450,19 +460,18 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt } // Data ended in chunk boundary.. just signal to start bulding tree - if n == 0 { + if readBytes == 0 { self.buildTree(isAppend, chunkLevel, parent, chunkWG, jobC, quitC, true, rootKey) break } else { - - pkey := self.enqueueDataChunk(chunkData, uint64(n), parent, chunkWG, jobC, quitC) + pkey := self.enqueueDataChunk(chunkData, uint64(readBytes), parent, chunkWG, jobC, quitC) // update tree related parent data structures - parent.subtreeSize += uint64(n) + parent.subtreeSize += uint64(readBytes) parent.branchCount++ // Data got exhausted... signal to send any parent tree related chunks - if int64(n) < self.chunkSize { + if int64(readBytes) < self.chunkSize { // only one data chunk .. so dont add any parent chunk if parent.branchCount <= 1 {