mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-25 06:06:44 +00:00
Merge branch 'swarm' of https://github.com/ethereum/go-ethereum into swarm
Conflicts: internal/web3ext/web3ext.go swarm/api/api.go swarm/api/filesystem.go swarm/api/storage.go swarm/api/testapi.go swarm/examples/album/file-manager.js swarm/examples/album/index.html swarm/examples/album/index.js swarm/network/hive.go swarm/network/kademlia/kademlia.go swarm/network/protocol.go swarm/services/swap/swap.go swarm/swarm.go swarm/test/syncing/00.sh
This commit is contained in:
Ihar Shaduryn
commit
f9e25a3db0
15 changed files with 150 additions and 164 deletions
|
|
@ -35,30 +35,6 @@ web3._extend({
|
|||
property: 'bzz',
|
||||
methods:
|
||||
[
|
||||
new web3._extend.Method({
|
||||
name: 'deposit',
|
||||
call: 'bzz_deposit',
|
||||
params: 1,
|
||||
inputFormatter: [null]
|
||||
}),
|
||||
new web3._extend.Method({
|
||||
name: 'info',
|
||||
call: 'bzz_info',
|
||||
params: 1,
|
||||
inputFormatter: [null]
|
||||
}),
|
||||
new web3._extend.Method({
|
||||
name: 'cash',
|
||||
call: 'bzz_cash',
|
||||
params: 1,
|
||||
inputFormatter: [null]
|
||||
}),
|
||||
new web3._extend.Method({
|
||||
name: 'issue',
|
||||
call: 'bzz_issue',
|
||||
params: 2,
|
||||
inputFormatter: [null, null]
|
||||
}),
|
||||
new web3._extend.Method({
|
||||
name: 'blockNetworkRead',
|
||||
call: 'bzz_blockNetworkRead',
|
||||
|
|
@ -122,6 +98,14 @@ web3._extend({
|
|||
],
|
||||
properties:
|
||||
[
|
||||
new web3._extend.Property({
|
||||
name: 'hive',
|
||||
getter: 'bzz_hive'
|
||||
}),
|
||||
new web3._extend.Property({
|
||||
name: 'info',
|
||||
getter: 'bzz_info',
|
||||
}),
|
||||
]
|
||||
});
|
||||
`
|
||||
|
|
|
|||
|
|
@ -56,7 +56,7 @@ type ErrResolve error
|
|||
// DNS Resolver
|
||||
func (self *Api) Resolve(hostPort string, nameresolver bool) (contentHash storage.Key, err error) {
|
||||
if hashMatcher.MatchString(hostPort) || self.dns == nil {
|
||||
glog.V(logger.Debug).Infof("[BZZ] host is a contentHash: '%v'", hostPort)
|
||||
glog.V(logger.Detail).Infof("[BZZ] host is a contentHash: '%v'", hostPort)
|
||||
return storage.Key(common.Hex2Bytes(hostPort)), nil
|
||||
}
|
||||
if !nameresolver {
|
||||
|
|
@ -66,9 +66,9 @@ func (self *Api) Resolve(hostPort string, nameresolver bool) (contentHash storag
|
|||
contentHash, err = self.dns.Resolve(hostPort)
|
||||
if err != nil {
|
||||
err = ErrResolve(err)
|
||||
glog.V(logger.Debug).Infof("[BZZ] DNS error : %v", err)
|
||||
glog.V(logger.Warn).Infof("[BZZ] DNS error : %v", err)
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] host lookup: %v -> %v", err)
|
||||
glog.V(logger.Detail).Infof("[BZZ] host lookup: %v -> %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -130,21 +130,21 @@ func (self *Api) Get(uri string, nameresolver bool) (reader storage.SectionReade
|
|||
|
||||
trie, err := loadManifest(self.dpa, key)
|
||||
if err != nil {
|
||||
glog.V(logger.Debug).Infof("[BZZ] Swarm: loadManifestTrie error: %v", err)
|
||||
glog.V(logger.Warn).Infof("[BZZ] Swarm: loadManifestTrie error: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
glog.V(logger.Debug).Infof("[BZZ] Swarm: getEntry(%s)", path)
|
||||
glog.V(logger.Detail).Infof("[BZZ] Swarm: getEntry(%s)", path)
|
||||
entry, _ := trie.getEntry(path)
|
||||
if entry != nil {
|
||||
key = common.Hex2Bytes(entry.Hash)
|
||||
status = entry.Status
|
||||
mimeType = entry.ContentType
|
||||
glog.V(logger.Debug).Infof("[BZZ] Swarm: content lookup key: '%v' (%v)", key, mimeType)
|
||||
glog.V(logger.Detail).Infof("[BZZ] Swarm: content lookup key: '%v' (%v)", key, mimeType)
|
||||
reader = self.dpa.Retrieve(key)
|
||||
} else {
|
||||
err = fmt.Errorf("manifest entry for '%s' not found", path)
|
||||
glog.V(logger.Debug).Infof("[BZZ] Swarm: %v", err)
|
||||
glog.V(logger.Warn).Infof("[BZZ] Swarm: %v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
|
|
|||
|
|
@ -113,7 +113,6 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
|
|||
mimeType = "text/css"
|
||||
}
|
||||
list[i].ContentType = mimeType
|
||||
//fmt.Printf("%v %v %v\n", entry.Path, mimeType, filepath.Ext(entry.Path))
|
||||
}
|
||||
}
|
||||
f.Close()
|
||||
|
|
@ -177,7 +176,7 @@ func (self *FileSystem) Download(bzzpath, localpath string) error {
|
|||
|
||||
trie, err := loadManifest(self.api.dpa, key)
|
||||
if err != nil {
|
||||
glog.V(logger.Debug).Infof("[BZZ] fs.Download: loadManifestTrie error: %v", err)
|
||||
glog.V(logger.Warn).Infof("[BZZ] fs.Download: loadManifestTrie error: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,12 +1,5 @@
|
|||
package api
|
||||
|
||||
import (
|
||||
// "github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/logger"
|
||||
"github.com/ethereum/go-ethereum/logger/glog"
|
||||
// "github.com/ethereum/go-ethereum/swarm/storage"
|
||||
)
|
||||
|
||||
type Response struct {
|
||||
MimeType string
|
||||
Status int
|
||||
|
|
@ -47,7 +40,6 @@ func (self *Storage) Get(bzzpath string) (*Response, error) {
|
|||
if int64(size) == expsize {
|
||||
err = nil
|
||||
}
|
||||
glog.V(logger.Detail).Infof("body: %s", body[:size])
|
||||
return &Response{mimeType, status, expsize, string(body[:size])}, err
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,8 +1,6 @@
|
|||
package api
|
||||
|
||||
import (
|
||||
// "fmt"
|
||||
|
||||
"github.com/ethereum/go-ethereum/swarm/network"
|
||||
)
|
||||
|
||||
|
|
@ -26,3 +24,7 @@ func (self *Control) SyncEnabled(on bool) {
|
|||
func (self *Control) SwapEnabled(on bool) {
|
||||
self.hive.SwapEnabled(on)
|
||||
}
|
||||
|
||||
func (self *Control) Hive() string {
|
||||
return self.hive.String()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -209,14 +209,14 @@ func (self *Hive) addPeer(p *peer) {
|
|||
|
||||
// called after peer disconnected
|
||||
func (self *Hive) removePeer(p *peer) {
|
||||
glog.V(logger.Detail).Infof("[BZZ] KΛÐΞMLIΛ hive: bee %v gone offline", p)
|
||||
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: bee %v removed", p)
|
||||
self.kad.Off(p, saveSync)
|
||||
select {
|
||||
case self.more <- true:
|
||||
default:
|
||||
}
|
||||
if self.kad.Count() == 0 {
|
||||
glog.V(logger.Detail).Infof("[BZZ] KΛÐΞMLIΛ hive: empty, all bees gone", p)
|
||||
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: empty, all bees gone")
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -232,7 +232,7 @@ func (self *Hive) getPeers(target storage.Key, max int) (peers []*peer) {
|
|||
|
||||
// disconnects all the peers
|
||||
func (self *Hive) DropAll() {
|
||||
glog.V(logger.Detail).Infof("[BZZ] KΛÐΞMLIΛ hive: dropping all bees")
|
||||
glog.V(logger.Info).Infof("[BZZ] KΛÐΞMLIΛ hive: dropping all bees")
|
||||
for _, node := range self.kad.FindClosest(kademlia.Address{}, 0) {
|
||||
node.Drop()
|
||||
}
|
||||
|
|
@ -284,7 +284,10 @@ func (self *peer) LastActive() time.Time {
|
|||
// reads the serialised form of sync state persisted as the 'Meta' attribute
|
||||
// and sets the decoded syncState on the online node
|
||||
func loadSync(record *kademlia.NodeRecord, node kademlia.Node) error {
|
||||
if p, ok := node.(*peer); ok {
|
||||
p, ok := node.(*peer)
|
||||
if !ok {
|
||||
return fmt.Errorf("invalid type")
|
||||
}
|
||||
if record.Meta == nil {
|
||||
glog.V(logger.Debug).Infof("no sync state for node record %v setting default", record)
|
||||
p.syncState = &syncState{DbSyncState: &storage.DbSyncState{}}
|
||||
|
|
@ -294,11 +297,9 @@ func loadSync(record *kademlia.NodeRecord, node kademlia.Node) error {
|
|||
if err != nil {
|
||||
return fmt.Errorf("error decoding kddb record meta info into a sync state: %v", err)
|
||||
}
|
||||
glog.V(logger.Debug).Infof("sync state for node record %v: %s -> %v", record, string(*(record.Meta)), state)
|
||||
glog.V(logger.Detail).Infof("sync state for node record %v read from Meta: %s", record, string(*(record.Meta)))
|
||||
p.syncState = state
|
||||
return err
|
||||
}
|
||||
return fmt.Errorf("invalid type")
|
||||
}
|
||||
|
||||
// callback when saving a sync state
|
||||
|
|
@ -309,8 +310,7 @@ func saveSync(record *kademlia.NodeRecord, node kademlia.Node) {
|
|||
glog.V(logger.Warn).Infof("error saving sync state for %v: %v", node, err)
|
||||
return
|
||||
}
|
||||
glog.V(logger.Warn).Infof("saving sync state for %v: %s", node, string(*meta))
|
||||
|
||||
glog.V(logger.Detail).Infof("saved sync state for %v: %s", node, string(*meta))
|
||||
record.Meta = meta
|
||||
}
|
||||
}
|
||||
|
|
@ -348,3 +348,7 @@ func (self *Hive) peers(req *retrieveRequestMsgData) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (self *Hive) String() string {
|
||||
return self.kad.String()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -130,7 +130,9 @@ func (self *KadDb) add(nrs []*NodeRecord, proximityBin func(Address) int) {
|
|||
n++
|
||||
}
|
||||
}
|
||||
glog.V(logger.Detail).Infof("[KΛÐ]: received %d node records, added %d new", len(nrs), n)
|
||||
if n > 0 {
|
||||
glog.V(logger.Debug).Infof("[KΛÐ]: %d/%d node records (new/known)", n, len(nrs))
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
@ -145,7 +147,8 @@ We check for missing online nodes in the buckets for 1 upto Max BucketSize round
|
|||
On each round we proceed from the low to high proximity order buckets.
|
||||
If the number of active nodes (=connected peers) is < rounds, then start looking
|
||||
for a known candidate. To determine if there is a candidate to recommend the
|
||||
node record database row corresponding to the bucket is checked.a
|
||||
node record database row corresponding to the bucket is checked.
|
||||
|
||||
If the row cursor is on position i, the ith element in the row is chosen.
|
||||
If the record is scheduled not to be retried before NOW, the next element is taken.
|
||||
If the record is scheduled can be retried, it is set as checked, scheduled for
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ const (
|
|||
var (
|
||||
purgeInterval = 42 * time.Hour
|
||||
initialRetryInterval = 42 * 100 * time.Millisecond
|
||||
maxIdleInterval = 42 * 10 * time.Second
|
||||
)
|
||||
|
||||
type KadParams struct {
|
||||
|
|
@ -112,13 +113,13 @@ func (self *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error
|
|||
// setting the node on the record, set it checked (for connectivity)
|
||||
record.node = node
|
||||
|
||||
glog.V(logger.Info).Infof("[KΛÐ]: add node record %v with node %v", record, node)
|
||||
if cb != nil {
|
||||
err = cb(record, node)
|
||||
glog.V(logger.Info).Infof("[KΛÐ]: cb(%v, %v) ->%v", record, node, err)
|
||||
glog.V(logger.Detail).Infof("[KΛÐ]: cb(%v, %v) ->%v", record, node, err)
|
||||
if err != nil {
|
||||
return fmt.Errorf("node %v not added: %v", node.Addr(), err)
|
||||
return fmt.Errorf("unable to add node %v, callback error: %v", node.Addr(), err)
|
||||
}
|
||||
glog.V(logger.Info).Infof("[KΛÐ]: add node record %v with node %v", record, node)
|
||||
}
|
||||
record.connected = true
|
||||
|
||||
|
|
@ -126,12 +127,18 @@ func (self *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error
|
|||
bucket := self.buckets[index]
|
||||
// if bucket is full insertion replaces the worst node
|
||||
// TODO: give priority to peers with active traffic
|
||||
if worst, pos := bucket.insert(node); worst != nil {
|
||||
glog.V(logger.Info).Infof("[KΛÐ]: replace node %v (%d) with node %v\n%v", worst, pos, node, self)
|
||||
return
|
||||
replaced, err := bucket.insert(node)
|
||||
if err != nil {
|
||||
glog.V(logger.Debug).Infof("[KΛÐ]: node %v not needed: %v", node, err)
|
||||
return err
|
||||
// no prox adjustment needed
|
||||
// do not change count
|
||||
}
|
||||
if replaced != nil {
|
||||
glog.V(logger.Debug).Infof("[KΛÐ]: node %v replaced by %v ", replaced, node)
|
||||
return
|
||||
}
|
||||
// new node added
|
||||
glog.V(logger.Info).Infof("[KΛÐ]: add node %v to table", node)
|
||||
self.count++
|
||||
self.setProxLimit(index, false)
|
||||
|
|
@ -327,17 +334,13 @@ func (h *nodesByDistance) push(node Node, max int) {
|
|||
// insert adds a peer to a bucket either by appending to existing items if
|
||||
// bucket length does not exceed bucketSize, or by replacing the worst
|
||||
// Node in the bucket
|
||||
func (self *bucket) insert(node Node) (dropped Node, pos int) {
|
||||
func (self *bucket) insert(node Node) (replaced Node, err error) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
if len(self.nodes) >= self.size { // >= allows us to add peers beyond the bucketsize limitation
|
||||
dropped, pos = self.worstNode()
|
||||
if dropped != nil {
|
||||
self.nodes[pos] = node
|
||||
glog.V(logger.Info).Infof("[KΛÐ] dropping node %v (%d)", dropped, pos)
|
||||
dropped.Drop()
|
||||
return
|
||||
}
|
||||
// dev p2p kicks out nodes idle for > 30 s, so here we never replace nodes if
|
||||
// bucket is full
|
||||
return nil, fmt.Errorf("bucket full")
|
||||
}
|
||||
self.nodes = append(self.nodes, node)
|
||||
return
|
||||
|
|
@ -349,20 +352,6 @@ func (self *bucket) length(node Node) int {
|
|||
return len(self.nodes)
|
||||
}
|
||||
|
||||
// worst expunges the single worst node in a row, where worst entry is the node
|
||||
// that has been inactive for the longests time
|
||||
func (self *bucket) worstNode() (node Node, pos int) {
|
||||
var oldest time.Time
|
||||
for p, n := range self.nodes {
|
||||
if (oldest == time.Time{}) || !oldest.Before(n.LastActive()) {
|
||||
oldest = n.LastActive()
|
||||
node = n
|
||||
pos = p
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
/*
|
||||
Taking the proximity order relative to a fix point x classifies the points in
|
||||
the space (n byte long byte sequences) into bins. Items in each are at
|
||||
|
|
@ -412,6 +401,7 @@ func (self *Kademlia) String() string {
|
|||
|
||||
var rows []string
|
||||
rows = append(rows, "=========================================================================")
|
||||
rows = append(rows, fmt.Sprintf("KΛÐΞMLIΛ hive: queen's address: %v, population: %d (%d)", self.addr, self.Count(), self.DBCount()))
|
||||
rows = append(rows, fmt.Sprintf("%v : MaxProx: %d, ProxBinSize: %d, BucketSize: %d, proxLimit: %d, proxSize: %d", time.Now(), self.MaxProx, self.ProxBinSize, self.BucketSize, self.proxLimit, self.proxSize))
|
||||
|
||||
for i, b := range self.buckets {
|
||||
|
|
|
|||
|
|
@ -77,7 +77,11 @@ func (self storeRequestMsgData) String() string {
|
|||
} else {
|
||||
from = self.from.Addr().String()
|
||||
}
|
||||
return fmt.Sprintf("from: %v, Key: %v; ID: %v, requestTimeout: %v, storageTimeout: %v, SData %x", from, self.Key, self.Id, self.requestTimeout, self.storageTimeout, self.SData[:10])
|
||||
end := len(self.SData)
|
||||
if len(self.SData) > 10 {
|
||||
end = 10
|
||||
}
|
||||
return fmt.Sprintf("from: %v, Key: %v; ID: %v, requestTimeout: %v, storageTimeout: %v, SData %x", from, self.Key, self.Id, self.requestTimeout, self.storageTimeout, self.SData[:end])
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import (
|
|||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/accounts/abi/bind"
|
||||
"github.com/ethereum/go-ethereum/errs"
|
||||
"github.com/ethereum/go-ethereum/logger"
|
||||
"github.com/ethereum/go-ethereum/logger/glog"
|
||||
|
|
@ -75,6 +76,7 @@ type bzz struct {
|
|||
peer *p2p.Peer // the p2p peer object
|
||||
rw p2p.MsgReadWriter // messageReadWriter to send messages to
|
||||
errors *errs.Errors // errors table
|
||||
backend bind.Backend
|
||||
|
||||
swap *swap.Swap // swap instance for the peer connection
|
||||
swapParams *bzzswap.SwapParams // swap settings both local and remote
|
||||
|
|
@ -106,7 +108,7 @@ on each peer connection
|
|||
The Run function of the Bzz protocol class creates a bzz instance
|
||||
which will represent the peer for the swarm hive and all peer-aware components
|
||||
*/
|
||||
func Bzz(cloud StorageHandler, hive *Hive, dbaccess *DbAccess, sp *bzzswap.SwapParams, sy *SyncParams) (p2p.Protocol, error) {
|
||||
func Bzz(cloud StorageHandler, backend bind.Backend, hive *Hive, dbaccess *DbAccess, sp *bzzswap.SwapParams, sy *SyncParams) (p2p.Protocol, error) {
|
||||
|
||||
// a single global request db is created for all peer connections
|
||||
// this is to persist delivery backlog and aid syncronisation
|
||||
|
|
@ -120,7 +122,7 @@ func Bzz(cloud StorageHandler, hive *Hive, dbaccess *DbAccess, sp *bzzswap.SwapP
|
|||
Version: Version,
|
||||
Length: ProtocolLength,
|
||||
Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
|
||||
return run(requestDb, cloud, hive, dbaccess, sp, sy, p, rw)
|
||||
return run(requestDb, cloud, backend, hive, dbaccess, sp, sy, p, rw)
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
|
@ -137,10 +139,11 @@ the main protocol loop that
|
|||
* whenever the loop terminates, the peer will disconnect with Subprotocol error
|
||||
* whenever handlers return an error the loop terminates
|
||||
*/
|
||||
func run(requestDb *storage.LDBDatabase, depo StorageHandler, hive *Hive, dbaccess *DbAccess, sp *bzzswap.SwapParams, sy *SyncParams, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
|
||||
func run(requestDb *storage.LDBDatabase, depo StorageHandler, backend bind.Backend, hive *Hive, dbaccess *DbAccess, sp *bzzswap.SwapParams, sy *SyncParams, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
|
||||
|
||||
self := &bzz{
|
||||
storage: depo,
|
||||
backend: backend,
|
||||
hive: hive,
|
||||
dbAccess: dbaccess,
|
||||
requestDb: requestDb,
|
||||
|
|
@ -187,7 +190,7 @@ func run(requestDb *storage.LDBDatabase, depo StorageHandler, hive *Hive, dbacce
|
|||
return
|
||||
}
|
||||
|
||||
// may need to implement protocol drop only? don't want to kick off the peer
|
||||
// TODO: may need to implement protocol drop only? don't want to kick off the peer
|
||||
// if they are useful for other protocols
|
||||
func (self *bzz) Drop() {
|
||||
self.peer.Disconnect(p2p.DiscSubprotocolError)
|
||||
|
|
@ -218,7 +221,7 @@ func (self *bzz) handle() error {
|
|||
// store requests are dispatched to netStore
|
||||
var req storeRequestMsgData
|
||||
if err := msg.Decode(&req); err != nil {
|
||||
return self.protoError(ErrDecode, "msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<- %v: %v", msg, err)
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] incoming store request: %s", req.String())
|
||||
// swap accounting is done within forwarding
|
||||
|
|
@ -228,7 +231,7 @@ func (self *bzz) handle() error {
|
|||
// retrieve Requests are dispatched to netStore
|
||||
var req retrieveRequestMsgData
|
||||
if err := msg.Decode(&req); err != nil {
|
||||
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<- %v: %v", msg, err)
|
||||
}
|
||||
req.from = &peer{bzz: self}
|
||||
// if request is lookup and not to be delivered
|
||||
|
|
@ -248,30 +251,30 @@ func (self *bzz) handle() error {
|
|||
// dispatches new peer data to the hive that adds them to KADDB
|
||||
var req peersMsgData
|
||||
if err := msg.Decode(&req); err != nil {
|
||||
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<- %v: %v", msg, err)
|
||||
}
|
||||
req.from = &peer{bzz: self}
|
||||
glog.V(logger.Debug).Infof("[BZZ] incoming peer addresses: %v", req)
|
||||
glog.V(logger.Debug).Infof("[BZZ] <- peer addresses: %v", req)
|
||||
self.hive.HandlePeersMsg(&req, &peer{bzz: self})
|
||||
|
||||
case syncRequestMsg:
|
||||
var req syncRequestMsgData
|
||||
if err := msg.Decode(&req); err != nil {
|
||||
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<- %v: %v", msg, err)
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] sync request received: %v", req)
|
||||
glog.V(logger.Debug).Infof("[BZZ] <- sync request: %v", req)
|
||||
self.sync(req.SyncState)
|
||||
|
||||
case unsyncedKeysMsg:
|
||||
// coming from parent node offering
|
||||
var req unsyncedKeysMsgData
|
||||
if err := msg.Decode(&req); err != nil {
|
||||
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<- %v: %v", msg, err)
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] incoming unsynced keys msg: %s", req.String())
|
||||
glog.V(logger.Debug).Infof("[BZZ] <- unsynced keys : %s", req.String())
|
||||
err := self.storage.HandleUnsyncedKeysMsg(&req, &peer{bzz: self})
|
||||
if err != nil {
|
||||
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<- %v: %v", msg, err)
|
||||
}
|
||||
|
||||
case deliveryRequestMsg:
|
||||
|
|
@ -279,12 +282,12 @@ func (self *bzz) handle() error {
|
|||
// also relays the last synced state to the source
|
||||
var req deliveryRequestMsgData
|
||||
if err := msg.Decode(&req); err != nil {
|
||||
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<-msg %v: %v", msg, err)
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] incoming delivery request: %s", req.String())
|
||||
glog.V(logger.Debug).Infof("[BZZ] <- delivery request: %s", req.String())
|
||||
err := self.storage.HandleDeliveryRequestMsg(&req, &peer{bzz: self})
|
||||
if err != nil {
|
||||
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<- %v: %v", msg, err)
|
||||
}
|
||||
|
||||
case paymentMsg:
|
||||
|
|
@ -292,9 +295,9 @@ func (self *bzz) handle() error {
|
|||
if self.swapEnabled {
|
||||
var req paymentMsgData
|
||||
if err := msg.Decode(&req); err != nil {
|
||||
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, "<- %v: %v", msg, err)
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] incoming payment: %s", req.String())
|
||||
glog.V(logger.Debug).Infof("[BZZ] <- payment: %s", req.String())
|
||||
self.swap.Receive(int(req.Units), req.Promise)
|
||||
}
|
||||
|
||||
|
|
@ -340,7 +343,7 @@ func (self *bzz) handleStatus() (err error) {
|
|||
|
||||
var status statusMsgData
|
||||
if err := msg.Decode(&status); err != nil {
|
||||
return self.protoError(ErrDecode, "msg %v: %v", msg, err)
|
||||
return self.protoError(ErrDecode, " %v: %v", msg, err)
|
||||
}
|
||||
|
||||
if status.NetworkId != NetworkId {
|
||||
|
|
@ -356,7 +359,7 @@ func (self *bzz) handleStatus() (err error) {
|
|||
|
||||
if self.swapEnabled {
|
||||
// set remote profile for accounting
|
||||
self.swap, err = bzzswap.NewSwap(self.swapParams, status.Swap, self)
|
||||
self.swap, err = bzzswap.NewSwap(self.swapParams, status.Swap, self.backend, self)
|
||||
if err != nil {
|
||||
return self.protoError(ErrSwap, "%v", err)
|
||||
}
|
||||
|
|
@ -385,7 +388,7 @@ func (self *bzz) sync(state *syncState) error {
|
|||
// an explicitly received nil syncstate disables syncronisation
|
||||
if state == nil {
|
||||
self.syncEnabled = false
|
||||
glog.V(logger.Info).Infof("[BZZ] syncronisation disabled for peer %v", self)
|
||||
glog.V(logger.Warn).Infof("[BZZ] syncronisation disabled for peer %v", self)
|
||||
state = &syncState{DbSyncState: &storage.DbSyncState{}, Synced: true}
|
||||
} else {
|
||||
state.synced = make(chan bool)
|
||||
|
|
@ -394,7 +397,7 @@ func (self *bzz) sync(state *syncState) error {
|
|||
state.Start = storage.Key(start[:])
|
||||
state.Stop = storage.Key(stop[:])
|
||||
}
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncronisation requested by peer %v at state %v", self, state)
|
||||
glog.V(logger.Debug).Infof("[BZZ] syncronisation requested by peer %v at state %v", self, state)
|
||||
}
|
||||
var err error
|
||||
self.syncer, err = newSyncer(
|
||||
|
|
@ -454,11 +457,11 @@ func (self *bzz) store(req *storeRequestMsgData) error {
|
|||
func (self *bzz) syncRequest() error {
|
||||
req := &syncRequestMsgData{}
|
||||
if self.hive.syncEnabled {
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncronisation request to peer %v at state %v", self, self.syncState)
|
||||
glog.V(logger.Debug).Infof("[BZZ] syncronisation request to peer %v at state %v", self, self.syncState)
|
||||
req.SyncState = self.syncState
|
||||
}
|
||||
if self.syncState == nil {
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncronisation disabled for peer %v at state %v", self, self.syncState)
|
||||
glog.V(logger.Warn).Infof("[BZZ] syncronisation disabled for peer %v at state %v", self, self.syncState)
|
||||
}
|
||||
return self.send(syncRequestMsg, req)
|
||||
}
|
||||
|
|
@ -513,7 +516,8 @@ func (self *bzz) send(msg uint64, data interface{}) error {
|
|||
if self.hive.blockWrite {
|
||||
return fmt.Errorf("network write blocked")
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] -> %v: %v (%T) to %v", msg, data, data, self)
|
||||
// self.messages = append(self.messages, "")
|
||||
glog.V(logger.Detail).Infof("[BZZ] -> %v: %v (%T) to %v", msg, data, data, self)
|
||||
err := p2p.Send(self.rw, msg, data)
|
||||
if err != nil {
|
||||
self.Drop()
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ func newSyncDb(db *storage.LDBDatabase, key storage.Key, priority uint, bufferSi
|
|||
batch: make(chan chan int),
|
||||
dbBatchSize: dbBatchSize,
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] syncDb[peer: %v, priority: %v] - initialised", key.Log(), priority)
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncDb[peer: %v, priority: %v] - initialised", key.Log(), priority)
|
||||
|
||||
// starts the main forever loop reading from buffer
|
||||
go syncdb.bufferRead(deliver)
|
||||
|
|
@ -159,14 +159,13 @@ LOOP:
|
|||
}
|
||||
self.dbTotal++
|
||||
self.total++
|
||||
// otherwise break after select
|
||||
|
||||
// otherwise break after selec
|
||||
case dbSize = <-self.batch:
|
||||
// explicit request for batch
|
||||
if inBatch == 0 && quit != nil {
|
||||
// there was no writes since the last batch so db depleted
|
||||
// switch to buffer mode
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncDb[%v] empty db: switching to buffer", self.priority)
|
||||
glog.V(logger.Debug).Infof("[BZZ] syncDb[%v] empty db: switching to buffer", self.priority)
|
||||
db = nil
|
||||
buffer = self.buffer
|
||||
dbSize <- 0 // indicates to 'caller' that batch has been written
|
||||
|
|
@ -175,9 +174,9 @@ LOOP:
|
|||
}
|
||||
binary.BigEndian.PutUint64(counterValue, counter)
|
||||
batch.Put(self.counterKey, counterValue)
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncDb[%v] write batch %v/%v - %x - %x", self.priority, inBatch, counter, self.counterKey, counterValue)
|
||||
glog.V(logger.Debug).Infof("[BZZ] syncDb[%v] write batch %v/%v - %x - %x", self.priority, inBatch, counter, self.counterKey, counterValue)
|
||||
batch = self.writeSyncBatch(batch)
|
||||
dbSize <- inBatch
|
||||
dbSize <- inBatch // indicates to 'caller' that batch has been written
|
||||
inBatch = 0
|
||||
continue LOOP
|
||||
|
||||
|
|
@ -222,7 +221,7 @@ LOOP:
|
|||
func (self *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch {
|
||||
err := self.db.Write(batch)
|
||||
if err != nil {
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncDb[%v] saving batch to db failed: %v", self.priority, err)
|
||||
glog.V(logger.Warn).Infof("[BZZ] syncDb[%v] saving batch to db failed: %v", self.priority, err)
|
||||
return batch
|
||||
}
|
||||
return new(leveldb.Batch)
|
||||
|
|
@ -275,7 +274,11 @@ func (self *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}
|
|||
// this could be called before all cnt items sent out
|
||||
// so that loop is not blocking while delivering
|
||||
// only relevant if cnt is large
|
||||
self.batch <- batchSizes
|
||||
select {
|
||||
case self.batch <- batchSizes:
|
||||
case <-self.quit:
|
||||
return
|
||||
}
|
||||
// wait for the write to finish and get the item count in the next batch
|
||||
cnt = <-batchSizes
|
||||
batches++
|
||||
|
|
@ -318,7 +321,7 @@ func (self *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}
|
|||
n++
|
||||
total++
|
||||
}
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncDb[%v/%v] - db session closed, batches: %v, total: %v, session total from db: %v/%v", self.key.Log(), self.priority, batches, total, self.dbTotal, self.total)
|
||||
glog.V(logger.Debug).Infof("[BZZ] syncDb[%v/%v] - db session closed, batches: %v, total: %v, session total from db: %v/%v", self.key.Log(), self.priority, batches, total, self.dbTotal, self.total)
|
||||
self.db.Write(del) // this could be async called only when db is idle
|
||||
it.Release()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -316,7 +316,7 @@ func (self *syncer) syncState(state *syncState) {
|
|||
// stop quits both request processor and saves the request cache to disk
|
||||
func (self *syncer) stop() {
|
||||
close(self.quit)
|
||||
glog.V(logger.Debug).Infof("[BZZ] syncer[%v]: save sync request db backlog", self.key.Log())
|
||||
glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: stopand save sync request db backlog", self.key.Log())
|
||||
for _, db := range self.queues {
|
||||
db.stop()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ type PayProfile struct {
|
|||
publicKey *ecdsa.PublicKey
|
||||
owner common.Address
|
||||
chbook *chequebook.Chequebook
|
||||
backend bind.Backend
|
||||
// backend bind.Backend
|
||||
lock sync.RWMutex
|
||||
}
|
||||
|
||||
|
|
@ -99,18 +99,18 @@ func DefaultSwapParams(contract common.Address, prvkey *ecdsa.PrivateKey) *SwapP
|
|||
// n < 0 called when receiving chunks = receiving delivery responses
|
||||
// OR receiving cheques.
|
||||
|
||||
func NewSwap(local *SwapParams, remote *SwapProfile, proto swap.Protocol) (self *swap.Swap, err error) {
|
||||
func NewSwap(local *SwapParams, remote *SwapProfile, backend bind.Backend, proto swap.Protocol) (self *swap.Swap, err error) {
|
||||
|
||||
// check if remote chequebook is valid
|
||||
// insolvent chequebooks suicide so will signal as invalid
|
||||
// TODO: monitoring a chequebooks events
|
||||
var in *chequebook.Inbox
|
||||
err = bind.Validate(remote.Contract, contract.ContractDeployedCode, local.backend)
|
||||
err = bind.Validate(remote.Contract, contract.ContractDeployedCode, backend)
|
||||
if err != nil {
|
||||
glog.V(logger.Info).Infof("[BZZ] SWAP invalid contract %v for peer %v: %v)", remote.Contract.Hex()[:8], proto, err)
|
||||
} else {
|
||||
// remote contract valid, create inbox
|
||||
in, err = chequebook.NewInbox(local.privateKey, remote.Contract, local.Beneficiary, crypto.ToECDSAPub(common.FromHex(remote.PublicKey)), local.backend)
|
||||
in, err = chequebook.NewInbox(local.privateKey, remote.Contract, local.Beneficiary, crypto.ToECDSAPub(common.FromHex(remote.PublicKey)), backend)
|
||||
if err != nil {
|
||||
glog.V(logger.Warn).Infof("[BZZ] SWAP unable to set up inbox for chequebook contract %v for peer %v: %v)", remote.Contract.Hex()[:8], proto, err)
|
||||
}
|
||||
|
|
@ -118,7 +118,7 @@ func NewSwap(local *SwapParams, remote *SwapProfile, proto swap.Protocol) (self
|
|||
|
||||
// cheque if local chequebook contract is valid
|
||||
var out *chequebook.Outbox
|
||||
err = bind.Validate(local.Contract, contract.ContractDeployedCode, local.backend)
|
||||
err = bind.Validate(local.Contract, contract.ContractDeployedCode, backend)
|
||||
if err != nil {
|
||||
glog.V(logger.Warn).Infof("[BZZ] SWAP unable to set up outbox for peer %v: chequebook contract (owner: %v): %v)", proto, local.owner.Hex(), err)
|
||||
} else {
|
||||
|
|
@ -159,17 +159,18 @@ func (self *SwapParams) Chequebook() *chequebook.Chequebook {
|
|||
return self.chbook
|
||||
}
|
||||
|
||||
func (self *SwapParams) Backend() bind.Backend {
|
||||
return self.backend
|
||||
}
|
||||
// func (self *SwapParams) Backend() bind.Backend {
|
||||
// return self.backend
|
||||
// }
|
||||
|
||||
func (self *SwapParams) PrivateKey() *ecdsa.PrivateKey {
|
||||
return self.privateKey
|
||||
}
|
||||
|
||||
func (self *SwapParams) PublicKey() *ecdsa.PublicKey {
|
||||
return self.publicKey
|
||||
}
|
||||
// func (self *SwapParams) PublicKey() *ecdsa.PublicKey {
|
||||
// return self.publicKey
|
||||
// }
|
||||
|
||||
func (self *SwapParams) SetKey(prvkey *ecdsa.PrivateKey) {
|
||||
self.privateKey = prvkey
|
||||
self.publicKey = &prvkey.PublicKey
|
||||
|
|
@ -182,10 +183,9 @@ func (self *SwapParams) SetChequebook(path string, backend bind.Backend) (done c
|
|||
self.lock.Lock()
|
||||
var valid bool
|
||||
done = make(chan bool)
|
||||
self.backend = backend
|
||||
err = bind.Validate(self.Contract, contract.ContractDeployedCode, backend)
|
||||
if err != nil {
|
||||
go self.deployChequebook(path, done)
|
||||
go self.deployChequebook(path, backend, done)
|
||||
} else {
|
||||
valid = true
|
||||
go func() {
|
||||
|
|
@ -210,14 +210,14 @@ func deploy(deployTransactor *bind.TransactOpts, backend bind.ContractBackend) (
|
|||
|
||||
// prvKey *ecdsa.PrivateKey, amount *big.Int,
|
||||
|
||||
func (self *SwapParams) deployChequebook(path string, done chan bool) {
|
||||
func (self *SwapParams) deployChequebook(path string, backend bind.Backend, done chan bool) {
|
||||
owner := crypto.PubkeyToAddress(*(self.publicKey))
|
||||
glog.V(logger.Info).Infof("[BZZ] SWAP Deploying new chequebook (owner: %v)", owner.Hex())
|
||||
var valid bool
|
||||
|
||||
transactOpts := bind.NewKeyedTransactor(self.privateKey)
|
||||
transactOpts.Value = self.AutoDepositBuffer
|
||||
contract, err := bind.Deploy(deploy, contract.ContractDeployedCode, transactOpts, bind.DefaultDeployOptions(), self.backend)
|
||||
contract, err := bind.Deploy(deploy, contract.ContractDeployedCode, transactOpts, bind.DefaultDeployOptions(), backend)
|
||||
if err != nil {
|
||||
glog.V(logger.Error).Infof("[BZZ] SWAP unable to deploy new chequebook: %v", err)
|
||||
return
|
||||
|
|
@ -226,7 +226,7 @@ func (self *SwapParams) deployChequebook(path string, done chan bool) {
|
|||
// need to save config at this point
|
||||
self.lock.Lock()
|
||||
self.Contract = contract
|
||||
err = self.newChequebookFromContract(path, self.backend)
|
||||
err = self.newChequebookFromContract(path, backend)
|
||||
self.lock.Unlock()
|
||||
if err != nil {
|
||||
glog.V(logger.Warn).Infof("[BZZ] SWAP error initialising cheque book (owner: %v): %v", owner.Hex(), err)
|
||||
|
|
|
|||
|
|
@ -46,8 +46,8 @@ type Swarm struct {
|
|||
cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
|
||||
hive *network.Hive // the logistic manager
|
||||
client *httpclient.HTTPClient // bzz capable light http client
|
||||
contractBackend bind.ContractBackend // abigen contract Backend
|
||||
backend bind.Backend // simple blockchain Backend
|
||||
privateKey *ecdsa.PrivateKey
|
||||
swapEnabled bool
|
||||
}
|
||||
|
||||
|
|
@ -60,8 +60,8 @@ type SwarmAPI struct {
|
|||
func (self *Swarm) API() *SwarmAPI {
|
||||
return &SwarmAPI{
|
||||
Api: self.api,
|
||||
Backend: self.config.Swap.Backend(),
|
||||
PrvKey: self.config.Swap.PrivateKey(),
|
||||
Backend: self.backend,
|
||||
PrvKey: self.privateKey,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -90,6 +90,7 @@ func NewSwarm(ctx *node.ServiceContext, config *api.Config, swapEnabled, syncEna
|
|||
ethereum: ethereum,
|
||||
swapEnabled: swapEnabled,
|
||||
client: ethereum.HTTPClient(),
|
||||
privateKey: config.Swap.PrivateKey(),
|
||||
}
|
||||
glog.V(logger.Debug).Infof("[BZZ] Setting up Swarm service components")
|
||||
|
||||
|
|
@ -138,7 +139,7 @@ func NewSwarm(ctx *node.ServiceContext, config *api.Config, swapEnabled, syncEna
|
|||
self.backend = eth.NewContractBackend(ethereum)
|
||||
|
||||
// set up high level api
|
||||
transactOpts := bind.NewKeyedTransactor(self.config.Swap.PrivateKey())
|
||||
transactOpts := bind.NewKeyedTransactor(self.privateKey)
|
||||
// backend := ethereum.ContractBackend()
|
||||
self.dns = ens.NewENS(transactOpts, ENSContractAddr, self.backend)
|
||||
glog.V(logger.Debug).Infof("[BZZ] -> Swarm Domain Name Registrar")
|
||||
|
|
@ -226,7 +227,7 @@ func (self *Swarm) Stop() error {
|
|||
|
||||
// implements the node.Service interface
|
||||
func (self *Swarm) Protocols() []p2p.Protocol {
|
||||
proto, err := network.Bzz(self.depo, self.hive, self.dbAccess, self.config.Swap, self.config.SyncParams)
|
||||
proto, err := network.Bzz(self.depo, self.backend, self.hive, self.dbAccess, self.config.Swap, self.config.SyncParams)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ echo " two nodes that sync (no swap and do not have any funds)"
|
|||
echo " can be in sync content with each other"
|
||||
|
||||
dir=`dirname $0`
|
||||
source $dir/../../cmd/swarm/tesst.sh
|
||||
source $dir/../../cmd/swarm/test.sh
|
||||
|
||||
mkdir -p /tmp/swarm-test-files
|
||||
FILE_00=/tmp/swarm-test-files/00
|
||||
|
|
|
|||
Loading…
Reference in a new issue