Add asynchronous curl backends for cats-effect and fs2 (curl_multi_socket_action)

build.sbt

@@ -432,7 +432,10 @@ lazy val cats = (projectMatrix in file("effects/cats"))
   )
   .nativePlatform(
     scalaVersions = scala2And3,
-    settings = commonNativeSettings ++ testServerSettings
+    settings = commonNativeSettings ++ testServerSettings ++ Seq(
+      // the async curl backend uses IO and the runtime's FileDescriptorPoller (epoll/kqueue)
+      libraryDependencies += "org.typelevel" %% "cats-effect" % catsEffect_3_version
+    )
   )
 
 lazy val fs2Ce2 = (projectMatrix in file("effects/fs2-ce2"))

core/src/main/resources/scala-native/ffi.c

@@ -9,6 +9,25 @@ const char* sttp_curl_get_version() {
 }
 int sttp_curl_getinfo_pointer(CURL *curl, CURLINFO info, void* arg) {return curl_easy_getinfo(curl, info, arg); }
 
+/* curl_multi_setopt is variadic, so it can't be bound directly */
+int sttp_curl_multi_setopt_pointer(CURLM *multi, CURLMoption opt, void *arg) { return curl_multi_setopt(multi, opt, arg); }
+int sttp_curl_multi_setopt_long(CURLM *multi, CURLMoption opt, long arg) { return curl_multi_setopt(multi, opt, arg); }
+
+/* Non-blocking readiness check of a socket: want = 1 (read), 2 (write), 3 (both).
+   Returns non-zero if the socket is ready (or in an error/hangup state). Used to drain edge-triggered notifications. */
+#ifdef _WIN32
+int sttp_curl_fd_ready(int fd, int want) { return 1; }
+#else
+#include <poll.h>
+int sttp_curl_fd_ready(int fd, int want) {
+    struct pollfd p;
+    p.fd = fd;
+    p.events = ((want & 1) ? POLLIN : 0) | ((want & 2) ? POLLOUT : 0);
+    p.revents = 0;
+    return poll(&p, 1, 0) > 0 && p.revents != 0;
+}
+#endif
+
 /* curl_multi wrappers */
 /* Most curl_multi functions are called directly via @extern bindings.
    This wrapper exists because CURLMsg contains a union that can't be

core/src/main/scalanative/sttp/client4/curl/internal/CCurl.scala

@@ -118,4 +118,17 @@ private[curl] trait CCurl {
   // Via C wrapper (union extraction) — see ffi.c
   @name("sttp_curl_multi_info_read_result")
   def multiInfoReadResult(multi: Ptr[CurlM], easy_out: Ptr[Ptr[Curl]]): CInt = extern
+
+  // Multi socket interface
+  @name("curl_multi_socket_action")
+  def multiSocketAction(multi: Ptr[CurlM], sockfd: CInt, evBitmask: CInt, runningHandles: Ptr[CInt]): CInt = extern
+
+  @name("sttp_curl_multi_setopt_pointer")
+  def multiSetoptPtr(multi: Ptr[CurlM], option: CInt, parameter: Ptr[_]): CInt = extern
+
+  @name("sttp_curl_multi_setopt_long")
+  def multiSetoptLong(multi: Ptr[CurlM], option: CInt, parameter: Long): CInt = extern
+
+  @name("sttp_curl_fd_ready")
+  def fdReady(fd: CInt, want: CInt): CInt = extern
 }

docs/backends/native/curl.md

@@ -64,4 +64,58 @@ object Main extends ZIOAppDefault:
     res <- basicRequest.get(uri"http://httpbin.org/ip").send(backend)
     _ <- Console.printLine(res)    
   yield ()
-```
+```
+
+## Cats Effect-based (asynchronous)
+
+To use in an sbt project, add the following dependency:
+
+```
+"com.softwaremill.sttp.client4" %%% "cats" % @VERSION@
+```
+
+`CurlCatsAsyncBackend` is a non-blocking backend, which uses libcurl's
+[multi socket interface](https://curl.se/libcurl/c/libcurl-multi.html) (`curl_multi_socket_action`). libcurl reports
+which sockets it wants to wait on and for how long; the backend delegates the waiting to the Cats Effect runtime's
+polling system (epoll on Linux, kqueue on macOS). No thread is blocked on network I/O, and curl shares the event loop
+with the rest of the application.
+
+```scala
+//> using platform native
+//> using scala 3
+//> using dep com.softwaremill.sttp.client4::cats::@VERSION@
+
+import cats.effect.{IO, IOApp}
+import sttp.client4.*
+import sttp.client4.curl.cats.CurlCatsAsyncBackend
+
+object Main extends IOApp.Simple:
+  def run: IO[Unit] =
+    CurlCatsAsyncBackend.resource().use { backend =>
+      basicRequest.get(uri"http://httpbin.org/ip").send(backend).flatMap(IO.println)
+    }
+```
+
+`CurlCatsAsyncBackend()` creates a backend directly; it should be closed after use. The `resource()` variant closes it
+for you.
+
+### Threading
+
+libcurl multi handles are not thread-safe, so each one is guarded by a fiber-aware mutex (fibers waiting for it are suspended, no thread is
+blocked), held only for the short, non-blocking
+libcurl calls. When Scala Native multithreading is enabled, requests are spread over several multi handles to avoid
+contention; the number can be set with the `parallelism` parameter (defaults to the number of available processors, or
+`1` in single-threaded mode):
+
+```scala
+CurlCatsAsyncBackend.resource(parallelism = 2)
+```
+
+### Requirements
+
+The backend relies on the `FileDescriptorPoller` of the runtime. The default `IORuntime` on Scala Native provides one
+(epoll/kqueue); if a custom runtime doesn't, each transfer is driven from the blocking thread pool instead.
+
+Limitations: WebSockets and `Stream`-based responses are not supported, as with the other curl backends. Responses read
+using `asInputStream` work, but they are driven by the shared blocking implementation (`curl_multi_perform` +
+`curl_multi_poll` on the calling thread), so they are not asynchronous.

effects/cats/src/main/scalanative/sttp/client4/curl/cats/CurlCatsAsyncBackend.scala

@@ -0,0 +1,101 @@
+package sttp.client4.curl.cats
+
+import cats.effect.{FileDescriptorPoller, IO, Resource}
+import sttp.client4.Backend
+import sttp.client4.curl.AbstractCurlBackend
+import sttp.client4.curl.internal.CurlApi._
+import sttp.client4.curl.internal.{CurlApi, CurlCode, CurlMCode}
+import sttp.client4.impl.cats.CatsMonadError
+import sttp.client4.wrappers.FollowRedirectsBackend
+
+import java.util.concurrent.atomic.AtomicInteger
+import scala.scalanative.libc.stdlib
+import scala.scalanative.meta.LinktimeInfo
+import scala.scalanative.unsafe._
+import scala.scalanative.unsigned._
+
+/** A curl backend which never blocks a thread on network I/O.
+  *
+  * It uses libcurl's multi socket interface (`curl_multi_socket_action`): libcurl reports which sockets it wants to
+  * wait on and for how long, and the backend delegates the waiting to the cats-effect runtime's polling system
+  * (epoll on Linux, kqueue on macOS), so curl shares the event loop with the rest of the application.
+  *
+  * libcurl multi handles aren't thread-safe, so each handle is guarded by a fiber-aware mutex (waiting fibers are
+  * suspended, no thread is blocked), which is only held for the short, non-blocking libcurl calls. In multi-threaded mode, requests are spread over `parallelism` multi handles to avoid
+  * contention; in single-threaded mode, a single handle is used.
+  *
+  * If the runtime has no [[FileDescriptorPoller]] (e.g. a custom `IORuntime` with the default sleep-based polling
+  * system), the backend falls back to driving each transfer from the blocking thread pool.
+  */
+class CurlCatsAsyncBackend private (drivers: Array[CurlMultiDriver], verbose: Boolean)
+    extends AbstractCurlBackend[IO](new CatsMonadError[IO], verbose)
+    with Backend[IO] {
+
+  private val next = new AtomicInteger(0)
+
+  override protected def performCurl(c: CurlHandle): IO[CurlCode.CurlCode] =
+    FileDescriptorPoller.find.flatMap {
+      case Some(poller) =>
+        val driver = drivers(Math.floorMod(next.getAndIncrement(), drivers.length))
+        driver.perform(poller, c).map(CurlCode(_))
+      case None => CurlCatsAsyncBackend.performBlocking(c)
+    }
+
+  override def close(): IO[Unit] = drivers.foldLeft(IO.unit)(_ *> _.close)
+}
+
+object CurlCatsAsyncBackend {
+
+  /** The default number of multi handles: one per available processor in multi-threaded mode, one otherwise. */
+  def defaultParallelism: Int =
+    if (LinktimeInfo.isMultithreadingEnabled) Math.max(1, Runtime.getRuntime.availableProcessors()) else 1
+
+  /** Creates a backend, which is closed when the resource is released.
+    *
+    * @param verbose
+    *   If true, logs request and response summary to the console.
+    * @param parallelism
+    *   The number of libcurl multi handles to spread the requests over.
+    */
+  def resource(verbose: Boolean = false, parallelism: Int = defaultParallelism): Resource[IO, Backend[IO]] =
+    Resource.make(IO(apply(verbose, parallelism)))(_.close())
+
+  /** Creates a backend. It should be closed after use. */
+  def apply(verbose: Boolean = false, parallelism: Int = defaultParallelism): Backend[IO] = {
+    require(parallelism > 0, "parallelism must be positive")
+    FollowRedirectsBackend(new CurlCatsAsyncBackend(Array.fill(parallelism)(CurlMultiDriver()), verbose))
+  }
+
+  /** Fallback when there's no poller: a dedicated multi handle per transfer, driven on the blocking pool in short
+    * slices, so that the fiber stays cancelable.
+    */
+  private def performBlocking(c: CurlHandle): IO[CurlCode.CurlCode] =
+    IO {
+      val multi = CurlApi.multiInit
+      val running = stdlib.calloc(1.toUSize, sizeof[CInt]).asInstanceOf[Ptr[CInt]]
+      val rc = multi.addHandle(c)
+      (multi, running, rc)
+    }.bracket { case (multi, running, rc) =>
+      def loop: IO[CurlCode.CurlCode] =
+        IO.blocking {
+          val pc = multi.perform(running)
+          if (pc != CurlMCode.Ok) Some(CurlCode.FailedInit)
+          else if (!running == 0) Some(CurlCode(Math.max(multi.infoReadResult(null), 0)))
+          else {
+            val _ = multi.poll(50, null)
+            None
+          }
+        }.flatMap {
+          case Some(code) => IO.pure(code)
+          case None       => IO.cede *> loop
+        }
+      if (rc != CurlMCode.Ok) IO.raiseError(new RuntimeException(s"curl_multi_add_handle failed with $rc"))
+      else loop
+    } { case (multi, running, _) =>
+      IO {
+        val _ = multi.removeHandle(c)
+        multi.cleanup()
+        stdlib.free(running.asInstanceOf[Ptr[Byte]])
+      }
+    }
+}