// Peer is an interface representing a peer connected on a reactor.
type Peer interface {
 service.Service
 FlushStop()

 ID() ID               // peer's cryptographic ID
 RemoteIP() net.IP     // remote IP of the connection
 RemoteAddr() net.Addr // remote address of the connection

 IsOutbound() bool   // did we dial the peer
 IsPersistent() bool // do we redial this peer when we disconnect

 CloseConn() error // close original connection

 NodeInfo() NodeInfo // peer's info
 Status() tmconn.ConnectionStatus
 SocketAddr() *NetAddress // actual address of the socket

 // Deprecated: entities looking to act as peers should implement SendEnvelope instead.
 // Send will be removed in v0.37.
 Send(byte, []byte) bool

 // Deprecated: entities looking to act as peers should implement TrySendEnvelope instead.
 // TrySend will be removed in v0.37.
 TrySend(byte, []byte) bool

 Set(string, interface{})
 Get(string) interface{}

 SetRemovalFailed()
 GetRemovalFailed() bool
}

type EnvelopeSender interface {
 SendEnvelope(Envelope) bool
 TrySendEnvelope(Envelope) bool
}

// peer implements Peer.
//
// Before using a peer, you will need to perform a handshake on connection.
type peer struct {
 service.BaseService

 // raw peerConn and the multiplex connection
 peerConn
 mconn *tmconn.MConnection

 // peer's node info and the channel it knows about
 // channels = nodeInfo.Channels
 // cached to avoid copying nodeInfo in hasChannel
 nodeInfo NodeInfo
 channels []byte

 // User data
 Data *cmap.CMap

 metrics       *Metrics
 metricsTicker *time.Ticker
 mlc           *metricsLabelCache

 // When removal of a peer fails, we set this flag
 removalAttemptFailed bool
}

//----------------------------------------------------------

// peerConn contains the raw connection and its config.
type peerConn struct {
 outbound   bool //表示该连接是否为主动发起的出站连接（即由本地节点主动连接远程节点）。
 persistent bool //表示该连接是否为持久化连接，即在断开连接后是否需要重新建立连接。
 conn       net.Conn // source connection 表示底层的网络连接。在通讯过程中，该对象会被封装成mconn连接。

 socketAddr *NetAddress //一个指向NetAddress类型的指针，表示连接的实际地址。

 // cached RemoteIP()
 ip net.IP //一个net.IP类型的对象，表示连接对应的远程IP地址
}

func newPeerConn(
 outbound, persistent bool,
 conn net.Conn,
 socketAddr *NetAddress,
) peerConn {

 return peerConn{
  outbound:   outbound,
  persistent: persistent,
  conn:       conn,
  socketAddr: socketAddr,
 }
}

// ID only exists for SecretConnection.
// NOTE: Will panic if conn is not *SecretConnection.
func (pc peerConn) ID() ID {
 return PubKeyToID(pc.conn.(*tmconn.SecretConnection).RemotePubKey())
}

// Return the IP from the connection RemoteAddr
func (pc peerConn) RemoteIP() net.IP {
 if pc.ip != nil {
  return pc.ip
 }

 host, _, err := net.SplitHostPort(pc.conn.RemoteAddr().String())
 if err != nil {
  panic(err)
 }

 ips, err := net.LookupIP(host)
 if err != nil {
  panic(err)
 }

 pc.ip = ips[0]

 return pc.ip
}

type PeerOption func(*peer)

func newPeer(
 pc peerConn,
 mConfig tmconn.MConnConfig,
 nodeInfo NodeInfo,
 reactorsByCh map[byte]Reactor,
 msgTypeByChID map[byte]proto.Message,
 chDescs []*tmconn.ChannelDescriptor,
 onPeerError func(Peer, interface{}),
 mlc *metricsLabelCache,
 options ...PeerOption,
) *peer {
 p := &peer{
  peerConn:      pc,
  nodeInfo:      nodeInfo,
  channels:      nodeInfo.(DefaultNodeInfo).Channels,
  Data:          cmap.NewCMap(),
  metricsTicker: time.NewTicker(metricsTickerDuration),
  metrics:       NopMetrics(),
  mlc:           mlc,
 }

 p.mconn = createMConnection(
  pc.conn,
  p,
  reactorsByCh,
  msgTypeByChID,
  chDescs,
  onPeerError,
  mConfig,
 )
 p.BaseService = *service.NewBaseService(nil, "Peer", p)
 for _, option := range options {
  option(p)
 }

 return p
}

//------------------------------------------------------------------
// helper funcs

func createMConnection(
 conn net.Conn,
 p *peer,
 reactorsByCh map[byte]Reactor,
 msgTypeByChID map[byte]proto.Message,
 chDescs []*tmconn.ChannelDescriptor,
 onPeerError func(Peer, interface{}),
 config tmconn.MConnConfig,
) *tmconn.MConnection {

 onReceive := func(chID byte, msgBytes []byte) {
  reactor := reactorsByCh[chID]
  if reactor == nil {
   // Note that its ok to panic here as it's caught in the conn._recover,
   // which does onPeerError.
   panic(fmt.Sprintf("Unknown channel %X", chID))
  }
  mt := msgTypeByChID[chID]
  msg := proto.Clone(mt)
  err := proto.Unmarshal(msgBytes, msg)
  if err != nil {
   panic(fmt.Errorf("unmarshaling message: %s into type: %s", err, reflect.TypeOf(mt)))
  }
  labels := []string{
   "peer_id", string(p.ID()),
   "chID", fmt.Sprintf("%#x", chID),
  }
  if w, ok := msg.(Unwrapper); ok {
   msg, err = w.Unwrap()
   if err != nil {
    panic(fmt.Errorf("unwrapping message: %s", err))
   }
  }
  p.metrics.PeerReceiveBytesTotal.With(labels...).Add(float64(len(msgBytes)))
  p.metrics.MessageReceiveBytesTotal.With("message_type", p.mlc.ValueToMetricLabel(msg)).Add(float64(len(msgBytes)))
  if nr, ok := reactor.(EnvelopeReceiver); ok {
   nr.ReceiveEnvelope(Envelope{
    ChannelID: chID,
    Src:       p,
    Message:   msg,
   })
  } else {
   reactor.Receive(chID, p, msgBytes)
  }
 }

 onError := func(r interface{}) {
  onPeerError(p, r)
 }

 return tmconn.NewMConnectionWithConfig(
  conn,
  chDescs,
  onReceive,
  onError,
  config,
 )
}

