churros 0.1.4.0 → 0.1.4.1

raw patch · 4 files changed

+116/−15 lines, 4 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−0
• churros.cabal +1/−1
• src/Control/Churro/Prelude.hs +98/−11
• src/Control/Churro/Types.hs +12/−3

@@ -1,5 +1,10 @@ # Revision history for Churros +## 0.1.4.1 -- 2020-10-20++* Adding `processes` function that runs a set of processes over the input.+* Adding `withChurro` helper function.+ ## 0.1.4.0 -- 2020-10-18  * Generalising list type to Foldable/Traversible where possible (`sources`).

@@ -1,6 +1,6 @@ cabal-version:       2.2 name:                churros-version:             0.1.4.0+version:             0.1.4.1 license-file:        LICENSE author:              Lyndon Maydwell maintainer:          lyndon@sordina.net

@@ -21,14 +21,15 @@ import           Control.Arrow            (arr) import           Control.Category         (id, (.), (>>>)) import           Control.Concurrent       (threadDelay)-import           Control.Concurrent.Async (cancel, Async, wait)+import           Control.Concurrent.Async (async, cancel, Async, wait) import           Control.Exception        (Exception, SomeException, try) import           Control.Monad            (replicateM_, when)-import           Data.Foldable            (toList, for_)+import           Data.Foldable            (foldMap', for_) import           Data.Maybe               (isJust) import           Data.Time                (NominalDiffTime) import           Data.Void                (Void) import           GHC.Natural              (Natural)+import Data.Traversable (for)   -- $setup@@ -38,6 +39,7 @@ -- >>> :set -XBlockArguments -- >>> import Control.Churro.Transport -- >>> import Data.Time.Clock+-- >>> import System.Timeout (timeout) --   -- * Runners@@ -134,19 +136,35 @@ --  -- Sends individual items downstream without attempting to combine them. -- --- Warning: Passing an empty list is unspecified.+-- >>> runWaitChan $ sources [pure 1, pure 1] >>> sinkPrint+-- 1+-- 1 -- --- TODO: Use NonEmptyList instead of []+-- Can combine results of sources with a Monoid instance, although this isn't very useful+-- when forming the start of a longer pipeline: -- --- >>> runWaitChan $ sources_ [pure 1, pure 1] >>> sinkPrint+-- >>> :{+-- do+--   r <- runWaitChan $ sources [sourceIO \_cb -> print 1 >> return "hello ", sourceIO \_cb -> print 1 >> return "world"]+--   print r+-- :} -- 1 -- 1-sources :: (Transport t, Foldable f, Traversable f) => f (Churro a t Void i) -> Churro () t Void i-sources ss = sourceIO \cb -> do-    asyncs <- mapM (\s -> run $ s >>>> sinkIO cb) ss-    let as = toList asyncs-    finally' (mapM_ cancel as) do-        mapM_ wait as+-- "hello world"+sources :: (Transport t, Traversable f, Monoid a) => f (Churro a t Void o) -> Churro a t Void o+sources ss = buildChurro \_i o -> do+    cs <- mapM runChurro ss+    finally' (mapM_ (\(_,_,a) -> cancel a) cs) do+        as <- for cs \(_i,o',a) ->+            async do+                yankAll' o' \v -> do+                    case v of+                        Nothing -> return ()+                        Just x  -> yeet o (Just x)+                wait a+        r <- foldMap' wait as+        yeet o Nothing+        return r  -- | Variant of `sources` with Async action of sources in argument specialised to `()`. -- @@ -247,6 +265,75 @@     buildChurro \i o -> do         yankAll i \x -> do mapM_ (yeet o . Just) =<< f x         yeet o Nothing++-- | Concatenates splits lists of items into individual items.+-- +concatC :: Transport t => Churro () t [o] o+concatC = processN (pure . id)++-- | Run a set of churros like a work-stealing queue for its inputs.+-- +-- Similar to ArrowChoice, but more straightforward due to unified output type and independent implementation.+-- +-- Note: `processes` makes no judgement about the ordering of outputs corresponding to the ordering of inputs.+-- +-- TODO: Figure out cancellation strategy.+-- TODO: Consider a binary combinator and this as a folded application.+-- +-- WARNING: This won't deterministically allocate work to idle workers unless a bounded channel is used.+-- +-- Sanity check - All items entering should propagate, independent of the number of processes:+-- +-- >>> runWaitListChan $ sourceList [1,1,1,1,1] >>> processes (replicate 3 (delay 0.1))+-- [1,1,1,1,1]+-- +-- This example creates a source of 10 values, then creates a process of 10 workers that all wait 1/2 a second.+-- If this works, then all ten values should be consumed and propagated in 1/2 a second by distributing the load+-- over the set of 10 workers:+-- +-- >>> :{+-- do+--   timeout 10000000 $ runWaitListChan $ sourceList (replicate 10 1) >>> processes (replicate 1 $ delay 0.05)+-- :}+-- Just [1,1,1,1,1,1,1,1,1,1]+-- +-- We could use different strategies such as round-robin, etc. to default to a more balanced allocation, but this wouldn't+-- be most efficient if each worker performed at different rates of consumption.+-- +processes :: (Traversable f, Transport t, Monoid a) => f (Churro a t i o) -> Churro a t i o+processes cs = Churro do+    (i,  o ) <- flex+    (i', o') <- flex++    let+        worker c = async do+            withChurro c \ci co ca -> do+                a' <- async do+                    c2c' co i'+                    yeet i Nothing -- Ensure other consumers aren't blocked. FIXME: This produces once more Nothing than is required.+                c2c id o ci+                wait a'+                wait ca++    as <- mapM worker cs++    a  <- async do+        r <- foldMap' wait as+        yeet i' Nothing -- Make sure to conclude the process once all the processes have finished consuming+        return r++    return (i,o',a)++    where+    -- Version of c2c that doesn't propagate Nothing once transport is consumed.+    -- This is required here since we don't want a worker to be able to prematurely terminate the processes+    -- While an earlier slower works still hasn't finished propagating its result.+    c2c' o i = yankAll o (yeet i . Just)++-- | Set up N worker churro processes to concurrently process the stream.+-- +thief :: (Transport t, Monoid a) => Int -> Churro a t i o -> Churro a t i o+thief n c = processes (replicate n c)  -- | Extract xs from (Just x)s. Similar to `catMaybes`. -- 

@@ -67,8 +67,8 @@ class Transport (t :: * -> *) where     data In  t :: * -> *     data Out t :: * -> *-    flex :: IO (In t a, Out t a)  -- ^ Create a new pair of transports.-    yank :: Out t a -> IO a       -- ^ Yank an item of the Transport+    flex :: IO (In t a, Out t a)  -- ^ Create a new pair of Transports.+    yank :: Out t a -> IO a       -- ^ Yank an item off the Transport     yeet :: In t a -> a -> IO ()  -- ^ Yeet an item onto the Transport  -- | Covariant functor instance for Churro - Maps over the output.@@ -243,6 +243,15 @@     a       <- async do cb ai ao bi     return (ai,bo,a) +-- | Helper. Finalises cancellation of async.+-- +-- Use instead of runChurro unless you want to directly manage cancellation.+-- +withChurro :: Churro a t i o -> (In t (Maybe i) -> Out t (Maybe o) -> Async a -> IO b) -> IO b+withChurro c f = do+    (i,o,a) <- runChurro c+    finally' (cancel a) do f i o a+ -- | Yeet all items from a list into a raw transport. --  -- WARNING: If you are using this to build a churro by hand make sure you yeet Nothing once you're finished.@@ -268,7 +277,7 @@  -- | Yank each raw item from a transport into a callback. -- --- The items are wrapped in Maybes and when all items are yanked, Nothing is fed to the callback.+-- The items are wrapped in Maybes and when Nothing is yanked, Nothing is fed to the callback and `yankAll'` completes. --  yankAll' :: (Transport t, Monoid b) => Out t (Maybe a) -> (Maybe a -> IO b) -> IO b yankAll' c f = do