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.

220 lines
6.3 KiB

7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
  1. package evidence
  2. import (
  3. "fmt"
  4. "reflect"
  5. "time"
  6. amino "github.com/tendermint/go-amino"
  7. clist "github.com/tendermint/tendermint/libs/clist"
  8. "github.com/tendermint/tendermint/libs/log"
  9. "github.com/tendermint/tendermint/p2p"
  10. "github.com/tendermint/tendermint/types"
  11. )
  12. const (
  13. EvidenceChannel = byte(0x38)
  14. maxMsgSize = 1048576 // 1MB TODO make it configurable
  15. broadcastEvidenceIntervalS = 60 // broadcast uncommitted evidence this often
  16. peerCatchupSleepIntervalMS = 100 // If peer is behind, sleep this amount
  17. )
  18. // EvidenceReactor handles evpool evidence broadcasting amongst peers.
  19. type EvidenceReactor struct {
  20. p2p.BaseReactor
  21. evpool *EvidencePool
  22. eventBus *types.EventBus
  23. }
  24. // NewEvidenceReactor returns a new EvidenceReactor with the given config and evpool.
  25. func NewEvidenceReactor(evpool *EvidencePool) *EvidenceReactor {
  26. evR := &EvidenceReactor{
  27. evpool: evpool,
  28. }
  29. evR.BaseReactor = *p2p.NewBaseReactor("EvidenceReactor", evR)
  30. return evR
  31. }
  32. // SetLogger sets the Logger on the reactor and the underlying Evidence.
  33. func (evR *EvidenceReactor) SetLogger(l log.Logger) {
  34. evR.Logger = l
  35. evR.evpool.SetLogger(l)
  36. }
  37. // GetChannels implements Reactor.
  38. // It returns the list of channels for this reactor.
  39. func (evR *EvidenceReactor) GetChannels() []*p2p.ChannelDescriptor {
  40. return []*p2p.ChannelDescriptor{
  41. &p2p.ChannelDescriptor{
  42. ID: EvidenceChannel,
  43. Priority: 5,
  44. },
  45. }
  46. }
  47. // AddPeer implements Reactor.
  48. func (evR *EvidenceReactor) AddPeer(peer p2p.Peer) {
  49. go evR.broadcastEvidenceRoutine(peer)
  50. }
  51. // RemovePeer implements Reactor.
  52. func (evR *EvidenceReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
  53. // nothing to do
  54. }
  55. // Receive implements Reactor.
  56. // It adds any received evidence to the evpool.
  57. func (evR *EvidenceReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
  58. msg, err := decodeMsg(msgBytes)
  59. if err != nil {
  60. evR.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes)
  61. evR.Switch.StopPeerForError(src, err)
  62. return
  63. }
  64. evR.Logger.Debug("Receive", "src", src, "chId", chID, "msg", msg)
  65. switch msg := msg.(type) {
  66. case *EvidenceListMessage:
  67. for _, ev := range msg.Evidence {
  68. err := evR.evpool.AddEvidence(ev)
  69. if err != nil {
  70. evR.Logger.Info("Evidence is not valid", "evidence", msg.Evidence, "err", err)
  71. // punish peer
  72. evR.Switch.StopPeerForError(src, err)
  73. }
  74. }
  75. default:
  76. evR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
  77. }
  78. }
  79. // SetEventSwitch implements events.Eventable.
  80. func (evR *EvidenceReactor) SetEventBus(b *types.EventBus) {
  81. evR.eventBus = b
  82. }
  83. // Modeled after the mempool routine.
  84. // - Evidence accumulates in a clist.
  85. // - Each peer has a routien that iterates through the clist,
  86. // sending available evidence to the peer.
  87. // - If we're waiting for new evidence and the list is not empty,
  88. // start iterating from the beginning again.
  89. func (evR *EvidenceReactor) broadcastEvidenceRoutine(peer p2p.Peer) {
  90. var next *clist.CElement
  91. for {
  92. // This happens because the CElement we were looking at got garbage
  93. // collected (removed). That is, .NextWait() returned nil. Go ahead and
  94. // start from the beginning.
  95. if next == nil {
  96. select {
  97. case <-evR.evpool.EvidenceWaitChan(): // Wait until evidence is available
  98. if next = evR.evpool.EvidenceFront(); next == nil {
  99. continue
  100. }
  101. case <-peer.Quit():
  102. return
  103. case <-evR.Quit():
  104. return
  105. }
  106. }
  107. ev := next.Value.(types.Evidence)
  108. msg, retry := evR.checkSendEvidenceMessage(peer, ev)
  109. if msg != nil {
  110. success := peer.Send(EvidenceChannel, cdc.MustMarshalBinaryBare(msg))
  111. retry = !success
  112. }
  113. if retry {
  114. time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
  115. continue
  116. }
  117. afterCh := time.After(time.Second * broadcastEvidenceIntervalS)
  118. select {
  119. case <-afterCh:
  120. // start from the beginning every tick.
  121. // TODO: only do this if we're at the end of the list!
  122. next = nil
  123. case <-next.NextWaitChan():
  124. // see the start of the for loop for nil check
  125. next = next.Next()
  126. case <-peer.Quit():
  127. return
  128. case <-evR.Quit():
  129. return
  130. }
  131. }
  132. }
  133. // Returns the message to send the peer, or nil if the evidence is invalid for the peer.
  134. // If message is nil, return true if we should sleep and try again.
  135. func (evR EvidenceReactor) checkSendEvidenceMessage(peer p2p.Peer, ev types.Evidence) (msg EvidenceMessage, retry bool) {
  136. // make sure the peer is up to date
  137. evHeight := ev.Height()
  138. peerState, ok := peer.Get(types.PeerStateKey).(PeerState)
  139. if !ok {
  140. evR.Logger.Info("Found peer without PeerState", "peer", peer)
  141. return nil, true
  142. }
  143. // NOTE: We only send evidence to peers where
  144. // peerHeight - maxAge < evidenceHeight < peerHeight
  145. maxAge := evR.evpool.State().ConsensusParams.EvidenceParams.MaxAge
  146. peerHeight := peerState.GetHeight()
  147. if peerHeight < evHeight {
  148. // peer is behind. sleep while he catches up
  149. return nil, true
  150. } else if peerHeight > evHeight+maxAge {
  151. // evidence is too old, skip
  152. // NOTE: if evidence is too old for an honest peer,
  153. // then we're behind and either it already got committed or it never will!
  154. evR.Logger.Info("Not sending peer old evidence", "peerHeight", peerHeight, "evHeight", evHeight, "maxAge", maxAge, "peer", peer)
  155. return nil, false
  156. }
  157. // send evidence
  158. msg = &EvidenceListMessage{[]types.Evidence{ev}}
  159. return msg, false
  160. }
  161. // PeerState describes the state of a peer.
  162. type PeerState interface {
  163. GetHeight() int64
  164. }
  165. //-----------------------------------------------------------------------------
  166. // Messages
  167. // EvidenceMessage is a message sent or received by the EvidenceReactor.
  168. type EvidenceMessage interface{}
  169. func RegisterEvidenceMessages(cdc *amino.Codec) {
  170. cdc.RegisterInterface((*EvidenceMessage)(nil), nil)
  171. cdc.RegisterConcrete(&EvidenceListMessage{},
  172. "tendermint/evidence/EvidenceListMessage", nil)
  173. }
  174. func decodeMsg(bz []byte) (msg EvidenceMessage, err error) {
  175. if len(bz) > maxMsgSize {
  176. return msg, fmt.Errorf("Msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
  177. }
  178. err = cdc.UnmarshalBinaryBare(bz, &msg)
  179. return
  180. }
  181. //-------------------------------------
  182. // EvidenceMessage contains a list of evidence.
  183. type EvidenceListMessage struct {
  184. Evidence []types.Evidence
  185. }
  186. // String returns a string representation of the EvidenceListMessage.
  187. func (m *EvidenceListMessage) String() string {
  188. return fmt.Sprintf("[EvidenceListMessage %v]", m.Evidence)
  189. }