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.

111 lines
2.7 KiB

8 years ago
8 years ago
8 years ago
  1. package consensus
  2. import (
  3. "time"
  4. auto "github.com/tendermint/go-autofile"
  5. . "github.com/tendermint/go-common"
  6. "github.com/tendermint/go-wire"
  7. "github.com/tendermint/tendermint/types"
  8. )
  9. //--------------------------------------------------------
  10. // types and functions for savings consensus messages
  11. type TimedWALMessage struct {
  12. Time time.Time `json:"time"`
  13. Msg WALMessage `json:"msg"`
  14. }
  15. type WALMessage interface{}
  16. var _ = wire.RegisterInterface(
  17. struct{ WALMessage }{},
  18. wire.ConcreteType{types.EventDataRoundState{}, 0x01},
  19. wire.ConcreteType{msgInfo{}, 0x02},
  20. wire.ConcreteType{timeoutInfo{}, 0x03},
  21. )
  22. //--------------------------------------------------------
  23. // Simple write-ahead logger
  24. // Write ahead logger writes msgs to disk before they are processed.
  25. // Can be used for crash-recovery and deterministic replay
  26. // TODO: currently the wal is overwritten during replay catchup
  27. // give it a mode so it's either reading or appending - must read to end to start appending again
  28. type WAL struct {
  29. BaseService
  30. group *auto.Group
  31. light bool // ignore block parts
  32. }
  33. func NewWAL(walFile string, light bool) (*WAL, error) {
  34. group, err := auto.OpenGroup(walFile)
  35. if err != nil {
  36. return nil, err
  37. }
  38. wal := &WAL{
  39. group: group,
  40. light: light,
  41. }
  42. wal.BaseService = *NewBaseService(log, "WAL", wal)
  43. _, err = wal.Start()
  44. return wal, err
  45. }
  46. func (wal *WAL) OnStart() error {
  47. size, err := wal.group.Head.Size()
  48. if err != nil {
  49. return err
  50. } else if size == 0 {
  51. wal.writeHeight(1)
  52. }
  53. _, err = wal.group.Start()
  54. return err
  55. }
  56. func (wal *WAL) OnStop() {
  57. wal.BaseService.OnStop()
  58. wal.group.Stop()
  59. }
  60. // called in newStep and for each pass in receiveRoutine
  61. func (wal *WAL) Save(wmsg WALMessage) {
  62. if wal == nil {
  63. return
  64. }
  65. if wal.light {
  66. // in light mode we only write new steps, timeouts, and our own votes (no proposals, block parts)
  67. if mi, ok := wmsg.(msgInfo); ok {
  68. if mi.PeerKey != "" {
  69. return
  70. }
  71. }
  72. }
  73. // Write #HEIGHT: XYZ if new height
  74. if edrs, ok := wmsg.(types.EventDataRoundState); ok {
  75. if edrs.Step == RoundStepNewHeight.String() {
  76. wal.writeHeight(edrs.Height)
  77. }
  78. }
  79. // Write the wal message
  80. var wmsgBytes = wire.JSONBytes(TimedWALMessage{time.Now(), wmsg})
  81. err := wal.group.WriteLine(string(wmsgBytes))
  82. if err != nil {
  83. PanicQ(Fmt("Error writing msg to consensus wal. Error: %v \n\nMessage: %v", err, wmsg))
  84. }
  85. // TODO: only flush when necessary
  86. if err := wal.group.Flush(); err != nil {
  87. PanicQ(Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
  88. }
  89. }
  90. func (wal *WAL) writeHeight(height int) {
  91. wal.group.WriteLine(Fmt("#HEIGHT: %v", height))
  92. // TODO: only flush when necessary
  93. if err := wal.group.Flush(); err != nil {
  94. PanicQ(Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
  95. }
  96. }