interprocess 0.1.0.0 → 0.2.0.0

raw patch · 17 files changed

+1731/−274 lines, 17 filesdep ~basenew-component:exe:concurrent-mallocnew-component:exe:wait-mvarnew-component:exe:wait-qsemPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• README.md +1/−1
• app/concurrent-malloc.hs +0/−159
• app/wait-qsem.hs +0/−87
• examples/concurrent-malloc.hs +159/−0
• examples/wait-mvar.hs +133/−0
• examples/wait-qsem.hs +89/−0
• interprocess.cabal +34/−10
• src/Control/Concurrent/Process/QSem.c +0/−1
• src/Control/Concurrent/Process/QSem.hs +2/−0
• src/Control/Concurrent/Process/StoredMVar.c +746/−0
• src/Control/Concurrent/Process/StoredMVar.hs +336/−0
• src/Foreign/SharedObjectName/Internal.hs +18/−1
• src/Foreign/SharedPtr.c +11/−0
• src/Foreign/SharedPtr/C.hs +22/−2
• src/Foreign/SharedPtrPosix.c +1/−1
• src/Foreign/SharedPtrWin32.c +11/−12
• test/StoredMVar.hs +168/−0

@@ -12,7 +12,7 @@         that can be accessed by multiple processes.   * [x] Semaphores   * [ ] Mutexes (not sure if need this)-  * [ ] Mutable variables via `Storable` instance plus garbage collection.+  * [x] Mutable variables via `Storable` instance plus garbage collection.   * [ ] Proper error messages   * [ ] Debug, verbose mode 

@@ -1,159 +0,0 @@-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

@@ -1,87 +0,0 @@-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:))

@@ -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+    execFile <- getExecutablePath+    args <- getArgs+    let processBConfig = setStdin createPipe+                       $ proc execFile ("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

@@ -0,0 +1,133 @@+module Main where++import           Control.Concurrent.Process.StoredMVar+import           Control.Exception                     (SomeException, catch,+                                                        displayException)+import           Control.Monad+import           Data.List                             (partition)+import           Data.Maybe                            (fromMaybe)+import           Data.Monoid                           (First (..))+import           Foreign.SharedObjectName+import           GHC.Environment                       (getFullArgs)+import           System.Environment+import           System.Exit+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+      i = fromMaybe 1 . getFirst $ foldMap (First . readMaybe) remargs+  if isSlave+  then runB i `catch` ( \e -> do+                          putStrLn $ "[" ++ show i ++ "] "+                                            ++ displayException (e :: SomeException)+                          exitFailure+                      )+  else runA i `catch` ( \e -> do+                          putStrLn $ "[A] " ++ displayException (e :: SomeException)+                          exitFailure+                      )++runA :: Int -> IO ()+runA n = do+    execFile <- getExecutablePath+    args <- getFullArgs+    let pconfs = flip map [1..n] $ \i -> setStdin createPipe+                                 $ proc execFile ("slave" : show i : args)+    mVar <- newEmptyMVar :: IO (StoredMVar Double)++    withProcesses pconfs $ \procs -> do++      let mvName = mVarName mVar+      report $ "Created mVar: " ++ show mvName++      -- send name of StoredMVar to every slave process+      forM_ procs $ \p -> do+        hPutSOName (getStdin p) mvName+        hFlush (getStdin p)++      report "Sent MVar name. Now putMVar n times"+      forM_ (zipWith const [1..] procs) $ \i -> do+        let x = recip i+        report $ "Putting " ++ show x+        putMVar mVar x+        report $ "Have put " ++ show x ++ " (" ++ show i ++ "-th)"++      report "Have put. Waiting"++    report "Finished successfully"+  where+    report s =  putStrLn $ "[A] " ++ s+++runB :: Int -> IO ()+runB i = do+    Just mVarRef <- hGetSOName stdin  -- get name of a StoredMVar+    report "Started."++    mVar <- lookupMVar mVarRef :: IO (StoredMVar Double)+    takePutMTimes mVar 20+    report "Taking last time."+    v <- takeMVar mVar+    report $ show v ++ " Taken last time."+    report "Finished successfully"+  where+    takePutMTimes mVar m = go (1 :: Int)+      where+        go k | k > m = return ()+             | mod k 5 == 3 = do+                report $ show k ++ " Swapping."+                a <- swapMVar mVar 17.5+                report $ show k ++ " Have swapped " ++ show a+                go (k+1)+             | mod k 3 == 2 = do+                report $ show k ++ " Reading."+                a <- readMVar mVar+                report $ show k ++ " Have read " ++ show a+                go (k+1)+             | mod k 7 == 6 = do+                report $ show k ++ " Try taking."+                ma <- tryTakeMVar mVar+                case ma of+                  Just a -> do+                    report $ show k ++ " Trytaken succesfully. Try putting."+                    putSuccess <- tryPutMVar mVar (a + 3)+                    if putSuccess+                      then do+                        report $ show k ++ " I'm lucky! Try swapping now."+                        mb <- trySwapMVar mVar (a - 7)+                        case mb of+                          Just b -> do+                            report $ show k ++ " Bingo! " ++ show (a, b)+                            go (k+1)+                          Nothing -> do+                            report $ show k ++ " Failed to tryswap."+                            go (k+1)+                      else do+                        report $ show k ++ " Could not tryput. Wait and put now."+                        putMVar mVar (a + 1)+                        go (k+1)+                  Nothing -> do+                    report $ show k ++ " Could not trytake. Repeat."+                    go k+             | otherwise = do+                report $ show k ++ " Taking."+                a <- takeMVar mVar+                report $ show k ++ " Taken " ++ show a ++ ". Putting."+                putMVar mVar (a * 2)+                report $ show k ++ " Have put."+                go (k+1)++    report s =  putStrLn $ "[" ++ show i ++ "] " ++ s++++withProcesses ::  [ProcessConfig stdin stdout stderr]+              -> ([Process stdin stdout stderr] -> IO a)+              -> IO a+withProcesses [] k = k []+withProcesses (conf:cfs) k = withProcesses cfs $ \ps -> withProcess_ conf $ k . (:ps)

@@ -0,0 +1,89 @@+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+    execFile <- getExecutablePath+    args <- getArgs+    let processBConfig = setStdin createPipe+                       $ proc execFile ("slave":args)++    putStrLn "[A] Starting..."+    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:))

