diff --git a/dynamic-pipeline.cabal b/dynamic-pipeline.cabal --- a/dynamic-pipeline.cabal +++ b/dynamic-pipeline.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: dynamic-pipeline -version: 0.2.0.0 +version: 0.3.0.0 synopsis: Library Type Safe implementation of Dynamic Pipeline Paradigm (DPP). description: @dynamic-pipeline@ is a __/Type Safe/__ Dynamic and Parallel Streaming Library, which is an implementation of __Dynamic Pipeline Paradigm (DPP)__ proposed in this paper [DPP](https://biblioteca.sistedes.es/articulo/the-dynamic-pipeline-paradigm/). @@ -110,6 +110,7 @@ Graph.ConnComp Graph.ConnectedComp Misc.RepeatedDP + Misc.RepeatedTwiceDP hs-source-dirs: examples default-extensions: diff --git a/examples/Graph/ConnectedComp.hs b/examples/Graph/ConnectedComp.hs --- a/examples/Graph/ConnectedComp.hs +++ b/examples/Graph/ConnectedComp.hs @@ -14,8 +14,8 @@ -- brittany-disable-next-binding type DPConnComp = Source (Channel (Edge :<+> ConnectedComponents :<+> Eof)) - :>> Generator (Channel (Edge :<+> ConnectedComponents :<+> Eof)) - :>> Sink + :=> Generator (Channel (Edge :<+> ConnectedComponents :<+> Eof)) + :=> Sink source' :: FilePath -> Stage @@ -24,7 +24,7 @@ $ \edgeOut _ -> unfoldFile filePath edgeOut (toEdge . decodeUtf8) sink' :: Stage (ReadChannel Edge -> ReadChannel ConnectedComponents -> DP st ()) -sink' = withSink @DPConnComp $ \_ cc -> withDP $ foldM cc print +sink' = withSink @DPConnComp $ \_ cc -> withDP $ foldM_ cc print generator' :: GeneratorStage DPConnComp ConnectedComponents Edge st generator' = @@ -41,7 +41,7 @@ -> WriteChannel ConnectedComponents -> StateT ConnectedComponents (DP st) () actor1 _ readEdge _ writeEdge _ = - foldM readEdge $ \e -> get >>= doActor e + foldM_ readEdge $ \e -> get >>= doActor e where doActor v conn | toConnectedComp v `intersect` conn = modify (toConnectedComp v <>) @@ -54,7 +54,7 @@ -> WriteChannel ConnectedComponents -> StateT ConnectedComponents (DP st) () actor2 _ _ readCC _ writeCC = do - foldM' readCC pushMemory $ \e -> get >>= doActor e + foldWithM_ readCC pushMemory $ \e -> get >>= doActor e where pushMemory = get >>= flip push writeCC @@ -73,7 +73,7 @@ genAction filter' readEdge readCC _ writeCC = do let unfoldFilter = mkUnfoldFilterForAll filter' toConnectedComp readEdge (readCC .*. HNil) results <- unfoldF unfoldFilter - foldM (hHead results) (`push` writeCC) + foldM_ (hHead results) (`push` writeCC) program :: FilePath -> IO () diff --git a/examples/Main.hs b/examples/Main.hs --- a/examples/Main.hs +++ b/examples/Main.hs @@ -2,11 +2,13 @@ import Graph.ConnectedComp as CC import Misc.RepeatedDP as Repeated +import Misc.RepeatedTwiceDP as RepeatedTwice import Options.Applicative as Opt import Relude data ProgramOptions = RepeatedElements + | RepeatedFeedback | ConnectedComponents FilePath deriving(Show) @@ -14,8 +16,11 @@ programDesc :: ParserInfo ProgramOptions programDesc = info (everyProgram <**> helper) (fullDesc <> header "Examples on dynamic-pipeline library") where - everyProgram = - subparser (command "repeated-elements" subCmdRepeated <> command "connected-components" subCmdConnectedComp) + everyProgram = subparser + ( command "repeated-elements" subCmdRepeated + <> command "repeated-feedback" subCmdRepeatedFeed + <> command "connected-components" subCmdConnectedComp + ) subCmdRepeated :: ParserInfo ProgramOptions subCmdRepeated = info @@ -24,6 +29,12 @@ "Given a list of 2000 repeated Integers filter and output unique 1000 integers" ) +subCmdRepeatedFeed :: ParserInfo ProgramOptions +subCmdRepeatedFeed = info + (pure RepeatedFeedback) + (fullDesc <> header "repeated-feedback - Dynamic Pipeline Examples" <> progDesc + "Given a list of 2000 repeated Integers filter and output unique 1000 integers and do it twice with feedback" + ) subCmdConnectedComp :: ParserInfo ProgramOptions subCmdConnectedComp = info @@ -45,6 +56,7 @@ main' :: ProgramOptions -> IO () main' = \case - RepeatedElements -> Repeated.program - ConnectedComponents file -> CC.program file + RepeatedElements -> Repeated.program + RepeatedFeedback -> RepeatedTwice.program + ConnectedComponents file -> CC.program file diff --git a/examples/Misc/RepeatedDP.hs b/examples/Misc/RepeatedDP.hs --- a/examples/Misc/RepeatedDP.hs +++ b/examples/Misc/RepeatedDP.hs @@ -11,7 +11,7 @@ import DynamicPipeline import Relude -type DPExample = Source (Channel (Int :<+> Eof)) :>> Generator (Channel (Int :<+> Eof)) :>> Sink +type DPExample = Source (Channel (Int :<+> Eof)) :=> Generator (Channel (Int :<+> Eof)) :=> Sink source' :: Stage (WriteChannel Int -> DP s ()) source' = withSource @DPExample $ \cout -> unfoldT ([1 .. 1000] <> [1 .. 1000]) cout identity @@ -36,10 +36,10 @@ -> ReadChannel Int -> WriteChannel Int -> StateT (Maybe Int) (DP s) () -actorRepeted i rc wc = foldM rc $ \e -> if e /= i then push e wc else pure () +actorRepeted i rc wc = foldM_ rc $ \e -> if e /= i then push e wc else pure () sink' :: Stage (ReadChannel Int -> DP s ()) -sink' = withSink @DPExample $ flip foldM print +sink' = withSink @DPExample $ flip foldM_ print program :: IO () program = runDP $ mkDP @DPExample source' generator' sink' diff --git a/examples/Misc/RepeatedTwiceDP.hs b/examples/Misc/RepeatedTwiceDP.hs new file mode 100644 --- /dev/null +++ b/examples/Misc/RepeatedTwiceDP.hs @@ -0,0 +1,84 @@ +-- | +-- Module : Misc.RepeatedTwiceDP +-- Copyright : (c) 2021 Juan Pablo Royo Sales +-- +-- License : BSD3 +-- Maintainer : juanpablo.royo@gmail.com +-- Stability : experimental +-- Portability : GHC +module Misc.RepeatedTwiceDP where + +import DynamicPipeline +import Relude + +type DPExample = Source (Channel (Int :<+> Int :<+> Double :<+> String :<+> Eof)) + :=> Generator (Channel (Int :<+> Int :<+> Double :<+> Eof)) + :=> FeedbackChannel (String :<+> Eof) + :=> Sink + +source' :: Stage (ReadChannel String -> WriteChannel Int -> WriteChannel Int -> WriteChannel Double -> WriteChannel String -> DP st ()) +source' = withSource @DPExample $ \feedback cout cout' cout'' toFilter -> do + unfoldT ([1 .. 10] <> [1 .. 10]) cout identity + finish cout >> finish cout' >> finish cout'' + foldM_ feedback (`push` toFilter) + +generator' :: GeneratorStage DPExample (Maybe Int) Int s +generator' = + let gen = withGenerator @DPExample genAction + in mkGenerator gen filterTemp + +genAction :: Filter DPExample (Maybe Int) Int st + -> ReadChannel Int + -> ReadChannel Int + -> ReadChannel Double + -> ReadChannel String + -> WriteChannel Int + -> WriteChannel Int + -> WriteChannel Double + -> WriteChannel String + -> DP st () +genAction filter' cin cin' cdn cin'' _ _ odn cout = do + let unfoldFilter = mkUnfoldFilterForAll filter' Just cin (cin' .*. cdn .*. cin'' .*. HNil) + HCons ft (HCons sec _) <- unfoldF unfoldFilter + foldM_ ft $ flip push cout . show + finish cout + foldM_ sec $ flip push odn + +filterTemp :: Filter DPExample (Maybe Int) Int s +filterTemp = mkFilter actorRepeted + +actorRepeted :: Int + -> ReadChannel Int + -> ReadChannel Int + -> ReadChannel Double + -> ReadChannel String + -> WriteChannel Int + -> WriteChannel Int + -> WriteChannel Double + -> WriteChannel String + -> StateT (Maybe Int) (DP s) () +actorRepeted i rc rc' rd rs wc wc' wd wc'' = do + foldM_ rc $ \e -> do + putTextLn $ "1) Elem: " <> show e <> " - Param: " <> show i + if e /= i then push e wc else pure () + finish wc + push i wc' + foldM_ rc' $ \e -> do + putTextLn $ "2) Elem: " <> show e <> " - Param: " <> show i + push e wc' + finish wc' + foldM_ rs $ \e -> + let x = maybe 0 identity $ readMaybe @Int e + in if i == x + then push (fromIntegral x) wd + else push e wc'' + finish wc'' + foldM_ rd $ flip push wd + finish wd + + +sink' :: Stage (ReadChannel Int -> ReadChannel Int -> ReadChannel Double -> DP s ()) +sink' = withSink @DPExample $ \_ _ ci -> foldM_ ci print + +program :: IO () +program = runDP $ mkDP @DPExample source' generator' sink' diff --git a/src/DynamicPipeline.hs b/src/DynamicPipeline.hs --- a/src/DynamicPipeline.hs +++ b/src/DynamicPipeline.hs @@ -22,7 +22,7 @@ -- @ -- import "DynamicPipeline" -- --- type DPExample = 'Source' ('Channel' (Int ':<+>' 'Eof')) ':>>' 'Generator' ('Channel' (Int ':<+>' 'Eof')) ':>>' 'Sink' +-- type DPExample = 'Source' ('Channel' (Int ':<+>' 'Eof')) ':=>' 'Generator' ('Channel' (Int ':<+>' 'Eof')) ':=>' 'Sink' -- -- source' :: 'Stage' ('WriteChannel' Int -> 'DP' s ()) -- source' = 'withSource' @DPExample $ \cout -> 'unfoldT' ([1 .. 1000] <> [1 .. 1000]) cout identity @@ -73,7 +73,8 @@ Generator, Source, Channel, - type (:>>)(..), + FeedbackChannel, + type (:=>)(..), type (:<+>)(..), -- * Smart Constructors DynamicPipeline, @@ -102,21 +103,22 @@ mkUnfoldFilter', mkUnfoldFilterForAll, mkUnfoldFilterForAll', - (.*.), HList(HNil), hHead, + (.*.), HList(HNil,HCons), hHead, hTail, -- * Channels ReadChannel, WriteChannel, - foldM, - foldM', + foldM_, + foldWithM_, push, pull, + finish, unfoldM, unfoldFile, unfoldT ) where -import Data.HList ((.*.), HList(HNil), hHead) +import Data.HList ((.*.), HList(HNil,HCons), hHead, hTail) import DynamicPipeline.Flow import DynamicPipeline.Channel import DynamicPipeline.Stage @@ -125,14 +127,14 @@ -- The following is the Regular Grammar allowed to build a /DPP/ Flow definition: -- -- @ --- __DP__ = 'Source' __CHANS__ ':>>' 'Generator' __CHANS__ ':>>' 'Sink' +-- __DP__ = 'Source' __CHANS__ ':=>' 'Generator' __CHANS__ ':=>' 'Sink' -- __CHANS__ = 'Channel' __CH__ -- __CH__ = 'Eof' | 'Type' ':<+>' __CH__ -- @ -- -- Example: -- --- @ 'Source' ('Channel' (Int ':<+>' Int)) ':>>' 'Generator' ('Channel' (Int ':<+>' Int)) ':>>' 'Sink' @ +-- @ 'Source' ('Channel' (Int ':<+>' Int)) ':=>' 'Generator' ('Channel' (Int ':<+>' Int)) ':=>' 'Sink' @ -- -- -- $dp @@ -149,7 +151,7 @@ -- -- >>> import Relude -- >>> import DynamicPipeline --- >>> type DPEx = Source (Channel (Int :<+> Eof)) :>> Generator (Channel (Int :<+> Eof)) :>> Sink +-- >>> type DPEx = Source (Channel (Int :<+> Eof)) :=> Generator (Channel (Int :<+> Eof)) :=> Sink -- >>> :t withSource @DPEx -- withSource @DPEx -- :: forall k (st :: k). diff --git a/src/DynamicPipeline/Channel.hs b/src/DynamicPipeline/Channel.hs --- a/src/DynamicPipeline/Channel.hs +++ b/src/DynamicPipeline/Channel.hs @@ -12,8 +12,8 @@ module DynamicPipeline.Channel ( ReadChannel , WriteChannel - , DynamicPipeline.Channel.foldM - , foldM' + , foldM_ + , foldWithM_ , push , pull , unfoldM @@ -21,6 +21,7 @@ , unfoldT , newChannel , end + , finish ) where import qualified Control.Concurrent as CC @@ -29,7 +30,8 @@ ) import Data.ByteString as B import Data.Foldable as F -import Data.HList +import Data.HList hiding ( foldM_ + ) import GHC.IO.Handle as H import Relude as R @@ -44,23 +46,24 @@ -- [@a@]: Type that this Channel can read newtype ReadChannel a = ReadChannel { unRead :: OutChan (Maybe a) } --- | 'foldM' is a /Catamorphism/ for consuming a 'ReadChannel' and do some Monadic @m@ computation with each element -{-# INLINE foldM #-} -foldM :: MonadIO m - => ReadChannel a -- ^'ReadChannel' - -> (a -> m ()) -- ^Computation to do with read element - -> m () -foldM = flip foldM' (pure ()) - --- | Idem 'foldM' but allows pass a monadic computation to perform at the end of the Channel -{-# INLINE foldM' #-} -foldM' :: MonadIO m +-- | 'foldM_' is a /Catamorphism/ for consuming a 'ReadChannel' and do some Monadic @m@ computation with each element +{-# INLINE foldM_ #-} +foldM_ :: MonadIO m => ReadChannel a -- ^'ReadChannel' - -> m () -- ^Computation to do at the end of the channel -> (a -> m ()) -- ^Computation to do with read element -> m () -foldM' = loop' where loop' c onNothing io = maybe onNothing (\e -> io e >> loop' c onNothing io) =<< liftIO (pull c) +foldM_ = flip foldWithM_ (pure ()) +-- | Idem 'foldM_' but allows pass a monadic computation to perform at the end of the Channel +{-# INLINE foldWithM_ #-} +foldWithM_ :: MonadIO m + => ReadChannel a -- ^'ReadChannel' + -> m () -- ^Computation to do at the end of the channel + -> (a -> m ()) -- ^Computation to do with read element + -> m () +foldWithM_ = loop' + where loop' c onNothing io = maybe onNothing (\e -> io e >> loop' c onNothing io) =<< liftIO (pull c) + -- | Push element @a@ into 'WriteChannel' {-# INLINE push #-} push :: MonadIO m => a -> WriteChannel a -> m () @@ -71,6 +74,11 @@ pull :: MonadIO m => ReadChannel a -> m (Maybe a) pull = liftIO . readChan (CC.threadDelay 100) . unRead +-- | Finalize Channel to indicate EOF mark and allow progress on following consumers +finish :: MonadIO m => WriteChannel a -> m () +finish = liftIO . end + + -- | Coalgebra with Monadic computation to Feed some 'WriteChannel' -- -- [@m@]: Monadic computation wrapping Coalgebra @@ -114,4 +122,5 @@ {-# INLINE end #-} end :: WriteChannel a -> IO () end = flip writeChan Nothing . unWrite + diff --git a/src/DynamicPipeline/Flow.hs b/src/DynamicPipeline/Flow.hs --- a/src/DynamicPipeline/Flow.hs +++ b/src/DynamicPipeline/Flow.hs @@ -15,32 +15,35 @@ Generator, Source, Channel, - type (:>>)(..), + FeedbackChannel, + type (:=>)(..), type (:<+>)(..), ChanIn, + ChanInIn, ChanOut, ChanOutIn, ChansFilter, ChanWriteSource, ChanReadWriteGen, ChanReadOut, + ChanRecord(..), + InOutChan(..), MkCh(..), MkChans(..), ExpandGenToCh, ExpandSinkToCh, ExpandSourceToCh, ExpandFilterToCh, - AllChans, - FilterChans, inGenOut, + toHList, makeChans, + makeChansF, getFilterChannels ) where import Control.Lens hiding ((<|)) import Data.Foldable as F import Data.HList -import Data.HList.Labelable import DynamicPipeline.Channel import Relude as R @@ -65,6 +68,12 @@ -- @ a ~ (Type ':<+>' Type ':<+>' ... ':<+>' Eof) @ data Channel (a :: Type) +-- |'FeedbackChannel' is the Container Type of /Open Union Type/ which is going to be defined with ':<+>' and indicates that this +-- | Channel is for feedback to Source +-- +-- @ a ~ (Type ':<+>' Type ':<+>' ... ':<+>' Eof) @ +data FeedbackChannel (a :: Type) + -- | This is the Type level function of the /Open Union Type/ for Channels. -- -- Channels forms an /Open Union Type/ in each stage because according to __DPP__ we can have multiple /In/ and /Out/ Channels @@ -83,10 +92,10 @@ -- -- This should have the form: -- --- @ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink' @ -data a :>> b = a :>> b +-- @ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink' @ +data a :=> b = a :=> b deriving (Typeable, Eq, Show, Functor, Traversable, Foldable, Bounded) -infixr 5 :>> +infixr 5 :=> -- Internal Data Types for expanding function based on Channel definitions {-# WARNING ChanIn "INTERNAL USE" #-} @@ -95,6 +104,8 @@ data ChanOut (a :: Type) {-# WARNING ChanOutIn "INTERNAL USE" #-} data ChanOutIn (a :: Type) (b :: Type) +{-# WARNING ChanInIn "INTERNAL USE" #-} +data ChanInIn (a :: Type) (b :: Type) {-# WARNING ChansFilter "INTERNAL USE" #-} data ChansFilter (a :: Type) {-# WARNING ChanWriteSource "INTERNAL USE" #-} @@ -107,7 +118,7 @@ -- Type encoding for Building Chans. Only for internal use in the Associated Type Family and combinators of MkCh and MkChans -- For accessing Dynamic Indexed Records of Channels {-# WARNING inLabel "INTERNAL USE" #-} -inLabel :: Label "Source" +inLabel :: Label "source" inLabel = Label {-# WARNING genLabel "INTERNAL USE" #-} @@ -115,7 +126,7 @@ genLabel = Label {-# WARNING outLabel "INTERNAL USE" #-} -outLabel :: Label "Sink" +outLabel :: Label "sink" outLabel = Label {-# WARNING inChLabel "INTERNAL USE" #-} @@ -134,10 +145,10 @@ mkCh :: Proxy a -> IO (HList (HChI a), HList (HChO a)) instance MkCh more => MkCh (a :<+> more) where - type HChI (a :<+> more) = WriteChannel a ': HChI more - type HChO (a :<+> more) = ReadChannel a ': HChO more + type HChI (a :<+> more) = ReadChannel a ': HChI more + type HChO (a :<+> more) = WriteChannel a ': HChO more mkCh _ = do - (i, o) <- newChannel @a + (o, i) <- newChannel @a (il, ol) <- mkCh (Proxy @more) return (i .*. il, o .*. ol) @@ -150,19 +161,34 @@ {-# WARNING ExpandToHList "INTERNAL USE" #-} type family ExpandToHList (a :: Type) (param :: Type) :: [Type] type instance ExpandToHList (ChanWriteSource ( Source (Channel inToGen) - :>> Generator (Channel genToOut) - :>> Sink ) - ) _ = HChI inToGen + :=> Generator (Channel genToOut) + :=> Sink ) + ) _ = HChO inToGen +type instance ExpandToHList (ChanWriteSource ( Source (Channel inToGen) + :=> Generator (Channel genToOut) + :=> FeedbackChannel toSource + :=> Sink ) + ) _ = HAppendListR (HChI toSource) (HChO inToGen) type instance ExpandToHList (ChanReadWriteGen ( Source (Channel inToGen) - :>> Generator (Channel genToOut) - :>> Sink) - ) filter = filter ': HAppendListR (HChO inToGen) (HChI genToOut) + :=> Generator (Channel genToOut) + :=> Sink) + ) filter = filter ': HAppendListR (HChI inToGen) (HChO genToOut) +type instance ExpandToHList (ChanReadWriteGen ( Source (Channel inToGen) + :=> Generator (Channel genToOut) + :=> FeedbackChannel toSource + :=> Sink) + ) filter = filter ': HAppendListR (HAppendListR (HChI inToGen) (HChO genToOut)) (HChO toSource) type instance ExpandToHList (ChanReadOut ( Source (Channel inToGen) - :>> Generator (Channel genToOut) - :>> Sink ) - ) filter = HChO genToOut + :=> Generator (Channel genToOut) + :=> Sink ) + ) filter = HChI genToOut +type instance ExpandToHList (ChanReadOut ( Source (Channel inToGen) + :=> Generator (Channel genToOut) + :=> FeedbackChannel toSource + :=> Sink ) + ) filter = HChI genToOut {-# WARNING ExpandSourceToCh "INTERNAL USE" #-} type ExpandSourceToCh a = ExpandToHList (ChanWriteSource a) Void @@ -173,164 +199,117 @@ {-# WARNING ExpandSinkToCh "INTERNAL USE" #-} type ExpandSinkToCh a = ExpandToHList (ChanReadOut a) Void +data InOutChan lr lw = InOutChan + { _iocReadChans :: HList lr + , _iocWriteChans :: HList lw + } + +toHList :: HAppendList lr lw => InOutChan lr lw -> HList (HAppendListR lr lw) +toHList InOutChan{..} = _iocReadChans `hAppend` _iocWriteChans + +data ChanRecord slr slw glr glw silr silw = ChanRecord + { _crSource :: InOutChan slr slw + , _crGenerator :: InOutChan glr glw + , _crSink :: InOutChan silr silw + } + -- Class for building Channels base on a DP Definition on `a` Type {-# WARNING MkChans "INTERNAL USE" #-} class MkChans (a :: Type) where type HChan a :: Type mkChans :: Proxy a -> IO (HChan a) --- Instance for Building Channels for all the Chain Source :>> Generator :>> Sink +-- Instance for Building Channels for all the Chain Source :=> Generator :=> Sink instance ( MkCh inToGen , MkCh genToOut) - => MkChans (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) where + => MkChans (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) where - type HChan (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) - = Record '[ Tagged "Source" (Record '[ Tagged "in-ch" (HList (HChI inToGen)) - , Tagged "out-ch" (HList (HChO inToGen)) - ] - ) - , Tagged "generator" (Record '[ Tagged "in-ch" (HList (HChI genToOut)) - , Tagged "out-ch" (HList (HChO genToOut)) - ] - ) - , Tagged "Sink" (Record '[ Tagged "in-ch" (HList (HChI genToOut))]) - ] + type HChan (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) + = ChanRecord '[] (HChO inToGen) (HChI inToGen) (HChO genToOut) (HChI genToOut) '[] mkChans _ = do - (ii, io) <- mkCh (Proxy @inToGen) - (gi, go) <- mkCh (Proxy @genToOut) - (oi, _) <- mkCh (Proxy @genToOut) - return $ (inLabel .=. (inChLabel .=. ii .*. outChLabel .=. io .*. emptyRecord)) - .*. - (genLabel .=. (inChLabel .=. gi .*. outChLabel .=. go .*. emptyRecord)) - .*. - (outLabel .=. (inChLabel .=. oi .*. emptyRecord)) - .*. - emptyRecord + (rs, ws) <- mkCh (Proxy @inToGen) + (rg, wg) <- mkCh (Proxy @genToOut) + return $ ChanRecord + { _crSource = InOutChan HNil ws + , _crGenerator = InOutChan rs wg + , _crSink = InOutChan rg HNil + } +instance ( MkCh inToGen + , MkCh genToOut + , MkCh toSource + , HAppendList (HChO genToOut) (HChO toSource)) + => MkChans (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) where + + type HChan (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) + = ChanRecord (HChI toSource) (HChO inToGen) + (HChI inToGen) (HAppendListR (HChO genToOut) (HChO toSource)) + (HChI genToOut) '[] + + mkChans _ = do + (rf, wf) <- mkCh (Proxy @toSource) + (rs, ws) <- mkCh (Proxy @inToGen) + (rg, wg) <- mkCh (Proxy @genToOut) + return $ ChanRecord + { _crSource = InOutChan rf ws + , _crGenerator = InOutChan rs (wg `hAppend` wf) + , _crSink = InOutChan rg HNil + } + -- Instance for Building Only Channels for Filters on each Generator action instance MkCh inToGen - => MkChans (ChansFilter (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink)) where + => MkChans (ChansFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink)) where - type HChan (ChansFilter (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink)) - = Record '[ Tagged "in-ch" (HList (HChO inToGen)) - , Tagged "out-ch" (HList (HChI inToGen)) - ] + type HChan (ChansFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink)) + = InOutChan (HChI inToGen) (HChO inToGen) - mkChans _ = do - (writes', reads') <- mkCh (Proxy @inToGen) - return $ mkRecord (inChLabel .=. reads' .*. outChLabel .=. writes' .*. HNil) + mkChans _ = uncurry InOutChan <$> mkCh (Proxy @inToGen) +instance ( MkCh inToGen + ) + => MkChans (ChansFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink)) where + type HChan (ChansFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink)) + = InOutChan (HChI inToGen) (HChO inToGen) + + mkChans _ = uncurry InOutChan <$> mkCh (Proxy @inToGen) + + {-# WARNING makeChans "INTERNAL USE" #-} {-# INLINE makeChans #-} -makeChans :: forall (a :: Type). MkChans a => IO (HChan a) +makeChans :: forall (a :: Type) slr slw glr glw silr silw. (MkChans a, HChan a ~ ChanRecord slr slw glr glw silr silw) => IO (ChanRecord slr slw glr glw silr silw) makeChans = mkChans (Proxy @a) +{-# WARNING makeChansF "INTERNAL USE" #-} +{-# INLINE makeChansF #-} +makeChansF :: forall (a :: Type) flr flw. (MkChans a, HChan a ~ InOutChan flr flw) => IO (InOutChan flr flw) +makeChansF = mkChans (Proxy @a) + -- Ugly Dynamic Indexed Record Viewer to generate specific list of channels {-# WARNING sourceChans "INTERNAL USE" #-} {-# INLINE sourceChans #-} -sourceChans :: ( LabeledOpticF (LabelableTy r1) (Const t1) - , LabeledOpticP (LabelableTy r1) (->) - , LabeledOpticTo (LabelableTy r1) "in-ch" (->) - , LabeledOpticF (LabelableTy r2) (Const t1) - , LabeledOpticP (LabelableTy r2) (->) - , LabeledOpticTo (LabelableTy r2) "Source" (->) - , Labelable "in-ch" r1 s t2 t1 t1 - , Labelable "Source" r2 t3 t3 (r1 s) (r1 t2)) - => r2 t3 -> t1 -sourceChans = let inl = hLens' inLabel - inch = hLens' inChLabel - in view (inl . inch) +sourceChans :: HAppendList slr slw => ChanRecord slr slw _ _ _ _ -> HList (HAppendListR slr slw) +sourceChans = toHList . _crSource {-# WARNING generatorChans "INTERNAL USE" #-} {-# INLINE generatorChans #-} -generatorChans :: ( LabeledOpticF (LabelableTy r1) (Const (HList l1)) - , LabeledOpticP (LabelableTy r1) (->) - , LabeledOpticTo (LabelableTy r1) "out-ch" (->) - , LabeledOpticF (LabelableTy r2) (Const (HList l2)) - , LabeledOpticP (LabelableTy r2) (->) - , LabeledOpticTo (LabelableTy r2) "in-ch" (->) - , LabeledOpticF (LabelableTy r3) (Const (HList l2)) - , LabeledOpticP (LabelableTy r3) (->) - , LabeledOpticTo (LabelableTy r3) "generator" (->) - , LabeledOpticF (LabelableTy r3) (Const (HList l1)) - , LabeledOpticTo (LabelableTy r3) "Source" (->) - , HAppendList l1 l2 - , Labelable "generator" r3 t1 t1 (r2 s1) (r2 t2) - , Labelable "in-ch" r2 s1 t2 (HList l2) (HList l2) - , Labelable "Source" r3 t1 t1 (r1 s2) (r1 t3) - , Labelable "out-ch" r1 s2 t3 (HList l1) (HList l1)) - => r3 t1 -> HList (HAppendListR l1 l2) -generatorChans ch = let inl = hLens' inLabel - genl = hLens' genLabel - inch = hLens' inChLabel - outch = hLens' outChLabel - outsIn = view (inl . outch) ch - insGen = view (genl . inch) ch - in outsIn `hAppendList` insGen +generatorChans :: HAppendList glr glw => ChanRecord _ _ glr glw _ _ -> HList (HAppendListR glr glw) +generatorChans = toHList . _crGenerator + {-# WARNING sinkChans "INTERNAL USE" #-} {-# INLINE sinkChans #-} -sinkChans :: ( LabeledOpticF (LabelableTy r1) (Const t1) - , LabeledOpticP (LabelableTy r1) (->) - , LabeledOpticTo (LabelableTy r1) "out-ch" (->) - , LabeledOpticF (LabelableTy r2) (Const t1) - , LabeledOpticP (LabelableTy r2) (->) - , LabeledOpticTo (LabelableTy r2) "generator" (->) - , Labelable "generator" r2 t2 t2 (r1 s) (r1 t3) - , Labelable "out-ch" r1 s t3 t1 t1) - => r2 t2 -> t1 -sinkChans = let genl = hLens' genLabel - outch = hLens' outChLabel - in view (genl . outch) - -{-# WARNING AllChans "INTERNAL USE" #-} -type AllChans r2 r3 l1 r4 l2 t2 s t1 s2 t5 l3 l4 = (LabeledOpticTo (LabelableTy r2) "in-ch" (->), - LabeledOpticF (LabelableTy r3) (Const (HList l3)), - LabeledOpticP (LabelableTy r3) (->), - LabeledOpticTo (LabelableTy r3) "Source" (->), - LabeledOpticF (LabelableTy r2) (Const (HList l1)), - LabeledOpticP (LabelableTy r2) (->), - LabeledOpticTo (LabelableTy r2) "out-ch" (->), - LabeledOpticTo (LabelableTy r4) "in-ch" (->), - LabeledOpticF (LabelableTy r3) (Const (HList l2)), - LabeledOpticTo (LabelableTy r3) "generator" (->), - LabeledOpticF (LabelableTy r3) (Const (HList l1)), - LabeledOpticF (LabelableTy r2) (Const (HList l3)), - LabeledOpticF (LabelableTy r4) (Const (HList l4)), - LabeledOpticP (LabelableTy r4) (->), - LabeledOpticTo (LabelableTy r4) "out-ch" (->), - LabeledOpticF (LabelableTy r3) (Const (HList l4)), - LabeledOpticF (LabelableTy r4) (Const (HList l2)), - HAppendList l1 l2, Labelable "generator" r3 t2 t2 (r4 s) (r4 t1), - Labelable "in-ch" r2 s2 t5 (HList l3) (HList l3), - Labelable "in-ch" r4 s t1 (HList l2) (HList l2), - Labelable "Source" r3 t2 t2 (r2 s2) (r2 t5), - Labelable "out-ch" r2 s2 t5 (HList l1) (HList l1), - Labelable "out-ch" r4 s t1 (HList l4) (HList l4)) +sinkChans :: HAppendList silr silw => ChanRecord _ _ _ _ silr silw -> HList (HAppendListR silr silw) +sinkChans = toHList . _crSink {-# WARNING inGenOut "INTERNAL USE" #-} {-# INLINE inGenOut #-} -inGenOut :: AllChans r2 r3 l1 r4 l2 t2 s t1 s2 t5 l3 l4 => r3 t2 -> (HList l3, HList (HAppendListR l1 l2), HList l4) +inGenOut :: (HAppendList slr slw, HAppendList glr glw, HAppendList silr silw) => ChanRecord slr slw glr glw silr silw -> (HList (HAppendListR slr slw), HList (HAppendListR glr glw), HList (HAppendListR silr silw)) inGenOut ch = (sourceChans ch, generatorChans ch, sinkChans ch) - -{-# WARNING FilterChans "INTERNAL USE" #-} -type FilterChans r b t a = (LabeledOpticF (LabelableTy r) (Const b), - LabeledOpticTo (LabelableTy r) "out-ch" (->), - Labelable "out-ch" r t t b b, - LabeledOpticF (LabelableTy r) (Const a), - LabeledOpticP (LabelableTy r) (->), - LabeledOpticTo (LabelableTy r) "in-ch" (->), - Labelable "in-ch" r t t a a) - {-# WARNING getFilterChannels "INTERNAL USE" #-} {-# INLINE getFilterChannels #-} -getFilterChannels :: FilterChans r b t a => r t -> (a, b) -getFilterChannels ch = - let inch = hLens' inChLabel - outch = hLens' outChLabel - reads' = ch^.inch - writes' = ch^.outch - in (reads', writes') +getFilterChannels :: InOutChan lr lw -> (HList lr, HList lw) +getFilterChannels InOutChan{..} = (_iocReadChans, _iocWriteChans) + diff --git a/src/DynamicPipeline/Stage.hs b/src/DynamicPipeline/Stage.hs --- a/src/DynamicPipeline/Stage.hs +++ b/src/DynamicPipeline/Stage.hs @@ -56,15 +56,21 @@ -- | FCF - Type Level Defunctionalization -- 'IsDP' Validates if /DP/ Flow Type Level Definition is Correct according to the Grammar -- --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- type family IsDP (dpDefinition :: k) :: Bool where IsDP (Source (Channel inToGen) - :>> Generator (Channel genToOut) - :>> Sink) + :=> Generator (Channel genToOut) + :=> Sink) = And (IsDP (Source (Channel inToGen))) (IsDP (Generator (Channel genToOut))) - IsDP (Source (Channel (a :<+> more))) = IsDP (Source (Channel more)) - IsDP (Source (Channel Eof)) = 'True + IsDP ( Source (Channel inToGen) + :=> Generator (Channel genToOut) + :=> FeedbackChannel toSource + :=> Sink + ) + = And (IsDP (Source (Channel inToGen))) (IsDP (Generator (Channel genToOut))) + IsDP (Source (Channel (a :<+> more))) = IsDP (Source (Channel more)) + IsDP (Source (Channel Eof)) = 'True IsDP (Generator (Channel (a :<+> more))) = IsDP (Generator (Channel more)) IsDP (Generator (Channel a)) = 'True IsDP x = 'False @@ -81,16 +87,18 @@ ValidDP 'False = TypeError ( 'Text "Invalid Semantic for Building DP Program" ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :>> Generator CHANS :>> Sink" + ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" ':$$: 'Text "CHANS = Channel CH" ':$$: 'Text "CH = Type | Type :<+> CH" - ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'" + ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) -- Inductive Type Family for Expanding and building Source, Generator, Filter and Sink Functions Signatures type family WithSource (dpDefinition :: Type) (monadicAction :: Type -> Type) :: Type where - WithSource (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) monadicAction - = WithSource (ChanIn inToGen) monadicAction + WithSource (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) monadicAction + = WithSource (ChanIn inToGen) monadicAction + WithSource (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) monadicAction + = WithSource (ChanOutIn toSource inToGen) monadicAction WithSource (ChanIn (dpDefinition :<+> more)) monadicAction = WriteChannel dpDefinition -> WithSource (ChanIn more) monadicAction WithSource (ChanIn Eof) monadicAction = monadicAction () WithSource (ChanOutIn (dpDefinition :<+> more) ins) monadicAction = ReadChannel dpDefinition -> WithSource (ChanOutIn more ins) monadicAction @@ -101,53 +109,70 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :>> Generator CHANS :>> Sink" + ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" ':$$: 'Text "CHANS = Channel CH" ':$$: 'Text "CH = Type | Type :<+> CH" - ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'" + ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) type family WithGenerator (a :: Type) (filter :: Type) (monadicAction :: Type -> Type) :: Type where - WithGenerator (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) filter monadicAction - = filter -> WithGenerator (ChanOutIn inToGen genToOut) filter monadicAction - WithGenerator (ChanIn (a :<+> more)) filter monadicAction = WriteChannel a -> WithGenerator (ChanIn more) filter monadicAction - WithGenerator (ChanIn Eof) filter monadicAction = monadicAction () - WithGenerator (ChanOutIn (a :<+> more) ins) filter monadicAction = ReadChannel a -> WithGenerator (ChanOutIn more ins) filter monadicAction - WithGenerator (ChanOutIn Eof ins) filter monadicAction = WithGenerator (ChanIn ins) filter monadicAction - WithGenerator dpDefinition _ _ = TypeError - ( 'Text "Invalid Semantic for Generator Stage" - ':$$: 'Text "in the DP Definition '" - ':<>: 'ShowType dpDefinition - ':<>: 'Text "'" - ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :>> Generator CHANS :>> Sink" - ':$$: 'Text "CHANS = Channel CH" - ':$$: 'Text "CH = Type | Type :<+> CH" - ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'" - ) - + WithGenerator (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) filter monadicAction + = filter -> WithGenerator (ChanOutIn inToGen genToOut) filter monadicAction + WithGenerator (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) filter monadicAction + = filter -> WithGenerator (ChanOutIn inToGen (ChanInIn genToOut toSource)) filter monadicAction + WithGenerator (ChanInIn (a :<+> more) ins) filter monadicAction = WriteChannel a -> WithGenerator (ChanInIn more ins) filter monadicAction + WithGenerator (ChanInIn Eof ins) filter monadicAction = WithGenerator (ChanIn ins) filter monadicAction + WithGenerator (ChanIn (a :<+> more)) filter monadicAction = WriteChannel a -> WithGenerator (ChanIn more) filter monadicAction + WithGenerator (ChanIn Eof) filter monadicAction = monadicAction () + WithGenerator (ChanOutIn (a :<+> more) ins) filter monadicAction = ReadChannel a -> WithGenerator (ChanOutIn more ins) filter monadicAction + WithGenerator (ChanOutIn Eof (ChanInIn ins ins2)) filter monadicAction = WithGenerator (ChanInIn ins ins2) filter monadicAction + WithGenerator (ChanOutIn Eof ins) filter monadicAction = WithGenerator (ChanIn ins) filter monadicAction + WithGenerator dpDefinition _ _ = TypeError + ( 'Text "Invalid Semantic for Generator Stage" + ':$$: 'Text "in the DP Definition '" + ':<>: 'ShowType dpDefinition + ':<>: 'Text "'" + ':$$: 'Text "Language Grammar:" + ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" + ':$$: 'Text "CHANS = Channel CH" + ':$$: 'Text "CH = Type | Type :<+> CH" + ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" + ) + type family WithFilter (dpDefinition :: Type) (param :: Type) (monadicAction :: Type -> Type) :: Type where - WithFilter (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) param monadicAction + WithFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) param monadicAction = param -> WithFilter (ChanOutIn inToGen genToOut) param monadicAction - WithFilter (ChanIn (dpDefinition :<+> more)) param monadicAction = WriteChannel dpDefinition -> WithFilter (ChanIn more) param monadicAction - WithFilter (ChanIn Eof) param monadicAction = monadicAction () - WithFilter (ChanOutIn (dpDefinition :<+> more) ins) param monadicAction = ReadChannel dpDefinition -> WithFilter (ChanOutIn more ins) param monadicAction + WithFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) param monadicAction + = param -> WithFilter (ChanOutIn inToGen (ChanInIn genToOut toSource)) param monadicAction + WithFilter (ChanInIn (a :<+> more) ins) param monadicAction + = WriteChannel a -> WithFilter (ChanInIn more ins) param monadicAction + WithFilter (ChanInIn Eof ins) param monadicAction + = WithFilter (ChanIn ins) param monadicAction + WithFilter (ChanIn (dpDefinition :<+> more)) param monadicAction + = WriteChannel dpDefinition -> WithFilter (ChanIn more) param monadicAction + WithFilter (ChanIn Eof) param monadicAction = monadicAction () + WithFilter (ChanOutIn (dpDefinition :<+> more) ins) param monadicAction + = ReadChannel dpDefinition -> WithFilter (ChanOutIn more ins) param monadicAction + WithFilter (ChanOutIn Eof (ChanInIn ins ins2)) param monadicAction + = WithFilter (ChanInIn ins ins2) param monadicAction WithFilter (ChanOutIn Eof ins) param m = WithFilter (ChanIn ins) param m - WithFilter dpDefinition _ _ = TypeError - ( 'Text "Invalid Semantic Semantic for Generator Stage" - ':$$: 'Text "in the DP Definition '" - ':<>: 'ShowType dpDefinition - ':<>: 'Text "'" - ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :>> Generator CHANS :>> Sink" - ':$$: 'Text "CHANS = Channel CH" - ':$$: 'Text "CH = Type | Type :<+> CH" - ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'" - ) + WithFilter dpDefinition _ _ = TypeError + ( 'Text "Invalid Semantic Semantic for Generator Stage" + ':$$: 'Text "in the DP Definition '" + ':<>: 'ShowType dpDefinition + ':<>: 'Text "'" + ':$$: 'Text "Language Grammar:" + ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" + ':$$: 'Text "CHANS = Channel CH" + ':$$: 'Text "CH = Type | Type :<+> CH" + ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" + ) type family WithSink (dpDefinition :: Type) (monadicAction :: Type -> Type) :: Type where - WithSink (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) monadicAction - = WithSink (ChanOut genToOut) monadicAction + WithSink (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) monadicAction + = WithSink (ChanOut genToOut) monadicAction + WithSink (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) monadicAction + = WithSink (ChanOut genToOut) monadicAction WithSink (ChanOut (dpDefinition :<+> more)) monadicAction = ReadChannel dpDefinition -> WithSink (ChanOut more) monadicAction WithSink (ChanOut Eof) monadicAction = monadicAction () WithSink dpDefinition _ = TypeError @@ -156,10 +181,10 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :>> Generator CHANS :>> Sink" + ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" ':$$: 'Text "CHANS = Channel CH" ':$$: 'Text "CH = Type | Type :<+> CH" - ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'" + ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -211,7 +236,7 @@ -- | 'DynamicPipeline' data type which contains all the three Stages definitions that have been generated by other combinators like 'withSource', -- 'withGenerator' and 'withSink'. -- --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filterState@]: State of the 'StateT' 'Monad' that is the local State of the Filter execution -- @@ -229,7 +254,7 @@ -- in orther to know how to spawn a new 'Filter' if it is needed, and the 'Stage' of the Generator to allow the user to perform some computation -- in that case. -- --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filterState@]: State of the 'StateT' 'Monad' that is the local State of the Filter execution -- @@ -249,7 +274,7 @@ -- -- * All the 'Filter' execution (a.k.a. @forM_ actors runStage@) executes in a 'StateT' 'Monad' to share an internal state among 'Actor's. -- --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filterState@]: State of the 'StateT' 'Monad' that is the local State of the Filter execution -- @@ -264,7 +289,7 @@ -- | 'Actor' Is a particular 'Stage' computation inside a 'Filter'. -- --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filterState@]: State of the 'StateT' 'Monad' that is the local State of the Filter execution -- @@ -343,7 +368,7 @@ -- | Combinator for Building a 'Source' Stage. It uses an Associated Type Class to deduce the Function Signature required to the user -- taken from /DP/ Type Level Flow Definition --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@st@]: Existential Scope of 'DP' 'Monad'. {-# INLINE withSource #-} @@ -354,7 +379,7 @@ -- | Combinator for Building a 'Generator' Stage. It uses an Associated Type Class to deduce the Function Signature required to the user -- taken from /DP/ Type Level Flow Definition --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filter@]: 'Filter' template type -- @@ -367,7 +392,7 @@ -- | Combinator for Building a 'Sink' Stage. It uses an Associated Type Class to deduce the Function Signature required to the user -- taken from /DP/ Type Level Flow Definition --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@st@]: Existential Scope of 'DP' 'Monad'. {-# INLINE withSink #-} @@ -385,31 +410,36 @@ mkDP' = DynamicPipeline @dpDefinition -- Hiding DP Constraint for running DP -type DPConstraint dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4 = - ( MkChans dpDefinition - , HChan dpDefinition ~ r3 t2 - , Filter dpDefinition filterState filterParam st ~ filter - , CloseList l3 - , CloseList l4 - , CloseList (HAppendListR l1 l2) - , iparams ~ WithSource dpDefinition (DP st) - , gparams ~ WithGenerator dpDefinition filter (DP st) - , oparams ~ WithSink dpDefinition (DP st) - , ArityRev iparams (HLength (ExpandSourceToCh dpDefinition)) - , ArityFwd iparams (HLength (ExpandSourceToCh dpDefinition)) - , HCurry' (HLength (ExpandSourceToCh dpDefinition)) iparams l3 (DP st ()) - , ArityRev gparams (HLength (ExpandGenToCh dpDefinition filter)) - , ArityFwd gparams (HLength (ExpandGenToCh dpDefinition filter)) - , HCurry' (HLength (ExpandGenToCh dpDefinition filter)) gparams (filter ': HAppendListR l1 l2) (DP st ()) - , ArityRev oparams (HLength (ExpandSinkToCh dpDefinition)) - , ArityFwd oparams (HLength (ExpandSinkToCh dpDefinition)) - , HCurry' (HLength (ExpandSinkToCh dpDefinition)) oparams l4 (DP st ()) - , AllChans r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4) +type DPConstraint dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi + = ( MkChans dpDefinition + , HChan dpDefinition ~ ChanRecord slr slw glr glw silr silw + , HAppendList slr slw + , HAppendList glr glw + , HAppendList silr silw + , Filter dpDefinition filterState filterParam st ~ filter + , ls ~ HAppendListR slr slw + , lsi ~ HAppendListR silr silw + , CloseList ls + , CloseList lsi + , CloseList (HAppendListR glr glw) + , iparams ~ WithSource dpDefinition (DP st) + , gparams ~ WithGenerator dpDefinition filter (DP st) + , oparams ~ WithSink dpDefinition (DP st) + , ArityRev iparams (HLength (ExpandSourceToCh dpDefinition)) + , ArityFwd iparams (HLength (ExpandSourceToCh dpDefinition)) + , HCurry' (HLength (ExpandSourceToCh dpDefinition)) iparams ls (DP st ()) + , ArityRev gparams (HLength (ExpandGenToCh dpDefinition filter)) + , ArityFwd gparams (HLength (ExpandGenToCh dpDefinition filter)) + , HCurry' (HLength (ExpandGenToCh dpDefinition filter)) gparams (filter ': HAppendListR glr glw) (DP st ()) + , ArityRev oparams (HLength (ExpandSinkToCh dpDefinition)) + , ArityFwd oparams (HLength (ExpandSinkToCh dpDefinition)) + , HCurry' (HLength (ExpandSinkToCh dpDefinition)) oparams lsi (DP st ()) + ) {-# INLINE buildDPProg #-} -buildDPProg :: forall dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4. - DPConstraint dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4 - => DynamicPipeline dpDefinition filterState filterParam st -> DP st () +buildDPProg :: forall dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi. + DPConstraint dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi => + DynamicPipeline dpDefinition filterState filterParam st -> DP st () buildDPProg DynamicPipeline{..} = do (cIns, cGen, cOut) <- inGenOut <$> withDP (makeChans @dpDefinition) let genWithFilter = _gsFilterTemplate generator .*. cGen @@ -419,8 +449,8 @@ -- | Smart constructor for 'DynamicPipeline' Definition {-# INLINE mkDP #-} -mkDP :: forall dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4. - DPConstraint dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4 +mkDP :: forall dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi. + DPConstraint dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi => Stage (WithSource dpDefinition (DP st)) -- ^ 'Source' Stage generated by 'withSource' combinator -> GeneratorStage dpDefinition filterState filterParam st -- ^ 'Generator' Stage generated by 'withGenerator' combinator -> Stage (WithSink dpDefinition (DP st)) -- ^ 'Sink' Stage generated by 'withSink' combinator @@ -454,15 +484,14 @@ close = const $ pure () -- | 'SpawnFilterConstraint' Constraint type alias -type SpawnFilterConstraint dpDefinition readElem st filterState filterParam l r t l1 b0 l2 l3 b2 b3 l4 = +type SpawnFilterConstraint dpDefinition readElem st filterState filterParam l l1 l2 l3 b2 b3 l4 = ( MkChans (ChansFilter dpDefinition) - , FilterChans r (HList l3) t (HList (ReadChannel readElem : l1)) , l1 ~ l , CloseList (ReadChannel readElem ': l4) , HAppendList l l3 , l4 ~ HAppendListR l l3 , l2 ~ (readElem ': ReadChannel readElem ': l4) - , HChan (ChansFilter dpDefinition) ~ r t + , HChan (ChansFilter dpDefinition) ~ InOutChan (ReadChannel readElem : l) l3 , WithFilter dpDefinition filterParam (StateT filterState (DP st)) ~ (b2 -> ReadChannel b2 -> b3) , HLength (ExpandFilterToCh dpDefinition filterParam) ~ HLength l2 , HCurry' (HLength l2) (WithFilter dpDefinition filterParam (StateT filterState (DP st))) l2 (StateT filterState (DP st) ()) @@ -474,7 +503,7 @@ -- The user will have the capability to select how those filters are going to be spawned, for example on each read element, how to setup -- initial states of 'StateT' Monad on 'Actor' computations in filters, among others. -- --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition +-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@readElem@]: Type of the element that is being read from the Selected Channel in the 'Generator' Stage -- @@ -537,8 +566,8 @@ -- | Run 'UnFoldFilter' {-# INLINE unfoldF #-} -unfoldF :: forall dpDefinition readElem st filterState filterParam l r t l1 b0 l2 l3 b2 b3 l4. - SpawnFilterConstraint dpDefinition readElem st filterState filterParam l r t l1 b0 l2 l3 b2 b3 l4 +unfoldF :: forall dpDefinition readElem st filterState filterParam l l1 l2 l3 b2 b3 l4. + SpawnFilterConstraint dpDefinition readElem st filterState filterParam l l1 l2 l3 b2 b3 l4 => UnFoldFilter dpDefinition readElem st filterState filterParam l -- ^ 'UnFoldFilter' -> DP st (HList l) -- ^Return the list of 'ReadChannel's with the results to be read for the 'Generator' at the end. You can use this to pass the results to 'Sink' unfoldF = loopSpawn @@ -551,7 +580,7 @@ _ufOnElem elem' if _ufSpawnIf elem' then do - (reads', writes' :: HList l3) <- getFilterChannels <$> DP (makeChans @(ChansFilter dpDefinition)) + (reads', writes' :: HList l3) <- getFilterChannels <$> DP (makeChansF @(ChansFilter dpDefinition)) let hlist = elem' .*. _ufReadChannel .*. (_ufRsChannels `hAppendList` writes') void $ runFilter _ufFilter (_ufInitState elem') hlist (_ufReadChannel .*. (_ufRsChannels `hAppendList` writes')) return $ uf { _ufReadChannel = hHead reads', _ufRsChannels = hTail reads' }