theatre-dev (empty) → 0.0.1

raw patch · 14 files changed

+1042/−0 lines, 14 filesdep +asyncdep +basedep +contravariant

Dependencies added: async, base, contravariant, hspec, rerebase, stm, theatre-dev, unagi-chan, vector

Files

• LICENSE +22/−0
• hspec/Main.hs +11/−0
• hspec/TheatreDev/StmBasedSpec.hs +46/−0
• library/TheatreDev/ExtrasFor/List.hs +12/−0
• library/TheatreDev/ExtrasFor/TBQueue.hs +29/−0
• library/TheatreDev/Perpetual.hs +108/−0
• library/TheatreDev/Prelude.hs +73/−0
• library/TheatreDev/StmBased.hs +231/−0
• library/TheatreDev/StmBased/StmStructures/Runner.hs +97/−0
• library/TheatreDev/StmBased/Tell.hs +33/−0
• library/TheatreDev/StmBased/Wait.hs +27/−0
• library/TheatreDev/Terminal/Actor.hs +200/−0
• library/TheatreDev/Terminal/StatefulActorSpec.hs +39/−0
• theatre-dev.cabal +114/−0

@@ -0,0 +1,22 @@+Copyright (c) 2022, Nikita Volkov++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.

@@ -0,0 +1,11 @@+module Main (main) where++import Test.Hspec+import qualified TheatreDev.StmBasedSpec+import Prelude++main :: IO ()+main =+  hspec do+    describe "TheatreDev" do+      describe "StmBased" TheatreDev.StmBasedSpec.spec