@@ -1,7 +1,7 @@ name:                interprocess-version:             0.1.0.0+version:             0.2.0.0 synopsis:            Shared memory and control structures for IPC-description:         Provides portable shared memory allocator and semaphores.+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@@ -18,6 +18,7 @@     cbits/SharedObjectName.c     include/SharedObjectName.h     include/SharedPtr.h+    src/Control/Concurrent/Process/StoredMVar.c     src/Control/Concurrent/Process/QSem.c     src/Foreign/SharedPtrPosix.c     src/Foreign/SharedPtrWin32.c@@ -39,11 +40,13 @@                        Foreign.SharedObjectName.Internal                        Foreign.SharedPtr.C                        Control.Concurrent.Process.QSem-    build-depends:     base >= 4.7 && < 5+                       Control.Concurrent.Process.StoredMVar+    build-depends:     base >= 4.8 && < 5     default-language:  Haskell2010     ghc-options:       -Wall     include-dirs:      include     c-sources:         cbits/SharedObjectName.c+                       src/Control/Concurrent/Process/StoredMVar.c                        src/Control/Concurrent/Process/QSem.c                        src/Foreign/SharedPtr.c     if os(windows)@@ -51,31 +54,52 @@     else       c-sources:       src/Foreign/SharedPtrPosix.c     if flag(dev)-        cpp-options:   -DSTDOUT_SYSCALL_DEBUG         ghc-options:   -O0     else         cpp-options:   -DNDEBUG+        cc-options:    -DNDEBUG         ghc-options:   -O2  -executable interprocess-concurrent-malloc+executable concurrent-malloc     if !flag(examples)       buildable:         False-    hs-source-dirs:      app+    hs-source-dirs:      examples     main-is:             concurrent-malloc.hs     default-language:    Haskell2010-    build-depends:       base >= 4.7 && < 5+    build-depends:       base                        , typed-process >= 0.2                        , interprocess     ghc-options:         -threaded -Wall -executable interprocess-wait-qsem+executable wait-qsem     if !flag(examples)       buildable:         False-    hs-source-dirs:      app+    hs-source-dirs:      examples     main-is:             wait-qsem.hs     default-language:    Haskell2010-    build-depends:       base >= 4.7 && < 5+    build-depends:       base+                       , typed-process >= 0.2+                       , interprocess+    ghc-options:         -threaded -Wall++executable wait-mvar+    if !flag(examples)+      buildable:         False+    hs-source-dirs:      examples+    main-is:             wait-mvar.hs+    default-language:    Haskell2010+    build-depends:       base+                       , typed-process >= 0.2+                       , interprocess+    ghc-options:         -threaded -Wall++test-suite StoredMVar+    type:                exitcode-stdio-1.0+    main-is:             StoredMVar.hs+    default-language:    Haskell2010+    hs-source-dirs:      test+    build-depends:       base                        , typed-process >= 0.2                        , interprocess     ghc-options:         -threaded -Wall

