diff --git a/hspec/Main.hs b/hspec/Main.hs --- a/hspec/Main.hs +++ b/hspec/Main.hs @@ -1,10 +1,10 @@ module Main (main) where import Test.Hspec -import TheatreDevSpec qualified +import TheatreDev.ActorSpec qualified import Prelude main :: IO () main = hspec do - describe "TheatreDev" TheatreDevSpec.spec + describe "TheatreDev.Actor" TheatreDev.ActorSpec.spec diff --git a/hspec/TheatreDev/ActorSpec.hs b/hspec/TheatreDev/ActorSpec.hs new file mode 100644 --- /dev/null +++ b/hspec/TheatreDev/ActorSpec.hs @@ -0,0 +1,166 @@ +{-# OPTIONS_GHC -Wno-unused-local-binds -Wno-unused-binds #-} + +module TheatreDev.ActorSpec (spec) where + +import Control.Concurrent.Async +import Data.IntMap.Strict qualified as IntMap +import Data.IntSet qualified as IntSet +import Test.Hspec +import Test.Hspec.QuickCheck +import Test.QuickCheck +import TheatreDev.Actor qualified as Actor +import TheatreDev.ActorSpec.IO qualified as IO +import TheatreDev.ActorSpec.Preferences qualified as Preferences +import Prelude + +spec :: Spec +spec = + do + describe "kill" do + describe "When full" do + it "Does not block" pending + it "Lets all messages scheduled before be processed" pending + it "Makes all messages scheduled after be ignored" pending + + describe "wait" do + describe "When full" do + it "Blocks until all messages are processed" pending + + describe "spawnStatefulBatched" do + let spawnIntUpdater step = Actor.spawnStatefulBatched @Int 0 (const (return ())) step + let spawnUnit step = Actor.spawnStatefulBatched () (const (return ())) step + + 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 + Actor.kill actor + + collectedBatches <- reverse . fmap toList <$> readIORef acc + shouldBe collectedBatches [[1], [2, 3], [4]] + + it "Threads the state" do + let input = [0 .. 9] + inputLength = length input + + resultsVar <- newTVarIO [] + actor <- Actor.spawnStatefulBatched [] (const (return ())) $ \state msgs -> do + let !newState = foldl' (flip (:)) state msgs + atomically $ writeTVar resultsVar newState + return newState + traverse_ (Actor.tell actor) input + + results <- atomically $ do + results <- readTVar resultsVar + if length results < inputLength + then retry + else return results + + shouldBe (reverse results) input + + it "Kill and wait" do + let input = [0 .. 9] + inputLength = length input + + resultVar <- newEmptyMVar + actor <- + Actor.spawnStatefulBatched + [] + ( \state -> do + threadDelay 1000 + putMVar resultVar state + ) + ( \state msgs -> return $ foldl' (flip (:)) state msgs + ) + traverse_ (Actor.tell actor) input + + Actor.kill actor + Actor.wait actor + + result <- takeMVar resultVar + shouldBe result $ reverse input + + describe "allOf" . modifyMaxSuccess (max 1000) $ do + it "Passes 1" do + let emittersNum = 2 + messagesNum = 10 + actorsNum = 3 + messages = [0 .. messagesNum - 1] + results <- fmap (fmap sort) (IO.simulateReduction actorsNum emittersNum Actor.allOf messages) + shouldBe results (replicate actorsNum (sort (concat (replicate emittersNum messages)))) + shouldBe (getSum (foldMap (Sum . length) results)) (actorsNum * emittersNum * messagesNum) + prop "" $ forAll (chooseInt (0, 99)) $ \size -> forAll arbitrary $ \(messages :: [Int]) -> + idempotentIOProperty do + results <- sort . concat <$> IO.simulateReduction Preferences.concurrency size Actor.allOf messages + return + $ conjoin + [ length results === length messages * size * Preferences.concurrency, + results === sort (concat (replicate (size * Preferences.concurrency) messages)) + ] + + describe "firstAvailableOneOf" . modifyMaxSuccess (max 1000) $ do + prop "Dispatches correctly" $ forAll (chooseInt (0, 99)) $ \size -> forAll arbitrary $ \(messages :: [Int]) -> + idempotentIOProperty do + results <- sort . concat <$> IO.simulateReduction Preferences.concurrency size Actor.firstAvailableOneOf messages + return + $ conjoin + [ length results === length messages * size, + results + === sort (concat (replicate (size) messages)) + ] + + describe "byKeyHashOneOf" . modifyMaxSuccess (max Preferences.largePropertyMaxSuccess) $ do + prop "Dispatches individually" $ forAll (chooseInt (0, 99)) $ \size -> forAll arbitrary $ \(messages :: [Int]) -> idempotentIOProperty $ do + resultsVar <- newTVarIO [] + actor <- + fmap (Actor.byKeyHashOneOf id) + $ replicateM size + $ Actor.spawnStatefulIndividual + IntMap.empty + ( \state -> + atomically + $ modifyTVar' resultsVar + $ (:) state + ) + ( \state msg -> + return $ IntMap.alter (Just . maybe 1 succ) msg state + ) + + mapConcurrently id + $ replicate Preferences.concurrency + $ for_ messages + $ Actor.tell actor + + Actor.kill actor + Actor.wait actor + + results <- readTVarIO resultsVar + + let allKeys = results >>= IntMap.keys + nubbedKeys = nub allKeys + + return + $ conjoin + [ allKeys === nubbedKeys, + if size == 0 + then nubbedKeys === [] + else IntSet.fromList nubbedKeys === IntSet.fromList messages + ] diff --git a/hspec/TheatreDev/ActorSpec/IO.hs b/hspec/TheatreDev/ActorSpec/IO.hs new file mode 100644 --- /dev/null +++ b/hspec/TheatreDev/ActorSpec/IO.hs @@ -0,0 +1,34 @@ +module TheatreDev.ActorSpec.IO where + +import Control.Concurrent.Async +import TheatreDev.Actor (Actor) +import TheatreDev.Actor qualified as Actor +import Prelude + +simulateReduction :: Int -> Int -> ([Actor a] -> Actor a) -> [a] -> IO [[a]] +simulateReduction actorsNum generatorsNum reducer messages = + do + resultsVar <- newTVarIO [] + + actor <- + fmap reducer + $ replicateM actorsNum + $ Actor.spawnStatefulIndividual + [] + ( \state -> + atomically + $ modifyTVar' resultsVar (reverse state :) + ) + ( \state msg -> + return $ msg : state + ) + + mapConcurrently id + $ replicate generatorsNum + $ for_ messages + $ Actor.tell actor + + Actor.kill actor + Actor.wait actor + + readTVarIO resultsVar diff --git a/hspec/TheatreDev/ActorSpec/Preferences.hs b/hspec/TheatreDev/ActorSpec/Preferences.hs new file mode 100644 --- /dev/null +++ b/hspec/TheatreDev/ActorSpec/Preferences.hs @@ -0,0 +1,9 @@ +module TheatreDev.ActorSpec.Preferences where + +import Prelude + +concurrency :: Int +concurrency = 7 + +largePropertyMaxSuccess :: Int +largePropertyMaxSuccess = 10000 diff --git a/hspec/TheatreDevSpec.hs b/hspec/TheatreDevSpec.hs deleted file mode 100644 --- a/hspec/TheatreDevSpec.hs +++ /dev/null @@ -1,166 +0,0 @@ -{-# OPTIONS_GHC -Wno-unused-local-binds -Wno-unused-binds #-} - -module TheatreDevSpec (spec) where - -import Control.Concurrent.Async -import Data.IntMap.Strict qualified as IntMap -import Data.IntSet qualified as IntSet -import Test.Hspec -import Test.Hspec.QuickCheck -import Test.QuickCheck -import TheatreDev qualified as Actor -import TheatreDevSpec.IO qualified as IO -import TheatreDevSpec.Preferences qualified as Preferences -import Prelude - -spec :: Spec -spec = - do - describe "kill" do - describe "When full" do - it "Does not block" pending - it "Lets all messages scheduled before be processed" pending - it "Makes all messages scheduled after be ignored" pending - - describe "wait" do - describe "When full" do - it "Blocks until all messages are processed" pending - - describe "spawnStatefulBatched" do - let spawnIntUpdater step = Actor.spawnStatefulBatched @Int 0 (const (return ())) step - let spawnUnit step = Actor.spawnStatefulBatched () (const (return ())) step - - 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 - Actor.kill actor - - collectedBatches <- reverse . fmap toList <$> readIORef acc - shouldBe collectedBatches [[1], [2, 3], [4]] - - it "Threads the state" do - let input = [0 .. 9] - inputLength = length input - - resultsVar <- newTVarIO [] - actor <- Actor.spawnStatefulBatched [] (const (return ())) $ \state msgs -> do - let !newState = foldl' (flip (:)) state msgs - atomically $ writeTVar resultsVar newState - return newState - traverse_ (Actor.tell actor) input - - results <- atomically $ do - results <- readTVar resultsVar - if length results < inputLength - then retry - else return results - - shouldBe (reverse results) input - - it "Kill and wait" do - let input = [0 .. 9] - inputLength = length input - - resultVar <- newEmptyMVar - actor <- - Actor.spawnStatefulBatched - [] - ( \state -> do - threadDelay 1000 - putMVar resultVar state - ) - ( \state msgs -> return $ foldl' (flip (:)) state msgs - ) - traverse_ (Actor.tell actor) input - - Actor.kill actor - Actor.wait actor - - result <- takeMVar resultVar - shouldBe result $ reverse input - - describe "allOf" . modifyMaxSuccess (max 1000) $ do - it "Passes 1" do - let emittersNum = 2 - messagesNum = 10 - actorsNum = 3 - messages = [0 .. messagesNum - 1] - results <- fmap (fmap sort) (IO.simulateReduction actorsNum emittersNum Actor.allOf messages) - shouldBe results (replicate actorsNum (sort (concat (replicate emittersNum messages)))) - shouldBe (getSum (foldMap (Sum . length) results)) (actorsNum * emittersNum * messagesNum) - prop "" $ forAll (chooseInt (0, 99)) $ \size -> forAll arbitrary $ \(messages :: [Int]) -> - idempotentIOProperty do - results <- sort . concat <$> IO.simulateReduction Preferences.concurrency size Actor.allOf messages - return - $ conjoin - [ length results === length messages * size * Preferences.concurrency, - results === sort (concat (replicate (size * Preferences.concurrency) messages)) - ] - - describe "firstAvailableOneOf" . modifyMaxSuccess (max 1000) $ do - prop "Dispatches correctly" $ forAll (chooseInt (0, 99)) $ \size -> forAll arbitrary $ \(messages :: [Int]) -> - idempotentIOProperty do - results <- sort . concat <$> IO.simulateReduction Preferences.concurrency size Actor.firstAvailableOneOf messages - return - $ conjoin - [ length results === length messages * size, - results - === sort (concat (replicate (size) messages)) - ] - - describe "byKeyHashOneOf" . modifyMaxSuccess (max Preferences.largePropertyMaxSuccess) $ do - prop "Dispatches individually" $ forAll (chooseInt (0, 99)) $ \size -> forAll arbitrary $ \(messages :: [Int]) -> idempotentIOProperty $ do - resultsVar <- newTVarIO [] - actor <- - fmap (Actor.byKeyHashOneOf id) - $ replicateM size - $ Actor.spawnStatefulIndividual - IntMap.empty - ( \state -> - atomically - $ modifyTVar' resultsVar - $ (:) state - ) - ( \state msg -> - return $ IntMap.alter (Just . maybe 1 succ) msg state - ) - - mapConcurrently id - $ replicate Preferences.concurrency - $ for_ messages - $ Actor.tell actor - - Actor.kill actor - Actor.wait actor - - results <- readTVarIO resultsVar - - let allKeys = results >>= IntMap.keys - nubbedKeys = nub allKeys - - return - $ conjoin - [ allKeys === nubbedKeys, - if size == 0 - then nubbedKeys === [] - else IntSet.fromList nubbedKeys === IntSet.fromList messages - ] diff --git a/hspec/TheatreDevSpec/IO.hs b/hspec/TheatreDevSpec/IO.hs deleted file mode 100644 --- a/hspec/TheatreDevSpec/IO.hs +++ /dev/null @@ -1,34 +0,0 @@ -module TheatreDevSpec.IO where - -import Control.Concurrent.Async -import TheatreDev (Actor) -import TheatreDev qualified as Actor -import Prelude - -simulateReduction :: Int -> Int -> ([Actor a] -> Actor a) -> [a] -> IO [[a]] -simulateReduction actorsNum generatorsNum reducer messages = - do - resultsVar <- newTVarIO [] - - actor <- - fmap reducer - $ replicateM actorsNum - $ Actor.spawnStatefulIndividual - [] - ( \state -> - atomically - $ modifyTVar' resultsVar (reverse state :) - ) - ( \state msg -> - return $ msg : state - ) - - mapConcurrently id - $ replicate generatorsNum - $ for_ messages - $ Actor.tell actor - - Actor.kill actor - Actor.wait actor - - readTVarIO resultsVar diff --git a/hspec/TheatreDevSpec/Preferences.hs b/hspec/TheatreDevSpec/Preferences.hs deleted file mode 100644 --- a/hspec/TheatreDevSpec/Preferences.hs +++ /dev/null @@ -1,9 +0,0 @@ -module TheatreDevSpec.Preferences where - -import Prelude - -concurrency :: Int -concurrency = 7 - -largePropertyMaxSuccess :: Int -largePropertyMaxSuccess = 10000 diff --git a/library/TheatreDev.hs b/library/TheatreDev.hs deleted file mode 100644 --- a/library/TheatreDev.hs +++ /dev/null @@ -1,234 +0,0 @@ -module TheatreDev - ( Actor, - - -- * Acquisition - spawnStatefulIndividual, - spawnStatefulBatched, - spawnStatelessIndividual, - spawnStatelessBatched, - - -- * Control - tell, - kill, - wait, - - -- * Composition - firstAvailableOneOf, - byKeyHashOneOf, - allOf, - ) -where - -import TheatreDev.Prelude -import TheatreDev.StmStructures.Runner (Runner) -import TheatreDev.StmStructures.Runner qualified as Runner -import TheatreDev.Tell (Tell) -import TheatreDev.Tell qualified as Tell -import TheatreDev.Wait qualified as Wait - --- | --- Controls of an actor, which processes the messages of type @message@. --- The processing runs on a dedicated green thread. --- --- 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', 'firstAvailableOneOf' and 'byKeyHashOneOf'. -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), - -- | IDs of the constituent actors. - -- Useful for debugging. - ids :: [UUID] - } - -instance Contravariant Actor where - contramap fn (Actor tell kill wait ids) = - Actor (tell . fn) kill wait ids - -instance Divisible Actor where - conquer = - Actor (const (return ())) (return ()) (return Nothing) [] - divide divisor (Actor lTell lKill lWait lIds) (Actor rTell rKill rWait rIds) = - Actor - { tell = \msg -> case divisor msg of (lMsg, rMsg) -> lTell lMsg >> rTell rMsg, - kill = lKill >> rKill, - wait = Wait.both lWait rWait, - ids = lIds <> rIds - } - -instance Decidable Actor where - lose fn = - Actor (const (return ()) . absurd . fn) (return ()) (return Nothing) [] - choose choice (Actor lTell lKill lWait lIds) (Actor rTell rKill rWait rIds) = - Actor - { tell = either lTell rTell . choice, - kill = lKill >> rKill, - wait = Wait.both lWait rWait, - ids = lIds <> rIds - } - --- * Composition - -fromRunner :: Runner a -> Actor a -fromRunner runner = - Actor - { tell = Runner.tell runner, - kill = Runner.kill runner, - wait = Runner.wait runner, - ids = [Runner.getId 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 = --- > firstAvailableOneOf <$> replicateM size spawn --- --- You can consider this being an interface to the Sum monoid. -firstAvailableOneOf :: [Actor message] -> Actor message -firstAvailableOneOf = tellComposition Tell.one - --- | --- 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. -byKeyHashOneOf :: - -- | Function extracting the key from the message and hashing it. - (message -> Int) -> - -- | Pool of actors. - [Actor message] -> - Actor message -byKeyHashOneOf = tellComposition . Tell.byKeyHashOneOf - --- | Distribute the message stream to all provided actors. --- --- You can consider this being an interface to the Product monoid. -allOf :: [Actor message] -> Actor message -allOf = tellComposition Tell.all - -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), - ids = foldMap (.ids) actors - } - --- * Acquisition - --- | Spawn an actor which processes messages in isolated executions. -spawnStatelessIndividual :: - -- | Clean up when killed or exception is thrown. - IO () -> - -- | Interpret a message. - (message -> IO ()) -> - -- | Fork a thread to run the handler loop on and produce a handle to control it. - IO (Actor message) -spawnStatelessIndividual cleaner interpreter = - -- TODO: Optimize by reimplementing directly. - spawnStatefulIndividual () (const cleaner) (const interpreter) - --- | Spawn an actor which processes all available messages in one execution. -spawnStatelessBatched :: - -- | Clean up when killed or exception is thrown. - IO () -> - -- | Interpret a batch of messages. - (NonEmpty message -> IO ()) -> - -- | Fork a thread to run the handler loop on and produce a handle to control it. - IO (Actor message) -spawnStatelessBatched cleaner interpreter = - -- TODO: Optimize by reimplementing directly. - spawnStatefulBatched () (const cleaner) (const interpreter) - --- | Spawn an actor which processes messages in isolated executions --- and threads state. -spawnStatefulIndividual :: - -- | Initial state. - state -> - -- | Clean up when killed or exception is thrown. - (state -> IO ()) -> - -- | Process a message and update state. - (state -> message -> IO state) -> - -- | Fork a thread to run the handler loop on and produce a handle to control it. - IO (Actor message) -spawnStatefulIndividual zero finalizer step = - spawnStatefulBatched zero finalizer $ foldM step - --- | Spawn an actor which processes all available messages in one execution --- and threads state. -spawnStatefulBatched :: - -- | Initial state. - state -> - -- | Clean up when killed or exception is thrown. - (state -> IO ()) -> - -- | Process a batch of messages and update state. - (state -> NonEmpty message -> IO state) -> - -- | Fork a thread to run the handler loop on and produce a handle to control it. - IO (Actor message) -spawnStatefulBatched zero finalizer step = - do - runner <- Runner.start - forkIOWithUnmask $ \unmask -> - let loop !state = - do - messages <- atomically $ Runner.receiveMultiple runner - case messages of - Just nonEmptyMessages -> - do - result <- try $ unmask $ step state nonEmptyMessages - case result of - Right newState -> - loop newState - Left exception -> - finally (finalizer state) - $ atomically - $ Runner.releaseWithException runner exception - -- Empty batch means that the runner is finished. - Nothing -> - finally (finalizer state) - $ atomically - $ Runner.releaseNormally runner - in loop zero - return $ fromRunner runner - --- * Control - --- | Add a message to the end of the queue of the --- messages to be processed by the provided actor. -tell :: Actor message -> message -> IO () -tell actor = - atomically . actor.tell - --- | Command the actor to stop registering new messages, --- process all the registered ones and execute the clean up action. --- --- This action executes immediately. --- If you want to block waiting for the actor to actually die, --- after 'kill' you can run 'wait'. -kill :: Actor message -> IO () -kill actor = - atomically actor.kill - --- | Block waiting for the actor to die either due to getting killed --- or due to its interpreter action throwing an exception. --- The exception will get rethrown here. -wait :: Actor message -> IO () -wait actor = - atomically actor.wait >>= maybe (pure ()) throwIO diff --git a/library/TheatreDev/Actor.hs b/library/TheatreDev/Actor.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/Actor.hs @@ -0,0 +1,236 @@ +module TheatreDev.Actor + ( Actor, + + -- * Acquisition + spawnStatefulIndividual, + spawnStatefulBatched, + spawnStatelessIndividual, + spawnStatelessBatched, + + -- * Control + tell, + kill, + wait, + + -- * Composition + firstAvailableOneOf, + byKeyHashOneOf, + allOf, + ) +where + +import TheatreDev.Prelude +import TheatreDev.StmStructures.Runner (Runner) +import TheatreDev.StmStructures.Runner qualified as Runner +import TheatreDev.Tell (Tell) +import TheatreDev.Tell qualified as Tell +import TheatreDev.Wait qualified as Wait + +-- | +-- Controls of an actor, which processes the messages of type @message@. +-- The processing runs on a dedicated green thread. +-- +-- 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', 'firstAvailableOneOf' and 'byKeyHashOneOf'. +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), + -- | IDs of the constituent actors. + -- Useful for debugging. + ids :: [UUID] + } + +instance Contravariant Actor where + contramap fn (Actor tell kill wait ids) = + Actor (tell . fn) kill wait ids + +instance Divisible Actor where + conquer = + Actor (const (return ())) (return ()) (return Nothing) [] + divide divisor (Actor lTell lKill lWait lIds) (Actor rTell rKill rWait rIds) = + Actor + { tell = \msg -> case divisor msg of (lMsg, rMsg) -> lTell lMsg >> rTell rMsg, + kill = lKill >> rKill, + wait = Wait.both lWait rWait, + ids = lIds <> rIds + } + +instance Decidable Actor where + lose fn = + Actor (const (return ()) . absurd . fn) (return ()) (return Nothing) [] + choose choice (Actor lTell lKill lWait lIds) (Actor rTell rKill rWait rIds) = + Actor + { tell = either lTell rTell . choice, + kill = lKill >> rKill, + wait = Wait.both lWait rWait, + ids = lIds <> rIds + } + +-- * Composition + +-- | 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 = +-- > firstAvailableOneOf <$> replicateM size spawn +-- +-- You can consider this being an interface to the Sum monoid. +firstAvailableOneOf :: [Actor message] -> Actor message +firstAvailableOneOf = tellComposition Tell.one + +-- | +-- 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. +byKeyHashOneOf :: + -- | Function extracting the key from the message and hashing it. + (message -> Int) -> + -- | Pool of actors. + [Actor message] -> + Actor message +byKeyHashOneOf = tellComposition . Tell.byKeyHashOneOf + +-- | Distribute the message stream to all provided actors. +-- +-- You can consider this being an interface to the Product monoid. +allOf :: [Actor message] -> Actor message +allOf = tellComposition Tell.all + +-- ** Helpers + +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), + ids = foldMap (.ids) actors + } + +fromRunner :: Runner a -> Actor a +fromRunner runner = + Actor + { tell = Runner.tell runner, + kill = Runner.kill runner, + wait = Runner.wait runner, + ids = [Runner.getId runner] + } + +-- * Acquisition + +-- | Spawn an actor which processes messages in isolated executions. +spawnStatelessIndividual :: + -- | Clean up when killed or exception is thrown. + IO () -> + -- | Interpret a message. + (message -> IO ()) -> + -- | Fork a thread to run the handler loop on and produce a handle to control it. + IO (Actor message) +spawnStatelessIndividual cleaner interpreter = + -- TODO: Optimize by reimplementing directly. + spawnStatefulIndividual () (const cleaner) (const interpreter) + +-- | Spawn an actor which processes all available messages in one execution. +spawnStatelessBatched :: + -- | Clean up when killed or exception is thrown. + IO () -> + -- | Interpret a batch of messages. + (NonEmpty message -> IO ()) -> + -- | Fork a thread to run the handler loop on and produce a handle to control it. + IO (Actor message) +spawnStatelessBatched cleaner interpreter = + -- TODO: Optimize by reimplementing directly. + spawnStatefulBatched () (const cleaner) (const interpreter) + +-- | Spawn an actor which processes messages in isolated executions +-- and threads state. +spawnStatefulIndividual :: + -- | Initial state. + state -> + -- | Clean up when killed or exception is thrown. + (state -> IO ()) -> + -- | Process a message and update state. + (state -> message -> IO state) -> + -- | Fork a thread to run the handler loop on and produce a handle to control it. + IO (Actor message) +spawnStatefulIndividual zero finalizer step = + spawnStatefulBatched zero finalizer $ foldM step + +-- | Spawn an actor which processes all available messages in one execution +-- and threads state. +spawnStatefulBatched :: + -- | Initial state. + state -> + -- | Clean up when killed or exception is thrown. + (state -> IO ()) -> + -- | Process a batch of messages and update state. + (state -> NonEmpty message -> IO state) -> + -- | Fork a thread to run the handler loop on and produce a handle to control it. + IO (Actor message) +spawnStatefulBatched zero finalizer step = + do + runner <- Runner.start + forkIOWithUnmask $ \unmask -> + let loop !state = + do + messages <- atomically $ Runner.receiveMultiple runner + case messages of + Just nonEmptyMessages -> + do + result <- try $ unmask $ step state nonEmptyMessages + case result of + Right newState -> + loop newState + Left exception -> + finally (finalizer state) + $ atomically + $ Runner.releaseWithException runner exception + -- Empty batch means that the runner is finished. + Nothing -> + finally (finalizer state) + $ atomically + $ Runner.releaseNormally runner + in loop zero + return $ fromRunner runner + +-- * Control + +-- | Add a message to the end of the queue of the +-- messages to be processed by the provided actor. +tell :: Actor message -> message -> IO () +tell actor = + atomically . actor.tell + +-- | Command the actor to stop registering new messages, +-- process all the registered ones and execute the clean up action. +-- +-- This action executes immediately. +-- If you want to block waiting for the actor to actually die, +-- after 'kill' you can run 'wait'. +kill :: Actor message -> IO () +kill actor = + atomically actor.kill + +-- | Block waiting for the actor to die either due to getting killed +-- or due to its interpreter action throwing an exception. +-- The exception will get rethrown here. +wait :: Actor message -> IO () +wait actor = + atomically actor.wait >>= maybe (pure ()) throwIO diff --git a/library/TheatreDev/Daemon.hs b/library/TheatreDev/Daemon.hs new file mode 100644 --- /dev/null +++ b/library/TheatreDev/Daemon.hs @@ -0,0 +1,86 @@ +module TheatreDev.Daemon + ( Daemon, + + -- * Acquisition + spawn, + + -- * Control + kill, + wait, + ) +where + +import TheatreDev.Prelude +import TheatreDev.Wait qualified as Wait + +data Config = forall state. + Config + { initialState :: state, + iterate :: state -> IO state, + cleanUp :: state -> IO () + } + +-- | +-- Think of an actor that does not process any messages and simply +-- interrupts between each iteration to check whether it's still alive. +data Daemon = Daemon + { -- | Kill the daemon. + kill :: STM (), + -- | Wait for the daemon to die due to error or being killed. + wait :: STM (Maybe SomeException) + } + +instance Semigroup Daemon where + left <> right = + Daemon + { kill = left.kill *> right.kill, + wait = Wait.both left.wait right.wait + } + +instance Monoid Daemon where + mempty = + Daemon + { kill = return (), + wait = return Nothing + } + mconcat daemons = + Daemon + { kill = traverse_ (.kill) daemons, + wait = Wait.all (fmap (.wait) daemons) + } + +spawn :: Config -> IO Daemon +spawn Config {..} = do + iteratingVar <- newTVarIO True + resultVar <- newEmptyTMVarIO + forkIOWithUnmask $ \unmask -> + let go !state = do + iterating <- readTVarIO iteratingVar + if iterating + then do + iterationAttemptResult <- try (unmask (iterate state)) + case iterationAttemptResult of + Right newState -> go newState + Left exception -> do + try @SomeException (unmask (cleanUp state)) + atomically (putTMVar resultVar (Just exception)) + else do + cleanUpResult <- try @SomeException (unmask (cleanUp state)) + case cleanUpResult of + Right () -> atomically (putTMVar resultVar Nothing) + Left exception -> atomically (putTMVar resultVar (Just exception)) + in go initialState + return + Daemon + { kill = writeTVar iteratingVar False, + wait = readTMVar resultVar + } + where + +kill :: Daemon -> IO () +kill daemon = + atomically daemon.kill + +wait :: Daemon -> IO () +wait daemon = + atomically daemon.wait >>= maybe (pure ()) throwIO diff --git a/library/TheatreDev/Prelude.hs b/library/TheatreDev/Prelude.hs --- a/library/TheatreDev/Prelude.hs +++ b/library/TheatreDev/Prelude.hs @@ -1,3 +1,5 @@ +{-# OPTIONS_GHC -Wno-dodgy-imports #-} + module TheatreDev.Prelude ( module Exports, ) diff --git a/theatre-dev.cabal b/theatre-dev.cabal --- a/theatre-dev.cabal +++ b/theatre-dev.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: theatre-dev -version: 0.3 +version: 0.4 category: Concurrency, Actors synopsis: Minimalistic actor library experiments description: @@ -82,7 +82,10 @@ library import: base hs-source-dirs: library - exposed-modules: TheatreDev + exposed-modules: + TheatreDev.Actor + TheatreDev.Daemon + other-modules: TheatreDev.ExtrasFor.List TheatreDev.ExtrasFor.TBQueue @@ -104,9 +107,9 @@ hs-source-dirs: hspec main-is: Main.hs other-modules: - TheatreDevSpec - TheatreDevSpec.IO - TheatreDevSpec.Preferences + TheatreDev.ActorSpec + TheatreDev.ActorSpec.IO + TheatreDev.ActorSpec.Preferences build-depends: , async