diff --git a/.github/workflows/haskell.yml b/.github/workflows/haskell.yml index 2a931ee..2b41063 100644 --- a/.github/workflows/haskell.yml +++ b/.github/workflows/haskell.yml @@ -44,9 +44,9 @@ jobs: - name: Install dependencies run: | cabal update - cabal build --only-dependencies --enable-tests + cabal build --only-dependencies --enable-tests --enable-benchmarks - name: Build - run: cabal build --enable-tests all --ghc-options=-Werror + run: cabal build --enable-tests --enable-benchmarks all --ghc-options=-Werror - name: Run tests run: cabal test all --ghc-options=-Werror --test-options='+RTS -N2 -RTS' - name: Package checks diff --git a/CHANGELOG.md b/CHANGELOG.md index d030445..81fc88e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,26 @@ # Revision history for stm-actor -## 0.4.0.0 -- UNRELEASED +## 0.4.0.0 -- 2026-08-23 + +* Add `actWith` and `ActorConfig`, which configure the mailbox capacity, the + completion handler, an `onUndelivered` handler that receives the messages + still queued when the actor stopped, and an `onEffectFailure` handler that + is called for each completion effect that throws. `act`, `actBounded`, + `actFinally`, and `actFinallyBounded` are specialisations. +* Make `addAfterEffect` lifecycle-checked by default, matching `send`; the + previous unchecked behaviour is available as `addAfterEffectUnchecked`, and + `addAfterEffectChecked` remains as an alias. +* Add `monitor` and `monitorSTM`, which deliver an actor's completion to the + monitoring actor's mailbox as an ordinary message instead of an asynchronous + exception. Monitoring an actor that has already stopped delivers the message + immediately. +* Add `awaitEffects`, which waits until the completion handler and + after-effects have finished running. +* Make `murder` a no-op once the actor has stopped, so completion effects are + not interrupted. +* Document that an actor blocked in `receive` whose handles have all been + dropped is stopped by the runtime with `BlockedIndefinitelyOnSTM`. +* Add a fan-in throughput benchmark comparing unbounded and bounded mailboxes. * Make linking to an already-stopped actor fail reliably rather than silently installing an after-effect that can never run. @@ -24,7 +44,10 @@ * Add opt-in bounded actor mailboxes with transactional backpressure through `actBounded` and `actFinallyBounded`. * Use `stm-queue`'s incremental real-time queue for unbounded and bounded - mailboxes while keeping compatibility with the published `stm-queue-0.2.0`. + mailboxes. Bounded mailboxes are `stm-queue-0.2.2`'s bounded queues, whose + split read and write credits let senders and the actor conflict on + capacity accounting once per `capacity` sends rather than on every message. +* Require `stm-queue >= 0.2.2.0`. * Add deterministic lifecycle regression tests and bounded test waits. * Run the concurrency suite with multiple runtime capabilities. * Validate the oldest compatible dependency plan in CI. diff --git a/README.md b/README.md index 32c1e5a..e9ac8f1 100644 --- a/README.md +++ b/README.md @@ -25,9 +25,12 @@ with other threads and actors. ## Mailboxes and sending `act` and `actFinally` create an unbounded FIFO mailbox. `actBounded capacity` -and `actFinallyBounded capacity` instead create a bounded FIFO mailbox. Both -use `stm-queue`'s incremental-rotation real-time queue; bounded mailboxes add -transactional occupancy accounting. Sending and lifecycle operations are STM +and `actFinallyBounded capacity` instead create a bounded FIFO mailbox. All +four are specialisations of `actWith`, which takes an `ActorConfig` describing +the mailbox capacity and the handlers run when the actor stops. Both kinds of +mailbox use `stm-queue`'s incremental-rotation real-time queue; bounded mailboxes are +its bounded queues, whose split read and write credits keep senders and the +actor from contending on capacity accounting for every message. Sending and lifecycle operations are STM transactions, so they can be combined atomically with application state. A committed `send` means that the actor was alive at that transaction's linearization point; it does not promise that the actor will eventually process @@ -62,7 +65,19 @@ Choose the sending operation based on how the caller handles lifecycle races: | `trySend actor message` | Returns `ActorStopped` | Returns `MailboxFull` | `trySend` returns `Sent` after enqueueing and never retries because of capacity. -Actor shutdown drains messages which were already queued. +Actor shutdown drains messages which were already queued and hands them, in +mailbox order, to the `onUndelivered` handler of the actor's `ActorConfig`: + +```haskell +worker <- actWith defaultActorConfig + { mailboxCapacity = Just 256 + , onUndelivered = mapM_ requeueElsewhere + } + workerLoop +``` + +Every committed `send` therefore either reaches a `receive` handler or reaches +`onUndelivered`; the default handler discards the messages. `receive` removes one message and then runs its handler. `receiveSTM` combines mailbox removal and a caller-supplied STM action in one transaction, so either @@ -75,29 +90,32 @@ An actor transitions exactly once from `Alive` to either `Completed` or `await` retries in STM until the action has stopped, so it composes with other transactions without polling. -The transition also closes checked after-effect registration. A registration -racing completion is therefore either committed and later run, or rejected; -it is never silently lost. As with sending, three registration modes are -available: +The transition also closes after-effect registration. A registration racing +completion is therefore either committed and later run, or rejected; it is +never silently lost. As with sending, three registration modes are available: | Operation | Result if the actor has already stopped | | --- | --- | -| `addAfterEffect` | Registers unchecked; the effect cannot run | -| `addAfterEffectChecked` | Throws `ActorDead` in STM | +| `addAfterEffect` | Throws `ActorDead` in STM | | `tryAddAfterEffect` | Returns `False` without registering | - -After the lifecycle transition, queued messages are drained and link -notifications are initiated. The completion handler passed to `actFinally` -then runs, followed by registered after-effects in registration order. Every -effect is attempted even if an earlier one throws; after draining the list, the -first effect exception is re-thrown in the actor's terminating thread. User -effects run sequentially, so a blocking effect delays later user effects. - -`await` returns after the actor's action result has been recorded and checked +| `addAfterEffectUnchecked` | Registers without checking; the effect cannot run | + +After the lifecycle transition, queued messages are drained and link and +monitor notifications are initiated. The completion handler then runs, followed +by `onUndelivered` if any messages were queued, and then registered +after-effects in registration order. Every effect is attempted even if an +earlier one throws: each failure is passed to the `onEffectFailure` handler, +which by default rethrows, so after draining the list the first exception is +re-thrown in the actor's terminating thread. Supply a logging handler to keep +effect failures out of the default uncaught-exception output. User effects run +sequentially, so a blocking effect delays later user effects. + +`await` returns after the actor's action result has been recorded and registration has closed. Completion handlers and after-effects may still be -running, and their failures do not change the recorded `Liveness` result. +running, and their failures do not change the recorded `Liveness` result; +`awaitEffects` additionally waits until every effect has finished. -## Links and cancellation +## Links, monitors, and cancellation `link target`, called inside an actor, establishes a one-way link: when `target` stops normally or exceptionally, the calling actor receives `LinkKill`. @@ -113,9 +131,37 @@ asynchronous exceptions from blocking the target's completion effects. The helper itself can remain blocked while the recipient uses `uninterruptibleMask`. +Links interrupt the recipient, which suits cancellation. To be told that an +actor stopped without being interrupted, use a monitor. `monitor target +toMessage`, called inside an actor, sends `toMessage completion` to the calling +actor's own mailbox when `target` stops, where `completion` is `Nothing` for +normal completion or `Just` the exception. The notification is handled like any +other message, in mailbox order: + +```haskell +data Message = Work Job | WorkerDown ThreadId (Maybe SomeException) + +supervisor worker = do + monitor worker (WorkerDown (threadId worker)) + receive $ \case + Work job -> ... + WorkerDown who reason -> ... +``` + +Monitoring an actor that has already stopped delivers the message immediately. +`monitorSTM recipient target toMessage` is the same operation in STM. + `murder` requests cancellation by synchronously using `throwTo` with a `MurderKill` exception. Like any synchronous `throwTo`, it can block while the -target is uninterruptibly masking asynchronous exceptions. +target is uninterruptibly masking asynchronous exceptions. Once the actor has +stopped, `murder` does nothing, so completion effects are not interrupted. + +An actor blocked in `receive` whose mailbox is no longer reachable from any +other thread can never receive another message. The runtime detects this at +the next major garbage collection and throws `BlockedIndefinitelyOnSTM` to the +actor, which stops with `ThrewException` and runs its links, monitors, and +completion effects like any other failure. Dropping every handle to an actor +therefore reclaims it, but the failure cascades through links. ## Scope and compatibility diff --git a/bench/Bench.hs b/bench/Bench.hs new file mode 100644 index 0000000..4580131 --- /dev/null +++ b/bench/Bench.hs @@ -0,0 +1,57 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE BlockArguments #-} +module Main where + +import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent.Actor +import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) +import Control.Concurrent.STM +import Control.Monad (forM_, forever, when) +import Control.Monad.IO.Class (liftIO) +import Data.IORef +import Numeric.Natural (Natural) +import System.Environment (getArgs) + +-- | Fan-in throughput: @senders@ threads send to one actor for one second. +-- Compares an unbounded mailbox with a bounded one, which exercises the +-- capacity accounting shared between senders and the actor. +-- +-- With more senders than capacity permits, the bounded figure is dominated by +-- STM wakeups: every receive wakes every sender blocked on the full mailbox. +-- That cost grows with the number of capabilities, so compare runs at a fixed +-- @+RTS -N@ setting. +fanIn :: Int -> String -> Maybe Natural -> IO () +fanIn senders label capacity = do + received <- newIORef (0 :: Int) + actor <- actWith defaultActorConfig { mailboxCapacity = capacity } $ + forever $ receive \() -> liftIO (modifyIORef' received (+ 1)) + stop <- newTVarIO False + finished <- newEmptyMVar + forM_ [1 .. senders] \_ -> forkIO do + let loop !sent = do + continue <- atomically do + stopped <- readTVar stop + if stopped then pure False else send actor () >> pure True + if continue then loop (sent + 1) else putMVar finished (sent :: Int) + loop 0 + threadDelay 1000000 + atomically (writeTVar stop True) + sent <- sum <$> mapM (const (takeMVar finished)) [1 .. senders] + murder actor + _ <- atomically (awaitEffects actor) + count <- readIORef received + putStrLn + ( label <> ": " <> show senders <> " senders, " + <> show count <> " received, " <> show sent <> " sent" + ) + +main :: IO () +main = do + args <- getArgs + let senderCounts = case args of + [] -> [1, 4, 16] + _ -> map read args + forM_ senderCounts \senders -> do + fanIn senders "unbounded mailbox" Nothing + fanIn senders "bounded mailbox (1024)" (Just 1024) + when (senders /= last senderCounts) (putStrLn "") diff --git a/cabal.project b/cabal.project index e6fdbad..45d93da 100644 --- a/cabal.project +++ b/cabal.project @@ -1 +1,8 @@ packages: . + +-- stm-queue-0.2.2.0 is not yet on Hackage. Remove this pin once it is +-- published; the sdist and oldest-dependency CI jobs will fail until then. +source-repository-package + type: git + location: https://github.com/SamuelSchlesinger/stm-queue.git + tag: b1c8f5653a03c93a3243cf7c93aa9c67e3d6a836 diff --git a/src/Control/Concurrent/Actor.hs b/src/Control/Concurrent/Actor.hs index 2dd3264..ece70a7 100644 --- a/src/Control/Concurrent/Actor.hs +++ b/src/Control/Concurrent/Actor.hs @@ -24,48 +24,73 @@ handler in the actor's thread. An actor has a single lifecycle transition from alive to stopped. The transition records whether its action completed normally or threw, atomically -closes normal sends and checked after-effect registration, and drains queued -messages. 'await' observes that transition without polling. Link notifications -are initiated next. The completion handler supplied to 'actFinally' and -registered user after-effects then run in the actor's terminating thread, with -every effect attempted in registration order. - -'send' is lifecycle-safe by default. The unchecked 'addAfterEffect' operation -avoids touching the lifecycle 'TVar'; use it only when the caller already -controls the actor lifecycle. +closes sends and after-effect registration, and drains queued messages. +'await' observes that transition without polling. Link and monitor +notifications are initiated next. The completion handler, the undelivered +message handler, and registered user after-effects then run in the actor's +terminating thread, with every effect attempted in registration order; +'awaitEffects' observes their completion. + +'send' and 'addAfterEffect' are lifecycle-safe by default. The unchecked +'addAfterEffectUnchecked' operation avoids touching the lifecycle 'TVar'; use +it only when the caller already controls the actor lifecycle. + +== Actors and garbage collection + +An actor blocked in 'receive' whose mailbox is reachable from no other thread +can never receive another message. The runtime system detects this at the next +major garbage collection and throws 'Control.Exception.BlockedIndefinitelyOnSTM' +to the actor, which then stops with 'ThrewException' and runs its links, +monitors, and completion effects like any other failure. Dropping every t'Actor' +handle therefore reclaims the actor, but the failure cascades through links. +Note that an after-effect or monitor which captures the handle keeps the actor +alive. -} module Control.Concurrent.Actor ( ActionT , Actor + -- * Creating actors +, act +, actBounded +, actFinally +, actFinallyBounded +, actWith +, ActorConfig(..) +, defaultActorConfig + -- * Sending , send , sendChecked , trySend , SendResult(..) -, addAfterEffect -, addAfterEffectChecked -, tryAddAfterEffect + -- * Receiving +, receive +, receiveSTM +, self +, hoistActionT + -- * Lifecycle , threadId , livenessCheck -, await -, withLivenessCheck , Liveness(..) +, await +, awaitEffects , ActorDead(..) -, actFinally -, actFinallyBounded -, act -, actBounded -, receiveSTM -, receive -, hoistActionT + -- * Completion effects +, addAfterEffect +, addAfterEffectChecked +, addAfterEffectUnchecked +, tryAddAfterEffect +, withLivenessCheck + -- * Links and monitors , link , linkSTM , LinkKill(..) -, self +, monitor +, monitorSTM + -- * Cancellation , murder , MurderKill(..) ) where --- This list was generated by compiling with the -ddump-minimal-imports flag. import Control.Applicative (Alternative) import Control.Concurrent (ThreadId, forkFinally, forkIO, myThreadId, throwTo) @@ -75,7 +100,6 @@ import Control.Concurrent.STM , atomically , check , modifyTVar' - , newTVar , newTVarIO , readTVar , retry @@ -83,7 +107,7 @@ import Control.Concurrent.STM , writeTVar ) import Control.Exception - (Exception, SomeException, catch, mask_, throwIO, try) + (Exception, SomeException, catch, finally, mask_, throwIO, try) import Control.Monad (void) import Control.Monad.Cont.Class (MonadCont) import Control.Monad.Error.Class (MonadError) @@ -95,7 +119,8 @@ import Control.Monad.RWS.Class (MonadRWS) import Control.Monad.State.Class (MonadState) import Control.Monad.Trans (MonadTrans(..)) import Control.Monad.Writer.Class (MonadWriter) -import Data.Queue (dequeue, enqueue, flush, newQueue) +import Data.Queue + (Queue, dequeue, enqueue, flush, newBoundedQueueIO, newQueueIO, tryEnqueue) import Data.Functor.Contravariant (Contravariant(contramap)) import Numeric.Natural (Natural) @@ -129,13 +154,6 @@ data ActorContext message = ActorContext , actorHandle :: Actor message } -data Mailbox message = Mailbox - { readMailbox :: STM message - , writeMailbox :: message -> STM () - , tryWriteMailbox :: message -> STM Bool - , clearMailbox :: STM () - } - -- | A handle used to send messages, inspect lifecycle state, register -- completion effects, and address the actor's thread. data Actor message = Actor @@ -145,6 +163,7 @@ data Actor message = Actor , send' :: message -> STM () , trySend' :: message -> STM Bool , actorState :: TVar ActorState + , effectsFinished :: TVar Bool } type Completion = Maybe SomeException @@ -171,7 +190,8 @@ livenessCheck actor = do -- atomically with other STM operations without polling or sleeping. -- -- The lifecycle transition happens before the completion handler and --- after-effects run, so this does not wait for those effects to finish. +-- after-effects run, so this does not wait for those effects to finish; see +-- 'awaitEffects'. -- -- @since 0.4.0.0 await :: Actor message -> STM Liveness @@ -179,19 +199,30 @@ await actor = livenessCheck actor >>= \case Alive -> retry stopped -> pure stopped --- | The exception thrown when an operation wrapped in 'withLivenessCheck' is --- attempted on an actor which has already stopped. 'Nothing' denotes normal --- completion; 'Just' contains the exception thrown by the actor's action. +-- | Wait until an actor has stopped and its completion handler, undelivered +-- message handler, and after-effects have all finished running, whether or +-- not any of them threw. Like 'await', this retries rather than polling. +-- +-- @since 0.4.0.0 +awaitEffects :: Actor message -> STM Liveness +awaitEffects actor = do + liveness <- await actor + readTVar (effectsFinished actor) >>= check + pure liveness + +-- | The exception thrown when a lifecycle-checked operation is attempted on an +-- actor which has already stopped. 'Nothing' denotes normal completion; 'Just' +-- contains the exception thrown by the actor's action. data ActorDead = ActorDead (Maybe SomeException) deriving Show instance Exception ActorDead --- | Wrap 'addAfterEffect' or another custom combinator in a liveness check. --- This adds contention on the lifecycle 'TVar', but prevents an operation from --- being accepted after the actor has stopped. If the t'Actor' is 'Completed' or --- 'ThrewException', this throws an t'ActorDead' exception with 'Nothing' or --- 'Just' the exception, respectively. +-- | Wrap 'addAfterEffectUnchecked' or another custom combinator in a liveness +-- check. This adds the lifecycle 'TVar' to the transaction's read set, but +-- prevents an operation from being accepted after the actor has stopped. If +-- the t'Actor' is 'Completed' or 'ThrewException', this throws an t'ActorDead' +-- exception with 'Nothing' or 'Just' the exception, respectively. withLivenessCheck :: (Actor message -> x -> STM ()) -> Actor message -> x -> STM () withLivenessCheck f actor x = ensureAlive actor >> f actor x @@ -200,30 +231,39 @@ ensureAlive actor = readTVar (actorState actor) >>= \case Running -> pure () Stopped completion -> throwSTM (ActorDead completion) --- | Once the t'Actor' stops, all of the effects that have been added via --- this function will run, in registration order. Later effects are still --- attempted if an earlier effect throws an exception. This is how you can --- implement your own functions like 'link' or 'linkSTM'. This operation does --- not itself check liveness; wrap it with 'withLivenessCheck' when registering --- against an actor that may already have stopped. Registering unchecked after --- completion stores an effect that can never run. +-- | Register an effect to run once the t'Actor' stops. All registered effects +-- run in registration order, after the completion handler; later effects are +-- still attempted if an earlier effect throws. This is how you can implement +-- your own functions like 'link', 'linkSTM', or 'monitorSTM'. +-- +-- If the actor has already stopped, throw t'ActorDead'. The liveness check +-- and registration are one STM transaction, so actor completion cannot race +-- between them. Use 'tryAddAfterEffect' for a non-throwing variant, or +-- 'addAfterEffectUnchecked' to skip the check entirely. addAfterEffect :: Actor message -> (Maybe SomeException -> IO ()) -> STM () -addAfterEffect = addAfterEffect' +addAfterEffect = withLivenessCheck addAfterEffectUnchecked --- | Register an after-effect only if the actor is alive. If the actor has --- already stopped, throw t'ActorDead'. The liveness check and registration are --- one STM transaction, so actor completion cannot race between them. +-- | Compatibility name for 'addAfterEffect'. -- -- @since 0.4.0.0 addAfterEffectChecked :: Actor message -> (Maybe SomeException -> IO ()) -> STM () -addAfterEffectChecked = withLivenessCheck addAfterEffect +addAfterEffectChecked = addAfterEffect + +-- | Register an after-effect without checking liveness. This avoids reading +-- the lifecycle 'TVar', but registering against an actor which has already +-- stopped stores an effect that can never run and is retained by the handle. +-- Use it only when the caller already controls the actor lifecycle. +-- +-- @since 0.4.0.0 +addAfterEffectUnchecked :: Actor message -> (Maybe SomeException -> IO ()) -> STM () +addAfterEffectUnchecked = addAfterEffect' -- | Attempt to register an after-effect. Return 'False' without registering it -- if the actor has already stopped. -- -- @since 0.4.0.0 tryAddAfterEffect :: Actor message -> (Maybe SomeException -> IO ()) -> STM Bool -tryAddAfterEffect = tryWhileAlive addAfterEffect +tryAddAfterEffect = tryWhileAlive addAfterEffectUnchecked -- | Retrieve the 'ThreadId' associated with this t'Actor'. threadId :: Actor message -> ThreadId @@ -287,13 +327,54 @@ instance Contravariant Actor where , send' = send' actor . f , trySend' = trySend' actor . f , actorState = actorState actor + , effectsFinished = effectsFinished actor } +-- | How to create an actor with 'actWith'. Start from 'defaultActorConfig' +-- and override fields with record update syntax. +-- +-- @since 0.4.0.0 +data ActorConfig message a = ActorConfig + { mailboxCapacity :: Maybe Natural + -- ^ 'Nothing' for an unbounded mailbox, or 'Just' the maximum number of + -- queued messages. The message currently being handled is no longer queued + -- and does not count against this capacity. A capacity of zero creates a + -- mailbox to which no send can commit. + , onCompletion :: Either SomeException a -> IO () + -- ^ Run in the actor's terminating thread with the result of its action, + -- after link and monitor notifications have been initiated and before the + -- undelivered message handler and user after-effects. + , onUndelivered :: [message] -> IO () + -- ^ Run after 'onCompletion' with the messages which were still queued when + -- the actor stopped, in mailbox order. It is not called when no messages + -- were queued. Every committed 'send' either reaches a handler or reaches + -- this function. + , onEffectFailure :: SomeException -> IO () + -- ^ Called in the actor's terminating thread for each completion effect + -- that throws, with that exception, before the remaining effects run. The + -- default rethrows, so after every effect has been attempted the + -- terminating thread rethrows the first such exception. + } + +-- | An unbounded mailbox and no-op handlers, except that effect failures are +-- rethrown. +-- +-- @since 0.4.0.0 +defaultActorConfig :: ActorConfig message a +defaultActorConfig = ActorConfig + { mailboxCapacity = Nothing + , onCompletion = const (pure ()) + , onUndelivered = const (pure ()) + , onEffectFailure = throwIO + } + -- | Perform some t'ActionT' in a new thread. Once the action stops, record its --- result, drain its mailbox, initiate link notifications, run the supplied --- completion handler, and then drain all registered user after-effects. +-- result, drain its mailbox, initiate link and monitor notifications, run the +-- supplied completion handler, and then drain all registered user +-- after-effects. actFinally :: (Either SomeException a -> IO ()) -> ActionT message IO a -> IO (Actor message) -actFinally = actFinallyWith newUnboundedMailbox +actFinally completionHandler = + actWith defaultActorConfig { onCompletion = completionHandler } -- | Like 'actFinally', but use a bounded FIFO mailbox with space for at most -- the given number of queued messages. Sends retry transactionally while the @@ -301,133 +382,113 @@ actFinally = actFinallyWith newUnboundedMailbox -- therefore does not count against this capacity. A capacity of zero creates a -- mailbox to which no send can commit. -- +-- The mailbox is a bounded @stm-queue@ queue, which tracks free capacity as +-- split read and write credits. Senders and the actor therefore conflict on +-- capacity accounting once per @capacity@ sends rather than on every message. +-- -- @since 0.4.0.0 actFinallyBounded :: Natural -> (Either SomeException a -> IO ()) -> ActionT message IO a -> IO (Actor message) -actFinallyBounded capacity = actFinallyWith (newBoundedMailbox capacity) +actFinallyBounded capacity completionHandler = actWith defaultActorConfig + { mailboxCapacity = Just capacity + , onCompletion = completionHandler + } -actFinallyWith - :: STM (Mailbox message) - -> (Either SomeException a -> IO ()) - -> ActionT message IO a - -> IO (Actor message) -actFinallyWith newMailbox completionHandler (ActionT actionT) = do +-- | Perform some t'ActionT' in a new thread. Use 'await' to observe when its +-- action stops. +act :: ActionT message IO a -> IO (Actor message) +act = actWith defaultActorConfig + +-- | Like 'act', but use a bounded FIFO mailbox with space for at most the +-- given number of queued messages. See 'actFinallyBounded'. +-- +-- @since 0.4.0.0 +actBounded :: Natural -> ActionT message IO a -> IO (Actor message) +actBounded capacity = actWith defaultActorConfig { mailboxCapacity = Just capacity } + +-- | Perform some t'ActionT' in a new thread, configured by an t'ActorConfig'. +-- The other creation functions are specialisations of this one. +-- +-- The mailbox is an @stm-queue@ queue. Its unbounded and bounded variants +-- share one type, so 'send' retries only when a bounded mailbox is full, +-- 'trySend' never retries, receiving releases bounded capacity, and shutdown +-- makes all capacity available again. +-- +-- @since 0.4.0.0 +actWith :: ActorConfig message a -> ActionT message IO a -> IO (Actor message) +actWith config (ActionT actionT) = do afterEffects <- newTVarIO [] terminationEffects <- newTVarIO [] - mailbox <- atomically newMailbox + mailbox <- maybe newQueueIO newBoundedQueueIO (mailboxCapacity config) stateVar <- newTVarIO Running - let registerEffect afterEffect = modifyTVar' afterEffects (afterEffect :) - registerTerminationEffect afterEffect = - modifyTVar' terminationEffects (afterEffect :) - enqueueMessage = writeMailbox mailbox - tryEnqueueMessage = tryWriteMailbox mailbox - makeActor actorThread = Actor - { addAfterEffect' = registerEffect - , addTerminationEffect' = registerTerminationEffect + finishedVar <- newTVarIO False + let makeActor actorThread = Actor + { addAfterEffect' = \afterEffect -> modifyTVar' afterEffects (afterEffect :) + , addTerminationEffect' = \afterEffect -> + modifyTVar' terminationEffects (afterEffect :) , threadId' = actorThread - , send' = enqueueMessage - , trySend' = tryEnqueueMessage + , send' = enqueue mailbox + , trySend' = tryEnqueue mailbox , actorState = stateVar + , effectsFinished = finishedVar } actorThread <- forkFinally (do currentThread <- myThreadId - actionT (ActorContext (readMailbox mailbox) (makeActor currentThread))) - (finishActor stateVar mailbox terminationEffects afterEffects completionHandler) + actionT (ActorContext (dequeue mailbox) (makeActor currentThread))) + (finishActor config stateVar finishedVar mailbox terminationEffects afterEffects) pure (makeActor actorThread) -newUnboundedMailbox :: STM (Mailbox message) -newUnboundedMailbox = do - queue <- newQueue - pure Mailbox - { readMailbox = dequeue queue - , writeMailbox = enqueue queue - , tryWriteMailbox = \message -> enqueue queue message >> pure True - , clearMailbox = void (flush queue) - } - -newBoundedMailbox :: Natural -> STM (Mailbox message) -newBoundedMailbox capacity = do - queue <- newQueue - size <- newTVar 0 - let reserveSlot = do - current <- readTVar size - check (current < capacity) - writeTVar size $! current + 1 - tryReserveSlot = do - current <- readTVar size - if current < capacity - then do - writeTVar size $! current + 1 - pure True - else pure False - releaseSlot = modifyTVar' size (subtract 1) - readMessage = do - message <- dequeue queue - releaseSlot - pure message - writeMessage message = reserveSlot >> enqueue queue message - tryWriteMessage message = tryReserveSlot >>= \case - True -> enqueue queue message >> pure True - False -> pure False - clear = void (flush queue) >> writeTVar size 0 - pure Mailbox - { readMailbox = readMessage - , writeMailbox = writeMessage - , tryWriteMailbox = tryWriteMessage - , clearMailbox = clear - } - finishActor - :: TVar ActorState - -> Mailbox message + :: ActorConfig message a + -> TVar ActorState + -> TVar Bool + -> Queue message -> TVar [AfterEffect] -> TVar [AfterEffect] - -> (Either SomeException a -> IO ()) -> Either SomeException a -> IO () -finishActor stateVar mailbox terminationEffects afterEffects completionHandler result = do - (earlyEffects, effects) <- atomically do +finishActor config stateVar finishedVar mailbox terminationEffects afterEffects result = do + (earlyEffects, effects, undelivered) <- atomically do writeTVar stateVar (Stopped completion) - clearMailbox mailbox + undelivered <- flush mailbox registeredEarly <- readTVar terminationEffects registered <- readTVar afterEffects writeTVar terminationEffects [] writeTVar afterEffects [] - pure (reverse registeredEarly, reverse registered) - runAllEffects - (map ($ completion) earlyEffects - <> (completionHandler result : map ($ completion) effects)) + pure (reverse registeredEarly, reverse registered, undelivered) + runAllEffects (onEffectFailure config) + ( map ($ completion) earlyEffects + <> [onCompletion config result] + <> [onUndelivered config undelivered | not (null undelivered)] + <> map ($ completion) effects + ) + `finally` atomically (writeTVar finishedVar True) where completion = either Just (const Nothing) result -runAllEffects :: [IO ()] -> IO () -runAllEffects effects = mask_ (go Nothing effects) +-- The terminating thread is already masked by 'forkFinally'; the explicit mask +-- keeps this function correct on its own. Asynchronous exceptions delivered to +-- an interruptible effect are caught like any other effect failure, so cleanup +-- is never truncated. +runAllEffects :: (SomeException -> IO ()) -> [IO ()] -> IO () +runAllEffects onFailure effects = mask_ (go Nothing effects) where go :: Maybe SomeException -> [IO ()] -> IO () go firstException [] = maybe (pure ()) throwIO firstException go firstException (effect : remaining) = try effect >>= \case - Left exception -> go (rememberFirst firstException exception) remaining + Left exception -> try (onFailure exception) >>= \case + Left handlerException -> + go (rememberFirst firstException handlerException) remaining + Right () -> go firstException remaining Right () -> go firstException remaining rememberFirst Nothing exception = Just exception rememberFirst remembered _ = remembered --- | Perform some t'ActionT' in a new thread. Use 'await' to observe when its --- action stops. -act :: ActionT message IO a -> IO (Actor message) -act = actFinally (const (pure ())) - --- | Like 'act', but use a bounded FIFO mailbox with space for at most the --- given number of queued messages. See 'actFinallyBounded'. --- --- @since 0.4.0.0 -actBounded :: Natural -> ActionT message IO a -> IO (Actor message) -actBounded capacity = actFinallyBounded capacity (const (pure ())) - -- | Receive a message and do some t'ActionT' with it. receive :: MonadIO m => (message -> ActionT message m a) -> ActionT message m a receive f = ActionT \ctx -> do @@ -453,6 +514,9 @@ instance Exception LinkKill -- stops, whether normally or exceptionally, it will throw a t'LinkKill' -- exception to us with its 'ThreadId' attached. Linking to an actor that has -- already stopped throws the same t'LinkKill' immediately. +-- +-- Links are for cancellation. To be told that an actor stopped without being +-- interrupted, use 'monitor'. link :: MonadIO m => Actor message -> ActionT message' m () link you = do me <- self @@ -480,6 +544,51 @@ signalLink alice bob = do _ <- forkIO $ throwTo (threadId alice) (LinkKill (threadId bob)) pure () +-- | Monitor the given actor from this one. When it stops, whether normally or +-- exceptionally, the message built from its completion is sent to our mailbox, +-- so it is handled like any other message rather than interrupting us. If the +-- given actor has already stopped, the message is sent immediately. +-- +-- @since 0.4.0.0 +monitor + :: MonadIO m + => Actor message' + -> (Maybe SomeException -> message) + -> ActionT message m () +monitor target toMessage = do + me <- self + liftIO (atomically (monitorSTM me target toMessage)) + +-- | Make the first actor monitor the second. When the second actor stops, the +-- message built from its completion ('Nothing' for normal completion, 'Just' +-- the exception otherwise) is sent to the first actor. If the second actor has +-- already stopped, the message is sent in this transaction, retrying like +-- 'send' if the first actor's bounded mailbox is full. If the first actor has +-- already stopped, throw t'ActorDead'. +-- +-- Like links, monitor notifications are initiated at the second actor's +-- lifecycle transition, before its completion handler and after-effects, and +-- are delivered from a helper thread which waits for mailbox capacity. A +-- notification to a first actor which has since stopped is dropped. +-- +-- @since 0.4.0.0 +monitorSTM + :: Actor message + -> Actor message' + -> (Maybe SomeException -> message) + -> STM () +monitorSTM recipient target toMessage = do + ensureAlive recipient + readTVar (actorState target) >>= \case + Stopped completion -> send' recipient (toMessage completion) + Running -> addTerminationEffect' target \completion -> + signalMonitor recipient (toMessage completion) + +signalMonitor :: Actor message -> message -> IO () +signalMonitor recipient message = void $ forkIO $ + atomically (send recipient message) + `catch` \(ActorDead _) -> pure () + -- | Returns the t'Actor' handle of the actor executing this action. self :: Applicative m => ActionT message m (Actor message) self = ActionT (pure . actorHandle) @@ -490,10 +599,15 @@ data MurderKill = MurderKill ThreadId instance Exception MurderKill --- | Synchronously throw a t'MurderKill' exception to the given t'Actor'. As --- with 'throwTo', this can block while the target is uninterruptibly masking --- asynchronous exceptions. +-- | Throw a t'MurderKill' exception to the given t'Actor' if it is still alive. +-- As with 'throwTo', this can block while the target is uninterruptibly +-- masking asynchronous exceptions. Once the actor has stopped this does +-- nothing, so completion effects are not interrupted; a murder that races the +-- lifecycle transition may still reach the terminating thread, where it is +-- recorded as an effect failure rather than truncating cleanup. murder :: MonadIO m => Actor message -> m () murder actor = liftIO do murderer <- myThreadId - throwTo (threadId actor) (MurderKill murderer) + atomically (livenessCheck actor) >>= \case + Alive -> throwTo (threadId actor) (MurderKill murderer) + _ -> pure () diff --git a/stm-actor.cabal b/stm-actor.cabal index 0c3aee7..ab2708a 100644 --- a/stm-actor.cabal +++ b/stm-actor.cabal @@ -41,7 +41,7 @@ library exposed-modules: Control.Concurrent.Actor build-depends: base >=4.18 && <5, stm >=2.5 && <2.6, - stm-queue >=0.2.0.0 && <0.3, + stm-queue >=0.2.2.0 && <0.3, mtl >=1.0 && <2.4, unliftio-core >=0.2 && <0.3, transformers >=0.2 && <0.7 @@ -58,6 +58,18 @@ test-suite stm-actor-test hspec >=2.7.4 && <2.12, mtl >=1.0 && <2.4, stm-actor - other-extensions: BlockArguments, LambdaCase + other-extensions: BlockArguments, LambdaCase, ScopedTypeVariables ghc-options: -threaded -rtsopts default-language: Haskell2010 + +benchmark stm-actor-benchmark + import: warnings + type: exitcode-stdio-1.0 + hs-source-dirs: bench + main-is: Bench.hs + build-depends: base >=4.18 && <5, + stm >=2.5 && <2.6, + stm-actor + other-extensions: BangPatterns, BlockArguments + ghc-options: -threaded -rtsopts "-with-rtsopts=-N" + default-language: Haskell2010 diff --git a/test/Test.hs b/test/Test.hs index ad289fb..42c7c63 100644 --- a/test/Test.hs +++ b/test/Test.hs @@ -1,5 +1,6 @@ {-# LANGUAGE BlockArguments #-} {-# LANGUAGE LambdaCase #-} +{-# LANGUAGE ScopedTypeVariables #-} module Main where import Control.Concurrent (forkIO) @@ -11,6 +12,7 @@ import Control.Concurrent.STM import Control.Exception ( ArithException(Underflow) , AsyncException(ThreadKilled) + , BlockedIndefinitelyOnSTM(BlockedIndefinitelyOnSTM) , SomeException , fromException , throwIO @@ -22,6 +24,7 @@ import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask, runReaderT) import Data.Functor.Contravariant (contramap) import Data.IORef +import System.Mem (performMajorGC) import System.Timeout import Test.Hspec @@ -76,6 +79,33 @@ main = hspec do within "large after-effect set" (takeMVar finished) `shouldReturn` () readIORef effectCount `shouldReturn` 10000 + it "is lifecycle-checked by default, with an unchecked variant" do + actor <- act (pure ()) + _ <- within "actor completion" (awaitStopped actor) + atomically (addAfterEffect actor (const (pure ()))) + `shouldThrow` isActorDead + atomically (addAfterEffectUnchecked actor (const (pure ()))) + `shouldReturn` () + + it "reports effect failures to the configured handler" do + release <- newEmptyMVar + failures <- newIORef [] + finished <- newEmptyMVar + actor <- actWith defaultActorConfig + { onEffectFailure = \exception -> + atomicModifyIORef' failures (\seen -> (exception : seen, ())) + } + (liftIO (takeMVar release)) + atomically do + addAfterEffect actor (const (throwIO Underflow)) + addAfterEffect actor (const (throwIO ThreadKilled)) + addAfterEffect actor (const (putMVar finished ())) + putMVar release () + within "effects after failures" (takeMVar finished) `shouldReturn` () + _ <- within "effect completion" (atomically (awaitEffects actor)) + seen <- reverse <$> readIORef failures + map isUnderflowException seen `shouldBe` [True, False] + it "offers atomic checked and non-throwing registration" do release <- newEmptyMVar effectRan <- newEmptyMVar @@ -159,9 +189,16 @@ main = hspec do status -> expectationFailure ("expected Completed, got " <> show status) atomically (addAfterEffectChecked actor (const (pure ()))) `shouldThrow` isActorDead + timeout 100000 (atomically (awaitEffects actor)) >>= \case + Nothing -> pure () + Just status -> expectationFailure + ("awaitEffects returned while an effect was blocked: " <> show status) putMVar releaseEffect () within "remaining after-effects" (takeMVar effectsDrained) `shouldReturn` () + within "awaitEffects" (atomically (awaitEffects actor)) >>= \case + Completed -> pure () + status -> expectationFailure ("expected Completed, got " <> show status) describe "sending" do it "sends while alive and rejects normal sends after completion" do @@ -249,6 +286,28 @@ main = hspec do ("expected ActorStopped, got " <> show sendResult) atomically (send actor "second") `shouldThrow` isActorDead + it "delivers messages from many senders exactly once and in order" do + let senders = 8 :: Int + perSender = 500 :: Int + received <- newIORef [] + drained <- newEmptyMVar + actor <- actBounded 4 do + replicateM_ (senders * perSender) $ receive \message -> + liftIO (atomicModifyIORef' received (\messages -> (message : messages, ()))) + liftIO (putMVar drained ()) + finished <- newEmptyMVar + forM_ [1 .. senders] \sender -> forkIO do + forM_ [1 .. perSender] \i -> atomically (send actor (sender, i)) + putMVar finished () + replicateM_ senders (within "sender completion" (takeMVar finished)) + within "bounded fan-in drain" (takeMVar drained) `shouldReturn` () + messages <- reverse <$> readIORef received + forM_ [1 .. senders] \sender -> + [i | (sender', i) <- messages, sender' == sender] + `shouldBe` [1 .. perSender] + _ <- within "fan-in actor completion" (awaitStopped actor) + pure () + it "wakes a blocked normal sender when the actor stops" do blocker <- newEmptyMVar actor <- actBounded 1 (liftIO (takeMVar blocker)) @@ -265,6 +324,28 @@ main = hspec do Left _ -> pure () Right () -> expectationFailure "send unexpectedly committed" + describe "undelivered messages" do + it "hands queued messages to the configured handler in order" do + blocker <- newEmptyMVar + undelivered <- newEmptyMVar + actor <- actWith defaultActorConfig + { mailboxCapacity = Just 8 + , onUndelivered = putMVar undelivered + } + (liftIO (takeMVar blocker)) + atomically (forM_ [1 .. 3 :: Int] (send actor)) + murder actor + within "undelivered messages" (takeMVar undelivered) + `shouldReturn` [1, 2, 3] + + it "does not call the handler when nothing was queued" do + called <- newIORef False + actor <- actWith defaultActorConfig + { onUndelivered = \(_ :: [Int]) -> writeIORef called True } + (pure ()) + _ <- within "empty-mailbox completion" (atomically (awaitEffects actor)) + readIORef called `shouldReturn` False + describe "receive" do it "can receive messages" do result <- newEmptyMVar @@ -293,6 +374,11 @@ main = hspec do `shouldReturn` True describe "murder" do + it "does nothing once the actor has stopped" do + actor <- act (pure ()) + _ <- within "actor completion" (awaitStopped actor) + within "murder of a stopped actor" (murder actor) `shouldReturn` () + it "kills actors" do blocker <- newEmptyMVar :: IO (MVar ()) result <- newEmptyMVar @@ -419,6 +505,58 @@ main = hspec do _ <- within "link target cleanup" (awaitStopped target) pure () + describe "monitor" do + it "delivers a message when the target completes normally" do + releaseTarget <- newEmptyMVar + result <- newEmptyMVar + target <- act (liftIO (takeMVar releaseTarget)) + _ <- act do + monitor target TargetDown + receive \(TargetDown completion) -> liftIO (putMVar result completion) + putMVar releaseTarget () + within "normal monitor notification" (takeMVar result) >>= \case + Nothing -> pure () + Just exception -> expectationFailure + ("expected normal completion, got " <> show exception) + + it "delivers the exception when the target fails" do + releaseTarget <- newEmptyMVar + result <- newEmptyMVar + target <- act do + liftIO (takeMVar releaseTarget) + liftIO (throwIO Underflow) + _ <- act do + monitor target TargetDown + receive \(TargetDown completion) -> liftIO (putMVar result completion) + putMVar releaseTarget () + within "failure monitor notification" (takeMVar result) >>= \case + Just exception | isUnderflowException exception -> pure () + completion -> expectationFailure + ("expected Underflow, got " <> show completion) + + it "notifies immediately about an already-stopped target" do + target <- act (pure ()) + _ <- within "target completion" (awaitStopped target) + result <- newEmptyMVar + _ <- act do + monitor target TargetDown + receive \(TargetDown completion) -> liftIO (putMVar result completion) + within "late monitor notification" (takeMVar result) >>= \case + Nothing -> pure () + Just exception -> expectationFailure + ("expected normal completion, got " <> show exception) + + it "rejects a stopped recipient and drops notifications to one" do + recipient <- act (pure ()) + _ <- within "recipient completion" (awaitStopped recipient) + targetBlocker <- newEmptyMVar + target <- act (liftIO (takeMVar targetBlocker)) + atomically (monitorSTM recipient target TargetDown) + `shouldThrow` isActorDead + murder target + _ <- within "monitor target cleanup" (awaitStopped target) + pure () + describe "self" do it "returns the actor's real handle" do result <- newEmptyMVar @@ -480,13 +618,34 @@ main = hspec do Completed -> pure () status -> expectationFailure ("expected Completed, got " <> show status) + describe "garbage collection" do + it "stops a receiver whose handle has been dropped" do + stopped <- newEmptyMVar + do + actor <- act (receive (\() -> pure ())) + atomically (addAfterEffect actor (putMVar stopped)) + let collect attempts = do + performMajorGC + timeout 200000 (takeMVar stopped) >>= \case + Just completion -> pure (Just completion) + Nothing + | attempts > (1 :: Int) -> collect (attempts - 1) + | otherwise -> pure Nothing + collect 25 >>= \case + Just (Just exception) + | Just BlockedIndefinitelyOnSTM <- fromException exception -> pure () + outcome -> expectationFailure + ("expected BlockedIndefinitelyOnSTM, got " <> show outcome) + describe "withLivenessCheck" do it "doesn't let you add after-effects to dead actors" do actor <- act (pure ()) _ <- within "target completion" (awaitStopped actor) - atomically (withLivenessCheck addAfterEffect actor (const (pure ()))) + atomically (withLivenessCheck addAfterEffectUnchecked actor (const (pure ()))) `shouldThrow` isActorDead +newtype TargetDown = TargetDown (Maybe SomeException) + within :: String -> IO a -> IO a within label action = timeout 5000000 action >>= \case Nothing -> expectationFailure message >> fail message