Add Parallel#resize(maximum)

Allows to grow and shrink a Parallel context. Takes care to replace
`@schedulers` instead of mutating it, and to modify #steal to read it
once, then always refer to the same array (never mutated) so we can
safely steal without acquiring the mutex.

Author: ysbaddaden

src/fiber/execution_context/parallel.cr

@@ -164,6 +164,61 @@ module Fiber::ExecutionContext
       end
     end
 
+    # Resizes the context to the new *maximum* parallelism.
+    #
+    # The new *maximum* can grow, in which case more schedulers are started to
+    # eventually increase the parallelism.
+    #
+    # The new *maximum* can also shrink, in which case the overflow schedulers
+    # are removed and told to shutdown immediately. The actual shutdown is
+    # cooperative, so running schedulers won't stop until their current fiber
+    # tries to switch to another fiber.
+    def resize(maximum : Int32) : Nil
+      maximum = maximum.clamp(1..) # FIXME: raise if maximum < 1
+      removed_schedulers = nil
+
+      @mutex.synchronize do
+        # can run in parallel to #steal that dereferences @schedulers (once)
+        # without locking the mutex, so we dup the schedulers, mutate the copy,
+        # and eventually assign the copy as @schedulers; this way #steal should
+        # only ever saw a complete and valid array.
+        new_schedulers = nil
+        new_capacity = maximum
+        old_threads = @threads
+        old_schedulers = @schedulers
+        old_capacity = capacity
+
+        if new_capacity > old_capacity
+          @schedulers = Array(Scheduler).new(new_capacity) do |index|
+            old_schedulers[index]? || Scheduler.new(self, "#{@name}-#{index}")
+          end
+          threads = Array(Thread).new(new_capacity)
+          old_threads.each { |thread| threads << thread }
+          @threads = threads
+        elsif new_capacity < old_capacity
+          # tell the overflow schedulers to shutdown
+          removed_schedulers = old_schedulers[new_capacity..]
+          removed_schedulers.each(&.shutdown(:now))
+
+          # resize
+          @schedulers = old_schedulers[0...new_capacity]
+          @threads = old_threads[0...new_capacity]
+
+          # wakeup all waiting schedulers so they can shutdown
+          @condition.broadcast
+          @event_loop.interrupt
+        end
+      end
+
+      return unless removed_schedulers
+
+      # drain the local queues of removed schedulers because they're no longer
+      # available for stealing
+      removed_schedulers.each do |scheduler|
+        scheduler.@runnables.drain
+      end
+    end
+
     # :nodoc:
     def spawn(*, name : String? = nil, same_thread : Bool, &block : ->) : Fiber
       raise ArgumentError.new("#{self.class.name}#spawn doesn't support same_thread:true") if same_thread
@@ -188,11 +243,12 @@ module Fiber::ExecutionContext
     protected def steal(& : Scheduler ->) : Nil
       return if capacity == 1
 
+      schedulers = @schedulers
       i = @rng.next_int
-      n = @schedulers.size
+      n = schedulers.size
 
       n.times do |j|
-        if scheduler = @schedulers[(i &+ j) % n]?
+        if scheduler = schedulers[(i &+ j) % n]?
           yield scheduler
         end
       end
@@ -270,11 +326,11 @@ module Fiber::ExecutionContext
       # check if we can start another thread; no need for atomics, the values
       # shall be rather stable over time and we check them again inside the
       # mutex
-      return if @threads.size == capacity
+      return if @threads.size >= capacity
 
       @mutex.synchronize do
         index = @threads.size
-        return if index == capacity # check again
+        return if index >= capacity # check again
 
         @threads << start_thread(@schedulers[index])
       end