integer = 1
# automatically queues a message in the inbox
r1 = Ractor.new(integer, name: 'r1') do |i|
  puts "Executing #{i} + 2"
  i + 2
end
r1.inspect # => "#<Ractor:#1 r1 (irb):8 terminated>"

# the Ractor already did its work and the in-port was never opened (value was copied on initialization)
r1.send(1)
# `send': The incoming-port is already closed (Ractor::ClosedError)

# The Ractor outport is open, waiting for us to `take` the result
r1.take
# => 3

# after we retrieve the result the outport is closes too
r1.take
# `take': The outgoing-port is already closed (Ractor::ClosedError)

# now the Ractor `r1` is no longer usable / executable and will be garbage collected

r1 = Ractor.new(name: 'r1') do
  loop do
    input = Ractor.receive
    result = input + 2
    puts "Executed - result will be: #{result}"
    Ractor.yield(result)
  end
end

r1.inspect
# => #<Ractor:#8 r1 (irb):12 blocking>

r1.send(1)
# Executed - result will be: 3
# => #<Ractor:#8 r1 (irb):41 blocking>

# we can add messages to the incoming queue
r1.send(2) # note this time it doesn't execute yet as we haven't taken the last result

# we fetch our first result and immediately the next message in the queue is executed
r1.take
# Executed - result will be: 4
# => 3

# now we get the last result, but nothing executes (since nothing is in the queue)
r1.take
# => 4

# we see that our Ractor is still waiting (blocking) for something to do
r1.inspect
# => "#<Ractor:#8 r1 (irb):41 blocking>"

# if we try to take now we will block our current thread too (the Ractor is waiting for an input and we are waiting for an answer)!
r1.take

integer = 'a'

# automatically queues a message in the inbox
r1 = Ractor.new(integer, name: 'doomed') { |i| i + 2 }
#<Thread:0x0000000104373ae8 run> terminated with exception (report_on_exception is true):
# (irb):16:in `+': no implicit conversion of Integer into String (TypeError) from (irb):16:in `block in <top (required)>'
# => #<Ractor:#4 doomed (irb):16 running>

# because the Ractor was build and immediately tried to process
# the return messages are in an odd order.
# actually checking the status confirms it died
r1.inspect
# => "#<Ractor:#9 r1 (irb):58 terminated>"

r1 = Ractor.new(name: 'r1') do
  loop do
    input = Ractor.receive
    result = input + 2
    puts "Executed - result will be: #{result}"
    Ractor.yield(result)
  end
end
=> #<Ractor:#11 r1 (irb):70 blocking>
r1.name
# => "r1"

def fibonacci(n)
  ans = recursion(n)
  result = "#{n} - #{ans}"
  puts result
  result
end

def recursion(n)
  n <= 1 ? n : recursion( n - 1 ) + recursion( n - 2 )
end

def fibonacci(n) = (0..n).inject([1,0]) { |(a,b), _| [b, a+b] }[0]

def fibonacci(n)
  ans = recursion(n)
  result = "#{n} - #{ans}"
  puts result # its nice to see the results as you go
  result
end

def recursion(n)
  n <= 1 ? n : recursion( n - 1 ) + recursion( n - 2 )
end

rounds = 40
pool_size = 6
results = []

# Single Thread
t1 = Time.now
rounds.times do |i|
  # trigger the calculations and collect our results
  results << fibonacci(i)
end
puts "Single thread - duration #{Time.now - t1}"

pp results

0 - 0
1 - 1
2 - 1
3 - 2
4 - 3
5 - 5
6 - 8
...
Single thread - duration 17.414451

t1 = Time.now; threads = []
results = [37].cycle(10).map { |i| Thread.new { [i, fib(i)] }.value }
puts "10 Ruby Threads Sequential - duration #{Time.now - t1}" # duration 24.546575

# 10 Threads
t1 = Time.now; threads = []
[37].cycle(10).each { |i| threads << Thread.new { Thread.current[:results] = [i, fib(i)] } }
results = threads.map { |t| t.join; t[:results] }
puts "10 Ruby Threads Sequential - duration #{Time.now - t1}" # duration 24.546575

def fibonacci(n)
  ans = recursion(n)
  result = "#{n} - #{ans}"
  puts result
  result
end

def recursion(n)
  n <= 1 ? n : recursion( n - 1 ) + recursion( n - 2 )
end

