transient-universe 0.3.1.2 → 0.3.2
raw patch · 5 files changed
+308/−232 lines, 5 filesdep +iproute
Dependencies added: iproute
Files
- src/Transient/MapReduce.hs +35/−25
- src/Transient/Move.hs +227/−161
- src/Transient/Move/Services.hs +25/−27
- src/Transient/Move/Utils.hs +19/−18
- transient-universe.cabal +2/−1
src/Transient/MapReduce.hs view
@@ -208,7 +208,7 @@ shuffler nodes = do ref@(Ref node path sav) <- mx return () -- !> ref - runAt node $ local $ runCloud ( foldAndSend nodes ref) -- <*** runCloud sendEnd + runAt node $ foldAndSend nodes ref stop @@ -222,26 +222,33 @@ -- foldAndSend :: (Hashable k, Distributable vector a)=> (Int,[(k,vector a)]) -> Cloud () foldAndSend nodes ref= do - nsent <- onAll $ liftIO $ newMVar 0 - pairs <- onAll $ getPartitionData1 ref - <|> return (error $ "DDS computed out of his node:"++ show ref) - let mpairs = groupByDestiny pairs - length <- local . return $ M.size mpairs - (i,folded) <- loggedc $ parallelize foldthem $ M.assocs mpairs + nsent <- onAll $ liftIO $ newMVar 0 + pairs <- onAll $ getPartitionData1 ref + <|> return (error $ "DDS computed out of his node:"++ show ref) + let mpairs = groupByDestiny pairs + length <- local . return $ M.size mpairs - runAt (nodes !! i) $ (local $ putMailbox box $ Reduce folded) - -- !> ("SEND REDUCE DATA",folded) - n <- lliftIO $ modifyMVar nsent $ \r -> return (r+1, r+1) - when (n == length) $ sendEnd nodes - where - foldthem (i,kvs)= local . async $ return (i,map (\(k,vs) -> (k,foldl1 red vs)) kvs) + mpart <- loggedc $ parallelize foldthem (M.assocs mpairs) <|> return Nothing + n <- lliftIO $ modifyMVar nsent $ \r -> return (r+1, r+1) + case mpart of + Just (i,folded) -> + runAt (nodes !! i) $ local $ putMailbox box $ Reduce folded + -- !> ("SEND REDUCE DATA",mynode)) + Nothing -> return () - sendEnd nodes = onNodes nodes . local $ putMailbox box (EndReduce `asTypeOf` paramOf dds) + when (n > length) $ sendEnd nodes - onNodes nodes f= foldr (<|>) empty $ map (\n -> runAt n f ) nodes + where + foldthem (i,kvs)= local . async . return . Just + $ (i,map (\(k,vs) -> (k,foldl1 red vs)) kvs) + + sendEnd nodes = onNodes nodes . local $ putMailbox box (EndReduce `asTypeOf` paramOf dds) + -- !> ("send ENDREDUCE",mynode) + 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 @@ -259,11 +266,12 @@ EndReduce -> do n <- liftIO $ modifyMVar numberSent $ \r -> return (r+1, r+1) - if n == lengthNodes -- !>("END REDUCE RECEIVED",n, lengthNodes) + + if n == lengthNodes -- !> ("END REDUCE RECEIVED",n, lengthNodes,mynode) then do cleanMailbox box (EndReduce `asTypeOf` paramOf dds) r <- liftIO $ readMVar reduceResults - return r -- !> ("reduceresult",r) + return r -- !> ("reduceresult",r) else stop @@ -276,7 +284,7 @@ return $ M.insert k (case maccum of Just accum -> red input accum Nothing -> input) map - mapM addIt (kvs `asTypeOf` paramOf' dds) -- !> ("Received Reduce",kvs) + mapM addIt (kvs `asTypeOf` paramOf' dds) -- !> ("Received Reduce",kvs) stop @@ -369,14 +377,16 @@ 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 - distribute'' xss nodes + 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'' :: (Loggable a, Distributable vector a) + => [vector a] -> [Node] -> Cloud (PartRef (vector a)) distribute'' xss nodes = parallelize move $ zip nodes xss -- !> show xss where @@ -390,21 +400,21 @@ getText :: (Loggable a, Distributable vector a) => (String -> [a]) -> String -> DDS (vector a) getText part str= DDS $ loggedc $ do nodes' <- local getNodes -- !> "DISTRIBUTE" - let nodes = filter (not . isWebNode) nodes' + let nodes = filter (not . isWebNode) nodes' let lnodes = length nodes parallelize (process lnodes) $ zip nodes [0..lnodes-1] where - isWebNode Node {nodeHost="webnode"}= True - isWebNode _= False + isWebNode node= "webnode" `elem` (map fst $ nodeServices node) + process lnodes (node,i)= do 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 - par <- generateRef xss - (return par) + generateRef xss + -- | get the worlds of an URL textUrl :: String -> DDS (DV.Vector Text.Text)
src/Transient/Move.hs view
@@ -30,7 +30,7 @@ setBuffSize, getBuffSize, #endif -createNode, createWebNode, getMyNode, getNodes, +createNode, createWebNode, createNodeServ, getMyNode, getNodes, addNodes, shuffleNodes, -- * low level @@ -42,28 +42,31 @@ ) where import Transient.Base -import Transient.Internals(killChildren,getCont,runCont,EventF(..),LogElem(..),Log(..) +import Transient.Internals(IDynamic(..),killChildren,getCont,runCont,EventF(..),LogElem(..),Log(..) ,onNothing,RemoteStatus(..),getCont,StateIO,readsPrec') import Transient.Logged +import Transient.Indeterminism(choose) import Transient.EVars import Data.Typeable import Control.Applicative #ifndef ghcjs_HOST_OS import Network import Network.Info +import qualified Data.IP as IP import qualified Network.Socket as NS import qualified Network.BSD as BSD import qualified Network.WebSockets as NWS(RequestHead(..)) import qualified Network.WebSockets.Connection as WS - import Network.WebSockets.Stream hiding(parse) -import Data.ByteString as B (ByteString,concat) +import qualified Data.ByteString as B (ByteString,concat) import qualified Data.ByteString.Char8 as BC -import qualified Data.ByteString.Lazy.Char8 as BLC +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(send,sendMany,sendAll,recv) +import qualified Network.Socket.ByteString.Lazy as SBSL import Data.CaseInsensitive(mk) import Data.Char(isSpace) --import GHCJS.Perch (JSString) @@ -102,7 +105,7 @@ import Data.IORef -import qualified Data.ByteString.Lazy.Char8 as BS + import System.IO import Control.Concurrent @@ -119,6 +122,7 @@ type HostName = String newtype PortID = PortNumber Int deriving (Read, Show, Eq, Typeable) #endif + data Node= Node{ nodeHost :: HostName , nodePort :: Int , connection :: MVar Pool @@ -130,23 +134,34 @@ 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) - - ---instance Alternative Cloud where --- empty= Cloud empty +newtype Cloud a= Cloud {runCloud ::TransIO a} deriving (Functor,Applicative,Monoid,Alternative, Monad, MonadState EventF) -- --- Cloud x <|> Cloud y= Cloud $ Transient $ do --- mx <- runTrans x -- !!> "RUNTRANS11111" --- was <- getData `onNothing` return NoRemote --- if was== WasRemote -- !> was --- then return Nothing --- else case mx of --- Nothing -> runTrans y -- !!> "RUNTRANS22222" --- justx -> return justx - +-- +--instance Applicative Cloud where +-- +-- pure = return +-- +-- x <*> y= Cloud . Transient $ do +-- l1 <- liftIO $ newIORef Nothing +-- l2 <- liftIO $ newIORef Nothing +-- runTrans $ do +-- Log _ _ full <- getData `onNothing` error "instance Applicative: no log" +-- r <- runCloud (eval l1 x) <*> runCloud (eval l2 y) +-- Just v1 <- localIO $ readIORef l1 +-- Just v2 <- localIO $ readIORef l2 +-- let full' = Var (toIDyn v1) : Var (toIDyn v2) : full +-- setData $ Log False full' full' +-- return r +-- where +-- eval l x= x >>= \v -> localIO (writeIORef l $ Just v) >> return v +-- +-- +--instance Monoid a => Monoid (Cloud a) where +-- mappend x y = mappend <$> x <*> y +-- mempty= return mempty -- | 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 @@ -241,9 +256,9 @@ -- | Within an open connection to other node opened by `wormhole`, it run the computation in the remote node and return -- the result back to the original node. atRemote proc= loggedc $ do - teleport -- !> "teleport 1111" + teleport -- !> "teleport 1111" r <- Cloud $ runCloud proc <** setData WasRemote - teleport -- !> "teleport 2222" + teleport -- !> "teleport 2222" return r -- | synonymous of `callTo` @@ -256,7 +271,7 @@ #ifndef ghcjs_HOST_OS -msend (Connection _(Just (Node2Node _ sock)) _ _ blocked _ _ ) r= do +msend (Connection _(Just (Node2Node _ sock _)) _ _ blocked _ _ ) r= do r <- liftIO $ do withMVar blocked $ const $ do @@ -363,12 +378,12 @@ getWebServerNode :: TransIO Node getWebServerNode = liftIO $ - createNode <$> ( fromJSValUnchecked js_hostname) + createNode <$> (fromJSValUnchecked js_hostname) <*> (fromIntegral <$> (fromJSValUnchecked js_port :: IO Int)) - <*> (return []) + hsonmessage ::WebSocket -> (MessageEvent ->IO()) -> IO () hsonmessage ws hscb= do cb <- makeCallback MessageEvent hscb @@ -396,7 +411,7 @@ #else -mread (Connection _(Just (Node2Node _ _)) _ _ blocked _ _ ) = parallelReadHandler -- !> "mread" +mread (Connection _(Just (Node2Node _ _ _)) _ _ blocked _ _ ) = parallelReadHandler -- !> "mread" mread (Connection node (Just (Node2Web sconn )) bufSize events blocked _ _)= @@ -407,6 +422,7 @@ -- `catch`(\(e ::SomeException) -> return $ SError e) +getWebServerNode :: TransIO Node getWebServerNode = getMyNode #endif @@ -415,7 +431,7 @@ _ -> error $ "reading " ++ s -- | 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 +-- `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 @@ -425,10 +441,10 @@ logdata@(Log rec log fulLog) <- getData `onNothing` return (Log False [][]) mynode <- runTrans getMyNode -- debug - if not rec -- !> ("wormhole recovery", rec) + if not rec -- !> ("wormhole recovery", rec) then runTrans $ (do - conn <- mconnect node -- !> (mynode,"connecting node ", node) + conn <- mconnect node -- !> (mynode,"connecting node ", node) setData conn{calling= True} #ifdef ghcjs_HOST_OS addPrefix -- for the DOM identifiers @@ -461,6 +477,7 @@ #endif newtype Prefix= Prefix JSString deriving(Read,Show) + addPrefix= Transient $ do r <- liftIO $ replicateM 5 (randomRIO ('a','z')) setData $ Prefix $ pack r @@ -477,7 +494,7 @@ -- send log with closure at head Log rec log fulLog <- getData `onNothing` return (Log False [][]) - if not rec -- !> ("teleport rec,loc fulLog=",rec,log,fulLog) + if not rec -- !> ("teleport rec,loc fulLog=",rec,log,fulLog) -- if is not recovering in the remote node then it is active then do conn@Connection{closures= closures,calling= calling} <- getData @@ -515,8 +532,8 @@ -- | copy a 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. +-- 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 @@ -538,12 +555,12 @@ #ifndef ghcjs_HOST_OS mclose (Connection _ - (Just (Node2Node _ sock )) _ _ _ _ _)= NS.sClose sock + (Just (Node2Node _ sock _ )) _ _ _ _ _)= NS.sClose sock mclose (Connection node (Just (Node2Web sconn )) bufSize events blocked _ _)= - WS.sendClose sconn ("closemsg" :: ByteString) + WS.sendClose sconn ("closemsg" :: BS.ByteString) #else @@ -560,8 +577,8 @@ case fnode of [] -> addNodes [node] >> mconnect node [Node host port pool _] -> do --- Log _ _ full <- getSData <|> error "mconnect log error" + plist <- liftIO $ readMVar pool case plist of handle:_ -> do @@ -577,9 +594,10 @@ conn <- liftIO $ do let size=8192 - sock <- connectTo' size host $ PortNumber $ fromIntegral port -- !!> ("CONNECTING "++ show port) + sock <- connectTo' size host $ PortNumber $ fromIntegral port -- !> ("CONNECTING ",node) - let conn= (defConnection 8100){myNode= my,comEvent= ev,connData= Just $ Node2Node u sock} + let conn= (defConnection 8100){myNode=my,comEvent= ev,connData= Just $ Node2Node u sock (error $ "addr: outgoing connection")} + SBS.send sock "CLOS a b\n\n" -- !> "sending CLOS" @@ -587,10 +605,11 @@ #else conn <- do + ws <- connectToWS host $ PortNumber $ fromIntegral port let conn= (defConnection 8100){comEvent= ev,connData= Just $ Web2Node ws} - return conn + return conn -- !> ("websocker CONNECION") #endif liftIO $ modifyMVar_ pool $ \plist -> return $ conn:plist @@ -643,9 +662,9 @@ data ConnectionData= #ifndef ghcjs_HOST_OS Node2Node{port :: PortID - - ,socket ::Socket - } + ,socket ::Socket + ,remoteNode :: NS.SockAddr + } | Node2Web{webSocket :: WS.Connection} #else @@ -655,12 +674,12 @@ -data Connection= Connection{myNode :: Node - ,connData :: (Maybe(ConnectionData)) - ,bufferSize ::BuffSize - ,comEvent :: IORef (M.Map T.Text (EVar Dynamic)) - ,blocked :: Blocked - ,calling :: Bool +data Connection= Connection{myNode :: Node + ,connData :: Maybe(ConnectionData) + ,bufferSize :: BuffSize + ,comEvent :: IORef (M.Map T.Text (EVar Dynamic)) + ,blocked :: Blocked + ,calling :: Bool ,closures :: MVar (M.Map Int ([LogElem], EventF))} deriving Typeable @@ -672,20 +691,20 @@ -- Internally, the mailbox is in a well known EVar stored by `listen` in the `Connection` state. newMailbox :: T.Text -> TransIO () newMailbox name= do - return () -- !> "newMailBox" +-- return () -- !> "newMailBox" Connection{comEvent= mv} <- getData `onNothing` errorMailBox - onFinish . const $ liftIO $ do - return () -- !> "NEWMAILBOX finish" - mailboxes <- readIORef mv - let me = M.lookup name mailboxes - case me of - Nothing -> empty -- !> "EMPTY" - Just (EVar id rn ref1) -> do - n <- atomically $ do - (n,n') <- readTVar rn - writeTVar rn (n-1,n'-1) -- !> ("decreased rn",n-1) - return $ n-1 - when (n==0) $ atomicModifyIORef mv $ \mboxes -> (M.delete name mboxes,()) +-- onFinish . const $ liftIO $ do +---- return () -- !> "NEWMAILBOX finish" +-- mailboxes <- readIORef mv +-- let me = M.lookup name mailboxes +-- case me of +-- Nothing -> empty -- !> "EMPTY" +-- Just (EVar id rn ref1) -> do +-- n <- atomically $ do +-- (n,n') <- readTVar rn +-- writeTVar rn (n-1,n'-1) -- !> ("decreased rn",n-1) +-- return $ n-1 +-- when (n==0) $ atomicModifyIORef mv $ \mboxes -> (M.delete name mboxes,()) ev <- newEVar liftIO $ atomicModifyIORef mv $ \mailboxes -> (M.insert name ev mailboxes,()) @@ -778,7 +797,7 @@ -- #ifndef ghcjs_HOST_OS defConnection size= - Connection (createNode "program" 0 []) Nothing size + Connection (createNode "program" 0) Nothing size (error "defConnection: accessing network events out of listen") (unsafePerformIO $ newMVar ()) False (unsafePerformIO $ newMVar M.empty) @@ -798,22 +817,22 @@ - listen :: Node -> Cloud () -listen (node@(Node _ ( port) _ _ )) = onAll $ do +listen (node@(Node _ port _ _ )) = onAll $ do addThreads 1 - addNodes [node] + setData $ Log False [] [] conn' <- getSData <|> return (defConnection 8192) ev <- liftIO $ newIORef M.empty - let conn= conn'{myNode= node , comEvent=ev} - setData conn + let conn= conn'{myNode=node, comEvent=ev} - mlog <- listenNew (fromIntegral port) conn <|> listenResponses + setData conn + addNodes [node] + mlog <- listenNew (fromIntegral port) conn <|> listenResponses - checkLog mlog + execLog mlog @@ -847,14 +866,14 @@ case method of "CLOS" -> do - return () -- !> "CLOS detected" - setData $ conn{connData=Just (Node2Node (PortNumber port) sock )} +-- return () !> "CLOS detected" + setData $ conn{connData=Just (Node2Node (PortNumber port) sock addr)} -- killOnFinish $ parallel $ readHandler -- !> "read Listen" -- :: TransIO (StreamData [LogElem]) parallelReadHandler _ -> do - sconn <- httpMode (method,uri, headers) sock -- stay serving pages until a websocket request is received + sconn <- httpMode (method, uri, headers) sock -- stay serving pages until a websocket request is received setData conn{connData= Just (Node2Web sconn )} @@ -870,12 +889,12 @@ -instance Read PortNumber where - readsPrec n str= let [(n,s)]= readsPrec n str in [(fromIntegral n,s)] +--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 +--deriving instance Read PortID +--deriving instance Typeable PortID #endif @@ -885,8 +904,9 @@ setData conn #ifndef ghcjs_HOST_OS case conn of - Connection _(Just (Node2Node _ sock)) _ _ _ _ _ -> - setData $ (ParseContext (giveData sock) "" :: ParseContext ByteString) + Connection _(Just (Node2Node _ sock _)) _ _ _ _ _ -> do + input <- liftIO $ SBSL.getContents sock + setData $ (ParseContext (error "SHOULD NOT READ 2") input :: ParseContext BS.ByteString) #endif initFinish onFinish $ const $ do @@ -906,7 +926,7 @@ data Closure= Closure IdClosure -checkLog mlog = Transient $ do +execLog mlog = Transient $ do case mlog of -- !> ("RECEIVED ", mlog ) of SError e -> do runTrans $ finish $ Just e @@ -950,10 +970,10 @@ listen node = onAll $ do addNodes [node] events <- liftIO $ newIORef M.empty - let conn= (defConnection 8192){myNode= node,comEvent=events}-- Connection node Nothing 8192 events (unsafePerformIO $ newMVar ()) False False 0 :: Connection xxxxxx + let conn= (defConnection 8192){myNode=node,comEvent=events} setData conn r <- listenResponses - checkLog r + execLog r #endif type Pool= [Connection] @@ -969,10 +989,16 @@ emptyPool :: MonadIO m => m (MVar Pool) emptyPool= liftIO $ newMVar [] -createNode :: HostName -> Integer -> [Service] -> Node -createNode h p svs= Node h ( fromInteger p) (unsafePerformIO emptyPool) svs +createNodeServ :: HostName -> Integer -> [Service] -> Node +createNodeServ h p svs= Node h ( fromInteger p) (unsafePerformIO emptyPool) svs + + + +createNode :: HostName -> Integer -> Node +createNode h p= createNodeServ h p [] + createWebNode :: Node createWebNode= Node "webnode" ( fromInteger 0) (unsafePerformIO emptyPool) [("webnode","")] @@ -986,6 +1012,7 @@ show (Node h p _ servs )= show (h,p, servs) + instance Read Node where readsPrec _ s= @@ -998,6 +1025,18 @@ empty= unsafePerformIO emptyPool +-- #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) @@ -1020,48 +1059,29 @@ -errorMyNode f= error $ f ++ ": Node not set. Use initConnData before listen" +errorMyNode f= error $ f ++ ": Node not set. initialize it with connect, listen, initNode..." --- #ifdef ghcjs_HOST_OS ---getMyNode :: TransIO Node ---getMyNode= return createWebNode --- -----initConnData :: Node -> TransIO () -----initConnData node= do ----- addNodes [node] ----- events <- newEVar ----- let conn= Connection () Nothing 8192 events (unsafePerformIO $ newMVar ()) False False 0 :: Connection ----- setData conn --- #else -getMyNode :: TransIO Node +getMyNode :: TransIO Node -- (MonadIO m, MonadState EventF m) => m Node getMyNode = do - Connection{myNode=node} <- getSData <|> errorMyNode "getMyNode" - + Connection{myNode= node} <- getSData <|> errorMyNode "getMyNode" :: TransIO Connection return node ---initConnData :: Node -> TransIO () ---initConnData node= do --- addNodes [node] --- events <- newEVar --- let conn= Connection node Nothing 8192 events (unsafePerformIO $ newMVar ()) False False 0 :: Connection --- setData conn --- return $ Just () - - -- | return the list of nodes connected to the local node getNodes :: MonadIO m => m [Node] getNodes = liftIO $ atomically $ readTVar nodeList -- | add nodes to the list of nodes -addNodes :: (MonadIO m, MonadState EventF m) => [Node] -> m () +addNodes :: [Node] -> TransIO () -- (MonadIO m, MonadState EventF m) => [Node] -> m () addNodes nodes= do + my <- getMyNode -- mynode must be first liftIO . atomically $ do prevnodes <- readTVar nodeList - writeTVar nodeList $ nub $ nodes ++ prevnodes + writeTVar nodeList $ my: (( nub $ nodes ++ prevnodes) \\[my]) + -- | set the list of nodes setNodes nodes= liftIO $ atomically $ writeTVar nodeList $ nodes @@ -1122,57 +1142,98 @@ -- | set the rest of the computation as a new node (first parameter) and connect it -- to an existing node (second parameter). then it uses `connect`` to synchronize the list of nodes + connect :: Node -> Node -> Cloud () + +#ifndef ghcjs_HOST_OS connect node remotenode = do listen node <|> return () -- listen1 node remotenode connect' remotenode -- | synchronize the list of nodes with a remote node and all the nodes connected to it --- the final effect is that all the nodes share the same list of nodes +-- the final effect is that all the nodes reachable share the same list of nodes connect' remotenode= do nodes <- local $ getNodes 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 --- conn <- onAll getSData <|> error "connect: no connnection data" - let nodeConnecting= head nodes - mclustered . local . addNodes $ nodes -- nodeConnecting{connection= unsafePerformIO (newMVar [conn])} : tail nodes - local . onFinish . const $ do - liftIO $ putStrLn "removing node: ">> print nodeConnecting - nodes <- getNodes - setNodes $ nodes \\ [nodeConnecting] + newNodes <- runAt remotenode $ do + local $ do + conn@(Connection _(Just (Node2Node _ _ _)) _ _ _ _ _) <- getSData <|> + error ("connect': need to be connected to a node: use wormhole/connect/listen") + let nodeConnecting= head nodes + liftIO $ modifyMVar_ (connection nodeConnecting) $ const $ return [conn] - local $ do - liftIO $ putStrLn "New nodes: " >> print nodes - n <- getNodes - my <- getMyNode - return $ n \\ [my] - let newNodes = remotenode : nodes' + onFinish . const $ do + liftIO $ putStrLn "removing node: ">> print nodeConnecting + nodes <- getNodes + setNodes $ nodes \\ [nodeConnecting] + +-- renameMyNode remotenode + return nodes + + mclustered . local . addNodes $ nodes + + + + local $ do + allNodes <- getNodes + liftIO $ putStrLn "Known nodes: " >> print allNodes + return allNodes + let n = newNodes \\ nodes when (not $ null n) $ mclustered $ local $ do - liftIO $ putStrLn "New nodes: " >> print n - addNodes n -- add the new discovered nodes to the already known ones. + liftIO $ putStrLn "New nodes: " >> print n + addNodes n -- add the new discovered nodes + local $ do - addNodes nodes - nodes <- getNodes - liftIO $ putStrLn "Known nodes: " >> print nodes + addNodes nodes + nodes <- getNodes + liftIO $ putStrLn "Known nodes: " >> print nodes +-- where +-- renameMyNode new= do +-- con <- getSData <|> error "connection not set. please initialize it" +-- mynode <- liftIO $ readIORef $ myNode con +-- liftIO $ writeIORef (myNode con) new +-- +-- nodes <- getNodes !> ("renaming", mynode, new) +-- setNodes $ new:(nodes \\[mynode]) +#else +connect _ _= empty +connect' _ = empty +#endif + -------------------------------------------- + #ifndef ghcjs_HOST_OS + +readFrom :: Socket -- ^ Connected socket + -> IO BS.ByteString -- ^ Data received +readFrom sock = loop where +-- do +-- s <- SBS.recv sock 40980 +-- BLC.Chunk <$> return s <*> return BLC.Empty + loop = unsafeInterleaveIO $ do + s <- SBS.recv sock 4098 + if BC.null s + then return BLC.Empty -- !> "EMPTY SOCK" + else BLC.Chunk s `liftM` loop + +toStrict= B.concat . BS.toChunks + httpMode (method,uri, headers) conn = do - return () -- !> ("HTTP request",method,uri, headers) +-- return () !> ("HTTP request",method,uri, headers) 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) + bs <- SBS.recv conn 4096 -- readFrom conn + 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) @@ -1180,7 +1241,7 @@ let pc = WS.PendingConnection { WS.pendingOptions = WS.defaultConnectionOptions - , WS.pendingRequest = NWS.RequestHead uri headers False -- RequestHead (BC.pack $ show uri) + , WS.pendingRequest = NWS.RequestHead uri headers False -- RequestHead (BC.pack $ show uri) -- (map parseh headers) False , WS.pendingOnAccept = \_ -> return () , WS.pendingStream = stream @@ -1220,17 +1281,18 @@ data ParseContext a = IsString a => ParseContext (IO a) a deriving Typeable -giveData s= do - r <- SBS.recv s 4096 - return r -- !> ( "giveData ", r) +--giveData s= do +-- r <- readFrom s -- 80000 +-- return r -- !> ( "giveData ", r) receiveHTTPHead s = do - return () - setData $ (ParseContext (giveData s) "" ::ParseContext ByteString) + input <- liftIO $ SBSL.getContents s + setData $ (ParseContext (error "request truncated. Maybe the browser program does not match the server one. \nRecompile the program again with ghcjs <prog> -o static/out") input + ::ParseContext BS.ByteString) (method, uri, vers) <- (,,) <$> getMethod <*> getUri <*> getVers headers <- many $ (,) <$> (mk <$> getParam) <*> getParamValue -- !> (method, uri, vers) - return (method, uri, headers) -- !> (method, uri, headers) + return (method, toStrict uri, headers) -- !> (method, uri, headers) where @@ -1240,37 +1302,39 @@ getParam= do dropSpaces r <- tTakeWhile (\x -> x /= ':' && not (endline x)) - if BC.null r || r=="\r" then empty else dropChar >> return r + if BS.null r || r=="\r" then empty else dropChar >> return(toStrict r) - getParamValue= dropSpaces >> tTakeWhile (\x -> not (endline x)) + getParamValue= toStrict <$> ( dropSpaces >> tTakeWhile (\x -> not (endline x))) -dropSpaces= parse $ \str ->((),BC.dropWhile isSpace str) +dropSpaces= parse $ \str ->((),BS.dropWhile isSpace str) -dropChar= parse $ \r -> ((), BC.tail r) +dropChar= parse $ \r -> ((), BS.tail r) endline c= c== '\n' || c =='\r' --tGetLine= tTakeWhile . not . endline +readStream :: Read a => BS.ByteString -> [StreamData a] +readStream s= readStream1 $ BS.unpack s + where + readStream1 s= + let [(x,r)] = reads s + in x : readStream1 r + + + parallelReadHandler :: Loggable a => TransIO (StreamData a) parallelReadHandler= do ParseContext readit str <- getSData <|> error "parallelReadHandler: ParseContext not found" - :: (TransIO (ParseContext ByteString)) - rbuff <- liftIO $ newIORef str - r <- killOnFinish $ parallel $ do - buff <- readIORef rbuff - str3 <- if buff /= mempty then return buff - else readit - let [(v,left)] = readsPrec' 0 $ BC.unpack str3 - writeIORef rbuff $ BC.pack left - return v - left <- liftIO $ readIORef rbuff - setData $ ParseContext readit left + :: (TransIO (ParseContext BS.ByteString)) + r <- killOnFinish $ choose $ readStream str + + return r --- !> ("readHandler",r) +-- !> ("read",r) -- !> "<-------<----------<--------<----------" @@ -1278,13 +1342,15 @@ dropSpaces tTakeWhile (not . isSpace) -tTakeWhile :: (Char -> Bool) -> TransIO BC.ByteString -tTakeWhile cond= parse (BC.span cond) +tTakeWhile :: (Char -> Bool) -> TransIO BS.ByteString +tTakeWhile cond= parse (BS.span cond) -parse :: Monoid b => (ByteString -> (b,ByteString)) -> TransIO b +parse :: Monoid b => (BS.ByteString -> (b, BS.ByteString)) -> TransIO b parse split= do - ParseContext readit str <- getSData <|> error "parse: ParseContext not found" + ParseContext readit str <- getSData + <|> error "parse: ParseContext not found" + :: TransIO (ParseContext BS.ByteString) if str == mempty then do str3 <- liftIO readit @@ -1292,8 +1358,8 @@ setData $ ParseContext readit str3 -- !> str3 if str3== mempty then empty else parse split - else if BC.take 2 str =="\n\n" then do setData $ ParseContext readit (BC.drop 2 str) ; empty - else if BC.take 4 str== "\r\n\r\n" then do setData $ ParseContext readit (BC.drop 4 str) ; empty + else if BS.take 2 str =="\n\n" then do setData $ ParseContext readit (BS.drop 2 str) ; empty + else if BS.take 4 str== "\r\n\r\n" then do setData $ ParseContext readit (BS.drop 4 str) ; empty else do let (ret,str3) = split str
src/Transient/Move/Services.hs view
@@ -17,9 +17,9 @@ import Transient.Base import Transient.Move import Transient.Logged(Loggable(..)) -import Transient.Internals(RemoteStatus(..)) +import Transient.Internals(RemoteStatus(..), Log(..)) import Transient.Move.Utils -import Transient.Internals(Log(..)) + import Transient.EVars import Transient.Indeterminism import Control.Monad.IO.Class @@ -51,7 +51,7 @@ -- createProcess $ shell prog -pathExe package program port= package++"/dist/build/"++package++"/"++program +pathExe package program port= {-"./"++package++"/dist/build/"++package++"/"++ -} program ++ " -p start/" ++ show port install :: String -> String -> Int -> Cloud () @@ -61,19 +61,18 @@ exist <- local $ liftIO $ doesDirectoryExist packagename when (not exist) $ local $ liftIO $ do callProcess "git" ["clone",package] - liftIO $ print "GIT DONE" + liftIO $ putStr package >> putStrLn " cloned" setCurrentDirectory packagename callProcess "cabal" ["install","--force-reinstalls"] setCurrentDirectory ".." return() let prog = pathExe packagename program port - lliftIO $ print prog - local $ liftIO $ do - createProcess $ shell program - return () + lliftIO $ print $ "executing "++ prog + local $ (async $ do createProcess $ shell prog ; return ()) <|> return () - return() + return() -- !> "INSTALLED" + name url= slash . slash . slash $ slash url where slash= tail1 . dropWhile (/='/') @@ -97,10 +96,10 @@ if yn then do port <- onAll freePort install package program port - nodeService thisNode port + nodeService thisNode port !> (thisNode,port) else empty local $ addNodes nodes - return $ head nodes + return $ head nodes -- !> ("GENERATED NODE", nodes) where nodeService (Node h _ _ _) port= local $ return [Node h port (unsafePerformIO $ newMVar []) [service] ] @@ -146,6 +145,7 @@ => String -> Service -> a -> Cloud b callService ident service params = do node <- initService ident service + localIO $ print node log <- onAll $ do log <- getSData <|> return emptyLog setData emptyLog @@ -177,7 +177,7 @@ runEmbeddedService :: (Loggable a, Loggable b) => Service -> (a -> Cloud b) -> Cloud b runEmbeddedService servname serv = do port <- lliftIO $ freePort - listen $ createNode "localhost" (fromIntegral port) [servname] + listen $ createNodeServ "localhost" (fromIntegral port) [servname] wormhole notused $ loggedc $ do x <- local $ return notused r <- onAll $ runCloud (serv x) <** setData WasRemote @@ -191,28 +191,26 @@ runService :: (Loggable a, Loggable b) => Service -> (a -> Cloud b) -> Cloud b runService servname serv = do - initNode [servname] - wormhole notused $ loggedc $ do - x <- local $ return notused + initNodeServ [servname] + wormhole (notused 1) $ loggedc $ do + x <- local $ return $ notused 2 r <- onAll $ runCloud (serv x) <** setData WasRemote local $ return r teleport return r where - notused= error "runService: variable should not be used" - initNode servs=do - port <- local getPort - let conn= defConnection 8192 - mynode= createNode "localhost" port servs + notused n= error $ "runService: "++ show (n::Int) ++ " variable should not be used" + initNodeServ servs=do + mynode <- local $ do + port <- getPort + return $ createNodeServ "localhost" port servs - listen mynode <|> return() + listen mynode -- <|> return() where getPort :: TransIO Integer - getPort = - if isBrowserInstance then return 0 else do - oneThread $ option "start" "re/start" - port <- input (const True) "port to listen? " - liftIO $ putStrLn "node started" - return port + getPort = if isBrowserInstance then return 0 else do + oneThread $ option "start" "re/start node" + input (const True) "port to listen? " +
src/Transient/Move/Utils.hs view
@@ -13,13 +13,15 @@ ----------------------------------------------------------------------------- module Transient.Move.Utils (initNode,inputNodes, simpleWebApp, initWebApp -, onServer, onBrowser, runNodes) +, onServer, onBrowser, runTestNodes) where import Transient.Base +--import Transient.Internals((!>)) import Transient.Move import Control.Applicative import Control.Monad.IO.Class +import Data.IORef -- | 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. @@ -48,18 +50,18 @@ -- initNode app= do - port <- getPort - initWebApp port app + node <- getPort + initWebApp node app where - getPort :: TransIO Integer + getPort :: TransIO Node getPort = - if isBrowserInstance then return 0 else do - oneThread $ option "start" "re/start" + if isBrowserInstance then return 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 $ putStrLn "node started" - return port + return $ createNode host port -- | ask for nodes to be added to the list of known nodes. it also ask to connect to the node to get -- his list of known nodes @@ -74,7 +76,7 @@ port <- local $ input (const True) "port?" connectit <- local $ input (\x -> x=="y" || x== "n") "connect to get his list of nodes?" - let nnode= createNode host port [] + let nnode= createNode host port if connectit== "y" then connect' nnode else local $ addNodes [nnode] empty @@ -98,14 +100,14 @@ -- -- simpleWebApp :: Integer -> Cloud () -> IO () -simpleWebApp port app = keep $ initWebApp port app +simpleWebApp port app = keep $ initWebApp (createNode "localhost" port) app -- | use this instead of smpleWebApp when you have to do some initializations in the server prior to the --- initialization of the web server. Otherwise, the behaviour is the same. -initWebApp :: Integer -> Cloud () -> TransIO () -initWebApp port app= do +-- initialization of the web server +initWebApp :: Node -> Cloud () -> TransIO () +initWebApp node app= do let conn= defConnection 8192 - setData $ conn{myNode= createNode "localhost" port []} + setData conn{myNode = node} serverNode <- getWebServerNode :: TransIO Node let mynode = if isBrowserInstance @@ -114,7 +116,7 @@ runCloud $ do listen mynode <|> return() - wormhole serverNode app + wormhole serverNode app -- !> ("servernode", serverNode) return () -- only execute if the the program is executing in the browser. The code inside can contain calls to the server. @@ -132,8 +134,7 @@ -- | 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. -runNodes ports= do - let nodes= map (\p -> createNode "localhost" p []) ports - onAll $ addNodes nodes +runTestNodes ports= do + let nodes= map (\p -> createNode "localhost" p) ports foldl (<|>) empty (map listen nodes) <|> return()
transient-universe.cabal view
@@ -1,5 +1,5 @@ name: transient-universe -version: 0.3.1.2 +version: 0.3.2 cabal-version: >=1.10 build-type: Simple license: MIT @@ -43,6 +43,7 @@ , network-uri -any , vector -any , websockets -any + , iproute -any exposed-modules: Transient.Move