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.

104 lines
2.6 KiB

8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
  1. package consensus
  2. import (
  3. "time"
  4. wire "github.com/tendermint/go-wire"
  5. "github.com/tendermint/tendermint/types"
  6. auto "github.com/tendermint/tmlibs/autofile"
  7. cmn "github.com/tendermint/tmlibs/common"
  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. cmn.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 = *cmn.NewBaseService(nil, "WAL", wal)
  43. return wal, nil
  44. }
  45. func (wal *WAL) OnStart() error {
  46. size, err := wal.group.Head.Size()
  47. if err != nil {
  48. return err
  49. } else if size == 0 {
  50. wal.writeEndHeight(0)
  51. }
  52. _, err = wal.group.Start()
  53. return err
  54. }
  55. func (wal *WAL) OnStop() {
  56. wal.BaseService.OnStop()
  57. wal.group.Stop()
  58. }
  59. // called in newStep and for each pass in receiveRoutine
  60. func (wal *WAL) Save(wmsg WALMessage) {
  61. if wal == nil {
  62. return
  63. }
  64. if wal.light {
  65. // in light mode we only write new steps, timeouts, and our own votes (no proposals, block parts)
  66. if mi, ok := wmsg.(msgInfo); ok {
  67. if mi.PeerKey != "" {
  68. return
  69. }
  70. }
  71. }
  72. // Write the wal message
  73. var wmsgBytes = wire.JSONBytes(TimedWALMessage{time.Now(), wmsg})
  74. err := wal.group.WriteLine(string(wmsgBytes))
  75. if err != nil {
  76. cmn.PanicQ(cmn.Fmt("Error writing msg to consensus wal. Error: %v \n\nMessage: %v", err, wmsg))
  77. }
  78. // TODO: only flush when necessary
  79. if err := wal.group.Flush(); err != nil {
  80. cmn.PanicQ(cmn.Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
  81. }
  82. }
  83. func (wal *WAL) writeEndHeight(height int) {
  84. wal.group.WriteLine(cmn.Fmt("#ENDHEIGHT: %v", height))
  85. // TODO: only flush when necessary
  86. if err := wal.group.Flush(); err != nil {
  87. cmn.PanicQ(cmn.Fmt("Error flushing consensus wal buf to file. Error: %v \n", err))
  88. }
  89. }