Suggested API change

[ColinDKelley]

Problem Statement

Recently I've worked with a team of ~6 developers converting two services from EventMachine + Synchrony to Async. That has gone well, but one detail of the Async API has tripped us all up at first and in some cases, more than once.

The confusion arises from Async { ... } being overloaded:

1. At the outer edge, Async { ... } starts up the reactor that must surround the concurrent code.
2. Inside the reactor block, Async { ... } creates a concurrent Async::Task within the reactor.

When writing concurrent code (case 2), it is necessary that it be surrounded by a reactor (case 1). If that is missing, case 2 will be misconstrued as case 1, and it won't be concurrent at all. This is an easy mistake to make, especially when writing tests.

Mitigation

To address this risk, we like to follow Design by Contract, which in this case means that concurrent code (case 2) should ensure that there is a surrounding reactor before it proceeds. Here's how we've done that, crudely:

def process_jobs(jobs)
  Async::Task.current? or raise ArgumentError, "must start up reactor first with surrounding 'Async { ... }"

  jobs
    .map { Async { _1.process! } }
    .map(&:wait)
end

The wrapper code that creates the reactor looks like:

Async do
  process_jobs(jobs)
end

Proposal

An ideal solution would separate the overloaded cases so that the contract can be implicitly enforced by the Async gem without clients needing to write manual code each time.

Since case 2 appears much more frequently in code than case 1, let's leave case 2 as-is, but change case 1 to Async.run { ... }. So the wrapper for the above code would read:

Async.run do
  process_jobs(jobs)
end

With this contract, case 2 could automatically raise an exception on this line if it is run without a surrounding reactor:

    .map { Async { _1.process! } }

=> Async::ReactorNotRunning: You must start a reactor first with `Async.run { ... }` surrounding this code.

Migration

We obviously would not want to introduce this proposal as a breaking change. Instead we could phase it in:

Phase 1

Introduce Async.run as a preferred but optional way to create the reactor at the outer level. Update all documentation to show it as the preferred way to create the reactor.

Phase 2

Deprecate Async { ... } being called without a surrounding reactor. In the deprecation warning, indicate that it should be written as Async.run { ... } if it is the outer level that is creating the reactor.

Phase 3

Complete the deprecation: If Async { ... } is called without a surrounding reactor, raise the Async::ReactorNotRunning exception as shown above.

[ioquatix]

I will think about this, but just as a short term idea, the mitigation can be simplified like this:

def process_jobs(jobs, parent: Async::Task.current)
  jobs
    .map { parent.async { _1.process! } }
    .map(&:wait)
end

This has almost identical behaviour but is the correct way to establish "Run in a parent context".

https://github.com/socketry/async/blob/main/lib/async/task.rb#L236-L238

Maybe we can adjust the error message to be a bit more meaningful. Additionally, maybe we can document this more clearly.

The reason why this is a good idea is because it enables you to control the scope of the child tasks (e.g. using a semaphore, or some other control structure).

Regarding your proposal, how does this work:

Async.run do
  Async.run do # (1) Is this an error?
  end

  Async do # (2) This isn't an error?
  end
end

Async do # (3) This is an error?
end

Sync do # (4) What about this?
end

[ColinDKelley]

Regarding your proposal, how does this work

  Async.run do # (1) Is this an error?
  end

Haha, I was going to ask you the same thing! Is it supported today to start a reactor nested inside another one? Whatever the answer, we can support it.

The nice thing about not overloading is that we can be precise in either allowing it or disallowing it.

[ColinDKelley]

Async do # (3) This is an error?
end

Yes, that is the error we want to catch (after Phase 3). It represents code that wants to be concurrent--and assumes that its caller has spun up a surrounding reactor--but the caller forgot to, or didn't understand that they needed to. This is an easy mistake to make, especially in tests.

[ColinDKelley]

Sync do # (4) What about this?
end

I'm a bit fuzzy on Sync, but if it needs a surrounding reactor, this would raise the same exception since there isn't one.

[ioquatix]

Sync do
  # It creates a reactor if required, otherwise it just executes the block.
end

The point of Sync is in places where you want a reactor, but don't need concurrency. In practice, most top level usage of Async is better written as Sync. It means synchronous execution with a reactor / event loop. I couldn't think of anything better and it seemed consistent with the intent.

[trevorturk]

Apologies because I think I'm misunderstanding the issue as initially described, but would using a top level Sync instead of Async produce the desired behavior? If not, would you mind trying to explain a bit more so I can try to understand more fully?

[ioquatix]

I think a top level Sync solves the problem, assuming the user is aware that it's required.

If they don't know, then the problem is, some internal Async block might not execute as desired - instead in that case, the request here is that it fails with an exception.

I think that as I outlined on the linked PR, I don't think it should be an error - if the user calls the top level function, probably it should have a top level Sync block... which avoids the error. But I don't know if this is an acceptable solution or not.

i.e. your proposal is, "Is this acceptable?":

def process_jobs(jobs)
  Sync do
    jobs
      .map { Async { _1.process! } }
      .map(&:wait)
  end
end

And I wonder if it is or not.

[ColinDKelley]

I just read up on Sync and found these two salient sentences:

This method will run your block in the current event loop if one exists, or create an event loop if not.
In other words, Sync{...} is very similar in behaviour to Async{...}.wait.

These tell me that Sync { ... } has the same overloading as Async { ... }... with the addition of the the implied .wait (but only in case 2!).

So my proposal logically extends here. The quoted sentences above would simplify to:

