forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-18 18:02:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/*: Golint fixes for swarm package

Browse source
This commit is contained in:
Eli 2018-05-01 16:41:20 -07:00
parent 1da33028ce
commit f63cea8d71
22 changed files with 342 additions and 345 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

79
swarm/api/api.go
View file

@ -46,7 +46,7 @@ var (
apiPutFail = metrics.NewRegisteredCounter("api.put.fail", nil) apiPutFail = metrics.NewRegisteredCounter("api.put.fail", nil)
apiGetCount = metrics.NewRegisteredCounter("api.get.count", nil) apiGetCount = metrics.NewRegisteredCounter("api.get.count", nil)
apiGetNotFound = metrics.NewRegisteredCounter("api.get.notfound", nil) apiGetNotFound = metrics.NewRegisteredCounter("api.get.notfound", nil)
apiGetHttp300 = metrics.NewRegisteredCounter("api.get.http.300", nil) apiGetHTTP300 = metrics.NewRegisteredCounter("api.get.http.300", nil)
apiModifyCount = metrics.NewRegisteredCounter("api.modify.count", nil) apiModifyCount = metrics.NewRegisteredCounter("api.modify.count", nil)
apiModifyFail = metrics.NewRegisteredCounter("api.modify.fail", nil) apiModifyFail = metrics.NewRegisteredCounter("api.modify.fail", nil)
apiAddFileCount = metrics.NewRegisteredCounter("api.addfile.count", nil) apiAddFileCount = metrics.NewRegisteredCounter("api.addfile.count", nil)
@ -146,7 +146,7 @@ type Api struct {
dns Resolver dns Resolver
} }
//the api constructor initialises //NewApi constructor initialises
func NewApi(dpa *storage.DPA, dns Resolver) (self *Api) { func NewApi(dpa *storage.DPA, dns Resolver) (self *Api) {
self = &Api{ self = &Api{
dpa: dpa, dpa: dpa,
@ -155,26 +155,26 @@ func NewApi(dpa *storage.DPA, dns Resolver) (self *Api) {
return return
} }
// to be used only in TEST // Upload to be used only in TEST
func (self *Api) Upload(uploadDir, index string) (hash string, err error) { func (api *Api) Upload(uploadDir, index string) (hash string, err error) {
fs := NewFileSystem(self) fs := NewFileSystem(api)
hash, err = fs.Upload(uploadDir, index) hash, err = fs.Upload(uploadDir, index)
return hash, err return hash, err
} }
// DPA reader API // Retrieve implements DPA reader API
func (self *Api) Retrieve(key storage.Key) storage.LazySectionReader { func (api *Api) Retrieve(key storage.Key) storage.LazySectionReader {
return self.dpa.Retrieve(key) return api.dpa.Retrieve(key)
} }
func (self *Api) Store(data io.Reader, size int64, wg *sync.WaitGroup) (key storage.Key, err error) { func (api *Api) Store(data io.Reader, size int64, wg *sync.WaitGroup) (key storage.Key, err error) {
return self.dpa.Store(data, size, wg, nil) return api.dpa.Store(data, size, wg, nil)
} }
type ErrResolve error type ErrResolve error
// DNS Resolver // Resolve implements DNS Resolver
func (self *Api) Resolve(uri *URI) (storage.Key, error) { func (api *Api) Resolve(uri *URI) (storage.Key, error) {
apiResolveCount.Inc(1) apiResolveCount.Inc(1)
log.Trace(fmt.Sprintf("Resolving : %v", uri.Addr)) log.Trace(fmt.Sprintf("Resolving : %v", uri.Addr))
@ -188,7 +188,7 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
} }
// if DNS is not configured, check if the address is a hash // if DNS is not configured, check if the address is a hash
if self.dns == nil { if api.dns == nil {
if !isHash { if !isHash {
apiResolveFail.Inc(1) apiResolveFail.Inc(1)
return nil, fmt.Errorf("no DNS to resolve name: %q", uri.Addr) return nil, fmt.Errorf("no DNS to resolve name: %q", uri.Addr)
@ -197,7 +197,7 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
} }
// try and resolve the address // try and resolve the address
resolved, err := self.dns.Resolve(uri.Addr) resolved, err := api.dns.Resolve(uri.Addr)
if err == nil { if err == nil {
return resolved[:], nil return resolved[:], nil
} else if !isHash { } else if !isHash {
@ -208,18 +208,18 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
} }
// Put provides singleton manifest creation on top of dpa store // Put provides singleton manifest creation on top of dpa store
func (self *Api) Put(content, contentType string) (storage.Key, error) { func (api *Api) Put(content, contentType string) (storage.Key, error) {
apiPutCount.Inc(1) apiPutCount.Inc(1)
r := strings.NewReader(content) r := strings.NewReader(content)
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
key, err := self.dpa.Store(r, int64(len(content)), wg, nil) key, err := api.dpa.Store(r, int64(len(content)), wg, nil)
if err != nil { if err != nil {
apiPutFail.Inc(1) apiPutFail.Inc(1)
return nil, err return nil, err
} }
manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType) manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType)
r = strings.NewReader(manifest) r = strings.NewReader(manifest)
key, err = self.dpa.Store(r, int64(len(manifest)), wg, nil) key, err = api.dpa.Store(r, int64(len(manifest)), wg, nil)
if err != nil { if err != nil {
apiPutFail.Inc(1) apiPutFail.Inc(1)
return nil, err return nil, err
@ -231,9 +231,9 @@ func (self *Api) Put(content, contentType string) (storage.Key, error) {
// Get uses iterative manifest retrieval and prefix matching // Get uses iterative manifest retrieval and prefix matching
// to resolve basePath to content using dpa retrieve // to resolve basePath to content using dpa retrieve
// it returns a section reader, mimeType, status and an error // 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 (api *Api) Get(key storage.Key, path string) (reader storage.LazySectionReader, mimeType string, status int, err error) {
apiGetCount.Inc(1) apiGetCount.Inc(1)
trie, err := loadManifest(self.dpa, key, nil) trie, err := loadManifest(api.dpa, key, nil)
if err != nil { if err != nil {
apiGetNotFound.Inc(1) apiGetNotFound.Inc(1)
status = http.StatusNotFound status = http.StatusNotFound
@ -249,13 +249,12 @@ func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionRe
key = common.Hex2Bytes(entry.Hash) key = common.Hex2Bytes(entry.Hash)
status = entry.Status status = entry.Status
if status == http.StatusMultipleChoices { if status == http.StatusMultipleChoices {
apiGetHttp300.Inc(1) apiGetHTTP300.Inc(1)
return return
} else {
mimeType = entry.ContentType
log.Trace(fmt.Sprintf("content lookup key: '%v' (%v)", key, mimeType))
reader = self.dpa.Retrieve(key)
} }
mimeType = entry.ContentType
log.Trace(fmt.Sprintf("content lookup key: '%v' (%v)", key, mimeType))
reader = api.dpa.Retrieve(key)
} else { } else {
status = http.StatusNotFound status = http.StatusNotFound
apiGetNotFound.Inc(1) apiGetNotFound.Inc(1)
@ -265,10 +264,10 @@ func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionRe
return return
} }
func (self *Api) Modify(key storage.Key, path, contentHash, contentType string) (storage.Key, error) { func (api *Api) Modify(key storage.Key, path, contentHash, contentType string) (storage.Key, error) {
apiModifyCount.Inc(1) apiModifyCount.Inc(1)
quitC := make(chan bool) quitC := make(chan bool)
trie, err := loadManifest(self.dpa, key, quitC) trie, err := loadManifest(api.dpa, key, quitC)
if err != nil { if err != nil {
apiModifyFail.Inc(1) apiModifyFail.Inc(1)
return nil, err return nil, err
@ -291,7 +290,7 @@ func (self *Api) Modify(key storage.Key, path, contentHash, contentType string)
return trie.hash, nil return trie.hash, nil
} }
func (self *Api) AddFile(mhash, path, fname string, content []byte, nameresolver bool) (storage.Key, string, error) { func (api *Api) AddFile(mhash, path, fname string, content []byte, nameresolver bool) (storage.Key, string, error) {
apiAddFileCount.Inc(1) apiAddFileCount.Inc(1)
uri, err := Parse("bzz:/" + mhash) uri, err := Parse("bzz:/" + mhash)
@ -299,7 +298,7 @@ func (self *Api) AddFile(mhash, path, fname string, content []byte, nameresolver
apiAddFileFail.Inc(1) apiAddFileFail.Inc(1)
return nil, "", err return nil, "", err
} }
mkey, err := self.Resolve(uri) mkey, err := api.Resolve(uri)
if err != nil { if err != nil {
apiAddFileFail.Inc(1) apiAddFileFail.Inc(1)
return nil, "", err return nil, "", err
@ -318,7 +317,7 @@ func (self *Api) AddFile(mhash, path, fname string, content []byte, nameresolver
ModTime: time.Now(), ModTime: time.Now(),
} }
mw, err := self.NewManifestWriter(mkey, nil) mw, err := api.NewManifestWriter(mkey, nil)
if err != nil { if err != nil {
apiAddFileFail.Inc(1) apiAddFileFail.Inc(1)
return nil, "", err return nil, "", err
@ -341,7 +340,7 @@ func (self *Api) AddFile(mhash, path, fname string, content []byte, nameresolver
} }
func (self *Api) RemoveFile(mhash, path, fname string, nameresolver bool) (string, error) { func (api *Api) RemoveFile(mhash, path, fname string, nameresolver bool) (string, error) {
apiRmFileCount.Inc(1) apiRmFileCount.Inc(1)
uri, err := Parse("bzz:/" + mhash) uri, err := Parse("bzz:/" + mhash)
@ -349,7 +348,7 @@ func (self *Api) RemoveFile(mhash, path, fname string, nameresolver bool) (strin
apiRmFileFail.Inc(1) apiRmFileFail.Inc(1)
return "", err return "", err
} }
mkey, err := self.Resolve(uri) mkey, err := api.Resolve(uri)
if err != nil { if err != nil {
apiRmFileFail.Inc(1) apiRmFileFail.Inc(1)
return "", err return "", err
@ -360,7 +359,7 @@ func (self *Api) RemoveFile(mhash, path, fname string, nameresolver bool) (strin
path = path[1:] path = path[1:]
} }
mw, err := self.NewManifestWriter(mkey, nil) mw, err := api.NewManifestWriter(mkey, nil)
if err != nil { if err != nil {
apiRmFileFail.Inc(1) apiRmFileFail.Inc(1)
return "", err return "", err
@ -382,7 +381,7 @@ func (self *Api) RemoveFile(mhash, path, fname string, nameresolver bool) (strin
return newMkey.String(), nil return newMkey.String(), nil
} }
func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, content []byte, oldKey storage.Key, offset int64, addSize int64, nameresolver bool) (storage.Key, string, error) { func (api *Api) AppendFile(mhash, path, fname string, existingSize int64, content []byte, oldKey storage.Key, offset int64, addSize int64, nameresolver bool) (storage.Key, string, error) {
apiAppendFileCount.Inc(1) apiAppendFileCount.Inc(1)
buffSize := offset + addSize buffSize := offset + addSize
@ -392,7 +391,7 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
buf := make([]byte, buffSize) buf := make([]byte, buffSize)
oldReader := self.Retrieve(oldKey) oldReader := api.Retrieve(oldKey)
io.ReadAtLeast(oldReader, buf, int(offset)) io.ReadAtLeast(oldReader, buf, int(offset))
newReader := bytes.NewReader(content) newReader := bytes.NewReader(content)
@ -406,7 +405,7 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
totalSize := int64(len(buf)) totalSize := int64(len(buf))
// TODO(jmozah): to append using pyramid chunker when it is ready // TODO(jmozah): to append using pyramid chunker when it is ready
//oldReader := self.Retrieve(oldKey) //oldReader := api.Retrieve(oldKey)
//newReader := bytes.NewReader(content) //newReader := bytes.NewReader(content)
//combinedReader := io.MultiReader(oldReader, newReader) //combinedReader := io.MultiReader(oldReader, newReader)
@ -415,7 +414,7 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
apiAppendFileFail.Inc(1) apiAppendFileFail.Inc(1)
return nil, "", err return nil, "", err
} }
mkey, err := self.Resolve(uri) mkey, err := api.Resolve(uri)
if err != nil { if err != nil {
apiAppendFileFail.Inc(1) apiAppendFileFail.Inc(1)
return nil, "", err return nil, "", err
@ -426,7 +425,7 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
path = path[1:] path = path[1:]
} }
mw, err := self.NewManifestWriter(mkey, nil) mw, err := api.NewManifestWriter(mkey, nil)
if err != nil { if err != nil {
apiAppendFileFail.Inc(1) apiAppendFileFail.Inc(1)
return nil, "", err return nil, "", err
@ -463,19 +462,19 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
} }
func (self *Api) BuildDirectoryTree(mhash string, nameresolver bool) (key storage.Key, manifestEntryMap map[string]*manifestTrieEntry, err error) { func (api *Api) BuildDirectoryTree(mhash string, nameresolver bool) (key storage.Key, manifestEntryMap map[string]*manifestTrieEntry, err error) {
uri, err := Parse("bzz:/" + mhash) uri, err := Parse("bzz:/" + mhash)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
key, err = self.Resolve(uri) key, err = api.Resolve(uri)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
quitC := make(chan bool) quitC := make(chan bool)
rootTrie, err := loadManifest(self.dpa, key, quitC) rootTrie, err := loadManifest(api.dpa, key, quitC)
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("can't load manifest %v: %v", key.String(), err) return nil, nil, fmt.Errorf("can't load manifest %v: %v", key.String(), err)
} }

4
swarm/api/api_test.go
View file

@ -29,7 +29,7 @@ import (
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
) )
func testApi(t *testing.T, f func(*Api)) { func testAPI(t *testing.T, f func(*Api)) {
datadir, err := ioutil.TempDir("", "bzz-test") datadir, err := ioutil.TempDir("", "bzz-test")
if err != nil { if err != nil {
t.Fatalf("unable to create temp dir: %v", err) t.Fatalf("unable to create temp dir: %v", err)
@ -106,7 +106,7 @@ func testGet(t *testing.T, api *Api, bzzhash, path string) *testResponse {
} }
func TestApiPut(t *testing.T) { func TestApiPut(t *testing.T) {
testApi(t, func(api *Api) { testAPI(t, func(api *Api) {
content := "hello" content := "hello"
exp := expResponse(content, "text/plain", 0) exp := expResponse(content, "text/plain", 0)
// exp := expResponse([]byte(content), "text/plain", 0) // exp := expResponse([]byte(content), "text/plain", 0)

26
swarm/api/config.go
View file

@ -37,7 +37,7 @@ const (
DefaultHTTPPort = "8500" DefaultHTTPPort = "8500"
) )
// separate bzz directories // Config separates bzz directories
// allow several bzz nodes running in parallel // allow several bzz nodes running in parallel
type Config struct { type Config struct {
// serialised/persisted fields // serialised/persisted fields
@ -63,10 +63,10 @@ type Config struct {
BootNodes string BootNodes string
} }
//create a default config with all parameters to set to defaults // NewDefaultConfig creates a default config with all parameters to set to defaults
func NewDefaultConfig() (self *Config) { func NewDefaultConfig() (config *Config) {
self = &Config{ config = &Config{
StoreParams: storage.NewDefaultStoreParams(), StoreParams: storage.NewDefaultStoreParams(),
ChunkerParams: storage.NewChunkerParams(), ChunkerParams: storage.NewChunkerParams(),
HiveParams: network.NewDefaultHiveParams(), HiveParams: network.NewDefaultHiveParams(),
@ -89,11 +89,11 @@ func NewDefaultConfig() (self *Config) {
//some config params need to be initialized after the complete //some config params need to be initialized after the complete
//config building phase is completed (e.g. due to overriding flags) //config building phase is completed (e.g. due to overriding flags)
func (self *Config) Init(prvKey *ecdsa.PrivateKey) { func (config *Config) Init(prvKey *ecdsa.PrivateKey) {
address := crypto.PubkeyToAddress(prvKey.PublicKey) address := crypto.PubkeyToAddress(prvKey.PublicKey)
self.Path = filepath.Join(self.Path, "bzz-"+common.Bytes2Hex(address.Bytes())) config.Path = filepath.Join(config.Path, "bzz-"+common.Bytes2Hex(address.Bytes()))
err := os.MkdirAll(self.Path, os.ModePerm) err := os.MkdirAll(config.Path, os.ModePerm)
if err != nil { if err != nil {
log.Error(fmt.Sprintf("Error creating root swarm data directory: %v", err)) log.Error(fmt.Sprintf("Error creating root swarm data directory: %v", err))
return return
@ -103,11 +103,11 @@ func (self *Config) Init(prvKey *ecdsa.PrivateKey) {
pubkeyhex := common.ToHex(pubkey) pubkeyhex := common.ToHex(pubkey)
keyhex := crypto.Keccak256Hash(pubkey).Hex() keyhex := crypto.Keccak256Hash(pubkey).Hex()
self.PublicKey = pubkeyhex config.PublicKey = pubkeyhex
self.BzzKey = keyhex config.BzzKey = keyhex
self.Swap.Init(self.Contract, prvKey) config.Swap.Init(config.Contract, prvKey)
self.SyncParams.Init(self.Path) config.SyncParams.Init(config.Path)
self.HiveParams.Init(self.Path) config.HiveParams.Init(config.Path)
self.StoreParams.Init(self.Path) config.StoreParams.Init(config.Path)
} }

14
swarm/api/filesystem.go
View file

@ -46,7 +46,7 @@ func NewFileSystem(api *Api) *FileSystem {
// TODO: localpath should point to a manifest // TODO: localpath should point to a manifest
// //
// DEPRECATED: Use the HTTP API instead // DEPRECATED: Use the HTTP API instead
func (self *FileSystem) Upload(lpath, index string) (string, error) { func (fs *FileSystem) Upload(lpath, index string) (string, error) {
var list []*manifestTrieEntry var list []*manifestTrieEntry
localpath, err := filepath.Abs(filepath.Clean(lpath)) localpath, err := filepath.Abs(filepath.Clean(lpath))
if err != nil { if err != nil {
@ -113,7 +113,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
stat, _ := f.Stat() stat, _ := f.Stat()
var hash storage.Key var hash storage.Key
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
hash, err = self.api.dpa.Store(f, stat.Size(), wg, nil) hash, err = fs.api.dpa.Store(f, stat.Size(), wg, nil)
if hash != nil { if hash != nil {
list[i].Hash = hash.String() list[i].Hash = hash.String()
} }
@ -142,7 +142,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
} }
trie := &manifestTrie{ trie := &manifestTrie{
dpa: self.api.dpa, dpa: fs.api.dpa,
} }
quitC := make(chan bool) quitC := make(chan bool)
for i, entry := range list { for i, entry := range list {
@ -173,7 +173,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
// under localpath // under localpath
// //
// DEPRECATED: Use the HTTP API instead // DEPRECATED: Use the HTTP API instead
func (self *FileSystem) Download(bzzpath, localpath string) error { func (fs *FileSystem) Download(bzzpath, localpath string) error {
lpath, err := filepath.Abs(filepath.Clean(localpath)) lpath, err := filepath.Abs(filepath.Clean(localpath))
if err != nil { if err != nil {
return err return err
@ -188,7 +188,7 @@ func (self *FileSystem) Download(bzzpath, localpath string) error {
if err != nil { if err != nil {
return err return err
} }
key, err := self.api.Resolve(uri) key, err := fs.api.Resolve(uri)
if err != nil { if err != nil {
return err return err
} }
@ -199,7 +199,7 @@ func (self *FileSystem) Download(bzzpath, localpath string) error {
} }
quitC := make(chan bool) quitC := make(chan bool)
trie, err := loadManifest(self.api.dpa, key, quitC) trie, err := loadManifest(fs.api.dpa, key, quitC)
if err != nil { if err != nil {
log.Warn(fmt.Sprintf("fs.Download: loadManifestTrie error: %v", err)) log.Warn(fmt.Sprintf("fs.Download: loadManifestTrie error: %v", err))
return err return err
@ -244,7 +244,7 @@ func (self *FileSystem) Download(bzzpath, localpath string) error {
} }
go func(i int, entry *downloadListEntry) { go func(i int, entry *downloadListEntry) {
defer wg.Done() defer wg.Done()
err := retrieveToFile(quitC, self.api.dpa, entry.key, entry.path) err := retrieveToFile(quitC, fs.api.dpa, entry.key, entry.path)
if err != nil { if err != nil {
select { select {
case errC <- err: case errC <- err:

21
swarm/api/http/error.go
View file

@ -15,7 +15,7 @@
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
/* /*
Show nicely (but simple) formatted HTML error pages (or respond with JSON Package http shows nicely (but simple) formatted HTML error pages (or respond with JSON
if the appropriate `Accept` header is set)) for the http package. if the appropriate `Accept` header is set)) for the http package.
*/ */
package http package http
@ -43,7 +43,7 @@ var (
jsonCounter = metrics.NewRegisteredCounter("api.http.errorpage.json.count", nil) jsonCounter = metrics.NewRegisteredCounter("api.http.errorpage.json.count", nil)
) )
//parameters needed for formatting the correct HTML page //ErrorParams needed for formatting the correct HTML page
type ErrorParams struct { type ErrorParams struct {
Msg string Msg string
Code int Code int
@ -52,8 +52,8 @@ type ErrorParams struct {
Details template.HTML Details template.HTML
} }
//a custom error case struct that would be used to store validators and //CaseError is a custom error case struct that would be used to store validators
//additional error info to display with client responses. //and additional error info to display with client responses.
type CaseError struct { type CaseError struct {
Validator func(*Request) bool Validator func(*Request) bool
Msg func(*Request) string Msg func(*Request) string
@ -107,7 +107,7 @@ func ValidateCaseErrors(r *Request) string {
return "" return ""
} }
//ShowMultipeChoices is used when a user requests a resource in a manifest which results //ShowMultipleChoices is used when a user requests a resource in a manifest which results
//in ambiguous results. It returns a HTML page with clickable links of each of the entry //in ambiguous results. It returns a HTML page with clickable links of each of the entry
//in the manifest which fits the request URI ambiguity. //in the manifest which fits the request URI ambiguity.
//For example, if the user requests bzz:/<hash>/read and that manifest contains entries //For example, if the user requests bzz:/<hash>/read and that manifest contains entries
@ -164,14 +164,14 @@ func ShowError(w http.ResponseWriter, r *Request, msg string, code int) {
func respond(w http.ResponseWriter, r *http.Request, params *ErrorParams) { func respond(w http.ResponseWriter, r *http.Request, params *ErrorParams) {
w.WriteHeader(params.Code) w.WriteHeader(params.Code)
if r.Header.Get("Accept") == "application/json" { if r.Header.Get("Accept") == "application/json" {
respondJson(w, params) respondJSON(w, params)
} else { } else {
respondHtml(w, params) respondHTML(w, params)
} }
} }
//return a HTML page //return a HTML page
func respondHtml(w http.ResponseWriter, params *ErrorParams) { func respondHTML(w http.ResponseWriter, params *ErrorParams) {
htmlCounter.Inc(1) htmlCounter.Inc(1)
err := params.template.Execute(w, params) err := params.template.Execute(w, params)
if err != nil { if err != nil {
@ -180,7 +180,7 @@ func respondHtml(w http.ResponseWriter, params *ErrorParams) {
} }
//return JSON //return JSON
func respondJson(w http.ResponseWriter, params *ErrorParams) { func respondJSON(w http.ResponseWriter, params *ErrorParams) {
jsonCounter.Inc(1) jsonCounter.Inc(1)
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(params) json.NewEncoder(w).Encode(params)
@ -190,7 +190,6 @@ func respondJson(w http.ResponseWriter, params *ErrorParams) {
func getTemplate(code int) *template.Template { func getTemplate(code int) *template.Template {
if val, tmpl := templateMap[code]; tmpl { if val, tmpl := templateMap[code]; tmpl {
return val return val
} else {
return templateMap[0]
} }
return templateMap[0]
} }

7
swarm/api/http/error_templates.go
View file

@ -24,9 +24,10 @@ they won't be found.
For this reason we resort to save the HTML error pages as strings, which then can be For this reason we resort to save the HTML error pages as strings, which then can be
parsed by Go's html/template package parsed by Go's html/template package
*/ */
package http package http
//This returns the HTML for generic errors // GetGenericErrorPage returns the HTML for generic errors
func GetGenericErrorPage() string { func GetGenericErrorPage() string {
page := ` page := `
<html> <html>
@ -206,7 +207,7 @@ func GetGenericErrorPage() string {
return page return page
} }
//This returns the HTML for a 404 Not Found error // GetNotFoundErrorPage returns the HTML for a 404 Not Found error
func GetNotFoundErrorPage() string { func GetNotFoundErrorPage() string {
page := ` page := `
<html> <html>
@ -386,7 +387,7 @@ func GetNotFoundErrorPage() string {
return page return page
} }
//This returns the HTML for a page listing disambiguation options //GetMultipleChoicesErrorPage returns the HTML for a page listing disambiguation options
//i.e. if user requested bzz:/<hash>/read and the manifest contains "readme.md" and "readinglist.txt", //i.e. if user requested bzz:/<hash>/read and the manifest contains "readme.md" and "readinglist.txt",
//this page is returned with a clickable list the existing disambiguation links in the manifest //this page is returned with a clickable list the existing disambiguation links in the manifest
func GetMultipleChoicesErrorPage() string { func GetMultipleChoicesErrorPage() string {

6
swarm/api/http/roundtripper.go
View file

@ -51,12 +51,12 @@ type RoundTripper struct {
Port string Port string
} }
func (self *RoundTripper) RoundTrip(req *http.Request) (resp *http.Response, err error) { func (rt *RoundTripper) RoundTrip(req *http.Request) (resp *http.Response, err error) {
host := self.Host host := rt.Host
if len(host) == 0 { if len(host) == 0 {
host = "localhost" host = "localhost"
} }
url := fmt.Sprintf("http://%s:%s/%s:/%s/%s", host, self.Port, req.Proto, req.URL.Host, req.URL.Path) url := fmt.Sprintf("http://%s:%s/%s:/%s/%s", host, rt.Port, req.Proto, req.URL.Host, req.URL.Path)
log.Info(fmt.Sprintf("roundtripper: proxying request '%s' to '%s'", req.RequestURI, url)) log.Info(fmt.Sprintf("roundtripper: proxying request '%s' to '%s'", req.RequestURI, url))
reqProxy, err := http.NewRequest(req.Method, url, req.Body) reqProxy, err := http.NewRequest(req.Method, url, req.Body)
if err != nil { if err != nil {

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

@ -15,7 +15,7 @@
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
/* /*
A simple http server interface to Swarm Package http implements a simple http server interface to Swarm
*/ */
package http package http
@ -78,7 +78,7 @@ type ServerConfig struct {
// electron (chromium) api for registering bzz url scheme handlers: // electron (chromium) api for registering bzz url scheme handlers:
// https://github.com/atom/electron/blob/master/docs/api/protocol.md // https://github.com/atom/electron/blob/master/docs/api/protocol.md
// starts up http server // StartHttpServer starts up http server
func StartHttpServer(api *api.Api, config *ServerConfig) { func StartHttpServer(api *api.Api, config *ServerConfig) {
var allowedOrigins []string var allowedOrigins []string
for _, domain := range strings.Split(config.CorsString, ",") { for _, domain := range strings.Split(config.CorsString, ",") {
@ -695,9 +695,8 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if uri.Raw() || uri.DeprecatedRaw() { if uri.Raw() || uri.DeprecatedRaw() {
ShowError(w, req, fmt.Sprintf("No PUT to %s allowed.", uri), http.StatusBadRequest) ShowError(w, req, fmt.Sprintf("No PUT to %s allowed.", uri), http.StatusBadRequest)
return return
} else {
s.HandlePostFiles(w, req)
} }
s.HandlePostFiles(w, req)
case "DELETE": case "DELETE":
if uri.Raw() || uri.DeprecatedRaw() { if uri.Raw() || uri.DeprecatedRaw() {

74
swarm/api/manifest.go
View file

@ -230,18 +230,18 @@ func readManifest(manifestReader storage.LazySectionReader, hash storage.Key, dp
return return
} }
func (self *manifestTrie) addEntry(entry *manifestTrieEntry, quitC chan bool) { func (mtrie *manifestTrie) addEntry(entry *manifestTrieEntry, quitC chan bool) {
self.hash = nil // trie modified, hash needs to be re-calculated on demand mtrie.hash = nil // trie modified, hash needs to be re-calculated on demand
if len(entry.Path) == 0 { if len(entry.Path) == 0 {
self.entries[256] = entry mtrie.entries[256] = entry
return return
} }
b := entry.Path[0] b := entry.Path[0]
oldentry := self.entries[b] oldentry := mtrie.entries[b]
if (oldentry == nil) || (oldentry.Path == entry.Path && oldentry.ContentType != ManifestType) { if (oldentry == nil) || (oldentry.Path == entry.Path && oldentry.ContentType != ManifestType) {
self.entries[b] = entry mtrie.entries[b] = entry
return return
} }
@ -251,7 +251,7 @@ func (self *manifestTrie) addEntry(entry *manifestTrieEntry, quitC chan bool) {
} }
if (oldentry.ContentType == ManifestType) && (cpl == len(oldentry.Path)) { if (oldentry.ContentType == ManifestType) && (cpl == len(oldentry.Path)) {
if self.loadSubTrie(oldentry, quitC) != nil { if mtrie.loadSubTrie(oldentry, quitC) != nil {
return return
} }
entry.Path = entry.Path[cpl:] entry.Path = entry.Path[cpl:]
@ -263,21 +263,21 @@ func (self *manifestTrie) addEntry(entry *manifestTrieEntry, quitC chan bool) {
commonPrefix := entry.Path[:cpl] commonPrefix := entry.Path[:cpl]
subtrie := &manifestTrie{ subtrie := &manifestTrie{
dpa: self.dpa, dpa: mtrie.dpa,
} }
entry.Path = entry.Path[cpl:] entry.Path = entry.Path[cpl:]
oldentry.Path = oldentry.Path[cpl:] oldentry.Path = oldentry.Path[cpl:]
subtrie.addEntry(entry, quitC) subtrie.addEntry(entry, quitC)
subtrie.addEntry(oldentry, quitC) subtrie.addEntry(oldentry, quitC)
self.entries[b] = newManifestTrieEntry(&ManifestEntry{ mtrie.entries[b] = newManifestTrieEntry(&ManifestEntry{
Path: commonPrefix, Path: commonPrefix,
ContentType: ManifestType, ContentType: ManifestType,
}, subtrie) }, subtrie)
} }
func (self *manifestTrie) getCountLast() (cnt int, entry *manifestTrieEntry) { func (mtrie *manifestTrie) getCountLast() (cnt int, entry *manifestTrieEntry) {
for _, e := range self.entries { for _, e := range mtrie.entries {
if e != nil { if e != nil {
cnt++ cnt++
entry = e entry = e
@ -286,27 +286,27 @@ func (self *manifestTrie) getCountLast() (cnt int, entry *manifestTrieEntry) {
return return
} }
func (self *manifestTrie) deleteEntry(path string, quitC chan bool) { func (mtrie *manifestTrie) deleteEntry(path string, quitC chan bool) {
self.hash = nil // trie modified, hash needs to be re-calculated on demand mtrie.hash = nil // trie modified, hash needs to be re-calculated on demand
if len(path) == 0 { if len(path) == 0 {
self.entries[256] = nil mtrie.entries[256] = nil
return return
} }
b := path[0] b := path[0]
entry := self.entries[b] entry := mtrie.entries[b]
if entry == nil { if entry == nil {
return return
} }
if entry.Path == path { if entry.Path == path {
self.entries[b] = nil mtrie.entries[b] = nil
return return
} }
epl := len(entry.Path) epl := len(entry.Path)
if (entry.ContentType == ManifestType) && (len(path) >= epl) && (path[:epl] == entry.Path) { if (entry.ContentType == ManifestType) && (len(path) >= epl) && (path[:epl] == entry.Path) {
if self.loadSubTrie(entry, quitC) != nil { if mtrie.loadSubTrie(entry, quitC) != nil {
return return
} }
entry.subtrie.deleteEntry(path[epl:], quitC) entry.subtrie.deleteEntry(path[epl:], quitC)
@ -317,13 +317,13 @@ func (self *manifestTrie) deleteEntry(path string, quitC chan bool) {
if lastentry != nil { if lastentry != nil {
lastentry.Path = entry.Path + lastentry.Path lastentry.Path = entry.Path + lastentry.Path
} }
self.entries[b] = lastentry mtrie.entries[b] = lastentry
} }
} }
} }
func (self *manifestTrie) recalcAndStore() error { func (mtrie *manifestTrie) recalcAndStore() error {
if self.hash != nil { if mtrie.hash != nil {
return nil return nil
} }
@ -331,7 +331,7 @@ func (self *manifestTrie) recalcAndStore() error {
buffer.WriteString(`{"entries":[`) buffer.WriteString(`{"entries":[`)
list := &Manifest{} list := &Manifest{}
for _, entry := range self.entries { for _, entry := range mtrie.entries {
if entry != nil { if entry != nil {
if entry.Hash == "" { // TODO: paralellize if entry.Hash == "" { // TODO: paralellize
err := entry.subtrie.recalcAndStore() err := entry.subtrie.recalcAndStore()
@ -352,22 +352,22 @@ func (self *manifestTrie) recalcAndStore() error {
sr := bytes.NewReader(manifest) sr := bytes.NewReader(manifest)
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
key, err2 := self.dpa.Store(sr, int64(len(manifest)), wg, nil) key, err2 := mtrie.dpa.Store(sr, int64(len(manifest)), wg, nil)
wg.Wait() wg.Wait()
self.hash = key mtrie.hash = key
return err2 return err2
} }
func (self *manifestTrie) loadSubTrie(entry *manifestTrieEntry, quitC chan bool) (err error) { func (mtrie *manifestTrie) loadSubTrie(entry *manifestTrieEntry, quitC chan bool) (err error) {
if entry.subtrie == nil { if entry.subtrie == nil {
hash := common.Hex2Bytes(entry.Hash) hash := common.Hex2Bytes(entry.Hash)
entry.subtrie, err = loadManifest(self.dpa, hash, quitC) entry.subtrie, err = loadManifest(mtrie.dpa, hash, quitC)
entry.Hash = "" // might not match, should be recalculated entry.Hash = "" // might not match, should be recalculated
} }
return return
} }
func (self *manifestTrie) listWithPrefixInt(prefix, rp string, quitC chan bool, cb func(entry *manifestTrieEntry, suffix string)) error { func (mtrie *manifestTrie) listWithPrefixInt(prefix, rp string, quitC chan bool, cb func(entry *manifestTrieEntry, suffix string)) error {
plen := len(prefix) plen := len(prefix)
var start, stop int var start, stop int
if plen == 0 { if plen == 0 {
@ -384,7 +384,7 @@ func (self *manifestTrie) listWithPrefixInt(prefix, rp string, quitC chan bool,
return fmt.Errorf("aborted") return fmt.Errorf("aborted")
default: default:
} }
entry := self.entries[i] entry := mtrie.entries[i]
if entry != nil { if entry != nil {
epl := len(entry.Path) epl := len(entry.Path)
if entry.ContentType == ManifestType { if entry.ContentType == ManifestType {
@ -393,7 +393,7 @@ func (self *manifestTrie) listWithPrefixInt(prefix, rp string, quitC chan bool,
l = epl l = epl
} }
if prefix[:l] == entry.Path[:l] { if prefix[:l] == entry.Path[:l] {
err := self.loadSubTrie(entry, quitC) err := mtrie.loadSubTrie(entry, quitC)
if err != nil { if err != nil {
return err return err
} }
@ -412,23 +412,23 @@ func (self *manifestTrie) listWithPrefixInt(prefix, rp string, quitC chan bool,
return nil return nil
} }
func (self *manifestTrie) listWithPrefix(prefix string, quitC chan bool, cb func(entry *manifestTrieEntry, suffix string)) (err error) { func (mtrie *manifestTrie) listWithPrefix(prefix string, quitC chan bool, cb func(entry *manifestTrieEntry, suffix string)) (err error) {
return self.listWithPrefixInt(prefix, "", quitC, cb) return mtrie.listWithPrefixInt(prefix, "", quitC, cb)
} }
func (self *manifestTrie) findPrefixOf(path string, quitC chan bool) (entry *manifestTrieEntry, pos int) { func (mtrie *manifestTrie) findPrefixOf(path string, quitC chan bool) (entry *manifestTrieEntry, pos int) {
log.Trace(fmt.Sprintf("findPrefixOf(%s)", path)) log.Trace(fmt.Sprintf("findPrefixOf(%s)", path))
if len(path) == 0 { if len(path) == 0 {
return self.entries[256], 0 return mtrie.entries[256], 0
} }
//see if first char is in manifest entries //see if first char is in manifest entries
b := path[0] b := path[0]
entry = self.entries[b] entry = mtrie.entries[b]
if entry == nil { if entry == nil {
return self.entries[256], 0 return mtrie.entries[256], 0
} }
epl := len(entry.Path) epl := len(entry.Path)
@ -436,7 +436,7 @@ func (self *manifestTrie) findPrefixOf(path string, quitC chan bool) (entry *man
if len(path) <= epl { if len(path) <= epl {
if entry.Path[:len(path)] == path { if entry.Path[:len(path)] == path {
if entry.ContentType == ManifestType { if entry.ContentType == ManifestType {
err := self.loadSubTrie(entry, quitC) err := mtrie.loadSubTrie(entry, quitC)
if err == nil && entry.subtrie != nil { if err == nil && entry.subtrie != nil {
subentries := entry.subtrie.entries subentries := entry.subtrie.entries
for i := 0; i < len(subentries); i++ { for i := 0; i < len(subentries); i++ {
@ -457,7 +457,7 @@ func (self *manifestTrie) findPrefixOf(path string, quitC chan bool) (entry *man
log.Trace(fmt.Sprintf("entry.ContentType = %v", entry.ContentType)) log.Trace(fmt.Sprintf("entry.ContentType = %v", entry.ContentType))
//the subentry is a manifest, load subtrie //the subentry is a manifest, load subtrie
if entry.ContentType == ManifestType && (strings.Contains(entry.Path, path) || strings.Contains(path, entry.Path)) { if entry.ContentType == ManifestType && (strings.Contains(entry.Path, path) || strings.Contains(path, entry.Path)) {
err := self.loadSubTrie(entry, quitC) err := mtrie.loadSubTrie(entry, quitC)
if err != nil { if err != nil {
return nil, 0 return nil, 0
} }
@ -495,10 +495,10 @@ func RegularSlashes(path string) (res string) {
return return
} }
func (self *manifestTrie) getEntry(spath string) (entry *manifestTrieEntry, fullpath string) { func (mtrie *manifestTrie) getEntry(spath string) (entry *manifestTrieEntry, fullpath string) {
path := RegularSlashes(spath) path := RegularSlashes(spath)
var pos int var pos int
quitC := make(chan bool) quitC := make(chan bool)
entry, pos = self.findPrefixOf(path, quitC) entry, pos = mtrie.findPrefixOf(path, quitC)
return entry, path[:pos] return entry, path[:pos]
} }

20
swarm/api/storage.go
View file

@ -26,7 +26,7 @@ type Response struct {
Content string Content string
} }
// implements a service // Storage implements a service
// //
// DEPRECATED: Use the HTTP API instead // DEPRECATED: Use the HTTP API instead
type Storage struct { type Storage struct {
@ -41,8 +41,8 @@ func NewStorage(api *Api) *Storage {
// its content type // its content type
// //
// DEPRECATED: Use the HTTP API instead // DEPRECATED: Use the HTTP API instead
func (self *Storage) Put(content, contentType string) (string, error) { func (s *Storage) Put(content, contentType string) (string, error) {
key, err := self.api.Put(content, contentType) key, err := s.api.Put(content, contentType)
if err != nil { if err != nil {
return "", err return "", err
} }
@ -57,16 +57,16 @@ func (self *Storage) Put(content, contentType string) (string, error) {
// size is resp.Size // size is resp.Size
// //
// DEPRECATED: Use the HTTP API instead // DEPRECATED: Use the HTTP API instead
func (self *Storage) Get(bzzpath string) (*Response, error) { func (s *Storage) Get(bzzpath string) (*Response, error) {
uri, err := Parse(path.Join("bzz:/", bzzpath)) uri, err := Parse(path.Join("bzz:/", bzzpath))
if err != nil { if err != nil {
return nil, err return nil, err
} }
key, err := self.api.Resolve(uri) key, err := s.api.Resolve(uri)
if err != nil { if err != nil {
return nil, err return nil, err
} }
reader, mimeType, status, err := self.api.Get(key, uri.Path) reader, mimeType, status, err := s.api.Get(key, uri.Path)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -83,20 +83,20 @@ func (self *Storage) Get(bzzpath string) (*Response, error) {
return &Response{mimeType, status, expsize, string(body[:size])}, err return &Response{mimeType, status, expsize, string(body[:size])}, err
} }
// Modify(rootHash, basePath, contentHash, contentType) takes th e manifest trie rooted in rootHash, // Modify takes the manifest trie rooted in rootHash,
// and merge on to it. creating an entry w conentType (mime) // and merge on to it. creating an entry w conentType (mime)
// //
// DEPRECATED: Use the HTTP API instead // DEPRECATED: Use the HTTP API instead
func (self *Storage) Modify(rootHash, path, contentHash, contentType string) (newRootHash string, err error) { func (s *Storage) Modify(rootHash, path, contentHash, contentType string) (newRootHash string, err error) {
uri, err := Parse("bzz:/" + rootHash) uri, err := Parse("bzz:/" + rootHash)
if err != nil { if err != nil {
return "", err return "", err
} }
key, err := self.api.Resolve(uri) key, err := s.api.Resolve(uri)
if err != nil { if err != nil {
return "", err return "", err
} }
key, err = self.api.Modify(key, path, contentHash, contentType) key, err = s.api.Modify(key, path, contentHash, contentType)
if err != nil { if err != nil {
return "", err return "", err
} }

16
swarm/api/testapi.go
View file

@ -29,18 +29,18 @@ func NewControl(api *Api, hive *network.Hive) *Control {
return &Control{api, hive} return &Control{api, hive}
} }
func (self *Control) BlockNetworkRead(on bool) { func (c *Control) BlockNetworkRead(on bool) {
self.hive.BlockNetworkRead(on) c.hive.BlockNetworkRead(on)
} }
func (self *Control) SyncEnabled(on bool) { func (c *Control) SyncEnabled(on bool) {
self.hive.SyncEnabled(on) c.hive.SyncEnabled(on)
} }
func (self *Control) SwapEnabled(on bool) { func (c *Control) SwapEnabled(on bool) {
self.hive.SwapEnabled(on) c.hive.SwapEnabled(on)
} }
func (self *Control) Hive() string { func (c *Control) Hive() string {
return self.hive.String() return c.hive.String()
} }

30
swarm/fuse/fuse_file.go
View file

@ -94,50 +94,50 @@ func (file *SwarmFile) Attr(ctx context.Context, a *fuse.Attr) error {
return nil return nil
} }
func (sf *SwarmFile) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error { func (file *SwarmFile) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
sf.lock.RLock() file.lock.RLock()
defer sf.lock.RUnlock() defer file.lock.RUnlock()
if sf.reader == nil { if file.reader == nil {
sf.reader = sf.mountInfo.swarmApi.Retrieve(sf.key) file.reader = file.mountInfo.swarmApi.Retrieve(file.key)
} }
buf := make([]byte, req.Size) buf := make([]byte, req.Size)
n, err := sf.reader.ReadAt(buf, req.Offset) n, err := file.reader.ReadAt(buf, req.Offset)
if err == io.ErrUnexpectedEOF || err == io.EOF { if err == io.ErrUnexpectedEOF || err == io.EOF {
err = nil err = nil
} }
resp.Data = buf[:n] resp.Data = buf[:n]
sf.reader = nil file.reader = nil
return err return err
} }
func (sf *SwarmFile) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error { func (file *SwarmFile) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
if sf.fileSize == 0 && req.Offset == 0 { if file.fileSize == 0 && req.Offset == 0 {
// A new file is created // A new file is created
err := addFileToSwarm(sf, req.Data, len(req.Data)) err := addFileToSwarm(file, req.Data, len(req.Data))
if err != nil { if err != nil {
return err return err
} }
resp.Size = len(req.Data) resp.Size = len(req.Data)
} else if req.Offset <= sf.fileSize { } else if req.Offset <= file.fileSize {
totalSize := sf.fileSize + int64(len(req.Data)) totalSize := file.fileSize + int64(len(req.Data))
if totalSize > MaxAppendFileSize { if totalSize > MaxAppendFileSize {
log.Warn("Append file size reached (%v) : (%v)", sf.fileSize, len(req.Data)) log.Warn("Append file size reached (%v) : (%v)", file.fileSize, len(req.Data))
return errFileSizeMaxLimixReached return errFileSizeMaxLimixReached
} }
err := appendToExistingFileInSwarm(sf, req.Data, req.Offset, int64(len(req.Data))) err := appendToExistingFileInSwarm(file, req.Data, req.Offset, int64(len(req.Data)))
if err != nil { if err != nil {
return err return err
} }
resp.Size = len(req.Data) resp.Size = len(req.Data)
} else { } else {
log.Warn("Invalid write request size(%v) : off(%v)", sf.fileSize, req.Offset) log.Warn("Invalid write request size(%v) : off(%v)", file.fileSize, req.Offset)
return errInvalidOffset return errInvalidOffset
} }

8
swarm/fuse/swarmfs.go
View file

@ -39,7 +39,7 @@ var (
) )
type SwarmFS struct { type SwarmFS struct {
swarmApi *api.Api swarmAPI *api.Api
activeMounts map[string]*MountInfo activeMounts map[string]*MountInfo
swarmFsLock *sync.RWMutex swarmFsLock *sync.RWMutex
} }
@ -47,7 +47,7 @@ type SwarmFS struct {
func NewSwarmFS(api *api.Api) *SwarmFS { func NewSwarmFS(api *api.Api) *SwarmFS {
swarmfsLock.Do(func() { swarmfsLock.Do(func() {
swarmfs = &SwarmFS{ swarmfs = &SwarmFS{
swarmApi: api, swarmAPI: api,
swarmFsLock: &sync.RWMutex{}, swarmFsLock: &sync.RWMutex{},
activeMounts: map[string]*MountInfo{}, activeMounts: map[string]*MountInfo{},
} }
@ -56,10 +56,10 @@ func NewSwarmFS(api *api.Api) *SwarmFS {
} }
// Inode numbers need to be unique, they are used for caching inside fuse // NewInode numbers need to be unique, they are used for caching inside fuse
func NewInode() uint64 { func NewInode() uint64 {
inodeLock.Lock() inodeLock.Lock()
defer inodeLock.Unlock() defer inodeLock.Unlock()
inode += 1 inode++
return inode return inode
} }

8
swarm/fuse/swarmfs_fallback.go
View file

@ -34,18 +34,18 @@ type MountInfo struct {
LatestManifest string LatestManifest string
} }
func (self *SwarmFS) Mount(mhash, mountpoint string) (*MountInfo, error) { func (sf *SwarmFS) Mount(mhash, mountpoint string) (*MountInfo, error) {
return nil, errNoFUSE return nil, errNoFUSE
} }
func (self *SwarmFS) Unmount(mountpoint string) (bool, error) { func (sf *SwarmFS) Unmount(mountpoint string) (bool, error) {
return false, errNoFUSE return false, errNoFUSE
} }
func (self *SwarmFS) Listmounts() ([]*MountInfo, error) { func (sf *SwarmFS) Listmounts() ([]*MountInfo, error) {
return nil, errNoFUSE return nil, errNoFUSE
} }
func (self *SwarmFS) Stop() error { func (sf *SwarmFS) Stop() error {
return nil return nil
} }

50
swarm/fuse/swarmfs_unix.go
View file

@ -48,14 +48,14 @@ func isFUSEUnsupportedError(err error) bool {
return err == fuse.ErrOSXFUSENotFound return err == fuse.ErrOSXFUSENotFound
} }
// information about every active mount // MountInfo is information about every active mount
type MountInfo struct { type MountInfo struct {
MountPoint string MountPoint string
StartManifest string StartManifest string
LatestManifest string LatestManifest string
rootDir *SwarmDir rootDir *SwarmDir
fuseConnection *fuse.Conn fuseConnection *fuse.Conn
swarmApi *api.Api swarmAPI *api.Api
lock *sync.RWMutex lock *sync.RWMutex
} }
@ -66,13 +66,13 @@ func NewMountInfo(mhash, mpoint string, sapi *api.Api) *MountInfo {
LatestManifest: mhash, LatestManifest: mhash,
rootDir: nil, rootDir: nil,
fuseConnection: nil, fuseConnection: nil,
swarmApi: sapi, swarmAPI: sapi,
lock: &sync.RWMutex{}, lock: &sync.RWMutex{},
} }
return newMountInfo return newMountInfo
} }
func (self *SwarmFS) Mount(mhash, mountpoint string) (*MountInfo, error) { func (sf *SwarmFS) Mount(mhash, mountpoint string) (*MountInfo, error) {
if mountpoint == "" { if mountpoint == "" {
return nil, errEmptyMountPoint return nil, errEmptyMountPoint
@ -82,25 +82,25 @@ func (self *SwarmFS) Mount(mhash, mountpoint string) (*MountInfo, error) {
return nil, err return nil, err
} }
self.swarmFsLock.Lock() sf.swarmFsLock.Lock()
defer self.swarmFsLock.Unlock() defer sf.swarmFsLock.Unlock()
noOfActiveMounts := len(self.activeMounts) noOfActiveMounts := len(sf.activeMounts)
if noOfActiveMounts >= maxFuseMounts { if noOfActiveMounts >= maxFuseMounts {
return nil, errMaxMountCount return nil, errMaxMountCount
} }
if _, ok := self.activeMounts[cleanedMountPoint]; ok { if _, ok := sf.activeMounts[cleanedMountPoint]; ok {
return nil, errAlreadyMounted return nil, errAlreadyMounted
} }
log.Info(fmt.Sprintf("Attempting to mount %s ", cleanedMountPoint)) log.Info(fmt.Sprintf("Attempting to mount %s ", cleanedMountPoint))
_, manifestEntryMap, err := self.swarmApi.BuildDirectoryTree(mhash, true) _, manifestEntryMap, err := sf.swarmAPI.BuildDirectoryTree(mhash, true)
if err != nil { if err != nil {
return nil, err return nil, err
} }
mi := NewMountInfo(mhash, cleanedMountPoint, self.swarmApi) mi := NewMountInfo(mhash, cleanedMountPoint, sf.swarmAPI)
dirTree := map[string]*SwarmDir{} dirTree := map[string]*SwarmDir{}
rootDir := NewSwarmDir("/", mi) rootDir := NewSwarmDir("/", mi)
@ -174,21 +174,21 @@ func (self *SwarmFS) Mount(mhash, mountpoint string) (*MountInfo, error) {
log.Info("Now serving swarm FUSE FS", "manifest", mhash, "mountpoint", cleanedMountPoint) log.Info("Now serving swarm FUSE FS", "manifest", mhash, "mountpoint", cleanedMountPoint)
} }
self.activeMounts[cleanedMountPoint] = mi sf.activeMounts[cleanedMountPoint] = mi
return mi, nil return mi, nil
} }
func (self *SwarmFS) Unmount(mountpoint string) (*MountInfo, error) { func (sf *SwarmFS) Unmount(mountpoint string) (*MountInfo, error) {
self.swarmFsLock.Lock() sf.swarmFsLock.Lock()
defer self.swarmFsLock.Unlock() defer sf.swarmFsLock.Unlock()
cleanedMountPoint, err := filepath.Abs(filepath.Clean(mountpoint)) cleanedMountPoint, err := filepath.Abs(filepath.Clean(mountpoint))
if err != nil { if err != nil {
return nil, err return nil, err
} }
mountInfo := self.activeMounts[cleanedMountPoint] mountInfo := sf.activeMounts[cleanedMountPoint]
if mountInfo == nil || mountInfo.MountPoint != cleanedMountPoint { if mountInfo == nil || mountInfo.MountPoint != cleanedMountPoint {
return nil, fmt.Errorf("%s is not mounted", cleanedMountPoint) return nil, fmt.Errorf("%s is not mounted", cleanedMountPoint)
@ -204,7 +204,7 @@ func (self *SwarmFS) Unmount(mountpoint string) (*MountInfo, error) {
} }
mountInfo.fuseConnection.Close() mountInfo.fuseConnection.Close()
delete(self.activeMounts, cleanedMountPoint) delete(sf.activeMounts, cleanedMountPoint)
succString := fmt.Sprintf("UnMounting %v succeeded", cleanedMountPoint) succString := fmt.Sprintf("UnMounting %v succeeded", cleanedMountPoint)
log.Info(succString) log.Info(succString)
@ -212,21 +212,21 @@ func (self *SwarmFS) Unmount(mountpoint string) (*MountInfo, error) {
return mountInfo, nil return mountInfo, nil
} }
func (self *SwarmFS) Listmounts() []*MountInfo { func (sf *SwarmFS) Listmounts() []*MountInfo {
self.swarmFsLock.RLock() sf.swarmFsLock.RLock()
defer self.swarmFsLock.RUnlock() defer sf.swarmFsLock.RUnlock()
rows := make([]*MountInfo, 0, len(self.activeMounts)) rows := make([]*MountInfo, 0, len(sf.activeMounts))
for _, mi := range self.activeMounts { for _, mi := range sf.activeMounts {
rows = append(rows, mi) rows = append(rows, mi)
} }
return rows return rows
} }
func (self *SwarmFS) Stop() bool { func (sf *SwarmFS) Stop() bool {
for mp := range self.activeMounts { for mp := range sf.activeMounts {
mountInfo := self.activeMounts[mp] mountInfo := sf.activeMounts[mp]
self.Unmount(mountInfo.MountPoint) sf.Unmount(mountInfo.MountPoint)
} }
return true return true
} }

41
swarm/network/depo.go
View file

@ -36,7 +36,7 @@ var (
syncSendNotFound = metrics.NewRegisteredCounter("network.sync.send.notfound", nil) syncSendNotFound = metrics.NewRegisteredCounter("network.sync.send.notfound", nil)
) )
// Handler for storage/retrieval related protocol requests // Depo is a handler for storage/retrieval related protocol requests
// implements the StorageHandler interface used by the bzz protocol // implements the StorageHandler interface used by the bzz protocol
type Depo struct { type Depo struct {
hashfunc storage.SwarmHasher hashfunc storage.SwarmHasher
@ -52,20 +52,20 @@ func NewDepo(hash storage.SwarmHasher, localStore, remoteStore storage.ChunkStor
} }
} }
// Handles UnsyncedKeysMsg after msg decoding - unsynced hashes upto sync state // HandleUnsyncedKeysMsg after msg decoding - unsynced hashes upto sync state
// * the remote sync state is just stored and handled in protocol // * the remote sync state is just stored and handled in protocol
// * filters through the new syncRequests and send the ones missing // * filters through the new syncRequests and send the ones missing
// * back immediately as a deliveryRequest message // * back immediately as a deliveryRequest message
// * empty message just pings back for more (is this needed?) // * empty message just pings back for more (is this needed?)
// * strict signed sync states may be needed. // * strict signed sync states may be needed.
func (self *Depo) HandleUnsyncedKeysMsg(req *unsyncedKeysMsgData, p *peer) error { func (depo *Depo) HandleUnsyncedKeysMsg(req *unsyncedKeysMsgData, p *peer) error {
unsynced := req.Unsynced unsynced := req.Unsynced
var missing []*syncRequest var missing []*syncRequest
var chunk *storage.Chunk var chunk *storage.Chunk
var err error var err error
for _, req := range unsynced { for _, req := range unsynced {
// skip keys that are found, // skip keys that are found,
chunk, err = self.localStore.Get(req.Key[:]) chunk, err = depo.localStore.Get(req.Key[:])
if err != nil || chunk.SData == nil { if err != nil || chunk.SData == nil {
missing = append(missing, req) missing = append(missing, req)
} }
@ -82,13 +82,13 @@ func (self *Depo) HandleUnsyncedKeysMsg(req *unsyncedKeysMsgData, p *peer) error
return nil return nil
} }
// Handles deliveryRequestMsg // HandleDeliveryRequestMsg does the following:
// * serves actual chunks asked by the remote peer // * serves actual chunks asked by the remote peer
// by pushing to the delivery queue (sync db) of the correct priority // by pushing to the delivery queue (sync db) of the correct priority
// (remote peer is free to reprioritize) // (remote peer is free to reprioritize)
// * the message implies remote peer wants more, so trigger for // * the message implies remote peer wants more, so trigger for
// * new outgoing unsynced keys message is fired // * new outgoing unsynced keys message is fired
func (self *Depo) HandleDeliveryRequestMsg(req *deliveryRequestMsgData, p *peer) error { func (depo *Depo) HandleDeliveryRequestMsg(req *deliveryRequestMsgData, p *peer) error {
deliver := req.Deliver deliver := req.Deliver
// queue the actual delivery of a chunk () // queue the actual delivery of a chunk ()
log.Trace(fmt.Sprintf("Depo.HandleDeliveryRequestMsg: received %v delivery requests: %v", len(deliver), deliver)) log.Trace(fmt.Sprintf("Depo.HandleDeliveryRequestMsg: received %v delivery requests: %v", len(deliver), deliver))
@ -96,7 +96,7 @@ func (self *Depo) HandleDeliveryRequestMsg(req *deliveryRequestMsgData, p *peer)
// TODO: look up in cache here or in deliveries // TODO: look up in cache here or in deliveries
// priorities are taken from the message so the remote party can // priorities are taken from the message so the remote party can
// reprioritise to at their leisure // reprioritise to at their leisure
// r = self.pullCached(sreq.Key) // pulls and deletes from cache // r = depo.pullCached(sreq.Key) // pulls and deletes from cache
Push(p, sreq.Key, sreq.Priority) Push(p, sreq.Key, sreq.Priority)
} }
@ -108,10 +108,10 @@ func (self *Depo) HandleDeliveryRequestMsg(req *deliveryRequestMsgData, p *peer)
// the entrypoint for store requests coming from the bzz wire protocol // the entrypoint for store requests coming from the bzz wire protocol
// if key found locally, return. otherwise // if key found locally, return. otherwise
// remote is untrusted, so hash is verified and chunk passed on to NetStore // remote is untrusted, so hash is verified and chunk passed on to NetStore
func (self *Depo) HandleStoreRequestMsg(req *storeRequestMsgData, p *peer) { func (depo *Depo) HandleStoreRequestMsg(req *storeRequestMsgData, p *peer) {
var islocal bool var islocal bool
req.from = p req.from = p
chunk, err := self.localStore.Get(req.Key) chunk, err := depo.localStore.Get(req.Key)
switch { switch {
case err != nil: case err != nil:
log.Trace(fmt.Sprintf("Depo.handleStoreRequest: %v not found locally. create new chunk/request", req.Key)) log.Trace(fmt.Sprintf("Depo.handleStoreRequest: %v not found locally. create new chunk/request", req.Key))
@ -133,7 +133,7 @@ func (self *Depo) HandleStoreRequestMsg(req *storeRequestMsgData, p *peer) {
//return //return
} }
hasher := self.hashfunc() hasher := depo.hashfunc()
hasher.Write(req.SData) hasher.Write(req.SData)
if !bytes.Equal(hasher.Sum(nil), req.Key) { if !bytes.Equal(hasher.Sum(nil), req.Key) {
// data does not validate, ignore // data does not validate, ignore
@ -150,12 +150,12 @@ func (self *Depo) HandleStoreRequestMsg(req *storeRequestMsgData, p *peer) {
chunk.Size = int64(binary.LittleEndian.Uint64(req.SData[0:8])) chunk.Size = int64(binary.LittleEndian.Uint64(req.SData[0:8]))
log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p)) log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p))
chunk.Source = p chunk.Source = p
self.netStore.Put(chunk) depo.netStore.Put(chunk)
} }
// entrypoint for retrieve requests coming from the bzz wire protocol // entrypoint for retrieve requests coming from the bzz wire protocol
// checks swap balance - return if peer has no credit // checks swap balance - return if peer has no credit
func (self *Depo) HandleRetrieveRequestMsg(req *retrieveRequestMsgData, p *peer) { func (depo *Depo) HandleRetrieveRequestMsg(req *retrieveRequestMsgData, p *peer) {
req.from = p req.from = p
// swap - record credit for 1 request // swap - record credit for 1 request
// note that only charge actual reqsearches // note that only charge actual reqsearches
@ -171,8 +171,8 @@ func (self *Depo) HandleRetrieveRequestMsg(req *retrieveRequestMsgData, p *peer)
// call storage.NetStore#Get which // call storage.NetStore#Get which
// blocks until local retrieval finished // blocks until local retrieval finished
// launches cloud retrieval // launches cloud retrieval
chunk, _ := self.netStore.Get(req.Key) chunk, _ := depo.netStore.Get(req.Key)
req = self.strategyUpdateRequest(chunk.Req, req) req = depo.strategyUpdateRequest(chunk.Req, req)
// check if we can immediately deliver // check if we can immediately deliver
if chunk.SData != nil { if chunk.SData != nil {
log.Trace(fmt.Sprintf("Depo.HandleRetrieveRequest: %v - content found, delivering...", req.Key.Log())) log.Trace(fmt.Sprintf("Depo.HandleRetrieveRequest: %v - content found, delivering...", req.Key.Log()))
@ -197,27 +197,26 @@ func (self *Depo) HandleRetrieveRequestMsg(req *retrieveRequestMsgData, p *peer)
} }
// add peer request the chunk and decides the timeout for the response if still searching // add peer request the chunk and decides the timeout for the response if still searching
func (self *Depo) strategyUpdateRequest(rs *storage.RequestStatus, origReq *retrieveRequestMsgData) (req *retrieveRequestMsgData) { func (depo *Depo) strategyUpdateRequest(rs *storage.RequestStatus, origReq *retrieveRequestMsgData) (req *retrieveRequestMsgData) {
log.Trace(fmt.Sprintf("Depo.strategyUpdateRequest: key %v", origReq.Key.Log())) log.Trace(fmt.Sprintf("Depo.strategyUpdateRequest: key %v", origReq.Key.Log()))
// we do not create an alternative one // we do not create an alternative one
req = origReq req = origReq
if rs != nil { if rs != nil {
self.addRequester(rs, req) depo.addRequester(rs, req)
req.setTimeout(self.searchTimeout(rs, req)) req.setTimeout(depo.searchTimeout(rs, req))
} }
return return
} }
// decides the timeout promise sent with the immediate peers response to a retrieve request // decides the timeout promise sent with the immediate peers response to a retrieve request
// if timeout is explicitly set and expired // if timeout is explicitly set and expired
func (self *Depo) searchTimeout(rs *storage.RequestStatus, req *retrieveRequestMsgData) (timeout *time.Time) { func (depo *Depo) searchTimeout(rs *storage.RequestStatus, req *retrieveRequestMsgData) (timeout *time.Time) {
reqt := req.getTimeout() reqt := req.getTimeout()
t := time.Now().Add(searchTimeout) t := time.Now().Add(searchTimeout)
if reqt != nil && reqt.Before(t) { if reqt != nil && reqt.Before(t) {
return reqt return reqt
} else {
return &t
} }
return &t
} }
/* /*
@ -225,7 +224,7 @@ adds a new peer to an existing open request
only add if less than requesterCount peers forwarded the same request id so far only add if less than requesterCount peers forwarded the same request id so far
note this is done irrespective of status (searching or found) note this is done irrespective of status (searching or found)
*/ */
func (self *Depo) addRequester(rs *storage.RequestStatus, req *retrieveRequestMsgData) { func (depo *Depo) addRequester(rs *storage.RequestStatus, req *retrieveRequestMsgData) {
log.Trace(fmt.Sprintf("Depo.addRequester: key %v - add peer to req.Id %v", req.Key.Log(), req.Id)) log.Trace(fmt.Sprintf("Depo.addRequester: key %v - add peer to req.Id %v", req.Key.Log(), req.Id))
list := rs.Requesters[req.Id] list := rs.Requesters[req.Id]
rs.Requesters[req.Id] = append(list, req) rs.Requesters[req.Id] = append(list, req)

18
swarm/network/forwarding.go
View file

@ -45,7 +45,7 @@ func NewForwarder(hive *Hive) *forwarder {
} }
// generate a unique id uint64 // generate a unique id uint64
func generateId() uint64 { func generateID() uint64 {
r := rand.New(rand.NewSource(time.Now().UnixNano())) r := rand.New(rand.NewSource(time.Now().UnixNano()))
return uint64(r.Int63()) return uint64(r.Int63())
} }
@ -54,8 +54,8 @@ var searchTimeout = 3 * time.Second
// forwarding logic // forwarding logic
// logic propagating retrieve requests to peers given by the kademlia hive // logic propagating retrieve requests to peers given by the kademlia hive
func (self *forwarder) Retrieve(chunk *storage.Chunk) { func (f *forwarder) Retrieve(chunk *storage.Chunk) {
peers := self.hive.getPeers(chunk.Key, 0) peers := f.hive.getPeers(chunk.Key, 0)
log.Trace(fmt.Sprintf("forwarder.Retrieve: %v - received %d peers from KΛÐΞMLIΛ...", chunk.Key.Log(), len(peers))) log.Trace(fmt.Sprintf("forwarder.Retrieve: %v - received %d peers from KΛÐΞMLIΛ...", chunk.Key.Log(), len(peers)))
OUT: OUT:
for _, p := range peers { for _, p := range peers {
@ -70,7 +70,7 @@ OUT:
} }
req := &retrieveRequestMsgData{ req := &retrieveRequestMsgData{
Key: chunk.Key, Key: chunk.Key,
Id: generateId(), Id: generateID(),
} }
var err error var err error
if p.swap != nil { if p.swap != nil {
@ -87,7 +87,7 @@ OUT:
// requests to specific peers given by the kademlia hive // requests to specific peers given by the kademlia hive
// except for peers that the store request came from (if any) // except for peers that the store request came from (if any)
// delivery queueing taken care of by syncer // delivery queueing taken care of by syncer
func (self *forwarder) Store(chunk *storage.Chunk) { func (f *forwarder) Store(chunk *storage.Chunk) {
var n int var n int
msg := &storeRequestMsgData{ msg := &storeRequestMsgData{
Key: chunk.Key, Key: chunk.Key,
@ -97,7 +97,7 @@ func (self *forwarder) Store(chunk *storage.Chunk) {
if chunk.Source != nil { if chunk.Source != nil {
source = chunk.Source.(*peer) source = chunk.Source.(*peer)
} }
for _, p := range self.hive.getPeers(chunk.Key, 0) { for _, p := range f.hive.getPeers(chunk.Key, 0) {
log.Trace(fmt.Sprintf("forwarder.Store: %v %v", p, chunk)) log.Trace(fmt.Sprintf("forwarder.Store: %v %v", p, chunk))
if p.syncer != nil && (source == nil || p.Addr() != source.Addr()) { if p.syncer != nil && (source == nil || p.Addr() != source.Addr()) {
@ -109,7 +109,7 @@ func (self *forwarder) Store(chunk *storage.Chunk) {
} }
// once a chunk is found deliver it to its requesters unless timed out // once a chunk is found deliver it to its requesters unless timed out
func (self *forwarder) Deliver(chunk *storage.Chunk) { func (f *forwarder) Deliver(chunk *storage.Chunk) {
// iterate over request entries // iterate over request entries
for id, requesters := range chunk.Req.Requesters { for id, requesters := range chunk.Req.Requesters {
counter := requesterCount counter := requesterCount
@ -137,14 +137,14 @@ func (self *forwarder) Deliver(chunk *storage.Chunk) {
} }
} }
// initiate delivery of a chunk to a particular peer via syncer#addRequest // Deliver initiates delivery of a chunk to a particular peer via syncer#addRequest
// depending on syncer mode and priority settings and sync request type // depending on syncer mode and priority settings and sync request type
// this either goes via confirmation roundtrip or queued or pushed directly // this either goes via confirmation roundtrip or queued or pushed directly
func Deliver(p *peer, req interface{}, ty int) { func Deliver(p *peer, req interface{}, ty int) {
p.syncer.addRequest(req, ty) p.syncer.addRequest(req, ty)
} }
// push chunk over to peer // Push chunk over to peer
func Push(p *peer, key storage.Key, priority uint) { func Push(p *peer, key storage.Key, priority uint) {
p.syncer.doDelivery(key, priority, p.syncer.quit) p.syncer.doDelivery(key, priority, p.syncer.quit)
} }

60
swarm/network/hive.go
View file

@ -92,8 +92,8 @@ func NewDefaultHiveParams() *HiveParams {
//this can only finally be set after all config options (file, cmd line, env vars) //this can only finally be set after all config options (file, cmd line, env vars)
//have been evaluated //have been evaluated
func (self *HiveParams) Init(path string) { func (params *HiveParams) Init(path string) {
self.KadDbPath = filepath.Join(path, "bzz-peers.json") params.KadDbPath = filepath.Join(path, "bzz-peers.json")
} }
func NewHive(addr common.Hash, params *HiveParams, swapEnabled, syncEnabled bool) *Hive { func NewHive(addr common.Hash, params *HiveParams, swapEnabled, syncEnabled bool) *Hive {
@ -108,53 +108,53 @@ func NewHive(addr common.Hash, params *HiveParams, swapEnabled, syncEnabled bool
} }
} }
func (self *Hive) SyncEnabled(on bool) { func (hive *Hive) SyncEnabled(on bool) {
self.syncEnabled = on hive.syncEnabled = on
} }
func (self *Hive) SwapEnabled(on bool) { func (hive *Hive) SwapEnabled(on bool) {
self.swapEnabled = on hive.swapEnabled = on
} }
func (self *Hive) BlockNetworkRead(on bool) { func (hive *Hive) BlockNetworkRead(on bool) {
self.blockRead = on hive.blockRead = on
} }
func (self *Hive) BlockNetworkWrite(on bool) { func (hive *Hive) BlockNetworkWrite(on bool) {
self.blockWrite = on hive.blockWrite = on
} }
// public accessor to the hive base address // public accessor to the hive base address
func (self *Hive) Addr() kademlia.Address { func (hive *Hive) Addr() kademlia.Address {
return self.addr return hive.addr
} }
// Start receives network info only at startup // Start receives network info only at startup
// listedAddr is a function to retrieve listening address to advertise to peers // listedAddr is a function to retrieve listening address to advertise to peers
// connectPeer is a function to connect to a peer based on its NodeID or enode URL // connectPeer is a function to connect to a peer based on its NodeID or enode URL
// there are called on the p2p.Server which runs on the node // there are called on the p2p.Server which runs on the node
func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPeer func(string) error) (err error) { func (hive *Hive) Start(id discover.NodeID, listenAddr func() string, connectPeer func(string) error) (err error) {
self.toggle = make(chan bool) hive.toggle = make(chan bool)
self.more = make(chan bool) hive.more = make(chan bool)
self.quit = make(chan bool) hive.quit = make(chan bool)
self.id = id hive.id = id
self.listenAddr = listenAddr hive.listenAddr = listenAddr
err = self.kad.Load(self.path, nil) err = hive.kad.Load(hive.path, nil)
if err != nil { if err != nil {
log.Warn(fmt.Sprintf("Warning: error reading kaddb '%s' (skipping): %v", self.path, err)) log.Warn(fmt.Sprintf("Warning: error reading kaddb '%s' (skipping): %v", hive.path, err))
err = nil err = nil
} }
// this loop is doing bootstrapping and maintains a healthy table // this loop is doing bootstrapping and maintains a healthy table
go self.keepAlive() go hive.keepAlive()
go func() { go func() {
// whenever toggled ask kademlia about most preferred peer // whenever toggled ask kademlia about most preferred peer
for alive := range self.more { for alive := range hive.more {
if !alive { if !alive {
// receiving false closes the loop while allowing parallel routines // receiving false closes the loop while allowing parallel routines
// to attempt to write to more (remove Peer when shutting down) // to attempt to write to more (remove Peer when shutting down)
return return
} }
node, need, proxLimit := self.kad.Suggest() node, need, proxLimit := hive.kad.Suggest()
if node != nil && len(node.Url) > 0 { if node != nil && len(node.Url) > 0 {
log.Trace(fmt.Sprintf("call known bee %v", node.Url)) log.Trace(fmt.Sprintf("call known bee %v", node.Url))
@ -164,10 +164,10 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee
} }
if need { if need {
// a random peer is taken from the table // a random peer is taken from the table
peers := self.kad.FindClosest(kademlia.RandomAddressAt(self.addr, rand.Intn(self.kad.MaxProx)), 1) peers := hive.kad.FindClosest(kademlia.RandomAddressAt(hive.addr, rand.Intn(hive.kad.MaxProx)), 1)
if len(peers) > 0 { if len(peers) > 0 {
// a random address at prox bin 0 is sent for lookup // a random address at prox bin 0 is sent for lookup
randAddr := kademlia.RandomAddressAt(self.addr, proxLimit) randAddr := kademlia.RandomAddressAt(hive.addr, proxLimit)
req := &retrieveRequestMsgData{ req := &retrieveRequestMsgData{
Key: storage.Key(randAddr[:]), Key: storage.Key(randAddr[:]),
} }
@ -181,11 +181,11 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee
log.Info(fmt.Sprintf("no need for more bees")) log.Info(fmt.Sprintf("no need for more bees"))
} }
select { select {
case self.toggle <- need: case hive.toggle <- need:
case <-self.quit: case <-hive.quit:
return return
} }
log.Debug(fmt.Sprintf("queen's address: %v, population: %d (%d)", self.addr, self.kad.Count(), self.kad.DBCount())) log.Debug(fmt.Sprintf("queen's address: %v, population: %d (%d)", hive.addr, hive.kad.Count(), hive.kad.DBCount()))
} }
}() }()
return return
@ -193,8 +193,8 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee
// keepAlive is a forever loop // keepAlive is a forever loop
// in its awake state it periodically triggers connection attempts // in its awake state it periodically triggers connection attempts
// by writing to self.more until Kademlia Table is saturated // by writing to hive.more until Kademlia Table is saturated
// wake state is toggled by writing to self.toggle // wake state is toggled by writing to hive.toggle
// it restarts if the table becomes non-full again due to disconnections // it restarts if the table becomes non-full again due to disconnections
func (self *Hive) keepAlive() { func (self *Hive) keepAlive() {
alarm := time.NewTicker(time.Duration(self.callInterval)).C alarm := time.NewTicker(time.Duration(self.callInterval)).C

10
swarm/network/kademlia/address.go
View file

@ -39,7 +39,7 @@ func (a *Address) UnmarshalJSON(value []byte) error {
return nil return nil
} }
// the string form of the binary representation of an address (only first 8 bits) // Bin returns the string form of the binary representation of an address (only first 8 bits)
func (a Address) Bin() string { func (a Address) Bin() string {
var bs []string var bs []string
for _, b := range a[:] { for _, b := range a[:] {
@ -75,7 +75,7 @@ func proximity(one, other Address) (ret int) {
return len(one) * 8 return len(one) * 8
} }
// Address.ProxCmp compares the distances a->target and b->target. // ProxCmp compares the distances a->target and b->target.
// Returns -1 if a is closer to target, 1 if b is closer to target // Returns -1 if a is closer to target, 1 if b is closer to target
// and 0 if they are equal. // and 0 if they are equal.
func (target Address) ProxCmp(a, b Address) int { func (target Address) ProxCmp(a, b Address) int {
@ -91,7 +91,7 @@ func (target Address) ProxCmp(a, b Address) int {
return 0 return 0
} }
// randomAddressAt(address, prox) generates a random address // RandomAddressAt generates a random address
// at proximity order prox relative to address // at proximity order prox relative to address
// if prox is negative a random address is generated // if prox is negative a random address is generated
func RandomAddressAt(self Address, prox int) (addr Address) { func RandomAddressAt(self Address, prox int) (addr Address) {
@ -116,7 +116,7 @@ func RandomAddressAt(self Address, prox int) (addr Address) {
return return
} }
// KeyRange(a0, a1, proxLimit) returns the address inclusive address // KeyRange returns the address inclusive address
// range that contain addresses closer to one than other // range that contain addresses closer to one than other
func KeyRange(one, other Address, proxLimit int) (start, stop Address) { func KeyRange(one, other Address, proxLimit int) (start, stop Address) {
prox := proximity(one, other) prox := proximity(one, other)
@ -167,7 +167,7 @@ func CommonBitsAddrByte(self, other Address, b byte, prox int) (addr Address) {
return CommonBitsAddrF(self, other, func() byte { return b }, prox) return CommonBitsAddrF(self, other, func() byte { return b }, prox)
} }
// randomAddressAt() generates a random address // RandomAddress generates a random address
func RandomAddress() Address { func RandomAddress() Address {
return RandomAddressAt(Address{}, -1) return RandomAddressAt(Address{}, -1)
} }

108
swarm/network/kademlia/kaddb.go
View file

@ -43,17 +43,17 @@ type NodeRecord struct {
node Node node Node
} }
func (self *NodeRecord) setSeen() { func (nr *NodeRecord) setSeen() {
t := time.Now() t := time.Now()
self.Seen = t nr.Seen = t
self.After = t nr.After = t
} }
func (self *NodeRecord) String() string { func (nr *NodeRecord) String() string {
return fmt.Sprintf("<%v>", self.Addr) return fmt.Sprintf("<%v>", nr.Addr)
} }
// persisted node record database () // KadDb is a persisted node record database ()
type KadDb struct { type KadDb struct {
Address Address Address Address
Nodes [][]*NodeRecord Nodes [][]*NodeRecord
@ -77,11 +77,11 @@ func newKadDb(addr Address, params *KadParams) *KadDb {
} }
} }
func (self *KadDb) findOrCreate(index int, a Address, url string) *NodeRecord { func (kdb *KadDb) findOrCreate(index int, a Address, url string) *NodeRecord {
defer self.lock.Unlock() defer kdb.lock.Unlock()
self.lock.Lock() kdb.lock.Lock()
record, found := self.index[a] record, found := kdb.index[a]
if !found { if !found {
record = &NodeRecord{ record = &NodeRecord{
Addr: a, Addr: a,
@ -89,8 +89,8 @@ func (self *KadDb) findOrCreate(index int, a Address, url string) *NodeRecord {
} }
log.Info(fmt.Sprintf("add new record %v to kaddb", record)) log.Info(fmt.Sprintf("add new record %v to kaddb", record))
// insert in kaddb // insert in kaddb
self.index[a] = record kdb.index[a] = record
self.Nodes[index] = append(self.Nodes[index], record) kdb.Nodes[index] = append(kdb.Nodes[index], record)
} else { } else {
log.Info(fmt.Sprintf("found record %v in kaddb", record)) log.Info(fmt.Sprintf("found record %v in kaddb", record))
} }
@ -102,26 +102,26 @@ func (self *KadDb) findOrCreate(index int, a Address, url string) *NodeRecord {
} }
// add adds node records to kaddb (persisted node record db) // add adds node records to kaddb (persisted node record db)
func (self *KadDb) add(nrs []*NodeRecord, proximityBin func(Address) int) { func (kdb *KadDb) add(nrs []*NodeRecord, proximityBin func(Address) int) {
defer self.lock.Unlock() defer kdb.lock.Unlock()
self.lock.Lock() kdb.lock.Lock()
var n int var n int
var nodes []*NodeRecord var nodes []*NodeRecord
for _, node := range nrs { for _, node := range nrs {
_, found := self.index[node.Addr] _, found := kdb.index[node.Addr]
if !found && node.Addr != self.Address { if !found && node.Addr != kdb.Address {
node.setSeen() node.setSeen()
self.index[node.Addr] = node kdb.index[node.Addr] = node
index := proximityBin(node.Addr) index := proximityBin(node.Addr)
dbcursor := self.cursors[index] dbcursor := kdb.cursors[index]
nodes = self.Nodes[index] nodes = kdb.Nodes[index]
// this is inefficient for allocation, need to just append then shift // this is inefficient for allocation, need to just append then shift
newnodes := make([]*NodeRecord, len(nodes)+1) newnodes := make([]*NodeRecord, len(nodes)+1)
copy(newnodes[:], nodes[:dbcursor]) copy(newnodes[:], nodes[:dbcursor])
newnodes[dbcursor] = node newnodes[dbcursor] = node
copy(newnodes[dbcursor+1:], nodes[dbcursor:]) copy(newnodes[dbcursor+1:], nodes[dbcursor:])
log.Trace(fmt.Sprintf("new nodes: %v, nodes: %v", newnodes, nodes)) log.Trace(fmt.Sprintf("new nodes: %v, nodes: %v", newnodes, nodes))
self.Nodes[index] = newnodes kdb.Nodes[index] = newnodes
n++ n++
} }
} }
@ -168,10 +168,10 @@ offline past peer)
The second argument returned names the first missing slot found The second argument returned names the first missing slot found
*/ */
func (self *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRecord, need bool, proxLimit int) { func (kdb *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRecord, need bool, proxLimit int) {
// return nil, proxLimit indicates that all buckets are filled // return nil, proxLimit indicates that all buckets are filled
defer self.lock.Unlock() defer kdb.lock.Unlock()
self.lock.Lock() kdb.lock.Lock()
var interval time.Duration var interval time.Duration
var found bool var found bool
@ -185,7 +185,7 @@ func (self *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRe
for rounds := 1; rounds <= maxBinSize; rounds++ { for rounds := 1; rounds <= maxBinSize; rounds++ {
ROUND: ROUND:
// iterate over rows from PO 0 upto MaxProx // iterate over rows from PO 0 upto MaxProx
for po, dbrow := range self.Nodes { for po, dbrow := range kdb.Nodes {
// if row has rounds connected peers, then take the next // if row has rounds connected peers, then take the next
if binSize(po) >= rounds { if binSize(po) >= rounds {
continue ROUND continue ROUND
@ -200,7 +200,7 @@ func (self *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRe
// there is a missing slot - finding a node to connect to // there is a missing slot - finding a node to connect to
// select a node record from the relavant kaddb row (of identical prox order) // select a node record from the relavant kaddb row (of identical prox order)
ROW: ROW:
for cursor = self.cursors[po]; !found && count < len(dbrow); cursor = (cursor + 1) % len(dbrow) { for cursor = kdb.cursors[po]; !found && count < len(dbrow); cursor = (cursor + 1) % len(dbrow) {
count++ count++
node = dbrow[cursor] node = dbrow[cursor]
@ -217,10 +217,10 @@ func (self *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRe
} }
delta = time.Since(node.Seen) delta = time.Since(node.Seen)
if delta < self.initialRetryInterval { if delta < kdb.initialRetryInterval {
delta = self.initialRetryInterval delta = kdb.initialRetryInterval
} }
if delta > self.purgeInterval { if delta > kdb.purgeInterval {
// remove node // remove node
purge[cursor] = true purge[cursor] = true
log.Debug(fmt.Sprintf("kaddb record %v (PO%03d:%d) unreachable since %v. Removed", node.Addr, po, cursor, node.Seen)) log.Debug(fmt.Sprintf("kaddb record %v (PO%03d:%d) unreachable since %v. Removed", node.Addr, po, cursor, node.Seen))
@ -230,15 +230,15 @@ func (self *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRe
log.Debug(fmt.Sprintf("kaddb record %v (PO%03d:%d) ready to be tried. seen at %v (%v ago), scheduled at %v", node.Addr, po, cursor, node.Seen, delta, node.After)) log.Debug(fmt.Sprintf("kaddb record %v (PO%03d:%d) ready to be tried. seen at %v (%v ago), scheduled at %v", node.Addr, po, cursor, node.Seen, delta, node.After))
// scheduling next check // scheduling next check
interval = delta * time.Duration(self.connRetryExp) interval = delta * time.Duration(kdb.connRetryExp)
after = time.Now().Add(interval) after = time.Now().Add(interval)
log.Debug(fmt.Sprintf("kaddb record %v (PO%03d:%d) selected as candidate connection %v. seen at %v (%v ago), selectable since %v, retry after %v (in %v)", node.Addr, po, cursor, rounds, node.Seen, delta, node.After, after, interval)) log.Debug(fmt.Sprintf("kaddb record %v (PO%03d:%d) selected as candidate connection %v. seen at %v (%v ago), selectable since %v, retry after %v (in %v)", node.Addr, po, cursor, rounds, node.Seen, delta, node.After, after, interval))
node.After = after node.After = after
found = true found = true
} // ROW } // ROW
self.cursors[po] = cursor kdb.cursors[po] = cursor
self.delete(po, purge) kdb.delete(po, purge)
if found { if found {
return node, need, proxLimit return node, need, proxLimit
} }
@ -251,33 +251,33 @@ func (self *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRe
// deletes the noderecords of a kaddb row corresponding to the indexes // deletes the noderecords of a kaddb row corresponding to the indexes
// caller must hold the dblock // caller must hold the dblock
// the call is unsafe, no index checks // the call is unsafe, no index checks
func (self *KadDb) delete(row int, purge []bool) { func (kdb *KadDb) delete(row int, purge []bool) {
var nodes []*NodeRecord var nodes []*NodeRecord
dbrow := self.Nodes[row] dbrow := kdb.Nodes[row]
for i, del := range purge { for i, del := range purge {
if i == self.cursors[row] { if i == kdb.cursors[row] {
//reset cursor //reset cursor
self.cursors[row] = len(nodes) kdb.cursors[row] = len(nodes)
} }
// delete the entry to be purged // delete the entry to be purged
if del { if del {
delete(self.index, dbrow[i].Addr) delete(kdb.index, dbrow[i].Addr)
continue continue
} }
// otherwise append to new list // otherwise append to new list
nodes = append(nodes, dbrow[i]) nodes = append(nodes, dbrow[i])
} }
self.Nodes[row] = nodes kdb.Nodes[row] = nodes
} }
// save persists kaddb on disk (written to file on path in json format. // save persists kaddb on disk (written to file on path in json format.
func (self *KadDb) save(path string, cb func(*NodeRecord, Node)) error { func (kdb *KadDb) save(path string, cb func(*NodeRecord, Node)) error {
defer self.lock.Unlock() defer kdb.lock.Unlock()
self.lock.Lock() kdb.lock.Lock()
var n int var n int
for _, b := range self.Nodes { for _, b := range kdb.Nodes {
for _, node := range b { for _, node := range b {
n++ n++
node.After = time.Now() node.After = time.Now()
@ -288,7 +288,7 @@ func (self *KadDb) save(path string, cb func(*NodeRecord, Node)) error {
} }
} }
data, err := json.MarshalIndent(self, "", " ") data, err := json.MarshalIndent(kdb, "", " ")
if err != nil { if err != nil {
return err return err
} }
@ -302,9 +302,9 @@ func (self *KadDb) save(path string, cb func(*NodeRecord, Node)) error {
} }
// Load(path) loads the node record database (kaddb) from file on path. // Load(path) loads the node record database (kaddb) from file on path.
func (self *KadDb) load(path string, cb func(*NodeRecord, Node) error) (err error) { func (kdb *KadDb) load(path string, cb func(*NodeRecord, Node) error) (err error) {
defer self.lock.Unlock() defer kdb.lock.Unlock()
self.lock.Lock() kdb.lock.Lock()
var data []byte var data []byte
data, err = ioutil.ReadFile(path) data, err = ioutil.ReadFile(path)
@ -312,13 +312,13 @@ func (self *KadDb) load(path string, cb func(*NodeRecord, Node) error) (err erro
return return
} }
err = json.Unmarshal(data, self) err = json.Unmarshal(data, kdb)
if err != nil { if err != nil {
return return
} }
var n int var n int
var purge []bool var purge []bool
for po, b := range self.Nodes { for po, b := range kdb.Nodes {
purge = make([]bool, len(b)) purge = make([]bool, len(b))
ROW: ROW:
for i, node := range b { for i, node := range b {
@ -333,9 +333,9 @@ func (self *KadDb) load(path string, cb func(*NodeRecord, Node) error) (err erro
if node.After.IsZero() { if node.After.IsZero() {
node.After = time.Now() node.After = time.Now()
} }
self.index[node.Addr] = node kdb.index[node.Addr] = node
} }
self.delete(po, purge) kdb.delete(po, purge)
} }
log.Info(fmt.Sprintf("loaded kaddb with %v nodes from %v", n, path)) log.Info(fmt.Sprintf("loaded kaddb with %v nodes from %v", n, path))
@ -343,8 +343,8 @@ func (self *KadDb) load(path string, cb func(*NodeRecord, Node) error) (err erro
} }
// accessor for KAD offline db count // accessor for KAD offline db count
func (self *KadDb) count() int { func (kdb *KadDb) count() int {
defer self.lock.Unlock() defer kdb.lock.Unlock()
self.lock.Lock() kdb.lock.Lock()
return len(self.index) return len(kdb.index)
} }

58
swarm/network/kademlia/kademlia.go
View file

@ -109,25 +109,25 @@ func New(addr Address, params *KadParams) *Kademlia {
} }
// accessor for KAD base address // accessor for KAD base address
func (self *Kademlia) Addr() Address { func (k *Kademlia) Addr() Address {
return self.addr return self.addr
} }
// accessor for KAD active node count // accessor for KAD active node count
func (self *Kademlia) Count() int { func (k *Kademlia) Count() int {
defer self.lock.Unlock() defer self.lock.Unlock()
self.lock.Lock() self.lock.Lock()
return self.count return self.count
} }
// accessor for KAD active node count // accessor for KAD active node count
func (self *Kademlia) DBCount() int { func (k *Kademlia) DBCount() int {
return self.db.count() return self.db.count()
} }
// On is the entry point called when a new nodes is added // On is the entry point called when a new nodes is added
// unsafe in that node is not checked to be already active node (to be called once) // unsafe in that node is not checked to be already active node (to be called once)
func (self *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error) { func (k *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error) {
log.Debug(fmt.Sprintf("%v", self)) log.Debug(fmt.Sprintf("%v", self))
defer self.lock.Unlock() defer self.lock.Unlock()
self.lock.Lock() self.lock.Lock()
@ -186,7 +186,7 @@ func (self *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error
} }
// Off is the called when a node is taken offline (from the protocol main loop exit) // Off is the called when a node is taken offline (from the protocol main loop exit)
func (self *Kademlia) Off(node Node, cb func(*NodeRecord, Node)) (err error) { func (k *Kademlia) Off(node Node, cb func(*NodeRecord, Node)) (err error) {
self.lock.Lock() self.lock.Lock()
defer self.lock.Unlock() defer self.lock.Unlock()
@ -218,23 +218,23 @@ func (self *Kademlia) Off(node Node, cb func(*NodeRecord, Node)) (err error) {
// 2) the sum of all items are the minimum possible but higher than ProxBinSize // 2) the sum of all items are the minimum possible but higher than ProxBinSize
// adjust Prox (proxLimit and proxSize after an insertion/removal of nodes) // adjust Prox (proxLimit and proxSize after an insertion/removal of nodes)
// caller holds the lock // caller holds the lock
func (self *Kademlia) setProxLimit(r int, on bool) { func (k *Kademlia) setProxLimit(r int, on bool) {
// if the change is outside the core (PO lower) // if the change is outside the core (PO lower)
// and the change does not leave a bucket empty then // and the change does not leave a bucket empty then
// no adjustment needed // no adjustment needed
if r < self.proxLimit && len(self.buckets[r]) > 0 { if r < self.proxLimit && len(k.buckets[r]) > 0 {
return return
} }
// if on=a node was added, then r must be within prox limit so increment cardinality // if on=a node was added, then r must be within prox limit so increment cardinality
if on { if on {
self.proxSize++ self.proxSize++
curr := len(self.buckets[self.proxLimit]) curr := len(k.buckets[self.proxLimit])
// if now core is big enough without the furthest bucket, then contract // if now core is big enough without the furthest bucket, then contract
// this can result in more than one bucket change // this can result in more than one bucket change
for self.proxSize >= self.ProxBinSize+curr && curr > 0 { for self.proxSize >= self.ProxBinSize+curr && curr > 0 {
self.proxSize -= curr self.proxSize -= curr
self.proxLimit++ self.proxLimit++
curr = len(self.buckets[self.proxLimit]) curr = len(k.buckets[self.proxLimit])
log.Trace(fmt.Sprintf("proxbin contraction (size: %v, limit: %v, bin: %v)", self.proxSize, self.proxLimit, r)) log.Trace(fmt.Sprintf("proxbin contraction (size: %v, limit: %v, bin: %v)", self.proxSize, self.proxLimit, r))
} }
@ -245,21 +245,21 @@ func (self *Kademlia) setProxLimit(r int, on bool) {
self.proxSize-- self.proxSize--
} }
// expand core by lowering prox limit until hit zero or cover the empty bucket or reached target cardinality // expand core by lowering prox limit until hit zero or cover the empty bucket or reached target cardinality
for (self.proxSize < self.ProxBinSize || r < self.proxLimit) && for (k.proxSize < self.ProxBinSize || r < self.proxLimit) &&
self.proxLimit > 0 { self.proxLimit > 0 {
// //
self.proxLimit-- self.proxLimit--
self.proxSize += len(self.buckets[self.proxLimit]) self.proxSize += len(k.buckets[self.proxLimit])
log.Trace(fmt.Sprintf("proxbin expansion (size: %v, limit: %v, bin: %v)", self.proxSize, self.proxLimit, r)) log.Trace(fmt.Sprintf("proxbin expansion (size: %v, limit: %v, bin: %v)", self.proxSize, self.proxLimit, r))
} }
} }
/* /*
returns the list of nodes belonging to the same proximity bin FindClosest returns the list of nodes belonging to the same proximity bin
as the target. The most proximate bin will be the union of the bins between as the target. The most proximate bin will be the union of the bins between
proxLimit and MaxProx. proxLimit and MaxProx.
*/ */
func (self *Kademlia) FindClosest(target Address, max int) []Node { func (k *Kademlia) FindClosest(target Address, max int) []Node {
self.lock.Lock() self.lock.Lock()
defer self.lock.Unlock() defer self.lock.Unlock()
@ -289,7 +289,7 @@ func (self *Kademlia) FindClosest(target Address, max int) []Node {
n++ n++
} }
// terminate if index reached the bottom or enough peers > min // terminate if index reached the bottom or enough peers > min
log.Trace(fmt.Sprintf("add %v -> %v (PO%02d, PO%03d)", len(self.buckets[index]), n, index, po)) log.Trace(fmt.Sprintf("add %v -> %v (PO%02d, PO%03d)", len(k.buckets[index]), n, index, po))
if n >= min && (step < 0 || max == 0) { if n >= min && (step < 0 || max == 0) {
break break
} }
@ -304,14 +304,14 @@ func (self *Kademlia) FindClosest(target Address, max int) []Node {
return r.nodes return r.nodes
} }
func (self *Kademlia) Suggest() (*NodeRecord, bool, int) { func (k *Kademlia) Suggest() (*NodeRecord, bool, int) {
defer self.lock.RUnlock() defer self.lock.RUnlock()
self.lock.RLock() self.lock.RLock()
return self.db.findBest(self.BucketSize, func(i int) int { return len(self.buckets[i]) }) return self.db.findBest(k.BucketSize, func(i int) int { return len(k.buckets[i]) })
} }
// adds node records to kaddb (persisted node record db) // Add node records to kaddb (persisted node record db)
func (self *Kademlia) Add(nrs []*NodeRecord) { func (k *Kademlia) Add(nrs []*NodeRecord) {
self.db.add(nrs, self.proximityBin) self.db.add(nrs, self.proximityBin)
} }
@ -369,8 +369,8 @@ a guaranteed constant maximum limit on the number of hops needed to reach one
node from the other. node from the other.
*/ */
func (self *Kademlia) proximityBin(other Address) (ret int) { func (k *Kademlia) proximityBin(other Address) (ret int) {
ret = proximity(self.addr, other) ret = proximity(k.addr, other)
if ret > self.MaxProx { if ret > self.MaxProx {
ret = self.MaxProx ret = self.MaxProx
} }
@ -378,24 +378,24 @@ func (self *Kademlia) proximityBin(other Address) (ret int) {
} }
// provides keyrange for chunk db iteration // provides keyrange for chunk db iteration
func (self *Kademlia) KeyRange(other Address) (start, stop Address) { func (k *Kademlia) KeyRange(other Address) (start, stop Address) {
defer self.lock.RUnlock() defer self.lock.RUnlock()
self.lock.RLock() self.lock.RLock()
return KeyRange(self.addr, other, self.proxLimit) return KeyRange(k.addr, other, self.proxLimit)
} }
// save persists kaddb on disk (written to file on path in json format. // save persists kaddb on disk (written to file on path in json format.
func (self *Kademlia) Save(path string, cb func(*NodeRecord, Node)) error { func (k *Kademlia) Save(path string, cb func(*NodeRecord, Node)) error {
return self.db.save(path, cb) return self.db.save(path, cb)
} }
// Load(path) loads the node record database (kaddb) from file on path. // Load(path) loads the node record database (kaddb) from file on path.
func (self *Kademlia) Load(path string, cb func(*NodeRecord, Node) error) (err error) { func (k *Kademlia) Load(path string, cb func(*NodeRecord, Node) error) (err error) {
return self.db.load(path, cb) return self.db.load(path, cb)
} }
// kademlia table + kaddb table displayed with ascii // kademlia table + kaddb table displayed with ascii
func (self *Kademlia) String() string { func (k *Kademlia) String() string {
defer self.lock.RUnlock() defer self.lock.RUnlock()
self.lock.RLock() self.lock.RLock()
defer self.db.lock.RUnlock() defer self.db.lock.RUnlock()
@ -404,7 +404,7 @@ func (self *Kademlia) String() string {
var rows []string var rows []string
rows = append(rows, "=========================================================================") rows = append(rows, "=========================================================================")
rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), self.addr.String()[:6])) rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), self.addr.String()[:6]))
rows = append(rows, fmt.Sprintf("population: %d (%d), proxLimit: %d, proxSize: %d", self.count, len(self.db.index), self.proxLimit, self.proxSize)) rows = append(rows, fmt.Sprintf("population: %d (%d), proxLimit: %d, proxSize: %d", self.count, len(k.db.index), self.proxLimit, self.proxSize))
rows = append(rows, fmt.Sprintf("MaxProx: %d, ProxBinSize: %d, BucketSize: %d", self.MaxProx, self.ProxBinSize, self.BucketSize)) rows = append(rows, fmt.Sprintf("MaxProx: %d, ProxBinSize: %d, BucketSize: %d", self.MaxProx, self.ProxBinSize, self.BucketSize))
for i, bucket := range self.buckets { for i, bucket := range self.buckets {
@ -425,7 +425,7 @@ func (self *Kademlia) String() string {
for ; k < 4; k++ { for ; k < 4; k++ {
row = append(row, " ") row = append(row, " ")
} }
row = append(row, fmt.Sprintf("| %2d %2d", len(self.db.Nodes[i]), self.db.cursors[i])) row = append(row, fmt.Sprintf("| %2d %2d", len(k.db.Nodes[i]), self.db.cursors[i]))
for j, p := range self.db.Nodes[i] { for j, p := range self.db.Nodes[i] {
row = append(row, p.Addr.String()[:6]) row = append(row, p.Addr.String()[:6])
@ -442,12 +442,12 @@ func (self *Kademlia) String() string {
} }
//We have to build up the array of counters for each index //We have to build up the array of counters for each index
func (self *Kademlia) initMetricsVariables() { func (k *Kademlia) initMetricsVariables() {
//create the arrays //create the arrays
bucketAddIndexCount = make([]metrics.Counter, self.MaxProx+1) bucketAddIndexCount = make([]metrics.Counter, self.MaxProx+1)
bucketRmIndexCount = make([]metrics.Counter, self.MaxProx+1) bucketRmIndexCount = make([]metrics.Counter, self.MaxProx+1)
//at each index create a metrics counter //at each index create a metrics counter
for i := 0; i < (self.KadParams.MaxProx + 1); i++ { for i := 0; i < (k.KadParams.MaxProx + 1); i++ {
bucketAddIndexCount[i] = metrics.NewRegisteredCounter(fmt.Sprintf("network.kademlia.bucket.add.%d.index", i), nil) bucketAddIndexCount[i] = metrics.NewRegisteredCounter(fmt.Sprintf("network.kademlia.bucket.add.%d.index", i), nil)
bucketRmIndexCount[i] = metrics.NewRegisteredCounter(fmt.Sprintf("network.kademlia.bucket.rm.%d.index", i), nil) bucketRmIndexCount[i] = metrics.NewRegisteredCounter(fmt.Sprintf("network.kademlia.bucket.rm.%d.index", i), nil)
} }

22
swarm/network/kademlia/kademlia_test.go
View file

@ -272,31 +272,31 @@ func TestSaveLoad(t *testing.T) {
} }
} }
func (self *Kademlia) proxCheck(t *testing.T) bool { func (k *Kademlia) proxCheck(t *testing.T) bool {
var sum int var sum int
for i, b := range self.buckets { for i, b := range k.buckets {
l := len(b) l := len(b)
// if we are in the high prox multibucket // if we are in the high prox multibucket
if i >= self.proxLimit { if i >= k.proxLimit {
sum += l sum += l
} else if l == 0 { } else if l == 0 {
t.Errorf("bucket %d empty, yet proxLimit is %d\n%v", len(b), self.proxLimit, self) t.Errorf("bucket %d empty, yet proxLimit is %d\n%v", len(b), k.proxLimit, k)
return false return false
} }
} }
// check if merged high prox bucket does not exceed size // check if merged high prox bucket does not exceed size
if sum > 0 { if sum > 0 {
if sum != self.proxSize { if sum != k.proxSize {
t.Errorf("proxSize incorrect, expected %v, got %v", sum, self.proxSize) t.Errorf("proxSize incorrect, expected %v, got %v", sum, k.proxSize)
return false return false
} }
last := len(self.buckets[self.proxLimit]) last := len(k.buckets[k.proxLimit])
if last > 0 && sum >= self.ProxBinSize+last { if last > 0 && sum >= k.ProxBinSize+last {
t.Errorf("proxLimit %v incorrect, redundant non-empty bucket %d added to proxBin with %v (target %v)\n%v", self.proxLimit, last, sum-last, self.ProxBinSize, self) t.Errorf("proxLimit %v incorrect, redundant non-empty bucket %d added to proxBin with %v (target %v)\n%v", k.proxLimit, last, sum-last, k.ProxBinSize, k)
return false return false
} }
if self.proxLimit > 0 && sum < self.ProxBinSize { if k.proxLimit > 0 && sum < k.ProxBinSize {
t.Errorf("proxLimit %v incorrect. proxSize %v is less than target %v, yet there is more peers", self.proxLimit, sum, self.ProxBinSize) t.Errorf("proxLimit %v incorrect. proxSize %v is less than target %v, yet there is more peers", k.proxLimit, sum, k.ProxBinSize)
return false return false
} }
} }