608d3ef · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Reorganize modules

608d3efHarendra Kumar9 years ago

src/Strands.hs−1

compare with master

unifiedsplit

@@ -16,5 +16,4 @@ module Strands

16 )

17where

19−import Strands.AsyncT

20import Strands.Threads

19import Strands.Threads

src/Strands/AsyncT.hs+1−124

@@ -11,11 +11,6 @@

12module Strands.AsyncT

13 ( AsyncT (..)

14− , waitAsync

15− , processOneEvent

16− , drainChildren

17− , waitForChildren

18− , getCtxResultDest

19 , (<**)

14 , (<**)

20 , onNothing

15 , onNothing

21 , dbg

16 , dbg

@@ -23,12 +18,8 @@ module Strands.AsyncT

23where

18where

25import Control.Applicative (Alternative (..))

20import Control.Applicative (Alternative (..))

26−import Control.Concurrent (ThreadId, killThread)

27−import Control.Concurrent.STM (TChan, atomically, newTChan,

28− readTChan, writeTChan)

29import Control.Monad.Base (MonadBase (..), liftBaseDefault)

21import Control.Monad.Base (MonadBase (..), liftBaseDefault)

30−import Control.Monad.Catch (MonadCatch, MonadThrow, try,

22+import Control.Monad.Catch (MonadThrow, throwM)

31− throwM, SomeException)

32import Control.Monad.State (MonadIO (..), MonadPlus (..),

23import Control.Monad.State (MonadIO (..), MonadPlus (..),

33 MonadState (..), StateT (..),

24 MonadState (..), StateT (..),

34 liftM, modify, runStateT, when)

25 liftM, modify, runStateT, when)

@@ -38,11 +29,6 @@ import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),

38 defaultLiftBaseWith,

29 defaultLiftBaseWith,

39 defaultRestoreM)

30 defaultRestoreM)

40import Data.Dynamic (Typeable)

31import Data.Dynamic (Typeable)

41−import Data.IORef (IORef, modifyIORef, newIORef, readIORef,

42− writeIORef)

43−import Data.List (delete)

44−import Data.Maybe (fromJust, isJust, isNothing)

45−import Unsafe.Coerce (unsafeCoerce)

47import Strands.Context

33import Strands.Context

@@ -157,115 +143,6 @@ instance (MonadBaseControl b m, MonadIO m) => MonadBaseControl b (AsyncT m) wher

157instance MonadThrow m => MonadThrow (AsyncT m) where

143instance MonadThrow m => MonadThrow (AsyncT m) where

158 throwM e = lift $ throwM e

144 throwM e = lift $ throwM e

159

145

160−------------------------------------------------------------------------------

161−-- Thread management

162−------------------------------------------------------------------------------

163−

164−-- XXX We are using unbounded channels so this will not block on writing to

165−-- pchan. We can use bounded channels to throttle the creation of threads based

166−-- on consumption rate.

167−processOneEvent :: MonadIO m

168− => ChildEvent a

169− -> Either (TChan (ChildEvent a)) (IORef [a])

170− -> [ThreadId]

171− -> m ([ThreadId], Maybe SomeException)

172−processOneEvent ev dest pending = do

173− -- Collect results unless we have already encountered an exception.

174− case ev of

175− ChildDone _ (Just e) -> handleException e

176− ChildResult (Left e) -> handleException e

177− ChildDone tid Nothing -> return (delete tid pending, Nothing)

178− ChildResult (Right r) -> do

179− case dest of

180− Left chan -> liftIO $ atomically $ writeTChan chan ev

181− Right ref ->

182− liftIO $ modifyIORef ref $ \rs -> unsafeCoerce r : rs

183− return (pending, Nothing)

184− where

185−

186− handleException e = do

187− liftIO $ mapM_ killThread pending

188− return (pending, Just e)

189−

190−drainChildren :: MonadIO m

191− => Either (TChan (ChildEvent a)) (IORef [a])

