transient-universe 0.5.0.0 → 0.6.0.0
raw patch · 17 files changed
+6179/−3262 lines, 17 filesdep +aesondep +base64-bytestringdep +old-timedep ~basedep ~networkdep ~processnew-component:exe:controlServicesnew-component:exe:executor
Dependencies added: aeson, base64-bytestring, old-time
Dependency ranges changed: base, network, process, transient, websockets
Files
- README.md +22/−0
- app/client/Transient/Move/Services/MonitorService.hs +0/−1
- app/client/Transient/Move/Services/void.hs +1/−0
- app/server/Transient/Move/Services/MonitorService.hs +356/−158
- app/server/Transient/Move/Services/MonitorService.hsvoid.hs +0/−0
- app/server/Transient/Move/Services/controlServices.hsvoid.hs +0/−0
- app/server/Transient/Move/Services/executor.hs +192/−0
- app/server/Transient/Move/Services/executor.hsvoid.hs +0/−0
- src/Transient/MapReduce.hs +588/−510
- src/Transient/Move.hs +8/−7
- src/Transient/Move/Internals.hs +3340/−2109
- src/Transient/Move/PubSub.hs +152/−0
- src/Transient/Move/Services.hs +792/−111
- src/Transient/Move/Services/Executor.hs +211/−0
- src/Transient/Move/Utils.hs +248/−187
- tests/TestSuite.hs +36/−24
- transient-universe.cabal +233/−155
README.md view
@@ -19,6 +19,10 @@ ======= ```haskell +import Transient.Base +import Transient.Move +import Control.Monad + main= keep . initNode $ inputNodes <|> mypPogram myProgram :: Cloud () @@ -43,6 +47,24 @@ Distributed Browser/server Widgets ------- Browser nodes can integrate a reactive client side library based in trasient (package [axiom](https://github.com/transient-haskell/axiom)). These widgets can create widgets with HTML form elements and control the server nodes. A computation can move from browser to server and back despite the different architecture. + +This program will obtain a string from the browser, will send it to the server, which will return three responses wich will be presented in the browser: + +```haskell +import Transient.Base +import Transient.Move +import Transient.Indeterminism +import GHCJS.HPlay.View + +main= keep . initNode $ myProgram + +myProgram :: Cloud () +myProgram= do + name <- local . render $ getString Nothing `fire` OnChange + r <- atRemote . local . choose . take 3 . repeat $ "hello "++ name + local . render . rawHtml $ h1 r +``` +See the package Axiom for instructions about how to compile and run this program. 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.
− app/client/Transient/Move/Services/MonitorService.hs
@@ -1,1 +0,0 @@-main= return ()
+ app/client/Transient/Move/Services/void.hs view
@@ -0,0 +1,1 @@+main= return ()
app/server/Transient/Move/Services/MonitorService.hs view
@@ -1,158 +1,356 @@------------------------------------------------------------------------------ --- --- Module : Transient.Move.Services.MonitorService --- Copyright : --- License : MIT --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | --- ------------------------------------------------------------------------------ -{-# LANGUAGE ScopedTypeVariables #-} -module Main where - -import Transient.Internals - -import Transient.Move.Internals -import Transient.Move.Utils -import Transient.Move.Services -import Control.Applicative -import Control.Monad.IO.Class -import Control.Exception(SomeException(..)) -import Control.Concurrent -import Control.Monad -import Data.List -import System.Process -import System.Directory -import Data.Monoid - -main = keep . runCloud $ do - runService monitorService 3000 $ \(ident,service,num) -> do - return () !> ("RUNSERVICE",ident, service, num) - nodes <- local $ findInNodes service >>= return . take num - -- onAll $ liftIO $ print ("NODES",nodes) - - - let n= num - length nodes - if n==0 then return nodes - else return nodes <> requestInstall ident service n - where - - requestInstall :: String -> Service -> Int -> Cloud [ Node] - requestInstall ident service num= do - ns <- local getEqualNodes - -- return () !> ("equal",ns) - auth <- callNodes' ns (<>) mempty $ localIO $ authorizeService ident service >>= \x -> return [x] - -- return () !> auth - let nodes = map fst $ filter snd $ zip ns auth - nnodes= length nodes - pernode= num `div` nnodes - lacking= num `rem` nnodes - (nodes1,nodes2)= splitAt lacking nodes - -- return () !> (pernode,lacking,nodes1,nodes2) - rs <- callNodes' nodes1 (<>) mempty (installHere ident service (pernode+1)) <> - callNodes' nodes2 (<>) mempty (installHere ident service pernode) - local $ addNodes rs - return rs -- !> ("MONITOR RETURN---------------------------------->", rs) - - -- installIt = installHere ident service <|> installThere ident service - installHere :: String -> Service -> Int -> Cloud [ Node] - installHere ident service n= local $ replicateM n installOne - where - installOne= do - port <- liftIO freePort - install service port - return () !> "INSTALLED" - - thisNode <- getMyNode - let node= Node (nodeHost thisNode) port Nothing service -- node to be published - nodelocal= Node "localhost" port Nothing [("externalNode", show $ node{nodeServices=[]})] -- local node - addNodes [node{nodeServices=("localNode", show nodelocal{nodeServices=[]}):nodeServices node},nodelocal ] - return node {nodeServices= nodeServices node ++ [("relay",show thisNode{nodeServices=[]})]} - `catcht` \(e :: SomeException) -> liftIO (putStr "INSTALLLLLLLLLLLLLLL2222222: " >> print e) >> empty - - - --- nodeService node@(Node h _ _ _) port service= Node h port Nothing $ service -- ++ [("relay",show $node{nodeServices=[]}) - - - -install :: Service -> Int -> TransIO () - -install service port= do - -- 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 program - <|> do tryInstall service ; tryExec program host port - -emptyIfNothing :: Maybe a -> TransIO a -emptyIfNothing = Transient . return - -tryInstall :: Service -> TransIO () -tryInstall service = do - package <- emptyIfNothing (lookup "package" service) - install package - where - install package - | "git:" `isPrefixOf` package= installGit package - | "https://github.com" `isPrefixOf` package = installGit package - | "http://github.com" `isPrefixOf` package = installGit package - - -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++ " " ++ program] - - -tryExec program host port= do - path <- Transient $ liftIO $ findExecutable program -- !> ("findExecutable", program) - spawnProgram program host port -- !>"spawn" - where - spawnProgram program host port= liftIO $ do - - let prog = pathExe program host port - putStr "executing: " >> putStrLn prog - let createprostruct= shell prog - createProcess $ createprostruct ; return () - - threadDelay 2000000 - - -- return() !> ("INSTALLED", program) - where - - pathExe program host port= - program ++ " -p start/" ++ show (host ::String) - ++"/" ++ show (port ::Int) ++ " > "++ program ++ host ++ show port ++ ".log" - - - - - - -installGit package = liftIO $ do - - let packagename = name package - when (null packagename) $ error $ "source for \""++package ++ "\" not found" - callProcess "git" ["clone",package] - liftIO $ putStr package >> putStrLn " cloned" - setCurrentDirectory packagename - callProcess "cabal" ["install","--force-reinstalls"] - setCurrentDirectory ".." - - - where - name url= slash . slash . slash $ slash url - where - slash= tail1 . dropWhile (/='/') - tail1 []=[] - tail1 x= tail x - +#!/usr/bin/env execthirdlinedocker.sh+-- info: use sed -i 's/\r//g' file if report "/usr/bin/env: ‘execthirdlinedocker.sh\r’: No such file or directory"+-- LIB="/projects/transient-stack" && runghc -DDEBUG -i${LIB}/transient/src -i${LIB}/transient-universe/src -i${LIB}/transient/src -i${LIB}/transient-universe-tls/src -i${LIB}/axiom/src $1 ${2} ${3}++-----------------------------------------------------------------------------+--+-- Module : Transient.Move.Services.MonitorService+-- Copyright :+-- License : MIT+--+-- Maintainer : agocorona@gmail.com+-- Stability :+-- Portability :+--+-- |+--+-----------------------------------------------------------------------------+{-# LANGUAGE ScopedTypeVariables #-}+module Main where++import Transient.Internals+import Transient.Mailboxes+import Transient.Logged+import Transient.Indeterminism(choose)+import Transient.Move.Internals+import Transient.Move.Utils+import Transient.Move.Services+import Control.Applicative+import Control.Monad.IO.Class+import Control.Exception(SomeException(..))+import Control.Concurrent+import Control.Monad+import Data.List+import System.IO+import System.Process+import System.Directory+import Data.Monoid+import Unsafe.Coerce+import System.IO.Unsafe+import Data.IORef+import qualified Data.Map as M+-- import GHC.Conc+import Data.Maybe(fromMaybe)+import Control.Exception+import qualified Data.ByteString.Lazy.Char8 as BS+import qualified Data.ByteString.Char8 as BSS+import System.Exit +++ +main = do+ putStrLn "Starting Transient monitor"+ keep $ runService monitorService 3000 + [serve receiveStatus+ ,serve returnInstances+ ,serve reReturnInstances+ ,serve receiveFromNodeStandardOutputIt+ ,serve sendToNodeStandardInputIt+ ,serve getLogIt+ ]+ (return ()) ++++{- ping is not used to determine healt of services. The client program notify the+ monitor wShen a service fails, with reInitService.+pings = do+ + localIO $ print $ "INITIATING PINGSSSSSSSSSSSSSSSSSSSSSSS"+ local $ threads 0 $ choose ([1..] :: [Int])++ nodes <- local getNodes + return () !> ("NODES=", length nodes)+ + localIO $ threadDelay 10000000 ++ local $ threads 1 $ runCloud $ mapM ping $ tail nodes+ empty+-}+ + +type Port= Int++-- | receive a status from an executable.+receiveStatus :: (Port, String) -> Cloud ()+receiveStatus (port, logLine)= do+ localIO $ appendFile ("log"++ show port) $ logLine++"\n"+ ++blockings= unsafePerformIO $ newIORef M.empty+++withBlockingService :: Service -> Cloud a -> Cloud a+withBlockingService serv proc= do+ beingDone <- localIO $ atomicModifyIORef blockings $ \map -> + let mv = M.lookup serv map+ in case mv of+ Nothing -> (M.insert serv () map,False)+ Just () -> (map,True)+ if beingDone + then do+ --localIO $ threadDelay 3000000+ withBlockingService serv proc+ else do+ r <- proc+ localIO $ atomicModifyIORef blockings $ \map -> (M.delete serv map,())+ return r++-- | gets a node with a service, which probably failed and return other n instances of the same service.+-- This is used to implement failover.+reReturnInstances :: (String, Node, Int) -> Cloud [Node] +reReturnInstances (ident, node, num)= do+ local $ delNodes [node]+ returnInstances (ident, head $ nodeServices node, num)++-- | install and return n instances of a service, distributed+-- among all the nodes which have monitoService executables running and connected +returnInstances :: (String, Service, Int) -> Cloud [Node] +returnInstances (ident, service, num)= withBlockingService service $ do+ nodes <- local $ findInNodes service >>= return . take num++ let n= num - length nodes+ if n <= 0 then return $ take num nodes + else return nodes <> requestInstall ident service n ++ where++ requestInstall :: String -> Service -> Int -> Cloud [ Node]+ requestInstall ident service num= do+ ns <- local getEqualNodes + return () !> ("monitors: ",map nodeHost ns) + auth <- callNodes' ns (<>) mempty $ localIO $ authorizeService ident service >>= \x -> return [x]+ return () !> ("authotorized: ",auth)+ let nodes = map fst $ filter snd $ zip ns auth + nnodes= length nodes+ pernode= num `div` nnodes+ lacking= num `rem` nnodes+ (nodes1,nodes2)= splitAt lacking nodes+ return () !> (pernode,lacking,nodes1,nodes2)+ rs <- callNodes' nodes1 (<>) mempty (installHere service (pernode+1)) <> + callNodes' nodes2 (<>) mempty (installHere service pernode)+ local $ addNodes rs + --ns <- onAll getNodes+ tr ("MONITOR RETURN---------------------------------->", rs)+ return rs + + -- installIt = installHere service <|> installThere service+ installHere :: Service -> Int -> Cloud [ Node]+ installHere service n= local $ replicateM n installOne+ where+ installOne= do+ port <- liftIO freePort+ install service port+ return () !> ("INSTALLED",n)++ thisNode <- getMyNode+ let node= Node (nodeHost thisNode) port Nothing ([service]) -- ++ [relayinfo thisNode]) -- node to be published+ addNodes [node] + return node+ `catcht` \(e :: SomeException) -> liftIO (putStr "INSTALL error: " >> print e) >> empty+ + relayinfo mon= if nodeHost mon /= "localhost" then [("relay",show(nodeHost mon,nodePort mon))] else []++++install :: Service -> Int -> TransIO ()++install service port= do+ -- return () !> "IIIIIIIIIIIIIIINSTALL"++ install' `catcht` \(e :: SomeException) -> liftIO (putStr "INSTALL error: " >> print e) >> empty + where+ install'= do+ my <- getMyNode+ let host= nodeHost my+ program <- return (lookup "executable" service) `onNothing` empty+ -- return () !> ("program",program)+ tryExec program host port <|> tryDocker service host port program+ <|> do tryInstall service ; tryExec program host port+++tryInstall :: Service -> TransIO ()+tryInstall service = do + package <- emptyIfNothing (lookup "package" service) + install package+ where + install package+ | "git:" `isPrefixOf` package= installGit package + | "https://github.com" `isPrefixOf` package = installGit package + | "http://github.com" `isPrefixOf` package = installGit package +++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++ " " ++ program]+++tryExec program host port= do+ path <- Transient $ liftIO $ findExecutable program -- would abandon (empty) if the executable is not found+ spawnProgram program host port -- !>"spawn"+ where+ spawnProgram program host port= do++ let prog = pathExe program host port+ liftIO $ putStr "executing: " >> putStrLn prog++ (networkExecuteStreamIt prog >> empty) <|> return () !> "INSTALLING"+ liftIO $ threadDelay 2000000++ return() !> ("INSTALLED", program,port)+ + +pathExe program host port=+ program ++ " -p start/" ++ (host ::String) + ++"/" ++ show (port ::Int) -- ++ " > "++ program ++ host ++ show port ++ ".log 2>&1"++++++ +installGit package = liftIO $ do+ let packagename = name package+ when (null packagename) $ error $ "source for \""++package ++ "\" not found"+ callProcess "git" ["clone",package]+ liftIO $ putStr package >> putStrLn " cloned"+ setCurrentDirectory packagename + callProcess "cabal" ["install","--force-reinstalls"]+ setCurrentDirectory ".."+++ where+ name url= slash . slash . slash $ slash url+ where+ slash= tail1 . dropWhile (/='/')+ tail1 []=[]+ tail1 x= tail x+++-------------------------execution ----------------------------++getLogIt :: GetLog -> Cloud BS.ByteString+getLogIt (GetLog node)= do+ let program = fromMaybe (error "no Executable in service "++ show (nodeServices node)) $+ lookup2 "executable" (nodeServices node)+ let expr = pathExe program (nodeHost node) (nodePort node)+ localIO $ BS.readFile $ logFileName expr+++sendToNodeStandardInputIt :: (Node, String) -> Cloud ()+sendToNodeStandardInputIt (node,inp)= do+ let program = fromMaybe (error "no Executable in service "++ show (nodeServices node)) $+ lookup2 "executable" (nodeServices node)+ expr= pathExe program (nodeHost node) (nodePort node)+ return () !> ("SEND TO NODE STANDARD INPUT", program, expr)+ sendExecuteStreamIt1 (expr, inp)+ where+ sendExecuteStreamIt1 (cmdline, inp)= localIO $ do+ map <- readIORef rinput + let input1= fromMaybe (error "this command line has not been opened") $ M.lookup cmdline map + hPutStrLn input1 inp + hFlush input1+ return()+ +receiveFromNodeStandardOutputIt :: ReceiveFromNodeStandardOutput -> Cloud String+receiveFromNodeStandardOutputIt (ReceiveFromNodeStandardOutput node ident) = local $ do+ let program = fromMaybe (error "no Executable in service "++ show (nodeServices node)) $+ lookup2 "executable" (nodeServices node)+ expr= pathExe program (nodeHost node) (nodePort node)+ return () !> ("RECEIVE FROM STANDARD OUTPUT",expr)+ labelState ident+ getMailbox' ("output"++ expr)++rinput :: IORef (M.Map String Handle)+rinput= unsafePerformIO $ newIORef M.empty +++logFolder= "./log/"++logFileName ('.':expr) = logFileName expr+logFileName expr= logFolder ++ subst expr ++ ".log"+ where+ subst []= [] + subst (' ':xs)= '-':subst xs+ subst ('/':xs)= '-':subst xs+ subst ('\"':xs)= '-':subst xs+ subst (x:xs)= x:subst xs ++-- | execute the shell command specified in a string and stream back at runtime -line by line- the standard output+-- as soon as there is any output. It also stream all the standard error in case of exiting with a error status.+-- to the service caller. invoked by `networkExecuteStream`.+++ +networkExecuteStreamIt :: String -> TransIO String+networkExecuteStreamIt expr = do+ liftIO $ createDirectoryIfMissing True logFolder+ blocked <- liftIO $ newMVar () + r <- liftIO $ createProcess $ (shell expr){std_in=CreatePipe,std_err=CreatePipe,std_out=CreatePipe}+ liftIO $ atomicModifyIORef rinput $ \map -> (M.insert expr (input1 r) map,())+ + let logfile= logFileName expr ++ hlog <- liftIO $ openFile logfile WriteMode + liftIO $ hPutStrLn hlog expr+ liftIO $ hClose hlog + + line <- watch (output r) <|> watch (err r) <|> watchExitError r + putMailbox' ("output" ++ expr) line+ liftIO $ withMVar blocked $ const $ do+ hlog <- openFile logfile AppendMode + hPutStrLn hlog line+ hClose hlog + return line+ where++ input1 r= inp where (Just inp,_,_,_)= r+ output r= out where (_,Just out,_,_)= r+ err r= err where (_,_,Just err,_)= r+ handle r= h where (_,_,_,h)= r++ watch :: Handle -> TransIO String+ watch h= do+ abduce+ mline <- threads 0 $ (parallel $ (SMore <$> hGetLine' h) `catch` \(e :: SomeException) -> return SDone)+ case mline of+ SDone -> empty+ SMore line -> return line+ + where++ hGetLine' h= do+ buff <- newIORef []+ getMore buff+ + where++ getMore buff= do+ b <- hWaitForInput h 10+ if not b+ then do+ r <-readIORef buff+ if null r then getMore buff else return r+ else do+ c <- hGetChar h+ if c== '\n' then readIORef buff else do+ modifyIORef buff $ \str -> str ++ [c]+ getMore buff++ watchExitError r= do -- make it similar to watch+ abduce+ liftIO $ waitForProcess $ handle r+ errors <- liftIO $ hGetContents (err r)+ return errors+
+ app/server/Transient/Move/Services/MonitorService.hsvoid.hs view
+ app/server/Transient/Move/Services/controlServices.hsvoid.hs view
+ app/server/Transient/Move/Services/executor.hs view
@@ -0,0 +1,192 @@+----------------------------------------------------------------------------- +-- +-- Module : Transient.Move.Services.Executor +-- Copyright : +-- License : MIT +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +--f +-- | +-- +----------------------------------------------------------------------------- +{-# LANGUAGE ScopedTypeVariables #-} +module Main where + +import Transient.Internals +import Transient.Mailboxes +import Transient.Move.Services.Executor +import Transient.Move.Internals +import Transient.Move.Utils +-- import Transient.Logged(maybeFromIDyn) +import Transient.Move.Services +import Control.Applicative +import Control.Monad.IO.Class +import Control.Exception(SomeException(..),catch) +import Control.Concurrent +import Control.Monad +import Data.List +import System.Process +import System.Directory +import Data.Monoid +import Data.IORef +import System.IO +import System.IO.Unsafe +import qualified Data.Map as M +import Data.Maybe +import qualified Data.ByteString.Lazy.Char8 as BS +import qualified Data.ByteString.Char8 as BSS +import Data.String +import Data.Time + + +main = do + putStrLn "Starting Transient Executor Service" + keep $ runService executorService 3005 + [ serve networkExecuteStreamIt + , serve networkExecuteIt + , serve sendExecuteStreamIt + , serve receiveExecuteStreamIt + , serve networkExecuteStreamIt' + , serve getLogIt + , serve getProcessesIt] + (return ()) + +getProcessesIt :: GetProcesses -> Cloud [String] +getProcessesIt _= localIO $ do + map1 <- readIORef rinput + return $ map fst $ M.toList map1 + +-- | send input to a remote process initiated with `networkExecuteStream` or `networkExecuteStream'` +sendExecuteStreamIt :: (String,String) -> Cloud () +sendExecuteStreamIt (cmdline, inp)= do + localIO $ do + map <- readIORef rinput + let input= fromMaybe (error "this command line has not been opened") $ M.lookup cmdline map + hPutStrLn input inp + hFlush input + return () + +-- receive input from a remote process initiated with `networkExecuteStream'` +receiveExecuteStreamIt :: ReceiveExecuteStream -> Cloud String +receiveExecuteStreamIt (ReceiveExecuteStream expr ident)= local $ do + labelState ident + getMailbox' ("output"++ expr) + +-- | execute a shell script and a input, and return all the output. Called externally by `networkExecute` +networkExecuteIt :: (String, String, ()) -> Cloud String +networkExecuteIt (expr, input,()) = localIO $ readCreateProcess (shell expr) input + +getLogIt :: GetLogCmd -> Cloud BS.ByteString +getLogIt (GetLogCmd cmd)= localIO $ BS.readFile $ logFileName cmd + + +logFileName ('.':expr) = logFileName expr +logFileName expr= logFolder ++ subst expr ++ ".log" + where + subst []= [] + subst (' ':xs)= '-':subst xs + subst ('/':xs)= '-':subst xs + subst ('\"':xs)= '-':subst xs + subst (x:xs)= x:subst xs + +networkExecuteStreamIt' :: ExecuteStream -> Cloud String +networkExecuteStreamIt' (ExecuteStream expr) = local $ do + + setRState False + + r <- executeStreamIt expr + + + + init <- getRState + if init then empty + else do + setRState True + return r -- return the first output line only + + +-- execute the shell command specified in a string and stream line by line the standard output/error +-- to the service caller. It also store the output in a logfile and update a mailbox that can be +-- inspected by `receiveExecuteStreamIt`. Invoked by `networkExecuteStream`. +-- The first result returned is the process identifier. +networkExecuteStreamIt :: String -> Cloud String +networkExecuteStreamIt expr = local $ executeStreamIt expr + +logFolder= "./.log/" + +executeStreamIt expr = do + liftIO $ createDirectoryIfMissing True logFolder + r <- liftIO $ createProcess $ (shell expr){std_in=CreatePipe,std_err=CreatePipe,std_out=CreatePipe} + + time <- liftIO $ getCurrentTime + let header= expr ++" "++ show time + abduce + labelState $ BSS.pack header + + + onException $ \(e :: SomeException) -> do + liftIO $ do + print ("watch:",e) + cleanupProcess r + atomicModifyIORef rinput $ \map -> (M.delete header map,()) + empty + + let logfile= logFileName header + let box= "output" ++ header + liftIO $ atomicModifyIORef rinput $ \map -> (M.insert header (input1 r) map,()) + + line <- async (return header) <|> watch (output r) <|> watch (err r) <|> watchExitError r + + putMailbox' box line + + hlog <- liftIO $ openFile logfile AppendMode + liftIO $ hPutStrLn hlog line + liftIO $ hClose hlog + return line + + where + + input1 r= inp where (Just inp,_,_,_)= r + output r= out where (_,Just out,_,_)= r + err r= err where (_,_,Just err,_)= r + handle r= h where (_,_,_,h)= r + + watch :: Handle -> TransIO String + watch h = do + abduce + mline <- threads 0 $ (parallel $ (SMore <$> hGetLine' h) `catch` \(e :: SomeException) -> return SDone) + case mline of + SDone -> empty + SError e -> do liftIO $ print ("watch:",e); empty + SMore line -> return line + + where + + hGetLine' h= do + buff <- newIORef [] + getMore buff + + where + + getMore buff= do + b <- hWaitForInput h 10 + if not b + then do + r <-readIORef buff + if null r then getMore buff else return r + else do + c <- hGetChar h + if c == '\n' then readIORef buff else do + modifyIORef buff $ \str -> str ++ [c] + getMore buff + + watchExitError r= do -- make it similar to watch + abduce + liftIO $ waitForProcess $ handle r + errors <- liftIO $ hGetContents (err r) + return errors + + +rinput= unsafePerformIO $ newIORef M.empty
+ app/server/Transient/Move/Services/executor.hsvoid.hs view
src/Transient/MapReduce.hs view
@@ -1,511 +1,589 @@-{-# LANGUAGE ExistentialQuantification, DeriveDataTypeable -, FlexibleInstances, MultiParamTypeClasses, OverloadedStrings, CPP #-} - - -module Transient.MapReduce -( -Distributable(..),distribute, getText, -getUrl, getFile,textUrl, textFile, -mapKeyB, mapKeyU, reduce,eval, ---v* internals -DDS(..),Partition(..),PartRef(..)) - where - -#ifdef ghcjs_HOST_OS -import Transient.Base -import Transient.Move hiding (pack) -import Transient.Logged --- dummy Transient.MapReduce module, -reduce _ _ = local stop :: Loggable a => Cloud a -mapKeyB _ _= undefined -mapKeyU _ _= undefined -distribute _ = undefined -getText _ _ = undefined -textFile _ = undefined -getUrl _ _ = undefined -textUrl _ = undefined -getFile _ _ = undefined -eval _= local stop -data Partition -data DDS= DDS -class Distributable -data PartRef a=PartRef a - -#else - -import Transient.Internals hiding (Ref) - -import Transient.Move.Internals hiding (pack) -import Transient.Indeterminism -import Control.Applicative -import System.Random -import Control.Monad.IO.Class - -import Control.Monad -import Data.Monoid - -import Data.Typeable -import Data.List hiding (delete, foldl') -import Control.Exception -import Control.Concurrent ---import Data.Time.Clock -import Network.HTTP -import Data.TCache hiding (onNothing) -import Data.TCache.Defs - -import Data.ByteString.Lazy.Char8 (pack,unpack) -import qualified Data.Map.Strict as M -import Control.Arrow (second) -import qualified Data.Vector.Unboxed as DVU -import qualified Data.Vector as DV -import Data.Hashable -import System.IO.Unsafe - -import qualified Data.Foldable as F -import qualified Data.Text as Text -import Data.IORef - -data DDS a= Loggable a => DDS (Cloud (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 b _)= keyp string b - - - -keyp s True= "PartP@"++s :: String -keyp s False="PartT@"++s - -instance Loggable a => IResource (Partition a) where - keyResource= key - readResourceByKey k= r - where - typePart :: IO (Maybe a) -> a - typePart = undefined - r = if k !! 4 /= 'P' then return Nothing else - defaultReadByKey (defPath (typePart r) ++ k) >>= return . fmap ( read . unpack) - writeResource (s@(Part _ _ save _))= - unless (not save) $ defaultWrite (defPath s ++ key s) (pack $ show s) - - -eval :: DDS a -> Cloud (PartRef a) -eval (DDS mx) = mx - - -type Path=String - - -instance F.Foldable DVU.Vector where - {-# INLINE foldr #-} - foldr = foldr - - {-# INLINE foldl #-} - foldl = foldl - - {-# INLINE foldr1 #-} - foldr1 = foldr1 - - {-# INLINE foldl1 #-} - foldl1 = foldl1 - ---foldlIt' :: V.Unbox a => (b -> a -> b) -> b -> V.Vector a -> b ---foldlIt' f z0 xs= V.foldr f' id xs z0 --- where f' x k z = k $! f z x --- ---foldlIt1 :: V.Unbox a => (a -> a -> a) -> V.Vector a -> a ---foldlIt1 f xs = fromMaybe (error "foldl1: empty structure") --- (V.foldl mf Nothing xs) --- where --- mf m y = Just (case m of --- Nothing -> y --- Just x -> f x y) - -class (F.Foldable c, Typeable c, Typeable a, Monoid (c a), Loggable (c a)) => Distributable c a where - singleton :: a -> c a - 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 - fromList = DV.fromList - -instance (Loggable a,DVU.Unbox a) => Distributable DVU.Vector a where - singleton= DVU.singleton - splitAt= DVU.splitAt - fromList= DVU.fromList - - - - --- | perform a map and partition the result with different keys using boxed vectors --- The final result will be used by reduce. -mapKeyB :: (Loggable a, Loggable b, Loggable k,Ord k) - => (a -> (k,b)) - -> DDS (DV.Vector a) - -> DDS (M.Map k(DV.Vector b)) -mapKeyB= mapKey - --- | perform a map and partition the result with different keys using unboxed vectors --- The final result will be used by reduce. -mapKeyU :: (Loggable a, DVU.Unbox a, Loggable b, DVU.Unbox b, Loggable k,Ord k) - => (a -> (k,b)) - -> DDS (DVU.Vector a) - -> DDS (M.Map k(DVU.Vector b)) -mapKeyU= mapKey - --- | perform a map and partition the result with different keys. --- The final result will be used by reduce. -mapKey :: (Distributable vector a,Distributable vector b, Loggable k,Ord k) - => (a -> (k,b)) - -> DDS (vector a) - -> DDS (M.Map k (vector b)) -mapKey f (DDS mx)= DDS $ loggedc $ do - refs <- mx - process refs -- !> ("process",refs) - - where --- process :: Partition a -> Cloud [Partition b] - process (ref@(Ref node path sav))= runAt node $ local $ do - xs <- getPartitionData ref -- !> ("CMAP", ref,node) - (generateRef $ map1 f xs) - - - --- map1 :: (Ord k, F.Foldable vector) => (a -> (k,b)) -> vector a -> M.Map k(vector b) - map1 f v= F.foldl' f1 M.empty v - where - f1 map x= - let (k,r) = f x - in M.insertWith (<>) k (Transient.MapReduce.singleton r) map - - - -data ReduceChunk a= EndReduce | Reduce a deriving (Typeable, Read, Show) - -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) - -reduce red (dds@(DDS mx))= loggedc $ do - - mboxid <- localIO $ atomicModifyIORef boxids $ \n -> let n'= n+1 in (n',n') - nodes <- local getEqualNodes - - let lengthNodes = length nodes - shuffler nodes = do - localIO $ threadDelay 100000 - ref@(Ref node path sav) <- mx -- return the resulting blocks of the map - - runAt node $ foldAndSend node nodes ref - - stop - --- groupByDestiny :: (Hashable k, Distributable vector a) => M.Map k (vector a) -> M.Map Int [(k ,vector a)] - groupByDestiny map = M.foldlWithKey' f M.empty map - where --- f :: M.Map Int [(k ,vector a)] -> k -> vector a -> M.Map Int [(k ,vector a)] - f map k vs= M.insertWith (<>) (hash1 k) [(k,vs)] map - hash1 k= abs $ hash k `rem` length nodes - - --- foldAndSend :: (Hashable k, Distributable vector a)=> (Int,[(k,vector a)]) -> Cloud () - foldAndSend node nodes ref= do - - pairs <- onAll $ getPartitionData1 ref - <|> return (error $ "DDS computed out of his node:"++ show ref ) - let mpairs = groupByDestiny pairs - - length <- local . return $ M.size mpairs - - let port2= nodePort node - - - if length == 0 then sendEnd nodes else do - - nsent <- onAll $ liftIO $ newMVar 0 - - (i,folded) <- local $ parallelize foldthem (M.assocs mpairs) - - n <- localIO $ modifyMVar nsent $ \r -> return (r+1, r+1) - - (runAt (nodes !! i) $ local $ putMailbox' mboxid (Reduce folded)) - !> ("send",n,length,i,folded) - --- return () !> (port,n,length) - - when (n == length) $ sendEnd nodes - empty - - where - - - foldthem (i,kvs)= async . return - $ (i,map (\(k,vs) -> (k,foldl1 red vs)) kvs) - - - sendEnd nodes = onNodes nodes $ local - $ putMailbox' mboxid (EndReduce `asTypeOf` paramOf dds) --- !> ("send ENDREDUCE ", port)) - - onNodes nodes f = foldr (<|>) empty $ map (\n -> runAt n f) nodes - - sumNodes nodes f= foldr (<>) mempty $ map (\n -> runAt n f) nodes - - reducer nodes= sumNodes nodes reduce1 -- a reduce1 process in each node, get the results and mappend them - --- reduce :: (Ord k) => Cloud (M.Map k v) - - reduce1 = local $ do - reduceResults <- liftIO $ newMVar M.empty - numberSent <- liftIO $ newMVar 0 - - minput <- getMailbox' mboxid -- get the chunk once it arrives to the mailbox - - case minput of - - EndReduce -> do - - n <- liftIO $ modifyMVar numberSent $ \r -> let r'= r+1 in return (r', r') - - - if n == lengthNodes --- !> ("END REDUCE RECEIVED",n, lengthNodes) - then do - cleanMailbox' mboxid (EndReduce `asTypeOf` paramOf dds) - r <- liftIO $ readMVar reduceResults - return r - - else stop - - Reduce kvs -> do - let addIt (k,inp) = do - let input= inp `asTypeOf` atype dds - liftIO $ modifyMVar_ reduceResults - $ \map -> do - let maccum = M.lookup k map - return $ M.insert k (case maccum of - Just accum -> red input accum - Nothing -> input) map - - mapM addIt (kvs `asTypeOf` paramOf' dds) - !> ("Received Reduce",kvs) - stop - - - reducer nodes <|> shuffler nodes - where - atype ::DDS(M.Map k (vector a)) -> a - atype = undefined -- type level - - paramOf :: DDS (M.Map k (vector a)) -> ReduceChunk [( k, a)] - paramOf = undefined -- type level - paramOf' :: DDS (M.Map k (vector a)) -> [( k, a)] - paramOf' = undefined -- type level - - - - --- parallelize :: Loggable b => (a -> Cloud b) -> [a] -> Cloud b -parallelize f xs = foldr (<|>) empty $ map f xs - -mparallelize f xs = loggedc $ foldr (<>) mempty $ map f xs - - -getPartitionData :: Loggable a => PartRef a -> TransIO a -getPartitionData (Ref node path save) = Transient $ do - mp <- (liftIO $ atomically - $ readDBRef - $ getDBRef - $ keyp path save) - `onNothing` error ("not found DDS data: "++ keyp path save) - case mp of - (Part _ _ _ xs) -> return $ Just xs - -getPartitionData1 :: Loggable a => PartRef a -> TransIO a -getPartitionData1 (Ref node path save) = Transient $ do - mp <- liftIO $ atomically - $ readDBRef - $ getDBRef - $ keyp path save - - case mp of - Just (Part _ _ _ xs) -> return $ Just xs - Nothing -> return Nothing - -getPartitionData2 :: Loggable a => PartRef a -> IO a -getPartitionData2 (Ref node path save) = do - mp <- ( atomically - $ readDBRef - $ getDBRef - $ keyp path save) - `onNothing` error ("not found DDS data: "++ keyp path save) - case mp of - (Part _ _ _ xs) -> return xs - --- 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 -> Cloud a -runAtP node f uuid= do - r <- runAt node $ onAll . 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 <- onAll getEqualNodes - let node'= head $ nodes \\ [node] - content <- onAll . liftIO $ readFile uuid - runAt node' $ local $ liftIO $ writeFile uuid content - -sendAnyError :: SomeException -> IO (StreamData a) -sendAnyError e= return $ SError e - - --- | distribute a vector of values among many nodes. --- If the vector is static and sharable, better use the get* primitives --- since each node will load the data independently. -distribute :: (Loggable a, Distributable vector a ) => vector a -> DDS (vector a) -distribute = DDS . distribute' - -distribute' xs= loggedc $ do - nodes <- local getEqualNodes -- !> "DISTRIBUTE" - let lnodes = length nodes - let size= case F.length xs `div` (length nodes) of 0 ->1 ; n -> n - xss= split size lnodes 1 xs -- !> size - r <- distribute'' xss nodes - return r - where - split n s s' xs | s==s' = [xs] - split n s s' xs= - let (h,t)= Transient.MapReduce.splitAt n xs - in h : split n s (s'+1) t - -distribute'' :: (Loggable a, Distributable vector a) - => [vector a] -> [Node] -> Cloud (PartRef (vector a)) -distribute'' xss nodes = - parallelize move $ zip nodes xss -- !> show xss - where - move (node, xs)= runAt node $ local $ do - par <- generateRef xs - return par - -- !> ("move", node,xs) - --- | input data from a text that must be static and shared by all the nodes. --- 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 lnodes = length nodes - - parallelize (process lnodes) $ zip nodes [0..lnodes-1] - where - - process lnodes (node,i)= - runAt node $ local $ do - let xs = part str - size= case length xs `div` lnodes of 0 ->1 ; n -> n - xss= Transient.MapReduce.fromList $ - if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs - generateRef xss - --- | get the worlds of an URL -textUrl :: String -> DDS (DV.Vector Text.Text) -textUrl= getUrl (map Text.pack . words) - --- | generate a DDS from the content of a URL. --- The first parameter is a function that divide the text in words -getUrl :: (Loggable a, Distributable vector a) => (String -> [a]) -> String -> DDS (vector a) -getUrl partitioner url= DDS $ do - nodes <- local getEqualNodes -- !> "DISTRIBUTE" - let lnodes = length nodes - - parallelize (process lnodes) $ zip nodes [0..lnodes-1] -- !> show xss - where - process lnodes (node,i)= runAt node $ local $ do - r <- liftIO . simpleHTTP $ getRequest url - body <- liftIO $ getResponseBody r - let xs = partitioner body - size= case length xs `div` lnodes of 0 ->1 ; n -> n - xss= Transient.MapReduce.fromList $ - if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs - - generateRef xss - - --- | get the words of a file -textFile :: String -> DDS (DV.Vector Text.Text) -textFile= getFile (map Text.pack . words) - --- | generate a DDS from a file. All the nodes must access the file with the same path --- the first parameter is the parser that generates elements from the content -getFile :: (Loggable a, Distributable vector a) => (String -> [a]) -> String -> DDS (vector a) -getFile partitioner file= DDS $ do - nodes <- local getEqualNodes -- !> "DISTRIBUTE" - let lnodes = length nodes - - parallelize (process lnodes) $ zip nodes [0..lnodes-1] -- !> show xss - where - process lnodes (node, i)= runAt node $ local $ do - content <- do - c <- liftIO $ readFile file - length c `seq` return c - let xs = partitioner content - - size= case length xs `div` lnodes of 0 ->1 ; n -> n - xss= Transient.MapReduce.fromList $ - if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs - - generateRef xss - - - -generateRef :: Loggable a => a -> TransIO (PartRef a) -generateRef x= do - node <- getMyNode - liftIO $ do - temp <- getTempName - let reg= Part node temp False x - atomically $ newDBRef reg --- syncCache - (return $ getRef reg) -- !> ("generateRef",reg,node) - -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.(to be tested) -streamDDS - :: (Loggable a, Distributable vector a) => - Integer -> IO (StreamData a) -> DDS (vector a) -streamDDS time io= DDS $ do - xs <- local . groupByTime time $ do - r <- parallel io - case r of - SDone -> empty - SLast x -> return x - SMore x -> return x - SError e -> error $ show e - distribute' $ Transient.MapReduce.fromList xs - - - - +{-# LANGUAGE ExistentialQuantification, DeriveDataTypeable+, FlexibleInstances, MultiParamTypeClasses, OverloadedStrings, CPP #-}+++module Transient.MapReduce+(+Distributable(..),distribute, getText,+getUrl, getFile,textUrl, textFile,+mapKeyB, mapKeyU, reduce,eval,+-- * internals+DDS(..),Partition(..),PartRef(..))+ where++#ifdef ghcjs_HOST_OS+import Transient.Base+import Transient.Move hiding (pack)+import Transient.Logged hiding (hash)+-- dummy Transient.MapReduce module,+reduce _ _ = local stop :: Loggable a => Cloud a+mapKeyB _ _= undefined+mapKeyU _ _= undefined+distribute _ = undefined+getText _ _ = undefined+textFile _ = undefined+getUrl _ _ = undefined+textUrl _ = undefined+getFile _ _ = undefined+eval _= local stop+data Partition+data DDS= DDS+class Distributable+data PartRef a=PartRef a++#else++import Transient.Internals hiding (Ref)+import Transient.Mailboxes+import Transient.Parse+import Transient.Logged+import Transient.Move.Internals hiding (pack)+import Transient.Indeterminism+import Control.Applicative+import System.Random+import Control.Monad.State++import Control.Monad+import Data.Monoid++import Data.Typeable+import Data.Foldable+import Data.List hiding (delete, foldl')+import Control.Exception+import Control.Concurrent+--import Data.Time.Clock+import Network.HTTP+import Data.TCache hiding (onNothing)+import Data.TCache.Defs hiding (serialize,deserialize)++import Data.ByteString.Lazy.Char8 (pack,unpack, toStrict)+import Data.ByteString.Builder+import qualified Data.Map.Strict as M+import Control.Arrow (second)+import qualified Data.Vector.Unboxed as DVU+import qualified Data.Vector as DV+import Data.Hashable+import System.IO.Unsafe++import qualified Data.Foldable as F+import qualified Data.Text as Text+import Data.Text.Encoding++import Data.IORef++-- | a DDS contains a distrib. computation which return a (non-deterministic/stream/set of) +-- link/s to the generated chunk/s of data, in different nodes, thanks to the non-deterministic+-- and multithreaded nature of the Transient/Cloud comp.+data DDS a= Loggable a => DDS (Cloud (PartRef a))++-- | a link to a chunk of data, located in a node+data PartRef a= Ref Node Path Save deriving (Typeable, Read, Show)++-- | the chunk of data loaded in memory+data Partition a= Part Node Path Save a deriving (Typeable,Read,Show)+type Save= Bool++instance Loggable Text.Text where+ serialize t= byteString (encodeUtf8 t) + deserialize = tTakeWhile (/= '/') >>= return . decodeUtf8 . toStrict++instance Typeable a => Loggable (PartRef a) where+ serialize(Ref node path save)= serialize node <> "/" <> serialize path <> "/" <> serialize save -- <> "/" + deserialize= Ref <$> (deserialize <* slash)+ <*> (deserialize <* slash)+ <*> (deserialize)++ where+ slash= tChar '/'+ +instance Indexable (Partition a) where+ key (Part _ string b _)= keyp string b++++keyp s True= "PartP@"++s :: String+keyp s False="PartT@"++s++instance Loggable a => IResource (Partition a) where+ keyResource= key+ readResourceByKey k= r+ where+ typePart :: IO (Maybe a) -> a+ typePart = undefined+ r = if k !! 4 /= 'P' then return Nothing else+ defaultReadByKey (defPath (typePart r) ++ k) >>= return . fmap ( read . unpack)+ writeResource (s@(Part _ _ save _))=+ unless (not save) $ defaultWrite (defPath s ++ key s) (pack $ show s)+++eval :: DDS a -> Cloud (PartRef a)+eval (DDS mx) = mx+++type Path=String+++instance F.Foldable DVU.Vector where+ {-# INLINE foldr #-}+ foldr = foldr++ {-# INLINE foldl #-}+ foldl = foldl++ {-# INLINE foldr1 #-}+ foldr1 = foldr1++ {-# INLINE foldl1 #-}+ foldl1 = foldl1++--foldlIt' :: V.Unbox a => (b -> a -> b) -> b -> V.Vector a -> b+--foldlIt' f z0 xs= V.foldr f' id xs z0+-- where f' x k z = k $! f z x+--+--foldlIt1 :: V.Unbox a => (a -> a -> a) -> V.Vector a -> a+--foldlIt1 f xs = fromMaybe (error "foldl1: empty structure")+-- (V.foldl mf Nothing xs)+-- where+-- mf m y = Just (case m of+-- Nothing -> y+-- Just x -> f x y)++class (F.Foldable c, Monoid (c a), Loggable (c a),Typeable c, Typeable a) => Distributable c a where+ singleton :: a -> c a+ splitAt :: Int -> c a -> (c a, c a)+ fromList :: [a] -> c a++instance Loggable a => Loggable (DV.Vector a) where+ serialize v= intDec (DV.length v) <> "/" <> foldl' (\s x -> s <> "/" <> serialize x ) mempty v + deserialize= do+ len <- int + DV.replicateM len $ tChar '/' *> deserialize ++instance (Typeable a, Loggable a) => Distributable DV.Vector a where+ singleton = DV.singleton+ splitAt= DV.splitAt+ fromList = DV.fromList+++instance (Loggable a,DVU.Unbox a) => Loggable (DVU.Vector a) where+ serialize v= intDec (DVU.length v) <> "/" <> serialize(v DVU.! 0) <> DVU.ifoldl' (\s _ x -> s <> "/" <> serialize x) mempty (DVU.slice 1 (DVU.length v -1) v)+ deserialize= do+ len <- int + DVU.replicateM len $ tChar '/' *> deserialize++instance (Typeable a, Loggable a, DVU.Unbox a) => Distributable DVU.Vector a where+ singleton= DVU.singleton+ splitAt= DVU.splitAt+ fromList= DVU.fromList+++++-- | perform a map and partition the result with different keys using boxed vectors+-- The final result will be used by reduce.+mapKeyB :: (Typeable a, Loggable a, Typeable b,Loggable b, Typeable k, Loggable k,Ord k)+ => (a -> (k,b))+ -> DDS (DV.Vector a)+ -> DDS (M.Map k(DV.Vector b))+mapKeyB= mapKey++-- | perform a map and partition the result with different keys using unboxed vectors+-- The final result will be used by reduce.+mapKeyU :: (Typeable a, Loggable a, DVU.Unbox a, Typeable b, Loggable b, DVU.Unbox b, Typeable k, Loggable k,Ord k)+ => (a -> (k,b))+ -> DDS (DVU.Vector a)+ -> DDS (M.Map k(DVU.Vector b))+mapKeyU= mapKey++-- | perform a map and partition the result with different keys.+-- The final result will be used by reduce.+mapKey :: (Distributable container a,Distributable container b, Typeable k, Loggable k,Ord k)+ => (a -> (k,b))+ -> DDS (container a)+ -> DDS (M.Map k (container b))+mapKey f (DDS mx)= DDS $ loggedc $ do+ refs <- mx+ process refs !> ("process",refs)++ where+-- process :: Partition a -> Cloud [Partition b]+ process (ref@(Ref node path sav))= runAt node $ local $ do+ xs <- getPartitionData ref !> ("CMAP", ref,node)+ (generateRef $ map1 f xs)++++-- map1 :: (Ord k, F.Foldable container) => (a -> (k,b)) -> container a -> M.Map k(container b)+ map1 f v= F.foldl' f1 M.empty v+ where+ f1 map x=+ let (k,r) = f x+ in M.insertWith (<>) k (Transient.MapReduce.singleton r) map+{-+map :: (Distributable container a,Distributable container b, Loggable k,Ord k)+ => (a -> b)+ -> DDS (container a)+ -> DDS (container b)+map f (DDS mx)= DDS $ loggedc $ do+ refs <- mx+ process refs -- !> ("process",refs)++ where+-- process :: Partition a -> Cloud [Partition b]+ process (ref@(Ref node path sav))= runAt node $ local $ do+ xs <- getPartitionData ref -- !> ("CMAP", ref,node)+ (generateRef $ map1 f xs) // xxx+ !> "MAP"+ + map1 :: (Ord k, F.Foldable container) => (a -> b) -> container a -> container b+ map1 f v= F.foldl' f1 M.empty v+ where+ f1 map x=+ let r = f x+ in M.insertWith (<>) k (Transient.MapReduce.singleton r) map+-}++data ReduceChunk a= EndReduce | Reduce a deriving (Typeable, Read, Show)++boxids= unsafePerformIO $ newIORef (0 :: Int)+++reduce :: (Hashable k,Ord k, Distributable container a, Typeable k, Loggable k, Typeable a, Loggable a)+ => (a -> a -> a) -> DDS (M.Map k (container a)) ->Cloud (M.Map k a)++reduce red (dds@(DDS mx))= loggedc $ do++ mboxid <- localIO $ atomicModifyIORef boxids $ \n -> let n'= n+1 in (n',n')+ nodes <- local getEqualNodes+ -- return () !> ("REDUCE NODES=", nodes)++ let lengthNodes = length nodes+ shuffler nodes = do++ localIO $ threadDelay 100000+ ref@(Ref node path sav) <- mx -- return the resulting blocks of the map++ runAt node $ do+ localIO $ return () !> ("FOLDANDSEND","REF",ref, "runAt", node)+ foldAndSend node nodes ref++ stop++-- groupByDestiny :: (Hashable k, Distributable container a) => M.Map k (container a) -> M.Map Int [(k ,container a)]+ groupByDestiny map = M.foldlWithKey' f M.empty map+ where+-- f :: M.Map Int [(k ,container a)] -> k -> container a -> M.Map Int [(k ,container a)]+ f map k vs= M.insertWith (<>) (hash1 k) [(k,vs)] map+ hash1 k= abs $ hash k `rem` length nodes+++-- foldAndSend :: (Hashable k, Distributable container a)=> (Int,[(k,container a)]) -> Cloud ()+ foldAndSend node nodes ref= do++ pairs <- onAll $ getPartitionData1 ref+ <|> return (error $ "DDS computed out of his node:"++ show ref )+ let mpairs = groupByDestiny pairs++ length <- local . return $ M.size mpairs++ let port2= nodePort node+++ if length == 0 then sendEnd nodes else do++ nsent <- onAll $ liftIO $ newMVar 0++ (i,folded) <- local $ parallelize foldthem (M.assocs mpairs)++ n <- localIO $ modifyMVar nsent $ \r -> return (r+1, r+1) :: IO (Int,Int)++ -- sourcenode <- local getMyNode -- XXX borrar, solo para debug+ --localIO $ return () !> ("PUTMAILBOX TOSEND from",sourcenode,n,length,i,folded)+ (runAt (nodes !! i) $ do+ return () !> "JUST BEFORE PUTMAILBOX"+ local $ (putMailbox' mboxid (Reduce folded `asTypeOf` paramOf dds))+ !> ("PUTMAILBOX SENT ",n,length,i,folded))+++ + when (n == length) $ sendEnd nodes+ return ()+ empty++ where+++ foldthem (i,kvs)= async . return+ $ (i,map (\(k,vs) -> (k,foldl1 red vs)) kvs)+++ sendEnd nodes = do+ -- node <- local getMyNode -- XXX quitar. solo para debug++ onNodes nodes $ local $ + putMailbox' mboxid (EndReduce `asTypeOf` paramOf dds)+ -- !> ("PUTMAILBOX ENDREDUCE FROM", node))++ onNodes nodes f = foldr (<|>) empty $ map (\n -> runAt n f ) nodes++ sumNodes nodes f= foldr (<>) mempty $ map (\n -> runAt n f ) nodes++ reducer nodes= sumNodes nodes reduce1 -- a reduce1 process in each node, get the results and mappend them++-- reduce1 :: (Ord k) => Cloud (M.Map k v)++ reduce1 =local $ do+ + reduceResults <- liftIO $ newMVar M.empty+ numberSent <- liftIO $ newMVar 0+ return () !> "GETMAILBOX"+ minput <- getMailbox' mboxid -- get the chunk once it arrives to the mailbox++ case minput of++ EndReduce -> do+ return () !> "ENDREDUCE"+ n <- liftIO $ modifyMVar numberSent $ \r -> let r'= r+1 in return (r', r')++ if n == lengthNodes+ !> ("END REDUCE RECEIVEDDDD",n, lengthNodes)+ then do+ deleteMailbox' mboxid (EndReduce `asTypeOf` paramOf dds)+ r <- liftIO $ readMVar reduceResults+++ return r !> ("RETURNING",r)++ else stop++ Reduce kvs -> do+ return () !> "REDUCE RECEIVEDDDDDDDDDDDD"+ let addIt (k,inp) = do+ let input= inp `asTypeOf` atype dds+ liftIO $ modifyMVar_ reduceResults+ $ \map -> do+ let maccum = M.lookup k map+ return $ M.insert k (case maccum of+ Just accum -> red input accum+ Nothing -> input) map ++ mapM addIt (kvs `asTypeOf` paramOf' dds)+ !> ("RECEIVED REDUCEEEEEEEEEEEEE",kvs)+ stop++ + r <- reducer nodes <|> shuffler nodes+ localIO $ return () !> "RETRETRET"+ return r+ where+ atype ::DDS(M.Map k (container a)) -> a+ atype = undefined -- type level++ paramOf :: DDS (M.Map k (container a)) -> ReduceChunk [( k, a)]+ paramOf = undefined -- type level+ paramOf' :: DDS (M.Map k (container a)) -> [( k, a)]+ paramOf' = undefined -- type level+++++-- parallelize :: Loggable b => (a -> Cloud b) -> [a] -> Cloud b+parallelize f xs = foldr (<|>) empty $ map f xs++mparallelize f xs = loggedc $ foldr (<>) mempty $ map f xs+++getPartitionData :: (Typeable a, Loggable a) => PartRef a -> TransIO a+getPartitionData (Ref node path save) = Transient $ do+ mp <- (liftIO $ atomically+ $ readDBRef+ $ getDBRef+ $ keyp path save)+ `onNothing` error ("not found DDS data: "++ keyp path save)+ case mp of+ (Part _ _ _ xs) -> return $ Just xs++getPartitionData1 :: (Typeable a, Loggable a) => PartRef a -> TransIO a+getPartitionData1 (Ref node path save) = Transient $ do+ mp <- liftIO $ atomically+ $ readDBRef+ $ getDBRef+ $ keyp path save++ case mp of+ Just (Part _ _ _ xs) -> return $ Just xs+ Nothing -> return Nothing++getPartitionData2 :: (Typeable a,Loggable a) => PartRef a -> IO a+getPartitionData2 (Ref node path save) = do+ mp <- ( atomically+ $ readDBRef+ $ getDBRef+ $ keyp path save)+ `onNothing` error ("not found DDS data: "++ keyp path save)+ case mp of+ (Part _ _ _ xs) -> return xs++-- 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 -> Cloud a+runAtP node f uuid= do+ r <- runAt node $ onAll . 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 <- onAll getEqualNodes+ let node'= head $ nodes \\ [node]+ content <- onAll . liftIO $ readFile uuid+ runAt node' $ local $ liftIO $ writeFile uuid content++sendAnyError :: SomeException -> IO (StreamData a)+sendAnyError e= return $ SError e+++-- | distribute a container of values among many nodes.+-- If the container is static and sharable, better use the get* primitives+-- since each node will load the data independently.+distribute :: (Loggable a, Distributable container a ) => container a -> DDS (container a)+distribute = DDS . distribute'++distribute' xs= loggedc $ do+ nodes <- local getEqualNodes -- !> "DISTRIBUTE"+ let lnodes = length nodes+ let size= case F.length xs `div` (length nodes) of 0 ->1 ; n -> n+ xss= split size lnodes 1 xs -- !> size+ r <- distribute'' xss nodes+ return r+ where+ split n s s' xs | s==s' = [xs]+ split n s s' xs=+ let (h,t)= Transient.MapReduce.splitAt n xs+ in h : split n s (s'+1) t++distribute'' :: (Loggable a, Distributable container a)+ => [container a] -> [Node] -> Cloud (PartRef (container a))+distribute'' xss nodes =+ parallelize move $ zip nodes xss -- !> show xss+ where+ move (node, xs)= runAt node $ local $ do+ par <- generateRef xs+ return par+ !> ("move", node,xs)++-- | input data from a text that must be static and shared by all the nodes.+-- The function parameter partition the text in words+getText :: (Loggable a, Distributable container a) => (String -> [a]) -> String -> DDS (container a)+getText part str= DDS $ loggedc $ do+ nodes <- local getEqualNodes !> "getText"++ return () !> ("DISTRIBUTE TEXT IN NODES:",nodes)+ let lnodes = length nodes++ parallelize (process lnodes) $ zip nodes [0..lnodes-1]+ where++ process lnodes (node,i)= + runAt node $ local $ do+ let xs = part str+ size= case length xs `div` lnodes of 0 ->1 ; n -> n+ xss= Transient.MapReduce.fromList $+ if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs+ generateRef xss + !> "GETTEXT PROCESS"++-- | get the worlds of an URL+textUrl :: String -> DDS (DV.Vector Text.Text)+textUrl= getUrl (map Text.pack . words)++-- | generate a DDS from the content of a URL.+-- The first parameter is a function that divide the text in words+getUrl :: (Loggable a, Distributable container a) => (String -> [a]) -> String -> DDS (container a)+getUrl partitioner url= DDS $ do+ nodes <- local getEqualNodes -- !> "DISTRIBUTE"+ let lnodes = length nodes++ parallelize (process lnodes) $ zip nodes [0..lnodes-1] -- !> show xss+ where+ process lnodes (node,i)= runAt node $ local $ do+ r <- liftIO . simpleHTTP $ getRequest url+ body <- liftIO $ getResponseBody r+ let xs = partitioner body+ size= case length xs `div` lnodes of 0 ->1 ; n -> n+ xss= Transient.MapReduce.fromList $+ if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs+ + generateRef xss+ !> "GETURL"++-- | get the words of a file+textFile :: String -> DDS (DV.Vector Text.Text)+textFile= getFile (map Text.pack . words)++-- | generate a DDS from a file. All the nodes must access the file with the same path+-- the first parameter is the parser that generates elements from the content+getFile :: (Loggable a, Distributable container a) => (String -> [a]) -> String -> DDS (container a)+getFile partitioner file= DDS $ do+ nodes <- local getEqualNodes -- !> "DISTRIBUTE"+ let lnodes = length nodes++ parallelize (process lnodes) $ zip nodes [0..lnodes-1] -- !> show xss+ where+ process lnodes (node, i)= runAt node $ local $ do+ content <- do+ c <- liftIO $ readFile file+ length c `seq` return c+ let xs = partitioner content+ + size= case length xs `div` lnodes of 0 ->1 ; n -> n+ xss= Transient.MapReduce.fromList $+ if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs+ + generateRef xss+ !> "GETFILE"+++generateRef :: (Typeable a, Loggable a) => a -> TransIO (PartRef a)+generateRef x= do+ node <- getMyNode+ liftIO $ do+ temp <- getTempName+ let reg= Part node temp True x -- False to not save+ atomically $ newDBRef reg+-- syncCache+ (return $ getRef reg) -- !> ("generateRef",reg,node)++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.(to be tested)+streamDDS+ :: (Loggable a, Distributable container a) =>+ Int -> IO (StreamData a) -> DDS (container a)+streamDDS time io= DDS $ do+ xs <- local . groupByTime time $ do+ r <- parallel io+ case r of+ SDone -> empty+ SLast x -> return [x]+ SMore x -> return [x]+ SError e -> error $ show e+ distribute' $ Transient.MapReduce.fromList xs++++ #endif
src/Transient/Move.hs view
@@ -58,27 +58,27 @@ -- ** Joining the cluster Transient.Move.Internals.connect, connect', listen, -- Low level APIs -addNodes, shuffleNodes, +addNodes, addThisNodeToRemote, shuffleNodes, --Connection(..), ConnectionData(..), defConnection, -- ** Querying nodes -getMyNode, getWebServerNode, getNodes, nodeList, isBrowserInstance, +getMyNode, getWebServerNode, getNodes, getEqualNodes, nodeList, isBrowserInstance, -- * Running Local Computations -local, onAll, lazy, fixRemote, loggedc, lliftIO, localIO, fullStop, +local, onAll, lazy, localFix, fixRemote, loggedc, lliftIO, localIO, -- * Moving Computations wormhole, teleport, copyData, fixClosure, -- * Running at a Remote Node -beamTo, forkTo, callTo, runAt, atRemote, +beamTo, forkTo, callTo, runAt, atRemote, setSynchronous, syncStream, -- * Running at Multiple Nodes -clustered, mclustered, callNodes, +clustered, mclustered, callNodes, callNodes', foldNet, exploreNet, exploreNetUntil, -- * Messaging -putMailbox, putMailbox',getMailbox,getMailbox',cleanMailbox,cleanMailbox', +putMailbox, putMailbox',getMailbox,getMailbox',deleteMailbox,deleteMailbox', -- * Thread Control single, unique, @@ -90,11 +90,12 @@ #ifndef ghcjs_HOST_OS -- * REST API -api, HTTPMethod(..), PostParams, +api, HTTPMethod(..), HTTPHeaders(..), PostParams, noHTTP #endif ) where import Transient.Move.Internals +import Transient.Mailboxes -- $cluster --
src/Transient/Move/Internals.hs view
@@ -1,2109 +1,3340 @@------------------------------------------------------------------------------ --- --- Module : Transient.Move.Internals --- Copyright : --- License : MIT --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | --- ------------------------------------------------------------------------------ - - -{-# LANGUAGE DeriveDataTypeable , ExistentialQuantification, OverloadedStrings - ,ScopedTypeVariables, StandaloneDeriving, RecordWildCards, FlexibleContexts, CPP - ,GeneralizedNewtypeDeriving #-} -module Transient.Move.Internals where - -import Transient.Internals -import Transient.Parse -import Transient.Logged -import Transient.Indeterminism --- import Transient.Backtrack -import Transient.EVars - - -import Data.Typeable -import Control.Applicative -import System.IO.Error - -#ifndef ghcjs_HOST_OS -import Network ---- import Network.Info -import Network.URI ---import qualified Data.IP as IP -import qualified Network.Socket as NS -import qualified Network.BSD as BSD -import qualified Network.WebSockets as NWS -- S(RequestHead(..)) - -import qualified Network.WebSockets.Connection as WS - -import Network.WebSockets.Stream hiding(parse) -import qualified Data.ByteString as B(ByteString,concat) -import qualified Data.ByteString.Char8 as BC -import qualified Data.ByteString.Lazy.Internal as BLC -import qualified Data.ByteString.Lazy as BL -import qualified Data.ByteString.Lazy.Char8 as BS -import Network.Socket.ByteString as SBS(sendMany,sendAll,recv) -import qualified Network.Socket.ByteString.Lazy as SBSL -import Data.CaseInsensitive(mk) -import Data.Char(isSpace) - --- import System.Random - -#else -import JavaScript.Web.WebSocket -import qualified JavaScript.Web.MessageEvent as JM -import GHCJS.Prim (JSVal) -import GHCJS.Marshal(fromJSValUnchecked) -import qualified Data.JSString as JS - - -import JavaScript.Web.MessageEvent.Internal -import GHCJS.Foreign.Callback.Internal (Callback(..)) -import qualified GHCJS.Foreign.Callback as CB -import Data.JSString (JSString(..), pack) - -#endif - - -import Control.Monad.State --- import System.IO -import Control.Exception hiding (onException,try) -import Data.Maybe ---import Data.Hashable - ---import System.Directory --- import Control.Monad - -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,(\\)) -- ,find, insert) -import Data.IORef - - - --- import System.IO - -import Control.Concurrent - - - --- import Data.Dynamic -import Data.String - -import System.Mem.StableName -import Unsafe.Coerce - - - ---import System.Random - -#ifdef ghcjs_HOST_OS -type HostName = String -newtype PortID = PortNumber Int deriving (Read, Show, Eq, Typeable) -#endif - -data Node= Node{ nodeHost :: HostName - , nodePort :: Int - , connection :: Maybe (MVar Pool) - , nodeServices :: Service - } - - deriving (Typeable) - -instance Ord Node where - compare node1 node2= compare (nodeHost node1,nodePort node1)(nodeHost node2,nodePort node2) - - --- The cloud monad is a thin layer over Transient in order to make sure that the type system --- forces the logging of intermediate results -newtype Cloud a= Cloud {runCloud' ::TransIO a} deriving (Functor,Monoid,Applicative,Alternative, Monad, Num, Fractional, MonadState EventF) - - - --- | Execute a distributed computation inside a TransIO computation. --- All the computations in the TransIO monad that enclose the cloud computation must be `logged` -runCloud :: Cloud a -> TransIO a - -runCloud x= do - closRemote <- getSData <|> return (Closure 0) - runCloud' x <*** setData closRemote - - ---instance Monoid a => Monoid (Cloud a) where --- mappend x y = mappend <$> x <*> y --- mempty= return mempty - -#ifndef ghcjs_HOST_OS - ---- empty Hooks for TLS - -{-# NOINLINE tlsHooks #-} -tlsHooks ::IORef (SData -> BS.ByteString -> IO () - ,SData -> IO B.ByteString - ,NS.Socket -> BS.ByteString -> TransIO () - ,String -> NS.Socket -> BS.ByteString -> TransIO ()) -tlsHooks= unsafePerformIO $ newIORef - ( notneeded - , notneeded - , \_ i -> tlsNotSupported i - , \_ _ _-> return()) - - where - notneeded= error "TLS hook function called" - - - - tlsNotSupported input = do - if ((not $ BL.null input) && BL.head input == 0x16) - then do - conn <- getSData - sendRaw conn $ BS.pack $ "HTTP/1.0 525 SSL Handshake Failed\r\nContent-Length: 0\nConnection: close\r\n\r\n" - else return () - -(sendTLSData,recvTLSData,maybeTLSServerHandshake,maybeClientTLSHandshake)= unsafePerformIO $ readIORef tlsHooks - - -#endif - --- | Means that this computation will be executed in the current node. the result will be logged --- so the closure will be recovered if the computation is translated to other node by means of --- primitives like `beamTo`, `forkTo`, `runAt`, `teleport`, `clustered`, `mclustered` etc -local :: Loggable a => TransIO a -> Cloud a -local = Cloud . logged - ---stream :: Loggable a => TransIO a -> Cloud (StreamVar a) ---stream= Cloud . transport - --- #ifndef ghcjs_HOST_OS --- | Run a distributed computation inside the IO monad. Enables asynchronous --- console input (see 'keep'). -runCloudIO :: Typeable a => Cloud a -> IO (Maybe a) -runCloudIO (Cloud mx)= keep mx - --- | Run a distributed computation inside the IO monad with no console input. -runCloudIO' :: Typeable a => Cloud a -> IO (Maybe a) -runCloudIO' (Cloud mx)= keep' mx - --- #endif - --- | alternative to `local` It means that if the computation is translated to other node --- this will be executed again if this has not been executed inside a `local` computation. --- --- > onAll foo --- > local foo' --- > local $ do --- > bar --- > runCloud $ do --- > onAll baz --- > runAt node .... --- > callTo node' ..... --- --- Here foo will be executed in node' but foo' bar and baz don't. --- --- However foo bar and baz will e executed in node. --- - -onAll :: TransIO a -> Cloud a -onAll = Cloud - --- | only executes if the result is demanded. It is useful when the conputation result is only used in --- the remote node, but it is not serializable. -lazy :: TransIO a -> Cloud a -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 ) - logged mx <*** setData closRemote - - -loggedc' :: Loggable a => Cloud a -> Cloud a -loggedc' (Cloud mx)= Cloud $ logged mx - - - - --- | the `Cloud` monad has no `MonadIO` instance. `lliftIO= local . liftIO` -lliftIO :: Loggable a => IO a -> Cloud a -lliftIO= local . liftIO - --- | `localIO = lliftIO` -localIO :: Loggable a => IO a -> Cloud a -localIO= lliftIO - --- | stop the current computation and does not execute any alternative computation -fullStop :: TransIO stop -fullStop= setData WasRemote >> stop - - --- | continue the execution in a new node -beamTo :: Node -> Cloud () -beamTo node = wormhole node teleport - - --- | execute in the remote node a process with the same execution state -forkTo :: Node -> Cloud () -forkTo node= beamTo node <|> return() - --- | open a wormhole to another node and executes an action on it. --- currently by default it keep open the connection to receive additional requests --- and responses (streaming) -callTo :: Loggable a => Node -> Cloud a -> Cloud a -callTo node remoteProc= wormhole node $ atRemote remoteProc - - -#ifndef ghcjs_HOST_OS --- | A connectionless version of callTo for long running remote calls -callTo' :: (Show a, Read a,Typeable a) => Node -> Cloud a -> Cloud a -callTo' node remoteProc= do - mynode <- local $ getNodes >>= return . head - beamTo node - r <- remoteProc - beamTo mynode - return r -#endif - --- | Within a connection to a node opened by `wormhole`, it run the computation in the remote node and return --- the result back to the original node. --- --- If `atRemote` is executed in the remote node, then the computation is executed in the original node --- --- > wormhole node2 $ do --- > t <- atRemote $ do --- > r <- foo -- executed in node2 --- > s <- atRemote bar r -- executed in the original node --- > baz s -- in node2 --- > bat t -- in 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" - lazy $ setData was - return r - --- | Execute a computation in the node that initiated the connection. --- --- if the sequence of connections is n1 -> n2 -> n3 then `atCallingNode $ atCallingNode foo` in n3 --- would execute `foo` in n1, -- while `atRemote $ atRemote foo` would execute it in n3 --- atCallingNode :: Loggable a => Cloud a -> Cloud a --- atCallingNode proc= connectCaller $ atRemote proc - --- | synonymous of `callTo` -runAt :: Loggable a => Node -> Cloud a -> Cloud a -runAt= callTo - - - --- | run a single thread with that action for each connection created. --- When the same action is re-executed within that connection, all the threads generated by the previous execution --- are killed --- --- > box <- foo --- > r <- runAt node . local . single $ getMailbox box --- > localIO $ print r --- --- if foo return differnt mainbox indentifiers, the above code would print the --- messages of the last one. --- Without single, it would print the messages of all of them. -single :: TransIO a -> TransIO a -single f= do - cutExceptions - Connection{closChildren=rmap} <- getSData <|> error "single: only works within a wormhole" - mapth <- liftIO $ readIORef rmap - id <- liftIO $ f `seq` makeStableName f >>= return . hashStableName - - - case M.lookup id mapth of - Just tv -> liftIO $ killBranch' tv -- !> "JUSTTTTTTTTT" - Nothing -> return () -- !> "NOTHING" - - - tv <- get - f <** do - id <- liftIO $ makeStableName f >>= return . hashStableName - liftIO $ modifyIORef rmap $ \mapth -> M.insert id tv mapth - - --- | run an unique continuation for each connection. The first thread that execute `unique` is --- executed for that connection. The rest are ignored. -unique :: a -> TransIO () -unique f= do - Connection{closChildren=rmap} <- getSData <|> error "unique: only works within a connection. Use wormhole" - mapth <- liftIO $ readIORef rmap - id <- liftIO $ f `seq` makeStableName f >>= return . hashStableName - - let mx = M.lookup id mapth - case mx of - Just _ -> empty - Nothing -> do - tv <- get - liftIO $ modifyIORef rmap $ \mapth -> M.insert id tv mapth - - ---data ParentConnection= ParentConnection Connection (Maybe Closure) deriving Typeable - --- | A wormhole opens a connection with another node anywhere in a computation. --- `teleport` uses this connection to translate the computation back and forth between the two nodes connected -wormhole :: Loggable a => Node -> Cloud a -> Cloud a -wormhole node (Cloud comp) = local $ Transient $ do - - moldconn <- getData :: StateIO (Maybe Connection) - mclosure <- getData :: StateIO (Maybe Closure) - -- when (isJust moldconn) . setState $ ParentConnection (fromJust moldconn) mclosure - - -- labelState $ "wormhole" ++ show node - Log rec _ _ _<- getData `onNothing` return (Log False [][] 0) - - if not rec - then runTrans $ (do - conn <- mconnect node - liftIO $ writeIORef (remoteNode conn) $ Just node - setData conn{calling= True} - - setData $ (Closure 0 ) - - comp ) - <*** do - when (isJust moldconn) . setData $ fromJust moldconn - when (isJust mclosure) . setData $ fromJust mclosure - -- <** is not enough since comp may be reactive - else do - let conn = fromMaybe (error "wormhole: no connection in remote node") moldconn - setData $ conn{calling= False} - runTrans $ comp - <*** do when (isJust mclosure) . setData $ fromJust mclosure - --- | connect to the caller node. --- connectCaller :: Loggable a => Cloud a -> Cloud a --- connectCaller (Cloud comp)= local $ do --- conn <- getState !> "CONNECTCALLER" --- case connData conn of --- Nothing -> empty --- Just Self -> empty --- _ -> if not $ calling conn !> ("calling", calling conn) then comp else do --- ParentConnection conn mmclosure <- getState <|> error "connectCaller: No connection defined: use wormhole" --- moldconn <- getData :: TransIO (Maybe Connection) --- mclosure <- getData :: TransIO (Maybe Closure) - --- -- labelState $ "wormhole" ++ show node --- Log rec _ _ <- getData `onNothing` return (Log False [][]) - - --- if not rec --- then do --- -- liftIO $ writeIORef (remoteNode conn) $ Just node --- setData conn{calling= True} --- setData $ if (isJust mmclosure) --- then fromJust mmclosure --- else Closure 0 - --- comp --- <*** do when (isJust moldconn) . setData $ fromJust moldconn --- when (isJust mclosure) . setData $ fromJust mclosure --- -- <** is not enough since comp may be reactive --- else do --- let conn = fromMaybe (error "wormhole: no connection in remote node") moldconn --- setData $ conn{calling= False} --- comp --- <*** do when (isJust mclosure) . setData $ fromJust mclosure - -#ifndef ghcjs_HOST_OS -type JSString= String -pack= id - - - -#endif - -data CloudException = CloudException Node IdClosure String deriving (Typeable, Show, Read) - -instance Exception CloudException - -teleport :: Cloud () -teleport = local $ do - Transient $ do - cont <- get - Log rec log fulLog closLocal <- getData `onNothing` return (Log False [][] 0) - - conn@Connection{connData=contype, localClosures= localClosures,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 - - --- when a node call itself, there is no need of socket communications --- #ifndef ghcjs_HOST_OS - case contype of - Just Self -> runTrans $ do - setData $ if (not calling) then WasRemote else WasParallel - abduce !> "SELF" -- call himself - liftIO $ do - remote <- readIORef $ remoteNode conn - writeIORef (myNode conn) $ fromMaybe (error "teleport: no connection?") remote - - - _ -> do - --- #endif - - --read this Closure - Closure closRemote <- getData `onNothing` return (Closure 0 ) - - - return () !> ("TELEPORTTTTTTTTTT", closLocal) - --set his own closure in his Node data - - -- closLocal <- liftIO $ randomRIO (0,1000000) --- node <- runTrans getMyNode - - - let tosend= reverse $ if closRemote==0 then fulLog else log - - liftIO $ modifyMVar_ localClosures $ \map -> return $ M.insert closLocal cont map - -- The log sent is in the order of execution. log is in reverse order - - -- send log with closure ids at head - runTrans $ msend conn $ SMore $ ClosureData closRemote closLocal tosend - !> ("teleport sending", SMore (unsafePerformIO $ readIORef $ remoteNode conn,closRemote,closLocal,tosend)) - !> "--------->------>---------->" - -- -- !> ("log",reverse fulLog) - - - 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 - return $ Just () - - -- code moved to reportBack - -- runTrans $ onException $ \(e :: SomeException) -> do - - -- Closure closRemote <- getData `onNothing` error "teleport: no closRemote" - -- node <- getMyNode - -- let msg= SError $ toException $ ErrorCall $ show $ show $ CloudException node closRemote $ show e - -- msend conn msg !> "MSEND" - - - - return () -- !> "TELEPORT remote" - --- | forward exceptions to the calling node -reportBack :: TransIO () -reportBack= onException $ \(e :: SomeException) -> do - conn<- getData `onNothing` error "reportBack: No connection defined: use wormhole" - Closure closRemote <- getData `onNothing` error "teleport: no closRemote" - node <- getMyNode - let msg= SError $ toException $ ErrorCall $ show $ show $ CloudException node closRemote $ show e - msend conn msg !> "MSEND" - - - --- | copy a session data variable from the local to the remote node. --- If there is none set in the local node, The parameter is the default value. --- In this case, the default value is also set in the local node. -copyData def = do - r <- local getSData <|> return def - onAll $ setData r - return r - - --- | write to the mailbox --- Mailboxes are node-wide, for all processes that share the same connection data, that is, are under the --- same `listen` or `connect` --- while EVars are only visible by the process that initialized it and his children. --- Internally, the mailbox is in a well known EVar stored by `listen` in the `Connection` state. -putMailbox :: Typeable val => val -> TransIO () -putMailbox = putMailbox' (0::Int) - --- | write to a mailbox identified by an identifier besides the type -putMailbox' :: (Typeable key, Ord key, Typeable val) => key -> val -> TransIO () -putMailbox' idbox dat= do - let name= MailboxId idbox $ typeOf dat - Connection{comEvent= mv} <- getData `onNothing` errorMailBox - mbs <- liftIO $ readIORef mv - let mev = M.lookup name mbs - case mev of - Nothing ->newMailbox name >> putMailbox' idbox dat - Just ev -> writeEVar ev $ unsafeCoerce dat - - -newMailbox :: MailboxId -> TransIO () -newMailbox name= do --- return () -- !> "newMailBox" - Connection{comEvent= mv} <- getData `onNothing` errorMailBox - ev <- newEVar - liftIO $ atomicModifyIORef mv $ \mailboxes -> (M.insert name ev mailboxes,()) - - -errorMailBox= error "MailBox: No connection open. Use wormhole" - --- | get messages from the mailbox that matches with the type expected. --- The order of reading is defined by `readTChan` --- This is reactive. it means that each new message trigger the execution of the continuation --- each message wake up all the `getMailbox` computations waiting for it. -getMailbox :: Typeable val => TransIO val -getMailbox = getMailbox' (0 :: Int) - --- | read from a mailbox identified by an identifier besides the type -getMailbox' :: (Typeable key, Ord key, Typeable val) => key -> TransIO val -getMailbox' mboxid = x where - x = do - - let name= MailboxId mboxid $ typeOf $ typeOf1 x - Connection{comEvent= mv} <- getData `onNothing` errorMailBox - mbs <- liftIO $ readIORef mv - let mev = M.lookup name mbs - case mev of - Nothing ->newMailbox name >> getMailbox' mboxid - Just ev ->unsafeCoerce $ readEVar ev - - typeOf1 :: TransIO a -> a - typeOf1 = undefined - --- | delete all subscriptions for that mailbox expecting this kind of data -cleanMailbox :: Typeable a => a -> TransIO () -cleanMailbox = cleanMailbox' 0 - --- | clean a mailbox identified by an Int and the type -cleanMailbox' :: Typeable a => Int -> a -> TransIO () -cleanMailbox' mboxid witness= do - let name= MailboxId mboxid $ typeOf witness - Connection{comEvent= mv} <- getData `onNothing` error "getMailBox: accessing network events out of listen" - mbs <- liftIO $ readIORef mv - let mev = M.lookup name mbs - case mev of - Nothing -> return() - Just ev -> do cleanEVar ev - liftIO $ atomicModifyIORef mv $ \mbs -> (M.delete name mbs,()) - --- | execute a Transient action in each of the nodes connected. --- --- The response of each node is received by the invoking node and processed by the rest of the procedure. --- By default, each 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 => Cloud a -> Cloud a -clustered proc= callNodes (<|>) empty proc - - --- A variant of `clustered` that wait for all the responses and `mappend` them -mclustered :: (Monoid a, Loggable a) => Cloud a -> Cloud a -mclustered proc= callNodes (<>) mempty proc - - -callNodes op init proc= loggedc' $ do - nodes <- local getEqualNodes - callNodes' nodes op init proc - - -callNodes' nodes op init proc= loggedc' $ foldr op init $ map (\node -> runAt node proc) nodes ------ -#ifndef ghcjs_HOST_OS -sendRaw (Connection _ _ _ (Just (Node2Web sconn )) _ _ _ _ _ _) r= - liftIO $ WS.sendTextData sconn r -- !> ("NOde2Web",r) - -sendRaw (Connection _ _ _ (Just (Node2Node _ sock _)) _ _ blocked _ _ _) r= - liftIO $ withMVar blocked $ const $ SBS.sendMany sock - (BL.toChunks r ) -- !> ("NOde2Node",r) - -sendRaw (Connection _ _ _(Just (TLSNode2Node ctx )) _ _ blocked _ _ _) r= - liftIO $ withMVar blocked $ const $ sendTLSData ctx r -- !> ("TLNode2Web",r) - -#else -sendRaw (Connection _ _ _ (Just (Web2Node sconn)) _ _ blocked _ _ _) r= liftIO $ - withMVar blocked $ const $ JavaScript.Web.WebSocket.send r sconn -- !!> "MSEND SOCKET" -#endif - -sendRaw _ _= error "No connection stablished" - -type LengthFulLog= Int -data NodeMSG= ClosureData IdClosure IdClosure CurrentPointer - | RelayMSG Node Node (StreamData NodeMSG) - deriving (Typeable, Read, Show) - -msend :: Connection -> StreamData NodeMSG -> TransIO () - -msend (Connection _ _ _ (Just Self) _ _ _ _ _ _) r= return () - -#ifndef ghcjs_HOST_OS - - -msend (Connection _ _ _ (Just (Node2Node _ sock _)) _ _ blocked _ _ _) r=do - liftIO $ withMVar blocked $ const $ SBS.sendAll sock $ BC.pack (show r) !> ("N2N SEND", r) - -msend (Connection _ _ _ (Just (TLSNode2Node ctx)) _ _ _ _ _ _) r= - liftIO $ sendTLSData ctx $ BS.pack (show r) !> "TLS SEND" - - -msend (Connection _ _ _ (Just (Node2Web sconn)) _ _ _ _ _ _) r=liftIO $ - {-withMVar blocked $ const $ -} WS.sendTextData sconn $ BS.pack (show r) !> "websockets send" - -msend(Connection _ myNode _ (Just (Relay conn remote )) _ _ _ _ _ _) r= do - origin <- liftIO $ readIORef myNode - msend conn $ SMore $ RelayMSG origin remote r - - -#else - -msend (Connection _ _ remoten (Just (Web2Node sconn)) _ _ blocked _ _ _) r= liftIO $ do - -- when (js_readystate sconn /= 1) $ do -- must try to reconnect - -- Just node <- liftIO $ readIORef remoten - -- ... - withMVar blocked $ const $ JavaScript.Web.WebSocket.send (JS.pack $ show r) sconn !> "MSEND SOCKET" - - - -#endif - -msend (Connection _ _ _ Nothing _ _ _ _ _ _) _= error "msend out of connection context: use wormhole to connect" - - - -mread :: Loggable a => Connection -> TransIO (StreamData a) - - -#ifdef ghcjs_HOST_OS - - -mread (Connection _ _ _ (Just (Web2Node sconn)) _ _ _ _ _ _)= wsRead sconn - - - -wsRead :: Loggable a => WebSocket -> TransIO a -wsRead ws= do - dat <- react (hsonmessage ws) (return ()) - case JM.getData dat of - JM.StringData str -> return (read' $ JS.unpack str) - !> ("Browser webSocket read", str) !> "<------<----<----<------" - JM.BlobData blob -> error " blob" - JM.ArrayBufferData arrBuffer -> error "arrBuffer" - - - -wsOpen :: JS.JSString -> TransIO WebSocket -wsOpen url= do - ws <- liftIO $ js_createDefault url -- !> ("wsopen",url) - react (hsopen ws) (return ()) -- !!> "react" - return ws -- !!> "AFTER ReACT" - -foreign import javascript safe - "window.location.hostname" - js_hostname :: JSVal - -foreign import javascript safe - "window.location.pathname" - js_pathname :: JSVal - -foreign import javascript safe - "window.location.protocol" - js_protocol :: JSVal - -foreign import javascript safe - "(function(){var res=window.location.href.split(':')[2];if (res === undefined){return 80} else return res.split('/')[0];})()" - js_port :: JSVal - -foreign import javascript safe - "$1.onmessage =$2;" - js_onmessage :: WebSocket -> JSVal -> IO () - - -getWebServerNode :: TransIO Node -getWebServerNode = liftIO $ do - h <- fromJSValUnchecked js_hostname - p <- fromIntegral <$> (fromJSValUnchecked js_port :: IO Int) - createNode h p - - -hsonmessage ::WebSocket -> (MessageEvent ->IO()) -> IO () -hsonmessage ws hscb= do - cb <- makeCallback MessageEvent hscb - js_onmessage ws cb - -foreign import javascript safe - "$1.onopen =$2;" - js_open :: WebSocket -> JSVal -> IO () - -foreign import javascript safe - "$1.readyState" - js_readystate :: WebSocket -> Int - -newtype OpenEvent = OpenEvent JSVal deriving Typeable -hsopen :: WebSocket -> (OpenEvent ->IO()) -> IO () -hsopen ws hscb= do - cb <- makeCallback OpenEvent hscb - js_open ws cb - -makeCallback :: (JSVal -> a) -> (a -> IO ()) -> IO JSVal - -makeCallback f g = do - Callback cb <- CB.syncCallback1 CB.ContinueAsync (g . f) - return cb - - -foreign import javascript safe - "new WebSocket($1)" js_createDefault :: JS.JSString -> IO WebSocket - - -#else -mread (Connection _ _ _ (Just (Node2Node _ _ _)) _ _ _ _ _ _) = parallelReadHandler -- !> "mread" - -mread (Connection _ _ _ (Just (TLSNode2Node _ )) _ _ _ _ _ _) = parallelReadHandler --- parallel $ do --- s <- recvTLSData ctx --- return . read' $ BC.unpack s - -mread (Connection _ _ _ (Just (Node2Web sconn )) _ _ _ _ _ _)= - parallel $ do - s <- WS.receiveData sconn - return . read' $ BS.unpack s - !> ("WS MREAD RECEIVED ----<----<------<--------", s) - -mread (Connection _ _ _ (Just (Relay conn _ )) _ _ _ _ _ _)= - mread conn -- !> "MREAD RELAY" - - - - -parallelReadHandler :: Loggable a => TransIO (StreamData a) -parallelReadHandler= do - str <- giveData :: TransIO BS.ByteString - r <- choose $ readStream str - - return r - !> ("parallel read handler read", r) - !> "<-------<----------<--------<----------" - where - readStream :: (Typeable a, Read a) => BS.ByteString -> [StreamData a] - readStream s= readStream1 $ BS.unpack s - where - - readStream1 s= - let [(x,r)] = reads s - in x : readStream1 r - - - -getWebServerNode :: TransIO Node -getWebServerNode = getNodes >>= return . head -#endif - - - ---release (Node h p rpool _) hand= liftIO $ do ----- print "RELEASED" --- atomicModifyIORef rpool $ \ hs -> (hand:hs,()) --- -- !!> "RELEASED" - -mclose :: Connection -> IO () - -#ifndef ghcjs_HOST_OS - -mclose (Connection _ _ _ - (Just (Node2Node _ sock _ )) _ _ _ _ _ _)= NS.close sock - -mclose (Connection _ _ _ - (Just (Node2Web sconn )) - _ _ _ _ _ _)= - WS.sendClose sconn ("closemsg" :: BS.ByteString) - -#else - -mclose (Connection _ _ _ (Just (Web2Node sconn)) _ _ blocked _ _ _)= - JavaScript.Web.WebSocket.close Nothing Nothing sconn - -#endif - - - -mconnect :: Node -> TransIO Connection -mconnect node'= do - node <- fixNode node' - nodes <- getNodes - return () !> ("mconnnect", nodePort node) - let fnode = filter (==node) nodes - case fnode of - [] -> mconnect1 node -- !> "NO NODE" - [node'@(Node _ _ pool _)] -> do - plist <- liftIO $ readMVar $ fromJust pool - case plist of -- !> ("length", length plist,nodePort node) of - (handle:_) -> do - delData $ Closure undefined - return handle - !> ("REUSED!", node) - - _ -> mconnect1 node' - where - - -#ifndef ghcjs_HOST_OS - mconnect1 (node@(Node host port _ _))= do - - return () !> ("MCONNECT1",host,port,nodeServices node) - (conn,parseContext) <- checkSelf node <|> - timeout 1000000 (connectNode2Node host port) <|> - timeout 1000000 (connectWebSockets host port) <|> - checkRelay <|> - (throwt $ ConnectionError "" node) - - setState conn - setState parseContext --- return () !> "CONNECTED AFTER TIMEOUT" - - -- write node connected in the connection - liftIO $ writeIORef (remoteNode conn) $ Just node - -- write connection in the node - liftIO $ modifyMVar_ (fromJust $ connection node) . const $ return [conn] - addNodes [node] - - case connData conn of - Just Self -> return() - _ -> watchConnection - delData $ Closure undefined - return conn - - where - checkSelf node= do - node' <- getMyNode - if node /= node' - then empty - else do - conn<- case connection node of - Nothing -> error "checkSelf error" - Just ref -> do - cnn <- getSData <|> error "chechself: no connection" - rnode <- liftIO $ newIORef node - conn <- defConnection >>= \c -> return c{myNode= rnode, comEvent=comEvent cnn,connData= Just Self} !> "DEFF1" - liftIO $ withMVar ref $ const $ return [conn] - return conn - - return (conn,(ParseContext (error "checkSelf parse error") (error "checkSelf parse error") - :: ParseContext BS.ByteString)) - - timeout t proc=do - r <- collect' 1 t proc - case r of - [] -> empty - r:_ -> return r - - checkRelay= do - return () !> "RELAY" - myNode <- getMyNode - if nodeHost node== nodeHost myNode - then - case lookup "localNode" $ nodeServices node of - Just snode -> do - con <- mconnect $ read snode - cont <- getSData <|> return noParseContext - return (con,cont) - Nothing -> empty - else do - - case lookup "relay" $ nodeServices node of - Nothing -> empty -- !> "NO RELAY" - Just relayInfo -> do - let relay= read relayInfo - conn <- mconnect relay -- !> ("RELAY",relay, node) - rem <- liftIO $ newIORef $ Just node - -- clos <- liftIO $ newMVar $ M.empty - let conn'= conn{connData= Just $ Relay conn node,remoteNode=rem} --,closures= clos} - - parseContext <- getState <|> return noParseContext - return (conn', parseContext) - - noParseContext= (ParseContext (error "relay error") (error "relay error") - :: ParseContext BS.ByteString) - - connectSockTLS host port= do - return () !> "connectSockTLS" - - let size=8192 - Connection{myNode=my,comEvent= ev} <- getSData <|> error "connect: listen not set for this node" - - sock <- liftIO $ connectTo' size host $ PortNumber $ fromIntegral port - - conn' <- defConnection >>= \c -> - return c{myNode=my, comEvent= ev,connData= - - Just $ (Node2Node u sock (error $ "addr: outgoing connection"))} - - setData conn' - input <- liftIO $ SBSL.getContents sock - - setData $ ParseContext (error "parse context: Parse error") input - - - maybeClientTLSHandshake host sock input - - - `catcht` \(_ :: SomeException) -> empty - - - connectNode2Node host port= do - return () !> "NODE 2 NODE" - connectSockTLS host port - - conn <- getSData <|> error "mconnect: no connection data" - sendRaw conn "CLOS a b\r\n\r\n" - r <- liftIO $ readFrom conn - - case r of - "OK" -> do - parseContext <- getState - return (conn,parseContext) - - _ -> do - let Connection{connData=cdata}= conn - case cdata of - Just(Node2Node _ s _) -> liftIO $ NS.close s -- since the HTTP firewall closes the connection --- Just(TLSNode2Node c) -> contextClose c -- TODO - empty - - - connectWebSockets host port = do - return () !> "WEBSOCKETS" - connectSockTLS host port -- a new connection - - never <- liftIO $ newEmptyMVar :: TransIO (MVar ()) - conn <- getSData <|> error "connectWebSockets: no connection" - stream <- liftIO $ makeWSStreamFromConn conn - wscon <- react (NWS.runClientWithStream stream (host++(':': show port)) "/" - WS.defaultConnectionOptions []) (takeMVar never) - - - return (conn{connData= Just $ (Node2Web wscon)}, noParseContext) - --- noConnection= error $ show node ++ ": no connection" - - - watchConnection= do - conn <- getSData - parseContext <- getSData <|> error "NO PARSE CONTEXT" - :: TransIO (ParseContext BS.ByteString) - chs <- liftIO $ newIORef M.empty - let conn'= conn{closChildren= chs} - -- liftIO $ modifyMVar_ (fromJust pool) $ \plist -> do - -- if not (null plist) then print "DUPLICATE" else return () - -- return $ conn':plist -- !> (node,"ADDED TO POOL") - - -- tell listenResponses to watch incoming responses - putMailbox ((conn',parseContext,node) - :: (Connection,ParseContext BS.ByteString,Node)) - liftIO $ threadDelay 100000 -- give time to initialize listenResponses - -#else - mconnect1 (node@(Node host port (Just pool) _))= do - conn' <- getSData <|> error "connect: listen not set for this node" - if nodeHost node== "webnode" then return conn'{connData= Just Self} else do - ws <- connectToWS host $ PortNumber $ fromIntegral port --- !> "CONNECTWS" - let conn= conn'{connData= Just (Web2Node ws)} --- !> ("websocker CONNECION") - let parseContext = - ParseContext (error "parsecontext not available in the browser") - ("" :: JSString) - - chs <- liftIO $ newIORef M.empty - let conn'= conn{closChildren= chs} - liftIO $ modifyMVar_ pool $ \plist -> return $ conn':plist - putMailbox (conn',parseContext,node) -- tell listenResponses to watch incoming responses - delData $ Closure undefined - return conn -#endif - - u= undefined - -data ConnectionError= ConnectionError String Node deriving Show - -instance Exception ConnectionError - --- mconnect _ = empty - - - -#ifndef ghcjs_HOST_OS -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 --- NS.setSocketOption sock NS.SendTimeOut 1000000 !> ("CONNECT",port) - - he <- BSD.getHostByName hostname - - NS.connect sock (NS.SockAddrInet port (BSD.hostAddress he)) - - return sock) - -#else -connectToWS h (PortNumber p) = do - protocol <- liftIO $ fromJSValUnchecked js_protocol - pathname <- liftIO $ fromJSValUnchecked js_pathname - return () !> ("PAHT",pathname) - let ps = case (protocol :: JSString)of "http:" -> "ws://"; "https:" -> "wss://" - wsOpen $ JS.pack $ ps++ h++ ":"++ show p ++ pathname -#endif - - - -type Blocked= MVar () -type BuffSize = Int -data ConnectionData= -#ifndef ghcjs_HOST_OS - Node2Node{port :: PortID - ,socket ::Socket - ,sockAddr :: NS.SockAddr - } - | TLSNode2Node{tlscontext :: SData} - | Node2Web{webSocket :: WS.Connection} --- | WS2Node{webSocketNode :: WS.Connection} - | Self - | Relay Connection Node -- (EVar (StreamData NodeMSG)) -#else - Self - | Web2Node{webSocket :: WebSocket} -#endif - -- deriving (Eq,Ord) - - -data MailboxId = forall a .(Typeable a, Ord a) => MailboxId a TypeRep - -instance Eq MailboxId where - id1 == id2 = id1 `compare` id2== EQ - -instance Ord MailboxId where - MailboxId n t `compare` MailboxId n' t'= - case typeOf n `compare` typeOf n' of - EQ -> case n `compare` unsafeCoerce n' of - EQ -> t `compare` t' - LT -> LT - GT -> GT - - other -> other - -data Connection= Connection{idConn :: Int - ,myNode :: IORef Node - ,remoteNode :: IORef (Maybe Node) - ,connData :: Maybe ConnectionData - ,bufferSize :: BuffSize - -- Used by getMailBox, putMailBox - ,comEvent :: IORef (M.Map MailboxId (EVar SData)) - -- multiple wormhole/teleport use the same connection concurrently - ,blocked :: Blocked - ,calling :: Bool - -- local localClosures with his log and his continuation - ,localClosures :: MVar (M.Map IdClosure EventF) - - -- for each remote closure that points to local closure 0, - -- a new container of child processes - -- in order to treat them separately - -- so that 'killChilds' do not kill unrelated processes - ,closChildren :: IORef (M.Map Int EventF)} - - deriving Typeable - - - - - - - - - -defConnection :: (MonadIO m, MonadState EventF m) => m Connection - - --- #ifndef ghcjs_HOST_OS -defConnection = do - idc <- genGlobalId - liftIO $ do - my <- newIORef (error "node in default connection") - x <- newMVar () - y <- newMVar M.empty - noremote <- newIORef Nothing - z <- return $ error "closchildren: newIORef M.empty" - return $ Connection idc my noremote Nothing 8192 - (error "defConnection: accessing network events out of listen") - x False y z - - - -#ifndef ghcjs_HOST_OS -setBuffSize :: Int -> TransIO () -setBuffSize size= Transient $ do - conn<- getData `onNothing` (defConnection !> "DEFF3") - setData $ conn{bufferSize= size} - return $ Just () - -getBuffSize= - (do getSData >>= return . bufferSize) <|> return 8192 - - - - --- | Setup the node to start listening for incoming connections. --- -listen :: Node -> Cloud () -listen (node@(Node _ port _ _ )) = onAll $ do - addThreads 1 - - setData $ Log False [] [] 0 - - conn' <- getSData <|> defConnection - ev <- liftIO $ newIORef M.empty - chs <- liftIO $ newIORef M.empty - let conn= conn'{connData=Just Self, comEvent=ev,closChildren=chs} - pool <- liftIO $ newMVar [conn] - - let node'= node{connection=Just pool} - liftIO $ writeIORef (myNode conn) node' - setData conn - - liftIO $ modifyMVar_ (fromJust $ connection node') $ const $ return [conn] - - addNodes [node'] - - mlog <- listenNew (fromIntegral port) conn <|> listenResponses :: TransIO (StreamData NodeMSG) - return () !> mlog - case mlog of - SMore (RelayMSG _ _ _) -> relay mlog - _ -> execLog mlog - `catcht` (\(e ::SomeException) -> liftIO $ print e) - - --- relayService :: TransIO () - -relay (SMore (RelayMSG origin destiny streamdata)) = do - nodes <- getNodes - my <- getMyNode -- !> "relayService" - if destiny== my - then do - case filter (==origin) nodes of - [node] -> do - (conn: _) <- liftIO $ readMVar $ fromJust $ connection node - setData conn - - [] -> do - conn@Connection{remoteNode= rorigin} <- getState - let conn'= conn{connData= Just $ Relay conn origin} -- !> ("Relay set with: ", origin, destiny) - pool <- liftIO $ newMVar [conn'] - addNodes [origin{connection= Just pool}] - setData conn' - execLog streamdata - - else do - -- search local node name if hostname is the same - - -- let destiny' = if nodeHost destiny== nodeHost my - -- then - -- case filter (==destiny) nodes of - -- [node] -> case lookup "localNode" $ nodeServices node of - -- Just snode -> read snode - -- Nothing -> destiny - -- _ -> destiny - -- else destiny - -- let origin'= if nodeHost origin == "localhost" - -- then case filter (==origin) nodes of - -- [node] ->case lookup "externalNode" $ nodeServices node of - -- Just snode -> read snode - -- Nothing -> origin - -- _ -> origin - -- else origin - - let (origin',destiny')= nat origin destiny my nodes - con <- mconnect destiny' - msend con . SMore $ RelayMSG origin' destiny' streamdata - return () !> ("SEND RELAY DATA",streamdata) - fullStop - - -relay _= empty - -nat origin destiny my nodes= - let destiny' = if nodeHost destiny== nodeHost my - then - case filter (==destiny) nodes of - [node] -> case lookup "localNode" $ nodeServices node of - Just snode -> read snode - Nothing -> destiny - _ -> destiny - else destiny - origin'= if nodeHost origin == "localhost" - then case filter (==origin) nodes of - [node] ->case lookup "externalNode" $ nodeServices node of - Just snode -> read snode - Nothing -> origin - _ -> origin - else origin - in (origin',destiny') - --- listen incoming requests -listenNew port conn'= do - - - sock <- liftIO . listenOn $ PortNumber port - - let bufSize= bufferSize conn' - liftIO $ do NS.setSocketOption sock NS.RecvBuffer bufSize - NS.setSocketOption sock NS.SendBuffer bufSize - - -- wait for connections. One thread per connection - (sock,addr) <- waitEvents $ NS.accept sock - chs <- liftIO $ newIORef M.empty --- case addr of --- NS.SockAddrInet port host -> liftIO $ print("connection from", port, host) --- NS.SockAddrInet6 a b c d -> liftIO $ print("connection from", a, b,c,d) - noNode <- liftIO $ newIORef Nothing - id1 <- genId - let conn= conn'{idConn=id1,closChildren=chs, remoteNode= noNode} - - input <- liftIO $ SBSL.getContents sock - - cutExceptions - - onException $ \(e :: IOException) -> - when (ioeGetLocation e=="Network.Socket.recvBuf") $ do - liftIO $ putStr "listen: " >> print e - - let Connection{remoteNode=rnode,localClosures=localClosures,closChildren= rmap} = conn - -- TODO How to close Connection by discriminating exceptions - mnode <- liftIO $ readIORef rnode - case mnode of - Nothing -> return () - Just node -> do - liftIO $ putStr "removing1 node: " >> print node - nodes <- getNodes - setNodes $ nodes \\ [node] - liftIO $ do - modifyMVar_ localClosures $ const $ return M.empty - writeIORef rmap M.empty - -- topState >>= showThreads - - killBranch - - - setData $ (ParseContext (NS.close sock >> error "Communication error" ) input - ::ParseContext BS.ByteString) - - setState conn{connData=Just (Node2Node (PortNumber port) sock addr)} - maybeTLSServerHandshake sock input - - - - -- (method,uri, headers) <- receiveHTTPHead - (method, uri, vers) <- getFirstLine - case method of - - "CLOS" -> - do - conn <- getSData - sendRaw conn "OK" -- !> "CLOS detected" - - mread conn - - _ -> do - let uri'= BC.tail $ uriPath uri !> uriPath uri - if "api/" `BC.isPrefixOf` uri' - then do - - log <- return $ Exec: (Var $ IDyns $ BS.unpack method):(map (Var . IDyns ) $ split $ BC.unpack $ BC.drop 4 uri') - - - str <- giveData <|> error "no api data" - headers <- getHeaders - maybeSetHost headers - log' <- case (method,lookup "Content-Type" headers) of - ("POST",Just "application/x-www-form-urlencoded") -> do - len <- read <$> BC.unpack - <$> (Transient $ return (lookup "Content-Length" headers)) - setData $ ParseContext (return mempty) $ BS.take len str - - postParams <- parsePostUrlEncoded <|> return [] - return $ log ++ [(Var . IDynamic $ postParams)] - - _ -> return $ log -- ++ [Var $ IDynamic str] - - return $ SMore $ ClosureData 0 0 log' - - else if "relay/" `BC.isPrefixOf` uri' then proxy sock method vers uri' - - else do - headers <- getHeaders - return () !> (method,uri') - -- stay serving pages until a websocket request is received - servePages (method, uri', headers) - conn <- getSData - sconn <- makeWebsocketConnection conn uri headers - -- websockets mode - - - let conn'= conn{connData= Just (Node2Web sconn) - , closChildren=chs} - setState conn' !> "WEBSOCKETS-----------------------------------------------" - onException $ \(e :: SomeException) -> do - cutExceptions - liftIO $ putStr "listen websocket:" >> print e - --liftIO $ mclose conn' - killBranch - empty --- async (return (SMore (0,0,[Exec]))) <|> do - do --- return () !> "WEBSOCKET" - r <- parallel $ do - msg <- WS.receiveData sconn - return () !> ("Server WebSocket msg read",msg) - !> "<-------<---------<--------------" - - case reads $ BS.unpack msg of - [] -> do - let log =Exec: [Var $ IDynamic (msg :: BS.ByteString)] - return $ SMore (ClosureData 0 0 log) - ((x ,_):_) -> return (x :: StreamData NodeMSG) -- StreamData (Int,Int,[LogElem])) - - case r of - SError e -> do --- liftIO $ WS.sendClose sconn ("error" :: BS.ByteString) - back e --- !> "FINISH1" - _ -> return r - - where - uriPath = BC.dropWhile (/= '/') - split []= [] - split ('/':r)= split r - split s= - let (h,t) = span (/= '/') s - in h: split t - - -- reverse proxy for urls that look like http://host:port/relay/otherhost/otherport - proxy sclient method vers uri' = do - -- get host port - let (host:port:_)= split $ BC.unpack $ BC.drop 6 uri' - sserver <- liftIO $ connectTo' 4096 host $ PortNumber $ fromIntegral $ read port - - rawHeaders <- getRawHeaders - let uri= BS.fromStrict $ let d x= BC.tail $ BC.dropWhile (/= '/') x in d . d $ d uri' - - let sent= method <> BS.pack " /" - <> uri - <> BS.cons ' ' vers - <> BS.pack "\r\n" - <> rawHeaders <> BS.pack "\r\n\r\n" - liftIO $ SBSL.send sserver sent - -- Connection{connData=Just (Node2Node _ sclient _)} <- getState <|> error "proxy: no connection" - cutExceptions - onException $ \(e:: SomeException ) -> liftIO $ do - putStr "Proxy: " >> print e - sClose sserver - sClose sclient - - send sclient sserver <|> send sserver sclient - empty - where - send f t= async $ mapData f t - mapData from to = do - content <- recv from 4096 - -- return () !> (" proxy received ", content) - if not $ BC.null content - then sendAll to content >> mapData from to - else finish - where - finish= sClose from >> sClose to - -- throw $ Finish "finish" - - - maybeSetHost headers= do - setHost <- liftIO $ readIORef rsetHost - when setHost $ do - - mnode <- liftIO $ do - let mhost= lookup "Host" headers - case mhost of - Nothing -> return Nothing - Just host -> atomically $ do - -- set the firt node (local node) as is called from outside - nodes <- readTVar nodeList - let (host1,port)= BC.span (/= ':') host - hostnode= (head nodes){nodeHost= BC.unpack host1 - ,nodePort= if BC.null port then 80 - else read $ BC.unpack $ BC.tail port} - writeTVar nodeList $ hostnode : tail nodes - return $ Just hostnode -- !> (host1,port) - - when (isJust mnode) $ do - conn <- getState - liftIO $ writeIORef (myNode conn) $fromJust mnode - liftIO $ writeIORef rsetHost False -- !> "HOSt SET" - -{-#NOINLINE rsetHost #-} -rsetHost= unsafePerformIO $ newIORef True - - - ---instance Read PortNumber where --- readsPrec n str= let [(n,s)]= readsPrec n str in [(fromIntegral n,s)] - - ---deriving instance Read PortID ---deriving instance Typeable PortID -#endif - -listenResponses :: Loggable a => TransIO (StreamData a) -listenResponses= do - (conn, parsecontext, node) <- getMailbox -- :: TransIO (Connection,ParseContext BS.ByteString,Node) - labelState $ "listen from: "++ show node --- return () !> ("LISTEN",case connData conn of Just (Relay _) -> "RELAY"; _ -> "OTHER") - setData conn - -#ifndef ghcjs_HOST_OS - setData (parsecontext :: ParseContext BS.ByteString) -#else - setData (parsecontext :: ParseContext JSString) -#endif - - - - cutExceptions - onException (\(e:: SomeException) -> do - liftIO $ putStr "ListenResponses: " >> print e - liftIO $ putStr "removing node: " >> print node - nodes <- getNodes - setNodes $ nodes \\ [node] - -- topState >>= showThreads - killChilds - let Connection{localClosures=localClosures}= conn - liftIO $ modifyMVar_ localClosures $ const $ return M.empty) - - - mread conn - - - - -type IdClosure= Int - --- The remote closure ids for each node connection -newtype Closure= Closure IdClosure -- deriving Show - - - - - -type RemoteClosure= (Node, IdClosure) - -newtype JobGroup= JobGroup (M.Map String RemoteClosure) deriving Typeable - --- | if there is a remote job identified by th string identifier, it stop that job, and set the --- current remote operation (if any) as the current remote job for this identifier. --- The purpose is to have a single remote job. --- to identify the remote job, it should be used after the `wormhole` and before the remote call: --- --- r <- wormhole node $ do --- stopRemoteJob "streamlog" --- atRemote myRemotejob --- --- So: --- --- runAtUnique ident node job= wormhole node $ do stopRemoteJob ident; aRemote job -stopRemoteJob :: String -> Cloud () -stopRemoteJob ident = do - local $ do - JobGroup map <- getRState <|> return (JobGroup M.empty) - let mj= M.lookup ident map - when (isJust mj) $ putMailbox $ fromJust mj - fixClosure - local $ do - JobGroup map <- getRState <|> return (JobGroup M.empty) - Closure closr <- getData `onNothing` error "resetRemote: Closure not set, use wormhole" - conn <- getData `onNothing` error "resetRemote: no connection set" - remote <- liftIO $ readIORef $ remoteNode conn - when (isJust remote) $ do - setRState $ JobGroup $ M.insert ident (fromJust remote,closr) map - putMailbox (fromJust remote, closr) - - --- kill the remote job. Usually, before starting a new one. -resetRemote :: Cloud () -resetRemote= local $ do - Closure clos <- getState `onNothing` return (Closure 0) - conn <- getData `onNothing` error "resetRemote: no connection set" - remote <- liftIO $ readIORef $ remoteNode conn - when (isJust remote) $ putMailbox (fromJust remote, clos) - --- | delete closures in a remote node when is requested by `resetRemote` or `stopRemoteJob`. --- This is necessary because a remote closure can be reactive or may take a long time. --- --- It should be located as an alternative computation to the program: --- --- > main= initNode $ inputNodes <|> manageClosures <|> myCloudCode -manageClosures = do - (remote, clos) <- local getMailbox - localIO $ print ("MANAGECLOSURESSSSSSSSSSSSSS", clos) - when (clos /= 0) $ runAt remote $ local $ do - conn@Connection {localClosures=localClosures} <- getData `onNothing` error "Listen: myNode not set" - mcont <- liftIO $ modifyMVar localClosures $ \map -> return ( M.delete clos map, M.lookup clos map) - case mcont of - Nothing -> error $ "closure not found: " ++ show clos - Just cont -> do - showThreads $ fromJust $ parent cont - liftIO $ killBranch' cont - return () - - - -execLog :: StreamData NodeMSG -> TransIO () -execLog mlog = Transient $ do - - return () !> "EXECLOG" - case mlog of - SError e -> do - case fromException e of - Just (ErrorCall str) -> do - case read str of - (e@(CloudException _ closl err)) -> do - process closl (error "closr: should not be used") (Left e) True - - - SDone -> runTrans(back $ ErrorCall "SDone") >> return Nothing -- TODO remove closure? - SMore (ClosureData closl closr log) -> process closl closr (Right log) False - SLast (ClosureData closl closr log) -> process closl closr (Right log) True - -- !> ("EXECLOG",mlog) - where - process :: IdClosure -> IdClosure -> (Either CloudException CurrentPointer) -> Bool -> StateIO (Maybe ()) - process closl closr mlog deleteClosure= do - conn@Connection {localClosures=localClosures} <- getData `onNothing` error "Listen: myNode not set" - if closl== 0 then case mlog of - Left except -> do - setData $ Log True [] [] - --setData $ Closure closr - return () !> "THROWWWW1" - runTrans $ throwt except - empty - Right log -> do - -- runTrans cutExceptions !> "CUTEXCEPTIONS" - setData $ Log True log (reverse log) 0 - setData $ Closure closr - - - return $ Just () -- !> "executing top level closure" - else do - - mcont <- liftIO $ modifyMVar localClosures - $ \map -> return (if deleteClosure then - M.delete closl map - else map, M.lookup closl map) - -- !> ("localClosures=", M.size map) - case mcont of - Nothing -> do --- --- if closl == 0 -- add what is after execLog as closure 0 --- then do --- setData $ Log True log $ reverse log --- setData $ Closure closr --- cont <- get !> ("CLOSL","000000000") --- liftIO $ modifyMVar localClosures --- $ \map -> return (M.insert closl ([],cont) map,()) --- return $ Just () --exec what is after execLog (closure 0) --- --- else do - runTrans $ msend conn $ SLast (ClosureData closr closl []) - -- to delete the remote closure - runTrans $ liftIO $ error ("request received for non existent closure: " - ++ show closl) - -- execute the closure - Just cont -> do -- remove fulLog? - liftIO $ runStateT (case mlog of - Right log -> do - Log _ _ fulLog hash <- getData `onNothing` return (Log True [] [] 0) - -- return() !> ("fullog in execlog", reverse fulLog) - let nlog= reverse log ++ fulLog - - setData $ Log True log nlog hash - setData $ Closure closr - - runContinuation cont () - - Left except -> do - setData $ Log True [] [] - --setData $ Closure closr - return () !> "THROWWWW2" - runTrans $ throwt except) cont - return Nothing - - -#ifdef ghcjs_HOST_OS -listen node = onAll $ do - addNodes [node] - - events <- liftIO $ newIORef M.empty - rnode <- liftIO $ newIORef node - conn <- defConnection >>= \c -> return c{myNode=rnode,comEvent=events} - setData conn - r <- listenResponses - execLog r -#endif - -type Pool= [Connection] -type Package= String -type Program= String -type Service= [(Package, Program)] - - - - - --------------------------------------------- - - -#ifndef ghcjs_HOST_OS - - --- maybeRead line= unsafePerformIO $ do --- let [(v,left)] = reads line ----- print v --- (v `seq` return [(v,left)]) --- `catch` (\(e::SomeException) -> do --- liftIO $ print $ "******readStream ERROR in: "++take 100 line --- maybeRead left) - - -readFrom Connection{connData= Just(TLSNode2Node ctx)}= recvTLSData ctx - -readFrom Connection{connData= Just(Node2Node _ sock _)} = toStrict <$> loop - - where - bufSize= 4098 - loop :: IO BL.ByteString - loop = unsafeInterleaveIO $ do - s <- SBS.recv sock bufSize - - if BC.length s < bufSize - then return $ BLC.Chunk s mempty - else BLC.Chunk s `liftM` loop - -readFrom _ = error "readFrom error" - -toStrict= B.concat . BS.toChunks - -makeWSStreamFromConn conn= do - let rec= readFrom conn - send= sendRaw conn - makeStream -- !!> "WEBSOCKETS request" - (do - bs <- rec -- SBS.recv sock 4098 - return $ if BC.null bs then Nothing else Just bs) - (\mbBl -> case mbBl of - Nothing -> return () - Just bl -> send bl) -- SBS.sendMany sock (BL.toChunks bl) >> return()) -- !!> show ("SOCK RESP",bl) - -makeWebsocketConnection conn uri headers= liftIO $ do - - stream <- makeWSStreamFromConn conn - let - pc = WS.PendingConnection - { WS.pendingOptions = WS.defaultConnectionOptions - , WS.pendingRequest = NWS.RequestHead uri headers False -- RequestHead (BC.pack $ show uri) - -- (map parseh headers) False - , WS.pendingOnAccept = \_ -> return () - , WS.pendingStream = stream - } - - - sconn <- WS.acceptRequest pc -- !!> "accept request" - WS.forkPingThread sconn 30 - return sconn - -servePages (method,uri, headers) = do --- return () !> ("HTTP request",method,uri, headers) - conn <- getSData <|> error " servePageMode: no connection" - - if isWebSocketsReq headers - then return () - - - - else do - - let file= if BC.null uri then "index.html" else uri - - {- TODO rendering in server - NEEDED: recodify View to use blaze-html in server. wlink to get path in server - does file exist? - if exist, send else do - store path, execute continuation - get the rendering - send trough HTTP - - put this logic as independent alternative programmer options - serveFile dirs <|> serveApi apis <|> serveNode nodeCode - -} - mcontent <- liftIO $ (Just <$> BL.readFile ( "./static/out.jsexe/"++ BC.unpack file) ) - `catch` (\(e:: SomeException) -> return Nothing) - --- return "Not found file: index.html<br/> please compile with ghcjs<br/> ghcjs program.hs -o static/out") - case mcontent of - Just content -> liftIO $ sendRaw conn $ - "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\nConnection: close\r\nContent-Length: " - <> BS.pack (show $ BL.length content) <>"\r\n\r\n" <> content - - Nothing ->liftIO $ sendRaw conn $ BS.pack $ - "HTTP/1.0 404 Not Found\nContent-Length: 13\nConnection: close\n\nNot Found 404" - empty - - ---counter= unsafePerformIO $ newMVar 0 -api :: TransIO BS.ByteString -> Cloud () -api w= Cloud $ do - conn <- getSData <|> error "api: Need a connection opened with initNode, listen, simpleWebApp" - let send= sendRaw conn - r <- w - send r -- !> r - - - - - - - - - -isWebSocketsReq = not . null - . filter ( (== mk "Sec-WebSocket-Key") . fst) - - -data HTTPMethod= GET | POST deriving (Read,Show,Typeable) - -getFirstLine= (,,) <$> getMethod <*> (toStrict <$> getUri) <*> getVers - where - getMethod= parseString - getUri= parseString - getVers= parseString - -getRawHeaders= dropSpaces >> parse (scan mempty) - -- rs <- manyTill line (string "\r\n\r\n") - -- return $ BS.concat rs - -- where - -- line= parse cond - where - scan res str - | "\r\n\r\n" `BS.isPrefixOf` str= (res, BS.drop 4 str) - | otherwise= scan ( BS.snoc res $ BS.head str) $ BS.tail str - -- line= do - -- dropSpaces - -- tTakeWhile (not . endline) - -type PostParams = [(BS.ByteString, String)] - -parsePostUrlEncoded :: TransIO PostParams -parsePostUrlEncoded= do - dropSpaces - many $ (,) <$> param <*> value - where - param= tTakeWhile' ( /= '=') - value= unEscapeString <$> BS.unpack <$> tTakeWhile' ( /= '&') - - - - -getHeaders = manyTill paramPair (string "\r\n\r\n") -- !> (method, uri, vers) - - where - - - paramPair= (,) <$> (mk <$> getParam) <*> getParamValue - - - getParam= do - dropSpaces - r <- tTakeWhile (\x -> x /= ':' && not (endline x)) - if BS.null r || r=="\r" then empty else dropChar >> return (toStrict r) - - getParamValue= toStrict <$> ( dropSpaces >> tTakeWhile (\x -> not (endline x))) - - - -#endif - - - -#ifdef ghcjs_HOST_OS -isBrowserInstance= True -api _= empty -#else --- | Returns 'True' if we are running in the browser. -isBrowserInstance= False - -#endif - - - - - -{-# NOINLINE emptyPool #-} -emptyPool :: MonadIO m => m (MVar Pool) -emptyPool= liftIO $ newMVar [] - - --- | Create a node from a hostname (or IP address), port number and a list of --- services. -createNodeServ :: HostName -> Int -> Service -> IO Node -createNodeServ h p svs= return $ Node h p Nothing svs - - -createNode :: HostName -> Int -> IO Node -createNode h p= createNodeServ h p [] - -createWebNode :: IO Node -createWebNode= do - pool <- emptyPool - return $ Node "webnode" 0 (Just pool) [("webnode","")] - - -instance Eq Node where - Node h p _ _ ==Node h' p' _ _= h==h' && p==p' - - -instance Show Node where - show (Node h p _ servs )= show (h,p, servs) - -instance Read Node where - readsPrec n s= - let r= readsPrec n s - in case r of - [] -> [] - [((h,p,ss),s')] -> [(Node h p Nothing ss ,s')] - - - - --- inst ghc-options: -threaded -rtsopts - -nodeList :: TVar [Node] -nodeList = unsafePerformIO $ newTVarIO [] - -deriving instance Ord PortID - ---myNode :: Int -> DBRef MyNode ---myNode= getDBRef $ key $ MyNode undefined - -errorMyNode f= error $ f ++ ": Node not set. initialize it with connect, listen, initNode..." - --- | Return the local node i.e. the node where this computation is running. -getMyNode :: TransIO Node -- (MonadIO m, MonadState EventF m) => m Node -getMyNode = do - Connection{myNode= node} <- getSData <|> errorMyNode "getMyNode" :: TransIO Connection - liftIO $ readIORef node - --- | Return the list of nodes in the cluster. -getNodes :: MonadIO m => m [Node] -getNodes = liftIO $ atomically $ readTVar nodeList - --- getEqualNodes= getNodes - -getEqualNodes = do - nodes <- getNodes - let srv= nodeServices $ head nodes - case srv of - [] -> return $ filter (null . nodeServices) nodes - (srv:_) -> return $ filter (\n -> head (nodeServices n) == srv ) nodes - -matchNodes f = do - nodes <- getNodes - return $ map (\n -> filter f $ nodeServices n) nodes - --- | Add a list of nodes to the list of existing cluster nodes. -addNodes :: [Node] -> TransIO () -- (MonadIO m, MonadState EventF m) => [Node] -> m () -addNodes nodes= do --- my <- getMyNode -- mynode must be first - nodes' <- mapM fixNode nodes - liftIO . atomically $ do - prevnodes <- readTVar nodeList - writeTVar nodeList $ nub $ prevnodes ++ nodes' - -fixNode n= case connection n of - Nothing -> do - pool <- emptyPool - return n{connection= Just pool} - Just _ -> return n - --- | set the list of nodes -setNodes nodes= liftIO $ atomically $ writeTVar nodeList $ nodes - - --- | Shuffle the list of cluster nodes and return the shuffled list. -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 --- ifs <- liftIO $ getNetworkInterfaces --- liftIO $ mapM_ (\(i,n) ->putStrLn $ show i ++ "\t"++ show (ipv4 n) ++ "\t"++name n)$ zip [0..] ifs --- liftIO $ putStrLn "Select one: " --- ind <- input ( < length ifs) --- return $ show . ipv4 $ ifs !! ind - - - - --- #ifndef ghcjs_HOST_OS ---instance Read NS.SockAddr where --- readsPrec _ ('[':s)= --- let (s',r1)= span (/=']') s --- [(port,r)]= readsPrec 0 $ tail $ tail r1 --- in [(NS.SockAddrInet6 port 0 (IP.toHostAddress6 $ read s') 0, r)] --- readsPrec _ s= --- let (s',r1)= span(/= ':') s --- [(port,r)]= readsPrec 0 $ tail r1 --- in [(NS.SockAddrInet port (IP.toHostAddress $ read s'),r)] --- #endif - ---newtype MyNode= MyNode Node deriving(Read,Show,Typeable) - - ---instance Indexable MyNode where key (MyNode Node{nodePort=port}) = "MyNode "++ show port --- ---instance Serializable MyNode where --- serialize= BS.pack . show --- deserialize= read . BS.unpack - - - --- | Add a node (first parameter) to the cluster using a node that is already --- part of the cluster (second parameter). The added node starts listening for --- incoming connections and the rest of the computation is executed on this --- newly added node. -connect :: Node -> Node -> Cloud () -#ifndef ghcjs_HOST_OS -connect node remotenode = do - listen node <|> return () - connect' remotenode - - - --- | Reconcile the list of nodes in the cluster using a remote node already --- part of the cluster. Reconciliation end up in each node in the cluster --- having the same list of nodes. -connect' :: Node -> Cloud () -connect' remotenode= loggedc $ do - nodes <- local getNodes - localIO $ putStr "connecting to: " >> print remotenode - - newNodes <- runAt remotenode $ interchange nodes - - local $ return () !> "interchange finish" - - -- add the new nodes to the local nodes in all the nodes connected previously - - let toAdd=remotenode:tail newNodes - callNodes' nodes (<>) mempty $ local $ do - liftIO $ putStr "New nodes: " >> print toAdd !> "NEWNODES" - addNodes toAdd - - where - -- receive new nodes and send their own - interchange nodes= - do - newNodes <- local $ do - conn@Connection{remoteNode=rnode, connData=Just cdata} <- getSData <|> - error ("connect': need to be connected to a node: use wormhole/connect/listen") - - - -- if is a websockets node, add only this node - -- let newNodes = case cdata of - -- Node2Web _ -> [(head nodes){nodeServices=[("relay",show remotenode)]}] - -- _ -> nodes - let newNodes= map (\n -> n{nodeServices= nodeServices n ++ [("relay",show (remotenode,n))]}) nodes - - callingNode<- fixNode $ head newNodes - - liftIO $ writeIORef rnode $ Just callingNode - - liftIO $ modifyMVar_ (fromJust $ connection callingNode) $ const $ return [conn] - - - -- onException $ \(e :: SomeException) -> do - -- liftIO $ putStr "connect:" >> print e - -- liftIO $ putStrLn "removing node: " >> print callingNode - -- -- topState >>= showThreads - -- nodes <- getNodes - -- setNodes $ nodes \\ [callingNode] - - return newNodes - - oldNodes <- local $ getNodes - - - mclustered . local $ do - liftIO $ putStrLn "New nodes: " >> print newNodes - - addNodes newNodes - - localIO $ atomically $ do - -- set the firt node (local node) as is called from outside --- return () !> "HOST2 set" - nodes <- readTVar nodeList - let nodes'= (head nodes){nodeHost=nodeHost remotenode - ,nodePort=nodePort remotenode}:tail nodes - writeTVar nodeList nodes' - - - return oldNodes - -#else -connect _ _= empty -connect' _ = empty -#endif - - - - - +--------------------------------------------------------------------------+--+-- Module : Transient.Move.Internals+-- Copyright :+-- License : MIT+--+-- Maintainer : agocorona@gmail.com+-- Stability :+-- Portability :+--+--+-----------------------------------------------------------------------------+{-# LANGUAGE DeriveDataTypeable , ExistentialQuantification, OverloadedStrings,FlexibleInstances, UndecidableInstances+ ,ScopedTypeVariables, StandaloneDeriving, RecordWildCards, FlexibleContexts, CPP+ ,GeneralizedNewtypeDeriving #-}+module Transient.Move.Internals where++import Prelude hiding(drop,length)++import Transient.Internals+import Transient.Parse+import Transient.Logged+import Transient.Indeterminism+import Transient.Mailboxes+++import Data.Typeable+import Control.Applicative+import System.Random+import Data.String+import qualified Data.ByteString.Char8 as BC+import qualified Data.ByteString.Lazy.Char8 as BS++import System.Time+import Data.ByteString.Builder+++#ifndef ghcjs_HOST_OS+import Network+--- import Network.Info+import Network.URI+--import qualified Data.IP as IP+import qualified Network.Socket as NS+import qualified Network.BSD as BSD+import qualified Network.WebSockets as NWS -- S(RequestHead(..))++import qualified Network.WebSockets.Connection as WS++import Network.WebSockets.Stream hiding(parse)++import qualified Data.ByteString as B(ByteString)+import qualified Data.ByteString.Lazy.Internal as BLC+import qualified Data.ByteString.Lazy as BL+import Network.Socket.ByteString as SBS(sendMany,sendAll,recv)+import qualified Network.Socket.ByteString.Lazy as SBSL+import Data.CaseInsensitive(mk,CI)+import Data.Char+import Data.Aeson+import qualified Data.ByteString.Base64.Lazy as B64++-- import System.Random++#else+import JavaScript.Web.WebSocket+import qualified JavaScript.Web.MessageEvent as JM+import GHCJS.Prim (JSVal)+import GHCJS.Marshal(fromJSValUnchecked)+import qualified Data.JSString as JS+-- import Data.Text.Encoding+import JavaScript.Web.MessageEvent.Internal+import GHCJS.Foreign.Callback.Internal (Callback(..))+import qualified GHCJS.Foreign.Callback as CB+--import Data.JSString (JSString(..), pack,drop,length)++#endif+++import Control.Monad.State+import Control.Monad.Fail+import Control.Exception hiding (onException,try)+import Data.Maybe+--import Data.Hashable+++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 (partition,union,(\\),length, nubBy,isPrefixOf) -- (nub,(\\),intersperse, find, union, length, partition)+--import qualified Data.List(length)+import Data.IORef++import Control.Concurrent++import System.Mem.StableName+import Unsafe.Coerce+import System.Environment++{- TODO+ timeout for closures: little smaller in sender than in receiver+-}++--import System.Random+pk= BS.pack+up= BS.unpack++#ifdef ghcjs_HOST_OS+type HostName = String+newtype PortID = PortNumber Int deriving (Read, Show, Eq, Typeable)+#endif++data Node= Node{ nodeHost :: HostName+ , nodePort :: Int+ , connection :: Maybe (MVar Pool)+ , nodeServices :: [Service]+ }++ deriving (Typeable)++instance Loggable Node++instance Ord Node where+ compare node1 node2= compare (nodeHost node1,nodePort node1)(nodeHost node2,nodePort node2)+++-- The cloud monad is a thin layer over Transient in order to make sure that the type system+-- forces the logging of intermediate results+newtype Cloud a= Cloud {runCloud' ::TransIO a} deriving (AdditionalOperators,Functor,+#ifdef MIN_VERSION_base(4,11,0)+ Semigroup,+#endif+ Monoid ,Applicative, Alternative,MonadFail, Monad, Num, Fractional, MonadState EventF)++{-+instance Applicative Cloud where+ pure a = Cloud $ return a++ Cloud mf <*> Cloud mx = do+ -- bp <- getData `onNothing` error "no backpoint"+ -- local $ onExceptionPoint bp $ \(CloudException _ _ _) -> continue+ r1 <- onAll . liftIO $ newIORef Nothing+ r2 <- onAll . liftIO $ newIORef Nothing+ onAll $ fparallel r1 r2 <|> xparallel r1 r2++ where++ fparallel r1 r2= do+ f <- mf+ liftIO $ (writeIORef r1 $ Just f)+ mr <- liftIO (readIORef r2)+ case mr of+ Nothing -> empty+ Just x -> return $ f x++ xparallel r1 r2 = do++ mr <- liftIO (readIORef r1)+ case mr of+ Nothing -> do++ p <- gets execMode++ if p== Serial then empty else do+ x <- mx+ liftIO $ (writeIORef r2 $ Just x)++ mr <- liftIO (readIORef r1)+ case mr of+ Nothing -> empty+ Just f -> return $ f x+++ Just f -> do+ x <- mx+ liftIO $ (writeIORef r2 $ Just x)+ return $ f x+-}+++type UPassword= BS.ByteString+type Host= BS.ByteString++type ProxyData= (UPassword,Host,Int)+rHTTPProxy= unsafePerformIO $ newIORef (Nothing :: Maybe (Maybe ProxyData, Maybe ProxyData))+++getHTTProxyParams t= do+ mp <- liftIO $ readIORef rHTTPProxy+ case mp of+ Just (p1,p2) -> return $ if t then p2 else p1+ Nothing -> do + ps <- (,) <$> getp "http" <*> getp "https"+ liftIO $ writeIORef rHTTPProxy $ Just ps+ getHTTProxyParams t+ where+ getp t= do+ let var= t ++ "_proxy"++ p<- liftIO $ lookupEnv var+ tr ("proxy",p)+ case p of+ Nothing -> return Nothing+ Just hp -> do + pr<- withParseString (BS.pack hp) $ do+ tDropUntilToken (BS.pack "//") <|> return ()+ (,,) <$> getUPass <*> tTakeWhile' (/=':') <*> int+ return $ Just pr+ getUPass= tTakeUntilToken "@" <|> return ""++-- | Execute a distributed computation inside a TransIO computation.+-- All the computations in the TransIO monad that enclose the cloud computation must be `logged`+runCloud :: Cloud a -> TransIO a++runCloud x= do+ closRemote <- getState <|> return (Closure 0)+ runCloud' x <*** setState closRemote+++--instance Monoid a => Monoid (Cloud a) where+-- f mappend x y = mappend <$> x <*> y+-- mempty= return mempty++#ifndef ghcjs_HOST_OS++--- empty Hooks for TLS++{-# NOINLINE tlsHooks #-}+tlsHooks ::IORef (Bool+ ,SData -> BS.ByteString -> IO ()+ ,SData -> IO B.ByteString+ ,NS.Socket -> BS.ByteString -> TransIO ()+ ,String -> NS.Socket -> BS.ByteString -> TransIO ()+ ,SData -> IO ())+tlsHooks= unsafePerformIO $ newIORef+ ( False+ , notneeded+ , notneeded+ , \_ i -> tlsNotSupported i+ , \_ _ _-> return()+ , \_ -> return())++ where+ notneeded= error "TLS hook function called"++++ tlsNotSupported input = do+ if ((not $ BL.null input) && BL.head input == 0x16)+ then do+ conn <- getSData+ sendRaw conn $ BS.pack $ "HTTP/1.0 525 SSL Handshake Failed\r\nContent-Length: 0\nConnection: close\r\n\r\n"+ else return ()++(isTLSIncluded,sendTLSData,recvTLSData,maybeTLSServerHandshake,maybeClientTLSHandshake,tlsClose)= unsafePerformIO $ readIORef tlsHooks+++#endif++-- | Means that this computation will be executed in the current node. the result will be logged+-- so the closure will be recovered if the computation is translated to other node by means of+-- primitives like `beamTo`, `forkTo`, `runAt`, `teleport`, `clustered`, `mclustered` etc+local :: Loggable a => TransIO a -> Cloud a+local = Cloud . logged++--stream :: Loggable a => TransIO a -> Cloud (StreamVar a)+--stream= Cloud . transport++-- #ifndef ghcjs_HOST_OS+-- | Run a distributed computation inside the IO monad. Enables asynchronous+-- console input (see 'keep').+runCloudIO :: Typeable a => Cloud a -> IO (Maybe a)+runCloudIO (Cloud mx)= keep mx++-- | Run a distributed computation inside the IO monad with no console input.+runCloudIO' :: Typeable a => Cloud a -> IO (Maybe a)+runCloudIO' (Cloud mx)= keep' mx++-- #endif++-- | alternative to `local` It means that if the computation is translated to other node+-- this will be executed again if this has not been executed inside a `local` computation.+--+-- > onAll foo+-- > local foo'+-- > local $ do+-- > bar+-- > runCloud $ do+-- > onAll baz+-- > runAt node ....+-- > runAt node' .....+--+-- foo bar and baz will e executed locally.+-- But foo will be executed remotely also in node' while foo' bar and baz don't.+--++--++onAll :: TransIO a -> Cloud a+onAll = Cloud++-- | only executes if the result is demanded. It is useful when the conputation result is only used in+-- the remote node, but it is not serializable. All the state changes executed in the argument with+-- `setData` `setState` etc. are lost+lazy :: TransIO a -> Cloud a+lazy mx= onAll $ do+ st <- get+ return $ fromJust $ 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++-- | 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 <- getState <|> return (Closure 0 )+ (fixRemote :: Maybe LocalFixData) <- getData+ logged mx <*** do setData closRemote+ when (isJust fixRemote) $ setState (fromJust fixRemote)++++loggedc' :: Loggable a => Cloud a -> Cloud a+loggedc' (Cloud mx)= Cloud $ do+ fixRemote :: Maybe LocalFixData <- getData+ logged mx <*** (when (isJust fixRemote) $ setState (fromJust fixRemote))+++++-- | the `Cloud` monad has no `MonadIO` instance. `lliftIO= local . liftIO`+lliftIO :: Loggable a => IO a -> Cloud a+lliftIO= local . liftIO++-- | `localIO = lliftIO`+localIO :: Loggable a => IO a -> Cloud a+localIO= lliftIO++++-- | continue the execution in a new node+beamTo :: Node -> Cloud ()+beamTo node = wormhole node teleport+++-- | execute in the remote node a process with the same execution state+forkTo :: Node -> Cloud ()+forkTo node= beamTo node <|> return()++-- | open a wormhole to another node and executes an action on it.+-- currently by default it keep open the connection to receive additional requests+-- and responses (streaming)+callTo :: Loggable a => Node -> Cloud a -> Cloud a+callTo node remoteProc= wormhole' node $ atRemote remoteProc+++#ifndef ghcjs_HOST_OS+-- | A connectionless version of callTo for long running remote calls+callTo' :: (Show a, Read a,Typeable a) => Node -> Cloud a -> Cloud a+callTo' node remoteProc= do+ mynode <- local $ getNodes >>= return . Prelude.head+ beamTo node+ r <- remoteProc+ beamTo mynode+ return r+#endif++-- | Within a connection to a node opened by `wormhole`, it run the computation in the remote node and return+-- the result back to the original node.+--+-- If `atRemote` is executed in the remote node, then the computation is executed in the original node+--+-- > wormhole node2 $ do+-- > t <- atRemote $ do+-- > r <- foo -- executed in node2+-- > s <- atRemote bar r -- executed in the original node+-- > baz s -- in node2+-- > bat t -- in the original node++atRemote :: Loggable a => Cloud a -> Cloud a+atRemote proc= loggedc' $ do+ --modify $ \s -> s{execMode=Parallel}+ teleport -- !> "teleport 1111"++ modify $ \s -> s{execMode= if execMode s== Parallel then Parallel else Serial} -- modifyData' f1 Serial+ {-+ local $ noTrans $ do+ cont <- get++ let loop=do+ chs <- liftIO $ readMVar $ children $ fromJust $ parent cont++ tr ("THREADS ***************", length chs)+ threadDelay 1000000+ loop++ liftIO $ forkIO loop++ return()+ -}+ r <- loggedc $ proc <** modify (\s -> s{execMode= Remote}) -- setData Remote++ teleport -- !> "teleport 2222"++ return r+++-- | Execute a computation in the node that initiated the connection.+--+-- if the sequence of connections is n1 -> n2 -> n3 then `atCallingNode $ atCallingNode foo` in n3+-- would execute `foo` in n1, -- while `atRemote $ atRemote foo` would execute it in n3+-- atCallingNode :: Loggable a => Cloud a -> Cloud a+-- atCallingNode proc= connectCaller $ atRemote proc++-- | synonymous of `callTo`+runAt :: Loggable a => Node -> Cloud a -> Cloud a+runAt= callTo+++-- | run a single thread with that action for each connection created.+-- When the same action is re-executed within that connection, all the threads generated by the previous execution+-- are killed+--+-- > box <- foo+-- > r <- runAt node . local . single $ getMailbox box+-- > localIO $ print r+--+-- if foo return different mainbox indentifiers, the above code would print the+-- messages of the last one.+-- Without single, it would print the messages of all of them since each call would install a new `getMailBox` for each one of them+single :: TransIO a -> TransIO a+single f= do+ cutExceptions+ Connection{closChildren=rmap} <- getSData <|> error "single: only works within a connection"+ mapth <- liftIO $ readIORef rmap+ id <- liftIO $ f `seq` makeStableName f >>= return . hashStableName++ case M.lookup id mapth of+ Just tv -> liftIO $ killBranch' tv+ Nothing -> return ()++ tv <- get+ f <** do+ id <- liftIO $ makeStableName f >>= return . hashStableName+ liftIO $ modifyIORef rmap $ \mapth -> M.insert id tv mapth+++-- | run an unique continuation for each connection. The first thread that execute `unique` is+-- executed for that connection. The rest are ignored.+unique :: TransIO a -> TransIO a+unique f= do+ Connection{closChildren=rmap} <- getSData <|> error "unique: only works within a connection. Use wormhole"+ mapth <- liftIO $ readIORef rmap+ id <- liftIO $ f `seq` makeStableName f >>= return . hashStableName++ let mx = M.lookup id mapth+ case mx of+ Just _ -> empty+ Nothing -> do+ tv <- get+ liftIO $ modifyIORef rmap $ \mapth -> M.insert id tv mapth+ f++-- | A wormhole opens a connection with another node anywhere in a computation.+-- `teleport` uses this connection to translate the computation back and forth between the two nodes connected.+-- If the connection fails, it search the network for suitable relay nodes to reach the destination node.+wormhole node comp= do+ onAll $ onException $ \(e@(ConnectionError "no connection" nodeerr)) ->+ if nodeerr== node then do runCloud' $ findRelay node ; continue else return ()+ wormhole' node comp+++ where+ findRelay node = do+ relaynode <- exploreNetUntil $ do+ nodes <- local getNodes+ let thenode= filter (== node) nodes+ if not (null thenode) && isJust(connection $ Prelude.head thenode ) then return $ Prelude.head nodes else empty+ local $ addNodes [node{nodeServices= {-nodeServices node ++ -} [[("relay", show (nodeHost (relaynode :: Node),nodePort relaynode ))]]}]++-- when the first teleport has been sent within a wormhole, the+-- log sent should be the segment not send in the previous teleport+newtype DialogInWormholeInitiated= DialogInWormholeInitiated Bool+++-- | wormhole without searching for relay nodes.+wormhole' :: Loggable a => Node -> Cloud a -> Cloud a+wormhole' node (Cloud comp) = local $ Transient $ do++ moldconn <- getData :: StateIO (Maybe Connection)+ mclosure <- getData :: StateIO (Maybe Closure)+ mdialog <- getData :: StateIO (Maybe ( Ref DialogInWormholeInitiated))+ -- when (isJust moldconn) . setState $ ParentConnection (fromJust moldconn) mclosure++ labelState $ "wormhole" <> BC.pack (show node)+ log <- getLog+ + if not $ recover log+ then runTrans $ (do+ conn <- mconnect node+ + liftIO $ writeIORef (remoteNode conn) $ Just node+ setData conn{synchronous= maybe False id $ fmap synchronous moldconn, calling= True}+++ setState $ Closure 0+ newRState $ DialogInWormholeInitiated False+ --lhls <- liftIO $ atomicModifyIORef (wormholes conn) $ \hls -> ((ref:hls),length hls)+ --tr ("LENGTH HLS",lhls)+++ comp )+ <*** do+ when (isJust moldconn) . setData $ fromJust moldconn+ when (isJust mclosure) . setData $ fromJust mclosure+ when (isJust mdialog) . setData $ fromJust mdialog+ -- <** is not enough since comp may be reactive+ else do+ -- tr "YES REC"+ let conn = fromMaybe (error "wormhole: no connection in remote node") moldconn+ setData $ conn{calling= False}+ runTrans $ comp+ <*** do when (isJust mclosure) . setData $ fromJust mclosure++++-- #ifndef ghcjs_HOST_OS+-- type JSString= String+-- pack= id+-- #endif++data CloudException = CloudException Node IdClosure String deriving (Typeable, Show, Read)++instance Exception CloudException++-- | set remote invocations synchronous+-- this is necessary when data is transfered very fast from node to node in a stream non-deterministically+-- in order to keep the continuation of the calling node unchanged until the arrival of the response+-- since all the calls share a single continuation in the calling node.+--+-- If there is no response from the remote node, the streaming is interrupted+--+-- > main= keep $ initNode $ onBrowser $ do+-- > local $ setSynchronous True+-- > line <- local $ threads 0 $ choose[1..10::Int]+-- > localIO $ print ("1",line)+-- > atRemote $ localIO $ print line+-- > localIO $ print ("2", line)++setSynchronous :: Bool -> TransIO ()+setSynchronous sync= do+ modifyData'(\con -> con{synchronous=sync}) (error "setSynchronous: no communication data")+ return ()++-- set synchronous mode for remote calls within a cloud computation and also avoid unnecessary+-- thread creation+syncStream :: Cloud a -> Cloud a+syncStream proc= do+ sync <- local $ do+ Connection{synchronous= synchronous} <- modifyData'(\con -> con{synchronous=True}) err+ return synchronous+ Cloud $ threads 0 $ runCloud' proc <*** modifyData'(\con -> con{synchronous=sync}) err+ where err= error "syncStream: no communication data"++++teleport :: Cloud ()+teleport = do++ modify $ \s -> s{execMode=if execMode s == Remote then Remote else Parallel}+ local $ do+ conn@Connection{connData=contype, synchronous=synchronous, localClosures= localClosures} <- getData+ `onNothing` error "teleport: No connection defined: use wormhole"+ -- onException $ \(e :: IOException ) -> do -- to retry the connection in case of failure+ -- tr ("teleport:", e) -- should be three tries at most+ -- liftIO $ writeIORef contype Nothing+ -- mclose conn -- msend will open a new connection. move that open here?+ -- continue++ Transient $ do+ labelState "teleport"++ cont <- get++ log <- getLog+++ if not $ recover log -- !> ("teleport rec,loc fulLog=",rec,log,fulLog)+ -- if is not recovering in the remote node then it is active++ then do++ -- when a node call itself, there is no need of socket communications+ ty <- liftIO $ readIORef contype+ case ty of+ Just Self -> runTrans $ do+ modify $ \s -> s{execMode= Parallel} -- setData Parallel+ abduce -- !> "SELF" -- call himself+ liftIO $ do+ remote <- readIORef $ remoteNode conn+ writeIORef (myNode conn) $ fromMaybe (error "teleport: no connection?") remote++ _ -> do++ --read this Closure++ DialogInWormholeInitiated initiated <- getRData `onNothing` return(DialogInWormholeInitiated True)+ --detecta si ya ha enviado en un mismo wormhole+ -- como detectar eso sin usar Closure?+ -- un Rflag en estado ejecución+ --tr("INITIATED",initiated,closRemote/=0)+ (closRemote',tosend) <- if initiated+ -- for localFix+ then do+ Closure closRemote <- getData `onNothing` return (Closure 0 )+ tr ("REMOTE CLOSURE",closRemote)+ return (closRemote, buildLog log)+ else do+ mfix <- getData -- mirar globalFix+ tr ("mfix", mfix)+ let droplog Nothing= return (0, fulLog log)+ droplog (Just localfix)= do+ sent <- liftIO $ atomicModifyIORef' (fixedConnections localfix) $ \list -> do+ let n= idConn conn+ if n `Prelude.elem` list+ then (list, True)+ else (n:list,False)+++ tr ("LOCALFIXXXXXXXXXX",localfix)+ let dropped= lazyByteString $ BS.drop (fromIntegral $ lengthFix localfix) $ toLazyByteString $ fulLog log+ if sent then return (closure localfix, dropped)+ else if isService localfix then return (0, dropped)+ else droplog $ prevFix localfix -- look for other previous closure sent+++ droplog mfix+++ let closLocal= hashClosure log+ map <- liftIO $ readMVar localClosures+ let mr = M.lookup closLocal map+ pair <- case mr of+ -- for synchronous streaming+ Just (chs,clos,mvar,_) -> do+ when synchronous $ liftIO $ takeMVar mvar+ -- tr ("TELEPORT removing", (Data.List.length $unsafePerformIO $ readMVar chs)-1)+ --ths <- liftIO $ readMVar (children cont)+ --liftIO $ when (length ths > 1)$ mapM_ (killChildren . children) $ tail ths+ --runTrans $ msend conn $ SLast (ClosureData closRemote' closLocal mempty)+ --no se llama se hace asincronamente en el blucle loopclosures+ return (children ${- fromJust $ parent -} cont,clos,mvar,cont)++ _ -> liftIO $ do mv <- newEmptyMVar; return ( children $ fromJust $ parent cont,closRemote',mv,cont)++ liftIO $ modifyMVar_ localClosures $ \map -> return $ M.insert closLocal pair map+++ -- The log sent is in the order of execution. log is in reverse order++ -- send log with closure ids at head+ --tr ("MSEND --------->------>", SMore (unsafePerformIO $ readIORef $ remoteNode conn,closRemote',closLocal,toLazyByteString tosend))+ runTrans $ msend conn $ SMore $ ClosureData closRemote' closLocal tosend+++ return Nothing++ else return $ Just ()+++++++{- |+One problem of forwarding closures for streaming is that it could transport not only the data but extra information that reconstruct the closure in the destination node. In a single in-single out interaction It may not be a problem, but think, for example, when I have to synchronize N editors by forwarding small modifications, or worst of all, when transmitting packets of audio or video. But the size of the closure, that is, the amount of variables that I have to transport increases when the code is more complex. But transient build closures upon closures, so It has to send only what has changed since the last interaction.++In one-to-one interactions whithin a wormhole, this is automatic, but when there are different wormholes involved, it is necessary+to tell explicitly what is the closure that will continue the execution. this is what `localFix` does. otherwise it will use the closure 0.++> main= do+> filename <- local input+> source <- atServer $ local $ readFile filename+> local $ render source inEditor+> -- send upto here one single time please, so I only stream the deltas+> localFix+> delta <- react onEachChange+> forallNodes $ update delta++if forwardChanges send to all the nodes editing the document, the data necessary to reconstruct the+closure would include even the source code of the file on EACH change.+Fortunately it is possible to fix a closure that will not change in all the remote nodes so after that,+I only have to send the only necessary variable, the delta. This is as efficient as an hand-made+socket write/forkThread/readSocket loop for each node.+-}+localFix= localFixServ False False+type ConnectionId= Int+type HasClosed= Bool+-- for each connection, the list of closures fixed and the list of connections which created that closure in the remote node+-- unificar para todas las conexiones+-- pero como se sabe si una closure global aplica a un envio despues de una desconexion?+-- el programa tiene que pasar por esa globalClosure,+-- si solo se ha perdido la conexión, tiene estado y puede utilizarla+-- si ha rearrancado, ha ejecutado hasta ahi y tiene que reconstruir su estado de localFix++globalFix = unsafePerformIO $ newIORef (M.empty :: M.Map ConnectionId (HasClosed,[(IdClosure, IORef [ConnectionId ])]))+-- how to signal that was closed?++data LocalFixData= LocalFixData{ isService :: Bool+ , lengthFix :: Int+ , closure :: Int+ , fixedConnections :: IORef [ConnectionId] -- List of connections that created+ -- that closure in the remote node++ , prevFix :: Maybe LocalFixData} deriving Show++instance Show a => Show (IORef a) where+ show r= show $ unsafePerformIO $ readIORef r++-- data LocalFixData= LocalFixData Bool Int Int (IORef (M.Map Int Int))++-- first flag=True assumes that the localFix closure has been created otherwise+-- the first request invoke closure 0 and create the localFix closure+-- further request will invoque this closure+--+-- the second flag creates a closure that is invoked ever, even if localfix is re-executed.+--If this second flag is false,+-- a reexecution of localFix will recreate the remote closure, perhaps with different variables.+localFixServ isService isGlobal= Cloud $ noTrans $ do+ log <- getLog+ Connection{..} <- getData `onNothing` error "teleport: No connection set: use initNode"++ if recover log+ then do+ cont <- get+ mv <- liftIO newEmptyMVar+ liftIO $ modifyMVar_ localClosures $ \map -> return $ M.insert (hashClosure log) ( children $ fromJust $ parent cont,0,mv,cont) map++ else do++ mprevFix <- getData+++ ref <- liftIO $ if not $ isGlobal then newIORef [] else do+ map <- readIORef globalFix+ return $ do+ (_,l) <- M.lookup idConn map+ lookup (hashClosure log) l++ `onNothing` do+ ref <- newIORef []+ modifyIORef globalFix $ \map ->+ let (closed,l)= fromMaybe (False,[]) $ M.lookup idConn map+ in M.insert idConn (closed,(hashClosure log, ref):l) map+ return ref+ mmprevFix <- liftIO $ readIORef ref >>= \l -> return $ if Prelude.null l then Nothing else mprevFix+ let newfix =LocalFixData{ isService = isService+ , lengthFix = fromIntegral $ BS.length $ toLazyByteString $ fulLog log+ , closure = hashClosure log+ , fixedConnections = ref+ , prevFix = mmprevFix}+ setState newfix+++ !> ("SET LOCALFIX", newfix )+++-- | forward exceptions back to the calling node+reportBack :: TransIO ()+reportBack= onException $ \(e :: SomeException) -> do+ conn <- getData `onNothing` error "reportBack: No connection defined: use wormhole"+ Closure closRemote <- getData `onNothing` error "teleport: no closRemote"+ node <- getMyNode+ let msg= SError $ toException $ ErrorCall $ show $ show $ CloudException node closRemote $ show e+ msend conn msg !> "MSEND"++++-- | copy a session data variable from the local to the remote node.+-- If there is none set in the local node, The parameter is the default value.+-- In this case, the default value is also set in the local node.+copyData def = do+ r <- local getSData <|> return def+ onAll $ setData r+ return r+++-- | execute a Transient action in each of the nodes connected.+--+-- The response of each node is received by the invoking node and processed by the rest of the procedure.+-- By default, each 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 => Cloud a -> Cloud a+clustered proc= callNodes (<|>) empty proc+++-- A variant of `clustered` that wait for all the responses and `mappend` them+mclustered :: (Monoid a, Loggable a) => Cloud a -> Cloud a+mclustered proc= callNodes (<>) mempty proc+++callNodes op init proc= loggedc' $ do+ nodes <- local getEqualNodes+ callNodes' nodes op init proc+++callNodes' nodes op init proc= loggedc' $ Prelude.foldr op init $ Prelude.map (\node -> runAt node proc) nodes+-----+#ifndef ghcjs_HOST_OS+sendRawRecover con r= do + + c <- liftIO $ readIORef $ connData con+ con' <- case c of+ Nothing -> do+ tr "CLOSED CON"+ n <- liftIO $ readIORef $ remoteNode con+ case n of+ Nothing -> error "connection closed by caller"+ Just node -> do+ r <- mconnect' node+ return r++ Just _ -> return con+ sendRaw con' r++ `whileException` \(SomeException _)->+ liftIO$ writeIORef (connData con) Nothing++sendRaw con r= do+ let blocked= isBlocked con+ c <- liftIO $ readIORef $ connData con+ liftIO $ modifyMVar_ blocked $ const $ do+ tr ("sendRaw",r)+ case c of+ Just (Node2Web sconn ) -> liftIO $ WS.sendTextData sconn r+ Just (Node2Node _ sock _) ->+ SBS.sendMany sock (BL.toChunks r )++ Just (TLSNode2Node ctx ) ->+ sendTLSData ctx r+ _ -> error "No connection stablished"+ TOD time _ <- getClockTime+ return $ Just time++{-++sendRaw (Connection _ _ _ (Just (Node2Web sconn )) _ _ _ _ _ _ _) r=+ liftIO $ WS.sendTextData sconn r -- !> ("NOde2Web",r)++sendRaw (Connection _ _ _ (Just (Node2Node _ sock _)) _ _ blocked _ _ _ _) r=+ liftIO $ withMVar blocked $ const $ SBS.sendMany sock+ (BL.toChunks r ) -- !> ("NOde2Node",r)++sendRaw (Connection _ _ _(Just (TLSNode2Node ctx )) _ _ blocked _ _ _ _) r=+ liftIO $ withMVar blocked $ const $ sendTLSData ctx r !> ("TLNode2Web",r)+-}+#else+sendRaw con r= do+ c <- liftIO $ readIORef $ connData con+ case c of+ Just (Web2Node sconn) ->+ JavaScript.Web.WebSocket.send r sconn+ _ -> error "No connection stablished"++{-+sendRaw (Connection _ _ _ (Just (Web2Node sconn)) _ _ blocked _ _ _ _) r= liftIO $+ withMVar blocked $ const $ JavaScript.Web.WebSocket.send r sconn -- !> "MSEND SOCKET"+-}++#endif+++++data NodeMSG= ClosureData IdClosure IdClosure Builder deriving (Read, Show)+++instance Loggable NodeMSG where+ serialize (ClosureData clos clos' build)= intDec clos <> "/" <> intDec clos' <> "/" <> build+ deserialize= ClosureData <$> (int <* tChar '/') <*> (int <* tChar '/') <*> restOfIt+ where+ restOfIt= lazyByteString <$> giveParseString++instance Show Builder where+ show b= BS.unpack $ toLazyByteString b++instance Read Builder where+ readsPrec _ str= [(lazyByteString $ BS.pack $ read str,"")]+++instance Loggable a => Loggable (StreamData a) where+ serialize (SMore x)= byteString "SMore/" <> serialize x+ serialize (SLast x)= byteString "SLast/" <> serialize x+ serialize SDone= byteString "SDone"+ serialize (SError e)= byteString "SError/" <> serialize e++ deserialize = smore <|> slast <|> sdone <|> serror+ where+ smore = symbol "SMore/" >> (SMore <$> deserialize)+ slast = symbol "SLast/" >> (SLast <$> deserialize)+ sdone = symbol "SDone" >> return SDone+ serror= symbol "SError/" >> (SError <$> deserialize)+{-+ en msend escribir la longitud del paquete y el paquete+ en mread cojer la longitud y el mensaje+++data Packet= Packet Int BS.ByteString deriving (Read,Show)+instance Loggable Packet where+ serialize (Packet len msg) = intDec len <> lazyByteString msg+ deserialize = do+ len <- int+ Packet len <$> tTake (fromIntegral len)++-}+++msend :: Connection -> StreamData NodeMSG -> TransIO ()++-- msend (Connection _ _ _ (Just Self) _ _ _ _ _ _ _) r= return ()++#ifndef ghcjs_HOST_OS++msend con r= do+ tr ("MSEND --------->------>", r)+ c <- liftIO $ readIORef $ connData con+ con' <- case c of+ Nothing -> do+ tr "CLOSED CON"+ n <- liftIO $ readIORef $ remoteNode con+ case n of+ Nothing -> error "connection closed by caller"+ Just node -> do+ r <- mconnect node++ return r+ --case r of+ -- Nothing -> error $ "can not reconnect with " ++ show n+ -- Just c -> return c+ Just _ -> return con+ let blocked= isBlocked con'+ c <- liftIO $ readIORef $ connData con'+ let bs = toLazyByteString $ serialize r++ do+ --liftIO $ do+ case c of++ Just (TLSNode2Node ctx) -> liftIO $ modifyMVar_ blocked $ const $ do+ tr "TLSSSSSSSSSSS SEND"+ sendTLSData ctx $ toLazyByteString $ int64Dec $ BS.length bs+ sendTLSData ctx bs+ TOD time _ <- getClockTime+ return $ Just time++ Just (Node2Node _ sock _) -> liftIO $ modifyMVar_ blocked $ const $ do+ tr "NODE2NODE SEND"+ SBSL.send sock $ toLazyByteString $ int64Dec $ BS.length bs+ SBSL.sendAll sock bs+ TOD time _ <- getClockTime+ return $ Just time++ Just (HTTP2Node _ sock _) -> liftIO $ modifyMVar_ blocked $ const $ do+ tr "HTTP2NODE SEND"+ SBSL.sendAll sock $ bs <> "\r\n"+ TOD time _ <- getClockTime+ return $ Just time++ Just (HTTPS2Node ctx) -> liftIO $ modifyMVar_ blocked $ const $ do+ tr "HTTPS2NODE SEND"+ sendTLSData ctx $ bs <> "\r\n"+ TOD time _ <- getClockTime+ return $ Just time++ Just (Node2Web sconn) -> do+ tr "NODE2WEB"+ -- {-withMVar blocked $ const $ -} WS.sendTextData sconn $ serialize r -- BS.pack (show r) !> "websockets send"+ liftIO $ do+ let bs = toLazyByteString $ serialize r+ -- WS.sendTextData sconn $ toLazyByteString $ int64Dec $ BS.length bs+ tr "ANTES SEND"++ WS.sendTextData sconn bs -- !> ("N2N SEND", bd)+ tr "AFTER SEND"+ Just Self -> error "connection to the same node shouldn't happen, file a bug please"+ _ -> error "msend out of connection context: use wormhole to connect"++ + -- return()++{-+msend (Connection _ _ _ (Just (Node2Node _ sock _)) _ _ blocked _ _ _ _) r=do+ liftIO $ withMVar blocked $ const $ do+ let bs = toLazyByteString $ serialize r+ SBSL.send sock $ toLazyByteString $ int64Dec $ BS.length bs+ SBSL.sendAll sock bs -- !> ("N2N SEND", bd)++msend (Connection _ _ _ (Just (HTTP2Node _ sock _)) _ _ blocked _ _ _ _) r=do+ liftIO $ withMVar blocked $ const $ do+ let bs = toLazyByteString $ serialize r+ let len= BS.length bs+ lenstr= toLazyByteString $ int64Dec $ len++ SBSL.send sock $ "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\nContent-Length: "+ <> lenstr+ -- <>"\r\n" <> "Set-Cookie:" <> "cookie=" <> cook -- <> "\r\n"+ <>"\r\n\r\n"++ SBSL.sendAll sock bs -- !> ("N2N SEND", bd)++msend (Connection _ _ _ (Just (TLSNode2Node ctx)) _ _ _ _ _ _ _) r= liftIO $ do+ let bs = toLazyByteString $ serialize r+ sendTLSData ctx $ toLazyByteString $ int64Dec $ BS.length bs+ sendTLSData ctx bs -- !> "TLS SEND"+++msend (Connection _ _ _ (Just (Node2Web sconn)) _ _ _ _ _ _ _) r=+ -- {-withMVar blocked $ const $ -} WS.sendTextData sconn $ serialize r -- BS.pack (show r) !> "websockets send"+ liftIO $ do+ let bs = toLazyByteString $ serialize r+ WS.sendTextData sconn $ toLazyByteString $ int64Dec $ BS.length bs+ WS.sendTextData sconn bs -- !> ("N2N SEND", bd)++-}++#else+msend con r= do+ tr ("MSEND --------->------>", r)++ let blocked= isBlocked con+ c <- liftIO $ readIORef $ connData con+ case c of+ Just (Web2Node sconn) -> liftIO $ do+ tr "MSEND BROWSER"+ --modifyMVar_ (isBlocked con) $ const $ do + let bs = toLazyByteString $ serialize r+ JavaScript.Web.WebSocket.send (JS.pack $ BS.unpack bs) sconn -- TODO OPTIMIZE THAT!+ tr "AFTER MSEND"+ --TOD time _ <- getClockTime+ --return $ Just time++++ _ -> error "msend out of connection context: use wormhole to connect"+{-+msend (Connection _ _ remoten (Just (Web2Node sconn)) _ _ blocked _ _ _ _) r= liftIO $ do++ withMVar blocked $ const $ do -- JavaScript.Web.WebSocket.send (serialize r) sconn -- (JS.pack $ show r) sconn !> "MSEND SOCKET"+ let bs = toLazyByteString $ serialize r+ JavaScript.Web.WebSocket.send (toLazyByteString $ int64Dec $ BS.length bs) sconn+ JavaScript.Web.WebSocket.send bs sconn++-}++#endif+++#ifdef ghcjs_HOST_OS++mread con= do+ labelState "mread"+ sconn <- liftIO $ readIORef $ connData con+ case sconn of+ Just (Web2Node sconn) -> wsRead sconn+ Nothing -> error "connection not opened"+++--mread (Connection _ _ _ (Just (Web2Node sconn)) _ _ _ _ _ _ _)= wsRead sconn++++wsRead :: Loggable a => WebSocket -> TransIO a+wsRead ws= do+ dat <- react (hsonmessage ws) (return ())+ tr "received"++ case JM.getData dat of+ JM.StringData ( text) -> do+ setParseString $ BS.pack . JS.unpack $ text -- TODO OPTIMIZE THAT++ --len <- integer+ tr ("Browser webSocket read", text) !> "<------<----<----<------"+ deserialize -- return (read' $ JS.unpack str)+ + JM.BlobData blob -> error " blob"+ JM.ArrayBufferData arrBuffer -> error "arrBuffer"++++wsOpen :: JS.JSString -> TransIO WebSocket+wsOpen url= do+ ws <- liftIO $ js_createDefault url -- !> ("wsopen",url)+ react (hsopen ws) (return ()) -- !!> "react"+ return ws -- !!> "AFTER ReACT"++foreign import javascript safe+ "window.location.hostname"+ js_hostname :: JSVal++foreign import javascript safe+ "window.location.pathname"+ js_pathname :: JSVal++foreign import javascript safe+ "window.location.protocol"+ js_protocol :: JSVal++foreign import javascript safe+ "(function(){var res=window.location.href.split(':')[2];if (res === undefined){return 80} else return res.split('/')[0];})()"+ js_port :: JSVal++foreign import javascript safe+ "$1.onmessage =$2;"+ js_onmessage :: WebSocket -> JSVal -> IO ()+++getWebServerNode :: TransIO Node+getWebServerNode = liftIO $ do+ h <- fromJSValUnchecked js_hostname+ p <- fromIntegral <$> (fromJSValUnchecked js_port :: IO Int)+ createNode h p+++hsonmessage ::WebSocket -> (MessageEvent ->IO()) -> IO ()+hsonmessage ws hscb= do+ cb <- makeCallback1 MessageEvent hscb+ js_onmessage ws cb++foreign import javascript safe+ "$1.onopen =$2;"+ js_open :: WebSocket -> JSVal -> IO ()++foreign import javascript safe+ "$1.readyState"+ js_readystate :: WebSocket -> Int++newtype OpenEvent = OpenEvent JSVal deriving Typeable+hsopen :: WebSocket -> (OpenEvent ->IO()) -> IO ()+hsopen ws hscb= do+ cb <- makeCallback1 OpenEvent hscb+ js_open ws cb++makeCallback1 :: (JSVal -> a) -> (a -> IO ()) -> IO JSVal++makeCallback1 f g = do+ Callback cb <- CB.syncCallback1 CB.ContinueAsync (g . f)+ return cb++-- makeCallback :: IO () -> IO ()+makeCallback f = do+ Callback cb <- CB.syncCallback CB.ContinueAsync f+ return cb++++foreign import javascript safe+ "new WebSocket($1)" js_createDefault :: JS.JSString -> IO WebSocket+++#else+++mread conn= do+ cc <- liftIO $ readIORef $ connData conn+ case cc of+ Just (Node2Node _ _ _) -> parallelReadHandler conn+ Just (TLSNode2Node _ ) -> parallelReadHandler conn+ -- the rest of the cases are managed by listenNew+ -- Just (Node2Web sconn ) -> do+ -- ss <- parallel $ receiveData' conn sconn+ -- case ss of+ -- SDone -> empty+ -- SMore s -> do+ -- tr ("WEBSOCKET RECEIVED", s)+ -- setParseString s+ -- --integer+ -- TOD t _ <- liftIO getClockTime+ -- liftIO $ modifyMVar_ (isBlocked conn) $ const $ Just <$> return t+ -- deserialize+ -- where+ -- perform timeouts and cleanup of the server when connections + +receiveData' a b= NWS.receiveData b+-- receiveData' :: Connection -> NWS.Connection -> IO BS.ByteString+-- receiveData' c conn = do+-- msg <- WS.receive conn+-- tr ("RECEIVED",msg)+-- case msg of+-- NWS.DataMessage _ _ _ am -> return $ NWS.fromDataMessage am+-- NWS.ControlMessage cm -> case cm of+-- NWS.Close i closeMsg -> do+-- hasSentClose <- readIORef $ WS.connectionSentClose conn+-- unless hasSentClose $ WS.send conn msg+-- writeIORef (connData c) Nothing+-- cleanConnectionData c+-- empty++-- NWS.Pong _ -> do+-- TOD t _ <- liftIO getClockTime+-- liftIO $ modifyMVar_ (isBlocked c) $ const $ Just <$> return t+-- receiveData' c conn+-- --NWS.connectionOnPong (WS.connectionOptions conn)+-- --NWS.receiveDataMessage conn+-- NWS.Ping pl -> do+-- TOD t _ <- liftIO getClockTime+-- liftIO $ modifyMVar_ (isBlocked c) $ const $ Just <$> return t+-- WS.send conn (NWS.ControlMessage (NWS.Pong pl))+-- receiveData' c conn+++-- --WS.receiveDataMessage conn+{-+mread (Connection _ _ _ (Just (Node2Node _ _ _)) _ _ _ _ _ _ _) = parallelReadHandler -- !> "mread"++mread (Connection _ _ _ (Just (TLSNode2Node _ )) _ _ _ _ _ _ _) = parallelReadHandler+-- parallel $ do+-- s <- recvTLSData ctx+-- return . read' $ BC.unpack s++mread (Connection _ _ _ (Just (Node2Web sconn )) _ _ _ _ _ _ _)= do++ s <- waitEvents $ WS.receiveData sconn+ setParseString s+ integer+ deserialize+{-+ parallel $ do+ s <- WS.receiveData sconn+ return . read' $ BS.unpack s+ !> ("WS MREAD RECEIVED ----<----<------<--------", s)+-}++-}++many' p= p <|> many' p++parallelReadHandler :: Loggable a => Connection -> TransIO (StreamData a)+parallelReadHandler conn= do+ onException $ \(e:: IOError) -> empty+ many' extractPacket+ where+ extractPacket= do+ len <- integer <|> (do s <- getParseBuffer; if BS.null s then empty else error $ show $ ("malformed data received: expected Int, received: ", BS.take 10 s))+ str <- tTake (fromIntegral len)+ tr ("MREAD <-------<-------",str)+ TOD t _ <- liftIO $ getClockTime+ liftIO $ modifyMVar_ (isBlocked conn) $ const $ Just <$> return t++ abduce++ setParseString str+ deserialize+++{-+parallelReadHandler= do+ str <- giveParseString :: TransIO BS.ByteString++ r <- choose $ readStream str++ return r+ !> ("parallel read handler read", r)+ !> "<-------<----------<--------<----------"+ where+ readStream :: (Typeable a, Read a) => BS.ByteString -> [StreamData a]+ readStream s= readStream1 $ BS.unpack s+ where++ readStream1 s=+ let [(x,r)] = reads s+ in x : readStream1 r++-}++getWebServerNode :: TransIO Node+getWebServerNode = getNodes >>= return . Prelude.head+#endif++++mclose :: MonadIO m => Connection -> m ()++#ifndef ghcjs_HOST_OS++mclose con= do+ + --c <- liftIO $ readIORef $ connData con+ c <- liftIO $ atomicModifyIORef (connData con) $ \c -> (Nothing,c)++ case c of+ Just (TLSNode2Node ctx) -> liftIO $ withMVar (isBlocked con) $ const $ liftIO $ tlsClose ctx+ Just (Node2Node _ sock _ ) -> liftIO $ withMVar (isBlocked con) $ const $ liftIO $ NS.close sock !> "SOCKET CLOSE"++ Just (Node2Web sconn ) -> liftIO $ WS.sendClose sconn ("closemsg" :: BS.ByteString) !> "WEBSOCkET CLOSE"+ _ -> return()+ cleanConnectionData con+{-+mclose (Connection _ _ _+ (Just (Node2Node _ sock _ )) _ _ _ _ _ _ _)= liftIO $ NS.close sock++mclose (Connection _ _ _+ (Just (Node2Web sconn ))+ _ _ _ _ _ _ _)=+ liftIO $ WS.sendClose sconn ("closemsg" :: BS.ByteString)+-}+#else++mclose con= do+ --c <- liftIO $ readIORef $ connData con+ c <- liftIO $ atomicModifyIORef (connData con) $ \c -> (Nothing,c)++ case c of+ Just (Web2Node sconn)->+ liftIO $ JavaScript.Web.WebSocket.close Nothing Nothing sconn+{-+mclose (Connection _ _ _ (Just (Web2Node sconn)) _ _ blocked _ _ _ _)=+ liftIO $ JavaScript.Web.WebSocket.close Nothing Nothing sconn+-}+#endif++#ifndef ghcjs_HOST_OS+-- connection cookie+rcookie= unsafePerformIO $ newIORef $ BS.pack "cookie1"+#endif++conSection= unsafePerformIO $ newMVar ()+exclusiveCon mx= do+ liftIO $ takeMVar conSection+ r <- mx+ liftIO $ putMVar conSection ()+ return r++-- check for cached connection and return it, otherwise tries to connect with connect1 without cookie check+mconnect' :: Node -> TransIO Connection+mconnect' node'= exclusiveCon $ do + conn <- do + node <- fixNode node'+ nodes <- getNodes++ let fnode = filter (==node) nodes+ case fnode of+ [] -> mconnect1 node -- !> "NO NODE"+ (node'@(Node _ _ pool _):_) -> do+ plist <- liftIO $ readMVar $ fromJust pool+ case plist of -- !> ("length", length plist,nodePort node) of+ (handle:_) -> do++ c <- liftIO $ readIORef $ connData handle+ if isNothing c -- was closed by timeout+ then mconnect1 node+ else return handle+ !> ("REUSED!", nodeHost node, nodePort node)+ _ -> do+ delNodes [node]+ r <- mconnect1 node+ tr "after mconnect1"+ return r+ -- ctx <- liftIO $ readIORef $ istream conn+ -- modify $ \s -> s{parseContext= ctx}+ -- liftIO $ print "SET PARSECONTEXT"+ setState conn+ + return conn++++#ifndef ghcjs_HOST_OS+-- effective connect trough different methods+mconnect1 (node@(Node host port _ services ))= do++ return () !> ("MCONNECT1",host,port,isTLSIncluded)+ {-+ onException $ \(ConnectionError msg node) -> do+ liftIO $ do+ putStr msg+ putStr " connecting "+ print node+ continue+ empty+ -}+ let types=mapMaybe (lookup "type") services -- need to look in all services+ needTLS <- if "HTTP" `elem` types then return False+ else if "HTTPS" `elem` types then+ if not isTLSIncluded then error "no 'initTLS'. This is necessary for https connections. Please include it: main= do{ initTLS; keep ...."+ else return True+ else return isTLSIncluded+ -- case lookup "type" services of+ -- Just "HTTP" -> return False;+ -- Just "HTTPS" -> + -- if not isTLSIncluded then error "no 'initTLS'. This is necessary for https connections. Please include it: main= do{ initTLS; keep ...."+ -- else return True+ -- _ -> return isTLSIncluded + + tr ("NEED TLS",needTLS)+ (conn,parseContext) <- checkSelf node <|>+ timeout 10000000 (connectNode2Node host port needTLS) <|>+ timeout 1000000 (connectWebSockets host port needTLS) <|>+ timeout 1000000 (checkRelay needTLS) <|>+ (throw $ ConnectionError "no connection" node)++ setState conn+ modify $ \s -> s{execMode=Serial,parseContext= parseContext}++ -- "write node connected in the connection"+ liftIO $ writeIORef (remoteNode conn) $ Just node+ -- "write connection in the node"+ liftIO $ modifyMVar_ (fromJust $ connection node) . const $ return [conn]++ addNodes [node]+ + return conn+++ where+ checkSelf node= do+ tr "CHECKSELF"+ node' <- getMyNodeMaybe + guard $ isJust (connection node')+ v <- liftIO $ readMVar (fromJust $ connection node') -- to force connection in case of calling a service of itself+ tr "IN CHECKSELF"+ if node /= node' || null v+ then empty+ else do+ conn<- case connection node of+ Nothing -> error "checkSelf error"+ Just ref -> do+ rnode <- liftIO $ newIORef node'+ cdata <- liftIO $ newIORef $ Just Self+ conn <- defConnection >>= \c -> return c{myNode= rnode, connData= cdata}+ liftIO $ withMVar ref $ const $ return [conn]+ return conn++ return (conn, noParseContext)++ timeout t proc= do+ r <- collect' 1 t proc+ case r of+ [] -> empty !> "TIMEOUT EMPTY"+ mr:_ -> case mr of+ Nothing -> throw $ ConnectionError "Bad cookie" node+ Just r -> return r++ checkRelay needTLS= do+ case lookup "relay" $ map head (nodeServices node) of+ Nothing -> empty -- !> "NO RELAY"+ Just relayinfo -> do+ let (h,p)= read relayinfo+ connectWebSockets1 h p ("/relay/" ++ h ++ "/" ++ show p ++ "/") needTLS+++ connectSockTLS host port needTLS= do+ return () !> "connectSockTLS"++ let size=8192+ c@Connection{myNode=my,connData=rcdata} <- getSData <|> defConnection + tr "BEFORE HANDSHAKE"++ sock <- liftIO $ connectTo' size host $ PortNumber $ fromIntegral port+ let cdata= (Node2Node u sock (error $ "addr: outgoing connection"))+ cdata' <- liftIO $ readIORef rcdata++ --input <- liftIO $ SBSL.getContents sock+ -- let pcontext= ParseContext (do mclose c; return SDone) input (unsafePerformIO $ newIORef False)++ pcontext <- makeParseContext $ SBSL.recv sock 4096+ + conn' <- if isNothing cdata' -- lost connection, reconnect+ + then do + liftIO $ writeIORef rcdata $ Just cdata+ liftIO $ writeIORef (istream c) pcontext+ return c !> "RECONNECT"+ else do+ c <- defConnection+ rcdata' <- liftIO $ newIORef $ Just cdata+ liftIO $ writeIORef (istream c) pcontext++ return c{myNode=my,connData= rcdata'} !> "CONNECT"++ setData conn'++ --modify $ \s ->s{parseContext=ParseContext (do NS.close sock ; return SDone) input} --throw $ ConnectionError "connection closed" node) input}+ modify $ \s ->s{execMode=Serial,parseContext=pcontext}+ --modify $ \s ->s{execMode=Serial,parseContext=ParseContext (SMore . BL.fromStrict <$> recv sock 1000) mempty}++ when (isTLSIncluded && needTLS) $ maybeClientTLSHandshake host sock mempty++++ connectNode2Node host port needTLS= do+ -- onException $ \(e :: SomeException) -> empty+ tr "NODE 2 NODE"+ mproxy <- getHTTProxyParams needTLS+ let (upass,h',p) = case (mproxy) of+ Just p -> p+ _ -> ("",BS.pack host,port)+ h= BS.unpack h'++ if (isLocal host || h == host && p == port) then+ connectSockTLS h p needTLS++ + else do+ let connect = + "CONNECT "<> pk host <> ":" <> pk (show port) <> " HTTP/1.1\r\n" <>+ "Host: "<> pk host <> ":" <> BS.pack (show port) <> "\r\n" <> ++ "User-Agent: transient\r\n" <>+ (if BS.null upass then "" else "Proxy-Authorization: Basic " <> (B64.encode upass)<> "\r\n") <>+ "Proxy-Connection: Keep-Alive\r\n" <>+ "\r\n" + tr connect+ connectSockTLS h p False+ conn <- getSData <|> error "mconnect: no connection data"++ sendRaw conn $ connect+ first@(vers,code,_) <- getFirstLineResp -- tTakeUntilToken (BS.pack "\r\n\r\n")+ tr ("PROXY RESPONSE=",first) + guard (BC.head code== '2') + <|> do + headers <- getHeaders+ Raw body <- parseBody headers+ error $ show (headers,body) -- decode the body and print++ when (isTLSIncluded && needTLS) $ do+ Just(Node2Node{socket=sock}) <- liftIO $ readIORef $ connData conn+ maybeClientTLSHandshake h sock mempty++ conn <- getSData <|> error "mconnect: no connection data"++ --mynode <- getMyNode+ parseContext <- gets parseContext + return $ Just(conn,parseContext)+ ++ connectWebSockets host port needTLS= connectWebSockets1 host port "/" needTLS+ connectWebSockets1 host port verb needTLS= do+ -- onException $ \(e :: SomeException) -> empty+ tr "WEBSOCKETS"+ connectSockTLS host port needTLS -- a new connection++ never <- liftIO $ newEmptyMVar :: TransIO (MVar ())+ conn <- getSData <|> error "connectWebSockets: no connection"++ stream <- liftIO $ makeWSStreamFromConn conn+ co <- liftIO $ readIORef rcookie+ let hostport= host++(':': show port)+ headers= [("cookie", "cookie=" <> BS.toStrict co)] -- if verb =="/" then [("Host",fromString hostport)] else []+++ onException $ \(NWS.CloseRequest code msg) -> do+ conn <- getSData+ cleanConnectionData conn+ -- throw $ ConnectionError (BS.unpack msg) node+ empty++ wscon <- react (NWS.runClientWithStream stream hostport verb+ WS.defaultConnectionOptions headers)+ (takeMVar never)+++ msg <- liftIO $ WS.receiveData wscon+ tr "WS RECEIVED"+ case msg of+++ ("OK" :: BS.ByteString) -> do+ tr "return connectWebSockets"+ cdata <- liftIO $ newIORef $ Just $ (Node2Web wscon)+ return $ Just (conn{connData= cdata}, noParseContext)++ _ -> do tr "RECEIVED CLOSE"; liftIO $ WS.sendClose wscon ("" ::BS.ByteString); return Nothing++isLocal:: String ->Bool+isLocal host= host=="localhost" || + (or $ map (flip isPrefixOf host) + ["0.0","10.","100", "127", "169", "172", "192", "198", "203"]) ||+ isAlphaNum (head host) && not ('.' `elem` host) -- is not a host address with dot inside: www.host.com++-- >>> isLocal "titan"+-- True+--+++makeParseContext rec= liftIO $ do+ done <- newIORef False+ let receive= liftIO $ do + d <- readIORef done+ if d then return SDone + + else (do+ r<- rec+ if BS.null r then liftIO $ do writeIORef done True; return SDone+ else return $ SMore r)++ `catch` \(SomeException e) -> do liftIO $ writeIORef done True+ putStr "Parse: "+ print e+ return SDone++ return $ ParseContext receive mempty done++ +#else+ +mconnect1 (node@(Node host port (Just pool) _))= do+ conn <- getSData <|> error "connect: listen not set for this node"+ if nodeHost node== "webnode"+ then do+ liftIO $ writeIORef (connData conn) $ Just Self+ return conn+ else do+ ws <- connectToWS host $ PortNumber $ fromIntegral port+-- !> "CONNECTWS"+ liftIO $ writeIORef (connData conn) $ Just (Web2Node ws)++-- !> ("websocker CONNECION")+ let parseContext =+ ParseContext (error "parsecontext not available in the browser")+ "" (unsafePerformIO $ newIORef False)++ chs <- liftIO $ newIORef M.empty+ let conn'= conn{closChildren= chs}+ liftIO $ modifyMVar_ pool $ \plist -> return $ conn':plist+ + return conn'+#endif++u= undefined++data ConnectionError= ConnectionError String Node deriving (Show , Read)++instance Exception ConnectionError++-- check for cached connect, if not, it connects and check cookie with mconnect2+mconnect node'= do+ node <- fixNode node'+ nodes <- getNodes++ let fnode = filter (==node) nodes+ case fnode of+ [] -> mconnect2 node -- !> "NO NODE"+ [node'@(Node _ _ pool _)] -> do+ plist <- liftIO $ readMVar $ fromJust pool+ case plist of -- !> ("length", length plist,nodePort node) of+ (handle:_) -> do++ c <- liftIO $ readIORef $ connData handle+ if isNothing c -- was closed by timeout+ then mconnect2 node+ else return handle+ -- !> ("REUSED!", node)+ _ -> do+ delNodes [node]+ mconnect2 node+ where+ -- connect and check for connection cookie among nodes+ mconnect2 node= do+ conn <- mconnect1 node+-- `catcht` \(e :: SomeException) -> empty+ cd <- liftIO $ readIORef $ connData conn+ case cd of+#ifndef ghcjs_HOST_OS+ Just Self -> return()+ Just (TLSNode2Node _ ) -> do+ checkCookie conn+ watchConnection conn node+ Just (Node2Node _ _ _) -> do+ checkCookie conn+ watchConnection conn node+#endif+ _ -> watchConnection conn node+ return conn+#ifndef ghcjs_HOST_OS + checkCookie conn= do + cookie <- liftIO $ readIORef rcookie+ mynode <- getMyNode+ sendRaw conn $ "CLOS " <> cookie <> --" b \r\nField: value\r\n\r\n" -- TODO put it standard: Set-Cookie:...+ " b \r\nHost: " <> BS.pack (nodeHost mynode) <> "\r\nPort: " <> BS.pack (show $ nodePort mynode) <> "\r\n\r\n"++ r <- liftIO $ readFrom conn++ case r of+ "OK" -> return ()+ + _ -> do+ let Connection{connData=rcdata}= conn+ cdata <- liftIO $ readIORef rcdata+ case cdata of+ Just(Node2Node _ s _) -> liftIO $ NS.close s -- since the HTTP firewall closes the connection+ Just(TLSNode2Node c) -> liftIO $ tlsClose c+ empty+#endif++ watchConnection conn node= do+ liftIO $ atomicModifyIORef connectionList $ \m -> (conn:m,())++ parseContext <- gets parseContext -- getSData <|> error "NO PARSE CONTEXT"+ :: TransIO ParseContext+ chs <- liftIO $ newIORef M.empty+ --whls <- liftIO $ newIORef []+ let conn'= conn{closChildren= chs} --, wormholes= whls}+ -- liftIO $ modifyMVar_ (fromJust pool) $ \plist -> do+ -- if not (null plist) then print "DUPLICATE" else return ()+ -- return $ conn':plist -- !> (node,"ADDED TO POOL")++ -- tell listenResponses to watch incoming responses+ putMailbox ((conn',parseContext,node) :: (Connection,ParseContext,Node))+ liftIO $ threadDelay 100000 -- give time to initialize listenResponses++++++#ifndef ghcjs_HOST_OS+close1 sock= do++ NS.setSocketOption sock NS.Linger 0+ NS.close sock++connectTo' bufSize hostname (PortNumber port) = do+ proto <- BSD.getProtocolNumber "tcp"+ bracketOnError+ (NS.socket NS.AF_INET NS.Stream proto)+ (NS.close) -- only done if there's an error+ (\sock -> do+ NS.setSocketOption sock NS.RecvBuffer bufSize+ NS.setSocketOption sock NS.SendBuffer bufSize++++-- NS.setSocketOption sock NS.SendTimeOut 1000000 !> ("CONNECT",port)++ he <- BSD.getHostByName hostname++ NS.connect sock (NS.SockAddrInet port (BSD.hostAddress he))++ return sock)++#else+connectToWS h (PortNumber p) = do+ protocol <- liftIO $ fromJSValUnchecked js_protocol+ pathname <- liftIO $ fromJSValUnchecked js_pathname+ tr ("PAHT",pathname)+ let ps = case (protocol :: JS.JSString)of "http:" -> "ws://"; "https:" -> "wss://"+ wsOpen $ JS.pack $ ps++ h++ ":"++ show p ++ pathname+#endif+++-- last usage+ blocking semantics for sending+type Blocked= MVar (Maybe Integer)+type BuffSize = Int+data ConnectionData=+#ifndef ghcjs_HOST_OS+ Node2Node{port :: PortID+ ,socket ::Socket+ ,sockAddr :: NS.SockAddr+ }+ | TLSNode2Node{tlscontext :: SData}+ | HTTPS2Node{tlscontext :: SData}+ | Node2Web{webSocket :: WS.Connection}+ | HTTP2Node{port :: PortID+ ,socket ::Socket+ ,sockAddr :: NS.SockAddr}+ | Self++#else+ Self+ | Web2Node{webSocket :: WebSocket}+#endif+ -- deriving (Eq,Ord)++++data Connection= Connection{idConn :: Int+ ,myNode :: IORef Node+ ,remoteNode :: IORef (Maybe Node)+ ,connData :: IORef (Maybe ConnectionData)+ ,istream :: IORef ParseContext+ ,bufferSize :: BuffSize+ -- multiple wormhole/teleport use the same connection concurrently+ ,isBlocked :: Blocked+ ,calling :: Bool+ ,synchronous :: Bool+ -- local localClosures with his continuation and a blocking MVar+ -- another MVar with the children created by the closure+ -- also has the id of the remote closure connected with+ ,localClosures :: MVar (M.Map IdClosure (MVar[EventF],IdClosure, MVar (),EventF))++ -- for each remote closure that points to local closure 0,+ -- a new container of child processes+ -- in order to treat them separately+ -- so that 'killChilds' do not kill unrelated processes+ -- used by `single` and `unique`+ ,closChildren :: IORef (M.Map Int EventF)}++ deriving Typeable++connectionList :: IORef [Connection]+connectionList= unsafePerformIO $ newIORef []++++++defConnection :: TransIO Connection++noParseContext= let err= error "parseContext not set" in+ ParseContext err err err ++-- #ifndef ghcjs_HOST_OS+defConnection = do+ idc <- genGlobalId+ liftIO $ do+ my <- createNode "localhost" 0 >>= newIORef+ x <- newMVar Nothing+ y <- newMVar M.empty+ ref <- newIORef Nothing+ z <- newIORef M.empty+ noconn <- newIORef Nothing+ np <- newIORef noParseContext+ -- whls <- newIORef []+ return $ Connection idc my ref noconn np 8192+ --(error "defConnection: accessing network events out of listen")+ x False False y z -- whls+++#ifndef ghcjs_HOST_OS++setBuffSize :: Int -> TransIO ()+setBuffSize size= do+ conn<- getData `onNothing` (defConnection !> "DEFF3")+ setData $ conn{bufferSize= size}+++getBuffSize=+ (do getSData >>= return . bufferSize) <|> return 8192+++-- | Setup the node to start listening for incoming connections.+--+listen :: Node -> Cloud ()+listen (node@(Node _ port _ _ )) = onAll $ do+ labelState "listen"++ {-+ st <- get+ onException $ \(e :: SomeException) -> do+ case fromException e of+ Just (CloudException _ _ _) -> return()+ _ -> do+ cutExceptions+ liftIO $ print "EXCEPTION: KILLING"+ topState >>= showThreads+ -- liftIO $ killBranch' st+ -- Closure closRemote <- getData `onNothing` error "teleport: no closRemote"+ -- conn <- getData `onNothing` error "reportBack: No connection defined: use wormhole"+ -- liftIO $ putStrLn "Closing connection"+ -- mclose conn+ -- msend conn $ SError $ toException $ ErrorCall $ show $ show $ CloudException node closRemote $ show e+ empty+ -}+ -- ex <- exceptionPoint :: TransIO (BackPoint SomeException)+ -- setData ex+ onException $ \(ConnectionError msg node) -> empty++ --addThreads 2+ fork connectionTimeouts+ fork loopClosures++ setData $ Log{recover=False, buildLog= mempty, fulLog= mempty, lengthFull= 0, hashClosure= 0}++ conn' <- getSData <|> defConnection+ chs <- liftIO $ newIORef M.empty+ cdata <- liftIO $ newIORef $ Just Self+ let conn= conn'{connData=cdata,closChildren=chs}+ pool <- liftIO $ newMVar [conn]++ let node'= node{connection=Just pool}+ liftIO $ writeIORef (myNode conn) node'+ setData conn++ liftIO $ modifyMVar_ (fromJust $ connection node') $ const $ return [conn]++ addNodes [node']+ setRState(JobGroup M.empty) --used by resetRemote++ ex <- exceptionPoint :: TransIO (BackPoint SomeException)+ setData ex++ mlog <- listenNew (fromIntegral port) conn <|> listenResponses :: TransIO (StreamData NodeMSG)+ execLog mlog+ --showNext "after listen" 10+ tr "END LISTEN"++-- listen incoming requests++listenNew port conn'= do+ sock <- liftIO $ listenOn $ PortNumber port++ liftIO $ do+ let bufSize= bufferSize conn'+ NS.setSocketOption sock NS.RecvBuffer bufSize+ NS.setSocketOption sock NS.SendBuffer bufSize++ -- wait for connections. One thread per connection+ liftIO $ do putStr "Connected to port: "; print port+ (sock,addr) <- waitEvents $ NS.accept sock++ chs <- liftIO $ newIORef M.empty+-- case addr of+-- NS.SockAddrInet port host -> liftIO $ print("connection from", port, host)+-- NS.SockAddrInet6 a b c d -> liftIO $ print("connection from", a, b,c,d)+ noNode <- liftIO $ newIORef Nothing+ id1 <- genGlobalId+ let conn= conn'{idConn=id1,closChildren=chs, remoteNode= noNode}++ --liftIO $ atomicModifyIORef connectionList $ \m -> (conn: m,()) -- TODO++ input <- liftIO $ SBSL.getContents sock+ --tr "SOME INPUT"+ -- cutExceptions++ -- onException $ \(e :: IOException) ->+ -- when (ioeGetLocation e=="Network.Socket.recvBuf") $ do+ -- liftIO $ putStr "listen: " >> print e++ -- let Connection{remoteNode=rnode,localClosures=localClosures,closChildren= rmap} = conn+ -- mnode <- liftIO $ readIORef rnode+ -- case mnode of+ -- Nothing -> return ()+ -- Just node -> do+ -- liftIO $ putStr "removing1 node: " >> print node+ -- nodes <- getNodes+ -- setNodes $ nodes \\ [node]+ -- liftIO $ do+ -- modifyMVar_ localClosures $ const $ return M.empty+ -- writeIORef rmap M.empty+ -- -- topState >>= showThreads++ -- killBranch+++ let nod = unsafePerformIO $ liftIO $ createNode "incoming connection" 0 in+ modify $ \s -> s{execMode=Serial,parseContext= (ParseContext + (liftIO $ NS.close sock >> throw (ConnectionError "connection closed" nod)) + input (unsafePerformIO $ newIORef False)+ ::ParseContext )}+ cdata <- liftIO $ newIORef $ Just (Node2Node (PortNumber port) sock addr)+ let conn'= conn{connData=cdata}+ setState conn'+ liftIO $ atomicModifyIORef connectionList $ \m -> (conn': m,()) -- TODO++ maybeTLSServerHandshake sock input+ -- tr "AFTER HANDSHAKE"++ firstLine@(method, uri, vers) <- getFirstLine+ headers <- getHeaders+ + setState $ HTTPHeaders firstLine headers+ -- tr ("HEADERS", headers)+ -- string "\r\n\r\n"+ -- tr (method, uri,vers)+ case (method, uri) of++ ("CLOS", hisCookie) -> do++ conn <- getSData+ tr "CONNECTING"+ let host = BC.unpack $ fromMaybe (error "no host in header")$ lookup "Host" headers+ port = read $ BC.unpack $ fromMaybe (error "no port in header")$ lookup "Port" headers+ remNode' <- liftIO $ createNode host port+++ rc <- liftIO $ newMVar [conn]+ let remNode= remNode'{connection= Just rc}+ liftIO $ writeIORef (remoteNode conn) $ Just remNode+ tr ("ADDED NODE", remNode)+ addNodes [remNode]+ myCookie <- liftIO $ readIORef rcookie+ if BS.toStrict myCookie /= hisCookie+ then do+ sendRaw conn "NOK"++ mclose conn+ error "connection attempt with bad cookie"+ else do++ sendRaw conn "OK" -- !> "CLOS detected"+ --async (return (SMore $ ClosureData 0 0[Exec])) <|> mread conn++ mread conn++ _ -> do+ -- it is a HTTP request+ -- process the current request in his own thread and then (<|>) any other request that arrive in the same connection+ cutBody method headers <|> many' cutHTTPRequest++ HTTPHeaders (method,uri,vers) headers <- getState <|> error "HTTP: no headers?"++ let uri'= BC.tail $ uriPath uri !> uriPath uri+ tr ("uri'", uri')++ case BC.span (/= '/') uri' of++ ("api",_) -> do+ -- if "api" `BC.isPrefixOf` uri'+ --then do++ --log <- return $ Exec:Exec: (Var $ IDyns $ up method):(map (Var . IDyns ) $ split $ BC.unpack $ BC.drop 4 uri')++ let log= exec <> lazyByteString method <> byteString "/" <> byteString (BC.drop 4 uri')++++ maybeSetHost headers+ tr ("HEADERS", headers)+ str <- giveParseString <|> error "no api data"+ if lookup "Transfer-Encoding" headers == Just "chunked" then error $ "chunked not supported" else do++ len <- (read <$> BC.unpack+ <$> (Transient $ return (lookup "Content-Length" headers)))+ <|> return 0+ log' <- case lookup "Content-Type" headers of++ Just "application/json" -> do++ let toDecode= BS.take len str+ -- tr ("TO DECODE", log <> lazyByteString toDecode)++ setParseString $ BS.take len str+ return $ log <> "/" <> lazyByteString toDecode -- [(Var $ IDynamic json)] -- TODO hande this serialization++ Just "application/x-www-form-urlencoded" -> do++ tr ("POST HEADERS=", BS.take len str)++ setParseString $ BS.take len str+ --postParams <- parsePostUrlEncoded <|> return []+ return $ log <> lazyByteString ( BS.take len str) -- [(Var . IDynamic $ postParams)] TODO: solve deserialization++ Just x -> do+ tr ("POST HEADERS=", BS.take len str)+ let str= BS.take len str+ return $ log <> lazyByteString str -- ++ [Var $ IDynamic str]++ _ -> return $ log++ setParseString $ toLazyByteString log'+ return $ SMore $ ClosureData 0 0 log' !> ("APIIIII", log')++ --else if "relay" `BC.isPrefixOf` uri' then proxy sock method vers uri'+ ("relay",_) -> proxy sock method vers uri'++ (h,rest) -> do+ if BC.null rest || h== "file" then do+ --headers <- getHeaders+ maybeSetHost headers+ let uri= if BC.null h || BC.null rest then uri' else BC.tail rest+ tr (method,uri)+ -- stay serving pages until a websocket request is received+ servePages (method, uri, headers)++ -- when servePages finish, is because a websocket request has arrived+ conn <- getSData+ sconn <- makeWebsocketConnection conn uri headers++ -- websockets mode+ -- para qué reiniciarlo todo????+ rem <- liftIO $ newIORef Nothing+ chs <- liftIO $ newIORef M.empty+ cls <- liftIO $ newMVar M.empty+ cme <- liftIO $ newIORef M.empty+ cdata <- liftIO $ newIORef $ Just (Node2Web sconn)+ let conn'= conn{connData= cdata+ , closChildren=chs,localClosures=cls, remoteNode=rem} --,comEvent=cme}+ setState conn' !> "WEBSOCKETS CONNECTION"++ co <- liftIO $ readIORef rcookie++ let receivedCookie= lookup "cookie" headers+++ tr ("cookie", receivedCookie)+ rcookie <- case receivedCookie of+ Just str-> Just <$> do+ withParseString (BS.fromStrict str) $ do+ tDropUntilToken "cookie="+ tTakeWhile (not . isSpace)+ Nothing -> return Nothing+ tr ("RCOOKIE", rcookie)+ if rcookie /= Nothing && rcookie /= Just co + then do+ node <- getMyNode+ --let msg= SError $ toException $ ErrorCall $ show $ show $ CloudException node 0 $ show $ ConnectionError "bad cookie" node+ tr "SENDINg"+++ liftIO $ WS.sendClose sconn ("Bad Cookie" :: BS.ByteString) !> "SendClose Bad cookie"+ empty++ else do++ liftIO $ WS.sendTextData sconn ("OK" :: BS.ByteString)++++ -- a void message is sent to the application signaling the beginning of a connection+ -- async (return (SMore $ ClosureData 0 0[Exec])) <|> do++ + tr "WEBSOCKET"+ -- onException $ \(e :: SomeException) -> do+ -- liftIO $ putStr "listen websocket:" >> print e+ -- -- liftIO $ mclose conn'+ -- -- killBranch+ -- -- empty+++ s <- waitEvents $ receiveData' conn' sconn :: TransIO BS.ByteString+ setParseString s+ tr ("WEBSOCKET RECEIVED <-----------",s)+ -- integer+ deserialize + -- a void message is sent to the application signaling the beginning of a connection+ <|> (return $ SMore (ClosureData 0 0 (exec <> lazyByteString s))) + else do+ let uriparsed= BS.pack $ unEscapeString $ BC.unpack uri'+ setParseString uriparsed !> ("uriparsed",uriparsed)+ remoteClosure <- deserialize :: TransIO Int+ tChar '/'+ thisClosure <- deserialize :: TransIO Int+ tChar '/'+ --cdata <- liftIO $ newIORef $ Just (HTTP2Node (PortNumber port) sock addr)+ conn <- getSData+ liftIO $ atomicModifyIORef' (connData conn) $ \cdata -> case cdata of+ Just(Node2Node port sock addr) -> (Just $ HTTP2Node port sock addr,())+ Just(TLSNode2Node ctx) -> (Just $ HTTPS2Node ctx,())++ --setState conn{connData=cdata}+ s <- giveParseString+ cook <- liftIO $ readIORef rcookie++ liftIO $ SBSL.sendAll sock $ "HTTP/1.0 200 OK\r\nContent-Type: text/plain\r\n\r\n"+ return $ SMore $ ClosureData remoteClosure thisClosure $ lazyByteString s++++++ where+ cutHTTPRequest = do+ first@(method,_,_) <- getFirstLine+ -- tr ("after getfirstLine", method, uri, vers)+ headers <- getHeaders+ setState $ HTTPHeaders first headers+ cutBody method headers++++ cutBody method headers= do+ if method == "POST" then+ case fmap (read . BC.unpack) $ lookup "Content-Length" headers of+ Nothing -> return () -- most likely chunked encoding+ Just len -> do+ str <- tTake (fromIntegral len)+ abduce+ setParseString str+ else abduce+++ uriPath = BC.dropWhile (/= '/')+ split []= []+ split ('/':r)= split r+ split s=+ let (h,t) = Prelude.span (/= '/') s+ in h: split t++ -- reverse proxy for urls that look like http://host:port/relay/otherhost/otherport/+ proxy sclient method vers uri' = do+ let (host:port:_)= split $ BC.unpack $ BC.drop 6 uri'+ tr ("RELAY TO",host, port)+ --liftIO $ threadDelay 1000000+ sserver <- liftIO $ connectTo' 4096 host $ PortNumber $ fromIntegral $ read port+ tr "CONNECTED"+ rawHeaders <- getRawHeaders+ tr ("RAWHEADERS",rawHeaders)+ let uri= BS.fromStrict $ let d x= BC.tail $ BC.dropWhile (/= '/') x in d . d $ d uri'++ let sent= method <> BS.pack " /"+ <> uri+ <> BS.cons ' ' vers+ <> BS.pack "\r\n"+ <> rawHeaders <> BS.pack "\r\n\r\n"+ tr ("SENT",sent)+ liftIO $ SBSL.send sserver sent+ -- Connection{connData=Just (Node2Node _ sclient _)} <- getState <|> error "proxy: no connection"+++ (send sclient sserver <|> send sserver sclient)+ `catcht` \(e:: SomeException ) -> liftIO $ do+ putStr "Proxy: " >> print e+ NS.close sserver+ NS.close sclient+ empty++ empty+ where+ send f t= async $ mapData f t+ mapData from to = do+ content <- recv from 4096+ tr (" proxy received ", content)+ if not $ BC.null content+ then sendAll to content >> mapData from to+ else finish+ where+ finish= NS.close from >> NS.close to+ -- throw $ Finish "finish"+++ maybeSetHost headers= do+ setHost <- liftIO $ readIORef rsetHost+ when setHost $ do++ mnode <- liftIO $ do+ let mhost= lookup "Host" headers+ case mhost of+ Nothing -> return Nothing+ Just host -> atomically $ do+ -- set the first node (local node) as is called from outside+ nodes <- readTVar nodeList+ let (host1,port)= BC.span (/= ':') host+ hostnode= (Prelude.head nodes){nodeHost= BC.unpack host1+ ,nodePort= if BC.null port then 80+ else read $ BC.unpack $ BC.tail port}+ writeTVar nodeList $ hostnode : Prelude.tail nodes+ return $ Just hostnode -- !> (host1,port)++ when (isJust mnode) $ do+ conn <- getState+ liftIO $ writeIORef (myNode conn) $fromJust mnode+ liftIO $ writeIORef rsetHost False -- !> "HOSt SET"++{-#NOINLINE rsetHost #-}+rsetHost= unsafePerformIO $ newIORef True++++--instance Read PortNumber where+-- readsPrec n str= let [(n,s)]= readsPrec n str in [(fromIntegral n,s)]+++--deriving instance Read PortID+--deriving instance Typeable PortID++-- | filter out HTTP requests+noHTTP= onAll $ do+ conn <- getState+ cdata <- liftIO $ readIORef $ connData conn+ case cdata of+ Just (HTTPS2Node ctx) -> do+ liftIO $ sendTLSData ctx $ "HTTP/1.1 403 Forbidden\r\nConnection: close\r\nContent-Length: 11\r\n\r\nForbidden\r\n"+ liftIO $ tlsClose ctx+ Just (HTTP2Node _ sock _) -> do+ liftIO $ SBSL.sendAll sock $ "HTTP/1.1 403 Forbidden\r\nConnection: close\r\nContent-Length: 11\r\n\r\nForbidden\r\n"+ liftIO $ NS.close sock++ empty+ _ -> return ()++{-+-- filter out WebSockets connections(usually coming from a web node)+noWebSockets= onAll $ do+ conn <- getState+ cdata <- liftIO $ readIORef $ connData conn+ case cdata of+ Just (Web2Node _) -> empty+ _ -> return()+-}+#endif++++listenResponses :: Loggable a => TransIO (StreamData a)+listenResponses= do+ labelState "listen responses"++ (conn, parsecontext, node) <- getMailbox :: TransIO (Connection,ParseContext,Node)+ labelState . fromString $ "listen from: "++ show node+ setData conn+ tr ("CONNECTION RECEIVED","listen from: "++ show node)+ modify $ \s-> s{execMode=Serial,parseContext = parsecontext}++ -- cutExceptions+ -- onException $ \(e:: SomeException) -> do+-- liftIO $ putStr "ListenResponses: " >> print e+-- liftIO $ putStr "removing node: " >> print node+-- nodes <- getNodes+-- setNodes $ nodes \\ [node]+-- -- topState >>= showThreads+-- killChilds+-- let Connection{localClosures=localClosures}= conn+-- liftIO $ modifyMVar_ localClosures $ const $ return M.empty+-- empty+++ mread conn+++++type IdClosure= Int++-- The remote closure ids for each node connection+newtype Closure= Closure IdClosure deriving (Read,Show,Typeable)++++++type RemoteClosure= (Node, IdClosure)+++newtype JobGroup= JobGroup (M.Map BC.ByteString RemoteClosure) deriving Typeable++-- | if there is a remote job identified by th string identifier, it stop that job, and set the+-- current remote operation (if any) as the current remote job for this identifier.+-- The purpose is to have a single remote job.+-- to identify the remote job, it should be used after the `wormhole` and before the remote call:+--+-- > r <- wormhole node $ do+-- > stopRemoteJob "streamlog"+-- > atRemote myRemotejob+--+-- So:+--+-- > runAtUnique ident node job= wormhole node $ do stopRemoteJob ident; atRemote job++-- This program receive a stream of "hello" from a second node when the option "hello" is entered in the keyboard+-- If you enter "world", the "hello" stream from the second node+-- will stop and will print an stream of "world" from a third node:+-- Entering "hello" again will stream "hello" again from the second node and so on:+++-- > main= keep $ initNode $ inputNodes <|> do+-- >+-- > local $ option "init" "init"+-- > nodes <- local getNodes+-- > r <- proc (nodes !! 1) "hello" <|> proc (nodes !! 2) "world"+-- > localIO $ print r+-- > return ()+-- >+-- > proc node par = do+-- > v <- local $ option par par+-- > runAtUnique "name" node $ local $ do+-- > abduce+-- > r <- threads 0 $ choose $ repeat v+-- > liftIO $ threadDelay 1000000+-- > return r++-- the nodes could be started from the command line as such in different terminals:++-- > program -p start/localhost/8000+-- > program -p start/localhost/8002+-- > program -p start/localhost/8001/add/localhost/8000/n/add/localhost/8002/n/init++-- The third is the one wich has the other two connected and can execute the two options.++stopRemoteJob :: BC.ByteString -> Cloud ()++instance Loggable Closure++stopRemoteJob ident = do+ resetRemote ident+ Closure closr <- local $ getData `onNothing` error "stopRemoteJob: Connection not set, use wormhole"+ tr ("CLOSRRRRRRRR", closr)+ fixClosure+ local $ do+ Closure closr <- getData `onNothing` error "stopRemoteJob: Connection not set, use wormhole"+ conn <- getData `onNothing` error "stopRemoteJob: Connection not set, use wormhole"+ remote <- liftIO $ readIORef $ remoteNode conn+ return (remote,closr) !> ("REMOTE",remote)++ JobGroup map <- getRState <|> return (JobGroup M.empty)+ setRState $ JobGroup $ M.insert ident (fromJust remote,closr) map++++-- |kill the remote job. Usually, before starting a new one.+resetRemote :: BC.ByteString -> Cloud ()+resetRemote ident = do+ mj <- local $ do+ JobGroup map <- getRState <|> return (JobGroup M.empty)+ return $ M.lookup ident map++ when (isJust mj) $ do+ let (remote,closr)= fromJust mj+ --do -- when (closr /= 0) $ do+ runAt remote $ local $ do+ conn@Connection {localClosures=localClosures} <- getData `onNothing` error "Listen: myNode not set"+ mcont <- liftIO $ modifyMVar localClosures $ \map -> return ( M.delete closr map, M.lookup closr map)+ case mcont of+ Nothing -> error $ "closure not found: " ++ show closr+ Just (_,_,_,cont) -> do+ topState >>= showThreads+ liftIO $ killBranch' cont+ return ()++++execLog :: StreamData NodeMSG -> TransIO ()+execLog mlog =Transient $ do+ tr "EXECLOG"+ case mlog of+ SError e -> do+ return() !> ("SERROR",e)+ case fromException e of+ Just (ErrorCall str) -> do++ case read str of+ (e@(CloudException _ closl err)) -> do++ process closl (error "closr: should not be used") (Left e) True+++ SDone -> runTrans(back $ ErrorCall "SDone") >> return Nothing -- TODO remove closure?+ SMore (ClosureData closl closr log) -> process closl closr (Right log) False+ SLast (ClosureData closl closr log) -> process closl closr (Right log) True+ where+ process :: IdClosure -> IdClosure -> (Either CloudException Builder) -> Bool -> StateIO (Maybe ())+ process closl closr mlog deleteClosure= do+ conn@Connection {localClosures=localClosures} <- getData `onNothing` error "Listen: myNode not set"+ if closl== 0 then do++ case mlog of+ Left except -> do+ setData emptyLog+ tr "Exception received from network 1"+ runTrans $ throwt except+ empty+ Right log -> do+ tr ("CLOSURE 0",log)+ setData Log{recover= True, buildLog= mempty, fulLog= log, lengthFull= 0, hashClosure= 0} --Log True [] []++ setState $ Closure closr+ setRState $ DialogInWormholeInitiated True+ -- setParseString $ toLazyByteString log -- not needed it is has the log from the request, still not parsed++ return $ Just () -- !> "executing top level closure"+ else do+ mcont <- liftIO $ modifyMVar localClosures+ $ \map -> return (if deleteClosure then+ M.delete closl map+ else map, M.lookup closl map)+ -- !> ("localClosures=", M.size map)++ case mcont of+ Nothing -> do++ node <- liftIO $ readIORef (remoteNode conn) `onNothing` error "mconnect: no remote node?"+ let e = "request received for non existent closure. Perhaps the connection was closed by timeout and reopened"+ let err= CloudException node closl $ show e+++ throw err+ -- execute the closure+ Just (chs,closLocal, mv,cont) -> do+ when deleteClosure $ do+ -- liftIO $ killChildren chs++ empty -- last message received+++ liftIO $ tryPutMVar mv ()+ void $ liftIO $ runStateT (case mlog of+ Right log -> do+ -- Log _ _ fulLog hashClosure <- getData `onNothing` return (Log True [] [] 0)+ Log{fulLog=fulLog, hashClosure=hashClosure} <- getLog+ -- return() !> ("fullog in execlog", reverse fulLog)++ let nlog= fulLog <> log -- let nlog= reverse log ++ fulLog+ setData $ Log{recover= True, buildLog= mempty, fulLog= nlog, lengthFull=error "lengthFull TODO", hashClosure= hashClosure} -- TODO hashClosure must change?+ setState $ Closure closr+ setRState $ DialogInWormholeInitiated True+ setParseString $ toLazyByteString log+ --restrs <- giveParseString+ --tr ("rs' in execlog =", fmap (BS.take 4) restrs)+ runContinuation cont ()++ Left except -> do+ setData emptyLog+ tr ("Exception received from the network", except)+ runTrans $ throwt except) cont++ return Nothing++#ifdef ghcjs_HOST_OS+listen node = onAll $ do+ addNodes [node]+ setRState(JobGroup M.empty)+ -- ex <- exceptionPoint :: TransIO (BackPoint SomeException)+ -- setData ex++ events <- liftIO $ newIORef M.empty+ rnode <- liftIO $ newIORef node+ conn <- defConnection >>= \c -> return c{myNode=rnode} -- ,comEvent=events}+ liftIO $ atomicModifyIORef connectionList $ \m -> (conn: m,())++ setData conn+ r <- listenResponses+ execLog r++#endif++type Pool= [Connection]+type SKey= String+type SValue= String+type Service= [(SKey, SValue)]++lookup2 key doubleList= + let r= mapMaybe(lookup key ) doubleList+ in if null r then Nothing else Just $ head r++filter2 key doubleList= mapMaybe(lookup key ) doubleList+++--------------------------------------------+++#ifndef ghcjs_HOST_OS+++-- maybeRead line= unsafePerformIO $ do+-- let [(v,left)] = reads line+---- print v+-- (v `seq` return [(v,left)])+-- `catch` (\(e::SomeException) -> do+-- liftIO $ print $ "******readStream ERROR in: "++take 100 line+-- maybeRead left)++{-+readFrom Connection{connData= Just(TLSNode2Node ctx)}= recvTLSData ctx++readFrom Connection{connData= Just(Node2Node _ sock _)} = toStrict <$> loop++readFrom _ = error "readFrom error"+-}++readFrom con= do+ cd <- readIORef $ connData con+ case cd of+ Just(TLSNode2Node ctx) -> recvTLSData ctx+ Just(Node2Node _ sock _) -> BS.toStrict <$> loop sock+ _ -> error "readFrom error"+ where+ bufSize= 4098+ loop sock = loop1+ where+ loop1 :: IO BL.ByteString+ loop1 = unsafeInterleaveIO $ do+ s <- SBS.recv sock bufSize++ if BC.length s < bufSize+ then return $ BLC.Chunk s mempty+ else BLC.Chunk s `liftM` loop1++++-- toStrict= B.concat . BS.toChunks++makeWSStreamFromConn conn= do+ tr "WEBSOCKETS request"+ let rec= readFrom conn+ send= sendRaw conn+ makeStream+ (do+ bs <- rec -- SBS.recv sock 4098+ return $ if BC.null bs then Nothing else Just bs)+ (\mbBl -> case mbBl of+ Nothing -> return ()+ Just bl -> send bl) -- SBS.sendMany sock (BL.toChunks bl) >> return()) -- !!> show ("SOCK RESP",bl)++makeWebsocketConnection conn uri headers= liftIO $ do++ stream <- makeWSStreamFromConn conn+ let+ pc = WS.PendingConnection+ { WS.pendingOptions = WS.defaultConnectionOptions -- {connectionOnPong=xxx}+ , WS.pendingRequest = NWS.RequestHead uri headers False -- RequestHead (BC.pack $ show uri)+ -- (map parseh headers) False+ , WS.pendingOnAccept = \_ -> return ()+ , WS.pendingStream = stream+ }++ sconn <- WS.acceptRequest pc -- !!> "accept request"+ WS.forkPingThread sconn 30+ return sconn++servePages (method,uri, headers) = do+-- return () !> ("HTTP request",method,uri, headers)+ conn <- getSData <|> error " servePageMode: no connection"++ if isWebSocketsReq headers+ then return ()++++ else do++ let file= if BC.null uri then "index.html" else uri++ {- TODO rendering in server+ NEEDED: recodify View to use blaze-html in server. wlink to get path in server+ does file exist?+ if exist, send else do+ store path, execute continuation+ get the rendering+ send trough HTTP+ - put this logic as independent alternative programmer options+ serveFile dirs <|> serveApi apis <|> serveNode nodeCode+ -}+ mcontent <- liftIO $ (Just <$> BL.readFile ( "./static/out.jsexe/"++ BC.unpack file) )+ `catch` (\(e:: SomeException) -> return Nothing)++-- return "Not found file: index.html<br/> please compile with ghcjs<br/> ghcjs program.hs -o static/out")+ case mcontent of+ Just content -> do+ cook <- liftIO $ readIORef rcookie+ liftIO $ sendRaw conn $+ "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\nConnection: close\r\nContent-Length: "+ <> BS.pack (show $ BL.length content) <>"\r\n"+ <> "Set-Cookie:" <> "cookie=" <> cook -- <> "\r\n"+ <>"\r\n\r\n" <> content++ Nothing ->liftIO $ sendRaw conn $ BS.pack $+ "HTTP/1.0 404 Not Found\nContent-Length: 13\r\nConnection: close\r\n\r\nNot Found 404"+ empty+++-- | forward all the result of the Transient computation to the opened connection+api :: TransIO BS.ByteString -> Cloud ()+api w= Cloud $ do+ log <- getLog+ if not $ recover log then empty else do+ HTTPHeaders (_,_,vers) hdrs <- getState <|> error "api: no HTTP headers???"+ let closeit= lookup "Connection" hdrs == Just "close"+ conn <- getState <|> error "api: Need a connection opened with initNode, listen, simpleWebApp"+ let send= sendRaw conn++ r <- w+ tr ("response",r)+ send r++ tr (vers, hdrs)+ when (vers == http10 ||+ BS.isPrefixOf http10 r ||+ lookup "Connection" hdrs == Just "close" ||+ closeInResponse r)+ $ liftIO $ mclose conn+ where+ closeInResponse r=+ let rest= findSubstring "Connection:" r+ rest' = BS.dropWhile (==' ') rest+ in if BS.isPrefixOf "close" rest' then True else False++ where+ findSubstring sub str+ | BS.null str = str+ | BS.isPrefixOf sub str = BS.drop (BS.length sub) str+ | otherwise= findSubstring sub (BS.tail str)++http10= "HTTP/1.0"++++++isWebSocketsReq = not . null+ . filter ( (== mk "Sec-WebSocket-Key") . fst)+++data HTTPMethod= GET | POST deriving (Read,Show,Typeable,Eq)++instance Loggable HTTPMethod++getFirstLine= (,,) <$> getMethod <*> (BS.toStrict <$> getUri) <*> getVers+ where+ getMethod= parseString+ + getUri= parseString+ getVers= parseString++getRawHeaders= dropSpaces >> (withGetParseString $ \s -> return $ scan mempty s)++ where+ scan res str++ | "\r\n\r\n" `BS.isPrefixOf` str = (res, BS.drop 4 str)+ | otherwise= scan ( BS.snoc res $ BS.head str) $ BS.tail str+ -- line= do+ -- dropSpaces+ -- tTakeWhile (not . endline)++type PostParams = [(BS.ByteString, String)]++parsePostUrlEncoded :: TransIO PostParams+parsePostUrlEncoded= do+ dropSpaces+ many $ (,) <$> param <*> value+ where+ param= tTakeWhile' ( /= '=') !> "param"+ value= unEscapeString <$> BS.unpack <$> tTakeWhile' (/= '&' )+++++getHeaders = manyTill paramPair (string "\r\n\r\n") -- !> (method, uri, vers)++ where+++ paramPair= (,) <$> (mk <$> getParam) <*> getParamValue+++ getParam= do+ dropSpaces+ r <- tTakeWhile (\x -> x /= ':' && not (endline x))+ if BS.null r || r=="\r" then empty else anyChar >> return (BS.toStrict r)+ where+ endline c= c== '\r' || c =='\n'++ getParamValue= BS.toStrict <$> ( dropSpaces >> tTakeWhile (\x -> not (endline x)))+ where+ endline c= c== '\r' || c =='\n'++++#endif++++#ifdef ghcjs_HOST_OS+isBrowserInstance= True+api _= empty+#else+-- | Returns 'True' if we are running in the browser.+isBrowserInstance= False++#endif++++++{-# NOINLINE emptyPool #-}+emptyPool :: MonadIO m => m (MVar Pool)+emptyPool= liftIO $ newMVar []+++-- | Create a node from a hostname (or IP address), port number and a list of+-- services.+createNodeServ :: HostName -> Int -> [Service] -> IO Node+createNodeServ h p svs= return $ Node h p Nothing svs+++createNode :: HostName -> Int -> IO Node+createNode h p= createNodeServ h p []++createWebNode :: IO Node+createWebNode= do+ pool <- emptyPool+ port <- randomIO+ return $ Node "webnode" port (Just pool) [[("webnode","")]]+++instance Eq Node where+ Node h p _ _ ==Node h' p' _ _= h==h' && p==p'+++instance Show Node where+ show (Node h p _ servs )= show (h,p, servs)++instance Read Node where+ readsPrec n s=+ let r= readsPrec n s+ in case r of+ [] -> []+ [((h,p,ss),s')] -> [(Node h p Nothing ss ,s')]++++nodeList :: TVar [Node]+nodeList = unsafePerformIO $ newTVarIO []++deriving instance Ord PortID++--myNode :: Int -> DBRef MyNode+--myNode= getDBRef $ key $ MyNode undefined++errorMyNode f= error $ f ++ ": Node not set. initialize it with connect, listen, initNode..."++-- | Return the local node i.e. the node where this computation is running.+getMyNode :: TransIO Node+getMyNode = do+ Connection{myNode= node} <- getSData <|> errorMyNode "getMyNode" :: TransIO Connection+ liftIO $ readIORef node++-- | empty if the node is not set+getMyNodeMaybe= do+ Connection{myNode= node} <- getSData+ liftIO $ readIORef node++-- | Return the list of nodes in the cluster.+getNodes :: MonadIO m => m [Node]+getNodes = liftIO $ atomically $ readTVar nodeList+++-- | get the nodes that have the same service definition that the calling node+getEqualNodes = do+ nodes <- getNodes++ let srv= nodeServices $ Prelude.head nodes+ case srv of+ [] -> return $ filter (null . nodeServices) nodes++ (srv:_) -> return $ filter (\n -> (not $ null $ nodeServices n) && Prelude.head (nodeServices n) == srv ) nodes++getWebNodes :: MonadIO m => m [Node]+getWebNodes = do+ nodes <- getNodes+ return $ filter ( (==) "webnode" . nodeHost) nodes++matchNodes f = do+ nodes <- getNodes+ return $ Prelude.map (\n -> filter f $ nodeServices n) nodes++-- | Add a list of nodes to the list of existing nodes know locally.+-- If the node is already present, It add his services to the already present node+-- services which have the first element equal (usually the "name" field) will be substituted if the match+addNodes :: [Node] -> TransIO ()+addNodes nodes= liftIO $ do+ -- the local node should be the first+ nodes' <- mapM fixNode nodes+ atomically $ mapM_ insert nodes'+ where+ insert node= do+ prevnodes <- readTVar nodeList -- !> ("ADDNODES", nodes)++ let mn = filter(==node) prevnodes++ case mn of+ [] -> do tr "NUEVO NODO"; writeTVar nodeList $ (prevnodes) ++ [node]+++ [n] ->do+ let nservices= nubBy (\s s' -> head s== head s') $ nodeServices node++ nodeServices n+ writeTVar nodeList $ ((prevnodes) \\ [node]) ++ [n{nodeServices=nservices}]++ _ -> error $ "duplicated node: " ++ show node+ + --writeTVar nodeList $ (prevnodes \\ nodes') ++ nodes'++delNodes nodes= liftIO $ atomically $ do+ nodes' <- readTVar nodeList+ writeTVar nodeList $ nodes' \\ nodes++fixNode n= case connection n of+ Nothing -> do+ pool <- emptyPool+ return n{connection= Just pool}+ Just _ -> return n++-- | set the list of nodes+setNodes nodes= liftIO $ do+ nodes' <- mapM fixNode nodes+ atomically $ writeTVar nodeList nodes'+++-- | Shuffle the list of cluster nodes and return the shuffled list.+shuffleNodes :: MonadIO m => m [Node]+shuffleNodes= liftIO . atomically $ do+ nodes <- readTVar nodeList+ let nodes'= Prelude.tail nodes ++ [Prelude.head nodes]+ writeTVar nodeList nodes'+ return nodes'+++--getInterfaces :: TransIO TransIO HostName+--getInterfaces= do+-- host <- logged $ do+-- ifs <- liftIO $ getNetworkInterfaces+-- liftIO $ mapM_ (\(i,n) ->putStrLn $ show i ++ "\t"++ show (ipv4 n) ++ "\t"++name n)$ zip [0..] ifs+-- liftIO $ putStrLn "Select one: "+-- ind <- input ( < length ifs)+-- return $ show . ipv4 $ ifs !! ind+++++-- #ifndef ghcjs_HOST_OS+--instance Read NS.SockAddr where+-- readsPrec _ ('[':s)=+-- let (s',r1)= span (/=']') s+-- [(port,r)]= readsPrec 0 $ tail $ tail r1+-- in [(NS.SockAddrInet6 port 0 (IP.toHostAddress6 $ read s') 0, r)]+-- readsPrec _ s=+-- let (s',r1)= span(/= ':') s+-- [(port,r)]= readsPrec 0 $ tail r1+-- in [(NS.SockAddrInet port (IP.toHostAddress $ read s'),r)]+-- #endif++-- | add this node to the list of know nodes in the remote node connected by a `wormhole`.+-- This is useful when the node is called back by the remote node.+-- In the case of web nodes with webSocket connections, this is the way to add it to the list of+-- known nodes in the server.+addThisNodeToRemote= do+ n <- local getMyNode+ atRemote $ local $ do+ n' <- setConnectionIn n+ addNodes [n']++setConnectionIn node=do+ conn <- getState <|> error "addThisNodeToRemote: connection not found"+ ref <- liftIO $ newMVar [conn]+ return node{connection=Just ref}++-- | Add a node (first parameter) to the cluster using a node that is already+-- part of the cluster (second parameter). The added node starts listening for+-- incoming connections and the rest of the computation is executed on this+-- newly added node.+connect :: Node -> Node -> Cloud ()+#ifndef ghcjs_HOST_OS+connect node remotenode = do+ listen node <|> return ()+ connect' remotenode++++-- | Reconcile the list of nodes in the cluster using a remote node already+-- part of the cluster. Reconciliation end up in each node in the cluster+-- having the same list of nodes.+connect' :: Node -> Cloud ()+connect' remotenode= loggedc $ do+ nodes <- local getNodes+ localIO $ putStr "connecting to: " >> print remotenode++ newNodes <- runAt remotenode $ interchange nodes++ --return () !> "interchange finish"++ -- add the new nodes to the local nodes in all the nodes connected previously++ let toAdd=remotenode:Prelude.tail newNodes+ callNodes' nodes (<>) mempty $ local $ do+ liftIO $ putStr "New nodes: " >> print toAdd !> "NEWNODES"+ addNodes toAdd++ where+ -- receive new nodes and send their own+ interchange nodes=+ do+ newNodes <- local $ do++ conn@Connection{remoteNode=rnode} <- getSData <|>+ error ("connect': need to be connected to a node: use wormhole/connect/listen")+++ -- if is a websockets node, add only this node+ -- let newNodes = case cdata of+ -- Node2Web _ -> [(head nodes){nodeServices=[("relay",show remotenode)]}]+ -- _ -> nodes++ let newNodes= nodes -- map (\n -> n{nodeServices= nodeServices n ++ [("relay",show (remotenode,n))]}) nodes++ callingNode<- fixNode $ Prelude.head newNodes++ liftIO $ writeIORef rnode $ Just callingNode++ liftIO $ modifyMVar_ (fromJust $ connection callingNode) $ const $ return [conn]+++ -- onException $ \(e :: SomeException) -> do+ -- liftIO $ putStr "connect:" >> print e+ -- liftIO $ putStrLn "removing node: " >> print callingNode+ -- -- topState >>= showThreads+ -- nodes <- getNodes+ -- setNodes $ nodes \\ [callingNode]++ return newNodes++ oldNodes <- local $ getNodes+++ mclustered . local $ do+ liftIO $ putStrLn "New nodes: " >> print newNodes++ addNodes newNodes++ localIO $ atomically $ do+ -- set the first node (local node) as is called from outside+-- tr "HOST2 set"+ nodes <- readTVar nodeList+ let nodes'= (Prelude.head nodes){nodeHost=nodeHost remotenode+ ,nodePort=nodePort remotenode}:Prelude.tail nodes+ writeTVar nodeList nodes'+++ return oldNodes++#else+connect _ _= empty+connect' _ = empty+#endif+++#ifndef ghcjs_HOST_OS+------------------------------- HTTP client ---------------+++instance {-# Overlapping #-} Loggable Value where+ serialize= return . lazyByteString =<< encode+ deserialize = decodeIt+ where+ jsElem :: TransIO BS.ByteString -- just delimites the json string, do not parse it+ jsElem= dropSpaces >> (jsonObject <|> array <|> atom)+ atom= elemString+ array= (brackets $ return "[" <> return "{}" <> chainSepBy mappend (return "," <> jsElem) (tChar ',')) <> return "]"+ jsonObject= (braces $ return "{" <> chainMany mappend jsElem) <> return "}"+ elemString= do+ dropSpaces+ tTakeWhile (\c -> c /= '}' && c /= ']' )++++ decodeIt= do+ s <- jsElem+ tr ("decode",s)++ case eitherDecode s !> "DECODE" of+ Right x -> return x+ Left err -> empty++++++data HTTPHeaders= HTTPHeaders (BS.ByteString, B.ByteString, BS.ByteString) [(CI BC.ByteString,BC.ByteString)] deriving Show++++rawHTTP :: Loggable a => Node -> String -> TransIO a+rawHTTP node restmsg = do+ abduce -- is a parallel operation+ tr ("***********************rawHTTP",nodeHost node)+ --sock <- liftIO $ connectTo' 8192 (nodeHost node) (PortNumber $ fromIntegral $ nodePort node)+ mcon <- getData :: TransIO (Maybe Connection)+ c <- do++ c <- mconnect' node+ tr "after mconnect'"+ sendRawRecover c $ BS.pack restmsg++ c <- getState <|> error "rawHTTP: no connection?"+ let blocked= isBlocked c -- TODO: the same flag is used now for sending and receiving+ tr "before blocked"+ liftIO $ takeMVar blocked+ tr "after blocked"+ ctx <- liftIO $ readIORef $ istream c++ liftIO $ writeIORef (done ctx) False+ modify $ \s -> s{parseContext= ctx} -- actualize the parse context++ return c+ `while` \c ->do+ is <- isTLS c+ px <- getHTTProxyParams is+ tr ("PX=", px)+ (if isJust px then return True else do c <- anyChar ; tPutStr $ BS.singleton c; tr "anyChar"; return True) <|> do+ TOD t _ <- liftIO $ getClockTime+ -- ("PUTMVAR",nodeHost node)+ liftIO $ putMVar (isBlocked c) $ Just t+ liftIO (writeIORef (connData c) Nothing) + mclose c+ tr "CONNECTION EXHAUSTED,RETRYING WITH A NEW CONNECTION"+ return False++ modify $ \s -> s{execMode=Serial}+ let blocked= isBlocked c -- TODO: the same flag is used now for sending and receiving+ tr "after send"+ --showNext "NEXT" 100+ --try (do r <-tTake 10;liftIO $ print "NOTPARSED"; liftIO $ print r; empty) <|> return()+ first@(vers,code,_) <- getFirstLineResp <|> do + r <- notParsed+ error $ "No HTTP header received:\n"++ up r+ tr ("FIRST line",first)+ headers <- getHeaders+ let hdrs= HTTPHeaders first headers+ setState hdrs++--tr ("HEADERS", first, headers)+ + guard (BC.head code== '2') + <|> do Raw body <- parseBody headers+ error $ show (hdrs,body) -- decode the body and print++ result <- parseBody headers++ when (vers == http10 ||+ -- BS.isPrefixOf http10 str ||+ lookup "Connection" headers == Just "close" )+ $ do+ TOD t _ <- liftIO $ getClockTime++ liftIO $ putMVar blocked $ Just t+ liftIO $ mclose c+ liftIO $ takeMVar blocked+ return()+ + --tr ("result", result)+ + + --when (not $ null rest) $ error "THERE WERE SOME REST"+ ctx <- gets parseContext+ -- "SET PARSECONTEXT PREVIOUS"+ liftIO $ writeIORef (istream c) ctx + + ++ TOD t _ <- liftIO $ getClockTime+ -- ("PUTMVAR",nodeHost node)+ liftIO $ putMVar blocked $ Just t++ + if (isJust mcon) then setData (fromJust mcon) else delData c+ return result+ where+ isTLS c= liftIO $ do+ cdata <- readIORef $ connData c+ case cdata of+ Just(TLSNode2Node _) -> return True+ _ -> return False++ while act fix= do r <- act; b <- fix r; if b then return r else act++parseBody headers= case lookup "Transfer-Encoding" headers of+ Just "chunked" -> dechunk |- deserialize++ _ -> case fmap (read . BC.unpack) $ lookup "Content-Length" headers of++ Just length -> do+ msg <- tTake length+ tr ("GOT", length)+ withParseString msg deserialize+ _ -> do+ str <- notParsed -- TODO: must be strict to avoid premature close+ BS.length str `seq` withParseString str deserialize+++getFirstLineResp= do++ -- showNext "getFirstLineResp" 20+ (,,) <$> httpVers <*> (BS.toStrict <$> getCode) <*> getMessage+ where+ httpVers= tTakeUntil (BS.isPrefixOf "HTTP" ) >> parseString+ getCode= parseString+ getMessage= tTakeUntilToken ("\r\n")+ --con<- getState <|> error "rawHTTP: no connection?"+ --mclose con xxx+ --maybeClose vers headers c str+++++dechunk= do++ n<- numChars+ if n== 0 then do string "\r\n"; return SDone else do+ r <- tTake $ fromIntegral n !> ("numChars",n)+ --tr ("message", r)+ trycrlf+ tr "SMORE1"+ return $ SMore r++ <|> return SDone !> "SDone in dechunk"+ + where+ trycrlf= try (string "\r\n" >> return()) <|> return ()+ numChars= do l <- hex ; tDrop 2 >> return l++#endif+++-- | crawl the nodes executing the same action in each node and accumulate the results using a binary operator++foldNet :: Loggable a => (Cloud a -> Cloud a -> Cloud a) -> Cloud a -> Cloud a -> Cloud a+foldNet op init action = atServer $ do+ ref <- onAll $ liftIO $ newIORef Nothing -- eliminate additional results due to unneded parallelism when using (<|>)+ r <- exploreNetExclude []+ v <-localIO $ atomicModifyIORef ref $ \v -> (Just r, v)+ case v of+ Nothing -> return r+ Just _ -> empty+ where+ exploreNetExclude nodesexclude = loggedc $ do+ local $ tr "EXPLORENETTTTTTTTTTTT"+ action `op` otherNodes+ where+ otherNodes= do+ node <- local getMyNode+ nodes <- local getNodes'+ tr ("NODES to explore",nodes)+ let nodesToExplore= Prelude.tail nodes \\ (node:nodesexclude)+ callNodes' nodesToExplore op init $+ exploreNetExclude (union (node:nodesexclude) nodes)++ getNodes'= getEqualNodes -- if isBrowserInstance then return <$>getMyNode -- getEqualNodes+ -- else (++) <$> getEqualNodes <*> getWebNodes+++exploreNet :: (Loggable a,Monoid a) => Cloud a -> Cloud a+exploreNet = foldNet mappend mempty++exploreNetUntil :: (Loggable a) => Cloud a -> Cloud a+exploreNetUntil = foldNet (<|>) empty+++-- | only execute if the the program is executing in the browser. The code inside can contain calls to the server.+-- Otherwise return empty (so it stop the computation and may execute alternative computations).+onBrowser :: Cloud a -> Cloud a+onBrowser x= do+ r <- local $ return isBrowserInstance+ if r then x else empty++-- | only executes the computaion if it is in the server, but the computation can call the browser. Otherwise return empty+onServer :: Cloud a -> Cloud a+onServer x= do+ r <- local $ return isBrowserInstance+ if not r then x else empty+++-- | If the computation is running in the server, translates i to the browser and return back.+-- If it is already in the browser, just execute it+atBrowser :: Loggable a => Cloud a -> Cloud a+atBrowser x= do+ r <- local $ return isBrowserInstance+ if r then x else atRemote x++-- | If the computation is running in the browser, translates i to the server and return back.+-- If it is already in the server, just execute it+atServer :: Loggable a => Cloud a -> Cloud a+atServer x= do+ r <- local $ return isBrowserInstance+ tr ("AT SERVER",r)+ if not r then x else atRemote x++------------------- timeouts -----------------------+-- delete connections.+-- delete receiving closures before sending closures++delta= 60 -- 3*60+connectionTimeouts :: TransIO ()+connectionTimeouts= do+ labelState "loop connections"++ threads 0 $ waitEvents $ return () --loop+ liftIO $ threadDelay 10000000+ -- tr "timeouts"+ TOD time _ <- liftIO $ getClockTime+ toClose <- liftIO $ atomicModifyIORef connectionList $ \ cons ->+ Data.List.partition (\con ->+ let mc= unsafePerformIO $ readMVar $ isBlocked con++ in isNothing mc || -- check that is not doing some IO+ ((time - fromJust mc) < delta) ) cons -- !> Data.List.length cons+ -- time etc are in a IORef+ forM_ toClose $ \c -> liftIO $ do++ tr "close "+ tr $ idConn c+ when (calling c) $ mclose c+ cleanConnectionData c -- websocket connections close everithing on timeout++cleanConnectionData c= liftIO $ do+ -- reset the remote accessible closures+ modifyIORef globalFix $ \m -> M.insert (idConn c) (False,[]) m+ modifyMVar_ (localClosures c) $ const $ return M.empty+ modifyIORef globalFix $ \m -> M.insert (idConn c) (True,[]) m++loopClosures= do+ labelState "loop closures"++ threads 0 $ do -- in the current thread+ waitEvents $ threadDelay 5000000 -- every 5 seconds++ nodes <- getNodes -- get the nodes known+ node <- choose $ tail nodes -- walk trough them, except my own node++ guard (isJust $ connection node) -- when a node has connections+ nc <- liftIO $ readMVar $ fromJust (connection node) -- get them+ conn <- choose nc -- and walk trough them+ lcs <- liftIO $ readMVar $ localClosures conn -- get the local endpoints of this node for that connection+ (closLocal,(mv,clos,_,cont)) <- choose $ M.toList lcs -- walk trough them+ chs <- liftIO $ readMVar $ children $ fromJust $ parent cont -- get the threads spawned by requests to this endpoint+ return()+ --return ("NUMBER=",length chs)+ + guard (null chs) -- when there is no activity+ tr ("REMOVING", closLocal)+ liftIO $ modifyMVar (localClosures conn) $ \lcs -> return $ (M.delete closLocal lcs,()) -- remove the closure+ msend conn $ SLast $ ClosureData clos closLocal mempty -- notify the remote node++ -- tr ("THREADS ***************", length chs)+++
+ src/Transient/Move/PubSub.hs view
@@ -0,0 +1,152 @@+{-# LANGUAGE CPP, TypeSynonymInstances,FlexibleInstances #-} +module Transient.Move.PubSub where +import Transient.Base +import Transient.Internals ((!>)) +import Transient.Move +import Transient.Move.Utils +import qualified Data.Map as M +import Control.Applicative +import Control.Monad +import Data.List +import Data.Maybe +import Data.IORef +import System.IO.Unsafe +import Control.Monad.IO.Class (liftIO) +import Data.ByteString.Lazy.Char8 (pack, unpack) +import Data.Typeable +#ifndef ghcjs_HOST_OS +import Data.TCache +import Data.TCache.DefaultPersistence +#endif + + + + +type Suscribed = M.Map String [Node] + +#ifndef ghcjs_HOST_OS + +instance Indexable Suscribed where + key _= "#suscribed" + + +instance Serializable Suscribed where + serialize= pack . show + deserialize= read . unpack + + + + +suscribed= getDBRef "#suscribed" :: DBRef Suscribed + +atomicModifyDBRef :: DBRef Suscribed -> (Suscribed -> (Suscribed,a)) -> IO a +atomicModifyDBRef ref proc= atomically $ do + x <- readDBRef ref `onNothing` return M.empty + let (r,y) = proc x + writeDBRef ref r + return y + + + +#else + +suscribed= undefined + +atomicModifyDBRef a b= return () + + + +#endif + +suscribe :: (Typeable a,Loggable a) => String -> Cloud a +suscribe key= do + node <- local getMyNode + local (getMailbox' key) <|> notifySuscribe key node + + +notifySuscribe key node = atServer (do + localIO $ atomicModifyDBRef suscribed $ \ss -> (insert key [ node] ss,()) + susc node) + where + susc node=do + exploreNet $ localIO $ liftIO $ atomicModifyDBRef suscribed $ \ss -> (insert key [node] ss,()) + + + empty + + insert h node susc= + let ns = fromMaybe [] $ M.lookup h susc + in M.insert h (union node ns) susc + + + + +unsuscribe key withness= do + node <- local getMyNode + local $ deleteMailbox' key withness + atServer $ exploreNet $ localIO $ atomicModifyDBRef suscribed $ \ss -> (delete key [node] ss,()) + + + where + delete h nodes susc= + let ns = fromMaybe [] $ M.lookup h susc + in M.insert h (ns \\ nodes) susc + + + +publish :: (Typeable a, Loggable a) => String -> a -> Cloud () +publish key dat= do + n <- local getMyNode + publishExclude [n] key dat + where + -- publishExclude :: Loggable a => [Node] -> String -> a -> Cloud () + publishExclude excnodes key dat= foldPublish (<|>) empty excnodes key $ local $ do + putMailbox' key dat + return () !> "PUTMAILBOX" + empty + return() + +-- | executes `proc` in all the nodes suscribed to `key` +foldPublish op init excnodes key proc= atServer $ do +#ifndef ghcjs_HOST_OS + nodes <- localIO $ atomically ((readDBRef suscribed) `onNothing` return M.empty) + >>= return . fromMaybe [] . M.lookup key +#else + nodes <- localIO empty +#endif + let unodes= union nodes excnodes + return() !> ("NODES PUB",nodes \\ excnodes) + foldr op init $ map pub (nodes \\ excnodes) + empty + + where + + pub node= runAt node $ proc + + + +{- +examples +main = keep $ initNode $ inputNodes <|> (onBrowser $ do + + --addWebNode + + --local $ optionn ("f" :: String) "fire" + -- crawl the cloud to list all the nodes connected + --r <- exploreNet $ local $ return <$> getMyNode :: Cloud [Node] + --localIO $ print r + --empty + + wnode <- local getMyNode + atRemote $ local $ updateConnectionInfo wnode "" >> return () + + + r <- suscribe "hello" <|> do + local $ optionn ("f" :: String) "fire" + publish ("hello" ::String) ("world" :: String) + empty + + local $ render $ rawHtml $ p (r :: String) ) + + +-}
src/Transient/Move/Services.hs view
@@ -1,3 +1,4 @@+ ----------------------------------------------------------------------------- -- -- Module : Transient.Move.Services @@ -11,27 +12,86 @@ -- | -- ----------------------------------------------------------------------------- -{-# LANGUAGE ScopedTypeVariables, CPP, FlexibleInstances, UndecidableInstances #-} +{-# LANGUAGE ScopedTypeVariables, CPP, FlexibleInstances + , FlexibleContexts, UndecidableInstances, RecordWildCards + , MultiParamTypeClasses, ExistentialQuantification #-} +{- +TODO: + service=[("runsource", "this")] + + send the execution arguments, the source code to all monitors + compile it using the command arguments + find the host:port and set up them for each node + +generate a web interface for each service: + get the type of the argument + parse the type and generate axiom source code. +-} + + + +module Transient.Move.Services( +runService,callService, callService',callServiceFail,serve,ping +, monitorNode, monitorService, setRemoteJob,killRemoteJob + #ifndef ghcjs_HOST_OS -module Transient.Move.Services where +,initService,authorizeService,requestInstance,requestInstanceFail,requestInstanceHost +,findInNodes,endMonitor,freePort, controlNodeService, controlNode +-- * implementation details +,GetNodes(..) +,GetLog (..) +,ReceiveFromNodeStandardOutput (..) +,controlToken +#endif +) where + import Transient.Internals +import Transient.Logged +import Transient.Parse import Transient.Move.Internals --- import Transient.Backtrack --- import Transient.Internals(RemoteStatus(..), Log(..)) import Transient.Move.Utils -import Control.Monad.IO.Class +import Control.Monad.State +import System.IO (hFlush,stdout) import System.IO.Unsafe import Control.Concurrent.MVar import Control.Applicative -import System.Process + import Control.Concurrent(threadDelay) import Control.Exception hiding(onException) import Data.IORef +import Control.Monad(when) +import Data.Typeable +import System.Random +import Data.Maybe +import qualified Data.Map as M +import System.Environment +import Data.List(isPrefixOf) +import Unsafe.Coerce +import Data.Monoid +import Data.String +import Data.Char +import qualified Data.ByteString.Char8 as BSS +import qualified Data.ByteString.Lazy.Char8 as BS +#ifndef ghcjs_HOST_OS +import System.Directory +import GHC.IO.Handle +#else +import qualified Data.JSString as JS +#endif + + + +#ifndef ghcjs_HOST_OS +import System.Process +#endif + + + monitorService= [("service","monitor") ,("executable", "monitorService") ,("package","https://github.com/transient-haskell/transient-universe")] @@ -39,32 +99,111 @@ monitorPort= 3000 -initService :: String -> Service -> Cloud Node -initService ident service= - cached <|> installIt +#ifndef ghcjs_HOST_OS + +reInitService :: Node -> Cloud Node +reInitService node= loggedc $ cached <|> installIt where cached= local $ do - ns <- findInNodes service + ns <- findInNodes $ head $ nodeServices node if null ns then empty - else return $ head ns - installIt= do - ns <- requestInstance ident service 1 + else do + ind <- liftIO $ randomRIO(0,length ns-1) + return $ ns !! ind + installIt= do -- TODO block by service name, to avoid double initializations + ns <- requestInstanceFail node 1 if null ns then empty else return $ head ns + + +-- | initService search for any node in the list of nodes that the local node may know, for that service, instead of calling +-- the monitor. if there is no such node, it request an instance from the monitor `requestInstance`. `initService` is used by `callService` +initService :: Service -> Cloud Node +initService service= loggedc $ cached <|> installed <|> installIt + where + installed= local $ do + --if has host-port key it has been installed manually + host <- emptyIfNothing $ lookup "nodehost" service + port <- emptyIfNothing $ lookup "nodeport" service + + node <- liftIO $ createNodeServ host (read' port) [service] + addNodes [node] + return node + + cached= local $ do + ns <- findInNodes service + if null ns then empty + else do + ind <- liftIO $ randomRIO(0,length ns-1) + return $ ns !! ind + + installIt= do -- TODO(DONE) block by service name, to avoid double initializations + ns <- requestInstance service 1 + tr ("CALLING NODE: INSTALLED",ns) + if null ns then empty else return $ head ns -requestInstance :: String -> Service -> Int -> Cloud [Node] -requestInstance ident service num= loggedc $ do - -- return () !> "requestInstance" - local $ onException $ \(e:: ConnectionError) -> startMonitor >> continue -- !> ("EXCEPTIOOOOOOOOOOON",e) - nodes <- callService' ident monitorNode (ident,service,num) - local $ addNodes nodes -- !> ("ADDNODES",service) +-- | receives the specification of a service and install (if necessary) and run it (if necessary) +-- if the servi ce has been started previously, it returns the node immediately. +-- if the monitor service executable is not running `requestInstace` initiates it. +-- Instances are provisioned among the available nodes +-- The returned nodes are added to the list of known nodes. + +requestInstance :: Service -> Int -> Cloud [Node] +requestInstance service num= loggedc $ do + local $ onException $ \(e:: ConnectionError) -> do + liftIO $ putStrLn $ show ("Monitor was not running. STARTING MONITOR for this machine",e) + continue + startMonitor + + + nodes <- callService' monitorNode ("",service, num ) + local $ addNodes nodes -- !> ("ADDNODES",service) return nodes + +requestInstanceHost :: String -> Service -> Cloud Node +requestInstanceHost hostname service= do + monitorHost <- localIO $ createNodeServ hostname + (fromIntegral monitorPort) + [monitorService] + -startMonitor :: MonadIO m => m () -startMonitor= liftIO $ do - (_,_,_,h) <- createProcess . shell $ "monitorService -p start/localhost/"++ show monitorPort + nodes@[node] <- callService' monitorHost ("",service, 1::Int) + local $ addNodes nodes + return node + +requestInstanceFail :: Node -> Int -> Cloud [Node] +requestInstanceFail node num= loggedc $ do + return () !> "REQUEST INSTANCEFAIL" + local $ delNodes [node] + local $ onException $ \(e:: ConnectionError) -> do + liftIO $ putStrLn "Monitor was not running. STARTING MONITOR" + continue + startMonitor !> ("EXCEPTIOOOOOOOOOOON",e) + + + nodes <- callService' monitorNode ("", node, num ) !> "CALLSERVICE'" + local $ addNodes nodes !> ("ADDNODES") + return nodes + + +rmonitor= unsafePerformIO $ newMVar () -- to avoid races starting the monitor +startMonitor :: TransIO () +startMonitor = ( liftIO $ do + return () !> "START MONITOR" + b <- tryTakeMVar rmonitor + when (b== Just()) $ do + + r <- findExecutable "monitorService" + when ( r == Nothing) $ error "monitor not found" + (_,_,_,h) <- createProcess $ (shell $ "monitorService -p start/localhost/"++ show monitorPort ++ " > monitor.log 2>&1"){std_in=NoStream} + writeIORef monitorHandle $ Just h - threadDelay 2000000 + putMVar rmonitor () + threadDelay 2000000) + `catcht` \(e :: SomeException) -> do + liftIO $ putStrLn "'monitorService' binary should be in some folder included in the $PATH variable. Computation aborted" + empty + monitorHandle= unsafePerformIO $ newIORef Nothing endMonitor= do @@ -75,19 +214,37 @@ findInNodes :: Service -> TransIO [Node] findInNodes service = do - -- return () !> "FINDINNODES" + return () !> "FINDINNODES" nodes <- getNodes - return $ filter (\node -> head service == head (nodeServices node)) nodes + + return $ filter (hasService service) nodes + where + head1 []= (mempty,mempty) + head1 x= head x + hasService service node= not $ null $ filter (\s -> head s==head service) $ nodeServices node +-- >>> :t head $ nodeServices(undefined :: Node) +-- head $ nodeServices(undefined :: Node) :: (Package, Program) +-- +-- nodeServices :: Node -> Service +-- + + + + + + rfriends = unsafePerformIO $ newIORef ([] ::[String]) rservices = unsafePerformIO $ newIORef ([] ::[Service]) ridentsBanned = unsafePerformIO $ newIORef ([] ::[String]) rServicesBanned = unsafePerformIO $ newIORef ([] ::[Service]) -inputAuthorizations= do +inputAuthorizations :: Cloud () +inputAuthorizations= onServer $ Cloud $ do + abduce oneThread $ option "auth" "add authorizations for users and services" showPerm <|> friends <|> services <|> identBanned <|> servicesBanned empty @@ -127,13 +284,14 @@ liftIO $ putStr "services banned: " >> print servicesBanned rfreePort :: MVar Int -rfreePort = unsafePerformIO $ newMVar (monitorPort +1) +rfreePort = unsafePerformIO $ newMVar (monitorPort +2) -- executor use 3001 by default freePort :: MonadIO m => m Int -freePort= liftIO $ modifyMVar rfreePort $ \ n -> return (n+1,n) +freePort= liftIO $ modifyMVar rfreePort $ \ n -> return (n+1,n) -authorizeService :: MonadIO m => String -> Service -> m Bool + +authorizeService :: MonadIO m => String -> Service -> m Bool authorizeService ident service= do friends <- liftIO $ readIORef rfriends @@ -150,118 +308,641 @@ notElem a b= not $ elem a b +runEmbeddedService :: (Loggable a, Loggable b) => Service -> (a -> Cloud b) -> Cloud b +runEmbeddedService servname serv = do + node <- localIO $ do + port <- freePort + createNodeServ "localhost" (fromIntegral port) [servname] + listen node + wormhole' (notused 4) $ loggedc $ do + x <- local $ return (notused 0) + r <- onAll $ runCloud (serv x) <** modify (\s -> s{execMode= Remote}) --setData Remote + local $ return r + teleport + return r + +#endif + +-- | call a service. If the service is not running in some node, the monitor service would install +-- and run it. The first parameter is a weak password. + +#ifndef ghcjs_HOST_OS callService + :: (Subst1 a String, Loggable a,Loggable b) + => Service -> a -> Cloud b +callService service params = loggedc $ do + let type1 = fromMaybe "" $ lookup "type" service + + service'= case map toUpper type1 of + "HTTP" -> service ++[("nodeport", "80")] + "HTTPS" -> service ++[("nodeport", "443")] + _ -> service + node <- initService service' -- !> ("callservice initservice", service) + + if take 4 type1=="HTTP" + then callHTTPService node service' params + else callService' node params -- !> ("NODE FOR SERVICE",node) +#else +callService :: (Loggable a, Loggable b) - => String -> Service -> a -> Cloud b -callService ident service params = do - node <- initService ident service -- !> ("callservice initservice", service) - callService' ident node params -- !> ("NODE FOR SERVICE",node) + => Service -> a -> Cloud b +callService service params = local $ empty +#endif +setRemoteJob :: BSS.ByteString -> Node -> TransIO () +setRemoteJob thid node= do + JobGroup map <- getRState <|> return (JobGroup M.empty) + setRState $ JobGroup $ M.insert thid (node,0) map + +data KillRemoteJob = KillRemoteJob BSS.ByteString deriving (Read,Show, Typeable) +instance Loggable KillRemoteJob + +killRemoteJob :: Node -> BSS.ByteString -> Cloud () +killRemoteJob node thid= callService' node (KillRemoteJob thid) + + +killRemoteJobIt :: KillRemoteJob -> Cloud () +killRemoteJobIt (KillRemoteJob thid)= local $ do + st <- findState match =<< topState + liftIO $ killBranch' st + where + match st= do + (_,lab) <-liftIO $ readIORef $ labelth st + return $ if lab == thid then True else False + + +-- | notify the the monitor that a node has failed for a service and reclaim another +-- to execute the request. If the service is not running in some node, the monitor service would install +-- and run it. The first parameter is a weak password. +callServiceFail + :: (Typeable a , Typeable b, Loggable a, Loggable b) + => Node -> a -> Cloud b +#ifndef ghcjs_HOST_OS +callServiceFail node params = loggedc $ do + node <- reInitService node + callService' node params +#else +callServiceFail node params = local empty +#endif + monitorNode= unsafePerformIO $ createNodeServ "localhost" (fromIntegral monitorPort) - monitorService + [monitorService] -callService' ident node params = do - log <- onAll $ do - log <- getSData <|> return emptyLog - setData emptyLog - return log - r <- wormhole node $ do - local $ return params - + +-- | call a service located in a node +callService' :: (Loggable a, Loggable b) => Node -> a -> Cloud b +#ifndef ghcjs_HOST_OS +callService' node params = loggedc $ do + tr "callService'" + onAll abduce -- is asynchronous + + my <- onAll getMyNode -- to force connection when calling himself + if node== my + then onAll $ do + svs <- liftIO $ readIORef selfServices + + modifyData' (\log -> log{buildLog=mempty,recover=True}) $ error "No log????" + withParseString (toLazyByteString $ serialize params <> byteString (BSS.pack "/")) $ runCloud' svs + modifyData' (\log -> log{recover=True}) $ error "No log????" + log <- getState + setParseString $ toLazyByteString $ buildLog log + r <- logged empty + + return r + + else do + + localFixServ True False + local $ return () + + r <- wormhole' node $ do + + local $ return params + teleport - -- local empty `asTypeOf` typea params - local empty + + r <- local empty -- read the response + onAll $ symbol $ BS.pack "e/" + return r + delData (undefined :: LocalFixData) + return r - restoreLog log -- !> "RESTORELOG" - return r - where - typea :: a -> Cloud a - typea = undefined - restoreLog (Log _ _ logw hash)= onAll $ do + + + + + + + + -- on exception, callService is called to reclaim a new node to the monitor if necessary + + ---- `catchc` \(e :: SomeException ) -> do onAll $ delNodes [node] ; callServiceFail node params + + {- + typea :: a -> Cloud a + typea = undefined + restoreLog (Log _ _ logw hash)= onAll $ do Log _ _ logw' hash' <- getSData <|> return emptyLog let newlog= reverse logw' ++ logw -- return () !> ("newlog", logw,logw') setData $ Log False newlog newlog (hash + hash') +-} +#else +callService' node params = local empty +#endif - emptyLog= Log False [] [] 0 +sendStatusToMonitor :: String -> Cloud () +#ifndef ghcjs_HOST_OS +sendStatusToMonitor status= loggedc $ do + local $ onException $ \(e:: ConnectionError) -> continue >> startMonitor -- !> ("EXCEPTIOOOOOOOOOOON",e) + nod <- local getMyNode + callService' monitorNode (nodePort nod, status) -- <|> return() +#else +sendStatusToMonitor status= local $ return () --- catchc :: Exception e => Cloud a -> (e -> Cloud a) -> Cloud a --- catchc a b= Cloud $ catcht (runCloud' a) (\e -> runCloud' $ b e) +inputAuthorizations :: Cloud () +inputAuthorizations= empty +#endif -runEmbeddedService :: (Loggable a, Loggable b) => Service -> (a -> Cloud b) -> Cloud b -runEmbeddedService servname serv = do - node <- localIO $ do - port <- freePort - createNodeServ "localhost" (fromIntegral port) servname - listen node - wormhole (notused 4) $ loggedc $ do - x <- local $ return (notused 0) - r <- onAll $ runCloud (serv x) <** setData WasRemote - local $ return r - teleport - return r - -notused n= error $ "runService: "++ show (n::Int) ++ " variable should not be used" +catchc :: Exception e => Cloud a -> (e -> Cloud a) -> Cloud a +catchc a b= Cloud $ catcht (runCloud' a) (\e -> runCloud' $ b e) -runService :: (Loggable a, Loggable b) => Service -> Int -> (a -> Cloud b) -> Cloud b -runService servname defPort serv = do - onAll $ onException $ \(e :: SomeException)-> liftIO $ print e - initNodeServ servname - service - where - service= - wormhole (notused 1) $ do - x <- local . return $ notused 2 - r <- local $ runCloud (serv x) -- <** setData WasRemote - setData emptyLog - local $ return r - teleport - return r - emptyLog= Log False [] [] 0 +selfServices= unsafePerformIO $ newIORef empty +notused n= error $ "runService: " ++ show (n :: Int) ++ " variable should not be used" - initNodeServ servs=do - mynode <- local getNode +-- | executes a program that export endpoints that can be called with `callService` primitives. +-- It receives the service description, a default port, the services to set up and the computation to start. +-- for example the monitor exposes two services, and is started with: +-- +-- > main = keep $ runService monitorService 3000 $ +-- > [serve returnInstances +-- > ,serve addToLog] someComp +-- +-- every service incorporates a ping service and a error service. The later invoqued when the parameter received +-- do not match with any of the endpoints implemented. +runService :: Loggable a => Service -> Int -> [Cloud ()] -> Cloud a -> TransIO () +runService servDesc defPort servs proc= runCloud $ + runService' servDesc defPort servAll proc + where + servAll :: Cloud () + servAll = foldr (<|>) empty $ servs + ++ [ serve killRemoteJobIt + , serve ping + , serve (local . addNodes) + , serve getNodesIt +#ifndef ghcjs_HOST_OS + , serve redirectOutputIt + , serve sendToInputIt +#endif + , serveerror] - local $ do - conn <- defConnection - liftIO $ writeIORef (myNode conn) mynode - setState conn - onAll inputAuthorizations <|> (inputNodes >> empty) <|> return () - listen mynode + ping :: () -> Cloud () + ping = const $ return() !> "PING" - where - getNode :: TransIO Node - getNode = if isBrowserInstance then liftIO createWebNode else do - oneThread $ option "start" "re/start node" - host <- input' (Just "localhost") (const True) "hostname of this node (must be reachable) (\"localhost\"): " - port <- input' (Just 3000) (const True) "port to listen? (3000) " - liftIO $ createNodeServ host port servs + serveerror = empty -- :: Raw -> Cloud() + -- serveerror (Raw p)= error $ "parameter mismatch calling service (parameter,service): "++ show (p,servDesc) + - inputNodes= do - onServer $ do - local $ option "add" "add a new monitor node" - host <- local $ do - r <- input (const True) "Host to connect to: (none): " - if r == "" then stop else return r +data GetNodes = GetNodes deriving(Read,Show, Typeable) +instance Loggable GetNodes - port <- local $ input (const True) "port? " +-- | return the list of nodes known by the service +getNodesIt :: GetNodes -> Cloud [Node] +getNodesIt _ = local getNodes - nnode <- localIO $ createNodeServ host port monitorService - local $ do - liftIO $ putStr "Added node: ">> print nnode - addNodes [nnode] - empty + + +runService' :: Loggable a => Service -> Int -> Cloud () -> Cloud a -> Cloud () +runService' servDesc defPort servAll proc= do + + onAll $ liftIO $ writeIORef selfServices servAll + serverNode <- initNodeServ servDesc + wormhole' serverNode $ inputNodes <|> proc >> empty >> return() + return () !> "ENTER SERVALL" + onAll $ symbol $ BS.pack "e/" + servAll + tr "before teleport" + onAll $ setRState $ DialogInWormholeInitiated True + teleport + + where + + servAll' = servAll + + `catchc` \(e:: SomeException ) -> do + setState emptyLog + return () !> ("ERRORRRRRR:",e) + node <- local getMyNode + sendStatusToMonitor $ show e + + local $ do + Closure closRemote <- getData `onNothing` error "teleport: no closRemote" + conn <- getData `onNothing` error "reportBack: No connection defined: use wormhole" + msend conn $ SError $ toException $ ErrorCall $ show $ show $ CloudException node closRemote $ show e + empty -- return $ toIDyn () + + + initNodeServ servs=do + (mynode,serverNode) <- onAll $ do + node <- getNode "localhost" defPort [servDesc] + addNodes [node] + serverNode <- getWebServerNode + mynode <- if isBrowserInstance + then do + addNodes [serverNode] + return node + else return serverNode + + conn <- defConnection + liftIO $ writeIORef (myNode conn) mynode + + setState conn + return (mynode,serverNode) + + inputAuthorizations <|> return () + + listen mynode <|> return () + return serverNode + + where + + -- getNode :: TransIO Node + getNode host port servs= def <|> getNodeParams <|> getCookie + where + def= do + args <- liftIO getArgs + + if "-p" `elem` args then empty else liftIO $ createNodeServ host port servs + getNodeParams= + if isBrowserInstance then liftIO createWebNode else do + oneThread $ option "start" "re/start node" + host <- input' (Just "localhost") (const True) "hostname of this node (must be reachable) (\"localhost\"): " + port <- input' (Just 3000) (const True) "port to listen? (3000) " + liftIO $ createNodeServ host port servs + +#ifndef ghcjs_HOST_OS + getCookie= do + if isBrowserInstance then return() else do + option "cookie" "set the cookie" + c <- input (const True) "cookie: " + liftIO $ writeIORef rcookie c + empty #else -requestInstance :: String -> Service -> Int -> Cloud [Node] -requestInstance ident service num= logged empty + getCookie= empty #endif +-- | ping a service in a node. since services now try in other nodes created by the monitor until succees, ping can be +-- used to preemptively assure that there is a node ready for the service. +ping node= callService' node () :: Cloud () +sendToNodeStandardInput :: Node -> String -> Cloud () +sendToNodeStandardInput node cmd= callService' (monitorOfNode node) (node,cmd) :: Cloud () + +-- | monitor for a node is the monitor process that is running in his host +monitorOfNode node= + case lookup "relay" $ map head (nodeServices node) of + Nothing -> node{nodePort= 3000, nodeServices=[monitorService]} + Just info -> let (h,p)= read info + in Node h p Nothing [monitorService] + +data ReceiveFromNodeStandardOutput= ReceiveFromNodeStandardOutput Node BSS.ByteString deriving (Read,Show,Typeable) +instance Loggable ReceiveFromNodeStandardOutput + +receiveFromNodeStandardOutput :: Node -> BSS.ByteString -> Cloud String +receiveFromNodeStandardOutput node ident= callService' (monitorOfNode node) $ ReceiveFromNodeStandardOutput node ident + + + +-- | execute the individual services. A service within a program is invoked if the types of +-- the parameters received match with what the service expect. See `runService` for a usage example + +serve :: (Loggable a, Loggable b) => (a -> Cloud b) -> Cloud () +serve serv= do + modify $ \s -> s{execMode= Serial} + p <- onAll deserialize -- empty if the parameter does not match + modifyData' (\log -> log{recover=False}) $ error "serve: error" + loggedc $ serv p + + tr ("SERVE") + + return() + + +#ifndef ghcjs_HOST_OS + + +-- callHTTPService :: (Subst1 a String, fromJSON b) => Node -> String -> a -> Cloud ( BS.ByteString) +callHTTPService node service vars= local $ do + newVar "hostnode" $ nodeHost node + newVar "hostport" $ nodePort node + + + callString <- emptyIfNothing $ lookup "HTTPstr" service + + let calls = subst callString vars + restmsg <- replaceVars calls + --return () !> ("restmsg",restmsg) + --prox <- getProxyNode node $ map toLower $ fromJust $ lookup "type" service + rawHTTP node restmsg + {- + where + + getProxyNode nod t= do + let var= t ++ "_proxy" + + p<- liftIO $ lookupEnv var + tr ("proxy",p) + case p of + Nothing -> return nod + Just hp -> do + (upass,h,p )<- withParseString (BS.pack hp) $ do + tDropUntilToken (BS.pack "//") <|> return () + (,,) <$> tTakeWhile' (/= '@') <*> tTakeWhile' (/=':') <*> int + nod<- liftIO $ createNodeServ (BS.unpack h) p [[("type","HTTP")]] + tr upass + when (t == "https") $ do + connect <- replaceVars$ subst + ("CONNECT $hostnode:$hostport HTTP/1.1\r\n" + <> "Host: $hostnode:$hostport\r\n" + <> "Proxy-Authorization: Basic "++ BS.unpack(encode upass)++"\r\n" + <> "\r\n" :: String) vars + con <- mconnect' nod + sendRaw con $ BS.pack connect + resp <- tTakeUntilToken (BS.pack "\r\n") + tr resp + + + return nod +-} + +controlNodeService node= send <|> receive + where + send= do + local abduce + local $ do + let nname= nodeHost node ++":" ++ show(nodePort node) + + liftIO $ putStr "Controlling node " >> print nname + liftIO $ writeIORef lineprocessmode True + oldprompt <- liftIO $ atomicModifyIORef rprompt $ \oldp -> ( nname++ "> ",oldp) + cbs <- liftIO $ atomicModifyIORef rcb $ \cbs -> ([],cbs) -- remove local node options + setState (oldprompt,cbs) -- store them + + + endcontrol <|> log <|> inputs + empty + + endcontrol= do + + local $ option "endcontrol" "end controlling node" + killRemoteJob (monitorOfNode node) $ controlToken + local $ do + liftIO $ writeIORef lineprocessmode False + liftIO $ putStrLn "end controlling remote node" + (oldprompt,cbs) <- getState + liftIO $ writeIORef rcb cbs -- restore local node options + liftIO $ writeIORef rprompt oldprompt + + log = do + local $ option "log" "display the log of the node" + log <- Transient.Move.Services.getLog node + localIO $ do + + putStr "\n\n------------- LOG OF NODE: ">> print node >> putStrLn "" + mapM_ BS.putStrLn $ BS.lines log + putStrLn "------------- END OF LOG" + + inputs= do + line <- local $ inputf False "input" "" Nothing (const True) + sendToNodeStandardInput node line + + + receive= do + local $ setRemoteJob controlToken $ monitorOfNode node + r <- receiveFromNodeStandardOutput node $ controlToken + when (not $ null r) $ localIO $ putStrLn r + empty + + +controlNode node= send <|> receive + where + send= do + local abduce + local $ do + let nname= nodeHost node ++":" ++ show(nodePort node) + liftIO $ writeIORef lineprocessmode True + liftIO $ putStr "Controlling node " >> print nname + + oldprompt <- liftIO $ atomicModifyIORef rprompt $ \oldp -> ( nname++ "> ",oldp) + cbs <- liftIO $ atomicModifyIORef rcb $ \cbs -> ([],cbs) -- remove local node options + setState (oldprompt,cbs) -- store them + + + endcontrol <|> log <|> inputs + empty + + endcontrol= do + local $ option "endcontrol" "end controlling node" + killRemoteJob node $ controlToken + local $ do + liftIO $ writeIORef lineprocessmode False + liftIO $ putStrLn "end controlling remote node" + (oldprompt,cbs) <- getState + liftIO $ writeIORef rcb cbs -- restore local node options + liftIO $ writeIORef rprompt oldprompt + + log = do + local $ option "log" "display the log of the node" + log <- Transient.Move.Services.getLog node + localIO $ do + + putStr "\n\n------------- LOG OF NODE: " >> print node >> putStrLn "" + mapM_ BS.putStrLn $ BS.lines log + putStrLn "------------- END OF LOG" + + inputs= do + line <- local $ inputf False "input" "" Nothing (const True) + callService' node $ SendToInput line :: Cloud () + + + receive= do + local $ setRemoteJob controlToken $ monitorOfNode node + r <- callService' node $ RedirectOutput $ controlToken + localIO $ putStrLn r + empty + +{-# NOINLINE controlToken#-} +controlToken :: BSS.ByteString +controlToken= fromString "#control" <> fromString (show (unsafePerformIO $ (randomIO :: IO Int))) + +newtype RedirectOutput= RedirectOutput BSS.ByteString deriving (Read,Show,Typeable) +instance Loggable RedirectOutput + +newtype SendToInput= SendToInput String deriving (Read,Show,Typeable) +instance Loggable SendToInput + +sendToInputIt :: SendToInput -> Cloud () +sendToInputIt (SendToInput input)= localIO $ processLine input >> hFlush stdout -- to force flush stdout + +redirectOutputIt (RedirectOutput label)= local $ do + + (rr,ww) <- liftIO createPipe + stdout_dup <- liftIO $ hDuplicate stdout + liftIO $ hDuplicateTo ww stdout + finish stdout_dup + labelState label + read rr + where + read rr = waitEvents $ hGetLine rr + + finish stdout_dup = onException $ \(e :: SomeException) -> do + + liftIO $ hDuplicateTo stdout_dup stdout + liftIO $ putStrLn "restored control" + empty + + + + +newtype GetLog= GetLog Node deriving (Read,Show, Typeable) +instance Loggable GetLog + +getLog :: Node -> Cloud BS.ByteString +getLog node= callService' (monitorOfNode node) (GetLog node) + + +-------------------cloudshell vars ------------------------- +data LocalVars = LocalVars (M.Map String String) deriving (Typeable, Read, Show) + + +newVar :: (Show a, Typeable a) => String -> a -> TransIO () +newVar name val= noTrans $ do + LocalVars map <- getData `onNothing` return (LocalVars M.empty) + setState $ LocalVars $ M.insert name (show1 val) map + +replaceVars :: String -> TransIO String +replaceVars []= return [] +replaceVars ('$':str)= do + LocalVars localvars <- getState <|> return (LocalVars M.empty) + let (var,rest')= break (\c -> c=='-' || c==':' || c==' ' || c=='\r' || c == '\n' ) str + (manifest, rest)= if null rest' || head rest'=='-' + then break (\c -> c=='\r' || c =='\n' || c==' ') $ tailSafe rest' + else ("", rest') + + if var== "port"&& null manifest then (++) <$> (show <$> freePort) <*> replaceVars rest -- $host variable + else if var== "host" && null manifest then (++) <$> (nodeHost <$> getMyNode) <*> replaceVars rest + else if null manifest then + case M.lookup var localvars of + Just v -> do + v' <- processVar v + (++) <$> return (show1 v') <*> replaceVars rest + Nothing -> (:) <$> return '$' <*> replaceVars rest + else do + map <- liftIO $ readFile manifest >>= return . toMap + let mval = lookup var map + case mval of + Nothing -> error $ "Not found variable: "++ "$" ++ var ++ manifest + Just val -> (++) <$> return val <*> replaceVars rest + where + tailSafe []=[] + tailSafe xs= tail xs + + processVar= return . id + + toMap :: String -> [(String, String)] + toMap desc= map break1 $ lines desc + where + break1 line= + let (k,v1)= break (== ' ') line + in (k,dropWhile (== ' ') v1) + +replaceVars (x:xs) = (:) <$> return x <*> replaceVars xs + +---------------- substitution --------------------------------------------- + +subst :: Subst1 a r => String -> a -> r +subst expr= subst1 expr 1 + + +class Subst1 a r where + subst1 :: String -> Int -> a -> r + + +instance (Show b, Typeable b, Subst1 a r) => Subst1 b (a -> r) where + subst1 str n x = \a -> subst1 (subst1 str n x) (n+1) a + +instance {-# Overlaps #-} (Show a,Typeable a, Show b, Typeable b) => Subst1 (a,b) String where + subst1 str n (x,y)= subst str x y + +instance {-# Overlaps #-} (Show a,Typeable a, Show b, Typeable b + ,Show c, Typeable c) => Subst1 (a,b,c) String where + subst1 str n (x,y,z)= subst str x y z + +instance {-# Overlaps #-} (Show a,Typeable a, Show b, Typeable b + ,Show c,Typeable c, Show d, Typeable d) + => Subst1 (a,b,c,d) String where + subst1 str n (x,y,z,t)= subst str x y z t + +instance {-# Overlaps #-} (Show a,Typeable a, Show b, Typeable b + ,Show c,Typeable c, Show d, Typeable d + ,Show e,Typeable e) + => Subst1 (a,b,c,d,e) String where + subst1 str n (x,y,z,t,u)= subst str x y z t u + +instance {-# Overlaps #-} (Show a,Typeable a, Show b, Typeable b + ,Show c,Typeable c, Show d, Typeable d + ,Show e,Typeable e, Show f, Typeable f) + => Subst1 (a,b,c,d,e,f) String where + subst1 str n (x,y,z,t,u,v)= subst str x y z t u v + +instance {-# Overlaps #-} (Show a,Typeable a, Show b, Typeable b + ,Show c,Typeable c, Show d, Typeable d + ,Show e,Typeable e, Show f, Typeable f + ,Show g,Typeable g) + => Subst1 (a,b,c,d,e,f,g) String where + subst1 str n (x,y,z,t,u,v,s)= subst str x y z t u v s + +instance {-# Overlaps #-} (Show a,Typeable a, Show b, Typeable b + ,Show c,Typeable c, Show d, Typeable d + ,Show e,Typeable e, Show f, Typeable f + ,Show g,Typeable g, Show h, Typeable h) + => Subst1 (a,b,c,d,e,f,g,h) String where + subst1 str n (x,y,z,t,u,v,s,r)= subst str x y z t u v s r + + +instance {-# Overlaps #-} (Show a,Typeable a, Show b, Typeable b + ,Show c,Typeable c, Show d, Typeable d + ,Show e,Typeable e, Show f, Typeable f + ,Show g,Typeable g, Show h, Typeable h + ,Show i, Typeable i) + => Subst1 (a,b,c,d,e,f,g,h,i) String where + subst1 str n (a,b,c,d,e,f,g,h,i)= subst str a b c d e f g h i + +instance {-# Overlaps #-} (Show a,Typeable a) => Subst1 a String where + subst1 str n x= subst2 str n x + +subst2 str n x= replaces str ('$' : show n ) x + +replaces str var x= replace var (show1 x) str + +replace _ _ [] = [] +replace a b s@(x:xs) = + if isPrefixOf a s + then b++replace a b (drop (length a) s) + else x:replace a b xs + + + + +show1 :: (Show a, Typeable a) => a -> String +show1 x | typeOf x == typeOf (""::String)= unsafeCoerce x + | otherwise= show x +#endif
+ src/Transient/Move/Services/Executor.hs view
@@ -0,0 +1,211 @@+module Transient.Move.Services.Executor where + +import Transient.Internals +import Transient.Move.Internals +import Transient.Logged +import Transient.Move.Services +import Data.IORef +import System.IO.Unsafe +import qualified Data.Map as M +import qualified Data.ByteString.Lazy.Char8 as BS +import qualified Data.ByteString.Char8 as BSS +import Data.String +import Data.Typeable +import Control.Applicative +import Control.Monad +import Control.Monad.State (liftIO) +import Data.Maybe(mapMaybe) + +executorService = [("service","executor") + ,("executable", "executor") + ,("package","https://github.com/transient-haskell/transient-universe")] + + + +-- initialize N instances, of the executor service. The monitor would spread them among the nodes available. +-- the number N should be less of equal than the number of phisical machines. +-- Since the executor serivice can execute any number of processes, it sould be at most one per machine. + +initExecute number= requestInstance executorService number + +-- | execute a command in some node by an executor service, and return the result when the program finishes +networkExecute :: String -> String -> Cloud String +networkExecute cmdline input= + callService executorService (cmdline, input,()) + + + +-- | execute a process in some machine trough the local monitor and the executor. +-- This call return a process identifier +-- The process can be controlled with other services like `controlNodeProcess` +networkExecuteStream' :: String -> Cloud String +networkExecuteStream' cmdline= do + -- callService executorService cmdline + node <- initService executorService + return () !> ("STORED NODE", node) + name <- callService' node $ ExecuteStream cmdline + localIO $ print ("NAME", name) + localIO $ atomicModifyIORef rnodecmd $ \map -> (M.insert name node map,()) + local $ setRemoteJob (BSS.pack name) node -- so it can be stopped by `killRemoteJob` + return name + +-- | execute a shell command in some node using the executor service. +-- The response is received as an stream of responses, one per line +networkExecuteStream :: String -> Cloud String -- '[Multithreaded,Streaming] +networkExecuteStream cmdline= do + node <- initService executorService + flag <- onAll $ liftIO $ newIORef False + r <- callService' node cmdline + init <- onAll $ liftIO $ readIORef flag + when (not init) $ do + onAll $ liftIO $ writeIORef flag True -- get the first line (header) as the name of the process + local $ setRemoteJob (BSS.pack r) node -- so it can be stopped by `killRemoteJob` + localIO $ atomicModifyIORef rnodecmd $ \map -> (M.insert r node map,()) + return r + +rnodecmd= unsafePerformIO $ newIORef M.empty + +-- | send a message that will be read by the standard input of the program initiated by `networkExecuteStream`, identified by the command line. +-- the stream of responses is returned by that primitive. `sendExecuteStream` never return anything, since it is asynchronous +sendExecuteStream :: String -> String -> Cloud () -- '[Asynchronous] +sendExecuteStream cmdline msg= do + + return () !> ("SENDEXECUTE", cmdline) + node <- nodeForProcess cmdline + --localIO $ do + -- map <- readIORef rnodecmd + -- let mn = M.lookup cmdline map + -- case mn of + -- Nothing -> error $ "sendExecuteStream: no node executing the command: "++ cmdline + -- Just n -> return n + return () !> ("NODE", node) + callService' node (cmdline, msg) + + +controlNodeProcess cmdline= do + exnode <- nodeForProcess cmdline + --local $ do + -- map <- readIORef rinput + -- let mn = M.lookup cmdline map + -- return $ case mn of + -- Nothing -> error $ "sendExecuteStream: no node executing the command: "++ cmdline + -- Just n -> n + + send exnode <|> receive exnode + where + + send exnode= do + local abduce + local $ do + liftIO $ writeIORef lineprocessmode True + oldprompt <- liftIO $ atomicModifyIORef rprompt $ \oldp -> ( takeWhile (/= ' ') cmdline++ "> ",oldp) + cbs <- liftIO $ atomicModifyIORef rcb $ \cbs -> ([],cbs) -- remove local node options + setState (oldprompt,cbs) -- store them + + + endcontrolop exnode <|> kill exnode <|> log exnode <|> inputs exnode + empty + + kill exnode= do + local $ option "kill" "kill the process" + localIO $ putStrLn "process terminated" + killRemoteJob exnode $ fromString cmdline + endcontrol exnode + + endcontrolop exnode= do + local $ option "endcontrol" "end controlling the process" + localIO $ putStrLn "end controlling the process" + endcontrol exnode + + endcontrol exnode= do + localIO $ writeIORef lineprocessmode False + killRemoteJob exnode controlToken + local $ do + + (oldprompt,cbs) <- getState + liftIO $ writeIORef rcb cbs -- restore local node options + liftIO $ writeIORef rprompt oldprompt + + log exnode = do + local $ option "log" "display the log of the node" + log <- getLogCmd cmdline exnode + localIO $ do + + putStr "\n\n------------- LOG OF PROCESS: ">> print cmdline >> putStrLn "" + mapM_ BS.putStrLn $ BS.lines log + putStrLn "------------- END OF LOG" + + inputs exnode= do + + line <- local $ inputf False "input" "" Nothing (const True) + sendExecuteStream cmdline line + + + receive exnode= do + r <- receiveExecuteStream cmdline exnode + when (not $ null r) $ localIO $ putStrLn r + empty + + receiveExecuteStream cmd node=do + local $ setRemoteJob controlToken node + callService' node $ ReceiveExecuteStream cmd controlToken + + getLogCmd :: String -> Node -> Cloud BS.ByteString + getLogCmd cmd node= callService' node (GetLogCmd cmd) + +newtype GetLogCmd= GetLogCmd String deriving (Read, Show, Typeable) +instance Loggable GetLogCmd + +newtype ExecuteStream= ExecuteStream String deriving (Read, Show, Typeable) +instance Loggable ExecuteStream + +data ReceiveExecuteStream= ReceiveExecuteStream String BSS.ByteString deriving (Read, Show, Typeable) +instance Loggable ReceiveExecuteStream + +data GetProcesses= GetProcesses deriving (Read, Show, Typeable) +instance Loggable GetProcesses + +getProcesses :: Node -> Cloud [String] +getProcesses node= callService' node GetProcesses + + + + + +-- | get the executor that executes a process + + +nodeForProcess :: String -> Cloud Node +nodeForProcess process= do + + callService monitorService () :: Cloud () -- start/ping monitor if not started + + nods <- squeezeMonitor [] monitorNode + case nods of + [] -> error $ "no node running: "++ process + nod:_ -> return nod + where + squeezeMonitor :: [Node] -> Node -> Cloud [Node] + squeezeMonitor exc nod= do + if nod `elem` exc then return [] else do + nodes <- callService' nod GetNodes :: Cloud [Node] + return . concat =<< mapM squeeze (tail nodes) + + where + squeeze :: Node -> Cloud [Node] + squeeze node= do + + case lookup2 "service" $ nodeServices node of + + Just "monitor" -> squeezeMonitor (nod:exc) node + + Just "executor" -> do + + procs <- callService' node GetProcesses :: Cloud [String] + if process `elem` procs then return [node] else return [] + + _ -> return [] + + + +
src/Transient/Move/Utils.hs view
@@ -1,187 +1,248 @@------------------------------------------------------------------------------ --- --- Module : Transient.Move.Utils --- Copyright : --- License : MIT --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | --- ------------------------------------------------------------------------------ -{-# LANGUAGE ScopedTypeVariables #-} -module Transient.Move.Utils (initNode,initNodeDef, initNodeServ, inputNodes, simpleWebApp, initWebApp -, onServer, onBrowser, atServer, atBrowser, runTestNodes) - where - ---import Transient.Base -import Transient.Internals -import Transient.Move.Internals -import Control.Applicative -import Control.Monad.State -import Data.IORef -import System.Environment - --- | ask in the console for the port number and initializes a node in the port specified --- It needs the application to be initialized with `keep` to get input from the user. --- the port can be entered in the command line with "<program> -p start/<PORT>" --- --- A node is also a web server that send to the browser the program if it has been --- compiled to JavaScript with ghcjs. `initNode` also initializes the web nodes. --- --- This sequence compiles to JScript and executes the program with a node in the port 8080 --- --- > ghc program.hs --- > ghcjs program.hs -o static/out --- > ./program -p start/myhost/8080 --- --- `initNode`, when the application has been loaded and executed in the browser, will perform a `wormhole` to his server node. --- So the application run within this wormhole. --- --- Since the code is executed both in server node and browser node, to avoid confusion and in order --- to execute in a single logical thread, use `onServer` for code that you need to execute only in the server --- node, and `onBrowser` for code that you need in the browser, although server code could call the browser --- and vice-versa. --- --- To invoke from browser to server and vice-versa, use `atRemote`. --- --- To translate the code from the browser to the server node, use `teleport`. --- -initNode :: Loggable a => Cloud a -> TransIO a -initNode app= do - node <- getNodeParams - --abduce - initWebApp node app - - -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? " - liftIO $ createNode host port - -initNodeDef :: Loggable a => String -> Int -> Cloud a -> TransIO a -initNodeDef host port app= do - node <- def <|> getNodeParams - initWebApp node app - where - def= do - args <- liftIO getArgs - if null args then liftIO $ createNode host port else empty - -initNodeServ :: Loggable a => Service -> String -> Int -> Cloud a -> TransIO a -initNodeServ services host port app= do - node <- def <|> getNodeParams - let node'= node{nodeServices=services} - initWebApp node' $ app - where - def= do - args <- liftIO getArgs - if null args then liftIO $ createNode host port else empty - --- | 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. It returns empty -inputNodes :: Cloud empty -inputNodes= onServer $ do - --local abduce - listNodes <|> addNew - where - addNew= do - local $ oneThread $ option "add" "add a new node" - host <- local $ do - r <- input (const True) "Hostname of the node (none): " - if r == "" then stop else return r - - port <- local $ input (const True) "port? " - - services <- local $ input' (Just []) (const True) "services? ([]) " - - connectit <- local $ input (\x -> x=="y" || x== "n") "connect to the node to interchange node lists? (n) " - nnode <- localIO $ createNodeServ host port services - if connectit== "y" then connect' nnode - else local $ do - liftIO $ putStr "Added node: ">> print nnode - addNodes [nnode] - empty - - listNodes= do - local $ option "list" "list nodes" - local $ getNodes >>= liftIO . print - empty - --- | executes the application in the server and the Web browser. --- the browser must point to http://hostname:port where port is the first parameter. --- It creates a wormhole to the server. --- The code of the program after `simpleWebApp` run in the browser unless `teleport` translates the execution to the server. --- To run something in the server and get the result back to the browser, use `atRemote` --- This last also works in the other side; If the application was teleported to the server, `atRemote` will --- execute his parameter in the browser. --- --- It is necesary to compile the application with ghcjs: --- --- > ghcjs program.js --- > ghcjs program.hs -o static/out --- --- > ./program --- --- -simpleWebApp :: Loggable a => Integer -> Cloud a -> IO () -simpleWebApp port app = do - node <- createNode "localhost" $ fromIntegral port - keep $ initWebApp node app - return () - --- | use this instead of smpleWebApp when you have to do some initializations in the server prior to the --- initialization of the web server -initWebApp :: Loggable a => Node -> Cloud a -> TransIO a -initWebApp node app= do - conn <- defConnection - liftIO $ writeIORef (myNode conn) node - addNodes [node] - serverNode <- getWebServerNode :: TransIO Node - - mynode <- if isBrowserInstance - then liftIO $ createWebNode - else return serverNode - runCloud $ do - listen mynode <|> return() - wormhole serverNode app - --- | only execute if the the program is executing in the browser. The code inside can contain calls to the server. --- Otherwise return empty (so it stop the computation and may execute alternative computations). -onBrowser :: Cloud a -> Cloud a -onBrowser x= do - r <- local $ return isBrowserInstance - if r then x else empty - --- | only executes the computaion if it is in the server, but the computation can call the browser. Otherwise return empty -onServer :: Cloud a -> Cloud a -onServer x= do - r <- local $ return isBrowserInstance - if not r then x else empty - - --- | If the computation is running in the server, translates i to the browser and return back. --- If it is already in the browser, just execute it -atBrowser :: Loggable a => Cloud a -> Cloud a -atBrowser x= do - r <- local $ return isBrowserInstance - if r then x else atRemote x - --- | If the computation is running in the browser, translates i to the server and return back. --- If it is already in the server, just execute it -atServer :: Loggable a => Cloud a -> Cloud a -atServer x= do - r <- local $ return isBrowserInstance - if not r then x else atRemote x - --- | 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 - nodes <- onAll $ mapM (\p -> liftIO $ createNode "localhost" p) ports - foldl (<|>) empty (map listen nodes) <|> return() - +-----------------------------------------------------------------------------+--+-- Module : Transient.Move.Utils+-- Copyright :+-- License : MIT+--+-- Maintainer : agocorona@gmail.com+-- Stability :+-- Portability :+--+-- |+--+-----------------------------------------------------------------------------+{-# LANGUAGE CPP, ScopedTypeVariables #-}+module Transient.Move.Utils (initNode,initNodeDef, initNodeServ, inputNodes, simpleWebApp, initWebApp+, onServer, onBrowser, atServer, atBrowser, runTestNodes, showURL)+ where++--import Transient.Base+import Transient.Internals+import Transient.Logged+import Transient.Move.Internals+import Control.Applicative+import Control.Monad.State+import Data.IORef+import System.Environment+import System.IO.Error+import Data.Typeable+import Data.List((\\), isPrefixOf)+import qualified Data.ByteString.Char8 as BS+import Control.Exception hiding(onException)+import System.IO.Unsafe++rretry= unsafePerformIO $ newIORef False++-- | ask in the console for the port number and initializes a node in the port specified+-- It needs the application to be initialized with `keep` to get input from the user.+-- the port can be entered in the command line with "<program> -p start/<PORT>"+--+-- A node is also a web server that send to the browser the program if it has been+-- compiled to JavaScript with ghcjs. `initNode` also initializes the web nodes.+--+-- This sequence compiles to JScript and executes the program with a node in the port 8080+--+-- > ghc program.hs+-- > ghcjs program.hs -o static/out+-- > ./program -p start/myhost/8080+--+-- `initNode`, when the application has been loaded and executed in the browser, will perform a `wormhole` to his server node.+-- So the application run within this wormhole.+--+-- Since the code is executed both in server node and browser node, to avoid confusion and in order+-- to execute in a single logical thread, use `onServer` for code that you need to execute only in the server+-- node, and `onBrowser` for code that you need in the browser, although server code could call the browser+-- and vice-versa.+--+-- To invoke from browser to server and vice-versa, use `atRemote`.+--+-- To translate the code from the browser to the server node, use `teleport`.+--+initNode :: Loggable a => Cloud a -> TransIO a+initNode app= do+ node <- getNodeParams ++ rport <- liftIO $ newIORef $ nodePort node+ node' <- return node `onException'` ( \(e :: IOException) -> do+ if (ioeGetErrorString e == "resource busy") + then do+ liftIO $ putStr "Port busy: " >> print (nodePort node)+ retry <- liftIO $ readIORef rretry+ if retry then do liftIO $ print "retrying with next port" ;continue else empty+ port <- liftIO $ atomicModifyIORef rport $ \p -> (p+1,p+1)+ return node{nodePort= port}+ + else return node )+ return () !> ("NODE", node')+ initWebApp node' app++++getNodeParams :: TransIO Node+getNodeParams =+ if isBrowserInstance then liftIO createWebNode else+#ifdef ghcjs_HOST_OS+ empty+#else+ do+ oneThread $ option "start" "re/start node"++ host <- input' (Just "localhost") (const True) "hostname of this node. (Must be reachable, default:localhost)? "+ retry <-input' (Just "n") (== "retry") "if you want to retry with port+1 when fail, write 'retry': "+ when (retry == "retry") $ liftIO $ writeIORef rretry True+ port <- input (const True) "port to listen? "+ liftIO $ createNode host port+ <|> getCookie+ where+ getCookie= do+ if isBrowserInstance then return() else do+ option "cookie" "set the cookie"+ c <- input (const True) "cookie: "+ liftIO $ writeIORef rcookie c+ empty+#endif+ +initNodeDef :: Loggable a => String -> Int -> Cloud a -> TransIO a+initNodeDef host port app= do+ node <- def <|> getNodeParams -- <|> maybeRetry+ initWebApp node app+ where+ def= do+ args <- liftIO getArgs+ if null args then liftIO $ createNode host port else empty++initNodeServ :: Loggable a => Service -> String -> Int -> Cloud a -> TransIO a+initNodeServ services host port app= do+ node <- def <|> getNodeParams+ let node'= node{nodeServices=[services]}+ initWebApp node' $ app+ where+ def= do+ args <- liftIO getArgs+ if null args then liftIO $ createNode host port else empty++-- | 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. It returns empty.+-- to input a node, enter "add" then the host and the port, the service description (if any) and "y" or "n"+-- to either connect to that node and synchronize their lists of nodes or not.+--+-- A typical sequence of initiation of an application that includes `initNode` and `inputNodes` is:+--+-- > program -p start/host/8000/add/host2/8001/n/add/host3/8005/y+--+-- "start/host/8000" is read by `initNode`. The rest is initiated by `inputNodes` in this case two nodes are added.+-- the first of the two is not connected to synchronize their list of nodes. The second does.+inputNodes :: Cloud empty+inputNodes= onServer $ do + local $ abduce >> labelState (BS.pack "inputNodes")+ listNodes <|> addNew+ where+ addNew= do+ local $ do+ option "add" "add a new node"+ return ()+ host <- local $ do+ r <- input (const True) "Hostname of the node (none): "+ if r == "" then stop else return r++ port <- local $ input (const True) "port? "+ serv <- local $ nodeServices <$> getMyNode + services <- local $ input' (Just serv) (const True) ("services? ("++ show serv ++ ") ")++ connectit <- local $ input (\x -> x=="y" || x== "n") "connect to the node to interchange node lists? (n) "+ ++ nnode <- localIO $ createNodeServ host port services+ if connectit== "y" then connect' nnode+ else local $ do+ liftIO $ putStr "Added node: ">> print nnode+ addNodes [nnode]+ empty++ listNodes= do+ local $ option "nodes" "list nodes"+ local $ do+ nodes <- getNodes+ liftIO $ putStrLn "list of nodes known in this node:"+ liftIO $ mapM (\(i,n) -> do putStr (show i); putChar('\t'); print n) $ zip [0..] nodes+ empty+ +-- | show the URL that may be called to access that functionality within a program +showURL= onAll$ do + Closure closRemote <- getSData <|> return (Closure 0 )--get myclosure+ --get remoteclosure+ log <- getLog --get path + n <- getMyNode+ liftIO $ do+ putStr "'http://"+ putStr $ nodeHost n+ putStr ":"+ putStr $show $ nodePort n+ putStr "/"+ putStr $ show 0+ putStr "/"+ putStr $ show closRemote+ putStr "/"+ putStr $ show $ fulLog log+ putStrLn "'"++ +-- | executes the application in the server and the Web browser.+-- the browser must point to http://hostname:port where port is the first parameter.+-- It creates a wormhole to the server.+-- The code of the program after `simpleWebApp` run in the browser unless `teleport` translates the execution to the server.+-- To run something in the server and get the result back to the browser, use `atRemote`+-- This last also works in the other side; If the application was teleported to the server, `atRemote` will+-- execute his parameter in the browser.+--+-- It is necesary to compile the application with ghcjs:+--+-- > ghcjs program.js+-- > ghcjs program.hs -o static/out+--+-- > ./program+--+--+simpleWebApp :: (Typeable a, Loggable a) => Integer -> Cloud a -> IO ()+simpleWebApp port app = do+ node <- createNode "localhost" $ fromIntegral port+ keep $ initWebApp node app+ return ()++-- | use this instead of simpleWebApp when you have to do some initializations in the server prior to the+-- initialization of the web server+initWebApp :: Loggable a => Node -> Cloud a -> TransIO a+initWebApp node app= do++ conn <- defConnection+ liftIO $ writeIORef (myNode conn) node+ setNodes [node]+ serverNode <- getWebServerNode :: TransIO Node+ mynode <- if isBrowserInstance+ then do+ addNodes [serverNode]+ return node+ else return serverNode++ runCloud' $ do+ listen mynode <|> return()+ serverNode <- onAll getWebServerNode+ wormhole serverNode app+ ++ +-- | 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+ nodes <- onAll $ mapM (\p -> liftIO $ createNode "localhost" p) ports+ onAll $ addNodes nodes+ foldl (<|>) empty (map listen1 nodes) <|> return()+ where + listen1 n= do+ listen n+ onAll $ do+ ns <- getNodes+ addNodes $ n: (ns \\[n])+ conn <- getState <|> error "runTestNodes error"+ liftIO $ writeIORef (myNode conn) n+
tests/TestSuite.hs view
@@ -10,7 +10,7 @@ import Transient.Base import Transient.Internals import Transient.Indeterminism -import Transient.Move +import Transient.Move.Internals import Transient.Move.Utils import Transient.Move.Services import Transient.MapReduce @@ -20,17 +20,22 @@ import Control.Monad.State import Control.Exception --- #define _UPK_(x) {-# UNPACK #-} !(x) +import Control.Concurrent(threadDelay ) -shouldRun x= local $ getMyNode >>= \p -> assert ( p == (x)) (return ()) +#define SHOULDRUNIN(x) (local $ do p <-getMyNode; liftIO $ print (p,x) ;assert ( p == (x)) (liftIO $ print p)) + +-- #define _UPK_(x) {-# UNPACK #-} !(x) + +-- SHOULDRUNIN x= local $ getMyNode >>= \p -> assert ( p == (x)) (liftIO $ print p) + service= [("service","test suite") ,("executable", "test-transient1") ,("package","https://github.com/agocorona/transient-universe")] main= do - mr <- keep $ test `catcht` \(e:: SomeException) -> liftIO (putStr "EXCEPTiON: " >> print e) >> exit (Just e) + mr <- keep test endMonitor case mr of @@ -38,43 +43,49 @@ 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" + local $ guard (nodePort node0== 8080) -- only executes locally in node 8080 - - [node1, node2] <- requestInstance "PIN1" service 2 - + [node1, node2] <- requestInstance service 2 - local ( option "f" "fire") <|> return "" -- to repeat the test, remove exit + local ( option "f" "fire") <|> return "" -- to repeat the tests, remove the "exit" at the end + + + localIO $ putStrLn "------checking empty in remote node when the remote call back to the caller #46 --------" + + r <- runAt node1 $ do + SHOULDRUNIN(node1) + runAt node2 $ (runAt node1 $ SHOULDRUNIN(node1) >> empty ) <|> (SHOULDRUNIN(node2) >> return "world") + localIO $ print r + + localIO $ putStrLn "------checking Alternative distributed--------" r <- local $ collect 3 $ - runCloud $ (runAt node0 (shouldRun( node0) >> return "hello" )) - <|> (runAt node1 (shouldRun( node1) >> return "world" )) - <|> (runAt node2 (shouldRun( node2) >> return "world2" )) + runCloud $ (runAt node0 (SHOULDRUNIN( node0) >> return "hello" )) + <|> (runAt node1 (SHOULDRUNIN( node1) >> return "world" )) + <|> (runAt node2 (SHOULDRUNIN( node2) >> return "world2" )) - assert(sort r== ["hello", "world","world2"]) $ localIO $ print r + assert(sort r== ["hello", "world","world2"]) $ localIO $ print r localIO $ putStrLn "--------------checking Applicative distributed--------" - r <- loggedc $(runAt node0 (shouldRun( node0) >> return "hello ")) - <> (runAt node1 (shouldRun( node1) >> return "world " )) - <> (runAt node2 (shouldRun( node2) >> return "world2" )) + r <- loggedc $(runAt node0 (SHOULDRUNIN( node0) >> return "hello ")) + <> (runAt node1 (SHOULDRUNIN( node1) >> return "world " )) + <> (runAt node2 (SHOULDRUNIN( node2) >> return "world2" )) assert(r== "hello world world2") $ localIO $ print r localIO $ putStrLn "----------------checking monadic, distributed-------------" - r <- runAt node0 (shouldRun(node0) - >> runAt node1 (shouldRun (node1) - >> runAt node2 (shouldRun(node2) >> (return "HELLO" )))) + r <- runAt node0 (SHOULDRUNIN(node0) + >> runAt node1 (SHOULDRUNIN(node1) + >> runAt node2 (SHOULDRUNIN(node2) >> (return "HELLO" )))) assert(r== "HELLO") $ localIO $ print r @@ -85,7 +96,8 @@ assert (sort (M.toList r) == sort [("hello",2::Int),("world",1)]) $ return r - local $ exit (Nothing :: Maybe SomeException) -- remove this to repeat the test + onAll $ exit (Nothing :: Maybe SomeException) -- remove this to repeat the test +
transient-universe.cabal view
@@ -1,155 +1,233 @@-name: transient-universe -version: 0.5.0.0 -cabal-version: >=1.10 -build-type: Simple -license: MIT -license-file: LICENSE -maintainer: agocorona@gmail.com -homepage: https://github.com/transient-haskell/transient-universe -bug-reports: https://github.com/transient-haskell/transient-universe/issues -synopsis: Remote execution and map-reduce: distributed computing for Transient -description: - See <http://github.com/transient-haskell/transient>. -category: Control, Distributed Computing -author: Alberto G. Corona -extra-source-files: - ChangeLog.md README.md - app/client/Transient/Move/Services/MonitorService.hs - app/server/Transient/Move/Services/MonitorService.hs - -source-repository head - type: git - location: https://github.com/transient-haskell/transient-universe - -library - - if !impl(ghcjs >=0.1) - exposed-modules: - Transient.Move.Services - - if impl(ghcjs >=0.1) - build-depends: - ghcjs-base -any, - ghcjs-prim -any - else - build-depends: - HTTP -any, - TCache >= 0.12, - case-insensitive -any, - directory -any, - filepath -any, - hashable -any, - iproute -any, - network -any, - network-info -any, - network-uri -any, - vector -any, - websockets -any, - process -any, - random -any, - text -any - - exposed-modules: - Transient.Move - Transient.MapReduce - Transient.Move.Internals - Transient.Move.Utils - build-depends: - base >4 && <5, - bytestring -any, - containers -any, - mtl -any, - stm -any, - time -any, - transformers -any, - transient >= 0.6.0.0 - default-language: Haskell2010 - hs-source-dirs: src . - -executable monitorService - - if !impl(ghcjs >=0.1) - build-depends: - transformers -any, - transient >= 0.6.0.0, - transient-universe, - process, - directory - hs-source-dirs: app/server/Transient/Move/Services - else - hs-source-dirs: app/client/Transient/Move/Services - main-is: MonitorService.hs - build-depends: - base >4 && <5 - - - default-language: Haskell2010 - ghc-options: -threaded -rtsopts - -executable test-transient1 - - if !impl(ghcjs >=0.1) - build-depends: - mtl -any, - transient >= 0.5.9.2, - random -any, - text -any, - containers -any, - directory -any, - filepath -any, - stm -any, - HTTP -any, - network -any, - transformers -any, - process -any, - network -any, - network-info -any, - bytestring -any, - time -any, - vector -any, - TCache >= 0.12, - websockets -any, - network-uri -any, - case-insensitive -any, - hashable -any - main-is: TestSuite.hs - build-depends: - base >4 - default-language: Haskell2010 - hs-source-dirs: tests src . - ghc-options: -threaded -rtsopts - - -test-suite test-transient - - if !impl(ghcjs >=0.1) - build-depends: - mtl -any, - transient >= 0.5.9.2, - random -any, - text -any, - containers -any, - directory -any, - filepath -any, - stm -any, - HTTP -any, - network -any, - transformers -any, - process -any, - network -any, - network-info -any, - bytestring -any, - time -any, - vector -any, - TCache >= 0.12, - websockets -any, - network-uri -any, - case-insensitive -any, - hashable -any - type: exitcode-stdio-1.0 - main-is: TestSuite.hs - build-depends: - base >4 - default-language: Haskell2010 - hs-source-dirs: tests src . - ghc-options: -threaded -rtsopts +name: transient-universe+version: 0.6.0.0+cabal-version: >=1.10+build-type: Simple+license: MIT+license-file: LICENSE+maintainer: agocorona@gmail.com+homepage: https://github.com/transient-haskell/transient-universe+bug-reports: https://github.com/transient-haskell/transient-universe/issues+synopsis: fully composable remote execution for the creation of distributed systems+description: fully composable remote execution for the creation of distributed systems across Web clients and servers using sockets, websockets and HTTP. Web API compatible, map-reduce implementation.+ See <http://github.com/transient-haskell/transient-stack/transient-universe>.+category: Control, Distributed Computing+author: Alberto G. Corona+extra-source-files:+ ChangeLog.md README.md+ app/client/Transient/Move/Services/void.hs+ app/server/Transient/Move/Services/MonitorService.hs+ app/server/Transient/Move/Services/executor.hs++source-repository head+ type: git+ location: https://github.com/transient-haskell/transient-universe++library+ + if !impl(ghcjs >=0.1)+ exposed-modules:+ Transient.Move.Services.Executor++ if impl(ghcjs >=0.1)+ build-depends:+ -- ghcjs-base should be installed with+ -- > git clone https://github.com/ghcjs/ghcjs-base+ -- > cd ghcjs-base+ -- > cabal install --ghcjs --constraint 'primitive < 0.6.4'+ ghcjs-base -any,+ ghcjs-prim -any,+ random -any+ else+ build-depends:+ base64-bytestring,+ HTTP -any,+ TCache >= 0.12,+ case-insensitive -any,+ directory -any,+ filepath -any,+ hashable -any,+ iproute -any,+ network >=2.8.0.0 && < 3.0.0.0,+ network-info -any,+ network-uri -any,+ vector -any,+ websockets >= 0.12.7.1 ,+ process -any,+ random -any,+ text -any,+ aeson -any+ --primitive < 0.6.4.0+ -- entropy <= 0.3.6, + build-depends:+ old-time+ +++ exposed-modules:+ Transient.Move+ Transient.MapReduce+ Transient.Move.Internals+ Transient.Move.Utils+ Transient.Move.Services+ Transient.Move.PubSub+ build-depends:+ base >4 && <5,+ bytestring -any,+ containers,+ mtl -any,+ stm -any,+ time -any,+ transformers -any,+ transient >= 0.7.0.0+ default-language: Haskell2010+ hs-source-dirs: src .+ ghc-options: ++executable monitorService++ if !impl(ghcjs >=0.1)+ build-depends:+ transformers -any,+ containers,+ transient >= 0.7.0.0,+ transient-universe,+ process,+ directory,+ bytestring+ + hs-source-dirs: app/server/Transient/Move/Services+ main-is: MonitorService.hs+ else+ hs-source-dirs: app/client/Transient/Move/Services+ main-is: void.hs+ build-depends:+ base >4 && <5+++ default-language: Haskell2010+ ghc-options: -threaded -rtsopts+ ++executable executor+ if !impl(ghcjs >=0.1)+ build-depends:+ containers,+ transformers -any,+ transient >= 0.7.0.0,+ transient-universe,+ process >= 1.6.4.0,+ directory,+ bytestring, + aeson,+ time+ + hs-source-dirs: app/server/Transient/Move/Services+ main-is: executor.hs+ else+ hs-source-dirs: app/client/Transient/Move/Services+ main-is: void.hs+ build-depends:+ base >4 && <5+++ default-language: Haskell2010+ ghc-options: -threaded -rtsopts++executable controlServices+ if !impl(ghcjs >=0.1)+ build-depends:+ containers,+ transformers -any,+ transient >= 0.7.0.0,+ transient-universe,+ process >= 1.6.4.0,+ directory,+ bytestring, + aeson,+ time+ + hs-source-dirs: app/server/Transient/Move/Services+ main-is: controlServices.hs+ else+ hs-source-dirs: app/client/Transient/Move/Services+ main-is: void.hs+ build-depends:+ base >4 && <5+++ default-language: Haskell2010+ ghc-options: -threaded -rtsopts++executable test-transient1++ if !impl(ghcjs >=0.1)+ build-depends:+ mtl -any,+ transient >= 0.7.0.0,+ random -any,+ text -any,+ containers -any,+ directory -any,+ filepath -any,+ stm -any,+ base64-bytestring,+ HTTP -any,+ network >=2.8.0.0 && < 3.0.0.0,+ transformers -any,+ process -any,+ network-info -any,+ bytestring -any,+ time -any,+ vector -any,+ TCache >= 0.12,+ websockets >= 0.12.7.1 ,+ network-uri -any,+ case-insensitive -any,+ hashable -any,+ aeson,+ old-time+ + + main-is: TestSuite.hs+ build-depends:+ base >4+ default-language: Haskell2010+ hs-source-dirs: tests src .+ ghc-options: -threaded -rtsopts -fno-ignore-asserts+++test-suite test-transient++ if !impl(ghcjs >=0.1)+ build-depends:+ mtl -any,+ transient >= 0.7.0.0,+ random -any,+ text -any,+ containers -any,+ directory -any,+ filepath -any,+ stm -any,+ base64-bytestring,+ HTTP -any,+ network >=2.8.0.0 && < 3.0.0.0,+ transformers -any,+ process -any,+ network-info -any,+ bytestring -any,+ time -any,+ vector -any,+ TCache >= 0.12,+ websockets >= 0.12.7.1 ,+ network-uri -any,+ case-insensitive -any,+ hashable -any,+ aeson,+ old-time+ type: exitcode-stdio-1.0+ main-is: TestSuite.hs+ build-depends:+ base >4+ default-language: Haskell2010+ hs-source-dirs: tests src .+ ghc-options: -threaded -rtsopts -fno-ignore-asserts