From ce53889384231bca9c527f8a1cc0a0ec9bdc17f5 Mon Sep 17 00:00:00 2001 From: "Daniel A. Nagy" Date: Tue, 10 Feb 2015 20:08:56 +0100 Subject: [PATCH] Double-reading from Chunker eliminated. Join not fixed. --- bzz/chunker.go | 38 ++++++++++++++---------------- bzz/dpa.go | 13 ++++------ bzz/httpaccess.go | 60 +++++++++++++++++++++++++++++++++++++++++++++-- eth/backend.go | 10 ++++++-- 4 files changed, 88 insertions(+), 33 deletions(-) diff --git a/bzz/chunker.go b/bzz/chunker.go index 7081cb3d1e..4272c84b13 100644 --- a/bzz/chunker.go +++ b/bzz/chunker.go @@ -26,7 +26,6 @@ import ( "crypto" "encoding/binary" "fmt" - "io" "sync" "time" ) @@ -124,24 +123,16 @@ func (self *Chunk) String() string { var slice []byte var n int var err error - if self.Reader != nil { - size = 32 // we are printing 32 bytes of the data - slice = make([]byte, size) - n, err = self.Reader.ReadAt(slice, 0) - if err != nil && err != io.EOF { - slice = []byte(fmt.Sprintf("ERROR: %v", err)) - } - } - return fmt.Sprintf("Key: [%x..] TreeSize: %v Chunksize: %v Data: %x\n", self.Key[:4], self.Size, size, slice[:n]) + return fmt.Sprintf("Key: [%x..] TreeSize: %v Chunksize: %v Data: %x\n", self.Key[:4], self.Size, size, self.Data[: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 SectionReader) []byte { +func (self *TreeChunker) Hash(size int64, input []byte) []byte { hasher := self.HashFunc.New() binary.Write(hasher, binary.LittleEndian, size) - io.Copy(hasher, input) // it uses WriteTo if available, ChunkReader implements io.WriterTo + hasher.Write(input) return hasher.Sum(nil) } @@ -218,12 +209,14 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR if depth == 0 { // leaf nodes -> content chunks - hash = self.Hash(size, data) + chunkData := make([]byte, data.Size()) + data.ReadAt(chunkData, 0) + hash = self.Hash(size, chunkData) // dpaLogger.Debugf("content chunk: max subtree size: %v, data size: %v", treeSize, size) newChunk = &Chunk{ - Key: hash, - Reader: data, - Size: size, + Key: hash, + Data: chunkData, + Size: size, } } else { // intermediate chunk containing child nodes hashes @@ -257,11 +250,14 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data SectionR childrenWg.Wait() // now we got the hashes in the chunk, then hash the chunk chunkReader := NewChunkReaderFromBytes(chunk) // bytes.Reader almost implements SectionReader - hash = self.Hash(size, chunkReader) + chunkData := make([]byte, chunkReader.Size()) + chunkReader.ReadAt(chunkData, 0) + + hash = self.Hash(size, chunkData) newChunk = &Chunk{ - Key: hash, - Reader: chunkReader, - Size: size, + Key: hash, + Data: chunkData, + Size: size, } } // send off new chunk to storage @@ -357,7 +353,7 @@ func (self *TreeChunker) join(depth int, treeSize int64, key Key, chunkC chan *C } if depth == 0 { - r = LazyReader(chunk.Reader) + r = LazyReader(NewByteSliceReader(chunk.Data)) return // simply give back the chunks reader for content chunks } diff --git a/bzz/dpa.go b/bzz/dpa.go index 8a0a416740..39f82e675f 100644 --- a/bzz/dpa.go +++ b/bzz/dpa.go @@ -56,12 +56,11 @@ 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 { - Reader SectionReader // nil if request, to be supplied by dpa - 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 // + 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 // } type ChunkStore interface { @@ -171,8 +170,6 @@ func (self *DPA) storeLoop() { go func() { STORE: for chunk := range self.storeC { - chunk.Data = make([]byte, chunk.Reader.Size()) - chunk.Reader.ReadAt(chunk.Data, 0) self.ChunkStore.Put(chunk) select { case <-self.quitC: diff --git a/bzz/httpaccess.go b/bzz/httpaccess.go index 072fa9fc23..824667596f 100644 --- a/bzz/httpaccess.go +++ b/bzz/httpaccess.go @@ -4,7 +4,9 @@ A simple http server interface to Swarm package bzz import ( + "fmt" "github.com/ethereum/go-ethereum/ethutil" + "io" "net/http" "regexp" "time" @@ -18,10 +20,64 @@ var ( uriMatcher = regexp.MustCompile("^/raw/[0-9A-Fa-f]{64}$") ) +type sequentialReader struct { + reader io.Reader + pos int64 + ahead map[int64](chan bool) +} + +func (self *sequentialReader) ReadAt(target []byte, off int64) (n int, err error) { + if self.pos != off { + dpaLogger.Debugf("Swarm: deferred read in POST at position %d, offset %d.", + self.pos, off) + wait := make(chan bool) + self.ahead[off] = wait + if <-wait { + // failed read behind + n = 0 + err = io.ErrUnexpectedEOF + return + } + } + n, err = self.reader.Read(target) + dpaLogger.Debugf("Swarm: Read %d bytes into buffer size %d from POST, error %v.", + n, len(target), err) + if err != nil { + for i := range self.ahead { + self.ahead[i] <- true + self.ahead[i] = nil + } + } + self.pos += int64(n) + wait := self.ahead[self.pos] + if wait != nil { + dpaLogger.Debugf("Swarm: deferred read in POST at position %d triggered.", + self.pos) + self.ahead[self.pos] = nil + close(wait) + } + return +} + func handler(w http.ResponseWriter, r *http.Request, dpa *DPA) { uri := r.RequestURI switch { - case r.Method == "PUT": + case r.Method == "POST": + if uri == "/raw" { + dpaLogger.Debugf("Swarm: POST request received.") + key, err := dpa.Store(io.NewSectionReader(&sequentialReader{ + reader: r.Body, + ahead: make(map[int64]chan bool), + }, 0, r.ContentLength)) + if err == nil { + fmt.Fprintf(w, "%064x", key) + dpaLogger.Debugf("Swarm: Object %064x stored", key) + } else { + http.Error(w, err.Error(), http.StatusBadRequest) + } + } else { + http.Error(w, "No POST to "+uri+" allowed.", http.StatusBadRequest) + } case r.Method == "GET": if uriMatcher.MatchString(uri) { name := uri[5:] @@ -31,7 +87,7 @@ func handler(w http.ResponseWriter, r *http.Request, dpa *DPA) { http.Error(w, "Object "+uri+" not found.", http.StatusNotFound) } default: - http.Error(w, "Method "+r.Method+" is not supported.", http.StatusBadRequest) + http.Error(w, "Method "+r.Method+" is not supported.", http.StatusMethodNotAllowed) } } diff --git a/eth/backend.go b/eth/backend.go index a6d117eee4..367ca7ec68 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -141,8 +141,14 @@ func New(config *Config) (*Ethereum, error) { eth.whisper.Protocol(), bzz.BzzProtocol(netStore), } - // dpa := - bzz.StartHttpServer(nil) + chunker := &bzz.TreeChunker{} + chunker.Init() + dpa := &bzz.DPA{ + Chunker: chunker, + ChunkStore: netStore, + } + dpa.Start() + bzz.StartHttpServer(dpa) nat, err := p2p.ParseNAT(config.NATType, config.PMPGateway) if err != nil {