[ base ] Add non-blocking and timeout variants for channelGet

CHANGELOG_NEXT.md

@@ -254,6 +254,10 @@ This CHANGELOG describes the merged but unreleased changes. Please see [CHANGELO
 
 * Several functions like `pop`, `differenceMap` and `toSortedMap` were added to `Data.Sorted{Map|Set}`
 
+* Added `System.Concurrency.channelGetNonBlocking` for the chez backend.
+
+* Added `System.Concurrency.channelGetWithTimeout` for the chez backend.
+
 #### Contrib
 
 * `Data.List.Lazy` was moved from `contrib` to `base`.

libs/base/System/Concurrency.idr

@@ -72,19 +72,15 @@ data Condition : Type where [external]
 %foreign "scheme,racket:blodwen-make-cv"
          "scheme,chez:blodwen-make-condition"
 prim__makeCondition : PrimIO Condition
-
 %foreign "scheme,racket:blodwen-cv-wait"
          "scheme,chez:blodwen-condition-wait"
 prim__conditionWait : Condition -> Mutex -> PrimIO ()
-
 %foreign "scheme,chez:blodwen-condition-wait-timeout"
 --         "scheme,racket:blodwen-cv-wait-timeout"
 prim__conditionWaitTimeout : Condition -> Mutex -> Int -> PrimIO ()
-
 %foreign "scheme,racket:blodwen-cv-signal"
          "scheme,chez:blodwen-condition-signal"
 prim__conditionSignal : Condition -> PrimIO ()
-
 %foreign "scheme,racket:blodwen-cv-broadcast"
          "scheme,chez:blodwen-condition-broadcast"
 prim__conditionBroadcast : Condition -> PrimIO ()
@@ -183,10 +179,169 @@ barrierWait barrier = primIO (prim__barrierWait barrier)
 export
 data Channel : Type -> Type where [external]
 
+data ChannelObj : Type where [external]
+
+data ChannelSchemeObj : Type where
+  Null       : ChannelSchemeObj
+  Cons       : ChannelSchemeObj -> ChannelSchemeObj -> ChannelSchemeObj
+  IntegerVal : Integer -> ChannelSchemeObj
+  FloatVal   : Double -> ChannelSchemeObj
+  StringVal  : String -> ChannelSchemeObj
+  CharVal    : Char -> ChannelSchemeObj
+  Symbol     : String -> ChannelSchemeObj
+  Box        : ChannelSchemeObj -> ChannelSchemeObj
+  Vector     : Integer -> List ChannelSchemeObj -> ChannelSchemeObj
+  Procedure  : ChannelObj -> ChannelSchemeObj
+
+export
+interface Scheme a where
+  fromScheme : ChannelSchemeObj -> Maybe a
+
+export
+Scheme Integer where
+  fromScheme (IntegerVal x) = Just x
+  fromScheme _              = Nothing
+
+export
+Scheme Nat where
+  fromScheme (IntegerVal x) = Just $ integerToNat x
+  fromScheme _              = Nothing
+
+export
+Scheme Int where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme Int8 where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme Int16 where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme Int32 where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme Int64 where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme Bits8 where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme Bits16 where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme Bits32 where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme Bits64 where
+  fromScheme (IntegerVal x) = Just (cast x)
+  fromScheme _              = Nothing
+
+export
+Scheme String where
+  fromScheme (StringVal x) = Just x
+  fromScheme _             = Nothing
+
+export
+Scheme Double where
+  fromScheme (FloatVal x) = Just x
+  fromScheme _            = Nothing
+
+export
+Scheme Char where
+  fromScheme (CharVal x) = Just x
+  fromScheme _           = Nothing
+
+export
+Scheme Bool where
+  fromScheme (IntegerVal 0) = Just False
+  fromScheme (IntegerVal 1) = Just True
+  fromScheme _              = Nothing
+
+export
+Scheme a => Scheme (List a) where
+  fromScheme Null        = Just []
+  fromScheme (Cons x xs) = Just $ !(fromScheme x) :: !(fromScheme xs)
+  fromScheme _           = Nothing
+
+export
+(Scheme a, Scheme b) => Scheme (a, b) where
+  fromScheme (Cons x y) = Just (!(fromScheme x), !(fromScheme y))
+  fromScheme _          = Nothing
+
+export
+Scheme a => Scheme (Maybe a) where
+  fromScheme Null    = Just Nothing
+  fromScheme (Box x) = Just $ Just !(fromScheme x)
+  fromScheme _       = Nothing
+
+%foreign "scheme:blodwen-is-number"
+prim_isNumber : ChannelObj -> Int
+%foreign "scheme:blodwen-is-integer"
+prim_isInteger : ChannelObj -> Int
+%foreign "scheme:blodwen-is-float"
+prim_isFloat : ChannelObj -> Int
+%foreign "scheme:blodwen-is-char"
+prim_isChar : ChannelObj -> Int
+%foreign "scheme:blodwen-is-string"
+prim_isString : ChannelObj -> Int
+%foreign "scheme:blodwen-is-procedure"
+prim_isProcedure : ChannelObj -> Int
+%foreign "scheme:blodwen-is-symbol"
+prim_isSymbol : ChannelObj -> Int
+%foreign "scheme:blodwen-is-nil"
+prim_isNil : ChannelObj -> Int
+%foreign "scheme:blodwen-is-pair"
+prim_isPair : ChannelObj -> Int
+%foreign "scheme:blodwen-is-vector"
+prim_isVector : ChannelObj -> Int
+%foreign "scheme:blodwen-id"
+unsafeGetInteger : ChannelObj -> Integer
+%foreign "scheme:blodwen-id"
+unsafeGetString : ChannelObj -> String
+%foreign "scheme:blodwen-id"
+unsafeGetFloat : ChannelObj -> Double
+%foreign "scheme:blodwen-id"
+unsafeGetChar : ChannelObj -> Char
+%foreign "scheme:car"
+unsafeFst : ChannelObj -> ChannelObj
+%foreign "scheme:cdr"
+unsafeSnd : ChannelObj -> ChannelObj
+%foreign "scheme:blodwen-vector-ref"
+unsafeVectorRef : ChannelObj -> Integer -> ChannelObj
+%foreign "scheme:blodwen-unbox"
+unsafeUnbox : ChannelObj -> ChannelObj
+%foreign "scheme:blodwen-vector-length"
+unsafeVectorLength : ChannelObj -> Integer
+%foreign "scheme:blodwen-vector-list"
+unsafeVectorToList : ChannelObj -> List ChannelObj
+%foreign "scheme:blodwen-read-symbol"
+unsafeReadSymbol : ChannelObj -> String
+%foreign "scheme:blodwen-is-box"
+prim_isBox : ChannelObj -> Int
 %foreign "scheme:blodwen-make-channel"
 prim__makeChannel : PrimIO (Channel a)
 %foreign "scheme:blodwen-channel-get"
 prim__channelGet : Channel a -> PrimIO a
+%foreign "scheme:blodwen-channel-get-non-blocking"
+prim__channelGetNonBlocking : Channel a -> PrimIO ChannelObj
+%foreign "scheme:blodwen-channel-get-with-timeout"
+prim__channelGetWithTimeout : Channel a -> Nat -> PrimIO ChannelObj
 %foreign "scheme:blodwen-channel-put"
 prim__channelPut : Channel a -> a -> PrimIO ()
 
@@ -208,6 +363,68 @@ export
 channelGet : HasIO io => (chan : Channel a) -> io a
 channelGet chan = primIO (prim__channelGet chan)
 
+||| Non-blocking version of channelGet.
+|||
+||| @ chan the channel to receive on
+partial
+export
+channelGetNonBlocking : HasIO io => Scheme a => (chan : Channel a) -> io (Maybe a)
+channelGetNonBlocking chan =
+  pure $ (fromScheme . decodeObj) !(primIO (prim__channelGetNonBlocking chan))
+  where
+    decodeObj : ChannelObj -> ChannelSchemeObj
+    decodeObj obj =
+      if prim_isInteger obj == 1 then IntegerVal (unsafeGetInteger obj)
+      else if prim_isVector obj == 1 then Vector (unsafeGetInteger (unsafeVectorRef obj 0))
+                                                 (readVector (unsafeVectorLength obj) 1 obj)
+      else if prim_isPair obj == 1 then Cons (decodeObj (unsafeFst obj))
+                                             (decodeObj (unsafeSnd obj))
+      else if prim_isFloat obj == 1 then FloatVal (unsafeGetFloat obj)
+      else if prim_isString obj == 1 then StringVal (unsafeGetString obj)
+      else if prim_isChar obj == 1 then CharVal (unsafeGetChar obj)
+      else if prim_isSymbol obj == 1 then Symbol (unsafeReadSymbol obj)
+      else if prim_isProcedure obj == 1 then Procedure obj
+      else if prim_isBox obj == 1 then Box (decodeObj (unsafeUnbox obj))
+      else Null
+      where
+        readVector : Integer -> Integer -> ChannelObj -> List ChannelSchemeObj
+        readVector len i obj
+          = if len == i
+              then []
+              else decodeObj (unsafeVectorRef obj i) :: readVector len (i + 1) obj
+
+||| Timeout version of channelGet.
+||| Continously loops with 1ms delays until `seconds` has elapsed, or a value is provided through `chan`.
+|||
+||| @ chan the channel to receive on
+||| @ seconds how many seconds to wait until timeout
+partial
+export
+channelGetWithTimeout : HasIO io => Scheme a => (chan : Channel a) -> (seconds : Nat) -> io (Maybe a)
+channelGetWithTimeout chan seconds =
+  pure $ (fromScheme . decodeObj) !(primIO (prim__channelGetWithTimeout chan seconds))
+  where
+    decodeObj : ChannelObj -> ChannelSchemeObj
+    decodeObj obj =
+      if prim_isInteger obj == 1 then IntegerVal (unsafeGetInteger obj)
+      else if prim_isVector obj == 1 then Vector (unsafeGetInteger (unsafeVectorRef obj 0))
+                                                 (readVector (unsafeVectorLength obj) 1 obj)
+      else if prim_isPair obj == 1 then Cons (decodeObj (unsafeFst obj))
+                                             (decodeObj (unsafeSnd obj))
+      else if prim_isFloat obj == 1 then FloatVal (unsafeGetFloat obj)
+      else if prim_isString obj == 1 then StringVal (unsafeGetString obj)
+      else if prim_isChar obj == 1 then CharVal (unsafeGetChar obj)
+      else if prim_isSymbol obj == 1 then Symbol (unsafeReadSymbol obj)
+      else if prim_isProcedure obj == 1 then Procedure obj
+      else if prim_isBox obj == 1 then Box (decodeObj (unsafeUnbox obj))
+      else Null
+      where
+        readVector : Integer -> Integer -> ChannelObj -> List ChannelSchemeObj
+        readVector len i obj
+          = if len == i
+              then []
+              else decodeObj (unsafeVectorRef obj i) :: readVector len (i + 1) obj
+
 ||| Puts a value on the given channel.
 |||
 ||| @ chan the `Channel` to send the value over

support/chez/support.ss

@@ -439,6 +439,51 @@
     (condition-signal read-cv)
     the-val))
 
