Personal tools

SafeConcurrent

From HaskellWiki

(Difference between revisions)
Jump to: navigation, search
(Propose new code to replace some Control.Concurrent modules)
m
Line 7: Line 7:
 
Specifically, both the wait and signal operations on a semaphore may block. These may then be interrupted by a killThread or other asynchronous exception. Exception safety means that this will never leave the semaphore in a broken state. Exception correctness means that the semaphore does not lose any of its quantity if the waiter is interrupted before the wait operation finished.
 
Specifically, both the wait and signal operations on a semaphore may block. These may then be interrupted by a killThread or other asynchronous exception. Exception safety means that this will never leave the semaphore in a broken state. Exception correctness means that the semaphore does not lose any of its quantity if the waiter is interrupted before the wait operation finished.
   
MSem and MSemN are the proposed replacements for QSem and QSemN.
+
MSem is the proposed replacements for QSem.
  +
  +
A replacement for QSemN is in progress.
   
 
The SampleVar code is also not exception safe. The replacement has not yet been written.
 
The SampleVar code is also not exception safe. The replacement has not yet been written.
Line 13: Line 13:
 
= MSem =
 
= MSem =
   
This code should be exception safe and exception correct. It is derived from the MSemN code by specializing the quantity to 1.
+
This code should be exception safe and exception correct.
   
 
Note that it does not allocate any MVars to manage the waiting queue. Only newMSem allocates them. This should be more efficient than QSem.
 
Note that it does not allocate any MVars to manage the waiting queue. Only newMSem allocates them. This should be more efficient than QSem.
Line 92: Line 92:
 
else throwIO . MSem'Exception $ "MSem.signalMSem: impossible happened, the headWait MVar was full"
 
else throwIO . MSem'Exception $ "MSem.signalMSem: impossible happened, the headWait MVar was full"
 
else return (m { avail = avail m + 1 })
 
else return (m { avail = avail m + 1 })
</haskell>
 
 
= MSemN =
 
 
This is an exception safe and exception correct replacement for QSemN.
 
 
Note that it does not allocate any MVars to manage the waiting queue. Only newMSemN allocates them. This should be more efficient than QSemN.
 
 
<haskell>
 
