Merge pull request #15 from scala-native/use-scala-native-loop

Use default Scala Native ExecutionContext instead of custom

Author: lolgab

README.md

@@ -6,7 +6,8 @@ Extensible event loop and async-oriented IO for Scala Native; powered by libuv.
 If you're looking for the new 0.4 rewrite, check the `04` branch.  The current state of master is mostly extracted from the book [Modern Systems Programming in Scala Native](https://pragprog.com/book/rwscala/modern-systems-programming-with-scala-native).
 
 ## What is it?
-native-loop provides a real, asynchronous ExecutionContext/Future implementation for Scala Native.
+
+scala-native-loop provides a real, asynchronous ExecutionContext implementation for Scala Native.
 It's backed by libuv, the same C library that the node.js ecosystem runs on; in addition to basic 
 Future dispatching, we can also use libuv to provide other basic functionality, like:
 
@@ -19,7 +20,7 @@ Future dispatching, we can also use libuv to provide other basic functionality,
 To provide a working API for practical, async Scala Native programs, we have two subprojects,
 `client` and `server`, which provide an async HTTP client and server, respectively, by integrating addtional C libraries: [nodejs/http-parser](https://github.com/nodejs/http-parser) for request parsing, and [curl](https://github.com/curl/curl) for a full featured client with HTTPS support.
 
-That said - providing a full-featured ecosystem in a single library isn't feasible - instead, we provide a `LoopExtension` trait that allows other C libaries to be integrated to the underlying event loop, in the same way that libcurl and http-parser are integrated; this opens up the possiblity of fully asynchronous bindings for postgres, redis, and many others.
+That said - providing a full-featured ecosystem in a single library isn't feasible - instead, we provide a `LoopExtension` trait that allows other C libraries to be integrated to the underlying event loop, in the same way that libcurl and http-parser are integrated; this opens up the possiblity of fully asynchronous bindings for postgres, redis, and many others.
 
 ## Why is this here?
 

build.sbt

@@ -3,11 +3,11 @@ licenses := Seq(
   "Apache 2.0" -> url("http://www.apache.org/licenses/LICENSE-2.0")
 )
 publishMavenStyle := true
-publishArtifact in Test := false
+Test / publishArtifact := false
 pomIncludeRepository := { _ =>
   false
 }
-publishTo in ThisBuild := {
+ThisBuild / publishTo := {
   val nexus = "https://oss.sonatype.org/"
   if (isSnapshot.value)
     Some("snapshots" at nexus + "content/repositories/snapshots")
@@ -34,58 +34,60 @@ lazy val commonSettings = Seq(
   version := "0.1.1-SNAPSHOT",
   scalaVersion := "2.11.12",
   scalacOptions ++= Seq(
-    "-feature"
+    "-feature",
+    "-Ywarn-unused-import",
+    "-Xfatal-warnings"
   ),
-  skip in publish := true,
-  skip in publishLocal := true
+  publish / skip := true,
+  publishLocal / skip := true
 )
 
-lazy val core = (project in file("core"))
+lazy val core = project.in(file("core"))
   .settings(name := "native-loop-core")
-  .settings(commonSettings: _*)
-  .settings(skip in publish := false)
-  .settings(skip in publishLocal := false)
+  .settings(commonSettings)
+  .settings(publish / skip := false)
+  .settings(publishLocal / skip := false)
   .enablePlugins(ScalaNativePlugin)
 
-lazy val pipe = (project in file("pipe"))
-  .settings(commonSettings: _*)
+lazy val pipe = project.in(file("pipe"))
+  .settings(commonSettings)
   .enablePlugins(ScalaNativePlugin)
   .dependsOn(core)
 
-lazy val client = (project in file("client"))
-  .settings(commonSettings: _*)
+lazy val client = project.in(file("client"))
+  .settings(commonSettings)
   .enablePlugins(ScalaNativePlugin)
   .dependsOn(core)
 
-lazy val server = (project in file("server"))
-  .settings(commonSettings: _*)
+lazy val server = project.in(file("server"))
+  .settings(commonSettings)
   .enablePlugins(ScalaNativePlugin)
   .dependsOn(core)
 
-lazy val scalaJsCompat = (project in file("scalajs-compat"))
+lazy val scalaJsCompat = project.in(file("scalajs-compat"))
   .settings(name := "native-loop-js-compat")
-  .settings(commonSettings: _*)
-  .settings(skip in publish := false)
-  .settings(skip in publishLocal := false)
+  .settings(commonSettings)
+  .settings(publish / skip := false)
+  .settings(publishLocal / skip := false)
   .enablePlugins(ScalaNativePlugin)
   .dependsOn(core)
 
-lazy val serverExample = (project in file("examples/server"))
-  .settings(commonSettings: _*)
+lazy val serverExample = project.in(file("examples/server"))
+  .settings(commonSettings)
   .enablePlugins(ScalaNativePlugin)
   .dependsOn(core, server, client)
 
-lazy val pipeExample = (project in file("examples/pipe"))
-  .settings(commonSettings: _*)
+lazy val pipeExample = project.in(file("examples/pipe"))
+  .settings(commonSettings)
   .enablePlugins(ScalaNativePlugin)
   .dependsOn(core, pipe, client)
 
-lazy val curlExample = (project in file("examples/curl"))
-  .settings(commonSettings: _*)
+lazy val curlExample = project.in(file("examples/curl"))
+  .settings(commonSettings)
   .enablePlugins(ScalaNativePlugin)
   .dependsOn(core, client)
 
-lazy val timerExample = (project in file("examples/timer"))
-  .settings(commonSettings: _*)
+lazy val timerExample = project.in(file("examples/timer"))
+  .settings(commonSettings)
   .enablePlugins(ScalaNativePlugin)
   .dependsOn(core)

client/curl.scala

@@ -1,6 +1,5 @@
 package scala.scalanative.loop
 import scala.scalanative.unsafe._
-import scala.scalanative.unsigned._
 import scala.collection.mutable
 import scala.scalanative.libc.stdlib._
 import scala.scalanative.libc.string._

core/loop.scala

@@ -1,77 +1,38 @@
 package scala.scalanative.loop
 import scala.scalanative.unsafe._
-import scala.scalanative.libc.stdlib
-
-import scala.collection.mutable.ListBuffer
-import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
-import scala.concurrent.Future
-import scala.concurrent.Promise
-import scala.util.{Try, Success}
-import scala.Option
-import LibUV.Buffer
-import LibUVConstants.check
-
-object EventLoop extends ExecutionContextExecutor {
-  import LibUV._, LibUVConstants._
-
-  val loop              = uv_default_loop()
-  private val taskQueue = ListBuffer[Runnable]()
-  val handle            = stdlib.malloc(uv_handle_size(UV_PREPARE_T))
 
-  private def initDispatcher(loop: LibUV.Loop): PrepareHandle = {
-    check(uv_prepare_init(loop, handle), "uv_prepare_init")
-    return handle
-  }
-
-  val prepareCallback = new PrepareCB {
-    def apply(handle: PrepareHandle) = {
-      while (taskQueue.nonEmpty) {
-        val runnable = taskQueue.remove(0)
-        try {
-          runnable.run()
-        } catch {
-          case t: Throwable => reportFailure(t)
-        }
-      }
-      if (taskQueue.isEmpty) {
-        LibUV.uv_prepare_stop(handle)
-      }
-    }
-  }
+object EventLoop {
+  import LibUV._, LibUVConstants._
 
-  private val dispatcher = initDispatcher(loop)
+  val loop: LibUV.Loop = uv_default_loop()
 
   // Schedule loop execution after main ends
   scalanative.runtime.ExecutionContext.global.execute(
     new Runnable {
+      /**
+        * This is the implementation of the event loop
+        * that integrates with libuv. The logic is the
+        * following:
+        * - First we run all Scala futures in the default
+        *   execution context
+        * - Then in loop:
+        *   - we check if they generated IO calls on
+        *     the event loop
+        *   - If it's the case we run libuv's event loop
+        *     using UV_RUN_ONCE that blocks only once
+        *   - We run the default execution context again
+        *     in case the callbacks generated new Futures
+        */
       def run(): Unit = {
-        val returnCode = EventLoop.run()
-        if (returnCode != 0) {
-          Zone { implicit z =>
-            System.err.println(fromCString(uv_err_name(returnCode)))
-          }
-          System.exit(returnCode)
+        scala.scalanative.runtime.loop()
+        while (uv_loop_alive(loop) != 0) {
+          uv_run(loop, UV_RUN_ONCE)
+          scala.scalanative.runtime.loop()
         }
+        uv_loop_close(loop)
       }
     }
   )
-
-  def execute(runnable: Runnable): Unit = {
-    taskQueue += runnable
-    check(uv_prepare_start(handle, prepareCallback), "uv_prepare_start")
-  }
-
-  def reportFailure(t: Throwable): Unit = {
-    t.printStackTrace()
-  }
-
-  def run(mode: Int = UV_RUN_DEFAULT): Int = {
-    var continue = 1
-    while (continue != 0) {
-      continue = uv_run(loop, mode)
-    }
-    continue
-  }
 }
 
 @link("uv")
@@ -103,6 +64,8 @@ object LibUV {
 
   def uv_default_loop(): Loop                                     = extern
   def uv_loop_size(): CSize                                       = extern
+  def uv_loop_alive(loop: Loop): CInt                             = extern
+  def uv_loop_close(loop: Loop): CInt                             = extern
   def uv_is_active(handle: Ptr[Byte]): Int                        = extern
   def uv_handle_size(h_type: Int): CSize                          = extern
   def uv_req_size(r_type: Int): CSize                             = extern

core/timer.scala

@@ -2,11 +2,8 @@ package scala.scalanative.loop
 import scala.scalanative.unsafe._
 import scala.scalanative.libc.stdlib
 import scala.collection.mutable
-import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
 import scala.concurrent.Future
 import scala.concurrent.Promise
-import scala.util.{Try, Success}
-import scala.Option
 import scala.concurrent.duration._
 import LibUV._, LibUVConstants._
 

examples/curl/main.scala

@@ -1,16 +1,11 @@
 import scala.scalanative.loop._
-import scala.concurrent._
-import scala.concurrent.duration._
 import LibCurlConstants._
+import scala.concurrent.ExecutionContext.Implicits.global
 
 object Main {
-  implicit val ec: ExecutionContext = EventLoop
-
   def main(args: Array[String]): Unit = {
     Curl.startRequest(GET, "http://www.example.com", Seq()).map { response =>
       println(s"got response: $response")
     }
-
-    EventLoop.run()
   }
 }

examples/server/main.scala

@@ -1,10 +1,6 @@
 import scala.scalanative.loop._
-import scala.concurrent._
-import scala.concurrent.duration._
 
 object Main {
-  implicit val ec: ExecutionContext = EventLoop
-
   def main(args: Array[String]): Unit = {
     Server.init(9999) { (r, c) =>
       println(s"received request $r on connection $c")
@@ -16,7 +12,5 @@ object Main {
         "hello!"
       )
     }
-
-    EventLoop.run()
   }
 }

examples/timer/main.scala

@@ -1,10 +1,8 @@
 import scala.scalanative.loop._
-import scala.concurrent._
 import scala.concurrent.duration._
+import scala.concurrent.ExecutionContext.Implicits.global
 
 object Main {
-  implicit val ec: ExecutionContext = EventLoop
-
   def main(args: Array[String]): Unit = {
     Timer
       .delay(3.seconds)
@@ -19,7 +17,5 @@ object Main {
       .onComplete { _ =>
         println("done")
       }
-
-    EventLoop.run()
   }
 }

pipe/pipe.scala

@@ -1,15 +1,13 @@
 package scala.scalanative.loop
 import scala.scalanative.unsafe._
 import scala.scalanative.libc.stdlib
-import scala.scalanative.libc.string
 
 import scala.collection.mutable
-import scala.util.{Try, Success, Failure}
-import scala.concurrent.{Future, ExecutionContext}
+import scala.concurrent.Future
 import scala.concurrent.{Promise}
 
 case class Handle(serial: Long, handle: Ptr[Byte]) {
-  import LibUV._, LibUVConstants._
+  import LibUV._
 
   def stream(
       itemHandler: StreamIO.ItemHandler,

scalajs-compat/Handles.scala

@@ -12,7 +12,6 @@
 
 package scala.scalajs.js.timers
 
-import scalanative.unsafe.Zone
 import scalanative.loop.LibUV.TimerHandle
 
 /** <span class="badge badge-non-std" style="float: right;">Non-Standard</span>