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.

186 lines
5.0 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
9 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() int
  86. }
  87. // Peer describes a peer.
  88. type Peer interface {
  89. IsRunning() bool
  90. Send(byte, interface{}) bool
  91. Get(string) interface{}
  92. }
  93. // Send new mempool txs to peer.
  94. // TODO: Handle mempool or reactor shutdown?
  95. // As is this routine may block forever if no new txs come in.
  96. func (memR *MempoolReactor) broadcastTxRoutine(peer Peer) {
  97. if !memR.config.Broadcast {
  98. return
  99. }
  100. var next *clist.CElement
  101. for {
  102. if !memR.IsRunning() || !peer.IsRunning() {
  103. return // Quit!
  104. }
  105. if next == nil {
  106. // This happens because the CElement we were looking at got
  107. // garbage collected (removed). That is, .NextWait() returned nil.
  108. // Go ahead and start from the beginning.
  109. next = memR.Mempool.TxsFrontWait() // Wait until a tx is available
  110. }
  111. memTx := next.Value.(*mempoolTx)
  112. // make sure the peer is up to date
  113. height := memTx.Height()
  114. if peerState_i := peer.Get(types.PeerStateKey); peerState_i != nil {
  115. peerState := peerState_i.(PeerState)
  116. if peerState.GetHeight() < height-1 { // Allow for a lag of 1 block
  117. time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
  118. continue
  119. }
  120. }
  121. // send memTx
  122. msg := &TxMessage{Tx: memTx.tx}
  123. success := peer.Send(MempoolChannel, struct{ MempoolMessage }{msg})
  124. if !success {
  125. time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
  126. continue
  127. }
  128. next = next.NextWait()
  129. continue
  130. }
  131. }
  132. //-----------------------------------------------------------------------------
  133. // Messages
  134. const (
  135. msgTypeTx = byte(0x01)
  136. )
  137. // MempoolMessage is a message sent or received by the MempoolReactor.
  138. type MempoolMessage interface{}
  139. var _ = wire.RegisterInterface(
  140. struct{ MempoolMessage }{},
  141. wire.ConcreteType{&TxMessage{}, msgTypeTx},
  142. )
  143. // DecodeMessage decodes a byte-array into a MempoolMessage.
  144. func DecodeMessage(bz []byte) (msgType byte, msg MempoolMessage, err error) {
  145. msgType = bz[0]
  146. n := new(int)
  147. r := bytes.NewReader(bz)
  148. msg = wire.ReadBinary(struct{ MempoolMessage }{}, r, maxMempoolMessageSize, n, &err).(struct{ MempoolMessage }).MempoolMessage
  149. return
  150. }
  151. //-------------------------------------
  152. // TxMessage is a MempoolMessage containing a transaction.
  153. type TxMessage struct {
  154. Tx types.Tx
  155. }
  156. // String returns a string representation of the TxMessage.
  157. func (m *TxMessage) String() string {
  158. return fmt.Sprintf("[TxMessage %v]", m.Tx)
  159. }