-- | Internal implementation details for "Data.Pool".
--
-- This module is intended for internal use only, and may change without warning
-- in subsequent releases.
{-# OPTIONS_HADDOCK not-home #-}
module Data.Pool.Internal where

import Control.Concurrent
import Control.Exception
import Control.Monad
import Data.IORef
import Data.Primitive.SmallArray
import GHC.Clock
import qualified Data.List as L

-- | Striped resource pool based on "Control.Concurrent.QSem".
--
-- The number of stripes is arranged to be equal to the number of capabilities
-- so that they never compete over access to the same stripe. This results in a
-- very good performance in a multi-threaded environment.
data Pool a = Pool
  { Pool a -> PoolConfig a
poolConfig   :: !(PoolConfig a)
  , Pool a -> SmallArray (LocalPool a)
localPools   :: !(SmallArray (LocalPool a))
  , Pool a -> IORef ()
reaperRef    :: !(IORef ())
  }

-- | A single, capability-local pool.
data LocalPool a = LocalPool
  { LocalPool a -> Int
stripeId   :: !Int
  , LocalPool a -> MVar (Stripe a)
stripeVar  :: !(MVar (Stripe a))
  , LocalPool a -> IORef ()
cleanerRef :: !(IORef ())
  }

-- | Stripe of a resource pool. If @available@ is 0, the list of threads waiting
-- for a resource (each with an associated 'MVar') is @queue ++ reverse queueR@.
data Stripe a = Stripe
  { Stripe a -> Int
available :: !Int
  , Stripe a -> [Entry a]
cache     :: ![Entry a]
  , Stripe a -> Queue a
queue     :: !(Queue a)
  , Stripe a -> Queue a
queueR    :: !(Queue a)
  }

-- | An existing resource currently sitting in a pool.
data Entry a = Entry
  { Entry a -> a
entry    :: a
  , Entry a -> Double
lastUsed :: !Double
  }

-- | A queue of MVarS corresponding to threads waiting for resources.
--
-- Basically a monomorphic list to save two pointer indirections.
data Queue a = Queue !(MVar (Maybe a)) (Queue a) | Empty

-- | Configuration of a 'Pool'.
data PoolConfig a = PoolConfig
  { PoolConfig a -> IO a
createResource :: !(IO a)
    -- ^ The action that creates a new resource.
  , PoolConfig a -> a -> IO ()
freeResource :: !(a -> IO ())
    -- ^ The action that destroys an existing resource.
  , PoolConfig a -> Double
poolCacheTTL :: !Double
  -- ^ The amount of seconds for which an unused resource is kept around. The
  -- smallest acceptable value is @0.5@.
  --
  -- /Note:/ the elapsed time before destroying a resource may be a little
  -- longer than requested, as the collector thread wakes at 1-second intervals.
  , PoolConfig a -> Int
poolMaxResources :: !Int
  -- ^ The maximum number of resources to keep open across all stripes. The
  -- smallest acceptable value is @1@.
  --
  -- /Note:/ for each stripe the number of resources is divided by the number of
  -- capabilities and rounded up. Therefore the pool might end up creating up to
  -- @N - 1@ resources more in total than specified, where @N@ is the number of
  -- capabilities.
  }

-- | Create a new striped resource pool.
--
-- The number of stripes is equal to the number of capabilities.
--
-- /Note:/ although the runtime system will destroy all idle resources when the
-- pool is garbage collected, it's recommended to manually call
-- 'destroyAllResources' when you're done with the pool so that the resources
-- are freed up as soon as possible.
newPool :: PoolConfig a -> IO (Pool a)
newPool :: PoolConfig a -> IO (Pool a)
newPool PoolConfig a
pc = do
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (PoolConfig a -> Double
forall a. PoolConfig a -> Double
poolCacheTTL PoolConfig a
pc Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
< Double
0.5) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    [Char] -> IO ()
forall a. HasCallStack => [Char] -> a
error [Char]
"poolCacheTTL must be at least 0.5"
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (PoolConfig a -> Int
forall a. PoolConfig a -> Int
poolMaxResources PoolConfig a
pc Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
1) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    [Char] -> IO ()
forall a. HasCallStack => [Char] -> a
error [Char]
"poolMaxResources must be at least 1"
  Int
numStripes <- IO Int
getNumCapabilities
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
numStripes Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
1) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    [Char] -> IO ()
forall a. HasCallStack => [Char] -> a
error [Char]
"numStripes must be at least 1"
  SmallArray (LocalPool a)