192− -> TChan (ChildEvent a)

193− -> [ThreadId]

194− -> m ([ThreadId], Maybe SomeException)

195−drainChildren dest cchan pending =

196− case pending of

197− [] -> return (pending, Nothing)

198− _ -> do

199− ev <- liftIO $ atomically $ readTChan cchan

200− (p, e) <- processOneEvent ev dest pending

201− maybe (drainChildren dest cchan p) (const $ return (p, e)) e

202−

203−waitForChildren :: MonadIO m => Context -> m (Maybe SomeException)

204−waitForChildren ctx = do

205− let pendingRef = pendingThreads ctx

206− pending <- liftIO $ readIORef pendingRef

207− (p, e) <- drainChildren (getCtxResultDest ctx) (childChannel ctx) pending

208− liftIO $ writeIORef pendingRef p

209− return e

210−

211−getCtxResultDest :: Context -> Either (TChan (ChildEvent a)) (IORef [a])

212−getCtxResultDest ctx =

213− maybe (Right $ unsafeCoerce $ fromJust $ accumResults ctx)

214− (Left . unsafeCoerce) (parentChannel ctx)

215−

216−------------------------------------------------------------------------------

217−-- Running the monad

218−------------------------------------------------------------------------------

219−

220−collectResult :: MonadIO m => a -> StateT Context m ()

221−collectResult r = do

222− ctx <- get

223− case parentChannel ctx of

224− Nothing -> do

225− let ref = fromJust $ accumResults ctx

226− liftIO $ modifyIORef ref $ \rs -> unsafeCoerce r : rs

227− Just chan -> do

228− -- XXX can we pass the result directly to the root thread

229− -- instead of passing through all the parents? We can let the

230− -- parent go away and handle the ChildDone events as well in

231− -- the root thread.

232− liftIO $ atomically $ writeTChan chan

233− (ChildResult (Right (unsafeCoerce r)))

234−

235−-- | Invoked to store the result of the computation in the context and finish

236−-- the computation when the computation is done

237−finishComputation :: MonadIO m => a -> AsyncT m b

238−finishComputation x = AsyncT $ do

239− collectResult x

240− return Nothing

241−

242−-- XXX pass a collector function and return a Traversable.

243−-- XXX Ideally it should be a non-empty list instead.

244−-- | Run an 'AsyncT m' computation. Returns a list of results of the

245−-- computation or may throw an exception.

246−waitAsync :: forall m a. (MonadIO m, MonadCatch m)

247− => AsyncT m a -> m [a]

248−waitAsync m = do

249− childChan <- liftIO $ atomically newTChan

250− pendingRef <- liftIO $ newIORef []

251− resultsRef <- liftIO $ newIORef []

252− credit <- liftIO $ newIORef maxBound

253−

254− let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit

255− finishComputation resultsRef

256−

257− r <- try $ runStateT (runAsyncT $ m >>= finishComputation) ctx

258−

259− case r of

260− Left (exc :: SomeException) -> do

261− liftIO $ readIORef pendingRef >>= mapM_ killThread

262− throwM exc

263− Right _ -> do

264− e <- waitForChildren ctx

265− case e of

266− Just (exc :: SomeException) -> throwM exc

267− Nothing -> liftIO $ readIORef resultsRef

268−

269------------------------------------------------------------------------------

146------------------------------------------------------------------------------

270-- * Extensible State: Session Data Management

147-- * Extensible State: Session Data Management

271------------------------------------------------------------------------------

148------------------------------------------------------------------------------

src/Strands/Context.hs+7

@@ -11,6 +11,7 @@ module Strands.Context

11 , resumeContext

12 , setContextMailBox

13 , takeContextMailBox

14+ , getCtxResultDest

14 , Location(..)

15 , Location(..)

15 , getLocation

16 , getLocation

16 , setLocation

17 , setLocation

@@ -30,6 +31,7 @@ import qualified Control.Monad.Trans.State.Lazy as Lazy (get, gets, modify, put)

