@@ -18,7 +18,7 @@ import Control.Concurrent
18import Control.Concurrent.STM
18import Control.Concurrent.STM
19 (TChan, atomically, newTChan, readTChan, tryReadTChan, writeTChan)
19 (TChan, atomically, newTChan, readTChan, tryReadTChan, writeTChan)
20import Control.Exception (SomeException(..), ErrorCall(..), catch)
20import Control.Exception (SomeException(..), ErrorCall(..), catch)
21−import qualified Control.Exception.Lifted as E
21+import qualified Control.Exception.Lifted as EL
22import Data.Dynamic (Typeable)
22import Data.Dynamic (Typeable)
23import Data.IORef
23import Data.IORef
24 (IORef, atomicModifyIORef, readIORef, writeIORef, modifyIORef,
24 (IORef, atomicModifyIORef, readIORef, writeIORef, modifyIORef,
@@ -102,20 +102,19 @@ instance Read SomeException where
102forkFinally1 :: (MonadIO m, MonadBaseControl IO m) =>
102forkFinally1 :: (MonadIO m, MonadBaseControl IO m) =>
103 m a -> (Either SomeException a -> IO ()) -> m ThreadId
103 m a -> (Either SomeException a -> IO ()) -> m ThreadId
104forkFinally1 action and_then =
104forkFinally1 action and_then =
105− E.mask $ \restore ->
105+ EL.mask $ \restore ->
106 liftBaseWith $ \runInIO -> forkIO $ do
106 liftBaseWith $ \runInIO -> forkIO $ do
107− runInIO $ E.try (restore action) >>= liftIO . and_then
107+ _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then
108 return ()
108 return ()
110forkIt :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m t) -> m ()
110forkIt :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m t) -> m ()
111forkIt current proc = do
111forkIt current proc = do
112 child <- newChildCont current
112 child <- newChildCont current
113 tid <- forkFinally1 (runInChild proc child)
113 tid <- forkFinally1 (runInChild proc child)
114− $ \res -> do
114+ $ \res -> do
115− thId <- myThreadId
115+ thId <- myThreadId
116− beforeExit child{threadId = thId} res
116+ beforeExit child{threadId = thId} res
117 updatePendingThreads current child{threadId = tid}
117 updatePendingThreads current child{threadId = tid}
@@ -151,6 +150,8 @@ forkIt current proc = do
152 waitForChildren (zombieChannel cur) (pendingThreads cur)
151 waitForChildren (zombieChannel cur) (pendingThreads cur)
153 signalQSemB (threadCredit cur)
152 signalQSemB (threadCredit cur)
153+ -- We are guaranteed to have a parent because we have been explicitly
154+ -- forked by some parent.
154 let p = fromJust (parent cur)
155 let p = fromJust (parent cur)
155 liftIO $ atomically $ writeTChan (zombieChannel p) (threadId cur)
156 liftIO $ atomically $ writeTChan (zombieChannel p) (threadId cur)
@@ -182,10 +183,14 @@ data StreamData a =
182 | SError SomeException -- ^ An error occurred
183 | SError SomeException -- ^ An error occurred
183 deriving (Typeable, Show,Read)
184 deriving (Typeable, Show,Read)
186+-- XXX instead of starting a new thread every time, reuse the existing child
187+-- threads and send them work via a shared channel. When there is no more work
188+-- available we need a way to close the channel and wakeup all waiters so that
185-- | Execute the IO action and the continuation
191-- | Execute the IO action and the continuation
186genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => EventF -> IO (StreamData t) -> m ()
192genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => EventF -> IO (StreamData t) -> m ()
187−genAsyncEvents parentc rec =
193+genAsyncEvents parentc rec = forkMaybe parentc loop
188− forkMaybe parentc $ \childcont -> loop childcont >> return ()