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