forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-21 20:26:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

added size to chunk data

Browse source
This commit is contained in:
zsfelfoldi 2015-02-13 15:15:19 +01:00
parent 41c4e97182
commit 41cc4c239d
8 changed files with 116 additions and 105 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

76
bzz/chunker.go
View file

@ -33,7 +33,7 @@ import (
const (
hasherfunc crypto.Hash = crypto.SHA256 // http://golang.org/pkg/hash/#Hash
branches int64 = 128
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,
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,14 +254,14 @@ 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
/* chunkReader := NewChunkReaderFromBytes(chunk) // bytes.Reader almost implements SectionReader
chunkData := make([]byte, chunkReader.Size())
chunkReader.ReadAt(chunkData, 0)
chunkReader.ReadAt(chunkData, 0)*/
hash = self.Hash(size, chunkData)
hash = self.Hash(chunk)
newChunk = &Chunk{
Key: hash,
Data: chunkData,
SData: chunk,
Size: size,
wg: swg,
}
@ -268,6 +270,7 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR
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
}

13
bzz/chunker_test.go
View file

@ -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)

3
bzz/common_test.go
View file

@ -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)

16
bzz/dbstore.go
View file

@ -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,7 +108,7 @@ func encodeIndex(index *dpaDBIndex) []byte {
func encodeData(chunk *Chunk) []byte {
var rlpEntry struct {
/* var rlpEntry struct {
Data []byte
Size uint64
}
@ -118,7 +117,9 @@ func encodeData(chunk *Chunk) []byte {
rlpEntry.Size = uint64(chunk.Size)
data, _ := rlp.EncodeToBytes(rlpEntry)
return data
return data*/
return chunk.SData
}
@ -131,7 +132,7 @@ func decodeIndex(data []byte, index *dpaDBIndex) {
func decodeData(data []byte, chunk *Chunk) {
var rlpEntry struct {
/* var rlpEntry struct {
Data []byte
Size uint64
}
@ -142,7 +143,10 @@ func decodeData(data []byte, chunk *Chunk) {
panic(err.Error())
}
chunk.Data = rlpEntry.Data
chunk.Size = int64(rlpEntry.Size)
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 {

4
bzz/dpa.go
View file

@ -56,7 +56,7 @@ 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
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
@ -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)

6
bzz/memstore.go
View file

@ -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
@ -255,7 +255,7 @@ func (s *memStore) Get(hash Key) (chunk *Chunk, err error) {
node.updateAccess(s.accessCnt)
chunk = &Chunk{
Key: hash,
Data: node.entry.Data,
SData: node.entry.SData,
Size: node.entry.Size,
}
if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt {

28
bzz/netstore.go
View file

@ -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
@ -86,12 +87,12 @@ func (self *NetStore) addStoreRequest(req *storeRequestMsgData) {
if err != nil {
chunk = &Chunk{
Key: req.Key,
Data: req.Data,
Size: int64(req.Size),
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
@ -241,8 +242,7 @@ func (self *NetStore) propagateResponse(chunk *Chunk) {
dpaLogger.Debugf("NetStore.propagateResponse id %064x", id)
msg := &storeRequestMsgData{
Key: chunk.Key,
Data: chunk.Data,
Size: uint64(chunk.Size),
SData: chunk.SData,
Id: uint64(id),
}
for _, req := range requesters {
@ -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
@ -275,9 +274,8 @@ func (self *NetStore) store(chunk *Chunk) {
id := generateId()
req := &storeRequestMsgData{
Key: chunk.Key,
Data: chunk.Data,
SData: chunk.SData,
Id: uint64(id),
Size: uint64(chunk.Size),
}
for _, peer := range self.hive.getPeers(chunk.Key) {
go peer.store(req)

5
bzz/protocol.go
View file

@ -74,8 +74,7 @@ type statusMsgData struct {
*/
type storeRequestMsgData struct {
Key Key // hash of datasize | data
Size uint64 // size of data in bytes
Data []byte // is this needed?
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) {