diff --git a/SafeSemaphore.cabal b/SafeSemaphore.cabal --- a/SafeSemaphore.cabal +++ b/SafeSemaphore.cabal @@ -1,8 +1,10 @@ Name: SafeSemaphore -Version: 0.7.0 +Version: 0.9.0 Synopsis: Much safer replacement for QSemN, QSem, and SampleVar -Description: This provides a much safer semaphore than the QSem in base. Performance has not been compared. In the source is a tests/TestKillSem.hs executable (run by cabal test) that shows the problem with QSem. -Homepage: http://hackage.haskell.org/package/SafeSemaphore +Description: This provides a much safer semaphore than the QSem, QSemN, SampleVar in base. + Those base modules are not exception safe and can be broken by killThread. + See for more details. +Homepage: https://github.com/ChrisKuklewicz/SafeSemaphore License: BSD3 License-file: LICENSE Author: Chris Kuklewicz @@ -16,9 +18,13 @@ ghc-options: -Wall -O2 -funbox-strict-fields Exposed-modules: Control.Concurrent.MSem Control.Concurrent.MSemN + Control.Concurrent.MSemN2 Control.Concurrent.MSampleVar Control.Concurrent.FairRWLock - Build-depends: base < 5, containers + Control.Concurrent.SSem + Control.Concurrent.STM.SSem + Other-modules: Control.Concurrent.STM.SSemInternals + Build-depends: base < 5, containers, stm Test-Suite TestSafeSemaphore type: exitcode-stdio-1.0 diff --git a/src/Control/Concurrent/FairRWLock.hs b/src/Control/Concurrent/FairRWLock.hs --- a/src/Control/Concurrent/FairRWLock.hs +++ b/src/Control/Concurrent/FairRWLock.hs @@ -1,51 +1,50 @@ {-# LANGUAGE DeriveDataTypeable, PatternGuards #-} -{-| Provides a fair RWLock, similar to one from Java. - +{-| Provides a fair RWLock, similar to one from Java, which is itself documented at - There are complicated policy choices that have to be made. This policy choices here are different - from the ones for the RWLock in concurrent-extras. - - The preferred way to use this API is sticking to 'new', 'withRead', and 'withWrite'. + There are complicated policy choices that have to be made. The policy choices here are different +from the ones for the RWLock in concurrent-extras. - The readers and writers are always identified by their ThreadId. Each thread that calls - acquireRead must later call releaseRead from the same thread. Each thread that calls acquireWrite - must later call releaseWrite from the same thread. + The 'FairRWLock' may be in a free unlocked state, it may be in a read locked state, or it may be a +write locked state. Many running threads may hold the read lock and execute concurrently. Only one +running thread may hold the write lock. The scheduling is a fair FIFO queue that avoids starvation. - The main way to misuse a FairRWLock is to call a release without having called an acquire. This is - reported in the (Left error) outcomes from releaseRead and releaseWrite. If the FairRWLock has a - bug and finds itself in an impossible state then it will throw an error. + When in the read lock state the first 'acquireWrite' will block, and subsequent 'acquireRead' and +'acquireWrite' will queue in order. When in the write locked state all other threads trying to +'acquireWrite' or 'acquireRead' will queue in order. - The FairRWLock may be in a free unlocked state, it may be in a read locked state, and it may be a - write locked state. Many running threads may hold the read lock and execute concurrently. Only - one running thread may hold the write lock. The scheduling is a fair FIFO queue that avoids - starvation. + 'FairRWLock' allows recursive write locks, and it allows recursive read locks, and it allows the +write lock holding thread to acquire read locks. When the current writer also holds read locks and +then releases its last write lock it will immediately convert to the read locked state (and other +waiting readers may join it). When a reader acquires a write lock it will (1) release all its read +locks, (2) wait to acquire the write lock, (3) retake the same number of read locks released in (1). - When in the read lock state the first acquireWrite will block, and subsequent acquireRead and - acquireWrite will queue in order. When in the write locked state all other threads trying to - acquireWrite or acquireRead will queue in order. + The preferred way to use this API is sticking to 'new', 'withRead', and 'withWrite'. - FairRWLock allows recursive write locks, and it allows recursive read locks, and it allows the - write lock holding thread to acquire read locks. When the current writer also holds read locks and - then releases its last write lock it will immediately convert to the read locked state (and other - waiting readers may join it). When a reader acquires a write lock it will (1) release all its read - locks, (2) wait to acquire the write lock, (3) retake the same number of read locks released in - (1). + No sequence of calling acquire on a single RWLock should lead to deadlock. Exceptions, espcially +from 'killThread', do not break 'withRead' or 'withWrite'. The 'withRead' and 'withWrite' ensure +all locks get released when exiting due to an exception. - No sequence of calling acquire on a single RWLock should lead to deadlock. + The readers and writers are always identified by their 'ThreadId'. Each thread that calls +'acquireRead' must later call 'releaseRead' from the same thread. Each thread that calls +'acquireWrite' must later call 'releaseWrite' from the same thread. The main way to misuse a +FairRWLock is to call a release without having called an acquire. This is reported in the (Left +error) outcomes from 'releaseRead' and 'releaseWrite'. Only if the 'FairRWLock' has a bug and finds +itself in an impossible state then it will throw an error. -} module Control.Concurrent.FairRWLock ( RWLock, RWLockException(..), RWLockExceptionKind(..),FRW(..),LockKind(..),TMap,TSet - , new, peekLock, checkLock + , new + , withRead, withWrite , acquireRead, acquireWrite , releaseRead, releaseWrite - , withRead, withWrite + , peekLock, checkLock ) where import Control.Applicative(liftA2) import Control.Concurrent -import Control.Exception +import Control.Exception(Exception,bracket_,onException,evaluate,uninterruptibleMask_,mask_,throw) import Control.Monad((>=>),join,forM_) import Data.Sequence((<|),(|>),(><),Seq,ViewL(..),ViewR(..)) import qualified Data.Sequence as Seq(empty,viewl,viewr,breakl,spanl) @@ -90,16 +89,16 @@ data RWLockException = RWLockException ThreadId RWLockExceptionKind String deriving (Show,Typeable) --- | Operation in which error arose +-- | Operation in which error arose, data RWLockExceptionKind = RWLock'acquireWrite | RWLock'releaseWrite | RWLock'acquireRead | RWLock'releaseRead deriving (Show,Typeable) instance Exception RWLockException --- | Observable state of holders of lock. The W returns a pair of Ints where the first is number of +-- | Observable state of holder(s) of lock(s). The W returns a pair of Ints where the first is number of -- read locks (at least 0) and the second is the number of write locks held (at least 1). The R --- returns a map from thread id to the positive number of read locks held. +-- returns a map from thread id to the number of read locks held (at least 1). data FRW = F | R TMap | W (ThreadId,(Int,Int)) deriving (Show) -- | Create a new RWLock which starts in a free and unlocked state. @@ -160,23 +159,27 @@ -- -- This can block but cannot be interrupted. releaseRead :: RWLock -> IO (Either RWLockException ()) -releaseRead (RWL rwlVar) = uninterruptibleMask_ $ do +releaseRead (RWL rwlVar) = mask_ $ do me <- myThreadId - releaseRead' False me rwlVar -- False to indicate called from releaseRead + releaseRead' False me rwlVar -- False to indicate call is from releaseRead --- Eleven non-impossible cases, plus one impossible case --- Lock is Free, error or impossible --- I have write lock, I have no read lock, error or impossible --- , I have at least one read lock, just decrement the counter --- Someone else has write lock, abandoning my acquireWrite --- , releaseWrite called in error --- Read lock held, I have 1 read lock, no other readers, change to FreeLock --- , change to next Writer --- , remove and leave to other readers --- , I have more than one read lock, just decrement the counter --- , I have no read lock, abandoning with no queue is IMPOSSIBLE --- , abandoning from queue past next writer --- , releaseRead called in error +-- The (abandon :: Bool) is False if called from releaseRead (from user API). +-- The (abandon :: Bool) is True if called as handler when acquireRead[Priority] interrupted by exception (internal use). +-- +-- There are 14 cases. +-- Four ERROR cases from misuse of releaseRead, Three IMPOSSIBLE cases (from interruptions), Seven normal cases: +-- Lock is Free, ERROR if releaseRead or IMPOSSIBLE if interrupted -- 1 and 2 +-- I have write lock, I have no read lock, ERROR if releaseRead or IMPOSSIBLE if interrupted -- 3 and 4 +-- , I have at least one read lock, just decrement the counter -- 5 +-- Someone else has write lock, abandoning my acquireWrite -- 6 +-- , releaseRead called in ERROR -- 7 +-- Read lock held, I have 1 read lock, no other readers, change to FreeLock -- 8 +-- , change to next Writer -- 9 +-- , remove and leave to other readers -- 10 +-- , I have more than one read lock, just decrement the counter -- 11 +-- , I have no read lock, abandoning with no queue is IMPOSSIBLE -- 12 +-- , abandoning from queue past next writer -- 13 +-- , releaseRead called in ERROR -- 14 releaseRead' :: Bool -> ThreadId -> MVar LockUser -> IO (Either RWLockException ()) releaseRead' abandon me rwlVar = uninterruptibleMask_ . modifyMVar rwlVar $ \ rwd -> do let impossible :: Show x => String -> x -> IO a @@ -202,24 +205,29 @@ evaluate $ if Set.null rcs' then pre >< post else pre >< ((ReaderKind rcs',mblock) <| post) case rwd of - FreeLock | abandon -> + FreeLock | abandon -> {- 1 -} impossible "acquireRead interrupted with unlocked RWLock" me - | otherwise -> + + | otherwise -> {- 2 -} err "cannot releaseRead lock from unlocked RWLock" me w@(Writer { writerID=it, readerCount=rc, queue=q }) | it==me -> do case rc of - 0 | abandon -> impossible "acquireRead interrupted with write lock but not read lock" (me,it) - | otherwise -> err "releaseRead when holding write lock but not read lock" (me,it) - _ -> do + 0 | abandon -> {- 3 -} + impossible "acquireRead interrupted with write lock but not read lock" (me,it) + + | otherwise -> {- 4 -} + err "releaseRead when holding write lock but not read lock" (me,it) + + _ -> do {- 5 -} rc' <- evaluate $ pred rc ret (w { readerCount=rc' }) - {-ditto-} | abandon -> do + {-ditto-} | abandon -> do {- 6 -} q' <- dropReader q ret (w { queue=q' }) - {-ditto-} | otherwise -> + {-ditto-} | otherwise -> {- 7 -} err "releaseRead called when not read locked " me r@(Readers { readerCounts=rcs,queueR=qR }) -> @@ -228,27 +236,30 @@ let rcs' = Map.delete me rcs if Map.null rcs' then case qR of - Nothing -> + Nothing -> {- 8 -} ret FreeLock - Just ((wid,mblock),q) -> do + + Just ((wid,mblock),q) -> do {- 9 -} putMVar mblock () ret (Writer { writerID=wid, writerCount=1, readerCount=0, queue=q }) - else ret (r { readerCounts=rcs' }) - Just rc -> do + else ret (r { readerCounts=rcs' }) {- 10 -} + + Just rc -> do {- 11 -} rc' <- evaluate $ pred rc rcs' <- evaluate $ Map.insert me rc' rcs ret (r { readerCounts=rcs' }) Nothing | abandon -> case qR of - Nothing -> + Nothing -> {- 12 -} impossible "acquireRead interrupted not holding lock and with no queue" (me,rcs) - Just (w,q) -> do + + Just (w,q) -> {- 13 -} do q' <- dropReader q ret (r { queueR = Just (w,q') }) - {-ditto-} | otherwise -> + {-ditto-} | otherwise -> {- 14 -} err "releaseRead called with read lock held by others" (me,rcs) -- | A thread that calls acquireWrite must later call releaseWrite once for each call to acquireWrite. @@ -258,9 +269,9 @@ -- -- This can block but cannot be interrupted. releaseWrite :: RWLock -> IO (Either RWLockException ()) -releaseWrite (RWL rwlVar) = uninterruptibleMask_ $ do +releaseWrite (RWL rwlVar) = mask_ $ do me <- myThreadId - releaseWrite' False me rwlVar -- False to indicate called from releaseWrite + releaseWrite' False me rwlVar -- False to indicate call is from releaseWrite -- Nine non-impossible cases, plus one impossible case -- Lock is Free @@ -298,6 +309,7 @@ case rwd of FreeLock | abandon -> impossible "acquireWrite interrupted with unlocked RWLock" me + | otherwise -> err "cannot releaseWrite lock from unlocked RWLock" me @@ -366,13 +378,13 @@ isReader (ReaderKind {},_) = True isReader _ = False --- Six cases: +-- Six cases below: -- Lock is Free -- I already have write lock --- Someone else has write lock, mblock +-- Someone else has write lock, leads to mblock -- I alread have read lock -- Someone else has read lock, no pending write lock --- Someone else has read lock, there is a pending write lock, mblock +-- Someone else has read lock, there is a pending write lock, leads to mblock -- | Any thread may call acquireRead (even ones holding write locks). This read lock may be -- acquired multiple times, requiring an identical number of releaseRead calls. @@ -397,7 +409,8 @@ rc' <- evaluate $ succ rc return ( w { readerCount=rc' } , return () ) - | otherwise -> do + + {- ditto -} | otherwise -> do (q',mblock) <- enterQueueR q me return ( w { queue = q' } , safeBlock mblock ) @@ -434,6 +447,7 @@ addMe rcs | Set.member me rcs = error (imp "enterQueueR.addMe when already in set" me) | otherwise = return (Set.insert me rcs) +-- Five cases. -- This is not exported. This has uninterruptibleMask_. It is used to restore read locks released -- during acquireWrite when acquireWrite is called while holding read locks. If this acquireWrite -- upgrade is going well then this thread holds the Writer lock and acquireReadPriority is identical @@ -486,13 +500,13 @@ addMe rcs | Set.member me rcs = error (imp "enterQueueL.addMe when already in set" me) | otherwise = return (Set.insert me rcs) --- Six cases: +-- Six cases below: -- Lock is Free -- I already have write lock --- Someone else has write lock +-- Someone else has write lock, leads to waiting -- I already have read lock --- Someone else has read lock, there is no pending write lock --- Someone else has read lock, there is a pending write lock +-- Someone else has read lock, there is no pending write lock, wait +-- Someone else has read lock, there is a pending write lock, wait -- | Any thread may call acquireWrite (even ones holding read locks, but see below for interrupted -- behavior). This write lock may be acquired multiple times, requiring an identical number of @@ -565,7 +579,7 @@ subtle bug #1: -When converting from a read lock holding 'rc' read locks to a also holding a write lock, I first wrote: +When converting from a read lock holding rc > 0 read locks to also holding a write lock, I first wrote: replicateM_ rc (releaseRead rwl >>= either throw return) acquireWrite rwl diff --git a/src/Control/Concurrent/MSem.hs b/src/Control/Concurrent/MSem.hs deleted file mode 100644 --- a/src/Control/Concurrent/MSem.hs +++ /dev/null @@ -1,132 +0,0 @@ -{-# LANGUAGE DeriveDataTypeable #-} --- | --- Module : Control.Concurrent.MSem --- Copyright : (c) Chris Kuklewicz 2011 --- License : 3 clause BSD-style (see the file LICENSE) --- --- Maintainer : haskell@list.mightyreason.com --- Stability : experimental --- Portability : non-portable (concurrency) --- --- A semaphore in which operations may 'wait' for or 'signal' single units of value. This modules --- is intended to improve on "Control.Concurrent.QSem". --- --- This semaphore gracefully handles threads which die while blocked waiting. The fairness --- guarantee is that blocked threads are FIFO. --- --- If 'with' is used to guard a critical section then no quantity of the semaphore will be lost if --- the activity throws an exception. 'new' can initialize the semaphore to negative, zero, or --- positive quantity. 'wait' always leaves the 'MSem' with non-negative quantity. --- --- The functions below are generic in (Integral i) with specialization to Int and Integer. --- --- Overflow warning: These operations do not check for overflow errors. If the Integral type is too --- small to accept the new total then the behavior of these operations is undefined. Using (MSem --- Integer) prevents the possibility of an overflow error. -module Control.Concurrent.MSem - (MSem - ,new - ,with - ,wait - ,signal - ,peekAvail - ) where - -import Control.Monad(join) -import Control.Concurrent.MVar(MVar,withMVar,modifyMVar,modifyMVar_,newMVar,newEmptyMVar,putMVar,takeMVar,tryTakeMVar,tryPutMVar) -import Control.Exception(bracket_,uninterruptibleMask_,mask_) -import Data.Typeable(Typeable) - -{- design notes are in MSemN.hs -} - --- | A 'MSem' is a semaphore in which the available quantity can be added and removed in single --- units, and which can start with positive, zero, or negative value. -data MSem i = MSem { mSem :: !(MVar i) -- ^ Used to lock access to state of semaphore quantity. Never updated. - , queueWait :: !(MVar ()) -- ^ Used as FIFO queue for waiter, held by head of queue. Never updated. - , headWait :: !(MVar ()) -- ^ The head of the waiter queue blocks on headWait. Never updated. - } - deriving (Eq,Typeable) - --- |'new' allows positive, zero, and negative initial values. The initial value is forced here to --- better localize errors. --- --- The only way to acheive a negative value with MSem is to start negative with 'new'. Once the quantity -new :: Integral i => i -> IO (MSem i) -{-# SPECIALIZE new :: Int -> IO (MSem Int) #-} -{-# SPECIALIZE new :: Integer -> IO (MSem Integer) #-} -new initial = do - newMS <- newMVar $! initial - newQueueWait <- newMVar () - newHeadWait <- newEmptyMVar - return (MSem { mSem = newMS - , queueWait = newQueueWait - , headWait = newHeadWait }) - --- | 'with' takes a unit of value from the semaphore to hold while performing the provided --- operation. 'with' ensures the quantity of the sempahore cannot be lost if there are exceptions. --- --- 'with' uses 'bracket_' to ensure 'wait' and 'signal' get called correctly. -with :: Integral i => MSem i -> IO a -> IO a -{-# SPECIALIZE with :: MSem Int -> IO a -> IO a #-} -{-# SPECIALIZE with :: MSem Integer -> IO a -> IO a #-} -with m = bracket_ (wait m) (signal m) - --- |'wait' will take one unit of value from the sempahore, but will block if the quantity available --- is not positive. --- --- If 'wait' returns without interruption then it left the 'MSem' with a remaining quantity that was --- greater than or equal to zero. If 'wait' is interrupted then no quantity is lost. If 'wait' --- returns without interruption then it is known that each earlier waiter has definitely either been --- interrupted or has retured without interruption. -wait :: Integral i => MSem i -> IO () -{-# SPECIALIZE wait :: MSem Int -> IO () #-} -{-# SPECIALIZE wait :: MSem Integer -> IO () #-} -wait m = mask_ . withMVar (queueWait m) $ \ () -> do - join . modifyMVar (mSem m) $ \ ms -> do - mayGrab <- tryTakeMVar (headWait m) - case mayGrab of - Just () -> return (ms,return ()) - Nothing -> if 1 <= ms - then let ms' = pred ms -- ms' is never negative - in seq ms' $ return (ms', return ()) - else return (ms, takeMVar (headWait m)) - -- mask_ is needed above because we may have just decremented 'avail' and we must finished 'wait' - -- without being interrupted so that a 'bracket' can ensure a matching 'signal' can be ensured. - -- - -- join (takeMVar ..) actually may or may not block, a 'signal' could have already arrived or this - -- thread might have an pending throwTo/killThread exception. - --- | 'signal' adds one unit to the sempahore. --- --- 'signal' may block, but it cannot be interrupted, which allows it to dependably restore value to --- the 'MSem'. All 'signal', 'peekAvail', and the head waiter may momentarily block in a fair FIFO --- manner. -signal :: Integral i => MSem i -> IO () -{-# SPECIALIZE signal :: MSem Int -> IO () #-} -{-# SPECIALIZE signal :: MSem Integer -> IO () #-} -signal m = uninterruptibleMask_ . modifyMVar_ (mSem m) $ \ ms -> do - -- mask_ might be as good as uninterruptibleMask_ since nothing below can block - if ms < 0 - then return $! succ ms - else do - didPlace <- tryPutMVar (headWait m) () -- ms is never negative - if didPlace - then return ms - else return $! succ ms - --- | 'peekAvail' skips the queue of any blocked 'wait' threads, but may momentarily block on --- 'signal', other 'peekAvail', and the head waiter. This returns the amount of value available to --- be taken. Using this value without producing unwanted race conditions is left up to the --- programmer. --- --- Note that "Control.Concurrent.MSemN" offers a more powerful API for making decisions based on the --- available amount. -peekAvail :: Integral i => MSem i -> IO i -{-# SPECIALIZE peekAvail :: MSem Int -> IO Int #-} -{-# SPECIALIZE peekAvail :: MSem Integer -> IO Integer #-} -peekAvail m = mask_ $ withMVar (mSem m) $ \ ms -> do - extraFlag <- tryTakeMVar (headWait m) - case extraFlag of - Nothing -> return ms - Just () -> do putMVar (headWait m) () -- cannot block - return $! succ ms diff --git a/src/Control/Concurrent/MSem.lhs b/src/Control/Concurrent/MSem.lhs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/MSem.lhs @@ -0,0 +1,306 @@ +> {-# LANGUAGE DeriveDataTypeable #-} +> -- | +> -- Module : Control.Concurrent.MSem +> -- Copyright : (c) Chris Kuklewicz 2011 +> -- License : 3 clause BSD-style (see the file LICENSE) +> -- +> -- Maintainer : haskell@list.mightyreason.com +> -- Stability : experimental +> -- Portability : non-portable (concurrency) +> -- +> -- This is a literate haskell version of Control.Concurrent.MSem for increased clarity. +> -- +> -- A semaphore in which operations may 'wait' for or 'signal' single units of value. This modules +> -- is intended to improve on "Control.Concurrent.QSem". +> -- +> -- This semaphore gracefully handles threads which die while blocked waiting. The fairness +> -- guarantee is that blocked threads are servied in a FIFO order. +> -- +> -- If 'with' is used to guard a critical section then no quantity of the semaphore will be lost if +> -- the activity throws an exception or if this thread is killed by the rest of the program. +> -- +> -- 'new' can initialize the semaphore to negative, zero, or positive quantity. +> -- 'wait' always leaves the 'MSem' with non-negative quantity. +> -- 'signal' alawys adds one to the quantity. +> -- +> -- The functions below are generic in (Integral i) with specialization to Int, Word, and Integer. +> -- +> -- Overflow warning: These operations do not check for overflow errors. If the Integral type is too +> -- small to accept the new total then the behavior of 'signal' is undefined. Using (MSem +> -- Integer) prevents the possibility of an overflow error. [ A version of 'signal' that checks the upper +> -- bound could be added, but how would it report failure and how would you use this sanely? ] +> -- +> +> module Control.Concurrent.MSem +> (MSem -- do not export the constructor, kept abstract +> , new -- :: Integral i => i -> IO (MSem i) +> , with -- :: Integral i => MSem i -> IO a -> IO a +> , wait -- :: Integral i => MSem i -> IO () +> , signal -- :: Integral i => MSem i -> IO () +> , peekAvail -- :: Integral i => MSem i -> IO i +> ) where + +The above export list shows the API. + +The amount of value in the orignal QSem is always of type Int. This module +generalizes the type to any Integral, where comparison (<) to 'fromIntegral 0' +and 'pred' and 'succ' are employed. + +The 'new', 'wait', and 'signal' operations mimic the QSem API. The peekAvail +query is also provided, primarily for monitoring or debugging purposes. The +with combinator is used to safely and conveniently bracket operations. + +> import Prelude( Integral,Eq,IO,Int,Integer,Maybe(Just,Nothing) +> , seq,pred,succ,return +> , (.),(<),($),($!) ) +> import Control.Concurrent.MVar( MVar +> , withMVar,modifyMVar,modifyMVar_,tryPutMVar +> , newMVar,newEmptyMVar,putMVar,takeMVar,tryTakeMVar) +> import Control.Exception(bracket_,uninterruptibleMask_,mask_) +> import Control.Monad(join) +> import Data.Typeable(Typeable) +> import Data.Word(Word) + +The import list shows that most of the power of MVar's will be exploited, and +that the rather dangerous uninterruptibleMask_ will be employed (in 'signal'). + +A new semaphore is created with a specified avaiable quantity. The mutable +available quantity will be called the value of the semaphore for brevity's +sake. + +The use of a semaphore involves multiple threads executing 'wait' and 'signal' +commands. This stream of wait and 'signal' commands will be executed as if +they arrive in some sequential, non-overlapping, order which is an interleaving +of the commands from each thread. + +From the local perspective of a single thread the semantics are simple to +specify. The 'signal' command will find the MSem to have a value and mutate +this to add one to the value. The 'wait' command will find the MSem to have a +value and if this is greater than zero it will mutate this to be one less and +finish, otherwise the value is negative or zero and the execution of the 'wait' +thread will block. Eventually another thread executes 'signal' and raises the +value to be positive, at this point the blocked 'wait' thread will reduce the +value by one and finish executing the 'wait' command. + +From a broader perspective there is a question of precedence and starvation. +If there is a blocked wait thread and a second 'wait' command starts to execute +then will the second thread "find the MSem to have a value" before or after the +orignal blocked thread has finished? If there are several blocked 'wait' +threads and a 'signal' arrives then which blocked thread has priority to take +the quatity and finish waiting? Are there any fairness guarantees or might a +blocked thread never get priority over its bretheren leading to starvation? + +I have designed this module to provide a fair semaphore: multiple 'wait' +threads are serviced in FIFO order. All 'signal' operations, while they may +block, are individually quick. + +There are precisely three components, all MVars alloced by 'new': queueWait, +quantityStore, and headWait. + +1) The 'wait' operations are forced into a FIFO queue by taking an (MVar ()) +called queueWait during their operation. The thread holding this token is the +"head" waiter. + +2) The 'signal' operations are forced into a FIFO queue by taking the MVar +called quantityStore which holds an integral value. + +3) The logical value stored in the semaphore might be represented by one of two +different states of the semaphore data structure, depending on whether +'headWait :: MVar ()' is empty or full. In this module a full headWait +reprents a single unit of value stored in the semaphore. + +> -- | A 'MSem' is a semaphore in which the available quantity can be added and removed in single +> -- units, and which can start with positive, zero, or negative value. +> data MSem i = MSem { quantityStore :: !(MVar i) -- ^ Used to lock access to state of semaphore quantity. Never updated. +> , queueWait :: !(MVar ()) -- ^ Used as FIFO queue for waiter, held by head of queue. Never updated. +> , headWait :: !(MVar ()) -- ^ The head of the waiter queue blocks on headWait. Never updated. +> } +> deriving (Eq,Typeable) +> +> -- |'new' allows positive, zero, and negative initial values. The initial value is forced here to +> -- better localize errors. +> -- +> -- The only way to achieve a negative value with MSem is to start negative with 'new'. Once a negative quantity becomes non-negative +> -- by use of 'signal' it will never later be negative. +> new :: Integral i => i -> IO (MSem i) +> {-# SPECIALIZE new :: Int -> IO (MSem Int) #-} +> {-# SPECIALIZE new :: Word -> IO (MSem Word) #-} +> {-# SPECIALIZE new :: Integer -> IO (MSem Integer) #-} +> new initial = do +> newQuantityStore <- newMVar $! initial +> newQueueWait <- newMVar () +> newHeadWait <- newEmptyMVar +> return (MSem { quantityStore = newQuantityStore +> , queueWait = newQueueWait +> , headWait = newHeadWait }) +> + +Note that the only MVars that get allocated are all by these three commands in +'new'. The other commands change the stored values but do not allocate new +mutable storage. None of these three MVars can be simply replaced by an IORef +because the possibility of blocking on each of them is used in the design. A +design with two MVar is possible but I think it would have more contention +between threads and be more complex to ensure thread safety. + +There are four operations on the semaphore leading to two possible states for +headWait: + +1) If the most recent operation to finish was 'new' then headWait is definitely +empty and the value of the MSem is the quantity in quantityStore. + +2) If the most recent operation to finish was 'wait' then headWait is +definitely empty and the value of the MSem is the quantity in quantityStore. + +3) If the most recent operation to finish was a 'signal' and the new value is +positive then headWait is definitely full and the value of the MSem is the +quantity in quantityStore PLUS ONE. + +4) If the most recent operation to finish was a 'signal' and the new value is +non-positive then headWait is definitely empty and the value of the MSem is the +quantity in quantityStore. + +If the "head" 'wait' thread finds a non-positive value then it will need to +sleep until being awakened by a future 'signal'. This sleeping is accomplished +by the head waiter taking an empty headWait. + +All uses of the semaphore API to guard execution of an action should use 'with' +to simplify ensuring exceptions are safely handled. Other uses should use +still try and use combinators in Control.Exception to ensure that no 'signal' +commands get lost so that no quantity of the semaphore leaks when exceptions +occur. + +> -- | 'with' takes a unit of value from the semaphore to hold while performing the provided +> -- operation. 'with' ensures the quantity of the sempahore cannot be lost if there are exceptions or +> -- if killThread is used. +> -- +> -- 'with' uses 'bracket_' to ensure 'wait' and 'signal' get called correctly. +> with :: Integral i => MSem i -> IO a -> IO a +> {-# SPECIALIZE with :: MSem Int -> IO a -> IO a #-} +> {-# SPECIALIZE with :: MSem Word -> IO a -> IO a #-} +> {-# SPECIALIZE with :: MSem Integer -> IO a -> IO a #-} +> with m = bracket_ (wait m) (signal m) + +> -- |'wait' will take one unit of value from the sempahore, but will block if the quantity available +> -- is not positive. +> -- +> -- If 'wait' returns normally (not interrupted) then it left the 'MSem' with a remaining quantity that was +> -- greater than or equal to zero. If 'wait' is interrupted then no quantity is lost. If 'wait' +> -- returns without interruption then it is known that each earlier waiter has definitely either been +> -- interrupted or has retured without interruption (the FIFO guarantee). +> wait :: Integral i => MSem i -> IO () +> {-# SPECIALIZE wait :: MSem Int -> IO () #-} +> {-# SPECIALIZE wait :: MSem Word -> IO () #-} +> {-# SPECIALIZE wait :: MSem Integer -> IO () #-} +> wait m = mask_ . withMVar (queueWait m) $ \ () -> do +> join . modifyMVar (quantityStore m) $ \ quantity -> do +> mayGrab <- tryTakeMVar (headWait m) -- First try optimistic grab on (headWait w) +> case mayGrab of +> Just () -> return (quantity,return ()) -- Took unit of value, done +> Nothing -> if 0 < quantity -- Did not take unit of value, check quantity +> then let quantity' = pred quantity -- quantity' is never negative +> in seq quantity' $ return (quantity', return ()) +> else return (quantity, takeMVar (headWait m)) -- go to sleep + +The needed invariant is that 'wait' takes a unit of value iff it returns +normally (i.e. it is not interrupted). The 'mask_' is needed above because we +may decrement 'headWait' with 'tryTakeMVar' and must then finished the +'withMVar' without being interrupted. Under the 'mask_' the 'wait' might block +and then be interruptable at one or more of + +1) 'withMVar (queueWait m)' : the 'wait' dies before becoming head waiter while +blocked by previous 'wait'. + +2) 'modifyMVar (quantityStore m)' : the 'wait' dies as head waiter while +blocked by previous 'signal'. + +3) 'takeMVar (headWait m)' from 'join' : the 'wait' dies as head waiter while +sleeping on 'headWait'. + +All three of those are safe places to die. The unsafe possibilities would be +to die after a 'tryTakeMVar (headWait m)' returns 'Just ()' or after +'modifyMVar' puts the decremented quantity into (quantityStore m). These are +prevented by the 'mask_'. + +Note that the head waiter must also get to the front of the FIFO queue of +signals to get the value of 'quantityStore'. Only the head waiter competes +with the 'signal' & peek threads for obtaining 'quantityStore'. + +> -- | 'signal' adds one unit to the sempahore. Overflow is not checked. +> -- +> -- 'signal' may block, but it cannot be interrupted, which allows it to dependably restore value to +> -- the 'MSem'. All 'signal', 'peekAvail', and the head waiter may momentarily block in a fair FIFO +> -- manner. +> signal :: Integral i => MSem i -> IO () +> {-# SPECIALIZE signal :: MSem Int -> IO () #-} +> {-# SPECIALIZE signal :: MSem Word -> IO () #-} +> {-# SPECIALIZE signal :: MSem Integer -> IO () #-} +> signal m = uninterruptibleMask_ . modifyMVar_ (quantityStore m) $ \ quantity -> do +> if quantity < 0 +> then return $! succ quantity +> else do +> didPlace <- tryPutMVar (headWait m) () -- quantity is never negative +> if didPlace +> then return quantity +> else return $! succ quantity + +The 'signal' operation first has the FIFO grab of (quantityStore m). If +'tryPutMVar' returns True then a currently sleeping head waiter will be woken +up. + +The 'modifyMVar_' will block until prior 'signal' and 'peek' threads and +perhaps a prior head 'wait' finish. This is the only point that may block. +Thus 'uninterruptibleMask_' only differs from 'mask_' in that once 'signal' +starts executing it cannot be interrupted before returning the unit of value to +the MSem. All the operations 'signal' would be waiting for are quick and are +themselves non-blocking, so the uninterruptible operation here should finish +without arbitrary delay. + +Consider 'with m act = bracket_ (wait m) (signal m) act', refer to +http://www.haskell.org/ghc/docs/latest/html/libraries/base/src/Control-Exception-Base.html#bracket_ +for the details. Specifically a killThread arrives at one of these points: + +1) during (wait m) the exception is masked by both 'bracket' and 'wait' so this +occurs at one of the blocking points mentioned above. This does not affect the +MSe, and aborts the 'bracket_' without calling act or (signal m). + +2) during (restore act) the `onException` in the definition of 'bracket' will +shift control to (signal m). + +3) during (signal m) regardless of how act exited. Here we know (wait m) +exited normally and thus took a unit of value from the MSem. The mask_ of +'bracket' ensures that the uninterruptibleMask_ in 'signal' ensures that the +unit of value is returned to MSem even if 'signal' blocks on 'modifyMVar_ +(quantityStore m)'. + +4) Outside of any of the above the mask_ in 'bracket' prevents the killThread +from being recognized until one of the above or until the 'bracket' finishes. + +If 'signal' did not use 'uninterruptibleMask_' then point (3) could be +interrupted without returning the value to the MSem. Avoiding losing quantity +is the primary design criterion for this semaphore library, and I think it +requires this apparantly safe use of uninterruptibleMask_ to ensure that +'signal' can and will succeed. + +> -- | 'peekAvail' skips the queue of any blocked 'wait' threads, but may momentarily block on +> -- 'signal', other 'peekAvail', and the head waiter. This returns the amount of value available to +> -- be taken. Using this value without producing unwanted race conditions is left up to the +> -- programmer. +> -- +> -- Note that "Control.Concurrent.MSemN" offers a more powerful API for making decisions based on the +> -- available amount. +> peekAvail :: Integral i => MSem i -> IO i +> {-# SPECIALIZE peekAvail :: MSem Int -> IO Int #-} +> {-# SPECIALIZE peekAvail :: MSem Word -> IO Word #-} +> {-# SPECIALIZE peekAvail :: MSem Integer -> IO Integer #-} +> peekAvail m = mask_ $ withMVar (quantityStore m) $ \ quantity -> do +> extraFlag <- tryTakeMVar (headWait m) +> case extraFlag of +> Nothing -> return quantity +> Just () -> do putMVar (headWait m) () -- cannot block +> return $! succ quantity + +The implementaion of peekAvail is slightly complicated by the interplay of +tryTakeMVar and putMVar. Only this thread will be holding the lock on +quantityStore and the putMVar only runs to put a () just taken from headWait. +Thus the putMVar will never block. The 'mask_' ensures that there can be no +external interruption between a tryTakeMVar and putMVar. diff --git a/src/Control/Concurrent/MSemN.hs b/src/Control/Concurrent/MSemN.hs deleted file mode 100644 --- a/src/Control/Concurrent/MSemN.hs +++ /dev/null @@ -1,223 +0,0 @@ -{-# LANGUAGE DeriveDataTypeable #-} --- | --- Module : Control.Concurrent.MSemN --- Copyright : (c) Chris Kuklewicz 2011 --- License : 3 clause BSD-style (see the file LICENSE) --- --- Maintainer : haskell@list.mightyreason.com --- Stability : experimental --- Portability : non-portable (concurrency) --- --- Quantity semaphores in which each thread may wait for an arbitrary amount. This modules is --- intended to improve on "Control.Concurrent.QSemN". --- --- This semaphore gracefully handles threads which die while blocked waiting for quantity. The --- fairness guarantee is that blocked threads are FIFO. An early thread waiting for a large --- quantity will prevent a later thread waiting for a small quantity from jumping the queue. --- --- If 'with' is used to guard a critical section then no quantity of the semaphore will be lost --- if the activity throws an exception. --- --- The functions below are generic in (Integral i) with specialization to Int and Integer. --- --- Overflow warning: These operations do not check for overflow errors. If the Integral type is too --- small to accept the new total then the behavior of these operations is undefined. Using (MSem --- Integer) prevents the possibility of an overflow error. -module Control.Concurrent.MSemN - (MSemN - ,new - ,with - ,wait - ,signal - ,withF - ,waitF - ,signalF - ,peekAvail - ) where - -import Control.Monad(when) -import Control.Concurrent.MVar(MVar,withMVar,modifyMVar,modifyMVar_,newMVar,newEmptyMVar,putMVar,takeMVar,tryTakeMVar) -import Control.Exception(bracket,uninterruptibleMask_,onException,evaluate,mask_) -import Data.Typeable(Typeable) - -{- - -The only MVars allocated are the three created be 'new'. Their three roles are -1) to have a FIFO queue of waiters -2) for the head waiter to block on -3) to protect the quantity state of the semaphore and the head waiter - -subtle design notes: - -with, wait, and signal pattern match the quantity against 0 which has two effect: it avoids locking -in the easy case and it ensures strict evaluation of the quantity before any locks are taken. - -Originally withF, waitF, and signal did not strictly evalaute the function they are passed before -locks are taken because there is no real point since the function may throw an error when computing -the size. But then I realized forcing 'f' might run forever with the locks held and I could move -this particular hang outside the locks by first evaluating 'f'. - --} - --- MS has an invariant that "maybe True (> avail) headWants" is always True. -data MS i = MS { avail :: !i -- ^ This is the quantity available to be taken from the semaphore. - , headWants :: !(Maybe i) -- ^ If there is waiter then this is Just the amount being waited for. - } - deriving (Eq,Typeable) - --- | A 'MSemN' is a quantity semaphore, in which the available quantity may be signalled or --- waited for in arbitrary amounts. -data MSemN i = MSemN { mSem :: !(MVar (MS i)) -- ^ Used to lock access to state of semaphore quantity. - , queueWait :: !(MVar ()) -- ^ Used as FIFO queue for waiter, held by head of queue. - , headWait :: !(MVar ()) -- ^ The head of the waiter queue blocks on headWait. - } - deriving (Eq,Typeable) - --- |'new' allows positive, zero, and negative initial values. The initial value is forced here to --- better localize errors. -new :: Integral i => i -> IO (MSemN i) -{-# SPECIALIZE new :: Int -> IO (MSemN Int) #-} -{-# SPECIALIZE new :: Integer -> IO (MSemN Integer) #-} -new initial = do - newMS <- newMVar $! (MS { avail = initial - , headWants = Nothing }) - newQueueWait <- newMVar () - newHeadWait <- newEmptyMVar - return (MSemN { mSem = newMS - , queueWait = newQueueWait - , headWait = newHeadWait }) - --- | 'with' takes a quantity of the semaphore to take and hold while performing the provided --- operation. 'with' ensures the quantity of the sempahore cannot be lost if there are exceptions. --- This uses 'bracket' to ensure 'wait' and 'signal' get called correctly. -with :: Integral i => (MSemN i) -> i -> IO a -> IO a -{-# SPECIALIZE with :: MSemN Int -> Int -> IO a -> IO a #-} -{-# SPECIALIZE with :: MSemN Integer -> Integer -> IO a -> IO a #-} -with _ 0 = id -with m wanted = bracket (wait m wanted) (\() -> signal m wanted) . const - --- | 'withF' takes a pure function and an operation. The pure function converts the available --- quantity to a pair of the wanted quantity and a returned value. The operation takes the result --- of the pure function. 'withF' ensures the quantity of the sempahore cannot be lost if there --- are exceptions. This uses 'bracket' to ensure 'waitF' and 'signal' get called correctly. --- --- Note: A long running pure function will block all other access to the 'MSemN' while it is --- evaluated. -withF :: Integral i => (MSemN i) -> (i -> (i,b)) -> ((i,b) -> IO a) -> IO a -{-# SPECIALIZE withF :: MSemN Int -> (Int -> (Int,b)) -> ((Int,b) -> IO a) -> IO a #-} -{-# SPECIALIZE withF :: MSemN Integer -> (Integer -> (Integer,b)) -> ((Integer,b) -> IO a) -> IO a #-} -withF m f = bracket (waitF m f) (\(wanted,_) -> signal m wanted) - --- |'wait' allow positive, zero, and negative wanted values. Waiters may block, and will be handled --- fairly in FIFO order. --- --- If 'wait' returns without interruption then it left the 'MSemN' with a remaining quantity that was --- greater than or equal to zero. If 'wait' is interrupted then no quantity is lost. If 'wait' --- returns without interruption then it is known that each earlier waiter has definitely either been --- interrupted or has retured without interruption. -wait :: Integral i => (MSemN i) -> i -> IO () -{-# SPECIALIZE wait :: MSemN Int -> Int -> IO () #-} -{-# SPECIALIZE wait :: MSemN Integer -> Integer -> IO () #-} -wait _ 0 = return () -wait m wanted = fmap snd $ waitF m (const (wanted,())) - --- | 'waitWith' takes the 'MSemN' and a pure function that takes the available quantity and computes the --- amount wanted and a second value. The value wanted is stricly evaluated but the second value is --- returned lazily. --- --- 'waitF' allow positive, zero, and negative wanted values. Waiters may block, and will be handled --- fairly in FIFO order. --- --- If 'waitF' returns without interruption then it left the 'MSemN' with a remaining quantity that was --- greater than or equal to zero. If 'waitF' or the provided function are interrupted then no --- quantity is lost. If 'waitF' returns without interruption then it is known that each previous --- waiter has each definitely either been interrupted or has retured without interruption. --- --- Note: A long running pure function will block all other access to the 'MSemN' while it is --- evaluated. -waitF :: Integral i => (MSemN i) -> (i -> (i,b)) -> IO (i,b) -{-# SPECIALIZE waitF :: MSemN Int -> (Int -> (Int,b)) -> IO (Int,b) #-} -{-# SPECIALIZE waitF :: MSemN Integer -> (Integer -> (Integer,b)) -> IO (Integer,b) #-} -waitF m f = mask_ . withMVar (queueWait m) $ \ () -> do - (out@(wanted,_),mustWait) <- modifyMVar (mSem m) $ \ ms -> do - let outVal@(wantedVal,_) = f (avail ms) - -- assert that headDown is Nothing (from prior 'new' or 'signal' or 'cleanup') - -- wantedVal gets forced by the (<=) condition here: - if wantedVal <= avail ms - then do - let avail'down = avail ms - wantedVal -- avail'down is never negative, barring overflow - ms' <- evaluate ms { avail = avail'down } - return (ms', (outVal,False)) - else do - ms' <- evaluate ms { headWants = Just wantedVal } - return (ms', (outVal,True)) - -- mask_ is needed above because either (Just wantedVal) may be set here and this means we need to - -- get the `onException` setup without being interrupted, or avail'down was set and we must finish - -- 'waitF' without being interrupted so that a 'bracket' can ensure a matching 'signal' can - -- protect the returned quantity. - when mustWait $ do - let cleanup = uninterruptibleMask_ $ modifyMVar_ (mSem m) $ \ms -> do - mStale <- tryTakeMVar (headWait m) - let avail' = avail ms + maybe 0 (const wanted) mStale - evaluate ms {avail = avail', headWants = Nothing} - takeMVar (headWait m) `onException` cleanup -- may not block if a 'signal' or exception has already arrived. - return out - --- |'signal' allows positive, zero, and negative values, thus this is also way to remove quantity --- that skips any threads in the 'wait'/'waitF' queue. If the new total is greater than the next --- value being waited for (if present) then the first waiter is woken. If there are queued waiters --- then the next one will wake after a waiter has proceeded and notice the remaining value; thus a --- single 'signal' may result in several waiters obtaining values. Waking waiting threads is --- asynchronous. --- --- 'signal' may block, but it cannot be interrupted, which allows it to dependably restore value to --- the 'MSemN'. All 'signal', 'signalF', 'peekAvail', and the head waiter may momentarily block in a --- fair FIFO manner. -signal :: Integral i => (MSemN i) -> i -> IO () -{-# SPECIALIZE signal :: MSemN Int -> Int -> IO () #-} -{-# SPECIALIZE signal :: MSemN Integer -> Integer -> IO () #-} -signal _ 0 = return () -signal m size = uninterruptibleMask_ $ fmap snd $ signalF m (const (size,())) - --- | Instead of providing a fixed change to the available quantity, 'signalF' applies a provided --- pure function to the available quantity to compute the change and a second value. The --- requested change is stricly evaluated but the second value is returned lazily. If the new total is --- greater than the next value being waited for then the first waiter is woken. If there are queued --- waiters then the next one will wake after a waiter has proceeded and notice the remaining value; --- thus a single 'signalF' may result in several waiters obtaining values. Waking waiting threads --- is asynchronous. --- --- 'signalF' may block, and it can be safely interrupted. If the provided function throws an error --- or is interrupted then it leaves the 'MSemN' unchanged. All 'signal', 'signalF', 'peekAvail', and --- the head waiter may momentarily block in a fair FIFO manner. --- --- Note: A long running pure function will block all other access to the 'MSemN' while it is --- evaluated. -signalF :: Integral i => (MSemN i) -> (i -> (i,b)) -> IO (i,b) -{-# SPECIALIZE signalF :: MSemN Int -> (Int -> (Int,b)) -> IO (Int,b) #-} -{-# SPECIALIZE signalF :: MSemN Integer -> (Integer -> (Integer,b)) -> IO (Integer,b) #-} -signalF m f = mask_ . modifyMVar (mSem m) $ \ ms -> do - -- Nothing below blocks, not even the putMVar - let out@(size,_) = f (avail ms) - avail' <- evaluate $ avail ms + size -- this forces 'size' - ms' <- case headWants ms of - Just wanted | wanted <= avail' -> do - putMVar (headWait m) () - let avail'down = avail' - wanted -- avail'down is never negative, barring overflow - evaluate ms { avail = avail'down, headWants = Nothing } - _ -> evaluate ms { avail = avail' } - return (ms',out) - --- | 'peekAvail' skips the queue of any blocked 'wait' and 'waitF' threads, but may momentarily --- block on 'signal', 'signalF', other 'peekAvail', and the head waiter. This returns the amount of --- value available to be taken. Using this value without producing unwanted race conditions is left --- up to the programmer. --- --- 'peekAvail' is an optimized form of \"signalF m (\x -> (0,x))\". --- --- A version of 'peekAvail' that joins the FIFO queue of 'wait' and 'waitF' can be acheived by --- \"waitF m (\x -> (0,x))\" -peekAvail :: Integral i => (MSemN i) -> IO i -{-# SPECIALIZE peekAvail :: MSemN Int -> IO Int #-} -{-# SPECIALIZE peekAvail :: MSemN Integer -> IO Integer #-} -peekAvail m = withMVar (mSem m) (return . avail) diff --git a/src/Control/Concurrent/MSemN.lhs b/src/Control/Concurrent/MSemN.lhs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/MSemN.lhs @@ -0,0 +1,319 @@ +> {-# LANGUAGE DeriveDataTypeable #-} +> -- | +> -- Module : Control.Concurrent.MSemN +> -- Copyright : (c) Chris Kuklewicz 2011 +> -- License : 3 clause BSD-style (see the file LICENSE) +> -- +> -- Maintainer : haskell@list.mightyreason.com +> -- Stability : experimental +> -- Portability : non-portable (concurrency) +> -- +> -- Quantity semaphores in which each thread may wait for an arbitrary amount. This modules is +> -- intended to improve on "Control.Concurrent.QSemN". +> -- +> -- This semaphore gracefully handles threads which die while blocked waiting for quantity. The +> -- fairness guarantee is that blocked threads are FIFO. An early thread waiting for a large +> -- quantity will prevent a later thread waiting for a small quantity from jumping the queue. +> -- +> -- If 'with' is used to guard a critical section then no quantity of the semaphore will be lost +> -- if the activity throws an exception. +> -- +> -- The functions below are generic in (Integral i) with specialization to Int and Integer. +> -- +> -- Overflow warning: These operations do not check for overflow errors. If the Integral type is too +> -- small to accept the new total then the behavior of these operations is undefined. Using (MSem +> -- Integer) prevents the possibility of an overflow error. +> +> module Control.Concurrent.MSemN +> (MSemN +> ,new +> ,with +> ,wait +> ,signal +> ,withF +> ,waitF +> ,signalF +> ,peekAvail +> ) where +> +> import Prelude( Integral,Eq,IO,Int,Integer,Maybe(Just,Nothing),Num((+),(-)),Bool(False,True) +> , return,id,const,fmap,snd,maybe,seq +> , (.),(<=),($),($!) ) +> import Control.Concurrent.MVar( MVar +> , withMVar,modifyMVar,modifyMVar_,newMVar +> , newEmptyMVar,tryPutMVar,takeMVar,tryTakeMVar ) +> import Control.Exception(bracket,bracket_,uninterruptibleMask_,onException,evaluate,mask_) +> import Control.Monad(when) +> import Data.Typeable(Typeable) +> import Data.Word(Word) +> + +The only MVars allocated are the three created be 'new'. Their three roles are +1) to have a FIFO queue of waiters (queueWait) +2) for the head waiter to block on, if necessary (headWait) +3) to protect the actual state of the semaphore (quantityStore) + +> -- MS has an invariant that "maybe True (> avail) headWants" is always True. +> data MS i = MS { avail :: !i -- ^ This is the quantity available to be taken from the semaphore. +> , headWants :: !(Maybe i) -- ^ If there is waiter then this is Just the amount being waited for. +> } +> deriving (Eq,Typeable) + +> -- | A 'MSemN' is a quantity semaphore, in which the available quantity may be signalled or +> -- waited for in arbitrary amounts. +> data MSemN i = MSemN { quantityStore :: !(MVar (MS i)) -- ^ Used to lock access to state of semaphore quantity. +> , queueWait :: !(MVar ()) -- ^ Used as FIFO queue for waiter, held by head of queue. +> , headWait :: !(MVar ()) -- ^ The head of the waiter queue blocks on headWait. +> } +> deriving (Eq,Typeable) + +The data structure for 'MSemN' is slightly more complicated than the one in 'MSem'. Here the +quantityStore holds not just a value of type 'i' but also a 'Maybe i' called 'headWants'. + +'headWants' is Nothing when there are no blocked threads waiting on quantity. 'headWants' is (Just +x) when there is at least one blocked thread and the head of the queue needs positive quantity x to +proceed. + +There are two possible lifecycles of a wait request. Like in MSem, all waiters do all work while +holding queueWait. This is what forces the waiters into a FIFO order. + +The first is when the waiter gets to head of the queue and finds that the quantityStore has enough +in 'avail' to be satisfied. This waiter subtracts its wanted value from 'avail' and returns. + +The second is when the waiter does not find a larger enough value in 'avail' must block. It sets +headWants from Nothing to 'Just wanted' and then releases quantityStore, followed by blocked in +headWait. When a signal arrives that puts the available quantity above the value in 'headWants' +then it puts () into 'headWait' to wake the blocked waiting thread. Here the subtraction of the +value in 'Just wanted' from the available quantity is handled by the signalling thread. + +The difficulty is maintaining the desired invariants in the face of exceptions. If a frustrated +waiter dies before the 'takeMVar' on 'headWait' succeeds then the waiter's changes to +'quantityStore' must be undone! This requires the 'uninterruptibleMask_' around the onException +action in 'waitF'. + +When the head waiter releases the queueWait MVar, either by succeeding or being interrupted, there +are three invariants: + +(wait invariant 1) The headWait MVar must be empty. + +(wait invariant 2) The headWants value is Nothing. + +This means that when a waiter first acquires the queueWait MVar both the above hold. If the waiter +succeeded then there is a progress invariant: + +(wait progress invariant) The value of 'avail' is non-negative when wait succeeds. + +When the signal operation release the quantityStore MVar then one of three situations holds: + +(signal possibility 1) headWants was Nothing and it and headWait are unchanged, or + +(signal possibility 2) headWants was (Just x) and it and headWait are unchanged, or + +(signal possibility 3) headWants was (Just x) and is changed to Nothing and headWait has () put into it. + +If headWait had () put into it then headWants is Nothing. The only way headWants can change back to +(Just x) is if a new waiter does it. This requires the original waiter to hand over the queueWait +MVar, and we can be certain that (wait invariant 1) means that the () put into headWait is taken out +before this handoff. + +Thus when a signal first acquires the quantityStore MVar there is a dynamically maintained invariant: + +(signal invariant 1) A signal that finds headWants of (Just x) also finds headWait empty. + +Note that a () put into headWait signifies amount: it is worth the quantity x in the (Just x) in +headWants that was just changed to Nothing. After (signal possibility 3) only the receiving waiting +thread knows the amount that this () in headWait represents, and only this thread can fix the MSemN +if an exception occurs. The waitF function below is careful to fix MSemN. + +> -- |'new' allows positive, zero, and negative initial values. The initial value is forced here to +> -- better localize errors. +> new :: Integral i => i -> IO (MSemN i) +> {-# SPECIALIZE new :: Int -> IO (MSemN Int) #-} +> {-# SPECIALIZE new :: Word -> IO (MSemN Word) #-} +> {-# SPECIALIZE new :: Integer -> IO (MSemN Integer) #-} +> new initial = do +> newMS <- newMVar $! (MS { avail = initial -- this forces initial +> , headWants = Nothing }) +> newQueueWait <- newMVar () +> newHeadWait <- newEmptyMVar +> return (MSemN { quantityStore = newMS +> , queueWait = newQueueWait +> , headWait = newHeadWait }) +> +> -- | 'with' takes a quantity of the semaphore to take and hold while performing the provided +> -- operation. 'with' ensures the quantity of the sempahore cannot be lost if there are exceptions. +> -- This uses 'bracket' to ensure 'wait' and 'signal' get called correctly. +> with :: Integral i => MSemN i -> i -> IO a -> IO a +> {-# SPECIALIZE with :: MSemN Int -> Int -> IO a -> IO a #-} +> {-# SPECIALIZE with :: MSemN Word -> Word -> IO a -> IO a #-} +> {-# SPECIALIZE with :: MSemN Integer -> Integer -> IO a -> IO a #-} +> with m wanted = seq wanted $ bracket_ (wait m wanted) (signal m wanted) +> +> -- | 'withF' takes a pure function and an operation. The pure function converts the available +> -- quantity to a pair of the wanted quantity and a returned value. The operation takes the result +> -- of the pure function. 'withF' ensures the quantity of the sempahore cannot be lost if there +> -- are exceptions. This uses 'bracket' to ensure 'waitF' and 'signal' get called correctly. +> -- +> -- Note: A long running pure function will block all other access to the 'MSemN' while it is +> -- evaluated. +> withF :: Integral i +> => MSemN i +> -> (i -> (i,b)) +> -> ((i,b) -> IO a) +> -> IO a +> {-# SPECIALIZE withF :: MSemN Int -> (Int -> (Int,b)) -> ((Int,b) -> IO a) -> IO a #-} +> {-# SPECIALIZE withF :: MSemN Word -> (Word -> (Word,b)) -> ((Word,b) -> IO a) -> IO a #-} +> {-# SPECIALIZE withF :: MSemN Integer -> (Integer -> (Integer,b)) -> ((Integer,b) -> IO a) -> IO a #-} +> withF m f = bracket (waitF m f) (\(wanted,_) -> signal m wanted) +> +> -- |'wait' allow positive, zero, and negative wanted values. Waiters may block, and will be handled +> -- fairly in FIFO order. Waiters will succeed when the wanted value is less than or equal to the +> -- available value. The FIFO order means that a 'wait' for a large quantity that blocks will prevent later +> -- requests from being considered even if the later requests would be for a small quantity that could be fulfilled. +> -- +> -- If 'wait' returns without interruption then it left the 'MSemN' with a remaining quantity that was +> -- greater than or equal to zero. If 'wait' is interrupted then no quantity is lost. If 'wait' +> -- returns without interruption then it is known that each earlier waiter has definitely either been +> -- interrupted or has retured without interruption. +> wait :: Integral i => MSemN i -> i -> IO () +> {-# SPECIALIZE wait :: MSemN Int -> Int -> IO () #-} +> {-# SPECIALIZE wait :: MSemN Word -> Word -> IO () #-} +> {-# SPECIALIZE wait :: MSemN Integer -> Integer -> IO () #-} +> wait m wanted = seq wanted $ fmap snd $ waitF m (const (wanted,())) +> +> -- | 'waitWith' takes the 'MSemN' and a pure function that takes the available quantity and computes the +> -- amount wanted and a second value. The value wanted is stricly evaluated but the second value is +> -- returned lazily. +> -- +> -- 'waitF' allow positive, zero, and negative wanted values. Waiters may block, and will be handled +> -- fairly in FIFO order. Waiters will succeed when the wanted value is less than or equal to the +> -- available value. The FIFO order means that a 'waitF' for a large quantity that blocks will prevent later +> -- requests from being considered even if the later requests would be for a small quantity that could be fulfilled. +> -- +> -- If 'waitF' returns without interruption then it left the 'MSemN' with a remaining quantity that was +> -- greater than or equal to zero. If 'waitF' or the provided function are interrupted then no +> -- quantity is lost. If 'waitF' returns without interruption then it is known that each previous +> -- waiter has each definitely either been interrupted or has retured without interruption. +> -- +> -- Note: A long running pure function will block all other access to the 'MSemN' while it is +> -- evaluated. +> waitF :: Integral i => MSemN i -> (i -> (i,b)) -> IO (i,b) +> {-# SPECIALIZE waitF :: MSemN Int -> (Int -> (Int,b)) -> IO (Int,b) #-} +> {-# SPECIALIZE waitF :: MSemN Word -> (Word -> (Word,b)) -> IO (Word,b) #-} +> {-# SPECIALIZE waitF :: MSemN Integer -> (Integer -> (Integer,b)) -> IO (Integer,b) #-} +> waitF m f = seq f $ mask_ . withMVar (queueWait m) $ \ () -> do +> -- Assume when queueWait taken: (headWait is empty) AND (headWants is Nothing) +> (out@(wanted,_),mustWait) <- modifyMVar (quantityStore m) $ \ ms -> do +> -- Nothing in this scope can block +> let outVal@(wantedVal,_) = f (avail ms) +> -- assert that headDown is Nothing (from prior 'new' or 'signal' or 'cleanup') +> -- wantedVal gets forced by the (<=) condition here: +> if wantedVal <= avail ms +> then do +> let avail'down = avail ms - wantedVal -- avail'down is never negative, barring overflow +> ms' <- evaluate ms { avail = avail'down } +> return (ms', (outVal,False)) +> else do +> ms' <- evaluate ms { headWants = Just wantedVal } +> return (ms', (outVal,True)) +> -- quantityStore is now released, queueWait is still held, race with signal now possible +> -- Assert: (headWait is empty) AND (mustWait == (headWants is Just)) at release (point X) +> -- Proof: (headWait is empty) was assumed and is unchanged, and +> -- either mustWait is False and assumed (headWants is Nothing) is unchanged, +> -- or mustWait is True and headWants was set to Just wantedVal +> when mustWait $ do +> let cleanup = uninterruptibleMask_ $ modifyMVar_ (quantityStore m) $ \ms -> do +> recovered <- tryTakeMVar (headWait m) +> let total = avail ms + maybe 0 (const wanted) recovered +> evaluate MS {avail = total, headWants = Nothing} +> takeMVar (headWait m) `onException` cleanup -- takeMVar might not block if a 'signal' or exception has already arrived. +> return out +> -- Invariant when queueWait released: (headWait is empty) AND (headWants is Nothing) +> -- Proof: 1) mustWait is false, so (headWants is Just) was false +> -- so (headWait is empty) AND (headWants is Nothing) was true at (point X) +> -- by LEMMA under signalF this is unchanged by signalF; there has been no race condition +> -- 2) mustWait is true, so (headWants is Just) was true +> -- 2a) takeMVar succeeded so headWait became full since (point X) +> -- this implies signal filled headWait and thus signal ended with (headWait is full) +> -- signal invariant ((headWait is empty) OR (headWants is Nothing)) implies (headWants is Nothing) was set +> -- (headWait is empty) by takeMVar and (headWants is Nothing) by implication +> -- 2b) takeMVar was interrupted, then onException ran cleanup, by uninterruptibleMask_ it succeeded +> -- cleanup's tryTakeMVar ensured (headWait is empty), and +> -- cleanup's modifyMVar_ ensured (headWants is Nothing) + +> +> -- |'signal' allows positive, zero, and negative values, thus this is also way to remove quantity +> -- that skips any threads in the 'wait'/'waitF' queue. If the new total is greater than the next +> -- value being waited for (if present) then the first waiter is woken. If there are queued waiters +> -- then the next one will wake after a waiter has proceeded and notice the remaining value; thus a +> -- single 'signal' may result in several waiters obtaining values. Waking waiting threads is +> -- asynchronous. +> -- +> -- 'signal' may block, but it cannot be interrupted, which allows it to dependably restore value to +> -- the 'MSemN'. All 'signal', 'signalF', 'peekAvail', and the head waiter may momentarily block in a +> -- fair FIFO manner. +> signal :: Integral i => MSemN i -> i -> IO () +> {-# SPECIALIZE signal :: MSemN Int -> Int -> IO () #-} +> {-# SPECIALIZE signal :: MSemN Word -> Word -> IO () #-} +> {-# SPECIALIZE signal :: MSemN Integer -> Integer -> IO () #-} +> signal _ 0 = return () -- this also forces 'size' +> signal m size = uninterruptibleMask_ $ fmap snd $ signalF m (const (size,())) +> +> -- | Instead of providing a fixed change to the available quantity, 'signalF' applies a provided +> -- pure function to the available quantity to compute the change and a second value. The +> -- requested change is stricly evaluated but the second value is returned lazily. If the new total is +> -- greater than the next value being waited for then the first waiter is woken. If there are queued +> -- waiters then the next one will wake after a waiter has proceeded and notice the remaining value; +> -- thus a single 'signalF' may result in several waiters obtaining values. Waking waiting threads +> -- is asynchronous. +> -- +> -- 'signalF' may block, and it can be safely interrupted. If the provided function throws an error +> -- or is interrupted then it leaves the 'MSemN' unchanged. All 'signal', 'signalF', 'peekAvail', and +> -- the head waiter may momentarily block in a fair FIFO manner. +> -- +> -- Note: A long running pure function will block all other access to the 'MSemN' while it is +> -- evaluated. +> signalF :: Integral i +> => MSemN i +> -> (i -> (i,b)) +> -> IO (i,b) +> {-# SPECIALIZE signalF :: MSemN Int -> (Int -> (Int,b)) -> IO (Int,b) #-} +> {-# SPECIALIZE signalF :: MSemN Word -> (Word -> (Word,b)) -> IO (Word,b) #-} +> {-# SPECIALIZE signalF :: MSemN Integer -> (Integer -> (Integer,b)) -> IO (Integer,b) #-} +> signalF m f = seq f $ mask_ . modifyMVar (quantityStore m) $ \ ms -> do +> -- Assume: ((headWait is empty) OR (headWants is Nothing)) +> -- Nothing below can block +> let out@(size,_) = f (avail ms) +> total <- evaluate $ avail ms + size -- this forces 'size' +> ms' <- case headWants ms of +> Just wanted | wanted <= total -> do +> -- Assumption implies headWait is empty, using putMVar below would never block +> didPlace <- tryPutMVar (headWait m) () +> evaluate $ if didPlace +> then MS { avail = total - wanted, headWants = Nothing } -- always this case +> else MS { avail = total, headWants = Nothing } -- impossible case +> _ -> evaluate ms { avail = total } +> return (ms',out) +> -- Invariant: ((headWait is empty) OR (headWants is Nothing)) +> -- Proof: 1) originally (headWants is Nothing), headWait and headWants unchanged, invariant still holds +> -- 2) orignal (Just wanted) +> -- 2a) wanted <= total, headWait becomes filled and headWants becomes Nothing, invariant holds +> -- 2b) wanted > total, headWait and headWants unchanged, invariant still holds +> +> -- LEMMA: if (headWait is empty) AND (headWants is Nothing) holds before signalF then it holds after signalF +> -- Proof: When (headWants is Nothing) both headWait and headWants are unchanged (proof case 1 above) + +> -- | 'peekAvail' skips the queue of any blocked 'wait' and 'waitF' threads, but may momentarily +> -- block on 'signal', 'signalF', other 'peekAvail', and the head waiter. This returns the amount of +> -- value available to be taken. Using this value without producing unwanted race conditions is left +> -- up to the programmer. +> -- +> -- 'peekAvail' is an optimized form of \"signalF m (\x -> (0,x))\". +> -- +> -- A version of 'peekAvail' that joins the FIFO queue of 'wait' and 'waitF' can be acheived by +> -- \"waitF m (\x -> (0,x))\" +> peekAvail :: Integral i => MSemN i -> IO i +> {-# SPECIALIZE peekAvail :: MSemN Int -> IO Int #-} +> {-# SPECIALIZE peekAvail :: MSemN Word -> IO Word #-} +> {-# SPECIALIZE peekAvail :: MSemN Integer -> IO Integer #-} +> peekAvail m = withMVar (quantityStore m) (return . avail) diff --git a/src/Control/Concurrent/MSemN2.hs b/src/Control/Concurrent/MSemN2.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/MSemN2.hs @@ -0,0 +1,262 @@ +{-# LANGUAGE DeriveDataTypeable #-} +-- | +-- Module : Control.Concurrent.MSemN2 +-- Copyright : (c) Chris Kuklewicz 2011 +-- License : 3 clause BSD-style (see the file LICENSE) +-- +-- Maintainer : haskell@list.mightyreason.com +-- Stability : experimental +-- Portability : non-portable (concurrency) +-- +-- Quantity semaphores in which each thread may wait for an arbitrary amount. This modules is +-- intended to improve on "Control.Concurrent.QSemN". +-- +-- This semaphore gracefully handles threads which die while blocked waiting for quantity. The +-- fairness guarantee is that blocked threads are FIFO. An early thread waiting for a large +-- quantity will prevent a later thread waiting for a small quantity from jumping the queue. +-- +-- If 'with' is used to guard a critical section then no quantity of the semaphore will be lost +-- if the activity throws an exception. +-- +-- The functions below are generic in (Integral i) with specialization to Int and Integer. +-- +-- Overflow warning: These operations do not check for overflow errors. If the Integral type is too +-- small to accept the new total then the behavior of these operations is undefined. Using (MSem +-- Integer) prevents the possibility of an overflow error. +module Control.Concurrent.MSemN2 + (MSemN + ,new + ,with + ,wait + ,signal + ,withF + ,waitF + ,signalF + ,peekAvail + ) where + +import Prelude( Integral,Eq,IO,Int,Integer,Maybe(Just,Nothing),Num((+),(-)),Bool(False,True) + , return,id,const,fmap,snd,seq + , (.),(<=),($),($!) ) +import Control.Concurrent.MVar( MVar + , withMVar,modifyMVar,newMVar + , newEmptyMVar,tryPutMVar,takeMVar,tryTakeMVar ) +import Control.Exception(bracket,bracket_,uninterruptibleMask_,evaluate,mask_) +import Control.Monad(when,void) +import Data.Maybe(fromMaybe) +import Data.Typeable(Typeable) +import Data.Word(Word) + +{- + +The only MVars allocated are the three created be 'new'. Their three roles are +1) to have a FIFO queue of waiters +2) for the head waiter to block on +3) to protect the quantity state of the semaphore and the head waiter + +-} + +-- MS has an invariant that "maybe True (> avail) headWants" is always True. +data MS i = MS { avail :: !i -- ^ This is the quantity available to be taken from the semaphore. + , headWants :: !(Maybe i) -- ^ If there is waiter then this is Just the amount being waited for. + } + deriving (Eq,Typeable) + +-- | A 'MSemN' is a quantity semaphore, in which the available quantity may be signalled or +-- waited for in arbitrary amounts. +data MSemN i = MSemN { quantityStore :: !(MVar (MS i)) -- ^ Used to lock access to state of semaphore quantity. + , queueWait :: !(MVar ()) -- ^ Used as FIFO queue for waiter, held by head of queue. + , headWait :: !(MVar i) -- ^ The head of the waiter queue blocks on headWait. + } + deriving (Eq,Typeable) + +-- |'new' allows positive, zero, and negative initial values. The initial value is forced here to +-- better localize errors. +new :: Integral i => i -> IO (MSemN i) +{-# SPECIALIZE new :: Int -> IO (MSemN Int) #-} +{-# SPECIALIZE new :: Word -> IO (MSemN Word) #-} +{-# SPECIALIZE new :: Integer -> IO (MSemN Integer) #-} +new initial = do + newMS <- newMVar $! (MS { avail = initial -- this forces 'initial' + , headWants = Nothing }) + newQueueWait <- newMVar () + newHeadWait <- newEmptyMVar + return (MSemN { quantityStore = newMS + , queueWait = newQueueWait + , headWait = newHeadWait }) + +-- | 'with' takes a quantity of the semaphore to take and hold while performing the provided +-- operation. 'with' ensures the quantity of the sempahore cannot be lost if there are exceptions. +-- This uses 'bracket' to ensure 'wait' and 'signal' get called correctly. +with :: Integral i => MSemN i -> i -> IO a -> IO a +{-# SPECIALIZE with :: MSemN Int -> Int -> IO a -> IO a #-} +{-# SPECIALIZE with :: MSemN Word -> Word -> IO a -> IO a #-} +{-# SPECIALIZE with :: MSemN Integer -> Integer -> IO a -> IO a #-} +with m wanted = seq wanted $ bracket_ (wait m wanted) (uninterruptibleMask_ $ signal m wanted) + +-- | 'withF' takes a pure function and an operation. The pure function converts the available +-- quantity to a pair of the wanted quantity and a returned value. The operation takes the result +-- of the pure function. 'withF' ensures the quantity of the sempahore cannot be lost if there +-- are exceptions. This uses 'bracket' to ensure 'waitF' and 'signal' get called correctly. +-- +-- Note: A long running pure function will block all other access to the 'MSemN' while it is +-- evaluated. +withF :: Integral i + => MSemN i + -> (i -> (i,b)) + -> ((i,b) -> IO a) + -> IO a +{-# SPECIALIZE withF :: MSemN Int -> (Int -> (Int,b)) -> ((Int,b) -> IO a) -> IO a #-} +{-# SPECIALIZE withF :: MSemN Word -> (Word -> (Word,b)) -> ((Word,b) -> IO a) -> IO a #-} +{-# SPECIALIZE withF :: MSemN Integer -> (Integer -> (Integer,b)) -> ((Integer,b) -> IO a) -> IO a #-} +withF m f = bracket (waitF m f) (\(wanted,_) -> uninterruptibleMask_ $ signal m wanted) + +-- |'wait' allow positive, zero, and negative wanted values. Waiters may block, and will be handled +-- fairly in FIFO order. +-- +-- If 'wait' returns without interruption then it left the 'MSemN' with a remaining quantity that was +-- greater than or equal to zero. If 'wait' is interrupted then no quantity is lost. If 'wait' +-- returns without interruption then it is known that each earlier waiter has definitely either been +-- interrupted or has retured without interruption. +wait :: Integral i => MSemN i -> i -> IO () +{-# SPECIALIZE wait :: MSemN Int -> Int -> IO () #-} +{-# SPECIALIZE wait :: MSemN Word -> Word -> IO () #-} +{-# SPECIALIZE wait :: MSemN Integer -> Integer -> IO () #-} +wait m wanted = seq wanted $ fmap snd $ waitF m (const (wanted,())) + +-- | 'waitWith' takes the 'MSemN' and a pure function that takes the available quantity and computes the +-- amount wanted and a second value. The value wanted is stricly evaluated but the second value is +-- returned lazily. +-- +-- 'waitF' allow positive, zero, and negative wanted values. Waiters may block, and will be handled +-- fairly in FIFO order. +-- +-- If 'waitF' returns without interruption then it left the 'MSemN' with a remaining quantity that was +-- greater than or equal to zero. If 'waitF' or the provided function are interrupted then no +-- quantity is lost. If 'waitF' returns without interruption then it is known that each previous +-- waiter has each definitely either been interrupted or has retured without interruption. +-- +-- Note: A long running pure function will block all other access to the 'MSemN' while it is +-- evaluated. +waitF :: Integral i => MSemN i -> (i -> (i,b)) -> IO (i,b) +{-# SPECIALIZE waitF :: MSemN Int -> (Int -> (Int,b)) -> IO (Int,b) #-} +{-# SPECIALIZE waitF :: MSemN Word -> (Word -> (Word,b)) -> IO (Word,b) #-} +{-# SPECIALIZE waitF :: MSemN Integer -> (Integer -> (Integer,b)) -> IO (Integer,b) #-} +waitF m f = seq f $ mask_ . withMVar (queueWait m) $ \ () -> do + (out,mustWait) <- modifyMVar (quantityStore m) $ \ ms -> do + -- Assume: ((headWait is empty) OR (headWants is Nothing)) + -- Nothing in this scope can block + -- + -- headWait might be full here if the predecessor waitF blocked and died and signal (tried to) + -- feed it. + recovered <- fmap (fromMaybe 0) (tryTakeMVar (headWait m)) + let total = avail ms + recovered + outVal@(wantedVal,_) = f total + if wantedVal <= total -- forces wantedVal + then do + ms' <- evaluate MS { avail = total - wantedVal, headWants = Nothing } + return (ms', (outVal,False)) + else do + ms' <- evaluate MS { avail = total, headWants = Just wantedVal } + return (ms', (outVal,True)) + -- quantityStore is now released, queueWait is still held, race with signal now possible + -- Assert: (headWait is empty) AND (mustWait == (headWants is Just)) at release + -- Proof: tryTakeMVar forced (headWait is empty), and + -- the if-then-else branches ensured (mustWait == (headWants is Just)) + -- This assertion implies ((headWait is empty) OR (headWants is Nothing)) invariant holds (point X) + when mustWait (void (takeMVar (headWait m))) + return out + -- Invariant: ((headWait is empty) OR (headWants is Nothing)) + -- Proof: 1) mustWait was false + -- nothing happened since (point X) except perhaps race with signal + -- signal maintained invariant + -- 2) mustWait was true + -- 2a) takeMVar succeeded so headWait became full since (point X) + -- this implies signal filled headWait and thus signal ended with (headWait is full) + -- signal invariant ((headWait is empty) OR (headWants is Nothing)) implies (headWants is Nothing) was set + -- (headWait is empty) by takeMVar and (headWants is Nothing) by implication + -- 2b) takeMVar was interrupted and thus did nothing + -- nothing happened since (point X) except perhaps race with signal + -- signal maintained invariant + +-- |'signal' allows positive, zero, and negative values, thus this is also way to remove quantity +-- that skips any threads in the 'wait'/'waitF' queue. If the new total is greater than the next +-- value being waited for (if present) then the first waiter is woken. If there are queued waiters +-- then the next one will wake after a waiter has proceeded and notice the remaining value; thus a +-- single 'signal' may result in several waiters obtaining values. Waking waiting threads is +-- asynchronous. +-- +-- 'signal' may block, but it cannot be interrupted, which allows it to dependably restore value to +-- the 'MSemN'. All 'signal', 'signalF', 'peekAvail', and the head waiter may momentarily block in a +-- fair FIFO manner. +signal :: Integral i => MSemN i -> i -> IO () +{-# SPECIALIZE signal :: MSemN Int -> Int -> IO () #-} +{-# SPECIALIZE signal :: MSemN Word -> Word -> IO () #-} +{-# SPECIALIZE signal :: MSemN Integer -> Integer -> IO () #-} +signal _ 0 = return () -- this case forces 'size' +signal m size = fmap snd $ signalF m (const (size,())) + +-- | Instead of providing a fixed change to the available quantity, 'signalF' applies a provided +-- pure function to the available quantity to compute the change and a second value. The +-- requested change is stricly evaluated but the second value is returned lazily. If the new total is +-- greater than the next value being waited for then the first waiter is woken. If there are queued +-- waiters then the next one will wake after a waiter has proceeded and notice the remaining value; +-- thus a single 'signalF' may result in several waiters obtaining values. Waking waiting threads +-- is asynchronous. +-- +-- 'signalF' may block, and it can be safely interrupted. If the provided function throws an error +-- or is interrupted then it leaves the 'MSemN' unchanged. All 'signal', 'signalF', 'peekAvail', and +-- the head waiter may momentarily block in a fair FIFO manner. +-- +-- Note: A long running pure function will block all other access to the 'MSemN' while it is +-- evaluated. +signalF :: Integral i + => MSemN i + -> (i -> (i,b)) + -> IO (i,b) +{-# SPECIALIZE signalF :: MSemN Int -> (Int -> (Int,b)) -> IO (Int,b) #-} +{-# SPECIALIZE signalF :: MSemN Word -> (Word -> (Word,b)) -> IO (Word,b) #-} +{-# SPECIALIZE signalF :: MSemN Integer -> (Integer -> (Integer,b)) -> IO (Integer,b) #-} +signalF m f = seq f $ mask_ . modifyMVar (quantityStore m) $ \ ms -> do + -- Assume: ((headWait is empty) OR (headWants is Nothing)) + -- Nothing in this scope can block + let out@(size,_) = f (avail ms) + ms' <- case headWants ms of + Nothing -> evaluate ms { avail = avail ms + size } + Just wantedVal -> do + -- Because headWants is Just _ the assumption implies headWait is empty + let total = avail ms + size + if wantedVal <= total + then do + _didPlace <- tryPutMVar (headWait m) wantedVal -- _didPlace is always True + evaluate MS { avail = total - wantedVal, headWants = Nothing } + else do + evaluate ms { avail = total } + return (ms',out) + -- Invariant: ((headWait is empty) OR (headWants is Nothing)) + -- Proof: Assume invariant originally holds when taking quantityStore + -- 1) headWants originally Nothing, headWants and headWait unchanged, invariant still holds + -- 2) headWants originally Just _ implies, by assumption, that (headWait is empty) + -- if-then-branch: headWants changed to Nothing and headWait changed to filled, invariant satisfied + -- if-else-branch: headWants and headWait unchanged, invariant still holds + +-- | 'peekAvail' skips the queue of any blocked 'wait' and 'waitF' threads, but may momentarily +-- block on 'signal', 'signalF', other 'peekAvail', and the head waiter. This returns the amount of +-- value available to be taken. Using this value without producing unwanted race conditions is left +-- up to the programmer. +-- +-- 'peekAvail' is an optimized form of \"signalF m (\x -> (0,x))\". +-- +-- Quantity that has been passed to a blocked waiter but not picked up is not counted. If the +-- blocked waiter is killed before picking it up then the passed quantity will be recovered by the +-- next waiter. In this exceptional case this next waiter may see an available total that is +-- different than returned by peekAvail. +-- +-- A version of 'peekAvail' that joins the FIFO queue of 'wait' and 'waitF' can be acheived by +-- \"waitF m (\x -> (0,x))\" but this will block if x is negative. On the other hand this method +-- will see the total including any recovered quantity. +peekAvail :: Integral i => MSemN i -> IO i +{-# SPECIALIZE peekAvail :: MSemN Int -> IO Int #-} +{-# SPECIALIZE peekAvail :: MSemN Word -> IO Word #-} +{-# SPECIALIZE peekAvail :: MSemN Integer -> IO Integer #-} +peekAvail m = withMVar (quantityStore m) (return . avail) diff --git a/src/Control/Concurrent/SSem.hs b/src/Control/Concurrent/SSem.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/SSem.hs @@ -0,0 +1,77 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.SSem +-- Copyright : (c) Chris Kuklewicz, 2012 +-- License : BSD-style +-- +-- Maintainer : haskell@list.mightyreason.com +-- Stability : experimental +-- Portability : non-portable (concurrency) +-- +-- Very simple quantity semaphore. +-- +----------------------------------------------------------------------------- +module Control.Concurrent.SSem( SSem,new + , withSem,wait,signal,tryWait + , withSemN,waitN,signalN,tryWaitN + , getValue) where + +import Control.Concurrent.STM.SSemInternals(SSem(SSem)) +import qualified Control.Concurrent.STM.SSem as S(wait,signal,tryWait,waitN,signalN,tryWaitN,getValue) +import Control.Concurrent.STM.TVar(newTVarIO) +import Control.Exception(bracket_) +import Control.Monad.STM(atomically) + +-- | Create a new semaphore with the given argument as the initially available quantity. This +-- allows new semaphores to start with a negative, zero, or positive quantity. +new :: Int -> IO SSem +new = fmap SSem . newTVarIO + +-- | It is recommended that all paired uses of 'wait' and 'signal' use the 'with' bracketed form +-- to ensure exceptions safety. +withSem :: SSem -> IO a -> IO a +withSem s = bracket_ (wait s) (signal s) + +-- | It is recommended that all paired uses of 'waitN' and 'signalN' use the 'withN' +-- bracketed form to ensure exceptions safety. +withSemN :: SSem -> Int -> IO a -> IO a +withSemN s i = bracket_ (waitN s i) (signalN s i) + +-- | Try to take a unit of value from the semaphore. This succeeds when the current quantity is +-- positive, and then reduces the quantity by one. Otherwise this will block and 'retry' until it +-- succeeds or is killed. This will never result in a negative quantity. If several threads are +-- retying then which one succeeds next is undefined -- an unlucky thread might starve. +wait :: SSem -> IO () +wait = atomically . S.wait + +-- | Try to take the given value from the semaphore. This succeeds when the quantity is greater or +-- equal to the given value, and then subtracts the given value from the quantity. Otherwise this +-- will block and 'retry' until it succeeds or is killed. This will never result in a negative +-- quantity. If several threads are retrying then which one succeeds next is undefined -- an +-- unlucky thread might starve. +waitN :: SSem -> Int-> IO () +waitN s i = atomically (S.waitN s i) + +-- | Signal that single unit of the semaphore is available. This increases the available quantity +-- by one. +signal :: SSem -> IO () +signal = atomically . S.signal + +-- | Signal that many units of the semaphore are available. This changes the available quantity by +-- adding the passed size. +signalN :: SSem-> Int -> IO () +signalN s i = atomically (S.signalN s i) + +-- | Non-waiting version of wait. `tryWait s` is defined as `tryWaitN s 1` +tryWait :: SSem -> IO (Maybe Int) +tryWait = atomically . S.tryWait + +-- | Non-waiting version of waitN. It either takes the quantity from the semaphore like +-- waitN and returns `Just value taken` or finds insufficient quantity to take and returns +-- Nothing +tryWaitN :: SSem -> Int -> IO (Maybe Int) +tryWaitN s i = atomically (S.tryWaitN s i) + +-- | This returns the current quantity in the semaphore. This is diffucult to use due to race conditions. +getValue :: SSem -> IO Int +getValue = atomically . S.getValue diff --git a/src/Control/Concurrent/STM/SSem.hs b/src/Control/Concurrent/STM/SSem.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/STM/SSem.hs @@ -0,0 +1,75 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.STM.SSem +-- Copyright : (c) Chris Kuklewicz, 2012 +-- License : BSD-style +-- +-- Maintainer : haskell@list.mightyreason.com +-- Stability : experimental +-- Portability : non-portable (concurrency) +-- +-- Very simple quantity semaphore. +-- +----------------------------------------------------------------------------- +module Control.Concurrent.STM.SSem(SSem, new, wait, signal, tryWait + , waitN, signalN, tryWaitN + , getValue) where + +import Control.Monad.STM(STM,retry) +import Control.Concurrent.STM.TVar(newTVar,readTVar,writeTVar) +import Control.Concurrent.STM.SSemInternals(SSem(SSem)) + +-- | Create a new semaphore with the given argument as the initially available quantity. This +-- allows new semaphores to start with a negative, zero, or positive quantity. +new :: Int -> STM SSem +new = fmap SSem . newTVar + +-- | Try to take a unit of value from the semaphore. This succeeds when the current quantity is +-- positive, and then reduces the quantity by one. Otherwise this will 'retry'. This will never +-- result in a negative quantity. If several threads are retying then which one succeeds next is +-- undefined -- an unlucky thread might starve. +wait :: SSem -> STM () +wait = flip waitN 1 + +-- | Try to take the given value from the semaphore. This succeeds when the quantity is greater or +-- equal to the given value, and then subtracts the given value from the quantity. Otherwise this +-- will 'retry'. This will never result in a negative quantity. If several threads are retrying +-- then which one succeeds next is undefined -- an unlucky thread might starve. +waitN :: SSem -> Int -> STM () +waitN (SSem s) i = do + v <- readTVar s + if v >= i + then writeTVar s $! v-i + else retry + +-- | Signal that single unit of the semaphore is available. This increases the available quantity +-- by one. +signal :: SSem -> STM () +signal = flip signalN 1 + +-- | Signal that many units of the semaphore are available. This changes the available quantity by +-- adding the passed size. +signalN :: SSem -> Int -> STM () +signalN (SSem s) i = do + v <- readTVar s + writeTVar s $! v+i + +-- | Non-retrying version of 'wait'. `tryWait s` is defined as `tryN s 1` +tryWait :: SSem -> STM (Maybe Int) +tryWait = flip tryWaitN 1 + +-- | Non-retrying version of waitN. It either takes the quantity from the semaphore like +-- waitN and returns `Just value taken` or finds insufficient quantity to take and returns +-- Nothing +tryWaitN :: SSem -> Int -> STM (Maybe Int) +tryWaitN (SSem s) i = do + v <- readTVar s + if v >= i + then do writeTVar s $! v-i + return (Just i) + else return Nothing + +-- | Return the current quantity in the semaphore. This is potentially useful in a larger STM +-- transaciton and less useful as `atomically getValueSem :: IO Int` due to race conditions. +getValue :: SSem -> STM Int +getValue (SSem s) = readTVar s diff --git a/src/Control/Concurrent/STM/SSemInternals.hs b/src/Control/Concurrent/STM/SSemInternals.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/STM/SSemInternals.hs @@ -0,0 +1,27 @@ +{-# LANGUAGE CPP, StandaloneDeriving, DeriveDataTypeable #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.STM.SSemInternals +-- Copyright : (c) Chris Kuklewicz, 2012 +-- License : BSD-style +-- +-- Maintainer : haskell@list.mightyreason.com +-- Stability : experimental +-- Portability : non-portable (concurrency) +-- +-- Very simple quantity semaphore. Declared here so that private constructor +-- can be shared in both STM and IO APIs but hidden from user. +-- +----------------------------------------------------------------------------- + +module Control.Concurrent.STM.SSemInternals(SSem(SSem)) where + +import Control.Concurrent.STM.TVar(TVar) +import Data.Typeable(Typeable) -- Typeable(typeOf),TyCon,mkTyCon,mkTyConApp) + +#include "Typeable.h" + +newtype SSem = SSem (TVar Int) deriving (Eq) + +INSTANCE_TYPEABLE0(SSem,semTc,"SSem") diff --git a/tests/TestKillSem.hs b/tests/TestKillSem.hs new file mode 100644 --- /dev/null +++ b/tests/TestKillSem.hs @@ -0,0 +1,249 @@ +{- some output from log of "cabal test", three old modules fail, three new modules pass: + +Test SampleVar +0: forkIO read thread 1 +0: stop thread 1 +1: read interrupted +0: write sv #1 +0: write sv #2 with timeout +0: timeout triggered, write sv #2 blocked, FAIL + + +Test QSem +0: forkIO wait thread 1 +0: stop thread 1 +1: wait interrupted +0: signal q #1 +0: forkIO wait thread 2 +0: forkIO wait thread 3 +0: signal q #2 +2: wait done +0: stop thread 2 +0: stop thread 3 +3: wait interrupted (QUANTITY LOST) FAIL +False + + +Test QSemN +0: forkIO wait thread 1 +0: stop thread 1 +1: wait interrupted +0: signal q #1 +0: forkIO wait thread 2 +0: forkIO wait thread 3 +0: signal q #2 +2: wait done +0: stop thread 2 +0: stop thread 3 +3: wait interrupted (QUANTITY LOST) FAIL +False +Expected 3 Failures for above code + + + +Test MSampleVar +0: forkIO read thread 1 +0: stop thread 1 +1: read interrupted +0: write sv #1 +0: write sv #2 with timeout +0: write sv #2 returned, PASS + + +Test MSem +0: forkIO wait thread 1 +0: stop thread 1 +1: wait interrupted +0: signal q #1 +0: forkIO wait thread 2 +2: wait done +0: forkIO wait thread 3 +0: signal q #2 +3: wait done (QUANTITY CONSERVED) PASS +0: stop thread 2 +0: stop thread 3 +True + + +Test MSemN +0: forkIO wait thread 1 +0: stop thread 1 +1: wait interrupted +0: signal q #1 +0: forkIO wait thread 2 +2: wait done +0: forkIO wait thread 3 +0: signal q #2 +3: wait done (QUANTITY CONSERVED) PASS +0: stop thread 2 +0: stop thread 3 +True +Test suite TestSafeSemaphore: PASS +Test suite logged to: dist/test/SafeSemaphore-0.8.0-TestSafeSemaphore.log + +-} +module Main where + +import Prelude hiding (read) +import Control.Concurrent +import Control.Exception +import Control.Concurrent.QSem +import Control.Concurrent.QSemN +import qualified Control.Concurrent.MSem as MSem +import qualified Control.Concurrent.MSemN as MSemN +import qualified Control.Concurrent.MSemN2 as MSemN2 +import qualified Control.Concurrent.SSem as SSem +import Control.Concurrent.MVar +import Test.HUnit +import System.Exit +import Control.Concurrent.SampleVar +import Control.Concurrent.MSampleVar as MSV +import System.Timeout + +delay = threadDelay (1000*100) +--delay = yield -- now causes tests to fail in ghc 7.4 + +fork x = do m <- newEmptyMVar + t <- forkIO (finally x (putMVar m ())) + delay + return (t,m) + +stop (t,m) = do killThread t + delay + takeMVar m + +-- True if test passed, False if test failed +-- This expects FIFO semantics for the waiters +testSem :: Integral n + => String + -> (n -> IO a) + -> (a->IO ()) + -> (a -> IO ()) + -> IO Bool +testSem name new wait signal = do + putStrLn ("\n\nTest "++ name) + q <- new 0 + + putStrLn "0: forkIO wait thread 1" + (t1,m1) <- fork $ do + wait q `onException` (putStrLn "1: wait interrupted") + putStrLn "1: wait done UNEXPECTED" + putStrLn "0: stop thread 1" + stop (t1,m1) + putStrLn "0: signal q #1" + signal q + delay + + putStrLn "0: forkIO wait thread 2" + (t2,m2) <- fork $ do + wait q `onException` (putStrLn "2: wait interrupted UNEXPECTED") + putStrLn "2: wait done" + delay + + result <- newEmptyMVar + putStrLn "0: forkIO wait thread 3" + (t3,m3) <- fork $ do + wait q `onException` (putStrLn "3: wait interrupted (QUANTITY LOST) FAIL" >> putMVar result False) + putStrLn "3: wait done (QUANTITY CONSERVED) PASS" + putMVar result True + putStrLn "0: signal q #2" + signal q + delay + + putStrLn "0: stop thread 2" + stop (t2,m2) + putStrLn "0: stop thread 3" + stop (t3,m3) + r <- takeMVar result + print r + return r + +testSV name newEmpty read write = do + putStrLn ("\n\nTest "++ name) + sv <- newEmpty + putStrLn "0: forkIO read thread 1" + (t1,m1) <- fork $ do + read sv `onException` (putStrLn "1: read interrupted") + putStrLn "1: read done UNEXPECTED" + putStrLn "0: stop thread 1" + stop (t1,m1) + putStrLn "0: write sv #1" + write sv 1 + putStrLn "0: write sv #2 with timeout" + m <- timeout (1000*100) (write sv 2) + case m of + Nothing -> do + putStrLn "0: timeout triggered, write sv #2 blocked, FAIL" + return False + Just () -> do + putStrLn "0: write sv #2 returned, PASS" + return True + +-- True if test passed, False if test failed +-- This does not expect FIFO semantics for the waiters, uses getValue instead +testSSem :: Integral n + => String + -> (n -> IO a) + -> (a->IO ()) + -> (a -> IO ()) + -> (a -> IO Int) + -> IO Bool +testSSem name new wait signal getValue = do + putStrLn ("\n\nTest "++ name) + q <- new 0 + + putStrLn "0: forkIO wait thread 1" + (t1,m1) <- fork $ do + wait q `onException` (putStrLn "1: wait interrupted") + putStrLn "1: wait done UNEXPECTED" + putStrLn "0: stop thread 1" + stop (t1,m1) + putStrLn "0: signal q #1" + signal q + delay + + putStrLn "0: forkIO wait thread 2" + (t2,m2) <- fork $ do + wait q `onException` (putStrLn "2: wait interrupted") + putStrLn "2: wait done" + delay + + putStrLn "0: forkIO wait thread 3" + (t3,m3) <- fork $ do + wait q `onException` (putStrLn "3: wait interrupted") + putStrLn "3: wait done" + delay + + putStrLn "0: signal q #2" + signal q + delay + + putStrLn "0: stop thread 2" + stop (t2,m2) + putStrLn "0: stop thread 3" + stop (t3,m3) + r <- getValue q + putStrLn $ "Final Value "++show r + return (r==0) + +testOldSV = test $ testSV "SampleVar" newEmptySampleVar readSampleVar writeSampleVar +testNewSV = test $ testSV "MSampleVar" newEmptySV readSV writeSV + +testsQ = TestList . (testOldSV:) . map test $ + [ testSem "QSem" newQSem waitQSem signalQSem + , testSem "QSemN" newQSemN (flip waitQSemN 1) (flip signalQSemN 1) + ] + +testsM = TestList . (testNewSV:) . map test $ + [ testSem "MSem" MSem.new MSem.wait MSem.signal + , testSem "MSemN" MSemN.new (flip MSemN.wait 1) (flip MSemN.signal 1) + , testSem "MSemN2" MSemN2.new (flip MSemN2.wait 1) (flip MSemN2.signal 1) + , testSSem "SSem" SSem.new SSem.wait SSem.signal SSem.getValue + ] + +-- This is run by "cabal test" +main = do + runTestTT testsQ + putStrLn "Expected 3 Failures for above code\n" + c <- runTestTT testsM + if failures c == 0 then exitSuccess else exitFailure