"golang.org/x/sys/unix"
)
-// syncer makes the writes of a chat durable by group commit. Each write goes to
+// openSyncer returns the syncer of the file system that holds dir. It syncs
+// with syncfs, which writes back only that file system.
+func openSyncer(dir string) (*Syncer, error) {
+ file, err := os.Open(dir)
+ if err != nil {
+ return nil, fmt.Errorf("open chat directory: %w", err)
+ }
+
+ s := newSyncer(func() error {
+ return unix.Syncfs(int(file.Fd()))
+ })
+ s.file = file
+
+ return s, nil
+}
+
+// Syncer makes the writes of a chat durable by group commit. Each write goes to
// the operating system at once, so a crash of the process loses nothing. After
// each write, Written starts a background sync of the file system of the chat,
// unless one runs already. Then the next sync starts when the running one ends.
// One sync makes every write before its start durable, so a sync covers many
-// writes, and there is never more than one sync at a time. A power loss loses at
-// most the writes after the start of the last finished sync.
+// writes, and there is never more than one sync at a time. A power loss loses
+// at most the writes after the start of the last finished sync.
//
// A failed sync can mean that the kernel dropped written data, so the error
// stays, and each later Flush returns it.
-type syncer struct {
+type Syncer struct {
mutex sync.Mutex
sync func() error
file *os.File
workers sync.WaitGroup
}
-// openSyncer returns the syncer of the file system that holds dir. It syncs
-// with syncfs, which writes back only that file system.
-func openSyncer(dir string) (*syncer, error) {
- file, err := os.Open(dir)
- if err != nil {
- return nil, fmt.Errorf("open chat directory: %w", err)
- }
-
- s := newSyncer(func() error {
- return unix.Syncfs(int(file.Fd()))
- })
- s.file = file
-
- return s, nil
-}
-
-// newSyncer returns a syncer that makes the writes durable with sync.
-func newSyncer(sync func() error) *syncer {
- return &syncer{
- sync: sync,
- done: make(chan struct{}),
- }
-}
-
// Written tells the syncer that a write went to the operating system, and
// starts a sync when none runs.
-func (s *syncer) Written() {
+func (s *Syncer) Written() {
s.mutex.Lock()
defer s.mutex.Unlock()
s.start()
}
-// start starts a sync of the writes so far when none runs and a write is not
-// durable. The caller holds the mutex.
-func (s *syncer) start() {
- if s.running || s.durable == s.written || s.failure != nil {
- return
- }
-
- s.running = true
- s.workers.Add(1)
- go s.run(s.written)
-}
-
-// run syncs once, so the writes up to target become durable, and starts the
-// next sync when writes came during this one.
-func (s *syncer) run(target uint64) {
- defer s.workers.Done()
-
- err := s.sync()
-
- s.mutex.Lock()
- defer s.mutex.Unlock()
-
- s.running = false
- if err != nil {
- s.failure = fmt.Errorf("sync chat files: %w", err)
- } else {
- s.durable = target
- }
-
- close(s.done)
- s.done = make(chan struct{})
- s.start()
-}
-
// Flush returns when every write before the call is durable.
-func (s *syncer) Flush(ctx context.Context) error {
+func (s *Syncer) Flush(ctx context.Context) error {
s.mutex.Lock()
+
target := s.written
s.mutex.Unlock()
for {
s.mutex.Lock()
+
reached := s.durable >= target
failure := s.failure
done := s.done
// Close makes every write durable, waits for the syncs, and closes the
// directory.
-func (s *syncer) Close() error {
+func (s *Syncer) Close() error {
err := s.Flush(context.Background())
s.workers.Wait()
+
if s.file != nil {
err = errors.Join(err, s.file.Close())
s.file = nil
return err
}
+
+// start starts a sync of the writes so far when none runs and a write is not
+// durable. The caller holds the mutex.
+func (s *Syncer) start() {
+ if s.running || s.durable == s.written || s.failure != nil {
+ return
+ }
+
+ s.running = true
+ s.workers.Add(1)
+
+ go s.run(s.written)
+}
+
+// run syncs once, so the writes up to target become durable, and starts the
+// next sync when writes came during this one.
+func (s *Syncer) run(target uint64) {
+ defer s.workers.Done()
+
+ err := s.sync()
+
+ s.mutex.Lock()
+ defer s.mutex.Unlock()
+
+ s.running = false
+ if err != nil {
+ s.failure = fmt.Errorf("sync chat files: %w", err)
+ } else {
+ s.durable = target
+ }
+
+ close(s.done)
+
+ s.done = make(chan struct{})
+ s.start()
+}
+
+// newSyncer returns a syncer that makes the writes durable with sync.
+func newSyncer(sync func() error) *Syncer {
+ return &Syncer{
+ sync: sync,
+ done: make(chan struct{}),
+ }
+}