pooled-io 0.0.1.1 → 0.0.1.2
raw patch · 3 files changed
+33/−32 lines, 3 filesdep +concurrent-splitPVP ok
version bump matches the API change (PVP)
Dependencies added: concurrent-split
API changes (from Hackage documentation)
Files
- pooled-io.cabal +4/−3
- src/Control/Concurrent/PooledIO/Independent.hs +14/−14
- src/Control/Concurrent/PooledIO/Monad.hs +15/−15
pooled-io.cabal view
@@ -1,5 +1,5 @@ Name: pooled-io-Version: 0.0.1.1+Version: 0.0.1.2 License: BSD3 License-File: LICENSE Author: Henning Thielemann <haskell@henning-thielemann.de>@@ -33,7 +33,7 @@ * @async@: start threads and wait for their results, forward exceptions, but do not throttle concurrency with respect to number of available cores .- * @parallel-tasks@+ * @parallel-tasks@: Tested-With: GHC==7.4.1 Cabal-Version: >=1.8 Build-Type: Simple@@ -43,7 +43,7 @@ default: False Source-Repository this- Tag: 0.0.1.1+ Tag: 0.0.1.2 Type: darcs Location: http://code.haskell.org/~thielema/pooled-io/ @@ -53,6 +53,7 @@ Library Build-Depends:+ concurrent-split >=0.0 && <0.1, transformers >=0.2.2 && <0.5, deepseq >=1.3 && <1.5, unsafe >=0.0 && <0.1,
src/Control/Concurrent/PooledIO/Independent.hs view
@@ -9,7 +9,7 @@ import Control.Concurrent.PooledIO.Monad (withNumCapabilities, chooseNumCapabilities, forkFinally, forkTry, takeMVarTry, runTry)-import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar)+import qualified Control.Concurrent.Split.MVar as MVar import Control.Exception (evaluate) import Control.Monad (replicateM_)@@ -28,10 +28,10 @@ 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+ (mvarIn, mvarOut) <- MVar.newEmpty+ mapM_ (forkFinally mvarIn) start+ mapM_ (\act -> MVar.take mvarOut >> forkFinally mvarIn act) queue+ replicateM_ n $ MVar.take mvarOut {- | Execute all actions parallelly without a bound an the number of threads.@@ -39,13 +39,13 @@ -} runUnlimited :: [IO ()] -> IO () runUnlimited acts =- mapM_ takeMVar =<< mapM fork acts+ mapM_ MVar.take =<< mapM fork acts -fork :: IO () -> IO (MVar ())+fork :: IO () -> IO (MVar.Out ()) fork act = do- mvar <- newEmptyMVar- forkFinally mvar act- return mvar+ (mvarIn, mvarOut) <- MVar.newEmpty+ forkFinally mvarIn act+ return mvarOut {- |@@ -63,8 +63,8 @@ runOneBreaksAll numCaps acts = do let (start, queue) = splitAt numCaps acts n <- evaluate $ length start- mvar <- newEmptyMVar+ (mvarIn, mvarOut) <- MVar.newEmpty runTry $ do- mapM_ (forkTry mvar) start- mapM_ (\act -> takeMVarTry mvar >> forkTry mvar act) queue- replicateM_ n $ takeMVarTry mvar+ mapM_ (forkTry mvarIn) start+ mapM_ (\act -> takeMVarTry mvarOut >> forkTry mvarIn act) queue+ replicateM_ n $ takeMVarTry mvarOut
src/Control/Concurrent/PooledIO/Monad.hs view
@@ -1,7 +1,7 @@ {-# LANGUAGE Safe #-} module Control.Concurrent.PooledIO.Monad where -import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)+import qualified Control.Concurrent.Split.MVar as MVar import Control.Concurrent (ThreadId, forkIO, getNumCapabilities, myThreadId, killThread) import Control.DeepSeq (NFData, ($!!)) import Control.Exception (SomeException, finally, try, throw)@@ -18,33 +18,33 @@ import qualified Data.Set as Set; import Data.Set (Set) -type T = MR.ReaderT (MVar ()) (MS.StateT Int IO)+type T = MR.ReaderT (MVar.In (), MVar.Out ()) (MS.StateT Int IO) fork :: (NFData a) => IO a -> T (IO a) fork act = do- complete <- MR.ask+ (completeIn, completeOut) <- MR.ask initial <- MT.lift MS.get if initial>0 then MT.lift $ MS.put (initial-1)- else liftIO $ takeMVar complete+ else liftIO $ MVar.take completeOut liftIO $ do- result <- newEmptyMVar- forkFinally complete $ (putMVar result $!!) =<< act- return $ takeMVar result+ (resultIn, resultOut) <- MVar.newEmpty+ forkFinally completeIn $ (MVar.put resultIn $!!) =<< act+ return $ MVar.take resultOut -forkFinally :: MVar () -> IO () -> IO ()+forkFinally :: MVar.In () -> IO () -> IO () forkFinally mvar act =- void $ forkIO $ finally act $ putMVar mvar ()+ void $ forkIO $ finally act $ MVar.put mvar () forkTry :: (NFData a) =>- MVar (ThreadId, Either SomeException a) -> IO a ->+ MVar.In (ThreadId, Either SomeException a) -> IO a -> MS.StateT (Set ThreadId) IO () forkTry mvar act = do thread <- liftIO $ forkIO $- applyStrictRight (putMVar mvar) =<< liftM2 (,) myThreadId (try act)+ applyStrictRight (MVar.put mvar) =<< liftM2 (,) myThreadId (try act) MS.modify (Set.insert thread) applyStrictRight :: (NFData a) => ((id, Either e a) -> b) -> (id, Either e a) -> b@@ -54,10 +54,10 @@ Right a -> f . (,) thread . Right $!! a takeMVarTry ::- MVar (ThreadId, Either SomeException a) ->+ MVar.Out (ThreadId, Either SomeException a) -> MS.StateT (Set ThreadId) IO a takeMVarTry mvar = do- (thread, ee) <- liftIO $ takeMVar mvar+ (thread, ee) <- liftIO $ MVar.take mvar MS.modify (Set.delete thread) case ee of Left e -> do liftIO . Fold.mapM_ killThread =<< MS.get; liftIO $ throw e@@ -77,8 +77,8 @@ runLimited :: Int -> T a -> IO a runLimited maxThreads m = do- complete <- newEmptyMVar+ complete@(_, completeOut) <- MVar.newEmpty (result, uninitialized) <- MS.runStateT (MR.runReaderT m complete) maxThreads- replicateM_ (maxThreads-uninitialized) $ takeMVar complete+ replicateM_ (maxThreads-uninitialized) $ MVar.take completeOut return result