glue (empty) → 0.1.0.0
raw patch · 16 files changed
+782/−0 lines, 16 filesdep +QuickCheckdep +asyncdep +basesetup-changed
Dependencies added: QuickCheck, async, base, ekg-core, hashable, hspec, lifted-base, monad-control, quickcheck-instances, text, time, transformers, transformers-base, unordered-containers
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- example/Glue/Example/BatcherExample.hs +47/−0
- example/Main.hs +7/−0
- glue.cabal +88/−0
- src/Glue.hs +46/−0
- src/Glue/Batcher.hs +90/−0
- src/Glue/Caching.hs +49/−0
- src/Glue/CircuitBreaker.hs +80/−0
- src/Glue/DogpileProtection.hs +37/−0
- src/Glue/Failover.hs +37/−0
- src/Glue/Retry.hs +49/−0
- src/Glue/Stats.hs +109/−0
- src/Glue/Timeout.hs +45/−0
- src/Glue/Types.hs +52/−0
- test/Main.hs +14/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015, Sean Parsons++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Sean Parsons nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ example/Glue/Example/BatcherExample.hs view
@@ -0,0 +1,47 @@+{-# LANGUAGE OverloadedStrings #-}++module Glue.Example.BatcherExample(+ batcherExample+) where++import Glue+import Control.Concurrent.Async+import Control.Concurrent.Lifted+import Data.IORef.Lifted+import qualified Data.List as L+import Data.Text+import Data.Traversable+import System.Metrics+import qualified System.Metrics.Distribution as D+import qualified Data.HashSet as S+import qualified Data.HashMap.Strict as M+import Text.Printf++printStats :: Store -> Text -> IO ()+printStats store name = do+ samples <- sampleAll store+ let possibleValue = M.lookup name samples+ case possibleValue of+ (Just (Distribution stats)) -> printf "%s - sum : %f - count : %d" (unpack name) (D.sum stats) (D.count stats) >> putStrLn ""+ otherwise -> return ()++runTest :: Store -> Text -> Bool -> IO ()+runTest store name shouldBatch = do+ counter <- newIORef 0+ let listOfNums = [1..50]+ let requests = fmap (S.fromList . L.take 10) $ L.tails listOfNums :: [S.HashSet Int]+ let service request = do+ atomicModifyIORef' counter (\c -> (c + (S.size request), ()))+ threadDelay (500 * S.size request)+ return $ M.fromList $ fmap (\r -> (r, r * 2)) $ S.toList request+ statsWrappedService <- recordDistribution store name service+ possiblyBatchedService <- if shouldBatch then return statsWrappedService else fmap (\(_, b) -> b) $ batchingService defaultBatchingOptions statsWrappedService+ (mapConcurrently possiblyBatchedService requests)+ printStats store name++-- Tests the batchingService by comparing the stats with and without it.+batcherExample :: IO ()+batcherExample = do+ store <- newStore+ runTest store "Unbatched Service" False+ runTest store "Batched Service" True
+ example/Main.hs view
@@ -0,0 +1,7 @@+module Main where++import Glue.Example.BatcherExample++main :: IO ()+main = do+ batcherExample
+ glue.cabal view
@@ -0,0 +1,88 @@+name: glue+version: 0.1.0.0+synopsis: Make better services.+description: Implements common patterns used in building services that run smoothly and efficiently.+license: BSD3+license-file: LICENSE+author: Sean Parsons+maintainer: github@futurenotfound.com+category: Network+build-type: Simple+-- extra-source-files: +cabal-version: >=1.10++source-repository head+ type: git+ location: git://github.com/seanparsons/glue.git++library+ exposed-modules: Glue+ Glue.Types+ Glue.Failover+ Glue.Caching+ Glue.CircuitBreaker+ Glue.DogpileProtection+ Glue.Timeout+ Glue.Retry+ Glue.Batcher+ Glue.Stats+ -- other-modules: + -- other-extensions: + build-depends: base ==4.*,+ transformers,+ transformers-base,+ lifted-base,+ time,+ monad-control,+ unordered-containers,+ hashable,+ ekg-core,+ text+ ghc-options: -rtsopts+ -Wall+ hs-source-dirs: src+ default-language: Haskell2010++executable example+ main-is: Main.hs+ hs-source-dirs: src,+ example+ other-modules: Glue.Example.BatcherExample+ build-depends: base ==4.*,+ transformers,+ transformers-base,+ lifted-base,+ time,+ monad-control,+ unordered-containers,+ hashable,+ ekg-core,+ text,+ async+ default-language: Haskell2010++test-suite tests+ build-depends: base ==4.*,+ QuickCheck -any,+ quickcheck-instances,+ hspec -any,+ transformers,+ transformers-base,+ lifted-base,+ time,+ monad-control,+ unordered-containers,+ hashable,+ ekg-core,+ text,+ async+ ghc-options: -rtsopts+ -Wall+ -O2+ -threaded+ type: exitcode-stdio-1.0+ main-is: Main.hs+ buildable: True+ default-language: Haskell2010+ hs-source-dirs: test,+ src
+ src/Glue.hs view
@@ -0,0 +1,46 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module that re-exports the entirety of the library.+module Glue+ ( module Glue.Types+ , module Glue.Failover+ , module Glue.Caching+ , module Glue.CircuitBreaker+ , module Glue.DogpileProtection+ , module Glue.Timeout+ , module Glue.Retry+ , module Glue.Batcher+ , module Glue.Stats+) where++import Glue.Types+import Glue.Failover+import Glue.Caching+import Glue.CircuitBreaker+import Glue.DogpileProtection+import Glue.Timeout+import Glue.Retry+import Glue.Batcher+import Glue.Stats++{-+Done:+- Caching.+- Fail fast/circuit breaker.+- Failover.+- Dogpile protection.+- Timeouts.+- Retries.+- Multi-get optimisations.+- Performance stats.++Todo:+- Load balancer, maybe just a simple round robin to start with.+- Concurrent request constraint, possibly X inflight and Y queued.+- Switching service, allowing different implementations to be switched in and out at runtime.+- Routing service? Possibly something akin to HTTP routing in a lot of web frameworks.+- Request duplication, with completion based on the first successful result.+- Graceful service shutdown, reject new requests with a default value, wait for a period of time while existing requests continue. Then possibly a cancel action?+- Caching that supports refreshing invisibly to the consumer?+-}
+ src/Glue/Batcher.hs view
@@ -0,0 +1,90 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module for creating a service that batches calls across requests.+module Glue.Batcher(+ BatchingOptions+ , batchingService+ , defaultBatchingOptions+ , batchWindowMs+) where++import Control.Applicative+import Control.Concurrent.Lifted+import Control.Exception.Lifted+import Control.Monad+import Control.Monad.Trans.Control+import qualified Data.HashSet as S+import qualified Data.HashMap.Strict as M+import Data.Foldable+import Data.Hashable+import Data.IORef.Lifted+import Glue.Types++-- | Options for configuring the batching. +data BatchingOptions = BatchingOptions {+ batchWindowMs :: Int -- ^ Window in milliseconds over which to batch.+} deriving (Eq, Show)++-- | Default instance of 'BatchingOptions' with a batching window of 10ms.+defaultBatchingOptions :: BatchingOptions+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))++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++emptyBatch :: (Eq a, Hashable a) => RequestBatch a b+emptyBatch = RequestBatch [] S.empty++makeCall :: (Eq a, Hashable a, MonadBaseControl IO m) => MultiGetService m a b -> S.HashSet a -> m (Either SomeException (M.HashMap a b))+makeCall service requests = catch (fmap Right $ service requests) (\(e :: SomeException) -> return $ Left e) ++processCalls :: (Eq a, Hashable a, MonadBaseControl IO m) => MultiGetService m a b -> RequestBatch a b -> m ()+processCalls service (RequestBatch pendings requests) = do+ 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++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 ()))++singleService :: (Eq a, Hashable a, MonadBaseControl IO 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+multiService options service ref requests = do+ 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 options service = do+ ref <- newIORef emptyBatch+ return (singleService options service ref, multiService options service ref)
+ src/Glue/Caching.hs view
@@ -0,0 +1,49 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module supporting the caching of a service.+module Glue.Caching(+ cacheWithBasic+ , cacheWithMulti+) where++import Data.Hashable+import Glue.Types+import qualified Data.HashSet as S+import qualified Data.HashMap.Strict as M++-- | 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 + cachedService request = do+ fromCache <- lookupWith request+ maybe (fallback request) return 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+cacheWithMulti lookupWith insertWith service = + let fallback request = do+ result <- service request+ insertWith result+ return 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
+ src/Glue/CircuitBreaker.hs view
@@ -0,0 +1,80 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DeriveDataTypeable #-}++-- | 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+ , CircuitBreakerStatus+ , CircuitBreakerException(..)+ , defaultCircuitBreakerOptions+ , circuitBreaker+ , maxBreakerFailures+ , resetTimeoutSecs+ , breakerDescription+) 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 Glue.Types++-- | 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.+}++-- | Defaulted options for the circuit breaker with 3 failures over 60 seconds.+defaultCircuitBreakerOptions :: CircuitBreakerOptions+defaultCircuitBreakerOptions = CircuitBreakerOptions { maxBreakerFailures = 3, resetTimeoutSecs = 60, breakerDescription = "Circuit breaker open." }++-- | Status indicating if the circuit is open.+data CircuitBreakerStatus = CircuitBreakerClosed Int | CircuitBreakerOpen Int++-- | Exception thrown when the circuit is open.+data CircuitBreakerException = CircuitBreakerException String deriving (Eq, Show, Typeable)+instance Exception CircuitBreakerException++-- 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 (IORef CircuitBreakerStatus, BasicService m a b)+circuitBreaker options service = + let getCurrentTime = 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 $ CircuitBreakerClosed 0+ return result+ incErrors ref = do+ currentTime <- getCurrentTime+ atomicModifyIORef' ref $ \status -> case status of+ (CircuitBreakerClosed errorCount) -> (if errorCount >= failureMax then CircuitBreakerOpen (currentTime + (resetTimeoutSecs options)) else CircuitBreakerClosed (errorCount + 1), ())+ other -> (other, ())+ + failingCall = throw $ CircuitBreakerException $ breakerDescription options+ callIfOpen request ref = do+ currentTime <- getCurrentTime+ canaryRequest <- atomicModifyIORef' ref $ \status -> case status of + (CircuitBreakerClosed _) -> (status, False)+ (CircuitBreakerOpen time) -> if currentTime > time then ((CircuitBreakerOpen (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 + (CircuitBreakerClosed _) -> callIfClosed request ref+ (CircuitBreakerOpen _) -> callIfOpen request ref+ + in do+ ref <- newIORef $ CircuitBreakerClosed 0+ return (ref, breakerService ref)
+ src/Glue/DogpileProtection.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module supporting the dogpile protection of a service, see <http://en.wikipedia.org/wiki/Cache_stampede http://en.wikipedia.org/wiki/Cache_stampede>.+module Glue.DogpileProtection(+ dogpileProtect+) where++import Control.Concurrent.Lifted+import Control.Exception.Lifted+import Control.Monad.Trans.Control+import Data.Hashable+import qualified Data.HashMap.Strict as M+import Data.IORef.Lifted+import Glue.Types++-- TODO: Should make this return just BasicService, hiding the HashMap.+-- | Dogpile protection of a service, to prevent multiple calls for the same value being submitted.+-- | Loses the values held within m.+dogpileProtect :: (MonadBaseControl IO m, MonadBaseControl IO n, Eq a, Hashable a) + => BasicService m a b -- ^ The service to protect.+ -> n (IORef (M.HashMap a (ResultVar b)), BasicService m a b)+dogpileProtect service = do+ mapRef <- newIORef M.empty+ let protectedService request = do+ firstRequestMVar <- newEmptyMVar + resultAction <- atomicModifyIORef' mapRef (\refMap -> + let removeFromMap = atomicModifyIORef' mapRef (\m -> (M.delete request m, ()))+ invokeService = do + result <- bracketOnError (return ()) (\_ -> removeFromMap) (\_ -> service request)+ putMVar firstRequestMVar $ Right result+ return result+ updateMap mvar = (M.insert request mvar refMap, getResult mvar)+ addToMap = (M.insert request firstRequestMVar refMap, invokeService)+ in maybe addToMap updateMap $ M.lookup request refMap)+ resultAction+ return (mapRef, protectedService)
+ src/Glue/Failover.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE FlexibleContexts #-}+{-# 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+ , defaultFailoverOptions+ , failover+ , maxFailovers+ , transformFailoverRequest+) where++import Glue.Types+import Control.Exception.Lifted hiding(throw)+import Control.Monad.Trans.Control++-- | 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.+}++-- | Simple defaults that results in retrying 3 times in a dumb way immediately without transforming the request.+defaultFailoverOptions :: FailoverOptions a+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 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+ in invokeService 0
+ src/Glue/Retry.hs view
@@ -0,0 +1,49 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DeriveDataTypeable #-}++-- | Module containing retry functionality, allowing the construction of services that attempt multiple times in case of transient failure.+module Glue.Retry(+ RetryOptions+ , defaultRetryOptions+ , retryingService+ , retryAllowed+ , retryInitialWaitTimeMs+ , maximumRetries+ , retryWaitTimeMultiplier+) where++import Glue.Types+import Control.Concurrent.Lifted+import Control.Exception.Lifted+import Control.Monad.Trans.Control++-- | 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.+}++-- | Defaulted options for retrying 3 times with no wait time.+defaultRetryOptions :: RetryOptions a+defaultRetryOptions = RetryOptions { + retryAllowed = (\_ -> True)+ , retryInitialWaitTimeMs = 0+ , maximumRetries = 3+ , retryWaitTimeMultiplier = 0+ }++-- | Retries a call to a service multiple times, potentially backing off wait times between subsequent calls.+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+retryingService options service =+ let attempt retryCount request = if (retryAllowed options) request && maxRetries > retryCount+ then catch (service request) (\(_ :: SomeException) -> (wait (retryCount + 1)) >> (attempt (retryCount + 1) request))+ else service request+ maxRetries = maximumRetries options+ wait retryCount = threadDelay $ round $ fromIntegral (retryInitialWaitTimeMs options) * ((retryWaitTimeMultiplier options) ^ retryCount)+ in attempt 0
+ src/Glue/Stats.hs view
@@ -0,0 +1,109 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Module supporting the recording of various stats about a service using "System.Metrics". +module Glue.Stats(+ recordDistribution+ , recordAttempts+ , recordSuccesses+ , recordFailures+ , recordLastRequest+ , recordLastResult+) where++import Control.Exception.Lifted+import Control.Monad.Base+import Data.Text+import Data.Time.Clock.POSIX+import Glue.Types+import Control.Monad.Trans.Control+import System.Metrics+import qualified System.Metrics.Distribution as MD+import qualified System.Metrics.Counter as MC+import qualified System.Metrics.Label as ML++currentTime :: (MonadBaseControl IO m) => m Double+currentTime = do+ time <- liftBase $ getPOSIXTime+ return $ realToFrac time++-- | Record the timings for service invocations with a 'Distribution' held in the passed in 'Store'..+recordDistribution :: (MonadBaseControl IO m, MonadBaseControl IO n)+ => Store -- ^ 'Store' where the 'Distribution' will reside.+ -> Text -- ^ The name to associate the 'Distribution' with.+ -> BasicService m a b -- ^ Base service to record stats for.+ -> n (BasicService m a b)+recordDistribution store name service = do+ dist <- liftBase $ createDistribution name store+ let timedService req = do+ before <- currentTime+ let recordTime = do+ after <- currentTime+ liftBase $ MD.add dist (after - before)+ finally (service req) recordTime + return timedService++-- | Increments a counter with a 'Counter' held in the passed in 'Store' for each time the service is called.+recordAttempts :: (MonadBaseControl IO m, MonadBaseControl IO n) + => Store -- ^ 'Store' where the 'Counter' will reside.+ -> Text -- ^ The name to associate the 'Counter' with.+ -> BasicService m a b -- ^ Base service to record stats for.+ -> n (BasicService m a b)+recordAttempts store name service = do+ counter <- liftBase $ createCounter name store+ let countedService req = do+ liftBase $ MC.inc counter+ service req+ return countedService++-- | Increments a counter with a 'Counter' held in the passed in 'Store' for each time the service successfully returns.+recordSuccesses :: (MonadBaseControl IO m, MonadBaseControl IO n) + => Store -- ^ 'Store' where the 'Counter' will reside.+ -> Text -- ^ The name to associate the 'Counter' with.+ -> BasicService m a b -- ^ Base service to record stats for.+ -> n (BasicService m a b)+recordSuccesses store name service = do+ counter <- liftBase $ createCounter name store+ let countedService req = do+ result <- service req+ liftBase $ MC.inc counter+ return result+ return countedService++-- | Increments a counter with a 'Counter' held in the passed in 'Store' for each time the service fails.+recordFailures :: (MonadBaseControl IO m, MonadBaseControl IO n) + => Store -- ^ 'Store' where the 'Counter' will reside.+ -> Text -- ^ The name to associate the 'Counter' with.+ -> BasicService m a b -- ^ Base service to record stats for.+ -> n (BasicService m a b)+recordFailures store name service = do+ counter <- liftBase $ createCounter name store+ let countedService req = onException (service req) (liftBase $ MC.inc counter)+ return countedService++-- | Sets a 'Label' held in the passed in 'Store' for each request the service receives.+recordLastRequest :: (MonadBaseControl IO m, MonadBaseControl IO n, Show a) + => Store -- ^ 'Store' where the 'Label' will reside.+ -> Text -- ^ The name to associate the 'Label' with.+ -> BasicService m a b -- ^ Base service to record stats for.+ -> n (BasicService m a b)+recordLastRequest store name service = do+ label <- liftBase $ createLabel name store+ let requestRecordingService req = do+ liftBase $ ML.set label $ pack $ show req+ service req+ return requestRecordingService++-- | Sets a 'Label' held in the passed in 'Store' for each successful result the service returns.+recordLastResult :: (MonadBaseControl IO m, MonadBaseControl IO n, Show b) + => Store -- ^ 'Store' where the 'Label' will reside.+ -> Text -- ^ The name to associate the 'Label' with.+ -> BasicService m a b -- ^ Base service to record stats for.+ -> n (BasicService m a b)+recordLastResult store name service = do+ label <- liftBase $ createLabel name store+ let resultRecordingService req = do+ result <- service req+ liftBase $ ML.set label $ pack $ show result+ return result + return resultRecordingService
+ src/Glue/Timeout.hs view
@@ -0,0 +1,45 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DeriveDataTypeable #-}++-- | Module supporting adding timeouts to a given service.+module Glue.Timeout(+ TimeoutOptions+ , TimeoutException(..)+ , defaultTimeoutOptions+ , addTimeout+ , timeoutDescription+ , timeoutLimitMs+) where++import Data.Typeable+import Glue.Types+import Control.Concurrent.Lifted+import Control.Exception.Lifted+import Control.Monad.Trans.Control++-- | 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.+}++-- | Default instance of 'TimeoutOptions' with a timeout of 30 seconds.+defaultTimeoutOptions :: TimeoutOptions+defaultTimeoutOptions = TimeoutOptions { timeoutDescription = "Service call timed out.", timeoutLimitMs = 30000 }++-- | Exception thrown when the timeout is exceeded.+data TimeoutException = TimeoutException String deriving (Eq, Show, Typeable)+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))
+ src/Glue/Types.hs view
@@ -0,0 +1,52 @@+{-# LANGUAGE FlexibleContexts #-}++-- | Module containing the root types and some support functionality.+module Glue.Types(+ BasicService+ , MultiGetService+ , MultiGetRequest+ , MultiGetResponse+ , ResultVar+ , multiGetToBasic+ , basicToMultiGet+ , getResult+) where++import Control.Applicative+import Data.Hashable+import Control.Concurrent+import qualified Control.Concurrent.MVar.Lifted as MV+import Control.Exception.Base hiding(throw, throwIO)+import Control.Exception.Lifted hiding(throw)+import Control.Monad.Trans.Control+import qualified Data.HashSet as S+import qualified Data.HashMap.Strict as M++-- | Type alias for the most basic form of a service supported.+type BasicService m a b = a -> m b+-- | Type alias for the request portion of a `MultiGetService`.+type MultiGetRequest a = S.HashSet a+-- | Type alias for the response portion of a `MultiGetService`.+type MultiGetResponse a b = M.HashMap a b+-- | Type alias for a service that looks up multiple values and returns multiple results.+type MultiGetService m a b = BasicService m (MultiGetRequest a) (MultiGetResponse a b)+-- | Type alias for the common container of a result used in asynchronous calls.+type ResultVar a = MVar (Either SomeException a)++-- | Convert a 'MultiGetService' into a 'BasicService' that looks up a single key, returning a `Maybe` which determines if the value was present.+multiGetToBasic :: (Hashable a, Eq a, Monad m) => MultiGetService m a b -> BasicService m a (Maybe b)+multiGetToBasic service = (\r -> do+ mapResult <- service (S.singleton r)+ return $ M.lookup r mapResult)++-- | Convert a 'BasicService' into a 'MultiGetService'+basicToMultiGet :: (Hashable a, Eq a, Applicative m) => BasicService m a b -> MultiGetService m a b+basicToMultiGet service = + let callService resultMap request = liftA2 (flip $ M.insert request) resultMap (service request)+ in S.foldl' callService (pure M.empty)++-- | Obtain a result from a 'ResultVar' in the 'Monad' of your choice.+getResult :: (MonadBaseControl IO m) => ResultVar a -> m a+getResult var = do+ result <- MV.readMVar var+ either throwIO return result
+ test/Main.hs view
@@ -0,0 +1,14 @@+module Main where++import Test.Hspec.Runner+import qualified Spec++customConfig :: Config+customConfig = defaultConfig + { configColorMode = ColorAlways+ , configPrintCpuTime = True+ }+++main :: IO ()+main = hspecWith customConfig Spec.spec