concurrent-utilities (empty) → 0.1.0.0

raw patch · 9 files changed

+435/−0 lines, 9 filesdep +basesetup-changed

Dependencies added: base

Files

• LICENSE +30/−0
• Setup.hs +2/−0
• concurrent-utilities.cabal +52/−0
• src/Control/Concurrent/Datastructures/BlockingConcurrentQueue.hs +69/−0
• src/Control/Concurrent/Datastructures/ThreadWaitQueue.hs +91/−0
• src/Control/Concurrent/ExceptionCollection.hs +42/−0
• src/Control/Concurrent/ExceptionUtility.hs +55/−0
• src/Control/Concurrent/SafePrint.hs +24/−0
• src/Control/Concurrent/Thread.hs +70/−0

@@ -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.

@@ -0,0 +1,2 @@+import Distribution.Simple
+main = defaultMain

@@ -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
+  

@@ -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)

@@ -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
+

@@ -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

@@ -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)
+                                                  )

@@ -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 ()
+       

@@ -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