transient 0.5.5 → 0.5.6
raw patch · 10 files changed
+2597/−2534 lines, 10 filesdep ~basesetup-changedPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependency ranges changed: base
API changes (from Hackage documentation)
+ Transient.Base: catcht :: Exception e => TransIO b -> (e -> TransIO b) -> TransIO b
+ Transient.Base: input' :: (Typeable a, Read a, Show a) => Maybe a -> (a -> Bool) -> String -> TransIO a
+ Transient.Base: throwt :: Exception e => e -> TransIO a
+ Transient.Internals: input' :: (Typeable a, Read a, Show a) => Maybe a -> (a -> Bool) -> String -> TransIO a
+ Transient.Internals: throwt :: Exception e => e -> TransIO a
- Transient.Base: input :: (Typeable a, Read a, Show a) => (a -> Bool) -> String -> TransIO a
+ Transient.Base: input :: (Show a, Read a, Typeable * a) => (a -> Bool) -> String -> TransIO a
- Transient.Internals: input :: (Typeable a, Read a, Show a) => (a -> Bool) -> String -> TransIO a
+ Transient.Internals: input :: (Show a, Read a, Typeable * a) => (a -> Bool) -> String -> TransIO a
- Transient.Internals: labelState :: String -> TransIO ()
+ Transient.Internals: labelState :: (MonadIO m, MonadState EventF m) => String -> m ()
Files
- LICENSE +17/−17
- README.md +62/−62
- Setup.hs +2/−2
- src/Transient/Base.hs +296/−296
- src/Transient/EVars.hs +84/−84
- src/Transient/Indeterminism.hs +153/−153
- src/Transient/Internals.hs +1622/−1634
- src/Transient/Logged.hs +230/−230
- tests/TestSuite.hs +52/−0
- transient.cabal +79/−56
LICENSE view
@@ -1,18 +1,18 @@-Copyright © 2014-2016 Alberto G. Corona <https://github.com/agocorona> - -Permission is hereby granted, free of charge, to any person obtaining a copy of -this software and associated documentation files (the "Software"), to deal in -the Software without restriction, including without limitation the rights to -use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of -the Software, and to permit persons to whom the Software is furnished to do so, -subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS -FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR -COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER -IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +Copyright © 2014-2016 Alberto G. Corona <https://github.com/agocorona>++Permission is hereby granted, free of charge, to any person obtaining a copy of+this software and associated documentation files (the "Software"), to deal in+the Software without restriction, including without limitation the rights to+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of+the Software, and to permit persons to whom the Software is furnished to do so,+subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
README.md view
@@ -1,62 +1,62 @@- -========= - -[](http://hackage.haskell.org/package/transient) -[](http://stackage.org/lts/package/transient) -[](http://stackage.org/nightly/package/transient) -[](https://travis-ci.org/transient-haskell/transient) -[](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link) - -NOTE: distributed computing and web primitives have been moved to [transient-universe](https://github.com/agocorona/transient-universe) and [ghcjs-hplay](https://github.com/agocorona/ghcjs-hplay) - - -## Some feedback on `transient`: - -1. Rahul Muttineni @rahulmutt nov. 09 2016 03:40 Lead developper of ETA (the JVM Haskell compiler) - - *It's a bit mind bending in that it's like using a higher-level list monad, but it's very, very cool. For beginning Haskellers, what would be really useful is a visualisation of what happens when you do various distributed/parallel stuff.* **It's almost shocking how effortlessly you can run computations across threads/nodes.** - - *The cool part is the composability in the distributed setting. *You can make higher-order monadic functions that allow you to compose & reuse a long chain of distributed transactions via `wormhole` and `teleport`*. Another benefit is that the transaction becomes first class and* **you can see exactly what's going on in one place** *instead of distributing the logic across actors making the code equivalent to event callbacks, as you've stated.* - - https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?at=58228caa35e6cf054773303b - -## What is Transient? - -One of the dreams of software engineering is unrestricted composability. - -This may be put in these terms: - -let `ap1` and `ap2` two applications with arbitrary complexity, with all effects including multiple threads, asynchronous IO, indeterminism, events and perhaps, distributed computing. - -Then the combinations: - - - ap1 <|> ap2 -- Alternative expression - - ap1 >>= \x -> ap2 -- monadic sequence - - ap1 <> ap2 -- monoidal expression - - (,) <$> ap1 <*> ap2 -- Applicative expression - -are possible if the types match, and generate new applications that are composable as well. - -Transient does exactly that. - -The operators `<$>` `<*>` and `<>` express concurrency, the operator `<|>` express parallelism and `>>=` for sequencing of threads and/or distributed processes. So even in the presence of these effects and others, everything is composable. - -For this purpose transient is an extensible effects monad with all major effects and primitives for parallelism, events, asynchronous IO, early termination, non-determinism logging and distributed computing. Since it is possible to extend it with more effects without adding monad transformers, the composability is assured. - -Documentation -============= - -The [Wiki](https://github.com/agocorona/transient/wiki) is more user oriented - -My video sessions in [livecoding.tv](https://www.livecoding.tv/agocorona/videos/) not intended as tutorials or presentations, but show some of the latest features running. - -The articles are more technical: - -- [Philosophy, async, parallelism, thread control, events, Session state](https://www.fpcomplete.com/user/agocorona/EDSL-for-hard-working-IT-programmers?show=tutorials) -- [Backtracking and undoing IO transactions](https://www.fpcomplete.com/user/agocorona/the-hardworking-programmer-ii-practical-backtracking-to-undo-actions?show=tutorials) -- [Non-deterministic list like processing, multithreading](https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii?show=tutorials) -- [Distributed computing](https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv?show=tutorials) -- [Publish-Subscribe variables](https://www.schoolofhaskell.com/user/agocorona/publish-subscribe-variables-transient-effects-v) -- [Distributed streaming, map-reduce](https://www.schoolofhaskell.com/user/agocorona/estimation-of-using-distributed-computing-streaming-transient-effects-vi-1) - -These articles contain executable examples (not now, since the site no longer support the execution of haskell snippets). ++=========++[](http://hackage.haskell.org/package/transient)+[](http://stackage.org/lts/package/transient)+[](http://stackage.org/nightly/package/transient)+[](https://travis-ci.org/transient-haskell/transient)+[](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link)++NOTE: distributed computing and web primitives have been moved to [transient-universe](https://github.com/agocorona/transient-universe) and [ghcjs-hplay](https://github.com/agocorona/ghcjs-hplay)+++## Some feedback on `transient`:++1. Rahul Muttineni @rahulmutt nov. 09 2016 03:40 Lead developper of ETA (the JVM Haskell compiler)++ *It's a bit mind bending in that it's like using a higher-level list monad, but it's very, very cool. For beginning Haskellers, what would be really useful is a visualisation of what happens when you do various distributed/parallel stuff.* **It's almost shocking how effortlessly you can run computations across threads/nodes.**++ *The cool part is the composability in the distributed setting. *You can make higher-order monadic functions that allow you to compose & reuse a long chain of distributed transactions via `wormhole` and `teleport`*. Another benefit is that the transaction becomes first class and* **you can see exactly what's going on in one place** *instead of distributing the logic across actors making the code equivalent to event callbacks, as you've stated.*++ https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?at=58228caa35e6cf054773303b++## What is Transient?++One of the dreams of software engineering is unrestricted composability.++This may be put in these terms:++let `ap1` and `ap2` two applications with arbitrary complexity, with all effects including multiple threads, asynchronous IO, indeterminism, events and perhaps, distributed computing.++Then the combinations:++ - ap1 <|> ap2 -- Alternative expression+ - ap1 >>= \x -> ap2 -- monadic sequence+ - ap1 <> ap2 -- monoidal expression+ - (,) <$> ap1 <*> ap2 -- Applicative expression++are possible if the types match, and generate new applications that are composable as well.++Transient does exactly that.++The operators `<$>` `<*>` and `<>` express concurrency, the operator `<|>` express parallelism and `>>=` for sequencing of threads and/or distributed processes. So even in the presence of these effects and others, everything is composable.++For this purpose transient is an extensible effects monad with all major effects and primitives for parallelism, events, asynchronous IO, early termination, non-determinism logging and distributed computing. Since it is possible to extend it with more effects without adding monad transformers, the composability is assured.++Documentation+=============++The [Wiki](https://github.com/agocorona/transient/wiki) is more user oriented++My video sessions in [livecoding.tv](https://www.livecoding.tv/agocorona/videos/) not intended as tutorials or presentations, but show some of the latest features running.++The articles are more technical:++- [Philosophy, async, parallelism, thread control, events, Session state](https://www.fpcomplete.com/user/agocorona/EDSL-for-hard-working-IT-programmers?show=tutorials)+- [Backtracking and undoing IO transactions](https://www.fpcomplete.com/user/agocorona/the-hardworking-programmer-ii-practical-backtracking-to-undo-actions?show=tutorials)+- [Non-deterministic list like processing, multithreading](https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii?show=tutorials)+- [Distributed computing](https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv?show=tutorials)+- [Publish-Subscribe variables](https://www.schoolofhaskell.com/user/agocorona/publish-subscribe-variables-transient-effects-v)+- [Distributed streaming, map-reduce](https://www.schoolofhaskell.com/user/agocorona/estimation-of-using-distributed-computing-streaming-transient-effects-vi-1)++These articles contain executable examples (not now, since the site no longer support the execution of haskell snippets).
Setup.hs view
@@ -1,2 +1,2 @@-import Distribution.Simple -main = defaultMain +import Distribution.Simple+main = defaultMain
src/Transient/Base.hs view
@@ -1,296 +1,296 @@-{-# LANGUAGE ScopedTypeVariables #-} ------------------------------------------------------------------------------ --- --- Module : Base --- Copyright : --- License : MIT --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | Transient provides high level concurrency allowing you to do concurrent --- processing without requiring any knowledge of threads or synchronization. --- From the programmer's perspective, the programming model is single threaded. --- Concurrent tasks are created and composed seamlessly resulting in highly --- modular and composable concurrent programs. Transient has diverse --- applications from simple concurrent applications to massively parallel and --- distributed map-reduce problems. If you are considering Apache Spark or --- Cloud Haskell then transient might be a simpler yet better solution for you --- (see --- <https://github.com/transient-haskell/transient-universe transient-universe>). --- Transient makes it easy to write composable event driven reactive UI --- applications. For example, <https://hackage.haskell.org/package/axiom Axiom> --- is a transient based unified client and server side framework that provides --- a better programming model and composability compared to frameworks like --- ReactJS. --- --- = Overview --- --- The 'TransientIO' monad allows you to: --- --- * Split a problem into concurrent task sets --- * Compose concurrent task sets using non-determinism --- * Collect and combine results of concurrent tasks --- --- You can think of 'TransientIO' as a concurrent list transformer monad with --- many other features added on top e.g. backtracking, logging and recovery to --- move computations across machines for distributed processing. --- --- == Non-determinism --- --- In its non-concurrent form, the 'TransientIO' monad behaves exactly like a --- <http://hackage.haskell.org/package/list-transformer list transformer monad>. --- It is like a list whose elements are generated using IO effects. It composes --- in the same way as a list monad. Let's see an example: --- --- @ --- import Control.Concurrent (threadDelay) --- import Control.Monad.IO.Class (liftIO) --- import System.Random (randomIO) --- import Transient.Base (keep, threads, waitEvents) --- --- main = keep $ threads 0 $ do --- x <- waitEvents (randomIO :: IO Int) --- liftIO $ threadDelay 1000000 --- liftIO $ putStrLn $ show x --- @ --- --- 'keep' runs the 'TransientIO' monad. The 'threads' primitive limits the --- number of threads to force non-concurrent operation. The 'waitEvents' --- primitive generates values (list elements) in a loop using the 'randomIO' IO --- action. The above code behaves like a list monad as if we are drawing --- elements from a list generated by 'waitEvents'. The sequence of actions --- following 'waitEvents' is executed for each element of the list. We see a --- random value printed on the screen every second. As you can see this --- behavior is identical to a list transformer monad. --- --- == Concurrency --- --- 'TransientIO' monad is a concurrent list transformer i.e. each element of --- the generated list can be processed concurrently. In the previous example --- if we change the number of threads to 10 we can see concurrency in action: --- --- @ --- ... --- main = keep $ threads 10 $ do --- ... --- @ --- --- Now each element of the list is processed concurrently in a separate thread, --- up to 10 threads are used. Therefore we see 10 results printed every second --- instead of 1 in the previous version. --- --- In the above examples the list elements are generated using a synchronous IO --- action. These elements can also be asynchronous events, for example an --- interactive user input. In transient, the elements of the list are known as --- tasks. The tasks terminology is general and intuitive in the context of --- transient as tasks can be triggered by asynchronous events and multiple of --- them can run simultaneously in an unordered fashion. --- --- == Composing Tasks --- --- The type @TransientIO a@ represents a /task set/ with each task in --- the set returning a value of type @a@. A task set could be /finite/ or --- /infinite/; multiple tasks could run simultaneously. The absence of a task, --- a void task set or failure is denoted by a special value 'empty' in an --- 'Alternative' composition, or the 'stop' primitive in a monadic composition. --- In the transient programming model the programmer thinks in terms of tasks --- and composes tasks. Whether the tasks run synchronously or concurrently does --- not matter; concurrency is hidden from the programmer for the most part. In --- the previous example the code written for a single threaded list transformer --- works concurrently as well. --- --- We have already seen that the 'Monad' instance provides a way to compose the --- tasks in a sequential, non-deterministic and concurrent manner. When a void --- task set is encountered, the monad stops processing any further computations --- as we have nothing to do. The following example does not generate any --- output after "stop here": --- --- @ --- main = keep $ threads 0 $ do --- x <- waitEvents (randomIO :: IO Int) --- liftIO $ threadDelay 1000000 --- liftIO $ putStrLn $ "stop here" --- stop --- liftIO $ putStrLn $ show x --- @ --- --- When a task creation primitive creates a task concurrently in a new thread --- (e.g. 'waitEvents'), it returns a void task set in the current thread --- making it stop further processing. However, processing resumes from the same --- point onwards with the same state in the new task threads as and when they --- are created; as if the current thread along with its state has branched into --- multiple threads, one for each new task. In the following example you can --- see that the thread id changes after the 'waitEvents' call: --- --- @ --- main = keep $ threads 1 $ do --- mainThread <- liftIO myThreadId --- liftIO $ putStrLn $ "Main thread: " ++ show mainThread --- x <- waitEvents (randomIO :: IO Int) --- --- liftIO $ threadDelay 1000000 --- evThread <- liftIO myThreadId --- liftIO $ putStrLn $ "Event thread: " ++ show evThread --- @ --- --- Note that if we use @threads 0@ then the new task thread is the same as the --- main thread because 'waitEvents' falls back to synchronous non-concurrent --- mode, and therefore returns a non void task set. --- --- In an 'Alternative' composition, when a computation results in 'empty' --- the next alternative is tried. When a task creation primitive creates a --- concurrent task, it returns 'empty' allowing tasks to run concurrently when --- composed with the '<|>' combinator. The following example combines two --- single concurrent tasks generated by 'async': --- --- @ --- main = keep $ do --- x <- event 1 \<|\> event 2 --- liftIO $ putStrLn $ show x --- where event n = async (return n :: IO Int) --- @ --- --- Note that availability of threads can impact the behavior of an application. --- An infinite task set generator (e.g. 'waitEvents' or 'sample') running --- synchronously (due to lack of threads) can block all other computations in --- an 'Alternative' composition. The following example does not trigger the --- 'async' task unless we increase the number of threads to make 'waitEvents' --- asynchronous: --- --- @ --- main = keep $ threads 0 $ do --- x <- waitEvents (randomIO :: IO Int) \<|\> async (return 0 :: IO Int) --- liftIO $ threadDelay 1000000 --- liftIO $ putStrLn $ show x --- @ --- --- == Parallel Map Reduce --- --- The following example uses 'choose' to send the items in a list to parallel --- tasks for squaring and then folds the results of those tasks using 'collect'. --- --- @ --- import Control.Monad.IO.Class (liftIO) --- import Data.List (sum) --- import Transient.Base (keep) --- import Transient.Indeterminism (choose, collect) --- --- main = keep $ do --- collect 100 squares >>= liftIO . putStrLn . show . sum --- where --- squares = do --- x <- choose [1..100] --- return (x * x) --- @ --- --- == State Isolation --- --- State is inherited but never shared. A transient application is written as --- a composition of task sets. New concurrent tasks can be triggered from --- inside a task. A new task inherits the state of the monad at the point --- where it got started. However, the state of a task is always completely --- isolated from other tasks irrespective of whether it is started in a new --- thread or not. The state is referentially transparent i.e. any changes to --- the state creates a new copy of the state. Therefore a programmer does not --- have to worry about synchronization or unintended side effects. --- --- The monad starts with an empty state. At any point you can add ('setData'), --- retrieve ('getSData') or delete ('delData') a data item to or from the --- current state. Creation of a task /branches/ the computation, inheriting --- the previous state, and collapsing (e.g. 'collect') discards the state of --- the tasks being collapsed. If you want to use the state in the results you --- will have to pass it as part of the results of the tasks. --- --- = Reactive Applications --- --- A popular model to handle asynchronous events in imperative languages is the --- callback model. The control flow of the program is driven by events and --- callbacks; callbacks are event handlers that are hooked into the event --- generation code and are invoked every time an event happens. This model --- makes the overall control flow hard to understand resulting into a "callback --- hell" because the logic is distributed across various isolated callback --- handlers, and many different event threads work on the same global state. --- --- Transient provides a better programming model for reactive applications. In --- contrast to the callback model, transient transparently moves the relevant --- state to the respective event threads and composes the results to arrive at --- the new state. The programmer is not aware of the threads, there is no --- shared state to worry about, and a seamless sequential flow enabling easy --- reasoning and composable application components. --- <https://hackage.haskell.org/package/axiom Axiom> is a client and server --- side web UI and reactive application framework built using the transient --- programming model. --- --- = Further Reading --- --- * <https://github.com/transient-haskell/transient/wiki/Transient-tutorial Tutorial> --- * <https://github.com/transient-haskell/transient-examples Examples> --- ------------------------------------------------------------------------------ - -module Transient.Base( --- * The Monad -TransIO(..), TransientIO - --- * Task Composition Operators -, (**>), (<**), (<***), (<|) - --- * Running the monad -,keep, keep', stop, exit - --- * Asynchronous console IO -,option, input - --- * Task Creation --- $taskgen -, StreamData(..) -,parallel, async, waitEvents, sample, spawn, react - --- * State management -,setData, getSData, getData, delData, modifyData, try, setState, getState, delState, modifyState - --- * Thread management -, threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds - --- * Exceptions --- $exceptions - -,onException, cutExceptions, continue - --- * Utilities -,genId -) - -where - - -import Transient.Internals - --- $taskgen --- --- These primitives are used to create asynchronous and concurrent tasks from --- an IO action. --- - --- $exceptions --- --- Exception handlers are implemented using the backtracking mechanism. --- (see 'Transient.Backtrack.back'). Several exception handlers can be --- installed using 'onException'; handlers are run in reverse order when an --- exception is raised. The following example prints "3" and then "2". --- --- @ --- {-\# LANGUAGE ScopedTypeVariables #-} --- import Transient.Base (keep, onException, cutExceptions) --- import Control.Monad.IO.Class (liftIO) --- import Control.Exception (ErrorCall) --- --- main = keep $ do --- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "1" --- cutExceptions --- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "2" --- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "3" --- liftIO $ error "Raised ErrorCall exception" >> return () --- @ +{-# LANGUAGE ScopedTypeVariables #-}+-----------------------------------------------------------------------------+--+-- Module : Base+-- Copyright :+-- License : MIT+--+-- Maintainer : agocorona@gmail.com+-- Stability :+-- Portability :+--+-- | Transient provides high level concurrency allowing you to do concurrent+-- processing without requiring any knowledge of threads or synchronization.+-- From the programmer's perspective, the programming model is single threaded.+-- Concurrent tasks are created and composed seamlessly resulting in highly+-- modular and composable concurrent programs. Transient has diverse+-- applications from simple concurrent applications to massively parallel and+-- distributed map-reduce problems. If you are considering Apache Spark or+-- Cloud Haskell then transient might be a simpler yet better solution for you+-- (see+-- <https://github.com/transient-haskell/transient-universe transient-universe>).+-- Transient makes it easy to write composable event driven reactive UI+-- applications. For example, <https://hackage.haskell.org/package/axiom Axiom>+-- is a transient based unified client and server side framework that provides+-- a better programming model and composability compared to frameworks like+-- ReactJS.+--+-- = Overview+--+-- The 'TransientIO' monad allows you to:+--+-- * Split a problem into concurrent task sets+-- * Compose concurrent task sets using non-determinism+-- * Collect and combine results of concurrent tasks+--+-- You can think of 'TransientIO' as a concurrent list transformer monad with+-- many other features added on top e.g. backtracking, logging and recovery to+-- move computations across machines for distributed processing.+--+-- == Non-determinism+--+-- In its non-concurrent form, the 'TransientIO' monad behaves exactly like a+-- <http://hackage.haskell.org/package/list-transformer list transformer monad>.+-- It is like a list whose elements are generated using IO effects. It composes+-- in the same way as a list monad. Let's see an example:+--+-- @+-- import Control.Concurrent (threadDelay)+-- import Control.Monad.IO.Class (liftIO)+-- import System.Random (randomIO)+-- import Transient.Base (keep, threads, waitEvents)+--+-- main = keep $ threads 0 $ do+-- x <- waitEvents (randomIO :: IO Int)+-- liftIO $ threadDelay 1000000+-- liftIO $ putStrLn $ show x+-- @+--+-- 'keep' runs the 'TransientIO' monad. The 'threads' primitive limits the+-- number of threads to force non-concurrent operation. The 'waitEvents'+-- primitive generates values (list elements) in a loop using the 'randomIO' IO+-- action. The above code behaves like a list monad as if we are drawing+-- elements from a list generated by 'waitEvents'. The sequence of actions+-- following 'waitEvents' is executed for each element of the list. We see a+-- random value printed on the screen every second. As you can see this+-- behavior is identical to a list transformer monad.+--+-- == Concurrency+--+-- 'TransientIO' monad is a concurrent list transformer i.e. each element of+-- the generated list can be processed concurrently. In the previous example+-- if we change the number of threads to 10 we can see concurrency in action:+--+-- @+-- ...+-- main = keep $ threads 10 $ do+-- ...+-- @+--+-- Now each element of the list is processed concurrently in a separate thread,+-- up to 10 threads are used. Therefore we see 10 results printed every second+-- instead of 1 in the previous version.+--+-- In the above examples the list elements are generated using a synchronous IO+-- action. These elements can also be asynchronous events, for example an+-- interactive user input. In transient, the elements of the list are known as+-- tasks. The tasks terminology is general and intuitive in the context of+-- transient as tasks can be triggered by asynchronous events and multiple of+-- them can run simultaneously in an unordered fashion.+--+-- == Composing Tasks+--+-- The type @TransientIO a@ represents a /task set/ with each task in+-- the set returning a value of type @a@. A task set could be /finite/ or+-- /infinite/; multiple tasks could run simultaneously. The absence of a task,+-- a void task set or failure is denoted by a special value 'empty' in an+-- 'Alternative' composition, or the 'stop' primitive in a monadic composition.+-- In the transient programming model the programmer thinks in terms of tasks+-- and composes tasks. Whether the tasks run synchronously or concurrently does+-- not matter; concurrency is hidden from the programmer for the most part. In+-- the previous example the code written for a single threaded list transformer+-- works concurrently as well.+--+-- We have already seen that the 'Monad' instance provides a way to compose the+-- tasks in a sequential, non-deterministic and concurrent manner. When a void+-- task set is encountered, the monad stops processing any further computations+-- as we have nothing to do. The following example does not generate any+-- output after "stop here":+--+-- @+-- main = keep $ threads 0 $ do+-- x <- waitEvents (randomIO :: IO Int)+-- liftIO $ threadDelay 1000000+-- liftIO $ putStrLn $ "stop here"+-- stop+-- liftIO $ putStrLn $ show x+-- @+--+-- When a task creation primitive creates a task concurrently in a new thread+-- (e.g. 'waitEvents'), it returns a void task set in the current thread+-- making it stop further processing. However, processing resumes from the same+-- point onwards with the same state in the new task threads as and when they+-- are created; as if the current thread along with its state has branched into+-- multiple threads, one for each new task. In the following example you can+-- see that the thread id changes after the 'waitEvents' call:+--+-- @+-- main = keep $ threads 1 $ do+-- mainThread <- liftIO myThreadId+-- liftIO $ putStrLn $ "Main thread: " ++ show mainThread+-- x <- waitEvents (randomIO :: IO Int)+--+-- liftIO $ threadDelay 1000000+-- evThread <- liftIO myThreadId+-- liftIO $ putStrLn $ "Event thread: " ++ show evThread+-- @+--+-- Note that if we use @threads 0@ then the new task thread is the same as the+-- main thread because 'waitEvents' falls back to synchronous non-concurrent+-- mode, and therefore returns a non void task set.+--+-- In an 'Alternative' composition, when a computation results in 'empty'+-- the next alternative is tried. When a task creation primitive creates a+-- concurrent task, it returns 'empty' allowing tasks to run concurrently when+-- composed with the '<|>' combinator. The following example combines two+-- single concurrent tasks generated by 'async':+--+-- @+-- main = keep $ do+-- x <- event 1 \<|\> event 2+-- liftIO $ putStrLn $ show x+-- where event n = async (return n :: IO Int)+-- @+--+-- Note that availability of threads can impact the behavior of an application.+-- An infinite task set generator (e.g. 'waitEvents' or 'sample') running+-- synchronously (due to lack of threads) can block all other computations in+-- an 'Alternative' composition. The following example does not trigger the+-- 'async' task unless we increase the number of threads to make 'waitEvents'+-- asynchronous:+--+-- @+-- main = keep $ threads 0 $ do+-- x <- waitEvents (randomIO :: IO Int) \<|\> async (return 0 :: IO Int)+-- liftIO $ threadDelay 1000000+-- liftIO $ putStrLn $ show x+-- @+--+-- == Parallel Map Reduce+--+-- The following example uses 'choose' to send the items in a list to parallel+-- tasks for squaring and then folds the results of those tasks using 'collect'.+--+-- @+-- import Control.Monad.IO.Class (liftIO)+-- import Data.List (sum)+-- import Transient.Base (keep)+-- import Transient.Indeterminism (choose, collect)+--+-- main = keep $ do+-- collect 100 squares >>= liftIO . putStrLn . show . sum+-- where+-- squares = do+-- x <- choose [1..100]+-- return (x * x)+-- @+--+-- == State Isolation+--+-- State is inherited but never shared. A transient application is written as+-- a composition of task sets. New concurrent tasks can be triggered from+-- inside a task. A new task inherits the state of the monad at the point+-- where it got started. However, the state of a task is always completely+-- isolated from other tasks irrespective of whether it is started in a new+-- thread or not. The state is referentially transparent i.e. any changes to+-- the state creates a new copy of the state. Therefore a programmer does not+-- have to worry about synchronization or unintended side effects.+--+-- The monad starts with an empty state. At any point you can add ('setData'),+-- retrieve ('getSData') or delete ('delData') a data item to or from the+-- current state. Creation of a task /branches/ the computation, inheriting+-- the previous state, and collapsing (e.g. 'collect') discards the state of+-- the tasks being collapsed. If you want to use the state in the results you+-- will have to pass it as part of the results of the tasks.+--+-- = Reactive Applications+--+-- A popular model to handle asynchronous events in imperative languages is the+-- callback model. The control flow of the program is driven by events and+-- callbacks; callbacks are event handlers that are hooked into the event+-- generation code and are invoked every time an event happens. This model+-- makes the overall control flow hard to understand resulting into a "callback+-- hell" because the logic is distributed across various isolated callback+-- handlers, and many different event threads work on the same global state.+--+-- Transient provides a better programming model for reactive applications. In+-- contrast to the callback model, transient transparently moves the relevant+-- state to the respective event threads and composes the results to arrive at+-- the new state. The programmer is not aware of the threads, there is no+-- shared state to worry about, and a seamless sequential flow enabling easy+-- reasoning and composable application components.+-- <https://hackage.haskell.org/package/axiom Axiom> is a client and server+-- side web UI and reactive application framework built using the transient+-- programming model.+--+-- = Further Reading+--+-- * <https://github.com/transient-haskell/transient/wiki/Transient-tutorial Tutorial>+-- * <https://github.com/transient-haskell/transient-examples Examples>+--+-----------------------------------------------------------------------------++module Transient.Base(+-- * The Monad+TransIO(..), TransientIO++-- * Task Composition Operators+, (**>), (<**), (<***), (<|)++-- * Running the monad+,keep, keep', stop, exit++-- * Asynchronous console IO+,option, input,input'++-- * Task Creation+-- $taskgen+, StreamData(..)+,parallel, async, waitEvents, sample, spawn, react++-- * State management+,setData, getSData, getData, delData, modifyData, try, setState, getState, delState, modifyState++-- * Thread management+, threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds++-- * Exceptions+-- $exceptions++,onException, cutExceptions, continue, catcht, throwt++-- * Utilities+,genId+)++where+++import Transient.Internals++-- $taskgen+--+-- These primitives are used to create asynchronous and concurrent tasks from+-- an IO action.+--++-- $exceptions+--+-- Exception handlers are implemented using the backtracking mechanism.+-- (see 'Transient.Backtrack.back'). Several exception handlers can be+-- installed using 'onException'; handlers are run in reverse order when an+-- exception is raised. The following example prints "3" and then "2".+--+-- @+-- {-\# LANGUAGE ScopedTypeVariables #-}+-- import Transient.Base (keep, onException, cutExceptions)+-- import Control.Monad.IO.Class (liftIO)+-- import Control.Exception (ErrorCall)+--+-- main = keep $ do+-- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "1"+-- cutExceptions+-- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "2"+-- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "3"+-- liftIO $ error "Raised ErrorCall exception" >> return ()+-- @
src/Transient/EVars.hs view
@@ -1,84 +1,84 @@-{-# LANGUAGE DeriveDataTypeable #-} -module Transient.EVars where - -import Transient.Base -import Transient.Internals(runTransState,onNothing, EventF(..), killChildren) -import qualified Data.Map as M -import Data.Typeable - -import Control.Concurrent -import Control.Applicative -import Control.Concurrent.STM -import Control.Monad.IO.Class -import Control.Exception(SomeException) - -import Data.List(nub) -import Control.Monad.State -import Data.IORef - - - -data EVar a= EVar (TChan (StreamData a)) deriving Typeable - - --- | creates an EVar. --- --- Evars are event vars. `writeEVar` trigger the execution of all the continuations associated to the `readEVar` of this variable --- (the code that is after them). --- --- It is like the publish-subscribe pattern but without inversion of control, since a readEVar can be inserted at any place in the --- Transient flow. --- --- EVars are created upstream and can be used to communicate two sub-threads of the monad. Following the Transient philosophy they --- do not block his own thread if used with alternative operators, unlike the IORefs and TVars. And unlike STM vars, that are composable, --- they wait for their respective events, while TVars execute the whole expression when any variable is modified. --- --- The execution continues after the writeEVar when all subscribers have been executed. --- --- Now the continuations are executed in parallel. --- --- see https://www.fpcomplete.com/user/agocorona/publish-subscribe-variables-transient-effects-v --- - -newEVar :: TransIO (EVar a) -newEVar = Transient $ do - ref <-liftIO newBroadcastTChanIO - return . Just $ EVar ref - --- | delete al the subscriptions for an evar. -cleanEVar :: EVar a -> TransIO () -cleanEVar (EVar ref1)= liftIO $ atomically $ writeTChan ref1 SDone - - --- | read the EVar. It only succeed when the EVar is being updated --- The continuation gets registered to be executed whenever the variable is updated. --- --- if readEVar is re-executed in any kind of loop, since each continuation is different, this will register --- again. The effect is that the continuation will be executed multiple times --- To avoid multiple registrations, use `cleanEVar` -readEVar :: EVar a -> TransIO a -readEVar (EVar ref1)= do - tchan <- liftIO . atomically $ dupTChan ref1 - r <- parallel $ atomically $ readTChan tchan - - case r of - SDone -> empty - SMore x -> return x - SLast x -> return x - SError e -> empty --- error $ "readEVar: "++ show e - --- | update the EVar and execute all readEVar blocks with "last in-first out" priority --- -writeEVar (EVar ref1) x= liftIO $ atomically $ do - writeTChan ref1 $ SMore x - - --- | write the EVar and drop all the `readEVar` handlers. --- --- It is like a combination of `writeEVar` and `cleanEVar` -lastWriteEVar (EVar ref1) x= liftIO $ atomically $ do - writeTChan ref1 $ SLast x - - - +{-# LANGUAGE DeriveDataTypeable #-}+module Transient.EVars where++import Transient.Base+import Transient.Internals(runTransState,onNothing, EventF(..), killChildren)+import qualified Data.Map as M+import Data.Typeable++import Control.Concurrent+import Control.Applicative+import Control.Concurrent.STM+import Control.Monad.IO.Class+import Control.Exception(SomeException)++import Data.List(nub)+import Control.Monad.State+import Data.IORef++++data EVar a= EVar (TChan (StreamData a)) deriving Typeable+++-- | creates an EVar.+--+-- Evars are event vars. `writeEVar` trigger the execution of all the continuations associated to the `readEVar` of this variable+-- (the code that is after them).+--+-- It is like the publish-subscribe pattern but without inversion of control, since a readEVar can be inserted at any place in the+-- Transient flow.+--+-- EVars are created upstream and can be used to communicate two sub-threads of the monad. Following the Transient philosophy they+-- do not block his own thread if used with alternative operators, unlike the IORefs and TVars. And unlike STM vars, that are composable,+-- they wait for their respective events, while TVars execute the whole expression when any variable is modified.+--+-- The execution continues after the writeEVar when all subscribers have been executed.+--+-- Now the continuations are executed in parallel.+--+-- see https://www.fpcomplete.com/user/agocorona/publish-subscribe-variables-transient-effects-v+--++newEVar :: TransIO (EVar a)+newEVar = Transient $ do+ ref <-liftIO newBroadcastTChanIO+ return . Just $ EVar ref++-- | delete al the subscriptions for an evar.+cleanEVar :: EVar a -> TransIO ()+cleanEVar (EVar ref1)= liftIO $ atomically $ writeTChan ref1 SDone+++-- | read the EVar. It only succeed when the EVar is being updated+-- The continuation gets registered to be executed whenever the variable is updated.+--+-- if readEVar is re-executed in any kind of loop, since each continuation is different, this will register+-- again. The effect is that the continuation will be executed multiple times+-- To avoid multiple registrations, use `cleanEVar`+readEVar :: EVar a -> TransIO a+readEVar (EVar ref1)= do+ tchan <- liftIO . atomically $ dupTChan ref1+ r <- parallel $ atomically $ readTChan tchan++ case r of+ SDone -> empty+ SMore x -> return x+ SLast x -> return x+ SError e -> empty+-- error $ "readEVar: "++ show e++-- | update the EVar and execute all readEVar blocks with "last in-first out" priority+--+writeEVar (EVar ref1) x= liftIO $ atomically $ do+ writeTChan ref1 $ SMore x+++-- | write the EVar and drop all the `readEVar` handlers.+--+-- It is like a combination of `writeEVar` and `cleanEVar`+lastWriteEVar (EVar ref1) x= liftIO $ atomically $ do+ writeTChan ref1 $ SLast x+++
src/Transient/Indeterminism.hs view
@@ -1,153 +1,153 @@------------------------------------------------------------------------------ --- --- Module : Transient.Indeterminism --- Copyright : --- License : MIT --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | see <https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii> --- ------------------------------------------------------------------------------ -{-# LANGUAGE ScopedTypeVariables #-} -module Transient.Indeterminism ( -choose, choose', collect, collect', group, groupByTime -) where - -import Transient.Internals hiding (retry) - -import Data.IORef -import Control.Applicative -import Data.Monoid -import Control.Concurrent -import Data.Typeable -import Control.Monad.State -import GHC.Conc -import Data.Time.Clock -import Control.Exception - - --- | Converts a list of pure values into a transient task set. You can use the --- 'threads' primitive to control the parallelism. --- -choose :: Show a => [a] -> TransIO a -choose []= empty -choose xs = do - evs <- liftIO $ newIORef xs - r <- parallel $ do - es <- atomicModifyIORef' evs $ \es -> let tes= tail es in (tes,es) - case es of - [x] -> x `seq` return $ SLast x - x:_ -> x `seq` return $ SMore x - checkFinalize r - --- | Same as 'choose' except that the 'threads' combinator cannot be used, --- instead the parent thread's limit applies. --- -choose' :: [a] -> TransIO a -choose' xs = foldl (<|>) empty $ map (async . return) xs - - --- | Collect the results of a task set in groups of @n@ elements. --- -group :: Int -> TransIO a -> TransIO [a] -group num proc = do - v <- liftIO $ newIORef (0,[]) - x <- proc - - mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> - let n'=n +1 - in if n'== num - - then ((0,[]), Just $ x:xs) - else ((n', x:xs),Nothing) - case mn of - Nothing -> stop - Just xs -> return xs - --- | Collect the results of a task set, grouping all results received within --- every time interval specified by the first parameter as `diffUTCTime`. --- -groupByTime :: Integer -> TransIO a -> TransIO [a] - -groupByTime time proc = do - v <- liftIO $ newIORef (0,[]) - t <- liftIO getCurrentTime - x <- proc - t' <- liftIO getCurrentTime - mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> let n'=n +1 - in - if diffUTCTime t' t < fromIntegral time - then ((n', x:xs),Nothing) - else ((0,[]), Just $ x:xs) - case mn of - Nothing -> stop - Just xs -> return xs - - - --- XXX Collect might return before collecting @n@ results if the runtime --- detects that there is absolutely no possibility of more tasks to be --- collected (using 'BlockedIndefinitelyOnMVar'). This is inconsistent with the --- behavior of group. We should perhaps return 'stop' in the failure case to --- keep it consistent with group. --- --- | Collect the results of the first @n@ tasks. Synchronizes concurrent tasks --- to collect the results safely and kills all the non-free threads before --- returning the results. Results are returned in the thread where 'collect' --- is called. --- -collect :: Int -> TransIO a -> TransIO [a] -collect n = collect' n 0 - --- | Like 'collect' but with a timeout. When the timeout is zero it behaves --- exactly like 'collect'. If the timeout (second parameter) is non-zero, --- collection stops after the timeout and the results collected till now are --- returned. --- -collect' :: Int -> Int -> TransIO a -> TransIO [a] -collect' n t search= do - - rv <- liftIO $ newEmptyMVar -- !> "NEWMVAR" - - results <- liftIO $ newIORef (0,[]) - - let worker = do - r <- abduce >> search - liftIO $ putMVar rv $ Just r - stop - - timer= do - when (t > 0) . async $ threadDelay t >>putMVar rv Nothing -- readIORef results >>= return . snd - empty - - monitor= liftIO loop - - where - loop = do - mr <- takeMVar rv - (n',rs) <- readIORef results - case mr of - Nothing -> return rs - Just r -> do - let n''= n' +1 - let rs'= r:rs - writeIORef results (n'',rs') - - t' <- getCurrentTime - if (n > 0 && n'' >= n) - then return (rs') - else loop - `catch` \(e :: BlockedIndefinitelyOnMVar) -> - readIORef results >>= return . snd - - - r <- oneThread $ worker <|> timer <|> monitor - - return r - - - - +-----------------------------------------------------------------------------+--+-- Module : Transient.Indeterminism+-- Copyright :+-- License : MIT+--+-- Maintainer : agocorona@gmail.com+-- Stability :+-- Portability :+--+-- | see <https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii>+--+-----------------------------------------------------------------------------+{-# LANGUAGE ScopedTypeVariables #-}+module Transient.Indeterminism (+choose, choose', collect, collect', group, groupByTime+) where++import Transient.Internals hiding (retry)++import Data.IORef+import Control.Applicative+import Data.Monoid+import Control.Concurrent+import Data.Typeable+import Control.Monad.State+import GHC.Conc+import Data.Time.Clock+import Control.Exception+++-- | Converts a list of pure values into a transient task set. You can use the+-- 'threads' primitive to control the parallelism.+--+choose :: Show a => [a] -> TransIO a+choose []= empty+choose xs = do+ evs <- liftIO $ newIORef xs+ r <- parallel $ do+ es <- atomicModifyIORef' evs $ \es -> let tes= tail es in (tes,es)+ case es of+ [x] -> x `seq` return $ SLast x+ x:_ -> x `seq` return $ SMore x+ checkFinalize r++-- | Same as 'choose' except that the 'threads' combinator cannot be used,+-- instead the parent thread's limit applies.+--+choose' :: [a] -> TransIO a+choose' xs = foldl (<|>) empty $ map (async . return) xs+++-- | Collect the results of a task set in groups of @n@ elements.+--+group :: Int -> TransIO a -> TransIO [a]+group num proc = do+ v <- liftIO $ newIORef (0,[])+ x <- proc++ mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) ->+ let n'=n +1+ in if n'== num++ then ((0,[]), Just $ x:xs)+ else ((n', x:xs),Nothing)+ case mn of+ Nothing -> stop+ Just xs -> return xs++-- | Collect the results of a task set, grouping all results received within+-- every time interval specified by the first parameter as `diffUTCTime`.+--+groupByTime :: Integer -> TransIO a -> TransIO [a]++groupByTime time proc = do+ v <- liftIO $ newIORef (0,[])+ t <- liftIO getCurrentTime+ x <- proc+ t' <- liftIO getCurrentTime+ mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> let n'=n +1+ in+ if diffUTCTime t' t < fromIntegral time+ then ((n', x:xs),Nothing)+ else ((0,[]), Just $ x:xs)+ case mn of+ Nothing -> stop+ Just xs -> return xs++++-- XXX Collect might return before collecting @n@ results if the runtime+-- detects that there is absolutely no possibility of more tasks to be+-- collected (using 'BlockedIndefinitelyOnMVar'). This is inconsistent with the+-- behavior of group. We should perhaps return 'stop' in the failure case to+-- keep it consistent with group.+--+-- | Collect the results of the first @n@ tasks. Synchronizes concurrent tasks+-- to collect the results safely and kills all the non-free threads before+-- returning the results. Results are returned in the thread where 'collect'+-- is called.+--+collect :: Int -> TransIO a -> TransIO [a]+collect n = collect' n 0++-- | Like 'collect' but with a timeout. When the timeout is zero it behaves+-- exactly like 'collect'. If the timeout (second parameter) is non-zero,+-- collection stops after the timeout and the results collected till now are+-- returned.+--+collect' :: Int -> Int -> TransIO a -> TransIO [a]+collect' n t search= do++ rv <- liftIO $ newEmptyMVar -- !> "NEWMVAR"++ results <- liftIO $ newIORef (0,[])++ let worker = do+ r <- abduce >> search+ liftIO $ putMVar rv $ Just r+ stop++ timer= do+ when (t > 0) . async $ threadDelay t >>putMVar rv Nothing -- readIORef results >>= return . snd+ empty++ monitor= liftIO loop++ where+ loop = do+ mr <- takeMVar rv+ (n',rs) <- readIORef results+ case mr of+ Nothing -> return rs+ Just r -> do+ let n''= n' +1+ let rs'= r:rs+ writeIORef results (n'',rs')++ t' <- getCurrentTime+ if (n > 0 && n'' >= n)+ then return (rs')+ else loop+ `catch` \(e :: BlockedIndefinitelyOnMVar) ->+ readIORef results >>= return . snd+++ r <- oneThread $ worker <|> timer <|> monitor++ return r++++
src/Transient/Internals.hs view
@@ -1,1634 +1,1622 @@------------------------------------------------------------------------------ --- --- Module : Base --- Copyright : --- License : MIT --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | See http://github.com/agocorona/transient --- Everything in this module is exported in order to allow extensibility. ------------------------------------------------------------------------------ -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE UndecidableInstances #-} -{-# LANGUAGE Rank2Types #-} -{-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE CPP #-} -{-# LANGUAGE InstanceSigs #-} -{-# LANGUAGE ConstraintKinds #-} - -module Transient.Internals where - -import Control.Applicative -import Control.Monad.State -import Data.Dynamic -import qualified Data.Map as M -import System.IO.Unsafe -import Unsafe.Coerce -import Control.Exception hiding (try,onException) -import qualified Control.Exception (try) -import Control.Concurrent -import GHC.Conc(unsafeIOToSTM) -import Control.Concurrent.STM hiding (retry) -import qualified Control.Concurrent.STM as STM (retry) -import System.Mem.StableName -import Data.Maybe - -import Data.List -import Data.IORef -import System.Environment -import System.IO - -import qualified Data.ByteString.Char8 as BS - -#ifdef DEBUG - -import Data.Monoid -import Debug.Trace -import System.Exit - -{-# INLINE (!>) #-} -(!>) :: Show a => b -> a -> b -(!>) x y = trace (show y) x -infixr 0 !> - -#else - -{-# INLINE (!>) #-} -(!>) :: a -> b -> a -(!>) = const - -#endif - -type StateIO = StateT EventF IO - -newtype TransIO a = Transient { runTrans :: StateIO (Maybe a) } - -type SData = () - -type EventId = Int - -type TransientIO = TransIO - -data LifeCycle = Alive | Parent | Listener | Dead - deriving (Eq, Show) - --- | EventF describes the context of a TransientIO computation: -data EventF = forall a b. EventF - { meffects :: () - - , event :: Maybe SData - -- ^ Not yet consumed result (event) from the last asynchronous run of the - -- computation - - , xcomp :: TransIO a - , fcomp :: [b -> TransIO b] - -- ^ List of continuations - - , mfData :: M.Map TypeRep SData - -- ^ State data accessed with get or put operations - - , mfSequence :: Int - , threadId :: ThreadId - , freeTh :: Bool - -- ^ When 'True', threads are not killed using kill primitives - - , parent :: Maybe EventF - -- ^ The parent of this thread - - , children :: MVar [EventF] - -- ^ Forked child threads, used only when 'freeTh' is 'False' - - , maxThread :: Maybe (IORef Int) - -- ^ Maximum number of threads that are allowed to be created - - , labelth :: IORef (LifeCycle, BS.ByteString) - -- ^ Label the thread with its lifecycle state and a label string - } deriving Typeable - -type Effects = - forall a b c. TransIO a -> TransIO a -> (a -> TransIO b) - -> StateIO (StateIO (Maybe c) -> StateIO (Maybe c), Maybe a) - -instance MonadState EventF TransIO where - get = Transient $ get >>= return . Just - put x = Transient $ put x >> return (Just ()) - state f = Transient $ do - s <- get - let ~(a, s') = f s - put s' - return $ Just a - --- | Run a "non transient" computation within the underlying state monad, so it is --- guaranteed that the computation neither can stop neither can trigger additional --- events/threads. -noTrans :: StateIO x -> TransIO x -noTrans x = Transient $ x >>= return . Just - -emptyEventF :: ThreadId -> IORef (LifeCycle, BS.ByteString) -> MVar [EventF] -> EventF -emptyEventF th label childs = - EventF { meffects = mempty - , event = mempty - , xcomp = empty - , fcomp = [] - , mfData = mempty - , mfSequence = 0 - , threadId = th - , freeTh = False - , parent = Nothing - , children = childs - , maxThread = Nothing - , labelth = label } - --- | Run a transient computation with a default initial state -runTransient :: TransIO a -> IO (Maybe a, EventF) -runTransient t = do - th <- myThreadId - label <- newIORef $ (Alive, BS.pack "top") - childs <- newMVar [] - runStateT (runTrans t) $ emptyEventF th label childs - --- | Run a transient computation with a given initial state -runTransState :: EventF -> TransIO x -> IO (Maybe x, EventF) -runTransState st x = runStateT (runTrans x) st - --- | Get the continuation context: closure, continuation, state, child threads etc -getCont :: TransIO EventF -getCont = Transient $ Just <$> get - --- | Run the closure and the continuation using the state data of the calling thread -runCont :: EventF -> StateIO (Maybe a) -runCont EventF { xcomp = x, fcomp = fs } = runTrans $ do - r <- unsafeCoerce x - compose fs r - --- | Run the closure and the continuation using its own state data. -runCont' :: EventF -> IO (Maybe a, EventF) -runCont' cont = runStateT (runCont cont) cont - --- | Warning: Radically untyped stuff. handle with care -getContinuations :: StateIO [a -> TransIO b] -getContinuations = do - EventF { fcomp = fs } <- get - return $ unsafeCoerce fs - -{- -runCont cont = do - mr <- runClosure cont - case mr of - Nothing -> return Nothing - Just r -> runContinuation cont r --} - --- | Compose a list of continuations. -compose :: [a -> TransIO a] -> (a -> TransIO b) -compose [] = const empty -compose (f:fs) = \x -> f x >>= compose fs - --- | Run the closure (the 'x' in 'x >>= f') of the current bind operation. -runClosure :: EventF -> StateIO (Maybe a) -runClosure EventF { xcomp = x } = unsafeCoerce (runTrans x) - --- | Run the continuation (the 'f' in 'x >>= f') of the current bind operation with the current state. -runContinuation :: EventF -> a -> StateIO (Maybe b) -runContinuation EventF { fcomp = fs } = - runTrans . (unsafeCoerce $ compose $ fs) - --- | Save a closure and a continuation ('x' and 'f' in 'x >>= f'). -setContinuation :: TransIO a -> (a -> TransIO b) -> [c -> TransIO c] -> StateIO () -setContinuation b c fs = do - modify $ \EventF{..} -> EventF { xcomp = b - , fcomp = unsafeCoerce c : fs - , .. } - --- | Save a closure and continuation, run the closure, restore the old continuation. --- | NOTE: The old closure is discarded. -withContinuation :: b -> TransIO a -> TransIO a -withContinuation c mx = do - EventF { fcomp = fs, .. } <- get - put $ EventF { xcomp = mx - , fcomp = unsafeCoerce c : fs - , .. } - r <- mx - restoreStack fs - return r - --- | Restore the continuations to the provided ones. --- | NOTE: Events are also cleared out. -restoreStack :: MonadState EventF m => [a -> TransIO a] -> m () -restoreStack fs = modify $ \EventF {..} -> EventF { event = Nothing, fcomp = fs, .. } - --- | Run a chain of continuations. --- WARNING: It is up to the programmer to assure that each continuation typechecks --- with the next, and that the parameter type match the input of the first --- continuation. --- NOTE: Normally this makes sense to stop the current flow with `stop` after the --- invocation. -runContinuations :: [a -> TransIO b] -> c -> TransIO d -runContinuations fs x = compose (unsafeCoerce fs) x - --- Instances for Transient Monad - -instance Functor TransIO where - fmap f mx = do - x <- mx - return $ f x - -instance Applicative TransIO where - pure a = Transient . return $ Just a - - f <*> g = Transient $ do - rf <- liftIO $ newIORef (Nothing,[]) - rg <- liftIO $ newIORef (Nothing,[]) -#ifdef DEBUG - !> "NEWIOREF" -#endif - - fs <- getContinuations - - let hasWait (_:Wait:_) = True - hasWait _ = False - - appf k = Transient $ do - Log rec _ full <- getData `onNothing` return (Log False [] []) - (liftIO $ writeIORef rf (Just k,full)) -#ifdef DEBUG - !> ( show $ unsafePerformIO myThreadId) ++"APPF" -#endif - (x, full2)<- liftIO $ readIORef rg - when (hasWait full ) $ -#ifdef DEBUG - (!> (hasWait full,"full",full, "\nfull2",full2)) $ -#endif - let full'= head full: full2 - in (setData $ Log rec full' full') -#ifdef DEBUG - !> ("result1",full') -#endif - return $ Just k <*> x - - appg x = Transient $ do - Log rec _ full <- getData `onNothing` return (Log False [] []) - liftIO $ writeIORef rg (Just x, full) -#ifdef DEBUG - !> ( show $ unsafePerformIO myThreadId) ++ "APPG" -#endif - (k,full1) <- liftIO $ readIORef rf - when (hasWait full) $ -#ifdef DEBUG - (!> ("full", full, "\nfull1",full1)) $ -#endif - let full'= head full: full1 - in (setData $ Log rec full' full') -#ifdef DEBUG - !> ("result2",full') -#endif - return $ k <*> Just x - - setContinuation f appf fs - - k <- runTrans f -#ifdef DEBUG - !> ( show $ unsafePerformIO myThreadId)++ "RUN f" -#endif - was <- getData `onNothing` return NoRemote - when (was == WasParallel) $ setData NoRemote - - Log recovery _ full <- getData `onNothing` return (Log False [] []) - - - - if was== WasRemote || (not recovery && was == NoRemote && isNothing k ) -#ifdef DEBUG - !> ("was,recovery,isNothing=",was,recovery, isNothing k) -#endif - -- if the first operand was a remote request - -- (so this node is not master and hasn't to execute the whole expression) - -- or it was not an asyncronous term (a normal term without async or parallel - -- like primitives) and is nothing - then do - restoreStack fs - return Nothing - else do - when (isJust k) $ liftIO $ writeIORef rf (k,full) - -- when necessary since it maybe WasParallel and Nothing - - setContinuation g appg fs - - x <- runTrans g -#ifdef DEBUG - !> ( show $ unsafePerformIO myThreadId) ++ "RUN g" -#endif - Log recovery _ full' <- getData `onNothing` return (Log False [] []) - liftIO $ writeIORef rg (x,full') - restoreStack fs - k'' <- if was== WasParallel - then do - (k',_) <- liftIO $ readIORef rf -- since k may have been updated by a parallel f - return k' - else return k - return $ k'' <*> x - -instance Monad TransIO where - return = pure - x >>= f = Transient $ do - c <- setEventCont x f - mk <- runTrans x - resetEventCont mk c - case mk of - Just k -> runTrans (f k) - Nothing -> return Nothing - -instance MonadIO TransIO where - liftIO mx = do - ex <- liftIO' $ (mx >>= return . Right) `catch` - (\(e :: SomeException) -> return $ Left e) - case ex of - Left e -> back e -- finish $ Just e - Right x -> return x - where liftIO' x = Transient $ liftIO x >>= return . Just - -- let x= liftIO io in x `seq` lift x - -instance Monoid a => Monoid (TransIO a) where - mappend x y = mappend <$> x <*> y - mempty = return mempty - -instance Alternative TransIO where - empty = Transient $ return Nothing - (<|>) = mplus - -instance MonadPlus TransIO where - mzero = empty - mplus x y = Transient $ do - mx <- runTrans x -#ifdef DEBUG - !> "RUNTRANS11111" -#endif - was <- getData `onNothing` return NoRemote - if was == WasRemote -#ifdef DEBUG - !> was -#endif - then return Nothing - else case mx of - Nothing -> runTrans y -#ifdef DEBUG - !> "RUNTRANS22222" -#endif - justx -> return justx - -readWithErr :: (Typeable a, Read a) => String -> IO [(a, String)] -readWithErr line = - (v `seq` return [(v, left)]) - `catch` (\(e :: SomeException) -> - error $ "read error trying to read type: \"" ++ show (typeOf v) - ++ "\" in: " ++ " <" ++ show line ++ "> ") - where [(v, left)] = readsPrec 0 line - -readsPrec' _ = unsafePerformIO . readWithErr - --- | Constraint type synonym for a value that can be logged. -type Loggable a = (Show a, Read a, Typeable a) - --- | Dynamic serializable data for logging. -data IDynamic = - IDyns String - | forall a. Loggable a => IDynamic a - -instance Show IDynamic where - show (IDynamic x) = show (show x) - show (IDyns s) = show s - -instance Read IDynamic where - readsPrec n str = map (\(x,s) -> (IDyns x,s)) $ readsPrec' n str - -type Recover = Bool -type CurrentPointer = [LogElem] -type LogEntries = [LogElem] - -data LogElem = Wait | Exec | Var IDynamic - deriving (Read, Show) - -data Log = Log Recover CurrentPointer LogEntries - deriving (Typeable, Show) - -data RemoteStatus = WasRemote | WasParallel | NoRemote - deriving (Typeable, Eq, Show) - --- | A synonym of 'empty' that can be used in a monadic expression. It stops --- the computation, which allows the next computation in an 'Alternative' --- ('<|>') composition to run. -stop :: Alternative m => m stopped -stop = empty - ---instance (Num a,Eq a,Fractional a) =>Fractional (TransIO a)where --- mf / mg = (/) <$> mf <*> mg --- fromRational (x:%y) = fromInteger x % fromInteger y - - -instance (Num a, Eq a) => Num (TransIO a) where - fromInteger = return . fromInteger - mf + mg = (+) <$> mf <*> mg - mf * mg = (*) <$> mf <*> mg - negate f = f >>= return . negate - abs f = f >>= return . abs - signum f = f >>= return . signum - -class AdditionalOperators m where - - -- | Run @m a@ discarding its result before running @m b@. - (**>) :: m a -> m b -> m b - - -- | Run @m b@ discarding its result, after the whole task set @m a@ is - -- done. - (<**) :: m a -> m b -> m a - - atEnd' :: m a -> m b -> m a - atEnd' = (<**) - - -- | Run @m b@ discarding its result, once after each task in @m a@, and - -- once again after the whole task set is done. - (<***) :: m a -> m b -> m a - - atEnd :: m a -> m b -> m a - atEnd = (<***) - -instance AdditionalOperators TransIO where - - (**>) :: TransIO a -> TransIO b -> TransIO b - (**>) x y = - Transient $ do - runTrans x - runTrans y - - (<***) :: TransIO a -> TransIO b -> TransIO a - (<***) ma mb = - Transient $ do - fs <- getContinuations - setContinuation ma (\x -> mb >> return x) fs - a <- runTrans ma - runTrans mb - restoreStack fs - return a - - (<**) :: TransIO a -> TransIO b -> TransIO a - (<**) ma mb = - Transient $ do - a <- runTrans ma -#ifdef DEBUG - !> "ma" -#endif - - runTrans mb -#ifdef DEBUG - !> "mb" -#endif - return a - -infixr 1 <***, <**, **> - --- | Run @b@ once, discarding its result when the first task in task set @a@ --- has finished. Useful to start a singleton task after the first task has been --- setup. -(<|) :: TransIO a -> TransIO b -> TransIO a -(<|) ma mb = Transient $ do - fs <- getContinuations - ref <- liftIO $ newIORef False - setContinuation ma (cont ref) fs - r <- runTrans ma - restoreStack fs - return r - where cont ref x = Transient $ do - n <- liftIO $ readIORef ref - if n == True - then return $ Just x - else do liftIO $ writeIORef ref True - runTrans mb - return $ Just x - --- | Set the current closure and continuation for the current statement -setEventCont :: TransIO a -> (a -> TransIO b) -> StateIO EventF -setEventCont x f = do - EventF { fcomp = fs, .. } <- get -#ifdef DEBUG - !> "SET" -#endif - let cont = EventF { xcomp = x - , fcomp = unsafeCoerce f : fs - , .. } - put cont - return cont - --- | Reset the closure and continuation. Remove inner binds than the previous --- computations may have stacked in the list of continuations. --- resetEventCont :: Maybe a -> EventF -> StateIO (TransIO b -> TransIO b) -resetEventCont mx _ = do - EventF { fcomp = fs, .. } <- get -#ifdef DEBUG - !> "reset" -#endif - let f mx = case mx of - Nothing -> empty - Just x -> unsafeCoerce (head fs) x - put $ EventF { xcomp = f mx - , fcomp = tailsafe fs - , .. } - return id - --- | Total variant of `tail` that returns an empty list when given an empty list. -tailsafe :: [a] -> [a] -tailsafe [] = [] -tailsafe (x:xs) = xs - ---refEventCont= unsafePerformIO $ newIORef baseEffects - --- effects not used --- {-# INLINE baseEffects #-} ---baseEffects :: Effects --- ---baseEffects x x' f' = do --- c <- setEventCont x' f' --- mk <- runTrans x --- t <- resetEventCont mk c --- return (t,mk) - - ---instance MonadTrans (Transient ) where --- lift mx = Transient $ mx >>= return . Just - --- * Threads - -waitQSemB sem = atomicModifyIORef sem $ \n -> - if n > 0 then(n - 1, True) else (n, False) -signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ()) - --- | Sets the maximum number of threads that can be created for the given task --- set. When set to 0, new tasks start synchronously in the current thread. --- New threads are created by 'parallel', and APIs that use parallel. -threads :: Int -> TransIO a -> TransIO a -threads n process = do - msem <- gets maxThread - sem <- liftIO $ newIORef n - modify $ \s -> s { maxThread = Just sem } - r <- process <** (modify $ \s -> s { maxThread = msem }) -- restore it - return r - --- | Terminate all the child threads in the given task set and continue --- execution in the current thread. Useful to reap the children when a task is --- done. --- -oneThread :: TransIO a -> TransIO a -oneThread comp = do - st <- get - chs <- liftIO $ newMVar [] - label <- liftIO $ newIORef (Alive, BS.pack "oneThread") - let st' = st { parent = Just st - , children = chs - , labelth = label } - liftIO $ hangThread st st' - put st' - x <- comp - th <- liftIO myThreadId -#ifdef DEBUG - !> ("FATHER:", threadId st) -#endif - chs <- liftIO $ readMVar chs -- children st' - liftIO $ mapM_ (killChildren1 th) chs - return x - where killChildren1 :: ThreadId -> EventF -> IO () - killChildren1 th state = do - ths' <- modifyMVar (children state) $ \ths -> do - let (inn, ths')= partition (\st -> threadId st == th) ths - return (inn, ths') - mapM_ (killChildren1 th) ths' - mapM_ (killThread . threadId) ths' -#ifdef DEBUG - !> ("KILLEVENT1 ", map threadId ths' ) -#endif - --- | Add a label to the current passing threads so it can be printed by debugging calls like `showThreads` -labelState :: String -> TransIO () -labelState l = do - st <- get - liftIO $ atomicModifyIORef (labelth st) $ \(status,_) -> ((status, BS.pack l), ()) - -printBlock :: MVar () -printBlock = unsafePerformIO $ newMVar () - --- | Show the tree of threads hanging from the state. -showThreads :: MonadIO m => EventF -> m () -showThreads st = liftIO $ withMVar printBlock $ const $ do - mythread <- myThreadId -#ifdef DEBUG - !> "showThreads" -#endif - putStrLn "---------Threads-----------" - let showTree n ch = do - liftIO $ do - putStr $ take n $ repeat ' ' - (state, label) <- readIORef $ labelth ch - if BS.null label - then putStr . show $ threadId ch - else do BS.putStr label; putStr . drop 8 . show $ threadId ch - when (state == Dead) $ putStr " dead" - putStrLn $ if mythread == threadId ch then " <--" else "" - chs <- readMVar $ children ch - mapM_ (showTree $ n + 2) $ reverse chs - showTree 0 st - --- | Return the state of the thread that initiated the transient computation -topState :: TransIO EventF -topState = do - st <- get - return $ toplevel st - where toplevel st = case parent st of - Nothing -> st - Just p -> toplevel p - --- | Return the state variable of the type desired with which a thread, identified by his number in the treee was initiated -showState :: (Typeable a, MonadIO m, Alternative m) => String -> EventF -> m (Maybe a) -showState th top = resp - where resp = do - let thstring = drop 9 . show $ threadId top - if thstring == th - then getstate top - else do - sts <- liftIO $ readMVar $ children top - foldl (<|>) empty $ map (showState th) sts - getstate st = - case M.lookup (typeOf $ typeResp resp) $ mfData st of - Just x -> return . Just $ unsafeCoerce x - Nothing -> return Nothing - typeResp :: m (Maybe x) -> x - typeResp = undefined - --- | Add n threads to the limit of threads. If there is no limit, the limit is set. -addThreads' :: Int -> TransIO () -addThreads' n= noTrans $ do - msem <- gets maxThread - case msem of - Just sem -> liftIO $ modifyIORef sem $ \n' -> n + n' - Nothing -> do - sem <- liftIO (newIORef n) - modify $ \ s -> s { maxThread = Just sem } - --- | Ensure that at least n threads are available for the current task set. -addThreads :: Int -> TransIO () -addThreads n = noTrans $ do - msem <- gets maxThread - case msem of - Nothing -> return () - Just sem -> liftIO $ modifyIORef sem $ \n' -> if n' > n then n' else n - ---getNonUsedThreads :: TransIO (Maybe Int) ---getNonUsedThreads= Transient $ do --- msem <- gets maxThread --- case msem of --- Just sem -> liftIO $ Just <$> readIORef sem --- Nothing -> return Nothing - --- | Disable tracking and therefore the ability to terminate the child threads. --- By default, child threads are terminated automatically when the parent --- thread dies, or they can be terminated using the kill primitives. Disabling --- it may improve performance a bit, however, all threads must be well-behaved --- to exit on their own to avoid a leak. -freeThreads :: TransIO a -> TransIO a -freeThreads process = Transient $ do - st <- get - put st { freeTh = True } - r <- runTrans process - modify $ \s -> s { freeTh = freeTh st } - return r - --- | Enable tracking and therefore the ability to terminate the child threads. --- This is the default but can be used to re-enable tracking if it was --- previously disabled with 'freeThreads'. -hookedThreads :: TransIO a -> TransIO a -hookedThreads process = Transient $ do - st <- get - put st {freeTh = False} - r <- runTrans process - modify $ \st -> st { freeTh = freeTh st } - return r - --- | Kill all the child threads of the current thread. -killChilds :: TransIO () -killChilds = noTrans $ do - cont <- get - liftIO $ do - killChildren $ children cont - writeIORef (labelth cont) (Alive, mempty) -#ifdef DEBUG - !> (threadId cont,"relabeled") -#endif - return () - --- | Kill the current thread and the childs. -killBranch :: TransIO () -killBranch = noTrans $ do - st <- get - liftIO $ killBranch' st - --- | Kill the childs and the thread of an state -killBranch' :: EventF -> IO () -killBranch' cont = do - killChildren $ children cont - let thisth = threadId cont - mparent = parent cont - when (isJust mparent) $ - modifyMVar_ (children $ fromJust mparent) $ \sts -> - return $ filter (\st -> threadId st /= thisth) sts - killThread $ thisth - --- * Extensible State: Session Data Management - --- | Same as 'getSData' but with a more general type. If the data is found, a --- 'Just' value is returned. Otherwise, a 'Nothing' value is returned. -getData :: (MonadState EventF m, Typeable a) => m (Maybe a) -getData = resp - where resp = do - list <- gets mfData - case M.lookup (typeOf $ typeResp resp) list of - Just x -> return . Just $ unsafeCoerce x - Nothing -> return Nothing - typeResp :: m (Maybe x) -> x - typeResp = undefined - --- | Retrieve a previously stored data item of the given data type from the --- monad state. The data type to retrieve is implicitly determined from the --- requested type context. --- If the data item is not found, an 'empty' value (a void event) is returned. --- Remember that an empty value stops the monad computation. If you want to --- print an error message or a default value in that case, you can use an --- 'Alternative' composition. For example: --- --- > getSData <|> error "no data" --- > getInt = getSData <|> return (0 :: Int) -getSData :: Typeable a => TransIO a -getSData = Transient getData - --- | Same as `getSData` -getState :: Typeable a => TransIO a -getState = getSData - --- | 'setData' stores a data item in the monad state which can be retrieved --- later using 'getData' or 'getSData'. Stored data items are keyed by their --- data type, and therefore only one item of a given type can be stored. A --- newtype wrapper can be used to distinguish two data items of the same type. --- --- @ --- import Control.Monad.IO.Class (liftIO) --- import Transient.Base --- import Data.Typeable --- --- data Person = Person --- { name :: String --- , age :: Int --- } deriving Typeable --- --- main = keep $ do --- setData $ Person "Alberto" 55 --- Person name age <- getSData --- liftIO $ print (name, age) --- @ -setData :: (MonadState EventF m, Typeable a) => a -> m () -setData x = modify $ \st -> st { mfData = M.insert t (unsafeCoerce x) (mfData st) } - where t = typeOf x - --- | Accepts a function that takes the current value of the stored data type --- and returns the modified value. If the function returns 'Nothing' the value --- is deleted otherwise updated. -modifyData :: (MonadState EventF m, Typeable a) => (Maybe a -> Maybe a) -> m () -modifyData f = modify $ \st -> st { mfData = M.alter alterf t (mfData st) } - where typeResp :: (Maybe a -> b) -> a - typeResp = undefined - t = typeOf (typeResp f) - alterf mx = unsafeCoerce $ f x' - where x' = case mx of - Just x -> Just $ unsafeCoerce x - Nothing -> Nothing - --- | Same as modifyData -modifyState :: (MonadState EventF m, Typeable a) => (Maybe a -> Maybe a) -> m () -modifyState = modifyData - --- | Same as 'setData' -setState :: (MonadState EventF m, Typeable a) => a -> m () -setState = setData - --- | Delete the data item of the given type from the monad state. -delData :: (MonadState EventF m, Typeable a) => a -> m () -delData x = modify $ \st -> st { mfData = M.delete (typeOf x) (mfData st) } - --- | Same as 'delData' -delState :: (MonadState EventF m, Typeable a) => a -> m () -delState = delData - --- | Run an action, if the result is a void action undo any state changes --- that it might have caused. -try :: TransIO a -> TransIO a -try mx = do - sd <- gets mfData - mx <|> (modify (\s -> s { mfData = sd }) >> empty) - --- | Executes the computation and reset the state either if it fails or not -sandbox :: TransIO a -> TransIO a -sandbox mx = do - sd <- gets mfData - mx <*** modify (\s ->s { mfData = sd}) - --- | Generator of identifiers that are unique within the current monadic --- sequence They are not unique in the whole program. -genId :: MonadState EventF m => m Int -genId = do - st <- get - let n = mfSequence st - put st { mfSequence = n + 1 } - return n - -getPrevId :: MonadState EventF m => m Int -getPrevId = gets mfSequence - -instance Read SomeException where - readsPrec n str = [(SomeException $ ErrorCall s, r)] - where [(s , r)] = read str - --- | 'StreamData' represents a task in a task stream being generated. -data StreamData a = - SMore a -- ^ More tasks to come - | SLast a -- ^ This is the last task - | SDone -- ^ No more tasks, we are done - | SError SomeException -- ^ An error occurred - deriving (Typeable, Show,Read) - --- | An task stream generator that produces an infinite stream of tasks by --- running an IO computation in a loop. A task is triggered carrying the output --- of the computation. See 'parallel' for notes on the return value. -waitEvents :: IO a -> TransIO a -waitEvents io = do - mr <- parallel (SMore <$> io) - case mr of - SMore x -> return x - SError e -> back e - --- | Run an IO computation asynchronously and generate a single task carrying --- the result of the computation when it completes. See 'parallel' for notes on --- the return value. -async :: IO a -> TransIO a -async io = do - mr <- parallel (SLast <$> io) - case mr of - SLast x -> return x - SError e -> back e - --- | Force an async computation to run synchronously. It can be useful in an --- 'Alternative' composition to run the alternative only after finishing a --- computation. Note that in Applicatives it might result in an undesired --- serialization. -sync :: TransIO a -> TransIO a -sync x = do - setData WasRemote - r <- x - delData WasRemote - return r - --- | @spawn = freeThreads . waitEvents@ -spawn :: IO a -> TransIO a -spawn = freeThreads . waitEvents - --- | An task stream generator that produces an infinite stream of tasks by --- running an IO computation periodically at the specified time interval. The --- task carries the result of the computation. A new task is generated only if --- the output of the computation is different from the previous one. See --- 'parallel' for notes on the return value. -sample :: Eq a => IO a -> Int -> TransIO a -sample action interval = do - v <- liftIO action - prev <- liftIO $ newIORef v - waitEvents (loop action prev) <|> async (return v) - where loop action prev = loop' - where loop' = do - threadDelay interval - v <- action - v' <- readIORef prev - if v /= v' then writeIORef prev v >> return v else loop' - ---serial :: IO (StreamData b) -> TransIO (StreamData b) ---serial ioaction= Transient $ do --- cont <- get -- !> "PARALLEL" --- case event cont of --- j@(Just _) -> do --- put cont{event=Nothing} --- return $ unsafeCoerce j --- Nothing -> do --- liftIO $ loop cont ioaction --- return Nothing --- --- where loop cont ioaction= do --- let iocont dat= do --- runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat} --- return () --- mdat <- ioaction `catch` \(e :: SomeException) -> return $ SError e --- case mdat of --- se@(SError _) -> iocont se --- SDone -> iocont SDone --- last@(SLast _) -> iocont last --- --- more@(SMore _) -> do --- iocont more --- loop cont ioaction - --- | Run an IO action one or more times to generate a stream of tasks. The IO --- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new --- task is triggered with the result value. If the return value is 'SMore', the --- action is run again to generate the next task, otherwise task creation --- stops. --- --- Unless the maximum number of threads (set with 'threads') has been reached, --- the task is generated in a new thread and the current thread returns a void --- task. -parallel :: IO (StreamData b) -> TransIO (StreamData b) -parallel ioaction = Transient $ do - cont <- get -#ifdef DEBUG - !> "PARALLEL" -#endif - case event cont of - j@(Just _) -> do - put cont { event = Nothing } - return $ unsafeCoerce j - Nothing -> do - liftIO $ atomicModifyIORef (labelth cont) $ \(_, lab) -> ((Parent, lab), ()) - liftIO $ loop cont ioaction - was <- getData `onNothing` return NoRemote - when (was /= WasRemote) $ setData WasParallel --- th <- liftIO myThreadId --- return () !> ("finish",th) - return Nothing - --- | Execute the IO action and the continuation -loop :: EventF -> IO (StreamData t) -> IO () -loop parentc rec = forkMaybe parentc $ \cont -> do - -- Execute the IO computation and then the closure-continuation - liftIO $ atomicModifyIORef (labelth cont) $ const ((Listener,BS.pack "wait"),()) - let loop'= do - mdat <- rec `catch` \(e :: SomeException) -> return $ SError e - case mdat of - se@(SError _) -> setworker cont >> iocont se cont - SDone -> setworker cont >> iocont SDone cont - last@(SLast _) -> setworker cont >> iocont last cont - - more@(SMore _) -> do - forkMaybe cont $ iocont more - loop' - - where - setworker cont= liftIO $ atomicModifyIORef (labelth cont) $ const ((Alive,BS.pack "work"),()) - - iocont dat cont = do - - let cont'= cont{event= Just $ unsafeCoerce dat} - runStateT (runCont cont') cont' - return () - - - loop' - return () - where - - forkMaybe parent proc = do - case maxThread parent of - Nothing -> forkIt parent proc - Just sem -> do - dofork <- waitQSemB sem - if dofork then forkIt parent proc else proc parent - - - forkIt parent proc= do - chs <- liftIO $ newMVar [] - - label <- newIORef (Alive, BS.pack "work") - let cont = parent{parent=Just parent,children= chs, labelth= label} - - forkFinally1 (do - th <- myThreadId - let cont'= cont{threadId=th} - when(not $ freeTh parent )$ hangThread parent cont' - -- !> ("thread created: ",th,"in",threadId parent ) - - proc cont') - $ \me -> do - --- case me of -- !> "THREAD END" of --- Left e -> do ----- when (fromException e /= Just ThreadKilled)$ --- liftIO $ print e --- killChildren $ children cont ----- !> "KILL RECEIVED" ++ (show $ unsafePerformIO myThreadId) --- --- Right _ -> - - - - case maxThread cont of - Just sem -> signalQSemB sem -- !> "freed thread" - Nothing -> when(not $ freeTh parent ) $ do -- if was not a free thread - - th <- myThreadId - (can,label) <- atomicModifyIORef (labelth cont) $ \(l@(status,label)) -> - ((if status== Alive then Dead else status, label),l) - - - when (can/= Parent ) $ free th parent - return () - - - forkFinally1 :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId - forkFinally1 action and_then = - mask $ \restore -> forkIO $ Control.Exception.try (restore action) >>= and_then - -free th env= do --- return () !> ("freeing",th,"in",threadId env) - let sibling= children env - - (sbs',found) <- modifyMVar sibling $ \sbs -> do - let (sbs', found) = drop [] th sbs - return (sbs',(sbs',found)) - - - - if found - then do - --- !> ("new list for",threadId env,map threadId sbs') - (typ,_) <- readIORef $ labelth env - if (null sbs' && typ /= Listener && isJust (parent env)) - -- free the parent - then free (threadId env) ( fromJust $ parent env) - else return () - --- return env - else return () -- putMVar sibling sbs - -- !> (th,"orphan") - - where - drop processed th []= (processed,False) - drop processed th (ev:evts)| th == threadId ev= (processed ++ evts, True) - | otherwise= drop (ev:processed) th evts - - - -hangThread parentProc child = do - - let headpths= children parentProc - - modifyMVar_ headpths $ \ths -> return (child:ths) --- ths <- takeMVar headpths --- putMVar headpths (child:ths) - - - -- !> ("hang", threadId child, threadId parentProc,map threadId ths,unsafePerformIO $ readIORef $ labelth parentProc) - --- | kill all the child threads associated with the continuation context -killChildren childs = do - - - ths <- modifyMVar childs $ \ths -> return ([],ths) --- ths <- takeMVar childs --- putMVar childs [] - - mapM_ (killChildren . children) ths - - - mapM_ (killThread . threadId) ths -- !> ("KILL", map threadId ths ) - - - - - --- | Make a transient task generator from an asynchronous callback handler. --- --- The first parameter is a callback. The second parameter is a value to be --- returned to the callback; if the callback expects no return value it --- can just be a @return ()@. The callback expects a setter function taking the --- @eventdata@ as an argument and returning a value to the callback; this --- function is supplied by 'react'. --- --- Callbacks from foreign code can be wrapped into such a handler and hooked --- into the transient monad using 'react'. Every time the callback is called it --- generates a new task for the transient monad. --- -react - :: Typeable eventdata - => ((eventdata -> IO response) -> IO ()) - -> IO response - -> TransIO eventdata -react setHandler iob= Transient $ do - cont <- get - case event cont of - Nothing -> do - liftIO $ setHandler $ \dat ->do - runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat} - iob - was <- getData `onNothing` return NoRemote - when (was /= WasRemote) $ setData WasParallel - return Nothing - - j@(Just _) -> do - put cont{event=Nothing} - return $ unsafeCoerce j - --- | Runs a computation asynchronously without generating any events. Returns --- 'empty' in an 'Alternative' composition. - -abduce = Transient $ do - st <- get - case event st of - Just _ -> do - put st{event=Nothing} - return $ Just () - Nothing -> do - chs <- liftIO $ newMVar [] - - label <- liftIO $ newIORef (Alive, BS.pack "abduce") - liftIO $ forkIO $ do - th <- myThreadId - let st' = st{event= Just (),parent=Just st,children= chs, threadId=th,labelth= label} - liftIO $ hangThread st st' - - runCont' st' - return() - return Nothing - - --- * non-blocking keyboard input - -getLineRef= unsafePerformIO $ newTVarIO Nothing - - -roption= unsafePerformIO $ newMVar [] - --- | Waits on stdin in a loop and triggers a new task every time the input data --- matches the first parameter. The value contained by the task is the matched --- value i.e. the first argument itself. The second parameter is a label for --- the option. The label is displayed on the console when the option is --- activated. --- --- Note that if two independent invocations of 'option' are expecting the same --- input, only one of them gets it and triggers a task. It cannot be --- predicted which one gets it. --- -option :: (Typeable b, Show b, Read b, Eq b) => - b -> String -> TransIO b -option ret message= do - let sret= show ret - - liftIO $ putStrLn $ "Enter "++sret++"\tto: " ++ message - liftIO $ modifyMVar_ roption $ \msgs-> return $ sret:msgs - waitEvents $ getLine' (==ret) - liftIO $ putStr "\noption: " >> putStrLn (show ret) - return ret - - --- | Waits on stdin and triggers a task when a console input matches the --- predicate specified in the first argument. The second parameter is a string --- to be displayed on the console before waiting. --- -input :: (Typeable a, Read a,Show a) => (a -> Bool) -> String -> TransIO a -input cond prompt= Transient . liftIO $do - putStr prompt >> hFlush stdout - atomically $ do - mr <- readTVar getLineRef - case mr of - Nothing -> STM.retry - Just r -> - case reads1 r of - (s,_):_ -> if cond s -- !> show (cond s) - then do - unsafeIOToSTM $ print s - writeTVar getLineRef Nothing -- !>"match" - return $ Just s - - else return Nothing - _ -> return Nothing - --- | Non blocking `getLine` with a validator -getLine' cond= do - atomically $ do - mr <- readTVar getLineRef - case mr of - Nothing -> STM.retry - Just r -> - case reads1 r of -- !> ("received " ++ show r ++ show (unsafePerformIO myThreadId)) of - (s,_):_ -> if cond s -- !> show (cond s) - then do - writeTVar getLineRef Nothing -- !>"match" - return s - - else STM.retry - _ -> STM.retry - -reads1 s=x where - x= if typeOf(typeOfr x) == typeOf "" then unsafeCoerce[(s,"")] else readsPrec' 0 s - typeOfr :: [(a,String)] -> a - typeOfr = undefined - - -inputLoop= do - r<- getLine - -- XXX hoping that the previous value has been consumed by now. - -- otherwise its just lost by overwriting. - atomically $ writeTVar getLineRef Nothing - processLine r - inputLoop - -processLine r= do - - let rs = breakSlash [] r - - -- XXX this blocks forever if an input is not consumed by any consumer. - -- e.g. try this "xxx/xxx" on the stdin - liftIO $ mapM_ (\ r -> - atomically $ do --- threadDelay 1000000 - t <- readTVar getLineRef - when (isJust t) STM.retry - writeTVar getLineRef $ Just r ) rs - - - where - breakSlash :: [String] -> String -> [String] - breakSlash [] ""= [""] - breakSlash s ""= s - breakSlash res ('\"':s)= - let (r,rest) = span(/= '\"') s - in breakSlash (res++[r]) $ tail1 rest - - breakSlash res s= - let (r,rest) = span(/= '/') s - in breakSlash (res++[r]) $ tail1 rest - - tail1 []=[] - tail1 x= tail x - - - - --- | Wait for the execution of `exit` and return the result or the exhaustion of thread activity - -stay rexit= takeMVar rexit - `catch` \(e :: BlockedIndefinitelyOnMVar) -> return Nothing - -newtype Exit a= Exit a deriving Typeable - --- | Runs the transient computation in a child thread and keeps the main thread --- running until all the user threads exit or some thread invokes 'exit'. --- --- The main thread provides facilities to accept keyboard input in a --- non-blocking but line-oriented manner. The program reads the standard input --- and feeds it to all the async input consumers (e.g. 'option' and 'input'). --- All async input consumers contend for each line entered on the standard --- input and try to read it atomically. When a consumer consumes the input --- others do not get to see it, otherwise it is left in the buffer for others --- to consume. If nobody consumes the input, it is discarded. --- --- A @/@ in the input line is treated as a newline. --- --- When using asynchronous input, regular synchronous IO APIs like getLine --- cannot be used as they will contend for the standard input along with the --- asynchronous input thread. Instead you can use the asynchronous input APIs --- provided by transient. --- --- A built-in interactive command handler also reads the stdin asynchronously. --- All available commands handled by the command handler are displayed when the --- program is run. The following commands are available: --- --- 1. @ps@: show threads --- 2. @log@: inspect the log of a thread --- 3. @end@, @exit@: terminate the program --- --- An input not handled by the command handler can be handled by the program. --- --- The program's command line is scanned for @-p@ or @--path@ command line --- options. The arguments to these options are injected into the async input --- channel as keyboard input to the program. Each line of input is separated by --- a @/@. For example: --- --- > foo -p ps/end --- -keep :: Typeable a => TransIO a -> IO (Maybe a) -keep mx = do - - liftIO $ hSetBuffering stdout LineBuffering - rexit <- newEmptyMVar - forkIO $ do --- liftIO $ putMVar rexit $ Right Nothing - runTransient $ do - st <- get - - setData $ Exit rexit - (async (return ()) >> labelState "input" >> liftIO inputLoop) - - <|> do - option "ps" "show threads" - liftIO $ showThreads st - empty - <|> do - option "log" "inspect the log of a thread" - th <- input (const True) "thread number>" - ml <- liftIO $ showState th st - liftIO $ print $ fmap (\(Log _ _ log) -> reverse log) ml - empty - <|> do - option "end" "exit" - killChilds - liftIO $ putMVar rexit Nothing - empty - <|> mx - return () - threadDelay 10000 - execCommandLine - stay rexit - - where - type1 :: TransIO a -> Either String (Maybe a) - type1= undefined - --- | Same as `keep` but does not read from the standard input, and therefore --- the async input APIs ('option' and 'input') cannot be used in the monad. --- However, keyboard input can still be passed via command line arguments as --- described in 'keep'. Useful for debugging or for creating background tasks, --- as well as to embed the Transient monad inside another computation. It --- returns either the value returned by `exit`. or Nothing, when there are no --- more threads running --- -keep' :: Typeable a => TransIO a -> IO (Maybe a) -keep' mx = do - liftIO $ hSetBuffering stdout LineBuffering - rexit <- newEmptyMVar - forkIO $ do - runTransient $ do - setData $ Exit rexit - mx - - return () - threadDelay 10000 - forkIO $ execCommandLine - stay rexit - - -execCommandLine= do - args <- getArgs - let mindex = findIndex (\o -> o == "-p" || o == "--path" ) args - when (isJust mindex) $ do - let i= fromJust mindex +1 - when (length args >= i) $ do - let path= args !! i - putStr "Executing: " >> print path - processLine path - --- | Exit the main thread, and thus all the Transient threads (and the --- application if there is no more code) -exit :: Typeable a => a -> TransIO a -exit x= do - Exit rexit <- getSData <|> error "exit: not the type expected" `asTypeOf` type1 x - liftIO $ putMVar rexit $ Just x - stop - where - type1 :: a -> TransIO (Exit (MVar (Maybe a))) - type1= undefined - - - --- | If the first parameter is 'Nothing' return the second parameter otherwise --- return the first parameter.. -onNothing :: Monad m => m (Maybe b) -> m b -> m b -onNothing iox iox'= do - mx <- iox - case mx of - Just x -> return x - Nothing -> iox' - - - - - -----------------------------------backtracking ------------------------ - - -data Backtrack b= Show b =>Backtrack{backtracking :: Maybe b - ,backStack :: [EventF] } - deriving Typeable - - - --- | Delete all the undo actions registered till now for the given track id. -backCut :: (Typeable b, Show b) => b -> TransientIO () -backCut reason= Transient $ do - delData $ Backtrack (Just reason) [] - return $ Just () - --- | 'backCut' for the default track; equivalent to @backCut ()@. -undoCut :: TransientIO () -undoCut = backCut () - --- | Run the action in the first parameter and register the second parameter as --- the undo action. On undo ('back') the second parameter is called with the --- undo track id as argument. --- -{-# NOINLINE onBack #-} -onBack :: (Typeable b, Show b) => TransientIO a -> ( b -> TransientIO a) -> TransientIO a -onBack ac bac = registerBack (typeof bac) $ Transient $ do - Backtrack mreason _ <- getData `onNothing` backStateOf (typeof bac) - runTrans $ case mreason of - Nothing -> ac - Just reason -> bac reason - where - typeof :: (b -> TransIO a) -> b - typeof = undefined - --- | 'onBack' for the default track; equivalent to @onBack ()@. -onUndo :: TransientIO a -> TransientIO a -> TransientIO a -onUndo x y= onBack x (\() -> y) - - --- | Register an undo action to be executed when backtracking. The first --- parameter is a "witness" whose data type is used to uniquely identify this --- backtracking action. The value of the witness parameter is not used. --- -{-# NOINLINE registerUndo #-} -registerBack :: (Typeable b, Show b) => b -> TransientIO a -> TransientIO a -registerBack witness f = Transient $ do - cont@(EventF _ _ x _ _ _ _ _ _ _ _ _) <- get -- !!> "backregister" - - md <- getData `asTypeOf` (Just <$> backStateOf witness) - - case md of - Just (bss@(Backtrack b (bs@((EventF _ _ x' _ _ _ _ _ _ _ _ _):_)))) -> - when (isNothing b) $ do - addrx <- addr x - addrx' <- addr x' -- to avoid duplicate backtracking points - setData $ if addrx == addrx' then bss else Backtrack mwit (cont:bs) - Nothing -> setData $ Backtrack mwit [cont] - - runTrans f - where - mwit= Nothing `asTypeOf` (Just witness) - addr x = liftIO $ return . hashStableName =<< (makeStableName $! x) - - -registerUndo :: TransientIO a -> TransientIO a -registerUndo f= registerBack () f - --- XXX Should we enforce retry of the same track which is being undone? If the --- user specifies a different track would it make sense? --- --- | For a given undo track id, stop executing more backtracking actions and --- resume normal execution in the forward direction. Used inside an undo --- action. --- -forward :: (Typeable b, Show b) => b -> TransIO () -forward reason= Transient $ do - Backtrack _ stack <- getData `onNothing` (backStateOf reason) - setData $ Backtrack(Nothing `asTypeOf` Just reason) stack - return $ Just () - --- | 'forward' for the default undo track; equivalent to @forward ()@. -retry= forward () - --- | Abort finish. Stop executing more finish actions and resume normal --- execution. Used inside 'onFinish' actions. --- -noFinish= forward (FinishReason Nothing) - --- | Start the undo process for the given undo track id. Performs all the undo --- actions registered till now in reverse order. An undo action can use --- 'forward' to stop the undo process and resume forward execution. If there --- are no more undo actions registered execution stops and a 'stop' action is --- returned. --- -back :: (Typeable b, Show b) => b -> TransientIO a -back reason = Transient $ do - bs <- getData `onNothing` backStateOf reason -- !!>"GOBACK" - goBackt bs - - where - - goBackt (Backtrack _ [] )= return Nothing -- !!> "END" - goBackt (Backtrack b (stack@(first : bs)) )= do - (setData $ Backtrack (Just reason) stack) - - mr <- runClosure first -- !> "RUNCLOSURE" - - Backtrack back _ <- getData `onNothing` backStateOf reason - -- !> "END RUNCLOSURE" --- case back of --- Nothing -> case mr of --- Nothing -> return empty -- !> "FORWARD END" --- Just x -> runContinuation first x -- !> "FORWARD EXEC" --- justreason -> goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back) - - case mr of - Nothing -> return empty -- !> "END EXECUTION" - Just x -> case back of - Nothing -> runContinuation first x -- !> "FORWARD EXEC" - justreason -> goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back) - -backStateOf :: (Monad m, Show a, Typeable a) => a -> m (Backtrack a) -backStateOf reason= return $ Backtrack (Nothing `asTypeOf` (Just reason)) [] - --- | 'back' for the default undo track; equivalent to @back ()@. --- -undo :: TransIO a -undo= back () - - ------- finalization - -newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show) - --- | Clear all finish actions registered till now. -initFinish= backCut (FinishReason Nothing) - --- | Register an action that to be run when 'finish' is called. 'onFinish' can --- be used multiple times to register multiple actions. Actions are run in --- reverse order. Used in infix style. --- -onFinish :: ((Maybe SomeException) ->TransIO ()) -> TransIO () -onFinish f= onFinish' (return ()) f - - --- | Run the action specified in the first parameter and register the second --- parameter as a finish action to be run when 'finish' is called. Used in --- infix style. --- -onFinish' ::TransIO a ->((Maybe SomeException) ->TransIO a) -> TransIO a -onFinish' proc f= proc `onBack` \(FinishReason reason) -> - f reason - - --- | Execute all the finalization actions registered up to the last --- 'initFinish', in reverse order. Either an exception or 'Nothing' can be --- passed to 'finish'. The argument passed is made available in the 'onFinish' --- actions invoked. --- -finish :: Maybe SomeException -> TransIO a -finish reason= back (FinishReason reason) - - - --- | trigger finish when the stream of data ends -checkFinalize v= - case v of - SDone -> stop - SLast x -> return x - SError e -> back e - SMore x -> return x - ------- exceptions --- --- --- | Install an exception handler. On exception, currently installed handlers --- are executed in reverse (i.e. last in first out) order. Note that multiple --- handlers can be installed for the same exception type. --- -onException :: Exception e => (e -> TransIO ()) -> TransIO () -onException exc= return () `onException'` exc - - -onException' :: Exception e => TransIO a -> (e -> TransIO a) -> TransIO a -onException' mx f= onAnyException mx $ \e -> - case fromException e of - Nothing -> empty - Just e' -> f e' - where - onAnyException :: TransIO a -> (SomeException ->TransIO a) -> TransIO a - onAnyException mx f= mx `onBack` f - --- | Delete all the exception handlers registered till now. -cutExceptions= backCut (undefined :: SomeException) - --- | Used inside an exception handler. Stop executing any further exception --- handlers and resume normal execution from this point on. --- -continue = forward (undefined :: SomeException) - - -catcht mx exc= sandbox $ do - cutExceptions - onException' mx exc - where - sandbox mx= do - exState <- getState <|> backStateOf (undefined :: SomeException) - mx - <*** setState exState - - - - +-----------------------------------------------------------------------------+--+-- Module : Base+-- Copyright :+-- License : MIT+--+-- Maintainer : agocorona@gmail.com+-- Stability :+-- Portability :+--+-- | See http://github.com/agocorona/transient+-- Everything in this module is exported in order to allow extensibility.+-----------------------------------------------------------------------------+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE Rank2Types #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE ConstraintKinds #-}++module Transient.Internals where++import Control.Applicative+import Control.Monad.State+import Data.Dynamic+import qualified Data.Map as M+import System.IO.Unsafe+import Unsafe.Coerce+import Control.Exception hiding (try,onException)+import qualified Control.Exception (try)+import Control.Concurrent+import GHC.Conc(unsafeIOToSTM)+import Control.Concurrent.STM hiding (retry)+import qualified Control.Concurrent.STM as STM (retry)+import System.Mem.StableName+import Data.Maybe++import Data.List+import Data.IORef+import System.Environment+import System.IO++import qualified Data.ByteString.Char8 as BS++#ifdef DEBUG++import Data.Monoid+import Debug.Trace+import System.Exit++{-# INLINE (!>) #-}+(!>) :: Show a => b -> a -> b+(!>) x y = trace (show y) x+infixr 0 !>++#else++{-# INLINE (!>) #-}+(!>) :: a -> b -> a+(!>) = const+++#endif++type StateIO = StateT EventF IO++newtype TransIO a = Transient { runTrans :: StateIO (Maybe a) }++type SData = ()++type EventId = Int++type TransientIO = TransIO++data LifeCycle = Alive | Parent | Listener | Dead+ deriving (Eq, Show)++-- | EventF describes the context of a TransientIO computation:+data EventF = forall a b. EventF+ { meffects :: ()++ , event :: Maybe SData+ -- ^ Not yet consumed result (event) from the last asynchronous run of the+ -- computation++ , xcomp :: TransIO a+ , fcomp :: [b -> TransIO b]+ -- ^ List of continuations++ , mfData :: M.Map TypeRep SData+ -- ^ State data accessed with get or put operations++ , mfSequence :: Int+ , threadId :: ThreadId+ , freeTh :: Bool+ -- ^ When 'True', threads are not killed using kill primitives++ , parent :: Maybe EventF+ -- ^ The parent of this thread++ , children :: MVar [EventF]+ -- ^ Forked child threads, used only when 'freeTh' is 'False'++ , maxThread :: Maybe (IORef Int)+ -- ^ Maximum number of threads that are allowed to be created++ , labelth :: IORef (LifeCycle, BS.ByteString)+ -- ^ Label the thread with its lifecycle state and a label string+ } deriving Typeable++type Effects =+ forall a b c. TransIO a -> TransIO a -> (a -> TransIO b)+ -> StateIO (StateIO (Maybe c) -> StateIO (Maybe c), Maybe a)++instance MonadState EventF TransIO where+ get = Transient $ get >>= return . Just+ put x = Transient $ put x >> return (Just ())+ state f = Transient $ do+ s <- get+ let ~(a, s') = f s+ put s'+ return $ Just a++-- | Run a "non transient" computation within the underlying state monad, so it is+-- guaranteed that the computation neither can stop neither can trigger additional+-- events/threads.+noTrans :: StateIO x -> TransIO x+noTrans x = Transient $ x >>= return . Just++emptyEventF :: ThreadId -> IORef (LifeCycle, BS.ByteString) -> MVar [EventF] -> EventF+emptyEventF th label childs =+ EventF { meffects = mempty+ , event = mempty+ , xcomp = empty+ , fcomp = []+ , mfData = mempty+ , mfSequence = 0+ , threadId = th+ , freeTh = False+ , parent = Nothing+ , children = childs+ , maxThread = Nothing+ , labelth = label }++-- | Run a transient computation with a default initial state+runTransient :: TransIO a -> IO (Maybe a, EventF)+runTransient t = do+ th <- myThreadId+ label <- newIORef $ (Alive, BS.pack "top")+ childs <- newMVar []+ runStateT (runTrans t) $ emptyEventF th label childs++-- | Run a transient computation with a given initial state+runTransState :: EventF -> TransIO x -> IO (Maybe x, EventF)+runTransState st x = runStateT (runTrans x) st++-- | Get the continuation context: closure, continuation, state, child threads etc+getCont :: TransIO EventF+getCont = Transient $ Just <$> get++-- | Run the closure and the continuation using the state data of the calling thread+runCont :: EventF -> StateIO (Maybe a)+runCont EventF { xcomp = x, fcomp = fs } = runTrans $ do+ r <- unsafeCoerce x+ compose fs r++-- | Run the closure and the continuation using its own state data.+runCont' :: EventF -> IO (Maybe a, EventF)+runCont' cont = runStateT (runCont cont) cont++-- | Warning: Radically untyped stuff. handle with care+getContinuations :: StateIO [a -> TransIO b]+getContinuations = do+ EventF { fcomp = fs } <- get+ return $ unsafeCoerce fs++{-+runCont cont = do+ mr <- runClosure cont+ case mr of+ Nothing -> return Nothing+ Just r -> runContinuation cont r+-}++-- | Compose a list of continuations.+compose :: [a -> TransIO a] -> (a -> TransIO b)+compose [] = const empty+compose (f:fs) = \x -> f x >>= compose fs++-- | Run the closure (the 'x' in 'x >>= f') of the current bind operation.+runClosure :: EventF -> StateIO (Maybe a)+runClosure EventF { xcomp = x } = unsafeCoerce (runTrans x)++-- | Run the continuation (the 'f' in 'x >>= f') of the current bind operation with the current state.+runContinuation :: EventF -> a -> StateIO (Maybe b)+runContinuation EventF { fcomp = fs } =+ runTrans . (unsafeCoerce $ compose $ fs)++-- | Save a closure and a continuation ('x' and 'f' in 'x >>= f').+setContinuation :: TransIO a -> (a -> TransIO b) -> [c -> TransIO c] -> StateIO ()+setContinuation b c fs = do+ modify $ \EventF{..} -> EventF { xcomp = b+ , fcomp = unsafeCoerce c : fs+ , .. }++-- | Save a closure and continuation, run the closure, restore the old continuation.+-- | NOTE: The old closure is discarded.+withContinuation :: b -> TransIO a -> TransIO a+withContinuation c mx = do+ EventF { fcomp = fs, .. } <- get+ put $ EventF { xcomp = mx+ , fcomp = unsafeCoerce c : fs+ , .. }+ r <- mx+ restoreStack fs+ return r++-- | Restore the continuations to the provided ones.+-- | NOTE: Events are also cleared out.+restoreStack :: MonadState EventF m => [a -> TransIO a] -> m ()+restoreStack fs = modify $ \EventF {..} -> EventF { event = Nothing, fcomp = fs, .. }++-- | Run a chain of continuations.+-- WARNING: It is up to the programmer to assure that each continuation typechecks+-- with the next, and that the parameter type match the input of the first+-- continuation.+-- NOTE: Normally this makes sense to stop the current flow with `stop` after the+-- invocation.+runContinuations :: [a -> TransIO b] -> c -> TransIO d+runContinuations fs x = compose (unsafeCoerce fs) x++-- Instances for Transient Monad++instance Functor TransIO where+ fmap f mx = do+ x <- mx+ return $ f x++instance Applicative TransIO where+ pure a = Transient . return $ Just a++ f <*> g = Transient $ do+ rf <- liftIO $ newIORef (Nothing,[])+ rg <- liftIO $ newIORef (Nothing,[])+++ fs <- getContinuations++ let hasWait (_:Wait:_) = True+ hasWait _ = False++ appf k = Transient $ do+ Log rec _ full <- getData `onNothing` return (Log False [] [])+ (liftIO $ writeIORef rf (Just k,full))+ -- !> ( show $ unsafePerformIO myThreadId) ++"APPF"++ (x, full2)<- liftIO $ readIORef rg+ when (hasWait full ) $+ -- (!> (hasWait full,"full",full, "\nfull2",full2)) $+ let full'= head full: full2+ in (setData $ Log rec full' full')+ -- !> ("result1",full')++ return $ Just k <*> x++ appg x = Transient $ do+ Log rec _ full <- getData `onNothing` return (Log False [] [])+ liftIO $ writeIORef rg (Just x, full)++ -- !> ( show $ unsafePerformIO myThreadId) ++ "APPG"++ (k,full1) <- liftIO $ readIORef rf+ when (hasWait full) $++ -- (!> ("full", full, "\nfull1",full1)) $++ let full'= head full: full1+ in (setData $ Log rec full' full')+ -- !> ("result2",full')++ return $ k <*> Just x++ setContinuation f appf fs++ k <- runTrans f+ -- !> ( show $ unsafePerformIO myThreadId)++ "RUN f"++ was <- getData `onNothing` return NoRemote+ when (was == WasParallel) $ setData NoRemote++ Log recovery _ full <- getData `onNothing` return (Log False [] [])++++ if was== WasRemote || (not recovery && was == NoRemote && isNothing k )+ -- !> ("was,recovery,isNothing=",was,recovery, isNothing k)++ -- if the first operand was a remote request+ -- (so this node is not master and hasn't to execute the whole expression)+ -- or it was not an asyncronous term (a normal term without async or parallel+ -- like primitives) and is nothing+ then do+ restoreStack fs+ return Nothing+ else do+ when (isJust k) $ liftIO $ writeIORef rf (k,full)+ -- when necessary since it maybe WasParallel and Nothing++ setContinuation g appg fs++ x <- runTrans g+ -- !> ( show $ unsafePerformIO myThreadId) ++ "RUN g"++ Log recovery _ full' <- getData `onNothing` return (Log False [] [])+ liftIO $ writeIORef rg (x,full')+ restoreStack fs+ k'' <- if was== WasParallel+ then do+ (k',_) <- liftIO $ readIORef rf -- since k may have been updated by a parallel f+ return k'+ else return k+ return $ k'' <*> x++instance Monad TransIO where+ return = pure+ x >>= f = Transient $ do+ c <- setEventCont x f+ mk <- runTrans x+ resetEventCont mk c+ case mk of+ Just k -> runTrans (f k)+ Nothing -> return Nothing++instance MonadIO TransIO where+ -- liftIO mx = do+ -- ex <- liftIO' $ (mx >>= return . Right) `catch`+ -- (\(e :: SomeException) -> return $ Left e)+ -- case ex of+ -- Left e -> back e -- finish $ Just e+ -- Right x -> return x+ -- where + liftIO x = Transient $ liftIO x >>= return . Just+ -- let x= liftIO io in x `seq` lift x++instance Monoid a => Monoid (TransIO a) where+ mappend x y = mappend <$> x <*> y+ mempty = return mempty++instance Alternative TransIO where+ empty = Transient $ return Nothing+ (<|>) = mplus++instance MonadPlus TransIO where+ mzero = empty+ mplus x y = Transient $ do+ mx <- runTrans x++ was <- getData `onNothing` return NoRemote+ if was == WasRemote++ then return Nothing+ else case mx of+ Nothing -> runTrans y++ justx -> return justx++readWithErr :: (Typeable a, Read a) => String -> IO [(a, String)]+readWithErr line =+ (v `seq` return [(v, left)])+ `catch` (\(e :: SomeException) ->+ error $ "read error trying to read type: \"" ++ show (typeOf v)+ ++ "\" in: " ++ " <" ++ show line ++ "> ")+ where [(v, left)] = readsPrec 0 line++readsPrec' _ = unsafePerformIO . readWithErr++-- | Constraint type synonym for a value that can be logged.+type Loggable a = (Show a, Read a, Typeable a)++-- | Dynamic serializable data for logging.+data IDynamic =+ IDyns String+ | forall a. Loggable a => IDynamic a++instance Show IDynamic where+ show (IDynamic x) = show (show x)+ show (IDyns s) = show s++instance Read IDynamic where+ readsPrec n str = map (\(x,s) -> (IDyns x,s)) $ readsPrec' n str++type Recover = Bool+type CurrentPointer = [LogElem]+type LogEntries = [LogElem]++data LogElem = Wait | Exec | Var IDynamic+ deriving (Read, Show)++data Log = Log Recover CurrentPointer LogEntries+ deriving (Typeable, Show)++data RemoteStatus = WasRemote | WasParallel | NoRemote+ deriving (Typeable, Eq, Show)++-- | A synonym of 'empty' that can be used in a monadic expression. It stops+-- the computation, which allows the next computation in an 'Alternative'+-- ('<|>') composition to run.+stop :: Alternative m => m stopped+stop = empty++--instance (Num a,Eq a,Fractional a) =>Fractional (TransIO a)where+-- mf / mg = (/) <$> mf <*> mg+-- fromRational (x:%y) = fromInteger x % fromInteger y+++instance (Num a, Eq a) => Num (TransIO a) where+ fromInteger = return . fromInteger+ mf + mg = (+) <$> mf <*> mg+ mf * mg = (*) <$> mf <*> mg+ negate f = f >>= return . negate+ abs f = f >>= return . abs+ signum f = f >>= return . signum++class AdditionalOperators m where++ -- | Run @m a@ discarding its result before running @m b@.+ (**>) :: m a -> m b -> m b++ -- | Run @m b@ discarding its result, after the whole task set @m a@ is+ -- done.+ (<**) :: m a -> m b -> m a++ atEnd' :: m a -> m b -> m a+ atEnd' = (<**)++ -- | Run @m b@ discarding its result, once after each task in @m a@, and+ -- once again after the whole task set is done.+ (<***) :: m a -> m b -> m a++ atEnd :: m a -> m b -> m a+ atEnd = (<***)++instance AdditionalOperators TransIO where++ (**>) :: TransIO a -> TransIO b -> TransIO b+ (**>) x y =+ Transient $ do+ runTrans x+ runTrans y++ (<***) :: TransIO a -> TransIO b -> TransIO a+ (<***) ma mb =+ Transient $ do+ fs <- getContinuations+ setContinuation ma (\x -> mb >> return x) fs+ a <- runTrans ma+ runTrans mb+ restoreStack fs+ return a++ (<**) :: TransIO a -> TransIO b -> TransIO a+ (<**) ma mb =+ Transient $ do+ a <- runTrans ma+ runTrans mb+ return a++infixr 1 <***, <**, **>++-- | Run @b@ once, discarding its result when the first task in task set @a@+-- has finished. Useful to start a singleton task after the first task has been+-- setup.+(<|) :: TransIO a -> TransIO b -> TransIO a+(<|) ma mb = Transient $ do+ fs <- getContinuations+ ref <- liftIO $ newIORef False+ setContinuation ma (cont ref) fs+ r <- runTrans ma+ restoreStack fs+ return r+ where cont ref x = Transient $ do+ n <- liftIO $ readIORef ref+ if n == True+ then return $ Just x+ else do liftIO $ writeIORef ref True+ runTrans mb+ return $ Just x++-- | Set the current closure and continuation for the current statement+setEventCont :: TransIO a -> (a -> TransIO b) -> StateIO EventF+setEventCont x f = do+ EventF { fcomp = fs, .. } <- get+ let cont = EventF { xcomp = x+ , fcomp = unsafeCoerce f : fs+ , .. }+ put cont+ return cont++-- | Reset the closure and continuation. Remove inner binds than the previous+-- computations may have stacked in the list of continuations.+-- resetEventCont :: Maybe a -> EventF -> StateIO (TransIO b -> TransIO b)+resetEventCont mx _ = do+ EventF { fcomp = fs, .. } <- get+ let f mx = case mx of+ Nothing -> empty+ Just x -> unsafeCoerce (head fs) x+ put $ EventF { xcomp = f mx+ , fcomp = tailsafe fs+ , .. }+ return id++-- | Total variant of `tail` that returns an empty list when given an empty list.+tailsafe :: [a] -> [a]+tailsafe [] = []+tailsafe (x:xs) = xs++--refEventCont= unsafePerformIO $ newIORef baseEffects++-- effects not used+-- {-# INLINE baseEffects #-}+--baseEffects :: Effects+--+--baseEffects x x' f' = do+-- c <- setEventCont x' f'+-- mk <- runTrans x+-- t <- resetEventCont mk c+-- return (t,mk)+++--instance MonadTrans (Transient ) where+-- lift mx = Transient $ mx >>= return . Just++-- * Threads++waitQSemB sem = atomicModifyIORef sem $ \n ->+ if n > 0 then(n - 1, True) else (n, False)+signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())++-- | Sets the maximum number of threads that can be created for the given task+-- set. When set to 0, new tasks start synchronously in the current thread.+-- New threads are created by 'parallel', and APIs that use parallel.+threads :: Int -> TransIO a -> TransIO a+threads n process = do+ msem <- gets maxThread+ sem <- liftIO $ newIORef n+ modify $ \s -> s { maxThread = Just sem }+ r <- process <** (modify $ \s -> s { maxThread = msem }) -- restore it+ return r++-- | Terminate all the child threads in the given task set and continue+-- execution in the current thread. Useful to reap the children when a task is+-- done.+--+oneThread :: TransIO a -> TransIO a+oneThread comp = do+ st <- get+ chs <- liftIO $ newMVar []+ label <- liftIO $ newIORef (Alive, BS.pack "oneThread")+ let st' = st { parent = Just st+ , children = chs+ , labelth = label }+ liftIO $ hangThread st st'+ put st'+ x <- comp+ th <- liftIO myThreadId+ -- !> ("FATHER:", threadId st)+ chs <- liftIO $ readMVar chs -- children st'+ liftIO $ mapM_ (killChildren1 th) chs+ return x+ where killChildren1 :: ThreadId -> EventF -> IO ()+ killChildren1 th state = do+ ths' <- modifyMVar (children state) $ \ths -> do+ let (inn, ths')= partition (\st -> threadId st == th) ths+ return (inn, ths')+ mapM_ (killChildren1 th) ths'+ mapM_ (killThread . threadId) ths'+ -- !> ("KILLEVENT1 ", map threadId ths' )++-- | Add a label to the current passing threads so it can be printed by debugging calls like `showThreads`+labelState :: (MonadIO m,MonadState EventF m) => String -> m ()+labelState l = do+ st <- get+ liftIO $ atomicModifyIORef (labelth st) $ \(status,_) -> ((status, BS.pack l), ())++printBlock :: MVar ()+printBlock = unsafePerformIO $ newMVar ()++-- | Show the tree of threads hanging from the state.+showThreads :: MonadIO m => EventF -> m ()+showThreads st = liftIO $ withMVar printBlock $ const $ do+ mythread <- myThreadId++ putStrLn "---------Threads-----------"+ let showTree n ch = do+ liftIO $ do+ putStr $ take n $ repeat ' '+ (state, label) <- readIORef $ labelth ch+ if BS.null label+ then putStr . show $ threadId ch+ else do BS.putStr label; putStr . drop 8 . show $ threadId ch+ when (state == Dead) $ putStr " dead"+ putStrLn $ if mythread == threadId ch then " <--" else ""+ chs <- readMVar $ children ch+ mapM_ (showTree $ n + 2) $ reverse chs+ showTree 0 st++-- | Return the state of the thread that initiated the transient computation+topState :: TransIO EventF+topState = do+ st <- get+ return $ toplevel st+ where toplevel st = case parent st of+ Nothing -> st+ Just p -> toplevel p++-- | Return the state variable of the type desired with which a thread, identified by his number in the treee was initiated+showState :: (Typeable a, MonadIO m, Alternative m) => String -> EventF -> m (Maybe a)+showState th top = resp+ where resp = do+ let thstring = drop 9 . show $ threadId top+ if thstring == th+ then getstate top+ else do+ sts <- liftIO $ readMVar $ children top+ foldl (<|>) empty $ map (showState th) sts+ getstate st =+ case M.lookup (typeOf $ typeResp resp) $ mfData st of+ Just x -> return . Just $ unsafeCoerce x+ Nothing -> return Nothing+ typeResp :: m (Maybe x) -> x+ typeResp = undefined++-- | Add n threads to the limit of threads. If there is no limit, the limit is set.+addThreads' :: Int -> TransIO ()+addThreads' n= noTrans $ do+ msem <- gets maxThread+ case msem of+ Just sem -> liftIO $ modifyIORef sem $ \n' -> n + n'+ Nothing -> do+ sem <- liftIO (newIORef n)+ modify $ \ s -> s { maxThread = Just sem }++-- | Ensure that at least n threads are available for the current task set.+addThreads :: Int -> TransIO ()+addThreads n = noTrans $ do+ msem <- gets maxThread+ case msem of+ Nothing -> return ()+ Just sem -> liftIO $ modifyIORef sem $ \n' -> if n' > n then n' else n++--getNonUsedThreads :: TransIO (Maybe Int)+--getNonUsedThreads= Transient $ do+-- msem <- gets maxThread+-- case msem of+-- Just sem -> liftIO $ Just <$> readIORef sem+-- Nothing -> return Nothing++-- | Disable tracking and therefore the ability to terminate the child threads.+-- By default, child threads are terminated automatically when the parent+-- thread dies, or they can be terminated using the kill primitives. Disabling+-- it may improve performance a bit, however, all threads must be well-behaved+-- to exit on their own to avoid a leak.+freeThreads :: TransIO a -> TransIO a+freeThreads process = Transient $ do+ st <- get+ put st { freeTh = True }+ r <- runTrans process+ modify $ \s -> s { freeTh = freeTh st }+ return r++-- | Enable tracking and therefore the ability to terminate the child threads.+-- This is the default but can be used to re-enable tracking if it was+-- previously disabled with 'freeThreads'.+hookedThreads :: TransIO a -> TransIO a+hookedThreads process = Transient $ do+ st <- get+ put st {freeTh = False}+ r <- runTrans process+ modify $ \st -> st { freeTh = freeTh st }+ return r++-- | Kill all the child threads of the current thread.+killChilds :: TransIO ()+killChilds = noTrans $ do+ cont <- get+ liftIO $ do+ killChildren $ children cont+ writeIORef (labelth cont) (Alive, mempty)+ -- !> (threadId cont,"relabeled")+ return ()++-- | Kill the current thread and the childs.+killBranch :: TransIO ()+killBranch = noTrans $ do+ st <- get+ liftIO $ killBranch' st++-- | Kill the childs and the thread of an state+killBranch' :: EventF -> IO ()+killBranch' cont = do+ killChildren $ children cont+ let thisth = threadId cont+ mparent = parent cont+ when (isJust mparent) $+ modifyMVar_ (children $ fromJust mparent) $ \sts ->+ return $ filter (\st -> threadId st /= thisth) sts+ killThread $ thisth++-- * Extensible State: Session Data Management++-- | Same as 'getSData' but with a more general type. If the data is found, a+-- 'Just' value is returned. Otherwise, a 'Nothing' value is returned.+getData :: (MonadState EventF m, Typeable a) => m (Maybe a)+getData = resp+ where resp = do+ list <- gets mfData+ case M.lookup (typeOf $ typeResp resp) list of+ Just x -> return . Just $ unsafeCoerce x+ Nothing -> return Nothing+ typeResp :: m (Maybe x) -> x+ typeResp = undefined++-- | Retrieve a previously stored data item of the given data type from the+-- monad state. The data type to retrieve is implicitly determined from the+-- requested type context.+-- If the data item is not found, an 'empty' value (a void event) is returned.+-- Remember that an empty value stops the monad computation. If you want to+-- print an error message or a default value in that case, you can use an+-- 'Alternative' composition. For example:+--+-- > getSData <|> error "no data"+-- > getInt = getSData <|> return (0 :: Int)+getSData :: Typeable a => TransIO a+getSData = Transient getData++-- | Same as `getSData`+getState :: Typeable a => TransIO a+getState = getSData++-- | 'setData' stores a data item in the monad state which can be retrieved+-- later using 'getData' or 'getSData'. Stored data items are keyed by their+-- data type, and therefore only one item of a given type can be stored. A+-- newtype wrapper can be used to distinguish two data items of the same type.+--+-- @+-- import Control.Monad.IO.Class (liftIO)+-- import Transient.Base+-- import Data.Typeable+--+-- data Person = Person+-- { name :: String+-- , age :: Int+-- } deriving Typeable+--+-- main = keep $ do+-- setData $ Person "Alberto" 55+-- Person name age <- getSData+-- liftIO $ print (name, age)+-- @+setData :: (MonadState EventF m, Typeable a) => a -> m ()+setData x = modify $ \st -> st { mfData = M.insert t (unsafeCoerce x) (mfData st) }+ where t = typeOf x++-- | Accepts a function that takes the current value of the stored data type+-- and returns the modified value. If the function returns 'Nothing' the value+-- is deleted otherwise updated.+modifyData :: (MonadState EventF m, Typeable a) => (Maybe a -> Maybe a) -> m ()+modifyData f = modify $ \st -> st { mfData = M.alter alterf t (mfData st) }+ where typeResp :: (Maybe a -> b) -> a+ typeResp = undefined+ t = typeOf (typeResp f)+ alterf mx = unsafeCoerce $ f x'+ where x' = case mx of+ Just x -> Just $ unsafeCoerce x+ Nothing -> Nothing++-- | Same as modifyData+modifyState :: (MonadState EventF m, Typeable a) => (Maybe a -> Maybe a) -> m ()+modifyState = modifyData++-- | Same as 'setData'+setState :: (MonadState EventF m, Typeable a) => a -> m ()+setState = setData++-- | Delete the data item of the given type from the monad state.+delData :: (MonadState EventF m, Typeable a) => a -> m ()+delData x = modify $ \st -> st { mfData = M.delete (typeOf x) (mfData st) }++-- | Same as 'delData'+delState :: (MonadState EventF m, Typeable a) => a -> m ()+delState = delData++-- | Run an action, if the result is a void action undo any state changes+-- that it might have caused.+try :: TransIO a -> TransIO a+try mx = do+ sd <- gets mfData+ mx <|> (modify (\s -> s { mfData = sd }) >> empty)++-- | Executes the computation and reset the state either if it fails or not+sandbox :: TransIO a -> TransIO a+sandbox mx = do+ sd <- gets mfData+ mx <*** modify (\s ->s { mfData = sd})++-- | Generator of identifiers that are unique within the current monadic+-- sequence They are not unique in the whole program.+genId :: MonadState EventF m => m Int+genId = do+ st <- get+ let n = mfSequence st+ put st { mfSequence = n + 1 }+ return n++getPrevId :: MonadState EventF m => m Int+getPrevId = gets mfSequence++instance Read SomeException where+ readsPrec n str = [(SomeException $ ErrorCall s, r)]+ where [(s , r)] = read str++-- | 'StreamData' represents a task in a task stream being generated.+data StreamData a =+ SMore a -- ^ More tasks to come+ | SLast a -- ^ This is the last task+ | SDone -- ^ No more tasks, we are done+ | SError SomeException -- ^ An error occurred+ deriving (Typeable, Show,Read)++-- | An task stream generator that produces an infinite stream of tasks by+-- running an IO computation in a loop. A task is triggered carrying the output+-- of the computation. See 'parallel' for notes on the return value.+waitEvents :: IO a -> TransIO a+waitEvents io = do+ mr <- parallel (SMore <$> io)+ case mr of+ SMore x -> return x+ SError e -> back e++-- | Run an IO computation asynchronously and generate a single task carrying+-- the result of the computation when it completes. See 'parallel' for notes on+-- the return value.+async :: IO a -> TransIO a+async io = do+ mr <- parallel (SLast <$> io)+ case mr of+ SLast x -> return x+ SError e -> back e++-- | Force an async computation to run synchronously. It can be useful in an+-- 'Alternative' composition to run the alternative only after finishing a+-- computation. Note that in Applicatives it might result in an undesired+-- serialization.+sync :: TransIO a -> TransIO a+sync x = do+ setData WasRemote+ r <- x+ delData WasRemote+ return r++-- | @spawn = freeThreads . waitEvents@+spawn :: IO a -> TransIO a+spawn = freeThreads . waitEvents++-- | An task stream generator that produces an infinite stream of tasks by+-- running an IO computation periodically at the specified time interval. The+-- task carries the result of the computation. A new task is generated only if+-- the output of the computation is different from the previous one. See+-- 'parallel' for notes on the return value.+sample :: Eq a => IO a -> Int -> TransIO a+sample action interval = do+ v <- liftIO action+ prev <- liftIO $ newIORef v+ waitEvents (loop action prev) <|> async (return v)+ where loop action prev = loop'+ where loop' = do+ threadDelay interval+ v <- action+ v' <- readIORef prev+ if v /= v' then writeIORef prev v >> return v else loop'++--serial :: IO (StreamData b) -> TransIO (StreamData b)+--serial ioaction= Transient $ do+-- cont <- get -- !> "PARALLEL"+-- case event cont of+-- j@(Just _) -> do+-- put cont{event=Nothing}+-- return $ unsafeCoerce j+-- Nothing -> do+-- liftIO $ loop cont ioaction+-- return Nothing+--+-- where loop cont ioaction= do+-- let iocont dat= do+-- runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat}+-- return ()+-- mdat <- ioaction `catch` \(e :: SomeException) -> return $ SError e+-- case mdat of+-- se@(SError _) -> iocont se+-- SDone -> iocont SDone+-- last@(SLast _) -> iocont last+--+-- more@(SMore _) -> do+-- iocont more+-- loop cont ioaction++-- | Run an IO action one or more times to generate a stream of tasks. The IO+-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new+-- task is triggered with the result value. If the return value is 'SMore', the+-- action is run again to generate the next task, otherwise task creation+-- stops.+--+-- Unless the maximum number of threads (set with 'threads') has been reached,+-- the task is generated in a new thread and the current thread returns a void+-- task.+parallel :: IO (StreamData b) -> TransIO (StreamData b)+parallel ioaction = Transient $ do+ cont <- get+ -- !> "PARALLEL"+ case event cont of+ j@(Just _) -> do+ put cont { event = Nothing }+ return $ unsafeCoerce j+ Nothing -> do+ liftIO $ atomicModifyIORef (labelth cont) $ \(_, lab) -> ((Parent, lab), ())+ liftIO $ loop cont ioaction+ was <- getData `onNothing` return NoRemote+ when (was /= WasRemote) $ setData WasParallel+-- th <- liftIO myThreadId+-- return () !> ("finish",th)+ return Nothing++-- | Execute the IO action and the continuation+loop :: EventF -> IO (StreamData t) -> IO ()+loop parentc rec = forkMaybe parentc $ \cont -> do+ -- Execute the IO computation and then the closure-continuation+ liftIO $ atomicModifyIORef (labelth cont) $ const ((Listener,BS.pack "wait"),())+ let loop'= do+ mdat <- rec `catch` \(e :: SomeException) -> return $ SError e+ case mdat of+ se@(SError _) -> setworker cont >> iocont se cont+ SDone -> setworker cont >> iocont SDone cont+ last@(SLast _) -> setworker cont >> iocont last cont++ more@(SMore _) -> do+ forkMaybe cont $ iocont more+ loop'++ where+ setworker cont= liftIO $ atomicModifyIORef (labelth cont) $ const ((Alive,BS.pack "work"),())++ iocont dat cont = do++ let cont'= cont{event= Just $ unsafeCoerce dat}+ runStateT (runCont cont') cont'+ return ()+++ loop'+ return ()+ where++ forkMaybe parent proc = do+ case maxThread parent of+ Nothing -> forkIt parent proc+ Just sem -> do+ dofork <- waitQSemB sem+ if dofork then forkIt parent proc else proc parent+++ forkIt parent proc= do+ chs <- liftIO $ newMVar []++ label <- newIORef (Alive, BS.pack "work")+ let cont = parent{parent=Just parent,children= chs, labelth= label}++ forkFinally1 (do+ th <- myThreadId+ let cont'= cont{threadId=th}+ when(not $ freeTh parent )$ hangThread parent cont'+ -- !> ("thread created: ",th,"in",threadId parent )++ proc cont')+ $ \me -> do++-- case me of -- !> "THREAD END" of+-- Left e -> do+---- when (fromException e /= Just ThreadKilled)$+-- liftIO $ print e+-- killChildren $ children cont+---- !> "KILL RECEIVED" ++ (show $ unsafePerformIO myThreadId)+--+-- Right _ ->++++ case maxThread cont of+ Just sem -> signalQSemB sem -- !> "freed thread"+ Nothing -> when(not $ freeTh parent ) $ do -- if was not a free thread++ th <- myThreadId+ (can,label) <- atomicModifyIORef (labelth cont) $ \(l@(status,label)) ->+ ((if status== Alive then Dead else status, label),l)+++ when (can/= Parent ) $ free th parent+ return ()+++ forkFinally1 :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId+ forkFinally1 action and_then =+ mask $ \restore -> forkIO $ Control.Exception.try (restore action) >>= and_then++free th env= do+-- return () !> ("freeing",th,"in",threadId env)+ let sibling= children env++ (sbs',found) <- modifyMVar sibling $ \sbs -> do+ let (sbs', found) = drop [] th sbs+ return (sbs',(sbs',found))++++ if found+ then do++-- !> ("new list for",threadId env,map threadId sbs')+ (typ,_) <- readIORef $ labelth env+ if (null sbs' && typ /= Listener && isJust (parent env))+ -- free the parent+ then free (threadId env) ( fromJust $ parent env)+ else return ()++-- return env+ else return () -- putMVar sibling sbs+ -- !> (th,"orphan")++ where+ drop processed th []= (processed,False)+ drop processed th (ev:evts)| th == threadId ev= (processed ++ evts, True)+ | otherwise= drop (ev:processed) th evts++++hangThread parentProc child = do++ let headpths= children parentProc++ modifyMVar_ headpths $ \ths -> return (child:ths)+-- ths <- takeMVar headpths+-- putMVar headpths (child:ths)+++ -- !> ("hang", threadId child, threadId parentProc,map threadId ths,unsafePerformIO $ readIORef $ labelth parentProc)++-- | kill all the child threads associated with the continuation context+killChildren childs = do+++ ths <- modifyMVar childs $ \ths -> return ([],ths)+-- ths <- takeMVar childs+-- putMVar childs []++ mapM_ (killChildren . children) ths+++ mapM_ (killThread . threadId) ths -- !> ("KILL", map threadId ths )++++++-- | Make a transient task generator from an asynchronous callback handler.+--+-- The first parameter is a callback. The second parameter is a value to be+-- returned to the callback; if the callback expects no return value it+-- can just be a @return ()@. The callback expects a setter function taking the+-- @eventdata@ as an argument and returning a value to the callback; this+-- function is supplied by 'react'.+--+-- Callbacks from foreign code can be wrapped into such a handler and hooked+-- into the transient monad using 'react'. Every time the callback is called it+-- generates a new task for the transient monad.+--++react+ :: Typeable eventdata+ => ((eventdata -> IO response) -> IO ())+ -> IO response+ -> TransIO eventdata+react setHandler iob= Transient $ do+ cont <- get+ case event cont of+ Nothing -> do+ liftIO $ setHandler $ \dat ->do+ runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat}+ iob+ was <- getData `onNothing` return NoRemote+ when (was /= WasRemote) $ setData WasParallel+ return Nothing++ j@(Just _) -> do+ put cont{event=Nothing}+ return $ unsafeCoerce j++-- | Runs a computation asynchronously without generating any events. Returns+-- 'empty' in an 'Alternative' composition.++abduce = async $ return ()+-- Transient $ do+-- st <- get+-- case event st of+-- Just _ -> do+-- put st{event=Nothing}+-- return $ Just ()+-- Nothing -> do+-- chs <- liftIO $ newMVar []+--+-- label <- liftIO $ newIORef (Alive, BS.pack "abduce")+-- liftIO $ forkIO $ do+-- th <- myThreadId+-- let st' = st{event= Just (),parent=Just st,children= chs, threadId=th,labelth= label}+-- liftIO $ hangThread st st'+--+-- runCont' st'+-- return()+-- return Nothing+++-- * non-blocking keyboard input++getLineRef= unsafePerformIO $ newTVarIO Nothing+++roption= unsafePerformIO $ newMVar []++-- | Waits on stdin in a loop and triggers a new task every time the input data+-- matches the first parameter. The value contained by the task is the matched+-- value i.e. the first argument itself. The second parameter is a label for+-- the option. The label is displayed on the console when the option is+-- activated.+--+-- Note that if two independent invocations of 'option' are expecting the same+-- input, only one of them gets it and triggers a task. It cannot be+-- predicted which one gets it.+--+option :: (Typeable b, Show b, Read b, Eq b) =>+ b -> String -> TransIO b+option ret message= do+ let sret= show ret++ liftIO $ putStrLn $ "Enter "++sret++"\tto: " ++ message+ liftIO $ modifyMVar_ roption $ \msgs-> return $ sret:msgs+ waitEvents $ getLine' (==ret)+ liftIO $ putStr "\noption: " >> putStrLn (show ret)+ return ret+++-- | Waits on stdin and triggers a task when a console input matches the+-- predicate specified in the first argument. The second parameter is a string+-- to be displayed on the console before waiting.+--+input cond prompt= input' Nothing cond prompt++input' :: (Typeable a, Read a,Show a) => Maybe a -> (a -> Bool) -> String -> TransIO a+input' mv cond prompt= Transient . liftIO $do+ putStr prompt >> hFlush stdout+ atomically $ do+ mr <- readTVar getLineRef+ case mr of+ Nothing -> STM.retry+ Just r ->+ case reads1 r of+ (s,_):_ -> if cond s -- !> show (cond s)+ then do+ unsafeIOToSTM $ print s+ writeTVar getLineRef Nothing -- !>"match"+ return $ Just s++ else return mv+ _ -> return mv+++-- | Non blocking `getLine` with a validator+getLine' cond= do+ atomically $ do+ mr <- readTVar getLineRef+ case mr of+ Nothing -> STM.retry+ Just r ->+ case reads1 r of -- !> ("received " ++ show r ++ show (unsafePerformIO myThreadId)) of+ (s,_):_ -> if cond s -- !> show (cond s)+ then do+ writeTVar getLineRef Nothing -- !>"match"+ return s++ else STM.retry+ _ -> STM.retry++reads1 s=x where+ x= if typeOf(typeOfr x) == typeOf "" then unsafeCoerce[(s,"")] else readsPrec' 0 s+ typeOfr :: [(a,String)] -> a+ typeOfr = undefined+++inputLoop= do+ r <- getLine+ -- XXX hoping that the previous value has been consumed by now.+ -- otherwise its just lost by overwriting.+ atomically $ writeTVar getLineRef Nothing+ processLine r+ inputLoop++processLine r= do++ let rs = breakSlash [] r++ -- XXX this blocks forever if an input is not consumed by any consumer.+ -- e.g. try this "xxx/xxx" on the stdin+ liftIO $ mapM_ (\ r ->+ atomically $ do+-- threadDelay 1000000+ t <- readTVar getLineRef+ when (isJust t) STM.retry+ writeTVar getLineRef $ Just r ) rs+++ where+ breakSlash :: [String] -> String -> [String]+ breakSlash [] ""= [""]+ breakSlash s ""= s+ breakSlash res ('\"':s)=+ let (r,rest) = span(/= '\"') s+ in breakSlash (res++[r]) $ tail1 rest++ breakSlash res s=+ let (r,rest) = span(/= '/') s+ in breakSlash (res++[r]) $ tail1 rest++ tail1 []=[]+ tail1 x= tail x+++++-- | Wait for the execution of `exit` and return the result or the exhaustion of thread activity++stay rexit= takeMVar rexit+ `catch` \(e :: BlockedIndefinitelyOnMVar) -> return Nothing++newtype Exit a= Exit a deriving Typeable++-- | Runs the transient computation in a child thread and keeps the main thread+-- running until all the user threads exit or some thread invokes 'exit'.+--+-- The main thread provides facilities to accept keyboard input in a+-- non-blocking but line-oriented manner. The program reads the standard input+-- and feeds it to all the async input consumers (e.g. 'option' and 'input').+-- All async input consumers contend for each line entered on the standard+-- input and try to read it atomically. When a consumer consumes the input+-- others do not get to see it, otherwise it is left in the buffer for others+-- to consume. If nobody consumes the input, it is discarded.+--+-- A @/@ in the input line is treated as a newline.+--+-- When using asynchronous input, regular synchronous IO APIs like getLine+-- cannot be used as they will contend for the standard input along with the+-- asynchronous input thread. Instead you can use the asynchronous input APIs+-- provided by transient.+--+-- A built-in interactive command handler also reads the stdin asynchronously.+-- All available commands handled by the command handler are displayed when the+-- program is run. The following commands are available:+--+-- 1. @ps@: show threads+-- 2. @log@: inspect the log of a thread+-- 3. @end@, @exit@: terminate the program+--+-- An input not handled by the command handler can be handled by the program.+--+-- The program's command line is scanned for @-p@ or @--path@ command line+-- options. The arguments to these options are injected into the async input+-- channel as keyboard input to the program. Each line of input is separated by+-- a @/@. For example:+--+-- > foo -p ps/end+--+keep :: Typeable a => TransIO a -> IO (Maybe a)+keep mx = do++ liftIO $ hSetBuffering stdout LineBuffering+ rexit <- newEmptyMVar+ forkIO $ do+-- liftIO $ putMVar rexit $ Right Nothing+ runTransient $ do+ st <- get++ setData $ Exit rexit+ (async (return ()) >> labelState "input" >> liftIO inputLoop)++ <|> do+ option "ps" "show threads"+ liftIO $ showThreads st+ empty+ <|> do+ option "log" "inspect the log of a thread"+ th <- input (const True) "thread number>"+ ml <- liftIO $ showState th st+ liftIO $ print $ fmap (\(Log _ _ log) -> reverse log) ml+ empty+ <|> do+ option "end" "exit"+ killChilds+ liftIO $ putMVar rexit Nothing+ empty+ <|> mx+ return ()+ threadDelay 10000+ execCommandLine+ stay rexit++ where+ type1 :: TransIO a -> Either String (Maybe a)+ type1= undefined++-- | Same as `keep` but does not read from the standard input, and therefore+-- the async input APIs ('option' and 'input') cannot be used in the monad.+-- However, keyboard input can still be passed via command line arguments as+-- described in 'keep'. Useful for debugging or for creating background tasks,+-- as well as to embed the Transient monad inside another computation. It+-- returns either the value returned by `exit`. or Nothing, when there are no+-- more threads running+--+keep' :: Typeable a => TransIO a -> IO (Maybe a)+keep' mx = do+ liftIO $ hSetBuffering stdout LineBuffering+ rexit <- newEmptyMVar+ forkIO $ do+ runTransient $ do+ setData $ Exit rexit+ mx++ return ()+ threadDelay 10000+ forkIO $ execCommandLine+ stay rexit+++execCommandLine= do+ args <- getArgs+ let mindex = findIndex (\o -> o == "-p" || o == "--path" ) args+ when (isJust mindex) $ do+ let i= fromJust mindex +1+ when (length args >= i) $ do+ let path= args !! i+ putStr "Executing: " >> print path+ processLine path++-- | Exit the main thread, and thus all the Transient threads (and the+-- application if there is no more code)+exit :: Typeable a => a -> TransIO a+exit x= do+ Exit rexit <- getSData <|> error "exit: not the type expected" `asTypeOf` type1 x+ liftIO $ putMVar rexit $ Just x+ stop+ where+ type1 :: a -> TransIO (Exit (MVar (Maybe a)))+ type1= undefined++++-- | If the first parameter is 'Nothing' return the second parameter otherwise+-- return the first parameter..+onNothing :: Monad m => m (Maybe b) -> m b -> m b+onNothing iox iox'= do+ mx <- iox+ case mx of+ Just x -> return x+ Nothing -> iox'++++++----------------------------------backtracking ------------------------+++data Backtrack b= Show b =>Backtrack{backtracking :: Maybe b+ ,backStack :: [EventF] }+ deriving Typeable++++-- | Delete all the undo actions registered till now for the given track id.+backCut :: (Typeable b, Show b) => b -> TransientIO ()+backCut reason= Transient $ do+ delData $ Backtrack (Just reason) []+ return $ Just ()++-- | 'backCut' for the default track; equivalent to @backCut ()@.+undoCut :: TransientIO ()+undoCut = backCut ()++-- | Run the action in the first parameter and register the second parameter as+-- the undo action. On undo ('back') the second parameter is called with the+-- undo track id as argument.+--+{-# NOINLINE onBack #-}+onBack :: (Typeable b, Show b) => TransientIO a -> ( b -> TransientIO a) -> TransientIO a+onBack ac bac = registerBack (typeof bac) $ Transient $ do+ Backtrack mreason stack <- getData `onNothing` backStateOf (typeof bac)+ runTrans $ case mreason of+ Nothing -> ac+ Just reason -> do+ setState $ Backtrack mreason $ tail stack -- to avoid recursive call tot he same handler+ bac reason+ where+ typeof :: (b -> TransIO a) -> b+ typeof = undefined++-- | 'onBack' for the default track; equivalent to @onBack ()@.+onUndo :: TransientIO a -> TransientIO a -> TransientIO a+onUndo x y= onBack x (\() -> y)++++-- | Register an undo action to be executed when backtracking. The first+-- parameter is a "witness" whose data type is used to uniquely identify this+-- backtracking action. The value of the witness parameter is not used.+--+{-# NOINLINE registerUndo #-}+registerBack :: (Typeable b, Show b) => b -> TransientIO a -> TransientIO a+registerBack witness f = Transient $ do+ cont@(EventF _ _ x _ _ _ _ _ _ _ _ _) <- get -- !!> "backregister"++ md <- getData `asTypeOf` (Just <$> backStateOf witness)++ case md of+ Just (bss@(Backtrack b (bs@((EventF _ _ x' _ _ _ _ _ _ _ _ _):_)))) ->+ when (isNothing b) $ do+ addrx <- addr x+ addrx' <- addr x' -- to avoid duplicate backtracking points+ setData $ if addrx == addrx' then bss else Backtrack mwit (cont:bs)+ Nothing -> setData $ Backtrack mwit [cont]++ runTrans f+ where+ mwit= Nothing `asTypeOf` (Just witness)+ addr x = liftIO $ return . hashStableName =<< (makeStableName $! x)+++registerUndo :: TransientIO a -> TransientIO a+registerUndo f= registerBack () f++-- XXX Should we enforce retry of the same track which is being undone? If the+-- user specifies a different track would it make sense?+--+-- | For a given undo track id, stop executing more backtracking actions and+-- resume normal execution in the forward direction. Used inside an undo+-- action.+--+forward :: (Typeable b, Show b) => b -> TransIO ()+forward reason= Transient $ do+ Backtrack _ stack <- getData `onNothing` (backStateOf reason)+ setData $ Backtrack(Nothing `asTypeOf` Just reason) stack+ return $ Just ()++-- | 'forward' for the default undo track; equivalent to @forward ()@.+retry= forward ()++-- | Abort finish. Stop executing more finish actions and resume normal+-- execution. Used inside 'onFinish' actions.+--+noFinish= forward (FinishReason Nothing)++-- | Start the undo process for the given undo track id. Performs all the undo+-- actions registered till now in reverse order. An undo action can use+-- 'forward' to stop the undo process and resume forward execution. If there+-- are no more undo actions registered execution stops and a 'stop' action is+-- returned.+--+back :: (Typeable b, Show b) => b -> TransientIO a+back reason = Transient $ do+ bs <- getData `onNothing` backStateOf reason -- !!>"GOBACK"+ goBackt bs++ where++ goBackt (Backtrack _ [] )= return Nothing -- !!> "END"+ goBackt (Backtrack b (stack@(first : bs)) )= do+ setData $ Backtrack (Just reason) stack++ mr <- runClosure first -- !> "RUNCLOSURE"++ Backtrack back _ <- getData `onNothing` backStateOf reason+ -- !> "END RUNCLOSURE"++ case mr of+ Nothing -> return empty -- !> "END EXECUTION"+ Just x -> case back of+ Nothing -> runContinuation first x -- !> "FORWARD EXEC"+ justreason -> goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back)++backStateOf :: (Monad m, Show a, Typeable a) => a -> m (Backtrack a)+backStateOf reason= return $ Backtrack (Nothing `asTypeOf` (Just reason)) []++-- | 'back' for the default undo track; equivalent to @back ()@.+--+undo :: TransIO a+undo= back ()+++------ finalization++newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show)++-- | Clear all finish actions registered till now.+initFinish= backCut (FinishReason Nothing)++-- | Register an action that to be run when 'finish' is called. 'onFinish' can+-- be used multiple times to register multiple actions. Actions are run in+-- reverse order. Used in infix style.+--+onFinish :: ((Maybe SomeException) ->TransIO ()) -> TransIO ()+onFinish f= onFinish' (return ()) f+++-- | Run the action specified in the first parameter and register the second+-- parameter as a finish action to be run when 'finish' is called. Used in+-- infix style.+--+onFinish' ::TransIO a ->((Maybe SomeException) ->TransIO a) -> TransIO a+onFinish' proc f= proc `onBack` \(FinishReason reason) ->+ f reason+++-- | Execute all the finalization actions registered up to the last+-- 'initFinish', in reverse order. Either an exception or 'Nothing' can be+-- passed to 'finish'. The argument passed is made available in the 'onFinish'+-- actions invoked.+--+finish :: Maybe SomeException -> TransIO a+finish reason= back (FinishReason reason)++++-- | trigger finish when the stream of data ends+checkFinalize v=+ case v of+ SDone -> stop+ SLast x -> return x+ SError e -> back e+ SMore x -> return x++------ exceptions ---+--+-- | Install an exception handler. On exception, currently installed handlers+-- are executed in reverse (i.e. last in first out) order. Note that multiple+-- handlers can be installed for the same exception type.+--+onException :: Exception e => (e -> TransIO ()) -> TransIO ()+onException exc= return () `onException'` exc+++onException' :: Exception e => TransIO a -> (e -> TransIO a) -> TransIO a+onException' mx f= onAnyException mx $ \e ->+ case fromException e of+ Nothing -> empty+ Just e' -> f e'+ where+ onAnyException :: TransIO a -> (SomeException ->TransIO a) -> TransIO a+ onAnyException mx f= ioexp `onBack` f+ + ioexp = Transient $ do + st <- get+ (mx,st') <- liftIO $ (runStateT (do++ case event st of + Nothing -> do+ r <- runTrans mx -- !> "mx"++ modify $ \s -> s{event= Just $ unsafeCoerce r}+ + runCont st -- !> "runcont"+ was <- getData `onNothing` return NoRemote+ when (was /= WasRemote) $ setData WasParallel++ return Nothing+ Just r -> do+ modify $ \s -> s{event=Nothing} + return $ unsafeCoerce r) st)+ `catch` \(e ::SomeException) -> runStateT ( runTrans $ back e ) st+ put st'+ return mx++-- | Delete all the exception handlers registered till now.+cutExceptions= backCut (undefined :: SomeException)++-- | Used inside an exception handler. Stop executing any further exception+-- handlers and resume normal execution from this point on.+--+continue = forward (undefined :: SomeException)+++catcht mx exc= sandbox $ do+ cutExceptions+ onException' mx exc+ where+ sandbox mx= do+ exState <- getState <|> backStateOf (undefined :: SomeException)+ mx+ <*** setState exState++throwt :: Exception e => e -> TransIO a+throwt= back . toException+
src/Transient/Logged.hs view
@@ -1,230 +1,230 @@------------------------------------------------------------------------------ --- --- Module : Transient.Logged --- Copyright : --- License : MIT --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | The 'logged' primitive is used to save the results of the subcomputations --- of a transient computation (including all its threads) in a log buffer. At --- any point, a 'suspend' or 'checkpoint' can be used to save the accumulated --- log on a persistent storage. A 'restore' reads the saved logs and resumes --- the computation from the saved checkpoint. On resumption, the saved results --- are used for the computations which have already been performed. The log --- contains purely application level state, and is therefore independent of the --- underlying machine architecture. The saved logs can be sent across the wire --- to another machine and the computation can then be resumed on that machine. --- We can also save the log to gather diagnostic information, especially in --- 'finish' blocks. --- --- The following example illustrates the APIs. In its first run 'suspend' saves --- the state in a directory named @logs@ and exits, in the second run it --- resumes from that point and then stops at the 'checkpoint', in the third run --- it resumes from the checkpoint and then finishes. --- --- @ --- main= keep $ restore $ do --- r <- logged $ choose [1..10 :: Int] --- logged $ liftIO $ print (\"A",r) --- suspend () --- logged $ liftIO $ print (\"B",r) --- checkpoint --- liftIO $ print (\"C",r) --- @ ------------------------------------------------------------------------------ -{-# LANGUAGE CPP,ExistentialQuantification, FlexibleInstances, ScopedTypeVariables, UndecidableInstances #-} -module Transient.Logged( -Loggable, logged, received, param - -#ifndef ghcjs_HOST_OS -, suspend, checkpoint, restore -#endif -) where - -import Data.Typeable -import Unsafe.Coerce -import Transient.Base -import Transient.Internals(Loggable) -import Transient.Indeterminism(choose) -import Transient.Internals(onNothing,reads1,IDynamic(..),Log(..),LogElem(..),RemoteStatus(..),StateIO) -import Control.Applicative -import Control.Monad.IO.Class -import System.Directory -import Control.Exception -import Control.Monad - - - -#ifndef ghcjs_HOST_OS -import System.Random -#endif - - - -#ifndef ghcjs_HOST_OS -logs= "logs/" - --- | Reads the saved logs from the @logs@ subdirectory of the current --- directory, restores the state of the computation from the logs, and runs the --- computation. The log files are removed after the state has been restored. --- -restore :: TransIO a -> TransIO a -restore proc= do - liftIO $ createDirectory logs `catch` (\(e :: SomeException) -> return ()) - list <- liftIO $ getDirectoryContents logs - `catch` (\(e::SomeException) -> return []) - if length list== 2 then proc else do - - let list'= filter ((/=) '.' . head) list - file <- choose list' -- !> list' - - logstr <- liftIO $ readFile (logs++file) - let log= length logstr `seq` read' logstr - - log `seq` setData (Log True (reverse log) log) - liftIO $ remove $ logs ++ file - proc - where - read'= fst . head . reads1 - - remove f= removeFile f `catch` (\(e::SomeException) -> remove f) - - - --- | Saves the logged state of the current computation that has been --- accumulated using 'logged', and then 'exit's using the passed parameter as --- the exit code. Note that all the computations before a 'suspend' must be --- 'logged' to have a consistent log state. The logs are saved in the @logs@ --- subdirectory of the current directory. Each thread's log is saved in a --- separate file. --- -suspend :: Typeable a => a -> TransIO a -suspend x= do - Log recovery _ log <- getData `onNothing` return (Log False [] []) - if recovery then return x else do - logAll log - exit x - --- | Saves the accumulated logs of the current computation, like 'suspend', but --- does not exit. -checkpoint :: TransIO () -checkpoint = do - Log recovery _ log <- getData `onNothing` return (Log False [] []) - if recovery then return () else logAll log - - -logAll log= do - newlogfile <- liftIO $ (logs ++) <$> replicateM 7 (randomRIO ('a','z')) - liftIO $ writeFile newlogfile $ show log - :: TransIO () - -#endif - - -fromIDyn :: Loggable a => IDynamic -> a -fromIDyn (IDynamic x)=r where r= unsafeCoerce x -- !> "coerce" ++ " to type "++ show (typeOf r) - -fromIDyn (IDyns s)=r `seq`r where r= read s -- !> "read " ++ s ++ " to type "++ show (typeOf r) - - - -toIDyn x= IDynamic x - - - --- | Run the computation, write its result in a log in the parent computation --- and return the result. If the log already contains the result of this --- computation ('restore'd from previous saved state) then that result is used --- instead of running the computation again. --- --- 'logged' can be used for computations inside a 'logged' computation. Once --- the parent computation is finished its internal (subcomputation) logs are --- discarded. --- -logged :: Loggable a => TransientIO a -> TransientIO a -logged mx = Transient $ do - Log recover rs full <- getData `onNothing` return ( Log False [][]) - runTrans $ - case (recover ,rs) of -- !> ("logged enter",recover,rs) of - (True, Var x: rs') -> do - setData $ Log True rs' full - return $ fromIDyn x --- !> ("Var:", x) - - (True, Exec:rs') -> do - setData $ Log True rs' full - mx --- !> "Exec" - - (True, Wait:rs') -> do - setData (Log True rs' full) -- !> "Wait" - empty - - _ -> do --- let add= Exec: full - setData $ Log False (Exec : rs) (Exec: full) -- !> ("setLog False", Exec:rs) - - r <- mx <** ( do -- when p1 <|> p2, to avoid the re-execution of p1 at the - -- recovery when p1 is asynchronous - r <- getSData <|> return NoRemote - case r of - WasParallel -> --- let add= Wait: full - setData $ Log False (Wait: rs) (Wait: full) - _ -> return ()) - - Log recoverAfter lognew _ <- getData `onNothing` return ( Log False [][]) - let add= Var (toIDyn r): full - if recoverAfter && (not $ null lognew) -- !> ("recoverAfter", recoverAfter) - then (setData $ Log True lognew (reverse lognew ++ add) ) - -- !> ("recover",reverse lognew ,add) - else if recoverAfter && (null lognew) then - setData $ Log False [] add - else - (setData $ Log False (Var (toIDyn r):rs) add) -- !> ("restore", (Var (toIDyn r):rs)) - return r - - - - --------- parsing the log for API's - -received :: Loggable a => a -> TransIO () -received n=Transient $ do - Log recover rs full <- getData `onNothing` return ( Log False [][]) - case rs of - [] -> return Nothing - Var (IDyns s):t -> if s == show1 n - then do - setData $ Log recover t full - return $ Just () - else return Nothing - _ -> return Nothing - where - show1 x= if typeOf x == typeOf "" then unsafeCoerce x else show x - -param :: Loggable a => TransIO a -param= res where - res= Transient $ do - Log recover rs full <- getData `onNothing` return ( Log False [][]) - case rs of - [] -> return Nothing - Var (IDynamic v):t ->do - setData $ Log recover t full - return $ cast v - Var (IDyns s):t -> do - let mr = reads1 s `asTypeOf` type1 res - - case mr of - [] -> return Nothing - (v,r):_ -> do - setData $ Log recover t full - return $ Just v - _ -> return Nothing - - where - type1 :: TransIO a -> [(a,String)] - type1= error "type1: typelevel" +-----------------------------------------------------------------------------+--+-- Module : Transient.Logged+-- Copyright :+-- License : MIT+--+-- Maintainer : agocorona@gmail.com+-- Stability :+-- Portability :+--+-- | The 'logged' primitive is used to save the results of the subcomputations+-- of a transient computation (including all its threads) in a log buffer. At+-- any point, a 'suspend' or 'checkpoint' can be used to save the accumulated+-- log on a persistent storage. A 'restore' reads the saved logs and resumes+-- the computation from the saved checkpoint. On resumption, the saved results+-- are used for the computations which have already been performed. The log+-- contains purely application level state, and is therefore independent of the+-- underlying machine architecture. The saved logs can be sent across the wire+-- to another machine and the computation can then be resumed on that machine.+-- We can also save the log to gather diagnostic information, especially in+-- 'finish' blocks.+--+-- The following example illustrates the APIs. In its first run 'suspend' saves+-- the state in a directory named @logs@ and exits, in the second run it+-- resumes from that point and then stops at the 'checkpoint', in the third run+-- it resumes from the checkpoint and then finishes.+--+-- @+-- main= keep $ restore $ do+-- r <- logged $ choose [1..10 :: Int]+-- logged $ liftIO $ print (\"A",r)+-- suspend ()+-- logged $ liftIO $ print (\"B",r)+-- checkpoint+-- liftIO $ print (\"C",r)+-- @+-----------------------------------------------------------------------------+{-# LANGUAGE CPP,ExistentialQuantification, FlexibleInstances, ScopedTypeVariables, UndecidableInstances #-}+module Transient.Logged(+Loggable, logged, received, param++#ifndef ghcjs_HOST_OS+, suspend, checkpoint, restore+#endif+) where++import Data.Typeable+import Unsafe.Coerce+import Transient.Base+import Transient.Internals(Loggable)+import Transient.Indeterminism(choose)+import Transient.Internals(onNothing,reads1,IDynamic(..),Log(..),LogElem(..),RemoteStatus(..),StateIO)+import Control.Applicative+import Control.Monad.IO.Class+import System.Directory+import Control.Exception+import Control.Monad++++#ifndef ghcjs_HOST_OS+import System.Random+#endif++++#ifndef ghcjs_HOST_OS+logs= "logs/"++-- | Reads the saved logs from the @logs@ subdirectory of the current+-- directory, restores the state of the computation from the logs, and runs the+-- computation. The log files are removed after the state has been restored.+--+restore :: TransIO a -> TransIO a+restore proc= do+ liftIO $ createDirectory logs `catch` (\(e :: SomeException) -> return ())+ list <- liftIO $ getDirectoryContents logs+ `catch` (\(e::SomeException) -> return [])+ if length list== 2 then proc else do++ let list'= filter ((/=) '.' . head) list+ file <- choose list' -- !> list'++ logstr <- liftIO $ readFile (logs++file)+ let log= length logstr `seq` read' logstr++ log `seq` setData (Log True (reverse log) log)+ liftIO $ remove $ logs ++ file+ proc+ where+ read'= fst . head . reads1++ remove f= removeFile f `catch` (\(e::SomeException) -> remove f)++++-- | Saves the logged state of the current computation that has been+-- accumulated using 'logged', and then 'exit's using the passed parameter as+-- the exit code. Note that all the computations before a 'suspend' must be+-- 'logged' to have a consistent log state. The logs are saved in the @logs@+-- subdirectory of the current directory. Each thread's log is saved in a+-- separate file.+--+suspend :: Typeable a => a -> TransIO a+suspend x= do+ Log recovery _ log <- getData `onNothing` return (Log False [] [])+ if recovery then return x else do+ logAll log+ exit x++-- | Saves the accumulated logs of the current computation, like 'suspend', but+-- does not exit.+checkpoint :: TransIO ()+checkpoint = do+ Log recovery _ log <- getData `onNothing` return (Log False [] [])+ if recovery then return () else logAll log+++logAll log= do+ newlogfile <- liftIO $ (logs ++) <$> replicateM 7 (randomRIO ('a','z'))+ liftIO $ writeFile newlogfile $ show log+ :: TransIO ()++#endif+++fromIDyn :: Loggable a => IDynamic -> a+fromIDyn (IDynamic x)=r where r= unsafeCoerce x -- !> "coerce" ++ " to type "++ show (typeOf r)++fromIDyn (IDyns s)=r `seq`r where r= read s -- !> "read " ++ s ++ " to type "++ show (typeOf r)++++toIDyn x= IDynamic x++++-- | Run the computation, write its result in a log in the parent computation+-- and return the result. If the log already contains the result of this+-- computation ('restore'd from previous saved state) then that result is used+-- instead of running the computation again.+--+-- 'logged' can be used for computations inside a 'logged' computation. Once+-- the parent computation is finished its internal (subcomputation) logs are+-- discarded.+--+logged :: Loggable a => TransientIO a -> TransientIO a+logged mx = Transient $ do+ Log recover rs full <- getData `onNothing` return ( Log False [][])+ runTrans $+ case (recover ,rs) of -- !> ("logged enter",recover,rs) of+ (True, Var x: rs') -> do+ setData $ Log True rs' full+ return $ fromIDyn x+-- !> ("Var:", x)++ (True, Exec:rs') -> do+ setData $ Log True rs' full+ mx+-- !> "Exec"++ (True, Wait:rs') -> do+ setData (Log True rs' full) -- !> "Wait"+ empty++ _ -> do+-- let add= Exec: full+ setData $ Log False (Exec : rs) (Exec: full) -- !> ("setLog False", Exec:rs)++ r <- mx <** ( do -- when p1 <|> p2, to avoid the re-execution of p1 at the+ -- recovery when p1 is asynchronous+ r <- getSData <|> return NoRemote+ case r of+ WasParallel ->+-- let add= Wait: full+ setData $ Log False (Wait: rs) (Wait: full)+ _ -> return ())++ Log recoverAfter lognew _ <- getData `onNothing` return ( Log False [][])+ let add= Var (toIDyn r): full+ if recoverAfter && (not $ null lognew) -- !> ("recoverAfter", recoverAfter)+ then (setData $ Log True lognew (reverse lognew ++ add) )+ -- !> ("recover",reverse lognew ,add)+ else if recoverAfter && (null lognew) then+ setData $ Log False [] add+ else+ (setData $ Log False (Var (toIDyn r):rs) add) -- !> ("restore", (Var (toIDyn r):rs))+ return r+++++-------- parsing the log for API's++received :: Loggable a => a -> TransIO ()+received n=Transient $ do+ Log recover rs full <- getData `onNothing` return ( Log False [][])+ case rs of+ [] -> return Nothing+ Var (IDyns s):t -> if s == show1 n+ then do+ setData $ Log recover t full+ return $ Just ()+ else return Nothing+ _ -> return Nothing+ where+ show1 x= if typeOf x == typeOf "" then unsafeCoerce x else show x++param :: Loggable a => TransIO a+param= res where+ res= Transient $ do+ Log recover rs full <- getData `onNothing` return ( Log False [][])+ case rs of+ [] -> return Nothing+ Var (IDynamic v):t ->do+ setData $ Log recover t full+ return $ cast v+ Var (IDyns s):t -> do+ let mr = reads1 s `asTypeOf` type1 res++ case mr of+ [] -> return Nothing+ (v,r):_ -> do+ setData $ Log recover t full+ return $ Just v+ _ -> return Nothing++ where+ type1 :: TransIO a -> [(a,String)]+ type1= error "type1: typelevel"
+ tests/TestSuite.hs view
@@ -0,0 +1,52 @@+#!/usr/bin/env ./execthirdline.sh+-- development+-- set -e && docker run -it -v /c/Users/magocoal/OneDrive/Haskell/devel:/devel agocorona/transient:05-02-2017 bash -c "runghc -j2 -isrc -i/devel/transient/src /devel/transient/tests/$1 $2 $3 $4"++-- compile and run within a docker image+-- set -e && executable=`basename -s .hs ${1}` && docker run -it -v $(pwd):/work agocorona/transient:05-02-2017 bash -c "ghc /work/${1} && /work/${executable} ${2} ${3}"+++import Transient.Base+import Transient.EVars+import Transient.Indeterminism+import Transient.Backtrack+import System.Exit+import Data.Monoid+import Control.Applicative+import Control.Monad.IO.Class+import System.Random+import Control.Concurrent+import Control.Exception.Base+import Data.List+++instance Monoid Int where+ mempty= 0+ mappend = (+)++main= do+ keep $ do+ let genElem :: a -> TransIO a+ genElem x= do+ isasync <- liftIO randomIO+ delay <- liftIO $ randomRIO (1, 1000)+ liftIO $ threadDelay delay+ if isasync then async $ return x else return x++ liftIO $ putStrLn "--Testing thread control + Monoid + Applicative + async + indetermism---"++ collect 100 $ do+ i <- threads 0 $ choose [1..100]+ nelems <- liftIO $ randomRIO (1, 10) :: TransIO Int+ nthreads <- liftIO $ randomRIO (1,nelems)+ r <- threads nthreads $ foldr (<>) mempty $map genElem [1..nelems]+ assert (r == sum[1..nelems]) $ return ()++ liftIO $ putStrLn "--------------checking parallel execution, Alternative, events --------"+ ev <- newEVar+ r <- collect 3 $ readEVar ev <|> ((choose [1..3] >>= writeEVar ev) >> stop)+ assert (sort r== [1,2,3]) $ return ()+ liftIO $ print "SUCCESS"+ exit ()++ exitSuccess
transient.cabal view
@@ -1,56 +1,79 @@-name: transient -version: 0.5.5 -author: Alberto G. Corona -extra-source-files: - ChangeLog.md README.md -maintainer: agocorona@gmail.com -cabal-version: >=1.10 -build-type: Simple -license: MIT -license-file: LICENSE -homepage: http://www.fpcomplete.com/user/agocorona -bug-reports: https://github.com/agocorona/transient/issues -synopsis: composing programs with multithreading, events and distributed computing -description: See <http://github.com/agocorona/transient> - Distributed primitives are in the transient-universe package. Web primitives are in the axiom package. -category: Control, Concurrency -data-dir: "" - -flag debug - description: Enable debugging outputs - default: False - manual: True - -library - -- Note: `stack sdist/upload` will add missing bounds (via "pvp-bounds: both") in `build-depends` - -- support GHC 7.10.3 and later; lower bounds below denote GHC 7.10.3's bundled versions - build-depends: base >= 4.8.1 && < 5 - , containers >= 0.5.6 - , transformers >= 0.4.2 - , time >= 1.5 - , directory >= 1.2.2 - , bytestring >= 0.10.6 - - -- libraries not bundled w/ GHC - , mtl - , stm - , random - - exposed-modules: Transient.Backtrack - Transient.Base - Transient.EVars - Transient.Indeterminism - Transient.Internals - Transient.Logged - - exposed: True - buildable: True - exposed: True - default-language: Haskell2010 - hs-source-dirs: src . - if flag(debug) - cpp-options: -DDEBUG - -source-repository head - type: git - location: https://github.com/agocorona/transient +name: transient+version: 0.5.6+author: Alberto G. Corona+extra-source-files:+ ChangeLog.md README.md+maintainer: agocorona@gmail.com+cabal-version: >=1.10+build-type: Simple+license: MIT+license-file: LICENSE+homepage: http://www.fpcomplete.com/user/agocorona+bug-reports: https://github.com/agocorona/transient/issues+synopsis: composing programs with multithreading, events and distributed computing+description: See <http://github.com/agocorona/transient>+ Distributed primitives are in the transient-universe package. Web primitives are in the axiom package.+category: Control, Concurrency+data-dir: ""++flag debug+ description: Enable debugging outputs+ default: False+ manual: True++library+ -- Note: `stack sdist/upload` will add missing bounds (via "pvp-bounds: both") in `build-depends`+ -- support GHC 7.10.3 and later; lower bounds below denote GHC 7.10.3's bundled versions+ build-depends: base >= 4.8.1 && < 5+ , containers >= 0.5.6+ , transformers >= 0.4.2+ , time >= 1.5+ , directory >= 1.2.2+ , bytestring >= 0.10.6++ -- libraries not bundled w/ GHC+ , mtl+ , stm+ , random++ exposed-modules: Transient.Backtrack+ Transient.Base+ Transient.EVars+ Transient.Indeterminism+ Transient.Internals+ Transient.Logged++ exposed: True+ buildable: True+ exposed: True+ default-language: Haskell2010+ hs-source-dirs: src .+ if flag(debug)+ cpp-options: -DDEBUG++source-repository head+ type: git+ location: https://github.com/agocorona/transient++test-suite test-transient++ if !impl(ghcjs >=0.1)+ build-depends:+ base >= 4.8.1 && < 5+ , containers >= 0.5.6+ , transformers >= 0.4.2+ , time >= 1.5+ , directory >= 1.2.2+ , bytestring >= 0.10.6++ -- libraries not bundled w/ GHC+ , mtl+ , stm+ , random+ type: exitcode-stdio-1.0+ main-is: TestSuite.hs+ build-depends:+ base >4+ default-language: Haskell2010+ hs-source-dirs: tests src .+ ghc-options: -threaded -rtsopts