Merge pull request #3674 from rubinius/fibers

WIP Rewrite Fibers to use pthreads

Author: brixen

configure

@@ -72,7 +72,6 @@ class Configure
     @libc         = nil
     @x86_32       = false
     @x86_64       = false
-    @fibers       = false
     @dtrace       = false
     @dtrace_const = false
     @have_lchmod  = false
@@ -458,7 +457,6 @@ class Configure
     feature "vendor-zlib", false
     feature "vendor-libsodium", true
     feature "alloc-tracking", false
-    feature "fibers", true
     feature "dtrace", false
     feature "rpath", true
 
@@ -1198,10 +1196,6 @@ int main() { return tgetnum(""); }
       @defines << "RBX_ALLOC_TRACKING"
     end
 
-    if @features["fibers"].value
-      @fibers = true if @x86_32 or @x86_64
-    end
-
     if @features["dtrace"].value and has_dtrace
       @defines << "HAVE_DTRACE"
     end
@@ -1583,7 +1577,6 @@ int main() { return tgetnum(""); }
       :x86_64             => @x86_64,
       :dtrace             => @dtrace,
       :dtrace_const       => @dtrace_const,
-      :fibers             => @fibers,
       :debug_build        => @debug_build,
       :sourcedir          => @sourcedir,
       :stagingdir         => @stagingdir,
@@ -1690,10 +1683,6 @@ int main() { return tgetnum(""); }
         end
       end
 
-      if @fibers
-        f.puts "#define RBX_FIBER_ENABLED    1"
-      end
-
       f.puts "#define RBX_DTRACE_CONST     #{@dtrace_const ? "const" : ""}"
 
       write_have_defines f

core/enumerator.rb

@@ -98,19 +98,7 @@ def each_with_index
     def next
       return @lookahead.shift unless @lookahead.empty?
 
-      unless @generator
-        # Allow #to_generator to return nil, indicating it has none for
-        # this method.
-        if @object.respond_to? :to_generator
-          @generator = @object.to_generator(@iter)
-        end
-
-        if !@generator and gen = FiberGenerator
-          @generator = gen.new(self)
-        else
-          @generator = ThreadGenerator.new(self, @object, @iter, @args)
-        end
-      end
+      @generator ||= Iterator.new self
 
       begin
         return @generator.next if @generator.next?
@@ -198,6 +186,7 @@ def <<(*args)
 
     class Generator
       include Enumerable
+
       def initialize(&block)
         raise LocalJumpError, "Expected a block to be given" unless block_given?
 
@@ -447,123 +436,41 @@ def zip(*lists)
       end
     end
 
-    if Rubinius::Fiber::ENABLED
-      class FiberGenerator
-        STACK_SIZE = 1_048_576
-
-        attr_reader :result
+    class Iterator
+      STACK_SIZE = 1_048_576
 
-        def initialize(obj)
-          @object = obj
-          rewind
-        end
-
-        def next?
-          !@done
-        end
-
-        def next
-          reset unless @fiber
-
-          val = @fiber.resume
-
-          raise StopIteration, "iteration has ended" if @done
-
-          return val
-        end
-
-        def rewind
-          @fiber = nil
-          @done = false
-        end
-
-        def reset
-          @done = false
-          @fiber = Rubinius::Fiber.new stack_size: STACK_SIZE do
-            obj = @object
-            @result = obj.each do |*val|
-              Rubinius::Fiber.yield *val
-            end
-            @done = true
-          end
-        end
-      end
-    else
-      FiberGenerator = nil
-    end
-
-    class ThreadGenerator
       attr_reader :result
 
-      def initialize(enum, obj, meth, args)
+      def initialize(obj)
         @object = obj
-        @method = meth
-        @args = args
-
-        ObjectSpace.define_finalizer(enum, method(:kill))
-
         rewind
       end
 
-      # Used to cleanup the background thread when the enumerator
-      # is GC'd.
-      def kill(obj_id)
-        if @thread
-          @thread.kill
-        end
-      end
-
       def next?
-        if @done
-          @thread.join if @thread
-          @thread = nil
-          return false
-        end
-
-        true
+        !@done
       end
 
       def next
