dynamic-pipeline 0.3.0.0 → 0.3.1.0
raw patch · 5 files changed
+131/−37 lines, 5 filesPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
API changes (from Hackage documentation)
+ DynamicPipeline: (|=>) :: MonadIO m => ReadChannel a -> WriteChannel a -> m ()
+ DynamicPipeline: (|=>|) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> b) -> m ()
+ DynamicPipeline: (|>=>) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()
+ DynamicPipeline: (|>=>|) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()
+ DynamicPipeline: infixl 5 |=>|
+ DynamicPipeline: mapF_ :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> b) -> m ()
+ DynamicPipeline: mapMF_ :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()
+ DynamicPipeline: mapM_ :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()
+ DynamicPipeline: map_ :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> b) -> m ()
+ DynamicPipeline.Channel: (|=>) :: MonadIO m => ReadChannel a -> WriteChannel a -> m ()
+ DynamicPipeline.Channel: (|=>|) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> b) -> m ()
+ DynamicPipeline.Channel: (|>=>) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()
+ DynamicPipeline.Channel: (|>=>|) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()
+ DynamicPipeline.Channel: infixl 5 |=>|
+ DynamicPipeline.Channel: infixr 5 |>=>|
+ DynamicPipeline.Channel: mapF_ :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> b) -> m ()
+ DynamicPipeline.Channel: mapMF_ :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()
+ DynamicPipeline.Channel: mapM_ :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()
+ DynamicPipeline.Channel: map_ :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> b) -> m ()
- DynamicPipeline: infixr 5 |>>
+ DynamicPipeline: infixr 5 |>=>|
Files
- dynamic-pipeline.cabal +1/−1
- examples/Misc/RepeatedTwiceDP.hs +11/−15
- src/DynamicPipeline.hs +16/−3
- src/DynamicPipeline/Channel.hs +77/−2
- src/DynamicPipeline/Stage.hs +26/−16
dynamic-pipeline.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: dynamic-pipeline-version: 0.3.0.0+version: 0.3.1.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/).
examples/Misc/RepeatedTwiceDP.hs view
@@ -18,9 +18,9 @@ source' :: Stage (ReadChannel String -> WriteChannel Int -> WriteChannel Int -> WriteChannel Double -> WriteChannel String -> DP st ()) source' = withSource @DPExample $ \feedback cout cout' cout'' toFilter -> do + finish cout' >> finish cout'' unfoldT ([1 .. 10] <> [1 .. 10]) cout identity- finish cout >> finish cout' >> finish cout''- foldM_ feedback (`push` toFilter)+ feedback |=> toFilter generator' :: GeneratorStage DPExample (Maybe Int) Int s generator' =@@ -40,9 +40,8 @@ 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 + ft |=>| cout $ show+ sec |=>| odn $ id filterTemp :: Filter DPExample (Maybe Int) Int s filterTemp = mkFilter actorRepeted@@ -58,23 +57,20 @@ -> 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+ rc |>=>| wc $ \e -> do + -- putTextLn $ "1) Elem: " <> show e <> " - Param: " <> show i+ if e /= i then pure $ Just e else pure Nothing push i wc'- foldM_ rc' $ \e -> do - putTextLn $ "2) Elem: " <> show e <> " - Param: " <> show i- push e wc'- finish wc'+ rc' |>=>| wc' $ \e -> do + -- putTextLn $ "2) Elem: " <> show e <> " - Param: " <> show i+ pure $ Just e 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+ rd |=>| wd $ id sink' :: Stage (ReadChannel Int -> ReadChannel Int -> ReadChannel Double -> DP s ())
src/DynamicPipeline.hs view
@@ -107,6 +107,14 @@ -- * Channels ReadChannel, WriteChannel,+ (|=>),+ (|=>|),+ (|>=>),+ (|>=>|),+ mapF_,+ map_,+ mapM_,+ mapMF_, foldM_, foldWithM_, push,@@ -127,15 +135,20 @@ -- The following is the Regular Grammar allowed to build a /DPP/ Flow definition: -- -- @--- __DP__ = 'Source' __CHANS__ ':=>' 'Generator' __CHANS__ ':=>' 'Sink'--- __CHANS__ = 'Channel' __CH__--- __CH__ = 'Eof' | 'Type' ':<+>' __CH__+-- __DP__ -> 'Source' __CHANS__ ':=>' 'Generator' __CHANS__ ':=>' 'Sink'+-- __DP__ -> 'Source' __CHANS ':=>' 'Generator' __CHANS__ ':=>' __FEEDBACK__ ':=>' 'Sink'+-- __CHANS__ -> 'Channel' __CH__+-- __FEEDBACK__ -> 'FeedbackChannel' __CH__+-- __CH__ -> 'Type' ':<+>' __CH__ | 'Eof' -- @ -- -- Example: -- -- @ 'Source' ('Channel' (Int ':<+>' Int)) ':=>' 'Generator' ('Channel' (Int ':<+>' Int)) ':=>' 'Sink' @ --+-- Or with Feedback Channel to retrofit Streamming+--+-- @ 'Source' ('Channel' (Int ':<+>' Int)) ':=>' 'Generator' ('Channel' (Int ':<+>' Int)) ':=>' 'FeedbackChannel' ('String' ':<+>' 'Eof') ':=>' 'Sink' @ -- -- $dp -- 'DynamicPipeline' Data type is the point where all the information is contained in order the library can run our /DP/ Algorithm.
src/DynamicPipeline/Channel.hs view
@@ -12,6 +12,14 @@ module DynamicPipeline.Channel ( ReadChannel , WriteChannel+ , (|=>)+ , (|=>|)+ , (|>=>)+ , (|>=>|)+ , mapF_+ , map_+ , mapM_+ , mapMF_ , foldM_ , foldWithM_ , push@@ -30,10 +38,15 @@ ) import Data.ByteString as B import Data.Foldable as F+ hiding ( mapM_+ ) import Data.HList hiding ( foldM_+ , mapM_ ) import GHC.IO.Handle as H import Relude as R+ hiding ( mapM_+ ) -- | 'WriteChannel' can only write values into some Channel Queue@@ -46,6 +59,68 @@ -- [@a@]: Type that this Channel can read newtype ReadChannel a = ReadChannel { unRead :: OutChan (Maybe a) } +-- | 'map_' is a /Natural Transformation/ from consumer 'ReadChannel' to some producer 'WriteChannel' applying a transformation with function @f@+{-# INLINE map_ #-}+map_ :: MonadIO m+ => ReadChannel a -- ^'ReadChannel'+ -> WriteChannel b -- ^'ReadChannel'+ -> (a -> b) -- ^Monadic Transformation to do with read element+ -> m ()+map_ rc wc f = foldM_ rc $ flip push wc . f++-- | Same as 'map_' but with 'id' combinator+(|=>) :: MonadIO m => ReadChannel a -> WriteChannel a -> m ()+(|=>) rc wc = map_ rc wc id++infixl 5 |=>++-- | Same as 'map_' but mark Eof Channel after all processing+{-# INLINE mapF_ #-}+mapF_ :: MonadIO m+ => ReadChannel a -- ^'ReadChannel'+ -> WriteChannel b -- ^'ReadChannel'+ -> (a -> b) -- ^Monadic Transformation to do with read element+ -> m ()+mapF_ rc wc f = map_ rc wc f >> finish wc++-- | Alias 'mapF_'+(|=>|) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> b) -> m ()+(|=>|) = mapF_++infixl 5 |=>|+++-- | Same as 'map_' But applying a Monadic mapping+{-# INLINE mapM_ #-}+mapM_ :: MonadIO m+ => ReadChannel a -- ^'ReadChannel'+ -> WriteChannel b -- ^'ReadChannel'+ -> (a -> m (Maybe b)) -- ^Monadic Transformation to do with read element+ -> m ()+mapM_ rc wc f = foldM_ rc $ maybe (pure ()) (`push` wc) <=< f++-- | Alias 'mapM_'+(|>=>) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()+(|>=>) = mapM_++infixr 5 |>=>++-- | Same as 'mapM_' but mark Eof Channel after all processing+{-# INLINE mapMF_ #-}+mapMF_ :: MonadIO m+ => ReadChannel a -- ^'ReadChannel'+ -> WriteChannel b -- ^'ReadChannel'+ -> (a -> m (Maybe b)) -- ^Monadic Transformation to do with read element+ -> m ()+mapMF_ rc wc f = mapM_ rc wc f >> finish wc++-- | Alias 'mapMF_'+(|>=>|) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m ()+(|>=>|) = mapMF_++infixr 5 |>=>|++ -- | 'foldM_' is a /Catamorphism/ for consuming a 'ReadChannel' and do some Monadic @m@ computation with each element {-# INLINE foldM_ #-} foldM_ :: MonadIO m@@ -97,7 +172,7 @@ unfoldM = loop' where loop' seed fn stopIfM writeChannel =- ifM stopIfM (pure ()) (seed >>= flip push writeChannel . fn >> loop' seed fn stopIfM writeChannel)+ ifM stopIfM (finish writeChannel) (seed >>= flip push writeChannel . fn >> loop' seed fn stopIfM writeChannel) -- | Using 'unfoldM', unfold from file {-# INLINE unfoldFile #-} unfoldFile :: MonadIO m@@ -111,7 +186,7 @@ -- | Idem 'unfoldM' but for 'Foldable', for example a List @[a]@. Useful for testing purpose {-# INLINE unfoldT #-} unfoldT :: (MonadIO m, Foldable t) => t a -> WriteChannel b -> (a -> b) -> m ()-unfoldT ts writeChannel fn = forM_ ts (flip push writeChannel . fn)+unfoldT ts writeChannel fn = forM_ ts (flip push writeChannel . fn) >> finish writeChannel {-# WARNING newChannel "INTERNAL USE" #-} {-# NOINLINE newChannel #-}
src/DynamicPipeline/Stage.hs view
@@ -42,7 +42,7 @@ import Control.Lens hiding ((<|)) import Data.HList import Data.List.NonEmpty-import DynamicPipeline.Channel+import DynamicPipeline.Channel hiding (mapM_) import DynamicPipeline.Flow import GHC.TypeLits import Relude as R@@ -87,9 +87,11 @@ ValidDP 'False = TypeError ( 'Text "Invalid Semantic for Building DP Program" ':$$: 'Text "Language Grammar:"- ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink"- ':$$: 'Text "CHANS = Channel CH"- ':$$: 'Text "CH = Type | Type :<+> CH"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink"+ ':$$: 'Text "CHANS -> Channel CH"+ ':$$: 'Text "FEEDBACK -> FeedbackChannel CH"+ ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -109,9 +111,11 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:"- ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink"- ':$$: 'Text "CHANS = Channel CH"- ':$$: 'Text "CH = Type | Type :<+> CH"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink"+ ':$$: 'Text "CHANS -> Channel CH"+ ':$$: 'Text "FEEDBACK -> FeedbackChannel CH"+ ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -133,9 +137,11 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:"- ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink"- ':$$: 'Text "CHANS = Channel CH"- ':$$: 'Text "CH = Type | Type :<+> CH"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink"+ ':$$: 'Text "CHANS -> Channel CH"+ ':$$: 'Text "FEEDBACK -> FeedbackChannel CH"+ ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -162,9 +168,11 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:"- ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink"- ':$$: 'Text "CHANS = Channel CH"- ':$$: 'Text "CH = Type | Type :<+> CH"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink"+ ':$$: 'Text "CHANS -> Channel CH"+ ':$$: 'Text "FEEDBACK -> FeedbackChannel CH"+ ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -181,9 +189,11 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:"- ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink"- ':$$: 'Text "CHANS = Channel CH"- ':$$: 'Text "CH = Type | Type :<+> CH"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink"+ ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink"+ ':$$: 'Text "CHANS -> Channel CH"+ ':$$: 'Text "FEEDBACK -> FeedbackChannel CH"+ ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" )