diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/hspec/Main.hs b/hspec/Main.hs new file mode 100644 --- /dev/null +++ b/hspec/Main.hs @@ -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 diff --git a/hspec/TheatreDev/StmBasedSpec.hs b/hspec/TheatreDev/StmBasedSpec.hs new file mode 100644 --- /dev/null +++ b/hspec/TheatreDev/StmBasedSpec.hs @@ -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 diff --git a/library/TheatreDev/ExtrasFor/List.hs b/library/TheatreDev/ExtrasFor/List.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/ExtrasFor/List.hs @@ -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, []) diff --git a/library/TheatreDev/ExtrasFor/TBQueue.hs b/library/TheatreDev/ExtrasFor/TBQueue.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/ExtrasFor/TBQueue.hs @@ -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 diff --git a/library/TheatreDev/Perpetual.hs b/library/TheatreDev/Perpetual.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/Perpetual.hs @@ -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 diff --git a/library/TheatreDev/Prelude.hs b/library/TheatreDev/Prelude.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/Prelude.hs @@ -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, (.)) diff --git a/library/TheatreDev/StmBased.hs b/library/TheatreDev/StmBased.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/StmBased.hs @@ -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 diff --git a/library/TheatreDev/StmBased/StmStructures/Runner.hs b/library/TheatreDev/StmBased/StmStructures/Runner.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/StmBased/StmStructures/Runner.hs @@ -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) diff --git a/library/TheatreDev/StmBased/Tell.hs b/library/TheatreDev/StmBased/Tell.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/StmBased/Tell.hs @@ -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 diff --git a/library/TheatreDev/StmBased/Wait.hs b/library/TheatreDev/StmBased/Wait.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/StmBased/Wait.hs @@ -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 diff --git a/library/TheatreDev/Terminal/Actor.hs b/library/TheatreDev/Terminal/Actor.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/Terminal/Actor.hs @@ -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 + } diff --git a/library/TheatreDev/Terminal/StatefulActorSpec.hs b/library/TheatreDev/Terminal/StatefulActorSpec.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/Terminal/StatefulActorSpec.hs @@ -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) + } diff --git a/theatre-dev.cabal b/theatre-dev.cabal new file mode 100644 --- /dev/null +++ b/theatre-dev.cabal @@ -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 +maintainer: Nikita Volkov +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