forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-25 06:06:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/storage, swarm/api: fix chunker joiner and fs api

Browse source 
* SectionReader -> LazySectionReader: Size method signature change
* introduce abort channel to joiner to fix process leak
* add abort channel   context to all manifest retrieval methods
* global waitgroup fixes intermittent upload test failures due to unfinished storage of chunks
* add back final slash to paths in manifest matching
* simplify downloader code and fix process leak
* fix filesystem api tests
* streamline joiner logic, contexts now unique to each readAt call
* LazyChunkReader seeker complains about missing size only if whence=2
This commit is contained in:
zelig 2016-06-28 22:35:32 +02:00
parent 58397953bc
commit 0de16bd17e
14 changed files with 318 additions and 240 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
swarm/api/api.go
View file

@ -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()

21
swarm/api/api_test.go
View file

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

86
swarm/api/filesystem.go
View file

@ -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,46 +189,47 @@ 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
if err != nil {
return err
}
cnt := len(list)
errors := make([]error, cnt)
wg := sync.WaitGroup{}
errC := make(chan error)
done := make(chan bool, maxParallelFiles)
dcnt := 0
for i, entry := range list {
if i >= dcnt+maxParallelFiles {
<-done
dcnt++
select {
case done <- true:
wg.Add(1)
case <-quitC:
return fmt.Errorf("aborted")
}
go func(i int, entry *downloadListEntry, done chan bool) {
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)
_, err = io.CopyN(writer, reader, reader.Size()) // TODO: handle errors
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
@ -234,23 +239,26 @@ func (self *FileSystem) Download(bzzpath, localpath string) error {
err = err2
}
}
errors[i] = err
done <- true
}(i, entry, done)
}
for dcnt < cnt {
<-done
dcnt++
}
if err != nil {
select {
case errC <- err:
case <-quitC:
}
return
}
<-done
}(i, entry)
}
go func() {
wg.Wait()
close(errC)
}()
select {
case err = <-errC:
return err
case <-quitC:
return fmt.Errorf("aborted")
}
for i, _ := range list {
if errors[i] != nil {
return errors[i]
}
}
return err
}

42
swarm/api/filesystem_test.go
View file

@ -1,15 +1,13 @@
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 (
@ -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 := expResponse(content, "text/html; charset=utf-8", 0)
t.Logf("index.html (size%v=?=%v)", resp.Size, exp.Size)
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
}

10
swarm/api/http/server.go
View file

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

72
swarm/api/manifest.go
View file

@ -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]
}

7
swarm/api/manifest_test.go
View file

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

6
swarm/api/storage.go
View file

@ -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 {

211
swarm/storage/chunker.go
View file

@ -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,
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
}
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
chunk := retrieve(self.key, self.chunkC, quitC)
if chunk == nil {
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())
case <-quitC:
return 0, errors.New("aborted")
default:
return 0, fmt.Errorf("root chunk not found for %v", self.key.Hex())
}
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 {
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) {
// 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
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 <-self.quitC:
// this is how we control process leakage (quitC is closed once join is finished (after timeout))
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
}

8
swarm/storage/chunker_test.go
View file

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

9
swarm/storage/common_test.go
View file

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

4
swarm/storage/dpa.go
View file

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

5
swarm/storage/localstore.go
View file

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

16
swarm/storage/types.go
View file

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