Open file with blocking: nil or false in another thread

We can't reliably detect without significant overhead whether the file at
*path* might block for a while before calling open; while O_NONBLOCK has
no effect on regular disk files, special file types are a different story.

Open with O_NONBLOCK will fail with ENXIO for O_WRONLY (no connected
reader) but it will always succeed for O_RDONLY (regardless of a connected
writer or not), then any attempt to read will return EOF, leaving no means
to wait until a writer connects.

We thus rely on the *blocking* arg: when false the file might be a special
file type, so we check it; if it's a fifo (named pipe) or a character
device, we open in another thread so we don't risk blocking the current
thread (and thus other fibers) until a reader or writer is also connected.

We need preview_mt to safely re-enqueue the current fiber from the thread.

Author: ysbaddaden

spec/std/file_spec.cr

@@ -94,19 +94,25 @@ describe "File" do
           ret = LibC.mkfifo(path, File::DEFAULT_CREATE_PERMISSIONS)
           raise RuntimeError.from_errno("mkfifo") unless ret == 0
 
-          # FIXME: open(2) will block when opening a fifo file until another
-          #        thread or process also opened the file; we should pass
-          #        O_NONBLOCK to the open(2) call itself, not afterwards
-          file = nil
-          new_thread { file = File.new(path, "w", blocking: nil) }
+          # open(2) blocks when opening a pipe/fifo or chardev file until
+          # another thread or process also opened the file in the opposite mode
+          w = nil
+          {% if flag?(:preview_mt) %}
+            # the evloop handles opening the fifo file without blocking the
+            # current thread (only the current fiber)
+            spawn { w = File.open(path, "w", blocking: nil) }
+          {% else %}
+            # we must explicitly spawn a thread
+            new_thread { w = File.new(path, "w") }
+          {% end %}
 
           begin
-            File.open(path, "r", blocking: false) do |file|
-              file.blocking.should be_false
+            File.open(path, "r", blocking: false) do |r|
+              r.blocking.should be_false
             end
           ensure
             File.delete(path)
-            file.try(&.close)
+            w.try(&.close)
           end
         end
       {% end %}

src/crystal/event_loop/libevent.cr

@@ -111,10 +111,37 @@ class Crystal::EventLoop::LibEvent < Crystal::EventLoop
   def open(path : String, flags : Int32, permissions : File::Permissions, blocking : Bool?) : {System::FileDescriptor::Handle, Bool} | Errno
     path.check_no_null_byte
 
-    fd = LibC.open(path, flags | LibC::O_CLOEXEC, permissions)
-    return Errno.value if fd == -1
+    fd = 0
+    flags |= LibC::O_CLOEXEC
+    blocking = !System::File.special_type?(path) if blocking.nil?
+
+    # We can't reliably detect without significant overhead whether the file at
+    # *path* might block for a while before calling open; while O_NONBLOCK has
+    # no effect on regular disk files, special file types are a different story.
+    # Open with O_NONBLOCK will fail with ENXIO for O_WRONLY (no connected
+    # reader) but it will always succeed for O_RDONLY (regardless of a connected
+    # writer or not), then any attempt to read will return EOF, leaving no means
+    # to wait until a writer connects.
+    #
+    # We thus rely on the *blocking* arg: when false the file might be a special
+    # file type, so we check it; if it's a fifo (named pipe) or a character
+    # device, we open in another thread so we don't risk blocking the current
+    # thread (and thus other fibers) until a reader or writer is also connected.
+    #
+    # We need preview_mt to safely re-enqueue the current fiber from the thread.
+    {% if flag?(:preview_mt) && !flag?(:interpreted) %}
+      if blocking
+        fd = LibC.open(path, flags, permissions)
+        return Errno.value if fd == -1
+      else
+        fd, errno = System::File.async_open(path, flags, permissions)
+        return errno if fd == -1
+      end
+    {% else %}
+      fd = LibC.open(path, flags, permissions)
+      return Errno.value if fd == -1
+    {% end %}
 
