e603579 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add concurrent function application operators

e603579Harendra Kumar8 years ago

Changelog.md+5−1

compare with master

unifiedsplit

@@ -2,8 +2,10 @@

3### Breaking changes

5−* Some prelude functions, to which concurrency capability has been added, will

5+* Some prelude functions, to whom concurrency capability has been added, will

6 now require a `MonadAsync` constraint.

7+* Remove the deprecated "<|" operator and reuse it for reverse concurrent fold

8+ application operator.

8### Bug Fixes

10### Bug Fixes

@@ -22,6 +24,8 @@

22 concurrently when used with concurrent stream types.

24 concurrently when used with concurrent stream types.

23* Monad transformation functions `mapM` and `sequence` can now map actions

25* Monad transformation functions `mapM` and `sequence` can now map actions

24 concurrently when used at appropriate stream types.

26 concurrently when used at appropriate stream types.

27+* Added concurrent function application operators to run stages of a

28+ stream processing function application pipeline concurrently.

26## 0.2.0

30## 0.2.0

src/Streamly.hs+34−14

@@ -80,7 +80,18 @@ module Streamly

80 , ZipSerialM

81 , ZipAsyncM

83− -- * Polymorphic Sum Operations

83+ -- * Running Streams

84+ , runStream

85+

86+ -- * Parallel Function Application

87+ -- $application

88+ , (|$)

89+ , (|&)

90+ , (<|)

91+ , (|>)

92+ , mkAsync

93+

94+ -- * Merging Streams

84 -- $sum

95 -- $sum

85 , serial

96 , serial

86 , wSerial

97 , wSerial

@@ -89,6 +100,12 @@ module Streamly

89 , wAsync

100 , wAsync

90 , parallel

101 , parallel

102

103+ -- * Folding Containers of Streams

104+ -- $foldutils

105+ , foldWith

106+ , foldMapWith

107+ , forEachWith

108+

92 -- * Stream Type Adapters

109 -- * Stream Type Adapters

93 -- $adapters

110 -- $adapters

94 , IsStream ()

111 , IsStream ()

@@ -113,18 +130,6 @@ module Streamly

113 , ZipSerial

130 , ZipSerial

114 , ZipAsync

131 , ZipAsync

115

132

116− -- * Running Streams

117− , runStream

118−

119− -- * Transformation

120− , mkAsync

121−

122− -- * Polymorphic Fold Utilities

123− -- $foldutils

124− , foldWith

125− , foldMapWith

126− , forEachWith

127−

128 -- * Re-exports

133 -- * Re-exports

129 , Semigroup (..)

134 , Semigroup (..)

130 -- * Deprecated

135 -- * Deprecated

@@ -143,7 +148,6 @@ module Streamly

143 , zipping

148 , zipping

144 , zippingAsync

149 , zippingAsync

145 , (<=>)

150 , (<=>)

146− , (<|)

147 )

151 )

148where

152where

149

153

@@ -174,6 +178,22 @@ import Data.Semigroup (Semigroup(..))

174-- corresponding elements of two streams together. Note that these types are

178-- corresponding elements of two streams together. Note that these types are

175-- not monads.

179-- not monads.

176

180

181+-- $application

182+--

183+-- Several stream processing functions can be called in a chain using regular

184+-- function application, using the '$' operator or reverse function application

185+-- operator '&'. This section provides corresponding operators that can apply

186+-- stream processing functions such that ach stage of the stream can run in

187+-- parallel.

188+--

189+-- Imports for the code snippets below:

190+--

191+-- @

192+-- import Streamly

193+-- import qualified Streamly.Prelude as S

194+-- import Control.Concurrent

195+-- @

196+

177-- $sum

197-- $sum

178-- The 'Semigroup' operation '<>' of each stream type combines two streams in a

198-- The 'Semigroup' operation '<>' of each stream type combines two streams in a

179-- type specific manner. This section provides polymorphic versions of '<>'

199-- type specific manner. This section provides polymorphic versions of '<>'

