entwine 0.0.2 → 0.0.3
raw patch · 16 files changed
+110/−54 lines, 16 filesdep +clockdep +retrydep +semigroupsdep ~stmdep ~transformers
Dependencies added: clock, retry, semigroups
Dependency ranges changed: stm, transformers
Files
- Changes.md +4/−0
- README.md +1/−1
- entwine.cabal +9/−5
- src/Entwine/Async.hs +4/−5
- src/Entwine/Data/Duration.hs +2/−0
- src/Entwine/Data/Finalizer.hs +1/−1
- src/Entwine/Data/Gate.hs +3/−1
- src/Entwine/Data/Parallel.hs +13/−6
- src/Entwine/Data/Pin.hs +5/−2
- src/Entwine/Data/Queue.hs +15/−5
- src/Entwine/Guard.hs +2/−0
- src/Entwine/Loop.hs +0/−2
- src/Entwine/Parallel.hs +38/−19
- src/Entwine/Snooze.hs +5/−3
- test/Test/Entwine/Parallel.hs +8/−2
- test/bench.hs +0/−2
Changes.md view
@@ -1,3 +1,7 @@+* 0.0.3+ - Add support for 7.10+ - Upgrades STM version to 2.5 (which made changes in their API from Int to Natural for some parameters)+ * 0.0.2 - Update async to 2.2. This version of async changed the exceptions thrown, see https://hackage.haskell.org/package/async-2.2/changelog for more details.
README.md view
@@ -1,4 +1,4 @@-Twine+Entwine ===== Concurrency tools.
entwine.cabal view
@@ -1,5 +1,5 @@ name: entwine-version: 0.0.2+version: 0.0.3 license: BSD3 license-file: LICENSE author: Ambiata <info@ambiata.com>@@ -11,9 +11,9 @@ category: System cabal-version: >= 1.8 build-type: Simple-description: entwine - Concurrency tools+description: Entwine provides concurrency types and tools for building correct software. -tested-with: GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.4+tested-with: GHC == 7.10.3, GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.4 extra-source-files: README.md Changes.md@@ -26,11 +26,14 @@ build-depends: base >= 3 && < 5 , async == 2.2.*+ , clock >= 0.6 && < 0.8 , containers >= 0.5 && < 0.7 , exceptions >= 0.6 && < 0.9 , monad-loops == 0.4.*+ , retry , SafeSemaphore == 0.10.*- , stm == 2.4.*+ , semigroups == 0.18.*+ , stm == 2.5.* , text == 1.2.* , time >= 1.4 && < 1.9 , transformers >= 0.3 && < 0.6@@ -54,8 +57,8 @@ Entwine.Data.Pin Entwine.Data.Queue Entwine.Loop- Entwine.Parallel Entwine.P+ Entwine.Parallel Entwine.Snooze Entwine.Guard @@ -119,6 +122,7 @@ , process >= 1.2 && < 1.7 , text , time+ , transformers , transformers-either , QuickCheck == 2.10.* , quickcheck-instances == 0.3.*
src/Entwine/Async.hs view
@@ -14,8 +14,7 @@ import Control.Concurrent.Async (async, cancel, wait, AsyncCancelled) import Control.Concurrent.STM (atomically, orElse, retry) import Control.Monad.Catch (catches, throwM, Handler(..))-import Control.Monad.IO.Class (liftIO)-import Control.Monad.Trans.Either+import Control.Monad.Trans.Either (EitherT, left) import Data.IORef (newIORef, readIORef, writeIORef) import qualified Data.Text as T@@ -25,7 +24,7 @@ import System.IO (IO) -data AsyncTimeout =+newtype AsyncTimeout = AsyncTimeout Duration deriving (Eq, Show) @@ -46,12 +45,12 @@ e <- liftIO $ waitEither a s case e of Left a' ->- pure $ a'+ pure a' Right _ -> do liftIO $ writeIORef r True liftIO $ cancel a (liftIO $ wait a)- `catches` [ Handler (\ (ax :: AsyncCancelled) -> do+ `catches` [ Handler (\ (ax :: AsyncCancelled) -> liftIO (readIORef r) >>= bool (liftIO $ throwM ax) (left $ AsyncTimeout d) )]
src/Entwine/Data/Duration.hs view
@@ -1,6 +1,8 @@ {-# LANGUAGE NoImplicitPrelude #-} module Entwine.Data.Duration (+ -- * Types Duration+ -- * Functions , microseconds , milliseconds , seconds
src/Entwine/Data/Finalizer.hs view
@@ -5,7 +5,7 @@ import Entwine.P -import System.IO+import System.IO (IO) -- |
src/Entwine/Data/Gate.hs view
@@ -1,6 +1,8 @@ {-# LANGUAGE NoImplicitPrelude #-} module Entwine.Data.Gate (+ -- * Types Gate+ -- * Functions , newGate , isOpen , close@@ -10,7 +12,7 @@ import Entwine.P -import System.IO+import System.IO (IO) -- | -- A gate is an abstract type, representing a simple barrier
src/Entwine/Data/Parallel.hs view
@@ -1,7 +1,9 @@ {-# LANGUAGE NoImplicitPrelude #-} module Entwine.Data.Parallel (+ -- * Types Workers , Result+ -- * Functions , newResult , emptyResult , addResult@@ -15,12 +17,15 @@ ) where import Control.Concurrent.Async (Async, cancel, wait)-import Control.Concurrent.MVar+import Control.Concurrent.MVar (MVar, newEmptyMVar, newMVar, modifyMVar+ , modifyMVar_, readMVar) import Entwine.P -import System.IO+import System.IO (IO) +-- |+-- newtype Result a = Result (MVar a) @@ -35,12 +40,14 @@ addResult :: Monad m => Result (m a) -> m a -> IO (m a) addResult (Result r) w =- modifyMVar r (\x -> pure $ ((x >> w), x))+ modifyMVar r (\x -> pure (x >> w, x)) getResult :: Result a -> IO a getResult (Result r) = readMVar r +-- |+-- newtype Workers a = Workers (MVar [Async a]) @@ -57,14 +64,14 @@ getWorkers (Workers w) = readMVar w -addWorker :: (Workers a) -> Async a -> IO ()+addWorker :: Workers a -> Async a -> IO () addWorker (Workers w) r = modifyMVar_ w $ pure . (:) r -waitForWorkers :: (Workers a) -> IO ()+waitForWorkers :: Workers a -> IO () waitForWorkers (Workers w) = readMVar w >>= mapM_ wait -waitForWorkers' :: (Workers a) -> IO [a]+waitForWorkers' :: Workers a -> IO [a] waitForWorkers' (Workers w) = readMVar w >>= mapM wait
src/Entwine/Data/Pin.hs view
@@ -1,17 +1,20 @@ {-# LANGUAGE NoImplicitPrelude #-} module Entwine.Data.Pin (+ -- * Types Pin+ -- * Functions , newPin , checkPin , waitForPin , pullPin ) where -import Control.Concurrent.MVar+import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar+ , readMVar, tryPutMVar) import Entwine.P -import System.IO+import System.IO (IO) -- | -- A pin is an abstract type, representing a simple barrier
src/Entwine/Data/Queue.hs view
@@ -8,21 +8,31 @@ , isQueueEmpty ) where -import Control.Concurrent.STM.TBQueue ( TBQueue, newTBQueue, tryReadTBQueue- , readTBQueue, writeTBQueue, isEmptyTBQueue)+import Control.Concurrent.STM.TBQueue (TBQueue, isEmptyTBQueue, newTBQueue, readTBQueue, tryReadTBQueue,+ writeTBQueue) +import Entwine.P+ import GHC.Conc (atomically) -import Entwine.P+import Numeric.Natural (Natural) -import System.IO+import System.IO (IO) +-- |+--+-- A queue is an abstract type, representing a simple bounded FIFO channel+-- that can only have its state queried in a non-blocking manner.+--+-- This useful for implementing things like passing data between a+-- producer and consumer process with back presure.+-- newtype Queue a = Queue { queue :: TBQueue a } -newQueue :: Int -> IO (Queue a)+newQueue :: Natural -> IO (Queue a) newQueue i = atomically $ Queue <$> newTBQueue i
src/Entwine/Guard.hs view
@@ -1,7 +1,9 @@ {-# LANGUAGE NoImplicitPrelude #-} module Entwine.Guard (+ -- * Types TerminationAction (..) , TerminationHandler (..)+ -- * Functions , guarded , repeatedly ) where
src/Entwine/Loop.hs view
@@ -4,8 +4,6 @@ loop ) where -import Control.Monad.IO.Class (MonadIO, liftIO)- import Entwine.Data.Gate import Entwine.Data.Duration import Entwine.P
src/Entwine/Parallel.hs view
@@ -1,42 +1,46 @@+{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE ExplicitForAll #-}-{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE ScopedTypeVariables #-} module Entwine.Parallel (+ -- * Types RunError (..)+ -- * Functions , renderRunError- , consume_ , consume+ , consume_ , waitEitherBoth ) where import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (async, cancel, poll, waitBoth, wait, waitEither)+import Control.Concurrent.Async (async, cancel, poll, wait, waitBoth, 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.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Monad.Catch (Exception (..), SomeException, catch, catchAll, finally, throwM) import Control.Monad.Loops (untilM_)-import Control.Monad.Trans.Either+import Control.Monad.Trans.Either (EitherT, pattern EitherT, bimapEitherT, runEitherT) import qualified Data.Text as T-import Data.Typeable+import Data.Typeable (Typeable) import Entwine.Async (waitEitherBoth) import Entwine.Data.Parallel import Entwine.Data.Queue import Entwine.P -import System.IO+import Numeric.Natural (Natural) +import System.IO (IO)+ -- | Provide a producer and an action to be run across the result -- of that producer in parallel.---+-- For a version that doesn't ignore the results see 'Entwine.Parallel.consume'. -- -- Common usage:+-- -- @ -- let producer :: Address -> Queue Address -> IO () -- producer prefix q =@@ -45,7 +49,7 @@ -- consume producer 100 (\(a :: Address) -> doThis) -- @ ---consume_ :: MonadIO m => (Queue b -> IO a) -> Int -> (b -> EitherT e IO ()) -> EitherT (RunError e) m a+consume_ :: MonadIO m => (Queue b -> IO a) -> Natural -> (b -> EitherT e IO ()) -> EitherT (RunError e) m a consume_ pro fork action = EitherT . liftIO $ do q <- newQueue fork @@ -67,7 +71,7 @@ threadDelay 1000 {-- 1 ms --} p <- poll producer e <- isQueueEmpty q- pure $ (isJust p) && e+ pure $ isJust p && e submitter <- async $ untilM_ spawn check @@ -82,7 +86,7 @@ waitForWorkers workers pure i - (waiter >>= \i -> getResult result >>= pure . second (const $ i))+ (waiter >>= \i -> getResult result >>= pure . second (const i)) `catchAll` (\z -> failWorkers workers >> getResult result >>= \w ->@@ -106,8 +110,23 @@ 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+-- | Provide a producer and an action to be run across the results+-- of that producer in parallel and collect the results.+-- For a version that ignores the results see 'Entwine.Parallel.consume_'.+--+-- Common usage:+--+-- @+-- let producer :: Address -> Queue Address -> IO Int+-- producer prefix q =+-- list' prefix $$ writeQueue q+--+-- consume producer 100 (\(a :: Address) -> doThis)+-- @+--+consume :: MonadIO m => (Queue b -> IO a) -> Natural -> (b -> IO (Either e c))+ -> m (Either (RunError e) (a, [c]))+consume pro fork action = liftIO . flip catchAll (pure . Left . BlowUpError) $ do q <- newQueue fork -- not fork producer <- async $ pro q workers <- (emptyWorkers :: IO (Workers c))@@ -136,7 +155,7 @@ threadDelay 1000 {-- 1 ms --} p <- poll producer e <- isQueueEmpty q- pure $ (isJust p) && e+ pure $ isJust p && e submitter <- async $ untilM_ spawn check @@ -145,8 +164,8 @@ ws <- liftIO $ getWorkers workers ii <- mapM (bimapEitherT WorkerError id . EitherT . waitEither terminator) ws- pure $ (i, ii)+ pure (i, ii) waiter `catch` (\(_ :: EarlyTermination) -> failWorkers workers >>- (Left . WorkerError) <$> wait terminator)+ Left . WorkerError <$> wait terminator)
src/Entwine/Snooze.hs view
@@ -1,16 +1,18 @@ {-# LANGUAGE NoImplicitPrelude #-} module Entwine.Snooze ( snooze- , module Entwine.Data.Duration+ , module X ) where -import Control.Concurrent+import Control.Concurrent (threadDelay) import Entwine.P-import Entwine.Data.Duration+import Entwine.Data.Duration as X import System.IO (IO) +-- | Snooze for a short duration+-- snooze :: Duration -> IO () snooze = threadDelay . toMicroseconds
test/Test/Entwine/Parallel.hs view
@@ -79,9 +79,15 @@ work = const $ pure () r <- runEitherT $ consume_ pro 1 work- pure $ (isRight r) === True+ pure $ isRight r === True +prop_empty_producer_consume = testIO $ do+ let pro = const . pure . Right $ (1 :: Int)+ work = const . pure . Right $ (1 :: Int) + r <- runEitherT $ consume pro 1 work+ pure $ isRight r === True+ return [] tests :: IO Bool-tests = $quickCheckAll+tests = $forAllProperties $ quickCheckWithResult (stdArgs { maxSuccess = 10 })
test/bench.hs view
@@ -1,7 +1,6 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE NoImplicitPrelude #-} import Control.Concurrent-import Control.Monad.IO.Class (liftIO) import Control.Monad.Trans.Either import Criterion.Main@@ -9,7 +8,6 @@ import Entwine.Data import Entwine.Parallel- import Entwine.P import System.IO