From 41cc4c239d32c61b9396251d629c740b18482fac Mon Sep 17 00:00:00 2001 From: zsfelfoldi Date: Fri, 13 Feb 2015 15:15:19 +0100 Subject: [PATCH] added size to chunk data --- bzz/chunker.go | 90 +++++++++++++++++++++++++-------------------- bzz/chunker_test.go | 13 ++++--- bzz/common_test.go | 3 +- bzz/dbstore.go | 46 ++++++++++++----------- bzz/dpa.go | 14 +++---- bzz/memstore.go | 10 ++--- bzz/netstore.go | 38 +++++++++---------- bzz/protocol.go | 7 ++-- 8 files changed, 116 insertions(+), 105 deletions(-) diff --git a/bzz/chunker.go b/bzz/chunker.go index 1838242d94..28b3ba663c 100644 --- a/bzz/chunker.go +++ b/bzz/chunker.go @@ -32,8 +32,8 @@ import ( ) const ( - hasherfunc crypto.Hash = crypto.SHA256 // http://golang.org/pkg/hash/#Hash - branches int64 = 128 + hasherfunc crypto.Hash = crypto.SHA256 // http://golang.org/pkg/hash/#Hash + defaultBranches int64 = 128 ) var ( @@ -99,7 +99,7 @@ func (self *TreeChunker) Init() { self.HashFunc = hasherfunc } if self.Branches == 0 { - self.Branches = branches + self.Branches = defaultBranches } if self.JoinTimeout == 0 { self.JoinTimeout = joinTimeout @@ -121,15 +121,14 @@ func (self *TreeChunker) KeySize() int64 { func (self *Chunk) String() string { var size int64 var n int - return fmt.Sprintf("Key: [%x..] TreeSize: %v Chunksize: %v Data: %x\n", self.Key[:4], self.Size, size, self.Data[:n]) + return fmt.Sprintf("Key: [%x..] TreeSize: %v Chunksize: %v Data: %x\n", self.Key[:4], self.Size, size, self.SData[:n]) } // 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(size int64, input []byte) []byte { +func (self *TreeChunker) Hash(input []byte) []byte { hasher := self.HashFunc.New() - binary.Write(hasher, binary.LittleEndian, size) hasher.Write(input) return hasher.Sum(nil) } @@ -212,23 +211,26 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR if depth == 0 { // leaf nodes -> content chunks - chunkData := make([]byte, data.Size()) - data.ReadAt(chunkData, 0) - hash = self.Hash(size, chunkData) + chunkData := make([]byte, data.Size()+8) + binary.LittleEndian.PutUint64(chunkData[0:8], uint64(size)) + data.ReadAt(chunkData[8:], 0) + hash = self.Hash(chunkData) // dpaLogger.Debugf("content chunk: max subtree size: %v, data size: %v", treeSize, size) newChunk = &Chunk{ - Key: hash, - Data: chunkData, - Size: size, + Key: hash, + SData: chunkData, + Size: size, } } else { // intermediate chunk containing child nodes hashes - branchCnt := int64((size-1)/treeSize) + 1 + branchCnt := int64((size + treeSize - 1) / treeSize) // dpaLogger.Debugf("intermediate node: setting branches: %v, depth: %v, max subtree size: %v, data size: %v", branches, depth, treeSize, size) - var chunk []byte = make([]byte, branches*self.hashSize) + var chunk []byte = make([]byte, branchCnt*self.hashSize+8) var pos, i int64 + binary.LittleEndian.PutUint64(chunk[0:8], uint64(size)) + childrenWg := &sync.WaitGroup{} var secSize int64 for i < branchCnt { @@ -241,7 +243,7 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR // take the section of the data corresponding encoded in the subTree subTreeData := NewChunkReader(data, pos, secSize) // the hash of that data - subTreeKey := chunk[i*self.hashSize : (i+1)*self.hashSize] + 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) @@ -252,22 +254,23 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR // 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 chunk - chunkReader := NewChunkReaderFromBytes(chunk) // bytes.Reader almost implements SectionReader - chunkData := make([]byte, chunkReader.Size()) - chunkReader.ReadAt(chunkData, 0) + /* chunkReader := NewChunkReaderFromBytes(chunk) // bytes.Reader almost implements SectionReader + chunkData := make([]byte, chunkReader.Size()) + chunkReader.ReadAt(chunkData, 0)*/ - hash = self.Hash(size, chunkData) + hash = self.Hash(chunk) newChunk = &Chunk{ - Key: hash, - Data: chunkData, - Size: size, - wg: swg, + Key: hash, + SData: chunk, + Size: size, + wg: swg, } if swg != nil { swg.Add(1) } } + // send off new chunk to storage if chunkC != nil { chunkC <- newChunk @@ -306,24 +309,25 @@ 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) - // dpaLogger.Debugf("readAt %x into %d bytes at %d.", chunk.Key[:4], len(b), off) + dpaLogger.Debugf("readAt %x into %d bytes at %d.", chunk.Key[:4], len(b), off) // waiting for the chunk retrieval select { case <-self.quitC: // this is how we control process leakage (quitC is closed once join is finished (after timeout)) - // dpaLogger.Debugf("quit") + dpaLogger.Debugf("quit") return case <-chunk.C: // bells are ringing, data have been delivered - // dpaLogger.Debugf("chunk data received for %x", chunk.Key[:4]) + dpaLogger.Debugf("chunk data received for %x", chunk.Key[:4]) } - if len(chunk.Data) == 0 { - // dpaLogger.Debugf("No payload.") + if len(chunk.SData) == 0 { + dpaLogger.Debugf("No payload.") return 0, notFound } + chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) self.size = chunk.Size if b == nil { - // dpaLogger.Debugf("Size query for %x.", chunk.Key[:4]) + dpaLogger.Debugf("Size query for %x.", chunk.Key[:4]) return } want := int64(len(b)) @@ -346,14 +350,14 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { }() select { case err = <-self.errC: - // dpaLogger.Debugf("ReadAt received %v.", err) + dpaLogger.Debugf("ReadAt received %v.", err) read = len(b) if off+int64(read) == self.size { err = io.EOF } - // dpaLogger.Debugf("ReadAt returning with %d, %v.", read, err) + dpaLogger.Debugf("ReadAt returning with %d, %v.", read, err) case <-self.quitC: - // dpaLogger.Debugf("ReadAt aborted with %d, %v.", read, err) + dpaLogger.Debugf("ReadAt aborted with %d, %v.", read, err) } return } @@ -361,7 +365,9 @@ 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) { defer parentWg.Done() - // dpaLogger.Debugf("depth: %v, loff: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, off, eoff, chunk.Size, treeSize) + dpaLogger.Debugf("depth: %v, loff: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, off, eoff, chunk.Size, treeSize) + + chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) // find appropriate block level for chunk.Size < treeSize && depth > 0 { @@ -370,9 +376,13 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr } if depth == 0 { - // dpaLogger.Debugf("len(b): %v, off: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, len(b), off, eoff, chunk.Size, treeSize) + dpaLogger.Debugf("len(b): %v, off: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, len(b), off, eoff, chunk.Size, treeSize) + if int64(len(b)) != eoff-off { + //fmt.Printf("len(b) = %v off = %v eoff = %v", len(b), off, eoff) + panic("len(b) does not match") + } - copy(b, chunk.Data[off:eoff]) + copy(b, chunk.SData[8+off:8+eoff]) return // simply give back the chunks reader for content chunks } @@ -396,14 +406,14 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr wg.Add(1) go func(j int64) { - childKey := chunk.Data[j*self.chunker.hashSize : (j+1)*self.chunker.hashSize] - // dpaLogger.Debugf("subtree index: %v -> %x", j, childKey[:4]) + childKey := chunk.SData[8+j*self.chunker.hashSize : 8+(j+1)*self.chunker.hashSize] + dpaLogger.Debugf("subtree index: %v -> %x", j, childKey[:4]) ch := &Chunk{ Key: childKey, C: make(chan bool), // close channel to signal data delivery } - // dpaLogger.Debugf("chunk data sent for %x (key interval in chunk %v-%v)", ch.Key[:4], j*self.chunker.hashSize, (j+1)*self.chunker.hashSize) + dpaLogger.Debugf("chunk data sent for %x (key interval in chunk %v-%v)", ch.Key[:4], j*self.chunker.hashSize, (j+1)*self.chunker.hashSize) self.chunkC <- ch // submit retrieval request, someone should be listening on the other side (or we will time out globally) // waiting for the chunk retrieval @@ -412,12 +422,12 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr // this is how we control process leakage (quitC is closed once join is finished (after timeout)) return case <-ch.C: // bells are ringing, data have been delivered - // dpaLogger.Debugf("chunk data received") + dpaLogger.Debugf("chunk data received") } if soff < off { soff = off } - if len(ch.Data) == 0 { + if len(ch.SData) == 0 { self.errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize) return } diff --git a/bzz/chunker_test.go b/bzz/chunker_test.go index d533317f0e..5238f11915 100644 --- a/bzz/chunker_test.go +++ b/bzz/chunker_test.go @@ -103,8 +103,7 @@ func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int) SectionRea for _, ch := range self.chunks { if bytes.Equal(chunk.Key, ch.Key) { found = true - chunk.Data = ch.Data - chunk.Size = ch.Size + chunk.SData = ch.SData break } } @@ -122,14 +121,16 @@ func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int) SectionRea func testRandomData(chunker *TreeChunker, tester *chunkerTester, n int, chunks int, t *testing.T) { key, input := tester.Split(chunker, n) + t.Logf(" Key = %x\n", key) + tester.checkChunks(t, chunks) time.Sleep(100 * time.Millisecond) reader := tester.Join(chunker, key, 0) output := make([]byte, n) - _, err := reader.Read(output) - if err != io.EOF { - t.Errorf("read error %v\n", err) + 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.Logf(" IN: %x\nOUT: %x\n", input, output) if !bytes.Equal(output, input) { @@ -146,7 +147,7 @@ func TestRandomData(t *testing.T) { } chunker.Init() tester := &chunkerTester{} - testRandomData(chunker, tester, 70, 3, t) + testRandomData(chunker, tester, 60, 1, t) testRandomData(chunker, tester, 179, 5, t) testRandomData(chunker, tester, 253, 7, t) // t.Logf("chunks %v", tester.chunks) diff --git a/bzz/common_test.go b/bzz/common_test.go index a378d8fb2f..edc7b3e3e7 100644 --- a/bzz/common_test.go +++ b/bzz/common_test.go @@ -72,8 +72,7 @@ SPLIT: } else if err != nil { dpaLogger.DebugDetailf("error retrieving chunk %x: %v", chunk.Key, err) } else { - chunk.Data = storedChunk.Data - chunk.Size = storedChunk.Size + chunk.SData = storedChunk.SData } dpaLogger.DebugDetailf("chunk '%x' not found", chunk.Key[:4]) close(chunk.C) diff --git a/bzz/dbstore.go b/bzz/dbstore.go index f94ed80dae..b235a0e5a1 100644 --- a/bzz/dbstore.go +++ b/bzz/dbstore.go @@ -1,5 +1,4 @@ -// disk storage layer for the package blockhash -// inefficient work-in-progress version +// disk storage layer for the package bzz package bzz @@ -109,16 +108,18 @@ func encodeIndex(index *dpaDBIndex) []byte { func encodeData(chunk *Chunk) []byte { - var rlpEntry struct { - Data []byte - Size uint64 - } + /* var rlpEntry struct { + Data []byte + Size uint64 + } - rlpEntry.Data = chunk.Data - rlpEntry.Size = uint64(chunk.Size) + rlpEntry.Data = chunk.Data + rlpEntry.Size = uint64(chunk.Size) - data, _ := rlp.EncodeToBytes(rlpEntry) - return data + data, _ := rlp.EncodeToBytes(rlpEntry) + return data*/ + + return chunk.SData } @@ -131,18 +132,21 @@ func decodeIndex(data []byte, index *dpaDBIndex) { func decodeData(data []byte, chunk *Chunk) { - var rlpEntry struct { - Data []byte - Size uint64 - } + /* var rlpEntry struct { + Data []byte + Size uint64 + } - dec := rlp.NewStream(bytes.NewReader(data)) - err := dec.Decode(&rlpEntry) - if err != nil { - panic(err.Error()) - } - chunk.Data = rlpEntry.Data - chunk.Size = int64(rlpEntry.Size) + dec := rlp.NewStream(bytes.NewReader(data)) + err := dec.Decode(&rlpEntry) + if err != nil { + panic(err.Error()) + } + chunk.Data = rlpEntry.Data + chunk.Size = int64(rlpEntry.Size)*/ + + chunk.SData = data + chunk.Size = int64(binary.LittleEndian.Uint64(data[0:8])) } func gcListPartition(list []*gcItem, left int, right int, pivotIndex int) int { diff --git a/bzz/dpa.go b/bzz/dpa.go index fa29f14cd4..8f3ba50e7c 100644 --- a/bzz/dpa.go +++ b/bzz/dpa.go @@ -56,12 +56,12 @@ type DPA struct { // but the size of the subtree encoded in the chunk // 0 if request, to be supplied by the dpa type Chunk struct { - Data []byte // nil if request, to be supplied by dpa - Size int64 // size of the data covered by the subtree encoded in this chunk - Key Key // always - C chan bool // to signal data delivery by the dpa - req *requestStatus // - wg *sync.WaitGroup + SData []byte // nil if request, to be supplied by dpa + Size int64 // size of the data covered by the subtree encoded in this chunk + Key Key // always + C chan bool // to signal data delivery by the dpa + req *requestStatus // + wg *sync.WaitGroup } type ChunkStore interface { @@ -134,7 +134,7 @@ func (self *DPA) retrieveLoop() { } else if err != nil { dpaLogger.DebugDetailf("error retrieving chunk %x: %v", chunk.Key, err) } else { - chunk.Data = storedChunk.Data + chunk.SData = storedChunk.SData chunk.Size = storedChunk.Size } close(chunk.C) diff --git a/bzz/memstore.go b/bzz/memstore.go index ce012222d4..8975e80d49 100644 --- a/bzz/memstore.go +++ b/bzz/memstore.go @@ -193,9 +193,9 @@ func (s *memStore) Put(entry *Chunk) { if node.entry.Key.isEqual(entry.Key) { node.updateAccess(s.accessCnt) - if node.entry.Data == nil { + if node.entry.SData == nil { node.entry.Size = entry.Size - node.entry.Data = entry.Data + node.entry.SData = entry.SData } if node.entry.req == nil { node.entry.req = entry.req @@ -254,9 +254,9 @@ func (s *memStore) Get(hash Key) (chunk *Chunk, err error) { s.accessCnt++ node.updateAccess(s.accessCnt) chunk = &Chunk{ - Key: hash, - Data: node.entry.Data, - Size: node.entry.Size, + Key: hash, + SData: node.entry.SData, + Size: node.entry.Size, } if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt { s.dbAccessCnt++ diff --git a/bzz/netstore.go b/bzz/netstore.go index edb37bce96..aa3e867659 100644 --- a/bzz/netstore.go +++ b/bzz/netstore.go @@ -1,6 +1,7 @@ package bzz import ( + "encoding/binary" "math/rand" "sync" "time" @@ -57,8 +58,8 @@ func (self *NetStore) Put(entry *Chunk) { dpaLogger.Debugf("NetStore.Put: localStore.Get returned with %v.", err) if err != nil { chunk = entry - } else if chunk.Data == nil { - chunk.Data = entry.Data + } else if chunk.SData == nil { + chunk.SData = entry.SData chunk.Size = entry.Size } else { return @@ -85,13 +86,13 @@ func (self *NetStore) addStoreRequest(req *storeRequestMsgData) { // we assume that a returned chunk is the one stored in the memory cache if err != nil { chunk = &Chunk{ - Key: req.Key, - Data: req.Data, - Size: int64(req.Size), + Key: req.Key, + SData: req.SData, + Size: int64(binary.LittleEndian.Uint64(req.SData[0:8])), } - } else if chunk.Data == nil { - chunk.Data = req.Data - chunk.Size = int64(req.Size) + } else if chunk.SData == nil { + chunk.SData = req.SData + chunk.Size = int64(binary.LittleEndian.Uint64(req.SData[0:8])) } else { return } @@ -103,7 +104,7 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) { chunk = self.get(key) id := generateId() timeout := time.Now().Add(searchTimeout) - if chunk.Data == nil { + if chunk.SData == nil { self.startSearch(chunk, id, &timeout) } else { return @@ -146,7 +147,7 @@ func (self *NetStore) addRetrieveRequest(req *retrieveRequestMsgData) { defer self.lock.Unlock() chunk := self.get(req.Key) - if chunk.Data == nil { + if chunk.SData == nil { chunk.req.status = reqSearching } else { chunk.req.status = reqFound @@ -240,10 +241,9 @@ func (self *NetStore) propagateResponse(chunk *Chunk) { counter := requesterCount dpaLogger.Debugf("NetStore.propagateResponse id %064x", id) msg := &storeRequestMsgData{ - Key: chunk.Key, - Data: chunk.Data, - Size: uint64(chunk.Size), - Id: uint64(id), + Key: chunk.Key, + SData: chunk.SData, + Id: uint64(id), } for _, req := range requesters { if req.timeout.After(time.Now()) { @@ -262,8 +262,7 @@ func (self *NetStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) { storeReq := &storeRequestMsgData{ Key: req.Key, Id: req.Id, - Data: chunk.Data, - Size: uint64(chunk.Size), + SData: chunk.SData, requestTimeout: req.timeout, // // StorageTimeout *time.Time // expiry of content // Metadata metaData @@ -274,10 +273,9 @@ func (self *NetStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) { func (self *NetStore) store(chunk *Chunk) { id := generateId() req := &storeRequestMsgData{ - Key: chunk.Key, - Data: chunk.Data, - Id: uint64(id), - Size: uint64(chunk.Size), + Key: chunk.Key, + SData: chunk.SData, + Id: uint64(id), } for _, peer := range self.hive.getPeers(chunk.Key) { go peer.store(req) diff --git a/bzz/protocol.go b/bzz/protocol.go index c0f024071c..d8dbdfe016 100644 --- a/bzz/protocol.go +++ b/bzz/protocol.go @@ -73,9 +73,8 @@ type statusMsgData struct { if the storage request is sufficiently close (within our proximity range (the last row of the routing table), then sending it to all peers will not guarantee convergence, so there needs to be an absolute expiry of the request too. Maybe the protocol should specify a forward probability exponentially declining with age. */ type storeRequestMsgData struct { - Key Key // hash of datasize | data - Size uint64 // size of data in bytes - Data []byte // is this needed? + Key Key // hash of datasize | data + SData []byte // is this needed? // optional Id uint64 // requestTimeout *time.Time // expiry for forwarding @@ -289,7 +288,7 @@ func (self *bzzProtocol) retrieve(req *retrieveRequestMsgData) { } func (self *bzzProtocol) store(req *storeRequestMsgData) { - p2p.EncodeMsg(self.rw, storeRequestMsg, req.Key, req.Size, req.Data, req.Id) + p2p.EncodeMsg(self.rw, storeRequestMsg, req.Key, req.SData, req.Id) } func (self *bzzProtocol) peers(req *peersMsgData) {