diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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_). diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/src/Control/Monad/Generator.hs b/src/Control/Monad/Generator.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/Generator.hs @@ -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 diff --git a/src/Control/Monad/SD.hs b/src/Control/Monad/SD.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/SD.hs @@ -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 diff --git a/src/Control/Monad/SD/Case.hs b/src/Control/Monad/SD/Case.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/SD/Case.hs @@ -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_)] diff --git a/src/Control/Monad/SD/Combinator.hs b/src/Control/Monad/SD/Combinator.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/SD/Combinator.hs @@ -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 #-} diff --git a/src/Control/Monad/SD/FRP.hs b/src/Control/Monad/SD/FRP.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/SD/FRP.hs @@ -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 diff --git a/src/Control/Monad/SD/Ohua.hs b/src/Control/Monad/SD/Ohua.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/SD/Ohua.hs @@ -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 diff --git a/src/Control/Monad/SD/STCLang.hs b/src/Control/Monad/SD/STCLang.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/SD/STCLang.hs @@ -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 diff --git a/src/Control/Monad/SD/Smap.hs b/src/Control/Monad/SD/Smap.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/SD/Smap.hs @@ -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'' diff --git a/src/Control/Monad/Stream.hs b/src/Control/Monad/Stream.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/Stream.hs @@ -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 () diff --git a/src/Control/Monad/Stream/Chan.hs b/src/Control/Monad/Stream/Chan.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/Stream/Chan.hs @@ -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 diff --git a/src/Control/Monad/Stream/Par.hs b/src/Control/Monad/Stream/Par.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/Stream/Par.hs @@ -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 diff --git a/src/Control/Monad/Stream/PinnedChan.hs b/src/Control/Monad/Stream/PinnedChan.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/Stream/PinnedChan.hs @@ -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 () diff --git a/src/Data/Dynamic2.hs b/src/Data/Dynamic2.hs new file mode 100644 --- /dev/null +++ b/src/Data/Dynamic2.hs @@ -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) diff --git a/src/Data/StateElement.hs b/src/Data/StateElement.hs new file mode 100644 --- /dev/null +++ b/src/Data/StateElement.hs @@ -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 diff --git a/src/Type/Magic.hs b/src/Type/Magic.hs new file mode 100644 --- /dev/null +++ b/src/Type/Magic.hs @@ -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 diff --git a/src/Type/Magic/GHC8.hs b/src/Type/Magic/GHC8.hs new file mode 100644 --- /dev/null +++ b/src/Type/Magic/GHC8.hs @@ -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 diff --git a/src/Type/Magic/OldGHC.hs b/src/Type/Magic/OldGHC.hs new file mode 100644 --- /dev/null +++ b/src/Type/Magic/OldGHC.hs @@ -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 diff --git a/stc-lang.cabal b/stc-lang.cabal new file mode 100644 --- /dev/null +++ b/stc-lang.cabal @@ -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 +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 diff --git a/stream-bench/CampaignProcMap.hs b/stream-bench/CampaignProcMap.hs new file mode 100644 --- /dev/null +++ b/stream-bench/CampaignProcMap.hs @@ -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 diff --git a/stream-bench/MutableNFMap.hs b/stream-bench/MutableNFMap.hs new file mode 100644 --- /dev/null +++ b/stream-bench/MutableNFMap.hs @@ -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 diff --git a/stream-bench/MutableSet.hs b/stream-bench/MutableSet.hs new file mode 100644 --- /dev/null +++ b/stream-bench/MutableSet.hs @@ -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 _ = () diff --git a/stream-bench/algo.hs b/stream-bench/algo.hs new file mode 100644 --- /dev/null +++ b/stream-bench/algo.hs @@ -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)) diff --git a/test/FakeComputation.hs b/test/FakeComputation.hs new file mode 100644 --- /dev/null +++ b/test/FakeComputation.hs @@ -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) diff --git a/test/SD/Correctness.hs b/test/SD/Correctness.hs new file mode 100644 --- /dev/null +++ b/test/SD/Correctness.hs @@ -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 + ] diff --git a/test/SD/Performance.hs b/test/SD/Performance.hs new file mode 100644 --- /dev/null +++ b/test/SD/Performance.hs @@ -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] diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -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