transient 0.4.4.1 → 0.5.1

raw patch · 12 files changed

+2170/−1792 lines, 12 filesdep +bytestringsetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: bytestring

API changes (from Hackage documentation)

Files

• ChangeLog.md +0/−0
• LICENSE +17/−17
• README.md +45/−0
• Setup.hs +2/−2
• src/Transient/Backtrack.hs +141/−139
• src/Transient/Base.hs +49/−35
• src/Transient/EVars.hs +84/−86
• src/Transient/Indeterminism.hs +132/−137
• src/Transient/Internals.hs +1418/−1070
• src/Transient/Logged.hs +227/−179
• src/Transient/Stream/Resource.hs +0/−76
• transient.cabal +55/−51

@@ -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.

@@ -0,0 +1,45 @@+![](https://raw.githubusercontent.com/agocorona/transient/master/logo.png)
+
+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)
+
+transient
+=========
+
+One of the biggest 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 `<>` can be used for concurrency, the operator `<|>` can be used for 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).

@@ -1,2 +1,2 @@-import Distribution.Simple-main = defaultMain+import Distribution.Simple
+main = defaultMain

@@ -10,11 +10,11 @@ onBack, back, forward, backCut,registerBack,
 
 -- * finalization primitives
-finish, onFinish, onFinish' ,initFinish , noFinish, killOnFinish ,checkFinalize , FinishReason
+finish, onFinish, onFinish' ,initFinish , noFinish,checkFinalize , FinishReason
 ) where
 
-import Transient.Base
-import Transient.Internals(EventF(..),killChildren,onNothing,runClosure,runContinuation)
+import Transient.Internals
+
 import Data.Typeable
 import Control.Applicative
 import Control.Monad.State
@@ -23,141 +23,143 @@ import Control.Exception
 import Control.Concurrent.STM hiding (retry)
 import Data.Maybe
-
-data Backtrack b= Show b =>Backtrack{backtracking :: Maybe b
-                                    ,backStack :: [EventF] }
-                                    deriving Typeable
-
--- | assures that backtracking will not go further back
-backCut :: (Typeable reason, Show reason) => reason -> TransientIO ()
-backCut reason= Transient $ do
-     delData $ Backtrack (Just reason)  []
-     return $ Just ()
-
-undoCut ::  TransientIO ()
-undoCut = backCut ()
-
--- | the second parameter will be executed when backtracking
-{-# 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
-
-onUndo ::  TransientIO a -> TransientIO a -> TransientIO a
-onUndo x y= onBack x (\() -> y)
-
-
--- | register an action that will be executed when backtracking
-{-# 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
-
--- | restart the flow forward from this point on
-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 ()
-
-retry= forward ()
-
-noFinish= forward (FinishReason Nothing)
-
--- | execute backtracking. It execute the registered actions in reverse order.
 --
--- If the backtracking flag is changed the flow proceed  forward from that point on.
+--data Backtrack b= Show b =>Backtrack{backtracking :: Maybe b
+--                                    ,backStack :: [EventF] }
+--                                    deriving Typeable
 --
--- If the backtrack stack is finished or undoCut executed, `undo` will stop.
-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)
-
-backStateOf :: (Monad m, Show a, Typeable a) => a -> m (Backtrack a)
-backStateOf reason= return $ Backtrack (Nothing `asTypeOf` (Just reason)) []
-
-undo ::  TransIO a
-undo= back ()
-
------- finalization
-
-newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show)
-
--- | initialize the event variable for finalization.
--- all the following computations in different threads will share it
--- it also isolate this event from other branches that may have his own finish variable
-initFinish= backCut (FinishReason Nothing)
-
--- | set a computation to be called when the finish event happens
-onFinish :: ((Maybe SomeException) ->TransIO ()) -> TransIO ()
-onFinish f= onFinish' (return ()) f
-
-
--- | set a computation to be called when the finish event happens this only apply for
-onFinish' ::TransIO a ->((Maybe SomeException) ->TransIO a) -> TransIO a
-onFinish' proc f= proc `onBack`   \(FinishReason reason) ->
-    f reason
-
-
--- | trigger the event, so this closes all the resources
-finish :: Maybe SomeException -> TransIO a
-finish reason= back (FinishReason reason)
-
-
--- | kill all the processes generated by the parameter when finish event occurs
-killOnFinish comp= do
-   chs <- liftIO $ newTVarIO []
-   onFinish $ const $ liftIO $ killChildren chs   -- !> "killOnFinish event"
-   r <- comp
-   modify $ \ s -> s{children= chs}
-   return r
-
--- | trigger finish when the stream of data ends
-checkFinalize v=
-           case v of
-              SDone ->  finish Nothing >> stop
-              SLast x ->  return x
-              SError e -> liftIO ( print e) >> finish  Nothing >> stop
-              SMore x -> return x
+--
+--
+---- | assures that backtracking will not go further back
+--backCut :: (Typeable reason, Show reason) => reason -> TransientIO ()
+--backCut reason= Transient $ do
+--     delData $ Backtrack (Just reason)  []
+--     return $ Just ()
+--
+--undoCut ::  TransientIO ()
+--undoCut = backCut ()
+--
+---- | the second parameter will be executed when backtracking
+--{-# 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
+--
+--onUndo ::  TransientIO a -> TransientIO a -> TransientIO a
+--onUndo x y= onBack x (\() -> y)
+--
+--
+---- | register an action that will be executed when backtracking
+--{-# 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
+--
+---- | restart the flow forward from this point on
+--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 ()
+--
+--retry= forward ()
+--
+--noFinish= forward (FinishReason Nothing)
+--
+---- | execute backtracking. It execute the registered actions in reverse order.
+----
+---- If the backtracking flag is changed the flow proceed  forward from that point on.
+----
+---- If the backtrack stack is finished or undoCut executed, `undo` will stop.
+--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)
+--
+--backStateOf :: (Monad m, Show a, Typeable a) => a -> m (Backtrack a)
+--backStateOf reason= return $ Backtrack (Nothing `asTypeOf` (Just reason)) []
+--
+--undo ::  TransIO a
+--undo= back ()
+--
+-------- finalization
+--
+--newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show)
+--
+---- | initialize the event variable for finalization.
+---- all the following computations in different threads will share it
+---- it also isolate this event from other branches that may have his own finish variable
+--initFinish= backCut (FinishReason Nothing)
+--
+---- | set a computation to be called when the finish event happens
+--onFinish :: ((Maybe SomeException) ->TransIO ()) -> TransIO ()
+--onFinish f= onFinish' (return ()) f
+--
+--
+---- | set a computation to be called when the finish event happens this only apply for
+--onFinish' ::TransIO a ->((Maybe SomeException) ->TransIO a) -> TransIO a
+--onFinish' proc f= proc `onBack`   \(FinishReason reason) ->
+--    f reason
+--
+--
+---- | trigger the event, so this closes all the resources
+--finish :: Maybe SomeException -> TransIO a
+--finish reason= back (FinishReason reason)
+--
+--
+---- | kill all the processes generated by the parameter when finish event occurs
+--killOnFinish comp= do
+--   chs <- liftIO $ newTVarIO []
+--   onFinish $ const $ liftIO $ killChildren chs   -- !> "killOnFinish event"
+--   r <- comp
+--   modify $ \ s -> s{children= chs}
+--   return r
+--
+---- | trigger finish when the stream of data ends
+--checkFinalize v=
+--           case v of
+--              SDone ->  finish Nothing >> stop
+--              SLast x ->  return x
+--              SError e -> liftIO ( print e) >> finish  Nothing >> stop
+--              SMore x -> return x