30import Data.Dynamic (TypeRep, Typeable, typeOf)

31import Data.Dynamic (TypeRep, Typeable, typeOf)

31import Data.IORef (IORef)

32import Data.IORef (IORef)

32import qualified Data.Map as M

33import qualified Data.Map as M

34+import Data.Maybe (fromJust)

33import Unsafe.Coerce (unsafeCoerce)

35import Unsafe.Coerce (unsafeCoerce)

34import GHC.Prim (Any)

36import GHC.Prim (Any)

@@ -213,6 +215,11 @@ takeContextMailBox = do

213 Lazy.put ctx { mailBox = Nothing }

215 Lazy.put ctx { mailBox = Nothing }

214 return $ Right (unsafeCoerce x)

216 return $ Right (unsafeCoerce x)

215

217

218+getCtxResultDest :: Context -> Either (TChan (ChildEvent a)) (IORef [a])

219+getCtxResultDest ctx =

220+ maybe (Right $ unsafeCoerce $ fromJust $ accumResults ctx)

221+ (Left . unsafeCoerce) (parentChannel ctx)

222+

216------------------------------------------------------------------------------

223------------------------------------------------------------------------------

217-- * Extensible State: Session Data Management

224-- * Extensible State: Session Data Management

218------------------------------------------------------------------------------

225------------------------------------------------------------------------------

src/Strands/Threads.hs+104−2

@@ -9,10 +9,11 @@ module Strands.Threads

9 , sync

10 --, react

11 , threads

12+ , waitAsync

12 )

13 )

13where

14where

15−import Control.Applicative ((<|>))

16+import Control.Applicative ((<|>), empty)

16import Control.Concurrent (ThreadId, forkIO, killThread,

17import Control.Concurrent (ThreadId, forkIO, killThread,

17 myThreadId, threadDelay)

18 myThreadId, threadDelay)

18import Control.Concurrent.STM (TChan, atomically, newTChan,

19import Control.Concurrent.STM (TChan, atomically, newTChan,

@@ -21,7 +22,8 @@ import Control.Concurrent.STM (TChan, atomically, newTChan,

21import Control.Exception (ErrorCall (..),

22import Control.Exception (ErrorCall (..),

22 SomeException (..), catch)

23 SomeException (..), catch)

23import qualified Control.Exception.Lifted as EL

24import qualified Control.Exception.Lifted as EL

24−import Control.Monad.Catch (MonadThrow, throwM)

25+import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,

26+ try)

25import Control.Monad.IO.Class (MonadIO (..))

27import Control.Monad.IO.Class (MonadIO (..))

26import Control.Monad.State (get, gets, modify,

28import Control.Monad.State (get, gets, modify,

27 runStateT, when, StateT)

29 runStateT, when, StateT)

@@ -80,6 +82,53 @@ runContext ctx = do

80-- Thread Management (creation, reaping and killing)

82-- Thread Management (creation, reaping and killing)

81------------------------------------------------------------------------------

83------------------------------------------------------------------------------

85+-- XXX We are using unbounded channels so this will not block on writing to

86+-- pchan. We can use bounded channels to throttle the creation of threads based

87+-- on consumption rate.

88+processOneEvent :: MonadIO m

89+ => ChildEvent a

90+ -> Either (TChan (ChildEvent a)) (IORef [a])

91+ -> [ThreadId]

92+ -> m ([ThreadId], Maybe SomeException)

93+processOneEvent ev dest pending = do

94+ -- Collect results unless we have already encountered an exception.

95+ case ev of

96+ ChildDone _ (Just e) -> handleException e

97+ ChildResult (Left e) -> handleException e

98+ ChildDone tid Nothing -> return (delete tid pending, Nothing)

99+ ChildResult (Right r) -> do

100+ case dest of

101+ Left chan -> liftIO $ atomically $ writeTChan chan ev

102+ Right ref ->

103+ liftIO $ modifyIORef ref $ \rs -> unsafeCoerce r : rs

104+ return (pending, Nothing)

105+ where

106+

107+ handleException e = do

108+ liftIO $ mapM_ killThread pending

109+ return (pending, Just e)

110+

111+drainChildren :: MonadIO m

112+ => Either (TChan (ChildEvent a)) (IORef [a])

113+ -> TChan (ChildEvent a)

114+ -> [ThreadId]

115+ -> m ([ThreadId], Maybe SomeException)

116+drainChildren dest cchan pending =

117+ case pending of

118+ [] -> return (pending, Nothing)

119+ _ -> do

120+ ev <- liftIO $ atomically $ readTChan cchan

121+ (p, e) <- processOneEvent ev dest pending

122+ maybe (drainChildren dest cchan p) (const $ return (p, e)) e

123+

124+waitForChildren :: MonadIO m => Context -> m (Maybe SomeException)

125+waitForChildren ctx = do

126+ let pendingRef = pendingThreads ctx

127+ pending <- liftIO $ readIORef pendingRef

128+ (p, e) <- drainChildren (getCtxResultDest ctx) (childChannel ctx) pending

129+ liftIO $ writeIORef pendingRef p

130+ return e

131+

83tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()

132tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()

84tryReclaimZombies ctx = do

133tryReclaimZombies ctx = do

85 let dest = getCtxResultDest ctx

134 let dest = getCtxResultDest ctx

@@ -466,3 +515,56 @@ noTrans x = AsyncT $ x >>= return . Just

466-- modifyThreads :: MonadIO m => (Int -> Int) -> AsyncT m ()

515-- modifyThreads :: MonadIO m => (Int -> Int) -> AsyncT m ()

467-- modifyThreads f =

516-- modifyThreads f =

468-}

517-}

