connection.go

package conn

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"runtime/debug"

	"fmt"
	"io"
	"math"
	"net"
	"reflect"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pkg/errors"

	amino "github.com/tendermint/go-amino"

	flow "github.com/okx/okbchain/libs/tendermint/libs/flowrate"
	"github.com/okx/okbchain/libs/tendermint/libs/log"
	tmmath "github.com/okx/okbchain/libs/tendermint/libs/math"
	"github.com/okx/okbchain/libs/tendermint/libs/service"
	"github.com/okx/okbchain/libs/tendermint/libs/timer"
)

const (
	defaultMaxPacketMsgPayloadSize = 1024

	numBatchPacketMsgs = 10
	minReadBufferSize  = 1024
	minWriteBufferSize = 65536
	updateStats        = 2 * time.Second

	// some of these defaults are written in the user config
	// flushThrottle, sendRate, recvRate
	// TODO: remove values present in config
	defaultFlushThrottle = 100 * time.Millisecond

	defaultSendQueueCapacity   = 1
	defaultRecvBufferCapacity  = 4096
	defaultRecvMessageCapacity = 22020096      // 21MB
	defaultSendRate            = int64(512000) // 500KB/s
	defaultRecvRate            = int64(512000) // 500KB/s
	defaultSendTimeout         = 10 * time.Second
	defaultPingInterval        = 60 * time.Second
	defaultPongTimeout         = 45 * time.Second
)

type logData struct {
	Params [6]interface{}
	Bytes  bytesHexStringer
	Packet PacketMsg
}

var logDataPool = &sync.Pool{
	New: func() interface{} {
		return &logData{}
	},
}

type receiveCbFunc func(chID byte, msgBytes []byte)
type errorCbFunc func(interface{})

type bytesHexStringer []byte

func (b bytesHexStringer) String() string {
	return fmt.Sprintf("%X", []byte(b))
}

/*
Each peer has one `MConnection` (multiplex connection) instance.

__multiplex__ *noun* a system or signal involving simultaneous transmission of
several messages along a single channel of communication.

Each `MConnection` handles message transmission on multiple abstract communication
`Channel`s.  Each channel has a globally unique byte id.
The byte id and the relative priorities of each `Channel` are configured upon
initialization of the connection.

There are two methods for sending messages:
	func (m MConnection) Send(chID byte, msgBytes []byte) bool {}
	func (m MConnection) TrySend(chID byte, msgBytes []byte}) bool {}

`Send(chID, msgBytes)` is a blocking call that waits until `msg` is
successfully queued for the channel with the given id byte `chID`, or until the
request times out.  The message `msg` is serialized using Go-Amino.

`TrySend(chID, msgBytes)` is a nonblocking call that returns false if the
channel's queue is full.

Inbound message bytes are handled with an onReceive callback function.
*/
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 sync.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
}

// MConnConfig is a MConnection configuration.
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"`
}

// DefaultMConnConfig returns the default config.
func DefaultMConnConfig() MConnConfig {
	return MConnConfig{
		SendRate:                defaultSendRate,
		RecvRate:                defaultRecvRate,
		MaxPacketMsgPayloadSize: defaultMaxPacketMsgPayloadSize,
		FlushThrottle:           defaultFlushThrottle,
		PingInterval:            defaultPingInterval,
		PongTimeout:             defaultPongTimeout,
	}
}

// NewMConnection wraps net.Conn and creates multiplex connection
func NewMConnection(
	conn net.Conn,
	chDescs []*ChannelDescriptor,
	onReceive receiveCbFunc,
	onError errorCbFunc,
) *MConnection {
	return NewMConnectionWithConfig(
		conn,
		chDescs,
		onReceive,
		onError,
		DefaultMConnConfig())
}

// 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
}

func (c *MConnection) SetLogger(l log.Logger) {
	c.BaseService.SetLogger(l)
	for _, ch := range c.channels {
		ch.SetLogger(l)
	}
}

