Easy Racer implementation

[jackgene]

Hi there,
I have contributed an Easy Racer implementation using Gears.

While working on this, a couple of questions did come up which I'm hoping to be able to get answered here:

Cancellable HTTP client?

Many Easy Racer scenarios have been designed around being able to cancel HTTP requests, and it is unclear to me if Gears currently has a preferred solution to this. There's an example of this in the Supervision with Retries and Timeouts section of the document:

Async.blocking:
  withTimeout(60.seconds):
    val body = request.get("https://google.com")
    // ...

But it is unclear what HTTP client implementation is being used here, or if the request would get cancelled in the event of a timeout.

To get around this, I was able to wrap OkHttp in a Gears Cancellable like so, but I'm curious if there's a preferred/more idiomatic way I should be approaching this.

Cancellable blocking CPU-bound operations?

Easy Racer's Scenario 10 requires us to implement a CPU intensive operation that needs to be cancellable.

I found two ways to approach this:

1. Implement a custom Cancellable where the blocking operation periodically checks for cancellation (implemented as an AtomicBoolean here). I like this solution in that it fits in the cancellable model, but it does seem to require a fair bit of boilerplate.
2. Just implement the blocking operation directly in a Future, but introduce a suspension point by calling AsyncOperations.sleep(0) periodically. This is more succinct, but it may not be the most intuitive what the AsyncOperation.sleep(0) is meant to do (if this is meant to be the preferred solution, perhaps introducing an AsyncOperation.checkCancellation() might help make the intent clearer).

Again, I'm curious which is the preferred solution, or if I'm missing a completely different solution.

I'm also open to any feedback the team may have on this implementation in general.

Thank you!

[m8nmueller]

Hi Jack,
thanks for your input! Let me answer your questions one by one:

Cancellable HTTP client

For your HTTP client, I would suggest looking at gears.async.cancellationScope. The idea is to wrap long-running, but not awaiting code with a guard that can be used to cancel the operation inside. This could resemble this:

  cancellationScope(() => call.cancel()):
    val response = call.execute()
    cancellationScope(() => response.close()):
      response.body().string()

Note though that it should generally be avoided to have such long-running, not awaiting (especially: blocking) code, because it occupies a carrier thread. You should instead try to wrap the async API which does lend more to our idioms:

extension (call: Call)
  def asyncCall()(using Async): Try[Response] =
    Future
      .withResolver[Response]: resolver =>
        call.enqueue(
          new Callback:
            def onFailure(th: Throwable): Unit = resolver.reject(th)
            def onResponse(response: Response) = resolver.resolve(response)
        )
        resolver.onCancel(() => call.cancel())
      .link()
      .awaitResult

Cancellable blocking CPU-bound operations

I would suggest to use a separate thread pool for CPU-bound tasks, for similar reasons as outlined above (occupies carrier thread). It's just not something where Futures (or virtual threads) are good at. Again, you can expose the result via a withResolver Future. You could employ executor service task cancellation that maps to thread interruption of the worker. Then, if the worker has to explicitly check interruption, it would use Thread.currentThread().isInterrupted().

[jackgene]

Thanks for the quick feedback, Maximilian!

I'll attempt to implement these over the weekend.

I am curious about this:

Note though that it should generally be avoided to have such long-running, not awaiting (especially: blocking) code, because it occupies a carrier thread.

To make sure I understand, is the concern here because a Loom virtual thread, while lighter weight than a platform thread, is still relatively heavy, compared to things like Futures?

[m8nmueller]

Hi, sorry for being unclear. Futures are basically just virtual threads in the JVM implementation.

But no matter whether you take this implementation or the SCNative or even another (yet to be written) alternative, the idea is that these contain IO/async-bound code and schedulers (JVM-internal or custom) are optimized to do this. We should definitely make this more clear in the documentation, though, thanks for pointing out!

[jackgene]

... and schedulers (JVM-internal or custom) are optimized to do this.

To make sure I understand, did you mean to say "are not optimized to do this."?

[m8nmueller]

