async-pool 0.9.0 → 0.9.0.1

raw patch · 3 files changed

+53/−16 lines, 3 filesdep ~basePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• Control/Concurrent/Async/Pool/Async.hs +32/−7
• Control/Concurrent/Async/Pool/Internal.hs +19/−7
• async-pool.cabal +2/−2

@@ -1,4 +1,4 @@-{-# LANGUAGE CPP, MagicHash, UnboxedTuples, RankNTypes #-}+{-# LANGUAGE CPP, MagicHash, UnboxedTuples, RankNTypes, GADTs #-} #if __GLASGOW_HASKELL__ >= 701 {-# LANGUAGE Trustworthy #-} #endif@@ -107,10 +107,21 @@  -- | A 'Handle' is a unique identifier for a task submitted to a 'Pool'. type Handle    = Node-data State     = Ready | Starting | Started ThreadId deriving (Eq, Show)+data State     = Ready | Starting | forall a. Started ThreadId (TMVar a) data Status    = Pending | Completed deriving (Eq, Show) type TaskGraph = Gr (TVar State) Status +instance Eq State where+    Ready        == Ready        = True+    Starting     == Starting     = True+    Started n1 _ == Started n2 _ = n1 == n2+    _            == _            = False++instance Show State where+    show Ready         = "Ready"+    show Starting      = "Starting"+    show (Started n _) = "Started " ++ show n+ -- | A 'Pool' manages a collection of possibly interdependent tasks, such that --   tasks await execution until the tasks they depend on have finished (and --   tasks may depend on an arbitrary number of other tasks), while@@ -141,11 +152,25 @@       -- ^ Tokens identify tasks, and are provisioned monotonically.     } +waitTMVar :: TMVar a -> STM ()+waitTMVar tv = do+    _ <- readTMVar tv+    return ()++syncPool :: Pool -> STM ()+syncPool p = do+    g <- readTVar (tasks p)+    forM_ (labNodes g) $ \(_h, st) -> do+        x <- readTVar st+        case x of+            Started _tid v -> waitTMVar v+            _ -> retry+ data TaskGroup = TaskGroup     { pool    :: Pool     , avail   :: TVar Int       -- ^ The number of available execution slots in the pool.-    , pending :: TVar (IntMap (IO ThreadId))+    , pending :: forall a. TVar (IntMap (IO ThreadId, TMVar a))       -- ^ Nodes in the task graph that are waiting to start.     } @@ -171,9 +196,9 @@ getThreadId g h = do     status <- readTVar (getTaskVar g h)     case status of-        Ready     -> return Nothing-        Starting  -> retry-        Started x -> return $ Just x+        Ready       -> return Nothing+        Starting    -> retry+        Started x _ -> return $ Just x  instance Eq (Async a) where   Async _ a _ == Async _ b _  =  a == b@@ -218,7 +243,7 @@             doFork $ try (restore (action `finally` cleanup h))                 >>= atomically . putTMVar var -    modifyTVar (pending p) (IntMap.insert h start)+    modifyTVar (pending p) (IntMap.insert h (start, var))     tv <- newTVar Ready     modifyTVar (tasks (pool p)) (insNode (h, tv)) 

@@ -23,10 +23,11 @@ import           Data.Monoid (Monoid(mempty), (<>)) import           Data.Traversable (Traversable(sequenceA), forM) import           Prelude hiding (mapM_, mapM, foldr, all, any, concatMap, foldl1)+import           Unsafe.Coerce  -- | Return a list of actions ready for execution, by checking the graph to --   ensure all dependencies have completed.-getReadyNodes :: TaskGroup -> TaskGraph -> STM (IntMap (IO ThreadId))+getReadyNodes :: TaskGroup -> TaskGraph -> STM (IntMap (IO ThreadId, TMVar a)) getReadyNodes p g = do     availSlots <- readTVar (avail p)     check (availSlots > 0)@@ -49,7 +50,7 @@  -- | Return a list of tasks ready to execute, and their related state --   variables from the dependency graph.-getReadyTasks :: TaskGroup -> STM [(TVar State, IO ThreadId)]+getReadyTasks :: TaskGroup -> STM [(TVar State, (IO ThreadId, TMVar a))] getReadyTasks p = do     g <- readTVar (tasks (pool p))     map (first (getTaskVar g)) . IntMap.toList <$> getReadyNodes p g@@ -60,12 +61,23 @@ createPool = Pool <$> newTVarIO Gr.empty                   <*> newTVarIO 0 +-- | Use a task pool for a bounded region. At the end of the region,+-- 'withPool' will block until all tasks have completed.+withPool :: (Pool -> IO a) -> IO a+withPool f = do+    p <- createPool+    x <- f p+    atomically $ syncPool p+    return x+ -- | Create a task group for executing interdependent tasks concurrently.  The --   number of available slots governs how many tasks may run at one time. createTaskGroup :: Pool -> Int -> IO TaskGroup-createTaskGroup p cnt = TaskGroup <$> pure p-                                  <*> newTVarIO cnt-                                  <*> newTVarIO mempty+createTaskGroup p cnt = do+    c <- newTVarIO cnt+    m <- newTVarIO IntMap.empty+    -- Prior to GHC 8, this call to unsafeCoerce was not necessary.+    return $ TaskGroup p c (unsafeCoerce m)  -- | Execute tasks in a given task group.  The number of slots determines how --   many threads may execute concurrently.@@ -78,9 +90,9 @@         check (not (null ready))         forM_ ready $ \(tv, _) -> writeTVar tv Starting         return ready-    forM_ ready $ \(tv, go) -> do+    forM_ ready $ \(tv, (go, var)) -> do         t <- go-        atomically $ swapTVar tv $ Started t+        atomically $ swapTVar tv $ Started t var  -- | Create a task group within the given pool having a specified number of --   execution slots, but with a bounded lifetime.  Leaving the block cancels

@@ -1,5 +1,5 @@ Name:           async-pool-Version:        0.9.0+Version:        0.9.0.1 Synopsis:       A modified version of async that supports worker groups and many-to-many task dependencies License-file:   LICENSE License:        MIT@@ -20,7 +20,7 @@     default-language: Haskell98     ghc-options:      -Wall     build-depends:-        base                 >= 3 && < 5+        base >= 3 && < 4.10       , fgl       , async       , stm