Module: AsyncFutures::IOAsync

Defined in:: lib/async_futures/io_async.rb

Overview

This is a simple example mixin module to add async IO to the IO and OpenSSL::SSLSocket classes. You can also include this in other classes so long as they have the methods write_nonblock and read_nonblock and they behave the same way as IO and/or OpenSSL::SSLSocket.

All reads and writes are done on (at least) one background worker thread.

This is not the most efficient implementation. It is just meant to be an example of how one can use the Future class outside of an Executor implementation.

It's only real efficiencies, so to speak, are that it doesn't use a thread per future (which would potentially use up a lot of memory), nor do futures need to wait until a thread opens up on a worker pool (which would block newer IO work until older IO work completely finished). Instead work is picked up immediately on at least one background worker thread, and attempts to read/write are started immediately via read_nonblock and write_nonblock. If read_nonblock/write_nonblock cannot proceed (because they would block) for any particular IO object then the next operation for the next IO object is attempted, and so forth, until all the work is completed (or a timeout happens).

However, all of this is accomplished via a simplistic, unoptimized busy loop. This is less than ideal. There are some simple sleeps and timeouts added to avoid completely eating up the CPU, but this is still a very naive approach.

A better implementation would utilize higher performance OS specific features like FreeBSD's kqueue/aio or Linux's epoll/io_uring. However, the logic for integrating these is beyond the scope of this example code.

This could probably be done using the FFI library fiddle, which is bundled with ruby, so it wouldn't need to reach outside the standard library. However I have a very good reason for not doing that right now: I don't want to.

Instance Method Summary collapse

#read_async(maxlen, timeout = nil, sleep_timeout: 0.001) ⇒ Object
Return an incomplete future that will eventually contain the string value read from the IO object or an exception if the IO object could not be read from for some reason.
#write_async(string, timeout = nil, sleep_timeout: 0.001) ⇒ Object
Return an incomplete future that will eventually contain an integer with the number of bytes written or an exception if the string could not be written for some reason.

Instance Method Details

#read_async(maxlen, timeout = nil, sleep_timeout: 0.001) ⇒ `Object`

Return an incomplete future that will eventually contain the string value read from the IO object or an exception if the IO object could not be read from for some reason.

A string up to maxlen in length is read in a nonblocking fashion on a background worker thread.

The optional timeout argument causes the work to finish the future exceptionally with Timeout::Error if it takes longer than timeout seconds to complete. This is used to avoid having background work that spins forever on IO that may never complete. If nil or no value is given, this means no timeout (i.e. potentially spin indefinitely).

This should not be confused with the Timeout::Error raised via the timeout argument on Future.result and Future.exception. If it matters for your purposes to differentiate between the two, you can do something like the following:

# `join` returns `nil` on timeout
result = if future.join(1.0)
           # If `Timeout::Error` is raised here,
           # it is from the `timeout` parameter to the `*_async` method.
           future.result
         else
           raise Timeout::Error.new('Timed out on `join`')
         end

The optional sleep_timeout keyword argument is used to determine how quickly the worker thread stops polling the input work queue and how much sleep time happens between failed nonblocking IO attempts. It defaults to 1ms. If existing worker(s) have already been spawned, then this argument isn't used.

If the process shuts down before the future can be fully completed, the work may be abandoned even if it partially completed.

# File 'lib/async_futures/io_async.rb', line 172

def read_async(maxlen, timeout = nil, sleep_timeout: 0.001) # rubocop:disable Metrics/AbcSize,Metrics/CyclomaticComplexity
  Future.new.tap do |ftr|
    clock_timeout = timeout && (Time.now.to_f + timeout)

    work_proc = proc do
      # :nocov:
      break ftr unless ftr.set_running_or_notify_cancel(set_context: true)
      # :nocov:

      to_err = Timeout::Error.new('execution expired')
      to_read_length = maxlen
      retrieved_str = String.new

      loop do
        cur_to = timeout && (Time.now.to_f - clock_timeout)
        break ftr.tap { ftr.set_exception(to_err) } unless timeout.nil? || cur_to.positive?

        retrieved_str << read_nonblock(to_read_length)
        to_read_length = maxlen - retrieved_str.size
        break ftr.tap { ftr.set_result(retrieved_str) } if to_read_length.zero?
      rescue IO::WaitReadable, IO::WaitWritable, Errno::EINTR
        Fiber.yield nil
        retry
      rescue EOFError
        break ftr.tap { ftr.set_result(retrieved_str) }
      rescue Exception => e # rubocop:disable Lint/RescueException
        break ftr.tap { ftr.set_exception(e) }
      end
    end

    io_async_queue.push(work_proc)
    maybe_spawn_worker(sleep_timeout)
  end
end

#write_async(string, timeout = nil, sleep_timeout: 0.001) ⇒ `Object`

Return an incomplete future that will eventually contain an integer with the number of bytes written or an exception if the string could not be written for some reason.

The string argument is written in a nonblocking fashion on a background worker thread.

# `join` returns `nil` on timeout
result = if future.join(1.0)
           # If `Timeout::Error` is raised here,
           # it is from the `timeout` parameter to the `*_async` method.
           future.result
         else
           raise Timeout::Error.new('Timed out on `join`')
         end