test/Streamly/Test/Data/Stream/Concurrent.hs · streamly

git.iko.soy/streamly

on abdc6e1 help sign in

streamly/test/Streamly/Test/Data/Stream/Concurrent.hs

363 lines 13 kB history raw

1-- |

2-- Module : Streamly.Test.Data.Stream.Concurrent

4--

5-- License : BSD-3-Clause

6-- Maintainer : streamly@composewell.com

7-- Stability : experimental

8-- Portability : GHC

10module Streamly.Test.Data.Stream.Concurrent (main) where

12#ifdef DEVBUILD

13import Control.Concurrent (threadDelay)

14#endif

15import Control.Exception (Exception, try)

16import Control.Monad (replicateM)

17import Control.Monad.Catch (throwM)

18import Data.List (sort)

19import Data.Word (Word8)

20import Streamly.Data.Stream (Stream)

21import Test.Hspec.QuickCheck

22import Test.QuickCheck (Testable, Property, choose, forAll, withMaxSuccess)

23import Test.QuickCheck.Monadic (monadicIO, run)

24import Test.Hspec as H

26import qualified Streamly.Data.Fold as Fold ( toList )

27import qualified Streamly.Data.Stream as Stream

28 ( replicate, fromEffect, fromPure, fromList, fold, take, nil )

29import qualified Streamly.Internal.Data.Stream.Concurrent as Async

31import Streamly.Test.Common (listEquals)

33moduleName :: String

34moduleName = "Data.Stream.Concurrent"

36-------------------------------------------------------------------------------

37-- Utilities

38-------------------------------------------------------------------------------

40sortEq :: Ord a => [a] -> [a] -> Bool

41sortEq a b = sort a == sort b

43cmp :: (Show a, Ord a) => ([a] -> [a] -> Bool) -> [a] -> Stream IO a -> Property

44cmp eq list s =

45 monadicIO $ do

46 stream <- run $ sort <$> Stream.fold Fold.toList s

47 listEquals eq stream list

49prop1 :: Testable prop => String -> prop -> SpecWith ()

50prop1 x y = modifyMaxSuccess (const 1) $ prop x y

52-- Coverage build takes too long with default number of tests

53maxTestCount :: Int

54#ifdef DEVBUILD

55maxTestCount = 100

56#else

57maxTestCount = 10

58#endif

60transformCombineFromList ::

61 ([Int] -> Stream IO Int)

62 -> ([Int] -> [Int] -> Bool)

63 -> ([Int] -> [Int])

64 -> (Stream IO Int -> Stream IO Int)

65 -> [Int]

66 -> [Int]

67 -> [Int]

68 -> Property

69transformCombineFromList constr eq listOp op a b c =

70 withMaxSuccess maxTestCount $

71 monadicIO $ do

72 let s1 = op (Async.parList id [constr b, constr c])

73 let s2 = Async.parList id [constr a, s1]

74 stream <- run (Stream.fold Fold.toList s2)

75 let list = a <> listOp (b <> c)

76 listEquals eq stream list

78commonOpConfigs :: [(String, Async.Config -> Async.Config)]

79commonOpConfigs =

80 [ ("default", id)

81#ifndef COVERAGE_BUILD

82 , ("rate AvgRate 10000", Async.avgRate 10000)

83 , ("rate Nothing", Async.rate Nothing)

84 , ("maxBuffer 0", Async.maxBuffer 0)

85 , ("maxThreads 0", Async.maxThreads 0)

86 , ("maxThreads 1", Async.maxThreads 1)

87 , ("eager", Async.eager True)

88 -- XXX Need to use an unsorted eq operation for ahead

89 , ("ordered", Async.ordered True)

90#ifdef USE_LARGE_MEMORY

91 , ("maxThreads -1", Async.maxThreads (-1))

92#endif

93#endif

94 ]

96opConfigs :: [(String, Async.Config -> Async.Config)]

97opConfigs = commonOpConfigs

98 ++ [

99#ifndef COVERAGE_BUILD

100 ("maxBuffer 1", Async.maxBuffer 1)

101#endif

102 ]

103

104makeSpec :: [(String, a)] -> (a -> Spec) -> Spec

105makeSpec cfg spec = mapM_ (\(desc, arg) -> describe desc $ spec arg) cfg

106

107asyncSpec :: ((Async.Config -> Async.Config) -> Spec) -> Spec

108asyncSpec =

109 makeSpec $ opConfigs

110#ifndef COVERAGE_BUILD

111 <> [("maxBuffer (-1)", Async.maxBuffer (-1))]

112#endif

113

114-------------------------------------------------------------------------------

115-- Compose with MonadThrow

116-------------------------------------------------------------------------------

117

118newtype ExampleException = ExampleException String deriving (Eq, Show, Ord)

