{-# LANGUAGE MagicHash #-}
{-# LANGUAGE QuantifiedConstraints #-}
{-# LANGUAGE UnboxedTuples #-}
{-# OPTIONS_HADDOCK not-home #-}
-- | Implementation of sequential and concurrent unlifts.
--
-- This module is intended for internal use only, and may change without warning
-- in subsequent releases.
module Effectful.Internal.Unlift
  ( -- * Unlifting strategies
    UnliftStrategy(..)
  , Persistence(..)
  , Limit(..)

    -- * Unlifting functions
  , ephemeralConcLimitedUnlift
  , ephemeralConcUnlimitedUnlift
  , persistentConcUnlift
  , persistentConcSingleUnlift
  , persistentConcUnlifts
  , persistentConcSingleUnlifts
  ) where

import Control.Concurrent
import Control.Concurrent.MVar.Strict qualified as S
import Control.Monad
import Data.Coerce
import Data.Word
import GHC.Conc.Sync (ThreadId(..))
import GHC.Exts (mkWeak#, mkWeakNoFinalizer#)
import GHC.Generics (Generic)
import GHC.IO (IO(..))
import GHC.Stack (HasCallStack)
import GHC.Weak (Weak(..))
import System.Mem.Weak (deRefWeak)

import Effectful.Internal.Env
import Effectful.Internal.Utils
import Effectful.Internal.Utils.Word64Map qualified as M

----------------------------------------
-- Unlift strategies

-- | The strategy to use when unlifting 'Effectful.Eff' computations via
-- 'Effectful.withEffToIO' or the 'Effectful.Dispatch.Dynamic.localUnlift'
-- family.
data UnliftStrategy
  = SeqUnlift
  -- ^ The sequential strategy is the fastest and a default setting for
  -- t'Effectful.IOE'. Any attempt of calling the unlifting function in threads
  -- distinct from its creator will result in a runtime error.
  | SeqForkUnlift
  -- ^ Like 'SeqUnlift', but all unlifted actions will be executed in a cloned
  -- environment.
  --
  -- The main consequence is that thread local state is forked at the point of
  -- creation of the unlifting function and its modifications in unlifted
  -- actions will not affect the main thread of execution (and vice versa):
  --
  -- >>> import Effectful
  -- >>> import Effectful.State.Dynamic
  -- >>> :{
  --  action :: (IOE :> es, State Int :> es) => Eff es ()
  --  action = do
  --    modify @Int (+1)
  --    withEffToIO SeqForkUnlift $ \unlift -> unlift $ modify @Int (+2)
  --    modify @Int (+4)
  -- :}
  --
  -- >>> runEff . execStateLocal @Int 0 $ action
  -- 5
  --
  -- >>> runEff . execStateShared @Int 0 $ action
  -- 7
  --
  -- Because of this it's possible to safely use the unlifting function outside
  -- of the scope of effects it captures, e.g. by creating an @IO@ action that
  -- executes effectful operations and running it later:
  --
  -- >>> :{
  --   delayed :: UnliftStrategy -> IO (IO String)
  --   delayed strategy = runEff . evalStateLocal "Hey" $ do
  --     r <- withEffToIO strategy $ \unlift -> pure $ unlift get
  --     modify (++ "!!!")
  --     pure r
  -- :}
  --
  -- This doesn't work with the 'SeqUnlift' strategy because when the returned
  -- action runs, @State@ is no longer in scope:
  --
  -- >>> join $ delayed SeqUnlift
  -- *** Exception: version (...) /= storageVersion (0)
  -- ...
  --
  -- However, it does with the 'SeqForkUnlift' strategy:
  --
  -- >>> join $ delayed SeqForkUnlift
  -- "Hey"
  --
  | ConcUnlift !Persistence !Limit
  -- ^ The concurrent strategy makes it possible for the unlifting function to
  -- be called in threads distinct from its creator. See 'Persistence' and
  -- 'Limit' settings for more information.
  deriving stock (UnliftStrategy -> UnliftStrategy -> Bool
(UnliftStrategy -> UnliftStrategy -> Bool)
-> (UnliftStrategy -> UnliftStrategy -> Bool) -> Eq UnliftStrategy
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: UnliftStrategy -> UnliftStrategy -> Bool
== :: UnliftStrategy -> UnliftStrategy -> Bool
$c/= :: UnliftStrategy -> UnliftStrategy -> Bool
/= :: UnliftStrategy -> UnliftStrategy -> Bool
Eq, (forall x. UnliftStrategy -> Rep UnliftStrategy x)
-> (forall x. Rep UnliftStrategy x -> UnliftStrategy)
-> Generic UnliftStrategy
forall x. Rep UnliftStrategy x -> UnliftStrategy
forall x. UnliftStrategy -> Rep UnliftStrategy x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. UnliftStrategy -> Rep UnliftStrategy x
from :: forall x. UnliftStrategy -> Rep UnliftStrategy x
$cto :: forall x. Rep UnliftStrategy x -> UnliftStrategy
to :: forall x. Rep UnliftStrategy x -> UnliftStrategy
Generic, Eq UnliftStrategy
Eq UnliftStrategy =>
(UnliftStrategy -> UnliftStrategy -> Ordering)
-> (UnliftStrategy -> UnliftStrategy -> Bool)
-> (UnliftStrategy -> UnliftStrategy -> Bool)
-> (UnliftStrategy -> UnliftStrategy -> Bool)
-> (UnliftStrategy -> UnliftStrategy -> Bool)
-> (UnliftStrategy -> UnliftStrategy -> UnliftStrategy)
-> (UnliftStrategy -> UnliftStrategy -> UnliftStrategy)
-> Ord UnliftStrategy
UnliftStrategy -> UnliftStrategy -> Bool
UnliftStrategy -> UnliftStrategy -> Ordering
UnliftStrategy -> UnliftStrategy -> UnliftStrategy
forall a.
Eq a =>
(a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
$ccompare :: UnliftStrategy -> UnliftStrategy -> Ordering
compare :: UnliftStrategy -> UnliftStrategy -> Ordering
$c< :: UnliftStrategy -> UnliftStrategy -> Bool
< :: UnliftStrategy -> UnliftStrategy -> Bool
$c<= :: UnliftStrategy -> UnliftStrategy -> Bool
<= :: UnliftStrategy -> UnliftStrategy -> Bool
$c> :: UnliftStrategy -> UnliftStrategy -> Bool
> :: UnliftStrategy -> UnliftStrategy -> Bool
$c>= :: UnliftStrategy -> UnliftStrategy -> Bool
>= :: UnliftStrategy -> UnliftStrategy -> Bool
$cmax :: UnliftStrategy -> UnliftStrategy -> UnliftStrategy
max :: UnliftStrategy -> UnliftStrategy -> UnliftStrategy
$cmin :: UnliftStrategy -> UnliftStrategy -> UnliftStrategy
min :: UnliftStrategy -> UnliftStrategy -> UnliftStrategy
Ord, Int -> UnliftStrategy -> ShowS
[UnliftStrategy] -> ShowS
UnliftStrategy -> String
(Int -> UnliftStrategy -> ShowS)
-> (UnliftStrategy -> String)
-> ([UnliftStrategy] -> ShowS)
-> Show UnliftStrategy
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> UnliftStrategy -> ShowS
showsPrec :: Int -> UnliftStrategy -> ShowS
$cshow :: UnliftStrategy -> String
show :: UnliftStrategy -> String
$cshowList :: [UnliftStrategy] -> ShowS
showList :: [UnliftStrategy] -> ShowS
Show)

-- | Persistence setting for the 'ConcUnlift' strategy.
--
-- Different functions require different persistence strategies. Examples:
--
-- - Lifting 'pooledMapConcurrentlyN' from the @unliftio@ library requires the
--   'Ephemeral' strategy as we don't want jobs to share environment changes
--   made by previous jobs run in the same worker thread.
--
-- - Lifting 'Control.Concurrent.forkIOWithUnmask' requires the 'Persistent'
--   strategy, otherwise the unmasking function would start with a fresh
--   environment each time it's called.
--
-- Both cases come down to what happens when the unlifting function is called
-- more than once in the same thread. If a thread calls it only once, the
-- 'Persistence' setting makes no observable difference.
--
-- === Example 1
--
-- Consider a thread that modifies thread local state, then inspects it with a
-- second call to the unlifting function:
--
-- >>> import Control.Concurrent
-- >>> import Control.Monad
-- >>> import Effectful
-- >>> import Effectful.State.Dynamic
--
-- >>> :{
--   modifyThenGet :: UnliftStrategy -> IO Int
--   modifyThenGet strategy = runEff . evalStateLocal @Int 0 $ do
--     withEffToIO strategy $ \unlift -> do
--       result <- newEmptyMVar
--       void . forkIO $ do
--         unlift $ modify @Int (+1)
--         putMVar result =<< unlift (get @Int)
--       takeMVar result
-- :}
--
-- With the 'Persistent' strategy the unlifting function keeps the environment
-- between the calls, so the second call sees the modification from the first
-- one:
--
-- >>> modifyThenGet $ ConcUnlift Persistent (Limited 1)
-- 1
--
-- On the other hand, with 'Ephemeral' each call to the unlifting function
-- starts with a fresh copy of the environment, so the modification is silently
-- lost:
--
-- >>> modifyThenGet $ ConcUnlift Ephemeral (Limited 2)
-- 0
--
-- This also showcases the limit meaning different things for the two settings:
-- for the 'Persistent' strategy it limits the number of threads the unlifting
-- can happen in, for 'Ephemeral' it limits the number of calls to the unlifting
-- function.
--
-- === Example 2
--
-- Consider a situation where a single worker thread runs multiple independent
-- jobs:
--
-- >>> :{
--   twoJobs :: UnliftStrategy -> IO [Int]
--   twoJobs strategy = runEff . evalStateLocal @Int 0 $ do
--     withEffToIO strategy $ \unlift -> do
--       result <- newEmptyMVar
--       void . forkIO $ do
--         let job = unlift $ modify @Int (+1) >> get @Int
--         putMVar result =<< sequence [job, job]
--       takeMVar result
-- :}
--
-- With 'Ephemeral' both jobs start from the environment as it was when the
-- unlifting function was created:
--
-- >>> twoJobs $ ConcUnlift Ephemeral Unlimited
-- [1,1]
--
-- With 'Persistent' the second job inherits changes made by the first one, even
-- though the user would most likely expect them to be independent:
--
-- >>> twoJobs $ ConcUnlift Persistent Unlimited
-- [1,2]
data Persistence
  = Ephemeral
  -- ^ Don't persist the environment between calls to the unlifting function in
  -- threads distinct from its creator.
  | Persistent
  -- ^ Persist the environment between calls to the unlifting function within a
  -- particular thread.
  deriving stock (Persistence -> Persistence -> Bool
(Persistence -> Persistence -> Bool)
-> (Persistence -> Persistence -> Bool) -> Eq Persistence
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Persistence -> Persistence -> Bool
== :: Persistence -> Persistence -> Bool
$c/= :: Persistence -> Persistence -> Bool
/= :: Persistence -> Persistence -> Bool
Eq, (forall x. Persistence -> Rep Persistence x)
-> (forall x. Rep Persistence x -> Persistence)
-> Generic Persistence
forall x. Rep Persistence x -> Persistence
forall x. Persistence -> Rep Persistence x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. Persistence -> Rep Persistence x
from :: forall x. Persistence -> Rep Persistence x
$cto :: forall x. Rep Persistence x -> Persistence
to :: forall x. Rep Persistence x -> Persistence
Generic, Eq Persistence
Eq Persistence =>
(Persistence -> Persistence -> Ordering)
-> (Persistence -> Persistence -> Bool)
-> (Persistence -> Persistence -> Bool)
-> (Persistence -> Persistence -> Bool)
-> (Persistence -> Persistence -> Bool)
-> (Persistence -> Persistence -> Persistence)
-> (Persistence -> Persistence -> Persistence)
-> Ord Persistence
Persistence -> Persistence -> Bool
Persistence -> Persistence -> Ordering
Persistence -> Persistence -> Persistence
forall a.
Eq a =>
(a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
$ccompare :: Persistence -> Persistence -> Ordering
compare :: Persistence -> Persistence -> Ordering
$c< :: Persistence -> Persistence -> Bool
< :: Persistence -> Persistence -> Bool
$c<= :: Persistence -> Persistence -> Bool
<= :: Persistence -> Persistence -> Bool
$c> :: Persistence -> Persistence -> Bool
> :: Persistence -> Persistence -> Bool
$c>= :: Persistence -> Persistence -> Bool
>= :: Persistence -> Persistence -> Bool
$cmax :: Persistence -> Persistence -> Persistence
max :: Persistence -> Persistence -> Persistence
$cmin :: Persistence -> Persistence -> Persistence
min :: Persistence -> Persistence -> Persistence
Ord, Int -> Persistence -> ShowS
[Persistence] -> ShowS
Persistence -> String
(Int -> Persistence -> ShowS)
-> (Persistence -> String)
-> ([Persistence] -> ShowS)
-> Show Persistence
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Persistence -> ShowS
showsPrec :: Int -> Persistence -> ShowS
$cshow :: Persistence -> String
show :: Persistence -> String
$cshowList :: [Persistence] -> ShowS
showList :: [Persistence] -> ShowS
Show)

-- | Limit setting for the 'ConcUnlift' strategy.
data Limit
  = Limited !Int
  -- ^ Behavior dependent on the 'Persistence' setting.
  --
  -- For 'Ephemeral', it limits the amount of uses of the unlifting function in
  -- threads distinct from its creator to @N@. The unlifting function will
  -- create @N@ copies of the environment when called @N@ times and @K+1@ copies
  -- when called @K < N@ times.
  --
  -- For 'Persistent', it limits the amount of threads, distinct from the
  -- creator of the unlifting function, it can be called in to @N@. The amount
  -- of calls to the unlifting function within a particular threads is
  -- unlimited. The unlifting function will create @N@ copies of the environment
  -- when called in @N@ threads and @K+1@ copies when called in @K < N@ threads.
  | Unlimited
  -- ^ Unlimited use of the unlifting function.
  deriving stock (Limit -> Limit -> Bool
(Limit -> Limit -> Bool) -> (Limit -> Limit -> Bool) -> Eq Limit
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Limit -> Limit -> Bool
== :: Limit -> Limit -> Bool
$c/= :: Limit -> Limit -> Bool
/= :: Limit -> Limit -> Bool
Eq, (forall x. Limit -> Rep Limit x)
-> (forall x. Rep Limit x -> Limit) -> Generic Limit
forall x. Rep Limit x -> Limit
forall x. Limit -> Rep Limit x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. Limit -> Rep Limit x
from :: forall x. Limit -> Rep Limit x
$cto :: forall x. Rep Limit x -> Limit
to :: forall x. Rep Limit x -> Limit
Generic, Eq Limit
Eq Limit =>
(Limit -> Limit -> Ordering)
-> (Limit -> Limit -> Bool)
-> (Limit -> Limit -> Bool)
-> (Limit -> Limit -> Bool)
-> (Limit -> Limit -> Bool)
-> (Limit -> Limit -> Limit)
-> (Limit -> Limit -> Limit)
-> Ord Limit
Limit -> Limit -> Bool
Limit -> Limit -> Ordering
Limit -> Limit -> Limit
forall a.
Eq a =>
(a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
$ccompare :: Limit -> Limit -> Ordering
compare :: Limit -> Limit -> Ordering
$c< :: Limit -> Limit -> Bool
< :: Limit -> Limit -> Bool
$c<= :: Limit -> Limit -> Bool
<= :: Limit -> Limit -> Bool
$c> :: Limit -> Limit -> Bool
> :: Limit -> Limit -> Bool
$c>= :: Limit -> Limit -> Bool
>= :: Limit -> Limit -> Bool
$cmax :: Limit -> Limit -> Limit
max :: Limit -> Limit -> Limit
$cmin :: Limit -> Limit -> Limit
min :: Limit -> Limit -> Limit
Ord, Int -> Limit -> ShowS
[Limit] -> ShowS
Limit -> String
(Int -> Limit -> ShowS)
-> (Limit -> String) -> ([Limit] -> ShowS) -> Show Limit
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Limit -> ShowS
showsPrec :: Int -> Limit -> ShowS
$cshow :: Limit -> String
show :: Limit -> String
$cshowList :: [Limit] -> ShowS
showList :: [Limit] -> ShowS
Show)

----------------------------------------
-- Unlift functions

-- | Concurrent unlift with limited uses that doesn't preserve the environment
-- between calls to the unlifting function in threads other than its creator.
--
-- @since 2.7.0.0
ephemeralConcLimitedUnlift
  :: (HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r))
  => Env es
  -> Int
  -- ^ Number of permitted uses of the unlift function.
  -> ((forall r. effEs r -> IO r) -> IO a)
  -> IO a
ephemeralConcLimitedUnlift :: forall (effEs :: Type -> Type) (es :: [Effect]) a.
(HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r)) =>
Env es -> Int -> ((forall r. effEs r -> IO r) -> IO a) -> IO a
ephemeralConcLimitedUnlift Env es
es0 Int
uses (forall r. effEs r -> IO r) -> IO a
k = do
  Bool -> IO () -> IO ()
forall (f :: Type -> Type). Applicative f => Bool -> f () -> f ()
unless (Int
uses Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    String -> IO ()
forall a. HasCallStack => String -> a
error (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"Invalid number of uses: " String -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show Int
uses
  ThreadId
tid0 <- IO ThreadId
myThreadId
  -- Create a copy of the environment as a template for the other threads to
  -- use. This can't be done from inside the callback as the environment might
  -- have already changed by then.
  Env es
esTemplate <- Env es -> IO (Env es)
forall (es :: [Effect]). HasCallStack => Env es -> IO (Env es)
cloneEnv Env es
es0
  MVar Int
mvUses <- Int -> IO (MVar Int)
forall a. a -> IO (MVar a)
S.newMVar Int
uses
  let getEs :: IO (Env es)
getEs = IO ThreadId
myThreadId IO ThreadId -> (ThreadId -> IO (Env es)) -> IO (Env es)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        ThreadId
tid | ThreadId
tid0 ThreadId -> ThreadId -> Bool
forall a. Eq a => a -> a -> Bool
== ThreadId
tid -> Env es -> IO (Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure Env es
es0
        ThreadId
_ -> MVar Int -> (Int -> IO (Int, Env es)) -> IO (Env es)
forall a b. MVar a -> (a -> IO (a, b)) -> IO b
S.modifyMVar MVar Int
mvUses ((Int -> IO (Int, Env es)) -> IO (Env es))
-> (Int -> IO (Int, Env es)) -> IO (Env es)
forall a b. (a -> b) -> a -> b
$ \case
          Int
0 -> String -> IO (Int, Env es)
forall a. HasCallStack => String -> a
error
             (String -> IO (Int, Env es)) -> String -> IO (Int, Env es)
forall a b. (a -> b) -> a -> b
$ String
"Number of permitted calls (" String -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show Int
uses String -> ShowS
forall a. [a] -> [a] -> [a]
++ String
") to the unlifting "
            String -> ShowS
forall a. [a] -> [a] -> [a]
++ String
"function in other threads was exceeded. Please increase the limit "
            String -> ShowS
forall a. [a] -> [a] -> [a]
++ String
"or use the unlimited variant."
          Int
1 -> (Int, Env es) -> IO (Int, Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (Int
0, Env es
esTemplate)
          Int
n -> do
            Env es
es <- Env es -> IO (Env es)
forall (es :: [Effect]). HasCallStack => Env es -> IO (Env es)
cloneEnv Env es
esTemplate
            (Int, Env es) -> IO (Int, Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1, Env es
es)
  (forall r. effEs r -> IO r) -> IO a
k ((forall r. effEs r -> IO r) -> IO a)
-> (forall r. effEs r -> IO r) -> IO a
forall a b. (a -> b) -> a -> b
$ \effEs r
action -> effEs r -> Env es -> IO r
forall a b. Coercible a b => a -> b
coerce effEs r
action (Env es -> IO r) -> IO (Env es) -> IO r
forall (m :: Type -> Type) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO (Env es)
getEs
{-# INLINE ephemeralConcLimitedUnlift #-}

-- | Concurrent unlift with unlimited uses that doesn't preserve the environment
-- between calls to the unlifting function in threads other than its creator.
--
-- @since 2.7.0.0
ephemeralConcUnlimitedUnlift
  :: (HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r))
  => Env es
  -> ((forall r. effEs r -> IO r) -> IO a)
  -> IO a
ephemeralConcUnlimitedUnlift :: forall (effEs :: Type -> Type) (es :: [Effect]) a.
(HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r)) =>
Env es -> ((forall r. effEs r -> IO r) -> IO a) -> IO a
ephemeralConcUnlimitedUnlift Env es
es0 (forall r. effEs r -> IO r) -> IO a
k = do
  ThreadId
tid0 <- IO ThreadId
myThreadId
  -- Create a copy of the environment as a template for the other threads to
  -- use. This can't be done from inside the callback as the environment might
  -- have already changed by then.
  Env es
esTemplate <- Env es -> IO (Env es)
forall (es :: [Effect]). HasCallStack => Env es -> IO (Env es)
cloneEnv Env es
es0
  let getEs :: IO (Env es)
getEs = IO ThreadId
myThreadId IO ThreadId -> (ThreadId -> IO (Env es)) -> IO (Env es)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        ThreadId
tid | ThreadId
tid0 ThreadId -> ThreadId -> Bool
forall a. Eq a => a -> a -> Bool
== ThreadId
tid -> Env es -> IO (Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure Env es
es0
        ThreadId
_ -> Env es -> IO (Env es)
forall (es :: [Effect]). HasCallStack => Env es -> IO (Env es)
cloneEnv Env es
esTemplate
  (forall r. effEs r -> IO r) -> IO a
k ((forall r. effEs r -> IO r) -> IO a)
-> (forall r. effEs r -> IO r) -> IO a
forall a b. (a -> b) -> a -> b
$ \effEs r
action -> effEs r -> Env es -> IO r
forall a b. Coercible a b => a -> b
coerce effEs r
action (Env es -> IO r) -> IO (Env es) -> IO r
forall (m :: Type -> Type) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO (Env es)
getEs
{-# INLINE ephemeralConcUnlimitedUnlift #-}

-- | Concurrent unlift that preserves the environment between calls to the
-- unlifting function within a particular thread.
persistentConcUnlift
  :: (HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r))
  => Env es
  -> Bool
  -> Int
  -- ^ Number of threads that are allowed to use the unlift function.
  -> ((forall r. effEs r -> IO r) -> IO a)
  -> IO a
persistentConcUnlift :: forall (effEs :: Type -> Type) (es :: [Effect]) a.
(HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r)) =>
Env es
-> Bool -> Int -> ((forall r. effEs r -> IO r) -> IO a) -> IO a
persistentConcUnlift Env es
es0 Bool
cleanUp Int
threads (forall r. effEs r -> IO r) -> IO a
k = do
  Bool -> IO () -> IO ()
forall (f :: Type -> Type). Applicative f => Bool -> f () -> f ()
unless (Int
threads Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    String -> IO ()
forall a. HasCallStack => String -> a
error (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"Invalid number of threads: " String -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show Int
threads
  ThreadId
tid0 <- IO ThreadId
myThreadId
  -- Create a copy of the environment as a template for the other threads to
  -- use. This can't be done from inside the callback as the environment might
  -- have already changed by then.
  Env es
esTemplate <- Env es -> IO (Env es)
forall (es :: [Effect]). HasCallStack => Env es -> IO (Env es)
cloneEnv Env es
es0
  MVar (ThreadEntries (Env es))
mvEntries <- ThreadEntries (Env es) -> IO (MVar (ThreadEntries (Env es)))
forall a. a -> IO (MVar a)
S.newMVar (ThreadEntries (Env es) -> IO (MVar (ThreadEntries (Env es))))
-> ThreadEntries (Env es) -> IO (MVar (ThreadEntries (Env es)))
forall a b. (a -> b) -> a -> b
$ Int -> Word64Map (Weak (Env es)) -> ThreadEntries (Env es)
forall a. Int -> Word64Map (Weak a) -> ThreadEntries a
ThreadEntries Int
threads Word64Map (Weak (Env es))
forall a. Word64Map a
M.empty
  let getEs :: IO (Env es)
getEs = IO ThreadId
myThreadId IO ThreadId -> (ThreadId -> IO (Env es)) -> IO (Env es)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        ThreadId
tid | ThreadId
tid0 ThreadId -> ThreadId -> Bool
forall a. Eq a => a -> a -> Bool
== ThreadId
tid -> Env es -> IO (Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure Env es
es0
        ThreadId
tid -> do
          ThreadEntries (Env es)
te0 <- MVar (ThreadEntries (Env es)) -> IO (ThreadEntries (Env es))
forall a. MVar a -> IO a
S.readMVar MVar (ThreadEntries (Env es))
mvEntries
          let wkTid :: Word64
wkTid = ThreadId -> Word64
weakThreadId ThreadId
tid
          case Word64
wkTid Word64 -> Word64Map (Weak (Env es)) -> Maybe (Weak (Env es))
forall a. Word64 -> Word64Map a -> Maybe a
`M.lookup` ThreadEntries (Env es)
te0.entries of
            Just Weak (Env es)
wkEs -> Weak (Env es) -> IO (Env es)
forall a. HasCallStack => Weak a -> IO a
getWkTidEnv Weak (Env es)
wkEs
            -- If the environment is not in the map, there is no point checking
            -- again within modifyMVar below, because this is the only thread
            -- that can put it there.
            Maybe (Weak (Env es))
Nothing -> MVar (ThreadEntries (Env es))
-> (ThreadEntries (Env es) -> IO (ThreadEntries (Env es), Env es))
-> IO (Env es)
forall a b. MVar a -> (a -> IO (a, b)) -> IO b
S.modifyMVar MVar (ThreadEntries (Env es))
mvEntries ((ThreadEntries (Env es) -> IO (ThreadEntries (Env es), Env es))
 -> IO (Env es))
-> (ThreadEntries (Env es) -> IO (ThreadEntries (Env es), Env es))
-> IO (Env es)
forall a b. (a -> b) -> a -> b
$ \ThreadEntries (Env es)
te -> case ThreadEntries (Env es)
te.capacity of
              Int
0 -> Int -> IO (ThreadEntries (Env es), Env es)
forall a. HasCallStack => Int -> a
noCapacityError Int
threads
              Int
1 -> do
                Weak (Env es)
wkTidEs <- ThreadId
-> Word64
-> Env es
-> MVar (ThreadEntries (Env es))
-> Bool
-> IO (Weak (Env es))
forall a.
ThreadId
-> Word64 -> a -> MVar (ThreadEntries a) -> Bool -> IO (Weak a)
mkWeakThreadIdEnv ThreadId
tid Word64
wkTid Env es
esTemplate MVar (ThreadEntries (Env es))
mvEntries Bool
cleanUp
                let newEntries :: ThreadEntries (Env es)
newEntries = ThreadEntries
                      { $sel:capacity:ThreadEntries :: Int
capacity = ThreadEntries (Env es)
te.capacity Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1
                      , $sel:entries:ThreadEntries :: Word64Map (Weak (Env es))
entries  = Word64
-> Weak (Env es)
-> Word64Map (Weak (Env es))
-> Word64Map (Weak (Env es))
forall a. Word64 -> a -> Word64Map a -> Word64Map a
M.insert Word64
wkTid Weak (Env es)
wkTidEs ThreadEntries (Env es)
te.entries
                      }
                (ThreadEntries (Env es), Env es)
-> IO (ThreadEntries (Env es), Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (ThreadEntries (Env es)
newEntries, Env es
esTemplate)
              Int
_ -> do
                Env es
es <- Env es -> IO (Env es)
forall (es :: [Effect]). HasCallStack => Env es -> IO (Env es)
cloneEnv Env es
esTemplate
                Weak (Env es)
wkTidEs <- ThreadId
-> Word64
-> Env es
-> MVar (ThreadEntries (Env es))
-> Bool
-> IO (Weak (Env es))
forall a.
ThreadId
-> Word64 -> a -> MVar (ThreadEntries a) -> Bool -> IO (Weak a)
mkWeakThreadIdEnv ThreadId
tid Word64
wkTid Env es
es MVar (ThreadEntries (Env es))
mvEntries Bool
cleanUp
                let newEntries :: ThreadEntries (Env es)
newEntries = ThreadEntries
                      { $sel:capacity:ThreadEntries :: Int
capacity = ThreadEntries (Env es)
te.capacity Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1
                      , $sel:entries:ThreadEntries :: Word64Map (Weak (Env es))
entries  = Word64
-> Weak (Env es)
-> Word64Map (Weak (Env es))
-> Word64Map (Weak (Env es))
forall a. Word64 -> a -> Word64Map a -> Word64Map a
M.insert Word64
wkTid Weak (Env es)
wkTidEs ThreadEntries (Env es)
te.entries
                      }
                (ThreadEntries (Env es), Env es)
-> IO (ThreadEntries (Env es), Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (ThreadEntries (Env es)
newEntries, Env es
es)
  (forall r. effEs r -> IO r) -> IO a
k ((forall r. effEs r -> IO r) -> IO a)
-> (forall r. effEs r -> IO r) -> IO a
forall a b. (a -> b) -> a -> b
$ \effEs r
action -> effEs r -> Env es -> IO r
forall a b. Coercible a b => a -> b
coerce effEs r
action (Env es -> IO r) -> IO (Env es) -> IO r
forall (m :: Type -> Type) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO (Env es)
getEs
{-# INLINE persistentConcUnlift #-}

-- | Variant of 'persistentConcUnlift' for a single other thread that doesn't
-- need ThreadEntries.
--
-- @since 2.7.0.0
persistentConcSingleUnlift
  :: ( HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r))
  => Env es
  -> ((forall r. effEs r -> IO r) -> IO a)
  -> IO a
persistentConcSingleUnlift :: forall (effEs :: Type -> Type) (es :: [Effect]) a.
(HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r)) =>
Env es -> ((forall r. effEs r -> IO r) -> IO a) -> IO a
persistentConcSingleUnlift Env es
es0 (forall r. effEs r -> IO r) -> IO a
k = do
  ThreadId
tid0 <- IO ThreadId
myThreadId
  -- Create a copy of the environment for the other thread to use. This can't be
  -- done from inside the callback as the environment might have already changed
  -- by then.
  Env es
es <- Env es -> IO (Env es)
forall (es :: [Effect]). HasCallStack => Env es -> IO (Env es)
cloneEnv Env es
es0
  -- GHC never labels threads as 0.
  MVar Word64
mvWeakTid <- Word64 -> IO (MVar Word64)
forall a. a -> IO (MVar a)
S.newMVar Word64
0
  let getEs :: IO (Env es)
getEs = IO ThreadId
myThreadId IO ThreadId -> (ThreadId -> IO (Env es)) -> IO (Env es)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        ThreadId
tid | ThreadId
tid0 ThreadId -> ThreadId -> Bool
forall a. Eq a => a -> a -> Bool
== ThreadId
tid -> Env es -> IO (Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure Env es
es0
        ThreadId
tid -> do
          let wkTid :: Word64
wkTid = ThreadId -> Word64
weakThreadId ThreadId
tid
          MVar Word64 -> IO Word64
forall a. MVar a -> IO a
S.readMVar MVar Word64
mvWeakTid IO Word64 -> (Word64 -> IO (Env es)) -> IO (Env es)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
            Word64
0 -> MVar Word64 -> (Word64 -> IO (Word64, Env es)) -> IO (Env es)
forall a b. MVar a -> (a -> IO (a, b)) -> IO b
S.modifyMVar MVar Word64
mvWeakTid ((Word64 -> IO (Word64, Env es)) -> IO (Env es))
-> (Word64 -> IO (Word64, Env es)) -> IO (Env es)
forall a b. (a -> b) -> a -> b
$ \case
              Word64
0 -> (Word64, Env es) -> IO (Word64, Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (Word64
wkTid, Env es
es)
              Word64
_ -> Int -> IO (Word64, Env es)
forall a. HasCallStack => Int -> a
noCapacityError Int
1
            Word64
v | Word64
v Word64 -> Word64 -> Bool
forall a. Eq a => a -> a -> Bool
== Word64
wkTid -> Env es -> IO (Env es)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure Env es
es
              | Bool
otherwise -> Int -> IO (Env es)
forall a. HasCallStack => Int -> a
noCapacityError Int
1
  (forall r. effEs r -> IO r) -> IO a
k ((forall r. effEs r -> IO r) -> IO a)
-> (forall r. effEs r -> IO r) -> IO a
forall a b. (a -> b) -> a -> b
$ \effEs r
action -> effEs r -> Env es -> IO r
forall a b. Coercible a b => a -> b
coerce effEs r
action (Env es -> IO r) -> IO (Env es) -> IO r
forall (m :: Type -> Type) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO (Env es)
getEs
{-# INLINE persistentConcSingleUnlift #-}

-- | Variant of 'persistentConcUnlift' producing two unlifting functions that
-- share the effect storage in each thread.
--
-- @since 2.7.0.0
persistentConcUnlifts
  :: ( HasCallStack
     , forall r. Coercible (effEs r) (Env es -> IO r)
     , forall r. Coercible (effLocalEs r) (Env localEs -> IO r)
     )
  => Env es
  -> Env localEs
  -> Bool
  -> Int
  -- ^ Number of threads that are allowed to use the unlift function.
  -> ((forall r. effEs r -> IO r) -> (forall r. effLocalEs r -> IO r) -> IO a)
  -> IO a
persistentConcUnlifts :: forall (effEs :: Type -> Type) (es :: [Effect])
       (effLocalEs :: Type -> Type) (localEs :: [Effect]) a.
(HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r),
 forall r. Coercible (effLocalEs r) (Env localEs -> IO r)) =>
Env es
-> Env localEs
-> Bool
-> Int
-> ((forall r. effEs r -> IO r)
    -> (forall r. effLocalEs r -> IO r) -> IO a)
-> IO a
persistentConcUnlifts Env es
es0 Env localEs
les0 Bool
cleanUp Int
threads (forall r. effEs r -> IO r)
-> (forall r. effLocalEs r -> IO r) -> IO a
k = do
  Bool -> IO () -> IO ()
forall (f :: Type -> Type). Applicative f => Bool -> f () -> f ()
unless (Int
threads Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    String -> IO ()
forall a. HasCallStack => String -> a
error (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"Invalid number of threads: " String -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show Int
threads
  ThreadId
tid0 <- IO ThreadId
myThreadId
  -- Create a copy of the environments sharing the effect storage as a template
  -- for the other threads to use. This can't be done from inside the callback
  -- as the environment might have already changed by then.
  IORef Storage
storageTemplate <- HasCallStack => IORef Storage -> IO (IORef Storage)
IORef Storage -> IO (IORef Storage)
cloneStorage Env es
es0.storage
  Env es
esTemplate <- Env es -> IORef Storage -> IO (Env es)
forall (es :: [Effect]). Env es -> IORef Storage -> IO (Env es)
replaceStorage Env es
es0 IORef Storage
storageTemplate
  Env localEs
lesTemplate <- Env localEs -> IORef Storage -> IO (Env localEs)
forall (es :: [Effect]). Env es -> IORef Storage -> IO (Env es)
replaceStorage Env localEs
les0 IORef Storage
storageTemplate
  MVar (ThreadEntries (Env es, Env localEs))
mvEntries <- ThreadEntries (Env es, Env localEs)
-> IO (MVar (ThreadEntries (Env es, Env localEs)))
forall a. a -> IO (MVar a)
S.newMVar (ThreadEntries (Env es, Env localEs)
 -> IO (MVar (ThreadEntries (Env es, Env localEs))))
-> ThreadEntries (Env es, Env localEs)
-> IO (MVar (ThreadEntries (Env es, Env localEs)))
forall a b. (a -> b) -> a -> b
$ Int
-> Word64Map (Weak (Env es, Env localEs))
-> ThreadEntries (Env es, Env localEs)
forall a. Int -> Word64Map (Weak a) -> ThreadEntries a
ThreadEntries Int
threads Word64Map (Weak (Env es, Env localEs))
forall a. Word64Map a
M.empty
  let getEsLes :: IO (Env es, Env localEs)
getEsLes = IO ThreadId
myThreadId IO ThreadId
-> (ThreadId -> IO (Env es, Env localEs))
-> IO (Env es, Env localEs)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        ThreadId
tid | ThreadId
tid0 ThreadId -> ThreadId -> Bool
forall a. Eq a => a -> a -> Bool
== ThreadId
tid -> (Env es, Env localEs) -> IO (Env es, Env localEs)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (Env es
es0, Env localEs
les0)
        ThreadId
tid -> do
          ThreadEntries (Env es, Env localEs)
te0 <- MVar (ThreadEntries (Env es, Env localEs))
-> IO (ThreadEntries (Env es, Env localEs))
forall a. MVar a -> IO a
S.readMVar MVar (ThreadEntries (Env es, Env localEs))
mvEntries
          let wkTid :: Word64
wkTid = ThreadId -> Word64
weakThreadId ThreadId
tid
          case Word64
wkTid Word64
-> Word64Map (Weak (Env es, Env localEs))
-> Maybe (Weak (Env es, Env localEs))
forall a. Word64 -> Word64Map a -> Maybe a
`M.lookup` ThreadEntries (Env es, Env localEs)
te0.entries of
            Just Weak (Env es, Env localEs)
wkEsLes -> Weak (Env es, Env localEs) -> IO (Env es, Env localEs)
forall a. HasCallStack => Weak a -> IO a
getWkTidEnv Weak (Env es, Env localEs)
wkEsLes
            -- If the environments are not in the map, there is no point
            -- checking again within modifyMVar below, because this is the only
            -- thread that can put them there.
            Maybe (Weak (Env es, Env localEs))
Nothing -> MVar (ThreadEntries (Env es, Env localEs))
-> (ThreadEntries (Env es, Env localEs)
    -> IO (ThreadEntries (Env es, Env localEs), (Env es, Env localEs)))
-> IO (Env es, Env localEs)
forall a b. MVar a -> (a -> IO (a, b)) -> IO b
S.modifyMVar MVar (ThreadEntries (Env es, Env localEs))
mvEntries ((ThreadEntries (Env es, Env localEs)
  -> IO (ThreadEntries (Env es, Env localEs), (Env es, Env localEs)))
 -> IO (Env es, Env localEs))
-> (ThreadEntries (Env es, Env localEs)
    -> IO (ThreadEntries (Env es, Env localEs), (Env es, Env localEs)))
-> IO (Env es, Env localEs)
forall a b. (a -> b) -> a -> b
$ \ThreadEntries (Env es, Env localEs)
te -> case ThreadEntries (Env es, Env localEs)
te.capacity of
              Int
0 -> Int
-> IO (ThreadEntries (Env es, Env localEs), (Env es, Env localEs))
forall a. HasCallStack => Int -> a
noCapacityError Int
threads
              Int
1 -> do
                Weak (Env es, Env localEs)
wkTidEsLes <- ThreadId
-> Word64
-> (Env es, Env localEs)
-> MVar (ThreadEntries (Env es, Env localEs))
-> Bool
-> IO (Weak (Env es, Env localEs))
forall a.
ThreadId
-> Word64 -> a -> MVar (ThreadEntries a) -> Bool -> IO (Weak a)
mkWeakThreadIdEnv ThreadId
tid Word64
wkTid (Env es
esTemplate, Env localEs
lesTemplate) MVar (ThreadEntries (Env es, Env localEs))
mvEntries Bool
cleanUp
                let newEntries :: ThreadEntries (Env es, Env localEs)
newEntries = ThreadEntries
                      { $sel:capacity:ThreadEntries :: Int
capacity = ThreadEntries (Env es, Env localEs)
te.capacity Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1
                      , $sel:entries:ThreadEntries :: Word64Map (Weak (Env es, Env localEs))
entries  = Word64
-> Weak (Env es, Env localEs)
-> Word64Map (Weak (Env es, Env localEs))
-> Word64Map (Weak (Env es, Env localEs))
forall a. Word64 -> a -> Word64Map a -> Word64Map a
M.insert Word64
wkTid Weak (Env es, Env localEs)
wkTidEsLes ThreadEntries (Env es, Env localEs)
te.entries
                      }
                (ThreadEntries (Env es, Env localEs), (Env es, Env localEs))
-> IO (ThreadEntries (Env es, Env localEs), (Env es, Env localEs))
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (ThreadEntries (Env es, Env localEs)
newEntries, (Env es
esTemplate, Env localEs
lesTemplate))
              Int
_ -> do
                IORef Storage
storage <- HasCallStack => IORef Storage -> IO (IORef Storage)
IORef Storage -> IO (IORef Storage)
cloneStorage IORef Storage
storageTemplate
                Env es
es <- Env es -> IORef Storage -> IO (Env es)
forall (es :: [Effect]). Env es -> IORef Storage -> IO (Env es)
replaceStorage Env es
esTemplate IORef Storage
storage
                Env localEs
les <- Env localEs -> IORef Storage -> IO (Env localEs)
forall (es :: [Effect]). Env es -> IORef Storage -> IO (Env es)
replaceStorage Env localEs
lesTemplate IORef Storage
storage
                Weak (Env es, Env localEs)
wkTidEsLes <- ThreadId
-> Word64
-> (Env es, Env localEs)
-> MVar (ThreadEntries (Env es, Env localEs))
-> Bool
-> IO (Weak (Env es, Env localEs))
forall a.
ThreadId
-> Word64 -> a -> MVar (ThreadEntries a) -> Bool -> IO (Weak a)
mkWeakThreadIdEnv ThreadId
tid Word64
wkTid (Env es
es, Env localEs
les) MVar (ThreadEntries (Env es, Env localEs))
mvEntries Bool
cleanUp
                let newEntries :: ThreadEntries (Env es, Env localEs)
newEntries = ThreadEntries
                      { $sel:capacity:ThreadEntries :: Int
capacity = ThreadEntries (Env es, Env localEs)
te.capacity Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1
                      , $sel:entries:ThreadEntries :: Word64Map (Weak (Env es, Env localEs))
entries  = Word64
-> Weak (Env es, Env localEs)
-> Word64Map (Weak (Env es, Env localEs))
-> Word64Map (Weak (Env es, Env localEs))
forall a. Word64 -> a -> Word64Map a -> Word64Map a
M.insert Word64
wkTid Weak (Env es, Env localEs)
wkTidEsLes ThreadEntries (Env es, Env localEs)
te.entries
                      }
                (ThreadEntries (Env es, Env localEs), (Env es, Env localEs))
-> IO (ThreadEntries (Env es, Env localEs), (Env es, Env localEs))
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (ThreadEntries (Env es, Env localEs)
newEntries, (Env es
es, Env localEs
les))
  (forall r. effEs r -> IO r)
-> (forall r. effLocalEs r -> IO r) -> IO a
k (\effEs r
action -> effEs r -> Env es -> IO r
forall a b. Coercible a b => a -> b
coerce effEs r
action (Env es -> IO r)
-> ((Env es, Env localEs) -> Env es)
-> (Env es, Env localEs)
-> IO r
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Env es, Env localEs) -> Env es
forall a b. (a, b) -> a
fst ((Env es, Env localEs) -> IO r) -> IO (Env es, Env localEs) -> IO r
forall (m :: Type -> Type) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO (Env es, Env localEs)
getEsLes)
    (\effLocalEs r
action -> effLocalEs r -> Env localEs -> IO r
forall a b. Coercible a b => a -> b
coerce effLocalEs r
action (Env localEs -> IO r)
-> ((Env es, Env localEs) -> Env localEs)
-> (Env es, Env localEs)
-> IO r
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Env es, Env localEs) -> Env localEs
forall a b. (a, b) -> b
snd ((Env es, Env localEs) -> IO r) -> IO (Env es, Env localEs) -> IO r
forall (m :: Type -> Type) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO (Env es, Env localEs)
getEsLes)
{-# INLINE persistentConcUnlifts #-}

-- | Variant of 'persistentConcUnlifts' for a single other thread that doesn't
-- need ThreadEntries.
--
-- @since 2.7.0.0
persistentConcSingleUnlifts
  :: ( HasCallStack
     , forall r. Coercible (effEs r) (Env es -> IO r)
     , forall r. Coercible (effLocalEs r) (Env localEs -> IO r)
     )
  => Env es
  -> Env localEs
  -> ((forall r. effEs r -> IO r) -> (forall r. effLocalEs r -> IO r) -> IO a)
  -> IO a
persistentConcSingleUnlifts :: forall (effEs :: Type -> Type) (es :: [Effect])
       (effLocalEs :: Type -> Type) (localEs :: [Effect]) a.
(HasCallStack, forall r. Coercible (effEs r) (Env es -> IO r),
 forall r. Coercible (effLocalEs r) (Env localEs -> IO r)) =>
Env es
-> Env localEs
-> ((forall r. effEs r -> IO r)
    -> (forall r. effLocalEs r -> IO r) -> IO a)
-> IO a
persistentConcSingleUnlifts Env es
es0 Env localEs
les0 (forall r. effEs r -> IO r)
-> (forall r. effLocalEs r -> IO r) -> IO a
k = do
  ThreadId
tid0 <- IO ThreadId
myThreadId
  -- Create a copy of the environments sharing the effect storage for the other
  -- thread to use. This can't be done from inside the callback as the
  -- environment might have already changed by then.
  IORef Storage
storage <- HasCallStack => IORef Storage -> IO (IORef Storage)
IORef Storage -> IO (IORef Storage)
cloneStorage Env es
es0.storage
  Env es
es <- Env es -> IORef Storage -> IO (Env es)
forall (es :: [Effect]). Env es -> IORef Storage -> IO (Env es)
replaceStorage Env es
es0 IORef Storage
storage
  Env localEs
les <- Env localEs -> IORef Storage -> IO (Env localEs)
forall (es :: [Effect]). Env es -> IORef Storage -> IO (Env es)
replaceStorage Env localEs
les0 IORef Storage
storage
  -- GHC never labels threads as 0.
  MVar Word64
mvWeakTid <- Word64 -> IO (MVar Word64)
forall a. a -> IO (MVar a)
S.newMVar Word64
0
  let getEsLes :: IO (Env es, Env localEs)
getEsLes = IO ThreadId
myThreadId IO ThreadId
-> (ThreadId -> IO (Env es, Env localEs))
-> IO (Env es, Env localEs)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        ThreadId
tid | ThreadId
tid0 ThreadId -> ThreadId -> Bool
forall a. Eq a => a -> a -> Bool
== ThreadId
tid -> (Env es, Env localEs) -> IO (Env es, Env localEs)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (Env es
es0, Env localEs
les0)
        ThreadId
tid -> do
          let wkTid :: Word64
wkTid = ThreadId -> Word64
weakThreadId ThreadId
tid
          MVar Word64 -> IO Word64
forall a. MVar a -> IO a
S.readMVar MVar Word64
mvWeakTid IO Word64
-> (Word64 -> IO (Env es, Env localEs)) -> IO (Env es, Env localEs)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
            Word64
0 -> MVar Word64
-> (Word64 -> IO (Word64, (Env es, Env localEs)))
-> IO (Env es, Env localEs)
forall a b. MVar a -> (a -> IO (a, b)) -> IO b
S.modifyMVar MVar Word64
mvWeakTid ((Word64 -> IO (Word64, (Env es, Env localEs)))
 -> IO (Env es, Env localEs))
-> (Word64 -> IO (Word64, (Env es, Env localEs)))
-> IO (Env es, Env localEs)
forall a b. (a -> b) -> a -> b
$ \case
              Word64
0 -> (Word64, (Env es, Env localEs))
-> IO (Word64, (Env es, Env localEs))
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (Word64
wkTid, (Env es
es, Env localEs
les))
              Word64
_ -> Int -> IO (Word64, (Env es, Env localEs))
forall a. HasCallStack => Int -> a
noCapacityError Int
1
            Word64
v | Word64
v Word64 -> Word64 -> Bool
forall a. Eq a => a -> a -> Bool
== Word64
wkTid -> (Env es, Env localEs) -> IO (Env es, Env localEs)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (Env es
es, Env localEs
les)
              | Bool
otherwise -> Int -> IO (Env es, Env localEs)
forall a. HasCallStack => Int -> a
noCapacityError Int
1
  (forall r. effEs r -> IO r)
-> (forall r. effLocalEs r -> IO r) -> IO a
k (\effEs r
action -> effEs r -> Env es -> IO r
forall a b. Coercible a b => a -> b
coerce effEs r
action (Env es -> IO r)
-> ((Env es, Env localEs) -> Env es)
-> (Env es, Env localEs)
-> IO r
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Env es, Env localEs) -> Env es
forall a b. (a, b) -> a
fst ((Env es, Env localEs) -> IO r) -> IO (Env es, Env localEs) -> IO r
forall (m :: Type -> Type) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO (Env es, Env localEs)
getEsLes)
    (\effLocalEs r
action -> effLocalEs r -> Env localEs -> IO r
forall a b. Coercible a b => a -> b
coerce effLocalEs r
action (Env localEs -> IO r)
-> ((Env es, Env localEs) -> Env localEs)
-> (Env es, Env localEs)
-> IO r
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Env es, Env localEs) -> Env localEs
forall a b. (a, b) -> b
snd ((Env es, Env localEs) -> IO r) -> IO (Env es, Env localEs) -> IO r
forall (m :: Type -> Type) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO (Env es, Env localEs)
getEsLes)
{-# INLINE persistentConcSingleUnlifts #-}

----------------------------------------
-- Internal helpers

noCapacityError :: HasCallStack => Int -> a
noCapacityError :: forall a. HasCallStack => Int -> a
noCapacityError Int
threads = String -> a
forall a. HasCallStack => String -> a
error
  (String -> a) -> String -> a
forall a b. (a -> b) -> a -> b
$ String
"Number of other threads (" String -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show Int
threads String -> ShowS
forall a. [a] -> [a] -> [a]
++ String
") permitted to "
  String -> ShowS
forall a. [a] -> [a] -> [a]
++ String
"use the unlifting function was exceeded. Please increase the "
  String -> ShowS
forall a. [a] -> [a] -> [a]
++ String
"limit or use the unlimited variant."

getWkTidEnv :: HasCallStack => Weak a -> IO a
getWkTidEnv :: forall a. HasCallStack => Weak a -> IO a
getWkTidEnv Weak a
wkTidEnv = Weak a -> IO (Maybe a)
forall v. Weak v -> IO (Maybe v)
deRefWeak Weak a
wkTidEnv IO (Maybe a) -> (Maybe a -> IO a) -> IO a
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: Type -> Type) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
  Maybe a
Nothing -> String -> IO a
forall a. HasCallStack => String -> a
error String
"Impossible, thread alive but its weak ref dead"
  Just a
env -> a -> IO a
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure a
env

data ThreadEntries a = ThreadEntries
  { forall a. ThreadEntries a -> Int
capacity :: !Int
  , forall a. ThreadEntries a -> Word64Map (Weak a)
entries  :: !(M.Word64Map (Weak a))
  }

mkWeakThreadIdEnv
  :: ThreadId
  -> Word64
  -> a
  -> S.MVar (ThreadEntries a)
  -> Bool
  -> IO (Weak a)
mkWeakThreadIdEnv :: forall a.
ThreadId
-> Word64 -> a -> MVar (ThreadEntries a) -> Bool -> IO (Weak a)
mkWeakThreadIdEnv (ThreadId ThreadId#
t#) Word64
wkTid a
es MVar (ThreadEntries a)
v = \case
  Bool
True -> (State# RealWorld -> (# State# RealWorld, Weak a #)) -> IO (Weak a)
forall a. (State# RealWorld -> (# State# RealWorld, a #)) -> IO a
IO ((State# RealWorld -> (# State# RealWorld, Weak a #))
 -> IO (Weak a))
-> (State# RealWorld -> (# State# RealWorld, Weak a #))
-> IO (Weak a)
forall a b. (a -> b) -> a -> b
$ \State# RealWorld
s0 ->
    case ThreadId#
-> a
-> (State# RealWorld -> (# State# RealWorld, () #))
-> State# RealWorld
-> (# State# RealWorld, Weak# a #)
forall a b c.
a
-> b
-> (State# RealWorld -> (# State# RealWorld, c #))
-> State# RealWorld
-> (# State# RealWorld, Weak# b #)
mkWeak# ThreadId#
t# a
es State# RealWorld -> (# State# RealWorld, () #)
finalizer State# RealWorld
s0 of
      (# State# RealWorld
s1, Weak# a
w #) -> (# State# RealWorld
s1, Weak# a -> Weak a
forall v. Weak# v -> Weak v
Weak Weak# a
w #)
  Bool
False -> (State# RealWorld -> (# State# RealWorld, Weak a #)) -> IO (Weak a)
forall a. (State# RealWorld -> (# State# RealWorld, a #)) -> IO a
IO ((State# RealWorld -> (# State# RealWorld, Weak a #))
 -> IO (Weak a))
-> (State# RealWorld -> (# State# RealWorld, Weak a #))
-> IO (Weak a)
forall a b. (a -> b) -> a -> b
$ \State# RealWorld
s0 ->
    case ThreadId#
-> a -> State# RealWorld -> (# State# RealWorld, Weak# a #)
forall a b.
a -> b -> State# RealWorld -> (# State# RealWorld, Weak# b #)
mkWeakNoFinalizer# ThreadId#
t# a
es State# RealWorld
s0 of
      (# State# RealWorld
s1, Weak# a
w #) -> (# State# RealWorld
s1, Weak# a -> Weak a
forall v. Weak# v -> Weak v
Weak Weak# a
w #)
  where
    -- The finalizer runs only if the corresponding entry is in the map. It
    -- might not be there for two reasons:
    --
    -- 1. Registration of the thread was interrupted by an asynchronous
    --    exception after the finalizer was attached, but before the update of
    --    the map was committed. The commit was rolled back, so there is
    --    nothing to clean up (and if the thread registered successfully
    --    afterwards, the entry belongs to the finalizer attached then).
    --
    -- 2. The thread registered successfully after one or more interrupted
    --    attempts, so multiple finalizers run on its death and another one
    --    already cleaned up the entry.
    IO State# RealWorld -> (# State# RealWorld, () #)
finalizer = MVar (ThreadEntries a)
-> (ThreadEntries a -> IO (ThreadEntries a)) -> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
S.modifyMVar_ MVar (ThreadEntries a)
v ((ThreadEntries a -> IO (ThreadEntries a)) -> IO ())
-> (ThreadEntries a -> IO (ThreadEntries a)) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ThreadEntries a
te -> do
      ThreadEntries a -> IO (ThreadEntries a)
forall a. a -> IO a
forall (f :: Type -> Type) a. Applicative f => a -> f a
pure (ThreadEntries a -> IO (ThreadEntries a))
-> ThreadEntries a -> IO (ThreadEntries a)
forall a b. (a -> b) -> a -> b
$ case (Word64 -> Weak a -> Maybe (Weak a))
-> Word64
-> Word64Map (Weak a)
-> (Maybe (Weak a), Word64Map (Weak a))
forall a.
(Word64 -> a -> Maybe a)
-> Word64 -> Word64Map a -> (Maybe a, Word64Map a)
M.updateLookupWithKey (\Word64
_ Weak a
_ -> Maybe (Weak a)
forall a. Maybe a
Nothing) Word64
wkTid ThreadEntries a
te.entries of
        (Maybe (Weak a)
Nothing, Word64Map (Weak a)
_) -> ThreadEntries a
te
        (Just Weak a
_, Word64Map (Weak a)
newEntries) -> ThreadEntries
          { $sel:capacity:ThreadEntries :: Int
capacity = case ThreadEntries a
te.capacity of
              -- If the template copy of the environment hasn't been consumed
              -- yet, the capacity can be restored.
              Int
0 -> Int
0
              Int
n -> Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1
          , $sel:entries:ThreadEntries :: Word64Map (Weak a)
entries = Word64Map (Weak a)
newEntries
          }