Patch CVs and sleep in Racket (#1059)

CHANGELOG.md

@@ -13,6 +13,32 @@ REPL/IDE mode changes:
 * Added `:search` command, which searches for functions by type
 * `:load`/`:l` and `:cd` commands now only accept paths surrounded by double quotes
 
+Compiler changes:
+
+* Racket codegen now always uses `blodwen-sleep` instead of `idris2_sleep` in
+  order to not block the Racket runtime when `sleep` is called.
+
+Library changes:
+
+* Redid condition variables in the Racket codegen based on page 5 of the
+  Microsoft [Implementing CVs paper](https://www.microsoft.com/en-us/research/wp-content/uploads/2004/12/ImplementingCVs.pdf).
+  Previously, they were based on an implementation using semaphores and
+  asynchronous channels, which worked apart from `broadcast`. The rework fixes
+  `broadcast` at the cost of losing `wait-timeout` due to increased complexity
+  of their internals and interactions between their associated functions.
+
+Other changes:
+
+* The `version` field in `.ipkg` files is now used. Packages are installed into
+  a directory which includes its version number, and dependencies can have
+  version number ranges using `<=`, `<`, `>=`, `>`, `==` to express version
+  constraints. Version numbers must be in the form of integers, separated by
+  dots (e.g. `1.0`, `0.3.0`, `3.1.4.1.5` etc)
+* Idris now looks in the current working directory, under a subdirectory
+  `depends` for local installations of packages before looking globally.
+* Added an environment variable `IDRIS2_PACKAGE_PATH` for extending where to
+  look for packages.
+
 Other changes:
 
 * The `version` field in `.ipkg` files is now used. Packages are installed into

CONTRIBUTORS

@@ -54,8 +54,10 @@ tensorknower69
 then0rTh
 Theo Butler
 Thomas Dziedzic
+Thomas E. Hansen
 Tim Süberkrüb
 Timmy Jose
+Wen Kokke
 
 Apologies to anyone we've missed - let us know and we'll correct it (or just
 send a PR with the correction).

libs/base/System.idr

@@ -10,9 +10,17 @@ support fn = "C:" ++ fn ++ ", libidris2_support"
 libc : String -> String
 libc fn = "C:" ++ fn ++ ", libc 6"
 
-%foreign support "idris2_sleep"
+-- `sleep` and `usleep` need to be tied to `blodwen-[u]sleep` for threading
+-- reasons (see support/racket/support.rkt)
+
+%foreign "scheme,racket:blodwen-sleep"
+         support "idris2_sleep"
+--         "C:idris2_sleep, libidris2_support"
 prim__sleep : Int -> PrimIO ()
-%foreign support "idris2_usleep"
+
+%foreign "scheme,racket:blodwen-usleep"
+         support "idris2_usleep"
+--         "C:idris2_usleep, libidris2_support"
 prim__usleep : Int -> PrimIO ()
 
 export

libs/base/System/Concurrency.idr

@@ -62,15 +62,24 @@ mutexRelease m = primIO (prim__mutexRelease m)
 export
 data Condition : Type where [external]
 
-%foreign "scheme:blodwen-make-condition"
+%foreign "scheme,racket:blodwen-make-cv"
+         "scheme,chez:blodwen-make-condition"
 prim__makeCondition : PrimIO Condition
-%foreign "scheme:blodwen-condition-wait"
+
+%foreign "scheme,racket:blodwen-cv-wait"
+         "scheme,chez:blodwen-condition-wait"
 prim__conditionWait : Condition -> Mutex -> PrimIO ()
-%foreign "scheme:blodwen-condition-wait-timeout"
+
+%foreign "scheme,chez:blodwen-condition-wait-timeout"
+--         "scheme,racket:blodwen-cv-wait-timeout"
 prim__conditionWaitTimeout : Condition -> Mutex -> Int -> PrimIO ()
-%foreign "scheme:blodwen-condition-signal"
+
+%foreign "scheme,racket:blodwen-cv-signal"
+         "scheme,chez:blodwen-condition-signal"
 prim__conditionSignal : Condition -> PrimIO ()
-%foreign "scheme:blodwen-condition-broadcast"
+
+%foreign "scheme,racket:blodwen-cv-broadcast"
+         "scheme,chez:blodwen-condition-broadcast"
 prim__conditionBroadcast : Condition -> PrimIO ()
 
 

support/racket/support.rkt

@@ -195,6 +195,11 @@
 
 ;; Threads
 
+;; NB: Racket threads are green/virtual threads meaning extra caution is to be
+;; taken when using FFI functions in combination with threads. The *entire*
+;; Racket runtime blocks on a foreign call, meaning no threads will progress
+;; until the foreign call returns.
+
 (define (blodwen-thread proc)
   (thread (lambda () (proc (vector 0)))))
 
@@ -267,40 +272,127 @@
       (semaphore-post sema)))
 
 ;; Condition Variables
