git.iko.soy/odd-jobs
on master help sign in
Merge pull request #92 from SupercedeTech/expose-runner-env-upstream
583f720Jappie Klooster
src/OddJobs/Job.hs+66−55
compare with master
unifiedsplit
@@ -38,6 +38,7 @@ module OddJobs.Job
38 , AllJobTypes(..)
38 , AllJobTypes(..)
39 , ResourceId(..)
39 , ResourceId(..)
40 , FunctionName
40 , FunctionName
41+ , RunnerEnv(..)
41
42
42 -- ** Structured logging
43 -- ** Structured logging
43 --
44 --
@@ -52,6 +53,7 @@ module OddJobs.Job
52 , jobEventListener
53 , jobEventListener
53 , jobPoller
54 , jobPoller
54 , jobPollingSql
55 , jobPollingSql
56+ , pollRunJob
55 , JobRunner
57 , JobRunner
56 , HasJobRunner (..)
58 , HasJobRunner (..)
57
59
@@ -374,6 +376,9 @@ updateJobHelper tname conn (newStatus, existingStates, mRunAt, jid) =
374data TimeoutException = TimeoutException deriving (Eq, Show)
376data TimeoutException = TimeoutException deriving (Eq, Show)
375instance Exception TimeoutException
377instance Exception TimeoutException
376
378
379+-- | this spawns the job on a a seperate thread, and
380+-- also spawns a monitoring thread, whichever finishes
381+-- first kills the other.
377runJobWithTimeout :: (HasJobRunner m)
382runJobWithTimeout :: (HasJobRunner m)
378 => Seconds
383 => Seconds
379 -> Job
384 -> Job
@@ -397,6 +402,7 @@ runJobWithTimeout timeoutSec job = do
397 (atomicModifyIORef' threadsRef $ \threads -> (DL.delete a threads, ()))
402 (atomicModifyIORef' threadsRef $ \threads -> (DL.delete a threads, ()))
398
403
399
404
405+-- | runs a job, blocks for as long as it's in progress
400runJob :: (HasJobRunner m) => JobId -> m ()
406runJob :: (HasJobRunner m) => JobId -> m ()
401runJob jid = do
407runJob jid = do
402 findJobById jid >>= \case
408 findJobById jid >>= \case
@@ -559,60 +565,67 @@ jobPollingIO pollerDbConn processName tname lockTimeout = do
559jobPoller :: (HasJobRunner m) => m ()
565jobPoller :: (HasJobRunner m) => m ()
560jobPoller = do
566jobPoller = do
561 processName <- liftIO jobWorkerName
567 processName <- liftIO jobWorkerName
562− pool <- getDbPool
563− tname <- getTableName
564− lockTimeout <- getDefaultJobTimeout
565− traceM "jobPoller just before log statement"
566 log LevelInfo $ LogText $ toS $ "Starting the job monitor via DB polling with processName=" <> processName
568 log LevelInfo $ LogText $ toS $ "Starting the job monitor via DB polling with processName=" <> processName
567−
568− let poll conn mResCfg = join $ mask_ $ do
569− log LevelDebug $ LogText $ toS $ "[" <> processName <> "] Polling the job queue.."
570− t <- liftIO getCurrentTime
571− r <- case mResCfg of
572− Nothing -> liftIO $
573− PGS.query conn jobPollingSql
574− ( tname
575− , Locked
576− , t
577− , processName
578− , tname
579− , t
580− , In [Queued, Retry]
581− , Locked
582− , addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t)
583− Just ResourceCfg{..} -> liftIO $
584− PGS.query conn jobPollingWithResourceSql
585− ( tname
586− , Locked
587− , t
588− , processName
589− , tname
590− , t
591− , In [Queued, Retry]
592− , Locked
593− , addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t
594− , resCfgCheckResourceFunction
595− )
596− case r of
597− -- When we don't have any jobs to run, we can relax a bit...
598− [] -> pure delayAction
599−
600− -- When we find a job to run, fork and try to find the next job without any delay...
601− [Only (jid :: JobId)] -> do
602− void $ async $ runJob jid
603− pure noDelayAction
604−
605− x -> error $ "WTF just happened? I was supposed to get only a single row, but got: " ++ show x
606−
607 concurrencyControlFn <- getConcurrencyControlFn
569 concurrencyControlFn <- getConcurrencyControlFn
608− withResource pool $ \pollerDbConn -> forever $ concurrencyControlFn pollerDbConn >>= \case
570+ pool <- getDbPool
609− DontPoll -> do
571+ forever $ do
572+ concurencyPolicy <- withResource pool concurrencyControlFn
573+ case concurencyPolicy of
610− log LevelWarn $ LogText "NOT polling the job queue due to concurrency control"
574+ DontPoll -> log LevelWarn $ LogText "NOT polling the job queue due to concurrency control"
611− -- If we can't run any jobs ATM, relax and wait for resources to free up
575+ PollAny -> void $ pollRunJob processName Nothing
576+ PollWithResources resCfg -> void $ pollRunJob processName (Just resCfg)
577+
578+-- | Polls a job and runs it, or executes a delay action if no job was found
579+--
580+-- returns an async to the task that was started, allowing you
581+-- to block until it's finished
582+pollRunJob :: (HasJobRunner m) => String -> Maybe ResourceCfg -> m (Maybe (Async ()))
583+pollRunJob processName mResCfg = do
584+ tname <- getTableName
585+ -- note it's better to use fine grained pool connection,
586+ -- since data.pool already has it's internal resource cache.
587+ -- this way the user can configure how long a connection
588+ -- needs to remain open.
589+ pool <- getDbPool
590+ lockTimeout <- getDefaultJobTimeout
591+ join $ withResource pool $ \pollerDbConn -> mask_ $ do
592+ log LevelDebug $ LogText $ toS $ "[" <> processName <> "] Polling the job queue.."
593+ t <- liftIO getCurrentTime
594+ r <- case mResCfg of
595+ Nothing -> liftIO $
596+ PGS.query pollerDbConn jobPollingSql
597+ ( tname
598+ , Locked
599+ , t
600+ , processName
601+ , tname
602+ , t
603+ , In [Queued, Retry]
604+ , Locked
605+ , addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t)
606+ Just ResourceCfg{..} -> liftIO $
607+ PGS.query pollerDbConn jobPollingWithResourceSql
608+ ( tname
609+ , Locked
610+ , t
611+ , processName
612+ , tname
613+ , t
614+ , In [Queued, Retry]
615+ , Locked
616+ , addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t
617+ , resCfgCheckResourceFunction
618+ )
619+ case r of
620+ -- When we don't have any jobs to run, we can relax a bit...
612− delayAction
621+ [] -> pure (Nothing <$ delayAction)
613− PollAny -> poll pollerDbConn Nothing
622+
614− PollWithResources resCfg -> poll pollerDbConn (Just resCfg)
623+ -- When we find a job to run, fork and try to find the next job without any delay...
615−
624+ [Only (jid :: JobId)] -> do
625+ x <- async $ runJob jid
626+ pure $ Just x <$ noDelayAction
627+
628+ x -> error $ "WTF just happened? I was supposed to get only a single row, but got: " ++ show x
616 where
629 where
617 delayAction = delaySeconds =<< getPollingInterval
630 delayAction = delaySeconds =<< getPollingInterval
618 noDelayAction = pure ()
631 noDelayAction = pure ()
@@ -740,7 +753,7 @@ scheduleJob conn tname payload runAt = do
740 case rs of
753 case rs of
741 [] -> Prelude.error . (<> "Not expecting a blank result set when creating a job. Query=") <$> queryFormatter
754 [] -> Prelude.error . (<> "Not expecting a blank result set when creating a job. Query=") <$> queryFormatter
742 [r] -> pure r
755 [r] -> pure r
743− _ -> Prelude.error . (<> "Not expecting multiple rows when creating a single job. Query=") <$> queryFormatter
756+ _ -> Prelude.error . (<> "Not expecting multiple rows when creating a single job. Query=") <$> queryFormatter
744
757
745type ResourceList = [(ResourceId, Int)]
758type ResourceList = [(ResourceId, Int)]
746
759
@@ -786,7 +799,6 @@ scheduleJobWithResources conn tname ResourceCfg{..} payload resources runAt = do
786
799
787 pure job
800 pure job
788
801
789−
790-- getRunnerEnv :: (HasJobRunner m) => m RunnerEnv
802-- getRunnerEnv :: (HasJobRunner m) => m RunnerEnv
791-- getRunnerEnv = ask
803-- getRunnerEnv = ask
792
804
@@ -840,4 +852,3 @@ fetchAllJobRunners :: (MonadIO m)
840 -> m [JobRunnerName]
852 -> m [JobRunnerName]
841fetchAllJobRunners UIConfig{uicfgTableName, uicfgDbPool} = liftIO $ withResource uicfgDbPool $ \conn -> do
853fetchAllJobRunners UIConfig{uicfgTableName, uicfgDbPool} = liftIO $ withResource uicfgDbPool $ \conn -> do
842 mapMaybe fromOnly <$> PGS.query conn "select distinct locked_by from ?" (Only uicfgTableName)
854 mapMaybe fromOnly <$> PGS.query conn "select distinct locked_by from ?" (Only uicfgTableName)
843−
test/Test.hs+107−70
@@ -47,6 +47,7 @@ import Data.Maybe (fromMaybe)
47import qualified OddJobs.ConfigBuilder as Job
47import qualified OddJobs.ConfigBuilder as Job
48import UnliftIO
48import UnliftIO
49import Control.Exception (ArithException)
49import Control.Exception (ArithException)
50+import Data.Bifunctor(first)
50
51
51$(Aeson.deriveJSON Aeson.defaultOptions ''Seconds)
52$(Aeson.deriveJSON Aeson.defaultOptions ''Seconds)
52
53
@@ -137,7 +138,7 @@ jobRunner Job{jobPayload, jobAttempts} = case fromJSON jobPayload of
137 Success (j :: JobPayload) ->
138 Success (j :: JobPayload) ->
138 let recur pload idx = case pload of
139 let recur pload idx = case pload of
139 PayloadAlwaysFail delay -> delaySeconds delay >> error ("Forced error after " <> show delay <> " seconds")
140 PayloadAlwaysFail delay -> delaySeconds delay >> error ("Forced error after " <> show delay <> " seconds")
140− PayloadSucceed delay -> void $ delaySeconds delay
141+ PayloadSucceed delay -> delaySeconds delay
141 PayloadFail delay innerpload -> if idx<jobAttempts
142 PayloadFail delay innerpload -> if idx<jobAttempts
142 then recur innerpload (idx + 1)
143 then recur innerpload (idx + 1)
143 else delaySeconds delay >> error ("Forced error after " <> show delay <> " seconds. step=" <> show idx)
144 else delaySeconds delay >> error ("Forced error after " <> show delay <> " seconds. step=" <> show idx)
@@ -235,8 +236,35 @@ withRandomTable jobPool action = do
235 (liftIO $ Pool.withResource jobPool (\conn -> liftIO $ Migrations.createJobTable conn tname) >> runInIO (action tname))
236 (liftIO $ Pool.withResource jobPool (\conn -> liftIO $ Migrations.createJobTable conn tname) >> runInIO (action tname))
236 (liftIO $ Pool.withResource jobPool $ \conn -> liftIO $ void $ PGS.execute conn "drop table if exists ?" (Only tname))
237 (liftIO $ Pool.withResource jobPool $ \conn -> liftIO $ void $ PGS.execute conn "drop table if exists ?" (Only tname))
237
238
238−withRandomResourceTables :: (MonadBaseControl
239+runSingleJobFromQueueBlock :: Job.Config -> IO (Maybe (Either SomeException ()))
240+runSingleJobFromQueueBlock cfg =
241+ traverse waitCatch =<< runSingleJobFromQueue cfg
242+
243+-- | waits untill the job has started
244+runSingleJobFromQueue :: Job.Config -> IO (Maybe (Async ()))
245+runSingleJobFromQueue config' = do
246+ r <- liftIO $ newIORef mempty
247+ waitTillJobStart <- newEmptyMVar
248+ let config = config' {
249+ Job.cfgPollingInterval = 0
250+ , Job.cfgJobRunner = \job -> do
251+ putMVar waitTillJobStart ()
252+ jobRunner job
253+ }
254+ let monitorEnv = Job.RunnerEnv
255+ { Job.envConfig = config
256+ , Job.envJobThreadsRef = r
257+ }
258+ result <- runReaderT (Job.pollRunJob "runSingleJobFromQueue" $ readResourceConfig config) monitorEnv
259+ forM_ result $ const $ readMVar waitTillJobStart
260+ pure result
261+
262+readResourceConfig :: Job.Config -> Maybe Job.ResourceCfg
263+readResourceConfig cfg = case Job.cfgConcurrencyControl cfg of
264+ Job.ResourceLimits res -> Just res
265+ _ -> Nothing
266+
239− IO m, MonadUnliftIO m) => Int -> Pool Connection -> Job.TableName -> (Job.ResourceCfg -> m a) -> m a
267+withRandomResourceTables :: (MonadBaseControl IO m, MonadUnliftIO m) => Int -> Pool Connection -> Job.TableName -> (Job.ResourceCfg -> m a) -> m a
240withRandomResourceTables defaultLimit jobPool tname action = do
268withRandomResourceTables defaultLimit jobPool tname action = do
241 resCfgResourceTable <- liftIO $ fromString . ("resources_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
269 resCfgResourceTable <- liftIO $ fromString . ("resources_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
242 resCfgUsageTable <- liftIO $ fromString . ("usage_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
270 resCfgUsageTable <- liftIO $ fromString . ("usage_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
@@ -262,7 +290,13 @@ withNewJobMonitor jobPool actualTest = do
262 withRandomTable jobPool $ \tname -> do
290 withRandomTable jobPool $ \tname -> do
263 withNamedJobMonitor tname jobPool Prelude.id (actualTest tname)
291 withNamedJobMonitor tname jobPool Prelude.id (actualTest tname)
264
292
293+withNamedJobMonitor :: Job.TableName -> Pool Connection -> (Job.Config -> Job.Config) -> (IORef [Job.LogEvent] -> IO a) -> IO a
265−withNamedJobMonitor tname jobPool cfgFn actualTest = do
294+withNamedJobMonitor tname jobPool cfgFn actualTest =
295+ withConfig tname jobPool cfgFn $ \logRef cfg -> do
296+ withAsync (Job.startJobRunner cfg) (const $ actualTest logRef)
297+
298+withConfig :: Job.TableName -> Pool Connection -> (Job.Config -> Job.Config) -> (IORef [Job.LogEvent] -> Job.Config -> IO a) -> IO a
299+withConfig tname jobPool cfgFn configConsumer = do
266 logRef :: IORef [Job.LogEvent] <- newIORef []
300 logRef :: IORef [Job.LogEvent] <- newIORef []
267 tcache <- newTimeCache simpleTimeFormat'
301 tcache <- newTimeCache simpleTimeFormat'
268 withTimedFastLogger tcache LogNone $ \tlogger -> do
302 withTimedFastLogger tcache LogNone $ \tlogger -> do
@@ -270,7 +304,7 @@ withNamedJobMonitor tname jobPool cfgFn actualTest = do
270 tlogger $ \t -> toLogStr t <> " | " <> Job.defaultLogStr Job.defaultJobType logLevel logEvent
304 tlogger $ \t -> toLogStr t <> " | " <> Job.defaultLogStr Job.defaultJobType logLevel logEvent
271 atomicModifyIORef' logRef (\logs -> (logEvent:logs, ()))
305 atomicModifyIORef' logRef (\logs -> (logEvent:logs, ()))
272 cfg = Job.mkConfig flogger tname jobPool Job.UnlimitedConcurrentJobs jobRunner (\cfg' -> cfgFn $ cfg'{Job.cfgDefaultMaxAttempts=testMaxAttempts})
306 cfg = Job.mkConfig flogger tname jobPool Job.UnlimitedConcurrentJobs jobRunner (\cfg' -> cfgFn $ cfg'{Job.cfgDefaultMaxAttempts=testMaxAttempts})
273− withAsync (Job.startJobRunner cfg) (const $ actualTest logRef)
307+ configConsumer logRef cfg
274
308
275payloadGen :: MonadGen m => m JobPayload
309payloadGen :: MonadGen m => m JobPayload
276payloadGen = Gen.recursive Gen.choice nonRecursive recursive
310payloadGen = Gen.recursive Gen.choice nonRecursive recursive
@@ -300,43 +334,22 @@ testEnsureShutdown appPool jobPool = testCase "ensure shutdown" $ do
300 assertJobIdStatus conn tname logRef "Job is scheduled in future, should still be queueud" Job.Queued jobId
334 assertJobIdStatus conn tname logRef "Job is scheduled in future, should still be queueud" Job.Queued jobId
301 pure (jobId, logRef)
335 pure (jobId, logRef)
302
336
303−testGracefulShutdown appPool jobPool = testCase "ensure graceful shutdown" $ do
337+testGracefulShutdown :: Pool Connection -> Pool Connection -> TestTree
338+testGracefulShutdown appPool jobPool =
339+ testCase "ensure wait for all jobs is correct" $
304− withRandomTable jobPool $ \tname -> do
340+ withRandomTable jobPool $ \tname -> do
305− setupPreconditions 3 tname >>= \case
341+ waitTillJobStart <- newEmptyMVar
306− Nothing -> fail "Unable to setup preconditions for this test, even after trying thrice"
307− Just (j1, j2, logRef) -> do
308 Pool.withResource appPool $ \conn -> do
342 Pool.withResource appPool $ \conn -> do
343+ j1 <- Job.createJob conn tname (PayloadSucceed $ 4 * Job.defaultPollingInterval)
344+ time<- getCurrentTime
345+ j2 <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral $ unSeconds $ Job.defaultPollingInterval * 2) time)
346+ logRef <- withNamedJobMonitor tname jobPool (\cfg -> cfg { Job.cfgOnJobStart = \ _ -> do
347+ putMVar waitTillJobStart ()
348+ } ) $ \logRef -> do
349+ readMVar waitTillJobStart -- as soon as the job has started we quit the monitor:w
350+ pure logRef -- eg closing the withNamedJobMonitor closure shutsdwon the monitor
309 assertJobIdStatus conn tname logRef "Expecting the first job to be completed successfully if graceful shutdown is implemented correctly" Job.Success (jobId j1)
351 assertJobIdStatus conn tname logRef "Expecting the first job to be completed successfully if graceful shutdown is implemented correctly" Job.Success (jobId j1)
310 assertJobIdStatus conn tname logRef "Expecting the second job to be queued because no new job should be picked up during graceful shutdown" Job.Queued (jobId j2)
352 assertJobIdStatus conn tname logRef "Expecting the second job to be queued because no new job should be picked up during graceful shutdown" Job.Queued (jobId j2)
311− where
312− -- When this block completes execution, the graceful shutdown should have happened.
313− setupPreconditions i tname =
314− if i == (0 :: Integer)
315− then pure Nothing
316− else do
317− withNamedJobMonitor tname jobPool Prelude.id $ \logRef -> do
318− Pool.withResource appPool $ \conn -> do
319− t <- getCurrentTime
320− j1 <- Job.createJob conn tname (PayloadSucceed $ 4 * Job.defaultPollingInterval)
321− j2 <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral $ unSeconds $ Job.defaultPollingInterval * 2) t)
322− let waitForPrecondition = do
323− k1 <- Job.findJobByIdIO conn tname (jobId j1)
324− k2 <- Job.findJobByIdIO conn tname (jobId j2)
325− case (jobStatus <$> k1, jobStatus <$> k2) of
326− (Nothing, _) -> do
327− traceM "Not expecting first job to be completed so quickly. Trying again."
328− setupPreconditions (i-1) tname
329− (_, Nothing) -> do
330− traceM "Not expecting second job to be completed so quickly. Trying again."
331− setupPreconditions (i-1) tname
332− (Just Job.Locked, Just Job.Queued) -> pure $ Just (j1, j2, logRef) -- precondition met
333− (Just Job.Success, _) -> setupPreconditions (i-1) tname
334− (_, Just Job.Success) -> setupPreconditions (i-1) tname
335− (Just Job.Queued, Just Job.Queued) -> do
336− delaySeconds 1
337− waitForPrecondition
338− x -> fail $ "Unexpeted job statuses while setting up the test: " <> show x
339− waitForPrecondition
340
353
341testJobScheduling appPool jobPool = testCase "job scheduling" $ do
354testJobScheduling appPool jobPool = testCase "job scheduling" $ do
342 withNewJobMonitor jobPool $ \tname logRef -> do
355 withNewJobMonitor jobPool $ \tname logRef -> do
@@ -447,8 +460,8 @@ testRetryBackoff appPool jobPool = testCase "retry backoff" $ do
447 modifyRetryBackoff cfg
460 modifyRetryBackoff cfg
448 = cfg { Job.cfgDefaultRetryBackoff = pure . backoff }
461 = cfg { Job.cfgDefaultRetryBackoff = pure . backoff }
449
462
450−testResourceLimitedScheduling appPool jobPool = testGroup "concurrency control by resources"
463+testResourceLimitedScheduling _appPool jobPool = testGroup "concurrency control by resources"
451− [ testCase "resources control how jobs run" $ setup $ \tname resCfg logRef conn -> do
464+ [ testCase "resources control how jobs run" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
452 let firstResource = [(Job.ResourceId "first", 1)]
465 let firstResource = [(Job.ResourceId "first", 1)]
453 secondResource = [(Job.ResourceId "second", 1)]
466 secondResource = [(Job.ResourceId "second", 1)]
454
467
@@ -456,22 +469,25 @@ testResourceLimitedScheduling appPool jobPool = testGroup "concurrency control b
456 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
469 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
457 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) secondResource
470 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) secondResource
458
471
459− delaySeconds $ Job.defaultPollingInterval + Seconds 2
472+ _jone <- runSingleJobFromQueue cfg
473+ _jtwo <- runSingleJobFromQueue cfg
474+ _jthree <- runSingleJobFromQueue cfg
460
475
461 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
476 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
462 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
477 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
463 assertJobIdStatus conn tname logRef "Job uses different resources - it should be running" Job.Locked thirdJob
478 assertJobIdStatus conn tname logRef "Job uses different resources - it should be running" Job.Locked thirdJob
464
479
465− , testCase "resource-less jobs run normally" $ setup $ \tname resCfg logRef conn -> do
480+ , testCase "resource-less jobs run normally" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
466 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) []
481 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) []
467 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) []
482 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) []
468
483
469− delaySeconds $ Job.defaultPollingInterval + Seconds 2
484+ _jone <- runSingleJobFromQueue cfg
485+ _jtwo <- runSingleJobFromQueue cfg
470
486
471 assertJobIdStatus conn tname logRef "First job uses no resources - it should be running" Job.Locked firstJob
487 assertJobIdStatus conn tname logRef "First job uses no resources - it should be running" Job.Locked firstJob
472 assertJobIdStatus conn tname logRef "Second job uses no resources - it should be running" Job.Locked secondJob
488 assertJobIdStatus conn tname logRef "Second job uses no resources - it should be running" Job.Locked secondJob
473
489
474− , testCase "jobs can hold multiple resources" $ setup $ \tname resCfg logRef conn -> do
490+ , testCase "jobs can hold multiple resources" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
475 let firstResource = [(Job.ResourceId "first", 1)]
491 let firstResource = [(Job.ResourceId "first", 1)]
476 secondResource = [(Job.ResourceId "second", 1)]
492 secondResource = [(Job.ResourceId "second", 1)]
477
493
@@ -479,57 +495,70 @@ testResourceLimitedScheduling appPool jobPool = testGroup "concurrency control b
479 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
495 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
480 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) secondResource
496 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) secondResource
481
497
482− delaySeconds $ Job.defaultPollingInterval + Seconds 2
498+ jone <- runSingleJobFromQueue cfg
499+ jtwo <- runSingleJobFromQueue cfg
500+ jthree <- runSingleJobFromQueue cfg
501+ assertEqual "first job claimed the work" (void jone) (Just ())
502+ assertEqual "second thread was blocked" (void jtwo) Nothing
503+ assertEqual "third thread was blocked " (void jthree) Nothing
483
504
484 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
505 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
485 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
506 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
486 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued thirdJob
507 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued thirdJob
487
508
488− , testCase "resources are freed when jobs succeed" $ setup $ \tname resCfg logRef conn -> do
509+ , testCase "resources are freed when jobs succeed" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
489 let firstResource = [(Job.ResourceId "first", 1)]
510 let firstResource = [(Job.ResourceId "first", 1)]
490
511
491 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
512 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
492 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
513 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
493
514
494− delaySeconds $ Job.defaultPollingInterval + Seconds 2
515+ jone <- runSingleJobFromQueue cfg
516+ _jtwo <- runSingleJobFromQueue cfg
495
517
496 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
518 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
497 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
519 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
498
520
499− delaySeconds $ (Job.defaultPollingInterval * 2) + Seconds 8
521+ forM_ jone wait
522+
523+ _three <- runSingleJobFromQueue cfg
500
524
501 assertJobIdStatus conn tname logRef "First job should have succeeded by now" Job.Success firstJob
525 assertJobIdStatus conn tname logRef "First job should have succeeded by now" Job.Success firstJob
502 assertJobIdStatus conn tname logRef "Second job should be running after first job suceeded" Job.Locked secondJob
526 assertJobIdStatus conn tname logRef "Second job should be running after first job suceeded" Job.Locked secondJob
503
527
504− , testCase "resources are freed when jobs fail" $ setup $ \tname resCfg logRef conn -> do
528+ , testCase "resources are freed when jobs fail" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
505 let firstResource = [(Job.ResourceId "first", 1)]
529 let firstResource = [(Job.ResourceId "first", 1)]
506
530
507 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadAlwaysFail 10) firstResource
531 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadAlwaysFail 10) firstResource
508 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
532 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
509
533
510− delaySeconds $ Job.defaultPollingInterval + Seconds 2
534+ jone <- runSingleJobFromQueue cfg
535+ _jtwo <- runSingleJobFromQueue cfg
511
536
512 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
537 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
513 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
538 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
514
539
515− delaySeconds $ (Job.defaultPollingInterval * 2) + Seconds 8
540+ jobRes <- forM jone waitCatch
541+ assertEqual "should've thrown an exception" (Just (Left ())) (first (const ()) <$> jobRes)
542+
543+ _three <- runSingleJobFromQueue cfg
516
544
517 assertJobIdStatus conn tname logRef "First job should have failed by now" Job.Failed firstJob
545 assertJobIdStatus conn tname logRef "First job should have failed by now" Job.Failed firstJob
518 assertJobIdStatus conn tname logRef "Second job should be running after first job failed" Job.Locked secondJob
546 assertJobIdStatus conn tname logRef "Second job should be running after first job failed" Job.Locked secondJob
519
547
520− , testCase "resources allow for multiple usage" $ setup $ \tname resCfg logRef conn -> do
548+ , testCase "resources allow for multiple usage" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
521 let firstResource = [(Job.ResourceId "first", 1)]
549 let firstResource = [(Job.ResourceId "first", 1)]
522 resCfg' = resCfg { Job.resCfgDefaultLimit = 2 }
550 resCfg' = resCfg { Job.resCfgDefaultLimit = 2 }
523
551
524 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) firstResource
552 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) firstResource
525 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) firstResource
553 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) firstResource
526
554
527− delaySeconds $ Job.defaultPollingInterval + Seconds 2
555+ _jone <- runSingleJobFromQueue cfg
556+ _jtwo <- runSingleJobFromQueue cfg
528
557
529 assertJobIdStatus conn tname logRef "First job uses some resources - it should be running" Job.Locked firstJob
558 assertJobIdStatus conn tname logRef "First job uses some resources - it should be running" Job.Locked firstJob
530 assertJobIdStatus conn tname logRef "Second job uses some resources - it should be running" Job.Locked secondJob
559 assertJobIdStatus conn tname logRef "Second job uses some resources - it should be running" Job.Locked secondJob
531
560
532− , testCase "resource usage can differ by job" $ setup $ \tname resCfg logRef conn -> do
561+ , testCase "resource usage can differ by job" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
533 let firstResource n = [(Job.ResourceId "first", n)]
562 let firstResource n = [(Job.ResourceId "first", n)]
534 resCfg' = resCfg { Job.resCfgDefaultLimit = 5 }
563 resCfg' = resCfg { Job.resCfgDefaultLimit = 5 }
535
564
@@ -537,57 +566,65 @@ testResourceLimitedScheduling appPool jobPool = testGroup "concurrency control b
537 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 3)
566 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 3)
538 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 2)
567 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 2)
539
568
540− delaySeconds $ Job.defaultPollingInterval + Seconds 2
569+ jone <- runSingleJobFromQueue cfg
570+ _jtwo <- runSingleJobFromQueue cfg
571+ _jthree <- runSingleJobFromQueue cfg
541
572
573+ -- TODO, so it's not locked but already succeeded, which is werid, but the rest is valid?
542 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
574 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
543 assertJobIdStatus conn tname logRef "Job doesn't fit alongside first job - it shouldn't be running" Job.Queued secondJob
575 assertJobIdStatus conn tname logRef "Job doesn't fit alongside first job - it shouldn't be running" Job.Queued secondJob
544 assertJobIdStatus conn tname logRef "Job doesn't fit alongside first job - it shouldn't be running" Job.Queued thirdJob
576 assertJobIdStatus conn tname logRef "Job doesn't fit alongside first job - it shouldn't be running" Job.Queued thirdJob
545
577
546− delaySeconds $ (Job.defaultPollingInterval * 2) + Seconds 8
578+ forM_ jone wait
579+
580+ _jfour <- runSingleJobFromQueue cfg
581+ _jfive <- runSingleJobFromQueue cfg
547
582
548 assertJobIdStatus conn tname logRef "First job should have succeeded by now" Job.Success firstJob
583 assertJobIdStatus conn tname logRef "First job should have succeeded by now" Job.Success firstJob
549 assertJobIdStatus conn tname logRef "Second job should be running after first job suceeded" Job.Locked secondJob
584 assertJobIdStatus conn tname logRef "Second job should be running after first job suceeded" Job.Locked secondJob
550 assertJobIdStatus conn tname logRef "Third job fits alongside second job - it should be running after first job suceeded" Job.Locked thirdJob
585 assertJobIdStatus conn tname logRef "Third job fits alongside second job - it should be running after first job suceeded" Job.Locked thirdJob
551
586
552− , testCase "resource limits permanently block too-big jobs" $ setup $ \tname resCfg logRef conn -> do
587+ , testCase "resource limits permanently block too-big jobs" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
553 let firstResource = [(Job.ResourceId "first", 2)]
588 let firstResource = [(Job.ResourceId "first", 2)]
554
589
555 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
590 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
556
591
557− delaySeconds $ Job.defaultPollingInterval + Seconds 2
592+ void $ runSingleJobFromQueueBlock cfg
558
593
559 assertJobIdStatus conn tname logRef "Job requires too many resources - shouldn't be running" Job.Queued firstJob
594 assertJobIdStatus conn tname logRef "Job requires too many resources - shouldn't be running" Job.Queued firstJob
560
595
561− , testCase "resources with 0 limit block jobs" $ setup' 0 $ \tname resCfg logRef conn -> do
596+ , testCase "resources with 0 limit block jobs" $ setup' jobPool 0 $ \tname resCfg logRef conn cfg -> do
562 let firstResource = [(Job.ResourceId "first", 1)]
597 let firstResource = [(Job.ResourceId "first", 1)]
563
598
564 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
599 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
565
600
566− delaySeconds $ Job.defaultPollingInterval + Seconds 2
601+ void $ runSingleJobFromQueueBlock cfg
567
602
568 assertJobIdStatus conn tname logRef "Job can't access 0-limited resources - shouldn't be running" Job.Queued firstJob
603 assertJobIdStatus conn tname logRef "Job can't access 0-limited resources - shouldn't be running" Job.Queued firstJob
569
604
570− , testCase "resources with negative limit block jobs" $ setup' (-5) $ \tname resCfg logRef conn -> do
605+ , testCase "resources with negative limit block jobs" $ setup' jobPool (-5) $ \tname resCfg logRef conn cfg -> do
571 let firstResource = [(Job.ResourceId "first", 1)]
606 let firstResource = [(Job.ResourceId "first", 1)]
572
607
573 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
608 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
574
609
575− delaySeconds $ Job.defaultPollingInterval + Seconds 2
610+ void $ runSingleJobFromQueueBlock cfg
576
611
577 assertJobIdStatus conn tname logRef "Job can't access negative-limited resources - shouldn't be running" Job.Queued firstJob
612 assertJobIdStatus conn tname logRef "Job can't access negative-limited resources - shouldn't be running" Job.Queued firstJob
578 ]
613 ]
579− where
580− setup = setup' 1
581
614
582− setup' defaultLimit action =
615+setup jobPool = setup' jobPool 1
583− withRandomTable jobPool $ \tname ->
584− withRandomResourceTables defaultLimit jobPool tname $ \resCfg ->
585− withNamedJobMonitor tname jobPool (cfgFn resCfg) $ \logRef -> do
586− Pool.withResource appPool $ \conn -> action tname resCfg logRef conn
587
616
617+setup' jobPool defaultLimit action =
618+ withRandomTable jobPool $ \tname ->
619+ withRandomResourceTables defaultLimit jobPool tname $ \resCfg ->
620+ withConfig tname jobPool (cfgFn resCfg) $ \logRef cfg -> do
621+ Pool.withResource jobPool $ \conn -> action tname resCfg logRef conn cfg
622+
623+ where
588 cfgFn resCfg cfg = cfg
624 cfgFn resCfg cfg = cfg
589 { Job.cfgDefaultMaxAttempts = 1 -- Simplifies some tests where we have a failing job
625 { Job.cfgDefaultMaxAttempts = 1 -- Simplifies some tests where we have a failing job
590 , Job.cfgConcurrencyControl = Job.ResourceLimits resCfg
626 , Job.cfgConcurrencyControl = Job.ResourceLimits resCfg
627+ , Job.cfgDeleteSuccessfulJobs = False
591 }
628 }
592
629
593data JobEvent = JobStart
630data JobEvent = JobStart