diff --git a/Setup.hs b/Setup.hs deleted file mode 100644 --- a/Setup.hs +++ /dev/null @@ -1,2 +0,0 @@ -import Distribution.Simple -main = defaultMain diff --git a/glue-core.cabal b/glue-core.cabal --- a/glue-core.cabal +++ b/glue-core.cabal @@ -1,78 +1,92 @@ -name: glue-core -version: 0.6.3 -synopsis: Make better services and clients. -description: Combinator library to enhance the general functionality of services and clients. -license: BSD3 -license-file: LICENSE -author: Sean Parsons -maintainer: github@futurenotfound.com -category: Network -build-type: Simple -cabal-version: >=1.10 +cabal-version: 3.0 +name: glue-core +version: 0.7.0 +synopsis: Make better services and clients. +description: Combinator library to enhance the general functionality of services and clients. +license: BSD-3-Clause +license-file: LICENSE +author: Sean Parsons +maintainer: github@futurenotfound.com +category: Network +build-type: Simple +tested-with: GHC == 9.6.7, GHC == 9.8.4, GHC == 9.10.3, GHC == 9.12.3 source-repository head - type: git - location: git://github.com/seanparsons/glue.git + type: git + location: https://github.com/seanparsons/glue.git +common common-options + default-language: Haskell2010 + ghc-options: + -Wall + -Wcompat + -Widentities + -Wincomplete-record-updates + -Wincomplete-uni-patterns + -Wmissing-home-modules + -Wpartial-fields + -Wredundant-constraints + library - exposed-modules: Glue.Failover - Glue.Caching - Glue.CircuitBreaker - Glue.DogpileProtection - Glue.Timeout - Glue.Retry - Glue.Batcher - Glue.Preload - Glue.Switching - -- other-extensions: - build-depends: base >=4.6 && <5, - glue-common, - transformers, - transformers-base, - lifted-base, - time, - monad-control, - unordered-containers, - hashable, - text - ghc-options: -Wall - hs-source-dirs: src - default-language: Haskell2010 + import: common-options + hs-source-dirs: src + exposed-modules: + Glue.Batcher + Glue.Caching + Glue.CircuitBreaker + Glue.ConcurrencyLimit + Glue.DogpileProtection + Glue.Failover + Glue.Hedging + Glue.LoadBalancer + Glue.Preload + Glue.Retry + Glue.Shutdown + Glue.Switching + Glue.Timeout + build-depends: + , base >= 4.16 && < 5 + , glue-common ^>= 0.7.0 + , hashable + , time + , unliftio >= 0.2 + , unordered-containers test-suite glue-core-tests - build-depends: base >=4.6 && <5, - glue-common , - glue-core, - QuickCheck -any, - quickcheck-instances, - hspec, - transformers, - transformers-base, - lifted-base, - time, - monad-control, - unordered-containers, - hashable, - ekg-core, - text, - async - other-modules: Glue.CachingSpec - Glue.FailoverSpec - Glue.RetrySpec - Glue.DogpileProtectionSpec - Glue.TimeoutSpec - Glue.CircuitBreakerSpec - Glue.BatcherSpec - Glue.PreloadSpec - Glue.SwitchingSpec - Spec - ghc-options: -rtsopts - -Wall - -O2 - -threaded - build-tools: hspec-discover - type: exitcode-stdio-1.0 - main-is: Main.hs - buildable: True - default-language: Haskell2010 - hs-source-dirs: test + import: common-options + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Main.hs + other-modules: + Glue.BatcherSpec + Glue.CachingSpec + Glue.CircuitBreakerSpec + Glue.ConcurrencyLimitSpec + Glue.DogpileProtectionSpec + Glue.FailoverSpec + Glue.HedgingSpec + Glue.LoadBalancerSpec + Glue.PreloadSpec + Glue.RetrySpec + Glue.ShutdownSpec + Glue.SwitchingSpec + Glue.TimeoutSpec + Spec + build-tool-depends: hspec-discover:hspec-discover + ghc-options: + -threaded + -rtsopts + -with-rtsopts=-N + -O2 + build-depends: + , async + , base >= 4.16 && < 5 + , glue-common + , glue-core + , hashable + , hspec + , QuickCheck + , quickcheck-instances + , time + , unliftio >= 0.2 + , unordered-containers diff --git a/src/Glue/Batcher.hs b/src/Glue/Batcher.hs --- a/src/Glue/Batcher.hs +++ b/src/Glue/Batcher.hs @@ -2,85 +2,100 @@ {-# LANGUAGE ScopedTypeVariables #-} -- | Module for creating a service that batches calls across requests. -module Glue.Batcher( - BatchingOptions +module Glue.Batcher + ( BatchingOptions (..) , batchingService , defaultBatchingOptions - , batchWindowMs -) where + ) where -import Control.Concurrent.Lifted -import Control.Exception.Lifted -import Control.Monad -import Control.Monad.Trans.Control -import Data.Foldable -import Data.Hashable -import qualified Data.HashMap.Strict as M -import qualified Data.HashSet as S -import Data.IORef.Lifted +import Control.Monad (join, void) +import Data.Foldable (traverse_) +import Data.Hashable (Hashable) +import qualified Data.HashMap.Strict as M +import qualified Data.HashSet as S import Glue.Types +import UnliftIO (MonadIO, MonadUnliftIO, SomeException) +import UnliftIO.Concurrent (forkIO, threadDelay) +import UnliftIO.IORef (IORef, atomicModifyIORef', newIORef) +import UnliftIO.MVar (newEmptyMVar, putMVar) -- | Options for configuring the batching. -data BatchingOptions = BatchingOptions { - batchWindowMs :: Int -- ^ Window in milliseconds over which to batch. -} deriving (Eq, Show) +newtype BatchingOptions = BatchingOptions + { batchWindowMs :: Int + -- ^ Window in milliseconds over which to batch. + } + deriving (Eq, Show) --- | Default instance of 'BatchingOptions' with a batching window of 10ms. +-- | Default instance of t'BatchingOptions' with a batching window of 10ms. defaultBatchingOptions :: BatchingOptions -defaultBatchingOptions = BatchingOptions { - batchWindowMs = 10 -} +defaultBatchingOptions = + BatchingOptions + { batchWindowMs = 10 + } -- | Cumulative request batch of all the outstanding calls. data RequestBatch a b = RequestBatch [PendingRequest a b] (MultiGetRequest a) -- | Individual request to satisfy, for either a single key or a multi request. -data PendingRequest a b = - SingleRequest a (ResultVar (Maybe b)) | - MultiRequest (MultiGetRequest a) (ResultVar (MultiGetResponse a b)) +data PendingRequest a b + = SingleRequest a (ResultVar (Maybe b)) + | MultiRequest (MultiGetRequest a) (ResultVar (MultiGetResponse a b)) -applyToPending :: (Eq a, Hashable a, MonadBaseControl IO m) => Either SomeException (M.HashMap a b) -> PendingRequest a b -> m () -applyToPending (Left e) (SingleRequest _ var) = putMVar var $ Left e -applyToPending (Right results) (SingleRequest a var) = putMVar var $ Right $ M.lookup a results -applyToPending (Left e) (MultiRequest _ var) = putMVar var $ Left e -applyToPending (Right results) (MultiRequest as var) = putMVar var $ Right $ M.filterWithKey (\k -> \_ -> S.member k as) results +applyToPending :: (Hashable a, MonadIO m) => Either SomeException (M.HashMap a b) -> PendingRequest a b -> m () +applyToPending (Left e) (SingleRequest _ var) = putMVar var $ Left e +applyToPending (Right results) (SingleRequest a var) = putMVar var $ Right $ M.lookup a results +applyToPending (Left e) (MultiRequest _ var) = putMVar var $ Left e +applyToPending (Right results) (MultiRequest as var) = putMVar var $ Right $ M.filterWithKey (\k _ -> S.member k as) results -emptyBatch :: (Eq a, Hashable a) => RequestBatch a b +emptyBatch :: RequestBatch a b emptyBatch = RequestBatch [] S.empty -processCalls :: (Eq a, Hashable a, MonadBaseControl IO m) => MultiGetService m a b -> RequestBatch a b -> m () +processCalls :: (Hashable a, MonadUnliftIO m) => MultiGetService m a b -> RequestBatch a b -> m () processCalls service (RequestBatch pendings requests) = do - result <- makeCall service requests - traverse_ (applyToPending result) pendings + result <- makeCall service requests + traverse_ (applyToPending result) pendings -startBatch :: (Eq a, Hashable a, MonadBaseControl IO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> m () -startBatch options service ref = fmap (\_ -> ()) $ fork $ do - threadDelay (1000 * batchWindowMs options) - pendings <- atomicModifyIORef' ref (\p -> (emptyBatch, p)) - processCalls service pendings +startBatch :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> m () +startBatch options service ref = + void $ forkIO $ do + threadDelay (1000 * batchWindowMs options) + pendings <- atomicModifyIORef' ref (\p -> (emptyBatch, p)) + processCalls service pendings -addPending :: (Eq a, Hashable a, MonadBaseControl IO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> PendingRequest a b -> m () -addPending options service ref p@(SingleRequest request _) = join $ atomicModifyIORef' ref (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.insert request rs), if null ps then startBatch options service ref else return ())) -addPending options service ref p@(MultiRequest requests _) = join $ atomicModifyIORef' ref (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.union rs requests), if null ps then startBatch options service ref else return ())) +addPending :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> PendingRequest a b -> m () +addPending options service ref p@(SingleRequest request _) = + join $ + atomicModifyIORef' + ref + (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.insert request rs), if null ps then startBatch options service ref else pure ())) +addPending options service ref p@(MultiRequest requests _) = + join $ + atomicModifyIORef' + ref + (\(RequestBatch ps rs) -> (RequestBatch (p : ps) (S.union rs requests), if null ps then startBatch options service ref else pure ())) -singleService :: (Eq a, Hashable a, MonadBaseControl IO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> BasicService m a (Maybe b) +singleService :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> BasicService m a (Maybe b) singleService options service ref request = do - mvar <- newEmptyMVar - let pending = SingleRequest request mvar - addPending options service ref pending - getResult mvar -multiService :: (Eq a, Hashable a, MonadBaseControl IO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> MultiGetService m a b + mvar <- newEmptyMVar + let pending = SingleRequest request mvar + addPending options service ref pending + getResult mvar + +multiService :: (Hashable a, MonadUnliftIO m) => BatchingOptions -> MultiGetService m a b -> IORef (RequestBatch a b) -> MultiGetService m a b multiService options service ref requests = do - mvar <- newEmptyMVar - let pending = MultiRequest requests mvar - addPending options service ref pending - getResult mvar + mvar <- newEmptyMVar + let pending = MultiRequest requests mvar + addPending options service ref pending + getResult mvar -- | Function for constructing a batching service. -batchingService :: (Eq a, Hashable a, MonadBaseControl IO m, Applicative m, MonadBaseControl IO n) - => BatchingOptions -- ^ Options to configure the batched service. - -> MultiGetService m a b -- ^ Service around which to batch requests. - -> n (BasicService m a (Maybe b), MultiGetService m a b) +batchingService + :: (Hashable a, MonadUnliftIO m, MonadIO n) + => BatchingOptions + -- ^ Options to configure the batched service. + -> MultiGetService m a b + -- ^ Service around which to batch requests. + -> n (BasicService m a (Maybe b), MultiGetService m a b) batchingService options service = do ref <- newIORef emptyBatch - return (singleService options service ref, multiService options service ref) + pure (singleService options service ref, multiService options service ref) diff --git a/src/Glue/Caching.hs b/src/Glue/Caching.hs --- a/src/Glue/Caching.hs +++ b/src/Glue/Caching.hs @@ -2,48 +2,58 @@ {-# LANGUAGE ScopedTypeVariables #-} -- | Module supporting the caching of a service. -module Glue.Caching( - cacheWithBasic +module Glue.Caching + ( cacheWithBasic , cacheWithMulti -) where + ) where -import Data.Hashable +import Data.Hashable (Hashable) import qualified Data.HashMap.Strict as M import qualified Data.HashSet as S import Glue.Types --- | Caching of a `BasicService` instance, that defers to external functions for the actual caching. --- | Note: Values within m will be lost for calls that hit the cache. -cacheWithBasic :: (Monad m) - => (a -> m (Maybe b)) -- ^ Cache lookup function, used before potentially invoking the fallback service. - -> (a -> b -> m ()) -- ^ Cache write function, used after invoking the fallback service to populate the cache. - -> BasicService m a b -- ^ The service to cache. - -> BasicService m a b +-- | Caching of a 'BasicService' instance, that defers to external functions for the actual caching. +-- Note: Values within @m@ will be lost for calls that hit the cache. +cacheWithBasic + :: Monad m + => (a -> m (Maybe b)) + -- ^ Cache lookup function, used before potentially invoking the fallback service. + -> (a -> b -> m ()) + -- ^ Cache write function, used after invoking the fallback service to populate the cache. + -> BasicService m a b + -- ^ The service to cache. + -> BasicService m a b cacheWithBasic lookupWith insertWith service = let fallback request = do - result <- service request - insertWith request result - return result + result <- service request + insertWith request result + pure result cachedService request = do - fromCache <- lookupWith request - maybe (fallback request) return fromCache - in cachedService + fromCache <- lookupWith request + maybe (fallback request) pure fromCache + in cachedService --- | Caching of a `MultiGetService` instance, that defers to external functions for the actual caching. --- | Partial responses will result in partial fallback calls that get just the missing keys. --- | Values within m will be lost for calls that hit the cache. -cacheWithMulti :: (Monad m, Functor m, Eq a, Hashable a) - => ((MultiGetRequest a) -> m (MultiGetResponse a b)) -- ^ Cache lookup function, used before potentially invoking the fallback service. - -> ((MultiGetResponse a b) -> m ()) -- ^ Cache write function, used after invoking the fallback service to populate the cache. - -> MultiGetService m a b -- ^ The service to cache. - -> MultiGetService m a b +-- | Caching of a 'MultiGetService' instance, that defers to external functions for the actual caching. +-- Partial responses will result in partial fallback calls that get just the missing keys. +-- Values within @m@ will be lost for calls that hit the cache. +cacheWithMulti + :: (Monad m, Hashable a) + => (MultiGetRequest a -> m (MultiGetResponse a b)) + -- ^ Cache lookup function, used before potentially invoking the fallback service. + -> (MultiGetResponse a b -> m ()) + -- ^ Cache write function, used after invoking the fallback service to populate the cache. + -> MultiGetService m a b + -- ^ The service to cache. + -> MultiGetService m a b cacheWithMulti lookupWith insertWith service = let fallback request = do - result <- service request - insertWith result - return result + result <- service request + insertWith result + pure result cachedService request = do - fromCache <- lookupWith request - let uncachedKeys = S.difference request (S.fromList $ M.keys fromCache) - if (S.null uncachedKeys) then (return fromCache) else (fmap (M.union fromCache) $ fallback uncachedKeys) - in cachedService + fromCache <- lookupWith request + let uncachedKeys = S.difference request (S.fromList $ M.keys fromCache) + if S.null uncachedKeys + then pure fromCache + else M.union fromCache <$> fallback uncachedKeys + in cachedService diff --git a/src/Glue/CircuitBreaker.hs b/src/Glue/CircuitBreaker.hs --- a/src/Glue/CircuitBreaker.hs +++ b/src/Glue/CircuitBreaker.hs @@ -1,55 +1,68 @@ -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} --- | Module containing circuit breaker functionality, which is the ability to open a circuit once a number of failures have occurred, thereby preventing later calls from attempting to make unsuccessful calls. --- | Often this is useful if the underlying service were to repeatedly time out, so as to reduce the number of calls inflight holding up upstream callers. -module Glue.CircuitBreaker( - CircuitBreakerOptions +-- | Module containing circuit breaker functionality, which is the ability to open a circuit +-- once a number of failures have occurred, thereby preventing later calls from attempting +-- to make unsuccessful calls. +-- Often this is useful if the underlying service were to repeatedly time out, so as to +-- reduce the number of calls inflight holding up upstream callers. +module Glue.CircuitBreaker + ( CircuitBreakerOptions (..) , CircuitBreakerState - , CircuitBreakerException(..) + , CircuitBreakerException (..) , combineCircuitBreakerStates , isCircuitBreakerOpen , isCircuitBreakerClosed , defaultCircuitBreakerOptions , circuitBreaker - , maxBreakerFailures - , resetTimeoutSecs - , breakerDescription -) where + ) where -import Control.Exception.Lifted -import Control.Monad.Base -import Control.Monad.Trans.Control -import Data.IORef.Lifted -import Data.Time.Clock.POSIX -import Data.Typeable +import Data.Time.Clock.POSIX (getPOSIXTime) import Glue.Types +import UnliftIO (Exception, MonadIO, MonadUnliftIO, + liftIO) +import UnliftIO.Exception (bracketOnError, throwIO) +import UnliftIO.IORef (IORef, atomicModifyIORef', newIORef, + readIORef, writeIORef) -- | Options for determining behaviour of circuit breaking services. -data CircuitBreakerOptions = CircuitBreakerOptions { - maxBreakerFailures :: Int -- ^ How many times the underlying service must fail in the given window before the circuit opens. - , resetTimeoutSecs :: Int -- ^ The window of time in which the underlying service must fail for the circuit to open. - , breakerDescription :: String -- ^ Description that is attached to the failure so as to identify the particular circuit. -} +data CircuitBreakerOptions = CircuitBreakerOptions + { maxBreakerFailures :: Int + -- ^ How many times the underlying service must fail in the given window before the circuit opens. + , resetTimeoutSecs :: Int + -- ^ The window of time in which the underlying service must fail for the circuit to open. + , breakerDescription :: String + -- ^ Description that is attached to the failure so as to identify the particular circuit. + } + deriving (Eq, Show) -- | Defaulted options for the circuit breaker with 3 failures over 60 seconds. defaultCircuitBreakerOptions :: CircuitBreakerOptions -defaultCircuitBreakerOptions = CircuitBreakerOptions { maxBreakerFailures = 3, resetTimeoutSecs = 60, breakerDescription = "Circuit breaker open." } +defaultCircuitBreakerOptions = + CircuitBreakerOptions + { maxBreakerFailures = 3 + , resetTimeoutSecs = 60 + , breakerDescription = "Circuit breaker open." + } -- | Status indicating if the circuit is open. -data BreakerStatus = BreakerClosed Int | BreakerOpen Int deriving (Eq, Show) +data BreakerStatus + = BreakerClosed Int + | BreakerOpen Int + deriving (Eq, Show) -- | Representation of the state the circuit breaker is currently in. -data CircuitBreakerState = CircuitBreakerState [IORef BreakerStatus] +newtype CircuitBreakerState = CircuitBreakerState [IORef BreakerStatus] -- | Exception thrown when the circuit is open. -data CircuitBreakerException = CircuitBreakerException String deriving (Eq, Show, Typeable) +newtype CircuitBreakerException = CircuitBreakerException String + deriving (Eq, Show) + instance Exception CircuitBreakerException -- | Combines multiple states together. -combineCircuitBreakerStates :: (Foldable t) => t CircuitBreakerState -> CircuitBreakerState -combineCircuitBreakerStates states = CircuitBreakerState $ foldMap (\(CircuitBreakerState refs) -> refs) states +combineCircuitBreakerStates :: Foldable t => t CircuitBreakerState -> CircuitBreakerState +combineCircuitBreakerStates = CircuitBreakerState . foldMap (\(CircuitBreakerState refs) -> refs) -- | Determines if a specific status is open. isStatusOpen :: BreakerStatus -> Bool @@ -62,45 +75,55 @@ isStatusClosed (BreakerClosed _) = True -- | Determines if a circuit breaker is open. -isCircuitBreakerOpen :: (MonadBaseControl IO m) => CircuitBreakerState -> m Bool -isCircuitBreakerOpen (CircuitBreakerState states) = fmap or $ traverse (\ref -> fmap isStatusOpen $ readIORef ref) states +isCircuitBreakerOpen :: MonadIO m => CircuitBreakerState -> m Bool +isCircuitBreakerOpen (CircuitBreakerState states) = + or <$> traverse (fmap isStatusOpen . readIORef) states -- | Determines if a circuit breaker is closed. -isCircuitBreakerClosed :: (MonadBaseControl IO m) => CircuitBreakerState -> m Bool -isCircuitBreakerClosed (CircuitBreakerState states) = fmap and $ traverse (\ref -> fmap isStatusClosed $ readIORef ref) states +isCircuitBreakerClosed :: MonadIO m => CircuitBreakerState -> m Bool +isCircuitBreakerClosed (CircuitBreakerState states) = + and <$> traverse (fmap isStatusClosed . readIORef) states --- TODO: Check that values within m aren't lost on a successful call. -- | Circuit breaking services can be constructed with this function. -circuitBreaker :: (MonadBaseControl IO m, MonadBaseControl IO n) - => CircuitBreakerOptions -- ^ Options for specifying the circuit breaker behaviour. - -> BasicService m a b -- ^ Service to protect with the circuit breaker. - -> n (CircuitBreakerState, BasicService m a b) -circuitBreaker options service = - let retrieveTime = liftBase $ round `fmap` getPOSIXTime - failureMax = maxBreakerFailures options - callIfClosed request ref = bracketOnError (return ()) (\_ -> incErrors ref) (\_ -> service request) - canaryCall request ref = do - result <- callIfClosed request ref - writeIORef ref $ BreakerClosed 0 - return result - incErrors ref = do - currentTime <- retrieveTime - atomicModifyIORef' ref $ \status -> case status of - (BreakerClosed errorCount) -> (if errorCount >= failureMax then BreakerOpen (currentTime + (resetTimeoutSecs options)) else BreakerClosed (errorCount + 1), ()) - other -> (other, ()) - failingCall = throw $ CircuitBreakerException $ breakerDescription options - callIfOpen request ref = do - currentTime <- retrieveTime - canaryRequest <- atomicModifyIORef' ref $ \status -> case status of - (BreakerClosed _) -> (status, False) - (BreakerOpen time) -> if currentTime > time then ((BreakerOpen (currentTime + (resetTimeoutSecs options))), True) else (status, False) - if canaryRequest then canaryCall request ref else failingCall - breakerService ref request = do - status <- readIORef ref - case status of - (BreakerClosed _) -> callIfClosed request ref - (BreakerOpen _) -> callIfOpen request ref - - in do - ref <- newIORef $ BreakerClosed 0 - return (CircuitBreakerState [ref], breakerService ref) +circuitBreaker + :: (MonadUnliftIO m, MonadIO n) + => CircuitBreakerOptions + -- ^ Options for specifying the circuit breaker behaviour. + -> BasicService m a b + -- ^ Service to protect with the circuit breaker. + -> n (CircuitBreakerState, BasicService m a b) +circuitBreaker options service = do + let retrieveTime = liftIO $ round <$> getPOSIXTime + failureMax = maxBreakerFailures options + callIfClosed request ref = bracketOnError (pure ()) (\_ -> incErrors ref) (\_ -> service request) + canaryCall request ref = do + result <- callIfClosed request ref + writeIORef ref $ BreakerClosed 0 + pure result + incErrors ref = do + currentTime <- retrieveTime + atomicModifyIORef' ref $ \status -> case status of + BreakerClosed errorCount -> + ( if errorCount >= failureMax + then BreakerOpen (currentTime + resetTimeoutSecs options) + else BreakerClosed (errorCount + 1) + , () + ) + other -> (other, ()) + failingCall = throwIO $ CircuitBreakerException $ breakerDescription options + callIfOpen request ref = do + currentTime <- retrieveTime + canaryRequest <- atomicModifyIORef' ref $ \status -> case status of + BreakerClosed _ -> (status, False) + BreakerOpen time -> + if currentTime > time + then (BreakerOpen (currentTime + resetTimeoutSecs options), True) + else (status, False) + if canaryRequest then canaryCall request ref else failingCall + breakerService ref request = do + status <- readIORef ref + case status of + BreakerClosed _ -> callIfClosed request ref + BreakerOpen _ -> callIfOpen request ref + ref <- newIORef $ BreakerClosed 0 + pure (CircuitBreakerState [ref], breakerService ref) diff --git a/src/Glue/ConcurrencyLimit.hs b/src/Glue/ConcurrencyLimit.hs new file mode 100644 --- /dev/null +++ b/src/Glue/ConcurrencyLimit.hs @@ -0,0 +1,87 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Module supporting concurrent request constraints (bulkhead pattern), +-- limiting the number of in-flight and optionally queued requests. +module Glue.ConcurrencyLimit + ( ConcurrencyLimitOptions (..) + , ConcurrencyLimitExceededException (..) + , defaultConcurrencyLimitOptions + , limitConcurrency + ) where + +import Glue.Types +import UnliftIO (Exception, MonadIO, MonadUnliftIO) +import UnliftIO.Exception (bracketOnError, finally, throwIO) +import UnliftIO.STM (atomically, modifyTVar', newTVarIO, + readTVar, retrySTM, writeTVar) + +-- | Options for configuring concurrent request limits. +data ConcurrencyLimitOptions = ConcurrencyLimitOptions + { maxInFlight :: Int + -- ^ Maximum number of concurrent requests allowed to execute the underlying service simultaneously. + , maxQueued :: Maybe Int + -- ^ Optional maximum number of requests allowed to wait in the queue (`Nothing` for unbounded queue, `Just 0` for no queueing). + , concurrencyLimitDescription :: String + -- ^ Description attached to t'ConcurrencyLimitExceededException' when a request is rejected. + } + deriving (Eq, Show) + +-- | Default concurrency limit options allowing 10 in-flight requests and an unbounded queue. +defaultConcurrencyLimitOptions :: ConcurrencyLimitOptions +defaultConcurrencyLimitOptions = + ConcurrencyLimitOptions + { maxInFlight = 10 + , maxQueued = Nothing + , concurrencyLimitDescription = "Concurrency limit exceeded." + } + +-- | Exception thrown when both in-flight and queue limits have been reached. +newtype ConcurrencyLimitExceededException = ConcurrencyLimitExceededException String + deriving (Eq, Show) + +instance Exception ConcurrencyLimitExceededException + +-- | Wraps a service with a concurrency limit restricting in-flight and queued requests. +limitConcurrency + :: (MonadUnliftIO m, MonadIO n) + => ConcurrencyLimitOptions + -- ^ Options configuring the concurrency and queue bounds. + -> BasicService m a b + -- ^ Service to protect. + -> n (BasicService m a b) +limitConcurrency options service = do + let maxActive = max 1 (maxInFlight options) + inFlightVar <- newTVarIO (0 :: Int) + queuedVar <- newTVarIO (0 :: Int) + let releaseInFlight = atomically $ modifyTVar' inFlightVar (subtract 1) + releaseQueued = atomically $ modifyTVar' queuedVar (subtract 1) + waitForSlot = atomically $ do + inFlight <- readTVar inFlightVar + if inFlight < maxActive + then do + modifyTVar' queuedVar (subtract 1) + writeTVar inFlightVar (inFlight + 1) + else retrySTM + limitedService request = do + admission <- atomically $ do + inFlight <- readTVar inFlightVar + queued <- readTVar queuedVar + if inFlight < maxActive && queued == 0 + then do + writeTVar inFlightVar (inFlight + 1) + pure (Right True) + else case maxQueued options of + Just maxQ | queued >= maxQ -> pure (Left ()) + _ -> do + writeTVar queuedVar (queued + 1) + pure (Right False) + case admission of + Left () -> + throwIO $ ConcurrencyLimitExceededException (concurrencyLimitDescription options) + Right True -> + finally (service request) releaseInFlight + Right False -> do + bracketOnError (pure ()) (\_ -> releaseQueued) (\_ -> waitForSlot) + finally (service request) releaseInFlight + pure limitedService diff --git a/src/Glue/DogpileProtection.hs b/src/Glue/DogpileProtection.hs --- a/src/Glue/DogpileProtection.hs +++ b/src/Glue/DogpileProtection.hs @@ -1,57 +1,94 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} --- | Module supporting the dogpile protection of a service, see . -module Glue.DogpileProtection( - dogpileProtect -) where +-- | Module supporting the dogpile protection of a service, +-- see . +module Glue.DogpileProtection + ( DogpileOptions (..) + , defaultDogpileOptions + , dogpileProtectWith + , dogpileProtect + ) where -import Control.Concurrent -import Control.Exception -import Data.Hashable +import Control.Monad (void) +import Data.Hashable (Hashable) import qualified Data.HashMap.Strict as M -import Data.IORef import Glue.Types +import UnliftIO (MonadIO, MonadUnliftIO, SomeException, + liftIO) +import UnliftIO.Concurrent (forkIO, threadDelay) +import UnliftIO.Exception (onException, throwIO, toException, tryAny) +import UnliftIO.IORef (IORef, atomicModifyIORef', newIORef) +import UnliftIO.MVar (MVar, newEmptyMVar, putMVar, readMVar, + tryPutMVar) -data DogpileResult b = CachedValue (Either SomeException b) - | RequestInProgress (IO b) +-- | Options for configuring dogpile protection. +newtype DogpileOptions = DogpileOptions + { dogpileCacheTimeMs :: Int + -- ^ How long (in milliseconds) to retain a completed result for subsequent callers. + } + deriving (Eq, Show) -type ResultMap a b = M.HashMap a (DogpileResult b) +-- | Default options for dogpile protection, caching completed values for 5 seconds (5000ms). +defaultDogpileOptions :: DogpileOptions +defaultDogpileOptions = + DogpileOptions + { dogpileCacheTimeMs = 5000 + } -type ResultRef a b = IORef (ResultMap a b) +data DogpileResult b + = CachedValue (Either SomeException b) + | RequestInProgress (MVar (Either SomeException b)) -getDogpileResult :: DogpileResult b -> IO b -getDogpileResult (RequestInProgress requestWait) = requestWait -getDogpileResult (CachedValue result) = either throw return result +type ResultMap a b = M.HashMap a (DogpileResult b) +type ResultRef a b = IORef (ResultMap a b) -waitForMVar :: MVar (Either SomeException b) -> IO b -waitForMVar mvar = do - result <- readMVar mvar - either throw return result +getDogpileResult :: MonadIO m => DogpileResult b -> m b +getDogpileResult (RequestInProgress mvar) = readMVar mvar >>= either throwIO pure +getDogpileResult (CachedValue result) = either throwIO pure result -getProtectedResult :: (Eq a, Hashable a) - => a -> BasicService IO a b -> MVar (Either SomeException b) -> ResultRef a b -> ResultMap a b -> (ResultMap a b, IO b) -getProtectedResult request service mvar mapRef resultMap = - let mvarWait = waitForMVar mvar - invokeService = do - result <- try $ service request - putMVar mvar result - atomicModifyIORef' mapRef (\mapToUpdate -> (M.insert request (CachedValue result) mapToUpdate, ())) - forkIO $ do - threadDelay (5 * 1000 * 1000) - atomicModifyIORef' mapRef (\mapToUpdate -> (M.delete request mapToUpdate, ())) - inCache = M.lookup request resultMap - in maybe (M.insert request (RequestInProgress mvarWait) resultMap, invokeService >> mvarWait) (\r -> (resultMap, getDogpileResult r)) inCache +-- | Dogpile protection of a service with configurable options. +dogpileProtectWith + :: forall m n a b. + (MonadUnliftIO m, MonadIO n, Hashable a) + => DogpileOptions + -- ^ Options configuring dogpile protection. + -> BasicService m a b + -- ^ The service to protect. + -> n (BasicService m a b) +dogpileProtectWith options service = do + mapRef <- newIORef M.empty :: n (ResultRef a b) + let ttlMs = dogpileCacheTimeMs options + protectedService request = do + newVar <- newEmptyMVar + actionOrExisting <- atomicModifyIORef' mapRef $ \resultMap -> + case M.lookup request resultMap of + Just existing -> (resultMap, Right existing) + Nothing -> (M.insert request (RequestInProgress newVar) resultMap, Left newVar) + case actionOrExisting of + Right existing -> getDogpileResult existing + Left mvar -> do + let cleanupOnAbort = do + atomicModifyIORef' mapRef (\m -> (M.delete request m, ())) + void $ tryPutMVar mvar (Left (toException (userError "Dogpile leader aborted"))) + result <- tryAny (service request) `onException` cleanupOnAbort + putMVar mvar result + if ttlMs > 0 + then do + atomicModifyIORef' mapRef (\m -> (M.insert request (CachedValue result) m, ())) + void $ liftIO $ forkIO $ do + threadDelay (ttlMs * 1000) + atomicModifyIORef' mapRef (\m -> (M.delete request m, ())) + else + atomicModifyIORef' mapRef (\m -> (M.delete request m, ())) + either throwIO pure result + pure protectedService --- | Dogpile protection of a service, to prevent multiple calls for the same value being submitted. --- | Loses the values held within m. -dogpileProtect :: (Eq a, Hashable a) - => BasicService IO a b -- ^ The service to protect. - -> IO (BasicService IO a b) -dogpileProtect service = do - mapRef <- newIORef M.empty - let protectedService request = do - mvar <- newEmptyMVar - resultAction <- atomicModifyIORef' mapRef $ getProtectedResult request service mvar mapRef - resultAction - return protectedService +-- | Dogpile protection of a service using 'defaultDogpileOptions', +-- preventing multiple concurrent calls for the same value from being submitted. +dogpileProtect + :: (MonadUnliftIO m, MonadIO n, Hashable a) + => BasicService m a b + -- ^ The service to protect. + -> n (BasicService m a b) +dogpileProtect = dogpileProtectWith defaultDogpileOptions diff --git a/src/Glue/Failover.hs b/src/Glue/Failover.hs --- a/src/Glue/Failover.hs +++ b/src/Glue/Failover.hs @@ -2,36 +2,45 @@ {-# LANGUAGE ScopedTypeVariables #-} -- | Module supporting failover of a service by transforming the request each time a failure occurs up to a fixed number of attempts. -module Glue.Failover( - FailoverOptions +module Glue.Failover + ( FailoverOptions (..) , defaultFailoverOptions , failover - , maxFailovers - , transformFailoverRequest -) where + ) where -import Control.Exception.Lifted hiding (throw) -import Control.Monad.Trans.Control import Glue.Types +import UnliftIO (MonadUnliftIO) +import UnliftIO.Exception (catchAny) -- | Options for determining behaviour of failover services. -data FailoverOptions a = FailoverOptions { - maxFailovers :: Int -- ^ The maximum numer of times the service will failover and make another attempt. - , transformFailoverRequest :: a -> a -- ^ Each time the service performs a failover, transform the request with this function. -} +data FailoverOptions a = FailoverOptions + { maxFailovers :: Int + -- ^ The maximum number of times the service will failover and make another attempt. + , transformFailoverRequest :: a -> a + -- ^ Each time the service performs a failover, transform the request with this function. + } --- | Simple defaults that results in retrying 3 times in a dumb way immediately without transforming the request. +-- | Simple defaults that result in retrying 3 times immediately without transforming the request. defaultFailoverOptions :: FailoverOptions a -defaultFailoverOptions = FailoverOptions { maxFailovers = 3, transformFailoverRequest = id } +defaultFailoverOptions = + FailoverOptions + { maxFailovers = 3 + , transformFailoverRequest = id + } -- | Creates a service that can transform requests with each subsequent failure. -failover :: (MonadBaseControl IO m) - => FailoverOptions a -- ^ Instance of 'FailoverOptions' to configure the failover functionality. - -> BasicService m a b -- ^ The service to perform failover around. - -> BasicService m a b +failover + :: MonadUnliftIO m + => FailoverOptions a + -- ^ Instance of t'FailoverOptions' to configure the failover functionality. + -> BasicService m a b + -- ^ The service to perform failover around. + -> BasicService m a b failover options service = let invokeService failCount request = - let afterFail = (\(_ :: SomeException) -> invokeService (failCount + 1) ((transformFailoverRequest options) request)) - invoke = service request - in if failCount >= (maxFailovers options) then invoke else catch invoke afterFail + let afterFail _ = invokeService (failCount + 1) (transformFailoverRequest options request) + invoke = service request + in if failCount >= maxFailovers options + then invoke + else catchAny invoke afterFail in invokeService 0 diff --git a/src/Glue/Hedging.hs b/src/Glue/Hedging.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Hedging.hs @@ -0,0 +1,81 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Module supporting request hedging (duplication), dispatching multiple concurrent +-- requests (optionally staggered) and completing with the first successful result. +module Glue.Hedging + ( HedgingOptions (..) + , defaultHedgingOptions + , hedgingService + ) where + +import Control.Monad (when) +import Data.Maybe (fromMaybe) +import Glue.Types +import UnliftIO (Async, MonadUnliftIO, SomeException, + withAsync) +import UnliftIO.Concurrent (threadDelay) +import UnliftIO.Exception (throwIO, tryAny) +import UnliftIO.STM (atomically, newTVarIO, readTVar, retrySTM, + writeTVar) + +-- | Options for configuring hedged (duplicated) service calls. +data HedgingOptions = HedgingOptions + { hedgingCount :: Int + -- ^ Total number of concurrent attempts to make (including the primary call). + , hedgingDelayMs :: Int + -- ^ Stagger delay in milliseconds before each subsequent hedged call (`0` to fire all simultaneously). + } + deriving (Eq, Show) + +-- | Default hedging options that fire 2 concurrent requests with no stagger delay. +defaultHedgingOptions :: HedgingOptions +defaultHedgingOptions = + HedgingOptions + { hedgingCount = 2 + , hedgingDelayMs = 0 + } + +withAsyncs :: MonadUnliftIO m => [m a] -> ([Async a] -> m b) -> m b +withAsyncs [] k = k [] +withAsyncs (m:ms) k = withAsync m $ \a -> withAsyncs ms (\as -> k (a : as)) + +-- | Creates a service that duplicates requests up to 'hedgingCount' times and returns +-- the first successful response, automatically cancelling remaining in-flight attempts. +-- If all attempts fail, the first encountered exception is rethrown. +hedgingService + :: forall m a b. + MonadUnliftIO m + => HedgingOptions + -- ^ Options configuring the number of hedged calls and stagger delay. + -> BasicService m a b + -- ^ Underlying service to invoke. + -> BasicService m a b +hedgingService options service request + | hedgingCount options <= 1 = service request + | otherwise = do + let total = hedgingCount options + delayUs = max 0 (hedgingDelayMs options) * 1000 + stateVar <- newTVarIO (Nothing :: Maybe b, total, Nothing :: Maybe SomeException) + let runAttempt :: Int -> m () + runAttempt idx = do + when (idx > 0 && delayUs > 0) $ + threadDelay (idx * delayUs) + res <- tryAny (service request) + atomically $ do + (mWin, remaining, mErr) <- readTVar stateVar + case res of + Right val -> writeTVar stateVar (Just (fromMaybe val mWin), remaining - 1, mErr) + Left err -> writeTVar stateVar (mWin, remaining - 1, Just (fromMaybe err mErr)) + withAsyncs [runAttempt i | i <- [0 .. total - 1]] $ \_ -> do + outcome <- atomically $ do + (mWin, remaining, mErr) <- readTVar stateVar + case mWin of + Just val -> pure (Right val) + Nothing + | remaining <= 0 -> pure (Left mErr) + | otherwise -> retrySTM + case outcome of + Right val -> pure val + Left (Just err) -> throwIO err + Left Nothing -> service request diff --git a/src/Glue/LoadBalancer.hs b/src/Glue/LoadBalancer.hs new file mode 100644 --- /dev/null +++ b/src/Glue/LoadBalancer.hs @@ -0,0 +1,51 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Module providing round-robin load balancing across multiple services. +module Glue.LoadBalancer + ( NoServicesAvailableException (..) + , roundRobinLoadBalancer + , roundRobinLoadBalancerNE + ) where + +import Data.Foldable (toList) +import Data.List.NonEmpty (NonEmpty (..)) +import qualified Data.List.NonEmpty as NE +import Glue.Types +import UnliftIO (Exception, MonadIO) +import UnliftIO.Exception (throwIO) +import UnliftIO.IORef (atomicModifyIORef', newIORef) + +-- | Exception thrown when attempting to call a load balancer configured with an empty set of services. +data NoServicesAvailableException = NoServicesAvailableException + deriving (Eq, Show) + +instance Exception NoServicesAvailableException + +-- | Creates a service that distributes requests across a 'NonEmpty' list of backend services in round-robin order. +roundRobinLoadBalancerNE + :: (MonadIO m, MonadIO n) + => NonEmpty (BasicService m a b) + -- ^ Non-empty collection of backend services to balance across. + -> n (BasicService m a b) +roundRobinLoadBalancerNE services = do + let svcList = NE.toList services + count = length svcList + indexRef <- newIORef (0 :: Int) + let balanced request = do + idx <- atomicModifyIORef' indexRef (\i -> ((i + 1) `mod` count, i)) + let target = svcList !! idx + target request + pure balanced + +-- | Creates a service that distributes requests across a foldable collection of backend services in round-robin order. +-- If the collection is empty, invoking the returned service throws t'NoServicesAvailableException'. +roundRobinLoadBalancer + :: (MonadIO m, MonadIO n, Foldable t) + => t (BasicService m a b) + -- ^ Collection of backend services to balance across. + -> n (BasicService m a b) +roundRobinLoadBalancer services = + case toList services of + [] -> pure $ const $ throwIO NoServicesAvailableException + (x:xs) -> roundRobinLoadBalancerNE (x :| xs) diff --git a/src/Glue/Preload.hs b/src/Glue/Preload.hs --- a/src/Glue/Preload.hs +++ b/src/Glue/Preload.hs @@ -1,109 +1,148 @@ {-# LANGUAGE BangPatterns #-} -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} -- | Module containing a form of caching where values for given keys are preloaded ahead of time. --- | Once warmed up requests for preloaded keys will be instant, with the values refreshed in the background. -module Glue.Preload( - PreloadedOptions +-- Once warmed up, requests for preloaded keys will be instant, with the values refreshed in the background. +module Glue.Preload + ( PreloadedOptions (..) + , PreloadStoppedBeforeExecutedException (..) , defaultPreloadedOptions , preloadingService , preloadingCall - , preloadedKeys - , preloadingRefreshTimeMs - , preloadingRun -) where + ) where -import Control.Concurrent.Lifted -import Control.Exception.Base hiding (throwIO) -import Control.Exception.Lifted -import Control.Monad.IO.Class -import Control.Monad.Trans.Control -import Data.Hashable -import qualified Data.HashMap.Strict as M -import qualified Data.HashSet as S -import Data.IORef.Lifted -import Data.Maybe -import Data.Typeable +import Control.Monad (void) +import Data.Hashable (Hashable) +import qualified Data.HashMap.Strict as M +import qualified Data.HashSet as S +import Data.Maybe (fromJust) import Glue.Types +import UnliftIO (Exception, MonadIO, MonadUnliftIO, + SomeException (..), liftIO) +import UnliftIO.Async (race_) +import UnliftIO.Concurrent (forkIO, threadDelay) +import UnliftIO.Exception (throwIO) +import UnliftIO.IORef (IORef, atomicModifyIORef', newIORef, + readIORef) +import UnliftIO.MVar (MVar, newEmptyMVar, readMVar, tryPutMVar) -- | Options for determining behaviour of preloading services. -data PreloadedOptions m a b = PreloadedOptions { - preloadedKeys :: S.HashSet a, -- ^ Keys to preload. - preloadingRefreshTimeMs :: Int, -- ^ Amount of time between refreshes. - preloadingRun :: MToIO m -- ^ Get an IO of the response for the caching. -} +data PreloadedOptions m a b = PreloadedOptions + { preloadedKeys :: S.HashSet a + -- ^ Keys to preload. + , preloadingRefreshTimeMs :: Int + -- ^ Amount of time between refreshes. + , preloadingRun :: MToIO m + -- ^ Get an 'IO' of the response for the caching. + } --- | Defaulted options for preloading a HashSet of keys with a 30 second refresh time. +-- | Defaulted options for preloading a 'S.HashSet' of keys with a 30 second refresh time. defaultPreloadedOptions :: S.HashSet a -> MToIO m -> PreloadedOptions m a b -defaultPreloadedOptions toPreload pRun = PreloadedOptions { - preloadedKeys = toPreload, - preloadingRefreshTimeMs = 30 * 1000, - preloadingRun = pRun -} +defaultPreloadedOptions toPreload pRun = + PreloadedOptions + { preloadedKeys = toPreload + , preloadingRefreshTimeMs = 30 * 1000 + , preloadingRun = pRun + } -data PreloadedState a b = PreloadedStarted - | PreloadedWithResult (FailOrSuccess a b) Bool +data PreloadedState a b + = PreloadedStarted + | PreloadedWithResult (FailOrSuccess a b) Bool -data PreloadStoppedBeforeExecutedException = PreloadStoppedBeforeExecutedException deriving (Eq, Show, Typeable) +-- | Exception thrown when preloading is stopped before the initial preload has executed. +data PreloadStoppedBeforeExecutedException = PreloadStoppedBeforeExecutedException + deriving (Eq, Show) + instance Exception PreloadStoppedBeforeExecutedException -modifyStateWithResult :: forall a b . Either SomeException (MultiGetResponse a b) -> PreloadedState a b -> (PreloadedState a b, Bool) -modifyStateWithResult result PreloadedStarted = (PreloadedWithResult result True, True) -modifyStateWithResult result (PreloadedWithResult _ c) = (PreloadedWithResult result c, c) +modifyStateWithResult :: Either SomeException (MultiGetResponse a b) -> PreloadedState a b -> (PreloadedState a b, Bool) +modifyStateWithResult result PreloadedStarted = (PreloadedWithResult result True, True) +modifyStateWithResult result (PreloadedWithResult _ c) = (PreloadedWithResult result c, c) -applyResultToState :: forall m a b . MonadBaseControl IO m => IORef (PreloadedState a b) -> FailOrSuccess a b -> m Bool -applyResultToState stateRef result = atomicModifyIORef' stateRef $ modifyStateWithResult result +applyResultToState :: MonadIO m => IORef (PreloadedState a b) -> MVar () -> FailOrSuccess a b -> m Bool +applyResultToState stateRef readyVar result = do + continue <- atomicModifyIORef' stateRef $ modifyStateWithResult result + void $ tryPutMVar readyVar () + pure continue -waitForResult :: forall m a b . (MonadBaseControl IO m) => IORef (PreloadedState a b) -> m (MultiGetResponse a b) -waitForResult stateRef = do - state <- readIORef stateRef - let tryAgainLater = threadDelay 1000 >> waitForResult stateRef - case state of - PreloadedStarted -> tryAgainLater - PreloadedWithResult (Right success) _ -> return success - PreloadedWithResult (Left failure) _ -> throwIO failure +waitForResult :: MonadIO m => IORef (PreloadedState a b) -> MVar () -> m (MultiGetResponse a b) +waitForResult stateRef readyVar = do + readMVar readyVar + state <- readIORef stateRef + case state of + PreloadedStarted -> throwIO PreloadStoppedBeforeExecutedException + PreloadedWithResult (Right success) _ -> pure success + PreloadedWithResult (Left failure) _ -> throwIO failure markAsFinished :: PreloadedState a b -> (PreloadedState a b, ()) -markAsFinished PreloadedStarted = (PreloadedWithResult (Left (SomeException PreloadStoppedBeforeExecutedException)) False, ()) +markAsFinished PreloadedStarted = (PreloadedWithResult (Left (SomeException PreloadStoppedBeforeExecutedException)) False, ()) markAsFinished (PreloadedWithResult r _) = (PreloadedWithResult r False, ()) -- | Preloads the results of calls for given keys. -preloadingService :: forall m n a b . (MonadIO m, MonadIO n, MonadBaseControl IO m, MonadBaseControl IO n, Eq a, Hashable a, Show a) - => PreloadedOptions m a b -- ^ Instance of 'PreloadedOptions' to configure the preloading functionality. - -> MultiGetService m a b -- ^ The service to perform preloading of. - -> n (MultiGetService m a b, () -> n ()) +preloadingService + :: forall m n a b. + (MonadUnliftIO m, MonadIO n, Hashable a) + => PreloadedOptions m a b + -- ^ Instance of t'PreloadedOptions' to configure the preloading functionality. + -> MultiGetService m a b + -- ^ The service to perform preloading of. + -> n (MultiGetService m a b, () -> n ()) preloadingService PreloadedOptions{..} service = do !stateIORef <- newIORef PreloadedStarted - let stop _ = atomicModifyIORef' stateIORef markAsFinished + !readyVar <- newEmptyMVar + !stopVar <- newEmptyMVar + let stop _ = do + atomicModifyIORef' stateIORef markAsFinished + void $ tryPutMVar readyVar () + void $ tryPutMVar stopVar () let runUpdate = do - result <- makeCall service preloadedKeys - applyResultToState stateIORef result + result <- makeCall service preloadedKeys + applyResultToState stateIORef readyVar result let updatePreloaded = do - continue <- liftIO $ preloadingRun $ runUpdate - if continue then threadDelay (preloadingRefreshTimeMs * 1000) >> updatePreloaded else return () + continue <- liftIO $ preloadingRun runUpdate + if continue + then do + stopped <- liftIO $ raceWait stopVar (preloadingRefreshTimeMs * 1000) + if stopped then pure () else updatePreloaded + else pure () let plService request = do - let fromPreloadKeys = S.intersection request preloadedKeys - let fromServiceKeys = S.difference request preloadedKeys - !fromPreload <- if S.null fromPreloadKeys then return M.empty else fmap (M.filterWithKey (\k -> \_ -> S.member k fromPreloadKeys)) $ waitForResult stateIORef - !fromService <- if S.null fromServiceKeys then return M.empty else service fromServiceKeys - return $ M.union fromService fromPreload - _ <- fork updatePreloaded - return (plService, stop) + let fromPreloadKeys = S.intersection request preloadedKeys + let fromServiceKeys = S.difference request preloadedKeys + !fromPreload <- + if S.null fromPreloadKeys + then pure M.empty + else M.filterWithKey (\k _ -> S.member k fromPreloadKeys) <$> waitForResult stateIORef readyVar + !fromService <- + if S.null fromServiceKeys + then pure M.empty + else service fromServiceKeys + pure $ M.union fromService fromPreload + void $ liftIO $ forkIO updatePreloaded + pure (plService, stop) + where + raceWait :: MVar () -> Int -> IO Bool + raceWait stopVar delayUs = do + race_ (readMVar stopVar) (threadDelay delayUs) + not <$> tryPutMVar stopVar () -- | Preloads a single parameter-less call. -preloadingCall :: forall m n b . (MonadIO m, MonadIO n, MonadBaseControl IO m, MonadBaseControl IO n) - => MToIO m -- ^ Get an IO of the call for the caching. - -> Int -- ^ Amount of time between refreshes. - -> m b -- ^ The call to perform preloading of. - -> n (m b, () -> n ()) +preloadingCall + :: forall m n b. + (MonadUnliftIO m, MonadIO n) + => MToIO m + -- ^ Get an 'IO' of the call for the caching. + -> Int + -- ^ Amount of time between refreshes. + -> m b + -- ^ The call to perform preloading of. + -> n (m b, () -> n ()) preloadingCall toIO timeBetweenRefreshes call = do let multiGetService = basicToMultiGet (\_ -> call) :: MultiGetService m () b let options = PreloadedOptions (S.singleton ()) timeBetweenRefreshes toIO (preloading, stop) <- preloadingService options multiGetService let preloadingBasicService = multiGetToBasic preloading - let preloadingCallResult = fmap (fromJust) $ preloadingBasicService () - return (preloadingCallResult, stop) + let preloadingCallResult = fromJust <$> preloadingBasicService () + pure (preloadingCallResult, stop) diff --git a/src/Glue/Retry.hs b/src/Glue/Retry.hs --- a/src/Glue/Retry.hs +++ b/src/Glue/Retry.hs @@ -1,56 +1,72 @@ -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} -- | Module containing retry functionality, allowing the construction of services that attempt multiple times in case of transient failure. -module Glue.Retry( - RetryOptions +module Glue.Retry + ( RetryOptions (..) , defaultRetryOptions , retryingService - , retryAllowed - , retryInitialWaitTimeMs - , maximumRetries - , retryWaitTimeMultiplier -) where + ) where -import Control.Concurrent.Lifted -import Control.Exception.Lifted -import Control.Monad.Trans.Control +import Control.Monad (when) import Glue.Types +import UnliftIO (MonadUnliftIO, SomeAsyncException, + SomeException, fromException) +import UnliftIO.Concurrent (threadDelay) +import UnliftIO.Exception (catch, throwIO) -- | Options for determining behaviour of retrying services. -data RetryOptions a = RetryOptions { - retryAllowed :: a -> Bool, -- ^ Predicate for determining if we can retry a call, can be used to prevent retries on non-idempotent operations. - retryInitialWaitTimeMs :: Int, -- ^ Amount of time to wait after the first failure. - maximumRetries :: Int, -- ^ The upper bound on how many attempts to make when invoking the service. - retryWaitTimeMultiplier :: Double -- ^ How much to multiply 'retryInitialWaitTimeMs' by for each number of times the service has retried. -} +data RetryOptions a = RetryOptions + { retryAllowed :: a -> Bool + -- ^ Predicate for determining if we can retry a call, can be used to prevent retries on non-idempotent operations. + , retryInitialWaitTimeMs :: Int + -- ^ Amount of time (in milliseconds) to wait after the first failure. + , maximumRetries :: Int + -- ^ The upper bound on how many retry attempts to make when invoking the service. + , retryWaitTimeMultiplier :: Double + -- ^ How much to multiply 'retryInitialWaitTimeMs' by for each subsequent retry. + } -- | Defaulted options for retrying 3 times with no wait time. defaultRetryOptions :: RetryOptions a -defaultRetryOptions = RetryOptions { - retryAllowed = (\_ -> True) - , retryInitialWaitTimeMs = 0 - , maximumRetries = 3 - , retryWaitTimeMultiplier = 0 - } +defaultRetryOptions = + RetryOptions + { retryAllowed = const True + , retryInitialWaitTimeMs = 0 + , maximumRetries = 3 + , retryWaitTimeMultiplier = 0 + } possibleAsyncException :: SomeException -> Maybe SomeAsyncException possibleAsyncException = fromException -- | Retries a call to a service multiple times, potentially backing off wait times between subsequent calls. --- | Asynchronous exceptions don't result in a retry, they are immediately rethrown. -retryingService :: (MonadBaseControl IO m) - => RetryOptions a -- ^ Instance of 'RetryOptions' to configure the retry functionality. - -> BasicService m a b -- ^ The service to perform retries of. - -> BasicService m a b +-- Asynchronous exceptions don't result in a retry; they are immediately rethrown. +retryingService + :: forall m a b. + MonadUnliftIO m + => RetryOptions a + -- ^ Instance of t'RetryOptions' to configure the retry functionality. + -> BasicService m a b + -- ^ The service to perform retries of. + -> BasicService m a b retryingService options service = - let catchHandler rc r e = case possibleAsyncException e of - Just ae -> throw ae - Nothing -> wait (rc + 1) >> attempt (rc + 1) r - attempt retryCount request = if (retryAllowed options) request && maxRetries > retryCount - then catch (service request) (catchHandler retryCount request) - else service request - maxRetries = maximumRetries options - wait retryCount = threadDelay $ round $ fromIntegral (retryInitialWaitTimeMs options) * ((retryWaitTimeMultiplier options) ^ retryCount) + let catchHandler rc r e = + case possibleAsyncException e of + Just ae -> throwIO ae + Nothing -> wait rc >> attempt (rc + 1) r + attempt retryCount request = + if retryAllowed options request && maxRetries > retryCount + then catch (service request) (catchHandler retryCount request) + else service request + maxRetries = maximumRetries options + initialUs = retryInitialWaitTimeMs options * 1000 + multiplier = + if retryWaitTimeMultiplier options <= 0 + then 1.0 + else retryWaitTimeMultiplier options + wait :: Int -> m () + wait retryCount = + when (initialUs > 0) $ + threadDelay $ round $ fromIntegral initialUs * (multiplier ^ retryCount) in attempt 0 diff --git a/src/Glue/Shutdown.hs b/src/Glue/Shutdown.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Shutdown.hs @@ -0,0 +1,128 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Module supporting graceful service shutdown, rejecting new requests with either +-- a fallback value or t'ServiceShuttingDownException', waiting for in-flight requests +-- to drain within a configurable grace period, and cancelling any remaining calls if +-- the grace period expires. +module Glue.Shutdown + ( ShutdownOptions (..) + , ServiceShuttingDownException (..) + , ShutdownHandle (..) + , defaultShutdownOptions + , gracefulShutdown + ) where + +import Glue.Types +import UnliftIO (Exception, MonadIO, MonadUnliftIO) +import UnliftIO.Async (race) +import UnliftIO.Exception (finally, throwIO) +import UnliftIO.STM (atomically, modifyTVar', newTVarIO, + readTVar, retrySTM, writeTVar) +import UnliftIO.Timeout (timeout) + +-- | Options for configuring graceful service shutdown. +data ShutdownOptions b = ShutdownOptions + { shutdownFallback :: Maybe b + -- ^ Optional default value returned for rejected requests after shutdown is initiated. + -- If 'Nothing', t'ServiceShuttingDownException' is thrown instead. + , shutdownGracePeriodMs :: Int + -- ^ Maximum time in milliseconds to wait for existing in-flight requests to complete before cancelling them. + , shutdownDescription :: String + -- ^ Description attached to t'ServiceShuttingDownException'. + } + deriving (Eq, Show) + +-- | Default shutdown options: throws t'ServiceShuttingDownException' on rejected calls and waits up to 5 seconds for in-flight calls. +defaultShutdownOptions :: ShutdownOptions b +defaultShutdownOptions = + ShutdownOptions + { shutdownFallback = Nothing + , shutdownGracePeriodMs = 5000 + , shutdownDescription = "Service is shutting down." + } + +-- | Exception thrown when a service is shutting down and no 'shutdownFallback' is configured. +newtype ServiceShuttingDownException = ServiceShuttingDownException String + deriving (Eq, Show) + +instance Exception ServiceShuttingDownException + +-- | Handle used to control and inspect a service wrapped with 'gracefulShutdown'. +data ShutdownHandle m = ShutdownHandle + { initiateShutdown :: m Bool + -- ^ Initiates shutdown: immediately rejects new requests, waits up to 'shutdownGracePeriodMs' + -- for in-flight requests to finish, and cancels any remaining calls if the grace period expires. + -- Returns 'True' if all in-flight calls drained cleanly within the grace period, or 'False' if forced cancellation occurred. + , isShuttingDown :: m Bool + -- ^ Returns 'True' if shutdown has been initiated. + , inFlightCount :: m Int + -- ^ Returns the current number of in-flight requests executing the service. + } + +-- | Wraps a service with graceful shutdown support. +gracefulShutdown + :: forall m n a b. + (MonadUnliftIO m, MonadIO n) + => ShutdownOptions b + -- ^ Options configuring fallback behaviour and grace period. + -> BasicService m a b + -- ^ Service to wrap. + -> n (BasicService m a b, ShutdownHandle m) +gracefulShutdown options service = do + shuttingDownVar <- newTVarIO False + forceCancelVar <- newTVarIO False + inFlightVar <- newTVarIO (0 :: Int) + + let rejectRequest :: m b + rejectRequest = + case shutdownFallback options of + Just fallback -> pure fallback + Nothing -> throwIO $ ServiceShuttingDownException (shutdownDescription options) + + waitForForceCancel :: m () + waitForForceCancel = atomically $ do + cancelled <- readTVar forceCancelVar + if cancelled then pure () else retrySTM + + waitDrained :: m () + waitDrained = atomically $ do + active <- readTVar inFlightVar + if active <= 0 then pure () else retrySTM + + wrappedService :: BasicService m a b + wrappedService request = do + accepted <- atomically $ do + shuttingDown <- readTVar shuttingDownVar + if shuttingDown + then pure False + else do + modifyTVar' inFlightVar (+ 1) + pure True + if not accepted + then rejectRequest + else do + res <- + race waitForForceCancel (service request) + `finally` atomically (modifyTVar' inFlightVar (subtract 1)) + case res of + Right val -> pure val + Left () -> rejectRequest + + handle = + ShutdownHandle + { initiateShutdown = do + atomically $ writeTVar shuttingDownVar True + let graceUs = max 0 (shutdownGracePeriodMs options) * 1000 + drained <- timeout graceUs waitDrained + case drained of + Just () -> pure True + Nothing -> do + atomically $ writeTVar forceCancelVar True + waitDrained + pure False + , isShuttingDown = atomically $ readTVar shuttingDownVar + , inFlightCount = atomically $ readTVar inFlightVar + } + + pure (wrappedService, handle) diff --git a/src/Glue/Switching.hs b/src/Glue/Switching.hs --- a/src/Glue/Switching.hs +++ b/src/Glue/Switching.hs @@ -1,26 +1,28 @@ {-# LANGUAGE BangPatterns #-} -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} -- | Module containing a switchable service, allowing the possibility of changing --- | a service "live". -module Glue.Switching( - switchingService -) where +-- a service "live". +module Glue.Switching + ( switchingService + ) where -import Control.Monad.Base -import Data.IORef.Lifted import Glue.Types +import UnliftIO (MonadIO) +import UnliftIO.IORef (newIORef, readIORef, writeIORef) -- | Provides a switchable service. -switchingService :: (MonadBase IO m, MonadBase IO n) => MultiGetService m a b -- ^ The service to initialise the switching service with. - -> n (MultiGetService m a b, (MultiGetService m a b) -> n ()) +-- Works for both 'BasicService' and 'MultiGetService' (since 'MultiGetService' is a 'BasicService'). +switchingService + :: (MonadIO m, MonadIO n) + => BasicService m a b + -- ^ The service to initialise the switching service with. + -> n (BasicService m a b, BasicService m a b -> n ()) switchingService service = do !serviceRef <- newIORef service let switching rs = do - serviceToUse <- readIORef serviceRef - serviceToUse rs + serviceToUse <- readIORef serviceRef + serviceToUse rs let switchingUpdate newService = writeIORef serviceRef newService - return (switching, switchingUpdate) + pure (switching, switchingUpdate) diff --git a/src/Glue/Timeout.hs b/src/Glue/Timeout.hs --- a/src/Glue/Timeout.hs +++ b/src/Glue/Timeout.hs @@ -1,45 +1,50 @@ -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} -- | Module supporting adding timeouts to a given service. -module Glue.Timeout( - TimeoutOptions - , TimeoutException(..) +module Glue.Timeout + ( TimeoutOptions (..) + , TimeoutException (..) , defaultTimeoutOptions , addTimeout - , timeoutDescription - , timeoutLimitMs -) where + ) where -import Control.Concurrent.Lifted -import Control.Exception.Lifted -import Control.Monad.Trans.Control -import Data.Typeable import Glue.Types +import UnliftIO (Exception, MonadUnliftIO) +import UnliftIO.Exception (throwIO) +import UnliftIO.Timeout (timeout) -- | Options for determining behaviour of services with a timeout. -data TimeoutOptions = TimeoutOptions { - timeoutDescription :: String -- ^ Description added to the 'TimeoutException' thrown when the timeout is exceeded. - , timeoutLimitMs :: Int -- ^ Timeout in milliseconds. -} +data TimeoutOptions = TimeoutOptions + { timeoutDescription :: String + -- ^ Description added to the t'TimeoutException' thrown when the timeout is exceeded. + , timeoutLimitMs :: Int + -- ^ Timeout in milliseconds. + } + deriving (Eq, Show) --- | Default instance of 'TimeoutOptions' with a timeout of 30 seconds. +-- | Default instance of t'TimeoutOptions' with a timeout of 30 seconds. defaultTimeoutOptions :: TimeoutOptions -defaultTimeoutOptions = TimeoutOptions { timeoutDescription = "Service call timed out.", timeoutLimitMs = 30000 } +defaultTimeoutOptions = + TimeoutOptions + { timeoutDescription = "Service call timed out." + , timeoutLimitMs = 30000 + } -- | Exception thrown when the timeout is exceeded. -data TimeoutException = TimeoutException String deriving (Eq, Show, Typeable) +newtype TimeoutException = TimeoutException String + deriving (Eq, Show) + instance Exception TimeoutException -- | Function for producing services protected with a timeout. -addTimeout :: (MonadBaseControl IO m) - => TimeoutOptions -- ^ Options to configure the timeout. - -> BasicService m a b -- ^ Service to protect with a timeout. - -> BasicService m a b -addTimeout options service = (\request -> do - currentThreadId <- myThreadId - timeoutThreadId <- fork $ do - threadDelay (1000 * timeoutLimitMs options) - throwTo currentThreadId (TimeoutException $ timeoutDescription options) - finally (service request) (killThread timeoutThreadId)) +addTimeout + :: MonadUnliftIO m + => TimeoutOptions + -- ^ Options to configure the timeout. + -> BasicService m a b + -- ^ Service to protect with a timeout. + -> BasicService m a b +addTimeout options service request = do + maybeResult <- timeout (1000 * timeoutLimitMs options) (service request) + maybe (throwIO $ TimeoutException $ timeoutDescription options) pure maybeResult diff --git a/test/Glue/BatcherSpec.hs b/test/Glue/BatcherSpec.hs --- a/test/Glue/BatcherSpec.hs +++ b/test/Glue/BatcherSpec.hs @@ -1,10 +1,8 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.BatcherSpec where +module Glue.BatcherSpec (spec) where -import Control.Concurrent.Async +import Control.Concurrent.Async (mapConcurrently) import qualified Data.HashMap.Strict as M import qualified Data.HashSet as S import Glue.Batcher @@ -15,12 +13,19 @@ spec :: Spec spec = do describe "batchingService" $ do - it "Requests should receive their full results" $ do + it "Multi-get requests should receive their full results" $ property $ \(requests :: [[Int]]) -> do - let requestsAsSets = fmap (S.fromList) requests - let serviceFunctionality rs = M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList rs - let service = (return . serviceFunctionality) :: MultiGetService IO Int Int - (_, multiService) <- batchingService defaultBatchingOptions service + let requestsAsSets = fmap S.fromList requests + serviceFunctionality rs = M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList rs + service = (pure . serviceFunctionality) :: MultiGetService IO Int Int + (_, multiService) <- batchingService (BatchingOptions 2) service let expectedResults = fmap serviceFunctionality requestsAsSets - (mapConcurrently multiService requestsAsSets) `shouldReturn` expectedResults + mapConcurrently multiService requestsAsSets `shouldReturn` expectedResults + it "Single-key requests should receive their individual lookup results" $ + property $ \(requests :: [Int]) -> do + let serviceFunctionality rs = M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList rs + service = (pure . serviceFunctionality) :: MultiGetService IO Int Int + (singleService, _) <- batchingService (BatchingOptions 2) service + let expectedResults = fmap (\r -> Just (r * 2)) requests + mapConcurrently singleService requests `shouldReturn` expectedResults diff --git a/test/Glue/CachingSpec.hs b/test/Glue/CachingSpec.hs --- a/test/Glue/CachingSpec.hs +++ b/test/Glue/CachingSpec.hs @@ -1,22 +1,20 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.CachingSpec where +module Glue.CachingSpec (spec) where -import Control.Exception.Base hiding (throw, throwIO) -import Control.Exception.Lifted import qualified Data.HashMap.Strict as M import qualified Data.HashSet as S import Data.IORef -import Data.Typeable import Glue.Caching import Glue.Types import Test.Hspec import Test.QuickCheck import Test.QuickCheck.Instances () +import UnliftIO.Exception (Exception, throwIO) -data CachingTestException = CachingTestException Int deriving (Eq, Show, Typeable) +newtype CachingTestException = CachingTestException Int + deriving (Eq, Show) + instance Exception CachingTestException type TestResp = MultiGetResponse Int Int @@ -24,35 +22,35 @@ spec :: Spec spec = do describe "cacheWithBasic" $ do - it "For a second request, the value comes from the cache" $ do + it "For a second request, the value comes from the cache" $ property $ \(request :: Int, result :: Int) -> do ref <- newIORef (0 :: Int) let service req = do - if req == request then return () else throwIO (CachingTestException (-1)) - callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) - if callCount > 1 then throwIO (CachingTestException callCount) else return result + if req == request then pure () else throwIO (CachingTestException (-1)) + callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) + if callCount > 1 then throwIO (CachingTestException callCount) else pure result cache <- newIORef M.empty - let lookupWith r = fmap (M.lookup r) $ readIORef cache + let lookupWith r = M.lookup r <$> readIORef cache let insertWith req resp = atomicModifyIORef' cache (\c -> (M.insert req resp c, ())) let cachedService = cacheWithBasic lookupWith insertWith service - (cachedService request) `shouldReturn` result - (cachedService request) `shouldReturn` result + cachedService request `shouldReturn` result + cachedService request `shouldReturn` result describe "cacheWithMulti" $ do - it "For a second request, the value comes from the cache" $ do + it "For a second request, the value comes from the cache" $ property $ \(result :: TestResp) -> do let request = S.fromList $ M.keys result ref <- newIORef (0 :: Int) let service req = do - if req == request then return () else throwIO (CachingTestException (-1)) - callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) - if callCount > 1 then throwIO (CachingTestException callCount) else return result + if req == request then pure () else throwIO (CachingTestException (-1)) + callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) + if callCount > 1 then throwIO (CachingTestException callCount) else pure result cache <- newIORef M.empty - let lookupWith rs = fmap (M.filterWithKey (\k -> \_ -> S.member k rs)) $ readIORef cache + let lookupWith rs = M.filterWithKey (\k _ -> S.member k rs) <$> readIORef cache let insertWith resp = atomicModifyIORef' cache (\c -> (M.union resp c, ())) let cachedService = cacheWithMulti lookupWith insertWith service - (cachedService request) `shouldReturn` result - (cachedService request) `shouldReturn` result - it "Merges the cached values as appropriate from subsequent requests" $ do + cachedService request `shouldReturn` result + cachedService request `shouldReturn` result + it "Merges the cached values as appropriate from subsequent requests" $ property $ \(first :: TestResp, second :: TestResp, both :: TestResp) -> do let uniqueFirst = M.difference first (M.union second both) let uniqueSecond = M.difference second (M.union first both) @@ -64,15 +62,15 @@ let secondRequest = S.fromList $ M.keys secondResult ref <- newIORef (0 :: Int) let service req = do - callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) - if callCount > 2 then throwIO (CachingTestException callCount) else return $ M.filterWithKey (\k -> \_ -> S.member k req) uniqueAll + callCount <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) + if callCount > 2 + then throwIO (CachingTestException callCount) + else pure $ M.filterWithKey (\k _ -> S.member k req) uniqueAll cache <- newIORef M.empty - let lookupWith rs = fmap (M.filterWithKey (\k -> \_ -> S.member k rs)) $ readIORef cache + let lookupWith rs = M.filterWithKey (\k _ -> S.member k rs) <$> readIORef cache let insertWith resp = atomicModifyIORef' cache (\c -> (M.union resp c, ())) let cachedService = cacheWithMulti lookupWith insertWith service - (cachedService firstRequest) `shouldReturn` firstResult - (cachedService secondRequest) `shouldReturn` secondResult - (cachedService firstRequest) `shouldReturn` firstResult - (cachedService secondRequest) `shouldReturn` secondResult - - + cachedService firstRequest `shouldReturn` firstResult + cachedService secondRequest `shouldReturn` secondResult + cachedService firstRequest `shouldReturn` firstResult + cachedService secondRequest `shouldReturn` secondResult diff --git a/test/Glue/CircuitBreakerSpec.hs b/test/Glue/CircuitBreakerSpec.hs --- a/test/Glue/CircuitBreakerSpec.hs +++ b/test/Glue/CircuitBreakerSpec.hs @@ -1,47 +1,58 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.CircuitBreakerSpec where +module Glue.CircuitBreakerSpec (spec) where -import Control.Exception.Base hiding (throw, throwIO, try) -import Control.Exception.Lifted import Data.IORef -import Data.Typeable import Glue.CircuitBreaker import Test.Hspec import Test.QuickCheck +import UnliftIO.Exception (Exception, throwIO, try) -data CircuitBreakerTestException = CircuitBreakerTestException deriving (Eq, Show, Typeable) +data CircuitBreakerTestException = CircuitBreakerTestException + deriving (Eq, Show) + instance Exception CircuitBreakerTestException requests :: [Int] -requests = [1..10] +requests = [1 .. 10] spec :: Spec spec = do describe "circuitBreaker" $ do - it "Multiple failures prevent subsequent calls" $ do + it "Multiple failures prevent subsequent calls" $ property $ \failureMax -> do - let positiveFailureMax = (abs failureMax) `mod` 5 - let options = defaultCircuitBreakerOptions { maxBreakerFailures = positiveFailureMax } - ref <- newIORef (0 :: Int) - let service _ = atomicModifyIORef' ref (\c -> (c + 1, ())) >> throwIO CircuitBreakerTestException :: IO Int - (s, circuitBreakerService) <- circuitBreaker options service - results <- traverse (\req -> try $ try $ circuitBreakerService req) requests :: IO [Either CircuitBreakerException (Either CircuitBreakerTestException Int)] - let expectedResults = (replicate (positiveFailureMax + 1) (Right $ Left $ CircuitBreakerTestException)) ++ (replicate (10 - positiveFailureMax - 1) (Left $ CircuitBreakerException "Circuit breaker open.")) + let positiveFailureMax = abs failureMax `mod` 5 + options = defaultCircuitBreakerOptions { maxBreakerFailures = positiveFailureMax } + ref <- newIORef (0 :: Int) + let service _ = atomicModifyIORef' ref (\c -> (c + 1, ())) >> throwIO CircuitBreakerTestException :: IO Int + (s, circuitBreakerService) <- circuitBreaker options service + results <- traverse (try . try . circuitBreakerService) requests :: IO [Either CircuitBreakerException (Either CircuitBreakerTestException Int)] + let expectedResults = replicate (positiveFailureMax + 1) (Right $ Left CircuitBreakerTestException) + ++ replicate (10 - positiveFailureMax - 1) (Left $ CircuitBreakerException "Circuit breaker open.") results `shouldBe` expectedResults - (isCircuitBreakerOpen s) `shouldReturn` True - (isCircuitBreakerClosed s) `shouldReturn` False - (readIORef ref) `shouldReturn` (positiveFailureMax + 1) - it "Successful calls pass straight through" $ do + isCircuitBreakerOpen s `shouldReturn` True + isCircuitBreakerClosed s `shouldReturn` False + readIORef ref `shouldReturn` (positiveFailureMax + 1) + + it "Successful calls pass straight through" $ property $ \(failureMax :: Int, requestIDs :: [Int]) -> do - let positiveFailureMax = abs failureMax - let options = defaultCircuitBreakerOptions { maxBreakerFailures = positiveFailureMax } - let service req = return $ req + 1 - (s, circuitBreakerService) <- circuitBreaker options service - results <- traverse (\req -> circuitBreakerService req) requestIDs - let expectedResults = fmap (+ 1) requestIDs + let positiveFailureMax = abs failureMax + options = defaultCircuitBreakerOptions { maxBreakerFailures = positiveFailureMax } + service req = pure $ req + 1 + (s, circuitBreakerService) <- circuitBreaker options service + results <- traverse circuitBreakerService requestIDs + let expectedResults = fmap (+ 1) requestIDs results `shouldBe` expectedResults - (isCircuitBreakerOpen s) `shouldReturn` False - (isCircuitBreakerClosed s) `shouldReturn` True + isCircuitBreakerOpen s `shouldReturn` False + isCircuitBreakerClosed s `shouldReturn` True + + it "combineCircuitBreakerStates reports open if any breaker is open" $ do + let opts = defaultCircuitBreakerOptions { maxBreakerFailures = 0 } + (s1, svc1) <- circuitBreaker opts (\(_ :: Int) -> throwIO CircuitBreakerTestException :: IO Int) + (s2, _) <- circuitBreaker opts (\(x :: Int) -> pure (x + 1) :: IO Int) + let combined = combineCircuitBreakerStates [s1, s2] + isCircuitBreakerOpen combined `shouldReturn` False + isCircuitBreakerClosed combined `shouldReturn` True + _ <- try (svc1 1) :: IO (Either CircuitBreakerTestException Int) + isCircuitBreakerOpen combined `shouldReturn` True + isCircuitBreakerClosed combined `shouldReturn` False diff --git a/test/Glue/ConcurrencyLimitSpec.hs b/test/Glue/ConcurrencyLimitSpec.hs new file mode 100644 --- /dev/null +++ b/test/Glue/ConcurrencyLimitSpec.hs @@ -0,0 +1,65 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module Glue.ConcurrencyLimitSpec (spec) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async, mapConcurrently, wait) +import Data.IORef +import Glue.ConcurrencyLimit +import Test.Hspec +import UnliftIO.Exception (Exception, throwIO, try) + +data ConcurrencyTestException = ConcurrencyTestException + deriving (Eq, Show) + +instance Exception ConcurrencyTestException + +spec :: Spec +spec = do + describe "limitConcurrency" $ do + it "Never exceeds maxInFlight concurrent executions" $ do + currentInFlight <- newIORef (0 :: Int) + peakInFlight <- newIORef (0 :: Int) + let maxAllowed = 3 + service (req :: Int) = do + cur <- atomicModifyIORef' currentInFlight (\c -> (c + 1, c + 1)) + atomicModifyIORef' peakInFlight (\p -> (max p cur, ())) + threadDelay (20 * 1000) + atomicModifyIORef' currentInFlight (\c -> (c - 1, ())) + pure (req * 2) + limited <- limitConcurrency (defaultConcurrencyLimitOptions { maxInFlight = maxAllowed }) service + results <- mapConcurrently limited [1 .. 12] + results `shouldBe` fmap (* 2) [1 .. 12] + peak <- readIORef peakInFlight + peak `shouldSatisfy` (<= maxAllowed) + peak `shouldSatisfy` (> 1) + + it "Rejects requests when maxInFlight and maxQueued are full" $ do + let opts = ConcurrencyLimitOptions + { maxInFlight = 1 + , maxQueued = Just 1 + , concurrencyLimitDescription = "Busy" + } + service (req :: Int) = do + threadDelay (60 * 1000) + pure req + limited <- limitConcurrency opts service + a1 <- async (try (limited 1) :: IO (Either ConcurrencyLimitExceededException Int)) + threadDelay (10 * 1000) + a2 <- async (try (limited 2) :: IO (Either ConcurrencyLimitExceededException Int)) + threadDelay (10 * 1000) + -- 3rd request should be rejected immediately since 1 is in-flight and 1 is queued + r3 <- try (limited 3) :: IO (Either ConcurrencyLimitExceededException Int) + r1 <- wait a1 + r2 <- wait a2 + r3 `shouldBe` Left (ConcurrencyLimitExceededException "Busy") + r1 `shouldBe` Right 1 + r2 `shouldBe` Right 2 + + it "Releases slot when underlying service throws an exception" $ do + let opts = defaultConcurrencyLimitOptions { maxInFlight = 1, maxQueued = Just 0 } + failingOrOk (req :: Int) = + if req == 0 then throwIO ConcurrencyTestException else pure (req + 10) + limited <- limitConcurrency opts failingOrOk + limited 0 `shouldThrow` (== ConcurrencyTestException) + limited 5 `shouldReturn` 15 diff --git a/test/Glue/DogpileProtectionSpec.hs b/test/Glue/DogpileProtectionSpec.hs --- a/test/Glue/DogpileProtectionSpec.hs +++ b/test/Glue/DogpileProtectionSpec.hs @@ -1,47 +1,46 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.DogpileProtectionSpec where +module Glue.DogpileProtectionSpec (spec) where -import Control.Concurrent -import Control.Concurrent.Async -import Control.Exception.Base hiding (throw, throwIO) -import Control.Exception.Lifted +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async, wait) import Data.IORef -import Data.Traversable -import Data.Typeable import Glue.DogpileProtection -import Prelude hiding (sequence) import Test.Hspec +import UnliftIO.Exception (Exception, throwIO) -data DogpileProtectionTestException = DogpileProtectionTestException deriving (Eq, Show, Typeable) +data DogpileProtectionTestException = DogpileProtectionTestException + deriving (Eq, Show) + instance Exception DogpileProtectionTestException requests :: [Int] -requests = take 10 $ repeat 1 +requests = replicate 10 1 delayTime :: Int -delayTime = 1000 * 1000 +delayTime = 50 * 1000 spec :: Spec spec = do describe "dogpileProtect" $ do it "With multiple calls to a slow service only one actually gets through" $ do counter <- newIORef (0 :: Int) - let service request = atomicModifyIORef' counter (\n -> (n + 1, ())) >> threadDelay delayTime >> return (request * 2) + let service request = do + atomicModifyIORef' counter (\n -> (n + 1, ())) + threadDelay delayTime + pure (request * 2) protectedService <- dogpileProtect service asyncResults <- traverse (async . protectedService) requests - let results = traverse wait asyncResults - results `shouldReturn` (fmap (*2) requests) - (readIORef counter) `shouldReturn` 1 + traverse wait asyncResults `shouldReturn` fmap (* 2) requests + readIORef counter `shouldReturn` 1 + it "With multiple calls to a slow failing service only one actually gets through" $ do counter <- newIORef (0 :: Int) - let service _ = atomicModifyIORef' counter (\n -> (n + 1, ())) >> threadDelay delayTime >> throwIO DogpileProtectionTestException :: IO Int + let service _ = do + atomicModifyIORef' counter (\n -> (n + 1, ())) + threadDelay delayTime + throwIO DogpileProtectionTestException :: IO Int protectedService <- dogpileProtect service asyncResults <- traverse (async . protectedService) requests - let results = traverse wait asyncResults - results `shouldThrow` (== DogpileProtectionTestException) - (readIORef counter) `shouldReturn` 1 - - + traverse wait asyncResults `shouldThrow` (== DogpileProtectionTestException) + readIORef counter `shouldReturn` 1 diff --git a/test/Glue/FailoverSpec.hs b/test/Glue/FailoverSpec.hs --- a/test/Glue/FailoverSpec.hs +++ b/test/Glue/FailoverSpec.hs @@ -1,28 +1,26 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts #-} -module Glue.FailoverSpec where +module Glue.FailoverSpec (spec) where -import Control.Exception.Base hiding (throw, throwIO) -import Control.Exception.Lifted -import Data.Typeable import Glue.Failover import Test.Hspec import Test.QuickCheck +import UnliftIO.Exception (Exception, throwIO) -data FailoverTestException = FailoverTestException deriving (Eq, Show, Typeable) +data FailoverTestException = FailoverTestException + deriving (Eq, Show) + instance Exception FailoverTestException spec :: Spec spec = do describe "failover" $ do - it "Failover handles potentially multiple errors" $ do + it "Failover handles potentially multiple errors" $ property $ \(request, failureMax) -> - let positiveFailureMax = (abs failureMax) `mod` 10 - service req = if req >= 10 then return (req + 100) else throwIO FailoverTestException - options = defaultFailoverOptions { transformFailoverRequest = (+1), maxFailovers = positiveFailureMax } - failOverService = failover options service - successCase = (failOverService request) `shouldReturn` ((if request <= 10 then 10 else request) + 100) - failureCase = (failOverService request) `shouldThrow` (== FailoverTestException) - in if request + positiveFailureMax >= 10 then successCase else failureCase + let positiveFailureMax = abs failureMax `mod` 10 + service req = if req >= 10 then pure (req + 100) else throwIO FailoverTestException + options = defaultFailoverOptions { transformFailoverRequest = (+ 1), maxFailovers = positiveFailureMax } + failOverService = failover options service + successCase = failOverService request `shouldReturn` ((if request <= 10 then 10 else request) + 100) + failureCase = failOverService request `shouldThrow` (== FailoverTestException) + in if request + positiveFailureMax >= 10 then successCase else failureCase diff --git a/test/Glue/HedgingSpec.hs b/test/Glue/HedgingSpec.hs new file mode 100644 --- /dev/null +++ b/test/Glue/HedgingSpec.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module Glue.HedgingSpec (spec) where + +import Control.Concurrent (threadDelay) +import Data.IORef +import Glue.Hedging +import Test.Hspec +import UnliftIO.Exception (Exception, throwIO) + +data HedgingTestException = HedgingTestException + deriving (Eq, Show) + +instance Exception HedgingTestException + +spec :: Spec +spec = do + describe "hedgingService" $ do + it "Completes with the fast hedged call when the primary call is slow" $ do + ref <- newIORef (0 :: Int) + let service (req :: Int) = do + attemptNum <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) + if attemptNum == 1 + then threadDelay (500 * 1000) >> pure (req + 100) + else pure (req * 2) + hedged = hedgingService (HedgingOptions { hedgingCount = 2, hedgingDelayMs = 5 }) service + hedged 21 `shouldReturn` 42 + + it "Succeeds if one attempt fails immediately and another succeeds" $ do + ref <- newIORef (0 :: Int) + let service (req :: Int) = do + attemptNum <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) + if attemptNum == 1 + then throwIO HedgingTestException + else pure (req * 3) + hedged = hedgingService (HedgingOptions { hedgingCount = 2, hedgingDelayMs = 0 }) service + hedged 10 `shouldReturn` 30 + + it "Rethrows the exception if all hedged attempts fail" $ do + let service (_ :: Int) = throwIO HedgingTestException :: IO Int + hedged = hedgingService (HedgingOptions { hedgingCount = 3, hedgingDelayMs = 0 }) service + hedged 1 `shouldThrow` (== HedgingTestException) diff --git a/test/Glue/LoadBalancerSpec.hs b/test/Glue/LoadBalancerSpec.hs new file mode 100644 --- /dev/null +++ b/test/Glue/LoadBalancerSpec.hs @@ -0,0 +1,46 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module Glue.LoadBalancerSpec (spec) where + +import Control.Concurrent.Async (mapConcurrently) +import Data.IORef +import Data.List.NonEmpty (NonEmpty (..)) +import Glue.LoadBalancer +import Glue.Types +import Test.Hspec +import Test.QuickCheck + +spec :: Spec +spec = do + describe "roundRobinLoadBalancer" $ do + it "Throws NoServicesAvailableException when given an empty list" $ do + lb <- roundRobinLoadBalancer ([] :: [BasicService IO Int Int]) + lb 42 `shouldThrow` (== NoServicesAvailableException) + + it "Distributes requests evenly in round-robin order" $ + property $ \(reqs :: [Int]) -> do + ref0 <- newIORef (0 :: Int) + ref1 <- newIORef (0 :: Int) + ref2 <- newIORef (0 :: Int) + let mkSvc tag ref req = do + atomicModifyIORef' ref (\c -> (c + 1, ())) + pure (tag, req * 2) + lb <- roundRobinLoadBalancerNE + ( mkSvc (0 :: Int) ref0 + :| [mkSvc 1 ref1, mkSvc 2 ref2] + ) + results <- traverse lb reqs + let expectedTags = zipWith (\i r -> (i `mod` 3, r * 2)) [0 ..] reqs + results `shouldBe` expectedTags + + it "Handles concurrent requests across all backends" $ do + ref0 <- newIORef (0 :: Int) + ref1 <- newIORef (0 :: Int) + let svc ref req = atomicModifyIORef' ref (\c -> (c + 1, ())) >> pure (req + 1) + lb <- roundRobinLoadBalancer [svc ref0, svc ref1] + results <- mapConcurrently lb ([1 .. 20] :: [Int]) + results `shouldBe` [2 .. 21] + c0 <- readIORef ref0 + c1 <- readIORef ref1 + c0 `shouldBe` 10 + c1 `shouldBe` 10 diff --git a/test/Glue/PreloadSpec.hs b/test/Glue/PreloadSpec.hs --- a/test/Glue/PreloadSpec.hs +++ b/test/Glue/PreloadSpec.hs @@ -1,18 +1,15 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.PreloadSpec where +module Glue.PreloadSpec (spec) where -import Control.Exception.Base import qualified Data.HashMap.Strict as M import qualified Data.HashSet as S -import Data.Typeable import Glue.Preload import Glue.Types import Test.Hspec import Test.QuickCheck import Test.QuickCheck.Instances () +import UnliftIO.Exception (Exception, finally, throwIO) serviceFunctionality :: MultiGetRequest Int -> MultiGetResponse Int Int serviceFunctionality rs = M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList rs @@ -20,41 +17,44 @@ failingService :: MultiGetRequest Int -> IO (MultiGetResponse Int Int) failingService _ = throwIO PreloadTestException -successfullCall :: IO Int -successfullCall = return 12 +successfulCall :: IO Int +successfulCall = pure 12 failingCall :: IO Int failingCall = throwIO PreloadTestException -data PreloadTestException = PreloadTestException deriving (Eq, Show, Typeable) +data PreloadTestException = PreloadTestException + deriving (Eq, Show) + instance Exception PreloadTestException spec :: Spec spec = do describe "preloadingService" $ do - it "Requests should work the same regardless of whether or not parts are preloaded" $ do + it "Requests should work the same regardless of whether or not parts are preloaded" $ property $ \(preload :: S.HashSet Int, nonPreload :: S.HashSet Int) -> do - let service = (return . serviceFunctionality) :: MultiGetService IO Int Int + let service = (pure . serviceFunctionality) :: MultiGetService IO Int Int (preloadedService, disable) <- preloadingService (defaultPreloadedOptions preload id) service let expectedResults = serviceFunctionality (S.union preload nonPreload) - actualResults <- preloadedService (S.union preload nonPreload) - disable () + actualResults <- preloadedService (S.union preload nonPreload) `finally` disable () actualResults `shouldBe` expectedResults - it "Exceptions should propagate" $ do + + it "Exceptions should propagate" $ property $ \(preload :: S.HashSet Int, nonPreload :: S.HashSet Int) -> do - (preloadedService, _) <- preloadingService (defaultPreloadedOptions preload id) failingService + (preloadedService, disable) <- preloadingService (defaultPreloadedOptions preload id) failingService let expectedResults = serviceFunctionality (S.union preload nonPreload) - let goodPath = preloadedService (S.union preload nonPreload) `shouldReturn` expectedResults - let badPath = preloadedService (S.union preload nonPreload) `shouldThrow` (== PreloadTestException) - if (S.null preload && S.null nonPreload) then goodPath else badPath + goodPath = preloadedService (S.union preload nonPreload) `shouldReturn` expectedResults + badPath = preloadedService (S.union preload nonPreload) `shouldThrow` (== PreloadTestException) + (if S.null preload && S.null nonPreload then goodPath else badPath) `finally` disable () + describe "preloadingCall" $ do - it "Requests should work" $ do + it "Requests should work" $ property $ do - (preloadedCall, disable) <- preloadingCall id 1000 successfullCall - actualResults <- preloadedCall - disable () + (preloadedCall, disable) <- preloadingCall id 1000 successfulCall + actualResults <- preloadedCall `finally` disable () actualResults `shouldBe` 12 - it "Exceptions should propagate" $ do + + it "Exceptions should propagate" $ property $ do - (preloadedCall, _) <- preloadingCall id 1000 failingCall - preloadedCall `shouldThrow` (== PreloadTestException) + (preloadedCall, disable) <- preloadingCall id 1000 failingCall + (preloadedCall `shouldThrow` (== PreloadTestException)) `finally` disable () diff --git a/test/Glue/RetrySpec.hs b/test/Glue/RetrySpec.hs --- a/test/Glue/RetrySpec.hs +++ b/test/Glue/RetrySpec.hs @@ -1,54 +1,53 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE OverloadedStrings #-} - -module Glue.RetrySpec where +module Glue.RetrySpec (spec) where -import Control.Concurrent -import Control.Exception.Base hiding (throw, throwIO, throwTo) -import Control.Exception.Lifted hiding (throwTo) -import Control.Monad.IO.Class +import Control.Concurrent (forkIO, myThreadId, threadDelay) +import Control.Exception.Base (AsyncException (UserInterrupt), + throwTo) +import Control.Monad.IO.Class (liftIO) import Data.IORef -import Data.Typeable import Glue.Retry import Test.Hspec import Test.QuickCheck +import UnliftIO.Exception (Exception, throwIO) -newtype SmallInt = SmallInt Int deriving (Eq, Show) +newtype SmallInt = SmallInt Int + deriving (Eq, Show) instance Arbitrary SmallInt where arbitrary = sized $ \s -> do - n <- choose (0, s `min` 10) - return $ SmallInt n + n <- choose (0, s `min` 10) + pure $ SmallInt n -data RetryTestException = RetryTestException deriving (Eq, Show, Typeable) +data RetryTestException = RetryTestException + deriving (Eq, Show) + instance Exception RetryTestException spec :: Spec spec = do describe "retryingService" $ do - it "Attempts a service call multiple times" $ do - property $ \(request, (SmallInt failures), (SmallInt retries)) -> - do - ref <- liftIO $ newIORef 0 - let service req = do - counter <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) - if counter > failures then return req else throwIO RetryTestException - let options = defaultRetryOptions { maximumRetries = retries } - let retryService = retryingService options service - let successCase = (retryService request) `shouldReturn` (request :: Int) - let failureCase = (retryService request) `shouldThrow` (== RetryTestException) - if retries >= failures then successCase else failureCase - it "Asynchronous exceptions are rethrown" $ do - property $ \(request, retries) -> - do - ref <- liftIO $ newIORef (0 :: Int) - let service req = do - atomicModifyIORef' ref (\c -> (c + 1, ())) - threadId <- myThreadId - _ <- forkIO (threadDelay 10000 >> throwTo threadId UserInterrupt) - threadDelay 1000000 - return (req * 2 :: Int) - let options = defaultRetryOptions { maximumRetries = retries } - let retryService = retryingService options service - (retryService request) `shouldThrow` (== UserInterrupt) - (readIORef ref) `shouldReturn` 1 + it "Attempts a service call multiple times" $ + property $ \(request, SmallInt failures, SmallInt retries) -> do + ref <- liftIO $ newIORef 0 + let service req = do + counter <- atomicModifyIORef' ref (\c -> (c + 1, c + 1)) + if counter > failures then pure req else throwIO RetryTestException + options = defaultRetryOptions { maximumRetries = retries } + retryService = retryingService options service + successCase = retryService request `shouldReturn` (request :: Int) + failureCase = retryService request `shouldThrow` (== RetryTestException) + if retries >= failures then successCase else failureCase + + it "Asynchronous exceptions are rethrown" $ + property $ \(request, retries) -> do + ref <- liftIO $ newIORef (0 :: Int) + let service req = do + atomicModifyIORef' ref (\c -> (c + 1, ())) + threadId <- myThreadId + _ <- forkIO (threadDelay 2000 >> throwTo threadId UserInterrupt) + threadDelay 1000000 + pure (req * 2 :: Int) + options = defaultRetryOptions { maximumRetries = retries } + retryService = retryingService options service + retryService request `shouldThrow` (== UserInterrupt) + readIORef ref `shouldReturn` 1 diff --git a/test/Glue/ShutdownSpec.hs b/test/Glue/ShutdownSpec.hs new file mode 100644 --- /dev/null +++ b/test/Glue/ShutdownSpec.hs @@ -0,0 +1,43 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module Glue.ShutdownSpec (spec) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async, wait) +import Glue.Shutdown +import Test.Hspec + +spec :: Spec +spec = do + describe "gracefulShutdown" $ do + it "Allows in-flight requests to drain cleanly within the grace period and rejects new requests" $ do + let service (req :: Int) = do + threadDelay (30 * 1000) + pure (req * 2) + (wrapped, handle) <- gracefulShutdown (defaultShutdownOptions { shutdownGracePeriodMs = 500 }) service + isShuttingDown handle `shouldReturn` False + inFlight <- async (wrapped 21) + threadDelay (5 * 1000) + inFlightCount handle `shouldReturn` 1 + drainedCleanly <- initiateShutdown handle + drainedCleanly `shouldBe` True + isShuttingDown handle `shouldReturn` True + wait inFlight `shouldReturn` 42 + wrapped 99 `shouldThrow` (== ServiceShuttingDownException "Service is shutting down.") + + it "Returns shutdownFallback for rejected calls when configured" $ do + let opts = defaultShutdownOptions { shutdownFallback = Just (-1 :: Int), shutdownGracePeriodMs = 100 } + (wrapped, handle) <- gracefulShutdown opts (\x -> pure (x * 2)) + wrapped 10 `shouldReturn` 20 + _ <- initiateShutdown handle + wrapped 10 `shouldReturn` (-1) + + it "Cancels stuck in-flight requests when grace period expires" $ do + let opts = defaultShutdownOptions { shutdownFallback = Just (-999 :: Int), shutdownGracePeriodMs = 20 } + slowService (_ :: Int) = threadDelay (500 * 1000) >> pure 100 + (wrapped, handle) <- gracefulShutdown opts slowService + inFlight <- async (wrapped 1) + threadDelay (5 * 1000) + drainedCleanly <- initiateShutdown handle + drainedCleanly `shouldBe` False + wait inFlight `shouldReturn` (-999) diff --git a/test/Glue/SwitchingSpec.hs b/test/Glue/SwitchingSpec.hs --- a/test/Glue/SwitchingSpec.hs +++ b/test/Glue/SwitchingSpec.hs @@ -1,8 +1,6 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} -module Glue.SwitchingSpec where +module Glue.SwitchingSpec (spec) where import Glue.Switching import Glue.Types @@ -13,15 +11,15 @@ spec :: Spec spec = do describe "switchingService" $ do - it "Requests should be handled by the service used" $ do + it "Requests should be handled by the service used" $ property $ \(firstResult :: MultiGetResponse Int Int, secondResult :: MultiGetResponse Int Int, request :: MultiGetRequest Int) -> do - let first _ = return firstResult - let second _ = return secondResult + let first _ = pure firstResult + second _ = pure secondResult (switchedService, serviceUpdate) <- switchingService first - actualFirstResult <- switchedService request + actualFirstResult <- switchedService request expectedFirstResult <- first request serviceUpdate second - actualSecondResult <- switchedService request + actualSecondResult <- switchedService request expectedSecondResult <- second request actualFirstResult `shouldBe` expectedFirstResult actualSecondResult `shouldBe` expectedSecondResult diff --git a/test/Glue/TimeoutSpec.hs b/test/Glue/TimeoutSpec.hs --- a/test/Glue/TimeoutSpec.hs +++ b/test/Glue/TimeoutSpec.hs @@ -1,9 +1,6 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE OverloadedStrings #-} - -module Glue.TimeoutSpec where +module Glue.TimeoutSpec (spec) where -import Control.Concurrent +import Control.Concurrent (threadDelay) import Glue.Timeout import Test.Hspec import Test.QuickCheck @@ -14,10 +11,10 @@ spec :: Spec spec = do describe "addTimeout" $ do - it "Services taking too long should fail with a timeout" $ do + it "Services taking too long should fail with a timeout" $ property $ \request -> - let service req = (if req > 0 then return () else threadDelay (100 * 1000)) >> return (req + 100) - timeoutService = addTimeout timeoutOptions service - successCase = (timeoutService request) `shouldReturn` (request + 100) - failureCase = (timeoutService request) `shouldThrow` (== TimeoutException(timeoutDescription timeoutOptions)) - in if request > (0 :: Int) then successCase else failureCase + let service req = (if req > 0 then pure () else threadDelay (50 * 1000)) >> pure (req + 100) + timeoutService = addTimeout timeoutOptions service + successCase = timeoutService request `shouldReturn` (request + 100) + failureCase = timeoutService request `shouldThrow` (== TimeoutException (timeoutDescription timeoutOptions)) + in if request > (0 :: Int) then successCase else failureCase