Double-reading from Chunker eliminated. Join not fixed.

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
 		}
 

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:

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

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 {