transient-universe 0.4.6 → 0.4.6.1

raw patch · 8 files changed

+88/−69 lines, 8 filesdep ~transientPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: transient

API changes (from Hackage documentation)

Files

• README.md +1/−1
• app/server/Transient/Move/Services/MonitorService.hs +9/−9
• src/Transient/MapReduce.hs +7/−6
• src/Transient/Move.hs +2/−2
• src/Transient/Move/Internals.hs +45/−27
• src/Transient/Move/Services.hs +14/−13
• tests/TestSuite.hs +5/−6
• transient-universe.cabal +5/−5

@@ -22,7 +22,7 @@ 
 Distributed Browser/server Widgets
 -------
-Browser nodes can integrate Hplayground for ghcjs, a reactive client side library based in trasient (package ghcjs-hplay) they can create widgets with HTML form elements and control the server nodes. A computation can move from browser to server and back at runtime despite the different architecture.
+Browser nodes can integrate Hplayground for ghcjs, a reactive client side library based in trasient (package axiom) they can create widgets with HTML form elements and control the server nodes. A computation can move from browser to server and back at runtime despite the different architecture.
 
 Widgets with code running in browser and servers can compose with other widgets. A Browser node can gain access to many server nodes trough the  server that delivered the web application.
 

@@ -56,7 +56,7 @@       rs <- callNodes' nodes1 (<>) mempty (installHere ident service (pernode+1)) <>           
             callNodes' nodes2 (<>) mempty (installHere ident service pernode)
       local $ addNodes rs 
-      return rs  !>  ("MONITOR RETURN---------------------------------->", rs)
+      return rs   -- !>  ("MONITOR RETURN---------------------------------->", rs)
        
     -- installIt = installHere ident service <|> installThere ident service
     installHere  :: String -> Service -> Int -> Cloud [ Node]
@@ -83,14 +83,14 @@ install ::  Service  -> Int -> TransIO ()
 
 install  service port= do
-    return () !> "IIIIIIIIIIIIIIINSTALL"
+    -- return () !> "IIIIIIIIIIIIIIINSTALL"
     install'  `catcht` \(e :: SomeException) -> liftIO (putStr "INSTALL error: " >> print e) >> empty 
     where
     install'= do
         let host= "localhost"
         program <- return (lookup "executable" service) `onNothing` empty
-        return ()  !> ("program",program)
-        tryExec program host port  <|> tryDocker service host port 
+        -- return ()  !> ("program",program)
+        tryExec program host port  <|> tryDocker service host port program
                                    <|> do tryInstall service  ; tryExec program host port
 
 emptyIfNothing :: Maybe a -> TransIO a
@@ -104,18 +104,18 @@     install package
         | "git:" `isPrefixOf` package= installGit package  
         | "https://github.com" `isPrefixOf` package =  installGit package  
-        | "http://github.com" `isPrefixOf` package =  installGit package  
+        | "http://github.com"  `isPrefixOf` package =  installGit package  
 
 
-tryDocker service host port= do
+tryDocker service host port program= do
      image <- emptyIfNothing $ lookup "image" service
      path <- Transient $ liftIO $ findExecutable "docker"    -- return empty if not found
-     liftIO $ callProcess path ["run", image,"-p","start/"++host++"/"++ show port]
+     liftIO $ callProcess path ["run", image,"-p"," start/"++host++"/"++ show port++ " " ++ program]
 
 
 tryExec program host port= do
-     path <-  Transient $ liftIO $ findExecutable program   !> ("findExecutable", program)
-     spawnProgram program host port    !>"spawn"
+     path <-  Transient $ liftIO $ findExecutable program  -- !> ("findExecutable", program)
+     spawnProgram program host port  --  !>"spawn"
      where
      spawnProgram  program host port= liftIO $ do
 

@@ -7,7 +7,8 @@ Distributable(..),distribute, getText,
 getUrl, getFile,textUrl, textFile,
 mapKeyB, mapKeyU, reduce,eval,
-PartRef)
+--v* internals
+DDS(..),Partition(..),PartRef(..))
  where
 
 #ifdef ghcjs_HOST_OS
