Monte carlo: try experimental stream (no fusion yet)

conduit/Data/Conduit/Internal.hs

@@ -4,6 +4,8 @@ module Data.Conduit.Internal
       module Data.Conduit.Internal.Pipe
       -- * Conduit
     , module Data.Conduit.Internal.Conduit
+      -- * Fusion (highly experimental!!!)
+    , module Data.Conduit.Internal.Fusion
     ) where
 
 import           Data.Conduit.Internal.Conduit hiding (addCleanup, await,
@@ -12,3 +14,4 @@ import           Data.Conduit.Internal.Conduit hiding (addCleanup, await,
                                                 mapOutputMaybe, transPipe,
                                                 yield, yieldM, yieldOr)
 import           Data.Conduit.Internal.Pipe
+import           Data.Conduit.Internal.Fusion

conduit/Data/Conduit/Internal/Fusion.hs

@@ -0,0 +1,48 @@
+{-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE RankNTypes #-}
+module Data.Conduit.Internal.Fusion
+    ( -- ** Types
+      Step (..)
+    , Stream (..)
+      -- ** Producer
+    , streamProducerM
+    , streamProducerId
+    ) where
+
+import Data.Conduit.Internal.Conduit
+import Data.Conduit.Internal.Pipe (Pipe (..))
+import Data.Functor.Identity (Identity (runIdentity))
+
+-- | This is the same as stream fusion\'s Step. Constructors are renamed to
+-- avoid confusion with conduit names.
+data Step s o
+    = Emit s o
+    | Skip s
+    | Stop
+
+data Stream m o = forall s. Stream
+    (s -> m (Step s o))
+    (m s)
+
+streamProducerM :: Monad m => Stream m o -> Producer m o
+streamProducerM (Stream step ms0) =
+    ConduitM $ PipeM $ ms0 >>= loop
+  where
+    loop s = do
+        res <- step s
+        case res of
+            Emit s' o -> return $ HaveOutput (PipeM $ loop s') (return ()) o
+            Skip s' -> loop s'
+            Stop -> return $ Done ()
+{-# INLINE [0] streamProducerM #-}
+
+streamProducerId :: Monad m => Stream Identity o -> Producer m o
+streamProducerId (Stream step ms0) =
+    ConduitM $ loop $ runIdentity ms0
+  where
+    loop s =
+        case runIdentity $ step s of
+            Emit s' o -> HaveOutput (loop s') (return ()) o
+            Skip s' -> loop s'
+            Stop -> Done ()
+{-# INLINE [0] streamProducerId #-}

conduit/benchmarks/optimize-201408.hs

@@ -2,6 +2,7 @@
 {-# LANGUAGE ExistentialQuantification #-}
 {-# LANGUAGE FlexibleContexts          #-}
 {-# LANGUAGE MultiParamTypeClasses     #-}
+{-# LANGUAGE TupleSections             #-}
 -- Collection of three benchmarks: a simple integral sum, monte carlo analysis,
 -- and sliding vector.
 import           Control.DeepSeq
@@ -120,6 +121,14 @@ monteCarloTB = return $ TBGroup "monte carlo"
                         | otherwise = t
                 go (i - 1) t'
         go count (0 :: Int)
+    , TBIOTest "conduit, stream" closeEnough $ do
+        successes <- sourceRandomNStream count
+                  $$ CL.fold (\t (x, y) ->
+                                if (x*x + y*(y :: Double) < 1)
+                                    then t + 1
+                                    else t)
+                        (0 :: Int)
+        return $ fromIntegral successes / fromIntegral count * 4
     , TBIOTest "conduit, ConduitM primitives" closeEnough $ do
         successes <- sourceRandomN count
                   $$ CL.fold (\t (x, y) ->
@@ -176,6 +185,15 @@ sourceRandomN cnt0 = do
             liftIO (MWC.uniform gen) >>= yield >> loop (cnt - 1)
     loop cnt0
 
+sourceRandomNStream :: (MWC.Variate a, MonadIO m) => Int -> Source m a
+sourceRandomNStream cnt0 =
+    CI.streamProducerM $ CI.Stream step (liftIO $ fmap (, cnt0) MWC.createSystemRandom)
+  where
+    step (_, 0) = return CI.Stop
+    step (gen, i) = do
+        o <- liftIO $ MWC.uniform gen
+        return $ CI.Emit (gen, i - 1) o
+
 sourceRandomNBind :: (MWC.Variate a, MonadIO m) => Int -> Source m a
 sourceRandomNBind cnt0 = lift (liftIO MWC.createSystemRandom) >>= \gen ->
     let loop 0 = return ()

conduit/conduit.cabal

@@ -21,6 +21,7 @@ Library
                        Data.Conduit.Lift
   other-modules:       Data.Conduit.Internal.Pipe
                        Data.Conduit.Internal.Conduit
+                       Data.Conduit.Internal.Fusion
   Build-depends:       base                     >= 4.3          && < 5
                      , resourcet                >= 1.1          && < 1.2
                      , exceptions