518+

519+------------------------------------------------------------------------------

520+-- Running the monad

521+------------------------------------------------------------------------------

522+

523+collectResult :: MonadIO m => a -> StateT Context m ()

524+collectResult r = do

525+ ctx <- get

526+ case parentChannel ctx of

527+ Nothing -> do

528+ let ref = fromJust $ accumResults ctx

529+ liftIO $ modifyIORef ref $ \rs -> unsafeCoerce r : rs

530+ Just chan -> do

531+ -- XXX can we pass the result directly to the root thread

532+ -- instead of passing through all the parents? We can let the

533+ -- parent go away and handle the ChildDone events as well in

534+ -- the root thread.

535+ liftIO $ atomically $ writeTChan chan

536+ (ChildResult (Right (unsafeCoerce r)))

537+

538+-- | Invoked to store the result of the computation in the context and finish

539+-- the computation when the computation is done

540+finishComputation :: MonadIO m => a -> AsyncT m b

541+finishComputation x = AsyncT $ do

542+ collectResult x

543+ return Nothing

544+

545+-- XXX pass a collector function and return a Traversable.

546+-- XXX Ideally it should be a non-empty list instead.

547+-- | Run an 'AsyncT m' computation. Returns a list of results of the

548+-- computation or may throw an exception.

549+waitAsync :: forall m a. (MonadIO m, MonadCatch m)

550+ => AsyncT m a -> m [a]

551+waitAsync m = do

552+ childChan <- liftIO $ atomically newTChan

553+ pendingRef <- liftIO $ newIORef []

554+ resultsRef <- liftIO $ newIORef []

555+ credit <- liftIO $ newIORef maxBound

556+

557+ let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit

558+ finishComputation resultsRef

559+

560+ r <- try $ runStateT (runAsyncT $ m >>= finishComputation) ctx

561+

562+ case r of

563+ Left (exc :: SomeException) -> do

564+ liftIO $ readIORef pendingRef >>= mapM_ killThread

565+ throwM exc

566+ Right _ -> do

567+ e <- waitForChildren ctx

568+ case e of

569+ Just (exc :: SomeException) -> throwM exc

570+ Nothing -> liftIO $ readIORef resultsRef