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.

786 lines
21 KiB

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