diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Artem Chirkin (c) 2018 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Artem Chirkin nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,24 @@ +# interprocess + +WIP: platform-independent interprocess communication. + +The core feature is shared memory allocator implemented using POSIX mmap and Windows CreateFileMapping. +You can create as many allocators as you want (as your RAM can afford) and +concurrently malloc and free memory in different processes using them. + +Features and TODO: + + * [x] `Foreign.SharedPtr` -- `malloc`, `realloc` and `free` in the shared memory region + that can be accessed by multiple processes. + * [x] Semaphores + * [ ] Mutexes (not sure if need this) + * [ ] Mutable variables via `Storable` instance plus garbage collection. + * [ ] Proper error messages + * [ ] Debug, verbose mode + +The idea of the library is to address GHC stop-the-world GC problem: + + 1. Create several instances of your program in different isolated processes + using e.g. `typed-process` library. + 2. Establish shared memory and semaphore usage via this program + 3. Garbage collection events in one process do not affect another one at all. Profit! diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/app/concurrent-malloc.hs b/app/concurrent-malloc.hs new file mode 100644 --- /dev/null +++ b/app/concurrent-malloc.hs @@ -0,0 +1,159 @@ +module Main where + +import Control.Monad +import Data.List (partition) +import Data.Maybe (fromMaybe) +import Data.Monoid (First (..)) +import Foreign.Marshal.Alloc +import Foreign.SharedPtr as Shared +import Foreign.Storable +import System.Environment +import System.Exit +import System.IO +import System.Process.Typed +import Text.Read (readMaybe) +import Data.IORef + +-- | Supply integer argument to a program to set the allocation size. +--- Number 10000 is default, corresponds to argound 800MB of memory and very fast +-- Number 25000 corresponds to around 5GB of memory +main :: IO () +main = do + args <- getArgs + let (isSlaveL, remargs) = partition ("slave"==) args + isSlave = not $ null isSlaveL + n = fromMaybe 10000 . getFirst $ foldMap (First . readMaybe) remargs + if isSlave + then runB n + else runA n + + +runA :: Int -> IO () +runA n = do + progName <- getProgName + args <- getArgs + let processBConfig = setStdin createPipe + $ proc progName ("slave":args) + + ec <- withNewAllocator $ \sa -> do + + putStrLn $ "[A] Allocator addr: " ++ show sa + ptr <- Shared.malloc sa + putStrLn $ "[A] Malloc'ed addr: " ++ show ptr + poke ptr n + + ec' <- withProcess_ processBConfig $ \procB -> do + + putStrLn $ "[A] Sending store name: " + ++ show (allocStoreName sa) + hPutStorable (getStdin procB) (allocStoreName sa) + putStrLn $ "[A] Sending SharedPtr: " + ++ show (toSharedPtr sa ptr) + hPutStorable (getStdin procB) + (toSharedPtr sa ptr) + + runMallocs "A" sa n + + putStrLn "[A] Now, waiting for process B to finish..." + ec'' <- waitExitCode procB + putStrLn "[A] Finished running B" + return ec'' + n' <- peek ptr + putStrLn $ "[A] Read back: " ++ show n' + Shared.free sa ptr + putStrLn "[A] Finished successfully" + return ec' + + exitWith ec + + +runB :: Int -> IO () +runB n = do + let inputH = stdin + Just sname <- hGetStorable inputH + putStrLn $ "[B] Received allocator store name: " ++ show sname + Just xptr <- hGetStorable inputH :: IO (Maybe (SharedPtr Int)) + putStrLn $ "[B] Received SharedPtr: " ++ show xptr + + withAllocator sname $ \sa -> do + let ptr = fromSharedPtr sa xptr + putStrLn $ "[B] Decoded SharedPtr: " ++ show ptr + n' <- peek ptr + putStrLn $ "[B] Compare N received through pipes (" ++ show n + ++ ") and N read from shared memory (" ++ show n' + ++ "): " ++ show (compare n n') + poke ptr (n + 777) + putStrLn $ "[B] updated pointer value: " ++ show (n + 777) + + runMallocs "B" sa ((n * 11) `div` 10) + + putStrLn "[B] Finished successfully" + + + +-- | Run malloc on increasingly big size, 8 + i bytes, where i = [1..n]. +runMallocs :: String -> Allocator -> Int -> IO () +runMallocs runnerName a n = do + reqBytes <- newIORef 0 + -- run malloc + ptrs <- forM [1..n] $ \i -> do + p <- Shared.mallocBytes a (8+i) + modifyIORef' reqBytes (+(8+i)) + when (mod i (n `div` thisManyMallocFreeReports) == 7) $ + putStrLn $ "[" ++ runnerName ++ "] Malloced: " ++ show p + forM_ [0 .. i `div` 8] $ \j -> pokeElemOff p j (i - j) + return p + -- run realloc + ptrs' <- forM (zip [1..] ptrs) $ \(i, p) -> do + let j = newElemLength i + p' <- Shared.realloc a p (8*j) + modifyIORef' reqBytes (+(8*j)) + when (mod i (n `div` thisManyMallocFreeReports) == 7) $ + putStrLn $ "[" ++ runnerName ++ "] Realloced: " ++ show p' + return p' + + -- run free + r <- foldM (\(y, i) p -> do + let j = newElemLength i + -- generate some work so it needs time to finish + x <- foldM (const $ peekElemOff p) 0 [j-1, j-2 .. 0] + Shared.free a p + when (mod i (n `div` thisManyMallocFreeReports) == 7) $ + putStrLn $ "[" ++ runnerName ++ "] Liberated ptr and read value: " + ++ show x + return (y + x, i + 1) + ) (0, 1 :: Int) ptrs' + putStrLn $ "[" ++ runnerName ++ "] Validate results: " + ++ show (r, sum [1..n]) + reqBytesVal <- readIORef reqBytes + putStrLn $ "[" ++ runnerName ++ "] TotalMemory requested in the loops (MB): " + ++ show (fromIntegral ((reqBytesVal * 1000 * 1000) `div` (1024 * 1024)) + / (1000 * 1000) :: Double) + where + thisManyMallocFreeReports = 10 + newElemLength i = case mod i 9 of + 1 -> 3 + 2 -> 1 + 3 -> min 5 (div i 3) + 4 -> i + 7 + 5 -> i * 2 + 6 -> div (i * 3) 2 + _ -> i + + +hPutStorable :: Storable a => Handle -> a -> IO () +hPutStorable h a = alloca $ \ptr -> do + poke ptr a + hPutBuf h ptr (sizeOf a) + hFlush h + +hGetStorable :: Storable a => Handle -> IO (Maybe a) +hGetStorable h = go undefined + where + go :: Storable a => a -> IO (Maybe a) + go a = alloca $ \ptr -> do + let n = sizeOf a + n' <- hGetBuf h ptr n + if n' < n + then return Nothing + else Just <$> peek ptr diff --git a/app/wait-qsem.hs b/app/wait-qsem.hs new file mode 100644 --- /dev/null +++ b/app/wait-qsem.hs @@ -0,0 +1,87 @@ +module Main where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Process.QSem +import Control.Monad +import Data.List (partition) +import Data.Maybe (fromMaybe) +import Data.Monoid (First (..)) +import Foreign.SharedObjectName +import System.Environment +import System.IO +import System.Process.Typed +import Text.Read (readMaybe) + +main :: IO () +main = do + args <- getArgs + let (isSlaveL, remargs) = partition ("slave"==) args + isSlave = not $ null isSlaveL + n = fromMaybe 4 . getFirst $ foldMap (First . readMaybe) remargs + if isSlave + then runB + else runA n + + +runA :: Int -> IO () +runA n = do + progName <- getProgName + args <- getArgs + let processBConfig = setStdin createPipe + $ proc progName ("slave":args) + + withNProcesses n processBConfig $ \procs -> do + + qSem <- newQSem 0 + + forM_ (zip [99 :: Int, 98..] procs) $ \(i, p) -> do + hPutSOName (getStdin p) (qSemName qSem) + hPrint (getStdin p) $ 100 - i + hPutStrLn (getStdin p) $ "Say " ++ show i ++ " bottles of rum!" + hFlush (getStdin p) + + threadDelay 100000 + putStrLn "[A] Done! Signal semaphore available to other threads" + replicateM_ n $ threadDelay 100000 >> signalQSem qSem + + putStrLn "[A] Finished successfully" + + + +runB :: IO () +runB = do + let inputH = stdin + Just qSemRef <- hGetSOName inputH -- get name of a semaphore + qSem <- lookupQSem qSemRef + i <- read <$> hGetLine inputH -- get id of a spawned process + instruction <- hGetLine inputH -- some arbitrary text + putStrLn $ "[B] " ++ instruction + threadDelay 500000 + if mod i 7 == (2 :: Int) + then + let procedure = do + wasAvailable <- tryWaitQSem qSem + if wasAvailable + then + putStrLn $ "[B] (" ++ show i ++ ") Was available - " ++ reverse instruction + else do + putStrLn $ "[B] (" ++ show i ++ ") Ha-ha, I am not blocked!" + waitQSem qSem + putStrLn $ "[B] (" ++ show i ++ ") Woke up" + signalQSem qSem + threadDelay 100000 + procedure + in procedure + else do + waitQSem qSem + putStrLn $ "[B] " ++ reverse instruction + putStrLn "[B] Finished successfully" + + +withNProcesses :: Int + -> ProcessConfig stdin stdout stderr + -> ([Process stdin stdout stderr] -> IO a) + -> IO a +withNProcesses 0 _ k = k [] +withNProcesses n conf k = withProcess_ conf $ \p -> + withNProcesses (n-1) conf (k . (p:)) diff --git a/cbits/SharedObjectName.c b/cbits/SharedObjectName.c new file mode 100644 --- /dev/null +++ b/cbits/SharedObjectName.c @@ -0,0 +1,50 @@ +#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/include/SharedObjectName.h b/include/SharedObjectName.h new file mode 100644 --- /dev/null +++ b/include/SharedObjectName.h @@ -0,0 +1,10 @@ +#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 new file mode 100644 --- /dev/null +++ b/include/SharedPtr.h @@ -0,0 +1,64 @@ +#ifndef __HSSHAREDPTR_H__ +#define __HSSHAREDPTR_H__ + +#include +#include "HsFFI.h" +#include "MachDeps.h" +#include "SharedObjectName.h" + +#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || defined(mingw32_HOST_OS) +#include +typedef struct SharedMutex { + SharedObjectName mutexName; +} SharedMutex; +#else +#include +typedef struct SharedMutex { + pthread_mutex_t mutVal; + pthread_mutexattr_t mutAttr; +} SharedMutex; +#endif + +typedef struct SharedAllocator SharedAllocator; +// Internal pointer representation is a single word; +typedef HsWord SharedPtr; + +// Create a new memory allocator +// Returns NULL on failure +SharedAllocator *shared_createAllocator(); +// Create a new memory allocator for an existing memory pool +// Returns NULL on failure +SharedAllocator *shared_lookupAllocator(const char *storeName); +// Destroy a memory allocator. +// 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); +// 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); +// May return NULL on failure. +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. +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 _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); + + +#endif /* __HSSHAREDPTR_H__ */ diff --git a/interprocess.cabal b/interprocess.cabal new file mode 100644 --- /dev/null +++ b/interprocess.cabal @@ -0,0 +1,81 @@ +name: interprocess +version: 0.1.0.0 +synopsis: Shared memory and control structures for IPC +description: Provides portable shared memory allocator and semaphores. + Can be used for interprocess communication. + Refer to README.md for further information. +homepage: https://github.com/achirkin/interprocess +license: BSD3 +license-file: LICENSE +author: Artem Chirkin +maintainer: chirkin@arch.ethz.ch +copyright: (c) 2018 Artem Chirkin +category: system +build-type: Simple +cabal-version: >=1.10 +extra-source-files: + README.md + cbits/SharedObjectName.c + include/SharedObjectName.h + include/SharedPtr.h + src/Control/Concurrent/Process/QSem.c + src/Foreign/SharedPtrPosix.c + src/Foreign/SharedPtrWin32.c + +flag examples + description: + Build example-test programs + default: False + +flag dev + description: + Development build with debugging and with no optimization + default: False + +library + hs-source-dirs: src + exposed-modules: Foreign.SharedPtr + Foreign.SharedObjectName + Foreign.SharedObjectName.Internal + Foreign.SharedPtr.C + Control.Concurrent.Process.QSem + build-depends: base >= 4.7 && < 5 + default-language: Haskell2010 + ghc-options: -Wall + include-dirs: include + c-sources: cbits/SharedObjectName.c + src/Control/Concurrent/Process/QSem.c + src/Foreign/SharedPtr.c + if os(windows) + c-sources: src/Foreign/SharedPtrWin32.c + else + c-sources: src/Foreign/SharedPtrPosix.c + if flag(dev) + cpp-options: -DSTDOUT_SYSCALL_DEBUG + ghc-options: -O0 + else + cpp-options: -DNDEBUG + ghc-options: -O2 + + +executable interprocess-concurrent-malloc + if !flag(examples) + buildable: False + hs-source-dirs: app + main-is: concurrent-malloc.hs + default-language: Haskell2010 + build-depends: base >= 4.7 && < 5 + , typed-process >= 0.2 + , interprocess + ghc-options: -threaded -Wall + +executable interprocess-wait-qsem + if !flag(examples) + buildable: False + hs-source-dirs: app + main-is: wait-qsem.hs + default-language: Haskell2010 + build-depends: base >= 4.7 && < 5 + , typed-process >= 0.2 + , interprocess + ghc-options: -threaded -Wall diff --git a/src/Control/Concurrent/Process/QSem.c b/src/Control/Concurrent/Process/QSem.c new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Process/QSem.c @@ -0,0 +1,191 @@ +#include "SharedObjectName.h" +#include "HsFFI.h" +#include + +typedef struct QSem QSem; + +QSem *qsem_new(HsInt count); +QSem *qsem_lookup(const char *name); +void qsem_close(QSem *qsem); +int qsem_signal(QSem *qsem); +int qsem_wait(QSem *qsem); +int qsem_trywait(QSem *qsem); +void qsem_name(QSem *qsem, char * const name); + + + +#if defined(WIN32) || defined(_WIN32) || defined(__WIN32) || defined(mingw32_HOST_OS) +#include +#include + +typedef struct QSem { + HANDLE mainSem; + SharedObjectName mainName; +} QSem; + +QSem *qsem_new(HsInt count) { + QSem *r = malloc(sizeof(QSem)); + if (r == NULL) { + return NULL; + } + genSharedObjectName(r->mainName); + + HANDLE semPtr = CreateSemaphoreA( NULL, (LONG)count, LONG_MAX, r->mainName ); + + if (semPtr == NULL) { + free(r); + return NULL; + } + + r->mainSem = semPtr; + return r; +} + +QSem *qsem_lookup(const char *name) { + QSem *r = malloc(sizeof(QSem)); + if (r == NULL) { + return NULL; + } + + HANDLE semPtr = OpenSemaphoreA( SEMAPHORE_MODIFY_STATE | SYNCHRONIZE, TRUE, name ); + + if (semPtr == NULL) { + free(r); + return NULL; + } + + r->mainSem = semPtr; + strcpy(r->mainName, name); + return r; +} + +void qsem_close(QSem *qsem) { + CloseHandle(qsem->mainSem); + free(qsem); +} + +int qsem_signal(QSem *qsem) { + return (ReleaseSemaphore(qsem->mainSem, 1, NULL) == 0); +} + +int qsem_wait(QSem *qsem) { + return (WaitForSingleObject(qsem->mainSem, INFINITE) != WAIT_OBJECT_0); +} + +int qsem_trywait(QSem *qsem) { + return (WaitForSingleObject(qsem->mainSem, 0) != WAIT_OBJECT_0); +} + + +void qsem_name(QSem *qsem, char * const name) { + strcpy(name, qsem->mainName); +} + + +#else +#include /* For O_* constants */ +#include +#include +#include "HsFFI.h" + +#define GuardNameSuffix "X" + +typedef struct QSem { + sem_t *mainSem; + sem_t *guardSem; + /* Second semaphore is used to count number of + extra users. + That is, the quantity of a guardSem is (N-1) where N is number of processes + currently having the semaphore initialized. + When executing qsem_close, sem_trywait result defines if the semaphore + should be unlinked or not. + */ + SharedObjectName mainName; +} QSem; + +QSem *qsem_new(HsInt count) { + QSem *r = malloc(sizeof(QSem)); + if (r == NULL) { + return NULL; + } + genSharedObjectName(r->mainName); + + sem_t *mPtr = sem_open (r->mainName, O_CREAT | O_EXCL, 0600, count); + if (mPtr == SEM_FAILED) { + free(r); + return NULL; + } + SharedObjectName guardName = {0}; + strcpy (guardName, r->mainName); + strcat (guardName, GuardNameSuffix); + sem_t *cPtr = sem_open (guardName, O_CREAT | O_EXCL, 0600, 0); + if (mPtr == SEM_FAILED) { + free(r); + if (sem_close(mPtr) == 0){ + sem_unlink(r->mainName); + } + return NULL; + } + r->mainSem = mPtr; + r->guardSem = cPtr; + return r; +} + +QSem *qsem_lookup(const char *name) { + QSem *r = malloc(sizeof(QSem)); + if (r == NULL) { + return NULL; + } + sem_t *mPtr = sem_open (name, 0); + if (mPtr == SEM_FAILED) { + free(r); + return NULL; + } + SharedObjectName guardName = {0}; + strcpy (guardName, name); + strcat (guardName, GuardNameSuffix); + sem_t *cPtr = sem_open (guardName, 0); + if (mPtr == SEM_FAILED || sem_post(cPtr) != 0) { + free(r); + sem_close(mPtr); // don't unlink semaphore in case it is used somewhere else. + return NULL; + } + r->mainSem = mPtr; + r->guardSem = cPtr; + strcpy(r->mainName, name); + return r; +} + +void qsem_close(QSem *qsem) { + sem_close(qsem->mainSem); + if ( sem_trywait(qsem->guardSem) != 0 ) { + sem_unlink(qsem->mainName); + sem_close(qsem->guardSem); + SharedObjectName guardName = {0}; + strcpy (guardName, qsem->mainName); + strcat (guardName, GuardNameSuffix); + sem_unlink(guardName); + } else { + sem_close(qsem->guardSem); + } + free(qsem); +} + +int qsem_signal(QSem *qsem) { + return sem_post(qsem->mainSem); +} + +int qsem_wait(QSem *qsem) { + return sem_wait(qsem->mainSem); +} + +int qsem_trywait(QSem *qsem) { + return sem_trywait(qsem->mainSem); +} + +void qsem_name(QSem *qsem, char * const name) { + strcpy(name, qsem->mainName); +} + + +#endif diff --git a/src/Control/Concurrent/Process/QSem.hs b/src/Control/Concurrent/Process/QSem.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Process/QSem.hs @@ -0,0 +1,117 @@ +{-# OPTIONS_GHC -funbox-strict-fields #-} +-- | Simple interprocess quantity semaphores +-- +-- Based on POSIX or Win32 C semaphores +module Control.Concurrent.Process.QSem + ( QSem, newQSem, lookupQSem, waitQSem, tryWaitQSem, signalQSem, qSemName + ) where + +import Control.Monad (when) +import Foreign.C.Error +import Foreign.C.String +import Foreign.C.Types +import Foreign.ForeignPtr +import Foreign.Ptr +import Foreign.SharedObjectName.Internal + + +-- | Opaque implementation-dependent semaphore +data QSemT + +-- | 'QSem' is a quantity semaphore in which the resource is aqcuired +-- and released in units of one. +data QSem = QSem !(SOName QSem) !(ForeignPtr QSemT) + +-- | Build a new 'QSem' with a supplied initial quantity. +-- The initial quantity must be at least 0. +-- +-- This function throws an exception +-- if an underlying platform-dependent function fails. +newQSem :: Int -> IO QSem +newQSem initial + | initial < 0 = fail "newQSem: Initial quantity must be non-negative" + | otherwise = do + qsem <- checkNullPointer "newQSem" $ c'qsem_new initial + n <- newEmptySOName + unsafeWithSOName n $ c'qsem_name qsem + QSem n <$> newForeignPtr p'qsem_close qsem + +-- | Lookup QSem by its name in the global namespace. +-- Use this function to init several entangled semaphores in different processes. +-- +-- This function throws an exception if no `QSem` with this name exist, +-- or if an underlying platform-dependent function fails. +lookupQSem :: SOName QSem -> IO QSem +lookupQSem n = do + qsem <- unsafeWithSOName n $ checkNullPointer "lookupQSem" . c'qsem_lookup + QSem n <$> newForeignPtr p'qsem_close qsem + +-- | Get a global reference to the semaphore. +-- Send this reference to another process to lookup this semaphore and +-- start interprocess communication. +qSemName :: QSem -> SOName QSem +qSemName (QSem r _) = r + +-- | Wait for a unit to become available +-- +-- This function throws an exception +-- if an underlying platform-dependent function fails. +waitQSem :: QSem -> IO () +waitQSem (QSem _ p) = withForeignPtr p $ checkZeroReturn "waitQSem" . c'qsem_wait + +-- | Try to take a unit of the `QSem`. +-- +-- This function does not wait, in fact. Sorry for naming. +-- +-- Returns: +-- +-- * @True@ if successfully took a unit of `QSem` (it is decremented) +-- * @False@ if number of available units is less than @1@ (it is not decremented) +-- +-- This function does not throw an exception. +tryWaitQSem :: QSem -> IO Bool +tryWaitQSem (QSem _ p) = withForeignPtr p $ fmap (0 ==) . c'qsem_trywait + + +-- | Signal that a unit of the 'QSem' is available +-- +-- This function throws an exception +-- if an underlying platform-dependent function fails. +signalQSem :: QSem -> IO () +signalQSem (QSem _ p) = withForeignPtr p $ checkZeroReturn "signalQSem" . c'qsem_signal + + +checkNullPointer :: String -> IO (Ptr a) -> IO (Ptr a) +checkNullPointer s k = do + p <- k + if p == nullPtr + then throwErrno (s ++ " returned NULL pointer.") + else return p + +checkZeroReturn :: String -> IO CInt -> IO () +checkZeroReturn s k = do + p <- k + when (p /= 0) $ + throwErrno (s ++ " returned non-zero result.") + + +foreign import ccall unsafe "qsem_new" + c'qsem_new :: Int -> IO (Ptr QSemT) + +foreign import ccall unsafe "qsem_lookup" + c'qsem_lookup :: CString -> IO (Ptr QSemT) + +foreign import ccall unsafe "&qsem_close" + p'qsem_close :: FunPtr (Ptr QSemT -> IO ()) + +foreign import ccall unsafe "qsem_signal" + c'qsem_signal :: Ptr QSemT -> IO CInt + +foreign import ccall unsafe "qsem_wait" + c'qsem_wait :: Ptr QSemT -> IO CInt + +foreign import ccall unsafe "qsem_trywait" + c'qsem_trywait :: Ptr QSemT -> IO CInt + +foreign import ccall unsafe "qsem_name" + c'qsem_name :: Ptr QSemT -> CString -> IO () diff --git a/src/Foreign/SharedObjectName.hs b/src/Foreign/SharedObjectName.hs new file mode 100644 --- /dev/null +++ b/src/Foreign/SharedObjectName.hs @@ -0,0 +1,16 @@ +-- | Globally unique names to be used as references in the interprocess communication. +-- +-- The implementation must guarantee that no two processes can generate +-- the same unique @SOName@ independently at the same time. +-- To ensure this, the implementation uses three different seeds: +-- +-- 1. @C@ @rand()@ with a time-dependent seed @srand(time(NULL))@ +-- 2. Process id taken from @getpid@ or @GetCurrentProcessId@ (unique across process). +-- 3. A global auto-incremented variable (unique within a process). +-- +module Foreign.SharedObjectName + ( SOName (), hPutSOName, hGetSOName, unsafeWithSOName + ) where + + +import Foreign.SharedObjectName.Internal diff --git a/src/Foreign/SharedObjectName/Internal.hs b/src/Foreign/SharedObjectName/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Foreign/SharedObjectName/Internal.hs @@ -0,0 +1,103 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE RoleAnnotations #-} +-- | Exposed internals of `Foreign.SharedObjectName`. +-- +module Foreign.SharedObjectName.Internal + ( SOName (..), hPutSOName, hGetSOName, unsafeWithSOName + , genSOName, newEmptySOName + ) where + +import Foreign.C.String +import Foreign.C.Types +import Foreign.ForeignPtr +import Foreign.ForeignPtr.Unsafe +import Foreign.Ptr +import Foreign.Storable +import System.IO +import System.IO.Unsafe + +#define HS_IMPORT_CONSTANTS_ONLY +#include "SharedObjectName.h" +#include "MachDeps.h" + +-- | Reference to a shared object; can be sent to other processes. +newtype SOName a = SOName (ForeignPtr CChar) +type role SOName phantom + +instance Show (SOName a) where + showsPrec d (SOName a) + = showParen (d >= 10) $ showString "SOName " . showsPrec 10 getstr + where + getstr = unsafePerformIO $ withForeignPtr a peekCString + {-# NOINLINE getstr #-} + +instance Eq (SOName a) where + (SOName a) == (SOName b) + = cmpCStrings (unsafeForeignPtrToPtr a) (unsafeForeignPtrToPtr b) == EQ + +instance Ord (SOName a) where + compare (SOName a) (SOName b) + = cmpCStrings (unsafeForeignPtrToPtr a) (unsafeForeignPtrToPtr b) + +instance Storable (SOName a) where + sizeOf _ = SHARED_OBJECT_NAME_LENGTH + alignment _ = SIZEOF_HSWORD + poke p (SOName qp) = withForeignPtr qp + $ \q -> c_memcpy p q SHARED_OBJECT_NAME_LENGTH + peek p = do + qp <- mallocForeignPtrBytes SHARED_OBJECT_NAME_LENGTH + withForeignPtr qp + $ \q -> c_memcpy q p SHARED_OBJECT_NAME_LENGTH + return $ SOName qp + +-- | Write a shared object name into somwhere referenced by a handle. +-- Useful for sending references to other processes via pipes. +hPutSOName :: Handle -> SOName a -> IO () +hPutSOName h (SOName q) + = withForeignPtr q $ flip (hPutBuf h) SHARED_OBJECT_NAME_LENGTH + +-- | Read a shared object name from somwhere referenced by a handle. +-- Returns @Nothing@ if @hGetBuf@ gets less than @SHARED_OBJECT_NAME_LENGTH@ bytes. +-- Useful for sending references to other processes via pipes. +hGetSOName :: Handle -> IO (Maybe (SOName a)) +hGetSOName h = do + let n = SHARED_OBJECT_NAME_LENGTH + q <- mallocForeignPtrBytes n + n' <- withForeignPtr q $ \p -> hGetBuf h p n + return $ + if n' < n + then Nothing + else Just (SOName q) + +-- | Generate a new unique shared object name. +genSOName :: IO (SOName a) +genSOName = do + fp <- mallocForeignPtrBytes SHARED_OBJECT_NAME_LENGTH + withForeignPtr fp c_genSharedObjectName + return $ SOName fp + +-- | Allocate a new shared object name. +newEmptySOName :: IO (SOName a) +newEmptySOName = SOName <$> mallocForeignPtrBytes SHARED_OBJECT_NAME_LENGTH + + +-- | Use a pointer to a C string to pass to some low-level (e.g. foreign) functions. +-- `SOName` is asserted immutable, so do not modify it! +unsafeWithSOName :: SOName a -> (CString -> IO b) -> IO b +unsafeWithSOName (SOName fp) = withForeignPtr fp + +-- | Check first if two CString point to the same memory location. +-- Otherwise, compare them using C @strcmp@ function. +cmpCStrings :: CString -> CString -> Ordering +cmpCStrings a b + | a == b = EQ + | otherwise = c_strcmp a b `compare` 0 + +foreign import ccall unsafe "strcmp" + c_strcmp :: CString -> CString -> CInt + +foreign import ccall unsafe "memcpy" + c_memcpy :: Ptr a -> Ptr b -> CInt -> IO () + +foreign import ccall unsafe "genSharedObjectName" + c_genSharedObjectName :: CString -> IO () diff --git a/src/Foreign/SharedPtr.c b/src/Foreign/SharedPtr.c new file mode 100644 --- /dev/null +++ b/src/Foreign/SharedPtr.c @@ -0,0 +1,558 @@ +/* Modified version of "Buddy system" from + "The Art of Computer Programming" by D.Knuth, vol 1, ch 2.5. + + Shared memory functions align all allocated memory to the system memory page size, + which is usually 4KB. + This has two implications for the algorithm: + 1. Minimum storage block size is equal to 4096 + 2. All allocated memory is aligned to min(2^k, 4096) bytes, where k is + minimum power such that allocsize < 2^k + + */ +#include "SharedPtr.h" +#include +#include +#include + +/* Default store size is at least the page size, which usualy is equal to 4KB. + */ +#define DEFAULT_STORE_SIZE_FACTOR 12 +#define DEFAULT_STORE_SIZE 4096 +/* + SharedPtrs define the store id and offset in the store; + storeId == 0 is reserved for SharedAllocData; + AvailableStorage must be not bigger than DEFAULT_STORE_SIZE to be addressable. + + MIN_ALLOC_FACTOR: k where 2^k = 2*sizeof(HsWord)+1 + MAX_STORES: (WORD_SIZE_IN_BITS - DEFAULT_STORE_SIZE_FACTOR + 1) + */ +#if SIZEOF_HSWORD == 4 +#define MIN_ALLOC_FACTOR 4 +#elif SIZEOF_HSWORD == 8 +#define MIN_ALLOC_FACTOR 5 +#else +#error Unknown WORD size architecture. Cannot proceed anymore. +#endif + +typedef struct SharedMutex SharedMutex; + + +/* Determine necessary memory block size based on the requested size; + the block size must be: + 1. power of two; + 2. >= 2*sizeof(HsWord) + 1 + */ +HsWord8 allocFactor(size_t size){ + HsWord8 r = MIN_ALLOC_FACTOR; + size_t x = size >> MIN_ALLOC_FACTOR; + while(x > 0){ + r++; + x = x >> 1; + } + return r; +} + +/* Get storeId from the sharedPtr. + StoreId is defined by the largest bit of the pointer. + */ +HsWord8 getStoreId(const SharedPtr ptr){ + if (ptr < DEFAULT_STORE_SIZE ) { + return 0; + } + HsWord8 r = DEFAULT_STORE_SIZE_FACTOR; + SharedPtr x = ptr >> (DEFAULT_STORE_SIZE_FACTOR+1); + while(x > 0){ + r++; + x = x >> 1; + } + return r; +} + +/* Tag byte is to the left from allocation block, + to preserve data alignment, it has a negative position (-1). + The only exception is that tag for the first block is kept in the last + allocated memory addresed in a corresponding storage. + */ +#define STORAGE_TAG_BYTE_PTR(storePtr, storeSize, ptr) \ + ( (HsWord8*) \ + ( ( ( (((HsWord)ptr) - ((HsWord)storePtr)) + (((HsWord)storeSize) - ((HsWord)1)) ) \ + % ((HsWord)storeSize) \ + ) + ((HsWord)storePtr) \ + ) \ + ) \ + +/* This linked list contains pointers to available chunks of allocated memory + and an id of used store. + */ +typedef struct ListNode { + SharedPtr linkB; // previous free block + SharedPtr linkF; // next free block +} ListNode; + + +/* + Store a table of linked lists, one per each memory size. + + */ +typedef ListNode AvailableStorage[WORD_SIZE_IN_BITS]; + + +typedef struct SharedAllocData { + // Table of linked lists, one for each power-of-two memory size. + // Must go first, to be within first DEFAULT_STORE_SIZE of the shared memory block + AvailableStorage availStorage; + // Names of mmap-allocated storage chunks, + // storeNames[0] is the name of storage used to keep SharedAllocData itself + SharedObjectName storeNames[WORD_SIZE_IN_BITS]; + // How many allocators point to this data? only the last one should unlink an allocated memory. + HsWord usersN; + // Current largest allocated storage chunk + HsWord8 largestStoreId; + // Protect shared data acces + SharedMutex mutex; +} SharedAllocData; + +/* Allocator keeps part of its data in private memory, part in shared memory. + .storeAddrs[0] === .sharedAllocData + */ +typedef struct SharedAllocator{ + union { + SharedAllocData *sharedAllocData; + HsPtr storeAddrs[WORD_SIZE_IN_BITS]; + }; + // ids of stores sorted by their mmapped addressed ASC + HsWord8 storeIdsSorted[WORD_SIZE_IN_BITS]; + // Total number of initialized stores, including the main one. + HsWord8 storeN; + // Windows-use only: HANDLE to mutex + void *mutexPrivateHandle; + // Windows-use only: HANDLEs to stores + void *storePrivateHandles[WORD_SIZE_IN_BITS]; +} SharedAllocator; + +// returns NULL if failed +SharedAllocData *_SharedAllocData_init(void **privateStoreHandle, void **mutexPrivateHandle){ + SharedObjectName stName = { 0 }; + genSharedObjectName(stName); + SharedAllocData *sdataPtr + = (SharedAllocData*)_store_alloc(stName, privateStoreHandle, sizeof(SharedAllocData)); + if (sdataPtr == NULL) { + return NULL; + } + memset(sdataPtr, 0, sizeof(SharedAllocData)); + SharedPtr curNode; + for(HsInt i = 0; i < WORD_SIZE_IN_BITS; i++) { + curNode = ((SharedPtr) &(sdataPtr->availStorage[i])) - ((SharedPtr) sdataPtr); + sdataPtr->availStorage[i] + = (struct ListNode) + { .linkB = curNode + , .linkF = curNode + }; + } + memcpy(sdataPtr->storeNames[0], stName, sizeof(SharedObjectName)); + sdataPtr->largestStoreId = 0; + _SharedMutex_init(&(sdataPtr->mutex), mutexPrivateHandle, 1); + return sdataPtr; +} + +void _SharedAllocData_destroy( SharedAllocData *sdataPtr + , void **privateStoreHandle + , void **mutexPrivateHandle + , const _Bool isLastUser + ){ + _SharedMutex_destroy(isLastUser ? &(sdataPtr->mutex) : NULL, mutexPrivateHandle); + SharedObjectName myName = {0}; + memcpy(myName, sdataPtr->storeNames[0], sizeof(SharedObjectName)); + _store_free( myName + , privateStoreHandle + , sdataPtr + , sizeof(SharedAllocData) + , isLastUser + ); +} + +SharedAllocator *shared_createAllocator(){ + SharedAllocator *aptr = (SharedAllocator*)malloc(sizeof(SharedAllocator)); + if ( aptr == NULL ) { + return NULL; + } + memset(aptr, 0, sizeof(SharedAllocator)); + SharedAllocData *sdataPtr + = _SharedAllocData_init(&(aptr->storePrivateHandles[0]), &(aptr->mutexPrivateHandle)); + if ( sdataPtr == NULL ) { + free(aptr); + return NULL; + } + aptr->sharedAllocData = sdataPtr; + aptr->storeN = 1; + aptr->sharedAllocData->usersN = 1; + return aptr; +} + +SharedAllocator *shared_lookupAllocator(const char *storeName){ + SharedAllocator *aptr = (SharedAllocator*)malloc(sizeof(SharedAllocator)); + if ( aptr == NULL ) { + return NULL; + } + memset(aptr, 0, sizeof(SharedAllocator)); + SharedAllocData *sdataPtr = (SharedAllocData*) + _store_alloc(storeName, &(aptr->storePrivateHandles[0]), sizeof(SharedAllocData)); + if ( sdataPtr == NULL ) { + free(aptr); + return NULL; + } + aptr->sharedAllocData = sdataPtr; + aptr->storeN = 1; + sdataPtr->usersN++; + _SharedMutex_init(&(sdataPtr->mutex), &(aptr->mutexPrivateHandle), 0); + return aptr; +} + +void shared_destroyAllocator(SharedAllocator *aptr) { + SharedAllocData *sdataPtr = aptr->sharedAllocData; + int mlock = _SharedMutex_lock(&(sdataPtr->mutex), &(aptr->mutexPrivateHandle)); + int remainingUsers = --(sdataPtr->usersN); + if(mlock == 0){ + _SharedMutex_unlock(&(sdataPtr->mutex), &(aptr->mutexPrivateHandle)); + } + + // release all stores except the main one + for (HsWord8 storeId = DEFAULT_STORE_SIZE_FACTOR; + storeId <= sdataPtr->largestStoreId; + storeId++) { + _store_free( sdataPtr->storeNames[storeId] + , &(aptr->storePrivateHandles[storeId]) + , aptr->storeAddrs[storeId] + , ((size_t)1) << storeId + , remainingUsers == 0 + ); + } + _SharedAllocData_destroy( sdataPtr + , &(aptr->storePrivateHandles[0]) + , &(aptr->mutexPrivateHandle) + , remainingUsers == 0 + ); + free(aptr); +} + +char *shared_getStoreName(SharedAllocator *aptr){ + return aptr->sharedAllocData->storeNames[0]; +} + + +SharedPtr shared_ptrToShPtr(SharedAllocator *aptr, void *ptr) { + HsWord8 i = aptr->storeN - 1; + HsWord8 storeId = aptr->storeIdsSorted[i]; + while (i > 0 && aptr->storeAddrs[storeId] > ptr) { + storeId = aptr->storeIdsSorted[--i]; + } + + SharedPtr base = (SharedPtr)ptr - (SharedPtr)(aptr->storeAddrs[storeId]); + if (storeId == 0) { + assert(base < DEFAULT_STORE_SIZE); + return base; + } else { + assert(storeId >= DEFAULT_STORE_SIZE_FACTOR && storeId < WORD_SIZE_IN_BITS); + assert(base < (((SharedPtr)1) << storeId)); + return base | (((SharedPtr)1) << storeId); + } +} + +int _shared_initStore(SharedAllocator * const aptr, const HsWord8 storeId); + +// Must return a valid pointer if the system is intact. +// This function returning NULL with valid allocator means implementation is broken. +// If the allocator is erroneous, behavior is undefined. +void *shared_shPtrToPtr(SharedAllocator *aptr, SharedPtr ptr){ + HsWord8 storeId = getStoreId(ptr); + SharedPtr storeAddr = (SharedPtr)(aptr->storeAddrs[storeId]); + if(storeAddr == 0){ + int initSuccess = _shared_initStore(aptr, storeId); + assert(initSuccess == 0); + storeAddr = (SharedPtr)(aptr->storeAddrs[storeId]); + } + if (storeId == 0) { + assert(ptr < DEFAULT_STORE_SIZE); + return (void*)(storeAddr + ptr); + } else { + assert(storeId >= DEFAULT_STORE_SIZE_FACTOR && storeId < WORD_SIZE_IN_BITS); + return (void*)(storeAddr + (ptr & ~(((SharedPtr)1)< 0 +// returns: +// 0 on success +// 1 otherwise +int _shared_initStore(SharedAllocator * const aptr, const HsWord8 storeId){ + // do nothing if store already exists + if ( aptr->storeAddrs[storeId] != 0 || storeId == 0) { + return 0; + } + assert(storeId >= DEFAULT_STORE_SIZE_FACTOR && storeId < WORD_SIZE_IN_BITS); + SharedAllocData *sdataPtr = aptr->sharedAllocData; + HsPtr storePtr; + size_t storeSize = ((size_t)1) << storeId; + // if store already exists, just add it to local context + if (sdataPtr->storeNames[storeId][0] != '\0'){ + storePtr + = _store_alloc( sdataPtr->storeNames[storeId], &(aptr->storePrivateHandles[storeId]), storeSize ); + if (storePtr == NULL) { + return 1; + } + } else { + // generate a new unique name for the store + genSharedObjectName( sdataPtr->storeNames[storeId] ); + // allocate store + storePtr + = _store_alloc( sdataPtr->storeNames[storeId], &(aptr->storePrivateHandles[storeId]), storeSize ); + if (storePtr == NULL) { + return 1; + } + // init first free block in the newly created store + ListNode *firstNode = &(sdataPtr->availStorage[storeId]); + ListNode *secondNode = (ListNode*)shared_shPtrToPtr(aptr, firstNode->linkF); + // last byte of the storage stands for (storePtr-1) + *((HsWord8*)storePtr + storeSize - 1) + = storeId | 0x80; // first bit is flag, others are free storage size (==storeSize) + ListNode *newNode = (ListNode*)storePtr; + SharedPtr newNodePtr = (SharedPtr)storeSize; + newNode->linkF = firstNode->linkF; + newNode->linkB = secondNode->linkB; + // insert an availBlock node into the list + firstNode->linkF = newNodePtr; + secondNode->linkB = newNodePtr; + // insert update shared data + if (storeId > sdataPtr->largestStoreId) { + sdataPtr->largestStoreId = storeId; + } + } + aptr->storeAddrs[storeId] = storePtr; + // bubble-sort-insert new store into the sorted pool + HsWord8 i = aptr->storeN; + while (i > 0 && storePtr + < (aptr->storeAddrs[aptr->storeIdsSorted[i-1]]) ) { + aptr->storeIdsSorted[i] = aptr->storeIdsSorted[i-1]; + i--; + } + aptr->storeIdsSorted[i] = storeId; + aptr->storeN++; + return 0; +} + + + +static inline void *_shared_malloc(SharedAllocator *aptr, size_t size) { + HsWord8 storeId; + HsWord8 allocSizeBit = allocFactor(size); + HsWord8 availSizeBit = allocSizeBit; + ListNode* nptr = &(aptr->sharedAllocData->availStorage[availSizeBit]); + // if SharedPtr value is less than DEFAULT_STORE_SIZE, it points to the default store, + // which implies the list is empty. + // {while list is empty and we cannot init store of this size} + while ( nptr->linkF < DEFAULT_STORE_SIZE + && ( availSizeBit < DEFAULT_STORE_SIZE_FACTOR + || aptr->sharedAllocData->storeNames[availSizeBit][0] != '\0' + ) + ) { + availSizeBit++; + nptr = &(aptr->sharedAllocData->availStorage[availSizeBit]); + } + // By now, either we found a free block, or we can create one + if ( nptr->linkF < DEFAULT_STORE_SIZE ) { + assert(availSizeBit >= DEFAULT_STORE_SIZE_FACTOR); + if (_shared_initStore(aptr,availSizeBit) != 0) { + return NULL; + } + nptr = &(aptr->sharedAllocData->availStorage[availSizeBit]); + } + storeId = getStoreId(nptr->linkF); + assert(storeId >= DEFAULT_STORE_SIZE_FACTOR && storeId < WORD_SIZE_IN_BITS); + size_t storeSize = ((size_t)1) << storeId; + // By now, nptr list is not empty, and all lists allocSizeBit..availSizeBit-1 are empty. + ListNode *allocNodePtr, *secondNodePtr; + HsWord8* storePtr = aptr->storeAddrs[storeId]; + if (storePtr == 0) { + if (_shared_initStore(aptr, storeId) != 0) { + return NULL; + } else { + storePtr = aptr->storeAddrs[storeId]; + } + } + // Now, nptr points to the beginning of non-empty list of free nodes of required size. + // Allocate a node! + allocNodePtr = (ListNode*)shared_shPtrToPtr(aptr, nptr->linkF); + + // remove node from the list + secondNodePtr = (ListNode*)shared_shPtrToPtr(aptr, allocNodePtr->linkF); + secondNodePtr->linkB = allocNodePtr->linkB; + nptr->linkF = allocNodePtr->linkF; + + // Check if we need to split the allocated region into buddies + ListNode *buddyNodePtr; + SharedPtr buddySharedPtr; + while ( availSizeBit > allocSizeBit ) { + + availSizeBit--; + nptr = &(aptr->sharedAllocData->availStorage[availSizeBit]); + // add buddy to corresponding list + buddyNodePtr = (ListNode*)((SharedPtr)allocNodePtr + (((SharedPtr)1) << availSizeBit)); + buddySharedPtr = (SharedPtr)(((size_t)buddyNodePtr - (size_t)storePtr) | storeSize); + *buddyNodePtr = (struct ListNode) + { .linkB = nptr->linkB + , .linkF = nptr->linkF + }; + nptr->linkF = buddySharedPtr; + nptr->linkB = buddySharedPtr; + *(((HsWord8*)buddyNodePtr) - 1) = availSizeBit | 0x80; + } + // set flag to occupied + *STORAGE_TAG_BYTE_PTR(storePtr, storeSize, allocNodePtr) = allocSizeBit; + // return it! + return (void*)allocNodePtr; +} + +static inline void _shared_free(SharedAllocator *aptr, void *ptr) { + ListNode *myPtr = (ListNode*)ptr; + SharedPtr mySharedPtr = shared_ptrToShPtr(aptr, ptr); + HsWord8 storeId = getStoreId(mySharedPtr); + assert(storeId >= DEFAULT_STORE_SIZE_FACTOR && storeId < WORD_SIZE_IN_BITS); + size_t storeSize = ((size_t)1) << storeId; // guaranteed: storeId != 0 + HsWord8 *storePtr = aptr->storeAddrs[storeId]; + HsWord8 nodeSizeBit = *STORAGE_TAG_BYTE_PTR(storePtr, storeSize, ptr); + assert(nodeSizeBit >= MIN_ALLOC_FACTOR && nodeSizeBit <= storeId); + // join buddies if needed + ListNode *buddyPtr + = (ListNode*)( (((SharedPtr)myPtr - (SharedPtr)storePtr) ^ (((SharedPtr)1) << nodeSizeBit)) + + (SharedPtr)storePtr ); + + while ( nodeSizeBit < storeId + && *STORAGE_TAG_BYTE_PTR(storePtr, storeSize, buddyPtr) == (nodeSizeBit | 0x80) + ) { + // remove buddy from corresponding list; + ListNode *tmpNode; + tmpNode = (ListNode *)shared_shPtrToPtr(aptr, buddyPtr->linkF); + tmpNode->linkB = buddyPtr->linkB; + tmpNode = (ListNode *)shared_shPtrToPtr(aptr, buddyPtr->linkB); + tmpNode->linkF = buddyPtr->linkF; + // take the left one further + if(buddyPtr < myPtr) { + myPtr = buddyPtr; + mySharedPtr = ((SharedPtr)myPtr - (SharedPtr)storePtr) | (SharedPtr)storeSize; + } + nodeSizeBit++; + buddyPtr + = (ListNode*)( (((SharedPtr)myPtr - (SharedPtr)storePtr) ^ (((SharedPtr)1) << nodeSizeBit)) + + (SharedPtr)storePtr ); + } + *STORAGE_TAG_BYTE_PTR(storePtr, storeSize, myPtr) = nodeSizeBit | 0x80; + // Now, there are no buddies to join; add the node to the pool + ListNode *firstNodePtr, *secondNodePtr; + firstNodePtr = &(aptr->sharedAllocData->availStorage[nodeSizeBit]); + secondNodePtr = (ListNode*)shared_shPtrToPtr(aptr, firstNodePtr->linkF); + myPtr->linkF = firstNodePtr->linkF; + myPtr->linkB = secondNodePtr->linkB; + firstNodePtr->linkF = mySharedPtr; + secondNodePtr->linkB = mySharedPtr; +} + + +static inline void *_shared_realloc(SharedAllocator *aptr, void *ptr, size_t size) { + SharedPtr nodeSharedPtr = shared_ptrToShPtr(aptr, ptr); + HsWord8 storeId = getStoreId(nodeSharedPtr); + size_t storeSize = ((size_t)1) << storeId; // guaranteed: storeId != 0 + HsWord8 *storePtr = aptr->storeAddrs[storeId]; + HsWord8 nodeSizeBit = *STORAGE_TAG_BYTE_PTR(storePtr, storeSize, ptr); + HsWord8 allocSizeBit = allocFactor(size); + + void *buddyPtr; + // free extra memory if not needed anymore + while ( nodeSizeBit > allocSizeBit) { + nodeSizeBit--; + buddyPtr + = (void*)( ( ( (SharedPtr)ptr - (SharedPtr)storePtr ) + ^ (((SharedPtr)1) << nodeSizeBit) + ) + (SharedPtr)storePtr ); + if ( buddyPtr < ptr ) { + nodeSizeBit++; + break; + } + *STORAGE_TAG_BYTE_PTR(storePtr, storeSize, buddyPtr) = nodeSizeBit; + *STORAGE_TAG_BYTE_PTR(storePtr, storeSize, ptr) = nodeSizeBit; + _shared_free(aptr, buddyPtr); + } + // use buddy memory if available + if (allocSizeBit <= storeId) { + ListNode *buddyPrev, *buddyNext; + while ( nodeSizeBit < allocSizeBit ) { + buddyPtr + = (void*)( ( ( (SharedPtr)ptr - (SharedPtr)storePtr ) + ^ (((SharedPtr)1) << nodeSizeBit) + ) + (SharedPtr)storePtr ); + if ( buddyPtr < ptr || (*STORAGE_TAG_BYTE_PTR(storePtr, storeSize, buddyPtr) != (nodeSizeBit | 0x80)) ) { + break; + } + // remove node from the list + buddyPrev = (ListNode*)shared_shPtrToPtr(aptr, ((ListNode*)buddyPtr)->linkB); + buddyNext = (ListNode*)shared_shPtrToPtr(aptr, ((ListNode*)buddyPtr)->linkF); + buddyPrev->linkF = ((ListNode*)buddyPtr)->linkF; + buddyNext->linkB = ((ListNode*)buddyPtr)->linkB; + nodeSizeBit++; + *STORAGE_TAG_BYTE_PTR(storePtr, storeSize, ptr) = nodeSizeBit; + } + } + // return the same pointer if the size matches + if ( nodeSizeBit == allocSizeBit ) { + return ptr; + } + // copy data because cheaper options failed + void *rptr = _shared_malloc(aptr, size); + if ( rptr == NULL ) { + return NULL; + } + size_t toCopy = (((size_t)1) << nodeSizeBit) - 1; + if (toCopy > size) { + toCopy = size; + } + memcpy( rptr, ptr, toCopy); + _shared_free(aptr, ptr); + + return rptr; +} + + + + +void *shared_malloc(SharedAllocator *aptr, size_t size){ + int mlock = _SharedMutex_lock(&(aptr->sharedAllocData->mutex), &(aptr->mutexPrivateHandle)); + if ( mlock != 0 ) { + return NULL; + } + void* r = _shared_malloc(aptr, size); + mlock = _SharedMutex_unlock(&(aptr->sharedAllocData->mutex), &(aptr->mutexPrivateHandle)); + if ( mlock != 0 ) { + return NULL; + } + return r; +} + +void *shared_realloc(SharedAllocator *aptr, void *ptr, size_t size) { + int mlock = _SharedMutex_lock(&(aptr->sharedAllocData->mutex), &(aptr->mutexPrivateHandle)); + if ( mlock != 0 ) { + return NULL; + } + void* r = _shared_realloc(aptr, ptr, size); + mlock = _SharedMutex_unlock(&(aptr->sharedAllocData->mutex), &(aptr->mutexPrivateHandle)); + if ( mlock != 0 ) { + return NULL; + } + return r; +} + +void shared_free(SharedAllocator *aptr, void *ptr) { + int mlock = _SharedMutex_lock(&(aptr->sharedAllocData->mutex), &(aptr->mutexPrivateHandle)); + _shared_free(aptr, ptr); + if ( mlock == 0 ) { + _SharedMutex_unlock(&(aptr->sharedAllocData->mutex), &(aptr->mutexPrivateHandle)); + } +} diff --git a/src/Foreign/SharedPtr.hs b/src/Foreign/SharedPtr.hs new file mode 100644 --- /dev/null +++ b/src/Foreign/SharedPtr.hs @@ -0,0 +1,91 @@ +module Foreign.SharedPtr + ( SharedPtr (), toSharedPtr, fromSharedPtr + , Allocator + , createAllocator, lookupAllocator, destroyAllocator + , withNewAllocator, withAllocator, allocStoreName + , malloc, mallocBytes, realloc, free + ) where + +import Control.Exception (bracket) +import Foreign.C.Error +import Foreign.Marshal.Utils +import Foreign.Ptr +import Foreign.SharedObjectName.Internal +import Foreign.SharedPtr.C +import Foreign.Storable +import System.IO.Unsafe (unsafePerformIO) + + +-- | Make a portable shared pointer out of a regular pointer. +-- The result can be transfered to another process and re-created using +-- the shared `Allocator`. +toSharedPtr :: Allocator -> Ptr a -> SharedPtr a +toSharedPtr = c'shared_ptrToShPtr + +-- | Reconstruct a regular pointer from a portable shared pointer. +-- Returns @NULL@ if shared pointer or allocator are not valid. +fromSharedPtr :: Allocator -> SharedPtr a -> Ptr a +fromSharedPtr = c'shared_shPtrToPtr + +-- | Create a new `Allocator`. +createAllocator :: IO Allocator +createAllocator = checkNullPointer "SharedPtr.createAllocator" + c'shared_createAllocator +{-# INLINE createAllocator #-} + + +-- | Lookup a `Allocator` by its name. +-- Use this to share one allocator between multiple processes. +lookupAllocator :: SOName Allocator -> IO Allocator +lookupAllocator = checkNullPointer "SharedPtr.lookupAllocator" + . flip unsafeWithSOName c'shared_lookupAllocator +{-# INLINE lookupAllocator #-} + +-- | Destroy allocator instance. +-- Note: memory is fully unlinked and released only after +-- the last allocator sharing the memory is destroyed. +destroyAllocator :: Allocator -> IO () +destroyAllocator = c'shared_destroyAllocator +{-# INLINE destroyAllocator #-} + +withNewAllocator :: (Allocator -> IO a) -> IO a +withNewAllocator = bracket createAllocator destroyAllocator +{-# INLINE withNewAllocator #-} + +withAllocator :: SOName Allocator -> (Allocator -> IO a) -> IO a +withAllocator s = bracket (lookupAllocator s) destroyAllocator +{-# INLINE withAllocator #-} + +allocStoreName :: Allocator -> SOName Allocator +allocStoreName a = unsafePerformIO $ do + n <- newEmptySOName + unsafeWithSOName n $ + \p -> copyBytes p (c'shared_getStoreName a) (sizeOf n) + return n +{-# NOINLINE allocStoreName #-} + + +malloc :: Storable a => Allocator -> IO (Ptr a) +malloc a = go undefined + where + go :: Storable b => b -> IO (Ptr b) + go x = mallocBytes a (sizeOf x) + +mallocBytes :: Allocator -> Int -> IO (Ptr a) +mallocBytes a = checkNullPointer "SharedPtr.malloc" + . c'shared_malloc a . fromIntegral + +realloc :: Allocator -> Ptr a -> Int -> IO (Ptr a) +realloc a p = checkNullPointer "SharedPtr.realloc" + . c'shared_realloc a p . fromIntegral + +free :: Allocator -> Ptr a -> IO () +free = c'shared_free + + +checkNullPointer :: String -> IO (Ptr a) -> IO (Ptr a) +checkNullPointer s k = do + p <- k + if p == nullPtr + then throwErrno (s ++ " returned NULL pointer.") + else return p diff --git a/src/Foreign/SharedPtr/C.hs b/src/Foreign/SharedPtr/C.hs new file mode 100644 --- /dev/null +++ b/src/Foreign/SharedPtr/C.hs @@ -0,0 +1,107 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE RoleAnnotations #-} +module Foreign.SharedPtr.C + ( SharedPtr (), Allocator + + , c'shared_createAllocator, c'shared_lookupAllocator + , c'shared_destroyAllocator, c'shared_getStoreName + , c'shared_ptrToShPtr, c'shared_shPtrToPtr + , c'shared_malloc, c'shared_realloc, c'shared_free + + , p'shared_createAllocator, p'shared_lookupAllocator + , p'shared_destroyAllocator, p'shared_getStoreName + , p'shared_ptrToShPtr, p'shared_shPtrToPtr + , p'shared_malloc, p'shared_realloc, p'shared_free + ) where + +import Data.Data (Data) +import Foreign.C.String +import Foreign.C.Types +import Foreign.Ptr +import Foreign.Storable +import GHC.Generics (Generic) + +type role SharedPtr phantom +-- | Special pointer format to pass between memory spaces of processes. +newtype SharedPtr a = SharedPtr ( Ptr a ) + deriving (Eq, Ord, Show, Data, Generic, Storable) + +data AllocatorT +-- | Opaque pointer to the allocator type defined in C code. +type Allocator = Ptr AllocatorT + + +foreign import ccall unsafe "&shared_createAllocator" + p'shared_createAllocator + :: FunPtr (IO Allocator) + +foreign import ccall unsafe "&shared_lookupAllocator" + p'shared_lookupAllocator + :: FunPtr (CString -> IO Allocator) + +foreign import ccall unsafe "&shared_destroyAllocator" + p'shared_destroyAllocator + :: FunPtr (Allocator -> IO ()) + +foreign import ccall unsafe "&shared_getStoreName" + p'shared_getStoreName + :: FunPtr (Allocator -> CString) + +foreign import ccall unsafe "&shared_ptrToShPtr" + p'shared_ptrToShPtr + :: FunPtr (Allocator -> Ptr a -> SharedPtr a) + +foreign import ccall unsafe "&shared_shPtrToPtr" + p'shared_shPtrToPtr + :: FunPtr (Allocator -> SharedPtr a -> Ptr a) + +foreign import ccall unsafe "&shared_malloc" + p'shared_malloc + :: FunPtr (Allocator -> CSize -> IO (Ptr a)) + +foreign import ccall unsafe "&shared_realloc" + p'shared_realloc + :: FunPtr (Allocator -> Ptr a -> CSize -> IO (Ptr a)) + +foreign import ccall unsafe "&shared_free" + p'shared_free + :: FunPtr (Allocator -> Ptr a -> IO ()) + + +foreign import ccall unsafe "shared_createAllocator" + c'shared_createAllocator + :: IO Allocator + +foreign import ccall unsafe "shared_lookupAllocator" + c'shared_lookupAllocator + :: CString -> IO Allocator + +foreign import ccall unsafe "shared_destroyAllocator" + c'shared_destroyAllocator + :: Allocator -> IO () + +foreign import ccall unsafe "shared_getStoreName" + c'shared_getStoreName + :: Allocator -> CString + +foreign import ccall unsafe "shared_ptrToShPtr" + c'shared_ptrToShPtr + :: Allocator -> Ptr a -> SharedPtr a + +foreign import ccall unsafe "shared_shPtrToPtr" + c'shared_shPtrToPtr + :: Allocator -> SharedPtr a -> Ptr a + +foreign import ccall unsafe "shared_malloc" + c'shared_malloc + :: Allocator -> CSize -> IO (Ptr a) + +foreign import ccall unsafe "shared_realloc" + c'shared_realloc + :: Allocator -> Ptr a -> CSize -> IO (Ptr a) + +foreign import ccall unsafe "shared_free" + c'shared_free + :: Allocator -> Ptr a -> IO () diff --git a/src/Foreign/SharedPtrPosix.c b/src/Foreign/SharedPtrPosix.c new file mode 100644 --- /dev/null +++ b/src/Foreign/SharedPtrPosix.c @@ -0,0 +1,67 @@ +#include "SharedPtr.h" + +#include +#include +#include +#include + + +void _SharedMutex_init(SharedMutex *mptr, void **privateMutexHandle, const int createNew) { + *privateMutexHandle = (void*)mptr; + if (createNew != 0) { + pthread_mutexattr_init(&(mptr->mutAttr)); + // pthread_mutexattr_settype(&(mptr->mutAttr), PTHREAD_MUTEX_ERRORCHECK); + pthread_mutexattr_setpshared(&(mptr->mutAttr), PTHREAD_PROCESS_SHARED); + pthread_mutex_init(&(mptr->mutVal), &(mptr->mutAttr)); + } +} + +void _SharedMutex_destroy(SharedMutex *mptr, void **privateMutexHandle) { + *privateMutexHandle = NULL; + if ( mptr != NULL ) { + pthread_mutex_destroy(&(mptr->mutVal)); + pthread_mutexattr_destroy(&(mptr->mutAttr)); + } +} + +int _SharedMutex_lock(SharedMutex *mptr, void **privateMutexHandle) { + return pthread_mutex_lock(&(mptr->mutVal)); +} + +int _SharedMutex_unlock(SharedMutex *mptr, void **privateMutexHandle) { + return pthread_mutex_unlock(&(mptr->mutVal)); +} + +// returns NULL if failed +HsPtr _store_alloc(const char *memBlockName, void **privateStoreHandle, size_t size) { + int memFd = shm_open(memBlockName, O_CREAT | O_RDWR, S_IRWXU); + if (memFd < 0) { + return NULL; + } + int res = ftruncate(memFd, size); + if (res != 0) { + shm_unlink(memBlockName); + return NULL; + } + HsPtr r = mmap( NULL + , size + , PROT_READ | PROT_WRITE + , MAP_SHARED + , memFd, 0); + if (r == MAP_FAILED) { + shm_unlink(memBlockName); + return NULL; + } + return r; +} + +// failures of system calls are ignored +void _store_free( const char *memBlockName, void **privateStoreHandle, HsPtr addr + , size_t size, _Bool unlinkToo) { + if(addr != 0){ + munmap(addr, size); + } + if(unlinkToo && memBlockName[0] != '\0') { + shm_unlink(memBlockName); + } +} diff --git a/src/Foreign/SharedPtrWin32.c b/src/Foreign/SharedPtrWin32.c new file mode 100644 --- /dev/null +++ b/src/Foreign/SharedPtrWin32.c @@ -0,0 +1,91 @@ +#include "SharedPtr.h" +#ifdef STDOUT_SYSCALL_DEBUG +#include +#include +#endif + +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 ); +#ifdef STDOUT_SYSCALL_DEBUG + if (*privateMutexHandle == NULL) { + _tprintf(TEXT("CreateMutex error: %d\n"), GetLastError()); + } +#endif +} + +void _SharedMutex_destroy(SharedMutex *mptr, void **privateMutexHandle) { + CloseHandle(*privateMutexHandle); +} + +int _SharedMutex_lock(SharedMutex *mptr, void **privateMutexHandle) { + DWORD r = WaitForSingleObject(*privateMutexHandle, INFINITE); + if (r != WAIT_OBJECT_0) { +#ifdef STDOUT_SYSCALL_DEBUG + _tprintf(TEXT("WaitForSingleObject mutex error: return %d; error code %d.\n"), r, GetLastError()); +#endif + return 1; + } else { + return 0; + } +} + +int _SharedMutex_unlock(SharedMutex *mptr, void **privateMutexHandle) { + DWORD r = ReleaseMutex(*privateMutexHandle); + if(r == 0) { +#ifdef STDOUT_SYSCALL_DEBUG + _tprintf(TEXT("ReleaseMutex error: %d\n"), GetLastError()); +#endif + return 1; + } else { + 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 + + if (*privateMutexHandle == NULL) { +#ifdef STDOUT_SYSCALL_DEBUG + _tprintf(TEXT("Could not create file mapping object (%d).\n"), GetLastError()); +#endif + return NULL; + } + void *rptr = MapViewOfFile + ( *privateMutexHandle // handle to map object + , FILE_MAP_ALL_ACCESS // read/write permission + , 0 + , 0 + , size); + + if (rptr == NULL) { +#ifdef STDOUT_SYSCALL_DEBUG + _tprintf(TEXT("Could not map view of file (%d).\n"), GetLastError()); +#endif + CloseHandle(*privateMutexHandle); + return NULL; + } + + return rptr; +} + +// failures of system calls are ignored +void _store_free( const char *memBlockName, void **privateStoreHandle, HsPtr addr + , size_t size, _Bool unlinkToo) { + if(addr != 0){ + UnmapViewOfFile(addr); + CloseHandle(*privateStoreHandle); + } +}