[WIP] Windows

src/callstack.cr

@@ -1,4 +1,6 @@
-require "c/dlfcn"
+{% if !flag?(:windows) %}
+  require "c/dlfcn"
+{% end %}
 require "c/stdio"
 require "c/string"
 require "callstack/lib_unwind"
@@ -293,16 +295,19 @@ struct CallStack
   {% end %}
 
   protected def self.decode_frame(ip, original_ip = ip)
-    if LibC.dladdr(ip, out info) != 0
-      offset = original_ip - info.dli_saddr
+    {% if flag?(:windows) %}
+    {% else %}
+      if LibC.dladdr(ip, out info) != 0
+        offset = original_ip - info.dli_saddr
 
-      if offset == 0
-        return decode_frame(ip - 1, original_ip)
-      end
+        if offset == 0
+          return decode_frame(ip - 1, original_ip)
+        end
 
-      unless info.dli_sname.null?
-        {offset, info.dli_sname}
+        unless info.dli_sname.null?
+          {offset, info.dli_sname}
+        end
       end
-    end
+    {% end %}
   end
 end

src/concurrent.cr

@@ -1,5 +1,8 @@
 require "fiber"
-require "./concurrent/*"
+require "./concurrent/channel"
+require "./concurrent/error"
+require "./concurrent/future"
+require "./concurrent/scheduler"
 
 # Blocks the current fiber for the specified number of seconds.
 #

src/concurrent/scheduler.cr

@@ -1,89 +1,5 @@
-require "event"
-
-# :nodoc:
-class Scheduler
-  @@runnables = Deque(Fiber).new
-  @@eb = Event::Base.new
-
-  def self.reschedule
-    if runnable = @@runnables.shift?
-      runnable.resume
-    else
-      loop_fiber.resume
-    end
-    nil
-  end
-
-  def self.loop_fiber
-    @@loop_fiber ||= Fiber.new { @@eb.run_loop }
-  end
-
-  def self.after_fork
-    @@eb.reinit
-  end
-
-  def self.create_resume_event(fiber)
-    @@eb.new_event(-1, LibEvent2::EventFlags::None, fiber) do |s, flags, data|
-      data.as(Fiber).resume
-    end
-  end
-
-  def self.create_fd_write_event(io : IO::FileDescriptor, edge_triggered : Bool = false)
-    flags = LibEvent2::EventFlags::Write
-    flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered
-    event = @@eb.new_event(io.fd, flags, io) do |s, flags, data|
-      fd_io = data.as(IO::FileDescriptor)
-      if flags.includes?(LibEvent2::EventFlags::Write)
-        fd_io.resume_write
-      elsif flags.includes?(LibEvent2::EventFlags::Timeout)
-        fd_io.write_timed_out = true
-        fd_io.resume_write
-      end
-    end
-    event
-  end
-
-  def self.create_fd_read_event(io : IO::FileDescriptor, edge_triggered : Bool = false)
-    flags = LibEvent2::EventFlags::Read
-    flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered
-    event = @@eb.new_event(io.fd, flags, io) do |s, flags, data|
-      fd_io = data.as(IO::FileDescriptor)
-      if flags.includes?(LibEvent2::EventFlags::Read)
-        fd_io.resume_read
-      elsif flags.includes?(LibEvent2::EventFlags::Timeout)
-        fd_io.read_timed_out = true
-        fd_io.resume_read
-      end
-    end
-    event
-  end
-
-  def self.create_signal_event(signal : Signal, chan)
-    flags = LibEvent2::EventFlags::Signal | LibEvent2::EventFlags::Persist
-    event = @@eb.new_event(Int32.new(signal.to_i), flags, chan) do |s, flags, data|
-      ch = data.as(Channel::Buffered(Signal))
-      sig = Signal.new(s)
-      ch.send sig
-    end
-    event.add
-    event
-  end
-
-  @@dns_base : Event::DnsBase?
-
-  private def self.dns_base
-    @@dns_base ||= @@eb.new_dns_base
-  end
-
-  def self.create_dns_request(nodename, servname, hints, data, &callback : LibEvent2::DnsGetAddrinfoCallback)
-    dns_base.getaddrinfo(nodename, servname, hints, data, &callback)
-  end
-
-  def self.enqueue(fiber : Fiber)
-    @@runnables << fiber
-  end
-
-  def self.enqueue(fibers : Enumerable(Fiber))
-    @@runnables.concat fibers
-  end
-end
+{% if flag?(:windows) %}
+  require "./scheduler.windows.cr"
+{% else %}
+  require "./scheduler.posix.cr"
+{% end %}

src/concurrent/scheduler.posix.cr