src/Streamly/Core.hs+86−29

@@ -48,6 +48,10 @@ module Streamly.Core

48 , wAsync

49 , parallel

51+ -- * applications

52+ , applyWith

53+ , runWith

54+

51 -- * zip

55 -- * zip

52 , zipWith

56 , zipWith

53 , zipAsyncWith

57 , zipAsyncWith

@@ -66,18 +70,10 @@ module Streamly.Core

66 )

70 )

67where

71where

69−-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark,

70−-- but keep it on for now to debug problems quickly if and when they happen.

71−#define DIAGNOSTICS

72−

73import Control.Concurrent (ThreadId, myThreadId, threadDelay)

74import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

75 tryPutMVar, takeMVar, putMVar)

76import Control.Exception (SomeException (..))

77−#ifdef DIAGNOSTICS

78−import Control.Exception (catch,

79− BlockedIndefinitelyOnMVar(..))

80−#endif

81import qualified Control.Exception.Lifted as EL

77import qualified Control.Exception.Lifted as EL

82import Control.Monad (when)

78import Control.Monad (when)

83import Control.Monad.Catch (MonadThrow, throwM)

79import Control.Monad.Catch (MonadThrow, throwM)

@@ -105,6 +101,17 @@ import GHC.Exts

105import GHC.Conc (ThreadId(..))

101import GHC.Conc (ThreadId(..))

106import GHC.IO (IO(..))

102import GHC.IO (IO(..))

107

103

104+-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark,

105+-- but keep it on for now to debug problems quickly if and when they happen.

106+#define DIAGNOSTICS

107+

108+#ifdef DIAGNOSTICS

109+import Control.Exception (catches, throwIO, Handler(..),

110+ BlockedIndefinitelyOnMVar(..),

111+ BlockedIndefinitelyOnSTM(..))

112+import System.IO (hPutStrLn, stderr)

113+#endif

114+

108------------------------------------------------------------------------------

115------------------------------------------------------------------------------

109-- Parent child thread communication type

116-- Parent child thread communication type

110------------------------------------------------------------------------------

117------------------------------------------------------------------------------

@@ -193,41 +200,63 @@ data SVar m a =

