interprocess 0.2.0.1 → 0.2.1.0
raw patch · 23 files changed
+1580/−1100 lines, 23 filesdep +asyncdep +bytestringdep +containersdep ~basedep ~typed-processPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependencies added: async, bytestring, containers, stm, temporary
Dependency ranges changed: base, typed-process
API changes (from Hackage documentation)
- Foreign.SharedObjectName.Internal: SOName :: (ForeignPtr CChar) -> SOName a
+ Foreign.SharedObjectName.Internal: SOName :: ForeignPtr CChar -> SOName a
Files
- README.md +3/−3
- cbits/SharedObjectName.c +0/−50
- cbits/common.c +58/−0
- examples/concurrent-malloc.hs +1/−1
- examples/wait-mvar.hs +1/−1
- examples/wait-qsem.hs +1/−1
- include/SharedObjectName.h +0/−10
- include/SharedPtr.h +16/−18
- include/common.h +42/−0
- interprocess.cabal +30/−16
- src/Control/Concurrent/Process/QSem.c +2/−3
- src/Control/Concurrent/Process/StoredMVar.c +0/−746
- src/Control/Concurrent/Process/StoredMVar.hs +59/−47
- src/Control/Concurrent/Process/StoredMVarPosix.c +413/−0
- src/Control/Concurrent/Process/StoredMVarPrim.cmm +37/−0
- src/Control/Concurrent/Process/StoredMVarWin32.c +538/−0
- src/Foreign/SharedObjectName/Internal.hs +1/−1
- src/Foreign/SharedPtr.c +1/−1
- src/Foreign/SharedPtrWin32.c +33/−42
- test/StoredMVar.hs +77/−160
- test/Tools/Handle.hs +36/−0
- test/Tools/Runner.hs +207/−0
- test/Tools/TestResult.hs +24/−0
README.md view
@@ -1,6 +1,6 @@ # interprocess [](https://hackage.haskell.org/package/interprocess)-[](http://travis-ci.org/achirkin/interprocess)+[](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
@@ -1,50 +0,0 @@-#include "SharedObjectName.h"-#include <limits.h>-#include <time.h>-#include <stdlib.h>-#include <string.h>--#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || defined(mingw32_HOST_OS)-#include <windows.h>-#define GetMyPid abs((int)GetCurrentProcessId())-#else-#include <unistd.h>-#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];- }-}
+ cbits/common.c view
@@ -0,0 +1,58 @@+#include "common.h"++#include <limits.h>+#include <stdatomic.h>+#include <string.h>+#include <time.h>++#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || \+ defined(mingw32_HOST_OS)+#include <windows.h>+#define GetMyPid abs((int)GetCurrentProcessId())+#else+#include <unistd.h>+#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;+}
examples/concurrent-malloc.hs view
@@ -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)
examples/wait-mvar.hs view
@@ -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)
examples/wait-qsem.hs view
@@ -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:))
@@ -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__ */
@@ -1,12 +1,11 @@ #ifndef __HSSHAREDPTR_H__ #define __HSSHAREDPTR_H__+#include "common.h" #include <stddef.h>-#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 <windows.h> typedef struct SharedMutex { SharedObjectName mutexName;@@ -14,7 +13,7 @@ #else #include <pthread.h> 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__ */
+ include/common.h view
@@ -0,0 +1,42 @@+#ifndef __HSCOMMON_H__+#define __HSCOMMON_H__++#define SHARED_OBJECT_NAME_LENGTH 32++#ifndef HS_IMPORT_CONSTANTS_ONLY+#include <Rts.h>++/**+ * 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 <stdio.h>+#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__ */
interprocess.cabal view
@@ -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
src/Control/Concurrent/Process/QSem.c view
@@ -1,6 +1,5 @@-#include "SharedObjectName.h"-#include "HsFFI.h"-#include <stdlib.h>+#include <common.h>+ typedef struct QSem QSem;
− src/Control/Concurrent/Process/StoredMVar.c
@@ -1,746 +0,0 @@-#include "SharedObjectName.h"-#include <stdlib.h>-#include <string.h>-#ifndef NDEBUG-// a bit of printf to track important events-#include <stdio.h>-#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 <windows.h>-#include <assert.h>-#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 <pthread.h>-#include <fcntl.h>-#include <sys/mman.h>-#include <unistd.h>---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);-}
src/Control/Concurrent/Process/StoredMVar.hs view
@@ -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
+ src/Control/Concurrent/Process/StoredMVarPosix.c view
@@ -0,0 +1,413 @@+#include <common.h>++#include <assert.h>+#include <errno.h>+#include <fcntl.h>+#include <pthread.h>+#include <stdatomic.h>+#include <string.h>+#include <sys/mman.h>+#include <time.h>+#include <unistd.h>++#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); }
+ src/Control/Concurrent/Process/StoredMVarPrim.cmm view
@@ -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));+}
+ src/Control/Concurrent/Process/StoredMVarWin32.c view
@@ -0,0 +1,538 @@+#include <common.h>++#include <assert.h>+#include <errno.h>+#include <limits.h>+#include <stdatomic.h>+#include <windows.h>++#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); }
@@ -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)
@@ -10,7 +10,7 @@ */ #include "SharedPtr.h"-#include <stdlib.h>+ #include <string.h> #include <assert.h>
@@ -1,19 +1,16 @@ #include "SharedPtr.h"-#ifndef NDEBUG-#include <stdio.h>-#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); }
test/StoredMVar.hs view
@@ -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]
+ test/Tools/Handle.hs view
@@ -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
+ test/Tools/Runner.hs view
@@ -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
+ test/Tools/TestResult.hs view
@@ -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