diff --git a/bzz/chunker.go b/bzz/chunker.go index 28b3ba663c..32015bf29e 100644 --- a/bzz/chunker.go +++ b/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])) diff --git a/bzz/chunker_test.go b/bzz/chunker_test.go index 5238f11915..1cff67383b 100644 --- a/bzz/chunker_test.go +++ b/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) } } diff --git a/bzz/httpaccess.go b/bzz/httpaccess.go index 9563713d7d..1b95b0c335 100644 --- a/bzz/httpaccess.go +++ b/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) diff --git a/bzz/memstore.go b/bzz/memstore.go index 8975e80d49..0c5f17d5be 100644 --- a/bzz/memstore.go +++ b/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 diff --git a/bzz/netstore.go b/bzz/netstore.go index aa3e867659..7053ddd596 100644 --- a/bzz/netstore.go +++ b/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) diff --git a/cmd/ethereum/flags.go b/cmd/ethereum/flags.go index d69ea5cfb6..b3be4de885 100644 --- a/cmd/ethereum/flags.go +++ b/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)") diff --git a/cmd/ethereum/main.go b/cmd/ethereum/main.go index ea53b9a304..8a549dbea0 100644 --- a/cmd/ethereum/main.go +++ b/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) diff --git a/cmd/utils/cmd.go b/cmd/utils/cmd.go index 06e3fb93f1..f6e7999e75 100644 --- a/cmd/utils/cmd.go +++ b/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) } diff --git a/eth/backend.go b/eth/backend.go index f5163bdfd5..4c6cd22d06 100644 --- a/eth/backend.go +++ b/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) diff --git a/p2p/message.go b/p2p/message.go index 7eaa884069..cabd1940bd 100644 --- a/p2p/message.go +++ b/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 }