stc-lang (empty) → 1.0.0

raw patch · 29 files changed

+3149/−0 lines, 29 filesdep +BoundedChandep +HUnitdep +abstract-parsetup-changed

Dependencies added: BoundedChan, HUnit, abstract-par, aeson, base, bytestring, clock, deepseq, ghc-prim, hashable, hashtables, hedis, hw-kafka-client, microlens, microlens-aeson, monad-par, monad-par-extras, mtl, random, stc-lang, test-framework, test-framework-hunit, text, time, transformers, uuid-types, vector, yaml

Files

• LICENSE +30/−0
• README.md +26/−0
• Setup.hs +2/−0
• src/Control/Monad/Generator.hs +270/−0
• src/Control/Monad/SD.hs +33/−0
• src/Control/Monad/SD/Case.hs +49/−0
• src/Control/Monad/SD/Combinator.hs +123/−0
• src/Control/Monad/SD/FRP.hs +92/−0
• src/Control/Monad/SD/Ohua.hs +341/−0
• src/Control/Monad/SD/STCLang.hs +48/−0
• src/Control/Monad/SD/Smap.hs +215/−0
• src/Control/Monad/Stream.hs +16/−0
• src/Control/Monad/Stream/Chan.hs +24/−0
• src/Control/Monad/Stream/Par.hs +53/−0
• src/Control/Monad/Stream/PinnedChan.hs +40/−0
• src/Data/Dynamic2.hs +167/−0
• src/Data/StateElement.hs +30/−0
• src/Type/Magic.hs +13/−0
• src/Type/Magic/GHC8.hs +59/−0
• src/Type/Magic/OldGHC.hs +52/−0
• stc-lang.cabal +140/−0
• stream-bench/CampaignProcMap.hs +41/−0
• stream-bench/MutableNFMap.hs +48/−0
• stream-bench/MutableSet.hs +55/−0
• stream-bench/algo.hs +803/−0
• test/FakeComputation.hs +60/−0
• test/SD/Correctness.hs +251/−0
• test/SD/Performance.hs +56/−0
• test/Spec.hs +12/−0

@@ -0,0 +1,30 @@+Copyright Norman Rink, Sebastian Ertel, Justus Adam (c) 2017-2019++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Author name here nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,26 @@+# STCLang: A library for implicit monadic dataflow parallelism++STCLang is a library that enables stateful, implicit, monadic parallel+computation in Haskell. The core ideas come from the+[ohua](https://ohua-dev.github.io) project.++STCLang lets you create parallel dataflows with stateful nodes without having to+explicitly wire complex graph structures. Instead the program is written with an+embedded, monadic DSL and automatically transformed into a graph and executed in+parallel.++On top of the base abstraction we have also built an FRP (functional reactive+programming) interface. This allows you to run reactive programs on sequential+streams of values and leverage pipeline parallelism to peed up computation.++We also [published](#publication) the theory and concepts behind this library.++## Publication++We documented the principles in this library in a paper at Haskell'2019.++A link to the publication will appear here once we have one, e.t.a. is 22th of+August (date of the conference). Should it be after this date now, but there's+still no link, I probably forgot. In that case open an issue, shoot+[me](https://github.com/JustusAdam) an email or tweet me+[@justusadam_](https://twitter.com/justusadam_).

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,270 @@+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE MonadComprehensions #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE CPP #-}++module Control.Monad.Generator+    ( IsGenerator(..)+    , liftIO+    , foldlGenerator+    , foldlGeneratorT+    , foldlGenerator_+    , foldlGeneratorT_+    , chanToGenerator+    , ioReaderToGenerator+    , foldableGenerator+    , foldableGenerator'+    , foldableGenerator''+    , foldableGeneratorEval+    , listGenerator+    , stateToGenerator+    , Generator+    -- ** A mutable IO generator variable+    , GenVar, newGenVar, pull+    ) where++import Control.Applicative+import Control.Arrow+import Control.Concurrent.Chan+import Control.Concurrent.MVar+import Control.DeepSeq+import Control.Monad.State+import Data.Foldable (foldr')+import Data.Tuple+import qualified GHC.Exts (IsList(..))+++-- | A natural transformation+type f ~> g = forall x. f x -> g x++------------------------------------------------------------------+--+-- The generator+--+------------------------------------------------------------------+-- | There are three basic ways to construct this generator, which+-- correspond to the (not exposed) constructors of this+-- type. 1. 'finish' this marks the end of a generator, 2. 'yield' is+-- used to return a value and continue with the computation afterwards+-- and finally using 'liftIO' you can execute any IO action and then+-- continue.+data Generator m a+    = Finished+    | Yield (Generator m a)+            a+    | NeedM (m (Generator m a))++-- At first I implemented all of the `Applicative` and `Monad`+-- functions fully, that is to say with all the recursion down the+-- source generator.  I was already suspecting a pattern there,+-- because the recursion for both looked very similar, I just couldn't+-- quite figure it out.  When suddenly something *magical* happened.+-- I was looking at `Alternative`, because that instance is needed for+-- `MonadComprehensions` and I was again reminded of just how similar+-- `Alternative` is to `Monoid` and I realized that `Generator` is in+-- fact a `Monoid`. The empty element is `Finished` because it can be+-- appended or prepended to any generator without changing its meaning+-- and `mappend` is simply exhausting the first generator first,+-- followed by the second one.  After having that realization+-- implementing `Monad` and `Applicative` became really easy because+-- you just simply create a new generator by applying the function and+-- then prepend a recursion of the respective operation (`>>=` or+-- `<*>`)+-- You can see the generator in action by running the examples at the bottom in ghci with `runGenerator`+mappendGen :: Functor m => Generator m a -> Generator m a -> Generator m a+Finished `mappendGen` gen2 = gen2+NeedM sc `mappendGen` gen2 = NeedM $ (`mappendGen` gen2) <$> sc+Yield g v `mappendGen` gen2 = Yield (g `mappendGen` gen2) v++instance Functor m => Monoid (Generator m a) where+    mempty = Finished+#if MIN_VERSION_base(4,11,0)+instance Functor m => Semigroup (Generator m a) where+    (<>) = mappendGen+#else+    mappend = mappendGen+#endif+instance Functor m => Functor (Generator m) where+    fmap _ Finished = Finished+    fmap f (NeedM m) = NeedM $ fmap (fmap f) m+    fmap f (Yield g a) = Yield (fmap f g) (f a)++instance Functor m => Applicative (Generator m) where+    pure = Yield Finished+    Finished <*> _ = Finished+    NeedM m <*> v = NeedM $ (<*> v) <$> m+    Yield fg f <*> v = fmap f v `mappend` (fg <*> v)++instance Functor m => Monad (Generator m) where+    return = pure+    Finished >>= _ = Finished+    NeedM m >>= f = NeedM $ (>>= f) <$> m+    Yield cont a >>= f = f a `mappend` (cont >>= f)++-- | This is needed to get the Monad comprehensions+instance Functor m => Alternative (Generator m) where+    empty = mempty+    (<|>) = mappend++-- | IO can be embedded easily+instance MonadIO m => MonadIO (Generator m) where+    liftIO = needM . liftIO++instance Monad m => GHC.Exts.IsList (Generator m a) where+    type Item (Generator m a) = a+    fromList = listGenerator+    toList _ = error "toList: need monad to evaluate generator"++foldlGeneratorT ::+       (IsGenerator g f, Monad m)+    => (f ~> m)+    -> (b -> a -> m b)+    -> b+    -> g a+    -> m b+foldlGeneratorT trans ac = flip go+  where+    go gen seed' =+        trans (step gen) >>=+        maybe (pure seed') (\(a, gen') -> go gen' =<< ac seed' a)++foldlGenerator ::+       (IsGenerator g m, Monad m) => (b -> a -> m b) -> b -> g a -> m b+foldlGenerator = foldlGeneratorT id++foldlGeneratorT_ ::+       (IsGenerator g f, Monad m) => (f ~> m) -> (a -> m ()) -> g a -> m ()+foldlGeneratorT_ trans f = foldlGeneratorT trans (\() a -> f a) ()++foldlGenerator_ :: (IsGenerator g m, Monad m) => (a -> m ()) -> g a -> m ()+foldlGenerator_ = foldlGeneratorT_ id++ioReaderToGenerator :: (IsGenerator g m, Monad g) => m (Maybe a) -> g a+ioReaderToGenerator reader = recur+  where+    recur = maybe finish (`yield` recur) =<< needM reader++chanToGenerator ::+       (MonadIO m, IsGenerator g m, Monad g) => Chan (Maybe a) -> g a+chanToGenerator = ioReaderToGenerator . liftIO . readChan++foldableGenerator :: (Foldable f, IsGenerator g m) => f a -> g a+foldableGenerator = foldr' yield finish++foldableGeneratorEval ::+       (Foldable f, IsGenerator g m) => (forall b. a -> b -> b) -> f a -> g a+foldableGeneratorEval eval = foldr (\a rest -> a `eval` yield a rest) finish++foldableGenerator' :: (Foldable f, IsGenerator g m) => f a -> g a+foldableGenerator' = foldableGeneratorEval seq++foldableGenerator'' :: (Foldable f, IsGenerator g m, NFData a) => f a -> g a+foldableGenerator'' = foldableGeneratorEval deepseq++-----------------------------------------------------------------+--+-- Creating generators+--+------------------------------------------------------------------+listGenerator :: IsGenerator g m => [a] -> g a+listGenerator = foldableGenerator++-- | A generator crated with this will run until it returns `Nothing` in which case the generator finishes+stateToGenerator ::+       (Monad g, IsGenerator g m) => StateT s m (Maybe a) -> s -> g a+stateToGenerator st s = do+    (a, s') <- needM $ runStateT st s+    maybe finish (`yield` stateToGenerator st s') a++------------------------------------------------------------------+--+-- Some more fun stuff that can be done with them+--+------------------------------------------------------------------+-- One fun thing we can do in IO is put the generator in a mutable variable and then just pull values from that.+type GenVar a = MVar (Generator IO a)++newGenVar :: Generator IO a -> IO (GenVar a)+newGenVar = newMVar++-- | This pulls a new value from this var (if possible) and updates its state+pull :: GenVar a -> IO (Maybe a)+pull =+    flip modifyMVar $+    fmap (maybe (Finished, Nothing) (second Just . swap)) . step++------------------------------------------------------------------+--+-- Some examples of comprehensions, composition and state embedding+--+------------------------------------------------------------------+permutations :: Generator IO (Int, Char)+permutations =+    [(i, c) | i <- listGenerator [0 .. 9], c <- listGenerator ['a' .. 'f']]++nonReflexivePermutations :: Int -> Generator IO (Int, Int)+nonReflexivePermutations i = [(a, b) | a <- ints, b <- ints, a /= b]+  where+    ints = listGenerator [0 .. i]++justSomeStuffWithInts :: Generator IO Int+justSomeStuffWithInts =+    flip stateToGenerator 0 $ do+        s <- get+        if s < 100+            then do+                modify (+ 4)+                pure $ Just s+            else do+                liftIO $ putStrLn "We have reached 100" -- It can do IO as well ;)+                pure Nothing++-- and they are all compatible and can be joined together (and depend on each other)+-- Probably dont run this ... it creates a **lot** of output+crazy :: Generator IO (Int, Int, Char)+crazy =+    [ (a + b, a * d, c)+    | i <- justSomeStuffWithInts+    , (b, d) <- nonReflexivePermutations i+    , (a, c) <- permutations+    ]++------------------------------------------------------------------+--+-- A generator interface+--+------------------------------------------------------------------+-- Something I thought of this morning.+-- There could also be a generic interface for generators+-- | A generator @g@ that runs in the monad @m@+class IsGenerator g m | g -> m where+    yield :: a -> g a -> g a+    finish :: g a+    needM :: m a -> g a+    isFinished :: g a -> Bool+    default isFinished :: Eq (g a) =>+        g a -> Bool+    isFinished = (== finish)+    -- | Run until the generator yields its first value or finishes.+    -- Returns the created value and a new generator which represents its updated internal state.+    step :: g a -> m (Maybe (a, g a))+    -- | Run a generator producing a list of output values+    toList :: g a -> m [a]+    default toList :: Monad m =>+        g a -> m [a]+    toList = foldlGenerator (\b a -> pure $ a : b) []++instance Monad m => IsGenerator (Generator m) m where+    yield = flip Yield+    finish = Finished+    needM = NeedM . fmap pure+    isFinished Finished = True+    isFinished _ = False+    step Finished = pure Nothing+    step (NeedM ac) = ac >>= step+    step (Yield g a) = pure $ Just (a, g)+    toList Finished = pure []+    toList (NeedM ac) = ac >>= toList+    toList (Yield g a) = (a :) <$> toList g

@@ -0,0 +1,33 @@+module Control.Monad.SD+  -- | Base functionality+    ( case_+    , if_+    , smap+    , runOhuaM+    , liftWithIndex+    , OhuaM+    , SF+    , SFM+  -- | STCLang re-exports+    , runSTCLang+    , liftWithState+    , STCLang+    , CollSt(..)+    , smapSTC+  -- | Signals re-exports+    , liftSignal+    , runSignals+    , filterSignalM+    , filterSignal+    , Signals+  -- | Combinators+    , mapReduce+    , mapReduceRangeThresh+    ) where++import Control.Monad.SD.Case+import Control.Monad.SD.Combinator+import Control.Monad.SD.FRP+import Control.Monad.SD.Ohua+import Control.Monad.SD.STCLang+import Control.Monad.SD.Smap

@@ -0,0 +1,49 @@+module Control.Monad.SD.Case+    ( case_+    , if_+    ) where++import Control.Monad+import Control.Monad.IO.Class+import Control.Monad.Par.Class as PC+import Control.Monad.SD.Ohua+import Data.StateElement++import Data.List as List+import Data.Maybe++case_ ::+       forall a p. (NFData a, Show a, Eq p)+    => p+    -> [(p, OhuaM a)]+    -> OhuaM a+case_ cond patternsAndBranches = OhuaM moveState comp+  where+    moveState ::+           forall ivar m. (ParIVar ivar m, MonadIO m)+        => GlobalState ivar+        -> m (GlobalState ivar)+    moveState gs =+        (foldM (flip moveStateForward) gs . map snd) patternsAndBranches+    comp ::+           forall ivar m. (ParIVar ivar m, Monad m, MonadIO m, NFData (ivar S))+        => GlobalState ivar+        -> m (a, GlobalState ivar)+    comp gs+      -- find the first pattern that matches+     = do+        let idx = List.findIndex ((cond ==) . fst) patternsAndBranches+        let ith = fromMaybe (error "No pattern found for condition.") idx+      -- one could of course do the following in parallel but it is not a performance bottleneck as of now.+        let trueBranch = patternsAndBranches !! ith+        let falseBranches =+                ((\(before, _:after) -> before ++ after) . List.splitAt ith)+                    patternsAndBranches+        gs' <- foldM (flip moveStateForward) gs $ map snd falseBranches+        (result, gs'') <- runOhua (snd trueBranch) gs'+        return (result, gs'')++if_ :: (Show a, NFData a) => OhuaM Bool -> OhuaM a -> OhuaM a -> OhuaM a+if_ cond then_ else_ = do+    i <- cond+    case_ i [(True, then_), (False, else_)]

