From 4b6892c6270c689b80927bafa8e4070927a8467c Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Wed, 10 Oct 2018 12:42:14 -0500 Subject: [PATCH] swarm/swap: leaner first phase, bug fixes and PR comments --- metrics/counter.go | 34 ++++- p2p/protocols/accounting.go | 72 +++++++--- swarm/network/stream/stream.go | 20 +-- swarm/swap/api.go | 108 --------------- swarm/swap/chequemanager.go | 63 --------- swarm/swap/protocol.go | 203 ---------------------------- swarm/swap/swap.go | 139 +------------------- swarm/swap/swap_test.go | 233 ++------------------------------- swarm/swap_test.go | 29 ++-- swarm/swarm.go | 7 - 10 files changed, 122 insertions(+), 786 deletions(-) delete mode 100644 swarm/swap/api.go delete mode 100644 swarm/swap/chequemanager.go delete mode 100644 swarm/swap/protocol.go diff --git a/metrics/counter.go b/metrics/counter.go index c7f2b4bd3a..2f78c90d5c 100644 --- a/metrics/counter.go +++ b/metrics/counter.go @@ -1,6 +1,8 @@ package metrics -import "sync/atomic" +import ( + "sync/atomic" +) // Counters hold an int64 value that can be incremented and decremented. type Counter interface { @@ -20,6 +22,17 @@ func GetOrRegisterCounter(name string, r Registry) Counter { return r.GetOrRegister(name, NewCounter).(Counter) } +// GetOrRegisterCounterForced returns an existing Counter or constructs and registers a +// new Counter no matter the global switch is enabled or not. +// Be sure to unregister the counter from the registry once it is of no use to +// allow for garbage collection. +func GetOrRegisterCounterForced(name string, r Registry) Counter { + if nil == r { + r = DefaultRegistry + } + return r.GetOrRegister(name, NewCounterForced).(Counter) +} + // NewCounter constructs a new StandardCounter. func NewCounter() Counter { if !Enabled { @@ -28,6 +41,12 @@ func NewCounter() Counter { return &StandardCounter{0} } +// NewCounterForced constructs a new StandardCounter and returns it no matter if +// the global switch is enabled or not. +func NewCounterForced() Counter { + return &StandardCounter{0} +} + // NewRegisteredCounter constructs and registers a new StandardCounter. func NewRegisteredCounter(name string, r Registry) Counter { c := NewCounter() @@ -38,6 +57,19 @@ func NewRegisteredCounter(name string, r Registry) Counter { return c } +// NewRegisteredCounterForced constructs and registers a new StandardCounter +// and launches a goroutine no matter the global switch is enabled or not. +// Be sure to unregister the counter from the registry once it is of no use to +// allow for garbage collection. +func NewRegisteredCounterForced(name string, r Registry) Counter { + c := NewCounterForced() + if nil == r { + r = DefaultRegistry + } + r.Register(name, c) + return c +} + // CounterSnapshot is a read-only copy of another Counter. type CounterSnapshot int64 diff --git a/p2p/protocols/accounting.go b/p2p/protocols/accounting.go index 3395588d62..2319b6a99b 100644 --- a/p2p/protocols/accounting.go +++ b/p2p/protocols/accounting.go @@ -18,25 +18,41 @@ package protocols import ( "sync" + + "github.com/ethereum/go-ethereum/metrics" +) + +var ( + //NOTE: these metrics just define the interfaces and are currently *NOT persisted* over sessions + mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", nil) + mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", nil) + mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", nil) + mBytesDebit = metrics.NewRegisteredCounterForced("account.bytes.debit", nil) + mMsgCredit = metrics.NewRegisteredCounterForced("account.msg.credit", nil) + mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", nil) + mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", nil) + mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", nil) + mChequesIssued = metrics.NewRegisteredCounterForced("account.cheques.issued", nil) + mChequesReceived = metrics.NewRegisteredCounterForced("account.cheques.received", nil) ) type PriceOracle interface { Price(uint32, interface{}) (EntryDirection, uint64) - Accountable(interface{}) bool } type BalanceManager interface { //Credit is crediting the peer, charging local node - Credit(peer *Peer, amount uint64, size uint32) error + Credit(peer *Peer, amount uint64) error //Debit is crediting the local node, charging the remote peer - Debit(peer *Peer, amount uint64, size uint32) error + Debit(peer *Peer, amount uint64) error } -type EntryDirection bool +type EntryDirection uint const ( - ChargeSender EntryDirection = true - ChargeReceiver EntryDirection = false + ChargeSender EntryDirection = 1 + ChargeReceiver EntryDirection = 2 + ChargeNone EntryDirection = 3 ) type AccountingHook struct { @@ -57,14 +73,15 @@ func (ah *AccountingHook) Send(peer *Peer, size uint32, msg interface{}) error { ah.lock.Lock() defer ah.lock.Unlock() var err error - if !ah.PriceOracle.Accountable(msg) { - return nil - } direction, price := ah.PriceOracle.Price(size, msg) if direction == ChargeSender { - err = ah.BalanceManager.Debit(peer, price, size) - } else { - err = ah.BalanceManager.Credit(peer, price, size) + err = ah.BalanceManager.Debit(peer, price) + ah.debitMetrics(price, size, err) + } else if direction == ChargeReceiver { + err = ah.BalanceManager.Credit(peer, price) + ah.creditMetrics(price, size, err) + } else if direction == ChargeNone { + return nil } return err } @@ -73,14 +90,33 @@ func (ah *AccountingHook) Receive(peer *Peer, size uint32, msg interface{}) erro ah.lock.Lock() defer ah.lock.Unlock() var err error - if !ah.PriceOracle.Accountable(msg) { - return nil - } direction, price := ah.PriceOracle.Price(size, msg) if direction == ChargeReceiver { - err = ah.BalanceManager.Debit(peer, price, size) - } else { - err = ah.BalanceManager.Credit(peer, price, size) + err = ah.BalanceManager.Debit(peer, price) + ah.debitMetrics(price, size, err) + } else if direction == ChargeSender { + err = ah.BalanceManager.Credit(peer, price) + ah.creditMetrics(price, size, err) + } else if direction == ChargeNone { + return nil } return err } + +func (ah *AccountingHook) debitMetrics(price uint64, size uint32, err error) { + mBalanceDebit.Inc(int64(price)) + mBytesDebit.Inc(int64(size)) + mMsgDebit.Inc(1) + if err != nil { + mSelfDrops.Inc(1) + } +} + +func (ah *AccountingHook) creditMetrics(price uint64, size uint32, err error) { + mBalanceCredit.Inc(int64(price)) + mBytesCredit.Inc(int64(size)) + mMsgCredit.Inc(1) + if err != nil { + mPeerDrops.Inc(1) + } +} diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index ede6cea3a9..b4fb3fe9eb 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -728,24 +728,18 @@ type StreamerPriceOracle struct { registry *Registry } -func (spo *StreamerPriceOracle) Accountable(msg interface{}) bool { - code, ok := spo.registry.spec.GetCode(msg) - if !ok { - return false - } - if _, ok = spo.priceMatrix[code]; ok { - return true - } - return false -} - func (spo *StreamerPriceOracle) Price(size uint32, msg interface{}) (direction protocols.EntryDirection, price uint64) { code, ok := spo.registry.spec.GetCode(msg) + direction = protocols.ChargeNone if !ok { - panic("Attempting to get message code for an expected message type, but code not found") + //this could be handled more severely (panic) but let's be gentle? + return } - tag := spo.priceMatrix[code] + tag, ok := spo.priceMatrix[code] + if !ok { + return + } price = tag.price if tag.sizeBased { price = price * uint64(size) diff --git a/swarm/swap/api.go b/swarm/swap/api.go deleted file mode 100644 index 70aecb876c..0000000000 --- a/swarm/swap/api.go +++ /dev/null @@ -1,108 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package swap - -import ( - "context" - "errors" - - "github.com/ethereum/go-ethereum/p2p/enode" -) - -var ( - ErrNoSuchPeerAccounting = errors.New("No accounting with that peer") -) - -//This is the API definition to access swarm swap accounting data via RPC -type API struct { - swap *Swap -} - -//Get metrics about swap for this node -//The current metrics are for accounted message types only -//(i.e. BytesTransferred is amount of bytes sent but only for a -//accounted message types) -type Metrics struct { - BalanceCredited uint64 - BalanceDebited uint64 - BytesCredited uint64 - BytesDebited uint64 - MsgCredited uint64 - MsgDebited uint64 - ChequesIssued uint64 - ChequesReceived uint64 - PeerDrops uint64 - SelfDrops uint64 -} - -//Create a new API instance -func NewAPI(swap *Swap) *API { - return &API{swap: swap} -} - -//Get the balance for this node with a specific peer -func (swapapi *API) BalanceWithPeer(ctx context.Context, peer enode.ID) (balance int64, err error) { - var ok bool - balance, ok = swapapi.swap.balances[peer] - if !ok { - err = ErrNoSuchPeerAccounting - } - return -} - -//Get the overall balance of the node -//Iterates over all peers this node is having accounted interaction -//and just adds up balances. -//It assumes that a disfavorable balance is represented as a negative value -func (swapapi *API) Balance(ctx context.Context) (balance int64, err error) { - balance = 0 - for _, peerBalance := range swapapi.swap.balances { - balance += peerBalance - } - return -} - -//Just return the Swap metrics -func (swapapi *API) GetSwapMetricsForPeer(ctx context.Context, peer enode.ID) (*Metrics, error) { - var ok bool - var metrics *Metrics - metrics, ok = swapapi.swap.metrics[peer] - if !ok { - return nil, ErrNoSuchPeerAccounting - } - return metrics, nil -} - -//Just return the Swap metrics -func (swapapi *API) GetSwapMetrics(ctx context.Context, peer enode.ID) (*Metrics, error) { - var metrics *Metrics - - for _, m := range swapapi.swap.metrics { - metrics.BalanceCredited += m.BalanceCredited - metrics.BalanceDebited += m.BalanceDebited - metrics.BytesCredited += m.BytesCredited - metrics.BytesDebited += m.BytesDebited - metrics.ChequesIssued += m.ChequesIssued - metrics.ChequesReceived += m.ChequesReceived - metrics.MsgCredited += m.MsgCredited - metrics.MsgDebited += m.MsgDebited - metrics.PeerDrops += m.PeerDrops - metrics.SelfDrops += m.SelfDrops - } - - return metrics, nil -} diff --git a/swarm/swap/chequemanager.go b/swarm/swap/chequemanager.go deleted file mode 100644 index 8e272a6426..0000000000 --- a/swarm/swap/chequemanager.go +++ /dev/null @@ -1,63 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package swap - -import ( - "github.com/ethereum/go-ethereum/p2p/enode" - "github.com/ethereum/go-ethereum/swarm/state" -) - -type ChequeManager struct { - stateStore state.Store - serialPerNode map[enode.ID]uint64 - openDebitCheques map[enode.ID][]*Cheque - openCreditCheques map[enode.ID][]*Cheque -} - -type Cheque struct { - serial uint64 - timeout uint64 - amount int64 - sumCumulated int64 - beneficiary enode.ID //this should probably be common.Address? -} - -func NewChequeManager(stateStore state.Store) *ChequeManager { - return &ChequeManager{ - stateStore: stateStore, - //TODO: restore from state store - serialPerNode: make(map[enode.ID]uint64), - openDebitCheques: make(map[enode.ID][]*Cheque), - openCreditCheques: make(map[enode.ID][]*Cheque), - } -} - -func (mgr *ChequeManager) CreateCheque(beneficiary enode.ID, amount int64) *Cheque { - mgr.serialPerNode[beneficiary]++ - cheque := &Cheque{ - serial: mgr.serialPerNode[beneficiary], - beneficiary: beneficiary, - amount: amount, - } - openCheques := mgr.openDebitCheques[beneficiary] - if openCheques == nil { - openCheques = make([]*Cheque, 0) - } - openCheques = append(openCheques, cheque) - mgr.openDebitCheques[beneficiary] = openCheques - return cheque -} diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go deleted file mode 100644 index 36d5448e66..0000000000 --- a/swarm/swap/protocol.go +++ /dev/null @@ -1,203 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package swap - -import ( - "context" - "fmt" - "sync" - - "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/enode" - "github.com/ethereum/go-ethereum/p2p/protocols" - "github.com/ethereum/go-ethereum/rpc" - "github.com/ethereum/go-ethereum/swarm/log" -) - -const ( - IsActiveProtocol = true -) - -//Here we define the Swap p2p.protocol -//We use it to standardize the interaction between nodes who need clear their accounts. -//Accounting itself is separate (see swarm/swap/swap.go) -//This protocol focuses on sending and receiving cheques and -//cheque management/clearing -//For a better understanding, please read the Swap network paper "Generalized Swap swear and swindle games" -type Protocol struct { - peersMu sync.RWMutex - peers map[enode.ID]*Peer -} - -//This is the peer representing a participant in this protocol -type Peer struct { - *protocols.Peer -} - -//Create a new protocol instance -func NewProtocol() *Protocol { - proto := &Protocol{ - peers: make(map[enode.ID]*Peer), - } - return proto -} - -// NewPeer is the constructor for the protocol Peer -func NewPeer(peer *protocols.Peer) *Peer { - p := &Peer{ - Peer: peer, - } - return p -} - -//In a peer exchange, if node A gets too indebted with node B, -//node A issues a cheque and sends it to B -type IssueChequeMsg struct { - Cheque *Cheque -} - -//In a scenario where B received a cheque from A, node B can -//redeem a cheque, which means it kicks off the process to cash it in. -//In this case, this message is sent to peer A -type RedeemChequeMsg struct { - Cheque *Cheque -} - -///////////////////////////////////////////////////////////////////// -// SECTION: node.Service interface -///////////////////////////////////////////////////////////////////// -func (s *Swap) Start(srv *p2p.Server) error { - log.Debug("Started swap") - return nil -} - -func (s *Swap) Stop() error { - log.Info("swap shutting down") - return nil -} - -var swapSpec = &protocols.Spec{ - Name: swapProtocolName, - Version: swapVersion, - MaxMsgSize: defaultMaxMsgSize, - Messages: []interface{}{ - IssueChequeMsg{}, - RedeemChequeMsg{}, - }, -} - -func (s *Swap) Protocols() []p2p.Protocol { - return []p2p.Protocol{ - { - Name: swapSpec.Name, - Version: swapSpec.Version, - Length: swapSpec.Length(), - Run: s.run, - }, - } -} - -func (s *Swap) APIs() []rpc.API { - apis := []rpc.API{ - { - Namespace: "swap", - Version: "1.0", - Service: NewAPI(s), - Public: true, - }, - } - return apis -} - -///////////////////////////////////////////////////////////////////// -// SECTION: p2p.protocol interface -///////////////////////////////////////////////////////////////////// -func (s *Swap) run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { - p := protocols.NewPeer(peer, rw, swapSpec) - sp := NewPeer(p) - s.protocol.setPeer(sp) - defer s.protocol.deletePeer(sp) - defer s.protocol.Close() - return sp.Run(sp.handleSwapMsg) -} - -func (s *Swap) NodeInfo() interface{} { - return nil -} - -func (s *Swap) PeerInfo(id enode.ID) interface{} { - return nil -} - -//------------------------------------------------------------------------------------------ -func (proto *Protocol) Close() { -} - -func (proto *Protocol) getPeer(peerId enode.ID) *Peer { - proto.peersMu.RLock() - defer proto.peersMu.RUnlock() - - return proto.peers[peerId] -} - -func (proto *Protocol) setPeer(peer *Peer) { - proto.peersMu.Lock() - defer proto.peersMu.Unlock() - - proto.peers[peer.ID()] = peer -} - -func (proto *Protocol) deletePeer(peer *Peer) { - proto.peersMu.Lock() - defer proto.peersMu.Unlock() - - delete(proto.peers, peer.ID()) -} - -func (proto *Protocol) peersCount() (c int) { - proto.peersMu.Lock() - c = len(proto.peers) - proto.peersMu.Unlock() - return -} - -//Protocol message handler for handling cheque messages -func (p *Peer) handleSwapMsg(ctx context.Context, msg interface{}) error { - switch msg := msg.(type) { - - case *IssueChequeMsg: - return p.handleIssueChequeMsg(ctx, msg) - - case *RedeemChequeMsg: - return p.handleRedeemChequeMsg(ctx, msg) - - default: - return fmt.Errorf("unknown message type: %T", msg) - } -} - -//A IssueChequeMsg has been received -func (p *Peer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) { - log.Debug("SwapProtocolPeer: handleIssueChequeMsg") - return err -} - -//A RedeemChequeMsg has been received -func (p *Peer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) { - log.Debug("SwapProtocolPeer: handleRedeemChequeMsg") - return err -} diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 65a9e63c51..fa7874d8a0 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -17,10 +17,8 @@ package swap import ( - "context" "errors" "fmt" - "math" "strconv" "sync" @@ -30,34 +28,18 @@ import ( "github.com/ethereum/go-ethereum/swarm/state" ) -const ( - defaultMaxMsgSize = 1024 * 1024 - swapProtocolName = "swap" - swapVersion = 1 -) - -var ( - payAt = int64(-4096 * 10000000) // threshold that triggers payment {request} (bytes) - dropAt = int64(-4096 * 12000000) // threshold that triggers disconnect (bytes) - - ErrInsufficientFunds = errors.New("Insufficient funds") -) - // SwAP Swarm Accounting Protocol // a peer to peer micropayment system // A node maintains an individual balance with every peer // Only messages which have a price will be accounted for type Swap struct { - chequeManager *ChequeManager //cheque manager keeps track of issued cheques - stateStore state.Store //stateStore is needed in order to keep balances across sessions - lock sync.RWMutex //lock the balances - balances map[enode.ID]int64 //map of balances for each peer - metrics map[enode.ID]*Metrics //map of metrics for each peer - protocol *Protocol //reference to the cheque exchange protocol + stateStore state.Store //stateStore is needed in order to keep balances across sessions + lock sync.RWMutex //lock the balances + balances map[enode.ID]int64 //map of balances for each peer } //Credit us and debit remote -func (s *Swap) Credit(peer *protocols.Peer, amount uint64, size uint32) (err error) { +func (s *Swap) Credit(peer *protocols.Peer, amount uint64) (err error) { s.lock.Lock() defer s.lock.Unlock() @@ -67,29 +49,12 @@ func (s *Swap) Credit(peer *protocols.Peer, amount uint64, size uint32) (err err peerBalance := s.balances[peer.ID()] s.stateStore.Put(peer.ID().String(), &peerBalance) - if float64(peerBalance) > math.Abs(float64(payAt)) { - ctx := context.TODO() - //WARNING: Should we do this? Otherwise anyone could create a WantChequeMsg requiring us to handle special situations... - err := s.wantCheque(ctx, peer.ID()) - if err != nil { - //TODO: special error handling, as at this point the accounting has been done - //but the cheque could not be sent? - log.Warn("Payment threshold exceeded, but error sending cheque!", "err", err) - } - } - if float64(peerBalance) > math.Abs(float64(dropAt)) { - s.metrics[peer.ID()].PeerDrops += 1 - return ErrInsufficientFunds - } - //TODO: size for metrics is currently misleading: should only account for size based messages(?) - s.updatePeerMetrics(peer, true, amount, size) - log.Debug(fmt.Sprintf("balance for peer %s: %s", peer.ID().String(), strconv.FormatInt(peerBalance, 10))) return err } //Debit us and credit remote -func (s *Swap) Debit(peer *protocols.Peer, amount uint64, size uint32) (err error) { +func (s *Swap) Debit(peer *protocols.Peer, amount uint64) (err error) { s.lock.Lock() defer s.lock.Unlock() @@ -100,23 +65,6 @@ func (s *Swap) Debit(peer *protocols.Peer, amount uint64, size uint32) (err erro peerBalance := s.balances[peer.ID()] s.stateStore.Put(peer.ID().String(), &peerBalance) - if peerBalance < payAt { - ctx := context.TODO() - err := s.issueCheque(ctx, peer.ID()) - s.metrics[peer.ID()].ChequesIssued += 1 - if err != nil { - //TODO: special error handling, as at this point the accounting has been done - //but the cheque could not be sent? - log.Warn("Payment threshold exceeded, but error sending cheque!", "err", err) - } - } - if peerBalance < dropAt { - s.metrics[peer.ID()].SelfDrops += 1 - return ErrInsufficientFunds - } - - s.updatePeerMetrics(peer, false, amount, size) - log.Debug(fmt.Sprintf("balance for peer %s: %s", peer.ID().String(), strconv.FormatInt(peerBalance, 10))) return nil } @@ -131,94 +79,21 @@ func (swap *Swap) GetPeerBalance(peer enode.ID) (int64, error) { return 0, errors.New("Peer not found") } -func (swap *Swap) GetPeerMetrics(peer enode.ID) (*Metrics, error) { - swap.lock.RLock() - defer swap.lock.RUnlock() - if p, ok := swap.metrics[peer]; ok { - return p, nil - } - return nil, errors.New("Peer not found") -} - func (s *Swap) loadState(peer *protocols.Peer) { var peerBalance int64 - var peerMetrics *Metrics peerID := peer.ID() - if _, ok := s.metrics[peerID]; !ok { - s.stateStore.Get("metrics"+peerID.String(), &peerMetrics) - if peerMetrics == nil { - peerMetrics = &Metrics{ - BalanceCredited: 0, - BalanceDebited: 0, - BytesCredited: 0, - BytesDebited: 0, - MsgCredited: 0, - MsgDebited: 0, - ChequesIssued: 0, - ChequesReceived: 0, - PeerDrops: 0, - SelfDrops: 0, - } - s.metrics[peerID] = peerMetrics - } - } if _, ok := s.balances[peerID]; !ok { s.stateStore.Get(peerID.String(), &peerBalance) s.balances[peerID] = peerBalance } } -//local node is being credited(in favor of local node), so the balance increases -func (s *Swap) updatePeerMetrics(peer *protocols.Peer, credit bool, amount uint64, size uint32) { - - metrics := s.metrics[peer.ID()] - - if credit { - metrics.BalanceCredited += amount - metrics.BytesCredited += uint64(size) - metrics.MsgCredited += 1 - } else { - metrics.BalanceDebited += amount - metrics.BytesDebited += uint64(size) - metrics.MsgDebited += 1 - } - - s.stateStore.Put("metrics"+peer.ID().String(), metrics) -} - -//Issue a cheque for the remote peer. Happens if we are indebted with the peer -//and crossed the payment threshold -func (s *Swap) issueCheque(ctx context.Context, id enode.ID) error { - cheque := s.chequeManager.CreateCheque(id, int64(math.Abs(float64(payAt)))) - _ = IssueChequeMsg{ - Cheque: cheque, - } - p := s.protocol.getPeer(id) - if p == nil { - return fmt.Errorf("wanting to send to non-connected peer!") - } - return nil - //TODO: Don't actually send any cheques yet - //return p.Send(ctx, msg) -} - -//Issue a cheque for the remote peer. Happens if we are indebted with the peer -//and crossed the payment threshold -func (s *Swap) wantCheque(ctx context.Context, id enode.ID) error { - //TODO: Don't actually initially any real message exchange yet - //return p.Send(ctx, msg) - return nil -} - // New - swap constructor func New(stateStore state.Store) (swap *Swap) { swap = &Swap{ - chequeManager: NewChequeManager(stateStore), - stateStore: stateStore, - balances: make(map[enode.ID]int64), - metrics: make(map[enode.ID]*Metrics), - protocol: NewProtocol(), + stateStore: stateStore, + balances: make(map[enode.ID]int64), } return } diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 61e6955a83..3ccc363813 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -22,29 +22,22 @@ import ( "flag" "fmt" "io/ioutil" - "math" mrand "math/rand" "os" - "path/filepath" "testing" "time" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/rlp" - "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/state" colorable "github.com/mattn/go-colorable" ) var ( - p2pPort = 30100 - ipcpath = ".swarm.ipc" - datadirPrefix = ".data_" - loglevel = flag.Int("loglevel", 2, "verbosity of logs") + loglevel = flag.Int("loglevel", 2, "verbosity of logs") ) type testPriceOracle struct{} @@ -56,10 +49,6 @@ func (tpo *testPriceOracle) Accountable(msg interface{}) bool { func (tpo *testPriceOracle) Price(size uint32, msg interface{}) (protocols.EntryDirection, uint64) { switch msg := msg.(type) { - case *testExceedsPayAtMsg: - return msg.Price(size) - case *testExceedsDropAtMsg: - return msg.Price(size) case *testCheapMsg: return msg.Price(size) case *testSizeBasedMsg: @@ -67,7 +56,7 @@ func (tpo *testPriceOracle) Price(size uint32, msg interface{}) (protocols.Entry case *testChargeRecvMsg: return msg.Price(size) } - return false, 0 + return protocols.ChargeNone, 0 } var testSpec = &protocols.Spec{ @@ -75,8 +64,6 @@ var testSpec = &protocols.Spec{ Version: 1, MaxMsgSize: 10 * 1024 * 1024, Messages: []interface{}{ - testExceedsPayAtMsg{}, - testExceedsDropAtMsg{}, testCheapMsg{}, testSizeBasedMsg{}, testChargeRecvMsg{}, @@ -102,24 +89,12 @@ func (d *dummyRW) ReadMsg() (p2p.Msg, error) { } //define a couple of messages for tests -type testExceedsPayAtMsg struct{} -type testExceedsDropAtMsg struct{} type testCheapMsg struct{} type testSizeBasedMsg struct { Data []byte } type testChargeRecvMsg struct{} -//this message is just one unit more expensive than the payment threshold -func (tmsg *testExceedsPayAtMsg) Price(size uint32) (protocols.EntryDirection, uint64) { - return protocols.ChargeSender, uint64(math.Abs(float64(payAt)) + float64(1)) -} - -//this message is just one unit more expensive than the disconnect threshold -func (tmsg *testExceedsDropAtMsg) Price(size uint32) (protocols.EntryDirection, uint64) { - return protocols.ChargeSender, uint64(math.Abs(float64(dropAt)) + float64(1)) -} - //a message with an arbitrary cost func (tmsg *testCheapMsg) Price(size uint32) (protocols.EntryDirection, uint64) { return protocols.ChargeSender, uint64(10) @@ -142,13 +117,6 @@ func init() { log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true)))) } -//check that the disconnect threshold is below the payment threshold -func TestLimits(t *testing.T) { - if dropAt >= payAt { - t.Fatal(fmt.Sprintf("dropAt limit is not lower than payAt limit, dropAt: %d, payAt: %d", dropAt, payAt)) - } -} - //check that the disconnect threshold is below the payment threshold func TestRepeatedBookings(t *testing.T) { //create a test swap account @@ -156,12 +124,10 @@ func TestRepeatedBookings(t *testing.T) { defer os.RemoveAll(testDir) testPeer := newDummyPeer() - //size is irrelevant for this test - size := uint32(0) amount := mrand.Intn(100) cnt := 1 + mrand.Intn(10) for i := 0; i < cnt; i++ { - swap.Credit(testPeer.Peer.Peer, uint64(amount), size) + swap.Credit(testPeer.Peer, uint64(amount)) } expectedBalance := int64(cnt * amount) realBalance := swap.balances[testPeer.ID()] @@ -173,7 +139,7 @@ func TestRepeatedBookings(t *testing.T) { amount = mrand.Intn(100) cnt = 1 + mrand.Intn(10) for i := 0; i < cnt; i++ { - swap.Debit(testPeer2.Peer.Peer, uint64(amount), size) + swap.Debit(testPeer2.Peer, uint64(amount)) } expectedBalance = int64(0 - (cnt * amount)) realBalance = swap.balances[testPeer2.ID()] @@ -185,9 +151,9 @@ func TestRepeatedBookings(t *testing.T) { amount1 := mrand.Intn(100) amount2 := mrand.Intn(100) amount3 := mrand.Intn(100) - swap.Credit(testPeer2.Peer.Peer, uint64(amount1), size) - swap.Credit(testPeer2.Peer.Peer, uint64(amount2), size) - swap.Debit(testPeer2.Peer.Peer, uint64(amount3), size) + swap.Credit(testPeer2.Peer, uint64(amount1)) + swap.Credit(testPeer2.Peer, uint64(amount2)) + swap.Debit(testPeer2.Peer, uint64(amount3)) expectedBalance = expectedBalance + int64(amount1+amount2-amount3) realBalance = swap.balances[testPeer2.ID()] @@ -197,66 +163,6 @@ func TestRepeatedBookings(t *testing.T) { } } -//unit test for exceeds pay limit -//when the payment threshold is reached, a cheque will be issued -//this test checks that a cheque is present if a message is sent -//which exceeds the payment threshold -//(note: the details of cheque handling will need to be fleshed out -//in future iterations, current implementation is very primitive) -func TestExceedsPayAt(t *testing.T) { - //create a test swap account - swap, testDir := createTestSwap(t) - defer os.RemoveAll(testDir) - - msg := &testExceedsPayAtMsg{} - ctx := context.Background() - testPeer := newDummyPeer() - testPeer.Send(ctx, msg) - - _, price := msg.Price(0) - if swap.balances[testPeer.ID()] != int64(0-price) { - t.Fatalf("Expected balance to be %d, but is %d", price, swap.balances[testPeer.ID()]) - } - - //check that a cheque has been created - cheques := swap.chequeManager.openDebitCheques[testPeer.ID()] - if cheques == nil { - t.Fatal("Expected cheques for this peer to be present, but are nil") - } - if len(cheques) == 0 { - t.Fatal("Expected a cheque to have arrived, but len is zero") - } - if cheques[0].serial != 1 { - t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial)) - } - if float64(cheques[0].amount) != float64(math.Abs(float64(payAt))) { - t.Fatal(fmt.Sprintf("Expected cheques amount to be equal to payAt limit, but it is: %d", cheques[0].amount)) - } -} - -//unit test for exceeds drop limit -//tests that a message is being sent which crosses the drop limit -//in that case, we should receive a InsufficientFunds error -func TestExceedsDropAt(t *testing.T) { - //create a test swap account - swap, testDir := createTestSwap(t) - defer os.RemoveAll(testDir) - - msg := &testExceedsDropAtMsg{} - ctx := context.Background() - testPeer := newDummyPeer() - err := testPeer.Send(ctx, msg) - - _, price := msg.Price(0) - if swap.balances[testPeer.ID()] != int64(0-price) { - t.Fatalf("Expected balance to be %d, but is %d", price, swap.balances[testPeer.ID()]) - } - - if err != ErrInsufficientFunds { - t.Fatal("Expected this test to fail with insufficient funds, but it did not") - } -} - //send a message with cost, //then check that the balance has the expected amount func TestSendCheapMessage(t *testing.T) { @@ -400,106 +306,12 @@ func dummyMsgHandler(ctx context.Context, msg interface{}) error { return nil } -func createAndStartSvcNode(swap *Swap, t *testing.T) *node.Node { - stack, err := newServiceNode(p2pPort, 0, 0) - if err != nil { - t.Fatal("Create servicenode #1 fail", "err", err) - } - - swapsvc := func(ctx *node.ServiceContext) (node.Service, error) { - return swap, nil - } - - err = stack.Register(swapsvc) - if err != nil { - t.Fatal("Register service in servicenode #1 fail", "err", err) - } - - // start the nodes - err = stack.Start() - if err != nil { - t.Fatal("servicenode #1 start failed", "err", err) - } - - return stack -} - -//tests some basic things over RPC -func TestSwapRPC(t *testing.T) { - - swap, testDir := createTestSwap(t) - defer os.RemoveAll(testDir) - - stack := createAndStartSvcNode(swap, t) - defer stack.Stop() - defer os.RemoveAll(stack.DataDir()) - - // connect to the servicenode RPCs - rpcclient, err := rpc.Dial(filepath.Join(stack.DataDir(), ipcpath)) - if err != nil { - t.Fatal("connect to servicenode IPC fail", "err", err) - } - defer os.RemoveAll(stack.DataDir()) - - var balance int64 - err = rpcclient.Call(&balance, "swap_balance") - if err != nil { - t.Fatal("servicenode RPC failed", "err", err) - } - log.Debug("servicenode balance", "balance", balance) - - if balance != 0 { - t.Fatal("Expected balance to be 0 but it is not") - } - - dummyPeer1 := newDummyPeer() - dummyPeer2 := newDummyPeer() - id1 := dummyPeer1.ID() - id2 := dummyPeer2.ID() - - fake1 := int64(234) - fake2 := int64(-100) - - swap.balances[id1] = fake1 - swap.balances[id2] = fake2 - - err = rpcclient.Call(&balance, "swap_balanceWithPeer", id1) - if err != nil { - t.Fatal("servicenode RPC failed", "err", err) - } - log.Debug("balance1", "balance-1", balance) - if balance != fake1 { - t.Fatal(fmt.Sprintf("Expected balance %d to be equal to fake balance %d, but it is not", balance, fake1)) - } - - err = rpcclient.Call(&balance, "swap_balanceWithPeer", id2) - if err != nil { - t.Fatal("servicenode RPC failed", "err", err) - } - log.Debug("balance2", "balance-2", balance) - if balance != fake2 { - t.Fatal(fmt.Sprintf("Expected balance %d to be equal to fake balance %d, but it is not", balance, fake2)) - } - - err = rpcclient.Call(&balance, "swap_balance") - if err != nil { - t.Fatal("servicenode RPC failed", "err", err) - } - log.Debug("balance", "balance", balance) - - fakeSum := fake1 + fake2 - if balance != fakeSum { - t.Fatal(fmt.Sprintf("Expected balance %d to be equal to sum %d, but it is not", balance, fakeSum)) - } -} - type dummyPeer struct { - *Peer + *protocols.Peer testFunc func(error) } func (dp *dummyPeer) Drop(err error) { - fmt.Println("DD") dp.testFunc(err) } @@ -508,34 +320,7 @@ func newDummyPeer() *dummyPeer { id := adapters.RandomNodeConfig().ID protoPeer := protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), &dummyRW{}, testSpec) dummy := &dummyPeer{ - Peer: NewPeer(protoPeer), + Peer: protoPeer, } return dummy } - -//creates a p2p.Service node stub -func newServiceNode(port int, httpport int, wsport int, modules ...string) (*node.Node, error) { - cfg := &node.DefaultConfig - cfg.P2P.ListenAddr = fmt.Sprintf(":%d", port) - cfg.P2P.EnableMsgEvents = true - cfg.P2P.NoDiscovery = true - cfg.IPCPath = ipcpath - cfg.DataDir = fmt.Sprintf("%s%d", datadirPrefix, port) - if httpport > 0 { - cfg.HTTPHost = node.DefaultHTTPHost - cfg.HTTPPort = httpport - } - if wsport > 0 { - cfg.WSHost = node.DefaultWSHost - cfg.WSPort = wsport - cfg.WSOrigins = []string{"*"} - for i := 0; i < len(modules); i++ { - cfg.WSModules = append(cfg.WSModules, modules[i]) - } - } - stack, err := node.New(cfg) - if err != nil { - return nil, fmt.Errorf("ServiceNode create fail: %v", err) - } - return stack, nil -} diff --git a/swarm/swap_test.go b/swarm/swap_test.go index dc47ffd85e..82aad6939e 100644 --- a/swarm/swap_test.go +++ b/swarm/swap_test.go @@ -29,6 +29,7 @@ import ( "time" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" @@ -172,23 +173,6 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { } else { log.Debug(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) } - metrics, err := swarm.swap.GetPeerMetrics(n) - if err == nil && *printStats { - fmt.Println(fmt.Sprintf("********** Metrics for node %s with node %s: *************", node.TerminalString(), n.TerminalString())) - fmt.Println(fmt.Sprintf("Total units credited: %d", metrics.BalanceCredited)) - fmt.Println(fmt.Sprintf("Total units debited: %d", metrics.BalanceDebited)) - fmt.Println(fmt.Sprintf("Total bytes credited: %d", metrics.BytesCredited)) - fmt.Println(fmt.Sprintf("Total bytes debited: %d", metrics.BytesDebited)) - fmt.Println(fmt.Sprintf("Cheques issued: %d", metrics.ChequesIssued)) - fmt.Println(fmt.Sprintf("Cheques received: %d", metrics.ChequesReceived)) - fmt.Println(fmt.Sprintf("Number of messages credited: %d", metrics.MsgCredited)) - fmt.Println(fmt.Sprintf("Number of messages debited: %d", metrics.MsgDebited)) - fmt.Println(fmt.Sprintf("Peers dropped: %d", metrics.PeerDrops)) - fmt.Println(fmt.Sprintf("Number of times node dropped itself: %d", metrics.SelfDrops)) - } else { - //not all peers have metrics with every node, so probably can be ignored - log.Debug("Error getting metrics", "err", err) - } } //update the map for this node balancesMap[node] = subBalances @@ -202,6 +186,17 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { fmt.Println(fmt.Sprintf(".........with node %s: balance %d", kk.TerminalString(), vv)) } } + //NOTE: this are currently metrics over ALL nodes, not per node + fmt.Println(fmt.Sprintf("Total units credited: %d", metrics.Get("account.balance.credit").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Total units debited: %d", metrics.Get("account.balance.debit").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Total bytes credited: %d", metrics.Get("account.bytes.credit").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Total bytes debited: %d", metrics.Get("account.bytes.debit").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Cheques issued: %d", metrics.Get("account.cheques.issued").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Cheques received: %d", metrics.Get("account.cheques.received").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Number of messages credited: %d", metrics.Get("account.msg.credit").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Number of messages debited: %d", metrics.Get("account.msg.debit").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Peers dropped: %d", metrics.Get("account.peerdrops").(metrics.Counter).Count())) + fmt.Println(fmt.Sprintf("Number of times node dropped itself: %d", metrics.Get("account.selfdrops").(metrics.Counter).Count())) } //now iterate the whole map diff --git a/swarm/swarm.go b/swarm/swarm.go index 8c0a327fad..24bbfa6418 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -367,10 +367,6 @@ func (self *Swarm) Start(srv *p2p.Server) error { } log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.BaseAddr())) - if self.swap != nil { - self.swap.Start(srv) - } - if self.ps != nil { self.ps.Start(srv) } @@ -450,9 +446,6 @@ func (self *Swarm) Protocols() (protos []p2p.Protocol) { protos = append(protos, self.ps.Protocols()...) } - if self.swap != nil { - protos = append(protos, self.swap.Protocols()...) - } return }