// OnStart implements BaseService.
func (p *peer) OnStart() error {
 if err := p.BaseService.OnStart(); err != nil {
  return err
 }

 if err := p.mconn.Start(); err != nil {
  return err
 }

 go p.metricsReporter()
 return nil
}

// FlushStop mimics OnStop but additionally ensures that all successful
// SendEnvelope() calls will get flushed before closing the connection.
// NOTE: it is not safe to call this method more than once.
func (p *peer) FlushStop() {
 p.metricsTicker.Stop()
 p.BaseService.OnStop()
 p.mconn.FlushStop() // stop everything and close the conn
}

// OnStop implements BaseService.
func (p *peer) OnStop() {
 p.metricsTicker.Stop()
 p.BaseService.OnStop()
 if err := p.mconn.Stop(); err != nil { // stop everything and close the conn
  p.Logger.Debug("Error while stopping peer", "err", err)
 }
}

type MConnection struct {
 service.BaseService

 conn          net.Conn
 bufConnReader *bufio.Reader
 bufConnWriter *bufio.Writer
 sendMonitor   *flow.Monitor
 recvMonitor   *flow.Monitor
 send          chan struct{}
 pong          chan struct{}
 channels      []*Channel
 channelsIdx   map[byte]*Channel
 onReceive     receiveCbFunc
 onError       errorCbFunc
 errored       uint32
 config        MConnConfig

 // Closing quitSendRoutine will cause the sendRoutine to eventually quit.
 // doneSendRoutine is closed when the sendRoutine actually quits.
 quitSendRoutine chan struct{}
 doneSendRoutine chan struct{}

 // Closing quitRecvRouting will cause the recvRouting to eventually quit.
 quitRecvRoutine chan struct{}

 // used to ensure FlushStop and OnStop
 // are safe to call concurrently.
 stopMtx tmsync.Mutex

 flushTimer *timer.ThrottleTimer // flush writes as necessary but throttled.
 pingTimer  *time.Ticker         // send pings periodically

 // close conn if pong is not received in pongTimeout
 pongTimer     *time.Timer
 pongTimeoutCh chan bool // true - timeout, false - peer sent pong

 chStatsTimer *time.Ticker // update channel stats periodically

 created time.Time // time of creation

 _maxPacketMsgSize int
}

type MConnConfig struct {
 SendRate int64 `mapstructure:"send_rate"`
 RecvRate int64 `mapstructure:"recv_rate"`

 // Maximum payload size
 MaxPacketMsgPayloadSize int `mapstructure:"max_packet_msg_payload_size"`

 // Interval to flush writes (throttled)
 FlushThrottle time.Duration `mapstructure:"flush_throttle"`

 // Interval to send pings
 PingInterval time.Duration `mapstructure:"ping_interval"`

 // Maximum wait time for pongs
 PongTimeout time.Duration `mapstructure:"pong_timeout"`
}

// NewMConnectionWithConfig wraps net.Conn and creates multiplex connection with a config
func NewMConnectionWithConfig(
 conn net.Conn,
 chDescs []*ChannelDescriptor,
 onReceive receiveCbFunc,
 onError errorCbFunc,
 config MConnConfig,
) *MConnection {
 if config.PongTimeout >= config.PingInterval {
  panic("pongTimeout must be less than pingInterval (otherwise, next ping will reset pong timer)")
 }

 mconn := &MConnection{
  conn:          conn,
  bufConnReader: bufio.NewReaderSize(conn, minReadBufferSize),
  bufConnWriter: bufio.NewWriterSize(conn, minWriteBufferSize),
  sendMonitor:   flow.New(0, 0),
  recvMonitor:   flow.New(0, 0),
  send:          make(chan struct{}, 1),
  pong:          make(chan struct{}, 1),
  onReceive:     onReceive,
  onError:       onError,
  config:        config,
  created:       time.Now(),
 }

 // Create channels
 var channelsIdx = map[byte]*Channel{}
 var channels = []*Channel{}

 for _, desc := range chDescs {
  channel := newChannel(mconn, *desc)
  channelsIdx[channel.desc.ID] = channel
  channels = append(channels, channel)
 }
 mconn.channels = channels
 mconn.channelsIdx = channelsIdx

 mconn.BaseService = *service.NewBaseService(nil, "MConnection", mconn)

 // maxPacketMsgSize() is a bit heavy, so call just once
 mconn._maxPacketMsgSize = mconn.maxPacketMsgSize()

 return mconn
}

// TODO: lowercase.
// NOTE: not goroutine-safe.
type Channel struct {
 conn          *MConnection
 desc          ChannelDescriptor
 sendQueue     chan []byte
 sendQueueSize int32 // atomic.
 recving       []byte
 sending       []byte
 recentlySent  int64 // exponential moving average

 maxPacketMsgPayloadSize int

 Logger log.Logger
}

func newChannel(conn *MConnection, desc ChannelDescriptor) *Channel {
 desc = desc.FillDefaults()
 if desc.Priority <= 0 {
  panic("Channel default priority must be a positive integer")
 }
 return &Channel{
  conn:                    conn,
  desc:                    desc,
  sendQueue:               make(chan []byte, desc.SendQueueCapacity),
  recving:                 make([]byte, 0, desc.RecvBufferCapacity),
  maxPacketMsgPayloadSize: conn.config.MaxPacketMsgPayloadSize,
 }
}

