churros 0.1.4.1 → 0.1.5.0

raw patch · 4 files changed

+41/−12 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 +23/−11
• src/Control/Churro/Transport/Unagi/Bounded.hs +12/−0

@@ -1,5 +1,10 @@ # Revision history for Churros +## 0.1.5.1 -- 2020-10-20++* Generalising processes to allow for different transport for process group.+* Added `processesUnagi` function.+ ## 0.1.4.1 -- 2020-10-20  * Adding `processes` function that runs a set of processes over the input.

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

@@ -275,16 +275,22 @@ --  -- 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.+-- * NOTE: This makes no judgement about the ordering of outputs corresponding to the ordering of inputs.+-- * NOTE: You will need to specialise the transport of the processes. This is deliberate as it allows you+--         to use a bounded channel that ensures allocation to idle processes.+--         Use the `processesUnagi` variant from `Control.Churro.Transport.Unagi.Bounded` to default to+--         a buffer size of 1. --  -- WARNING: This won't deterministically allocate work to idle workers unless a bounded channel is used. -- +-- * TODO: Figure out cancellation strategy.+-- * TODO: Consider a binary combinator and this as a folded application.+-- +-- >>> import Control.Churro.Transport.Unagi.Bounded (processesUnagi)+--  -- 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))+-- >>> runWaitListChan $ sourceList [1,1,1,1,1] >>> processesUnagi (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.@@ -293,14 +299,14 @@ --  -- >>> :{ -- do---   timeout 10000000 $ runWaitListChan $ sourceList (replicate 10 1) >>> processes (replicate 1 $ delay 0.05)+--   timeout 10000000 $ runWaitListChan $ sourceList (replicate 10 1) >>> processesUnagi (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 :: (Traversable f, Transport t1, Transport t2, Monoid a) => f (Churro a t1 i o) -> Churro a t2 i o processes cs = Churro do     (i,  o ) <- flex     (i', o') <- flex@@ -310,8 +316,9 @@             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+                    yeet i Nothing -- Ensure other consumers aren't blocked. FIXME: This produces one more `Nothing` than is required.+                c2c' o ci+                yeet ci Nothing                 wait a'                 wait ca @@ -327,12 +334,17 @@     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.+    -- while an earlier slower worker still hasn't finished propagating its result.+    -- Also allows two different types of Transport.+    -- +    c2c' :: (Transport t1, Transport t2) => Out t1 (Maybe i) -> In t2 (Maybe i) -> IO ()     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+-- Consider using `thiefUnagi` unless you have a requirement for controlling the transport of the process group.+-- +thief :: (Transport t1, Transport t2, Monoid a) => Int -> Churro a t1 i o -> Churro a t2 i o thief n c = processes (replicate n c)  -- | Extract xs from (Just x)s. Similar to `catMaybes`.

@@ -49,3 +49,15 @@ -- [2,3,4] runWaitListUnagi :: KnownNat n => ChurroUnagiBounded () n Void o -> IO [o] runWaitListUnagi = runWaitList++-- A version of processes that mandates the use of a single item buffer for the internal processes.+-- This is useful in order to prevent allocating to processes that are not yet idle.+-- +processesUnagi :: (Traversable f, Transport t, Monoid a) => f (ChurroUnagiBounded a 1 i o) -> Churro a t i o+processesUnagi = processes++-- A version of thief that mandates the use of a single item buffer for the internal processes.+-- This is useful in order to prevent allocating to processes that are not yet idle.+-- +thiefUnagi :: (Transport t, Monoid a) => Int -> ChurroUnagiBounded a 1 i o -> Churro a t i o+thiefUnagi n c = processes (replicate n c)