4adbddf · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add context save, restore and continue functions

4adbddfHarendra Kumar9 years ago

duct.cabal+2−1

compare with master

unifiedsplit

@@ -20,7 +20,7 @@ cabal-version: >= 1.10

20library

21 hs-source-dirs: src

22 exposed-modules: Duct

23− , Duct.Event

23+ , Duct.Context

24 , Duct.Log

25 , Duct.AsyncT

26 , Duct.Threads

@@ -33,6 +33,7 @@ library

33 , containers >= 0.5.6

34 , time >= 1.5

35 , transformers >= 0.4.2

36+ , ghc-prim

37 -- libraries not bundled with GHC

38 -- libraries not bundled with GHC

38 , lifted-base

39 , lifted-base

src/Duct/AsyncT.hs+9−43

@@ -14,7 +14,6 @@ module Duct.AsyncT

14 , waitAsync

15 , (<**)

16 , onNothing

17− , RemoteStatus (..)

18 , waitForChildren

17 , waitForChildren

19 , dbg

18 , dbg

20 )

19 )

@@ -39,21 +38,13 @@ import Data.IORef (IORef, newIORef, readIORef,

39 writeIORef)

38 writeIORef)

40import Data.List (delete)

39import Data.List (delete)

41import Data.Maybe (isJust, isNothing)

40import Data.Maybe (isJust, isNothing)

42−import Unsafe.Coerce (unsafeCoerce)

44−import Duct.Event

42+import Duct.Context

46-- import Debug.Trace

44-- import Debug.Trace

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

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

50−------------------------------------------------------------------------------

51−-- Remote data types

52−------------------------------------------------------------------------------

53−

54−data RemoteStatus = WasRemote | WasParallel | NoRemote

55− deriving (Typeable, Eq, Show)

56−

57------------------------------------------------------------------------------

48------------------------------------------------------------------------------

58-- Utilities

49-- Utilities

59------------------------------------------------------------------------------

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

@@ -96,12 +87,10 @@ instance Monad m => Alternative (AsyncT m) where

96 empty = AsyncT $ return Nothing

87 empty = AsyncT $ return Nothing

97 (<|>) x y = AsyncT $ do

88 (<|>) x y = AsyncT $ do

98 mx <- runAsyncT x

89 mx <- runAsyncT x

99− was <- getData `onNothing` return NoRemote

90+ loc <- getLocation

100− if was == WasRemote

101− then return Nothing

102− else case mx of

91+ case loc of

92+ RemoteNode -> return Nothing

103− Nothing -> runAsyncT y

93+ _ -> maybe (runAsyncT y) (return . Just) mx

104− justx -> return justx

105

106-- | A synonym of 'empty' that can be used in a monadic expression. It stops

95-- | A synonym of 'empty' that can be used in a monadic expression. It stops

107-- the computation, which allows the next computation in an 'Alternative'

96-- the computation, which allows the next computation in an 'Alternative'

@@ -113,38 +102,15 @@ stop = empty

113-- Monad

102-- Monad

114------------------------------------------------------------------------------

103------------------------------------------------------------------------------

115

104

116−-- | Total variant of `tail` that returns an empty list when given an empty list.

117−tailsafe :: [a] -> [a]

118−tailsafe [] = []

119−tailsafe (_:xs) = xs

120−

121instance Monad m => Monad (AsyncT m) where

105instance Monad m => Monad (AsyncT m) where

122− return = pure

106+ return = pure

123−

124− m >>= f = AsyncT $ do

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

126− -- bind operation completes, we run the operation manually in the new

127− -- thread.

128− modify $ \Context { fstack = fs, .. } ->

129− Context { currentm = unsafeCoerce m

130− , fstack = unsafeCoerce f : fs

131− , .. }

132

107

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

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

134 -- a new thread somewhere in the middle we unwind the fstack and run these

109 -- a new thread somewhere in the middle we unwind the fstack and run these

135 -- functions manually after migration.

110 -- functions manually after migration.

136− mres <- runAsyncT m

111+ m >>= f = AsyncT $ do

137−

112+ saveContext m f

138− let res = case mres of

139− Nothing -> empty

140− Just x -> f x

141−

142− modify $ \Context { fstack = fs, .. } ->

143− Context { currentm = unsafeCoerce res

144− , fstack = tailsafe fs

145− , .. }

146−

147− runAsyncT res

113+ runAsyncT m >>= restoreContext >>= runAsyncT

148

114

149instance Monad m => MonadPlus (AsyncT m) where

115instance Monad m => MonadPlus (AsyncT m) where

150 mzero = empty

116 mzero = empty

src/Duct/Event.hs → src/Duct/Context.hsrenamed+77−11

@@ -1,9 +1,16 @@