// Queues message to send to this channel.
// Goroutine-safe
// Times out (and returns false) after defaultSendTimeout
func (ch *Channel) sendBytes(bytes []byte) bool {
 select {
 case ch.sendQueue <- bytes:
  atomic.AddInt32(&ch.sendQueueSize, 1)
  return true
 case <-time.After(defaultSendTimeout):
  return false
 }
}

// Queues message to send to this channel.
// Nonblocking, returns true if successful.
// Goroutine-safe
func (ch *Channel) trySendBytes(bytes []byte) bool {
 select {
 case ch.sendQueue <- bytes:
  atomic.AddInt32(&ch.sendQueueSize, 1)
  return true
 default:
  return false
 }
}

// Goroutine-safe
func (ch *Channel) loadSendQueueSize() (size int) {
 return int(atomic.LoadInt32(&ch.sendQueueSize))
}

// Goroutine-safe
// Use only as a heuristic.
func (ch *Channel) canSend() bool {
 return ch.loadSendQueueSize() < defaultSendQueueCapacity
}

// Returns true if any PacketMsgs are pending to be sent.
// Call before calling nextPacketMsg()
// Goroutine-safe
func (ch *Channel) isSendPending() bool {
 if len(ch.sending) == 0 {
  if len(ch.sendQueue) == 0 {
   return false
  }
  ch.sending = <-ch.sendQueue
 }
 return true
}

// Creates a new PacketMsg to send.
// Not goroutine-safe
func (ch *Channel) nextPacketMsg() tmp2p.PacketMsg {
 packet := tmp2p.PacketMsg{ChannelID: int32(ch.desc.ID)}
 maxSize := ch.maxPacketMsgPayloadSize
 packet.Data = ch.sending[:tmmath.MinInt(maxSize, len(ch.sending))]
 if len(ch.sending) <= maxSize {
  packet.EOF = true
  ch.sending = nil
  atomic.AddInt32(&ch.sendQueueSize, -1) // decrement sendQueueSize
 } else {
  packet.EOF = false
  ch.sending = ch.sending[tmmath.MinInt(maxSize, len(ch.sending)):]
 }
 return packet
}

// Writes next PacketMsg to w and updates c.recentlySent.
// Not goroutine-safe
func (ch *Channel) writePacketMsgTo(w io.Writer) (n int, err error) {
 packet := ch.nextPacketMsg()
 n, err = protoio.NewDelimitedWriter(w).WriteMsg(mustWrapPacket(&packet))
 atomic.AddInt64(&ch.recentlySent, int64(n))
 return
}

// Handles incoming PacketMsgs. It returns a message bytes if message is
// complete. NOTE message bytes may change on next call to recvPacketMsg.
// Not goroutine-safe
func (ch *Channel) recvPacketMsg(packet tmp2p.PacketMsg) ([]byte, error) {
 ch.Logger.Debug("Read PacketMsg", "conn", ch.conn, "packet", packet)
 var recvCap, recvReceived = ch.desc.RecvMessageCapacity, len(ch.recving) + len(packet.Data)
 if recvCap < recvReceived {
  return nil, fmt.Errorf("received message exceeds available capacity: %v < %v", recvCap, recvReceived)
 }
 ch.recving = append(ch.recving, packet.Data...)
 if packet.EOF {
  msgBytes := ch.recving

  // clear the slice without re-allocating.
  // http://stackoverflow.com/questions/16971741/how-do-you-clear-a-slice-in-go
  //   suggests this could be a memory leak, but we might as well keep the memory for the channel until it closes,
  // at which point the recving slice stops being used and should be garbage collected
  ch.recving = ch.recving[:0] // make([]byte, 0, ch.desc.RecvBufferCapacity)
  return msgBytes, nil
 }
 return nil, nil
}

// Call this periodically to update stats for throttling purposes.
// Not goroutine-safe
func (ch *Channel) updateStats() {
 // Exponential decay of stats.
 // TODO: optimize.
 atomic.StoreInt64(&ch.recentlySent, int64(float64(atomic.LoadInt64(&ch.recentlySent))*0.8))
}

// OnStart implements BaseService
func (c *MConnection) OnStart() error {
 if err := c.BaseService.OnStart(); err != nil {
  return err
 }
 c.flushTimer = timer.NewThrottleTimer("flush", c.config.FlushThrottle)
 c.pingTimer = time.NewTicker(c.config.PingInterval)
 c.pongTimeoutCh = make(chan bool, 1)
 c.chStatsTimer = time.NewTicker(updateStats)
 c.quitSendRoutine = make(chan struct{})
 c.doneSendRoutine = make(chan struct{})
 c.quitRecvRoutine = make(chan struct{})
 go c.sendRoutine()
 go c.recvRoutine()
 return nil
}

// stopServices stops the BaseService and timers and closes the quitSendRoutine.
// if the quitSendRoutine was already closed, it returns true, otherwise it returns false.
// It uses the stopMtx to ensure only one of FlushStop and OnStop can do this at a time.
func (c *MConnection) stopServices() (alreadyStopped bool) {
 c.stopMtx.Lock()
 defer c.stopMtx.Unlock()

 select {
 case <-c.quitSendRoutine:
  // already quit
  return true
 default:
 }

 select {
 case <-c.quitRecvRoutine:
  // already quit
  return true
 default:
 }

 c.BaseService.OnStop()
 c.flushTimer.Stop()
 c.pingTimer.Stop()
 c.chStatsTimer.Stop()

 // inform the recvRouting that we are shutting down
 close(c.quitRecvRoutine)
 close(c.quitSendRoutine)
 return false
}