Schedulers are (or will be ..) optimized for tasks that do a bit of computing mixed with awaiting operations. As IO is usually somehow time-y in nature (and backends therefore blocking or async), we presume that IO is done using await (wrapping async APIs).

[jackgene]

I was going through my existing implementation, and found a bug in scenario 4

  def scenario4(scenarioUrl: Int => String)(using Async.Spawn): String =
    val url = scenarioUrl(4)
    def req = Future(scenarioRequest(url).execute().link().body().string())
    Seq(withTimeout(1.second)(req), req).awaitFirstWithCancel

Tests are passing some of the times, but it turns out that's because the default OkHttp timeout is set pretty low, and depending on which of the requests times out first, we may end up with the correct behavior.

If I understand correctly, I believe the issue is that withTimeout is not cancelling the operation it's guarding.

I was able to implement the timeout using awaitFirstWithCancel/orWithCancel:

    def req = Future:
      Try(scenarioRequest(url).execute().link().body().string())
    def timeout(duration: FiniteDuration) = Future:
      AsyncOperations.sleep(duration)
      Failure(TimeoutException())
    Seq(
      Future:
        Seq(req, timeout(1.second)).awaitFirstWithCancel.get,
      Future(req.await.get)
    ).awaitFirstWithCancel

But I'd like to know if there is a better way to accomplish this, as what I currently have it very verbose? Alternatively, do you have plans to add something like a withTimeoutWithCancel?

[m8nmueller]

Actually, withTimeout does cancel (as say its docs ;))

Just note that when you create a (Runnable, i.e., with Future.apply) Future, you get a handle to the concurrent computation, that is actively started in that very moment. So you are awaiting the same result twice.

Maybe, Task could help you ;)

[jackgene]

Ah! Good catch, I actually tried a few variants, but I see what I got wrong now. The following works:

  def scenario4(scenarioUrl: Int => String)(using Async.Spawn): String =
    val url = scenarioUrl(4)
    def req(using Async) = scenarioRequest(url).asyncGet.getResponseBody
    Seq(Future(withTimeout(1.second)(req)), Future(req)).awaitFirstWithCancel

[jackgene]

Hey @m8nmueller,
I've updated the implementation to use AsyncHttpClient as even though OkHttp has an async API, it isn't actually async (making 10k connections, I can see that it's firing up 10k platform threads). I'm still curious about the rationale for preferring async I/O though, given Gears appear to be build on Loom virtual threads.

Also, as mentioned above, I notice withTimeout isn't cancelling an operation upon timeout, do you have plans to add something like withTimeoutWithCancel? I tried implementing something along the lines of:

// Like `gears.async.withTimeout`, but cancels operation upon timeout
def withTimeoutWithCancel[T](timeout: FiniteDuration)(op: Async ?=> T)(using Async): T =
  Async.group:
    val timeoutFut = Future:
      AsyncOperations.sleep(timeout)
      Failure(TimeoutException())
    Seq(timeoutFut, Future(Try(op))).awaitFirstWithCancel.get

But it doesn't seem to be working (the operation isn't being cancelled).

[m8nmueller]

Oh, that’s sad. It looks like they don’t actually use platform’s async IO.

With the current implementation and on JVM only, it is therefore probably fine (performance/scheduling-wise). But it has a couple downsides:

• The calls and their users are not recognizable as suspending, so you should add a possibly unused (using Async)
• You should look out for non-Vthread compatible blocking stuff, such as native methods and object monitor synchronization (synchronized) in the HTTP library.
• Both of those points are also important when you switch the library.
• It is not portable (to SCNative or other non-VThread-based backends). One should at least have a very large warning about that. Furthermore, there could be implications with future versions of Gears, because of the assertion that no async task blocks or suspends secretly.

Because of the model-incompatibilities, the cancellation becomes a bit harder. You can see a similar scenario in our sleep implementation on JVM: https://github.com/lampepfl/gears/blob/main/jvm/src/main/scala/async/JvmAsyncOperations.scala . Being part of the well-known backend, we do exactly this kind of optimization that you have in mind.

Maybe, if OkHttp does consider that threading reasonable, you could follow them. Or take another library with real AIO; there’s even something for Gears in the making, but I don’t know details.