parallel-tasks (empty) → 4.0.0.0
raw patch · 6 files changed
+619/−0 lines, 6 filesdep +basedep +bytestringdep +cerealsetup-changed
Dependencies added: base, bytestring, cereal, deepseq, here, old-locale, stm, time, transformers, vector, vector-algorithms
Files
- Control/Concurrent/ParallelTasks.hs +60/−0
- Control/Concurrent/ParallelTasks/Base.hs +238/−0
- Control/Concurrent/ParallelTasks/Cache.hs +249/−0
- LICENSE +31/−0
- Setup.hs +2/−0
- parallel-tasks.cabal +39/−0
+ Control/Concurrent/ParallelTasks.hs view
@@ -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
+ Control/Concurrent/ParallelTasks/Base.hs view
@@ -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)
+ Control/Concurrent/ParallelTasks/Cache.hs view
@@ -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++
+ LICENSE view
@@ -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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ parallel-tasks.cabal view
@@ -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 <nccb@kent.ac.uk>+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.*