packages feed

theatre-dev 0.3 → 0.4

raw patch · 12 files changed

+543/−450 lines, 12 files

Files

hspec/Main.hs view
@@ -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
+ hspec/TheatreDev/ActorSpec.hs view
@@ -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+            ]
+ hspec/TheatreDev/ActorSpec/IO.hs view
@@ -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
+ hspec/TheatreDev/ActorSpec/Preferences.hs view
@@ -0,0 +1,9 @@+module TheatreDev.ActorSpec.Preferences where++import Prelude++concurrency :: Int+concurrency = 7++largePropertyMaxSuccess :: Int+largePropertyMaxSuccess = 10000
− hspec/TheatreDevSpec.hs 
@@ -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-            ]
− hspec/TheatreDevSpec/IO.hs 
@@ -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
− hspec/TheatreDevSpec/Preferences.hs 
@@ -1,9 +0,0 @@-module TheatreDevSpec.Preferences where--import Prelude--concurrency :: Int-concurrency = 7--largePropertyMaxSuccess :: Int-largePropertyMaxSuccess = 10000
− library/TheatreDev.hs 
@@ -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
+ library/TheatreDev/Actor.hs view
@@ -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
+ library/TheatreDev/Daemon.hs view
@@ -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
library/TheatreDev/Prelude.hs view
@@ -1,3 +1,5 @@+{-# OPTIONS_GHC -Wno-dodgy-imports #-}+ module TheatreDev.Prelude   ( module Exports,   )
theatre-dev.cabal view
@@ -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