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.

376 lines
9.5 KiB

  1. package consensus
  2. import (
  3. "bufio"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "os"
  8. "reflect"
  9. "strconv"
  10. "strings"
  11. "time"
  12. auto "github.com/tendermint/go-autofile"
  13. . "github.com/tendermint/go-common"
  14. "github.com/tendermint/go-wire"
  15. sm "github.com/tendermint/tendermint/state"
  16. "github.com/tendermint/tendermint/types"
  17. )
  18. // Unmarshal and apply a single message to the consensus state
  19. // as if it were received in receiveRoutine
  20. // Lines that start with "#" are ignored.
  21. // NOTE: receiveRoutine should not be running
  22. func (cs *ConsensusState) readReplayMessage(msgBytes []byte, newStepCh chan interface{}) error {
  23. // Skip over empty and meta lines
  24. if len(msgBytes) == 0 || msgBytes[0] == '#' {
  25. return nil
  26. }
  27. var err error
  28. var msg TimedWALMessage
  29. wire.ReadJSON(&msg, msgBytes, &err)
  30. if err != nil {
  31. fmt.Println("MsgBytes:", msgBytes, string(msgBytes))
  32. return fmt.Errorf("Error reading json data: %v", err)
  33. }
  34. // for logging
  35. switch m := msg.Msg.(type) {
  36. case types.EventDataRoundState:
  37. log.Notice("Replay: New Step", "height", m.Height, "round", m.Round, "step", m.Step)
  38. // these are playback checks
  39. ticker := time.After(time.Second * 2)
  40. if newStepCh != nil {
  41. select {
  42. case mi := <-newStepCh:
  43. m2 := mi.(types.EventDataRoundState)
  44. if m.Height != m2.Height || m.Round != m2.Round || m.Step != m2.Step {
  45. return fmt.Errorf("RoundState mismatch. Got %v; Expected %v", m2, m)
  46. }
  47. case <-ticker:
  48. return fmt.Errorf("Failed to read off newStepCh")
  49. }
  50. }
  51. case msgInfo:
  52. peerKey := m.PeerKey
  53. if peerKey == "" {
  54. peerKey = "local"
  55. }
  56. switch msg := m.Msg.(type) {
  57. case *ProposalMessage:
  58. p := msg.Proposal
  59. log.Notice("Replay: Proposal", "height", p.Height, "round", p.Round, "header",
  60. p.BlockPartsHeader, "pol", p.POLRound, "peer", peerKey)
  61. case *BlockPartMessage:
  62. log.Notice("Replay: BlockPart", "height", msg.Height, "round", msg.Round, "peer", peerKey)
  63. case *VoteMessage:
  64. v := msg.Vote
  65. log.Notice("Replay: Vote", "height", v.Height, "round", v.Round, "type", v.Type,
  66. "blockID", v.BlockID, "peer", peerKey)
  67. }
  68. cs.handleMsg(m, cs.RoundState)
  69. case timeoutInfo:
  70. log.Notice("Replay: Timeout", "height", m.Height, "round", m.Round, "step", m.Step, "dur", m.Duration)
  71. cs.handleTimeout(m, cs.RoundState)
  72. default:
  73. return fmt.Errorf("Replay: Unknown TimedWALMessage type: %v", reflect.TypeOf(msg.Msg))
  74. }
  75. return nil
  76. }
  77. // replay only those messages since the last block.
  78. // timeoutRoutine should run concurrently to read off tickChan
  79. func (cs *ConsensusState) catchupReplay(csHeight int) error {
  80. // set replayMode
  81. cs.replayMode = true
  82. defer func() { cs.replayMode = false }()
  83. // Ensure that height+1 doesn't exist
  84. gr, found, err := cs.wal.group.Search("#HEIGHT: ", makeHeightSearchFunc(csHeight+1))
  85. if found {
  86. return errors.New(Fmt("WAL should not contain height %d.", csHeight+1))
  87. }
  88. if gr != nil {
  89. gr.Close()
  90. }
  91. // Search for height marker
  92. gr, found, err = cs.wal.group.Search("#HEIGHT: ", makeHeightSearchFunc(csHeight))
  93. if err == io.EOF {
  94. return nil
  95. } else if err != nil {
  96. return err
  97. }
  98. if !found {
  99. return errors.New(Fmt("WAL does not contain height %d.", csHeight))
  100. }
  101. defer gr.Close()
  102. log.Notice("Catchup by replaying consensus messages", "height", csHeight)
  103. for {
  104. line, err := gr.ReadLine()
  105. if err != nil {
  106. if err == io.EOF {
  107. break
  108. } else {
  109. return err
  110. }
  111. }
  112. // NOTE: since the priv key is set when the msgs are received
  113. // it will attempt to eg double sign but we can just ignore it
  114. // since the votes will be replayed and we'll get to the next step
  115. if err := cs.readReplayMessage([]byte(line), nil); err != nil {
  116. return err
  117. }
  118. }
  119. log.Notice("Replay: Done")
  120. return nil
  121. }
  122. //--------------------------------------------------------
  123. // replay messages interactively or all at once
  124. // Interactive playback
  125. func (cs ConsensusState) ReplayConsole(file string) error {
  126. return cs.replay(file, true)
  127. }
  128. // Full playback, with tests
  129. func (cs ConsensusState) ReplayMessages(file string) error {
  130. return cs.replay(file, false)
  131. }
  132. // replay all msgs or start the console
  133. func (cs *ConsensusState) replay(file string, console bool) error {
  134. if cs.IsRunning() {
  135. return errors.New("cs is already running, cannot replay")
  136. }
  137. if cs.wal != nil {
  138. return errors.New("cs wal is open, cannot replay")
  139. }
  140. cs.startForReplay()
  141. // ensure all new step events are regenerated as expected
  142. newStepCh := subscribeToEvent(cs.evsw, "replay-test", types.EventStringNewRoundStep(), 1)
  143. // just open the file for reading, no need to use wal
  144. fp, err := os.OpenFile(file, os.O_RDONLY, 0666)
  145. if err != nil {
  146. return err
  147. }
  148. pb := newPlayback(file, fp, cs, cs.state.Copy())
  149. defer pb.fp.Close()
  150. var nextN int // apply N msgs in a row
  151. for pb.scanner.Scan() {
  152. if nextN == 0 && console {
  153. nextN = pb.replayConsoleLoop()
  154. }
  155. if err := pb.cs.readReplayMessage(pb.scanner.Bytes(), newStepCh); err != nil {
  156. return err
  157. }
  158. if nextN > 0 {
  159. nextN -= 1
  160. }
  161. pb.count += 1
  162. }
  163. return nil
  164. }
  165. //------------------------------------------------
  166. // playback manager
  167. type playback struct {
  168. cs *ConsensusState
  169. fp *os.File
  170. scanner *bufio.Scanner
  171. count int // how many lines/msgs into the file are we
  172. // replays can be reset to beginning
  173. fileName string // so we can close/reopen the file
  174. genesisState *sm.State // so the replay session knows where to restart from
  175. }
  176. func newPlayback(fileName string, fp *os.File, cs *ConsensusState, genState *sm.State) *playback {
  177. return &playback{
  178. cs: cs,
  179. fp: fp,
  180. fileName: fileName,
  181. genesisState: genState,
  182. scanner: bufio.NewScanner(fp),
  183. }
  184. }
  185. // go back count steps by resetting the state and running (pb.count - count) steps
  186. func (pb *playback) replayReset(count int, newStepCh chan interface{}) error {
  187. pb.cs.Stop()
  188. pb.cs.Wait()
  189. newCS := NewConsensusState(pb.cs.config, pb.genesisState.Copy(), pb.cs.proxyAppConn, pb.cs.blockStore, pb.cs.mempool)
  190. newCS.SetEventSwitch(pb.cs.evsw)
  191. newCS.startForReplay()
  192. pb.fp.Close()
  193. fp, err := os.OpenFile(pb.fileName, os.O_RDONLY, 0666)
  194. if err != nil {
  195. return err
  196. }
  197. pb.fp = fp
  198. pb.scanner = bufio.NewScanner(fp)
  199. count = pb.count - count
  200. log.Notice(Fmt("Reseting from %d to %d", pb.count, count))
  201. pb.count = 0
  202. pb.cs = newCS
  203. for i := 0; pb.scanner.Scan() && i < count; i++ {
  204. if err := pb.cs.readReplayMessage(pb.scanner.Bytes(), newStepCh); err != nil {
  205. return err
  206. }
  207. pb.count += 1
  208. }
  209. return nil
  210. }
  211. func (cs *ConsensusState) startForReplay() {
  212. // don't want to start full cs
  213. cs.BaseService.OnStart()
  214. // since we replay tocks we just ignore ticks
  215. // TODO:!
  216. /*
  217. go func() {
  218. for {
  219. select {
  220. case <-cs.tickChan:
  221. case <-cs.Quit:
  222. return
  223. }
  224. }
  225. }()*/
  226. }
  227. // console function for parsing input and running commands
  228. func (pb *playback) replayConsoleLoop() int {
  229. for {
  230. fmt.Printf("> ")
  231. bufReader := bufio.NewReader(os.Stdin)
  232. line, more, err := bufReader.ReadLine()
  233. if more {
  234. Exit("input is too long")
  235. } else if err != nil {
  236. Exit(err.Error())
  237. }
  238. tokens := strings.Split(string(line), " ")
  239. if len(tokens) == 0 {
  240. continue
  241. }
  242. switch tokens[0] {
  243. case "next":
  244. // "next" -> replay next message
  245. // "next N" -> replay next N messages
  246. if len(tokens) == 1 {
  247. return 0
  248. } else {
  249. i, err := strconv.Atoi(tokens[1])
  250. if err != nil {
  251. fmt.Println("next takes an integer argument")
  252. } else {
  253. return i
  254. }
  255. }
  256. case "back":
  257. // "back" -> go back one message
  258. // "back N" -> go back N messages
  259. // NOTE: "back" is not supported in the state machine design,
  260. // so we restart and replay up to
  261. // ensure all new step events are regenerated as expected
  262. newStepCh := subscribeToEvent(pb.cs.evsw, "replay-test", types.EventStringNewRoundStep(), 1)
  263. if len(tokens) == 1 {
  264. pb.replayReset(1, newStepCh)
  265. } else {
  266. i, err := strconv.Atoi(tokens[1])
  267. if err != nil {
  268. fmt.Println("back takes an integer argument")
  269. } else if i > pb.count {
  270. fmt.Printf("argument to back must not be larger than the current count (%d)\n", pb.count)
  271. } else {
  272. pb.replayReset(i, newStepCh)
  273. }
  274. }
  275. case "rs":
  276. // "rs" -> print entire round state
  277. // "rs short" -> print height/round/step
  278. // "rs <field>" -> print another field of the round state
  279. rs := pb.cs.RoundState
  280. if len(tokens) == 1 {
  281. fmt.Println(rs)
  282. } else {
  283. switch tokens[1] {
  284. case "short":
  285. fmt.Printf("%v/%v/%v\n", rs.Height, rs.Round, rs.Step)
  286. case "validators":
  287. fmt.Println(rs.Validators)
  288. case "proposal":
  289. fmt.Println(rs.Proposal)
  290. case "proposal_block":
  291. fmt.Printf("%v %v\n", rs.ProposalBlockParts.StringShort(), rs.ProposalBlock.StringShort())
  292. case "locked_round":
  293. fmt.Println(rs.LockedRound)
  294. case "locked_block":
  295. fmt.Printf("%v %v\n", rs.LockedBlockParts.StringShort(), rs.LockedBlock.StringShort())
  296. case "votes":
  297. fmt.Println(rs.Votes.StringIndented(" "))
  298. default:
  299. fmt.Println("Unknown option", tokens[1])
  300. }
  301. }
  302. case "n":
  303. fmt.Println(pb.count)
  304. }
  305. }
  306. return 0
  307. }
  308. //--------------------------------------------------------------------------------
  309. // Parses marker lines of the form:
  310. // #HEIGHT: 12345
  311. func makeHeightSearchFunc(height int) auto.SearchFunc {
  312. return func(line string) (int, error) {
  313. line = strings.TrimRight(line, "\n")
  314. parts := strings.Split(line, " ")
  315. if len(parts) != 2 {
  316. return -1, errors.New("Line did not have 2 parts")
  317. }
  318. i, err := strconv.Atoi(parts[1])
  319. if err != nil {
  320. return -1, errors.New("Failed to parse INFO: " + err.Error())
  321. }
  322. if height < i {
  323. return 1, nil
  324. } else if height == i {
  325. return 0, nil
  326. } else {
  327. return -1, nil
  328. }
  329. }
  330. }