@@ -86,7 +86,6 @@ #include <fcntl.h>           /* For O_* constants */ #include <semaphore.h> #include <string.h>-#include "HsFFI.h"  #define GuardNameSuffix "X" 

@@ -7,6 +7,7 @@   ) where  import           Control.Monad                     (when)+import           Data.Data                         (Typeable) import           Foreign.C.Error import           Foreign.C.String import           Foreign.C.Types@@ -21,6 +22,7 @@ -- | 'QSem' is a quantity semaphore in which the resource is aqcuired --   and released in units of one. data QSem = QSem !(SOName QSem) !(ForeignPtr QSemT)+  deriving (Eq, Typeable)  -- | Build a new 'QSem' with a supplied initial quantity. --   The initial quantity must be at least 0.

@@ -0,0 +1,746 @@+#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);+}

@@ -0,0 +1,336 @@+{-# LANGUAGE InterruptibleFFI    #-}+{-# LANGUAGE ScopedTypeVariables #-}+-----------------------------------------------------------------------------+-- |+--   This module is an adaptation of `Control.Concurrent.MVar` to an+--   interprocess communication (IPC).+--   The IPC setting implies a few changes to the interface.+--+--   1. `StoredMVar` resides in a shared memory region.+--+--   2. We use `Storable` instance to serialize and deserialize a value.+--+--   3. Point (2) implies the value is always fully evaluated before being stored.+--+--   4. Scheduling is done by OS, thus the module does not guarantee FIFO order.+--+--   5. Using `StoredMVar` is only safe if `Storable` instance for its content+--      is correct and `peek` does not throw exceptions.+--      If `peek` throws an exception inside `takeMVar` or `swapMVar`,+--      the original content of `StoredMVar` is not restored+--+-----------------------------------------------------------------------------+module Control.Concurrent.Process.StoredMVar+  ( StoredMVar (), mVarName+  , newEmptyMVar, newMVar, lookupMVar+  , takeMVar, putMVar, readMVar, swapMVar+  , tryTakeMVar, tryPutMVar, tryReadMVar, trySwapMVar+  , isEmptyMVar+  , withMVar, withMVarMasked+  , 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++-- | Opaque implementation-dependent StoredMVar+data StoredMVarT++-- | An 'StoredMVar' is a synchronising variable, used+--   for communication between concurrent processes or threads.+--   It can be thought of as a a box, which may be empty or full.+--+--   @StoredMVar@ tries to mimic vanilla `MVar`, though it behaves quite differently.+--   It uses `Storable` instance to make the value accessible in different memory spaces.+--   Thus, the content of @StoredMVar@ is forced to be fully evaluated and serialized.+data StoredMVar a+  = StoredMVar !(SOName (StoredMVar a)) !(ForeignPtr StoredMVarT)+  deriving (Eq, Typeable)+++-- | 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)+    n <- newEmptySOName+    unsafeWithSOName n $ c'mvar_name mvar+    StoredMVar n <$> newForeignPtr p'mvar_destroy mvar++-- | Create a 'StoredMVar' which is initially empty.+newMVar :: Storable a => a -> IO (StoredMVar a)+newMVar value = do+    x <- newEmptyMVar+    putMVar x value+    return x+++-- | Find a `StoredMVar` created in another process ot thread by its reference.+lookupMVar :: Storable a => SOName (StoredMVar a) ->  IO (StoredMVar a)+lookupMVar n = mask_ $ do+    mvar <- unsafeWithSOName n $ checkNullPointer "lookupMVar". c'mvar_lookup+    StoredMVar n <$> newForeignPtr p'mvar_destroy mvar++-- | Get a global reference to the `StoredMVar`.+--   Send this reference to another process to lookup this `StoredMVar` and+--   start interprocess communication.+mVarName :: StoredMVar a -> SOName (StoredMVar a)+mVarName (StoredMVar r _) = r+{-# INLINE mVarName #-}++-- | Check whether a given 'StoredMVar' is empty.+--+--   Notice that the boolean value returned  is just a snapshot of+--   the state of the MVar. By the time you get to react on its result,+--   the MVar may have been filled (or emptied) - so be extremely+--   careful when using this operation.  Use 'tryTakeMVar' instead if possible.+isEmptyMVar :: StoredMVar a -> IO Bool+isEmptyMVar (StoredMVar _ fp) = withForeignPtr fp $ fmap (0 /=) . c'mvar_isempty+{-# INLINE isEmptyMVar #-}+++-- | Return the contents of the 'StoredMVar'.  If the 'StoredMVar' is currently+--   empty, 'takeMVar' will wait until it is full.  After a 'takeMVar',+--   the 'StoredMVar' is left empty.+--+--+--   * 'takeMVar' is single-wakeup.  That is, if there are multiple+--     processes blocked in 'takeMVar', and the 'StoredMVar' becomes full,+--     only one thread will be woken up.+--+--   * The library makes no guarantees about the order in which processes+--     are woken up. This is all up to implementation-dependent OS scheduling.+--+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+{-# INLINE takeMVar #-}+++-- | Atomically read the contents of an 'StoredMVar'.  If the 'StoredMVar' is+--   currently empty, 'readMVar' will wait until its full.+--   'readMVar' is guaranteed to receive the next 'putMVar'.+--+--  'readMVar' is multiple-wakeup, so when multiple readers are+--    blocked on an 'StoredMVar', all of them are woken up at the same time.+--+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+{-# INLINE readMVar #-}+++-- | Atomically take a value from an 'StoredMVar', put a new value into the 'StoredMVar' and+--   return the value taken.+swapMVar :: Storable a => StoredMVar a -> a -> IO a+swapMVar (StoredMVar _ fp) x+  = 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+{-# INLINE swapMVar #-}+++-- | Put a value into an 'StoredMVar'.  If the 'StoredMVar' is currently full,+--   'putMVar' will wait until it becomes empty.+--+--+--   * 'putMVar' is single-wakeup.  That is, if there are multiple threads+--     or processes blocked in 'putMVar', and the 'StoredMVar' becomes empty,+--     only one thread will be woken up.+--+--   * The library makes no guarantees about the order in which processes+--     are woken up. This is all up to implementation-dependent OS scheduling.+--+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+{-# NOINLINE putMVar #-}++-- | A non-blocking version of 'takeMVar'.  The 'tryTakeMVar' function+--   returns immediately, with 'Nothing' if the 'StoredMVar' was empty, or+--   @'Just' a@ if the 'StoredMVar' was full with contents @a@.+--   After 'tryTakeMVar', the 'StoredMVar' is left empty.+tryTakeMVar :: Storable a => StoredMVar a -> IO (Maybe a)+tryTakeMVar (StoredMVar _ fp) = mask_ $ withForeignPtr fp $ \p -> alloca $ \lp -> do+    r <- c'mvar_trytake p lp+    if r == 0 then Just <$> peek lp+              else return Nothing+{-# INLINE tryTakeMVar #-}++-- | A non-blocking version of 'readMVar'.+--   The 'tryReadMVar' function+--   returns immediately, with 'Nothing' if the 'StoredMVar' was empty, or+--   @'Just' a@ if the 'StoredMVar' was full with contents @a@.+--+tryReadMVar :: Storable a => StoredMVar a -> IO (Maybe a)+tryReadMVar (StoredMVar _ fp) = mask_ $ withForeignPtr fp $ \p -> alloca $ \lp -> do+    r <- c'mvar_tryread p lp+    if r == 0 then Just <$> peek lp+              else return Nothing+{-# INLINE tryReadMVar #-}++-- | A non-blocking version of 'putMVar'.+--   The 'tryPutMVar' function+--   attempts to put the value @a@ into the 'StoredMVar', returning 'True' if+--   it was successful, or 'False' otherwise.+tryPutMVar  :: Storable a => StoredMVar a -> a -> IO Bool+tryPutMVar (StoredMVar _ fp) x = mask_ $ withForeignPtr fp $ \p -> alloca $ \lp -> do+    poke lp x+    r <- c'mvar_tryput p lp+    return $ r == 0+{-# INLINE tryPutMVar #-}++-- | A non-blocking version of 'swapMVar'.+--   Atomically attempt take a value from an 'StoredMVar', put a new value into the 'StoredMVar' and+--   return the value taken (thus, leave the `StoredMVar` full).+--   Return @Nothing@ if the `StoredMVar` was empty (and leave it empty).+trySwapMVar :: Storable a => StoredMVar a -> a -> IO (Maybe a)+trySwapMVar (StoredMVar _ fp) x+  = mask_ $ withForeignPtr fp $ \p -> allocaArray 2 $ \inp -> do+    let outp = advancePtr inp 1+    poke inp x+    r <- c'mvar_tryswap p inp outp+    if r == 0+    then Just <$> peek outp+    else return Nothing+{-# INLINE trySwapMVar #-}++checkNullPointer :: String -> IO (Ptr a) -> IO (Ptr a)+checkNullPointer s k = do+  p <- k+  if p == nullPtr+  then throwErrno ("StoredMVar." ++ s ++ ": FFI returned NULL pointer.")+  else return p+{-# INLINE checkNullPointer #-}+++foreign import ccall unsafe "mvar_new"+  c'mvar_new :: CSize -> IO (Ptr StoredMVarT)++foreign import ccall unsafe "mvar_lookup"+  c'mvar_lookup :: CString -> IO (Ptr StoredMVarT)++foreign import ccall unsafe "&mvar_destroy"+  p'mvar_destroy :: FunPtr (Ptr StoredMVarT -> IO ())++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+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++++-- | 'withMVar' is an exception-safe wrapper for operating on the contents+--   of an 'StoredMVar'.  This operation is exception-safe: it will replace the+--   original contents of the 'StoredMVar' if an exception is raised (see+--   "Control.Exception").  However, it is only atomic if there are no+--  other producers for this 'StoredMVar'.+withMVar :: Storable a => StoredMVar a -> (a -> IO b) -> IO b+withMVar m io = mask $ \restore -> do+    a <- takeMVar m+    b <- restore (io a) `onException` putMVar m a+    putMVar m a+    return b+{-# INLINE withMVar #-}+++-- | Like 'withMVar', but the @IO@ action in the second argument is executed+--   with asynchronous exceptions masked.+withMVarMasked :: Storable a => StoredMVar a -> (a -> IO b) -> IO b+withMVarMasked m io = mask_ $ do+    a <- takeMVar m+    b <- io a `onException` putMVar m a+    putMVar m a+    return b+{-# INLINE withMVarMasked #-}+++-- | An exception-safe wrapper for modifying the contents of an 'StoredMVar'.+--   Like 'withMVar', 'modifyMVar' will replace the original contents of+--   the 'StoredMVar' if an exception is raised during the operation.  This+--   function is only atomic if there are no other producers for this+--   'StoredMVar'.+modifyMVar_ :: Storable a => StoredMVar a -> (a -> IO a) -> IO ()+modifyMVar_ m io = mask $ \restore -> do+    a  <- takeMVar m+    a' <- restore (io a) `onException` putMVar m a+    putMVar m a'+{-# INLINE modifyMVar_ #-}+++-- | A slight variation on 'modifyMVar_' that allows a value to be+--   returned (@b@) in addition to the modified value of the 'StoredMVar'.+modifyMVar :: Storable a => StoredMVar a -> (a -> IO (a,b)) -> IO b+modifyMVar m io = mask $ \restore -> do+    a      <- takeMVar m+    (a',b) <- restore (io a >>= evaluate) `onException` putMVar m a+    putMVar m a'+    return b+{-# INLINE modifyMVar #-}+++-- | Like 'modifyMVar_', but the @IO@ action in the second argument is executed with+--   asynchronous exceptions masked.+modifyMVarMasked_ :: Storable a => StoredMVar a -> (a -> IO a) -> IO ()+modifyMVarMasked_ m io = mask_ $ do+    a  <- takeMVar m+    a' <- io a `onException` putMVar m a+    putMVar m a'+{-# INLINE modifyMVarMasked_ #-}+++-- | Like 'modifyMVar', but the @IO@ action in the second argument is executed with+--   asynchronous exceptions masked.+modifyMVarMasked :: Storable a => StoredMVar a -> (a -> IO (a,b)) -> IO b+modifyMVarMasked m io = mask_ $ do+    a      <- takeMVar m+    (a',b) <- (io a >>= evaluate) `onException` putMVar m a+    putMVar m a'+    return b+{-# INLINE modifyMVarMasked #-}

@@ -15,6 +15,7 @@ import           Foreign.Storable import           System.IO import           System.IO.Unsafe+import           Text.Read  #define HS_IMPORT_CONSTANTS_ONLY #include "SharedObjectName.h"@@ -28,8 +29,24 @@     showsPrec d (SOName a)       = showParen (d >= 10) $ showString "SOName " . showsPrec 10 getstr       where-        getstr = unsafePerformIO $ withForeignPtr a peekCString+        getstr = unsafePerformIO $ withForeignPtr a peekCAString         {-# NOINLINE getstr #-}++instance Read (SOName a) where+    readPrec = parens $ prec 10 $ do+        Ident "SOName" <- lexP+        s <- step readPrec+        return $ putstr s+      where+        writeStr [] n     ptr+          = pokeElemOff ptr n 0 -- put end of string character+        writeStr (c:cs) n ptr+          = pokeElemOff ptr n (castCharToCChar c) >> writeStr cs (n+1) ptr+        putstr s = unsafePerformIO $ do+          n <- newEmptySOName+          unsafeWithSOName n $ writeStr s 0+          return n+        {-# NOINLINE putstr #-}  instance Eq (SOName a) where     (SOName a) == (SOName b)

@@ -14,6 +14,17 @@ #include <string.h> #include <assert.h> ++void *vk_shared_malloc(void *pUserData, size_t size, size_t alignment, int32_t allocationScope) {+  return shared_malloc((SharedAllocator *)pUserData, size);+}+void *vk_shared_realloc(void *pUserData, void* pOriginal, size_t size, size_t alignment, int32_t allocationScope) {+  return shared_realloc((SharedAllocator *)pUserData, pOriginal, size);+}+void  vk_shared_free(void *pUserData, void* pMemory) {+  return shared_free((SharedAllocator *)pUserData, pMemory);+}+ /* Default store size is at least the page size, which usualy is equal to 4KB.  */ #define DEFAULT_STORE_SIZE_FACTOR 12

@@ -3,7 +3,7 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE RoleAnnotations            #-} module Foreign.SharedPtr.C-  ( SharedPtr (), Allocator+  ( SharedPtr (), Allocator, AllocatorT    , c'shared_createAllocator, c'shared_lookupAllocator   , c'shared_destroyAllocator, c'shared_getStoreName@@ -14,9 +14,12 @@   , p'shared_destroyAllocator, p'shared_getStoreName   , p'shared_ptrToShPtr, p'shared_shPtrToPtr   , p'shared_malloc, p'shared_realloc, p'shared_free++  , p'vk_shared_malloc, p'vk_shared_realloc, p'vk_shared_free   ) where  import           Data.Data        (Data)+import           Data.Void        (Void) import           Foreign.C.String import           Foreign.C.Types import           Foreign.Ptr@@ -28,8 +31,9 @@ newtype SharedPtr a = SharedPtr ( Ptr a )   deriving (Eq, Ord, Show, Data, Generic, Storable) +-- | @C@ structure, should not be inspected from Haskell code data AllocatorT--- | Opaque pointer to the allocator type defined in C code.+-- | Opaque pointer to the allocator type defined in @C@ code. type Allocator = Ptr AllocatorT  @@ -105,3 +109,19 @@ foreign import ccall unsafe "shared_free"   c'shared_free     :: Allocator -> Ptr a -> IO ()++++foreign import ccall unsafe "&vk_shared_malloc"+  p'vk_shared_malloc+    :: FunPtr+        (Ptr Void -> CSize -> CSize -> allocScope -> IO (Ptr Void))++foreign import ccall unsafe "&vk_shared_realloc"+  p'vk_shared_realloc+    :: FunPtr+        (Ptr Void -> Ptr Void -> CSize -> CSize -> allocScope -> IO (Ptr Void))++foreign import ccall unsafe "&vk_shared_free"+  p'vk_shared_free+    :: FunPtr (Ptr Void -> Ptr Void -> IO ())

@@ -45,7 +45,7 @@   }   HsPtr r = mmap( NULL                 , size-                , PROT_READ | PROT_WRITE+                , PROT_READ | PROT_WRITE | PROT_EXEC                 , MAP_SHARED                 , memFd, 0);   if (r == MAP_FAILED) {

@@ -1,7 +1,6 @@ #include "SharedPtr.h"-#ifdef STDOUT_SYSCALL_DEBUG+#ifndef NDEBUG #include <stdio.h>-#include <tchar.h> #endif  void _SharedMutex_init(SharedMutex *mptr, void **privateMutexHandle, const int createNew) {@@ -12,9 +11,9 @@     ( NULL    // default security attributes     , FALSE   // initially not owned     , mptr->mutexName );-#ifdef STDOUT_SYSCALL_DEBUG+#ifndef NDEBUG   if (*privateMutexHandle == NULL) {-    _tprintf(TEXT("CreateMutex error: %d\n"), GetLastError());+    printf("CreateMutex error: %d\n", GetLastError());   } #endif }@@ -26,8 +25,8 @@ 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());+#ifndef NDEBUG+    printf("WaitForSingleObject mutex error: return %d; error code %d.\n", r, GetLastError()); #endif     return 1;   } else {@@ -38,8 +37,8 @@ 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());+#ifndef NDEBUG+    printf("ReleaseMutex error: %d\n", GetLastError()); #endif 	  return 1;   } else {@@ -58,8 +57,8 @@     , memBlockName);             // name of mapping object    if (*privateMutexHandle == NULL) {-#ifdef STDOUT_SYSCALL_DEBUG-    _tprintf(TEXT("Could not create file mapping object (%d).\n"), GetLastError());+#ifndef NDEBUG+    printf("Could not create file mapping object (%d).\n", GetLastError()); #endif     return NULL;   }@@ -71,8 +70,8 @@     , size);    if (rptr == NULL) {-#ifdef STDOUT_SYSCALL_DEBUG-	  _tprintf(TEXT("Could not map view of file (%d).\n"), GetLastError());+#ifndef NDEBUG+	  printf("Could not map view of file (%d).\n", GetLastError()); #endif     CloseHandle(*privateMutexHandle);     return NULL;

@@ -0,0 +1,168 @@+module Main (main) where++import           Control.Concurrent.Process.StoredMVar+import qualified Control.Concurrent.MVar as Vanilla+import           Control.Concurrent (forkIO)+import Control.Monad (void, forM)+-- import           Control.Exception                     (SomeException, catch,+--                                                         displayException)+import           Foreign.SharedObjectName+-- import           GHC.Environment                       (getFullArgs)+import           System.Environment+import           System.Exit+import Data.Monoid+import Text.Read (readMaybe)+import           System.Process.Typed+import System.IO+import Control.Exception+import System.IO.Unsafe+import System.Mem++-- | 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++data BasicRole = Master | Slave+  deriving (Eq, Ord, Show, Read)++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 Success++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` ())+      )+  ]++++data TestResult+  = Success+  | Failure String+  deriving (Eq, Ord, Show, Read)++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 ("  " <>) . 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+  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)++        rx <- unsafeInterleaveIO $ Vanilla.takeMVar mr+        rxs <- go xs+        return $ rx <> rxs+      where+        withI s = "[" ++ show i ++ "] " ++ s