{-# LANGUAGE DeriveDataTypeable #-}
 
-- |This modules is intended to replace "Control.Concurrent.QSemN". Unlike QSemN, this MSemN module
 
-- should be exception safe and correct. This means that when signalMSemN and waitQSemN operations
 
-- receive an asynchronous exception such as killThread they will leave the MSemN in a non-broken
 
-- state, and will not lose any quantity of the semaphore's value.
 
--
 
-- TODO : drop the MSem suffix from the operations.
 
--
 
-- Author : Chris Kuklewicz < haskell @at@ list .dot. mightyreason .dot. com >
 
-- Copyright : BSD3 2009
 
module MSemN(MSemN,newMSemN,signalMSemN,waitMSemN,MSemN'Exception) where
 
 
import Control.Concurrent.MVar
 
import Control.Exception(Exception,throwIO,block)
 
import Data.Maybe(fromMaybe)
 
import Data.Typeable(Typeable)
 
 
newtype MSemN = MSemN (MVar M)
 
 
data M = M { avail :: Int
 
, headWants :: Maybe Int
 
, headWait :: MVar Int
 
, tailWait :: MVar () }
 
 
newtype MSemN'Exception = MSemN'Exception String deriving (Show,Typeable)
 
instance Exception MSemN'Exception
 
 
-- |'newSemN' allows positive, zero, and negative initial values.
 
newMSemN initial = do
 
newHeadWait <- newEmptyMVar
 
newTailWait <- newMVar ()
 
let m = M { avail = initial
 
, headWants = Nothing
 
, headWait = newHeadWait
 
, tailWait = newTailWait }
 
sem <- newMVar m
 
return (MSemN sem)
 
 
-- |'waitMSemN' allow positive, zero, and negative wanted values. Waiters block in FIFO order.
 
-- This returns when it is the front waiter and the available value is not less than the wanted
 
-- value. If this throws an exception then no quantity of semaphore will be lost.
 
waitMSemN :: MSemN -> Int -> IO ()
 
waitMSemN (MSemN sem) wanted = block $ do
 
-- sem throw?
 
advance <- withMVar sem $ \ m -> return (tailWait m)
 
-- advance throw?
 
withMVar advance $ \ _ -> do
 
-- sem throw? withMVar cleans advance
 
todo <- modifyMVar sem $ \ m -> do
 
-- clean up if previous waiter died
 
mStale <- tryTakeMVar (headWait m)
 
let avail' = avail m + fromMaybe 0 mStale
 
-- ensure the sem is in a sane state
 
if avail' >= wanted
 
then do return (m { avail = avail' - wanted, headWants = Nothing }, Nothing)
 
else do return (m { avail = avail', headWants = Just wanted }, Just (headWait m))
 
case todo of
 
Nothing -> return ()
 
Just wait -> getWanted wait
 
where
 
getWanted wait = do
 
-- takeMVar throw? clean up with next waiter
 
given <- takeMVar wait
 
if given == wanted
 
then return ()
 
else throwIO . MSemN'Exception $ "MSemN.waitMSemN: impossible happened, (wanted,given) == "++ show (wanted,given)
 
 
-- |'signalMSemN' allows positive, zero, and negative size values. If the new total is greater than
 
-- the value waited for then the first waiter is woken. This may momentarily block, and thus may
 
-- throw an exception and leave then MSemN unchanged.
 
signalMSemN :: MSemN -> Int -> IO ()
 
signalMSemN _ 0 = return ()
 
signalMSemN msem@(MSemN sem) size = block $ modifyMVar_ sem $ \ m -> do
 
case headWants m of
 
Nothing -> return (m { avail = avail m + size })
 
Just wanted -> do
 
let avail' = avail m + size
 
if avail' >= wanted
 
then do
 
ok <- tryPutMVar (headWait m) wanted
 
if ok then return (m { avail = avail' - wanted, headWants = Nothing })
 
else throwIO . MSemN'Exception $ "MSemN.signalMSemN: impossible happened, the headWait MVar was full"
 
else return (m { avail = avail' })
 
 
{-
 
 
-- |'queryMSemN' returns two value, the first is the available value in the semaphore. The second
 
-- value, if not Nothing, is Just the value wanted by the first blocked waiter. If the second value
 
-- is Nothing that does not imply there are no blocked waiters.
 
--
 
-- Warning: the first value may be momentarily wrong (and the second Nothing) if the previous waiter
 
-- died between being signaled and receiving its wanted value.
 
queryMSemN :: MSemN -> IO (Int,Maybe Int)
 
queryMSemN (MSemN sem) = withMVar sem $ \ m -> return (avail m, headWants m)
 
 
-}
 
 
</haskell>
 
</haskell>

Revision as of 10:54, 11 April 2009


1 Motivation

The base package (version 3.0.3.1) code for Control.Concurrent.QSem and QSemN and SamepleVar is not exception safe. This page is for holding proposed replacement code.

Specifically, both the wait and signal operations on a semaphore may block. These may then be interrupted by a killThread or other asynchronous exception. Exception safety means that this will never leave the semaphore in a broken state. Exception correctness means that the semaphore does not lose any of its quantity if the waiter is interrupted before the wait operation finished.

MSem is the proposed replacements for QSem.

A replacement for QSemN is in progress.

The SampleVar code is also not exception safe. The replacement has not yet been written.

2 MSem

This code should be exception safe and exception correct.

Note that it does not allocate any MVars to manage the waiting queue. Only newMSem allocates them. This should be more efficient than QSem.

{-# LANGUAGE DeriveDataTypeable #-}
-- |This modules is intended to replace "Control.Concurrent.QSem".  Unlike QSem, this MSem module
-- should be exception safe and correct.  This means that when signalMSemN and waitQSem operations
-- receive an asynchronous exception such as killThread they will leave the MSem in a non-broken
-- state, and will not lose any quantity of the semaphore's value.
--
-- TODO : drop the MSem suffix from the operations.
--
-- Author : Chris Kuklewicz < haskell @at@ list .dot. mightyreason .dot. com >
-- Copyright : BSD3 2009
module MSem(MSem,newMSem,signalMSem,waitMSem,MSem'Exception) where
 
import Control.Concurrent.MVar
import Control.Exception(Exception,throwIO,block)
import Data.Maybe(fromMaybe)
import Data.Typeable(Typeable)
 
newtype MSem = MSem (MVar M)
 
data M = M { avail :: Int
           , headWants :: Bool
           , headWait :: MVar ()
           , tailWait :: MVar () }
 
newtype MSem'Exception = MSem'Exception String deriving (Show,Typeable)
instance Exception MSem'Exception
 
-- |'newSem' allows positive, zero, and negative initial values.
newMSem initial = do
  newHeadWait <- newEmptyMVar
  newTailWait <- newMVar ()
  let m = M { avail = initial
            , headWants = False
            , headWait = newHeadWait
            , tailWait = newTailWait }
  sem <- newMVar m
  return (MSem sem)
 
-- |Waiters block in FIFO order.  This returns when it is the front waiter and the available value
-- is positive.  If this throws an exception then no quantity of semaphore will be lost.
waitMSem :: MSem -> IO ()
waitMSem (MSem sem) = block $ do
  -- sem throw?
  advance <- withMVar sem $ \ m -> return (tailWait m)
  -- advance throw?
  withMVar advance $ \ _ -> do
    -- sem throw? withMVar cleans advance
    todo <- modifyMVar sem $ \ m -> do
              -- clean up if previous waiter died
              mStale <- tryTakeMVar (headWait m)
              let avail' = avail m + maybe 0 (const 1) mStale
              -- ensure the sem is in a sane state
              if avail' >= 1
                then do return (m { avail = avail' - 1, headWants = False }, Nothing)
                else do return (m { avail = avail', headWants = True }, Just (headWait m))
    case todo of
      Nothing -> return ()
      Just wait -> do
        -- takeMVar throw? the headWants is still set to True, withMVar cleans advance
        takeMVar wait
 
-- |Add one to the semaphore, if the new value is greater than 0 then the first waiter is worken.
-- This may momentarily block, and thus may throw an exception and leave then MSem unchanged.
signalMSem :: MSem -> IO ()
signalMSem msem@(MSem sem) = block $ modifyMVar_ sem $ \ m -> do
  case headWants m of
    False -> return (m { avail = avail m + 1 })
    True ->
      if avail m >= 0
        then do
          ok <- tryPutMVar (headWait m) ()
          if ok then return (m { headWants = False })
            else throwIO . MSem'Exception $ "MSem.signalMSem: impossible happened, the headWait MVar was full"
        else return (m { avail = avail m + 1 })