# ThreadPool- Naive Source: https://rossta.net/blog/a-ruby-antihero-thread-pool.html
class ThreadPoolNaive
  def initialize(size:)
    @pool = []
  end

  def schedule(*args, &block)
    @pool << Thread.new { block.call(args) }
  end

  def shutdown
    @pool.map(&:join)
  end
end

rounds = 40
pool_size = 4
results = []

# Thread Pool
pool = ThreadPoolNaive.new(size: pool_size)
t1 = Time.now
rounds.times do |i|
  pool.schedule {
    results << fibonacci(i)
  }
end
pool.shutdown
puts "Concurrent Threads with #{pool_size} threads - duration #{Time.now - t1}"
pp results

0 - 0
2 - 1
3 - 2
31 - 1346269
32 - 2178309
6 - 8
7 - 13
...
Concurrent Threads with 4 threads - duration 17.351042

def fibonacci(n)
  ans = recursion(n)
  result = "#{n} - #{ans}"
  puts result # its nice to see the results as you go
  result
end

def recursion(n)
  n <= 1 ? n : recursion( n - 1 ) + recursion( n - 2 )
end

rounds = 40
pool_size = 4
results = []

# generate a Ractor that will act as a worker pool
pool = Ractor.new {
  loop { Ractor.yield(Ractor.receive) }
}

# generate Workers that will do the work
workers = (1..pool_size).map do |i|
  Ractor.new(pool) do |input_pipe|
    loop { Ractor.yield(fibonacci(input_pipe.take)) }
  end
end

t1 = Time.now
rounds.times { |i| pool.send(i) }   # send requests
rounds.times {
  answer = Ractor.select(*workers)
  # returns: [#<Ractor:#6 r4 slow_ractor_pool.rb:18 blocking>, "3 - 2"]
  results << answer.last # .last because we only want the result
}
puts "Parallel Ractors - pool size: #{pool_size} - duration #{Time.now - t1}"
pp results

2 - 1
0 - 0
1 - 1
3 - 2
6 - 8
4 - 3
8 - 21
...
Four Parallel Ractors - pool size: 4 - duration 7.912549

def fib n
  if n < 2
    1
  else
    fib(n-2) + fib(n-1)
  end
end

max = 39
rs = max.downto(1).map { |i| i }.map do |i|
  Ractor.new i do |i|
    [i, fib(i)]
  end
end

t1 = Time.now
until rs.empty?
  r, v = Ractor.select(*rs)
  rs.delete r
  p answer: v
end
puts "Parallel Ractors - full cpu usage - duration #{Time.now - t1}"

object = 'Hi'
object.object_id # 25940

r1 = Ractor.new do
  # ruby compiler doesn't allow the same name in a Ractor as an external object!
  object_in = Ractor.receive
  p object_in.object_id # 47640
  object_in # result given to the outbox
end

# we push `send` our message
r1.send(object)
# 47640  # print executes

# given there is no loop the incoming port closes after the first message
r1.send('Hoi')
# `send': The incoming-port is already closed (Ractor::ClosedError)

# collect our result with:
r1.take
# 'Hi'

# again, without a loop Ractors close the outbox when emptied
r1.take
# `take': The outgoing-port is already closed (Ractor::ClosedError)

r1 = Ractor.new do
  obj = Ractor.receive # gets sent information
  obj # result available in the outport
end

rt.take # waits for the outbox, but has not input so you will wait forever!
# ctrl-c

r1 = Ractor.new do
  3.times { |i| Ractor.yield(i) }
end
3.times { p r1.take }
# 0
# 1
# 2
# => 3 # like normal ruby code the last value is returned

# because 3 was the last result will also be returned
r1.take
# => 3

r1.take # now the outbox is empty
# `take': The outgoing-port is already closed (Ractor::ClosedError)

r1 = Ractor.new do
  3.times { |i| Ractor.yield(i) }
  :r1_done
end
3.times { p r1.take }
# 0
# 1
# 2
# => :r1_done
r1.take
# => :r1_done

max_fib = 39

def fib(n) = n < 2 ? 1 : fib(n-2) + fib(n-1)

# we create an Array of Ractors to do the work needed.
work_queue = (1..max_fib).map do |i|
  Ractor.new(i) { |i| [i, fib(i)] }
end

t1 = Time.now

# we wait here in this loop for our queued work to finish (using `select`)
until work_queue.empty?
  # 'select' returns both the Ractor and the value
  terminated_ractor, computed_value = Ractor.select(*work_queue)
  # the Ractor is now terminated - so we remove it from our work queue
  work_queue.delete terminated_ractor
  p answer: computed_value
