| 205 | } |
| 206 | |
| 207 | func (c *Checkpointer) Start() { |
| 208 | // Start a time-based checkpointer goroutine |
| 209 | if c.checkpointInterval > 0 { |
| 210 | c.closeWg.Add(1) |
| 211 | go func() { |
| 212 | defer c.closeWg.Done() |
| 213 | ticker := time.NewTicker(c.checkpointInterval) |
| 214 | defer ticker.Stop() |
| 215 | for { |
| 216 | select { |
| 217 | case <-ticker.C: |
| 218 | base.TracefCtx(c.ctx, base.KeyReplicate, "calling checkpoint now. context is not cancelled here") |
| 219 | c.CheckpointNow() |
| 220 | case <-c.ctx.Done(): |
| 221 | base.DebugfCtx(c.ctx, base.KeyReplicate, "checkpointer goroutine stopped") |
| 222 | return |
| 223 | } |
| 224 | } |
| 225 | }() |
| 226 | } |
| 227 | } |
| 228 | |
| 229 | // CheckpointNow forces the checkpointer to send a checkpoint, and blocks until it has finished. |
| 230 | func (c *Checkpointer) CheckpointNow() { |