MonadPar WIP

Author: garyb

bower.json

@@ -20,7 +20,8 @@
     "purescript-console": "^1.0.0",
     "purescript-exceptions": "^1.0.0",
     "purescript-functions": "^1.0.0",
-    "purescript-transformers": "^1.0.0"
+    "purescript-transformers": "^1.0.0",
+    "purescript-parallel": "^1.0.0"
   },
   "devDependencies": {
     "purescript-partial": "^1.1.2"

src/Control/Monad/Aff.js

@@ -320,3 +320,11 @@ exports._tailRecM = function (isLeft, f, a) {
     }(a);
   };
 };
+
+
+var avar = require("../Control.Monad.Aff.AVar/foreign.js");
+
+exports._makeVar = avar._makeVar;
+exports._takeVar = avar._takeVar;
+exports._putVar = avar._putVar;
+exports._killVar = avar._killVar;

src/Control/Monad/Aff.purs

@@ -21,6 +21,7 @@ module Control.Monad.Aff
   where
 
 import Prelude
+
 import Control.Alt (class Alt)
 import Control.Alternative (class Alternative)
 import Control.Monad.Cont.Class (class MonadCont)
@@ -30,7 +31,9 @@ import Control.Monad.Eff.Exception (Error, EXCEPTION, throwException, error)
 import Control.Monad.Error.Class (class MonadError, throwError)
 import Control.Monad.Rec.Class (class MonadRec)
 import Control.MonadPlus (class MonadZero, class MonadPlus)
+import Control.Parallel.Class (class MonadRace, class MonadPar)
 import Control.Plus (class Plus)
+
 import Data.Either (Either(..), either, isLeft)
 import Data.Foldable (class Foldable, foldl)
 import Data.Function.Uncurried (Fn2, Fn3, runFn2, runFn3)
@@ -202,6 +205,56 @@ instance semigroupCanceler :: Semigroup (Canceler e) where
 instance monoidCanceler :: Monoid (Canceler e) where
   mempty = Canceler (const (pure true))
 
+instance monadParAff :: MonadPar (Aff e) where
+  par f ma mb = do
+    va <- _makeVar nonCanceler
+    vb <- _makeVar nonCanceler
+    c1 <- forkAff (putOrKill va =<< attempt ma)
+    c2 <- forkAff (putOrKill vb =<< attempt mb)
+    f <$> (takeVar va) <*> (takeVar vb)
+    where
+    putOrKill :: forall a. AVar a -> Either Error a -> Aff e Unit
+    putOrKill v = either (killVar v) (putVar v)
+
+instance monadRaceAff :: MonadRace (Aff e) where
+  stall = throwError $ error "Stalled"
+  race a1 a2 = do
+    va <- _makeVar nonCanceler -- the `a` value
+    ve <- _makeVar nonCanceler -- the error count (starts at 0)
+    putVar ve 0
+    c1 <- forkAff $ either (maybeKill va ve) (putVar va) =<< attempt a1
+    c2 <- forkAff $ either (maybeKill va ve) (putVar va) =<< attempt a2
+    takeVar va `cancelWith` (c1 <> c2)
+    where
+    maybeKill :: forall a. AVar a -> AVar Int -> Error -> Aff e Unit
+    maybeKill va ve err = do
+      e <- takeVar ve
+      if e == 1 then killVar va err else pure unit
+      putVar ve (e + 1)
+
+--------------------------------
+
+foreign import data AVar :: * -> *
+
+takeVar :: forall e a. AVar a -> Aff e a
+takeVar q = runFn2 _takeVar nonCanceler q
+
+putVar :: forall e a. AVar a -> a -> Aff e Unit
+putVar q a = runFn3 _putVar nonCanceler q a
+
+killVar :: forall e a. AVar a -> Error -> Aff e Unit
+killVar q e = runFn3 _killVar nonCanceler q e
+
+foreign import _makeVar :: forall e a. Canceler e -> Aff e (AVar a)
+
+foreign import _takeVar :: forall e a. Fn2 (Canceler e) (AVar a) (Aff e a)
+
+foreign import _putVar :: forall e a. Fn3 (Canceler e) (AVar a) a (Aff e Unit)
+
+foreign import _killVar :: forall e a. Fn3 (Canceler e) (AVar a) Error (Aff e Unit)
+
+--------------------------------
+
 foreign import _cancelWith :: forall e a. Fn3 (Canceler e) (Aff e a) (Canceler e) (Aff e a)
 
 foreign import _setTimeout :: forall e a. Fn3 (Canceler e) Int (Aff e a) (Aff e a)

test/Test/Main.purs

@@ -4,10 +4,10 @@ import Prelude
 
 import Control.Alt ((<|>))
 import Control.Apply ((*>))