end
puts "Parallel Ractors - full cpu usage - duration #{Time.now - t1}"

def increment(n) = n + 1

r1 = Ractor.new do
  loop do
    incoming_r1 = Ractor.receive
    incremented_once = increment(incoming_r1)
    Ractor.yield(incremented_once)
  end
end

r2 = Ractor.new(r1) do |r1|
  loop do
    incoming_r2 = r1.take
    incremented_twice = increment(incoming_r2)
    Ractor.yield(incremented_twice)
  end
end

# Main Thread (Main Ractor)
r1.send(1)
r2.take
# => 3 - since 1 was incremented twice

def fibonacci(n)
  ans = (0..n).inject([1,0]) { |(a,b), _| [b, a+b] }[0]
  result = "#{n} - #{ans}"
  puts result
  result
end

pool = Ractor.new do
  loop do
    input = Ractor.receive
    Ractor.yield(input)
  end
end

# 4 workers
workers = (1..4).map do |i|
  Ractor.new(pool, name: "r#{i}") do |p|
    loop do
      input = p.take # from pool
      output_value = fibonacci(input)
      Ractor.yield(output_value)
    end
  end
end

results = []

# send 8 Requests
40.times { |i| pool.send(i) }
# notice the first 4 values are computed immediately, and then it stops since the outports are full (queue of 1)

# collect our Results - wait for multiple
40.times { results << Ractor.select(*workers) }
# now that we collect the results the inboxes will be processed

# if we are uncertain how

# notice select returns the Ractor that did the processing and the result
pp results

# collect just the results and sort them since they are processed in any random order
pp results.map { |r| r.last }.sort

def task(char) = char + '2'

r1 = Ractor.new(name: 'r1') do
  task(Ractor.receive)
end

r1.send('1')
p r1.take # => '12'

# without loop after first usage it closes
begin
  r1.send('2')
rescue Ractor::ClosedError => e
  p e
  p e.cause
  p e.ractor
end
# => #<Ractor::ClosedError: The incoming-port is already closed>

p r1
#<Ractor:#2 r1 (irb):3 terminated>

r1 = Ractor.new(name: 'r1') do
  loop { task(receive) }
end

# if wrong type it closes with an error
begin
  r1.send(1)
  p r1.take
rescue Ractor::RemoteError => e
  p e        #<Ractor::RemoteError: thrown by remote Ractor.>
  p e.cause  #<TypeError: String can't be coerced into Integer>
  p e.ractor #<Ractor:#22 r1 (irb):236 terminated>
end

r1.send('2')
# => #<Ractor::ClosedError: The incoming-port is already closed>

def increment1(n) = n + 1
def increment2(n) = n + 2

r1 = Ractor.new(name: 'r1') do
  loop do
    incoming_r1 = Ractor.receive
    incremented_once = increment1(incoming_r1)
    Ractor.yield(incremented_once)
  end
end

r2 = Ractor.new(r1, name: 'r2') do |r1|
  loop do
    incoming_r2 = r1.take
    incremented_twice = increment2(incoming_r2)
    Ractor.yield(incremented_twice)
  end
end

# Main Thread (Main Ractor)
r1.send(1)
r2.take # => 4

r1.send(2)
# if we don't collect (take) our result it gets lost with the propagated crash

r1.send('c')
#<Thread:0x00000001081260c0 run> terminated with exception (report_on_exception is true):

r2.take # <internal:ractor>:698:in `take': thrown by remote Ractor. (Ractor::RemoteError)

def task(char) = char + '2'

def r1_init = Ractor.new(name: 'r1') { loop { Ractor.yield task(Ractor.receive) } }

r1 = r1_init

r1.send('1')
r1.take # => '12'

input = 1
begin
  r1.send(input)
  p r1.take #<Thread:0x0000000103feca28 run> terminated with exception (report_on_exception is true):
rescue Ractor::RemoteError => e
  pp e.cause
  pp e.ractor
  pp e.ractor.name

  # restart ractor
  r1 = r1_init

  # retry 'sending' & 'taking' after input coercion
  input = input.to_s
  retry
end
# #<TypeError: String can't be coerced into Integer>
#<Ractor:#2 r1 (irb):21 terminated>
# "12"

p r1
#<Ractor:#3 r1 (irb):21 blocking>

