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.

179 lines
4.9 KiB

10 years ago
10 years ago
8 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
8 years ago
10 years ago
10 years ago
7 years ago
10 years ago
7 years ago
10 years ago
7 years ago
8 years ago
10 years ago
10 years ago
10 years ago
10 years ago
8 years ago
8 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
  1. package mempool
  2. import (
  3. "bytes"
  4. "fmt"
  5. "reflect"
  6. "time"
  7. abci "github.com/tendermint/abci/types"
  8. wire "github.com/tendermint/go-wire"
  9. "github.com/tendermint/tmlibs/clist"
  10. "github.com/tendermint/tmlibs/log"
  11. cfg "github.com/tendermint/tendermint/config"
  12. "github.com/tendermint/tendermint/p2p"
  13. "github.com/tendermint/tendermint/types"
  14. )
  15. const (
  16. MempoolChannel = byte(0x30)
  17. maxMempoolMessageSize = 1048576 // 1MB TODO make it configurable
  18. peerCatchupSleepIntervalMS = 100 // If peer is behind, sleep this amount
  19. )
  20. // MempoolReactor handles mempool tx broadcasting amongst peers.
  21. type MempoolReactor struct {
  22. p2p.BaseReactor
  23. config *cfg.MempoolConfig
  24. Mempool *Mempool
  25. }
  26. // NewMempoolReactor returns a new MempoolReactor with the given config and mempool.
  27. func NewMempoolReactor(config *cfg.MempoolConfig, mempool *Mempool) *MempoolReactor {
  28. memR := &MempoolReactor{
  29. config: config,
  30. Mempool: mempool,
  31. }
  32. memR.BaseReactor = *p2p.NewBaseReactor("MempoolReactor", memR)
  33. return memR
  34. }
  35. // SetLogger sets the Logger on the reactor and the underlying Mempool.
  36. func (memR *MempoolReactor) SetLogger(l log.Logger) {
  37. memR.Logger = l
  38. memR.Mempool.SetLogger(l)
  39. }
  40. // GetChannels implements Reactor.
  41. // It returns the list of channels for this reactor.
  42. func (memR *MempoolReactor) GetChannels() []*p2p.ChannelDescriptor {
  43. return []*p2p.ChannelDescriptor{
  44. {
  45. ID: MempoolChannel,
  46. Priority: 5,
  47. },
  48. }
  49. }
  50. // AddPeer implements Reactor.
  51. // It starts a broadcast routine ensuring all txs are forwarded to the given peer.
  52. func (memR *MempoolReactor) AddPeer(peer p2p.Peer) {
  53. go memR.broadcastTxRoutine(peer)
  54. }
  55. // RemovePeer implements Reactor.
  56. func (memR *MempoolReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
  57. // broadcast routine checks if peer is gone and returns
  58. }
  59. // Receive implements Reactor.
  60. // It adds any received transactions to the mempool.
  61. func (memR *MempoolReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
  62. _, msg, err := DecodeMessage(msgBytes)
  63. if err != nil {
  64. memR.Logger.Error("Error decoding message", "err", err)
  65. return
  66. }
  67. memR.Logger.Debug("Receive", "src", src, "chId", chID, "msg", msg)
  68. switch msg := msg.(type) {
  69. case *TxMessage:
  70. err := memR.Mempool.CheckTx(msg.Tx, nil)
  71. if err != nil {
  72. memR.Logger.Info("Could not check tx", "tx", msg.Tx, "err", err)
  73. }
  74. // broadcasting happens from go routines per peer
  75. default:
  76. memR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
  77. }
  78. }
  79. // BroadcastTx is an alias for Mempool.CheckTx. Broadcasting itself happens in peer routines.
  80. func (memR *MempoolReactor) BroadcastTx(tx types.Tx, cb func(*abci.Response)) error {
  81. return memR.Mempool.CheckTx(tx, cb)
  82. }
  83. // PeerState describes the state of a peer.
  84. type PeerState interface {
  85. GetHeight() int64
  86. }
  87. // Send new mempool txs to peer.
  88. // TODO: Handle mempool or reactor shutdown - as is this routine
  89. // may block forever if no new txs come in.
  90. func (memR *MempoolReactor) broadcastTxRoutine(peer p2p.Peer) {
  91. if !memR.config.Broadcast {
  92. return
  93. }
  94. var next *clist.CElement
  95. for {
  96. if !memR.IsRunning() || !peer.IsRunning() {
  97. return // Quit!
  98. }
  99. if next == nil {
  100. // This happens because the CElement we were looking at got
  101. // garbage collected (removed). That is, .NextWait() returned nil.
  102. // Go ahead and start from the beginning.
  103. next = memR.Mempool.TxsFrontWait() // Wait until a tx is available
  104. }
  105. memTx := next.Value.(*mempoolTx)
  106. // make sure the peer is up to date
  107. height := memTx.Height()
  108. if peerState_i := peer.Get(types.PeerStateKey); peerState_i != nil {
  109. peerState := peerState_i.(PeerState)
  110. if peerState.GetHeight() < height-1 { // Allow for a lag of 1 block
  111. time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
  112. continue
  113. }
  114. }
  115. // send memTx
  116. msg := &TxMessage{Tx: memTx.tx}
  117. success := peer.Send(MempoolChannel, struct{ MempoolMessage }{msg})
  118. if !success {
  119. time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
  120. continue
  121. }
  122. next = next.NextWait()
  123. continue
  124. }
  125. }
  126. //-----------------------------------------------------------------------------
  127. // Messages
  128. const (
  129. msgTypeTx = byte(0x01)
  130. )
  131. // MempoolMessage is a message sent or received by the MempoolReactor.
  132. type MempoolMessage interface{}
  133. var _ = wire.RegisterInterface(
  134. struct{ MempoolMessage }{},
  135. wire.ConcreteType{&TxMessage{}, msgTypeTx},
  136. )
  137. // DecodeMessage decodes a byte-array into a MempoolMessage.
  138. func DecodeMessage(bz []byte) (msgType byte, msg MempoolMessage, err error) {
  139. msgType = bz[0]
  140. n := new(int)
  141. r := bytes.NewReader(bz)
  142. msg = wire.ReadBinary(struct{ MempoolMessage }{}, r, maxMempoolMessageSize, n, &err).(struct{ MempoolMessage }).MempoolMessage
  143. return
  144. }
  145. //-------------------------------------
  146. // TxMessage is a MempoolMessage containing a transaction.
  147. type TxMessage struct {
  148. Tx types.Tx
  149. }
  150. // String returns a string representation of the TxMessage.
  151. func (m *TxMessage) String() string {
  152. return fmt.Sprintf("[TxMessage %v]", m.Tx)
  153. }