diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2015, Sebastiaan la Fleur + +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 Sebastiaan la Fleur 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/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/concurrent-utilities.cabal b/concurrent-utilities.cabal new file mode 100644 --- /dev/null +++ b/concurrent-utilities.cabal @@ -0,0 +1,52 @@ +name: concurrent-utilities + +-- The package version. See the Haskell package versioning policy (PVP) +-- for standards guiding when and how versions should be incremented. +-- http://www.haskell.org/haskellwiki/Package_versioning_policy +-- PVP summary: +-+------- breaking API changes +-- | | +----- non-breaking API additions +-- | | | +--- code changes with no API change +version: 0.1.0.0 +synopsis: More utilities and broad-used datastructures for concurrency. +description: More utilities and broad-used datastructures for concurrency. +homepage: - +license: BSD3 +license-file: LICENSE +author: Sebastiaan la Fleur +maintainer: sebastiaan.la.fleur@gmail.com +copyright: University of Twente 2015 | Sebastiaan la Fleur 2015 +category: Concurrency +build-type: Simple + +-- Extra files to be distributed with the package, such as examples or a +-- README. +-- extra-source-files: + +-- Constraint on the version of Cabal needed to build this package. +cabal-version: >=1.10 + + +library + -- Modules exported by the library. + exposed-modules: Control.Concurrent.ExceptionCollection, + Control.Concurrent.ExceptionUtility, + Control.Concurrent.SafePrint, + Control.Concurrent.Thread, + Control.Concurrent.Datastructures.ThreadWaitQueue, + Control.Concurrent.Datastructures.BlockingConcurrentQueue + + -- Modules included in this library but not exported. + -- other-modules: + + -- LANGUAGE extensions used by modules in this package. + -- other-extensions: + + -- Other library packages from which modules are imported. + build-depends: base >=4.8 && <4.9 + + -- Directories containing source files. + hs-source-dirs: src + + -- Base language which the package is written in. + default-language: Haskell2010 + diff --git a/src/Control/Concurrent/Datastructures/BlockingConcurrentQueue.hs b/src/Control/Concurrent/Datastructures/BlockingConcurrentQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Datastructures/BlockingConcurrentQueue.hs @@ -0,0 +1,69 @@ +module Control.Concurrent.Datastructures.BlockingConcurrentQueue where + +import Control.Concurrent +import Control.Concurrent.MVar +import Control.Concurrent.Datastructures.ThreadWaitQueue + + +data BlockingConcurrentQueue a = BlockingConcurrentQueue { queue :: MVar [a] + , threadWaitQueue :: ThreadWaitQueue -- Used to signal when to try again if the queue was empty and we wanted to take one + } + + +showBlockingConcurrentQueue :: (Show a) => BlockingConcurrentQueue a -> IO [Char] +showBlockingConcurrentQueue blockingConcurrentQueue + = do + els <- readAllFromBlockingConcurrentQueue blockingConcurrentQueue + return (show els) + + +createBlockingConcurrentQueue :: IO (BlockingConcurrentQueue a) +createBlockingConcurrentQueue + = do + newQueue <- newMVar [] + threadWaitQueue <- createWaitQueue + return (BlockingConcurrentQueue newQueue threadWaitQueue) + + +putInBlockingConcurrentQueue :: BlockingConcurrentQueue a -> a -> IO () +putInBlockingConcurrentQueue blockingConcurrentQueue el + = do + oldQueue <- takeMVar (queue blockingConcurrentQueue) + putMVar (queue blockingConcurrentQueue) (oldQueue ++ [el]) + openAndRecloseWaitQueue (threadWaitQueue blockingConcurrentQueue) + + +putAllInBlockingConcurrentQueue :: BlockingConcurrentQueue a -> [a] -> IO () +putAllInBlockingConcurrentQueue blockingConcurrentQueue els + = do + oldQueue <- takeMVar (queue blockingConcurrentQueue) + putMVar (queue blockingConcurrentQueue) (oldQueue ++ els) + openAndRecloseWaitQueue (threadWaitQueue blockingConcurrentQueue) + + +takeFromBlockingConcurrentQueue :: BlockingConcurrentQueue a -> IO a +takeFromBlockingConcurrentQueue blockingConcurrentQueue + = do + oldQueue <- takeMVar (queue blockingConcurrentQueue) + case oldQueue of + [] -> do + queueTicket <- getQueueTicket (threadWaitQueue blockingConcurrentQueue) + putMVar (queue blockingConcurrentQueue) [] + enterWaitQueueWithTicket queueTicket + takeFromBlockingConcurrentQueue blockingConcurrentQueue + (el:els) -> do + putMVar (queue blockingConcurrentQueue) els + return el + + +takeAllFromBlockingConcurrentQueue :: BlockingConcurrentQueue a -> IO [a] +takeAllFromBlockingConcurrentQueue blockingConcurrentQueue + = do + els <- takeMVar (queue blockingConcurrentQueue) + putMVar (queue blockingConcurrentQueue) [] + return els + + +readAllFromBlockingConcurrentQueue :: BlockingConcurrentQueue a -> IO [a] +readAllFromBlockingConcurrentQueue blockingConcurrentQueue + = readMVar (queue blockingConcurrentQueue) diff --git a/src/Control/Concurrent/Datastructures/ThreadWaitQueue.hs b/src/Control/Concurrent/Datastructures/ThreadWaitQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Datastructures/ThreadWaitQueue.hs @@ -0,0 +1,91 @@ +module Control.Concurrent.Datastructures.ThreadWaitQueue where + +import Control.Concurrent +import Control.Concurrent.MVar + +type WaitOnToken = MVar () +type ThreadWaitQueue = MVar WaitOnToken + + +createWaitQueue :: IO ThreadWaitQueue +createWaitQueue + = do + waitOnToken <- newEmptyMVar + threadWaitQueue <- newMVar waitOnToken + return threadWaitQueue + + +getQueueTicket :: ThreadWaitQueue -> IO WaitOnToken +getQueueTicket threadWaitQueue + = readMVar threadWaitQueue + + +enterWaitQueueWithTicket :: WaitOnToken -> IO () +enterWaitQueueWithTicket waitOnToken + = do + yield + {- Do not remove this yield. This yield is the product of many many tears + and a newfound disgust of the concurrent forkIO Haskell system. + Now more serious: + forkIO creates lightweight threads that are NOT interrupted when they are + performing a computation. The switch happens somewhere else (probably after a + task has been completed). So a very deep recursion tree is not interrupted until it + is solved. + So picture the following case: the queue is open and a lightweight thread tried + to perform something that the queue was necessary for and finds that the condition isn't + set yet and returns to the queue. But the queue is open... so we start all over again + and again and again... and because the computation is not "complete" the thread is never + switched out and thus we have a blocking thread that causes the entire system to halt. + This yield interrupts that process and gives other threads actually the chance to set the + condition the queue is needed for. + -} + readMVar waitOnToken + + +enterWaitQueue :: ThreadWaitQueue -> IO () +enterWaitQueue threadWaitQueue + = getQueueTicket threadWaitQueue >>= enterWaitQueueWithTicket + + +{- | Open the queue. If the queue is already opened, nothing happens. +-} +openWaitQueue :: ThreadWaitQueue -> IO () +openWaitQueue threadWaitQueue + = do + waitOnToken <- takeMVar threadWaitQueue + _openIfNeeded waitOnToken + putMVar threadWaitQueue waitOnToken + + +{- Not thread safe! Should never be called if we do not have the rights +for the ThreadWaitQueue +-} +_openIfNeeded :: WaitOnToken -> IO () +_openIfNeeded waitOnToken + = do + notOpenYet <- isEmptyMVar waitOnToken + case notOpenYet of + True -> putMVar waitOnToken () + False -> return () + + +recloseWaitQueue :: ThreadWaitQueue -> IO () +recloseWaitQueue threadWaitQueue + = do + waitOnToken' <- newEmptyMVar + _ <- takeMVar threadWaitQueue + putMVar threadWaitQueue waitOnToken' + + +{- | Atomic +-} +openAndRecloseWaitQueue :: ThreadWaitQueue -> IO () +openAndRecloseWaitQueue threadWaitQueue + = do + waitOnToken1 <- takeMVar threadWaitQueue + waitOnToken2 <- newEmptyMVar + -- First open the old queue if needed + _openIfNeeded waitOnToken1 + -- Now set new close queue + putMVar threadWaitQueue waitOnToken2 + diff --git a/src/Control/Concurrent/ExceptionCollection.hs b/src/Control/Concurrent/ExceptionCollection.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/ExceptionCollection.hs @@ -0,0 +1,42 @@ +module Control.Concurrent.ExceptionCollection + ( ExceptionCollection(..) + , createExceptionCollection + , logException + , readExceptions + , collectExceptions + ) where + +import Control.Concurrent.MVar + +data ExceptionCollection e = ExceptionCollection (MVar [e]) + + +createExceptionCollection :: IO (ExceptionCollection e) +createExceptionCollection + = do + exceptionsM <- newMVar [] + return (ExceptionCollection exceptionsM) + + +logException :: ExceptionCollection e + -> e + -> IO () +logException (ExceptionCollection exceptionsM) e + = do + raisedExceptions <- takeMVar exceptionsM + putMVar exceptionsM (raisedExceptions ++ [e]) + + +readExceptions :: ExceptionCollection e + -> IO [e] +readExceptions (ExceptionCollection exceptionsM) + = readMVar exceptionsM + + +collectExceptions :: ExceptionCollection e + -> IO [e] +collectExceptions (ExceptionCollection exceptionsM) + = do + raisedExceptions <- takeMVar exceptionsM + putMVar exceptionsM [] + return raisedExceptions diff --git a/src/Control/Concurrent/ExceptionUtility.hs b/src/Control/Concurrent/ExceptionUtility.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/ExceptionUtility.hs @@ -0,0 +1,55 @@ +module Control.Concurrent.ExceptionUtility where + +import Control.Exception + +{- | Utility function that uses an uninterruptable mask to make sure a resource +is always acquired and released even with asynchronous exceptions. You can define +the acquire, release and update actions. The acquire function should deliver the resource. +The update function receives the acquired value and tries to update it but may run into (asynchronous) exceptions. +And lastly the release function receives the updated value (or the old value in case of an exception) and can release +the resource with it. + +This function is primarily used in situations with concurrent data where some value is always required to reside +in the concurrent datastructure (ex. MVars). + +The acquire and release reside under the uninterruptable mask, the update function does not and may be interrupted +or may cause exceptions. + +-} +uninterruptibleUpdateResource + :: IO a -- ^ Acquire resource + -> (a -> IO b) -- ^ Release resource with either updated or old value + -> (a -> IO a) -- ^ How to update the value + -> IO a -- ^ Updated value +uninterruptibleUpdateResource acquire release update + = uninterruptibleMask $ \restore -> do + aOld <- acquire + catch ( do + aNew <- restore (update aOld) + _ <- release aNew + return aNew + ) + ( \exception -> do + _ <- release aOld + throwIO (exception :: SomeException) + ) +{- | A version of uninterruptibleUpdateResource where you can also calculate a new value + +-} +uninterruptibleUpdateResourceAndCalculate + :: IO a -- ^ Acquire resource + -> (a -> IO b) -- ^ Release resource with either updated or old value + -> (a -> IO (a, c)) -- ^ How to update the value and calculate the result + -> IO (a, c) -- ^ Updated value +uninterruptibleUpdateResourceAndCalculate acquire release update + = uninterruptibleMask $ \restore -> do + aOld <- acquire + catch ( do + (aNew, c) <- restore (update aOld) + _ <- release aNew + return (aNew, c) + ) + ( \exception -> do + _ <- release aOld + throwIO (exception :: SomeException) + ) diff --git a/src/Control/Concurrent/SafePrint.hs b/src/Control/Concurrent/SafePrint.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/SafePrint.hs @@ -0,0 +1,24 @@ +module Control.Concurrent.SafePrint where + +import Control.Concurrent.MVar + +type SafePrintToken = MVar () + +createSafePrintToken :: IO SafePrintToken +createSafePrintToken = newMVar () + +safePrint :: SafePrintToken -> String -> IO () +safePrint safePrintToken message + = do + takeMVar safePrintToken + putStr message + putMVar safePrintToken () + + +safePrintLn :: SafePrintToken -> String -> IO () +safePrintLn safePrintToken message + = do + takeMVar safePrintToken + putStrLn message + putMVar safePrintToken () + diff --git a/src/Control/Concurrent/Thread.hs b/src/Control/Concurrent/Thread.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Thread.hs @@ -0,0 +1,70 @@ +module Control.Concurrent.Thread + ( Thread + , forkThread + , isTerminated + , joinThread + , joinThreads + , getThreadId + , terminateThread + ) where + +import Control.Exception +import Control.Concurrent +import Control.Concurrent.MVar + +type WaitOn = MVar () +data Thread = Thread ThreadId WaitOn + +instance Show Thread where + show (Thread threadId _) = show threadId + + +forkThread :: IO () -> IO Thread +forkThread action + = do + waitOn <- newEmptyMVar + tid <- forkFinally action (_setTerminated waitOn) + return (Thread tid waitOn) + + +_setTerminated :: WaitOn -> (Either SomeException a) -> IO () +_setTerminated waitOn (Right _) + = putMVar waitOn () + +_setTerminated waitOn (Left someException) + = do + putMVar waitOn () + throw someException + +isTerminated :: Thread + -> IO Bool +isTerminated (Thread tid waitOn) + = do + isTerm <- isEmptyMVar waitOn + return (not isTerm) + + +joinThread :: Thread -> IO () +joinThread t@(Thread threadId waitOn) + = do + isTerm <- isTerminated t + case isTerm of + True -> return () + False -> do + _ <- takeMVar waitOn + return () + + +joinThreads :: [Thread] -> IO () +joinThreads threads = foldr ((>>).joinThread) (return ()) threads + + +getThreadId :: Thread -> ThreadId +getThreadId (Thread tid _) + = tid + + +terminateThread :: Thread -> IO () +terminateThread (Thread tid _) + = do + killThread tid