@@ -0,0 +1,123 @@+module Control.Monad.SD.Combinator where++import Control.Monad+import Control.Monad.Generator+import Control.Monad.Par.Class as PC+import Control.Monad.Par.Combinator (InclusiveRange, InclusiveRange(..))+import Control.Monad.SD.Ohua+import Control.Monad.SD.STCLang+import Control.Monad.SD.Smap+import Data.Dynamic2+import Data.StateElement++import Control.Monad.State as S+import Data.List as List+  ----+  -- The below comes originally from: https://hackage.haskell.org/package/monad-par-extras-0.3.3/docs/src/Control-Monad-Par-Combinator.html#parMapReduceRangeThresh+  ----+  -- | Computes a binary map\/reduce over a finite range.  The range is+  -- recursively split into two, the result for each half is computed in+  -- parallel, and then the two results are combined.  When the range+  -- reaches the threshold size, the remaining elements of the range are+  -- computed sequentially.+  --+  -- For example, the following is a parallel implementation of+  --+  -- >  foldl (+) 0 (map (^2) [1..10^6])+  --+  -- > parMapReduceRangeThresh 100 (InclusiveRange 1 (10^6))+  -- >        (\x -> return (x^2))+  -- >        (\x y -> return (x+y))+  -- >        0+  --+  -- parMapReduceRangeThresh ::+  --      (NFData a, ParFuture iv p)+  --   => Int -- ^ threshold+  --   -> InclusiveRange -- ^ range over which to calculate+  --   -> (Int -> p a) -- ^ compute one result+  --   -> (a -> a -> p a) -- ^ combine two results (associative)+  --   -> a -- ^ initial result+  --   -> p a+  -- parMapReduceRangeThresh threshold range fn binop init =+  -- loop min max+  -- where+  --   loop min max+  --     | max - min <= threshold =+  --       let mapred a b = do+  --             x <- fn b+  --             result <- a `binop` x+  --             return result+  --        in foldM mapred init [min .. max]+  --     | otherwise = do+  --       let mid = min + ((max - min) `quot` 2)+  --       rght <- spawn $ loop (mid + 1) max+  --       l <- loop min mid+  --       r <- get rght+  --       l `binop` r++instance Show InclusiveRange++mapReduceRangeThresh ::+       (NFData a, Typeable a, Show a)+    => Int -- ^ threshold+    -> InclusiveRange -- ^ range over which to calculate+    -> (Int -> a) -- ^ compute one result+    -> (a -> a -> a) -- ^ combine two results (associative)+    -> a -- ^ initial result+    -> IO a+mapReduceRangeThresh threshold range fn binop init+    -- sadly I could not use STCLang to build this :(+    -- reason: it must be STCLang a b to implement liftSignal instead of just+    -- STCLang b just like OhuaM b+    -- (_, [reduceState]) <- runOhuaM mapReduce [toS init]+    -- return $ fromS reduceState+ = do+    (_, [reduceState]) <- runSTCLang mapReduce chunkGenerator+    return $ fromS reduceState+  where+    mapReduce = do+        reduceST <- liftWithState (return init) reduce+        -- return $\x -> smapGen ((pure . mapAndCombine) >=> reduceST) x+        return $ smapGen ((pure . mapAndCombine) >=> reduceST)+      -- mapReduce = do+      --   smapGen+      --     ((pure . mapAndCombine) >=> liftWithIndex 0 reduce)+      --     chunkGenerator+    chunkGenerator :: Generator IO InclusiveRange+    chunkGenerator =+        flip stateToGenerator range $ do+            (InclusiveRange mi ma) <- S.get+            if mi >= ma+                then return Nothing+                else let mi' = min (mi + threshold) ma+                      in do S.put $ InclusiveRange (mi' + 1) ma+                            return $ Just $ InclusiveRange mi mi'+    list (InclusiveRange mi ma)+        | mi >= ma = []+        | otherwise = InclusiveRange mi mi' : list (InclusiveRange (mi' + 1) ma)+      where+        mi' = min (mi + threshold) ma+    mapAndCombine (InclusiveRange mi ma) =+        let mapred a b =+                let x = fn b+                    result = a `binop` x+                 in result+         in List.foldl mapred init [mi .. ma]+    reduce v = S.get >>= (S.put . (`binop` v))+  --{-# INLINE parMapReduceRangeThresh #-}+  -- streams output from the map phase to the reduce phase++mapReduce ::+       (NFData a, NFData b, Typeable b, Show a, Show b)+    => (a -> b)+    -> (b -> b -> b)+    -> b+    -> [a]+    -> IO b+mapReduce mapper reducer init xs = do+    (_, [reduceState]) <- runOhuaM algo [toS init]+    return $ fromS reduceState+  where+    algo = smapGen (pure . mapper >=> liftWithIndex 0 reduce) $ listGenerator xs+    reduce v = S.get >>= (S.put . (`reducer` v))+  --{-# INLINE mapReduce #-}

@@ -0,0 +1,92 @@+module Control.Monad.SD.FRP+    ( liftSignal+    , runSignals+    , filterSignalM+    , filterSignal+    , Signals+    ) where++import Control.Monad.Generator+import Control.Monad.SD.Case+import Control.Monad.SD.Ohua+import Control.Monad.SD.STCLang+import Control.Monad.SD.Smap+import Data.Dynamic2+import Data.StateElement++import qualified Control.Concurrent as Conc+import qualified Control.Concurrent.BoundedChan as BC+import Control.DeepSeq (NFData)+import Control.Exception (bracket)+import Control.Monad.State as S+import System.IO (hPutStrLn, stderr)++type Signal = IO++type Signals = (Int, S)++instance Show S where+    show _ = "S"++liftSignal :: (Typeable a, NFData a) => Signal a -> IO a -> STCLang Signals a+liftSignal s0 init = do+    idx <-+        S.state $ \s@CollSt {signals} ->+            (length signals, s {signals = signals ++ [toS <$> s0]})+    liftWithState init $ \(i, s) ->+        if i == idx+            then do+                let my = fromS s+                S.put my+                pure my+            else S.get++debugSignals :: Bool+debugSignals = True++printSignalD :: MonadIO m => String -> m ()+printSignalD+    | debugSignals = liftIO . hPutStrLn stderr+    | otherwise = const $ pure ()++runSignals :: NFData a => STCLang Signals a -> IO ([a], [S])+runSignals comp = do+    printSignalD "Running STCLang"+    (comp', s) <- S.runStateT comp mempty+    chan <- BC.newBoundedChan 100+    bracket+        (do printSignalD "Starting signals... "+            forM (zip [0 ..] $ signals s) $ \(idx, sig) ->+                Conc.forkIO $+                forever $ do+                    event <- sig+                    BC.writeChan chan $ Just (idx, event))+        (\threads -> do+             printSignalD "Killing signal threads"+             mapM_ Conc.killThread threads)+        (\_ -> do+             putStrLn "signals done"+             let signalGen = ioReaderToGenerator (BC.readChan chan)+             runOhuaM (smapGen comp' signalGen) $ states s)++filterSignalM ::+       (Show b, NFData a, NFData b)+    => (a -> OhuaM Bool)+    -> (a -> OhuaM b)+    -> STCLang a (Maybe b)+filterSignalM cond f =+    pure $ \item -> if_ (cond item) (Just <$> f item) (pure Nothing)+    -- | @filter init p f@ applies @f@ to only those values @a@ that satisfy the+    -- predicate @p@. For values not satisfying it returns the last computed value+    -- (initially @init@)++filterSignal ::+       (Show b, Typeable b, NFData b, NFData a)+    => IO b -- Initial value for the output+    -> (a -> OhuaM Bool) -- predicate+    -> (a -> OhuaM b) -- computation to perform on `a`+    -> STCLang a b+filterSignal init cond f = do+    g <- liftWithState init $ maybe S.get (\i -> S.put i >> pure i)+    fil <- filterSignalM cond f+    return $ fil >=> g

