hworker (empty) → 0.1.0.0
raw patch · 8 files changed
+1058/−0 lines, 8 filesdep +HUnitdep +aesondep +attoparsecsetup-changed
Dependencies added: HUnit, aeson, attoparsec, base, bytestring, hedis, hspec, hspec-contrib, text, time, uuid
Files
- CHANGELOG.md +3/−0
- LICENSE +13/−0
- README.md +146/−0
- Setup.hs +2/−0
- hworker.cabal +45/−0
- src/Data/Aeson/Helpers.hs +20/−0
- src/System/Hworker.hs +440/−0
- test/Spec.hs +389/−0
+ CHANGELOG.md view
@@ -0,0 +1,3 @@+* 0.1.0.0 Daniel Patterson <dbp@dbpmail.net> 2015-10-25++ Initial release.
+ LICENSE view
@@ -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.
+ README.md view
@@ -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++[](https://circleci.com/gh/dbp/hworker)
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ hworker.cabal view
@@ -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
+ src/Data/Aeson/Helpers.hs view
@@ -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
+ src/System/Hworker.hs view
@@ -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 ()
+ test/Spec.hs view
@@ -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