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.

732 lines
20 KiB

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