p2p/protocols: addressed @nonsense PR comments

This commit is contained in:
Fabio Barone 2018-10-24 08:30:05 -05:00
parent 47a57216de
commit 01470a7125
3 changed files with 25 additions and 11 deletions

View file

@ -61,7 +61,13 @@ type Price struct {
Payer Payer Payer Payer
} }
//ForSender gives back the price for sending a message //For gives back the price for a message
//A protocol provides the message price in absolute value
//This method then returns the correct signed amount,
//depending on who pays, which is identified by the `payer` argument:
//`Send` will pass a `Sender` payer, `Receive` will pass the `Receiver` argument.
//Thus: If Sending and sender pays, amount positive, otherwise negative
//If Receiving, and receiver pays, amount positive, otherwise negative
func (p *Price) For(payer Payer, size uint32) int64 { func (p *Price) For(payer Payer, size uint32) int64 {
price := p.Value price := p.Value
if p.PerByte { if p.PerByte {
@ -114,7 +120,7 @@ func (ah *Accounting) Send(peer *Peer, size uint32, msg interface{}) error {
costToLocalNode := price.For(Sender, size) costToLocalNode := price.For(Sender, size)
//do the accounting //do the accounting
err := ah.Add(costToLocalNode, peer) err := ah.Add(costToLocalNode, peer)
//record metrics //record metrics: just increase counters for user-facing metrics
ah.doMetrics(costToLocalNode, size, err) ah.doMetrics(costToLocalNode, size, err)
return err return err
} }
@ -134,18 +140,24 @@ func (ah *Accounting) Receive(peer *Peer, size uint32, msg interface{}) error {
costToLocalNode := price.For(Receiver, size) costToLocalNode := price.For(Receiver, size)
//do the accounting //do the accounting
err := ah.Add(costToLocalNode, peer) err := ah.Add(costToLocalNode, peer)
//record metrics //record metrics: just increase counters for user-facing metrics
ah.doMetrics(costToLocalNode, size, err) ah.doMetrics(costToLocalNode, size, err)
return err return err
} }
//record some metrics //record some metrics
//this is not an error handling. `err` is returned by both `Send` and `Receive`
//`err` will only be non-nil if a limit has been violated (overdraft), in which case the peer has been dropped.
//if the limit has been violated and `err` is thus not nil:
// * if the price is positive, local node has been credited; thus `err` implicitly signals the REMOTE has been dropped
// * if the price is negative, local node has been debited, thus `err` implicitly signals LOCAL node "overdraft"
func (ah *Accounting) doMetrics(price int64, size uint32, err error) { func (ah *Accounting) doMetrics(price int64, size uint32, err error) {
if price > 0 { if price > 0 {
mBalanceCredit.Inc(price) mBalanceCredit.Inc(price)
mBytesCredit.Inc(int64(size)) mBytesCredit.Inc(int64(size))
mMsgCredit.Inc(1) mMsgCredit.Inc(1)
if err != nil { if err != nil {
//increase the number of times a remote node has been dropped due to "overdraft"
mPeerDrops.Inc(1) mPeerDrops.Inc(1)
} }
} else { } else {
@ -153,6 +165,7 @@ func (ah *Accounting) doMetrics(price int64, size uint32, err error) {
mBytesDebit.Inc(int64(size)) mBytesDebit.Inc(int64(size))
mMsgDebit.Inc(1) mMsgDebit.Inc(1)
if err != nil { if err != nil {
//increase the number of times the local node has done an "overdraft" in respect to other nodes
mSelfDrops.Inc(1) mSelfDrops.Inc(1)
} }
} }

View file

@ -32,7 +32,6 @@ import (
"bufio" "bufio"
"bytes" "bytes"
"context" "context"
"errors"
"fmt" "fmt"
"io" "io"
"reflect" "reflect"
@ -361,11 +360,6 @@ func (p *Peer) handleIncoming(handle func(ctx context.Context, msg interface{})
//if the accounting hook is set, call it //if the accounting hook is set, call it
if p.spec.Hook != nil { if p.spec.Hook != nil {
if wmsg.Size != uint32(len(wmsg.Payload)) {
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) err := p.spec.Hook.Receive(p, wmsg.Size, val)
if err != nil { if err != nil {
return err return err

View file

@ -235,9 +235,13 @@ func TestProtocolHook(t *testing.T) {
runFunc := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { runFunc := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := NewPeer(p, rw, spec) peer := NewPeer(p, rw, spec)
ctx := context.TODO() ctx := context.TODO()
peer.Send(ctx, &dummyMsg{ err := peer.Send(ctx, &dummyMsg{
Content: "handshake"}) Content: "handshake"})
if err != nil {
t.Fatal(err)
}
handle := func(ctx context.Context, msg interface{}) error { handle := func(ctx context.Context, msg interface{}) error {
return nil return nil
} }
@ -335,7 +339,10 @@ func TestNoHook(t *testing.T) {
ctx := context.TODO() ctx := context.TODO()
msg := &perBytesMsgSenderPays{Content: "testBalance"} msg := &perBytesMsgSenderPays{Content: "testBalance"}
//send a message //send a message
peer.Send(ctx, msg) err := peer.Send(ctx, msg)
if err != nil {
t.Fatal(err)
}
//simulate receiving a message //simulate receiving a message
rw.msg = msg rw.msg = msg
peer.handleIncoming(func(ctx context.Context, msg interface{}) error { peer.handleIncoming(func(ctx context.Context, msg interface{}) error {