193{-# NOINLINE dumpSVar #-}

200{-# NOINLINE dumpSVar #-}

194dumpSVar :: SVar m a -> IO String

201dumpSVar :: SVar m a -> IO String

195dumpSVar sv = do

202dumpSVar sv = do

203+ tid <- myThreadId

196 (oqList, oqLen) <- readIORef $ outputQueue sv

204 (oqList, oqLen) <- readIORef $ outputQueue sv

197 db <- tryTakeMVar $ doorBell sv

205 db <- tryTakeMVar $ doorBell sv

198 srn <- tryTakeMVar $ siren sv

206 srn <- tryTakeMVar $ siren sv

207+ aheadDump <-

208+ if svarStyle sv == AheadVar

209+ then do

199− (oheap, oheapSize) <- readIORef $ outputHeap sv

210+ (oheap, oheapSize) <- readIORef $ outputHeap sv

200− hsrn <- tryTakeMVar $ heapSiren sv

211+ hsrn <- tryTakeMVar $ heapSiren sv

201− (wq, wqSize) <- readIORef $ workQueue sv

212+ (wq, wqSize) <- readIORef $ workQueue sv

213+ return $ unlines

214+ [ "heap length computed = " ++ show (H.size oheap)

215+ , "heap length maintained = " ++ show oheapSize

216+ , "heap siren = " ++ show hsrn

217+ , "work queue length computed = " ++ show (length wq)

218+ , "work queue length maintained = " ++ show wqSize

219+ ]

220+ else return []

221+

202 waiting <- readIORef $ waitingForWork sv

222 waiting <- readIORef $ waitingForWork sv

203 rthread <- readIORef $ runningThreads sv

223 rthread <- readIORef $ runningThreads sv

204 workers <- readIORef $ activeWorkers sv

224 workers <- readIORef $ activeWorkers sv

205 -- XXX queueEmpty should be made IO return type

225 -- XXX queueEmpty should be made IO return type

206

226

207 return $ unlines

227 return $ unlines

228+ [ "tid = " ++ show tid

208− [ "style = " ++ show (svarStyle sv)

229+ , "style = " ++ show (svarStyle sv)

209 , "outputQueue length computed = " ++ show (length oqList)

230 , "outputQueue length computed = " ++ show (length oqList)

210 , "outputQueue length maintained = " ++ show oqLen

231 , "outputQueue length maintained = " ++ show oqLen

211 , "output doorBell = " ++ show db

232 , "output doorBell = " ++ show db

212 , "worker siren = " ++ show srn

233 , "worker siren = " ++ show srn

213− , "heap length computed = " ++ show (H.size oheap)

234+ ]

214− , "heap length maintained = " ++ show oheapSize

235+ ++ aheadDump ++ unlines

215− , "heap siren = " ++ show hsrn

216− , "work queue length computed = " ++ show (length wq)

217− , "work queue length maintained = " ++ show wqSize

218− , "waitingForWork = " ++ show waiting

236+ [ "waitingForWork = " ++ show waiting

219 , "running threads = " ++ show rthread

237 , "running threads = " ++ show rthread

220 , "running thread count = " ++ show workers

238 , "running thread count = " ++ show workers

221 ]

239 ]

222

240

223{-# NOINLINE mvarExcHandler #-}

241{-# NOINLINE mvarExcHandler #-}

224mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()

242mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()

225−mvarExcHandler sv label BlockedIndefinitelyOnMVar = do

243+mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do

226 svInfo <- dumpSVar sv

244 svInfo <- dumpSVar sv

227− error $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo

245+ hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo

246+ throwIO e

247+

248+{-# NOINLINE stmExcHandler #-}

249+stmExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnSTM -> IO ()

250+stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do

251+ svInfo <- dumpSVar sv

252+ hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo

253+ throwIO e

228

254

229withDBGMVar :: SVar m a -> String -> IO () -> IO ()

255withDBGMVar :: SVar m a -> String -> IO () -> IO ()

230−withDBGMVar sv label action = action `catch` mvarExcHandler sv label

256+withDBGMVar sv label action =

257+ action `catches` [ Handler (mvarExcHandler sv label)

258+ , Handler (stmExcHandler sv label)

259+ ]

231#else

260#else

232withDBGMVar :: SVar m a -> String -> IO () -> IO ()

261withDBGMVar :: SVar m a -> String -> IO () -> IO ()

233withDBGMVar _ _ action = action

262withDBGMVar _ _ action = action

@@ -759,7 +788,7 @@ delThread sv tid =

759-- If present then delete else add. This takes care of out of order add and

788-- If present then delete else add. This takes care of out of order add and

760-- delete i.e. a delete arriving before we even added a thread.

789-- delete i.e. a delete arriving before we even added a thread.

761-- This occurs when the forked thread is done even before the 'addThread' right

790-- This occurs when the forked thread is done even before the 'addThread' right

762−-- after the fork is executed.

791+-- after the fork gets a chance to run.

763{-# INLINE modifyThread #-}

792{-# INLINE modifyThread #-}

764modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()

793modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()

765modifyThread sv tid = do

794modifyThread sv tid = do

@@ -797,7 +826,10 @@ pushWorker sv = do

797-- using a CAS based modification.

826-- using a CAS based modification.

798{-# NOINLINE pushWorkerPar #-}

827{-# NOINLINE pushWorkerPar #-}

799pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()

828pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()

800−pushWorkerPar sv m =

829+pushWorkerPar sv m = do

830+ -- We do not use activeWorkers in case of ParallelVar but still there is no

831+ -- harm in maintaining it correctly.

832+ liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1

801 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv

833 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv

802

834

803-- XXX When the queue is LIFO we can put a limit on the number of dispatches.

835-- XXX When the queue is LIFO we can put a limit on the number of dispatches.

@@ -1050,8 +1082,11 @@ newStreamVar1 style m = do

1050 -- Note: We must have all the work on the queue before sending the

1082 -- Note: We must have all the work on the queue before sending the

1051 -- pushworker, otherwise the pushworker may exit before we even get a

1083 -- pushworker, otherwise the pushworker may exit before we even get a

1052 -- chance to push.

1084 -- chance to push.

1085+ if style == ParallelVar

1086+ then pushWorkerPar sv m

1087+ else do

1053− liftIO $ (enqueue sv) m

1088+ liftIO $ (enqueue sv) m

1054− pushWorker sv

1089+ pushWorker sv

1055 return sv

1090 return sv

1056

1091

1057-- | Create a new SVar and enqueue one stream computation on it.

1092-- | Create a new SVar and enqueue one stream computation on it.

@@ -1245,6 +1280,26 @@ _alt m1 m2 = Stream $ \_ stp sng yld ->

1245 let stop = runStream m2 Nothing stp sng yld

1280 let stop = runStream m2 Nothing stp sng yld

1246 in runStream m1 Nothing stop sng yld

1281 in runStream m1 Nothing stop sng yld

1247

1282

1283+------------------------------------------------------------------------------

1284+-- Stream to stream function application

1285+------------------------------------------------------------------------------

1286+

1287+applyWith :: MonadAsync m

1288+ => SVarStyle -> (Stream m a -> Stream m b) -> Stream m a -> Stream m b

1289+applyWith style f m = Stream $ \svr stp sng yld -> do

1290+ sv <- newStreamVar1 style m

1291+ runStream (f $ fromStreamVar sv) svr stp sng yld

1292+

1293+------------------------------------------------------------------------------

1294+-- Stream runner function application

1295+------------------------------------------------------------------------------

1296+

1297+runWith :: MonadAsync m

1298+ => SVarStyle -> (Stream m a -> m b) -> Stream m a -> m b

1299+runWith style f m = do

1300+ sv <- newStreamVar1 style m

1301+ f $ fromStreamVar sv

1302+

1248------------------------------------------------------------------------------

1303------------------------------------------------------------------------------

1249-- Zipping

1304-- Zipping

1250------------------------------------------------------------------------------

1305------------------------------------------------------------------------------

@@ -1261,10 +1316,6 @@ zipWith f m1 m2 = go m1 m2

1261 yield1 a ra = merge a ra

1316 yield1 a ra = merge a ra

1262 (runStream mx) Nothing stp single1 yield1

1317 (runStream mx) Nothing stp single1 yield1

1263

1318

1264−mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)

1265−mkAsync m = newStreamVar1 AsyncVar m

1266− >>= return . fromStreamVar

1267−

1268zipAsyncWith :: MonadAsync m

1319zipAsyncWith :: MonadAsync m

1269 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

1320 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

1270zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do

1321zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do

@@ -1272,6 +1323,12 @@ zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do

1272 mb <- mkAsync m2

1323 mb <- mkAsync m2

1273 (runStream (zipWith f ma mb)) Nothing stp sng yld

1324 (runStream (zipWith f ma mb)) Nothing stp sng yld

1274

1325

1326+ where

1327+

1328+ mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)

1329+ mkAsync m = newStreamVar1 AsyncVar m

1330+ >>= return . fromStreamVar

1331+

1275-------------------------------------------------------------------------------

1332-------------------------------------------------------------------------------

1276-- Transformers

1333-- Transformers

1277-------------------------------------------------------------------------------

1334-------------------------------------------------------------------------------

src/Streamly/Prelude.hs+47

@@ -191,6 +191,13 @@ unfoldr step = fromStream . go

191-- 2

192-- 3

193-- @

194+-- When run concurrently, the next unfold can run concurrently while previously

195+-- generated value is being processed by the next stream stage.

196+--

197+-- @

198+-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)

199+-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()

200+-- @

194--

201--

195-- /Concurrent/

202-- /Concurrent/

196--

203--

@@ -215,6 +222,11 @@ fromFoldable = Prelude.foldr cons nil

215

222

216-- | Construct a stream from a 'Foldable' containing monadic actions.

223-- | Construct a stream from a 'Foldable' containing monadic actions.

217--

224--

225+-- @

226+-- runStream $ serially $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1)

227+-- runStream $ asyncly $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1)

228+-- @

229+--

218-- /Concurrent (do not use with 'parallely' on infinite containers)/

230-- /Concurrent (do not use with 'parallely' on infinite containers)/

219--

231--

220-- @since 0.3.0

232-- @since 0.3.0

@@ -245,6 +257,12 @@ once = fromStream . S.once

245

257

246-- | Generate a stream by performing a monadic action @n@ times.

258-- | Generate a stream by performing a monadic action @n@ times.

247--

259--

260+--

261+-- @

262+-- runStream $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)

263+-- runStream $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)

264+-- @

265+--

248-- /Concurrent/

266-- /Concurrent/

249--

267--

250-- @since 0.1.1

268-- @since 0.1.1

@@ -255,6 +273,11 @@ replicateM n m = go n

255

273

256-- | Generate a stream by repeatedly executing a monadic action forever.

274-- | Generate a stream by repeatedly executing a monadic action forever.

257--

275--

276+-- @

277+-- runStream $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)

278+-- runStream $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)

