diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/example/Glue/Example/BatcherExample.hs b/example/Glue/Example/BatcherExample.hs new file mode 100644 --- /dev/null +++ b/example/Glue/Example/BatcherExample.hs @@ -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 diff --git a/example/Main.hs b/example/Main.hs new file mode 100644 --- /dev/null +++ b/example/Main.hs @@ -0,0 +1,7 @@ +module Main where + +import Glue.Example.BatcherExample + +main :: IO () +main = do + batcherExample diff --git a/glue.cabal b/glue.cabal new file mode 100644 --- /dev/null +++ b/glue.cabal @@ -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 diff --git a/src/Glue.hs b/src/Glue.hs new file mode 100644 --- /dev/null +++ b/src/Glue.hs @@ -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? +-} diff --git a/src/Glue/Batcher.hs b/src/Glue/Batcher.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Batcher.hs @@ -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) diff --git a/src/Glue/Caching.hs b/src/Glue/Caching.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Caching.hs @@ -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 diff --git a/src/Glue/CircuitBreaker.hs b/src/Glue/CircuitBreaker.hs new file mode 100644 --- /dev/null +++ b/src/Glue/CircuitBreaker.hs @@ -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) diff --git a/src/Glue/DogpileProtection.hs b/src/Glue/DogpileProtection.hs new file mode 100644 --- /dev/null +++ b/src/Glue/DogpileProtection.hs @@ -0,0 +1,37 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Module supporting the dogpile protection of a service, see . +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) diff --git a/src/Glue/Failover.hs b/src/Glue/Failover.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Failover.hs @@ -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 diff --git a/src/Glue/Retry.hs b/src/Glue/Retry.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Retry.hs @@ -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 diff --git a/src/Glue/Stats.hs b/src/Glue/Stats.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Stats.hs @@ -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 diff --git a/src/Glue/Timeout.hs b/src/Glue/Timeout.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Timeout.hs @@ -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)) diff --git a/src/Glue/Types.hs b/src/Glue/Types.hs new file mode 100644 --- /dev/null +++ b/src/Glue/Types.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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