A Hash-like async primitive similar to Notification or Variable

[postmodern]

Hello, I was searching the async gem's API looking for a primitive that would allow me to manage async RPC requests and the resulting responses. While the Async API already provides Notification and Variable, I think it would be useful to provide a Hash-like object for registering a notification using a unique key, and then allowing other code to trigger it's notification once a value has been returned for that unique key. This would help developers implement a "cubby-hole" style notification system for requested values and the eventual returned values.

[ioquatix]

Sounds interesting, can you give me a simple example of the usage?

[postmodern]

In my use case I want to send and receive the RPC requests and responses asynchronously, but return the response for a given request synchronously.

@inbox = Async::HashLikeNotifier.new
@requests = Async::Queue.new
@responses = Async::Queue.new
@request_id = 0

def next_id
  @request_id += 1
end

def send_request(name,*args)
  request_id = next_id

  @requests.enqueue(Request.new(name,args, id: request_id))

  # indicate that we are waiting for a value for the requestID
  @inbox.register(request_id)
  return request_id
end

Async do |task|
  task.async do
    while (line = stdin.gets)
      name, *args = line.split

      request_id = send_request(name,*args)

      # wait for a value to be returned
      value = @inbox[request_id]
      puts "=> #{value.inspect}"
    end
  end

  task.async do
    # imagine that other RPC requests are also being sent from other network connections or UNIX sockets
  end

  task.async do
    # imagine that responses are read from a socket or an HTTP session and enqueued onto @responses
  end

  task.async do
    while (response = @responses.deq)
      # notify the waiting fiber of the received value
      @inbox[response.id] = response.value
    end
  end
end

Effectively the HashLikeNotifier is a routing table for which fiber is waiting for which response, and where to route a response value to.

[trevorturk]

I could see using something like this, and my use case isn't far off, but I'm curious if you can accomplish basically the same thing with an async barrier? I'm collecting tasks in a barrier and then waiting on the result. Perhaps this hash inbox would be a bit easier to work with, or avoiding the wait would be more efficient?

[postmodern]

but I'm curious if you can accomplish basically the same thing with an async barrier?

I'm not sure how one would use Async::Barrier to accomplish this. The idea is to allow multiple notifications or variables to be registered, and allow them to individually trigger the code which is waiting on them. The idea is sort of like old fashioned mailbox cubbies in an office, where mail is sorted into individual slots for workers, and then the individual office workers pick up their mail from their assigned slot.

[trevorturk]

Yeah, I'm not totally sure. In my case, I'm getting a request, issuing a bunch of HTTP requests to upstream APIs, putting them all into a barrier, then waiting for them all to finish before sending a response to the client. In your case, I'm wondering if you add the tasks that need to complete before issuing a response into a barrier, then wait on that before returning the response? I do think this hash notification system seems useful, and perhaps I could use it, I'm just trying to tease out if there might be some way to construct things with the barrier primitive that could even be used to build the new feature.

[postmodern]

@trevorturk the scenario you are describing is different from what I'm describing. In my scenario I have multiple connected clients sending requests to various other upstream sessions and waiting for their individual responses, and each response is returned to the requesting client when the upstream transmits back the response message to the corresponding request. Sort of like a load balancing reverse proxy, except the request can pick which internal server the request is routed to.

[emiltin]

Could you use a hash of Async::Queue or Async::Condition?

require 'async'
require 'async/queue'

class Cubbies
  def initialize() @inboxes = {} end
  def box(id) @inboxes[id] ||= Async::Queue.new end
  def drop(id) @inboxes.delete(id) end
  def place(id, value) box(id).enqueue value end

  def fetch id
    value = box(id).dequeue
    drop(id) if box(id).empty?
    value
  end
end

cubbies = Cubbies.new
names = [:anna,:john,:lily,:dave]
fruits = [:apple,:orange,:banana,:pear,:mango]

Async do
  Async do
    loop do
      person, fruit = names.sample, fruits.sample
      puts "#{fruit} ready for #{person}"
      cubbies.place(person,fruit)
      sleep rand(10)*0.1
    end
  end

  Async do
    loop do
      person = names.sample
      puts "#{person} is hungry"
      fruit = cubbies.fetch(person)
      puts "#{person} eats #{fruit}"
      sleep rand(10)*0.1
    end
  end
end

=>
banana ready for anna
dave is hungry
banana ready for john
banana ready for lily
pear ready for dave
dave eats pear
dave is hungry
apple ready for anna
mango ready for lily
pear ready for john
pear ready for dave
dave eats pear
banana ready for anna
lily is hungry
lily eats banana
banana ready for anna
john is hungry
john eats banana