MAX_CPUS = 4.freeze; MAX_FIB_NUM = 39.freeze
def fib(n) = n < 2 ? 1 : fib(n-2) + fib(n-1)
def make_pool = Ractor.new { loop { Ractor.yield(Ractor.receive) } }
def make_worker(pool)
  Ractor.new(pool) do |p|
    loop {
      input = p.take
      fib_num = fib(input)
      result = [input, fib_num]
      Ractor.yield(result)
    }
  end
end
def make_workers(pool, count)
  (1..count).map do |i|
    make_worker(pool)
  end
end
def send_work(input, pool)
  Array(input).each do |i|
    pool.send(i)
  end
end
def collect_results(input, pool, workers)
  results = []
  input.count.times do
    begin
      worker, answer = Ractor.select(*workers)
      results << answer
    rescue Ractor::RemoteError => e
      worker = e.ractor
      pp worker
      workers.delete(worker)
      # restore worker
      workers << make_worker(pool)
    end
  end
  results
end

def run_fib(input, pool, workers)
  send_work(input, pool)
  collect_results(input, pool, workers)
end

pool = make_pool
workers =  make_workers(pool, 4)
pp workers

input_list = [29, 28, 27, 26, 25]
answers = run_fib(input_list, pool, workers)
pp answers.count
pp answers
pp workers # workers are the same as above

input_list = [24, 23, '21', '22', 20]
answers = run_fib(input_list, pool, workers)
pp answers.count
pp answers
pp workers # 2 workers now have higher numbers

# We can keep using our pool even after some crashed
input_list = [34, 33, 32, 31, 30]
answers = run_fib(input_list, pool, workers)
pp answers.count
pp answers
pp workers # workers are the same as above

# source: https://kirshatrov.com/posts/ractor-web-server-part-two/
require 'webrick'

def respond(c, req)
  # process request
  path = req.path
  query = req.query
  body = req.body

  # log request
  puts
  puts req.inspect
  puts
  puts "path:"
  puts path.inspect
  puts "query:"
  puts query.inspect
  puts "Body:"
  puts body.inspect
  puts "=" * 150

  # respond to Connection
  c.print "HTTP/1.1 200\r\n"
  c.print "Content-Type: text/html\r\n"
  c.print "\r\n"
  c.print "<h1>Hello #{query['name'] || 'world'}</h1>"
  c.print "<h2>Your path: #{path}</h2>"
  c.print "<br>"
  c.print "<hr>"
  c.print "<br>"
  c.print req.inspect
end

# need to Freeze WEBrick objects that need to be passed - there is a PR to fix this, but not yet merged.
# PR: https://github.com/ruby/webrick/pull/65
Ractor.make_shareable(WEBrick::Config::HTTP)
Ractor.make_shareable(WEBrick::LF)
Ractor.make_shareable(WEBrick::CRLF)
Ractor.make_shareable(WEBrick::HTTPRequest::BODY_CONTAINABLE_METHODS)
Ractor.make_shareable(WEBrick::HTTPStatus::StatusMessage)

def make_pipe
  Ractor.new do
    loop do
      Ractor.yield(Ractor.receive, move: true)
    end
  end
end

def make_worker(pipe_line)
  Ractor.new(pipe_line) do |pipe|
    loop do
      # capture input
      s = pipe.take

      # process input
      req = WEBrick::HTTPRequest.new(
              WEBrick::Config::HTTP.merge(RequestTimeout: nil)
            )
      req.parse(s)

      respond(s, req)

      s.close
    end
  end
end

def make_workers(pipe, cpu_count)
  cpu_count.times.map { make_worker(pipe) }
end

def make_listener(pipe_line)
  Ractor.new(pipe_line) do |pipe|
    server = TCPServer.new(8080)
    loop do
      conn, _ = server.accept
      pipe.send(conn, move: true)
    end
  end
end


CPU_COUNT = 4.freeze
pipe = make_pipe
workers = make_workers(pipe, CPU_COUNT)
listener = make_listener(pipe)

loop do
  begin
    Ractor.select(listener, *workers)
  # Worker Supervision
  rescue Ractor::RemoteError => e
    # identify crashed worker
    worker = e.ractor
    # remove crashed worker
    workers.delete(worker)
    # restore a new worker
    workers << make_worker(pipe_line)
  end
end

Hello world
Your path: /

# along with all the session info

Hello bill
Your path: /hoi

# along with all the session info