From 09713314c7c9121433c5fad06d9c8b866beb7e22 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Sun, 22 Apr 2018 13:10:08 +0200 Subject: [PATCH 01/10] swarm/storage: more error logging for Read and ReadAt --- swarm/storage/chunker.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index c2bac98593..7e2b3be646 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -393,7 +393,7 @@ type LazyChunkReader struct { } // implements the Joiner interface -func (self *TreeChunker) Join() LazySectionReader { +func (self *TreeChunker) Join() *LazyChunkReader { return &LazyChunkReader{ key: self.key, chunkSize: self.chunkSize, @@ -436,6 +436,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { quitC := make(chan bool) size, err := self.Size(quitC) if err != nil { + log.Error("lazychunkreader.readat.size", "size", size, "err", err) return 0, err } @@ -464,6 +465,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { err = <-errC if err != nil { + log.Error("lazychunkreader.readat.errc", "err", err) close(quitC) return 0, err } @@ -517,6 +519,7 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr childKey := chunkData[8+j*self.hashSize : 8+(j+1)*self.hashSize] chunkData, err := self.getter.Get(Reference(childKey)) if err != nil { + log.Error("lazychunkreader.join", "key", childKey, "err", err) select { case errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize): case <-quitC: @@ -535,6 +538,9 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr func (self *LazyChunkReader) Read(b []byte) (read int, err error) { log.Debug("lazychunkreader.read", "key", self.key) read, err = self.ReadAt(b, self.off) + if err != nil { + log.Error("lazychunkreader.readat", "read", read, "err", err) + } self.off += int64(read) return From 426bc4f26b2694c174fd8bd800add37ef2283ea6 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Sun, 22 Apr 2018 13:28:16 +0200 Subject: [PATCH 02/10] swarm: for loop on reader --- swarm/api/http/server.go | 18 +++++++++++++++++- swarm/storage/chunker.go | 2 +- swarm/storage/dpa.go | 2 +- 3 files changed, 19 insertions(+), 3 deletions(-) diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index c5f97ab775..4ce832863f 100644 --- a/swarm/api/http/server.go +++ b/swarm/api/http/server.go @@ -574,7 +574,23 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) { contentType = typ } w.Header().Set("Content-Type", contentType) - http.ServeContent(w, &r.Request, "", time.Now(), reader) + + var res []byte + var err error + for { + res, err = ioutil.ReadAll(reader) + if err != nil { + log.Error("handle.get", "ruid", r.ruid, "error", err) + time.Sleep(200 * time.Millisecond) + continue + } + log.Debug("handle.get.readall success", "ruid", r.ruid) + break + } + + rdr := bytes.NewReader(res) + + http.ServeContent(w, &r.Request, "", time.Now(), rdr) case r.uri.Hash(): w.Header().Set("Content-Type", "text/plain") w.WriteHeader(http.StatusOK) diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index 7e2b3be646..f9b3fe2746 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -131,7 +131,7 @@ type TreeChunker struct { The chunks are not meant to be validated by the chunker when joining. This is because it is left to the DPA to decide which sources are trusted. */ -func TreeJoin(key Key, getter Getter, depth int) LazySectionReader { +func TreeJoin(key Key, getter Getter, depth int) *LazyChunkReader { return NewTreeJoiner(NewJoinerParams(key, getter, depth, DefaultChunkSize)).Join() } diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index edf81ad5cb..2257df877c 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -84,7 +84,7 @@ func NewDPA(store ChunkStore, params *DPAParams) *DPA { // Chunk retrieval blocks on netStore requests with a timeout so reader will // report error if retrieval of chunks within requested range time out. // It returns a reader with the chunk data and whether the content was encrypted -func (self *DPA) Retrieve(key Key) (reader LazySectionReader, isEncrypted bool) { +func (self *DPA) Retrieve(key Key) (reader *LazyChunkReader, isEncrypted bool) { isEncrypted = len(key) > self.hashFunc().Size() getter := NewHasherStore(self.ChunkStore, self.hashFunc, isEncrypted) reader = TreeJoin(key, getter, 0) From b775c55622bc9ae1f4b95bf48032b2e9e49d8de7 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Sun, 22 Apr 2018 13:42:09 +0200 Subject: [PATCH 03/10] swarm: remove spamy logs --- swarm/network/stream/peer.go | 2 +- swarm/network/stream/syncer.go | 3 +-- swarm/storage/memstore.go | 3 +-- 3 files changed, 3 insertions(+), 5 deletions(-) diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 79afd5d425..be46f32df0 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -120,7 +120,7 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { To: to, Stream: s.stream, } - log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to) + //log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to) return p.SendPriority(msg, s.priority) } diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index 9ecd2fe25a..5da3eb4881 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -21,7 +21,6 @@ import ( "strconv" "time" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -130,7 +129,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 ticker.Stop() } - log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po)) + //log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po)) return batch, from, to, nil, nil } diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go index 7266bc92be..d1b9521ede 100644 --- a/swarm/storage/memstore.go +++ b/swarm/storage/memstore.go @@ -21,7 +21,6 @@ package storage import ( "sync" - "github.com/ethereum/go-ethereum/log" lru "github.com/hashicorp/golang-lru" ) @@ -55,7 +54,7 @@ func NewMemStore(params *StoreParams, _ *LDBStore) (m *MemStore) { } requestEvicted := func(key interface{}, value interface{}) { - log.Error("evict called on outgoing request") + //log.Error("evict called on outgoing request") } r, err := lru.NewWithEvict(int(params.ChunkRequestsCacheCapacity), requestEvicted) if err != nil { From f88ac098cbc5c2b4d6931cdd443888c7ee738f23 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Sun, 22 Apr 2018 14:11:52 +0200 Subject: [PATCH 04/10] swarm/api: for loop for GetFile --- swarm/api/api.go | 2 +- swarm/api/http/server.go | 16 +++++++++++++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/swarm/api/api.go b/swarm/api/api.go index 1e08b71d35..84ac95769b 100644 --- a/swarm/api/api.go +++ b/swarm/api/api.go @@ -308,7 +308,7 @@ func (self *Api) Put(content, contentType string, toEncrypt bool) (k storage.Key // Get uses iterative manifest retrieval and prefix matching // to resolve basePath to content using dpa retrieve // it returns a section reader, mimeType, status and an error -func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionReader, mimeType string, status int, err error) { +func (self *Api) Get(key storage.Key, path string) (reader *storage.LazyChunkReader, mimeType string, status int, err error) { log.Debug("api.get", "key", key, "path", path) apiGetCount.Inc(1) trie, err := loadManifest(self.dpa, key, nil) diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index 4ce832863f..aab253e12b 100644 --- a/swarm/api/http/server.go +++ b/swarm/api/http/server.go @@ -844,9 +844,23 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) { return } + var res []byte + for { + res, err = ioutil.ReadAll(reader) + if err != nil { + log.Error("handle.get.file", "ruid", r.ruid, "error", err) + time.Sleep(200 * time.Millisecond) + continue + } + log.Debug("handle.get.file.readall success", "ruid", r.ruid) + break + } + + rdr := bytes.NewReader(res) + w.Header().Set("Content-Type", contentType) - http.ServeContent(w, &r.Request, "", time.Now(), reader) + http.ServeContent(w, &r.Request, "", time.Now(), rdr) } func (s *Server) ServeHTTP(rw http.ResponseWriter, r *http.Request) { From b68c2d35d86f404ced59af22982a72424efcbb01 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Sun, 22 Apr 2018 14:26:07 +0200 Subject: [PATCH 05/10] swarm/api: get rid of endless loop on readall --- swarm/api/http/server.go | 30 ++++++++++++------------------ 1 file changed, 12 insertions(+), 18 deletions(-) diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index aab253e12b..3e0c13470e 100644 --- a/swarm/api/http/server.go +++ b/swarm/api/http/server.go @@ -577,16 +577,13 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) { var res []byte var err error - for { - res, err = ioutil.ReadAll(reader) - if err != nil { - log.Error("handle.get", "ruid", r.ruid, "error", err) - time.Sleep(200 * time.Millisecond) - continue - } - log.Debug("handle.get.readall success", "ruid", r.ruid) - break + res, err = ioutil.ReadAll(reader) + if err != nil { + log.Error("handle.get", "ruid", r.ruid, "error", err) + Respond(w, r, fmt.Sprintf("chunk not found: %s", err), http.StatusNotFound) + return } + log.Debug("handle.get.readall success", "ruid", r.ruid) rdr := bytes.NewReader(res) @@ -845,16 +842,13 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) { } var res []byte - for { - res, err = ioutil.ReadAll(reader) - if err != nil { - log.Error("handle.get.file", "ruid", r.ruid, "error", err) - time.Sleep(200 * time.Millisecond) - continue - } - log.Debug("handle.get.file.readall success", "ruid", r.ruid) - break + res, err = ioutil.ReadAll(reader) + if err != nil { + log.Error("handle.get.file", "ruid", r.ruid, "error", err) + Respond(w, r, fmt.Sprintf("chunk not found %s: %s", r.uri, err), http.StatusNotFound) + return } + log.Debug("handle.get.file.readall success", "ruid", r.ruid) rdr := bytes.NewReader(res) From 2b5c0d0820c48c3d7735d6ff6be839b4b9762888 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Sun, 22 Apr 2018 15:18:53 +0200 Subject: [PATCH 06/10] swarm/api: comment out ioutil.ReadAll --- cmd/swarm/swarm-smoke/upload_and_sync.go | 2 +- swarm/api/http/server.go | 20 ++++++++++---------- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/cmd/swarm/swarm-smoke/upload_and_sync.go b/cmd/swarm/swarm-smoke/upload_and_sync.go index 542c4508db..1733d19bdc 100644 --- a/cmd/swarm/swarm-smoke/upload_and_sync.go +++ b/cmd/swarm/swarm-smoke/upload_and_sync.go @@ -69,7 +69,7 @@ func fetch(hash string, endpoint string, original []byte, ruid string) error { time.Sleep(10 * time.Second) log.Trace("http get request", "ruid", ruid, "api", endpoint, "hash", hash) - res, err := http.Get(endpoint + "/bzz:/" + hash) + res, err := http.Get(endpoint + "/bzz:/" + hash + "/") if err != nil { log.Warn(err.Error(), "ruid", ruid) return err diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index 3e0c13470e..54d68ca81f 100644 --- a/swarm/api/http/server.go +++ b/swarm/api/http/server.go @@ -841,20 +841,20 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) { return } - var res []byte - res, err = ioutil.ReadAll(reader) - if err != nil { - log.Error("handle.get.file", "ruid", r.ruid, "error", err) - Respond(w, r, fmt.Sprintf("chunk not found %s: %s", r.uri, err), http.StatusNotFound) - return - } - log.Debug("handle.get.file.readall success", "ruid", r.ruid) + //var res []byte + //res, err = ioutil.ReadAll(reader) + //if err != nil { + //log.Error("handle.get.file", "ruid", r.ruid, "error", err) + //Respond(w, r, fmt.Sprintf("chunk not found %s: %s", r.uri, err), http.StatusNotFound) + //return + //} + //log.Debug("handle.get.file.readall success", "ruid", r.ruid) - rdr := bytes.NewReader(res) + //rdr := bytes.NewReader(res) w.Header().Set("Content-Type", contentType) - http.ServeContent(w, &r.Request, "", time.Now(), rdr) + http.ServeContent(w, &r.Request, "", time.Now(), reader) } func (s *Server) ServeHTTP(rw http.ResponseWriter, r *http.Request) { From 2ead337b48fc6288f22008f3451069df63c1afdc Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Sun, 22 Apr 2018 15:41:18 +0200 Subject: [PATCH 07/10] swarm/storage: improve tracing for missing reference --- swarm/storage/chunker.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index f9b3fe2746..40af0499fa 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -519,9 +519,9 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr childKey := chunkData[8+j*self.hashSize : 8+(j+1)*self.hashSize] chunkData, err := self.getter.Get(Reference(childKey)) if err != nil { - log.Error("lazychunkreader.join", "key", childKey, "err", err) + log.Error("lazychunkreader.join", "key", fmt.Sprintf("%x", childKey), "err", err) select { - case errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize): + case errC <- fmt.Errorf("chunk %v-%v not found; key: %s", off, off+treeSize, fmt.Sprintf("%x", childKey)): case <-quitC: } return From 7785f29b6bd080841deb78d7759a577736864b19 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Sun, 22 Apr 2018 16:08:57 +0200 Subject: [PATCH 08/10] swarm/storage: EOF is not an error --- cmd/swarm/swarm-smoke/upload_and_sync.go | 8 +++++++- swarm/storage/chunker.go | 2 +- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/cmd/swarm/swarm-smoke/upload_and_sync.go b/cmd/swarm/swarm-smoke/upload_and_sync.go index 1733d19bdc..b490098809 100644 --- a/cmd/swarm/swarm-smoke/upload_and_sync.go +++ b/cmd/swarm/swarm-smoke/upload_and_sync.go @@ -40,6 +40,12 @@ func cliUploadAndSync(c *cli.Context) error { log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fhash)) + if filesize < 10 { + time.Sleep(15 * time.Second) + } else { + time.Sleep(2 * time.Duration(filesize) * time.Second) + } + wg := sync.WaitGroup{} for _, endpoint := range endpoints { endpoint := endpoint @@ -66,7 +72,7 @@ func cliUploadAndSync(c *cli.Context) error { // fetch is getting the requested `hash` from the `endpoint` and compares it with the `original` file func fetch(hash string, endpoint string, original []byte, ruid string) error { log.Trace("sleeping", "ruid", ruid) - time.Sleep(10 * time.Second) + time.Sleep(1 * time.Second) log.Trace("http get request", "ruid", ruid, "api", endpoint, "hash", hash) res, err := http.Get(endpoint + "/bzz:/" + hash + "/") diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index 40af0499fa..b01d724e15 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -538,7 +538,7 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr func (self *LazyChunkReader) Read(b []byte) (read int, err error) { log.Debug("lazychunkreader.read", "key", self.key) read, err = self.ReadAt(b, self.off) - if err != nil { + if err != nil && err != io.EOF { log.Error("lazychunkreader.readat", "read", read, "err", err) } From 3006eb768fad3da2e8534d15786d390f21d0af80 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Mon, 23 Apr 2018 12:16:37 +0200 Subject: [PATCH 09/10] swarm: remove commented out code --- swarm/api/http/server.go | 26 +------------------------- swarm/network/stream/peer.go | 2 +- swarm/network/stream/syncer.go | 3 ++- swarm/storage/memstore.go | 3 ++- 4 files changed, 6 insertions(+), 28 deletions(-) diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index 54d68ca81f..c5f97ab775 100644 --- a/swarm/api/http/server.go +++ b/swarm/api/http/server.go @@ -574,20 +574,7 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) { contentType = typ } w.Header().Set("Content-Type", contentType) - - var res []byte - var err error - res, err = ioutil.ReadAll(reader) - if err != nil { - log.Error("handle.get", "ruid", r.ruid, "error", err) - Respond(w, r, fmt.Sprintf("chunk not found: %s", err), http.StatusNotFound) - return - } - log.Debug("handle.get.readall success", "ruid", r.ruid) - - rdr := bytes.NewReader(res) - - http.ServeContent(w, &r.Request, "", time.Now(), rdr) + http.ServeContent(w, &r.Request, "", time.Now(), reader) case r.uri.Hash(): w.Header().Set("Content-Type", "text/plain") w.WriteHeader(http.StatusOK) @@ -841,17 +828,6 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) { return } - //var res []byte - //res, err = ioutil.ReadAll(reader) - //if err != nil { - //log.Error("handle.get.file", "ruid", r.ruid, "error", err) - //Respond(w, r, fmt.Sprintf("chunk not found %s: %s", r.uri, err), http.StatusNotFound) - //return - //} - //log.Debug("handle.get.file.readall success", "ruid", r.ruid) - - //rdr := bytes.NewReader(res) - w.Header().Set("Content-Type", contentType) http.ServeContent(w, &r.Request, "", time.Now(), reader) diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index be46f32df0..79afd5d425 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -120,7 +120,7 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { To: to, Stream: s.stream, } - //log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to) + log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to) return p.SendPriority(msg, s.priority) } diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index 5da3eb4881..ae4071ba4e 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -21,6 +21,7 @@ import ( "strconv" "time" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -129,7 +130,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 ticker.Stop() } - //log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po)) + log.Trace("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po)) return batch, from, to, nil, nil } diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go index d1b9521ede..7266bc92be 100644 --- a/swarm/storage/memstore.go +++ b/swarm/storage/memstore.go @@ -21,6 +21,7 @@ package storage import ( "sync" + "github.com/ethereum/go-ethereum/log" lru "github.com/hashicorp/golang-lru" ) @@ -54,7 +55,7 @@ func NewMemStore(params *StoreParams, _ *LDBStore) (m *MemStore) { } requestEvicted := func(key interface{}, value interface{}) { - //log.Error("evict called on outgoing request") + log.Error("evict called on outgoing request") } r, err := lru.NewWithEvict(int(params.ChunkRequestsCacheCapacity), requestEvicted) if err != nil { From 312f8120e118ecd4d6db363746e4a70ab656cb6f Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Mon, 23 Apr 2018 17:49:35 +0200 Subject: [PATCH 10/10] swarm/storage: remove redundant comment --- swarm/storage/chunker.go | 1 - 1 file changed, 1 deletion(-) diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index b01d724e15..41c0c17c12 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -392,7 +392,6 @@ type LazyChunkReader struct { getter Getter } -// implements the Joiner interface func (self *TreeChunker) Join() *LazyChunkReader { return &LazyChunkReader{ key: self.key,