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.

247 lines
6.9 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
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. {
  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. // Receive implements Reactor.
  52. // It adds any received evidence to the evpool.
  53. func (evR *EvidenceReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
  54. msg, err := decodeMsg(msgBytes)
  55. if err != nil {
  56. evR.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes)
  57. evR.Switch.StopPeerForError(src, err)
  58. return
  59. }
  60. if err = msg.ValidateBasic(); err != nil {
  61. evR.Logger.Error("Peer sent us invalid msg", "peer", src, "msg", msg, "err", err)
  62. evR.Switch.StopPeerForError(src, err)
  63. return
  64. }
  65. evR.Logger.Debug("Receive", "src", src, "chId", chID, "msg", msg)
  66. switch msg := msg.(type) {
  67. case *EvidenceListMessage:
  68. for _, ev := range msg.Evidence {
  69. err := evR.evpool.AddEvidence(ev)
  70. if err != nil {
  71. evR.Logger.Info("Evidence is not valid", "evidence", msg.Evidence, "err", err)
  72. // punish peer
  73. evR.Switch.StopPeerForError(src, err)
  74. }
  75. }
  76. default:
  77. evR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
  78. }
  79. }
  80. // SetEventSwitch implements events.Eventable.
  81. func (evR *EvidenceReactor) SetEventBus(b *types.EventBus) {
  82. evR.eventBus = b
  83. }
  84. // Modeled after the mempool routine.
  85. // - Evidence accumulates in a clist.
  86. // - Each peer has a routien that iterates through the clist,
  87. // sending available evidence to the peer.
  88. // - If we're waiting for new evidence and the list is not empty,
  89. // start iterating from the beginning again.
  90. func (evR *EvidenceReactor) broadcastEvidenceRoutine(peer p2p.Peer) {
  91. var next *clist.CElement
  92. for {
  93. // This happens because the CElement we were looking at got garbage
  94. // collected (removed). That is, .NextWait() returned nil. Go ahead and
  95. // start from the beginning.
  96. if next == nil {
  97. select {
  98. case <-evR.evpool.EvidenceWaitChan(): // Wait until evidence is available
  99. if next = evR.evpool.EvidenceFront(); next == nil {
  100. continue
  101. }
  102. case <-peer.Quit():
  103. return
  104. case <-evR.Quit():
  105. return
  106. }
  107. }
  108. ev := next.Value.(types.Evidence)
  109. msg, retry := evR.checkSendEvidenceMessage(peer, ev)
  110. if msg != nil {
  111. success := peer.Send(EvidenceChannel, cdc.MustMarshalBinaryBare(msg))
  112. retry = !success
  113. }
  114. if retry {
  115. time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
  116. continue
  117. }
  118. afterCh := time.After(time.Second * broadcastEvidenceIntervalS)
  119. select {
  120. case <-afterCh:
  121. // start from the beginning every tick.
  122. // TODO: only do this if we're at the end of the list!
  123. next = nil
  124. case <-next.NextWaitChan():
  125. // see the start of the for loop for nil check
  126. next = next.Next()
  127. case <-peer.Quit():
  128. return
  129. case <-evR.Quit():
  130. return
  131. }
  132. }
  133. }
  134. // Returns the message to send the peer, or nil if the evidence is invalid for the peer.
  135. // If message is nil, return true if we should sleep and try again.
  136. func (evR EvidenceReactor) checkSendEvidenceMessage(
  137. peer p2p.Peer,
  138. ev types.Evidence,
  139. ) (msg EvidenceMessage, retry bool) {
  140. // make sure the peer is up to date
  141. evHeight := ev.Height()
  142. peerState, ok := peer.Get(types.PeerStateKey).(PeerState)
  143. if !ok {
  144. // Peer does not have a state yet. We set it in the consensus reactor, but
  145. // when we add peer in Switch, the order we call reactors#AddPeer is
  146. // different every time due to us using a map. Sometimes other reactors
  147. // will be initialized before the consensus reactor. We should wait a few
  148. // milliseconds and retry.
  149. return nil, true
  150. }
  151. // NOTE: We only send evidence to peers where
  152. // peerHeight - maxAge < evidenceHeight < peerHeight
  153. maxAge := evR.evpool.State().ConsensusParams.Evidence.MaxAge
  154. peerHeight := peerState.GetHeight()
  155. if peerHeight < evHeight {
  156. // peer is behind. sleep while he catches up
  157. return nil, true
  158. } else if peerHeight > evHeight+maxAge {
  159. // evidence is too old, skip
  160. // NOTE: if evidence is too old for an honest peer,
  161. // then we're behind and either it already got committed or it never will!
  162. evR.Logger.Info(
  163. "Not sending peer old evidence",
  164. "peerHeight", peerHeight,
  165. "evHeight", evHeight,
  166. "maxAge", maxAge,
  167. "peer", peer,
  168. )
  169. return nil, false
  170. }
  171. // send evidence
  172. msg = &EvidenceListMessage{[]types.Evidence{ev}}
  173. return msg, false
  174. }
  175. // PeerState describes the state of a peer.
  176. type PeerState interface {
  177. GetHeight() int64
  178. }
  179. //-----------------------------------------------------------------------------
  180. // Messages
  181. // EvidenceMessage is a message sent or received by the EvidenceReactor.
  182. type EvidenceMessage interface {
  183. ValidateBasic() error
  184. }
  185. func RegisterEvidenceMessages(cdc *amino.Codec) {
  186. cdc.RegisterInterface((*EvidenceMessage)(nil), nil)
  187. cdc.RegisterConcrete(&EvidenceListMessage{},
  188. "tendermint/evidence/EvidenceListMessage", nil)
  189. }
  190. func decodeMsg(bz []byte) (msg EvidenceMessage, err error) {
  191. if len(bz) > maxMsgSize {
  192. return msg, fmt.Errorf("msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
  193. }
  194. err = cdc.UnmarshalBinaryBare(bz, &msg)
  195. return
  196. }
  197. //-------------------------------------
  198. // EvidenceListMessage contains a list of evidence.
  199. type EvidenceListMessage struct {
  200. Evidence []types.Evidence
  201. }
  202. // ValidateBasic performs basic validation.
  203. func (m *EvidenceListMessage) ValidateBasic() error {
  204. for i, ev := range m.Evidence {
  205. if err := ev.ValidateBasic(); err != nil {
  206. return fmt.Errorf("invalid evidence (#%d): %v", i, err)
  207. }
  208. }
  209. return nil
  210. }
  211. // String returns a string representation of the EvidenceListMessage.
  212. func (m *EvidenceListMessage) String() string {
  213. return fmt.Sprintf("[EvidenceListMessage %v]", m.Evidence)
  214. }