SafeSemaphore 0.6.0 → 0.7.0

raw patch · 4 files changed

+711/−94 lines, 4 filesdep +containersPVP ok

version bump matches the API change (PVP)

Dependencies added: containers

API changes (from Hackage documentation)

Files

• SafeSemaphore.cabal +4/−2
• src/Control/Concurrent/FairRWLock.hs +582/−0
• src/Control/Concurrent/MSem.hs +54/−40
• src/Control/Concurrent/MSemN.hs +71/−52

@@ -1,5 +1,5 @@ Name:                SafeSemaphore-Version:             0.6.0+Version:             0.7.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@@ -17,9 +17,11 @@   Exposed-modules: Control.Concurrent.MSem                    Control.Concurrent.MSemN                    Control.Concurrent.MSampleVar-  Build-depends:  base < 5+                   Control.Concurrent.FairRWLock+  Build-depends:  base < 5, containers  Test-Suite TestSafeSemaphore   type:          exitcode-stdio-1.0   main-is:       tests/TestKillSem.hs   build-depends: base < 5, SafeSemaphore, HUnit+

@@ -0,0 +1,582 @@+{-# LANGUAGE DeriveDataTypeable, PatternGuards #-}+{-| Provides a fair RWLock, similar to one from Java.++ <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'.++ 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.  If the FairRWLock has a+ bug and finds itself in an impossible state then it will throw an error.++ 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.++ 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.++ 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.++-}+module Control.Concurrent.FairRWLock+  ( RWLock, RWLockException(..), RWLockExceptionKind(..),FRW(..),LockKind(..),TMap,TSet+  , new, peekLock, checkLock+  , acquireRead, acquireWrite+  , releaseRead, releaseWrite+  , withRead, withWrite+  ) where++import Control.Applicative(liftA2)+import Control.Concurrent+import Control.Exception+import Control.Monad((>=>),join,forM_)+import Data.Sequence((<|),(|>),(><),Seq,ViewL(..),ViewR(..))+import qualified Data.Sequence as Seq(empty,viewl,viewr,breakl,spanl)+import qualified Data.Foldable as F(toList)+import Data.Map(Map)+import qualified Data.Map as Map+import Data.Set(Set)+import qualified Data.Set as Set+import Data.Typeable(Typeable)++-- Try to make most impossible data states unrepresentable+type TMap = Map ThreadId Int -- nonempty, all values > 0+type TSet = Set ThreadId     -- nonempty++data LockKind = ReaderKind { unRK :: TSet }+              | WriterKind { unWK :: ThreadId }+  deriving (Eq,Ord,Show)++-- LockQ may be empty+-- No duplicate ThreadIds in LockKinds+-- MVar () will be created empty, released once with putMVar+type LockQ = Seq (LockKind,MVar ())++data LockUser = +    FreeLock+  | Readers { readerCounts :: TMap -- re-entrant count of reader locks held be each thread+            , queueR :: Maybe ( (ThreadId,MVar ())    -- empty or queue with leading Writer+                              , LockQ )+            }+  | Writer { writerID :: ThreadId+           , writerCount           -- re-entrant writer locks held by writerID, at least 1+           , readerCount :: !Int   -- re-entrant reader locks held by writerID, at least 0+           , queue :: LockQ }+  deriving (Eq,Typeable)+++-- | Opaque type of the fair RWLock.+newtype RWLock = RWL (MVar LockUser)++-- | Exception type thrown or returned by this module.  \"Impossible\" conditions get the error thrown+--  and usage problems get the error returned.+data RWLockException = RWLockException ThreadId RWLockExceptionKind String+  deriving (Show,Typeable)++-- | 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+-- 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.+data FRW = F | R TMap | W (ThreadId,(Int,Int)) deriving (Show)++-- | Create a new RWLock which starts in a free and unlocked state.+new :: IO RWLock+new = fmap RWL (newMVar FreeLock)++-- | This is by far the preferred way to acquire a read lock.  This uses bracket_ to ensure+-- acquireRead and releaseRead are called correctly around the passed command.+--+-- This ought to ensure releaseRead will not return a (Left error), but if it does then this error+-- will be thrown.+--+-- This can block and be safely interrupted.+withRead :: RWLock -> IO a -> IO a+withRead = liftA2 bracket_ acquireRead (releaseRead >=> either throw return)++-- | This is by far the preferred way to acquire a write lock.  This uses bracket_ to ensure+-- acquireWrite and releaseWrite are called correctly around the passed command.+--+-- This ought to ensure releaseWrite will not return a (Left error), but if it does then this error+-- will be thrown.+--+-- This can block and be safely interrupted.+withWrite :: RWLock -> IO a -> IO a+withWrite = liftA2 bracket_ acquireWrite (releaseWrite >=> either throw return)++-- | Observe which threads are holding the lock and which threads are waiting (in order).  This is+-- particularly useful for writing tests.+peekLock :: RWLock -> IO (FRW,[LockKind])+peekLock (RWL rwlVar) = withMVar rwlVar $ \ rwd -> return $+  case rwd of+    FreeLock -> (F,[])+    Readers { readerCounts=rcs, queueR=qr } -> (R rcs,maybe [] (\((t,_),q) -> WriterKind t : map fst (F.toList q)) qr)+    Writer { writerID=it, writerCount=wc, readerCount=rc, queue=q } -> (W (it,(rc,wc)), map fst (F.toList q))++-- | checkLocks return a pair of numbers, the first is the count of read locks this thread holds,+-- the second is the number of write locks that this thread holds.  This may be useful for sanity+-- checking complex usage of RWLocks.+--+-- This may block and be safely interrupted.+checkLock :: RWLock -> IO (Int,Int)+checkLock (RWL rwlVar) = do+  me <- myThreadId+  withMVar rwlVar $ \ rwd -> return $+    case rwd of+      FreeLock -> (0,0)+      Readers { readerCounts=rcs } ->+        case Map.lookup me rcs of+          Nothing -> (0,0)+          Just rc -> (rc,0)+      Writer { writerID=it, writerCount=wc, readerCount=rc } ->+        if it==me then (rc,wc) else (0,0)++-- | A thread that calls acquireRead must later call releaseRead once for each call to acquireRead.+--+-- If this thread has not previous called acquireRead then releaseRead will do nothing and return a+-- (Left error).+--+-- This can block but cannot be interrupted.+releaseRead :: RWLock -> IO (Either RWLockException ())+releaseRead (RWL rwlVar) = uninterruptibleMask_ $ do+  me <- myThreadId+  releaseRead' False me rwlVar -- False to indicate called 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+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+      impossible s x = throw+        (RWLockException me (if abandon then RWLock'acquireRead else RWLock'releaseRead) (imp s x))+      err :: Show x => String -> x -> IO (LockUser,Either RWLockException ())+      err s x = return . ((,) rwd) . Left $+        (RWLockException me (if abandon then RWLock'acquireRead else RWLock'releaseRead) (s++" : "++show x))+      ret :: LockUser -> IO (LockUser,Either RWLockException ())+      ret x = return (x,Right ())++      -- if there is a bug then dropReader may find an impossible situation when abandoning a thread, and throw an error+      dropReader :: LockQ -> IO LockQ+      dropReader q = do+        let inR (ReaderKind rcs,_) = Set.member me rcs+            inR _ = False+            (pre,myselfPost) = Seq.breakl inR q+        case Seq.viewl myselfPost of+          EmptyL ->+            impossible "failure to abandon acquireRead, RWLock locked by other thread(s) and this thread is not in queue" me+          (myself,mblock) :< post -> do+            let rcs' = Set.delete me (unRK myself) -- safe unRK call+            evaluate $ if Set.null rcs' then pre >< post else pre >< ((ReaderKind rcs',mblock) <| post)++  case rwd of+    FreeLock | abandon ->+      impossible "acquireRead interrupted with unlocked RWLock" me+             | otherwise ->+      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+          rc' <- evaluate $ pred rc+          ret (w { readerCount=rc' })++    {-ditto-}                                           | abandon -> do+      q' <- dropReader q+      ret (w { queue=q' })++    {-ditto-}                                           | otherwise ->+      err "releaseRead called when not read locked " me++    r@(Readers { readerCounts=rcs,queueR=qR }) ->+      case Map.lookup me rcs of+        Just 1 -> do+          let rcs' = Map.delete me rcs+          if Map.null rcs'+            then case qR of+                   Nothing ->+                     ret FreeLock+                   Just ((wid,mblock),q) -> do+                     putMVar mblock ()+                     ret (Writer { writerID=wid, writerCount=1, readerCount=0, queue=q })+            else ret (r { readerCounts=rcs' })++        Just rc -> do+          rc' <- evaluate $ pred rc+          rcs' <- evaluate $ Map.insert me rc' rcs+          ret (r { readerCounts=rcs' })++        Nothing   | abandon ->+          case qR of+            Nothing ->+              impossible "acquireRead interrupted not holding lock and with no queue" (me,rcs)+            Just (w,q) -> do+              q' <- dropReader q+              ret (r { queueR = Just (w,q') })++        {-ditto-} | otherwise -> +          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.+--+-- If this thread has not previous called acquireWrite then releaseWrite will do nothing and return+-- a (Left error).+--+-- This can block but cannot be interrupted.+releaseWrite :: RWLock -> IO (Either RWLockException ())+releaseWrite (RWL rwlVar) = uninterruptibleMask_ $ do+  me <- myThreadId+  releaseWrite' False me rwlVar  -- False to indicate called from releaseWrite++-- Nine non-impossible cases, plus one impossible case+-- Lock is Free+-- I have write lock, I only had 1 write lock and no read locks, promote from LockQ+--                  , I only had 1 write lock and some read locks, convert me to reader and promote leading readers+--                  , I have many write locks, just decrement the counter+-- Someone else has write lock, abandoning my acquireWrite+--                            , releaseWrite called in error+-- Read lock held, releaseWrite called in error+--               , with no queue, abandoning acquireWrite is IMPOSSIBLE+--               , abandoning my leading acquireWrite+--               , abandoning my non-leading acquireWrite+releaseWrite' :: Bool -> ThreadId -> MVar LockUser -> IO (Either RWLockException ())+releaseWrite' abandon me rwlVar = uninterruptibleMask_ . modifyMVar rwlVar $ \ rwd -> do+  let impossible :: Show x => String -> x -> IO a+      impossible s x = throw+        (RWLockException me (if abandon then RWLock'acquireWrite else RWLock'releaseWrite) (imp s x))+      err :: Show x => String -> x -> IO (LockUser,Either RWLockException ())+      err s x = return . ((,) rwd) . Left $+        (RWLockException me (if abandon then RWLock'acquireWrite else RWLock'releaseWrite) (s++" : "++show x))+      ret :: LockUser -> IO (LockUser,Either RWLockException ())+      ret x = return (x,Right ())++      dropWriter :: LockQ -> IO LockQ+      dropWriter q = do+        let inW (WriterKind it,_) = me==it+            inW _ = False+            (pre,myselfPost) = Seq.breakl inW q+        case Seq.viewl myselfPost of+          EmptyL -> +            impossible "failure to abandon acquireWrite, RWLock locked by other and not in queue" me+          _ :< post ->+            evaluate $ pre><post++  case rwd of+    FreeLock | abandon ->+     impossible "acquireWrite interrupted with unlocked RWLock" me+             | otherwise ->+     err "cannot releaseWrite lock from unlocked RWLock" me++    w@(Writer { writerID=it, writerCount=wc, readerCount=rc, queue=q }) | it==me -> do+      case (wc,rc) of+        (1,0) -> ret =<< promote q  -- if abandon then this is the only valid case+        _ | abandon -> impossible "acquireWrite interrupted with write lock and bad RWLock state" (me,it,wc,rc)+        (1,_) -> ret =<< promoteReader rc q+        (_,_) -> ret (w { writerCount=(pred wc) })++    {-ditto-}                                                          | abandon -> do+      q' <- dropWriter q+      ret (w { queue=q' })++    {-ditto-}                                                          | otherwise  -> do+      err "cannot releaseWrite when not not holding the write lock" (me,it)++    Readers { readerCounts=rcs} | not abandon ->+      err "cannot releaseWrite when RWLock is read locked" (me,rcs)+          +    Readers { readerCounts=rcs, queueR=Nothing } ->+      impossible "failure to abandon acquireWrite, RWLock read locked and no queue" (me,rcs)++    r@(Readers { readerCounts=rcs, queueR=Just (w@(it,_),q) }) | it==me -> do+      (rcs'new,qr) <- splitReaders q+      ret (r { readerCounts=Map.union rcs rcs'new, queueR=qr })++    {- ditto -}                                                | otherwise -> do+      q' <- dropWriter q+      ret (r { queueR=Just (w,q') })++ where+  -- | promote when converting from write lock straight to read lock+  promoteReader :: Int -> LockQ -> IO LockUser+  promoteReader rc q = do+    (rcs'new, qr) <- splitReaders q+    let rcs = Map.insert me rc rcs'new+    return (Readers { readerCounts=rcs, queueR=qr })++  -- | promote from releasing write lock+  promote :: LockQ -> IO LockUser+  promote qIn = do+    case Seq.viewl qIn of+      EmptyL -> return FreeLock++      (WriterKind it,mblock) :< qOut -> do+        putMVar mblock ()+        return (Writer { writerID=it, writerCount=1, readerCount=0, queue=qOut })++      _ -> do+        (rcs,qr) <- splitReaders qIn+        return (Readers { readerCounts=rcs, queueR=qr })++  -- | Merge (and wake) any and all readers on left end of LockQ, and return queueR value+  splitReaders :: LockQ -> IO (TMap,Maybe ((ThreadId,MVar ()),LockQ))+  splitReaders qIn = do+    let (more'Readers,qTail) = Seq.spanl isReader qIn+        (rks,mblocks) = unzip (F.toList more'Readers)+        rcs = Map.fromDistinctAscList . map (\k -> (k,1)) . F.toList . Set.unions . map unRK $ rks+        qr = case Seq.viewl qTail of+              EmptyL -> Nothing+              (wk,mblock) :< qOut -> Just ((unWK wk,mblock),qOut) -- unWK safe+    forM_ mblocks (\mblock -> putMVar mblock ())+    return (rcs,qr)+   where+    isReader (ReaderKind {},_) = True+    isReader _ = False++-- Six cases:+-- Lock is Free+-- I already have write lock+-- Someone else has write lock, 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++-- | 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.+--+-- All previous calls to acquireWrite by other threads will have succeeded and been released (or+-- interrupted) before this acquireRead will return.+--+-- The best way to use acquireRead is to use withRead instead to ensure releaseRead will be called+-- exactly once.+--+-- This may block and be safely interrupted.  If interrupted then the RWLock will be left unchanged.+acquireRead :: RWLock -> IO ()+acquireRead (RWL rwlVar) = mask_ . join . modifyMVar rwlVar $ \ rwd -> do+  me <- myThreadId+  let safeBlock mblock = (readMVar mblock) `onException` (releaseRead' True me rwlVar)+  case rwd of+    FreeLock -> +      return ( Readers { readerCounts=Map.singleton me 1, queueR=Nothing }+             , return () )++    w@(Writer { writerID=it, readerCount=rc, queue=q }) | it == me -> do+      rc' <- evaluate $ succ rc+      return ( w { readerCount=rc' }+             , return () )+                                                        | otherwise -> do+      (q',mblock) <- enterQueueR q me+      return ( w { queue = q' }+             , safeBlock mblock )++    r@(Readers { readerCounts=rcs }) | Just rc <- Map.lookup me rcs -> do+      rc' <- evaluate $ succ rc+      rcs' <- evaluate $ Map.insert me rc' rcs+      return ( r { readerCounts=rcs' }+             , return () )++    r@(Readers { readerCounts=rcs, queueR=Nothing }) -> do+      rcs' <- evaluate $ Map.insert me 1 rcs+      return ( r { readerCounts=rcs' }+             , return () )++    r@(Readers { queueR=Just (w,q) }) -> do+      (q',mblock) <- enterQueueR q me+      return ( r { queueR=Just (w,q') }+             , safeBlock mblock )+ where+  -- Merge adjacent readers when adding to right end of LockQ+  enterQueueR :: LockQ -> ThreadId -> IO (LockQ,MVar ())+  enterQueueR qIn me = do+    case Seq.viewr qIn of+      pre :> (ReaderKind rcs,mblock) -> do+        rcs' <- addMe rcs+        return (pre |> (ReaderKind rcs', mblock),mblock)+      _ -> do+        mblock <- newEmptyMVar+        return (qIn |> (ReaderKind (Set.singleton me),mblock), mblock)+   where+    -- Paranoid check of design assertion, TODO: remove check+    addMe :: TSet -> IO TSet+    addMe rcs | Set.member me rcs = error (imp "enterQueueR.addMe when already in set" me)+              | otherwise = return (Set.insert me rcs)++-- 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+-- to acquireRead.  If this acquireWrite gets interrupted then acquireReadPriority will to obtain+-- the read lock or put itself at the front of the queue if another thread holds the write lock.+acquireReadPriority :: RWLock -> IO ()+acquireReadPriority (RWL rwlVar) = uninterruptibleMask_ . join . modifyMVar rwlVar $ \ rwd -> do+  me <- myThreadId+  let safeBlock mblock = (readMVar mblock) `onException` (releaseRead' True me rwlVar)+  case rwd of+    FreeLock -> +      return ( Readers { readerCounts=Map.singleton me 1, queueR=Nothing }+             , return () )++    w@(Writer { writerID=it, readerCount=rc, queue=q }) | it == me -> do+      rc' <- evaluate $ succ rc+      return ( w { readerCount=rc' }+             , return () )+                                                        | otherwise -> do+      (q',mblock) <- enterQueueL me q+      return ( w { queue = q' }+             , safeBlock mblock )++    r@(Readers { readerCounts=rcs }) -> do+      case Map.lookup me rcs of+        Just rc -> do+          rc' <- evaluate $ succ rc+          rcs' <- evaluate $ Map.insert me rc' rcs+          return ( r { readerCounts=rcs' }+                 , return () )++        Nothing -> do+          rcs' <- evaluate $ Map.insert me 1 rcs+          return ( r { readerCounts=rcs' }+                 , return () )+ where+  -- Merge adjacent readers when adding to right end of LockQ+  enterQueueL :: ThreadId -> LockQ -> IO (LockQ,MVar ())+  enterQueueL me qIn = do+    case Seq.viewl qIn of+      (ReaderKind rcs,mblock) :< post -> do+        rcs' <- addMe rcs+        return ((ReaderKind rcs', mblock) <| post,mblock)+      _ -> do+        mblock <- newEmptyMVar+        return ((ReaderKind (Set.singleton me),mblock) <| qIn , mblock)+   where+    -- Paranoid check of design assertion, TODO: remove check+    addMe :: TSet -> IO TSet+    addMe rcs | Set.member me rcs = error (imp "enterQueueL.addMe when already in set" me)+              | otherwise = return (Set.insert me rcs)++-- Six cases:+-- Lock is Free+-- I already have write lock+-- Someone else has write lock+-- 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++-- | 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+-- releaseWrite calls.+--+-- All previous calls to acquireRead by other threads will have succeeded and been released (or+-- interrupted) before this acquireWrite will return.+--+-- The best way to use acquireWrite is to use withWrite instead to ensure releaseWrite will be+-- called exactly once.+--+-- This may block and usually be safely interrupted.  If interrupted then the RWLock will be left+-- unchanged.  The exception to being able to interrupted when this blocks is very subtle: if this+-- thread holds read locks and calls acquireWrite then it will release those read locks and go to+-- the back of the queue to acquire the write lock (it does not get to skip the queue).  While+-- blocking waiting for the write lock to be available this thread may be interrupted.  If not+-- interrupted then the write lock will eventually be acquired, followed by re-acquiring the+-- original number of read locks.  But if acquireWrite is interrupted after releasing read locks+-- then it MUST restore those read locks on the way out.  To do this the internal error handler will+-- use 'uninterruptibleMask_' and a special version of acquireRead that skips to the front of the+-- queue; when the current lock state is a reader this works instantly but when the current lock+-- state is a writer this thread will block in an UNINTERRUPTIBLE state until the current writer is+-- finished.  Once this other writer is finished the error handler will obtain the read locks it+-- needs to allow the error propagation to continue.+acquireWrite :: RWLock -> IO ()+acquireWrite rwl@(RWL rwlVar) = mask_ . join . modifyMVar rwlVar $ \ rwd -> do+  me <- myThreadId+  let safeBlock mblock = (takeMVar mblock) `onException` (releaseWrite' True me rwlVar)+  case rwd of+    FreeLock ->+      return ( Writer { writerID=me, writerCount=1, readerCount=0, queue=Seq.empty }+             , return () )++    w@(Writer { writerID=it, writerCount=wc, queue=q }) | it==me ->+      return ( w { writerCount=(succ wc) }+             , return () )++    {-ditto-}                                           | otherwise -> do+      mblock <- newEmptyMVar+      q' <- evaluate $ q |> (WriterKind me,mblock)+      return ( w { queue=q' }+             , safeBlock mblock )++    Readers { readerCounts=rcs } | Just rc <- Map.lookup me rcs -> do+      return ( rwd+             , withoutReads rc (acquireWrite rwl) )++    r@(Readers { queueR=Nothing }) -> do+      mblock <- newEmptyMVar+      let qr = Just ((me,mblock),Seq.empty)+      return ( r { queueR=qr }+             , safeBlock mblock )++    r@(Readers { queueR=Just (w,q) }) -> do+      mblock <- newEmptyMVar+      q' <- evaluate $ q |> (WriterKind me,mblock)+      return ( r { queueR=Just (w,q') }+             , safeBlock mblock )+ where+  withoutReads :: Int -> IO a -> IO a+  withoutReads n x = foldr (.) id (replicate n withoutRead) $ x+  withoutRead :: IO a -> IO a+  withoutRead = bracket_ (releaseRead rwl >>= either throw return) (acquireReadPriority rwl)++-- format impossible error strings to include standard description prefix+imp :: Show x => String -> x -> String+imp s x = "FairRWLock impossible error: "++s++" : "++show x++{-++subtle bug #1:++When converting from a read lock holding 'rc' read locks to a also holding a write lock, I first wrote:++replicateM_ rc (releaseRead rwl >>= either throw return)+acquireWrite rwl+replicateM_ rc (acquireRead rwl)++Imagine there are rc copies of withRead wrapped around the above:+withRead = liftA2 bracket_ acquireRead (releaseRead >=> either throw return)++Then the acquireWrite blocks and gets interrupted.+The releaseReads in the withRead will see a strange situation (not locked!) and call throw.++What is the answer? reverse the bracket for the release/acquire? Hmm..++-}

@@ -17,6 +17,12 @@ -- 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@@ -26,40 +32,43 @@     ,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_,evaluate,mask_)+import Control.Exception(bracket_,uninterruptibleMask_,mask_) import Data.Typeable(Typeable)  {- design notes are in MSemN.hs -} -data MS = MS { avail :: !Integer     -- ^ This is the quantity available to be taken from the semaphore. Often updated.-             , headWait :: MVar ()   -- ^ The head of the waiter queue blocks on headWait. Never updated.-             }-  deriving (Eq,Typeable)- -- | 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 = MSem { mSem :: !(MVar MS)      -- ^ 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.-                 }+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.-new :: Integer -> IO MSem+--+-- 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-  newHeadWait <- newEmptyMVar+  newMS <- newMVar $! initial   newQueueWait <- newMVar ()-  newMS <- newMVar $! (MS { avail = initial-                          , headWait = newHeadWait })+  newHeadWait <- newEmptyMVar   return (MSem { mSem = newMS-               , queueWait = newQueueWait })+               , 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 :: MSem -> IO a -> IO a+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@@ -69,50 +78,55 @@ -- 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 :: MSem -> IO ()-wait (MSem sem advance) = mask_ $ withMVar advance $ \ () -> do-  todo <- mask_ $ modifyMVar sem $ \ m -> do+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 (m,Nothing)-      Nothing -> if 1 <= avail m-                   then do-                     m' <- evaluate $ m { avail = avail m - 1 }-                     return (m', Nothing)-                   else do-                     return (m, Just (headWait m))+      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.-  case todo of-    Nothing -> return ()-    Just hw -> takeMVar hw -- actually may or may not block, a 'signal' could have already arrived.+  --+  -- 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 :: MSem -> IO ()-signal (MSem sem _) = uninterruptibleMask_ $ modifyMVar_ sem $ \ m -> do+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 avail m < 0-    then evaluate m { avail = avail m + 1 }+  if ms < 0+    then return $! succ ms     else do-      didPlace <- tryPutMVar (headWait m) ()+      didPlace <- tryPutMVar (headWait m) ()  -- ms is never negative       if didPlace-        then return m-        else evaluate m { avail = avail m + 1 }+        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 :: MSem -> IO Integer-peekAvail (MSem sem _) = mask_ $ withMVar sem $  \ m -> do+-- 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 (avail m)+    Nothing -> return ms     Just () -> do putMVar (headWait m) () -- cannot block-                  return (1 + avail m)+                  return $! succ ms

