e5da40c · streamly · git.iko.soy

git.iko.soy/streamly

Add some doc updates

commite5da40c94c52cb44e58f5b3facd1fec8c77e9d97 authorHarendra Kumar <harendra.kumar@gmail.com> dateFri 14 Jul 2017, 11:07 (9 years ago) parent2756fbd

3 files changed+172−51

compare with master

unifiedsplit

src/Asyncly.hs+118−14 src/Asyncly/AsyncT.hs+37−30 src/Asyncly/Threads.hs+17−7

src/Asyncly.hs+118−14

@@ -7,38 +7,142 @@

77-- Stability : experimental

88-- Portability : GHC

99--

10+--

11+-- Asyncly allows expressing and composing state machines and event driven

12+-- programs in a straightforward manner. It allows waiting for events to occur

13+-- in a non-blocking manner and process them asynchronously; multiple events

14+-- can be processed in parallel. Tasks can be easily split into smaller tasks,

15+-- processed in parallel and results combined.

16+--

17+-- The 'Alternative' composition @(\<|\>)@ is used to express asynchronous or

18+-- parallel tasks. The following example demonstrates generation and printing

19+-- of random numbers in parallel:

20+--

21+-- @

22+-- import Control.Applicative ((\<|\>))

23+-- import Control.Concurrent (threadDelay)

24+-- import Control.Monad.IO.Class (liftIO)

25+-- import System.Random (randomIO)

26+-- import Asyncly

27+--

28+-- main = wait_ $ do

29+-- x <- loop

30+-- liftIO $ print x

31+--

32+-- where

33+--

34+-- loop = do

35+-- liftIO $ threadDelay 1000000

36+-- x <- liftIO (randomIO :: IO Int)

37+-- return x \<|\> loop

38+-- @

39+--

40+-- Here two random number generation loops run in parallel so two numbers are

41+-- printed every second. Note that the threadId printed for each is different:

42+--

43+-- @

44+-- import Control.Applicative ((\<|\>))

45+-- import Control.Concurrent (myThreadId, threadDelay)

46+-- import Control.Monad.IO.Class (liftIO)

47+-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

48+-- import System.Random (randomIO)

49+-- import Asyncly

50+--

51+-- main = wait_ $ do

52+-- liftIO $ hSetBuffering stdout LineBuffering

53+-- x <- loop \"A" \<|\> loop \"B"

54+-- liftIO $ myThreadId >>= putStr . show

55+-- >> putStr " "

56+-- >> print x

57+--

58+-- where

59+--

60+-- loop name = do

61+-- liftIO $ threadDelay 1000000

62+-- rnd <- liftIO (randomIO :: IO Int)

63+-- return (name, rnd) \<|\> loop name

64+-- @

65+-- Here the two loops are serially composed. For each value yielded by loop A,

66+-- loop B is executed. Four results are printed, all four run in separate

67+-- parallel threads. The composition is like ListT except that this is

68+-- concurrent:

69+--

70+-- @

71+-- import Control.Applicative ((\<|\>), empty)

72+-- import Control.Concurrent (myThreadId)

73+-- import Control.Monad.IO.Class (liftIO)

74+-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

75+-- import System.Random (randomIO)

76+-- import Asyncly

77+--

78+-- main = wait_ $ do

79+-- liftIO $ hSetBuffering stdout LineBuffering

80+-- x <- loop "A " 2

81+-- y <- loop "B " 2

82+-- liftIO $ myThreadId >>= putStr . show

83+-- >> putStr " "

84+-- >> print (x, y)

85+--

86+-- where

87+--

88+-- loop name n = do

89+-- rnd <- liftIO (randomIO :: IO Int)

90+-- let result = (name ++ show rnd)

91+-- repeat = if n > 1 then loop name (n - 1) else empty

92+-- in (return result) \<|\> repeat

93+-- @

94+--

95+-- Here we perform a simple multi-threaded map-reduce by squaring each number

96+-- in a separate thread and then summing the squares:

97+--

98+-- @

99+-- import Control.Applicative ((\<|\>), empty)

100+-- import Data.List (sum)

101+-- import Asyncly

102+--

103+-- main = do

104+-- squares <- wait $ do

105+-- x <- foldl (\<|\>) empty $ map return [1..100]

106+-- return (x * x)

107+-- print . sum $ squares

108+-- @

109+--

110+-- 'Applicative' and 'Monoid' compositions work as expected.

10111

