diff --git a/Control/Concurrent/ParallelTasks.hs b/Control/Concurrent/ParallelTasks.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelTasks.hs @@ -0,0 +1,60 @@ +-- | The parallel functions in this module all use the same underlying behaviour. You supply a list +-- of tasks that you wish performed, either in the @IO@ monad or some other @MonadIO m => m@ monad. +-- This library starts up a limited number of threads (by default, one per capability, i.e. one per +-- available processor/core) and then executes the given work queue across the threads. This is better +-- than simply starting all the jobs in parallel and waiting, because in the case where you have +-- thousands or millions of jobs, but only say 16 cores, you do not want the overheads of switching +-- between all those contending threads. +-- +-- The default behaviour of these functions is to put useful progress reports onto stderr while +-- it is running (number of tasks completed, estimate of final completion time). +-- The library is aimed at millions of jobs taking several hours to complete; hence built-in output +-- is very useful for you, while you wait. You can customise this behaviour by using the primed +-- version of each of these functions and supplying a customised options record. +-- +-- The only difference between the functions @parallelList@, @parallelVec@ and @parallelIOVec@ is the type of the results returned. The +-- closest to the underlying behaviour is @parallelIOVec'@; the other functions are simply convenience wrappers that freeze/convert +-- the IOVector into a Vector or list. +-- +-- /Note/: make sure you compile your program with the @-threaded -with-rtsopts=-N@ options (e.g. in the ghc-options field in your +-- cabal file), or else you will not get any parallel execution in your program! +module Control.Concurrent.ParallelTasks ( + -- * The main parallel processing functions. + parallelList, parallelVec, parallelIOVec, + -- * The configurable versions of the functions. + -- | These versions can take place in a monad other than IO, and can configure other options (such as killing off long-running tasks). + -- See the documentation for 'ParTaskOpts'. + parallelList', parallelVec', parallelIOVec', + -- * The options available to configure the functions. + SimpleParTaskOpts(..), ParTaskOpts(..), defaultParTaskOpts) where + +import Control.Monad (liftM) +import Control.Monad.IO.Class (MonadIO, liftIO) +import qualified Data.Vector as V (Vector, toList, unsafeFreeze) +import qualified Data.Vector.Mutable as V (IOVector) + +import Control.Concurrent.ParallelTasks.Base (SimpleParTaskOpts(..), ParTaskOpts(..), defaultExtendedParTaskOpts, defaultParTaskOpts, parallelTasks) + +-- | As 'parallelList', but returns the results in a mutable IOVector. +-- +-- Defined as @parallelIOVec' defaultParTaskOpts@ +parallelIOVec :: [IO a] -> IO (V.IOVector a) +parallelIOVec = parallelIOVec' defaultParTaskOpts +parallelIOVec' :: MonadIO m => ParTaskOpts m a -> [m a] -> m (V.IOVector a) +parallelIOVec' o = parallelTasks (defaultExtendedParTaskOpts o) + +-- | As 'parallelList', but returns the results in an immutable Vector. +-- +-- Defined as @parallelVec' defaultParTaskOpts@ +parallelVec :: [IO a] -> IO (V.Vector a) +parallelVec = parallelVec' defaultParTaskOpts +parallelVec' :: MonadIO m => ParTaskOpts m a -> [m a] -> m (V.Vector a) +parallelVec' o t = parallelIOVec' o t >>= liftIO . V.unsafeFreeze + +-- | Runs the list of tasks in parallel (a few at a time), and returns the results in a list (with the corresponding order to the input list, i.e. the first task produces the first result in the list.) See the module description for more details. +-- +-- Defined as: @parallelList' defaultParTaskOpts@ +parallelList :: [IO a] -> IO [a] +parallelList = parallelList' defaultParTaskOpts +parallelList' :: MonadIO m => ParTaskOpts m a -> [m a] -> m [a] +parallelList' o t = V.toList `liftM` parallelVec' o t diff --git a/Control/Concurrent/ParallelTasks/Base.hs b/Control/Concurrent/ParallelTasks/Base.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelTasks/Base.hs @@ -0,0 +1,238 @@ +{-# LANGUAGE DeriveDataTypeable, QuasiQuotes, Rank2Types, ScopedTypeVariables #-} +-- FlexibleInstances, TypeSynonymInstances #-} + +-- | Module with the internal workhorse for the library, 'parallelTasks'. You only +-- need to use this module if you want to alter 'ExtendedParTaskOpts', which allows +-- you to redirect the logging output or store information about task timing. +module Control.Concurrent.ParallelTasks.Base (ExtendedParTaskOpts(..), ParTaskOpts(..), SimpleParTaskOpts(..), TaskOutcome(Success, TookTooLong), defaultExtendedParTaskOpts, defaultParTaskOpts, parallelTasks) where + +import Control.Applicative ((<$), (<$>)) +import Control.Concurrent (forkIO, getNumCapabilities, killThread, myThreadId, threadDelay) +import Control.Concurrent.STM (TVar, atomically, newTVarIO, readTVar, retry, writeTVar) +import Control.Concurrent.STM.TMVar (newTMVar, putTMVar, takeTMVar) +import Control.Exception.Base (Exception, bracket_, evaluate, handle, onException, throwTo) +import Control.Monad (replicateM_, when) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Maybe (isJust) +import Data.String.Here.Interpolated (i) +import Data.Time.Clock (UTCTime, addUTCTime, diffUTCTime, getCurrentTime) +import Data.Time.Format (formatTime) +import Data.Typeable +import qualified Data.Vector.Mutable as V (IOVector, new, write) +import System.IO (Handle, hFlush, hPutStrLn, stderr) +import System.Locale (defaultTimeLocale) + +data SimpleParTaskOpts = SimpleParTaskOpts { + -- | Number of worker threads to use. When this is Nothing, defaults to number of capabilities (see @numCapabilities@) + numberWorkers :: Maybe Int, + -- | How often to print the progress of the tasks. E.g. when Just 100, print a message roughly + -- after the completion of every 100 tasks. + printProgress :: Maybe Int, + -- | How often to print an estimate of the estimated completion time. E.g. when Just 100, + -- print an estimate after the completion of every 100 tasks. + printEstimate :: Maybe Int + } + +-- | Options controlling the general running of parallel tasks. The @m@ parameter is the monad (which must be an instance +-- of 'MonadIO') in which the tasks will be run, and the @a@ parameter is the return value of the tasks. +data ParTaskOpts m a = ParTaskOpts { + -- | The simple options. + simpleOpts :: SimpleParTaskOpts, + -- | Function to use to run the @m@ monad on top of IO. The returned function is run at least once per worker, so should support + -- being run multiple times in parallel, and should clean up after itself. Suitable instance for IO is simply @return id@. + wrapWorker :: forall r. m (m r -> IO r), + -- | When Just, the number of microseconds to let each task run for, before assuming it will + -- not complete, and killing it off. In the case that the task is killed off, the second + -- part of the pair is the value that will be stored in the vector. + timeLimit :: Maybe (Integer, a) + } + +-- | Advanced options controlling the behaviour of parallel tasks. The @m@ parameter +-- is the monad that the tasks execute in, the @a@ parameter is the output value of the +-- tasks, and the @b@ parameter is the type that is stored in the results array. It is +-- common that either @b = a@ or @b = Maybe a@. +data ExtendedParTaskOpts m a = ExtendedParTaskOpts { + -- | Core options + coreOpts :: ParTaskOpts m a, + -- | Function that supplies a handle to an inner block to write messages to. + -- To use stdout or stderr, you can just supply @($ stdout)@. To write to a file, + -- use @withFile \"blah\" WriteMode@. + printTo :: forall r. (Handle -> IO r) -> IO r, + -- | Function used to store the outcome of the task. Arguments are (in order): + -- + -- * Time that the task took to complete (in seconds) + -- + -- * Index at which to store the result (same as index of the task in the original tasks list) + -- + -- * The outcome of the task + -- + -- If a String is returned, it is logged + afterFinish :: Double -> Int -> TaskOutcome -> IO (Maybe String) + } + +-- | Value indicating whether a task successfully completed, or was killed off for taking too long +data TaskOutcome = Success | TookTooLong + +data TookTooLongException = TookTooLongException deriving (Show, Typeable) +instance Exception TookTooLongException + +-- | A version of threadDelay that accommodates delays longer than @maxBound :: Int@. +threadDelay' :: Integer -> IO () +threadDelay' target + | target > mx = threadDelay maxBound >> threadDelay' (target - mx) + | otherwise = threadDelay (fromInteger target) + where + mx :: Integer + mx = toInteger (maxBound :: Int) + +-- | Default extended options. Prints messages to stderr, and writes a message when a +-- task is killed +defaultExtendedParTaskOpts :: MonadIO m => ParTaskOpts m a -> ExtendedParTaskOpts m a +defaultExtendedParTaskOpts opts = ExtendedParTaskOpts { coreOpts = opts, printTo = ($ stderr), afterFinish = printKill } + where + printKill _ n TookTooLong = return $ Just [i|*** Killed task ${n} for taking too long|] + printKill _ _ _ = return Nothing + +-- | Default parallel task options. The number of workers defaults to the number of capabilities, +-- with no time limit, and printing progress every 50 tasks and an estimated time every 200 +defaultParTaskOpts :: ParTaskOpts IO a +defaultParTaskOpts = ParTaskOpts { simpleOpts = SimpleParTaskOpts {numberWorkers = Nothing, printProgress = Just 50, printEstimate = Just 200 }, wrapWorker = return id, timeLimit = Nothing } + +-- | Runs the given set of computations in parallel, and once they are all finished, returns their results. +-- Note that they won't all be run in parallel from the start; rather, a set of +-- workers will be spawned that work their way through the (potentially large) set of jobs. +parallelTasks :: MonadIO m => ExtendedParTaskOpts m a -> [m a] -> m (V.IOVector a) +parallelTasks _ [] = liftIO $ V.new 0 +parallelTasks opts tasks = wrapWorker (coreOpts opts) >>= \run -> liftIO $ printTo opts $ \h -> do + let numTasks = length tasks + numWorkers <- maybe getNumCapabilities return (numberWorkers $ simpleOpts $ coreOpts opts) + vValue <- V.new numTasks + tvWork <- newTVarIO (numTasks, zip [0..] tasks) + tvDone <- newTVarIO numWorkers + startTime <- getCurrentTime + let printStartEnd = isJust $ printProgress $ simpleOpts $ coreOpts opts + when printStartEnd $ + hPrintTime h [i|Total tasks: ${numTasks}, starting at: |] + hMutex <- atomically $ newTMVar () + let safeLog s = bracket_ (atomically $ takeTMVar hMutex) (atomically $ putTMVar hMutex ()) (hPutStrLnFlush h s) + replicateM_ numWorkers $ forkIO $ + worker (timeLimit $ coreOpts opts) run + (\t n x o -> V.write vValue n x >> afterFinish opts t n o >>= maybe (return ()) safeLog) + tvWork tvDone + waitForWorkers (simpleOpts $ coreOpts opts) safeLog startTime numTasks tvWork tvDone + when printStartEnd $ + hPrintTime h "Finished at: " + return vValue + +hPrintTime :: Handle -> String -> IO () +hPrintTime h msg = getCurrentTime >>= hPutStrLnFlush h . (msg ++) . show + +-- | Primarily waits for the last TVar to hit zero, but in the mean time prints details +-- of the tasks remaining and the estimated completion time +waitForWorkers :: SimpleParTaskOpts -> (String -> IO ()) -> UTCTime -> Int -> TVar (Int, _x) -> TVar Int -> IO () +waitForWorkers opts safeLog startTime totalTasks tvWork tvDone = go True totalTasks totalTasks + where + go False _ _ = return () + go True lastProgress lastETA = do + (workersRemaining, tasksRemaining, timeToPrintProgress, timeToPrintETA) <- atomically $ do + tasksRemaining <- fst <$> readTVar tvWork + workersRemaining <- readTVar tvDone + let timeToPrintProgress = case printProgress opts of + Nothing -> False + Just n -> tasksRemaining <= lastProgress - n + timeToPrintETA = case printEstimate opts of + Nothing -> False + Just n -> tasksRemaining <= lastETA - n + when (workersRemaining > 0 && not timeToPrintETA && not timeToPrintProgress) retry + return (workersRemaining, tasksRemaining, timeToPrintProgress, timeToPrintETA) + curTime <- getCurrentTime + safeLog $ concat + [if timeToPrintProgress then [i|Tasks remaining: ${tasksRemaining} |] else "" + ,if timeToPrintETA then [i|ETA: ${eta startTime totalTasks curTime tasksRemaining}|] else ""] + go (workersRemaining > 0) + (if timeToPrintProgress then tasksRemaining else lastProgress) + (if timeToPrintETA then tasksRemaining else lastETA) + +eta :: UTCTime -> Int -> UTCTime -> Int -> String +eta startTime totalTasks curTime tasksRemaining + = [i|${timeLeft} seconds (${showTime timeFinish})|] + where + timeSoFar = curTime `diffUTCTime` startTime + timePerTask = timeSoFar / fromIntegral (totalTasks - tasksRemaining) + timeLeft = timePerTask * fromIntegral tasksRemaining + timeFinish = timeLeft `addUTCTime` curTime + + showTime :: UTCTime -> String + showTime = formatTime defaultTimeLocale "%T %F" + + +-- | Repeatedly picks next job from queue and executes it. +worker :: forall m a. MonadIO m => + -- Limit in microseconds for the computations: + Maybe (Integer, a) -> + -- Function for running the monad: + (m () -> IO ()) -> + -- Function for storing the result of the computation: + (Double -> Int -> a -> TaskOutcome -> IO ()) -> + -- Variable to grab the next work item from: + TVar (Int, [(Int, m a)]) -> + -- Variable to decrement when the worker finishes: + TVar Int -> + IO () +worker mlimit run store tvWork tvDone = case mlimit of + Just _limit -> handle (\TookTooLongException -> return ()) + (run $ go True) `onException` finish + -- We do not call finish when we receive a TookTooLongException, + -- because another worker will have replaced us + Nothing -> (run $ go True) `onException` finish + where + finish = atomically $ readTVar tvDone >>= writeTVar tvDone . pred + + go :: Bool -> m () + go False = liftIO finish + go True = do nextWork <- liftIO $ atomically $ do + (count, work) <- readTVar tvWork + case work of + [] -> return Nothing + (w:ws) -> Just w <$ writeTVar tvWork (pred count, ws) + keepGoing <- case (nextWork, mlimit) of + (Nothing, _) -> return False + (Just (n, work), Nothing) -> do + (x, t) <- withTime work + liftIO $ store t n x Success + return True + (Just (n, work), Just (limit, def)) -> do + myId <- liftIO myThreadId + start <- liftIO getCurrentTime + -- Fork a watchdog to watch for if we have been executing too long: + theirId <- liftIO $ forkIO $ do + threadDelay' limit + end <- getCurrentTime + store (end `timeDiff` start) n def TookTooLong + -- We fork a new worker, because we cannot guarantee we will ever kill + -- the old one (e.g. if it is blocked, or not allocating memory) + -- TODO there is a slim chance that the worker can start and then we are killed + -- first: + _ <- forkIO $ worker mlimit run store tvWork tvDone + -- This call may block (effectively) forever, waiting for the other thread to die: + throwTo myId TookTooLongException + (x, t) <- withTime work + liftIO $ killThread theirId -- Kill the watchdog + liftIO $ store t n x Success + return True + go keepGoing + +-- | A version of putStrLn that flushes after writing (and works in any MonadIO monad) +hPutStrLnFlush :: MonadIO m => Handle -> String -> m () +hPutStrLnFlush h s = liftIO $ hPutStrLn h s >> hFlush h + +withTime :: MonadIO m => m a -> m (a, Double) +withTime m = do + start <- liftIO getCurrentTime + x <- m + end <- liftIO getCurrentTime + t <- liftIO $ evaluate $ end `timeDiff` start + return (x, t) + +timeDiff :: UTCTime -> UTCTime -> Double +timeDiff end start = fromRational $ toRational (end `diffUTCTime` start) diff --git a/Control/Concurrent/ParallelTasks/Cache.hs b/Control/Concurrent/ParallelTasks/Cache.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelTasks/Cache.hs @@ -0,0 +1,249 @@ +{-# LANGUAGE DeriveFunctor, QuasiQuotes, Rank2Types, ScopedTypeVariables #-} + +-- | A module with a function to support caching the output of your parallel tasks. +module Control.Concurrent.ParallelTasks.Cache (parMapCache) where + +import Control.Applicative ((<$>), (<*>)) +import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM.TMVar (newTMVar, putTMVar, takeTMVar) +import Control.DeepSeq (NFData, force) +import Control.Exception as E(catch, evaluate, IOException) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.ST (ST, runST) +import qualified Data.ByteString as BS +import Data.Int (Int64) +import Data.Serialize +import Data.String.Here.Interpolated (i) +import Data.Time.Clock (getCurrentTime) +import qualified Data.Vector.Unboxed as U +import qualified Data.Vector.Unboxed.Mutable as MU +import qualified Data.Vector.Mutable as MV +import qualified Data.Vector as V +import Data.Vector.Algorithms.Intro as MU +import System.IO (Handle, IOMode(..), SeekMode(..), hClose, hSeek, hTell, openFile, withFile) +import System.IO (hFlush, hPutStrLn) + +import Control.Concurrent.ParallelTasks.Base (ExtendedParTaskOpts(..), ParTaskOpts(..), TaskOutcome(..), parallelTasks) + +type Location = (Int64, Int64) -- start, length + +type CacheStem = String + +data CacheOutcome key = CacheHit | CacheMissSuccess | CacheMissTookTooLong key + +isCacheHit :: CacheOutcome a -> Bool +isCacheHit CacheHit = True +isCacheHit _ = False + +isCacheMissSuccess :: CacheOutcome a -> Bool +isCacheMissSuccess CacheMissSuccess = True +isCacheMissSuccess _ = False + +-- Cache index file structure: +-- Int64 (number of keys) +-- Then, that many lots of: +-- key, Int64 (start, relative to payload), Int64 (end) +-- Cache payload file structure: +-- Payload + +indexFile :: CacheStem -> FilePath +indexFile = (++ "-index") + +payloadFile :: CacheStem -> FilePath +payloadFile = (++ "-payload") + +readKeysFromCache :: (U.Unbox key, Serialize key) => CacheStem -> IO (U.Vector (key, Location)) +readKeysFromCache cacheStem = (readKeysFromCache' <$> BS.readFile (indexFile cacheStem)) `E.catch` (\(_e :: IOException) -> return $ U.fromList []) + +readKeysFromCache' :: (U.Unbox key, Serialize key) => BS.ByteString -> U.Vector (key, Location) +readKeysFromCache' origFull = + let (count, table) = BS.splitAt 8 origFull + keysAmount :: Int64 + keysAmount = either error id $ runGet get count + in either (const U.empty) id $ runGet (U.replicateM (fromIntegral keysAmount) getKeyLocation) table + where + getKeyLocation = (,) <$> get <*> ((,) <$> get <*> get) + +-- Makes keys available during run, and on exit, adds new values to cache file +withCache :: forall key value a. (Ord key, MU.Unbox key, Serialize key, NFData value, Serialize value) => + CacheStem -> Handle -> Int -> (U.Vector (key, Location) -> (Location -> IO value) -> (key -> value -> IO ()) -> IO a) -> IO a +withCache cacheStem logHandle maxNewKeys inner = withFile (payloadFile cacheStem) ReadWriteMode $ \payloadHandle -> do + -- Get existing keys: + prevKeys <- readKeysFromCache cacheStem + -- Allocate space for as many new keys as we might need (length of tasks array) + newKeys <- MU.new maxNewKeys + (newKeysVar, mutex) <- atomically $ (,) <$> newTMVar 0 <*> newTMVar () + let readValue :: Location -> IO value + readValue (start, len) = do + -- With the mutex, seek and read from the cache payload file: + atomically $ takeTMVar mutex + hSeek payloadHandle AbsoluteSeek (toInteger start) + val <- BS.hGet payloadHandle (fromIntegral len) + atomically $ putTMVar mutex () + -- Force evaluation to make sure the conversion is done now, not ages down the line: + evaluate $ either error force $ runGet get val + writeValue :: key -> value -> IO () + writeValue k v = do + -- With the mutex, seek and write to the cache payload file: + newPayload <- evaluate $ runPut (put v) + n <- atomically $ takeTMVar mutex >> takeTMVar newKeysVar + hSeek payloadHandle SeekFromEnd 0 + start <- hTell payloadHandle + BS.hPut payloadHandle newPayload + MU.write newKeys n (k, (fromInteger start, fromIntegral $ BS.length newPayload)) + atomically $ putTMVar mutex () >> putTMVar newKeysVar (succ n) + result <- inner prevKeys readValue writeValue + -- At the end, we get all the keys together, sort them and write them all out to the index file: + printTime logHandle "Combining keys " + numNewKeys <- atomically $ takeTMVar newKeysVar + let endPrevKeys = U.length prevKeys + joinedKeys <- flip MU.unsafeGrow numNewKeys =<< U.unsafeThaw prevKeys + mapM_ (\n -> MU.read newKeys n >>= MU.write joinedKeys (n + endPrevKeys)) [0 .. numNewKeys - 1] + printTime logHandle "Sorting keys " + MU.sort joinedKeys + frozenJoinedKeys <- U.unsafeFreeze joinedKeys + printTime logHandle "Writing index " + withFile (indexFile cacheStem) WriteMode $ \indexHandle -> do + BS.hPut indexHandle $ runPut $ put (fromIntegral (U.length frozenJoinedKeys) :: Int64) + U.mapM_ (BS.hPut indexHandle . runPut . (\(k, l) -> put k >> put (fst l) >> put (snd l))) frozenJoinedKeys + return result + +printTime :: Handle -> String -> IO () +printTime h msg = getCurrentTime >>= hPutStrLnFlush h . (msg ++) . show + + +binarySearch :: (Ord key, MU.Unbox key, MU.Unbox v) => key -> U.Vector (key, v) -> Maybe v +binarySearch tgt v = go 0 (U.length v - 1) + where + go imin imax + | imax < imin = Nothing + | otherwise = let imid = (imin + imax) `div` 2 -- Could overflow on *very* large caches + (k, x) = v U.! imid + in case compare k tgt of + GT -> go imin (imid - 1) + LT -> go (imid + 1) imax + EQ -> Just x + +-- | A function that performs caching (between runs of the same tasks) to help when running the same analysis task +-- many times. +-- +-- Imagine that you have a program where you want to some map-reduce work. The mapping takes a long time, but you +-- are working on the reduce part. You don't want to have to redo the mapping every time you run your program; +-- you can use this cache functionality to save the results of the mapping between program runs. Alternatively, you +-- may want to analyse only part of your data at first (for speed) then slowly expand to the rest of the data set. +-- Caching allows you to re-use the results you have already calculated. +-- +-- There are three main concepts in the type signature. @input@ is a type containing all the information needed +-- to perform the task and produce the output. This may involve file handles or functions or whatever. The @key@ +-- type is generally smaller, and is the smallest possible unique identifier for a corresponding output. This might +-- be the primary key of a database record, or an input filename. (Obviously, in some cases, @input = key@; that +-- makes life easy). The @output@ type is the output of the task. +-- +-- In order to serialise the cache to a file, both @key@ and @output@ have to be instances of @Serialize@. To allow +-- efficient unboxing of a vector, we require an @Unbox@ instance for @key@ (contact me if you think this is too onerous), +-- and to ensure strict reading from the cache we require @NFData@ for output. +-- +-- Remember that @parMapCache@ doesn't know when your cache is invalid (e.g. because you've altered the processing algorithm +-- that you are passing to this function), and will blindly use it if it finds it. It's your responsibility to remove +-- the cache when it becomes invalid. +parMapCache :: forall input output key m. (MonadIO m, Ord key, Show key, MU.Unbox key, NFData output, Serialize key, Serialize output) => + ParTaskOpts m output + -- ^ The parallel task options for running these tasks in parallel + -> FilePath + -- ^ The directory in which to store the cache files (\"cache-index\" and \"cache-payload\") + -- and the log file (\"parmap-log\"). If you have multiple distinct parMapCache tasks + -- and you don't want them overlapping, pass a different directory for each. + -- (This is definitely a good idea, because if your two functions have an identical + -- serialised @key@ value, you'll be in all sorts of trouble!) + -> (input -> key) + -- ^ The function to map inputs to keys + -> (input -> m output) + -- ^ The actual function to calculate an output from an input. Note that despite + -- the NFData instance on output, we do not force the evaluation of output; + -- that is left to you to do inside this function. + -> [input] + -- ^ The list of inputs to process + -> m (MV.IOVector output) + -- ^ The vector of outputs. +parMapCache opts dir getKey process inputs + -- vOutcome is for statistics, holds CacheOutcome values + = do vOutcome <- liftIO $ MV.new (length inputs) + logFile <- liftIO $ openFile (dir ++ "/parmap-log") WriteMode + let fullOpts = (ExtendedParTaskOpts opts + ($ logFile) + -- When we write to the results array, we also write to our outcomes array: + (\t n outcome -> case outcome of + Success -> do (_, x) <- MV.read vOutcome n + MV.write vOutcome n (t, x) + return Nothing + TookTooLong -> do let key = getKey $ inputs !! n + MV.write vOutcome n (t, CacheMissTookTooLong key) + return $ Just [i|*** Killed task with key ${show key} for taking too long|] + )) + run <- wrapWorker opts + results <- liftIO $ withCache (dir ++ "/cache") logFile (length inputs) $ + \cachedKeys readValue saveResult -> run $ + parallelTasks fullOpts (zipWith (processWithCache vOutcome cachedKeys readValue saveResult) [0..] inputs) + -- Print all the statistics at the end: + liftIO $ do + outcomes <- V.unsafeFreeze vOutcome + let hits = fstFilter isCacheHit outcomes + missSuccesses = fstFilter isCacheMissSuccess outcomes + hPutStrLn logFile [i|Complete; hits: ${V.length hits}, misses: ${V.length missSuccesses}, timed out: ${V.length outcomes - V.length hits - V.length missSuccesses}|] + hPutStrLn logFile [i|Average cache hit time: ${average hits}|] + hPutStrLn logFile [i|Average successful task (cache miss) time: ${average missSuccesses}|] + hPutStrLn logFile [i|Median successful task (cache miss) time: ${median missSuccesses}|] + hPutStrLn logFile [i|Longest successful task (cache miss) time: ${maximumV missSuccesses}|] + hPutStrLn logFile "Details of killed tasks:" + sequence_ [hPutStrLn logFile [i| Killed task, key: ${show k}|] | (_, CacheMissTookTooLong k) <- V.toList outcomes] + hClose logFile + return results + where + -- Looks for a given key in the cache. If it finds it, reads the associated value and returns it. + -- If it doesn't find it, calculates it and marks it for addition to the cache. + processWithCache :: MV.IOVector (Double, CacheOutcome key) -> U.Vector (key, Location) -> ((Int64, Int64) -> IO output) -> (key -> output -> IO ()) -> Int -> input -> m output + processWithCache vOutcome cachedKeys readValue saveResult n x = case binarySearch theKey cachedKeys of + Just resultLoc -> liftIO $ do MV.write vOutcome n (0, CacheHit) + readValue resultLoc + Nothing -> do result <- process x + liftIO $ MV.write vOutcome n (0, CacheMissSuccess) + liftIO $ saveResult theKey result + return result + where + theKey = getKey x + +average :: V.Vector Double -> Double +average xs = V.foldr (+) 0 xs / fromIntegral (V.length xs) + +maximumV :: V.Vector Double -> Double +maximumV = V.foldr max 0 + +median :: V.Vector Double -> Double +median = median' . (\v -> runST (stSort v)) + where + stSort :: V.Vector Double -> (forall s. ST s (V.Vector Double)) + stSort orig = do + copy <- V.thaw orig + MU.sort copy + V.unsafeFreeze copy + + median' :: V.Vector Double -> Double + median' v + | V.null v = 1 / 0 -- NaN + | V.length v `mod` 2 == 1 = v V.! (V.length v `div` 2) + | otherwise = ((v V.! (V.length v `div` 2)) + (v V.! ((V.length v `div` 2) + 1))) / 2 + +fstFilter :: (b -> Bool) -> V.Vector (a, b) -> V.Vector a +fstFilter f v = V.unfoldr build 0 + where + build n + | n >= V.length v = Nothing + | f (snd x) = Just (fst x, succ n) + | otherwise = build (succ n) + where x = v V.! n + +-- | A version of putStrLn that flushes after writing (and works in any MonadIO monad) +hPutStrLnFlush :: MonadIO m => Handle -> String -> m () +hPutStrLnFlush h s = liftIO $ hPutStrLn h s >> hFlush h + + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,31 @@ +parallel-tasks library +Copyright (c) 2013, Neil Brown + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Neil Brown nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/parallel-tasks.cabal b/parallel-tasks.cabal new file mode 100644 --- /dev/null +++ b/parallel-tasks.cabal @@ -0,0 +1,39 @@ +name: parallel-tasks +version: 4.0.0.0 +-- synopsis: +description: This library is useful for running a large amount of parallel tasks + that run on top of the IO monad, executing them in batches from a work queue. + . + It has several features aimed at monitoring the progress of the tasks + and tries to be reasonably efficient (in space and time) for large + numbers (millions) of tasks. There is also caching support available so that the results of + running the task can be preserved between runs of the same program, which + is useful for doing scientific analysis. +license: BSD3 +license-file: LICENSE +author: Neil Brown +maintainer: nccb@kent.ac.uk +-- copyright: +-- category: +build-type: Simple +cabal-version: >=1.8 + +Library + Exposed-modules: Control.Concurrent.ParallelTasks, + Control.Concurrent.ParallelTasks.Base, + Control.Concurrent.ParallelTasks.Cache + ghc-options: -Wall -fwarn-tabs -fwarn-wrong-do-bind + -- -ddump-simpl -dsuppress-all + build-depends: base == 4.*, + bytestring >= 0.9 && < 0.11, + cereal >= 0.3 && < 0.5, + deepseq == 1.3.*, + here == 1.2.*, +-- lifted-base == 0.2.*, +-- monad-control == 0.3.*, + old-locale == 1.0.*, + stm == 2.4.*, + time == 1.4.*, + transformers == 0.3.*, + vector >= 0.7 && < 0.11, + vector-algorithms == 0.5.*