merge

Merge branch 'bzz' of https://github.com/ethersphere/go-ethereum into bzz

Conflicts:
	bzz/chunker.go
	bzz/memstore.go
	bzz/netstore.go

bzz/chunker.go

@@ -321,7 +321,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 		dpaLogger.Debugf("chunk data received for %x", chunk.Key[:4])
 	}
 	if len(chunk.SData) == 0 {
-		dpaLogger.Debugf("No payload.")
+		dpaLogger.Debugf("No payload in %x.", chunk.Key)
 		return 0, notFound
 	}
 	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))

bzz/chunker_test.go

@@ -178,15 +178,25 @@ func readAll(reader SectionReader, result []byte) {
 	}
 }
 
+func benchReadAll(reader SectionReader) {
+	size := reader.Size()
+	output := make([]byte, 1000)
+	for pos := int64(0); pos < size; pos += 1000 {
+		reader.ReadAt(output, pos)
+	}
+}
+
 func benchmarkJoinRandomData(n int, chunks int, t *testing.B) {
+	t.StopTimer()
 	for i := 0; i < t.N; i++ {
-		t.StopTimer()
+		// fmt.Printf("round %v\n", i)
 		chunker, tester := chunkerAndTester()
 		key, _ := tester.Split(chunker, n)
+		// fmt.Printf("split done %v, joining...\n", i)
 		t.StartTimer()
 		reader := tester.Join(chunker, key, i)
-		result := make([]byte, n)
-		readAll(reader, result)
+		// fmt.Printf("join done %v, reading...\n", i)
+		benchReadAll(reader)
 	}
 }
 

bzz/httpaccess.go