// 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() // nolint: errcheck

	// 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() // nolint: errcheck

	// 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.
}

func (c *MConnection) String() string {
	return fmt.Sprintf("MConn{%v}", c.conn.RemoteAddr())
}

func (c *MConnection) flush() {
	c.Logger.Debug("Flush", "conn", c)
	err := c.bufConnWriter.Flush()
	if err != nil {
		c.Logger.Error("MConnection flush failed", "err", err)
	}
}

// Catch panics, usually caused by remote disconnects.
func (c *MConnection) _recover() {
	if r := recover(); r != nil {
		c.Logger.Error("MConnection panicked", "err", r, "stack", string(debug.Stack()))
		c.stopForError(errors.Errorf("recovered from panic: %v", r))
	}
}

func (c *MConnection) stopForError(r interface{}) {
	c.Stop()
	if atomic.CompareAndSwapUint32(&c.errored, 0, 1) {
		if c.onError != nil {
			c.onError(r)
		}
	}
}

func (c *MConnection) logSendData(msg string, chID byte, msgBytes []byte) {
	logParams := logDataPool.Get().(*logData)
	logParams.Bytes = msgBytes
	params := &logParams.Params
	params[0] = "channel"
	params[1] = chID
	params[2] = "conn"
	params[3] = c
	params[4] = "msgBytes"
	params[5] = &logParams.Bytes

	c.Logger.Debug(msg, logParams.Params[:]...)
	logDataPool.Put(logParams)
}

// Queues a message to be sent to channel.
func (c *MConnection) Send(chID byte, msgBytes []byte) bool {
	if !c.IsRunning() {
		return false
	}

	c.logSendData("Send", chID, msgBytes)

	// Send message to channel.
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.Logger.Error(fmt.Sprintf("Cannot send bytes, unknown channel %X", chID))
		return false
	}

	success := channel.sendBytes(msgBytes)
	if success {
		// Wake up sendRoutine if necessary
		select {
		case c.send <- struct{}{}:
		default:
		}
	} else {
		c.logSendData("Send failed", chID, msgBytes)
	}
	return success
}

// Queues a message to be sent to channel.
// Nonblocking, returns true if successful.
func (c *MConnection) TrySend(chID byte, msgBytes []byte) bool {
	if !c.IsRunning() {
		return false
	}

	c.Logger.Debug("TrySend", "channel", chID, "conn", c, "msgBytes", fmt.Sprintf("%X", msgBytes))

	// Send message to channel.
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.Logger.Error(fmt.Sprintf("Cannot send bytes, unknown channel %X", chID))
		return false
	}

	ok = channel.trySendBytes(msgBytes)
	if ok {
		// Wake up sendRoutine if necessary
		select {
		case c.send <- struct{}{}:
		default:
		}
	}

	return ok
}

// CanSend returns true if you can send more data onto the chID, false
// otherwise. Use only as a heuristic.
func (c *MConnection) CanSend(chID byte) bool {
	if !c.IsRunning() {
		return false
	}

	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.Logger.Error(fmt.Sprintf("Unknown channel %X", chID))
		return false
	}
	return channel.canSend()
}

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