// FlushStop replicates the logic of OnStop.
// It additionally ensures that all successful
// .Send() calls will get flushed before closing
// the connection.
func (c *MConnection) FlushStop() {
 if c.stopServices() {
  return
 }

 // this block is unique to FlushStop
 {
  // wait until the sendRoutine exits
  // so we dont race on calling sendSomePacketMsgs
  <-c.doneSendRoutine

  // Send and flush all pending msgs.
  // Since sendRoutine has exited, we can call this
  // safely
  eof := c.sendSomePacketMsgs()
  for !eof {
   eof = c.sendSomePacketMsgs()
  }
  c.flush()

  // Now we can close the connection
 }

 c.conn.Close()

 // We can't close pong safely here because
 // recvRoutine may write to it after we've stopped.
 // Though it doesn't need to get closed at all,
 // we close it @ recvRoutine.

 // c.Stop()
}

// OnStop implements BaseService
func (c *MConnection) OnStop() {
 if c.stopServices() {
  return
 }

 c.conn.Close()

 // We can't close pong safely here because
 // recvRoutine may write to it after we've stopped.
 // Though it doesn't need to get closed at all,
 // we close it @ recvRoutine.
}

// sendRoutine polls for packets to send from channels.
func (c *MConnection) sendRoutine() {
 defer c._recover()

 protoWriter := protoio.NewDelimitedWriter(c.bufConnWriter)

FOR_LOOP:
 for {
  var _n int
  var err error
 SELECTION:
  select {
  case <-c.flushTimer.Ch:
   // NOTE: flushTimer.Set() must be called every time
   // something is written to .bufConnWriter.
   c.flush()
  case <-c.chStatsTimer.C:
   for _, channel := range c.channels {
    channel.updateStats()
   }
  case <-c.pingTimer.C:
   c.Logger.Debug("Send Ping")
   _n, err = protoWriter.WriteMsg(mustWrapPacket(&tmp2p.PacketPing{}))
   if err != nil {
    c.Logger.Error("Failed to send PacketPing", "err", err)
    break SELECTION
   }
   c.sendMonitor.Update(_n)
   c.Logger.Debug("Starting pong timer", "dur", c.config.PongTimeout)
   c.pongTimer = time.AfterFunc(c.config.PongTimeout, func() {
    select {
    case c.pongTimeoutCh <- true:
    default:
    }
   })
   c.flush()
  case timeout := <-c.pongTimeoutCh:
   if timeout {
    c.Logger.Debug("Pong timeout")
    err = errors.New("pong timeout")
   } else {
    c.stopPongTimer()
   }
  case <-c.pong:
   c.Logger.Debug("Send Pong")
   _n, err = protoWriter.WriteMsg(mustWrapPacket(&tmp2p.PacketPong{}))
   if err != nil {
    c.Logger.Error("Failed to send PacketPong", "err", err)
    break SELECTION
   }
   c.sendMonitor.Update(_n)
   c.flush()
  case <-c.quitSendRoutine:
   break FOR_LOOP
  case <-c.send:
   // Send some PacketMsgs
   eof := c.sendSomePacketMsgs()
   if !eof {
    // Keep sendRoutine awake.
    select {
    case c.send <- struct{}{}:
    default:
    }
   }
  }

  if !c.IsRunning() {
   break FOR_LOOP
  }
  if err != nil {
   c.Logger.Error("Connection failed @ sendRoutine", "conn", c, "err", err)
   c.stopForError(err)
   break FOR_LOOP
  }
 }

 // Cleanup
 c.stopPongTimer()
 close(c.doneSendRoutine)
}

// Returns true if messages from channels were exhausted.
// Blocks in accordance to .sendMonitor throttling.
func (c *MConnection) sendSomePacketMsgs() bool {
 // Block until .sendMonitor says we can write.
 // Once we're ready we send more than we asked for,
 // but amortized it should even out.
 c.sendMonitor.Limit(c._maxPacketMsgSize, atomic.LoadInt64(&c.config.SendRate), true)

 // Now send some PacketMsgs.
 for i := 0; i < numBatchPacketMsgs; i++ {
  if c.sendPacketMsg() {
   return true
  }
 }
 return false
}

// Returns true if messages from channels were exhausted.
func (c *MConnection) sendPacketMsg() bool {
 // Choose a channel to create a PacketMsg from.
 // The chosen channel will be the one whose recentlySent/priority is the least.
 var leastRatio float32 = math.MaxFloat32
 var leastChannel *Channel
 for _, channel := range c.channels {
  // If nothing to send, skip this channel
  if !channel.isSendPending() {
   continue
  }
  // Get ratio, and keep track of lowest ratio.
  ratio := float32(channel.recentlySent) / float32(channel.desc.Priority)
  if ratio < leastRatio {
   leastRatio = ratio
   leastChannel = channel
  }
 }

 // Nothing to send?
 if leastChannel == nil {
  return true
 }
 // c.Logger.Info("Found a msgPacket to send")

 // Make & send a PacketMsg from this channel
 _n, err := leastChannel.writePacketMsgTo(c.bufConnWriter)
 if err != nil {
  c.Logger.Error("Failed to write PacketMsg", "err", err)
  c.stopForError(err)
  return true
 }
 c.sendMonitor.Update(_n)
 c.flushTimer.Set()
 return false
}

