diff --git a/p2p/protocols/accounting.go b/p2p/protocols/accounting.go index 74f8cd2d7b..67a8f20384 100644 --- a/p2p/protocols/accounting.go +++ b/p2p/protocols/accounting.go @@ -20,29 +20,47 @@ import ( "github.com/ethereum/go-ethereum/metrics" ) +//define some 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) + //All metrics are cumulative + + //total amount of units credited + mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", nil) + //total amount of units debited + mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", nil) + //total amount of bytes credited + mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", nil) + //total amount of bytes debited + mBytesDebit = metrics.NewRegisteredCounterForced("account.bytes.debit", nil) + //total amount of credited messages + mMsgCredit = metrics.NewRegisteredCounterForced("account.msg.credit", nil) + //total amount of debited messages + mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", nil) + //how many times local node had to drop remote peers + mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", nil) + //how many times local node overdrafted and dropped + mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", nil) + //how many cheques have been issued + //mChequesIssued = metrics.NewRegisteredCounterForced("account.cheques.issued", nil) + //how many cheques have been received + //mChequesReceived = metrics.NewRegisteredCounterForced("account.cheques.received", nil) ) +//Prices defines how prices are being passed on to the accounting instance type Prices interface { + //Return the Price for a message Price(interface{}) *Price } +//Price represents the costs of a message type Price struct { Value int64 //Positive if sender pays, negative if receiver pays PerByte bool //True if the price is per byte or for unit } +//Price can be different depending on if it is the sender or receiver who pays +//ForSender gives back the price for sending a message func (p *Price) ForSender(size uint32) int64 { price := p.Value if p.PerByte { @@ -51,6 +69,8 @@ func (p *Price) ForSender(size uint32) int64 { return price } +//Price can be different depending on if it is the sender or receiver who pays +//ForReceiver gives back the price for receiving a message func (p *Price) ForReceiver(size uint32) int64 { price := p.Value if p.PerByte { @@ -59,6 +79,9 @@ func (p *Price) ForReceiver(size uint32) int64 { return 0 - price } +//Balance is the actual accounting instance +//Balance defines the operations needed for accounting +//Implementations internally maintain the balance for every peer type Balance interface { //Adds amount to the local balance with remote node `peer`; //positive amount = credit local node @@ -66,9 +89,12 @@ type Balance interface { Add(amount int64, peer *Peer) error } +//Accounting implements the Hook interface +//It interfaces to the balances through the Balance interface, +//while interfacing with protocols and its prices through the Prices interface type Accounting struct { - Balance - Prices + Balance //interface to accounting logic + Prices //interface to prices logic } func NewAccounting(mgr Balance, po Prices) *Accounting { @@ -79,28 +105,41 @@ func NewAccounting(mgr Balance, po Prices) *Accounting { return ah } +//Implement Hook.Send func (ah *Accounting) Send(peer *Peer, size uint32, msg interface{}) error { + //get the price for a message (through the protocol spec) price := ah.Price(msg) + //this message doesn't need accounting if price == nil { return nil } + //evaluate the price for sending messages finalPrice := price.ForSender(size) + //do the accounting err := ah.Add(finalPrice, peer) + //record metrics ah.doMetrics(finalPrice, size, err) return err } +//Implement Hook.Receive func (ah *Accounting) Receive(peer *Peer, size uint32, msg interface{}) error { + //get the price for a message (through the protocol spec) price := ah.Price(msg) + //this message doesn't need accounting if price == nil { return nil } + //evaluate the price for receiving messages finalPrice := price.ForReceiver(size) + //do the accounting err := ah.Add(finalPrice, peer) + //record metrics ah.doMetrics(finalPrice, size, err) return err } +//record some metrics func (ah *Accounting) doMetrics(price int64, size uint32, err error) { if price > 0 { mBalanceCredit.Inc(int64(price)) diff --git a/p2p/protocols/accounting_test.go b/p2p/protocols/accounting_test.go index 9c16699bfc..7e4f85446e 100644 --- a/p2p/protocols/accounting_test.go +++ b/p2p/protocols/accounting_test.go @@ -27,27 +27,36 @@ import ( "github.com/ethereum/go-ethereum/rlp" ) +//dummy Balance implementation type dummyBalance struct { amount int64 peer *Peer } +//dummy Prices implementation type dummyPrices struct{} +//a dummy message which needs size based accounting type perBytesMsg struct { Content string } +//a dummy message which is paid for per unit type perUnitMsg struct{} + +//a dummy message which doesn't need any payment type zeroMsg struct{} +//return the price for the defined messages func (d *dummyPrices) Price(msg interface{}) *Price { switch msg.(type) { + //size based message cost, sender pays case *perBytesMsg: return &Price{ PerByte: true, Value: int64(100), } + //unitary cost, receiver pays case *perUnitMsg: return &Price{ PerByte: false, @@ -57,71 +66,173 @@ func (d *dummyPrices) Price(msg interface{}) *Price { return nil } +//dummy accounting implementation, only stores values for later check func (d *dummyBalance) Add(amount int64, peer *Peer) error { d.amount = amount d.peer = peer return nil } +//We need to test that if the hook is not defined, then message infrastructure +//(send,receive) still works func TestNoHook(t *testing.T) { + //create a test spec spec := createTestSpec() + //a random node id := adapters.RandomNodeConfig().ID + //a peer p := p2p.NewPeer(id, "testPeer", nil) - peer := NewPeer(p, &dummyRW{}, spec) + rw := &dummyRW{} + peer := NewPeer(p, rw, spec) ctx := context.TODO() msg := &perBytesMsg{Content: "testBalance"} + //send a message peer.Send(ctx, msg) + //simulate receiving a message + rw.msg = msg peer.handleIncoming(func(ctx context.Context, msg interface{}) error { return nil }) + //all should just work and not result in any error } -func TestSendBalance(t *testing.T) { +//lowest level unit test +func TestSend(t *testing.T) { + //create instances balance := &dummyBalance{} prices := &dummyPrices{} - + //create the spec spec := createTestSpec() + //create the accounting hook for the spec spec.Hook = NewAccounting(balance, prices) + //create a peer + id := adapters.RandomNodeConfig().ID + p := p2p.NewPeer(id, "testPeer", nil) + peer := NewPeer(p, &dummyRW{}, spec) + msg := &perBytesMsg{Content: "testBalance"} + size, _ := rlp.EncodeToBytes(msg) + spec.Hook.Send(peer, uint32(len(size)), msg) + if balance.amount != int64(len(size)*100) { + t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount) + } + + //send a unitary, receiver pays message + msg2 := &perUnitMsg{} + spec.Hook.Send(peer, 0, msg2) + //check that the balance is actually the expected + if balance.amount != int64(-99) { + t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount) + } + + //set arbitrary balance + balance.amount = 77 + //send a non accounted message, balance should not change + msg3 := &zeroMsg{} + spec.Hook.Send(peer, 0, msg3) + if balance.amount != int64(77) { + t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount) + } +} + +//lowest level unit test +func TestReceive(t *testing.T) { + //create instances + balance := &dummyBalance{} + prices := &dummyPrices{} + //create the spec + spec := createTestSpec() + //create the accounting hook for the spec + spec.Hook = NewAccounting(balance, prices) + //create a peer + id := adapters.RandomNodeConfig().ID + p := p2p.NewPeer(id, "testPeer", nil) + peer := NewPeer(p, &dummyRW{}, spec) + msg := &perBytesMsg{Content: "testBalance"} + + size, _ := rlp.EncodeToBytes(msg) + spec.Hook.Receive(peer, uint32(len(size)), msg) + if balance.amount != int64(len(size)*-100) { + t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount) + } + + //send a unitary, receiver pays message + msg2 := &perUnitMsg{} + spec.Hook.Receive(peer, 0, msg2) + //check that the balance is actually the expected + if balance.amount != int64(99) { + t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount) + } + + //set arbitrary balance + balance.amount = 77 + //send a non accounted message, balance should not change + msg3 := &zeroMsg{} + spec.Hook.Receive(peer, 0, msg3) + if balance.amount != int64(77) { + t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount) + } +} + +//Test sending with a peer, higher level +func TestSendWithPeer(t *testing.T) { + //create instances + balance := &dummyBalance{} + prices := &dummyPrices{} + //create the spec + spec := createTestSpec() + //create the accounting hook for the spec + spec.Hook = NewAccounting(balance, prices) + //create a peer id := adapters.RandomNodeConfig().ID p := p2p.NewPeer(id, "testPeer", nil) peer := NewPeer(p, &dummyRW{}, spec) ctx := context.TODO() msg := &perBytesMsg{Content: "testBalance"} size, _ := rlp.EncodeToBytes(msg) + //send a size based message peer.Send(ctx, msg) + //check that the balance is actually the expected if balance.amount != int64((len(size) * 100)) { - t.Fatalf("Expected price to be %d but is %d", (len(size) * 100), balance.amount) + t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount) } + //send a unitary, receiver pays message msg2 := &perUnitMsg{} peer.Send(ctx, msg2) + //check that the balance is actually the expected if balance.amount != int64(-99) { - t.Fatalf("Expected price to be %d but is %d", -99, balance.amount) + t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount) } + //set arbitrary balance balance.amount = 77 + //send a non accounted message, balance should not change msg3 := &zeroMsg{} peer.Send(ctx, msg3) if balance.amount != int64(77) { - t.Fatalf("Expected price to be %d but is %d", 77, balance.amount) + t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount) } } -func TestReceiveBalance(t *testing.T) { +//Test receiving with a peer, higher level +func TestReceiveWithPeer(t *testing.T) { + //create instances balance := &dummyBalance{} prices := &dummyPrices{} - + //create the spec spec := createTestSpec() + //create the accounting hook for the spec spec.Hook = NewAccounting(balance, prices) - + //create a peer id := adapters.RandomNodeConfig().ID p := p2p.NewPeer(id, "testPeer", nil) rw := &dummyRW{} peer := NewPeer(p, rw, spec) + + //simulate receiving a size based, sender-pays message msg := &perBytesMsg{Content: "testBalance"} size, _ := rlp.EncodeToBytes(msg) - rw.msg = msg rw.code, _ = spec.GetCode(msg) err := peer.handleIncoming(func(ctx context.Context, msg interface{}) error { @@ -131,9 +242,10 @@ func TestReceiveBalance(t *testing.T) { t.Fatalf("Expected no error, but got error: %v", err) } if balance.amount != int64((len(size) * (-100))) { - t.Fatalf("Expected price to be %d but is %d", (len(size) * (-100)), balance.amount) + t.Fatalf("Expected balance to be %d but is %d", (len(size) * (-100)), balance.amount) } + //simulate receiving a size based, sender-pays message msg2 := &perUnitMsg{} rw.msg = msg2 rw.code, _ = spec.GetCode(msg2) @@ -144,23 +256,24 @@ func TestReceiveBalance(t *testing.T) { t.Fatalf("Expected no error, but got error: %v", err) } if balance.amount != int64(99) { - t.Fatalf("Expected price to be %d but is %d", 99, balance.amount) + t.Fatalf("Expected balance to be %d but is %d", 99, balance.amount) } + //set arbitrary balance an msg3 := &zeroMsg{} rw.msg = msg3 rw.code, _ = spec.GetCode(msg3) //need to reset cause no accounting won't overwrite balance.amount = -888 + //simulate receiving a non accounted message, balance should not change err = peer.handleIncoming(func(ctx context.Context, msg interface{}) error { return nil }) if err != nil { t.Fatalf("Expected no error, but got error: %v", err) } - if balance.amount != int64(-888) { - t.Fatalf("Expected price to be %d but is %d", -888, balance.amount) + t.Fatalf("Expected balance to be %d but is %d", -888, balance.amount) } } @@ -200,6 +313,7 @@ func (d *dummyRW) getDummyMsg() []byte { return rr } +//create a test spec func createTestSpec() *Spec { spec := &Spec{ Name: "test", diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index f3908bf940..4376f2b3a6 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -123,9 +123,14 @@ type WrappedMsg struct { Payload []byte } +//For accounting, the design is to allow the Spec to describe which and how its messages are priced +//To access this functionality, we provide a Hook interface which will call accounting methods +//NOTE: there could be more such (horizontal) hooks in the future type Hook interface { - Send(*Peer, uint32, interface{}) error - Receive(*Peer, uint32, interface{}) error + //A hook for sending messages + Send(peer *Peer, size uint32, msg interface{}) error + //A hook for receiving messages + Receive(peer *Peer, size uint32, msg interface{}) error } // Spec is a protocol specification including its name and version as well as @@ -147,6 +152,7 @@ type Spec struct { // each message must have a single unique data type Messages []interface{} + //hook for accounting (could be extended to multiple hooks in the future) Hook Hook initOnce sync.Once @@ -287,6 +293,7 @@ func (p *Peer) Send(ctx context.Context, msg interface{}) error { return errorf(ErrInvalidMsgType, "%v", code) } + //if the accounting hook is set, call it if p.spec.Hook != nil { err := p.spec.Hook.Send(p, wmsg.Size, msg) if err != nil { @@ -353,10 +360,12 @@ func (p *Peer) handleIncoming(handle func(ctx context.Context, msg interface{}) return errorf(ErrDecode, "<= %v: %v", msg, err) } + //if the accounting hook is set, call it if p.spec.Hook != nil { if wmsg.Size != uint32(len(wmsg.Payload)) { - log.Warn("Advertised message size and payload length don't match") - p.Drop(errors.New("message size and payload length don't match")) + errMsg := "Advertised message size and payload length don't match" + log.Warn(errMsg) + p.Drop(errors.New(errMsg)) } err := p.spec.Hook.Receive(p, wmsg.Size, val) if err != nil { diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index bff207a983..e610e1483b 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -38,15 +38,18 @@ type Swap struct { balances map[enode.ID]int64 //map of balances for each peer } -//Credit us and debit remote +//Swap implements the protocols.Balance interface +//Add is the (sole) accounting function func (s *Swap) Add(amount int64, peer *protocols.Peer) (err error) { s.lock.Lock() defer s.lock.Unlock() + //load existing balances from the state store s.loadState(peer) - + //adjust the balance + //if amount is negative, it will decrease, otherwise increase s.balances[peer.ID()] += amount - + //save the new balance to the state store peerBalance := s.balances[peer.ID()] s.stateStore.Put(peer.ID().String(), &peerBalance) @@ -54,7 +57,7 @@ func (s *Swap) Add(amount int64, peer *protocols.Peer) (err error) { return err } -//get a peer's balance +//Get a peer's balance func (swap *Swap) GetPeerBalance(peer enode.ID) (int64, error) { swap.lock.RLock() defer swap.lock.RUnlock() @@ -64,9 +67,12 @@ func (swap *Swap) GetPeerBalance(peer enode.ID) (int64, error) { return 0, errors.New("Peer not found") } +//load balances from the state store (persisted) func (s *Swap) loadState(peer *protocols.Peer) { var peerBalance int64 peerID := peer.ID() + //only load if the current instance doesn't already have this peer's + //balance in memory if _, ok := s.balances[peerID]; !ok { s.stateStore.Get(peerID.String(), &peerBalance) s.balances[peerID] = peerBalance diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 2708725c38..d1b48a2f4f 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -43,7 +43,7 @@ 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 +//Test that repeated bookings do correct accounting func TestRepeatedBookings(t *testing.T) { //create a test swap account swap, testDir := createTestSwap(t)