-        reset unless @thread
+        reset unless @fiber
 
-        @hold_channel << nil
-        vals = @channel.receive
+        val = @fiber.resume
 
         raise StopIteration, "iteration has ended" if @done
 
-        # return *[1] == [1], unfortunately.
-        return vals.size == 1 ? vals.first : vals
+        return val
       end
 
       def rewind
-        if @thread
-          @thread.kill
-        end
-
+        @fiber = nil
         @done = false
-        @thread = nil
       end
 
       def reset
         @done = false
-        @channel = Rubinius::Channel.new
-        @hold_channel = Rubinius::Channel.new
-
-        @thread = Thread.new do
-          @result = @object.__send__(@method, *@args) do |*vals|
-            @hold_channel.receive
-            @channel << vals
-          end
-
-          # Hold to indicate done to avoid race conditions setting
-          # the ivar.
-          @hold_channel.receive
+        @fiber = Fiber.new stack_size: STACK_SIZE do
+          obj = @object
+          @result = obj.each { |*val| Fiber.yield *val }
           @done = true
-
-          # Put an extra value in the channel, so that main
-          # thread doesn't accidentally block if it doesn't
-          # detect @done in time.
-          @channel << nil
         end
       end
     end

core/fiber.rb

@@ -1,48 +1,63 @@
-module Rubinius
-  class Fiber
-    attr_reader :stack_size
-    attr_reader :thread_name
-    attr_reader :fiber_id
-    attr_reader :source
+class Fiber
+  attr_reader :stack_size
+  attr_reader :thread_name
+  attr_reader :fiber_id
+  attr_reader :source
+
+  def self.new(**kw, &block)
+    if block.nil?
+      raise ArgumentError, "Fiber.new requires a block"
+    end
 
-    def self.new(**kw, &block)
-      if block.nil?
-        raise ArgumentError, "Fiber.new requires a block"
-      end
+    stack_size = Rubinius::Type.try_convert kw[:stack_size], Fixnum, :to_int
 
-      stack_size = Rubinius::Type.try_convert kw[:stack_size], Fixnum, :to_int
+    Rubinius.invoke_primitive :fiber_new, stack_size, block, self
+  end
 
-      Rubinius.invoke_primitive :fiber_new, stack_size, block, self
-    end
+  def self.current
+    Rubinius.primitive :fiber_s_current
+    raise PrimitiveFailure, "Fiber.current primitive failed"
+  end
 
-    def self.current
-      Rubinius.primitive :fiber_s_current
-      raise PrimitiveFailure, "Rubinius::Fiber.current failed"
-    end
+  def self.yield(*args)
+    Rubinius.primitive :fiber_s_yield
+    raise PrimitiveFailure, "Fiber.yield primitive failed"
+  end
 
-    def self.yield(*args)
-      Rubinius.primitive :fiber_s_yield
-      raise PrimitiveFailure, "Rubinius::Fiber.yield failed"
-    end
+  def self.list
+    Rubinius.primitive :fiber_s_list
+    raise PrimitiveFailure, "Fiber.list primitive failed"
+  end
 
-    def resume(*args)
-      Rubinius.primitive :fiber_resume
-      raise PrimitiveFailure, "Rubinius::Fiber#resume failed"
-    end
+  def self.main
+    Rubinius.primitive :fiber_s_main
+    raise PrimitiveFailure, "Fiber.main primitive failed"
+  end
 
-    def transfer(*args)
-      Rubinius.primitive :fiber_transfer
-      raise PrimitiveFailure, "Rubinius::Fiber#transfer failed"
-    end
+  def status
+    Rubinius.primitive :fiber_status
+    raise PrimitiveFailure, "Fiber#status primitive failed"
+  end
 
-    def alive?
-      !@dead
-    end
+  def resume(*args)
+    Rubinius.primitive :fiber_resume
+    raise PrimitiveFailure, "Fiber#resume primitive failed"
+  end
 
