diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 7139ec67ad..8f9d720838 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -77,7 +77,7 @@ type HiveParams struct { *kademlia.KadParams } -//create default params +// NewDefaultHiveParams creates default params func NewDefaultHiveParams() *HiveParams { kad := kademlia.NewDefaultKadParams() // kad.BucketSize = bucketSize @@ -124,7 +124,7 @@ func (hive *Hive) BlockNetworkWrite(on bool) { hive.blockWrite = on } -// public accessor to the hive base address +// Addr is the public accessor to the hive base address func (hive *Hive) Addr() kademlia.Address { return hive.addr } @@ -196,57 +196,57 @@ func (hive *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee // by writing to hive.more until Kademlia Table is saturated // wake state is toggled by writing to hive.toggle // it restarts if the table becomes non-full again due to disconnections -func (self *Hive) keepAlive() { - alarm := time.NewTicker(time.Duration(self.callInterval)).C +func (hive *Hive) keepAlive() { + alarm := time.NewTicker(time.Duration(hive.callInterval)).C for { - peersNumGauge.Update(int64(self.kad.Count())) + peersNumGauge.Update(int64(hive.kad.Count())) select { case <-alarm: - if self.kad.DBCount() > 0 { + if hive.kad.DBCount() > 0 { select { - case self.more <- true: + case hive.more <- true: log.Debug(fmt.Sprintf("buzz wakeup")) default: } } - case need := <-self.toggle: + case need := <-hive.toggle: if alarm == nil && need { - alarm = time.NewTicker(time.Duration(self.callInterval)).C + alarm = time.NewTicker(time.Duration(hive.callInterval)).C } if alarm != nil && !need { alarm = nil } - case <-self.quit: + case <-hive.quit: return } } } -func (self *Hive) Stop() error { +func (hive *Hive) Stop() error { // closing toggle channel quits the updateloop - close(self.quit) - return self.kad.Save(self.path, saveSync) + close(hive.quit) + return hive.kad.Save(hive.path, saveSync) } // called at the end of a successful protocol handshake -func (self *Hive) addPeer(p *peer) error { +func (hive *Hive) addPeer(p *peer) error { addPeerCounter.Inc(1) defer func() { select { - case self.more <- true: + case hive.more <- true: default: } }() log.Trace(fmt.Sprintf("hi new bee %v", p)) - err := self.kad.On(p, loadSync) + err := hive.kad.On(p, loadSync) if err != nil { return err } - // self lookup (can be encoded as nil/zero key since peers addr known) + no id () + // hive lookup (can be encoded as nil/zero key since peers addr known) + no id () // the most common way of saying hi in bzz is initiation of gossip // let me know about anyone new from my hood , here is the storageradius - // to send the 6 byte self lookup + // to send the 6 byte hive lookup // we do not record as request or forward it, just reply with peers p.retrieve(&retrieveRequestMsgData{}) log.Trace(fmt.Sprintf("'whatsup wheresdaparty' sent to %v", p)) @@ -255,33 +255,33 @@ func (self *Hive) addPeer(p *peer) error { } // called after peer disconnected -func (self *Hive) removePeer(p *peer) { +func (hive *Hive) removePeer(p *peer) { removePeerCounter.Inc(1) log.Debug(fmt.Sprintf("bee %v removed", p)) - self.kad.Off(p, saveSync) + hive.kad.Off(p, saveSync) select { - case self.more <- true: + case hive.more <- true: default: } - if self.kad.Count() == 0 { + if hive.kad.Count() == 0 { log.Debug(fmt.Sprintf("empty, all bees gone")) } } // Retrieve a list of live peers that are closer to target than us -func (self *Hive) getPeers(target storage.Key, max int) (peers []*peer) { +func (hive *Hive) getPeers(target storage.Key, max int) (peers []*peer) { var addr kademlia.Address copy(addr[:], target[:]) - for _, node := range self.kad.FindClosest(addr, max) { + for _, node := range hive.kad.FindClosest(addr, max) { peers = append(peers, node.(*peer)) } return } -// disconnects all the peers -func (self *Hive) DropAll() { +// DropAll disconnects all the peers +func (hive *Hive) DropAll() { log.Info(fmt.Sprintf("dropping all bees")) - for _, node := range self.kad.FindClosest(kademlia.Address{}, 0) { + for _, node := range hive.kad.FindClosest(kademlia.Address{}, 0) { node.Drop() } } @@ -298,10 +298,10 @@ func newNodeRecord(addr *peerAddr) *kademlia.NodeRecord { } } -// called by the protocol when receiving peerset (for target address) +// HandlePeersMsg called by the protocol when receiving peerset (for target address) // peersMsgData is converted to a slice of NodeRecords for Kademlia // this is to store all thats needed -func (self *Hive) HandlePeersMsg(req *peersMsgData, from *peer) { +func (hive *Hive) HandlePeersMsg(req *peersMsgData, from *peer) { var nrs []*kademlia.NodeRecord for _, p := range req.Peers { if err := netutil.CheckRelayIP(from.remoteAddr.IP, p.IP); err != nil { @@ -310,7 +310,7 @@ func (self *Hive) HandlePeersMsg(req *peersMsgData, from *peer) { } nrs = append(nrs, newNodeRecord(p)) } - self.kad.Add(nrs) + hive.kad.Add(nrs) } // peer wraps the protocol instance to represent a connected peer @@ -320,17 +320,17 @@ type peer struct { } // protocol instance implements kademlia.Node interface (embedded peer) -func (self *peer) Addr() kademlia.Address { - return self.remoteAddr.Addr +func (p *peer) Addr() kademlia.Address { + return p.remoteAddr.Addr } -func (self *peer) Url() string { - return self.remoteAddr.String() +func (p *peer) Url() string { + return p.remoteAddr.String() } // TODO take into account traffic -func (self *peer) LastActive() time.Time { - return self.lastActive +func (p *peer) LastActive() time.Time { + return p.lastActive } // reads the serialised form of sync state persisted as the 'Meta' attribute @@ -370,19 +370,19 @@ func saveSync(record *kademlia.NodeRecord, node kademlia.Node) { // the immediate response to a retrieve request, // sends relevant peer data given by the kademlia hive to the requester // TODO: remember peers sent for duration of the session, only new peers sent -func (self *Hive) peers(req *retrieveRequestMsgData) { +func (hive *Hive) peers(req *retrieveRequestMsgData) { if req != nil { var addrs []*peerAddr if req.timeout == nil || time.Now().Before(*(req.timeout)) { key := req.Key - // self lookup from remote peer + // hive lookup from remote peer if storage.IsZeroKey(key) { addr := req.from.Addr() key = storage.Key(addr[:]) req.Key = nil } // get peer addresses from hive - for _, peer := range self.getPeers(key, int(req.MaxPeers)) { + for _, peer := range hive.getPeers(key, int(req.MaxPeers)) { addrs = append(addrs, peer.remoteAddr) } log.Debug(fmt.Sprintf("Hive sending %d peer addresses to %v. req.Id: %v, req.Key: %v", len(addrs), req.from, req.Id, req.Key.Log())) @@ -398,6 +398,6 @@ func (self *Hive) peers(req *retrieveRequestMsgData) { } } -func (self *Hive) String() string { - return self.kad.String() +func (hive *Hive) String() string { + return hive.kad.String() } diff --git a/swarm/network/messages.go b/swarm/network/messages.go index d920def959..3869272250 100644 --- a/swarm/network/messages.go +++ b/swarm/network/messages.go @@ -64,8 +64,8 @@ type statusMsgData struct { NetworkId uint64 } -func (self *statusMsgData) String() string { - return fmt.Sprintf("Status: Version: %v, ID: %v, Addr: %v, Swap: %v, NetworkId: %v", self.Version, self.ID, self.Addr, self.Swap, self.NetworkId) +func (data *statusMsgData) String() string { + return fmt.Sprintf("Status: Version: %v, ID: %v, Addr: %v, Swap: %v, NetworkId: %v", data.Version, data.ID, data.Addr, data.Swap, data.NetworkId) } /* @@ -86,18 +86,18 @@ type storeRequestMsgData struct { from *peer // [not serialised] protocol registers the requester } -func (self storeRequestMsgData) String() string { +func (data storeRequestMsgData) String() string { var from string - if self.from == nil { - from = "self" + if data.from == nil { + from = "data" } else { - from = self.from.Addr().String() + from = data.from.Addr().String() } - end := len(self.SData) - if len(self.SData) > 10 { + end := len(data.SData) + if len(data.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]) + return fmt.Sprintf("from: %v, Key: %v; ID: %v, requestTimeout: %v, storageTimeout: %v, SData %x", from, data.Key, data.Id, data.requestTimeout, data.storageTimeout, data.SData[:end]) } /* @@ -118,7 +118,7 @@ prompting a peers response but not launching a search. Lookup requests are meant to be used to bootstrap kademlia tables. In the special case that the key is the zero value as well, the remote peer's -address is assumed (the message is to be handled as a self lookup request). +address is assumed (the message is to be handled as a data lookup request). The response is a PeersMsg with the peers in the kademlia proximity bin corresponding to the address. */ @@ -133,40 +133,40 @@ type retrieveRequestMsgData struct { from *peer // } -func (self *retrieveRequestMsgData) String() string { +func (data *retrieveRequestMsgData) String() string { var from string - if self.from == nil { + if data.from == nil { from = "ourselves" } else { - from = self.from.Addr().String() + from = data.from.Addr().String() } var target []byte - if len(self.Key) > 3 { - target = self.Key[:4] + if len(data.Key) > 3 { + target = data.Key[:4] } - return fmt.Sprintf("from: %v, Key: %x; ID: %v, MaxSize: %v, MaxPeers: %d", from, target, self.Id, self.MaxSize, self.MaxPeers) + return fmt.Sprintf("from: %v, Key: %x; ID: %v, MaxSize: %v, MaxPeers: %d", from, target, data.Id, data.MaxSize, data.MaxPeers) } // lookups are encoded by missing request ID -func (self *retrieveRequestMsgData) isLookup() bool { - return self.Id == 0 +func (data *retrieveRequestMsgData) isLookup() bool { + return data.Id == 0 } // sets timeout fields -func (self *retrieveRequestMsgData) setTimeout(t *time.Time) { - self.timeout = t +func (data *retrieveRequestMsgData) setTimeout(t *time.Time) { + data.timeout = t if t != nil { - self.Timeout = uint64(t.UnixNano()) + data.Timeout = uint64(t.UnixNano()) } else { - self.Timeout = 0 + data.Timeout = 0 } } -func (self *retrieveRequestMsgData) getTimeout() (t *time.Time) { - if self.Timeout > 0 && self.timeout == nil { - timeout := time.Unix(int64(self.Timeout), 0) +func (data *retrieveRequestMsgData) getTimeout() (t *time.Time) { + if data.Timeout > 0 && data.timeout == nil { + timeout := time.Unix(int64(data.Timeout), 0) t = &timeout - self.timeout = t + data.timeout = t } return } @@ -180,10 +180,10 @@ type peerAddr struct { } // peerAddr pretty prints as enode -func (self *peerAddr) String() string { +func (addr *peerAddr) String() string { var nodeid discover.NodeID - copy(nodeid[:], self.ID) - return discover.NewNode(nodeid, self.IP, 0, self.Port).String() + copy(nodeid[:], addr.ID) + return discover.NewNode(nodeid, addr.IP, 0, addr.Port).String() } /* @@ -199,7 +199,7 @@ the timeout or not. NodeID serves as the owner of payment contracts and signer of proofs of transfer. The Key is the target (if response to a retrieval request) or missing (zero value) -peers address (hash of NodeID) if retrieval request was a self lookup. +peers address (hash of NodeID) if retrieval request was a data lookup. Peers message is requested by retrieval requests with a missing or zero value request ID */ @@ -213,26 +213,26 @@ type peersMsgData struct { } // peers msg pretty printer -func (self *peersMsgData) String() string { +func (data *peersMsgData) String() string { var from string - if self.from == nil { + if data.from == nil { from = "ourselves" } else { - from = self.from.Addr().String() + from = data.from.Addr().String() } var target []byte - if len(self.Key) > 3 { - target = self.Key[:4] + if len(data.Key) > 3 { + target = data.Key[:4] } - return fmt.Sprintf("from: %v, Key: %x; ID: %v, Peers: %v", from, target, self.Id, self.Peers) + return fmt.Sprintf("from: %v, Key: %x; ID: %v, Peers: %v", from, target, data.Id, data.Peers) } -func (self *peersMsgData) setTimeout(t *time.Time) { - self.timeout = t +func (data *peersMsgData) setTimeout(t *time.Time) { + data.timeout = t if t != nil { - self.Timeout = uint64(t.UnixNano()) + data.Timeout = uint64(t.UnixNano()) } else { - self.Timeout = 0 + data.Timeout = 0 } } @@ -248,8 +248,8 @@ type syncRequestMsgData struct { SyncState *syncState `rlp:"nil"` } -func (self *syncRequestMsgData) String() string { - return fmt.Sprintf("%v", self.SyncState) +func (data *syncRequestMsgData) String() string { + return fmt.Sprintf("%v", data.SyncState) } /* @@ -265,8 +265,8 @@ type deliveryRequestMsgData struct { Deliver []*syncRequest } -func (self *deliveryRequestMsgData) String() string { - return fmt.Sprintf("sync request for new chunks\ndelivery request for %v chunks", len(self.Deliver)) +func (data *deliveryRequestMsgData) String() string { + return fmt.Sprintf("sync request for new chunks\ndelivery request for %v chunks", len(data.Deliver)) } /* @@ -287,8 +287,8 @@ type unsyncedKeysMsgData struct { State *syncState } -func (self *unsyncedKeysMsgData) String() string { - return fmt.Sprintf("sync: keys of %d new chunks (state %v) => synced: %v", len(self.Unsynced), self.State, self.State.Synced) +func (data *unsyncedKeysMsgData) String() string { + return fmt.Sprintf("sync: keys of %d new chunks (state %v) => synced: %v", len(data.Unsynced), data.State, data.State.Synced) } /* @@ -303,6 +303,6 @@ type paymentMsgData struct { Promise *chequebook.Cheque // payment with cheque } -func (self *paymentMsgData) String() string { - return fmt.Sprintf("payment for %d units: %v", self.Units, self.Promise) +func (data *paymentMsgData) String() string { + return fmt.Sprintf("payment for %d units: %v", data.Units, data.Promise) } diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 1cbe00a973..4dadbf6d61 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -193,13 +193,13 @@ func run(requestDb *storage.LDBDatabase, depo StorageHandler, backend chequebook // 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) +func (bzz *bzz) Drop() { + bzz.peer.Disconnect(p2p.DiscSubprotocolError) } // one cycle of the main forever loop that handles and dispatches incoming messages -func (self *bzz) handle() error { - msg, err := self.rw.ReadMsg() +func (bzz *bzz) handle() error { + msg, err := bzz.rw.ReadMsg() log.Debug(fmt.Sprintf("<- %v", msg)) if err != nil { return err @@ -232,7 +232,7 @@ func (self *bzz) handle() error { self.lastActive = time.Now() log.Trace(fmt.Sprintf("incoming store request: %s", req.String())) // swap accounting is done within forwarding - self.storage.HandleStoreRequestMsg(&req, &peer{bzz: self}) + bzz.storage.HandleStoreRequestMsg(&req, &peer{bzz: bzz}) case retrieveRequestMsg: // retrieve Requests are dispatched to netStore @@ -241,7 +241,7 @@ func (self *bzz) handle() error { if err := msg.Decode(&req); err != nil { return fmt.Errorf("<- %v: %v", msg, err) } - req.from = &peer{bzz: self} + req.from = &peer{bzz: bzz} // if request is lookup and not to be delivered if req.isLookup() { log.Trace(fmt.Sprintf("self lookup for %v: responding with peers only...", req.from)) @@ -249,10 +249,10 @@ func (self *bzz) handle() error { return fmt.Errorf("protocol handler: req.Key == nil || req.Timeout == nil") } else { // swap accounting is done within netStore - self.storage.HandleRetrieveRequestMsg(&req, &peer{bzz: self}) + bzz.storage.HandleRetrieveRequestMsg(&req, &peer{bzz: bzz}) } // direct response with peers, TODO: sort this out - self.hive.peers(&req) + bzz.hive.peers(&req) case peersMsg: // response to lookups and immediate response to retrieve requests @@ -262,9 +262,9 @@ func (self *bzz) handle() error { if err := msg.Decode(&req); err != nil { return fmt.Errorf("<- %v: %v", msg, err) } - req.from = &peer{bzz: self} + req.from = &peer{bzz: bzz} log.Trace(fmt.Sprintf("<- peer addresses: %v", req)) - self.hive.HandlePeersMsg(&req, &peer{bzz: self}) + bzz.hive.HandlePeersMsg(&req, &peer{bzz: bzz}) case syncRequestMsg: syncRequestMsgCounter.Inc(1) @@ -273,8 +273,8 @@ func (self *bzz) handle() error { return fmt.Errorf("<- %v: %v", msg, err) } log.Debug(fmt.Sprintf("<- sync request: %v", req)) - self.lastActive = time.Now() - self.sync(req.SyncState) + bzz.lastActive = time.Now() + bzz.sync(req.SyncState) case unsyncedKeysMsg: // coming from parent node offering @@ -284,8 +284,8 @@ func (self *bzz) handle() error { return fmt.Errorf("<- %v: %v", msg, err) } log.Debug(fmt.Sprintf("<- unsynced keys : %s", req.String())) - err := self.storage.HandleUnsyncedKeysMsg(&req, &peer{bzz: self}) - self.lastActive = time.Now() + err := bzz.storage.HandleUnsyncedKeysMsg(&req, &peer{bzz: bzz}) + bzz.lastActive = time.Now() if err != nil { return fmt.Errorf("<- %v: %v", msg, err) } @@ -299,8 +299,8 @@ func (self *bzz) handle() error { return fmt.Errorf("<-msg %v: %v", msg, err) } log.Debug(fmt.Sprintf("<- delivery request: %s", req.String())) - err := self.storage.HandleDeliveryRequestMsg(&req, &peer{bzz: self}) - self.lastActive = time.Now() + err := bzz.storage.HandleDeliveryRequestMsg(&req, &peer{bzz: bzz}) + bzz.lastActive = time.Now() if err != nil { return fmt.Errorf("<- %v: %v", msg, err) } @@ -308,13 +308,13 @@ func (self *bzz) handle() error { case paymentMsg: // swap protocol message for payment, Units paid for, Cheque paid with paymentMsgCounter.Inc(1) - if self.swapEnabled { + if bzz.swapEnabled { var req paymentMsgData if err := msg.Decode(&req); err != nil { return fmt.Errorf("<- %v: %v", msg, err) } log.Debug(fmt.Sprintf("<- payment: %s", req.String())) - self.swap.Receive(int(req.Units), req.Promise) + bzz.swap.Receive(int(req.Units), req.Promise) } default: @@ -325,27 +325,27 @@ func (self *bzz) handle() error { return nil } -func (self *bzz) handleStatus() (err error) { +func (bzz *bzz) handleStatus() (err error) { handshake := &statusMsgData{ Version: uint64(Version), ID: "honey", - Addr: self.selfAddr(), - NetworkId: self.NetworkId, + Addr: bzz.selfAddr(), + NetworkId: bzz.NetworkId, Swap: &bzzswap.SwapProfile{ - Profile: self.swapParams.Profile, - PayProfile: self.swapParams.PayProfile, + Profile: bzz.swapParams.Profile, + PayProfile: bzz.swapParams.PayProfile, }, } - err = p2p.Send(self.rw, statusMsg, handshake) + err = p2p.Send(bzz.rw, statusMsg, handshake) if err != nil { return err } // read and handle remote status var msg p2p.Msg - msg, err = self.rw.ReadMsg() + msg, err = bzz.rw.ReadMsg() if err != nil { return err } @@ -365,52 +365,52 @@ func (self *bzz) handleStatus() (err error) { return fmt.Errorf("<- %v: %v", msg, err) } - if status.NetworkId != self.NetworkId { - return fmt.Errorf("network id mismatch: %d (!= %d)", status.NetworkId, self.NetworkId) + if status.NetworkId != bzz.NetworkId { + return fmt.Errorf("network id mismatch: %d (!= %d)", status.NetworkId, bzz.NetworkId) } if Version != status.Version { return fmt.Errorf("protocol version mismatch: %d (!= %d)", status.Version, Version) } - self.remoteAddr = self.peerAddr(status.Addr) - log.Trace(fmt.Sprintf("self: advertised IP: %v, peer advertised: %v, local address: %v\npeer: advertised IP: %v, remote address: %v\n", self.selfAddr(), self.remoteAddr, self.peer.LocalAddr(), status.Addr.IP, self.peer.RemoteAddr())) + bzz.remoteAddr = bzz.peerAddr(status.Addr) + log.Trace(fmt.Sprintf("bzz: advertised IP: %v, peer advertised: %v, local address: %v\npeer: advertised IP: %v, remote address: %v\n", bzz.selfAddr(), bzz.remoteAddr, bzz.peer.LocalAddr(), status.Addr.IP, bzz.peer.RemoteAddr())) - if self.swapEnabled { + if bzz.swapEnabled { // set remote profile for accounting - self.swap, err = bzzswap.NewSwap(self.swapParams, status.Swap, self.backend, self) + bzz.swap, err = bzzswap.NewSwap(bzz.swapParams, status.Swap, bzz.backend, bzz) if err != nil { return err } } - log.Info(fmt.Sprintf("Peer %08x is capable (%d/%d)", self.remoteAddr.Addr[:4], status.Version, status.NetworkId)) - err = self.hive.addPeer(&peer{bzz: self}) + log.Info(fmt.Sprintf("Peer %08x is capable (%d/%d)", bzz.remoteAddr.Addr[:4], status.Version, status.NetworkId)) + err = bzz.hive.addPeer(&peer{bzz: bzz}) if err != nil { return err } // hive sets syncstate so sync should start after node added - log.Info(fmt.Sprintf("syncronisation request sent with %v", self.syncState)) - self.syncRequest() + log.Info(fmt.Sprintf("syncronisation request sent with %v", bzz.syncState)) + bzz.syncRequest() return nil } -func (self *bzz) sync(state *syncState) error { +func (bzz *bzz) sync(state *syncState) error { // syncer setup - if self.syncer != nil { + if bzz.syncer != nil { return errors.New("sync request can only be sent once") } - cnt := self.dbAccess.counter() - remoteaddr := self.remoteAddr.Addr - start, stop := self.hive.kad.KeyRange(remoteaddr) + cnt := bzz.dbAccess.counter() + remoteaddr := bzz.remoteAddr.Addr + start, stop := bzz.hive.kad.KeyRange(remoteaddr) // an explicitly received nil syncstate disables syncronisation if state == nil { - self.syncEnabled = false - log.Warn(fmt.Sprintf("syncronisation disabled for peer %v", self)) + bzz.syncEnabled = false + log.Warn(fmt.Sprintf("syncronisation disabled for peer %v", bzz)) state = &syncState{DbSyncState: &storage.DbSyncState{}, Synced: true} } else { state.synced = make(chan bool) @@ -419,31 +419,31 @@ func (self *bzz) sync(state *syncState) error { state.Start = storage.Key(start[:]) state.Stop = storage.Key(stop[:]) } - log.Debug(fmt.Sprintf("syncronisation requested by peer %v at state %v", self, state)) + log.Debug(fmt.Sprintf("syncronisation requested by peer %v at state %v", bzz, state)) } var err error - self.syncer, err = newSyncer( - self.requestDb, + bzz.syncer, err = newSyncer( + bzz.requestDb, storage.Key(remoteaddr[:]), - self.dbAccess, - self.unsyncedKeys, self.store, - self.syncParams, state, func() bool { return self.syncEnabled }, + bzz.dbAccess, + bzz.unsyncedKeys, bzz.store, + bzz.syncParams, state, func() bool { return bzz.syncEnabled }, ) if err != nil { return nil } - log.Trace(fmt.Sprintf("syncer set for peer %v", self)) + log.Trace(fmt.Sprintf("syncer set for peer %v", bzz)) return nil } -func (self *bzz) String() string { - return self.remoteAddr.String() +func (bzz *bzz) String() string { + return bzz.remoteAddr.String() } // repair reported address if IP missing -func (self *bzz) peerAddr(base *peerAddr) *peerAddr { +func (bzz *bzz) peerAddr(base *peerAddr) *peerAddr { if base.IP.IsUnspecified() { - host, _, _ := net.SplitHostPort(self.peer.RemoteAddr().String()) + host, _, _ := net.SplitHostPort(bzz.peer.RemoteAddr().String()) base.IP = net.ParseIP(host) } return base @@ -452,12 +452,12 @@ func (self *bzz) peerAddr(base *peerAddr) *peerAddr { // returns self advertised node connection info (listening address w enodes) // IP will get repaired on the other end if missing // or resolved via ID by discovery at dialout -func (self *bzz) selfAddr() *peerAddr { - id := self.hive.id - host, port, _ := net.SplitHostPort(self.hive.listenAddr()) +func (bzz *bzz) selfAddr() *peerAddr { + id := bzz.hive.id + host, port, _ := net.SplitHostPort(bzz.hive.listenAddr()) intport, _ := strconv.Atoi(port) addr := &peerAddr{ - Addr: self.hive.addr, + Addr: bzz.hive.addr, ID: id[:], IP: net.ParseIP(host), Port: uint16(intport), @@ -467,68 +467,68 @@ func (self *bzz) selfAddr() *peerAddr { // outgoing messages // send retrieveRequestMsg -func (self *bzz) retrieve(req *retrieveRequestMsgData) error { - return self.send(retrieveRequestMsg, req) +func (bzz *bzz) retrieve(req *retrieveRequestMsgData) error { + return bzz.send(retrieveRequestMsg, req) } // send storeRequestMsg -func (self *bzz) store(req *storeRequestMsgData) error { - return self.send(storeRequestMsg, req) +func (bzz *bzz) store(req *storeRequestMsgData) error { + return bzz.send(storeRequestMsg, req) } -func (self *bzz) syncRequest() error { +func (bzz *bzz) syncRequest() error { req := &syncRequestMsgData{} - if self.hive.syncEnabled { - log.Debug(fmt.Sprintf("syncronisation request to peer %v at state %v", self, self.syncState)) - req.SyncState = self.syncState + if bzz.hive.syncEnabled { + log.Debug(fmt.Sprintf("syncronisation request to peer %v at state %v", bzz, bzz.syncState)) + req.SyncState = bzz.syncState } - if self.syncState == nil { - log.Warn(fmt.Sprintf("syncronisation disabled for peer %v at state %v", self, self.syncState)) + if bzz.syncState == nil { + log.Warn(fmt.Sprintf("syncronisation disabled for peer %v at state %v", bzz, bzz.syncState)) } - return self.send(syncRequestMsg, req) + return bzz.send(syncRequestMsg, req) } // queue storeRequestMsg in request db -func (self *bzz) deliveryRequest(reqs []*syncRequest) error { +func (bzz *bzz) deliveryRequest(reqs []*syncRequest) error { req := &deliveryRequestMsgData{ Deliver: reqs, } - return self.send(deliveryRequestMsg, req) + return bzz.send(deliveryRequestMsg, req) } // batch of syncRequests to send off -func (self *bzz) unsyncedKeys(reqs []*syncRequest, state *syncState) error { +func (bzz *bzz) unsyncedKeys(reqs []*syncRequest, state *syncState) error { req := &unsyncedKeysMsgData{ Unsynced: reqs, State: state, } - return self.send(unsyncedKeysMsg, req) + return bzz.send(unsyncedKeysMsg, req) } // send paymentMsg -func (self *bzz) Pay(units int, promise swap.Promise) { +func (bzz *bzz) Pay(units int, promise swap.Promise) { req := &paymentMsgData{uint(units), promise.(*chequebook.Cheque)} - self.payment(req) + bzz.payment(req) } // send paymentMsg -func (self *bzz) payment(req *paymentMsgData) error { - return self.send(paymentMsg, req) +func (bzz *bzz) payment(req *paymentMsgData) error { + return bzz.send(paymentMsg, req) } // sends peersMsg -func (self *bzz) peers(req *peersMsgData) error { - return self.send(peersMsg, req) +func (bzz *bzz) peers(req *peersMsgData) error { + return bzz.send(peersMsg, req) } -func (self *bzz) send(msg uint64, data interface{}) error { - if self.hive.blockWrite { +func (bzz *bzz) send(msg uint64, data interface{}) error { + if bzz.hive.blockWrite { return fmt.Errorf("network write blocked") } - log.Trace(fmt.Sprintf("-> %v: %v (%T) to %v", msg, data, data, self)) - err := p2p.Send(self.rw, msg, data) + log.Trace(fmt.Sprintf("-> %v: %v (%T) to %v", msg, data, data, bzz)) + err := p2p.Send(bzz.rw, msg, data) if err != nil { - self.Drop() + bzz.Drop() } return err } diff --git a/swarm/network/syncdb.go b/swarm/network/syncdb.go index 88b4b68dd0..2ef2bbf986 100644 --- a/swarm/network/syncdb.go +++ b/swarm/network/syncdb.go @@ -108,7 +108,7 @@ It is automatically started when syncdb is initialised. It saves the buffer to db upon receiving quit signal. syncDb#stop() */ -func (self *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) { +func (db *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) { var buffer, db chan interface{} // channels representing the two read modes var more bool var req interface{} @@ -116,18 +116,18 @@ func (self *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) { var inBatch, inDb int batch := new(leveldb.Batch) var dbSize chan int - quit := self.quit + quit := db.quit counterValue := make([]byte, 8) // counter is used for keeping the items in order, persisted to db // start counter where db was at, 0 if not found - data, err := self.db.Get(self.counterKey) + data, err := db.db.Get(db.counterKey) var counter uint64 if err == nil { counter = binary.BigEndian.Uint64(data) - log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter read from db at %v", self.key.Log(), self.priority, counter)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter read from db at %v", db.key.Log(), db.priority, counter)) } else { - log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter starts at %v", self.key.Log(), self.priority, counter)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter starts at %v", db.key.Log(), db.priority, counter)) } LOOP: @@ -139,26 +139,26 @@ LOOP: // deliver request : this is blocking on network write so // it is passed the quit channel as argument, so that it returns // if syncdb is stopped. In this case we need to save the item to the db - more = deliver(req, self.quit) + more = deliver(req, db.quit) if !more { - log.Debug(fmt.Sprintf("syncDb[%v/%v] quit: switching to db. session tally (db/total): %v/%v", self.key.Log(), self.priority, self.dbTotal, self.total)) + log.Debug(fmt.Sprintf("syncDb[%v/%v] quit: switching to db. session tally (db/total): %v/%v", db.key.Log(), db.priority, db.dbTotal, db.total)) // received quit signal, save request currently waiting delivery // by switching to db mode and closing the buffer buffer = nil - db = self.buffer + db = db.buffer close(db) quit = nil // needs to block the quit case in select break // break from select, this item will be written to the db } - self.total++ - log.Trace(fmt.Sprintf("syncDb[%v/%v] deliver (db/total): %v/%v", self.key.Log(), self.priority, self.dbTotal, self.total)) + db.total++ + log.Trace(fmt.Sprintf("syncDb[%v/%v] deliver (db/total): %v/%v", db.key.Log(), db.priority, db.dbTotal, db.total)) // by the time deliver returns, there were new writes to the buffer // if buffer contention is detected, switch to db mode which drains // the buffer so no process will block on pushing store requests if len(buffer) == cap(buffer) { - log.Debug(fmt.Sprintf("syncDb[%v/%v] buffer full %v: switching to db. session tally (db/total): %v/%v", self.key.Log(), self.priority, cap(buffer), self.dbTotal, self.total)) + log.Debug(fmt.Sprintf("syncDb[%v/%v] buffer full %v: switching to db. session tally (db/total): %v/%v", db.key.Log(), db.priority, cap(buffer), db.dbTotal, db.total)) buffer = nil - db = self.buffer + db = db.buffer } continue LOOP @@ -167,30 +167,30 @@ LOOP: if !more { // only if quit is called, saved all the buffer binary.BigEndian.PutUint64(counterValue, counter) - batch.Put(self.counterKey, counterValue) // persist counter in batch - self.writeSyncBatch(batch) // save batch - log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save current batch to db", self.key.Log(), self.priority)) + batch.Put(db.counterKey, counterValue) // persist counter in batch + db.writeSyncBatch(batch) // save batch + log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save current batch to db", db.key.Log(), db.priority)) break LOOP } - self.dbTotal++ - self.total++ + db.dbTotal++ + db.total++ // otherwise break after select - case dbSize = <-self.batch: + case dbSize = <-db.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 - log.Debug(fmt.Sprintf("syncDb[%v/%v] empty db: switching to buffer", self.key.Log(), self.priority)) + log.Debug(fmt.Sprintf("syncDb[%v/%v] empty db: switching to buffer", db.key.Log(), db.priority)) db = nil - buffer = self.buffer + buffer = db.buffer dbSize <- 0 // indicates to 'caller' that batch has been written inDb = 0 continue LOOP } binary.BigEndian.PutUint64(counterValue, counter) - batch.Put(self.counterKey, counterValue) - log.Debug(fmt.Sprintf("syncDb[%v/%v] write batch %v/%v - %x - %x", self.key.Log(), self.priority, inBatch, counter, self.counterKey, counterValue)) - batch = self.writeSyncBatch(batch) + batch.Put(db.counterKey, counterValue) + log.Debug(fmt.Sprintf("syncDb[%v/%v] write batch %v/%v - %x - %x", db.key.Log(), db.priority, inBatch, counter, db.counterKey, counterValue)) + batch = db.writeSyncBatch(batch) dbSize <- inBatch // indicates to 'caller' that batch has been written inBatch = 0 continue LOOP @@ -198,45 +198,45 @@ LOOP: // closing syncDb#quit channel is used to signal to all goroutines to quit case <-quit: // need to save backlog, so switch to db mode - db = self.buffer + db = db.buffer buffer = nil quit = nil - log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save buffer to db", self.key.Log(), self.priority)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save buffer to db", db.key.Log(), db.priority)) close(db) continue LOOP } // only get here if we put req into db - entry, err = self.newSyncDbEntry(req, counter) + entry, err = db.newSyncDbEntry(req, counter) if err != nil { - log.Warn(fmt.Sprintf("syncDb[%v/%v] saving request %v (#%v/%v) failed: %v", self.key.Log(), self.priority, req, inBatch, inDb, err)) + log.Warn(fmt.Sprintf("syncDb[%v/%v] saving request %v (#%v/%v) failed: %v", db.key.Log(), db.priority, req, inBatch, inDb, err)) continue LOOP } batch.Put(entry.key, entry.val) - log.Trace(fmt.Sprintf("syncDb[%v/%v] to batch %v '%v' (#%v/%v/%v)", self.key.Log(), self.priority, req, entry, inBatch, inDb, counter)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] to batch %v '%v' (#%v/%v/%v)", db.key.Log(), db.priority, req, entry, inBatch, inDb, counter)) // if just switched to db mode and not quitting, then launch dbRead // in a parallel go routine to send deliveries from db if inDb == 0 && quit != nil { - log.Trace(fmt.Sprintf("syncDb[%v/%v] start dbRead", self.key.Log(), self.priority)) - go self.dbRead(true, counter, deliver) + log.Trace(fmt.Sprintf("syncDb[%v/%v] start dbRead", db.key.Log(), db.priority)) + go db.dbRead(true, counter, deliver) } inDb++ inBatch++ counter++ // need to save the batch if it gets too large (== dbBatchSize) - if inBatch%int(self.dbBatchSize) == 0 { - batch = self.writeSyncBatch(batch) + if inBatch%int(db.dbBatchSize) == 0 { + batch = db.writeSyncBatch(batch) } } - log.Info(fmt.Sprintf("syncDb[%v:%v]: saved %v keys (saved counter at %v)", self.key.Log(), self.priority, inBatch, counter)) - close(self.done) + log.Info(fmt.Sprintf("syncDb[%v:%v]: saved %v keys (saved counter at %v)", db.key.Log(), db.priority, inBatch, counter)) + close(db.done) } // writes the batch to the db and returns a new batch object -func (self *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch { - err := self.db.Write(batch) +func (db *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch { + err := db.db.Write(batch) if err != nil { - log.Warn(fmt.Sprintf("syncDb[%v/%v] saving batch to db failed: %v", self.key.Log(), self.priority, err)) + log.Warn(fmt.Sprintf("syncDb[%v/%v] saving batch to db failed: %v", db.key.Log(), db.priority, err)) return batch } return new(leveldb.Batch) @@ -247,8 +247,8 @@ type syncDbEntry struct { key, val []byte } -func (self syncDbEntry) String() string { - return fmt.Sprintf("key: %x, value: %x", self.key, self.val) +func (entry syncDbEntry) String() string { + return fmt.Sprintf("key: %x, value: %x", entry.key, entry.val) } /* @@ -272,9 +272,9 @@ dbRead needs a boolean to indicate if on first round all the historical record is synced. Second argument to indicate current db counter The third is the function to apply */ -func (self *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}, chan bool) bool) { +func (db *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}, chan bool) bool) { key := make([]byte, 42) - copy(key, self.start) + copy(key, db.start) binary.BigEndian.PutUint64(key[34:], counter) var batches, n, cnt, total int var more bool @@ -290,8 +290,8 @@ func (self *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{} // so that loop is not blocking while delivering // only relevant if cnt is large select { - case self.batch <- batchSizes: - case <-self.quit: + case db.batch <- batchSizes: + case <-db.quit: return } // wait for the write to finish and get the item count in the next batch @@ -302,31 +302,31 @@ func (self *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{} return } } - it = self.db.NewIterator() + it = db.db.NewIterator() it.Seek(key) if !it.Valid() { - copy(key, self.start) + copy(key, db.start) useBatches = true continue } del = new(leveldb.Batch) - log.Trace(fmt.Sprintf("syncDb[%v/%v]: new iterator: %x (batch %v, count %v)", self.key.Log(), self.priority, key, batches, cnt)) + log.Trace(fmt.Sprintf("syncDb[%v/%v]: new iterator: %x (batch %v, count %v)", db.key.Log(), db.priority, key, batches, cnt)) for n = 0; !useBatches || n < cnt; it.Next() { copy(key, it.Key()) if len(key) == 0 || key[0] != 0 { - copy(key, self.start) + copy(key, db.start) useBatches = true break } val := make([]byte, 40) copy(val, it.Value()) entry = &syncDbEntry{key, val} - // log.Trace(fmt.Sprintf("syncDb[%v/%v] - %v, batches: %v, total: %v, session total from db: %v/%v", self.key.Log(), self.priority, self.key.Log(), batches, total, self.dbTotal, self.total)) - more = fun(entry, self.quit) + // log.Trace(fmt.Sprintf("syncDb[%v/%v] - %v, batches: %v, total: %v, session total from db: %v/%v", db.key.Log(), db.priority, db.key.Log(), batches, total, db.dbTotal, db.total)) + more = fun(entry, db.quit) if !more { // quit received when waiting to deliver entry, the entry will not be deleted - log.Trace(fmt.Sprintf("syncDb[%v/%v] batch %v quit after %v/%v items", self.key.Log(), self.priority, batches, n, cnt)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] batch %v quit after %v/%v items", db.key.Log(), db.priority, batches, n, cnt)) break } // since subsequent batches of the same db session are indexed incrementally @@ -336,22 +336,22 @@ func (self *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{} n++ total++ } - log.Debug(fmt.Sprintf("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 + log.Debug(fmt.Sprintf("syncDb[%v/%v] - db session closed, batches: %v, total: %v, session total from db: %v/%v", db.key.Log(), db.priority, batches, total, db.dbTotal, db.total)) + db.db.Write(del) // this could be async called only when db is idle it.Release() } } // -func (self *syncDb) stop() { - close(self.quit) - <-self.done +func (db *syncDb) stop() { + close(db.quit) + <-db.done } // calculate a dbkey for the request, for the db to work // see syncdb for db key structure // polimorphic: accepted types, see syncer#addRequest -func (self *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDbEntry, err error) { +func (db *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDbEntry, err error) { var key storage.Key var chunk *storage.Chunk var id uint64 @@ -377,7 +377,7 @@ func (self *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *sync dbval := make([]byte, 40) // encode key - copy(dbkey[:], self.start[:34]) // db peer + copy(dbkey[:], db.start[:34]) // db peer binary.BigEndian.PutUint64(dbkey[34:], counter) // encode value copy(dbval, key[:]) diff --git a/swarm/network/syncdb_test.go b/swarm/network/syncdb_test.go index be21d156f9..7b70ee0f5e 100644 --- a/swarm/network/syncdb_test.go +++ b/swarm/network/syncdb_test.go @@ -71,25 +71,25 @@ func newTestSyncDb(priority, bufferSize, batchSize int, dbdir string, t *testing } -func (self *testSyncDb) close() { - self.db.Close() - os.RemoveAll(self.dbdir) +func (db *testSyncDb) close() { + db.db.Close() + os.RemoveAll(db.dbdir) } -func (self *testSyncDb) push(n int) { +func (db *testSyncDb) push(n int) { for i := 0; i < n; i++ { - self.buffer <- storage.Key(crypto.Keccak256([]byte{byte(self.c)})) - self.sent = append(self.sent, self.c) - self.c++ + db.buffer <- storage.Key(crypto.Keccak256([]byte{byte(db.c)})) + db.sent = append(db.sent, db.c) + db.c++ } log.Debug(fmt.Sprintf("pushed %v requests", n)) } -func (self *testSyncDb) draindb() { - it := self.db.NewIterator() +func (db *testSyncDb) draindb() { + it := db.db.NewIterator() defer it.Release() for { - it.Seek(self.start) + it.Seek(db.start) if !it.Valid() { return } @@ -98,44 +98,44 @@ func (self *testSyncDb) draindb() { return } it.Release() - it = self.db.NewIterator() + it = db.db.NewIterator() } } -func (self *testSyncDb) deliver(req interface{}, quit chan bool) bool { +func (db *testSyncDb) deliver(req interface{}, quit chan bool) bool { _, db := req.(*syncDbEntry) key, _, _, _, err := parseRequest(req) if err != nil { - self.t.Fatalf("unexpected error of key %v: %v", key, err) + db.t.Fatalf("unexpected error of key %v: %v", key, err) } - self.delivered = append(self.delivered, key) + db.delivered = append(db.delivered, key) select { - case self.fromDb <- db: + case db.fromDb <- db: return true case <-quit: return false } } -func (self *testSyncDb) expect(n int, db bool) { +func (db *testSyncDb) expect(n int, db bool) { var ok bool // for n items for i := 0; i < n; i++ { - ok = <-self.fromDb - if self.at+1 > len(self.delivered) { - self.t.Fatalf("expected %v, got %v", self.at+1, len(self.delivered)) + ok = <-db.fromDb + if db.at+1 > len(db.delivered) { + db.t.Fatalf("expected %v, got %v", db.at+1, len(db.delivered)) } - if len(self.sent) > self.at && !bytes.Equal(crypto.Keccak256([]byte{byte(self.sent[self.at])}), self.delivered[self.at]) { - self.t.Fatalf("expected delivery %v/%v/%v to be hash of %v, from db: %v = %v", i, n, self.at, self.sent[self.at], ok, db) - log.Debug(fmt.Sprintf("%v/%v/%v to be hash of %v, from db: %v = %v", i, n, self.at, self.sent[self.at], ok, db)) + if len(db.sent) > db.at && !bytes.Equal(crypto.Keccak256([]byte{byte(db.sent[db.at])}), db.delivered[db.at]) { + db.t.Fatalf("expected delivery %v/%v/%v to be hash of %v, from db: %v = %v", i, n, db.at, db.sent[db.at], ok, db) + log.Debug(fmt.Sprintf("%v/%v/%v to be hash of %v, from db: %v = %v", i, n, db.at, db.sent[db.at], ok, db)) } if !ok && db { - self.t.Fatalf("expected delivery %v/%v/%v from db", i, n, self.at) + db.t.Fatalf("expected delivery %v/%v/%v from db", i, n, db.at) } if ok && !db { - self.t.Fatalf("expected delivery %v/%v/%v from cache", i, n, self.at) + db.t.Fatalf("expected delivery %v/%v/%v from cache", i, n, db.at) } - self.at++ + db.at++ } } diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index 6d729fcb9e..8b5c6f80df 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -77,13 +77,13 @@ func NewDbAccess(loc *storage.LocalStore) *DbAccess { } // to obtain the chunks from key or request db entry only -func (self *DbAccess) get(key storage.Key) (*storage.Chunk, error) { - return self.loc.Get(key) +func (dba *DbAccess) get(key storage.Key) (*storage.Chunk, error) { + return dba.loc.Get(key) } // current storage counter of chunk db -func (self *DbAccess) counter() uint64 { - return self.db.Counter() +func (dba *DbAccess) counter() uint64 { + return dba.db.Counter() } // implemented by dbStoreSyncIterator @@ -92,30 +92,29 @@ type keyIterator interface { } // generator function for iteration by address range and storage counter -func (self *DbAccess) iterator(s *syncState) keyIterator { - it, err := self.db.NewSyncIterator(*(s.DbSyncState)) +func (dba *DbAccess) iterator(s *syncState) keyIterator { + it, err := dba.db.NewSyncIterator(*(s.DbSyncState)) if err != nil { return nil } return keyIterator(it) } -func (self syncState) String() string { - if self.Synced { +func (state syncState) String() string { + if state.Synced { return fmt.Sprintf( "session started at: %v, last seen at: %v, latest key: %v", - self.SessionAt, self.LastSeenAt, - self.Latest.Log(), - ) - } else { - return fmt.Sprintf( - "address: %v-%v, index: %v-%v, session started at: %v, last seen at: %v, latest key: %v", - self.Start.Log(), self.Stop.Log(), - self.First, self.Last, - self.SessionAt, self.LastSeenAt, - self.Latest.Log(), + state.SessionAt, state.LastSeenAt, + state.Latest.Log(), ) } + return fmt.Sprintf( + "address: %v-%v, index: %v-%v, session started at: %v, last seen at: %v, latest key: %v", + state.Start.Log(), state.Stop.Log(), + state.First, state.Last, + state.SessionAt, state.LastSeenAt, + state.Latest.Log(), + ) } // syncer parameters (global, not peer specific) @@ -145,8 +144,8 @@ func NewDefaultSyncParams() *SyncParams { //this can only finally be set after all config options (file, cmd line, env vars) //have been evaluated -func (self *SyncParams) Init(path string) { - self.RequestDbPath = filepath.Join(path, "requests") +func (params *SyncParams) Init(path string) { + params.RequestDbPath = filepath.Join(path, "requests") } // syncer is the agent that manages content distribution/storage replication/chunk storeRequest forwarding @@ -266,21 +265,21 @@ func decodeSync(meta *json.RawMessage) (*syncState, error) { sync is called from the syncer constructor and is not supposed to be used externally */ -func (self *syncer) sync() { - state := self.state +func (sync *syncer) sync() { + state := sync.state // sync finished - defer close(self.syncStates) + defer close(sync.syncStates) // 0. first replay stale requests from request db if state.SessionAt == 0 { - log.Debug(fmt.Sprintf("syncer[%v]: nothing to sync", self.key.Log())) + log.Debug(fmt.Sprintf("syncer[%v]: nothing to sync", sync.key.Log())) return } - log.Debug(fmt.Sprintf("syncer[%v]: start replaying stale requests from request db", self.key.Log())) + log.Debug(fmt.Sprintf("syncer[%v]: start replaying stale requests from request db", sync.key.Log())) for p := priorities - 1; p >= 0; p-- { - self.queues[p].dbRead(false, 0, self.replay()) + sync.queues[p].dbRead(false, 0, sync.replay()) } - log.Debug(fmt.Sprintf("syncer[%v]: done replaying stale requests from request db", self.key.Log())) + log.Debug(fmt.Sprintf("syncer[%v]: done replaying stale requests from request db", sync.key.Log())) // unless peer is synced sync unfinished history beginning on if !state.Synced { @@ -289,9 +288,9 @@ func (self *syncer) sync() { if !storage.IsZeroKey(state.Latest) { // 1. there is unfinished earlier sync state.Start = state.Latest - log.Debug(fmt.Sprintf("syncer[%v]: start syncronising backlog (unfinished sync: %v)", self.key.Log(), state)) + log.Debug(fmt.Sprintf("syncer[%v]: start syncronising backlog (unfinished sync: %v)", sync.key.Log(), state)) // blocks while the entire history upto state is synced - self.syncState(state) + sync.syncState(state) if state.Last < state.SessionAt { state.First = state.Last + 1 } @@ -301,8 +300,8 @@ func (self *syncer) sync() { // 2. sync up to last disconnect1 if state.First < state.LastSeenAt { state.Last = state.LastSeenAt - log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history upto last disconnect at %v: %v", self.key.Log(), state.LastSeenAt, state)) - self.syncState(state) + log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history upto last disconnect at %v: %v", sync.key.Log(), state.LastSeenAt, state)) + sync.syncState(state) state.First = state.LastSeenAt } state.Latest = storage.ZeroKey @@ -316,28 +315,28 @@ func (self *syncer) sync() { // if there have been new chunks since last session if state.LastSeenAt < state.SessionAt { state.Last = state.SessionAt - log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history since last disconnect at %v up until session start at %v: %v", self.key.Log(), state.LastSeenAt, state.SessionAt, state)) + log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history since last disconnect at %v up until session start at %v: %v", sync.key.Log(), state.LastSeenAt, state.SessionAt, state)) // blocks until state syncing is finished - self.syncState(state) + sync.syncState(state) } - log.Info(fmt.Sprintf("syncer[%v]: syncing all history complete", self.key.Log())) + log.Info(fmt.Sprintf("syncer[%v]: syncing all history complete", sync.key.Log())) } // wait till syncronised block uptil state is synced -func (self *syncer) syncState(state *syncState) { - self.syncStates <- state +func (sync *syncer) syncState(state *syncState) { + sync.syncStates <- state select { case <-state.synced: - case <-self.quit: + case <-sync.quit: } } // stop quits both request processor and saves the request cache to disk -func (self *syncer) stop() { - close(self.quit) - log.Trace(fmt.Sprintf("syncer[%v]: stop and save sync request db backlog", self.key.Log())) - for _, db := range self.queues { +func (sync *syncer) stop() { + close(sync.quit) + log.Trace(fmt.Sprintf("syncer[%v]: stop and save sync request db backlog", sync.key.Log())) + for _, db := range sync.queues { db.stop() } } @@ -348,11 +347,11 @@ type syncRequest struct { Priority uint } -func (self *syncRequest) String() string { - return fmt.Sprintf("", self.Key.Log(), self.Priority) +func (req *syncRequest) String() string { + return fmt.Sprintf("", req.Key.Log(), req.Priority) } -func (self *syncer) newSyncRequest(req interface{}, p int) (*syncRequest, error) { +func (sync *syncer) newSyncRequest(req interface{}, p int) (*syncRequest, error) { key, _, _, _, err := parseRequest(req) // TODO: if req has chunk, it should be put in a cache // create @@ -366,11 +365,11 @@ func (self *syncer) newSyncRequest(req interface{}, p int) (*syncRequest, error) // * read is on demand, blocking unless history channel is read // * accepts sync requests (syncStates) to create new db iterator // * closes the channel one iteration finishes -func (self *syncer) syncHistory(state *syncState) chan interface{} { +func (sync *syncer) syncHistory(state *syncState) chan interface{} { var n uint history := make(chan interface{}) - log.Debug(fmt.Sprintf("syncer[%v]: syncing history between %v - %v for chunk addresses %v - %v", self.key.Log(), state.First, state.Last, state.Start, state.Stop)) - it := self.dbAccess.iterator(state) + log.Debug(fmt.Sprintf("syncer[%v]: syncing history between %v - %v for chunk addresses %v - %v", sync.key.Log(), state.First, state.Last, state.Start, state.Stop)) + it := sync.dbAccess.iterator(state) if it != nil { go func() { // signal end of the iteration ended @@ -385,22 +384,22 @@ func (self *syncer) syncHistory(state *syncState) chan interface{} { // blocking until history channel is read from case history <- key: n++ - log.Trace(fmt.Sprintf("syncer[%v]: history: %v (%v keys)", self.key.Log(), key.Log(), n)) + log.Trace(fmt.Sprintf("syncer[%v]: history: %v (%v keys)", sync.key.Log(), key.Log(), n)) state.Latest = key - case <-self.quit: + case <-sync.quit: return } } - log.Debug(fmt.Sprintf("syncer[%v]: finished syncing history between %v - %v for chunk addresses %v - %v (at %v) (chunks = %v)", self.key.Log(), state.First, state.Last, state.Start, state.Stop, state.Latest, n)) + log.Debug(fmt.Sprintf("syncer[%v]: finished syncing history between %v - %v for chunk addresses %v - %v (at %v) (chunks = %v)", sync.key.Log(), state.First, state.Last, state.Start, state.Stop, state.Latest, n)) }() } return history } // triggers key syncronisation -func (self *syncer) sendUnsyncedKeys() { +func (sync *syncer) sendUnsyncedKeys() { select { - case self.deliveryRequest <- true: + case sync.deliveryRequest <- true: default: } } @@ -411,7 +410,7 @@ func (self *syncer) sendUnsyncedKeys() { // historical data is used so historical items are lower priority within // their priority group. // * Order of historical data is unspecified -func (self *syncer) syncUnsyncedKeys() { +func (sync *syncer) syncUnsyncedKeys() { // send out new var unsynced []*syncRequest var more, justSynced bool @@ -419,12 +418,12 @@ func (self *syncer) syncUnsyncedKeys() { var history chan interface{} priority := High - keys := self.keys[priority] + keys := sync.keys[priority] var newUnsyncedKeys, deliveryRequest chan bool keyCounts := make([]int, priorities) - histPrior := self.SyncPriorities[HistoryReq] - syncStates := self.syncStates - state := self.state + histPrior := sync.SyncPriorities[HistoryReq] + syncStates := sync.syncStates + state := sync.state LOOP: for { @@ -440,15 +439,15 @@ LOOP: PRIORITIES: for priority = High; priority >= 0; priority-- { // the first priority channel that is non-empty will be assigned to keys - if len(self.keys[priority]) > 0 { - log.Trace(fmt.Sprintf("syncer[%v]: reading request with priority %v", self.key.Log(), priority)) - keys = self.keys[priority] + if len(sync.keys[priority]) > 0 { + log.Trace(fmt.Sprintf("syncer[%v]: reading request with priority %v", sync.key.Log(), priority)) + keys = sync.keys[priority] break PRIORITIES } - log.Trace(fmt.Sprintf("syncer[%v/%v]: queue: [%v, %v, %v]", self.key.Log(), priority, len(self.keys[High]), len(self.keys[Medium]), len(self.keys[Low]))) + log.Trace(fmt.Sprintf("syncer[%v/%v]: queue: [%v, %v, %v]", sync.key.Log(), priority, len(sync.keys[High]), len(sync.keys[Medium]), len(sync.keys[Low]))) // if the input queue is empty on this level, resort to history if there is any if uint(priority) == histPrior && history != nil { - log.Trace(fmt.Sprintf("syncer[%v]: reading history for %v", self.key.Log(), self.key)) + log.Trace(fmt.Sprintf("syncer[%v]: reading history for %v", sync.key.Log(), sync.key)) keys = history break PRIORITIES } @@ -458,8 +457,8 @@ LOOP: // if peer ready to receive but nothing to send if keys == nil && deliveryRequest == nil { // if no items left and switch to waiting mode - log.Trace(fmt.Sprintf("syncer[%v]: buffers consumed. Waiting", self.key.Log())) - newUnsyncedKeys = self.newUnsyncedKeys + log.Trace(fmt.Sprintf("syncer[%v]: buffers consumed. Waiting", sync.key.Log())) + newUnsyncedKeys = sync.newUnsyncedKeys } // send msg iff @@ -470,48 +469,48 @@ LOOP: if deliveryRequest == nil && (justSynced || len(unsynced) > 0 && keys == nil || - len(unsynced) == int(self.SyncBatchSize)) { + len(unsynced) == int(sync.SyncBatchSize)) { justSynced = false // listen to requests - deliveryRequest = self.deliveryRequest + deliveryRequest = sync.deliveryRequest newUnsyncedKeys = nil // not care about data until next req comes in // set sync to current counter // (all nonhistorical outgoing traffic sheduled and persisted - state.LastSeenAt = self.dbAccess.counter() + state.LastSeenAt = sync.dbAccess.counter() state.Latest = storage.ZeroKey - log.Trace(fmt.Sprintf("syncer[%v]: sending %v", self.key.Log(), unsynced)) - // send the unsynced keys + log.Trace(fmt.Sprintf("syncer[%v]: sending %v", sync.key.Log(), unsynced)) + // send the unsynced keyssync stateCopy := *state - err := self.unsyncedKeys(unsynced, &stateCopy) + err := sync.unsyncedKeys(unsynced, &stateCopy) if err != nil { - log.Warn(fmt.Sprintf("syncer[%v]: unable to send unsynced keys: %v", self.key.Log(), err)) + log.Warn(fmt.Sprintf("syncer[%v]: unable to send unsynced keys: %v", sync.key.Log(), err)) } - self.state = state - log.Debug(fmt.Sprintf("syncer[%v]: --> %v keys sent: (total: %v (%v), history: %v), sent sync state: %v", self.key.Log(), len(unsynced), keyCounts, keyCount, historyCnt, stateCopy)) + sync.state = state + log.Debug(fmt.Sprintf("syncer[%v]: --> %v keys sent: (total: %v (%v), history: %v), sent sync state: %v", sync.key.Log(), len(unsynced), keyCounts, keyCount, historyCnt, stateCopy)) unsynced = nil keys = nil } // process item and add it to the batch select { - case <-self.quit: + case <-sync.quit: break LOOP case req, more = <-keys: if keys == history && !more { - log.Trace(fmt.Sprintf("syncer[%v]: syncing history segment complete", self.key.Log())) + log.Trace(fmt.Sprintf("syncer[%v]: syncing history segment complete", sync.key.Log())) // history channel is closed, waiting for new state (called from sync()) - syncStates = self.syncStates + syncStates = sync.syncStates state.Synced = true // this signals that the current segment is complete select { case state.synced <- false: - case <-self.quit: + case <-sync.quit: break LOOP } justSynced = true history = nil } case <-deliveryRequest: - log.Trace(fmt.Sprintf("syncer[%v]: peer ready to receive", self.key.Log())) + log.Trace(fmt.Sprintf("syncer[%v]: peer ready to receive", sync.key.Log())) // this 1 cap channel can wake up the loop // signaling that peer is ready to receive unsynced Keys @@ -519,23 +518,23 @@ LOOP: deliveryRequest = nil case <-newUnsyncedKeys: - log.Trace(fmt.Sprintf("syncer[%v]: new unsynced keys available", self.key.Log())) + log.Trace(fmt.Sprintf("syncer[%v]: new unsynced keys available", sync.key.Log())) // this 1 cap channel can wake up the loop // signals that data is available to send if peer is ready to receive newUnsyncedKeys = nil - keys = self.keys[High] + keys = sync.keys[High] case state, more = <-syncStates: // this resets the state if !more { - state = self.state - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing complete upto %v)", self.key.Log(), priority, state)) + state = sync.state + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing complete upto %v)", sync.key.Log(), priority, state)) state.Synced = true syncStates = nil } else { - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing history upto %v priority %v)", self.key.Log(), priority, state, histPrior)) + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing history upto %v priority %v)", sync.key.Log(), priority, state, histPrior)) state.Synced = false - history = self.syncHistory(state) + history = sync.syncHistory(state) // only one history at a time, only allow another one once the // history channel is closed syncStates = nil @@ -545,19 +544,19 @@ LOOP: continue LOOP } - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) added to unsynced keys: %v", self.key.Log(), priority, req)) + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) added to unsynced keys: %v", sync.key.Log(), priority, req)) keyCounts[priority]++ keyCount++ if keys == history { - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) history item %v (synced = %v)", self.key.Log(), priority, req, state.Synced)) + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) history item %v (synced = %v)", sync.key.Log(), priority, req, state.Synced)) historyCnt++ } - if sreq, err := self.newSyncRequest(req, priority); err == nil { + if sreq, err := sync.newSyncRequest(req, priority); err == nil { // extract key from req - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v): request %v (synced = %v)", self.key.Log(), priority, req, state.Synced)) + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v): request %v (synced = %v)", sync.key.Log(), priority, req, state.Synced)) unsynced = append(unsynced, sreq) } else { - log.Warn(fmt.Sprintf("syncer[%v]: (priority %v): error creating request for %v: %v)", self.key.Log(), priority, req, err)) + log.Warn(fmt.Sprintf("syncer[%v]: (priority %v): error creating request for %v: %v)", sync.key.Log(), priority, req, err)) } } @@ -566,7 +565,7 @@ LOOP: // delivery loop // takes into account priority, send store Requests with chunk (delivery) // idle blocking if no new deliveries in any of the queues -func (self *syncer) syncDeliveries() { +func (sync *syncer) syncDeliveries() { var req *storeRequestMsgData p := High var deliveries chan *storeRequestMsgData @@ -577,7 +576,7 @@ func (self *syncer) syncDeliveries() { var total, success uint for { - deliveries = self.deliveries[p] + deliveries = sync.deliveries[p] select { case req = <-deliveries: n[p]++ @@ -586,13 +585,13 @@ func (self *syncer) syncDeliveries() { if p == Low { // blocking, depletion on all channels, no preference for priority select { - case req = <-self.deliveries[High]: + case req = <-sync.deliveries[High]: n[High]++ - case req = <-self.deliveries[Medium]: + case req = <-sync.deliveries[Medium]: n[Medium]++ - case req = <-self.deliveries[Low]: + case req = <-sync.deliveries[Low]: n[Low]++ - case <-self.quit: + case <-sync.quit: return } p = High @@ -602,20 +601,20 @@ func (self *syncer) syncDeliveries() { } } total++ - msg, err = self.newStoreRequestMsgData(req) + msg, err = sync.newStoreRequestMsgData(req) if err != nil { - log.Warn(fmt.Sprintf("syncer[%v]: failed to create store request for %v: %v", self.key.Log(), req, err)) + log.Warn(fmt.Sprintf("syncer[%v]: failed to create store request for %v: %v", sync.key.Log(), req, err)) } else { - err = self.store(msg) + err = sync.store(msg) if err != nil { - log.Warn(fmt.Sprintf("syncer[%v]: failed to deliver %v: %v", self.key.Log(), req, err)) + log.Warn(fmt.Sprintf("syncer[%v]: failed to deliver %v: %v", sync.key.Log(), req, err)) } else { success++ - log.Trace(fmt.Sprintf("syncer[%v]: %v successfully delivered", self.key.Log(), req)) + log.Trace(fmt.Sprintf("syncer[%v]: %v successfully delivered", sync.key.Log(), req)) } } - if total%self.SyncBatchSize == 0 { - log.Debug(fmt.Sprintf("syncer[%v]: deliver Total: %v, Success: %v, High: %v/%v, Medium: %v/%v, Low %v/%v", self.key.Log(), total, success, c[High], n[High], c[Medium], n[Medium], c[Low], n[Low])) + if total%sync.SyncBatchSize == 0 { + log.Debug(fmt.Sprintf("syncer[%v]: deliver Total: %v, Success: %v, High: %v/%v, Medium: %v/%v, Low %v/%v", sync.key.Log(), total, success, c[High], n[High], c[Medium], n[Medium], c[Low], n[Low])) } } } @@ -635,28 +634,28 @@ func (self *syncer) syncDeliveries() { If sync mode is off then, requests are directly sent to deliveries */ -func (self *syncer) addRequest(req interface{}, ty int) { +func (sync *syncer) addRequest(req interface{}, ty int) { // retrieve priority for request type name int8 - priority := self.SyncPriorities[ty] + priority := sync.SyncPriorities[ty] // sync mode for this type ON - if self.syncF() || ty == DeliverReq { - if self.SyncModes[ty] { - self.addKey(req, priority, self.quit) + if sync.syncF() || ty == DeliverReq { + if sync.SyncModes[ty] { + sync.addKey(req, priority, sync.quit) } else { - self.addDelivery(req, priority, self.quit) + sync.addDelivery(req, priority, sync.quit) } } } // addKey queues sync request for sync confirmation with given priority // ie the key will go out in an unsyncedKeys message -func (self *syncer) addKey(req interface{}, priority uint, quit chan bool) bool { +func (sync *syncer) addKey(req interface{}, priority uint, quit chan bool) bool { select { - case self.keys[priority] <- req: + case sync.keys[priority] <- req: // this wakes up the unsynced keys loop if idle select { - case self.newUnsyncedKeys <- true: + case sync.newUnsyncedKeys <- true: default: } return true @@ -668,9 +667,9 @@ func (self *syncer) addKey(req interface{}, priority uint, quit chan bool) bool // addDelivery queues delivery request for with given priority // ie the chunk will be delivered ASAP mod priority queueing handled by syncdb // requests are persisted across sessions for correct sync -func (self *syncer) addDelivery(req interface{}, priority uint, quit chan bool) bool { +func (sync *syncer) addDelivery(req interface{}, priority uint, quit chan bool) bool { select { - case self.queues[priority].buffer <- req: + case sync.queues[priority].buffer <- req: return true case <-quit: return false @@ -679,14 +678,14 @@ func (self *syncer) addDelivery(req interface{}, priority uint, quit chan bool) // doDelivery delivers the chunk for the request with given priority // without queuing -func (self *syncer) doDelivery(req interface{}, priority uint, quit chan bool) bool { - msgdata, err := self.newStoreRequestMsgData(req) +func (sync *syncer) doDelivery(req interface{}, priority uint, quit chan bool) bool { + msgdata, err := sync.newStoreRequestMsgData(req) if err != nil { log.Warn(fmt.Sprintf("unable to deliver request %v: %v", msgdata, err)) return false } select { - case self.deliveries[priority] <- msgdata: + case sync.deliveries[priority] <- msgdata: return true case <-quit: return false @@ -695,9 +694,9 @@ func (self *syncer) doDelivery(req interface{}, priority uint, quit chan bool) b // returns the delivery function for given priority // passed on to syncDb -func (self *syncer) deliver(priority uint) func(req interface{}, quit chan bool) bool { +func (sync *syncer) deliver(priority uint) func(req interface{}, quit chan bool) bool { return func(req interface{}, quit chan bool) bool { - return self.doDelivery(req, priority, quit) + return sync.doDelivery(req, priority, quit) } } @@ -705,25 +704,23 @@ func (self *syncer) deliver(priority uint) func(req interface{}, quit chan bool) // depending on sync mode settings for BacklogReq, // re play of request db backlog sends items via confirmation // or directly delivers -func (self *syncer) replay() func(req interface{}, quit chan bool) bool { - sync := self.SyncModes[BacklogReq] - priority := self.SyncPriorities[BacklogReq] +func (sync *syncer) replay() func(req interface{}, quit chan bool) bool { + sync := sync.SyncModes[BacklogReq] + priority := sync.SyncPriorities[BacklogReq] // sync mode for this type ON if sync { return func(req interface{}, quit chan bool) bool { - return self.addKey(req, priority, quit) + return sync.addKey(req, priority, quit) } - } else { - return func(req interface{}, quit chan bool) bool { - return self.doDelivery(req, priority, quit) - } - + } + return func(req interface{}, quit chan bool) bool { + return sync.doDelivery(req, priority, quit) } } // given a request, extends it to a full storeRequestMsgData // polimorphic: see addRequest for the types accepted -func (self *syncer) newStoreRequestMsgData(req interface{}) (*storeRequestMsgData, error) { +func (sync *syncer) newStoreRequestMsgData(req interface{}) (*storeRequestMsgData, error) { key, id, chunk, sreq, err := parseRequest(req) if err != nil { @@ -733,7 +730,7 @@ func (self *syncer) newStoreRequestMsgData(req interface{}) (*storeRequestMsgDat if sreq == nil { if chunk == nil { var err error - chunk, err = self.dbAccess.get(key) + chunk, err = sync.dbAccess.get(key) if err != nil { return nil, err } diff --git a/swarm/services/swap/swap.go b/swarm/services/swap/swap.go index 1f9b22b904..4c03964e60 100644 --- a/swarm/services/swap/swap.go +++ b/swarm/services/swap/swap.go @@ -80,7 +80,7 @@ type PayProfile struct { lock sync.RWMutex } -//create params with default values +// NewDefaultSwapParams creates params with default values func NewDefaultSwapParams() *SwapParams { return &SwapParams{ PayProfile: &PayProfile{}, @@ -104,10 +104,10 @@ func NewDefaultSwapParams() *SwapParams { //this can only finally be set after all config options (file, cmd line, env vars) //have been evaluated -func (self *SwapParams) Init(contract common.Address, prvkey *ecdsa.PrivateKey) { +func (params *SwapParams) Init(contract common.Address, prvkey *ecdsa.PrivateKey) { pubkey := &prvkey.PublicKey - self.PayProfile = &PayProfile{ + params.PayProfile = &PayProfile{ PublicKey: common.ToHex(crypto.FromECDSAPub(pubkey)), Contract: contract, Beneficiary: crypto.PubkeyToAddress(*pubkey), @@ -184,44 +184,44 @@ func NewSwap(local *SwapParams, remote *SwapProfile, backend chequebook.Backend, return } -func (self *SwapParams) Chequebook() *chequebook.Chequebook { - defer self.lock.Unlock() - self.lock.Lock() - return self.chbook +func (params *SwapParams) Chequebook() *chequebook.Chequebook { + defer params.lock.Unlock() + params.lock.Lock() + return params.chbook } -func (self *SwapParams) PrivateKey() *ecdsa.PrivateKey { - return self.privateKey +func (params *SwapParams) PrivateKey() *ecdsa.PrivateKey { + return params.privateKey } -// func (self *SwapParams) PublicKey() *ecdsa.PublicKey { -// return self.publicKey +// func (params *SwapParams) PublicKey() *ecdsa.PublicKey { +// return params.publicKey // } -func (self *SwapParams) SetKey(prvkey *ecdsa.PrivateKey) { - self.privateKey = prvkey - self.publicKey = &prvkey.PublicKey +func (params *SwapParams) SetKey(prvkey *ecdsa.PrivateKey) { + params.privateKey = prvkey + params.publicKey = &prvkey.PublicKey } -// setChequebook(path, backend) wraps the +// SetChequebook wraps the // chequebook initialiser and sets up autoDeposit to cover spending. -func (self *SwapParams) SetChequebook(ctx context.Context, backend chequebook.Backend, path string) error { - self.lock.Lock() - contract := self.Contract - self.lock.Unlock() +func (params *SwapParams) SetChequebook(ctx context.Context, backend chequebook.Backend, path string) error { + params.lock.Lock() + contract := params.Contract + params.lock.Unlock() valid, err := chequebook.ValidateCode(ctx, backend, contract) if err != nil { return err } else if valid { - return self.newChequebookFromContract(path, backend) + return params.newChequebookFromContract(path, backend) } - return self.deployChequebook(ctx, backend, path) + return params.deployChequebook(ctx, backend, path) } -func (self *SwapParams) deployChequebook(ctx context.Context, backend chequebook.Backend, path string) error { - opts := bind.NewKeyedTransactor(self.privateKey) - opts.Value = self.AutoDepositBuffer +func (params *SwapParams) deployChequebook(ctx context.Context, backend chequebook.Backend, path string) error { + opts := bind.NewKeyedTransactor(params.privateKey) + opts.Value = params.AutoDepositBuffer opts.Context = ctx log.Info(fmt.Sprintf("Deploying new chequebook (owner: %v)", opts.From.Hex())) @@ -233,10 +233,10 @@ func (self *SwapParams) deployChequebook(ctx context.Context, backend chequebook log.Info(fmt.Sprintf("new chequebook deployed at %v (owner: %v)", contract.Hex(), opts.From.Hex())) // need to save config at this point - self.lock.Lock() - self.Contract = contract - err = self.newChequebookFromContract(path, backend) - self.lock.Unlock() + params.lock.Lock() + params.Contract = contract + err = params.newChequebookFromContract(path, backend) + params.lock.Unlock() if err != nil { log.Warn(fmt.Sprintf("error initialising cheque book (owner: %v): %v", opts.From.Hex(), err)) } @@ -265,26 +265,26 @@ func deployChequebookLoop(opts *bind.TransactOpts, backend chequebook.Backend) ( // initialise the chequebook from a persisted json file or create a new one // caller holds the lock -func (self *SwapParams) newChequebookFromContract(path string, backend chequebook.Backend) error { - hexkey := common.Bytes2Hex(self.Contract.Bytes()) +func (params *SwapParams) newChequebookFromContract(path string, backend chequebook.Backend) error { + hexkey := common.Bytes2Hex(params.Contract.Bytes()) err := os.MkdirAll(filepath.Join(path, "chequebooks"), os.ModePerm) if err != nil { return fmt.Errorf("unable to create directory for chequebooks: %v", err) } chbookpath := filepath.Join(path, "chequebooks", hexkey+".json") - self.chbook, err = chequebook.LoadChequebook(chbookpath, self.privateKey, backend, true) + params.chbook, err = chequebook.LoadChequebook(chbookpath, params.privateKey, backend, true) if err != nil { - self.chbook, err = chequebook.NewChequebook(chbookpath, self.Contract, self.privateKey, backend) + params.chbook, err = chequebook.NewChequebook(chbookpath, params.Contract, params.privateKey, backend) if err != nil { - log.Warn(fmt.Sprintf("unable to initialise chequebook (owner: %v): %v", self.owner.Hex(), err)) - return fmt.Errorf("unable to initialise chequebook (owner: %v): %v", self.owner.Hex(), err) + log.Warn(fmt.Sprintf("unable to initialise chequebook (owner: %v): %v", params.owner.Hex(), err)) + return fmt.Errorf("unable to initialise chequebook (owner: %v): %v", params.owner.Hex(), err) } } - self.chbook.AutoDeposit(self.AutoDepositInterval, self.AutoDepositThreshold, self.AutoDepositBuffer) - log.Info(fmt.Sprintf("auto deposit ON for %v -> %v: interval = %v, threshold = %v, buffer = %v)", crypto.PubkeyToAddress(*(self.publicKey)).Hex()[:8], self.Contract.Hex()[:8], self.AutoDepositInterval, self.AutoDepositThreshold, self.AutoDepositBuffer)) + params.chbook.AutoDeposit(params.AutoDepositInterval, params.AutoDepositThreshold, params.AutoDepositBuffer) + log.Info(fmt.Sprintf("auto deposit ON for %v -> %v: interval = %v, threshold = %v, buffer = %v)", crypto.PubkeyToAddress(*(params.publicKey)).Hex()[:8], params.Contract.Hex()[:8], params.AutoDepositInterval, params.AutoDepositThreshold, params.AutoDepositBuffer)) return nil } diff --git a/swarm/services/swap/swap/swap.go b/swarm/services/swap/swap/swap.go index a78f1f0e2a..5ca5eaf743 100644 --- a/swarm/services/swap/swap/swap.go +++ b/swarm/services/swap/swap/swap.go @@ -82,7 +82,7 @@ type InPayment interface { Stop() } -// swap is the swarm accounting protocol instance +// Swap is the swarm accounting protocol instance // * pairwise accounting and payments type Swap struct { lock sync.Mutex // mutex for balance access @@ -114,139 +114,139 @@ func New(local *Params, pm Payment, proto Protocol) (self *Swap, err error) { } // entry point for setting remote swap profile (e.g from handshake or other message) -func (self *Swap) SetRemote(remote *Profile) { - defer self.lock.Unlock() - self.lock.Lock() +func (swap *Swap) SetRemote(remote *Profile) { + defer swap.lock.Unlock() + swap.lock.Lock() - self.remote = remote - if self.Sells && (remote.BuyAt.Sign() <= 0 || self.local.SellAt.Sign() <= 0 || remote.BuyAt.Cmp(self.local.SellAt) < 0) { - self.Out.Stop() - self.Sells = false + swap.remote = remote + if swap.Sells && (remote.BuyAt.Sign() <= 0 || swap.local.SellAt.Sign() <= 0 || remote.BuyAt.Cmp(swap.local.SellAt) < 0) { + swap.Out.Stop() + swap.Sells = false } - if self.Buys && (remote.SellAt.Sign() <= 0 || self.local.BuyAt.Sign() <= 0 || self.local.BuyAt.Cmp(self.remote.SellAt) < 0) { - self.In.Stop() - self.Buys = false + if swap.Buys && (remote.SellAt.Sign() <= 0 || swap.local.BuyAt.Sign() <= 0 || swap.local.BuyAt.Cmp(swap.remote.SellAt) < 0) { + swap.In.Stop() + swap.Buys = false } - log.Debug(fmt.Sprintf("<%v> remote profile set: pay at: %v, drop at: %v, buy at: %v, sell at: %v", self.proto, remote.PayAt, remote.DropAt, remote.BuyAt, remote.SellAt)) + log.Debug(fmt.Sprintf("<%v> remote profile set: pay at: %v, drop at: %v, buy at: %v, sell at: %v", swap.proto, remote.PayAt, remote.DropAt, remote.BuyAt, remote.SellAt)) } // to set strategy dynamically -func (self *Swap) SetParams(local *Params) { - defer self.lock.Unlock() - self.lock.Lock() - self.local = local - self.setParams(local) +func (swap *Swap) SetParams(local *Params) { + defer swap.lock.Unlock() + swap.lock.Lock() + swap.local = local + swap.setParams(local) } // caller holds the lock -func (self *Swap) setParams(local *Params) { +func (swap *Swap) setParams(local *Params) { - if self.Sells { - self.In.AutoCash(local.AutoCashInterval, local.AutoCashThreshold) - log.Info(fmt.Sprintf("<%v> set autocash to every %v, max uncashed limit: %v", self.proto, local.AutoCashInterval, local.AutoCashThreshold)) + if swap.Sells { + swap.In.AutoCash(local.AutoCashInterval, local.AutoCashThreshold) + log.Info(fmt.Sprintf("<%v> set autocash to every %v, max uncashed limit: %v", swap.proto, local.AutoCashInterval, local.AutoCashThreshold)) } else { - log.Info(fmt.Sprintf("<%v> autocash off (not selling)", self.proto)) + log.Info(fmt.Sprintf("<%v> autocash off (not selling)", swap.proto)) } - if self.Buys { - self.Out.AutoDeposit(local.AutoDepositInterval, local.AutoDepositThreshold, local.AutoDepositBuffer) - log.Info(fmt.Sprintf("<%v> set autodeposit to every %v, pay at: %v, buffer: %v", self.proto, local.AutoDepositInterval, local.AutoDepositThreshold, local.AutoDepositBuffer)) + if swap.Buys { + swap.Out.AutoDeposit(local.AutoDepositInterval, local.AutoDepositThreshold, local.AutoDepositBuffer) + log.Info(fmt.Sprintf("<%v> set autodeposit to every %v, pay at: %v, buffer: %v", swap.proto, local.AutoDepositInterval, local.AutoDepositThreshold, local.AutoDepositBuffer)) } else { - log.Info(fmt.Sprintf("<%v> autodeposit off (not buying)", self.proto)) + log.Info(fmt.Sprintf("<%v> autodeposit off (not buying)", swap.proto)) } } // Add(n) // n > 0 called when promised/provided n units of service // n < 0 called when used/requested n units of service -func (self *Swap) Add(n int) error { - defer self.lock.Unlock() - self.lock.Lock() - self.balance += n - if !self.Sells && self.balance > 0 { - log.Trace(fmt.Sprintf("<%v> remote peer cannot have debt (balance: %v)", self.proto, self.balance)) - self.proto.Drop() - return fmt.Errorf("[SWAP] <%v> remote peer cannot have debt (balance: %v)", self.proto, self.balance) +func (swap *Swap) Add(n int) error { + defer swap.lock.Unlock() + swap.lock.Lock() + swap.balance += n + if !swap.Sells && swap.balance > 0 { + log.Trace(fmt.Sprintf("<%v> remote peer cannot have debt (balance: %v)", swap.proto, swap.balance)) + swap.proto.Drop() + return fmt.Errorf("[SWAP] <%v> remote peer cannot have debt (balance: %v)", swap.proto, swap.balance) } - if !self.Buys && self.balance < 0 { - log.Trace(fmt.Sprintf("<%v> we cannot have debt (balance: %v)", self.proto, self.balance)) - return fmt.Errorf("[SWAP] <%v> we cannot have debt (balance: %v)", self.proto, self.balance) + if !swap.Buys && swap.balance < 0 { + log.Trace(fmt.Sprintf("<%v> we cannot have debt (balance: %v)", swap.proto, swap.balance)) + return fmt.Errorf("[SWAP] <%v> we cannot have debt (balance: %v)", swap.proto, swap.balance) } - if self.balance >= int(self.local.DropAt) { - log.Trace(fmt.Sprintf("<%v> remote peer has too much debt (balance: %v, disconnect threshold: %v)", self.proto, self.balance, self.local.DropAt)) - self.proto.Drop() - return fmt.Errorf("[SWAP] <%v> remote peer has too much debt (balance: %v, disconnect threshold: %v)", self.proto, self.balance, self.local.DropAt) - } else if self.balance <= -int(self.remote.PayAt) { - self.send() + if swap.balance >= int(swap.local.DropAt) { + log.Trace(fmt.Sprintf("<%v> remote peer has too much debt (balance: %v, disconnect threshold: %v)", swap.proto, swap.balance, swap.local.DropAt)) + swap.proto.Drop() + return fmt.Errorf("[SWAP] <%v> remote peer has too much debt (balance: %v, disconnect threshold: %v)", swap.proto, swap.balance, swap.local.DropAt) + } else if swap.balance <= -int(swap.remote.PayAt) { + swap.send() } return nil } -func (self *Swap) Balance() int { - defer self.lock.Unlock() - self.lock.Lock() - return self.balance +func (swap *Swap) Balance() int { + defer swap.lock.Unlock() + swap.lock.Lock() + return swap.balance } // send(units) is called when payment is due // In case of insolvency no promise is issued and sent, safe against fraud // No return value: no error = payment is opportunistic = hang in till dropped -func (self *Swap) send() { - if self.local.BuyAt != nil && self.balance < 0 { - amount := big.NewInt(int64(-self.balance)) - amount.Mul(amount, self.remote.SellAt) - promise, err := self.Out.Issue(amount) +func (swap *Swap) send() { + if swap.local.BuyAt != nil && swap.balance < 0 { + amount := big.NewInt(int64(-swap.balance)) + amount.Mul(amount, swap.remote.SellAt) + promise, err := swap.Out.Issue(amount) if err != nil { - log.Warn(fmt.Sprintf("<%v> cannot issue cheque (amount: %v, channel: %v): %v", self.proto, amount, self.Out, err)) + log.Warn(fmt.Sprintf("<%v> cannot issue cheque (amount: %v, channel: %v): %v", swap.proto, amount, swap.Out, err)) } else { - log.Warn(fmt.Sprintf("<%v> cheque issued (amount: %v, channel: %v)", self.proto, amount, self.Out)) - self.proto.Pay(-self.balance, promise) - self.balance = 0 + log.Warn(fmt.Sprintf("<%v> cheque issued (amount: %v, channel: %v)", swap.proto, amount, swap.Out)) + swap.proto.Pay(-swap.balance, promise) + swap.balance = 0 } } } // receive(units, promise) is called by the protocol when a payment msg is received // returns error if promise is invalid. -func (self *Swap) Receive(units int, promise Promise) error { +func (swap *Swap) Receive(units int, promise Promise) error { if units <= 0 { return fmt.Errorf("invalid units: %v <= 0", units) } price := new(big.Int).SetInt64(int64(units)) - price.Mul(price, self.local.SellAt) + price.Mul(price, swap.local.SellAt) - amount, err := self.In.Receive(promise) + amount, err := swap.In.Receive(promise) if err != nil { err = fmt.Errorf("invalid promise: %v", err) } else if price.Cmp(amount) != 0 { // verify amount = units * unit sale price - return fmt.Errorf("invalid amount: %v = %v * %v (units sent in msg * agreed sale unit price) != %v (signed in cheque)", price, units, self.local.SellAt, amount) + return fmt.Errorf("invalid amount: %v = %v * %v (units sent in msg * agreed sale unit price) != %v (signed in cheque)", price, units, swap.local.SellAt, amount) } if err != nil { - log.Trace(fmt.Sprintf("<%v> invalid promise (amount: %v, channel: %v): %v", self.proto, amount, self.In, err)) + log.Trace(fmt.Sprintf("<%v> invalid promise (amount: %v, channel: %v): %v", swap.proto, amount, swap.In, err)) return err } // credit remote peer with units - self.Add(-units) - log.Trace(fmt.Sprintf("<%v> received promise (amount: %v, channel: %v): %v", self.proto, amount, self.In, promise)) + swap.Add(-units) + log.Trace(fmt.Sprintf("<%v> received promise (amount: %v, channel: %v): %v", swap.proto, amount, swap.In, promise)) return nil } // stop() causes autocash loop to terminate. // Called after protocol handle loop terminates. -func (self *Swap) Stop() { - defer self.lock.Unlock() - self.lock.Lock() - if self.Buys { - self.Out.Stop() +func (swap *Swap) Stop() { + defer swap.lock.Unlock() + swap.lock.Lock() + if swap.Buys { + swap.Out.Stop() } - if self.Sells { - self.In.Stop() + if swap.Sells { + swap.In.Stop() } } diff --git a/swarm/services/swap/swap/swap_test.go b/swarm/services/swap/swap/swap_test.go index 222e0770f3..d0e34f8a54 100644 --- a/swarm/services/swap/swap/swap_test.go +++ b/swarm/services/swap/swap/swap_test.go @@ -34,20 +34,20 @@ type testPromise struct { amount *big.Int } -func (self *testInPayment) Receive(promise Promise) (*big.Int, error) { +func (test *testInPayment) Receive(promise Promise) (*big.Int, error) { p := promise.(*testPromise) - self.received = append(self.received, p) + test.received = append(test.received, p) return p.amount, nil } -func (self *testInPayment) AutoCash(interval time.Duration, limit *big.Int) { - self.autocashInterval = interval - self.autocashLimit = limit +func (test *testInPayment) AutoCash(interval time.Duration, limit *big.Int) { + test.autocashInterval = interval + test.autocashLimit = limit } -func (self *testInPayment) Cash() (string, error) { return "", nil } +func (test *testInPayment) Cash() (string, error) { return "", nil } -func (self *testInPayment) Stop() {} +func (test *testInPayment) Stop() {} type testOutPayment struct { deposits []*big.Int @@ -56,22 +56,22 @@ type testOutPayment struct { autodepositBuffer *big.Int } -func (self *testOutPayment) Issue(amount *big.Int) (promise Promise, err error) { +func (test *testOutPayment) Issue(amount *big.Int) (promise Promise, err error) { return &testPromise{amount}, nil } -func (self *testOutPayment) Deposit(amount *big.Int) (string, error) { - self.deposits = append(self.deposits, amount) +func (test *testOutPayment) Deposit(amount *big.Int) (string, error) { + test.deposits = append(test.deposits, amount) return "", nil } -func (self *testOutPayment) AutoDeposit(interval time.Duration, threshold, buffer *big.Int) { - self.autodepositInterval = interval - self.autodepositThreshold = threshold - self.autodepositBuffer = buffer +func (test *testOutPayment) AutoDeposit(interval time.Duration, threshold, buffer *big.Int) { + test.autodepositInterval = interval + test.autodepositThreshold = threshold + test.autodepositBuffer = buffer } -func (self *testOutPayment) Stop() {} +func (test *testOutPayment) Stop() {} type testProtocol struct { drop bool @@ -79,18 +79,18 @@ type testProtocol struct { promises []*testPromise } -func (self *testProtocol) Drop() { - self.drop = true +func (test *testProtocol) Drop() { + test.drop = true } -func (self *testProtocol) String() string { +func (test *testProtocol) String() string { return "" } -func (self *testProtocol) Pay(amount int, promise Promise) { +func (test *testProtocol) Pay(amount int, promise Promise) { p := promise.(*testPromise) - self.promises = append(self.promises, p) - self.amounts = append(self.amounts, amount) + test.promises = append(test.promises, p) + test.amounts = append(test.amounts, amount) } func TestSwap(t *testing.T) { diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index 2b397f8010..2956d489ea 100644 --- a/swarm/storage/chunker.go +++ b/swarm/storage/chunker.go @@ -91,13 +91,13 @@ func NewTreeChunker(params *ChunkerParams) (self *TreeChunker) { return } -// func (self *TreeChunker) KeySize() int64 { -// return self.hashSize +// func (tc *TreeChunker) KeySize() int64 { +// return tc.hashSize // } // String() for pretty printing -func (self *Chunk) String() string { - return fmt.Sprintf("Key: %v TreeSize: %v Chunksize: %v", self.Key.Log(), self.Size, len(self.SData)) +func (c *Chunk) String() string { + return fmt.Sprintf("Key: %v TreeSize: %v Chunksize: %v", c.Key.Log(), c.Size, len(c.SData)) } type hashJob struct { @@ -107,26 +107,26 @@ type hashJob struct { parentWg *sync.WaitGroup } -func (self *TreeChunker) incrementWorkerCount() { - self.workerLock.Lock() - defer self.workerLock.Unlock() - self.workerCount += 1 +func (tc *TreeChunker) incrementWorkerCount() { + tc.workerLock.Lock() + defer tc.workerLock.Unlock() + tc.workerCount++ } -func (self *TreeChunker) getWorkerCount() int64 { - self.workerLock.RLock() - defer self.workerLock.RUnlock() - return self.workerCount +func (tc *TreeChunker) getWorkerCount() int64 { + tc.workerLock.RLock() + defer tc.workerLock.RUnlock() + return tc.workerCount } -func (self *TreeChunker) decrementWorkerCount() { - self.workerLock.Lock() - defer self.workerLock.Unlock() - self.workerCount -= 1 +func (tc *TreeChunker) decrementWorkerCount() { + tc.workerLock.Lock() + defer tc.workerLock.Unlock() + tc.workerCount-- } -func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, swg, wwg *sync.WaitGroup) (Key, error) { - if self.chunkSize <= 0 { +func (tc *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, swg, wwg *sync.WaitGroup) (Key, error) { + if tc.chunkSize <= 0 { panic("chunker must be initialised") } @@ -140,23 +140,23 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s wwg.Add(1) } - self.incrementWorkerCount() - go self.hashWorker(jobC, chunkC, errC, quitC, swg, wwg) + tc.incrementWorkerCount() + go tc.hashWorker(jobC, chunkC, errC, quitC, swg, wwg) depth := 0 - treeSize := self.chunkSize + treeSize := tc.chunkSize // takes lowest depth such that chunksize*HashCount^(depth+1) > size // power series, will find the order of magnitude of the data size in base hashCount or numbers of levels of branching in the resulting tree. - for ; treeSize < size; treeSize *= self.branches { + for ; treeSize < size; treeSize *= tc.branches { depth++ } - key := make([]byte, self.hashFunc().Size()) + key := make([]byte, tc.hashFunc().Size()) // this waitgroup member is released after the root hash is calculated wg.Add(1) //launch actual recursive function passing the waitgroups - go self.split(depth, treeSize/self.branches, key, data, size, jobC, chunkC, errC, quitC, wg, swg, wwg) + go tc.split(depth, treeSize/tc.branches, key, data, size, jobC, chunkC, errC, quitC, wg, swg, wwg) // closes internal error channel if all subprocesses in the workgroup finished go func() { @@ -182,12 +182,12 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s return key, nil } -func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reader, size int64, jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, parentWg, swg, wwg *sync.WaitGroup) { +func (tc *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reader, size int64, jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, parentWg, swg, wwg *sync.WaitGroup) { // for depth > 0 && size < treeSize { - treeSize /= self.branches + treeSize /= tc.branches depth-- } @@ -214,7 +214,7 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade // intermediate chunk containing child nodes hashes branchCnt := (size + treeSize - 1) / treeSize - var chunk = make([]byte, branchCnt*self.hashSize+8) + var chunk = make([]byte, branchCnt*tc.hashSize+8) var pos, i int64 binary.LittleEndian.PutUint64(chunk[0:8], uint64(size)) @@ -229,10 +229,10 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade secSize = treeSize } // the hash of that data - subTreeKey := chunk[8+i*self.hashSize : 8+(i+1)*self.hashSize] + subTreeKey := chunk[8+i*tc.hashSize : 8+(i+1)*tc.hashSize] childrenWg.Add(1) - self.split(depth-1, treeSize/self.branches, subTreeKey, data, secSize, jobC, chunkC, errC, quitC, childrenWg, swg, wwg) + tc.split(depth-1, treeSize/tc.branches, subTreeKey, data, secSize, jobC, chunkC, errC, quitC, childrenWg, swg, wwg) i++ pos += treeSize @@ -242,13 +242,13 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade // go func() { childrenWg.Wait() - worker := self.getWorkerCount() + worker := tc.getWorkerCount() if int64(len(jobC)) > worker && worker < ChunkProcessors { if wwg != nil { wwg.Add(1) } - self.incrementWorkerCount() - go self.hashWorker(jobC, chunkC, errC, quitC, swg, wwg) + tc.incrementWorkerCount() + go tc.hashWorker(jobC, chunkC, errC, quitC, swg, wwg) } select { @@ -257,10 +257,10 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade } } -func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, swg, wwg *sync.WaitGroup) { - defer self.decrementWorkerCount() +func (tc *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, swg, wwg *sync.WaitGroup) { + defer tc.decrementWorkerCount() - hasher := self.hashFunc() + hasher := tc.hashFunc() if wwg != nil { defer wwg.Done() } @@ -272,7 +272,7 @@ func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC return } // now we got the hashes in the chunk, then hash the chunks - self.hashChunk(hasher, job, chunkC, swg) + tc.hashChunk(hasher, job, chunkC, swg) case <-quitC: return } @@ -282,7 +282,7 @@ func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC // The treeChunkers own Hash hashes together // - the size (of the subtree encoded in the Chunk) // - the Chunk, ie. the contents read from the input reader -func (self *TreeChunker) hashChunk(hasher SwarmHash, job *hashJob, chunkC chan *Chunk, swg *sync.WaitGroup) { +func (tc *TreeChunker) hashChunk(hasher SwarmHash, job *hashJob, chunkC chan *Chunk, swg *sync.WaitGroup) { hasher.ResetWithLength(job.chunk[:8]) // 8 bytes of length hasher.Write(job.chunk[8:]) // minus 8 []byte length h := hasher.Sum(nil) @@ -316,7 +316,7 @@ func (self *TreeChunker) hashChunk(hasher SwarmHash, job *hashJob, chunkC chan * } } -func (self *TreeChunker) Append(key Key, data io.Reader, chunkC chan *Chunk, swg, wwg *sync.WaitGroup) (Key, error) { +func (tc *TreeChunker) Append(key Key, data io.Reader, chunkC chan *Chunk, swg, wwg *sync.WaitGroup) (Key, error) { return nil, errAppendOppNotSuported } @@ -331,45 +331,45 @@ type LazyChunkReader struct { hashSize int64 // inherit from chunker } -// implements the Joiner interface -func (self *TreeChunker) Join(key Key, chunkC chan *Chunk) LazySectionReader { +// Join implements the Joiner interface +func (tc *TreeChunker) Join(key Key, chunkC chan *Chunk) LazySectionReader { return &LazyChunkReader{ key: key, chunkC: chunkC, - chunkSize: self.chunkSize, - branches: self.branches, - hashSize: self.hashSize, + chunkSize: tc.chunkSize, + branches: tc.branches, + hashSize: tc.hashSize, } } // Size is meant to be called on the LazySectionReader -func (self *LazyChunkReader) Size(quitC chan bool) (n int64, err error) { - if self.chunk != nil { - return self.chunk.Size, nil +func (reader *LazyChunkReader) Size(quitC chan bool) (n int64, err error) { + if reader.chunk != nil { + return reader.chunk.Size, nil } - chunk := retrieve(self.key, self.chunkC, quitC) + chunk := retrieve(reader.key, reader.chunkC, quitC) if chunk == nil { select { case <-quitC: return 0, errors.New("aborted") default: - return 0, fmt.Errorf("root chunk not found for %v", self.key.Hex()) + return 0, fmt.Errorf("root chunk not found for %v", reader.key.Hex()) } } - self.chunk = chunk + reader.chunk = chunk return chunk.Size, nil } -// read at can be called numerous times +// ReadAt can be called numerous times // concurrent reads are allowed // Size() needs to be called synchronously on the LazyChunkReader first -func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { +func (reader *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { // this is correct, a swarm doc cannot be zero length, so no EOF is expected if len(b) == 0 { return 0, nil } quitC := make(chan bool) - size, err := self.Size(quitC) + size, err := reader.Size(quitC) if err != nil { return 0, err } @@ -380,13 +380,13 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { var treeSize int64 var depth int // calculate depth and max treeSize - treeSize = self.chunkSize - for ; treeSize < size; treeSize *= self.branches { + treeSize = reader.chunkSize + for ; treeSize < size; treeSize *= reader.branches { depth++ } wg := sync.WaitGroup{} wg.Add(1) - go self.join(b, off, off+int64(len(b)), depth, treeSize/self.branches, self.chunk, &wg, errC, quitC) + go reader.join(b, off, off+int64(len(b)), depth, treeSize/reader.branches, reader.chunk, &wg, errC, quitC) go func() { wg.Wait() close(errC) @@ -404,7 +404,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) { return len(b), nil } -func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) { +func (reader *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunk *Chunk, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) { defer parentWg.Done() // return NewDPA(&LocalStore{}) @@ -412,7 +412,7 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr // find appropriate block level for chunk.Size < treeSize && depth > 0 { - treeSize /= self.branches + treeSize /= reader.branches depth-- } @@ -449,8 +449,8 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr } wg.Add(1) go func(j int64) { - childKey := chunk.SData[8+j*self.hashSize : 8+(j+1)*self.hashSize] - chunk := retrieve(childKey, self.chunkC, quitC) + childKey := chunk.SData[8+j*reader.hashSize : 8+(j+1)*reader.hashSize] + chunk := retrieve(childKey, reader.chunkC, quitC) if chunk == nil { select { case errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize): @@ -461,7 +461,7 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr if soff < off { soff = off } - self.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/self.branches, chunk, wg, errC, quitC) + reader.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/reader.branches, chunk, wg, errC, quitC) }(i) } //for } @@ -496,10 +496,10 @@ func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk { } // Read keeps a cursor so cannot be called simulateously, see ReadAt -func (self *LazyChunkReader) Read(b []byte) (read int, err error) { - read, err = self.ReadAt(b, self.off) +func (reader *LazyChunkReader) Read(b []byte) (read int, err error) { + read, err = reader.ReadAt(b, reader.off) - self.off += int64(read) + reader.off += int64(read) return } @@ -507,27 +507,27 @@ func (self *LazyChunkReader) Read(b []byte) (read int, err error) { var errWhence = errors.New("Seek: invalid whence") var errOffset = errors.New("Seek: invalid offset") -func (s *LazyChunkReader) Seek(offset int64, whence int) (int64, error) { +func (reader *LazyChunkReader) Seek(offset int64, whence int) (int64, error) { switch whence { default: return 0, errWhence case 0: offset += 0 case 1: - offset += s.off + offset += reader.off case 2: - if s.chunk == nil { //seek from the end requires rootchunk for size. call Size first - _, err := s.Size(nil) + if reader.chunk == nil { //seek from the end requires rootchunk for size. call Size first + _, err := reader.Size(nil) if err != nil { return 0, fmt.Errorf("can't get size: %v", err) } } - offset += s.chunk.Size + offset += reader.chunk.Size } if offset < 0 { return 0, errOffset } - s.off = offset + reader.off = offset return offset, nil } diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go index c8bd12299a..f34a31fb27 100644 --- a/swarm/storage/chunker_test.go +++ b/swarm/storage/chunker_test.go @@ -45,12 +45,12 @@ type chunkerTester struct { t test } -func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, chunkC chan *Chunk, swg *sync.WaitGroup, expectedError error) (key Key, err error) { +func (tester *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, chunkC chan *Chunk, swg *sync.WaitGroup, expectedError error) (key Key, err error) { // reset - self.chunks = make(map[string]*Chunk) + tester.chunks = make(map[string]*Chunk) - if self.inputs == nil { - self.inputs = make(map[uint64][]byte) + if tester.inputs == nil { + tester.inputs = make(map[uint64][]byte) } quitC := make(chan bool) @@ -64,8 +64,8 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c case <-quitC: return nil case chunk := <-chunkC: - // self.chunks = append(self.chunks, chunk) - self.chunks[chunk.Key.String()] = chunk + // tester.chunks = append(tester.chunks, chunk) + tester.chunks[chunk.Key.String()] = chunk if chunk.wg != nil { chunk.wg.Done() } @@ -89,7 +89,7 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c return key, err } -func (self *chunkerTester) Append(chunker Splitter, rootKey Key, data io.Reader, chunkC chan *Chunk, swg *sync.WaitGroup, expectedError error) (key Key, err error) { +func (tester *chunkerTester) Append(chunker Splitter, rootKey Key, data io.Reader, chunkC chan *Chunk, swg *sync.WaitGroup, expectedError error) (key Key, err error) { quitC := make(chan bool) timeout := time.After(60 * time.Second) if chunkC != nil { @@ -102,10 +102,10 @@ func (self *chunkerTester) Append(chunker Splitter, rootKey Key, data io.Reader, return nil case chunk := <-chunkC: if chunk != nil { - stored, success := self.chunks[chunk.Key.String()] + stored, success := tester.chunks[chunk.Key.String()] if !success { // Requesting data - self.chunks[chunk.Key.String()] = chunk + tester.chunks[chunk.Key.String()] = chunk if chunk.wg != nil { chunk.wg.Done() } @@ -135,7 +135,7 @@ func (self *chunkerTester) Append(chunker Splitter, rootKey Key, data io.Reader, return key, err } -func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Chunk, quitC chan bool) LazySectionReader { +func (tester *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Chunk, quitC chan bool) LazySectionReader { // reset but not the chunks reader := chunker.Join(key, chunkC) @@ -153,7 +153,7 @@ func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Ch return nil } // this just mocks the behaviour of a chunk store retrieval - stored, success := self.chunks[chunk.Key.String()] + stored, success := tester.chunks[chunk.Key.String()] if !success { return errors.New("Not found") } diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index cd4c2ef139..07e2502bbb 100644 --- a/swarm/storage/common_test.go +++ b/swarm/storage/common_test.go @@ -46,12 +46,12 @@ func testDataReader(l int) (r io.Reader) { return io.LimitReader(rand.Reader, int64(l)) } -func (self *brokenLimitedReader) Read(buf []byte) (int, error) { - if self.off+len(buf) > self.errAt { +func (reader *brokenLimitedReader) Read(buf []byte) (int, error) { + if reader.off+len(buf) > reader.errAt { return 0, fmt.Errorf("Broken reader") } - self.off += len(buf) - return self.lr.Read(buf) + reader.off += len(buf) + return reader.lr.Read(buf) } func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) { diff --git a/swarm/storage/database.go b/swarm/storage/database.go index f2ceb94e46..acb976047a 100644 --- a/swarm/storage/database.go +++ b/swarm/storage/database.go @@ -45,27 +45,27 @@ func NewLDBDatabase(file string) (*LDBDatabase, error) { return database, nil } -func (self *LDBDatabase) Put(key []byte, value []byte) { - err := self.db.Put(key, value, nil) +func (db *LDBDatabase) Put(key []byte, value []byte) { + err := db.db.Put(key, value, nil) if err != nil { fmt.Println("Error put", err) } } -func (self *LDBDatabase) Get(key []byte) ([]byte, error) { - dat, err := self.db.Get(key, nil) +func (db *LDBDatabase) Get(key []byte) ([]byte, error) { + dat, err := db.db.Get(key, nil) if err != nil { return nil, err } return dat, nil } -func (self *LDBDatabase) Delete(key []byte) error { - return self.db.Delete(key, nil) +func (db *LDBDatabase) Delete(key []byte) error { + return db.db.Delete(key, nil) } -func (self *LDBDatabase) LastKnownTD() []byte { - data, _ := self.Get([]byte("LTD")) +func (db *LDBDatabase) LastKnownTD() []byte { + data, _ := db.Get([]byte("LTD")) if len(data) == 0 { data = []byte{0x0} @@ -74,15 +74,15 @@ func (self *LDBDatabase) LastKnownTD() []byte { return data } -func (self *LDBDatabase) NewIterator() iterator.Iterator { - return self.db.NewIterator(nil, nil) +func (db *LDBDatabase) NewIterator() iterator.Iterator { + return db.db.NewIterator(nil, nil) } -func (self *LDBDatabase) Write(batch *leveldb.Batch) error { - return self.db.Write(batch, nil) +func (db *LDBDatabase) Write(batch *leveldb.Batch) error { + return db.db.Write(batch, nil) } -func (self *LDBDatabase) Close() { +func (db *LDBDatabase) Close() { // Close the leveldb database - self.db.Close() + db.db.Close() } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 1ff42a0c05..b63a147723 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -201,12 +201,10 @@ func gcListSelect(list []*gcItem, left int, right int, n int) int { pivotIndex = gcListPartition(list, left, right, pivotIndex) if n == pivotIndex { return n + } else if n < pivotIndex { + return gcListSelect(list, left, pivotIndex-1, n) } else { - if n < pivotIndex { - return gcListSelect(list, left, pivotIndex-1, n) - } else { - return gcListSelect(list, pivotIndex+1, right, n) - } + return gcListSelect(list, pivotIndex+1, right, n) } } @@ -539,7 +537,7 @@ func (s *DbStore) Close() { s.db.Close() } -// describes a section of the DbStore representing the unsynced +// DbSyncState describes a section of the DbStore representing the unsynced // domain relevant to a peer // Start - Stop designate a continuous area Keys in an address space // typically the addresses closer to us than to the peer but not closer @@ -558,13 +556,13 @@ type dbSyncIterator struct { DbSyncState } -// initialises a sync iterator from a syncToken (passed in with the handshake) -func (self *DbStore) NewSyncIterator(state DbSyncState) (si *dbSyncIterator, err error) { +// NewSyncIterator initialises a sync iterator from a syncToken (passed in with the handshake) +func (s *DbStore) NewSyncIterator(state DbSyncState) (si *dbSyncIterator, err error) { if state.First > state.Last { return nil, fmt.Errorf("no entries found") } si = &dbSyncIterator{ - it: self.db.NewIterator(), + it: s.db.NewIterator(), DbSyncState: state, } si.it.Seek(getIndexKey(state.Start)) @@ -573,28 +571,28 @@ func (self *DbStore) NewSyncIterator(state DbSyncState) (si *dbSyncIterator, err // walk the area from Start to Stop and returns items within time interval // First to Last -func (self *dbSyncIterator) Next() (key Key) { - for self.it.Valid() { - dbkey := self.it.Key() +func (itr *dbSyncIterator) Next() (key Key) { + for itr.it.Valid() { + dbkey := itr.it.Key() if dbkey[0] != 0 { break } key = Key(make([]byte, len(dbkey)-1)) copy(key[:], dbkey[1:]) - if bytes.Compare(key[:], self.Start) <= 0 { - self.it.Next() + if bytes.Compare(key[:], itr.Start) <= 0 { + itr.it.Next() continue } - if bytes.Compare(key[:], self.Stop) > 0 { + if bytes.Compare(key[:], itr.Stop) > 0 { break } var index dpaDBIndex - decodeIndex(self.it.Value(), &index) - self.it.Next() - if (index.Idx >= self.First) && (index.Idx < self.Last) { + decodeIndex(itr.it.Value(), &index) + itr.it.Next() + if (index.Idx >= itr.First) && (index.Idx < itr.Last) { return } } - self.it.Release() + itr.it.Release() return nil } diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index 44a2669f12..c43d4540df 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -62,7 +62,7 @@ type DPA struct { quitC chan bool } -// for testing locally +// NewLocalDPA used for testing locally func NewLocalDPA(datadir string) (*DPA, error) { hash := MakeHashFunc("SHA256") @@ -90,53 +90,53 @@ func NewDPA(store ChunkStore, params *ChunkerParams) *DPA { // FS-aware API and httpaccess // Chunk retrieval blocks on netStore requests with a timeout so reader will // report error if retrieval of chunks within requested range time out. -func (self *DPA) Retrieve(key Key) LazySectionReader { - return self.Chunker.Join(key, self.retrieveC) +func (dpa *DPA) Retrieve(key Key) LazySectionReader { + return dpa.Chunker.Join(key, dpa.retrieveC) } // Public API. Main entry point for document storage directly. Used by the // FS-aware API and httpaccess -func (self *DPA) Store(data io.Reader, size int64, swg *sync.WaitGroup, wwg *sync.WaitGroup) (key Key, err error) { - return self.Chunker.Split(data, size, self.storeC, swg, wwg) +func (dpa *DPA) Store(data io.Reader, size int64, swg *sync.WaitGroup, wwg *sync.WaitGroup) (key Key, err error) { + return dpa.Chunker.Split(data, size, dpa.storeC, swg, wwg) } -func (self *DPA) Start() { - self.lock.Lock() - defer self.lock.Unlock() - if self.running { +func (dpa *DPA) Start() { + dpa.lock.Lock() + defer dpa.lock.Unlock() + if dpa.running { return } - self.running = true - self.retrieveC = make(chan *Chunk, retrieveChanCapacity) - self.storeC = make(chan *Chunk, storeChanCapacity) - self.quitC = make(chan bool) - self.storeLoop() - self.retrieveLoop() + dpa.running = true + dpa.retrieveC = make(chan *Chunk, retrieveChanCapacity) + dpa.storeC = make(chan *Chunk, storeChanCapacity) + dpa.quitC = make(chan bool) + dpa.storeLoop() + dpa.retrieveLoop() } -func (self *DPA) Stop() { - self.lock.Lock() - defer self.lock.Unlock() - if !self.running { +func (dpa *DPA) Stop() { + dpa.lock.Lock() + defer dpa.lock.Unlock() + if !dpa.running { return } - self.running = false - close(self.quitC) + dpa.running = false + close(dpa.quitC) } // retrieveLoop dispatches the parallel chunk retrieval requests received on the // retrieve channel to its ChunkStore (NetStore or LocalStore) -func (self *DPA) retrieveLoop() { +func (dpa *DPA) retrieveLoop() { for i := 0; i < maxRetrieveProcesses; i++ { - go self.retrieveWorker() + go dpa.retrieveWorker() } log.Trace(fmt.Sprintf("dpa: retrieve loop spawning %v workers", maxRetrieveProcesses)) } -func (self *DPA) retrieveWorker() { - for chunk := range self.retrieveC { +func (dpa *DPA) retrieveWorker() { + for chunk := range dpa.retrieveC { log.Trace(fmt.Sprintf("dpa: retrieve loop : chunk %v", chunk.Key.Log())) - storedChunk, err := self.Get(chunk.Key) + storedChunk, err := dpa.Get(chunk.Key) if err == notFound { log.Trace(fmt.Sprintf("chunk %v not found", chunk.Key.Log())) } else if err != nil { @@ -148,7 +148,7 @@ func (self *DPA) retrieveWorker() { close(chunk.C) select { - case <-self.quitC: + case <-dpa.quitC: return default: } @@ -157,24 +157,24 @@ func (self *DPA) retrieveWorker() { // storeLoop dispatches the parallel chunk store request processors // received on the store channel to its ChunkStore (NetStore or LocalStore) -func (self *DPA) storeLoop() { +func (dpa *DPA) storeLoop() { for i := 0; i < maxStoreProcesses; i++ { - go self.storeWorker() + go dpa.storeWorker() } log.Trace(fmt.Sprintf("dpa: store spawning %v workers", maxStoreProcesses)) } -func (self *DPA) storeWorker() { +func (dpa *DPA) storeWorker() { - for chunk := range self.storeC { - self.Put(chunk) + for chunk := range dpa.storeC { + dpa.Put(chunk) if chunk.wg != nil { log.Trace(fmt.Sprintf("dpa: store processor %v", chunk.Key.Log())) chunk.wg.Done() } select { - case <-self.quitC: + case <-dpa.quitC: return default: } @@ -198,14 +198,14 @@ func NewDpaChunkStore(localStore, netStore ChunkStore) *dpaChunkStore { // Get is the entrypoint for local retrieve requests // waits for response or times out -func (self *dpaChunkStore) Get(key Key) (chunk *Chunk, err error) { - chunk, err = self.netStore.Get(key) +func (s *dpaChunkStore) Get(key Key) (chunk *Chunk, err error) { + chunk, err = s.netStore.Get(key) // timeout := time.Now().Add(searchTimeout) if chunk.SData != nil { log.Trace(fmt.Sprintf("DPA.Get: %v found locally, %d bytes", key.Log(), len(chunk.SData))) return } - // TODO: use self.timer time.Timer and reset with defer disableTimer + // TODO: use s.timer time.Timer and reset with defer disableTimer timer := time.After(searchTimeout) select { case <-timer: @@ -218,8 +218,8 @@ func (self *dpaChunkStore) Get(key Key) (chunk *Chunk, err error) { } // Put is the entrypoint for local store requests coming from storeLoop -func (self *dpaChunkStore) Put(entry *Chunk) { - chunk, err := self.localStore.Get(entry.Key) +func (s *dpaChunkStore) Put(entry *Chunk) { + chunk, err := s.localStore.Get(entry.Key) if err != nil { log.Trace(fmt.Sprintf("DPA.Put: %v new chunk. call netStore.Put", entry.Key.Log())) chunk = entry @@ -232,10 +232,10 @@ func (self *dpaChunkStore) Put(entry *Chunk) { return } // from this point on the storage logic is the same with network storage requests - log.Trace(fmt.Sprintf("DPA.Put %v: %v", self.n, chunk.Key.Log())) - self.n++ - self.netStore.Put(chunk) + log.Trace(fmt.Sprintf("DPA.Put %v: %v", s.n, chunk.Key.Log())) + s.n++ + s.netStore.Put(chunk) } // Close chunk store -func (self *dpaChunkStore) Close() {} +func (s *dpaChunkStore) Close() {} diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index ece0c86155..6444e1d54d 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -34,7 +34,7 @@ type LocalStore struct { DbStore ChunkStore } -// This constructor uses MemStore and DbStore as components +// NewLocalStore constructor uses MemStore and DbStore as components func NewLocalStore(hash SwarmHasher, params *StoreParams) (*LocalStore, error) { dbStore, err := NewDbStore(params.ChunkDbPath, hash, params.DbCapacity, params.Radius) if err != nil { @@ -46,48 +46,48 @@ func NewLocalStore(hash SwarmHasher, params *StoreParams) (*LocalStore, error) { }, nil } -func (self *LocalStore) CacheCounter() uint64 { - return uint64(self.memStore.(*MemStore).Counter()) +func (s *LocalStore) CacheCounter() uint64 { + return uint64(s.memStore.(*MemStore).Counter()) } -func (self *LocalStore) DbCounter() uint64 { - return self.DbStore.(*DbStore).Counter() +func (s *LocalStore) DbCounter() uint64 { + return s.DbStore.(*DbStore).Counter() } // LocalStore is itself a chunk store // unsafe, in that the data is not integrity checked -func (self *LocalStore) Put(chunk *Chunk) { +func (s *LocalStore) Put(chunk *Chunk) { chunk.dbStored = make(chan bool) - self.memStore.Put(chunk) + s.memStore.Put(chunk) if chunk.wg != nil { chunk.wg.Add(1) } go func() { dbStorePutCounter.Inc(1) - self.DbStore.Put(chunk) + s.DbStore.Put(chunk) if chunk.wg != nil { chunk.wg.Done() } }() } -// Get(chunk *Chunk) looks up a chunk in the local stores +// Get looks up a chunk in the local stores // This method is blocking until the chunk is retrieved // so additional timeout may be needed to wrap this call if // ChunkStores are remote and can have long latency -func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { - chunk, err = self.memStore.Get(key) +func (s *LocalStore) Get(key Key) (chunk *Chunk, err error) { + chunk, err = s.memStore.Get(key) if err == nil { return } - chunk, err = self.DbStore.Get(key) + chunk, err = s.DbStore.Get(key) if err != nil { return } chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - self.memStore.Put(chunk) + s.memStore.Put(chunk) return } // Close local store -func (self *LocalStore) Close() {} +func (s *LocalStore) Close() {} diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go index 0552b84efa..f6a710efbd 100644 --- a/swarm/storage/netstore.go +++ b/swarm/storage/netstore.go @@ -41,7 +41,7 @@ type NetStore struct { cloud CloudStore } -// backend engine for cloud store +// CloudStore backend engine // It can be aggregate dispatching to several parallel implementations: // bzz/network/forwarder. forwarder or IPFS or IPΞS type CloudStore interface { @@ -58,7 +58,7 @@ type StoreParams struct { } //create params with default values -func NewDefaultStoreParams() (self *StoreParams) { +func NewDefaultStoreParams() (params *StoreParams) { return &StoreParams{ DbCapacity: defaultDbCapacity, CacheCapacity: defaultCacheCapacity, @@ -68,11 +68,11 @@ func NewDefaultStoreParams() (self *StoreParams) { //this can only finally be set after all config options (file, cmd line, env vars) //have been evaluated -func (self *StoreParams) Init(path string) { - self.ChunkDbPath = filepath.Join(path, "chunks") +func (params *StoreParams) Init(path string) { + params.ChunkDbPath = filepath.Join(path, "chunks") } -// netstore contructor, takes path argument that is used to initialise dbStore, +// NewNetStore contructs NetStore, takes path argument that is used to initialise dbStore, // the persistent (disk) storage component of LocalStore // the second argument is the hive, the connection/logistics manager for the node func NewNetStore(hash SwarmHasher, lstore *LocalStore, cloud CloudStore, params *StoreParams) *NetStore { @@ -92,8 +92,8 @@ var ( // ~ unsafe put in localdb no check if exists no extra copy no hash validation // the chunk is forced to propagate (Cloud.Store) even if locally found! // caller needs to make sure if that is wanted -func (self *NetStore) Put(entry *Chunk) { - self.localStore.Put(entry) +func (s *NetStore) Put(entry *Chunk) { + s.localStore.Put(entry) // handle deliveries if entry.Req != nil { @@ -102,19 +102,19 @@ func (self *NetStore) Put(entry *Chunk) { // that the chunk is has been retrieved close(entry.Req.C) // deliver the chunk to requesters upstream - go self.cloud.Deliver(entry) + go s.cloud.Deliver(entry) } else { log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v stored locally", entry.Key.Log())) // handle propagating store requests - // go self.cloud.Store(entry) - go self.cloud.Store(entry) + // go s.cloud.Store(entry) + go s.cloud.Store(entry) } } -// retrieve logic common for local and network chunk retrieval requests -func (self *NetStore) Get(key Key) (*Chunk, error) { +// Get logic common for local and network chunk retrieval requests +func (s *NetStore) Get(key Key) (*Chunk, error) { var err error - chunk, err := self.localStore.Get(key) + chunk, err := s.localStore.Get(key) if err == nil { if chunk.Req == nil { log.Trace(fmt.Sprintf("NetStore.Get: %v found locally", key)) @@ -127,10 +127,10 @@ func (self *NetStore) Get(key Key) (*Chunk, error) { // no data and no request status log.Trace(fmt.Sprintf("NetStore.Get: %v not found locally. open new request", key)) chunk = NewChunk(key, newRequestStatus(key)) - self.localStore.memStore.Put(chunk) - go self.cloud.Retrieve(chunk) + s.localStore.memStore.Put(chunk) + go s.cloud.Retrieve(chunk) return chunk, nil } // Close netstore -func (self *NetStore) Close() {} +func (s *NetStore) Close() {} diff --git a/swarm/storage/pyramid.go b/swarm/storage/pyramid.go index 19d493405a..493f9d6196 100644 --- a/swarm/storage/pyramid.go +++ b/swarm/storage/pyramid.go @@ -136,44 +136,44 @@ func NewPyramidChunker(params *ChunkerParams) (self *PyramidChunker) { return } -func (self *PyramidChunker) Join(key Key, chunkC chan *Chunk) LazySectionReader { +func (c *PyramidChunker) Join(key Key, chunkC chan *Chunk) LazySectionReader { return &LazyChunkReader{ key: key, chunkC: chunkC, - chunkSize: self.chunkSize, - branches: self.branches, - hashSize: self.hashSize, + chunkSize: c.chunkSize, + branches: c.branches, + hashSize: c.hashSize, } } -func (self *PyramidChunker) incrementWorkerCount() { - self.workerLock.Lock() - defer self.workerLock.Unlock() - self.workerCount += 1 +func (c *PyramidChunker) incrementWorkerCount() { + c.workerLock.Lock() + defer c.workerLock.Unlock() + c.workerCount++ } -func (self *PyramidChunker) getWorkerCount() int64 { - self.workerLock.Lock() - defer self.workerLock.Unlock() - return self.workerCount +func (c *PyramidChunker) getWorkerCount() int64 { + c.workerLock.Lock() + defer c.workerLock.Unlock() + return c.workerCount } -func (self *PyramidChunker) decrementWorkerCount() { - self.workerLock.Lock() - defer self.workerLock.Unlock() - self.workerCount -= 1 +func (c *PyramidChunker) decrementWorkerCount() { + c.workerLock.Lock() + defer c.workerLock.Unlock() + c.workerCount-- } -func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, storageWG, processorWG *sync.WaitGroup) (Key, error) { +func (c *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, storageWG, processorWG *sync.WaitGroup) (Key, error) { jobC := make(chan *chunkJob, 2*ChunkProcessors) wg := &sync.WaitGroup{} errC := make(chan error) quitC := make(chan bool) - rootKey := make([]byte, self.hashSize) - chunkLevel := make([][]*TreeEntry, self.branches) + rootKey := make([]byte, c.hashSize) + chunkLevel := make([][]*TreeEntry, c.branches) wg.Add(1) - go self.prepareChunks(false, chunkLevel, data, rootKey, quitC, wg, jobC, processorWG, chunkC, errC, storageWG) + go c.prepareChunks(false, chunkLevel, data, rootKey, quitC, wg, jobC, processorWG, chunkC, errC, storageWG) // closes internal error channel if all subprocesses in the workgroup finished go func() { @@ -204,20 +204,20 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk } -func (self *PyramidChunker) Append(key Key, data io.Reader, chunkC chan *Chunk, storageWG, processorWG *sync.WaitGroup) (Key, error) { +func (c *PyramidChunker) Append(key Key, data io.Reader, chunkC chan *Chunk, storageWG, processorWG *sync.WaitGroup) (Key, error) { quitC := make(chan bool) - rootKey := make([]byte, self.hashSize) - chunkLevel := make([][]*TreeEntry, self.branches) + rootKey := make([]byte, c.hashSize) + chunkLevel := make([][]*TreeEntry, c.branches) // Load the right most unfinished tree chunks in every level - self.loadTree(chunkLevel, key, chunkC, quitC) + c.loadTree(chunkLevel, key, chunkC, quitC) jobC := make(chan *chunkJob, 2*ChunkProcessors) wg := &sync.WaitGroup{} errC := make(chan error) wg.Add(1) - go self.prepareChunks(true, chunkLevel, data, rootKey, quitC, wg, jobC, processorWG, chunkC, errC, storageWG) + go c.prepareChunks(true, chunkLevel, data, rootKey, quitC, wg, jobC, processorWG, chunkC, errC, storageWG) // closes internal error channel if all subprocesses in the workgroup finished go func() { @@ -245,10 +245,10 @@ func (self *PyramidChunker) Append(key Key, data io.Reader, chunkC chan *Chunk, } -func (self *PyramidChunker) processor(id int64, jobC chan *chunkJob, chunkC chan *Chunk, errC chan error, quitC chan bool, swg, wwg *sync.WaitGroup) { - defer self.decrementWorkerCount() +func (c *PyramidChunker) processor(id int64, jobC chan *chunkJob, chunkC chan *Chunk, errC chan error, quitC chan bool, swg, wwg *sync.WaitGroup) { + defer c.decrementWorkerCount() - hasher := self.hashFunc() + hasher := c.hashFunc() if wwg != nil { defer wwg.Done() } @@ -259,14 +259,14 @@ func (self *PyramidChunker) processor(id int64, jobC chan *chunkJob, chunkC chan if !ok { return } - self.processChunk(id, hasher, job, chunkC, swg) + c.processChunk(id, hasher, job, chunkC, swg) case <-quitC: return } } } -func (self *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJob, chunkC chan *Chunk, swg *sync.WaitGroup) { +func (c *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJob, chunkC chan *Chunk, swg *sync.WaitGroup) { hasher.ResetWithLength(job.chunk[:8]) // 8 bytes of length hasher.Write(job.chunk[8:]) // minus 8 []byte length h := hasher.Sum(nil) @@ -294,7 +294,7 @@ func (self *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJ } } -func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC chan *Chunk, quitC chan bool) error { +func (c *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC chan *Chunk, quitC chan bool) error { // Get the root chunk to get the total size chunk := retrieve(key, chunkC, quitC) if chunk == nil { @@ -302,13 +302,13 @@ func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC } //if data size is less than a chunk... add a parent with update as pending - if chunk.Size <= self.chunkSize { + if chunk.Size <= c.chunkSize { newEntry := &TreeEntry{ level: 0, branchCount: 1, subtreeSize: uint64(chunk.Size), - chunk: make([]byte, self.chunkSize+8), - key: make([]byte, self.hashSize), + chunk: make([]byte, c.chunkSize+8), + key: make([]byte, c.hashSize), index: 0, updatePending: true, } @@ -319,13 +319,13 @@ func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC var treeSize int64 var depth int - treeSize = self.chunkSize - for ; treeSize < chunk.Size; treeSize *= self.branches { + treeSize = c.chunkSize + for ; treeSize < chunk.Size; treeSize *= c.branches { depth++ } // Add the root chunk entry - branchCount := int64(len(chunk.SData)-8) / self.hashSize + branchCount := int64(len(chunk.SData)-8) / c.hashSize newEntry := &TreeEntry{ level: depth - 1, branchCount: branchCount, @@ -343,14 +343,14 @@ func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC //TODO(jmozah): instead of loading finished branches and then trim in the end, //avoid loading them in the first place for _, ent := range chunkLevel[lvl] { - branchCount = int64(len(ent.chunk)-8) / self.hashSize + branchCount = int64(len(ent.chunk)-8) / c.hashSize for i := int64(0); i < branchCount; i++ { - key := ent.chunk[8+(i*self.hashSize) : 8+((i+1)*self.hashSize)] + key := ent.chunk[8+(i*c.hashSize) : 8+((i+1)*c.hashSize)] newChunk := retrieve(key, chunkC, quitC) if newChunk == nil { return errLoadingTreeChunk } - bewBranchCount := int64(len(newChunk.SData)-8) / self.hashSize + bewBranchCount := int64(len(newChunk.SData)-8) / c.hashSize newEntry := &TreeEntry{ level: lvl - 1, branchCount: bewBranchCount, @@ -365,7 +365,7 @@ func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC } // We need to get only the right most unfinished branch.. so trim all finished branches - if int64(len(chunkLevel[lvl-1])) >= self.branches { + if int64(len(chunkLevel[lvl-1])) >= c.branches { chunkLevel[lvl-1] = nil } } @@ -374,7 +374,7 @@ func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC return nil } -func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEntry, data io.Reader, rootKey []byte, quitC chan bool, wg *sync.WaitGroup, jobC chan *chunkJob, processorWG *sync.WaitGroup, chunkC chan *Chunk, errC chan error, storageWG *sync.WaitGroup) { +func (c *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEntry, data io.Reader, rootKey []byte, quitC chan bool, wg *sync.WaitGroup, jobC chan *chunkJob, processorWG *sync.WaitGroup, chunkC chan *Chunk, errC chan error, storageWG *sync.WaitGroup) { defer wg.Done() chunkWG := &sync.WaitGroup{} @@ -385,10 +385,10 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt processorWG.Add(1) } - self.incrementWorkerCount() - go self.processor(self.workerCount, jobC, chunkC, errC, quitC, storageWG, processorWG) + c.incrementWorkerCount() + go c.processor(c.workerCount, jobC, chunkC, errC, quitC, storageWG, processorWG) - parent := NewTreeEntry(self) + parent := NewTreeEntry(c) var unFinishedChunk *Chunk if isAppend && len(chunkLevel[0]) != 0 { @@ -396,7 +396,7 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt lastIndex := len(chunkLevel[0]) - 1 ent := chunkLevel[0][lastIndex] - if ent.branchCount < self.branches { + if ent.branchCount < c.branches { parent = &TreeEntry{ level: 0, branchCount: ent.branchCount, @@ -408,10 +408,10 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt } lastBranch := parent.branchCount - 1 - lastKey := parent.chunk[8+lastBranch*self.hashSize : 8+(lastBranch+1)*self.hashSize] + lastKey := parent.chunk[8+lastBranch*c.hashSize : 8+(lastBranch+1)*c.hashSize] unFinishedChunk = retrieve(lastKey, chunkC, quitC) - if unFinishedChunk.Size < self.chunkSize { + if unFinishedChunk.Size < c.chunkSize { parent.subtreeSize = parent.subtreeSize - uint64(unFinishedChunk.Size) parent.branchCount = parent.branchCount - 1 @@ -425,7 +425,7 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt var n int var err error - chunkData := make([]byte, self.chunkSize+8) + chunkData := make([]byte, c.chunkSize+8) if unFinishedChunk != nil { copy(chunkData, unFinishedChunk.SData) n, err = data.Read(chunkData[8+unFinishedChunk.Size:]) @@ -441,7 +441,7 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt if parent.branchCount == 1 { // Data is exactly one chunk.. pick the last chunk key as root chunkWG.Wait() - lastChunksKey := parent.chunk[8 : 8+self.hashSize] + lastChunksKey := parent.chunk[8 : 8+c.hashSize] copy(rootKey, lastChunksKey) break } @@ -453,18 +453,18 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt // Data ended in chunk boundary.. just signal to start bulding tree if n == 0 { - self.buildTree(isAppend, chunkLevel, parent, chunkWG, jobC, quitC, true, rootKey) + c.buildTree(isAppend, chunkLevel, parent, chunkWG, jobC, quitC, true, rootKey) break } else { - pkey := self.enqueueDataChunk(chunkData, uint64(n), parent, chunkWG, jobC, quitC) + pkey := c.enqueueDataChunk(chunkData, uint64(n), parent, chunkWG, jobC, quitC) // update tree related parent data structures parent.subtreeSize += uint64(n) parent.branchCount++ // Data got exhausted... signal to send any parent tree related chunks - if int64(n) < self.chunkSize { + if int64(n) < c.chunkSize { // only one data chunk .. so dont add any parent chunk if parent.branchCount <= 1 { @@ -473,39 +473,39 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt break } - self.buildTree(isAppend, chunkLevel, parent, chunkWG, jobC, quitC, true, rootKey) + c.buildTree(isAppend, chunkLevel, parent, chunkWG, jobC, quitC, true, rootKey) break } - if parent.branchCount == self.branches { - self.buildTree(isAppend, chunkLevel, parent, chunkWG, jobC, quitC, false, rootKey) - parent = NewTreeEntry(self) + if parent.branchCount == c.branches { + c.buildTree(isAppend, chunkLevel, parent, chunkWG, jobC, quitC, false, rootKey) + parent = NewTreeEntry(c) } } - workers := self.getWorkerCount() + workers := c.getWorkerCount() if int64(len(jobC)) > workers && workers < ChunkProcessors { if processorWG != nil { processorWG.Add(1) } - self.incrementWorkerCount() - go self.processor(self.workerCount, jobC, chunkC, errC, quitC, storageWG, processorWG) + c.incrementWorkerCount() + go c.processor(c.workerCount, jobC, chunkC, errC, quitC, storageWG, processorWG) } } } -func (self *PyramidChunker) buildTree(isAppend bool, chunkLevel [][]*TreeEntry, ent *TreeEntry, chunkWG *sync.WaitGroup, jobC chan *chunkJob, quitC chan bool, last bool, rootKey []byte) { +func (c *PyramidChunker) buildTree(isAppend bool, chunkLevel [][]*TreeEntry, ent *TreeEntry, chunkWG *sync.WaitGroup, jobC chan *chunkJob, quitC chan bool, last bool, rootKey []byte) { chunkWG.Wait() - self.enqueueTreeChunk(chunkLevel, ent, chunkWG, jobC, quitC, last) + c.enqueueTreeChunk(chunkLevel, ent, chunkWG, jobC, quitC, last) compress := false - endLvl := self.branches - for lvl := int64(0); lvl < self.branches; lvl++ { + endLvl := c.branches + for lvl := int64(0); lvl < c.branches; lvl++ { lvlCount := int64(len(chunkLevel[lvl])) - if lvlCount >= self.branches { + if lvlCount >= c.branches { endLvl = lvl + 1 compress = true break @@ -527,9 +527,9 @@ func (self *PyramidChunker) buildTree(isAppend bool, chunkLevel [][]*TreeEntry, return } - for startCount := int64(0); startCount < lvlCount; startCount += self.branches { + for startCount := int64(0); startCount < lvlCount; startCount += c.branches { - endCount := startCount + self.branches + endCount := startCount + c.branches if endCount > lvlCount { endCount = lvlCount } @@ -545,18 +545,18 @@ func (self *PyramidChunker) buildTree(isAppend bool, chunkLevel [][]*TreeEntry, level: int(lvl + 1), branchCount: 0, subtreeSize: 0, - chunk: make([]byte, self.chunkSize+8), - key: make([]byte, self.hashSize), + chunk: make([]byte, c.chunkSize+8), + key: make([]byte, c.hashSize), index: int(nextLvlCount), updatePending: true, } for index := int64(0); index < lvlCount; index++ { updateEntry.branchCount++ updateEntry.subtreeSize += chunkLevel[lvl][index].subtreeSize - copy(updateEntry.chunk[8+(index*self.hashSize):8+((index+1)*self.hashSize)], chunkLevel[lvl][index].key[:self.hashSize]) + copy(updateEntry.chunk[8+(index*c.hashSize):8+((index+1)*c.hashSize)], chunkLevel[lvl][index].key[:c.hashSize]) } - self.enqueueTreeChunk(chunkLevel, updateEntry, chunkWG, jobC, quitC, last) + c.enqueueTreeChunk(chunkLevel, updateEntry, chunkWG, jobC, quitC, last) } else { @@ -565,8 +565,8 @@ func (self *PyramidChunker) buildTree(isAppend bool, chunkLevel [][]*TreeEntry, level: int(lvl + 1), branchCount: noOfBranches, subtreeSize: 0, - chunk: make([]byte, (noOfBranches*self.hashSize)+8), - key: make([]byte, self.hashSize), + chunk: make([]byte, (noOfBranches*c.hashSize)+8), + key: make([]byte, c.hashSize), index: int(nextLvlCount), updatePending: false, } @@ -575,11 +575,11 @@ func (self *PyramidChunker) buildTree(isAppend bool, chunkLevel [][]*TreeEntry, for i := startCount; i < endCount; i++ { entry := chunkLevel[lvl][i] newEntry.subtreeSize += entry.subtreeSize - copy(newEntry.chunk[8+(index*self.hashSize):8+((index+1)*self.hashSize)], entry.key[:self.hashSize]) + copy(newEntry.chunk[8+(index*c.hashSize):8+((index+1)*c.hashSize)], entry.key[:c.hashSize]) index++ } - self.enqueueTreeChunk(chunkLevel, newEntry, chunkWG, jobC, quitC, last) + c.enqueueTreeChunk(chunkLevel, newEntry, chunkWG, jobC, quitC, last) } @@ -595,7 +595,7 @@ func (self *PyramidChunker) buildTree(isAppend bool, chunkLevel [][]*TreeEntry, } -func (self *PyramidChunker) enqueueTreeChunk(chunkLevel [][]*TreeEntry, ent *TreeEntry, chunkWG *sync.WaitGroup, jobC chan *chunkJob, quitC chan bool, last bool) { +func (c *PyramidChunker) enqueueTreeChunk(chunkLevel [][]*TreeEntry, ent *TreeEntry, chunkWG *sync.WaitGroup, jobC chan *chunkJob, quitC chan bool, last bool) { if ent != nil { // wait for data chunks to get over before processing the tree chunk @@ -604,10 +604,10 @@ func (self *PyramidChunker) enqueueTreeChunk(chunkLevel [][]*TreeEntry, ent *Tre } binary.LittleEndian.PutUint64(ent.chunk[:8], ent.subtreeSize) - ent.key = make([]byte, self.hashSize) + ent.key = make([]byte, c.hashSize) chunkWG.Add(1) select { - case jobC <- &chunkJob{ent.key, ent.chunk[:ent.branchCount*self.hashSize+8], int64(ent.subtreeSize), chunkWG, TreeChunk, 0}: + case jobC <- &chunkJob{ent.key, ent.chunk[:ent.branchCount*c.hashSize+8], int64(ent.subtreeSize), chunkWG, TreeChunk, 0}: case <-quitC: } @@ -622,9 +622,9 @@ func (self *PyramidChunker) enqueueTreeChunk(chunkLevel [][]*TreeEntry, ent *Tre } } -func (self *PyramidChunker) enqueueDataChunk(chunkData []byte, size uint64, parent *TreeEntry, chunkWG *sync.WaitGroup, jobC chan *chunkJob, quitC chan bool) Key { +func (c *PyramidChunker) enqueueDataChunk(chunkData []byte, size uint64, parent *TreeEntry, chunkWG *sync.WaitGroup, jobC chan *chunkJob, quitC chan bool) Key { binary.LittleEndian.PutUint64(chunkData[:8], size) - pkey := parent.chunk[8+parent.branchCount*self.hashSize : 8+(parent.branchCount+1)*self.hashSize] + pkey := parent.chunk[8+parent.branchCount*c.hashSize : 8+(parent.branchCount+1)*c.hashSize] chunkWG.Add(1) select { diff --git a/swarm/storage/swarmhasher.go b/swarm/storage/swarmhasher.go index 38b86373c5..4b1ffc3901 100644 --- a/swarm/storage/swarmhasher.go +++ b/swarm/storage/swarmhasher.go @@ -34,7 +34,7 @@ type HashWithLength struct { hash.Hash } -func (self *HashWithLength) ResetWithLength(length []byte) { - self.Reset() - self.Write(length) +func (len *HashWithLength) ResetWithLength(length []byte) { + len.Reset() + len.Write(length) } diff --git a/swarm/storage/types.go b/swarm/storage/types.go index d35f1f9294..5befaa3cf9 100644 --- a/swarm/storage/types.go +++ b/swarm/storage/types.go @@ -243,6 +243,6 @@ type LazyTestSectionReader struct { *io.SectionReader } -func (self *LazyTestSectionReader) Size(chan bool) (int64, error) { - return self.SectionReader.Size(), nil +func (r *LazyTestSectionReader) Size(chan bool) (int64, error) { + return r.SectionReader.Size(), nil } diff --git a/swarm/swarm.go b/swarm/swarm.go index 0a120db1fa..6f36ca8ece 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -57,7 +57,7 @@ var ( cacheSizeGauge = metrics.NewRegisteredGauge("storage.db.cache.size", nil) ) -// the swarm stack +// Swarm stack type Swarm struct { config *api.Config // swarm configuration api *api.Api // high level api layer (fs/manifest) @@ -82,15 +82,15 @@ type SwarmAPI struct { PrvKey *ecdsa.PrivateKey } -func (self *Swarm) API() *SwarmAPI { +func (s *Swarm) API() *SwarmAPI { return &SwarmAPI{ - Api: self.api, - Backend: self.backend, - PrvKey: self.privateKey, + Api: s.api, + Backend: s.backend, + PrvKey: s.privateKey, } } -// creates a new swarm service instance +// NewSwarm creates a new swarm service instance // implements node.Service func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.Config) (self *Swarm, err error) { if bytes.Equal(common.FromHex(config.PublicKey), storage.ZeroKey) { @@ -272,7 +272,7 @@ Start is called when the stack is started * TODO: start subservices like sword, swear, swarmdns */ // implements the node.Service interface -func (self *Swarm) Start(srv *p2p.Server) error { +func (s *Swarm) Start(srv *p2p.Server) error { startTime = time.Now() connectPeer := func(url string) error { node, err := discover.ParseNode(url) @@ -283,119 +283,120 @@ func (self *Swarm) Start(srv *p2p.Server) error { return nil } // set chequebook - if self.swapEnabled { + if s.swapEnabled { ctx := context.Background() // The initial setup has no deadline. - err := self.SetChequebook(ctx) + err := s.SetChequebook(ctx) if err != nil { return fmt.Errorf("Unable to set chequebook for SWAP: %v", err) } - log.Debug(fmt.Sprintf("-> cheque book for SWAP: %v", self.config.Swap.Chequebook())) + log.Debug(fmt.Sprintf("-> cheque book for SWAP: %v", s.config.Swap.Chequebook())) } else { log.Debug(fmt.Sprintf("SWAP disabled: no cheque book set")) } log.Warn(fmt.Sprintf("Starting Swarm service")) - self.hive.Start( + s.hive.Start( discover.PubkeyID(&srv.PrivateKey.PublicKey), func() string { return srv.ListenAddr }, connectPeer, ) - log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.Addr())) + log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", s.hive.Addr())) - self.dpa.Start() + s.dpa.Start() log.Debug(fmt.Sprintf("Swarm DPA started")) // start swarm http proxy server - if self.config.Port != "" { - addr := net.JoinHostPort(self.config.ListenAddr, self.config.Port) - go httpapi.StartHttpServer(self.api, &httpapi.ServerConfig{ + if s.config.Port != "" { + addr := net.JoinHostPort(s.config.ListenAddr, s.config.Port) + go httpapi.StartHttpServer(s.api, &httpapi.ServerConfig{ Addr: addr, - CorsString: self.corsString, + CorsString: s.corsString, }) log.Info(fmt.Sprintf("Swarm http proxy started on %v", addr)) - if self.corsString != "" { - log.Debug(fmt.Sprintf("Swarm http proxy started with corsdomain: %v", self.corsString)) + if s.corsString != "" { + log.Debug(fmt.Sprintf("Swarm http proxy started with corsdomain: %v", s.corsString)) } } - self.periodicallyUpdateGauges() + s.periodicallyUpdateGauges() startCounter.Inc(1) return nil } -func (self *Swarm) periodicallyUpdateGauges() { +func (s *Swarm) periodicallyUpdateGauges() { ticker := time.NewTicker(updateGaugesPeriod) go func() { for range ticker.C { - self.updateGauges() + s.updateGauges() } }() } -func (self *Swarm) updateGauges() { - dbSizeGauge.Update(int64(self.lstore.DbCounter())) - cacheSizeGauge.Update(int64(self.lstore.CacheCounter())) +func (s *Swarm) updateGauges() { + dbSizeGauge.Update(int64(s.lstore.DbCounter())) + cacheSizeGauge.Update(int64(s.lstore.CacheCounter())) uptimeGauge.Update(time.Since(startTime).Nanoseconds()) } -// implements the node.Service interface +// Stop implements the node.Service interface // stops all component services. -func (self *Swarm) Stop() error { - self.dpa.Stop() - err := self.hive.Stop() - if ch := self.config.Swap.Chequebook(); ch != nil { +func (s *Swarm) Stop() error { + s.dpa.Stop() + err := s.hive.Stop() + if ch := s.config.Swap.Chequebook(); ch != nil { ch.Stop() ch.Save() } - if self.lstore != nil { - self.lstore.DbStore.Close() + if s.lstore != nil { + s.lstore.DbStore.Close() } - self.sfs.Stop() + s.sfs.Stop() stopCounter.Inc(1) return err } -// implements the node.Service interface -func (self *Swarm) Protocols() []p2p.Protocol { - proto, err := network.Bzz(self.depo, self.backend, self.hive, self.dbAccess, self.config.Swap, self.config.SyncParams, self.config.NetworkId) +// Protocols implements the node.Service interface +func (s *Swarm) Protocols() []p2p.Protocol { + proto, err := network.Bzz(s.depo, s.backend, s.hive, s.dbAccess, s.config.Swap, s.config.SyncParams, s.config.NetworkId) if err != nil { return nil } return []p2p.Protocol{proto} } + +// APIs returns the RPC Api descriptors the Swarm implementation offers // implements node.Service -// Apis returns the RPC Api descriptors the Swarm implementation offers -func (self *Swarm) APIs() []rpc.API { +func (s *Swarm) APIs() []rpc.API { return []rpc.API{ // public APIs { Namespace: "bzz", Version: "0.1", - Service: &Info{self.config, chequebook.ContractParams}, + Service: &Info{s.config, chequebook.ContractParams}, Public: true, }, // admin APIs { Namespace: "bzz", Version: "0.1", - Service: api.NewControl(self.api, self.hive), + Service: api.NewControl(s.api, s.hive), Public: false, }, { Namespace: "chequebook", Version: chequebook.Version, - Service: chequebook.NewApi(self.config.Swap.Chequebook), + Service: chequebook.NewApi(s.config.Swap.Chequebook), Public: false, }, { Namespace: "swarmfs", Version: fuse.Swarmfs_Version, - Service: self.sfs, + Service: s.sfs, Public: false, }, // storage APIs @@ -403,35 +404,35 @@ func (self *Swarm) APIs() []rpc.API { { Namespace: "bzz", Version: "0.1", - Service: api.NewStorage(self.api), + Service: api.NewStorage(s.api), Public: true, }, { Namespace: "bzz", Version: "0.1", - Service: api.NewFileSystem(self.api), + Service: api.NewFileSystem(s.api), Public: false, }, - // {Namespace, Version, api.NewAdmin(self), false}, + // {Namespace, Version, api.NewAdmin(s), false}, } } -func (self *Swarm) Api() *api.Api { - return self.api +func (s *Swarm) Api() *api.Api { + return s.api } // SetChequebook ensures that the local checquebook is set up on chain. -func (self *Swarm) SetChequebook(ctx context.Context) error { - err := self.config.Swap.SetChequebook(ctx, self.backend, self.config.Path) +func (s *Swarm) SetChequebook(ctx context.Context) error { + err := s.config.Swap.SetChequebook(ctx, s.backend, s.config.Path) if err != nil { return err } - log.Info(fmt.Sprintf("new chequebook set (%v): saving config file, resetting all connections in the hive", self.config.Swap.Contract.Hex())) - self.hive.DropAll() + log.Info(fmt.Sprintf("new chequebook set (%v): saving config file, resetting all connections in the hive", s.config.Swap.Contract.Hex())) + s.hive.DropAll() return nil } -// Local swarm without netStore +// NewLocalSwarm without netStore func NewLocalSwarm(datadir, port string) (self *Swarm, err error) { prvKey, err := crypto.GenerateKey() @@ -463,6 +464,6 @@ type Info struct { *chequebook.Params } -func (self *Info) Info() *Info { - return self +func (info *Info) Info() *Info { + return info }