Chaining together async. about async HOT 4 CLOSED

@ioquatix I noticed you wrote code like this in your Ruby Gems Source Code Analysis and was thinking it'd be useful if this were documented.

My feedback is that I think we would have a need for this. It's useful to be able to set a limit on the number of simultaneous async operations, for various reasons including:

• prevent DOS on myself by creating an arbitrarily large number of fibers
• prevent DOS on servers that my code calls

I do find this API pretty hard to wrap my head around, though. I see how it works but I'd have to think about it every time I see the code.

from async.

I do find this API pretty hard to wrap my head around, though. I see how it works but I'd have to think about it every time I see the code.

This was my first impression as well. If this is expected to be a common pattern, would it be worth supporting it more directly? Maybe some new object that behaves like a queue but also has a Barrier and Semaphore internally:

instance = Async::QueueLikeObject.new(10)

instance.async do |item|
  ...
end

instance.wait

from async.

I'm definitely a fan of composition of simple objects, so I'd avoid things like QueueLikeObject in the core async code, but have no issue of building those more complex abstractions in user code where it makes sense.

The reason why I ended up with this kind of composition is it naturally emerged from the need to control flow and limits...

The reason why I was unsure about it, is because injecting a barrier into a semaphore kind of seems odd to me. It's actually not necessary as long as you only want to go one level deep, e.g.

barrier = Async::Barrier.new
semaphore = Async::Semaphore.new(10)

queue.each do |item|
	semaphore.async(parent: barrier) do	
		# Do something with item...
	end
end

# Wait for all outstanding tasks.
barrier.wait

But as soon as you want to do more complex things, or even just inject a barrier into code that takes a parent task, it can get more complicated:

	x.report("async-http (pipelined)") do
		Async do |task|
			internet = Async::HTTP::Internet.new
			semaphore = Async::Semaphore.new(100, parent: task)
			barrier = Async::Barrier.new(parent: semaphore)
			
			# Warm up the connection pool...
			response = internet.get(URL)
			response.read
			
			i = 0
			while i < REPEATS
				barrier.async do
					response = internet.get(URL)
					
					response.read
				end
				
				i += 1
			end
			
			barrier.wait
		ensure
			internet&.close
		end
	end

In the above benchmark, we bind the semaphore to the task. Then we bind the barrier to the semaphore.

That way, we avoid dynamically looking up Task.current for every iteration, and we also bind to a specific task from the POV of the task hierarchy. It's suggested that most top level entry points take a parent: Task.current keyword argument, which allows users to control the task tree and resource usage, and now it's relatively trivial to provide a semaphore or barrier or any kind of nested task allocation strategy.

from async.

Okay, I am planning to merge this.

from async.