+import Control.Parallel.Class (parallel, runParallel)
 import Control.Monad.Aff (Aff, runAff, makeAff, later, later', forkAff, forkAll, Canceler(..), cancel, attempt, finally, apathize)
 import Control.Monad.Aff.AVar (AVAR, makeVar, makeVar', putVar, modifyVar, takeVar, killVar)
 import Control.Monad.Aff.Console (log)
-import Control.Monad.Aff.Par (Par(..), runPar)
 import Control.Monad.Cont.Class (callCC)
 import Control.Monad.Eff (Eff)
 import Control.Monad.Eff.Console (CONSOLE)
@@ -40,8 +40,8 @@ test_makeAff = unsafePartial do
   asyncF <- attempt $ makeAff \reject resolve -> reject (error "ok")
   log $ "makeAff asynchronous failure is " <> message (fromLeft asyncF)
 
-  asyncF <- attempt $ makeAff \reject resolve -> synchronousUnexpectedThrowError
-  log $ "makeAff synchronous failure is " <> message (fromLeft asyncF)
+  asyncF' <- attempt $ makeAff \reject resolve -> synchronousUnexpectedThrowError
+  log $ "makeAff synchronous failure is " <> message (fromLeft asyncF')
 
   log "Success: makeAff is ok"
 
@@ -99,25 +99,25 @@ test_finally = do
 
 test_parRace :: TestAVar Unit
 test_parRace = do
-  s <- runPar (Par (later' 100 $ pure "Success: Early bird got the worm") <|>
-               Par (later' 200 $ pure "Failure: Late bird got the worm"))
+  s <- runParallel (parallel (later' 100 $ pure "Success: Early bird got the worm") <|>
+               parallel (later' 200 $ pure "Failure: Late bird got the worm"))
   log s
 
 test_parError :: TestAVar Unit
 test_parError = do
-  e <- attempt $ runPar (Par (throwError (error ("Oh noes!"))) *> pure unit)
+  e <- attempt $ runParallel (parallel (throwError (error ("Oh noes!"))) *> pure unit)
   either (const $ log "Success: Exception propagated") (const $ log "Failure: Exception missing") e
 
 test_parRaceKill1 :: TestAVar Unit
 test_parRaceKill1 = do
-  s <- runPar (Par (later' 100 $ throwError (error ("Oh noes!"))) <|>
-               Par (later' 200 $ pure "Success: Early error was ignored in favor of late success"))
+  s <- runParallel (parallel (later' 100 $ throwError (error ("Oh noes!"))) <|>
+               parallel (later' 200 $ pure "Success: Early error was ignored in favor of late success"))
   log s
 
 test_parRaceKill2 :: TestAVar Unit
 test_parRaceKill2 = do
-  e <- attempt $ runPar (Par (later' 100 $ throwError (error ("Oh noes!"))) <|>
-                         Par (later' 200 $ throwError (error ("Oh noes!"))))
+  e <- attempt $ runParallel (parallel (later' 100 $ throwError (error ("Oh noes!"))) <|>
+                         parallel (later' 200 $ throwError (error ("Oh noes!"))))
   either (const $ log "Success: Killing both kills it dead") (const $ log "Failure: It's alive!!!") e
 
 test_semigroupCanceler :: Test Unit
@@ -137,13 +137,13 @@ test_cancelLater = do
   v <- cancel c (error "Cause")
   log (if v then "Success: Canceled later" else "Failure: Did not cancel later")
 
-test_cancelPar :: TestAVar Unit
-test_cancelPar = do
-  c  <- forkAff <<< runPar $ Par (later' 100 $ log "Failure: #1 should not get through") <|>
-                             Par (later' 100 $ log "Failure: #2 should not get through")
+test_cancelParallel :: TestAVar Unit
+test_cancelParallel = do
+  c  <- forkAff <<< runParallel $ parallel (later' 100 $ log "Failure: #1 should not get through") <|>
+                             parallel (later' 100 $ log "Failure: #2 should not get through")
   v  <- c `cancel` (error "Must cancel")
-  log (if v then "Success: Canceling composite of two Par succeeded"
-                   else "Failure: Canceling composite of two Par failed")
+  log (if v then "Success: Canceling composite of two Parallel succeeded"
+                   else "Failure: Canceling composite of two Parallel failed")
 
 test_syncTailRecM :: TestAVar Unit
 test_syncTailRecM = do
@@ -224,20 +224,20 @@ main = runAff throwException (const (pure unit)) $ do
   log "Testing finally"
   test_finally
 
-  log "Test Par (*>)"
+  log "Test Parallel (*>)"
   test_parError
 
-  log "Testing Par (<|>)"
+  log "Testing Parallel (<|>)"
   test_parRace
 
-  log "Testing Par (<|>) - kill one"
+  log "Testing Parallel (<|>) - kill one"
   test_parRaceKill1
 
-  log "Testing Par (<|>) - kill two"
+  log "Testing Parallel (<|>) - kill two"
   test_parRaceKill2
 
-  log "Testing cancel of Par (<|>)"
-  test_cancelPar
+  log "Testing cancel of Parallel (<|>)"
+  test_cancelParallel
 
   log "Testing synchronous tailRecM"
   test_syncTailRecM