@@ -0,0 +1,341 @@+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE CPP #-}++--- this implementation does not rely on channels. it builds on futures!+module Control.Monad.SD.Ohua+    ( liftWithIndex+    , liftWithIndex'+    , SF+    , SFM+    , runOhuaM+    , OhuaM(..)+    , GlobalState(..)+    ) where++import Control.Monad++-- import           Control.Monad.Par       as P+import Control.Arrow (first)+import Control.Monad.Par.Class as PC++import Control.Monad.Par.IO as PIO+#ifdef DEBUG_SCHED+import qualified Control.Monad.Par.Scheds.TraceDebuggable as TDB+#endif+import Control.Monad.State as S++--+-- for debugging only:+-- import Debug.Scheduler as P+--+-- import           Control.Parallel    (pseq)+import Data.Dynamic2+import Data.StateElement++import Control.DeepSeq (deepseq)++import GHC.Generics (Generic)+import GHC.Stack (HasCallStack)++-- type SFM s b = State s b+type SFM s b = StateT s IO b++type SF s a b = a -> SFM s b++-- runSF :: SFM s b -> s -> (b,s)+-- runSF = runState+runSF :: SFM s b -> s -> IO (b, s)+runSF = runStateT++-- data OhuaM m globalState result = OhuaM {+--                               moveStateForward :: globalState -> m globalState,+--                               runOhua :: globalState -> m (result, globalState)+--                              }+-- this existential quantification essentially hides the types for ivar and m.+-- this forces somebody with a variable of that type to apply it only to a predefined+-- function that knows what the type of 'ivar' and 'm' is.+-- this way, the types are entirely hidden inside that module and restrict the user/caller+-- to a very specific function, i.e., runOhua and moveStateForward.+-- I love that!+-- sources: https://prime.haskell.org/wiki/ExistentialQuantification+--          https://stackoverflow.com/questions/12031878/what-is-the-purpose-of-rank2types#12033549+-- data OhuaM state result = forall ivar m. (ParIVar ivar m)+--                         => OhuaM {+--                              moveStateForward :: GlobalState ivar state -> m (GlobalState ivar state),+--                              runOhua :: GlobalState ivar state -> m (result, GlobalState ivar state)+--                             }+-- when the data constructor OhuaM is called then the type variables are+-- captured with the according types. when the according functions are called later on, then+-- the input to that call must match the captured types now.+-- the above version quantifies over the whole creation of the data type. it becomes:+-- forall ivar m. (ParIVar ivar m) => ((GlobalState ivar state) -> m (GlobalState ivar state))+--                                 -> ((GlobalState ivar state) -> m (result, GlobalState ivar state))+--                                 -> OhuaM state result+-- but we want to have Rank2Types instead to hide ivar and m! (see the example below!)+data OhuaM result = OhuaM+    { moveStateForward :: forall ivar m. (ParIVar ivar m, MonadIO m) =>+                                             GlobalState ivar -> m (GlobalState ivar)+    , runOhua :: forall ivar m. ( ParIVar ivar m+                                , MonadIO m+                                , NFData (ivar S) -- FIXME giving the MonadIO constraint here seems weird to me because then it totally breaks the abstraction and could write ParIO directly.+                                ) =>+                                    GlobalState ivar -> m ( result+                                                          , GlobalState ivar)+    }++-- Example: ExistentialQuantification vs Rank2Types+-- Prelude> set: -XExistentialQuantification+-- Prelude> data T s r = forall ivar m. (Show ivar, Monad m) => TR { f :: (s,ivar) -> m (ivar,s) }+-- Prelude> :t TR+-- TR :: (Monad m, Show ivar) => ((s, ivar) -> m (ivar, s)) -> T s r+-- that is:+-- TR :: forall ivar m. (Monad m, Show ivar) => ((s, ivar) -> m (ivar, s)) -> T s r+-- BUT:+-- Prelude> set: -Rank2Types+-- Prelude> data T s r = TR { f :: forall ivar m. (Show ivar, Monad m) => (s,ivar) -> m (ivar,s) }+-- translates to:+-- Prelude> :t TR+-- TR :: (forall ivar (m :: * -> *). (Show ivar, Monad m) => (s, ivar) -> m (ivar, s)) -> T s r+--+-- ExistentialQuantification makes only sense when we quantify over the output of a function (i.e.)+-- the type of a record. that is because each function captures its own type variable so you can not+-- compose such data as I tried in <*> or =<< with GlobalState (which came from another data).+data GlobalState ivar = GlobalState+    { input :: [ivar S]+    , result :: [ivar S]+    } deriving (Generic)++-- data GlobalState ivar = GlobalState [ivar S] [ivar S] deriving (Generic)+instance (NFData (ivar S)) => NFData (GlobalState ivar)++--+-- shortcoming: this monad implementation is strict because bind requests the+--              actual value. consider the following situation:+--              do+--                 x1 <- a 5+--                 x2 <- b 5+--                 x3 <- c 5+--              this monad will run these 3 statements in sequence because bind+--              always wants the concrete value although it may not actually be+--              used by the directly following computation. to circumvent this+--              case, one would have to use an applicative here:+--              do+--                (x1,x2,x3) <- (,,) <$> a 5 <*> a 5 <*> a 5+--+instance Functor OhuaM where+    fmap f g = OhuaM (moveStateForward g) $ fmap (first f) . runOhua g++instance Applicative OhuaM where+    pure = return+  -- TODO (<*>) = Control.Monad.ap  this is a requirement if the functor is also a monad.+  -- this is the case so we should create a new functor that is not a monad but only an applicative.+  -- in order to do so we need to provide a OhuaM computation in the new applicative functor that+  -- can be ready executed via runOhua! - (Haxl doesn't care)+    (<*>) :: forall a b. OhuaM (a -> b) -> OhuaM a -> OhuaM b+    f <*> a = OhuaM moveState comp+      where+        moveState ::+               forall ivar m. (ParIVar ivar m, MonadIO m)+            => GlobalState ivar+            -> m (GlobalState ivar)+        moveState gs+        -- there is really no computation here, so no need to spawn anything+         = do+            gs' <- moveStateForward a gs+            moveStateForward f gs'+        -- there is no state change here really. I could have returned gs' as well, I suppose.+        comp ::+               forall ivar m. (ParIVar ivar m, MonadIO m, NFData (ivar S))+            => GlobalState ivar+            -> m (b, GlobalState ivar)+        comp gs+        -- run the action first. in the final monad code for OhuaM, the outermost <*>+        -- will execute first. as a result of this code, we will recursively go and+        -- spawn the tasks for the arguments which can happily execute in parallel+        -- until we reach the bottom of the recursion, i.e., the pure function.+        -- then the recursion unwinds again gathering all the results.+         = do+            aVar <- PC.spawn_ $ runOhua a gs -- TODO force evaluation here+        -- run the function+            (fResult, _) <- runOhua f gs+        -- wrap it up by applying the function to the result of the action+            (r, gs') <- PC.get aVar+            return (fResult r, gs')+  -- mf@(OhuaM _) <*> mv@(OhuaM _) = Collected mf [mv]+  -- mf@(OhuaM _) <*> (Collected pf sfs) = Collected mf (pf : sfs)+  -- (Collected pf sfs) <*> mv@(OhuaM sf) = Collected pf sfs ++ [mv]+  -- (Collected pf1 sfs1) <*> (Collected pf2 sfs2) = Collected pf1 (sfs1 ++ (pf2:sfs2))+  --  -- this collecting is only stopped by the monadic bind operator!++instance Monad OhuaM+  --{-# NOINLINE return #-}+                            where+    return :: forall a. a -> OhuaM a+    return v = OhuaM return $ \s -> return (v, s)+    {-# NOINLINE (>>=) #-}+    (>>=) :: HasCallStack => OhuaM a -> (a -> OhuaM b) -> OhuaM b+    f >>= g =+            OhuaM moveState comp+          where+        moveState ::+               forall ivar m. (ParIVar ivar m, MonadIO m, HasCallStack)+            => GlobalState ivar+            -> m (GlobalState ivar)+        moveState gs = do+            gs' <- moveStateForward f gs+            flip moveStateForward gs' $+                g $+                error+                    "Invariant broken: Don't touch me, state forward moving code!"+      -- comp ::+      --      forall ivar m. (ParIVar ivar m, MonadIO m, NFData (ivar S))+      --   => GlobalState ivar+      --   -> m (b, GlobalState ivar)+        comp gs+          -- there is no need to spawn here!+          -- pipeline parallelism is solely created by smap.+          -- task-level parallelism is solely created by <*>+         = do+            (result0, gs') <- runOhua f gs+            (result1, gs'') <- runOhua (g result0) gs'+            return (result1, gs'')+        {-# INLINE comp #-}++instance MonadIO OhuaM where+    liftIO :: IO a -> OhuaM a+    liftIO ioAction = OhuaM return $ \s -> (, s) <$> liftIO ioAction+    {-# INLINE liftIO #-}++--{-# NOINLINE liftWithIndex #-}+{-# INLINE liftWithIndex #-}+liftWithIndex ::+       (NFData a, Show a, NFData s, Typeable s)+    => Int+    -> SF s a b+    -> a+    -> OhuaM b+liftWithIndex = liftWithIndexS++--liftWithIndex i f d = liftWithIndex' i $ f d+liftWithIndexS ::+       forall a s b. (Show a, NFData s, Typeable s, NFData a)+    => Int+    -> SF s a b+    -> a+    -> OhuaM b+liftWithIndexS i f d = OhuaM (moveState d) (compAndMoveState $ f d)+  where+    compAndMoveState ::+           forall ivar m a. (ParIVar ivar m, MonadIO m)+        => SFM s a+        -> GlobalState ivar+        -> m (a, GlobalState ivar)+    compAndMoveState sf (GlobalState gsIn gsOut)+      -- we define the proper order on the private state right here!+     = do+        let ithIn = gsIn !! i+            ithOut = gsOut !! i+      -- if we do not deepseq here then a previous parallel stage will get+      -- serialized at this point because the monadic operation will always+      -- be evaluated first and then the computation that computes the input+      -- for this algo.+        d `deepseq` pure ()+        localState <- getState ithIn -- this synchronizes access to the local state+        (d', localState') <- liftIO $ runSF sf $ fromS localState+        release ithOut $ toS localState'+        return (d', GlobalState gsIn gsOut)+    moveState ::+           forall ivar m a. (ParIVar ivar m, MonadIO m)+        => a+        -> GlobalState ivar+        -> m (GlobalState ivar)+    moveState token (GlobalState gsIn gsOut) = do+        let ithIn = gsIn !! i+            ithOut = gsOut !! i+        localState <- getState ithIn+        (_, localState') <- return (d, localState) -- id+        release ithOut localState'+      -- I'd love to be able to do something like this, but I can't catch exceptions here.+      -- release ithOut localState'  `catch` \e@ErrorCall{} ->+      --     if isMultiplePutErr e+      --     then error $ "Double use of index " ++ show i ++ " detected"+      --     else throw e+        return $ GlobalState gsIn gsOut+    idSf :: SFM s ()+    idSf = return ()+    {-# INLINE idSf #-}+    -- This match is extracted from the `shed` function in+    -- `Control.Monad.Par.Scheds.TraceInternal`+    -- isMultiplePutErr (ErrorCall msg) = msg == "multiple put"++{-# INLINE liftWithIndex' #-}+liftWithIndex' ::+       forall s b. (NFData s, Typeable s)+    => Int+    -> SFM s b+    -> OhuaM b+liftWithIndex' i comp =+    OhuaM (fmap snd . compAndMoveState idSf) (compAndMoveState comp)+  where+    compAndMoveState ::+           forall ivar m a. (ParIVar ivar m, MonadIO m)+        => SFM s a+        -> GlobalState ivar+        -> m (a, GlobalState ivar)+    compAndMoveState sf (GlobalState gsIn gsOut)+      -- we define the proper order on the private state right here!+     = do+        let ithIn = gsIn !! i+            ithOut = gsOut !! i+        localState <- getState ithIn -- this synchronizes access to the local state+        (d', localState') <- liftIO $ runSF sf $ fromS localState+        release ithOut $ toS localState'+        return (d', GlobalState gsIn gsOut)+    idSf :: SFM s ()+    idSf = return ()+    {-# INLINE idSf #-}++--{-# NOINLINE release #-}+release :: (NFData s, ParIVar ivar m) => ivar s -> s -> m ()+release = updateState++{-# INLINE release #-}+{-# INLINE updateState #-}+{-# INLINE getState #-}+updateState :: (NFData s, ParIVar ivar m) => ivar s -> s -> m ()+updateState = PC.put++getState :: (ParFuture ivar m) => ivar s -> m s+getState = PC.get -- will wait for the value+#ifdef DEBUG_SCHED+-- for debugging the scheduler+runParComp = TDB.runParIO+#else+runParComp = runParIO+#endif+runOhuaM :: (NFData a) => OhuaM a -> [S] -> IO (a, [S])+runOhuaM comp initialState =+    runParComp $ do+        inState <- mapM PC.newFull initialState+        outState <- forM initialState $ const PC.new+        (result, _) <- runOhua comp $ GlobalState inState outState+        finalState <- mapM getState outState+        return (result, finalState)+-- envisioned API:+--+-- s1 = liftWithIndex 5 $ \ x -> ....+-- OhuaM ..+-- do+--   r0 <- a x+--   r1 <- b x+--   r2 <- c x+--   xs <- d r2+--   <- smap c xs+--+--   where c x = do+--                r01 <- e x+--                r02 <- f r01+--                return r02+--+-- runOhua m s

@@ -0,0 +1,48 @@+module Control.Monad.SD.STCLang+    ( STCLang+    , liftWithState+    , runSTCLang+    , CollSt(..)+    , smapSTC+    ) where++import Control.Monad.SD.Ohua+import Control.Monad.SD.Smap+import Data.Dynamic2+import Data.StateElement++import Control.DeepSeq (NFData)+import Control.Monad.State as S++data CollSt = CollSt+    { states :: [S]+    , signals :: [IO S]+    }++instance Monoid CollSt where+    mempty = CollSt [] []+    CollSt st1 si1 `mappend` CollSt st2 si2 =+        CollSt (st1 `mappend` st2) (si1 `mappend` si2)++type STCLang a b = StateT CollSt IO (a -> OhuaM b)++liftWithState ::+       (Typeable s, NFData a, NFData s, Show a)+    => IO s+    -> (a -> StateT s IO b)+    -> STCLang a b+liftWithState state stateThread = do+    s0 <- lift state+    l <- S.state $ \s -> (length $ states s, s {states = states s ++ [toS s0]})+    pure $ liftWithIndex l stateThread++runSTCLang :: (NFData b) => STCLang a b -> a -> IO (b, [S])+runSTCLang langComp a = do+    (comp, gs) <- S.runStateT langComp mempty+    runOhuaM (comp a) $ states gs++smapSTC ::+       forall a b. (NFData b, Show a)+    => STCLang a b+    -> STCLang [a] [b]+smapSTC comp = smap <$> comp

@@ -0,0 +1,215 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE CPP #-}++module Control.Monad.SD.Smap+    ( smap+    , smapGen+    ) where++import Control.Monad+import Control.Monad.Generator+import Control.Monad.IO.Class+import Control.Monad.Par.Class as PC+import Control.Monad.SD.Ohua+import Data.StateElement++-- FIXME this should be based on smapGen!+-- this spawns the computations for the elements but integrates the+-- state dependencies!+-- version used for debugging:+-- smap :: (NFData b, NFData s, Show a, ParIVar ivar m, NFData (ivar s)) => (Int -> a -> OhuaM m (GlobalState ivar s) b) -> [a] -> OhuaM m (GlobalState ivar s) [b]+--{-# NOINLINE smap #-}+--{-# INLINE smap #-}+smap ::+       forall a b. (NFData b, Show a)+    => (a -> OhuaM b)+    -> [a]+    -> OhuaM [b]+smap algo xs =+    case xs of+        [] -> OhuaM moveState (fmap (([] :: [b]), ) . moveState) -- if no data was given then just move the state.+        _ -> OhuaM moveState comp+    -- all we need to do is to move the state once, no need to do it for each+    -- of the elements in the array!+  where+    moveState ::+           forall ivar m. (ParIVar ivar m, MonadIO m)+        => GlobalState ivar+        -> m (GlobalState ivar)+    moveState = moveStateForward $ algo (error "I do not want to be touched!")+    comp ::+           forall ivar m. (ParIVar ivar m, MonadIO m, NFData (ivar S))+        => GlobalState ivar+        -> m ([b], GlobalState ivar)+    comp (GlobalState gsIn gsOut) = do+        futures <- smap' algo gsOut gsIn xs+        results <- forM futures PC.get -- collect the results+        let result = map fst results+        return (result, GlobalState gsIn gsOut)+    -- This function replicates the state as many times as their are values in+    -- the list and spawns the computation.+    smap' ::+           (NFData b, Show a, ParIVar ivar m, MonadIO m, NFData (ivar S))+        => (a -> OhuaM b)+        -> [ivar S]+        -> [ivar S]+        -> [a]+        -> m [ivar (b, GlobalState ivar)]+    smap' f originalOut initialState = go initialState+      where+        newEmptyStateVec = sequence $ replicate stateVSize PC.new -- create the new output state+        stateVSize = length initialState+        go prevState l =+            case l of+                [] -> error "I should be unreachable"+                [y] -> pure <$> spawnComp y originalOut+                (y:ys) -> do+                    stateVec <- newEmptyStateVec+                    (:) <$> spawnComp y stateVec <*> go stateVec ys+          where+            spawnComp e stateVec =+                PC.spawn $ runOhua (f e) $ GlobalState prevState stateVec++type AlgoRunner m ivar t result+     --(ParIVar ivar m, MonadIO m, MonadIO ivar) =>+     = t -> [ivar S] -> [ivar S] -> m (ivar (result, GlobalState ivar))++type PipelineStrategy a b+     = forall m ivar. (ParIVar ivar m, MonadIO m) =>+                          AlgoRunner m ivar a b -- algo runner+                           -> Int -- state vector size+                               -> [ivar S] -- final state vector+                                   -> [ivar S] -- current state vector+                                       -> Generator IO a -> a -> m [ivar ( b+                                                                         , GlobalState ivar)]++-- TODO: Check if this can deal with empty generators. Furthermore+-- it always advances the generator one position more than what it+-- currently processes to find the end of the generator before the+-- last item is processed so that it can spawn that computation with+-- the original output state vector.+smapGen ::+       forall a b. (NFData b, Show a)+    => (a -> OhuaM b)+    -> Generator IO a+    -> OhuaM [b]+#ifdef UNTHROTTLED+smapGen = smapGenInternal unthrottledPipe+#else+smapGen = smapGenInternal throttledPipe+#endif+smapGenInternal ::+       forall a b. (NFData b, Show a)+    => PipelineStrategy a b+    -> (a -> OhuaM b)+    -> Generator IO a+    -> OhuaM [b]+smapGenInternal pipelineStrategy algo gen =+    OhuaM moveState $ \g@(GlobalState gsIn gsOut) ->+        liftIO (step gen) >>= \case+            Nothing -> fmap (([] :: [b]), ) $ moveState g+            Just (a, gen') -> do+                futures <- spawnFutures gsOut gsIn gen' a+                values <- mapM PC.get futures+                pure (map fst values, GlobalState gsIn gsOut)+  where+    spawnFutures lastStateOut = pipelineStrategy runAlgo stateVSize lastStateOut+      where+        stateVSize = length lastStateOut+        runAlgo e stateIn stateOut =+            PC.spawn $ runOhua (algo e) $ GlobalState stateIn stateOut+    moveState ::+           forall ivar m. (ParIVar ivar m, MonadIO m)+        => GlobalState ivar+        -> m (GlobalState ivar)+    moveState = moveStateForward $ algo (undefined :: a)++newEmptyStateVec size = sequence $ replicate size PC.new++unthrottledPipe :: PipelineStrategy a b+unthrottledPipe runAlgo stateVSize lastStateOut stateIn gen' a =+    liftIO (step gen') >>= \case+        Nothing -> pure <$> runLastAlgo+        Just (a', gen'') -> do+            newStateVec <- newEmptyStateVec stateVSize+      -- the parallelism is in the applicative.+      -- runAlgo immediately returns and gives me an IVar.+      -- go is the recursion.+      -- I need to change `go` to take the current list of IVars.+      -- Then a simple version of throttling becomes totally easy.+      -- I just need to check the length of the list and once it has+      -- reached the predefined threshold, I need to stop and wait for+      -- the IVar at the head of the list before contiuing to spawn.+      -- (This assumes that the head is the one finishing first.)+            (:) <$> runAlgo a stateIn newStateVec <*>+                unthrottledPipe+                    runAlgo+                    stateVSize+                    lastStateOut+                    newStateVec+                    gen''+                    a'+  where+    runLastAlgo = runAlgo a stateIn lastStateOut++limit :: Int+limit = 10++throttledPipe :: PipelineStrategy a b+throttledPipe runAlgo stateVSize lastStateOut stateIn gen a+    -- 1. get the first n+ = do+    (genLimited, a', lastLimitOut, firstResults) <-+        unthrottled limit [] stateIn gen a+    -- 2. get on head of results before spawning a new computation+    throttled firstResults 0 lastLimitOut genLimited a'+    -- unthrottled ::+    --      Int+    --   -> [ivar (b, GlobalState ivar)]+    --   -> [ivar S]+    --   -> [ivar S]+    --   -> Generator IO a+    --   -> a+    --   -> m (Generator IO a, a, [ivar S], [ivar (b, GlobalState ivar)])+  where+    unthrottled l results sIn gen' a' = do+        if l == 0+            then return (gen', a', sIn, results)+            else do+                liftIO (step gen') >>= \case+                    Nothing -> do+                        res <- runAlgo a' sIn lastStateOut+              -- from now on the generator always returns NOTHING, so it is+              -- ok to use it as the state input vector to the next iteration.+                        return (gen', a', lastStateOut, results ++ [res])+                    Just (a'', gen'') -> do+                        newStateVec <- newEmptyStateVec stateVSize+                        resultFuture <- runAlgo a' sIn newStateVec+                        unthrottled+                            (l - 1)+                            (results ++ [resultFuture])+                            newStateVec+                            gen''+                            a''+    -- throttled ::+      --    [ivar (b, GlobalState ivar)]+      -- -> Int+      -- -> [ivar S]+      -- -> Generator IO a+      -- -> a+      -- -> m [ivar (b, GlobalState ivar)]+    throttled results lastPending sIn gen' a' = do+        _ <- PC.get $ results !! lastPending -- throttling+        liftIO (step gen') >>= \case+            Nothing -> do+                res <- runAlgo a' sIn lastStateOut+                return $ results ++ [res]+            Just (a'', gen'') -> do+                newStateVec <- newEmptyStateVec stateVSize+                ivar <- runAlgo a' sIn newStateVec+                throttled+                    (results ++ [ivar])+                    (lastPending + 1)+                    newStateVec+                    gen''+                    a''

