Integrate Crystal::FdLock into IO::FileDescriptor, File and Socket

Only operations that can affect the file descriptor are counted, for
example read or write, truncating a file or changing file permissions.

Mere queries with no side effects go through normally because at worst
they will fail (they would have anyway).

Author: ysbaddaden

spec/std/crystal/fd_lock_spec.cr

@@ -1,5 +1,4 @@
 require "spec"
-require "../../support/interpreted.cr"
 require "crystal/fd_lock"
 require "wait_group"
 
@@ -92,8 +91,7 @@ describe Crystal::FdLock do
       called.should eq(1)
     end
 
-    # FIXME: the interpreter segfaults while running this spec (NULL pointer)
-    pending_interpreted "waits for all references to return" do
+    it "waits for all references to return" do
       lock = Crystal::FdLock.new
 
       ready = WaitGroup.new(10)

src/crystal/fd_lock.cr

@@ -1,5 +1,118 @@
-{% if flag?(:preview_mt) %}
-  require "./fd_lock_mt"
-{% else %}
-  require "./fd_lock_no_mt"
-{% end %}
+# The general design is influenced by fdMutex in Go (LICENSE: BSD 3-Clause,
+# Copyright Google):
+# https://github.com/golang/go/blob/go1.25.1/src/internal/poll/fd_mutex.go
+
+# :nodoc:
+#
+# Tracks active references over a system file descriptor (fd) and serializes
+# reads and writes.
+#
+# Every access to the fd that may affect its system state or system buffers must
+# acquire a shared lock.
+#
+# The fdlock can be closed at any time, but the actual system close will wait
+# until there are no more references left. This avoids potential races when a
+# thread might try to read a fd that has been closed and has been reused by the
+# OS for example.
+struct Crystal::FdLock
+  CLOSED = 1_u32 << 0 # the fdlock has been closed
+  REF    = 1_u32 << 1 # the reference counter increment
+  MASK   = ~(REF - 1) # mask for the reference counter
+
+  @m = Atomic(UInt32).new(0_u32)
+  @closing : Fiber?
+
+  # Borrows a reference for the duration of the block. Raises if the fdlock is
+  # closed while trying to borrow.
+  def reference(& : -> F) : F forall F
+    m, success = @m.compare_and_set(0_u32, REF, :acquire, :relaxed)
+    increment_slow(m) unless success
+
+    begin
+      yield
+    ensure
+      m = @m.sub(REF, :release)
+      handle_last_ref(m)
+    end
+  end
+
+  private def increment_slow(m)
+    while true
+      if (m & CLOSED) == CLOSED
+        raise IO::Error.new("Closed")
+      end
+      m, success = @m.compare_and_set(m, m + REF, :acquire, :relaxed)
+      break if success
+    end
+  end
+
+  private def handle_last_ref(m)
+    return unless (m & CLOSED) == CLOSED # is closed?
+    return unless (m & MASK) == REF      # was the last ref?
+
+    # the last ref after close is responsible to resume the closing fiber
+    if fiber = @closing
+      fiber.enqueue
+    else
+      raise NilAssertionError.new("BUG: expected a closing fiber to resume.")
+    end
+  end
+
+  # Closes the fdlock. Blocks for as long as there are references.
+  #
+  # The *callback* block must cancel any external waiters (e.g. pending evloop
+  # reads or writes).
+  #
+  # Returns true if the fdlock has been closed: no fiber can acquire a reference
+  # anymore, the calling fiber fully owns the fd and can safely close it.
+  #
+  # Returns false if the fdlock has already been closed: the calling fiber
+  # doesn't own the fd and musn't close it, as there might still be active
+  # references and another fiber will close anyway.
+  def try_close?(&callback : ->) : Bool
+    attempts = 0
+
+    while true
+      m = @m.get(:relaxed)
+
+      if (m & CLOSED) == CLOSED
+        # already closed: abort
+        return false
+      end
+
+      # close + increment ref
+      m, success = @m.compare_and_set(m, (m + REF) | CLOSED, :acquire, :relaxed)
+      break if success
+
+      attempts = Thread.delay(attempts)
+    end
+
+    # set the current fiber as the closing fiber (to be resumed by the last ref)
+    @closing = Fiber.current
+
+    # decrement the last ref
+    m = @m.sub(REF, :release)
+
+    begin
+      yield
+    ensure
+      # wait for the last ref... unless we're the last ref!
+      Fiber.suspend unless (m & MASK) == REF
+    end
+
+    @closing = nil
+    true
+  end
+
+  # Resets the fdlock back to its pristine state so it can be used again.
+  # Assumes the caller owns the fdlock. This is required by
+  # `TCPSocket#initialize`.
+  def reset : Nil
+    @m.lazy_set(0_u32)
+    @closing = nil
+  end
+
+  def closed? : Bool
+    (@m.get(:relaxed) & CLOSED) == CLOSED
+  end
+end

src/crystal/fd_lock_mt.cr

@@ -109,7 +109,7 @@ module Crystal::System::File
   end
 
   private def system_chown(uid : Int, gid : Int)
-    ret = LibC.fchown(fd, uid, gid)
+    ret = @fd_lock.reference { LibC.fchown(fd, uid, gid) }
     raise ::File::Error.from_errno("Error changing owner", file: path) if ret == -1
   end
 
@@ -120,7 +120,7 @@ module Crystal::System::File
   end
 
   private def system_chmod(mode)
-    if LibC.fchmod(fd, mode) == -1
+    if @fd_lock.reference { LibC.fchmod(fd, mode) } == -1
       raise ::File::Error.from_errno("Error changing permissions", file: path)
     end
   end
@@ -201,12 +201,12 @@ module Crystal::System::File
             timespecs = uninitialized LibC::Timespec[2]
             timespecs[0] = Crystal::System::Time.to_timespec(atime)
             timespecs[1] = Crystal::System::Time.to_timespec(mtime)
-            LibC.futimens(fd, timespecs)
+            @fd_lock.reference { LibC.futimens(fd, timespecs) }
           {% elsif LibC.has_method?("futimes") %}
             timevals = uninitialized LibC::Timeval[2]
             timevals[0] = Crystal::System::Time.to_timeval(atime)
             timevals[1] = Crystal::System::Time.to_timeval(mtime)
-            LibC.futimes(fd, timevals)
+            @fd_lock.reference { LibC.futimes(fd, timevals) }
           {% else %}
             {% raise "Missing futimens & futimes" %}
           {% end %}
@@ -218,7 +218,7 @@ module Crystal::System::File
 
   private def system_truncate(size) : Nil
     flush
-    code = LibC.ftruncate(fd, size)
+    code = @fd_lock.reference { LibC.ftruncate(fd, size) }
     if code != 0
       raise ::File::Error.from_errno("Error truncating file", file: path)
     end