semaphore-compat 1.0.0 → 2.0.0
raw patch · 14 files changed
+3051/−438 lines, 14 filesdep +containersdep +directorydep +filepathdep ~basesetup-changednew-component:exe:semaphore-helpernew-uploader
Dependencies added: containers, directory, filepath, process, semaphore-compat, stm, tasty, tasty-expected-failure, tasty-hunit
Dependency ranges changed: base
Files
- Setup.hs +6/−6
- cbits/domain.c +161/−0
- changelog.md +39/−3
- helper/Main.hs +6/−0
- readme.md +28/−16
- semaphore-compat.cabal +157/−62
- src/System/Semaphore.hs +187/−351
- src/System/Semaphore/Internal/Common.hs +142/−0
- src/System/Semaphore/Internal/DomainSocket.hs +101/−0
- src/System/Semaphore/Internal/Posix.hs +447/−0
- src/System/Semaphore/Internal/Unsupported.hs +67/−0
- src/System/Semaphore/Internal/Win32.hs +218/−0
- test-common/System/Semaphore/TestHelper.hs +277/−0
- test/Main.hs +1215/−0
Setup.hs view
@@ -1,6 +1,6 @@-module Main (main) where - -import Distribution.Simple - -main :: IO () -main = defaultMain +module Main (main) where++import Distribution.Simple++main :: IO ()+main = defaultMain
+ cbits/domain.c view
@@ -0,0 +1,161 @@+#include <errno.h>+#include <fcntl.h>+#include <poll.h>+#include <string.h>+#include <sys/socket.h>+#include <sys/types.h>+#include <sys/un.h>+#include <unistd.h>++int hs_listen_domain_socket(const char *path) {+ if (strlen(path) >= sizeof(((struct sockaddr_un *)0)->sun_path)) {+ errno = ENAMETOOLONG;+ return -1;+ }++#ifdef SOCK_NONBLOCK+ int fd = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0);+#else+ int fd = socket(AF_UNIX, SOCK_STREAM, 0);+#endif+ if (fd == -1) {+ return -1;+ }+#ifndef SOCK_NONBLOCK+ int flags = fcntl(fd, F_GETFL, 0);+ if (flags == -1 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -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));+ server_addr.sun_family = AF_UNIX;+ strncpy(server_addr.sun_path, path, sizeof(server_addr.sun_path) - 1);++ if (bind(fd, (struct sockaddr *) &server_addr, sizeof(struct sockaddr_un)) == -1) {+ int saved_errno = errno;+ close(fd);+ errno = saved_errno;+ return -1;+ }++ if (listen(fd, SOMAXCONN) == -1) {+ int saved_errno = errno;+ close(fd);+ /* bind(2) succeeded so the socket file exists on disk; unlink it. */+ unlink(path);+ errno = saved_errno;+ return -1;+ }++ return fd;+}++int hs_connect_domain_socket(const char *path) {+ if (strlen(path) >= sizeof(((struct sockaddr_un *)0)->sun_path)) {+ errno = ENAMETOOLONG;+ return -1;+ }++ int fd = socket(AF_UNIX, SOCK_STREAM, 0);+ if (fd == -1) {+ return -1;+ }++ struct sockaddr_un server_addr;+ memset(&server_addr, 0, sizeof(struct sockaddr_un));+ server_addr.sun_family = AF_UNIX;+ strncpy(server_addr.sun_path, path, sizeof(server_addr.sun_path) - 1);++ if (connect(fd, (struct sockaddr *) &server_addr, sizeof(struct sockaddr_un)) == -1) {+ if (errno != EINTR) {+ int saved_errno = errno;+ close(fd);+ errno = saved_errno;+ return -1;+ }+ /* See http://www.madore.org/~david/computers/connect-intr.html */+ struct pollfd unix_really_sucks;+ int some_more_junk;+ socklen_t yet_more_useless_junk;+ unix_really_sucks.fd = fd;+ unix_really_sucks.events = POLLOUT;+ while (poll(&unix_really_sucks, 1, -1) == -1) {+ if (errno != EINTR) {+ int saved_errno = errno;+ close(fd);+ errno = saved_errno;+ return -1;+ }+ }+ yet_more_useless_junk = sizeof(some_more_junk);+ if (getsockopt(fd, SOL_SOCKET, SO_ERROR,+ &some_more_junk, &yet_more_useless_junk) == -1) {+ int saved_errno = errno;+ close(fd);+ errno = saved_errno;+ return -1;+ }+ if (some_more_junk != 0) {+ close(fd);+ errno = some_more_junk;+ return -1;+ }+ }+ /* At this point, fd is connected. */++ return fd;+}++/*+ * Cooperative interruptible accept.+ *+ * Blocks in poll(2) until either a connection arrives on listen_fd or+ * cancel_fd becomes readable. Spins until cancel_fd is written to,+ * or reading from listen_fd fails for some reason. + *+ * This avoids interruptible FFI (which is subject to GHC #27110 and #27113).+ *+ * Returns:+ * >= 0 accepted fd+ * -1 error (with errno)+ * -2 cancelled (cancel_fd was signalled or closed)+ *+ */+int hs_poll_accept(int listen_fd, int cancel_fd) {+ struct pollfd fds[2];+ fds[0].fd = listen_fd;+ fds[0].events = POLLIN;+ fds[1].fd = cancel_fd;+ fds[1].events = POLLIN;++ for (;;) {+ int r = poll(fds, 2, -1);+ if (r == -1) {+ if (errno == EINTR) continue;+ return -1;+ }+ /* Check cancel fd first */+ if (fds[1].revents & (POLLIN | POLLHUP | POLLERR))+ return -2;+ /* Listen fd error (e.g. closed/invalid): report as error, don't spin */+ if (fds[0].revents & (POLLERR | POLLHUP | POLLNVAL)) {+ errno = EBADF;+ return -1;+ }+ if (fds[0].revents & POLLIN) {+ int afd = accept(listen_fd, NULL, NULL);+ if (afd >= 0) 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)+ continue;+ return -1;+ }+ /* spurious wakeup — retry */+ }+}
changelog.md view
@@ -1,3 +1,39 @@-### 1.0.0 (March 13, 2023) - -- First version of the `semaphore-compat` package. +### 2.0.0 (May 22, 2026)++- Major release introducing protocol v2:+ - On Linux and other POSIX platforms, the implementation now uses+ Unix domain sockets in place of POSIX named semaphores+ (`sem_open(3)`), avoiding the libc-ABI issues that affected the+ v1 implementation when a jobserver and its clients were built+ against different C standard libraries (see+ [GHC #25087](https://gitlab.haskell.org/ghc/ghc/-/issues/25087)+ and [cabal #9993](https://github.com/haskell/cabal/issues/9993)).+ - Windows is unaffected and continues to use Win32 named+ semaphores; its reported protocol version remains v1.+- API overhaul:+ - `Semaphore` has been renamed to `ClientSemaphore`.+ - `ServerSemaphore` is a new type representing the server-side+ identity of a semaphore, with full bookkeeping for crash+ recovery on POSIX. Use `serverClientSemaphore` to recover the+ `ClientSemaphore` from a `ServerSemaphore`.+ - `SemaphoreToken` is a new type representing a single held token;+ `waitOnSemaphore` returns one and `releaseSemaphoreToken` (or+ `withSemaphoreToken`) releases one.+ - `SemaphoreName` is now a record carrying the protocol version+ alongside the unversioned name; use `semaphoreIdentifier` to derive+ the serialised name for transport between processes via flags like+ `-jsem`.+- `versionsAreCompatible :: SemaphoreProtocolVersion ->+ SemaphoreProtocolVersion -> Bool` exposed for use by jobservers+ (e.g. cabal-install) to detect protocol mismatches before invoking a+ jobclient (e.g. GHC).+- `waitOnSemaphore` is now interruptible: an asynchronous exception+ delivered while a thread is blocked on the semaphore aborts the wait+ safely.+- POSIX server-side crash recovery: when a client disconnects without+ explicitly returning its tokens, the server returns them to the pool+ automatically. This is not available on Windows.++### 1.0.0 (March 13, 2023)++- First version of the `semaphore-compat` package.
+ helper/Main.hs view
@@ -0,0 +1,6 @@+module Main where++import System.Semaphore.TestHelper (helperMain)++main :: IO ()+main = helperMain
readme.md view
@@ -1,16 +1,28 @@-# semaphore-compat - -`semaphore-compat` provides a cross-platform implementation of system semaphores -that abstracts over the `unix` and `Win32` libraries. - -It supports: - - - Creating (`createSemaphore`, `freshSemaphore`), opening (`openSemaphore`) - and closing (`destroySemaphore`) semaphores. - - Waiting on a semaphore: - - without blocking with `tryWaitOnSemaphore`, - - blocking forever, with `waitOnSemaphore`, - - blocking, in a separate thread and allowing interruption, with - `forkWaitOnSemaphoreInterruptible` and `interruptWaitOnSemaphore`. - - Releasing tokens to a semaphore (`releaseSemaphore`). - - Querying the semaphore for its current value (`getSemaphoreValue`). +# semaphore-compat++`semaphore-compat` provides cross-platform semaphores that can be shared across processes.++It uses different backends on different platforms:++ - POSIX: Unix domain sockets with automatic return of tokens from+ disconnected clients.+ - Windows: Win32 named semaphores. Tokens held by crashed clients are not+ returned automatically.+ - wasm/JavaScript: buildable stubs that throw unsupported-operation errors.+ Avoid calling this code on these backends.++It supports:++ - Creating server semaphores (`createSemaphore`, `freshSemaphore`).+ - A server is the process that creates the semaphore and must stay alive for+ the duration of its use.+ - A server process can also act as a client and request/release tokens from the+ semaphore.+ - Opening client semaphores (`openSemaphore`).+ - Waiting with `tryWaitOnSemaphore` or interruptible `waitOnSemaphore`,+ which return `SemaphoreToken`s.+ - Releasing tokens with `releaseSemaphoreToken` or `withSemaphoreToken`.+ - Destroying semaphores with `destroyClientSemaphore` and+ `destroyServerSemaphore`.+ - Serialising names and checking versions with `semaphoreIdentifier` and+ `versionsAreCompatible`.
semaphore-compat.cabal view
@@ -1,62 +1,157 @@-cabal-version: 3.0 - -name: - semaphore-compat -version: - 1.0.0 -license: - BSD-3-Clause - -author: - The GHC team -maintainer: - ghc-devs@haskell.org -homepage: - https://gitlab.haskell.org/ghc/semaphore-compat -bug-reports: - https://gitlab.haskell.org/ghc/ghc/issues/new - -category: - System -synopsis: - Cross-platform abstraction for system semaphores -description: - This package provides a cross-platform implementation of system semaphores - that abstracts over the `unix` and `Win32` libraries. - -build-type: - Simple - -extra-source-files: - changelog.md - , readme.md - - -source-repository head - type: git - location: https://gitlab.haskell.org/ghc/semaphore-compat.git - -library - hs-source-dirs: - src - - exposed-modules: - System.Semaphore - - build-depends: - base - >= 4.12 && < 4.20 - , exceptions - >= 0.7 && < 0.11 - - if os(windows) - build-depends: - Win32 - >= 2.13.4.0 && < 2.14 - else - build-depends: - unix - >= 2.8.1.0 && < 2.9 - - default-language: - Haskell2010 +cabal-version: 3.0++name:+ semaphore-compat+version:+ 2.0.0+license:+ BSD-3-Clause++author:+ The GHC team+maintainer:+ ghc-devs@haskell.org+homepage:+ https://gitlab.haskell.org/ghc/semaphore-compat+bug-reports:+ https://gitlab.haskell.org/ghc/ghc/issues/new++category:+ System+synopsis:+ Cross-platform abstraction for system semaphores+description:+ Cross-platform semaphores for managing resources across processes,+ abstracting over Win32 named semaphores on Windows and Unix domain sockets+ on POSIX.++build-type:+ Simple++extra-source-files:+ changelog.md+ , readme.md+++flag build-testing+ description: Build test helper library, executable, and test suite+ default: False+ manual: True++source-repository head+ type: git+ location: https://gitlab.haskell.org/ghc/semaphore-compat.git++library+ hs-source-dirs:+ src++ exposed-modules:+ System.Semaphore+ System.Semaphore.Internal.Common++ build-depends:+ base+ >= 4.11 && < 4.24+ , directory+ >= 1.3 && < 1.4+ , exceptions+ >= 0.7 && < 0.11+ , filepath+ >= 1.4 && < 1.6++ if os(windows)+ build-depends:+ Win32+ >= 2.13.4.0 && < 2.15+ exposed-modules:+ System.Semaphore.Internal.Win32+ elif arch(wasm32) || arch(javascript)+ exposed-modules:+ System.Semaphore.Internal.Unsupported+ else+ build-depends:+ unix+ >= 2.8.1.0 && < 2.9+ , stm+ >= 2.4 && < 2.6+ , containers+ >= 0.5 && < 0.9+ c-sources:+ cbits/domain.c+ exposed-modules:+ System.Semaphore.Internal.DomainSocket+ System.Semaphore.Internal.Posix++ default-language:+ Haskell2010++library semaphore-test-common+ visibility:+ private+ hs-source-dirs:+ test-common+ exposed-modules:+ System.Semaphore.TestHelper+ build-depends:+ base+ >= 4.11 && < 4.23+ , semaphore-compat+ , process+ >= 1.6.16 && < 1.8+ if !os(windows)+ build-depends:+ unix+ >= 2.8.1.0 && < 2.9+ if !flag(build-testing)+ buildable: False+ ghc-options:+ -Wall+ default-language:+ Haskell2010++executable semaphore-helper+ hs-source-dirs:+ helper+ main-is:+ Main.hs+ build-depends:+ base+ >= 4.11 && < 4.23+ , semaphore-compat:semaphore-test-common+ if !flag(build-testing)+ buildable: False+ ghc-options:+ -threaded+ default-language:+ Haskell2010++test-suite semaphore-compat-test+ type:+ exitcode-stdio-1.0+ hs-source-dirs:+ test+ main-is:+ Main.hs+ build-depends:+ base+ >= 4.11 && < 4.23+ , semaphore-compat+ , semaphore-compat:semaphore-test-common+ , process+ >= 1.6.16 && < 1.8+ , tasty+ , tasty-expected-failure+ , tasty-hunit+ if !os(windows)+ build-depends:+ unix+ >= 2.8.1.0 && < 2.9+ build-tool-depends:+ semaphore-compat:semaphore-helper+ if !flag(build-testing)+ buildable: False+ ghc-options:+ -threaded -rtsopts "-with-rtsopts=-N"+ default-language:+ Haskell2010
src/System/Semaphore.hs view
@@ -1,351 +1,187 @@-{-# LANGUAGE BangPatterns #-} -{-# LANGUAGE CPP #-} -{-# LANGUAGE MagicHash #-} -{-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE TypeApplications #-} - -module System.Semaphore - ( -- * System semaphores - Semaphore(..), SemaphoreName(..) - , createSemaphore, freshSemaphore, openSemaphore - , waitOnSemaphore, tryWaitOnSemaphore - , WaitId(..) - , forkWaitOnSemaphoreInterruptible - , interruptWaitOnSemaphore - , getSemaphoreValue - , releaseSemaphore - , destroySemaphore - - -- * Abstract semaphores - , AbstractSem(..) - , withAbstractSem - ) where - --- base -import Control.Concurrent -import Control.Monad -import Data.List.NonEmpty ( NonEmpty(..) ) -import GHC.Exts ( Char(..), Int(..), indexCharOffAddr# ) - --- exceptions -import qualified Control.Monad.Catch as MC - -#if defined(mingw32_HOST_OS) --- Win32 -import qualified System.Win32.Event as Win32 - ( createEvent, setEvent - , waitForSingleObject, waitForMultipleObjects - , wAIT_OBJECT_0 ) -import qualified System.Win32.File as Win32 - ( closeHandle ) -import qualified System.Win32.Process as Win32 - ( iNFINITE ) -import qualified System.Win32.Semaphore as Win32 - ( Semaphore(..), sEMAPHORE_ALL_ACCESS - , createSemaphore, openSemaphore, releaseSemaphore ) -import qualified System.Win32.Time as Win32 - ( FILETIME(..), getSystemTimeAsFileTime ) -import qualified System.Win32.Types as Win32 - ( HANDLE, errorWin ) -#else --- base -import Foreign.C.Types - ( CClock(..) ) - --- unix -import qualified System.Posix.Semaphore as Posix - ( Semaphore, OpenSemFlags(..) - , semOpen, semWaitInterruptible, semTryWait, semThreadWait - , semGetValue, semPost, semUnlink ) -import qualified System.Posix.Files as Posix - ( stdFileMode ) -import qualified System.Posix.Process as Posix - ( ProcessTimes(systemTime), getProcessTimes ) -#endif - ---------------------------------------- --- System-specific semaphores - -newtype SemaphoreName = - SemaphoreName { getSemaphoreName :: String } - deriving Eq - --- | A system semaphore (POSIX or Win32). -data Semaphore = - Semaphore - { semaphoreName :: !SemaphoreName - , semaphore :: -#if defined(mingw32_HOST_OS) - !Win32.Semaphore -#else - !Posix.Semaphore -#endif - } - --- | Create a new semaphore with the given name and initial amount of --- available resources. --- --- Throws an error if a semaphore by this name already exists. -createSemaphore :: SemaphoreName - -> Int -- ^ number of tokens on the semaphore - -> IO Semaphore -createSemaphore (SemaphoreName sem_name) init_toks = do - mb_sem <- create_sem sem_name init_toks - case mb_sem of - Left err -> err - Right sem -> return sem - --- | Create a fresh semaphore with the given amount of tokens. --- --- Its name will start with the given prefix, but will have a random suffix --- appended to it. -freshSemaphore :: String -- ^ prefix - -> Int -- ^ number of tokens on the semaphore - -> IO Semaphore -freshSemaphore prefix init_toks = do - suffixes <- random_strings - go 0 suffixes - where - go :: Int -> NonEmpty String -> IO Semaphore - go i (suffix :| suffs) = do - mb_sem <- create_sem (prefix ++ "_" ++ suffix) init_toks - case mb_sem of - Right sem -> return sem - Left err - | next : nexts <- suffs - , i < 32 -- give up after 32 attempts - -> go (i+1) (next :| nexts) - | otherwise - -> err - -create_sem :: String -> Int -> IO (Either (IO Semaphore) Semaphore) -create_sem sem_str init_toks = do -#if defined(mingw32_HOST_OS) - let toks = fromIntegral init_toks - mb_sem <- MC.try @_ @MC.SomeException $ - Win32.createSemaphore Nothing toks toks (Just sem_str) - return $ case mb_sem of - Right (sem, exists) - | exists - -> Left (Win32.errorWin $ "semaphore-compat: semaphore " ++ sem_str ++ " already exists") - | otherwise - -> Right $ mk_sem sem - Left err - -> Left $ MC.throwM err -#else - let flags = - Posix.OpenSemFlags - { Posix.semCreate = True - , Posix.semExclusive = True } - mb_sem <- MC.try @_ @MC.SomeException $ - Posix.semOpen sem_str flags Posix.stdFileMode init_toks - return $ case mb_sem of - Left err -> Left $ MC.throwM err - Right sem -> Right $ mk_sem sem -#endif - where - sem_nm = SemaphoreName sem_str - mk_sem sem = - Semaphore - { semaphore = sem - , semaphoreName = sem_nm } - --- | Open a semaphore with the given name. --- --- If no such semaphore exists, throws an error. -openSemaphore :: SemaphoreName -> IO Semaphore -openSemaphore nm@(SemaphoreName sem_name) = do -#if defined(mingw32_HOST_OS) - sem <- Win32.openSemaphore Win32.sEMAPHORE_ALL_ACCESS True sem_name -#else - let - flags = Posix.OpenSemFlags - { Posix.semCreate = False - , Posix.semExclusive = False } - sem <- Posix.semOpen sem_name flags Posix.stdFileMode 0 -#endif - return $ - Semaphore - { semaphore = sem - , semaphoreName = nm } - --- | Indefinitely wait on a semaphore. --- --- If you want to be able to cancel a wait operation, use --- 'forkWaitOnSemaphoreInterruptible' instead. -waitOnSemaphore :: Semaphore -> IO () -waitOnSemaphore (Semaphore { semaphore = sem }) = -#if defined(mingw32_HOST_OS) - MC.mask_ $ do - () <$ Win32.waitForSingleObject (Win32.semaphoreHandle sem) Win32.iNFINITE -#else - Posix.semThreadWait sem -#endif - --- | Try to obtain a token from the semaphore, without blocking. --- --- Immediately returns 'False' if no resources are available. -tryWaitOnSemaphore :: Semaphore -> IO Bool -tryWaitOnSemaphore (Semaphore { semaphore = sem }) = -#if defined(mingw32_HOST_OS) - MC.mask_ $ do - wait_res <- Win32.waitForSingleObject (Win32.semaphoreHandle sem) 0 - return $ wait_res == Win32.wAIT_OBJECT_0 -#else - Posix.semTryWait sem -#endif - --- | Release a semaphore: add @n@ to its internal counter. --- --- No-op when `n <= 0`. -releaseSemaphore :: Semaphore -> Int -> IO () -releaseSemaphore (Semaphore { semaphore = sem }) n - | n <= 0 - = return () - | otherwise - = MC.mask_ $ do -#if defined(mingw32_HOST_OS) - void $ Win32.releaseSemaphore sem (fromIntegral n) -#else - replicateM_ n (Posix.semPost sem) -#endif - --- | Destroy the given semaphore. -destroySemaphore :: Semaphore -> IO () -destroySemaphore sem = -#if defined(mingw32_HOST_OS) - Win32.closeHandle (Win32.semaphoreHandle $ semaphore sem) -#else - Posix.semUnlink (getSemaphoreName $ semaphoreName sem) -#endif - --- | 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 :: Semaphore -> IO Int -getSemaphoreValue (Semaphore { semaphore = sem }) = -#if defined(mingw32_HOST_OS) - MC.mask_ $ do - wait_res <- Win32.waitForSingleObject (Win32.semaphoreHandle sem) 0 - if wait_res == Win32.wAIT_OBJECT_0 - -- We were able to acquire a resource from the semaphore without waiting: - -- release it immediately, thus obtaining the total number of available - -- resources. - then - (+1) . fromIntegral <$> Win32.releaseSemaphore sem 1 - else - return 0 -#else - Posix.semGetValue sem -#endif - --- | 'WaitId' stores the information we need to cancel a thread --- which is waiting on a semaphore. --- --- See 'forkWaitOnSemaphoreInterruptible' and 'interruptWaitOnSemaphore'. -data WaitId = WaitId { waitingThreadId :: ThreadId -#if defined(mingw32_HOST_OS) - , cancelHandle :: Win32.HANDLE -#endif - } - --- | Spawn a thread that waits on the given semaphore. --- --- In this thread, asynchronous exceptions will be masked. --- --- The waiting operation can be interrupted using the --- 'interruptWaitOnSemaphore' function. -forkWaitOnSemaphoreInterruptible - :: Semaphore - -> ( Either MC.SomeException Bool -> IO () ) -- ^ wait result action - -> IO WaitId -forkWaitOnSemaphoreInterruptible - (Semaphore { semaphore = sem }) - wait_result_action = do -#if defined(mingw32_HOST_OS) - cancelHandle <- Win32.createEvent Nothing True False "" -#endif - let - interruptible_wait :: IO Bool - interruptible_wait = -#if defined(mingw32_HOST_OS) - -- Windows: wait on both the handle used for cancelling the wait - -- and on the semaphore. - do - wait_res <- - Win32.waitForMultipleObjects - [ Win32.semaphoreHandle sem - , cancelHandle ] - False -- False <=> WaitAny - Win32.iNFINITE - return $ wait_res == Win32.wAIT_OBJECT_0 - -- Only in the case that the wait result is WAIT_OBJECT_0 will - -- we have succeeded in obtaining a token from the semaphore. -#else - -- POSIX: use the 'semWaitInterruptible' interruptible FFI call - -- that can be interrupted when we send a killThread signal. - Posix.semWaitInterruptible sem -#endif - waitingThreadId <- forkIO $ MC.mask_ $ do - wait_res <- MC.try interruptible_wait - wait_result_action wait_res - return $ WaitId { .. } - --- | Interrupt a semaphore wait operation initiated by --- 'forkWaitOnSemaphoreInterruptible'. -interruptWaitOnSemaphore :: WaitId -> IO () -interruptWaitOnSemaphore ( WaitId { .. } ) = do -#if defined(mingw32_HOST_OS) - Win32.setEvent cancelHandle - -- On Windows, we signal to stop waiting. -#endif - killThread waitingThreadId - -- On POSIX, killing the thread will cancel the wait on the semaphore - -- due to the FFI call being interruptible ('semWaitInterruptible'). - ---------------------------------------- --- Abstract semaphores - --- | Abstraction over the operations of a semaphore. -data AbstractSem = - AbstractSem - { acquireSem :: IO () - , releaseSem :: IO () - } - -withAbstractSem :: AbstractSem -> IO b -> IO b -withAbstractSem sem = MC.bracket_ (acquireSem sem) (releaseSem sem) - ---------------------------------------- --- Utility - -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"# - -random_strings :: IO (NonEmpty String) -random_strings = do -#if defined(mingw32_HOST_OS) - Win32.FILETIME t <- Win32.getSystemTimeAsFileTime -#else - CClock t <- Posix.systemTime <$> Posix.getProcessTimes -#endif - return $ fmap ( \ i -> iToBase62 (i + fromIntegral t) ) (0 :| [1..]) +{-# LANGUAGE CPP #-}++module System.Semaphore+ ( -- * System semaphores++ -- $server-vs-client++ ClientSemaphore, ServerSemaphore, SemaphoreName(..)++ -- | The name of a client semaphore+ , clientSemaphoreName+ , semaphoreIdentifier+ -- | Retrieve the client semaphore corresponding to a server semaphore+ , serverClientSemaphore++ -- ** Creating a semaphore+ , createSemaphore, freshSemaphore++ -- ** Opening a semaphore+ , SemaphoreToken+ , openSemaphore+ , SemaphoreIdentifier, parseSemaphoreIdentifier+ , SemaphoreProtocolVersion(..)+ , semaphoreVersion+ , versionsAreCompatible+ , SemaphoreError(..)++ -- ** Requesting a token+ , waitOnSemaphore, tryWaitOnSemaphore+ , withSemaphoreToken+ , getSemaphoreValue++ -- ** Releasing resources+ , releaseSemaphoreToken++ -- $destroying+ , destroyClientSemaphore, destroyServerSemaphore++ -- * Abstract semaphores+ , AbstractSem(..)+ , withAbstractSem+ ) where++{- $server-vs-client++Since version 2 of @semaphore-compat@, we distinguish 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'.++When the jobserver wants to also act as a jobclient, it can use+'serverClientSemaphore' to obtain the 'ClientSemaphore' corresponding+to its 'ServerSemaphore'.++This architecture allows the jobserver to keep full accounting of+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'.+-}++-- base+import Data.List.NonEmpty ( NonEmpty(..) )++-- exceptions+import qualified Control.Monad.Catch as MC++import System.Semaphore.Internal.Common++#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++---------------------------------------+-- Version compatibility++-- | Check whether two semaphore protocol versions are compatible.+-- Only identical versions are compatible.+versionsAreCompatible :: SemaphoreProtocolVersion -> SemaphoreProtocolVersion -> Bool+versionsAreCompatible a b = 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)+createSemaphore label init_toks = do+ let sem_nm = SemaphoreName+ { semaphoreProtocolVersion = semaphoreVersion+ , unversionedSemaphoreNameString = label+ }+ create_sem sem_nm init_toks++-- | Create a fresh semaphore with a unique name and the given token count.+--+-- The name is derived from the given prefix with a random suffix.+freshSemaphore :: String -- ^ label prefix+ -> Int -- ^ number of tokens on the semaphore+ -> IO (Either SemaphoreError ServerSemaphore)+freshSemaphore prefix init_toks = do+ seed <- getTimeSeed+ go 0 (seedStrings seed)+ where+ go :: Int -> NonEmpty String -> IO (Either SemaphoreError ServerSemaphore)+ go i (suffix :| suffs) = do+ let sem_str = prefix ++ "_" ++ suffix+ sem_nm = SemaphoreName+ { semaphoreProtocolVersion = semaphoreVersion+ , unversionedSemaphoreNameString = sem_str+ }+ mb_sem <- create_sem sem_nm init_toks+ case mb_sem of+ Right sem -> return (Right sem)+ Left err+ | next : nexts <- suffs+ , i < 32 -- give up after 32 attempts+ -> go (i+1) (next :| nexts)+ | otherwise+ -> return (Left err)++-- | Open a semaphore from its 'SemaphoreIdentifier'.+--+-- 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@.+openSemaphore :: SemaphoreIdentifier -> IO (Either SemaphoreError ClientSemaphore)+openSemaphore ident =+ case parseSemaphoreIdentifier ident of+ Nothing+ | versionsAreCompatible v1 semaphoreVersion ->+ open_sem_raw (SemaphoreName { semaphoreProtocolVersion = v1+ , unversionedSemaphoreNameString = ident })+ | otherwise ->+ return $ Left $ semVerError v1+ Just nm+ | not (versionsAreCompatible (semaphoreProtocolVersion nm) semaphoreVersion) ->+ return $ Left $ semVerError (semaphoreProtocolVersion nm)+ | otherwise ->+ open_sem_raw nm+ where+ v1 = SemaphoreProtocolVersion 1+ semVerError ver = SemaphoreIncompatibleVersion ver semaphoreVersion++-- | 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++seedStrings :: Int -> NonEmpty String+seedStrings seed = fmap ( \ i -> iToBase62 (i + seed) ) (0 :| [1..])++---------------------------------------+-- Abstract semaphores++-- | Abstraction over the operations of a semaphore.+data AbstractSem =+ AbstractSem+ { acquireSem :: IO ()+ , releaseSem :: IO ()+ }++withAbstractSem :: AbstractSem -> IO b -> IO b+withAbstractSem s = MC.bracket_ (acquireSem s) (releaseSem s)
+ src/System/Semaphore/Internal/Common.hs view
@@ -0,0 +1,142 @@+{-# 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)
+ src/System/Semaphore/Internal/DomainSocket.hs view
@@ -0,0 +1,101 @@+{-# LANGUAGE ForeignFunctionInterface #-}++module System.Semaphore.Internal.DomainSocket+ ( connectDomainSocket+ , listenDomainSocket+ , pollAcceptSocket, AcceptResult(..)+ , fdReadByte, fdWriteByte+ , fdShutdown+ ) where++-- base+import Data.Word ( Word8 )+import Foreign.C.Error ( throwErrnoIfMinus1Retry, throwErrnoIfMinus1Retry_, throwErrno )+import Foreign.C.String ( CString, withCString )+import Foreign.C.Types ( CInt(..), CSize(..) )+import Foreign.Marshal.Alloc ( allocaBytes )+import Foreign.Ptr ( Ptr )+import Foreign.Storable ( peek, poke )+import GHC.IO.Exception ( IOErrorType(EOF), IOException(..) )+import GHC.Stack ( HasCallStack, callStack, prettyCallStack )++-- unix+import System.Posix.Types ( Fd(..) )++foreign import ccall safe "hs_connect_domain_socket"+ c_connectDomainSocket :: CString -> IO CInt++connectDomainSocket :: FilePath -> IO Fd+connectDomainSocket path =+ withCString path $ fmap Fd . throwErrnoIfMinus1Retry "connectDomainSocket" . c_connectDomainSocket++foreign import ccall safe "hs_listen_domain_socket"+ c_listenDomainSocket :: CString -> IO CInt++-- | Open a socket in non blocking mode (O_NONBLOCK)+listenDomainSocket :: FilePath -> IO Fd+listenDomainSocket path =+ withCString path $ fmap Fd . throwErrnoIfMinus1Retry "listenDomainSocket" . c_listenDomainSocket++foreign import ccall safe "read"+ c_read :: CInt -> Ptr Word8 -> CSize -> IO CInt++foreign import ccall safe "write"+ c_write :: CInt -> Ptr Word8 -> CSize -> IO CInt++-- | Read a single byte from a file descriptor.+-- Throws an EOF 'IOError' if the peer has disconnected.+fdReadByte :: HasCallStack => Fd -> IO Word8+fdReadByte (Fd fd) =+ allocaBytes 1 $ \buf -> do+ rc <- throwErrnoIfMinus1Retry ("fdReadByte(fd=" ++ show fd ++ ")") $+ c_read fd buf 1+ if rc == 0+ then ioError $ IOError Nothing EOF+ (prettyCallStack callStack)+ ("fd=" ++ show fd)+ Nothing Nothing+ else peek buf++-- | Write a single byte to a file descriptor.+fdWriteByte :: HasCallStack => Fd -> Word8 -> IO ()+fdWriteByte (Fd fd) byte =+ allocaBytes 1 $ \buf -> do+ poke buf byte+ _ <- throwErrnoIfMinus1Retry ("fdWriteByte(fd=" ++ show fd ++ ")") $+ c_write fd buf 1+ return ()++foreign import ccall safe "shutdown"+ c_shutdown :: CInt -> CInt -> IO CInt++-- | Shut down a socket for both reading and writing.+-- A concurrent 'fdReadByte' on the same fd will return immediately+-- Used to cancel threads blocked in read()+fdShutdown :: Fd -> IO ()+fdShutdown (Fd fd) =+ throwErrnoIfMinus1Retry_ "fdShutdown" $ c_shutdown fd 2 -- SHUT_RDWR++-- | Result of 'pollAcceptSocket'.+data AcceptResult+ = AcceptedFd !Fd -- ^ A client connected; here is the fd.+ | AcceptCancelled -- ^ The cancel pipe was signalled.++foreign import ccall safe "hs_poll_accept"+ c_pollAccept :: CInt -> CInt -> IO CInt++-- | Block until either a client connects or the cancel fd is written to.+--+-- Relies on cooperative cancellation implemented in hs_poll_accept using+-- @poll(2)@ + @accept(2)@ via safe FFI to avoid GHC #27110 and #27113.+--+-- Must be called from a masked context. The caller is responsible for+-- installing an exception handler that closes all 3 fds (inputs and outputs).+pollAcceptSocket :: Fd -> Fd -> IO AcceptResult+pollAcceptSocket (Fd listenFd) (Fd cancelFd) = do+ r <- c_pollAccept listenFd cancelFd+ if r == -2+ then return AcceptCancelled+ else if r == -1+ then throwErrno "pollAcceptSocket"+ else return (AcceptedFd (Fd r))
+ src/System/Semaphore/Internal/Posix.hs view
@@ -0,0 +1,447 @@+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module System.Semaphore.Internal.Posix+ ( ClientSemaphore(..), ServerSemaphore(..)+ , SemaphoreToken(..)+ , create_sem, open_sem_raw+ , waitOnSemaphore, tryWaitOnSemaphore+ , releaseSemaphoreToken+ , destroyClientSemaphore, destroyServerSemaphore+ , getSemaphoreValue+ , getTimeSeed+ ) where++-- base+import Control.Concurrent+ ( 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 )+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++-- stm+import Control.Concurrent.STM+ ( TVar, atomically, newTVarIO, readTVar, readTVarIO+ , modifyTVar', writeTVar, retry )++-- directory+import System.Directory ( doesPathExist )++-- unix+import System.Posix.IO ( closeFd, createPipe )+import System.Posix.Files ( removeLink )+import System.Posix.Types ( Fd )+import System.Posix.Process ( getProcessID )++import System.Semaphore.Internal.Common+import System.Semaphore.Internal.DomainSocket+ ( connectDomainSocket, listenDomainSocket+ , pollAcceptSocket, AcceptResult(..)+ , fdReadByte, fdWriteByte+ , fdShutdown )++-- | A semaphore 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.+newtype SemaphoreToken = SemaphoreToken+ { tokenFdLock :: MVar Fd+ }++-- | 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+ }++data ServerState = ServerState+ { serverListenFd :: !Fd+ , serverCancelFd :: !Fd+ -- ^ 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+ 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+ pool <- newTVarIO init_toks+ (cancelRd, cancelWr) <- createPipe+ tid <- forkIOWithUnmask $ \unmask ->+ unmask (serverLoop pool listenFd cancelRd)+ `MC.finally` closeFd cancelRd+ stateVar <- newMVar ServerState+ { serverListenFd = listenFd+ , serverCancelFd = cancelWr+ }+ return ServerSemaphore+ { serverClientSemaphore = ClientSemaphore { clientSemaphoreName = sem_nm+ , semSocketPath = socketPath }+ , serverThreadId = tid+ , serverPool = pool+ , serverState = stateVar+ }+ return $ case mb_res of+ Left err -> Left $ SemaphoreOtherError err+ Right sem -> Right sem++open_sem_raw :: SemaphoreName -> IO (Either SemaphoreError ClientSemaphore)+open_sem_raw nm = do+ mb_res <- MC.try @_ @IOException $ do+ socketPath <- getSemaphoreSocketPath nm+ exists <- doesPathExist socketPath+ return (socketPath, exists)+ return $ case mb_res of+ Left err -> Left $ SemaphoreOtherError err+ Right (_, False) -> Left $ SemaphoreDoesNotExist (semaphoreIdentifier nm)+ Right (socketPath, _) -> Right $+ ClientSemaphore+ { clientSemaphoreName = nm+ , 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+ fd <- connectDomainSocket (semSocketPath sem)+ -- The read() runs in a forked thread+ workerTid <- forkIOWithUnmask $ \_ -> do+ res <- MC.try @_ @MC.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+ -- shutdown(SHUT_RDWR) causes the worker's read() to return EOF immediately+ void $ MC.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+ -- 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+ case res of+ Right resp+ | resp == RspOk -> mkToken fd+ | otherwise -> do void $ MC.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++-- | 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+ fd <- connectDomainSocket (semSocketPath sem)+ resp <- flip MC.onException (closeFd fd) $ do+ fdWriteByte fd CmdTryWait+ fdReadByte fd+ case resp of+ RspOk -> Just <$> mkToken fd+ _ -> do void $ MC.try @_ @IOException $ closeFd fd+ return Nothing++mkToken :: Fd -> IO SemaphoreToken+mkToken fd = do+ fdVar <- newMVar fd+ _ <- mkWeakMVar fdVar $ do+ mb <- tryTakeMVar fdVar+ case mb of+ 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'+ 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+ 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)+ case resp of+ RspOk -> return ()+ -- myCount <= 0 on the server means the token is effectively+ -- already released; stay idempotent.+ 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+ -- 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,+ -- but otherwise it should exit promptly.+ --+ -- Without uninterruptibleMask_, we potentially leak serverListenFd and path if an+ -- exception arrives when we are in `killThread`.+ mbState <- tryTakeMVar (serverState server)+ case mbState of+ Nothing -> return () -- already destroyed+ 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+ killThread (serverThreadId server)+ void $ MC.try @_ @IOException $ closeFd serverListenFd+ void $ MC.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+ 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
+ src/System/Semaphore/Internal/Unsupported.hs view
@@ -0,0 +1,67 @@+{-# LANGUAGE EmptyDataDecls #-}+{-# LANGUAGE EmptyCase #-}+{-# LANGUAGE LambdaCase #-}++-- | 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+ , waitOnSemaphore, tryWaitOnSemaphore+ , releaseSemaphoreToken+ , destroyClientSemaphore, destroyServerSemaphore+ , getSemaphoreValue+ , getTimeSeed+ ) where++import Control.Exception ( IOException )+import GHC.IO.Exception ( unsupportedOperation )+import System.IO.Error ( ioeSetLocation )++import System.Semaphore.Internal.Common++data ClientSemaphore++data ServerSemaphore++data SemaphoreToken++clientSemaphoreName :: ClientSemaphore -> SemaphoreName+clientSemaphoreName = \case {}++serverClientSemaphore :: ServerSemaphore -> ClientSemaphore+serverClientSemaphore = \case {}++unsupported :: String -> IOException+unsupported op = ioeSetLocation unsupportedOperation ("semaphore-compat." ++ op)++create_sem :: SemaphoreName -> Int -> IO (Either SemaphoreError ServerSemaphore)+create_sem _ _ =+ return $ Left $ SemaphoreOtherError $ unsupported "createSemaphore"++open_sem_raw :: SemaphoreName -> IO (Either SemaphoreError ClientSemaphore)+open_sem_raw _ =+ return $ Left $ SemaphoreOtherError $ unsupported "openSemaphore"++waitOnSemaphore :: ClientSemaphore -> IO SemaphoreToken+waitOnSemaphore = \case {}++tryWaitOnSemaphore :: ClientSemaphore -> IO (Maybe SemaphoreToken)+tryWaitOnSemaphore = \case {}++releaseSemaphoreToken :: SemaphoreToken -> IO ()+releaseSemaphoreToken = \case {}++destroyClientSemaphore :: ClientSemaphore -> IO ()+destroyClientSemaphore = \case {}++destroyServerSemaphore :: ServerSemaphore -> IO ()+destroyServerSemaphore = \case {}++getSemaphoreValue :: ServerSemaphore -> IO Int+getSemaphoreValue = \case {}++getTimeSeed :: IO Int+getTimeSeed =+ ioError $ unsupported "getTimeSeed"
+ src/System/Semaphore/Internal/Win32.hs view
@@ -0,0 +1,218 @@+{-# LANGUAGE TypeApplications #-}++module System.Semaphore.Internal.Win32+ ( ClientSemaphore(..), ServerSemaphore(..)+ , SemaphoreToken(..)+ , create_sem, open_sem_raw+ , waitOnSemaphore, tryWaitOnSemaphore+ , releaseSemaphoreToken+ , destroyClientSemaphore, destroyServerSemaphore+ , getSemaphoreValue+ , getTimeSeed+ ) where++import Control.Concurrent+ ( forkIO, killThread )+import Control.Concurrent.MVar+ ( MVar, newEmptyMVar, newMVar, putMVar, takeMVar, tryReadMVar, tryTakeMVar )+import Control.Exception ( IOException, uninterruptibleMask_ )+import Control.Monad+import System.IO.Error ( isDoesNotExistError )++import qualified Control.Monad.Catch as MC++import qualified System.Win32.Event as Win32+ ( createEvent, setEvent+ , waitForSingleObject, waitForMultipleObjects+ , wAIT_OBJECT_0 )+import qualified System.Win32.File as Win32+ ( closeHandle )+import qualified System.Win32.Process as Win32+ ( iNFINITE )+import qualified System.Win32.Semaphore as Win32+ ( Semaphore(..), sEMAPHORE_ALL_ACCESS+ , createSemaphore, openSemaphore, releaseSemaphore )+import qualified System.Win32.Time as Win32+ ( FILETIME(..), getSystemTimeAsFileTime )++import System.Semaphore.Internal.Common++-- | 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.+ }++-- | A held semaphore token.+newtype SemaphoreToken = SemaphoreToken+ { tokenSemaphore :: Win32.Semaphore+ }++-- | 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+ let toks = fromIntegral init_toks+ mb_sem <- MC.try @_ @IOException $+ Win32.createSemaphore Nothing toks 0x7FFFFFFF (Just (semaphoreIdentifier sem_nm))+ case mb_sem of+ Left err ->+ return $ Left $ SemaphoreOtherError err+ 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)+ return $ Left $ SemaphoreAlreadyExists (semaphoreIdentifier sem_nm)+ Right (sem, False) -> do+ semVar <- newMVar sem+ return $ Right $+ ServerSemaphore+ { serverClientSemaphore =+ ClientSemaphore+ { clientSemaphoreName = sem_nm+ , semaphore = semVar }+ }++open_sem_raw :: SemaphoreName -> IO (Either SemaphoreError ClientSemaphore)+open_sem_raw nm = MC.mask_ $ do+ mb_res <- MC.try @_ @IOException $+ Win32.openSemaphore Win32.sEMAPHORE_ALL_ACCESS True (semaphoreIdentifier nm)+ case mb_res of+ Left err+ | isDoesNotExistError err+ -> return $ Left $ SemaphoreDoesNotExist (semaphoreIdentifier nm)+ | otherwise+ -> return $ Left $ SemaphoreOtherError err+ Right sem -> do+ semVar <- newMVar sem+ return $ Right $+ ClientSemaphore+ { clientSemaphoreName = nm+ , semaphore = semVar }++-- | Read the kernel handle from a 'ClientSemaphore'. Throws if the+-- client has been destroyed.+readClientHandle :: ClientSemaphore -> String -> IO Win32.Semaphore+readClientHandle client opName = do+ mb <- tryReadMVar (semaphore client)+ case mb of+ Just sem -> return sem+ 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+ sem <- readClientHandle client "waitOnSemaphore"+ cancelEvent <- Win32.createEvent Nothing True False ""+ workerTid <- (forkIO $ MC.mask_ $ do+ res <- MC.try @_ @MC.SomeException $ do+ wait_res <-+ Win32.waitForMultipleObjects+ [ Win32.semaphoreHandle sem+ , cancelEvent ]+ False -- WaitAny+ Win32.iNFINITE+ if wait_res == Win32.wAIT_OBJECT_0+ then return (SemaphoreToken sem)+ else ioError (userError "semaphore wait cancelled")+ putMVar resultVar res+ ) `MC.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+ case ok of+ Right () ->+ -- Cancel event signaled. Worker's WaitForMultipleObjects+ -- returns immediately. Safe to wait uninterruptibly.+ uninterruptibleMask_ $ killThread workerTid+ Left _ ->+ -- setEvent failed. We can't really do anything.+ -- The worker is in a safe ffi call so killThread+ -- will just block forever.+ return ()+ ) `MC.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+ _ -> return ()+ Win32.closeHandle cancelEvent)+ res <- takeMVar resultVar `MC.onException` cleanup+ Win32.closeHandle cancelEvent+ case res of+ Right tok -> return tok+ Left e -> MC.throwM 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+ sem <- readClientHandle client "tryWaitOnSemaphore"+ wait_res <- Win32.waitForSingleObject (Win32.semaphoreHandle sem) 0+ if wait_res == Win32.wAIT_OBJECT_0+ then return (Just (SemaphoreToken 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++-- | 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)+ case mb of+ 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+ sem <- readClientHandle (serverClientSemaphore server) "getSemaphoreValue"+ wait_res <- Win32.waitForSingleObject (Win32.semaphoreHandle sem) 0+ if wait_res == Win32.wAIT_OBJECT_0+ then+ (+1) . fromIntegral <$> Win32.releaseSemaphore sem 1+ else+ return 0++getTimeSeed :: IO Int+getTimeSeed = do+ Win32.FILETIME t <- Win32.getSystemTimeAsFileTime+ return (fromIntegral t)
+ test-common/System/Semaphore/TestHelper.hs view
@@ -0,0 +1,277 @@+{-# LANGUAGE CPP #-}++module System.Semaphore.TestHelper (+ -- Protocol+ Command(..), Response(..),+ encodeCommand, decodeCommand,+ encodeResponse, decodeResponse,+ -- Helper main loop (used by helper executable)+ helperMain,+ -- Test-side utilities (used by test suite)+ HelperHandle(..),+ withHelper,+ sendCommand, sendCommand_,+ sendCommandNoResponse,+ killHelper,+) where++import Control.Exception ( bracket, try, SomeException )+import Control.Monad ( when )+import Data.IORef ( IORef, newIORef, readIORef, writeIORef, modifyIORef' )+import Foreign.C.Types ( CInt(..) )+import System.Exit ( exitSuccess )+import System.IO+ ( Handle, hSetBuffering, hFlush, hGetLine, hPutStrLn, hClose+ , BufferMode(LineBuffering), stdin, stdout, hIsEOF )+import System.Process+ ( CreateProcess(..), StdStream(..), ProcessHandle+ , createProcess, proc, waitForProcess, getProcessExitCode+ , terminateProcess )++import System.Semaphore++#if !defined(mingw32_HOST_OS)+import System.Posix.Signals ( signalProcess, sigKILL )+import System.Process ( getPid )+#endif++--------------------------------------------------------------------------+-- Protocol types++data Command+ = CmdOpen String -- ^ open <name>+ | CmdWait -- ^ wait+ | CmdWaitN Int -- ^ wait-n <n>+ | CmdRelease Int -- ^ release <n>+ | CmdDestroy -- ^ destroy+ | CmdCrash -- ^ crash+ deriving (Show, Eq)++data Response+ = RspOk+ | RspInt Int+ | RspError String+ deriving (Show, Eq)++encodeCommand :: Command -> String+encodeCommand (CmdOpen name) = "open " ++ name+encodeCommand CmdWait = "wait"+encodeCommand (CmdWaitN n) = "wait-n " ++ show n+encodeCommand (CmdRelease n) = "release " ++ show n+encodeCommand CmdDestroy = "destroy"+encodeCommand CmdCrash = "crash"++decodeCommand :: String -> Maybe Command+decodeCommand "wait" = Just CmdWait+decodeCommand "destroy" = Just CmdDestroy+decodeCommand "crash" = Just CmdCrash+decodeCommand s+ | Just rest <- stripPrefix' "open " s+ = Just (CmdOpen rest)+ | Just rest <- stripPrefix' "wait-n " s+ , [(n, "")] <- reads rest+ = Just (CmdWaitN n)+ | Just rest <- stripPrefix' "release " s+ , [(n, "")] <- reads rest+ = Just (CmdRelease n)+ | otherwise+ = Nothing++encodeResponse :: Response -> String+encodeResponse RspOk = "ok"+encodeResponse (RspInt n) = "int " ++ show n+encodeResponse (RspError msg) = "error " ++ msg++decodeResponse :: String -> Maybe Response+decodeResponse "ok" = Just RspOk+decodeResponse s+ | Just rest <- stripPrefix' "int " s+ , [(n, "")] <- reads rest+ = Just (RspInt n)+ | Just rest <- stripPrefix' "error " s+ = Just (RspError rest)+ | otherwise+ = Nothing++stripPrefix' :: String -> String -> Maybe String+stripPrefix' [] ys = Just ys+stripPrefix' (_:_) [] = Nothing+stripPrefix' (x:xs) (y:ys)+ | x == y = stripPrefix' xs ys+ | otherwise = Nothing++--------------------------------------------------------------------------+-- FFI for _exit (used by crash command)++foreign import ccall "stdlib.h _exit" c_exit :: CInt -> IO ()++--------------------------------------------------------------------------+-- Helper main loop++-- Helper state: semaphore identity + list of held tokens.+data HelperState = HelperState+ { hsSem :: !(Maybe ClientSemaphore)+ , hsTokens :: ![SemaphoreToken]+ }++helperMain :: IO ()+helperMain = do+ hSetBuffering stdin LineBuffering+ hSetBuffering stdout LineBuffering+ ref <- newIORef (HelperState Nothing [])+ helperLoop ref++helperLoop :: IORef HelperState -> IO ()+helperLoop ref = do+ eof <- hIsEOF stdin+ if eof+ then do+ -- Clean exit on EOF: release all tokens and destroy semaphore+ st <- readIORef ref+ mapM_ releaseSemaphoreToken (hsTokens st)+ case hsSem st of+ Just sem -> destroyClientSemaphore sem+ Nothing -> return ()+ exitSuccess+ else do+ line <- hGetLine stdin+ case decodeCommand line of+ Nothing -> do+ respond (RspError ("unknown command: " ++ line))+ helperLoop ref+ Just cmd -> do+ executeCommand ref cmd++executeCommand :: IORef HelperState -> Command -> IO ()+executeCommand ref cmd = case cmd of+ CmdOpen name -> do+ result <- openSemaphore name+ case result of+ Right sem -> do+ modifyIORef' ref (\st -> st { hsSem = Just sem })+ respond RspOk+ Left err ->+ respond (RspError (show err))+ helperLoop ref++ CmdWait -> do+ st <- readIORef ref+ case hsSem st of+ Nothing -> do+ respond (RspError "no semaphore open")+ helperLoop ref+ Just sem -> do+ tok <- waitOnSemaphore sem+ writeIORef ref (st { hsTokens = tok : hsTokens st })+ respond RspOk+ helperLoop ref++ CmdWaitN n -> do+ st <- readIORef ref+ case hsSem st of+ Nothing -> do+ respond (RspError "no semaphore open")+ helperLoop ref+ Just sem -> do+ toks <- sequence [ waitOnSemaphore sem | _ <- [1..n] ]+ writeIORef ref (st { hsTokens = toks ++ hsTokens st })+ respond RspOk+ helperLoop ref++ CmdRelease n -> do+ st <- readIORef ref+ let (toRelease, toKeep) = splitAt n (hsTokens st)+ mapM_ releaseSemaphoreToken toRelease+ writeIORef ref (st { hsTokens = toKeep })+ respond RspOk+ helperLoop ref++ CmdDestroy -> do+ st <- readIORef ref+ mapM_ releaseSemaphoreToken (hsTokens st)+ case hsSem st of+ Just sem -> destroyClientSemaphore sem+ Nothing -> return ()+ writeIORef ref (HelperState Nothing [])+ respond RspOk+ exitSuccess++ CmdCrash -> do+ c_exit 0++respond :: Response -> IO ()+respond r = do+ hPutStrLn stdout (encodeResponse r)+ hFlush stdout++--------------------------------------------------------------------------+-- Test-side: HelperHandle and utilities++data HelperHandle = HelperHandle+ { hStdin :: Handle+ , hStdout :: Handle+ , hProcess :: ProcessHandle+ }++startHelper :: IO HelperHandle+startHelper = do+ (Just hin, Just hout, _, 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)++cleanupHelper :: HelperHandle -> IO ()+cleanupHelper h = do+ ec <- getProcessExitCode (hProcess h)+ when (ec == Nothing) $ do+ _ <- try (terminateProcess (hProcess h)) :: IO (Either SomeException ())+ _ <- waitForProcess (hProcess h)+ return ()+ _ <- try (hClose (hStdin h)) :: IO (Either SomeException ())+ _ <- try (hClose (hStdout h)) :: IO (Either SomeException ())+ return ()++withHelper :: (HelperHandle -> IO a) -> IO a+withHelper = bracket startHelper cleanupHelper++-- | Send a command and wait for a response.+sendCommand :: HelperHandle -> Command -> IO Response+sendCommand h cmd = do+ hPutStrLn (hStdin h) (encodeCommand cmd)+ hFlush (hStdin h)+ line <- hGetLine (hStdout h)+ case decodeResponse line of+ Just r -> return r+ Nothing -> return (RspError ("bad response: " ++ line))++-- | Send a command and expect 'RspOk'. Throws on error.+sendCommand_ :: HelperHandle -> Command -> IO ()+sendCommand_ h cmd = do+ r <- sendCommand h cmd+ case r of+ RspOk -> return ()+ _ -> error ("sendCommand_: expected ok, got " ++ show r)++-- | Send a command but don't read a response (for blocking commands like wait).+sendCommandNoResponse :: HelperHandle -> Command -> IO ()+sendCommandNoResponse h cmd = do+ hPutStrLn (hStdin h) (encodeCommand cmd)+ hFlush (hStdin h)++-- | Kill the helper process with SIGKILL (Unix) or terminateProcess (Windows).+killHelper :: HelperHandle -> IO ()+killHelper h = do+#if !defined(mingw32_HOST_OS)+ mbPid <- getPid (hProcess h)+ case mbPid of+ Just pid -> signalProcess sigKILL pid+ Nothing -> terminateProcess (hProcess h)+#else+ terminateProcess (hProcess h)+#endif+ _ <- waitForProcess (hProcess h)+ return ()
+ test/Main.hs view
@@ -0,0 +1,1215 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Main (main) where++import Control.Concurrent+ ( forkIO, killThread, myThreadId, threadDelay+ , newEmptyMVar, putMVar, takeMVar )+import Control.Exception+ ( SomeException, AsyncException(ThreadKilled)+ , bracket, mask, mask_, throwTo, try )+import System.Exit ( ExitCode )+import System.Mem ( performMajorGC )+import Control.Monad+ ( void, forM_, replicateM, replicateM_ )+import Data.IORef+ ( IORef, newIORef, readIORef, writeIORef, atomicModifyIORef' )+import Data.List ( isPrefixOf )+import System.Process+ ( waitForProcess )+import System.Timeout ( timeout )++import Test.Tasty+import Test.Tasty.ExpectedFailure (expectFailBecause)+import Test.Tasty.HUnit++import System.Semaphore+import qualified System.Semaphore.Internal.Common as Internal+import System.Semaphore.TestHelper++#if !defined(mingw32_HOST_OS)+import Control.Concurrent.MVar ( tryTakeMVar )+import System.Semaphore.Internal.Posix ( SemaphoreToken(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+-- sending a release command. The server should detect the disconnect+-- and return the token to the pool.+crashToken :: SemaphoreToken -> IO ()+#if !defined(mingw32_HOST_OS)+crashToken tok = do+ mb <- tryTakeMVar (tokenFdLock tok)+ case mb of+ Just fd -> closeFd fd+ Nothing -> return () -- already released+#else+crashToken _tok = return () -- Win32: no fd to close; crash recovery not supported+#endif++-- | Shorthand for constructing 'SemaphoreProtocolVersion' in tests.+v :: Int -> SemaphoreProtocolVersion+v = SemaphoreProtocolVersion++--------------------------------------------------------------------------+-- Test helpers++unwrapServer :: Either SemaphoreError ServerSemaphore -> IO ServerSemaphore+unwrapServer (Right srv) = return srv+unwrapServer (Left err) = assertFailure ("unexpected SemaphoreError: " ++ show err)+ >> error "unreachable"++unwrapSem :: Either SemaphoreError ClientSemaphore -> IO ClientSemaphore+unwrapSem (Right sem) = return sem+unwrapSem (Left err) = assertFailure ("unexpected SemaphoreError: " ++ show err)+ >> error "unreachable"++withSemaphore :: String -> Int -> (ServerSemaphore -> IO a) -> IO a+withSemaphore _label toks =+ bracket (freshSemaphore "sem_test" toks >>= unwrapServer)+ destroyServerSemaphore++withFreshSemaphore :: Int -> (ServerSemaphore -> IO a) -> IO a+withFreshSemaphore toks =+ bracket (freshSemaphore "sem_test" toks >>= unwrapServer)+ destroyServerSemaphore++-- | Assert that an action completes within the given number of+-- microseconds.+assertCompletes :: String -> Int -> IO a -> IO a+assertCompletes msg us action = do+ result <- timeout us action+ case result of+ Just a -> return a+ Nothing -> do+ assertFailure (msg ++ ": timed out")+ error "unreachable"++-- | Assert that an action does NOT complete within 500 ms.+assertBlocks :: String -> IO a -> IO ()+assertBlocks msg action = do+ result <- timeout 500000 action+ case result of+ Nothing -> return ()+ Just _ -> assertFailure (msg ++ ": expected to block but completed")++-- | Wait (with timeout) for the semaphore pool to reach the expected value.+-- Polls every 50ms for up to 5 seconds.+awaitPoolValue :: ServerSemaphore -> Int -> IO ()+awaitPoolValue srv expected = go (100 :: Int) -- 100 * 50ms = 5s+ where+ go 0 = do+ v <- getSemaphoreValue srv+ assertBool ("pool value: expected " ++ show expected ++ ", got " ++ show v) (v == expected)+ go n = do+ v <- getSemaphoreValue srv+ if v == expected+ then return ()+ else threadDelay 50000 >> go (n - 1)++-- | On Windows, Win32 named semaphores do not support crash recovery:+-- CloseHandle does not return tokens acquired via WaitForSingleObject.+-- Tests that expect crash recovery are marked as expected failures.+expectCrashRecoveryFail :: TestTree -> TestTree+#if defined(mingw32_HOST_OS)+expectCrashRecoveryFail = expectFailBecause+ "Win32 named semaphores do not support crash recovery"+#else+expectCrashRecoveryFail = id+#endif++--------------------------------------------------------------------------+-- Main++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests = testGroup "semaphore-compat"+ [ internalTests+ , versionCompatTests+ , createSemaphoreTests+ , freshSemaphoreTests+ , openSemaphoreTests+ , waitOnSemaphoreTests+ , tryWaitOnSemaphoreTests+ , releaseSemaphoreTests+ , getSemaphoreValueTests+ , destroySemaphoreTests+ , interruptibleWaitTests+ , semaphoreErrorTests+ , abstractSemTests+ , multiClientTests+ , sharedHandleTests+ , robustnessTests+ , subprocessTests+ , regressionTests+ ]++--------------------------------------------------------------------------+-- 1. Internal.Common (pure unit tests)++internalTests :: TestTree+internalTests = testGroup "Internal.Common"+ [ testGroup "parseSemaphoreIdentifier"+ [ testCase "valid: v2-foobar" $+ parseSemaphoreIdentifier "v2-foobar" @?= Just (SemaphoreName (v 2) "foobar")+ , testCase "valid: multi-digit version" $+ parseSemaphoreIdentifier "v12-bar" @?= Just (SemaphoreName (v 12) "bar")+ , testCase "invalid: empty" $+ parseSemaphoreIdentifier "" @?= Nothing+ , testCase "invalid: foo" $+ parseSemaphoreIdentifier "foo" @?= Nothing+ , testCase "invalid: v-foo (no digit)" $+ parseSemaphoreIdentifier "v-foo" @?= Nothing+ , testCase "invalid: v2foo (no dash)" $+ parseSemaphoreIdentifier "v2foo" @?= Nothing+ , testCase "invalid: 2-foo (no v)" $+ parseSemaphoreIdentifier "2-foo" @?= Nothing+ , testCase "invalid: vx-foo (non-numeric)" $+ parseSemaphoreIdentifier "vx-foo" @?= Nothing+ ]+ , testGroup "semaphoreIdentifier"+ [ testCase "v2 includes prefix" $+ semaphoreIdentifier (SemaphoreName (v 2) "foobar") @?= "v2-foobar"+ , testCase "v1 has no prefix" $+ semaphoreIdentifier (SemaphoreName (v 1) "foobar") @?= "foobar"+ , testCase "round-trips with parseSemaphoreIdentifier for v2" $+ fmap semaphoreIdentifier (parseSemaphoreIdentifier "v2-foobar")+ @?= Just "v2-foobar"+ ]+ , testGroup "iToBase62"+ [ testCase "0 -> \"0\"" $ Internal.iToBase62 0 @?= "0"+ , testCase "10 -> \"a\"" $ Internal.iToBase62 10 @?= "a"+ , testCase "36 -> \"A\"" $ Internal.iToBase62 36 @?= "A"+ , testCase "61 -> \"Z\"" $ Internal.iToBase62 61 @?= "Z"+ , testCase "62 -> \"10\"" $ Internal.iToBase62 62 @?= "10"+ , testCase "negative values produce valid output" $+ let result = Internal.iToBase62 (-42)+ in assertBool "should produce non-empty string" (not (null result))+ , testCase "output chars in valid range" $+ let valid = ['0'..'9'] ++ ['a'..'z'] ++ ['A'..'Z']+ result = Internal.iToBase62 12345+ in assertBool "all chars should be base62" (all (`elem` valid) result)+ ]+ ]++--------------------------------------------------------------------------+-- 1b. Version compatibility++versionCompatTests :: TestTree+versionCompatTests = testGroup "versionsAreCompatible"+ [ testCase "same version is always compatible" $+ versionsAreCompatible (v 2) (v 2) @?= True+ , testCase "same version (v1)" $+ versionsAreCompatible (v 1) (v 1) @?= True+ , testCase "different versions incompatible" $+ versionsAreCompatible (v 1) (v 2) @?= False+ , testCase "different versions incompatible (reverse)" $+ versionsAreCompatible (v 2) (v 1) @?= False+ , testCase "openSemaphore: unversioned (v1) name" $ do+ result <- openSemaphore "no_prefix_compat"+#if defined(mingw32_HOST_OS)+ -- On Windows, semaphoreVersion = 1, so v1 names are compatible.+ -- openSemaphore tries to open (fails because semaphore doesn't exist).+ case result of+ Left (SemaphoreDoesNotExist _) -> return ()+ Left err -> assertFailure ("expected SemaphoreDoesNotExist, got " ++ show err)+ Right _ -> assertFailure "expected Left, got Right"+#else+ -- On POSIX, semaphoreVersion = 2, so v1 names are incompatible.+ case result of+ Left (SemaphoreIncompatibleVersion actual expected) -> do+ actual @?= v 1+ expected @?= semaphoreVersion+ Left err -> assertFailure ("expected SemaphoreIncompatibleVersion, got " ++ show err)+ Right _ -> assertFailure "expected Left, got Right"+#endif+ , testCase "openSemaphore: wrong version prefix" $ do+ result <- openSemaphore "v99-wrong_ver_compat"+ case result of+ Left (SemaphoreIncompatibleVersion actual expected) -> do+ actual @?= v 99+ expected @?= semaphoreVersion+ Left err -> assertFailure ("expected SemaphoreIncompatibleVersion, got " ++ show err)+ Right _ -> assertFailure "expected Left, got Right"+ ]++--------------------------------------------------------------------------+-- 2. createSemaphore++createSemaphoreTests :: TestTree+createSemaphoreTests = testGroup "createSemaphore"+ [ testCase "zero tokens" $ withSemaphore "create_zero" 0 $ \srv -> do+ v <- getSemaphoreValue srv+ v @?= 0+ , testCase "name structure" $ withSemaphore "create_struct" 1 $ \srv -> do+ let nameStr = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ ver = getSemaphoreProtocolVersion semaphoreVersion+ if ver > 1+ then assertBool "should have version prefix" (("v" ++ show ver ++ "-") `isPrefixOf` nameStr)+ else assertBool "should be bare name" (head nameStr /= 'v' || '-' `notElem` take 4 nameStr)+ ]++--------------------------------------------------------------------------+-- 3. freshSemaphore++freshSemaphoreTests :: TestTree+freshSemaphoreTests = testGroup "freshSemaphore"+ [ testCase "two calls produce different names" $ do+ s1 <- freshSemaphore "fresh_a" 1 >>= unwrapServer+ s2 <- freshSemaphore "fresh_b" 1 >>= unwrapServer+ let n1 = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore s1))+ n2 = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore s2))+ assertBool "names should differ" (n1 /= n2)+ destroyServerSemaphore s1+ destroyServerSemaphore s2+ , testCase "name has version prefix" $ withFreshSemaphore 1 $ \srv -> do+ let nameStr = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ ver = getSemaphoreProtocolVersion semaphoreVersion+ if ver > 1+ then assertBool "should have version prefix" (("v" ++ show ver ++ "-") `isPrefixOf` nameStr)+ else assertBool "should be bare name" (head nameStr /= 'v' || '-' `notElem` take 4 nameStr)+ ]++--------------------------------------------------------------------------+-- 4. openSemaphore++openSemaphoreTests :: TestTree+openSemaphoreTests = testGroup "openSemaphore"+ [ testCase "opens existing semaphore and shares state" $+ withSemaphore "open_share" 3 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ child <- openSemaphore (name) >>= unwrapSem+ -- child acquires a token; parent sees the change+ tok <- waitOnSemaphore child+ v <- getSemaphoreValue srv+ v @?= 2+ releaseSemaphoreToken tok+ destroyClientSemaphore child+ ]++--------------------------------------------------------------------------+-- 5. waitOnSemaphore++waitOnSemaphoreTests :: TestTree+waitOnSemaphoreTests = testGroup "waitOnSemaphore"+ [ testCase "basic: acquires token, value decrements" $+ withSemaphore "wait_basic" 3 $ \srv -> do+ let sem = serverClientSemaphore srv+ tok <- waitOnSemaphore sem+ v <- getSemaphoreValue srv+ v @?= 2+ releaseSemaphoreToken tok+ , testCase "blocks when exhausted, completes after release" $+ withSemaphore "wait_block" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ name = semaphoreIdentifier (clientSemaphoreName sem)+ tok <- waitOnSemaphore sem -- take the only token+ started <- newEmptyMVar+ done <- newEmptyMVar+ _ <- forkIO $ do+ child <- openSemaphore (name) >>= unwrapSem+ childTok <- waitOnSemaphore child+ releaseSemaphoreToken childTok+ destroyClientSemaphore child+ putMVar done ()+ putMVar started ()+ takeMVar started+ threadDelay 100000 -- let thread reach the wait+ assertBlocks "wait should block when no tokens" (takeMVar done)+ releaseSemaphoreToken tok+ assertCompletes "wait should complete after release" 2000000 (takeMVar done)+ , testCase "multiple waiters all served" $+ withSemaphore "wait_multi" 3 $ \srv -> do+ let sem = serverClientSemaphore srv+ name = semaphoreIdentifier (clientSemaphoreName sem)+ numWaiters = 3+ -- Acquire all 3 tokens first, then fork waiters, then release.+ toks <- replicateM numWaiters (waitOnSemaphore sem)+ dones <- replicateM numWaiters newEmptyMVar+ forM_ dones $ \done -> forkIO $ do+ child <- openSemaphore (name) >>= unwrapSem+ childTok <- waitOnSemaphore child+ releaseSemaphoreToken childTok+ destroyClientSemaphore child+ putMVar done ()+ threadDelay 300000 -- let all waiters block+ mapM_ releaseSemaphoreToken toks+ forM_ dones $ \done ->+ assertCompletes "waiter should complete" 2000000 (takeMVar done)+ ]++--------------------------------------------------------------------------+-- 6. tryWaitOnSemaphore++tryWaitOnSemaphoreTests :: TestTree+tryWaitOnSemaphoreTests = testGroup "tryWaitOnSemaphore"+ [ testCase "returns Just when tokens available" $+ withSemaphore "try_true" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ r <- tryWaitOnSemaphore sem+ case r of+ Just tok -> releaseSemaphoreToken tok+ Nothing -> assertFailure "expected Just, got Nothing"+ , testCase "non-blocking on empty semaphore" $+ withSemaphore "try_nonblock" 0 $ \srv -> do+ r <- assertCompletes "tryWait should not block" 1000000+ (tryWaitOnSemaphore (serverClientSemaphore srv))+ case r of+ Nothing -> return ()+ Just _ -> assertFailure "expected Nothing, got Just"+ , testCase "successive calls deplete tokens" $+ withSemaphore "try_deplete" 3 $ \srv -> do+ let sem = serverClientSemaphore srv+ r1 <- tryWaitOnSemaphore sem+ r2 <- tryWaitOnSemaphore sem+ r3 <- tryWaitOnSemaphore sem+ r4 <- tryWaitOnSemaphore sem+ -- r1..r3 should be Just, r4 should be Nothing+ case r1 of+ Just _ -> return ()+ Nothing -> assertFailure "r1: expected Just"+ case r2 of+ Just _ -> return ()+ Nothing -> assertFailure "r2: expected Just"+ case r3 of+ Just _ -> return ()+ Nothing -> assertFailure "r3: expected Just"+ case r4 of+ Nothing -> return ()+ Just _ -> assertFailure "r4: expected Nothing"+ -- Release acquired tokens+ mapM_ (\r -> case r of Just tok -> releaseSemaphoreToken tok; Nothing -> return ()) [r1, r2, r3]+ ]++--------------------------------------------------------------------------+-- 7. releaseSemaphoreToken++releaseSemaphoreTests :: TestTree+releaseSemaphoreTests = testGroup "releaseSemaphoreToken"+ [ testCase "increments value" $ withSemaphore "rel_inc" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ tok <- waitOnSemaphore sem+ releaseSemaphoreToken tok+ v <- getSemaphoreValue srv+ v @?= 1+ , testCase "release N tokens adds N tokens" $ withSemaphore "rel_n" 3 $ \srv -> do+ let sem = serverClientSemaphore srv+ toks <- replicateM 3 (waitOnSemaphore sem)+ mapM_ releaseSemaphoreToken toks+ v <- getSemaphoreValue srv+ v @?= 3+ ]++--------------------------------------------------------------------------+-- 8. getSemaphoreValue++getSemaphoreValueTests :: TestTree+getSemaphoreValueTests = testGroup "getSemaphoreValue"+ [ testCase "reports initial count" $ withSemaphore "getval_init" 5 $ \srv -> do+ v <- getSemaphoreValue srv+ v @?= 5+ ]++--------------------------------------------------------------------------+-- 9. destroySemaphore++destroySemaphoreTests :: TestTree+destroySemaphoreTests = testGroup "destroySemaphore"+ [ testCase "destroyed semaphore cannot be used" $ do+ srv <- freshSemaphore "destroy_open" 1 >>= unwrapServer+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ destroyServerSemaphore srv+ -- openSemaphore must fail on a destroyed semaphore.+ result <- openSemaphore name+ case result of+ Left (SemaphoreDoesNotExist _) -> return ()+ Left err -> assertFailure ("unexpected error from openSemaphore: " ++ show err)+ Right _ -> assertFailure "expected openSemaphore to fail on destroyed semaphore"+ , expectCrashRecoveryFail $+ testCase "client disconnect returns held tokens" $+ withSemaphore "destroy_client" 3 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ child <- openSemaphore (name) >>= unwrapSem+ tok1 <- waitOnSemaphore child+ tok2 <- waitOnSemaphore child+ v1 <- getSemaphoreValue srv+ v1 @?= 1+ -- Simulate crash: close the token fds without sending release.+ crashToken tok1+ crashToken tok2+ destroyClientSemaphore child+ awaitPoolValue srv 3+ ]++--------------------------------------------------------------------------+-- 10. SemaphoreError (failure cases)++semaphoreErrorTests :: TestTree+semaphoreErrorTests = testGroup "SemaphoreError"+ [ testCase "openSemaphore: wrong version returns error" $ do+ result <- openSemaphore "v1-err_wrongver"+ case result of+#if defined(mingw32_HOST_OS)+ -- On Windows, all versions are compatible, so it tries to open+ Left (SemaphoreDoesNotExist _) -> return ()+#else+ Left (SemaphoreIncompatibleVersion actual expected) -> do+ actual @?= v 1+ expected @?= semaphoreVersion+#endif+ Left err -> assertFailure ("unexpected error: " ++ show err)+ Right _ -> assertFailure "expected Left, got Right"+ , testCase "openSemaphore: missing prefix returns version error on POSIX" $ do+ result <- openSemaphore "no_prefix_here"+ case result of+#if defined(mingw32_HOST_OS)+ -- On Windows, v1 is compatible, so it tries to open and fails+ Left (SemaphoreDoesNotExist _) -> return ()+#else+ Left (SemaphoreIncompatibleVersion actual expected) -> do+ actual @?= v 1+ expected @?= semaphoreVersion+#endif+ Left err -> assertFailure ("unexpected error: " ++ show err)+ Right _ -> assertFailure "expected Left, got Right"+ , testCase "openSemaphore: nonexistent semaphore fails" $ do+ let ident = semaphoreIdentifier (SemaphoreName semaphoreVersion "err_noexist")+ result <- openSemaphore ident+ case result of+ Left (SemaphoreDoesNotExist _) -> return ()+ Left err -> assertFailure ("unexpected error: " ++ show err)+ Right _ -> assertFailure "expected openSemaphore to fail on nonexistent semaphore"+ , testCase "openSemaphore: empty string returns version error on POSIX" $ do+ result <- openSemaphore ""+ case result of+#if defined(mingw32_HOST_OS)+ -- On Windows, v1 is compatible, so it tries to open the empty name.+ -- Win32 OpenSemaphore() rejects the empty name with ERROR_INVALID_HANDLE+ -- (not ERROR_FILE_NOT_FOUND), so we get SemaphoreOtherError here.+ Left (SemaphoreOtherError _) -> return ()+#else+ Left (SemaphoreIncompatibleVersion actual expected) -> do+ actual @?= v 1+ expected @?= semaphoreVersion+#endif+ Left err -> assertFailure ("unexpected error: " ++ show err)+ Right _ -> assertFailure "expected Left, got Right"+ ]++--------------------------------------------------------------------------+-- 11. Interruptible waitOnSemaphore++interruptibleWaitTests :: TestTree+interruptibleWaitTests = testGroup "interruptible waitOnSemaphore"+ [ testCase "succeeds when token available" $+ withSemaphore "int_ok" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ resultVar <- newEmptyMVar+ _ <- forkIO $ do+ r <- try (waitOnSemaphore sem) :: IO (Either SomeException SemaphoreToken)+ putMVar resultVar r+ result <- assertCompletes "should get token" 2000000+ (takeMVar resultVar)+ case result of+ Right tok -> releaseSemaphoreToken tok+ Left e -> assertFailure ("expected Right token, got Left " ++ show e)+ , testCase "can be cancelled on 0-token semaphore" $+ withSemaphore "int_cancel" 0 $ \srv -> do+ let sem = serverClientSemaphore srv+ resultVar <- newEmptyMVar+ tid <- forkIO $ do+ r <- try (waitOnSemaphore sem) :: IO (Either SomeException SemaphoreToken)+ putMVar resultVar r+ threadDelay 100000 -- let it block+ assertCompletes "killThread should not hang" 2000000+ (killThread tid)+ result <- assertCompletes "should get result" 2000000+ (takeMVar resultVar)+ case result of+ Left _ -> return ()+ Right _ -> assertFailure "expected Left (interrupted), got Right"+ , testCase "cancelled wait does not leak tokens" $+ withSemaphore "int_noleak" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ -- take the only token+ tok <- waitOnSemaphore sem+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ -- fork a wait that will block (no tokens available)+ tid <- forkIO $ void (waitOnSemaphore sem)+ threadDelay 100000+ -- cancel the wait+ assertCompletes "killThread should not hang" 2000000+ (killThread tid)+ -- release our token+ releaseSemaphoreToken tok+ -- The server should detect the cancelled wait's disconnect+ -- and return any transiently acquired token.+ tok' <- assertCompletes "token should be available" 5000000 $+ waitOnSemaphore sem+ v1 <- getSemaphoreValue srv+ v1 @?= 0+ releaseSemaphoreToken tok'+ ]++--------------------------------------------------------------------------+-- 12. AbstractSem++abstractSemTests :: TestTree+abstractSemTests = testGroup "AbstractSem"+ [ testCase "withAbstractSem acquires and releases" $+ withSemaphore "absem_basic" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ tokRef <- newIORef (undefined :: SemaphoreToken)+ let absSem = AbstractSem+ (waitOnSemaphore sem >>= writeIORef tokRef)+ (readIORef tokRef >>= releaseSemaphoreToken)+ v0 <- getSemaphoreValue srv+ v0 @?= 1+ withAbstractSem absSem $ do+ v1 <- getSemaphoreValue srv+ v1 @?= 0+ v2 <- getSemaphoreValue srv+ v2 @?= 1+ , testCase "releases on exception (bracket safety)" $+ withSemaphore "absem_exc" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ tokRef <- newIORef (undefined :: SemaphoreToken)+ let absSem = AbstractSem+ (waitOnSemaphore sem >>= writeIORef tokRef)+ (readIORef tokRef >>= releaseSemaphoreToken)+ result <- try $ withAbstractSem absSem $+ error "boom"+ case result of+ Left (_ :: SomeException) -> return ()+ Right (_ :: ()) -> assertFailure "expected exception"+ v <- getSemaphoreValue srv+ v @?= 1+ ]++--------------------------------------------------------------------------+-- 12. Multi-client (freshSemaphore, concurrency limiting)++multiClientTests :: TestTree+multiClientTests = testGroup "multi-client"+ [ testCase "freshSemaphore: children open by name and respect limit" $+ withFreshSemaphore 3 $ \srv -> do+ let nameStr = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ ref <- newIORef (0 :: Int, 0 :: Int)+ let numChildren = 6+ dones <- replicateM numChildren newEmptyMVar+ forM_ (zip [1::Int ..] dones) $ \(_n, done) -> forkIO $ do+ child <- openSemaphore (nameStr) >>= unwrapSem+ tok <- waitOnSemaphore child+ atomicModifyIORef' ref+ (\(cur, peak) -> let cur' = cur + 1 in ((cur', max peak cur'), ()))+ threadDelay 80000 -- 80 ms of work+ atomicModifyIORef' ref (\(cur, peak) -> ((cur - 1, peak), ()))+ releaseSemaphoreToken tok+ destroyClientSemaphore child+ putMVar done ()+ mapM_ takeMVar dones+ (_, peak) <- readIORef ref+ assertBool ("peak " ++ show peak ++ " should be <= 3") (peak <= 3)+ val <- getSemaphoreValue srv+ val @?= 3+ ]++--------------------------------------------------------------------------+-- 14. Shared-handle (multithreaded use of a single Semaphore value)++sharedHandleTests :: TestTree+sharedHandleTests = testGroup "shared handle"+ [ testCase "concurrent blocking wait + tryWait on same handle" $+ withFreshSemaphore 1 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ forM_ [1..50 :: Int] $ \_ -> do+ -- Use a helper connection to drain the token so pool = 0.+ helper <- openSemaphore name >>= unwrapSem+ helperTok <- waitOnSemaphore helper+ -- Shared client identity used by two threads concurrently.+ client <- openSemaphore name >>= unwrapSem+ -- Thread 1: blocking wait (opens fresh connection, blocks)+ t1done <- newEmptyMVar+ _ <- forkIO $ do+ r <- try (waitOnSemaphore client) :: IO (Either SomeException SemaphoreToken)+ putMVar t1done r+ threadDelay 50000 -- let thread 1 block+ -- Main thread: tryWait on same identity (opens fresh connection).+ t2done <- newEmptyMVar+ _ <- forkIO $ do+ r <- try (tryWaitOnSemaphore client) :: IO (Either SomeException (Maybe SemaphoreToken))+ putMVar t2done r+ r2 <- timeout 5000000 (takeMVar t2done)+ releaseSemaphoreToken helperTok+ r1 <- timeout 5000000 (takeMVar t1done)+ -- Check that thread 1 succeeded.+ case r1 of+ Just (Right tok) -> releaseSemaphoreToken tok+ Just (Left e) -> assertFailure ("waitOnSemaphore failed: " ++ show e)+ Nothing -> assertFailure "thread 1 deadlocked"+ -- Check that tryWait returned Nothing (no tokens available at time of try).+ case r2 of+ Just (Right Nothing) -> return ()+ Just (Right (Just tok)) -> releaseSemaphoreToken tok -- got lucky, that's ok too+ Just (Left e) -> assertFailure ("tryWait exception: " ++ show e)+ Nothing -> assertFailure "tryWait deadlocked"+ destroyClientSemaphore client+ destroyClientSemaphore helper+ , testCase "release from one connection unblocks wait on another" $+ withFreshSemaphore 1 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ -- Connection A acquires the only token.+ connA <- openSemaphore name >>= unwrapSem+ tokA <- waitOnSemaphore connA+ -- Connection B waits (will block -- pool is 0).+ connB <- openSemaphore name >>= unwrapSem+ done <- newEmptyMVar+ _ <- forkIO $ do+ tokB <- waitOnSemaphore connB+ putMVar done tokB+ threadDelay 100000 -- let thread reach the wait+ -- Connection A releases; connection B should be unblocked.+ releaseSemaphoreToken tokA+ tokB <- assertCompletes "connB should complete after release" 5000000+ (takeMVar done)+ -- connB now holds the token; release it.+ releaseSemaphoreToken tokB+ v <- getSemaphoreValue srv+ v @?= 1+ destroyClientSemaphore connA+ destroyClientSemaphore connB+ , testCase "Cabal-like worker pattern: wait then fork release" $+ withFreshSemaphore 2 $ \srv -> do+ let sem = serverClientSemaphore srv+ numJobs = 6+ dones <- replicateM numJobs newEmptyMVar+ -- Spawn workers: each acquires a token, forks work that releases it.+ forM_ (zip [1::Int ..] dones) $ \(_n, done) -> forkIO $ do+ tok <- waitOnSemaphore sem+ _ <- forkIO $ do+ threadDelay 50000 -- simulate work+ releaseSemaphoreToken tok+ putMVar done ()+ return ()+ forM_ dones $ \done ->+ assertCompletes "job should complete" 10000000 (takeMVar done)+ v <- getSemaphoreValue srv+ v @?= 2+ , expectCrashRecoveryFail $+ testCase "disconnect without release returns tokens, waiters on other connections unblocked" $+ withFreshSemaphore 2 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ -- Client A acquires both tokens (each waitOnSemaphore opens a fresh connection)+ clientA <- openSemaphore (name) >>= unwrapSem+ tokA1 <- waitOnSemaphore clientA+ tokA2 <- waitOnSemaphore clientA+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ -- Client B waits on a separate connection (will block)+ clientB <- openSemaphore (name) >>= unwrapSem+ done <- newEmptyMVar+ _ <- forkIO $ do+ tokB <- waitOnSemaphore clientB+ putMVar done tokB+ threadDelay 200000 -- let B block+ -- Simulate crash: close the token fds without sending release.+ crashToken tokA1+ crashToken tokA2+ destroyClientSemaphore clientA+ -- Client B should be unblocked by the returned tokens.+ tokB <- assertCompletes "client B should complete after A crashes" 5000000+ (takeMVar done)+ releaseSemaphoreToken tokB+ destroyClientSemaphore clientB+ ]++--------------------------------------------------------------------------+-- 15. Robustness (crash recovery, async exceptions, masking)++robustnessTests :: TestTree+robustnessTests = testGroup "robustness"+ [ testGroup "crash recovery"+ [ expectCrashRecoveryFail $+ testCase "multiple clients crash, all tokens recovered" $+ withSemaphore "robust_multi_crash" 5 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ -- Each child opens, acquires a token, and crashes (fd closed+ -- without sending release).+ replicateM_ 5 $ do+ child <- openSemaphore (name) >>= unwrapSem+ tok <- waitOnSemaphore child+ crashToken tok+ destroyClientSemaphore child+ awaitPoolValue srv 5+ , expectCrashRecoveryFail $+ testCase "client holding many tokens crashes" $+ withSemaphore "robust_many_toks" 10 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ child <- openSemaphore (name) >>= unwrapSem+ toks <- replicateM 10 (waitOnSemaphore child)+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ mapM_ crashToken toks+ destroyClientSemaphore child+ awaitPoolValue srv 10+ ]+ , testGroup "async exceptions"+ [ expectCrashRecoveryFail $+ testCase "throwTo kills thread holding token, no leak" $+ withFreshSemaphore 1 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ child <- openSemaphore (name) >>= unwrapSem+ tidVar <- newEmptyMVar+ tokRef <- newIORef Nothing+ _ <- forkIO $ do+ tid <- myThreadId+ putMVar tidVar tid+ tok <- waitOnSemaphore child -- acquires the token+ writeIORef tokRef (Just tok)+ threadDelay 10000000 -- long sleep; throwTo lands here (unmasked)+ tid <- takeMVar tidVar+ threadDelay 200000 -- let thread acquire and enter sleep+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ throwTo tid ThreadKilled+ -- Thread is dead; simulate crash by closing the token fd.+ mbTok <- readIORef tokRef+ case mbTok of+ Just tok -> crashToken tok+ Nothing -> return ()+ destroyClientSemaphore child+ awaitPoolValue srv 1+#if !defined(mingw32_HOST_OS)+ -- On Win32, crashToken is a no-op (no crash recovery), so when+ -- throwTo lands between mask_ blocks, the unreleased tokens are+ -- permanently lost. This makes the test flaky — skip it.+ , testCase "throwTo during releaseSemaphoreToken (masked)" $+ withSemaphore "robust_throwto_rel" 5 $ \srv -> do+ let sem = serverClientSemaphore srv+ toks <- replicateM 5 (waitOnSemaphore sem)+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ -- Track which tokens have been released.+ remainingRef <- newIORef toks+ tidVar <- newEmptyMVar+ _ <- forkIO $ do+ tid <- myThreadId+ putMVar tidVar tid+ let releaseAll [] = return ()+ releaseAll (t:ts) = do+ -- mask_ ensures the IORef update is atomic with the+ -- release, so remainingRef always accurately reflects+ -- which tokens are still open (prevents double-close+ -- of recycled fd numbers).+ mask_ $ do+ releaseSemaphoreToken t+ writeIORef remainingRef ts+ releaseAll ts+ releaseAll toks+ tid <- takeMVar tidVar+ throwTo tid ThreadKilled+ -- Crash-close any tokens the thread didn't get to release.+ remaining <- readIORef remainingRef+ mapM_ crashToken remaining+ -- Either mask_ protected the release, or crash recovery+ -- returns unreleased tokens; either way all 5 are restored.+ awaitPoolValue srv 5+#endif+ ]+ , testGroup "interruptible wait edge cases"+ [ testCase "double interrupt is safe" $+ withSemaphore "robust_dbl_int" 0 $ \srv -> do+ let sem = serverClientSemaphore srv+ doneVar <- newEmptyMVar+ tid <- forkIO $ do+ void (try (waitOnSemaphore sem) :: IO (Either SomeException SemaphoreToken))+ putMVar doneVar ()+ threadDelay 100000+ killThread tid+ void $ assertCompletes "thread should be dead" 2000000+ (takeMVar doneVar)+ -- second killThread: thread is dead, should not hang+ _ <- assertCompletes "second killThread should not hang" 2000000+ (try (killThread tid) :: IO (Either SomeException ()))+ return ()+ , testCase "killThread after natural completion" $+ withSemaphore "robust_int_done" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ resultVar <- newEmptyMVar+ tid <- forkIO $ do+ r <- try (waitOnSemaphore sem) :: IO (Either SomeException SemaphoreToken)+ putMVar resultVar r+ result <- assertCompletes "should complete" 2000000+ (takeMVar resultVar)+ case result of+ Right tok -> releaseSemaphoreToken tok+ Left e -> assertFailure ("expected Right token, got Left " ++ show e)+ threadDelay 100000+ -- killThread after completion: should not crash+ _ <- try (killThread tid) :: IO (Either SomeException ())+ return ()+ , testCase "many cancelled waits don't leak" $+ withSemaphore "robust_cancel_many" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ -- Acquire the token so waits will block.+ tok <- waitOnSemaphore sem+ tids <- replicateM 10 $+ forkIO $ void (waitOnSemaphore sem)+ threadDelay 200000+ forM_ tids $ \tid ->+ void (try (killThread tid) :: IO (Either SomeException ()))+ -- Wait for all cancelled waits to be cleaned up, then release.+ awaitPoolValue srv 0+ releaseSemaphoreToken tok+ awaitPoolValue srv 1+ , testCase "cancel-vs-complete race: no token leak" $+ withFreshSemaphore 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ forM_ [1..10 :: Int] $ \_ -> do+ conn <- openSemaphore name >>= unwrapSem+ connTok <- waitOnSemaphore conn+ resultVar <- newEmptyMVar+ tid <- forkIO $ mask $ \unmask -> do+ r <- try (unmask $ waitOnSemaphore conn) :: IO (Either SomeException SemaphoreToken)+ putMVar resultVar r+ threadDelay 50000 -- let thread block+ -- Race: release and cancel concurrently+ releaseDone <- newEmptyMVar+ _ <- forkIO $ releaseSemaphoreToken connTok >> putMVar releaseDone ()+ _ <- try (killThread tid) :: IO (Either SomeException ())+ result <- assertCompletes "should get result" 2000000+ (takeMVar resultVar)+ -- If the thread got a token, release it+ case result of+ Right tok -> releaseSemaphoreToken tok+ Left _ -> return ()+ -- Wait for the forked release to complete before destroying+ -- the connection. Without this, destroyClientSemaphore can+ -- close the semaphore handle while releaseSemaphore is still+ -- running on another thread, losing the token.+ takeMVar releaseDone+ awaitPoolValue srv 1+ destroyClientSemaphore conn+ ]+ , testGroup "concurrent stress"+ [ testCase "rapid create-destroy cycle" $+ forM_ [1..50 :: Int] $ \_ -> do+ srv <- freshSemaphore "robust_rapid" 1 >>= unwrapServer+ destroyServerSemaphore srv+ , testCase "destroyServerSemaphore is idempotent" $ do+ srv <- freshSemaphore "robust_idempotent_destroy" 1 >>= unwrapServer+ destroyServerSemaphore srv+ -- Second call is a no-op.+ assertCompletes "second destroyServerSemaphore should not hang" 2000000+ (destroyServerSemaphore srv)+ ]+ ]++--------------------------------------------------------------------------+-- 15. Subprocess robustness (real process death via helper executable)++subprocessTests :: TestTree+subprocessTests = testGroup "subprocess robustness"+ [ testGroup "crash recovery"+ [ expectCrashRecoveryFail $+ testCase "multiple processes crash, all tokens recovered" $+ withSemaphore "sub_multi_crash" 5 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ -- Spawn 5 helpers, each opens and acquires one token+ replicateM_ 5 $ withHelper $ \h -> do+ sendCommand_ h (CmdOpen name)+ sendCommand_ h CmdWait+ sendCommandNoResponse h CmdCrash+ _ <- try (waitForProcess (hProcess h)) :: IO (Either SomeException ExitCode)+ return ()+ -- Wait for server to detect disconnects+ threadDelay 1000000+ v1 <- getSemaphoreValue srv+ v1 @?= 5+ , expectCrashRecoveryFail $+ testCase "process holding many tokens crashes" $+ withSemaphore "sub_many_toks" 10 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ withHelper $ \h -> do+ sendCommand_ h (CmdOpen name)+ sendCommand_ h (CmdWaitN 10)+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ sendCommandNoResponse h CmdCrash+ _ <- try (waitForProcess (hProcess h)) :: IO (Either SomeException ExitCode)+ return ()+ threadDelay 1000000+ v1 <- getSemaphoreValue srv+ v1 @?= 10+ ]+ , testGroup "process kill"+ [ expectCrashRecoveryFail $+ testCase "process killed while holding token" $+ withSemaphore "sub_kill_hold" 1 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ withHelper $ \h -> do+ sendCommand_ h (CmdOpen name)+ sendCommand_ h CmdWait+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ killHelper h+ threadDelay 1000000+ v1 <- getSemaphoreValue srv+ v1 @?= 1+ , testCase "process killed during blocked wait, no leak" $+ withSemaphore "sub_kill_block" 0 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ withHelper $ \h -> do+ sendCommand_ h (CmdOpen name)+ -- Send wait but don't read response (it will block)+ sendCommandNoResponse h CmdWait+ threadDelay 200000 -- let helper block+ killHelper h+ threadDelay 1000000+ v <- getSemaphoreValue srv+ v @?= 0+ ]+ , testGroup "multi-process edge cases"+ [ expectCrashRecoveryFail $+ testCase "kill-vs-complete race" $+ withSemaphore "sub_kill_race" 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ name = semaphoreIdentifier (clientSemaphoreName sem)+ -- Acquire the token so the helper's wait will block.+ tok <- waitOnSemaphore sem+ withHelper $ \h -> do+ sendCommand_ h (CmdOpen name)+ sendCommandNoResponse h CmdWait+ threadDelay 200000+ -- Release a token so the helper can acquire it, then+ -- wait until the helper has acquired before killing.+ releaseSemaphoreToken tok+ -- Spin until the helper has consumed the token.+ let waitUntilAcquired = do+ v <- getSemaphoreValue srv+ if v == 0 then return ()+ else threadDelay 10000 >> waitUntilAcquired+ assertCompletes "helper should acquire" 5000000 waitUntilAcquired+ killHelper h+ threadDelay 1000000+ v <- getSemaphoreValue srv+ -- The helper acquired the token and was killed.+ -- On Posix, crash recovery returns the token: pool = 1.+ -- On Win32, the token is lost: pool = 0.+ v @?= 1+ ]+ , testGroup "concurrent waiter crash interaction"+ [ expectCrashRecoveryFail $+ testCase "process crash returns tokens, unblocking waiting process" $+ withSemaphore "sub_crash_unblock" 2 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ -- Use withHelper to ensure cleanup on test failure (expected on Windows)+ withHelper $ \hA -> withHelper $ \hB -> do+ -- Process A acquires both tokens+ sendCommand_ hA (CmdOpen name)+ sendCommand_ hA (CmdWaitN 2)+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ -- Process B sends a blocking wait+ sendCommand_ hB (CmdOpen name)+ sendCommandNoResponse hB CmdWait+ threadDelay 200000 -- let B's wait reach the server+ -- Crash process A: tokens returned to pool, B should unblock+ sendCommandNoResponse hA CmdCrash+ _ <- try (waitForProcess (hProcess hA)) :: IO (Either SomeException ExitCode)+ -- Wait until B has acquired the token (pool goes from 2 to 1)+ let waitUntilAcquired = do+ v <- getSemaphoreValue srv+ if v == 1 then return ()+ else threadDelay 10000 >> waitUntilAcquired+ assertCompletes "process B should acquire after A crashes" 5000000+ waitUntilAcquired+ killHelper hB+ threadDelay 1000000+ v <- getSemaphoreValue srv+ v @?= 2+ , testCase "process killed with pending wait, no token leak" $+ withSemaphore "sub_kill_pending" 0 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ -- Process sends a wait that will block (0 tokens)+ withHelper $ \hA -> do+ sendCommand_ hA (CmdOpen name)+ sendCommandNoResponse hA CmdWait+ threadDelay 200000 -- let wait reach the server+ -- Helper is blocked in waitOnSemaphore so it can't read+ -- CmdCrash from stdin; use killHelper (SIGKILL) instead.+ killHelper hA+ threadDelay 1000000+ -- No tokens were held, pending wait discarded, pool still 0+ v <- getSemaphoreValue srv+ v @?= 0+ , expectCrashRecoveryFail $+ testCase "process killed with held tokens and pending wait" $+ withSemaphore "sub_kill_held_pending" 1 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ -- Process acquires the token, then sends another wait (will block)+ withHelper $ \hA -> do+ sendCommand_ hA (CmdOpen name)+ sendCommand_ hA CmdWait -- acquires (pool=0)+ sendCommandNoResponse hA CmdWait -- pending (pool=0)+ threadDelay 200000 -- let pending wait register+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ -- Helper is blocked in waitOnSemaphore; use killHelper.+ -- 1 held token returned, pending wait discarded.+ killHelper hA+ threadDelay 1000000+ v <- getSemaphoreValue srv+ v @?= 1+ ]+ , testGroup "cross-process concurrency"+ [ testCase "cross-process concurrency limiting" $+ withSemaphore "sub_conc_limit" 4 $ \srv -> do+ let name = semaphoreIdentifier (clientSemaphoreName (serverClientSemaphore srv))+ dones <- replicateM 10 $ do+ done <- newEmptyMVar+ _ <- forkIO $ withHelper $ \h -> do+ sendCommand_ h (CmdOpen name)+ sendCommand_ h CmdWait+ sendCommand_ h (CmdRelease 1)+ sendCommand_ h CmdDestroy+ putMVar done ()+ return done+ forM_ dones $ \done ->+ assertCompletes "helper should complete" 10000000 (takeMVar done)+ v <- getSemaphoreValue srv+ v @?= 4+ ]+ ]++--------------------------------------------------------------------------+-- Regression tests.++regressionTests :: TestTree+regressionTests = testGroup "regression"+ [+#if !defined(mingw32_HOST_OS)+ -- Path length check: paths longer than sun_path must be rejected,+ -- not silently truncated.+ testCase "path too long is rejected" $ do+ let longName = replicate 200 'x'+ result <- createSemaphore longName 1+ case result of+ Left _ -> return ()+ Right s -> do destroyServerSemaphore s+ assertFailure "expected failure for path exceeding sun_path"++ -- Release guard: a CmdRelease on a connection that hasn't acquired+ -- any tokens must not increment the pool past its initial value.+ , testCase "release without acquire does not corrupt pool" $+ withFreshSemaphore 2 $ \srv -> do+ let path = Internal.getSemaphoreSocketPath+ (clientSemaphoreName $ serverClientSemaphore srv)+ socketPath <- path+ -- Open a raw connection and send CmdRelease without CmdWait.+ fd <- connectDomainSocket socketPath+ fdWriteByte fd 0x2B -- '+'+ resp <- fdReadByte fd+ closeFd fd+ -- Server should reject: response is '!' (RspFail)+ resp @?= 0x21+ -- Pool must still be 2, not 3.+ v <- getSemaphoreValue srv+ v @?= 2++ -- destroyServerSemaphore must not hang when serve threads are+ -- blocked in read.+ , testCase "destroyServerSemaphore does not hang with active connections" $ do+ srv <- freshSemaphore "reg_hang" 1 >>= unwrapServer+ -- Acquire a token but do NOT release it. The serve thread is+ -- now blocked in fdReadByte waiting for the next command on+ -- this connection.+ _tok <- waitOnSemaphore (serverClientSemaphore srv)+ -- destroyServerSemaphore must kill this blocked serve thread.+ result <- timeout 5000000 $ destroyServerSemaphore srv+ case result of+ Just _ -> return ()+ Nothing -> assertFailure+ "destroyServerSemaphore hung (serve thread not interruptible)"++ -- Rapid create/acquire/destroy must not corrupt fds (e.g. orphaned+ -- serve threads closing recycled fds).+ , testCase "rapid create-acquire-destroy does not corrupt fds" $+ forM_ [1..100 :: Int] $ \_ -> do+ srv <- freshSemaphore "reg_rapid" 1 >>= unwrapServer+ tok <- waitOnSemaphore (serverClientSemaphore srv)+ releaseSemaphoreToken tok+ destroyServerSemaphore srv++ -- Crash after acquire must return the token to the pool.+ , testCase "crash after acquire returns token to pool" $+ withFreshSemaphore 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ let path = Internal.getSemaphoreSocketPath (clientSemaphoreName sem)+ socketPath <- path+ -- Open a raw connection and acquire one token.+ fd <- connectDomainSocket socketPath+ fdWriteByte fd 0x2D -- '-' = CmdWait+ resp <- fdReadByte fd -- should get '.' = RspOk+ resp @?= 0x2E+ -- Pool should now be 0.+ v0 <- getSemaphoreValue srv+ v0 @?= 0+ -- Crash: close the fd without sending release.+ closeFd fd+ -- Server should detect disconnect and return the token.+ awaitPoolValue srv 1++ -- Cancel a blocked wait then immediately acquire. Exercises the+ -- shutdown-based cancellation under rapid repeated use to ensure+ -- no fd corruption or token leak.+ , testCase "cancel wait then immediately acquire" $+ withFreshSemaphore 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ -- Hold the token so waits block.+ tok <- waitOnSemaphore sem+ forM_ [1..50 :: Int] $ \_ -> do+ tid <- forkIO $ void (waitOnSemaphore sem)+ threadDelay 10000+ killThread tid+ -- The original token should still work (its fd wasn't corrupted).+ releaseSemaphoreToken tok+ awaitPoolValue srv 1++ -- Token-per-connection: each waitOnSemaphore opens its own connection,+ -- so concurrent waits must not interfere.+ , testCase "concurrent waits get independent connections" $+ withFreshSemaphore 10 $ \srv -> do+ let sem = serverClientSemaphore srv+ doneVars <- replicateM 10 $ do+ done <- newEmptyMVar+ _ <- forkIO $ do+ tok <- waitOnSemaphore sem+ releaseSemaphoreToken tok+ putMVar done ()+ return done+ forM_ doneVars $ \done ->+ assertCompletes "concurrent wait should complete" 5000000+ (takeMVar done)+ awaitPoolValue srv 10++ -- A SemaphoreToken that becomes unreachable without being released+ -- must eventually return its token to the pool via the MVar finalizer.+ , testCase "GC finalizer returns dropped token to pool" $+ withFreshSemaphore 1 $ \srv -> do+ let sem = serverClientSemaphore srv+ -- Acquire a token in a scope where the binding is immediately lost.+ void $ waitOnSemaphore sem+ performMajorGC+ awaitPoolValue srv 1+#endif+ ]