279+-- @

280+--

258-- /Concurrent, infinite (do not use with 'parallely')/

281-- /Concurrent, infinite (do not use with 'parallely')/

259--

282--

260-- @since 0.2.0

283-- @since 0.2.0

@@ -273,6 +296,14 @@ iterate step = fromStream . go

273-- | Iterate a monadic function from a seed value, streaming the results

296-- | Iterate a monadic function from a seed value, streaming the results

274-- forever.

297-- forever.

275--

298--

299+-- @

300+-- runStream $ serially $ S.take 10 $ S.iterateM

301+-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0

302+--

303+-- runStream $ asyncly $ S.take 10 $ S.iterateM

304+-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0

305+-- @

306+--

276-- /Concurrent, infinite (do not use with 'parallely')/

307-- /Concurrent, infinite (do not use with 'parallely')/

277--

308--

278-- @since 0.1.2

309-- @since 0.1.2

@@ -699,6 +730,14 @@ maximum m = go Nothing (toStream m)

699-- | Replace each element of the stream with the result of a monadic action

730-- | Replace each element of the stream with the result of a monadic action

700-- applied on the element.

731-- applied on the element.

701--

732--

733+-- @

734+-- runStream $ S.replicateM 10 (return 1)

735+-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))

736+--

737+-- runStream $ S.replicateM 10 (return 1)