@@ -0,0 +1,16 @@+{-# LANGUAGE TypeFamilies #-}+module Control.Monad.Stream where++import Control.Monad.IO.Class++class MonadIO m =>+      MonadStream m+    where+    type Sender m :: * -> *+    type Reciever m :: * -> *+    -- | Create a stream with a end that can only be sent to and one+    -- that can only be read from.+    createStream :: m (Sender m a, Reciever m a)+    send :: a -> Sender m a -> m ()+    recieve :: Reciever m a -> m a+    spawn :: m () -> m ()

@@ -0,0 +1,24 @@+{-# LANGUAGE TupleSections, TypeFamilies, RankNTypes, GeneralizedNewtypeDeriving #-}+module Control.Monad.Stream.Chan where++import Control.Concurrent.Chan+import Control.Concurrent+import Control.Arrow ((&&&))+import Control.Monad (void)+import Control.Monad.IO.Class++import Control.Monad.Stream++newtype ChanM a = ChanM+    { runChanM :: (IO a)+    } deriving (Functor, Applicative, Monad, MonadIO)++newtype S a = S { unS :: a -> ChanM () }++instance MonadStream ChanM where+    type Sender ChanM = S+    type Reciever ChanM = ChanM+    createStream = (S . (liftIO .) . writeChan &&& liftIO . readChan) <$> liftIO newChan+    send a f = (unS f) a+    recieve ac = ac+    spawn = ChanM . void . forkIO . runChanM

@@ -0,0 +1,53 @@+{-# LANGUAGE TypeFamilies #-}+module Control.Monad.Stream.Par+    ( S+    , ParIO+    , runParIO+    ) where++import Control.Monad.Par.IO as P+import Control.Monad.Par.Class as P+import Data.IORef+import Control.Monad.IO.Class++import Control.Monad.Stream++data Cons a = Cons a (IVar (Cons a)) ++newtype S a = S { unS :: IORef (IVar (Cons a)) }++instance MonadStream ParIO where+    type Sender ParIO = S+    type Reciever ParIO = S+    createStream = do+        v <- new+        in_ <- liftIO $ newIORef v+        out <- liftIO $ newIORef v+        pure (S in_, S out)++    -- This implementation is inherently unsafe. We use IORef and+    -- non-atomic operations to update them. This only works if the+    -- sender and receiver are never accessed simultaneously from two+    -- threads, which shouldn't happen because of the way the runtime+    -- is written. It could be made safe by using a Maybe and atomic+    -- operations, however I expect this will cause slowdown which, as+    -- the runtime should make sure this thing is impossible anyways,+    -- I am not willing to risk.+    send v =+        withS $ \var -> do+            next <- new+            put_ var $ Cons v next+            pure ((), next)+    recieve =+        withS $ \var -> do+            Cons v next <- get var+            pure (v, next)+    spawn = P.fork+++withS :: MonadIO m => (IVar (Cons a) -> m (b, IVar (Cons a))) -> S a -> m b+withS ac (S ref) = do+    var <- liftIO $ readIORef ref+    (val, var') <- ac var+    liftIO $ writeIORef ref var'+    pure val

@@ -0,0 +1,40 @@+{-# LANGUAGE TupleSections, TypeFamilies, RankNTypes, InstanceSigs, FlexibleInstances #-}+module Control.Monad.Stream.PinnedChan where++import Control.Concurrent.Chan+import Control.Concurrent+import Control.Arrow ((&&&))+import Control.Monad.IO.Class++import Control.Monad.Stream++import Control.Monad.State.Lazy++type PChanM = StateT Int IO++newtype S a = S { unS :: a -> PChanM () }++instance MonadStream PChanM where+    type Sender PChanM = S+    -- FIXME Receiver+    type Reciever PChanM = PChanM++    createStream :: PChanM (Sender PChanM a, Reciever PChanM a)+    createStream = (S . (liftIO .) . writeChan &&& liftIO . readChan) <$> liftIO newChan++    send :: a -> Sender PChanM a -> PChanM ()+    send a f = (unS f) a++    -- FIXME receive+    recieve :: Reciever PChanM a -> PChanM a+    recieve ac = ac++    spawn :: PChanM () -> PChanM ()+    spawn comp = do+      let ioComp = evalStateT comp 0+      cap    <- get+      defCap <- liftIO $ getNumCapabilities+      -- liftIO $ putStrLn $ "forking on cap num: " ++ (show (cap `mod` defCap) )+      _      <- liftIO $ forkOn cap ioComp+      put $ cap + 1+      return ()

@@ -0,0 +1,167 @@+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE CPP #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Data.Dynamic+-- Copyright   :  (c) The University of Glasgow 2001+-- License     :  BSD-style (see the file libraries/base/LICENSE)+--+-- Maintainer  :  libraries@haskell.org+-- Stability   :  experimental+-- Portability :  portable+--+-- The Dynamic interface provides basic support for dynamic types.+--+-- Operations for injecting values of arbitrary type into+-- a dynamically typed value, Dynamic, are provided, together+-- with operations for converting dynamic values into a concrete+-- (monomorphic) type.+--+-----------------------------------------------------------------------------+module Data.Dynamic2+        -- Module Data.Typeable re-exported for convenience+    ( module Data.Typeable+        -- * The @Dynamic@ type+    , Dynamic(..) -- abstract, instance of: Show, Typeable+        -- * Converting to and from @Dynamic@+    , toDyn+    , fromDyn+    , fromDynamic+        -- * Applying functions of dynamic type+    , dynApply+    , dynApp+    , dynTypeRep+    , forceDynamic+    , TypeCastException(..)+    ) where++import Data.Maybe+import Data.Typeable+import Unsafe.Coerce++import GHC.Base+import GHC.Exception+import GHC.Show++-------------------------------------------------------------+--+--              The type Dynamic+--+-------------------------------------------------------------+{-|+  A value of type 'Dynamic' is an object encapsulated together with its type.++  A 'Dynamic' may only represent a monomorphic value; an attempt to+  create a value of type 'Dynamic' from a polymorphically-typed+  expression will result in an ambiguity error (see 'toDyn').++  'Show'ing a value of type 'Dynamic' returns a pretty-printed representation+  of the object\'s type; useful for debugging.+-}+data Dynamic =+    Dynamic TypeRep+            Obj++instance Show Dynamic+   -- the instance just prints the type representation.+                                                        where+    showsPrec _ (Dynamic t _) =+        showString "<<" . showsPrec 0 t . showString ">>"++-- here so that it isn't an orphan:+instance Exception Dynamic++type Obj = Any+ -- Use GHC's primitive 'Any' type to hold the dynamically typed value.+ --+ -- In GHC's new eval/apply execution model this type must not look+ -- like a data type.  If it did, GHC would use the constructor convention+ -- when evaluating it, and this will go wrong if the object is really a+ -- function.  Using Any forces GHC to use+ -- a fallback convention for evaluating it that works for all types.++-- | Converts an arbitrary value into an object of type 'Dynamic'.+--+-- The type of the object must be an instance of 'Typeable', which+-- ensures that only monomorphically-typed objects may be converted to+-- 'Dynamic'.  To convert a polymorphic object into 'Dynamic', give it+-- a monomorphic type signature.  For example:+--+-- >    toDyn (id :: Int -> Int)+--+toDyn :: Typeable a => a -> Dynamic+toDyn v = Dynamic (typeOf v) (unsafeCoerce v)++-- | Converts a 'Dynamic' object back into an ordinary Haskell value of+-- the correct type.  See also 'fromDynamic'.+fromDyn ::+       Typeable a+    => Dynamic -- ^ the dynamically-typed object+    -> a -- ^ a default value+    -> a -- ^ returns: the value of the first argument, if+                        -- it has the correct type, otherwise the value of+                        -- the second argument.+fromDyn (Dynamic t v) def+    | typeOf def == t = unsafeCoerce v+    | otherwise = def++-- | Converts a 'Dynamic' object back into an ordinary Haskell value of+-- the correct type.  See also 'fromDyn'.+fromDynamic ::+       Typeable a+    => Dynamic -- ^ the dynamically-typed object+    -> Maybe a -- ^ returns: @'Just' a@, if the dynamically-typed+                        -- object has the correct type (and @a@ is its value),+                        -- or 'Nothing' otherwise.+fromDynamic (Dynamic t v) =+    case unsafeCoerce v of+        r+            | t == typeOf r -> Just r+            | otherwise -> Nothing++-- (f::(a->b)) `dynApply` (x::a) = (f a)::b+dynApply :: Dynamic -> Dynamic -> Maybe Dynamic+dynApply (Dynamic t1 f) (Dynamic t2 x) =+    case funResultTy t1 t2 of+        Just t3 -> Just (Dynamic t3 ((unsafeCoerce f) x))+        Nothing -> Nothing++dynApp :: Dynamic -> Dynamic -> Dynamic+dynApp f x =+    case dynApply f x of+        Just r -> r+        Nothing ->+            errorWithoutStackTrace+                ("Type error in dynamic application.\n" +++                 "Can't apply function " ++ show f ++ " to argument " ++ show x)+#if !MIN_VERSION_base(4,9,0)+errorWithoutStackTrace = error+#endif+dynTypeRep :: Dynamic -> TypeRep+dynTypeRep (Dynamic tr _) = tr++data TypeCastException =+    TypeCastException TypeRep+                      TypeRep+    deriving (Typeable)++instance Show TypeCastException where+    show (TypeCastException expected recieved) =+        "TypeCastexception: Expected " +++        show expected ++ " got " ++ show recieved++instance Exception TypeCastException++-- | Coerce a dynamic to a value.+-- If the expected type is not the one inside the 'Dynamic' it throws an error showing both types.+forceDynamic ::+       forall a. Typeable a+    => Dynamic+    -> a+forceDynamic dyn+    | Just a <- fromDynamic dyn = a+    | otherwise = throw $ TypeCastException rep (dynTypeRep dyn)+  where+    rep = typeRep (Proxy :: Proxy a)

@@ -0,0 +1,30 @@+{-# LANGUAGE InstanceSigs #-}++module Data.StateElement where++import Data.Dynamic2++import Control.DeepSeq++--+-- Support for heterogeneous lists.+--+data S =+    forall a. Typeable a =>+              S (a -> ())+                Dynamic++toS :: forall a. (Typeable a, NFData a)+    => a+    -> S+toS a = S rnf' (toDyn a)+  where+    rnf' :: a -> ()+    rnf' = rnf++fromS :: Typeable a => S -> a+fromS (S _ a) = forceDynamic a++instance NFData S where+    rnf :: S -> ()+    rnf (S toRnf d) = toRnf $ forceDynamic d

@@ -0,0 +1,13 @@+{-# LANGUAGE CPP #-}++module Type.Magic+    ( injectList+    , extractList+    , extractFunctor+    , injectFunctor+    ) where+#if MIN_VERSION_base(4,10,0)+import Type.Magic.GHC8 as X+#else+import Type.Magic.OldGHC as X+#endif

@@ -0,0 +1,59 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE TypeApplications #-}++module Type.Magic.GHC8+    ( injectList+    , extractFunctor+    , extractList+    , injectFunctor+    ) where++import Control.Exception+import Data.Dynamic2 (Dynamic(..), TypeCastException(..))+import Data.Kind+import Type.Reflection+import Unsafe.Coerce++extractFunctor ::+       forall f. (Typeable f, Functor f)+    => Dynamic+    -> f Dynamic+extractFunctor =+    \case+        Dynamic (SomeTypeRep (App con tra)) dl ->+            case con `eqTypeRep` targetRep of+                Just HRefl -> fmap f $ unsafeCoerce dl+                Nothing ->+                    throw $+                    TypeCastException (SomeTypeRep con) (SomeTypeRep targetRep)+            where f = Dynamic (SomeTypeRep tra)+        Dynamic tr _ -> throw $ TypeCastException tr (SomeTypeRep targetRep)+  where+    targetRep = typeRep @f++extractList :: Dynamic -> [Dynamic]+extractList = extractFunctor++injectFunctor ::+       forall f. (Typeable f, Functor f)+    => Dynamic+    -> f Dynamic+    -> Dynamic+injectFunctor (Dynamic (SomeTypeRep tra) _) l =+    Dynamic tr $ unsafeCoerce $ fmap unwrap l+  where+    unwrap (Dynamic _ v) = v+    tr =+        case traKind `eqTypeRep` kindStar of+            Just HRefl -> SomeTypeRep $ App (typeRep @f) tra+            Nothing ->+                throw $+                TypeCastException (SomeTypeRep traKind) (SomeTypeRep kindStar)+      where+        traKind = typeRepKind tra+        kindStar = typeRep @Type++injectList :: [Dynamic] -> Dynamic+injectList [] = error "cannot convert empty list"+injectList l@(x:_) = injectFunctor x l

@@ -0,0 +1,52 @@+module Type.Magic.OldGHC+    ( extractList+    , injectList+    , extractFunctor+    , injectFunctor+    ) where++import Control.Exception+import Data.Dynamic2 (Dynamic(..), TypeCastException(..))+import Data.Typeable+import Unsafe.Coerce++extractFunctor ::+       forall f. (Typeable f, Functor f)+    => Dynamic+    -> f Dynamic+extractFunctor =+    \(Dynamic trl dl) ->+        let (tyCon, tyArgs) = splitTyConApp trl+         in if tyCon == expectedTyCon+                then case tyArgs of+                         [] ->+                             error "Constructor must be at least of kind * -> *"+                         types -> fmap f l+                             where f = Dynamic $ last types+                                   l = unsafeCoerce dl+                         _ ->+                             error $ "Wrong kind for constructor " ++ show tyCon+                else throw $ TypeCastException expectedTy (mkTyConApp tyCon [])+  where+    !expectedTy = typeRep (undefined :: Proxy f)+    !expectedTyCon = typeRepTyCon expectedTy++extractList :: Dynamic -> [Dynamic]+extractList = extractFunctor++injectFunctor ::+       forall f. (Typeable f, Functor f)+    => Dynamic+    -> f Dynamic+    -> Dynamic+injectFunctor =+    \(Dynamic tra _) l -> Dynamic (mkTy tra) $ unsafeCoerce $ fmap unwrap l+  where+    unwrap (Dynamic _ v) = v+    !targetTyRep = typeRep (undefined :: Proxy f)+    !(!con, !args) = splitTyConApp targetTyRep+    mkTy tra = mkTyConApp con (args ++ [tra])++injectList :: [Dynamic] -> Dynamic+injectList [] = error "Cannot convert empty list yet"+injectList l@(x:_) = injectFunctor x l

@@ -0,0 +1,140 @@+cabal-version: 1.12+name: stc-lang+version: 1.0.0+license: BSD3+license-file: LICENSE+copyright: 2017-2019 Norman Rink, Sebastian Ertel, Justus Adam+maintainer: sebastian.ertel@tu-dresden.de+author: Norman Rink, Sebastian Ertel, Justus Adam+homepage: https://github.com/ohua-dev/stc-lang#readme+synopsis: A library for implicit, monadic dataflow parallelism+description:+    See the <https://github.com/ohua-dev/stc-lang#readme readme>+category: Concurrency, Development+build-type: Simple+extra-source-files:+    README.md++source-repository head+    type: git+    location: https://github.com/ohua-dev/stc-lang++flag debug-sched+    description:+        Enable the debuggable scheduler+    default: False+    manual: True++library+    exposed-modules:+        Control.Monad.SD+        Data.Dynamic2+        Data.StateElement+        Type.Magic+        Control.Monad.Generator+        Control.Monad.Stream+        Control.Monad.Stream.Chan+        Control.Monad.Stream.Par+        Control.Monad.Stream.PinnedChan+    hs-source-dirs: src+    other-modules:+        Control.Monad.SD.Case+        Control.Monad.SD.Combinator+        Control.Monad.SD.FRP+        Control.Monad.SD.Ohua+        Control.Monad.SD.STCLang+        Control.Monad.SD.Smap+    default-language: Haskell2010+    default-extensions: DeriveGeneric ExistentialQuantification+                        ExplicitForAll FlexibleContexts FlexibleInstances+                        ScopedTypeVariables TupleSections LambdaCase RankNTypes+                        NamedFieldPuns MultiParamTypeClasses RecordWildCards+                        TypeSynonymInstances BangPatterns DeriveFunctor RecordWildCards+    ghc-options: -Wall -O2 -fPIC -fno-cse+    build-depends:+        BoundedChan >=1.0.3.0,+        abstract-par >=0.3.3,+        base >=4.7 && <5,+        bytestring >=0.10.8.2,+        deepseq >=1.4.3.0,+        microlens >=0.4.8.3,+        monad-par >=0.3.4.8,+        monad-par-extras >=0.3.3,+        mtl >=2.2.1,+        transformers >=0.5.2.0+    +    if flag(debug-sched)+        cpp-options: -DDEBUG_SCHED+    +    if impl(ghc >=8.0.0)+        other-modules:+            Type.Magic.GHC8+    else+        other-modules:+            Type.Magic.OldGHC++executable ohua-stream-bench+    main-is: algo.hs+    hs-source-dirs: stream-bench+    other-modules:+        CampaignProcMap+        MutableNFMap+        MutableSet+        Paths_stc_lang+    default-language: Haskell2010+    default-extensions: DeriveGeneric ExistentialQuantification+                        ExplicitForAll FlexibleContexts FlexibleInstances+                        ScopedTypeVariables TupleSections LambdaCase RankNTypes+                        NamedFieldPuns MultiParamTypeClasses RecordWildCards+                        TypeSynonymInstances BangPatterns DeriveFunctor RecordWildCards+    ghc-options: -Wall -O2 -threaded -with-rtsopts=-N+    build-depends:+        BoundedChan >=1.0.3.0,+        aeson >=1.2.3.0,+        base >=4.7 && <5,+        bytestring >=0.10.8.2,+        clock >=0.7.2,+        deepseq >=1.4.3.0,+        hashable >=1.2.6.1,+        hashtables >=1.2.2.1,+        hedis >=0.9.12,+        hw-kafka-client >=2.6.0,+        microlens >=0.4.8.3,+        microlens-aeson >=2.2.0.2,+        mtl >=2.2.1,+        random >=1.1,+        stc-lang -any,+        text >=1.2.2.2,+        transformers >=0.5.2.0,+        uuid-types >=1.0.3,+        vector >=0.12.0.1,+        yaml >=0.8.28++test-suite statefulness-test+    type: exitcode-stdio-1.0+    main-is: Spec.hs+    hs-source-dirs: test+    other-modules:+        FakeComputation+        SD.Correctness+        SD.Performance+        Paths_stc_lang+    default-language: Haskell2010+    default-extensions: DeriveGeneric ExistentialQuantification+                        ExplicitForAll FlexibleContexts FlexibleInstances+                        ScopedTypeVariables TupleSections LambdaCase RankNTypes+                        NamedFieldPuns MultiParamTypeClasses RecordWildCards+                        TypeSynonymInstances BangPatterns DeriveFunctor RecordWildCards+    ghc-options: -Wall -O2 -threaded -rtsopts -with-rtsopts=-N4+    build-depends:+        HUnit >=1.6.0.0,+        base >=4.7 && <5,+        deepseq >=1.4.3.0,+        ghc-prim >=0.5.1.1,+        microlens >=0.4.8.3,+        mtl >=2.2.1,+        stc-lang -any,+        test-framework >=0.8.1.1,+        test-framework-hunit >=0.3.0.2,+        time >=1.8.0.2,+        transformers >=0.5.2.0

@@ -0,0 +1,41 @@+{-# LANGUAGE ConstraintKinds #-}++module CampaignProcMap+    ( Map+    , new+    , insert+    , mapM_+    ) where++import Control.DeepSeq+import Control.Monad.IO.Class+import qualified Data.HashTable.IO as MHT+import Data.Hashable (Hashable)+import qualified MutableSet as Set+import Prelude hiding (mapM_)++type Constraint a = (Hashable a, Eq a)++newtype Map k v = Map+    { unwrap :: MHT.BasicHashTable k (Set.Set v)+    }++-- This may seem like an odd instance for NFData, but as with the `HashSet` type+-- my reasoning is that 'HashTable' is strict in the keys, thus we don't need to+-- force them. And since the values are 'HashSet', which is also completely+-- strict already we don't need to specially evaluate it.+instance NFData (Map k v) where+    rnf _ = ()++new :: IO (Map k v)+new = Map <$> MHT.new++insert ::+       (Constraint k, Set.Constraint v, MonadIO m) => k -> v -> Map k v -> m ()+insert k v m =+    liftIO $ do+        set <- maybe Set.new pure =<< MHT.lookup (unwrap m) k+        Set.insert v set++mapM_ :: MonadIO m => ((k, Set.Set v) -> IO a) -> Map k v -> m ()+mapM_ f = liftIO . MHT.mapM_ f . unwrap

@@ -0,0 +1,48 @@+{-# LANGUAGE ConstraintKinds #-}++module MutableNFMap+    ( Map+    , new+    , insert+    , delete+    , lookup+    , mapM_+    , size+    ) where++import Control.DeepSeq+import Control.Monad.IO.Class+import qualified Data.HashTable.IO as MHT+import Data.Hashable (Hashable)+import Prelude hiding (lookup, mapM_)++type Constraint a = (Hashable a, Eq a)++newtype Map k v = Map+    { unwrap :: MHT.BasicHashTable k v+    }++-- | This instance does nothing, because the functions exposed force the keys+-- and values automatically+instance NFData (Map k v) where+    rnf _ = ()++new :: MonadIO m => m (Map k v)+new = liftIO $ Map <$> MHT.new++-- | I only require an 'NFData' instance for the values, because the assumption+-- is that calculating the hash for the keys will force them.+insert :: (Constraint k, NFData v, MonadIO m) => k -> v -> Map k v -> m ()+insert k v m = liftIO $ v `deepseq` MHT.insert (unwrap m) k v++delete :: (Constraint k, MonadIO m) => k -> Map k v -> m ()+delete k m = liftIO $ MHT.delete (unwrap m) k++lookup :: (Constraint k, MonadIO m) => k -> Map k v -> m (Maybe v)+lookup k m = liftIO $ MHT.lookup (unwrap m) k++mapM_ :: MonadIO m => ((k, v) -> IO a) -> Map k v -> m ()+mapM_ f = liftIO . MHT.mapM_ f . unwrap++size :: MonadIO m => Map k v -> m Word+size = liftIO . MHT.foldM (\a _ -> pure $ a + 1) 0 . unwrap

@@ -0,0 +1,55 @@+{-# LANGUAGE ConstraintKinds #-}++module MutableSet+    ( Set+    , Constraint+    , new+    , delete+    , insert+    , member+    , mapM_+    , size+    , toList+    ) where++import Control.DeepSeq+import Control.Monad.IO.Class (MonadIO, liftIO)+import qualified Data.HashTable.IO as HT+import Data.Hashable (Hashable)+import Data.Maybe (isJust)+import Prelude hiding (mapM_)++type HashSetInner a = HT.BasicHashTable a ()++newtype Set a = Set+    { unwrap :: HashSetInner a+    }++type Constraint a = (Hashable a, Eq a)++new :: MonadIO m => m (Set a)+new = liftIO $ Set <$> HT.new++insert :: (MonadIO m, Constraint a) => a -> Set a -> m ()+insert item t = liftIO $ HT.insert (unwrap t) item ()++delete :: (MonadIO m, Constraint a) => a -> Set a -> m ()+delete item set = liftIO $ HT.delete (unwrap set) item++member :: (MonadIO m, Constraint a) => a -> Set a -> m Bool+member i t = liftIO $ isJust <$> HT.lookup (unwrap t) i++mapM_ :: MonadIO m => (a -> IO b) -> Set a -> m ()+mapM_ f = liftIO . HT.mapM_ (f . fst) . unwrap++size :: MonadIO m => Set a -> m Word+size = liftIO . HT.foldM (\a _ -> pure $ a + 1) 0 . unwrap++toList :: (MonadIO m, Constraint a) => Set a -> m [a]+toList = liftIO . fmap (map fst) . HT.toList . unwrap++-- This is a weird NFData instance, but the assumption is that HashMaps are+-- strict in the keys, because computing the hash forces the key. And since a+-- set does not have non-unit values, the values need not be forced.+instance NFData (Set i) where+    rnf _ = ()

@@ -0,0 +1,803 @@+{-# LANGUAGE ConstraintKinds, TypeApplications, OverloadedStrings,+  PartialTypeSignatures, OverloadedLists, TypeFamilies, MultiWayIf+  #-}++import Control.Concurrent (forkIO, killThread, newEmptyMVar, threadDelay)+import qualified Control.Concurrent as Conc+import qualified Control.Concurrent.BoundedChan as BC+import Control.Concurrent.MVar (putMVar, takeMVar)+import Control.DeepSeq (NFData, deepseq)+import Control.Exception (assert, bracket)+import Control.Monad+    ( (<=<)+    , (>=>)+    , forM+    , forM_+    , forever+    , join+    , replicateM+    , unless+    , void+    , when+    )+import Control.Monad.Generator (Generator, foldlGeneratorT, ioReaderToGenerator)+import Control.Monad.IO.Class (MonadIO(liftIO))+import Control.Monad.List (ListT(ListT), runListT)+import Control.Monad.Reader (ReaderT, ask, lift, runReaderT)+import Control.Monad.SD+import Control.Monad.State.Class as S+    ( MonadState+    , get+    , gets+    , modify+    , put+    , state+    )+import Control.Monad.State.Lazy as S (StateT, runStateT)+import Data.Aeson ((.:), decode)+import qualified Data.Aeson as AE+import qualified Data.Aeson.Types as AE+import Data.Aeson.Types (parseMaybe)+import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy as LBS+import Data.Hashable+import Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef)+import Data.Int (Int64)+import Data.List (nub)+import Data.Maybe (fromJust, fromMaybe, isNothing)+import Data.Monoid ((<>))+import Data.StateElement+import Data.String (IsString)+import Data.Text (Text)+import qualified Data.Text as Tx+import qualified Data.Text.Encoding as Tx+import qualified Data.Text.IO as Tx+import Data.Typeable (Typeable)+import qualified Data.UUID.Types as UUID+import qualified Data.Vector as V+import qualified Data.Vector.Mutable as MV+import qualified Data.Vector.Unboxed as UV+import qualified Data.Vector.Unboxed.Mutable as UMV+import Data.Word (Word32)+import qualified Data.Yaml as Yaml+import qualified Debug.Trace as Debug+import GHC.Exts (IsList, Item)+import GHC.Generics (Generic)+import Lens.Micro+import Lens.Micro.Aeson+import qualified MutableNFMap as NFMap+import qualified MutableSet as Set+import Prelude hiding (String, show)+import qualified Prelude as P+import qualified System.Clock as Clock+import System.Environment+import System.IO (Handle, IOMode(WriteMode), withFile)+import System.IO.Unsafe (unsafePerformIO)+import System.Random (randomIO)+import System.Random (randomIO, randomRIO)+import Text.Printf (hPrintf, printf)++import qualified Database.Redis as Redis++import qualified Kafka.Consumer as K++type String = BS.ByteString++type Message = LBS.ByteString++type Long = Int64++data Window = Window+    { seenCount :: IORef Long+    , timestamp :: Text+    } deriving (Eq, Generic)++instance Hashable Window where+    hashWithSalt s w = hashWithSalt s (timestamp w)++instance NFData Window++instance AE.FromJSON String where+    parseJSON = AE.withText "Expected String" $ pure . Tx.encodeUtf8++data Collector a = Collector+    { counts :: IORef (UMV.IOVector a)+    , lastIndex :: IORef Int+    }++collectorInitialSize :: Int+collectorInitialSize = 400++newCollector :: (MonadIO m, UMV.Unbox a) => m (Collector a)+newCollector =+    liftIO $ Collector <$> (newIORef =<< UMV.unsafeNew collectorInitialSize) <*>+    newIORef 0++push :: (MonadIO m, UMV.Unbox a) => Collector a -> a -> m ()+push Collector {..} i =+    liftIO $ do+        indx <- readIORef lastIndex+        c <- readIORef counts+        let l = UMV.length c+        assert (indx <= l && indx > 0) (pure ())+        c' <-+            if l == indx+                then do+                    new <- UMV.unsafeGrow c l+                    writeIORef counts new+                    pure new+                else pure c+        UMV.write c' indx i+        modifyIORef' lastIndex succ++unsafeRead :: (MonadIO m, UMV.Unbox a) => Collector a -> m (UV.Vector a)+unsafeRead Collector {..} =+    liftIO $ do+        idx <- readIORef lastIndex+        UV.unsafeFreeze . UMV.slice 0 idx =<< readIORef counts++data Statistics = Statistics+    { fetcherEventCount :: Collector Int+    , pipelineEventCount :: Collector Word32+    , redisWritesCount :: Collector Int+    }++initStatistics :: IO Statistics+initStatistics = Statistics <$> newCollector <*> newCollector <*> newCollector++writeStatistics :: MonadIO m => Statistics -> Handle -> m ()+writeStatistics Statistics {..} h =+    liftIO $ do+        hPrintf+            h+            "%15s %15s %15s\n"+            ("Fetcher Events" :: P.String)+            ("Pipeline Events" :: P.String)+            ("Write events" :: P.String)+        fec <- unsafeRead fetcherEventCount+        pec <- unsafeRead pipelineEventCount+        rwc <- unsafeRead redisWritesCount+        UV.forM_ (UV.zip3 fec pec rwc) $ \(fcount, pcount, rcount) ->+            hPrintf h "%-15d %-15d %-15d\n" fcount pcount rcount+        let l1 = UV.length fec+            l2 = UV.length pec+            l3 = UV.length rwc+        unless (l1 == l2 && l2 == l3) $+            printf "Lengths were unequal: fec=%7d pec=%7d rwc=%7d\n" l1 l2 l3++fromRight :: Show a => Either a b -> b+fromRight = either (error . P.show) id++show :: Show a => a -> Text+show = Tx.pack . P.show++timeDivisor :: Long+timeDivisor = 10 * 1000++-- | Timeout for the timer trigger. In the actual benchmarks this is one second.+-- I choose 10 seconds here because in one second only very little happens+timeout :: Int+timeout = 1000 * 1000 * 10++eventGenerationStep = 100 * 1000++kafkaEventCount = 10 * 1000 * 1000++numCampaigns = 100++currentMilliSecs :: IO Long+currentMilliSecs =+    (`div` (1000 * 1000)) . fromIntegral . Clock.toNanoSecs <$>+    Clock.getTime Clock.Realtime++-- | Do this for `kafkaEventCount` many times+generateKafkaEvents :: V.Vector Text -> IO [AE.Value]+generateKafkaEvents ads = do+    startTime <-+        fromIntegral . (* eventGenerationStep) . Clock.toNanoSecs <$>+        Clock.getTime Clock.Realtime+    runListT $ do+        n <- ListT $ pure [0 .. 10000 :: Word]+        userId <- liftIO $ randomIO @UUID.UUID+        pageId <- liftIO $ randomIO @UUID.UUID+        ad <- randomIn ads+        adType <- randomIn adTypes+        eventType <- randomIn eventTypes+        pure $+            AE.object+                [ "user_id" AE..= userId+                , "page_id" AE..= pageId+                , "ad_id" AE..= ad+                , "ad_type" AE..= adType+                , "event_type" AE..= eventType+                , "event_time" AE..= (startTime + (n * 10) + skew + lateBy)+                , "ip_address" AE..= ("1.2.3.4" :: Text)+                ]+  where+    adTypes =+        V.fromList+            ["banner", "modal", "sponsored-search", "mail", "mobile" :: Text]+    eventTypes = V.fromList ["view", "click", "purchase" :: Text]+    skew = 0+    lateBy = 0+    randomIn l = (l V.!) <$> liftIO (randomRIO (0, V.length l - 1))++eventGenerationLoop :: ([Message] -> IO ()) -> IO ()+eventGenerationLoop writer = do+    ads <-+        V.fromList <$> replicateM (10 * numCampaigns) (UUID.toText <$> randomIO)+    --ads <- Tx.lines <$> Tx.readFile "ad-ids.txt"+    forM_ @[] @IO @Word [0,fromIntegral eventGenerationStep .. kafkaEventCount] $ \_ -> do+        evs <- generateKafkaEvents ads+        let evaluated = map AE.encode evs+        evaluated `deepseq` pure ()+        putStrLn "Writing Events"+        writer evaluated++-- NOTE I guessed this number. I have not seen any timeout specification in the+-- benchmark. This is in milliseconds.+kafkaTimeout :: K.Timeout+kafkaTimeout = K.Timeout 4000++-- NOTE I also guessed this number. No idea whether this value is appropriate.+kafkaBatchSize :: K.BatchSize+kafkaBatchSize = K.BatchSize 400++readKafka :: K.KafkaConsumer -> Statistics -> IO [Message]+readKafka con stats = do+    subsState <- fromRight <$> K.subscription con+    --putStrLn $ "Subscription state: \n" <> P.show subsState+    resps <- K.pollMessageBatch con kafkaTimeout (K.BatchSize batchSize)+    oldCount <- readIORef msgsRead+    time <- currentMilliSecs+    oldTime <- readIORef timeFrame+    let msgs =+            map+                (LBS.fromStrict . fromJust . K.crValue .+                 (\a -> assert (isNothing $ K.crKey a) a) .+                 fromRight)+                resps+    let !newCount = oldCount + length msgs+    if (time - oldTime > 10000)+        then do+            writeIORef timeFrame time+            writeIORef msgsRead 0+            putStrLn $ "Fetched " <> P.show newCount <>+                " messages in the last ~10 seconds"+            push (fetcherEventCount stats) newCount+        else writeIORef msgsRead newCount+               -- I just put this assert in here for now so that we can reason+               -- better about the structure of the kafka message+    --msgs `deepseq` putStrLn "done"+    pure msgs+  where+    {-# NOINLINE msgsRead #-}+    {-# NOINLINE timeFrame #-}+    msgsRead = unsafePerformIO $ newIORef 0+    timeFrame = unsafePerformIO $ newIORef =<< currentMilliSecs+    batchSize = 400++getIndex :: Int -> [a] -> Maybe a+getIndex n _+    | n < 0 = Nothing+getIndex _ [] = Nothing+getIndex 0 (x:_) = Just x+getIndex n (_:xs) = getIndex (n - 1) xs++writeRedis :: _ -> _ -> Redis.Redis _+writeRedis campaign window = do+    redisResponse <- getUUID (Tx.encodeUtf8 $ timestamp window)+    windowUUID <-+        case redisResponse of+            Nothing -> do+                windowUUID <- encodeUUID <$> liftIO randomIO+                checkR_ $+                    Redis.hset+                        campaign+                        (Tx.encodeUtf8 $ timestamp window)+                        windowUUID+                redisResponse2 <- getUUID "windows"+                windowListUUID <-+                    case redisResponse2 of+                        Nothing -> do+                            rand <- encodeUUID <$> liftIO randomIO+                            checkR_ $ Redis.hset campaign "windows" rand+                            pure rand+                        Just uuid -> pure uuid+                checkR_ $+                    Redis.lpush+                        windowListUUID+                        [Tx.encodeUtf8 $ timestamp window]+                pure windowUUID+            Just uuid -> pure uuid+    checkR_ $ Redis.hincrby windowUUID "seen_count" . fromIntegral =<<+        liftIO (readIORef (seenCount window))+    liftIO $ writeIORef (seenCount window) 0+    time <- BS.pack . P.show <$> liftIO currentMilliSecs+    checkR_ $ Redis.hset windowUUID "time_updated" time+    Redis.lpush "time_updated" [time]+    -- NOTE This function is not necessary. It "only" forces the errors from the+    -- redis database. You can drop it and all its uses if you want to describe+    -- the algorithm. The reason I have it is so we notice if something goes+    -- wrong with the redis database.+  where+    checkR_ :: (Monad f, Show err) => f (Either err a) -> f ()+    checkR_ = (either (error . P.show) (const $ pure ()) =<<)+    getUUID field =+        either (const Nothing) (join . getIndex 0) <$>+        Redis.hmget campaign [field]+    encodeUUID = BS.pack . P.show :: UUID.UUID -> BS.ByteString++redisGet :: MonadIO m => _ -> BS.ByteString -> m (Maybe BS.ByteString)+redisGet redisConn =+    liftIO . Redis.runRedis redisConn . fmap fromRight . Redis.get++isTheSame :: (Show a, Typeable a, Eq a, NFData a) => IO a -> STCLang a Bool+isTheSame init =+    liftWithState init $ \new -> do+        old <- get+        let isOld = old == new+        unless isOld $ put new+        pure $ isOld++redisJoinStateInit :: IO (NFMap.Map _ _)+redisJoinStateInit = NFMap.new++newWindow :: MonadIO m => Long -> m Window+newWindow timeBucket =+    liftIO $ Window <$> newIORef 0 <*> pure (show $ timeBucket * timeDivisor)++-- NOTE This function doesn't do much. I did make an extra function for this+-- because I am not sure why they changed this function to be so simple, and why+-- they removed the call to redis that I assume was in here. I have a suspicion,+-- that the simplification here means that this benchmark is not as it was+-- described in the paper and because of that I leave the function in so we+-- remember to check it later.+redisGetWindow :: MonadIO m => Long -> m (Maybe Window)+redisGetWindow timeBucket = Just <$> newWindow timeBucket++getWindow ::+       (MonadIO m, MonadState (w, NFMap.Map Long (NFMap.Map String Window)) m)+    => Long+    -> String+    -> m Window+getWindow timeBucket campaignId = do+    campaignWindows <- gets snd+    bucketMapE <-+        NFMap.lookup timeBucket campaignWindows >>= \case+            Just m -> pure $ Left m+            Nothing ->+                redisGetWindow timeBucket >>= \case+                    Nothing -> do+                        m <- NFMap.new+                        NFMap.insert timeBucket m campaignWindows+                        pure $ Left m+                    Just redisWindow -> do+                        m <- NFMap.new+                        NFMap.insert timeBucket m campaignWindows+                        pure $ Right redisWindow+    case bucketMapE of+        Right w -> pure w+        Left bucketMap ->+            NFMap.lookup campaignId bucketMap >>= \case+                Nothing -> do+                    window <-+                        maybe (newWindow timeBucket) pure =<<+                        redisGetWindow timeBucket+                    NFMap.insert campaignId window bucketMap+                    pure window+                Just window -> pure window++-- | These are necessary because the Kafka client is an older version (0.8.2.1)+-- and does not support the `ApiVersionRequest` that the C-client library we use+-- under the hood sends in the beginning.+extraKafkaProperties ::+       (IsList l, Item l ~ (s0, s1), IsString s1, IsString s0) => l+extraKafkaProperties =+    [("api.version.request", "false"), ("broker.version.fallback", "0.8.2.1")]++type KafkaReader = IO LBS.ByteString++type CloseKafka = IO ()++type KafkaActions = (KafkaReader, CloseKafka)++cachedBackoffReader :: _ -> IO [a] -> IO (IO a)+cachedBackoffReader backoff refetch = do+    cache <- newIORef []+    let go =+            readIORef cache >>= \case+                [] -> do+                    new <- refetch+                    if null new+                        then putStrLn "Refetch returned empty response" >>+                             threadDelay backoff+                        else writeIORef cache new+                    go+                (x:xs) -> writeIORef cache xs >> pure x+    pure go++setupMockKafka :: _ -> IO KafkaActions+setupMockKafka _conf = do+    kafkaVar <- newEmptyMVar+    let kafkaWriter m = putMVar kafkaVar m+    kafkaReader <- cachedBackoffReader 100 (takeMVar kafkaVar)+    writerThread <- forkIO $ eventGenerationLoop kafkaWriter+    pure (kafkaReader, killThread writerThread)++setupKafka :: Statistics -> AE.Value -> IO KafkaActions+setupKafka stats conf = do+    print topic+    print conf+    print $ K.cpProps props+    cons <- fromRight <$> (K.newConsumer props sub)+    reader <- cachedBackoffReader 100 $ readKafka cons stats+    pure (reader, maybe (pure ()) (error . P.show) =<< K.closeConsumer cons)+  where+    topic = K.TopicName $ conf ^?! key "kafka.topic" . _String+    sub = K.topics [topic] <> K.offsetReset K.Latest+    props+            -- NOTE If I do not assign a group it fails immediately with+            -- "unknown group".+     =+        K.groupId (K.ConsumerGroupId "ohua-stream-bench-group") <>+        --K.extraProps extraKafkaProperties <>+        --K.debugOptions [K.DebugAll] <>+        K.brokersList+            (map (\host ->+                      K.BrokerAddress $ host <> ":" <>+                      show (conf ^?! key "kafka.port" . _Integer))+                 (conf ^.. key "kafka.brokers" . values . _String))++-- | Reads the config file at the specified path and creates the connection+-- objects we need+setup :: FilePath -> IO (KafkaActions, Redis.Connection, Statistics)+setup loc = do+    conf <- fromRight <$> Yaml.decodeFileEither @AE.Value loc+    let rinfo =+            Redis.defaultConnectInfo+                { Redis.connectHost =+                      conf ^?! key "redis.host" . _String . to Tx.unpack+                }+    -- cons <- fmap fromRight (K.newConsumer props sub)+    stats <- initStatistics+    (,,) <$> setupKafka stats conf <*> Redis.checkedConnect rinfo <*> pure stats++withInitial msg ac = do+    putStrLn $ "Doing initial " <> msg <> "..."+    r <- ac+    putStrLn $ msg <> " done"+    pure r++traceM :: Monad m => P.String -> m ()+traceM msg = Debug.trace msg $ pure ()++-- NOTE in the original implementation of the algorithm `rebalance` is the first+-- function called on the input stream. This distributes the messages+-- round-robin. This means we could also spawn multiple algorithm instances and+-- process multiple messages in parallel. But we'd have to ensure the timer+-- events are *not* round robin distributed!+algo kafkaReader redisConn stats+    -- Allocate the basic functions --+ = do+    traceM "Start allocations"+    let deserialize o =+            either (error . ("Decoding error: " <>)) id $ do+                result <- AE.eitherDecode o+                flip AE.parseEither result $ \o ->+                    (,,,,,,) <$> (o .: "user_id" :: _ String) <*>+                    (o .: "page_id" :: _ String) <*>+                    (o .: "ad_id" :: _ String) <*>+                    (o .: "ad_type" :: _ String) <*>+                    (o .: "event_type" :: _ String) <*>+                    (read <$> o .: "event_time") <*>+                    (o .: "ip_address" :: _ String)+    let evFilterFunc ~(_, _, _, _, t, _, _) = pure $ t == "view"+    let project ~(_, _, i2, _, _, i5, _) = pure (i2, i5)+    traceM "Allocating redis"+    redisJoin <-+        liftWithState redisJoinStateInit $ \(adId, v2) ->+            fmap (, adId, v2) <$> do+                st <- get+                NFMap.lookup adId st >>= \case+                    Just cid -> pure $ Just cid+                    Nothing -> do+                        mcid <- redisGet redisConn adId+                        maybe+                            (liftIO $ putStrLn "Ad campaign not found in redis")+                            (\cid -> NFMap.insert adId cid st)+                            mcid+                        pure mcid+    -- The campaign processor wrapped in the logic to separate handling of the+    -- timing event, regular and filtered data.+    emitCounter <-+        liftWithState (pure 0 :: _ Word32) $ \case+            Right _ -> modify succ+            Left _ -> do+                c <- get+                put 0+                liftIO $ putStrLn $ "Saw " <> P.show c <>+                    " events total in time window"+                push (pipelineEventCount stats) c+    traceM "Allocating campaign"+    processCampaign <-+        liftWithState ((,) <$> Set.new <*> NFMap.new) $ \case+            Right (Just ev@(campaignId, _adId, eventTime)) -> do+                flushCache <- gets fst+                let timeBucket = eventTime `div` timeDivisor+                window <- getWindow timeBucket campaignId+                liftIO $ modifyIORef' (seenCount window) (+ 1)+                let value = (campaignId, window)+                Set.insert value flushCache+            Left timerTrigger -> do+                (s, cache) <- get+                newCache <- Set.new+                modify (\(_, o) -> (newCache, o))+                asList <- Set.toList s+                --liftIO $ printf "I touched %d campaigns" (length $ nub $ map fst asList)+                let l = length asList+                push (redisWritesCount stats) l+                liftIO $ do+                    putStrLn $ "Initiated redis write of " <> P.show l <>+                        " events"+                    Redis.runRedis redisConn $+                        mapM_+                            (\(cid, window) -> do+                                 c <- liftIO $ readIORef $ seenCount window+                                   --liftIO $ printf "Writing count %4d for timestamp %v\n" c (timestamp window)+                                 writeRedis cid window)+                            asList+            _ -> pure ()+    -- The condition we will use later for the if+    evCheck <- isTheSame currentMilliSecs+    -- Allocate signals+    traceM "Allocating timer"+    timerSig <-+        liftSignal+            (threadDelay timeout >> currentMilliSecs)+            (withInitial "time" currentMilliSecs)+    -- Not sure if this is a good idea but I initialize here by polling the+    -- first message. Perhaps we should use a `Maybe` instead, however its not+    -- particularly convenient yet in our model so I do this.+    traceM "Allocating Kafka reader"+    msgSig <- liftSignal kafkaReader (withInitial "read kafka" kafkaReader)+    traceM "Allocating preprocessor"+    filteredProcessor+        -- NOTE keyBy partitions the operator state according to some key. If we+        -- have time we should implement that too+         <-+        filterSignalM+            evFilterFunc+            (project >=> redisJoin+                                       -- >=> keyBy 0+             )+    -- The actual algorithm+    return $ \src -> do+        timerEv <- timerSig src+        msgEv <- msgSig src+        -- Fork on whether this is a timing event+        procInput <-+            if_+                (evCheck timerEv)+                (Right <$> do+                     msg <- pure $ deserialize msgEv+                     join <$> filteredProcessor msg)+                (pure $ Left timerEv)+        emitCounter procInput+        processCampaign procInput++-- data CollSt = CollSt+--   { states :: [S]+--   , signals :: [IO S]+--   }+--+-- instance Monoid CollSt where+--   mempty = CollSt [] []+--   CollSt st1 si1 `mappend` CollSt st2 si2 =+--     CollSt (st1 `mappend` st2) (si1 `mappend` si2)+algoSeq kafkaReader redisConn stats+    -- Allocate the basic functions --+ =+    runSignals $ do+        traceM "Start allocations"+        let deserialize o =+                either (error . ("Decoding error: " <>)) id $ do+                    result <- AE.eitherDecode o+                    flip AE.parseEither result $ \o ->+                        (,,,,,,) <$> (o .: "user_id" :: _ String) <*>+                        (o .: "page_id" :: _ String) <*>+                        (o .: "ad_id" :: _ String) <*>+                        (o .: "ad_type" :: _ String) <*>+                        (o .: "event_type" :: _ String) <*>+                        (read <$> o .: "event_time") <*>+                        (o .: "ip_address" :: _ String)+        let evFilterFunc ~(_, _, _, _, t, _, _) = pure $ t == "view"+        let project ~(_, _, i2, _, _, i5, _) = pure (i2, i5)+        traceM "Allocating redis"+        redisJoin <-+            liftWithState redisJoinStateInit $ \(adId, v2) ->+                fmap (, adId, v2) <$> do+                    st <- get+                    NFMap.lookup adId st >>= \case+                        Just cid -> pure $ Just cid+                        Nothing -> do+                            mcid <- redisGet redisConn adId+                            maybe+                                (liftIO $+                                 putStrLn "Ad campaign not found in redis")+                                (\cid -> NFMap.insert adId cid st)+                                mcid+                            pure mcid+    -- The campaign processor wrapped in the logic to separate handling of the+    -- timing event, regular and filtered data.+        emitCounter <-+            liftWithState (pure 0 :: _ Word32) $ \case+                Right _ -> modify succ+                Left _ -> do+                    c <- get+                    put 0+                    liftIO $ putStrLn $ "Saw " <> P.show c <>+                        " events total in time window"+                    push (pipelineEventCount stats) c+        traceM "Allocating campaign"+        processCampaign <-+            liftWithState ((,) <$> Set.new <*> NFMap.new) $ \case+                Right (Just ev@(campaignId, _adId, eventTime)) -> do+                    flushCache <- gets fst+                    let timeBucket = eventTime `div` timeDivisor+                    window <- getWindow timeBucket campaignId+                    liftIO $ modifyIORef' (seenCount window) (+ 1)+                    let value = (campaignId, window)+                    Set.insert value flushCache+                Left timerTrigger -> do+                    (s, cache) <- get+                    newCache <- Set.new+                    modify (\(_, o) -> (newCache, o))+                    asList <- Set.toList s+                --liftIO $ printf "I touched %d campaigns" (length $ nub $ map fst asList)+                    let l = length asList+                    push (redisWritesCount stats) l+                    liftIO $ do+                        putStrLn $ "Initiated redis write of " <> P.show l <>+                            " events"+                        Redis.runRedis redisConn $+                            mapM_+                                (\(cid, window) -> do+                                     c <- liftIO $ readIORef $ seenCount window+                                   --liftIO $ printf "Writing count %4d for timestamp %v\n" c (timestamp window)+                                     writeRedis cid window)+                                asList+                _ -> pure ()+    -- The condition we will use later for the if+        evCheck <- isTheSame currentMilliSecs+    -- Allocate signals+        traceM "Allocating timer"+        timerSig <-+            liftSignal+                (threadDelay timeout >> currentMilliSecs)+                (withInitial "time" currentMilliSecs)+    -- Not sure if this is a good idea but I initialize here by polling the+    -- first message. Perhaps we should use a `Maybe` instead, however its not+    -- particularly convenient yet in our model so I do this.+        traceM "Allocating Kafka reader"+        msgSig <- liftSignal kafkaReader (withInitial "read kafka" kafkaReader)+        traceM "Allocating preprocessor"+        filteredProcessor+        -- NOTE keyBy partitions the operator state according to some key. If we+        -- have time we should implement that too+             <-+            filterSignalM+                evFilterFunc+                (project >=> redisJoin+                                       -- >=> keyBy 0+                 )+    -- The actual algorithm+        pure $ \src -> do+            timerEv <- timerSig src+            msgEv <- msgSig src+        -- Fork on whether this is a timing event+            procInput <-+                if_+                    (evCheck timerEv)+                    (Right <$> do+                         msg <- pure $ deserialize msgEv+                         join <$> filteredProcessor msg)+                    (pure $ Left timerEv)+            emitCounter procInput+            processCampaign procInput+  where+    isTheSame init =+        liftWithState init $ \new -> do+            old <- get+            let isOld = old == new+            unless isOld $ put new+            pure $ isOld+    printSignalD = putStrLn+    runOhuaM comp states = do+        v <- V.thaw (V.fromList states)+        a <- runReaderT comp v+        (a, ) . V.toList <$> V.freeze v+    if_ c t e = do+        c' <- c+        if c'+            then t+            else e+    filterSignalM cond f =+        pure $ \item -> if_ (cond item) (Just <$> f item) (pure Nothing)+    liftWithState ::+           (Typeable s, NFData s)+        => IO s+        -> (SF s a b)+        -> BuildSeqOhua (a -> SeqOhua b)+    liftWithState state stateThread = do+        s0 <- lift state+        l <-+            S.state $ \s ->+                (length $ states s, s {states = states s ++ [toS s0]})+        pure $ \a -> do+            v <- ask+            liftIO $ do+                s <- fromS <$> MV.read v l+                (b, s') <- runStateT (stateThread a) s+                MV.write v l $ toS s'+                pure b+    liftSignal ::+           (Typeable a, NFData a)+        => IO a+        -> IO a+        -> BuildSeqOhua (Signals -> SeqOhua a)+    liftSignal s0 init = do+        idx <-+            S.state $ \s@CollSt {signals} ->+                (length signals, s {signals = signals ++ [toS <$> s0]})+        liftWithState init $ \(i, s) ->+            if i == idx+                then do+                    let my = fromS s+                    S.put my+                    pure my+                else S.get+    runSignals comp = do+        (comp', s) <- S.runStateT comp mempty+        chan <- BC.newBoundedChan 100+        bracket+            (do forM (zip [0 ..] $ signals s) $ \(idx, sig) ->+                    Conc.forkIO $ forever $ do+                        event <- sig+                        BC.writeChan chan $ Just (idx, event))+            (\threads -> do+                 printSignalD "Killing signal threads"+                 mapM_ Conc.killThread threads)+            (\_ -> do+                 printSignalD "signals done"+                 let signalGen =+                         ioReaderToGenerator @(Generator IO) (BC.readChan chan)+                 runOhuaM (smapGen comp' signalGen) $ states s)+    smapGen f = foldlGeneratorT lift (\acc i -> (: acc) <$> f i) []++type BuildSeqOhua a = StateT CollSt IO a++type SeqOhua a = ReaderT (MV.IOVector S) IO a++main, main0 :: IO ()+main = main0++main0 = do+    [confPath] <- getArgs+    runSequential <- maybe False (/= "") <$> lookupEnv "SEQUENTIAL"+    void $+        bracket+            (setup confPath)+            (\((_, closeKafka), _redisConn, stats) -> do+                 putStrLn "Closing resources"+                 closeKafka+                 putStrLn "Writing Statistics"+                 withFile "ohua-statistics.txt" WriteMode $ \h ->+                     writeStatistics stats h+             -- Redis.disconnect redisConn+             )+            (\((kafkaReader, _), redisConn, stats) ->+                 putStrLn "Starting execution" >>+                 if runSequential+                     then algoSeq kafkaReader redisConn stats+                     else runSignals (algo kafkaReader redisConn stats))

@@ -0,0 +1,60 @@+module FakeComputation where++-- | Original source https://github.com/iu-parfunc/lvars/tree/master/archived_old/fhpc13-lvars/benchmarks+import Control.DeepSeq+import Control.Monad.State++import Control.Concurrent (myThreadId)++-- Iterates the sin function n times on its input and returns the sum+-- of all iterations.+sin_iter :: Int -> Float -> Float+sin_iter 0 x = x+sin_iter n !x = sin_iter (n - 1) (x + sin x)++cos_iter :: Int -> Float -> Float+cos_iter 0 x = x+cos_iter n !x = cos_iter (n - 1) (x + cos x)++tan_iter :: Int -> Float -> Float+tan_iter 0 x = x+tan_iter n !x = tan_iter (n - 1) (x + tan x)++wrk_sins :: Int -> Float -> Float+wrk_sins num sin_wrk =+    let res = sin_iter num (2.222 + sin_wrk)+     in force res++gwrk :: Int -> Float -> (Int -> Float -> Float) -> Float+gwrk num wrk f =+    let res = f num (2.222 + wrk)+     in force res++type ID = Int++work :: Float -> StateT (ID, Int) IO (Float, Float)+work wrk+  -- liftIO $ putStrLn $ "work: " ++ (show wrk)+ = do+    (identifier, numIter) <- get+    tId <- liftIO myThreadId+  -- liftIO $ putStrLn $ "start: " ++ (show identifier) ++ " on thread: " ++ (show tId)+  -- let r = wrk_sins numIter wrk+    let r = sin_iter numIter (2.222 + wrk)+  -- r <- liftIO $ evaluate $ force $ sin_iter numIter (2.222 + wrk) -- this is the solution!+  -- liftIO $ putStrLn $ "stop: " ++ (show identifier)+  -- liftIO $ putStrLn $ "result: " ++ (show r)+    return (wrk, r)++gwork :: (Int -> Float -> Float) -> Float -> StateT (ID, Int) IO (Float, Float)+gwork f wrk+  -- liftIO $ putStrLn $ "work: " ++ (show wrk)+ = do+    (identifier, numIter) <- get+    tId <- liftIO myThreadId+  -- liftIO $ putStrLn $ "start: " ++ (show identifier) ++ " on thread: " ++ (show tId)+    let r = gwrk numIter wrk f+  -- r <- liftIO $ evaluate $ force $ f numIter (2.222 + wrk) -- this is the solution!+  -- liftIO $ putStrLn $ "stop: " ++ (show identifier)+  -- liftIO $ putStrLn $ "result: " ++ (show r)+    return (wrk, r)

@@ -0,0 +1,251 @@+module SD.Correctness (testSuite) where++import Control.Monad.State++import Test.Framework+import Test.Framework.Providers.HUnit+import Test.HUnit hiding (State)++-- import Data.Monoid+-- import Utils+import Control.Monad.SD+import Data.StateElement++foo :: Int -> StateT Int IO Int+foo x = do+    s <- get+    put $ s + 2+    return $ x + 2++bar :: Int -> StateT Int IO Int+bar x = do+    s <- get+    put $ s + 3+    return $ x * 3++barFloat :: Int -> StateT Float IO Int+barFloat x = do+    s <- get+    put $ s + 3.34+    return $ x * 3++-- simpleAlgo :: Int -> OhuaM ([LocalStateBox Int], [LocalStateBox Int]) Int+simpleComposition v = do+    c <- return v+    r0 <- liftWithIndex 0 foo c+    r1 <- liftWithIndex 1 bar r0+    return r1++packagedSimpleComposition = do+    f0 <- liftWithState (return 0) foo+    f1 <- liftWithState (return 0) bar+    return $ f1 <=< f0++simpleCompositionHetState v = do+    c <- return v+    r0 <- liftWithIndex 0 foo c+    r1 <- liftWithIndex 1 barFloat r0+    return r1++simpleSMap = smap simpleComposition++smapWithContext v = do+    c <- return v+    r0 <- liftWithIndex 2 foo c+    r1 <- liftWithIndex 3 bar r0+    r2 <- smap simpleComposition [r0, r1]+    return r2++smapResultUsed v = do+    c <- return v+    r0 <- liftWithIndex 2 foo c+    r1 <- liftWithIndex 3 bar r0+    r2 <- smap simpleComposition [r0, r1]+    r3 <- liftWithIndex 4 foo $ r2 !! 0+    r4 <- liftWithIndex 5 bar $ r2 !! 1+    return (r3, r4)++packagedSmapResultUsed = do+    f0 <- liftWithState (return 0) foo+    f1 <- liftWithState (return 0) bar+    f2 <- smapSTC packagedSimpleComposition+    f3 <- liftWithState (return 0) foo+    f4 <- liftWithState (return 0) bar+    return $ \v -> do+        r0 <- f0 v+        r1 <- f1 r0+        r2 <- f2 [r0, r1]+        r3 <- f3 $ r2 !! 0+        r4 <- f4 $ r2 !! 1+        return (r3, r4)++smapOverEmptyList = do+    r1 <- smap simpleComposition []+    smap someComp [length r1]+  where+    someComp i = do+        r0 <- liftWithIndex 2 foo i+        liftWithIndex 3 bar r0++smapOverEmptyList2 = smap (smap simpleComposition) [[], [1 .. 3]]++simpleCompositionPackaged v = do+    c <- return v+    r0 <- liftWithIndex 0 foo c+    r1 <- liftWithIndex 1 bar r0+    return r1++caseComp idxFoo idxBranch1 idxBranch2 v = do+    c <- liftWithIndex idxFoo foo v+    o <- case_ c [(4, branch1 c), (8, branch2 c)]+    return o+  where+    branch1 = liftWithIndex idxBranch1 bar+    branch2 = liftWithIndex idxBranch2 bar++caseComposition = caseComp 0 1 2++smapWithCase = smap caseComposition++nestedCase v = do+    o <- case_ v [(2, caseComp 0 1 2 v), (6, caseComp 3 4 5 v)]+    return o++ret = return (10 :: Int)++returnTest :: Assertion+returnTest = do+    (result, s) <- runOhuaM ret []+    assertEqual "result was wrong." (10 :: Int) result+    assertEqual "state was wrong." [] (map fromS s :: [Int])++bindTest :: Assertion+bindTest = do+    (result, s) <- runOhuaM (simpleComposition 10) $ map toS [0 :: Int, 0]+    assertEqual "result was wrong." 36 result+    assertEqual "state was wrong." [2, 3] (map fromS s :: [Int])++hetStateTest :: Assertion+hetStateTest = do+    (result, s1:(s2:_)) <-+        runOhuaM+            (simpleCompositionHetState 10)+            [toS (0 :: Int), toS (2.5 :: Float)]+    assertEqual "result was wrong." 36 result+    assertEqual "state was wrong." 2 (fromS s1 :: Int)+    assertEqual "state was wrong." 5.84 (fromS s2 :: Float)+  -- assertEqual "state was wrong." 2 (toConcrete s1 :: Int)+  -- assertEqual "state was wrong." 5.84 (toConcrete s2 :: Float)++pipeSMapTest :: Assertion+pipeSMapTest = do+    (result, s) <- runOhuaM (simpleSMap [10, 10]) $ map toS [0 :: Int, 0]+    assertEqual "result was wrong." [36, 36] result+    assertEqual "state was wrong." [4, 6] (map fromS s :: [Int])++smapContextTest :: Assertion+smapContextTest = do+    (result, s) <- runOhuaM (smapWithContext 10) $ map toS [0 :: Int, 0, 0, 0]+    assertEqual "result was wrong." [42, 114] result+    assertEqual "state was wrong." [4, 6, 2, 3] (map fromS s :: [Int])++smapResultUsedTest :: Assertion+smapResultUsedTest = do+    (result, s) <-+        runOhuaM (smapResultUsed 10) $ map toS [0 :: Int, 0, 0, 0, 0, 0]+    assertEqual "result was wrong." (44, 342) result+    assertEqual "state was wrong." [4, 6, 2, 3, 2, 3] (map fromS s :: [Int])++-- | Basically the same as 'smapResultUsed' but order of elements in the state+-- is different, because of the order in which the state monad collects the+-- indices.+packagedSmapResultUsedTest :: Assertion+packagedSmapResultUsedTest = do+    (result, s) <- runSTCLang packagedSmapResultUsed 10+    assertEqual "result was wrong." (44, 342) result+    assertEqual "state was wrong." [2, 3, 4, 6, 2, 3] (map fromS s :: [Int])++packagedBindTest :: Assertion+packagedBindTest = do+    (result, s) <-+        runOhuaM (simpleCompositionPackaged 10) $ map toS [0 :: Int, 0]+    assertEqual "result was wrong." 36 result+    assertEqual "state was wrong." [2, 3] (map fromS s :: [Int])++caseTest :: Assertion+caseTest+  -- "true" branch+ = do+    (result, s) <- runOhuaM (caseComposition 2) $ map toS [0 :: Int, 0, 0]+    assertEqual "result was wrong." 12 result+    assertEqual "state was wrong." [2, 3, 0] (map fromS s :: [Int])+  -- "false" branch+    (result', s') <- runOhuaM (caseComposition 6) $ map toS [0 :: Int, 0, 0]+    assertEqual "result was wrong." 24 result'+    assertEqual "state was wrong." [2, 0, 3] (map fromS s' :: [Int])++caseSmapTest :: Assertion+caseSmapTest+  -- "true" branch+ = do+    (result, s) <- runOhuaM (smapWithCase [2, 6]) $ map toS [0 :: Int, 0, 0]+    assertEqual "result was wrong." [12, 24] result+    assertEqual "state was wrong." [4, 3, 3] (map fromS s :: [Int])+  -- execute only once+    (result, s) <- runOhuaM (smapWithCase [2]) $ map toS [0 :: Int, 0, 0]+    assertEqual "result was wrong." [12] result+  --assertEqual "state was wrong." [4,3,3] (map fromS s :: [Int])++nestedCaseTest :: Assertion+nestedCaseTest+  -- "true" branch+ = do+    (result, s) <- runOhuaM (nestedCase 2) $ map toS [0 :: Int, 0, 0, 0, 0, 0]+    assertEqual "result was wrong." 12 result+    assertEqual "state was wrong." [2, 3, 0, 0, 0, 0] (map fromS s :: [Int])+  -- "false" branch+    (result', s') <- runOhuaM (nestedCase 6) $ map toS [0 :: Int, 0, 0, 0, 0, 0]+    assertEqual "result was wrong." 24 result'+    assertEqual "state was wrong." [0, 0, 0, 2, 0, 3] (map fromS s' :: [Int])++tooMuchStateTest :: Assertion+tooMuchStateTest = do+    (result, s) <- runOhuaM ret $ map toS [0 :: Int]+    assertEqual "result was wrong." (10 :: Int) result+    assertEqual "state was wrong." [0] (map fromS s :: [Int])++notEnoughStateTest :: Assertion+notEnoughStateTest = do+    (result, s) <- runOhuaM (simpleComposition 10) $ map toS [0 :: Int]+    assertEqual "result was wrong." 36 result+    assertEqual "state was wrong." [2, 3] (map fromS s :: [Int])++smapHandlesEmptyList :: Assertion+smapHandlesEmptyList =+    void (runOhuaM smapOverEmptyList (map toS [0 .. 3 :: Int]))++smapHandlesEmptyList2 :: Assertion+smapHandlesEmptyList2 = do+    (res, _) <- runOhuaM smapOverEmptyList2 (map toS [0 .. 1 :: Int])+    assertEqual "lengths differ" [0, 3] (map length res)++testSuite :: Test.Framework.Test+testSuite =+    testGroup+        "Futures"+        [ testCase "checking monadic return" returnTest+        , testCase "checking monadic bind" bindTest+        , testCase "checking simple pipe smap" pipeSMapTest+        , testCase "checking smap with context" smapContextTest+        , testCase "checking smap result used" smapResultUsedTest+        , testCase "smap over empty list" smapHandlesEmptyList+        , testCase "nested smap over empty list" smapHandlesEmptyList2+        , testCase "checking packaged version" packagedBindTest+        , testCase "checking case statement" caseTest+        , testCase "checking smap-case composition" caseSmapTest+        , testCase "simple nested case composition" nestedCaseTest+        , testCase "heterogeneous state" hetStateTest+        , testCase "test packaged state" packagedSmapResultUsedTest+            -- , testCase "Futures: too much state" tooMuchStateTest --> this turns into an Error in monad-par that says: "no result"+            -- , testCase "Futures: not enough state" notEnoughStateTest --> turns into the error: Prelude.!!: index too large+        ]

@@ -0,0 +1,56 @@+module SD.Performance (testSuite) where++import Test.Framework+import Test.Framework.Providers.HUnit+import Test.HUnit hiding (State)++import FakeComputation (work, wrk_sins)++import Control.Monad.SD+import Data.StateElement++import Data.Time.Clock.POSIX++import GHC.Conc (getNumCapabilities, numCapabilities, setNumCapabilities)++currentTimeMillis = round . (* 1000) <$> getPOSIXTime++pipeline v = do+    c <- return v+    (_, r0) <- liftWithIndex 0 work c+    (_, r1) <- liftWithIndex 1 work r0+    (_, r2) <- liftWithIndex 2 work r1+    (_, r3) <- liftWithIndex 3 work r2+    return r3++fourStepPipeline = smap pipeline++-- Beware: You need to recompile with "-threaded" in order to  enable concurrency!+--         Just changing the cabal file and running `stack test` won't work.+--         Instead always do `stack clean && stack test`+pipeSMapTest :: Assertion+pipeSMapTest = do+    let a = 3000000 :: Float+    let b = 2000000 :: Int+    let inputs = replicate 4 a+    let r = wrk_sins b a+    let expectedOutputs = replicate 4 r+    putStrLn $ "num cores (RTS option): " ++ (show numCapabilities)+    (\x -> putStrLn $ "num cores: " ++ show x) =<< getNumCapabilities+    start <- currentTimeMillis+    (result, _) <-+        runOhuaM (fourStepPipeline inputs) $+        map toS $ [(0 :: Int, b), (1, b), (2, b), (3, b)]+    stop <- currentTimeMillis+    putStrLn $ "Exec time [ms]: " ++ (show $ stop - start)+    assertEqual "result was wrong." expectedOutputs result++coresTest = mapM_ runTest [1 .. 4]+  where+    runTest numCores = do+        setNumCapabilities numCores+        pipeSMapTest+      -- TODO validation needed! (for now, check the exec times)++testSuite :: Test.Framework.Test+testSuite = testGroup "Performance FBM" [testCase "4-step pipeline" coresTest]

@@ -0,0 +1,12 @@+{-# LANGUAGE OverloadedLists #-}+import Test.Framework++import SD.Correctness as FBM+import SD.Performance as PFBM+++main :: IO ()+main =+    defaultMain+        [FBM.testSuite, PFBM.testSuite]+-- main = flip defaultMainWithOpts mempty FBM.testSuite