// recvRoutine reads PacketMsgs and reconstructs the message using the channels' "recving" buffer.
// After a whole message has been assembled, it's pushed to onReceive().
// Blocks depending on how the connection is throttled.
// Otherwise, it never blocks.
func (c *MConnection) recvRoutine() {
 defer c._recover()

 protoReader := protoio.NewDelimitedReader(c.bufConnReader, c._maxPacketMsgSize)

FOR_LOOP:
 for {
  // Block until .recvMonitor says we can read.
  c.recvMonitor.Limit(c._maxPacketMsgSize, atomic.LoadInt64(&c.config.RecvRate), true)

  // Peek into bufConnReader for debugging
  /*
   if numBytes := c.bufConnReader.Buffered(); numBytes > 0 {
    bz, err := c.bufConnReader.Peek(tmmath.MinInt(numBytes, 100))
    if err == nil {
     // return
    } else {
     c.Logger.Debug("Error peeking connection buffer", "err", err)
     // return nil
    }
    c.Logger.Info("Peek connection buffer", "numBytes", numBytes, "bz", bz)
   }
  */

  // Read packet type
  var packet tmp2p.Packet

  _n, err := protoReader.ReadMsg(&packet)
  c.recvMonitor.Update(_n)
  if err != nil {
   // stopServices was invoked and we are shutting down
   // receiving is excpected to fail since we will close the connection
   select {
   case <-c.quitRecvRoutine:
    break FOR_LOOP
   default:
   }

   if c.IsRunning() {
    if err == io.EOF {
     c.Logger.Info("Connection is closed @ recvRoutine (likely by the other side)", "conn", c)
    } else {
     c.Logger.Debug("Connection failed @ recvRoutine (reading byte)", "conn", c, "err", err)
    }
    c.stopForError(err)
   }
   break FOR_LOOP
  }

  // Read more depending on packet type.
  switch pkt := packet.Sum.(type) {
  case *tmp2p.Packet_PacketPing:
   // TODO: prevent abuse, as they cause flush()'s.
   // https://github.com/tendermint/tendermint/issues/1190
   c.Logger.Debug("Receive Ping")
   select {
   case c.pong <- struct{}{}:
   default:
    // never block
   }
  case *tmp2p.Packet_PacketPong:
   c.Logger.Debug("Receive Pong")
   select {
   case c.pongTimeoutCh <- false:
   default:
    // never block
   }
  case *tmp2p.Packet_PacketMsg:
   channelID := byte(pkt.PacketMsg.ChannelID)
   channel, ok := c.channelsIdx[channelID]
   if pkt.PacketMsg.ChannelID < 0 || pkt.PacketMsg.ChannelID > math.MaxUint8 || !ok || channel == nil {
    err := fmt.Errorf("unknown channel %X", pkt.PacketMsg.ChannelID)
    c.Logger.Debug("Connection failed @ recvRoutine", "conn", c, "err", err)
    c.stopForError(err)
    break FOR_LOOP
   }

   msgBytes, err := channel.recvPacketMsg(*pkt.PacketMsg)
   if err != nil {
    if c.IsRunning() {
     c.Logger.Debug("Connection failed @ recvRoutine", "conn", c, "err", err)
     c.stopForError(err)
    }
    break FOR_LOOP
   }
   if msgBytes != nil {
    c.Logger.Debug("Received bytes", "chID", channelID, "msgBytes", msgBytes)
    // NOTE: This means the reactor.Receive runs in the same thread as the p2p recv routine
    c.onReceive(channelID, msgBytes)
   }
  default:
   err := fmt.Errorf("unknown message type %v", reflect.TypeOf(packet))
   c.Logger.Error("Connection failed @ recvRoutine", "conn", c, "err", err)
   c.stopForError(err)
   break FOR_LOOP
  }
 }

 // Cleanup
 close(c.pong)
 for range c.pong {
  // Drain
 }
}

// accept is the container to carry the upgraded connection and NodeInfo from an
// asynchronously running routine to the Accept method.
type accept struct {
 netAddr  *NetAddress
 conn     net.Conn
 nodeInfo NodeInfo
 err      error
}

// peerConfig is used to bundle data we need to fully setup a Peer with an
// MConn, provided by the caller of Accept and Dial (currently the Switch). This
// a temporary measure until reactor setup is less dynamic and we introduce the
// concept of PeerBehaviour to communicate about significant Peer lifecycle
// events.
// TODO(xla): Refactor out with more static Reactor setup and PeerBehaviour.
type peerConfig struct {
 chDescs     []*conn.ChannelDescriptor
 onPeerError func(Peer, interface{})
 outbound    bool
 // isPersistent allows you to set a function, which, given socket address
 // (for outbound peers) OR self-reported address (for inbound peers), tells
 // if the peer is persistent or not.
 isPersistent  func(*NetAddress) bool
 reactorsByCh  map[byte]Reactor
 msgTypeByChID map[byte]proto.Message
 metrics       *Metrics
 mlc           *metricsLabelCache
}

// Transport emits and connects to Peers. The implementation of Peer is left to
// the transport. Each transport is also responsible to filter establishing
// peers specific to its domain.
type Transport interface {
 // Listening address.
 NetAddress() NetAddress

 // Accept returns a newly connected Peer.
 Accept(peerConfig) (Peer, error)

 // Dial connects to the Peer for the address.
 Dial(NetAddress, peerConfig) (Peer, error)

 // Cleanup any resources associated with Peer.
 Cleanup(Peer)
}

// transportLifecycle bundles the methods for callers to control start and stop
// behaviour.
type transportLifecycle interface {
 Close() error
 Listen(NetAddress) error
}

// Test multiplexTransport for interface completeness.
var _ Transport = (*MultiplexTransport)(nil)
var _ transportLifecycle = (*MultiplexTransport)(nil)

// ConnFilterFunc to be implemented by filter hooks after a new connection has
// been established. The set of exisiting connections is passed along together
// with all resolved IPs for the new connection.
type ConnFilterFunc func(ConnSet, net.Conn, []net.IP) error

// ConnDuplicateIPFilter resolves and keeps all ips for an incoming connection
// and refuses new ones if they come from a known ip.
func ConnDuplicateIPFilter() ConnFilterFunc {
 return func(cs ConnSet, c net.Conn, ips []net.IP) error {
  for _, ip := range ips {
   if cs.HasIP(ip) {
    return ErrRejected{
     conn:        c,
     err:         fmt.Errorf("ip<%v> already connected", ip),
     isDuplicate: true,
    }
   }
  }

  return nil
 }
}