@@ -25,7 +26,7 @@ textUrl _ = undefined
 getFile _ _ = undefined
 eval _= local stop
-
+data Partition
 data DDS= DDS
 class Distributable
 data PartRef a=PartRef a
@@ -127,6 +128,7 @@    splitAt :: Int -> c a -> (c a, c a)
    fromList :: [a] -> c a
 
+
 instance (Loggable a) => Distributable DV.Vector a where
    singleton = DV.singleton
    splitAt= DV.splitAt
@@ -187,6 +189,7 @@ 
 boxids= unsafePerformIO $ newIORef 0
 
+
 reduce ::  (Hashable k,Ord k, Distributable vector a, Loggable k,Loggable a)
              => (a -> a -> a) -> DDS (M.Map k (vector a)) ->Cloud (M.Map k a)
 
@@ -408,15 +411,13 @@ -- The function parameter partition the text in words
 getText  :: (Loggable a, Distributable vector a) => (String -> [a]) -> String -> DDS (vector a)
 getText part str= DDS $ loggedc $ do
-   nodes' <- local getEqualNodes                                        -- !> "getText"
-   let nodes  = filter (not . isWebNode) nodes'
+   nodes <- local getEqualNodes                                        -- !> "getText"
    let lnodes = length nodes
 
    parallelize  (process lnodes)  $ zip nodes [0..lnodes-1]
    where
-   isWebNode node= "webnode" `elem` (map fst $ nodeServices node)
 
-   process lnodes (node,i)= do
+   process lnodes (node,i)= 
       runAt node $ local $ do
             let xs = part str
                 size= case length xs `div` lnodes of 0 ->1 ; n -> n

@@ -66,10 +66,10 @@ 
 
 -- * Running Local Computations
-local, onAll, lazy, loggedc, lliftIO, localIO, fullStop,
+local, onAll, lazy, fixRemote, loggedc, lliftIO, localIO, fullStop, 
 
 -- * Moving Computations
-wormhole, teleport, copyData,
+wormhole, teleport, copyData, fixClosure,
 
 -- * Running at a Remote Node
 beamTo, forkTo, callTo, runAt, atRemote,

@@ -128,7 +128,7 @@ runCloud :: Cloud a -> TransIO a
 
 runCloud x= do
-       closRemote  <- getSData <|> return (Closure 0)
+       closRemote  <- getSData <|> return (Closure  0)
        runCloud' x <*** setData  closRemote
 
 
@@ -214,11 +214,22 @@ lazy mx= onAll $ getCont >>= \st -> Transient $
         return $ unsafePerformIO $  runStateT (runTrans mx) st >>=  return .fst
 
+-- | executes a non-serilizable action  in the remote node, whose result can be used by subsequent remote invocations
+fixRemote mx= do
+             r <- lazy mx
+             fixClosure
+             return r
+        
+-- | experimental: subsequent remote invocatioms will send logs to this closure. Therefore logs will be shorter. 
+--
+-- Also, non serializable statements before it will not be re-executed
+fixClosure= atRemote $ local $ async $ return ()
+
 -- log the result a cloud computation. like `loogged`, this erases all the log produced by computations
 -- inside and substitute it for that single result when the computation is completed.
 loggedc :: Loggable a => Cloud a -> Cloud a
 loggedc (Cloud mx)= Cloud $ do
-    closRemote  <- getSData <|> return (Closure 0 )
+    closRemote  <- getSData <|> return (Closure  0 )
     logged mx <*** setData  closRemote
 
 
@@ -282,9 +293,11 @@ -- the result back  to the original node.
 atRemote :: Loggable a => Cloud a -> Cloud a
 atRemote proc= loggedc' $ do
+     was <- lazy $ getSData <|> return NoRemote
      teleport                                              -- !> "teleport 1111"
      r <- Cloud $ runCloud proc <** setData WasRemote
      teleport                                              -- !> "teleport 2222"
+     onAll $ setData was
      return r
 
 -- | synonymous of `callTo`
@@ -345,7 +358,7 @@ wormhole node (Cloud comp) = local $ Transient $ do
    moldconn <- getData :: StateIO (Maybe Connection)
    mclosure <- getData :: StateIO (Maybe Closure)
