SafeSemaphore 0.7.0 → 0.9.0

raw patch · 11 files changed

+1411/−431 lines, 11 filesdep +stmdep ~basePVP ok

version bump matches the API change (PVP)

Dependencies added: stm

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• SafeSemaphore.cabal +10/−4
• src/Control/Concurrent/FairRWLock.hs +86/−72
• src/Control/Concurrent/MSem.hs +0/−132
• src/Control/Concurrent/MSem.lhs +306/−0
• src/Control/Concurrent/MSemN.hs +0/−223
• src/Control/Concurrent/MSemN.lhs +319/−0
• src/Control/Concurrent/MSemN2.hs +262/−0
• src/Control/Concurrent/SSem.hs +77/−0
• src/Control/Concurrent/STM/SSem.hs +75/−0
• src/Control/Concurrent/STM/SSemInternals.hs +27/−0
• tests/TestKillSem.hs +249/−0

@@ -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 <https://github.com/ChrisKuklewicz/SafeSemaphore> for more details.+Homepage:            https://github.com/ChrisKuklewicz/SafeSemaphore License:             BSD3 License-file:        LICENSE Author:              Chris Kuklewicz <haskell@list.mightyreason.com>@@ -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

@@ -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  <http://download.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantReadWriteLock.html> - 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

@@ -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

@@ -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.

@@ -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)

@@ -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)

@@ -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)

@@ -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

@@ -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

@@ -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")

@@ -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