priority-sync 0.1.0.1 → 0.2.1.0

raw patch · 23 files changed

+999/−753 lines, 23 filesdep +PSQueuedep −heapdep ~basenew-component:exe:_PrioritySync.Internal_Tests

Dependencies added: PSQueue

Dependencies removed: heap

Dependency ranges changed: base

Files

• Control/Concurrent/Priority/Queue.hs +0/−248
• Control/Concurrent/Priority/Room.hs +0/−124
• Control/Concurrent/Priority/RoomConstraint.hs +0/−48
• Control/Concurrent/Priority/RoomCore.hs +0/−121
• Control/Concurrent/Priority/Schedule.hs +0/−55
• Control/Concurrent/Priority/TaskPool.hs +0/−78
• PrioritySync/Internal/ClaimContext.hs +25/−0
• PrioritySync/Internal/Constrained.hs +26/−0
• PrioritySync/Internal/Dispatch.hs +40/−0
• PrioritySync/Internal/Prioritized.hs +18/−0
• PrioritySync/Internal/Queue.hs +273/−0
• PrioritySync/Internal/Receipt.hs +33/−0
• PrioritySync/Internal/Room.hs +39/−0
• PrioritySync/Internal/RoomConstraint.hs +48/−0
• PrioritySync/Internal/RoomCore.hs +144/−0
• PrioritySync/Internal/RoomGroup.hs +21/−0
• PrioritySync/Internal/Schedule.hs +44/−0
• PrioritySync/Internal/TaskPool.hs +90/−0
• PrioritySync/Internal/Unconstrained.hs +25/−0
• PrioritySync/Internal/UserData.hs +10/−0
• PrioritySync/PrioritySync.hs +77/−0
• Tests.hs +66/−39
• priority-sync.cabal +20/−40

@@ -1,248 +0,0 @@-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

@@ -1,124 +0,0 @@-{-# 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-

@@ -1,48 +0,0 @@-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)-

@@ -1,121 +0,0 @@-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

