diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # interprocess [![Hackage](https://img.shields.io/hackage/v/interprocess.svg)](https://hackage.haskell.org/package/interprocess) -[![Build Status](https://secure.travis-ci.org/achirkin/interprocess.svg)](http://travis-ci.org/achirkin/interprocess) +[![Haskell CI](https://github.com/achirkin/interprocess/actions/workflows/build.yml/badge.svg)](https://github.com/achirkin/interprocess/actions/workflows/build.yml) Platform-independent interprocess communication. @@ -43,9 +43,9 @@ * [x] Mutable variables (`MVar`-like) via `Storable` instance. * [ ] `Control.Concurrent.Chan`-like channels * [ ] More tests - * [ ] Ensure Win32 waiting on events interruptible - (Custom interrupt signal handler + WaitForMultipleObjects) + * [x] Ensure MVar waits are interruptible w.r.t. async exceptions on all platforms * [ ] Benchmarks + ### Think about it diff --git a/cbits/SharedObjectName.c b/cbits/SharedObjectName.c deleted file mode 100644 --- a/cbits/SharedObjectName.c +++ /dev/null @@ -1,50 +0,0 @@ -#include "SharedObjectName.h" -#include -#include -#include -#include - -#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || defined(mingw32_HOST_OS) -#include -#define GetMyPid abs((int)GetCurrentProcessId()) -#else -#include -#define GetMyPid (int)getpid() -#endif - -static int _unique_seed = 0; -static int _my_pid = 0; -static const char keytable[] - = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"; -static const char prefix[] - = "/HsIPC."; -#define keytableLength 62 -#define prefixLength 7 - - -void genSharedObjectName(char * const name) { - // init this once per process - if(_unique_seed == 0){ - srand(time(NULL)); - _my_pid = GetMyPid; - } - // clear variable - memset(name, 0, sizeof(SharedObjectName)); - memcpy(name, prefix, sizeof(prefix)); - // put three chars at a time - int c = INT_MAX, i = prefixLength; - div_t randV = { .quot = rand() ^ 0x19a628f6 - , .rem = 0 } - , globV = { .quot = _my_pid ^ 0x003772a1 - , .rem = 0 } - , procV = { .quot = (_unique_seed++) ^ 0x028ab100 - , .rem = 0 }; - while( (c = c / keytableLength) > 0 && i < (SHARED_OBJECT_NAME_LENGTH-3) ) { - randV = div(randV.quot, keytableLength); - globV = div(globV.quot, keytableLength); - procV = div(procV.quot, keytableLength); - name[i++] = keytable[randV.rem]; - name[i++] = keytable[globV.rem]; - name[i++] = keytable[procV.rem]; - } -} diff --git a/cbits/common.c b/cbits/common.c new file mode 100644 --- /dev/null +++ b/cbits/common.c @@ -0,0 +1,58 @@ +#include "common.h" + +#include +#include +#include +#include + +#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || \ + defined(mingw32_HOST_OS) +#include +#define GetMyPid abs((int)GetCurrentProcessId()) +#else +#include +#define GetMyPid (int)getpid() +#endif + +static atomic_int process_seed = 0; +static _Thread_local bool initialized = false; +static _Thread_local int process_id = 0; +static const char keytable[] = + "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"; +static const char prefix[] = "/HsIPC."; +#define keytableLength 62 +#define prefixLength 7 +// Leave a few symbol placeholders empty to be able to append some +// short suffixes to the object name in-place. +#define suffixLength 1 + +void genSharedObjectName(char* const name) { + int seed = atomic_fetch_add_explicit(&process_seed, 1, memory_order_relaxed); + // Init this once per thread + if (!initialized) { + initialized = true; + srand(time(NULL)); + process_id = GetMyPid; + } + // Clear the name + memset(name, 0, sizeof(SharedObjectName)); + memcpy(name, prefix, sizeof(prefix)); + // Combine three sources of uniqueness + const int rands_num = 3; + div_t rands[] = {{.quot = rand() ^ 0x19a628f6, .rem = 0}, + {.quot = process_id ^ 0x003772a1, .rem = 0}, + {.quot = seed ^ 0x028ab100, .rem = 0}}; + for (int i = prefixLength, j = 0, k = 0; + i < SHARED_OBJECT_NAME_LENGTH - 1 - suffixLength && k < rands_num; + i++, j = (j + 1) % rands_num) { + rands[j] = div(rands[j].quot, keytableLength); + name[i] = keytable[rands[j].rem]; + if (rands[j].quot == 0) k++; + } +} + +bool has_blocked_exceptions(StgStablePtr tso) { + return (typeof(END_TSO_QUEUE))( + ((StgTSO*)deRefStablePtr(tso))->blocked_exceptions) != + END_TSO_QUEUE; +} diff --git a/examples/concurrent-malloc.hs b/examples/concurrent-malloc.hs --- a/examples/concurrent-malloc.hs +++ b/examples/concurrent-malloc.hs @@ -42,7 +42,7 @@ putStrLn $ "[A] Malloc'ed addr: " ++ show ptr poke ptr n - ec' <- withProcess_ processBConfig $ \procB -> do + ec' <- withProcessWait_ processBConfig $ \procB -> do putStrLn $ "[A] Sending store name: " ++ show (allocStoreName sa) diff --git a/examples/wait-mvar.hs b/examples/wait-mvar.hs --- a/examples/wait-mvar.hs +++ b/examples/wait-mvar.hs @@ -130,4 +130,4 @@ -> ([Process stdin stdout stderr] -> IO a) -> IO a withProcesses [] k = k [] -withProcesses (conf:cfs) k = withProcesses cfs $ \ps -> withProcess_ conf $ k . (:ps) +withProcesses (conf:cfs) k = withProcesses cfs $ \ps -> withProcessWait_ conf $ k . (:ps) diff --git a/examples/wait-qsem.hs b/examples/wait-qsem.hs --- a/examples/wait-qsem.hs +++ b/examples/wait-qsem.hs @@ -85,5 +85,5 @@ -> ([Process stdin stdout stderr] -> IO a) -> IO a withNProcesses 0 _ k = k [] -withNProcesses n conf k = withProcess_ conf $ \p -> +withNProcesses n conf k = withProcessWait_ conf $ \p -> withNProcesses (n-1) conf (k . (p:)) diff --git a/include/SharedObjectName.h b/include/SharedObjectName.h deleted file mode 100644 --- a/include/SharedObjectName.h +++ /dev/null @@ -1,10 +0,0 @@ -#ifndef __HSSHAREDOBJECTNAME_H__ -#define __HSSHAREDOBJECTNAME_H__ - -#define SHARED_OBJECT_NAME_LENGTH 32 -#ifndef HS_IMPORT_CONSTANTS_ONLY -typedef char SharedObjectName[SHARED_OBJECT_NAME_LENGTH]; -void genSharedObjectName(char * const name); -#endif - -#endif /* __HSSHAREDOBJECTNAME_H__ */ diff --git a/include/SharedPtr.h b/include/SharedPtr.h --- a/include/SharedPtr.h +++ b/include/SharedPtr.h @@ -1,12 +1,11 @@ #ifndef __HSSHAREDPTR_H__ #define __HSSHAREDPTR_H__ +#include "common.h" #include -#include "HsFFI.h" -#include "MachDeps.h" -#include "SharedObjectName.h" -#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || defined(mingw32_HOST_OS) +#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || \ + defined(mingw32_HOST_OS) #include typedef struct SharedMutex { SharedObjectName mutexName; @@ -14,7 +13,7 @@ #else #include typedef struct SharedMutex { - pthread_mutex_t mutVal; + pthread_mutex_t mutVal; pthread_mutexattr_t mutAttr; } SharedMutex; #endif @@ -33,32 +32,31 @@ // The allocator keeps a number of its instances; // If you create+lookup an allocator N times, you need to destroy it N times. // Actual desctuction of shared data happens only on the last usage. -void shared_destroyAllocator(SharedAllocator *aptr); +void shared_destroyAllocator(SharedAllocator *aptr); // Get shared store name, to pass it to another process and lookup an allocator. char *shared_getStoreName(SharedAllocator *aptr); // Convert a pointer to a tagged portable shared pointer. // The result pointer cannot be used directly, but can be transfered between // different memory spaces. -SharedPtr shared_ptrToShPtr(SharedAllocator *aptr, void *ptr); +SharedPtr shared_ptrToShPtr(SharedAllocator *aptr, void *ptr); // May return NULL on failure. -void *shared_shPtrToPtr(SharedAllocator *aptr, SharedPtr ptr); +void *shared_shPtrToPtr(SharedAllocator *aptr, SharedPtr ptr); // Allocate memory. -// Memory is aligned to page size or closest bigger power of two to the allocation size. -// Returns NULL on failure. +// Memory is aligned to page size or closest bigger power of two to the +// allocation size. Returns NULL on failure. void *shared_malloc(SharedAllocator *aptr, size_t size); void *shared_realloc(SharedAllocator *aptr, void *ptr, size_t size); -void shared_free(SharedAllocator *aptr, void *ptr); - - - +void shared_free(SharedAllocator *aptr, void *ptr); -void _SharedMutex_init(SharedMutex *mptr, void **privateMutexHandle, const int createNew); +void _SharedMutex_init(SharedMutex *mptr, void **privateMutexHandle, + const int createNew); void _SharedMutex_destroy(SharedMutex *mptr, void **privateMutexHandle); int _SharedMutex_lock(SharedMutex *mptr, void **privateMutexHandle); int _SharedMutex_unlock(SharedMutex *mptr, void **privateMutexHandle); -HsPtr _store_alloc(const char *memBlockName, void **privateStoreHandle, size_t size); -void _store_free(const char *memBlockName, void **privateStoreHandle, HsPtr addr, size_t size, _Bool unlinkToo); - +HsPtr _store_alloc(const char *memBlockName, void **privateStoreHandle, + size_t size); +void _store_free(const char *memBlockName, void **privateStoreHandle, + HsPtr addr, size_t size, _Bool unlinkToo); #endif /* __HSSHAREDPTR_H__ */ diff --git a/include/common.h b/include/common.h new file mode 100644 --- /dev/null +++ b/include/common.h @@ -0,0 +1,42 @@ +#ifndef __HSCOMMON_H__ +#define __HSCOMMON_H__ + +#define SHARED_OBJECT_NAME_LENGTH 32 + +#ifndef HS_IMPORT_CONSTANTS_ONLY +#include + +/** + * The input is an StgTSO* wrapped in a stable pointer. + * The StgTSO object can be moved during GC, which can happen while we're + * in a safe/interruptible foreign call. + * + * By performing this check, I can find out if there are any pending async + * exceptions in the calling (and suspended) Haskell thread. I need this, + * because on Windows there seem to be no way to interrupt / check the + * interruption status in between `WaitForSingleObject` or alike using the + * provided GHC machinery (CancelSynchronousIo). + * + * @param [in] tso a stable pointer to an StgTSO object. + * @return whether there are any pending blocked exceptions on this Haskell + * thread. + */ +bool has_blocked_exceptions(StgStablePtr tso); + +typedef char SharedObjectName[SHARED_OBJECT_NAME_LENGTH]; + +void genSharedObjectName(char* const name); + +#ifdef INTERPROCESS_DEBUG +#include +#define INTERPROCESS_LOG_DEBUG(...) \ + do { \ + printf("%s:%d ", __FILE__, __LINE__); \ + printf(__VA_ARGS__); \ + } while (0) +#else +#define INTERPROCESS_LOG_DEBUG(...) (void)0 +#endif + +#endif /* HS_IMPORT_CONSTANTS_ONLY */ +#endif /* __HSCOMMON_H__ */ diff --git a/interprocess.cabal b/interprocess.cabal --- a/interprocess.cabal +++ b/interprocess.cabal @@ -1,37 +1,39 @@ +cabal-version: 3.0 name: interprocess -version: 0.2.0.1 +version: 0.2.1.0 synopsis: Shared memory and control structures for IPC description: Provides portable shared memory allocator and some synchronization primitives. Can be used for interprocess communication. Refer to README.md for further information. homepage: https://github.com/achirkin/interprocess -license: BSD3 +license: BSD-3-Clause license-file: LICENSE author: Artem Chirkin -maintainer: chirkin@arch.ethz.ch -copyright: (c) 2018 Artem Chirkin +maintainer: achirkin@users.noreply.github.com +copyright: (c) 2018-2022 Artem Chirkin category: system build-type: Simple -cabal-version: >=1.10 extra-source-files: README.md - cbits/SharedObjectName.c - include/SharedObjectName.h + include/common.h include/SharedPtr.h - src/Control/Concurrent/Process/StoredMVar.c src/Control/Concurrent/Process/QSem.c src/Foreign/SharedPtrPosix.c src/Foreign/SharedPtrWin32.c + src/Control/Concurrent/Process/StoredMVarPosix.c + src/Control/Concurrent/Process/StoredMVarWin32.c flag examples description: Build example-test programs default: False + manual: True flag dev description: Development build with debugging and with no optimization default: False + manual: True library hs-source-dirs: src @@ -45,16 +47,20 @@ default-language: Haskell2010 ghc-options: -Wall include-dirs: include - c-sources: cbits/SharedObjectName.c - src/Control/Concurrent/Process/StoredMVar.c + c-sources: cbits/common.c src/Control/Concurrent/Process/QSem.c src/Foreign/SharedPtr.c + cmm-sources: src/Control/Concurrent/Process/StoredMVarPrim.cmm if os(windows) c-sources: src/Foreign/SharedPtrWin32.c + src/Control/Concurrent/Process/StoredMVarWin32.c else c-sources: src/Foreign/SharedPtrPosix.c + src/Control/Concurrent/Process/StoredMVarPosix.c if flag(dev) - ghc-options: -O0 + ghc-options: -O0 -debug -dcmm-lint + cpp-options: -DINTERPROCESS_DEBUG + cc-options: -DINTERPROCESS_DEBUG else cpp-options: -DNDEBUG cc-options: -DNDEBUG @@ -68,7 +74,7 @@ main-is: concurrent-malloc.hs default-language: Haskell2010 build-depends: base - , typed-process >= 0.2 + , typed-process >= 0.2.5 , interprocess ghc-options: -threaded -Wall @@ -79,7 +85,7 @@ main-is: wait-qsem.hs default-language: Haskell2010 build-depends: base - , typed-process >= 0.2 + , typed-process >= 0.2.5 , interprocess ghc-options: -threaded -Wall @@ -90,16 +96,24 @@ main-is: wait-mvar.hs default-language: Haskell2010 build-depends: base - , typed-process >= 0.2 + , typed-process >= 0.2.5 , interprocess ghc-options: -threaded -Wall test-suite StoredMVar type: exitcode-stdio-1.0 main-is: StoredMVar.hs + other-modules: Tools.Handle + Tools.TestResult + Tools.Runner default-language: Haskell2010 hs-source-dirs: test - build-depends: base - , typed-process >= 0.2 + build-depends: base >= 4.11 + , typed-process >= 0.2.5 + , async + , containers + , bytestring + , stm + , temporary , interprocess ghc-options: -threaded -Wall diff --git a/src/Control/Concurrent/Process/QSem.c b/src/Control/Concurrent/Process/QSem.c --- a/src/Control/Concurrent/Process/QSem.c +++ b/src/Control/Concurrent/Process/QSem.c @@ -1,6 +1,5 @@ -#include "SharedObjectName.h" -#include "HsFFI.h" -#include +#include + typedef struct QSem QSem; diff --git a/src/Control/Concurrent/Process/StoredMVar.c b/src/Control/Concurrent/Process/StoredMVar.c deleted file mode 100644 --- a/src/Control/Concurrent/Process/StoredMVar.c +++ /dev/null @@ -1,746 +0,0 @@ -#include "SharedObjectName.h" -#include -#include -#ifndef NDEBUG -// a bit of printf to track important events -#include -#endif - -typedef struct MVar MVar; - -MVar *mvar_new(size_t byteSize); -MVar *mvar_lookup(const char *name); -void mvar_destroy(MVar *mvar); - -int mvar_take (MVar *mvar, void *localDataPtr); -int mvar_trytake(MVar *mvar, void *localDataPtr); -int mvar_put (MVar *mvar, void *localDataPtr); -int mvar_tryput (MVar *mvar, void *localDataPtr); -int mvar_read (MVar *mvar, void *localDataPtr); -int mvar_tryread(MVar *mvar, void *localDataPtr); -int mvar_swap (MVar *mvar, void *inPtr, void *outPtr); -int mvar_tryswap(MVar *mvar, void *inPtr, void *outPtr); -int mvar_isempty(MVar *mvar); - - -#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || defined(mingw32_HOST_OS) -#include -#include -#ifndef WAIT_OBJECT_1 -#define WAIT_OBJECT_1 ((WAIT_OBJECT_0 ) + 1 ) -#endif - -typedef struct MVarState { - size_t dataSize; - int pendingReaders; -} MVarState; - - -typedef struct MVar { - /* Semaphore: readers wait on this. - mvar_put releases 2n-1 units of semaphore, and every reader takes two units, - the last reader fails to take a second unit and sets canTakeE. - */ - HANDLE canReadS; - /* Auto-reset event: takers wait on this - */ - HANDLE canTakeE; - /* Auto-reset event: putters wait on this - */ - HANDLE canPutE; - /* Mutex: protect the number of pending readers - */ - HANDLE protectReaders; - /* FileMapping: keep the data by this handle - */ - HANDLE dataStoreH; - /* Address of the data store - */ - MVarState *storePtr; - /* Actual data is stored next to the MVarState - */ - void *dataPtr; - /* Base name of the shared memory region, used to share mvar across processes. - Secondary objects are contstructed by appending a single char to the name. - */ - SharedObjectName mvarName; -} MVar; - - -MVar *mvar_new(size_t byteSize) { - // allocate MVar - MVar *mvar = malloc(sizeof(MVar)); - if (mvar == NULL) { - return NULL; - } - genSharedObjectName(mvar->mvarName); - - mvar->dataStoreH = CreateFileMappingA - ( INVALID_HANDLE_VALUE // use paging file - , NULL // default security - , PAGE_READWRITE // read/write access - , (DWORD)(byteSize >> 32) // maximum object size (high-order DWORD) - , (DWORD)((byteSize + 64) & 0xFFFFFFFF) // maximum object size (low-order DWORD) - , mvar->mvarName); // name of mapping object - - if (mvar->dataStoreH == NULL) { -#ifdef NDEBUG - printf("Could not create file mapping object (%d).\n", GetLastError()); -#endif - free(mvar); - return NULL; - } - mvar->storePtr = (MVarState*)MapViewOfFile - ( mvar->dataStoreH // handle to map object - , FILE_MAP_ALL_ACCESS // read/write permission - , 0 - , 0 - , byteSize + 64); - - if (mvar->storePtr == NULL) { -#ifdef NDEBUG - printf("Could not map view of file (%d).\n", GetLastError()); -#endif - CloseHandle(mvar->dataStoreH); - free(mvar); - return NULL; - } - mvar->dataPtr = ((void*)(mvar->storePtr)) + 64; - - // now, create all synchronization objects - // TODO: NULL value checking? - SharedObjectName objName = {0}; - strcpy (objName, mvar->mvarName); - strcat (objName, "T"); - mvar->canTakeE = CreateEventA( NULL, FALSE, FALSE, objName); - strcpy (objName, mvar->mvarName); - strcat (objName, "P"); - mvar->canPutE = CreateEventA( NULL, FALSE, TRUE, objName); - strcpy (objName, mvar->mvarName); - strcat (objName, "R"); - mvar->canReadS = CreateSemaphoreA( NULL, 0, LONG_MAX, objName); - strcpy (objName, mvar->mvarName); - strcat (objName, "M"); - mvar->protectReaders = CreateMutexA( NULL, FALSE, objName); - - // zero readers pending - *(mvar->storePtr) = (struct MVarState) - { .dataSize = byteSize - , .pendingReaders = 0 - }; - - return mvar; -} - -MVar *mvar_lookup(const char *name) { - // allocate MVar - MVar *mvar = malloc(sizeof(MVar)); - if (mvar == NULL) { - return NULL; - } - strcpy(mvar->mvarName, name); - - mvar->dataStoreH = OpenFileMappingA( FILE_MAP_ALL_ACCESS, FALSE, mvar->mvarName); - - if (mvar->dataStoreH == NULL) { -#ifdef NDEBUG - printf("Could not open file mapping object (%d).\n", GetLastError()); -#endif - free(mvar); - return NULL; - } - mvar->storePtr = (MVarState*)MapViewOfFile - ( mvar->dataStoreH // handle to map object - , FILE_MAP_READ // read/write permission - , 0 - , 0 - , 64); - - if (mvar->storePtr == NULL) { -#ifdef NDEBUG - printf("Could not map view of file (%d).\n", GetLastError()); -#endif - CloseHandle(mvar->dataStoreH); - free(mvar); - return NULL; - } - size_t storeSize = mvar->storePtr->dataSize + 64; - UnmapViewOfFile(mvar->storePtr); - mvar->storePtr = (MVarState*)MapViewOfFile - ( mvar->dataStoreH // handle to map object - , FILE_MAP_ALL_ACCESS // read/write permission - , 0 - , 0 - , storeSize); - - if (mvar->storePtr == NULL) { -#ifdef NDEBUG - printf("Could not map view of file (%d).\n", GetLastError()); -#endif - CloseHandle(mvar->dataStoreH); - free(mvar); - return NULL; - } - mvar->dataPtr = (void*)(mvar->storePtr) + 64; - - // now, create all synchronization objects - // TODO: NULL value checking? - SharedObjectName objName = {0}; - strcpy (objName, mvar->mvarName); - strcat (objName, "T"); - mvar->canTakeE = OpenEventA( EVENT_MODIFY_STATE | SYNCHRONIZE, FALSE, objName); - strcpy (objName, mvar->mvarName); - strcat (objName, "P"); - mvar->canPutE = OpenEventA( EVENT_MODIFY_STATE | SYNCHRONIZE, FALSE, objName); - strcpy (objName, mvar->mvarName); - strcat (objName, "R"); - mvar->canReadS = OpenSemaphoreA( SEMAPHORE_MODIFY_STATE | SYNCHRONIZE, FALSE, objName); - strcpy (objName, mvar->mvarName); - strcat (objName, "M"); - mvar->protectReaders = OpenMutexA( SYNCHRONIZE, FALSE, objName); - - return mvar; -} - -void mvar_destroy(MVar *mvar) { - UnmapViewOfFile(mvar->storePtr); - CloseHandle(mvar->canTakeE); - CloseHandle(mvar->canPutE); - CloseHandle(mvar->canReadS); - CloseHandle(mvar->protectReaders); - CloseHandle(mvar->dataStoreH); - free(mvar); -} - -int mvar_take (MVar *mvar, void *localDataPtr) { - DWORD r = WaitForSingleObject(mvar->canTakeE, INFINITE); - if (r != WAIT_OBJECT_0) { -#ifdef NDEBUG - printf("WaitForSingleObject canTakeE error: return %d; error code %d.\n", r, GetLastError()); -#endif - return 1; - } else { - memcpy(localDataPtr, mvar->dataPtr, mvar->storePtr->dataSize); - SetEvent(mvar->canPutE); - return 0; - } -} - -int mvar_trytake(MVar *mvar, void *localDataPtr) { - DWORD r = WaitForSingleObject(mvar->canTakeE, 0); - switch (r) { - case WAIT_OBJECT_0: - memcpy(localDataPtr, mvar->dataPtr, mvar->storePtr->dataSize); - SetEvent(mvar->canPutE); - return 0; - case WAIT_TIMEOUT: - return 1; - default: -#ifdef NDEBUG - printf("WaitForSingleObject canTakeE error: return %d; error code %d.\n", r, GetLastError()); -#endif - return 1; - } -} - -int mvar_put (MVar *mvar, void *localDataPtr) { - DWORD r = WaitForSingleObject(mvar->canPutE, INFINITE); - if (r != WAIT_OBJECT_0) { -#ifdef NDEBUG - printf("WaitForSingleObject canPutE error: return %d; error code %d.\n", r, GetLastError()); -#endif - return 1; - } else { - memcpy(mvar->dataPtr, localDataPtr, mvar->storePtr->dataSize); - // first check readers, and only then, maybe allow takers - r = WaitForSingleObject(mvar->protectReaders, INFINITE); - assert( r == WAIT_OBJECT_0 ); - int remRdrs = mvar->storePtr->pendingReaders; - r = ReleaseMutex(mvar->protectReaders); - assert( r != 0 ); - if (remRdrs == 0) { - SetEvent(mvar->canTakeE); - } else { - ReleaseSemaphore(mvar->canReadS, 2 * (LONG) remRdrs - 1, NULL); - } - return 0; - } -} - -int mvar_tryput (MVar *mvar, void *localDataPtr) { - DWORD r = WaitForSingleObject(mvar->canPutE, 0); - switch (r) { - case WAIT_OBJECT_0: - memcpy(mvar->dataPtr, localDataPtr, mvar->storePtr->dataSize); - // first check readers, and only then, maybe allow takers - r = WaitForSingleObject(mvar->protectReaders, INFINITE); - assert( r == WAIT_OBJECT_0 ); - int remRdrs = mvar->storePtr->pendingReaders; - r = ReleaseMutex(mvar->protectReaders); - assert( r != 0 ); - if (remRdrs == 0) { - SetEvent(mvar->canTakeE); - } else { - ReleaseSemaphore(mvar->canReadS, 2 * (LONG) remRdrs - 1, NULL); - } - return 0; - case WAIT_TIMEOUT: - return 1; - default: -#ifdef NDEBUG - printf("WaitForSingleObject canTakeE error: return %d; error code %d.\n", r, GetLastError()); -#endif - return 1; - } -} - -int mvar_read (MVar *mvar, void *localDataPtr) { - DWORD r = WaitForSingleObject(mvar->protectReaders, INFINITE); - assert( r == WAIT_OBJECT_0 ); - mvar->storePtr->pendingReaders++; - r = ReleaseMutex(mvar->protectReaders); - assert( r != 0 ); - DWORD signaled = WaitForMultipleObjects(2, (HANDLE*)mvar, FALSE, INFINITE); - switch (signaled) { - case WAIT_OBJECT_0: - case WAIT_OBJECT_1: - memcpy(localDataPtr, mvar->dataPtr, mvar->storePtr->dataSize); - r = WaitForSingleObject(mvar->protectReaders, INFINITE); - assert( r == WAIT_OBJECT_0 ); - --(mvar->storePtr->pendingReaders); - r = ReleaseMutex(mvar->protectReaders); - assert( r != 0 ); - if ( signaled == WAIT_OBJECT_0 ) { - if (WaitForSingleObject(mvar->canReadS, 0) != WAIT_OBJECT_0) { - SetEvent(mvar->canTakeE); - } - } else if (signaled == WAIT_OBJECT_1 ) { - SetEvent(mvar->canTakeE); - } - return 0; - default: -#ifdef NDEBUG - printf("(mvar_read) WaitForMultipleObjects error: return %d; error code %d.\n", r, GetLastError()); -#endif - return 1; - } -} - -int mvar_tryread(MVar *mvar, void *localDataPtr) { - DWORD r = WaitForSingleObject(mvar->canTakeE, 0); - switch (r) { - case WAIT_OBJECT_0: - memcpy(localDataPtr, mvar->dataPtr, mvar->storePtr->dataSize); - SetEvent(mvar->canTakeE); - return 0; - case WAIT_TIMEOUT: - return 1; - default: -#ifdef NDEBUG - printf("WaitForSingleObject canTakeE error: return %d; error code %d.\n", r, GetLastError()); -#endif - return 1; - } -} - -int mvar_swap (MVar *mvar, void *inPtr, void *outPtr) { - DWORD r = WaitForSingleObject(mvar->canTakeE, INFINITE); - if (r != WAIT_OBJECT_0) { -#ifdef NDEBUG - printf("WaitForSingleObject canTakeE error: return %d; error code %d.\n", r, GetLastError()); -#endif - return 1; - } else { - memcpy(outPtr, mvar->dataPtr, mvar->storePtr->dataSize); - memcpy(mvar->dataPtr, inPtr , mvar->storePtr->dataSize); - SetEvent(mvar->canTakeE); - return 0; - } -} - -int mvar_tryswap(MVar *mvar, void *inPtr, void *outPtr) { - DWORD r = WaitForSingleObject(mvar->canTakeE, 0); - switch (r) { - case WAIT_OBJECT_0: - memcpy(outPtr, mvar->dataPtr, mvar->storePtr->dataSize); - memcpy(mvar->dataPtr, inPtr , mvar->storePtr->dataSize); - SetEvent(mvar->canTakeE); - return 0; - case WAIT_TIMEOUT: - return 1; - default: -#ifdef NDEBUG - printf("WaitForSingleObject canTakeE error: return %d; error code %d.\n", r, GetLastError()); -#endif - return 1; - } -} - -int mvar_isempty(MVar *mvar) { - DWORD r = WaitForSingleObject(mvar->canPutE, 0); - if (r == WAIT_TIMEOUT) { - return 1; - } else { - SetEvent(mvar->canPutE); - return 0; - } -} - - - -#else -#include -#include -#include -#include - - -typedef struct MVarState { - pthread_mutex_t mvMut; - pthread_cond_t canPutC; - pthread_cond_t canTakeC; - size_t dataSize; - pthread_mutexattr_t mvMAttr; - pthread_condattr_t condAttr; - int isFull; - int pendingReaders; - int totalUsers; -} MVarState; - -typedef struct MVar { - /* State is stored in the shared data area, accessed by all threads. - It has a fixed size and kept at the beginning of a shared memory region. - */ - MVarState *statePtr; - /* Actual data is stored next to the MVarState - */ - void *dataPtr; - /* Name of the shared memory region, used to share mvar across processes. - */ - SharedObjectName mvarName; -} MVar; - -// ensure 64 byte alignment (maximum possible we can think of, e.g. AVX512) -size_t mvar_state_size64() { - size_t x = sizeof(MVarState); - size_t r = x % 64; - return r == 0 ? x : (x + 64 - r); -} - -MVar *mvar_new(size_t byteSize) { - int r = 0; - // allocate MVar - MVar *mvar = malloc(sizeof(MVar)); - if (mvar == NULL) { - return NULL; - } - genSharedObjectName(mvar->mvarName); - - // allocate shared memory for MVarState and data - size_t dataShift = mvar_state_size64(); - size_t totalSize = dataShift + byteSize; - int memFd = shm_open(mvar->mvarName, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR); - if (memFd < 0) { - free(mvar); - return NULL; - } - r = ftruncate(memFd, totalSize); - if (r != 0) { - shm_unlink(mvar->mvarName); - free(mvar); - return NULL; - } - mvar->statePtr = (MVarState*) mmap( NULL, totalSize - , PROT_READ | PROT_WRITE - , MAP_SHARED, memFd, 0); - if (mvar->statePtr == MAP_FAILED) { - shm_unlink(mvar->mvarName); - free(mvar); - return NULL; - } - mvar->dataPtr = ((void*)(mvar->statePtr)) + dataShift; - - // setup state - *(mvar->statePtr) = (struct MVarState) - { .isFull = 0 - , .pendingReaders = 0 - , .dataSize = byteSize - , .totalUsers = 1 - }; - - // init mutex and condition variables - r = pthread_mutexattr_init(&(mvar->statePtr->mvMAttr)); -#ifndef NDEBUG - if ( r == 0) r = pthread_mutexattr_settype(&(mvar->statePtr->mvMAttr), PTHREAD_MUTEX_ERRORCHECK); -#endif - if ( r == 0) r = pthread_mutexattr_setpshared(&(mvar->statePtr->mvMAttr), PTHREAD_PROCESS_SHARED); - if ( r == 0) r = pthread_mutex_init(&(mvar->statePtr->mvMut), &(mvar->statePtr->mvMAttr)); - if ( r == 0) r = pthread_condattr_init(&(mvar->statePtr->condAttr)); - if ( r == 0) r = pthread_condattr_setpshared(&(mvar->statePtr->condAttr), PTHREAD_PROCESS_SHARED); - if ( r == 0) r = pthread_cond_init(&(mvar->statePtr->canPutC), &(mvar->statePtr->condAttr)); - if ( r == 0) r = pthread_cond_init(&(mvar->statePtr->canTakeC), &(mvar->statePtr->condAttr)); - if ( r != 0 ) { - munmap(mvar->statePtr, totalSize); - shm_unlink(mvar->mvarName); - free(mvar); - return NULL; - } - - return mvar; -} - -MVar *mvar_lookup(const char *name) { - int memFd = shm_open(name, O_RDWR, S_IRUSR | S_IWUSR); - if (memFd < 0) return NULL; - - // first, map only sizeof(MVarState) bytes - // then, read the actual size and remap memory - void *mvs0 = mmap( NULL, sizeof(MVarState) - , PROT_READ, MAP_SHARED, memFd, 0); - if (mvs0 == MAP_FAILED) return NULL; - size_t dataShift = mvar_state_size64(), - storeSize = dataShift + ((MVarState*)mvs0)->dataSize; - munmap(mvs0, sizeof(MVarState)); // don't really care if it is failed - MVarState *mvs = (MVarState*) mmap( NULL, storeSize - , PROT_READ | PROT_WRITE - , MAP_SHARED, memFd, 0); - if (mvs == MAP_FAILED) return NULL; - // setup MVar struct - MVar *mvar = malloc(sizeof(MVar)); - if (mvar == NULL) { - munmap(mvs, storeSize); - return NULL; - } - mvar->statePtr = (MVarState*)mvs; - mvar->dataPtr = ((void*)(mvar->statePtr)) + dataShift; - strcpy(mvar->mvarName, name); - - // update state - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 || mvar->statePtr->totalUsers == 0 ) { - munmap(mvar->statePtr, storeSize); - free(mvar); - return NULL; - } - mvar->statePtr->totalUsers++; - r = pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) { - munmap(mvar->statePtr, storeSize); - free(mvar); - return NULL; - } - - return mvar; -} - -void mvar_destroy(MVar *mvar) { - int usersLeft; - size_t storeSize = mvar->statePtr->dataSize + mvar_state_size64(); - pthread_mutex_lock(&(mvar->statePtr->mvMut)); - usersLeft = --(mvar->statePtr->totalUsers); - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - if ( usersLeft > 0 ) { -#ifndef NDEBUG - printf( "Destroying local instance of mvar %s, %d users left.\n" - , mvar->mvarName, usersLeft); -#endif - munmap(mvar->statePtr, storeSize); - } else { -#ifndef NDEBUG - printf( "Destroying mvar %s globally (no other users left).\n", mvar->mvarName); -#endif - pthread_cond_destroy(&(mvar->statePtr->canTakeC)); - pthread_cond_destroy(&(mvar->statePtr->canPutC)); - pthread_condattr_destroy(&(mvar->statePtr->condAttr)); - pthread_mutex_destroy(&(mvar->statePtr->mvMut)); - pthread_mutexattr_destroy(&(mvar->statePtr->mvMAttr)); - munmap(mvar->statePtr, storeSize); - shm_unlink(mvar->mvarName); - } - free(mvar); -} - - -int mvar_take(MVar *mvar, void *localDataPtr) { - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) return r; - while ( !(mvar->statePtr->isFull) || mvar->statePtr->pendingReaders > 0 ) { - if ( mvar->statePtr->pendingReaders > 0 ) { - pthread_cond_broadcast(&(mvar->statePtr->canTakeC)); - } - r = pthread_cond_wait(&(mvar->statePtr->canTakeC), &(mvar->statePtr->mvMut)); - if ( r != 0 ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return r; - } - } - memcpy(localDataPtr, mvar->dataPtr, mvar->statePtr->dataSize); - mvar->statePtr->isFull = 0; - pthread_cond_signal(&(mvar->statePtr->canPutC)); - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 0; -} - -int mvar_trytake(MVar *mvar, void *localDataPtr) { - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) return r; - // shortcut if is empty - if ( !(mvar->statePtr->isFull) ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 1; - } - while ( mvar->statePtr->pendingReaders > 0 ) { - // make sure readers do not sleep - pthread_cond_broadcast(&(mvar->statePtr->canTakeC)); - // last reader should wake me up, wait for it. - r = pthread_cond_wait(&(mvar->statePtr->canTakeC), &(mvar->statePtr->mvMut)); - if ( r != 0 ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return r; - } - // repeat emptyness check (if another trytake was faster) - if ( !(mvar->statePtr->isFull) ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 1; - } - } - memcpy(localDataPtr, mvar->dataPtr, mvar->statePtr->dataSize); - mvar->statePtr->isFull = 0; - pthread_cond_signal(&(mvar->statePtr->canPutC)); - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 0; -} - -int mvar_put(MVar *mvar, void *localDataPtr) { - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) return r; - while ( mvar->statePtr->isFull ) { - r = pthread_cond_wait(&(mvar->statePtr->canPutC), &(mvar->statePtr->mvMut)); - if ( r != 0 ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return r; - } - } - memcpy(mvar->dataPtr, localDataPtr, mvar->statePtr->dataSize); - mvar->statePtr->isFull = 1; - pthread_cond_broadcast(&(mvar->statePtr->canTakeC)); - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 0; -} - -int mvar_tryput(MVar *mvar, void *localDataPtr) { - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) return r; - // shortcut if is full - if ( mvar->statePtr->isFull ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 1; - } - memcpy(mvar->dataPtr, localDataPtr, mvar->statePtr->dataSize); - mvar->statePtr->isFull = 1; - pthread_cond_broadcast(&(mvar->statePtr->canTakeC)); - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 0; -} - -int mvar_read(MVar *mvar, void *localDataPtr) { - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) return r; - mvar->statePtr->pendingReaders++; - while ( !(mvar->statePtr->isFull) ) { - r = pthread_cond_wait(&(mvar->statePtr->canTakeC), &(mvar->statePtr->mvMut)); - if ( r != 0 ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return r; - } - } - memcpy(localDataPtr, mvar->dataPtr, mvar->statePtr->dataSize); - if ( (--(mvar->statePtr->pendingReaders)) == 0 ) { - pthread_cond_signal(&(mvar->statePtr->canTakeC)); - } - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 0; -} - -int mvar_tryread(MVar *mvar, void *localDataPtr) { - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) return r; - // shortcut if is empty - if ( !(mvar->statePtr->isFull) ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 1; - } - memcpy(localDataPtr, mvar->dataPtr, mvar->statePtr->dataSize); - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 0; -} - -int mvar_swap(MVar *mvar, void *inPtr, void *outPtr) { - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) return r; - while ( !(mvar->statePtr->isFull) || mvar->statePtr->pendingReaders > 0 ) { - if ( mvar->statePtr->pendingReaders > 0 ) { - pthread_cond_broadcast(&(mvar->statePtr->canTakeC)); - } - r = pthread_cond_wait(&(mvar->statePtr->canTakeC), &(mvar->statePtr->mvMut)); - if ( r != 0 ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return r; - } - } - memcpy(outPtr, mvar->dataPtr, mvar->statePtr->dataSize); - memcpy(mvar->dataPtr, inPtr , mvar->statePtr->dataSize); - pthread_cond_signal(&(mvar->statePtr->canTakeC)); - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 0; -} - -int mvar_tryswap(MVar *mvar, void *inPtr, void *outPtr) { - int r = 0; - r = pthread_mutex_lock(&(mvar->statePtr->mvMut)); - if ( r != 0 ) return r; - // shortcut if is empty - if ( !(mvar->statePtr->isFull) ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 1; - } - while ( mvar->statePtr->pendingReaders > 0 ) { - // make sure readers do not sleep - pthread_cond_broadcast(&(mvar->statePtr->canTakeC)); - // last reader should wake me up, wait for it. - r = pthread_cond_wait(&(mvar->statePtr->canTakeC), &(mvar->statePtr->mvMut)); - if ( r != 0 ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return r; - } - // repeat emptyness check (if another trytake was faster) - if ( !(mvar->statePtr->isFull) ) { - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 1; - } - } - memcpy(outPtr, mvar->dataPtr, mvar->statePtr->dataSize); - memcpy(mvar->dataPtr, inPtr , mvar->statePtr->dataSize); - pthread_cond_signal(&(mvar->statePtr->canPutC)); - pthread_mutex_unlock(&(mvar->statePtr->mvMut)); - return 0; -} - -int mvar_isempty(MVar *mvar) { - return mvar->statePtr->isFull == 0; -} - -#endif - - -void mvar_name(MVar *mvar, char * const name) { - strcpy(name, mvar->mvarName); -} diff --git a/src/Control/Concurrent/Process/StoredMVar.hs b/src/Control/Concurrent/Process/StoredMVar.hs --- a/src/Control/Concurrent/Process/StoredMVar.hs +++ b/src/Control/Concurrent/Process/StoredMVar.hs @@ -1,5 +1,9 @@ -{-# LANGUAGE InterruptibleFFI #-} -{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE GHCForeignImportPrim #-} +{-# LANGUAGE MagicHash #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE UnboxedTuples #-} +{-# LANGUAGE UnliftedFFITypes #-} + ----------------------------------------------------------------------------- -- | -- This module is an adaptation of `Control.Concurrent.MVar` to an @@ -30,17 +34,25 @@ , modifyMVar, modifyMVar_, modifyMVarMasked, modifyMVarMasked_ ) where -import Control.Exception -import Control.Monad (when) -import Data.Data (Typeable) -import Foreign.C -import Foreign.ForeignPtr -import Foreign.Marshal.Alloc (alloca) -import Foreign.Marshal.Array (advancePtr, allocaArray) -import Foreign.Ptr -import Foreign.SharedObjectName.Internal -import Foreign.Storable +import Control.Exception +import Control.Monad (when) +import Data.Data (Typeable) +import Data.Maybe (fromMaybe) +import Foreign.C +import Foreign.ForeignPtr +import Foreign.Marshal.Alloc (alloca) +import Foreign.Marshal.Array (advancePtr, allocaArray) +import Foreign.Ptr +import Foreign.SharedObjectName.Internal +import Foreign.Storable +import System.Environment (lookupEnv) +import Text.Read (readMaybe) +import GHC.Exts (Int(I#), Addr#, Int#, RealWorld, State#) +import GHC.IO (IO (..)) +import GHC.Ptr (Ptr (..)) + + -- | Opaque implementation-dependent StoredMVar data StoredMVarT @@ -59,8 +71,8 @@ -- | Create a 'StoredMVar' which is initially empty. newEmptyMVar :: forall a . Storable a => IO (StoredMVar a) newEmptyMVar = mask_ $ do - mvar <- checkNullPointer "newEmptyMVar" - . c'mvar_new . fromIntegral $ sizeOf (undefined :: a) + t <- fromMaybe 100 . (>>= readMaybe) <$> lookupEnv "INTEPROCESS_MAX_ERR_WAIT_MS" + mvar <- checkNullPointer "newEmptyMVar" $ c'mvar_new (fromIntegral (sizeOf (undefined :: a))) t n <- newEmptySOName unsafeWithSOName n $ c'mvar_name mvar StoredMVar n <$> newForeignPtr p'mvar_destroy mvar @@ -111,10 +123,8 @@ -- takeMVar :: Storable a => StoredMVar a -> IO a takeMVar (StoredMVar _ fp) = mask_ $ withForeignPtr fp $ \p -> alloca $ \lp -> do - r <- c'mvar_take p lp - if r == 0 - then peek lp - else throwErrno $ "takeMVar failed with code " ++ show r + checkErrorCode "takeMVar" $ cmmOp (cmm'mvar_take (unptr p) (unptr lp)) + peek lp {-# INLINE takeMVar #-} @@ -127,10 +137,8 @@ -- readMVar :: Storable a => StoredMVar a -> IO a readMVar (StoredMVar _ fp) = mask_ $ withForeignPtr fp $ \p -> alloca $ \lp -> do - r <- c'mvar_read p lp - if r == 0 - then peek lp - else throwErrno $ "readMVar failed with code " ++ show r + checkErrorCode "readMVar" $ cmmOp (cmm'mvar_read (unptr p) (unptr lp)) + peek lp {-# INLINE readMVar #-} @@ -141,10 +149,8 @@ = mask_ $ withForeignPtr fp $ \p -> allocaArray 2 $ \inp -> do let outp = advancePtr inp 1 poke inp x - r <- c'mvar_swap p inp outp - if r == 0 - then peek outp - else throwErrno $ "swapMVar failed with code " ++ show r + checkErrorCode "swapMVar" $ cmmOp (cmm'mvar_swap (unptr p) (unptr inp) (unptr outp)) + peek outp {-# INLINE swapMVar #-} @@ -162,8 +168,7 @@ putMVar :: Storable a => StoredMVar a -> a -> IO () putMVar (StoredMVar _ fp) x = mask_ $ withForeignPtr fp $ \p -> alloca $ \lp -> do poke lp x - r <- c'mvar_put p lp - when (r /= 0) $ throwErrno $ "putMVar failed with code " ++ show r + checkErrorCode "putMVar" $ cmmOp (cmm'mvar_put (unptr p) (unptr lp)) {-# NOINLINE putMVar #-} -- | A non-blocking version of 'takeMVar'. The 'tryTakeMVar' function @@ -219,13 +224,20 @@ checkNullPointer s k = do p <- k if p == nullPtr - then throwErrno ("StoredMVar." ++ s ++ ": FFI returned NULL pointer.") + then ioError $ errnoToIOError ("StoredMVar." ++ s ++ ": FFI returned NULL pointer.") eINVAL Nothing Nothing else return p {-# INLINE checkNullPointer #-} +checkErrorCode :: String -> IO CInt -> IO () +checkErrorCode s k = do + e <- k + when (e /= 0) $ + ioError $ errnoToIOError ("StoredMVar." ++ s ++ ": FFI failed with code " ++ show e) (Errno e) Nothing Nothing +{-# INLINE checkErrorCode #-} + foreign import ccall unsafe "mvar_new" - c'mvar_new :: CSize -> IO (Ptr StoredMVarT) + c'mvar_new :: CSize -> CInt -> IO (Ptr StoredMVarT) foreign import ccall unsafe "mvar_lookup" c'mvar_lookup :: CString -> IO (Ptr StoredMVarT) @@ -236,35 +248,35 @@ foreign import ccall unsafe "mvar_name" c'mvar_name :: Ptr StoredMVarT -> CString -> IO () --- | Waits a lot and should be interruptible -foreign import ccall interruptible "mvar_take" - c'mvar_take :: Ptr StoredMVarT -> Ptr a -> IO CInt --- | Waits a bit and may be unsafe +-- These wait a bit and thus may be unsafe foreign import ccall unsafe "mvar_trytake" c'mvar_trytake :: Ptr StoredMVarT -> Ptr a -> IO CInt --- | Waits a lot and should be interruptible -foreign import ccall interruptible "mvar_put" - c'mvar_put :: Ptr StoredMVarT -> Ptr a -> IO CInt --- | Waits a bit and may be unsafe foreign import ccall unsafe "mvar_tryput" c'mvar_tryput :: Ptr StoredMVarT -> Ptr a -> IO CInt --- | Waits a lot and should be interruptible -foreign import ccall interruptible "mvar_read" - c'mvar_read :: Ptr StoredMVarT -> Ptr a -> IO CInt --- | Does not wait and can be unsafe foreign import ccall unsafe "mvar_tryread" c'mvar_tryread :: Ptr StoredMVarT -> Ptr a -> IO CInt --- | Waits a lot and should be interruptible -foreign import ccall interruptible "mvar_swap" - c'mvar_swap :: Ptr StoredMVarT -> Ptr a -> Ptr a -> IO CInt --- | Waits a bit and may be unsafe foreign import ccall unsafe "mvar_tryswap" c'mvar_tryswap :: Ptr StoredMVarT -> Ptr a -> Ptr a -> IO CInt --- | Does not wait and can be unsafe foreign import ccall unsafe "mvar_isempty" c'mvar_isempty :: Ptr StoredMVarT -> IO CInt +-- These wait a lot and should be interruptible. +-- I need the thread state inside these on the C side, so there are C-- helpers. +foreign import prim "cmm_mvar_take" + cmm'mvar_take :: Addr# -> Addr# -> State# RealWorld -> (# State# RealWorld, Int# #) +foreign import prim "cmm_mvar_put" + cmm'mvar_put :: Addr# -> Addr# -> State# RealWorld -> (# State# RealWorld, Int# #) +foreign import prim "cmm_mvar_read" + cmm'mvar_read :: Addr# -> Addr# -> State# RealWorld -> (# State# RealWorld, Int# #) +foreign import prim "cmm_mvar_swap" + cmm'mvar_swap :: Addr# -> Addr# -> Addr# -> State# RealWorld -> (# State# RealWorld, Int# #) + +cmmOp :: (State# RealWorld -> (# State# RealWorld, Int# #)) -> IO CInt +cmmOp op = IO (\s0 -> case op s0 of (# s1, code #) -> (# s1, fromIntegral (I# code) #)) + +unptr :: Ptr a -> Addr# +unptr (Ptr p) = p -- | 'withMVar' is an exception-safe wrapper for operating on the contents -- of an 'StoredMVar'. This operation is exception-safe: it will replace the diff --git a/src/Control/Concurrent/Process/StoredMVarPosix.c b/src/Control/Concurrent/Process/StoredMVarPosix.c new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Process/StoredMVarPosix.c @@ -0,0 +1,413 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef INTERPROCESS_DEBUG +#define ASSERT_ZERO(op) assert(op == 0) +#else +#define ASSERT_ZERO(op) op +#endif + +// NB: use robust mutexes +// https://man7.org/linux/man-pages/man3/pthread_mutexattr_setrobust.3.html + +typedef struct MVar MVar; + +MVar *mvar_new(size_t byteSize, int max_wait_ms); +MVar *mvar_lookup(const char *name); +void mvar_destroy(MVar *mvar); + +int mvar_take(MVar *mvar, void *localDataPtr, StgStablePtr tso); +int mvar_trytake(MVar *mvar, void *localDataPtr); +int mvar_put(MVar *mvar, void *localDataPtr, StgStablePtr tso); +int mvar_tryput(MVar *mvar, void *localDataPtr); +int mvar_read(MVar *mvar, void *localDataPtr, StgStablePtr tso); +int mvar_tryread(MVar *mvar, void *localDataPtr); +int mvar_swap(MVar *mvar, void *inPtr, void *outPtr, StgStablePtr tso); +int mvar_tryswap(MVar *mvar, void *inPtr, void *outPtr); +int mvar_isempty(MVar *mvar); + +typedef struct MVarState { + size_t dataSize; + /** + * Maximum wait time in milliseconds. + * This determines maximum possible delay the mvar operation cancels in + * the event of an async exception thrown to the calling Haskell thread. + */ + int maxWaitMs; + int pendingReaders; + + // following is OS-specific + + int isFull; + atomic_int totalUsers; + pthread_mutex_t mvMut; + pthread_cond_t canPutC; + pthread_cond_t canTakeC; +} MVarState; + +typedef struct MVar { + /** + * State is stored in the shared data area, accessed by all processes. + * It has a fixed size and kept at the beginning of a shared memory region. + */ + MVarState *statePtr; + /** Actual data is stored next to the MVarState */ + void *dataPtr; + /** Name of the shared memory region, used to share mvar across processes. */ + SharedObjectName mvarName; +} MVar; + +// ensure 64 byte alignment (maximum possible we can think of, e.g. AVX512) +size_t mvar_state_size64() { + size_t x = sizeof(MVarState); + return (x + 63) & (~63); +} + +/** + * Increase the total users counter by one and return zero; + * if there are no users left (counter == 0), fail (return non-zero). + */ +static inline int inc_totalUsers(MVar *mvar) { + atomic_int *p = &(mvar->statePtr->totalUsers); + int exp = atomic_load_explicit(p, memory_order_relaxed); + while (exp >= 1) { + if (atomic_compare_exchange_weak_explicit( + p, &exp, exp + 1, memory_order_relaxed, memory_order_relaxed)) + return 0; + } + return -1; +} + +MVar *mvar_new(size_t byteSize, int max_wait_ms) { + int r = 0; + // allocate MVar + MVar *mvar = malloc(sizeof(MVar)); + if (mvar == NULL) goto failed; + genSharedObjectName(mvar->mvarName); + + // allocate shared memory for MVarState and data + size_t dataShift = mvar_state_size64(); + size_t totalSize = dataShift + byteSize; + int memFd = shm_open(mvar->mvarName, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR); + if (memFd < 0) { + INTERPROCESS_LOG_DEBUG("Could not open shared memory segment (%d).\n", + errno); + goto failed_after_mvar_allocated; + } + + if (ftruncate(memFd, totalSize) != 0) { + INTERPROCESS_LOG_DEBUG( + "Could not allocate shared memory (ftruncate) (%d).\n", errno); + goto failed_after_shmem_open; + } + mvar->statePtr = (MVarState *)mmap(NULL, totalSize, PROT_READ | PROT_WRITE, + MAP_SHARED, memFd, 0); + if (mvar->statePtr == MAP_FAILED) { + INTERPROCESS_LOG_DEBUG("Could not map shared memory (mmap) (%d).\n", errno); + goto failed_after_shmem_open; + } + mvar->dataPtr = ((void *)(mvar->statePtr)) + dataShift; + + // setup state + *(mvar->statePtr) = (struct MVarState){.isFull = 0, + .pendingReaders = 0, + .dataSize = byteSize, + .totalUsers = 1, + .maxWaitMs = max_wait_ms}; + + // init mutex and condition variables + pthread_mutexattr_t mvMAttr; + r = pthread_mutexattr_init(&mvMAttr); + if (r != 0) goto failed_on_mutex_init; +#ifdef INTERPROCESS_DEBUG + r = pthread_mutexattr_settype(&mvMAttr, PTHREAD_MUTEX_ERRORCHECK); + if (r != 0) goto failed_on_mutex_init; +#endif + r = pthread_mutexattr_setpshared(&mvMAttr, PTHREAD_PROCESS_SHARED); + if (r != 0) goto failed_on_mutex_init; + r = pthread_mutex_init(&(mvar->statePtr->mvMut), &mvMAttr); + if (r != 0) goto failed_on_mutex_init; + r = pthread_mutexattr_destroy(&mvMAttr); + if (r != 0) { + failed_on_mutex_init: + INTERPROCESS_LOG_DEBUG("Could not init a mutex (%d).\n", r); + goto failed_after_mmap; + } + + pthread_condattr_t condAttr; + r = pthread_condattr_init(&condAttr); + if (r != 0) goto failed_on_cond_init; + r = pthread_condattr_setpshared(&condAttr, PTHREAD_PROCESS_SHARED); + if (r != 0) goto failed_on_cond_init; + r = pthread_cond_init(&(mvar->statePtr->canPutC), &condAttr); + if (r != 0) goto failed_on_cond_init; + r = pthread_cond_init(&(mvar->statePtr->canTakeC), &condAttr); + if (r != 0) goto failed_on_cond_init; + pthread_condattr_destroy(&condAttr); + if (r != 0) { + failed_on_cond_init: + INTERPROCESS_LOG_DEBUG("Could not init a condition variable (%d).\n", r); + goto failed_after_mmap; + } + + return mvar; + +failed_after_mmap: + munmap(mvar->statePtr, totalSize); +failed_after_shmem_open: + shm_unlink(mvar->mvarName); +failed_after_mvar_allocated: + free(mvar); +failed: + return NULL; +} + +MVar *mvar_lookup(const char *name) { + // allocate MVar + MVar *mvar = malloc(sizeof(MVar)); + if (mvar == NULL) goto failed; + strcpy(mvar->mvarName, name); + + int memFd = shm_open(name, O_RDWR, S_IRUSR | S_IWUSR); + if (memFd < 0) { + INTERPROCESS_LOG_DEBUG("Could not open shared memory segment (%d).\n", + errno); + goto failed_after_mvar_allocated; + } + + // first, map only sizeof(MVarState) bytes + // then, read the actual size and remap memory + mvar->statePtr = (MVarState *)mmap(NULL, sizeof(MVarState), PROT_READ, + MAP_SHARED, memFd, 0); + if (mvar->statePtr == MAP_FAILED) { + INTERPROCESS_LOG_DEBUG("Could not map shared memory (mmap) (%d).\n", errno); + goto failed_after_mvar_allocated; + } + size_t dataShift = mvar_state_size64(); + size_t totalSize = dataShift + mvar->statePtr->dataSize; + ASSERT_ZERO(munmap(mvar->statePtr, sizeof(MVarState))); + + // now we know the actual size, map the memory again + mvar->statePtr = (MVarState *)mmap(NULL, totalSize, PROT_READ | PROT_WRITE, + MAP_SHARED, memFd, 0); + if (mvar->statePtr == MAP_FAILED) { + INTERPROCESS_LOG_DEBUG("Could not map shared memory (mmap) (%d).\n", errno); + goto failed_after_mvar_allocated; + } + mvar->dataPtr = ((void *)(mvar->statePtr)) + dataShift; + + // update state + if (inc_totalUsers(mvar) != 0) { + INTERPROCESS_LOG_DEBUG( + "Found that the mvar was already destroyed when updating the total " + "users counter.\n"); + goto failed_after_mmap; + } + return mvar; + +failed_after_mmap: + munmap(mvar->statePtr, totalSize); +failed_after_mvar_allocated: + free(mvar); +failed: + return NULL; +} + +void mvar_destroy(MVar *mvar) { + size_t storeSize = mvar->statePtr->dataSize + mvar_state_size64(); + int usersLeft = atomic_fetch_sub_explicit(&(mvar->statePtr->totalUsers), 1, + memory_order_relaxed) - + 1; + if (usersLeft > 0) { + INTERPROCESS_LOG_DEBUG( + "Destroying local instance of mvar %s, %d users left.\n", + mvar->mvarName, usersLeft); + munmap(mvar->statePtr, storeSize); + } else { + INTERPROCESS_LOG_DEBUG( + "Destroying mvar %s globally (no other users left).\n", mvar->mvarName); + pthread_cond_destroy(&(mvar->statePtr->canTakeC)); + pthread_cond_destroy(&(mvar->statePtr->canPutC)); + pthread_mutex_destroy(&(mvar->statePtr->mvMut)); + munmap(mvar->statePtr, storeSize); + shm_unlink(mvar->mvarName); + } + free(mvar); +} + +static inline void finilize_USE_MUTEX(pthread_mutex_t **mp) { + pthread_mutex_unlock(*mp); +} + +#define USE_MUTEX(state) \ + { \ + int r = 0; \ + r = pthread_mutex_lock(&(state->mvMut)); \ + if (r != 0) return r; \ + } \ + __attribute__((cleanup(finilize_USE_MUTEX))) pthread_mutex_t *_mut = \ + &(state->mvMut); + +int interprocess_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mut, + int millis) { + struct timespec time_to_wait; + clock_gettime(CLOCK_REALTIME, &time_to_wait); + ldiv_t t = ldiv(millis * 1000 + (time_to_wait.tv_nsec / 1000), 1000000); + time_to_wait.tv_sec += t.quot; + time_to_wait.tv_nsec = t.rem * 1000; + int r = pthread_cond_timedwait(cond, mut, &time_to_wait); +#ifdef __APPLE__ + // Workaround for macOS failing in timedwait for absolutely no reason + if (r == EINVAL) { + r = pthread_mutex_unlock(mut); + if (r != 0) return r; + t = ldiv(millis, 1000); + time_to_wait.tv_sec = t.quot; + time_to_wait.tv_nsec = t.rem * 1000000; + nanosleep(&time_to_wait, NULL); + r = pthread_mutex_lock(mut); + } +#endif + return r; +} + +#define WAIT_A_BIT(cond, mut, ms) \ + { \ + int r = interprocess_cond_timedwait(cond, mut, ms); \ + if (r != 0 && r != ETIMEDOUT) return r; \ + } + +int mvar_take(MVar *mvar, void *localDataPtr, StgStablePtr tso) { + MVarState *state = mvar->statePtr; + USE_MUTEX(state); + while (!(state->isFull) || state->pendingReaders > 0) { + if (state->pendingReaders > 0) pthread_cond_broadcast(&(state->canTakeC)); + WAIT_A_BIT(&(state->canTakeC), &(state->mvMut), state->maxWaitMs); + if (has_blocked_exceptions(tso)) return EINTR; + } + memcpy(localDataPtr, mvar->dataPtr, state->dataSize); + state->isFull = 0; + pthread_cond_signal(&(state->canPutC)); + return 0; +} + +int mvar_trytake(MVar *mvar, void *localDataPtr) { + MVarState *state = mvar->statePtr; + USE_MUTEX(state); + // shortcut if is empty + if (!(state->isFull)) return 1; + while (state->pendingReaders > 0) { + // make sure readers do not sleep + pthread_cond_broadcast(&(state->canTakeC)); + // last reader should wake me up, wait for it. + WAIT_A_BIT(&(state->canTakeC), &(state->mvMut), state->maxWaitMs); + // repeat emptyness check (if another trytake was faster) + if (!(state->isFull)) { + return 1; + } + } + memcpy(localDataPtr, mvar->dataPtr, state->dataSize); + state->isFull = 0; + pthread_cond_signal(&(state->canPutC)); + return 0; +} + +int mvar_put(MVar *mvar, void *localDataPtr, StgStablePtr tso) { + MVarState *state = mvar->statePtr; + USE_MUTEX(state); + while (state->isFull) { + WAIT_A_BIT(&(state->canPutC), &(state->mvMut), state->maxWaitMs); + if (has_blocked_exceptions(tso)) return EINTR; + } + memcpy(mvar->dataPtr, localDataPtr, state->dataSize); + state->isFull = 1; + pthread_cond_broadcast(&(state->canTakeC)); + return 0; +} + +int mvar_tryput(MVar *mvar, void *localDataPtr) { + MVarState *state = mvar->statePtr; + USE_MUTEX(state); + // shortcut if is full + if (state->isFull) return 1; + memcpy(mvar->dataPtr, localDataPtr, state->dataSize); + state->isFull = 1; + pthread_cond_broadcast(&(state->canTakeC)); + return 0; +} + +// Split mvar_read/worker to make sure pendingReaders counter is never leaked +static inline int mvar_read_loop(MVarState *state, StgStablePtr tso) { + while (!(state->isFull)) { + WAIT_A_BIT(&(state->canTakeC), &(state->mvMut), state->maxWaitMs); + if (has_blocked_exceptions(tso)) return EINTR; + } + return 0; +} + +int mvar_read(MVar *mvar, void *localDataPtr, StgStablePtr tso) { + MVarState *state = mvar->statePtr; + USE_MUTEX(state); + state->pendingReaders++; + int r = mvar_read_loop(state, tso); + if (r == 0) memcpy(localDataPtr, mvar->dataPtr, state->dataSize); + if ((--(state->pendingReaders)) == 0) pthread_cond_signal(&(state->canTakeC)); + return r; +} + +int mvar_tryread(MVar *mvar, void *localDataPtr) { + MVarState *state = mvar->statePtr; + USE_MUTEX(state); + // shortcut if is empty + if (!(state->isFull)) return 1; + memcpy(localDataPtr, mvar->dataPtr, state->dataSize); + return 0; +} + +int mvar_swap(MVar *mvar, void *inPtr, void *outPtr, StgStablePtr tso) { + MVarState *state = mvar->statePtr; + USE_MUTEX(state); + while (!(state->isFull) || state->pendingReaders > 0) { + if (state->pendingReaders > 0) { + pthread_cond_broadcast(&(state->canTakeC)); + } + WAIT_A_BIT(&(state->canTakeC), &(state->mvMut), state->maxWaitMs); + if (has_blocked_exceptions(tso)) return EINTR; + } + memcpy(outPtr, mvar->dataPtr, state->dataSize); + memcpy(mvar->dataPtr, inPtr, state->dataSize); + pthread_cond_signal(&(state->canTakeC)); + return 0; +} + +int mvar_tryswap(MVar *mvar, void *inPtr, void *outPtr) { + MVarState *state = mvar->statePtr; + USE_MUTEX(state); + // shortcut if is empty + if (!(state->isFull)) return 1; + while (state->pendingReaders > 0) { + // make sure readers do not sleep + pthread_cond_broadcast(&(state->canTakeC)); + // last reader should wake me up, wait for it. + WAIT_A_BIT(&(state->canTakeC), &(state->mvMut), state->maxWaitMs); + // repeat emptyness check (if another trytake was faster) + if (!(state->isFull)) return 1; + } + memcpy(outPtr, mvar->dataPtr, state->dataSize); + memcpy(mvar->dataPtr, inPtr, state->dataSize); + pthread_cond_signal(&(state->canPutC)); + return 0; +} + +int mvar_isempty(MVar *mvar) { return mvar->statePtr->isFull == 0; } + +void mvar_name(MVar *mvar, char *const name) { strcpy(name, mvar->mvarName); } diff --git a/src/Control/Concurrent/Process/StoredMVarPrim.cmm b/src/Control/Concurrent/Process/StoredMVarPrim.cmm new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Process/StoredMVarPrim.cmm @@ -0,0 +1,37 @@ +#include "Cmm.h" + +cmm_mvar_take(W_ mvar, W_ localDataPtr) { + W_ sptr; + CInt r; + ("ptr" sptr) = ccall getStablePtr(CurrentTSO "ptr") "unsafe"; + (r) = ccall mvar_take (mvar "ptr", localDataPtr "ptr", sptr "ptr") "safe"; + ccall freeStablePtr (sptr "ptr") "unsafe"; + return (TO_W_(r)); +} + +cmm_mvar_put(W_ mvar, W_ localDataPtr) { + W_ sptr; + CInt r; + ("ptr" sptr) = ccall getStablePtr(CurrentTSO "ptr") "unsafe"; + (r) = ccall mvar_put (mvar "ptr", localDataPtr "ptr", sptr "ptr") "safe"; + ccall freeStablePtr (sptr "ptr") "unsafe"; + return (TO_W_(r)); +} + +cmm_mvar_read(W_ mvar, W_ localDataPtr) { + W_ sptr; + CInt r; + ("ptr" sptr) = ccall getStablePtr(CurrentTSO "ptr") "unsafe"; + (r) = ccall mvar_read (mvar "ptr", localDataPtr "ptr", sptr "ptr") "safe"; + ccall freeStablePtr (sptr "ptr") "unsafe"; + return (TO_W_(r)); +} + +cmm_mvar_swap(W_ mvar, W_ inPtr, W_ outPtr) { + W_ sptr; + CInt r; + ("ptr" sptr) = ccall getStablePtr(CurrentTSO "ptr") "unsafe"; + (r) = ccall mvar_swap (mvar "ptr", inPtr "ptr", outPtr "ptr", sptr "ptr") "safe"; + ccall freeStablePtr (sptr "ptr") "unsafe"; + return (TO_W_(r)); +} diff --git a/src/Control/Concurrent/Process/StoredMVarWin32.c b/src/Control/Concurrent/Process/StoredMVarWin32.c new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Process/StoredMVarWin32.c @@ -0,0 +1,538 @@ +#include + +#include +#include +#include +#include +#include + +#ifndef WAIT_OBJECT_1 +#define WAIT_OBJECT_1 ((WAIT_OBJECT_0) + 1) +#endif + +#ifndef WAIT_ABANDONED_1 +#define WAIT_ABANDONED_1 ((WAIT_ABANDONED_0) + 1) +#endif + +#ifdef INTERPROCESS_DEBUG +#define ASSERT_TRUE(op) assert(op) +#else +#define ASSERT_TRUE(op) op +#endif + +typedef struct MVar MVar; + +MVar *mvar_new(size_t byteSize, int max_wait_ms); +MVar *mvar_lookup(const char *name); +void mvar_destroy(MVar *mvar); + +int mvar_take(MVar *mvar, void *localDataPtr, StgStablePtr tso); +int mvar_trytake(MVar *mvar, void *localDataPtr); +int mvar_put(MVar *mvar, void *localDataPtr, StgStablePtr tso); +int mvar_tryput(MVar *mvar, void *localDataPtr); +int mvar_read(MVar *mvar, void *localDataPtr, StgStablePtr tso); +int mvar_tryread(MVar *mvar, void *localDataPtr); +int mvar_swap(MVar *mvar, void *inPtr, void *outPtr, StgStablePtr tso); +int mvar_tryswap(MVar *mvar, void *inPtr, void *outPtr); +int mvar_isempty(MVar *mvar); + +typedef struct MVarState { + size_t dataSize; + /** + * Maximum wait time in milliseconds. + * This determines maximum possible delay the mvar operation cancels in + * the event of an async exception thrown to the calling Haskell thread. + */ + int maxWaitMs; + atomic_int pendingReaders; +} MVarState; + +typedef struct MVar { + /** Manual-reset event: takers and readers wait on this. */ + HANDLE canTakeE; + /** + * Mutex: guard the critical region executed by multiple processes. + * It's placed between the events to allow easy passing of the + * mutex+event pair to the WaitForMultipleObjects function. + */ + HANDLE guardM; + /** Manual-reset event: putters wait on this. */ + HANDLE canPutE; + /** FileMapping: keep the data by this handle. */ + HANDLE dataStoreFM; + /** + * State is stored in the shared data area, accessed by all processes. + * It has a fixed size and kept at the beginning of a shared memory region. + */ + MVarState *statePtr; + /** Actual data is stored next to the MVarState */ + void *dataPtr; + /** + * Secondary objects are contstructed by appending a single char to the name. + */ + SharedObjectName mvarName; +} MVar; + +/** Get the pending readers counter value. */ +static inline int get_pendingReaders(MVar *mvar) { + return atomic_load_explicit(&(mvar->statePtr->pendingReaders), + memory_order_relaxed); +} + +/** Set the pending readers counter value to zero. */ +static inline int reset_pendingReaders(MVar *mvar) { + atomic_store_explicit(&(mvar->statePtr->pendingReaders), 0, + memory_order_relaxed); +} + +/** Atomically increment the pending readers counter. */ +static inline void inc_pendingReaders(MVar *mvar) { + atomic_fetch_add_explicit(&(mvar->statePtr->pendingReaders), 1, + memory_order_relaxed); +} + +/** + * Atomically decrement the pending readers counter, protected against + * decrementing below zero. + */ +static inline void dec_pendingReaders(MVar *mvar) { + atomic_int *p = &(mvar->statePtr->pendingReaders); + int exp = atomic_load_explicit(p, memory_order_relaxed); + while (exp >= 1) { + if (atomic_compare_exchange_weak_explicit( + p, &exp, exp - 1, memory_order_relaxed, memory_order_relaxed)) + return; + } +} + +/** Check if an event is in the signalled state. */ +static inline bool is_signalled(HANDLE event) { + DWORD r = WaitForSingleObject(event, 0); +#ifdef INTERPROCESS_DEBUG + if (r != WAIT_TIMEOUT && r != WAIT_OBJECT_0) + INTERPROCESS_LOG_DEBUG( + "Warning: WaitForSingleObject produced unexpected return code %d; " + "error code %d. Continue as if the object is not signalled...\n", + r, GetLastError()); +#endif + return r == WAIT_OBJECT_0; +} + +/** + * @brief Check the result of waiting and possibly release the mutex. + * + * @param [in] debug_name provide a text name for debug logging + * @param [in] mvar MVar + * @param [in] code wait return code + * @param [in] primaryE primary event (canTakeE, canPutE, or NULL as default - + * signalled). + * @param [in] secondaryE secondary event (canPutE, canTakeR, or NULL as default + * - non-signalled). + * @return true The mutex is owned; the wait is successful and the primary event + * is signalled. + * @return false The mutex is not owned; the wait failed or the primary event is + * not signalled. + */ +static inline bool interpret_wait(const char *debug_name, MVar *mvar, + DWORD code, HANDLE primaryE, + HANDLE secondaryE) { + bool primaryR = true, secondaryR = false; + switch (code) { + // success, but another process is crashed + case WAIT_ABANDONED_0: + case WAIT_ABANDONED_1: + secondaryR = secondaryE != NULL && is_signalled(secondaryE); + + // success + case WAIT_OBJECT_0: + case WAIT_OBJECT_1: + primaryR = primaryE == NULL || is_signalled(primaryE); + if (secondaryR && primaryR) { + /* + If a process crashes while executing one of the MVar operations, + the state of MVar may become corrupted. This may be the case if the + mutex is in the abandoned state. + To recover the state of the MVar, we look at the two events, which + are carefully set and reset in an order that guarantees the following: + + canTakeE canPutE data state + + + valid incorrect + + - valid correct + - + ? correct + - - ? impossible + + Also the `pendingReaders` counter may be invalid, but the `take` + functions detect and reset that. + + The enclosing `if` block can be entered only when the mutex is + abandoned by another thread, and we can reset either of the two + events, because the data inside is valid. We choose to reset the + secondary event to reduce the waiting time and the chance of + deadlocks. + */ + ASSERT_TRUE(ResetEvent(secondaryE)); + } + if (!primaryR) { + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + } + return primaryR; + + // proper failure + case WAIT_FAILED: + INTERPROCESS_LOG_DEBUG("Error in %s: wait failed; error code %d.\n", + debug_name, GetLastError()); + return false; + + // bug-like failure + default: + INTERPROCESS_LOG_DEBUG( + "Error in %s: unexpected return code %d; error code %d.\n", + debug_name, code, GetLastError()); + ReleaseMutex(mvar->guardM); + return false; + } +} + +/** Copy the data from the mvar and switch the events. */ +static inline void load_data(MVar *mvar, void *ptr) { + /* + The order of set/reset event is important; it ensures that canTakeE and + canPutE are never both non-signalled to avoid deadlocks. If a process + crashes between SetEvent and ResetEvent, other processes can detect and fix + that. See the note in `interpret_wait` for more details. + */ + memcpy(ptr, mvar->dataPtr, mvar->statePtr->dataSize); + ASSERT_TRUE(SetEvent(mvar->canPutE)); + ASSERT_TRUE(ResetEvent(mvar->canTakeE)); +} + +/** Copy the data to the mvar and switch the events. */ +static inline void save_data(MVar *mvar, void *ptr) { + // See the note in `load_data` + memcpy(mvar->dataPtr, ptr, mvar->statePtr->dataSize); + ASSERT_TRUE(SetEvent(mvar->canTakeE)); + ASSERT_TRUE(ResetEvent(mvar->canPutE)); +} + +/** + * @brief Call `WaitForMultipleObjects` interrupting on exceptions in a calling + * Haskell thread. + * + * This function calls `WaitForMultipleObjects` in a loop with a timeout defined + * at the mvar creation. After each iteration, it checks if there are any + * pending blocked exceptions in the calling Haskell thread. If that is the + * case, it sets the `LastError` to `EINTR` and returns `WAIT_FAILED`. You can + * view this function as `WaitForMultipleObjects` with the infinite timeout. + * + * @param mvar + * @param tso Stable pointer for the Haskell thread state object (StgTSO). + * @param handles Two waitable objects, one of which is a mutex. + * @return DWORD Return code of `WaitForMultipleObjects`. + */ +static inline DWORD wait_interruptible(MVar *mvar, StgStablePtr tso, + HANDLE *handles) { + DWORD r; + do { + r = WaitForMultipleObjects(2, handles, TRUE, mvar->statePtr->maxWaitMs); + if (r == WAIT_TIMEOUT && has_blocked_exceptions(tso)) { + SetLastError(EINTR); + return WAIT_FAILED; + } + } while (r == WAIT_TIMEOUT); + return r; +} + +// ensure 64 byte alignment (maximum possible we can think of, e.g. AVX512) +static inline size_t mvar_state_size64() { + size_t x = sizeof(MVarState); + return (x + 63) & (~63); +} + +MVar *mvar_new(size_t byteSize, int max_wait_ms) { + // allocate MVar + MVar *mvar = malloc(sizeof(*mvar)); + if (mvar == NULL) goto failed; + genSharedObjectName(mvar->mvarName); + + size_t dataShift = mvar_state_size64(); + size_t totalSize = dataShift + byteSize; + mvar->dataStoreFM = CreateFileMappingA( + /* use paging file */ + INVALID_HANDLE_VALUE, + /* default security */ + NULL, + /* read/write access */ + PAGE_READWRITE, + /* maximum object size (high-order DWORD) */ + (DWORD)(totalSize >> 32), + /* maximum object size (low-order DWORD) */ + (DWORD)(totalSize & 0xFFFFFFFF), + /* name of mapping object */ + mvar->mvarName); + + if (mvar->dataStoreFM == NULL) { + INTERPROCESS_LOG_DEBUG("Could not create file mapping object (%d).\n", + GetLastError()); + goto failed_after_mvar_allocated; + } + mvar->statePtr = (MVarState *)MapViewOfFile( + /* handle to map object */ + mvar->dataStoreFM, + /* read/write permission */ + FILE_MAP_ALL_ACCESS, 0, 0, totalSize); + + if (mvar->statePtr == NULL) { + INTERPROCESS_LOG_DEBUG("Could not map view of file (%d).\n", + GetLastError()); + goto failed_after_fm_created; + } + mvar->dataPtr = ((void *)(mvar->statePtr)) + dataShift; + + // create all synchronization objects + SharedObjectName objName = {0}; + strcpy(objName, mvar->mvarName); + strcat(objName, "T"); + mvar->canTakeE = CreateEventA(NULL, TRUE, FALSE, objName); + if (mvar->canTakeE == NULL) { + INTERPROCESS_LOG_DEBUG("Could not create canTakeE (%d).\n", GetLastError()); + goto failed_after_fm_mapped; + } + strcpy(objName, mvar->mvarName); + strcat(objName, "P"); + mvar->canPutE = CreateEventA(NULL, TRUE, TRUE, objName); + if (mvar->canPutE == NULL) { + INTERPROCESS_LOG_DEBUG("Could not create canPutE (%d).\n", GetLastError()); + goto failed_after_cantake; + } + strcpy(objName, mvar->mvarName); + strcat(objName, "M"); + mvar->guardM = CreateMutexA(NULL, FALSE, objName); + if (mvar->guardM == NULL) { + INTERPROCESS_LOG_DEBUG("Could not create guardM (%d).\n", GetLastError()); + goto failed_after_canput; + } + + // zero readers pending + *(mvar->statePtr) = (struct MVarState){ + .dataSize = byteSize, .maxWaitMs = max_wait_ms, .pendingReaders = 0}; + + return mvar; + +failed_after_canput: + CloseHandle(mvar->canPutE); +failed_after_cantake: + CloseHandle(mvar->canTakeE); +failed_after_fm_mapped: + UnmapViewOfFile(mvar->statePtr); +failed_after_fm_created: + CloseHandle(mvar->dataStoreFM); +failed_after_mvar_allocated: + free(mvar); +failed: + return NULL; +} + +MVar *mvar_lookup(const char *name) { + // allocate MVar + MVar *mvar = malloc(sizeof(*mvar)); + if (mvar == NULL) goto failed; + strcpy(mvar->mvarName, name); + + mvar->dataStoreFM = + OpenFileMappingA(FILE_MAP_ALL_ACCESS, FALSE, mvar->mvarName); + + if (mvar->dataStoreFM == NULL) { + INTERPROCESS_LOG_DEBUG("Could not open file mapping object (%d).\n", + GetLastError()); + goto failed_after_mvar_allocated; + } + + size_t dataShift = mvar_state_size64(); + mvar->statePtr = (MVarState *)MapViewOfFile( + /* handle to map object */ + mvar->dataStoreFM, + /* First time just read the header */ + FILE_MAP_READ, 0, 0, dataShift); + + if (mvar->statePtr == NULL) { + INTERPROCESS_LOG_DEBUG("Could not map view of file (%d).\n", + GetLastError()); + goto failed_after_fm_created; + } + size_t totalSize = mvar->statePtr->dataSize + dataShift; + if (UnmapViewOfFile(mvar->statePtr) == 0) { + INTERPROCESS_LOG_DEBUG("Could not unmap view of file during lookup (%d).\n", + GetLastError()); + } + + mvar->statePtr = (MVarState *)MapViewOfFile( + /* handle to map object */ + mvar->dataStoreFM, + /* read/write permission */ + FILE_MAP_ALL_ACCESS, 0, 0, totalSize); + + if (mvar->statePtr == NULL) { + INTERPROCESS_LOG_DEBUG("Could not map view of file (%d).\n", + GetLastError()); + goto failed_after_fm_created; + } + mvar->dataPtr = (void *)(mvar->statePtr) + dataShift; + + // lookup all synchronization objects + SharedObjectName objName = {0}; + strcpy(objName, mvar->mvarName); + strcat(objName, "T"); + mvar->canTakeE = OpenEventA(EVENT_MODIFY_STATE | SYNCHRONIZE, FALSE, objName); + if (mvar->canTakeE == NULL) { + INTERPROCESS_LOG_DEBUG("Could not open canTakeE (%d).\n", GetLastError()); + goto failed_after_fm_mapped; + } + strcpy(objName, mvar->mvarName); + strcat(objName, "P"); + mvar->canPutE = OpenEventA(EVENT_MODIFY_STATE | SYNCHRONIZE, FALSE, objName); + if (mvar->canPutE == NULL) { + INTERPROCESS_LOG_DEBUG("Could not open canPutE (%d).\n", GetLastError()); + goto failed_after_cantake; + } + strcpy(objName, mvar->mvarName); + strcat(objName, "M"); + mvar->guardM = OpenMutexA(SYNCHRONIZE, FALSE, objName); + if (mvar->guardM == NULL) { + INTERPROCESS_LOG_DEBUG("Could not open guardM (%d).\n", GetLastError()); + goto failed_after_canput; + } + + return mvar; + +failed_after_canput: + CloseHandle(mvar->canPutE); +failed_after_cantake: + CloseHandle(mvar->canTakeE); +failed_after_fm_mapped: + UnmapViewOfFile(mvar->statePtr); +failed_after_fm_created: + CloseHandle(mvar->dataStoreFM); +failed_after_mvar_allocated: + free(mvar); +failed: + return NULL; +} + +void mvar_destroy(MVar *mvar) { + CloseHandle(mvar->canTakeE); + CloseHandle(mvar->canPutE); + CloseHandle(mvar->guardM); + UnmapViewOfFile(mvar->statePtr); + CloseHandle(mvar->dataStoreFM); + free(mvar); +} + +int mvar_take(MVar *mvar, void *localDataPtr, StgStablePtr tso) { + int pr_prev = INT_MAX, pr_cur; + do { + if (!interpret_wait("mvar_take", mvar, + wait_interruptible(mvar, tso, &(mvar->canTakeE)), NULL, + mvar->canPutE)) + return 1; + pr_cur = get_pendingReaders(mvar); + if (pr_cur >= pr_prev) { + pr_cur == 0; + reset_pendingReaders(mvar); + } + pr_prev = pr_cur; + if (pr_cur == 0) load_data(mvar, localDataPtr); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + } while (pr_cur != 0); + return 0; +} + +int mvar_trytake(MVar *mvar, void *localDataPtr) { + int pr_prev = INT_MAX, pr_cur; + do { + if (!interpret_wait("mvar_trytake", mvar, + WaitForSingleObject(mvar->guardM, INFINITE), + mvar->canTakeE, mvar->canPutE)) + return 1; + pr_cur = get_pendingReaders(mvar); + if (pr_cur >= pr_prev) { + pr_cur == 0; + reset_pendingReaders(mvar); + } + pr_prev = pr_cur; + if (pr_cur == 0) load_data(mvar, localDataPtr); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + } while (pr_cur != 0); + return 0; +} + +int mvar_put(MVar *mvar, void *localDataPtr, StgStablePtr tso) { + if (!interpret_wait("mvar_put", mvar, + wait_interruptible(mvar, tso, &(mvar->guardM)), NULL, + mvar->canTakeE)) + return 1; + save_data(mvar, localDataPtr); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + return 0; +} + +int mvar_tryput(MVar *mvar, void *localDataPtr) { + if (!interpret_wait("mvar_tryput", mvar, + WaitForSingleObject(mvar->guardM, INFINITE), + mvar->canPutE, mvar->canTakeE)) + return 1; + save_data(mvar, localDataPtr); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + return 0; +} + +int mvar_read(MVar *mvar, void *localDataPtr, StgStablePtr tso) { + DWORD r = WaitForSingleObject(mvar->guardM, INFINITE); + if (r == WAIT_OBJECT_0 || r == WAIT_ABANDONED_0) { + if (!is_signalled(mvar->canTakeE)) { + inc_pendingReaders(mvar); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + r = wait_interruptible(mvar, tso, &(mvar->canTakeE)); + dec_pendingReaders(mvar); + } + } + if (!interpret_wait("mvar_read", mvar, r, NULL, mvar->canPutE)) return 1; + memcpy(localDataPtr, mvar->dataPtr, mvar->statePtr->dataSize); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + return 0; +} + +int mvar_tryread(MVar *mvar, void *localDataPtr) { + if (!interpret_wait("mvar_tryread", mvar, + WaitForSingleObject(mvar->guardM, INFINITE), + mvar->canTakeE, mvar->canPutE)) + return 1; + memcpy(localDataPtr, mvar->dataPtr, mvar->statePtr->dataSize); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + return 0; +} + +int mvar_swap(MVar *mvar, void *inPtr, void *outPtr, StgStablePtr tso) { + if (!interpret_wait("mvar_swap", mvar, + wait_interruptible(mvar, tso, &(mvar->canTakeE)), NULL, + mvar->canPutE)) + return 1; + load_data(mvar, outPtr); + save_data(mvar, inPtr); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + return 0; +} + +int mvar_tryswap(MVar *mvar, void *inPtr, void *outPtr) { + if (!interpret_wait("mvar_tryswap", mvar, + WaitForSingleObject(mvar->guardM, INFINITE), + mvar->canTakeE, mvar->canPutE)) + return 1; + load_data(mvar, outPtr); + save_data(mvar, inPtr); + ASSERT_TRUE(ReleaseMutex(mvar->guardM)); + return 0; +} + +int mvar_isempty(MVar *mvar) { return is_signalled(mvar->canTakeE) ? 0 : 1; } + +void mvar_name(MVar *mvar, char *const name) { strcpy(name, mvar->mvarName); } diff --git a/src/Foreign/SharedObjectName/Internal.hs b/src/Foreign/SharedObjectName/Internal.hs --- a/src/Foreign/SharedObjectName/Internal.hs +++ b/src/Foreign/SharedObjectName/Internal.hs @@ -18,8 +18,8 @@ import Text.Read #define HS_IMPORT_CONSTANTS_ONLY -#include "SharedObjectName.h" #include "MachDeps.h" +#include "common.h" -- | Reference to a shared object; can be sent to other processes. newtype SOName a = SOName (ForeignPtr CChar) diff --git a/src/Foreign/SharedPtr.c b/src/Foreign/SharedPtr.c --- a/src/Foreign/SharedPtr.c +++ b/src/Foreign/SharedPtr.c @@ -10,7 +10,7 @@ */ #include "SharedPtr.h" -#include + #include #include diff --git a/src/Foreign/SharedPtrWin32.c b/src/Foreign/SharedPtrWin32.c --- a/src/Foreign/SharedPtrWin32.c +++ b/src/Foreign/SharedPtrWin32.c @@ -1,19 +1,16 @@ #include "SharedPtr.h" -#ifndef NDEBUG -#include -#endif -void _SharedMutex_init(SharedMutex *mptr, void **privateMutexHandle, const int createNew) { +void _SharedMutex_init(SharedMutex *mptr, void **privateMutexHandle, + const int createNew) { if (createNew != 0) { genSharedObjectName(mptr->mutexName); } - *privateMutexHandle = CreateMutexA - ( NULL // default security attributes - , FALSE // initially not owned - , mptr->mutexName ); -#ifndef NDEBUG + *privateMutexHandle = + CreateMutexA(/** default security attributes*/ NULL, + /** initially not owned */ FALSE, mptr->mutexName); +#ifdef INTERPROCESS_DEBUG if (*privateMutexHandle == NULL) { - printf("CreateMutex error: %d\n", GetLastError()); + INTERPROCESS_LOG_DEBUG("CreateMutex error: %d\n", GetLastError()); } #endif } @@ -25,9 +22,9 @@ int _SharedMutex_lock(SharedMutex *mptr, void **privateMutexHandle) { DWORD r = WaitForSingleObject(*privateMutexHandle, INFINITE); if (r != WAIT_OBJECT_0) { -#ifndef NDEBUG - printf("WaitForSingleObject mutex error: return %d; error code %d.\n", r, GetLastError()); -#endif + INTERPROCESS_LOG_DEBUG( + "WaitForSingleObject mutex error: return %d; error code %d.\n", r, + GetLastError()); return 1; } else { return 0; @@ -36,43 +33,37 @@ int _SharedMutex_unlock(SharedMutex *mptr, void **privateMutexHandle) { DWORD r = ReleaseMutex(*privateMutexHandle); - if(r == 0) { -#ifndef NDEBUG - printf("ReleaseMutex error: %d\n", GetLastError()); -#endif - return 1; + if (r == 0) { + INTERPROCESS_LOG_DEBUG("ReleaseMutex error: %d\n", GetLastError()); + return 1; } else { - return 0; + return 0; } } // returns NULL if failed -HsPtr _store_alloc(const char *memBlockName, void **privateMutexHandle, size_t size){ - *privateMutexHandle = CreateFileMappingA - ( INVALID_HANDLE_VALUE // use paging file - , NULL // default security - , PAGE_READWRITE // read/write access - , (DWORD)(size >> 32) // maximum object size (high-order DWORD) - , (DWORD)(size & 0xFFFFFFFF) // maximum object size (low-order DWORD) - , memBlockName); // name of mapping object +HsPtr _store_alloc(const char *memBlockName, void **privateMutexHandle, + size_t size) { + *privateMutexHandle = CreateFileMappingA( + /** use paging file */ INVALID_HANDLE_VALUE, + /** default security */ NULL, + /** read/write access */ PAGE_READWRITE, + /** maximum object size (high-order DWORD) */ (DWORD)(size >> 32), + /** maximum object size (low-order DWORD) */ (DWORD)(size & 0xFFFFFFFF), + /** name of mapping object */ memBlockName); if (*privateMutexHandle == NULL) { -#ifndef NDEBUG - printf("Could not create file mapping object (%d).\n", GetLastError()); -#endif + INTERPROCESS_LOG_DEBUG("Could not create file mapping object (%d).\n", + GetLastError()); return NULL; } - void *rptr = MapViewOfFile - ( *privateMutexHandle // handle to map object - , FILE_MAP_ALL_ACCESS // read/write permission - , 0 - , 0 - , size); + void *rptr = MapViewOfFile(/** handle to map object */ *privateMutexHandle, + /** read/write permission */ FILE_MAP_ALL_ACCESS, + 0, 0, size); if (rptr == NULL) { -#ifndef NDEBUG - printf("Could not map view of file (%d).\n", GetLastError()); -#endif + INTERPROCESS_LOG_DEBUG("Could not map view of file (%d).\n", + GetLastError()); CloseHandle(*privateMutexHandle); return NULL; } @@ -81,9 +72,9 @@ } // failures of system calls are ignored -void _store_free( const char *memBlockName, void **privateStoreHandle, HsPtr addr - , size_t size, _Bool unlinkToo) { - if(addr != 0){ +void _store_free(const char *memBlockName, void **privateStoreHandle, + HsPtr addr, size_t size, _Bool unlinkToo) { + if (addr != 0) { UnmapViewOfFile(addr); CloseHandle(*privateStoreHandle); } diff --git a/test/StoredMVar.hs b/test/StoredMVar.hs --- a/test/StoredMVar.hs +++ b/test/StoredMVar.hs @@ -1,30 +1,11 @@ -{-# LANGUAGE CPP #-} +{-# LANGUAGE TupleSections #-} module Main (main) where -import Control.Concurrent (forkIO) -import qualified Control.Concurrent.MVar as Vanilla -import Control.Concurrent.Process.StoredMVar -import Control.Exception -import Control.Monad (forM, void) -import Data.Monoid (First (..), Monoid (..)) -#if __GLASGOW_HASKELL__ >= 800 -import Data.Semigroup (Semigroup (..)) -#endif -import Foreign.SharedObjectName -import System.Environment -import System.Exit -import System.IO -import System.IO.Unsafe -import System.Mem -import System.Process.Typed -import Text.Read (readMaybe) - --- | A number of processes trying to do something concurrently. --- For example, read from the same StoredMVar. --- Actual number of processes may vary depending on a test case. --- Minimum allowed number is 1. -concProcBaseN :: Int -concProcBaseN = 20 +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async +import Control.Concurrent.Process.StoredMVar +import Tools.Runner +import Tools.TestResult data BasicRole = Master | Slave deriving (Eq, Ord, Show, Read) @@ -32,145 +13,81 @@ data ThreeWayRole = Reader | Taker | Putter deriving (Eq, Ord, Show, Read) -data TestSpec - = SimpleTakePut (SOName (StoredMVar Double)) BasicRole - | ReadersTakers (SOName (StoredMVar Double)) ThreeWayRole - deriving (Eq, Ord, Show, Read) - - -run :: TestSpec -> IO TestResult - -run (SimpleTakePut ref Master) = do - mVar <- lookupMVar ref - putMVar mVar 42 - putMVar mVar 17 - return Success -run (SimpleTakePut ref Slave) = do - mVar <- lookupMVar ref - a <- takeMVar mVar - b <- takeMVar mVar - return $ if (a + b) == (42 + 17) - then Success - else Failure $ show (a + b) ++ " /= 42 + 17" - -run (ReadersTakers ref Putter) = do - mVar <- lookupMVar ref - putMVar mVar 177 - putMVar mVar 178 - putMVar mVar 179 - putMVar mVar 777 - return Success -run (ReadersTakers ref Taker) = do - mVar <- lookupMVar ref - a <- takeMVar mVar - b <- takeMVar mVar - c <- takeMVar mVar - putStrLn $ "Taking: " ++ show (a,b,c) - return $ if a < b && b < c - then Success - else Failure "Three taken numbers must go ordered!" -run (ReadersTakers ref Reader) = do - mVar <- lookupMVar ref - a <- readMVar mVar - b <- readMVar mVar - c <- readMVar mVar - putStrLn $ "Reading: " ++ show (a,b,c) - return $ if a <= b && b <= c - then Success - else Failure "Three taken numbers must go ordered!" - -tests :: [IO ([TestSpec], IO ())] -tests = - [ do - mVar <- newEmptyMVar - return - ( [ SimpleTakePut (mVarName mVar) Slave - , SimpleTakePut (mVarName mVar) Master - ] - , return (mVar `seq` ()) - ) - , do - mVar <- newEmptyMVar - return - ( replicate concProcBaseN (ReadersTakers (mVarName mVar) Reader) - ++ - [ ReadersTakers (mVarName mVar) Taker - , ReadersTakers (mVarName mVar) Putter - ] - , return (mVar `seq` ()) - ) +simpleTakePut :: TestSpec +simpleTakePut = Repeat 100 $ TestSpec "SimpleTakePut" + [ (Master, run) + , (Slave, run) ] + where + run :: BasicRole -> StoredMVar Double -> IO TestResult + run Master mVar = do + putMVar mVar 42 + putMVar mVar 17 + return Success + run Slave mVar = do + a <- takeMVar mVar + b <- takeMVar mVar + return $ + if (a + b) == (42 + 17) + then Success + else Failure $ show (a + b) ++ " /= 42 + 17" +readersTakers :: TestSpec +readersTakers = Repeat 100 $ TestSpec "ReadersTakers" $ + (, run) <$> (replicate 20 Reader <> [Taker, Putter]) + where + run :: ThreeWayRole -> StoredMVar Int -> IO TestResult + run Putter mVar = do + let putLoud x = do + putStrLn $ "Putting " ++ show x ++ "..." + putMVar mVar x + putLoud 177 + putLoud 178 + putLoud 179 + putLoud 777 + return Success + run Taker mVar = do + a <- takeMVar mVar + b <- takeMVar mVar + c <- takeMVar mVar + putStrLn $ "Taking: " ++ show (a, b, c) + return $ + if a < b && b < c + then Success + else Failure "Three taken numbers must go ordered!" + run Reader mVar = do + a <- readMVar mVar + b <- readMVar mVar + c <- readMVar mVar + putStrLn $ "Reading: " ++ show (a, b, c) + return $ + if a <= b && b <= c + then Success + else Failure "Three taken numbers must go ordered!" -data TestResult - = Success - | Failure String - deriving (Eq, Ord, Show, Read) -#if __GLASGOW_HASKELL__ >= 800 -instance Semigroup TestResult where - (<>) = mappend -#endif - -instance Monoid TestResult where - mempty = Success - mappend Success a = a - mappend (Failure s) Success = Failure s - mappend (Failure s) (Failure t) = Failure $ unlines [s,t] - -displayResult :: TestResult -> String -displayResult Success = "OK." -displayResult (Failure s) = unlines $ ("Failure":) . map (mappend " ") . filter (not . null) $ lines s - -finish :: TestResult -> IO a -finish Success = exitSuccess -finish (Failure s) = die s - -main :: IO () -main = do - execFile <- getExecutablePath - args <- getArgs - case getFirst $ foldMap (First . readMaybe) args of - -- run particular test routine - Just spec -> run spec >>= finish - -- run all tests - Nothing -> do - results <- forM tests $ \iox -> do - (ts, fin) <- iox - r <-runSpecs execFile ts - fin - return r - case foldMap id results of - Success -> exitSuccess - Failure _ -> exitFailure - -runSpecs :: FilePath -> [TestSpec] -> IO TestResult -runSpecs f specs = do - putStrLn "" - r <- go (zip [1 :: Int ..] specs) >>= evaluate - performGC - putStrLn $ "Result: " ++ displayResult r - return r +asyncException :: TestSpec +asyncException = Repeat 10 $ WithTimeLimit 2000 $ TestSpec "AsyncException" [((), run)] where - conf ts = setStdout createPipe - $ setStderr createPipe - $ proc f [show ts] - - go [] = return Success - go ((i,x):xs) = do - mr <- Vanilla.newEmptyMVar - void . forkIO $ withProcess (conf x) $ \p -> do - hGetContents (getStdout p) >>= mapM_ (putStrLn . withI) . lines - errs <- hGetContents (getStderr p) - ecode <- waitExitCode p - evaluate $ foldr seq () errs - Vanilla.putMVar mr $! case ecode of - ExitSuccess -> Success - ExitFailure _ -> Failure (unlines . map withI $ lines errs) + run :: () -> StoredMVar Int -> IO TestResult + run _ mvar = do + let timeMs = 50 :: Int + locked <- async $ takeMVar mvar + threadDelay $ timeMs * 1000 + killed <- async $ cancel locked + let pollStatus t = do + let t1 = t + timeMs + threadDelay $ timeMs * 1000 + r <- poll killed + case r of + Nothing -> pollStatus t1 + Just _ -> return t1 + -- NB: the whole thread is gonna be killed by the test runner + -- if pollStatus does not finish within the time defined in SpecParams. + elapsed <- pollStatus 0 + putStrLn $ "Cancelled a thread within " ++ show elapsed ++ " ms" + return Success - rx <- unsafeInterleaveIO $ Vanilla.takeMVar mr - rxs <- go xs - return $ mappend rx rxs - where - withI s = "[" ++ show i ++ "] " ++ s +main :: IO () +main = runTests [simpleTakePut, readersTakers, asyncException] diff --git a/test/Tools/Handle.hs b/test/Tools/Handle.hs new file mode 100644 --- /dev/null +++ b/test/Tools/Handle.hs @@ -0,0 +1,36 @@ +{-# LANGUAGE LambdaCase #-} +module Tools.Handle (redirectHandle) where + +import Control.Exception (bracket) +import Data.ByteString.Char8 (ByteString, hGetContents) +import GHC.IO.Handle (hDuplicate, hDuplicateTo, hFlushAll) +import GHC.IO.Handle.Types (Handle (DuplexHandle, FileHandle)) +import System.IO hiding (hGetContents) +import System.IO.Temp (withSystemTempFile) + +-- | Temporary redirect the output of a handle to an output ByteString (e.g. `stdout` or `stderr`). +redirectHandle :: Handle -> IO a -> IO (ByteString, a) +redirectHandle origHandle action = withSystemTempFile "interprocess.out" $ \_ tempFile -> do + start <- hGetPosn tempFile + hGetEncoding origHandle >>= \case + Just e -> hSetEncoding tempFile e + Nothing -> hSetBinaryMode tempFile True + hSetBuffering tempFile NoBuffering + hSetNewlineMode tempFile noNewlineTranslation + r <- bracket (redirect $ hGetWriteEnd tempFile) unredirect (const action) + r `seq` hFlushAll tempFile + hSetPosn start + out <- hGetContents tempFile + return (out, r) + where + redirect tempFile = do + origDup <- hDuplicate origHandleOut + hDuplicateTo tempFile origHandleOut + return origDup + + unredirect origDup = hDuplicateTo origDup origHandleOut >> hClose origDup + + origHandleOut = hGetWriteEnd origHandle + + hGetWriteEnd (DuplexHandle p _ w) = FileHandle p w + hGetWriteEnd h = h diff --git a/test/Tools/Runner.hs b/test/Tools/Runner.hs new file mode 100644 --- /dev/null +++ b/test/Tools/Runner.hs @@ -0,0 +1,207 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Tools.Runner (runTests, TestSpec (..)) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async +import qualified Control.Concurrent.MVar as StdMVar +import Control.Concurrent.Process.StoredMVar +import Control.Exception (SomeException, catch, + evaluate) +import Control.Monad (foldM, forever, when) +import Control.Monad.STM (atomically) +import qualified Data.ByteString.Char8 as BS +import Data.ByteString.Lazy.Char8 (unpack) +import Data.Foldable (fold) +import Data.IORef (atomicModifyIORef', + newIORef) +import qualified Data.List as List +import qualified Data.Map as Map +import Data.Tuple (swap) +import Foreign.Storable (Storable) +import System.Environment (getArgs, + getExecutablePath) +import System.Exit +import System.IO +import System.Mem (performMajorGC) +import System.Process.Typed +import Text.Read (readMaybe) + +import Tools.Handle +import Tools.TestResult + +{- | + Specification of a single test. + Defines the test name and the list of roles to be given to subprocesses. + + The test spawns as many subprocesses as there are elements in the roles list. + All of them share a single IPC MVar for communication. In addition, the role + actions are parameterized by the role key `k`, which is passed via the command-line + arguments using the Show-Read instances. + -} +data TestSpec where + TestSpec :: forall k ctx + . (Show k, Read k, Eq k, Storable ctx) + => String -> [(k, k -> StoredMVar ctx -> IO TestResult)] -> TestSpec + -- | Set the maximum time the test can take in milliseconds. + -- The test process will be forcefully terminated after the time limit elapses. + WithTimeLimit :: Int -> TestSpec -> TestSpec + -- | Repeat the test the specified number of times. + Repeat :: Int -> TestSpec -> TestSpec + +-- | Get the name of the test. +testName :: TestSpec -> String +testName (TestSpec name _) = name +testName (WithTimeLimit _ s) = testName s +testName (Repeat _ s) = testName s + +stripSpec :: TestSpec -> TestSpec +stripSpec (WithTimeLimit _ s) = stripSpec s +stripSpec (Repeat _ s) = stripSpec s +stripSpec s = s + +data SpecParams = SpecParams + { exec :: String -- ^ Executable name + , timeout :: Int -- ^ Terminate process after this amount of milliseconds + , processStartTimeout :: Int -- ^ Fail if the child process not all child processes start within this time + } + +defaultSpecParams :: String -> SpecParams +defaultSpecParams execFile = SpecParams + { exec = execFile + , timeout = 1000 + , processStartTimeout = 10000 + } + +-- | A random string for a handshake between the processes. +startToken :: String +startToken = "KxmpaDIlwnSlfp01==123=123=1" + +-- | Run all tests in the list. Use it as the `main`: +-- +-- @ +-- main :: IO () +-- main = runTests [...] +-- @ +runTests :: [TestSpec] -> IO () +runTests specs = do + execFile <- getExecutablePath + args <- getArgs + case args of + -- run all tests + [] -> do + results <- mapM (runSpec $ defaultSpecParams execFile) specs + case fold results of + Success -> exitSuccess + Failure _ -> exitFailure + + -- run particular test routine + [n, sk, sm] -> do + hSetBuffering stdout LineBuffering + putStrLn startToken + returnedToken <- getLine + when (startToken /= returnedToken) $ fail "Child start handshake failed." + TestSpec _ roles <- case List.find ((== n) . testName) specs of + Nothing -> fail $ "Unknown test name: " ++ show n + Just spec -> pure $ stripSpec spec + k <- case readMaybe sk of + Nothing -> fail $ "Couldn't read the role " ++ show sk + Just x -> pure x + m <- case readMaybe sm of + Nothing -> fail $ "Couldn't read the mvar name " ++ show sm + Just ref -> lookupMVar ref + case List.find ((== k) . fst) roles of + Nothing -> fail $ "Unknown test role " ++ show k ++ " for test " ++ show n + Just (_ , action) -> action k m >>= finish + + -- unexpected + _ -> fail $ "Expected zero or three arguments, got " ++ show args + + +runSpec :: SpecParams -> TestSpec -> IO TestResult +runSpec pams (WithTimeLimit t s) = runSpec (pams {timeout = t}) s +runSpec pams (Repeat n s) = do + putStrLn $ "[" ++ testName s ++ "] Running " ++ show n ++ " times..." + r <- foldM go Success [1..n] + putStrLn $ "[" ++ testName s ++ "] Result: " ++ displayResult r + return r + where + go Success i = do + (out, r) <- redirectHandle stdout $ do + r <- runSpec pams s + r `seq` performMajorGC + return r + case r of + Success -> pure () + _ -> do + putStrLn $ "[" ++ testName s ++ "] Failed on test iteration " ++ show i + BS.putStrLn out + return r + go failure _ = pure failure +runSpec pams (TestSpec name (roles :: [(k, k -> StoredMVar ctx -> IO TestResult)])) = do + putStrLn $ "[" ++ name ++ "] Running..." + comms <- newEmptyMVar :: IO (StoredMVar ctx) + childrenLeft <- newIORef $ length roles + readyToGo <- StdMVar.newEmptyMVar :: IO (StdMVar.MVar ()) + let k = show $ mVarName comms + r <- runConcurrently $ foldMap (\(l, r) -> go childrenLeft readyToGo l [name, r, k]) labelsroles + r `seq` comms `seq` performMajorGC + putStrLn $ "[" ++ name ++ "] Result: " ++ displayResult r + return r + where + roleCount :: Map.Map String Int + roleCount = Map.filter (> 1) $ + List.foldl' (\m (k, _) -> Map.alter (Just . maybe 1 succ) (show k) m) mempty roles + + labelsroles :: [(String, String)] + labelsroles = snd $ List.mapAccumR (\m (k, _) -> swap $ Map.alterF ( + -- Maybe Int -> ((String, String), Maybe Int) + maybe ((show k, show k), Nothing) (\i -> ((show k ++ "-" ++ show i, show k), Just (pred i))) + ) (show k) m ) roleCount roles + + withLabel label s = " [" ++ label ++ "] " ++ s + + channelHandle i handle = forever (hGetLine handle >>= (putStrLn . withLabel i)) + `catch` (const $ return mempty :: IOError -> IO ()) + + mkConf x + = setStdout createPipe + $ setStdin createPipe + $ setStderr byteStringOutput + $ proc (exec pams) x + + errToResult :: String -> SomeException -> IO TestResult + errToResult label = return . Failure . withLabel label . show + + messWithProcess label x action = withProcessTerm (mkConf x) action `catch` errToResult label + + go childrenLeft readyToGo label x = Concurrently + $ messWithProcess label x $ \p -> hGetLine (getStdout p) >>= \handshake -> + if handshake /= startToken + then return (Failure $ withLabel label ("Unexpected starting token: " ++ handshake)) + else withAsyncBound (channelHandle label (getStdout p)) $ \sink -> do + iAmLast <- atomicModifyIORef' childrenLeft (\n -> (n - 1, n == 1)) + when iAmLast $ StdMVar.putMVar readyToGo () + ready <- race (threadDelay $ 1000 * processStartTimeout pams) (StdMVar.readMVar readyToGo) + case ready of + Left () -> return $ Failure $ withLabel label "child process didn't start in time" + Right () -> do + hPutStrLn (getStdin p) handshake + hClose (getStdin p) + r <- race (threadDelay $ 1000 * timeout pams) $ do + ecode <- (Right <$> waitExitCode p) `catch` (return . Left) + case ecode of + Right ExitSuccess -> pure Success + Right (ExitFailure _) -> do + errs <- atomically (getStderr p) + return $ Failure (unlines . map (withLabel label) $ lines $ unpack errs) + Left e -> errToResult label e + _ <- evaluate r + wait sink + hFlush stdout + hFlush stderr + case r of + Left () -> return $ Failure $ withLabel label $ + "the process did not finish in " ++ show (timeout pams) ++ " milliseconds" + Right s -> return s diff --git a/test/Tools/TestResult.hs b/test/Tools/TestResult.hs new file mode 100644 --- /dev/null +++ b/test/Tools/TestResult.hs @@ -0,0 +1,24 @@ +module Tools.TestResult (TestResult(..), displayResult, finish) where + +import System.Exit (die, exitSuccess) + +data TestResult + = Success + | Failure String + deriving (Eq, Ord, Show, Read) + +instance Semigroup TestResult where + Success <> a = a + Failure s <> Success = Failure s + Failure s <> Failure t = Failure $ unlines [s, t] + +instance Monoid TestResult where + mempty = Success + +displayResult :: TestResult -> String +displayResult Success = "OK." +displayResult (Failure s) = unlines $ ("Failure" :) . map (mappend " ") . filter (not . null) $ lines s + +finish :: TestResult -> IO a +finish Success = exitSuccess +finish (Failure s) = die s