@@ -1,3 +1,4 @@
1+{-# OPTIONS_GHC -fno-warn-unused-imports #-}
2-- Module : Asyncly.Tutorial
3-- Module : Asyncly.Tutorial
3-- Copyright : (c) 2017 Harendra Kumar
4-- Copyright : (c) 2017 Harendra Kumar
@@ -5,141 +6,321 @@
5-- License : BSD3
6-- License : BSD3
6-- Maintainer : harendra.kumar@gmail.com
7-- Maintainer : harendra.kumar@gmail.com
8−-- 'AsyncT' is like a list transformer ('ListT' or 'LogicT') that supports
9−-- concurrent composition. Lists or streams can be generated using serial or
10−-- concurrent monadic actions. A stream could be a stream of data or a stream
11−-- of asynchronous events in an event driven program. Streams can be appended
12−-- serially (<>) or interleaved fairly (\<=>), concurrent streams can be
13−-- unioned like sets (\<|) or interleaved fairly (\<|>). Streams can be
14−-- multiplied as in a list monad using the bind (>>=) operation.
16−-- Here is an example program that repeatedly reads strings from standard input
10+-- Asyncly combines streaming, logic programming, concurrency and reactive
17−-- and echos them to the standard output:
11+-- programming under a single abstraction with a concise API, giving
12+-- programmers a powerful tool for wide variety of applications.
14+-- The 'AsyncT' monad operates on streams. Multiple streams of the same type
15+-- can be summed using different sum style operators. For example, '<>' appends
16+-- streams serially, '<=>' interleaves streams, '<|' combines streams in
17+-- parallel and '<|>' combines streams in a fairly parallel manner. Fold
18+-- utilities are provided to choose a fold style and an operator to fold a
19+-- container in a desired manner producing an AsyncT stream.
21+-- Streams of the same or different types can be zipped or transformed using
22+-- applicative or monadic composition. The monadic composition is exactly the
23+-- same as a list transformer except that it can be concurrent. It can be
24+-- likened to nested loops in the imperative programming parlance. In addition
25+-- to the regular monadic bind operator '>>=', other operators are provided to
26+-- compose streams in alternate manners. For example, '>->' interleaves the
27+-- iterations of outer and inner loops, '>>|' runs outer and inner loops in
28+-- parallel and '>|>' runs the outer and inner loops in fairly parallel manner.
30+-- With asyncly you can write concurrent and streaming applications without
31+-- being aware of threads or synchronization. No explicit thread control is
32+-- needed, where applicable the concurrency rate is automatically controlled
33+-- based on the demand by the consumer. However, combinators are provided to
34+-- fine tune the concurrency control. Streaming and concurrency allows
35+-- expressing reactive applications conveniently. See 'Asyncly.Examples' for a
36+-- simple SDL based FRP example.
38+-- For examples and other ways to use the library see 'Asyncly.Examples' as
41+module Asyncly.Tutorial
43+ -- * AsyncT Monad Transformer
46+ -- * Generating Streams (Monoidal Sum Composition)
49+ -- ** Serial composition (Append)
52+ -- ** Interleaved composition
55+ -- ** Parallel composition
58+ -- ** Fairly Parallel composition
61+ -- ** Custom composition
64+ -- * Looping Over Streams (Monadic Product Composition)
66+ -- ** List Transformer
69+ -- ** Interleaved Nesting
70+ -- $interleavedNesting
72+ -- ** Concurrent Nesting
73+ -- $concurrentNesting
75+ -- ** Fairly Concurrent Nesting
76+ -- $fairlyConcurrentNesting
78+ -- * Applicative Composition
81+ -- * Reactive Programming
84+ -- * State Machine Model
87+ -- * Interworking with Streaming Libraries
96+import Asyncly.Prelude
97+-- import Asyncly.Examples
99+import Control.Applicative
100+import Control.Monad.IO.Class (MonadIO(..))
101+import Control.Monad.Trans.Class (MonadTrans (lift))
105+-- 'AsyncT' is a monad transformer, the type 'AsyncT m a' represents a stream
106+-- of values of type 'a' in the underlying monad 'm'. As usual you can use the
107+-- 'lift' combinator to lift values from the immediately lower monad in the
108+-- transformer stack or 'liftIO' to lift values from IO. 'runAsyncly' runs the
109+-- monad discarding any results, 'toList' can be used to collect the results in
110+-- a list. You can also use 'uncons' to extract results from 'AsyncT'. For
20-- import Asyncly
114-- import Asyncly
21−-- import Data.Semigroup (cycle1)
23−-- main = runAsyncly $ cycle1 (liftIO getLine) >>= liftIO . putStrLn
117+-- runAsyncly $ liftIO $ putStrLn "Hello world!"
118+-- xs \<- toList $ return 1 <> return 2
26−-- You can fold with (\<|) or (\<|>) to run actions concurrently. For
124+-- AsyncT streams are generated using either 'return' on pure values or by
27−-- example to find the fastest search engine:
125+-- using 'lift' or a variant of it or by folding existing 'AsyncT' streams with
126+-- a suitable sum operator. Some fold utilities are provided by the library for
129+-- * 'foldWith' folds a Foldable container of AsyncT computations using the
131+-- * 'foldMapWith' folds like foldWith but also maps a function before folding.
132+-- * 'forEachWith' is like foldMapwith but the container argument comes before
133+-- the function argument.
30-- import Asyncly
136-- import Asyncly
31−-- import Network.HTTP.Simple
33−-- main = runAsyncly $ google \<|> bing \<|> duckduckgo
138+-- main = do
34−-- where
139+-- runAsyncly $ liftIO $ putStrLn "Hello world!"
35−-- google = get "https://www.google.com/search?q=haskell"
140+-- xs \<- toList $ return 1 <> return 2; print xs
36−-- bing = get "https://www.bing.com/search?q=haskell"
141+-- xs \<- toList $ foldWith (<>) (map return [1..10]); print xs
37−-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
142+-- xs \<- toList $ foldMapWith (<>) return [1..10]; print xs
38−-- get s = liftIO (httpNoBody (parseRequest_ s) >> putStrLn (show s))
143+-- xs \<- toList $ forEachWith (<>) [1..10] return; print xs
145+-- The following sections illustrate different operators that we can use to
146+-- fold AsyncT streams. We will use a trivial example of composing a few timer
147+-- using the timer function defined below. In general these could be
148+-- arbitrarily complex computations and any number of them.
41−-- The above examples just used applicative or alternative compositions. They
150+-- @
42−-- can be done equivalently using async package as well. However the real power
151+-- import Asyncly
43−-- of asyncly is in the monadic composition. Monadic composition allows
152+-- import Control.Concurrent
44−-- creating state machines or event networks or multiplying streams whatever
45−-- way you want to think about it. It is a very powerful general, concurrent
46−-- and reactive programming tool.
48−-- Let us now see an applicative example of the same:
154+-- traced m = liftIO (myThreadId >>= print) >> m
50−-- Let's take an example of listing a directory tree:
156+-- timer n = liftIO $ do
157+-- threadDelay (n * 1000000)
158+-- tid \<- myThreadId
159+-- putStrLn (show tid ++ ": Timer " ++ show n)
163+-- The '<>' operator composes multiple AsyncT streams in series i.e. the next
164+-- event happens after the previous one is completed. The following example
165+-- prints the sequence 3, 2, 1 and takes a total of 6 seconds:
53−-- import Path.IO (listDir, getCurrentDir)
168+-- main = runAsyncly $ timer 3 <> timer 2 <> timer 1
54−-- import Asyncly
169+-- @
171+-- ThreadId 36: Timer 3
172+-- ThreadId 36: Timer 2
173+-- ThreadId 36: Timer 1
177+-- The '<=>' operator interleaves the two computations i.e. it yields one item
178+-- from the first stream and then one item from the second stream. The
179+-- following example should print the sequence 1, 3, 2, 4:
56−-- main = runAsyncly $ getCurrentDir >>= readdir
57−-- where readdir d = do
58−-- (dirs, files) <- lift $ listDir d
59−-- liftIO $ mapM_ putStrLn $ map show files
60−-- foldr (<>) mempty $ map readdir dirs
182+-- main = runAsyncly $ (timer 1 <> timer 2) \<=> (timer 3 <> timer 4)
185+-- ThreadId 36: Timer 1
186+-- ThreadId 36: Timer 3
187+-- ThreadId 36: Timer 2
188+-- ThreadId 36: Timer 4
193+-- The '<|' operator runs computations in parallel, preferring the first
194+-- computation over the second. The second computation is run in parallel with
195+-- the first only if the first computation is not producing enough output to
196+-- keep the consumer busy otherwise the second computation is run serially
197+-- after the first. We maintain a pool of minimum number of threads, for
198+-- running the computations, that is big enough to match the pull rate of the
199+-- consumer of the stream.
200+-- Note that the left bias of the operator '<|' is suggested by its shape.
201+-- It is also half cut '<|>' which is fairly parallel in contrast.
203+-- The following example runs all the parallel computations in a single thread
204+-- one after another, because none of them blocks. Note that if the consumer
205+-- were faster than the producer this would start new threads for each
206+-- computation to keep up even if none of them blocks:
63−-- Just think that you had applicative and no monad in Haskell. We have
208+-- @
64−-- streaming libraries that are equivalent to applicative and asyncly is the
209+-- main = runAsyncly $ traced (sqrt 9) <| traced (sqrt 16) <| traced (sqrt 25)
65−-- equivalent of the monadic composition of streams. We have had that in th
210+-- @
66−-- form of half hearted list transformers but asyncly provides a full fledged
211+-- @
67−-- concurent list transformer.
212+-- ThreadId 40
69−-- Let us see a reactive programming example:
217+-- In the following example since the first computation blocks we start the
218+-- next one in a separate thread and so on:
72−-- import Asyncly
221+-- main = runAsyncly $ timer 3 <| timer 2 <| timer 1
73−-- import Control.Concurrent (threadDelay)
222+-- @
74−-- import Control.Monad (when)
223+-- @
75−-- import Control.Monad.State
224+-- ThreadId 42: Timer 1
76−-- import Data.Semigroup (cycle1)
225+-- ThreadId 41: Timer 2
226+-- ThreadId 40: Timer 3
78−-- data Event = Harm Int | Heal Int | Quit deriving (Show)
229+-- When we have a hierarchy of computations this operator follows DFS style:
80−-- userAction :: MonadIO m => AsyncT m Event
231+-- @
81−-- userAction = cycle1 $ liftIO askUser
232+-- main = runAsyncly $ (p 1 <| p 2) <| (p 3 <| p 4)
82−-- where
233+-- where p = liftIO . print
83−-- askUser = do
234+-- @
84−-- command <- getLine
235+-- @
85−-- case command of
236+-- 1
86−-- "potion" -> return (Heal 10)
237+-- 2
87−-- "quit" -> return Quit
238+-- 3
88−-- _ -> putStrLn "What?" >> askUser
239+-- 4
90−-- acidRain :: MonadIO m => AsyncT m Event
242+-- Note that since this operator is left biased it should not be used when the
91−-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
243+-- computations have timers that are relative to each other because all
244+-- computations may not be started at the same time and therefore timers in all
245+-- of them will not start at the same time. When relative timing among all
246+-- computations is important or when we need to start all computations at once
247+-- for some reason use '<|>' instead. However, '<|' is useful in situations
248+-- when we want to optimally utilize the resources and we know that the
249+-- computations can run in parallel but we do not care if they actually run in
250+-- parallel or not, that decision is left to the scheduler. Also, note that
251+-- this operator can be used to fold infinite containers in contrast to '<|>',
252+-- because it does not require us to run all of them at the same time.
93−-- game :: (MonadAsync m, MonadState Int m) => AsyncT m ()
256+-- The 'Alternative' composition operator '<|>', like '<|', runs the composed
94−-- game = do
257+-- computations in parallel. However, unlike '<|' it runs all of the
95−-- event <- userAction <|> acidRain
258+-- computations in fairly parallel manner using a round robin scheduling. Note
96−-- case event of
259+-- that this should not be used on infinite containers, as it will lead to an
97−-- Harm n -> modify $ \h -> h - n
260+-- infinite scheduling queue.
98−-- Heal n -> modify $ \h -> h + n
99−-- Quit -> fail "quit"
101−-- h <- get
262+-- The following example sends a query to three search engines and prints the
102−-- when (h <= 0) $ fail "You die!"
263+-- name of the search engine as a response arrives:
103−-- liftIO $ putStrLn $ "Health = " ++ show h
106−-- putStrLn "Your health is deteriorating due to acid rain,\
107−-- \ type \"potion\" or \"quit\""
108−-- runStateT (runAsyncly game) 60
267+-- import Network.HTTP.Simple
111−-- You can create trees of demand driven parallel computations. Parallel
269+-- main = runAsyncly $ google \<|> bing \<|> duckduckgo
112−-- computations are started faster if the consumer is pulling faster. The
270+-- where
113−-- number of threads are automatically adjusted and are kept at a minimum
271+-- google = get "https://www.google.com/search?q=haskell"
114−-- required depending on the consumer rate.
272+-- bing = get "https://www.bing.com/search?q=haskell"
273+-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
274+-- get s = liftIO (httpNoBody (parseRequest_ s) >> putStrLn (show s))
280+-- The 'async' API can be used to create references to asynchronously running
281+-- 'AsyncT' computations. We can then use 'uncons' to explore the
282+-- streams arbitrarily and then recompose them to create a new 'AsyncT' stream.
283+-- This way we can dynamically decide which stream to explore at any given time.
284+-- Take an example of a merge sort of two sorted streams. We need to keep
285+-- consuming items from the stream which has the lowest item in the sort order.
286+-- This can be achieved using async references to streams. See
287+-- 'Asyncly.Examples.MergeSortedStreams'.
116−-- Asyncly allows expressing and composing state machines and event driven
289+-- $listt
117−-- programs in a straightforward manner. It allows waiting for events to occur
118−-- in a non-blocking manner and process them asynchronously; multiple events
119−-- can be processed in parallel. Tasks can be easily split into smaller tasks,
120−-- processed in parallel and results combined.
122−-- The 'Alternative' composition @(\<|\>)@ is used to express asynchronous or
291+-- The previous section discussed ways to purely compose streams using sum
123−-- parallel tasks. The following example demonstrates generation and printing
292+-- style operators. In this section we will explore how to transform streams
124−-- of random numbers happening in parallel:
293+-- using monadic composition.
294+-- The above examples just used applicative or alternative compositions. They
295+-- can be done equivalently using async package as well. However the real power
296+-- of asyncly is in the monadic composition. Monadic composition allows
297+-- creating state machines or event networks or multiplying streams whatever
298+-- way you want to think about it. It is a very powerful general, concurrent
299+-- and reactive programming tool.
302+-- Here is an example program that repeatedly reads strings from standard input
303+-- and echoes them to the standard output:
127−-- import Control.Applicative ((\<|\>))
128−-- import Control.Concurrent (threadDelay)
129−-- import Control.Monad.IO.Class (liftIO)
130−-- import System.Random (randomIO)
131-- import Asyncly
306-- import Asyncly
307+-- import Data.Semigroup (cycle1)
133−-- main = wait_ $ do
309+-- main = runAsyncly $ cycle1 (liftIO getLine) >>= liftIO . putStrLn
135−-- liftIO $ print x
311+-- Here we perform a simple multi-threaded map-reduce by squaring each number
312+-- in a separate thread and then summing the squares:
315+-- import Control.Applicative ((\<|\>), empty)
316+-- import Data.List (sum)
139−-- loop = do
319+-- main = do
140−-- liftIO $ threadDelay 1000000
320+-- squares <- wait $ do
141−-- x <- liftIO (randomIO :: IO Int)
321+-- x <- foldl (\<|\>) empty $ map return [1..100]
142−-- return x \<|\> loop
322+-- return (x * x)
323+-- print . sum $ squares
145-- Here two random number generation loops run in parallel so two numbers are
326-- Here two random number generation loops run in parallel so two numbers are
@@ -167,6 +348,24 @@
167-- rnd <- liftIO (randomIO :: IO Int)
348-- rnd <- liftIO (randomIO :: IO Int)
168-- return (name, rnd) \<|\> loop name
349-- return (name, rnd) \<|\> loop name
352+-- $interleavedNesting
353+-- interleavedNesting
355+-- $concurrentNesting
357+-- Let's take an example of listing a directory tree:
360+-- import Path.IO (listDir, getCurrentDir)
363+-- main = runAsyncly $ getCurrentDir >>= readdir
364+-- where readdir d = do
365+-- (dirs, files) <- lift $ listDir d
366+-- liftIO $ mapM_ putStrLn $ map show files
367+-- foldr (<>) mempty $ map readdir dirs
170-- Here the two loops are serially composed. For each value yielded by loop A,
369-- Here the two loops are serially composed. For each value yielded by loop A,
171-- loop B is executed. Four results are printed, all four run in separate
370-- loop B is executed. Four results are printed, all four run in separate
172-- parallel threads. The composition is like ListT except that this is
371-- parallel threads. The composition is like ListT except that this is
@@ -196,34 +395,56 @@
196-- repeat = if n > 1 then loop name (n - 1) else empty
395-- repeat = if n > 1 then loop name (n - 1) else empty
197-- in (return result) \<|\> repeat
396-- in (return result) \<|\> repeat
399+-- $fairlyConcurrentNesting
400+-- fairlyConcurrentNesting
403+-- Applicative composition
200−-- Here we perform a simple multi-threaded map-reduce by squaring each number
407+-- Let us see a reactive programming example:
201−-- in a separate thread and then summing the squares:
204−-- import Control.Applicative ((\<|\>), empty)
205−-- import Data.List (sum)
206-- import Asyncly
410-- import Asyncly
411+-- import Control.Concurrent (threadDelay)
412+-- import Control.Monad (when)
413+-- import Control.Monad.State
414+-- import Data.Semigroup (cycle1)
416+-- data Event = Harm Int | Heal Int | Quit deriving (Show)
418+-- userAction :: MonadIO m => AsyncT m Event
419+-- userAction = cycle1 $ liftIO askUser
422+-- command <- getLine
424+-- "potion" -> return (Heal 10)
425+-- "quit" -> return Quit
426+-- _ -> putStrLn "What?" >> askUser
428+-- acidRain :: MonadIO m => AsyncT m Event
429+-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
431+-- game :: (MonadAsync m, MonadState Int m) => AsyncT m ()
433+-- event <- userAction <|> acidRain
435+-- Harm n -> modify $ \h -> h - n
436+-- Heal n -> modify $ \h -> h + n
437+-- Quit -> fail "quit"
440+-- when (h <= 0) $ fail "You die!"
441+-- liftIO $ putStrLn $ "Health = " ++ show h
208-- main = do
443-- main = do
209−-- squares <- wait $ do
444+-- putStrLn "Your health is deteriorating due to acid rain,\
210−-- x <- foldl (\<|\>) empty $ map return [1..100]
445+-- \ type \"potion\" or \"quit\""
211−-- return (x * x)
446+-- runStateT (runAsyncly game) 60
212−-- print . sum $ squares
215−-- 'Applicative' and 'Monoid' compositions work as expected.
217−-- It can be thought of as a non-deterministic continuation monad or a
449+-- $statemachine
218−-- combination of ContT and ListT. It allows to capture the state of the
450+-- State machine stuff
219−-- application at any point and trigger arbitrary number of continuations from
220−-- the capture point. Non-determinism or parallel continuations are introduced
221−-- using the <|> operator. It provides a convenient way to implement and
222−-- compose state machines without using callbacks. An event in the state
223−-- machine corresponds to a continuation. There are no cycles in the state
224−-- machine as each transition in the state machine is an independent instance
225−-- of the state machine. It is an immutable state machine!
227−module Asyncly.Tutorial