priority-sync (empty) → 0.1.0.0
raw patch · 10 files changed
+961/−0 lines, 10 filesdep +basedep +containersdep +heapsetup-changed
Dependencies added: base, containers, heap, parallel, random, stm
Files
- Control/Concurrent/Priority/Queue.hs +248/−0
- Control/Concurrent/Priority/Room.hs +124/−0
- Control/Concurrent/Priority/RoomConstraint.hs +48/−0
- Control/Concurrent/Priority/RoomCore.hs +121/−0
- Control/Concurrent/Priority/Schedule.hs +55/−0
- Control/Concurrent/Priority/TaskPool.hs +78/−0
- LICENSE +26/−0
- Setup.hs +5/−0
- Tests.hs +176/−0
- priority-sync.cabal +80/−0
+ Control/Concurrent/Priority/Queue.hs view
@@ -0,0 +1,248 @@+module Control.Concurrent.Priority.Queue+ (Queue,+ TaskHandle,+ QueueOrder(..),+ QueueConfigurationRecord(..),+ fair_queue_configuration, fast_queue_configuration,+ newQueue,+ taskPriority,+ taskQueue,+ pendingTasks,+ isTopOfQueue,+ hasCompleted,+ putTask,+ pullTask,+ pullFromTop,+ pullSpecificTasks,+ dispatchTasks,+ flushQueue,+ load)+ where++import Data.Heap as Heap+import Data.List as List (sort,sortBy,groupBy,drop)+import GHC.Conc+import Control.Monad+import Data.Unique+import Data.Ord+import Data.Maybe++-- | A prioritized 'Queue'. Prioritization is least-first, i.e. larger values are nicer.+--+-- A 'Queue' is not associated with any working thread, therefore, it is the client\'s responsibility to make sure that every pushed+-- task is also pulled, or the 'Queue' will stall. There are several ways to accomplish this:+--+-- * Call 'pullTask' at least once for every call to 'putTask'.+--+-- * Use 'dispatchTasks' to push every task.+--+-- * Use 'flushQueue' whenever the 'Queue' is not empty.+data (Ord a) => Queue a = Queue {+ queue_configuration :: !(QueueConfigurationRecord a),+ queue_unique :: Unique,+ pending_tasks :: TVar (MinHeap (TaskHandle a)),+ task_counter :: TVar Integer }++data QueueOrder = FIFO | FILO++-- | Configuration options for a 'Queue'. A 'Queue' blocks on a number of predicates when dispatching a job. Generally, 'fair_queue_configuration'+-- should work well for long-running batch jobs and 'fast_queue_configuration' should work for rapid paced jobs.+--+-- * A single STM predicate for the entire 'Queue'. This blocks the entire 'Queue' until the predicate is satisfied.+--+-- * A STM predicate parameterized by priority. This blocks a single priority level, and the 'Queue' will skip all tasks at that priority.+--+-- * Each task is itself an STM transaction, and can block itself.+--+-- * Pure constraints on priority and ordering inversion.+--+-- If a task is blocked for any reason, the task is skipped and the next task attempted, in priority order.++data (Ord a) => QueueConfigurationRecord a = QueueConfigurationRecord {+ -- | A predicate that must hold before any task may be pulled from a 'Queue'.+ queue_predicate :: STM (),+ -- | A predicate that must hold before any priority level may be pulled from a 'Queue'.+ priority_indexed_predicate :: (a -> STM ()),+ -- | Constrains the greatest allowed difference between the priority of the top-of-queue task and the priority of a task to be pulled.+ allowed_priority_inversion :: a -> a -> Bool,+ -- | The greatest allowed difference between the ideal prioritized FILO/FIFO ordering of tasks and the actual ordering of tasks.+ -- Setting this too high can introduce a lot of overhead in the presence of a lot of short-running tasks.+ -- Setting this to zero turns off the predicate failover feature, i.e. only the top of queue task will ever be pulled.+ allowed_ordering_inversion :: Int,+ -- | Should the 'Queue' run in FILO or FIFO order. Ordering takes place after prioritization, and won't have much effect if priorities are very fine-grained.+ queue_order :: !QueueOrder }++-- | A queue tuned for high throughput and fairness when processing moderate to long running tasks.+fair_queue_configuration :: (Ord a) => QueueConfigurationRecord a+fair_queue_configuration = QueueConfigurationRecord {+ queue_predicate = return (),+ priority_indexed_predicate = const $ return (),+ allowed_priority_inversion = const $ const $ True,+ allowed_ordering_inversion = numCapabilities*5,+ queue_order = FIFO }++-- | A queue tuned for high responsiveness and low priority inversion, but may have poorer long-term throughput and potential to starve some tasks compared to 'fair_queue_configuration'.+fast_queue_configuration :: (Ord a) => QueueConfigurationRecord a+fast_queue_configuration = fair_queue_configuration {+ allowed_priority_inversion = (==),+ allowed_ordering_inversion = numCapabilities,+ queue_order = FILO }++instance (Ord a) => Eq (Queue a) where+ (==) l r = queue_unique l == queue_unique r++instance (Ord a) => Ord (Queue a) where+ compare l r = compare (queue_unique l) (queue_unique r)++data TaskHandle a = TaskHandle {+ task_action :: STM (),+ is_top_of_queue :: TVar Bool,+ has_completed :: TVar Bool,+ task_counter_index :: !Integer,+ task_priority :: !a,+ task_queue :: Queue a }++instance (Ord a,Eq a) => Eq (TaskHandle a) where+ (==) l r = (==) (taskOrd l) (taskOrd r)++instance (Ord a) => Ord (TaskHandle a) where+ compare l r = compare (taskOrd l) (taskOrd r)++taskOrd :: TaskHandle a -> (a,Integer,Queue a)+taskOrd t = (task_priority t,task_counter_index t,task_queue t)++-- | True iff this task is poised at the top of it's 'Queue'.+isTopOfQueue :: TaskHandle a -> STM Bool+isTopOfQueue task = readTVar (is_top_of_queue task)++hasCompleted :: TaskHandle a -> STM Bool+hasCompleted task = readTVar (has_completed task)++taskPriority :: TaskHandle a -> a+taskPriority = task_priority++taskQueue :: TaskHandle a -> Queue a+taskQueue = task_queue++pendingTasks :: (Ord a) => Queue a -> STM [TaskHandle a]+pendingTasks = liftM Heap.toList . readTVar . pending_tasks++-- | Create a new 'Queue'. +newQueue :: (Ord a) => QueueConfigurationRecord a -> IO (Queue a)+newQueue config = + do pending_tasks_var <- newTVarIO empty+ counter <- newTVarIO 0+ uniq <- newUnique+ return Queue {+ queue_configuration = config,+ queue_unique = uniq,+ pending_tasks = pending_tasks_var,+ task_counter = counter }++-- | Put a task with it's priority value onto this queue. Returns a handle to the task.+putTask :: (Ord a) => Queue a -> a -> STM () -> STM (TaskHandle a)+putTask q prio actionSTM = + do count <- readTVar (task_counter q)+ writeTVar (task_counter q) $ (case (queue_order $ queue_configuration q) of FIFO -> (+ 1); FILO -> (subtract 1)) count+ false_top_of_queue <- newTVar False+ false_has_completed <- newTVar False+ let task = TaskHandle {+ task_action = actionSTM,+ is_top_of_queue = false_top_of_queue,+ has_completed = false_has_completed,+ task_counter_index = count,+ task_priority = prio,+ task_queue = q }+ watchingTopOfQueue q $ writeTVar (pending_tasks q) . insert task =<< readTVar (pending_tasks q)+ return task++-- | The number of tasks pending on this Queue.+load :: (Ord a) => Queue a -> STM Int +load q = liftM size $ readTVar (pending_tasks q)++-- | Pull and commit a task from this 'Queue'.+pullTask :: (Ord a) => Queue a -> STM (TaskHandle a)+pullTask q = watchingTopOfQueue q $ + do queue_predicate $ queue_configuration q+ (task,rest) <- pullTask_ (queue_configuration q) empty =<< readTVar (pending_tasks q)+ writeTVar (pending_tasks q) rest+ writeTVar (has_completed task) True+ return task++pullTask_ :: (Ord a) => QueueConfigurationRecord a -> MinHeap (TaskHandle a) -> MinHeap (TaskHandle a) -> STM (TaskHandle a,MinHeap (TaskHandle a))+pullTask_ config faltered_tasks untried_tasks =+ do when (Heap.size faltered_tasks > allowed_ordering_inversion config) retry+ (task,rest) <- maybe retry return $ view untried_tasks+ let top_prio = taskPriority $ maybe task fst $ view $ faltered_tasks+ unless (allowed_priority_inversion config top_prio (taskPriority task)) retry+ let predicateFailed = do let (same_prios,remaining_prios) = Heap.span ((== (task_priority task)) . task_priority) rest+ pullTask_ config (insert task faltered_tasks `union` fromList same_prios) remaining_prios+ let taskFailed = do pullTask_ config (insert task faltered_tasks) rest+ prio_ok <- ((priority_indexed_predicate config $ task_priority task) >> return True) `orElse` (return False)+ case prio_ok of+ False -> predicateFailed+ True -> (task_action task >> return (task,faltered_tasks `union` rest)) `orElse` taskFailed++-- | Pull this task from the top of a 'Queue', if it is already there.+-- If this task is top-of-queue, but it's predicates fail, then 'pullFromTop' may instead pull a lower-priority 'TaskHandle'.+pullFromTop :: (Ord a) => TaskHandle a -> STM (TaskHandle a)+pullFromTop task = + do b <- hasCompleted task+ if b then return task else+ do flip unless retry =<< isTopOfQueue task+ pullTask (taskQueue task)++-- | Don't return until the given 'TaskHandle' has been pulled from its associated 'Queue'.+-- This doesn't guarantee that the 'TaskHandle' will ever be pulled, even when the 'TaskHandle' and 'Queue' are both viable.+-- You must concurrently arrange for every other 'TaskHandle' associated with the same 'Queue' to be pulled, or the 'Queue' will stall.+pullSpecificTask :: (Ord a) => TaskHandle a -> IO ()+pullSpecificTask task =+ do actual_task <- atomically $ pullFromTop task+ unless (actual_task == task) $ pullSpecificTask task++-- | Don't return until the given 'TaskHandle's have been pulled from their associated 'Queue's.+-- This doesn't guarantee that the 'TaskHandle' will ever be pulled, even when the 'TaskHandle' and 'Queue' are both viable.+-- You must concurrently arrange for every other 'TaskHandle' associated with the same 'Queue' to be pulled, or the 'Queue' will stall.+-- 'pullSpecificTasks' can handle lists 'TaskHandle's that are distributed among several 'Queue's, as well as a 'TaskHandle's that have+-- already completed or complete concurrently from another thread.+pullSpecificTasks :: (Ord a) => [TaskHandle a] -> IO ()+pullSpecificTasks tasks =+ do queue_groups <- mapM (\g -> liftM ((,) g) newEmptyMVar) $ map sort $ groupBy (\x y -> taskQueue x == taskQueue y) $ sortBy (comparing taskQueue) tasks+ let pullTaskGroup (g,m) = mapM pullSpecificTask g >> putMVar m ()+ mapM (forkIO . pullTaskGroup) (List.drop 1 queue_groups)+ maybe (return ()) pullTaskGroup $ listToMaybe queue_groups+ mapM_ (takeMVar . snd) queue_groups++-- | \"Fire and forget\" some tasks on a separate thread.+dispatchTasks :: (Ord a) => [(Queue a,a,STM ())] -> IO [TaskHandle a]+dispatchTasks task_records = + do tasks <- mapM (\(q,a,actionSTM) -> atomically $ putTask q a actionSTM) task_records+ forkIO $ pullSpecificTasks tasks+ return tasks++-- | Process a 'Queue' until it is empty.+flushQueue :: (Ord a) => Queue a -> IO ()+flushQueue q =+ do want_zero <- atomically $ + do l <- load q+ when (l > 0) $ pullTask q >> return ()+ return l+ unless (want_zero == 0) $ flushQueue q++setTopOfQueue :: (Ord a) => Queue a -> Bool -> STM Bool+setTopOfQueue q t =+ do m_view <- liftM view $ readTVar (pending_tasks q)+ case m_view of+ Nothing -> return True+ Just (task,_) -> + do previous_t <- readTVar (is_top_of_queue task)+ writeTVar (is_top_of_queue task) t+ return previous_t++watchingTopOfQueue :: (Ord a) => Queue a -> STM b -> STM b+watchingTopOfQueue q actionSTM =+ do should_be_true <- setTopOfQueue q False+ unless should_be_true $ error "watchingTopOfQueue: not reentrant"+ result <- actionSTM+ setTopOfQueue q True+ return result
+ Control/Concurrent/Priority/Room.hs view
@@ -0,0 +1,124 @@+{-# LANGUAGE MultiParamTypeClasses, TypeFamilies, FlexibleInstances, FlexibleContexts, UndecidableInstances #-}++module Control.Concurrent.Priority.Room+ (Room,+ newRoom,+ inUse,+ Claim,+ claimedRoom,+ claimedThread,+ userData,+ UserData,+ RoomGroup(..),+ RoomConstraint(..),+ BaseRoomContext(..),+ RoomContext(..),+ MaxThreads(..),+ ClaimMode(..),+ DefaultRoomContext(..),+ UnconstrainedRoomContext(..),+ claim,+ approveClaims)+ where++import Control.Concurrent.Priority.RoomCore as RoomCore+import Control.Concurrent.Priority.RoomConstraint+import Control.Concurrent.STM+import Control.Monad+import Data.Map as Map+import Data.List as List++-- | Require that all 'RoomConstraint's be satisfied when acquiring a 'Room'. This is the default.+data DefaultRoomContext u = Default++-- | Don't check any 'RoomConstraint's when acquiring a 'Room'.+data UnconstrainedRoomContext u = Unconstrained++type family UserData u :: *++type instance UserData (Room u) = u+type instance UserData [Room u] = u+type instance UserData (DefaultRoomContext u) = u+type instance UserData (UnconstrainedRoomContext u) = u+type instance UserData (c,m) = UserData c++class RoomGroup m where+ roomsOf :: m -> [Room (UserData m)]++instance RoomGroup (Room u) where+ roomsOf m = [m]++instance RoomGroup [Room u] where+ roomsOf = id++instance RoomGroup (DefaultRoomContext u) where+ roomsOf = const []++instance RoomGroup (UnconstrainedRoomContext u) where+ roomsOf = const []++instance (UserData c ~ UserData m,RoomGroup c,RoomGroup m) => RoomGroup (c,m) where+ roomsOf (c,m) = roomsOf c ++ roomsOf m++-- | Rules for calling 'claim_'. The two major contexts are 'DefaultRoomContext', which uses 'RoomConstraint's to+-- determine which 'Room's are available, and 'UnconstrainedRoomContext', which does not place any constraints on any 'Room'.+class BaseRoomContext c where+ type BaseRoomContextData c :: *+ -- | Should approve a some claims before entering a critical section, as described by 'claim_'.+ approveClaimsEntering :: c -> [Claim (UserData c)] -> STM (BaseRoomContextData c)+ -- | Should approve a some claims before exiting a critical section, as described by 'claim_'.+ approveClaimsExiting :: c -> [Claim (UserData c)] -> STM (BaseRoomContextData c)+ -- | A waiting transaction, as described by 'claim_'.+ waitingAction :: c -> (BaseRoomContextData c) -> STM ()++instance (RoomConstraint u) => BaseRoomContext (DefaultRoomContext u) where+ type BaseRoomContextData (DefaultRoomContext u) = ()+ approveClaimsEntering _ cs = approveClaims cs >> return ()+ approveClaimsExiting _ cs = approveClaims cs >> return ()+ waitingAction _ () = return ()++instance BaseRoomContext (UnconstrainedRoomContext u) where+ type BaseRoomContextData (UnconstrainedRoomContext u) = ()+ approveClaimsEntering _ cs = mapM_ approve cs >> return ()+ approveClaimsExiting _ cs = mapM_ approve cs >> return ()+ waitingAction _ _ = return ()++instance (BaseRoomContext c,Base m ~ DefaultRoomContext (UserData m)) => BaseRoomContext (c,m) where+ type BaseRoomContextData (c,m) = BaseRoomContextData c+ approveClaimsEntering = approveClaimsEntering . fst+ approveClaimsExiting = approveClaimsExiting . fst+ waitingAction = waitingAction . fst++-- | An indirect reference to a 'BaseRoomContext'.+class RoomContext c where+ type Base c :: *+ baseContext :: c -> Base c++instance (RoomConstraint u) => RoomContext (Room u) where+ type Base (Room u) = DefaultRoomContext u+ baseContext = const Default++instance (RoomConstraint u) => RoomContext [Room u] where+ type Base [Room u] = DefaultRoomContext u+ baseContext = const Default++instance (BaseRoomContext c,Base m ~ DefaultRoomContext (UserData m)) => RoomContext (c,m) where+ type Base (c,m) = c+ baseContext = fst++-- | Temporarily 'Acquire', and then release, or 'Release', and then acquire, some 'Room's for the duration of a critical section.+-- A simple example where a room might be used to prevent interleaving of 'stdout':+--+-- > room <- newRoom (MaxThreads 1)+-- > forkIO $ claim Acquire room $ putStrLn "Hello World!"+-- > forkIO $ claim Acquire room $ putStrLn "Foo! Bar!"+claim :: (RoomGroup c,RoomContext c,BaseRoomContext (Base c),UserData c ~ UserData (Base c)) => ClaimMode -> c -> IO a -> IO a+claim claim_mode c actionIO = + do let c' = baseContext c+ room_context_data <- newTVarIO (error "claim: BaseRoomContextData not yet available (please report a bug against the priority package)")+ claim_ (Map.fromList $ Prelude.map (flip (,) claim_mode) $ roomsOf c) + (\cs -> writeTVar room_context_data =<< approveClaimsEntering c' cs) + (\cs -> writeTVar room_context_data =<< approveClaimsExiting c' cs)+ (waitingAction c' =<< readTVar room_context_data)+ actionIO+
+ Control/Concurrent/Priority/RoomConstraint.hs view
@@ -0,0 +1,48 @@+module Control.Concurrent.Priority.RoomConstraint+ (RoomConstraint(..),+ MaxThreads(..),+ approveClaims)+ where++import Control.Concurrent.Priority.RoomCore+import Control.Concurrent.STM+import Control.Monad+import Data.Set as Set++class RoomConstraint u where+ -- | Should either 'approve' or 'retry' each claim.+ approveConstraint :: Claim a -> u -> STM ()++instance RoomConstraint () where+ approveConstraint c () = approve c++instance RoomConstraint Bool where -- this is pointless but means we support RoomConstraint (STM Bool)+ approveConstraint c True = approve c+ approveConstraint _ False = retry++-- | A maximum limit on the number of threads allowed to claim a room.+newtype MaxThreads = MaxThreads Int++instance RoomConstraint MaxThreads where+ approveConstraint c (MaxThreads n) =+ do s <- liftM (Set.size . Set.insert (claimedThread c)) $ inUse $ claimedRoom c+ approveConstraint c $ s <= n++instance (RoomConstraint u) => RoomConstraint (STM u) where+ approveConstraint c actionSTM = approveConstraint c =<< actionSTM++instance (RoomConstraint a,RoomConstraint b) => RoomConstraint (a,b) where+ approveConstraint c (a,b) =+ do approveConstraint c a+ approveConstraint c b++instance (RoomConstraint a,RoomConstraint b) => RoomConstraint (Either a b) where+ approveConstraint c = either (approveConstraint c) (approveConstraint c)++instance (RoomConstraint a) => RoomConstraint (Maybe a) where+ approveConstraint c = maybe (approveConstraint c ()) $ approveConstraint c++-- | 'approve' some claims according to their constraints.+approveClaims :: (RoomConstraint u) => [Claim u] -> STM ()+approveClaims = mapM_ (\c -> approveConstraint c $ userData $ claimedRoom c)+
+ Control/Concurrent/Priority/RoomCore.hs view
@@ -0,0 +1,121 @@+module Control.Concurrent.Priority.RoomCore+ (Room,+ newRoom,+ Claim,+ ClaimMode(..),+ claimedRoom,+ claimedThread,+ userData,+ approve,+ claim_,+ inUse)+ where++import Data.Unique+import Data.Set as Set+import Data.Map as Map+import GHC.Conc+import Control.Monad+import Control.Exception++-- | A resource pool, parameterized against arbitrary user data.+data Room u = Room (u,Unique) (TVar (Set ThreadId))++-- | A 'Claim', or attempt to acquire or release a 'Room'.+data Claim u = Claim (Room u) ThreadId ClaimMode (TVar Bool)++data ClaimMode = Acquire | Release deriving (Eq)++instance Eq (Room u) where+ (==) (Room u1 _) (Room u2 _) = snd u1 == snd u2++instance Ord (Room u) where+ compare (Room u1 _) (Room u2 _) = compare (snd u1) (snd u2)++-- | Create a new Room with some arbitrary user data.+newRoom :: u -> IO (Room u)+newRoom u = return Room `ap` liftM ((,) u) newUnique `ap` atomically (newTVar Set.empty)++-- | Get the user data associated with a 'Room'.+userData :: Room u -> u+userData (Room (u,_) _) = u++-- | Whether a Claim is to acquire or release a room.+claimMode :: Claim u -> ClaimMode+claimMode (Claim _ _ b _) = b++-- | Get the 'Room' target of a 'Claim'.+claimedRoom :: Claim u -> Room u+claimedRoom (Claim m _ _ _) = m++-- | Get the thread attempting a 'Claim'.+claimedThread :: Claim u -> ThreadId+claimedThread (Claim _ t _ _) = t++-- | Approve a claim. This actually acquires a 'Room'.+approve :: Claim u -> STM ()+approve (Claim (Room _ m) me want claim_var) =+ do claim_state <- readTVar claim_var+ when (claim_state == False) $ + do writeTVar claim_var True+ writeTVar m . (case want of Acquire -> Set.insert; Release -> Set.delete) me =<< readTVar m++-- | Acquire and/or release some rooms for the duration of a critical section.+--+-- * which 'Room's to 'Acquire', and later release, or 'Release' and later reacquire for the duration of the critical section.+--+-- * a transaction to 'approve' all entering 'Claim's+--+-- * a transaction to 'approve' all exiting 'Claim's+--+-- * a transaction to run one or more times if and only if this thread is waiting for approval+--+-- * the critical section+--+-- A separate 'Claim' is generated each time a Room needs to be acquired. The critical+-- section will not enter until every claim has been 'approve'd.+--+-- When the critical section exits, an inverse group of claims will be generated, and the critical+-- section will not exit until those claims have been 'approve'd.+--+-- It is guaranteed that when and only when all 'Claim's have been 'approve'd, the waiting thread will enter+-- (or exit) the critical section. The lock on each 'Room' is acquired when it's 'Claim' is 'approve'd,+-- not when the critical section is entered.+--+-- 'Claim's may be 'approve'd from any transaction, even from another thread.+--+claim_ :: Map (Room u) ClaimMode -> ([Claim u] -> STM ()) -> ([Claim u] -> STM ()) -> STM () -> IO a -> IO a+claim_ entering_rooms_map approveEnteringSTM approveExitingSTM waitingSTM actionIO =+ do let entering_rooms = Map.toList entering_rooms_map+ me <- myThreadId+ -- transition: generate and request (but do not wait for) approval for all claims+ let transition rooms approveSTM = atomically $+ (\claims -> approveSTM (Prelude.filter ((== Acquire) . claimMode) claims) >> return claims) =<< -- request approval of Acquire claims.+ (mapM $ \c -> when (claimMode c == Release) (approve c) >> return c) =<< --auto-approve Release claims.+ (mapM $ \(m,want) -> liftM (Claim m me want) (newTVar False)) =<< -- build claims+ filterM (\(m,want) -> liftM ((/= (want == Acquire)) . Set.member me) $ inUse m) rooms -- get the difference between the rooms we want and the rooms we have+ -- confirm: wait for all claims to be approved+ let confirm claims = forM_ claims $ \(Claim _ _ _ claim_var) ->+ do claim_state <- readTVar claim_var+ unless claim_state retry+ -- confirm or perform the user's wait action, return True iff we have confirmed+ let confirmWithWaitAction claims=+ do done <- atomically $ (confirm claims >> return True) `orElse` (waitingSTM >> return False)+ unless done $ confirmWithWaitAction claims+ -- when entering we signal the claims (transition) and then wait on the approval of the claims (confirm)+ -- then we generate a list of inverse claims for when it comes time to exit the critical section+ -- If an exception is thrown from the approve*STM or waitingSTM, this would leave dangling + -- room locks, therefore, we force the rooms into their former state, ignoring any + -- constraints that might have been placed on them. On the theory that it's better to violate + -- constraints than to leave dangling locks.+ let transitionAndConfirm approveSTM rooms = flip finally (transition rooms $ mapM_ approve) $+ do claims <- transition rooms approveSTM+ confirmWithWaitAction claims+ return $ Prelude.map (\(Claim m _ want _) -> (m,case want of Acquire -> Release; Release -> Acquire)) claims+ bracket (transitionAndConfirm approveEnteringSTM entering_rooms)+ (transitionAndConfirm approveExitingSTM)+ (const actionIO)++-- | Get all 'ThreadId's that are currently claimimg this 'Room'.+inUse :: Room u -> STM (Set ThreadId)+inUse (Room _ m) = readTVar m
+ Control/Concurrent/Priority/Schedule.hs view
@@ -0,0 +1,55 @@+{-# LANGUAGE TypeFamilies, UndecidableInstances #-}++module Control.Concurrent.Priority.Schedule+ (Schedule(..))+ where++import Control.Concurrent.Priority.Room+import Control.Concurrent.Priority.Queue+import Control.Concurrent.STM+import Control.Monad+import Data.List++-- | Schedule a task to run from a prioritized 'Queue'.+--+-- Tasks that do not actually make claims against any of the 'Schedule's internal 'Room's will skip scheduling and the 'Room's will be claimed immediately using 'DefaultRoomContext'. This is usually+-- what you want, in particular in the case where no rooms are actually being claimed, e.g. reentrant scheduling.+--+-- In other words:+--+-- Always wrong:+--+-- > (Schedule q 2 Default,[room1,room2])+--+-- Right:+--+-- > Schedule q 2 (Default,[room1,room2])+--+-- Alternately, if you only want to schedule access to @room1@, you can place @room1@ internally and @room2@ externally. 'Schedule' will be smart about when to schedule and when not to schedule:+--+-- > (Schedule q 2 (Default,room1), room2)+--+-- The 'Default' applies internally and externally to the 'Schedule'. In the following example, 'Unconstrained' applies to both @room1@ and @room2@:+--+-- > (Schedule q 2 (Unconstrained,room1), room2)+data Schedule p c = Schedule (Queue p) p c++type instance UserData (Schedule p c) = UserData c++instance (RoomGroup c) => RoomGroup (Schedule p c) where+ roomsOf (Schedule _ _ c) = roomsOf c++instance (Ord p,RoomGroup c,BaseRoomContext c,BaseRoomContextData c ~ ()) => BaseRoomContext (Schedule p c) where+ type BaseRoomContextData (Schedule p c) = Maybe (TaskHandle p)+ approveClaimsEntering = scheduleClaims approveClaimsEntering+ approveClaimsExiting = scheduleClaims approveClaimsExiting+ waitingAction (Schedule _ _ c) Nothing = waitingAction c ()+ waitingAction (Schedule _ _ c) (Just task) = flip unless retry . or =<< mapM (\m -> m >> return True `orElse` return False) [pullFromTop task >> return (), waitingAction c ()]++scheduleClaims :: (Ord p,RoomGroup c,BaseRoomContext c,BaseRoomContextData c ~ ()) => (c -> [Claim (UserData c)] -> STM ()) -> Schedule p c -> [Claim (UserData c)] -> STM (Maybe (TaskHandle p))+scheduleClaims approveClaimsX (Schedule _ _ c) cs | null (intersect (map claimedRoom cs) $ roomsOf c) = approveClaimsX c cs >> return Nothing+scheduleClaims approveClaimsX (Schedule q p c) cs = liftM Just $ putTask q p (approveClaimsX c cs)+ +instance (BaseRoomContext (Schedule p c)) => RoomContext (Schedule p c) where+ type Base (Schedule p c) = Schedule p c+ baseContext = id
+ Control/Concurrent/Priority/TaskPool.hs view
@@ -0,0 +1,78 @@+{-# LANGUAGE TypeFamilies, FlexibleInstances #-}++-- | A prioritized TaskPool. This consists of a 'Queue', which prioritizes tasks, and a 'Room' which restricts the number of tasks that may execute at one time.+module Control.Concurrent.Priority.TaskPool+ (TaskPool,+ Control.Concurrent.Priority.TaskPool.schedule,+ newTaskPool,+ simpleTaskPool,+ poolRoom,+ poolQueue,+ startQueue,+ stopQueue,+ activity)+ where++import Control.Concurrent.Priority.Room+import Control.Concurrent.Priority.Queue+import Control.Concurrent.Priority.Schedule+import Control.Monad+import Data.Set as Set+import GHC.Conc++data TaskPool p u = TaskPool {+ pool_on :: TVar Bool,+ pool_queue :: Queue p,+ pool_room :: Room (TaskPoolConstraint u) }++type TaskPoolConstraint u = (Maybe MaxThreads, u)++type instance UserData (TaskPool p u) = TaskPoolConstraint u++instance RoomGroup (TaskPool p u) where+ roomsOf (TaskPool _ _ m) = [m]++instance RoomContext (TaskPool () u) where+ type Base (TaskPool () u) = Schedule () (DefaultRoomContext (TaskPoolConstraint u),Room (TaskPoolConstraint u))+ baseContext tp = schedule tp ()++-- | A 'RoomContext' for a task pool.+schedule :: TaskPool p u -> p -> (Schedule p (DefaultRoomContext (TaskPoolConstraint u),Room (TaskPoolConstraint u)))+schedule (TaskPool _ q m) p = Schedule q p (Default,m)++-- | Create a new 'TaskPool'. 'TaskPool's begin stopped, use 'startQueue' to start.+--+-- * A 'QueueConfigurationRecord' for the backing 'Queue'. A typical value is 'simple_queue_configuration' or 'fast_queue_configuration'.+--+-- * The user data for the backing 'Room'. A typical value is @'MaxThreads' 'GHC.Conc.numCapabilities'@.+--+-- Consider using 'simpleTaskPool' if you have no special needs.+--+newTaskPool :: (Ord p) => QueueConfigurationRecord p -> Int -> u -> IO (TaskPool p u)+newTaskPool config n u =+ do on <- newTVarIO False + m <- newRoom $ (Just $ MaxThreads n,u)+ q <- newQueue $ config { queue_predicate = (flip when retry . not =<< readTVar on) >> + (flip when retry . (>= n) . Set.size =<< inUse m) >> + queue_predicate config }+ return $ TaskPool on q m++-- | Just create a new 'TaskPool'. The task pool is constrained by the number of capabilities indicated by 'GHC.Conc.numCapabilities'.+simpleTaskPool :: (Ord p) => IO (TaskPool p ())+simpleTaskPool = newTaskPool fair_queue_configuration numCapabilities ()++poolRoom :: TaskPool p u -> Room (TaskPoolConstraint u)+poolRoom = pool_room++poolQueue :: TaskPool p u -> Queue p+poolQueue = pool_queue++startQueue :: TaskPool p u -> IO ()+startQueue tp = atomically $ writeTVar (pool_on tp) True++stopQueue :: TaskPool p u -> IO ()+stopQueue tp = atomically $ writeTVar (pool_on tp) False++-- | The number of threads participating in this 'ThreadPool'.+activity :: (Ord p) => TaskPool p u -> STM Int+activity tp = liftM2 (+) (load $ poolQueue tp) (liftM size $ inUse $ poolRoom tp)
+ LICENSE view
@@ -0,0 +1,26 @@+Copyright (c) 2009, Christopher Lane Hinson+All rights reserved.++Redistribution and use in source and binary forms, with or without modification, are +permitted provided that the following conditions are met:++Redistributions of source code must retain the above copyright notice, this list of +conditions and the following disclaimer.++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.++Neither the name of Christopher Lane Hinson nor the names of its contributors may be used +to endorse or promote products derived from this software without specific prior written +permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 +COPYRIGHT OWNER 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.hs view
@@ -0,0 +1,5 @@+#!/usr/bin/runhaskell++import Distribution.Simple++main = defaultMainWithHooks simpleUserHooks
+ Tests.hs view
@@ -0,0 +1,176 @@+{-# LANGUAGE RecursiveDo, ScopedTypeVariables #-}+module Main (main) where++import Control.Concurrent.Priority.Room+import Control.Concurrent.Priority.Queue+import Control.Concurrent.Priority.TaskPool+import Control.Concurrent.MVar+import Control.Monad+import System.Random+import Data.Set as Set+import GHC.Conc+import System.Environment+import System.IO.Unsafe+import System.Exit++{-# NOINLINE fail_strs #-}+fail_strs :: MVar [String]+fail_strs = unsafePerformIO $ newMVar []++failed :: String -> IO ()+failed s = modifyMVar_ fail_strs $ \strs ->+ do putStrLn s+ return $ strs ++ [s]++testRoom :: IO ()+testRoom =+ do putStrLn "testRoom"+ putStrLn "Simple test of room reentrancy."+ m <- newRoom ()+ me <- myThreadId+ let check s b = do ok <- atomically $ liftM ((== b) . member me) $ inUse m+ when (not ok) $ failed $ "testRoom: " ++ s+ check "testRoom-1" False+ claim Acquire [m] $ check "testRoom-2" True+ check "testRoom-3" False+ claim Release [m] $ check "testRoom-4" False+ check "testRoom-5" False+ claim Acquire [m] $ claim Acquire [m] (check "testRoom-6" True) >> check "testRoom-7" True >> claim Release [m] (check "testRoom-8" False)++testMaxThreads :: IO ()+testMaxThreads =+ do putStrLn "testMaxThreads"+ putStrLn "Various threads run in a pair of rooms. The large room has four slots, while the small room has two slots."+ putStrLn $ "12 large, 4 small, 8 large+small, 4 unconstrained that occupy a slot in large and small"+ io_sem <- newMVar ()+ c <- newMVar 0+ let runThread s = do threadDelay 2000000+ modifyMVar_ c (return . (+1))+ withMVar io_sem $ const $ putStrLn s+ large <- newRoom (MaxThreads 4)+ small <- newRoom (MaxThreads 2)+ claim Acquire [large,small] $+ do forM_ [1..8] $ const $ forkIO $ claim Acquire [large,small] $ runThread "large+small"+ forM_ [1..12] $ const $ forkIO $ claim Acquire [large] $ runThread "large"+ forM_ [1..4] $ const $ forkIO $ claim Acquire [small] $ runThread "small"+ forM_ [1..4] $ const $ forkIO $ claim Acquire (Unconstrained,[large,small]) $ runThread "unconstrained occupant (large+small)"+ threadDelay 3000000+ withMVar c $ \x -> when (x < 4) $ failed "testMaxThreads: should have completed at least 4 tasks within 3 seconds"+ withMVar c $ \x -> when (x > 10) $ failed "testMaxThreads: should not have completed more than 10 tasks within 3 seconds"+ withMVar io_sem $ const $ putStrLn "--"+ threadDelay 3000000+ withMVar io_sem $ const $ putStrLn "--"+ threadDelay 3000000+ withMVar io_sem $ const $ putStrLn "--"+ threadDelay 3000000+ withMVar io_sem $ const $ putStrLn "--"+ threadDelay 3000000+ withMVar c $ \x -> when (x /= 28) $ failed "testMaxThreads: did not complete after 15 seconds."++testQueue :: IO ()+testQueue =+ mdo putStrLn "testQueue"+ putStrLn "Perform some tasks in priority order, with constraints enforced at queue-level (to govern input), priority level (priority-1 tasks require small load),"+ putStrLn "and task level (priority-2 tasks only work when the counter is even)."+ need_to_print <- newTVarIO False+ value_to_print <- newTVarIO ""+ q <- newQueue $ fair_queue_configuration {+ queue_predicate = flip when retry =<< readTVar need_to_print, + priority_indexed_predicate = \x -> do l <- load q; if x == 1 && l > 10 then retry else return () }+ counter <- newTVarIO 0+ str <- newTVarIO ""+ let incCounter x s = + do n <- readTVar counter+ writeTVar counter $ 1 + n+ writeTVar need_to_print True+ writeTVar value_to_print (s ++ " " ++ show n)+ writeTVar str . (++ show x) =<< readTVar str+ atomically $ + do forM [1..4] $ const $ putTask q 0 $ incCounter 0 "priority-0"+ forM [1..4] $ const $ putTask q 1 $ incCounter 1 "priority-1, load <= 10"+ forM [1..4] $ const $ putTask q 2 $ + do n <- readTVar counter+ when (n `mod` 2 /= 0) retry+ incCounter 2 "priority-2, counter is even"+ forM [1..4] $ const $ putTask q 3 $ incCounter 3 "priority-3"+ forM_ [1..32] $ const $+ do m_s <- atomically $ (do b <- readTVar need_to_print; if b then liftM Just (readTVar value_to_print) else retry) `orElse` (pullTask q >> return Nothing)+ maybe (return ()) (\s -> putStrLn s >> atomically (writeTVar need_to_print False)) m_s+ ok <- atomically $ liftM (== "0000231111232323") $ readTVar str+ when (not ok) $ failed "testQueue"++testTaskPool :: IO ()+testTaskPool =+ do putStrLn "testTaskPool"+ putStrLn "Threads should complete in priority order over a duration of one and a half seconds after a one second delay."+ putStrLn "Room has two open slots, so order of evaluation may be off by one task."+ pool <- newTaskPool fair_queue_configuration 2 ()+ m_inversions <- newMVar 0+ m_count <- newMVar 0+ m_last_prio <- newMVar 0+ let testPrio n = modifyMVar_ m_last_prio $ \last_prio ->+ do when (last_prio > n) $ modifyMVar_ m_inversions (return . (+1))+ modifyMVar_ m_count (return . (+1))+ return n+ forM_ [1..10] $ const $ forkIO $ claim Acquire (schedule pool 4) $ testPrio 4 >> threadDelay 200000 >> putStrLn "finished-4"+ forM_ [1..4] $ const $ forkIO $ claim Acquire (schedule pool 2) $ testPrio 2 >> threadDelay 200000 >> putStrLn "finished-2"+ forkIO $ claim Acquire (schedule pool 1) $ testPrio 1 >> threadDelay 200000 >> putStrLn "finished-1"+ threadDelay 1000000+ putStrLn "Starting testTaskPool:"+ startQueue pool+ threadDelay 4000000+ stopQueue pool+ forkIO $ claim Acquire (schedule pool 0) $ failed "testTaskPool: This task should never run!"+ withMVar m_inversions $ \inversions -> when (inversions > 2) $ failed "testTaskPool: too many priority inversions"+ withMVar m_count $ \count -> when (count /= 15) $ failed "testTaskPool: did not complete all tasks within 4 seconds"+ putStrLn "Finished testTaskPool:"++stress :: forall a. (Ord a) => QueueConfigurationRecord a -> (IO a) -> IO ()+stress config prioIO =+ do putStrLn "stressTest"+ putStrLn "Create 10,000 threads in a room of size 100, each test needs half a second to complete, and see what happens."+ threadDelay 3000000+ pool <- newTaskPool config 100 ()+ startQueue pool+ counter <- newMVar 0+ forM_ [1..10000] $ \_ ->+ do prio <- prioIO+ forkIO $ claim Acquire (schedule pool prio) $ threadDelay 500000 >> modifyMVar_ counter (return . (+1))+ threadDelay 50000000+ atomically $ flip unless retry . (== 0) =<< activity pool+ withMVar counter $ putStrLn . show++_example1 :: IO ()+_example1 =+ do (pool :: TaskPool () ()) <- simpleTaskPool+ forkIO $ claim Acquire pool $ putStrLn "Hello world!"+ forkIO $ claim Acquire pool $ putStrLn "Goodbye world!"+ startQueue pool+ +_example2 :: IO ()+_example2 =+ do prio_pool <- simpleTaskPool+ forkIO $ claim Acquire (schedule prio_pool 1) $ putStrLn "Hello world!"+ forkIO $ claim Acquire (schedule prio_pool 2) $ putStrLn "Goodbye world!"+ startQueue prio_pool++main :: IO ()+main =+ do args <- liftM (\args -> if Prelude.null args then ["help"] else args) getArgs+ let shouldRun s@('s':'t':'r':'e':'s':'s':_) = s `elem` args+ shouldRun s = s `elem` args || "all" `elem` args+ when (shouldRun "help") $ putStrLn "tests: all, testRoom, testMaxThreads, testQueue, testTaskPool, stressInt, stressIntFair, stressInt2, stressUnit, stressUnitFILO, stressUnitFair"+ when (shouldRun "testRoom") testRoom+ when (shouldRun "testMaxThreads") testMaxThreads+ when (shouldRun "testQueue") testQueue+ when (shouldRun "testTaskPool") testTaskPool+ when (shouldRun "stressInt") $ stress fast_queue_configuration $ randomRIO (0,1000 :: Int)+ when (shouldRun "stressIntFair") $ stress fair_queue_configuration $ randomRIO (0,1000 :: Int)+ when (shouldRun "stressInt2") $ stress fast_queue_configuration $ randomRIO (0,2 :: Int)+ when (shouldRun "stressUnit") $ stress fast_queue_configuration $ return ()+ when (shouldRun "stressUnitFILO") $ stress (fast_queue_configuration { queue_order = FILO }) $ return ()+ when (shouldRun "stressUnitFair") $ stress fair_queue_configuration $ return ()+ withMVar fail_strs $ \strs -> + do forM strs $ \s -> putStrLn $ "FAILED: " ++ s+ when (not $ Prelude.null strs) $ exitFailure+ putStrLn "Done."
+ priority-sync.cabal view
@@ -0,0 +1,80 @@+name: priority-sync+version: 0.1.0.0+license: BSD3+license-file: LICENSE+author: Christopher Lane Hinson+maintainer: Christopher Lane Hinson <lane@downstairspeople.org>+stability: Unstable++category: Concurrency+synopsis: Task prioritization.+description: Implements cooperative task prioritization with room synchronization.+ .+ In the simplest usage, for an unprioritized FILO queue, only three operations are needed: 'simpleTaskPool', 'claim', and 'startQueue'.+ .+ @+ (pool :: TaskPool () ()) <- simpleTaskPool+ forkIO $ claim Acquire pool $ putStrLn "Hello world!"+ forkIO $ claim Acquire pool $ putStrLn "Goodbye world!"+ startQueue pool+ @+ .+ For a simple prioritized queue, the 'schedule' operation introduces the priority. Prioritization is always least-first.+ .+ @+ prio_pool <- simpleTaskPool+ forkIO $ claim Acquire (schedule prio_pool 1) $ putStrLn "Hello world!"+ forkIO $ claim Acquire (schedule prio_pool 2) $ putStrLn "Goodbye world!"+ startQueue prio_pool+ @+ .+ Note that if you run these programs with @+RTS -N2@ or greater, the 'claim' operations may be processed in any order, since 'simpleTaskQueue' detects+ the number of capabilities and schedules tasks on each.+ .+ 'TaskPool's are not thread pools. The concept is similar to IO Completion Ports. There are no worker threads. If a number of threads are waiting,+ the thread that is most likely to be processed next is woken and temporarily serves as a working thread. 'TaskPool's are backed by carefully+ written STM (software transactional memory) transactions.+ .+ A salient feature is that, because any thread can participate, a 'TaskPool' supports both bound threads and threads created with 'forkOnIO'.+ .+ For applications that have complex resource constraints, it is possible to create a 'Room' to model each constraint. 'Room's are fully reentrant,+ and an arbitrary number of threads can 'claim' a 'Room' according to arbitrary rules, or 'RoomConstraint's. In the simple usage above,+ a single room represents the number of capabilities available to the GHC runtime.+ .+ Whenever a thread attempts to acquire a 'Room', a 'Claim' is generated that represents that attempt. The 'Claim' can be approved immediately,+ or it can be approved at the whim of another thread that has access to that 'Claim'. This means that 'Room's can be constructed in such+ a way that 'Claim's are approved only when a third party thread sees that the resource constraint modeled by that 'Room' has been satisfied.+ .+ The rules for generating and approving 'Claim's are described by a 'RoomContext'. By default, 'Claim's are approved immediately if their+ associated 'RoomConstraint's have been satisfied, but when a 'TaskPool' is introduced approval is deferred for prioritization.+ .+ 'Room' constraints are completely advisory: any task may claim any 'Room' without restriction at any time by using the 'UnconstrainedRoomContext'.+ .+ 'Queue's are used to prioritize tasks. Even if you have no need for prioritization, a 'Queue' ensures that only one thread is woken up+ when a 'Room' becomes available. A 'Queue' systematically examines to a configurable depth all waiting threads with their priorities+ and constraints and wakes the most eagerly prioritized thread whose constraints can be satisfied.+ .+ A 'TaskPool' combines 'Room's and 'Queue's in an efficient, easy-to-use interface.+ .+ The git repository is available at <http://www.downstairspeople.org/git/priority-sync.git>.++cabal-version: >= 1.2+build-type: Simple+tested-with: GHC==6.10.1++Library+ exposed-modules: Control.Concurrent.Priority.Room,+ Control.Concurrent.Priority.Queue, + Control.Concurrent.Priority.RoomConstraint,+ Control.Concurrent.Priority.Schedule, + Control.Concurrent.Priority.TaskPool+ other-modules: Control.Concurrent.Priority.RoomCore+ ghc-options: -Wall -fno-warn-type-defaults+ ghc-prof-options: -prof -auto-all+ build-depends: base>3, containers >= 0.1.0.1, heap, parallel >= 1.0.0.0, stm >= 2.1.1.2, random++Executable _Control_Concurrent_Priority_Tests+ Main-Is: Tests.hs+ ghc-options: -Wall -threaded -fno-warn-type-defaults+ ghc-prof-options: -prof -auto-all+ build-depends: base>3