git.giacomolli.uk - onechat/blob - syncer.go

   1 package main
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "os"
   8         "sync"
   9
  10         "golang.org/x/sys/unix"
  11 )
  12
  13 // openSyncer returns the syncer of the file system that holds dir. It syncs
  14 // with syncfs, which writes back only that file system.
  15 func openSyncer(dir string) (*Syncer, error) {
  16         file, err := os.Open(dir)
  17         if err != nil {
  18                 return nil, fmt.Errorf("open chat directory: %w", err)
  19         }
  20
  21         s := newSyncer(func() error {
  22                 return unix.Syncfs(int(file.Fd()))
  23         })
  24         s.file = file
  25
  26         return s, nil
  27 }
  28
  29 // Syncer makes the writes of a chat durable by group commit. Each write goes to
  30 // the operating system at once, so a crash of the process loses nothing. After
  31 // each write, Written starts a background sync of the file system of the chat,
  32 // unless one runs already. Then the next sync starts when the running one ends.
  33 // One sync makes every write before its start durable, so a sync covers many
  34 // writes, and there is never more than one sync at a time. A power loss loses
  35 // at most the writes after the start of the last finished sync.
  36 //
  37 // A failed sync can mean that the kernel dropped written data, so the error
  38 // stays, and each later Flush returns it.
  39 type Syncer struct {
  40         mutex   sync.Mutex
  41         sync    func() error
  42         file    *os.File
  43         written uint64
  44         durable uint64
  45         running bool
  46         done    chan struct{}
  47         failure error
  48         workers sync.WaitGroup
  49 }
  50
  51 // Written tells the syncer that a write went to the operating system, and
  52 // starts a sync when none runs.
  53 func (s *Syncer) Written() {
  54         s.mutex.Lock()
  55         defer s.mutex.Unlock()
  56
  57         s.written++
  58         s.start()
  59 }
  60
  61 // Flush returns when every write before the call is durable.
  62 func (s *Syncer) Flush(ctx context.Context) error {
  63         s.mutex.Lock()
  64
  65         target := s.written
  66         s.mutex.Unlock()
  67
  68         for {
  69                 s.mutex.Lock()
  70
  71                 reached := s.durable >= target
  72                 failure := s.failure
  73                 done := s.done
  74                 s.mutex.Unlock()
  75
  76                 if reached {
  77                         return nil
  78                 }
  79
  80                 if failure != nil {
  81                         return failure
  82                 }
  83
  84                 select {
  85                 case <-done:
  86                 case <-ctx.Done():
  87                         return ctx.Err()
  88                 }
  89         }
  90 }
  91
  92 // Close makes every write durable, waits for the syncs, and closes the
  93 // directory.
  94 func (s *Syncer) Close() error {
  95         err := s.Flush(context.Background())
  96         s.workers.Wait()
  97
  98         if s.file != nil {
  99                 err = errors.Join(err, s.file.Close())
 100                 s.file = nil
 101         }
 102
 103         return err
 104 }
 105
 106 // start starts a sync of the writes so far when none runs and a write is not
 107 // durable. The caller holds the mutex.
 108 func (s *Syncer) start() {
 109         if s.running || s.durable == s.written || s.failure != nil {
 110                 return
 111         }
 112
 113         s.running = true
 114         s.workers.Add(1)
 115
 116         go s.run(s.written)
 117 }
 118
 119 // run syncs once, so the writes up to target become durable, and starts the
 120 // next sync when writes came during this one.
 121 func (s *Syncer) run(target uint64) {
 122         defer s.workers.Done()
 123
 124         err := s.sync()
 125
 126         s.mutex.Lock()
 127         defer s.mutex.Unlock()
 128
 129         s.running = false
 130         if err != nil {
 131                 s.failure = fmt.Errorf("sync chat files: %w", err)
 132         } else {
 133                 s.durable = target
 134         }
 135
 136         close(s.done)
 137
 138         s.done = make(chan struct{})
 139         s.start()
 140 }
 141
 142 // newSyncer returns a syncer that makes the writes durable with sync.
 143 func newSyncer(sync func() error) *Syncer {
 144         return &Syncer{
 145                 sync: sync,
 146                 done: make(chan struct{}),
 147         }
 148 }