@@ -0,0 +1,89 @@
+require "event"
+
+# :nodoc:
+class Scheduler
+  @@runnables = Deque(Fiber).new
+  @@eb = Event::Base.new
+
+  def self.reschedule
+    if runnable = @@runnables.shift?
+      runnable.resume
+    else
+      loop_fiber.resume
+    end
+    nil
+  end
+
+  def self.loop_fiber
+    @@loop_fiber ||= Fiber.new { @@eb.run_loop }
+  end
+
+  def self.after_fork
+    @@eb.reinit
+  end
+
+  def self.create_resume_event(fiber)
+    @@eb.new_event(-1, LibEvent2::EventFlags::None, fiber) do |s, flags, data|
+      data.as(Fiber).resume
+    end
+  end
+
+  def self.create_fd_write_event(io : IO::FileDescriptor, edge_triggered : Bool = false)
+    flags = LibEvent2::EventFlags::Write
+    flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered
+    event = @@eb.new_event(io.fd, flags, io) do |s, flags, data|
+      fd_io = data.as(IO::FileDescriptor)
+      if flags.includes?(LibEvent2::EventFlags::Write)
+        fd_io.resume_write
+      elsif flags.includes?(LibEvent2::EventFlags::Timeout)
+        fd_io.write_timed_out = true
+        fd_io.resume_write
+      end
+    end
+    event
+  end
+
+  def self.create_fd_read_event(io : IO::FileDescriptor, edge_triggered : Bool = false)
+    flags = LibEvent2::EventFlags::Read
+    flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered
+    event = @@eb.new_event(io.fd, flags, io) do |s, flags, data|
+      fd_io = data.as(IO::FileDescriptor)
+      if flags.includes?(LibEvent2::EventFlags::Read)
+        fd_io.resume_read
+      elsif flags.includes?(LibEvent2::EventFlags::Timeout)
+        fd_io.read_timed_out = true
+        fd_io.resume_read
+      end
+    end
+    event
+  end
+
+  def self.create_signal_event(signal : Signal, chan)
+    flags = LibEvent2::EventFlags::Signal | LibEvent2::EventFlags::Persist
+    event = @@eb.new_event(Int32.new(signal.to_i), flags, chan) do |s, flags, data|
+      ch = data.as(Channel::Buffered(Signal))
+      sig = Signal.new(s)
+      ch.send sig
+    end
+    event.add
+    event
+  end
+
+  @@dns_base : Event::DnsBase?
+
+  private def self.dns_base
+    @@dns_base ||= @@eb.new_dns_base
+  end
+
+  def self.create_dns_request(nodename, servname, hints, data, &callback : LibEvent2::DnsGetAddrinfoCallback)
+    dns_base.getaddrinfo(nodename, servname, hints, data, &callback)
+  end
+
+  def self.enqueue(fiber : Fiber)
+    @@runnables << fiber
+  end
+
+  def self.enqueue(fibers : Enumerable(Fiber))
+    @@runnables.concat fibers
+  end
+end

src/concurrent/scheduler.windows.cr

