diff --git a/cbits/domain.c b/cbits/domain.c --- a/cbits/domain.c +++ b/cbits/domain.c @@ -7,17 +7,29 @@ #include #include +/* Some platforms may not have SOCK_CLOEXEC available as an option to socket(), + * so we need to set it explicitly with fcntl() + * */ +static int set_cloexec(int fd) { + int flags = fcntl(fd, F_GETFD, 0); + if (flags == -1) return -1; + return fcntl(fd, F_SETFD, flags | FD_CLOEXEC); +} + int hs_listen_domain_socket(const char *path) { if (strlen(path) >= sizeof(((struct sockaddr_un *)0)->sun_path)) { errno = ENAMETOOLONG; return -1; } + int listen_type = SOCK_STREAM; #ifdef SOCK_NONBLOCK - int fd = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0); -#else - int fd = socket(AF_UNIX, SOCK_STREAM, 0); + listen_type |= SOCK_NONBLOCK; #endif +#ifdef SOCK_CLOEXEC + listen_type |= SOCK_CLOEXEC; +#endif + int fd = socket(AF_UNIX, listen_type, 0); if (fd == -1) { return -1; } @@ -30,6 +42,14 @@ return -1; } #endif +#ifndef SOCK_CLOEXEC + if (set_cloexec(fd) == -1) { + int saved_errno = errno; + close(fd); + errno = saved_errno; + return -1; + } +#endif struct sockaddr_un server_addr; memset(&server_addr, 0, sizeof(struct sockaddr_un)); @@ -61,10 +81,22 @@ return -1; } +#ifdef SOCK_CLOEXEC + int fd = socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); +#else int fd = socket(AF_UNIX, SOCK_STREAM, 0); +#endif if (fd == -1) { return -1; } +#ifndef SOCK_CLOEXEC + if (set_cloexec(fd) == -1) { + int saved_errno = errno; + close(fd); + errno = saved_errno; + return -1; + } +#endif struct sockaddr_un server_addr; memset(&server_addr, 0, sizeof(struct sockaddr_un)); @@ -149,7 +181,13 @@ } if (fds[0].revents & POLLIN) { int afd = accept(listen_fd, NULL, NULL); - if (afd >= 0) return afd; + if (afd >= 0) { + int fl = fcntl(afd, F_GETFL, 0); + if (fl != -1) + fcntl(afd, F_SETFL, fl & ~O_NONBLOCK); + set_cloexec(afd); + return afd; + } /* Listen fd is O_NONBLOCK. an EAGAIN/EINTR here means the queued * connection was droppped poll and accept. */ if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,13 @@ +### 2.0.1 (June 30, 2026) + +- Fix a bug on Darwin where accepted fds would inherit O_NONBLOCK from the listen fd, + causing subsequent reads on the server to fail instantly. +- Fix fds opened by `semaphore-compat` leaking across `exec`. +- Avoid leaking fds from stderr handles in the testsuite. +- Improve documentation +- Drop dependency on `exceptions` +- `releaseSemaphoreToken` is now idempotent on Windows. + ### 2.0.0 (May 22, 2026) - Major release introducing protocol v2: diff --git a/semaphore-compat.cabal b/semaphore-compat.cabal --- a/semaphore-compat.cabal +++ b/semaphore-compat.cabal @@ -3,7 +3,7 @@ name: semaphore-compat version: - 2.0.0 + 2.0.1 license: BSD-3-Clause @@ -48,15 +48,14 @@ exposed-modules: System.Semaphore - System.Semaphore.Internal.Common + System.Semaphore.Internal + System.Semaphore.Internal.Version build-depends: base >= 4.11 && < 4.24 , directory >= 1.3 && < 1.4 - , exceptions - >= 0.7 && < 0.11 , filepath >= 1.4 && < 1.6 @@ -82,6 +81,7 @@ exposed-modules: System.Semaphore.Internal.DomainSocket System.Semaphore.Internal.Posix + System.Semaphore.Internal.Posix.Server default-language: Haskell2010 diff --git a/src/System/Semaphore.hs b/src/System/Semaphore.hs --- a/src/System/Semaphore.hs +++ b/src/System/Semaphore.hs @@ -1,17 +1,44 @@ -{-# LANGUAGE CPP #-} +-- | This library provides a cross-platform implementation of semaphores. +-- +-- Its main role is to provide a cross-platform notion of semaphore that can be +-- used to implement the jobserver\/jobclient model of +-- [GHC proposal #540](https://ghc-proposals.readthedocs.io/en/latest/proposals/0540-jsem.html). +-- +-- Typical usage: +-- +-- - The jobserver creates a 'ServerSemaphore', e.g. using 'createSemaphore'/'freshSemaphore'. +-- - The jobserver retrieves the serialisable 'SemaphoreIdentifier' of this +-- 'ServerSemaphore' by using 'serverClientSemaphore', 'clientSemaphoreName' +-- and 'semaphoreIdentifier'. +-- - The jobserver passes the 'SemaphoreIdentifier' to the jobclient via IPC. +-- - The jobclient uses 'openSemaphore' to obtain the 'ClientSemaphore' +-- corresponding to the 'SemaphoreIdentifier' it has been given. +-- - The jobclient can then request resources from the semaphore using +-- 'waitOnSemaphore' (and return them using 'releaseSemaphoreToken'), or by +-- using the 'withAbstractSem' bracket pattern. +-- - Once the jobclient is done, it uses 'destroyClientSemaphore' to ensure +-- all resources are released (including any remaining tokens). +-- - Once the jobserver is done, it cleans up the resources underlying the +-- semaphore using 'destroyServerSemaphore'. +-- +-- NB: the above usage outline deliberately omits any mention of implicit +-- semaphore tokens: such a notion does not exist in this library, as they are +-- purely an abstraction layer used by the jobserver implementation. +-- See 'destroyClientSemaphore' for more details. module System.Semaphore ( -- * System semaphores -- $server-vs-client - ClientSemaphore, ServerSemaphore, SemaphoreName(..) + ClientSemaphore, ServerSemaphore + , serverClientSemaphore - -- | The name of a client semaphore + , SemaphoreName(..) , clientSemaphoreName + + , SemaphoreIdentifier , semaphoreIdentifier - -- | Retrieve the client semaphore corresponding to a server semaphore - , serverClientSemaphore -- ** Creating a semaphore , createSemaphore, freshSemaphore @@ -19,7 +46,7 @@ -- ** Opening a semaphore , SemaphoreToken , openSemaphore - , SemaphoreIdentifier, parseSemaphoreIdentifier + , parseSemaphoreIdentifier , SemaphoreProtocolVersion(..) , semaphoreVersion , versionsAreCompatible @@ -33,23 +60,33 @@ -- ** Releasing resources , releaseSemaphoreToken - -- $destroying , destroyClientSemaphore, destroyServerSemaphore + -- $implicit-tokens -- * Abstract semaphores , AbstractSem(..) , withAbstractSem ) where +-- base +import Control.Exception ( bracket, bracket_ ) +import Data.List.NonEmpty ( NonEmpty(..) ) + +-- semaphore-compat +import System.Semaphore.Internal +import System.Semaphore.Internal.Version + +-------------------------------------------------------------------------------- + {- $server-vs-client -Since version 2 of @semaphore-compat@, we distinguish between two kinds -of semaphores: +Since version 2 of @semaphore-compat@, the library distinguishes between two +kinds of semaphores: - When a jobserver creates a semaphore via 'createSemaphore', it obtains a 'ServerSemaphore'. - - Jobclients (which open a pre-existing semaphore via 'openSemaphore') - obtain a 'ClientSemaphore'. + - Jobclients, which are passed the 'SemaphoreIdentifier' of a pre-existing + semaphore, obtain a 'ClientSemaphore' via 'openSemaphore'. When the jobserver wants to also act as a jobclient, it can use 'serverClientSemaphore' to obtain the 'ClientSemaphore' corresponding @@ -59,51 +96,42 @@ semaphore resources held by all clients. -} -{- $destroying - -Destroying a semaphore releases the implicit token held by the -semaphore. - -For example, when @cabal-install@ invokes @ghc@, the @ghc@ process -automatically has one semaphore token at the start (the implicit -token), and can request further tokens with 'waitOnSemaphore'. The -tokens acquired by 'waitOnSemaphore' are released by -'releaseSemaphoreToken', while the implicit token is released by -'destroyClientSemaphore'. --} +{- $implicit-tokens --- base -import Data.List.NonEmpty ( NonEmpty(..) ) +The GHC jobserver protocol described in [GHC proposal #540](https://ghc-proposals.readthedocs.io/en/latest/proposals/0540-jsem.html) +specifies the notion of __implicit semaphore tokens__. --- exceptions -import qualified Control.Monad.Catch as MC + 1. To invoke a jobclient, the jobserver must be in possession of at least one + semaphore token. We consider that the jobserver passes on ownership of this + token to the jobclient, even though no interaction with the semaphore takes + place. This is the __implicit token__. + 2. Each jobclient thus starts with one available token (the implicit token), + and can request more tokens by waiting on the semaphore. -import System.Semaphore.Internal.Common +A jobclient calling 'destroyClientSemaphore' signals to the jobserver that the +client is done doing work. The jobserver thus regains control of the implicit +token, which it can now use to spawn other jobclients. -#if defined(mingw32_HOST_OS) -import System.Semaphore.Internal.Win32 -#elif defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) -import System.Semaphore.Internal.Unsupported -#else -import System.Semaphore.Internal.Posix -#endif +Implicit tokens are purely an accounting abstraction that we superimpose on top +of semaphore token ownership for the purposes of implementing the jobserver +protocol. They are not part of this library. +-} --------------------------------------- -- Version compatibility -- | Check whether two semaphore protocol versions are compatible. --- Only identical versions are compatible. versionsAreCompatible :: SemaphoreProtocolVersion -> SemaphoreProtocolVersion -> Bool -versionsAreCompatible a b = a == b +versionsAreCompatible a b = + -- For now, only identical versions are compatible, but keeping this + -- check abstract (via 'versionsAreCompatible') gives us more flexibility + -- when evolving the library. + a == b --------------------------------------- -- System-specific semaphores -- | Create a new semaphore with the given label and initial token count. --- --- On POSIX, crash recovery is automatic: disconnected clients' tokens --- are returned to the pool. On Windows, tokens held by a crashed --- client are permanently lost. createSemaphore :: String -- ^ label -> Int -- ^ number of tokens on the semaphore -> IO (Either SemaphoreError ServerSemaphore) @@ -121,7 +149,7 @@ -> Int -- ^ number of tokens on the semaphore -> IO (Either SemaphoreError ServerSemaphore) freshSemaphore prefix init_toks = do - seed <- getTimeSeed + seed <- get_time_seed go 0 (seedStrings seed) where go :: Int -> NonEmpty String -> IO (Either SemaphoreError ServerSemaphore) @@ -141,13 +169,10 @@ | otherwise -> return (Left err) --- | Open a semaphore from its 'SemaphoreIdentifier'. +-- | Open a pre-existing semaphore. -- --- The identifier should normally begin with a version prefix @v\-@. --- An unversioned identifier is treated as v1 for backwards --- compatibility. Returns @Left SemaphoreIncompatibleVersion@ if the --- identifier's protocol version is not compatible with this build of --- @semaphore-compat@. +-- Returns @Left SemaphoreIncompatibleVersion@ if the semaphore protocol +-- version is not compatible with the current version of @semaphore-compat@. openSemaphore :: SemaphoreIdentifier -> IO (Either SemaphoreError ClientSemaphore) openSemaphore ident = case parseSemaphoreIdentifier ident of @@ -168,7 +193,7 @@ -- | Acquire a token, run an action, then release the token. Exception safe. withSemaphoreToken :: ClientSemaphore -> (SemaphoreToken -> IO a) -> IO a -withSemaphoreToken sem = MC.bracket (waitOnSemaphore sem) releaseSemaphoreToken +withSemaphoreToken sem = bracket (waitOnSemaphore sem) releaseSemaphoreToken seedStrings :: Int -> NonEmpty String seedStrings seed = fmap ( \ i -> iToBase62 (i + seed) ) (0 :| [1..]) @@ -183,5 +208,10 @@ , releaseSem :: IO () } +-- | Acquire/release bracket pattern: acquire a token, perform an action, +-- and release the token. +-- +-- Guarantees that the token is released in case that the inner action throws +-- an exception. withAbstractSem :: AbstractSem -> IO b -> IO b -withAbstractSem s = MC.bracket_ (acquireSem s) (releaseSem s) +withAbstractSem s = bracket_ (acquireSem s) (releaseSem s) diff --git a/src/System/Semaphore/Internal.hs b/src/System/Semaphore/Internal.hs new file mode 100644 --- /dev/null +++ b/src/System/Semaphore/Internal.hs @@ -0,0 +1,136 @@ +{-# LANGUAGE CPP #-} + +-- This module selects the internal implementation (POSIX, Windows, wasm/js) +-- with CPP and wraps it with user-facing types and documentation. +-- +-- Benefits: +-- +-- - All the documentation lives in a single place, instead of being duplicated +-- across every implementation. +-- - We can document platform-dependent behaviour, as opposed to e.g. the +-- Hackage page only showing POSIX-specific documentation. + +-- | Platform-independent internal API. +module System.Semaphore.Internal + ( ClientSemaphore(..), ServerSemaphore(..), SemaphoreToken(..) + , clientSemaphoreName, serverClientSemaphore + + , waitOnSemaphore, tryWaitOnSemaphore + , releaseSemaphoreToken + + , destroyClientSemaphore, destroyServerSemaphore + + , getSemaphoreValue + + -- * Internals + , create_sem, open_sem_raw + , Impl.get_time_seed + ) where + +-- base +import Data.Coerce ( coerce ) +import GHC.Stack ( HasCallStack ) + +-- semaphore-compat +import System.Semaphore.Internal.Version + +#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) +import qualified System.Semaphore.Internal.Unsupported as Impl +#elif defined(mingw32_HOST_OS) +import qualified System.Semaphore.Internal.Win32 as Impl +#else +import qualified System.Semaphore.Internal.Posix as Impl +#endif + +--------------------------------------- +-- User-facing API + +-- | A client-side semaphore: the handle a jobclient uses to acquire and release +-- tokens from a semaphore created by a jobserver. +-- +-- Retrieve the underlying name with 'clientSemaphoreName'. +newtype ClientSemaphore = ClientSemaphore Impl.ClientSemaphore + +-- | A server-side semaphore: a 'ClientSemaphore' together with the book-keeping +-- needed to track resource usage by all clients. +-- +-- Retrieve the corresponding client semaphore with 'serverClientSemaphore'. +newtype ServerSemaphore = ServerSemaphore Impl.ServerSemaphore + +-- | A held semaphore token: evidence of one unit of resource acquired from a +-- semaphore. +-- +-- Use 'releaseSemaphoreToken' or 'withSemaphoreToken' to ensure prompt release +-- of semaphore tokens. +-- +-- Platform-dependent behaviour: +-- +-- - On POSIX, if all references to a 'SemaphoreToken' are dropped without +-- being released, a finalizer returns the token to the semaphore. +-- - On Windows, tokens held by a crashed client are permanently lost. +newtype SemaphoreToken = SemaphoreToken Impl.SemaphoreToken + +-- | Retrieve the underlying name of a client-side semaphore. +clientSemaphoreName :: ClientSemaphore -> SemaphoreName +clientSemaphoreName = coerce Impl.clientSemaphoreName + +-- | Retrieve the client-side semaphore corresponding to a server-side semaphore. +serverClientSemaphore :: ServerSemaphore -> ClientSemaphore +serverClientSemaphore = coerce Impl.serverClientSemaphore + +-- | Acquire a token from the semaphore, blocking until one is available. +-- +-- The returned 'SemaphoreToken' must be released with 'releaseSemaphoreToken'. +-- For prompt and predictable release of resources, callers should use +-- 'System.Semaphore.withSemaphoreToken' or 'releaseSemaphoreToken'. +-- +-- This operation is interruptible: it can be cancelled by +-- 'Control.Concurrent.throwTo', 'Control.Concurrent.killThread', etc. +-- If interrupted, any transiently acquired token is automatically returned. +waitOnSemaphore :: HasCallStack => ClientSemaphore -> IO SemaphoreToken +waitOnSemaphore = coerce Impl.waitOnSemaphore + +-- | Try to acquire a token from the semaphore without blocking. +-- +-- Returns @Just token@ if a token was available, @Nothing@ otherwise. +-- +-- This is __not__ an interruptible operation, but that should not be a problem: +-- this function is not expected to block for long, as the server is supposed +-- to respond immediately. +tryWaitOnSemaphore :: HasCallStack => ClientSemaphore -> IO (Maybe SemaphoreToken) +tryWaitOnSemaphore = coerce Impl.tryWaitOnSemaphore + +-- | Release a semaphore token, returning it to the pool. +-- +-- Idempotent: a second call on the same token is a safe no-op. +releaseSemaphoreToken :: HasCallStack => SemaphoreToken -> IO () +releaseSemaphoreToken = coerce Impl.releaseSemaphoreToken + +-- | Destroy a client-side semaphore. +-- +-- This is an idempotent operation: double calls to 'destroyClientSemaphore' +-- do not cause any problems. +destroyClientSemaphore :: ClientSemaphore -> IO () +destroyClientSemaphore = coerce Impl.destroyClientSemaphore + +-- | Destroy a server-side semaphore. +-- +-- Idempotent. Not interruptible. +destroyServerSemaphore :: ServerSemaphore -> IO () +destroyServerSemaphore = coerce Impl.destroyServerSemaphore + +-- | Query the current semaphore value (how many tokens it has available). +-- +-- This is mainly for debugging use, as it is easy to introduce race conditions +-- when nontrivial program logic depends on the value returned by this function. +getSemaphoreValue :: ServerSemaphore -> IO Int +getSemaphoreValue = coerce Impl.getSemaphoreValue + +-------------------------------------------------------------------------------- +-- Internals + +create_sem :: SemaphoreName -> Int -> IO (Either SemaphoreError ServerSemaphore) +create_sem = coerce Impl.create_sem + +open_sem_raw :: SemaphoreName -> IO (Either SemaphoreError ClientSemaphore) +open_sem_raw = coerce Impl.open_sem_raw diff --git a/src/System/Semaphore/Internal/Common.hs b/src/System/Semaphore/Internal/Common.hs deleted file mode 100644 --- a/src/System/Semaphore/Internal/Common.hs +++ /dev/null @@ -1,142 +0,0 @@ -{-# LANGUAGE BangPatterns #-} -{-# LANGUAGE CPP #-} -{-# LANGUAGE MagicHash #-} - -module System.Semaphore.Internal.Common - ( SemaphoreName(..) - , SemaphoreIdentifier - , SemaphoreProtocolVersion(..) - , SemaphoreError(..) - , semaphoreVersion - , semaphoreIdentifier - , parseSemaphoreIdentifier - , getSemaphoreSocketPath - , iToBase62 - ) where - --- base -import Control.Exception ( Exception, IOException ) -import GHC.Exts ( Char(..), Int(..), indexCharOffAddr# ) - --- directory -import System.Directory ( getTemporaryDirectory, createDirectoryIfMissing ) - --- filepath -import System.FilePath ( () ) - ---------------------------------------- --- Types - --- | The protocol version of a semaphore. -newtype SemaphoreProtocolVersion = - SemaphoreProtocolVersion { getSemaphoreProtocolVersion :: Int } - deriving (Eq, Ord, Show) - --- | A semaphore name: a protocol version and an unversioned name string. -data SemaphoreName = SemaphoreName - { semaphoreProtocolVersion :: !SemaphoreProtocolVersion - , unversionedSemaphoreNameString :: !String - } deriving (Eq, Show) - --- | The identifier string of a semaphore, as serialised for transport --- between processes (e.g. on a command line via @-jsem@). --- --- For version 1 this is a bare name; for version @N@ (with @N >= 2@) --- this is @\"v\-\\"@. -type SemaphoreIdentifier = String - --- | Errors that can occur when creating or opening a semaphore. -data SemaphoreError - = SemaphoreAlreadyExists - { semaphoreErrorIdentifier :: !SemaphoreIdentifier - } - | SemaphoreDoesNotExist - { semaphoreErrorIdentifier :: !SemaphoreIdentifier - } - | SemaphoreIncompatibleVersion - { semaphoreErrorActualVersion :: !SemaphoreProtocolVersion - , semaphoreErrorExpectedVersion :: !SemaphoreProtocolVersion - } - | SemaphoreOtherError - { semaphoreOtherError :: !IOException - } - deriving (Show, Eq) - -instance Exception SemaphoreError - ---------------------------------------- --- Version negotiation - --- | The protocol version on this platform. --- --- The version tracks the IPC mechanism, not the library version: --- --- * __POSIX:__ 2 (domain sockets, replacing v1 system semaphores). --- * __Windows:__ 1 (Win32 named semaphores, unchanged from v1). --- * __Unsupported platforms:__ 0 (no compatible IPC backend). --- --- Because the version is 1 on Windows, 'semaphoreIdentifier' produces --- a bare name (no @v\-@ prefix), matching the v1 format. -semaphoreVersion :: SemaphoreProtocolVersion -#if defined(mingw32_HOST_OS) -semaphoreVersion = SemaphoreProtocolVersion 1 -#elif defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) -semaphoreVersion = SemaphoreProtocolVersion 0 -#else -semaphoreVersion = SemaphoreProtocolVersion 2 -#endif - - --- | The serialised identifier of a 'SemaphoreName' for transport --- between processes. --- --- For version 1 this is the bare name; for version @N@ (@N >= 2@) the --- name is prefixed with @v\-@. -semaphoreIdentifier :: SemaphoreName -> SemaphoreIdentifier -semaphoreIdentifier sn - | ver <= 1 = unversionedSemaphoreNameString sn - | otherwise = "v" ++ show ver ++ "-" ++ unversionedSemaphoreNameString sn - where - ver = getSemaphoreProtocolVersion (semaphoreProtocolVersion sn) - --- | Parse a 'SemaphoreIdentifier' into a 'SemaphoreName'. --- --- Returns @Nothing@ for unversioned strings (which should be treated as --- v1 by the caller's compatibility logic). -parseSemaphoreIdentifier :: SemaphoreIdentifier -> Maybe SemaphoreName -parseSemaphoreIdentifier ('v':rest) = - case reads rest of - [(n, '-':name)] -> Just (SemaphoreName (SemaphoreProtocolVersion n) name) - _ -> Nothing -parseSemaphoreIdentifier _ = Nothing - ---------------------------------------- --- Utilities - --- | Convert an 'Int' to a base-62 string (digits @0@–@9@, @a@–@z@, @A@–@Z@). -iToBase62 :: Int -> String -iToBase62 m = go m' "" - where - m' - | m == minBound = maxBound - | otherwise = abs m - go n cs | n < 62 - = let !c = chooseChar62 n - in c : cs - | otherwise - = let !(!q, r) = quotRem n 62 - !c = chooseChar62 r - in go q (c : cs) - - chooseChar62 :: Int -> Char - {-# INLINE chooseChar62 #-} - chooseChar62 (I# n) = C# (indexCharOffAddr# chars62 n) - chars62 = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"# - --- | The socket file path for a semaphore. -getSemaphoreSocketPath :: SemaphoreName -> IO FilePath -getSemaphoreSocketPath sn = do - tmp <- getTemporaryDirectory - let dir = tmp "semaphore-compat" - createDirectoryIfMissing True dir - return (dir unversionedSemaphoreNameString sn) diff --git a/src/System/Semaphore/Internal/DomainSocket.hs b/src/System/Semaphore/Internal/DomainSocket.hs --- a/src/System/Semaphore/Internal/DomainSocket.hs +++ b/src/System/Semaphore/Internal/DomainSocket.hs @@ -1,5 +1,6 @@ {-# LANGUAGE ForeignFunctionInterface #-} +-- | FFI glue code for Unix domain sockets. module System.Semaphore.Internal.DomainSocket ( connectDomainSocket , listenDomainSocket @@ -21,6 +22,8 @@ -- unix import System.Posix.Types ( Fd(..) ) + +-------------------------------------------------------------------------------- foreign import ccall safe "hs_connect_domain_socket" c_connectDomainSocket :: CString -> IO CInt diff --git a/src/System/Semaphore/Internal/Posix.hs b/src/System/Semaphore/Internal/Posix.hs --- a/src/System/Semaphore/Internal/Posix.hs +++ b/src/System/Semaphore/Internal/Posix.hs @@ -1,18 +1,19 @@ {-# LANGUAGE PatternSynonyms #-} -{-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeApplications #-} +-- | POSIX backend: semaphores implemented on top of Unix domain sockets. module System.Semaphore.Internal.Posix ( ClientSemaphore(..), ServerSemaphore(..) , SemaphoreToken(..) - , create_sem, open_sem_raw , waitOnSemaphore, tryWaitOnSemaphore , releaseSemaphoreToken , destroyClientSemaphore, destroyServerSemaphore , getSemaphoreValue - , getTimeSeed + + , create_sem, open_sem_raw + , get_time_seed ) where -- base @@ -20,93 +21,92 @@ ( ThreadId, forkIOWithUnmask, killThread ) import Control.Concurrent.MVar ( MVar, mkWeakMVar, newEmptyMVar, newMVar, putMVar - , readMVar, takeMVar, tryTakeMVar ) -import Control.Exception ( IOException ) -import Control.Monad -import Data.Word ( Word8 ) + , takeMVar, tryTakeMVar ) +import Control.Exception + ( IOException, SomeException + , finally, mask_, onException, throw, try, uninterruptibleMask_ + ) +import Control.Monad ( void ) import Data.Bits ( xor ) -import Foreign.C.Error ( Errno(Errno), eCONNABORTED ) import GHC.Clock ( getMonotonicTimeNSec ) -import GHC.IO.Exception ( ioe_errno ) import GHC.Stack ( HasCallStack ) -import System.IO.Error ( isFullError ) --- exceptions -import qualified Control.Monad.Catch as MC +-- directory +import System.Directory ( doesPathExist ) -- stm import Control.Concurrent.STM - ( TVar, atomically, newTVarIO, readTVar, readTVarIO - , modifyTVar', writeTVar, retry ) - --- directory -import System.Directory ( doesPathExist ) + ( TVar, newTVarIO, readTVarIO ) -- unix -import System.Posix.IO ( closeFd, createPipe ) +import System.Posix.IO ( closeFd, createPipe, setFdOption, FdOption(CloseOnExec) ) import System.Posix.Files ( removeLink ) import System.Posix.Types ( Fd ) import System.Posix.Process ( getProcessID ) -import System.Semaphore.Internal.Common +-- semaphore-compat +import System.Semaphore.Internal.Version import System.Semaphore.Internal.DomainSocket ( connectDomainSocket, listenDomainSocket - , pollAcceptSocket, AcceptResult(..) , fdReadByte, fdWriteByte , fdShutdown ) +import System.Semaphore.Internal.Posix.Server + ( serverLoop + , pattern CmdWait, pattern CmdTryWait, pattern CmdRelease + , pattern RspOk, pattern RspFail + ) --- | A semaphore identity (name + socket path). --- Each operation that needs a connection opens one internally. +-------------------------------------------------------------------------------- + +-- A client is just an identity (name + socket path); each operation that needs +-- a connection opens one internally. data ClientSemaphore = ClientSemaphore { clientSemaphoreName :: !SemaphoreName , semSocketPath :: !FilePath } --- | A held semaphore token, bound to one acquired resource. --- --- If all references to the 'SemaphoreToken' are dropped without being --- released, a finalizer closes the underlying connection and the server --- returns the token to the pool. Use 'releaseSemaphoreToken' or --- 'System.Semaphore.withSemaphoreToken' for prompt release rather than --- relying on GC timing. --- --- The fd is held in an internal 'MVar' so 'releaseSemaphoreToken' takes --- ownership atomically: a second (erroneous) release is a safe no-op. +-- The fd is held in the 'MVar' so 'releaseSemaphoreToken' takes ownership +-- atomically: a second (erroneous) release is a safe no-op. newtype SemaphoreToken = SemaphoreToken { tokenFdLock :: MVar Fd + -- NB: the 'Fd' is held in an 'MVar' to allow 'releaseSemaphoreToken' + -- to takes ownership of the 'Fd' atomically. This ensures that a + -- second (erroneous) release is a safe no-op (avoiding Fd double-close bugs). } --- | A server-side semaphore (owns the server thread, listen socket, and token pool). data ServerSemaphore = ServerSemaphore - { serverClientSemaphore :: !ClientSemaphore - , serverThreadId :: !ThreadId - , serverPool :: !(TVar Int) - , serverState :: !(MVar ServerState) -- ^ MVar is emptied when resources/fds are freed to prevent double close + { serverClientSemaphore :: !ClientSemaphore + , serverThreadId :: !ThreadId + , serverPool :: !(TVar Int) + , serverState :: !(MVar ServerState) + -- NB: the MVar is emptied when resources/fds are freed to prevent double close } data ServerState = ServerState { serverListenFd :: !Fd , serverCancelFd :: !Fd - -- ^ Write end of the cancel pipe. Writing a byte signals the + -- Write end of the cancel pipe. Writing a byte signals the -- server loop to exit its poll-accept. } create_sem :: SemaphoreName -> Int -> IO (Either SemaphoreError ServerSemaphore) create_sem sem_nm init_toks = do - mb_res <- MC.try @_ @IOException $ MC.mask_ $ do + mb_res <- try @IOException $ mask_ $ do socketPath <- getSemaphoreSocketPath sem_nm listenFd <- listenDomainSocket socketPath -- ^^ creates the socket file, unlink it too on failure. let cleanupListen = do - void $ MC.try @_ @IOException $ closeFd listenFd - void $ MC.try @_ @IOException $ removeLink socketPath - flip MC.onException cleanupListen $ do + void $ try @IOException $ closeFd listenFd + void $ try @IOException $ removeLink socketPath + flip onException cleanupListen $ do pool <- newTVarIO init_toks (cancelRd, cancelWr) <- createPipe + setFdOption cancelRd CloseOnExec True + setFdOption cancelWr CloseOnExec True tid <- forkIOWithUnmask $ \unmask -> unmask (serverLoop pool listenFd cancelRd) - `MC.finally` closeFd cancelRd + `finally` closeFd cancelRd stateVar <- newMVar ServerState { serverListenFd = listenFd , serverCancelFd = cancelWr @@ -124,7 +124,7 @@ open_sem_raw :: SemaphoreName -> IO (Either SemaphoreError ClientSemaphore) open_sem_raw nm = do - mb_res <- MC.try @_ @IOException $ do + mb_res <- try @IOException $ do socketPath <- getSemaphoreSocketPath nm exists <- doesPathExist socketPath return (socketPath, exists) @@ -137,65 +137,50 @@ , semSocketPath = socketPath } --- | Acquire a token from the semaphore, blocking until one is available. --- --- This operation is interruptible: it can be cancelled by --- 'Control.Concurrent.throwTo', 'Control.Concurrent.killThread', etc. If --- interrupted, any transiently acquired token is automatically returned to the --- pool. --- --- For prompt and predictable release of resources, callers should use --- 'System.Semaphore.withSemaphoreToken' or `releaseSemaphoreToken'. waitOnSemaphore :: HasCallStack => ClientSemaphore -> IO SemaphoreToken waitOnSemaphore sem = do resultVar <- newEmptyMVar -- Mask exceptions until we get to the interruptible takeMVar - MC.mask_ $ do + mask_ $ do fd <- connectDomainSocket (semSocketPath sem) -- The read() runs in a forked thread workerTid <- forkIOWithUnmask $ \_ -> do - res <- MC.try @_ @MC.SomeException $ do + res <- try @SomeException $ do fdWriteByte fd CmdWait fdReadByte fd putMVar resultVar res -- uninterruptibleMask_: killThread is interruptible, and a -- second async between killThread and closeFd would leak fd. - let cleanup = MC.uninterruptibleMask_ $ do + let cleanup = uninterruptibleMask_ $ do -- shutdown(SHUT_RDWR) causes the worker's read() to return EOF immediately - void $ MC.try @_ @IOException $ fdShutdown fd + void $ try @IOException $ fdShutdown fd -- Wait for the worker to exit before closing the fd. -- this prevents double close bugs killThread workerTid -- Finally close the fd - void $ MC.try @_ @IOException $ closeFd fd + void $ try @IOException $ closeFd fd -- We achieve interruptiblity by relying on the interruptiblity of takeMVar -- The worker thread is blocked on read(), fdShutdown in cleanup interrupts -- it if we are interrupt by an async exception here. - res <- takeMVar resultVar `MC.onException` cleanup + res <- takeMVar resultVar `onException` cleanup case res of Right resp | resp == RspOk -> mkToken fd - | otherwise -> do void $ MC.try @_ @IOException $ closeFd fd + | otherwise -> do void $ try @IOException $ closeFd fd fail $ "semaphore-compat: unexpected response in waitOnSemaphore: " ++ show resp - Left e -> do void $ MC.try @_ @IOException $ closeFd fd - MC.throwM e + Left e -> do void $ try @IOException $ closeFd fd + throw e --- | Try to acquire a token from the semaphore without blocking. --- --- Returns @Just token@ if a token was available, @Nothing@ otherwise. --- --- Not interruptible, but this shouldn't block for long as the server is --- supposed to respond immediately. tryWaitOnSemaphore :: HasCallStack => ClientSemaphore -> IO (Maybe SemaphoreToken) tryWaitOnSemaphore sem = - MC.mask_ $ do + mask_ $ do fd <- connectDomainSocket (semSocketPath sem) - resp <- flip MC.onException (closeFd fd) $ do + resp <- flip onException (closeFd fd) $ do fdWriteByte fd CmdTryWait fdReadByte fd case resp of RspOk -> Just <$> mkToken fd - _ -> do void $ MC.try @_ @IOException $ closeFd fd + _ -> do void $ try @IOException $ closeFd fd return Nothing mkToken :: Fd -> IO SemaphoreToken @@ -207,25 +192,19 @@ Nothing -> return () -- already released -- Closing the fd triggers the server's disconnect handling, -- which returns the token to the pool. - Just fd' -> void $ MC.try @_ @IOException $ closeFd fd' + Just fd' -> void $ try @IOException $ closeFd fd' return (SemaphoreToken fdVar) --- | Release a semaphore token, returning it to the pool. --- --- Sends a release command on the token's connection, then closes it. --- Idempotent: a second call on the same token is a safe no-op. --- --- Not interruptible; only returns when the release has succeeded. releaseSemaphoreToken :: HasCallStack => SemaphoreToken -> IO () releaseSemaphoreToken (SemaphoreToken fdVar) = - MC.mask_ $ do + mask_ $ do mb <- tryTakeMVar fdVar case mb of Nothing -> return () -- already released Just fd -> do resp <- (do fdWriteByte fd CmdRelease fdReadByte fd - ) `MC.finally` (void $ MC.try @_ @IOException $ closeFd fd) + ) `finally` (void $ try @IOException $ closeFd fd) case resp of RspOk -> return () -- myCount <= 0 on the server means the token is effectively @@ -233,19 +212,12 @@ RspFail -> return () _ -> fail "semaphore-compat: unexpected response in releaseSemaphoreToken" --- | Destroy a client-side semaphore. --- -- On POSIX this is a no-op: 'ClientSemaphore' holds no live connection. destroyClientSemaphore :: ClientSemaphore -> IO () destroyClientSemaphore _ = return () --- | Destroy a server-side semaphore. --- --- Idempotent. Subsequent calls after the first one are no-ops --- Not interruptible. Only returns when the server and all resources --- have been cleaned up destroyServerSemaphore :: ServerSemaphore -> IO () -destroyServerSemaphore server = MC.uninterruptibleMask_ $ do +destroyServerSemaphore server = uninterruptibleMask_ $ do -- we justify the uninterruptibleMask_ here by analysis of the server, which must exit -- in a bounded amount of time once it receives the cancel signal on serverCancelFd -- It has a bounded number of child threads it needs to cleanup before returning, @@ -259,189 +231,17 @@ Just ServerState{..} -> do let path = semSocketPath $ serverClientSemaphore server -- Signal the server loop to exit pollAcceptSocket, then wait for it. - void $ MC.try @_ @IOException $ fdWriteByte serverCancelFd 0 - void $ MC.try @_ @IOException $ closeFd serverCancelFd + void $ try @IOException $ fdWriteByte serverCancelFd 0 + void $ try @IOException $ closeFd serverCancelFd killThread (serverThreadId server) - void $ MC.try @_ @IOException $ closeFd serverListenFd - void $ MC.try @_ @IOException $ removeLink path + void $ try @IOException $ closeFd serverListenFd + void $ try @IOException $ removeLink path --- | Query the current semaphore value (how many tokens it has available). --- --- This is mainly for debugging use, as it is easy to introduce race conditions --- when nontrivial program logic depends on the value returned by this function. getSemaphoreValue :: ServerSemaphore -> IO Int getSemaphoreValue server = readTVarIO (serverPool server) -getTimeSeed :: IO Int -getTimeSeed = do +get_time_seed :: IO Int +get_time_seed = do ns <- getMonotonicTimeNSec pid <- getProcessID return $ fromIntegral ns `xor` fromIntegral pid - ---------------------------------------- --- Server (Unix domain socket) --- --- The server manages a shared token pool (TVar Int) and accepts multiple --- client connections, each served on its own thread. --- --- Protocol (SOCK_STREAM, one byte per command): --- --- "-" Wait (blocking acquire). Decrements semaphore; replies ".". --- "?" Try-wait. Decrements if positive and replies "."; otherwise replies "!". --- "+" Release. Increments pool; replies ".". Rejected with "!" if --- myCount <= 0 (client has not acquired any tokens on this connection). --- --- Unrecognised bytes are rejected with RspFail. --- --- Per-connection token tracking: the server counts tokens held by each --- connection (myCount). On disconnect (EOF / ResourceVanished) held --- tokens are returned to the pool, so a crashing client cannot leak --- tokens. --- --- Connections are tied to SemaphoreToken, so each token (obtained by waitOnSemaphore) --- has its own connection --- ---------------------------------------- --- Protocol byte constants - -pattern CmdWait, CmdTryWait, CmdRelease :: Word8 -pattern CmdWait = 0x2D -- '-' -pattern CmdTryWait = 0x3F -- '?' -pattern CmdRelease = 0x2B -- '+' - -pattern RspOk, RspFail :: Word8 -pattern RspOk = 0x2E -- '.' -pattern RspFail = 0x21 -- '!' - --- | Children of the server, to be cleaned up --- childFdLock is full when the 'Fd' is still known to be valid, --- and subject to being shutdown or closed -data Child = Child - { childThread :: !(MVar ThreadId) - , childFdLock :: !(MVar Fd) - } - -serverLoop :: TVar Int -> Fd -> Fd -> IO () -serverLoop pool listenFd cancelFd = do - children <- newTVarIO ([] :: [Child]) - loop children `MC.finally` killChildren children - where - loop children = do - continueLoop <- MC.mask_ $ do - r <- acceptWithRetry - case r of - AcceptCancelled -> return False - AcceptedFd clientFd -> do - forkServeChild children clientFd - return True - when continueLoop $ loop children - - -- pollAcceptSocket only returns if the accept succeeded, or cancellation - -- was signalled via the cancel pipe. - acceptWithRetry :: IO AcceptResult - acceptWithRetry = pollAcceptSocket listenFd cancelFd `MC.catch` handleIOError - - -- Retry accept on transient errors. - handleIOError :: IOException -> IO AcceptResult - handleIOError e - -- 'isFullError' catches ResourceExhausted (EMFILE/ENFILE/ENOBUFS/ENOMEM and more that accept doesn't produce but are harmless to retry). - | isFullError e = acceptWithRetry - -- ECONNABORTED is also transient but categorised as OtherError, we additionaly match that. - | Just err <- ioe_errno e - , Errno err == eCONNABORTED = acceptWithRetry - -- EINTR is absorbed in hs_poll_accept. - -- everything else (EBADF, EINVAL, ENOTSOCK, ...) is rethrown. - | otherwise = MC.throwM e - - forkServeChild children clientFd = do - fdLock <- newMVar clientFd - tidVar <- newEmptyMVar - let child = Child tidVar fdLock - atomically $ modifyTVar' children (child :) - childTid <- forkIOWithUnmask $ \unmask -> - serve unmask pool children clientFd child - putMVar tidVar childTid - - -- Interrupt all children blocked on a read from the FD, then kill them. - killChildren :: TVar [Child] -> IO () - killChildren children = do - kids <- readTVarIO children - forM_ kids $ \child -> do - -- If the child is in a read(), interrupt it - -- by calling 'fdShutdown'. - mb <- tryTakeMVar (childFdLock child) - case mb of - Just cfd -> do - void $ MC.try @_ @IOException $ fdShutdown cfd - putMVar (childFdLock child) cfd - Nothing -> return () -- serve thread already closing - -- No children blocked on read: terminate them all. - -- childThread was filled by the parent before mask exit; readMVar of - -- a definitely-full MVar is non-interruptible. - forM_ kids $ \child -> do - tid <- readMVar (childThread child) - killThread tid - --- | Per-connection server loop. -serve :: (forall a. IO a -> IO a) - -> TVar Int -> TVar [Child] -> Fd -> Child - -> IO () -serve restore pool children fd (Child _ fdLock) = do - myCount <- newTVarIO (0 :: Int) - let loop = forever $ MC.mask $ \restoreInner -> do - -- fdReadByte is a safe-FFI read(2), interrupted by - -- fdShutdown from killChildren, not by throwTo. - msg <- fdReadByte fd - case msg of - CmdWait -> do - -- Block until a token is available. - -- restoreInner keeps retry interruptible under mask. - restoreInner $ atomically $ do - n <- readTVar pool - when (n <= 0) retry - writeTVar pool (n - 1) - modifyTVar' myCount (+ 1) - fdWriteByte fd RspOk - - CmdRelease -> do - ok <- atomically $ do - mc <- readTVar myCount - if mc > 0 - then do - modifyTVar' pool (+ 1) - modifyTVar' myCount (subtract 1) - return True - else return False - fdWriteByte fd (if ok then RspOk else RspFail) - - CmdTryWait -> do - acquired <- atomically $ do - n <- readTVar pool - if n > 0 - then do - writeTVar pool (n - 1) - modifyTVar' myCount (+ 1) - return True - else return False - if acquired - then fdWriteByte fd RspOk - else fdWriteByte fd RspFail - - -- Unknown command: reply so the client doesn't hang in read(). - _ -> fdWriteByte fd RspFail - - cleanup = do - -- Return tokens to the pool and remove from children list. - atomically $ do - n <- readTVar myCount - when (n > 0) $ modifyTVar' pool (+ n) - modifyTVar' children (filter (\c -> childFdLock c /= fdLock)) - -- Take fd ownership and close. - -- prevents killChildren from double closing fd - void $ takeMVar fdLock - void $ MC.try @_ @IOException $ closeFd fd - - -- restore so thread can be killed in between loop iterations - -- Catch IOException (EOF/disconnect) silently. - (restore loop `MC.catch` \(_ :: IOException) -> return ()) - `MC.finally` cleanup diff --git a/src/System/Semaphore/Internal/Posix/Server.hs b/src/System/Semaphore/Internal/Posix/Server.hs new file mode 100644 --- /dev/null +++ b/src/System/Semaphore/Internal/Posix/Server.hs @@ -0,0 +1,214 @@ +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeApplications #-} + +-- | The Unix domain socket server backing the POSIX semaphore implementation. +-- +-- The server manages a shared token pool and accepts multiple client +-- connections, each served on its own thread. +-- +-- Protocol (SOCK_STREAM, one byte per command): +-- +-- "-" Wait (blocking acquire). Decrements semaphore; replies ".". +-- "?" Try-wait. Decrements if positive and replies "."; otherwise replies "!". +-- "+" Release. Increments pool; replies ".". Rejected with "!" if +-- myCount <= 0 (client has not acquired any tokens on this connection). +-- +-- Unrecognised bytes are rejected with 'RspFail'. +-- +-- Per-connection token tracking: the server counts tokens held by each +-- connection (myCount). On disconnect (EOF / ResourceVanished) held +-- tokens are returned to the pool, so a crashing client cannot leak +-- tokens. +-- +-- Connections are tied to a token, so each token (obtained by +-- @waitOnSemaphore@) has its own connection. +module System.Semaphore.Internal.Posix.Server + ( serverLoop + , pattern CmdWait, pattern CmdTryWait, pattern CmdRelease + , pattern RspOk, pattern RspFail + ) where + +-- base +import Control.Concurrent + ( ThreadId, forkIOWithUnmask, killThread ) +import Control.Concurrent.MVar + ( MVar, newEmptyMVar, newMVar, putMVar + , readMVar, takeMVar, tryTakeMVar ) +import Control.Exception + ( IOException, SomeException + , catch, finally, mask, mask_ + , onException, throw, try, uninterruptibleMask_ + ) +import Control.Monad + ( forM_, forever, void, when ) +import Data.Word ( Word8 ) +import Foreign.C.Error ( Errno(Errno), eCONNABORTED ) +import GHC.IO.Exception ( ioe_errno ) +import System.IO.Error ( isFullError ) + +-- stm +import Control.Concurrent.STM + ( TVar, atomically, newTVarIO, readTVar, readTVarIO + , modifyTVar', writeTVar, retry ) + +-- unix +import System.Posix.IO ( closeFd ) +import System.Posix.Types ( Fd ) + +import System.Semaphore.Internal.DomainSocket + ( pollAcceptSocket, AcceptResult(..) + , fdReadByte, fdWriteByte + , fdShutdown ) + +--------------------------------------- +-- Protocol byte constants + +pattern CmdWait, CmdTryWait, CmdRelease :: Word8 +pattern CmdWait = 0x2D -- '-' +pattern CmdTryWait = 0x3F -- '?' +pattern CmdRelease = 0x2B -- '+' + +pattern RspOk, RspFail :: Word8 +pattern RspOk = 0x2E -- '.' +pattern RspFail = 0x21 -- '!' + +-- | Children of the server, to be cleaned up +-- childFdLock is full when the 'Fd' is still known to be valid, +-- and subject to being shutdown or closed +data Child = Child + { childThread :: !(MVar ThreadId) + , childFdLock :: !(MVar Fd) + } + +-- | Run the server accept loop on a listening socket until cancellation is +-- signalled via the cancel pipe. All accepted connections are served on +-- child threads, which are cleaned up before this returns. +serverLoop :: TVar Int -- ^ shared token pool + -> Fd -- ^ listening socket + -> Fd -- ^ read end of the cancel pipe + -> IO () +serverLoop pool listenFd cancelFd = do + children <- newTVarIO ([] :: [Child]) + loop children `finally` killChildren children + where + loop children = do + continueLoop <- mask_ $ do + r <- acceptWithRetry + case r of + AcceptCancelled -> return False + AcceptedFd clientFd -> do + forkServeChild children clientFd + return True + when continueLoop $ loop children + + -- pollAcceptSocket only returns if the accept succeeded, or cancellation + -- was signalled via the cancel pipe. + acceptWithRetry :: IO AcceptResult + acceptWithRetry = pollAcceptSocket listenFd cancelFd `catch` handleIOError + + -- Retry accept on transient errors. + handleIOError :: IOException -> IO AcceptResult + handleIOError e + -- 'isFullError' catches ResourceExhausted (EMFILE/ENFILE/ENOBUFS/ENOMEM and more that accept doesn't produce but are harmless to retry). + | isFullError e = acceptWithRetry + -- ECONNABORTED is also transient but categorised as OtherError, we additionaly match that. + | Just err <- ioe_errno e + , Errno err == eCONNABORTED = acceptWithRetry + -- EINTR is absorbed in hs_poll_accept. + -- everything else (EBADF, EINVAL, ENOTSOCK, ...) is rethrown. + | otherwise = throw e + + forkServeChild children clientFd = do + fdLock <- newMVar clientFd + tidVar <- newEmptyMVar + let child = Child tidVar fdLock + atomically $ modifyTVar' children (child :) + childTid <- forkIOWithUnmask $ \unmask -> + serve unmask pool children clientFd child + putMVar tidVar childTid + + -- Interrupt all children blocked on a read from the FD, then kill them. + killChildren :: TVar [Child] -> IO () + killChildren children = do + kids <- readTVarIO children + forM_ kids $ \child -> do + -- If the child is in a read(), interrupt it + -- by calling 'fdShutdown'. + mb <- tryTakeMVar (childFdLock child) + case mb of + Just cfd -> do + void $ try @IOException $ fdShutdown cfd + putMVar (childFdLock child) cfd + Nothing -> return () -- serve thread already closing + -- No children blocked on read: terminate them all. + -- childThread was filled by the parent before mask exit; readMVar of + -- a definitely-full MVar is non-interruptible. + forM_ kids $ \child -> do + tid <- readMVar (childThread child) + killThread tid + +-- | Per-connection server loop. +serve :: (forall a. IO a -> IO a) + -> TVar Int -> TVar [Child] -> Fd -> Child + -> IO () +serve restore pool children fd (Child _ fdLock) = do + myCount <- newTVarIO (0 :: Int) + let loop = forever $ mask $ \restoreInner -> do + -- fdReadByte is a safe-FFI read(2), interrupted by + -- fdShutdown from killChildren, not by throwTo. + msg <- fdReadByte fd + case msg of + CmdWait -> do + -- Block until a token is available. + -- restoreInner keeps retry interruptible under mask. + restoreInner $ atomically $ do + n <- readTVar pool + when (n <= 0) retry + writeTVar pool (n - 1) + modifyTVar' myCount (+ 1) + fdWriteByte fd RspOk + + CmdRelease -> do + ok <- atomically $ do + mc <- readTVar myCount + if mc > 0 + then do + modifyTVar' pool (+ 1) + modifyTVar' myCount (subtract 1) + return True + else return False + fdWriteByte fd (if ok then RspOk else RspFail) + + CmdTryWait -> do + acquired <- atomically $ do + n <- readTVar pool + if n > 0 + then do + writeTVar pool (n - 1) + modifyTVar' myCount (+ 1) + return True + else return False + if acquired + then fdWriteByte fd RspOk + else fdWriteByte fd RspFail + + -- Unknown command: reply so the client doesn't hang in read(). + _ -> fdWriteByte fd RspFail + + cleanup = do + -- Return tokens to the pool and remove from children list. + atomically $ do + n <- readTVar myCount + when (n > 0) $ modifyTVar' pool (+ n) + modifyTVar' children (filter (\c -> childFdLock c /= fdLock)) + -- Take fd ownership and close. + -- prevents killChildren from double closing fd + void $ takeMVar fdLock + void $ try @IOException $ closeFd fd + + -- restore so thread can be killed in between loop iterations + -- Catch IOException (EOF/disconnect) silently. + (restore loop `catch` \(_ :: IOException) -> return ()) + `finally` cleanup diff --git a/src/System/Semaphore/Internal/Unsupported.hs b/src/System/Semaphore/Internal/Unsupported.hs --- a/src/System/Semaphore/Internal/Unsupported.hs +++ b/src/System/Semaphore/Internal/Unsupported.hs @@ -5,22 +5,27 @@ -- | Stub backend for wasm and targets without POSIX/Win32 sockets/semaphores. module System.Semaphore.Internal.Unsupported ( ClientSemaphore, ServerSemaphore - , clientSemaphoreName, serverClientSemaphore , SemaphoreToken - , create_sem, open_sem_raw + , clientSemaphoreName, serverClientSemaphore , waitOnSemaphore, tryWaitOnSemaphore , releaseSemaphoreToken , destroyClientSemaphore, destroyServerSemaphore , getSemaphoreValue - , getTimeSeed + + , create_sem, open_sem_raw + , get_time_seed ) where +-- base import Control.Exception ( IOException ) import GHC.IO.Exception ( unsupportedOperation ) import System.IO.Error ( ioeSetLocation ) -import System.Semaphore.Internal.Common +-- semaphore-compat +import System.Semaphore.Internal.Version +-------------------------------------------------------------------------------- + data ClientSemaphore data ServerSemaphore @@ -62,6 +67,5 @@ getSemaphoreValue :: ServerSemaphore -> IO Int getSemaphoreValue = \case {} -getTimeSeed :: IO Int -getTimeSeed = - ioError $ unsupported "getTimeSeed" +get_time_seed :: IO Int +get_time_seed = ioError $ unsupported "get_time_seed" diff --git a/src/System/Semaphore/Internal/Version.hs b/src/System/Semaphore/Internal/Version.hs new file mode 100644 --- /dev/null +++ b/src/System/Semaphore/Internal/Version.hs @@ -0,0 +1,153 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE MagicHash #-} + +-- | Semaphore protocol version and versioned semaphore identifiers. +module System.Semaphore.Internal.Version + ( SemaphoreName(..) + , SemaphoreIdentifier + , SemaphoreProtocolVersion(..) + , SemaphoreError(..) + , semaphoreVersion + , semaphoreIdentifier + , parseSemaphoreIdentifier + , getSemaphoreSocketPath + , iToBase62 + ) where + +-- base +import Control.Exception ( Exception, IOException ) +import GHC.Exts ( Char(..), Int(..), indexCharOffAddr# ) + +-- directory +import System.Directory ( getTemporaryDirectory, createDirectoryIfMissing ) + +-- filepath +import System.FilePath ( () ) + +--------------------------------------- +-- Types + +-- | The semaphore protocol version currently being used. +-- +-- The version tracks the IPC mechanism, not the library version: +-- +-- * __POSIX:__ 2 (domain sockets, replacing v1 system semaphores). +-- * __Windows:__ 1 (Win32 named semaphores, unchanged from v1). +-- * __Unsupported platforms:__ 0 (no compatible IPC backend). +newtype SemaphoreProtocolVersion = + SemaphoreProtocolVersion { getSemaphoreProtocolVersion :: Int } + deriving (Eq, Ord, Show) + +-- | A semaphore name: a protocol version and an unversioned name string. +data SemaphoreName = SemaphoreName + { semaphoreProtocolVersion :: !SemaphoreProtocolVersion + , unversionedSemaphoreNameString :: !String + } deriving (Eq, Show) + +-- | The identifier string of a semaphore: a serialised 'SemaphoreName' used +-- for inter-process communication (e.g. as a command line argument). +-- +-- The specific format of this name string depends on the __protocol__ version: +-- +-- - For version 1, this is the unadorned semaphore name. +-- - For version @N >= 2@, the semaphore name is prefixed with @v\-@. +type SemaphoreIdentifier = String + +-- | Errors that can occur when creating or opening a semaphore. +data SemaphoreError + -- | Can't create a semaphore: a semaphore with this name already exists. + = SemaphoreAlreadyExists + { semaphoreErrorIdentifier :: !SemaphoreIdentifier + } + -- | Can't open a semaphore: no semaphore with this name exists. + | SemaphoreDoesNotExist + { semaphoreErrorIdentifier :: !SemaphoreIdentifier + } + -- | Protocol version incompatibility: the semaphore identifier uses a + -- different protocol version than the library is currently using. + | SemaphoreIncompatibleVersion + { semaphoreErrorActualVersion :: !SemaphoreProtocolVersion + , semaphoreErrorExpectedVersion :: !SemaphoreProtocolVersion + } + -- | An 'IOException' was raised when creating or opening the semaphore. + | SemaphoreOtherError + { semaphoreOtherError :: !IOException + } + deriving (Show, Eq) + +instance Exception SemaphoreError + +--------------------------------------- +-- Version negotiation + +-- | The protocol version used on this host platform with this version of +-- @semaphore-compat@. +-- +-- See 'SemaphoreProtocolVersion'. +semaphoreVersion :: SemaphoreProtocolVersion +#if defined(mingw32_HOST_OS) +semaphoreVersion = SemaphoreProtocolVersion 1 +#elif defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) +semaphoreVersion = SemaphoreProtocolVersion 0 +#else +semaphoreVersion = SemaphoreProtocolVersion 2 +#endif + + +-- | The serialised identifier of a 'SemaphoreName' for inter-process +-- communication. +-- +-- See 'SemaphoreIdentifier' for more information. +semaphoreIdentifier :: SemaphoreName -> SemaphoreIdentifier +semaphoreIdentifier sn + | ver <= 1 = unversionedSemaphoreNameString sn + | otherwise = "v" ++ show ver ++ "-" ++ unversionedSemaphoreNameString sn + where + ver = getSemaphoreProtocolVersion (semaphoreProtocolVersion sn) + + + +-- TODO: I think 'parseSemaphoreIdentifier' is unused (neither Cabal nor GHC use it). + +-- | Parse a 'SemaphoreIdentifier' into a 'SemaphoreName'. +-- +-- Returns @Nothing@ for unversioned strings (which should be treated as +-- v1 by the caller's compatibility logic). +parseSemaphoreIdentifier :: SemaphoreIdentifier -> Maybe SemaphoreName +parseSemaphoreIdentifier ('v':rest) = + case reads rest of + [(n, '-':name)] -> Just (SemaphoreName (SemaphoreProtocolVersion n) name) + _ -> Nothing +parseSemaphoreIdentifier _ = Nothing + +--------------------------------------- +-- Utilities + +-- | Convert an 'Int' to a base-62 string (digits @0@–@9@, @a@–@z@, @A@–@Z@). +iToBase62 :: Int -> String +iToBase62 m = go m' "" + where + m' + | m == minBound = maxBound + | otherwise = abs m + go n cs | n < 62 + = let !c = chooseChar62 n + in c : cs + | otherwise + = let !(!q, r) = quotRem n 62 + !c = chooseChar62 r + in go q (c : cs) + + chooseChar62 :: Int -> Char + {-# INLINE chooseChar62 #-} + chooseChar62 (I# n) = C# (indexCharOffAddr# chars62 n) + chars62 = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"# + +-- | The socket file path for a semaphore. +getSemaphoreSocketPath :: SemaphoreName -> IO FilePath +getSemaphoreSocketPath sn = do + tmp <- getTemporaryDirectory + let dir = tmp "semaphore-compat" + createDirectoryIfMissing True dir + return (dir unversionedSemaphoreNameString sn) diff --git a/src/System/Semaphore/Internal/Win32.hs b/src/System/Semaphore/Internal/Win32.hs --- a/src/System/Semaphore/Internal/Win32.hs +++ b/src/System/Semaphore/Internal/Win32.hs @@ -1,26 +1,31 @@ {-# LANGUAGE TypeApplications #-} +-- | Windows backend: semaphores implemented on top of Win32 named semaphores. module System.Semaphore.Internal.Win32 ( ClientSemaphore(..), ServerSemaphore(..) , SemaphoreToken(..) - , create_sem, open_sem_raw , waitOnSemaphore, tryWaitOnSemaphore , releaseSemaphoreToken , destroyClientSemaphore, destroyServerSemaphore , getSemaphoreValue - , getTimeSeed + + , create_sem, open_sem_raw + , get_time_seed ) where +-- base import Control.Concurrent ( forkIO, killThread ) import Control.Concurrent.MVar ( MVar, newEmptyMVar, newMVar, putMVar, takeMVar, tryReadMVar, tryTakeMVar ) -import Control.Exception ( IOException, uninterruptibleMask_ ) +import Control.Exception + ( IOException, SomeException + , finally, mask_, onException, throw, try, uninterruptibleMask_ + ) import Control.Monad import System.IO.Error ( isDoesNotExistError ) -import qualified Control.Monad.Catch as MC - +-- win32 import qualified System.Win32.Event as Win32 ( createEvent, setEvent , waitForSingleObject, waitForMultipleObjects @@ -35,33 +40,31 @@ import qualified System.Win32.Time as Win32 ( FILETIME(..), getSystemTimeAsFileTime ) -import System.Semaphore.Internal.Common +-- semaphore-compat +import System.Semaphore.Internal.Version --- | A client-side semaphore (semaphore for a jobclient). +-------------------------------------------------------------------------------- + data ClientSemaphore = ClientSemaphore { clientSemaphoreName :: !SemaphoreName , semaphore :: !(MVar Win32.Semaphore) - -- ^ MVar is emptied when the handle is closed to prevent double close. + -- NB: the MVar is emptied when the handle is closed to prevent double close. } --- | A held semaphore token. newtype SemaphoreToken = SemaphoreToken - { tokenSemaphore :: Win32.Semaphore + { tokenSemaphore :: MVar Win32.Semaphore + -- NB: the MVar is emptied when the token is released to prevent double release. } --- | A server-side semaphore: a 'ClientSemaphore' with additional --- book-keeping to keep track of resource usage by all clients. --- --- Use 'serverClientSemaphore' to retrieve the client-side semaphore. newtype ServerSemaphore = ServerSemaphore { serverClientSemaphore :: ClientSemaphore } create_sem :: SemaphoreName -> Int -> IO (Either SemaphoreError ServerSemaphore) -create_sem sem_nm init_toks = MC.mask_ $ do +create_sem sem_nm init_toks = mask_ $ do let toks = fromIntegral init_toks - mb_sem <- MC.try @_ @IOException $ + mb_sem <- try @IOException $ Win32.createSemaphore Nothing toks 0x7FFFFFFF (Just (semaphoreIdentifier sem_nm)) case mb_sem of Left err -> @@ -69,7 +72,7 @@ Right (sem, True) -> do -- Semaphore already existed; createSemaphore still returned a valid -- handle to it, which we must close to avoid leaking it. - void $ MC.try @_ @IOException $ Win32.closeHandle (Win32.semaphoreHandle sem) + void $ try @IOException $ Win32.closeHandle (Win32.semaphoreHandle sem) return $ Left $ SemaphoreAlreadyExists (semaphoreIdentifier sem_nm) Right (sem, False) -> do semVar <- newMVar sem @@ -78,12 +81,13 @@ { serverClientSemaphore = ClientSemaphore { clientSemaphoreName = sem_nm - , semaphore = semVar } + , semaphore = semVar + } } open_sem_raw :: SemaphoreName -> IO (Either SemaphoreError ClientSemaphore) -open_sem_raw nm = MC.mask_ $ do - mb_res <- MC.try @_ @IOException $ +open_sem_raw nm = mask_ $ do + mb_res <- try @IOException $ Win32.openSemaphore Win32.sEMAPHORE_ALL_ACCESS True (semaphoreIdentifier nm) case mb_res of Left err @@ -96,9 +100,10 @@ return $ Right $ ClientSemaphore { clientSemaphoreName = nm - , semaphore = semVar } + , semaphore = semVar + } --- | Read the kernel handle from a 'ClientSemaphore'. Throws if the +-- Read the kernel handle from a 'ClientSemaphore'. Throws if the -- client has been destroyed. readClientHandle :: ClientSemaphore -> String -> IO Win32.Semaphore readClientHandle client opName = do @@ -108,20 +113,14 @@ Nothing -> ioError $ userError $ "semaphore-compat." ++ opName ++ ": semaphore has been destroyed" --- | Acquire a token from the semaphore, blocking until one is available. --- --- This operation is interruptible. If interrupted after the token was acquired, --- the token is automatically released back to the pool. --- --- Returns a 'SemaphoreToken' that must be released with 'releaseSemaphoreToken'. waitOnSemaphore :: ClientSemaphore -> IO SemaphoreToken waitOnSemaphore client = do resultVar <- newEmptyMVar - MC.mask_ $ do + mask_ $ do sem <- readClientHandle client "waitOnSemaphore" cancelEvent <- Win32.createEvent Nothing True False "" - workerTid <- (forkIO $ MC.mask_ $ do - res <- MC.try @_ @MC.SomeException $ do + workerTid <- (forkIO $ mask_ $ do + res <- try @SomeException $ do wait_res <- Win32.waitForMultipleObjects [ Win32.semaphoreHandle sem @@ -129,13 +128,13 @@ False -- WaitAny Win32.iNFINITE if wait_res == Win32.wAIT_OBJECT_0 - then return (SemaphoreToken sem) + then SemaphoreToken <$> newMVar sem else ioError (userError "semaphore wait cancelled") putMVar resultVar res - ) `MC.onException` Win32.closeHandle cancelEvent + ) `onException` Win32.closeHandle cancelEvent let cleanup = -- Signal the cancel event and wait for the worker to exit. - (do ok <- MC.try @_ @IOException $ Win32.setEvent cancelEvent + (do ok <- try @IOException $ Win32.setEvent cancelEvent case ok of Right () -> -- Cancel event signaled. Worker's WaitForMultipleObjects @@ -146,43 +145,39 @@ -- The worker is in a safe ffi call so killThread -- will just block forever. return () - ) `MC.finally` (do + ) `finally` (do -- Recover any transiently acquired token. If the worker -- acquired the semaphore before seeing the cancel event, -- release it back to the pool. mbRes <- tryTakeMVar resultVar case mbRes of Just (Right tok) -> - void $ MC.try @_ @MC.SomeException $ releaseSemaphoreToken tok + void $ try @SomeException $ releaseSemaphoreToken tok _ -> return () Win32.closeHandle cancelEvent) - res <- takeMVar resultVar `MC.onException` cleanup + res <- takeMVar resultVar `onException` cleanup Win32.closeHandle cancelEvent case res of Right tok -> return tok - Left e -> MC.throwM e + Left e -> throw e --- | Try to acquire a token from the semaphore without blocking. --- --- Returns @Just token@ if a token was available, @Nothing@ otherwise. tryWaitOnSemaphore :: ClientSemaphore -> IO (Maybe SemaphoreToken) tryWaitOnSemaphore client = - MC.mask_ $ do + mask_ $ do sem <- readClientHandle client "tryWaitOnSemaphore" wait_res <- Win32.waitForSingleObject (Win32.semaphoreHandle sem) 0 if wait_res == Win32.wAIT_OBJECT_0 - then return (Just (SemaphoreToken sem)) + then Just . SemaphoreToken <$> newMVar sem else return Nothing --- | Release a semaphore token, returning it to the pool. releaseSemaphoreToken :: SemaphoreToken -> IO () -releaseSemaphoreToken (SemaphoreToken sem) = - MC.mask_ $ do - void $ Win32.releaseSemaphore sem 1 +releaseSemaphoreToken (SemaphoreToken v) = + mask_ $ do + mb <- tryTakeMVar v + case mb of + Nothing -> return () + Just sem -> void $ Win32.releaseSemaphore sem 1 --- | Destroy a client-side semaphore. --- --- Idempotent: subsequent calls after the first are no-ops. destroyClientSemaphore :: ClientSemaphore -> IO () destroyClientSemaphore client = uninterruptibleMask_ $ do mb <- tryTakeMVar (semaphore client) @@ -190,20 +185,13 @@ Nothing -> return () -- already destroyed Just sem -> Win32.closeHandle (Win32.semaphoreHandle sem) --- | Destroy a server-side semaphore. --- --- Idempotent: subsequent calls after the first are no-ops. destroyServerSemaphore :: ServerSemaphore -> IO () destroyServerSemaphore server = destroyClientSemaphore (serverClientSemaphore server) --- | Query the current semaphore value (how many tokens it has available). --- --- This is mainly for debugging use, as it is easy to introduce race conditions --- when nontrivial program logic depends on the value returned by this function. getSemaphoreValue :: ServerSemaphore -> IO Int getSemaphoreValue server = - MC.mask_ $ do + mask_ $ do sem <- readClientHandle (serverClientSemaphore server) "getSemaphoreValue" wait_res <- Win32.waitForSingleObject (Win32.semaphoreHandle sem) 0 if wait_res == Win32.wAIT_OBJECT_0 @@ -212,7 +200,7 @@ else return 0 -getTimeSeed :: IO Int -getTimeSeed = do +get_time_seed :: IO Int +get_time_seed = do Win32.FILETIME t <- Win32.getSystemTimeAsFileTime return (fromIntegral t) diff --git a/test-common/System/Semaphore/TestHelper.hs b/test-common/System/Semaphore/TestHelper.hs --- a/test-common/System/Semaphore/TestHelper.hs +++ b/test-common/System/Semaphore/TestHelper.hs @@ -210,19 +210,20 @@ data HelperHandle = HelperHandle { hStdin :: Handle , hStdout :: Handle + , hStderr :: Handle , hProcess :: ProcessHandle } startHelper :: IO HelperHandle startHelper = do - (Just hin, Just hout, _, ph) <- createProcess (proc "semaphore-helper" []) + (Just hin, Just hout, Just herr, ph) <- createProcess (proc "semaphore-helper" []) { std_in = CreatePipe , std_out = CreatePipe , std_err = CreatePipe } hSetBuffering hin LineBuffering hSetBuffering hout LineBuffering - return (HelperHandle hin hout ph) + return (HelperHandle hin hout herr ph) cleanupHelper :: HelperHandle -> IO () cleanupHelper h = do @@ -233,6 +234,7 @@ return () _ <- try (hClose (hStdin h)) :: IO (Either SomeException ()) _ <- try (hClose (hStdout h)) :: IO (Either SomeException ()) + _ <- try (hClose (hStderr h)) :: IO (Either SomeException ()) return () withHelper :: (HelperHandle -> IO a) -> IO a diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -25,17 +25,16 @@ import Test.Tasty.HUnit import System.Semaphore -import qualified System.Semaphore.Internal.Common as Internal +import qualified System.Semaphore.Internal.Version as Internal import System.Semaphore.TestHelper +import System.Semaphore.Internal ( SemaphoreToken(..) ) #if !defined(mingw32_HOST_OS) import Control.Concurrent.MVar ( tryTakeMVar ) -import System.Semaphore.Internal.Posix ( SemaphoreToken(tokenFdLock) ) +import System.Semaphore.Internal.Posix ( tokenFdLock ) import System.Semaphore.Internal.DomainSocket ( connectDomainSocket, fdReadByte, fdWriteByte ) import System.Posix.IO ( closeFd ) -#else -import System.Semaphore.Internal.Win32 ( SemaphoreToken(..) ) #endif -- | Simulate a crash: close the token's underlying connection without @@ -43,7 +42,7 @@ -- and return the token to the pool. crashToken :: SemaphoreToken -> IO () #if !defined(mingw32_HOST_OS) -crashToken tok = do +crashToken (SemaphoreToken tok) = do mb <- tryTakeMVar (tokenFdLock tok) case mb of Just fd -> closeFd fd