transient-universe 0.3.2.1 → 0.3.2.2

raw patch · 4 files changed

+109/−121 lines, 4 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• src/Transient/Move.hs +34/−75
• src/Transient/Move/Services.hs +58/−31
• src/Transient/Move/Utils.hs +16/−14
• transient-universe.cabal +1/−1

@@ -18,18 +18,23 @@ Cloud(..),runCloudIO, runCloudIO',local,onAll,lazy, loggedc, lliftIO,localIO,
 listen, Transient.Move.connect, connect', fullStop,
 
+-- * primitives for communication
 wormhole, teleport, copyData,
 
+-- * single node invocation
 beamTo, forkTo, streamFrom, callTo, runAt, atRemote,
 
-clustered, mclustered,
+-- * invocation of many nodes
+clustered, mclustered, callNodes,
 
+-- * messaging
 newMailbox, putMailbox,getMailbox,cleanMailbox,
 
 #ifndef ghcjs_HOST_OS
 setBuffSize, getBuffSize,
 #endif
 
+-- * node management
 createNode, createWebNode, createNodeServ, getMyNode, getNodes,
 addNodes, shuffleNodes,
 
@@ -444,7 +449,7 @@    if not rec                                    --  !> ("wormhole recovery", rec)
             then runTrans $ (do
 
-                    conn <-  mconnect node       --  !> (mynode,"connecting node ",  node)
+                    conn <-  mconnect node         -- !> (mynode,"connecting node ",  node)
                     setData  conn{calling= True}
 #ifdef ghcjs_HOST_OS
                     addPrefix    -- for the DOM identifiers
@@ -571,7 +576,7 @@ 
 mconnect :: Node -> TransIO  Connection
 mconnect  node@(Node _ _ _ _ )=  do
-  nodes <- getNodes
+  nodes <- getNodes                                 --  !> ("connecting node", node)
 
   let fnode =  filter (==node) nodes
   case fnode of
@@ -583,7 +588,7 @@     case plist  of
       handle:_ -> do
                   delData $ Closure undefined
-                  return  handle                        -- !>   "REUSED!"
+                  return  handle                       --  !>   ("REUSED!", node)
 
       _ -> do
 --        liftIO $ putStr "*****CONNECTING NODE: " >> print node
@@ -594,9 +599,10 @@ 
         conn <- liftIO $ do
           let size=8192
-          sock <-  connectTo' size  host $ PortNumber $ fromIntegral port  --   !> ("CONNECTING ",node)
+          sock <-  connectTo' size  host $ PortNumber $ fromIntegral port
+                      -- !> ("CONNECTING ",port)
 
-          let conn= (defConnection 8100){myNode=my,comEvent= ev,connData= Just $ Node2Node u  sock (error $ "addr: outgoing connection")}
+          conn <- defConnection >>= \c -> return c{myNode=my,comEvent= ev,connData= Just $ Node2Node u  sock (error $ "addr: outgoing connection")}
 
           SBS.send sock "CLOS a b\n\n"   -- !> "sending CLOS"
 
@@ -607,7 +613,7 @@         conn <- do
 
           ws <- connectToWS host $ PortNumber $ fromIntegral port
-          let conn= (defConnection 8100){comEvent= ev,connData= Just $ Web2Node ws}
+          conn <- defConnection >>= \c -> return c{comEvent= ev,connData= Just $ Web2Node ws}
 
           return conn    -- !>  ("websocker CONNECION")
 #endif
@@ -693,18 +699,6 @@ newMailbox name= do
 --   return ()  -- !> "newMailBox"
    Connection{comEvent= mv} <- getData `onNothing` errorMailBox
---   onFinish . const $ liftIO $ do
-----         return ()                        --  !> "NEWMAILBOX finish"
---         mailboxes <- readIORef mv
---         let me = M.lookup name  mailboxes
---         case me of
---             Nothing -> empty             -- !> "EMPTY"
---             Just (EVar id rn ref1) -> do
---                 n <- atomically $ do
---                      (n,n') <- readTVar rn
---                      writeTVar rn (n-1,n'-1)        -- !> ("decreased rn",n-1)
---                      return $ n-1
---                 when (n==0) $ atomicModifyIORef mv $ \mboxes -> (M.delete name mboxes,())
    ev <- newEVar
    liftIO $ atomicModifyIORef mv $ \mailboxes ->   (M.insert name ev mailboxes,())
 
@@ -738,45 +732,6 @@              Nothing -> empty
              Just x -> return x
 
---getMailBox :: Typeable a => String -> TransIO a
---getMailBox name= Transient $ do
---     return () !> "getMailBox"
---     Connection{comEvent=(EVar id rn ref1)} <- getData `onNothing` error "getMailBox: accessing network events out of listen"
---     runTrans  $ do
---         liftIO $ atomically $ readTVar rn >>= \(n,n') -> writeTVar rn (n +1,n'+1)
---         r <- parallel $  do
---                  n' <- atomically $ do
---                              (n,n') <- readTVar rn
---                              writeTVar rn $ if n' > 1 then   (n,n'-1) else (n, n)
---                              return n'
---                  return () !> ("rear rn",name,n')
---                  atomically $ do
---                    d <- if n'> 1 then  peekTChan ref1 else readTChan ref1
---
---
---                    return () !> ("readmailbox",name,n')
---                    case d of
---                        SDone -> return SDone
---                        SMore x ->  case fromDynamic x of
---                              Nothing -> return $ SMore Nothing
---                              Just (nam, Nothing) -> do
---                                readTVar rn >>= \(n,n') -> writeTVar rn $ (n,n' -1)
---                                return SDone
---                              Just (nam,Just dat) ->
---                                return $ SMore $  if nam /= name then  Nothing else Just dat
-----                        SLast x -> case fromDynamic x of
-----                              Nothing -> retry
-----                              Just (nam, Nothing) -> return SDone
-----                              Just (nam,Just dat) ->
-----                                if nam /= name !>(nam,name) then retry else return $ SLast dat
---                        SError e -> return $ SError e
---
---         case r of
---            SDone -> empty
---            SMore Nothing -> empty
---            SMore (Just x) -> return x
---            SLast (Just x) -> return x
---            SError e -> error $ show e
 
 -- | delete all subscriptions for that mailbox expecting this kind of data.
 cleanMailbox :: Typeable a =>  T.Text ->  a -> TransIO ()
@@ -793,21 +748,22 @@ 
 
 
-defConnection :: Int -> Connection
+defConnection :: MonadIO m => m Connection
 
 -- #ifndef ghcjs_HOST_OS
-defConnection size=
-  Connection (createNode "program" 0) Nothing  size
+defConnection = liftIO $ do
+  x <- newMVar ()
+  y <- newMVar M.empty
+  return $ Connection (error "node in default connection") Nothing  8192
                  (error "defConnection: accessing network events out of listen")
-                 (unsafePerformIO $ newMVar ())
-                 False (unsafePerformIO $ newMVar M.empty)
+                 x  False (y)
 
 
 
 #ifndef ghcjs_HOST_OS
 setBuffSize :: Int -> TransIO ()
 setBuffSize size= Transient $ do
-   conn<- getData `onNothing` return (defConnection 8192)
+   conn<- getData `onNothing`  defConnection
    setData $ conn{bufferSize= size}
    return $ Just ()
 
@@ -823,7 +779,7 @@ 
    setData $ Log False [] []
 
-   conn' <- getSData <|> return (defConnection 8192)
+   conn' <- getSData <|> defConnection
    ev <- liftIO $ newIORef M.empty
    let conn= conn'{myNode=node, comEvent=ev}
 
@@ -836,7 +792,7 @@ 
 
 
-listenNew port conn= do --  node bufSize events blocked port= do
+listenNew port conn= do
 
 
    sock <- liftIO . listenOn  $ PortNumber port
@@ -856,7 +812,7 @@ 
    initFinish
    onFinish $ const $ do
-             return()                                 -- !> "onFinish closures receivedd with LISTEN"
+--             return()                   !> "onFinish closures receivedd with LISTEN"
              let Connection{closures=closures}= conn  -- !> "listenNew closures empty"
              liftIO $ modifyMVar_ closures $ const $ return M.empty
 
@@ -990,18 +946,21 @@ emptyPool= liftIO $ newMVar  []
 
 
-createNodeServ :: HostName -> Integer -> [Service] -> Node
-createNodeServ h p svs= Node h ( fromInteger p) (unsafePerformIO emptyPool) svs
+createNodeServ ::  HostName -> Integer -> [Service] -> IO Node
+createNodeServ h p svs= do
+    pool <- emptyPool
+    return $ Node h ( fromInteger p) pool svs
 
 
 
 
-createNode :: HostName -> Integer -> Node
+createNode :: HostName -> Integer -> IO Node
 createNode h p= createNodeServ h p []
 
-createWebNode :: Node
-createWebNode= Node "webnode" ( fromInteger 0) (unsafePerformIO emptyPool)
-                      [("webnode","")]
+createWebNode :: IO Node
+createWebNode= do
+  pool <- emptyPool
+  return $ Node "webnode" ( fromInteger 0) pool  [("webnode","")]
 
 
 instance Eq Node where
@@ -1153,10 +1112,10 @@ -- | synchronize the list of nodes with a remote node and all the nodes connected to it
 -- the final effect is that all the nodes reachable share the same list of nodes
 connect'  remotenode= do
-    nodes <- local $ getNodes
+    nodes <- local getNodes
     local $ liftIO $ putStrLn $ "connecting to: "++ show remotenode
 
-    newNodes <- runAt remotenode $  do
+    newNodes <- runAt remotenode $ do
            local $ do
               conn@(Connection _(Just (Node2Node _ _ _)) _ _ _ _ _) <- getSData <|>
                error ("connect': need to be connected to a node: use wormhole/connect/listen")

@@ -56,10 +56,11 @@ 
 install :: String  -> String -> Int -> Cloud ()
 install package program port =  do
+     exist <- localIO $ findExecutable program -- liftIO $ doesDirectoryExist  packagename
      let packagename = name package
-     when (null packagename) $ error $ "source for \""++package ++ "\" not found"
-     exist <- local $ liftIO $ doesDirectoryExist  packagename
-     when (not exist) $ local $ liftIO $ do
+     when (isNothing exist) $ local $ liftIO $ do
+
+         when (null packagename) $ error $ "source for \""++package ++ "\" not found"
          callProcess  "git" ["clone",package]
          liftIO $ putStr package >> putStrLn " cloned"
          setCurrentDirectory packagename
@@ -68,7 +69,7 @@          return()
      let prog = pathExe packagename program port
      lliftIO $ print $ "executing "++ prog
-     local  $ (async $ do createProcess $ shell prog ; return ()) <|> return ()
+     localIO $ do createProcess $ shell prog ; return ()  -- ) <|> return ()
 
 
      return() -- !> "INSTALLED"
@@ -88,35 +89,44 @@ initService ident service@(package, program)= loggedc $ do
     nodes <- local getNodes
     case find (\node  -> service `elem` nodeServices node) nodes  of
-       Just node -> return node
+       Just node -> return node                 -- !> "found"
        Nothing -> do
 
-          nodes <- callOne $ \thisNode -> do
+          node <- runAt (head nodes)  $ do
+                    thisNode <- local getMyNode
                     yn<- requestService ident service
-                    if yn then do
+                    if yn   then do
                         port <- onAll freePort
+                        localIO $ putStr "installing " >> putStrLn package
                         install package program  port
                         nodeService thisNode port
                       else empty
-          local $ addNodes nodes
-          return $ head nodes   -- !> ("GENERATED NODE", nodes)
+          local $ addNodes [node]
+          return node
     where
-    nodeService (Node h _ _ _) port= local $
-       return [Node h port (unsafePerformIO $ newMVar []) [service] ]  -- !> (thisNode,port)
-
+    nodeService (Node h _ _ _) port= localIO $ do
+       pool <- newMVar []
+       return $ Node h port pool [service]
 
+callOne :: Loggable a =>  Cloud a ->  Cloud a
+callOne = callNodes after empty
 
-callOne mx= callNodes' (<>) empty mx
- where
+after mx my= waitone mx <|> my
+  where
+  waitone mx = local $ do
+       rs <- collect' 1 1 0 $ runCloud mx
+       return $ head rs
 
- callNodes' op init proc= loggedc $ do
-    nodes <-  local getNodes
-    let nodes' = filter (not . isWebNode) nodes
-    foldr op init $ map (\node -> runAt node $ proc node) nodes'  :: Cloud [Node]
-    where
-    isWebNode Node {nodeServices=srvs}
-         | ("webnode","") `elem` srvs = True
-         | otherwise = False
+-- where
+--
+-- callNodes' op init proc= loggedc $ do
+--    nodes <-  local getNodes
+--    let nodes' = filter (not . isWebNode) nodes
+--    foldr op init $ map (\node -> runAt node $ proc node) nodes'  :: Cloud [Node]
+--    where
+--    isWebNode Node {nodeServices=srvs}
+--         | ("webnode","") `elem` srvs = True
+--         | otherwise = False
 
 
 rfriends        =   unsafePerformIO $ newMVar []
@@ -143,8 +153,10 @@     :: (Loggable a, Loggable b)
     => String -> Service -> a  -> Cloud b
 callService ident service params = do
+
     node <-  initService ident service
-    localIO $ print node
+
+
     log <- onAll $ do
            log  <- getSData <|> return emptyLog
            setData emptyLog
@@ -155,12 +167,10 @@ 
              teleport
              local empty
---    return () !> ("r=",r)
+
     restoreLog log
---    local $ do
---       Log _ _ log <- getSData <|> return emptyLog
---       return() !> ("log after",log)
-    return  r -- (r `asTypeOf` witness)
+
+    return  r
     where
     restoreLog (Log _ _ logw)= onAll $ do
        Log _ _ logw' <- getSData <|> return emptyLog
@@ -175,8 +185,10 @@ 
 runEmbeddedService :: (Loggable a, Loggable b) =>  Service -> (a -> Cloud b) -> Cloud b
 runEmbeddedService servname serv =  do
-   port <- lliftIO $ freePort
-   listen $ createNodeServ "localhost" (fromIntegral port) [servname]
+   node <- localIO $ do
+          port <- freePort
+          createNodeServ "localhost" (fromIntegral port) [servname]
+   listen node
    wormhole notused $ loggedc $ do
       x <- local $ return notused
       r <- onAll $ runCloud (serv x) <** setData WasRemote
@@ -202,7 +214,7 @@    initNodeServ servs=do
       mynode <- local $ do
         port <-  getPort
-        return $ createNodeServ "localhost" port servs
+        liftIO $ createNodeServ "localhost" port servs
 
       listen mynode -- <|> return()
       where
@@ -210,6 +222,21 @@       getPort =  if isBrowserInstance then return 0 else do
           oneThread $ option "start" "re/start node"
           input (const True) "port to listen? "
+
+{- |
+a service called monitor:
+  runService
+  receive request for a service.
+  check service in list
+  if executing return node
+  when not installed install
+  execute
+  return node
+-}
+--localServiceMonitor ident service = keep $ runCloud  $
+--    runService ("https://github.com/agocorona/transient-universe","monitor") $ do
+--       initService ident service
+
 
 
 

@@ -48,20 +48,20 @@ --
 -- To translate the code from the browser to the server node, use `teleport`.
 --
-
+initNode :: Cloud () -> TransIO ()
 initNode app= do
-   node <- getPort
+   node <- getNodeParams
    initWebApp node  app
 
 
   where
-  getPort :: TransIO Node
-  getPort =
-      if isBrowserInstance then return createWebNode else do
+  getNodeParams  :: TransIO Node
+  getNodeParams  =
+      if isBrowserInstance then  liftIO createWebNode else do
           oneThread $ option "start" "re/start node"
           host <- input (const True) "hostname of this node (must be reachable): "
           port <- input (const True) "port to listen? "
-          return $ createNode host port
+          liftIO $ createNode host port
 
 -- | ask for nodes to be added to the list of known nodes. it also ask to connect to the node to get
 -- his list of known nodes
@@ -76,7 +76,7 @@           port <-  local $ input (const True) "port?"
 
           connectit <- local $ input (\x -> x=="y" || x== "n") "connect to get his list of nodes?"
-          let nnode= createNode host port
+          nnode <- localIO $ createNode host port
           if connectit== "y" then connect'  nnode
                              else local $ addNodes [nnode]
    empty
@@ -100,19 +100,21 @@ --
 --
 simpleWebApp :: Integer -> Cloud () -> IO ()
-simpleWebApp port app =  keep $ initWebApp (createNode "localhost" port) app
+simpleWebApp port app = do
+   node <- createNode "localhost" port
+   keep $ initWebApp node app
 
 -- | use this instead of smpleWebApp when you have to do some initializations in the server prior to the
 -- initialization of the web server
 initWebApp :: Node -> Cloud () -> TransIO ()
 initWebApp node app=  do
-    let conn= defConnection 8192
+    conn <- defConnection
     setData  conn{myNode = node}
-    serverNode  <-  getWebServerNode  :: TransIO Node
+    serverNode  <-   getWebServerNode  :: TransIO Node
 
-    let mynode = if isBrowserInstance
-                    then createWebNode
-                    else serverNode
+    mynode <- if isBrowserInstance
+                    then liftIO $ createWebNode
+                    else return serverNode
 
     runCloud $ do
         listen mynode <|> return()
@@ -135,6 +137,6 @@ -- | run N nodes (N ports to listen) in the same program. For testing purposes.
 -- It add them to the list of known nodes, so it is possible to perform `clustered` operations with them.
 runTestNodes ports= do
-    let nodes=  map (\p -> createNode "localhost" p) ports
+    nodes <- onAll $  mapM (\p -> liftIO $ createNode "localhost" p) ports
     foldl (<|>) empty (map listen nodes) <|> return()
 

@@ -1,5 +1,5 @@ name: transient-universe
-version: 0.3.2.1
+version: 0.3.2.2
 cabal-version: >=1.10
 build-type: Simple
 license: MIT