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
}