From 7cda5af9271d570aa4dfde4c4a6bd614e8f88ab3 Mon Sep 17 00:00:00 2001 From: Samuel Schlesinger Date: Sun, 23 Aug 2026 21:20:47 -0400 Subject: [PATCH 1/2] Use stm-queue's bounded queues for bounded mailboxes stm-actor implemented bounded mailboxes as an unbounded stm-queue plus a single occupancy TVar that every send and every receive wrote, so senders and the actor conflicted on every message. stm-queue-0.2.2.0 provides bounded queues whose free capacity is tracked as split read and write credits, so capacity accounting conflicts once per `capacity` sends instead. Delegate to those queues and drop the in-house Mailbox record: enqueue retries only on a full bounded queue, tryEnqueue never retries, dequeue releases capacity, and flush releases all of it. Construct mailboxes with newQueueIO and newBoundedQueueIO instead of an atomically block. Require stm-queue >= 0.2.2.0, pinned to its git commit until it is published on Hackage. Add a multi-sender bounded mailbox test. --- CHANGELOG.md | 5 +- README.md | 5 +- cabal.project | 7 +++ src/Control/Concurrent/Actor.hs | 82 ++++++++------------------------- stm-actor.cabal | 2 +- test/Test.hs | 22 +++++++++ 6 files changed, 56 insertions(+), 67 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d030445..808575a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,7 +24,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..9aad404 100644 --- a/README.md +++ b/README.md @@ -26,8 +26,9 @@ with other threads and actors. `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 +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 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..63afdcd 100644 --- a/src/Control/Concurrent/Actor.hs +++ b/src/Control/Concurrent/Actor.hs @@ -73,9 +73,7 @@ import Control.Concurrent.STM ( STM , TVar , atomically - , check , modifyTVar' - , newTVar , newTVarIO , readTVar , retry @@ -95,7 +93,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 +128,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 @@ -293,7 +285,7 @@ instance Contravariant Actor where -- result, drain its mailbox, initiate link 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 = actFinallyWith newQueueIO -- | 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,89 +293,53 @@ 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 = actFinallyWith (newBoundedQueueIO capacity) +-- The mailbox is an @stm-queue@ queue. Its unbounded and bounded variants +-- share one type, so 'enqueue' retries only when a bounded mailbox is full, +-- 'tryEnqueue' never retries, 'dequeue' releases bounded capacity, and 'flush' +-- makes all capacity available again. actFinallyWith - :: STM (Mailbox message) + :: IO (Queue message) -> (Either SomeException a -> IO ()) -> ActionT message IO a -> IO (Actor message) actFinallyWith newMailbox completionHandler (ActionT actionT) = do afterEffects <- newTVarIO [] terminationEffects <- newTVarIO [] - mailbox <- atomically newMailbox + mailbox <- newMailbox 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 , threadId' = actorThread - , send' = enqueueMessage - , trySend' = tryEnqueueMessage + , send' = enqueue mailbox + , trySend' = tryEnqueue mailbox , actorState = stateVar } actorThread <- forkFinally (do currentThread <- myThreadId - actionT (ActorContext (readMailbox mailbox) (makeActor currentThread))) + actionT (ActorContext (dequeue mailbox) (makeActor currentThread))) (finishActor stateVar mailbox terminationEffects afterEffects completionHandler) 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 + -> Queue message -> TVar [AfterEffect] -> TVar [AfterEffect] -> (Either SomeException a -> IO ()) @@ -392,7 +348,7 @@ finishActor finishActor stateVar mailbox terminationEffects afterEffects completionHandler result = do (earlyEffects, effects) <- atomically do writeTVar stateVar (Stopped completion) - clearMailbox mailbox + void (flush mailbox) registeredEarly <- readTVar terminationEffects registered <- readTVar afterEffects writeTVar terminationEffects [] diff --git a/stm-actor.cabal b/stm-actor.cabal index 0c3aee7..65a523c 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 diff --git a/test/Test.hs b/test/Test.hs index ad289fb..33ff9e1 100644 --- a/test/Test.hs +++ b/test/Test.hs @@ -249,6 +249,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)) From 13b379c57ba09c481d85f81dc10559d03844c63a Mon Sep 17 00:00:00 2001 From: Samuel Schlesinger Date: Sun, 23 Aug 2026 21:37:10 -0400 Subject: [PATCH 2/2] Add actor configuration, monitors, awaitEffects, and safer lifecycle defaults Introduce actWith and ActorConfig so an actor's mailbox capacity, completion handler, undelivered-message handler, and effect-failure handler are configured in one place; act, actBounded, actFinally, and actFinallyBounded become specialisations. Messages still queued when an actor stops are handed to onUndelivered in mailbox order instead of being discarded silently, and each completion effect that throws is reported to onEffectFailure, which rethrows by default to preserve the previous behaviour. Make addAfterEffect lifecycle-checked by default, matching send, with addAfterEffectUnchecked for callers that control the lifecycle. Add monitor and monitorSTM, which deliver an actor's completion to the monitoring actor's mailbox as an ordinary message rather than an asynchronous exception, and notify immediately about an already-stopped target. Add awaitEffects, which waits for the completion handler and after-effects to finish. Make murder a no-op once the actor has stopped so cleanup is not interrupted. Document that a receiver whose handles have all been dropped is stopped by the runtime with BlockedIndefinitelyOnSTM, and test it. Add a fan-in throughput benchmark comparing unbounded and bounded mailboxes. Date the 0.4.0.0 changelog entry for release. --- .github/workflows/haskell.yml | 4 +- CHANGELOG.md | 22 +- README.md | 87 ++++++-- bench/Bench.hs | 57 ++++++ src/Control/Concurrent/Actor.hs | 346 +++++++++++++++++++++++--------- stm-actor.cabal | 14 +- test/Test.hs | 139 ++++++++++++- 7 files changed, 549 insertions(+), 120 deletions(-) create mode 100644 bench/Bench.hs 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 808575a..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. diff --git a/README.md b/README.md index 9aad404..e9ac8f1 100644 --- a/README.md +++ b/README.md @@ -25,8 +25,10 @@ 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 are +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 @@ -63,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 @@ -76,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`. @@ -114,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/src/Control/Concurrent/Actor.hs b/src/Control/Concurrent/Actor.hs index 63afdcd..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) @@ -73,6 +98,7 @@ import Control.Concurrent.STM ( STM , TVar , atomically + , check , modifyTVar' , newTVarIO , readTVar @@ -81,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) @@ -137,6 +163,7 @@ data Actor message = Actor , send' :: message -> STM () , trySend' :: message -> STM Bool , actorState :: TVar ActorState + , effectsFinished :: TVar Bool } type Completion = Maybe SomeException @@ -163,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 @@ -171,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 @@ -192,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 @@ -279,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 newQueueIO +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 @@ -303,87 +392,103 @@ actFinallyBounded -> (Either SomeException a -> IO ()) -> ActionT message IO a -> IO (Actor message) -actFinallyBounded capacity = actFinallyWith (newBoundedQueueIO capacity) +actFinallyBounded capacity completionHandler = actWith defaultActorConfig + { mailboxCapacity = Just capacity + , onCompletion = completionHandler + } + +-- | 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 'enqueue' retries only when a bounded mailbox is full, --- 'tryEnqueue' never retries, 'dequeue' releases bounded capacity, and 'flush' +-- 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. -actFinallyWith - :: IO (Queue message) - -> (Either SomeException a -> IO ()) - -> ActionT message IO a - -> IO (Actor message) -actFinallyWith newMailbox completionHandler (ActionT actionT) = do +-- +-- @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 <- newMailbox + mailbox <- maybe newQueueIO newBoundedQueueIO (mailboxCapacity config) stateVar <- newTVarIO Running - let registerEffect afterEffect = modifyTVar' afterEffects (afterEffect :) - registerTerminationEffect afterEffect = - modifyTVar' terminationEffects (afterEffect :) - 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' = enqueue mailbox , trySend' = tryEnqueue mailbox , actorState = stateVar + , effectsFinished = finishedVar } actorThread <- forkFinally (do currentThread <- myThreadId actionT (ActorContext (dequeue mailbox) (makeActor currentThread))) - (finishActor stateVar mailbox terminationEffects afterEffects completionHandler) + (finishActor config stateVar finishedVar mailbox terminationEffects afterEffects) pure (makeActor actorThread) finishActor - :: TVar ActorState + :: 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) - void (flush 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 @@ -409,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 @@ -436,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) @@ -446,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 65a523c..ab2708a 100644 --- a/stm-actor.cabal +++ b/stm-actor.cabal @@ -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 33ff9e1..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 @@ -287,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 @@ -315,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 @@ -441,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 @@ -502,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