119

120instance Exception ExampleException

121

122exceptionPropagation ::

123 (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec

124exceptionPropagation f = do

125 it "append throwM, nil" $

126 try (tl (Stream.fromEffect (throwM (ExampleException "E")) `f` Stream.nil))

127 `shouldReturn`

128 (Left (ExampleException "E") :: Either ExampleException [Int])

129 it "append nil, throwM" $

130 try (tl (Stream.nil `f` Stream.fromEffect (throwM (ExampleException "E"))))

131 `shouldReturn`

132 (Left (ExampleException "E") :: Either ExampleException [Int])

133 it "append nested throwM" $ do

134 let nested =

135 (Stream.fromList [1..10])

136 `f` Stream.fromEffect (throwM (ExampleException "E"))

137 `f` (Stream.fromList [1..10])

138 try (tl (Stream.nil `f` nested `f` (Stream.fromList [1..10])))

139 `shouldReturn`

140 (Left (ExampleException "E")

141 :: Either ExampleException [Int])

142 it "sequence throwM" $

143 let stream = Stream.fromList [throwM (ExampleException "E")]

144 in try (tl (Stream.nil `f` Async.parSequence id stream))

145 `shouldReturn`

146 (Left (ExampleException "E") :: Either ExampleException [Int])

147

148 it "concatMap throwM" $ do

149 let s1 = Async.parList id $ fmap Stream.fromPure [1..4]

150 s2 = Async.parList id $ fmap Stream.fromPure [5..8]

151 try $ tl (

152 let bind = flip (Async.parConcatMap id)

153 in bind s1 $ \x ->

154 bind s2 $ \y ->

155 if x + y > 10

156 then Stream.fromEffect (throwM (ExampleException "E"))

157 else Stream.fromPure (x + y)

158 )

159 `shouldReturn`

160 (Left (ExampleException "E") :: Either ExampleException [Int])

161

162 where

163

164 tl = Stream.fold Fold.toList

165

166---------------------------------------------------------------------------

167-- Time ordering

168---------------------------------------------------------------------------

169

170#ifdef DEVBUILD

171timeOrdering :: ([Stream IO Int] -> Stream IO Int) -> Spec

172timeOrdering f = do

173 it "Parallel event ordering check" $

174 Stream.fold Fold.toList (f [event 4, event 3, event 2, event 1])

175 `shouldReturn` [1..4]

176

177 where event n = Stream.fromEffect (threadDelay (n * 200000) >> return n)

178#endif

179

180-------------------------------------------------------------------------------

181-- Some ad-hoc tests that failed at times

182-------------------------------------------------------------------------------

183

184takeCombined :: Int -> IO ()

185takeCombined n = do

186 let constr = Stream.fromList

187 let s = Async.parList id [constr ([] :: [Int]), constr ([] :: [Int])]

188 r <- Stream.fold Fold.toList $ Stream.take n s

189 r `shouldBe` []

190

191---------------------------------------------------------------------------

192-- Main

193---------------------------------------------------------------------------

194

195constructWithLenM

196 :: (Int -> Stream IO Int)

197 -> (Int -> IO [Int])

198 -> Word8

199 -> Property

200constructWithLenM mkStream mkList len =

201 withMaxSuccess maxTestCount

202 $ monadicIO $ do

203 stream <-

204 run

205 $ Stream.fold Fold.toList

206 $ mkStream (fromIntegral len)

207 list <- run $ mkList (fromIntegral len)

208 listEquals (==) stream list

209

210sequenceReplicate

211 :: (Async.Config -> Async.Config)

212 -> Word8

213 -> Property

214sequenceReplicate cfg = constructWithLenM stream list

215

216 where

217

218 list = flip replicateM (return 1 :: IO Int)

219 stream = Async.parSequence cfg . flip Stream.replicate (return 1 :: IO Int)

220

221main :: IO ()

222main = hspec

223 $ H.parallel

224#ifdef COVERAGE_BUILD

225 $ modifyMaxSuccess (const 10)

226#endif

227 $ describe moduleName $ do

228 let transform = transformCombineFromList Stream.fromList sortEq

229

230 prop "parEval" $

231 transform

232 (fmap (+2))

233 (fmap (+1) . Async.parEval id . fmap (+1))

234

235 asyncSpec $ prop "parSequence" . sequenceReplicate

236

237 asyncSpec $

238 prop "parMapM (+1)"

239 . transform (fmap (+1))

240 . (`Async.parMapM` (\x -> return (x + 1)))

241

242 -- XXX Need to use eq instead of sortEq for ahead oeprations

243 -- Binary append

244 asyncSpec $

245 let appWith cfg = Async.parList cfg [Stream.nil, Stream.nil]

246 in prop1 "parList [] []" . cmp sortEq ([] :: [Int]) . appWith

247

248 asyncSpec $

249 let appWith cfg = Async.parList cfg [Stream.nil, Stream.fromPure 1]

250 in prop1 "parList [] [1]" . cmp sortEq [1 :: Int] . appWith

251

252 asyncSpec $

253 let appWith cfg = Async.parList cfg [Stream.fromPure 1, Stream.nil]

254 in prop1 "parList [1] []" . cmp sortEq [1 :: Int] . appWith

255

256 asyncSpec $

257 let appWith cfg =

258 Async.parList cfg [Stream.fromPure 0, Stream.fromPure 1]

259 in prop1 "parList [0] [1]" . cmp sortEq [0, 1 :: Int] . appWith

260

261 asyncSpec $

262 let appWith cfg =

263 Async.parList

264 cfg [Stream.fromPure 0, Stream.nil, Stream.fromPure 1]

265 in prop1 "parList [0] [] [1]" . cmp sortEq [0, 1 :: Int] . appWith

266

267 asyncSpec $

268 let appWith cfg =

269 Async.parTwo cfg

270 (Async.parTwo cfg

271 (Async.parTwo cfg

272 (Stream.fromPure 0) (Stream.fromPure 1))

273 (Stream.fromPure 2))

274 (Stream.fromPure 3)

275 in prop1 "parTwo left associated"

276 . cmp sortEq [0, 1, 2, 3 :: Int] . appWith

277

278 asyncSpec $

279 let appWith cfg =

280 Async.parTwo cfg

281 (Stream.fromPure 0)

282 (Async.parTwo cfg

283 (Stream.fromPure 1)

284 (Async.parTwo cfg

285 (Stream.fromPure 2) (Stream.fromPure 3))

286 )

287 in prop1 "parTwo right associated"

288 . cmp sortEq [0, 1, 2, 3 :: Int] . appWith

289

290 asyncSpec $

291 let leaf x y cfg =

292 Async.parTwo cfg (Stream.fromPure x)

293 (Stream.fromPure y)

294 leaf11 cfg =

295 Async.parTwo cfg (leaf 0 1 cfg) $ leaf 2 (3 :: Int) cfg

296 leaf12 cfg =

297 Async.parTwo cfg (leaf 4 5 cfg) $ leaf 6 7 cfg

298 appWith cfg =

299 Async.parTwo cfg (leaf11 cfg) (leaf12 cfg)

300 in prop1 "parTwo balanced"

301 . cmp sortEq [0, 1, 2, 3, 4, 5, 6,7] . appWith

302

303 asyncSpec $

304 let appWith cfg =

305 Async.parTwo cfg

306 (Stream.fromList [1,2,3,4,5 :: Int])

307 (Stream.fromList [6,7,8,9,10])

308 in prop1 "parTwo" . cmp (==) [1,2,3,4,5,6,7,8,9,10] . appWith

309

310 asyncSpec $

311 let par2 cfg =

312 Async.parTwo

313 cfg

314 (Stream.fromPure 1)

315 (Stream.fromPure 2)

316 s1 cfg =

317 Async.parApply

318 cfg

319 (Stream.fromPure (,))

320 (par2 cfg)

321 s2 cfg =

322 Async.parApply

323 cfg

324 (s1 cfg)

325 (Stream.fromPure 3) :: Stream IO (Int, Int)

326 in prop1

327 "parApply (async arg1)" . cmp (==) ( [(1, 3), (2, 3)]) . s2

328

329 asyncSpec $

330 let par2 cfg =

331 Async.parTwo

332 cfg

333 (Stream.fromPure (2 :: Int))

334 (Stream.fromPure 3)

335 s1 = Stream.fromPure (1 :: Int,)

336 s2 cfg = Async.parApply cfg s1 (par2 cfg)

337 in prop1 "apply (async arg2)" . cmp (==) ([(1, 2), (1, 3)]) . s2

338

339 -- concat

340 asyncSpec $

341 let stream cfg =

342 Async.parConcat cfg

343 $ fmap Stream.fromPure

344 $ Stream.fromList [1..100 :: Int]

345 in prop1 "parConcat" . cmp sortEq [1..100] . stream

346

347 asyncSpec $

348 let f cfg =

349 forAll (choose (0, 100)) $ \n ->

350 transform

351 (concatMap (const [1..n]))

352 (Async.parConcatMap

353 cfg (const (Stream.fromList [1..n]))

354 )

355 in prop "parConcatMap" . f

356

357#ifdef DEVBUILD

358 describe "Time ordering" $ timeOrdering (Async.parList id)

359#endif

360 let async = Async.parTwo id

361 describe "Exception propagation" $ exceptionPropagation async

362 -- Ad-hoc tests

363 it "takes n from stream of streams" $ takeCombined 2