-    def inspect
-      str = "#<#{self.class}:0x#{object_id.to_s(16)} thread_name=#{@thread_name} fiber_id=#{@fiber_id} status=#{alive? ? "alive" : "dead"}"
-      str << " source=#{@source}" if @source
-      str << ">"
-    end
+  def transfer(*args)
+    Rubinius.primitive :fiber_transfer
+    raise PrimitiveFailure, "Fiber#transfer primitive failed"
   end
+
+  def alive?
+    status != "dead"
+  end
+
+  def inspect
+    str = "#<#{self.class}:0x#{object_id.to_s(16)} thread_name=#{@thread_name} fiber_id=#{@fiber_id} status=#{status}"
+    str << " source=#{@source}" if @source
+    str << ">"
+  end
+
+  alias_method :to_s, :inspect
 end

core/thread.rb

@@ -84,6 +84,11 @@ def self.stop
     nil
   end
 
+  def fiber_list
+    Rubinius.primitive :thread_fiber_list
+    Kernel.raise PrimitiveFailure, "Thread.fiber_list primitive failed"
+  end
+
   def wakeup
     Rubinius.primitive :thread_wakeup
     Kernel.raise ThreadError, "Thread#wakeup primitive failed, thread may be dead"
@@ -224,40 +229,45 @@ def raise(exc=undefined, msg=nil, trace=nil)
     end
   end
 
-  def [](key)
-    locals_aref(Rubinius::Type.coerce_to_symbol(key))
+  def thread_variable_get(key)
+    key = Rubinius::Type.coerce_to_symbol key
+    Rubinius.invoke_primitive :thread_variable_get, self, key
   end
 
-  def locals_aref(key)
-    Rubinius.primitive :thread_locals_aref
-    raise PrimitiveFailure, "Thread#locals_aref primitive failed"
+  def thread_variable_set(key, value)
+    key = Rubinius::Type.coerce_to_symbol key
+    Rubinius.invoke_primitive :thread_variable_set, self, key, value
   end
-  private :locals_aref
 
-  def []=(key, value)
-    locals_store(Rubinius::Type.coerce_to_symbol(key), value)
+  def thread_variable?(key)
+    key = Rubinius::Type.coerce_to_symbol key
+    Rubinius.invoke_primitive :thread_variable_key_p, self, key
   end
 
-  def locals_store(key, value)
-    Rubinius.primitive :thread_locals_store
-    raise PrimitiveFailure, "Thread#locals_store primitive failed"
+  def thread_variables
+    Rubinius.primitive :thread_variables
+    raise PrimitiveFailure, "Thread#thread_variables primitive failed"
   end
-  private :locals_store
 
-  def keys
-    Rubinius.primitive :thread_locals_keys
-    raise PrimitiveFailure, "Thread#keys primitive failed"
+  def [](key)
+    key = Rubinius::Type.coerce_to_symbol key
+    Rubinius.invoke_primitive :thread_fiber_variable_get, self, key
+  end
+
+  def []=(key, value)
+    key = Rubinius::Type.coerce_to_symbol key
+    Rubinius.invoke_primitive :thread_fiber_variable_set, self, key, value
   end
 
   def key?(key)
-    locals_key?(Rubinius::Type.coerce_to_symbol(key))
+    key = Rubinius::Type.coerce_to_symbol key
+    Rubinius.invoke_primitive :thread_fiber_variable_key_p, self, key
   end
 
-  def locals_key?(key)
-    Rubinius.primitive :thread_locals_has_key
-    raise PrimitiveFailure, "Thread#locals_key? primitive failed"
+  def keys
+    Rubinius.primitive :thread_fiber_variables
+    raise PrimitiveFailure, "Thread#keys primitive failed"
   end
-  private :locals_key?
 
   # Register another Thread object +thr+ as the Thread where the debugger
   # is running. When the current thread hits a breakpoint, it uses this

core/zed.rb

@@ -1062,8 +1062,6 @@ module Errno
   FFI = Rubinius::FFI
 end
 
-Fiber = Rubinius::Fiber
-
 class File < IO
   # these will be necessary when we run on Windows
   DOSISH = false # !!(RUBY_PLATFORM =~ /mswin/)