"golang.org/x/sys/unix"
)
-// 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,
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() {
// Flush returns when every write before the call is durable.
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
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
s.running = true
s.workers.Add(1)
+
go s.run(s.written)
}
}
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{}),
- }
-}