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.

112 lines
2.8 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(walDir string, light bool) (*WAL, error) {
  34. group, err := auto.OpenGroup(walDir + "/wal")
  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. wal.BaseService.OnStart()
  48. size, err := wal.group.Head.Size()
  49. if err != nil {
  50. return err
  51. } else if size == 0 {
  52. wal.writeHeight(1)
  53. }
  54. _, err = wal.group.Start()
  55. return err
  56. }
  57. func (wal *WAL) OnStop() {
  58. wal.BaseService.OnStop()
  59. wal.group.Stop()
  60. }
  61. // called in newStep and for each pass in receiveRoutine
  62. func (wal *WAL) Save(wmsg WALMessage) {
  63. if wal == nil {
  64. return
  65. }
  66. if wal.light {
  67. // in light mode we only write new steps, timeouts, and our own votes (no proposals, block parts)
  68. if mi, ok := wmsg.(msgInfo); ok {
  69. if mi.PeerKey != "" {
  70. return
  71. }
  72. }
  73. }
  74. // Write #HEIGHT: XYZ if new height
  75. if edrs, ok := wmsg.(types.EventDataRoundState); ok {
  76. if edrs.Step == RoundStepNewHeight.String() {
  77. wal.writeHeight(edrs.Height)
  78. }
  79. }
  80. // Write the wal message
  81. var wmsgBytes = wire.JSONBytes(TimedWALMessage{time.Now(), wmsg})
  82. err := wal.group.WriteLine(string(wmsgBytes))
  83. if err != nil {
  84. PanicQ(Fmt("Error writing msg to consensus wal. Error: %v \n\nMessage: %v", err, wmsg))
  85. }
  86. // TODO: only flush when necessary
  87. if err := wal.group.Flush(); err != nil {
  88. PanicQ(Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
  89. }
  90. }
  91. func (wal *WAL) writeHeight(height int) {
  92. wal.group.WriteLine(Fmt("#HEIGHT: %v", height))
  93. // TODO: only flush when necessary
  94. if err := wal.group.Flush(); err != nil {
  95. PanicQ(Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
  96. }
  97. }