@@ -0,0 +1,118 @@
+require "event"
+
+# :nodoc:
+class Scheduler
+  @@runnables = Deque(Fiber).new
+  @@timers = Hash(Fiber, LibWindows::Handle).new
+
+  def self.reschedule
+    if runnable = @@runnables.shift?
+      runnable.resume
+    else
+      loop_fiber.resume
+    end
+    nil
+  end
+
+  def self.completion_port
+    @@completion_port ||= LibWindows.create_io_completion_port(LibWindows::INVALID_HANDLE_VALUE, nil, nil, 0)
+  end
+
+  def self.loop_fiber
+    @@loop_fiber ||= Fiber.new do
+      loop do
+        bytes_transfered = 0u32
+        data = Pointer(Void).null
+        entry = uninitialized LibWindows::Overlapped*
+        if LibWindows.get_queued_completion_status(Scheduler.completion_port, pointerof(bytes_transfered), pointerof(data), pointerof(entry), LibWindows::INFINITY)
+          if entry.null?
+            # It is just a fiber wanting to be resumed
+            fiber = data.as(Fiber)
+            if timer_handle = @@timers[fiber]?
+              unless LibWindows.delete_timer_queue_timer(nil, timer_handle, nil)
+                error = WinError.new "DeleteTimerQueueTimer"
+                raise error if error.code != WinError::ERROR_IO_PENDING && error.code != WinError::ERROR_SUCCESS
+              end
+              @@timers.delete(fiber)
+            end
+            fiber.resume
+          end
+        end
+      end
+    end
+  end
+
+  def self.create_resume_event(fiber)
+    unless LibWindows.post_queued_completion_status(Scheduler.completion_port, 0, fiber.as(Void*), nil)
+      raise WinError.new "PostQueueCompletionStatus"
+    end
+  end
+
+  def self.create_sleep_event(fiber, seconds)
+    LibWindows.create_timer_queue_timer(out handle, nil, ->(data, fired) {
+      # this is run inside a thread pool managed by the system
+      Scheduler.create_resume_event(data.as(Fiber))
+    }, fiber.as(Void*), (seconds*1000).to_u32, 0, 0)
+
+    @@timers[fiber] = handle
+  end
+
+  # def self.create_fd_write_event(io : IO::FileDescriptor, edge_triggered : Bool = false)
+  #   flags = LibEvent2::EventFlags::Write
+  #   flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered
+  #   event = @@eb.new_event(io.fd, flags, io) do |s, flags, data|
+  #     fd_io = data.as(IO::FileDescriptor)
+  #     if flags.includes?(LibEvent2::EventFlags::Write)
+  #       fd_io.resume_write
+  #     elsif flags.includes?(LibEvent2::EventFlags::Timeout)
+  #       fd_io.write_timed_out = true
+  #       fd_io.resume_write
+  #     end
+  #   end
+  #   event
+  # end
+
+  # def self.create_fd_read_event(io : IO::FileDescriptor, edge_triggered : Bool = false)
+  #   flags = LibEvent2::EventFlags::Read
+  #   flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered
+  #   event = @@eb.new_event(io.fd, flags, io) do |s, flags, data|
+  #     fd_io = data.as(IO::FileDescriptor)
+  #     if flags.includes?(LibEvent2::EventFlags::Read)
+  #       fd_io.resume_read
+  #     elsif flags.includes?(LibEvent2::EventFlags::Timeout)
+  #       fd_io.read_timed_out = true
+  #       fd_io.resume_read
+  #     end
+  #   end
+  #   event
+  # end
+
+  # def self.create_signal_event(signal : Signal, chan)
+  #   flags = LibEvent2::EventFlags::Signal | LibEvent2::EventFlags::Persist
+  #   event = @@eb.new_event(Int32.new(signal.to_i), flags, chan) do |s, flags, data|
+  #     ch = data.as(Channel::Buffered(Signal))
+  #     sig = Signal.new(s)
+  #     ch.send sig
+  #   end
+  #   event.add
+  #   event
+  # end
+
+  # @@dns_base : Event::DnsBase?
+
+  # private def self.dns_base
+  #   @@dns_base ||= @@eb.new_dns_base
+  # end
+
+  # def self.create_dns_request(nodename, servname, hints, data, &callback : LibEvent2::DnsGetAddrinfoCallback)
+  #   dns_base.getaddrinfo(nodename, servname, hints, data, &callback)
+  # end
+
+  def self.enqueue(fiber : Fiber)
+    @@runnables << fiber
+  end
+
+  def self.enqueue(fibers : Enumerable(Fiber))
+    @@runnables.concat fibers
+  end
+end

src/event/lib_event2.cr

@@ -1,4 +1,9 @@
-require "c/netdb"
+require "c/time"
+{% if flag?(:windows) %}
+  require "lib_windows"
+{% else %}
+  require "c/netdb"
+{% end %}
 
 {% if flag?(:linux) %}
   @[Link("rt")]
@@ -14,7 +19,7 @@ lib LibEvent2
   alias Int = LibC::Int
 
   {% if flag?(:windows) %}
-    # TODO
+    alias EvutilSocketT = LibWindows::DWord
   {% else %}
     alias EvutilSocketT = Int
   {% end %}
@@ -58,11 +63,11 @@ lib LibEvent2
 
   EVUTIL_EAI_CANCEL = -90001
 
-  alias DnsGetAddrinfoCallback = (Int32, LibC::Addrinfo*, Void*) ->
+  # alias DnsGetAddrinfoCallback = (Int32, LibC::Addrinfo*, Void*) ->
 
-  fun evdns_base_new(base : EventBase, init : Int32) : DnsBase
-  fun evdns_base_free(base : DnsBase, fail_requests : Int32)
-  fun evdns_getaddrinfo(base : DnsBase, nodename : UInt8*, servname : UInt8*, hints : LibC::Addrinfo*, cb : DnsGetAddrinfoCallback, arg : Void*) : DnsGetAddrinfoRequest
-  fun evdns_getaddrinfo_cancel(DnsGetAddrinfoRequest)
-  fun evutil_freeaddrinfo(ai : LibC::Addrinfo*)
+  # fun evdns_base_new(base : EventBase, init : Int32) : DnsBase
+  # fun evdns_base_free(base : DnsBase, fail_requests : Int32)
+  # fun evdns_getaddrinfo(base : DnsBase, nodename : UInt8*, servname : UInt8*, hints : LibC::Addrinfo*, cb : DnsGetAddrinfoCallback, arg : Void*) : DnsGetAddrinfoRequest
+  # fun evdns_getaddrinfo_cancel(DnsGetAddrinfoRequest)
+  # fun evutil_freeaddrinfo(ai : LibC::Addrinfo*)
 end

src/event/signal_child_handler.cr

@@ -1,4 +1,6 @@
-require "c/sys/wait"
+{% if !flag?(:windows) %}
+  require "c/sys/wait"
+{% end %}
 
 # :nodoc:
 # Singleton that handles SIG_CHLD and queues events for Process#waitpid.