@@ -12,11 +12,9 @@
1212module Duct.AsyncT
1313 ( AsyncT (..)
1414 , waitAsync
15− , compose
1615 , (<**)
1716 , onNothing
1817 , RemoteStatus (..)
19− , back
2018 , waitForChildren
2119 , dbg
2220 )
@@ -44,13 +42,10 @@ import Data.Maybe (isJust, isNothing)
4442import Unsafe.Coerce (unsafeCoerce)
4543
4644import Duct.Event
47−import Duct.Log
4845
4946-- import Debug.Trace
5047
51−type StateM m = StateT EventF m
52−
53−newtype AsyncT m a = AsyncT { runAsyncT :: StateM m (Maybe a) }
48+newtype AsyncT m a = AsyncT { runAsyncT :: StateT EventF m (Maybe a) }
5449
5550------------------------------------------------------------------------------
5651-- Remote data types
@@ -89,6 +84,7 @@ instance Monad (AsyncT m) => Functor (AsyncT m) where
8984-- Applicative
9085------------------------------------------------------------------------------
9186
87+{-
9288-- | Warning: Radically untyped stuff. handle with care
9389getContinuations :: Monad m => StateM m [a -> AsyncT m b]
9490getContinuations = do
@@ -109,6 +105,8 @@ restoreStack :: MonadState EventF m => t -> m ()
109105restoreStack fs = modify $ \EventF {..} ->
110106 EventF { event = Nothing, fcomp = (unsafeCoerce fs), .. }
111107
108+-}
109+
112110instance Monad m => Applicative (AsyncT m) where
113111 pure a = AsyncT . return $ Just a
114112 m1 <*> m2 = do { x1 <- m1; x2 <- m2; return (x1 x2) }
@@ -145,24 +143,29 @@ tailsafe (_:xs) = xs
145143
146144instance Monad m => Monad (AsyncT m) where
147145 return = pure
148− m >>= f = AsyncT $ do
149− -- save m in xcomp and add f to the head of fcomp
150− -- _ <- setEventCont m f
151− modify $ \EventF { fcomp = fs, .. } ->
152− EventF { xcomp = unsafeCoerce m, fcomp = unsafeCoerce f : fs, .. }
153146
154− -- run m under this modified environment
147+ m >>= f = AsyncT $ do
148+ -- Save the 'm' and 'f', in case we migrate to a new thread before this
149+ -- bind operation completes, we run the operation manually in the new
150+ -- thread.
151+ modify $ \EventF { fstack = fs, .. } ->
152+ EventF { currentm = unsafeCoerce m
153+ , fstack = unsafeCoerce f : fs
154+ , .. }
155+
156+ -- Inner bind operations in 'm' add their 'f' to fstack. If we migrate to
157+ -- a new thread somewhere in the middle we unwind the fstack and run these
158+ -- functions manually after migration.
155159 mres <- runAsyncT m
156− -- remove the head of fcomp apply it to mres and that becomes the new xcomp.
157− -- is head of fcomp always the same as f?
160+
158161 let res = case mres of
159162 Nothing -> empty
160163 Just x -> f x
161164
162− -- XXX why would we have more than one continuation? Can we have just one
163− -- instead of a list and remove it after every bind?
164− modify $ \EventF { fcomp = fs, .. } ->
165− EventF { xcomp = unsafeCoerce res, fcomp = tailsafe fs, .. }
165+ modify $ \EventF { fstack = fs, .. } ->
166+ EventF { currentm = unsafeCoerce res
167+ , fstack = tailsafe fs
168+ , .. }
166169
167170 runAsyncT res
168171
@@ -178,15 +181,11 @@ instance (Monoid a, Monad (AsyncT m)) => Monoid (AsyncT m a) where
178181-- Backtracking
179182------------------------------------------------------------------------------
180183
184+{-
181185-- | Run the closure (the 'x' in 'x >>= f') of the current bind operation.
182186runClosure :: EventF -> StateM m (Maybe a)
183187runClosure EventF { xcomp = x } = runAsyncT (unsafeCoerce x)
184188
185−-- | Compose a list of continuations.
186−compose :: MonadIO m => [a -> AsyncT m a] -> (a -> AsyncT m b)
187−compose [] = const empty
188−compose (f:fs) = \x -> f x >>= compose fs
189−
190189-- | Run the continuation (the 'f' in 'x >>= f') of the current bind operation with the current state.
191190runContinuation :: MonadIO m => EventF -> a -> StateM m (Maybe b)
192191runContinuation EventF { fcomp = fs } =
@@ -226,18 +225,14 @@ back reason = AsyncT $ do
226225 Nothing -> runContinuation first x
227226 justreason -> goBackt $ Backtrack justreason bs
228227
228+-}
229+
229230------------------------------------------------------------------------------
230231-- MonadIO
231232------------------------------------------------------------------------------
232233
233234instance MonadIO m => MonadIO (AsyncT m) where
234− liftIO mx = do
235− ex <- liftIO' $ (mx >>= return . Right) `catch`
236− (\(e :: SomeException) -> return $ Left e)
237− case ex of
238− Left e -> back e
239− Right x -> return x
240− where liftIO' x = AsyncT $ liftIO x >>= return . Just
235+ liftIO mx = AsyncT $ liftIO mx >>= return . Just
241236
242237-------------------------------------------------------------------------------
243238-- AsyncT transformer
@@ -403,6 +398,7 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
403398 abs f = f >>= return . abs
404399 signum f = f >>= return . signum
405400
401+{-
406402class AdditionalOperators m where
407403
408404 -- | Run @m a@ discarding its result before running @m b@.
@@ -469,3 +465,12 @@ infixr 1 <***, <**, **>
469465 runAsyncT mb
470466 return $ Just x
471467
468+-}
469+
470+(<**) :: Monad m => AsyncT m a -> AsyncT m b -> AsyncT m a
471+(<**) ma mb = AsyncT $ do
472+ a <- runAsyncT ma
473+ runAsyncT mb
474+ return a
475+
476+infixr 1 <**