738+-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))

739+-- @

740+--

702-- /Concurrent (do not use with 'parallely' on infinite streams)/

741-- /Concurrent (do not use with 'parallely' on infinite streams)/

703--

742--

704-- @since 0.1.0

743-- @since 0.1.0

@@ -728,6 +767,14 @@ mapM_ f m = go (toStream m)

728-- | Reduce a stream of monadic actions to a stream of the output of those

767-- | Reduce a stream of monadic actions to a stream of the output of those

729-- actions.

768-- actions.

730--

769--

770+-- @

771+-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)

772+-- & (serially . S.sequence)

773+--

774+-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)

775+-- & (asyncly . S.sequence)

776+-- @

777+--

731-- /Concurrent (do not use with 'parallely' on infinite streams)/

778-- /Concurrent (do not use with 'parallely' on infinite streams)/

732--

779--

733-- @since 0.1.0

780-- @since 0.1.0

src/Streamly/Streams.hs+91−9

@@ -41,6 +41,10 @@ module Streamly.Streams

42 -- * Transformation

43 , mkAsync

44+ , (|$)

45+ , (|&)

46+ , (|>)

47+ , (<|)

45 -- * Merging Streams

49 -- * Merging Streams

46 , serial

50 , serial

@@ -50,7 +54,6 @@ module Streamly.Streams