@@ -0,0 +1,46 @@+{-# OPTIONS_GHC -Wno-unused-local-binds #-}++module TheatreDev.StmBasedSpec (spec) where++import Test.Hspec+import qualified TheatreDev.StmBased as Actor+import Prelude++spec :: Spec+spec =+  do+    describe "spawnStatelessBatched" do+      let spawnInt step = Actor.spawnStatefulBatched @Int 0 step (const (return ()))+      let spawnUnit step = Actor.spawnStatefulBatched () step (const (return ()))++      it "Works in batches" do+        acc <- newIORef []+        actorLock <- newEmptyMVar+        emitterLock <- newEmptyMVar+        actor <- spawnUnit $ \_ messages ->+          do+            modifyIORef' acc (messages :)+            putMVar emitterLock ()+            takeMVar actorLock++        Actor.tell actor 1++        takeMVar emitterLock+        Actor.tell actor 2+        Actor.tell actor 3+        putMVar actorLock ()++        takeMVar emitterLock+        Actor.tell actor 4+        putMVar actorLock ()++        takeMVar emitterLock++        collectedBatches <- reverse . fmap toList <$> readIORef acc+        shouldBe collectedBatches [[1], [2, 3], [4]]++      it "Threads the state" do+        pending++      it "Kill and wait" do+        pending

@@ -0,0 +1,12 @@+module TheatreDev.ExtrasFor.List where++import TheatreDev.Prelude++splitWhileJust :: [Maybe a] -> ([a], [Maybe a])+splitWhileJust = go []+  where+    go !acc = \case+      head : tail -> case head of+        Nothing -> (reverse acc, head : tail)+        Just liveHead -> go (liveHead : acc) tail+      [] -> (reverse acc, [])

@@ -0,0 +1,29 @@+module TheatreDev.ExtrasFor.TBQueue where++import Control.Concurrent.STM.TBQueue+import TheatreDev.Prelude++flushNonEmptyTBQueue :: TBQueue a -> STM (NonEmpty a)+flushNonEmptyTBQueue x = do+  head <- readTBQueue x+  tail <- flushTBQueue x+  return (head :| tail)++-- | Get a list of all entries in the queue without removing them.+inspectTBQueue :: TBQueue a -> STM [a]+inspectTBQueue queue = do+  list <- flushTBQueue queue+  forM_ list $ writeTBQueue queue+  return list++-- | Starting from \"stm\" 2.5.2.0 "flushTBQueue" is broken.+-- We're fixing it here.+simplerFlushTBQueue :: TBQueue a -> STM [a]+simplerFlushTBQueue queue =+  go []+  where+    go !acc = do+      element <- tryReadTBQueue queue+      case element of+        Just element -> go $ element : acc+        Nothing -> return $ reverse acc

@@ -0,0 +1,108 @@+-- |+-- Exploration of perpetual actors.+-- I.e., those that exist for the whole duration of the app.+--+-- This limitation provides for simpler API and most apps+-- are expected not to need more.+module TheatreDev.Perpetual+  ( Actor,+    spawnStateless,+    spawnStateful,+    tell,+  )+where++import qualified Control.Concurrent.Chan.Unagi as Unagi+import TheatreDev.Prelude++-- |+-- Actor, which processes the messages of type @msg@.+--+-- Provides abstraction over the communication channel and threads.+newtype Actor msg+  = Actor (msg -> IO ())++-- |+-- Distributes the message across the merged actors.+instance Semigroup (Actor msg) where+  Actor lTell <> Actor rTell =+    Actor $ \msg -> lTell msg >> rTell msg+  sconcat actors = Actor $ \msg -> forM_ actors $ \(Actor tell) -> tell msg+  stimes n (Actor tell) = Actor $ \msg -> replicateM_ (fromIntegral n) $ tell msg++-- |+-- Provides an identity for merging the actors,+-- which does nothing.+instance Monoid (Actor msg) where+  mempty = Actor (const (return ()))+  mconcat actors = Actor $ \msg -> forM_ actors $ \(Actor tell) -> tell msg++-- |+-- Maps the input message to a different type.+instance Contravariant Actor where+  contramap fn (Actor tell) =+    Actor (tell . fn)++-- |+-- Splits the message between actors.+instance Divisible Actor where+  conquer =+    mempty+  divide divisor (Actor lTell) (Actor rTell) =+    Actor $ \msg -> case divisor msg of+      (lMsg, rMsg) -> lTell lMsg >> rTell rMsg++-- |+-- Provides a choice between alternative actors to process the message.+instance Decidable Actor where+  lose _ =+    Actor $ const $ return ()+  choose decider (Actor lTell) (Actor rTell) =+    Actor $ either lTell rTell . decider++spawnStateless ::+  -- |+  -- Process the next message.+  -- Must not throw any exceptions.+  (msg -> IO ()) ->+  -- |+  -- Action forking a thread to run the actor loop and+  -- producing a handle for sending messages to it.+  IO (Actor msg)+spawnStateless process = do+  (inChan, outChan) <- Unagi.newChan+  forkIO+    $ let loop = do+            msg <- Unagi.readChan outChan+            process msg+            loop+       in loop+  return $ Actor $ Unagi.writeChan inChan++spawnStateful ::+  -- |+  -- Initial state.+  state ->+  -- |+  -- Process the next message updating the state.+  -- The IO action must not throw any exceptions.+  (state -> msg -> IO state) ->+  -- |+  -- Action forking a thread to run the actor loop and+  -- producing a handle for sending messages to it.+  IO (Actor msg)+spawnStateful state process = do+  (inChan, outChan) <- Unagi.newChan+  forkIO+    $ let loop !state = do+            msg <- Unagi.readChan outChan+            state <- process state msg+            loop state+       in loop state+  return $ Actor $ Unagi.writeChan inChan++-- |+-- Schedule a message for the actor to process+-- after the ones already scheduled.+tell :: Actor msg -> msg -> IO ()+tell = coerce

@@ -0,0 +1,73 @@+module TheatreDev.Prelude+  ( module Exports,+  )+where++import Control.Applicative as Exports+import Control.Arrow as Exports hiding (first, second)+import Control.Category as Exports+import Control.Concurrent as Exports+import Control.Concurrent.Async as Exports (Concurrently (..))+import Control.Exception as Exports+import Control.Monad as Exports hiding (fail, forM, forM_, mapM, mapM_, msum, sequence, sequence_)+import Control.Monad.Fail as Exports+import Control.Monad.Fix as Exports hiding (fix)+import Control.Monad.IO.Class as Exports+import Control.Monad.ST as Exports+import Data.Bifunctor as Exports+import Data.Bits as Exports+import Data.Bool as Exports+import Data.Char as Exports+import Data.Coerce as Exports+import Data.Complex as Exports+import Data.Data as Exports+import Data.Dynamic as Exports+import Data.Either as Exports+import Data.Fixed as Exports+import Data.Foldable as Exports+import Data.Function as Exports hiding (id, (.))+import Data.Functor as Exports hiding (unzip)+import Data.Functor.Contravariant as Exports+import Data.Functor.Contravariant.Divisible as Exports+import Data.IORef as Exports+import Data.Int as Exports+import Data.Ix as Exports+import Data.List as Exports hiding (all, and, any, concat, concatMap, elem, find, foldl, foldl', foldl1, foldr, foldr1, isSubsequenceOf, mapAccumL, mapAccumR, maximum, maximumBy, minimum, minimumBy, notElem, or, product, sortOn, sum, uncons)+import Data.List.NonEmpty as Exports (NonEmpty (..), nonEmpty)+import Data.Maybe as Exports+import Data.Monoid as Exports hiding (Alt, First (..), Last (..))+import Data.Ord as Exports+import Data.Proxy as Exports+import Data.Ratio as Exports+import Data.STRef as Exports+import Data.Semigroup as Exports+import Data.String as Exports+import Data.Traversable as Exports+import Data.Tuple as Exports+import Data.Unique as Exports+import Data.Version as Exports+import Data.Void as Exports+import Data.Word as Exports+import Debug.Trace as Exports+import Foreign.ForeignPtr as Exports+import Foreign.Ptr as Exports+import Foreign.StablePtr as Exports+import Foreign.Storable as Exports+import GHC.Conc as Exports hiding (threadWaitRead, threadWaitReadSTM, threadWaitWrite, threadWaitWriteSTM, withMVar)+import GHC.Exts as Exports (groupWith, inline, lazy, sortWith)+import GHC.Generics as Exports (Generic)+import GHC.IO.Exception as Exports+import Numeric as Exports+import System.Environment as Exports+import System.Exit as Exports+import System.IO as Exports (Handle, hClose)+import System.IO.Error as Exports+import System.IO.Unsafe as Exports+import System.Mem as Exports+import System.Mem.StableName as Exports+import System.Timeout as Exports+import Text.ParserCombinators.ReadPrec as Exports (ReadPrec, readP_to_Prec, readPrec_to_P, readPrec_to_S, readS_to_Prec)+import Text.Printf as Exports (hPrintf, printf)+import Text.Read as Exports (Read (..), readEither, readMaybe)+import Unsafe.Coerce as Exports+import Prelude as Exports hiding (all, and, any, concat, concatMap, elem, fail, foldl, foldl1, foldr, foldr1, id, mapM, mapM_, maximum, minimum, notElem, or, product, sequence, sequence_, sum, (.))

@@ -0,0 +1,231 @@+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE NoFieldSelectors #-}++module TheatreDev.StmBased+  ( Actor,++    -- * Acquisition+    spawnStatefulIndividual,+    spawnStatefulBatched,+    spawnStatelessIndividual,+    spawnStatelessBatched,++    -- * Control+    tell,+    kill,+    wait,++    -- * Composition+    oneOf,+    allOf,+    byKeyHash,+  )+where++import TheatreDev.Prelude+import TheatreDev.StmBased.StmStructures.Runner (Runner)+import qualified TheatreDev.StmBased.StmStructures.Runner as Runner+import TheatreDev.StmBased.Tell (Tell)+import qualified TheatreDev.StmBased.Tell as Tell+import qualified TheatreDev.StmBased.Wait as Wait++-- |+-- Controls of an actor, which processes the messages of type @message@.+--+-- Provides abstraction over the message channel, thread-forking and killing.+--+-- Monoid instance is not provided for the same reason it is not provided for numbers.+-- This type supports both sum and product composition. See 'allOf' and 'oneOf'.+data Actor message = Actor+  { -- | Send a message to the actor.+    tell :: message -> STM (),+    -- | Kill the actor.+    kill :: STM (),+    -- | Wait for the actor to die due to error or being killed.+    wait :: STM (Maybe SomeException)+  }++instance Contravariant Actor where+  contramap fn (Actor tell kill wait) =+    Actor (tell . fn) kill wait++instance Divisible Actor where+  conquer =+    Actor (const (return ())) (return ()) (return Nothing)+  divide divisor (Actor lTell lKill lWait) (Actor rTell rKill rWait) =+    Actor tell kill wait+    where+      tell msg = case divisor msg of (lMsg, rMsg) -> lTell lMsg >> rTell rMsg+      kill = lKill >> rKill+      wait = Wait.both lWait rWait++instance Decidable Actor where+  lose fn =+    Actor (const (return ()) . absurd . fn) (return ()) (return Nothing)+  choose choice (Actor lTell lKill lWait) (Actor rTell rKill rWait) =+    Actor tell kill wait+    where+      tell = either lTell rTell . choice+      kill = lKill >> rKill+      wait = Wait.both lWait rWait++-- * Composition++fromRunner :: Runner a -> Actor a+fromRunner runner =+  Actor+    { tell = Runner.tell runner,+      kill = Runner.kill runner,+      wait = Runner.wait runner+    }++-- | Distribute the message stream across actors.+-- The message gets delivered to the first available one.+--+-- E.g., using this combinator in combination with 'replicateM'+-- you can construct pools:+--+-- > spawnPool :: Int -> IO (Actor message) -> IO (Actor message)+-- > spawnPool size spawn =+-- >   oneOf <$> replicateM size spawn+--+-- You can consider this being an interface to the Sum monoid.+oneOf :: [Actor message] -> Actor message+oneOf = tellComposition Tell.one++-- |+--+-- You can consider this being an interface to the Product monoid.+allOf :: [Actor message] -> Actor message+allOf = tellComposition Tell.all++-- |+-- Dispatch the message across actors based on a key hash.+--+-- This lets you ensure of a property that messages with+-- the same key will arrive to the same actor,+-- letting you maintain a local associated state in the actors.+--+-- The implementation applies a modulo equal to the amount+-- of actors to the hash and thus determines the index+-- of the actor to dispatch the message to.+--+-- This is inspired by how partitioning is done in Kafka.+byKeyHash ::+  -- | Function extracting the key from the message and hashing it.+  (message -> Int) ->+  -- | Pool of actors.+  [Actor message] ->+  Actor message+byKeyHash = tellComposition . Tell.byKeyHash++tellComposition :: ([Tell message] -> Tell message) -> [Actor message] -> Actor message+tellComposition tellReducer actors =+  Actor+    { tell = tellReducer (fmap (.tell) actors),+      kill = traverse_ (.kill) actors,+      wait = Wait.all (fmap (.wait) actors)+    }++-- * Acquisition++-- |+-- Given an interpreter of messages,+-- fork a thread to run the handler daemon on and+-- produce a handle to control that actor.+--+-- Killing that actor will make it process all the messages in the queue first.+-- All the messages sent to it after killing won't be processed.+spawnStatelessIndividual ::+  -- | Interpreter of a message.+  (message -> IO ()) ->+  -- | Clean up when killed.+  IO () ->+  -- | Fork a thread to run the handler daemon on and+  -- produce a handle to control it.+  IO (Actor message)+spawnStatelessIndividual interpreter cleaner =+  -- TODO: Optimize by reimplementing directly.+  spawnStatefulIndividual () (const interpreter) (const cleaner)++spawnStatelessBatched ::+  -- | Interpreter of a batch of messages.+  (NonEmpty message -> IO ()) ->+  -- | Clean up when killed.+  IO () ->+  -- | Fork a thread to run the handler daemon on and+  -- produce a handle to control it.+  IO (Actor message)+spawnStatelessBatched interpreter cleaner =+  -- TODO: Optimize by reimplementing directly.+  spawnStatefulBatched () (const interpreter) (const cleaner)++spawnStatefulIndividual ::+  state ->+  (state -> message -> IO state) ->+  (state -> IO ()) ->+  IO (Actor message)+spawnStatefulIndividual zero step finalizer =+  do+    runner <- atomically Runner.start+    forkIOWithUnmask $ \unmask ->+      let loop !state =+            do+              message <- atomically $ Runner.receiveSingle runner+              case message of+                Just message ->+                  do+                    result <- try @SomeException $ unmask $ step state message+                    case result of+                      Right newState ->+                        loop newState+                      Left exception ->+                        do+                          atomically $ Runner.releaseWithException runner exception+                          finalizer state+                Nothing ->+                  finalizer state+       in loop zero+    return $ fromRunner runner++spawnStatefulBatched ::+  state ->+  (state -> NonEmpty message -> IO state) ->+  (state -> IO ()) ->+  IO (Actor message)+spawnStatefulBatched zero step finalizer =+  do+    runner <- atomically Runner.start+    forkIOWithUnmask $ \unmask ->+      let loop !state =+            do+              messages <- atomically $ Runner.receiveMultiple runner+              case messages of+                Just nonEmptyMessages ->+                  do+                    result <- try @SomeException $ unmask $ step state nonEmptyMessages+                    case result of+                      Right newState ->+                        loop newState+                      Left exception ->+                        do+                          atomically $ Runner.releaseWithException runner exception+                          finalizer state+                -- Empty batch means that the runner is finished.+                Nothing -> finalizer state+       in loop zero+    return $ fromRunner runner++-- * Control++tell :: Actor message -> message -> IO ()+tell actor =+  atomically . actor.tell++kill :: Actor message -> IO ()+kill actor =+  atomically actor.kill++wait :: Actor message -> IO ()+wait actor =+  atomically actor.wait >>= maybe (pure ()) throwIO

@@ -0,0 +1,97 @@+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE NoFieldSelectors #-}++module TheatreDev.StmBased.StmStructures.Runner+  ( Runner,+    start,+    tell,+    kill,+    wait,+    receiveSingle,+    receiveMultiple,+    releaseWithException,+  )+where++import Control.Concurrent.STM.TBQueue+import Control.Concurrent.STM.TMVar+import qualified TheatreDev.ExtrasFor.List as List+import TheatreDev.ExtrasFor.TBQueue+import TheatreDev.Prelude++data Runner a = Runner+  { queue :: TBQueue (Maybe a),+    aliveVar :: TVar Bool,+    resVar :: TMVar (Maybe SomeException)+  }++start :: STM (Runner a)+start =+  do+    queue <- newTBQueue 1000+    aliveVar <- newTVar True+    resVar <- newEmptyTMVar @(Maybe SomeException)+    return Runner {..}++tell :: Runner a -> a -> STM ()+tell Runner {..} message =+  do+    alive <- readTVar aliveVar+    when alive do+      writeTBQueue queue $ Just message++kill :: Runner a -> STM ()+kill Runner {..} =+  do+    alive <- readTVar aliveVar+    when alive do+      writeTBQueue queue Nothing++wait :: Runner a -> STM (Maybe SomeException)+wait Runner {..} =+  readTMVar resVar++receiveSingle ::+  Runner a ->+  -- | Action producing a message or nothing, after it's killed.+  STM (Maybe a)+receiveSingle Runner {..} =+  do+    message <- readTBQueue queue+    case message of+      Just message -> return (Just message)+      Nothing -> do+        writeTVar aliveVar False+        putTMVar resVar Nothing+        return Nothing++receiveMultiple ::+  Runner a ->+  STM (Maybe (NonEmpty a))+receiveMultiple Runner {..} =+  do+    (messages, remainingCommands) <- do+      queueLength <- lengthTBQueue queue+      head <- readTBQueue queue+      tail <- simplerFlushTBQueue queue+      return $ List.splitWhileJust $ head : tail+    case messages of+      -- Implies that the tail is not empty,+      -- because we have at least one element.+      -- And that it starts with a Nothing.+      [] -> do+        forM_ remainingCommands $ unGetTBQueue queue+        writeTVar aliveVar False+        putTMVar resVar Nothing+        return Nothing+      messagesHead : messagesTail -> do+        unless (null remainingCommands) do+          unGetTBQueue queue Nothing+        return $ Just $ messagesHead :| messagesTail++releaseWithException :: Runner a -> SomeException -> STM ()+releaseWithException Runner {..} exception =+  do+    simplerFlushTBQueue queue+    writeTVar aliveVar False+    putTMVar resVar (Just exception)

@@ -0,0 +1,33 @@+module TheatreDev.StmBased.Tell where++import qualified Data.Vector as Vector+import TheatreDev.Prelude++type Tell a = a -> STM ()++either :: Tell a -> Tell a -> Tell a+either lTell rTell msg =+  lTell msg <|> rTell msg++both :: Tell a -> Tell a -> Tell a+both lTell rTell msg =+  lTell msg >> rTell msg++one :: [Tell a] -> Tell a+one tells msg =+  asum $ fmap (\tell -> tell msg) tells++all :: [Tell a] -> Tell a+all tells msg =+  traverse_ (\tell -> tell msg) tells++byKeyHash :: (a -> Int) -> [Tell a] -> Tell a+byKeyHash proj tells =+  let vector = Vector.fromList tells+      vectorLength = Vector.length vector+   in case vectorLength of+        0 -> const (pure ())+        _ -> \msg ->+          let index = mod (proj msg) vectorLength+              tellAtIndex = Vector.unsafeIndex vector index+           in tellAtIndex msg

@@ -0,0 +1,27 @@+module TheatreDev.StmBased.Wait where++import TheatreDev.Prelude++type Wait = STM (Maybe SomeException)++both :: Wait -> Wait -> Wait+both left right =+  do+    firstResult <- Left <$> left <|> Right <$> right+    case firstResult of+      Left Nothing -> right+      Left (Just exception) -> return (Just exception)+      Right Nothing -> left+      Right (Just exception) -> return (Just exception)++all :: [Wait] -> Wait+all waits =+  getException <|> getNothing+  where+    getException =+      waits+        & fmap (>>= maybe empty pure)+        & asum+        & fmap Just+    getNothing =+      Nothing <$ sequence_ waits

@@ -0,0 +1,200 @@+module TheatreDev.Terminal.Actor+  ( Actor,++    -- * Manipulation+    adaptToList,++    -- * Acquisition+    spawnStatelessGranular,+    spawnStatefulGranular,+    spawnStatefulBatched,++    -- * Control+    tell,+    kill,+    wait,+  )+where++import qualified Control.Concurrent.Chan.Unagi as E+import Control.Concurrent.STM.TBQueue+import Control.Concurrent.STM.TMVar+import qualified TheatreDev.ExtrasFor.List as List+import TheatreDev.ExtrasFor.TBQueue+import TheatreDev.Prelude++-- |+-- Controls of an actor, which processes the messages of type @message@.+--+-- Abstraction over the message channel, thread-forking and killing.+data Actor message = Actor+  { -- | Send a message to the actor.+    tell :: message -> IO (),+    -- | Kill the actor.+    kill :: IO (),+    -- | Wait for the actor to die due to error or being killed.+    wait :: IO ()+  }++instance Semigroup (Actor message) where+  (<>) (Actor lTell lKill lWait) (Actor rTell rKill rWait) =+    Actor tell kill wait+    where+      tell msg = lTell msg >> rTell msg+      kill = lKill >> rKill+      wait = lWait >> rWait++instance Monoid (Actor message) where+  mempty =+    Actor (const (return ())) (return ()) (return ())++instance Contravariant Actor where+  contramap fn (Actor tell kill wait) =+    Actor (tell . fn) kill wait++instance Divisible Actor where+  conquer =+    mempty+  divide divisor (Actor lTell lKill lWait) (Actor rTell rKill rWait) =+    Actor tell kill wait+    where+      tell msg = case divisor msg of (lMsg, rMsg) -> lTell lMsg >> rTell rMsg+      kill = lKill >> rKill+      wait = lWait >> rWait++instance Decidable Actor where+  lose fn =+    Actor (const (return ()) . absurd . fn) (return ()) (return ())+  choose choice (Actor lTell lKill lWait) (Actor rTell rKill rWait) =+    Actor tell kill wait+    where+      tell = either lTell rTell . choice+      kill = lKill >> rKill+      wait = lWait >> rWait++-- |+-- Adapt the actor to be able to receive lists of messages.+adaptToList :: Actor message -> Actor [message]+adaptToList Actor {..} =+  case traverse_ tell of+    tell -> Actor {..}++-- |+-- Given an interpreter of messages,+-- fork a thread to run the handler daemon on and+-- produce a handle to control that actor.+--+-- Killing that actor will make it process all the messages in the queue first.+-- All the messages sent to it after killing won't be processed.+spawnStatelessGranular ::+  -- | Interpreter of a message.+  (message -> IO ()) ->+  -- | Clean up when killed.+  IO () ->+  -- | Fork a thread to run the handler daemon on and+  -- produce a handle to control it.+  IO (Actor message)+spawnStatelessGranular interpretMessage cleanUp =+  do+    (inChan, outChan) <- E.newChan+    lock <- newEmptyMVar+    spawningThreadId <- myThreadId+    forkIO+      $ let loop =+              {-# SCC "spawnStatelessGranular/loop" #-}+              do+                message <- E.readChan outChan+                case message of+                  Just payload ->+                    do+                      res <- try @SomeException $ interpretMessage payload+                      case res of+                        Right () -> loop+                        Left exc ->+                          do+                            cleanUp+                            putMVar lock ()+                            throwTo spawningThreadId exc+                  Nothing ->+                    do+                      cleanUp+                      putMVar lock ()+         in loop+    return+      ( Actor+          (E.writeChan inChan . Just)+          (E.writeChan inChan Nothing)+          (takeMVar lock)+      )++-- |+-- Actor with memory.+--+-- Threads a persistent state thru its iterations.+--+-- Given an interpreter of messages and initial state generator,+-- forks a thread to run the computation on and+-- produces a handle to address that actor.+--+-- Killing that actor will make it process all the messages in the queue first.+-- All the messages sent to it after killing won't be processed.+spawnStatefulGranular :: state -> (state -> message -> IO state) -> (state -> IO ()) -> IO (Actor message)+spawnStatefulGranular zero step finalizer =+  spawnStatefulBatched zero newStep finalizer+  where+    newStep =+      foldM step++spawnStatefulBatched :: state -> (state -> NonEmpty message -> IO state) -> (state -> IO ()) -> IO (Actor message)+spawnStatefulBatched zero step finalizer =+  do+    queue <- newTBQueueIO 1000+    aliveVar <- newTVarIO True+    resVar <- newEmptyTMVarIO @(Maybe SomeException)+    forkIOWithUnmask $ \unmask ->+      let loop !state =+            join $ atomically $ do+              flushing <- flushNonEmptyTBQueue queue+              let (messages, flushingTail) = List.splitWhileJust (toList flushing)+              case messages of+                -- Automatically means that the tail is not empty.+                [] -> do+                  writeTVar aliveVar False+                  putTMVar resVar Nothing+                  return $ do+                    finalizer state+                messagesHead : messagesTail ->+                  return $ do+                    result <- try @SomeException $ unmask $ step state (messagesHead :| messagesTail)+                    case result of+                      Right newState ->+                        case flushingTail of+                          [] -> loop newState+                          _ -> do+                            atomically $ do+                              writeTVar aliveVar False+                              putTMVar resVar Nothing+                            finalizer state+                      Left exception -> do+                        atomically $ do+                          writeTVar aliveVar False+                          putTMVar resVar Nothing+                        finalizer state+       in loop zero+    return+      Actor+        { tell = \message -> atomically $ do+            alive <- readTVar aliveVar+            when alive+              $ writeTBQueue queue+              $ Just message,+          kill = atomically $ do+            alive <- readTVar aliveVar+            when alive+              $ writeTBQueue queue Nothing,+          wait = do+            res <- atomically $ takeTMVar resVar+            case res of+              Nothing -> return ()+              Just exception -> throwIO exception+        }

@@ -0,0 +1,39 @@+module TheatreDev.Terminal.StatefulActorSpec where++import TheatreDev.Prelude++data StatefulActorSpec message = forall state.+  StatefulActorSpec+  { enter :: Concurrently state,+    step :: state -> NonEmpty message -> Concurrently state,+    exit :: state -> Concurrently ()+  }++instance Semigroup (StatefulActorSpec message) where+  StatefulActorSpec leftEnter leftStep leftExit <> StatefulActorSpec rightEnter rightStep rightExit =+    StatefulActorSpec+      { enter =+          (,) <$> leftEnter <*> rightEnter,+        step = \(leftState, rightState) messages ->+          (,)+            <$> leftStep leftState messages+            <*> rightStep rightState messages,+        exit = \(leftState, rightState) ->+          leftExit leftState *> rightExit rightState+      }++instance Monoid (StatefulActorSpec message) where+  mempty =+    StatefulActorSpec+      { enter = pure (),+        step = const $ const $ pure (),+        exit = const $ pure ()+      }++individual :: IO state -> (state -> message -> IO state) -> (state -> IO ()) -> StatefulActorSpec message+individual enter step exit =+  StatefulActorSpec+    { enter = Concurrently enter,+      step = \state messages -> Concurrently (foldM step state messages),+      exit = \state -> Concurrently (exit state)+    }

@@ -0,0 +1,114 @@+cabal-version: 3.0+name:          theatre-dev+version:       0.0.1+category:      Concurrency, Actors+synopsis:      Minimalistic actor library experiments+description:+  Design space exploration for the \"theatre\" library.+  Don\'t expect this lib to maintain a stable API.+  Once clearly useful abstractions emerge, they'll be moved to the+  \"theatre\" lib.+  The namespace "TheatreDev.*" implies the name of an experimental API.++stability:     experimental+homepage:      https://github.com/nikita-volkov/theatre-dev+bug-reports:   https://github.com/nikita-volkov/theatre-dev/issues+author:        Nikita Volkov <nikita.y.volkov@mail.ru>+maintainer:    Nikita Volkov <nikita.y.volkov@mail.ru>+copyright:     (c) 2022, Nikita Volkov+license:       MIT+license-file:  LICENSE++source-repository head+  type:     git+  location: git://github.com/nikita-volkov/theatre-dev.git++common base+  default-language:   Haskell2010+  default-extensions:+    NoImplicitPrelude+    NoMonomorphismRestriction+    ApplicativeDo+    Arrows+    BangPatterns+    BlockArguments+    ConstraintKinds+    DataKinds+    DefaultSignatures+    DeriveDataTypeable+    DeriveFoldable+    DeriveFunctor+    DeriveGeneric+    DeriveTraversable+    EmptyDataDecls+    FlexibleContexts+    FlexibleInstances+    FunctionalDependencies+    GADTs+    GeneralizedNewtypeDeriving+    LambdaCase+    LiberalTypeSynonyms+    MagicHash+    MultiParamTypeClasses+    MultiWayIf+    NamedFieldPuns+    OverloadedRecordDot+    OverloadedStrings+    ParallelListComp+    PatternGuards+    PatternSynonyms+    QuasiQuotes+    RankNTypes+    RecordWildCards+    ScopedTypeVariables+    StandaloneDeriving+    TemplateHaskell+    TupleSections+    TypeApplications+    TypeFamilies+    TypeOperators+    UnboxedTuples++common executable+  import:      base+  ghc-options: -O2 -threaded "-with-rtsopts=-N -I0 -qg"++common test+  import:      base+  ghc-options: -threaded -with-rtsopts=-N++library+  import:          base+  hs-source-dirs:  library+  exposed-modules:+    TheatreDev.Perpetual+    TheatreDev.StmBased+    TheatreDev.Terminal.Actor++  other-modules:+    TheatreDev.ExtrasFor.List+    TheatreDev.ExtrasFor.TBQueue+    TheatreDev.Prelude+    TheatreDev.StmBased.StmStructures.Runner+    TheatreDev.StmBased.Tell+    TheatreDev.StmBased.Wait+    TheatreDev.Terminal.StatefulActorSpec++  build-depends:+    , async >=2.2 && <3+    , base >=4.16 && <5+    , contravariant >=1.3 && <2+    , stm >=2.5.2 && <3+    , unagi-chan >=0.4.1.4 && <0.5+    , vector >=0.13 && <0.14++test-suite hspec+  import:         test+  type:           exitcode-stdio-1.0+  hs-source-dirs: hspec+  main-is:        Main.hs+  other-modules:  TheatreDev.StmBasedSpec+  build-depends:+    , hspec >=2.11.6 && <3+    , rerebase <2+    , theatre-dev