git.iko.soy/streamly
on master help sign in
Some cleanups
bb5a74dHarendra Kumar
src/Asyncly/AsyncT.hs+28−73
compare with master
unifiedsplit
@@ -1,14 +1,7 @@
1−{-# LANGUAGE ConstraintKinds #-}
2−{-# LANGUAGE ExistentialQuantification #-}
3{-# LANGUAGE FlexibleContexts #-}
1{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
2{-# LANGUAGE FlexibleInstances #-}
5−{-# LANGUAGE InstanceSigs #-}
6{-# LANGUAGE MultiParamTypeClasses #-}
3{-# LANGUAGE MultiParamTypeClasses #-}
7−{-# LANGUAGE Rank2Types #-}
4+{-# LANGUAGE RankNTypes #-}
8−{-# LANGUAGE RecordWildCards #-}
9−{-# LANGUAGE ScopedTypeVariables #-}
10−{-# LANGUAGE TypeFamilies #-}
11−{-# LANGUAGE UndecidableInstances #-}
12
5
13-- |
6-- |
14-- Module : Asyncly.AsyncT
7-- Module : Asyncly.AsyncT
@@ -24,34 +17,27 @@ module Asyncly.AsyncT
24 ( AsyncT (..)
17 ( AsyncT (..)
25-- , async
18-- , async
26-- , makeAsync
19-- , makeAsync
27−
28− -- internal
29− , dbg
30 )
20 )
31where
21where
32
22
33import Control.Applicative (Alternative (..))
23import Control.Applicative (Alternative (..))
34−import Control.Monad (ap, mzero, when)
24+import Control.Monad (ap, liftM, MonadPlus(..), mzero)
35−import Control.Monad.Base (MonadBase (..), liftBaseDefault)
25+--import Control.Monad.Base (MonadBase (..), liftBaseDefault)
36import Control.Monad.Catch (MonadThrow, throwM)
26import Control.Monad.Catch (MonadThrow, throwM)
37−import Control.Monad.State (MonadIO (..), MonadPlus (..),
27+import Control.Monad.IO.Class (MonadIO(..))
38− StateT (..), liftM, runStateT,
39− get, gets, put, modify)
40import Control.Monad.Trans.Class (MonadTrans (lift))
28import Control.Monad.Trans.Class (MonadTrans (lift))
29+{-
41import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
30import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
42 MonadTransControl (..),
31 MonadTransControl (..),
43 defaultLiftBaseWith,
32 defaultLiftBaseWith,
44 defaultRestoreM, liftBaseWith)
33 defaultRestoreM, liftBaseWith)
34+ -}
45import Data.Maybe (maybe)
35import Data.Maybe (maybe)
46−import Data.IORef (newIORef, readIORef, writeIORef)
47−import Unsafe.Coerce (unsafeCoerce)
48
36
49import Control.Monad.Trans.Recorder (MonadRecorder(..))
37import Control.Monad.Trans.Recorder (MonadRecorder(..))
50−--import Debug.Trace (traceM)
38+--import Asyncly.Threads (MonadAsync)
51
39
52−import Asyncly.Threads (MonadAsync, Context)
53−
54−-- The 'Maybe' is redundant as we can use 'stop' value in place of Nothing,
40+-- The 'Maybe' is redundant as we can use 'stop' value in the Nothing case,
55-- but it makes the fold using '<|>' 25% faster.
41-- but it makes the fold using '<|>' 25% faster.
56newtype AsyncT m a =
42newtype AsyncT m a =
57 AsyncT {
43 AsyncT {
@@ -61,42 +47,30 @@ newtype AsyncT m a =
61 -> m r
47 -> m r
62 }
48 }
63
49
64−------------------------------------------------------------------------------
65−-- Utilities
66−------------------------------------------------------------------------------
67−
68−-- Toggle the dbg definition for debug traces
69−--dbg :: Monad m => String -> m ()
70−--dbg = traceM
71−dbg :: Monad m => a -> m ()
72−dbg _ = return ()
73−
74------------------------------------------------------------------------------
50------------------------------------------------------------------------------
75-- Monad
51-- Monad
76------------------------------------------------------------------------------
52------------------------------------------------------------------------------
77
53
78−instance MonadAsync m => Monad (AsyncT m) where
54+instance Monad m => Monad (AsyncT m) where
79 return a = AsyncT $ \_ yld -> yld a Nothing
55 return a = AsyncT $ \_ yld -> yld a Nothing
80
56
81 AsyncT m >>= f = AsyncT $ \stp yld ->
57 AsyncT m >>= f = AsyncT $ \stp yld ->
58+ let run x = (runAsyncT x) stp yld
82− m stp (\a r ->
59+ in m stp $ \a r ->
83− case r of
60+ maybe (run $ f a) (\rx -> run $ f a <|> (rx >>= f)) r
84− Nothing -> (runAsyncT (f a)) stp yld
85− Just rx -> (runAsyncT ((f a) <|> (rx >>= f))) stp yld
86− )
87
61
88------------------------------------------------------------------------------
62------------------------------------------------------------------------------
89-- Functor
63-- Functor
90------------------------------------------------------------------------------
64------------------------------------------------------------------------------
91
65
92−instance MonadAsync m => Functor (AsyncT m) where
66+instance Monad m => Functor (AsyncT m) where
93 fmap = liftM
67 fmap = liftM
94
68
95------------------------------------------------------------------------------
69------------------------------------------------------------------------------
96-- Applicative
70-- Applicative
97------------------------------------------------------------------------------
71------------------------------------------------------------------------------
98
72
99−instance MonadAsync m => Applicative (AsyncT m) where
73+instance Monad m => Applicative (AsyncT m) where
100 pure = return
74 pure = return
101 (<*>) = ap
75 (<*>) = ap
102
76
@@ -104,40 +78,21 @@ instance MonadAsync m => Applicative (AsyncT m) where
104-- Alternative
78-- Alternative
105------------------------------------------------------------------------------
79------------------------------------------------------------------------------
106
80
107−-- Thread pool vs transient threads
108−--
109−-- The current model is to start a new thread for every task. The computation
110−-- input is provided at the time of the creation and therefore no
111−-- synchronization is needed compared to a pool of threads contending to get
112−-- the input from a channel. However the thread creation overhead may be more
113−-- than the synchronization cost needed in the case of pool?
114−--
115−-- When the task is over, the outputs need to be collected and that requires
116−-- synchronization irrespective of a thread pool model or per task new thread
117−-- model.
118−--
119−-- XXX instead of starting a new thread every time, reuse the existing child
120−-- threads and send them work via a shared channel. When there is no more work
121−-- available we need a way to close the channel and wakeup all waiters so that
122−-- they can go away rather than waiting indefinitely.
123−
124−instance MonadAsync m => Alternative (AsyncT m) where
81+instance Monad m => Alternative (AsyncT m) where
125 empty = AsyncT $ \stp _ -> stp
82 empty = AsyncT $ \stp _ -> stp
126
83
127 -- XXX need to wait for the async threads in case of a stop is returned due
84 -- XXX need to wait for the async threads in case of a stop is returned due
128 -- to async thread spawning.
85 -- to async thread spawning.
129 AsyncT m1 <|> m2 = AsyncT $ \stp yld ->
86 AsyncT m1 <|> m2 = AsyncT $ \stp yld ->
130− m1 ((runAsyncT m2) stp yld) (\a r ->
87+ m1 ((runAsyncT m2) stp yld) $ \a r ->
131− case r of
132− Nothing -> yld a (Just m2)
88+ let yield x = yld a (Just x)
133− Just rx -> yld a (Just (rx <|> m2))
89+ in maybe (yield m2) (\rx -> yield $ rx <|> m2) r
134− )
135
90
136−instance MonadAsync m => MonadPlus (AsyncT m) where
91+instance Monad m => MonadPlus (AsyncT m) where
137 mzero = empty
92 mzero = empty
138 mplus = (<|>)
93 mplus = (<|>)
139
94
140−instance (Monoid a, MonadAsync m) => Monoid (AsyncT m a) where
95+instance (Monoid a, Monad m) => Monoid (AsyncT m a) where
141 mappend x y = mappend <$> x <*> y
96 mappend x y = mappend <$> x <*> y
142 mempty = return mempty
97 mempty = return mempty
143
98
@@ -160,14 +115,14 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
160instance MonadTrans AsyncT where
115instance MonadTrans AsyncT where
161 lift mx = AsyncT $ \_ yld -> mx >>= (\a -> (yld a Nothing))
116 lift mx = AsyncT $ \_ yld -> mx >>= (\a -> (yld a Nothing))
162
117
118+{-
163−instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where
119+instance (MonadBase b m, Monad m) => MonadBase b (AsyncT m) where
164 liftBase = liftBaseDefault
120 liftBase = liftBaseDefault
165
121
166-------------------------------------------------------------------------------
122-------------------------------------------------------------------------------
167-- monad-control
123-- monad-control
168-------------------------------------------------------------------------------
124-------------------------------------------------------------------------------
169
125
170−{-
171instance MonadTransControl AsyncT where
126instance MonadTransControl AsyncT where
172 type StT AsyncT a = (Maybe a, Context)
127 type StT AsyncT a = (Maybe a, Context)
173 liftWith f = AsyncT $ StateT $ \s ->
128 liftWith f = AsyncT $ StateT $ \s ->
@@ -177,7 +132,7 @@ instance MonadTransControl AsyncT where
177 {-# INLINABLE liftWith #-}
132 {-# INLINABLE liftWith #-}
178 {-# INLINABLE restoreT #-}
133 {-# INLINABLE restoreT #-}
179
134
180−instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) where
135+instance (MonadBaseControl b m, Monad m) => MonadBaseControl b (AsyncT m) where
181 type StM (AsyncT m) a = ComposeSt AsyncT m a
136 type StM (AsyncT m) a = ComposeSt AsyncT m a
182 liftBaseWith = defaultLiftBaseWith
137 liftBaseWith = defaultLiftBaseWith
183 restoreM = defaultRestoreM
138 restoreM = defaultRestoreM
@@ -189,17 +144,17 @@ instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) w
189-- Standard transformer instances
144-- Standard transformer instances
190------------------------------------------------------------------------------
145------------------------------------------------------------------------------
191
146
192−instance MonadAsync m => MonadIO (AsyncT m) where
147+instance MonadIO m => MonadIO (AsyncT m) where
193 liftIO = lift . liftIO
148 liftIO = lift . liftIO
194
149
195−instance MonadAsync m => MonadThrow (AsyncT m) where
150+instance MonadThrow m => MonadThrow (AsyncT m) where
196 throwM = lift . throwM
151 throwM = lift . throwM
197
152
198------------------------------------------------------------------------------
153------------------------------------------------------------------------------
199-- MonadRecorder
154-- MonadRecorder
200------------------------------------------------------------------------------
155------------------------------------------------------------------------------
201
156
202−instance (MonadAsync m, MonadRecorder m) => MonadRecorder (AsyncT m) where
157+instance (Monad m, MonadRecorder m) => MonadRecorder (AsyncT m) where
203 getJournal = lift getJournal
158 getJournal = lift getJournal
204 putJournal = lift . putJournal
159 putJournal = lift . putJournal
205 play = lift . play
160 play = lift . play
@@ -227,9 +182,9 @@ instance (MonadAsync m, MonadRecorder m) => MonadRecorder (AsyncT m) where
227-- 'async' is not useful as the last action as the last action always runs in
182-- 'async' is not useful as the last action as the last action always runs in
228-- the current thread.
183-- the current thread.
229{-
184{-
230−async :: MonadAsync m => AsyncT m a -> AsyncT m a
185+async :: Monad m => AsyncT m a -> AsyncT m a
231async action = AsyncT $ runAsyncTask True (runAsyncT action)
186async action = AsyncT $ runAsyncTask True (runAsyncT action)
232
187
233−makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a
188+makeAsync :: Monad m => ((a -> m ()) -> m ()) -> AsyncT m a
234makeAsync = AsyncT . makeCont
189makeAsync = AsyncT . makeCont
235-}
190-}