50 , wAsync

54 , wAsync

51 , parallel

55 , parallel

52 , (<=>) --deprecated

56 , (<=>) --deprecated

53− , (<|) --deprecated

55 -- * IO Streams

58 -- * IO Streams

56 , Serial

59 , Serial

@@ -154,10 +157,18 @@ class IsStream t where

154 -- ["hello","world"]

157 -- ["hello","world"]

155 -- @

158 -- @

156 --

159 --

160+ -- @

161+ -- let delay = threadDelay 1000000 >> print 1

162+ -- runStream $ serially $ delay |: delay |: delay |: nil

163+ -- runStream $ parallely $ delay |: delay |: delay |: nil

164+ -- @

165+ --

157 -- /Concurrent (do not use 'parallely' to construct infinite streams)/

166 -- /Concurrent (do not use 'parallely' to construct infinite streams)/

158 --

167 --

159 -- @since 0.2.0

168 -- @since 0.2.0

160 (|:) :: MonadAsync m => m a -> t m a -> t m a

169 (|:) :: MonadAsync m => m a -> t m a -> t m a

170+ -- We can define (|:) just as 'consM' but it is defined explicitly for each

171+ -- type because we want to use SPECIALIZE pragma on the definition.

161

172

162-- | Same as 'IsStream'.

173-- | Same as 'IsStream'.

163--

174--

@@ -307,6 +318,85 @@ mkAsync m = do

307 sv <- S.newStreamVar1 AsyncVar (toStream m)

318 sv <- S.newStreamVar1 AsyncVar (toStream m)

308 return $ fromSVar sv

319 return $ fromSVar sv

309

320

