Concurrency utilities.
Actor pattern and some limited structured concurrency tools
om-fork
This package provides some concurrency abstractions.
Structured concurrency
This package provides some very limited tools for structured concurrency. These tools are "limited" in the sense that they target a very specific use case: making sure that if any one of a group of cooperating threads ends for any reason, then they all end. For a more complete treatment of structured concurrency, see the ki package.
Motivation
If you are using the actor model (see below) and one of your actors dies for whatever reason, you probably want to crash completely instead of ending up in some kind of "half-crashed" zombie state.
Example
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeApplications #-}
module Main (main) where
import Control.Concurrent (threadDelay)
import Control.Exception (SomeException, try)
import Control.Monad (void)
import Control.Monad.IO.Class (MonadIO(liftIO))
import Control.Monad.Logger (runStdoutLoggingT)
import OM.Fork (race, runRace, wait)
main :: IO ()
main =
void . try @SomeException $
runStdoutLoggingT $ do
runRace $ do
race "loop1" $ loop1
race "loop2" $ loop2
race "loopThatCrashes" $ loopThatCrashes
wait
loop1 :: (MonadIO m) => m void
loop1 = do
liftIO $ do
threadDelay 1_000_000
putStrLn "loop1 cycle"
loop1
loop2 :: (MonadIO m) => m void
loop2 = do
liftIO $ do
threadDelay 200_000
putStrLn "loop2 cycle"
loop2
loopThatCrashes :: (MonadIO m) => m void
loopThatCrashes = do
liftIO $ do
threadDelay 5_000_000
putStrLn "loopThatCrashes about to crash"
error "crash"
Actor model
This package provides an abstraction over the "actor model". In particular, the main thing of value is a way get heterogeneously typed blocking responses from the actor, using call.
Motivation
The actor model isn't suitable for every concurrency problem, but maybe you've been programming in Erlang and you have a problem which sits easily in your mind as an Actor Model problem. A full motivation of the actor model is beyond the scope of this package.
Anyway, you probably want a way to interact with your "actor" in a type safe way. This package provides convenient tools and patterns to do so.
Example
{-# LANGUAGE LambdaCase #-}
module Main (main) where
import Control.Concurrent (Chan, forkIO, newChan, readChan)
import Control.Monad (void)
import OM.Fork (Responder, call, cast, respond)
{- | Messages that can be sent to the actor. -}
data MyMsg
= ReverseEcho String (Responder String)
{- ^ A blocking message. Responds with a String -}
| Print String
{- ^ A non-blocking message. -}
| GetState (Responder Int)
{- ^ Another blocking message. Responds with an Int -}
{- | The "state" is just a count of how many messages we've seen so far. -}
actorLoop :: Int -> Chan MyMsg -> IO void
actorLoop state chan = do
readChan chan >>= \case
ReverseEcho str responder ->
void $ respond responder (reverse str)
Print str ->
putStrLn ("Printeded in background by actor thread: " <> str)
GetState responder ->
void $ respond responder state
actorLoop (succ state) chan
main :: IO ()
main = do
{- `Chan a` is an instance of `Actor` -}
chan <- newChan
void . forkIO $ actorLoop 0 chan
{- | Notice that the result of this `call` is a String. -}
putStrLn =<< call chan (ReverseEcho "hello world")
cast chan (Print "foo")
{- | Notice that the result of this `call` is an Int. -}
actorState <- call chan GetState
print (actorState * 2)