SafeConcurrent

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.

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 signalMSem 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 woken.
-- 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 })