diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,3 @@ +* 0.1.0.0 Daniel Patterson 2015-10-25 + + Initial release. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,13 @@ +Copyright (c) 2015 Daniel Patterson + +Permission to use, copy, modify, and/or distribute this software for any purpose +with or without fee is hereby granted, provided that the above copyright notice +and this permission notice appear in all copies. + +THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND +FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS +OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER +TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF +THIS SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,146 @@ +## About + +`hworker` is a Redis-backed persistent at-least-once queue library. It +is vaguely inspired by `sidekiq` for Ruby. It is intended to be a +simple reliable mechanism for processing background tasks. The jobs +can be created by a Haskell application or any application that can +push JSON data structures of the right shape into a Redis queue. The +application that processes the jobs need not be the same one as the +application that creates them (they just need to be able to talk to +the same Redis server, and use the same serialization to/from JSON). + +## Stability + +This has been running in one application sending email (using +`hworker-ses`) for several months. This is relatively low traffic +(transactional messages) most of the time, with spikes of 10k-30k +messages (mailing blasts). + +## Important Note + +The expiration of jobs is really important. It defaults to 120 +seconds, which may be short depending on your application (for things +like sending emails, it may be fine). **The reason why this timeout is +important is that if a job ever runs longer than this, the monitor +will think that the job failed** in some inexplicable way (like the +server running the job died) and will add the job back to the queue to +be run. Based on the semantics of this job processor, jobs running +multiple times is not a failure case, but it's obviously not something +you _want_ to happen, so be sure to set the timeout to something +reasonable for your application. + +## Overview + +To define jobs, you define a serialized representation of the job, and +a function that runs the job, which returns a status. The behavior of +uncaught exceptions is defined when you create the worker - it can be +either `Failure` or `Retry`. Jobs that return `Failure` are removed +from the queue, whereas jobs that return `Retry` are added again. The +only difference between a `Success` and a `Failure` is that a +`Failure` returns a message that is logged (ie, neither run again). + +## Example + +See the `example` directory in the repository. + +## Semantics + +This behavior of this queue processor is at-least-once. + +We rely on the defined behavior of Redis for reliability. Once a job +has been `queue`d, it is guaranteed to be run eventually, provided +some worker and monitor threads exist. If the worker thread that was +running a given job dies, the job will eventually be retried (if you +do not want this behavior, do not start any monitor threads). Once the +job completes, provided nothing kills the worker thread in the +intervening time, jobs that returned `Success` will not be run again, +jobs that return `Failure` will have their messages logged and will +not be run again, and jobs that return `Retry` will be queued +again. If something kills the worker thread before these +acknowledgements go through, the job will be retried. Exceptions +triggered within the job cannot affect the worker thread - what they +do to the job is defined at startup (they can cause either a `Failure` +or `Retry`). + +Any deviations from this behavior are considered bugs that will be fixed. + + +## Redis Operations + +Under the hood, we will have the following data structures in redis +(`name` is set when you create the `hworker` instance): + +`hworker-jobs-name`: list of json serialized job descriptions + +`hworker-progress-name`: a hash of jobs that are in progress, mapping to time started + +`hworker-broken-name`: a hash of jobs to time that couldn't be deserialized; most likely means you changed the serialization format with jobs still in queue, _or_ you pointed different applications at the same queues. + +`hworker-failed-queue`: a record of the jobs that failed (limited in size based on config). + +In the following pseudo-code, I'm using `MULTI`...`EXEC` to indicate +atomic blocks of code. These are actually implemented with lua and +`EVAL`, but I think it's easier to read this way. If you want to see +what's actually happening, just read the code - it's not very long! + +When a worker wants to do work, the following happens: + +``` +now = TIME +MULTI +v = RPOP hworker-jobs-name +if v + HSET hworker-progress-name v now +EXEC +v +``` + +When it completes the job, it does the following: + +``` +v = JOB +HDEL hwork-progress v +``` + +If the job returned `Retry`, the following occurs: + +``` +v = JOB +t = START_TIME +MULTI +LPUSH hwork-jobs v +HDEL hwork-progress t +EXEC +``` + +A monitor runs on another thread that will re-run jobs that stay in +progress for too long (as that indicates that something unknown went +wrong). The operation that it runs periodically is: + +``` +keys = HKEYS (or HSCAN) hwork-progress +for keys as v: + started = HGET hwork-progress v + if started < TIME - timeout + MULTI + RPUSH hwork-jobs v + HDEL hwork-progress v + EXEC +``` + +Note that what the monitor does and `Retry` is slightly different - +the monitor puts jobs on the front of the queue, whereas `Retry` puts +them on the back. + +## Primary Libraries Used + +- hedis +- aeson + +## Contributors + +- Daniel Patterson (@dbp - dbp@dbpmail.net) + +## Build Status + +[![Circle CI](https://circleci.com/gh/dbp/hworker.svg?style=svg&circle-token=b40a5b06c599d457cbaa4d1c00824c98d4768f2f)](https://circleci.com/gh/dbp/hworker) 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/hworker.cabal b/hworker.cabal new file mode 100644 --- /dev/null +++ b/hworker.cabal @@ -0,0 +1,45 @@ +name: hworker +version: 0.1.0.0 +synopsis: A reliable at-least-once job queue built on top of redis. +description: See README. +homepage: http://github.com/dbp/hworker +license: ISC +license-file: LICENSE +author: Daniel Patterson +maintainer: dbp@dbpmail.net +build-type: Simple +extra-source-files: README.md CHANGELOG.md +cabal-version: >=1.10 + +library + exposed-modules: System.Hworker + other-modules: Data.Aeson.Helpers + build-depends: base >= 4.7 && < 5 + , aeson + , hedis >= 0.6.5 + , text + , bytestring + , time >= 1.5 + , attoparsec + , uuid >= 1.2.6 + hs-source-dirs: src + default-language: Haskell2010 + ghc-options: -Wall + +Test-Suite hworker-test + type: exitcode-stdio-1.0 + hs-source-dirs: src test + main-is: Spec.hs + other-modules: Data.Aeson.Helpers + , System.Hworker + build-depends: base >= 4.7 && < 5 + , aeson + , hedis >= 0.6.5 + , text + , bytestring + , time >= 1.5 + , attoparsec + , uuid >= 1.2.6 + , hspec >= 2 + , hspec-contrib + , HUnit diff --git a/src/Data/Aeson/Helpers.hs b/src/Data/Aeson/Helpers.hs new file mode 100644 --- /dev/null +++ b/src/Data/Aeson/Helpers.hs @@ -0,0 +1,20 @@ +module Data.Aeson.Helpers where + +import Data.Aeson +import Data.Aeson.Parser (value) +import Data.Attoparsec.Lazy (Parser) +import qualified Data.Attoparsec.Lazy as L +import qualified Data.ByteString.Lazy as L + +-- NOTE(dbp 2015-06-14): Taken from Data.Aeson.Parser.Internal +decodeWith :: Parser Value -> (Value -> Result a) -> L.ByteString -> Maybe a +decodeWith p to s = + case L.parse p s of + L.Done _ v -> case to v of + Success a -> Just a + _ -> Nothing + _ -> Nothing +{-# INLINE decodeWith #-} + +decodeValue :: FromJSON t => L.ByteString -> Maybe t +decodeValue = decodeWith value fromJSON diff --git a/src/System/Hworker.hs b/src/System/Hworker.hs new file mode 100644 --- /dev/null +++ b/src/System/Hworker.hs @@ -0,0 +1,440 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FunctionalDependencies #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} + +{-| + +This module contains an at-least-once persistent job processing queue +backed by Redis. It depends upon Redis not losing data once it has +acknowledged it, and guaranteeing the atomicity that is specified for +commands like EVAL (ie, that if you do several things within an EVAL, +they will all happen or none will happen). Nothing has been tested +with Redis clusters (and it likely will not work). + +-} + +module System.Hworker + ( -- * Types + Result(..) + , Job(..) + , Hworker + , HworkerConfig(..) + , ExceptionBehavior(..) + , RedisConnection(..) + , defaultHworkerConfig + -- * Managing Workers + , create + , createWith + , destroy + , worker + , monitor + -- * Queuing Jobs + , queue + -- * Inspecting Workers + , jobs + , failed + , broken + -- * Debugging Utilities + , debugger + ) + where + +import Control.Arrow (second) +import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) +import Control.Exception (SomeException, catch) +import Control.Monad (forM, forever, void, when) +import Data.Aeson (FromJSON, ToJSON) +import qualified Data.Aeson as A +import Data.Aeson.Helpers +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as B8 +import qualified Data.ByteString.Lazy as LB +import Data.Either (isRight) +import Data.Maybe (fromJust, mapMaybe) +import Data.Monoid ((<>)) +import Data.Text (Text) +import qualified Data.Text as T +import qualified Data.Text.Encoding as T +import Data.Time.Calendar (Day (..)) +import Data.Time.Clock (NominalDiffTime, UTCTime (..), + diffUTCTime, getCurrentTime) +import qualified Data.UUID as UUID +import qualified Data.UUID.V4 as UUID +import qualified Database.Redis as R +import GHC.Generics (Generic) + +-- | Jobs can return 'Success', 'Retry' (with a message), or 'Failure' +-- (with a message). Jobs that return 'Failure' are stored in the +-- 'failed' queue and are not re-run. Jobs that return 'Retry' are re-run. +data Result = Success + | Retry Text + | Failure Text + deriving (Generic, Show) +instance ToJSON Result +instance FromJSON Result + +-- | Each Worker that you create will be responsible for one type of +-- job, defined by a 'Job' instance. +-- +-- The job can do many different things (as the value can be a +-- variant), but be careful not to break deserialization if you add +-- new things it can do. +-- +-- The job will take some state (passed as the `s` parameter), which +-- does not vary based on the job, and the actual job data +-- structure. The data structure (the `t` parameter) will be stored +-- and copied a few times in Redis while in the lifecycle, so +-- generally it is a good idea for it to be relatively small (and have +-- it be able to look up data that it needs while the job in running). +-- +-- Finally, while deriving FromJSON and ToJSON instances automatically +-- might seem like a good idea, you will most likely be better off +-- defining them manually, so you can make sure they are backwards +-- compatible if you change them, as any jobs that can't be +-- deserialized will not be run (and will end up in the 'broken' +-- queue). This will only happen if the queue is non-empty when you +-- replce the running application version, but this is obviously +-- possible and could be likely depending on your use. +class (FromJSON t, ToJSON t, Show t) => Job s t | s -> t where + job :: s -> t -> IO Result + +data JobData t = JobData UTCTime t + +-- | What should happen when an unexpected exception is thrown in a +-- job - it can be treated as either a 'Failure' (the default) or a +-- 'Retry' (if you know the only exceptions are triggered by +-- intermittent problems). +data ExceptionBehavior = RetryOnException | FailOnException + +hwlog :: Show a => Hworker s t -> a -> IO () +hwlog hw a = hworkerLogger hw (hworkerName hw, a) + +-- | The worker data type - it is parametrized be the worker +-- state (the `s`) and the job type (the `t`). +data Hworker s t = + Hworker { hworkerName :: ByteString + , hworkerState :: s + , hworkerConnection :: R.Connection + , hworkerExceptionBehavior :: ExceptionBehavior + , hworkerLogger :: forall a. Show a => a -> IO () + , hworkerJobTimeout :: NominalDiffTime + , hworkerFailedQueueSize :: Int + , hworkerDebug :: Bool + } + +-- | When configuring a worker, you can tell it to use an existing +-- redis connection pool (which you may have for the rest of your +-- application). Otherwise, you can specify connection info. By +-- default, hworker tries to connect to localhost, which may not be +-- true for your production application. +data RedisConnection = RedisConnectInfo R.ConnectInfo + | RedisConnection R.Connection + +-- | The main configuration for workers. +-- +-- Each pool of workers should have a unique `hwconfigName`, as the +-- queues are set up by that name, and if you have different types of +-- data written in, they will likely be unable to be deserialized (and +-- thus could end up in the 'broken' queue). +-- +-- The 'hwconfigLogger' defaults to writing to stdout, so you will +-- likely want to replace that with something appropriate (like from a +-- logging package). +-- +-- The `hwconfigTimeout` is really important. It determines the length +-- of time after a job is started before the 'monitor' will decide +-- that the job must have died and will restart it. If it is shorter +-- than the length of time that a normal job takes to complete, the +-- jobs _will_ be run multiple times. This is _semantically_ okay, as +-- this is an at-least-once processor, but obviously won't be +-- desirable. It defaults to 120 seconds. +-- +-- The 'hwconfigExceptionBehavior' controls what happens when an +-- exception is thrown within a job. +-- +-- 'hwconfigFailedQueueSize' controls how many 'failed' jobs will be +-- kept. It defaults to 1000. +data HworkerConfig s = + HworkerConfig { + hwconfigName :: Text + , hwconfigState :: s + , hwconfigRedisConnectInfo :: RedisConnection + , hwconfigExceptionBehavior :: ExceptionBehavior + , hwconfigLogger :: forall a. Show a => a -> IO () + , hwconfigTimeout :: NominalDiffTime + , hwconfigFailedQueueSize :: Int + , hwconfigDebug :: Bool + } + +-- | The default worker config - it needs a name and a state (as those +-- will always be unique). +defaultHworkerConfig :: Text -> s -> HworkerConfig s +defaultHworkerConfig name state = + HworkerConfig name + state + (RedisConnectInfo R.defaultConnectInfo) + FailOnException + print + 120 + 1000 + False + +-- | Create a new worker with the default 'HworkerConfig'. +-- +-- Note that you must create at least one 'worker' and 'monitor' for +-- the queue to actually process jobs (and for it to retry ones that +-- time-out). +create :: Job s t => Text -> s -> IO (Hworker s t) +create name state = createWith (defaultHworkerConfig name state) + +-- | Create a new worker with a specified 'HworkerConfig'. +-- +-- Note that you must create at least one 'worker' and 'monitor' for +-- the queue to actually process jobs (and for it to retry ones that +-- time-out). +createWith :: Job s t => HworkerConfig s -> IO (Hworker s t) +createWith HworkerConfig{..} = + do conn <- case hwconfigRedisConnectInfo of + RedisConnectInfo c -> R.connect c + RedisConnection c -> return c + return $ Hworker (T.encodeUtf8 hwconfigName) + hwconfigState + conn + hwconfigExceptionBehavior + hwconfigLogger + hwconfigTimeout + hwconfigFailedQueueSize + hwconfigDebug + +-- | Destroy a worker. This will delete all the queues, clearing out +-- all existing 'jobs', the 'broken' and 'failed' queues. There is no need +-- to do this in normal applications (and most likely, you won't want to). +destroy :: Job s t => Hworker s t -> IO () +destroy hw = void $ R.runRedis (hworkerConnection hw) $ + R.del [ jobQueue hw + , progressQueue hw + , brokenQueue hw + , failedQueue hw + ] + +jobQueue :: Hworker s t -> ByteString +jobQueue hw = "hworker-jobs-" <> hworkerName hw + +progressQueue :: Hworker s t -> ByteString +progressQueue hw = "hworker-progress-" <> hworkerName hw + +brokenQueue :: Hworker s t -> ByteString +brokenQueue hw = "hworker-broken-" <> hworkerName hw + +failedQueue :: Hworker s t -> ByteString +failedQueue hw = "hworker-failed-" <> hworkerName hw + +-- | Adds a job to the queue. Returns whether the operation succeeded. +queue :: Job s t => Hworker s t -> t -> IO Bool +queue hw j = + do job_id <- UUID.toString <$> UUID.nextRandom + isRight <$> R.runRedis (hworkerConnection hw) + (R.lpush (jobQueue hw) [LB.toStrict $ A.encode (job_id, j)]) + +-- | Creates a new worker thread. This is blocking, so you will want to +-- 'forkIO' this into a thread. You can have any number of these (and +-- on any number of servers); the more there are, the faster jobs will +-- be processed. +worker :: Job s t => Hworker s t -> IO () +worker hw = + do now <- getCurrentTime + r <- R.runRedis (hworkerConnection hw) $ + R.eval "local job = redis.call('rpop',KEYS[1])\n\ + \if job ~= nil then\n\ + \ redis.call('hset', KEYS[2], job, ARGV[1])\n\ + \ return job\n\ + \else\n\ + \ return nil\n\ + \end" + [jobQueue hw, progressQueue hw] + [LB.toStrict $ A.encode now] + case r of + Left err -> hwlog hw err >> delayAndRun + Right Nothing -> delayAndRun + Right (Just t) -> + do when (hworkerDebug hw) $ hwlog hw ("WORKER RUNNING", t) + case decodeValue (LB.fromStrict t) of + Nothing -> do hwlog hw ("BROKEN JOB", t) + now <- getCurrentTime + withNil hw (R.eval "local del = redis.call('hdel', KEYS[1], ARGV[1])\n\ + \if del == 1 then\n\ + \ redis.call('hset', KEYS[2], ARGV[1], ARGV[2])\n\ + \end\n\ + \return nil" + [progressQueue hw, brokenQueue hw] + [t, LB.toStrict $ A.encode now]) + delayAndRun + Just (_ :: String, j) -> do + result <- runJob (job (hworkerState hw) j) + case result of + Success -> + do when (hworkerDebug hw) $ hwlog hw ("JOB COMPLETE", t) + delete_res <- R.runRedis (hworkerConnection hw) + (R.hdel (progressQueue hw) [t]) + case delete_res of + Left err -> hwlog hw err >> delayAndRun + Right 1 -> justRun + Right n -> do hwlog hw ("Job done: did not delete 1, deleted " <> show n) + delayAndRun + Retry msg -> + do hwlog hw ("Retry: " <> msg) + withNil hw + (R.eval "local del = redis.call('hdel', KEYS[1], ARGV[1])\n\ + \if del == 1 then\n\ + \ redis.call('lpush', KEYS[2], ARGV[1])\n\ + \end\n\ + \return nil" + [progressQueue hw, jobQueue hw] + [t]) + delayAndRun + Failure msg -> + do hwlog hw ("Failure: " <> msg) + withNil hw + (R.eval "local del = redis.call('hdel', KEYS[1], ARGV[1])\n\ + \if del == 1 then\n\ + \ redis.call('lpush', KEYS[2], ARGV[1])\n\ + \ redis.call('ltrim', KEYS[2], 0, ARGV[2])\n\ + \end\n\ + \return nil" + [progressQueue hw, failedQueue hw] + [t, B8.pack (show (hworkerFailedQueueSize hw - 1))]) + void $ R.runRedis (hworkerConnection hw) + (R.hdel (progressQueue hw) [t]) + delayAndRun + where delayAndRun = threadDelay 10000 >> worker hw + justRun = worker hw + runJob v = + do x <- newEmptyMVar + jt <- forkIO (catch (v >>= putMVar x . Right) + (\(e::SomeException) -> + putMVar x (Left e))) + res <- takeMVar x + case res of + Left e -> + let b = case hworkerExceptionBehavior hw of + RetryOnException -> Retry + FailOnException -> Failure in + return (b ("Exception raised: " <> (T.pack . show) e)) + Right r -> return r + + +-- | Start a monitor. Like 'worker', this is blocking, so should be +-- started in a thread. This is responsible for retrying jobs that +-- time out (which can happen if the processing thread is killed, for +-- example). You need to have at least one of these running to have +-- the retry happen, but it is safe to have any number running. +monitor :: Job s t => Hworker s t -> IO () +monitor hw = + forever $ + do now <- getCurrentTime + withList hw (R.hkeys (progressQueue hw)) + (\jobs -> + void $ forM jobs $ \job -> + withMaybe hw (R.hget (progressQueue hw) job) + (\start -> + when (diffUTCTime now (fromJust $ decodeValue (LB.fromStrict start)) > hworkerJobTimeout hw) $ + do n <- + withInt hw + (R.eval "local del = redis.call('hdel', KEYS[2], ARGV[1])\n\ + \if del == 1 then\ + \ redis.call('rpush', KEYS[1], ARGV[1])\n\ \end\n\ + \return del" + [jobQueue hw, progressQueue hw] + [job]) + when (hworkerDebug hw) $ hwlog hw ("MONITOR RV", n) + when (hworkerDebug hw && n == 1) $ hwlog hw ("MONITOR REQUEUED", job))) + -- NOTE(dbp 2015-07-25): We check every 1/10th of timeout. + threadDelay (floor $ 100000 * hworkerJobTimeout hw) + +-- | Returns the jobs that could not be deserialized, most likely +-- because you changed the 'ToJSON'/'FromJSON' instances for you job +-- in a way that resulted in old jobs not being able to be converted +-- back from json. Another reason for jobs to end up here (and much +-- worse) is if you point two instances of 'Hworker', with different +-- job types, at the same queue (ie, you re-use the name). Then +-- anytime a worker from one queue gets a job from the other it would +-- think it is broken. +broken :: Hworker s t -> IO [(ByteString, UTCTime)] +broken hw = do r <- R.runRedis (hworkerConnection hw) (R.hgetall (brokenQueue hw)) + case r of + Left err -> hwlog hw err >> return [] + Right xs -> return (map (second parseTime) xs) + where parseTime = fromJust . decodeValue . LB.fromStrict + +jobsFromQueue :: Job s t => Hworker s t -> ByteString -> IO [t] +jobsFromQueue hw queue = + do r <- R.runRedis (hworkerConnection hw) (R.lrange queue 0 (-1)) + case r of + Left err -> hwlog hw err >> return [] + Right [] -> return [] + Right xs -> return $ mapMaybe (fmap (\(_::String, x) -> x) . decodeValue . LB.fromStrict) xs + +-- | Returns all pending jobs. +jobs :: Job s t => Hworker s t -> IO [t] +jobs hw = jobsFromQueue hw (jobQueue hw) + +-- | Returns all failed jobs. This is capped at the most recent +-- 'hworkerconfigFailedQueueSize' jobs that returned 'Failure' (or +-- threw an exception when 'hworkerconfigExceptionBehavior' is +-- 'FailOnException'). +failed :: Job s t => Hworker s t -> IO [t] +failed hw = jobsFromQueue hw (failedQueue hw) + +-- | Logs the contents of the jobqueue and the inprogress queue at +-- `microseconds` intervals. +debugger :: Job s t => Int -> Hworker s t -> IO () +debugger microseconds hw = + forever $ + do withList hw (R.hkeys (progressQueue hw)) + (\running -> + withList hw (R.lrange (jobQueue hw) 0 (-1)) + (\queued -> hwlog hw ("DEBUG", queued, running))) + threadDelay microseconds + +-- Redis helpers follow +withList hw a f = + do r <- R.runRedis (hworkerConnection hw) a + case r of + Left err -> hwlog hw err + Right [] -> return () + Right xs -> f xs + +withMaybe hw a f = + do r <- R.runRedis (hworkerConnection hw) a + case r of + Left err -> hwlog hw err + Right Nothing -> return () + Right (Just v) -> f v + +withNil hw a = + do r <- R.runRedis (hworkerConnection hw) a + case r of + Left err -> hwlog hw err + Right (Just ("" :: ByteString)) -> return () + Right _ -> return () + +withInt :: Hworker s t -> R.Redis (Either R.Reply Integer) -> IO Integer +withInt hw a = + do r <- R.runRedis (hworkerConnection hw) a + case r of + Left err -> hwlog hw err >> return (-1) + Right n -> return n + +withIgnore :: Hworker s t -> R.Redis (Either R.Reply a) -> IO () +withIgnore hw a = + do r <- R.runRedis (hworkerConnection hw) a + case r of + Left err -> hwlog hw err + Right _ -> return () diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,389 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +import Control.Concurrent (forkIO, killThread, threadDelay) +import Control.Concurrent.MVar (MVar, modifyMVarMasked_, newMVar, + readMVar, takeMVar) +import Control.Monad (replicateM_) +import Data.Aeson (FromJSON, ToJSON) +import qualified Data.Text as T +import GHC.Generics (Generic) +import System.Hworker +import System.IO + +import Test.Hspec +import Test.Hspec.Contrib.HUnit +import Test.HUnit + +data SimpleJob = SimpleJob deriving (Generic, Show, Eq) +data SimpleState = SimpleState { unSimpleState :: MVar Int } +instance ToJSON SimpleJob +instance FromJSON SimpleJob +instance Job SimpleState SimpleJob where + job (SimpleState mvar) SimpleJob = + do modifyMVarMasked_ mvar (return . (+1)) + return Success + +data ExJob = ExJob deriving (Generic, Show) +data ExState = ExState { unExState :: MVar Int } +instance ToJSON ExJob +instance FromJSON ExJob +instance Job ExState ExJob where + job (ExState mvar) ExJob = + do modifyMVarMasked_ mvar (return . (+1)) + v <- readMVar mvar + if v > 1 + then return Success + else error "ExJob: failing badly!" + +data RetryJob = RetryJob deriving (Generic, Show) +data RetryState = RetryState { unRetryState :: MVar Int } +instance ToJSON RetryJob +instance FromJSON RetryJob +instance Job RetryState RetryJob where + job (RetryState mvar) RetryJob = + do modifyMVarMasked_ mvar (return . (+1)) + v <- readMVar mvar + if v > 1 + then return Success + else return (Retry "RetryJob retries") + +data FailJob = FailJob deriving (Eq, Generic, Show) +data FailState = FailState { unFailState :: MVar Int } +instance ToJSON FailJob +instance FromJSON FailJob +instance Job FailState FailJob where + job (FailState mvar) FailJob = + do modifyMVarMasked_ mvar (return . (+1)) + v <- readMVar mvar + if v > 1 + then return Success + else return (Failure "FailJob fails") + +data AlwaysFailJob = AlwaysFailJob deriving (Eq, Generic, Show) +data AlwaysFailState = AlwaysFailState { unAlwaysFailState :: MVar Int } +instance ToJSON AlwaysFailJob +instance FromJSON AlwaysFailJob +instance Job AlwaysFailState AlwaysFailJob where + job (AlwaysFailState mvar) AlwaysFailJob = + do modifyMVarMasked_ mvar (return . (+1)) + return (Failure "AlwaysFailJob fails") + +data TimedJob = TimedJob Int deriving (Generic, Show, Eq) +data TimedState = TimedState { unTimedState :: MVar Int } +instance ToJSON TimedJob +instance FromJSON TimedJob +instance Job TimedState TimedJob where + job (TimedState mvar) (TimedJob delay) = + do threadDelay delay + modifyMVarMasked_ mvar (return . (+1)) + return Success + +data BigJob = BigJob T.Text deriving (Generic, Show, Eq) +data BigState = BigState { unBigState :: MVar Int } +instance ToJSON BigJob +instance FromJSON BigJob +instance Job BigState BigJob where + job (BigState mvar) (BigJob _) = + do modifyMVarMasked_ mvar (return . (+1)) + return Success + +nullLogger :: Show a => a -> IO () +nullLogger = const (return ()) + +print' :: Show a => a -> IO () +print' a = do print a + hFlush stdout + +conf n s = (defaultHworkerConfig n s) { + hwconfigLogger = nullLogger + , hwconfigExceptionBehavior = FailOnException + , hwconfigTimeout = 4 + } + +main :: IO () +main = hspec $ + do describe "Simple" $ + do it "should run and increment counter" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "simpleworker-1" + (SimpleState mvar)) + wthread <- forkIO (worker hworker) + queue hworker SimpleJob + threadDelay 30000 + killThread wthread + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 1 after job runs" 1 v + it "queueing 2 jobs should increment twice" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "simpleworker-2" + (SimpleState mvar)) + wthread <- forkIO (worker hworker) + queue hworker SimpleJob + queue hworker SimpleJob + threadDelay 40000 + killThread wthread + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 2 after 2 jobs run" 2 v + it "queueing 1000 jobs should increment 1000" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "simpleworker-3" + (SimpleState mvar)) + wthread <- forkIO (worker hworker) + replicateM_ 1000 (queue hworker SimpleJob) + threadDelay 2000000 + killThread wthread + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 1000 after 1000 job runs" 1000 v + it "should work with multiple workers" $ + -- NOTE(dbp 2015-07-12): This probably won't run faster, because + -- they are all blocking on the MVar, but that's not the point. + do mvar <- newMVar 0 + hworker <- createWith (conf "simpleworker-4" + (SimpleState mvar)) + wthread1 <- forkIO (worker hworker) + wthread2 <- forkIO (worker hworker) + wthread3 <- forkIO (worker hworker) + wthread4 <- forkIO (worker hworker) + replicateM_ 1000 (queue hworker SimpleJob) + threadDelay 1000000 + killThread wthread1 + killThread wthread2 + killThread wthread3 + killThread wthread4 + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 1000 after 1000 job runs" 1000 v + + describe "Exceptions" $ + do it "should be able to have exceptions thrown in jobs and retry the job" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "exworker-1" + (ExState mvar)) { + hwconfigExceptionBehavior = + RetryOnException + } + wthread <- forkIO (worker hworker) + queue hworker ExJob + threadDelay 40000 + killThread wthread + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 2, since the first run failed" 2 v + it "should not retry if mode is FailOnException" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "exworker-2" + (ExState mvar)) + wthread <- forkIO (worker hworker) + queue hworker ExJob + threadDelay 30000 + killThread wthread + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 1, since failing run wasn't retried" 1 v + + describe "Retry" $ + do it "should be able to return Retry and get run again" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "retryworker-1" + (RetryState mvar)) + wthread <- forkIO (worker hworker) + queue hworker RetryJob + threadDelay 50000 + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 2, since it got retried" 2 v + + describe "Fail" $ + do it "should not retry a job that Fails" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "failworker-1" + (FailState mvar)) + wthread <- forkIO (worker hworker) + queue hworker FailJob + threadDelay 30000 + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 1, since failing run wasn't retried" 1 v + it "should put a failed job into the failed queue" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "failworker-2" + (FailState mvar)) + wthread <- forkIO (worker hworker) + queue hworker FailJob + threadDelay 30000 + jobs <- failed hworker + destroy hworker + assertEqual "Should have failed job" [FailJob] jobs + it "should only store failedQueueSize failed jobs" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "failworker-3" + (AlwaysFailState mvar)) { + hwconfigFailedQueueSize = 2 + } + wthread <- forkIO (worker hworker) + queue hworker AlwaysFailJob + queue hworker AlwaysFailJob + queue hworker AlwaysFailJob + queue hworker AlwaysFailJob + threadDelay 100000 + jobs <- failed hworker + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 4, since all jobs were run" 4 v + assertEqual "Should only have stored 2" + [AlwaysFailJob,AlwaysFailJob] jobs + describe "Monitor" $ + do it "should add job back after timeout" $ + -- NOTE(dbp 2015-07-12): The timing on this test is somewhat + -- tricky. We want to get the job started with one worker, + -- then kill the worker, then start a new worker, and have + -- the monitor put the job back in the queue and have the + -- second worker finish it. It's important that the job + -- takes less time to complete than the timeout for the + -- monitor, or else it'll queue it forever. + -- + -- The timeout is 5 seconds. The job takes 1 seconds to run. + -- The worker is killed after 0.5 seconds, which should be + -- plenty of time for it to have started the job. Then after + -- the second worker is started, we wait 10 seconds, which + -- should be plenty; we expect the total run to take around 11. + do mvar <- newMVar 0 + hworker <- createWith (conf "timedworker-1" + (TimedState mvar)) { + hwconfigTimeout = 5 + } + wthread1 <- forkIO (worker hworker) + mthread <- forkIO (monitor hworker) + queue hworker (TimedJob 1000000) + threadDelay 500000 + killThread wthread1 + wthread2 <- forkIO (worker hworker) + threadDelay 10000000 + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 2, since monitor thinks it failed" 2 v + it "should add back multiple jobs after timeout" $ + -- NOTE(dbp 2015-07-23): Similar to the above test, but we + -- have multiple jobs started, multiple workers killed. + -- then one worker will finish both interrupted jobs. + do mvar <- newMVar 0 + hworker <- createWith (conf "timedworker-2" + (TimedState mvar)) { + hwconfigTimeout = 5 + } + wthread1 <- forkIO (worker hworker) + wthread2 <- forkIO (worker hworker) + mthread <- forkIO (monitor hworker) + queue hworker (TimedJob 1000000) + queue hworker (TimedJob 1000000) + threadDelay 500000 + killThread wthread1 + killThread wthread2 + wthread3 <- forkIO (worker hworker) + threadDelay 10000000 + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 4, since monitor thinks first 2 failed" 4 v + it "should work with multiple monitors" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "timedworker-3" + (TimedState mvar)) { + hwconfigTimeout = 5 + } + wthread1 <- forkIO (worker hworker) + wthread2 <- forkIO (worker hworker) + -- NOTE(dbp 2015-07-24): This might seem silly, but it + -- was actually sufficient to expose a race condition. + mthread1 <- forkIO (monitor hworker) + mthread2 <- forkIO (monitor hworker) + mthread3 <- forkIO (monitor hworker) + mthread4 <- forkIO (monitor hworker) + mthread5 <- forkIO (monitor hworker) + mthread6 <- forkIO (monitor hworker) + queue hworker (TimedJob 1000000) + queue hworker (TimedJob 1000000) + threadDelay 500000 + killThread wthread1 + killThread wthread2 + wthread3 <- forkIO (worker hworker) + threadDelay 30000000 + destroy hworker + v <- takeMVar mvar + assertEqual "State should be 4, since monitor thinks first 2 failed" 4 v + -- NOTE(dbp 2015-07-24): It would be really great to have a + -- test that went after a race between the retry logic and + -- the monitors (ie, assume that the job completed with + -- Retry, and it happened to complete right at the timeout + -- period). I'm not sure if I could get that sort of + -- precision without adding other delay mechanisms, or + -- something to make it more deterministic. + describe "Broken jobs" $ + it "should store broken jobs" $ + do -- NOTE(dbp 2015-08-09): The more common way this could + -- happen is that you change your serialization format. But + -- we can abuse this by creating two different workers + -- pointing to the same queue, and submit jobs in one, try + -- to run them in another, where the types are different. + mvar <- newMVar 0 + hworker1 <- createWith (conf "broken-1" + (TimedState mvar)) { + hwconfigTimeout = 5 + } + hworker2 <- createWith (conf "broken-1" + (SimpleState mvar)) { + hwconfigTimeout = 5 + } + wthread <- forkIO (worker hworker1) + queue hworker2 SimpleJob + threadDelay 100000 + jobs <- broken hworker2 + killThread wthread + destroy hworker1 + v <- takeMVar mvar + assertEqual "State should be 0, as nothing should have happened" 0 v + assertEqual "Should be one broken job, as serialization is wrong" 1 (length jobs) + describe "Dump jobs" $ do + it "should return the job that was queued" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "dump-1" + (SimpleState mvar)) { + hwconfigTimeout = 5 + } + queue hworker SimpleJob + res <- jobs hworker + destroy hworker + assertEqual "Should be [SimpleJob]" [SimpleJob] res + it "should return jobs in order (most recently added at front; worker pulls from back)" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "dump-2" + (TimedState mvar)) { + hwconfigTimeout = 5 + } + queue hworker (TimedJob 1) + queue hworker (TimedJob 2) + res <- jobs hworker + destroy hworker + assertEqual "Should by [TimedJob 2, TimedJob 1]" [TimedJob 2, TimedJob 1] res + describe "Large jobs" $ do + it "should be able to deal with lots of large jobs" $ + do mvar <- newMVar 0 + hworker <- createWith (conf "big-1" + (BigState mvar)) + wthread1 <- forkIO (worker hworker) + wthread2 <- forkIO (worker hworker) + wthread3 <- forkIO (worker hworker) + wthread4 <- forkIO (worker hworker) + let content = T.intercalate "\n" (take 1000 (repeat "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")) + replicateM_ 5000 (queue hworker (BigJob content)) + threadDelay 10000000 + killThread wthread1 + killThread wthread2 + killThread wthread3 + killThread wthread4 + destroy hworker + v <- takeMVar mvar + assertEqual "Should have processed 5000" 5000 v