70c2917 · streamly · git.iko.soy

git.iko.soy/streamly

rename EventF to Context

commit70c2917343211cd3a67ab6ce64f71d097b8fc976 authorHarendra Kumar <harendra.kumar@gmail.com> dateSat 10 Jun 2017, 16:41 (9 years ago) parent0a69aec

3 files changed+32−33

compare with master

unifiedsplit

src/Duct/AsyncT.hs+13−13 src/Duct/Event.hs+12−13 src/Duct/Threads.hs+7−7

src/Duct/AsyncT.hs+13−13

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

4545

4646-- import Debug.Trace

4747

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

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

4949

5050------------------------------------------------------------------------------

5151-- Remote data types

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

148148 -- Save the 'm' and 'f', in case we migrate to a new thread before this

149149 -- bind operation completes, we run the operation manually in the new

150150 -- thread.

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

152− EventF { currentm = unsafeCoerce m

153− , fstack = unsafeCoerce f : fs

154− , .. }

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

152+ Context { currentm = unsafeCoerce m

153+ , fstack = unsafeCoerce f : fs

154+ , .. }

155155

156156 -- Inner bind operations in 'm' add their 'f' to fstack. If we migrate to

157157 -- 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

162162 Nothing -> empty

163163 Just x -> f x

164164

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

166− EventF { currentm = unsafeCoerce res

167− , fstack = tailsafe fs

168− , .. }

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

166+ Context { currentm = unsafeCoerce res

167+ , fstack = tailsafe fs

168+ , .. }

169169

170170 runAsyncT res

171171

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

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

243243

244244instance MonadTransControl AsyncT where

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

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

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

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

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

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

284284------------------------------------------------------------------------------

285285

286286-- | 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)

288288runTransient t = do

289289 childChan <- liftIO $ atomically newTChan

290290 pendingRef <- liftIO $ newIORef []

291291 credit <- liftIO $ newIORef maxBound

292292

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

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

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

295295

296296 waitForChildren childChan pendingRef

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

362362-- MonadState for AsyncT m

363363------------------------------------------------------------------------------

364364

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

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

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

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

368368 state f = AsyncT $ do

src/Duct/Event.hs+12−13

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

22{-# LANGUAGE FlexibleContexts #-}

44module Duct.Event

5− ( EventF(..)

6− , initEventF

5+ ( Context(..)

6+ , initContext

77 , getData

88 , setData

99 , modifyData

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

2525

2626type SData = ()

2727

28−-- XXX rename to Context

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

30−data EventF = forall m a b. EventF

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

29+data Context = forall m a b. Context

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

3231 -- away with this?

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

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

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

8988 } deriving Typeable

9089

91−initEventF

90+initContext

9291 :: m a

9392 -> TChan ThreadId

9493 -> IORef [ThreadId]

9594 -> IORef Int

96− -> EventF

97−initEventF x childChan pending credit =

98− EventF { event = mempty

95+ -> Context

96+initContext x childChan pending credit =

97+ Context { event = mempty

9998 , currentm = x

10099 , fstack = []

101100 , mfData = mempty

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

111110

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

113112-- '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)

115114getData = resp

116115 where resp = do

117116 list <- gets mfData

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

122121 typeResp = undefined

123122

124123-- | 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 ()

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

127126 where t = typeOf x

128127

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

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

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

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

132131 typeResp = undefined

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

136135 Just x -> Just $ unsafeCoerce x

137136 Nothing -> Nothing

138137

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

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

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

141140

142141{-

src/Duct/Threads.hs+7−7

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

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

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

4747

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

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

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

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

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

5151

5252 where

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

8080 liftIO $ modifyIORef pendingRef $ delete tid

8181

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

83−killChildren :: EventF -> IO ()

83+killChildren :: Context -> IO ()

8484killChildren ctx = do

8585 ths <- readIORef (pendingThreads ctx)

8686 mapM_ killThread ths

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

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

125125 return ()

126126

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

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

128128forkIt current proc = do

129129 child <- newChildCont current

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

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

132132

133133 where

134134

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

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

136136 updatePendingThreads cur tid = do

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

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

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

164164 tid <- myThreadId

165165 liftIO $ atomically $ writeTChan p tid

166166

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

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

168168forkMaybe current toRun = do

169169 gotCredit <- liftIO $ waitQSemB (threadCredit current)

170170 pending <- liftIO $ readIORef $ pendingThreads current

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

196196-- they can go away.

197197--

198198-- | 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 ()

200200genAsyncEvents parentc rec = forkMaybe parentc loop

201201

202202 where