transient 0.1.0.4 → 0.1.0.8

raw patch · 11 files changed

+862/−789 lines, 11 filesdep +TCachedep +bytestringdep +timedep −transientdep ~basePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: TCache, bytestring, time

Dependencies removed: transient

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• Main.hs +0/−386
• move.hs +0/−178
• src/Transient/Backtrack.hs +5/−6
• src/Transient/Base.hs +138/−63
• src/Transient/DDS.hs +215/−0
• src/Transient/EVars.hs +17/−15
• src/Transient/Indeterminism.hs +47/−20
• src/Transient/Logged.hs +15/−2
• src/Transient/Move.hs +271/−84
• src/Transient/Stream/Resource.hs +119/−0
• transient.cabal +35/−35

@@ -1,386 +0,0 @@-
-
-{-# LANGUAGE ScopedTypeVariables #-}
-
-{-# LANGUAGE DeriveDataTypeable  #-}
-module Main where
-
-import           Data.Typeable
-import           Transient.Base
-import           Transient.Backtrack
-import           Transient.Indeterminism
-import           Transient.Logged
-import           Transient.Move
-import           Transient.EVars
-import           Control.Applicative
-import           Control.Concurrent
-import           Control.Exception
-import           Control.Monad.State
-import           Data.Monoid
-import           System.IO.Unsafe
-import           System.Directory
-import           System.FilePath
-import           Network.HTTP
-import           Network
-import           System.IO
-import           Data.IORef
-
-import Data.List hiding (find,map, group)
-
-
-main= keep $ do
-      oneThread $ option "main" "to kill previous spawned processes and return to the main menu"   <|> return ""
-      liftIO $ putStrLn "MAIN MENU"
-
-      nonDeterminsm <|> trans <|>
-             colors <|> app   <|>
-            futures <|> server <|>
-            distributed <|> pubSub
-
-solveConstraint=  do
-      x <- choose  [1,2,3]
-      y <- choose  [4,5,6]
-
-      guard $ x * y == 8
-
-      return (x,y)
-
-pythags = freeThreads $ do
-  x <- choose [1..50]
-  y <- choose ([1..x] :: [Int])
-  z <- choose [1..round $ sqrt(fromIntegral $ 2*x*x)]
-
-  guard (x*x+y*y==z*z)
-  th <- liftIO $  myThreadId
-  return (x, y, z, th)
-
-example1= do
-    option "ex1" "example 1"
-    r <- threads 4 solveConstraint
-    liftIO $ print r
-
-example2= do
-    option "pyt" "pythagoras"
-    r<- threads 2 pythags
-    liftIO $ print r
-
-collectSample= threads 4 $ do
-    option "coll" "group sample: return results in a list"
-    r <- collect 0 $ do
-      x <- choose  [1,2,3]
-      y <- choose  [4,5,6]
-      th <- liftIO $ threadDelay 1000 >> myThreadId
-
-      return (x,y,th)
-
-    liftIO $ print r
-
-threadSample= do
-     option "th" "threads sample"
-     liftIO $ print "number of threads? (< 10)"
-
-     n <- input  ( < 10)
-
-     threads n $ do
-        x <- choose  [1,2,3]
-        y <- choose  [4,5,6]
-        th <- liftIO $ myThreadId
-        liftIO $ print (x,y,th)
-
-nonDeterminsm= do
-      option "nondet" "Non determinism examples"
-      example1 <|> example2
-               <|> collectSample
-               <|> threadSample
-               <|> fileSearch
-
-
-
-find' :: String -> FilePath -> TransientIO FilePath
-find' s d = do
-  fs <- liftIO $ getDirectoryContents d
-       `catch` \(e:: SomeException) -> return []       -- 1
-  let fs' = sort $ filter (`notElem` [".",".."]) fs    -- 2
-  if any (== s) fs'                                    -- 3
-     then do
-       liftIO $ print $ d </> s
-       return $ d</> s
-     else do
-       f <- choose fs'                                 -- 4
-       let d' = d </> f                                -- 6
-       isdir <- liftIO $ doesDirectoryExist d'         -- 7
-       if isdir then find' s d'                        -- 8
-                else stop
-
-
-------------------
-
-fileSearch=   do
-    option "file" "example of file search"
-    r<- threads 3 $ collect 10 $ find' "Main.hs"  "."
-    liftIO $ putStrLn $ "SOLUTION= "++ show  r
---    exit
-
-
-
-
-
-
-
-trans= do
-       option "trans" "transaction examples with backtracking for undoing actions"
-       transaction <|> transaction2
-
-transaction=   do
-       option "back" "backtracking test"
-       productNavigation
-       reserve
-       payment
-
-transaction2= do
-       option "back2" "backtracking test 2"
-       productNavigation
-       reserveAndSendMsg
-       payment
-
-
-       liftIO $ print "done!"
-
-
-productNavigation = liftIO $ putStrLn "product navigation"
-
-reserve= liftIO (putStrLn "product reserved,added to cart")
-                 `onUndo` liftIO (putStrLn "product un-reserved")
-
-payment = do
-           liftIO $ putStrLn "Payment failed"
-           undo
-
-reserveAndSendMsg= do
-            reserve
-            liftIO  (putStrLn "update other database necesary for the reservation")
-                 `onUndo` liftIO (putStrLn "database update undone")
-
-
-
-colors :: TransientIO ()
-colors= do
-       option "colors" "choose between three colors"
-       r <-  color 1  "red"  <|> color 2 "green" <|> color 3 "blue"
-       liftIO $ print r
-       where
-       color :: Int -> String -> TransientIO String
-       color n str= do
-         option (show n) str
-         liftIO . print $ str ++ " color"
-         return  str
-
-app :: TransientIO ()
-app= do
-       option "app" "applicative expression that return a counter in 2-tuples every second"
-       liftIO $ putStrLn "to stop the sequence, write main(enter)"
-       counter <- liftIO $ newMVar 0
-
-       r <-  (,) <$>  number  counter 1 <*> number counter 1
-
-
-       liftIO $ putStrLn $ "result=" ++ show r
-       where
-       number counter n= waitEvents $ do
-          threadDelay $ n * 1000000
-          n <- takeMVar counter
-          putMVar counter (n+1)
-          return n
-
-futures= do
-       option "async" "for parallelization of IO actions with applicative and monoidal combinators"
-       sum1 <|> sum2
-
-sum1 :: TransientIO ()
-sum1= do
-       option "sum1" "access to two web pages concurrently and sum the number of words using Applicative"
-       liftIO $ print " downloading data..."
-       (r,r') <- (,) <$> async  (worker "http://www.haskell.org/")
-                     <*> async  (worker "http://www.google.com/")
-
-       liftIO $ putStrLn $ "result="  ++ show  (r + r')
-
-getURL= simpleHTTP . getRequest
-
-worker :: String -> IO Int
-worker url=do
-      r <- getURL url
-      body <- getResponseBody r
-      putStrLn $ "number of words in " ++ url ++" is: " ++ show(length (words body))
-      return . length . words $ body
-
-sum2 :: TransientIO ()
-sum2= do
-      option "sum2" "access to N web pages concurrenty and sum the number of words using map-fold"
-      liftIO $ print " downloading data..."
-      rs <- foldl (<>) (return 0) $ map (async . worker)
-                  [ "http://www.haskell.org/"
-                  , "http://www.google.com/"]
-
-      liftIO $ putStrLn $ "result="  ++ show rs
-
-instance Monoid Int where
-      mappend= (+)
-      mempty= 0
-
-server :: TransientIO ()
-server=  do
-       option "server" "A web server in the port 8080"
-       liftIO $ print "Server Stated"
-       sock <-  liftIO $  listenOn $ PortNumber 8080
-
-       (h,_,_) <- spawn $ accept sock `catch` (\(e::SomeException) -> sClose sock >> throw e)
-       liftIO $ do
-           hPutStr h msg
-           putStrLn "new request"
-           hFlush h
-           hClose h
-         `catch` (\(e::SomeException) -> sClose sock)
-
-msg = "HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\nPong!\r\n"
-
-
--- distributed computing
-
-distributed= do
-      option "distr" "examples of distributed computing"
-      let port1 = PortNumber 2000
-
-
-      addNodes [Node host port1 Nothing]
-      listen port1 <|> return ()-- conn port1 port1 <|> conn port2 port1
-
-      examples' host port1
-      where
-      host= "localhost"
-      conn p p'=  connect host p host p'
-
-examples' remoteHost remotePort= do
-   logged $ option "maind"  "to see this menu" <|> return ""
-   r <-logged    $ option "move" "move to another node"
-               <|> option "call" "call a function in another node"
-               <|> option "chat" "chat"
-               <|> option "netev" "events propagating trough the network"
-   case r of
-       "call"  -> callExample remoteHost remotePort
-       "move"  -> moveExample remoteHost remotePort
-       "chat"  -> chat
-       "netev" -> networkEvents remoteHost remotePort
-
-
-callExample host port= do
-   logged $ putStrLnhp  port "asking for the remote data"
-   s <- callTo host port $ liftIO $ do
-                       putStrLnhp  port "remote callTo request"
-                       readIORef environ
-
-
-   liftIO $ putStrLn $ "resp=" ++ show s
-
-
-environ= unsafePerformIO $ newIORef "Not Changed"
-
-moveExample host port= do
-   putStrLnhp  port "enter a string. It will be inserted in the other node by a migrating program"
-   name <- logged $ input (const True)
-   beamTo host port
-   putStrLnhp  port "moved!"
-   putStrLnhp  port $ "inserting "++ name ++" as new data in this node"
-   liftIO $ writeIORef environ name
-   return()
-
-
-chat ::  TransIO ()
-chat  = do
-    name  <- logged $ do liftIO $ putStrLn "Name?" ; input (const True)
-    text <- logged $  waitEvents  $ putStr ">" >> hFlush stdout >> getLine' (const True)
-    let line= name ++": "++ text
-    clustered $   liftIO $ putStrLn line
-
-
-networkEvents rh rp= do
-     logged $  do
-       putStrLnhp  rp "callTo is not  a simole remote call. it stablish a connection"
-       putStrLnhp  rp "between transient processes in different nodes"
-       putStrLnhp  rp "in this example, events are piped back from a remote node to the local node"
-
-     r <- callTo rh rp $ do
-                         option "fire"  "fire event"
-                         return "event fired"
-     putStrLnhp  rp $ r ++ " in remote node"
-
-putStrLnhp p msg= liftIO $ putStr (show p) >> putStr " ->" >> putStrLn msg
-
-
-pubSub=  do
-  option "pubs" "an example of publish-subscribe using Event Vars (EVars)"
-  v  <- newEVar  :: TransIO (EVar String)
-
-  v' <- newEVar 
-  subscribe v v' <|> publish v v'
-  where
-      
-  publish v v'= do
-    liftIO $ putStrLn "Enter a message to publish"
-    msg <-  input(const True)
-
-    writeEVar v msg
-    liftIO $ putStrLn "after writing first EVar\n"
-    writeEVar v' $ "second " ++ msg
-    liftIO $ putStrLn "after writing second EVar\n"
-    publish v v'
-
-
-  subscribe :: EVar String -> EVar String -> TransIO ()
-  subscribe v v'= do
-       r <- (,) <$> proc1 v  <*>  proc2 v'
-       liftIO $ do
-             putStr "applicative result= "
-             print r
-
-  susbcribe2 ::  EVar String -> EVar String -> TransIO ()
-  susbcribe2 v v'= do
-        x <- readEVar v
-        y <- readEVar v'
-        liftIO $ do
-                 putStr "monadic result"
-                 print (x,y)
-
-  proc1 v=  do
-    msg <- readEVar v
-    liftIO $ putStrLn $  "proc1 readed var: " ++ show msg
-    return msg
-
-  proc2 v= do
-    msg <- readEVar v
-    liftIO $ putStrLn $ "proc2 readed var: " ++ show msg
-    return msg
-
-
-
-
-
-
---main=do
---      r <- getURL "https://www.w3.org/services/html2txt?url=http%3A%2F%2Fwww.searchquotes.com%2Fsearch%2Ftransient%2F"
---      body <- getResponseBody r
---      print $ parse  quote' "" body
---      where
---      quote'= do
---          q <-  between(brackets natural) (brackets natural) string
---
---          if "http" `isPrefixOf` q
---            then quote'
---            else  return q
---
-
---main    = case (parse numbers "" "11, 2, 43") of
---            Left err  -> print err
---            Right xs  -> print (sum xs)
---
---numbers = commaSep integer

@@ -1,178 +0,0 @@-{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable #-}
-module Main where
-
-import Transient.Move
-import Transient.Logged
-import Transient.Base
-import Network
-import Control.Applicative
-
-import Control.Monad.IO.Class
-import System.Environment
-import System.IO.Unsafe
-import Data.Monoid
-import System.IO
-import Control.Monad
-import Data.Maybe
-import Control.Exception
-import Control.Concurrent (threadDelay)
-import Data.Typeable
-import Control.Concurrent.STM
-import Data.IORef
--- some tests for distributed computing
-
---main= do
---      let port1 = PortNumber 2000
---          port2 = PortNumber 2001
---
---
---      keep $  do
---        conn port1 port1 <|> conn port2 port1
---
---        examples' host port2
---      where
---      host= "localhost"
-----      delay = liftIO $ threadDelay 1000000
---      conn p p'=  connect host p host p'
---
---
---
---
---
---test = do
---   args <- getArgs
---   let ports= [("localhost",PortNumber 2000), ("localhost",PortNumber 2001)]
---
---   let [(_,port1), (_,port2)]= if null args  then ports else reverse ports
---   print [port1, port2]
---   let local= "localhost"
---   beamInit port1 $ do
---           logged $ option "call" "call"
---           callTo local port2 (liftIO $ print "HOLA")
---             <|> callTo local port2(liftIO $ print "HELLO")
-
-
--- to be executed with two or more nodes
-main = do
-  args <- getArgs
-  if length args < 2
-    then do
-     putStrLn "The program need at least two parameters:  localHost localPort  remoteHost RemotePort"
-     putStrLn "Start one node alone. The rest, connected to this node."
-     return ()
-    else keep $ do
-       let localHost= head args
-           localPort= PortNumber . fromInteger . read $ args !! 1
-           (remoteHost,remotePort) =
-             if length args >=4
-                then(args !! 2, PortNumber . fromInteger . read $ args !! 3)
-                else (localHost,localPort)
-       connect localHost localPort remoteHost remotePort
-       examples
-
-
-examples =  do
-   logged $ option "main"  "to see the menu" <|> return ""
-   r <-logged      $ option "move" "move to another node"
-               <|> option "call" "call a function in another node"
-               <|> option "chat" "chat"
-               <|> option "netev" "events propagating trough the network"
-   case r of
-       "call" -> callExample
-       "move" -> moveExample
-       "chat" -> chat
-       "netev" -> networkEvents
-
-data Environ= Environ (IORef String) deriving Typeable
-
-callExample = do
-   nodes <- logged getNodes
-   let Node host port _=  head $ tail nodes
-
-   logged $ putStrLnhp  port "asking for the remote data"
-   s <- callTo host port $  do
-                       putStrLnhp  port "remote callTo request"
-                       liftIO $ readIORef environ
-
-
-   liftIO $ putStrLn $ "resp=" ++ show s
-
-
-environ= unsafePerformIO $ newIORef "Not Changed"
-
-moveExample = do
-   nodes <- logged getNodes
-   let Node host port _=  head $ tail nodes
-
-   putStrLnhp  port "enter a string. It will be inserted in the other node by a migrating program"
-   name <- logged $ input (const True)
-   beamTo host port
-   putStrLnhp  port "moved!"
-   putStrLnhp  port $ "inserting "++ name ++" as new data in this node"
-
-
-   liftIO $ writeIORef environ name
-   return()
-
-
-chat ::  TransIO ()
-chat  = do
-    name  <- logged $ do liftIO $ putStrLn "Name?" ; input (const True)
-    text <- logged $  waitEvents  $ putStr ">" >> hFlush stdout >> getLine' (const True)
-    let line= name ++": "++ text
-    clustered $   liftIO $ putStrLn line
-
-
-networkEvents = do
-     nodes <- logged getNodes
-     let Node rh rp _=  head $ tail nodes
-     logged $  do
-       putStrLnhp  rp "callTo is not  a simole remote call. it stablish a connection"
-       putStrLnhp  rp "between transient processes in different nodes"
-       putStrLnhp  rp "in this example, events are piped back from a remote node to the local node"
-
-     r <- callTo rh rp $ do
-                         option "fire"  "fire event"
-                         return "event fired"
-     putStrLnhp  rp $ r ++ " in remote node"
-
-putStrLnhp p msg= liftIO $ putStr (show p) >> putStr " ->" >> putStrLn msg
-
-
-
-
-
---call host port proc params= do
---   port <- getPort
---   listen port <|> return
---   parms <- logged $ return params
---   callTo host port proc parms
---   close
---
---distribute proc= do
---   case dataFor proc
---    Nothing -> proc
---    Just dataproc -> do
---      (h,p) <- bestNode dataproc
---      callTo h p proc
---
---bestNode dataproc=
---   nodes <- getNodes
---      (h,p) <- bestMatch dataproc nodes   <- user defined
---
---bestMatch (DataProc nodesAccesed cpuLoad resourcesNeeded)  nodes= do
---   nodesAccesed: node, response
---
---bestMove= do
---   thisproc <- gets myProc
---   case dataFor thisproc
---      Nothing -> return ()
---      Just dataproc -> do
---        (h,p) <- bestNode dataproc
---        moveTo h p
---
---
---inNetwork= do
---   p <- getPort
---   listen p <|> return ()
-

@@ -25,20 +25,19 @@ -- | the secod parameter will be executed when backtracking
 {-# NOINLINE onUndo #-}
 onUndo :: TransientIO a -> TransientIO a -> TransientIO a
-onUndo ac bac= do
-   r<-registerUndo $ Transient $ do
+onUndo ac bac= registerUndo $ Transient $ do
      Backtrack back _ <- getSessionData `onNothing` return (Backtrack False [])
      runTrans $ if back then bac  else ac
-   return r
 
+
 -- | register an action that will be executed when backtracking
 {-# NOINLINE registerUndo #-}
 registerUndo :: TransientIO a -> TransientIO a
 registerUndo f  = Transient $ do
-   cont@(EventF x _ _ _ _ _ _ _ _)  <- get   !> "backregister"
+   cont@(EventF _ x _ _ _ _ _ _ _ _)  <- get   !> "backregister"
    md  <- getSessionData
    ss <- case md of
-        Just (bss@(Backtrack b (bs@((EventF x'  _ _ _ _ _ _ _ _):_)))) -> do
+        Just (bss@(Backtrack b (bs@((EventF _ x'  _ _ _ _ _ _ _ _):_)))) -> do
             addrx  <- addr x
             addrx' <- addr x'         -- to avoid duplicate backtracking points
             return $ if addrx == addrx' then bss else  Backtrack b $ cont:bs
@@ -67,7 +66,7 @@   where
   nullBack= Backtrack False []
   goBackt (Backtrack _ [])= return Nothing                     !> "END"
-  goBackt (Backtrack b (stack@(first@(EventF x fs _ _  _ _ _ _ _): bs)))= do
+  goBackt (Backtrack b (stack@(first@(EventF _ x fs _ _  _ _ _ _ _): bs)))= do
 --        put first{replay=True}
         setSData $ Backtrack True stack
         mr <-  runClosure first                                !> "RUNCLOSURE"

@@ -37,11 +37,14 @@ import           GHC.Conc
 import           Data.List
 import           Data.IORef
+import           System.Environment
 
 
 {-# INLINE (!>) #-}
 (!>) =  const . id -- flip trace
 infixr 0 !>
+(!!>) =   flip trace
+infixr 0 !!>
 
 data TransIO  x = Transient  {runTrans :: StateT EventF IO (Maybe x)}
 type SData= ()
@@ -50,7 +53,8 @@ 
 type TransientIO= TransIO
 
-data EventF  = forall a b . EventF{xcomp       :: TransientIO a
+data EventF  = forall a b . EventF{event       :: Maybe SData
+                                  ,xcomp       :: TransientIO a
                                   ,fcomp       :: [b -> TransientIO b]
                                   ,mfData      :: M.Map TypeRep SData
                                   ,mfSequence  :: Int
@@ -93,7 +97,7 @@ runTransient t= do
 
   th <- myThreadId
-  let eventf0=  EventF  empty [] M.empty 0
+  let eventf0=  EventF  Nothing empty [] M.empty 0
           th False  Nothing  (unsafePerformIO $ newTVarIO []) Nothing
 
 
@@ -108,7 +112,7 @@ 
 -- | run the continuation context
 runCont :: EventF -> StateIO ()
-runCont (EventF  x fs _ _  _ _  _ _ _)= runTrans ((unsafeCoerce x') >>= compose ( fs)) >> return ()
+runCont (EventF _ x fs _ _  _ _  _ _ _)= runTrans ((unsafeCoerce x') >>= compose ( fs)) >> return ()
     where
     x'=  do
 --           modify $ \s -> s{replay=True}
@@ -131,12 +135,19 @@ 
 -- | run the closure  (the 'x'  in 'x >>= f') of the current bind operation.
 runClosure :: EventF -> StateIO (Maybe a)
-runClosure (EventF x _ _ _ _ _ _ _ _) =  unsafeCoerce $ runTrans x
+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 _ _ _ _  _ _ _) x= runTrans $  (unsafeCoerce $ compose $  fs) x
+runContinuation (EventF _ _ fs _ _ _ _  _ _ _) x=
+   runTrans $  (unsafeCoerce $ compose $  fs)  x
 
+-- | run a chain of continuations. It is up to the programmer to assure by construction that
+--  each continuation type-check with the next and the parameter type match the input of the first
+-- continuation and that the output is of the type intended.
+runContinuations :: [a -> TransIO b] -> c -> TransIO d
+runContinuations fs x= (compose $ unsafeCoerce fs)  x
+
 instance   Functor TransientIO where
   fmap f mx=   -- Transient $ fmap (fmap f) $ runTrans mx
     do
@@ -152,7 +163,7 @@          rf <- liftIO $ newIORef Nothing
          rg <- liftIO $ newIORef Nothing   -- !> "NEWIOREF"
 
-         cont@(EventF _ fs a b c d peers children g1) <- get   -- !> "APLICATIVE DOIT"
+         cont@(EventF _ _ fs a b c d peers children g1) <- get   -- !> "APLICATIVE DOIT"
 
          let
              appg x = Transient $  do
@@ -168,7 +179,7 @@ 
 
 
-         put $ EventF f (unsafeCoerce appf:  fs)
+         put $ EventF Nothing f (unsafeCoerce appf:  fs)
                                           a b c d peers children g1
          k <- runTrans f
          was <- getSessionData `onNothing` return NoRemote
@@ -178,7 +189,7 @@              liftIO $ writeIORef rf  k -- :: StateIO ()
 
              mfdata <- gets mfData
-             put $ EventF g (unsafeCoerce appg :  fs) mfdata b c d peers  children g1
+             put $ EventF Nothing g (unsafeCoerce appg :  fs) mfdata b c d peers  children g1
 
 
              x <- runTrans g              !> "RUN g"
@@ -237,20 +248,20 @@ -- | set the current closure and continuation for the current statement
 setEventCont ::   TransientIO a -> (a -> TransientIO b) -> StateIO ()
 setEventCont x f  = do
-   st@(EventF   _ fs d n  r applic  ch rc bs)  <- get
-   put $ EventF x ( unsafeCoerce f : fs) d n  r applic  ch rc bs
+   st@(EventF  e _ fs d n  r applic  ch rc bs)  <- get
+   put $ EventF e x ( unsafeCoerce f : fs) d n  r applic  ch rc bs
 
 
 -- | reset the closure and continuation. remove inner binds than the prevous computations may have stacked
 -- in the list of continuations.
 resetEventCont :: Maybe a -> StateIO ()
 resetEventCont mx =do
-   st@(EventF _ fs d n  r nr  ch rc bs)  <- get
+   st@(EventF e _ fs d n  r nr  ch rc bs)  <- get
 
    let f= \mx ->  case mx of
                        Nothing -> empty
                        Just x  -> (unsafeCoerce $ head fs)  x
-   put $ EventF  (f mx) ( tailsafe fs)d n  r nr  ch rc bs
+   put $ EventF e (f mx) ( tailsafe fs)d n  r nr  ch rc bs
    where
    tailsafe []=[]
    tailsafe (x:xs)= xs
@@ -286,7 +297,8 @@ 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 for the
+-- | 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 -> TransientIO a -> TransientIO a
 threads n proc= Transient $ do
    msem <- gets maxThread
@@ -305,8 +317,31 @@    killChilds
    return r
 
+-- | add n threads to the limit of threads. If there is no limit, it set it
+addThreads' :: Int -> TransIO ()
+addThreads' n= 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}
 
+-- | assure that at least there are n threads left
+addThreads n= do
+   msem <- gets maxThread
+   case msem of
+     Nothing -> return ()
+     Just sem ->  liftIO $ modifyIORef sem $ \n' -> if n' > n then n' else  n
 
+getNonUsedThreads :: TransIO (Maybe Int)
+getNonUsedThreads= 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.
 freeThreads :: TransientIO a -> TransientIO a
 freeThreads proc= Transient $ do
@@ -316,7 +351,7 @@      modify $ \st -> st{freeTh= freeTh st}
      return r
 
--- | The threads will be killed when the parent thread dies. That is the default
+-- | 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 :: TransientIO a -> TransientIO a
 hookedThreads proc= Transient $ do
@@ -400,93 +435,114 @@ refSequence= unsafePerformIO $ newp 0
 
 
+-- | async calls
 
-data Loop= Once | Loop | Multithread deriving Eq
+data StreamData a=  SMore a | SLast a | SDone | SError String 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 -> TransientIO b
 waitEvents io= do
-   r <- parallel (Right <$> io)
+   SMore r <- parallel (SMore <$> io)
    killChilds
    return r
 
+-- Multithreaded version of `waitEvents` that do not kill the computations spawned by previous events
+waitEvents' ::   IO b -> TransientIO b
+waitEvents' io= do
+   SMore r <- parallel (SMore <$> io)
+   return r
+
 -- | variant of + parallel` that execute the IO computation once, and kill the previous child threads
 async  ::  IO b -> TransientIO b
 async io= do
-   r <- parallel  (Left <$>io)
+   SLast r <- parallel  (SLast <$>io)
    killChilds
    return r
 
+-- | variant that spawn free threads. Since there is not thread control, this is faster
 spawn ::  IO b -> TransientIO b
 spawn io= freeThreads $ do
-   r <- parallel (Right <$>io)
+
+   SMore r <- parallel (SMore <$>io)
    return r
 
-data EventValue= EventValue SData deriving Typeable
 
+
+
+
 -- |  return empty to the current thread and launch the IO action in a new thread and attaches the continuation after it.
 -- if the result of the action is `Right` the process is repeated. if not, it finish.
 --
 -- If the maximum number of threads, set with `threads` has been reached  `parallel` perform
 -- the work sequentially, in the current thread.
 --
--- when finish, increase the counter of threads available, if there is a limitation of them.
-
-parallel  ::    IO (Either b b) -> TransientIO b
-parallel  ioaction= Transient$   do
+-- When `parallel`finish, increase the counter of threads available, if there is a limitation of them.
+--
+-- The behaviour of `parallel` depend on `StreamData` if `SMore`, `parallel` will excute again the
+-- IO action. with `SLast`, `SDone` and `SError`, `parallel` will execute the continuation and will stop.
+parallel  ::    IO (StreamData b) -> TransientIO (StreamData b)
+parallel  ioaction= Transient $   do
     cont <- getCont                    -- !> "PARALLEL"
-    mv <- getSessionData
-    case mv  of
-     Just (EventValue v)  -> do
-        delSessionData $ EventValue () -- !> "ISJUST "
-        return  $ Just $ unsafeCoerce v
+    case event cont of
+     j@(Just _) -> do
+        put cont{event=Nothing}
+        return $ unsafeCoerce j
      Nothing -> do
-        liftIO  $ loop cont    ioaction
+        liftIO $ loop cont ioaction
         return Nothing
 
-
-
-
-loop (cont'@(EventF x fs a b c d peers childs g))  rec  =  do
+-- executes the an IO action and then the continuation included in the first parameter
+loop :: EventF -> IO (StreamData t) -> IO ()
+loop (cont'@(EventF e x fs a b c d peers childs g))  rec  =  do
   chs <- liftIO $ newTVarIO []
-  let cont = EventF x fs a b c d (Just cont') chs g
-      iocont dat=
-          runStateT ( do
-             setSessionData . EventValue $ unsafeCoerce dat
-             runCont cont
-             ) cont
-             >> return ()
-      loop'= do
-        mdat <- rec
-        case mdat of
-         Left dat -> iocont dat
+  let cont = EventF e x fs a b c d (Just cont') chs g
+      iocont dat= do
+          runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat}
+          return ()
 
-         Right dat -> do
-              forkMaybe cont $ iocont dat
-              loop'
+      -- execute the IO computation and then the closure-continuation
+      loop'= forkMaybe False cont $ do
+         mdat <- rec
+         case mdat of
+             se@(SError _) -> iocont se
+             SDone -> iocont SDone
+             last@(SLast _) -> iocont last
 
-  forkMaybe  cont loop'
+             more@(SMore _) -> do
+                  forkMaybe False cont $ iocont more
+                  loop'
+
+  loop'
+  return ()
   where
-      forkMaybe cont proc = do
-         dofork <- case maxThread cont of
+  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
                  th <- forkFinally1 proc $ \me -> do
-                         case me of -- !> "THREAD ENDED" of
+                         case me of -- !> "THREAD END" of
                           Left  e -> do
-                           when (fromException e /= Just ThreadKilled)$ liftIO $ print e
-                           killChildren  cont  !> "KILL RECEIVED" ++ (show $ unsafePerformIO myThreadId)
 
-                          Right _ -> do
+                             when (fromException e /= Just ThreadKilled)$ liftIO $ print e
+                             killChildren  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 list (with free)
                              --  and pass his active children to his parent
-                             return ()
+
                              th <- myThreadId
                              mparent <- free th cont
                              case mparent of
@@ -644,12 +700,17 @@ 
     inputLoop'= do
            r<- getLine
-           if r=="end" then putMVar rexit () else do
-              let rs = breakSlash [] r
-              mapM_ (\ r -> do threadDelay 1000
-                               atomically . writeTVar  getLineRef $ Just r) rs
-              inputLoop'
+           processLine r
+           inputLoop'
 
+processLine r= do
+--   when (r=="end") $ putMVar rexit ()
+   let rs = breakSlash [] r
+   mapM_ (\ r -> if (r=="end") then putMVar rexit () else do
+                    threadDelay 1000
+                    atomically . writeTVar  getLineRef $ Just r) rs
+
+    where
     breakSlash :: [String] -> String -> [String]
     breakSlash s ""= s
     breakSlash res s=
@@ -665,18 +726,32 @@ stay=  takeMVar rexit
 
 -- | keep the main thread running, initiate the asynchronous keyboard input and execute
--- the transient computation.
+-- the transient computation. It also read a slash separated list of string that are interpreted by
+-- `option` and `input` as if they were entered by the keyboard
+keep :: TransIO a -> IO a
 keep mx = do
    forkIO $ inputLoop
    forkIO $ runTransient mx  >> return ()
+   threadDelay 100000
+   args <- getArgs
+   let path = filter (\arg -> arg !! 0 == '/') args
+   when (not (null path)) $ do
+        putStr "Executing: " >> print (head path)
+        processLine $  head path
    stay
 
+-- | same than `keep`but do not initiate the asynchronous keyboard input.
+-- Useful for debugging
+keep' :: TransIO a -> IO a
+keep' mx  = do
+   forkIO $ runTransient mx  >> return ()
+   stay
+
 -- | force the finalization of the main thread and thus, all the application
 exit :: TransientIO a
 exit= do
-  liftIO $ putStrLn "Tempus fugit: exit"
   liftIO $ putMVar rexit   True
-  return undefined
+  stop
 
 -- | alternative operator for maybe values. Used  in infix mode
 onNothing :: Monad m => m (Maybe b) -> m b -> m b

@@ -0,0 +1,215 @@+{-# LANGUAGE  ExistentialQuantification, DeriveDataTypeable #-}
+
+module Transient.DDS(distribute, cmap, reduce) where
+import Transient.Base
+import Transient.Move
+import Transient.Logged
+import Transient.Indeterminism
+
+import Control.Applicative
+import System.Random
+import Control.Monad.IO.Class
+
+import System.IO
+import Control.Monad
+import Data.Monoid
+
+import Data.Typeable
+import Data.List hiding (delete)
+import Control.Exception
+import Control.Concurrent
+import Data.Time.Clock
+
+import Data.TCache
+import Data.TCache.Defs
+--import Data.TCache.DefaultPersistence
+import Data.ByteString.Lazy.Char8 (pack,unpack)
+import Control.Monad.STM
+
+
+data DDS a= Loggable a => DDS  (TransIO [PartRef a])
+data PartRef a= Ref Node Path Save deriving (Typeable, Read, Show)
+data Partition a=  Part Node Path Save [a] deriving (Typeable,Read,Show)
+type Save= Bool
+
+instance Indexable (Partition a) where
+    key (Part _ string True _)= "PartP@"++string
+    key (Part _ string False _)= "PartT@"++string
+
+instance Loggable a => IResource (Partition a) where
+    keyResource= key
+    readResourceByKey k= if k!! 5 /= 'P' then return Nothing
+        else defaultReadByKey k >>= return . fmap ( read . unpack)
+    writeResource (s@(Part _ _ save _))= if not save then return ()
+        else defaultWrite (defPath s ++ key s) (pack$ show s)
+--instance Loggable a => Serializable (Partition a) where
+--    serialize = pack . show
+--    deserialize = read . unpack
+--    setPersist (Part _ _ persist _) =
+--      Just Persist
+--        {readByKey= if persist then defaultReadByKey else const $ return Nothing
+--        ,write    =  if persist then defaultWrite else const $  const $ return ()
+--        ,delete   = defaultDelete}
+
+
+instance Loggable a => Monoid (DDS a) where
+   mempty= DDS mempty
+   mappend (DDS ta) (DDS tb)= DDS $ ta <> tb
+
+
+
+type Path=String
+
+cmap :: Loggable b => (a -> b) -> DDS a -> DDS b
+cmap f (DDS mx)= DDS $ logged $ do
+        refs <- logged mx
+        foldl (<>) mempty $  map process refs
+
+  where
+--  process ::  Partition a -> TransIO [Partition b]
+  process  (ref@(Ref node path sav))= runAt node $  do
+              xs <- getPartitionData ref mx
+              ref <- generateRef node $ map f xs
+              return [ref]
+
+
+
+
+
+reduce' :: (Loggable b, Monoid b) => ([a] -> b) -> DDS a -> TransientIO b
+reduce' f = reduce f mappend mempty
+
+reduce ::  Loggable b => ([a] -> b) -> (b -> b -> b)-> b -> DDS a ->TransientIO b
+reduce f f2 seed (DDS mx)= logged $ do
+     refs <- logged mx
+     logged $ foldl (\ x y -> f2 <$> x <*> y)(return seed) $  map process refs
+
+     where
+
+--     process :: Partition a -> TransIO b
+     process (ref@(Ref node _ _))= runAt node $ do
+               xs <- getPartitionData ref mx
+               return $ f xs
+
+getPartitionData (Ref node path save) mx= do
+    Just (Part _ _ _ xs) <- liftIO $ atomically
+                                       $ readDBRef
+                                       $ getDBRef
+                                       $ keyResource((Part node path save undefined)
+                                                    `asTypeOf` getPartitionType mx)
+    return xs
+    where
+    getPartitionType :: TransIO [PartRef a]-> Partition a
+    getPartitionType = undefined -- type level only
+
+-- en caso de fallo de Node, se lanza un clustered en busca del path
+--   si solo uno lo tiene, se copia a otro
+--   se pone ese nodo de referencia en Part
+runAtP :: Loggable a => Node  -> (Path -> IO a) -> Path -> TransIO a
+runAtP node f uuid= do
+   r <- streamFrom node $ liftIO $ (SLast <$> f uuid) `catch` sendAnyError
+   case r of
+     SLast r -> return r
+     SError e -> do
+         nodes <-  mclustered $ search uuid
+         when(length nodes < 1) $ asyncDuplicate node uuid
+         runAtP ( head nodes) f uuid
+
+search uuid= error $ "chunk failover not yet defined. Lookin for: "++ uuid
+
+asyncDuplicate node uuid= do
+    forkTo node
+    nodes <- getNodes
+    let node'= head $ nodes \\ [node]
+    content <- liftIO $ readFile uuid
+    runAt node' $ liftIO $ writeFile uuid content
+
+sendAnyError :: SomeException -> IO (StreamData a)
+sendAnyError e= return $ SError $ show e
+
+
+distribute :: Loggable a => [a] -> DDS a
+distribute = DDS  . logged . distribute'
+
+distribute' xs=  do
+   nodes <- logged getNodes
+   let size= length xs `div` length nodes
+       xss = split size xs
+   distribute'' xss nodes
+   where
+   split n []= []
+   split n xs=
+      let (h,t)= splitAt n xs
+      in h : split n t
+
+distribute'' :: Loggable a => [[a]] -> [Node] -> TransIO[PartRef a]
+distribute'' xss nodes =
+   foldl (<>) mempty $ zipWith move nodes xss   !> show xss
+   where
+   move node xs=  runAt node $ do
+                        par <- generateRef node xs
+                        return  [par]
+
+
+
+
+textFile name= DDS $ logged $ do
+   lines <- liftIO $ liftM lines (readFile name)
+   distribute' lines
+
+--getId :: DDS a -> TransIO String
+--getId (DDS mx)= do
+--     ids <- mx
+--     let ids' = map (\(Part _  path _ _) -> path) ids
+--     return $ "DDS@"++ intercalate ":" ids'
+
+
+--fromId :: String -> DDS a
+--fromId ('D':'D':'S':'@':id)= do
+--   let ids= wordsBy (==':') id
+--   nodes <- clustered' $ mapM readDBRef ids
+--   return
+--
+--   where
+--   wordsBy :: (a -> Bool) -> [a] -> [[a]]
+--   wordsBy f s = case dropWhile f s of
+--        [] -> []
+--        x:xs -> (x:w) : wordsBy f (drop1 z)
+--            where (w,z) = break f xs
+
+
+generateRef :: Loggable a => Node -> [a] -> TransIO (PartRef a)
+generateRef node x= liftIO $ do
+       temp <- getTempName
+       let reg=  Part node temp False  x
+       atomically $ newDBRef reg
+       return $ getRef reg
+
+getRef (Part n t s x)= Ref n t s
+
+getTempName :: IO String
+getTempName=  ("DDS/" ++) <$> replicateM  5 (randomRIO ('a','z'))
+
+
+-------------- Distributed  Datasource Streams ---------
+-- | produce a stream of DDS's that can be map-reduced. Similar to spark streams.
+-- each interval of time,a new DDS is produced.
+streamDDS
+  :: (Typeable a, Show a, Read a) =>
+     Integer -> IO (StreamData a) -> DDS a
+streamDDS time io= DDS $ do
+     xs <- groupByTime time $ do
+               r <- parallel io
+               case r of
+                    SDone -> stop
+                    SLast x -> return x
+                    SMore x -> return x
+                    SError e -> error e
+     distribute'  xs
+
+
+
+
+
+
+

@@ -14,24 +14,26 @@ 
 newtype EVars= EVars  (IORef (M.Map Int [EventF]))  deriving Typeable
 
-data EVar a= EVar Int (IORef (Maybe a))
+data EVar a= EVar Int (IORef (Maybe a)) deriving Typeable
 
--- * Evars are event vars. `writeEVar` trigger the execution of all the continuations associated to the  `readEVar` of this variable
+
+-- | 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) as  stack: the most recent reads are executed first.
 --
 -- 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 
+-- 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.
 --
 -- see https://www.fpcomplete.com/user/agocorona/publish-subscribe-variables-transient-effects-v
--- 
+--
 
--- | creates an EVar
 newEVar ::  TransientIO (EVar a)
 newEVar  = Transient $ do
    getSessionData `onNothing`  do -- initialize EVars
@@ -49,15 +51,15 @@ -- again the continuation. The effect is that the continuation will be executed multiple times
 -- To avoid multiple registrations, use `unsubscribe`
 readEVar :: EVar a -> TransIO a
-readEVar (EVar id ref1)= Transient $ do 
-   mr <- liftIO $ readIORef ref1
+readEVar (EVar id ref1)= Transient $ do
+   mr <- liftIO $ readIORef ref1 !> "READEVAR"
    case mr of
      Just _ -> return mr
      Nothing -> do
          cont <- getCont
          EVars ref <- getSessionData `onNothing` error "No Events context"
          map <- liftIO $ readIORef ref
-         let Just conts=  M.lookup  id map <|> Just []
+         let Just conts=  M.lookup id map <|> Just []
          liftIO $ writeIORef ref $  M.insert id (cont:conts) map
          return Nothing
 
@@ -68,12 +70,12 @@    map <- liftIO $ readIORef ref
    let Just conts = M.lookup id map <|> Just []
        len= length conts
-   runCont'  len id ref 
+   runCont'  len id ref
    liftIO $ writeIORef ref1 Nothing
    return $ Just ()
-   
+
    where
-   runCont'  0 _ _ = return () 
+   runCont'  0 _ _ = return ()
    runCont'  n id ref= do
        map <- liftIO $ readIORef ref
        let Just conts= M.lookup id map <|> Just []
@@ -82,7 +84,7 @@        runCont current
        map' <- liftIO $ readIORef ref
        let Just conts'= M.lookup id map' <|> Just []
-       if (length conts /= length conts') then return () else liftIO $ writeIORef ref $   M.insert id (nexts ++ [current]) map 
+       if (length conts /= length conts') then return () else liftIO $ writeIORef ref $   M.insert id (nexts ++ [current]) map
        runCont'  (n - 1) id ref
 
 -- | unsuscribe the last `readEVar` executed for this EVar
@@ -91,7 +93,7 @@    map <- liftIO $ readIORef ref
    let Just conts = M.lookup id map <|> Just []
    liftIO $ writeIORef ref $  M.insert id (tail conts) map
-   
+
    return $ Just ()
 
-   +

@@ -13,7 +13,7 @@ -----------------------------------------------------------------------------
 {-# LANGUAGE BangPatterns #-}
 module Transient.Indeterminism (
-choose, choose', collect, group --, found
+choose, choose', collect, collect', group, groupByTime
 ) where
 
 import Transient.Base
@@ -26,6 +26,7 @@ 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
@@ -33,13 +34,18 @@ choose  ::  [a] -> TransientIO a
 choose []= empty
 choose   xs = do
-    evs <- liftIO  $ newIORef xs
-    parallel   $ do
+    evs <- liftIO $ newIORef xs
+    r <- parallel $ do
            es <- atomicModifyIORef' evs $ \es -> let !tes= tail es in (tes,es)
            case es of
-            [x]  -> return $ Left $ head es
-            x:_  -> return $ Right x
+            [x]  -> return $ SLast $ head es
+            x:_  -> return $ SMore x
+    return $ toData r
 
+toData r= case r of
+      SMore x -> x
+      SLast x -> x
+
 -- | group the output of a possible multithreaded process in groups of n elements.
 group :: Int -> TransientIO a -> TransientIO [a]
 group num proc =  do
@@ -50,9 +56,21 @@       then stop
       else liftIO $ atomicModifyIORef v $ \(n,xs) ->  ((0,[]),xs)
 
+groupByTime :: Integer -> TransientIO a -> TransientIO [a]
+groupByTime time proc =  do
+    v  <- liftIO $ newIORef (0,[])
+    t  <- liftIO getCurrentTime
+    x  <- proc
+    n  <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> let !n'=n +1 in ((n', x:xs),n')
+    t' <- liftIO getCurrentTime
+    if diffUTCTime t' t < fromIntegral time
+      then stop
+      else liftIO $ atomicModifyIORef v $ \(n,xs) ->  ((0,[]),xs)
+
+
 -- | alternative definition with more parallelism
 choose' :: [a] -> TransientIO a
-choose'  xs = foldl (<|>) empty $ map (parallel . return . Left) xs
+choose'  xs = foldl (<|>) empty $ map (\x -> parallel (return (SLast x)) >>= return . toData) xs
 
 
 --newtype Collect a= Collect (MVar (Int, [a])) deriving Typeable
@@ -64,15 +82,21 @@ 
 
 -- | execute a process and get the first n solutions.
--- if the process end without finding the number of solutions requested, it return the fond ones
--- if he find the number of solutions requested, it kill the threads of the process and return
+-- 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 -> TransientIO a -> TransientIO [a]
-collect n search=  do
+collect n = collect' n 1000 0
+
+-- | search also between two time intervals. If the first interval has passed and there is no result, it stop
+-- after the second interval, it stop unconditionally. It stops as soon as it has enoug result.
+collect' :: Int -> NominalDiffTime -> NominalDiffTime -> TransientIO a -> TransientIO [a]
+collect' n t1 t2 search=  do
   rv <- liftIO $ atomically $ newTVar (0,[]) !> "NEWMVAR"
   endflag <- liftIO $ newTVarIO False
   st <- get
+  t <- liftIO getCurrentTime
   let any1 = do
         r <- search   !> "ANY"
         liftIO $ atomically $ do
@@ -80,16 +104,16 @@             writeTVar  rv (n1+1,r:rs) !> "MODIFY"
         stop
 
-      detect= freeThreads $ do
-          xs <- async $ do
-             threadDelay 1000 -- to allow some activity before monitoring it
-             atomically $ do
-                (n',xs) <- readTVar rv
-                ns <- readTVar $ children st
-
-                if (n > 0 && n' >= n) ||  null ns  !> show (n,n') !> (show $ length ns)
-                  then return xs
-                  else retry
+      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)) || -- !> show (n, n', length ns)
+                                 (t2 > 0 && diffUTCTime t' t > t2)
+                               then return xs else retry
 
           th <- liftIO $ myThreadId !> "KILL"
           stnow <- get
@@ -97,5 +121,8 @@           liftIO $ addThread st stnow
           return  xs
 
-  (any1 >> stop)  <|> detect
+  monitor <|> any1
+
+
+
 

@@ -11,7 +11,7 @@ -- |
 --
 -----------------------------------------------------------------------------
-{-# LANGUAGE  ExistentialQuantification #-}
+{-# LANGUAGE  ExistentialQuantification, FlexibleInstances, UndecidableInstances #-}
 module Transient.Logged  where
 
 import Data.Typeable
@@ -20,6 +20,17 @@ import Control.Applicative
 import Control.Monad.IO.Class
 
+
+--newtype Logged m a =  Logged {runLogged :: m a}
+--
+--data RLogged= forall a.(Read a, Show a) => RLogged  a
+--
+--instance Monad m => Monad (Logged m)  where
+--   return  x=  Logged $ return x
+--   Logged x >>= f =  Logged $ do
+--         r <- x
+--         runLogged $ f r
+
 --data IDynamic= IDyns String | forall a.(Read a, Show a,Typeable a) => IDynamic a
 
 --instance Show IDynamic where
@@ -29,6 +40,8 @@ --instance Read IDynamic where
 --  readsPrec n str= map (\(x,s) -> (IDyns x,s)) $ readsPrec n str
 
+class (Show a, Read a,Typeable a) => Loggable a
+instance (Show a, Read a,Typeable a) => Loggable a
 
 fromIDyn :: (Read a, Show a, Typeable a) => IDynamic -> a
 fromIDyn (IDynamic x)= unsafeCoerce x
@@ -43,7 +56,7 @@ 
 
 
--- | write the result of the computation in the log and return it.
+-- | 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.
 --

@@ -11,15 +11,17 @@ -- | see <https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv>
 -----------------------------------------------------------------------------
 {-# LANGUAGE DeriveDataTypeable , ExistentialQuantification
-    ,ScopedTypeVariables, StandaloneDeriving #-}
+    ,ScopedTypeVariables, StandaloneDeriving, RecordWildCards #-}
 module Transient.Move where
 import Transient.Base
 import Transient.Logged
 import Data.Typeable
 import Control.Applicative
 import Network
-import Network.HTTP
+
+
 import Control.Monad.IO.Class
+import Control.Monad.State
 import System.IO
 import Control.Exception
 import Data.Maybe
@@ -30,16 +32,28 @@ import Network.Info
 import System.IO.Unsafe
 import Control.Concurrent.STM as STM
+import Control.Concurrent.MVar
+
 import Data.Monoid
 import qualified Data.Map as M
-import Data.List (nub,(\\))
+import Data.List (nub,(\\),find)
 import Data.IORef
 
+import qualified Network.Socket as NS
+import qualified Network.BSD as BSD
 
+
+import qualified Data.ByteString.Char8 as BS
+import System.IO
+
+
+import Control.Concurrent
+
+
 -- | install in a remote node a haskell package with an executable transient service initialized with `listen`
 -- the package, the git repository and the main exectable must have the same name
-installService node port servport package= do
-  beamTo node port
+installService (node@(Node _ port _)) servport package= do
+  beamTo node
   liftIO $ do
      let packagename= name package
      exist <- doesDirectoryExist  packagename
@@ -47,108 +61,204 @@          runCommand $ "git clone "++ package
          runCommand $ "cd "++ packagename
          runCommand "cabal install"
-         createProcess $ shell $ "./dist/build/"++ packagename++"/"++packagename
+         return ()
+     createProcess $ shell $ "./dist/build/"++ packagename++"/"++packagename
                                        ++ " " ++ show port
-         return()
+     return()
   where
   name path=
      let x= dropWhile (/= '/') path
-     in if x== "" then tail path else name $ tail    x
+     in if x== "" then tail path else name $ tail x
 
 -- | continue the execution in a new node
 -- all the previous actions from `listen` to this statement must have been logged
-beamTo :: HostName -> PortID -> TransientIO ()
-beamTo host port= do
+beamTo :: Node -> TransientIO ()
+beamTo node =  do
   Log rec log _ <- getSData <|> return (Log False [][])
   if rec then return () else do
-      h <- liftIO $ connectTo host port
-      liftIO $ hSetBuffering h LineBuffering
-      liftIO $ hPutStrLn h (show $ reverse log) >> hFlush h
-      liftIO $ hClose h
-      delSData h
+      Connection _ bufSize <- getSData <|> return (Connection Nothing 8192)
+      h <-  assign bufSize node
+      liftIO $ hPutStrLn h (show $ SMore $ reverse log) >> hFlush h
+      release node h
+      let log'= WaitRemote: log
+      setSData $ Log rec log' log'
       stop
 
 -- | execute in the remote node a process with the same execution state
 -- all the previous actions from `listen` to this statement must have been logged
-forkTo  :: HostName -> PortID -> TransientIO ()
-forkTo host port= do
+forkTo  :: Node -> TransientIO ()
+forkTo node= do
   Log rec log _<- getSData <|> return (Log False [][])
   if rec then return () else do
-      h <- liftIO $ connectTo host port
-      liftIO $ hSetBuffering h LineBuffering
-      liftIO $ hPutStrLn h (show $ reverse log)  >> hFlush h
-      liftIO $ hClose h
-      delSData h
+      Connection _ bufSize <- getSData <|> return (Connection Nothing 8192)
+      h <-assign bufSize node
+      liftIO $ hPutStrLn h (show $ SMore $ reverse log)  >> hFlush h
+      release node h
 
 -- | executes an action in another node.
+-- All the previous actions from `listen` to this statement must have been logged
+callTo :: Loggable a => Node -> TransIO a -> TransIO a
+callTo n p = streamFrom n (SMore <$> p) >>= \(SMore x) -> return x
+
+
+-- | synonymous of `callTo`
 -- all the previous actions from `listen` to this statement must have been logged
-callTo :: (Show a, Read a,Typeable a) => HostName -> PortID -> TransIO a -> TransIO a
-callTo host port remoteProc= logged $ Transient $ do
---      liftIO $ print "callto"
+runAt :: Loggable a => Node -> TransIO a -> TransIO a
+runAt= callTo
+
+-- | `callTo` can stream data but can not inform the receiving process about the finalization. This call
+-- does it.
+--
+-- All the previous actions from `listen` to this statement must have been logged
+streamFrom :: Loggable a => Node -> TransIO (StreamData a) -> TransIO (StreamData a)
+streamFrom node remoteProc= logged $ Transient $ do
       Log rec log fulLog <- getSessionData `onNothing` return (Log False [][])
       if rec
          then
           runTrans $ do
+            rnum <- liftIO $ newMVar (0 :: Int)
+            Connection (Just (_, h, sock, blocked)) _ <- getSData  <|> error "callTo: no hander"
+            r <- remoteProc                       !> "executing remoteProc" !> "CALLTO REMOTE" -- LOg="++ show fulLog
+            n <- liftIO $ do
+--                modifyMVar_ rnum $ \n -> return (n+1)
+                withMVar blocked $ const $ hPutStrLn   h (show r)  `catch` (\(e::SomeException) -> sClose sock)
+                 -- !> "sent response, HANDLE="++ show h
+--                modifyMVar rnum $ \n -> return (n-1,n)
 
-               Connection port  h sock <- getSData <|> error "callto: no hander"
-               r <- remoteProc !> "executing remoteProc" !> "CALLTO REMOTE" -- LOg="++ show fulLog
-               liftIO $ hPutStrLn h (show r) --  `catch` (\(e::SomeException) -> sClose sock)
-                -- !> "sent response, HANDLE="++ show h
-               setSData WasRemote
-               stop
+ --           adjustSenderThreads n
 
+            setSData WasRemote
+            stop
 
          else do
+            Connection _ bufSize <- getSessionData `onNothing` return (Connection Nothing 8192)
+            h <- assign bufSize node
+            liftIO $ hSetBuffering h LineBuffering
+            liftIO $ hPutStrLn h ( show $ SLast $ reverse fulLog) {- >> hFlush h -} !> "CALLTO LOCAL" -- send "++ show  log
 
-            h <- liftIO $ connectTo host port
-            liftIO $ hPutStrLn h (show $ reverse fulLog) >> hFlush h !> "CALLTO LOCAL" -- send "++ show  log
+
             let log'= WaitRemote:tail log
             setSessionData $ Log rec log' log'
-            runTrans $ waitEvents $ do -- local side
-                   liftIO $ hSetBuffering h LineBuffering
-                   s <- hGetLine h
---                   hClose h
+            runTrans $ do
+              r<- parallel $ do -- local side
+                   r <- readHandler h
 
-                   let r = read s
+                   case r of
+                        SDone -> release node h >> return SDone
+                        other -> return other
 
-                   return r   !> "read: " ++ s ++" response type= "++show( typeOf r)
+--              adjustRecThreads h
+              case r of
+                SDone -> empty
+                other  -> return other
 
+--      where
+--      adjustRecThreads h= do
+--          b <- liftIO $ hWaitForInput  h 1
+--          addThreads' $ if b then 1 else 0
+--          liftIO $ putStrLn $ "REC "++ show (case b of True -> "INC" ; _ -> "DEC")
+--
+--      adjustSenderThreads n
+--         | n > 2 = addThreads' (-1)  >> liftIO (putStrLn ("SEND DEC"))
+--         | n==0 = addThreads' 1  >> liftIO (putStrLn ("SEND INC"))
+--         | otherwise= return () >> liftIO(myThreadId >>= \th -> (putStrLn ("SEND "++ show th)))
+
+
 -- | A connectionless version of callTo for long running remote calls
-callTo' :: (Show a, Read a,Typeable a) => HostName -> PortID -> TransIO a -> TransIO a
-callTo' rhost rport remoteProc= logged $ do
-    (host,port) <- getMyNode
-    logged $ beamTo rhost rport
+callTo' :: (Show a, Read a,Typeable a) => Node -> TransIO a -> TransIO a
+callTo' node remoteProc= logged $ do
+    mynode <- logged getMyNode
+    beamTo node
     r <- logged remoteProc
-    logged $ beamTo host port
+    beamTo mynode
     return r
 
-data Connection= Connection PortID Handle Socket deriving Typeable
+type Blocked= MVar ()
+type BuffSize = Int
+data Connection= Connection (Maybe(PortID, Handle, Socket, Blocked)) BuffSize
+                  deriving Typeable
 
+setBufSize :: Int -> TransIO ()
+setBufSize size= Transient $ do
+   Connection c _ <- getSessionData `onNothing` return (Connection Nothing  size)
+   setSessionData $ Connection c size
+   return $ Just ()
+
+
+readHandler h= do
+    line <- hGetLine h -- {-  readIORef leftover <> -} unsafeInterleaveIO (hGetLine h)
+--    liftIO $ print (line, show h)
+    let [(v,left)]= readsPrec 0 line
+--    length v `seq` writeIORef leftover left
+    return  v
+
+  `catch` (\(e::SomeException) -> do
+      hClose h
+      liftIO $ do
+         putStr "readHandler: "
+         print e
+         return SDone)
+   where
+--   hGetLine' h= do
+
+
+
+connectTo' bufSize hostname (PortNumber port) = do
+    proto <- BSD.getProtocolNumber "tcp"
+    bracketOnError
+        (NS.socket NS.AF_INET NS.Stream proto)
+        (sClose)  -- only done if there's an error
+        (\sock -> do
+          NS.setSocketOption sock NS.RecvBuffer bufSize
+          NS.setSocketOption sock NS.SendBuffer bufSize
+          he <- BSD.getHostByName hostname
+          NS.connect sock (NS.SockAddrInet port (BSD.hostAddress he))
+
+          NS.socketToHandle sock ReadWriteMode
+        )
+
 -- | Wait for messages and replay the rest of the monadic sequence with the log received.
-listen :: PortID ->  TransIO ()
-listen  port = do
-       setSData $ Log False [] []
-       sock <- liftIO $ withSocketsDo $ listenOn  port
+listen ::  Node ->  TransIO ()
+listen  (Node _  port _) = do
+   addThreads 1
+   setSData $ Log False [] []
+   Connection _ bufSize <- getSData <|> return (Connection Nothing 8192)
+   sock <- liftIO $ withSocketsDo $ listenOn  port
+   liftIO $ do NS.setSocketOption sock NS.RecvBuffer bufSize
+               NS.setSocketOption sock NS.SendBuffer bufSize
+   SMore(h,host,port1) <- parallel $ SMore <$> accept sock
+                          `catch` (\(e::SomeException) -> print "socket exception" >> sClose sock >> throw e)
 
-       (h,host,port1) <- parallel $ Right <$> accept sock
-                          `catch` (\(e::SomeException) -> sClose sock >> throw e)
+   setSData $ Connection (Just (port, h, sock, unsafePerformIO $ newMVar ())) bufSize -- !> "setdata port=" ++ show port
 
-       liftIO $  hSetBuffering h LineBuffering  -- !> "LISTEN in "++ show (h,host,port1)
+   liftIO $  hSetBuffering h LineBuffering -- !> "LISTEN in "++ show (h,host,port1)
 
-       slog <- Transient $ liftIO $ (Just <$> hGetLine  h)
-                          `catch` (\(e::SomeException) -> print "ERR" >>  return Nothing)
+   mlog <- parallel $ readHandler h
 
-       setSData $ Connection port h sock  -- !> "setdata port=" ++ show port
+   case  mlog  of
+         SError e -> do
+             liftIO $ do
+                hClose h
+                putStr "listen: "
+                print e
+             stop
 
-       let log= read slog   -- !> "read1 " ++ slog
-       setSData $ Log True log (reverse log)
+         SDone -> liftIO (hClose h) >> stop
+         SMore log -> setSData $ Log True log (reverse log)
+         SLast log -> setSData $ Log True log (reverse log)
 
 
+
+
+
+
+
+
 -- | init a Transient process in a interactive as well as in a replay mode.
 -- It is intended for twin processes that interact among them in different nodes.
-beamInit :: PortID -> TransIO a -> IO b
-beamInit port program=  keep $ do
-    listen port   <|> return ()
+beamInit :: Node  -> TransIO a -> IO a
+beamInit  node program=  keep $ do
+    listen  node   <|> return ()
     program
 
 
@@ -161,13 +271,59 @@ deriving instance Read PortID
 deriving instance Typeable PortID
 
+
+
+data Pool=  Pool{free :: [Handle], pending :: Int}
+
+data Node= Node{host :: HostName, port :: PortID, connection :: IORef Pool} deriving Typeable
+
+release (Node h p rpool) hand= liftIO $ do
+  mhs <- atomicModifyIORef rpool $
+            \(Pool hs pend) ->
+               if pend==0
+                 then (Pool [] 0,Just hs)
+                 else (Pool (hand:hs) pend,Nothing)
+  case mhs of
+    Nothing -> return ()
+    Just hs  -> mapM_ hClose hs
+
+
+assign bufSize (Node h p pool)= liftIO $ do
+    mh <- atomicModifyIORef pool $
+            \(Pool hs p) ->  if null hs then (Pool hs p, Nothing)
+                                        else (Pool (tail hs) p, Just(head hs)) !> "REUSED"
+    case mh of
+      Just handle -> liftIO (putStrLn "REUSED!") >> return handle
+      Nothing -> liftIO $ do
+                  h <- connectTo' bufSize h p     !>  "REOPEN"
+                  hSetBuffering h LineBuffering
+                  return h
+
+
+
+
 -- * Level 2: connections node lists and operations with the node list
 
-data Node= Node{host :: HostName, port :: PortID, connection :: Maybe(Handle,Socket,HostName,PortID)} deriving (Eq,Typeable)
 
+{-# NOINLINE emptyPool #-}
+emptyPool :: MonadIO m => m (IORef Pool)
+emptyPool= liftIO $ newIORef $ Pool [] 0
+
+createNode :: HostName -> Integer -> Node
+createNode h p= Node h ( PortNumber $ fromInteger p) (unsafePerformIO emptyPool)
+
+instance Eq Node where
+    Node h p _ ==Node h' p' _= h==h' && p==p'
+
 instance Show Node where show (Node h p _)= show (h,p)
-instance Read Node where readsPrec _ s= let [((h,p),s')]= readsPrec 0 s in [((Node h p Nothing),s')]
 
+instance Read Node where
+     readsPrec _ s=
+          let [((h,p),s')]= readsPrec 0 s
+          in [(Node h p empty,s')]
+          where
+          empty= unsafePerformIO  emptyPool
+
 nodeList :: TVar  [Node]
 nodeList = unsafePerformIO $ newTVarIO []
 
@@ -175,17 +331,24 @@ 
 myNode= unsafePerformIO $ newIORef Nothing
 
-setMyNode h p= liftIO $ writeIORef  myNode $ Just (h,p)
+setMyNode node= liftIO $ writeIORef  myNode $ Just node
 getMyNode= Transient $ liftIO $ readIORef myNode
 
-getNodes :: TransIO [Node]
-getNodes  = Transient $ Just <$> (liftIO $ atomically $ readTVar  nodeList)
+getNodes :: MonadIO m => m [Node]
+getNodes  = liftIO $ atomically $ readTVar  nodeList
 
-addNodes   nodes= Transient . liftIO . atomically $ do
+addNodes :: MonadIO m => [Node] -> m ()
+addNodes   nodes=  liftIO . atomically $ do
   prevnodes <- readTVar nodeList
-  writeTVar nodeList $ nub $ prevnodes ++  nodes
-  return $ Just ()
+  writeTVar nodeList $ nub $ nodes ++ prevnodes
 
+shuffleNodes :: MonadIO m => m [Node]
+shuffleNodes=  liftIO . atomically $ do
+  nodes <- readTVar nodeList
+  let nodes'= tail nodes ++ [head nodes]
+  writeTVar nodeList nodes'
+  return nodes'
+
 --getInterfaces :: TransIO TransIO HostName
 --getInterfaces= do
 --   host <- logged $ do
@@ -196,32 +359,56 @@ --      return $ show . ipv4 $ ifs !! ind
 
 
--- | execute a Transient action in each of the nodes connected. The results are aggregated with `mappend`
-clustered :: (Typeable a, Show a, Read a) => Monoid a => TransIO a -> TransIO a
+-- | execute a Transient action in each of the nodes connected.
+--
+-- The response of each node is returned and processed by the rest of the procedure.
+-- By default, the response is processed in a new thread. To restrict the number of threads
+-- use the thread control primitives.
+--
+-- this snippet receive a message from each of the simulated nodes:
+-- > main = keep $ do
+-- >    let nodes= map createLocalNode [2000..2005]
+-- >    addNodes nodes
+-- >    (foldl (<|>) empty $ map listen nodes) <|> return ()
+-- >
+-- >    r <- clustered $ do
+-- >               Connection (Just(PortNumber port, _, _, _)) _ <- getSData
+-- >               return $ "hi from " ++ show port++ "\n"
+-- >    liftIO $ putStrLn r
+-- >    where
+-- >    createLocalNode n= createNode "localhost" (PortNumber n)
+clustered :: Loggable a  => TransIO a -> TransIO a
 clustered proc= logged $ do
      nodes <- step getNodes
-     logged $ foldr (<>) mempty $ map (\(Node h p _) -> callTo h p proc) nodes !> "fold"
+     logged $ foldr (<|>) empty $ map (\node -> callTo node proc) nodes !> "fold"
 
 -- | a connectionless version of clustered for long running remote computations. Not tested
 clustered' proc= logged $ do
+     nodes <-  getNodes
+     logged $ mapM (\node -> callTo' node proc) $ nodes
+
+-- A variant of clustered that wait for all the responses and `mappend` them
+mclustered :: (Monoid a, Loggable a)  => TransIO a -> TransIO a
+mclustered proc= logged $ do
      nodes <- step getNodes
-     logged $ mapM (\(Node h p _) -> callTo' h p proc) $ nodes
+     logged $ foldr (<>) mempty $ map (\node -> callTo node proc) nodes !> "fold"
 
--- | Connect to a new node to another. The other node will notify about this connection to
--- all the nodes connected to him. the new connected node will receive the list of connected nodes
--- the nodes will be updated with this list. it can be retrieved with `getNodes`
-connect ::   HostName ->  PortID ->  HostName ->  PortID -> TransientIO ()
-connect host  port   remotehost remoteport=  do
-    listen port <|> return ()
+-- | Initiates the transient monad, initialize it as a new node (first parameter) and connect it
+-- to an existing node (second parameter).
+-- The other node will notify about this connection to
+-- all the nodes connected to him. this new connected node will receive the list of nodes
+-- the local list of nodes then is updated with this list. it can be retrieved with `getNodes`
+connect ::  Node ->  Node -> TransientIO ()
+connect  (node@(Node h port _))  remotenode=  do
+    listen node <|> return ()
     logged $ do
-        let n= Node host port Nothing
         logged $ do
-             setMyNode host port
-             addNodes [n]
-             liftIO $ putStrLn $ "connecting to: "++ show (remotehost,remoteport)
-        newnode <- logged $ return n -- must pass my node the remote node or else it will use his own
+             setMyNode node
+             addNodes [node]
+             liftIO $ putStrLn $ "connecting to: "++ show remotenode
+        newnode <- logged $ return node -- must pass my node the remote node or else it will use his own
         port <- logged $ return port
-        nodes <- callTo remotehost remoteport   $ do
+        nodes <- callTo remotenode $ do
                    clustered $  addNodes [newnode]
                    r <- getNodes
                    liftIO $ putStrLn $ "Connected to modes: " ++ show r

@@ -0,0 +1,119 @@+-----------------------------------------------------------------------------
+--
+-- Module      :  Transient.Stream.Resource
+-- Copyright   :
+-- License     :  GPL-3
+--
+-- Maintainer  :  agocorona@gmail.com
+-- Stability   :
+-- Portability :
+--
+-- |
+--
+-----------------------------------------------------------------------------
+{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable  #-}
+module Transient.Stream.Resource(sourceFile, sinkFile, process, finish, onFinish) where
+
+
+import Transient.Base hiding (loop)
+import Transient.EVars
+import Control.Exception
+import Control.Applicative
+import Control.Monad.IO.Class
+import Data.Typeable
+import Data.Char
+import System.IO
+
+
+
+
+
+
+-- | 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 (show 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(show 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
+-- EVar `Finish`.
+--
+-- when this variable is updated, the close section is called.
+--
+-- When the processing return `SDone` or `SError`, the `Finish` variable is updated so all the
+-- subscribed code that close the resources are executed.
+process
+  :: TransIO a
+     -> IO handle
+     -> (handle -> IO ())
+     -> (handle -> a -> TransIO (StreamData b))
+     -> TransIO b
+process input open close process=do
+   mh <- liftIO $ (Right <$> open)  `catch` (\(e::SomeException)-> return $ Left e)
+   case mh of
+      Left e -> liftIO (putStr "process: " >> print e) >> finish  >> stop
+      Right h -> do
+       onFinish (liftIO (close h) >> killChilds >> stop) <|> return()
+       some <- input
+       process' h  some
+       where
+       process' h something = do
+           v <- process h  something
+           checkFinalize v
+
+
+
+checkFinalize v=
+           case v of
+              SDone ->  finish  >> stop
+              SLast x ->  finish >> return x
+              SError e -> liftIO ( putStrLn e) >> finish  >> stop
+              SMore x -> return x
+
+
+
+newtype Finish= Finish (EVar Bool) deriving Typeable
+
+initFinish :: TransIO Finish
+initFinish= do
+      fin <- newEVar
+      let f = Finish fin
+      setSData  f
+      return f
+
+-- | suscribe a computation to be called when the finish event is triggered
+onFinish :: TransIO () -> TransIO a
+onFinish  close= do
+       Finish finish <- getSData <|> initFinish
+       readEVar finish
+       close
+       stop
+
+-- | trigger the event for the closing of all the resources
+finish :: TransIO ()
+finish = do
+    liftIO $ putStrLn "finish Called"
+    Finish finish <- getSData
+    writeEVar finish True
+
+
+
+
+
+
+

@@ -1,53 +1,53 @@ name: transient
-synopsis: A monad for extensible effects and primitives for unrestricted composability of applications
-homepage: http://www.fpcomplete.com/user/agocorona
-bug-reports: https://github.com/agocorona/transient/issues
-version: 0.1.0.4
+version: 0.1.0.8
 cabal-version: >=1.10
 build-type: Simple
 license: GPL-3
 license-file: LICENSE
 maintainer: agocorona@gmail.com
+homepage: http://www.fpcomplete.com/user/agocorona
+bug-reports: https://github.com/agocorona/transient/issues
+synopsis: A monad for extensible effects and primitives for unrestricted composability of applications
+description: see <http://github.com/agocorona/transient>
 category: Control
 author: Alberto G. Corona
-description:
-     see <http://github.com/agocorona/transient>
-
 data-dir: ""
 
-library
-    build-depends: base -any, mtl -any, random -any, containers -any,
-                   directory -any, filepath -any, stm -any, HTTP -any, network -any,
-                   transformers -any, process -any, network-info -any
+source-repository head
+    type: git
+    location: https://github.com/agocorona/transient
 
-    exposed-modules: Transient.Indeterminism Transient.Base Transient.EVars
-                     Transient.Backtrack Transient.Move Transient.Logged
+library
+    build-depends: base  >4 && <5 , mtl , random , containers ,
+                   directory , filepath , stm , HTTP , network ,
+                   transformers , process , network-info ,
+                   bytestring , time , TCache
+    exposed-modules: Transient.DDS Transient.Indeterminism
+                     Transient.Base Transient.EVars Transient.Backtrack Transient.Move
+                     Transient.Logged Transient.Stream.Resource
     exposed: True
     buildable: True
     default-language: Haskell2010
     hs-source-dirs: src .
-    other-modules:
 
-executable transient
-    build-depends: base >4 && <5, mtl -any, random -any, containers -any,
-                   directory -any, filepath -any, stm -any, HTTP -any, network -any,
-                   transformers -any, transient -any
-    main-is: Main.hs
-    buildable: True
-    default-language: Haskell2010
-    hs-source-dirs: .
-    other-modules:
+--executable transient
+--    build-depends: base >4 && <5, mtl , random ,
+--                   containers , directory , filepath , stm ,
+--                   HTTP , network , transformers , transient ,
+--                   bytestring
+--    main-is: examples/Main.hs
+--    buildable: True
+--    default-language: Haskell2010
+--    hs-source-dirs: .
+--    other-modules:
+--
+--executable distributed
+--    build-depends: base , mtl , random , containers ,
+--                   directory , filepath , stm , HTTP , network ,
+--                   transformers , transient , bytestring
+--    main-is: examples/distributedExamples.hs
+--    buildable: True
+--    default-language: Haskell2010
+--    hs-source-dirs: .
 
-executable move
-    build-depends: base -any, mtl -any, random -any, containers -any,
-                   directory -any, filepath -any, stm -any, HTTP -any, network -any,
-                   transformers -any, transient -any
-    main-is: move.hs
-    buildable: True
-    default-language: Haskell2010
-    hs-source-dirs: .
-    other-modules:
 
-source-repository head
-  type:     git
-  location: https://github.com/agocorona/transient