"sync"
log "github.com/sirupsen/logrus"
+ "github.com/tendermint/tmlibs/flowrate"
"gopkg.in/fatih/set.v0"
"github.com/bytom/consensus"
Addr() net.Addr
ID() string
ServiceFlag() consensus.ServiceFlag
+ TrafficStatus() (*flowrate.Status, *flowrate.Status)
TrySend(byte, interface{}) bool
}
// PeerInfo indicate peer status snap
type PeerInfo struct {
- ID string `json:"peer_id"`
- RemoteAddr string `json:"remote_addr"`
- Height uint64 `json:"height"`
- Delay uint32 `json:"delay"`
+ ID string `json:"peer_id"`
+ RemoteAddr string `json:"remote_addr"`
+ Height uint64 `json:"height"`
+ Ping string `json:"ping"`
+ Duration string `json:"duration"`
+ TotalSent int64 `json:"total_sent"`
+ TotalReceive int64 `json:"total_receive"`
+ AverageSentRate int64 `json:"average_sent_rate"`
+ AverageReceiveRate int64 `json:"average_receive_rate"`
+ CurrentSentRate int64 `json:"current_sent_rate"`
+ CurrentReceiveRate int64 `json:"current_receive_rate"`
}
type peer struct {
func (p *peer) getPeerInfo() *PeerInfo {
p.mtx.RLock()
defer p.mtx.RUnlock()
+
+ sentStatus, receiveStatus := p.TrafficStatus()
+ ping := sentStatus.Idle - receiveStatus.Idle
+ if receiveStatus.Idle > sentStatus.Idle {
+ ping = -ping
+ }
+
return &PeerInfo{
- ID: p.ID(),
- RemoteAddr: p.Addr().String(),
- Height: p.height,
+ ID: p.ID(),
+ RemoteAddr: p.Addr().String(),
+ Height: p.height,
+ Ping: ping.String(),
+ Duration: sentStatus.Duration.String(),
+ TotalSent: sentStatus.Bytes,
+ TotalReceive: receiveStatus.Bytes,
+ AverageSentRate: sentStatus.AvgRate,
+ AverageReceiveRate: receiveStatus.AvgRate,
+ CurrentSentRate: sentStatus.CurRate,
+ CurrentReceiveRate: receiveStatus.CurRate,
}
}
"time"
wire "github.com/tendermint/go-wire"
+ "github.com/tendermint/tmlibs/flowrate"
"github.com/bytom/consensus"
"github.com/bytom/protocol/bc"
p.remoteNode = node
}
+func (p *P2PPeer) TrafficStatus() (*flowrate.Status, *flowrate.Status) {
+ return nil, nil
+}
+
func (p *P2PPeer) TrySend(b byte, msg interface{}) bool {
msgBytes := wire.BinaryBytes(msg)
if p.async {
log "github.com/sirupsen/logrus"
wire "github.com/tendermint/go-wire"
cmn "github.com/tendermint/tmlibs/common"
- flow "github.com/tendermint/tmlibs/flowrate"
+ "github.com/tendermint/tmlibs/flowrate"
)
const (
conn net.Conn
bufReader *bufio.Reader
bufWriter *bufio.Writer
- sendMonitor *flow.Monitor
- recvMonitor *flow.Monitor
+ sendMonitor *flowrate.Monitor
+ recvMonitor *flowrate.Monitor
send chan struct{}
pong chan struct{}
channels []*channel
conn: conn,
bufReader: bufio.NewReaderSize(conn, minReadBufferSize),
bufWriter: bufio.NewWriterSize(conn, minWriteBufferSize),
- sendMonitor: flow.New(0, 0),
- recvMonitor: flow.New(0, 0),
+ sendMonitor: flowrate.New(0, 0),
+ recvMonitor: flowrate.New(0, 0),
send: make(chan struct{}, 1),
pong: make(chan struct{}, 1),
channelsIdx: map[byte]*channel{},
return true
}
+// TrafficStatus return the in and out traffic status
+func (c *MConnection) TrafficStatus() (*flowrate.Status, *flowrate.Status) {
+ sentStatus := c.sendMonitor.Status()
+ receiveStatus := c.recvMonitor.Status()
+ return &sentStatus, &receiveStatus
+}
+
// TrySend queues a message to be sent to channel(Nonblocking).
func (c *MConnection) TrySend(chID byte, msg interface{}) bool {
if !c.IsRunning() {
return err
}
if !compatible {
- return fmt.Errorf("Peer is on a different major version. Peer version: %v, node version: %v.", other.Version, info.Version)
+ return fmt.Errorf("Peer is on a different major version. Peer version: %v, node version: %v", other.Version, info.Version)
}
if info.Network != other.Network {
- return fmt.Errorf("Peer is on a different network. Peer network: %v, node network: %v.", other.Network, info.Network)
+ return fmt.Errorf("Peer is on a different network. Peer network: %v, node network: %v", other.Network, info.Network)
}
return nil
}
crypto "github.com/tendermint/go-crypto"
wire "github.com/tendermint/go-wire"
cmn "github.com/tendermint/tmlibs/common"
+ "github.com/tendermint/tmlibs/flowrate"
cfg "github.com/bytom/config"
"github.com/bytom/consensus"
return peerNodeInfo, nil
}
+// ID return the uuid of the peer
func (p *Peer) ID() string {
return p.Key
}
return p.mconn.Send(chID, msg)
}
+// ServiceFlag return the ServiceFlag of this peer
func (p *Peer) ServiceFlag() consensus.ServiceFlag {
services := consensus.SFFullNode
if len(p.Other) == 0 {
return fmt.Sprintf("Peer{%v %v in}", p.mconn, p.Key[:12])
}
+// TrafficStatus return the in and out traffic status
+func (p *Peer) TrafficStatus() (*flowrate.Status, *flowrate.Status) {
+ return p.mconn.TrafficStatus()
+
+}
+
// TrySend msg to the channel identified by chID byte. Immediately returns
// false if the send queue is full.
func (p *Peer) TrySend(chID byte, msg interface{}) bool {
}
}
+// SetDiscv connect the discv model to the switch
func (sw *Switch) SetDiscv(discv *discover.Network) {
sw.discv = discv
}
reactor.RemovePeer(peer, reason)
}
peer.Stop()
+
+ sentStatus, receiveStatus := peer.TrafficStatus()
+ log.WithFields(log.Fields{
+ "address": peer.Addr().String(),
+ "reason": reason,
+ "duration": sentStatus.Duration.String(),
+ "total_sent": sentStatus.Bytes,
+ "total_receive": receiveStatus.Bytes,
+ "average_sent_rate": sentStatus.AvgRate,
+ "average_receive_rate": receiveStatus.AvgRate,
+ }).Info("disconnect with peer")
}