Effectful.Concurrent

Lifted Control.Concurrent.

For functions that spawn threads, the order of preference for their usage is recommended as follows:

1) High level functions from Effectful.Concurrent.Async such as withAsync, concurrently or mapConcurrently.

2) Low level functions from Effectful.Concurrent.Async such as async.

3) Low level functions from Effectful.Concurrent such as forkIO.

Effect

data Concurrent :: Effect

Provide the ability to run Eff computations concurrently in multiple threads and communicate between them.

Warning: unless you stick to high level functions from the withAsync family, the Concurrent effect makes it possible to escape the scope of any scoped effect operation. Consider the following:

>>> import Effectful.Reader.Static qualified as R

>>> printAsk msg = liftIO . putStrLn . (msg ++) . (": " ++) =<< R.ask

>>> :{
  runEff . R.runReader "GLOBAL" . runConcurrent $ do
    a <- R.local (const "LOCAL") $ do
      a <- async $ do
        printAsk "child (first)"
        threadDelay 20000
        printAsk "child (second)"
      threadDelay 10000
      printAsk "parent (inside)"
      pure a
    printAsk "parent (outside)"
    wait a
:}
child (first): LOCAL
parent (inside): LOCAL
parent (outside): GLOBAL
child (second): LOCAL

Note that the asynchronous computation doesn't respect the scope of local, i.e. the child thread still behaves like it's inside the local block, even though the parent thread already got out of it.

This is because the value provided by the Reader effect is thread local, i.e. each thread manages its own version of it. For the Reader it is the only reasonable behavior, it wouldn't be very useful if its "read only" value was affected by calls to local from its parent or child threads.

However, the cut isn't so clear if it comes to effects that provide access to a mutable state. That's why statically dispatched State and Writer effects come in two flavors, local and shared:

>>> import Effectful.State.Static.Local qualified as SL
>>> :{
  runEff . SL.execState "Hi" . runConcurrent $ do
    replicateConcurrently_ 3 $ SL.modify (++ "!")
:}
"Hi"

>>> import Effectful.State.Static.Shared qualified as SS
>>> :{
  runEff . SS.execState "Hi" . runConcurrent $ do
    replicateConcurrently_ 3 $ SS.modify (++ "!")
:}
"Hi!!!"

In the first example state updates made concurrently are not reflected in the parent thread because the value is thread local, but in the second example they are, because the value is shared.

Instances

type DispatchOf Concurrent

Defined in Effectful.Concurrent.Effect

type DispatchOf Concurrent = 'Static 'WithSideEffects

data StaticRep Concurrent

Defined in Effectful.Concurrent.Effect

data StaticRep Concurrent = Concurrent

Handlers

runConcurrent :: (HasCallStack, IOE :> es) => Eff (Concurrent : es) a -> Eff es a

Run the Concurrent effect.

Basic concurrency operations

myThreadId :: Concurrent :> es => Eff es ThreadId

Lifted myThreadId.

forkIO :: (HasCallStack, Concurrent :> es) => Eff es () -> Eff es ThreadId

Lifted forkIO.

forkFinally :: (HasCallStack, Concurrent :> es) => Eff es a -> (Either SomeException a -> Eff es ()) -> Eff es ThreadId

Lifted forkFinally.

forkIOWithUnmask :: (HasCallStack, Concurrent :> es) => ((forall a. Eff es a -> Eff es a) -> Eff es ()) -> Eff es ThreadId

Lifted forkIOWithUnmask.

killThread :: Concurrent :> es => ThreadId -> Eff es ()

Lifted killThread.

throwTo :: (Concurrent :> es, Exception e) => ThreadId -> e -> Eff es ()

Lifted throwTo.

Threads with affinity

forkOn :: (HasCallStack, Concurrent :> es) => Int -> Eff es () -> Eff es ThreadId

Lifted forkOn.

forkOnWithUnmask :: (HasCallStack, Concurrent :> es) => Int -> ((forall a. Eff es a -> Eff es a) -> Eff es ()) -> Eff es ThreadId

Lifted forkOnWithUnmask.

getNumCapabilities :: Concurrent :> es => Eff es Int

Lifted getNumCapabilities.

setNumCapabilities :: Concurrent :> es => Int -> Eff es ()

Lifted setNumCapabilities.

getNumProcessors :: Concurrent :> es => Eff es Int

Lifted getNumProcessors.

threadCapability :: Concurrent :> es => ThreadId -> Eff es (Int, Bool)

Lifted threadCapability.

Scheduling

yield :: Concurrent :> es => Eff es ()

Lifted yield.

Waiting

threadDelay :: Concurrent :> es => Int -> Eff es ()

Lifted threadDelay.

threadWaitRead :: Concurrent :> es => Fd -> Eff es ()

Lifted threadWaitRead.

threadWaitWrite :: Concurrent :> es => Fd -> Eff es ()

Lifted threadWaitWrite.

threadWaitReadSTM :: Concurrent :> es => Fd -> Eff es (STM (), Eff es ())

Lifted threadWaitReadSTM.

threadWaitWriteSTM :: Concurrent :> es => Fd -> Eff es (STM (), Eff es ())

Lifted threadWaitWriteSTM.

Bound threads

forkOS :: (HasCallStack, Concurrent :> es) => Eff es () -> Eff es ThreadId

Lifted forkOS.

forkOSWithUnmask :: (HasCallStack, Concurrent :> es) => ((forall a. Eff es a -> Eff es a) -> Eff es ()) -> Eff es ThreadId

Lifted forkOSWithUnmask.

isCurrentThreadBound :: Concurrent :> es => Eff es Bool

Lifted isCurrentThreadBound.

runInBoundThread :: (HasCallStack, Concurrent :> es) => Eff es a -> Eff es a

Lifted runInBoundThread.

runInUnboundThread :: (HasCallStack, Concurrent :> es) => Eff es a -> Eff es a

Lifted runInUnboundThread.

Weak references to ThreadIds

mkWeakThreadId :: Concurrent :> es => ThreadId -> Eff es (Weak ThreadId)

Lifted mkWeakThreadId.

Re-exports

rtsSupportsBoundThreads :: Bool

True if bound threads are supported. If rtsSupportsBoundThreads is False, isCurrentThreadBound will always return False and both forkOS and runInBoundThread will fail.