@@ -136,7 +136,12 @@ func handler(w http.ResponseWriter, r *http.Request, dpa *DPA) {
 				size, err := manifestReader.Read(manifest)
 				if int64(size) < manifestReader.Size() {
 					dpaLogger.Debugf("Swarm: Manifest %s not found.", name)
-					http.Error(w, err.Error(), http.StatusNotFound)
+					if err == nil {
+						http.Error(w, "Manifest retrieval cut short: "+string(size)+"<"+string(manifestReader.Size()),
+							http.StatusNotFound)
+					} else {
+						http.Error(w, err.Error(), http.StatusNotFound)
+					}
 					return
 				}
 				dpaLogger.Debugf("Swarm: Manifest %s retrieved.", name)

bzz/memstore.go

@@ -159,6 +159,7 @@ func (s *memStore) getEntryCnt() uint {
 
 }
 
+// entry (not its copy) is going to be in memStore
 func (s *memStore) Put(entry *Chunk) {
 
 	if s.capacity == 0 {
@@ -193,13 +194,15 @@ func (s *memStore) Put(entry *Chunk) {
 
 		if node.entry.Key.isEqual(entry.Key) {
 			node.updateAccess(s.accessCnt)
-			if node.entry.SData == nil {
-				node.entry.Size = entry.Size
-				node.entry.SData = entry.SData
+			if entry.SData == nil {
+				entry.Size = node.entry.Size
+				entry.SData = node.entry.SData
 			}
-			if node.entry.req == nil {
-				node.entry.req = entry.req
+			if entry.req == nil {
+				entry.req = node.entry.req
 			}
+			entry.C = node.entry.C
+			node.entry = entry
 			return
 		}
 
@@ -253,11 +256,7 @@ func (s *memStore) Get(hash Key) (chunk *Chunk, err error) {
 	if node.entry.Key.isEqual(hash) {
 		s.accessCnt++
 		node.updateAccess(s.accessCnt)
-		chunk = &Chunk{
-			Key:   hash,
-			SData: node.entry.SData,
-			Size:  node.entry.Size,
-		}
+		chunk = node.entry
 		if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt {
 			s.dbAccessCnt++
 			node.lastDBaccess = s.dbAccessCnt

bzz/netstore.go

@@ -69,9 +69,10 @@ func (self *NetStore) Put(entry *Chunk) {
 
 func (self *NetStore) put(entry *Chunk) {
 	self.localStore.Put(entry)
-	dpaLogger.Debugf("NetStore.put: localStore.Put of %064x completed.", entry.Key)
-	self.store(entry)
+	dpaLogger.Debugf("NetStore.put: localStore.Put of %064x completed, %d bytes (%p).", entry.Key, len(entry.SData), entry)
+	go self.store(entry)
 	// only send responses once
+	dpaLogger.Debugf("NetStore.put: req: %#v", entry.req)
 	if entry.req != nil && entry.req.status == reqSearching {
 		entry.req.status = reqFound
 		close(entry.req.C)
@@ -82,7 +83,9 @@ func (self *NetStore) put(entry *Chunk) {
 func (self *NetStore) addStoreRequest(req *storeRequestMsgData) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
+	dpaLogger.Debugf("NetStore.addStoreRequest: req = %#v", req)
 	chunk, err := self.localStore.Get(req.Key)
+	dpaLogger.Debugf("NetStore.addStoreRequest: chunk reference %p", chunk)
 	// we assume that a returned chunk is the one stored in the memory cache
 	if err != nil {
 		chunk = &Chunk{
@@ -116,7 +119,7 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
 		dpaLogger.Debugf("NetStore.Get: %064x request time out ", key)
 		err = notFound
 	case <-chunk.req.C:
-		dpaLogger.Debugf("NetStore.get: %064x retrieved", key)
+		dpaLogger.Debugf("NetStore.Get: %064x retrieved, %d bytes (%p)", key, len(chunk.SData), chunk)
 
 	}
 	return
@@ -137,6 +140,7 @@ func (self *NetStore) get(key Key) (chunk *Chunk) {
 
 	if chunk.req == nil {
 		chunk.req = new(requestStatus)
+		chunk.req.C = make(chan bool)
 	}
 	return
 }
@@ -158,8 +162,6 @@ func (self *NetStore) addRetrieveRequest(req *retrieveRequestMsgData) {
 
 	send, timeout := self.strategyUpdateRequest(chunk.req, req) // may change req status
 
-	dpaLogger.Debugf("Is %v == %v?", send, storeRequestMsg)
-
 	if send == storeRequestMsg {
 		dpaLogger.Debugf("NetStore.addRetrieveRequest: %064x - content found, delivering...", req.Key)
 		self.deliver(req, chunk)

cmd/ethereum/flags.go

@@ -66,6 +66,7 @@ var (
 	Dial            bool
 	PrintVersion    bool
 	Peers           string
+	Pull            string
 )
 
 // flags specific to cli client
@@ -105,6 +106,7 @@ func Init() {
 	flag.BoolVar(&Dial, "dial", true, "dial out connections (on)")
 	flag.BoolVar(&GenAddr, "genaddr", false, "create a new priv/pub key")
 	flag.StringVar(&Peers, "peers", "", "imports the file given (hex or mnemonic formats)")
+	flag.StringVar(&Pull, "pull", "", "swarm pull key")
 	flag.StringVar(&SecretFile, "import", "", "imports the file given (hex or mnemonic formats)")
 	flag.StringVar(&ExportDir, "export", "", "exports the session keyring to files in the directory given")
 	flag.StringVar(&LogFile, "logfile", "", "log file (defaults to standard output)")

cmd/ethereum/main.go

@@ -134,7 +134,7 @@ func main() {
 		utils.StartWebSockets(ethereum)
 	}
 
-	utils.StartEthereum(ethereum, UseSeed, Peers)
+	utils.StartEthereum(ethereum, UseSeed, Peers, Pull)
 
 	if StartJsConsole {
 		InitJsConsole(ethereum)

cmd/utils/cmd.go

@@ -120,9 +120,9 @@ func exit(err error) {
 	os.Exit(status)
 }
 
-func StartEthereum(ethereum *eth.Ethereum, UseSeed bool, Peers string) {
+func StartEthereum(ethereum *eth.Ethereum, UseSeed bool, Peers string, Pull string) {
 	clilogger.Infof("Starting %s", ethereum.ClientIdentity())
-	err := ethereum.Start(UseSeed, Peers)
+	err := ethereum.Start(UseSeed, Peers, Pull)
 	if err != nil {
 		exit(err)
 	}

eth/backend.go

@@ -2,9 +2,11 @@ package eth
 
 import (
 	"fmt"
+	"io"
 	"net"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/bzz"
 	"github.com/ethereum/go-ethereum/core"
@@ -70,6 +72,7 @@ type Ethereum struct {
 
 	RpcServer  *rpc.JsonRpcServer
 	keyManager *crypto.KeyManager
+	dpa        *bzz.DPA
 
 	clientIdentity p2p.ClientIdentity
 	logger         ethlogger.LogSystem
@@ -144,12 +147,12 @@ func New(config *Config) (*Ethereum, error) {
 	}
 	chunker := &bzz.TreeChunker{}
 	chunker.Init()
-	dpa := &bzz.DPA{
+	eth.dpa = &bzz.DPA{
 		Chunker:    chunker,
 		ChunkStore: netStore,
 	}
-	dpa.Start()
-	go bzz.StartHttpServer(dpa)
+	eth.dpa.Start()
+	go bzz.StartHttpServer(eth.dpa)
 
 	nat, err := p2p.ParseNAT(config.NATType, config.PMPGateway)
 	if err != nil {
@@ -234,7 +237,7 @@ func (s *Ethereum) MaxPeers() int {
 }
 
 // Start the ethereum
-func (s *Ethereum) Start(seed bool, p string) error {
+func (s *Ethereum) Start(seed bool, p string, pull string) error {
 	err := s.net.Start()
 	if err != nil {
 		return err
@@ -264,6 +267,28 @@ func (s *Ethereum) Start(seed bool, p string) error {
 		}
 	}
 
+	if len(pull) > 0 {
+		go func() {
+			time.Sleep(30 * time.Second)
+			key := ethutil.Hex2Bytes(pull)
+			reader := s.dpa.Retrieve(key)
+			logger.Debugf("retrieved reader for %064x", key)
+			b := make([]byte, 0x1000)
+			var length int64
+			for {
+				n, err := reader.Read(b)
+				if err != nil {
+					if err != io.EOF {
+						logger.Debugf("read %v bytes. read error for %064x: %v", n, key, err)
+					}
+					return
+				}
+				length += int64(n)
+			}
+			logger.Debugf("read %v bytes from %064x: %v", length, key)
+		}()
+	}
+
 	// TODO: read peers here
 	if seed {
 		logger.Infof("Connect to seed node %v", seedNodeAddress)

p2p/message.go

@@ -102,17 +102,22 @@ func writeMsg(w io.Writer, msg Msg) error {
 	copy(start, magicToken)
 	binary.BigEndian.PutUint32(start[4:], payloadLen)
 
-	srvlog.Debugf("Sending message:")
-
+	if msg.Size > 1 {
+		srvlog.Debugf("Sending message (size %v):", msg.Size)
+	}
 	for _, b := range [][]byte{start, listhdr, code} {
-		srvlog.Debugf("  %x", b)
+		if msg.Size > 1 {
+			srvlog.Debugf("  %x", b)
+		}
 		if _, err := w.Write(b); err != nil {
 			return err
 		}
 	}
 	b := make([]byte, msg.Size)
 	msg.Payload.Read(b)
-	srvlog.Debugf(" %x", b)
+	if msg.Size > 1 {
+		srvlog.Debugf(" %x", b)
+	}
 	_, err := w.Write(b)
 	return err
 }