bench-stream.hs

{-# LANGUAGE RecordWildCards #-}
module Main where

import           Unsafe.Coerce
import           Transient.Internals
import           Transient.Indeterminism
import           Control.Monad.State
import           System.IO.Unsafe
import           Control.Monad
import           Control.Applicative
import qualified Streaming.Prelude as Str
import qualified System.IO.Streams as IOS
import           Conduit.Simple as S
import           Control.Exception
import           Criterion.Main
import           Data.Conduit as C
import           Data.Conduit.Combinators as C
--
-- import           Fusion as F hiding ((&))
import           Data.Function ((&))

import           Pipes as P
import qualified Pipes.Prelude as P
import           Test.Hspec hiding (parallel)
import           Test.Hspec.Expectations
import           Data.IORef
import           Data.Atomics 

main :: IO ()
main = do
    -- hspec $ do
    --     describe "basic tests" $
    --         it "passes tests" $ True `shouldBe` True

    defaultMain [
        bgroup "basic"  [ bench "transient"      $ nfIO transient_basic2
                        , bench "stream"         $ nfIO stream_basic
                        , bench "iostreams"      $ nfIO iostreams_basic
                        , bench "pipes"          $ nfIO pipes_basic
                        , bench "conduit"        $ nfIO conduit_basic
                    --    , bench "simple-conduit" $ nfIO simple_conduit_basic
                     --  , bench "fusion"         $ nfIO fusion_basic
                       
                       ]
        ]

         

transient_basic2 :: IO (Maybe Int)
transient_basic2= keep' $ threads 0 $ do
    xs <- group 499000  $  do
             liftIO $ writeIORef count 0
             r <- choose  [1..1000000::Int]
             guard $ even r
             let r'= r+1
             Main.drop' 1000
             let r''= r'+1
             guard $ r'' `mod` 2 == 0
             return r'

    assert (Prelude.length xs == 499000) $
        exit (Prelude.length xs)
    
transient_basic :: IO (Maybe Int)
transient_basic= keep' $ threads 0 $ do
    liftIO $ writeIORef count 0
    xs <- group 499000 $  choose [1..1000000::Int]
          >>= Main.filter even
          >>= Main.map  (+1)
          >>= Main.drop 1000
          >>= Main.map  (+1)
          >>= Main.filter (\x -> x `mod` 2 == 0)
    assert (Prelude.length xs == 499000) $
        exit (Prelude.length xs)
    

 
 

{-# INLINABLE map #-}
map f x= return $ f x

{-# INLINABLE filter #-}
filter cond x= case cond x of
       False -> empty
       True ->  return x

{-# NOINLINE count #-}
count= unsafePerformIO $ newIORef 0
 
drop' :: Int  ->  TransIO ()
drop' num  =  do
   

    mn <- liftIO $ atomicModifyIORefCAS count $ \n ->
            let n'=n +1
            in  if n' < num

              then (n+1,False)
              else (n+1,True)
    case mn of
      False -> stop
      True -> return ()

drop:: Int  -> a -> TransIO a
drop num x =  do
   

    mn <- liftIO $ atomicModifyIORef' count $ \n ->
            let n'=n +1
            in  if n' < num

              then (n+1,Nothing)
              else (n+1,Just x)
    case mn of
      Nothing -> stop
      Just x -> return x

pipes_basic :: IO Int
pipes_basic = do
    xs <- P.toListM $ P.each [1..1000000]
      >-> P.filter even
      >-> P.map (+1)
      >-> P.drop 1000
      >-> P.map (+1)
      >-> P.filter (\x -> x `mod` 2 == 0)
    assert (Prelude.length xs == 499000) $
        return (Prelude.length xs)

conduit_basic :: IO Int
conduit_basic = do
    xs <-   C.yieldMany [1..1000000]
      C.$= C.filter even
      C.$= C.map ((+1) :: Int -> Int)
      C.$= (C.drop 1000 >> C.awaitForever C.yield)
      C.$= C.map ((+1) :: Int -> Int)
      C.$= C.filter (\x -> x `mod` 2 == 0)
      C.$$ C.sinkList
    assert (Prelude.length xs == 499000) $
        return (Prelude.length (xs :: [Int]))

simple_conduit_basic :: IO Int
simple_conduit_basic = do
    xs <-   S.sourceList [1..1000000]
      S.$= S.filterC even
      S.$= S.mapC ((+1) :: Int -> Int)
      S.$= S.dropC 1000
      S.$= S.mapC ((+1) :: Int -> Int)
      S.$= S.filterC (\x -> x `mod` 2 == 0)
      S.$$ S.sinkList
    assert (Prelude.length xs == 499000) $
        return (Prelude.length (xs :: [Int]))



-- fusion_basic :: IO Int
-- fusion_basic = do
--     xs <- F.toListM $ F.each [1..1000000]
--       & F.filter even
--       & F.map (+1)
--       & F.drop 1000
--       & F.map (+1)
--       & F.filter (\x -> x `mod` 2 == 0)
--     assert (Prelude.length xs == 499000) $
--         return (Prelude.length (xs :: [Int]))

stream_basic :: IO Int
stream_basic = do
    xs <- Str.toList $ Str.each [1..1000000]
      & Str.filter even
      & Str.map (+1)
      & Str.drop 1000
      & Str.map (+1)
      & Str.filter (\x -> x `mod` 2 == 0)
    -- assert (Prelude.length xs == 499000) $
    --     return (Prelude.length (xs :: [Int]))
    return 1

iostreams_basic :: IO Int
iostreams_basic = do
  s0 <- IOS.fromList [1..1000000]
  s1 <- IOS.filter even s0
  s2 <- IOS.map (+1) s1
  s3 <- IOS.drop 1000 s2
  s4 <- IOS.map (+1) s3
  s5 <- IOS.filter (\x -> x `mod` 2 == 0) s4
  xs <- IOS.toList s5
  assert (Prelude.length xs == 499000) $
      return (Prelude.length (xs :: [Int]))