om-fork (empty) → 0.7.1.6

raw patch · 7 files changed

+535/−0 lines, 7 filesdep +aesondep +basedep +exceptionssetup-changed

Dependencies added: aeson, base, exceptions, ki, monad-logger, om-fork, om-show, text, unliftio

Files

LICENSE +19/−0
README.md +140/−0
Setup.hs +2/−0
examples/actor.hs +47/−0
examples/structured.hs +43/−0
om-fork.cabal +71/−0
src/OM/Fork.hs +213/−0

+ LICENSE view

@@ -0,0 +1,19 @@+Copyright 2022 Rick Owens++Permission is hereby granted, free of charge, to any person obtaining a+copy of this software and associated documentation files (the "Software"),+to deal in the Software without restriction, including without limitation+the rights to use, copy, modify, merge, publish, distribute, sublicense,+and/or sell copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL+THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR+OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.

+ README.md view

@@ -0,0 +1,140 @@+# om-fork++This package provides some concurrency abstractions.++- [Structured concurrency](#structured-concurrency)+    - [Motivation](#motivation)+    - [Example](#example)+- [Actor model](#actor-model)+    - [Motivation](#motivation)+    - [Example](#example)++## 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](https://hackage.haskell.org/package/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++```haskell+{-# 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+ ++```haskell+{-# 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)+```

+ Setup.hs view

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

+ examples/actor.hs view