-
+   
    -- labelState $ "wormhole" ++ show node
    logdata@(Log rec log fulLog) <- getData `onNothing` return (Log False [][])
 
@@ -356,6 +369,7 @@                     conn <-  mconnect node
                     liftIO $ writeIORef (remoteNode conn) $ Just node
                     setData  conn{calling= True}
+                    setData $ (Closure 0 )
 
                     comp )
                   <*** do when (isJust moldconn) . setData $ fromJust moldconn
@@ -387,14 +401,17 @@ teleport =  do
   local $ Transient $ do
      cont <- get
-     -- labelState "teleport"
-     -- send log with closure at head
      Log rec log fulLog <- getData `onNothing` return (Log False [][])
+
+     conn@Connection{connData=contype,closures= closures,calling= calling} <- getData
+             `onNothing` error "teleport: No connection defined: use wormhole"
+
      if not rec   -- !> ("teleport rec,loc fulLog=",rec,log,fulLog)
                   -- if is not recovering in the remote node then it is active
       then  do
-        conn@Connection{connData=contype,closures= closures,calling= calling} <- getData
-             `onNothing` error "teleport: No connection defined: use wormhole"
+
+ 
+
 #ifndef ghcjs_HOST_OS
         case contype of
          Just Self ->  runTrans $ do
@@ -412,31 +429,32 @@ #endif
 
          --read this Closure
-          Closure closRemote  <- getData `onNothing` return (Closure 0 )
+          Closure closRemote  <- getData `onNothing`  return (Closure 0 )
 
+          let closLocal = sum $ map ( \x -> case x of Wait -> 100000;
+                                                      Exec -> 1000
+                                                      _ -> 1) fulLog          
          --set his own closure in his Node data
 
-          let closLocal = sum $ map (\x-> case x of Wait -> 100000;
-                                                    Exec -> 1000
-                                                    _ -> 1) fulLog
 
---         closLocal  <-   liftIO $ randomRIO (0,1000000)
+          -- closLocal  <-   liftIO $ randomRIO (0,1000000)
 --          node <- runTrans getMyNode
           
           liftIO $ modifyMVar_ closures $ \map -> return $ M.insert closLocal (fulLog,cont) map
 
-          let tosend= reverse $ if closRemote==0 then fulLog else  log
-          
+          let tosend= reverse $ if closRemote==0 then fulLog     else log 
+
+          -- send log with closure ids at head
           runTrans $ do msend conn $ SMore $ ClosureData closRemote closLocal tosend
                                      !> ("teleport sending", SMore (closRemote,closLocal,tosend))
                                      !> "--------->------>---------->"
+                  
 
-          setData $ if (not calling) then  WasRemote else WasParallel
+          setData $ if (not calling) then  WasRemote else WasParallel !> "SET WASPAraLLEL"
 
           return Nothing
 
       else do
-
          delData WasRemote                -- !> "deleting wasremote in teleport"
                                           -- it is recovering, therefore it will be the
                                           -- local, not remote
@@ -463,7 +481,7 @@ putMailbox :: Typeable a => a -> TransIO ()
 putMailbox = putMailbox' (0::Int)
 
--- | write to a mailbox identified by an Integer besides the type
+-- | write to a mailbox identified by an identifier besides the type
 putMailbox' :: (Typeable b, Ord b, Typeable a) =>  b -> a -> TransIO ()
 putMailbox'  idbox dat= do
    let name= MailboxId idbox $ typeOf dat
@@ -492,7 +510,7 @@ getMailbox :: Typeable a => TransIO a
 getMailbox = getMailbox' (0 :: Int)
 
--- | read from a mailbox identified by a number besides the type
+-- | read from a mailbox identified by an identifier besides the type
 getMailbox' :: (Typeable b, Ord b, Typeable a) => b -> TransIO a
 getMailbox' mboxid = x where
  x = do
