swarm/swap: leaner first phase, bug fixes and PR comments

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
 

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
+	ChargeSender   EntryDirection = 1
-	ChargeReceiver EntryDirection = false
+	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)
+		err = ah.BalanceManager.Debit(peer, price)
-	} else {
+		ah.debitMetrics(price, size, err)
-		err = ah.BalanceManager.Credit(peer, price, size)
+	} 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)
+		err = ah.BalanceManager.Debit(peer, price)
-	} else {
+		ah.debitMetrics(price, size, err)
-		err = ah.BalanceManager.Credit(peer, price, size)
+	} 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)
+	}
+}

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)

swarm/swap/api.go

@@ -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 <http://www.gnu.org/licenses/>.
-
-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
-}

swarm/swap/chequemanager.go

@@ -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 <http://www.gnu.org/licenses/>.
-
-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
-}

swarm/swap/protocol.go

@@ -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 <http://www.gnu.org/licenses/>.
-
-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
-}

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
-	stateStore    state.Store           //stateStore is needed in order to keep balances across sessions
+	lock       sync.RWMutex       //lock the balances
-	lock          sync.RWMutex          //lock the balances
+	balances   map[enode.ID]int64 //map of balances for each peer
-	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
 }
 
 //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,
-		stateStore:    stateStore,
+		balances:   make(map[enode.ID]int64),
-		balances:      make(map[enode.ID]int64),
-		metrics:       make(map[enode.ID]*Metrics),
-		protocol:      NewProtocol(),
 	}
 	return
 }

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
+	loglevel = flag.Int("loglevel", 2, "verbosity of logs")
-	ipcpath       = ".swarm.ipc"
-	datadirPrefix = ".data_"
-	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, uint64(amount1))
-	swap.Credit(testPeer2.Peer.Peer, uint64(amount2), size)
+	swap.Credit(testPeer2.Peer, uint64(amount2))
-	swap.Debit(testPeer2.Peer.Peer, uint64(amount3), size)
+	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
-}

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

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
 }