pools <- ([LocalPool a] -> SmallArray (LocalPool a))
-> IO [LocalPool a] -> IO (SmallArray (LocalPool a))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (Int -> [LocalPool a] -> SmallArray (LocalPool a)
forall a. Int -> [a] -> SmallArray a
smallArrayFromListN Int
numStripes) (IO [LocalPool a] -> IO (SmallArray (LocalPool a)))
-> ((Int -> IO (LocalPool a)) -> IO [LocalPool a])
-> (Int -> IO (LocalPool a))
-> IO (SmallArray (LocalPool a))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Int] -> (Int -> IO (LocalPool a)) -> IO [LocalPool a]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM [Int
1..Int
numStripes] ((Int -> IO (LocalPool a)) -> IO (SmallArray (LocalPool a)))
-> (Int -> IO (LocalPool a)) -> IO (SmallArray (LocalPool a))
forall a b. (a -> b) -> a -> b
$ \Int
n -> do
    IORef ()
ref <- () -> IO (IORef ())
forall a. a -> IO (IORef a)
newIORef ()
    MVar (Stripe a)
stripe <- Stripe a -> IO (MVar (Stripe a))
forall a. a -> IO (MVar a)
newMVar Stripe :: forall a. Int -> [Entry a] -> Queue a -> Queue a -> Stripe a
Stripe
      { available :: Int
available = PoolConfig a -> Int
forall a. PoolConfig a -> Int
poolMaxResources PoolConfig a
pc Int -> Int -> Int
`quotCeil` Int
numStripes
      , cache :: [Entry a]
cache     = []
      , queue :: Queue a
queue     = Queue a
forall a. Queue a
Empty
      , queueR :: Queue a
queueR    = Queue a
forall a. Queue a
Empty
      }
    -- When the local pool goes out of scope, free its resources.
    IO (Weak (IORef ())) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Weak (IORef ())) -> IO ())
