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