Merge pull request #26 from scala-native/poll

Implement Poll abstraction

Author: lolgab

client/curl.scala

@@ -5,7 +5,6 @@ import scala.scalanative.libc.stdlib._
 import scala.scalanative.libc.string._
 import scala.concurrent._
 import scala.scalanative.runtime.Boxes
-import scala.scalanative.runtime.Intrinsics
 
 case class ResponseState(
     var code: Int = 200,
@@ -16,14 +15,11 @@ case class ResponseState(
 object Curl {
   import LibCurl._
   import LibCurlConstants._
-  import LibUV._
   import LibUVConstants._
 
   var serial = 0L
 
-  val loop                     = EventLoop.loop
-  var multi: MultiCurl         = null
-  val timerHandle: TimerHandle = malloc(uv_handle_size(UV_TIMER_T))
+  var multi: MultiCurl = null
 
   val requestPromises = mutable.Map[Long, Promise[ResponseState]]()
   val requests        = mutable.Map[Long, ResponseState]()
@@ -44,7 +40,6 @@ object Curl {
         multi_setopt_ptr(multi, TIMERFUNCTION, func_to_ptr(startTimerCB))
       println(s"timerCB: $startTimerCB")
 
-      check(uv_timer_init(loop, timerHandle), "uv_timer_init")
       initialized = true
       println("done")
     }
@@ -164,60 +159,52 @@ object Curl {
   val socketCB = new CurlSocketCallback {
     def apply(
         curl: Curl,
-        socket: Ptr[Byte],
+        socket: Int,
         action: Int,
         data: Ptr[Byte],
         socket_data: Ptr[Byte]
     ): Int = {
       println(s"socketCB called with action ${action}")
       val pollHandle = if (socket_data == null) {
         println(s"initializing handle for socket ${socket}")
-        val buf = malloc(uv_handle_size(UV_POLL_T)).asInstanceOf[Ptr[Ptr[Byte]]]
-        !buf = socket
-        check(uv_poll_init_socket(loop, buf, socket), "uv_poll_init_socket")
+        val poll = Poll(socket)
         check(
-          multi_assign(multi, socket, buf.asInstanceOf[Ptr[Byte]]),
+          multi_assign(multi, socket, poll.ptr),
           "multi_assign"
         )
-        buf
+        poll
       } else {
-        socket_data.asInstanceOf[Ptr[Ptr[Byte]]]
+        new Poll(socket_data)
       }
 
-      val events = action match {
-        case POLL_NONE   => None
-        case POLL_IN     => Some(UV_READABLE)
-        case POLL_OUT    => Some(UV_WRITABLE)
-        case POLL_INOUT  => Some(UV_READABLE | UV_WRITABLE)
-        case POLL_REMOVE => None
-      }
+      val readable = action == POLL_IN || action == POLL_INOUT
+      val writable = action == POLL_OUT || action == POLL_INOUT
 
-      events match {
-        case Some(ev) =>
-          println(s"starting poll with events $ev")
-          uv_poll_start(pollHandle, ev, pollCB)
-        case None =>
-          println("stopping poll")
-          uv_poll_stop(pollHandle)
-          startTimerCB(multi, 1, null)
+      if (readable || writable) {
+        println(
+          s"starting poll with readable = $readable and writable = $writable"
+        )
+        pollHandle.start(readable, writable) { (status, readable, writable) =>
+          println(
+            s"ready_for_curl fired with status $status and readable = $readable writable = $writable"
+          )
+          var actions = 0
+          if (readable) actions |= 1
+          if (writable) actions |= 2
+          val running_handles = stackalloc[Int]
+          val result =
+            multi_socket_action(multi, socket, actions, running_handles)
+          println("multi_socket_action", result)
+        }
+      } else {
+        println("stopping poll")
+        pollHandle.stop()
+        startTimerCB(multi, 1, null)
       }
       0
     }
   }
 
-  val pollCB = new PollCB {
-    def apply(pollHandle: PollHandle, status: Int, events: Int): Unit = {
-      println(
-        s"ready_for_curl fired with status ${status} and events ${events}"
-      )
-      val socket          = !(pollHandle.asInstanceOf[Ptr[Ptr[Byte]]])
-      val actions         = (events & 1) | (events & 2) // Whoa, nelly!
-      val running_handles = stackalloc[Int]
-      val result          = multi_socket_action(multi, socket, actions, running_handles)
-      println("multi_socket_action", result)
-    }
-  }
-
   val startTimerCB = new CurlTimerCallback {
     def apply(curl: MultiCurl, timeout_ms: Long, data: Ptr[Byte]): Int = {
       println(s"start_timer called with timeout ${timeout_ms} ms")
@@ -226,23 +213,19 @@ object Curl {
         1
       } else timeout_ms
       println("starting timer")
-      check(uv_timer_start(timerHandle, timeoutCB, time, 0), "uv_timer_start")
+      Timer.timeout(time) { () =>
+        println("in timeout callback")
+        val running_handles = stackalloc[Int]
+        multi_socket_action(multi, -1, 0, running_handles)
+        println(s"on_timer fired, ${!running_handles} sockets running")
+      }
       println("cleaning up requests")
       cleanup_requests()
       println("done")
       0
     }
   }
 
-  val timeoutCB = new TimerCB {
-    def apply(handle: TimerHandle): Unit = {
-      println("in timeout callback")
-      val running_handles = stackalloc[Int]
-      multi_socket_action(multi, int_to_ptr(-1), 0, running_handles)
-      println(s"on_timer fired, ${!running_handles} sockets running")
-    }
-  }
-
   def cleanup_requests(): Unit = {
     val messages                  = stackalloc[Int]
     val privateDataPtr            = stackalloc[Ptr[Long]]
@@ -299,14 +282,6 @@ object Curl {
     Boxes.boxToPtr[Byte](Boxes.unboxToCFuncRawPtr(f))
   }
 
-  def int_to_ptr(i: Int): Ptr[Byte] = {
-    Boxes.boxToPtr[Byte](Intrinsics.castIntToRawPtr(i))
-  }
-
-  def long_to_ptr(l: Long): Ptr[Byte] = {
-    Boxes.boxToPtr[Byte](Intrinsics.castLongToRawPtr(l))
-  }
-
 }
 
 object LibCurlConstants {
@@ -349,7 +324,7 @@ object LibCurlConstants {
 
   type CurlDataCallback = CFuncPtr4[Ptr[Byte], CSize, CSize, Ptr[Byte], CSize]
   type CurlSocketCallback =
-    CFuncPtr5[Curl, Ptr[Byte], CInt, Ptr[Byte], Ptr[Byte], CInt]
+    CFuncPtr5[Curl, CInt, CInt, Ptr[Byte], Ptr[Byte], CInt]
   type CurlTimerCallback = CFuncPtr3[MultiCurl, Long, Ptr[Byte], CInt]
 
   @name("curl_global_init")
@@ -408,14 +383,14 @@ object LibCurlConstants {
   @name("curl_multi_assign")
   def multi_assign(
       multi: MultiCurl,
-      socket: Ptr[Byte],
+      socket: Int,
       socket_data: Ptr[Byte]
   ): Int = extern
 
   @name("curl_multi_socket_action")
   def multi_socket_action(
       multi: MultiCurl,
-      socket: Ptr[Byte],
+      socket: Int,
       events: Int,
       numhandles: Ptr[Int]
   ): Int = extern

core/loop.scala renamed to core/src/main/scala/scala/scalanative/loop/Eventloop.scala

@@ -30,7 +30,7 @@ object EventLoop {
         @tailrec
         def runUv(): Unit = {
           val res = uv_run(loop, UV_RUN_ONCE)
-          if(res != 0) runUv()
+          if (res != 0) runUv()
         }
 
         scala.scalanative.runtime.loop()
@@ -49,7 +49,7 @@ object EventLoop {
 object LibUV {
   type UVHandle      = Ptr[Byte]
   type PipeHandle    = Ptr[Byte]
-  type PollHandle    = Ptr[Ptr[Byte]]
+  type PollHandle    = Ptr[Byte]
   type TCPHandle     = Ptr[Byte]
   type PrepareHandle = Ptr[Byte]
   type TimerHandle   = Ptr[Byte]
@@ -100,6 +100,8 @@ object LibUV {
   def uv_pipe_open(handle: PipeHandle, fd: Int): Int              = extern
   def uv_pipe_bind(handle: PipeHandle, socketName: CString): Int  = extern
 
+  def uv_poll_init(loop: Loop, handle: PollHandle, fd: Int): Int =
+    extern
   def uv_poll_init_socket(
       loop: Loop,
       handle: PollHandle,

core/src/main/scala/scala/scalanative/loop/Poll.scala

@@ -0,0 +1,45 @@
+package scala.scalanative.loop
+
+import scala.scalanative.libc.stdlib
+import LibUV._, LibUVConstants._
+import scala.scalanative.unsafe.Ptr
+import internals.HandleUtils
+
+@inline class Poll(val ptr: Ptr[Byte]) extends AnyVal {
+  def start(in: Boolean, out: Boolean)(
+      callback: (Int, Boolean, Boolean) => Unit
+  ): Unit = {
+    HandleUtils.setData(ptr, callback)
+    var events = 0
+    if (out) events |= UV_WRITABLE
+    if (in) events |= UV_READABLE
+    uv_poll_start(ptr, events, Poll.pollCB)
+  }
+
+  def stop(): Unit = {
+    uv_poll_stop(ptr)
+    HandleUtils.close(ptr)
+  }
+}
+
+object Poll {
+  private val pollCB = new PollCB {
+    def apply(handle: PollHandle, status: Int, events: Int): Unit = {
+      val callback =
+        HandleUtils.getData[(Int, Boolean, Boolean) => Unit](handle)
+      callback.apply(
+        status,
+        (events & UV_READABLE) != 0,
+        (events & UV_WRITABLE) != 0
+      )
+    }
+  }
+
+  private lazy val size = uv_handle_size(UV_POLL_T)
+
+  def apply(fd: Int): Poll = {
+    val pollHandle = stdlib.malloc(size)
+    uv_poll_init(EventLoop.loop, pollHandle, fd)
+    new Poll(pollHandle)
+  }
+}

core/src/main/scala/scala/scalanative/loop/Timer.scala

@@ -29,7 +29,11 @@ object Timer {
     }
   }
   @inline
-  private def startTimer(timeout: Long, repeat: Long, callback: () => Unit): Timer = {
+  private def startTimer(
+      timeout: Long,
+      repeat: Long,
+      callback: () => Unit
+  ): Timer = {
     val timerHandle = stdlib.malloc(uv_handle_size(UV_TIMER_T))
     uv_timer_init(EventLoop.loop, timerHandle)
     HandleUtils.setData(timerHandle, callback)

core/src/main/scala/scala/scalanative/loop/internals/HandleUtils.scala

@@ -7,18 +7,18 @@ import scala.scalanative.libc.stdlib
 import scala.collection.mutable
 import LibUV._
 
-private [loop] object HandleUtils {
+private[loop] object HandleUtils {
   private val references = mutable.Map.empty[Long, Int]
 
   @inline def getData[T <: Object](handle: Ptr[Byte]): T = {
-    val data = LibUV.uv_handle_get_data(handle)
+    val data   = LibUV.uv_handle_get_data(handle)
     val rawptr = castLongToRawPtr(data)
     castRawPtrToObject(rawptr).asInstanceOf[T]
   }
   @inline def setData[T <: Object](handle: Ptr[Byte], function: T): Unit = {
     val rawptr = castObjectToRawPtr(function)
-    val data = castRawPtrToLong(rawptr)
-    if(references.contains(data)) references(data) += 1
+    val data   = castRawPtrToLong(rawptr)
+    if (references.contains(data)) references(data) += 1
     else references(data) = 1
     LibUV.uv_handle_set_data(handle, data)
   }
@@ -29,9 +29,9 @@ private [loop] object HandleUtils {
   }
   @inline def close(handle: Ptr[Byte]): Unit = {
     uv_close(handle, onCloseCB)
-    val data = LibUV.uv_handle_get_data(handle)
+    val data    = LibUV.uv_handle_get_data(handle)
     val current = references(data)
-    if(current > 1) references(data) -= 1
+    if (current > 1) references(data) -= 1
     else references.remove(data)
   }
 }

core/src/test/scala/scala/scalanative/loop/TimerTests.scala

@@ -8,9 +8,9 @@ import scala.concurrent.Promise
 object TimerTests extends TestSuite {
   val tests = Tests {
     def now(): Duration = System.currentTimeMillis().millis
-    val d = 200.millis
+    val d               = 200.millis
     test("delay") {
-      var i = 0
+      var i         = 0
       val startTime = now()
       def checkDelay(time: Int) = {
         i += 1
@@ -25,13 +25,13 @@ object TimerTests extends TestSuite {
       } yield ()
     }
     test("repeat") {
-      var i = 0
-      val startTime = now()
-      val times = 3
-      val p = Promise[Unit]()
+      var i            = 0
+      val startTime    = now()
+      val times        = 3
+      val p            = Promise[Unit]()
       var timer: Timer = null.asInstanceOf[Timer]
       timer = Timer.repeat(d.toMillis) { () =>
-        if(i == times) {
+        if (i == times) {
           p.success(())
           timer.clear()
         } else i += 1
@@ -42,5 +42,14 @@ object TimerTests extends TestSuite {
         assert(took >= d * 3)
       }
     }
+    test("clear timeout") {
+      val handle = Timer.timeout(d.toMillis) { () =>
+        throw new Exception("This timeout should have not triggered")
+      }
+      handle.clear()
+      for {
+        () <- Timer.delay(d * 2)
+      } yield ()
+    }
   }
 }

scalajs-compat/RawTimers.scala

@@ -31,7 +31,10 @@ object RawTimers {
    *  @return A handle that can be used to cancel the timeout by passing it
    *          to [[clearTimeout]].
    */
-  @inline def setTimeout(handler: () => Unit, interval: Double): SetTimeoutHandle =
+  @inline def setTimeout(
+      handler: () => Unit,
+      interval: Double
+  ): SetTimeoutHandle =
     Timer.timeout(interval.toLong)(handler)
 
   /** Schedule `handler` for repeated execution every `interval`
@@ -42,6 +45,9 @@ object RawTimers {
    *  @return A handle that can be used to cancel the interval by passing it
    *          to [[clearInterval]].
    */
-  @inline def setInterval(handler: () => Unit, interval: Double): SetIntervalHandle =
+  @inline def setInterval(
+      handler: () => Unit,
+      interval: Double
+  ): SetIntervalHandle =
     Timer.repeat(interval.toLong)(handler)
 }

scalajs-compat/package.scala

@@ -26,8 +26,8 @@ package object timers {
 
   import scala.scalanative.loop.Timer
 
-  type SetTimeoutHandle = Timer 
-  type SetIntervalHandle = Timer 
+  type SetTimeoutHandle  = Timer
+  type SetIntervalHandle = Timer
 
   /** Schedule something for execution in `interval` milliseconds.
    *