packages feed

pooled-io (empty) → 0.0

raw patch · 8 files changed

+368/−0 lines, 8 filesdep +basedep +deepseqdep +pooled-iosetup-changed

Dependencies added: base, deepseq, pooled-io, transformers, unsafe, utility-ht

Files

+ LICENSE view
@@ -0,0 +1,27 @@+Copyright (c) Henning Thielemann 2014++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:+1. Redistributions of source code must retain the above copyright+   notice, this list of conditions and the following disclaimer.+2. 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.+3. Neither the name of the author nor the names of his contributors+   may be used to endorse or promote products derived from this software+   without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE REGENTS 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 AUTHORS 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.lhs view
@@ -0,0 +1,3 @@+#! /usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain
+ example/Main.hs view
@@ -0,0 +1,45 @@+{- |+This example shows parallel computation with data dependencies.+Since the actual results are stored in files+the 'createFile' function returns the filepath of the created file as result.+If you run the example with two threads (@+RTS -N2 -RTS@)+then you will see how the machine consumes 200% computation time.+That is, computations are run in parallel until all threads are busy.+Additionally you see that the final 'zipSum'+runs only after the three files are created.+This is the data dependency detection at work.+-}+module Main where++import qualified Control.Concurrent.PooledIO.InOrder as PooledIO+import qualified System.IO as IO++import Control.Monad (liftM3)+++createFile :: Int -> Int -> Int -> IO FilePath+createFile fileId number chunkSize = do+   let name = "test" ++ show fileId+   IO.withFile name IO.WriteMode $ \h -> do+      IO.hSetBuffering h IO.LineBuffering+      IO.hPutStr h $ unlines $+         map (\n -> show $ sum $ take chunkSize $ iterate (1+) (n::Integer)) $+         take number $ iterate (fromIntegral chunkSize +) (0::Integer)+   return name++zipSum :: FilePath -> FilePath -> FilePath -> FilePath -> IO ()+zipSum out in0 in1 in2 = do+   let readNums path = fmap (map read . lines) $ readFile path+   writeFile out . unlines . map show =<<+      liftM3+         (zipWith3 (\x y z -> x+y+z :: Integer))+         (readNums in0)+         (readNums in1)+         (readNums in2)++main :: IO ()+main = PooledIO.run $ do+   file0 <- PooledIO.fork $ createFile 0 100 2000000+   file1 <- PooledIO.fork $ createFile 1 100 1000000+   file2 <- PooledIO.fork $ createFile 2 100 2000000+   PooledIO.fork $ zipSum "total" file0 file1 file2
+ pooled-io.cabal view
@@ -0,0 +1,75 @@+Name:             pooled-io+Version:          0.0+License:          BSD3+License-File:     LICENSE+Author:           Henning Thielemann <haskell@henning-thielemann.de>+Maintainer:       Henning Thielemann <haskell@henning-thielemann.de>+Homepage:         http://code.haskell.org/~thielema/pooled-io/+Category:         Parallelism+Synopsis:         Run jobs on a limited number of threads and support data dependencies+Description:+  The motivation for this package was to run computations on multiple cores+  that need to write intermediate results to disk.+  The functions restrict the number of simultaneously running jobs+  to a user given number or to the number of capabilities+  the Haskell program was started with,+  i.e. the number after the RTS option @-N@.+  .+  There some flavors of this functionality:+  .+  * "Control.Concurrent.PooledIO.Independent":+    run independent actions without results in parallel+  .+  * "Control.Concurrent.PooledIO.Final":+    run independent actions with a final result in parallel+  .+  * "Control.Concurrent.PooledIO.InOrder":+    run jobs in parallel with data dependencies like @make -j n@+  .+  Related packages:+  .+  * @lazyio@: interleave IO actions in a single thread+Tested-With:      GHC==7.4.1+Cabal-Version:    >=1.8+Build-Type:       Simple++Flag buildExamples+  description: Build example executables+  default:     False++Source-Repository this+  Tag:         0.0+  Type:        darcs+  Location:    http://code.haskell.org/~thielema/pooled-io/++Source-Repository head+  Type:        darcs+  Location:    http://code.haskell.org/~thielema/pooled-io/++Library+  Build-Depends:+    transformers >=0.2.2 && <0.4,+    deepseq >=1.3 && <1.4,+    unsafe >=0.0 && <0.1,+    utility-ht >=0.0.9 && <0.1,+    base >=4 && <5++  GHC-Options:      -Wall+  Hs-Source-Dirs:   src+  Exposed-Modules:+    Control.Concurrent.PooledIO.Independent+    Control.Concurrent.PooledIO.Final+    Control.Concurrent.PooledIO.InOrder+  Other-Modules:+    Control.Concurrent.PooledIO.Monad++Executable pooled-io-demo+  If flag(buildExamples)+    Build-Depends:+      pooled-io,+      base+  Else+    Buildable: False++  GHC-Options:      -Wall -threaded+  Main-Is:          example/Main.hs
+ src/Control/Concurrent/PooledIO/Final.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{- |+This module implements something similar to+"Control.Concurrent.PooledIO.InOrder",+but since it is restricted to an 'Applicative' interface+we can implement it without 'unsafeInterleaveIO'.+-}+module Control.Concurrent.PooledIO.Final (+   T, run, runLimited, fork,+   ) where++import qualified Control.Concurrent.PooledIO.Monad as Pool+import Control.DeepSeq (NFData)++import Control.Monad (join)+import Control.Applicative (Applicative)+import Data.Functor.Compose (Compose(Compose))+++newtype T a = Cons (Compose Pool.T IO a)+   deriving (Functor, Applicative)+++{- |+This runs an action parallelly to the starting thread.+Since it is an Applicative Functor and not a Monad,+there are no data dependencies between the actions+and thus all actions in a 'T' can be run parallelly.+Only the 'IO' actions are parallelised+but not the combining function passed to 'liftA2' et.al.+That is, the main work must be done in the 'IO' actions+in order to benefit from parallelisation.+-}+fork :: (NFData a) => IO a -> T a+fork = Cons . Compose . Pool.fork++{- |+'runLimited' with a maximum of @numCapabilites@ threads.+-}+run :: T a -> IO a+run = Pool.withNumCapabilities runLimited++{- |+@runLimited n@ runs several actions in a pool with at most @n@ threads.+-}+runLimited :: Int -> T a -> IO a+runLimited maxThreads (Cons (Compose m)) =+   join $ Pool.runLimited maxThreads m
+ src/Control/Concurrent/PooledIO/InOrder.hs view
@@ -0,0 +1,78 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+module Control.Concurrent.PooledIO.InOrder (+   T, run, runLimited, fork,+   ) where++import qualified Control.Concurrent.PooledIO.Monad as Pool+import Control.DeepSeq (NFData)++import qualified System.Unsafe as Unsafe++import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Applicative (Applicative)+++newtype T a = Cons (Pool.T a)+   deriving (Functor, Applicative, Monad)++{-+The 'complete' MVar makes sure+that we do not run more threads than capabilities.+The 'result' MVar makes sure+that we run an action only after all of its inputs are evaluated.+-}+{- |+'fork' runs an IO action in parallel+while respecting a maximum number of threads.+Evaluating the result of 'T'+waits for the termination of the according thread.++Unfortunately, this means that sometimes threads are bored:++> foo a b = do+>    c <- fork $ f a+>    d <- fork $ g c+>    e <- fork $ h b++Here the execution of @g c@ reserves a thread+but starts with waiting for the evaluation of @c@.+It would be certainly better to execute @h b@ first.+You may relax this problem by moving dependent actions+away from another as much as possible.+It would be optimal to have an @OutOfOrder@ monad,+but this is more difficult to implement.++Although we fork all actions in order,+the fork itself might re-order the actions.+Thus the actions must not rely on a particular order+other than the order imposed by data dependencies.+We enforce with the 'NFData' constraint+that the computation is actually completed+when the thread terminates.++Currently the monad does not handle exceptions.+It's certainly best to use a package with explicit exception handling+like @explicit-exception@ in order to tunnel exception information+from the forked action to the main thread.++Although 'fork' has almost the same type signature as 'liftIO'+we do not define @instance MonadIO InOrder.T@+since this definition would not satisfy the laws required by the 'MonadIO' class.+-}+fork :: (NFData a) => IO a -> T a+fork act =+   Cons $+   liftIO . Unsafe.interleaveIO =<< Pool.fork act++{- |+'runLimited' with a maximum of @numCapabilites@ threads.+-}+run :: T a -> IO a+run = Pool.withNumCapabilities runLimited++{- |+@runLimited n@ runs several actions in a pool with at most @n@ threads.+-}+runLimited :: Int -> T a -> IO a+runLimited maxThreads (Cons m) =+   Pool.runLimited maxThreads m
+ src/Control/Concurrent/PooledIO/Independent.hs view
@@ -0,0 +1,43 @@+module Control.Concurrent.PooledIO.Independent (+   run,+   runLimited,+   runUnlimited,+   ) where++import Control.Concurrent.PooledIO.Monad (forkFinally, withNumCapabilities)+import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar)+import Control.Exception (evaluate)++import Control.Monad (replicateM_)+++{- |+Execute all actions parallelly+but run at most @numCapabilities@ threads at once.+Stop when all actions are finished.+-}+run :: [IO ()] -> IO ()+run = withNumCapabilities runLimited++runLimited :: Int -> [IO ()] -> IO ()+runLimited numCaps acts = do+   let (start, queue) = splitAt numCaps acts+   n <- evaluate $ length start+   mvar <- newEmptyMVar+   mapM_ (forkFinally mvar) start+   mapM_ (\act -> takeMVar mvar >> forkFinally mvar act) queue+   replicateM_ n $ takeMVar mvar++{- |+Execute all actions parallelly without a bound an the number of threads.+Stop when all actions are finished.+-}+runUnlimited :: [IO ()] -> IO ()+runUnlimited acts =+   mapM_ takeMVar =<< mapM fork acts++fork :: IO () -> IO (MVar ())+fork act = do+   mvar <- newEmptyMVar+   forkFinally mvar act+   return mvar
+ src/Control/Concurrent/PooledIO/Monad.hs view
@@ -0,0 +1,49 @@+module Control.Concurrent.PooledIO.Monad where++import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)+import Control.Concurrent (forkIO, getNumCapabilities)+import Control.DeepSeq (NFData, deepseq)+import Control.Exception (finally)++import qualified Control.Monad.Trans.State as MS+import qualified Control.Monad.Trans.Reader as MR+import qualified Control.Monad.Trans.Class as MT+import Control.Monad.IO.Class (MonadIO, liftIO)++import Control.Monad (replicateM_)+import Control.Functor.HT (void)+++type T = MR.ReaderT (MVar ()) (MS.StateT Int IO)+++fork :: (NFData a) => IO a -> T (IO a)+fork act = do+   complete <- MR.ask+   initial <- MT.lift MS.get+   if initial>0+     then MT.lift $ MS.put (initial-1)+     else liftIO $ takeMVar complete+   liftIO $ do+      result <- newEmptyMVar+      forkFinally complete $ do+         r <- act+         deepseq r $ putMVar result r+      return $ takeMVar result++forkFinally :: MVar () -> IO () -> IO ()+forkFinally mvar act =+   void $ forkIO $ finally act $ putMVar mvar ()++withNumCapabilities :: (Int -> a -> IO b) -> a -> IO b+withNumCapabilities run acts = do+   numCaps <- getNumCapabilities+   run numCaps acts++runLimited :: Int -> T a -> IO a+runLimited maxThreads m = do+   complete <- newEmptyMVar+   (result, uninitialized) <-+      MS.runStateT (MR.runReaderT m complete) maxThreads+   replicateM_ (maxThreads-uninitialized) $ takeMVar complete+   return result