Parallel scheduler for running actions at designated times

Question

I've been trying to learn Haskell on and off for a couple years. I had a project for building a parallel web scraper, and I figured I'd try to use Haskell for it. Here's the part used to schedule tasks: basically, it opens a Chan that can be used to send jobs, which is simply an IO action with the time when it should be run. I'd be thankful for any (constructive) remarks.

Here's the main Scheduler module:

{-|
Module      : Scheduler
Description : Run IO actions at a designated time, in parallel
-}
module Scheduler (startScheduler, InputJobChan, Job(..)) where

import System.Timeout (timeout)
import Control.Concurrent.Chan (Chan, newChan, readChan)
import Control.Concurrent (forkIO)
import Data.Time.Clock.POSIX (getPOSIXTime, POSIXTime)
import qualified Data.PQueue.Min as PQ

-- |Main job type used to describe jobs that are sent to the scheduler.
data Job = Job { action :: InputJobChan -> IO(), time :: POSIXTime }
instance Eq Job where
  (==) j1 j2 = time j1 == time j2
instance Ord Job where
  (<=) j1 j2 = time j1 <= time j2
-- |Queue type used to send jobs to the scheduler
type InputJobChan = Chan [Job]
-- |Internal priority queue used to find the next job to run
type JobPQueue = PQ.MinQueue Job

-- |Start the scheduler and return an input queue used to send jobs
startScheduler :: IO InputJobChan
startScheduler = do
  chan <- newChan
  _ <- forkIO $ processJobs_ chan PQ.empty
  return chan

-- |Internal function doing most of the work
processJobs_ :: InputJobChan -> JobPQueue -> IO()
processJobs_ chan jobs
  -- If we don't have any jobs in queue, wait until we get some
  | PQ.null jobs = do
    returnedJobs <- readChan chan
    print "Got new jobs!"
    processJobs_ chan (PQ.fromList returnedJobs)
  -- otherwise wait until it's time to run the first job
  | otherwise = do
      let (job, others) = PQ.deleteFindMin jobs
      print $ "current: " ++ show (time job) ++ " (" ++ show (PQ.size others) ++ " others in queue)"
      now <- getPOSIXTime
      -- delay in microseconds
      let microsecondTimeout = round $ 1000000 * (time job - now)
      print $ "preparing to wait " ++ show microsecondTimeout ++ " for new messages on the queue..."
      readResult <- timeoutReadChan chan microsecondTimeout
      case readResult of
        Nothing -> do
          print "no new jobs, running current job action..."
          action job chan
          processJobs_ chan others
        Just returnedJobs -> do
          print "Enqueuing new jobs ..."
          let newJobs = PQ.union jobs (PQ.fromList returnedJobs)
          processJobs_ chan newJobs

-- |Listen to a chan at most t microseconds, return Nothing if nothing was read
timeoutReadChan :: Chan a -> Int -> IO (Maybe a)
timeoutReadChan chan t
  | t <= 0    = return Nothing
  | otherwise = timeout t (readChan chan)

And here is some simple test code:

import Scheduler
import System.Random.MWC (createSystemRandom)
import System.Random.MWC.Distributions (exponential)
import Control.Monad
import Data.Time.Clock.POSIX
import Control.Concurrent.Chan (writeChan)
import Control.Concurrent (threadDelay)

-- |Create jobs at random, simulating a poisson process
createRandomJobs :: InputJobChan -> IO ()
createRandomJobs chan = do
  gen <- createSystemRandom
  forM_ [1..15::Int] $ \n -> do
    delay <- exponential 1 gen
    now <- getPOSIXTime
    let job = Job { action = \_ -> putStrLn $ "hello " ++ show n, time = now + 1 }
    writeChan chan [job]
    threadDelay $ round $ 1000000 * delay

main :: IO ()
main = do
  chan <- startScheduler
  createRandomJobs chan

Answer 1

processJobs_ logic

You want to make sure that microsecondTimeout is never negative, since according to the documentation a negative timeout means to wait indefinitely.

To guard against negative timeouts, I would write processJobs_ to have the following logic:

loop:
  if queue is empty:
    call readChan for a new job and add to queue
    goto loop
  else:
    now <- getPosixTime
    if queue minimum <= now:
      fire off all jobs in the queue with a time <= now
      goto loop
    else:
      readResult <- timeoutReadChan ...
      put new job onto the queue
      goto loop

priority queues

You might be interested in this SO question and answers discussing the merits of the various Haskell priority queue packages:

https://stackoverflow.com/questions/6976559/comparison-of-priority-queue-implementations-in-haskell