+;; As per p.5 of the MS paper
+;; https://www.microsoft.com/en-us/research/wp-content/uploads/2004/12/ImplementingCVs.pdf
 
-(define (blodwen-make-condition)
-  (make-async-channel))
+; The MS paper has the mutex be part of the CV, but that seems to be contrary to
+; most other implementations
+(struct cv (countingSem waitersLock waiters handshakeSem) #:mutable)
 
-(define (blodwen-condition-wait ach mutex)
-  ;; Pre-condition: this threads holds `mutex'.
-  (let [(sema (make-semaphore 0))]
-    (async-channel-put ach sema)
-    (blodwen-mutex-release mutex)
-    (sync sema)
-    (blodwen-mutex-acquire mutex)))
+; CONSTRUCTOR
+(define (blodwen-make-cv)
+  (let ([s (make-semaphore 0)]
+        [x (make-semaphore 1)]
+        [h (make-semaphore 0)])
+    (cv s x 0 h)))
 
-(define (blodwen-condition-wait-timeout ach mutex timeout)
-  ;; Pre-condition: this threads holds `mutex'.
-  (let [(sema (make-semaphore 0))]
-    (async-channel-put ach sema)
-    (blodwen-mutex-release mutex)
-    (sync/timeout (/ timeout 1000000) sema)
-    (blodwen-mutex-acquire mutex)))
+;; MS paper: sem.V() := sem-post  /* "sem.V() increments sem.count, atomically" */
+;;           sem.P() := sem-wait
+;; (turns out this is Dijkstra's fault: P and V match up with the Dutch
+;;  terminology)
 
-(define (blodwen-condition-signal ach)
-  (let [(sema (async-channel-try-get ach))]
-    (when sema (semaphore-post sema))))
+; WAIT
+(define (blodwen-cv-wait my-cv m)
+    ; atomically increment waiters
+    (semaphore-wait (cv-waitersLock my-cv))
+    (set-cv-waiters! my-cv (+ (cv-waiters my-cv) 1))
+    (semaphore-post (cv-waitersLock my-cv))
+    ; release the provided mutex
+    (blodwen-mutex-release m)
+    ; wait for the counting semaphore to let us through
+    (semaphore-wait (cv-countingSem my-cv))
+    ; signal to broadcast that we have proceeded past the critical point/have
+    ; been woken up successfully
+    (semaphore-post (cv-handshakeSem my-cv))
+    ; re-acquire the provided mutex
+    (blodwen-mutex-acquire m)
+    )
 
-(define (blodwen-condition-broadcast ach)
-  (letrec [(loop (lambda ()
-                   (let [(sema (async-channel-try-get ach))]
-                     (when sema ((semaphore-post sema) (loop))))))]
-    loop))
+; SIGNAL
+(define (blodwen-cv-signal my-cv)
+    ; lock access to waiters
+    (semaphore-wait (cv-waitersLock my-cv))
+    (let ([waiters (cv-waiters my-cv)])
+      (if (> waiters 0)
+
+        ; if we have waiting threads, signal one of them
+        (begin
+          (set-cv-waiters! my-cv (- waiters 1))
+          ; increment the counting semaphore to wake up a thread
+          (semaphore-post (cv-countingSem my-cv))
+          ; wait for the thread to tell us it's okay to proceed
+          (semaphore-wait (cv-handshakeSem my-cv))
+          )
+
+        ; otherwise, do nothing
+        (void)
+        )
+       ; unlock access to waiters
+       (semaphore-post (cv-waitersLock my-cv))
+       ))
+
+; BROADCAST HELPERS
+
+; for (int i = 0; i < waiters; i++) s.V();
+(define (broadcast-for-helper my-cv i)
+    (if (= i 0)
+      ; if i is zero, we're done
+      (void)
+      ; otherwise, we signal one waiting thread, decrement i, and keep going
+      (begin
+        (semaphore-post (cv-countingSem my-cv))
+
+        (broadcast-for-helper my-cv (- i 1))
+        )))
+
+; while (waiters > 0) { waiters--; h.P(); }
+(define (broadcast-while-helper my-cv waiters)
+    (if (= waiters 0)
+      ; if waiters is 0, we're done
+      (void)
+      ; otherwise, wait for "waiters" many threads to tell us they're awake
+      (begin
+        (semaphore-wait (cv-handshakeSem my-cv))
+        (broadcast-while-helper my-cv (- waiters 1))
+        )))
+
+; BROADCAST
+(define (blodwen-cv-broadcast my-cv)
+    ; lock access to waiters
+    (semaphore-wait (cv-waitersLock my-cv))
+    (let ([waiters (cv-waiters my-cv)])
+      ; signal "waiters" many threads; counting *until* 0 in the helper
+      ; function, hence "waiters" and NOT "waiters - 1"
+      (broadcast-for-helper my-cv waiters)
+      ; wait on "waiters" many threads to have been woken
+      (broadcast-while-helper my-cv waiters)
+      ; unlock access to waiters
+      (semaphore-post (cv-waitersLock my-cv))
+      ))
+
+; FIXME: Maybe later. Possibly difficult because of the handshake thingy?
+;(define (blodwen-cv-wait-timeout my-cv lockM timeout)
+;  ;; precondition: calling thread holds lockM
+;   (semaphore-wait (cv-waitersLock my-cv))                 ; x.P()
+;   (set-cv-waiters! my-cv (+ (cv-waiters my-cv) 1)) ; waiters++
+;   (semaphore-post (cv-waitersLock my-cv))                 ; x.V()
+;   (blodwen-mutex-release lockM)                    ; m.Release()
+;
+;   (sync/timeout (/ timeout 1000000) (cv-countingSem my-cv))
+;
+;   (semaphore-wait (cv-countingSem my-cv))                 ; s.P()
+;   (semaphore-post (cv-handshakeSem my-cv))                 ; h.V()
+;   (blodwen-mutex-acquire lockM)                    ; m.Acquire()
+;   )
 
 
 (define (blodwen-make-future work) (future work))
 (define (blodwen-await-future ty future) (touch future))
 
+;; NB: These should *ALWAYS* be used in multi-threaded programs since Racket
+;; threads are green/virtual threads and so using an external function will
+;; block the *entire* runtime until the function returns. This is fine for most
+;; things, but not for `sleep`.
 (define (blodwen-sleep s) (sleep s))
 (define (blodwen-usleep us) (sleep (* 0.000001 us)))
 

tests/Main.idr

@@ -195,6 +195,14 @@ racketTests = MkTestPool [Racket]
     [ "forkjoin001"
     , "semaphores001", "semaphores002"
     , "futures001"
+    , "mutex001", "mutex002", "mutex003", "mutex004", "mutex005"
+    , "conditions001"
+    , "conditions002"
+    , "conditions003"
+    , "conditions004"
+    , "conditions005"
+--    , "conditions006"
+--    , "conditions007"
     ]
 
 nodeTests : TestPool

tests/racket/conditions001/Main.idr

@@ -4,6 +4,8 @@ import Debug.Trace
 import System
 import System.Concurrency
 
+-- Signal from child
+
 main : IO ()
 main = do
     mutex <- makeMutex
@@ -11,6 +13,7 @@ main = do
 
     threadID <- fork $ do
         putStrLn "Hello"
+        sleep 1
         conditionSignal cond
 
     mutexAcquire mutex

tests/racket/conditions002/Main.idr

@@ -1,21 +1,20 @@
-module Main
+-- Idris2
 
-import Debug.Trace
 import System
 import System.Concurrency
 
+-- Test `conditionSignal` works for 1 main and 1 child thread
+
 main : IO ()
-main = do
-    mutex <- makeMutex
-    cond <- makeCondition
+main =
+  do cvMutex <- makeMutex
+     cv <- makeCondition
+     t <- fork $ do mutexAcquire cvMutex
+                    conditionWait cv cvMutex
+                    putStrLn "Hello mother"
+                    mutexRelease cvMutex
+     putStrLn "Hello child"
+     sleep 1
+     conditionSignal cv
+     threadWait t
 
-    threadID <- fork $ do
-        mutexAcquire mutex
-        conditionWait cond mutex
-        putStrLn "Goodbye"
-        mutexRelease mutex
-
-    putStrLn "Hello"
-    conditionSignal cond
-
-    threadWait threadID

tests/racket/conditions002/expected

@@ -1,2 +1,2 @@
-Hello
-Goodbye
+Hello child
+Hello mother

tests/racket/conditions003/Main.idr

@@ -1,28 +1,22 @@
-module Main
+-- Idris2
 
-import Debug.Trace
 import System
 import System.Concurrency
 
+-- Test `conditionSignal` wakes 1 thread for 1 main and N child threads
+
 main : IO ()
-main = do
-    mutex <- makeMutex
-    cond <- makeCondition
+main =
+  let n = 3 in
+  do cvMutex <- makeMutex
+     cv <- makeCondition
+     ts <- for [1..n] $ \_ => fork $ do mutexAcquire cvMutex
+                                        conditionWait cv cvMutex
+                                        putStrLn "Hello mother"
+                                        mutexRelease cvMutex
+     putStrLn "Hello child"
+     sleep 1
+     conditionSignal cv
+     -- don't threadWait since we don't know which thread got signalled
+     sleep 1
 
-    thread1 <- fork $ do
-        mutexAcquire mutex
-        conditionWait cond mutex
-        putStrLn "Goodbye"
-        mutexRelease mutex
-
-    thread2 <- fork $ do
-        mutexAcquire mutex
-        conditionWait cond mutex
-        putStrLn "Goodbye"
-        mutexRelease mutex
-
-    putStrLn "Hello"
-    conditionBroadcast cond
-
-    threadWait thread1
-    threadWait thread2

tests/racket/conditions003/expected

@@ -1,2 +1,2 @@
-Hello
-Goodbye
+Hello child
+Hello mother

tests/racket/conditions004/Main.idr

@@ -0,0 +1,19 @@
+-- Idris2
+
+import System
+import System.Concurrency
+
+-- Test `conditionBroadcast` wakes the child with 1 main and 1 child thread
+
+main : IO ()
+main =
+  do cvMutex <- makeMutex
+     cv <- makeCondition
+     t <- fork $ do mutexAcquire cvMutex
+                    conditionWait cv cvMutex
+                    putStrLn "Hello mother"
+     putStrLn "Hello child"
+     sleep 1
+     conditionBroadcast cv
+     threadWait t
+

tests/racket/conditions004/expected

@@ -0,0 +1,2 @@
+Hello child
+Hello mother

tests/racket/conditions004/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main

tests/racket/conditions005/Main.idr

@@ -0,0 +1,22 @@
+-- Idris2
+
+import System
+import System.Concurrency
+
+-- Test `conditionBroadcast` wakes all threads for 1 main and N child threads
+
+main : IO ()
+main =
+  let n = 3 in
+  do cvMutex <- makeMutex
+     cv <- makeCondition
+     ts <- for [1..n] $ \_ => fork $ do mutexAcquire cvMutex
+                                        conditionWait cv cvMutex
+                                        putStrLn "Hello mother"
+                                        mutexRelease cvMutex
+     putStrLn "Hello children"
+     sleep 1
+     conditionBroadcast cv
+     ignore $ for ts $ \t => threadWait t
+     sleep 1
+

tests/racket/conditions005/expected

@@ -0,0 +1,4 @@
+Hello children
+Hello mother
+Hello mother
+Hello mother

tests/racket/conditions005/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main

tests/racket/conditions006/Main.idr

@@ -0,0 +1,19 @@
+-- Idris2
+
+import System
+import System.Concurrency
+
+-- Test `conditionWaitTimeout` times out for 1 main and 1 child thread
+
+main : IO ()
+main =
+  do cvMutex <- makeMutex
+     cv <- makeCondition
+     t <- fork $ do mutexAcquire cvMutex
+                    conditionWaitTimeout cv cvMutex 1000000
+                    putStrLn "Where are you mother?"
+                    mutexRelease cvMutex
+     sleep 2
+     putStrLn "Sorry I'm late child!"
+     threadWait t
+

tests/racket/conditions006/expected

@@ -0,0 +1,2 @@
+Where are you mother?
+Sorry I'm late child!

tests/racket/conditions006/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main

tests/racket/conditions007/Main.idr

@@ -0,0 +1,36 @@
+-- Idris2
+
+import System
+import System.Concurrency
+
+-- Test `conditionWaitTimeout` times out m of n threads for 1 main and n child
+-- threads
+
+main : IO ()
+main =
+  let
+    n = 5
+    m = 3
+  in
+    do cvMutex <- makeMutex
+       cv <- makeCondition
+
+       -- spawn n-m inifinitely patient children
+       waiting <- for [1..(n - m)] $ \_ => fork $
+             do mutexAcquire cvMutex
+                conditionWait cv cvMutex
+                putStrLn "Woke up despite no timeout (SHOULDN'T HAPPEN)"
+                mutexRelease cvMutex
+
+       -- spawn m impatient children
+       impatients <- for [1..m] $ \_ => fork $
+             do mutexAcquire cvMutex
+                conditionWaitTimeout cv cvMutex 1000000
+                putStrLn "Where are you mother?"
+                mutexRelease cvMutex
+
+       sleep m
+       putStrLn "Sorry I'm late children! Weren't there more of you?..."
+       for impatients $ \t => threadWait t
+       sleep 1
+

tests/racket/conditions007/expected

@@ -0,0 +1,4 @@
+Where are you mother?
+Where are you mother?
+Where are you mother?
+Sorry I'm late children! Weren't there more of you?...

tests/racket/conditions007/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main

tests/racket/mutex001/Main.idr

@@ -0,0 +1,13 @@
+-- Idris2
+
+import System.Concurrency
+
+||| Test basic lock/acquire and unlock/release functionality
+main : IO ()
+main =
+  do m <- makeMutex
+     mutexAcquire m
+     putStrLn "Mutex acquired"
+     mutexRelease m
+     putStrLn "Mutex released"
+

tests/racket/mutex001/expected

@@ -0,0 +1,2 @@
+Mutex acquired
+Mutex released

tests/racket/mutex001/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main

tests/racket/mutex002/Main.idr

@@ -0,0 +1,14 @@
+-- Idris2
+
+import System.Concurrency
+
+||| Test basic lock/acquire and unlock/release functionality from child thread
+main : IO ()
+main =
+  do m <- makeMutex
+     t <- fork $ do mutexAcquire m
+                    putStrLn "Child acquired mutex"
+                    mutexRelease m
+                    putStrLn "Child released mutex"
+     threadWait t
+

tests/racket/mutex002/expected

@@ -0,0 +1,2 @@
+Child acquired mutex
+Child released mutex

tests/racket/mutex002/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main

tests/racket/mutex003/Main.idr

@@ -0,0 +1,12 @@
+-- Idris2
+
+import System
+import System.Concurrency
+
+||| Test releasing without acquiring errors correctly
+main : IO ()
+main =
+  do m <- makeMutex
+     mutexRelease m
+     putStrLn "Released w/o acquiring (SHOULDN'T HAPPEN)"
+

tests/racket/mutex003/expected

@@ -0,0 +1 @@
+Exception in mutexRelease: thread does not own mutex

tests/racket/mutex003/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main

tests/racket/mutex004/Main.idr

@@ -0,0 +1,16 @@
+-- Idris2
+
+import System
+import System.Concurrency
+
+||| Test double-release errors correctly
+main : IO ()
+main =
+  do m <- makeMutex
+     mutexAcquire m
+     putStrLn "Mutex acquired"
+     mutexRelease m
+     putStrLn "1st release"
+     mutexRelease m
+     putStrLn "2nd release (SHOULDN'T HAPPEN)"
+

tests/racket/mutex004/expected

@@ -0,0 +1,3 @@
+Exception in mutexRelease: thread does not own mutex
+Mutex acquired
+1st release

tests/racket/mutex004/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main

tests/racket/mutex005/Main.idr

@@ -0,0 +1,21 @@
+-- Idris2
+
+import System
+import System.Concurrency
+
+-- Test that acquiring the mutex in the parent, then the child blocks correctly
+main : IO ()
+main =
+  do m <- makeMutex
+     mutexAcquire m
+     putStrLn "Main acquired mutex"
+     t <- fork $ do mutexAcquire m
+                    putStrLn "Child acquired mutex"
+                    sleep 2
+                    mutexRelease m
+                    putStrLn "Child released mutex"
+     sleep 1
+     mutexRelease m
+     putStrLn "Main released mutex"
+     threadWait t
+

tests/racket/mutex005/expected

@@ -0,0 +1,4 @@
+Main acquired mutex
+Main released mutex
+Child acquired mutex
+Child released mutex

tests/racket/mutex005/run

@@ -0,0 +1 @@
+$1 --no-banner --no-color --console-width 0 --cg racket Main.idr --exec main