transient-universe 0.4.5 → 0.4.6
raw patch · 9 files changed
+161/−91 lines, 9 filesdep ~transientPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependency ranges changed: transient
API changes (from Hackage documentation)
- Transient.Move.Services: catchc :: Exception e => Cloud a -> (e -> Cloud a) -> Cloud a
Files
- README.md +3/−8
- app/server/Transient/Move/Services/MonitorService.hs +3/−3
- src/Transient/MapReduce.hs +2/−2
- src/Transient/Move.hs +76/−26
- src/Transient/Move/Internals.hs +49/−33
- src/Transient/Move/Services.hs +8/−8
- src/Transient/Move/Utils.hs +4/−4
- tests/TestSuite.hs +11/−2
- transient-universe.cabal +5/−5
README.md view
@@ -1,4 +1,4 @@- + ========= [](http://hackage.haskell.org/package/transient-universe) @@ -24,13 +24,13 @@ ------- Browser nodes can integrate Hplayground for ghcjs, a reactive client side library based in trasient (package ghcjs-hplay) they can create widgets with HTML form elements and control the server nodes. A computation can move from browser to server and back at runtime despite the different architecture. -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. +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. These features can make transient ideal for client as well as server side-driven applications, whenever distribution and push-driven reactivity is necessary either in the servers or in the browser clients. New === -The last release add +The last release add - Hooks for secure communications: with [transient-universe-tls package](https://github.com/transient-haskell/transient-universe-tls), a node can use TLS to connect with other nodes, including web nodes. If the connection of a web node is initiated with "https" the websocket connection uses secure communications (wss). The only primitive added is `initTLS`. - Client websocket connections to connect with nodes within firewalled servers: a server node can connect with another situated after a HTTP server. All the process is transparent and add no new primitive; First `connect` tries a TCP socket connection if it receives other message than "OK", it tries a connection as a websocket client. This is important for P2P connections where a central server acts as coordinator. websocket connections can use TLS communications too. @@ -98,8 +98,3 @@ -I plan to improve map-reduce to create a viable platform for serious data analysis and machine learning using haskell. It will have a web notebook running in the browser. -Create services and examples for general Web applications with distributed servers and create services for them - - - - -
app/server/Transient/Move/Services/MonitorService.hs view
@@ -56,7 +56,7 @@ rs <- callNodes' nodes1 (<>) mempty (installHere ident service (pernode+1)) <> callNodes' nodes2 (<>) mempty (installHere ident service pernode) local $ addNodes rs - return rs + return rs !> ("MONITOR RETURN---------------------------------->", rs) -- installIt = installHere ident service <|> installThere ident service installHere :: String -> Service -> Int -> Cloud [ Node] @@ -72,7 +72,7 @@ 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 + `catcht` \(e :: SomeException) -> liftIO (putStr "INSTALLLLLLLLLLLLLLL2222222: " >> print e) >> empty @@ -84,7 +84,7 @@ install service port= do return () !> "IIIIIIIIIIIIIIINSTALL" - install' `catcht` \(e :: SomeException) -> liftIO (putStr "INSTALL error: " >> print e) >> empty + install' `catcht` \(e :: SomeException) -> liftIO (putStr "INSTALL error: " >> print e) >> empty where install'= do let host= "localhost"
src/Transient/MapReduce.hs view
@@ -32,7 +32,7 @@ #else -import Transient.Internals +import Transient.Internals hiding (Ref) import Transient.Move.Internals hiding (pack) import Transient.Indeterminism @@ -216,7 +216,7 @@ foldAndSend node nodes ref= do pairs <- onAll $ getPartitionData1 ref - <|> return (error $ "DDS computed out of his node:"++ show ref) + <|> return (error $ "DDS computed out of his node:"++ show ref ) let mpairs = groupByDestiny pairs length <- local . return $ M.size mpairs
src/Transient/Move.hs view
@@ -2,41 +2,89 @@ -- -- Module : Transient.Move -- Copyright : --- License : GPL-3 +-- License : MIT -- -- Maintainer : agocorona@gmail.com -- Stability : -- Portability : -- --- | see <https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv> +-- | @transient-universe@ extends the seamless composability of concurrent +-- multi-threaded programs provided by +-- <https://github.com/transient-haskell/transient transient> +-- to a multi-node cloud. Distributed concurrent programs are created and +-- composed seamlessly and effortlessly as if they were written for a single +-- node. @transient-universe@ has diverse applications from simple distributed +-- applications to massively parallel and distributed map-reduce problems. If +-- you are considering Apache Spark or Cloud Haskell then transient might be a +-- simpler yet better solution for you. +-- +-- Transient makes it easy to write composable, distributed event driven +-- reactive UI applications with client side and server side code composed +-- freely in the same application. For example, +-- <https://hackage.haskell.org/package/axiom Axiom> is a transient based +-- unified client and server side web application framework that provides a +-- better programming model and composability compared to frameworks like +-- ReactJS. +-- +-- = Overview +-- +-- The 'Cloud' monad adds the following facilities to complement the 'TransIO' +-- monad: +-- +-- * Create a distributed compute cluster of nodes +-- * Move computations across nodes at any point during computation +-- * Run computations on multiple nodes in parallel +-- +-- = Further Reading +-- +-- * <https://github.com/transient-haskell/transient/wiki/Transient-tutorial Tutorial> +-- * <https://github.com/transient-haskell/transient-examples Examples> +-- * <https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv Blog post> +-- ----------------------------------------------------------------------------- {-# LANGUAGE CPP #-} module Transient.Move( --- * running the Cloud monad -Cloud(..),runCloud, runCloudIO, runCloudIO', local, onAll, lazy, loggedc, lliftIO,localIO, -listen, Transient.Move.Internals.connect, connect', fullStop, +-- * Running the Monad +Cloud(..),runCloud, runCloudIO, runCloudIO', --- * primitives for communication -wormhole, teleport, copyData, +-- * Node & Cluster Management +-- $cluster +Node(..), +-- ** Creating nodes +Service(), createNodeServ, createNode, createWebNode, +-- ** Joining the cluster +Transient.Move.Internals.connect, connect', listen, +-- Low level APIs +addNodes, shuffleNodes, +Connection(..), ConnectionData(..), defConnection, +-- ** Querying nodes +getMyNode, getWebServerNode, getNodes, nodeList, isBrowserInstance, --- * single node invocation + +-- * Running Local Computations +local, onAll, lazy, loggedc, lliftIO, localIO, fullStop, + +-- * Moving Computations +wormhole, teleport, copyData, + +-- * Running at a Remote Node beamTo, forkTo, callTo, runAt, atRemote, --- * invocation of many nodes +-- * Running at Multiple Nodes clustered, mclustered, callNodes, --- * messaging +-- * Messaging putMailbox, putMailbox',getMailbox,getMailbox',cleanMailbox,cleanMailbox', --- * thread control +-- * Thread Control single, unique, #ifndef ghcjs_HOST_OS --- * buffering control +-- * Buffering Control setBuffSize, getBuffSize, #endif @@ -44,17 +92,9 @@ -- * REST API api, HTTPMethod(..), PostParams, #endif --- * node management -createNode, createWebNode, createNodeServ, getMyNode, getNodes, -addNodes, shuffleNodes, --- * low level - - getWebServerNode, Node(..), nodeList, Connection(..), Service(), - isBrowserInstance, +-- * Low Level APIs - defConnection, - ConnectionData(..), #ifndef ghcjs_HOST_OS ParseContext(..) #endif @@ -63,8 +103,18 @@ import Transient.Move.Internals - - - - - +-- $cluster +-- +-- To join the cluster a node 'connect's to a well known node already part of +-- the cluster. +-- +-- @ +-- import Transient.Move (runCloudIO, lliftIO, createNode, connect, getNodes, onAll) +-- +-- main = runCloudIO $ do +-- this <- lliftIO (createNode "192.168.1.2" 8000) +-- master <- lliftIO (createNode "192.168.1.1" 8000) +-- connect this master +-- onAll getNodes >>= lliftIO . putStrLn . show +-- @ +--
src/Transient/Move/Internals.hs view
@@ -126,6 +126,7 @@ -- | 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 @@ -177,11 +178,12 @@ --stream= Cloud . transport -- #ifndef ghcjs_HOST_OS --- | run the cloud computation. +-- | 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 the cloud computation with no console input +-- | 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 @@ -343,10 +345,11 @@ wormhole node (Cloud comp) = local $ Transient $ do moldconn <- getData :: StateIO (Maybe Connection) mclosure <- getData :: StateIO (Maybe Closure) - labelState $ "wormhole" ++ show node - Log rec _ _ <- getData `onNothing` return (Log False [][]) + -- labelState $ "wormhole" ++ show node + logdata@(Log rec log fulLog) <- getData `onNothing` return (Log False [][]) + if not rec then runTrans $ (do @@ -384,7 +387,7 @@ teleport = do local $ Transient $ do cont <- get - labelState "teleport" + -- labelState "teleport" -- 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) @@ -407,6 +410,7 @@ #else do #endif + --read this Closure Closure closRemote <- getData `onNothing` return (Closure 0 ) @@ -422,9 +426,8 @@ liftIO $ modifyMVar_ closures $ \map -> return $ M.insert closLocal (fulLog,cont) map let tosend= reverse $ if closRemote==0 then fulLog else log - - - runTrans $ msend conn $ SMore $ ClosureData closRemote closLocal tosend + + runTrans $ do msend conn $ SMore $ ClosureData closRemote closLocal tosend !> ("teleport sending", SMore (closRemote,closLocal,tosend)) !> "--------->------>---------->" @@ -582,8 +585,8 @@ #ifndef ghcjs_HOST_OS -msend (Connection _ _ (Just (Node2Node _ sock _)) _ _ blocked _ _ _) r= - liftIO $ withMVar blocked $ const $ SBS.sendAll sock $ BC.pack (show r) -- !> "N2N SEND" +msend (Connection _ _ (Just (Node2Node _ sock _)) _ _ blocked _ _ _) r=do + liftIO $ withMVar blocked $ const $ SBS.sendAll sock $ BC.pack (show r) !> "N2N SEND" msend (Connection _ _ (Just (TLSNode2Node ctx)) _ _ _ _ _ _) r= liftIO $ sendTLSData ctx $ BS.pack (show r) -- !> "TLS SEND" @@ -1070,6 +1073,8 @@ +-- | Setup the node to start listening for incoming connections. +-- listen :: Node -> Cloud () listen (node@(Node _ port _ _ )) = onAll $ do addThreads 1 @@ -1092,7 +1097,7 @@ case mlog of SMore (RelayMSG _ _ _) ->relay mlog _ -> execLog mlog - `catcht` (\(e ::SomeException) -> liftIO (print e)) + `catcht` (\(e ::SomeException) -> liftIO $ print e) -- relayService :: TransIO () @@ -1182,9 +1187,11 @@ input <- liftIO $ SBSL.getContents sock cutExceptions + onException $ \(e :: SomeException) -> do - cutExceptions + -- cutExceptions liftIO $ putStr "listen: " >> print e + let Connection{remoteNode=rnode,closures=closures,closChildren= rmap} = conn -- TODO How to close Connection by discriminating exceptions mnode <- liftIO $ readIORef rnode @@ -1198,7 +1205,9 @@ modifyMVar_ closures $ const $ return M.empty writeIORef rmap M.empty -- topState >>= showThreads + -- cutExceptions killBranch + setData $ (ParseContext (NS.close sock >> error "Communication error" ) input ::ParseContext BS.ByteString) @@ -1258,8 +1267,7 @@ onException $ \(e :: SomeException) -> do cutExceptions liftIO $ putStr "listen websocket:" >> print e - continue - liftIO $ mclose conn' + --liftIO $ mclose conn' killBranch empty -- async (return (SMore (0,0,[Exec]))) <|> do @@ -1663,7 +1671,7 @@ isBrowserInstance= True api _= empty #else --- | True if it is running in the browser +-- | Returns 'True' if we are running in the browser. isBrowserInstance= False #endif @@ -1677,6 +1685,8 @@ 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 @@ -1722,6 +1732,7 @@ 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 @@ -1729,7 +1740,7 @@ --- | return the list of nodes connected to the local node +-- | Return the list of nodes in the cluster. getNodes :: MonadIO m => m [Node] getNodes = liftIO $ atomically $ readTVar nodeList @@ -1746,7 +1757,7 @@ nodes <- getNodes return $ map (\n -> filter f $ nodeServices n) nodes --- | add nodes to the list of 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 @@ -1764,6 +1775,8 @@ -- | 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 @@ -1806,8 +1819,10 @@ --- | set the rest of the computation as the code of a new node (first parameter) and connect it --- to an existing node (second parameter). then it uses `connect`` to synchronize the list of nodes +-- | 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 @@ -1816,27 +1831,28 @@ --- | synchronize the list of nodes with a remote node and all the nodes connected to it --- the final effect is that all the nodes reachable share the same list of nodes +-- | Reconcile the list of nodes in the cluster using a remote node already +-- part of the cluster. Reconciliation results in each node in the cluster +-- having exactly the same list of nodes. connect' :: Node -> Cloud () -connect' remotenode= do +connect' remotenode= loggedc $ do nodes <- local getNodes localIO $ putStr "connecting to: " >> print remotenode - newNodes <- runAt remotenode $ interchange remotenode nodes + newNodes <- runAt remotenode $ interchange nodes - -- local $ return () !> "interchange finish" + 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 + liftIO $ putStr "New nodes: " >> print toAdd !> "NEWNODES" addNodes toAdd where -- receive new nodes and send their own - interchange remotemode nodes= + interchange nodes= do newNodes <- local $ do conn@Connection{remoteNode=rnode, connData=Just cdata} <- getSData <|> @@ -1856,12 +1872,12 @@ 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] + -- onException $ \(e :: SomeException) -> do + -- liftIO $ putStr "connect:" >> print e + -- liftIO $ putStrLn "removing node: " >> print callingNode + -- -- topState >>= showThreads + -- nodes <- getNodes + -- setNodes $ nodes \\ [callingNode] return newNodes @@ -1869,7 +1885,7 @@ mclustered . local $ do - liftIO $ putStrLn "New nodes1111: " >> print newNodes + liftIO $ putStrLn "New nodes: " >> print newNodes addNodes newNodes
src/Transient/Move/Services.hs view
@@ -2,7 +2,7 @@ -- -- Module : Transient.Move.Services -- Copyright : --- License : GPL-3 +-- License : MIT -- -- Maintainer : agocorona@gmail.com -- Stability : @@ -61,7 +61,7 @@ requestInstance :: String -> Service -> Int -> Cloud [Node] requestInstance ident service num= loggedc $ do return () !> "requestInstance" - local $ onException $ \(e:: ConnectionError) -> startMonitor >> continue !> ("Exception",e) + local $ onException $ \(e:: ConnectionError) -> startMonitor >> continue !> ("EXCEPTIOOOOOOOOOOON",e) nodes <- callService' ident monitorNode (ident,service,num) local $ addNodes nodes -- !> ("ADDNODES",service) @@ -165,11 +165,12 @@ callService' ident node params = do log <- onAll $ do log <- getSData <|> return emptyLog - setData emptyLog + setData emptyLog return log r <- wormhole node $ do local $ return params + teleport -- local empty `asTypeOf` typea params local empty @@ -189,8 +190,8 @@ emptyLog= Log False [] [] -catchc :: Exception e => Cloud a -> (e -> Cloud a) -> Cloud a -catchc a b= Cloud $ catcht (runCloud' a) (\e -> runCloud' $ b e) +-- catchc :: Exception e => Cloud a -> (e -> Cloud a) -> Cloud a +-- catchc a b= Cloud $ catcht (runCloud' a) (\e -> runCloud' $ b e) runEmbeddedService :: (Loggable a, Loggable b) => Service -> (a -> Cloud b) -> Cloud b runEmbeddedService servname serv = do @@ -218,7 +219,6 @@ service= wormhole (notused 1) $ do x <- local . return $ notused 2 - -- setData emptyLog r <- local $ runCloud (serv x) -- <** setData WasRemote setData emptyLog local $ return r @@ -241,8 +241,8 @@ getNode :: TransIO Node getNode = if isBrowserInstance then liftIO createWebNode else do oneThread $ option "start" "re/start node" - host <- input (const True) "hostname of this node (must be reachable) (\"localhost\"): " - port <- input (const True) "port to listen? (3000) " + 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 inputNodes= do
src/Transient/Move/Utils.hs view
@@ -2,7 +2,7 @@ -- -- Module : Transient.Move.Utils -- Copyright : --- License : GPL-3 +-- License : MIT -- -- Maintainer : agocorona@gmail.com -- Stability : @@ -62,8 +62,8 @@ 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? " + 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 @@ -92,7 +92,7 @@ where addNew= do - local $ option "add" "add a new node" + local $ oneThread $ option "add" "add a new node" host <- local $ do r <- input (const True) "Hostname of the node (none): "
tests/TestSuite.hs view
@@ -49,9 +49,18 @@ -- local $ option "get" "get instances" + + localIO $ return () !> "REQUEST INSTANCE " - [node1,node2] <- requestInstance "PIN1" service 2 - + [node1, node2] <- requestInstance "PIN1" service 2 + -- localIO $ putStrLn "------checking connect --------" + -- connect' node1 + -- connect' node2 + -- runAt node1 $ local $ do + -- nodes <- getNodes + -- assert (length nodes== 3) return () + + local ( option "f" "fire") <|> return "" -- to repeat the test, remove exit
transient-universe.cabal view
@@ -1,5 +1,5 @@ name: transient-universe -version: 0.4.5 +version: 0.4.6 cabal-version: >=1.10 build-type: Simple license: MIT @@ -62,7 +62,7 @@ stm -any, time -any, transformers -any, - transient >= 0.5.6 + transient >= 0.5.8 default-language: Haskell2010 hs-source-dirs: src . @@ -71,7 +71,7 @@ if !impl(ghcjs >=0.1) build-depends: transformers -any, - transient >=0.5.6, + transient >=0.5.8, transient-universe, process, directory @@ -91,7 +91,7 @@ if !impl(ghcjs >=0.1) build-depends: mtl -any, - transient >= 0.5.6, + transient >= 0.5.8, random -any, text -any, containers -any, @@ -125,7 +125,7 @@ if !impl(ghcjs >=0.1) build-depends: mtl -any, - transient >= 0.5.6, + transient >= 0.5.8, random -any, text -any, containers -any,