diff --git a/internal/web3ext/web3ext.go b/internal/web3ext/web3ext.go index 6865c7d72f..c217443214 100644 --- a/internal/web3ext/web3ext.go +++ b/internal/web3ext/web3ext.go @@ -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', + }), ] }); ` diff --git a/swarm/api/api.go b/swarm/api/api.go index e1ab01318e..883f226cf2 100644 --- a/swarm/api/api.go +++ b/swarm/api/api.go @@ -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 } diff --git a/swarm/api/filesystem.go b/swarm/api/filesystem.go index 3bd4a318d3..002bb95273 100644 --- a/swarm/api/filesystem.go +++ b/swarm/api/filesystem.go @@ -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 } diff --git a/swarm/api/storage.go b/swarm/api/storage.go index 82b8f12fba..97b5590fdf 100644 --- a/swarm/api/storage.go +++ b/swarm/api/storage.go @@ -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 } diff --git a/swarm/api/testapi.go b/swarm/api/testapi.go index 339b623089..7e7229bbc8 100644 --- a/swarm/api/testapi.go +++ b/swarm/api/testapi.go @@ -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() +} diff --git a/swarm/network/hive.go b/swarm/network/hive.go index fd8012350a..9dccef1400 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -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,21 +284,22 @@ 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 { - 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{}} - return nil - } - state, err := decodeSync(record.Meta) - 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) - p.syncState = state - return err + p, ok := node.(*peer) + if !ok { + return fmt.Errorf("invalid type") } - 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{}} + return nil + } + state, err := decodeSync(record.Meta) + if err != nil { + return fmt.Errorf("error decoding kddb record meta info into a sync state: %v", err) + } + glog.V(logger.Detail).Infof("sync state for node record %v read from Meta: %s", record, string(*(record.Meta))) + p.syncState = state + return err } // 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() +} diff --git a/swarm/network/kademlia/kaddb.go b/swarm/network/kademlia/kaddb.go index e45a9dc48e..d7c6a5be13 100644 --- a/swarm/network/kademlia/kaddb.go +++ b/swarm/network/kademlia/kaddb.go @@ -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 diff --git a/swarm/network/kademlia/kademlia.go b/swarm/network/kademlia/kademlia.go index d7308c7bc3..602db1445d 100644 --- a/swarm/network/kademlia/kademlia.go +++ b/swarm/network/kademlia/kademlia.go @@ -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 { diff --git a/swarm/network/messages.go b/swarm/network/messages.go index 31bdf19b54..0d169b1b0a 100644 --- a/swarm/network/messages.go +++ b/swarm/network/messages.go @@ -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]) } /* diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index c1eac50ab6..6f27f68b26 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -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() diff --git a/swarm/network/syncdb.go b/swarm/network/syncdb.go index 39d5f1bffc..f91ed9a50d 100644 --- a/swarm/network/syncdb.go +++ b/swarm/network/syncdb.go @@ -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() } diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index 9a117be683..e540e236f5 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -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() } diff --git a/swarm/services/swap/swap.go b/swarm/services/swap/swap.go index 366b84c47b..816e7263c0 100644 --- a/swarm/services/swap/swap.go +++ b/swarm/services/swap/swap.go @@ -57,8 +57,8 @@ type PayProfile struct { publicKey *ecdsa.PublicKey owner common.Address chbook *chequebook.Chequebook - backend bind.Backend - lock sync.RWMutex + // backend bind.Backend + lock sync.RWMutex } func DefaultSwapParams(contract common.Address, prvkey *ecdsa.PrivateKey) *SwapParams { @@ -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) diff --git a/swarm/swarm.go b/swarm/swarm.go index 078495d73a..6be1159aa7 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -35,20 +35,20 @@ var ENSContractAddr = common.HexToAddress("0x504cdf3992d8f81a4182bd7b24e270d3a28 // the swarm stack type Swarm struct { - ethereum *eth.Ethereum - config *api.Config // swarm configuration - api *api.Api // high level api layer (fs/manifest) - dns api.Resolver // DNS registrar - dbAccess *network.DbAccess // access to local chunk db iterator and storage counter - storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends - dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support - depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage - 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 - swapEnabled bool + ethereum *eth.Ethereum + config *api.Config // swarm configuration + api *api.Api // high level api layer (fs/manifest) + dns api.Resolver // DNS registrar + dbAccess *network.DbAccess // access to local chunk db iterator and storage counter + storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends + dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support + depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage + cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud) + hive *network.Hive // the logistic manager + client *httpclient.HTTPClient // bzz capable light http client + backend bind.Backend // simple blockchain Backend + privateKey *ecdsa.PrivateKey + swapEnabled bool } type SwarmAPI struct { @@ -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 } diff --git a/swarm/test/syncing/00.sh b/swarm/test/syncing/00.sh index 2eb1cd140c..a793fea916 100644 --- a/swarm/test/syncing/00.sh +++ b/swarm/test/syncing/00.sh @@ -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