threadpool_map.cr

module Indexable(T)
  def threadpool_map(workers : Int = 8, chunk_size : Int? = nil, &func : T -> R) forall T, R
    mutex = Thread::Mutex.new
    cs = chunk_size || (self.size**0.5).ceil.to_i
    Array(R).build(self.size) do |result|
      index = 0
      threads = Array.new(workers) {
        Thread.new do
          a = b = 0
          loop do
            mutex.synchronize do
              a = index
              b = (index += cs)
            end
            if b > self.size
              break if a >= self.size
              b = self.size
            end
            (a...b).each do |i|
              result[i] = func.call(self[i])
            end
          end
        end
      }
      threads.map &.join
      self.size
    end
  end
end

z-benchmark.cr

require "./threadpool_map"


def is_prime(n)
  return false if n % 2 == 0

  sqrt_n = Math.sqrt(n).to_i
  (3..sqrt_n).step(2) do |i|
    return false if n % i == 0
  end

  true
end


require "benchmark"

[
  (1_000_000_001..1_000_100_000).to_a,
  (1..3_000_000).to_a,
  [337] * 10_000_000,
].each do |numbers|
  Benchmark.bm do |r|
    a = b = nil
    r.report "thr" do
      a = numbers.threadpool_map(&->is_prime(Int32))
    end
    r.report "map" do
      b = numbers.map(&->is_prime(Int32))
    end
    raise "error" unless a == b
  end
end