@@ -0,0 +1,47 @@+{-# 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)++

+ examples/structured.hs view

@@ -0,0 +1,43 @@+{-# 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"

+ om-fork.cabal view

@@ -0,0 +1,71 @@+cabal-version:       3.0+name:                om-fork+version:             0.7.1.6+synopsis:            Concurrency utilities.+description:         Actor pattern and some limited structured concurrency+                     tools+homepage:            https://github.com/owensmurray/om-fork+license:             MIT+license-file:        LICENSE+author:              Rick Owens+maintainer:          rick@owensmurray.com+copyright:           2018 Owens Murray, LLC.+-- category:            +build-type:          Simple+extra-source-files:+  README.md+  LICENSE++common dependencies+  build-depends:+    , aeson        >= 2.0.3.0  && < 2.1+    , base         >= 4.15.0.0 && < 4.16+    , exceptions   >= 0.10.4   && < 0.11+    , ki           >= 0.2.0.1  && < 0.3+    , monad-logger >= 0.3.36   && < 0.4+    , om-show      >= 0.1.2.0  && < 0.2+    , text         >= 1.2.5.0  && < 1.3+    , unliftio     >= 0.2.22.0 && < 0.3++common warnings+  ghc-options:+    -Wmissing-deriving-strategies+    -Wmissing-export-lists+    -Wmissing-import-lists+    -Wredundant-constraints+    -Wall++library+  import: warnings, dependencies+  exposed-modules:     +    OM.Fork+  -- other-modules:       +  -- other-extensions:    +  hs-source-dirs: src+  default-language: Haskell2010+++-- This isn't really a test, it is just used to make sure the example+-- compiles without clusttering up the package with a bunch of+-- executables.+test-suite actor-example+  import: warnings, dependencies+  main-is: actor.hs+  type: exitcode-stdio-1.0+  hs-source-dirs: examples+  default-language: Haskell2010+  build-depends:+    , om-fork++-- This isn't really a test, it is just used to make sure the example+-- compiles without clusttering up the package with a bunch of+-- executables.+test-suite structured-example+  import: warnings, dependencies+  main-is: structured.hs+  type: exitcode-stdio-1.0+  hs-source-dirs: examples+  default-language: Haskell2010+  build-depends:+    , om-fork+

+ src/OM/Fork.hs view

@@ -0,0 +1,213 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ImplicitParams #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeFamilies #-}++{- | Description: Thread utilities. -}+module OM.Fork (+  -- * Actor Communication.+  Actor(..),+  Responder,+  Responded,+  respond,+  call,+  cast,++  -- * Forking Background Processes.+  logUnexpectedTermination,+  ProcessName(..),+  Race,+  runRace,+  race,+  wait,+) where+++import Control.Concurrent (Chan, myThreadId, newEmptyMVar, putMVar,+  takeMVar, writeChan)+import Control.Monad (void)+import Control.Monad.Catch (MonadThrow(throwM), MonadCatch, SomeException,+  try)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Logger.CallStack (MonadLogger, logInfo, logWarn)+import Data.Aeson (ToJSON, toJSON)+import Data.String (IsString)+import Data.Text (Text)+import OM.Show (showt)+import UnliftIO (MonadUnliftIO, askRunInIO, throwString)+import qualified Ki+++{- | How to respond to a asynchronous message. -}+newtype Responder a = Responder {+    unResponder :: a -> IO ()+  }+instance ToJSON (Responder a) where+  toJSON _ = toJSON ("<Responder>" :: Text)+instance Show (Responder a) where+  show _ = "Responder"+++{- | The class of types that can act as the handle for an asynchronous actor. -}+class Actor a where+  {- | The type of messages associated with the actor. -}+  type Msg a+  {- | The channel through which messages can be sent to the actor. -}+  actorChan :: a -> Msg a -> IO ()+instance Actor (Chan m) where+  type Msg (Chan m) = m+  actorChan = writeChan+++{- | Respond to an asynchronous message. -}+respond :: (MonadIO m) => Responder a -> a -> m Responded+respond responder val = do+  liftIO (unResponder responder val)+  return Responded+++{- | Send a message to an actor, and wait for a response. -}+call+  :: ( Actor actor+     , MonadIO m+     )+  => actor {- ^ The actor to which we are sending a call request. -}+  -> (Responder a -> Msg actor)+     {- ^+       Given a way for the actor to respond to the message, construct+       a message that should be sent to the actor.++       Typically, your 'Msg' type will look something like this:++       > data MyMsg+       >   = MsgWithResponse SomeData (Responder ResponseType)+       >     -- In this example, this type of message requires a+       >     -- response. We package the responder up as part of the+       >     -- message itself. Idiomatically it is best to put the+       >     -- responder as the last argument so that it is easy to pass+       >     -- 'MsgWithResponse someData' to 'call'.+       >   | MsgWithoutResponse SomeData+       >     -- In this example, this type of message requires no response. It+       >     -- is a "fire and forget" message.++       you will call 'call' like this:++       > do+       >   response <- call actor (MsgWithResponse someData)+       >   -- response :: ResponseType++     -}+  -> m a+call actor mkMessage = liftIO $ do+  mVar <- newEmptyMVar+  actorChan actor (mkMessage (Responder (putMVar mVar)))+  takeMVar mVar+++{- | Send a message to an actor, but do not wait for a response. -}+cast :: (Actor actor, MonadIO m) => actor -> Msg actor -> m ()+cast actor = liftIO . actorChan actor+++{- |+  Proof that 'respond' was called. Clients can use this type in their+  type signatures when they require that 'respond' be called at least+  once, because calling 'respond' is the only way to generate values of+  this type.+-}+data Responded = Responded+++{- | Log (at WARN) when the action terminates for any reason. -}+logUnexpectedTermination :: (MonadLogger m, MonadCatch m)+  => ProcessName+  -> m a+  -> m a+logUnexpectedTermination (ProcessName name) action =+  try action >>= \case+    Left err -> do+      logWarn+        $ "Thread " <> name <> " finished with an error: " <> showt err+      throwM (err :: SomeException)+    Right v -> do+      logWarn $ "Thread " <> name <> " finished normally."+      return v+++{- |+  Run a thread race.++  Within the provided action, you can call 'race' to fork new background+  threads. When the action terminates, all background threads forked+  with 'race' are also terminated. Likewise, if any one of the racing+  threads terminates, then all other racing threads are terminated _and_+  'runRace' will throw an exception.++  In any event, when 'runRace' returns, all background threads forked+  by the @action@ using 'race' will have been terminated.+-}+runRace+  :: (MonadUnliftIO m)+  => (Race => m a) {- ^ - @action@: The provided "race" action. -}+  -> m a+runRace action = do+  runInIO <- askRunInIO+  liftIO . Ki.scoped $ \scope ->+    runInIO (let ?scope = scope in action)+++{- |+  This constraint indicates that we are in the context of a thread race. If any+  threads in the race terminate, then all threads in the race terminate.+  Threads are "in the race" if they were forked using 'race'.+-}+type Race = (?scope :: Ki.Scope)+++{- |+  Fork a new thread within the context of a race. This thread will be+  terminated when any other racing thread terminates, or else if this+  thread terminates first it will cause all other racing threads to+  be terminated.++  Generally, we normally expect that the thread is a "background thread"+  and will never terminate under "normal" conditions.+-}+race+  :: ( MonadCatch m+     , MonadLogger m+     , MonadUnliftIO m+     , Race+     )+  => ProcessName+  -> m a+  -> m ()+race name action = do+  runInIO <- askRunInIO+  liftIO+    . Ki.fork_ ?scope+    $ do+      tid <- myThreadId+      runInIO . logUnexpectedTermination name $ do+        logInfo $ "Starting thread (tid, name): " <> showt (tid, name)+        void action+      throwString $ "Thread Finished (tid, name): " <> show (tid, name)+++{- | The name of a process. -}+newtype ProcessName = ProcessName+  { unProcessName :: Text+  }+  deriving newtype (IsString, Semigroup, Monoid, Show)+++{- | Wait for all racing threads to terminate. -}+wait :: (MonadIO m, Race) => m ()+wait = liftIO $ Ki.wait ?scope++