-    blocking = !System::File.special_type?(fd) if blocking.nil?
     unless blocking
       status_flags = System::FileDescriptor.fcntl(fd, LibC::F_GETFL)
       System::FileDescriptor.fcntl(fd, LibC::F_SETFL, status_flags | LibC::O_NONBLOCK)

src/crystal/event_loop/polling.cr

@@ -157,10 +157,37 @@ abstract class Crystal::EventLoop::Polling < Crystal::EventLoop
   def open(path : String, flags : Int32, permissions : File::Permissions, blocking : Bool?) : {System::FileDescriptor::Handle, Bool} | Errno
     path.check_no_null_byte
 
-    fd = LibC.open(path, flags | LibC::O_CLOEXEC, permissions)
-    return Errno.value if fd == -1
+    fd = 0
+    flags |= LibC::O_CLOEXEC
+    blocking = !System::File.special_type?(path) if blocking.nil?
+
+    # We can't reliably detect without significant overhead whether the file at
+    # *path* might block for a while before calling open; while O_NONBLOCK has
+    # no effect on regular disk files, special file types are a different story.
+    # Open with O_NONBLOCK will fail with ENXIO for O_WRONLY (no connected
+    # reader) but it will always succeed for O_RDONLY (regardless of a connected
+    # writer or not), then any attempt to read will return EOF, leaving no means
+    # to wait until a writer connects.
+    #
+    # We thus rely on the *blocking* arg: when false the file might be a special
+    # file type, so we check it; if it's a fifo (named pipe) or a character
+    # device, we open in another thread so we don't risk blocking the current
+    # thread (and thus other fibers) until a reader or writer is also connected.
+    #
+    # We need preview_mt to safely re-enqueue the current fiber from the thread.
+    {% if flag?(:preview_mt) && !flag?(:interpreted) %}
+      if blocking
+        fd = LibC.open(path, flags, permissions)
+        return Errno.value if fd == -1
+      else
+        fd, errno = System::File.async_open(path, flags, permissions)
+        return errno if fd == -1
+      end
+    {% else %}
+      fd = LibC.open(path, flags, permissions)
+      return Errno.value if fd == -1
+    {% end %}
 
-    blocking = !System::File.special_type?(fd) if blocking.nil?
     unless blocking
       status_flags = System::FileDescriptor.fcntl(fd, LibC::F_GETFL)
       System::FileDescriptor.fcntl(fd, LibC::F_SETFL, status_flags | LibC::O_NONBLOCK)

src/crystal/system/unix/file.cr

@@ -14,12 +14,37 @@ module Crystal::System::File
     end
   end
 
+  {% if flag?(:preview_mt) && !flag?(:interpreted) %}
+    protected def self.async_open(filename, flags, permissions)
+      fd = -1
+      errno = Errno::NONE
+      fiber = ::Fiber.current
+
+      ::Thread.new do
+        fd = LibC.open(filename, flags, permissions)
+        errno = Errno.value if fd == -1
+
+        {% if flag?(:execution_context) %}
+          fiber.enqueue
+        {% else %}
+          # HACK: avoid Fiber#enqueue because it would lazily create a scheduler
+          # for the thread just to send the fiber to another scheduler
+          fiber.get_current_thread.not_nil!.scheduler.send_fiber(fiber)
+        {% end %}
+      end
+      ::Fiber.suspend
+
+      {fd, errno}
+    end
+  {% end %}
+
   protected def system_init(mode : String, blocking : Bool) : Nil
   end
 
-  def self.special_type?(fd)
+  def self.special_type?(path : String) : Bool
+    path.check_no_null_byte
     stat = uninitialized LibC::Stat
-    ret = fstat(fd, pointerof(stat))
+    ret = stat(path, pointerof(stat))
     ret != -1 && (stat.st_mode & LibC::S_IFMT).in?(LibC::S_IFCHR, LibC::S_IFIFO)
   end