-> (IO () -> IO (Weak (IORef ()))) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IORef () -> IO () -> IO (Weak (IORef ()))
forall a. IORef a -> IO () -> IO (Weak (IORef a))
mkWeakIORef IORef ()
ref (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ (Entry a -> Bool) -> (a -> IO ()) -> MVar (Stripe a) -> IO ()
forall a.
(Entry a -> Bool) -> (a -> IO ()) -> MVar (Stripe a) -> IO ()
cleanStripe (Bool -> Entry a -> Bool
forall a b. a -> b -> a
const Bool
True) (PoolConfig a -> a -> IO ()
forall a. PoolConfig a -> a -> IO ()
freeResource PoolConfig a
pc) MVar (Stripe a)
stripe
    LocalPool a -> IO (LocalPool a)
forall (f :: * -> *) a. Applicative f => a -> f a
pure LocalPool :: forall a. Int -> MVar (Stripe a) -> IORef () -> LocalPool a
LocalPool { stripeId :: Int
stripeId   = Int
n
                   , stripeVar :: MVar (Stripe a)
stripeVar  = MVar (Stripe a)
stripe
                   , cleanerRef :: IORef ()
cleanerRef = IORef ()
ref
                   }
  IO (Pool a) -> IO (Pool a)
forall a. IO a -> IO a
mask_ (IO (Pool a) -> IO (Pool a)) -> IO (Pool a) -> IO (Pool a)
forall a b. (a -> b) -> a -> b
$ do
    IORef ()
ref        <- () -> IO (IORef ())
forall a. a -> IO (IORef a)
newIORef ()
    ThreadId
collectorA <- ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId
forkIOWithUnmask (((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId)
-> ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
unmask -> IO () -> IO ()
forall a. IO a -> IO a
unmask (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ SmallArray (LocalPool a) -> IO ()
forall (t :: * -> *) b. Foldable t => t (LocalPool a) -> IO b
collector SmallArray (LocalPool a)
pools
    IO (Weak (IORef ())) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Weak (IORef ())) -> IO ())
-> (IO () -> IO (Weak (IORef ()))) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IORef () -> IO () -> IO (Weak (IORef ()))
forall a. IORef a -> IO () -> IO (Weak (IORef a))
mkWeakIORef IORef ()
ref (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      -- When the pool goes out of scope, stop the collector. Resources existing
      -- in stripes will be taken care by their cleaners.
      ThreadId -> IO ()
killThread ThreadId
collectorA
    Pool a -> IO (Pool a)
forall (f :: * -> *) a. Applicative f => a -> f a
pure Pool :: forall a.
PoolConfig a -> SmallArray (LocalPool a) -> IORef () -> Pool a
Pool { poolConfig :: PoolConfig a
poolConfig = PoolConfig a
pc
              , localPools :: SmallArray (LocalPool a)
localPools = SmallArray (LocalPool a)
pools
              , reaperRef :: IORef ()
reaperRef  = IORef ()
ref
              }
  where
    quotCeil :: Int -> Int -> Int
    quotCeil :: Int -> Int -> Int
quotCeil Int
x Int
y =
      -- Basically ceiling (x / y) without going through Double.
      let (Int
z, Int
r) = Int
x Int -> Int -> (Int, Int)
forall a. Integral a => a -> a -> (a, a)
`quotRem` Int
y in if Int
r Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 then Int
z else Int
z Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1

    -- Collect stale resources from the pool once per second.
    collector :: t (LocalPool a) -> IO b
collector t (LocalPool a)
pools = IO () -> IO b
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO b) -> IO () -> IO b
forall a b. (a -> b) -> a -> b
$ do
      Int -> IO ()
threadDelay Int
1000000
      Double
now <- IO Double
getMonotonicTime
      let isStale :: Entry a -> Bool
isStale Entry a
e = Double
now Double -> Double -> Double
forall a. Num a => a -> a -> a
- Entry a -> Double
forall a. Entry a -> Double
lastUsed Entry a
e Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
> PoolConfig a -> Double
forall a. PoolConfig a -> Double
poolCacheTTL PoolConfig a
pc
      (LocalPool a -> IO ()) -> t (LocalPool a) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ ((Entry a -> Bool) -> (a -> IO ()) -> MVar (Stripe a) -> IO ()
forall a.
(Entry a -> Bool) -> (a -> IO ()) -> MVar (Stripe a) -> IO ()
cleanStripe Entry a -> Bool
forall a. Entry a -> Bool
isStale (PoolConfig a -> a -> IO ()
forall a. PoolConfig a -> a -> IO ()
freeResource PoolConfig a
pc) (MVar (Stripe a) -> IO ())
-> (LocalPool a -> MVar (Stripe a)) -> LocalPool a -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. LocalPool a -> MVar (Stripe a)
forall a. LocalPool a -> MVar (Stripe a)
stripeVar) t (LocalPool a)
pools

-- | Destroy a resource.
--
-- Note that this will ignore any exceptions in the destroy function.
destroyResource :: Pool a -> LocalPool a -> a -> IO ()
destroyResource :: Pool a -> LocalPool a -> a -> IO ()
destroyResource Pool a
pool LocalPool a
lp a
a = do
  IO () -> IO ()
forall a. IO a -> IO a
uninterruptibleMask_ (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do -- Note [signal uninterruptible]
    Stripe a
stripe <- MVar (Stripe a) -> IO (Stripe a)
forall a. MVar a -> IO a
takeMVar (LocalPool a -> MVar (Stripe a)
forall a. LocalPool a -> MVar (Stripe a)
stripeVar LocalPool a
lp)
    Stripe a
newStripe <- Stripe a -> Maybe a -> IO (Stripe a)
forall a. Stripe a -> Maybe a -> IO (Stripe a)
signal Stripe a
stripe Maybe a
forall a. Maybe a
Nothing
    MVar (Stripe a) -> Stripe a -> IO ()
forall a. MVar a -> a -> IO ()
putMVar (LocalPool a -> MVar (Stripe a)
forall a. LocalPool a -> MVar (Stripe a)
stripeVar LocalPool a
lp) Stripe a
newStripe
    IO (Either SomeException ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Either SomeException ()) -> IO ())
-> (IO () -> IO (Either SomeException ())) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall a.
Exception SomeException =>
IO a -> IO (Either SomeException a)
forall e a. Exception e => IO a -> IO (Either e a)
try @SomeException (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ PoolConfig a -> a -> IO ()
forall a. PoolConfig a -> a -> IO ()
freeResource (Pool a -> PoolConfig a
forall a. Pool a -> PoolConfig a
poolConfig Pool a
pool) a
a

-- | Return a resource to the given 'LocalPool'.
putResource :: LocalPool a -> a -> IO ()
putResource :: LocalPool a -> a -> IO ()
putResource LocalPool a
lp a
a = do
  IO () -> IO ()
forall a. IO a -> IO a
uninterruptibleMask_ (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do -- Note [signal uninterruptible]
    Stripe a
stripe    <- MVar (Stripe a) -> IO (Stripe a)
forall a. MVar a -> IO a
takeMVar (LocalPool a -> MVar (Stripe a)
forall a. LocalPool a -> MVar (Stripe a)
stripeVar LocalPool a
lp)
    Stripe a
newStripe <- Stripe a -> Maybe a -> IO (Stripe a)
forall a. Stripe a -> Maybe a -> IO (Stripe a)
signal Stripe a
stripe (a -> Maybe a
forall a. a -> Maybe a
Just a
a)
    MVar (Stripe a) -> Stripe a -> IO ()
forall a. MVar a -> a -> IO ()
putMVar (LocalPool a -> MVar (Stripe a)
forall a. LocalPool a -> MVar (Stripe a)
stripeVar LocalPool a
lp) Stripe a
newStripe

-- | Destroy all resources in all stripes in the pool.
--
-- Note that this will ignore any exceptions in the destroy function.
--
-- This function is useful when you detect that all resources in the pool are
-- broken. For example after a database has been restarted all connections
-- opened before the restart will be broken. In that case it's better to close
-- those connections so that 'takeResource' won't take a broken connection from
-- the pool but will open a new connection instead.
--
-- Another use-case for this function is that when you know you are done with
-- the pool you can destroy all idle resources immediately instead of waiting on
-- the garbage collector to destroy them, thus freeing up those resources
-- sooner.
destroyAllResources :: Pool a -> IO ()
destroyAllResources :: Pool a -> IO ()
destroyAllResources Pool a
pool = SmallArray (LocalPool a) -> (LocalPool a -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ (Pool a -> SmallArray (LocalPool a)
forall a. Pool a -> SmallArray (LocalPool a)
localPools Pool a
pool) ((LocalPool a -> IO ()) -> IO ())
-> (LocalPool a -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \LocalPool a
lp -> do
  (Entry a -> Bool) -> (a -> IO ()) -> MVar (Stripe a) -> IO ()
forall a.
(Entry a -> Bool) -> (a -> IO ()) -> MVar (Stripe a) -> IO ()
cleanStripe (Bool -> Entry a -> Bool
forall a b. a -> b -> a
const Bool
True) (PoolConfig a -> a -> IO ()
forall a. PoolConfig a -> a -> IO ()
freeResource (Pool a -> PoolConfig a
forall a. Pool a -> PoolConfig a
poolConfig Pool a
pool)) (LocalPool a -> MVar (Stripe a)
forall a. LocalPool a -> MVar (Stripe a)
stripeVar LocalPool a
lp)

----------------------------------------
-- Helpers

-- | Get a capability-local pool.
getLocalPool :: SmallArray (LocalPool a) -> IO (LocalPool a)
getLocalPool :: SmallArray (LocalPool a) -> IO (LocalPool a)
getLocalPool SmallArray (LocalPool a)
pools = do
  (Int
cid, Bool
_) <- ThreadId -> IO (Int, Bool)
threadCapability (ThreadId -> IO (Int, Bool)) -> IO ThreadId -> IO (Int, Bool)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO ThreadId
myThreadId
  LocalPool a -> IO (LocalPool a)
forall (f :: * -> *) a. Applicative f => a -> f a
pure (LocalPool a -> IO (LocalPool a))
-> LocalPool a -> IO (LocalPool a)
forall a b. (a -> b) -> a -> b
$ SmallArray (LocalPool a)
pools SmallArray (LocalPool a) -> Int -> LocalPool a
forall a. SmallArray a -> Int -> a
`indexSmallArray` (Int
cid Int -> Int -> Int
forall a. Integral a => a -> a -> a
`rem` SmallArray (LocalPool a) -> Int
forall a. SmallArray a -> Int
sizeofSmallArray SmallArray (LocalPool a)
pools)

-- | Wait for the resource to be put into a given 'MVar'.
waitForResource :: MVar (Stripe a) -> MVar (Maybe a) -> IO (Maybe a)
waitForResource :: MVar (Stripe a) -> MVar (Maybe a) -> IO (Maybe a)
waitForResource MVar (Stripe a)
mstripe MVar (Maybe a)
q = MVar (Maybe a) -> IO (Maybe a)
forall a. MVar a -> IO a
takeMVar MVar (Maybe a)
q IO (Maybe a) -> IO () -> IO (Maybe a)
forall a b. IO a -> IO b -> IO a
`onException` IO ()
cleanup
  where
    cleanup :: IO ()
cleanup = IO () -> IO ()
forall a. IO a -> IO a
uninterruptibleMask_ (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do -- Note [signal uninterruptible]
      Stripe a
stripe    <- MVar (Stripe a) -> IO (Stripe a)
forall a. MVar a -> IO a
takeMVar MVar (Stripe a)
mstripe
      Stripe a
newStripe <- MVar (Maybe a) -> IO (Maybe (Maybe a))
forall a. MVar a -> IO (Maybe a)
tryTakeMVar MVar (Maybe a)
q IO (Maybe (Maybe a))
-> (Maybe (Maybe a) -> IO (Stripe a)) -> IO (Stripe a)
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        Just Maybe a
ma -> do
          -- Between entering the exception handler and taking ownership of
          -- the stripe we got the resource we wanted. We don't need it
          -- anymore though, so pass it to someone else.
          Stripe a -> Maybe a -> IO (Stripe a)
forall a. Stripe a -> Maybe a -> IO (Stripe a)
signal Stripe a
stripe Maybe a
ma
        Maybe (Maybe a)
Nothing -> do
          -- If we're still waiting, fill up the MVar with an undefined value
          -- so that 'signal' can discard our MVar from the queue.
          MVar (Maybe a) -> Maybe a -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar (Maybe a)
q (Maybe a -> IO ()) -> Maybe a -> IO ()
forall a b. (a -> b) -> a -> b
$ [Char] -> Maybe a
forall a. HasCallStack => [Char] -> a
error [Char]
"unreachable"
          Stripe a -> IO (Stripe a)
forall (f :: * -> *) a. Applicative f => a -> f a
pure Stripe a
stripe
      MVar (Stripe a) -> Stripe a -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar (Stripe a)
mstripe Stripe a
newStripe

-- | If an exception is received while a resource is being created, restore the
-- original size of the stripe.
restoreSize :: MVar (Stripe a) -> IO ()
restoreSize :: MVar (Stripe a) -> IO ()
restoreSize MVar (Stripe a)
mstripe = IO () -> IO ()
forall a. IO a -> IO a
uninterruptibleMask_ (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  -- 'uninterruptibleMask_' is used since 'takeMVar' might block.
  Stripe a
stripe <- MVar (Stripe a) -> IO (Stripe a)
forall a. MVar a -> IO a
takeMVar MVar (Stripe a)
mstripe
  MVar (Stripe a) -> Stripe a -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar (Stripe a)
mstripe (Stripe a -> IO ()) -> Stripe a -> IO ()
forall a b. (a -> b) -> a -> b
$! Stripe a
stripe { available :: Int
available = Stripe a -> Int
forall a. Stripe a -> Int
available Stripe a
stripe Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1 }

-- | Free resource entries in the stripes that fulfil a given condition.
cleanStripe
  :: (Entry a -> Bool)
  -> (a -> IO ())
  -> MVar (Stripe a)
  -> IO ()
cleanStripe :: (Entry a -> Bool) -> (a -> IO ()) -> MVar (Stripe a) -> IO ()
cleanStripe Entry a -> Bool
isStale a -> IO ()
free MVar (Stripe a)
mstripe = ((forall a. IO a -> IO a) -> IO ()) -> IO ()
forall b. ((forall a. IO a -> IO a) -> IO b) -> IO b
mask (((forall a. IO a -> IO a) -> IO ()) -> IO ())
-> ((forall a. IO a -> IO a) -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
unmask -> do
  -- Asynchronous exceptions need to be masked here to prevent leaking of
  -- 'stale' resources before they're freed.
  [a]
stale <- MVar (Stripe a) -> (Stripe a -> IO (Stripe a, [a])) -> IO [a]
forall a b. MVar a -> (a -> IO (a, b)) -> IO b
modifyMVar MVar (Stripe a)
mstripe ((Stripe a -> IO (Stripe a, [a])) -> IO [a])
-> (Stripe a -> IO (Stripe a, [a])) -> IO [a]
forall a b. (a -> b) -> a -> b
$ \Stripe a
stripe -> IO (Stripe a, [a]) -> IO (Stripe a, [a])
forall a. IO a -> IO a
unmask (IO (Stripe a, [a]) -> IO (Stripe a, [a]))
-> IO (Stripe a, [a]) -> IO (Stripe a, [a])
forall a b. (a -> b) -> a -> b
$ do
    let ([Entry a]
stale, [Entry a]
fresh) = (Entry a -> Bool) -> [Entry a] -> ([Entry a], [Entry a])
forall a. (a -> Bool) -> [a] -> ([a], [a])
L.partition Entry a -> Bool
isStale (Stripe a -> [Entry a]
forall a. Stripe a -> [Entry a]
cache Stripe a
stripe)
        -- There's no need to update 'available' here because it only tracks
        -- the number of resources taken from the pool.
        newStripe :: Stripe a
newStripe      = Stripe a
stripe { cache :: [Entry a]
cache = [Entry a]
fresh }
    Stripe a
newStripe Stripe a -> IO (Stripe a, [a]) -> IO (Stripe a, [a])
`seq` (Stripe a, [a]) -> IO (Stripe a, [a])
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Stripe a
newStripe, (Entry a -> a) -> [Entry a] -> [a]
forall a b. (a -> b) -> [a] -> [b]
map Entry a -> a
forall a. Entry a -> a
entry [Entry a]
stale)
  -- We need to ignore exceptions in the 'free' function, otherwise if an
  -- exception is thrown half-way, we leak the rest of the resources. Also,
  -- asynchronous exceptions need to be hard masked here since freeing a
  -- resource might in theory block.
  IO () -> IO ()
forall a. IO a -> IO a
uninterruptibleMask_ (IO () -> IO ())
-> ((a -> IO (Either SomeException ())) -> IO ())
-> (a -> IO (Either SomeException ()))
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [a] -> (a -> IO (Either SomeException ())) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [a]
stale ((a -> IO (Either SomeException ())) -> IO ())
-> (a -> IO (Either SomeException ())) -> IO ()
forall a b. (a -> b) -> a -> b
$ forall a.
Exception SomeException =>
IO a -> IO (Either SomeException a)
forall e a. Exception e => IO a -> IO (Either e a)
try @SomeException (IO () -> IO (Either SomeException ()))
-> (a -> IO ()) -> a -> IO (Either SomeException ())
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> IO ()
free

-- Note [signal uninterruptible]
--
--   If we have
--
--      bracket takeResource putResource (...)
--
--   and an exception arrives at the putResource, then we must not lose the
--   resource. The putResource is masked by bracket, but taking the MVar might
--   block, and so it would be interruptible. Hence we need an uninterruptible
--   variant of mask here.
signal :: Stripe a -> Maybe a -> IO (Stripe a)
signal :: Stripe a -> Maybe a -> IO (Stripe a)
signal Stripe a
stripe Maybe a
ma = if Stripe a -> Int
forall a. Stripe a -> Int
available Stripe a
stripe Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0
  then Queue a -> Queue a -> IO (Stripe a)
loop (Stripe a -> Queue a
forall a. Stripe a -> Queue a
queue Stripe a
stripe) (Stripe a -> Queue a
forall a. Stripe a -> Queue a
queueR Stripe a
stripe)
  else do
    [Entry a]
newCache <- case Maybe a
ma of
      Just a
a -> do
        Double
now <- IO Double
getMonotonicTime
        [Entry a] -> IO [Entry a]
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Entry a] -> IO [Entry a]) -> [Entry a] -> IO [Entry a]
forall a b. (a -> b) -> a -> b
$ a -> Double -> Entry a
forall a. a -> Double -> Entry a
Entry a
a Double
now Entry a -> [Entry a] -> [Entry a]
forall a. a -> [a] -> [a]
: Stripe a -> [Entry a]
forall a. Stripe a -> [Entry a]
cache Stripe a
stripe
      Maybe a
Nothing -> [Entry a] -> IO [Entry a]
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Entry a] -> IO [Entry a]) -> [Entry a] -> IO [Entry a]
forall a b. (a -> b) -> a -> b
$ Stripe a -> [Entry a]
forall a. Stripe a -> [Entry a]
cache Stripe a
stripe
    Stripe a -> IO (Stripe a)
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Stripe a -> IO (Stripe a)) -> Stripe a -> IO (Stripe a)
forall a b. (a -> b) -> a -> b
$! Stripe a
stripe { available :: Int
available = Stripe a -> Int
forall a. Stripe a -> Int
available Stripe a
stripe Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1
                   , cache :: [Entry a]
cache = [Entry a]
newCache
                   }
  where
    loop :: Queue a -> Queue a -> IO (Stripe a)
loop Queue a
Empty Queue a
Empty = do
      [Entry a]
newCache <- case Maybe a
ma of
        Just a
a -> do
          Double
now <- IO Double
getMonotonicTime
          [Entry a] -> IO [Entry a]
forall (f :: * -> *) a. Applicative f => a -> f a
pure [a -> Double -> Entry a
forall a. a -> Double -> Entry a
Entry a
a Double
now]
        Maybe a
Nothing -> [Entry a] -> IO [Entry a]
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
      Stripe a -> IO (Stripe a)
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Stripe a -> IO (Stripe a)) -> Stripe a -> IO (Stripe a)
forall a b. (a -> b) -> a -> b
$! Stripe :: forall a. Int -> [Entry a] -> Queue a -> Queue a -> Stripe a
Stripe { available :: Int
available = Int
1
                     , cache :: [Entry a]
cache = [Entry a]
newCache
                     , queue :: Queue a
queue = Queue a
forall a. Queue a
Empty
                     , queueR :: Queue a
queueR = Queue a
forall a. Queue a
Empty
                     }
    loop Queue a
Empty        Queue a
qR = Queue a -> Queue a -> IO (Stripe a)
loop (Queue a -> Queue a
forall a. Queue a -> Queue a
reverseQueue Queue a
qR) Queue a
forall a. Queue a
Empty
    loop (Queue MVar (Maybe a)
q Queue a
qs) Queue a
qR = MVar (Maybe a) -> Maybe a -> IO Bool
forall a. MVar a -> a -> IO Bool
tryPutMVar MVar (Maybe a)
q Maybe a
ma IO Bool -> (Bool -> IO (Stripe a)) -> IO (Stripe a)
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      -- This fails when 'waitForResource' went into the exception handler and
      -- filled the MVar (with an undefined value) itself. In such case we
      -- simply ignore it.
      Bool
False -> Queue a -> Queue a -> IO (Stripe a)
loop Queue a
qs Queue a
qR
      Bool
True  -> Stripe a -> IO (Stripe a)
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Stripe a -> IO (Stripe a)) -> Stripe a -> IO (Stripe a)
forall a b. (a -> b) -> a -> b
$! Stripe a
stripe { available :: Int
available = Int
0
                              , queue :: Queue a
queue = Queue a
qs
                              , queueR :: Queue a
queueR = Queue a
qR
                              }

reverseQueue :: Queue a -> Queue a
reverseQueue :: Queue a -> Queue a
reverseQueue = Queue a -> Queue a -> Queue a
forall a. Queue a -> Queue a -> Queue a
go Queue a
forall a. Queue a
Empty
  where
    go :: Queue a -> Queue a -> Queue a
go Queue a
acc = \case
      Queue a
Empty      -> Queue a
acc
      Queue MVar (Maybe a)
x Queue a
xs -> Queue a -> Queue a -> Queue a
go (MVar (Maybe a) -> Queue a -> Queue a
forall a. MVar (Maybe a) -> Queue a -> Queue a
Queue MVar (Maybe a)
x Queue a
acc) Queue a
xs