entwine (empty) → 0.0.1

raw patch · 22 files changed

+1384/−0 lines, 22 filesdep +QuickCheckdep +SafeSemaphoredep +asyncsetup-changed

Dependencies added: QuickCheck, SafeSemaphore, async, base, containers, criterion, directory, entwine, exceptions, monad-loops, process, quickcheck-instances, quickcheck-properties, quickcheck-text, random, stm, text, time, transformers, transformers-either

Files

• Changes.md +3/−0
• LICENSE +28/−0
• README.md +4/−0
• Setup.hs +2/−0
• entwine.cabal +126/−0
• src/Entwine.hs +8/−0
• src/Entwine/Async.hs +76/−0
• src/Entwine/Data.hs +9/−0
• src/Entwine/Data/Duration.hs +67/−0
• src/Entwine/Data/Finalizer.hs +31/−0
• src/Entwine/Data/Gate.hs +38/−0
• src/Entwine/Data/Parallel.hs +70/−0
• src/Entwine/Data/Pin.hs +41/−0
• src/Entwine/Data/Queue.hs +43/−0
• src/Entwine/Guard.hs +81/−0
• src/Entwine/Loop.hs +24/−0
• src/Entwine/P.hs +469/−0
• src/Entwine/Parallel.hs +152/−0
• src/Entwine/Snooze.hs +16/−0
• test/bench.hs +59/−0
• test/test-io.hs +14/−0
• test/test.hs +23/−0

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

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

@@ -0,0 +1,4 @@+Twine+=====++Concurrency tools.

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

@@ -0,0 +1,126 @@+name:                  entwine+version:               0.0.1+license:               BSD3+license-file:          LICENSE+author:                Ambiata <info@ambiata.com>+maintainer:            Tim McGilchrist <timmcgil@gmail.com>+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            

@@ -0,0 +1,8 @@+{-# LANGUAGE NoImplicitPrelude #-}+module Entwine (+    module Entwine.Data+  , module Entwine.Snooze+  ) where++import           Entwine.Data+import           Entwine.Snooze

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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