func (c *MConnection) sendMsgPacket() bool {
	var leastRatio float32 = math.MaxFloat32
	var leastChannel *Channel
	for _, channel := range c.channels {
		if !channel.isSendPending() {
			continue
		}
		ratio := float32(channel.recentlySent) / float32(channel.priority)
		if ratio < leastRatio {
			leastRatio = ratio
			leastChannel = channel
		}
	}
	if leastChannel == nil {
		return true
	}
	n, err := leastChannel.writeMsgPacketTo(c.bufWriter)
	if err != nil {
		c.logger.Warn("Failed to write msgPacket", zap.Error(err))
		c.stopForError(err)
		return true
	}
	c.sendMonitor.Update(int(n))
	c.flushTimer.Set()
	return false
}
Returns true if messages from channels were exhausted.
package p2p
import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"runtime/debug"
	"strings"
	"sync/atomic"
	"time"
	"github.com/spf13/viper"
	"go.uber.org/zap"
	. "github.com/annchain/annchain/module/lib/go-common"
	flow "github.com/annchain/annchain/module/lib/go-flowrate/flowrate"
	"github.com/annchain/annchain/module/xlib"
)
const (
	numBatchMsgPackets = 10
	minReadBufferSize  = 1024
	minWriteBufferSize = 65536
	idleTimeoutMinutes = 5
	updateStatsSeconds = 2
	pingTimeoutSeconds = 40
	flushThrottleMS    = 100
	defaultSendQueueCapacity   = 1
	defaultRecvBufferCapacity  = 4096
	defaultRecvMessageCapacity = 22020096  
	defaultSendTimeoutSeconds  = 10
)
type receiveCbFunc func(chID byte, msgBytes []byte)
type errorCbFunc func(interface{})
type MConnection struct {
	BaseService
	conn        net.Conn
	bufReader   *bufio.Reader
	bufWriter   *bufio.Writer
	sendMonitor *flow.Monitor
	recvMonitor *flow.Monitor
	sendRate    int64
	recvRate    int64
	send        chan struct{}
	pong        chan struct{}
	channels    []*Channel
	channelsIdx map[byte]*Channel
	onReceive   receiveCbFunc
	onError     errorCbFunc
	errored     uint32
	quit         chan struct{}
	flushTimer   *ThrottleTimer  
	pingTimer    *RepeatTimer    
	chStatsTimer *RepeatTimer    
	LocalAddress  *NetAddress
	RemoteAddress *NetAddress
	connectResetWait time.Duration
	logger  *zap.Logger
	slogger *zap.SugaredLogger
}
func NewMConnection(logger *zap.Logger, config *viper.Viper, conn net.Conn, chDescs []*ChannelDescriptor, onReceive receiveCbFunc, onError errorCbFunc) *MConnection {
	mconn := &MConnection{
		conn:        conn,
		bufReader:   bufio.NewReaderSize(conn, minReadBufferSize),
		bufWriter:   bufio.NewWriterSize(conn, minWriteBufferSize),
		sendMonitor: flow.New(0, 0),
		recvMonitor: flow.New(0, 0),
		sendRate:    config.GetInt64(configKeySendRate),
		recvRate:    config.GetInt64(configKeyRecvRate),
		send:        make(chan struct{}, 1),
		pong:        make(chan struct{}),
		onReceive:   onReceive,
		onError:     onError,
		quit:         nil,
		flushTimer:   nil,
		pingTimer:    nil,
		chStatsTimer: nil,
		LocalAddress:  NewNetAddress(logger, conn.LocalAddr()),
		RemoteAddress: NewNetAddress(logger, conn.RemoteAddr()),
		connectResetWait: time.Duration(config.GetInt("connection_reset_wait")) * time.Millisecond,
		logger:  logger,
		slogger: logger.Sugar(),
	}
	var channelsIdx = map[byte]*Channel{}
	var channels = []*Channel{}
	for _, desc := range chDescs {
		descCopy := *desc  
		channel := newChannel(mconn, &descCopy)
		channelsIdx[channel.id] = channel
		channels = append(channels, channel)
	}
	mconn.channels = channels
	mconn.channelsIdx = channelsIdx
	mconn.BaseService = *NewBaseService(logger, "MConnection", mconn)
	return mconn
}
func (c *MConnection) OnStart() error {
	c.BaseService.OnStart()
	c.quit = make(chan struct{})
	c.flushTimer = NewThrottleTimer("flush", flushThrottleMS*time.Millisecond)
	c.pingTimer = NewRepeatTimer("ping", pingTimeoutSeconds*time.Second)
	c.chStatsTimer = NewRepeatTimer("chStats", updateStatsSeconds*time.Second)
	go c.sendRoutine()
	go c.recvRoutine()
	return nil
}
func (c *MConnection) OnStop() {
	c.BaseService.OnStop()
	c.flushTimer.Stop()
	c.pingTimer.Stop()
	c.chStatsTimer.Stop()
	if c.quit != nil {
		close(c.quit)
	}
	c.conn.Close()
}
func (c *MConnection) String() string {
	return fmt.Sprintf("MConn{%v}", c.conn.RemoteAddr())
}
func (c *MConnection) flush() {
	err := c.bufWriter.Flush()
	if err != nil {
		c.logger.Warn("MConnection flush failed", zap.Error(err))
	}
}
func (c *MConnection) _recover() {
	if r := recover(); r != nil {
		stack := debug.Stack()
		c.stopForError(StackError{r, stack})
	}
}
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) Send(chID byte, msg []byte) bool {
	if !c.IsRunning() {
		return false
	}
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.logger.Error("Cannot send bytes, unknown channel", zap.ByteString("chID", []byte{chID}))
		return false
	}
	success := channel.sendBytes(msg)
	if success {
		select {
		case c.send <- struct{}{}:
		default:
		}
	} else {
		c.slogger.Warnw("Send failed", "channel", chID, "conn", c, "msg", msg)
	}
	return success
}
func (c *MConnection) TrySend(chID byte, msg []byte) bool {
	if !c.IsRunning() {
		return false
	}
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.logger.Error("Cannot send bytes, unknown channels", zap.ByteString("chID", []byte{chID}))
		return false
	}
	ok = channel.trySendBytes(msg)
	if ok {
		select {
		case c.send <- struct{}{}:
		default:
		}
	}
	return ok
}
func (c *MConnection) CanSend(chID byte) bool {
	if !c.IsRunning() {
		return false
	}
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.slogger.Errorf("Unknown channel %X", chID)
		return false
	}
	return channel.canSend()
}
func (c *MConnection) sendRoutine() {
	defer c._recover()
FOR_LOOP:
	for {
		var err error
		select {
		case <-c.flushTimer.Ch:
			c.flush()
		case <-c.chStatsTimer.Ch:
			for _, channel := range c.channels {
				channel.updateStats()
			}
		case <-c.pingTimer.Ch:
			xlib.BinWrite(c.bufWriter, packetTypePing)
			c.sendMonitor.Update(1)
			c.flush()
		case <-c.pong:
			xlib.BinWrite(c.bufWriter, packetTypePong)
			c.sendMonitor.Update(1)
			c.flush()
		case <-c.quit:
			break FOR_LOOP
		case <-c.send:
			eof := c.sendSomeMsgPackets()
			if !eof {
				select {
				case c.send <- struct{}{}:
				default:
				}
			}
		}
		if !c.IsRunning() {
			break FOR_LOOP
		}
		if err != nil {
			if strings.Contains(err.Error(), "connection reset by peer") {
				time.Sleep(c.connectResetWait)
			} else {
				c.slogger.Warnw("Connection failed @ sendRoutine", "conn", c, "error", err)
				c.stopForError(err)
				break FOR_LOOP
			}
		}
	}
}
func (c *MConnection) sendSomeMsgPackets() bool {
	c.sendMonitor.Limit(maxMsgPacketTotalSize, atomic.LoadInt64(&c.sendRate), true)
	for i := 0; i < numBatchMsgPackets; i++ {
		if c.sendMsgPacket() {
			return true
		}
	}
	return false
}