semaphore-compat 2.0.0 → 2.0.1

raw patch · 14 files changed

+775/−540 lines, 14 filesdep −exceptionsPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies removed: exceptions

API changes (from Hackage documentation)

Files

• cbits/domain.c +42/−4
• changelog.md +10/−0
• semaphore-compat.cabal +4/−4
• src/System/Semaphore.hs +80/−50
• src/System/Semaphore/Internal.hs +136/−0
• src/System/Semaphore/Internal/Common.hs +0/−142
• src/System/Semaphore/Internal/DomainSocket.hs +3/−0
• src/System/Semaphore/Internal/Posix.hs +66/−266
• src/System/Semaphore/Internal/Posix/Server.hs +214/−0
• src/System/Semaphore/Internal/Unsupported.hs +11/−7
• src/System/Semaphore/Internal/Version.hs +153/−0
• src/System/Semaphore/Internal/Win32.hs +48/−60
• test-common/System/Semaphore/TestHelper.hs +4/−2
• test/Main.hs +4/−5

@@ -7,17 +7,29 @@ #include <sys/un.h> #include <unistd.h> +/* 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)

@@ -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:

@@ -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

@@ -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\<N\>-@.--- 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)

@@ -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

@@ -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\<N\>-\<name\>\"@.-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\<N\>-@ 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\<N\>-@.-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)

@@ -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

@@ -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

@@ -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

@@ -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"

@@ -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\<N\>-@.+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)

@@ -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)

@@ -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

@@ -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