70c2917 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

rename EventF to Context

70c2917Harendra Kumar9 years ago

src/Duct/AsyncT.hs+13−13

compare with master

unifiedsplit

@@ -45,7 +45,7 @@ import Duct.Event

46-- import Debug.Trace

48−newtype AsyncT m a = AsyncT { runAsyncT :: StateT EventF m (Maybe a) }

48+newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }

50------------------------------------------------------------------------------

51-- Remote data types

@@ -148,10 +148,10 @@ instance Monad m => Monad (AsyncT m) where

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, .. } ->

151+ modify $ \Context { fstack = fs, .. } ->

152− EventF { currentm = unsafeCoerce m

152+ Context { currentm = unsafeCoerce m

153− , fstack = unsafeCoerce f : fs

153+ , fstack = unsafeCoerce f : fs

154− , .. }

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

@@ -162,10 +162,10 @@ instance Monad m => Monad (AsyncT m) where

162 Nothing -> empty

163 Just x -> f x

164

165− modify $ \EventF { fstack = fs, .. } ->

165+ modify $ \Context { fstack = fs, .. } ->

166− EventF { currentm = unsafeCoerce res

166+ Context { currentm = unsafeCoerce res

167− , fstack = tailsafe fs

167+ , fstack = tailsafe fs

168− , .. }

168+ , .. }

169

170 runAsyncT res

171

@@ -242,7 +242,7 @@ instance MonadTrans AsyncT where

242 lift mx = AsyncT $ lift mx >>= return . Just

243

244instance MonadTransControl AsyncT where

245− type StT AsyncT a = (Maybe a, EventF)

245+ type StT AsyncT a = (Maybe a, Context)

246 liftWith f = AsyncT $ StateT $ \s ->

247 liftM (\x -> (Just x, s))

248 (f $ \t -> runStateT (runAsyncT t) s)

@@ -284,13 +284,13 @@ waitForChildren chan pendingRef = do

284------------------------------------------------------------------------------

285

286-- | Run a transient computation with a default initial state

287−runTransient :: forall m a. MonadIO m => AsyncT m a -> m (Maybe a, EventF)

287+runTransient :: forall m a. MonadIO m => AsyncT m a -> m (Maybe a, Context)

288runTransient t = do

289 childChan <- liftIO $ atomically newTChan

290 pendingRef <- liftIO $ newIORef []

291 credit <- liftIO $ newIORef maxBound

292

293− r <- runStateT (runAsyncT t) $ initEventF

293+ r <- runStateT (runAsyncT t) $ initContext

294 (empty :: AsyncT m a) childChan pendingRef credit

295

296 waitForChildren childChan pendingRef

@@ -362,7 +362,7 @@ delSData x = AsyncT $ delData x >> return (Just ())

362-- MonadState for AsyncT m

363------------------------------------------------------------------------------

364

365−instance (Monad m, Monad (AsyncT m)) => MonadState EventF (AsyncT m) where

365+instance (Monad m, Monad (AsyncT m)) => MonadState Context (AsyncT m) where

366 get = AsyncT $ get >>= return . Just

367 put x = AsyncT $ put x >> return (Just ())

368 state f = AsyncT $ do

src/Duct/Event.hs+12−13

@@ -2,8 +2,8 @@

2{-# LANGUAGE FlexibleContexts #-}

4module Duct.Event

5− ( EventF(..)

5+ ( Context(..)

6− , initEventF

6+ , initContext

7 , getData

8 , setData

9 , modifyData

@@ -25,9 +25,8 @@ import Unsafe.Coerce (unsafeCoerce)

26type SData = ()

28−-- XXX rename to Context

29−-- | EventF describes the context of a TransientIO computation.

28+-- | Describes the context of a computation.

30−data EventF = forall m a b. EventF

29+data Context = forall m a b. Context

31 { event :: Maybe SData -- untyped, XXX rename to mailbox - can we do

30 { event :: Maybe SData -- untyped, XXX rename to mailbox - can we do

32 -- away with this?

31 -- away with this?

33 -- ^ event data to use in a continuation in a new thread

32 -- ^ event data to use in a continuation in a new thread

@@ -88,14 +87,14 @@ data EventF = forall m a b. EventF

88 -- ^ How many more threads are allowed to be created?

87 -- ^ How many more threads are allowed to be created?

89 } deriving Typeable

88 } deriving Typeable

91−initEventF

90+initContext

92 :: m a

91 :: m a

93 -> TChan ThreadId

92 -> TChan ThreadId

94 -> IORef [ThreadId]

93 -> IORef [ThreadId]

95 -> IORef Int

94 -> IORef Int

96− -> EventF

95+ -> Context

97−initEventF x childChan pending credit =

96+initContext x childChan pending credit =

98− EventF { event = mempty

97+ Context { event = mempty

99 , currentm = x

98 , currentm = x

100 , fstack = []

99 , fstack = []

101 , mfData = mempty

100 , mfData = mempty

@@ -111,7 +110,7 @@ initEventF x childChan pending credit =

111

110

112-- | Same as 'getSData' but with a more general type. If the data is found, a

111-- | Same as 'getSData' but with a more general type. If the data is found, a

113-- 'Just' value is returned. Otherwise, a 'Nothing' value is returned.

112-- 'Just' value is returned. Otherwise, a 'Nothing' value is returned.

114−getData :: (MonadState EventF m, Typeable a) => m (Maybe a)

113+getData :: (MonadState Context m, Typeable a) => m (Maybe a)

115getData = resp

114getData = resp

116 where resp = do

115 where resp = do

117 list <- gets mfData

116 list <- gets mfData

@@ -122,11 +121,11 @@ getData = resp

122 typeResp = undefined

121 typeResp = undefined

123

122

124-- | Same as 'setSData' but with a more general type.

123-- | Same as 'setSData' but with a more general type.

125−setData :: (MonadState EventF m, Typeable a) => a -> m ()

124+setData :: (MonadState Context m, Typeable a) => a -> m ()

126setData x = modify $ \st -> st { mfData = M.insert t (unsafeCoerce x) (mfData st) }

125setData x = modify $ \st -> st { mfData = M.insert t (unsafeCoerce x) (mfData st) }

127 where t = typeOf x

126 where t = typeOf x

128

127

129−modifyData :: (MonadState EventF m, Typeable a) => (Maybe a -> Maybe a) -> m ()

128+modifyData :: (MonadState Context m, Typeable a) => (Maybe a -> Maybe a) -> m ()

130modifyData f = modify $ \st -> st { mfData = M.alter alterf t (mfData st) }

129modifyData f = modify $ \st -> st { mfData = M.alter alterf t (mfData st) }

131 where typeResp :: (Maybe a -> b) -> a

130 where typeResp :: (Maybe a -> b) -> a

132 typeResp = undefined

131 typeResp = undefined

@@ -136,7 +135,7 @@ modifyData f = modify $ \st -> st { mfData = M.alter alterf t (mfData st) }

136 Just x -> Just $ unsafeCoerce x

135 Just x -> Just $ unsafeCoerce x

137 Nothing -> Nothing

136 Nothing -> Nothing

138

137

139−delData :: (MonadState EventF m, Typeable a) => a -> m ()

138+delData :: (MonadState Context m, Typeable a) => a -> m ()

140delData x = modify $ \st -> st { mfData = M.delete (typeOf x) (mfData st) }

139delData x = modify $ \st -> st { mfData = M.delete (typeOf x) (mfData st) }

141

140

142{-

141{-

src/Duct/Threads.hs+7−7

@@ -45,8 +45,8 @@ import Duct.Event

45-- to a new thread. Run the stack of pending functions in fstack. The types

46-- don't match. We just coerce the types here, we know that they actually match.

48−continue :: (MonadIO m, Monad (AsyncT m)) => EventF -> StateT EventF m (Maybe a)

48+continue :: (MonadIO m, Monad (AsyncT m)) => Context -> StateT Context m (Maybe a)

49−continue EventF { currentm = x, fstack = fs } =

49+continue Context { currentm = x, fstack = fs } =

50 runAsyncT $ unsafeCoerce x >>= runfStack (unsafeCoerce fs)

52 where

@@ -80,7 +80,7 @@ waitForOneChild chan pendingRef = do

80 liftIO $ modifyIORef pendingRef $ delete tid

82-- | kill all the child threads associated with the continuation context

83−killChildren :: EventF -> IO ()

83+killChildren :: Context -> IO ()

84killChildren ctx = do

85 ths <- readIORef (pendingThreads ctx)

86 mapM_ killThread ths

@@ -124,7 +124,7 @@ forkFinally1 action and_then =

124 _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then

125 return ()

126

127−forkIt :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m t) -> m ()

127+forkIt :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m t) -> m ()

128forkIt current proc = do

129 child <- newChildCont current

130 tid <- forkFinally1 (proc child) (beforeExit child)

@@ -132,7 +132,7 @@ forkIt current proc = do

132

133 where

134

135− updatePendingThreads :: MonadIO m => EventF -> ThreadId -> m ()

135+ updatePendingThreads :: MonadIO m => Context -> ThreadId -> m ()

136 updatePendingThreads cur tid = do

137 -- update the new thread before reclaiming zombies so that if it exited

138 -- already reclaim finds it in the list and does not panic.

@@ -164,7 +164,7 @@ forkIt current proc = do

164 tid <- myThreadId

165 liftIO $ atomically $ writeTChan p tid

166

167−forkMaybe :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m ()) -> m ()

167+forkMaybe :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m ()) -> m ()

168forkMaybe current toRun = do

169 gotCredit <- liftIO $ waitQSemB (threadCredit current)

170 pending <- liftIO $ readIORef $ pendingThreads current

@@ -196,7 +196,7 @@ data StreamData a =

196-- they can go away.

197--

198-- | Execute the IO action and the continuation

199−genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => EventF -> IO (StreamData t) -> m ()

199+genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => Context -> IO (StreamData t) -> m ()

200genAsyncEvents parentc rec = forkMaybe parentc loop

201

202 where