transient-universe 0.3.5.1 → 0.4.0
raw patch · 10 files changed
+2695/−2282 lines, 10 filesdep ~TCachedep ~transient
Dependency ranges changed: TCache, transient
Files
- ChangeLog.md +0/−0
- README.md +105/−0
- app/client/Transient/Move/Services/MonitorService.hs +1/−1
- app/server/Transient/Move/Services/MonitorService.hs +64/−65
- src/Transient/MapReduce.hs +510/−502
- src/Transient/Move.hs +58/−1418
- src/Transient/Move/Internals.hs +1669/−0
- src/Transient/Move/Utils.hs +143/−144
- tests/TestSuite.hs +30/−42
- transient-universe.cabal +115/−110
+ ChangeLog.md view
+ README.md view
@@ -0,0 +1,105 @@+ +========= + +[](http://hackage.haskell.org/package/transient-universe) +[](http://stackage.org/lts/package/transient-universe) +[](http://stackage.org/nightly/package/transient-universe) +[](https://travis-ci.org/agocorona/transient-universe) +[](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 + + + + +
app/client/Transient/Move/Services/MonitorService.hs view
@@ -1,1 +1,1 @@-main= return()+main= return ()
app/server/Transient/Move/Services/MonitorService.hs view
@@ -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
src/Transient/MapReduce.hs view
@@ -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
src/Transient/Move.hs view
@@ -10,1421 +10,61 @@ -- -- | see <https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv> ----------------------------------------------------------------------------- -{-# LANGUAGE DeriveDataTypeable , ExistentialQuantification, OverloadedStrings - ,ScopedTypeVariables, StandaloneDeriving, RecordWildCards, FlexibleContexts, CPP - ,GeneralizedNewtypeDeriving #-} -module Transient.Move( - -Cloud(..),runCloudIO, runCloudIO',local,onAll,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<br/> please compile with ghcjs<br/> 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 <prog> -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 + + + + + +
+ src/Transient/Move/Internals.hs view
@@ -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<br/> please compile with ghcjs<br/> ghcjs program.hs -o static/out") + + case mcontent of + Just content -> liftIO $ sendRaw conn $ + "HTTP/1.0 200 OK\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 + + + + +
src/Transient/Move/Utils.hs view
@@ -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 "<program> -p start/<PORT>"------ A node is also a web server that send to the browser the program if it has been--- compiled to JavaScript with ghcjs. `initNode` also initializes the web nodes.------ This sequence compiles to JScript and executes the program with a node in the port 8080------ > ghc program.hs--- > ghcjs program.hs -o static/out--- > ./program -p start/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 "<program> -p start/<PORT>" +-- +-- A node is also a web server that send to the browser the program if it has been +-- compiled to JavaScript with ghcjs. `initNode` also initializes the web nodes. +-- +-- This sequence compiles to JScript and executes the program with a node in the port 8080 +-- +-- > ghc program.hs +-- > ghcjs program.hs -o static/out +-- > ./program -p start/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() +
tests/TestSuite.hs view
@@ -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 - -
transient-universe.cabal view
@@ -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 <http://github.com/transient-haskell/transient>.-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 <http://github.com/agocorona/transient>. +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 .