packages feed

cio (empty) → 0.1.0

raw patch · 5 files changed

+320/−0 lines, 5 filesdep +basedep +monad-stmdep +mtlsetup-changed

Dependencies added: base, monad-stm, mtl, parallel-io, stm

Files

+ LICENSE view
@@ -0,0 +1,22 @@+Copyright (c) 2013, Nikita Volkov++Permission is hereby granted, free of charge, to any person+obtaining a copy of this software and associated documentation+files (the "Software"), to deal in the Software without+restriction, including without limitation the rights to use,+copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following+conditions:++The above copyright notice and this permission notice shall be+included in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT+HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ cio.cabal view
@@ -0,0 +1,80 @@+name:+  cio+version:+  0.1.0+synopsis:+  A monad for concurrent IO on a thread pool+description:+  Provides a monadic API for efficient concurrency based on a thread pool.+  The implementation is based on a "parallel-io" library.+category:+  Concurrency+homepage:+  https://github.com/nikita-volkov/cio +bug-reports:+  https://github.com/nikita-volkov/cio/issues +author:+  Nikita Volkov <nikita.y.volkov@mail.ru>+maintainer:+  Nikita Volkov <nikita.y.volkov@mail.ru>+copyright:+  (c) 2013, Nikita Volkov+license:+  MIT+license-file:+  LICENSE+build-type:+  Simple+cabal-version:+  >=1.10+++source-repository head+  type:+    git+  location:+    git://github.com/nikita-volkov/cio.git+++library+  hs-source-dirs:+    src+  exposed-modules:+    CIO+  other-modules:+    CIO.Prelude+  build-depends:+    parallel-io,+    monad-stm,+    stm,+    mtl,+    base >= 4.5 && < 5+  default-extensions:+    DeriveGeneric+    BangPatterns+    PatternGuards+    GADTs+    StandaloneDeriving+    MultiParamTypeClasses+    ScopedTypeVariables+    FlexibleInstances+    TypeFamilies+    TypeOperators+    FlexibleContexts+    NoImplicitPrelude+    EmptyDataDecls+    DataKinds+    NoMonomorphismRestriction+    RankNTypes+    ConstraintKinds+    DefaultSignatures+    TupleSections+    OverloadedStrings+    TemplateHaskell+    QuasiQuotes+    DeriveDataTypeable+    GeneralizedNewtypeDeriving+    RecordWildCards+    MultiWayIf+  default-language:+    Haskell2010
+ src/CIO.hs view
@@ -0,0 +1,155 @@+module CIO +  (+    CIO,+    runCIO,+    runCIO',+    MonadCIO(..),+    mapMConcurrently,+    mapMConcurrently',+    mapMConcurrently_,+    forMConcurrently,+    forMConcurrently',+    forMConcurrently_,+    distributeConcurrently,+    distributeConcurrently_,+  )+  where++import CIO.Prelude+import qualified Control.Concurrent.ParallelIO.Local as ParallelIO+++-- | Concurrent IO. A composable monad of IO actions executable in a shared pool of threads.+newtype CIO r = CIO (ReaderT (ParallelIO.Pool, Int) IO r)+  deriving (Functor, Applicative, Monad)++instance MonadIO CIO where+  liftIO io = CIO $ lift io++instance MonadSTM CIO where+  liftSTM = CIO . liftSTM++-- | Run with a pool of the specified size.+runCIO :: Int -> CIO r -> IO r+runCIO numCapabilities (CIO t) =+  ParallelIO.withPool numCapabilities $ \pool -> runReaderT t (pool, numCapabilities)++-- | Run with a pool the size of the amount of available processors.+runCIO' :: CIO r -> IO r+runCIO' cio = do+  numCapabilities <- getNumCapabilities+  runCIO numCapabilities cio+++class (Monad m) => MonadCIO m where+  -- | Get the maximum number of available threads, which is set in 'runCIO'.+  getPoolNumCapabilities :: m Int+  -- | Same as @Control.Monad.'Control.Monad.sequence'@, but performs concurrently. +  sequenceConcurrently :: [m a] -> m [a]+  -- | Same as 'sequenceConcurrently' with a difference that +  -- it does not maintain the order of results,+  -- which allows it to execute a bit more efficiently.+  sequenceConcurrently' :: [m a] -> m [a]+  -- | Same as @Control.Monad.'Control.Monad.sequence_'@, but performs concurrently. +  -- Blocks the calling thread until all actions are finished.+  sequenceConcurrently_ :: [m a] -> m ()++instance MonadCIO CIO where+  getPoolNumCapabilities =+    CIO $ do+      (_, z) <- ask+      return z+  sequenceConcurrently actions = +    CIO $ do+      env@(pool, _) <- ask+      lift $ ParallelIO.parallel pool $ map (envToCIOToIO env) actions +    where+      envToCIOToIO env (CIO t) = runReaderT t env+  sequenceConcurrently' actions = +    CIO $ do+      env@(pool, _) <- ask+      lift $ ParallelIO.parallelInterleaved pool $ map (envToCIOToIO env) actions +    where+      envToCIOToIO env (CIO t) = runReaderT t env+  sequenceConcurrently_ actions = +    CIO $ do+      env@(pool, _) <- ask+      lift $ ParallelIO.parallel_ pool $ map (envToCIOToIO env) actions +    where+      envToCIOToIO env (CIO t) = runReaderT t env++instance (MonadCIO m) => MonadCIO (ReaderT r m) where+  getPoolNumCapabilities = lift getPoolNumCapabilities+  sequenceConcurrently actions = do+    env <- ask+    let cioActions = map (flip runReaderT env) actions+    lift $ sequenceConcurrently cioActions+  sequenceConcurrently' actions = do+    env <- ask+    let cioActions = map (flip runReaderT env) actions+    lift $ sequenceConcurrently' cioActions+  sequenceConcurrently_ actions = do+    env <- ask+    let cioActions = map (flip runReaderT env) actions+    lift $ sequenceConcurrently_ cioActions++instance (MonadCIO m, Monoid w) => MonadCIO (WriterT w m) where+  getPoolNumCapabilities = lift getPoolNumCapabilities+  sequenceConcurrently actions = do+    let cioActions = map runWriterT actions+    WriterT $ do+      (as, ws) <- return . unzip =<< sequenceConcurrently cioActions+      return (as, mconcat ws)+  sequenceConcurrently' actions = do+    let cioActions = map runWriterT actions+    WriterT $ do+      (as, ws) <- return . unzip =<< sequenceConcurrently' cioActions+      return (as, mconcat ws)+  sequenceConcurrently_ actions = do+    let cioActions = map execWriterT actions+    WriterT $ do+      ws <- sequenceConcurrently' cioActions+      return ((), mconcat ws)+++mapMConcurrently :: (MonadCIO m) => (a -> m b) -> [a] -> m [b]+mapMConcurrently f = sequenceConcurrently . map f++mapMConcurrently' :: (MonadCIO m) => (a -> m b) -> [a] -> m [b]+mapMConcurrently' f = sequenceConcurrently' . map f++mapMConcurrently_ :: (MonadCIO m) => (a -> m b) -> [a] -> m ()+mapMConcurrently_ f = sequenceConcurrently_ . map f++forMConcurrently :: (MonadCIO m) => [a] -> (a -> m b) -> m [b]+forMConcurrently = flip mapMConcurrently++forMConcurrently' :: (MonadCIO m) => [a] -> (a -> m b) -> m [b]+forMConcurrently' = flip mapMConcurrently'++forMConcurrently_ :: (MonadCIO m) => [a] -> (a -> m b) -> m ()+forMConcurrently_ = flip mapMConcurrently_++replicateMConcurrently :: (MonadCIO m) => Int -> m a -> m [a]+replicateMConcurrently n = sequenceConcurrently . replicate n++replicateMConcurrently' :: (MonadCIO m) => Int -> m a -> m [a]+replicateMConcurrently' n = sequenceConcurrently' . replicate n++replicateMConcurrently_ :: (MonadCIO m) => Int -> m a -> m ()+replicateMConcurrently_ n = sequenceConcurrently_ . replicate n++-- |+-- Run the provided side-effecting action on all available threads and +-- collect the results. The order of results may vary from run to run.+distributeConcurrently :: (MonadCIO m) => m a -> m [a]+distributeConcurrently action = do+  n <- getPoolNumCapabilities+  replicateMConcurrently' n action++-- |+-- Run the provided side-effecting action on all available threads.+distributeConcurrently_ :: (MonadCIO m) => m a -> m ()+distributeConcurrently_ action = do+  n <- getPoolNumCapabilities+  replicateMConcurrently_ n action
+ src/CIO/Prelude.hs view
@@ -0,0 +1,61 @@+module CIO.Prelude +  ( +    module Exports,+    traceM,+  )+  where++-- base+import Prelude as Exports hiding (concat, foldr, mapM_, sequence_, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, mapM, sequence, FilePath, id, (.))+import Control.Monad as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Applicative as Exports+import Control.Arrow as Exports hiding (left, right)+import Control.Category as Exports+import Data.Monoid as Exports+import Data.Foldable as Exports+import Data.Traversable as Exports hiding (for)+import Data.Maybe as Exports+import Data.List as Exports hiding (concat, foldr, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, find, maximumBy, minimumBy, mapAccumL, mapAccumR, foldl')+import Data.Tuple as Exports+import Data.Ord as Exports (Down(..))+import Data.String as Exports+import Data.Int as Exports+import Data.Word as Exports+import Data.Ratio as Exports+import Data.Fixed as Exports+import Data.Ix as Exports+import Data.Data as Exports+import Text.Read as Exports (readMaybe, readEither)+import Control.Exception as Exports hiding (tryJust)+import Control.Concurrent as Exports hiding (yield)+import System.Mem.StableName as Exports+import System.Timeout as Exports+import System.Exit as Exports+import System.IO.Unsafe as Exports+import System.IO as Exports (Handle, hClose)+import System.IO.Error as Exports+import Unsafe.Coerce as Exports+import GHC.Exts as Exports (groupWith, sortWith)+import GHC.Generics as Exports (Generic)+import GHC.IO.Exception as Exports+import Debug.Trace as Exports+import Data.IORef as Exports+import Data.STRef as Exports+import Control.Monad.ST as Exports++-- mtl+import Control.Monad.Identity as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Monad.State as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Monad.Reader as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Monad.Writer as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Monad.Trans as Exports++-- stm+import Control.Concurrent.STM as Exports++-- monad-stm+import Control.Monad.STM.Class as Exports+++traceM :: (Monad m) => String -> m ()+traceM s = trace s $ return ()