+(define (blodwen-channel-get-non-blocking ty chan)
+  (if (mutex-acquire (channel-read-mut chan) #f)
+    (let* ([val-box  (channel-val-box  chan)]
+           [read-box (channel-read-box chan)]
+           [read-cv  (channel-read-cv  chan)]
+           [the-val  (unbox val-box)]
+          )
+      (if (null? the-val)
+          (begin
+           (mutex-release (channel-read-mut chan))
+           '())
+          (begin
+            (set-box! val-box '())
+            (set-box! read-box #t)
+            (mutex-release (channel-read-mut chan))
+            (condition-signal read-cv)
+            the-val)
+      ))
+  '()))
+
+(define (blodwen-channel-get-with-timeout ty chan seconds)
+  (let loop ([start-time (current-time)]
+            )
+    (let ([elapsed (time-difference (current-time) start-time)])
+      (if (time>=? elapsed (make-time 'time-duration 0 seconds))
+          '()
+          (if (mutex-acquire (channel-read-mut chan) #f)
+              (let* ([val-box  (channel-val-box chan)]
+                     [the-val  (unbox val-box)])
+                (if (null? the-val)
+                    (begin
+                      (mutex-release (channel-read-mut chan))
+                      (sleep (make-time 'time-duration 1000000 0))
+                      (loop start-time))
+                    (let* ([read-box (channel-read-box chan)]
+                           [read-cv  (channel-read-cv chan)])
+                      (set-box! val-box '())
+                      (set-box! read-box #t)
+                      (mutex-release (channel-read-mut chan))
+                      (condition-signal read-cv)
+                      the-val)))
+              (begin
+                (sleep (make-time 'time-duration 1000000 0))
+                (loop start-time)))))))
+
 ;; Mutex
 
 (define (blodwen-make-mutex)
@@ -499,7 +544,6 @@
 (define (blodwen-clock-second time) (time-second time))
 (define (blodwen-clock-nanosecond time) (time-nanosecond time))
 
-
 (define (blodwen-arg-count)
   (length (command-line)))
 

tests/allschemes/channels007/Main.idr

@@ -0,0 +1,23 @@
+import System
+import System.Concurrency
+
+-- Test that using channelGetNonBlocking works as expected.
+main : IO ()
+main = do
+  chan <- makeChannel
+  threadID <- fork $ do
+    channelPut chan "Hello"
+    channelPut chan "Goodbye"
+  sleep 1
+  case !(channelGetNonBlocking chan) of
+    Nothing   =>
+      putStrLn "Nothing"
+    Just val' =>
+      putStrLn val'
+  case !(channelGetNonBlocking chan) of
+    Nothing   =>
+      putStrLn "Nothing"
+    Just val' =>
+      putStrLn val'
+  sleep 1
+

tests/allschemes/channels007/expected

@@ -0,0 +1,2 @@
+Hello
+Goodbye

tests/allschemes/channels007/run

@@ -0,0 +1,3 @@
+. ../../testutils.sh
+
+run Main.idr

tests/allschemes/channels008/Main.idr

@@ -0,0 +1,31 @@
+import System
+import System.Concurrency
+
+-- Simple producing thread.
+producer : Channel Nat -> Nat -> IO ()
+producer c n = ignore $ producer' n
+  where
+    producer' : Nat -> IO ()
+    producer' Z     = pure ()
+    producer' (S n) = do
+      channelPut c n
+      sleep 1
+
+-- Test that channelGetWithTimeout works as expected.
+main : IO ()
+main =
+  do c    <- makeChannel
+     tids <- for [0..11] $ \n => fork $ producer c n
+     vals <- for [0..11] $ \_ => channelGetWithTimeout c 5
+     ignore $ traverse (\t => threadWait t) tids
+     let vals' = map (\val => case val of
+                        Nothing   =>
+                          0
+                        Just val' =>
+                          val'
+                     ) vals
+         s     = sum vals'
+     if s == 55
+        then putStrLn "Success!"
+        else putStrLn "How did we get here?"
+

tests/allschemes/channels008/expected

@@ -0,0 +1 @@
+Success!

tests/allschemes/channels008/run

@@ -0,0 +1,3 @@
+. ../../testutils.sh
+
+run Main.idr