diff --git a/Changes.md b/Changes.md new file mode 100644 --- /dev/null +++ b/Changes.md @@ -0,0 +1,3 @@ +* 0.0.1 + - Initial release and cleanup from ambiata/twine + - Rename to entwine from twine, due to existing project called twine. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,28 @@ +Copyright 2017, Ambiata, All Rights Reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + 1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + 2. 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. + + 3. Neither the name of the copyright holder nor the names of + its 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 +HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,4 @@ +Twine +===== + +Concurrency tools. 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/entwine.cabal b/entwine.cabal new file mode 100644 --- /dev/null +++ b/entwine.cabal @@ -0,0 +1,126 @@ +name: entwine +version: 0.0.1 +license: BSD3 +license-file: LICENSE +author: Ambiata +maintainer: Tim McGilchrist +homepage: https://github.com/tmcgilchrist/entwine +bug-reports: https://github.com/tmcgilchrist/entwine/issues +copyright: (c) 2015 Ambiata. +synopsis: entwine - Concurrency tools +category: System +cabal-version: >= 1.8 +build-type: Simple +description: entwine - Concurrency tools + +tested-with: GHC == 8.0.2, GHC == 8.2.2 +extra-source-files: + README.md + Changes.md + +source-repository head + type: git + location: https://github.com/tmcgilchrist/entwine.git + +library + build-depends: + base >= 3 && < 5 + , async >= 2.0 && < 2.2 + , containers >= 0.5 && < 0.7 + , exceptions >= 0.6 && < 0.9 + , monad-loops == 0.4.* + , SafeSemaphore == 0.10.* + , stm == 2.4.* + , text == 1.2.* + , time >= 1.4 && < 1.9 + , transformers >= 0.3 && < 0.6 + , transformers-either == 0.1.* + + ghc-options: + -Wall + + hs-source-dirs: + src + + + exposed-modules: + Entwine + Entwine.Async + Entwine.Data + Entwine.Data.Duration + Entwine.Data.Finalizer + Entwine.Data.Gate + Entwine.Data.Parallel + Entwine.Data.Pin + Entwine.Data.Queue + Entwine.Loop + Entwine.Parallel + Entwine.P + Entwine.Snooze + Entwine.Guard + +test-suite test + type: exitcode-stdio-1.0 + + main-is: test.hs + + ghc-options: -Wall + + hs-source-dirs: + test + + build-depends: + base >= 3 && < 5 + , entwine + , async >= 2.0 && < 2.3 + , directory >= 1.2 && < 1.4 + , exceptions >= 0.6 && < 0.9 + , process >= 1.2 && < 1.7 + , text == 1.2.* + , time + , transformers >= 0.3 && < 0.6 + , transformers-either == 0.1.* + , QuickCheck == 2.10.* + , quickcheck-instances == 0.3.* + +test-suite test-io + type: exitcode-stdio-1.0 + + main-is: test-io.hs + + ghc-options: -Wall + + hs-source-dirs: + test + + build-depends: + base >= 3 && < 5 + , entwine + , async >= 2.0 && < 2.3 + , directory >= 1.2 && < 1.4 + , exceptions + , process >= 1.2 && < 1.7 + , text + , time + , transformers-either + , QuickCheck == 2.10.* + , quickcheck-instances == 0.3.* + +benchmark bench + type: exitcode-stdio-1.0 + main-is: bench.hs + ghc-options: -Wall -threaded -O2 + hs-source-dirs: test + build-depends: base + , entwine + , criterion >= 1.2 && < 1.6 + , directory >= 1.2 && < 1.4 + , process >= 1.2 && < 1.7 + , QuickCheck + , quickcheck-instances == 0.3.* + , quickcheck-properties == 0.1.* + , quickcheck-text == 0.1.* + , random + , text + , transformers + , transformers-either diff --git a/src/Entwine.hs b/src/Entwine.hs new file mode 100644 --- /dev/null +++ b/src/Entwine.hs @@ -0,0 +1,8 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine ( + module Entwine.Data + , module Entwine.Snooze + ) where + +import Entwine.Data +import Entwine.Snooze diff --git a/src/Entwine/Async.hs b/src/Entwine/Async.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Async.hs @@ -0,0 +1,76 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Entwine.Async ( + AsyncTimeout (..) + , renderAsyncTimeout + , waitWithTimeout + , waitEitherBoth + ) where + + +import Control.Concurrent.Async (Async, waitSTM, waitEither) +import Control.Concurrent.Async (async, cancel, wait) +import Control.Concurrent.STM (atomically, orElse, retry) +import Control.Exception.Base (AsyncException (..)) +import Control.Monad.Catch (catch, throwM) +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Trans.Either + +import Data.IORef (newIORef, readIORef, writeIORef) +import qualified Data.Text as T + +import Entwine.P +import Entwine.Snooze + +import System.IO (IO) + +data AsyncTimeout = + AsyncTimeout Duration + deriving (Eq, Show) + +renderAsyncTimeout :: AsyncTimeout -> Text +renderAsyncTimeout e = + case e of + AsyncTimeout d -> + mconcat [ + "Async took greater than '" + , T.pack . show $ toSeconds d + , " seconds' to return." + ] + +waitWithTimeout :: Async a -> Duration -> EitherT AsyncTimeout IO a +waitWithTimeout a d = do + r <- liftIO $ newIORef False + s <- liftIO . async $ snooze d + e <- liftIO $ waitEither a s + case e of + Left a' -> + pure $ a' + Right _ -> do + liftIO $ writeIORef r True + liftIO $ cancel a + (liftIO $ wait a) + `catch` + (\(ae :: AsyncException) -> + case ae of + ThreadKilled -> do + liftIO (readIORef r) >>= + bool (liftIO $ throwM ae) (left $ AsyncTimeout d) + StackOverflow -> + liftIO $ throwM ae + HeapOverflow -> + liftIO $ throwM ae + UserInterrupt -> + liftIO $ throwM ae) + +waitEitherBoth :: Async a -> Async b -> Async c -> IO (Either a (b, c)) +waitEitherBoth a b c = + atomically $ do + let + l = waitSTM a + r = do + bb <- waitSTM b `orElse` (waitSTM c >> retry) + cc <- waitSTM c + return (bb, cc) + fmap Left l `orElse` fmap Right r diff --git a/src/Entwine/Data.hs b/src/Entwine/Data.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Data.hs @@ -0,0 +1,9 @@ +module Entwine.Data ( + module X + ) where + +import Entwine.Data.Duration as X +import Entwine.Data.Finalizer as X +import Entwine.Data.Parallel as X +import Entwine.Data.Pin as X +import Entwine.Data.Queue as X diff --git a/src/Entwine/Data/Duration.hs b/src/Entwine/Data/Duration.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Data/Duration.hs @@ -0,0 +1,67 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine.Data.Duration ( + Duration + , microseconds + , milliseconds + , seconds + , minutes + , hours + , toMicroseconds + , toMilliseconds + , toHours + , toMinutes + , toSeconds + ) where + +import Entwine.P + +-- | +-- A Duration is an abstract type, representing a short delay (in the +-- region of micro-seconds to a few minutes). +-- +-- This useful for implementing concurrency and process primitives +-- that need to wait for short periods to ensure fairness (for example). +-- +newtype Duration = + Duration { + duration :: Int + } deriving (Eq, Show) + +microseconds :: Int -> Duration +microseconds = + Duration + +milliseconds :: Int -> Duration +milliseconds = + microseconds . (*) 1000 + +seconds :: Int -> Duration +seconds = + milliseconds . (*) 1000 + +minutes :: Int -> Duration +minutes = + seconds . (*) 60 + +hours :: Int -> Duration +hours = minutes . (*) 60 + +toMicroseconds :: Duration -> Int +toMicroseconds = + duration + +toMilliseconds :: Duration -> Int +toMilliseconds = + flip div 1000 . toMicroseconds + +toSeconds :: Duration -> Int +toSeconds = + flip div 1000 . toMilliseconds + +toMinutes :: Duration -> Int +toMinutes = + flip div 60 . toSeconds + +toHours :: Duration -> Int +toHours = + flip div 60 . toMinutes diff --git a/src/Entwine/Data/Finalizer.hs b/src/Entwine/Data/Finalizer.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Data/Finalizer.hs @@ -0,0 +1,31 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine.Data.Finalizer ( + Finalizer (..) + ) where + +import Entwine.P + +import System.IO + + +-- | +-- A callback which can be invoked to cleanup a resource +-- +newtype Finalizer = + Finalizer { + finalize :: IO () + } + +instance Semigroup Finalizer where + {-# INLINE (<>) #-} + (<>) (Finalizer a) (Finalizer b) = + Finalizer $ a >> b + +instance Monoid Finalizer where + {-# INLINE mempty #-} + mempty = + Finalizer $ pure () + + {-# INLINE mappend #-} + mappend (Finalizer a) (Finalizer b) = + Finalizer $ a >> b diff --git a/src/Entwine/Data/Gate.hs b/src/Entwine/Data/Gate.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Data/Gate.hs @@ -0,0 +1,38 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine.Data.Gate ( + Gate + , newGate + , isOpen + , close + ) where + +import Data.IORef (IORef, newIORef, readIORef, atomicWriteIORef) + +import Entwine.P + +import System.IO + +-- | +-- A gate is an abstract type, representing a simple barrier +-- that can only have its state queried in a non-blocking +-- manner. +-- +-- This useful for implementing things like passing in a gate +-- to terminate or stop a loop. +-- +newtype Gate = + Gate { + gate :: IORef Bool + } + +newGate :: IO Gate +newGate = + Gate <$> newIORef True + +isOpen :: Gate -> IO Bool +isOpen = + readIORef . gate + +close :: Gate -> IO () +close = + flip atomicWriteIORef False . gate diff --git a/src/Entwine/Data/Parallel.hs b/src/Entwine/Data/Parallel.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Data/Parallel.hs @@ -0,0 +1,70 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine.Data.Parallel ( + Workers + , Result + , newResult + , emptyResult + , addResult + , getResult + , emptyWorkers + , failWorkers + , getWorkers + , addWorker + , waitForWorkers + , waitForWorkers' + ) where + +import Control.Concurrent.Async (Async, cancel, wait) +import Control.Concurrent.MVar + +import Entwine.P + +import System.IO + +newtype Result a = + Result (MVar a) + +newResult :: a -> IO (Result a) +newResult a = + Result <$> newMVar a + +-- fix pair with takeMVar +emptyResult :: IO (Result a) +emptyResult = + Result <$> newEmptyMVar + +addResult :: Monad m => Result (m a) -> m a -> IO (m a) +addResult (Result r) w = + modifyMVar r (\x -> pure $ ((x >> w), x)) + +getResult :: Result a -> IO a +getResult (Result r) = + readMVar r + +newtype Workers a = + Workers (MVar [Async a]) + +emptyWorkers :: IO (Workers a) +emptyWorkers = + Workers <$> newMVar [] + +failWorkers :: Workers a -> IO () +failWorkers (Workers w) = + readMVar w >>= + mapM_ cancel + +getWorkers :: Workers a -> IO [Async a] +getWorkers (Workers w) = + readMVar w + +addWorker :: (Workers a) -> Async a -> IO () +addWorker (Workers w) r = + modifyMVar_ w $ pure . (:) r + +waitForWorkers :: (Workers a) -> IO () +waitForWorkers (Workers w) = + readMVar w >>= mapM_ wait + +waitForWorkers' :: (Workers a) -> IO [a] +waitForWorkers' (Workers w) = + readMVar w >>= mapM wait diff --git a/src/Entwine/Data/Pin.hs b/src/Entwine/Data/Pin.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Data/Pin.hs @@ -0,0 +1,41 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine.Data.Pin ( + Pin + , newPin + , checkPin + , waitForPin + , pullPin + ) where + +import Control.Concurrent.MVar + +import Entwine.P + +import System.IO + +-- | +-- A pin is an abstract type, representing a simple barrier +-- that can only have its state queried in a non-blocking +-- manner. +-- +-- This useful for implementing things like passing in a hook +-- to terminate or stop waiting for a process. +-- +newtype Pin = + Pin { pin :: MVar () } + +newPin :: IO Pin +newPin = + Pin <$> newEmptyMVar + +checkPin :: Pin -> IO Bool +checkPin = + fmap isJust . tryTakeMVar . pin + +waitForPin :: Pin -> IO () +waitForPin = + void . readMVar . pin + +pullPin :: Pin -> IO () +pullPin = + void . flip tryPutMVar () . pin diff --git a/src/Entwine/Data/Queue.hs b/src/Entwine/Data/Queue.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Data/Queue.hs @@ -0,0 +1,43 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine.Data.Queue ( + Queue + , newQueue + , readQueue + , tryReadQueue + , writeQueue + , isQueueEmpty + ) where + +import Control.Concurrent.STM.TBQueue ( TBQueue, newTBQueue, tryReadTBQueue + , readTBQueue, writeTBQueue, isEmptyTBQueue) + +import GHC.Conc (atomically) + +import Entwine.P + +import System.IO + +newtype Queue a = + Queue { + queue :: TBQueue a + } + +newQueue :: Int -> IO (Queue a) +newQueue i = + atomically $ Queue <$> newTBQueue i + +readQueue :: Queue a -> IO a +readQueue = + atomically . readTBQueue . queue + +tryReadQueue :: Queue a -> IO (Maybe a) +tryReadQueue = + atomically . tryReadTBQueue . queue + +writeQueue :: Queue a -> a -> IO () +writeQueue q = + atomically . writeTBQueue (queue q) + +isQueueEmpty :: Queue a -> IO Bool +isQueueEmpty = + atomically . isEmptyTBQueue . queue diff --git a/src/Entwine/Guard.hs b/src/Entwine/Guard.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Guard.hs @@ -0,0 +1,81 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine.Guard ( + TerminationAction (..) + , TerminationHandler (..) + , guarded + , repeatedly + ) where + +import Control.Monad.Catch +import Control.Monad.Trans.Class +import Control.Monad.Trans.Either + +import System.IO + +import Entwine.P +import Entwine.Snooze + +data TerminationAction = + Restart + | Die + deriving (Eq, Show) + +data TerminationHandler e = + TerminationHandler { + onExplosion :: SomeException -> IO TerminationAction + , onError :: e -> IO TerminationAction + , onGraceful :: IO TerminationAction + } + +-- | Run an action forever, using termination handler to notify of +-- failure events. +-- +-- This is generally most useful to guard a thread against +-- un-noticed termination. +-- +-- A reasonable usage would be to add monitoring / notifications +-- in termination handler and just let this loop keep your code +-- alive. It is recommened that you use the termination handler +-- to control number of retried. +-- +-- An equally reasonable alternative is to call exitImmediately +-- (or some equivalent - perhaps you have an MVar controlling +-- program termination?), and use this to ensure that the entire +-- process dies if any of the supervised threads die. +-- +-- The action passed to 'guard' should run forever, if you +-- want to run a short-lived action repeatedly in a supervised +-- fashion see 'repeatedly'. This function will still work, but +-- the 'onGraceful' handler will be called (a lot). +-- +-- Common usage (where expectation is the do block never return): +-- @ +-- void . forkIO . guard (TerminationHandler ...) . forever $ do +-- doThis +-- andThis +-- @ +-- +-- +guarded :: TerminationHandler e -> EitherT e IO () -> IO () +guarded handler action = + let run = runEitherT action >>= either (onError handler) (const . onGraceful $ handler) + safe = run `catchAll` onExplosion handler `catchAll` (const . pure) Restart + in safe >>= \a -> when (a == Restart) $ guarded handler action + +-- | Run an action repeatedly with a fixed delay between runs, using +-- termination handler to notify of failure events. +-- +-- See 'guard' for more description. This offers the same +-- behaviour for actions that don't run forever. +-- +-- Common usage (where expectation is the do block does returns): +-- @ +-- void . forkIO . repeatedly (seconds 5) (TerminationHandler ...) $ do +-- doThis +-- andThis +-- @ +-- +repeatedly :: Duration -> TerminationHandler e -> EitherT e IO () -> IO () +repeatedly d handler action = + let ever = action >> lift (snooze d) >> ever + in guarded handler ever diff --git a/src/Entwine/Loop.hs b/src/Entwine/Loop.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Loop.hs @@ -0,0 +1,24 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE LambdaCase #-} +module Entwine.Loop ( + loop + ) where + +import Control.Monad.IO.Class (MonadIO, liftIO) + +import Entwine.Data.Gate +import Entwine.Data.Duration +import Entwine.P +import Entwine.Snooze + +-- | Loop with a delay until the gate is closed +-- +loop :: MonadIO m => Duration -> Gate -> m () -> m () +loop d c action = do + action + liftIO (isOpen c) >>= \case + False -> + return () + True -> do + liftIO $ snooze d + loop d c action diff --git a/src/Entwine/P.hs b/src/Entwine/P.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/P.hs @@ -0,0 +1,469 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE LambdaCase #-} +module Entwine.P ( + -- * Primitive types + -- ** Bool + Bool (..) + , bool + , (&&) + , (||) + , not + , otherwise + -- ** Char + , Char + -- ** Int + , Integer + , Int + , Int8 + , Int16 + , Int32 + , Int64 + , div + , toInteger + -- ** Word + , Word64 + -- ** Real + , fromIntegral + , fromRational + + , Double + + -- * Algebraic structures + -- ** Semigroup + , Semigroup (..) + -- ** Monoid + , Monoid (..) + -- ** Functor + , Functor (..) + , (<$>) + , ($>) + , void + , with + -- ** Bifunctor + , Bifunctor (..) + -- ** Applicative + , Applicative (..) + , (<**>) + , orA + , andA + , optional + -- ** Alternative + , Alternative (..) + , asum + -- ** Monad + , Monad (..) + , join + , bind + , when + , unless + , mapM_ + , forever + , unlessM + , whenM + , ifM + , guardM + , filterM + , (=<<) + , liftM + -- ** MonadPlus + , MonadPlus (..) + , guard + , msum + -- ** MonadIO + , MonadIO (..) + + -- * Data structures + -- ** Either + , Either (..) + , either + , isRight + -- ** Maybe + , Maybe (..) + , fromMaybe + , maybe + , isJust + , isNothing + , mapMaybe + , maybeToRight + , catMaybes + , listToMaybe + , rightToMaybe + -- ** Tuple + , fst + , snd + , curry + , uncurry + -- ** List + , module List + , ordNub + , ordNubBy + -- * Typeclasses + -- ** Enum + , Enum + -- ** Num + , Num (..) + -- ** Eq + , Eq (..) + -- ** Read + , Read (..) + , readEither + , readMaybe + -- ** Show + , Show (..) + -- *** ShowS + , ShowS + , showString + -- ** Foldable + , Foldable (..) + , for_ + , forM_ + , all + , head + , concat + , concatMap + -- ** Ord + , Ord (..) + , Ordering (..) + , comparing + -- ** Traversable + , Traversable (..) + , for + , forM + , traverse_ + + -- * Combinators + , id + , (.) + , ($) + , ($!) + , (&) + , const + , flip + , fix + , on + , seq + + -- * Text functions + , Text + -- * Partial functions + , undefined + , error + + -- * Debugging facilities + , trace + , traceM + , traceIO + ) where + + +import Control.Monad as Monad ( + Monad (..) + , MonadPlus (..) + , guard + , join + , msum + , when + , unless + , guard + , mapM_ + , forever + , (=<<) + , filterM + , liftM + ) +import Control.Monad.IO.Class ( + MonadIO (..) + ) +import Control.Applicative as Applicative ( + Applicative (..) + , (<**>) + , Alternative (..) + , empty + , liftA2 + , optional + ) + +import Data.Bifunctor as Bifunctor ( + Bifunctor (..) + ) +import Data.Bool as Bool ( + Bool (..) + , bool + , (&&) + , (||) + , not + , otherwise + ) +import Data.Char as Char ( + Char + ) +import Data.Either as Either ( + Either (..) + , either + , isRight + ) +import Data.Foldable as Foldable ( + Foldable (..) + , asum + , traverse_ + , for_ + , forM_ + , all + , concat + , concatMap + ) +import Data.Function as Function ( + id + , (.) + , ($) + , (&) + , const + , flip + , fix + , on + ) +import Data.Functor as Functor ( + Functor (..) + , (<$>) + , ($>) + , void + ) +import Data.Eq as Eq ( + Eq (..) + ) +import Data.Int as Int ( + Int + , Int8 + , Int16 + , Int32 + , Int64 + ) +import Data.Maybe as Maybe ( + Maybe (..) + , fromMaybe + , maybe + , isJust + , isNothing + , mapMaybe + , catMaybes + , listToMaybe + ) +import Data.Semigroup as Semigroup ( + Semigroup (..) + ) +import Data.Monoid as Monoid ( + Monoid (..) + ) +import Data.Ord as Ord ( + Ord (..) + , Ordering (..) + , comparing + ) +import Data.Traversable as Traversable ( + Traversable (..) + , for + , forM + , mapM + ) +import Data.Tuple as Tuple ( + fst + , snd + , curry + , uncurry + ) +import Data.Word as Word ( + Word64 + ) +import Data.List as List ( + intercalate + , isPrefixOf + , drop + , splitAt + , break + , filter + , reverse + , any + , and + , notElem +#if (__GLASGOW_HASKELL__ < 710) + , length + , null +#endif + ) +import qualified Debug.Trace as Trace +import GHC.Exts as Exts ( + Double + ) +import GHC.Real as Real ( + fromIntegral + , fromRational + , div + ) +#if MIN_VERSION_base(4,9,0) +import GHC.Stack (HasCallStack) +#endif + +import Prelude as Prelude ( + Enum + , Num (..) + , Integer + , toInteger + , seq + , ($!) + ) +import qualified Prelude as Unsafe + +import System.IO as IO ( + IO + ) +import Data.Text as Text (Text) +import Text.Read as Read ( + Read (..) + , readEither + , readMaybe + ) +import Text.Show as Show ( + Show (..) + , ShowS + , showString + ) +import qualified Data.Set as Set + +#if MIN_VERSION_base(4,9,0) +undefined :: HasCallStack => a +#else +undefined :: a +#endif +undefined = + Unsafe.undefined +{-# WARNING undefined "'undefined' is unsafe" #-} + +#if MIN_VERSION_base(4,9,0) +error :: HasCallStack => [Char] -> a +#else +error :: [Char] -> a +#endif +error = + Unsafe.error +{-# WARNING error "'error' is unsafe" #-} + +trace :: [Char] -> a -> a +trace = + Trace.trace +{-# WARNING trace "'trace' should only be used while debugging" #-} + +#if MIN_VERSION_base(4,9,0) +traceM :: Applicative f => [Char] -> f () +#else +traceM :: Monad m => [Char] -> m () +#endif +traceM = + Trace.traceM +{-# WARNING traceM "'traceM' should only be used while debugging" #-} + +traceIO :: [Char] -> IO () +traceIO = + Trace.traceIO +{-# WARNING traceIO "'traceIO' should only be used while debugging" #-} + +with :: Functor f => f a -> (a -> b) -> f b +with = + flip fmap +{-# INLINE with #-} + +whenM :: Monad m => m Bool -> m () -> m () +whenM p m = + p >>= flip when m + +unlessM :: Monad m => m Bool -> m () -> m () +unlessM p m = + p >>= flip unless m + +ifM :: Monad m => m Bool -> m a -> m a -> m a +ifM p x y = + p >>= \b -> if b then x else y + +guardM :: MonadPlus m => m Bool -> m () +guardM f = guard =<< f + +-- | Identifier version of 'Control.Monad.=<<'. +bind :: Monad m => (a -> m b) -> m a -> m b +bind = (=<<) + +head :: (Foldable f) => f a -> Maybe a +head = foldr (\x _ -> return x) Nothing + +-- | Logical disjunction. +orA :: Applicative f => f Bool -> f Bool -> f Bool +orA = + liftA2 (||) + +-- | Logical conjunction. +andA :: Applicative f => f Bool -> f Bool -> f Bool +andA = + liftA2 (&&) + +infixl 8 `andA`, `orA` + +maybeToRight :: l -> Maybe r -> Either l r +maybeToRight l = maybe (Left l) Right + +rightToMaybe :: Either l r -> Maybe r +rightToMaybe = either (const Nothing) Just + +-- | /O(n log n)/ Remove duplicate elements from a list. +-- +-- Unlike 'Data.List.nub', this version requires 'Ord' and runs in +-- /O(n log n)/ instead of /O(n²)/. Like 'Data.List.nub' the output +-- order is identical to the input order. +-- See 'sortNub' for `nub` behaviour with sorted output. +-- +-- > ordNub "foo bar baz" == "fo barz" +-- > ordNub [3,2,1,2,1] == [3,2,1] +-- > List.take 3 (ordNub [4,5,6,undefined]) == [4,5,6] +-- > ordNub xs == List.nub xs +-- +ordNub :: Ord a => [a] -> [a] +ordNub = + ordNubBy compare + +-- | /O(n log n)/ Behaves exactly like 'ordNub' except it uses a user-supplied +-- comparison function. +-- +-- > ordNubBy (comparing length) ["foo","bar","quux"] == ["foo","quux"] +-- > ordNubBy (comparing fst) [("foo", 10),("foo", 20),("bar", 30)] == [("foo", 10),("bar", 30)] +-- +ordNubBy :: (a -> a -> Ordering) -> [a] -> [a] +ordNubBy f = + let + loop seen = \case + [] -> + [] + x : xs -> + let + y = + UserOrd f x + in + if Set.member y seen then + loop seen xs + else + x : loop (Set.insert y seen) xs + in + loop Set.empty + +-- +-- Some machinery so we can use Data.Set with a custom comparator. +-- + +data UserOrd a = + UserOrd (a -> a -> Ordering) a + +instance Eq (UserOrd a) where + (==) (UserOrd f x) (UserOrd _ y) = + f x y == EQ + +instance Ord (UserOrd a) where + compare (UserOrd f x) (UserOrd _ y) = + f x y diff --git a/src/Entwine/Parallel.hs b/src/Entwine/Parallel.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Parallel.hs @@ -0,0 +1,152 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ExplicitForAll #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Entwine.Parallel ( + RunError (..) + , renderRunError + , consume_ + , consume + , waitEitherBoth + ) where + + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async, cancel, poll, waitBoth, wait, waitEither) +import Control.Concurrent.MSem (new, signal) +import qualified Control.Concurrent.MSem as M +import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar) +import Control.Monad.Catch +import Control.Monad.IO.Class +import Control.Monad.Loops (untilM_) +import Control.Monad.Trans.Either + +import qualified Data.Text as T +import Data.Typeable + +import Entwine.Async (waitEitherBoth) +import Entwine.Data.Parallel +import Entwine.Data.Queue +import Entwine.P + +import System.IO + +-- | Provide a producer and an action to be run across the result +-- of that producer in parallel. +-- +-- +-- Common usage: +-- @ +-- let producer :: Address -> Queue Address -> IO () +-- producer prefix q = +-- list' prefix $$ writeQueue q +-- +-- consume producer 100 (\(a :: Address) -> doThis) +-- @ +-- +consume_ :: MonadIO m => (Queue b -> IO a) -> Int -> (b -> EitherT e IO ()) -> EitherT (RunError e) m a +consume_ pro fork action = EitherT . liftIO $ do + q <- newQueue fork + + producer <- async $ pro q + + workers <- emptyWorkers + result <- newResult $ Right () + sem <- new fork + + let spawn = do + m <- tryReadQueue q + flip (maybe $ return ()) m $ \a -> do + w <- do + M.wait sem + async $ (runEitherT (action a) >>= void . addResult result . first WorkerError) `finally` signal sem + addWorker workers w + + let check = do + threadDelay 1000 {-- 1 ms --} + p <- poll producer + e <- isQueueEmpty q + pure $ (isJust p) && e + + submitter <- async $ untilM_ spawn check + + -- early termination + void . async . forever $ do + threadDelay 1000000 {-- 1 second --} + getResult result >>= + either (const $ cancel producer >> cancel submitter) pure + + let waiter = do + (i, _) <- waitBoth producer submitter + waitForWorkers workers + pure i + + (waiter >>= \i -> getResult result >>= pure . second (const $ i)) + `catchAll` (\z -> + failWorkers workers >> + getResult result >>= \w -> + pure (w >> Left (BlowUpError z))) + +data RunError a = + WorkerError a + | BlowUpError SomeException + deriving Show + +renderRunError :: RunError a -> (a -> Text) -> Text +renderRunError r render = + case r of + WorkerError a -> + "Worker failed: " <> render a + BlowUpError e -> + "An unknown exception was caught: " <> T.pack (show e) + +data EarlyTermination = + EarlyTermination deriving (Eq, Show, Typeable) + +instance Exception EarlyTermination + +consume :: forall a b c e . (Queue a -> IO b) -> Int -> (a -> IO (Either e c)) -> IO (Either (RunError e) (b, [c])) +consume pro fork action = flip catchAll (pure . Left . BlowUpError) $ do + q <- newQueue fork -- not fork + producer <- async $ pro q + workers <- (emptyWorkers :: IO (Workers c)) + sem <- new fork + early <- newEmptyMVar + + terminator <- async $ takeMVar early + + let spawn :: IO () + spawn = do + m <- tryReadQueue q + flip (maybe $ return ()) m $ \a -> do + w <- do + M.wait sem + async $ flip finally (signal sem) $ do + r <- action a + case r of + Left e -> + putMVar early e >> + throwM EarlyTermination + Right c -> + pure $! c + addWorker workers w + + let check = do + threadDelay 1000 {-- 1 ms --} + p <- poll producer + e <- isQueueEmpty q + pure $ (isJust p) && e + + submitter <- async $ untilM_ spawn check + + let waiter = runEitherT $ do + (i, _) <- bimapEitherT WorkerError id . EitherT $ waitEitherBoth terminator producer submitter + + ws <- liftIO $ getWorkers workers + ii <- mapM (bimapEitherT WorkerError id . EitherT . waitEither terminator) ws + pure $ (i, ii) + + waiter `catch` (\(_ :: EarlyTermination) -> + failWorkers workers >> + (Left . WorkerError) <$> wait terminator) diff --git a/src/Entwine/Snooze.hs b/src/Entwine/Snooze.hs new file mode 100644 --- /dev/null +++ b/src/Entwine/Snooze.hs @@ -0,0 +1,16 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Entwine.Snooze ( + snooze + , module Entwine.Data.Duration + ) where + +import Control.Concurrent + +import Entwine.P +import Entwine.Data.Duration + +import System.IO (IO) + +snooze :: Duration -> IO () +snooze = + threadDelay . toMicroseconds diff --git a/test/bench.hs b/test/bench.hs new file mode 100644 --- /dev/null +++ b/test/bench.hs @@ -0,0 +1,59 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE NoImplicitPrelude #-} +import Control.Concurrent +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Trans.Either + +import Criterion.Main +import Criterion.Types (Config (..)) + +import Entwine.Data +import Entwine.Parallel + +import Entwine.P + +import System.IO + +import Test.QuickCheck.Instances () + +run_ :: EitherT (RunError ()) IO a -> IO a +run_ = + eitherT (const $ fail "fail") pure + +run :: IO (Either (RunError ()) a) -> IO a +run = + (=<<) (either (const $ fail "fail") pure) + +main :: IO () +main = do + let pro n = \q -> forM_ [1..n] (writeQueue q) + + work_ :: Int -> EitherT () IO () + work_ = + EitherT . work + + work :: Int -> IO (Either () ()) + work _ = liftIO $ do + threadDelay 10000 {-- 10 ms --} + pure $ Right () + + let cfg = + defaultConfig { + reportFile = Just "dist/build/twine-bench.html" + } + + + defaultMainWith cfg [ + bgroup "sem-100" [ + bench "work-1000" (nfIO . run_ $ consume_ (pro 1000) 100 work_) + , bench "work-1000" (nfIO . run $ consume (pro 1000) 100 work) + ] + , bgroup "sem-1000" [ + bench "work-1000" (nfIO . run_ $ consume_ (pro 1000) 1000 work_) + , bench "work-1000" (nfIO . run $ consume (pro 1000) 1000 work) + , bench "work-10000" (nfIO . run_ $ consume_ (pro 10000) 1000 work_) + , bench "work-10000" (nfIO . run $ consume (pro 10000) 1000 work) + , bench "work-100000" (nfIO . run_ $ consume_ (pro 100000) 1000 work_) + , bench "work-100000" (nfIO . run $ consume (pro 100000) 1000 work) + ] + ] diff --git a/test/test-io.hs b/test/test-io.hs new file mode 100644 --- /dev/null +++ b/test/test-io.hs @@ -0,0 +1,14 @@ +import Test.Disorder (disorderMain) + +import qualified Test.IO.Entwine.Guard +import qualified Test.IO.Entwine.Loop +import qualified Test.IO.Entwine.Snooze + + +main :: IO () +main = + disorderMain [ + Test.IO.Entwine.Guard.tests + , Test.IO.Entwine.Loop.tests + , Test.IO.Entwine.Snooze.tests + ] diff --git a/test/test.hs b/test/test.hs new file mode 100644 --- /dev/null +++ b/test/test.hs @@ -0,0 +1,23 @@ +import Test.Disorder (disorderMain) + +import Test.Entwine.Async +import Test.Entwine.Data.Duration +import Test.Entwine.Data.Finalizer +import Test.Entwine.Data.Gate +import Test.Entwine.Data.Parallel +import Test.Entwine.Data.Pin +import Test.Entwine.Data.Queue +import Test.Entwine.Parallel + +main :: IO () +main = + disorderMain [ + Test.Entwine.Async.tests + , Test.Entwine.Data.Duration.tests + , Test.Entwine.Data.Finalizer.tests + , Test.Entwine.Data.Gate.tests + , Test.Entwine.Data.Parallel.tests + , Test.Entwine.Data.Pin.tests + , Test.Entwine.Data.Queue.tests + , Test.Entwine.Parallel.tests + ]