1{-# LANGUAGE ExistentialQuantification #-}

2{-# LANGUAGE FlexibleContexts #-}

3+{-# LANGUAGE RecordWildCards #-}

4−module Duct.Event

5+module Duct.Context

5 ( Context(..)

6 ( Context(..)

6 , initContext

7 , initContext

8+ , saveContext

9+ , restoreContext

10+ , continueContext

11+ , Location(..)

12+ , getLocation

13+ , setLocation

7 , getData

14 , getData

8 , setData

15 , setData

9 , modifyData

16 , modifyData

@@ -11,33 +18,39 @@ module Duct.Event

11 )

18 )

12where

19where

21+import Control.Applicative (Alternative(..))

14import Control.Concurrent (ThreadId)

22import Control.Concurrent (ThreadId)

15import Control.Concurrent.STM (TChan)

23import Control.Concurrent.STM (TChan)

16−import Control.Monad.State (MonadState, gets, modify)

24+import Control.Monad.State (MonadState, gets, modify, StateT, MonadPlus)

25+import qualified Control.Monad.Trans.State.Lazy as Lazy (get, gets, modify, put)

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

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

18import Data.IORef (IORef)

27import Data.IORef (IORef)

19import qualified Data.Map as M

28import qualified Data.Map as M

20import Unsafe.Coerce (unsafeCoerce)

29import Unsafe.Coerce (unsafeCoerce)

30+import GHC.Prim (Any)

22------------------------------------------------------------------------------

32------------------------------------------------------------------------------

23-- State of a continuation

33-- State of a continuation

24------------------------------------------------------------------------------

34------------------------------------------------------------------------------

26−type SData = ()

27−

28-- | Describes the context of a computation.

36-- | Describes the context of a computation.

29−data Context = forall m a b. Context

37+data Context = Context

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

38+ { event :: Maybe () -- untyped, XXX rename to mailbox - can we do

31 -- away with this?

39 -- away with this?

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

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

34 -- the 'm' and 'f' in an 'm >>= f' operation of the monad

42 -- the 'm' and 'f' in an 'm >>= f' operation of the monad

35 -- In nested binds we store the current m only, but the whole stack of fs

43 -- In nested binds we store the current m only, but the whole stack of fs

36− , currentm :: m a

44+ , currentm :: Any -- untyped, the real type is: m a

37− , fstack :: [a -> m b]

45+ , fstack :: [Any] -- untyped, the real type is: [a -> m b]

38 -- ^ List of continuations

46 -- ^ List of continuations

48+ -- log only when there is a restore or if we are teleporting

49+ -- We can use a HasLog constraint to statically enable/disable logging

50+ -- , journal :: Maybe Log

51+ , location :: Location

52+

40− , mfData :: M.Map TypeRep SData

53+ , mfData :: M.Map TypeRep Any -- untyped, type coerced

41 -- ^ State data accessed with get or put operations

54 -- ^ State data accessed with get or put operations

43 -- XXX All of the following can be removed

56 -- XXX All of the following can be removed

@@ -92,15 +105,68 @@ initContext

92 -> IORef Int

105 -> IORef Int

93 -> Context

106 -> Context

94initContext x childChan pending credit =

107initContext x childChan pending credit =

95− Context { event = mempty

108+ Context { event = mempty

96− , currentm = x

109+ , currentm = unsafeCoerce x

97 , fstack = []

110 , fstack = []

111+ , location = Local

98 , mfData = mempty

112 , mfData = mempty

99 , parentChannel = Nothing

113 , parentChannel = Nothing

100 , childChannel = childChan

114 , childChannel = childChan

101 , pendingThreads = pending

115 , pendingThreads = pending

102 , threadCredit = credit }

116 , threadCredit = credit }

103

117

118+------------------------------------------------------------------------------

119+-- Where is the computation running?

120+------------------------------------------------------------------------------

121+

122+data Location = Local | ForkedThread | RemoteNode

123+ deriving (Typeable, Eq, Show)

124+

125+getLocation :: Monad m => StateT Context m Location

126+getLocation = Lazy.gets location

127+

128+setLocation :: Monad m => Location -> StateT Context m ()

129+setLocation loc = Lazy.modify $ \Context { .. } -> Context { location = loc, .. }

130+

131+------------------------------------------------------------------------------

132+-- Saving and restoring the context of a computation

133+------------------------------------------------------------------------------

134+

135+-- Save the 'm' and 'f', in case we migrate to a new thread before the current

136+-- bind operation completes, we run the operation manually in the new thread

137+-- using this context.

138+saveContext :: Monad m => f a -> (a -> f b) -> StateT Context m ()

139+saveContext m f =

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

141+ Context { currentm = unsafeCoerce m

142+ , fstack = unsafeCoerce f : fs

143+ , .. }

144+

145+-- pop the top function from the continuation stack, create the next closure,

146+-- set it as the current closure and return it.

147+restoreContext :: (Alternative f, Monad m) => Maybe a -> StateT Context m (f b)

148+restoreContext x = do

149+ -- XXX fstack must be non-empty when this is called.

150+ ctx@Context { fstack = f:fs } <- Lazy.get

151+

152+ let mres = case x of

153+ Nothing -> empty

154+ Just y -> (unsafeCoerce f) y

155+

156+ Lazy.put ctx { currentm = unsafeCoerce mres, fstack = fs }

157+

158+ return mres

159+

160+continueContext :: MonadPlus m => Context -> m a

161+continueContext Context { currentm = m, fstack = fs } =

162+ unsafeCoerce m >>= runfStack (unsafeCoerce fs)

163+

164+ where

165+

166+ -- runfStack :: Monad m => [a -> AsyncT m a] -> a -> AsyncT m b

167+ runfStack [] _ = empty

168+ runfStack (f:ff) x = f x >>= runfStack ff

169+

104------------------------------------------------------------------------------

170------------------------------------------------------------------------------

105-- * Extensible State: Session Data Management

171-- * Extensible State: Session Data Management

106------------------------------------------------------------------------------

172------------------------------------------------------------------------------

src/Duct/Threads.hs+12−19

@@ -12,7 +12,7 @@ module Duct.Threads

12 )

13where

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

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

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

17 myThreadId, threadDelay)

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

@@ -35,7 +35,7 @@ import Data.Maybe (fromJust)

35import Unsafe.Coerce (unsafeCoerce)

37import Duct.AsyncT

38−import Duct.Event

38+import Duct.Context

40------------------------------------------------------------------------------

41-- Pick up from where we left in the previous thread

@@ -45,15 +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)) => Context -> StateT Context m (Maybe a)

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

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