// MultiplexTransport accepts and dials tcp connections and upgrades them to
// multiplexed peers.
type MultiplexTransport struct {
 netAddr                NetAddress
 listener               net.Listener
 maxIncomingConnections int // see MaxIncomingConnections

 acceptc chan accept
 closec  chan struct{}

 // Lookup table for duplicate ip and id checks.
 conns       ConnSet
 connFilters []ConnFilterFunc

 dialTimeout      time.Duration
 filterTimeout    time.Duration
 handshakeTimeout time.Duration
 nodeInfo         NodeInfo
 nodeKey          NodeKey
 resolver         IPResolver

 // TODO(xla): This config is still needed as we parameterise peerConn and
 // peer currently. All relevant configuration should be refactored into options
 // with sane defaults.
 mConfig conn.MConnConfig
}

// NewMultiplexTransport returns a tcp connected multiplexed peer.
func NewMultiplexTransport(
 nodeInfo NodeInfo,
 nodeKey NodeKey,
 mConfig conn.MConnConfig,
) *MultiplexTransport {
 return &MultiplexTransport{
  acceptc:          make(chan accept),
  closec:           make(chan struct{}),
  dialTimeout:      defaultDialTimeout,
  filterTimeout:    defaultFilterTimeout,
  handshakeTimeout: defaultHandshakeTimeout,
  mConfig:          mConfig,
  nodeInfo:         nodeInfo,
  nodeKey:          nodeKey,
  conns:            NewConnSet(),
  resolver:         net.DefaultResolver,
 }
}

// Accept implements Transport.
func (mt *MultiplexTransport) Accept(cfg peerConfig) (Peer, error) {
 select {
 // This case should never have any side-effectful/blocking operations to
 // ensure that quality peers are ready to be used.
 case a := <-mt.acceptc:
  if a.err != nil {
   return nil, a.err
  }

  cfg.outbound = false

  return mt.wrapPeer(a.conn, a.nodeInfo, cfg, a.netAddr), nil
 case <-mt.closec:
  return nil, ErrTransportClosed{}
 }
}

func (mt *MultiplexTransport) wrapPeer(
 c net.Conn,
 ni NodeInfo,
 cfg peerConfig,
 socketAddr *NetAddress,
) Peer {

 persistent := false
 if cfg.isPersistent != nil {
  if cfg.outbound {
   persistent = cfg.isPersistent(socketAddr)
  } else {
   selfReportedAddr, err := ni.NetAddress()
   if err == nil {
    persistent = cfg.isPersistent(selfReportedAddr)
   }
  }
 }

 peerConn := newPeerConn(
  cfg.outbound,
  persistent,
  c,
  socketAddr,
 )

 p := newPeer(
  peerConn,
  mt.mConfig,
  ni,
  cfg.reactorsByCh,
  cfg.msgTypeByChID,
  cfg.chDescs,
  cfg.onPeerError,
  cfg.mlc,
  PeerMetrics(cfg.metrics),
 )

 return p
}

// Dial implements Transport.
func (mt *MultiplexTransport) Dial(
 addr NetAddress,
 cfg peerConfig,
) (Peer, error) {
 c, err := addr.DialTimeout(mt.dialTimeout)
 if err != nil {
  return nil, err
 }

 // TODO(xla): Evaluate if we should apply filters if we explicitly dial.
 if err := mt.filterConn(c); err != nil {
  return nil, err
 }

 secretConn, nodeInfo, err := mt.upgrade(c, &addr)
 if err != nil {
  return nil, err
 }

 cfg.outbound = true

 p := mt.wrapPeer(secretConn, nodeInfo, cfg, &addr)

 return p, nil
}

func (mt *MultiplexTransport) filterConn(c net.Conn) (err error) {
 defer func() {
  if err != nil {
   _ = c.Close()
  }
 }()

 // Reject if connection is already present.
 if mt.conns.Has(c) {
  return ErrRejected{conn: c, isDuplicate: true}
 }

 // Resolve ips for incoming conn.
 ips, err := resolveIPs(mt.resolver, c)
 if err != nil {
  return err
 }

 errc := make(chan error, len(mt.connFilters))

 for _, f := range mt.connFilters {
  go func(f ConnFilterFunc, c net.Conn, ips []net.IP, errc chan<- error) {
   errc <- f(mt.conns, c, ips)
  }(f, c, ips, errc)
 }

 for i := 0; i < cap(errc); i++ {
  select {
  case err := <-errc:
   if err != nil {
    return ErrRejected{conn: c, err: err, isFiltered: true}
   }
  case <-time.After(mt.filterTimeout):
   return ErrFilterTimeout{}
  }

 }

 mt.conns.Set(c, ips)

 return nil
}

func resolveIPs(resolver IPResolver, c net.Conn) ([]net.IP, error) {
 host, _, err := net.SplitHostPort(c.RemoteAddr().String())
 if err != nil {
  return nil, err
 }

 addrs, err := resolver.LookupIPAddr(context.Background(), host)
 if err != nil {
  return nil, err
 }

 ips := []net.IP{}

 for _, addr := range addrs {
  ips = append(ips, addr.IP)
 }

 return ips, nil
}

