SafeSemaphore (empty) → 0.4

raw patch · 5 files changed

+378/−0 lines, 5 filesdep +HUnitdep +SafeSemaphoredep +basesetup-changed

Dependencies added: HUnit, SafeSemaphore, base

Files

• LICENSE +30/−0
• SafeSemaphore.cabal +24/−0
• Setup.hs +2/−0
• src/Control/Concurrent/MSem.hs +118/−0
• src/Control/Concurrent/MSemN.hs +204/−0

@@ -0,0 +1,30 @@+Copyright (c)2011, Chris Kuklewicz++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Chris Kuklewicz nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,24 @@+Name:                SafeSemaphore+Version:             0.4+Synopsis:            Much safer replacement for QSemN and QSem+Description:         This provides a much safer semaphore than the QSem in base.  Performance has not been compared.+Homepage:            get-URL-from-future+License:             BSD3+License-file:        LICENSE+Author:              Chris Kuklewicz <haskell@list.mightyreason.com>+Maintainer:          Chris Kuklewicz <haskell@list.mightyreason.com>+Category:            Concurrency+Build-type:          Simple+Cabal-version:       >=1.8++Library+  Hs-Source-Dirs: src+  ghc-options:  -Wall -O2 -funbox-strict-fields+  Exposed-modules: Control.Concurrent.MSem+                   Control.Concurrent.MSemN+  Build-depends:  base < 5++Test-Suite TestSafeSemaphore+  type:          exitcode-stdio-1.0+  main-is:       tests/TestKillSem.hs+  build-depends: base < 5, SafeSemaphore, HUnit

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,118 @@+{-# 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.+module Control.Concurrent.MSem+    (MSem+    ,new+    ,with+    ,wait+    ,signal+    ,peekAvail+    ) where++import Control.Concurrent.MVar(MVar,withMVar,modifyMVar,modifyMVar_,newMVar,newEmptyMVar,putMVar,takeMVar,tryTakeMVar,tryPutMVar)+import Control.Exception(bracket_,uninterruptibleMask_,evaluate,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.+                 }+  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+new initial = do+  newHeadWait <- newEmptyMVar+  newQueueWait <- newMVar ()+  newMS <- newMVar $! (MS { avail = initial+                          , headWait = newHeadWait })+  return (MSem { mSem = newMS+               , queueWait = newQueueWait })++-- | '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 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 :: MSem -> IO ()+wait (MSem sem advance) = mask_ $ withMVar advance $ \ () -> do+  todo <- mask_ $ modifyMVar sem $ \ m -> 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))+  -- 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.++-- | '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+  -- mask_ might be as good as uninterruptibleMask_ since nothing below can block+  if avail m < 0+    then evaluate m { avail = avail m + 1 }+    else do+      didPlace <- tryPutMVar (headWait m) ()+      if didPlace+        then return m+        else evaluate m { avail = avail m + 1 }++-- | '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+  extraFlag <- tryTakeMVar (headWait m)+  case extraFlag of+    Nothing -> return (avail m)+    Just () -> do putMVar (headWait m) () -- cannot block+                  return (1 + avail m)

@@ -0,0 +1,204 @@+{-# 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.+--+module Control.Concurrent.MSemN+    (MSemN+    ,new+    ,with+    ,wait+    ,signal+    ,withF+    ,waitF+    ,signalF+    ,peekAvail+    ) where++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 = 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.+             }+  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.+                   }+  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 initial = do+  newHeadWait <- newEmptyMVar+  newQueueWait <- newMVar ()+  newMS <- newMVar $! (MS { avail = initial+                          , headWants = Nothing+                          , headWait = newHeadWait })+  return (MSemN { mSem = newMS+                , queueWait = newQueueWait })++-- | '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 _ 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 :: MSemN -> (Integer -> (Integer,b)) -> ((Integer,b) -> IO a) -> IO a+withF m f = seq 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 :: MSemN -> 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 :: 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+    -- wantedVal gets forced by the (<=) condition here:+    if wantedVal <= avail m+      then do+        let avail'down = avail m - wantedVal+        m' <- evaluate $ m { avail = avail'down }+        return (m', (outVal,Nothing))+      else do+        m' <- evaluate $ m { headWants = Just wantedVal }+        return (m', (outVal,Just (headWait m)))+  -- 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.+  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 :: MSemN -> 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 :: 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)++-- | '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 :: MSemN -> IO Integer+peekAvail (MSemN sem _) = withMVar sem (return . avail)