Merge pull request #8 from remeike/create-scheduler-thread

Create separate scheduler thread

Author: remeike

src/System/Hworker.hs

@@ -64,6 +64,7 @@ module System.Hworker
   , createWith
   , destroy
   , worker
+  , scheduler
   , monitor
     -- * Queuing Jobs
   , queue
@@ -837,15 +838,12 @@ worker hw =
                 delayAndRun
 
 
--- | 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) and for pushing scheduled jobs to the queue at the expected time.
--- You need to have at least one of these running to have
--- the retry happen, but it is safe to have any number running.
+-- | Start a scheduler. Like 'worker', this is blocking, so should be
+-- started in a thread. This is responsible for pushing scheduled jobs
+-- to the queue at the expected time.
 
-monitor :: Job s t => Hworker s t -> IO ()
-monitor hw =
+scheduler :: Job s t => Hworker s t -> IO ()
+scheduler hw  =
   forever $ do
     now <- getCurrentTime
 
@@ -868,6 +866,20 @@ monitor hw =
           Left err ->
             liftIO $ hwlog hw err
 
+    threadDelay 500000 >> scheduler hw
+
+
+-- | 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
+
     runWithList hw (R.hkeys (progressQueue hw)) $ \js ->
       forM_ js $ \j ->
         runWithMaybe hw (R.hget (progressQueue hw) j) $

test/Spec.hs

@@ -475,7 +475,7 @@ main = hspec $ do
       mvar <- newMVar 0
       hworker <- createWith (conf "simpleworker-1" (SimpleState mvar))
       wthread <- forkIO (worker hworker)
-      mthread <- forkIO (monitor hworker)
+      sthread <- forkIO (scheduler hworker)
       time <- getCurrentTime
       queueScheduled hworker SimpleJob (addUTCTime 1 time)
       queueScheduled hworker SimpleJob (addUTCTime 2 time)
@@ -485,14 +485,14 @@ main = hspec $ do
       threadDelay 1000000 >> readMVar mvar >>= shouldBe 2
       threadDelay 1000000 >> readMVar mvar >>= shouldBe 3
       killThread wthread
-      killThread mthread
+      killThread sthread
       destroy hworker
 
     it "should execute a recurring job" $ do
       mvar <- newMVar 0
       hworker <- createWith (conf "recurringworker-1" (RecurringState mvar))
       wthread <- forkIO (worker hworker)
-      mthread <- forkIO (monitor hworker)
+      sthread <- forkIO (scheduler hworker)
       time <- getCurrentTime
       queueScheduled hworker RecurringJob (addUTCTime 2 time)
       threadDelay 3000000 >> readMVar mvar >>= shouldBe 1
@@ -501,7 +501,7 @@ main = hspec $ do
       threadDelay 2000000 >> readMVar mvar >>= shouldBe 4
       destroy hworker
       killThread wthread
-      killThread mthread
+      killThread sthread
 
   describe "Broken jobs" $
     it "should store broken jobs" $ do