func (mt *MultiplexTransport) upgrade(
 c net.Conn,
 dialedAddr *NetAddress,
) (secretConn *conn.SecretConnection, nodeInfo NodeInfo, err error) {
 defer func() {
  if err != nil {
   _ = mt.cleanup(c)
  }
 }()

 secretConn, err = upgradeSecretConn(c, mt.handshakeTimeout, mt.nodeKey.PrivKey)
 if err != nil {
  return nil, nil, ErrRejected{
   conn:          c,
   err:           fmt.Errorf("secret conn failed: %v", err),
   isAuthFailure: true,
  }
 }

 // For outgoing conns, ensure connection key matches dialed key.
 connID := PubKeyToID(secretConn.RemotePubKey())
 if dialedAddr != nil {
  if dialedID := dialedAddr.ID; connID != dialedID {
   return nil, nil, ErrRejected{
    conn: c,
    id:   connID,
    err: fmt.Errorf(
     "conn.ID (%v) dialed ID (%v) mismatch",
     connID,
     dialedID,
    ),
    isAuthFailure: true,
   }
  }
 }

 nodeInfo, err = handshake(secretConn, mt.handshakeTimeout, mt.nodeInfo)
 if err != nil {
  return nil, nil, ErrRejected{
   conn:          c,
   err:           fmt.Errorf("handshake failed: %v", err),
   isAuthFailure: true,
  }
 }

 if err := nodeInfo.Validate(); err != nil {
  return nil, nil, ErrRejected{
   conn:              c,
   err:               err,
   isNodeInfoInvalid: true,
  }
 }

 // Ensure connection key matches self reported key.
 if connID != nodeInfo.ID() {
  return nil, nil, ErrRejected{
   conn: c,
   id:   connID,
   err: fmt.Errorf(
    "conn.ID (%v) NodeInfo.ID (%v) mismatch",
    connID,
    nodeInfo.ID(),
   ),
   isAuthFailure: true,
  }
 }

 // Reject self.
 if mt.nodeInfo.ID() == nodeInfo.ID() {
  return nil, nil, ErrRejected{
   addr:   *NewNetAddress(nodeInfo.ID(), c.RemoteAddr()),
   conn:   c,
   id:     nodeInfo.ID(),
   isSelf: true,
  }
 }

 if err := mt.nodeInfo.CompatibleWith(nodeInfo); err != nil {
  return nil, nil, ErrRejected{
   conn:           c,
   err:            err,
   id:             nodeInfo.ID(),
   isIncompatible: true,
  }
 }

 return secretConn, nodeInfo, nil
}

func upgradeSecretConn(
 c net.Conn,
 timeout time.Duration,
 privKey crypto.PrivKey,
) (*conn.SecretConnection, error) {
 if err := c.SetDeadline(time.Now().Add(timeout)); err != nil {
  return nil, err
 }

 sc, err := conn.MakeSecretConnection(c, privKey)
 if err != nil {
  return nil, err
 }

 return sc, sc.SetDeadline(time.Time{})
}

func handshake(
 c net.Conn,
 timeout time.Duration,
 nodeInfo NodeInfo,
) (NodeInfo, error) {
 if err := c.SetDeadline(time.Now().Add(timeout)); err != nil {
  return nil, err
 }

 var (
  errc = make(chan error, 2)

  pbpeerNodeInfo tmp2p.DefaultNodeInfo
  peerNodeInfo   DefaultNodeInfo
  ourNodeInfo    = nodeInfo.(DefaultNodeInfo)
 )

 go func(errc chan<- error, c net.Conn) {
  _, err := protoio.NewDelimitedWriter(c).WriteMsg(ourNodeInfo.ToProto())
  errc <- err
 }(errc, c)
 go func(errc chan<- error, c net.Conn) {
  protoReader := protoio.NewDelimitedReader(c, MaxNodeInfoSize())
  _, err := protoReader.ReadMsg(&pbpeerNodeInfo)
  errc <- err
 }(errc, c)

 for i := 0; i < cap(errc); i++ {
  err := <-errc
  if err != nil {
   return nil, err
  }
 }

 peerNodeInfo, err := DefaultNodeInfoFromToProto(&pbpeerNodeInfo)
 if err != nil {
  return nil, err
 }

 return peerNodeInfo, c.SetDeadline(time.Time{})
}

func handshake(
 c net.Conn,
 timeout time.Duration,
 nodeInfo NodeInfo,
) (NodeInfo, error) {
 if err := c.SetDeadline(time.Now().Add(timeout)); err != nil {
  return nil, err
 }

 var (
  errc = make(chan error, 2)

  pbpeerNodeInfo tmp2p.DefaultNodeInfo
  peerNodeInfo   DefaultNodeInfo
  ourNodeInfo    = nodeInfo.(DefaultNodeInfo)
 )

 go func(errc chan<- error, c net.Conn) {
  _, err := protoio.NewDelimitedWriter(c).WriteMsg(ourNodeInfo.ToProto())
  errc <- err
 }(errc, c)
 go func(errc chan<- error, c net.Conn) {
  protoReader := protoio.NewDelimitedReader(c, MaxNodeInfoSize())
  _, err := protoReader.ReadMsg(&pbpeerNodeInfo)
  errc <- err
 }(errc, c)

 for i := 0; i < cap(errc); i++ {
  err := <-errc
  if err != nil {
   return nil, err
  }
 }

 peerNodeInfo, err := DefaultNodeInfoFromToProto(&pbpeerNodeInfo)
 if err != nil {
  return nil, err
 }

 return peerNodeInfo, c.SetDeadline(time.Time{})
}

// Cleanup removes the given address from the connections set and
// closes the connection.
func (mt *MultiplexTransport) Cleanup(p Peer) {
 mt.conns.RemoveAddr(p.RemoteAddr())
 _ = p.CloseConn()
}

func (mt *MultiplexTransport) cleanup(c net.Conn) error {
 mt.conns.Remove(c)

 return c.Close()
}

// Close implements transportLifecycle.
func (mt *MultiplexTransport) Close() error {
 close(mt.closec)

 if mt.listener != nil {
  return mt.listener.Close()
 }

 return nil
}