This method will run your block in the current reactor (event loop) and .wait on the result.
If there is no current reactor, Async::ReactorNotRunning: You must start a reactor first with 'Async.run { ... }' surrounding this code. will be raised.

[ColinDKelley]

BTW I spotted a mistake in my initial message for the exception. I had omitted the .run. I corrected it to:

Async::ReactorNotRunning: You must start a reactor first with Async.run { ... } surrounding this code.

[ColinDKelley]

@ioquatix

The point of Sync is in places where you want a reactor, but don't need concurrency. In practice, most top level usage of Async is better written as Sync. It means synchronous execution with a reactor / event loop.

To test my understanding:
Currently, at the top level, Async { ... } and Sync { ... } are identical, aren't they? Both spin up a reactor and block until it is finished?.

I find the "don't need concurrency" point confusing. Both offer concurrency inside their block, right? Concurrency with sibling blocks depends on what is enclosing this block and sibling blocks. Which gets us back to the importance, IMO, of using distinct APIs for these cases: spinning up a reactor vs. writing concurrent code inside a reactor.

[ColinDKelley]

@ioquatix Any thoughts on the above? ^^^

[ColinDKelley]

@trevorturk Now that I understand Sync semantics, I will try restate the problem to include it. Does this help?

The confusion arises from Async/Sync { ... } each being overloaded:

1. At the outer edge, Async/Sync { ... } starts up the reactor that must surround concurrent code. (There is no difference between Async and Sync here.)
2. Inside the reactor block, Async { ... } creates a concurrent Async::Task within the reactor. Sync { ... } does the same, plus it adds an implied .wait at the end.

When writing concurrent code (case 2), it is necessary that it be surrounded by a reactor (case 1). If that is missing, case 2 will be misconstrued as case 1, and it won't be concurrent at all. This is an easy mistake to make, especially when writing tests.

To address this risk, we like to follow Design by Contract, which in this case means that concurrent code (case 2) should ensure that there is a surrounding reactor before it proceeds.
...

Or, here's another way to restate the problem at a high level:

• To run concurrent code with async, it is necessary to create a surrounding reactor in advance of creating concurrent tasks.
• If you try to do that latter without doing the former, your code will not run concurrently as you intended. (Instead, it will run sequentially.)
• We can fix this by separating concerns: the API to create the surrounding reactor can be different from the API that creates concurrent tasks. This will simplify usage. Most importantly, it will allow us to follow Design by Contract: we can raise a helpful exception if someone tries to do the latter but didn't do the former.
• I noticed that this exactly parallels how EventMachine.next_tick et.al. will raise a helpful exception if a surrounding EventMachine.run has not been called at that point to start up the reactor. That was the inspiration to suggest Async.run as the API for starting up the reactor.

[trevorturk]

Thank you for the detailed explanation!

First, I think this helps clarify Sync for me, and makes me want to refactor my code to avoid Sync and instead use Async {}.wait simply because that might help my brain make better sense of things. (You're creating a reactor, and then waiting at the end of the block before continuing on... there's nothing much special about Sync... but then I see that Samuel is suggesting the otherwise, so I'm not quite settled here myself...)

Anyway, putting aside Async vs Sync, I did a simple test which I believe illustrates the issue you're pointing out in a way that might be helpful to others... or at least it helps me see the problem more clearly:

Async { puts "starting 1"; sleep 1; puts "1" }
Async { puts "starting 2"; sleep 2; puts "2" }

...produces:

starting 1
1
starting 2
2

..whereas:

Async do
  Async { puts "starting 1"; sleep 1; puts "1" }
  Async { puts "starting 2"; sleep 2; puts "2" }
end

...produces:

starting 1
starting 2
1
2

So, the issue is that, in the first example, you're not running the tasks concurrently and there is no warning/error, so it's an easy mistake to make.

If I'm understanding this correctly, I wonder if logging a warning might be enough... or perhaps having a global "strict" setting which would turn those warnings into errors? I see you're discussing more in-depth and perhaps better solutions here, but maybe there's room for a less controversial incremental change?

(I don't have a strong opinion here, I'm just trying to follow along, and hopefully this might be helpful in your discussion, and/or for future people stumbling across this issue!)

[ColinDKelley]

Thanks @trevorturk! That's a perfect illustration. It's exactly the trap that everyone on my team has stepped into at different times.

It's especially easy to have happen with specs. Let's suppose your code were in a method:

def sleep_1_and_2
  Async { puts "starting 1"; sleep 1; puts "1" }
  Async { puts "starting 2"; sleep 2; puts "2" }
end

The full program might correctly start the reactor around it:

Async do
  ...
  sleep_1_and_2
  ...
end

But the specs are often testing the pieces as independently as possible:

RSpec.describe 'sleep_1_and_2' do
  it 'calls sleep twice' do
    expect(Kernel).to receive(:sleep).with(1)
    expect(Kernel).to receive(:sleep).with(2)
    sleep_1_and_2
  end
end

The above spec will pass, but the code is not actually running concurrently as designed!

In my Proposal, the above mistake would cause a helpful exception to be raised:

Async::ReactorNotRunning: You must start a reactor first with `Async.run { ... }` surrounding this code.

and that would explain to us exactly what we needed to change to honor the contract:

RSpec.describe 'sleep_1_and_2' do
  it 'calls sleep twice' do
    expect(Kernel).to receive(:sleep).with(1)
    expect(Kernel).to receive(:sleep).with(2)
    Async.run do
      sleep_1_and_2
    end 
  end
end