You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

794 lines
21 KiB

  1. package conn
  2. import (
  3. "bufio"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "math"
  8. "net"
  9. "reflect"
  10. "sync/atomic"
  11. "time"
  12. amino "github.com/tendermint/go-amino"
  13. cmn "github.com/tendermint/tendermint/libs/common"
  14. flow "github.com/tendermint/tendermint/libs/flowrate"
  15. "github.com/tendermint/tendermint/libs/log"
  16. )
  17. const (
  18. defaultMaxPacketMsgPayloadSize = 1024
  19. numBatchPacketMsgs = 10
  20. minReadBufferSize = 1024
  21. minWriteBufferSize = 65536
  22. updateStats = 2 * time.Second
  23. // some of these defaults are written in the user config
  24. // flushThrottle, sendRate, recvRate
  25. // TODO: remove values present in config
  26. defaultFlushThrottle = 100 * time.Millisecond
  27. defaultSendQueueCapacity = 1
  28. defaultRecvBufferCapacity = 4096
  29. defaultRecvMessageCapacity = 22020096 // 21MB
  30. defaultSendRate = int64(512000) // 500KB/s
  31. defaultRecvRate = int64(512000) // 500KB/s
  32. defaultSendTimeout = 10 * time.Second
  33. defaultPingInterval = 60 * time.Second
  34. defaultPongTimeout = 45 * time.Second
  35. )
  36. type receiveCbFunc func(chID byte, msgBytes []byte)
  37. type errorCbFunc func(interface{})
  38. /*
  39. Each peer has one `MConnection` (multiplex connection) instance.
  40. __multiplex__ *noun* a system or signal involving simultaneous transmission of
  41. several messages along a single channel of communication.
  42. Each `MConnection` handles message transmission on multiple abstract communication
  43. `Channel`s. Each channel has a globally unique byte id.
  44. The byte id and the relative priorities of each `Channel` are configured upon
  45. initialization of the connection.
  46. There are two methods for sending messages:
  47. func (m MConnection) Send(chID byte, msgBytes []byte) bool {}
  48. func (m MConnection) TrySend(chID byte, msgBytes []byte}) bool {}
  49. `Send(chID, msgBytes)` is a blocking call that waits until `msg` is
  50. successfully queued for the channel with the given id byte `chID`, or until the
  51. request times out. The message `msg` is serialized using Go-Amino.
  52. `TrySend(chID, msgBytes)` is a nonblocking call that returns false if the
  53. channel's queue is full.
  54. Inbound message bytes are handled with an onReceive callback function.
  55. */
  56. type MConnection struct {
  57. cmn.BaseService
  58. conn net.Conn
  59. bufConnReader *bufio.Reader
  60. bufConnWriter *bufio.Writer
  61. sendMonitor *flow.Monitor
  62. recvMonitor *flow.Monitor
  63. send chan struct{}
  64. pong chan struct{}
  65. channels []*Channel
  66. channelsIdx map[byte]*Channel
  67. onReceive receiveCbFunc
  68. onError errorCbFunc
  69. errored uint32
  70. config MConnConfig
  71. quit chan struct{}
  72. flushTimer *cmn.ThrottleTimer // flush writes as necessary but throttled.
  73. pingTimer *cmn.RepeatTimer // send pings periodically
  74. // close conn if pong is not received in pongTimeout
  75. pongTimer *time.Timer
  76. pongTimeoutCh chan bool // true - timeout, false - peer sent pong
  77. chStatsTimer *cmn.RepeatTimer // update channel stats periodically
  78. created time.Time // time of creation
  79. _maxPacketMsgSize int
  80. }
  81. // MConnConfig is a MConnection configuration.
  82. type MConnConfig struct {
  83. SendRate int64 `mapstructure:"send_rate"`
  84. RecvRate int64 `mapstructure:"recv_rate"`
  85. // Maximum payload size
  86. MaxPacketMsgPayloadSize int `mapstructure:"max_packet_msg_payload_size"`
  87. // Interval to flush writes (throttled)
  88. FlushThrottle time.Duration `mapstructure:"flush_throttle"`
  89. // Interval to send pings
  90. PingInterval time.Duration `mapstructure:"ping_interval"`
  91. // Maximum wait time for pongs
  92. PongTimeout time.Duration `mapstructure:"pong_timeout"`
  93. }
  94. // DefaultMConnConfig returns the default config.
  95. func DefaultMConnConfig() MConnConfig {
  96. return MConnConfig{
  97. SendRate: defaultSendRate,
  98. RecvRate: defaultRecvRate,
  99. MaxPacketMsgPayloadSize: defaultMaxPacketMsgPayloadSize,
  100. FlushThrottle: defaultFlushThrottle,
  101. PingInterval: defaultPingInterval,
  102. PongTimeout: defaultPongTimeout,
  103. }
  104. }
  105. // NewMConnection wraps net.Conn and creates multiplex connection
  106. func NewMConnection(conn net.Conn, chDescs []*ChannelDescriptor, onReceive receiveCbFunc, onError errorCbFunc) *MConnection {
  107. return NewMConnectionWithConfig(
  108. conn,
  109. chDescs,
  110. onReceive,
  111. onError,
  112. DefaultMConnConfig())
  113. }
  114. // NewMConnectionWithConfig wraps net.Conn and creates multiplex connection with a config
  115. func NewMConnectionWithConfig(conn net.Conn, chDescs []*ChannelDescriptor, onReceive receiveCbFunc, onError errorCbFunc, config MConnConfig) *MConnection {
  116. if config.PongTimeout >= config.PingInterval {
  117. panic("pongTimeout must be less than pingInterval (otherwise, next ping will reset pong timer)")
  118. }
  119. mconn := &MConnection{
  120. conn: conn,
  121. bufConnReader: bufio.NewReaderSize(conn, minReadBufferSize),
  122. bufConnWriter: bufio.NewWriterSize(conn, minWriteBufferSize),
  123. sendMonitor: flow.New(0, 0),
  124. recvMonitor: flow.New(0, 0),
  125. send: make(chan struct{}, 1),
  126. pong: make(chan struct{}, 1),
  127. onReceive: onReceive,
  128. onError: onError,
  129. config: config,
  130. }
  131. // Create channels
  132. var channelsIdx = map[byte]*Channel{}
  133. var channels = []*Channel{}
  134. for _, desc := range chDescs {
  135. channel := newChannel(mconn, *desc)
  136. channelsIdx[channel.desc.ID] = channel
  137. channels = append(channels, channel)
  138. }
  139. mconn.channels = channels
  140. mconn.channelsIdx = channelsIdx
  141. mconn.BaseService = *cmn.NewBaseService(nil, "MConnection", mconn)
  142. // maxPacketMsgSize() is a bit heavy, so call just once
  143. mconn._maxPacketMsgSize = mconn.maxPacketMsgSize()
  144. return mconn
  145. }
  146. func (c *MConnection) SetLogger(l log.Logger) {
  147. c.BaseService.SetLogger(l)
  148. for _, ch := range c.channels {
  149. ch.SetLogger(l)
  150. }
  151. }
  152. // OnStart implements BaseService
  153. func (c *MConnection) OnStart() error {
  154. if err := c.BaseService.OnStart(); err != nil {
  155. return err
  156. }
  157. c.quit = make(chan struct{})
  158. c.flushTimer = cmn.NewThrottleTimer("flush", c.config.FlushThrottle)
  159. c.pingTimer = cmn.NewRepeatTimer("ping", c.config.PingInterval)
  160. c.pongTimeoutCh = make(chan bool, 1)
  161. c.chStatsTimer = cmn.NewRepeatTimer("chStats", updateStats)
  162. go c.sendRoutine()
  163. go c.recvRoutine()
  164. return nil
  165. }
  166. // OnStop implements BaseService
  167. func (c *MConnection) OnStop() {
  168. c.BaseService.OnStop()
  169. c.flushTimer.Stop()
  170. c.pingTimer.Stop()
  171. c.chStatsTimer.Stop()
  172. if c.quit != nil {
  173. close(c.quit)
  174. }
  175. c.conn.Close() // nolint: errcheck
  176. // We can't close pong safely here because
  177. // recvRoutine may write to it after we've stopped.
  178. // Though it doesn't need to get closed at all,
  179. // we close it @ recvRoutine.
  180. }
  181. func (c *MConnection) String() string {
  182. return fmt.Sprintf("MConn{%v}", c.conn.RemoteAddr())
  183. }
  184. func (c *MConnection) flush() {
  185. c.Logger.Debug("Flush", "conn", c)
  186. err := c.bufConnWriter.Flush()
  187. if err != nil {
  188. c.Logger.Error("MConnection flush failed", "err", err)
  189. }
  190. }
  191. // Catch panics, usually caused by remote disconnects.
  192. func (c *MConnection) _recover() {
  193. if r := recover(); r != nil {
  194. err := cmn.ErrorWrap(r, "recovered panic in MConnection")
  195. c.stopForError(err)
  196. }
  197. }
  198. func (c *MConnection) stopForError(r interface{}) {
  199. c.Stop()
  200. if atomic.CompareAndSwapUint32(&c.errored, 0, 1) {
  201. if c.onError != nil {
  202. c.onError(r)
  203. }
  204. }
  205. }
  206. // Queues a message to be sent to channel.
  207. func (c *MConnection) Send(chID byte, msgBytes []byte) bool {
  208. if !c.IsRunning() {
  209. return false
  210. }
  211. c.Logger.Debug("Send", "channel", chID, "conn", c, "msgBytes", fmt.Sprintf("%X", msgBytes))
  212. // Send message to channel.
  213. channel, ok := c.channelsIdx[chID]
  214. if !ok {
  215. c.Logger.Error(fmt.Sprintf("Cannot send bytes, unknown channel %X", chID))
  216. return false
  217. }
  218. success := channel.sendBytes(msgBytes)
  219. if success {
  220. // Wake up sendRoutine if necessary
  221. select {
  222. case c.send <- struct{}{}:
  223. default:
  224. }
  225. } else {
  226. c.Logger.Error("Send failed", "channel", chID, "conn", c, "msgBytes", fmt.Sprintf("%X", msgBytes))
  227. }
  228. return success
  229. }
  230. // Queues a message to be sent to channel.
  231. // Nonblocking, returns true if successful.
  232. func (c *MConnection) TrySend(chID byte, msgBytes []byte) bool {
  233. if !c.IsRunning() {
  234. return false
  235. }
  236. c.Logger.Debug("TrySend", "channel", chID, "conn", c, "msgBytes", fmt.Sprintf("%X", msgBytes))
  237. // Send message to channel.
  238. channel, ok := c.channelsIdx[chID]
  239. if !ok {
  240. c.Logger.Error(fmt.Sprintf("Cannot send bytes, unknown channel %X", chID))
  241. return false
  242. }
  243. ok = channel.trySendBytes(msgBytes)
  244. if ok {
  245. // Wake up sendRoutine if necessary
  246. select {
  247. case c.send <- struct{}{}:
  248. default:
  249. }
  250. }
  251. return ok
  252. }
  253. // CanSend returns true if you can send more data onto the chID, false
  254. // otherwise. Use only as a heuristic.
  255. func (c *MConnection) CanSend(chID byte) bool {
  256. if !c.IsRunning() {
  257. return false
  258. }
  259. channel, ok := c.channelsIdx[chID]
  260. if !ok {
  261. c.Logger.Error(fmt.Sprintf("Unknown channel %X", chID))
  262. return false
  263. }
  264. return channel.canSend()
  265. }
  266. // sendRoutine polls for packets to send from channels.
  267. func (c *MConnection) sendRoutine() {
  268. defer c._recover()
  269. FOR_LOOP:
  270. for {
  271. var _n int64
  272. var err error
  273. SELECTION:
  274. select {
  275. case <-c.flushTimer.Ch:
  276. // NOTE: flushTimer.Set() must be called every time
  277. // something is written to .bufConnWriter.
  278. c.flush()
  279. case <-c.chStatsTimer.Chan():
  280. for _, channel := range c.channels {
  281. channel.updateStats()
  282. }
  283. case <-c.pingTimer.Chan():
  284. c.Logger.Debug("Send Ping")
  285. _n, err = cdc.MarshalBinaryWriter(c.bufConnWriter, PacketPing{})
  286. if err != nil {
  287. break SELECTION
  288. }
  289. c.sendMonitor.Update(int(_n))
  290. c.Logger.Debug("Starting pong timer", "dur", c.config.PongTimeout)
  291. c.pongTimer = time.AfterFunc(c.config.PongTimeout, func() {
  292. select {
  293. case c.pongTimeoutCh <- true:
  294. default:
  295. }
  296. })
  297. c.flush()
  298. case timeout := <-c.pongTimeoutCh:
  299. if timeout {
  300. c.Logger.Debug("Pong timeout")
  301. err = errors.New("pong timeout")
  302. } else {
  303. c.stopPongTimer()
  304. }
  305. case <-c.pong:
  306. c.Logger.Debug("Send Pong")
  307. _n, err = cdc.MarshalBinaryWriter(c.bufConnWriter, PacketPong{})
  308. if err != nil {
  309. break SELECTION
  310. }
  311. c.sendMonitor.Update(int(_n))
  312. c.flush()
  313. case <-c.quit:
  314. break FOR_LOOP
  315. case <-c.send:
  316. // Send some PacketMsgs
  317. eof := c.sendSomePacketMsgs()
  318. if !eof {
  319. // Keep sendRoutine awake.
  320. select {
  321. case c.send <- struct{}{}:
  322. default:
  323. }
  324. }
  325. }
  326. if !c.IsRunning() {
  327. break FOR_LOOP
  328. }
  329. if err != nil {
  330. c.Logger.Error("Connection failed @ sendRoutine", "conn", c, "err", err)
  331. c.stopForError(err)
  332. break FOR_LOOP
  333. }
  334. }
  335. // Cleanup
  336. c.stopPongTimer()
  337. }
  338. // Returns true if messages from channels were exhausted.
  339. // Blocks in accordance to .sendMonitor throttling.
  340. func (c *MConnection) sendSomePacketMsgs() bool {
  341. // Block until .sendMonitor says we can write.
  342. // Once we're ready we send more than we asked for,
  343. // but amortized it should even out.
  344. c.sendMonitor.Limit(c._maxPacketMsgSize, atomic.LoadInt64(&c.config.SendRate), true)
  345. // Now send some PacketMsgs.
  346. for i := 0; i < numBatchPacketMsgs; i++ {
  347. if c.sendPacketMsg() {
  348. return true
  349. }
  350. }
  351. return false
  352. }
  353. // Returns true if messages from channels were exhausted.
  354. func (c *MConnection) sendPacketMsg() bool {
  355. // Choose a channel to create a PacketMsg from.
  356. // The chosen channel will be the one whose recentlySent/priority is the least.
  357. var leastRatio float32 = math.MaxFloat32
  358. var leastChannel *Channel
  359. for _, channel := range c.channels {
  360. // If nothing to send, skip this channel
  361. if !channel.isSendPending() {
  362. continue
  363. }
  364. // Get ratio, and keep track of lowest ratio.
  365. ratio := float32(channel.recentlySent) / float32(channel.desc.Priority)
  366. if ratio < leastRatio {
  367. leastRatio = ratio
  368. leastChannel = channel
  369. }
  370. }
  371. // Nothing to send?
  372. if leastChannel == nil {
  373. return true
  374. }
  375. // c.Logger.Info("Found a msgPacket to send")
  376. // Make & send a PacketMsg from this channel
  377. _n, err := leastChannel.writePacketMsgTo(c.bufConnWriter)
  378. if err != nil {
  379. c.Logger.Error("Failed to write PacketMsg", "err", err)
  380. c.stopForError(err)
  381. return true
  382. }
  383. c.sendMonitor.Update(int(_n))
  384. c.flushTimer.Set()
  385. return false
  386. }
  387. // recvRoutine reads PacketMsgs and reconstructs the message using the channels' "recving" buffer.
  388. // After a whole message has been assembled, it's pushed to onReceive().
  389. // Blocks depending on how the connection is throttled.
  390. // Otherwise, it never blocks.
  391. func (c *MConnection) recvRoutine() {
  392. defer c._recover()
  393. FOR_LOOP:
  394. for {
  395. // Block until .recvMonitor says we can read.
  396. c.recvMonitor.Limit(c._maxPacketMsgSize, atomic.LoadInt64(&c.config.RecvRate), true)
  397. // Peek into bufConnReader for debugging
  398. /*
  399. if numBytes := c.bufConnReader.Buffered(); numBytes > 0 {
  400. bz, err := c.bufConnReader.Peek(cmn.MinInt(numBytes, 100))
  401. if err == nil {
  402. // return
  403. } else {
  404. c.Logger.Debug("Error peeking connection buffer", "err", err)
  405. // return nil
  406. }
  407. c.Logger.Info("Peek connection buffer", "numBytes", numBytes, "bz", bz)
  408. }
  409. */
  410. // Read packet type
  411. var packet Packet
  412. var _n int64
  413. var err error
  414. _n, err = cdc.UnmarshalBinaryReader(c.bufConnReader, &packet, int64(c._maxPacketMsgSize))
  415. c.recvMonitor.Update(int(_n))
  416. if err != nil {
  417. if c.IsRunning() {
  418. c.Logger.Error("Connection failed @ recvRoutine (reading byte)", "conn", c, "err", err)
  419. c.stopForError(err)
  420. }
  421. break FOR_LOOP
  422. }
  423. // Read more depending on packet type.
  424. switch pkt := packet.(type) {
  425. case PacketPing:
  426. // TODO: prevent abuse, as they cause flush()'s.
  427. // https://github.com/tendermint/tendermint/issues/1190
  428. c.Logger.Debug("Receive Ping")
  429. select {
  430. case c.pong <- struct{}{}:
  431. default:
  432. // never block
  433. }
  434. case PacketPong:
  435. c.Logger.Debug("Receive Pong")
  436. select {
  437. case c.pongTimeoutCh <- false:
  438. default:
  439. // never block
  440. }
  441. case PacketMsg:
  442. channel, ok := c.channelsIdx[pkt.ChannelID]
  443. if !ok || channel == nil {
  444. err := fmt.Errorf("Unknown channel %X", pkt.ChannelID)
  445. c.Logger.Error("Connection failed @ recvRoutine", "conn", c, "err", err)
  446. c.stopForError(err)
  447. break FOR_LOOP
  448. }
  449. msgBytes, err := channel.recvPacketMsg(pkt)
  450. if err != nil {
  451. if c.IsRunning() {
  452. c.Logger.Error("Connection failed @ recvRoutine", "conn", c, "err", err)
  453. c.stopForError(err)
  454. }
  455. break FOR_LOOP
  456. }
  457. if msgBytes != nil {
  458. c.Logger.Debug("Received bytes", "chID", pkt.ChannelID, "msgBytes", fmt.Sprintf("%X", msgBytes))
  459. // NOTE: This means the reactor.Receive runs in the same thread as the p2p recv routine
  460. c.onReceive(pkt.ChannelID, msgBytes)
  461. }
  462. default:
  463. err := fmt.Errorf("Unknown message type %v", reflect.TypeOf(packet))
  464. c.Logger.Error("Connection failed @ recvRoutine", "conn", c, "err", err)
  465. c.stopForError(err)
  466. break FOR_LOOP
  467. }
  468. }
  469. // Cleanup
  470. close(c.pong)
  471. for range c.pong {
  472. // Drain
  473. }
  474. }
  475. // not goroutine-safe
  476. func (c *MConnection) stopPongTimer() {
  477. if c.pongTimer != nil {
  478. _ = c.pongTimer.Stop()
  479. c.pongTimer = nil
  480. }
  481. }
  482. // maxPacketMsgSize returns a maximum size of PacketMsg, including the overhead
  483. // of amino encoding.
  484. func (c *MConnection) maxPacketMsgSize() int {
  485. return len(cdc.MustMarshalBinary(PacketMsg{
  486. ChannelID: 0x01,
  487. EOF: 1,
  488. Bytes: make([]byte, c.config.MaxPacketMsgPayloadSize),
  489. })) + 10 // leave room for changes in amino
  490. }
  491. type ConnectionStatus struct {
  492. Duration time.Duration
  493. SendMonitor flow.Status
  494. RecvMonitor flow.Status
  495. Channels []ChannelStatus
  496. }
  497. type ChannelStatus struct {
  498. ID byte
  499. SendQueueCapacity int
  500. SendQueueSize int
  501. Priority int
  502. RecentlySent int64
  503. }
  504. func (c *MConnection) Status() ConnectionStatus {
  505. var status ConnectionStatus
  506. status.Duration = time.Since(c.created)
  507. status.SendMonitor = c.sendMonitor.Status()
  508. status.RecvMonitor = c.recvMonitor.Status()
  509. status.Channels = make([]ChannelStatus, len(c.channels))
  510. for i, channel := range c.channels {
  511. status.Channels[i] = ChannelStatus{
  512. ID: channel.desc.ID,
  513. SendQueueCapacity: cap(channel.sendQueue),
  514. SendQueueSize: int(channel.sendQueueSize), // TODO use atomic
  515. Priority: channel.desc.Priority,
  516. RecentlySent: channel.recentlySent,
  517. }
  518. }
  519. return status
  520. }
  521. //-----------------------------------------------------------------------------
  522. type ChannelDescriptor struct {
  523. ID byte
  524. Priority int
  525. SendQueueCapacity int
  526. RecvBufferCapacity int
  527. RecvMessageCapacity int
  528. }
  529. func (chDesc ChannelDescriptor) FillDefaults() (filled ChannelDescriptor) {
  530. if chDesc.SendQueueCapacity == 0 {
  531. chDesc.SendQueueCapacity = defaultSendQueueCapacity
  532. }
  533. if chDesc.RecvBufferCapacity == 0 {
  534. chDesc.RecvBufferCapacity = defaultRecvBufferCapacity
  535. }
  536. if chDesc.RecvMessageCapacity == 0 {
  537. chDesc.RecvMessageCapacity = defaultRecvMessageCapacity
  538. }
  539. filled = chDesc
  540. return
  541. }
  542. // TODO: lowercase.
  543. // NOTE: not goroutine-safe.
  544. type Channel struct {
  545. conn *MConnection
  546. desc ChannelDescriptor
  547. sendQueue chan []byte
  548. sendQueueSize int32 // atomic.
  549. recving []byte
  550. sending []byte
  551. recentlySent int64 // exponential moving average
  552. maxPacketMsgPayloadSize int
  553. Logger log.Logger
  554. }
  555. func newChannel(conn *MConnection, desc ChannelDescriptor) *Channel {
  556. desc = desc.FillDefaults()
  557. if desc.Priority <= 0 {
  558. cmn.PanicSanity("Channel default priority must be a positive integer")
  559. }
  560. return &Channel{
  561. conn: conn,
  562. desc: desc,
  563. sendQueue: make(chan []byte, desc.SendQueueCapacity),
  564. recving: make([]byte, 0, desc.RecvBufferCapacity),
  565. maxPacketMsgPayloadSize: conn.config.MaxPacketMsgPayloadSize,
  566. }
  567. }
  568. func (ch *Channel) SetLogger(l log.Logger) {
  569. ch.Logger = l
  570. }
  571. // Queues message to send to this channel.
  572. // Goroutine-safe
  573. // Times out (and returns false) after defaultSendTimeout
  574. func (ch *Channel) sendBytes(bytes []byte) bool {
  575. select {
  576. case ch.sendQueue <- bytes:
  577. atomic.AddInt32(&ch.sendQueueSize, 1)
  578. return true
  579. case <-time.After(defaultSendTimeout):
  580. return false
  581. }
  582. }
  583. // Queues message to send to this channel.
  584. // Nonblocking, returns true if successful.
  585. // Goroutine-safe
  586. func (ch *Channel) trySendBytes(bytes []byte) bool {
  587. select {
  588. case ch.sendQueue <- bytes:
  589. atomic.AddInt32(&ch.sendQueueSize, 1)
  590. return true
  591. default:
  592. return false
  593. }
  594. }
  595. // Goroutine-safe
  596. func (ch *Channel) loadSendQueueSize() (size int) {
  597. return int(atomic.LoadInt32(&ch.sendQueueSize))
  598. }
  599. // Goroutine-safe
  600. // Use only as a heuristic.
  601. func (ch *Channel) canSend() bool {
  602. return ch.loadSendQueueSize() < defaultSendQueueCapacity
  603. }
  604. // Returns true if any PacketMsgs are pending to be sent.
  605. // Call before calling nextPacketMsg()
  606. // Goroutine-safe
  607. func (ch *Channel) isSendPending() bool {
  608. if len(ch.sending) == 0 {
  609. if len(ch.sendQueue) == 0 {
  610. return false
  611. }
  612. ch.sending = <-ch.sendQueue
  613. }
  614. return true
  615. }
  616. // Creates a new PacketMsg to send.
  617. // Not goroutine-safe
  618. func (ch *Channel) nextPacketMsg() PacketMsg {
  619. packet := PacketMsg{}
  620. packet.ChannelID = byte(ch.desc.ID)
  621. maxSize := ch.maxPacketMsgPayloadSize
  622. packet.Bytes = ch.sending[:cmn.MinInt(maxSize, len(ch.sending))]
  623. if len(ch.sending) <= maxSize {
  624. packet.EOF = byte(0x01)
  625. ch.sending = nil
  626. atomic.AddInt32(&ch.sendQueueSize, -1) // decrement sendQueueSize
  627. } else {
  628. packet.EOF = byte(0x00)
  629. ch.sending = ch.sending[cmn.MinInt(maxSize, len(ch.sending)):]
  630. }
  631. return packet
  632. }
  633. // Writes next PacketMsg to w and updates c.recentlySent.
  634. // Not goroutine-safe
  635. func (ch *Channel) writePacketMsgTo(w io.Writer) (n int64, err error) {
  636. var packet = ch.nextPacketMsg()
  637. n, err = cdc.MarshalBinaryWriter(w, packet)
  638. ch.recentlySent += n
  639. return
  640. }
  641. // Handles incoming PacketMsgs. It returns a message bytes if message is
  642. // complete. NOTE message bytes may change on next call to recvPacketMsg.
  643. // Not goroutine-safe
  644. func (ch *Channel) recvPacketMsg(packet PacketMsg) ([]byte, error) {
  645. ch.Logger.Debug("Read PacketMsg", "conn", ch.conn, "packet", packet)
  646. var recvCap, recvReceived = ch.desc.RecvMessageCapacity, len(ch.recving) + len(packet.Bytes)
  647. if recvCap < recvReceived {
  648. return nil, fmt.Errorf("Received message exceeds available capacity: %v < %v", recvCap, recvReceived)
  649. }
  650. ch.recving = append(ch.recving, packet.Bytes...)
  651. if packet.EOF == byte(0x01) {
  652. msgBytes := ch.recving
  653. // clear the slice without re-allocating.
  654. // http://stackoverflow.com/questions/16971741/how-do-you-clear-a-slice-in-go
  655. // suggests this could be a memory leak, but we might as well keep the memory for the channel until it closes,
  656. // at which point the recving slice stops being used and should be garbage collected
  657. ch.recving = ch.recving[:0] // make([]byte, 0, ch.desc.RecvBufferCapacity)
  658. return msgBytes, nil
  659. }
  660. return nil, nil
  661. }
  662. // Call this periodically to update stats for throttling purposes.
  663. // Not goroutine-safe
  664. func (ch *Channel) updateStats() {
  665. // Exponential decay of stats.
  666. // TODO: optimize.
  667. ch.recentlySent = int64(float64(ch.recentlySent) * 0.8)
  668. }
  669. //----------------------------------------
  670. // Packet
  671. type Packet interface {
  672. AssertIsPacket()
  673. }
  674. func RegisterPacket(cdc *amino.Codec) {
  675. cdc.RegisterInterface((*Packet)(nil), nil)
  676. cdc.RegisterConcrete(PacketPing{}, "tendermint/p2p/PacketPing", nil)
  677. cdc.RegisterConcrete(PacketPong{}, "tendermint/p2p/PacketPong", nil)
  678. cdc.RegisterConcrete(PacketMsg{}, "tendermint/p2p/PacketMsg", nil)
  679. }
  680. func (_ PacketPing) AssertIsPacket() {}
  681. func (_ PacketPong) AssertIsPacket() {}
  682. func (_ PacketMsg) AssertIsPacket() {}
  683. type PacketPing struct {
  684. }
  685. type PacketPong struct {
  686. }
  687. type PacketMsg struct {
  688. ChannelID byte
  689. EOF byte // 1 means message ends here.
  690. Bytes []byte
  691. }
  692. func (mp PacketMsg) String() string {
  693. return fmt.Sprintf("PacketMsg{%X:%X T:%X}", mp.ChannelID, mp.Bytes, mp.EOF)
  694. }