@@ -1,35 +1,49 @@-{-# LANGUAGE ScopedTypeVariables #-}------------------------------------------------------------------------------------ Module      :  Base--- Copyright   :--- License     :  GPL (Just (Version {versionBranch = [3], versionTags = []}))------ Maintainer  :  agocorona@gmail.com--- Stability   :--- Portability :------ | See http://github.com/agocorona/transient--------------------------------------------------------------------------------module Transient.Base(--TransIO(..), TransientIO-,keep, keep', stop-,option, input, exit-,async,waitEvents, spawn, parallel, sample-,react--,setData,getData,getSData,delData--, threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds--, (**>), (<**),(<***), (<|)--, StreamData(..)-,genId)--where---import    Transient.Internals+{-# LANGUAGE ScopedTypeVariables #-}
+-----------------------------------------------------------------------------
+--
+-- Module      :  Base
+-- Copyright   :
+-- License     :  GPL (Just (Version {versionBranch = [3], versionTags = []}))
+--
+-- Maintainer  :  agocorona@gmail.com
+-- Stability   :
+-- Portability :
+--
+-- | See http://github.com/agocorona/transient
+-----------------------------------------------------------------------------
+
+module Transient.Base(
+-- * The Monad
+TransIO(..), TransientIO
+-- * Running the monad
+,keep, keep', stop
+
+-- * input
+,option, input, exit
+
+-- * Asynchronous operations
+,async,waitEvents, spawn, parallel, sample
+,react
+
+-- * State management
+,setState, setData, getState, getSData,getData,delState,delData, modifyData,modifyState
+
+-- * Thread management
+, threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds
+
+-- * Additional operators
+, (**>), (<**),(<***), (<|)
+
+-- * exceptions
+
+,onException, cutExceptions, continue
+
+-- * Utilities
+, StreamData(..)
+,genId
+)
+
+where
+
+
+import    Transient.Internals

@@ -1,86 +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-   id <- genId-   ref <-liftIO  newBroadcastTChanIO-   return . Just $ EVar  ref---- | delete al the subscriptions for an evar.-cleanEVar :: EVar a -> TransIO ()-cleanEVar (EVar  ref1)= liftIO $ atomically $ do-    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
+
+
+

@@ -1,137 +1,132 @@------------------------------------------------------------------------------------ Module      :  Transient.Indeterminism--- Copyright   :--- License     :  GPL (Just (Version {versionBranch = [3], versionTags = []}))------ Maintainer  :  agocorona@gmail.com--- Stability   :--- Portability :------ | see <https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii>----------------------------------------------------------------------------------{-# LANGUAGE BangPatterns #-}-module Transient.Indeterminism (-choose, choose', collect, collect', group, groupByTime-) where--import Transient.Base-import Transient.Backtrack(checkFinalize)-import Transient.Internals(killChildren, EventF(..),hangThread)-import Data.IORef-import Control.Applicative-import Data.Monoid-import Control.Concurrent-import Data.Typeable-import Control.Monad.State-import Control.Concurrent.STM as STM-import GHC.Conc-import Data.Time.Clock----- | slurp a list of values and process them in parallel . To limit the number of processing--- threads, use `threads`-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----- | group the output of a possible multithreaded process 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 xs)-              else ((n', x:xs),Nothing)-    case mn of-      Nothing -> stop-      Just xs -> return xs---- | group result for a time interval, measured with `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 xs)-    case mn of-      Nothing -> stop-      Just xs -> return xs---- | alternative definition with more parallelism, as the composition of n `async` sentences-choose' :: [a] -> TransIO a-choose' xs = foldl (<|>) empty $ map (async . return) xs----- collect the results of a search done in parallel, usually initiated by--- `choose` .------ execute a process and get at least the first n solutions (they could be more).--- if the process end without finding the number of solutions requested, it return the found ones--- if he find the number of solutions requested, it kill the non-free threads of the process and return--- It works monitoring the solutions found and the number of active threads.--- If the first parameter is 0, collect will return all the results-collect ::  Int -> TransIO a -> TransIO [a]-collect n = collect' n 0.1 0---- | search also between two time intervals. If the first interval has passed and there is no result,---it stops.--- After the second interval, it stop unconditionally and return the current results.--- It also stops as soon as there are enough results specified in the first parameter.-collect' :: Int -> NominalDiffTime -> NominalDiffTime -> TransIO a -> TransIO [a]-collect' n t1 t2 search= hookedThreads $  do-  rv <- liftIO $ atomically $ newTVar (0,[])    -- !> "NEWMVAR"-  endflag <- liftIO $ newTVarIO False-  st <-  newPool-  t <- liftIO getCurrentTime-  let worker = do-        r <- search-        liftIO $ atomically $ do-            (n1,rs) <- readTVar rv-            writeTVar  rv (n1+1,r:rs)           -- !> "MODIFY"-        stop--      monitor=  freeThreads $ do-          xs <- async $ atomically $-                          do (n', xs) <- readTVar rv-                             ns <- readTVar $ children st-                             t' <- unsafeIOToSTM getCurrentTime-                             if-                               (n > 0 && n' >= n) ||-                                 (null ns && (diffUTCTime t' t > t1))    ||-                                 (t2 > 0 && diffUTCTime t' t > t2)-                                         -- !>  (diffUTCTime t' t, n', length ns)-                               then return xs else retry---          liftIO . killChildren $ children st--          return  xs--  monitor <|> worker-  where-  newPool  =  do-       chs <- liftIO $ newTVarIO []-       s <- get-       let s'=  s{children= chs}-       put s'-       return s'+-----------------------------------------------------------------------------
+--
+-- Module      :  Transient.Indeterminism
+-- Copyright   :
+-- License     :  GPL (Just (Version {versionBranch = [3], versionTags = []}))
+--
+-- 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
+
+
+-- | slurp a list of values and process them in parallel . To limit the number of processing
+-- threads, use `threads`
+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
+
+-- | alternative definition with more parallelism, as the composition of n `async` sentences
+choose' :: [a] -> TransIO a
+choose' xs = foldl (<|>) empty $ map (async . return) xs
+
+
+-- | group the output of a possible multithreaded process 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
+
+-- | group result for a time interval, measured with `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
+
+
+
+-- collect the results of a search done in parallel, usually initiated by
+-- `choose` .
+--
+-- execute a process and get at least the first n solutions (they could be more).
+-- if he find the number of solutions requested, it kill the non-free threads of the process and return
+collect ::  Int -> TransIO a -> TransIO [a]
+collect n = collect' n 0
+
+-- | search with a timeout
+-- After the the timeout, it stop unconditionally and return the current results.
+-- It also stops as soon as there are enough results specified in the first parameter.
+collect' :: Int -> Int -> TransIO a -> TransIO [a]
+collect' n t2 search= do
+
+  rv <- liftIO $ newEmptyMVar     -- !> "NEWMVAR"
+
+
+  results <- liftIO $ newIORef (0,[])
+
+  let worker = do
+        r <- search
+        liftIO $  putMVar rv r
+        stop
+
+      timer= if t2 > 0 then async $ threadDelay t2 >> readIORef results >>= return . snd
+                       else empty
+
+      monitor=  async loop
+
+          where
+          loop = do
+                     r <- takeMVar rv
+                     (n',rs) <- readIORef results
+                     let n''= n' +1
+                     writeIORef results  (n'',r:rs)
+
+                     t' <-  getCurrentTime
+                     if (n > 0 && n'' >= n)
+                       then readIORef results >>= return . snd else loop
+                 `catch` \(e :: BlockedIndefinitelyOnMVar) ->
+                                   readIORef results >>= return . snd
+
+
+  r <- oneThread $ monitor <|> worker <|> timer
+
+  return r
+
+

@@ -1,1070 +1,1418 @@-{-# LANGUAGE ScopedTypeVariables #-}------------------------------------------------------------------------------------ Module      :  Base--- Copyright   :--- License     :  GPL (Just (Version {versionBranch = [3], versionTags = []}))------ Maintainer  :  agocorona@gmail.com--- Stability   :--- Portability :------ | See http://github.com/agocorona/transient--- everithing in this module is exported in order to allow extensibility.-------------------------------------------------------------------------------{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE FlexibleContexts          #-}-{-# LANGUAGE FlexibleInstances         #-}-{-# LANGUAGE MultiParamTypeClasses     #-}-{-# LANGUAGE DeriveDataTypeable        #-}-{-# LANGUAGE Rank2Types        #-}--- show-module Transient.Internals where--- /show--import           Control.Applicative-import           Control.Monad.State-import           Data.Dynamic-import qualified Data.Map               as M-import           Data.Monoid-import           Debug.Trace-import           System.IO.Unsafe-import           Unsafe.Coerce-import           Control.Exception-import           Control.Concurrent-import           Control.Concurrent.STM-import           System.Mem.StableName-import           Data.Maybe-import           GHC.Conc-import           Data.List-import           Data.IORef-import           System.Environment-import           System.IO (hFlush,stdout)-import           System.Exit-{-# INLINE (!>) #-}-(!>) :: Show a => b -> a -> b-(!>) x y=  trace (show y) x-infixr 0 !>---data TransIO  x = Transient  {runTrans :: StateT EventF IO (Maybe x)}-type SData= ()--type EventId= Int--type TransientIO= TransIO--data EventF  = forall a b . EventF{meffects     :: Effects-                                  ,event       :: Maybe SData-                                  ,xcomp       :: TransIO a-                                  ,fcomp       :: [b -> TransIO b]-                                  ,mfData      :: M.Map TypeRep SData-                                  ,mfSequence  :: Int-                                  ,threadId    :: ThreadId-                                  ,freeTh      :: Bool-                                  ,parent      :: Maybe EventF-                                  ,children    :: TVar[EventF]-                                  ,maxThread   :: Maybe (IORef Int)-                                  }-                                  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--type StateIO= StateT EventF IO----- | run the transient computation with a blank state-runTransient :: TransIO x -> IO (Maybe x, EventF)-runTransient t= do-  th <- myThreadId-  let eventf0=  EventF baseEffects Nothing empty [] M.empty 0-          th False  Nothing  (unsafePerformIO $ newTVarIO []) Nothing---  runStateT (runTrans t) eventf0---- | run the transient computation with an state-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 _ _ x fs _ _  _ _  _ _ _)= runTrans $ do-      r <- unsafeCoerce x-      compose fs r---- | run the closure and the continuation using his own state data-runCont' cont= runStateT (runCont cont) cont---- | warning: radiactive untyped stuff. handle with care-getContinuations :: StateIO [a -> TransIO b]-getContinuations= do-  EventF _ _ _ 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 []= 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 _ _ x _ _ _ _ _ _ _ _) =  unsafeCoerce $ runTrans x----- | run the continuation (the 'f' in 'x >>= f') of the current bind operation-runContinuation ::  EventF -> a -> StateIO (Maybe b)-runContinuation (EventF _ _ _ fs _ _ _ _  _ _ _) =-   runTrans . (unsafeCoerce $ compose $  fs)---setContinuation :: TransIO a -> (a -> TransIO b) -> [c -> TransIO c] -> StateIO ()-setContinuation  b c fs =  do-    (EventF eff ev _ _ d e f g h i j) <- get-    put $ EventF eff ev b ( unsafeCoerce c: fs) d e f g h i j--withContinuation  c mx= do-    EventF eff ev f1 fs d e f g h i j <- get-    put $ EventF eff ev mx ( unsafeCoerce c: fs) d e f g h i j-    r <- mx-    restoreStack fs-    return r---- | run a chain of continuations. It is up to the programmer to assure by construction that---  each continuation type-check with the next, that the parameter type match the input of the first--- continuation.--- Normally this makes sense if it stop the current flow with `stop` after the invocation-runContinuations :: [a -> TransIO b] -> c -> TransIO d-runContinuations fs x= (compose $ unsafeCoerce fs)  x--instance   Functor TransIO where-  fmap f mx=  --   Transient $ fmap (fmap f) $ runTrans 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,[])   -- !> "NEWIOREF"--         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--restoreStack fs=-       modify $ \(EventF eff _ f _ a b c d parent children g1) ->-               EventF eff Nothing f fs a b c d parent children g1--readWithErr line=-     let [(v,left)] = readsPrec 0 line-     in (v   `seq` return [(v,left)])-                    `catch` (\(e::SomeException) ->-                      error ("read error of " ++ show( typeOf v) ++ " in: "++ "\""++line++"\""))---readsPrec' _= unsafePerformIO . readWithErr------ | dynamic serializable data for logging-data IDynamic= IDyns String | forall a.(Read a, Show a,Typeable 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---instance Alternative TransIO where-    empty = Transient $ return  Nothing-    (<|>) = mplus---data RemoteStatus=   WasRemote | WasParallel | NoRemote deriving (Typeable, Eq, Show)--instance MonadPlus TransIO where-    mzero= empty-    mplus  x y=  Transient $ do-         mx <- runTrans x                -- !!> "RUNTRANS11111"-         was <- getData `onNothing` return NoRemote-         if was== WasRemote              -- !> was-           then return Nothing-           else-                 case mx of-                     Nothing -> runTrans y      --  !!> "RUNTRANS22222"-                     justx -> return justx---- | a sinonym of empty that can be used in a monadic expression. it stop the--- computation and execute the next alternative computation (composed with `<|>`)-stop :: Alternative m => m stopped-stop= empty--class AdditionalOperators m where--    -- | executes the second operand even if the frist return empty.-    -- A normal imperative (monadic) sequence uses the operator (>>) which in the-    -- Transient monad does not execute the next operand if the previous one return empty.-    (**>) :: m a -> m b -> m b--    -- | forces the execution of the second operand even if the first stop. It does not execute-    -- the second operand as result of internal events occuring in the first operand.-    -- Return the first result-    (<**) :: m a -> m b -> m a--    atEnd' ::m a -> m b -> m a-    atEnd' = (<**)--    -- | forces the execution of the second operand even if the first stop. Return the first result. The second-    -- operand is executed also when internal events happens in the first operand and it returns something-    (<***) :: 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    -- !> "ma"-                  runTrans  mb        -- !> "mb"-                  return a--infixr 1  <***  ,  <**, **>------ | when the first operand is an asynchronous operation, the second operand is executed once (one single time)--- when the first completes his first asyncronous operation.------ This is useful for spawning asynchronous or distributed tasks that are singletons and that should start--- when the first one is set up.------ for example a streaming where the event receivers are acivated before the senders.--(<|) :: 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--instance Monoid a => Monoid (TransIO a) where-  mappend x y = mappend <$> x <*> y-  mempty= return mempty---- | set the current closure and continuation for the current statement-setEventCont ::   TransIO a -> (a -> TransIO b) -> StateIO EventF-setEventCont x f  = do--   st@(EventF eff e _ fs d n  r applic  ch rc bs)  <- get  -- !> "SET"-   let cont=  EventF eff e x ( unsafeCoerce f : fs) d n  r applic  ch rc bs-   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-   st@(EventF eff e _ fs d n  r nr  ch rc bs)  <- get     -- !> "reset"-   let f= \mx ->  case mx of-                       Nothing -> empty-                       Just x  -> (unsafeCoerce $ head fs)  x-   put $ EventF eff e (f mx) ( tailsafe fs) d n  r nr  ch rc bs-   return  id--tailsafe []=[]-tailsafe (x:xs)= xs----refEventCont= unsafePerformIO $ newIORef baseEffects--{-# INLINE baseEffects #-}-baseEffects :: Effects--baseEffects x  x' f' = do-            c <-setEventCont x'  f'-            mk <- runTrans x-            t <- resetEventCont mk c-            return (t,mk)--instance Monad TransIO where--      return  = pure--      x >>= f  = Transient $ do---            effects <- gets effects -- liftIO $ readIORef refEventCont-            (t,mk) <- baseEffects x x f-            t $ case mk of-                 Just k  ->  runTrans (f k)--                 Nothing ->  return Nothing----instance MonadTrans (Transient ) where---  lift mx = Transient $ mx >>= return . Just--instance MonadIO TransIO where-  liftIO x = Transient $ liftIO x >>= return . Just --     let x= liftIO io in x `seq` lift x----- * Threads--waitQSemB sem= atomicModifyIORef sem $ \n -> if n > 0 then(n-1,True) else (n,False)-signalQSemB sem= atomicModifyIORef sem  $ \n ->  (n + 1,())---- | set the maximun number of threads for a procedure. It is useful to limit the--- parallelization of transient code that uses `parallel` `spawn` and `waitEvents`-threads :: Int -> TransIO a -> TransIO a-threads n proc= Transient $ do-   msem <- gets maxThread-   sem <- liftIO $ newIORef n-   modify $ \s -> s{maxThread= Just sem}-   r <- runTrans proc-   modify $ \s -> s{maxThread = msem} -- restore it-   return r---- | delete all the previous childs generated by the expression taken as parameter and continue execution--- of the current thread.-oneThread :: TransIO a -> TransientIO a-oneThread comp=  do-   chs <- liftIO $ newTVarIO []-   r <-  comp-   modify $ \ s -> s{children= chs}-   killChilds-   return r---showThreads :: TransIO empty-showThreads= do-   st' <- gets (fromJust . parent)-   liftIO $ showTree 0 st'-   stop-   where-   toplevel st =-      case parent st of-        Nothing ->  st-        Just p -> toplevel p--   showThreads' n rchs= do-      chs <- atomically $ readTVar rchs-      mapM_ (showTree n) chs--   showTree n ch=  do-         putStr $ take n $ repeat  ' '-         print $ threadId ch-         showThreads' (n+4) $ children ch----- | add n threads to the limit of threads. If there is no limit, it set it-addThreads' :: Int -> TransIO ()-addThreads' n= Transient $ 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}-   return $ Just ()---- | assure that at least there are n threads available-addThreads n= Transient $ do-   msem <- gets maxThread-   case msem of-     Nothing -> return ()-     Just sem ->  liftIO $ modifyIORef sem $ \n' -> if n' > n then n' else  n-   return $ Just ()---getNonUsedThreads :: TransIO (Maybe Int)---getNonUsedThreads= Transient $ do---   msem <- gets maxThread---   case msem of---    Just sem -> liftIO $ Just <$> readIORef sem---    Nothing -> return Nothing----- | The threads generated in the process passed as parameter will not be killed by `kill*` primitives-freeThreads :: TransIO a -> TransIO a-freeThreads proc= Transient $ do-     st <- get-     put st{freeTh= True}-     r <- runTrans proc-     modify $ \s -> s{freeTh= freeTh st}-     return r---- | The threads will be killed when the parent thread dies. That is the default.--- This can be invoked to revert the effect of `freeThreads`-hookedThreads :: TransIO a -> TransIO a-hookedThreads proc= Transient $ do-     st <- get-     put st{freeTh= False}-     r <- runTrans proc-     modify $ \st -> st{freeTh= freeTh st}-     return r---- | kill all the child threads of the current thread-killChilds :: TransientIO()-killChilds= Transient $  do-   cont <- get-   liftIO $  killChildren $ children cont-   return $ Just ()---- * extensible state: session data management---- | Get the state data for the desired type if there is any.-getData ::  (MonadState EventF m,Typeable a) =>  m (Maybe a)-getData =  resp where- resp= gets mfData >>= \list  ->-    case M.lookup ( typeOf $ typeResp resp ) list  of-      Just x  -> return . Just $ unsafeCoerce x-      Nothing -> return Nothing- typeResp :: m (Maybe x) -> x- typeResp= undefined----- | getData specialized for the Transient monad. if Nothing, the--- monadic computation does not continue.------ If there is no such data, `getSData`  silently stop the computation.--- That may or may not be the desired behaviour.--- To make sure that this does not get unnoticed, use this construction:------ >  getSData <|> error "no data"------ To have the same semantics and guarantees than `get`, use a default value:------ > getInt= getSData <|> return (0 :: Int)------ The default value (0 in this case) has the same role than the initial value in a state monad.--- The difference is that you can define as many `get` as you need for all your data types.------ To distingish two data with the same types, use newtype definitions.-getSData ::  Typeable a => TransIO  a-getSData= Transient getData------ | set session data for this type. retrieved with getData or getSData--- Note that this is data in a state monad, that means that the update only affect downstream--- in the monad execution. it is not a global state neither a per user or per thread state--- it is a monadic state like the one of a state monad.-setData ::  (MonadState EventF m, Typeable a) => a -> m ()-setData  x=-  let t= typeOf x in  modify $ \st -> st{mfData= M.insert  t (unsafeCoerce x) (mfData st)}---delData :: ( MonadState EventF m,Typeable a) => a -> m ()-delData x=  modify $ \st -> st{mfData= M.delete (typeOf x ) (mfData st)}-----withSData ::  ( MonadState EventF m,Typeable a) => (Maybe a -> a) -> m ()---withSData f= modify $ \st -> st{mfData=---    let dat = mfData st---        mx= M.lookup typeofx dat---        mx'= case mx of Nothing -> Nothing; Just x -> unsafeCoerce x---        fx=  f mx'---        typeofx= typeOf $ typeoff f---    in  M.insert typeofx  (unsafeCoerce fx) dat}---    where---    typeoff :: (Maybe a -> a) -> a---    typeoff = undefined--------- | generator of identifiers that are unique withing 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= do-      n <- gets mfSequence-      return n--instance Read SomeException where-   readsPrec n str=-      let [(s , r)]= read str in [(SomeException $ ErrorCall s,r)]---- | async calls--data StreamData a=  SMore a | SLast a | SDone | SError SomeException deriving (Typeable, Show,Read)----- | variant of `parallel` that repeatedly executes the IO computation and kill the previously created childs------ It is useful in single threaded problems where each event discard the computations spawned by--- previous events-waitEvents ::   IO b -> TransIO b-waitEvents io= do-   mr <- parallel (SMore <$> io)-   case mr of-     SMore x -> return x-     SError e -> throw e----- Multithreaded version of `waitEvents` that do not kill the computations spawned by previous events-waitEvents' ::   IO b -> TransIO b-waitEvents' io= do-   mr <- parallel (SMore <$> io)-   case mr of-     SMore x -> return x-     SError e -> throw e---- | variant of `parallel` that execute the IO computation once, and kill the previous child threads-async  ::  IO b -> TransIO b-async io= do-   mr <- parallel  (SLast <$> io)-   case mr of-     SLast x -> return x-     SError e -> throw e---- | variant of waitEvents that spawn free threads. It is a little faster at the cost of no thread control-spawn ::  IO b -> TransIO b-spawn io= freeThreads $ do-   mr <- parallel (SMore <$>io)-   case mr of-     SMore x -> return x-     SError e -> throw e----- | executes an IO action each certain interval of time and return his value if it changes-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'------- |  return empty to the current thread, in new thread, execute the IO action,--- this IO action modify an internal buffer. then, executes the closure where `parallel` is located--- In this new execution, since the buffer is filled, `parallel` return the content of this buffer.--- Then it launch the continuation after it with this new value returned by the closure.------ If the maximum number of threads, set with `threads` has been reached  `parallel` perform--- the work sequentially, in the current thread.--- So `parallel` means that 'it can be parallelized if there are thread available'------ if there is a limitation of threads, when a thread finish, the counter of threads available--- is increased so another `parallel` can make use of it.------ The behaviour of `parallel` depend on `StreamData`; If `SMore`, `parallel` will excute again the--- IO action. with `SLast`, `SDone` and `SError`, `parallel` will not repeat the IO action anymore.-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 $ loop cont ioaction-            was <- getData `onNothing` return NoRemote-            when (was /= WasRemote) $ setData WasParallel-            return Nothing----- executes the IO action and then the continuation included in the first parameter-loop :: EventF -> IO (StreamData t) -> IO ()-loop (cont'@(EventF eff e x fs a b c d _ childs g))  rec  =  do-  chs <- liftIO $ newTVarIO []-  let cont = EventF eff e x fs a b c d (Just cont') chs g-      iocont dat= do-          runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat}-          return ()--      -- execute the IO computation and then the closure-continuation-      loop'= forkMaybe False cont $ do-         mdat <- threadDelay 0 >> rec `catch` \(e :: SomeException) -> return $ SError e-         case mdat of-             se@(SError _) ->  iocont se-             SDone ->          iocont SDone-             last@(SLast _) -> iocont last--             more@(SMore _) -> do-                  forkMaybe False cont $ iocont more-                  loop'-  loop'-  return ()-  where-  forkMaybe True cont proc = forkMaybe' True cont proc-  forkMaybe False cont proc = do-     dofork <- case maxThread cont of-                  Nothing -> return True-                  Just sem -> do-                    dofork <- waitQSemB sem-                    if dofork then  return True else return False-     forkMaybe' dofork cont proc--  forkMaybe' dofork cont proc=-         if dofork-            then  do-                 forkFinally1 (do-                     th <- myThreadId-                     hangThread cont' cont{threadId=th}  -- !>  "thread created: "++ show th-                     proc)-                     $ \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 _ ->  when(not $ freeTh cont')  $ do -- if was not a free thread-                             --  if parent is alive-                             --  then remove himself from the parent list (with free)-                             --  and pass his active children to his parent--                             th <- myThreadId-                             mparent <- free th cont-                             return ()-                               -- pass the active children to the parent---                             case mparent of---                              Nothing  ->  return()---                              Just parent -> atomically $ do---                                     chs' <- readTVar $ children cont---                                     chs  <- (readTVar $ children parent)---                                     writeTVar (children parent)$ chs ++ chs'---                                     return ()--                         case maxThread cont of-                           Just sem -> signalQSemB sem  --   !> "freed thread"-                           Nothing -> return ()-                 return ()-----            else proc  -- !> "NO THREAD"--forkFinally1 :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId-forkFinally1 action and_then =-  mask $ \restore ->  forkIO $ try (restore action) >>= and_then--free th env= do-  if isNothing $ parent env-   then  return Nothing  -- !!>  show th ++ " orphan"-   else do-    let msibling= fmap children $ parent env--    case msibling of-     Nothing -> return Nothing-     Just sibling  -> do-       found <- atomically $ do-                sbs <- readTVar sibling-                let (sbs', found) = drop [] th  sbs    -- !!> "search "++show th ++ " in " ++ show (map threadId sbs)-                when found $ writeTVar sibling sbs'    -- !> ("new list",map threadId sbs')-                return found-       if (not found && isJust (parent env))-         then free th $ fromJust $ parent env         -- !!> "toparent"-         else return $ Just env--   where-   drop processed th []= (processed,False)-   drop processed th (ev:evts)| th ==  threadId ev= (processed ++ evts, True)-                    | otherwise= drop (ev:processed) th evts--hangThread parent child = when(not $ freeTh parent) $ do-   let headpths= children parent-   atomically $ do-       ths <- readTVar headpths-       writeTVar headpths $  child:ths   -- !!>  "thread added: "++ show (threadId child)---- | kill  all the child threads associated with the continuation context-killChildren childs  = do----     forkIO $ do-        ths <- atomically $ do-           ths <- readTVar childs-           writeTVar childs []-           return ths---        mapM_ killChildren ths       -- recursive not needed, event handlers do it--        mapM_ (killThread . threadId) ths   -- !!> ("KILLEVENT " ++ show (map threadId ths) ++---                                                        if length ths <20 then ""---                                                          else error "long list of threads" )---     return ()---type EventSetter eventdata response= (eventdata ->  IO response) -> IO ()-type ToReturn  response=  IO response---- | deinvert an event handler.------ The first parameter is the setter of the event handler  to be--- deinverted. Usually it is the primitive provided by a framework to set an event handler------ the second parameter is the value to return to the event handler. Usually it is `return()`------ it configures the event handler by calling the setter of the event--- handler with the current continuation-react-  :: Typeable eventdata-  => EventSetter eventdata response-  -> ToReturn  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----          Just dat -> do---             delData dat---             return (Just  dat)-----    case event cont of---     Nothing -> do---        liftIO $ loop cont ioaction---        was <- getData `onNothing` return NoRemote---        when (was /= WasRemote) $ setData WasParallel------        return Nothing---     j@(Just _) -> do---        put cont{event=Nothing}---        return $ unsafeCoerce j------ * non-blocking keyboard input--getLineRef= unsafePerformIO $ newTVarIO Nothing---roption= unsafePerformIO $ newMVar []---- | install a event receiver that wait for a string and trigger the continuation when this string arrives.-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 $ putStrLn $ show ret ++ " chosen"-    return ret----- | validates an input entered in the keyboard in non blocking mode. non blocking means that--- the user can enter also anything else to activate other option--- unlike `option`, wich watch continuously, input only wait for one valid response-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 -> 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 -> 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 retry-            _ -> 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---    putStrLn "Press end to exit"-    inputLoop'  -- !> "started inputLoop"-    where--    inputLoop'= do-           r<- getLine-           processLine r-           inputLoop'--processLine r= do---   when (r=="end") $ atomically $ writeTVar rexit ()-   let rs = breakSlash [] r-   mapM_ (\ r ->  -- if (r=="end") then exit' $ Left "terminated by user" else-                 do-                    threadDelay 100000-                    atomically . 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-stay rexit=   do-    mr <- takeMVar rexit-    case mr of-      Right Nothing -> stay rexit-      Right (Just r) -> return r-      Left msg -> putStrLn msg >> exitWith ExitSuccess---- | keep the main thread running, initiate the non blocking keyboard input and execute--- the transient computation.------ It also read a slash-separated list of string that are read by--- `option` and `input` as if they were entered by the keyboard------ >  foo  -p  options/to/be/read/by/option/and/input--newtype Exit a= Exit a deriving Typeable--keep :: Typeable a => TransIO a -> IO a-keep mx = do-   rexit <- newEmptyMVar-   forkIO $ do-       liftIO $ putMVar rexit  $ Right Nothing-       runTransient $ do-           setData $ Exit rexit-           async inputLoop-            <|> do mx  -- ; liftIO (putMVar rexit  $ Right Nothing)-                       -- to avoid "takeMVar blocked in a infinite loop" error-            <|> do-               option "end" "exit"-               killChilds-               exit' (Left "terminated by user" `asTypeOf`  (type1 mx))---       return ()-   threadDelay 10000-   execCommandLine-   stay rexit-   where-   type1 :: TransIO a -> Either String (Maybe a)-   type1= undefined---- | same than `keep`but do not initiate the asynchronous keyboard input.--- Useful for debugging or for creating background tasks.-keep' :: Typeable a => TransIO a -> IO a-keep' mx  = do-   rexit <- newEmptyMVar-   forkIO $ do-           runTransient $ do-              setData $ Exit rexit-              mx >> liftIO (putMVar rexit  $ Right Nothing)-           -- to avoid takeMVar in a infinite loop-           return ()-   threadDelay 10000-   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---- | force the finalization of the main thread and thus, all the Transient block (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 .  Right $ Just x-  stop-  where-  type1 :: a -> TransIO (Exit (MVar (Either String (Maybe a))))-  type1= undefined--exit' x= do-  Exit rexit <- getSData <|> error "exit: not type expected"-  liftIO $  putMVar rexit  x ; stop----- | alternative operator for maybe values. Used  in infix mode-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'+{-# LANGUAGE ScopedTypeVariables #-}
+-----------------------------------------------------------------------------
+--
+-- Module      :  Base
+-- Copyright   :
+-- License     :  GPL (Just (Version {versionBranch = [3], versionTags = []}))
+--
+-- Maintainer  :  agocorona@gmail.com
+-- Stability   :
+-- Portability :
+--
+-- | See http://github.com/agocorona/transient
+-- everithing in this module is exported in order to allow extensibility.
+-----------------------------------------------------------------------------
+{-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE FlexibleContexts          #-}
+{-# LANGUAGE FlexibleInstances         #-}
+{-# LANGUAGE MultiParamTypeClasses     #-}
+{-# LANGUAGE DeriveDataTypeable        #-}
+{-# LANGUAGE UndecidableInstances      #-}
+{-# LANGUAGE Rank2Types                #-}
+
+module Transient.Internals where
+
+
+
+import           Control.Applicative
+import           Control.Monad.State
+import           Data.Dynamic
+import qualified Data.Map               as M
+import           Data.Monoid
+import           Debug.Trace
+import           System.IO.Unsafe
+import           Unsafe.Coerce
+import           Control.Exception
+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           System.Exit
+
+import qualified Data.ByteString.Char8 as BS
+
+
+
+--{-# INLINE (!>) #-}
+--(!>) :: Show a => b -> a -> b
+--(!>) x y=  trace (show y) x
+--infixr 0 !>
+
+
+
+
+
+data TransIO  x = Transient  {runTrans :: StateT EventF IO (Maybe x)}
+type SData= ()
+
+type EventId= Int
+
+type TransientIO= TransIO
+
+data LifeCycle= Alive | Parent | Listener | Dead deriving (Eq,Show)
+
+data EventF  = forall a b . EventF{meffects     :: ()
+                                  ,event       :: Maybe SData
+                                  ,xcomp       :: TransIO a
+                                  ,fcomp       :: [b -> TransIO b]
+                                  ,mfData      :: M.Map TypeRep SData
+                                  ,mfSequence  :: Int
+                                  ,threadId    :: ThreadId
+                                  ,freeTh      :: Bool
+                                  ,parent      :: Maybe EventF
+                                  ,children    :: MVar[EventF]
+                                  ,maxThread   :: Maybe (IORef Int)
+                                  ,labelth     :: IORef (LifeCycle,BS.ByteString)
+                                  }
+                                  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
+
+type StateIO= StateT EventF IO
+
+-- | 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 x= Transient $ x >>= return . Just
+
+-- | Run the transient computation with a blank state
+runTransient :: TransIO x -> IO (Maybe x, EventF)
+runTransient t= do
+  th <- myThreadId
+  label <- newIORef $ (Alive,BS.pack "top")
+  childs <- newMVar []
+  let eventf0=  EventF () Nothing empty [] M.empty 0
+          th False  Nothing  childs Nothing  label
+
+  runStateT (runTrans t) eventf0
+
+-- | Run the transient computation with an state
+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 _ _ x fs _ _  _ _  _ _ _ _)= runTrans $ do
+      r <- unsafeCoerce x
+      compose fs r
+
+-- | Run the closure and the continuation using his own state data
+runCont' cont= runStateT (runCont cont) cont
+
+-- | Warning: radiactive untyped stuff. handle with care
+getContinuations :: StateIO [a -> TransIO b]
+getContinuations= do
+  EventF _ _ _ 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 []= 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 _ _ 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 _ _ _ fs _ _ _ _  _ _ _ _) =
+   runTrans . (unsafeCoerce $ compose $  fs)
+
+
+setContinuation :: TransIO a -> (a -> TransIO b) -> [c -> TransIO c] -> StateIO ()
+setContinuation  b c fs =  do
+    (EventF eff ev _ _ d e f g h i j k) <- get
+    put $ EventF eff ev b ( unsafeCoerce c: fs) d e f g h i j k
+
+withContinuation  c mx= do
+    EventF eff ev f1 fs d e f g h i j k<- get
+    put $ EventF eff ev mx ( unsafeCoerce c: fs) d e f g h i j k
+    r <- mx
+    restoreStack fs
+    return r
+
+-- | run a chain of continuations. It is up to the programmer to assure by construction that
+--  each continuation type-check with the next, that the parameter type match the input of the first
+-- continuation.
+-- Normally this makes sense if it stop the current flow with `stop` after the invocation
+runContinuations :: [a -> TransIO b] -> c -> TransIO d
+runContinuations fs x= (compose $ unsafeCoerce fs)  x
+
+instance   Functor TransIO where
+  fmap f mx=  --   Transient $ fmap (fmap f) $ runTrans 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,[])   -- !> "NEWIOREF"
+
+         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
+
+restoreStack fs=
+       modify $ \(EventF eff _ f _ a b c d parent children g1 la) ->
+               EventF eff Nothing f fs a b c d parent children g1 la
+
+readWithErr line=
+     let [(v,left)] = readsPrec 0 line
+     in (v   `seq` return [(v,left)])
+                    `catch` (\(e::SomeException) ->
+                      error ("read error trying to read type: \"" ++ show( typeOf v) ++ "\" in:  "++" <"++ show line++"> "))
+
+
+readsPrec' _= unsafePerformIO . readWithErr
+
+
+class (Show a, Read a, Typeable a) => Loggable a where
+
+
+instance (Show a, Read a,Typeable a) => Loggable a where
+
+-- | 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)
+
+
+instance Alternative TransIO where
+    empty = Transient $ return  Nothing
+    (<|>) = mplus
+
+
+data RemoteStatus=   WasRemote | WasParallel | NoRemote deriving (Typeable, Eq, Show)
+
+instance MonadPlus TransIO where
+    mzero= empty
+    mplus  x y=  Transient $ do
+         mx <- runTrans x                       --  !> "RUNTRANS11111"
+         was <- getData `onNothing` return NoRemote
+         if was== WasRemote                     -- !> was
+           then return Nothing
+           else
+                 case mx of
+                     Nothing -> runTrans y       -- !> "RUNTRANS22222"
+                     justx -> return justx
+
+
+
+
+-- | A sinonym of empty that can be used in a monadic expression. it stop the
+-- computation and execute the next alternative computation (composed with `<|>`)
+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
+
+    -- | Executes the second operand even if the frist return empty.
+    -- A normal imperative (monadic) sequence uses the operator (>>) which in the
+    -- Transient monad does not execute the next operand if the previous one return empty.
+    (**>) :: m a -> m b -> m b
+
+    -- | Forces the execution of the second operand even if the first stop. It does not execute
+    -- the second operand as result of internal events occuring in the first operand.
+    -- Return the first result
+    (<**) :: m a -> m b -> m a
+
+    atEnd' ::m a -> m b -> m a
+    atEnd' = (<**)
+
+    -- | Forces the execution of the second operand even if the first stop. Return the first result. The second
+    -- operand is executed also when internal events happens in the first operand and it returns something
+    (<***) :: 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    -- !> "ma"
+                  runTrans  mb        -- !> "mb"
+                  return a
+
+infixr 1  <***  ,  <**, **>
+
+
+
+-- | When the first operand is an asynchronous operation, the second operand is executed once (one single time)
+-- when the first completes his first asyncronous operation.
+--
+-- This is useful for spawning asynchronous or distributed tasks that are singletons and that should start
+-- when the first one is set up.
+--
+-- for example a streaming where the event receivers are acivated before the senders.
+
+(<|) :: 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
+
+instance Monoid a => Monoid (TransIO a) where
+  mappend x y = mappend <$> x <*> y
+  mempty= return mempty
+
+-- | Set the current closure and continuation for the current statement
+setEventCont ::   TransIO a -> (a -> TransIO b) -> StateIO EventF
+setEventCont x f  = do
+
+   st@(EventF eff e _ fs d n  r applic  ch rc bs la)  <- get  -- !> "SET"
+   let cont=  EventF eff e x ( unsafeCoerce f : fs) d n  r applic  ch rc bs la
+   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
+   st@(EventF eff e _ fs d n  r nr  ch rc bs la)  <- get     -- !> "reset"
+   let f= \mx ->  case mx of
+                       Nothing -> empty
+                       Just x  -> (unsafeCoerce $ head fs)  x
+   put $ EventF eff e (f mx) ( tailsafe fs) d n  r nr  ch rc bs la
+   return  id
+
+tailsafe []=[]
+tailsafe (x:xs)= xs
+
+--refEventCont= unsafePerformIO $ newIORef baseEffects
+
+{-# INLINE baseEffects #-}
+baseEffects :: Effects
+
+baseEffects x  x' f' = do
+            c <-setEventCont x'  f'
+            mk <- runTrans x
+            t <- resetEventCont mk c
+            return (t,mk)
+
+instance Monad TransIO where
+
+      return  = pure
+
+      x >>= f  = Transient $ do
+--            effects <- gets effects -- liftIO $ readIORef refEventCont
+            (t,mk) <- baseEffects x x f
+            t $ case mk of
+                 Just k  ->  runTrans (f k)
+
+                 Nothing ->  return Nothing
+
+--instance MonadTrans (Transient ) where
+--  lift mx = Transient $ mx >>= return . Just
+
+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
+
+
+-- * Threads
+
+waitQSemB sem= atomicModifyIORef sem $ \n -> if n > 0 then(n-1,True) else (n,False)
+signalQSemB sem= atomicModifyIORef sem  $ \n ->  (n + 1,())
+
+-- | Set the maximun number of threads for a procedure. It is useful to limit the
+-- parallelization of transient code that uses `parallel` `spawn` and `waitEvents`
+threads :: Int -> TransIO a -> TransIO a
+threads n proc=  do
+   msem <- gets maxThread
+   sem <- liftIO $ newIORef n
+   modify $ \s -> s{maxThread= Just sem}
+   r <- proc <** (modify $ \s -> s{maxThread = msem}) -- restore it
+   return r
+
+-- | Delete all the previous child threads generated by the expression taken as parameter and continue execution
+-- of the current thread.
+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 l=  do
+     st <- get
+     liftIO $ atomicModifyIORef (labelth st) $ \(status,_) -> ((status,BS.pack l),())
+
+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
+--                                       !> "showThreads"
+
+   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= do
+      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
+       where
+       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, it set it
+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}
+   return  ()
+
+-- | Assure that at least there are n threads available
+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
+   return  ()
+--getNonUsedThreads :: TransIO (Maybe Int)
+--getNonUsedThreads= Transient $ do
+--   msem <- gets maxThread
+--   case msem of
+--    Just sem -> liftIO $ Just <$> readIORef sem
+--    Nothing -> return Nothing
+
+
+-- | The threads generated in the process passed as parameter will not be killed by `kill*`
+--  primitives.
+--
+-- Since there is no thread control, the application run slightly faster.
+freeThreads :: TransIO a -> TransIO a
+freeThreads proc= Transient $ do
+     st <- get
+     put st{freeTh= True}
+     r <- runTrans proc
+     modify $ \s -> s{freeTh= freeTh st}
+     return r
+
+-- | The threads will be killed when the parent thread dies. That is the default.
+-- This can be invoked to revert the effect of `freeThreads`
+hookedThreads :: TransIO a -> TransIO a
+hookedThreads proc= Transient $ do
+     st <- get
+     put st{freeTh= False}
+     r <- runTrans proc
+     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= noTrans $ do
+        st <- get
+        liftIO $ killBranch' st
+
+-- | Kill the childs and the thread of an state
+killBranch' cont= liftIO $ do
+
+        killChildren $ children cont
+        let thisth= threadId cont
+        let 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
+
+-- | Get the state data for the desired type if there is any.
+getData ::  (MonadState EventF m,Typeable a) =>  m (Maybe a)
+getData =  resp where
+ resp= gets mfData >>= \list  ->
+    case M.lookup ( typeOf $ typeResp resp ) list  of
+      Just x  -> return . Just $ unsafeCoerce x
+      Nothing -> return Nothing
+ typeResp :: m (Maybe x) -> x
+ typeResp= undefined
+
+
+-- | getData specialized for the Transient monad. if Nothing, the
+-- monadic computation does not continue.
+--
+-- If there is no such data, `getSData`  silently stop the computation.
+-- That may or may not be the desired behaviour.
+-- To make sure that this does not get unnoticed, use this construction:
+--
+-- >  getSData <|> error "no data"
+--
+-- To have the same semantics and guarantees than `get`, use a default value:
+--
+-- > getInt= getSData <|> return (0 :: Int)
+--
+-- The default value (0 in this case) has the same role than the initial value in a state monad.
+-- The difference is that you can define as many `get` as you need for all your data types.
+--
+-- To distingish two data with the same types, use newtype definitions.
+getSData ::  Typeable a => TransIO  a
+getSData= Transient getData
+
+-- | Synonym for `getSData`
+getState ::  Typeable a => TransIO  a
+getState= getSData
+
+-- | Set session data for this type. retrieved with getData or getSData
+-- Note that this is data in a state monad, that means that the update only affect downstream
+-- in the monad execution. it is not a global state neither a per user or per thread state
+-- it is a monadic state like the one of a state monad.
+setData ::  (MonadState EventF m, Typeable a) => a -> m ()
+setData  x=
+  let t= typeOf x in  modify $ \st -> st{mfData= M.insert  t (unsafeCoerce x) (mfData st)}
+
+
+-- | Modify state data. It accept a function that get the current state (if exist) as parameter.
+-- The state will be deleted or changed depending on function result
+modifyData :: (MonadState EventF m, Typeable a)  => (Maybe a -> Maybe a) -> m ()
+modifyData f=  modify $ \st -> st{mfData=
+       let  t= typeOf $ typeResp f
+       in M.alter alterf  t  (mfData st)}
+   where
+   typeResp :: (Maybe a -> b) -> a
+   typeResp= undefined
+   alterf mx =
+      let x' = case mx of
+                  Just x  -> Just $ unsafeCoerce x
+                  Nothing -> Nothing
+      in   unsafeCoerce $ f x'
+
+-- | Synonym for modifyData
+modifyState :: (MonadState EventF m, Typeable a)  => (Maybe a -> Maybe a) -> m ()
+modifyState= modifyData
+
+-- | Synonym for `setData`
+setState  ::  (MonadState EventF m, Typeable a) => a -> m ()
+setState= setData
+
+delData :: ( MonadState EventF m,Typeable a) => a -> m ()
+delData x=  modify $ \st -> st{mfData= M.delete (typeOf x ) (mfData st)}
+
+delState :: ( MonadState EventF m,Typeable a) => a -> m ()
+delState= delData
+
+
+
+-- | Generator of identifiers that are unique withing 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= do
+      n <- gets mfSequence
+      return n
+
+instance Read SomeException where
+   readsPrec n str=
+      let [(s , r)]= read str in [(SomeException $ ErrorCall s,r)]
+
+-- | Async calls
+
+data StreamData a=  SMore a | SLast a | SDone | SError SomeException deriving (Typeable, Show,Read)
+
+
+-- | Variant of `parallel` that repeatedly executes the IO computation without end
+--
+waitEvents ::   IO b -> TransIO b
+waitEvents io= do
+   mr <- parallel (SMore <$> io)
+   case mr of
+     SMore x -> return x
+     SError e -> back  e
+
+-- | Variant of `parallel` that execute the IO computation once
+async  ::  IO b -> TransIO b
+async io= do
+   mr <- parallel  (SLast <$> io)
+   case mr of
+     SLast x -> return x
+     SError e -> back e
+
+-- | in an alternative computation it executes an async operations synchronously.
+-- This means that the alternatives do not execute until the async operation finishes.
+-- Do not use in Applicatives.
+sync :: TransIO a -> TransIO a
+sync x= do
+  setData WasRemote
+  r <- x
+  delData WasRemote
+  return r
+
+-- | `spawn= freeThreads . waitEvents`
+spawn= freeThreads . waitEvents
+
+-- | Executes an IO action each certain interval of time and return his value if it changes
+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
+
+
+-- |  Return empty to the current thread and execute the IO action in a new thread.
+-- When the IO action returns, the transient computation continues with this value as the result
+-- The IO action may be re-executed or not depending on the result. So parallel can spawn any
+-- number of threads/results.
+--
+-- If the maximum number of threads, set with `threads` has been reached  `parallel` perform
+-- the work sequentially, in the current thread.
+-- So `parallel` means that 'it can be parallelized if there are thread available'
+--
+-- if there is a limitation of threads, when a thread finish, the counter of threads available
+-- is increased so another `parallel` can make use of it.
+--
+-- The behaviour of `parallel` depend on `StreamData`; If `SMore`, `parallel` will excute again the
+-- IO action. With `SLast`, `SDone` and `SError`, `parallel` will not repeat the IO action anymore.
+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 $ 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))
+
+--       sbs <- takeMVar sibling
+--       let (sbs', found) = drop [] th  sbs   -- !> "search "++show th ++ " in " ++ show (map threadId sbs)
+
+       if found
+         then do
+--           putMVar sibling sbs'
+--                                             !> ("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 )
+
+
+
+
+
+-- | De-invert an event handler.
+--
+-- The first parameter is the setter of the event handler  to be
+-- deinverted. Usually it is the primitive provided by a framework to set an event handler
+--
+-- the second parameter is the value to return to the event handler. Usually it is `return()`
+--
+-- it configures the event handler by calling the setter of the event
+-- handler with the current continuation
+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
+
+
+
+-- * non-blocking keyboard input
+
+getLineRef= unsafePerformIO $ newTVarIO Nothing
+
+
+roption= unsafePerformIO $ newMVar []
+
+-- | Install a event receiver that wait for a string and trigger the continuation when this string arrives.
+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 "option: " >> putStrLn (show ret)
+    return ret
+
+
+-- | Validates an input entered in the keyboard in non blocking mode. non blocking means that
+-- the user can enter also anything else to activate other option
+-- unlike `option`, wich watch continuously, input only wait for one valid response
+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
+           atomically $ writeTVar getLineRef Nothing
+           processLine r
+           inputLoop
+
+processLine r= do
+
+   let rs = breakSlash [] r
+
+   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
+
+-- | Keep the main thread running, initiate the non blocking keyboard input and execute
+-- the transient computation.
+--
+-- It also read a slash-separated list of string that are read by
+-- `option` and `input` as if they were entered by the keyboard
+--
+-- >  foo  -p  options/to/be/read/by/option/and/input
+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 than `keep` but do not initiate the asynchronous keyboard input.
+-- 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 is 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
+
+-- | Force the finalization of the main thread and thus, all the Transient block (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
+
+
+
+-- | Alternative operator for maybe values. Used  in infix mode
+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
+
+
+
+-- | Assures that backtracking will not go further back
+backCut :: (Typeable reason, Show reason) => reason -> TransientIO ()
+backCut reason= Transient $ do
+     delData $ Backtrack (Just reason)  []
+     return $ Just ()
+
+undoCut ::  TransientIO ()
+undoCut = backCut ()
+
+-- | The second parameter will be executed when backtracking
+{-# 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
+
+onUndo ::  TransientIO a -> TransientIO a -> TransientIO a
+onUndo x y= onBack x (\() -> y)
+
+
+-- | Register an action that will be executed when backtracking
+{-# 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
+
+-- | backtracking is stopped. the exection continues forward from this point on.
+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 ()
+
+retry= forward ()
+
+noFinish= forward (FinishReason Nothing)
+
+-- | Execute backtracking. It execute the registered actions in reverse order.
+--
+-- If the backtracking flag is changed the flow proceed  forward from that point on.
+--
+-- If the backtrack stack is finished or undoCut executed, `undo` will stop.
+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)) []
+
+undo ::  TransIO a
+undo= back ()
+
+
+------ finalization
+
+newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show)
+
+-- | Initialize the event variable for finalization.
+-- all the following computations in different threads will share it
+-- it also isolate this event from other branches that may have his own finish variable
+initFinish= backCut (FinishReason Nothing)
+
+-- | Set a computation to be called when the finish event happens
+onFinish :: ((Maybe SomeException) ->TransIO ()) -> TransIO ()
+onFinish f= onFinish' (return ()) f
+
+
+-- | Set a computation to be called when the finish event happens this only apply for
+onFinish' ::TransIO a ->((Maybe SomeException) ->TransIO a) -> TransIO a
+onFinish' proc f= proc `onBack`   \(FinishReason reason) ->
+    f reason
+
+
+-- | Trigger the event, so this closes all the resources
+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 ---
+-- | When a exception is produced in the continuation, the handler is executed.
+-- | handlers are executed Last in first out.
+onException :: Exception e => (e -> TransIO ()) -> TransIO ()
+onException f= onAnyException $ \e ->
+    case fromException e of
+       Nothing -> return ()
+       Just e'  -> f e'
+  where
+  onAnyException :: (SomeException ->TransIO ()) -> TransIO ()
+  onAnyException  f= (return ()) `onBack` f
+
+-- | stop the backtracking mechanism to execute further handlers
+cutExceptions= backCut (undefined :: SomeException)
+
+-- | Resume to normal execution at this point
+continue = forward (undefined :: SomeException)
+
+

@@ -1,179 +1,227 @@------------------------------------------------------------------------------------ Module      :  Transient.Logged--- Copyright   :--- License     :  GPL-3------ Maintainer  :  agocorona@gmail.com--- Stability   :--- Portability :------ |----------------------------------------------------------------------------------{-# LANGUAGE  ExistentialQuantification, FlexibleInstances, ScopedTypeVariables, UndecidableInstances #-}-module Transient.Logged(restore,checkpoint,suspend,logged,Loggable) where--import Data.Typeable-import Unsafe.Coerce-import Transient.Base-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-import System.Random---class (Show a, Read a,Typeable a) => Loggable a-instance (Show a, Read a,Typeable a) => Loggable a--logs= "logs/"---- re-excutes all the threads whose state has been logged in the "./logs" folder--- .Each log is removed when it is executed.------ example: this program, if executed three times will first print hello <number> some times--- but `suspend` will kill the threads and exit it.---- The second time, it will print "world" <number> and "world22222" <number> and will stay.------ The third time that it is executed, it only present "world22222" <number> messages------ > main= keep $ restore  $ do--- >    r <- logged $ choose [1..10 :: Int]--- >    logged $ liftIO $ print ("hello",r)--- >    suspend ()--- >    logged $ liftIO $ print ("world",r)--- >    checkpoint--- >    logged $ liftIO $ print ("world22222",r)--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)------ | save the state of  the thread that execute it and exit the transient block initiated with `keep` or similar--- . `keep` will return the value passed by `suspend`.--- If the process is executed again with `restore` it will reexecute the thread from this point on.------ it is useful to insert it in `finish` blocks to gather error information,-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---- | Save the state of every thread at this point. If the process is re-executed with `restore` it will reexecute the thread from this point on..-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 ()------fromIDyn :: (Read a, Show a, Typeable 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--{- TODO add save/recover from log-rerun :: Log -> TransIO a -> TransIO a--getLog :: TransIO Log--}---- | write the result of the computation in  the log and return it.--- but if there is data in the internal log, it read the data from the log and--- do not execute the computation.------ It accept nested step's. The effect is that if the outer step is executed completely--- the log of the inner steps are erased. If it is not the case, the inner steps are logged--- this reduce the log of large computations to the minimum. That is a feature not present--- in the package Workflow.------ >  r <- logged $ do--- >          logged this :: TransIO ()--- >          logged that :: TransIO ()--- >          logged thatOther--- >  liftIO $ print r------  when `print` is executed, the log is just the value of r.------  but at the `thatOther` execution the log is: [Exec,(), ()]----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---                                   !> ("read in 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---+-----------------------------------------------------------------------------
+--
+-- Module      :  Transient.Logged
+-- Copyright   :
+-- License     :  GPL-3
+--
+-- Maintainer  :  agocorona@gmail.com
+-- Stability   :
+-- Portability :
+--
+-- |
+--
+-----------------------------------------------------------------------------
+{-# LANGUAGE  CPP,ExistentialQuantification, FlexibleInstances, ScopedTypeVariables, UndecidableInstances #-}
+module Transient.Logged(
+
+#ifndef ghcjs_HOST_OS
+restore,checkpoint,suspend,
+#endif
+
+logged, received,param, Loggable) 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/"
+
+-- re-excutes all the threads whose state has been logged in the "./logs" folder
+-- .Each log is removed when it is executed.
+--
+-- example: this program, if executed three times will first print hello <number> some times
+-- but `suspend` will kill the threads and exit it.
+
+-- The second time, it will print "world" <number> and "world22222" <number> and will stay.
+--
+-- The third time that it is executed, it only present "world22222" <number> messages
+--
+-- > main= keep $ restore  $ do
+-- >    r <- logged $ choose [1..10 :: Int]
+-- >    logged $ liftIO $ print ("hello",r)
+-- >    suspend ()
+-- >    logged $ liftIO $ print ("world",r)
+-- >    checkpoint
+-- >    logged $ liftIO $ print ("world22222",r)
+
+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)
+
+
+
+-- | save the state of  the thread that execute it and exit the transient block initiated with `keep` or similar
+-- . `keep` will return the value passed by `suspend`.
+-- If the process is executed again with `restore` it will reexecute the thread from this point on.
+--
+-- it is useful to insert it in `finish` blocks to gather error information,
+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
+
+-- | Save the state of every thread at this point. If the process is re-executed with `restore` it will reexecute the thread from this point on..
+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
+
+
+
+-- | write the result of the computation in  the log and return it.
+-- but if there is data in the internal log, it read the data from the log and
+-- do not execute the computation.
+--
+-- It accept nested step's. The effect is that if the outer step is executed completely
+-- the log of the inner steps are erased. If it is not the case, the inner steps are logged
+-- this reduce the log of large computations to the minimum. That is a feature not present
+-- in the package Workflow.
+--
+-- >  r <- logged $ do
+-- >          logged this :: TransIO ()
+-- >          logged that :: TransIO ()
+-- >          logged thatOther
+-- >  liftIO $ print r
+--
+--  when `print` is executed, the log is just the value of r.
+--
+--  but at the `thatOther` execution the log is: [Exec,(), ()]
+--
+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"

@@ -1,76 +0,0 @@------------------------------------------------------------------------------------ Module      :  Transient.Stream.Resource--- Copyright   :--- License     :  GPL-3------ Maintainer  :  agocorona@gmail.com--- Stability   :--- Portability :------ |----------------------------------------------------------------------------------{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable  #-}-module Transient.Stream.Resource(sourceFile, sinkFile, process, initFinish, finish, onFinish) where---import Transient.Base hiding (loop)-import Transient.Backtrack-import Control.Exception-import Control.Applicative-import Data.Typeable-import Data.Char-import System.IO--import Control.Concurrent--import Control.Concurrent.STM-import Control.Monad.State------ | Stream the input to a file-sinkFile :: TransIO String -> String -> TransIO ()-sinkFile input file= process input (openFile file WriteMode)  hClose' hPutStrLn'-  where-  hClose' h _= putStr "closing " >> putStrLn file >> hClose h-  hPutStrLn' h  x= liftIO $ (SMore <$>  hPutStrLn h x)-                  `catch` (\(e::SomeException)-> return $ SError e)---- | slurp input from a file a line at a time. It creates as much threads as possible.--- to allow single threaded processing, use it with `threads 0`-sourceFile :: String -> TransIO String-sourceFile file= process (return ()) (openFile file ReadMode)  hClose' read'-      where-      hGetLine' h= (SMore <$> hGetLine h)-                   `catch` (\(e::SomeException)-> return $ SError e)-      read' h _ =  parallel $ hGetLine' h---      hClose' h _= putStr "closing ">> putStrLn file >> hClose h---- | is the general operation for processing a streamed input, with opening  resources before--- processing and closing them when finish is called.  The process statements suscribe to the--- `Finish` EVar.------ When this variable is updated, the close procedure is called.------ When the processing return `SDone` or `SError`, the `Finish` variable is updated so all the--- subscribed code, that close the resources, is executed.-process-  :: TransIO a       -- ^ input computation-     -> IO handle    -- ^ open computation that gives resources to be used during the computation-     -> (handle -> Maybe SomeException -> IO ())   -- ^ close computation that frees the resources-     -> (handle -> a -> TransIO (StreamData b))   -- ^ process to be done-     -> TransIO b-process input open close proc=do-   mh <- liftIO $ (Right <$> open)  `catch` (\(e::SomeException)-> return $ Left e)-   case mh of-      Left e -> liftIO (putStr "process: " >> print e) >> finish  (Just e) >> stop-      Right h -> do-       onFinish (liftIO . close h)-       some <- input-       v <- proc h  some-       liftIO $ myThreadId >>= print-       checkFinalize v

@@ -1,51 +1,55 @@-name: transient--version: 0.4.4.1--author: Alberto G. Corona-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: Making composable programs with multithreading, events and distributed computing-description: See <http://github.com/agocorona/transient>-             In this release distributed primitives have been moved to the transient-universe package, and web primitives have been moved to the ghcjs-hplay package.-category: Control--data-dir: ""---library-    build-depends:     base          > 4  &&  < 5-                     , containers-                     , mtl-                     , transformers-                     , stm-                     , time-                     , directory-                     , random--    exposed-modules: Transient.Backtrack-                     Transient.Base-                     Transient.EVars-                     Transient.Indeterminism-                     Transient.Internals-                     Transient.Logged-                     Transient.Stream.Resource-    exposed: True-    buildable: True-    exposed: True-    default-language: Haskell2010-    hs-source-dirs: src .--source-repository head-    type: git-    location: https://github.com/agocorona/transient+name: transient
+
+version: 0.5.1
+
+
+author: Alberto G. Corona
+extra-source-files:
+    ChangeLog.md README.md
+
+cabal-version: >=1.10
+build-type: Simple
+
+license: MIT
+license-file: LICENSE
+
+maintainer: agocorona@gmail.com
+homepage: http://www.fpcomplete.com/user/agocorona
+bug-reports: https://github.com/agocorona/transient/issues
+
+synopsis: Making composable programs with multithreading, events and distributed computing
+description: See <http://github.com/agocorona/transient>
+             In this release distributed primitives have been moved to the transient-universe package, and web primitives have been moved to the ghcjs-hplay package.
+category: Control
+data-dir: ""
+
+
+library
+    build-depends:     base          > 4  &&  < 5
+                     , containers
+                     , mtl
+                     , transformers
+                     , stm
+                     , time
+                     , directory
+                     , random
+                     , bytestring
+
+
+    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 .
+
+
+source-repository head
+    type: git
+    location: https://github.com/agocorona/transient