49+continue = runAsyncT . continueContext

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

51−

52− where

53−

54− runfStack :: Monad m => [a -> AsyncT m a] -> a -> AsyncT m b

55− runfStack [] _ = empty

56− runfStack (f:fs) x = f x >>= runfStack fs

58------------------------------------------------------------------------------

51------------------------------------------------------------------------------

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

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

@@ -244,8 +237,8 @@ parallel ioaction = AsyncT $ do

244 -- in this thread or a new thread.

237 -- in this thread or a new thread.

245 Nothing -> do

238 Nothing -> do

246 lift $ genAsyncEvents cont ioaction

239 lift $ genAsyncEvents cont ioaction

247− was <- getData `onNothing` return NoRemote

240+ loc <- getLocation

248− when (was /= WasRemote) $ setData WasParallel

241+ when (loc /= RemoteNode) $ setLocation ForkedThread

249 return Nothing

242 return Nothing

250

243

251-- | An task stream generator that produces an infinite stream of tasks by

244-- | An task stream generator that produces an infinite stream of tasks by

@@ -273,10 +266,10 @@ async io = do

273-- computation. Note that in Applicatives it might result in an undesired

266-- computation. Note that in Applicatives it might result in an undesired

274-- serialization.

267-- serialization.

275sync :: MonadIO m => AsyncT m a -> AsyncT m a

268sync :: MonadIO m => AsyncT m a -> AsyncT m a

276−sync x = do

269+sync x = AsyncT $ do

277− setData WasRemote

270+ setLocation RemoteNode

278− r <- x

271+ r <- runAsyncT x

279− delData WasRemote

272+ setLocation Local

280 return r

273 return r

281

274

282-- | An task stream generator that produces an infinite stream of tasks by

275-- | An task stream generator that produces an infinite stream of tasks by

@@ -320,8 +313,8 @@ react setHandler iob = AsyncT $ do

320 lift $ setHandler $ \dat ->do

313 lift $ setHandler $ \dat ->do

321 runStateT (continue cont) cont{event= Just $ unsafeCoerce dat}

314 runStateT (continue cont) cont{event= Just $ unsafeCoerce dat}

322 liftIO iob

315 liftIO iob

323− was <- getData `onNothing` return NoRemote

316+ loc <- getLocation

324− when (was /= WasRemote) $ setData WasParallel

317+ when (loc /= RemoteNode) $ setLocation ForkedThread

325 return Nothing

318 return Nothing

326

319

327 j@(Just _) -> do

320 j@(Just _) -> do

test/1-event.hs+5−6

@@ -6,23 +6,22 @@ import Control.Concurrent.STM

6import Control.Monad.State

7import Data.IORef

9−import Duct.Event

9+import Duct.Context

11−f :: StateT EventF IO String

11+f :: StateT Context IO String

12f = do

13 setData "x"

14 Just x <- getData

15 return x

17−runEvent :: forall m a. (Alternative m, MonadIO m) => StateT EventF m a -> m a

17+runEvent :: forall m a. (Alternative m, MonadIO m) => StateT Context m a -> m a

18runEvent t = do

19 zombieChan <- liftIO $ atomically newTChan

20 pendingRef <- liftIO $ newIORef []

21 credit <- liftIO $ newIORef maxBound

22− th <- liftIO $ myThreadId

24− (r, _) <- runStateT t $ initEventF

23+ (r, _) <- runStateT t $ initContext

25− (empty :: m a) th zombieChan pendingRef credit

24+ (empty :: m a) zombieChan pendingRef credit

26 return r

25 return r

28main = do

27main = do