Reimplement um_op and related code

README.md

@@ -32,6 +32,8 @@ UringMachine is based on my experience marrying Ruby and io_uring:
 - [IOU](https://github.com/digital-fabric/iou) - a low-level asynchronous API
   for using io_uring from Ruby.
 
+### Learnings
+
 Some important learnings from those two projects, in no particular order:
 
 - Monkey-patching is not a good solution, long term. You need to deal with
@@ -73,6 +75,30 @@ based on the following principles:
 - Do not insist on structured concurrency, just provide the APIs necessary to
   create actors and to supervise the execution of fibers.
 
+### Cancellation
+
+When working with io_uring, managing the life cycle of asynchronous operations
+is quite tricky, especially with regards to cancellation. This is due to the
+fact each operation lives on both sides of the userspace-kernel divide. This
+means that when cancelling an operation, we cannot free, or dispose of any
+resources associated with the operation, until we know for sure that the kernel
+side is also done with the operation.
+
+As stated above, working with fibers allows us to keep operation metadata and
+associated data (such as buffers etc) on the stack, which can greatly simplify
+the managing of the operation's lifetime, as well as significantly reduce heap
+allocations.
+
+When a cancellation does occur, UringMachine issues a cancellation (using
+`io_uring_prep_cancel64`), and then waits for the corresponding CQE (with a
+`-ECANCELED` result).
+
+### No scheduler, no runqueue
+
+UringMachine manages the scheduling of fibers by relying solely on io_uring
+CQE's. There's no list of ready fibers, no runqueue. Manually scheduling fibers
+is done by issuing `NOP` operations and then processing CQE's one by one.
+
 ## Example
 
 ```ruby

TODO.md

@@ -1,45 +1,7 @@
-- futex wait wake
-
-```ruby
-# interface:
-f = UM::Mutex.new
-f.synchronize { ... }
-
-q = UM:Queue.new
-# API is similar to stock Ruby Queue class
-q << :foo # same as q.push(:foo)
-q.shift
-q.pop
-q.unshift(:foo)
-```
-
-But how do we use this in conjunction with a UringMachine instance?
-
-```ruby
-f = UM::Mutex.new
-machine.synchronize(futex) { ... } # looks good
-
-# or maybe:
-f.synchronize(machine) { ... } # looks a bit weird
-
-# how about queues?
-machine.push(q)
-machine.pop(q)
-machine.shift(q)
-machine.unshift(q)
-
-# what about events?
-UM::Event
-```
-
-
-- queues
-
-
 - splice / - tee
 - sendto
 - recvfrom
-- poll
+- poll_add / poll_remove / poll_multishot / poll_update
 - open / openat
 - fsync
 - mkdir / mkdirat
@@ -51,5 +13,4 @@ UM::Event
 - madvise
 - getxattr / setxattr
 - shutdown
-
 - send_bundle / recv_bundle (kernel >= 6.10)

examples/bm_snooze.rb

@@ -0,0 +1,89 @@
+# frozen_string_literal: true
+
+require 'bundler/inline'
+
+gemfile do
+  source 'https://rubygems.org'
+  gem 'uringmachine', path: '..'
+  gem 'benchmark-ips'
+end
+
+require 'benchmark/ips'
+require 'uringmachine'
+
+ITERATIONS = 1000
+
+$machine = UringMachine.new
+
+def run_snooze
+  count = 0
+  main = Fiber.current
+
+  f1 = Fiber.new do
+    loop do
+      count += 1
+      if count == ITERATIONS
+        $machine.schedule(main, nil)
+        break
+      else
+        $machine.snooze
+      end
+    end
+  end
+
+  f2 = Fiber.new do
+    loop do
+      count += 1
+      if count == ITERATIONS
+        $machine.schedule(main, nil)
+        break
+      else
+        $machine.snooze
+      end
+    end
+  end
+
+  $machine.schedule(f1, nil)
+  $machine.schedule(f2, nil)
+  $machine.yield
+end
+
+def run_raw_transfer
+  count = 0
+  main = Fiber.current
+  f2 = nil
+  f1 = Fiber.new do
+    loop do
+      count += 1
+      if count == ITERATIONS
+        main.transfer(nil)
+        break
+      else
+        f2.transfer(nil)
+      end
+    end
+  end
+
+  f2 = Fiber.new do
+    loop do
+      count += 1
+      if count == ITERATIONS
+        main.transfer(nil)
+        break
+      else
+        f1.transfer(nil)
+      end
+    end
+  end
+
+  f1.transfer(nil)
+end
+
+bm = Benchmark.ips do |x|
+  x.config(:time => 5, :warmup => 2)
+
+  x.report("snooze")        { run_snooze }
+  x.report("raw transfer")  { run_raw_transfer }
+
+  x.compare!
+end

examples/bm_write.rb

@@ -0,0 +1,56 @@
+# frozen_string_literal: true
+
+require 'bundler/inline'
+
+gemfile do
+  source 'https://rubygems.org'
+  gem 'uringmachine', path: '..'
+  gem 'benchmark'
+end
+
+require 'benchmark'
+require 'uringmachine'
+
+ITERATIONS = 100000
+BUF = ('*' * 8192).freeze
+FN = '/tmp/bm_write'
+
+def run_io_write(num_threads)
+  FileUtils.rm(FN) rescue nil
+  fio = File.open(FN, 'w')
+
+  threads = num_threads.times.map do |i|
+    Thread.new do
+      ITERATIONS.times { fio.write(BUF) }
+    end
+  end
+  threads.each(&:join)
+ensure
+  fio.close
+end
+
+def run_um_write(num_fibers)
+  FileUtils.rm(FN) rescue nil
+  fio = File.open(FN, 'w')
+  fd = fio.fileno
+
+  machine = UringMachine.new
+  done = UringMachine::Queue.new
+  num_fibers.times do
+    machine.spin do
+      ITERATIONS.times { machine.write(fd, BUF) }
+      machine.push(done, true) 
+    end
+  end
+  num_fibers.times { machine.pop(done) }
+ensure
+  fio.close
+end
+
+Benchmark.bm do |x|
+  [1, 2, 4, 8].each do |c|
+    x.report("IO (#{c} threads)")     { run_io_write(c) }
+    x.report("UM (#{c} fibers) ")     { run_um_write(c) }
+    puts
+  end
+end

examples/dns_client.rb

@@ -0,0 +1,12 @@
+# frozen_string_literal: true
+
+require_relative '../lib/uringmachine'
+require 'resolv'
+
+machine = UM.new
+
+addrs = machine.resolve('status.realiteq.net')
+
+puts '*' * 40
+puts addrs.join("\n")
+puts

examples/http_server.rb

@@ -23,9 +23,10 @@ def http_handle_connection(fd)
   #   parser << _1
   #   break if done
   # end
+
   # puts "Connection closed on fd #{fd}"
 rescue => e
-  puts "Error while handling connection on fd #{fd}: #{e.inspect}"
+  # puts "Error while handling connection on fd #{fd}: #{e.inspect}"
 ensure
   @machine.close(fd) rescue nil
 end
@@ -36,15 +37,21 @@ def http_send_response(fd, body)
   @machine.write(fd, msg)
 end
 
-trap('SIGINT') { exit! }
+trap('SIGINT') { exit }
 
 server_fd = @machine.socket(UM::AF_INET, UM::SOCK_STREAM, 0, 0)
+@machine.setsockopt(server_fd, UM::SOL_SOCKET, UM::SO_REUSEADDR, true)
 @machine.bind(server_fd, '127.0.0.1', 1234)
 @machine.listen(server_fd, UM::SOMAXCONN)
 puts 'Listening on port 1234'
 
-at_exit { @machine.close(server_fd) rescue nil }
+at_exit do
+  puts "Closing server FD"
+  @machine.close(server_fd) rescue nil
+  puts "done!"
+end
 
 @machine.accept_each(server_fd) do |fd|
   @machine.spin(fd) { http_handle_connection _1 }
 end
+p :post_accept_each

examples/server_client.rb

@@ -0,0 +1,64 @@
+# frozen_string_literal: true
+
+require_relative '../lib/uringmachine'
+
+PORT = 1234
+
+@machine = UM.new
+@bgid = @machine.setup_buffer_ring(4096, 1024)
+
+@counter = 0
+
+def handle_connection(fd)
+  buf = +''
+  loop do
+    res = @machine.recv(fd, buf, 8192, 0)
+    break if res == 0
+
+    @machine.write(fd, buf)
+    @counter += 2
+  end
+ensure
+  @machine.close(fd) rescue nil
+end
+
+def run_client
+  fd = @machine.socket(UM::AF_INET, UM::SOCK_STREAM, 0, 0)
+  @machine.connect(fd, '127.0.0.1', PORT)
+  msg = 'foo' * 30
+  buf = +''
+  loop do
+    @machine.send(fd, msg, msg.bytesize, 0)
+    res = @machine.recv(fd, buf, 8192, 0)
+    @counter += 2
+
+    break if res == 0
+    raise "Got #{res} bytes instead of #{msg.bytesize}" if res != msg.bytesize
+  end
+end
+
+trap('SIGINT') { exit }
+
+server_fd = @machine.socket(UM::AF_INET, UM::SOCK_STREAM, 0, 0)
+@machine.setsockopt(server_fd, UM::SOL_SOCKET, UM::SO_REUSEADDR, true)
+@machine.bind(server_fd, '127.0.0.1', PORT)
+@machine.listen(server_fd, UM::SOMAXCONN)
+puts "Listening on port #{PORT}"
+
+at_exit { @machine.close(server_fd) rescue nil }
+
+20.times do
+  @machine.spin { run_client }
+end
+
+@machine.spin do
+  @machine.accept_each(server_fd) do |fd|
+    @machine.spin(fd) { handle_connection _1 }
+  end
+end
+
+t0 = Time.now
+@machine.sleep 3
+t1 = Time.now  
+elapsed = t1 - t0
+puts "Did #{@counter} ops in #{elapsed} seconds (#{(@counter / elapsed)} ops/s)"

examples/snooze.rb

@@ -0,0 +1,44 @@
+# frozen_string_literal: true
+
+require_relative '../lib/uringmachine'
+
+@machine = UM.new
+
+@counter = 0
+@fiber_count = 0
+
+def start_fiber
+  @fiber_count += 1
+  @machine.spin do
+    max_count = @counter + rand(1000)
+    puts "Start #{Fiber.current} #{max_count - @counter}"
+    loop do
+      @machine.sleep 0.001
+      @counter += 1
+      break if @counter >= max_count
+    end
+    puts "Stop #{Fiber.current}"
+  ensure
+    @fiber_count -= 1
+  end
+end
+
+t0 = Time.now
+MAX_FIBERS = 20
+MAX_TIME = 10
+loop do
+  @machine.sleep 0.1
+  puts "pending: #{@machine.pending_count}"
+  break if (Time.now - t0) > MAX_TIME
+  start_fiber while @fiber_count < MAX_FIBERS
+end
+t1 = Time.now  
+elapsed = t1 - t0
+rate = @counter / elapsed
+puts "Did #{@counter} ops in #{elapsed} seconds (#{rate} ops/s)"
+
+puts "Waiting for fibers... (#{@fiber_count})"
+while @fiber_count > 0
+  @machine.sleep 0.1
+end
+puts "Done"