11−-- | Asyncly allows writing and composing asynchronous event driven

12−-- applications conveniently. It can be thought of as a non-deterministic

13−-- continuation monad or a combination of ContT and ListT. It allows to capture

14−-- the state of the application at any point and trigger arbitrary number of

15−-- continuations from the capture point. Non-determinism or parallel

16−-- continuations are introduced using the <|> operator. It provides a

17−-- convenient way to implement and compose state machines without using

18−-- callbacks. An event in the state machine corresponds to a continuation.

19−-- There are no cycles in the state machine as each transition in the state

20−-- machine is an independent instance of the state machine. It is an immutable

21−-- state machine!

112+-- It can be thought of as a non-deterministic continuation monad or a

113+-- combination of ContT and ListT. It allows to capture the state of the

114+-- application at any point and trigger arbitrary number of continuations from

115+-- the capture point. Non-determinism or parallel continuations are introduced

116+-- using the <|> operator. It provides a convenient way to implement and

117+-- compose state machines without using callbacks. An event in the state

118+-- machine corresponds to a continuation. There are no cycles in the state

119+-- machine as each transition in the state machine is an independent instance

120+-- of the state machine. It is an immutable state machine!

22121

23122module Asyncly

24123 ( AsyncT

25124 , MonadAsync

125+

126+ -- * Run

26127 , wait

27128 , wait_

28129 , threads

29130

131+ -- * Fork and Gather

30132 , async

31133 , makeAsync

32134 , each

33135 , gather

34136

35− , discardAndThen

137+ -- * Compose

138+ , before

36139 , (*>>)

37− , thenDiscard

140+ , thereafter

38141 , (>>*)

39− , afterFirst

142+ , afterfirst

40143 , (>>|)

41144

145+ -- * Record and Play

42146 , waitRecord

43147 , waitRecord_

44148 , playRecordings

src/Asyncly/AsyncT.hs+37−30

@@ -10,6 +10,16 @@

1010{-# LANGUAGE TypeFamilies #-}

1111{-# LANGUAGE UndecidableInstances #-}

1212

13+-- |

14+-- Module : Asyncly.AsyncT

16+--

17+-- License : MIT-style

18+-- Maintainer : harendra.kumar@gmail.com

19+-- Stability : experimental

20+-- Portability : GHC

21+--

22+--

1323module Asyncly.AsyncT

1424 ( AsyncT (..)

1525 , MonadAsync

@@ -27,11 +37,11 @@ module Asyncly.AsyncT

2737 , makeAsync

2838 , each

2939

30− , discardAndThen

40+ , before

3141 , (*>>)

32− , thenDiscard

42+ , thereafter

3343 , (>>*)

34− , afterFirst

44+ , afterfirst

3545 , (>>|)

3646

3747 -- internal

@@ -438,10 +448,11 @@ runAsyncTask forceAsync action = do

438448-- action was an infinite loop for example.

439449--

440450-- | In an 'Alternative' composition, force the action to run asynchronously.

441−-- The <|> operator implies "can be parallel", whereas async implies "must be

442−-- parallel". Note that async is not useful and should never be used outside an

443−-- 'Alternative' composition. Even in an 'Alternative' composition 'async' is

444−-- not useful in the last action.

451+-- The @\<|\>@ operator implies "can be parallel", whereas 'async' implies

452+-- "must be parallel". Note that outside an 'Alternative' composition 'async'

453+-- is not useful and should not be used. Even in an 'Alternative' composition

454+-- 'async' is not useful as the last action as the last action always runs in

455+-- the current thread.

445456async :: MonadAsync m => AsyncT m a -> AsyncT m a

446457async action = AsyncT $ runAsyncTask True action

447458

@@ -477,9 +488,9 @@ each xs = foldl (<|>) empty $ map return xs

477488------------------------------------------------------------------------------

478489

479490-- XXX Should n be Word32 instead?

480−-- | Runs a computation under a given thread limit. A limit of 0 means new

481−-- tasks start synchronously in the current thread. New threads are created by

482−-- 'parallel', and APIs that use parallel.

491+-- | Runs a computation under a given thread limit. A limit of 0 means all new

492+-- tasks start synchronously in the current thread unless overridden by

493+-- 'async'.

483494threads :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a

484495threads n process = AsyncT $ do

485496 oldCr <- gets threadCredit

@@ -563,44 +574,40 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where

563574infixr 1 >>*, *>>, >>|

564575

565576-- XXX This can be moved to utility functions as it is purely app level

566−-- | Run @b@ once, discarding its result when the first task in task set @a@

567−-- has finished. Useful to start a singleton task after the first task has been

568−-- setup.

569−afterFirst :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

570−afterFirst ma mb = do

577+-- | Run @m b@ right after the first event in @m a@ is generated but before it

578+-- is yielded. The result of @m b@ is discarded.

579+afterfirst :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

580+afterfirst ma mb = do

571581 ref <- liftIO $ newIORef False

572582 x <- ma

573583 done <- liftIO $ readIORef ref

574584 when (not done) $ (liftIO $ writeIORef ref True) >>* mb

575585 return x

576586

587+-- | Same as 'afterfirst'.

577588(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

578−(>>|) = afterFirst

589+(>>|) = afterfirst

579590

580−-- | Run 'm a' in "isolation" and discard its result, and then run 'm b' and

581−-- return its result. Isolation means that any alternative actions inside 'm

582−-- a' are not continued to 'm b'.

583−discardAndThen :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b

584−discardAndThen ma mb = AsyncT $ do

591+-- | Run @m a@ before running @m b@. The result of @m a@ is discarded.

592+before :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b

593+before ma mb = AsyncT $ do

585594 _ <- runAsyncT (ma >> mzero)

586595 runAsyncT mb

587596

588−-- | Same as 'discardAndThen'.

597+-- | Same as 'before'.

589598(*>>) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b

590−(*>>) = discardAndThen

599+(*>>) = before

591600

592−-- | Run 'm a' and then run 'm b' in "isolation" and return the result of 'm

593−-- a'. Isolation means that any alternative actions inside 'm a' are not

594−-- continued to 'm b' and the results of 'm b' are discarded.

595−thenDiscard :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

596−thenDiscard ma mb = AsyncT $ do

601+-- | Run @m b@ after running @m a@. The result of @m b@ is discarded.

602+thereafter :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

603+thereafter ma mb = AsyncT $ do

597604 a <- runAsyncT ma

598605 _ <- runAsyncT (mb >> mzero)

599606 return a

600607

601−-- | Same as 'thenDiscard'.

608+-- | Same as 'thereafter'.

602609(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

603−(>>*) = thenDiscard

610+(>>*) = thereafter

604611

605612-------------------------------------------------------------------------------

606613-- AsyncT transformer

src/Asyncly/Threads.hs+17−7

@@ -1,6 +1,15 @@

11{-# LANGUAGE FlexibleContexts #-}

22{-# LANGUAGE ScopedTypeVariables #-}

4+-- |

5+-- Module : Asyncly.Threads

7+--

8+-- License : MIT-style

9+-- Maintainer : harendra.kumar@gmail.com

10+-- Stability : experimental

11+-- Portability : GHC

12+--

413module Asyncly.Threads

514 ( wait

615 , wait_

@@ -81,14 +90,15 @@ playRecording :: (MonadAsync m, MonadRecorder m)

8190 => AsyncT m a -> Recording -> AsyncT m a

8291playRecording m recording = play recording >> m

8392

84−-- | Compose a computation using previously captured list of logs. Composes a

85−-- group of computations, one for each log.

93+-- | Resume an 'AsyncT' computation using previously recorded logs. The

94+-- recording consists of a list of journals one for each thread in the

95+-- computation.

8696playRecordings :: (MonadAsync m, MonadRecorder m)

8797 => AsyncT m a -> [Recording] -> AsyncT m a

8898playRecordings m logs = each logs >>= playRecording m

8999

90−-- | Run an 'AsyncT m' computation with logging enabled and collect the results

91−-- or logs generated by each thread of the computation.

100+-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish

101+-- returning results for completed threads and recordings for paused threads.

92102waitRecord :: (MonadAsync m, MonadCatch m)

93103 => AsyncT m a -> m ([a], [Recording])

94104waitRecord m = do

@@ -99,9 +109,9 @@ waitRecord m = do

99109 logs <- liftIO $ readIORef lref

100110 return (res, logs)

101111

102−-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish

103−-- and discard the results. If the computation suspends collect the logs to

104−-- replay later.

112+-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish

113+-- and discard the results and return the recordings for paused threads, if

114+-- any.

105115waitRecord_ :: (MonadAsync m, MonadCatch m)

106116 => AsyncT (RecorderT m) a -> m [Recording]

107117waitRecord_ m = do