321+{-# INLINE applyWith #-}

322+applyWith :: (IsStream t, MonadAsync m)

323+ => SVarStyle -> (t m a -> t m b) -> t m a -> t m b

324+applyWith style f x = fromStream $

325+ S.applyWith style (toStream . f . fromStream) (toStream x)

326+

327+{-# INLINE runWith #-}

328+runWith :: (IsStream t, MonadAsync m)

329+ => SVarStyle -> (t m a -> m b) -> t m a -> m b

330+runWith style f x = S.runWith style (f . fromStream) (toStream x)

331+

332+-- | Parallel function application operator for streams; just like the regular

333+-- function application operator '$' except that it is concurrent. The

334+-- following code prints a value every second even though each stage adds a 1

335+-- second delay.

336+--

337+--

338+-- @

339+-- runStream $

340+-- S.mapM (\\x -> threadDelay 1000000 >> print x)

341+-- |$ S.repeatM (threadDelay 1000000 >> return 1)

342+-- @

343+--

344+-- /Concurrent/

345+--

346+-- @since 0.3.0

347+{-# INLINE (|$) #-}

348+(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b

349+f |$ x = applyWith ParallelVar f x

350+

351+-- | Parallel reverse function application operator for streams; just like the

352+-- regular reverse function application operator '&' except that it is

353+-- concurrent.

354+--

355+-- @

356+-- runStream $

357+-- S.repeatM (threadDelay 1000000 >> return 1)

358+-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)

359+-- @

360+--

361+-- /Concurrent/

362+--

363+-- @since 0.3.0

364+{-# INLINE (|&) #-}

365+(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b

366+x |& f = f |$ x

367+

368+-- | Parallel function application operator; applies a @run@ or @fold@ function

369+-- to a stream such that the fold consumer and the stream producer run in

370+-- parallel. A @run@ or @fold@ function reduces the stream to a value in the

371+-- underlying monad.

372+--

373+-- @

374+-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()

375+-- <| S.repeatM (threadDelay 1000000 >> return 1)

376+-- @

377+--

378+-- /Concurrent/

379+--

380+-- @since 0.3.0

381+{-# INLINE (<|) #-}

382+(<|) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b

383+f <| x = runWith ParallelVar f x

384+

385+-- | Parallel reverse function application operator for applying a run or fold

386+-- functions to a stream. Just like '<|' except that the operands are reversed.

387+--

388+-- @

389+-- S.repeatM (threadDelay 1000000 >> return 1)

390+-- |> S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()

391+-- @

392+--

393+-- /Concurrent/

394+--

395+-- @since 0.3.0

396+{-# INLINE (|>) #-}

397+(|>) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b

398+x |> f = f <| x

399+

310------------------------------------------------------------------------------

400------------------------------------------------------------------------------

311-- CPP macros for common instances

401-- CPP macros for common instances

312------------------------------------------------------------------------------

402------------------------------------------------------------------------------

@@ -776,14 +866,6 @@ async m1 m2 = fromStream $ S.async (toStream m1) (toStream m2)

776instance MonadAsync m => Semigroup (AsyncT m a) where

866instance MonadAsync m => Semigroup (AsyncT m a) where

777 (<>) = async

867 (<>) = async

778

868

779−-- | Same as 'async'.

780−--

781−-- @since 0.1.0

782−{-# DEPRECATED (<|) "Please use 'async' instead." #-}

783−{-# INLINE (<|) #-}

784−(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

785−(<|) = async

786−

787------------------------------------------------------------------------------

869------------------------------------------------------------------------------

788-- Monoid

870-- Monoid

789------------------------------------------------------------------------------

871------------------------------------------------------------------------------

src/Streamly/Tutorial.hs+1

@@ -386,6 +386,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

386-- where required.

387--

388-- TBD add more details and examples about concurrent transformation.

389+-- TBD add details about concurrent function application.

389

390

390-- $semigroup

391-- $semigroup

391--

392--

test/Prop.hs+77−4

@@ -2,7 +2,8 @@

3module Main (main) where

5−import Control.Exception (BlockedIndefinitelyOnMVar(..), catch)

5+import Control.Exception (BlockedIndefinitelyOnMVar(..), catches,

6+ BlockedIndefinitelyOnSTM(..), Handler(..))

6import Control.Monad (when)

7import Control.Monad (when)

7import Control.Applicative (ZipList(..))

8import Control.Applicative (ZipList(..))

8import Control.Concurrent (MVar, takeMVar, putMVar, newEmptyMVar)

9import Control.Concurrent (MVar, takeMVar, putMVar, newEmptyMVar)

@@ -68,10 +69,19 @@ transformFromList constr eq listOp op a =

68 let list = listOp a

69 let list = listOp a

69 equals eq stream list

70 equals eq stream list

72+mvarExcHandler :: String -> BlockedIndefinitelyOnMVar -> IO ()

73+mvarExcHandler label BlockedIndefinitelyOnMVar = do

74+ error $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n"

75+

76+stmExcHandler :: String -> BlockedIndefinitelyOnSTM -> IO ()

77+stmExcHandler label BlockedIndefinitelyOnSTM = do

78+ error $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n"

79+

71−dbgMVar :: String -> IO a -> IO a

80+dbgMVar :: String -> IO () -> IO ()

72dbgMVar label action =

81dbgMVar label action =

73− action `catch` \BlockedIndefinitelyOnMVar ->

82+ action `catches` [ Handler (mvarExcHandler label)

74− error $ label ++ " " ++ "BlockedIndefinitelyOnMVar"

83+ , Handler (stmExcHandler label)

84+ ]

76-- | first n actions takeMVar and the last action performs putMVar n times

86-- | first n actions takeMVar and the last action performs putMVar n times

77mvarSequenceOp :: MVar () -> Word8 -> Word8 -> IO Word8

87mvarSequenceOp :: MVar () -> Word8 -> Word8 -> IO Word8

@@ -131,6 +141,7 @@ concurrentUnfoldrM

131 -> Property

141 -> Property

132concurrentUnfoldrM eq op n =

142concurrentUnfoldrM eq op n =

133 monadicIO $ do

143 monadicIO $ do

144+ -- XXX we should test empty list case as well

134 let list = [0..n]

145 let list = [0..n]

135 stream <- run $ do

146 stream <- run $ do

136 -- putStrLn $ "concurrentUnfoldrM: " ++ show n

147 -- putStrLn $ "concurrentUnfoldrM: " ++ show n

@@ -154,6 +165,61 @@ concurrentUnfoldrM eq op n =

154 return x

165 return x

155 equals eq stream list

166 equals eq stream list

156

167

168+concurrentApplication :: Word8 -> Property

169+concurrentApplication n =

170+ monadicIO $ do

171+ -- XXX we should test empty list case as well

172+ let list = [0..n]

173+ stream <- run $ do

174+ -- putStrLn $ "concurrentApplication: " ++ show n

175+ mv <- newEmptyMVar :: IO (MVar ())

176+ -- since unfoldr happens in parallel with the stream processing we

177+ -- can do two takeMVar in one iteration. If it is not parallel then

178+ -- this will not work and the test will fail.

179+ A.toList $ do

180+ sourceUnfoldrM mv n |&

181+ (A.mapM $ \x -> do

182+ let msg = show x ++ "/" ++ show n

183+ if even x

184+ then do

185+ dbgMVar ("first take concurrentApp " ++ msg)

186+ (takeMVar mv)

187+ if n > x

188+ then dbgMVar ("second take concurrentApp " ++ msg)

189+ (takeMVar mv)

190+ else return ()

191+ else return ()

192+ return x)

193+ equals (==) stream list

194+

195+sourceUnfoldrM1 :: IsStream t => Word8 -> t IO Word8

196+sourceUnfoldrM1 n = A.unfoldrM step 0

197+ where

198+ -- argument must be integer to avoid overflow of word8 at 255

199+ step :: Int -> IO (Maybe (Word8, Int))

200+ step cnt = do

201+ if cnt > fromIntegral n

202+ then return Nothing

203+ else return (Just (fromIntegral cnt, cnt + 1))

204+

205+concurrentFoldlApplication :: Word8 -> Property

206+concurrentFoldlApplication n =

207+ monadicIO $ do

208+ -- XXX we should test empty list case as well

209+ let list = [0..n]

210+ stream <- run $ do

211+ sourceUnfoldrM1 n |> A.foldlM' (\xs x -> return (x : xs)) []

212+ equals (==) (reverse stream) list

213+

214+concurrentFoldrApplication :: Word8 -> Property

215+concurrentFoldrApplication n =

216+ monadicIO $ do

217+ -- XXX we should test empty list case as well

218+ let list = [0..n]

219+ stream <- run $ do

220+ sourceUnfoldrM1 n |> A.foldrM (\x xs -> return (x : xs)) []

221+ equals (==) stream list

222+

157foldFromList

223foldFromList

158 :: ([Int] -> t IO Int)

224 :: ([Int] -> t IO Int)

159 -> (t IO Int -> SerialT IO Int)

225 -> (t IO Int -> SerialT IO Int)

@@ -564,6 +630,13 @@ main = hspec $ do

564 concurrentOps folded "wAsyncly folded" wAsyncly sortEq

630 concurrentOps folded "wAsyncly folded" wAsyncly sortEq

565 concurrentOps folded "parallely folded" parallely sortEq

631 concurrentOps folded "parallely folded" parallely sortEq

566

632

633+ prop "concurrent application" $ withMaxSuccess maxTestCount $

634+ concurrentApplication

635+ prop "concurrent foldr application" $ withMaxSuccess maxTestCount $

636+ concurrentFoldrApplication

637+ prop "concurrent foldl application" $ withMaxSuccess maxTestCount $

638+ concurrentFoldlApplication

639+

567 describe "Stream elimination operations" $ do

640 describe "Stream elimination operations" $ do

568 eliminationOps A.fromFoldable "serially" serially

641 eliminationOps A.fromFoldable "serially" serially

569 eliminationOps A.fromFoldable "aheadly" aheadly

642 eliminationOps A.fromFoldable "aheadly" aheadly