packages feed

theatre-dev 0.0.1 → 0.1

raw patch · 13 files changed

+278/−131 lines, 13 filesdep +QuickCheckdep +uuiddep ~asyncdep ~stm

Dependencies added: QuickCheck, uuid

Dependency ranges changed: async, stm

Files

hspec/Main.hs view
@@ -1,7 +1,7 @@ module Main (main) where  import Test.Hspec-import qualified TheatreDev.StmBasedSpec+import TheatreDev.StmBasedSpec qualified import Prelude  main :: IO ()
hspec/TheatreDev/StmBasedSpec.hs view
@@ -1,17 +1,25 @@-{-# OPTIONS_GHC -Wno-unused-local-binds #-}+{-# OPTIONS_GHC -Wno-unused-local-binds -Wno-unused-binds #-}  module TheatreDev.StmBasedSpec (spec) where +import Control.Concurrent.Async+import Data.IntMap.Strict qualified as IntMap+import Data.IntSet qualified as IntSet import Test.Hspec-import qualified TheatreDev.StmBased as Actor+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 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 ()))+    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 []@@ -35,12 +43,115 @@         putMVar actorLock ()          takeMVar emitterLock+        Actor.kill actor          collectedBatches <- reverse . fmap toList <$> readIORef acc         shouldBe collectedBatches [[1], [2, 3], [4]]        it "Threads the state" do-        pending+        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-        pending+        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 Actor.allOf actorsNum emittersNum 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+          return+            $ conjoin+              [ length results === length messages * size * Preferences.concurrency,+                results === sort (concat (replicate (size * Preferences.concurrency) messages))+              ]++    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+          return+            $ conjoin+              [ length results === length messages * size,+                results+                  === sort (concat (replicate (size) messages))+              ]++    describe "byKeyHash" . modifyMaxSuccess (max Preferences.largePropertyMaxSuccess) $ do+      prop "Dispatches individually" $ forAll (chooseInt (0, 99)) $ \size -> forAll arbitrary $ \(messages :: [Int]) -> idempotentIOProperty $ do+        resultsVar <- newTVarIO []+        actor <-+          fmap (Actor.byKeyHash 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/StmBasedSpec/IO.hs view
@@ -0,0 +1,34 @@+module TheatreDev.StmBasedSpec.IO where++import Control.Concurrent.Async+import TheatreDev.StmBased (Actor)+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 =+  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/StmBasedSpec/Preferences.hs view
@@ -0,0 +1,9 @@+module TheatreDev.StmBasedSpec.Preferences where++import Prelude++concurrency :: Int+concurrency = 7++largePropertyMaxSuccess :: Int+largePropertyMaxSuccess = 10000
library/TheatreDev/ExtrasFor/TBQueue.hs view
@@ -6,13 +6,13 @@ flushNonEmptyTBQueue :: TBQueue a -> STM (NonEmpty a) flushNonEmptyTBQueue x = do   head <- readTBQueue x-  tail <- flushTBQueue x+  tail <- simplerFlushTBQueue 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+  list <- simplerFlushTBQueue queue   forM_ list $ writeTBQueue queue   return list
library/TheatreDev/Perpetual.hs view
@@ -12,7 +12,7 @@   ) where -import qualified Control.Concurrent.Chan.Unagi as Unagi+import Control.Concurrent.Chan.Unagi qualified as Unagi import TheatreDev.Prelude  -- |
library/TheatreDev/Prelude.hs view
@@ -8,6 +8,7 @@ import Control.Category as Exports import Control.Concurrent as Exports import Control.Concurrent.Async as Exports (Concurrently (..))+import Control.Concurrent.STM as Exports 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@@ -44,6 +45,7 @@ import Data.String as Exports import Data.Traversable as Exports import Data.Tuple as Exports+import Data.UUID as Exports (UUID) import Data.Unique as Exports import Data.Version as Exports import Data.Void as Exports
library/TheatreDev/StmBased.hs view
@@ -22,12 +22,13 @@   ) where +import Data.UUID.V4 qualified as UuidV4 import TheatreDev.Prelude import TheatreDev.StmBased.StmStructures.Runner (Runner)-import qualified TheatreDev.StmBased.StmStructures.Runner as Runner+import TheatreDev.StmBased.StmStructures.Runner qualified as Runner import TheatreDev.StmBased.Tell (Tell)-import qualified TheatreDev.StmBased.Tell as Tell-import qualified TheatreDev.StmBased.Wait as Wait+import TheatreDev.StmBased.Tell qualified as Tell+import TheatreDev.StmBased.Wait qualified as Wait  -- | -- Controls of an actor, which processes the messages of type @message@.@@ -42,32 +43,37 @@     -- | Kill the actor.     kill :: STM (),     -- | Wait for the actor to die due to error or being killed.-    wait :: STM (Maybe SomeException)+    wait :: STM (Maybe SomeException),+    -- | IDs of the constituent actors.+    -- Useful for debugging.+    ids :: [UUID]   }  instance Contravariant Actor where-  contramap fn (Actor tell kill wait) =-    Actor (tell . fn) kill wait+  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) (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+    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) (Actor rTell rKill rWait) =-    Actor tell kill wait-    where-      tell = either lTell rTell . choice-      kill = lKill >> rKill-      wait = Wait.both lWait rWait+    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 @@ -76,7 +82,8 @@   Actor     { tell = Runner.tell runner,       kill = Runner.kill runner,-      wait = Runner.wait runner+      wait = Runner.wait runner,+      ids = [Runner.getId runner]     }  -- | Distribute the message stream across actors.@@ -109,7 +116,6 @@ -- 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.@@ -124,7 +130,8 @@   Actor     { tell = tellReducer (fmap (.tell) actors),       kill = traverse_ (.kill) actors,-      wait = Wait.all (fmap (.wait) actors)+      wait = Wait.all (fmap (.wait) actors),+      ids = foldMap (.ids) actors     }  -- * Acquisition@@ -137,65 +144,49 @@ -- 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 ()) ->+  (Show message) =>   -- | Clean up when killed.   IO () ->+  -- | Interpreter of a message.+  (message -> IO ()) ->   -- | Fork a thread to run the handler daemon on and   -- produce a handle to control it.   IO (Actor message)-spawnStatelessIndividual interpreter cleaner =+spawnStatelessIndividual cleaner interpreter =   -- TODO: Optimize by reimplementing directly.-  spawnStatefulIndividual () (const interpreter) (const cleaner)+  spawnStatefulIndividual () (const cleaner) (const interpreter)  spawnStatelessBatched ::-  -- | Interpreter of a batch of messages.-  (NonEmpty message -> IO ()) ->+  (Show message) =>   -- | Clean up when killed.   IO () ->+  -- | Interpreter of a batch of messages.+  (NonEmpty message -> IO ()) ->   -- | Fork a thread to run the handler daemon on and   -- produce a handle to control it.   IO (Actor message)-spawnStatelessBatched interpreter cleaner =+spawnStatelessBatched cleaner interpreter =   -- TODO: Optimize by reimplementing directly.-  spawnStatefulBatched () (const interpreter) (const cleaner)+  spawnStatefulBatched () (const cleaner) (const interpreter)  spawnStatefulIndividual ::+  (Show message) =>   state ->-  (state -> message -> IO state) ->   (state -> IO ()) ->+  (state -> message -> IO state) ->   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+spawnStatefulIndividual zero finalizer step =+  spawnStatefulBatched zero finalizer $ foldM step  spawnStatefulBatched ::+  (Show message) =>   state ->-  (state -> NonEmpty message -> IO state) ->   (state -> IO ()) ->+  (state -> NonEmpty message -> IO state) ->   IO (Actor message)-spawnStatefulBatched zero step finalizer =+spawnStatefulBatched zero finalizer step =   do-    runner <- atomically Runner.start+    runner <- Runner.start     forkIOWithUnmask $ \unmask ->       let loop !state =             do@@ -203,16 +194,19 @@               case messages of                 Just nonEmptyMessages ->                   do-                    result <- try @SomeException $ unmask $ step state nonEmptyMessages+                    result <- try $ unmask $ step state nonEmptyMessages                     case result of                       Right newState ->                         loop newState                       Left exception ->-                        do-                          atomically $ Runner.releaseWithException runner exception-                          finalizer state+                        finally (finalizer state)+                          $ atomically+                          $ Runner.releaseWithException runner exception                 -- Empty batch means that the runner is finished.-                Nothing -> finalizer state+                Nothing ->+                  finally (finalizer state)+                    $ atomically+                    $ Runner.releaseNormally runner        in loop zero     return $ fromRunner runner
library/TheatreDev/StmBased/StmStructures/Runner.hs view
@@ -10,27 +10,37 @@     receiveSingle,     receiveMultiple,     releaseWithException,+    releaseNormally,++    -- * Inspection+    getId,   ) where  import Control.Concurrent.STM.TBQueue import Control.Concurrent.STM.TMVar-import qualified TheatreDev.ExtrasFor.List as List+import Data.UUID.V4 qualified as UuidV4+import TheatreDev.ExtrasFor.List qualified as List import TheatreDev.ExtrasFor.TBQueue import TheatreDev.Prelude  data Runner a = Runner-  { queue :: TBQueue (Maybe a),+  { queue :: TBQueue a,     aliveVar :: TVar Bool,-    resVar :: TMVar (Maybe SomeException)+    resVar :: TMVar (Maybe SomeException),+    id :: UUID   } -start :: STM (Runner a)+getId :: Runner a -> UUID+getId = (.id)++start :: IO (Runner a) start =   do-    queue <- newTBQueue 1000-    aliveVar <- newTVar True-    resVar <- newEmptyTMVar @(Maybe SomeException)+    queue <- newTBQueueIO 1000+    aliveVar <- newTVarIO True+    resVar <- newEmptyTMVarIO+    id <- UuidV4.nextRandom     return Runner {..}  tell :: Runner a -> a -> STM ()@@ -38,17 +48,18 @@   do     alive <- readTVar aliveVar     when alive do-      writeTBQueue queue $ Just message+      writeTBQueue queue message  kill :: Runner a -> STM () kill Runner {..} =-  do-    alive <- readTVar aliveVar-    when alive do-      writeTBQueue queue Nothing+  writeTVar aliveVar False  wait :: Runner a -> STM (Maybe SomeException)-wait Runner {..} =+wait Runner {..} = do+  isAlive <- readTVar aliveVar+  when isAlive retry+  queueIsEmpty <- isEmptyTBQueue queue+  unless queueIsEmpty retry   readTMVar resVar  receiveSingle ::@@ -57,41 +68,33 @@   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+    alive <- readTVar aliveVar+    if alive+      then Just <$> readTBQueue queue+      else return Nothing  receiveMultiple ::+  (Show a) =>   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+    messages <- simplerFlushTBQueue queue     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+        alive <- readTVar aliveVar+        if alive+          then retry+          else return Nothing+      messagesHead : messagesTail ->         return $ Just $ messagesHead :| messagesTail  releaseWithException :: Runner a -> SomeException -> STM () releaseWithException Runner {..} exception =   do     simplerFlushTBQueue queue-    writeTVar aliveVar False     putTMVar resVar (Just exception)++releaseNormally :: Runner a -> STM ()+releaseNormally Runner {..} =+  putTMVar resVar Nothing <|> pure ()
library/TheatreDev/StmBased/Tell.hs view
@@ -1,6 +1,6 @@ module TheatreDev.StmBased.Tell where -import qualified Data.Vector as Vector+import Data.Vector qualified as Vector import TheatreDev.Prelude  type Tell a = a -> STM ()
library/TheatreDev/StmBased/Wait.hs view
@@ -6,22 +6,8 @@  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)+  liftA2 (<|>) left right  all :: [Wait] -> Wait all waits =-  getException <|> getNothing-  where-    getException =-      waits-        & fmap (>>= maybe empty pure)-        & asum-        & fmap Just-    getNothing =-      Nothing <$ sequence_ waits+  asum <$> sequence waits
library/TheatreDev/Terminal/Actor.hs view
@@ -16,10 +16,10 @@   ) where -import qualified Control.Concurrent.Chan.Unagi as E+import Control.Concurrent.Chan.Unagi qualified as E import Control.Concurrent.STM.TBQueue import Control.Concurrent.STM.TMVar-import qualified TheatreDev.ExtrasFor.List as List+import TheatreDev.ExtrasFor.List qualified as List import TheatreDev.ExtrasFor.TBQueue import TheatreDev.Prelude
theatre-dev.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name:          theatre-dev-version:       0.0.1+version:       0.1 category:      Concurrency, Actors synopsis:      Minimalistic actor library experiments description:@@ -46,6 +46,7 @@     FunctionalDependencies     GADTs     GeneralizedNewtypeDeriving+    ImportQualifiedPost     LambdaCase     LiberalTypeSynonyms     MagicHash@@ -98,8 +99,9 @@     , async >=2.2 && <3     , base >=4.16 && <5     , contravariant >=1.3 && <2-    , stm >=2.5.2 && <3+    , stm >=2.5 && <2.5.2 || >=2.5.2.2 && <3     , unagi-chan >=0.4.1.4 && <0.5+    , uuid >=1.3.15 && <2     , vector >=0.13 && <0.14  test-suite hspec@@ -107,8 +109,14 @@   type:           exitcode-stdio-1.0   hs-source-dirs: hspec   main-is:        Main.hs-  other-modules:  TheatreDev.StmBasedSpec+  other-modules:+    TheatreDev.StmBasedSpec+    TheatreDev.StmBasedSpec.IO+    TheatreDev.StmBasedSpec.Preferences+   build-depends:+    , async     , hspec >=2.11.6 && <3+    , QuickCheck >=2.14 && <3     , rerebase <2     , theatre-dev