Store shared private images by content hash

[onechat] / syncer.go

diff --git a/syncer.go b/syncer.go
index 6b69f860c59c18cb615d40b0ed37e53f39b299b2..3da6d7a6dec0daa7ad2d6a9ae4e004445f694e91 100644 (file)
--- a/syncer.go
+++ b/syncer.go
@@ -10,17 +10,33 @@ import (
        "golang.org/x/sys/unix"
 )
 
        "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
 // 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.
 //
 // 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
        mutex   sync.Mutex
        sync    func() error
        file    *os.File


@@ -32,33 +48,9 @@ type syncer struct {
        workers sync.WaitGroup
 }
 
        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.
 // 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.mutex.Lock()
        defer s.mutex.Unlock()
 


@@ -66,48 +58,16 @@ func (s *syncer) Written() {
        s.start()
 }
 
        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.
 // 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()
        s.mutex.Lock()

+

        target := s.written
        s.mutex.Unlock()
 
        for {
                s.mutex.Lock()
        target := s.written
        s.mutex.Unlock()
 
        for {
                s.mutex.Lock()

+

                reached := s.durable >= target
                failure := s.failure
                done := s.done
                reached := s.durable >= target
                failure := s.failure
                done := s.done


@@ -131,9 +91,10 @@ func (s *syncer) Flush(ctx context.Context) error {
 
 // Close makes every write durable, waits for the syncs, and closes the
 // directory.
 
 // 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()
        err := s.Flush(context.Background())
        s.workers.Wait()

+

        if s.file != nil {
                err = errors.Join(err, s.file.Close())
                s.file = nil
        if s.file != nil {
                err = errors.Join(err, s.file.Close())
                s.file = nil


@@ -141,3 +102,47 @@ func (s *syncer) Close() error {
 
        return err
 }
 
        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{}),
+       }
+}