@@ -628,7 +646,7 @@   dat <- react (hsonmessage ws) (return ())
   case JM.getData dat of
     JM.StringData str  ->  return (read' $ JS.unpack str)
-        --         !> ("Browser webSocket read", str)  !> "<------<----<----<------"
+                 !> ("Browser webSocket read", str)  !> "<------<----<----<------"
     JM.BlobData   blob -> error " blob"
     JM.ArrayBufferData arrBuffer -> error "arrBuffer"
 
@@ -702,7 +720,7 @@         parallel $ do
             s <- WS.receiveData sconn
             return . read' $  BS.unpack s
-              --  !>  ("WS MREAD RECEIVED ----<----<------<--------", s)
+                !>  ("WS MREAD RECEIVED ----<----<------<--------", s)
 
 mread (Connection  _ _ (Just (Relay conn _  )) _ _ _ _ _ _)=  
      mread conn  -- !> "MREAD RELAY"
@@ -1053,7 +1071,7 @@   x <- newMVar ()
   y <- newMVar M.empty
   noremote <- newIORef Nothing
-  z <-  return $ error "closchildren newIORef M.empty"
+  z <-  return $ error "closchildren: newIORef M.empty"
   return $ Connection my noremote Nothing  8192
                  (error "defConnection: accessing network events out of listen")
                  x  False y z
@@ -1241,7 +1259,7 @@ 
                log' <- case (method,lookup "Content-Type" headers) of
                        ("POST",Just "application/x-www-form-urlencoded") -> do
-                            len <- read <$>  BC.unpack
+                            len <- read <$> BC.unpack
                                         <$> (Transient $ return (lookup "Content-Length" headers))
                             setData $ ParseContext (return mempty) $ BS.take len str
 
@@ -1275,8 +1293,8 @@ --                     return ()                                                   !> "WEBSOCKET"
                      r <-  parallel $ do
                              msg <- WS.receiveData sconn
-                            --  return () !> ("Server WebSocket msg read",msg)
-                            --            !> "<-------<---------<--------------"
+                             return () !> ("Server WebSocket msg read",msg)
+                                         !> "<-------<---------<--------------"
 
                              case reads $ BS.unpack msg of
                                [] -> do
@@ -1371,7 +1389,7 @@ 
 type IdClosure= Int
 
-newtype Closure= Closure IdClosure deriving Show
+newtype Closure= Closure  IdClosure -- deriving Show
 
 execLog :: StreamData NodeMSG -> TransIO ()
 execLog  mlog = Transient $
@@ -1390,7 +1408,7 @@       conn@Connection {closures=closures} <- getData `onNothing` error "Listen: myNode not set"
       if closl== 0 then do
            setData $ Log True log  $ reverse log
-           setData $ Closure closr
+           setData $ Closure  closr
            return $ Just ()                  --  !> "executing top level closure"
        else do
 
@@ -1421,7 +1439,7 @@                         liftIO $ runStateT (do
                              let nlog= reverse log ++  fulLog
                              setData $ Log True  log  nlog
-                             setData $ Closure closr
+                             setData $ Closure  closr
 --                                                                !> ("SETCLOSURE",closr)
                              runContinuation cont ()) cont
                         return Nothing

@@ -11,7 +11,7 @@ -- |
 --
 -----------------------------------------------------------------------------
-{-# LANGUAGE ScopedTypeVariables, CPP #-}
+{-# LANGUAGE ScopedTypeVariables, CPP, FlexibleInstances, UndecidableInstances #-}
 
 #ifndef ghcjs_HOST_OS
 
@@ -19,22 +19,15 @@ 
 import Transient.Internals
 import Transient.Move.Internals
-import Transient.Logged
 -- import Transient.Backtrack
 -- import Transient.Internals(RemoteStatus(..), Log(..))
 import Transient.Move.Utils
 
--- import Transient.EVars
-import Transient.Indeterminism
 import Control.Monad.IO.Class
 import System.IO.Unsafe
 import Control.Concurrent.MVar
 import Control.Applicative
 import System.Process
-import Control.Monad
-import Data.List
-import Data.Maybe
-import Data.Monoid
 import Control.Concurrent(threadDelay)
 import Control.Exception hiding(onException)
 import Data.IORef
@@ -60,19 +53,25 @@ 
 requestInstance :: String -> Service -> Int -> Cloud [Node]
 requestInstance ident service num=  loggedc $ do
-       return () !> "requestInstance"
-       local $ onException $ \(e:: ConnectionError) ->  startMonitor >> continue     !> ("EXCEPTIOOOOOOOOOOON",e)
-
+    --    return () !> "requestInstance"
+       local $ onException $ \(e:: ConnectionError) ->  startMonitor >> continue   --   !> ("EXCEPTIOOOOOOOOOOON",e)
        nodes <- callService' ident monitorNode (ident,service,num)
        local $ addNodes nodes      -- !> ("ADDNODES",service)
        return nodes
 
 startMonitor :: MonadIO m => m ()
 startMonitor=  liftIO $ do
-        createProcess . shell $ "monitorService -p start/localhost/"++ show monitorPort
+        (_,_,_,h) <- createProcess . shell $ "monitorService -p start/localhost/"++ show monitorPort
+        writeIORef monitorHandle $ Just h
         threadDelay 2000000
 
+monitorHandle= unsafePerformIO $ newIORef Nothing
 
+endMonitor= do
+    mm <- readIORef monitorHandle
+    case mm of
+        Nothing -> return ()  
+        Just h  -> interruptProcessGroupOf h
 
 findInNodes :: Service -> TransIO [Node]
 findInNodes service =  do
@@ -263,4 +262,6 @@ #else
 requestInstance :: String -> Service -> Int -> Cloud [Node]
 requestInstance ident service num= logged empty
-#endif+#endif
+
+

@@ -5,8 +5,6 @@ 
 import           Control.Monad
 import           Control.Monad.IO.Class
-import           Data.IORef
-import           GHC.Conc
 import           Control.Applicative
 import           Data.Monoid
 import           Transient.Base
@@ -34,19 +32,21 @@ 
 main= do
      mr <- keep $ test   `catcht` \(e:: SomeException) -> liftIO (putStr "EXCEPTiON: " >> print e) >> exit (Just e)
+     endMonitor 
+
      case mr of
-       Nothing -> print "EXCEPTION" >> exitFailure
+       Nothing -> print "NO RESULT, NO THREADS RUNNING" >> exitFailure
        Just Nothing -> print "SUCCESS" >> exitSuccess 
        Just (Just e) -> putStr "FAIL: " >> print e >> exitFailure
 
 
+
 test=  initNodeServ service  "localhost" 8080 $ do
      
           
           node0 <- local getMyNode
           
           local $ guard (nodePort node0== 8080)       -- only executes in node 8080
-
           
       --  local $ option "get" "get instances"
 
@@ -91,8 +91,7 @@           r <- reduce  (+)  . mapKeyB (\w -> (w, 1 :: Int))  $ getText  words "hello world hello"
           localIO $  print r
           assert (sort (M.toList r) == sort [("hello",2::Int),("world",1)]) $ return r
-
-          return Nothing
+          
 
           local $ exit (Nothing  :: Maybe SomeException) -- remove this to repeat the test
  

@@ -1,5 +1,5 @@ name: transient-universe
-version: 0.4.6
+version: 0.4.6.1
 cabal-version: >=1.10
 build-type: Simple
 license: MIT
@@ -62,7 +62,7 @@         stm -any,
         time -any,
         transformers -any,
-        transient >= 0.5.8
+        transient >= 0.5.9.2
     default-language: Haskell2010
     hs-source-dirs: src .
 
@@ -71,7 +71,7 @@     if !impl(ghcjs >=0.1)
         build-depends:
             transformers -any,
-            transient >=0.5.8,
+            transient >=0.5.9.2,
             transient-universe,
             process,
             directory
@@ -91,7 +91,7 @@     if !impl(ghcjs >=0.1)
         build-depends:
             mtl -any,
-            transient >= 0.5.8,
+            transient >= 0.5.9.2,
             random -any,
             text -any,
             containers -any,
@@ -125,7 +125,7 @@     if !impl(ghcjs >=0.1)
         build-depends:
             mtl -any,
-            transient >= 0.5.8,
+            transient >= 0.5.9.2,
             random -any,
             text -any,
             containers -any,