diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,105 @@ +![Transient logo](https://raw.githubusercontent.com/transient-haskell/transient/master/logo.png) +========= + +[![Hackage](https://img.shields.io/hackage/v/transient-universe.svg)](http://hackage.haskell.org/package/transient-universe) +[![Stackage LTS](http://stackage.org/package/transient-universe/badge/lts)](http://stackage.org/lts/package/transient-universe) +[![Stackage Nightly](http://stackage.org/package/transient-universe/badge/nightly)](http://stackage.org/nightly/package/transient-universe) +[![Build Status](https://travis-ci.org/agocorona/transient-universe.png?branch=master)](https://travis-ci.org/agocorona/transient-universe) +[![Gitter](https://badges.gitter.im/theam/haskell-do.svg)](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link) + +See the [Wiki](https://github.com/agocorona/transient/wiki) + +transient-universe is the distributed computing extension of [transient](https://github.com/agocorona/transient). It support moving computations between Haskell closures in different computers in the network. Even among different architectures: Linux nodes can work with windows and browser nodes running haskell compiled with [ghcjs](https://github.com/ghcjs/ghcjs). + +The primitives that perform the moving of computations are called `wormhole` and `teleport`, the names express the semantics. Hence the name of the package. + +All the nodes run the same program compiled for different architectures. It defines a Cloud computation (monad). It is a thin layer on top of transient with additional primitives and services that run a single program in one or many nodes. + +Browser integration +================== + +Browser nodes, running transient programs compiled with ghcjs are integrated with server nodes, using websockets for communication. Just compile the program with ghcjs and point the browser to http://server:port. The server nodes have a HTTP server that will send the compiled program to the browser. + +Distributed Browser/server Widgets +------- +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. + +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 + + - Hooks for secure communications + - Client websocket connections to connect with nodes within firewalled servers + - No network traffic when a node invokes itself + +Map-reduce +========== +transient-universe implements map-reduce in the style of [spark](http://spark.apache.org) as a particular case. It is at the same time a hard test of the distributed primitives since it involves a complex choreography of movement of computations. It supports in memory operations and caching. Resilience (restart from the last checkpoint in case of failure) is not implemented but it is foreseen. + +Look at [this article](https://www.schoolofhaskell.com/user/agocorona/estimation-of-using-distributed-computing-streaming-transient-effects-vi-1#distributed-datasets) + +There is a runnable example: [DistrbDataSets.hs](https://github.com/agocorona/transient-universe/blob/master/examples/DistrbDataSets.hs) that you can executed with: + +> runghc ./examples/DistrbDataSets.hs + +It uses a number of simulated nodes to calculate the frequency of words in a long text. + +Services +======== +Services communicate two different transient applications. This allows to divide the running application in different independent tiers. No documentation is available yet. Sorry. + +General distributed primitives +============================= +`teleport` is a primitive that translates computations back and forth reusing an already opened connection. + +The connection is initiated by `wormhole` with another node. This can be done anywhere in a computation without breaking composability. As always, Everything is composable. + +Both primitives support also streaming among nodes in an efficient way. It means that a remote call can return not just a single response, but many of them. + +All the other distributed primitives: `runAt`, `streamFrom` `clustered` etc are rewritten in terms of these two. + +How to run the ghcjs example: +============================= + +See the distributed examples in the [transient-examples](https://github.com/transient-haskell/transient) repository + +See this [video](https://www.livecoding.tv/agocorona/videos/Ke1Qz-seamless-composable-web-programming) to see this example running: + +The test program run among other things, two copies of a widget that start, stop and display a counter that run in the server. + +Documentation +============= + +The [Wiki](https://github.com/agocorona/transient/wiki) is more user oriented + +My video sessions in [livecoding.tv](https://www.livecoding.tv/agocorona/videos/) not intended as tutorials or presentations, but show some of the latest features running. + +The articles are more technical: + +- [Philosophy, async, parallelism, thread control, events, Session state](https://www.fpcomplete.com/user/agocorona/EDSL-for-hard-working-IT-programmers?show=tutorials) +- [Backtracking and undoing IO transactions](https://www.fpcomplete.com/user/agocorona/the-hardworking-programmer-ii-practical-backtracking-to-undo-actions?show=tutorials) +- [Non-deterministic list like processing, multithreading](https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii?show=tutorials) +- [Distributed computing](https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv?show=tutorials) +- [Publish-Subscribe variables](https://www.schoolofhaskell.com/user/agocorona/publish-subscribe-variables-transient-effects-v) +- [Distributed streaming, map-reduce](https://www.schoolofhaskell.com/user/agocorona/estimation-of-using-distributed-computing-streaming-transient-effects-vi-1) + +These articles contain executable examples (not now, since the site no longer support the execution of Haskell snippets). + + + +Future plans +============ +The only way to improve it is using it. Please send me bugs and additional functionalities! + +-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 + + + + + diff --git a/app/client/Transient/Move/Services/MonitorService.hs b/app/client/Transient/Move/Services/MonitorService.hs --- a/app/client/Transient/Move/Services/MonitorService.hs +++ b/app/client/Transient/Move/Services/MonitorService.hs @@ -1,1 +1,1 @@ -main= return() +main= return () diff --git a/app/server/Transient/Move/Services/MonitorService.hs b/app/server/Transient/Move/Services/MonitorService.hs --- a/app/server/Transient/Move/Services/MonitorService.hs +++ b/app/server/Transient/Move/Services/MonitorService.hs @@ -1,65 +1,64 @@ ------------------------------------------------------------------------------ --- --- Module : Transient.Move.Services.MonitorService --- Copyright : --- License : MIT --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | --- ------------------------------------------------------------------------------ - -module Main where - -import Transient.Base - -import Transient.Move -import Transient.Move.Utils -import Transient.Move.Services -import Control.Applicative -import Control.Monad.IO.Class -import Data.List ((\\)) - -main = keep . runCloud $ do - runService monitorService $ \(ident,service) -> do - mnode <- (local $ findInNodes service >>= return . Just . head) <|> - requestInstall ident service - return (mnode :: Maybe Node) - where - installHere ident service@(package,program)= local $ do - thisNode <- getMyNode - yn<- authorizeService ident service -- !> "AUTHORIZE" - if yn - then do - node <- liftIO $ do - port <- freePort - putStr "Monitor: installing " >> putStrLn package - install package program (nodeHost thisNode) port - putStrLn "INSTALLED" - nodeService thisNode port service - addNodes [node] - return $ Just node - else return Nothing - - requestInstall :: String -> Service -> Cloud (Maybe Node) - requestInstall ident service= do - mnode <- installHere ident service -- !> "INSTALLHERE" - case mnode of - Nothing -> installThere ident service - justnode -> return justnode - - installThere ident service= do - nodes <- onAll $ findInNodes monitorService -- !> "installThere" - mynode <- onAll getMyNode -- !> nodes - request $ nodes \\ [mynode] - where - request []= empty - request (n:ns)= do - mnode <- callService' ident n (ident,service) -- !> ("calling",n) - case mnode of - Nothing -> request ns - justnode -> return justnode - +----------------------------------------------------------------------------- +-- +-- Module : Transient.Move.Services.MonitorService +-- Copyright : +-- License : MIT +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +-- +-- | +-- +----------------------------------------------------------------------------- + +module Main where + +import Transient.Base + +import Transient.Move +import Transient.Move.Utils +import Transient.Move.Services +import Control.Applicative +import Control.Monad.IO.Class +import Data.List ((\\)) + +main = keep . runCloud $ do + runService monitorService $ \(ident,service) -> do + mnode <- (local $ findInNodes service >>= return . Just . head) <|> + requestInstall ident service + return (mnode :: Maybe Node) + where + installHere ident service@(package,program)= local $ do + thisNode <- getMyNode + yn<- authorizeService ident service -- !> "AUTHORIZE" + if yn + then do + node <- liftIO $ do + port <- freePort + putStr "Monitor: installing " >> putStrLn package + install package program (nodeHost thisNode) port + putStrLn "INSTALLED" + nodeService thisNode port service + addNodes [node] + return $ Just node + else return Nothing + + requestInstall :: String -> Service -> Cloud (Maybe Node) + requestInstall ident service= do + mnode <- installHere ident service -- !> "INSTALLHERE" + case mnode of + Nothing -> installThere ident service + justnode -> return justnode + + installThere ident service= do + nodes <- onAll $ findInNodes monitorService -- !> "installThere" + mynode <- onAll getMyNode -- !> nodes + request $ nodes \\ [mynode] + where + request []= empty + request (n:ns)= do + mnode <- callService' ident n (ident,service) -- !> ("calling",n) + case mnode of + Nothing -> request ns + justnode -> return justnode diff --git a/src/Transient/MapReduce.hs b/src/Transient/MapReduce.hs --- a/src/Transient/MapReduce.hs +++ b/src/Transient/MapReduce.hs @@ -1,502 +1,510 @@ -{-# LANGUAGE ExistentialQuantification, DeriveDataTypeable -, FlexibleInstances, MultiParamTypeClasses, OverloadedStrings, CPP #-} - - -module Transient.MapReduce -( -Distributable(..),distribute, getText, -getUrl, getFile,textUrl, textFile, -mapKeyB, mapKeyU, reduce,eval, -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 DDS= DDS -class Distributable -data PartRef a=PartRef a - -#else - -import Transient.Base -import Transient.Internals((!>)) -import Transient.Move hiding (pack) -import Transient.Logged -import Transient.Indeterminism -import Control.Applicative -import System.Random -import Control.Monad.IO.Class - -import System.IO -import Control.Monad -import Data.Monoid -import Data.Maybe - -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 -import Data.TCache.Defs - -import Data.ByteString.Lazy.Char8 (pack,unpack) -import Control.Monad.STM -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 - -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 - - 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 - - - ---instance Show a => Show (MVar a) where --- show mvx= "MVar " ++ show (unsafePerformIO $ readMVar mvx) --- ---instance Read a => Read (MVar a) where --- readsPrec n ('M':'V':'a':'r':' ':s)= --- let [(x,s')]= readsPrec n s --- in [(unsafePerformIO $ newMVar x,s')] - -data ReduceChunk a= EndReduce | Reduce a deriving (Typeable, Read, Show) - -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 - - box <- lliftIO $ return . Text.pack =<< replicateM 10 (randomRIO ('a','z')) - nodes <- local getNodes - - let lengthNodes = length nodes - shuffler nodes = do - ref@(Ref node path sav) <- mx --- return () !> ref - runAt node $ foldAndSend 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 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 - - nsent <- onAll $ liftIO $ newMVar 0 - - (i,folded) <- local $ parallelize foldthem (M.assocs mpairs) - - n <- lliftIO $ modifyMVar nsent $ \r -> return (r+1, r+1) - runAt (nodes !! i) $ local $ putMailbox box $ Reduce folded - - when (n == length) $ sendEnd nodes - - where - count n proc proc2= do - nsent <- onAll $ liftIO $ newMVar 0 - proc - n' <- lliftIO $ modifyMVar nsent $ \r -> return (r+1, r+1) - when (n'==n) proc2 - - foldthem (i,kvs)= async . return - $ (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 - --- reduce :: (Ord k) => Cloud (M.Map k v) - - reduce1 = local $ do - reduceResults <- liftIO $ newMVar M.empty - numberSent <- liftIO $ newMVar 0 - - minput <- getMailbox box -- get the chunk once it arrives to the mailbox - - case minput of - - EndReduce -> do - - n <- liftIO $ modifyMVar numberSent $ \r -> return (r+1, r+1) - - 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) - - 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 <- streamFrom 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 getNodes - 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 getNodes -- !> "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 getNodes -- !> "DISTRIBUTE" - let nodes = filter (not . isWebNode) nodes' - let lnodes = length nodes - - parallelize (process lnodes) $ zip nodes [0..lnodes-1] - where - 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 - 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 getNodes -- !> "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 $ 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 getNodes -- !> "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 <- liftIO $ readFile file - let xs = partitioner content - size= case length xs `div` lnodes of 0 ->1 ; n -> n - xss=Transient.MapReduce.fromList $ take size $ drop (i * size) xs -- !> size - 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 - - - - -#endif +{-# LANGUAGE ExistentialQuantification, DeriveDataTypeable +, FlexibleInstances, MultiParamTypeClasses, OverloadedStrings, CPP #-} + + +module Transient.MapReduce +( +Distributable(..),distribute, getText, +getUrl, getFile,textUrl, textFile, +mapKeyB, mapKeyU, reduce,eval, +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 DDS= DDS +class Distributable +data PartRef a=PartRef a + +#else + +import Transient.Internals + +import Transient.Move hiding (pack) +import Transient.Logged +import Transient.Indeterminism +import Control.Applicative +import System.Random +import Control.Monad.IO.Class + +import System.IO +import Control.Monad +import Data.Monoid +import Data.Maybe + +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 Control.Monad.STM +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 getNodes + + let lengthNodes = length nodes + shuffler nodes = do + + 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,port,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 getNodes + 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 getNodes -- !> "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 getNodes -- !> "getText" + let nodes = filter (not . isWebNode) nodes' + let lnodes = length nodes + + parallelize (process lnodes) $ zip nodes [0..lnodes-1] + where + 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 + 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 getNodes -- !> "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 $ 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 getNodes -- !> "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 <- liftIO $ readFile file + let xs = partitioner content + size= case length xs `div` lnodes of 0 ->1 ; n -> n + xss=Transient.MapReduce.fromList $ take size $ drop (i * size) xs -- !> size + 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 + + + + +#endif diff --git a/src/Transient/Move.hs b/src/Transient/Move.hs --- a/src/Transient/Move.hs +++ b/src/Transient/Move.hs @@ -10,1421 +10,61 @@ -- -- | see ----------------------------------------------------------------------------- -{-# LANGUAGE DeriveDataTypeable , ExistentialQuantification, OverloadedStrings - ,ScopedTypeVariables, StandaloneDeriving, RecordWildCards, FlexibleContexts, CPP - ,GeneralizedNewtypeDeriving #-} -module Transient.Move( - -Cloud(..),runCloudIO, runCloudIO',local,onAll,lazy, loggedc, lliftIO,localIO, -listen, Transient.Move.connect, connect', fullStop, - --- * primitives for communication -wormhole, teleport, copyData, - --- * single node invocation -beamTo, forkTo, streamFrom, callTo, runAt, atRemote, - --- * invocation of many nodes -clustered, mclustered, callNodes, - --- * messaging -newMailbox, putMailbox,getMailbox,cleanMailbox, - --- * thread control -single, - -#ifndef ghcjs_HOST_OS -setBuffSize, getBuffSize, -#endif - --- * node management -createNode, createWebNode, createNodeServ, getMyNode, getNodes, -addNodes, shuffleNodes, - --- * low level - - getWebServerNode, Node(..), nodeList, Connection(..), Service(), - isBrowserInstance, Prefix(..), addPrefix - ,defConnection - - -) where - -import Transient.Internals hiding ((!>)) -import Transient.Logged -import Transient.Indeterminism(choose) -import Transient.Backtrack -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 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(send,sendMany,sendAll,recv) -import qualified Network.Socket.ByteString.Lazy as SBSL -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 qualified Data.Text as T -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, insert) -import Data.IORef - - - -import System.IO - -import Control.Concurrent - -import System.Random - - - -import Data.Dynamic -import Data.String - -import System.Mem.StableName -import Unsafe.Coerce - -#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 :: 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,Applicative,Monoid,Alternative, Monad, MonadState EventF) --- --- ---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 --- 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 the cloud computation. -runCloudIO :: Typeable a => Cloud a -> IO a -runCloudIO (Cloud mx)= keep mx - --- | run the cloud computation with no console input -runCloudIO' :: Typeable a => 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 - -lazy :: TransIO a -> Cloud a -lazy mx= onAll $ getCont >>= \st -> Transient $ - return $ unsafePerformIO $ runStateT (runTrans mx) st >>= return .fst - --- 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 :: 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 - --- | locally perform IO. `localIO = lliftIO` -localIO :: Loggable a => IO a -> Cloud a -localIO= lliftIO - ---remote :: Loggable a => TransIO a -> Cloud a ---remote x= Cloud $ step' x $ \full x -> Transient $ do --- let add= Wormhole: full --- setData $ Log False add add --- --- r <- runTrans x --- --- let add= WaitRemote: full --- (setData $ Log False add add) -- !!> "AFTER STEP" --- return r - --- | stop the current computation and does not execute any alternative computation -fullStop :: Cloud stop -fullStop= onAll $ setData WasRemote >> stop - - --- | 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 = 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 - --- | 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" - r <- Cloud $ runCloud proc <** setData WasRemote - teleport -- !> "teleport 2222" - return r - --- | synonymous of `callTo` -runAt :: Loggable a => Node -> Cloud a -> Cloud a -runAt= callTo - --- | execute a single thread for each connection --- --- > box <- foo --- > r <- runAt node . local . single $ getMailbox box --- > localIO $ print r --- --- if foo would return many values, the above code would monitor one remote mailbox --- each time: the last one entered. --- Without single, it would monitor all of them. - -single :: TransIO a -> TransIO a -single f= do - con@Connection{closChildren=rmap} <- getSData <|> error "single: only works within a wormhole" - mapth <- liftIO $ readIORef rmap - id <- liftIO $ makeStableName f >>= return . hashStableName - - chs <- - let mx = M.lookup id mapth - in case mx of - Just tv -> return tv - Nothing -> liftIO $ newTVarIO [] - - - modify $ \ s -> s{children= chs} -- to allow his own thread control - liftIO $ killChildren chs - f <** do - id <- liftIO $ makeStableName f >>= return . hashStableName - liftIO $ modifyIORef rmap $ \mapth -> M.insert id chs mapth - - -msend :: Loggable a => Connection -> StreamData a -> TransIO () - - -#ifndef ghcjs_HOST_OS - -msend (Connection _(Just (Node2Node _ sock _)) _ _ blocked _ _ _) r= do - r <- liftIO $ do - withMVar blocked $ - const $ do - SBS.send sock $ BC.pack (show r) - return Nothing - `catch` (\(e::SomeException) -> return $ Just e) - case r of - Nothing -> return() - juste -> finish juste - - -msend (Connection _(Just (Node2Web sconn)) _ _ blocked _ _ _) r=liftIO $ - withMVar blocked $ const $ WS.sendTextData sconn $ BS.pack (show r) - - -#else - -msend (Connection _ (Just (Web2Node sconn)) _ _ blocked _ _ _) r= liftIO $ - withMVar blocked $ const $ JavaScript.Web.WebSocket.send (JS.pack $ show r) sconn -- !!> "MSEND SOCKET" - - - -#endif - -msend (Connection _ Nothing _ _ _ _ _ _) _= error "msend out of wormhole context" - -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" - - -{- -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 (setData 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 :: 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 () - -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 _ _ _)) _ _ blocked _ _ _) = parallelReadHandler -- !> "mread" - - -mread (Connection node (Just (Node2Web sconn )) bufSize events blocked _ _ _)= - parallel $ do - s <- WS.receiveData sconn - return . read' $ BS.unpack s --- !> ("WS MREAD RECEIVED ---->", s) - --- `catch`(\(e ::SomeException) -> return $ SError e) - -getWebServerNode :: TransIO Node -getWebServerNode = getMyNode -#endif - -read' s= case readsPrec' 0 s of - [(x,"")] -> x - _ -> 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 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) - - logdata@(Log rec log fulLog) <- getData `onNothing` return (Log False [][]) - - mynode <- runTrans getMyNode -- debug - if not rec -- !> ("wormhole recovery", rec) - then runTrans $ (do - - conn <- mconnect node -- !> (mynode,"connecting node ", node) - setData conn{calling= True} -#ifdef ghcjs_HOST_OS - addPrefix -- for the DOM identifiers -#endif - comp ) - <*** do when (isJust moldconn) . setData $ fromJust moldconn - when (isJust mclosure). setData $ fromJust mclosure - - -- <** is not enough - else do - let conn = fromMaybe (error "wormhole: no connection in remote node") moldconn --- conn <- getData `onNothing` error "wormhole: no connection in remote node" - - setData $ conn{calling= False} - - runTrans $ comp - <*** do --- when (null log) $ setData WasRemote !> "NULLLOG" - when (isJust mclosure) . setData $ fromJust mclosure - - - - -#ifndef ghcjs_HOST_OS -type JSString= String -pack= id - - - -#endif - -newtype Prefix= Prefix JSString deriving(Read,Show) - -addPrefix= Transient $ do - r <- liftIO $ replicateM 5 (randomRIO ('a','z')) - setData $ Prefix $ pack r - return $ Just () - - --- | translates computations back and forth between two nodes --- reusing a connection opened by `wormhole` --- --- each teleport transport to the other node what is new in the log since the --- last teleport --- --- It is used trough other primitives like `runAt` which involves two teleports: --- --- runAt node= wormhole node $ loggedc $ do --- > teleport --- > r <- Cloud $ runCloud proc <** setData WasRemote --- > teleport --- > return r - -teleport :: Cloud () -teleport = do - local $ Transient $ do - - cont <- get - - -- 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 is not recovering in the remote node then it is active - then do - conn@Connection{closures= closures,calling= calling} <- getData - `onNothing` error "teleport: No connection defined: use wormhole" - - --read this Closure - Closure closRemote <- getData `onNothing` return (Closure 0 ) - - --set his own closure in his Node data - - let closLocal = sum $ map (\x-> case x of Wait -> 100000; - Exec -> 1000 - _ -> 1) fulLog --- closLocal <- liftIO $ randomRIO (0,1000000) - - liftIO $ modifyMVar_ closures $ \map -> return $ M.insert closLocal (fulLog,cont) map - - let tosend= reverse $ if closRemote==0 then fulLog else log -- drop offset $ reverse fulLog !> ("fulLog", fulLog) - - runTrans $ msend conn $ SMore (closRemote,closLocal, tosend ) --- !> ("teleport sending", tosend ) --- !> "--------->------>---------->" - - setData $ if (not calling) then WasRemote else WasParallel - - return Nothing - - else do - - delData WasRemote -- !> "deleting wasremote in teleport" - -- it is recovering, therefore it will be the - -- local, not remote - - return (Just ()) -- !> "TELEPORT remote" - - - - --- | 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 - --- | `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 - - ---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.sClose sock - -mclose (Connection node - (Just (Node2Web sconn )) - bufSize events blocked _ _ _)= - WS.sendClose sconn ("closemsg" :: BS.ByteString) - -#else - -mclose (Connection _ (Just (Web2Node sconn)) _ _ blocked _ _ _)= - JavaScript.Web.WebSocket.close Nothing Nothing sconn - -#endif - -liftIOF :: IO b -> TransIO b -liftIOF mx=do - ex <- liftIO $ (mx >>= return . Right) `catch` (\(e :: SomeException) -> return $ Left e) - case ex of - Left e -> finish $ Just e - Right x -> return x - -mconnect :: Node -> TransIO Connection -mconnect node@(Node _ _ _ _ )= do - nodes <- getNodes -- !> ("connecting node", node) - - let fnode = filter (==node) nodes - case fnode of - [] -> addNodes [node] >> mconnect node - [Node host port pool _] -> do - - - plist <- liftIO $ readMVar pool - case plist of - handle:_ -> do - delData $ Closure undefined - return handle -- !> ("REUSED!", node) - - _ -> do --- liftIO $ putStr "*****CONNECTING NODE: " >> print node - my <- getMyNode --- liftIO $ putStr "OPENING CONNECTION WITH :" >> print port - Connection{comEvent= ev} <- getSData <|> error "connect: listen not set for this node" - -#ifndef ghcjs_HOST_OS - - conn <- liftIOF $ do - let size=8192 - sock <- connectTo' size host $ PortNumber $ fromIntegral port - -- !> ("CONNECTING ",port) - - conn <- defConnection >>= \c -> return c{myNode=my,comEvent= ev,connData= Just $ Node2Node u sock (error $ "addr: outgoing connection")} - - SBS.send sock "CLOS a b\n\n" -- !> "sending CLOS" - - - return conn - -#else - conn <- do - - ws <- connectToWS host $ PortNumber $ fromIntegral port - conn <- defConnection >>= \c -> return c{comEvent= ev,connData= Just $ Web2Node ws} - - return conn -- !> ("websocker CONNECION") -#endif - chs <- liftIO $ newIORef M.empty - let conn'= conn{closChildren= chs} - liftIO $ modifyMVar_ pool $ \plist -> return $ conn':plist - - putMailbox "connections" (conn',node) - - delData $ Closure undefined - - - - - return conn - - 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)) - return sock) - -#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 -callTo' :: (Show a, Read a,Typeable a) => Node -> Cloud a -> Cloud a -callTo' node remoteProc= do - mynode <- local 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 - ,socket ::Socket - ,remoteNode :: NS.SockAddr - } - - | Node2Web{webSocket :: WS.Connection} -#else - - Web2Node{webSocket :: WebSocket} -#endif - - - -data Connection= Connection{myNode :: Node - ,connData :: Maybe(ConnectionData) - ,bufferSize :: BuffSize - -- Used by getMailBox, putMailBox - ,comEvent :: IORef (M.Map T.Text (EVar Dynamic)) - -- multiple wormhole/teleport use the same connection concurrently - ,blocked :: Blocked - ,calling :: Bool - -- local closures with his log and his continuation - ,closures :: MVar (M.Map IdClosure ([LogElem], 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 (TVar[EventF]))} - - deriving Typeable - - --- 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. -newMailbox :: T.Text -> TransIO () -newMailbox name= do --- return () -- !> "newMailBox" - Connection{comEvent= mv} <- getData `onNothing` errorMailBox - ev <- newEVar - liftIO $ atomicModifyIORef mv $ \mailboxes -> (M.insert name ev mailboxes,()) - - --- | write to the mailbox -putMailbox :: Typeable a => T.Text -> a -> TransIO () -putMailbox name dat= do -- sendNodeEvent (name, Just dat) - Connection{comEvent= mv} <- getData `onNothing` errorMailBox - mbs <- liftIO $ readIORef mv - let mev = M.lookup name mbs - case mev of - Nothing ->newMailbox name >> putMailbox name dat - Just ev -> writeEVar ev $ toDyn dat - -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 name= do - Connection{comEvent= mv} <- getData `onNothing` errorMailBox - mbs <- liftIO $ readIORef mv - let mev = M.lookup name mbs - case mev of - Nothing ->newMailbox name >> getMailbox name - Just ev ->do - d <- readEVar ev - case fromDynamic d of -- !> "getMailBox" of - Nothing -> empty - Just x -> return x - - --- | delete all subscriptions for that mailbox expecting this kind of data. -cleanMailbox :: Typeable a => T.Text -> a -> TransIO () -cleanMailbox name witness= do - 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,()) - - - - - -defConnection :: MonadIO m => m Connection - --- #ifndef ghcjs_HOST_OS -defConnection = liftIO $ do - x <- newMVar () - y <- newMVar M.empty - z <- return $ error "closchildren newIORef M.empty" - return $ Connection (error "node in default connection") 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 - setData $ conn{bufferSize= size} - return $ Just () - -getBuffSize= - (do getSData >>= return . bufferSize) <|> return 8192 - - - - -listen :: Node -> Cloud () -listen (node@(Node _ port _ _ )) = onAll $ do - addThreads 1 - - setData $ Log False [] [] - - conn' <- getSData <|> defConnection - ev <- liftIO $ newIORef M.empty - chs <- liftIO $ newIORef M.empty - let conn= conn'{myNode=node, comEvent=ev,closChildren=chs} - - setData conn - addNodes [node] - mlog <- listenNew (fromIntegral port) conn <|> listenResponses - - - execLog mlog - - - -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 - - - (sock,addr) <- waitEvents $ NS.accept sock -- !!> "BEFORE ACCEPT" - - 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) - - - initFinish - onFinish $ const $ do --- return() !> "onFinish closures receivedd with LISTEN" - let Connection{closures=closures}= conn -- !> "listenNew closures empty" - liftIO $ modifyMVar_ closures $ const $ return M.empty - - - (method,uri, headers) <- receiveHTTPHead sock - - case method of - "CLOS" -> - do --- return () !> "CLOS detected" - setData $ conn{connData=Just (Node2Node (PortNumber port) sock addr),closChildren=chs} - --- 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 - - setData conn{connData= Just (Node2Web sconn ),closChildren=chs} - - - killOnFinish $ parallel $ do - msg <- WS.receiveData sconn -- WebSockets - return . read $ BC.unpack msg --- !> ("Server WebSocket msg read",msg) !> "<-------<---------<--------------" - - - - - - - ---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= do - - (conn, node) <- getMailbox "connections" - setData conn -#ifndef ghcjs_HOST_OS - case conn of - 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 - liftIO $ putStrLn "removing node: ">> print node - nodes <- getNodes - setNodes $ nodes \\ [node] - - - let Connection{closures=closures}= conn - liftIO $ modifyMVar_ closures $ const $ return M.empty - - - killOnFinish $ mread conn - - -type IdClosure= Int - -data Closure= Closure IdClosure - -execLog mlog = Transient $ do - case mlog of -- !> ("RECEIVED ", mlog ) of - SError e -> do - runTrans $ finish $ Just e - return Nothing - - SDone -> runTrans(finish Nothing) >> return Nothing - SMore r -> process r False - SLast r -> process r True - - where - process (closl,closr,log) deleteClosure= do - conn@Connection {closures=closures,closChildren=mapThreads} <- getData `onNothing` error "Listen: myNode not set" - - if closl== 0 then do - setData $ Log True log $ reverse log - setData $ Closure closr --- --- chs <- liftIO $ atomicModifyIORef mapThreads --- $ \mapth -> --- let mx = M.lookup closr mapth --- in case mx of --- Just tv -> (mapth,tv) !> "FOUND" --- Nothing -> --- let tv = unsafePerformIO $ newTVarIO [] --- in (M.insert closr tv mapth, tv) --- --- modify $ \ s -> s{children= chs} -- to allow his own thread control - - return $ Just () -- !> "executing top level closure" - - else do - mcont <- liftIO $ modifyMVar closures - $ \map -> return (if deleteClosure then - M.delete closl map - else map, M.lookup closl map) - -- !> ("closures=", M.size map) - case mcont of - Nothing -> do - runTrans $ msend conn $ SLast (closr,closl, [] :: [()] ) - -- to delete the remote closure - error ("request received for non existent closure: " ++ show closl) - -- execute the closure - Just (fulLog,cont) -> liftIO $ runStateT (do - let nlog= reverse log ++ fulLog - setData $ Log True log nlog - - setData $ Closure closr - runCont cont) cont -- !> ("executing closure",closr) - - - return Nothing -- !> "FINISH CLOSURE" - -#ifdef ghcjs_HOST_OS -listen node = onAll $ do - addNodes [node] - - events <- liftIO $ newIORef M.empty - - conn <- defConnection >>= \c -> return c{myNode=node,comEvent=events} - setData conn - r <- listenResponses - execLog r -#endif - -type Pool= [Connection] -type Package= String -type Program= String -type Service= (Package, Program) - - --- * Level 2: connections node lists and operations with the node list - - -{-# NOINLINE emptyPool #-} -emptyPool :: MonadIO m => m (MVar Pool) -emptyPool= liftIO $ newMVar [] - - -createNodeServ :: HostName -> Integer -> [Service] -> IO Node -createNodeServ h p svs= do - pool <- emptyPool - return $ Node h ( fromInteger p) pool svs - - - - -createNode :: HostName -> Integer -> IO Node -createNode h p= createNodeServ h p [] - -createWebNode :: IO Node -createWebNode= do - pool <- emptyPool - return $ Node "webnode" ( fromInteger 0) 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 _ s= - let r= readsPrec' 0 s - in case r of - [] -> [] - [((h,p,ss),s')] -> [(Node h p empty - ( ss),s')] - where - 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) - - ---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. initialize it with connect, listen, initNode..." - - -getMyNode :: TransIO Node -- (MonadIO m, MonadState EventF m) => m Node -getMyNode = do - Connection{myNode= node} <- getSData <|> errorMyNode "getMyNode" :: TransIO Connection - return node - - - --- | 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 :: [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 $ my: (( nub $ nodes ++ prevnodes) \\[my]) - --- | set the list of nodes -setNodes nodes= liftIO $ atomically $ writeTVar nodeList $ nodes - - -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= 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 getNodes - let nodes' = filter (not . isWebNode) nodes - foldr op init $ map (\node -> runAt node proc) nodes' - where - isWebNode Node {nodeServices=srvs} - | ("webnode","") `elem` srvs = True - | otherwise = False - --- | 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 reachable share the same list of nodes -connect' remotenode= do - nodes <- local getNodes - local $ liftIO $ putStrLn $ "connecting to: "++ show remotenode - - 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] - - 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 - - local $ do - 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) - if isWebSocketsReq headers - then liftIO $ do - - stream <- makeStream -- !!> "WEBSOCKETS request" - (do - 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) - - 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 - - - - 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 file: index.html
please compile with ghcjs
ghcjs program.hs -o static/out") - - 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 ) - - - empty - - where - uriPath = BC.dropWhile (/= '/') - - - -isWebSocketsReq = not . null - . filter ( (== mk "Sec-WebSocket-Key") . fst) - - - -data ParseContext a = IsString a => ParseContext (IO a) a deriving Typeable - - ---giveData s= do --- r <- readFrom s -- 80000 --- return r -- !> ( "giveData ", r) - - -receiveHTTPHead s = do - 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 -o static/out") input - ::ParseContext BS.ByteString) - (method, uri, vers) <- (,,) <$> getMethod <*> getUri <*> getVers - headers <- many $ (,) <$> (mk <$> getParam) <*> getParamValue -- !> (method, uri, vers) - return (method, toStrict uri, headers) -- !> (method, uri, headers) - - where - - getMethod= getString - getUri= getString - getVers= getString - 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))) - - - - -dropSpaces= parse $ \str ->((),BS.dropWhile isSpace str) - -dropChar= parse $ \r -> ((), BS.tail r) - -endline c= c== '\n' || c =='\r' - ---tGetLine= tTakeWhile . not . endline - -readStream :: (Typeable a, Read a) => BS.ByteString -> [StreamData a] -readStream s= readStream1 $ BS.unpack s - where - readStream1 s= - let [(x,r)] = readsPrec' 0 s - in x : readStream1 r - - - -parallelReadHandler :: Loggable a => TransIO (StreamData a) -parallelReadHandler= do - ParseContext readit str <- getSData <|> error "parallelReadHandler: ParseContext not found" - :: (TransIO (ParseContext BS.ByteString)) - r <- killOnFinish $ choose $ readStream str - - - return r --- !> ("read",r) --- !> "<-------<----------<--------<----------" - - -getString= do - dropSpaces - tTakeWhile (not . isSpace) - -tTakeWhile :: (Char -> Bool) -> TransIO BS.ByteString -tTakeWhile cond= parse (BS.span cond) - - -parse :: Monoid b => (BS.ByteString -> (b, BS.ByteString)) -> TransIO b -parse split= do - ParseContext readit str <- getSData - <|> error "parse: ParseContext not found" - :: TransIO (ParseContext BS.ByteString) - if str == mempty - then do - str3 <- liftIO readit - - setData $ ParseContext readit str3 -- !> str3 - - if str3== mempty then empty else parse split - 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 - setData $ ParseContext readit str3 - - if str3== mempty - then return ret <> (parse split <|> return mempty) - - else return ret - - - - -#endif - - - -#ifdef ghcjs_HOST_OS -isBrowserInstance= True - -#else -isBrowserInstance= False -#endif +{-# 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, + +-- * primitives for communication +wormhole, teleport, copyData, + + + +-- * single node invocation +beamTo, forkTo, callTo, runAt, atRemote, + +-- * invocation of many nodes +clustered, mclustered, callNodes, + +-- * messaging +putMailbox, putMailbox',getMailbox,getMailbox',cleanMailbox,cleanMailbox', + +-- * thread control +single, unique, + +#ifndef ghcjs_HOST_OS +-- * buffering control +setBuffSize, getBuffSize, +#endif + +#ifndef ghcjs_HOST_OS +-- * REST API +api, HTTPMethod(..), PostParams, +#endif +-- * node management +createNode, createWebNode, createNodeServ, getMyNode, getNodes, +addNodes, shuffleNodes, + +-- * low level + + getWebServerNode, Node(..), nodeList, Connection(..), Service(), + isBrowserInstance, + + defConnection, + ConnectionData(..), +#ifndef ghcjs_HOST_OS + ParseContext(..) +#endif + +) where + +import Transient.Move.Internals + + + + + + diff --git a/src/Transient/Move/Internals.hs b/src/Transient/Move/Internals.hs new file mode 100644 --- /dev/null +++ b/src/Transient/Move/Internals.hs @@ -0,0 +1,1669 @@ +----------------------------------------------------------------------------- +-- +-- 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.Logged +import Transient.Indeterminism(choose) +import Transient.Backtrack +import Transient.EVars + + +import Data.Typeable +import Control.Applicative +#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(send,sendMany,sendAll,recv) +import qualified Network.Socket.ByteString.Lazy as SBSL +import Data.CaseInsensitive(mk) +import Data.Char(isSpace) + +#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 :: 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,Applicative,Monoid,Alternative, Monad, Num, MonadState EventF) + + +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 + +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 + , \s 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\nContent-Length: 0\nConnection: close\n\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 the cloud computation. +runCloudIO :: Typeable a => Cloud a -> IO (Maybe a) +runCloudIO (Cloud mx)= keep mx + +-- | run the cloud computation 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 + +lazy :: TransIO a -> Cloud a +lazy mx= onAll $ getCont >>= \st -> Transient $ + return $ unsafePerformIO $ runStateT (runTrans mx) st >>= return .fst + +-- 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 :: 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 + +-- | locally perform IO. `localIO = lliftIO` +localIO :: Loggable a => IO a -> Cloud a +localIO= lliftIO + +--remote :: Loggable a => TransIO a -> Cloud a +--remote x= Cloud $ step' x $ \full x -> Transient $ do +-- let add= Wormhole: full +-- setData $ Log False add add +-- +-- r <- runTrans x +-- +-- let add= WaitRemote: full +-- (setData $ Log False add add) -- !!> "AFTER STEP" +-- return r + +-- | stop the current computation and does not execute any alternative computation +fullStop :: Cloud stop +fullStop= onAll $ setData WasRemote >> stop + + +-- | 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 = 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 getMyNode + beamTo node + r <- remoteProc + beamTo mynode + return r +#endif + +-- | 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" + r <- Cloud $ runCloud proc <** setData WasRemote + teleport -- !> "teleport 2222" + return r + +-- | 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 + con@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 -> 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 + con@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 tv -> empty + Nothing -> do + tv <- get + liftIO $ modifyIORef rmap $ \mapth -> M.insert id tv mapth + + + + +-- | 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) + labelState $ "wormhole" ++ show node + logdata@(Log rec log fulLog) <- getData `onNothing` return (Log False [][]) + + mynode <- runTrans getMyNode -- debug + if not rec -- !> ("wormhole recovery", rec) + then runTrans $ (do + + conn <- mconnect node + + setData conn{calling= True} + + comp ) + <*** do when (isJust moldconn) . setData $ fromJust moldconn + when (isJust mclosure) . setData $ fromJust mclosure + + -- <** is not enough + else do + let conn = fromMaybe (error "wormhole: no connection in remote node") moldconn +-- conn <- getData `onNothing` error "wormhole: no connection in remote node" + + setData $ conn{calling= False} + + runTrans $ comp + <*** do +-- when (null log) $ setData WasRemote !> "NULLLOG" + when (isJust mclosure) . setData $ fromJust mclosure + + + + +#ifndef ghcjs_HOST_OS +type JSString= String +pack= id + + + +#endif +-- +--newtype Prefix= Prefix JSString deriving(Read,Show) +--{-#NOINLINE rprefix #-} +--rprefix= unsafePerformIO $ newIORef 0 +--addPrefix :: (MonadIO m, MonadState EventF m) => m () +--addPrefix= do +-- r <- liftIO $ atomicModifyIORef rprefix (\n -> (n+1,n)) -- liftIO $ replicateM 5 (randomRIO ('a','z')) +-- (setData $ Prefix $ pack $ show r) !> "addPrefix" +-- + + + +-- | translates computations back and forth between two nodes +-- reusing a connection opened by `wormhole` +-- +-- each teleport transport to the other node what is new in the log since the +-- last teleport +-- +-- It is used trough other primitives like `runAt` which involves two teleports: +-- +-- runAt node= wormhole node $ loggedc $ do +-- > teleport +-- > r <- Cloud $ runCloud proc <** setData WasRemote +-- > teleport +-- > return r + +teleport :: Cloud () +teleport = do + local $ Transient $ do + + cont <- get + 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) + -- if is not recovering in the remote node then it is active + then do + conn@Connection{connData=contype,closures= closures,calling= calling} <- getData + `onNothing` error "teleport: No connection defined: use wormhole" +#ifndef ghcjs_HOST_OS + case contype of + Just Self -> do + + setData $ if (not calling) then WasRemote else WasParallel + runTrans $ async $ return () -- !> "SELF" -- call himself + _ -> do +#else + do +#endif + --read this Closure + Closure closRemote <- getData `onNothing` return (Closure 0 ) + + --set his own closure in his Node data + + let closLocal = sum $ map (\x-> case x of Wait -> 100000; + Exec -> 1000 + _ -> 1) fulLog + +-- closLocal <- liftIO $ randomRIO (0,1000000) + node <- runTrans getMyNode + + 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 (closRemote,closLocal,tosend ) +-- !> ("teleport sending", SMore (closRemote,closLocal,tosend)) +-- !> "--------->------>---------->" +-- !> ("fulLog", fulLog) + + setData $ if (not calling) then WasRemote else WasParallel + + return Nothing + + else do + + delData WasRemote -- !> "deleting wasremote in teleport" + -- it is recovering, therefore it will be the + -- local, not remote + + return (Just ()) -- !> "TELEPORT remote" + + + + +-- | 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 a => a -> TransIO () +putMailbox = putMailbox' 0 + +-- | write to a mailbox identified by an Integer besides the type +putMailbox' :: Typeable a => Int -> a -> 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 a => TransIO a +getMailbox = getMailbox' 0 + +-- | read from a mailbox identified by a number besides the type +getMailbox' :: Typeable a => Int -> TransIO a +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 getNodes + let nodes' = filter (not . isWebNode) nodes + callNodes' nodes' op init proc + where + isWebNode Node {nodeServices=srvs} + | ("webnode","") `elem` srvs = True + | otherwise = False + +callNodes' nodes op init proc= 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 + + +msend :: Loggable a => Connection -> StreamData a -> TransIO () + +#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 (TLSNode2Node ctx)) _ _ blocked _ _ _) r= + liftIO $ sendTLSData ctx $ BS.pack (show r) -- !> "TLS SEND" + + +msend (Connection _(Just (Node2Web sconn)) _ _ blocked _ _ _) r=liftIO $ + {-withMVar blocked $ const $ -} WS.sendTextData sconn $ BS.pack (show r) -- !> "websockets send" + + +#else + +msend (Connection _ (Just (Web2Node sconn)) _ _ blocked _ _ _) r= liftIO $ + withMVar blocked $ const $ JavaScript.Web.WebSocket.send (JS.pack $ show r) sconn -- !!> "MSEND SOCKET" + + + +#endif + +msend (Connection _ Nothing _ _ _ _ _ _) _= error "msend out of wormhole context" + +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.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 () + +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 ctx)) _ _ _ _ _ _) = parallelReadHandler +-- parallel $ do +-- s <- recvTLSData ctx +-- return . read' $ BC.unpack s + +mread (Connection node (Just (Node2Web sconn )) _ _ _ _ _ _)= + parallel $ do + s <- WS.receiveData sconn + return . read' $ BS.unpack s +-- !> ("WS MREAD RECEIVED ---->", s) + + + +getWebServerNode :: TransIO Node +getWebServerNode = getMyNode +#endif + +read' s= case readsPrec' 0 s of + [(x,"")] -> x + _ -> error $ "reading " ++ s + +--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 node + (Just (Node2Web sconn )) + bufSize events blocked _ _ _)= + 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@(Node _ _ _ _ )= do + nodes <- getNodes + + let fnode = filter (==node) nodes + case fnode of + [] -> addNodes [node] >> mconnect node + [Node host port pool _] -> do + + + plist <- liftIO $ readMVar pool + case plist of -- !> ("length",length plist) of + handle:_ -> do + delData $ Closure undefined + return handle +-- !> ("REUSED!", node) + + _ -> mconnect1 host port pool -- !> ("MCONNECT1",host,port) + where + + + +#ifndef ghcjs_HOST_OS + mconnect1 host port pool= do + + connectNode2Node host port <|> connectWebSockets host port + + watchConnection + + + where + connectSockTLS host port= do + return () -- !> "connectSockTLS" + + my <- getMyNode + + let size=8192 + Connection{comEvent= ev} <- getSData <|> error "connect: listen not set for this node" + + + sock <- liftIO $connectTo' size host $ PortNumber $ fromIntegral port + + + conn' <- liftIO $ 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 "listenResponses: Parse error") input + + maybeClientTLSHandshake host sock input + + + + connectNode2Node host port= do + connectSockTLS host port +-- return () !> "CONNECT NODE2NODE" + conn <- getSData <|> error "mconnect: no connection data" + sendRaw conn "CLOS a b\r\n\r\n" + r <- -- async (threadDelay 2000000 >> return Nothing) <|> + liftIO $ readFrom conn +-- (liftIO $ NS.recv sock 1000 >>= return . Just ) + +-- return () !> ("READ",r) + + case r of + "OK" -> return() + _ -> do + let Connection{connData=cdata}= conn + case cdata of + Just(Node2Node _ s _) -> liftIO $ NS.close s -- since the HTTP firewall closes the connection +-- TLSNode2Node c -> contextClose c -- TODO + empty --- close socket,tls + + + connectWebSockets host port = do + liftIO $ print "Trying WebSockets" + connectSockTLS host port -- a new connection +-- return () !> "connected" + never<- liftIO $ newEmptyMVar :: TransIO (MVar ()) + conn <- getSData <|> error "connectWebSockets: no connection" + stream <- liftIO $ makeWSStreamFromConn conn + wscon <- react (NWS.runClientWithStream stream host "/" + WS.defaultConnectionOptions []) (takeMVar never) + + liftIO $ print "WebSockets connection" + modifyState $ \(Just c) -> Just c{connData= Just $ Node2Web wscon} + +-- noConnection= error $ show node ++ ": no connection" + + + watchConnection= do + conn <- getSData + parseContext <- getSData <|> error "NO PASE CONTEXT" + :: TransIO (ParseContext BS.ByteString) + chs <- liftIO $ newIORef M.empty + let conn'= conn{closChildren= chs} + (liftIO $ modifyMVar_ pool $ \plist -> return $ conn':plist) -- !> (node,"ADDED TO POOL") + + -- tell listenResponses to watch incoming responses + putMailbox ((conn',parseContext,node) -- !> "PUTMAILBOX" + :: (Connection,ParseContext BS.ByteString,Node)) + delData $ Closure undefined + return conn + +#else + mconnect1 host port pool= do + my <- getMyNode + + Connection{comEvent= ev} <- getSData <|> error "connect: listen not set for this node" + + do + ws <- connectToWS host $ PortNumber $ fromIntegral port +-- !> "CONNECTWS" + conn <- defConnection >>= \c -> return c{comEvent= ev,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 + + + +-- 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 + let ps = case (protocol :: JSString)of "http:" -> "ws://"; "https:" -> "wss://" + wsOpen $ JS.pack $ ps++ h++ ":"++ show p +#endif + + + +type Blocked= MVar () +type BuffSize = Int +data ConnectionData= +#ifndef ghcjs_HOST_OS + Node2Node{port :: PortID + ,socket ::Socket + ,remoteNode :: NS.SockAddr + } + | TLSNode2Node{tlscontext :: SData} + | Node2Web{webSocket :: WS.Connection} +-- | WS2Node{webSocketNode :: WS.Connection} + | Self +#else + Self + | Web2Node{webSocket :: WebSocket} +#endif + +data MailboxId= MailboxId Int TypeRep deriving (Eq,Ord) + +data Connection= Connection{myNode :: 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 closures with his log and his continuation + ,closures :: MVar (M.Map IdClosure ([LogElem], 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 => m Connection + +-- #ifndef ghcjs_HOST_OS +defConnection = liftIO $ do + x <- newMVar () + y <- newMVar M.empty + z <- return $ error "closchildren newIORef M.empty" + return $ Connection (error "node in default connection") 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 + setData $ conn{bufferSize= size} + return $ Just () + +getBuffSize= + (do getSData >>= return . bufferSize) <|> return 8192 + + + + +listen :: Node -> Cloud () +listen (node@(Node _ port _ _ )) = onAll $ do + addThreads 1 + + setData $ Log False [] [] + + conn' <- getSData <|> defConnection + ev <- liftIO $ newIORef M.empty + chs <- liftIO $ newIORef M.empty + let conn= conn'{connData=Just Self,myNode=node, comEvent=ev,closChildren=chs} + + setData conn + liftIO $ modifyMVar_ (connection node) $ const $ return [conn] + addNodes [node] + mlog <- listenNew (fromIntegral port) conn <|> listenResponses + + execLog mlog + +-- 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 + + (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) + + let conn= conn'{closChildren=chs} + + input <- liftIO $ SBSL.getContents sock + + setData $ (ParseContext (error "parsing request") input + ::ParseContext BS.ByteString) + + cutExceptions + + onException $ \(e :: SomeException) -> do + liftIO $ print e + let Connection{closures=closures,closChildren= rmap}= conn + liftIO $ do + modifyMVar_ closures $ const $ return M.empty + writeIORef rmap M.empty + topState >>= showThreads + killBranch + + setState conn{connData=Just (Node2Node (PortNumber port) sock addr)} + maybeTLSServerHandshake sock input + + + + (method,uri, headers) <- receiveHTTPHead + + case method of + + "CLOS" -> + do + conn <- getSData + sendRaw conn "OK" -- !> "CLOS detected" + + + parallelReadHandler + + _ -> do + let uri'= BC.tail $ 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" + + 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 (0,0, log' ) + + else do + -- stay serving pages until a websocket request is received + servePages (method, uri', headers) + conn <- getSData + sconn <- makeWebsocketConnection conn uri headers + -- websockets mode + setData conn{connData= Just (Node2Web sconn) ,closChildren=chs} +-- 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 (0,0,log) + ((x ,_):_) -> return (x :: StreamData (Int,Int,[LogElem])) + + case r of + SError e -> do +-- liftIO $ WS.sendClose sconn ("error" :: BS.ByteString) + finish (Just e) +-- !> "FINISH1" + _ -> return r + + where + uriPath = BC.dropWhile (/= '/') + split []= [] + split ('/':r)= split r + split s= + let (h,t) = span (/= '/') s + in h: split t + + + +--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 + + labelState $ "listen from: "++ show node + +-- return () !> ("LISTEN FROM",node) + + setData conn + +#ifndef ghcjs_HOST_OS + setData (parsecontext :: ParseContext BS.ByteString) +#else + setData (parsecontext :: ParseContext JSString) +#endif + + + + cutExceptions + onException (\(e:: SomeException) -> do + liftIO $ print e + liftIO $ putStr "removing2 node: " >> print node + nodes <- getNodes + setNodes $ nodes \\ [node] + killChilds + let Connection{closures=closures}= conn + liftIO $ modifyMVar_ closures $ const $ return M.empty) + + + mread conn + + +type IdClosure= Int + +newtype Closure= Closure IdClosure deriving Show + +execLog mlog = Transient $ + case mlog of + SError e -> do + runTrans $ back e + return Nothing + + SDone -> runTrans(back $ ErrorCall "SDone") >> return Nothing -- TODO remove closure? + SMore r -> process r False + SLast r -> process r True + + where + process (closl,closr,log) deleteClosure= do + conn@Connection {closures=closures} <- getData `onNothing` error "Listen: myNode not set" + + if closl== 0 then do + + setData $ Log True log $ reverse log + setData $ Closure closr + + return $ Just () -- !> "executing top level closure" + + else do +-- return () !> "execlog" + mcont <- liftIO $ modifyMVar closures + $ \map -> return (if deleteClosure then + M.delete closl map + else map, M.lookup closl map) + -- !> ("closures=", M.size map) + case mcont of + Nothing -> do + + runTrans $ msend conn $ SLast (closr,closl, [] :: [()] ) + -- to delete the remote closure + error ("request received for non existent closure: " + ++ show closl) + -- execute the closure + Just (fulLog,cont) -> liftIO $ runStateT (do + + let nlog= reverse log ++ fulLog + setData $ Log True log nlog + + setData $ Closure closr +-- !> ("SETCLOSURE",closr) + runCont cont) cont +-- !> ("executing closure",closl) + + + return Nothing +-- !> "FINISH CLOSURE" + +#ifdef ghcjs_HOST_OS +listen node = onAll $ do + addNodes [node] + + events <- liftIO $ newIORef M.empty + + conn <- defConnection >>= \c -> return c{myNode=node,comEvent=events} + setData conn + r <- listenResponses + execLog r +#endif + +type Pool= [Connection] +type Package= String +type Program= String +type Service= (Package, Program) + + + + + +-------------------------------------------- +data ParseContext a = IsString a => ParseContext (IO a) a deriving Typeable + + +#ifndef ghcjs_HOST_OS + +parallelReadHandler :: Loggable a => TransIO (StreamData a) +parallelReadHandler= do + str <- giveData :: TransIO BS.ByteString + + r <- choose $ readStream str +-- rest <- liftIO $ newIORef $ BS.unpack str +-- +-- r <- parallel $ readStream' rest + return r +-- !> ("read",r) +-- !> "<-------<----------<--------<----------" + where +-- readStream' :: (Loggable a) => IORef String -> IO(StreamData a) +-- readStream' rest = do +-- return () !> "reAD StrEAM" +-- s <- readIORef rest +-- liftIO $ print $ takeWhile (/= ')') s +-- [(x,r)] <- maybeRead s +-- writeIORef rest r +-- return x + + 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 + + +-- 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 renderin 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
please compile with ghcjs
ghcjs program.hs -o static/out") + + case mcontent of + Just content -> liftIO $ sendRaw conn $ + "HTTP/1.0 200 OK\nContent-Type: text/html\nConnection: close\nContent-Length: " + <> BS.pack (show $ BL.length content) <>"\n\n" <> content + -- (BL.toChunks content ) + Nothing ->liftIO $ sendRaw conn $ BS.pack $ "HTTP/1.0 404 Not Found\nContent-Length: 0\nConnection: close\n\n" + 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 + liftIO $ myThreadId >>= print + send r -- !> r + + + + + + + + + +isWebSocketsReq = not . null + . filter ( (== mk "Sec-WebSocket-Key") . fst) + + +data HTTPMethod= GET | POST deriving (Read,Show,Typeable) + +receiveHTTPHead = do + + (method, uri, vers) <- (,,) <$> getMethod <*> getUri <*> getVers + headers <- manyTill paramPair (string "\r\n\r\n") -- !> (method, uri, vers) + return (method, toStrict uri, headers) -- !> (method, uri, headers) + + where + string :: BS.ByteString -> TransIO BS.ByteString + string s=withData $ \str -> do + let len= BS.length s + ret@(s',str') = BS.splitAt len str + if s == s' + then return ret + else empty + + paramPair= (,) <$> (mk <$> getParam) <*> getParamValue + manyTill p end = scan + where + scan = do{ end; return [] } + <|> + do{ x <- p; xs <- scan; return (x:xs) } + + getMethod= getString + getUri= getString + getVers= getString + 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))) + +dropSpaces= parse $ \str ->((),BS.dropWhile isSpace str) + +dropChar= parse $ \r -> ((), BS.tail r) + +endline c= c== '\n' || c =='\r' + +--tGetLine= 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' ( /= '&') + + +getString= do + dropSpaces + tTakeWhile (not . isSpace) + +tTakeWhile :: (Char -> Bool) -> TransIO BS.ByteString +tTakeWhile cond= parse (BS.span cond) + +tTakeWhile' :: (Char -> Bool) -> TransIO BS.ByteString +tTakeWhile' cond= parse ((\(h,t) -> (h, if BS.null t then t else BS.tail t)) . BS.span cond) + +parse :: (BS.ByteString -> (b, BS.ByteString)) -> TransIO b +parse split= withData $ \str -> + + if str== mempty then empty + + else return $ split str + + +-- | bring the data of a parse context as a lazy byteString to a parser +-- and actualize the parse context with the result +withData :: (BS.ByteString -> TransIO (a,BS.ByteString)) -> TransIO a +withData parser= Transient $ do + ParseContext readMore s <- getData `onNothing` error "parser: no context" + let loop = unsafeInterleaveIO $ do + r <- readMore + (r <>) `liftM` loop + str <- liftIO $ (s <> ) `liftM` loop + mr <- runTrans $ parser str + case mr of + Nothing -> return Nothing + Just (v,str') -> do + setData $ ParseContext readMore str' + return $ Just v + +-- | bring the data of the parse context as a lazy byteString +giveData =noTrans $ do + ParseContext readMore s <- getData `onNothing` error "parser: no context" + :: StateIO (ParseContext BS.ByteString) -- change to strict BS + + let loop = unsafeInterleaveIO $ do + r <- readMore + (r <>) `liftM` loop + liftIO $ (s <> ) `liftM` loop + + +#endif + + + +#ifdef ghcjs_HOST_OS +isBrowserInstance= True +api _= empty +#else +-- | True if it is running in the browser +isBrowserInstance= False + +#endif + + + + + +{-# NOINLINE emptyPool #-} +emptyPool :: MonadIO m => m (MVar Pool) +emptyPool= liftIO $ newMVar [] + + +createNodeServ :: HostName -> Integer -> [Service] -> IO Node +createNodeServ h p svs= do + pool <- emptyPool + return $ Node h ( fromInteger p) pool svs + + + + +createNode :: HostName -> Integer -> IO Node +createNode h p= createNodeServ h p [] + +createWebNode :: IO Node +createWebNode= do + pool <- emptyPool + return $ Node "webnode" ( fromInteger 0) 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 _ s= + let r= readsPrec' 0 s + in case r of + [] -> [] + [((h,p,ss),s')] -> [(Node h p empty + ( ss),s')] + where + empty= unsafePerformIO emptyPool + +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..." + + +getMyNode :: TransIO Node -- (MonadIO m, MonadState EventF m) => m Node +getMyNode = do + Connection{myNode= node} <- getSData <|> errorMyNode "getMyNode" :: TransIO Connection + return node + + + +-- | 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 :: [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 $ my: (( nub $ nodes ++ prevnodes) \\[my]) + +-- | set the list of nodes +setNodes nodes= liftIO $ atomically $ writeTVar nodeList $ nodes + + +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 + + + +-- | 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 +connect :: Node -> Node -> Cloud () +#ifndef ghcjs_HOST_OS +connect node remotenode = do + listen node <|> return () + 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 reachable share the same list of nodes +connect' remotenode= do + nodes <- local getNodes + localIO $ putStrLn $ "connecting to: "++ show remotenode + + newNodes <- runAt remotenode $ do + local $ do + conn@Connection{} <- 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] + + onException $ \(e :: SomeException) -> do + liftIO $ putStrLn "removing node: ">> print nodeConnecting + nodes <- getNodes + setNodes $ nodes \\ [nodeConnecting] + +-- return nodes -- delete + + mclustered . local $ addNodes nodes + + + local $ do + allNodes <- getNodes + liftIO $ putStrLn "Known nodes: " >> print allNodes + return allNodes + + local $ addNodes[remotenode] + + + callNodes' nodes (<>) mempty $ local $ addNodes newNodes -- add the new discovered nodes + + local $ do + nodes <- getNodes + liftIO $ putStrLn "Known nodes: " >> print nodes + + +#else +connect _ _= empty +connect' _ = empty +#endif + + + + + diff --git a/src/Transient/Move/Utils.hs b/src/Transient/Move/Utils.hs --- a/src/Transient/Move/Utils.hs +++ b/src/Transient/Move/Utils.hs @@ -1,144 +1,143 @@ ------------------------------------------------------------------------------ --- --- Module : Transient.Move.Utils --- Copyright : --- License : GPL-3 --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | --- ------------------------------------------------------------------------------ - -module Transient.Move.Utils (initNode,inputNodes, simpleWebApp, initWebApp -, 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. --- the port can be entered in the command line with " -p start/" --- --- 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/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 :: Cloud () -> TransIO () -initNode app= do - node <- getNodeParams - initWebApp node app - - - where - 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 - --- | 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 -inputNodes= do - onServer $ do - local $ option "add" "add a new node at any moment" - - host <- local $ do - r <- input (const True) "Host to connect to: (none): " - if r == "" then stop else return r - - port <- local $ input (const True) "port? " - - connectit <- local $ input (\x -> x=="y" || x== "n") "connect to the node to interchange nodes lists?" - nnode <- localIO $ createNode host port - if connectit== "y" then connect' nnode - else local $ do - liftIO $ putStr "Added node: ">> print nnode - addNodes [nnode] - 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 :: Integer -> Cloud () -> IO () -simpleWebApp port app = do - node <- createNode "localhost" port - keep $ initWebApp node app - --- | use this instead of smpleWebApp when you have to do some initializations in the server prior to the --- initialization of the web server -initWebApp :: Node -> Cloud () -> TransIO () -initWebApp node app= do - conn <- defConnection - setData conn{myNode = node} - serverNode <- getWebServerNode :: TransIO Node - - mynode <- if isBrowserInstance - then liftIO $ createWebNode - else return serverNode - - runCloud $ do - listen mynode <|> return() - 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. --- Otherwise return empty (so it stop the computation). -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 - --- | 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 : GPL-3 +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +-- +-- | +-- +----------------------------------------------------------------------------- + +module Transient.Move.Utils (initNode,inputNodes, simpleWebApp, initWebApp +, 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. +-- the port can be entered in the command line with " -p start/" +-- +-- 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/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 :: Cloud () -> TransIO () +initNode app= do + node <- getNodeParams + initWebApp node app + + + where + 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 + +-- | 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 $ option "add" "add a new node at any moment" + + host <- local $ do + r <- input (const True) "Host to connect to: (none): " + if r == "" then stop else return r + + port <- local $ input (const True) "port? " + + connectit <- local $ input (\x -> x=="y" || x== "n") "connect to the node to interchange node lists? " + nnode <- localIO $ createNode host port + if connectit== "y" then connect' nnode + else local $ do + liftIO $ putStr "Added node: ">> print nnode + addNodes [nnode] + 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 :: Integer -> Cloud () -> IO () +simpleWebApp port app = do + node <- createNode "localhost" 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 :: Node -> Cloud () -> TransIO () +initWebApp node app= do + conn <- defConnection + setData conn{myNode = node} + serverNode <- getWebServerNode :: TransIO Node + + mynode <- if isBrowserInstance + then liftIO $ createWebNode + else return serverNode + runCloud $ do + listen mynode <|> return() + wormhole serverNode app + return () + +-- 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). +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 + +-- | 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() + diff --git a/tests/TestSuite.hs b/tests/TestSuite.hs --- a/tests/TestSuite.hs +++ b/tests/TestSuite.hs @@ -1,3 +1,11 @@ +-- #!/usr/bin/env ./execthirdline.sh +-- development: +-- set -e && docker run -it -v /c/Users/magocoal/OneDrive/Haskell/devel:/devel agocorona/transient:05-02-2017 bash -c "runghc -j2 -isrc -i/devel/transient/src -i/devel/transient-universe/src /devel/transient-universe/tests/$1 $2 $3 $4" + +-- compile and run within a docker image +-- set -e && executable=`basename -s .hs ${1}` && docker run -it -v $(pwd):/work agocorona/transient:05-02-2017 bash -c "ghc /work/${1} && /work/${executable} ${2} ${3}" + + {-# LANGUAGE CPP #-} module Main where @@ -14,15 +22,14 @@ import Transient.Indeterminism import Transient.Logged import Transient.Move -import Transient.Stream.Resource import Transient.MapReduce import Transient.EVars -import Control.Concurrent -import System.IO.Unsafe -import Data.List -import Control.Exception.Base +import Control.Concurrent +import System.IO.Unsafe +import Data.List +import Control.Exception.Base import qualified Data.Map as M -import System.Exit +import System.Exit import Control.Monad.State @@ -32,28 +39,30 @@ #define shouldRun(x) (local $ getMyNode >>= \(Node _ p _ _) -> assert ( p == (x)) (return ())) - main= do - - let numNodes = 4 + let numNodes = 3 ports = [2000 .. 2000 + numNodes - 1] createLocalNode = createNode "localhost" nodes <- mapM createLocalNode ports let n2000= head nodes n2001= nodes !! 1 n2002= nodes !! 2 - n2003= nodes !! 3 - r <-runCloudIO $ do + + + r <- keep' $ freeThreads $ runCloud $ do + runNodes nodes localIO $ putStrLn "------checking Alternative distributed--------" - r <- local $ collect 3 $ - runCloud $ (runAt n2000 (shouldRun(2000) >> return "hello")) + r <- local $ collect 3 $ + runCloud $ (runAt n2000 (shouldRun(2000) >> return "hello" )) <|> (runAt n2001 (shouldRun(2001) >> return "world" )) <|> (runAt n2002 (shouldRun(2002) >> return "world2" )) - loggedc $ assert(sort r== ["hello", "world","world2"]) $ lliftIO $ print r + + loggedc $ assert(sort r== ["hello", "world","world2"]) $ lliftIO $ print r + lliftIO $ putStrLn "--------------checking Applicative distributed--------" r <- loggedc $(runAt n2000 (shouldRun(2000) >> return "hello ")) <> (runAt n2001 (shouldRun(2001) >> return "world " )) @@ -61,8 +70,6 @@ assert(r== "hello world world2") $ lliftIO $ print r - - lliftIO $ putStrLn "----------------checking monadic, distributed-------------" r <- runAt n2000 (shouldRun(2000) >> runAt n2001 (shouldRun(2001) @@ -72,39 +79,20 @@ lliftIO $ putStrLn "----------------checking map-reduce -------------" - r <- reduce (+) . mapKeyB (\w -> (w, 1 :: Int)) $ getText words "hello world hello hi" - lliftIO $ putStr "SOLUTION: " >> print r - assert (sort (M.toList r) == sort [("hello",2::Int),("hi",1),("world",1)]) $ return r - - - - - - + r <- reduce (+) . mapKeyB (\w -> (w, 1 :: Int)) $ getText words "hello world hello" + lliftIO $ putStr "SOLUTION: " >> print r + assert (sort (M.toList r) == sort [("hello",2::Int),("world",1)]) $ return r - lliftIO $ print "SUCCES" local $ exit () - + print "SUCCESS" exitSuccess ---getEffects :: Loggable a => Cloud [(Node, a)] ---getEffects=lliftIO $ readIORef effects --- -runNodes nodes= foldl (<|>) empty (map listen nodes) <|> return () -- (onAll $ async $ return()) --- --- ---delEffects= lliftIO $ writeIORef effects [] ---effects= unsafePerformIO $ newIORef [] --- ---EFFECT x= do --- node <- onAll getMyNode --- lliftIO $ atomicModifyIORef effects $ \ xs -> ((node,x): xs,()) --- return() +runNodes nodes= foldr (<|>) empty (map listen nodes) <|> return () + + #else main= return () #endif - - diff --git a/transient-universe.cabal b/transient-universe.cabal --- a/transient-universe.cabal +++ b/transient-universe.cabal @@ -1,110 +1,115 @@ -name: transient-universe -version: 0.3.5.1 -cabal-version: >=1.10 -build-type: Simple -license: MIT -license-file: LICENSE - -author: Alberto G. Corona -maintainer: agocorona@gmail.com -homepage: http://www.fpcomplete.com/user/agocorona -bug-reports: https://github.com/transient-haskell/transient-universe/issues - -synopsis: Remote execution and map-reduce: distributed computing for Transient -description: - See . -category: Control - -extra-source-files: app/client/Transient/Move/Services/MonitorService.hs - app/server/Transient/Move/Services/MonitorService.hs - -source-repository head - type: git - location: https://github.com/agocorona/transient-universe - -library - exposed-modules: - Transient.Move - Transient.MapReduce - Transient.Move.Utils - if !impl(ghcjs >=0.1) - exposed-modules: - Transient.Move.Services - - build-depends: base >4 && <5, - bytestring -any, - containers -any, - mtl -any, - process -any, - random -any, - stm -any, - text -any, - time -any, - transformers -any, - transient >=0.4.4.1 - if impl(ghcjs >=0.1) - build-depends: ghcjs-base -any, - ghcjs-prim -any - else - build-depends: HTTP -any, - TCache -any, - case-insensitive -any, - directory -any, - filepath -any, - hashable -any, - iproute -any, - network -any, - network-info -any, - network-uri -any, - vector -any, - websockets -any - default-language: Haskell2010 - hs-source-dirs: src . - - -executable monitorService - build-depends: base >4 && <5 - if !impl(ghcjs >=0.1) - hs-source-dirs: app/server/Transient/Move/Services - build-depends: transformers - , transient >=0.4.4.1 - , transient-universe - else - hs-source-dirs: app/client/Transient/Move/Services - main-is: MonitorService.hs - default-language: Haskell2010 - ghc-options: -threaded -rtsopts - - -test-suite test-transient - build-depends: base > 4 - if !impl(ghcjs >=0.1) - build-depends: HTTP -any, - TCache -any, - bytestring -any, - case-insensitive -any, - containers -any, - directory -any, - filepath -any, - hashable -any, - mtl -any, - network -any, - network -any, - network-info -any, - network-uri -any, - process -any, - random -any, - stm -any, - text -any, - time -any, - transformers -any, - transient >=0.4.4.1, - vector -any, - websockets -any - - - type: exitcode-stdio-1.0 - main-is: TestSuite.hs - buildable: True - default-language: Haskell2010 - hs-source-dirs: tests src . +name: transient-universe +version: 0.4.0 +cabal-version: >=1.10 +build-type: Simple +license: MIT +license-file: LICENSE +maintainer: agocorona@gmail.com +homepage: http://www.fpcomplete.com/user/agocorona +bug-reports: https://github.com/agocorona/transient-universe/issues +synopsis: Remote execution and map-reduce: distributed computing for Transient +description: + See . +category: Control +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/agocorona/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 -any, + case-insensitive -any, + directory -any, + filepath -any, + hashable -any, + iproute -any, + network -any, + network-info -any, + network-uri -any, + vector -any, + websockets -any + exposed-modules: + Transient.Move + Transient.MapReduce + Transient.Move.Internals + Transient.Move.Utils + build-depends: + base >4 && <5, + bytestring -any, + containers -any, + mtl -any, + process -any, + random -any, + stm -any, + text -any, + time -any, + transformers -any, + transient >=0.5.1 + default-language: Haskell2010 + hs-source-dirs: src . + +executable monitorService + + if !impl(ghcjs >=0.1) + build-depends: + transformers -any, + transient >=0.4.4, + transient-universe -any + 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 + +test-suite test-transient + + if !impl(ghcjs >=0.1) + build-depends: + mtl -any, + transient -any, + 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 .