// Listen implements transportLifecycle.
func (mt *MultiplexTransport) Listen(addr NetAddress) error {
 ln, err := net.Listen("tcp", addr.DialString())
 if err != nil {
  return err
 }

 if mt.maxIncomingConnections > 0 {
  ln = netutil.LimitListener(ln, mt.maxIncomingConnections)
 }

 mt.netAddr = addr
 mt.listener = ln

 go mt.acceptPeers()

 return nil
}

func (mt *MultiplexTransport) acceptPeers() {
 for {
  c, err := mt.listener.Accept()
  if err != nil {
   // If Close() has been called, silently exit.
   select {
   case _, ok := <-mt.closec:
    if !ok {
     return
    }
   default:
    // Transport is not closed
   }

   mt.acceptc <- accept{err: err}
   return
  }

  // Connection upgrade and filtering should be asynchronous to avoid
  // Head-of-line blocking[0].
  // Reference:  https://github.com/tendermint/tendermint/issues/2047
  //
  // [0] https://en.wikipedia.org/wiki/Head-of-line_blocking
  go func(c net.Conn) {
   defer func() {
    if r := recover(); r != nil {
     err := ErrRejected{
      conn:          c,
      err:           fmt.Errorf("recovered from panic: %v", r),
      isAuthFailure: true,
     }
     select {
     case mt.acceptc <- accept{err: err}:
     case <-mt.closec:
      // Give up if the transport was closed.
      _ = c.Close()
      return
     }
    }
   }()

   var (
    nodeInfo   NodeInfo
    secretConn *conn.SecretConnection
    netAddr    *NetAddress
   )

   err := mt.filterConn(c)
   if err == nil {
    secretConn, nodeInfo, err = mt.upgrade(c, nil)
    if err == nil {
     addr := c.RemoteAddr()
     id := PubKeyToID(secretConn.RemotePubKey())
     netAddr = NewNetAddress(id, addr)
    }
   }

   select {
   case mt.acceptc <- accept{netAddr, secretConn, nodeInfo, err}:
    // Make the upgraded peer available.
   case <-mt.closec:
    // Give up if the transport was closed.
    _ = c.Close()
    return
   }
  }(c)
 }
}

type Switch struct {
 service.BaseService //表明该结构继承自service.BaseService接口，表示Switch是一个基于服务的模块，可以启动、停止等。

 config        *config.P2PConfig //存储Tendermint P2P配置参数的数据结构，例如最大连接数、最大广播范围等。
 reactors      map[string]Reactor //一个map[string]Reactor类型的映射表，将Reactor名称映射到对应的Reactor实例上。
 chDescs       []*conn.ChannelDescriptor //存储通道描述符（Channel Descriptor）的切片，用于描述不同通道的属性信息，例如带宽、延迟等。
 reactorsByCh  map[byte]Reactor //一个map[byte]Reactor类型的映射表，将每个通道ID映射到对应的Reactor实例上。
 msgTypeByChID map[byte]proto.Message //一个map[byte]Reactor类型的映射表，将每个通道ID映射到对应的Reactor实例上。
 peers         *PeerSet //存储所有已连接的节点信息的PeerSet实例，包括其地址、公钥、连接状态等。
 dialing       *cmap.CMap //一个并发安全的map，用于存储正在进行拨号的节点信息。
 reconnecting  *cmap.CMap //一个并发安全的map，用于存储正在尝试重新建立连接的节点信息。
 nodeInfo      NodeInfo // our node info 存储本地节点信息的结构体，包括节点ID、网络地址、版本等信息。
 nodeKey       *NodeKey // our node privkey 存储本地节点私钥的结构体，用于加密通信数据和签名共识信息等。
 addrBook      AddrBook //存储所有已知节点地址的AddrBook实例，用于管理节点的网络地址信息。
 // peers addresses with whom we'll maintain constant connection
 persistentPeersAddrs []*NetAddress //一个存储持久连接节点地址的NetAddress切片，表示需要保持长时间连接的节点信息。
 unconditionalPeerIDs map[ID]struct{} //一个存储无条件连接节点ID的映射表，表示不需要进行握手验证就可以连接的节点。

 transport Transport //P2P消息传输协议的实现Transport接口实例，例如TCP或QUIC等协议的实现。

 filterTimeout time.Duration //过滤器超时时间，即在多长时间内没有活动的连接要关闭。
 peerFilters   []PeerFilterFunc //一个PeerFilterFunc切片，存储了所有的PeerFilterFunc函数，用于过滤和处理不需要的消息。

 rng *rand.Rand // seed for randomizing dial times and orders 一个Rand结构体实例，用于随机化拨号时间和顺序等参数。

 metrics *Metrics //存储P2P性能指标的Metrics结构体实例，例如网络吞吐量、延迟、丢包率等。
 mlc     *metricsLabelCache //一个metricsLabelCache结构体实例，用于缓存P2P性能指标的标签信息。
}

// PeerFilterFunc to be implemented by filter hooks after a new Peer has been
// fully setup.
type PeerFilterFunc func(IPeerSet, Peer) error

// An AddrBook represents an address book from the pex package, which is used
// to store peer addresses.
type AddrBook interface {
 AddAddress(addr *NetAddress, src *NetAddress) error //添加一个节点网络地址到地址簿中，addr表示待添加的地址，src表示该地址的来源；该方法返回一个错误类型的结果。
 AddPrivateIDs([]string) //添加一些私有ID到地址簿中，例如本地节点的ID等。
 AddOurAddress(*NetAddress) //将一个本地节点的地址添加到地址簿中。
 OurAddress(*NetAddress) bool //检查一个网络地址是否为本地节点的地址，如果是则返回true，否则返回false。
 MarkGood(ID) //标记一个节点为可靠的（即“好”的）节点，以便在进行P2P连接时优先选择这些节点。
 RemoveAddress(*NetAddress) //从地址簿中移除一个网络地址。

 HasAddress(*NetAddress) bool //从地址簿中移除一个网络地址。

 Save() //从地址簿中移除一个网络地址。
}