@@ -18,6 +18,11 @@ -- 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@@ -30,6 +35,7 @@     ,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)@@ -54,35 +60,39 @@ -}  -- MS has an invariant that "maybe True (> avail) headWants" is always True.-data MS = MS { avail :: !Integer             -- ^ This is the quantity available to be taken from the semaphore. Often updated.-             , headWants :: !(Maybe Integer) -- ^ If there is waiter then this is Just the amount being waited for. Often updated.-             , headWait :: MVar ()           -- ^ The head of the waiter queue blocks on headWait. Never updated.-             }+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 = MSemN { mSem :: !(MVar MS)      -- ^ 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.-                   }+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 :: Integer -> IO MSemN+new :: Integral i => i -> IO (MSemN i)+{-# SPECIALIZE new :: Int -> IO (MSemN Int) #-}+{-# SPECIALIZE new :: Integer -> IO (MSemN Integer) #-} new initial = do-  newHeadWait <- newEmptyMVar-  newQueueWait <- newMVar ()   newMS <- newMVar $! (MS { avail = initial-                          , headWants = Nothing-                          , headWait = newHeadWait })+                          , headWants = Nothing })+  newQueueWait <- newMVar ()+  newHeadWait <- newEmptyMVar   return (MSemN { mSem = newMS-                , queueWait = newQueueWait })+                , 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 :: MSemN -> Integer -> IO a -> IO a+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 @@ -93,8 +103,10 @@ -- -- Note: A long running pure function will block all other access to the 'MSemN' while it is -- evaluated.-withF :: MSemN -> (Integer -> (Integer,b)) -> ((Integer,b) -> IO a) -> IO a-withF m f = seq f $ bracket (waitF m f)  (\(wanted,_) -> signal m wanted)+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.@@ -103,7 +115,9 @@ -- 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 :: MSemN -> Integer -> IO ()+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,())) @@ -121,32 +135,32 @@ -- -- Note: A long running pure function will block all other access to the 'MSemN' while it is -- evaluated.-waitF :: MSemN -> (Integer -> (Integer,b)) -> IO (Integer,b)-waitF (MSemN sem advance) f = seq f $ mask_ $ withMVar advance $ \ () -> do-  (out@(wanted,_),todo) <- modifyMVar sem $ \ m -> do-    let outVal@(wantedVal,_) = f (avail m)-    -- assert that headDown is Nothing via new or signal or cleanup+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 m+    if wantedVal <= avail ms       then do-        let avail'down = avail m - wantedVal-        m' <- evaluate $ m { avail = avail'down }-        return (m', (outVal,Nothing))+        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-        m' <- evaluate $ m { headWants = Just wantedVal }-        return (m', (outVal,Just (headWait m)))+        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.-  case todo of-    Nothing -> return ()-    Just hw -> do-      let cleanup = uninterruptibleMask_ $ modifyMVar_ sem $ \m -> do-            mStale <- tryTakeMVar (headWait  m)-            let avail' = avail m + maybe 0 (const wanted) mStale-            evaluate $ m {avail = avail', headWants = Nothing}-      takeMVar hw `onException` cleanup -- may not block if a 'signal' has already arrived.+  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@@ -159,7 +173,9 @@ -- '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 :: MSemN -> Integer -> IO ()+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,())) @@ -177,19 +193,20 @@ -- -- Note: A long running pure function will block all other access to the 'MSemN' while it is -- evaluated.-signalF :: MSemN -> (Integer -> (Integer,b)) -> IO (Integer,b)-signalF (MSemN sem _) f = seq f $ modifyMVar sem $ \ m -> do-  let out@(size,_) = f (avail m)-  avail' <- evaluate $ avail m + size -- this forces 'size'-  case headWants m of-    Just wanted | wanted <= avail' -> do-      let avail'down = avail' - wanted-      m' <- evaluate $ m { avail = avail'down, headWants = Nothing }-      putMVar (headWait m') () -- will always succeed without blocking-      return (m',out)-    _ -> do-      m' <- evaluate $ m { avail = avail' }-      return (m',out)+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@@ -200,5 +217,7 @@ -- -- A version of 'peekAvail' that joins the FIFO queue of 'wait' and 'waitF' can be acheived by -- \"waitF m (\x -> (0,x))\"-peekAvail :: MSemN -> IO Integer-peekAvail (MSemN sem _) = withMVar sem (return . avail)+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)