git.iko.soy/streamly
on master help sign in
Add some doc updates
e5da40cHarendra Kumar
src/Asyncly.hs+118−14
compare with master
unifiedsplit
@@ -7,38 +7,142 @@
7-- Stability : experimental
7-- Stability : experimental
8-- Portability : GHC
8-- Portability : GHC
9--
9--
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.
10
111
11−-- | Asyncly allows writing and composing asynchronous event driven
12−-- applications conveniently. It can be thought of as a non-deterministic
112+-- It can be thought of as a non-deterministic continuation monad or a
13−-- continuation monad or a combination of ContT and ListT. It allows to capture
113+-- combination of ContT and ListT. It allows to capture the state of the
14−-- the state of the application at any point and trigger arbitrary number of
114+-- application at any point and trigger arbitrary number of continuations from
15−-- continuations from the capture point. Non-determinism or parallel
115+-- the capture point. Non-determinism or parallel continuations are introduced
16−-- continuations are introduced using the <|> operator. It provides a
116+-- using the <|> operator. It provides a convenient way to implement and
17−-- convenient way to implement and compose state machines without using
117+-- compose state machines without using callbacks. An event in the state
18−-- callbacks. An event in the state machine corresponds to a continuation.
118+-- machine corresponds to a continuation. There are no cycles in the state
19−-- There are no cycles in the state machine as each transition in the state
119+-- machine as each transition in the state machine is an independent instance
20−-- machine is an independent instance of the state machine. It is an immutable
120+-- of the state machine. It is an immutable state machine!
21−-- state machine!
22
121
23module Asyncly
122module Asyncly
24 ( AsyncT
123 ( AsyncT
25 , MonadAsync
124 , MonadAsync
125+
126+ -- * Run
26 , wait
127 , wait
27 , wait_
128 , wait_
28 , threads
129 , threads
29
130
131+ -- * Fork and Gather
30 , async
132 , async
31 , makeAsync
133 , makeAsync
32 , each
134 , each
33 , gather
135 , gather
34
136
35− , discardAndThen
137+ -- * Compose
138+ , before
36 , (*>>)
139 , (*>>)
37− , thenDiscard
140+ , thereafter
38 , (>>*)
141 , (>>*)
39− , afterFirst
142+ , afterfirst
40 , (>>|)
143 , (>>|)
41
144
145+ -- * Record and Play
42 , waitRecord
146 , waitRecord
43 , waitRecord_
147 , waitRecord_
44 , playRecordings
148 , playRecordings
src/Asyncly/AsyncT.hs+37−30
@@ -10,6 +10,16 @@
10{-# LANGUAGE TypeFamilies #-}
10{-# LANGUAGE TypeFamilies #-}
11{-# LANGUAGE UndecidableInstances #-}
11{-# LANGUAGE UndecidableInstances #-}
12
12
13+-- |
14+-- Module : Asyncly.AsyncT
15+-- Copyright : (c) 2017 Harendra Kumar
16+--
17+-- License : MIT-style
18+-- Maintainer : harendra.kumar@gmail.com
19+-- Stability : experimental
20+-- Portability : GHC
21+--
22+--
13module Asyncly.AsyncT
23module Asyncly.AsyncT
14 ( AsyncT (..)
24 ( AsyncT (..)
15 , MonadAsync
25 , MonadAsync
@@ -27,11 +37,11 @@ module Asyncly.AsyncT
27 , makeAsync
37 , makeAsync
28 , each
38 , each
29
39
30− , discardAndThen
40+ , before
31 , (*>>)
41 , (*>>)
32− , thenDiscard
42+ , thereafter
33 , (>>*)
43 , (>>*)
34− , afterFirst
44+ , afterfirst
35 , (>>|)
45 , (>>|)
36
46
37 -- internal
47 -- internal
@@ -438,10 +448,11 @@ runAsyncTask forceAsync action = do
438-- action was an infinite loop for example.
448-- action was an infinite loop for example.
439--
449--
440-- | In an 'Alternative' composition, force the action to run asynchronously.
450-- | In an 'Alternative' composition, force the action to run asynchronously.
441−-- The <|> operator implies "can be parallel", whereas async implies "must be
451+-- The @\<|\>@ operator implies "can be parallel", whereas 'async' implies
442−-- parallel". Note that async is not useful and should never be used outside an
452+-- "must be parallel". Note that outside an 'Alternative' composition 'async'
443−-- 'Alternative' composition. Even in an 'Alternative' composition 'async' is
453+-- is not useful and should not be used. Even in an 'Alternative' composition
444−-- not useful in the last action.
454+-- 'async' is not useful as the last action as the last action always runs in
455+-- the current thread.
445async :: MonadAsync m => AsyncT m a -> AsyncT m a
456async :: MonadAsync m => AsyncT m a -> AsyncT m a
446async action = AsyncT $ runAsyncTask True action
457async action = AsyncT $ runAsyncTask True action
447
458
@@ -477,9 +488,9 @@ each xs = foldl (<|>) empty $ map return xs
477------------------------------------------------------------------------------
488------------------------------------------------------------------------------
478
489
479-- XXX Should n be Word32 instead?
490-- XXX Should n be Word32 instead?
480−-- | Runs a computation under a given thread limit. A limit of 0 means new
491+-- | Runs a computation under a given thread limit. A limit of 0 means all new
481−-- tasks start synchronously in the current thread. New threads are created by
492+-- tasks start synchronously in the current thread unless overridden by
482−-- 'parallel', and APIs that use parallel.
493+-- 'async'.
483threads :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
494threads :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
484threads n process = AsyncT $ do
495threads n process = AsyncT $ do
485 oldCr <- gets threadCredit
496 oldCr <- gets threadCredit
@@ -563,44 +574,40 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
563infixr 1 >>*, *>>, >>|
574infixr 1 >>*, *>>, >>|
564
575
565-- XXX This can be moved to utility functions as it is purely app level
576-- 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@
577+-- | Run @m b@ right after the first event in @m a@ is generated but before it
567−-- has finished. Useful to start a singleton task after the first task has been
578+-- is yielded. The result of @m b@ is discarded.
568−-- setup.
569−afterFirst :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
579+afterfirst :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
570−afterFirst ma mb = do
580+afterfirst ma mb = do
571 ref <- liftIO $ newIORef False
581 ref <- liftIO $ newIORef False
572 x <- ma
582 x <- ma
573 done <- liftIO $ readIORef ref
583 done <- liftIO $ readIORef ref
574 when (not done) $ (liftIO $ writeIORef ref True) >>* mb
584 when (not done) $ (liftIO $ writeIORef ref True) >>* mb
575 return x
585 return x
576
586
587+-- | Same as 'afterfirst'.
577(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
588(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
578−(>>|) = afterFirst
589+(>>|) = afterfirst
579
590
580−-- | Run 'm a' in "isolation" and discard its result, and then run 'm b' and
591+-- | Run @m a@ before running @m b@. The result of @m a@ is discarded.
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
592+before :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
584−discardAndThen ma mb = AsyncT $ do
593+before ma mb = AsyncT $ do
585 _ <- runAsyncT (ma >> mzero)
594 _ <- runAsyncT (ma >> mzero)
586 runAsyncT mb
595 runAsyncT mb
587
596
588−-- | Same as 'discardAndThen'.
597+-- | Same as 'before'.
589(*>>) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
598(*>>) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
590−(*>>) = discardAndThen
599+(*>>) = before
591
600
592−-- | Run 'm a' and then run 'm b' in "isolation" and return the result of 'm
601+-- | Run @m b@ after running @m a@. The result of @m b@ is discarded.
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
602+thereafter :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
596−thenDiscard ma mb = AsyncT $ do
603+thereafter ma mb = AsyncT $ do
597 a <- runAsyncT ma
604 a <- runAsyncT ma
598 _ <- runAsyncT (mb >> mzero)
605 _ <- runAsyncT (mb >> mzero)
599 return a
606 return a
600
607
601−-- | Same as 'thenDiscard'.
608+-- | Same as 'thereafter'.
602(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
609(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
603−(>>*) = thenDiscard
610+(>>*) = thereafter
604
611
605-------------------------------------------------------------------------------
612-------------------------------------------------------------------------------
606-- AsyncT transformer
613-- AsyncT transformer
src/Asyncly/Threads.hs+17−7
@@ -1,6 +1,15 @@
1{-# LANGUAGE FlexibleContexts #-}
1{-# LANGUAGE FlexibleContexts #-}
2{-# LANGUAGE ScopedTypeVariables #-}
2{-# LANGUAGE ScopedTypeVariables #-}
3
3
4+-- |
5+-- Module : Asyncly.Threads
6+-- Copyright : (c) 2017 Harendra Kumar
7+--
8+-- License : MIT-style
9+-- Maintainer : harendra.kumar@gmail.com
10+-- Stability : experimental
11+-- Portability : GHC
12+--
4module Asyncly.Threads
13module Asyncly.Threads
5 ( wait
14 ( wait
6 , wait_
15 , wait_
@@ -81,14 +90,15 @@ playRecording :: (MonadAsync m, MonadRecorder m)
81 => AsyncT m a -> Recording -> AsyncT m a
90 => AsyncT m a -> Recording -> AsyncT m a
82playRecording m recording = play recording >> m
91playRecording m recording = play recording >> m
83
92
84−-- | Compose a computation using previously captured list of logs. Composes a
93+-- | Resume an 'AsyncT' computation using previously recorded logs. The
85−-- group of computations, one for each log.
94+-- recording consists of a list of journals one for each thread in the
95+-- computation.
86playRecordings :: (MonadAsync m, MonadRecorder m)
96playRecordings :: (MonadAsync m, MonadRecorder m)
87 => AsyncT m a -> [Recording] -> AsyncT m a
97 => AsyncT m a -> [Recording] -> AsyncT m a
88playRecordings m logs = each logs >>= playRecording m
98playRecordings m logs = each logs >>= playRecording m
89
99
90−-- | Run an 'AsyncT m' computation with logging enabled and collect the results
100+-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish
91−-- or logs generated by each thread of the computation.
101+-- returning results for completed threads and recordings for paused threads.
92waitRecord :: (MonadAsync m, MonadCatch m)
102waitRecord :: (MonadAsync m, MonadCatch m)
93 => AsyncT m a -> m ([a], [Recording])
103 => AsyncT m a -> m ([a], [Recording])
94waitRecord m = do
104waitRecord m = do
@@ -99,9 +109,9 @@ waitRecord m = do
99 logs <- liftIO $ readIORef lref
109 logs <- liftIO $ readIORef lref
100 return (res, logs)
110 return (res, logs)
101
111
102−-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
112+-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish
103−-- and discard the results. If the computation suspends collect the logs to
113+-- and discard the results and return the recordings for paused threads, if
104−-- replay later.
114+-- any.
105waitRecord_ :: (MonadAsync m, MonadCatch m)
115waitRecord_ :: (MonadAsync m, MonadCatch m)
106 => AsyncT (RecorderT m) a -> m [Recording]
116 => AsyncT (RecorderT m) a -> m [Recording]
107waitRecord_ m = do
117waitRecord_ m = do