@@ -1,55 +0,0 @@-{-# 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

@@ -1,78 +0,0 @@-{-# 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)

@@ -0,0 +1,25 @@+{-# LANGUAGE TypeFamilies #-}+module PrioritySync.Internal.ClaimContext+    (ClaimContext(..))+    where++import PrioritySync.Internal.UserData+import PrioritySync.Internal.RoomCore+import Control.Concurrent.STM++-- | Rules for calling 'claim_'.+class ClaimContext c where+    type ClaimHandle c :: *+    -- | Should approve a some claims before entering a critical section, as described by 'claim_'.+    approveClaimsEntering :: c -> [Claim (UserData c)] -> STM (ClaimHandle c)+    -- | Should approve a some claims before exiting a critical section, as described by 'claim_'.+    approveClaimsExiting :: c -> [Claim (UserData c)] -> STM (ClaimHandle c)+    -- | A waiting transaction, as described by 'claim_'.+    waitingAction :: c -> ClaimHandle c -> STM ()++instance (ClaimContext c) => ClaimContext (c,m) where+    type ClaimHandle (c,m) = ClaimHandle c+    approveClaimsEntering = approveClaimsEntering . fst+    approveClaimsExiting = approveClaimsExiting . fst+    waitingAction = waitingAction . fst+

@@ -0,0 +1,26 @@+{-# LANGUAGE TypeFamilies #-}+module PrioritySync.Internal.Constrained+    (Constrained(..))+    where++import PrioritySync.Internal.UserData+import PrioritySync.Internal.RoomGroup+import PrioritySync.Internal.ClaimContext+import PrioritySync.Internal.Room+import PrioritySync.Internal.RoomConstraint++-- | Require that all 'RoomConstraint's be satisfied when acquiring a 'Room'.+data Constrained u = Constrained++type instance UserData (Constrained u) = u++instance RoomGroup (Constrained u) where+    roomsOf = const []++instance (RoomConstraint u) => ClaimContext (Constrained u) where+    type ClaimHandle (Constrained u) = ()+    approveClaimsEntering _ cs = approveClaims cs >> return ()+    approveClaimsExiting _ cs = approveClaims cs >> return ()+    waitingAction _ () = return ()++

@@ -0,0 +1,40 @@+{-# LANGUAGE TypeFamilies, FlexibleContexts #-}+module PrioritySync.Internal.Dispatch+    (dispatch,TaskHandle,reprioritize,getResult,tryGetResult)+    where++import PrioritySync.Internal.Prioritized+import PrioritySync.Internal.Receipt+import PrioritySync.Internal.RoomGroup+import PrioritySync.Internal.Room+import PrioritySync.Internal.ClaimContext+import Control.Concurrent+import Control.Concurrent.STM++data TaskHandle p a = TaskHandle {+    task_reprioritize :: (p -> p) -> STM (),+    task_result :: TVar (Maybe a) }++-- | Perform a task on another thread.  This task can be reprioritized and canceled.+dispatch :: (RoomGroup c,ClaimContext c,Prioritized (ClaimHandle c)) => +            c -> IO a -> IO (TaskHandle (Priority (ClaimHandle c)) a)+dispatch c actionM =+    do result <- newTVarIO Nothing+       receive_task_handle <- newTVarIO Nothing+       _ <- forkIO $ (atomically . writeTVar result . Just) =<< claim Acquire (Receipt c (writeTVar receive_task_handle . Just) (const $ return ())) actionM+       task_handle <- atomically $ maybe retry return =<< readTVar receive_task_handle+       return $ TaskHandle (reprioritize task_handle) result++-- | Change the priority of a task.  This will not work if the task has already started.+instance Prioritized (TaskHandle p a) where+    type Priority (TaskHandle p a) = p+    reprioritize = task_reprioritize++-- | Wait for the result from this task.+getResult :: TaskHandle p a -> STM a+getResult task = maybe retry return =<< readTVar (task_result task)++-- | Non-blocking version of 'getResult'.+tryGetResult :: TaskHandle p a -> STM (Maybe a)+tryGetResult = readTVar . task_result+

@@ -0,0 +1,18 @@+{-# LANGUAGE TypeFamilies #-}++module PrioritySync.Internal.Prioritized+    (Prioritized(..))+    where++import Control.Concurrent.STM++-- | Reprioritize a task.  This has no effect on a target that has already left the queue.+class Prioritized p where+    type Priority p :: *+    reprioritize :: p -> (Priority p -> Priority p) -> STM ()++instance (Prioritized p) => Prioritized (Maybe p) where+    type Priority (Maybe p) = Priority p+    reprioritize Nothing = const $ return ()+    reprioritize (Just p) = reprioritize p+

@@ -0,0 +1,273 @@+{-# LANGUAGE TypeFamilies #-}++module PrioritySync.Internal.Queue+    (Queue,+     TaskHandle,+     QueueOrder(..),+     QueueConfigurationRecord(..),+     fair_queue_configuration, fast_queue_configuration,+     newQueue,+     taskPriority,+     taskQueue,+     pendingTasks,+     isTopOfQueue,+     hasCompleted,+     putTask,+     pullTask,+     pullFromTop,+     pullSpecificTasks,+     dispatchTasks,+     flushQueue,+     load,+     PrioritySync.Internal.Queue.isEmpty)+    where++import PrioritySync.Internal.Prioritized+import qualified Data.PSQueue as PSQ+import Data.PSQueue (Binding(..)) +import Data.List as List (sort,sortBy,groupBy,drop,unfoldr)+import GHC.Conc+import Control.Monad+import Control.Concurrent.MVar+import Data.Unique+import Data.Ord+import Data.Maybe+import Control.Arrow (first)++-- | 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 (PSQ.PSQ (TaskHandle a) (a,Integer)),+    task_counter :: TVar Integer,+    queue_is_empty :: TVar Bool }++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.+--+-- * 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_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 -> (Integer,Queue a)+taskOrd 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)++-- | Get the priority of this task, which only exists if the task is still enqueued.+taskPriority :: (Ord a) => TaskHandle a -> STM (Maybe a)+taskPriority task = liftM (fmap fst . PSQ.lookup task) $ readTVar $ pending_tasks $ taskQueue task++instance (Ord a) => Prioritized (TaskHandle a) where+    type Priority (TaskHandle a) = a+    reprioritize task f = watchingTopOfQueue (taskQueue task) $+        do let pending_tvar = pending_tasks $ taskQueue task+           writeTVar pending_tvar =<< liftM (PSQ.adjust (first f) task) (readTVar pending_tvar)++-- | Get the 'Queue' associated with this task.+taskQueue :: TaskHandle a -> Queue a+taskQueue = task_queue++pendingTasks :: (Ord a) => Queue a -> STM [TaskHandle a]+pendingTasks = liftM PSQ.keys . readTVar . pending_tasks++-- | Create a new 'Queue'.  +newQueue :: (Ord a) => QueueConfigurationRecord a -> IO (Queue a)+newQueue config = +    do pending_tasks_var <- newTVarIO PSQ.empty+       counter <- newTVarIO 0+       uniq <- newUnique+       is_empty <- newTVarIO True+       return Queue {+           queue_configuration = config,+           queue_unique = uniq,+           pending_tasks = pending_tasks_var,+           task_counter = counter,+           queue_is_empty = is_empty }++-- | 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_queue = q }+       watchingTopOfQueue q $ writeTVar (pending_tasks q) . PSQ.insert task (prio,task_counter_index task) =<< readTVar (pending_tasks q)+       return task++-- | The number of tasks pending on this Queue.+load :: (Ord a) => Queue a -> STM Int +load = liftM PSQ.size . readTVar . pending_tasks++-- | True iff this Queue is empty.+isEmpty :: (Ord a) => Queue a -> STM Bool+isEmpty = readTVar . queue_is_empty++-- | 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_asc_list <- liftM (List.unfoldr PSQ.minView) (readTVar $ pending_tasks q)+       task <- pullTask_ (queue_configuration q) (fst $ PSQ.prio $ head task_asc_list) 0 task_asc_list+       writeTVar (pending_tasks q) =<< liftM (PSQ.delete task) (readTVar $ pending_tasks q)+       writeTVar (has_completed task) True+       return task++pullTask_ :: (Ord a) => QueueConfigurationRecord a ->+                        a ->+                        Int ->+                        [PSQ.Binding (TaskHandle a) (a,Integer)] ->+                        STM (TaskHandle a)+pullTask_ _ _ _ [] = retry+pullTask_ config top_prio faltered_tasks (task:untried_tasks) =+    do when (faltered_tasks > allowed_ordering_inversion config) retry+       unless (allowed_priority_inversion config top_prio $ fst $ PSQ.prio task) retry+       prio_ok <- ((priority_indexed_predicate config $ fst $ PSQ.prio task) >> return True) `orElse` (return False)+       case prio_ok of+           False -> pullTask_ config top_prio (succ faltered_tasks) $ dropWhile ((== PSQ.prio task) . PSQ.prio) untried_tasks+           True -> (task_action (PSQ.key task) >> return (PSQ.key task)) `orElse` pullTask_ config top_prio (succ faltered_tasks) untried_tasks++-- | 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_min <- liftM PSQ.findMin $ readTVar (pending_tasks q)+       case m_min 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+       pending <- readTVar (pending_tasks q)+       is_empty <- readTVar (queue_is_empty q)+       when (PSQ.null pending && not is_empty) $ writeTVar (queue_is_empty q) True+       when ((not $ PSQ.null pending) && is_empty) $ writeTVar (queue_is_empty q) False+       return result+

@@ -0,0 +1,33 @@+{-# LANGUAGE TypeFamilies, UndecidableInstances #-}++module PrioritySync.Internal.Receipt+    (Receipt(..))+    where++import PrioritySync.Internal.UserData+import PrioritySync.Internal.RoomGroup+import PrioritySync.Internal.ClaimContext+import Control.Concurrent.STM++-- | Get a notification when a claim is approved or scheduled.+data Receipt c = Receipt {+    receipt_base_context :: c,+    receipt_entering_callback, receipt_exiting_callback :: ClaimHandle c -> STM () }++type instance UserData (Receipt c) = UserData c++instance (RoomGroup c) => RoomGroup (Receipt c) where+    roomsOf r = roomsOf $ receipt_base_context r++instance (ClaimContext c) => ClaimContext (Receipt c) where+    type ClaimHandle (Receipt c) = ClaimHandle c+    approveClaimsEntering r cs =+        do result <- approveClaimsEntering (receipt_base_context r) cs+           receipt_entering_callback r $ result+           return result+    approveClaimsExiting r cs =+        do result <- approveClaimsExiting (receipt_base_context r) cs+           receipt_exiting_callback r $ result+           return result+    waitingAction r = waitingAction $ receipt_base_context r+

@@ -0,0 +1,39 @@+{-# LANGUAGE MultiParamTypeClasses, TypeFamilies, FlexibleInstances, FlexibleContexts, UndecidableInstances #-}++module PrioritySync.Internal.Room+    (Room,+     newRoom,+     RoomCore.userData,+     inUse,+     Claim,+     claimedRoom,+     claimedThread,+     ClaimMode(..),+     claim,+     approveClaims,+     approve,+     RoomCore.isEmpty)+    where++import PrioritySync.Internal.RoomGroup+import PrioritySync.Internal.ClaimContext+import PrioritySync.Internal.RoomCore as RoomCore+import PrioritySync.Internal.RoomConstraint+import Control.Concurrent.STM+import Data.Map as Map++-- | 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 (Default,room) $ putStrLn "Hello World!"+-- > forkIO $ claim Acquire (Default,room) $ putStrLn "Foo!  Bar!"+claim :: (RoomGroup c,ClaimContext c) => ClaimMode -> c -> IO a -> IO a+claim claim_mode c actionIO = +    do 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+

@@ -0,0 +1,48 @@+module PrioritySync.Internal.RoomConstraint+    (RoomConstraint(..),+     MaxThreads(..),+     approveClaims)+    where++import PrioritySync.Internal.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)+

@@ -0,0 +1,144 @@+{-# LANGUAGE TypeFamilies #-}+module PrioritySync.Internal.RoomCore+    (Room,+     newRoom,+     Claim,+     ClaimMode(..),+     claimedRoom,+     claimedThread,+     userData,+     approve,+     claim_,+     inUse,+     isEmpty)+    where++import PrioritySync.Internal.UserData+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 {+    room_user_data :: (u,Unique),+    room_occupants :: (TVar (Set ThreadId)),+    room_is_empty :: (TVar Bool) }++type instance UserData (Room u) = u++instance Eq (Room u) where+    (==) r1 r2 = snd (room_user_data r1) == snd (room_user_data r2)++instance Ord (Room u) where+    compare r1 r2 = compare (snd $ room_user_data r1) (snd $ room_user_data r2)++-- | A 'Claim', or attempt to acquire or release a 'Room'.+data Claim u = Claim (Room u) ThreadId ClaimMode (TVar Bool)++type instance UserData (Claim u) = u++data ClaimMode = Acquire | Release deriving (Eq)++-- | Create a new Room with some arbitrary user data.+newRoom :: u -> IO (Room u)+newRoom u = return Room `ap` liftM ((,) u) newUnique `ap` newTVarIO Set.empty `ap` newTVarIO True++-- | Get the user data associated with a 'Room'.+userData :: Room u -> u+userData = fst . room_user_data++-- | 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 or releases a 'Room'.+approve :: Claim u -> STM ()+approve (Claim r me want claim_var) =+    do claim_state <- readTVar claim_var+       when (claim_state == False) $ +           do writeTVar claim_var True+              writeTVar (room_occupants r) . (case want of Acquire -> Set.insert; Release -> Set.delete) me =<< readTVar (room_occupants r)+              watchRoom r++-- | Set the 'room_is_empty' flag.+watchRoom :: Room u -> STM ()+watchRoom r =+    do is_empty <- readTVar (room_is_empty r)+       occus <- readTVar (room_occupants r)+       when (is_empty && (not $ Set.null occus)) $ writeTVar (room_is_empty r) True+       when (not is_empty && Set.null occus) $ writeTVar (room_is_empty r) False++-- | 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 claiming this 'Room'.+inUse :: Room u -> STM (Set ThreadId)+inUse = readTVar . room_occupants++-- | True iff a Room is empty.+isEmpty :: Room u -> STM Bool+isEmpty = readTVar . room_is_empty

@@ -0,0 +1,21 @@+{-# LANGUAGE TypeFamilies, UndecidableInstances #-}+module PrioritySync.Internal.RoomGroup+    (RoomGroup(..))+    where++import PrioritySync.Internal.UserData+import PrioritySync.Internal.RoomCore++class RoomGroup m where+    roomsOf :: m -> [Room (UserData m)]++instance RoomGroup (Room u) where+    roomsOf m = [m]++instance (RoomGroup rs) => RoomGroup [rs] where+    roomsOf = concatMap roomsOf++instance (UserData c ~ UserData m,RoomGroup c,RoomGroup m) => RoomGroup (c,m) where+    roomsOf (c,m) = roomsOf c ++ roomsOf m++

@@ -0,0 +1,44 @@+{-# LANGUAGE TypeFamilies, UndecidableInstances #-}++module PrioritySync.Internal.Schedule+    (Schedule(..))+    where++import PrioritySync.Internal.UserData+import PrioritySync.Internal.Room+import PrioritySync.Internal.Queue+import PrioritySync.Internal.ClaimContext+import PrioritySync.Internal.RoomGroup+import Control.Concurrent.STM+import Control.Monad+import Data.List++-- | Schedule a task to run from a prioritized 'Queue'.  The task will wait until it arrives at (or, with failover, near) the top of queue.  Typical usage:+--+-- > Schedule q 2 room1+--+-- Only the rooms inside the 'Schedule' declaration are claimed with scheduling.  If access to a room doesn't need to be prioritized, it can be set outside+-- the schedule:+--+-- > (Schedule q 2 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,ClaimContext c,ClaimHandle c ~ ()) => ClaimContext (Schedule p c) where+    type ClaimHandle (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,ClaimContext c,ClaimHandle 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)+ 

@@ -0,0 +1,90 @@+{-# 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 PrioritySync.Internal.TaskPool+    (TaskPool,+     PrioritySync.Internal.TaskPool.schedule,+     newTaskPool,+     simpleTaskPool,+     poolRoom,+     poolQueue,+     startQueue,+     stopQueue,+     PrioritySync.Internal.TaskPool.isEmpty,+     waitUntilFinished)+    where++import PrioritySync.Internal.Room as Room+import PrioritySync.Internal.Queue as Queue+import PrioritySync.Internal.Schedule+import PrioritySync.Internal.UserData+import PrioritySync.Internal.RoomGroup+import PrioritySync.Internal.RoomConstraint+import PrioritySync.Internal.Constrained+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]++-- | A prioritized 'ClaimContext' for a task pool.+schedule :: TaskPool p u -> p -> (Schedule p (Constrained (TaskPoolConstraint u),Room (TaskPoolConstraint u)))+schedule (TaskPool _ q m) p = Schedule q p (Constrained,m)++-- | Create a new 'TaskPool'.  'TaskPool's begin stopped, use 'startQueue' to start.+--+-- * A 'QueueConfigurationRecord' for the backing 'Queue'.  A typical value is 'fair_queue_configuration'.+--+-- * The allowed number of threads that can access the 'TaskPool' simultaneously.+--+-- * The user data for the backing 'Room'.  This can be @()@.+--+-- 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 ()++-- | Get the 'Room' that primarily constrains this 'TaskPool'.+poolRoom :: TaskPool p u -> Room (TaskPoolConstraint u)+poolRoom = pool_room++-- | Get the 'Queue' that admits new tasks to this 'TaskPool'.+poolQueue :: TaskPool p u -> Queue p+poolQueue = pool_queue++-- | Start the 'TaskPool'.+startQueue :: TaskPool p u -> STM ()+startQueue tp = writeTVar (pool_on tp) True++-- | Stop all activity on this 'TaskPool'.+stopQueue :: TaskPool p u -> STM ()+stopQueue tp = writeTVar (pool_on tp) False++-- | True iff this 'TaskPool' is entirely empty and inactive.+isEmpty :: (Ord p) => TaskPool p u -> STM Bool+isEmpty tp = liftM2 (&&) (Queue.isEmpty $ poolQueue tp) (Room.isEmpty $ poolRoom tp)++-- | Wait until a queue is finished.+waitUntilFinished :: (Ord p) => TaskPool p u -> IO ()+waitUntilFinished tp = atomically $ flip unless retry =<< PrioritySync.Internal.TaskPool.isEmpty tp+

@@ -0,0 +1,25 @@+{-# LANGUAGE TypeFamilies #-}+module PrioritySync.Internal.Unconstrained+    (Unconstrained(..))+    where++import PrioritySync.Internal.UserData+import PrioritySync.Internal.RoomGroup+import PrioritySync.Internal.ClaimContext+import PrioritySync.Internal.Room++-- | Don't check any 'RoomConstraint's when acquiring a 'Room'.+data Unconstrained u = Unconstrained++type instance UserData (Unconstrained u) = u++instance RoomGroup (Unconstrained u) where+    roomsOf = const []++instance ClaimContext (Unconstrained u) where+    type ClaimHandle (Unconstrained u) = ()+    approveClaimsEntering _ cs = mapM_ approve cs >> return ()+    approveClaimsExiting _ cs = mapM_ approve cs >> return ()+    waitingAction _ _ = return ()++

@@ -0,0 +1,10 @@+{-# LANGUAGE TypeFamilies #-}+module PrioritySync.Internal.UserData+    (UserData)+    where++type family UserData u :: *++type instance UserData [u] = UserData u+type instance UserData (c,m) = UserData c+

@@ -0,0 +1,77 @@+module PrioritySync.PrioritySync+    (Dispatch.TaskHandle,+     Room.claim,+     Dispatch.dispatch,+     getResult,+     tryGetResult,+     Prioritized.Prioritized(),+     reprioritize,+     Constrained.Constrained(..),+     Unconstrained.Unconstrained(..),+     RoomConstraint.MaxThreads(..),+     Room.Room,+     Room.newRoom,+     Room.userData,+     Room.ClaimMode(..),+     load,+     Occupancy(..),+     Queue.QueueConfigurationRecord(..),+     Queue.fair_queue_configuration,+     Queue.fast_queue_configuration,+     Queue.QueueOrder(..),+     TaskPool.TaskPool,+     TaskPool.schedule,+     TaskPool.newTaskPool,+     TaskPool.simpleTaskPool,+     startQueue,+     stopQueue,+     TaskPool.waitUntilFinished)+    where++import qualified PrioritySync.Internal.Dispatch as Dispatch+import qualified PrioritySync.Internal.Prioritized as Prioritized+import qualified PrioritySync.Internal.Room as Room+import qualified PrioritySync.Internal.RoomConstraint as RoomConstraint+import qualified PrioritySync.Internal.Queue as Queue+import qualified PrioritySync.Internal.TaskPool as TaskPool+import qualified PrioritySync.Internal.Constrained as Constrained+import qualified PrioritySync.Internal.Unconstrained as Unconstrained++import Data.Set as Set+import Control.Concurrent.STM+import Control.Concurrent+import Control.Monad++getResult :: Dispatch.TaskHandle p a -> IO a+getResult task = atomically $ Dispatch.getResult task++tryGetResult :: Dispatch.TaskHandle p a -> IO (Maybe a)+tryGetResult task = atomically $ Dispatch.tryGetResult task++reprioritize :: Dispatch.TaskHandle p a -> (p -> p) -> IO ()+reprioritize task f = atomically $ Prioritized.reprioritize task f++-- | The number of tasks waiting on this 'TaskPool'.+load :: (Ord p) => TaskPool.TaskPool p u -> IO Int+load = atomically . Queue.load . TaskPool.poolQueue++-- | A convenience class to observe the currently running occupants of a 'Room' or 'TaskPool'.+class Occupancy o where+    inUse :: o -> IO (Set ThreadId)+    isEmpty :: o -> IO Bool+    isEmpty = liftM Set.null . inUse++instance (Ord p) => Occupancy (TaskPool.TaskPool p u) where+    inUse pool = atomically $ Room.inUse $ TaskPool.poolRoom pool+    isEmpty pool = atomically $ TaskPool.isEmpty pool++instance Occupancy (Room.Room u) where+    inUse = atomically . Room.inUse+    isEmpty = atomically . Room.isEmpty++startQueue :: TaskPool.TaskPool p a -> IO ()+startQueue = atomically . TaskPool.startQueue++stopQueue :: TaskPool.TaskPool p a -> IO ()+stopQueue = atomically . TaskPool.stopQueue+

@@ -1,17 +1,18 @@-{-# LANGUAGE RecursiveDo, ScopedTypeVariables #-}+{-# LANGUAGE DoRec, 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 PrioritySync.PrioritySync+import qualified PrioritySync.Internal.Queue as Queue++import Control.Concurrent+import Control.Concurrent.STM import Control.Monad import System.Random import Data.Set as Set-import GHC.Conc import System.Environment import System.IO.Unsafe import System.Exit+import Control.Exception  {-# NOINLINE fail_strs #-} fail_strs :: MVar [String]@@ -28,14 +29,14 @@        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)+       let f s b = do ok <- liftM ((== b) . member me) $ inUse m+                      when (not ok) $ failed $ "testRoom: " ++ s+       f "testRoom-1" False+       claim Acquire (Constrained,[m]) $ f "testRoom-2" True+       f "testRoom-3" False+       claim Release (Constrained,[m]) $ f "testRoom-4" False+       f "testRoom-5" False+       claim Acquire (Constrained,[m]) $ claim Acquire (Constrained,[m]) (f "testRoom-6" True) >> f "testRoom-7" True >> claim Release (Constrained,[m]) (f "testRoom-8" False)  testMaxThreads :: IO () testMaxThreads =@@ -49,10 +50,10 @@                             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"+       claim Acquire (Constrained,[large,small]) $+           do forM_ [1..8] $ const $ forkIO $ claim Acquire (Constrained,[large,small]) $ runThread "large+small"+              forM_ [1..12] $ const $ forkIO $ claim Acquire (Constrained,[large]) $ runThread "large"+              forM_ [1..4] $ const $ forkIO $ claim Acquire (Constrained,[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"@@ -69,14 +70,15 @@  testQueue :: IO () testQueue =-   mdo putStrLn "testQueue"+    do 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 () }+       rec q <- Queue.newQueue $ fair_queue_configuration {+                                     allowed_ordering_inversion = 15, -- will stall with the default value, b/c the failover requires a significant ordering inversion+                                     queue_predicate = flip when retry =<< readTVar need_to_print, +                                     priority_indexed_predicate = \x -> do l <- Queue.load q; if x == 1 && l > 10 then retry else return () }        counter <- newTVarIO 0        str <- newTVarIO ""        let incCounter x s = @@ -86,15 +88,16 @@                   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 forM_ [1..4] $ const $ Queue.putTask q 0 $ incCounter 0 "priority-0"+              forM_ [1..4] $ const $ Queue.putTask q 1 $ incCounter 1 "priority-1, load <= 10"+              forM_ [1..4] $ const $ Queue.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..4] $ const $ Queue.putTask q 3 $ incCounter 3 "priority-3"+              return ()        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)+           do m_s <- atomically $ (do b <- readTVar need_to_print; if b then liftM Just (readTVar value_to_print) else retry) `orElse` (Queue.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"@@ -107,24 +110,47 @@        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))+       m_greatest_prio <- newMVar 0+       let testPrio n = modifyMVar_ m_greatest_prio $ \greatest_prio ->+               do when (greatest_prio > n) $ modifyMVar_ m_inversions (return . (+1))                   modifyMVar_ m_count (return . (+1))-                  return n+                  return $ max greatest_prio 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"+       _ <- 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!"+       _ <- forkIO $ (claim Acquire (schedule pool 0) $ failed "testTaskPool: This task should never run!") `finally`+                     (putStrLn "testTaskPool: runtime discovered that never-running task was hung (this is good)")        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:" +testDispatch :: IO ()+testDispatch =+    do putStrLn "testDispatch"+       putStrLn "Dispatch 8 tasks into a room of size 3.  Reprioritize one task from 7 to 0."+       io_sem <- newMVar ()+       c <- newMVar 0+       let runThread s = do withMVar io_sem $ const $ putStrLn $ "{" ++ s+                            threadDelay 2000000+                            modifyMVar_ c (return . (+1))+                            withMVar io_sem $ const $ putStrLn $ s ++ "}"+       pool <- newTaskPool fair_queue_configuration 3 ()+       task_handles <- forM [1..8 :: Integer] $ \n -> dispatch (schedule pool n) (runThread $ "priority-" ++ show n)+       startQueue pool+       threadDelay 1000000+       reprioritize (task_handles !! 6) (const 0)+       putStrLn "Just reprioritized 7 -> 0 after waiting on second."+       threadDelay 2000000+       withMVar c $ \x -> when (x < 3) $ failed "testDispatch: should have completed at least 3 tasks within 3 seconds"+       withMVar c $ \x -> when (x > 5) $ failed "testDispatch: should not have completed more than 5 tasks within 3 seconds"+       threadDelay 7000000+       withMVar c $ \x -> when (x /= 8) $ failed "testDispatch: did not complete after 7 seconds."+ stress :: forall a. (Ord a) => QueueConfigurationRecord a -> (IO a) -> IO () stress config prioIO =     do putStrLn "stressTest"@@ -137,16 +163,16 @@            do prio <- prioIO               forkIO $ claim Acquire (schedule pool prio) $ threadDelay 500000 >> modifyMVar_ counter (return . (+1))        threadDelay 50000000-       atomically $ flip unless retry . (== 0) =<< activity pool+       waitUntilFinished pool        withMVar counter $ putStrLn . show  example :: IO () example =     do let expensiveTask = threadDelay 1000000        pool <- simpleTaskPool-       forkIO $ claim Acquire (schedule pool 1) $ putStrLn "Task 1 started . . ." >> expensiveTask >> putStrLn "Task 1 completed."-       forkIO $ claim Acquire (schedule pool 3) $ putStrLn "Task 3 started . . ." >> expensiveTask >> putStrLn "Task 3 completed."-       forkIO $ claim Acquire (schedule pool 2) $ putStrLn "Task 2 started . . ." >> expensiveTask >> putStrLn "Task 2 completed."+       _ <- forkIO $ claim Acquire (schedule pool 1) $ putStrLn "Task 1 started . . ." >> expensiveTask >> putStrLn "Task 1 completed."+       _ <- forkIO $ claim Acquire (schedule pool 3) $ putStrLn "Task 3 started . . ." >> expensiveTask >> putStrLn "Task 3 completed."+       _ <- forkIO $ claim Acquire (schedule pool 2) $ putStrLn "Task 2 started . . ." >> expensiveTask >> putStrLn "Task 2 completed."        threadDelay 100000  -- contrive to wait for all tasks to become enqueued        putStrLn "Starting pool: "        startQueue pool@@ -158,12 +184,13 @@        let shouldRun s@('s':'t':'r':'e':'s':'s':_) = s `elem` args            shouldRun "example" = "example" `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 "help") $ putStrLn "tests: all, testRoom, testMaxThreads, testQueue, testTaskPool, testDispatch, stressInt, stressIntFair, stressInt2, stressUnit, stressUnitFILO, stressUnitFair"        when (shouldRun "example") $ example        when (shouldRun "testRoom") testRoom        when (shouldRun "testMaxThreads") testMaxThreads        when (shouldRun "testQueue") testQueue        when (shouldRun "testTaskPool") testTaskPool+       when (shouldRun "testDispatch") testDispatch        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)@@ -171,6 +198,6 @@        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+           do forM_ strs $ \s -> putStrLn $ "FAILED: " ++ s               when (not $ Prelude.null strs) $ exitFailure        putStrLn "Done."

@@ -1,5 +1,5 @@ name:                priority-sync-version:             0.1.0.1+version:             0.2.1.0 license:             BSD3 license-file:        LICENSE author:              Christopher Lane Hinson@@ -8,55 +8,35 @@  category:            Concurrency synopsis:            Cooperative task prioritization.-description:         In a simple use case, we want to run some expensive tasks in prioritized order, so that only one task is running on each-                     CPU (or hardware thread) at any time.   For this simple case, four operations are needed: 'simpleTaskPool', -                     'schedule', 'claim', and 'startQueue'.-                     .-                     @-                     let expensiveTask = threadDelay 1000000-                     pool <- simpleTaskPool-                     forkIO $ claim Acquire (schedule pool 1) $ putStrLn \"Task 1 started . . .\" >> expensiveTask >> putStrLn \"Task 1 completed.\"-                     forkIO $ claim Acquire (schedule pool 3) $ putStrLn \"Task 3 started . . .\" >> expensiveTask >> putStrLn \"Task 3 completed.\"-                     forkIO $ claim Acquire (schedule pool 2) $ putStrLn \"Task 2 started . . .\" >> expensiveTask >> putStrLn \"Task 2 completed.\"-                     threadDelay 100000  -- contrive to wait for all tasks to become enqueued-                     putStrLn \"Starting pool: \"-                     startQueue pool-                     threadDelay 4000000 -- contrive to wait for all tasks to become dequeued-                     @-                     .-                     A 'TaskPool' combines 'Room's and 'Queue's in an efficient easy-to-use-interface.-                     .-                     'Room's provide fully reentrant synchronization to any number of threads based on arbitrary resource constraints.-                     For example, the 'Room' from a 'simpleTaskPool' is constrained by 'GHC.numCapabilities'.-                     .-                     'Queue's provide task prioritization.  A 'Queue' systematically examines (to a configurable depth) all waiting threads with their-                     priorities and resource constraints and wakes the most eagerly prioritized thread whose constraints can be satisfied.-                     .-                     '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.  -                     .-                     'Room's, 'Queue's, and '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'.+description:         A strategy to prioritize access to limited resources.                      .                      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+tested-with:         GHC==6.12.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+  exposed-modules:     PrioritySync.PrioritySync,+                       PrioritySync.Internal.Room,+                       PrioritySync.Internal.Queue, +                       PrioritySync.Internal.RoomConstraint,+                       PrioritySync.Internal.Schedule, +                       PrioritySync.Internal.TaskPool+                       PrioritySync.Internal.Receipt+                       PrioritySync.Internal.Prioritized+                       PrioritySync.Internal.Dispatch+                       PrioritySync.Internal.UserData+                       PrioritySync.Internal.RoomGroup+                       PrioritySync.Internal.ClaimContext+                       PrioritySync.Internal.Constrained+                       PrioritySync.Internal.Unconstrained+  other-modules:       PrioritySync.Internal.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+  build-depends:       base>4&&<5, containers >= 0.1.0.1, PSQueue, parallel >= 1.0.0.0, stm >= 2.1.1.2, random -Executable _Control_Concurrent_Priority_Tests+Executable _PrioritySync.Internal_Tests   Main-Is:             Tests.hs   ghc-options:         -Wall -threaded -fno-warn-type-defaults   ghc-prof-options:    -prof -auto-all