theatre-dev 0.1 → 0.1.1

raw patch · 5 files changed

+44/−27 lines, 5 files

Files

• hspec/TheatreDev/StmBasedSpec.hs +11/−4
• hspec/TheatreDev/StmBasedSpec/IO.hs +2/−2
• library/TheatreDev/StmBased.hs +30/−19
• library/TheatreDev/StmBased/StmStructures/Runner.hs +0/−1
• theatre-dev.cabal +1/−1

@@ -8,7 +8,6 @@ import Test.Hspec import Test.Hspec.QuickCheck import Test.QuickCheck-import TheatreDev.StmBased (Actor) import TheatreDev.StmBased qualified as Actor import TheatreDev.StmBasedSpec.IO qualified as IO import TheatreDev.StmBasedSpec.Preferences qualified as Preferences@@ -17,6 +16,14 @@ spec :: Spec spec =   do+    describe "kill" do+      describe "When full" do+        it "Does not" pending++    describe "kill" do+      describe "When full" do+        it "Blocks until a slot is freed up" pending+     describe "spawnStatefulBatched" do       let spawnIntUpdater step = Actor.spawnStatefulBatched @Int 0 (const (return ())) step       let spawnUnit step = Actor.spawnStatefulBatched () (const (return ())) step@@ -95,12 +102,12 @@             messagesNum = 10             actorsNum = 3             messages = [0 .. messagesNum - 1]-        results <- fmap (fmap sort) (IO.simulateReduction Actor.allOf actorsNum emittersNum messages)+        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 Actor.allOf Preferences.concurrency size messages+          results <- sort . concat <$> IO.simulateReduction Preferences.concurrency size Actor.allOf messages           return             $ conjoin               [ length results === length messages * size * Preferences.concurrency,@@ -110,7 +117,7 @@     describe "oneOf" . modifyMaxSuccess (max 1000) $ do       prop "Dispatches correctly" $ forAll (chooseInt (0, 99)) $ \size -> forAll arbitrary $ \(messages :: [Int]) ->         idempotentIOProperty do-          results <- sort . concat <$> IO.simulateReduction Actor.oneOf Preferences.concurrency size messages+          results <- sort . concat <$> IO.simulateReduction Preferences.concurrency size Actor.oneOf messages           return             $ conjoin               [ length results === length messages * size,

@@ -5,8 +5,8 @@ import TheatreDev.StmBased qualified as Actor import Prelude -simulateReduction :: (Show a) => ([Actor a] -> Actor a) -> Int -> Int -> [a] -> IO [[a]]-simulateReduction reducer actorsNum generatorsNum messages =+simulateReduction :: Int -> Int -> ([Actor a] -> Actor a) -> [a] -> IO [[a]]+simulateReduction actorsNum generatorsNum reducer messages =   do     resultsVar <- newTVarIO [] 

@@ -22,7 +22,6 @@   ) where -import Data.UUID.V4 qualified as UuidV4 import TheatreDev.Prelude import TheatreDev.StmBased.StmStructures.Runner (Runner) import TheatreDev.StmBased.StmStructures.Runner qualified as Runner@@ -32,6 +31,7 @@  -- | -- 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. --@@ -100,7 +100,7 @@ oneOf :: [Actor message] -> Actor message oneOf = tellComposition Tell.one --- |+-- | Distribute the message stream to all provided actors. -- -- You can consider this being an interface to the Product monoid. allOf :: [Actor message] -> Actor message@@ -136,53 +136,54 @@  -- * 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.+-- | Spawn an actor which processes messages in isolated executions. spawnStatelessIndividual ::-  (Show message) =>   -- | Clean up when killed.   IO () ->-  -- | Interpreter of a message.+  -- | Interpret a message.   (message -> IO ()) ->-  -- | Fork a thread to run the handler daemon on and-  -- produce a handle to control it.+  -- | 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 ::-  (Show message) =>   -- | Clean up when killed.   IO () ->-  -- | Interpreter of a batch of messages.+  -- | Interpret a batch of messages.   (NonEmpty message -> IO ()) ->-  -- | Fork a thread to run the handler daemon on and-  -- produce a handle to control it.+  -- | 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 ::-  (Show message) =>+  -- | 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 ::-  (Show message) =>+  -- | 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@@ -212,14 +213,24 @@  -- * 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. wait :: Actor message -> IO () wait actor =   atomically actor.wait >>= maybe (pure ()) throwIO

@@ -74,7 +74,6 @@       else return Nothing  receiveMultiple ::-  (Show a) =>   Runner a ->   STM (Maybe (NonEmpty a)) receiveMultiple Runner {..} =

@@ -1,6 +1,6 @@ cabal-version: 3.0 name:          theatre-dev-version:       0.1+version:       0.1.1 category:      Concurrency, Actors synopsis:      Minimalistic actor library experiments description: