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.

242 lines
7.0 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
abci: localClient improvements & bugfixes & pubsub Unsubscribe issues (#2748) * use READ lock/unlock in ConsensusState#GetLastHeight Refs #2721 * do not use defers when there's no need * fix peer formatting (output its address instead of the pointer) ``` [54310]: E[11-02|11:59:39.851] Connection failed @ sendRoutine module=p2p peer=0xb78f00 conn=MConn{74.207.236.148:26656} err="pong timeout" ``` https://github.com/tendermint/tendermint/issues/2721#issuecomment-435326581 * panic if peer has no state https://github.com/tendermint/tendermint/issues/2721#issuecomment-435347165 It's confusing that sometimes we check if peer has a state, but most of the times we expect it to be there 1. https://github.com/tendermint/tendermint/blob/add79700b5fe84417538202b6c927c8cc5383672/mempool/reactor.go#L138 2. https://github.com/tendermint/tendermint/blob/add79700b5fe84417538202b6c927c8cc5383672/rpc/core/consensus.go#L196 (edited) I will change everything to always assume peer has a state and panic otherwise that should help identify issues earlier * abci/localclient: extend lock on app callback App callback should be protected by lock as well (note this was already done for InitChainAsync, why not for others???). Otherwise, when we execute the block, tx might come in and call the callback in the same time we're updating it in execBlockOnProxyApp => DATA RACE Fixes #2721 Consensus state is locked ``` goroutine 113333 [semacquire, 309 minutes]: sync.runtime_SemacquireMutex(0xc00180009c, 0xc0000c7e00) /usr/local/go/src/runtime/sema.go:71 +0x3d sync.(*RWMutex).RLock(0xc001800090) /usr/local/go/src/sync/rwmutex.go:50 +0x4e github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).GetRoundState(0xc001800000, 0x0) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:218 +0x46 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusReactor).queryMaj23Routine(0xc0017def80, 0x11104a0, 0xc0072488f0, 0xc007248 9c0) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/reactor.go:735 +0x16d created by github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusReactor).AddPeer /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/reactor.go:172 +0x236 ``` because localClient is locked ``` goroutine 1899 [semacquire, 309 minutes]: sync.runtime_SemacquireMutex(0xc00003363c, 0xc0000cb500) /usr/local/go/src/runtime/sema.go:71 +0x3d sync.(*Mutex).Lock(0xc000033638) /usr/local/go/src/sync/mutex.go:134 +0xff github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/abci/client.(*localClient).SetResponseCallback(0xc0001fb560, 0xc007868540) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/abci/client/local_client.go:32 +0x33 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/proxy.(*appConnConsensus).SetResponseCallback(0xc00002f750, 0xc007868540) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/proxy/app_conn.go:57 +0x40 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/state.execBlockOnProxyApp(0x1104e20, 0xc002ca0ba0, 0x11092a0, 0xc00002f750, 0xc0001fe960, 0xc000bfc660, 0x110cfe0, 0xc000090330, 0xc9d12, 0xc000d9d5a0, ...) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/state/execution.go:230 +0x1fd github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/state.(*BlockExecutor).ApplyBlock(0xc002c2a230, 0x7, 0x0, 0xc000eae880, 0x6, 0xc002e52c60, 0x16, 0x1f927, 0xc9d12, 0xc000d9d5a0, ...) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/state/execution.go:96 +0x142 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).finalizeCommit(0xc001800000, 0x1f928) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:1339 +0xa3e github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).tryFinalizeCommit(0xc001800000, 0x1f928) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:1270 +0x451 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).enterCommit.func1(0xc001800000, 0x0, 0x1f928) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:1218 +0x90 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).enterCommit(0xc001800000, 0x1f928, 0x0) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:1247 +0x6b8 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).addVote(0xc001800000, 0xc003d8dea0, 0xc000cf4cc0, 0x28, 0xf1, 0xc003bc7ad0, 0xc003bc7b10) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:1659 +0xbad github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).tryAddVote(0xc001800000, 0xc003d8dea0, 0xc000cf4cc0, 0x28, 0xf1, 0xf1, 0xf1) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:1517 +0x59 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).handleMsg(0xc001800000, 0xd98200, 0xc0070dbed0, 0xc000cf4cc0, 0x28) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:660 +0x64b github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).receiveRoutine(0xc001800000, 0x0) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:617 +0x670 created by github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus.(*ConsensusState).OnStart /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/consensus/state.go:311 +0x132 ``` tx comes in and CheckTx is executed right when we execute the block ``` goroutine 111044 [semacquire, 309 minutes]: sync.runtime_SemacquireMutex(0xc00003363c, 0x0) /usr/local/go/src/runtime/sema.go:71 +0x3d sync.(*Mutex).Lock(0xc000033638) /usr/local/go/src/sync/mutex.go:134 +0xff github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/abci/client.(*localClient).CheckTxAsync(0xc0001fb0e0, 0xc002d94500, 0x13f, 0x280, 0x0) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/abci/client/local_client.go:85 +0x47 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/proxy.(*appConnMempool).CheckTxAsync(0xc00002f720, 0xc002d94500, 0x13f, 0x280, 0x1) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/proxy/app_conn.go:114 +0x51 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/mempool.(*Mempool).CheckTx(0xc002d3a320, 0xc002d94500, 0x13f, 0x280, 0xc0072355f0, 0x0, 0x0) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/mempool/mempool.go:316 +0x17b github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/rpc/core.BroadcastTxSync(0xc002d94500, 0x13f, 0x280, 0x0, 0x0, 0x0) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/rpc/core/mempool.go:93 +0xb8 reflect.Value.call(0xd85560, 0x10326c0, 0x13, 0xec7b8b, 0x4, 0xc00663f180, 0x1, 0x1, 0xc00663f180, 0xc00663f188, ...) /usr/local/go/src/reflect/value.go:447 +0x449 reflect.Value.Call(0xd85560, 0x10326c0, 0x13, 0xc00663f180, 0x1, 0x1, 0x0, 0x0, 0xc005cc9344) /usr/local/go/src/reflect/value.go:308 +0xa4 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/rpc/lib/server.makeHTTPHandler.func2(0x1102060, 0xc00663f100, 0xc0082d7900) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/rpc/lib/server/handlers.go:269 +0x188 net/http.HandlerFunc.ServeHTTP(0xc002c81f20, 0x1102060, 0xc00663f100, 0xc0082d7900) /usr/local/go/src/net/http/server.go:1964 +0x44 net/http.(*ServeMux).ServeHTTP(0xc002c81b60, 0x1102060, 0xc00663f100, 0xc0082d7900) /usr/local/go/src/net/http/server.go:2361 +0x127 github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/rpc/lib/server.maxBytesHandler.ServeHTTP(0x10f8a40, 0xc002c81b60, 0xf4240, 0x1102060, 0xc00663f100, 0xc0082d7900) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/rpc/lib/server/http_server.go:219 +0xcf github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/rpc/lib/server.RecoverAndLogHandler.func1(0x1103220, 0xc00121e620, 0xc0082d7900) /root/go/src/github.com/MinterTeam/minter-go-node/vendor/github.com/tendermint/tendermint/rpc/lib/server/http_server.go:192 +0x394 net/http.HandlerFunc.ServeHTTP(0xc002c06ea0, 0x1103220, 0xc00121e620, 0xc0082d7900) /usr/local/go/src/net/http/server.go:1964 +0x44 net/http.serverHandler.ServeHTTP(0xc001a1aa90, 0x1103220, 0xc00121e620, 0xc0082d7900) /usr/local/go/src/net/http/server.go:2741 +0xab net/http.(*conn).serve(0xc00785a3c0, 0x11041a0, 0xc000f844c0) /usr/local/go/src/net/http/server.go:1847 +0x646 created by net/http.(*Server).Serve /usr/local/go/src/net/http/server.go:2851 +0x2f5 ``` * consensus: use read lock in Receive#VoteMessage * use defer to unlock mutex because application might panic * use defer in every method of the localClient * add a changelog entry * drain channels before Unsubscribe(All) Read https://github.com/tendermint/tendermint/blob/55362ed76630f3e1ebec159a598f6a9fb5892cb1/libs/pubsub/pubsub.go#L13 for the detailed explanation of the issue. We'll need to fix it someday. Make sure to keep an eye on https://github.com/tendermint/tendermint/blob/master/docs/architecture/adr-033-pubsub.md * retry instead of panic when peer has no state in reactors other than consensus in /dump_consensus_state RPC endpoint, skip a peer with no state * rpc/core/mempool: simplify error messages * rpc/core/mempool: use time.After instead of timer also, do not log DeliverTx result (to be consistent with other memthods) * unlock before calling the callback in reqRes#SetCallback
6 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. if err = msg.ValidateBasic(); err != nil {
  65. evR.Logger.Error("Peer sent us invalid msg", "peer", src, "msg", msg, "err", err)
  66. evR.Switch.StopPeerForError(src, err)
  67. return
  68. }
  69. evR.Logger.Debug("Receive", "src", src, "chId", chID, "msg", msg)
  70. switch msg := msg.(type) {
  71. case *EvidenceListMessage:
  72. for _, ev := range msg.Evidence {
  73. err := evR.evpool.AddEvidence(ev)
  74. if err != nil {
  75. evR.Logger.Info("Evidence is not valid", "evidence", msg.Evidence, "err", err)
  76. // punish peer
  77. evR.Switch.StopPeerForError(src, err)
  78. }
  79. }
  80. default:
  81. evR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
  82. }
  83. }
  84. // SetEventSwitch implements events.Eventable.
  85. func (evR *EvidenceReactor) SetEventBus(b *types.EventBus) {
  86. evR.eventBus = b
  87. }
  88. // Modeled after the mempool routine.
  89. // - Evidence accumulates in a clist.
  90. // - Each peer has a routien that iterates through the clist,
  91. // sending available evidence to the peer.
  92. // - If we're waiting for new evidence and the list is not empty,
  93. // start iterating from the beginning again.
  94. func (evR *EvidenceReactor) broadcastEvidenceRoutine(peer p2p.Peer) {
  95. var next *clist.CElement
  96. for {
  97. // This happens because the CElement we were looking at got garbage
  98. // collected (removed). That is, .NextWait() returned nil. Go ahead and
  99. // start from the beginning.
  100. if next == nil {
  101. select {
  102. case <-evR.evpool.EvidenceWaitChan(): // Wait until evidence is available
  103. if next = evR.evpool.EvidenceFront(); next == nil {
  104. continue
  105. }
  106. case <-peer.Quit():
  107. return
  108. case <-evR.Quit():
  109. return
  110. }
  111. }
  112. ev := next.Value.(types.Evidence)
  113. msg, retry := evR.checkSendEvidenceMessage(peer, ev)
  114. if msg != nil {
  115. success := peer.Send(EvidenceChannel, cdc.MustMarshalBinaryBare(msg))
  116. retry = !success
  117. }
  118. if retry {
  119. time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
  120. continue
  121. }
  122. afterCh := time.After(time.Second * broadcastEvidenceIntervalS)
  123. select {
  124. case <-afterCh:
  125. // start from the beginning every tick.
  126. // TODO: only do this if we're at the end of the list!
  127. next = nil
  128. case <-next.NextWaitChan():
  129. // see the start of the for loop for nil check
  130. next = next.Next()
  131. case <-peer.Quit():
  132. return
  133. case <-evR.Quit():
  134. return
  135. }
  136. }
  137. }
  138. // Returns the message to send the peer, or nil if the evidence is invalid for the peer.
  139. // If message is nil, return true if we should sleep and try again.
  140. func (evR EvidenceReactor) checkSendEvidenceMessage(peer p2p.Peer, ev types.Evidence) (msg EvidenceMessage, retry bool) {
  141. // make sure the peer is up to date
  142. evHeight := ev.Height()
  143. peerState, ok := peer.Get(types.PeerStateKey).(PeerState)
  144. if !ok {
  145. // Peer does not have a state yet. We set it in the consensus reactor, but
  146. // when we add peer in Switch, the order we call reactors#AddPeer is
  147. // different every time due to us using a map. Sometimes other reactors
  148. // will be initialized before the consensus reactor. We should wait a few
  149. // milliseconds and retry.
  150. return nil, true
  151. }
  152. // NOTE: We only send evidence to peers where
  153. // peerHeight - maxAge < evidenceHeight < peerHeight
  154. maxAge := evR.evpool.State().ConsensusParams.Evidence.MaxAge
  155. peerHeight := peerState.GetHeight()
  156. if peerHeight < evHeight {
  157. // peer is behind. sleep while he catches up
  158. return nil, true
  159. } else if peerHeight > evHeight+maxAge {
  160. // evidence is too old, skip
  161. // NOTE: if evidence is too old for an honest peer,
  162. // then we're behind and either it already got committed or it never will!
  163. evR.Logger.Info("Not sending peer old evidence", "peerHeight", peerHeight, "evHeight", evHeight, "maxAge", maxAge, "peer", peer)
  164. return nil, false
  165. }
  166. // send evidence
  167. msg = &EvidenceListMessage{[]types.Evidence{ev}}
  168. return msg, false
  169. }
  170. // PeerState describes the state of a peer.
  171. type PeerState interface {
  172. GetHeight() int64
  173. }
  174. //-----------------------------------------------------------------------------
  175. // Messages
  176. // EvidenceMessage is a message sent or received by the EvidenceReactor.
  177. type EvidenceMessage interface {
  178. ValidateBasic() error
  179. }
  180. func RegisterEvidenceMessages(cdc *amino.Codec) {
  181. cdc.RegisterInterface((*EvidenceMessage)(nil), nil)
  182. cdc.RegisterConcrete(&EvidenceListMessage{},
  183. "tendermint/evidence/EvidenceListMessage", nil)
  184. }
  185. func decodeMsg(bz []byte) (msg EvidenceMessage, err error) {
  186. if len(bz) > maxMsgSize {
  187. return msg, fmt.Errorf("Msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
  188. }
  189. err = cdc.UnmarshalBinaryBare(bz, &msg)
  190. return
  191. }
  192. //-------------------------------------
  193. // EvidenceListMessage contains a list of evidence.
  194. type EvidenceListMessage struct {
  195. Evidence []types.Evidence
  196. }
  197. // ValidateBasic performs basic validation.
  198. func (m *EvidenceListMessage) ValidateBasic() error {
  199. for i, ev := range m.Evidence {
  200. if err := ev.ValidateBasic(); err != nil {
  201. return fmt.Errorf("Invalid evidence (#%d): %v", i, err)
  202. }
  203. }
  204. return nil
  205. }
  206. // String returns a string representation of the EvidenceListMessage.
  207. func (m *EvidenceListMessage) String() string {
  208. return fmt.Sprintf("[EvidenceListMessage %v]", m.Evidence)
  209. }