diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Revision history for churros + +## 0.1.0.0 -- YYYY-mm-dd + +* First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +Copyright 2020 Lyndon Maydwell + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,6 @@ + +import Distribution.Simple +main = defaultMain + +-- import Distribution.Extra.Doctest (defaultMainWithDoctests) +-- main = defaultMainWithDoctests "generic-data-doctest" diff --git a/churros.cabal b/churros.cabal new file mode 100644 --- /dev/null +++ b/churros.cabal @@ -0,0 +1,57 @@ +cabal-version: 2.2 +name: churros +version: 0.1.0.0 +license-file: LICENSE +author: Lyndon Maydwell +maintainer: lyndon@sordina.net +build-type: Simple +extra-source-files: CHANGELOG.md +license: MIT +synopsis: Churros: Channel/Arrow based streaming computation library. +category: Data, Control +description: The Churro library takes an opinionated approach to streaming + by focusing on IO processes and allowing different transport + options. + +source-repository head + type: git + location: git@github.com:sordina/churros.git + +-- custom-setup +-- setup-depends: +-- base, Cabal, cabal-doctest + +common churros-dependencies + default-language: Haskell2010 + build-depends: + base >= 4.10 && < 4.15, + time, + async, + stm, + containers, + random + +library churros-lib + import: churros-dependencies + hs-source-dirs: src + exposed-modules: + Control.Churro, + Control.Churro.Types, + Control.Churro.Prelude, + Control.Churro.Transport, + Control.Churro.Transport.Chan + +test-suite churros-test + import: churros-dependencies + type: exitcode-stdio-1.0 + hs-source-dirs: src,test + main-is: Churro/Test/Main.hs + build-depends: doctest + ghc-options: -threaded -rtsopts + other-modules: + Control.Churro, + Control.Churro.Types, + Control.Churro.Prelude, + Control.Churro.Transport, + Control.Churro.Transport.Chan, + Churro.Test.Examples diff --git a/src/Control/Churro.hs b/src/Control/Churro.hs new file mode 100644 --- /dev/null +++ b/src/Control/Churro.hs @@ -0,0 +1,57 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE NoMonomorphismRestriction #-} +{-# LANGUAGE Arrows #-} + +-- | Recommended entrypoint for the Churro library. +-- +-- Contains enough functionality to use the package as intended. +-- +-- See README.md for more information. +-- +module Control.Churro + + ( module Control.Churro + + -- ** Churro Packages + + , module Control.Churro.Types + , module Control.Churro.Prelude + , module Control.Churro.Transport + + -- ** Other Packages + + -- *** Re-exported from Control.Category + , Control.Category.id + + -- *** Re-exported from Control.Arrow + , Control.Arrow.arr + , Control.Arrow.first + , Control.Arrow.second + , (Control.Arrow.>>>) + , (Control.Arrow.<<<) + , (Control.Arrow.&&&) + , (Control.Arrow.***) + + -- *** Re-exported from Data.Void + , Void() + + -- *** Re-exported from GHC.Natural + , Natural() + ) + +where + +import Control.Churro.Types +import Control.Churro.Prelude +import Control.Churro.Transport + +import Data.Void +import GHC.Natural +import Control.Arrow +import Control.Category + +million :: Fractional p => p +million = 1e6 diff --git a/src/Control/Churro/Prelude.hs b/src/Control/Churro/Prelude.hs new file mode 100644 --- /dev/null +++ b/src/Control/Churro/Prelude.hs @@ -0,0 +1,323 @@ +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE BlockArguments #-} + +-- | Common transport-agnostic functions for using Churro. +-- +module Control.Churro.Prelude where + +import Control.Churro.Types + +import Prelude hiding (id, (.)) + +import Control.Arrow (arr) +import Control.Category (id, (.), (>>>)) +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (Async, wait) +import Control.Exception (Exception, SomeException, try) +import Control.Monad (replicateM_, when) +import Data.Foldable (for_) +import Data.IORef (newIORef, readIORef, writeIORef) +import Data.Maybe (isJust) +import Data.Time (NominalDiffTime) +import Data.Void (Void) +import GHC.Natural (Natural) + + +-- $setup +-- +-- The examples in this module require the following imports: +-- +-- >>> import Control.Churro.Transport +-- >>> import Data.Time.Clock +-- + +-- * Runners + +-- | Automatically wait for a churro to complete. +-- +runWait :: Transport t => Churro t Void Void -> IO () +runWait x = wait =<< run x + +-- | Read the output of a Churro into a list. +-- +-- Warning: This will block until the Churro terminates, +-- Accumulating items in memory. +-- Only use when you expect a finite amount of output. +-- Otherwise consider composing with a Sink and using `runWait`. +-- +-- >>> runWaitListChan $ sourceList [0..4] >>> arr succ +-- [1,2,3,4,5] +-- +runWaitList :: Transport t => Churro t Void o -> IO [o] +runWaitList x = do + t <- newIORef [] + + let + c = buildChurro \i _o -> do + l <- yankList i + writeIORef t l + + runWait $ x >>> c + + readIORef t + +-- | Run a sourced and sinked (double-dipped) churro and return an async action representing the in-flight processes. +-- +run :: Transport t => Churro t Void Void -> IO (Async ()) +run = run' + +-- | Run any churro, there is no check that this was spawned with a source, or terminated with a sink. +-- This is unsafe, since the pipeline may not generate or consume in a predictable way. +-- Use `run` instead unless you are confident you know what you're doing. +-- +run' :: Transport t => Churro t i o -> IO (Async ()) +run' c = do + -- Compose an empty sourceList to ensure termination + (_i,_o,a) <- runChurro (sourceList [] >>> c) + return a + +-- * Library + +-- ** Sources + +-- | A single items source. +-- +-- >>> runWaitChan $ sourceSingleton 13 >>> sinkPrint +-- 13 +-- +-- Equivalent to `pure` from `Applicative`. Redefined here in case you're looking for a source! +-- +-- >>> runWaitChan $ pure 23 >>> sinkPrint +-- 23 +sourceSingleton :: Transport t => o -> Churro t Void o +sourceSingleton x = sourceList [x] + +-- | Create a source from a list of items, sending each down the churro independently. +-- +-- >>> runWaitChan $ sourceList [4,2] >>> sinkPrint +-- 4 +-- 2 +sourceList :: (Transport t, Foldable f) => f o -> Churro t Void o +sourceList = sourceIO . for_ + +-- | Create a source from an IO action that is passed a function to yield new items. +-- +-- >>> runWaitChan $ sourceIO (\cb -> cb 4 >> cb 2) >>> sinkPrint +-- 4 +-- 2 +sourceIO :: Transport t => ((o -> IO ()) -> IO a2) -> Churro t Void o +sourceIO cb = + buildChurro \_i o -> do + cb (yeet o . Just) + yeet o Nothing + +-- ** Sinks + +-- | Consume a churro with an IO process. +-- +-- >>> runWaitChan $ pure 1 >>> sinkIO (\x -> print "hello" >> print (succ x)) +-- "hello" +-- 2 +sinkIO :: Transport t => (o -> IO ()) -> Churro t o Void +sinkIO cb = buildChurro \i _o -> yankAll i cb + +-- | Consume and print each item. Used in many examples, but not much use outside debugging! +-- +-- >>> runWaitChan $ pure "hi" >>> sinkPrint +-- "hi" +sinkPrint :: (Transport t, Show a) => Churro t a Void +sinkPrint = sinkIO print + +-- ** Churros + +-- | Process each item with an IO action. +-- Acts as a one-to-one process. +-- +-- >>> runWaitChan $ pure "hi" >>> process (\x -> print x >> return (reverse x)) >>> sinkPrint +-- "hi" +-- "ih" +process :: Transport t => (a -> IO b) -> Churro t a b +process f = processN (fmap pure . f) + +-- | Print each item then pass it on. +processPrint :: (Transport t, Show b) => Churro t b b +processPrint = process \x -> do print x >> return x + +-- | Print each item with an additional debugging label. +processDebug :: (Transport t, Show b) => String -> Churro t b b +processDebug d = process \x -> putStrLn ("Debugging [" <> d <> "]: " <> show x) >> return x + +-- | Process each item with an IO action and potentially yield many items as a result. +-- Acts as a one-to-many process. +-- +-- >>> runWaitChan $ pure 1 >>> processN (\x -> print (show x) >> return [x, succ x]) >>> sinkPrint +-- "1" +-- 1 +-- 2 +processN :: Transport t => (a -> IO [b]) -> Churro t a b +processN f = + buildChurro \i o -> do + yankAll i \x -> do mapM_ (yeet o . Just) =<< f x + yeet o Nothing + +-- | Extract xs from (Just x)s. Similar to `catMaybes`. +-- +-- >>> runWaitChan $ sourceList [Just 1, Nothing, Just 3] >>> justs >>> sinkPrint +-- 1 +-- 3 +justs :: Transport t => Churro t (Maybe a) a +justs = mapN (maybe [] pure) + +-- | Extract ls from (Left l)s. +-- +-- >>> runWaitChan $ sourceList [Left 1, Right 2, Left 3] >>> lefts >>> sinkPrint +-- 1 +-- 3 +lefts :: Transport t => Churro t (Either a b) a +lefts = mapN (either pure (const [])) + +-- | Extract rs from (Right r)s. +-- +-- >>> runWaitChan $ sourceList [Left 1, Right 2, Left 3] >>> rights >>> sinkPrint +-- 2 +rights :: Transport t => Churro t (Either a b) b +rights = mapN (either (const []) pure) + +-- | Take and yield the first n items. +-- +-- WARNING: This is intended to terminate upstream once the items have been consumed +-- downstream, but there is a bug preventing this from working at present! +-- +-- >>> runWaitChan $ sourceList [1..100] >>> takeC 2 >>> sinkPrint +-- 1 +-- 2 +-- +-- This implementation explicitly stops propagating when the Churro completes, +-- although this could be handled by downstream consumer composition terminating +-- the producer and just using replicateM. +takeC :: (Transport t, Integral n) => n -> Churro t a a +takeC n = buildChurro \i o -> go n i o + where + go t i o + | t <= 0 = yeet o Nothing + | otherwise = do + x <- yank i + yeet o x + when (isJust x) do go (pred t) i o + +-- | Drop the first n items. +-- +-- >>> runWaitChan $ sourceList [1..4] >>> dropC 2 >>> sinkPrint +-- 3 +-- 4 +dropC :: (Transport t, Integral n) => n -> Churro t a a +dropC n = buildChurro \i o -> do + replicateM_ (fromIntegral n) (yank i) -- TODO: Check the async behaviour of this... + c2c id i o + +-- | Filter items according to a predicate. +-- +-- >>> runWaitChan $ sourceList [1..5] >>> filterC (> 3) >>> sinkPrint +-- 4 +-- 5 +filterC :: Transport t => (a -> Bool) -> Churro t a a +filterC p = mapN (filter p . pure) + +-- | Run a pure function over items, producing multiple outputs. +-- +-- >>> runWaitChan $ pure 9 >>> mapN (\x -> [x,x*10]) >>> sinkPrint +-- 9 +-- 90 +mapN :: Transport t => (a -> [b]) -> Churro t a b +mapN f = processN (return . f) + +-- | Delay items from being sent downstream. +-- +-- Note: NominalDiffTime's Num instance interprets literals as seconds. +-- +-- >>> let sinkTimeCheck = process (const getCurrentTime) >>> withPrevious >>> arr (\(x,y) -> diffUTCTime y x > 0.01) >>> sinkPrint +-- +-- >>> runWaitChan $ sourceList [1..2] >>> sinkTimeCheck +-- False +-- +-- >>> runWaitChan $ sourceList [1..2] >>> delay 0.1 >>> sinkTimeCheck +-- True +delay :: Transport t => NominalDiffTime -> Churro t a a +delay = delayMicro . ceiling @Double . fromRational . (*1000000) . toRational + +-- | Delay items in microseconds. Works the same way as `delay`. +delayMicro :: Transport t => Int -> Churro t a a +delayMicro d = process \x -> do + threadDelay d + return x + +-- | Passes consecutive pairs of items downstream. +-- +-- >>> runWaitChan $ sourceList [1,2,3] >>> withPrevious >>> sinkPrint +-- (1,2) +-- (2,3) +withPrevious :: Transport t => Churro t a (a,a) +withPrevious = buildChurro \i o -> do + prog Nothing i o + yeet o Nothing + where + prog x i o = do + y <- yank i + case (x,y) of + (Just x', Just y') -> yeet o (Just (x',y')) >> prog y i o + (Nothing, Just y') -> prog (Just y') i o + _ -> return () + +-- | Requeue an item if it fails. Swallows exceptions and gives up after retries. +-- +-- Note: Process will always try once so if retries = 1 then a failing process will execute twice. +-- +-- The item is requeues on the input side of the churro, so if other items have +-- been passed in they will appear first! +-- +-- Catches all `SomeException`s. If you wish to narrow the execption type, consider +-- using the processRetry' variant composed with `rights`. +-- +-- Note: There is an edgecase with Chan transport where a queued retry may not execute +-- if a source completes and finalises before the item is requeued. +-- A different transport type may allow a modified retry function that requeues differently. +-- +-- >>> :{ +-- let +-- prog = processRetry 1 flakeyThing +-- flakeyThing x = do +-- if x > 1 +-- then print "GT" >> return x +-- else print "LTE" >> error ("oops! " <> show x) +-- in +-- runWaitChan $ sourceList [1,2] >>> delay 0.1 >>> prog >>> sinkPrint +-- :} +-- "LTE" +-- "LTE" +-- "GT" +-- 2 +-- +processRetry :: Transport t => Natural -> (i -> IO o) -> Churro t i o +processRetry retries f = processRetry' @SomeException retries f >>> rights + +-- | Raw version of `processRetry`. -- Polymorphic over exception type and forwards errors. +-- +processRetry' :: (Exception e, Transport t) => Natural -> (i -> IO o) -> Churro t i (Either e o) +processRetry' retries f = arr (0,) >>> processRetry'' retries f + +-- | Rawest version of `processRetry`. +-- Expects the incoming items to contain number of retries. +-- +-- Also polymorphic over exception type. And forwards errors. +-- +processRetry'' :: (Exception e, Transport t) => Natural -> (a -> IO o) -> Churro t (Natural, a) (Either e o) +processRetry'' retries f = + buildChurro \i o -> do + yankAll i \(n, y) -> do + r <- try do f y + yeet o (Just r) + case r of + Right _ -> return () + Left _ -> when (n < retries) do yeet i (Just (succ n, y)) + yeet o Nothing diff --git a/src/Control/Churro/Transport.hs b/src/Control/Churro/Transport.hs new file mode 100644 --- /dev/null +++ b/src/Control/Churro/Transport.hs @@ -0,0 +1,33 @@ +{-# LANGUAGE BlockArguments #-} + +-- | Re-exporting Transport instances. +-- +-- Also includes convenience functions for working directly with transports. +-- +module Control.Churro.Transport + + ( module Control.Churro.Transport.Chan + , module Control.Churro.Transport + ) + +where + +import Control.Churro.Transport.Chan +import Control.Churro.Types + +-- | Write a list to a raw Transport. +-- +-- >>> import Control.Concurrent.Chan +-- >>> :{ +-- do +-- c <- flex :: IO (Chan (Maybe (Maybe Int))) +-- l2c c (map Just [1,2] ++ [Nothing]) +-- yankAll' c print +-- :} +-- Just (Just 1) +-- Just (Just 2) +-- Just Nothing +-- Nothing +-- +l2c :: Transport t => t (Maybe a) -> [a] -> IO () +l2c c l = mapM_ (yeet c . Just) l >> yeet c Nothing diff --git a/src/Control/Churro/Transport/Chan.hs b/src/Control/Churro/Transport/Chan.hs new file mode 100644 --- /dev/null +++ b/src/Control/Churro/Transport/Chan.hs @@ -0,0 +1,29 @@ +{-# OPTIONS_GHC -Wno-orphans #-} +{-# LANGUAGE BlockArguments #-} + +-- | Chan Transport Instance. + +module Control.Churro.Transport.Chan where + +import Control.Churro.Types +import Control.Churro.Prelude + +import Control.Concurrent +import Data.Void + +instance Transport Chan where + flex = newChan + yank = readChan + yeet = writeChan + +type ChurroChan = Churro Chan + +-- | Convenience function for running a Churro with a Chan Transport. +-- +runWaitChan :: ChurroChan Void Void -> IO () +runWaitChan = runWait + +-- | Convenience function for running a Churro into a List with a Chan Transport. +-- +runWaitListChan :: ChurroChan Void o -> IO [o] +runWaitListChan = runWaitList diff --git a/src/Control/Churro/Types.hs b/src/Control/Churro/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Churro/Types.hs @@ -0,0 +1,236 @@ +{-# LANGUAGE InstanceSigs #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE BlockArguments #-} + +-- | Datatypes and definitions used by Churro library. +-- +-- Expand instances for additional documentation! + +module Control.Churro.Types where + +import Prelude hiding (id, (.)) +import Control.Arrow +import Control.Category +import Control.Concurrent.Async (cancel, wait, Async, async) +import Data.Void +import Control.Exception (finally) + +-- $setup +-- +-- We import the library for testing, although this would be a circular import in the module itself. +-- +-- >>> import Control.Churro + +-- ** Data, Classes and Instances + +-- | The core datatype for the library. +-- Parameters`t`, `i` and `o` represent the transport, +-- input and output types respectively. +-- +-- When building a program by composing Churros, the output Transport of one Churro is fed into the input Transports of other Churros. +-- +data Churro t i o = Churro { runChurro :: IO (t (Maybe i), t (Maybe o), Async ()) } + +-- | The transport method is abstracted via the Transport class +-- +-- This allows use of pure or impure channels, such as: +-- +-- * Chan (Included in `Control.Churro.Transport.Chan`) +-- * TChan +-- * Seq +-- * Various buffered options +-- +-- Transports used in conjunction with Churros wrap items in Maybe so that once a source has been depleted it can signal completion with +-- a Nothing item. +-- +class Transport t where + flex :: IO (t a) -- ^ Create a new Transport + yank :: t a -> IO a -- ^ Yank an item of the Transport + yeet :: t a -> a -> IO () -- ^ Yeet an item onto the Transport + +-- | Covariant functor instance for Churro - Maps over the output. +-- +-- >>> let s = sourceList [1,2] +-- >>> runWaitChan $ s >>> sinkPrint +-- 1 +-- 2 +-- +-- >>> runWaitChan $ fmap succ s >>> sinkPrint +-- 2 +-- 3 +instance Transport t => Functor (Churro t i) where + fmap f c = Churro do + (i,o,a) <- runChurro c + o' <- flex + a' <- async do + finally' (cancel a) do + c2c f o o' + wait a + return (i,o',a') + +-- | The Category instance allows for the creation of Churro pipelines. +-- +-- All other examples of the form `a >>> b` use this instance. +-- +-- The `id` method creates a passthrough arrow. +-- There isn't usually a reason to use `id` directly as it has no effect: +-- +-- >>> runWaitChan $ pure 1 >>> id >>> id >>> id >>> sinkPrint +-- 1 +instance Transport t => Category (Churro t) where + id = Churro do + a <- async (return ()) + c <- flex + return (c,c,a) + + g . f = Churro do + (fi, fo, fa) <- runChurro f + (gi, go, ga) <- runChurro g + a <- async do c2c id fo gi + b <- async do + finally' (cancel a >> cancel fa >> cancel ga) do + wait ga + cancel fa + cancel a + return (fi, go, b) + +-- | The Applicative instance allows for pairwise composition of Churro pipelines. +-- Once again this is covariat and the composition occurs on the output transports of the Churros. +-- +-- The `pure` method allows for the creation of a Churro yielding a single item. +-- +instance Transport t => Applicative (Churro t Void) where + pure x = buildChurro \_i o -> yeet o (Just x) >> yeet o Nothing + + f <*> g = buildChurro \_i o -> do + (_fi, fo, fa) <- runChurro f + (_gi, go, ga) <- runChurro g + + let + prog :: IO () + prog = do + fx <- yank fo + gx <- yank go + case (fx, gx) of + (Just f', Just g') -> (yeet o $ Just (f' g')) >> prog + _ -> return () + + prog + yeet o Nothing + wait fa + wait ga + +-- | The Arrow instance allows for building non-cyclic directed graphs of churros. +-- +-- The `arr` method allows for the creation of a that maps items with a pure function. +-- This is equivalent to `fmap f id`. +-- +-- >>> :set -XArrows +-- >>> :{ +-- let sect = process $ \x@(_x,_y,z) -> print x >> return z +-- graph = +-- proc i -> do +-- j <- arr succ -< i +-- k <- arr show -< j +-- l <- arr succ -< j +-- m <- arr (> 5) -< j +-- n <- sect -< (k,l,m) +-- o <- arr not -< n +-- sinkPrint -< o +-- in +-- runWaitChan $ sourceList [1,5,30] >>> graph +-- :} +-- ("2",3,False) +-- ("6",7,True) +-- ("31",32,True) +-- True +-- False +-- False +-- +-- The other Arrow methods are also usable: +-- +-- >>> runWaitChan $ pure 1 >>> (arr show &&& arr succ) >>> sinkPrint +-- ("1",2) +instance Transport t => Arrow (Churro t) where + arr = flip fmap id + + first c = Churro do + (i,o,a) <- runChurro c + i' <- flex + o' <- flex + + let go = do + is <- yank i' + yeet i (fmap fst is) + + os <- yank o + yeet o' $ (,) <$> os <*> fmap snd is + + case (is, os) of + (Just _, Just _) -> go + _ -> return () + + a' <- async do + go + yeet o' Nothing + wait a + + return (i',o',a') + +-- ** Helpers + +-- | A helper to facilitate constructing a Churro that makes new input and output transports available for manipulation. +-- +-- The manipulations performed are carried out in the async action associated with the Churro +-- +buildChurro :: Transport t => (t (Maybe i) -> t (Maybe o) -> IO ()) -> Churro t i o +buildChurro cb = Churro do + i <- flex + o <- flex + a <- async do cb i o + return (i,o,a) + +-- | Yeet all items from a list into a transport. +-- +yeetList :: (Foldable t1, Transport t2) => t2 a -> t1 a -> IO () +yeetList t = mapM_ (yeet t) + +-- | Yank all items from a Raw transport into a list. +-- +-- Won't terminate until the transport has been consumed. +-- +yankList :: Transport t => t (Maybe a) -> IO [a] +yankList t = do + x <- yank t + case x of + Nothing -> return [] + Just y -> (y :) <$> yankList t + +-- | Yank each item from a transport into a callback. +-- +yankAll :: Transport t => t (Maybe i) -> (i -> IO a) -> IO () +yankAll c f = do + x <- yank c + case x of + Nothing -> return () + Just y -> f y >> yankAll c f + +-- | 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. +-- +yankAll' :: Transport t => t (Maybe a) -> (Maybe a -> IO b) -> IO b +yankAll' c f = do + yankAll c (f . Just) + f Nothing + +-- | Yank then Yeet each item from one Transport into another. +-- +-- Raw items are used so `Nothing` should be Yeeted once the transport is depleted. +-- +c2c :: Transport t => (a1 -> a2) -> t (Maybe a1) -> t (Maybe a2) -> IO () +c2c f i o = yankAll' i (yeet o . fmap f) + +-- | Flipped `finally`. +finally' :: IO b -> IO a -> IO a +finally' = flip finally diff --git a/test/Churro/Test/Examples.hs b/test/Churro/Test/Examples.hs new file mode 100644 --- /dev/null +++ b/test/Churro/Test/Examples.hs @@ -0,0 +1,100 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE NoMonomorphismRestriction #-} +{-# LANGUAGE Arrows #-} + +-- | A set of examples of more complicated and problematic Churros. +-- +module Churro.Test.Examples where + +import Data.Map (fromList) +import Data.List + +import Prelude hiding (id, (.)) + +import Control.Churro + +-- $setup +-- +-- >>> import System.Timeout (timeout) +-- >>> import Control.Monad (forever) +-- >>> import Control.Concurrent (threadDelay) + +-- ** Tests + +-- | Checks that the IO nature of the churros doesn't duplicate operations. +-- Actions within a pipeline should only occur once no matter how the +-- pipeline is composed. +-- +-- >>> runWaitChan linear +-- Debugging [l1]: 1 +-- Debugging [l2]: 1 +-- Debugging [r1]: 1 +-- Debugging [r2]: 1 +-- 1 +-- +linear :: Transport t => Churro t Void Void +linear = sourceList [1::Int] + >>> ((processDebug "l1" >>> processDebug "l2") >>> processDebug "r1" >>> processDebug "r2") + >>> sinkPrint + +-- | A more complicated pipeline exampe involving maps. +-- +-- >>> runWaitChan pipeline +-- (fromList [(0,0),(1,1)],fromList [(1,1),(2,2)]) +-- (fromList [(1,1),(2,2)],fromList [(2,2),(3,3)]) +pipeline :: ChurroChan Void Void +pipeline = sourceList (take 3 maps) + >>> withPrevious + >>> takeC (10 :: Int) + >>> sinkPrint + where + maps = map fromList $ zipWith zip updates updates + updates = map (take 2) (tails [0 :: Int ..]) + +-- | Consumers terminaiting should kill sources from producing. +-- +-- This can fail in the following scenarios if cancellation isn't implemented correctly: +-- +-- >>> timeout 1500000 $ runWaitChan $ sourceList [1..5] >>> delay 1 >>> takeC 1 >>> sinkPrint +-- 1 +-- Just () +-- +-- Cancells infinite producer with inbuilt delays: +-- +-- >>> timeout 1500000 $ runWaitChan $ sourceIO (\cb -> forever (cb 1 >> threadDelay 100000)) >>> takeC 1 >>> sinkPrint +-- 1 +-- Just () +-- +-- Cancells upstream infinite producer with no inbuilt delay: +-- +-- >>> timeout 2500000 $ runWaitChan $ sourceList [1..] >>> delay 1 >>> takeC 1 >>> sinkPrint +-- 1 +-- Just () +-- +-- Note that the timeout here has to be sufficient for a thread switch to occor and the +-- action be cancelled! See what happens if the timeout is only 1.5s: +-- +-- >>> timeout 1500000 $ runWaitChan $ sourceList [1..] >>> delay 1 >>> takeC 1 >>> sinkPrint +-- 1 +-- Nothing +-- +-- What should happen is that the Category instance composition of: +-- +-- ... delay 1 >>> takeC 1 ... +-- ^^^ PRE ^^^^ ^^^ POST ^^^ +-- +-- When POST terminates it should cancel the computation in PRE. +-- +-- Failures may be caused by one of the following: +-- +-- * Nested Async actions don't cascade on cancellation (incorrect finally clauses) +-- * The associativity laws of Category are broken, meaning that cancellation doesn't behave as it should. +-- * Producer blocks cancellation from being requested +-- * Chan is blocked preventing indicating termination to consumers +-- * Infinite source is causing issues (ruled out with this test example) +-- +-- This is currently working, but tests here should check that the implementation +-- isn't broken. diff --git a/test/Churro/Test/Main.hs b/test/Churro/Test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Churro/Test/Main.hs @@ -0,0 +1,14 @@ + +module Main where + +import Test.DocTest + +-- import qualified Churro.Test.Examples as Ex +-- import qualified Build_doctests as DT + +-- TODO: Find out how to use a different test manager to allow these to be run independently +main :: IO () +main = do + doctest ["-isrc", "test/Churro/Test/Examples.hs"] + -- Ex.main +