unliftio 0.2.9.0 → 0.2.10

raw patch · 7 files changed

+477/−13 lines, 7 filesdep +containersPVP ok

version bump matches the API change (PVP)

Dependencies added: containers

API changes (from Hackage documentation)

Files

• ChangeLog.md +4/−0
• README.md +2/−2
• src/UnliftIO/Async.hs +14/−0
• src/UnliftIO/Internals/Async.hs +249/−7
• src/UnliftIO/STM.hs +0/−1
• test/UnliftIO/PooledAsyncSpec.hs +203/−0
• unliftio.cabal +5/−3

@@ -1,5 +1,9 @@ # Changelog for unliftio +## 0.2.10++* Add pooling related functions for unliftio+ ## 0.2.9.0  * Add the new `Conc` datatype as a more efficient alternative to `Concurrently`

@@ -82,7 +82,7 @@     (\h -> runReaderT (inner h) env) ``` -I dare you to try to and accomplish this with `MonadIO` and+I dare you to try and accomplish this with `MonadIO` and `liftIO`. It simply can't be done. (If you're looking for the technical reason, it's because `IO` appears in [negative/argument position](https://www.fpcomplete.com/blog/2016/11/covariance-contravariance)@@ -174,7 +174,7 @@ ```haskell timeout :: MonadUnliftIO m => Int -> m a -> m (Maybe a) timeout x y = do-  u <- askUnliftIO+  (u :: UnliftIO m) <- askUnliftIO   liftIO $ System.Timeout.timeout x $ unliftIO u y ``` 

@@ -43,6 +43,20 @@     -- ** Linking     link, link2, +    -- ** Pooled concurrency+    pooledMapConcurrentlyN,+    pooledMapConcurrently,+    pooledMapConcurrentlyN_,+    pooledMapConcurrently_,+    pooledForConcurrentlyN,+    pooledForConcurrently,+    pooledForConcurrentlyN_,+    pooledForConcurrently_,+    pooledReplicateConcurrentlyN,+    pooledReplicateConcurrently,+    pooledReplicateConcurrentlyN_,+    pooledReplicateConcurrently_,+     -- * Convenient utilities     race, race_,     concurrently, concurrently_,

@@ -1,4 +1,5 @@ {-# LANGUAGE CPP                 #-}+{-# LANGUAGE TupleSections       #-} {-# LANGUAGE DeriveDataTypeable  #-} {-# LANGUAGE DeriveFunctor       #-} {-# LANGUAGE DeriveGeneric       #-}@@ -10,7 +11,7 @@ module UnliftIO.Internals.Async where  import           Control.Applicative-import           Control.Concurrent       (threadDelay)+import           Control.Concurrent       (threadDelay, getNumCapabilities) import qualified Control.Concurrent       as C import           Control.Concurrent.Async (Async) import qualified Control.Concurrent.Async as A@@ -20,6 +21,7 @@ import           Control.Monad.IO.Unlift import           Data.Foldable            (for_, traverse_) import           Data.Typeable            (Typeable)+import           Data.IORef (IORef, readIORef, atomicWriteIORef, newIORef, atomicModifyIORef') import qualified UnliftIO.Exception       as UE  -- For the implementation of Conc below, we do not want any of the@@ -32,14 +34,10 @@ #if MIN_VERSION_base(4,9,0) import           Data.Semigroup #else-import           Data.Foldable            (Foldable) import           Data.Monoid              hiding (Alt)-import           Data.Traversable         (Traversable) #endif--#if MIN_VERSION_base(4,7,0)-import           Data.Traversable         (traverse)-#endif+import           Data.Foldable            (Foldable, toList)+import           Data.Traversable         (Traversable, for, traverse)  -- | Unlifted 'A.async'. --@@ -835,3 +833,247 @@ -------------------------------------------------------------------------------- #endif --------------------------------------------------------------------------------++-- | Like 'mapConcurrently' from async, but instead of one thread per+-- element, it does pooling from a set of threads. This is useful in+-- scenarios where resource consumption is bounded and for use cases+-- where too many concurrent tasks aren't allowed.+--+-- === __Example usage__+--+-- @+-- import Say+--+-- action :: Int -> IO Int+-- action n = do+--   tid <- myThreadId+--   sayString $ show tid+--   threadDelay (2 * 10^6) -- 2 seconds+--   return n+--+-- main :: IO ()+-- main = do+--   yx \<- pooledMapConcurrentlyN 5 (\\x -\> action x) [1..5]+--   print yx+-- @+--+-- On executing you can see that five threads have been spawned:+--+-- @+-- \$ ./pool+-- ThreadId 36+-- ThreadId 38+-- ThreadId 40+-- ThreadId 42+-- ThreadId 44+-- [1,2,3,4,5]+-- @+--+--+-- Let's modify the above program such that there are less threads+-- than the number of items in the list:+--+-- @+-- import Say+--+-- action :: Int -> IO Int+-- action n = do+--   tid <- myThreadId+--   sayString $ show tid+--   threadDelay (2 * 10^6) -- 2 seconds+--   return n+--+-- main :: IO ()+-- main = do+--   yx \<- pooledMapConcurrentlyN 3 (\\x -\> action x) [1..5]+--   print yx+-- @+-- On executing you can see that only three threads are active totally:+--+-- @+-- \$ ./pool+-- ThreadId 35+-- ThreadId 37+-- ThreadId 39+-- ThreadId 35+-- ThreadId 39+-- [1,2,3,4,5]+-- @+--+-- @since 0.2.10+pooledMapConcurrentlyN :: (MonadUnliftIO m, Traversable t)+                      => Int -- ^ Max. number of threads. Should not be less than 1.+                      -> (a -> m b) -> t a -> m (t b)+pooledMapConcurrentlyN numProcs f xs =+    withRunInIO $ \run -> pooledMapConcurrentlyIO numProcs (run . f) xs++-- | Similar to 'pooledMapConcurrentlyN' but with number of threads+-- set from 'getNumCapabilities'. Usually this is useful for CPU bound+-- tasks.+--+-- @since 0.2.10+pooledMapConcurrently :: (MonadUnliftIO m, Traversable t) => (a -> m b) -> t a -> m (t b)+pooledMapConcurrently f xs = do+  withRunInIO $ \run -> do+    numProcs <- getNumCapabilities+    pooledMapConcurrentlyIO numProcs (run . f) xs++-- | Similar to 'pooledMapConcurrentlyN' but with flipped arguments.+--+-- @since 0.2.10+pooledForConcurrentlyN :: (MonadUnliftIO m, Traversable t)+                      => Int -- ^ Max. number of threads. Should not be less than 1.+                      -> t a -> (a -> m b) -> m (t b)+pooledForConcurrentlyN numProcs = flip (pooledMapConcurrentlyN numProcs)++-- | Similar to 'pooledForConcurrentlyN' but with number of threads+-- set from 'getNumCapabilities'. Usually this is useful for CPU bound+-- tasks.+--+-- @since 0.2.10+pooledForConcurrently :: (MonadUnliftIO m, Traversable t) => t a -> (a -> m b) -> m (t b)+pooledForConcurrently = flip pooledMapConcurrently++pooledMapConcurrentlyIO :: Traversable t => Int -> (a -> IO b) -> t a -> IO (t b)+pooledMapConcurrentlyIO numProcs f xs =+    if (numProcs < 1)+    then error "pooledMapconcurrentlyIO: number of threads < 1"+    else pooledMapConcurrentlyIO' numProcs f xs++-- | Performs the actual pooling for the tasks. This function will+-- continue execution until the task queue becomes empty. When one of+-- the pooled thread finishes it's task, it will pickup the next task+-- from the queue if an job is available.+pooledConcurrently+  :: Int -- ^ Max. number of threads. Should not be less than 1.+  -> IORef [a] -- ^ Task queue. These are required as inputs for the jobs.+  -> (a -> IO ()) -- ^ The task which will be run concurrently (but+                 -- will be pooled properly).+  -> IO ()+pooledConcurrently numProcs jobsVar f = do+  replicateConcurrently_ numProcs $ do+    let loop = do+          mbJob :: Maybe a <- atomicModifyIORef' jobsVar $ \x -> case x of+            [] -> ([], Nothing)+            var : vars -> (vars, Just var)+          case mbJob of+            Nothing -> return ()+            Just x -> do+              f x+              loop+     in loop++pooledMapConcurrentlyIO' ::+    Traversable t => Int  -- ^ Max. number of threads. Should not be less than 1.+                  -> (a -> IO b)+                  -> t a+                  -> IO (t b)+pooledMapConcurrentlyIO' numProcs f xs = do+  -- prepare one IORef per result...+  jobs :: t (a, IORef b) <-+    for xs (\x -> (x, ) <$> newIORef (error "pooledMapConcurrentlyIO': empty IORef"))+  -- ...put all the inputs in a queue..+  jobsVar :: IORef [(a, IORef b)] <- newIORef (toList jobs)+  -- ...run `numProcs` threads in parallel, each+  -- of them consuming the queue and filling in+  -- the respective IORefs.+  pooledConcurrently numProcs jobsVar $ \ (x, outRef) -> f x >>= atomicWriteIORef outRef      -- Read all the IORefs+  for jobs (\(_, outputRef) -> readIORef outputRef)++pooledMapConcurrentlyIO_' ::+  Foldable t => Int -> (a -> IO ()) -> t a -> IO ()+pooledMapConcurrentlyIO_' numProcs f jobs = do+  jobsVar :: IORef [a] <- newIORef (toList jobs)+  pooledConcurrently numProcs jobsVar f++pooledMapConcurrentlyIO_ :: Foldable t => Int -> (a -> IO b) -> t a -> IO ()+pooledMapConcurrentlyIO_ numProcs f xs =+    if (numProcs < 1)+    then error "pooledMapconcurrentlyIO_: number of threads < 1"+    else pooledMapConcurrentlyIO_' numProcs (\x -> f x >> return ()) xs++-- | Like 'pooledMapConcurrentlyN' but with the return value+-- discarded.+--+-- @since 0.2.10+pooledMapConcurrentlyN_ :: (MonadUnliftIO m, Foldable f)+                        => Int -- ^ Max. number of threads. Should not be less than 1.+                        -> (a -> m b) -> f a -> m ()+pooledMapConcurrentlyN_ numProcs f t =+  withRunInIO $ \run -> pooledMapConcurrentlyIO_ numProcs (run . f) t++-- | Like 'pooledMapConcurrently' but with the return value discarded.+--+-- @since 0.2.10+pooledMapConcurrently_ :: (MonadUnliftIO m, Foldable f) => (a -> m b) -> f a -> m ()+pooledMapConcurrently_ f t =+  withRunInIO $ \run -> do+    numProcs <- getNumCapabilities+    pooledMapConcurrentlyIO_ numProcs (run . f) t++-- | Like 'pooledMapConcurrently_' but with flipped arguments.+--+-- @since 0.2.10+pooledForConcurrently_ :: (MonadUnliftIO m, Foldable f) => f a -> (a -> m b) -> m ()+pooledForConcurrently_ = flip pooledMapConcurrently_++-- | Like 'pooledMapConcurrentlyN_' but with flipped arguments.+--+-- @since 0.2.10+pooledForConcurrentlyN_ :: (MonadUnliftIO m, Foldable t)+                        => Int -- ^ Max. number of threads. Should not be less than 1.+                        -> t a -> (a -> m b) -> m ()+pooledForConcurrentlyN_ numProcs = flip (pooledMapConcurrentlyN_ numProcs)+++-- | Pooled version of 'replicateConcurrently'. Performs the action in+-- the pooled threads.+--+-- @since 0.2.10+pooledReplicateConcurrentlyN :: (MonadUnliftIO m)+                             => Int -- ^ Max. number of threads. Should not be less than 1.+                             -> Int -- ^ Number of times to perform the action.+                             -> m a -> m [a]+pooledReplicateConcurrentlyN numProcs cnt task =+    if cnt < 1+    then return []+    else pooledMapConcurrentlyN numProcs (\_ -> task) [1..cnt]++-- | Similar to 'pooledReplicateConcurrentlyN' but with number of+-- threads set from 'getNumCapabilities'. Usually this is useful for+-- CPU bound tasks.+--+-- @since 0.2.10+pooledReplicateConcurrently :: (MonadUnliftIO m)+                            => Int -- ^ Number of times to perform the action.+                            -> m a -> m [a]+pooledReplicateConcurrently cnt task =+    if cnt < 1+    then return []+    else pooledMapConcurrently (\_ -> task) [1..cnt]++-- | Pooled version of 'replicateConcurrently_'. Performs the action in+-- the pooled threads.+--+-- @since 0.2.10+pooledReplicateConcurrentlyN_ :: (MonadUnliftIO m)+                              => Int -- ^ Max. number of threads. Should not be less than 1.+                              -> Int -- ^ Number of times to perform the action.+                              -> m a -> m ()+pooledReplicateConcurrentlyN_ numProcs cnt task =+  if cnt < 1+  then return ()+  else pooledMapConcurrentlyN_ numProcs (\_ -> task) [1..cnt]++-- | Similar to 'pooledReplicateConcurrently_' but with number of+-- threads set from 'getNumCapabilities'. Usually this is useful for+-- CPU bound tasks.+--+-- @since 0.2.10+pooledReplicateConcurrently_ :: (MonadUnliftIO m)+                             => Int -- ^ Number of times to perform the action.+                             -> m a -> m ()+pooledReplicateConcurrently_ cnt task =+  if cnt < 1+  then return ()+  else pooledMapConcurrently_ (\_ -> task) [1..cnt]

@@ -84,7 +84,6 @@ import qualified Control.Concurrent.STM as STM import Control.Monad.IO.Unlift import System.Mem.Weak (Weak)- #if MIN_VERSION_base(4, 8, 0) import GHC.Natural (Natural) #else

@@ -0,0 +1,203 @@+{-#LANGUAGE DeriveDataTypeable#-}+{-#LANGUAGE BangPatterns#-}++module UnliftIO.PooledAsyncSpec (spec) where++import Test.Hspec+import Control.Concurrent+import Data.List (sort)+import Test.QuickCheck+import qualified Data.Set as Set+import Data.Functor ((<$>))+import UnliftIO++data MyPooledException = PoolHellException+                         deriving (Show, Typeable)++instance Exception MyPooledException++-- | Strip out duplicates. (Taken from rio)+nubOrd :: Ord a => [a] -> [a]+nubOrd =+  loop Set.empty+  where+    loop _ [] = []+    loop !s (a:as)+      | a `Set.member` s = loop s as+      | otherwise = a : loop (Set.insert a s) as++spec :: Spec+spec = do+  let exAction :: Int -> IO Int+      exAction x = do+        if (x == 2) then throwIO PoolHellException else return ()+        return x++      action :: Int -> IO ThreadId+      action x = do+         threadDelay (2 * 10^5)+         myThreadId++      myVar :: IO (TVar Int)+      myVar = atomically $ newTVar 0++      maxTVar :: Int -> TVar Int -> IO ()+      maxTVar cval tvar = do+         atomically $ do+           v <- readTVar tvar+           if cval >= v+           then writeTVar tvar cval+           else return ()++      poolException :: Selector MyPooledException+      poolException = const True++  describe "pooled mapConcurrencyN" $ do+    it "Throws exception properly" $ do+       (pooledMapConcurrentlyN 5 exAction [1..5]) `shouldThrow` poolException++    it "total thread should be >= 1" $ do+       (pooledMapConcurrentlyN 0 action [1..5]) `shouldThrow` anyErrorCall++    it "should not spawn more than five threads for five concurrent tasks" $ do+       xs <- (pooledMapConcurrentlyN 5 action [1..5])+       (length $ nubOrd xs) `shouldSatisfy` (<= (5 :: Int))++    it "should not spawn more than three threads for five concurrent tasks" $ do+       xs <- (pooledMapConcurrentlyN 3 action [1..5])+       (length $ nubOrd xs) `shouldSatisfy` (<= (3 :: Int))++    it "should spawn only one thread" $ do+       xs <- (pooledMapConcurrentlyN 1 action [1..5])+       (length $ nubOrd xs) `shouldBe` 1++    it "never uses more than the given number of pools and doesn't miss any return values" $+        forAllShrink ((+ 1) . abs <$> arbitrary) (filter (>= 1) . shrink) $ \threads ->+            property $ \list -> do+                threadIdsVar <- newTVarIO []+                let go :: Int -> IO Int+                    go i = do+                        tid <- myThreadId+                        atomically $ modifyTVar threadIdsVar (tid :)+                        return i+                list' <- pooledMapConcurrentlyN threads go list+                sort list' `shouldBe` sort list+                tids <- readTVarIO threadIdsVar+                length (nubOrd tids) `shouldSatisfy` (<= threads)++  describe "pooled mapConcurrencyN_" $ do+    it "Throws exception properly" $ do+       (pooledMapConcurrentlyN_ 5 exAction [1..5]) `shouldThrow` poolException++    it "total thread should be >= 1" $ do+       (pooledMapConcurrentlyN_ 0 action [1..5]) `shouldThrow` anyErrorCall++    it "find proper maximum value" $ do+       var <- myVar+       xs <- (pooledMapConcurrentlyN_ 5 (\x -> maxTVar x var) [1..5])+       newVar <- atomically $ readTVar var+       atomically $ writeTVar var 0+       newVar `shouldBe` 5++    it "find proper maximum value with 2 threads" $ do+       var <- myVar+       xs <- (pooledMapConcurrentlyN_ 2 (\x -> maxTVar x var) [1..5])+       newVar <- atomically $ readTVar var+       atomically $ writeTVar var 0+       newVar `shouldBe` 5++    it "find proper maximum value with 1 threads" $ do+       var <- myVar+       xs <- (pooledMapConcurrentlyN_ 1 (\x -> maxTVar x var) [1..5])+       newVar <- atomically $ readTVar var+       atomically $ writeTVar var 0+       newVar `shouldBe` 5++    it "make sure activity is happening in different threads" $ do+       let myThreads :: IO (TVar [ThreadId])+           myThreads = atomically $ newTVar []++           collectThreads :: TVar [ThreadId] -> IO ()+           collectThreads threadVar = do+             tid <- myThreadId+             atomically $ do+               tvar <- readTVar threadVar+               writeTVar threadVar (tid:tvar)+             threadDelay $ 2 * 10^5++       tid <- myThreads+       xs <- pooledMapConcurrentlyN_ 5 (\_ -> collectThreads tid) [1..5]+       tids <- atomically $ readTVar tid+       (length $ nubOrd tids) `shouldSatisfy` (<= 5)++    it "Not more than 5 threads will be spawned even if pooling is set to 8 " $ do+       let myThreads :: IO (TVar [ThreadId])+           myThreads = atomically $ newTVar []++           collectThreads :: TVar [ThreadId] -> IO ()+           collectThreads threadVar = do+             tid <- myThreadId+             atomically $ do+               tvar <- readTVar threadVar+               writeTVar threadVar (tid:tvar)+             threadDelay $ 2 * 10^5++       tid <- myThreads+       xs <- pooledMapConcurrentlyN_ 8 (\_ -> collectThreads tid) [1..5]+       tids <- atomically $ readTVar tid+       (length $ nubOrd tids) `shouldSatisfy` (<= 5)++  describe "replicate concurrencyN" $ do+    it "Throws exception properly" $ do+       (pooledReplicateConcurrentlyN 5 1 (exAction 2)) `shouldThrow` poolException++    it "total thread should be >= 1" $ do+       (pooledReplicateConcurrentlyN 0 1 (action 1)) `shouldThrow` anyErrorCall++    it "Read tvar value should be 100" $ do+       var <- myVar+       xs <- (pooledReplicateConcurrentlyN 5 5 (maxTVar 100 var))+       newVar <- atomically $ readTVar var+       atomically $ writeTVar var 0+       newVar `shouldBe` 100++    it "should not spawn more than five threads for five concurrent tasks" $ do+       xs <- (pooledReplicateConcurrentlyN 5 5 (action 1))+       (length $ nubOrd xs) `shouldSatisfy` (<= (5 :: Int))++    it "should not spawn more than three threads for five concurrent tasks" $ do+       xs <- (pooledReplicateConcurrentlyN 3 5 (action 1))+       (length $ nubOrd xs) `shouldSatisfy` (<= (3 :: Int))++    it "should spawn only one thread" $ do+       xs <- (pooledReplicateConcurrentlyN 1 5 (action 1))+       (length $ nubOrd xs) `shouldBe` 1++    it "should give empty list" $ do+       xs <- (pooledReplicateConcurrentlyN 3 0 (action 1))+       xs `shouldBe` []++    it "should give empty list for -ve count" $ do+       xs <- (pooledReplicateConcurrentlyN 3 (-3) (action 1))+       xs `shouldBe` []++  describe "pooled replicateConcurrencyN_" $ do+    it "Throws exception properly" $ do+       (pooledReplicateConcurrentlyN_ 5 1 (exAction 2)) `shouldThrow` poolException++    it "total thread should be >= 1" $ do+       (pooledReplicateConcurrentlyN_ 0 2 (action 1)) `shouldThrow` anyErrorCall++    it "find proper maximum value" $ do+       var <- myVar+       pooledReplicateConcurrentlyN_ 5 3 (maxTVar 200 var)+       newVar <- atomically $ readTVar var+       atomically $ writeTVar var 0+       newVar `shouldBe` 200++    it "Should be initial value" $ do+       var <- myVar+       pooledReplicateConcurrentlyN_ 5 (-2) (maxTVar 200 var)+       newVar <- atomically $ readTVar var+       atomically $ writeTVar var 0+       newVar `shouldBe` 0

@@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.30.0.+-- This file has been generated from package.yaml by hpack version 0.31.1. -- -- see: https://github.com/sol/hpack ----- hash: 6ffa5ab703752662320163d252f2b4f425c0b85a390c41b8c9e74b15fbf93936+-- hash: 806905923ede98dbf20fe24ac95d14b92b4a62626183173c24b86850dc5beada  name:           unliftio-version:        0.2.9.0+version:        0.2.10 synopsis:       The MonadUnliftIO typeclass for unlifting monads to IO (batteries included) description:    Please see the documentation and README at <https://www.stackage.org/package/unliftio> category:       Control@@ -86,6 +86,7 @@       QuickCheck     , async >2.1.1     , base >=4.7 && <5+    , containers     , deepseq     , directory     , filepath@@ -106,6 +107,7 @@       UnliftIO.ExceptionSpec       UnliftIO.IOSpec       UnliftIO.MemoizeSpec+      UnliftIO.PooledAsyncSpec       Paths_unliftio   default-language: Haskell2010