diff --git a/Setup.hs b/Setup.hs --- a/Setup.hs +++ b/Setup.hs @@ -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 diff --git a/cbits/domain.c b/cbits/domain.c new file mode 100644 --- /dev/null +++ b/cbits/domain.c @@ -0,0 +1,161 @@ +#include +#include +#include +#include +#include +#include +#include +#include + +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 */ + } +} diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -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. diff --git a/helper/Main.hs b/helper/Main.hs new file mode 100644 --- /dev/null +++ b/helper/Main.hs @@ -0,0 +1,6 @@ +module Main where + +import System.Semaphore.TestHelper (helperMain) + +main :: IO () +main = helperMain diff --git a/readme.md b/readme.md --- a/readme.md +++ b/readme.md @@ -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`. diff --git a/semaphore-compat.cabal b/semaphore-compat.cabal --- a/semaphore-compat.cabal +++ b/semaphore-compat.cabal @@ -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 diff --git a/src/System/Semaphore.hs b/src/System/Semaphore.hs --- a/src/System/Semaphore.hs +++ b/src/System/Semaphore.hs @@ -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\-@. +-- 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) diff --git a/src/System/Semaphore/Internal/Common.hs b/src/System/Semaphore/Internal/Common.hs new file mode 100644 --- /dev/null +++ b/src/System/Semaphore/Internal/Common.hs @@ -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\-\\"@. +type SemaphoreIdentifier = String + +-- | Errors that can occur when creating or opening a semaphore. +data SemaphoreError + = SemaphoreAlreadyExists + { semaphoreErrorIdentifier :: !SemaphoreIdentifier + } + | SemaphoreDoesNotExist + { semaphoreErrorIdentifier :: !SemaphoreIdentifier + } + | SemaphoreIncompatibleVersion + { semaphoreErrorActualVersion :: !SemaphoreProtocolVersion + , semaphoreErrorExpectedVersion :: !SemaphoreProtocolVersion + } + | SemaphoreOtherError + { semaphoreOtherError :: !IOException + } + deriving (Show, Eq) + +instance Exception SemaphoreError + +--------------------------------------- +-- Version negotiation + +-- | The protocol version on this platform. +-- +-- The version tracks the IPC mechanism, not the library version: +-- +-- * __POSIX:__ 2 (domain sockets, replacing v1 system semaphores). +-- * __Windows:__ 1 (Win32 named semaphores, unchanged from v1). +-- * __Unsupported platforms:__ 0 (no compatible IPC backend). +-- +-- Because the version is 1 on Windows, 'semaphoreIdentifier' produces +-- a bare name (no @v\-@ prefix), matching the v1 format. +semaphoreVersion :: SemaphoreProtocolVersion +#if defined(mingw32_HOST_OS) +semaphoreVersion = SemaphoreProtocolVersion 1 +#elif defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) +semaphoreVersion = SemaphoreProtocolVersion 0 +#else +semaphoreVersion = SemaphoreProtocolVersion 2 +#endif + + +-- | The serialised identifier of a 'SemaphoreName' for transport +-- between processes. +-- +-- For version 1 this is the bare name; for version @N@ (@N >= 2@) the +-- name is prefixed with @v\-@. +semaphoreIdentifier :: SemaphoreName -> SemaphoreIdentifier +semaphoreIdentifier sn + | ver <= 1 = unversionedSemaphoreNameString sn + | otherwise = "v" ++ show ver ++ "-" ++ unversionedSemaphoreNameString sn + where + ver = getSemaphoreProtocolVersion (semaphoreProtocolVersion sn) + +-- | Parse a 'SemaphoreIdentifier' into a 'SemaphoreName'. +-- +-- Returns @Nothing@ for unversioned strings (which should be treated as +-- v1 by the caller's compatibility logic). +parseSemaphoreIdentifier :: SemaphoreIdentifier -> Maybe SemaphoreName +parseSemaphoreIdentifier ('v':rest) = + case reads rest of + [(n, '-':name)] -> Just (SemaphoreName (SemaphoreProtocolVersion n) name) + _ -> Nothing +parseSemaphoreIdentifier _ = Nothing + +--------------------------------------- +-- Utilities + +-- | Convert an 'Int' to a base-62 string (digits @0@–@9@, @a@–@z@, @A@–@Z@). +iToBase62 :: Int -> String +iToBase62 m = go m' "" + where + m' + | m == minBound = maxBound + | otherwise = abs m + go n cs | n < 62 + = let !c = chooseChar62 n + in c : cs + | otherwise + = let !(!q, r) = quotRem n 62 + !c = chooseChar62 r + in go q (c : cs) + + chooseChar62 :: Int -> Char + {-# INLINE chooseChar62 #-} + chooseChar62 (I# n) = C# (indexCharOffAddr# chars62 n) + chars62 = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"# + +-- | The socket file path for a semaphore. +getSemaphoreSocketPath :: SemaphoreName -> IO FilePath +getSemaphoreSocketPath sn = do + tmp <- getTemporaryDirectory + let dir = tmp "semaphore-compat" + createDirectoryIfMissing True dir + return (dir unversionedSemaphoreNameString sn) diff --git a/src/System/Semaphore/Internal/DomainSocket.hs b/src/System/Semaphore/Internal/DomainSocket.hs new file mode 100644 --- /dev/null +++ b/src/System/Semaphore/Internal/DomainSocket.hs @@ -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)) diff --git a/src/System/Semaphore/Internal/Posix.hs b/src/System/Semaphore/Internal/Posix.hs new file mode 100644 --- /dev/null +++ b/src/System/Semaphore/Internal/Posix.hs @@ -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 diff --git a/src/System/Semaphore/Internal/Unsupported.hs b/src/System/Semaphore/Internal/Unsupported.hs new file mode 100644 --- /dev/null +++ b/src/System/Semaphore/Internal/Unsupported.hs @@ -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" diff --git a/src/System/Semaphore/Internal/Win32.hs b/src/System/Semaphore/Internal/Win32.hs new file mode 100644 --- /dev/null +++ b/src/System/Semaphore/Internal/Win32.hs @@ -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) diff --git a/test-common/System/Semaphore/TestHelper.hs b/test-common/System/Semaphore/TestHelper.hs new file mode 100644 --- /dev/null +++ b/test-common/System/Semaphore/TestHelper.hs @@ -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 + | CmdWait -- ^ wait + | CmdWaitN Int -- ^ wait-n + | CmdRelease Int -- ^ release + | 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 () diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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 + ]