-----------------------------------------------------------------------------
--
-- Module : Transient.Move
-- Copyright :
-- License : GPL-3
--
-- Maintainer : agocorona@gmail.com
-- Stability :
-- Portability :
--
-- | see <https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv>
-----------------------------------------------------------------------------
{-# LANGUAGE DeriveDataTypeable , ExistentialQuantification, OverloadedStrings
,ScopedTypeVariables, StandaloneDeriving, RecordWildCards, FlexibleContexts, CPP
,GeneralizedNewtypeDeriving #-}
module Transient.Move(
Cloud(..),runCloudIO, runCloudIO',local,onAll, loggedc, lliftIO,
listen, connect,
wormhole, teleport, copyData,
beamTo, forkTo, streamFrom, callTo, callTo',runAt,
clustered, mclustered,
newMailBox, putMailBox,getMailBox, sendNodeEvent, waitNodeEvents,
setBuffSize, getBuffSize,
createNode, createWebNode, getMyNode, setMyNode, getNodes,
addNodes, shuffleNodes,
getWebServerNode, Node(..), nodeList, Connection(..),MyNode(..), Service(),
isBrowserInstance
) where
import Transient.Base
import Transient.Internals(killChildren,getCont,runCont,EventF(..),LogElem(..),Log(..),onNothing,RemoteStatus(..),getCont)
import Transient.Logged
import Transient.EVars
import Transient.Stream.Resource
import Data.Typeable
import Control.Applicative
#ifndef ghcjs_HOST_OS
import Network
import Network.Info
import qualified Network.Socket as NS
import qualified Network.BSD as BSD
import qualified Network.WebSockets as NS(sendTextData,receiveData, Connection,RequestHead(..),sendClose)
import qualified Network.WebSockets.Connection as WS
import Network.WebSockets.Stream hiding(parse)
import Data.ByteString as B (ByteString,concat)
import qualified Data.ByteString.Char8 as BC
import qualified Data.ByteString.Lazy as BL
import Network.Socket.ByteString as SBS(send,sendMany,sendAll,recv)
import Data.CaseInsensitive(mk)
import Data.Char(isSpace)
--import GHCJS.Perch (JSString)
#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
import Data.Maybe
import Unsafe.Coerce
--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)
import Data.IORef
import qualified Data.ByteString.Lazy.Char8 as BS
import System.IO
import Control.Concurrent
import System.Random
import Data.Dynamic
import Data.String
#ifdef ghcjs_HOST_OS
type HostName = String
newtype PortID = PortNumber Int deriving (Read, Show, Eq, Typeable)
#endif
data Node= Node{ nodeHost :: HostName
, nodePort :: PortID
, connection :: IORef Pool
, services :: [Service]}
| WebNode{wconnection:: IORef Pool}
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,Applicative, Alternative, Monad, MonadState EventF, Monoid)
-- | 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
-- #ifndef ghcjs_HOST_OS
-- | run the cloud computation.
runCloudIO :: Cloud a -> IO a
runCloudIO (Cloud mx)= keep mx
-- | run the cloud computation with no console input
runCloudIO' :: Cloud a -> IO 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
-- log the result a cloud computation. like `loogged`, This eliminated all the log produced by computations
-- inside and substitute it for that single result when the computation is completed.
loggedc (Cloud mx)= Cloud $ logged mx
lliftIO :: Loggable a => IO a -> Cloud a
lliftIO= local . liftIO
--remote :: Loggable a => TransIO a -> Cloud a
--remote x= Cloud $ step' x $ \full x -> Transient $ do
-- let add= Wormhole: full
-- setSData $ Log False add add
--
-- r <- runTrans x
--
-- let add= WaitRemote: full
-- (setSData $ Log False add add) -- !!> "AFTER STEP"
-- return r
---- | stop the current computation
--stop :: Cloud a
--stop= empty
-- | continue the execution in a new node
-- all the previous actions from `listen` to this statement must have been logged
beamTo :: Node -> Cloud ()
beamTo node = local $ do
Log rec log _ <- getSData <|> return (Log False [][])
if rec
then do
setSData WasRemote
return ()
else do
msendToNode node $ SLast $ reverse log -- !> "BEAMTO" -- !> ("beamto send", log)
empty
-- | execute in the remote node a process with the same execution state
-- all the previous actions from `listen` to this statement must have been logged
forkTo :: Node -> Cloud ()
forkTo node= local $ do
Log rec log _<- getSData <|> return (Log False [][])
if rec
then do
setSData WasRemote
return ()
else do
msendToNode node $ SLast $ reverse log
-- | executes an action in another node.
callTo :: Loggable a => Node -> Cloud a -> Cloud a
callTo node remoteProc=
wormhole node $ do
teleport -- !!> "TELEPORT to remote"
r <- loggedc remoteProc
teleport -- !!> "TELEPORT from remote"
return r
-- | synonymous of `callTo`
-- all the previous actions from `listen` to this statement must have been logged
runAt :: Loggable a => Node -> Cloud a -> Cloud a
runAt= callTo
msendToNode node msg= do
conn <- mconnect node
liftIO $ msend conn msg
msend :: Loggable a => Connection -> StreamData a -> IO ()
#ifndef ghcjs_HOST_OS
msend (Connection _(Just (Node2Node _ h sock)) _ _ blocked _ _ ) r= liftIO $ do
withMVar blocked $
const $ do
hPutStrLn h "LOG a b"
hPutStrLn h ""
hPutStrLn h (show r)
hFlush h -- !> ("msend: ", r)
-- `catch` (\(e::SomeException) -> sClose sock)
msend (Connection _(Just (Node2Web sconn)) _ _ blocked _ _) r=
withMVar blocked $ const $ NS.sendTextData sconn $ BS.pack (show r)
#else
msend (Connection _ (Just (Web2Node sconn)) _ _ blocked _ _) r=
withMVar blocked $ const $ JavaScript.Web.WebSocket.send (JS.pack $ show r) sconn -- !!> "MSEND SOCKET"
#endif
msend (Connection _ Nothing _ _ _ _ _ ) _= error "calling msend: with no connection"
mread :: Loggable a => Connection -> TransIO (StreamData a)
#ifdef ghcjs_HOST_OS
mread (Connection _ (Just (Web2Node sconn)) _ _ _ _ _)= wsRead sconn
--wsAsk ws tosend= do
-- liftIO $ send ws tosend
-- wsRead ws
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) -- !!> ("WSREAD RECEIVED " ++ show str)
JM.BlobData blob -> error " blob"
JM.ArrayBufferData arrBuffer -> error "arrBuffer"
{-
wsRead1 :: Loggable a => WebSocket -> TransIO (StreamData a)
wsRead1 ws= do
reactStream (makeCallback MessageEvent) (js_onmessage ws) CB.releaseCallback (return ())
where
reactStream createHandler setHandler removeHandler iob= Transient $ do
cont <- getCont
hand <- liftIO . createHandler $ \dat ->do
runStateT (setSData dat >> runCont cont) cont
iob
mEvData <- getSessionData
case mEvData of
Nothing -> liftIO $ do
setHandler hand
return Nothing
Just dat -> do
liftIO $ print "callback called 2*****"
delSessionData dat
dat' <- case getData dat of
StringData str -> liftIO $ putStrLn "WSREAD RECEIVED " >> print str >> return (read $ JS.unpack str)
BlobData blob -> error " blob"
ArrayBufferData arrBuffer -> error "arrBuffer"
liftIO $ case dat' of
SDone -> do
removeHandler $ Callback hand
empty
sl@(SLast x) -> do
removeHandler $ Callback hand -- !!> "REMOVEHANDLER"
return $ Just sl
SError e -> do
removeHandler $ Callback hand
print e
empty
more -> return (Just more)
-}
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
"(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 _=
createNode <$> ( fromJSValUnchecked js_hostname)
<*> (fromIntegral <$> (fromJSValUnchecked js_port :: IO Int))
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 ()
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 _ h _)) _ _ blocked _ _ ) =
parallel $ do
hGetLine h -- to skip LOG header
hGetLine h
readHandler h
mread (Connection node (Just (Node2Web sconn )) bufSize events blocked _ _ )=
parallel $ do
s <- NS.receiveData sconn
return . read $ BS.unpack s -- !> ("WS MREAD RECEIVED ---->", s)
getWebServerNode port= return $ createNode "localhost" port
#endif
-- | A wormhole opens a connection with another node anywhere in a computation.
wormhole :: Loggable a => Node -> Cloud a -> Cloud a
wormhole node (Cloud comp) = local $ Transient $ do
moldconn <- getData -- !!> "wormhole"
Log rec log fulLog <- getData `onNothing` return (Log False [][])
-- initState
let lengthLog= length fulLog
if not rec -- !> ("recovery", rec)
then runTrans $ (do
conn <- mconnect node -- !> ("connecting node ", show node)
-- !> "wormhole local"
liftIO $ msend conn $ SLast $ reverse fulLog -- !> ("wh sending ", show fulLog) -- SLast will disengage the previous wormhole/listen
setSData $ conn{calling= True,offset= lengthLog}
(mread conn >>= check fulLog) <|> return () -- !> "MREAD local"
-- putState !!> "PUTSTATE"
#ifdef ghcjs_HOST_OS
addPrefix -- for the DOM identifiers
#endif
comp)
<** (when (isJust moldconn) $ setSData (fromJust moldconn))
-- <*** is not enough
else do
let oldconn = fromMaybe (error "wormhole: no connection in remote node") moldconn
if null log -- has recovered state already
then do
setData $ oldconn{calling= False,offset= lengthLog}
runTrans $ do
mlog <- mread oldconn
check fulLog mlog -- !> ("MREAD remote",mlog)
comp
<** do
setSData oldconn
setSData WasRemote -- !> " set wasremote in wormhole"
else do -- it is recovering a wormhole in the middle of a chain of nested wormholes
setData $ oldconn{calling= False,offset= lengthLog}
runTrans $ comp <** setSData oldconn
where
-- for state recovery after a remote call.
-- it has problems so now the state variables are forwarded to the remote host and returned back
-- initState= do
-- rstat <-liftIO $ newIORef undefined
-- setSData rstat
-- stat <- gets mfData
-- liftIO $ writeIORef rstat stat
--
-- putState = do
-- rstate <- getSData <|> error "rstate not defined" :: TransIO (IORef(M.Map TypeRep SData))
-- st <- get
-- log@(Log _ _ l) <- getSData :: TransIO Log
-- con <-getSData :: TransIO Connection
-- mfDat <- liftIO $ readIORef rstate -- !!> show ("LOG BEFORe", l)
-- put st {mfData= mfDat}
-- setSData log
-- setSData con
--
check fulLog mlog =
case mlog of -- !> ("RECEIVED ", mlog ) of
SError e -> do
finish $ Just e
empty
SDone -> finish Nothing >> empty
SMore log -> setSData (Log True log $ reverse log ++ fulLog ) -- !!> ("SETTING "++ show log)
SLast log -> setSData (Log True log $ reverse log ++ fulLog ) -- !!> ("SETTING "++ show log)
#ifndef ghcjs_HOST_OS
type JSString= String
pack= id
#endif
newtype Prefix= Prefix JSString deriving(Read,Show)
newtype IdLine= IdLine JSString deriving(Read,Show)
data Repeat= Repeat | RepeatHandled JSString deriving (Eq, Read, Show)
addPrefix= Transient $ do
r <- liftIO $ replicateM 5 (randomRIO ('a','z'))
-- n <- genId
-- Prefix s <- getData `onNothing` return ( Prefix "")
-- setSData $ Prefix (pack( 's': show n)<> s)
setSData $ Prefix $ pack r
return $ Just ()
-- | teleport is a new primitive that translates computations back and forth
-- reusing an already opened connection.
teleport :: Cloud ()
teleport = local $ Transient $ do
conn@Connection{calling= calling,offset= n} <- getData
`onNothing` error "teleport: No connection defined: use wormhole"
when (saveVars conn) $ runTrans (do
let copyCounter= do
r <- local $ gets mfSequence
onAll $ modify $ \s -> s{mfSequence= r}
runCloud $ do
copyData $ Prefix ""
copyData $ IdLine ""
copyData $ Repeat
copyCounter) >> return ()
Log rec log fulLog <- getData `onNothing` return (Log False [][]) -- !!> "TELEPORT"
if not rec
then do
liftIO $ msend conn $ SMore $ drop n $ reverse fulLog
-- !!> ("TELEPORT LOCAL sending" ++ show (drop n $ reverse fulLog))
-- will be read by wormhole remote
when (not calling) $ setData WasRemote -- !> "setting WasRemote in telport"
-- getState !!> "GETSTAT"
return Nothing
else do delSData WasRemote -- !> "deleting wasremote in teleport"
return (Just ()) -- !!> "TELEPORT remote"
where
#ifndef ghcjs_HOST_OS
saveVars Connection{connData= Just(Node2Node{})}= False
#endif
saveVars _ = True
-- | copy session data variable from the local to the remote node.
-- The parameter is the default value if there is none set in the local node.
-- Then the default value is also set in the local node.
copyData def = do
r <- local getSData <|> return def
onAll $ setSData r
return r
-- | `callTo` can stream data but can not inform the receiving process about the finalization. This call
-- does it.
streamFrom :: Loggable a => Node -> Cloud (StreamData a) -> Cloud (StreamData a)
streamFrom = callTo
{- All the previous actions from `listen` to this statement must have been logged
streamFrom1 :: Loggable a => Node -> TransIO (StreamData a) -> TransIO a -- (StreamData a)
streamFrom1 node remoteProc= logged $ Transient $ do
liftIO $ print "STREAMFROM"
Log rec log fulLog <- getData `onNothing` return (Log False [][])
if rec
then
runTrans $ do
liftIO $ print "callTo Remote executing"
conn <- getSData <|> error "callTo receive: no connection data"
r <- remoteProc -- !> "executing remoteProc" !> "CALLTO REMOTE" -- LOg="++ show fulLog
n <- liftIO $ msend conn r -- !> "sent response"
setSData WasRemote
stop
<|> do
setSData WasRemote
stop
else do
cont <- getCont
runTrans $ process (return()) (mconnect node) (mcloseRelease cont node) $ \conn _ -> do
liftIO $ msend conn (SLast $ reverse fulLog) !> "CALLTO LOCAL" -- send "++ show log
let log'= Wait:tail log
setData $ Log rec log' log'
liftIO $ print "mread in callTO"
mread conn
where
mcloseRelease :: EventF -> Node -> Connection -> Maybe SomeException -> IO ()
mcloseRelease cont node conn reason=
case reason of
Nothing -> release node conn
Just r -> do
forkIO $ mclose conn
killChildren cont
-}
{-
runTrans $ do
conn <- mconnect node !> "mconnect"
onFinish $ \_ -> do
liftIO $ print "MCLOSE"
liftIO $ mclose conn
c <- getCont
liftIO $ killChildren c -- liftIO $ myThreadId >>= \th -> liftIO (print th) >> killThread th
liftIO $ msend conn (SLast $ reverse fulLog) !> "CALLTO LOCAL" -- send "++ show log
let log'= Wait:tail log
setData $ Log rec log' log'
liftIO $ print "mread in callTO"
r <- mread conn
-- adjustRecThreads h
case r of
SError e -> do
liftIO $ do
release node conn
print e
stop
SDone -> release node conn >> empty
smore@(SMore x) -> return smore
other -> release node conn >> return other
-}
-- where
-- adjustRecThreads h= do
-- b <- liftIO $ hWaitForInput h 1
-- addThreads' $ if b then 1 else 0
-- liftIO $ putStrLn $ "REC "++ show (case b of True -> "INC" ; _ -> "DEC")999999*
--
-- adjustSenderThreads n
-- | n > 2 = addThreads' (-1) >> liftIO (putStrLn ("SEND DEC"))
-- | n==0 = addThreads' 1 >> liftIO (putStrLn ("SEND INC"))
-- | otherwise= return () >> liftIO(myThreadId >>= \th -> (putStrLn ("SEND "++ show th)))
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 _ h sock )) _ _ _ _ _ )= hClose h -- !!> "Handle closed"
mclose (Connection node (Just (Node2Web sconn )) bufSize events blocked _ _ )= NS.sendClose sconn ("closemsg" ::ByteString)
#else
mclose (Connection _ (Just (Web2Node sconn)) _ _ blocked _ _)= JavaScript.Web.WebSocket.close Nothing Nothing sconn
#endif
mconnect :: Node -> TransIO Connection
mconnect (Node host port pool _)= do
mh <- liftIO $ atomicModifyIORef pool $ \mh ->
case mh of
(handle:hs) -> (hs,Just handle)
[] -> ([],Nothing)
case mh of
Just handle -> return handle -- !!> "REUSED!"
Nothing -> do
Connection{comEvent= ev} <- getSData <|> error "connect: listen not set for this node"
#ifndef ghcjs_HOST_OS
liftIO $ do
let size=8192
h <- connectTo' size host port -- !!> ("CONNECTING "++ show port)
hSetBuffering h $ BlockBuffering $ Just size
let conn= (defConnection 8100){comEvent= ev,connData= Just $ Node2Node u h u}
-- writeIORef pool [conn]
return conn
#else
do
ws <- connectToWS host port
let conn= (defConnection 8100){myNode=(),comEvent= ev,connData= Just $ Web2Node ws}
-- liftIO $ writeIORef pool [conn]
return conn
#endif
where u= undefined
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
he <- BSD.getHostByName hostname
NS.connect sock (NS.SockAddrInet port (BSD.hostAddress he))
NS.socketToHandle sock ReadWriteMode
)
#else
connectToWS h (PortNumber p) =
wsOpen $ JS.pack $ "ws://"++ h++ ":"++ show p
#endif
#ifndef ghcjs_HOST_OS
-- | A connectionless version of callTo for long running remote calls
-- myNode should be set with `setMyNode`
callTo' :: (Show a, Read a,Typeable a) => Node -> Cloud a -> Cloud a
callTo' node remoteProc= do
mynode <- getMyNode
beamTo node
r <- remoteProc
beamTo mynode
return r
#endif
type Blocked= MVar ()
type BuffSize = Int
data ConnectionData=
#ifndef ghcjs_HOST_OS
Node2Node{port :: PortID
,handle :: Handle
,socket ::Socket
}
| Node2Web{webSocket :: NS.Connection}
#else
Web2Node{webSocket :: WebSocket}
#endif
#ifndef ghcjs_HOST_OS
data Connection= Connection{myNode :: TVar MyNode
#else
data Connection= Connection{myNode :: ()
#endif
,connData :: (Maybe(ConnectionData))
,bufferSize ::BuffSize
,comEvent :: EVar Dynamic
,blocked :: Blocked
,calling :: Bool
,offset :: Int}
deriving Typeable
-- | Updates the mailbox of another node. It means that the data is transported trough the network
-- The call does not end until the mailbox entry is porcessed in the remote node.
-- Mailboxes are node-wide, for all processes that share the same connection data, that is, are under the
-- same `listen` or `connect`
-- while EVars scope are visible for the process where they are initialized and their children.
-- Internally a mailbox is a well known EVar stored by `listen` in the `Connection` state.
sendRemoteNodeEvent :: Loggable a => Node -> a -> Cloud ()
sendRemoteNodeEvent node dat= runAt node $ local $ sendNodeEvent dat
newMailBox :: MonadIO m => m String
newMailBox= liftIO $ replicateM 10 (randomRIO ('a','z'))
putMailBox name dat= sendNodeEvent (name, dat)
getMailBox name= do
(nam, dat) <- waitNodeEvents
if nam /= name then empty else return dat
-- | updates the local mailbox.
sendNodeEvent :: Typeable a => a -> TransIO ()
sendNodeEvent dat= Transient $ do
Connection{comEvent=comEvent}<- getData
`onNothing` error "sendNodeEvent: accessing network events out of listen"
(runTrans $ writeEVar comEvent $ toDyn dat) -- !> "PUTMAILBOXX"
-- | wait until a message of the type expected appears in the mailbox. Then executes the continuation
-- When the message appears, all the waiting `waitNodeEvents` are executed from newer to the older
-- following the `readEVar` order.
waitNodeEvents :: Loggable a => TransIO a
waitNodeEvents = Transient $ do
Connection{comEvent=comEvent} <- getData `onNothing` error "waitNodeEvents: accessing network events out of listen"
runTrans $ do
d <- readEVar comEvent
case fromDynamic d of
Nothing -> empty
Just x -> do
writeEVar comEvent $ toDyn ()
return x
defConnection :: Int -> Connection
#ifndef ghcjs_HOST_OS
defConnection size=
Connection (unsafePerformIO $ newTVarIO $ MyNode $ createNode "invalid" 0) Nothing size
(error "defConnection: accessing network events out of listen")
(unsafePerformIO $ newMVar ())
False 0
#else
defConnection size= Connection () Nothing size
(error "defConnection: accessing network events out of listen")
(unsafePerformIO $ newMVar ())
False 0
#endif
#ifndef ghcjs_HOST_OS
setBuffSize :: Int -> TransIO ()
setBuffSize size= Transient $ do
conn<- getData `onNothing` return (defConnection 8192)
setData $ conn{bufferSize= size}
return $ Just ()
getBuffSize=
(do getSData >>= return . bufferSize) <|> return 8192
readHandler h= do
line <- hGetLine h
let [(v,left)]= readsPrec 0 line
return v -- !> line
`catch` (\(e::SomeException) -> return $ SError e)
-- where
-- hGetLine' h= do
--listen :: Node -> TransIO ()
--listen (node@(Node _ port _ _)) = do
-- addThreads 1
-- setMyNode node
-- setSData $ Log False [] []
--
-- Connection node _ bufSize events blocked <- getSData <|> return (defConnection 8192)
--
-- sock <- liftIO $ listenOn port
-- liftIO $ do NS.setSocketOption sock NS.RecvBuffer bufSize
-- NS.setSocketOption sock NS.SendBuffer bufSize
-- SMore(h,host,port1) <- parallel $ (SMore <$> accept sock)
-- `catch` (\(e::SomeException) -> do
-- print "socket exception"
-- sClose sock
-- return SDone)
--
--
-- setSData $ Connection node (Just (Node2Node port h sock )) bufSize events blocked
--
-- liftIO $ hSetBuffering h LineBuffering -- !> "LISTEN in "++ show (h,host,port1)
--
-- mlog <- parallel $ readHandler h
--
-- case mlog of
-- SError e -> do
-- liftIO $ do
-- hClose h
-- putStr "listen: "
-- print e
-- stop
--
-- SDone -> liftIO (hClose h) >> stop
-- SMore log -> setSData $ Log True log (reverse log)
-- SLast log -> setSData $ Log True log (reverse log)
listen :: Node -> Cloud ()
listen (node@(Node _ (PortNumber port) _ _)) = onAll $ do
addThreads 1
setMyNode node
setSData $ Log False [] []
Connection node _ bufSize events blocked _ _ <- getSData <|> return (defConnection 8192)
sock <- liftIO . listenOn $ PortNumber port
liftIO $ do NS.setSocketOption sock NS.RecvBuffer bufSize
NS.setSocketOption sock NS.SendBuffer bufSize
(conn,_) <- waitEvents $ NS.accept sock -- !!> "BEFORE ACCEPT"
h <- liftIO $ NS.socketToHandle conn ReadWriteMode -- !!> "NEW SOCKET CONNECTION"
-- let conn= Connection node (Just (Node2Node port h sock )) bufSize events blocked
-- setSData conn
-- mlog <- mread conn
(method,uri, headers) <- receiveHTTPHead h
-- liftIO $ print ("RECEIVED ---------->",method,uri, headers)
mlog <- case method of
"LOG" ->
do
setSData $ (Connection node (Just (Node2Node (PortNumber port) h sock ))
bufSize events blocked False 0 :: Connection)
parallel $ readHandler h -- !!> "read Listen" -- :: TransIO (StreamData [LogElem])
_ -> do
sconn <- httpMode (method,uri, headers) conn
setSData $ (Connection node (Just (Node2Web sconn ))
bufSize events blocked False 0 :: Connection)
parallel $ do
msg <- WS.receiveData sconn
-- liftIO $ print ("WS RECEIVED: ", msg)
return . read $ BC.unpack msg
-- liftIO $ putStr "LISTEN RECEIVED " >> print mlog
case mlog of
SError e -> do
liftIO $ do
hClose h
putStr "listen: "
print e
-- killChilds
c <- getCont
liftIO $ killChildren . fromJust $ parent c
empty
SDone -> empty -- liftIO (hClose h) >> stop
SMore log -> setSData $ Log True log (reverse log)
SLast log -> setSData $ Log True log (reverse log)
-- liftIO $ print "END LISTEN"
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
type Pool= [Connection] -- Pool{free :: [Handle], pending :: Int}
type Package= String
type Program= String
type Service= (Package, Program, Int)
-- * Level 2: connections node lists and operations with the node list
{-# NOINLINE emptyPool #-}
emptyPool :: MonadIO m => m (IORef Pool)
emptyPool= liftIO $ newIORef []
createNode :: HostName -> Integer -> Node
createNode h p= Node h ( PortNumber $ fromInteger p) (unsafePerformIO emptyPool) []
createWebNode= WebNode (unsafePerformIO emptyPool)
instance Eq Node where
Node h p _ _ ==Node h' p' _ _= h==h' && p==p'
_ == _ = False
instance Show Node where
show (Node h p _ servs)= show (h,p,servs)
show (WebNode _)= "webnode"
instance Read Node where
readsPrec _ ('w':'e':'b':'n':'o':'d':'e':xs)=
[(WebNode . unsafePerformIO $ emptyPool, xs)]
readsPrec _ s=
let r= readsPrec 0 s
in case r of
[] -> []
[((h,p,ss),s')] -> [(Node h p empty ss,s')]
where
empty= unsafePerformIO emptyPool
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
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. Use setMynode before listen"
#ifdef ghcjs_HOST_OS
getMyNode :: Cloud ()
getMyNode= return ()
setMyNode :: Node -> TransIO ()
setMyNode node= do
addNodes [node]
events <- newEVar
let conn= Connection () Nothing 8192 events (unsafePerformIO $ newMVar ()) False 0 :: Connection
setData conn
#else
getMyNode :: Cloud Node
getMyNode = local $ do
Connection{myNode=rnode} <- getSData <|> errorMyNode "getMyNode"
MyNode node <- liftIO $ atomically $ readTVar rnode -- `onNothing` errorMyNode "getMyNode"
return node
setMyNode :: Node -> TransIO ()
setMyNode node= do
addNodes [node]
events <- newEVar
rnode <- liftIO $ newTVarIO $ MyNode node
let conn= Connection rnode Nothing 8192 events (unsafePerformIO $ newMVar ()) False 0 :: Connection
setData conn
-- return $ Just ()
#endif
-- | return the list of nodes connected to the local node
getNodes :: MonadIO m => m [Node]
getNodes = liftIO $ atomically $ readTVar nodeList
addNodes :: (MonadIO m, MonadState EventF m) => [Node] -> m ()
addNodes nodes= do
#ifndef ghcjs_HOST_OS
mapM_ verifyNode nodes -- if the node is a web one, add his connection
#endif
liftIO . atomically $ do
prevnodes <- readTVar nodeList
writeTVar nodeList $ nub $ nodes ++ prevnodes
#ifndef ghcjs_HOST_OS
verifyNode (WebNode pool)= do
r <- getData `onNothing` error "adding web node without connection set"
case r of
conn@(Connection{connData= Just( Node2Web ws)}) ->
liftIO $ writeIORef pool [conn]
other -> return ()
verifyNode n= return ()
#endif
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
-- | 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= loggedc $ do
nodes <- onAll getNodes
foldr (<|>) empty $ map (\node -> runAt node proc) nodes -- !> ("clustered",nodes)
-- A variant of `clustered` that wait for all the responses and `mappend` them
mclustered :: (Monoid a, Loggable a) => Cloud a -> Cloud a
mclustered proc= loggedc $ do
nodes <- onAll getNodes
foldr (<>) mempty $ map (\node -> runAt node proc) nodes -- !> ("mclustered",nodes)
-- | Initiates the transient monad, initialize it as a new node (first parameter) and connect it
-- to an existing node (second parameter).
-- The other node will notify about this connection to
-- all the nodes connected to him. this new connected node will receive the list of nodes
-- the local list of nodes then is updated with this list. it can be retrieved with `getNodes`
connect :: Node -> Node -> Cloud ()
connect node remotenode = do
listen node <|> return () -- listen1 node remotenode
local $ liftIO $ putStrLn $ "connecting to: "++ show remotenode
newnode <- local $ return node -- must pass my node to the remote node or else it will use his own
nodes <- runAt remotenode $ do
mclustered $ onAll $ addNodes [newnode]
onAll $ do
liftIO $ putStrLn $ "Connected node: " ++ show node
getNodes
onAll $ liftIO $ putStrLn $ "Connected to nodes: " ++ show nodes
onAll $ addNodes nodes
--------------------------------------------
#ifndef ghcjs_HOST_OS
httpMode (method,uri, headers) conn = do
if isWebSocketsReq headers
then liftIO $ do
stream <- makeStream -- !!> "WEBSOCKETS request"
(do
bs <- SBS.recv conn 4096
return $ if BC.null bs then Nothing else Just bs)
(\mbBl -> case mbBl of
Nothing -> return ()
Just bl -> SBS.sendMany conn (BL.toChunks bl) >> return()) -- !!> show ("SOCK RESP",bl)
let
pc = WS.PendingConnection
{ WS.pendingOptions = WS.defaultConnectionOptions
, WS.pendingRequest = NS.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
else do
let uri'= BC.tail $ uriPath uri -- !> "HTTP REQUEST"
file= if BC.null uri' then "index.html" else uri'
content <- liftIO $ BL.readFile ( "static/out.jsexe/"++ BC.unpack file)
`catch` (\(e:: SomeException) -> return "NOT FOUND")
return ()
n <- liftIO $ SBS.sendMany conn $ ["HTTP/1.0 200 OK\nContent-Type: text/html\nConnection: close\nContent-Length: " <> BC.pack (show $ BL.length content) <>"\n\n"] ++
(BL.toChunks content )
return () -- !> "HTTP sent"
empty
where
uriPath = BC.dropWhile (/= '/')
isWebSocketsReq = not . null
. filter ( (== mk "Sec-WebSocket-Key") . fst)
data ParseContext a = ParseContext (IO a) a deriving Typeable
--giveData h= do
-- r <- BC.hGetLine h
-- return r !> ( "RECEIVED "++ show r)
giveData h= do
r <- readIORef rend
if r then return "" else do
r<- BC.hGetLine h -- !!> "GETLINE"
if r=="\r" || r == "" then do
writeIORef rend True
return ""
else return r
where
rend= unsafePerformIO $ newIORef False
receiveHTTPHead h = do
setSData $ ParseContext (giveData h) ""
(method, uri, vers) <- (,,) <$> getMethod <*> getUri <*> getVers
headers <- many $ (,) <$> (mk <$> getParam) <*> getParamValue -- !> (method, uri, vers)
return (method, uri, headers) -- !> (method, uri, headers)
where
getMethod= getString
getUri= getString
getVers= getString
getParam= do
dropSpaces
r <- tTakeWhile (\x -> x /= ':' && x /= '\r')
if BC.null r || r=="\r" then empty else dropChar >> return r
getParamValue= dropSpaces >> tTakeWhile (/= '\r')
dropSpaces= parse $ \str ->((),BC.dropWhile isSpace str)
dropChar= parse $ \r -> ((), BC.tail r)
getString= do
dropSpaces
tTakeWhile (not . isSpace)
tTakeWhile :: (Char -> Bool) -> TransIO BC.ByteString
tTakeWhile cond= parse (BC.span cond)
parse :: (Typeable a, Eq a, Show a, Monoid a,Monoid b) => (a -> (b,a)) -> TransIO b
parse split= do
ParseContext rh str <- getSData <|> error "parse: ParseContext not found"
if str == mempty then do
str3 <- liftIO rh
setSData $ ParseContext rh str3 -- !> str3
if str3== mempty then empty else parse split
else do
cont <- do
let (ret,str3) = split str
setSData $ ParseContext rh str3
if str3 == mempty
then return $ Left ret
else return $ Right ret
case cont of
Left r -> (<>) <$> return r <*> (parse split <|> return mempty)
Right r -> return r
#endif
#ifdef ghcjs_HOST_OS
isBrowserInstance= True
listen node = onAll $ setMyNode node
-- return () -- error "listen not implemented in browser"
#else
isBrowserInstance= False
#endif