FOR_LOOP:
	for {
		var _n int64
		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 = cdc.MarshalBinaryLengthPrefixedWriterWithRegiteredMarshaller(c.bufConnWriter, PacketPing{})
			if err != nil {
				_n, err = cdc.MarshalBinaryLengthPrefixedWriter(c.bufConnWriter, PacketPing{})
			}
			if err != nil {
				break SELECTION
			}
			c.sendMonitor.Update(int(_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 = cdc.MarshalBinaryLengthPrefixedWriterWithRegiteredMarshaller(c.bufConnWriter, PacketPong{})
			if err != nil {
				_n, err = cdc.MarshalBinaryLengthPrefixedWriter(c.bufConnWriter, PacketPong{})
			}
			if err != nil {
				break SELECTION
			}
			c.sendMonitor.Update(int(_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(int(_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()

	var packetMsg PacketMsg
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 Packet
		var _n int64
		var err error
		// _n, err = cdc.UnmarshalBinaryLengthPrefixedReader(c.bufConnReader, &packet, int64(c._maxPacketMsgSize))
		packet, _n, err = unmarshalPacketFromAminoReader(c.bufConnReader, int64(c._maxPacketMsgSize), &packetMsg)
		c.recvMonitor.Update(int(_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.Error("Connection failed @ recvRoutine (reading byte)", "conn", c, "err", err)
				}
				c.stopForError(err)
			}
			break FOR_LOOP
		}

		// Read more depending on packet type.
		switch pkt := packet.(type) {
		case 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 PacketPong:
			c.Logger.Debug("Receive Pong")
			select {
			case c.pongTimeoutCh <- false:
			default:
				// never block
			}
		case *PacketMsg:
			channel, ok := c.channelsIdx[pkt.ChannelID]
			if !ok || channel == nil {
				err := fmt.Errorf("unknown channel %X", pkt.ChannelID)
				c.Logger.Error("Connection failed @ recvRoutine", "conn", c, "err", err)
				c.stopForError(err)
				break FOR_LOOP
			}

			msgBytes, err := channel.recvPacketMsg(*pkt)
			if err != nil {
				if c.IsRunning() {
					c.Logger.Error("Connection failed @ recvRoutine", "conn", c, "err", err)
					c.stopForError(err)
				}
				break FOR_LOOP
			}
			if msgBytes != nil {
				c.logReceiveMsg(pkt.ChannelID, msgBytes)
				// NOTE: This means the reactor.Receive runs in the same thread as the p2p recv routine
				c.onReceive(pkt.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
	}
}

func (c *MConnection) logReceiveMsg(channelID byte, msgBytes []byte) {
	logParams := logDataPool.Get().(*logData)
	logParams.Bytes = msgBytes
	params := &logParams.Params
	params[0] = "chID"
	params[1] = channelID
	params[2] = "msgBytes"
	params[3] = &logParams.Bytes
	c.Logger.Debug("Received bytes", logParams.Params[:4]...)
	logDataPool.Put(logParams)
}

// not goroutine-safe
func (c *MConnection) stopPongTimer() {
	if c.pongTimer != nil {
		_ = c.pongTimer.Stop()
		c.pongTimer = nil
	}
}

// maxPacketMsgSize returns a maximum size of PacketMsg, including the overhead
// of amino encoding.
func (c *MConnection) maxPacketMsgSize() int {
	return len(cdc.MustMarshalBinaryLengthPrefixed(PacketMsg{
		ChannelID: 0x01,
		EOF:       1,
		Bytes:     make([]byte, c.config.MaxPacketMsgPayloadSize),
	})) + 10 // leave room for changes in amino
}

type ConnectionStatus struct {
	Duration    time.Duration
	SendMonitor flow.Status
	RecvMonitor flow.Status
	Channels    []ChannelStatus
}

type ChannelStatus struct {
	ID                byte
	SendQueueCapacity int
	SendQueueSize     int
	Priority          int
	RecentlySent      int64
}

func (c *MConnection) Status() ConnectionStatus {
	var status ConnectionStatus
	status.Duration = time.Since(c.created)
	status.SendMonitor = c.sendMonitor.Status()
	status.RecvMonitor = c.recvMonitor.Status()
	status.Channels = make([]ChannelStatus, len(c.channels))
	for i, channel := range c.channels {
		status.Channels[i] = ChannelStatus{
			ID:                channel.desc.ID,
			SendQueueCapacity: cap(channel.sendQueue),
			SendQueueSize:     int(atomic.LoadInt32(&channel.sendQueueSize)),
			Priority:          channel.desc.Priority,
			RecentlySent:      atomic.LoadInt64(&channel.recentlySent),
		}
	}
	return status
}

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

type ChannelDescriptor struct {
	ID                  byte
	Priority            int
	SendQueueCapacity   int
	RecvBufferCapacity  int
	RecvMessageCapacity int
}

func (chDesc ChannelDescriptor) FillDefaults() (filled ChannelDescriptor) {
	if chDesc.SendQueueCapacity == 0 {
		chDesc.SendQueueCapacity = defaultSendQueueCapacity
	}
	if chDesc.RecvBufferCapacity == 0 {
		chDesc.RecvBufferCapacity = defaultRecvBufferCapacity
	}
	if chDesc.RecvMessageCapacity == 0 {
		chDesc.RecvMessageCapacity = defaultRecvMessageCapacity
	}
	filled = chDesc
	return
}

// 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,
	}
}

func (ch *Channel) SetLogger(l log.Logger) {
	ch.Logger = l
}

var sendTimerPool = &sync.Pool{
	New: func() interface{} {
		return time.NewTimer(defaultSendTimeout)
	},
}

// Queues message to send to this channel.
// Goroutine-safe
// Times out (and returns false) after defaultSendTimeout
func (ch *Channel) sendBytes(bytes []byte) bool {
	sendTimer := sendTimerPool.Get().(*time.Timer)
	if !sendTimer.Stop() {
		select {
		case <-sendTimer.C:
		default:
		}
	}
	sendTimer.Reset(defaultSendTimeout)
	select {
	case ch.sendQueue <- bytes:
		sendTimerPool.Put(sendTimer)
		atomic.AddInt32(&ch.sendQueueSize, 1)
		return true
	case <-sendTimer.C:
		sendTimerPool.Put(sendTimer)
		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() PacketMsg {
	packet := PacketMsg{}
	packet.ChannelID = ch.desc.ID
	maxSize := ch.maxPacketMsgPayloadSize
	packet.Bytes = ch.sending[:tmmath.MinInt(maxSize, len(ch.sending))]
	if len(ch.sending) <= maxSize {
		packet.EOF = byte(0x01)
		ch.sending = nil
		atomic.AddInt32(&ch.sendQueueSize, -1) // decrement sendQueueSize
	} else {
		packet.EOF = byte(0x00)
		ch.sending = ch.sending[tmmath.MinInt(maxSize, len(ch.sending)):]
	}
	return packet
}

var packetBzSendPool = &sync.Pool{
	New: func() interface{} {
		return &bytes.Buffer{}
	},
}

// Writes next PacketMsg to w and updates c.recentlySent.
// Not goroutine-safe
func (ch *Channel) writePacketMsgTo(w io.Writer) (n int64, err error) {
	var packet = ch.nextPacketMsg()

	packetMsgTypePrefix := getPacketMsgAminoTypePrefix()
	bzSize := len(packetMsgTypePrefix) + packet.AminoSize(cdc)
	bzSizeWithLenPrefix := amino.UvarintSize(uint64(bzSize)) + bzSize

	// var buf = bytes.NewBuffer(make([]byte, 0, bzSizeWithLenPrefix))
	buf := packetBzSendPool.Get().(*bytes.Buffer)
	defer packetBzSendPool.Put(buf)
	buf.Reset()
	buf.Grow(bzSizeWithLenPrefix)

	err = amino.EncodeUvarintToBuffer(buf, uint64(bzSize))
	if err == nil {
		buf.Write(packetMsgTypePrefix)
		err = packet.MarshalAminoTo(cdc, buf)
		if err == nil && buf.Len() == bzSizeWithLenPrefix {
			bzNum := 0
			bzNum, err = w.Write(buf.Bytes())
			n = int64(bzNum)
			atomic.AddInt64(&ch.recentlySent, n)
			return
		}
	}

	n, err = cdc.MarshalBinaryLengthPrefixedWriterWithRegiteredMarshaller(w, packet)
	if err != nil {
		n, err = cdc.MarshalBinaryLengthPrefixedWriter(w, packet)
	}
	atomic.AddInt64(&ch.recentlySent, n)
	return
}

func (ch *Channel) logRecvPacketMsg(packet PacketMsg) {
	logParams := logDataPool.Get().(*logData)
	logParams.Packet = packet
	params := &logParams.Params
	params[0] = "conn"
	params[1] = ch.conn
	params[2] = "packet"
	params[3] = &logParams.Packet
	ch.Logger.Debug("Read PacketMsg", logParams.Params[:4]...)
	logDataPool.Put(logParams)
}

// 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 PacketMsg) ([]byte, error) {
	ch.logRecvPacketMsg(packet)

	var recvCap, recvReceived = ch.desc.RecvMessageCapacity, len(ch.recving) + len(packet.Bytes)
	if recvCap < recvReceived {
		return nil, fmt.Errorf("received message exceeds available capacity: %v < %v", recvCap, recvReceived)
	}
	ch.recving = append(ch.recving, packet.Bytes...)
	if packet.EOF == byte(0x01) {
		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))
}

//----------------------------------------
// Packet

type Packet interface {
	AssertIsPacket()
}

const (
	PacketPingName = "tendermint/p2p/PacketPing"
	PacketPongName = "tendermint/p2p/PacketPong"
	PacketMsgName  = "tendermint/p2p/PacketMsg"
)

func RegisterPacket(cdc *amino.Codec) {
	cdc.RegisterInterface((*Packet)(nil), nil)
	cdc.RegisterConcrete(PacketPing{}, PacketPingName, nil)
	cdc.RegisterConcrete(PacketPong{}, PacketPongName, nil)
	cdc.RegisterConcrete(&PacketMsg{}, PacketMsgName, nil)

	cdc.RegisterConcreteMarshaller(PacketPingName, func(_ *amino.Codec, i interface{}) ([]byte, error) {
		return []byte{}, nil
	})
	cdc.RegisterConcreteMarshaller(PacketPongName, func(_ *amino.Codec, i interface{}) ([]byte, error) {
		return []byte{}, nil
	})
	cdc.RegisterConcreteMarshaller(PacketMsgName, func(codec *amino.Codec, i interface{}) ([]byte, error) {
		if packet, ok := i.(PacketMsg); ok {
			return packet.MarshalToAmino(codec)
		} else if ppacket, ok := i.(*PacketMsg); ok {
			return ppacket.MarshalToAmino(codec)
		} else {
			return nil, fmt.Errorf("%v must be of type %v", i, PacketMsg{})
		}
	})
	cdc.RegisterConcreteUnmarshaller(PacketPingName, func(_ *amino.Codec, _ []byte) (interface{}, int, error) {
		return PacketPing{}, 0, nil
	})
	cdc.RegisterConcreteUnmarshaller(PacketPongName, func(_ *amino.Codec, _ []byte) (interface{}, int, error) {
		return PacketPong{}, 0, nil
	})
	cdc.RegisterConcreteUnmarshaller(PacketMsgName, func(codec *amino.Codec, data []byte) (interface{}, int, error) {
		var msg PacketMsg
		err := msg.UnmarshalFromAmino(codec, data)
		if err != nil {
			return nil, 0, err
		} else {
			return &msg, len(data), nil
		}
	})

	cdc.EnableBufferMarshaler(PacketMsg{})
}

func (PacketPing) AssertIsPacket() {}
func (PacketPong) AssertIsPacket() {}
func (PacketMsg) AssertIsPacket()  {}

type PacketPing struct {
}

func (p *PacketPing) UnmarshalFromAmino(_ *amino.Codec, _ []byte) error {
	return nil
}

func (PacketPing) MarshalToAmino(_ *amino.Codec) ([]byte, error) {
	return []byte{}, nil
}

type PacketPong struct {
}

func (PacketPong) MarshalToAmino(_ *amino.Codec) ([]byte, error) {
	return []byte{}, nil
}

func (p *PacketPong) UnmarshalFromAmino(_ *amino.Codec, _ []byte) error {
	return nil
}

type PacketMsg struct {
	ChannelID byte
	EOF       byte // 1 means message ends here.
	Bytes     []byte
}

func (mp PacketMsg) String() string {
	return fmt.Sprintf("PacketMsg{%X:%X T:%X}", mp.ChannelID, mp.Bytes, mp.EOF)
}

func (mp *PacketMsg) Reset() {
	mp.ChannelID = 0
	mp.EOF = 0
	mp.Bytes = mp.Bytes[:0]
}

func (mp PacketMsg) AminoSize(_ *amino.Codec) int {
	size := 0
	if mp.ChannelID != 0 {
		if mp.ChannelID <= 0b0111_1111 {
			size += 2
		} else {
			size += 3
		}
	}

	if mp.EOF != 0 {
		if mp.EOF <= 0b0111_1111 {
			size += 2
		} else {
			size += 3
		}
	}

	if len(mp.Bytes) != 0 {
		size += 1 + amino.ByteSliceSize(mp.Bytes)
	}

	return size
}

func (mp PacketMsg) MarshalToAmino(cdc *amino.Codec) ([]byte, error) {
	var buf bytes.Buffer
	buf.Grow(mp.AminoSize(cdc))
	err := mp.MarshalAminoTo(cdc, &buf)
	if err != nil {
		return nil, err
	}
	return buf.Bytes(), nil
}

func (mp PacketMsg) MarshalAminoTo(_ *amino.Codec, buf *bytes.Buffer) error {
	var err error
	// field 1
	if mp.ChannelID != 0 {
		const pbKey = 1 << 3
		buf.WriteByte(pbKey)
		if mp.ChannelID <= 0b0111_1111 {
			buf.WriteByte(mp.ChannelID)
		} else {
			buf.WriteByte(0b1000_0000 | (mp.ChannelID & 0x7F))
			buf.WriteByte(mp.ChannelID >> 7)
		}
	}

	// field 2
	if mp.EOF != 0 {
		const pbKey = 2 << 3
		buf.WriteByte(pbKey)
		if mp.EOF <= 0b0111_1111 {
			buf.WriteByte(mp.EOF)
		} else {
			buf.WriteByte(0b1000_0000 | (mp.EOF & 0x7F))
			buf.WriteByte(mp.EOF >> 7)
		}
	}

	// field 3
	if len(mp.Bytes) != 0 {
		const pbKey = 3<<3 | 2
		buf.WriteByte(pbKey)
		err = amino.EncodeUvarintToBuffer(buf, uint64(len(mp.Bytes)))
		if err != nil {
			return err
		}
		buf.Write(mp.Bytes)
	}

	return nil
}

func (mp *PacketMsg) UnmarshalFromAmino(_ *amino.Codec, data []byte) error {
	var dataLen uint64 = 0
	var subData []byte

	for {
		data = data[dataLen:]

		if len(data) == 0 {
			break
		}

		pos, aminoType, err := amino.ParseProtoPosAndTypeMustOneByte(data[0])
		if err != nil {
			return err
		}
		data = data[1:]

		if aminoType == amino.Typ3_ByteLength {
			var n int
			dataLen, n, err = amino.DecodeUvarint(data)
			if err != nil {
				return err
			}

			data = data[n:]
			if len(data) < int(dataLen) {
				return errors.New("invalid data len")
			}
			subData = data[:dataLen]
		}

		switch pos {
		case 1:
			vari, n, err := amino.DecodeUvarint(data)
			if err != nil {
				return err
			}
			mp.ChannelID = byte(vari)
			dataLen = uint64(n)
		case 2:
			vari, n, err := amino.DecodeUvarint(data)
			if err != nil {
				return err
			}
			mp.EOF = byte(vari)
			dataLen = uint64(n)
		case 3:
			if cap(mp.Bytes) >= len(subData) {
				mp.Bytes = mp.Bytes[:len(subData)]
			} else {
				mp.Bytes = make([]byte, len(subData))
			}
			copy(mp.Bytes, subData)
		}
	}
	return nil
}

var (
	PacketPingTypePrefix = []byte{0x15, 0xC3, 0xD2, 0x89}
	PacketPongTypePrefix = []byte{0x8A, 0x79, 0x7F, 0xE2}
	PacketMsgTypePrefix  = []byte{0xB0, 0x5B, 0x4F, 0x2C}

	packetPing Packet = PacketPing{}
	packetPong Packet = PacketPong{}

	packetBzPool = &sync.Pool{
		New: func() interface{} {
			return new(bytes.Buffer)
		},
	}
	packetLengthPrefixBzPool = &sync.Pool{
		New: func() interface{} {
			return &[binary.MaxVarintLen64]byte{}
		},
	}
)

func unmarshalPacketFromAminoReader(r io.Reader, maxSize int64, targetMsg *PacketMsg) (packet Packet, n int64, err error) {
	if maxSize < 0 {
		panic("maxSize cannot be negative.")
	}

	// Read byte-length prefix.
	var l int64
	var buf = packetLengthPrefixBzPool.Get().(*[binary.MaxVarintLen64]byte)
	defer packetLengthPrefixBzPool.Put(buf)
	for i := 0; i < len(buf); i++ {
		_, err = r.Read(buf[i : i+1])
		if err != nil {
			return
		}
		n += 1
		if buf[i]&0x80 == 0 {
			break
		}
		if n >= maxSize {
			err = fmt.Errorf("Read overflow, maxSize is %v but uvarint(length-prefix) is itself greater than maxSize.", maxSize)
		}
	}
	u64, _ := binary.Uvarint(buf[:])
	if err != nil {
		return
	}
	if maxSize > 0 {
		if uint64(maxSize) < u64 {
			err = fmt.Errorf("Read overflow, maxSize is %v but this amino binary object is %v bytes.", maxSize, u64)
			return
		}
		if (maxSize - n) < int64(u64) {
			err = fmt.Errorf("Read overflow, maxSize is %v but this length-prefixed amino binary object is %v+%v bytes.", maxSize, n, u64)
			return
		}
	}
	l = int64(u64)
	if l < 0 {
		err = fmt.Errorf("Read overflow, this implementation can't read this because, why would anyone have this much data? Hello from 2018.")
	}

	bbuf := packetBzPool.Get().(*bytes.Buffer)
	defer packetBzPool.Put(bbuf)
	bbuf.Grow(int(l))
	var bz = bbuf.Bytes()
	if int64(cap(bz)) >= l {
		bz = bz[:l]
	} else {
		bz = make([]byte, l)
	}

	// Read that many bytes.
	_, err = io.ReadFull(r, bz)
	if err != nil {
		return
	}
	n += l

	if bytes.Equal(PacketPingTypePrefix, bz) {
		packet = packetPing
		return
	} else if bytes.Equal(PacketPongTypePrefix, bz) {
		packet = packetPong
		return
	} else if bytes.Equal(PacketMsgTypePrefix, bz[0:4]) {
		var msg *PacketMsg
		if targetMsg != nil {
			msg = targetMsg
			msg.Reset()
		} else {
			msg = &PacketMsg{}
		}
		err = msg.UnmarshalFromAmino(cdc, bz[4:])
		if err == nil {
			packet = msg
			return
		}
	}
	var packet4amino Packet
	err = cdc.UnmarshalBinaryBare(bz, &packet4amino)
	packet = packet4amino
	return
}