ractor_bitonic_sort.rb

def ractor_bitonic_sort array
  size = array.size
  level = size.bit_length - 1
  raise 'size is not power of 2' unless size == 1 << level
  pipe = Ractor.new { loop { Ractor.yield Ractor.recv } }
  ractors = array.map.with_index do |v, i|
    Ractor.new pipe, v, i, size, level do |pipe, v, index, size, level|
      received = {}
      conditional_recv = -> key {
        return received.delete key if received.key? key
        loop do
          rkey, value = Ractor.recv
          return value if rkey == key
          received[rkey] = value
        end
      }
      # create link between 2**n gap nodes
      # links[i] is ractors[index + 2**(i-1)]
      # links[-i] is ractors[index - 2**(i-1)]
      links = [nil] * (2 * level + 1)
      (1..level).each do |l|
        next_worker = conditional_recv.call l
        next_worker.send [-l, self], move: true
        prev_worker = conditional_recv.call -l
        links[l] = next_worker
        links[-l] = prev_worker
        prev_worker.send [l + 1, next_worker], move: true if l != level
      end

      # bitonic
      (1..level).each do |step|
        block_size = 1 << step
        asc = (index / block_size).even?
        step.downto 1 do |l|
          upper = (index / (1 << (l - 1))).even?
          key = [step, l]
          links[upper ? l : -l].send [key, v]
          v2 = conditional_recv.call key
          v = asc ^ upper ? [v, v2].max : [v, v2].min
        end
      end
      pipe << [index, v]
    end
  end
  # start create link: notify ractors[i] that it's next_worker is ractors[i+1]
  ractors.each_with_index do |r, i|
    ractors[i - 1].send [1, r], move: true
  end
  result = []
  size.times do
    i, v = pipe.take
    result[i] = v
  end
  result
end

p ractor_bitonic_sort 128.times.to_a.shuffle

ractor_bitonic_sort2.rb

def normal_bitonic_sort(array)
  size = array.size
  level = size.bit_length - 1
  raise 'size is not power of 2' unless size == 1 << level
  level.times do |step|
    block_size = 2 << step
    step.downto 0 do |substep|
      gap = 1 << substep
      array.each_with_index do |v, i|
        asc = (i / block_size).even?
        next unless (i / gap).even?
        j = i + gap
        w = array[j]
        array[i], array[j] = w, v if asc ^ (v < w)
      end
    end
  end
  array
end

def ractor_bitonic_sort(array, num_ractors: 8)
  size = array.size
  level = size.bit_length - 1
  size_per_worker = (size + num_ractors - 1) / num_ractors
  raise 'size is not power of 2' unless size == 1 << level
  workers = num_ractors.times.map do |wi|
    start_index = size_per_worker * wi
    Ractor.new array[start_index, size_per_worker], level, start_index, size_per_worker, size do |array, level, start_index, size_per_worker, size|
      worker_index = start_index / size_per_worker
      received = {}
      conditional_recv = -> key {
        return received.delete key if received.key? key
        loop do
          rkey, value = Ractor.recv
          return value if rkey == key
          received[rkey] = value
        end
      }
      pipe, ractors = conditional_recv.call :init
      level.times do |step|
        block_size = 2 << step
        step.downto 0 do |substep|
          gap = 1 << substep
          sends = {}
          array.each_with_index do |v, ii|
            i = start_index + ii
            upper = (i / gap).even?
            j = upper ? i + gap : i - gap
            if j / size_per_worker != worker_index
              (sends[j / size_per_worker] ||= [i, []]).last << v
            end
          end
          sends.each do |wi, v|
            ractors[wi] << [[step, substep, worker_index], v]
          end
          recvs = {}
          sends.keys.each do |wi|
            recvs[wi] = conditional_recv.call [step, substep, wi]
          end
          array.each_with_index do |v, ii|
            i = start_index + ii
            asc = (i / block_size).even?
            upper = (i / gap).even?
            j = upper ? i + gap : i - gap
            if j / size_per_worker != worker_index
              w_start, w_array = recvs[j / size_per_worker]
              w = w_array[j - w_start]
              array[ii] = asc ^ upper ? [v, w].max : [v, w].min
            elsif upper
              jj = j - start_index
              w = array[jj]
              array[ii], array[jj] = w, v if asc ^ (v < w)
            end
          end
        end
      end
      pipe << array
    end
  end
  pipes = num_ractors.times.map { Ractor.new { Ractor.recv } }
  workers.zip pipes do |ractor, pipe|
    ractor.send [:init, [pipe, workers.dup]], move: true
  end
  pipes.map(&:take).inject(:+)
end

arr = 4096.times.to_a.shuffle
t=Time.now; a=arr.sort; ta = Time.now - t
t=Time.now; b=normal_bitonic_sort arr; tb = Time.now - t
t=Time.now; c=ractor_bitonic_sort arr; tc = Time.now - t
p a==b && b == c
p ta, tb, tc, tb / ta, tc / ta, tc / tb