diff --git a/swarm/api/api.go b/swarm/api/api.go index cc76b36ae5..321837bda8 100644 --- a/swarm/api/api.go +++ b/swarm/api/api.go @@ -43,7 +43,7 @@ func NewApi(dpa *storage.DPA, dns Resolver) (self *Api) { } // DPA reader API -func (self *Api) Retrieve(key storage.Key) storage.SectionReader { +func (self *Api) Retrieve(key storage.Key) storage.LazySectionReader { return self.dpa.Retrieve(key) } @@ -124,11 +124,11 @@ func (self *Api) Put(content, contentType string) (string, error) { // Get uses iterative manifest retrieval and prefix matching // to resolve path to content using dpa retrieve // it returns a section reader, mimeType, status and an error -func (self *Api) Get(uri string, nameresolver bool) (reader storage.SectionReader, mimeType string, status int, err error) { +func (self *Api) Get(uri string, nameresolver bool) (reader storage.LazySectionReader, mimeType string, status int, err error) { key, _, path, err := self.parseAndResolve(uri, nameresolver) - - trie, err := loadManifest(self.dpa, key) + quitC := make(chan bool) + trie, err := loadManifest(self.dpa, key, quitC) if err != nil { glog.V(logger.Warn).Infof("[BZZ] Swarm: loadManifestTrie error: %v", err) return @@ -151,7 +151,8 @@ func (self *Api) Get(uri string, nameresolver bool) (reader storage.SectionReade func (self *Api) Modify(uri, contentHash, contentType string, nameresolver bool) (newRootHash string, err error) { root, _, path, err := self.parseAndResolve(uri, nameresolver) - trie, err := loadManifest(self.dpa, root) + quitC := make(chan bool) + trie, err := loadManifest(self.dpa, root, quitC) if err != nil { return } @@ -162,9 +163,9 @@ func (self *Api) Modify(uri, contentHash, contentType string, nameresolver bool) Hash: contentHash, ContentType: contentType, } - trie.addEntry(entry) + trie.addEntry(entry, quitC) } else { - trie.deleteEntry(path) + trie.deleteEntry(path, quitC) } err = trie.recalcAndStore() diff --git a/swarm/api/api_test.go b/swarm/api/api_test.go index 7db25c1fda..0fb50e5793 100644 --- a/swarm/api/api_test.go +++ b/swarm/api/api_test.go @@ -6,6 +6,8 @@ import ( "os" "testing" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -27,7 +29,7 @@ func testApi(t *testing.T, f func(*Api)) { } type testResponse struct { - reader storage.SectionReader + reader storage.LazySectionReader *Response } @@ -52,10 +54,13 @@ func checkResponse(t *testing.T, resp *testResponse, exp *Response) { } if resp.Content != exp.Content { // if !bytes.Equal(resp.Content, exp.Content) - t.Errorf("incorrect content. expected '%s...', got '%s...'", string(exp.Content), string(resp.Con} + t.Errorf("incorrect content. expected '%s...', got '%s...'", string(exp.Content), string(resp.Content)) + } +} // func expResponse(content []byte, mimeType string, status int) *Response { func expResponse(content string, mimeType string, status int) *Response { + glog.V(logger.Detail).Infof("expected content (%v): %v ", len(content), content) return &Response{mimeType, status, int64(len(content)), content} } @@ -65,13 +70,19 @@ func testGet(t *testing.T, api *Api, bzzhash string) *testResponse { if err != nil { t.Fatalf("unexpected error: %v", err) } - - s := make([]byte, reader.Size()) + quitC := make(chan bool) + size, err := reader.Size(quitC) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + glog.V(logger.Detail).Infof("reader size: %v ", size) + s := make([]byte, size) _, err = reader.Read(s) if err != io.EOF { t.Fatalf("unexpected error: %v", err) } - return &testResponse{reader, &Response{mimeType, status, reader.Size(), string(s)}} + reader.Seek(0, 0) + return &testResponse{reader, &Response{mimeType, status, size, string(s)}} // return &testResponse{reader, &Response{mimeType, status, reader.Size(), nil}} } diff --git a/swarm/api/filesystem.go b/swarm/api/filesystem.go index 34f2a8f0c8..cc8fd2c818 100644 --- a/swarm/api/filesystem.go +++ b/swarm/api/filesystem.go @@ -86,6 +86,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) { errors := make([]error, cnt) done := make(chan bool, maxParallelFiles) dcnt := 0 + wg := &sync.WaitGroup{} for i, entry := range list { if i >= dcnt+maxParallelFiles { @@ -96,7 +97,6 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) { f, err := os.Open(entry.Path) if err == nil { stat, _ := f.Stat() - wg := &sync.WaitGroup{} var hash storage.Key hash, err = self.api.dpa.Store(f, stat.Size(), wg) if hash != nil { @@ -128,6 +128,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) { trie := &manifestTrie{ dpa: self.api.dpa, } + quitC := make(chan bool) for i, entry := range list { if errors[i] != nil { return "", errors[i] @@ -139,9 +140,9 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) { Hash: entry.Hash, ContentType: entry.ContentType, } - trie.addEntry(ientry) + trie.addEntry(ientry, quitC) } - trie.addEntry(entry) + trie.addEntry(entry, quitC) } err2 := trie.recalcAndStore() @@ -149,6 +150,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) { if err2 == nil { hs = trie.hash.String() } + wg.Wait() return hs, err2 } @@ -169,11 +171,13 @@ func (self *FileSystem) Download(bzzpath, localpath string) error { if err != nil { return err } - // if len(path) > 0 { - // path += "/" - // } - trie, err := loadManifest(self.api.dpa, key) + if len(path) > 0 { + path += "/" + } + + quitC := make(chan bool) + trie, err := loadManifest(self.api.dpa, key, quitC) if err != nil { glog.V(logger.Warn).Infof("[BZZ] fs.Download: loadManifestTrie error: %v", err) return err @@ -185,72 +189,76 @@ func (self *FileSystem) Download(bzzpath, localpath string) error { } var list []*downloadListEntry - var mde, mderr error + var mde error prevPath := lpath - err = trie.listWithPrefix(path, func(entry *manifestTrieEntry, suffix string) { // TODO: paralellize + err = trie.listWithPrefix(path, quitC, func(entry *manifestTrieEntry, suffix string) { glog.V(logger.Detail).Infof("[BZZ] fs.Download: %#v", entry) - key := common.Hex2Bytes(entry.Hash) + key = common.Hex2Bytes(entry.Hash) path := lpath + "/" + suffix dir := filepath.Dir(path) if dir != prevPath { mde = os.MkdirAll(dir, os.ModePerm) - if mde != nil { - mderr = mde - } prevPath = dir } if (mde == nil) && (path != dir+"/") { list = append(list, &downloadListEntry{key: key, path: path}) } }) - if err == nil { - err = mderr - } - - cnt := len(list) - errors := make([]error, cnt) - done := make(chan bool, maxParallelFiles) - dcnt := 0 - - for i, entry := range list { - if i >= dcnt+maxParallelFiles { - <-done - dcnt++ - } - go func(i int, entry *downloadListEntry, done chan bool) { - f, err := os.Create(entry.path) // TODO: path separators - if err == nil { - reader := self.api.dpa.Retrieve(entry.key) - writer := bufio.NewWriter(f) - _, err = io.CopyN(writer, reader, reader.Size()) // TODO: handle errors - err2 := writer.Flush() - if err == nil { - err = err2 - } - err2 = f.Close() - if err == nil { - err = err2 - } - } - - errors[i] = err - done <- true - }(i, entry, done) - } - for dcnt < cnt { - <-done - dcnt++ - } - if err != nil { return err } - for i, _ := range list { - if errors[i] != nil { - return errors[i] + + wg := sync.WaitGroup{} + errC := make(chan error) + done := make(chan bool, maxParallelFiles) + for i, entry := range list { + select { + case done <- true: + wg.Add(1) + case <-quitC: + return fmt.Errorf("aborted") } + go func(i int, entry *downloadListEntry) { + defer wg.Done() + f, err := os.Create(entry.path) // TODO: path separators + if err == nil { + + reader := self.api.dpa.Retrieve(entry.key) + writer := bufio.NewWriter(f) + size, err := reader.Size(quitC) + if err == nil { + _, err = io.CopyN(writer, reader, size) // TODO: handle errors + err2 := writer.Flush() + if err == nil { + err = err2 + } + err2 = f.Close() + if err == nil { + err = err2 + } + } + } + if err != nil { + select { + case errC <- err: + case <-quitC: + } + return + } + <-done + }(i, entry) } - return err + go func() { + wg.Wait() + close(errC) + }() + select { + case err = <-errC: + return err + case <-quitC: + return fmt.Errorf("aborted") + } + } diff --git a/swarm/api/filesystem_test.go b/swarm/api/filesystem_test.go index c6be66e68f..f26c5df0ab 100644 --- a/swarm/api/filesystem_test.go +++ b/swarm/api/filesystem_test.go @@ -1,16 +1,14 @@ package api import ( - "io" + "bytes" "io/ioutil" "os" "path" "runtime" + "sync" "testing" - -"github.com/ethereum/go-ethereum/logger" - "github.com/ethereum/go-ethereum/logger/glog" - ) +) var ( testDir string @@ -21,8 +19,8 @@ func init() { _, filename, _, _ := runtime.Caller(1) testDir = path.Join(path.Dir(filename), "../test") testDownloadDir, _ = ioutil.TempDir(os.TempDir(), "bzz-test") - glog.SetV(logger.Detail) - glog.SetV = nil +} + func testFileSystem(t *testing.T, f func(*FileSystem)) { testApi(t, func(api *Api) { f(NewFileSystem(api)) @@ -30,7 +28,6 @@ func testFileSystem(t *testing.T, f func(*FileSystem)) { } func readPath(t *testing.T, parts ...string) string { - // func readPath(t *testing.T, parts ...string) []byte { file := path.Join(parts...) content, err := ioutil.ReadFile(file) @@ -41,39 +38,28 @@ func readPath(t *testing.T, parts ...string) string { } func TestApiDirUpload0(t *testing.T) { - // t.Skip("FIXME") testFileSystem(t, func(fs *FileSystem) { api := fs.api bzzhash, err := fs.Upload(path.Join(testDir, "test0"), "") if err != nil { t.Fatalf("unexpected error: %v", err) } - content := readPath(t, testDir, "test0", "index.html") - t.Logf("content (%v): %v ", len(content), content) resp := testGet(t, api, bzzhash+"/index.html") - exp := expRes ponse(content, "text/html; charset=utf-8", 0) - t.Logf("index.html (size%v=?=%v)", resp.Size, exp.Size) + exp := expResponse(content, "text/html; charset=utf-8", 0) checkResponse(t, resp, exp) - t.FailNow() content = readPath(t, testDir, "test0", "index.css") resp = testGet(t, api, bzzhash+"/index.css") exp = expResponse(content, "text/css", 0) - t.Logf("index.css (size%v=?=%v)", resp.Size, exp.Size) checkResponse(t, resp, exp) - content = readPath(t, testDir, "test0", "img", "logo.png") - resp = testGet(t, api, bzzhash+"/img/logo.png") - exp = expResponse(content, "image/png", 0) - _, _, _, err = api.Get(bzzhash, true) if err == nil { t.Fatalf("expected error: %v", err) } downloadDir := path.Join(testDownloadDir, "test0") - os.RemoveAll(downloadDir) defer os.RemoveAll(downloadDir) err = fs.Download(bzzhash, downloadDir) if err != nil { @@ -86,12 +72,10 @@ func TestApiDirUpload0(t *testing.T) { if bzzhash != newbzzhash { t.Fatalf("download %v reuploaded has incorrect hash, expected %v, got %v", downloadDir, bzzhash, newbzzhash) } - }) } func TestApiDirUploadModify(t *testing.T) { - // t.Skip("FIXME") testFileSystem(t, func(fs *FileSystem) { api := fs.api bzzhash, err := fs.Upload(path.Join(testDir, "test0"), "") @@ -105,12 +89,24 @@ func TestApiDirUploadModify(t *testing.T) { t.Errorf("unexpected error: %v", err) return } - bzzhash, err = api.Modify(bzzhash+"/index2.html", "9ea1f60ebd80786d6005f6b256376bdb494a82496cd86fe8c307cdfb23c99e71", "text/html; charset=utf-8", true) + index, err := ioutil.ReadFile(path.Join(testDir, "test0", "index.html")) if err != nil { t.Errorf("unexpected error: %v", err) return } - bzzhash, err = api.Modify(bzzhash+"/img/logo.png", "9ea1f60ebd80786d6005f6b256376bdb494a82496cd86fe8c307cdfb23c99e71", "text/html; charset=utf-8", true) + wg := &sync.WaitGroup{} + hash, err := api.Store(bytes.NewReader(index), int64(len(index)), wg) + wg.Wait() + if err != nil { + t.Errorf("unexpected error: %v", err) + return + } + bzzhash, err = api.Modify(bzzhash+"/index2.html", hash.Hex(), "text/html; charset=utf-8", true) + if err != nil { + t.Errorf("unexpected error: %v", err) + return + } + bzzhash, err = api.Modify(bzzhash+"/img/logo.png", hash.Hex(), "text/html; charset=utf-8", true) if err != nil { t.Errorf("unexpected error: %v", err) return @@ -172,7 +168,7 @@ func TestApiFileUploadWithRootFile(t *testing.T) { testFileSystem(t, func(fs *FileSystem) { api := fs.api bzzhash, err := fs.Upload(path.Join(testDir, "test0", "index.html"), "index.html") - if err != io.EOF { + if err != nil { t.Errorf("unexpected error: %v", err) return } diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index 29945a9f77..f5810c5c0a 100644 --- a/swarm/api/http/server.go +++ b/swarm/api/http/server.go @@ -164,7 +164,9 @@ func handler(w http.ResponseWriter, r *http.Request, a *api.Api) { // retrieving content reader := a.Retrieve(key) - glog.V(logger.Debug).Infof("[BZZ] Swarm: Reading %d bytes.", reader.Size()) + quitC := make(chan bool) + size, err := reader.Size(quitC) + glog.V(logger.Debug).Infof("[BZZ] Swarm: Reading %d bytes.", size) // setting mime type qv := requestURL.Query() @@ -175,7 +177,7 @@ func handler(w http.ResponseWriter, r *http.Request, a *api.Api) { w.Header().Set("Content-Type", mimeType) http.ServeContent(w, r, uri, forever(), reader) - glog.V(logger.Debug).Infof("[BZZ] Swarm: Serve raw content '%s' (%d bytes) as '%s'", uri, reader.Size(), mimeType) + glog.V(logger.Debug).Infof("[BZZ] Swarm: Serve raw content '%s' (%d bytes) as '%s'", uri, size, mimeType) // retrieve path via manifest } else { @@ -202,7 +204,9 @@ func handler(w http.ResponseWriter, r *http.Request, a *api.Api) { } else { status = 200 } - glog.V(logger.Debug).Infof("[BZZ] Swarm: Served '%s' (%d bytes) as '%s' (status code: %v)", uri, reader.Size(), mimeType, status) + quitC := make(chan bool) + size, err := reader.Size(quitC) + glog.V(logger.Debug).Infof("[BZZ] Swarm: Served '%s' (%d bytes) as '%s' (status code: %v)", uri, size, mimeType, status) http.ServeContent(w, r, path, forever(), reader) diff --git a/swarm/api/manifest.go b/swarm/api/manifest.go index ab74a35bb0..a78d86f485 100644 --- a/swarm/api/manifest.go +++ b/swarm/api/manifest.go @@ -34,24 +34,24 @@ type manifestTrieEntry struct { subtrie *manifestTrie } -func loadManifest(dpa *storage.DPA, hash storage.Key) (trie *manifestTrie, err error) { // non-recursive, subtrees are downloaded on-demand +func loadManifest(dpa *storage.DPA, hash storage.Key, quitC chan bool) (trie *manifestTrie, err error) { // non-recursive, subtrees are downloaded on-demand glog.V(logger.Detail).Infof("[BZZ] manifest lookup key: '%v'.", hash.Log()) // retrieve manifest via DPA manifestReader := dpa.Retrieve(hash) - return readManifest(manifestReader, hash, dpa) + return readManifest(manifestReader, hash, dpa, quitC) } -func readManifest(manifestReader storage.SectionReader, hash storage.Key, dpa *storage.DPA) (trie *manifestTrie, err error) { // non-recursive, subtrees are downloaded on-demand +func readManifest(manifestReader storage.LazySectionReader, hash storage.Key, dpa *storage.DPA, quitC chan bool) (trie *manifestTrie, err error) { // non-recursive, subtrees are downloaded on-demand // TODO check size for oversized manifests - manifestData := make([]byte, manifestReader.Size()) - var size int - size, err = manifestReader.Read(manifestData) - if int64(size) < manifestReader.Size() { + size, err := manifestReader.Size(quitC) + manifestData := make([]byte, size) + read, err := manifestReader.Read(manifestData) + if int64(read) < size { glog.V(logger.Detail).Infof("[BZZ] Manifest %v not found.", hash.Log()) if err == nil { - err = fmt.Errorf("Manifest retrieval cut short: read %v, expect %v", size, manifestReader.Size()) + err = fmt.Errorf("Manifest retrieval cut short: read %v, expect %v", read, size) } return } @@ -71,12 +71,12 @@ func readManifest(manifestReader storage.SectionReader, hash storage.Key, dpa *s dpa: dpa, } for _, entry := range man.Entries { - trie.addEntry(entry) + trie.addEntry(entry, quitC) } return } -func (self *manifestTrie) addEntry(entry *manifestTrieEntry) { +func (self *manifestTrie) addEntry(entry *manifestTrieEntry, quitC chan bool) { self.hash = nil // trie modified, hash needs to be re-calculated on demand if len(entry.Path) == 0 { @@ -97,11 +97,11 @@ func (self *manifestTrie) addEntry(entry *manifestTrieEntry) { } if (oldentry.ContentType == manifestType) && (cpl == len(oldentry.Path)) { - if self.loadSubTrie(oldentry) != nil { + if self.loadSubTrie(oldentry, quitC) != nil { return } entry.Path = entry.Path[cpl:] - oldentry.subtrie.addEntry(entry) + oldentry.subtrie.addEntry(entry, quitC) oldentry.Hash = "" return } @@ -113,8 +113,8 @@ func (self *manifestTrie) addEntry(entry *manifestTrieEntry) { } entry.Path = entry.Path[cpl:] oldentry.Path = oldentry.Path[cpl:] - subtrie.addEntry(entry) - subtrie.addEntry(oldentry) + subtrie.addEntry(entry, quitC) + subtrie.addEntry(oldentry, quitC) self.entries[b] = &manifestTrieEntry{ Path: commonPrefix, @@ -134,7 +134,7 @@ func (self *manifestTrie) getCountLast() (cnt int, entry *manifestTrieEntry) { return } -func (self *manifestTrie) deleteEntry(path string) { +func (self *manifestTrie) deleteEntry(path string, quitC chan bool) { self.hash = nil // trie modified, hash needs to be re-calculated on demand if len(path) == 0 { @@ -154,10 +154,10 @@ func (self *manifestTrie) deleteEntry(path string) { epl := len(entry.Path) if (entry.ContentType == manifestType) && (len(path) >= epl) && (path[:epl] == entry.Path) { - if self.loadSubTrie(entry) != nil { + if self.loadSubTrie(entry, quitC) != nil { return } - entry.subtrie.deleteEntry(path[epl:]) + entry.subtrie.deleteEntry(path[epl:], quitC) entry.Hash = "" // remove subtree if it has less than 2 elements cnt, lastentry := entry.subtrie.getCountLast() @@ -205,16 +205,16 @@ func (self *manifestTrie) recalcAndStore() error { return err2 } -func (self *manifestTrie) loadSubTrie(entry *manifestTrieEntry) (err error) { +func (self *manifestTrie) loadSubTrie(entry *manifestTrieEntry, quitC chan bool) (err error) { if entry.subtrie == nil { hash := common.Hex2Bytes(entry.Hash) - entry.subtrie, err = loadManifest(self.dpa, hash) + entry.subtrie, err = loadManifest(self.dpa, hash, quitC) entry.Hash = "" // might not match, should be recalculated } return } -func (self *manifestTrie) listWithPrefixInt(prefix, rp string, cb func(entry *manifestTrieEntry, suffix string)) (err error) { +func (self *manifestTrie) listWithPrefixInt(prefix, rp string, quitC chan bool, cb func(entry *manifestTrieEntry, suffix string)) error { plen := len(prefix) var start, stop int if plen == 0 { @@ -226,6 +226,11 @@ func (self *manifestTrie) listWithPrefixInt(prefix, rp string, cb func(entry *ma } for i := start; i <= stop; i++ { + select { + case <-quitC: + return fmt.Errorf("aborted") + default: + } entry := self.entries[i] if entry != nil { epl := len(entry.Path) @@ -235,11 +240,13 @@ func (self *manifestTrie) listWithPrefixInt(prefix, rp string, cb func(entry *ma l = epl } if prefix[:l] == entry.Path[:l] { - sterr := self.loadSubTrie(entry) - if sterr == nil { - entry.subtrie.listWithPrefixInt(prefix[l:], rp+entry.Path[l:], cb) - } else { - err = sterr + err := self.loadSubTrie(entry, quitC) + if err != nil { + return err + } + err = entry.subtrie.listWithPrefixInt(prefix[l:], rp+entry.Path[l:], quitC, cb) + if err != nil { + return err } } } else { @@ -249,14 +256,14 @@ func (self *manifestTrie) listWithPrefixInt(prefix, rp string, cb func(entry *ma } } } - return + return nil } -func (self *manifestTrie) listWithPrefix(prefix string, cb func(entry *manifestTrieEntry, suffix string)) (err error) { - return self.listWithPrefixInt(prefix, "", cb) +func (self *manifestTrie) listWithPrefix(prefix string, quitC chan bool, cb func(entry *manifestTrieEntry, suffix string)) (err error) { + return self.listWithPrefixInt(prefix, "", quitC, cb) } -func (self *manifestTrie) findPrefixOf(path string) (entry *manifestTrieEntry, pos int) { +func (self *manifestTrie) findPrefixOf(path string, quitC chan bool) (entry *manifestTrieEntry, pos int) { glog.V(logger.Detail).Infof("[BZZ] findPrefixOf(%s)", path) @@ -274,10 +281,10 @@ func (self *manifestTrie) findPrefixOf(path string) (entry *manifestTrieEntry, p if (len(path) >= epl) && (path[:epl] == entry.Path) { glog.V(logger.Detail).Infof("[BZZ] entry.ContentType = %v", entry.ContentType) if entry.ContentType == manifestType { - if self.loadSubTrie(entry) != nil { + if self.loadSubTrie(entry, quitC) != nil { return nil, 0 } - entry, pos = entry.subtrie.findPrefixOf(path[epl:]) + entry, pos = entry.subtrie.findPrefixOf(path[epl:], quitC) if entry != nil { pos += epl } @@ -307,6 +314,7 @@ func RegularSlashes(path string) (res string) { func (self *manifestTrie) getEntry(spath string) (entry *manifestTrieEntry, fullpath string) { path := RegularSlashes(spath) var pos int - entry, pos = self.findPrefixOf(path) + quitC := make(chan bool) + entry, pos = self.findPrefixOf(path, quitC) return entry, path[:pos] } diff --git a/swarm/api/manifest_test.go b/swarm/api/manifest_test.go index 558bdfb51b..3143e5eaaf 100644 --- a/swarm/api/manifest_test.go +++ b/swarm/api/manifest_test.go @@ -10,18 +10,19 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -func manifest(paths ...string) (manifestReader storage.SectionReader) { +func manifest(paths ...string) (manifestReader storage.LazySectionReader) { var entries []string for _, path := range paths { entry := fmt.Sprintf(`{"path":"%s"}`, path) entries = append(entries, entry) } manifest := fmt.Sprintf(`{"entries":[%s]}`, strings.Join(entries, ",")) - return io.NewSectionReader(strings.NewReader(manifest), 0, int64(len(manifest))) + return &storage.LazyTestSectionReader{io.NewSectionReader(strings.NewReader(manifest), 0, int64(len(manifest)))} } func testGetEntry(t *testing.T, path, match string, paths ...string) *manifestTrie { - trie, err := readManifest(manifest(paths...), nil, nil) + quitC := make(chan bool) + trie, err := readManifest(manifest(paths...), nil, nil, quitC) if err != nil { t.Errorf("unexpected error making manifest: %v", err) } diff --git a/swarm/api/storage.go b/swarm/api/storage.go index 1fd3fa6008..3dfc37f1c2 100644 --- a/swarm/api/storage.go +++ b/swarm/api/storage.go @@ -34,7 +34,11 @@ func (self *Storage) Get(bzzpath string) (*Response, error) { if err != nil { return nil, err } - expsize := reader.Size() + quitC := make(chan bool) + expsize, err := reader.Size(quitC) + if err != nil { + return nil, err + } body := make([]byte, expsize) size, err := reader.Read(body) if int64(size) == expsize { diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index d58eed6381..490f5b432e 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -126,10 +126,10 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s } key := make([]byte, self.hashFunc().Size()) - glog.V(logger.Detail).Infof("[BZZ] split request received for data (%v bytes, depth: %v)", size, depth) + // glog.V(logger.Detail).Infof("[BZZ] split request received for data (%v bytes, depth: %v)", size, depth) // this waitgroup member is released after the root hash is calculated wg.Add(1) - //launch actual recursive function passing the workgroup + //launch actual recursive function passing the waitgroups go self.split(depth, treeSize/self.branches, key, data, size, jobC, chunkC, errC, wg, swg, wwg) // closes internal error channel if all subprocesses in the workgroup finished @@ -266,108 +266,117 @@ func (self *TreeChunker) hashChunk(hasher hash.Hash, job *hashJob, chunkC chan * } } +// LazyChunkReader implements LazySectionReader +type LazyChunkReader struct { + key Key // root key + chunkC chan *Chunk // chunk channel to send retrieve requests on + chunk *Chunk // size of the entire subtree + off int64 // offset + chunkSize int64 // inherit from chunker + branches int64 // inherit from chunker + hashSize int64 // inherit from chunker +} + // implements the Joiner interface -func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) SectionReader { +func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) LazySectionReader { return &LazyChunkReader{ - key: key, - chunkC: chunkC, - quitC: make(chan bool), - errC: make(chan error), - chunker: self, + key: key, + chunkC: chunkC, + chunkSize: self.chunkSize, + branches: self.branches, + hashSize: self.hashSize, } } -// LazyChunkReader implements Lazy.SectionReader -type LazyChunkReader struct { - key Key // root key - chunkC chan *Chunk // chunk channel to send retrieve requests on - size int64 // size of the entire subtree - off int64 // offset - quitC chan bool // channel to abort retrieval - errC chan error // error channel to monitor retrieve errors - chunker *TreeChunker // needs TreeChunker params TODO: should just take - // the chunkSize, branches etc as params +// Size is meant to be called on the LazySectionReader +func (self *LazyChunkReader) Size(quitC chan bool) (n int64, err error) { + if self.chunk != nil { + return self.chunk.Size, nil + } + chunk := retrieve(self.key, self.chunkC, quitC) + if chunk == nil { + select { + case <-quitC: + return 0, errors.New("aborted") + default: + return 0, fmt.Errorf("root chunk not found for %v", self.key.Hex()) + } + } + self.chunk = chunk + return chunk.Size, nil } +// read at can be called numerous times +// concurrent reads are allowed +// Size() needs to be called synchronously on the LazyChunkReader first func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { - self.errC = make(chan error) - chunk := &Chunk{ - Key: self.key, - C: make(chan bool), // close channel to signal data delivery - } - self.chunkC <- chunk // submit retrieval request, someone should be listening on the other side (or we will time out globally) - // glog.V(logger.Detail).Infof("[BZZ] readAt: reading %v into %d bytes at offset %d.", chunk.Key.Log(), len(b), off) - - // waiting for the chunk retrieval - select { - case <-self.quitC: - // this is how we control process leakage (quitC is closed once join is finished (after timeout)) - // glog.V(logger.Detail).Infof("[BZZ] quit") - return - case <-chunk.C: // bells are ringing, data have been delivered - // glog.V(logger.Detail).Infof("[BZZ] chunk data received for %v", chunk.Key.Log()) - } - if len(chunk.SData) == 0 { - // glog.V(logger.Detail).Infof("[BZZ] No payload in %v", chunk.Key.Log()) - return 0, notFound - } - chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - self.size = chunk.Size - if b == nil { + // this is correct, a swarm doc cannot be zero length, so no EOF is expected + if len(b) == 0 { // glog.V(logger.Detail).Infof("[BZZ] Size query for %v", chunk.Key.Log()) - return + return 0, nil } - want := int64(len(b)) - if off+want > self.size { - want = self.size - off + quitC := make(chan bool) + size, err := self.Size(quitC) + if err != nil { + return 0, err } + glog.V(logger.Detail).Infof("readAt: len(b): %v, off: %v, size: %v ", len(b), off, size) + + errC := make(chan error) + // glog.V(logger.Detail).Infof("[BZZ] readAt: reading %v into %d bytes at offset %d.", self.chunk.Key.Log(), len(b), off) + + // } + // glog.V(logger.Detail).Infof("-> want: %v, off: %v size: %v ", want, off, self.size) var treeSize int64 var depth int // calculate depth and max treeSize - treeSize = self.chunker.chunkSize - for ; treeSize < chunk.Size; treeSize *= self.chunker.branches { + treeSize = self.chunkSize + for ; treeSize < size; treeSize *= self.branches { depth++ } wg := sync.WaitGroup{} wg.Add(1) - go self.join(b, off, off+want, depth, treeSize/self.chunker.branches, chunk, &wg) + go self.join(b, off, off+int64(len(b)), depth, treeSize/self.branches, self.chunk, &wg, errC, quitC) go func() { wg.Wait() - close(self.errC) + close(errC) }() - err = <-self.errC + err = <-errC + if err != nil { + close(quitC) + + return 0, err + } // glog.V(logger.Detail).Infof("[BZZ] ReadAt received %v", err) - read = len(b) - if off+int64(read) == self.size { - err = io.EOF + glog.V(logger.Detail).Infof("end: len(b): %v, off: %v, size: %v ", len(b), off, size) + if off+int64(len(b)) >= size { + glog.V(logger.Detail).Infof(" len(b): %v EOF", len(b)) + return len(b), io.EOF } // glog.V(logger.Detail).Infof("[BZZ] ReadAt returning at %d: %v", read, err) - return + return len(b), nil } -func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup) { +func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) { defer parentWg.Done() + // return NewDPA(&LocalStore{}) + glog.V(logger.Detail).Infof("inh len(b): %v, off: %v eoff: %v ", len(b), off, eoff) // glog.V(logger.Detail).Infof("[BZZ] depth: %v, loff: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, off, eoff, chunk.Size, treeSize) - chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) + // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) // find appropriate block level for chunk.Size < treeSize && depth > 0 { - treeSize /= self.chunker.branches + treeSize /= self.branches depth-- } + // leaf chunk found if depth == 0 { - // glog.V(logger.Detail).Infof("[BZZ] depth: %v, len(b): %v, off: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, len(b), off, eoff, chunk.Size, treeSize) - if int64(len(b)) != eoff-off { - //fmt.Printf("len(b) = %v off = %v eoff = %v", len(b), off, eoff) - msg := fmt.Sprintf("len(b) = %v =?= %v = eoff-off", len(b), eoff-off) - panic(msg) - } - + glog.V(logger.Detail).Infof("[BZZ] depth: %v, len(b): %v, off: %v, eoff: %v, chunk.Size: %v, treeSize: %v", depth, len(b), off, eoff, chunk.Size, treeSize) copy(b, chunk.SData[8+off:8+eoff]) return // simply give back the chunks reader for content chunks } @@ -375,10 +384,12 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr // subtree start := off / treeSize end := (eoff + treeSize - 1) / treeSize + wg := &sync.WaitGroup{} + defer wg.Wait() + glog.V(logger.Detail).Infof("[BZZ] start %v,end %v", start, end) for i := start; i < end; i++ { - soff := i * treeSize roff := soff seoff := soff + treeSize @@ -394,51 +405,65 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr } wg.Add(1) go func(j int64) { - childKey := chunk.SData[8+j*self.chunker.hashSize : 8+(j+1)*self.chunker.hashSize] - // glog.V(logger.Detail).Infof("[BZZ] subtree index: %v -> %v", j, childKey.Log()) - - ch := &Chunk{ - Key: childKey, - C: make(chan bool), // close channel to signal data delivery - } - // glog.V(logger.Detail).Infof("[BZZ] chunk data sent for %v (key interval in chunk %v-%v)", ch.Key.Log(), j*self.chunker.hashSize, (j+1)*self.chunker.hashSize) - self.chunkC <- ch // submit retrieval request, someone should be listening on the other side (or we will time out globally) - - // waiting for the chunk retrieval - select { - case <-self.quitC: - // this is how we control process leakage (quitC is closed once join is finished (after timeout)) + childKey := chunk.SData[8+j*self.hashSize : 8+(j+1)*self.hashSize] + // glog.V(logger.Detail).Infof("[BZZ] subtree ind.ex: %v -> %v", j, childKey.Log()) + chunk := retrieve(childKey, self.chunkC, quitC) + if chunk == nil { + select { + case errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize): + case <-quitC: + } return - case <-ch.C: // bells are ringing, data have been delivered - // glog.V(logger.Detail).Infof("[BZZ] chunk data received") } if soff < off { soff = off } - if len(ch.SData) == 0 { - select { - case self.errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize): - case <-self.quitC: - } - return - } - self.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/self.chunker.branches, ch, wg) + self.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/self.branches, chunk, wg, errC, quitC) }(i) } //for - wg.Wait() } -func (self *LazyChunkReader) Size() (n int64) { - self.ReadAt(nil, 0) - return self.size +// the helper method submits chunks for a key to a oueue (DPA) and +// block until they time out or arrive +// abort if quitC is readable +func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk { + chunk := &Chunk{ + Key: key, + C: make(chan bool), // close channel to signal data delivery + } + // glog.V(logger.Detail).Infof("[BZZ] chunk data sent for %v (key interval in chunk %v-%v)", ch.Key.Log(), j*self.chunker.hashSize, (j+1)*self.chunker.hashSize) + // submit chunk for retrieval + select { + case chunkC <- chunk: // submit retrieval request, someone should be listening on the other side (or we will time out globally) + case <-quitC: + return nil + } + // waiting for the chunk retrieval + select { // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) + + case <-quitC: + // this is how we control process leakage (quitC is closed once join is finished (after timeout)) + return nil + case <-chunk.C: // bells are ringing, data have been delivered + // glog.V(logger.Detail).Infof("[BZZ] chunk data received") + } + if len(chunk.SData) == 0 { + return nil // chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) + + } + return chunk } +// Read keeps a cursor so cannot be called simulateously, see ReadAt func (self *LazyChunkReader) Read(b []byte) (read int, err error) { read, err = self.ReadAt(b, self.off) + glog.V(logger.Detail).Infof("[BZZ] read: %v, off: %v, error: %v", read, self.off, err) + self.off += int64(read) return } +// completely analogous to standard SectionReader implementation var errWhence = errors.New("Seek: invalid whence") var errOffset = errors.New("Seek: invalid offset") @@ -451,8 +476,12 @@ func (s *LazyChunkReader) Seek(offset int64, whence int) (int64, error) { case 1: offset += s.off case 2: - offset += s.size + if s.chunk == nil { + return 0, fmt.Errorf("seek from the end requires rootchunk for size. call Size first") + } + offset += s.chunk.Size } + if offset < 0 { return 0, errOffset } diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go index fbcf210588..cdf549ff45 100644 --- a/swarm/storage/chunker_test.go +++ b/swarm/storage/chunker_test.go @@ -83,7 +83,7 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c return } -func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int, chunkC chan *Chunk, quitC chan bool) SectionReader { +func (self *chunkerTester) Join(chunker *TreeChunker, key Key, c int, chunkC chan *Chunk, quitC chan bool) LazySectionReader { // reset but not the chunks reader := chunker.Join(key, chunkC) @@ -163,7 +163,7 @@ func TestRandomData(t *testing.T) { testRandomData(253, 7, t) } -func readAll(reader SectionReader, result []byte) { +func readAll(reader LazySectionReader, result []byte) { size := int64(len(result)) var end int64 @@ -177,8 +177,8 @@ func readAll(reader SectionReader, result []byte) { } } -func benchReadAll(reader SectionReader) { - size := reader.Size() +func benchReadAll(reader LazySectionReader) { + size, _ := reader.Size(nil) output := make([]byte, 1000) for pos := int64(0); pos < size; pos += 1000 { reader.ReadAt(output, pos) diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index b3fdc027b7..55fcbfd409 100644 --- a/swarm/storage/common_test.go +++ b/swarm/storage/common_test.go @@ -44,7 +44,6 @@ func testStore(m ChunkStore, l int64, branches int64, t *testing.T) { swg.Wait() close(chunkC) chunkC = make(chan *Chunk) - r := chunker.Join(key, chunkC) quit := make(chan bool) @@ -53,18 +52,20 @@ func testStore(m ChunkStore, l int64, branches int64, t *testing.T) { go func(chunk *Chunk) { storedChunk, err := m.Get(chunk.Key) if err == notFound { - glog.V(logger.Detail).Infof("[BZZ] chunk '%x' not found", chunk.Key) + glog.V(logger.Detail).Infof("[BZZ] chunk '%v' not found", chunk.Key.Log()) } else if err != nil { - glog.V(logger.Detail).Infof("[BZZ] error retrieving chunk %x: %v", chunk.Key, err) + glog.V(logger.Detail).Infof("[BZZ] error retrieving chunk %v: %v", chunk.Key.Log(), err) } else { chunk.SData = storedChunk.SData + chunk.Size = storedChunk.Size } - glog.V(logger.Detail).Infof("[BZZ] chunk '%x' not found", chunk.Key[:4]) + glog.V(logger.Detail).Infof("[BZZ] chunk '%v' not found", chunk.Key.Log()) close(chunk.C) }(ch) } close(quit) }() + r := chunker.Join(key, chunkC) b := make([]byte, l) n, err := r.ReadAt(b, 0) diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index ba8e0e2c56..34a4639a6d 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -73,7 +73,7 @@ func NewDPA(store ChunkStore, params *ChunkerParams) *DPA { // FS-aware API and httpaccess // Chunk retrieval blocks on netStore requests with a timeout so reader will // report error if retrieval of chunks within requested range time out. -func (self *DPA) Retrieve(key Key) SectionReader { +func (self *DPA) Retrieve(key Key) LazySectionReader { return self.Chunker.Join(key, self.retrieveC) } @@ -146,7 +146,7 @@ func (self *DPA) storeLoop() { go func(chunk *Chunk) { self.Put(chunk) if chunk.wg != nil { - glog.V(logger.Detail).Infof("[BZZ] DPA.storeLoop %v", chunk.Key.Log()) + glog.V(logger.Detail).Infof("[BZZ] dpa: store loop %v", chunk.Key.Log()) chunk.wg.Done() } }(ch) diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 81d4a6f2d0..68487368f1 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -1,5 +1,9 @@ package storage +import ( + "encoding/binary" +) + // LocalStore is a combination of inmemory db over a disk persisted db // implements a Get/Put with fallback (caching) logic using any 2 ChunkStores type LocalStore struct { @@ -48,6 +52,7 @@ func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { if err != nil { return } + chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) self.memStore.Put(chunk) return } diff --git a/swarm/storage/types.go b/swarm/storage/types.go index 0a40fb4614..8d7e7fdd38 100644 --- a/swarm/storage/types.go +++ b/swarm/storage/types.go @@ -14,6 +14,8 @@ import ( type Hasher func() hash.Hash +// Peer is the recorded as Source on the chunk +// should probably not be here? but network should wrap chunk object type Peer interface{} type Key []byte @@ -187,7 +189,7 @@ type Joiner interface { 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. */ - Join(key Key, chunkC chan *Chunk) SectionReader + Join(key Key, chunkC chan *Chunk) LazySectionReader } type Chunker interface { @@ -198,9 +200,17 @@ type Chunker interface { } // Size, Seek, Read, ReadAt -type SectionReader interface { - Size() int64 +type LazySectionReader interface { + Size(chan bool) (int64, error) io.Seeker io.Reader io.ReaderAt } + +type LazyTestSectionReader struct { + *io.SectionReader +} + +func (self *LazyTestSectionReader) Size(chan bool) (int64, error) { + return self.SectionReader.Size(), nil +}