distributed-process-zookeeper (empty) → 0.2.0.0
raw patch · 6 files changed
+1195/−0 lines, 6 filesdep +basedep +binarydep +bytestringsetup-changed
Dependencies added: base, binary, bytestring, containers, deepseq, distributed-process, distributed-process-monad-control, distributed-process-zookeeper, enclosed-exceptions, hspec, hzk, lifted-base, monad-control, mtl, network, network-transport, network-transport-tcp, transformers
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- distributed-process-zookeeper.cabal +101/−0
- examples/Boss.hs +69/−0
- src/Control/Distributed/Process/Zookeeper.hs +714/−0
- test/Spec.hs +279/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015, Jeremy Huffman++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Jeremy Huffman nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ distributed-process-zookeeper.cabal view
@@ -0,0 +1,101 @@+name: distributed-process-zookeeper+version: 0.2.0.0+cabal-version: >=1.10+build-type: Simple+license: BSD3+license-file: LICENSE+copyright: Jeremy Huffman+maintainer: jeremy@jeremyhuffman.com+homepage: https://github.com/jeremyjh/distributed-process-zookeeper+synopsis: A Zookeeper back-end for Cloud Haskell.+description:+ Provides service and node discovery for Cloud Haskell applications using+ a Zookeeper cluster for name registration, lookups and leader election.+ Uses the hzk bindings to the Zookeeper multi-threaded C library.+ .+ Objectives and features:+ .+ * Compatible with <http://hackage.haskell.org/package/distributed-process-p2p distributed-process-p2p> API - can work as a drop-in+ replacement.+ .+ * No dependencies beyond those already included by distributed-process, hzk and network-transport-tcp.+ .+ * Automatic registration of local names to Zookeeper.+ .+ * Global singleton processes with leader election and re-elections on leader exit.+ .+ .+ Check out the <https://github.com/jeremyjh/distributed-process-zookeeper/blob/master/examples/Boss.hs example>.++category: Control, Cloud Haskell+author: Jeremy Huffman+ +source-repository head+ type: git+ location: git://github.com/jeremyjh/distributed-process-zookeeper.git+ +flag zkexamples+ Description: Build example programs.+ Default: False+ +library+ build-depends:+ base >=4.6 && < 4.8,+ hzk >=2.1.0 && <2.2,+ distributed-process >=0.5.3 && <0.6,+ network-transport >=0.4.1.0 && <0.5,+ network-transport-tcp >=0.4.1 && <0.5,+ binary >=0.7.1.0 && <0.8,+ bytestring >=0.10.0.0 && <0.11,+ containers >=0.5.0.0 && <0.6,+ transformers >=0.4.2.0 && <0.5,+ mtl >=2.2.1 && <2.3,+ deepseq >=1.3.0.0 && <1.4,+ network >=2.6.0.2 && <2.7+ exposed-modules:+ Control.Distributed.Process.Zookeeper+ exposed: True+ buildable: True+ default-language: Haskell2010+ hs-source-dirs: src+ ghc-options: -Wall+ +executable boss+ build-depends:+ base -any,+ distributed-process-zookeeper -any,+ distributed-process -any+ + if flag(zkexamples)+ buildable: True+ else+ buildable: False+ main-is: Boss.hs+ buildable: True+ default-language: Haskell2010+ hs-source-dirs: examples+ ghc-options: -threaded -with-rtsopts=-N+ +test-suite spec+ build-depends:+ base -any,+ distributed-process-zookeeper -any,+ hspec >= 2.0.2,+ distributed-process-monad-control -any,+ lifted-base -any,+ enclosed-exceptions -any,+ network -any,+ distributed-process -any,+ deepseq -any,+ monad-control -any,+ network-transport -any,+ network-transport-tcp -any,+ transformers -any,+ hzk -any,+ bytestring -any+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ buildable: True+ default-language: Haskell2010+ hs-source-dirs: test+ ghc-options: -threaded -with-rtsopts=-N
+ examples/Boss.hs view
@@ -0,0 +1,69 @@+import Control.Distributed.Process+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Zookeeper+import Control.Monad (void)+import System.Environment (getArgs)++-- Run the program with "args" to request a job named "args" - run with no+-- args to startup a boss and each worker. Entering a new-line will+-- terminate Boss or Workers.+main :: IO ()+main =+ bootstrapWith defaultConfig --{logTrace = sayTrace}+ "localhost"+ "0"+ "localhost:2181"+ initRemoteTable $+ do args <- liftIO getArgs+ case args of+ job : more -> -- transient requestor - sends job and exits+ do Just boss <- whereisGlobal "boss"+ send boss (job ++ " " ++ unwords more)+ _ -> -- will be a boss or boss candidate + worker+ do say "Starting persistent process - press <Enter> to exit."+ Right boss <- registerCandidate "boss" $+ do say "I'm the boss!"+ workers <- getCapable "worker"+ case length workers of+ 0 -> say $ "I don't do any work! Start "+ ++ " a worker."+ n -> say $ "I've got " ++ show n+ ++ " workers right now."+ bossLoop 0+ self <- getSelfNode+ if self == processNodeId boss -- like a boss+ then void $ liftIO getLine+ else do worker <- getSelfPid+ register "worker" worker+ say "Worker waiting for task..."+ void . spawnLocal $ --wait for exit+ do void $ liftIO getLine+ send worker "stop"+ workLoop++-- Calling getCapable for every loop is no problem because the results+-- of this are cached until the worker node is updated in Zookeeper.+bossLoop :: Int -> Process ()+bossLoop i =+ do job <- expect :: Process String+ found <- getCapable "worker"+ case found of+ [] ->+ do say "Where is my workforce?!"+ bossLoop i+ workers ->+ do say $ "'Delegating' task: " ++ job ++ " to " ++ show (pick workers)+ send (pick workers) job >> bossLoop (i + 1)+ where pick workers+ | length workers > 1 = head $ drop (mod i (length workers)) workers+ | otherwise = head workers++workLoop :: Process ()+workLoop =+ do work <- expect :: Process String+ case take 4 work of+ "stop" ->+ say "No more work for me!"+ _ ->+ do say $ "Doing task: " ++ work+ workLoop
+ src/Control/Distributed/Process/Zookeeper.hs view
@@ -0,0 +1,714 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE BangPatterns #-}++-- | Provides service and node discovery for Cloud Haskell applications using+-- a Zookeeper cluster for name registration, lookups and leader election.+-- Uses the hzk bindings to the Zookeeper multi-threaded C library.+--+-- Objectives and features:+--+-- * Compatible with @distributed-process-p2p@ API - can work as a drop-in+-- replacement.+--+-- * No dependencies beyond those already included by distributed-process, hzk and network-transport-tcp.+--+-- * Automatic registration of local names to Zookeeper.+--+-- * Global singleton processes with leader election and re-elections on+-- leader exit.+module Control.Distributed.Process.Zookeeper+ (+ -- * Initialization+ bootstrap+ , bootstrapWith+ , zkController+ , zkControllerWith+ -- * API+ , registerZK+ , getCapable+ , nsendCapable+ , registerCandidate+ , whereisGlobal+ -- * Compatibility+ , getPeers+ , nsendPeers+ -- * Config+ , Config(..)+ , defaultConfig+ -- * Utility+ , nolog+ , sayTrace+ , waitController+ ) where++import Database.Zookeeper (AclList (..),+ CreateFlag (..),+ Event (..), Watcher,+ ZKError (..),+ Zookeeper)+import qualified Database.Zookeeper as ZK++import Control.Applicative ((<$>))+import Control.Concurrent (myThreadId,+ newEmptyMVar,+ putMVar, putMVar,+ takeMVar, takeMVar,+ threadDelay)+import Control.DeepSeq (deepseq, NFData(..))+import Control.Exception (bracket, throwIO,+ throwTo)+import Control.Monad (forM, join, void)+import Control.Monad.Except (ExceptT (..), lift)+import Control.Monad.IO.Class (MonadIO)+import Control.Monad.Trans.Except (runExceptT, throwE)+import Data.Binary (Binary, decode,+ encode)+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BL+import Data.Foldable (forM_)+import Data.List (isPrefixOf, sort)+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Maybe (fromMaybe)+import Data.Monoid (mempty)+import Data.Typeable (Typeable)+import GHC.Generics (Generic)++import Control.Distributed.Process hiding (bracket,+ proxy)+import Control.Distributed.Process.Management+import Control.Distributed.Process.Node (newLocalNode,+ runProcess)+import Control.Distributed.Process.Serializable+import Network (HostName)+import Network.Socket (ServiceName)+import Network.Transport (closeTransport)+import Network.Transport.TCP (createTransport,+ defaultTCPParameters)+++data Command = Register !String !ProcessId !(SendPort (Either String ()))+ | GlobalCandidate !String !ProcessId !(SendPort (Either String ProcessId))+ | CheckCandidate !String+ | ClearCache !String+ | GetRegistered !String !(SendPort [ProcessId])+ | GetGlobal !String !(SendPort (Maybe ProcessId))+ | Exit+ deriving (Show, Typeable, Generic)++instance Binary Command+instance NFData Command++data Elect = Elect deriving (Typeable, Generic)++instance Binary Elect+instance NFData Elect++data State = State+ {+ nodeCache :: Map String [ProcessId]+ -- service nodes to remove each pid from when it exits+ , monPids :: Map ProcessId ([String],[String])+ , candidates :: Map String (String, ProcessId)+ , spid :: ProcessId+ , proxy :: Process () -> IO ()+ , conn :: Zookeeper+ }+++instance Show State where+ show State{..} = show ("nodes", nodeCache, "monPids", monPids)++data Config = Config+ {+ -- | Only register locally registered process names with zookeeper+ -- if the name begins with the given prefix. Default is "" which will+ -- register every locally registered process in the Zookeeper services+ -- node.+ registerPrefix :: String+ -- | An operation that will be called for trace level logging.+ -- 'defaultConfig' uses 'nolog'.+ , logTrace :: String -> Process ()+ -- | An operation that will be called for error logging.+ -- 'defaultConfig' uses 'say'.+ , logError :: String -> Process ()+ -- | The log level for the C Zookeper library. 'defaultConfig' uses+ -- 'ZK.ZLogWarn'.+ , zLogLevel :: ZK.ZLogLevel+ -- | The ACL to use for every node - see hzk documentation for+ -- 'ZK.AclList'. Note that if your nodes do not+ -- connect with the same identity, every node will need at least Read+ -- permission to all nodes created by this package.+ , acl :: ZK.AclList+ -- | Credentials for Zookeeper, see hzk 'ZK.addAuth' for details.+ , credentials :: Maybe (ZK.Scheme, ByteString)+ }++-- | A no-op that can be used for either of the loggers in 'Config'.+-- Because no actual I/O is performed, it fully evaluates the message so+-- thunks do not build up.+nolog :: String -> Process ()+nolog m = m `deepseq` return ()++-- | Simple formatter for trace output through 'say'.+sayTrace :: String -> Process ()+sayTrace = say . ("[C.D.P.Zookeeper: TRACE] - " ++)++-- | By default all local names are registered with zookeeper, and only+-- error messages are logged through 'say'.+--+-- > defaultConfig = Config {+-- > registerPrefix = ""+-- > , logTrace = nolog+-- > , logError = say . ("[C.D.P.Zookeeper: ERROR] - " ++)+-- > , zLogLevel = ZK.ZLogWarn+-- > , acl = OpenAclUnsafe+-- > , credentials = Nothing+-- > }+defaultConfig :: Config+defaultConfig = Config {+ registerPrefix = ""+ , logTrace = nolog+ , logError = say . ("[C.D.P.Zookeeper: ERROR] - " ++)+ , zLogLevel = ZK.ZLogWarn+ , acl = OpenAclUnsafe+ , credentials = Nothing+ }++-- |Register a name and pid as a service in Zookeeper. The controller+-- will monitor the pid and remove its child node from Zookeeper when it+-- exits.+--+-- Names will be registered at "\/distributed-process\/services\/\<name\>\/\<pid\>"+--+-- Note: By default all locally registered names (using 'register') will be+-- registered in Zookeeper under the same name by an MxAgent process. Use+-- this function if you want to register an anonymous pid or use+-- a different name than is registered with the local Process, or when+-- you are using a 'registerPrefix' to exclude the automatic+-- registration (see 'Config').+registerZK :: String -> ProcessId -> Process (Either String ())+registerZK name rpid = callZK $ Register name rpid++-- | Get a list of nodes advertised in Zookeeper. These are registered+-- when 'zkController' starts in path+-- "\/distributed-process\/controllers\/\<pid\>".+--+-- Note: this is included for API compatibility with+-- @distributed-process-p2p@ but its usage+-- would suggest discovery patterns that could be made more efficient+-- when using Zookeeper - i.e. just use 'getCapable'.+getPeers :: Process [NodeId]+getPeers = fmap processNodeId <$> callZK (GetRegistered controllersNode)++-- | Returns list of pids registered with the service name.+--+-- Results are cached by the controller until they are invalidated by+-- subsequent changes to the service node in Zookeeper, which is+-- communicated through a 'Watcher'. Data will be fetched from Zookeeper+-- only when it is changed and then requested again.+getCapable :: String -> Process [ProcessId]+getCapable name =+ callZK (GetRegistered (servicesNode </> name))++-- | Broadcast a message to a specific service on all registered nodes.+--+-- Note: this is included for API compatibility with @distributed-process-p2p@ but its usage+-- would suggest discovery patterns that could be made more efficient+-- when using Zookeeper - i.e. just use 'nfSendCapable' to+-- nfSend a broadcast directly to the registered process on each node.+nsendPeers :: Serializable a => String -> a -> Process ()+nsendPeers service msg = getPeers >>= mapM_ (\peer -> nsendRemote peer service msg)++-- | Broadcast a message to all pids registered with a particular service+-- name.+nsendCapable :: Serializable a => String -> a -> Process ()+nsendCapable service msg = getCapable service >>= mapM_ (`send` msg)++-- | Register a candidate process for election of a single process+-- associated to the given global name, and returns the 'ProcessId' of the+-- elected global (which may or may not be on the local node). The @Process ()@ argument is+-- only evaluated if this node ends up being the elected host for the+-- global. Calling this function subsequently on the same node for the same+-- name will replace the current candidate computation with the new one.+registerCandidate :: String -> Process () -> Process (Either String ProcessId)+registerCandidate name proc = stage >>= callZK . GlobalCandidate name+ where stage = spawnLocal $ (expect :: Process Elect) >> proc++-- | Find a registered global by name - see 'registerCandidate'.+whereisGlobal :: String -> Process (Maybe ProcessId)+whereisGlobal = callZK . GetGlobal++-- | Run a Zookeeper service process, and installs an MXAgent to+-- automatically register all local names in Zookeeper using+-- 'defaultConfig'.+--+-- > zkController = zkControllerWith defaultConfig+zkController :: String -- ^ The Zookeeper endpoint(s) -- comma separated list of host:port+ -> Process ()+zkController = zkControllerWith defaultConfig++-- | As 'zkController' but accept 'Config' options rather than assuming+-- defaults.+zkControllerWith :: Config+ -> String -- ^ The Zookeeper endpoint(s) -- comma separated list of host:port+ -> Process ()+zkControllerWith config@Config{..} keepers =+ do run <- spawnLinkedProxy+ waitInit <- liftIO newEmptyMVar+ liftIO $ ZK.setDebugLevel zLogLevel+ mthread <- liftIO myThreadId+ liftIO $ ZK.withZookeeper keepers 1000 (Just $ inits mthread waitInit) Nothing $ \rzh ->+ do init' <- takeMVar waitInit+ case init' of+ Left reason -> run $ do logError $ "Init failed: " ++ show reason+ die reason+ Right () -> run (server rzh config)+ where+ inits _ waitInit rzh SessionEvent ZK.ConnectedState _ =+ do esetup <- runExceptT $ eauth >> dosetup+ case esetup of+ Right _ -> putMVar waitInit (Right ())+ Left reason -> putMVar waitInit (Left reason)+ where+ eauth = lift auth >>= hoistEither+ auth =+ case credentials of+ Nothing -> return (Right ())+ Just (scheme, creds) ->+ do waitAuth <- newEmptyMVar+ ZK.addAuth rzh scheme creds $ \ res ->+ case res of+ Left reason ->+ let msg = "Authentication to Zookeeper failed: " ++ show reason+ in putMVar waitAuth (Left msg)+ Right () -> putMVar waitAuth (Right ())+ takeMVar waitAuth+ dosetup =+ do esetup <- runExceptT $+ do createAssert rzh "/" Nothing acl []+ createAssert rzh rootNode Nothing acl []+ createAssert rzh servicesNode Nothing acl []+ createAssert rzh controllersNode Nothing acl []+ createAssert rzh globalsNode Nothing acl []+ createAssert rzh (globalsNode </> "blarg") Nothing acl []+ case esetup of+ Right _ -> return ()+ Left reason ->+ throwE $ "FATAL: could not create system nodes in Zookeeper: "+ ++ show reason++ inits zkthread _ _ SessionEvent ZK.ExpiredSessionState _ =+ --for some reason having a linked proxy die here was not getting it done+ throwTo zkthread $ userError "Zookeeper session expired."++ inits _ _ _ _ _ _ = return ()++server :: Zookeeper -> Config -> Process ()+server rzh config@Config{..} =+ do pid <- getSelfPid+ register controller pid+ watchRegistration config+ proxy' <- spawnLinkedProxy+ regself <- create rzh (controllersNode </> pretty pid)+ (pidBytes pid) acl [Ephemeral]+ case regself of+ Left reason ->+ let msg = "Could not register self with Zookeeper: " ++ show reason+ in logError msg >> die msg+ Right _ -> return ()+ let loop st =+ let recvCmd = match $ \command -> case command of+ Exit -> return ()+ _ ->+ do eresult <- runExceptT $ handle st command+ case eresult of+ Right st' ->+ do logTrace $ "State of: " ++ show st' ++ " - after - "+ ++ show command+ loop st'+ Left reason ->+ do logError $ "Error handling: " ++ show command+ ++ " : " ++ show reason+ loop st+ recvMon = match $ \(ProcessMonitorNotification _ dead _) ->+ reap st dead >>= loop+ in logTrace (show st) >> receiveWait [recvCmd, recvMon]+ void $ loop (State mempty mempty mempty pid proxy' rzh)+ where+ reap st@State{..} pid =+ do let (services, globals) = fromMaybe ([],[]) (Map.lookup pid monPids)+ forM_ services $ \name ->+ deleteNode (servicesNode </> name </> pretty pid)+ forM_ globals $ \name ->+ deleteNode (globalsNode </> name)++ return st{monPids = Map.delete pid monPids}+ where+ deleteNode node =+ do result <- liftIO $ ZK.delete rzh node Nothing+ case result of+ Left reason -> logError $ show reason+ ++ " - failed to delete "+ ++ node+ _ -> logTrace $ "Reaped " ++ node++ handle st@State{..} (Register name rpid reply) =+ do let node = servicesNode </> name </> pretty rpid+ createAssert rzh (servicesNode </> name) Nothing acl []+ result <- create rzh node (pidBytes rpid) acl [Ephemeral]+ case result of+ Right _ -> lift $+ do logTrace $ "Registered " ++ node+ nfSendChan reply (Right ())+ void $ monitor rpid+ let apService (a',b') (a, b) = (a' ++ a, b' ++ b)+ return st{monPids = Map.insertWith apService rpid ([name],[]) monPids}+ Left reason -> lift $+ do logError $ "Failed to register name: " ++ node ++ " - " ++ show reason+ nfSendChan reply (Left $ show reason)+ return st++++ handle st@State{..} (ClearCache node) =+ return st{nodeCache = Map.delete node nodeCache}++ handle st@State{..} (GetRegistered node reply) =+ do epids <- case Map.lookup node nodeCache of+ Just pids -> return (Right pids)+ Nothing -> getChildPids rzh node (Just $ watchCache st node)+ lift $ case epids of+ Right pids ->+ do nfSendChan reply pids+ return st{nodeCache = Map.insert node pids nodeCache}+ Left reason ->+ do logError $ "Retrieval failed for node: " ++ node ++ " - " ++ show reason+ nfSendChan reply []+ return st{nodeCache = Map.delete node nodeCache}++ handle st (GlobalCandidate n c r) = handleGlobalCandidate config st n c r++ handle st@State{..} (GetGlobal name reply) =+ let gname = globalsNode </> name in+ case Map.lookup gname nodeCache of+ Just (pid : _) -> lift $+ do nfSendChan reply (Just pid)+ return st+ _ ->+ do elected <- getElected+ lift $ nfSendChan reply elected+ return $ maybe st+ (\pid -> st {nodeCache = Map.insert gname [pid] nodeCache})+ elected+ where+ getElected =+ do children <- getGlobalIds conn name+ case children of+ [] -> return Nothing+ (first: _) ->+ Just <$> getPid conn (globalsNode </> name </> first)+ (Just $ watchCache st name)++ handle st@State{..} (CheckCandidate name) =+ case Map.lookup name candidates of+ Just (myid, staged) -> snd <$> mayElect st name myid staged+ Nothing -> return st++ handle st Exit = return st --satisfy exhaustiveness checker++ pretty pid = drop 6 (show pid)++handleGlobalCandidate :: Config -> State+ -> String -> ProcessId -> SendPort (Either String ProcessId)+ -> ExceptT ZKError Process State+handleGlobalCandidate Config{..} st@State{..} name proc reply+ | Just (myid, staged) <- Map.lookup name candidates =+ case Map.lookup (globalsNode </> name) nodeCache of+ Just (pid : _) -> lift $+ do exit staged "New candidate staged."+ nfSendChan reply (Right pid)+ return st {candidates = Map.insert name (myid, proc) candidates}++ _ -> respondElect myid proc (Just staged)+ | otherwise =+ do myid <- registerGlobalId proc+ respondElect myid proc Nothing+ where+ respondElect myid staged mprev = lift $+ do eresult <- runExceptT $ mayElect st name myid staged+ case eresult of+ Right (pid, st') ->+ do nfSendChan reply (Right pid)+ forM_ mprev (`exit` "New candidate staged.")+ return st'+ Left reason ->+ do nfSendChan reply (Left $ show reason)+ return st++ registerGlobalId staged =+ do let pname = globalsNode </> name+ createAssert conn pname Nothing acl []+ node <- create conn (pname </> "")+ (pidBytes staged)+ acl+ [Ephemeral, Sequence] >>= hoistEither+ return $ extractId node+ where+ extractId s = trimId (reverse s) ""+ where+ trimId [] _ = error $ "end of string without delimiter / in " ++ s+ trimId ('/' : _) str = str+ trimId (n : rest) str = trimId rest (n : str)++mayElect :: State -> String -> String -> ProcessId -> ExceptT ZKError Process (ProcessId, State)+mayElect st@State{..} name myid staged =+ do others <- getGlobalIds conn name+ let first : _ = others+ if myid == first+ then+ do lift $ nfSend staged Elect+ st' <- lift $ cacheNMonitor staged+ return (staged, st')+ else+ do let prev = findPrev others+ watchFirst <- if prev == first then return $ Just watchPid+ else do void $ liftIO (ZK.exists+ conn+ (globalsNode </> name </> prev)+ (Just watchPid)) >>= hoistEither+ return Nothing+ pid <- getPid conn (globalsNode </> name </> first) watchFirst+ return (pid, stCache)+ where+ findPrev (prev : next : rest) = if next == myid+ then prev+ else findPrev (next : rest)+ findPrev _ = error "impossible: couldn't find myself in election"+ watchPid _ ZK.DeletedEvent ZK.ConnectedState _ =+ proxy $ nfSend spid (CheckCandidate name)+ watchPid _ _ _ _ = return ()++ stCandidate = st{candidates = Map.insert name (myid, staged) candidates}+ stCache = stCandidate {nodeCache = Map.insert (globalsNode </> name) [staged] nodeCache}+ cacheNMonitor pid =+ do void $ monitor pid+ liftIO $ void $ ZK.exists conn (globalsNode </> name </> myid) (Just $ watchCache st (globalsNode </> name))+ return stCacheMon+ where+ stCacheMon =+ let apGlobal (a', b') (a, b) = (a' ++ a, b' ++ b) in+ st{ candidates = Map.delete name candidates+ , monPids = Map.insertWith apGlobal+ pid+ ([],[name </> myid]) monPids+ , nodeCache = Map.insert (globalsNode </> name) [pid] nodeCache+ }++getGlobalIds :: MonadIO m+ => Zookeeper -> String -> ExceptT ZKError m [String]+getGlobalIds conn name = liftIO $+ do echildren <- ZK.getChildren conn (globalsNode </> name) Nothing+ case echildren of+ Left NoNodeError -> return []+ Left reason ->+ do putStrLn $ "Could not fetch globals for " ++ name ++ " - " ++ show reason+ return []+ Right children -> return (sort children)++getPid :: MonadIO m+ => Zookeeper -> String -> Maybe Watcher+ -> ExceptT ZKError m ProcessId+getPid conn name watcher =+ do res <- liftIO $ ZK.get conn name watcher+ case res of+ Right (Just bs, _) -> return (decode $ BL.fromStrict bs)+ Right _ -> throwE NothingError+ Left reason -> throwE reason++getChildPids :: MonadIO m+ => Zookeeper -> String -> Maybe Watcher+ -> m (Either ZKError [ProcessId])+getChildPids rzh node watcher = liftIO $+ do enodes' <- ZK.getChildren rzh node watcher+ case enodes' of+ Left NoNodeError -> return $ Right []+ _ ->+ runExceptT $+ do children <- hoistEither enodes'+ forM children $ \child ->+ do eresult <- liftIO $ ZK.get rzh (node </> child) Nothing+ case eresult of+ Left reason -> throwE reason+ Right (Nothing, _) -> throwE NothingError+ Right (Just bs, _) -> return (decode $ BL.fromStrict bs)++watchCache :: State -> String -> a -> b -> ZK.State -> c -> IO ()+watchCache State{..} node _ _ ZK.ConnectedState _ =+ proxy $ nfSend spid (ClearCache node)++watchCache _ _ _ _ _ _ = return ()++watchRegistration :: Config -> Process ()+watchRegistration Config{..} = do+ let initState = [] :: [MxEvent]+ void $ mxAgent (MxAgentId "zookeeper:name:listener") initState [+ mxSink $ \ev -> do+ let act =+ case ev of+ (MxRegistered _ "zookeeper:name:listener") -> return ()+ (MxRegistered pid name')+ | prefixed name' -> liftMX $+ do ereg <- registerZK name' pid+ case ereg of+ Left reason ->+ logError $ "Automatic registration failed for name: "+ ++ name' ++ " - " ++ reason+ _ -> return ()+ | otherwise -> return ()+ _ -> return ()+ act >> mxReady ]+ liftIO $ threadDelay 10000+ where prefixed = isPrefixOf registerPrefix++-- | Wait for zkController to startup and register iteself.+-- This is only useful if you are *not* using a 'bootstrap'+-- function to start your node, but rather starting the node yourself+-- and using one of the 'zkController' functions.+waitController :: Int -> Process (Maybe ())+waitController timeout =+ do res <- whereis controller+ case res of+ Nothing ->+ do let timeleft = timeout - 10000+ if timeleft <= 0 then return Nothing+ else do liftIO (threadDelay 10000)+ waitController timeleft+ Just _ -> return (Just ())++stopController :: Process ()+stopController =+ do res <- whereis controller+ case res of+ Nothing -> say "Could not find controller to stop it."+ Just pid -> nfSend pid Exit++hoistEither :: Monad m => Either e a -> ExceptT e m a+hoistEither = ExceptT . return++(</>) :: String -> String -> String+l </> r = l ++ "/" ++ r++servicesNode :: String+servicesNode = rootNode </> "services"++controllersNode :: String+controllersNode = rootNode </> "controllers"++globalsNode :: String+globalsNode = rootNode </> "globals"++rootNode :: String+rootNode = "/distributed-process"++createAssert :: MonadIO m+ => Zookeeper -> String -> Maybe BS.ByteString -> AclList -> [CreateFlag]+ -> ExceptT ZKError m ()+createAssert z n d a f = create z n d a f >>= eitherExists+ where+ eitherExists (Right _) = return ()+ eitherExists (Left NodeExistsError) = return ()+ eitherExists (Left reason) = throwE reason++pidBytes :: ProcessId -> Maybe BS.ByteString+pidBytes = Just . BL.toStrict . encode++controller :: String+controller = "zookeeper:controller"++-- We should not have any big surprises due to serialization semantics+-- because all messaging will always be local in this module - it will never be necessary+-- to actually serialize. Still, we do want to+-- preserve expectations about which process is doing the evaluation.+nfSendChan :: (Binary a, Typeable a, NFData a) => SendPort a -> a -> Process ()+nfSendChan !port !msg = unsafeSendChan port (msg `deepseq` msg)++nfSend :: (Binary a, Typeable a, NFData a) => ProcessId -> a -> Process ()+nfSend !pid !msg = unsafeSend pid (msg `deepseq` msg)++callZK :: Serializable a => (SendPort a -> Command) -> Process a+callZK command =+ do Just pid <- whereis controller+ (nfSendCh, replyCh) <- newChan+ link pid+ nfSend pid (command nfSendCh)+ result <- receiveChan replyCh+ unlink pid+ return result++-- | Create a process runner that communicates+-- through an MVar - so you can call process actions from IO+--+-- This is linked bidirectionally - if either the proxy or parent exits the+-- other will as well.+spawnLinkedProxy :: Process (Process a -> IO a)+spawnLinkedProxy =+ do action <- liftIO newEmptyMVar+ result <- liftIO newEmptyMVar+ self <- getSelfPid+ pid <- spawnLocal $+ let loop = join (liftIO $ takeMVar action)+ >>= liftIO . putMVar result+ >> loop in link self >> loop+ link pid+ return (\f -> putMVar action f >> takeMVar result)++create :: MonadIO m => Zookeeper -> String -> Maybe BS.ByteString+ -> AclList -> [CreateFlag] -> m (Either ZKError String)+create z n d a = liftIO . ZK.create z n d a++-- | Create a new Cloud Haskell node on the provided IP/Port and start+-- a Zookeeper-backed controller process ('zkController') with a default configuration+-- connected to the provided Zookeeper server list.+-- Finally execute the supplied Process computation.+--+-- > bootstrap = bootstrapWith defaultConfig+bootstrap :: HostName -- ^ Hostname or IP this Cloud Haskell node will listen on.+ -> ServiceName -- ^ Port or port name this node will listen on.+ -> String -- ^ The Zookeeper endpoint(s) -- comma separated list of host:port+ -> RemoteTable -- ^ Cloud Haskell 'RemoteTable' to use in the new node.+ -> Process () -- ^ Process computation to run in the new node.+ -> IO ()+bootstrap = bootstrapWith defaultConfig++-- | Create a new Cloud Haskell node on the provided IP/Port and start+-- a Zookeeper-backed controller process ('zkController') connected to the provided+-- Zookeeper server list and finally execute the supplied Process computation.+bootstrapWith :: Config -- ^ controller configuration+ -> HostName -- ^ Hostname or IP this Cloud Haskell node will listen on.+ -> ServiceName -- ^ Port or port name this node will listen on.+ -> String -- ^ The Zookeeper endpoint(s) -- comma separated list of host:port+ -> RemoteTable -- ^ Cloud Haskell 'RemoteTable' to use in the new node.+ -> Process () -- ^ Process computation to run in the new node.+ -> IO ()+bootstrapWith config host port zservs rtable proc =+ bracket openTransport closeTransport exec+ where+ openTransport =+ do mtcp <- createTransport host port defaultTCPParameters+ case mtcp of+ Right tcp -> return tcp+ Left reason -> throwIO reason+ exec tcp =+ do node <- newLocalNode tcp rtable+ runProcess node $+ do zkpid <- spawnLocal $ zkControllerWith config zservs+ link zkpid+ found <- waitController 100000+ case found of+ Nothing -> die "Timeout waiting for Zookeeper controller to start."+ Just () -> proc+ stopController
+ test/Spec.hs view
@@ -0,0 +1,279 @@+{-# OPTIONS_GHC -fno-warn-missing-signatures #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE FlexibleContexts #-}++module Main (main, spec) where++import Test.Hspec+import Control.Concurrent.Lifted+import Control.Distributed.Process.MonadBaseControl()+import Control.Distributed.Process hiding (bracket)+import Control.Distributed.Process.Zookeeper+import Control.Distributed.Process.Serializable (Serializable)+import qualified Database.Zookeeper as ZK+import Control.Distributed.Process.Node (initRemoteTable)+import Control.DeepSeq (NFData, deepseq)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad (void)+import Control.Monad.IO.Class (MonadIO)+import Control.Exception.Enclosed (catchAny)+import Control.Exception.Lifted (throw)+import Data.ByteString.Char8 (pack)++main :: IO ()+main = hspec spec++spec :: Spec+spec = do+ describe "Test harness" $ do++ it "can do basic Process messaging" $ do+ testBoot $ do+ parent <- getSelfPid+ child <- spawnLocal $ do+ saysee <- texpect :: Process String+ send parent ("I said: " ++ saysee)+ send child "foo"+ said <- texpect :: Process String+ return said+ `shouldReturn` "I said: foo"++ describe "C.D.P. Zookeeper" $ do++ it "will register local names with zookeeper" $ do+ testBoot $ do+ self <- getSelfPid+ register "testy" self+ threadDelay 20000+ [found] <- getCapable "testy"+ return (found == self)+ `shouldReturn` True++ it "will only register prefixed local names if configured" $ do+ testBootWith defaultConfig {registerPrefix = "zk:"} zookeepers $ do+ self <- getSelfPid+ register "testr" self+ register "zk:testr2" self+ threadDelay 10000+ [] <- getCapable "testr"+ [found] <- getCapable "zk:testr2"+ return (found == self)+ `shouldReturn` True++ it "enables broadcast to specific services" $ do+ testBoot $ do+ one <- newEmptyMVar+ two <- newEmptyMVar+ let slave mv = void $ fork $ testBoot $+ do self <- getSelfPid+ register "test-broadcast" self+ threadDelay 50000 --finish registration+ putMVar mv ()+ "foundme" <- expect+ putMVar mv ()+ expect :: Process ()+ slave one+ slave two+ takeMVar one+ takeMVar two --hopefully they are both registered by now!+ nsendCapable "test-broadcast" "foundme"+ takeMVar one+ takeMVar two+ `shouldReturn` ()++ it "enables broadcast to all nodes" $ do+ testBoot $ do+ let slave mv =+ do self <- getSelfPid+ register "test-broadcast" self+ threadDelay 50000 --finish registration+ putMVar mv ()+ "foundme" <- expect+ putMVar mv ()+ one <- forkSlave slave+ two <- forkSlave slave+ takeMVar one+ takeMVar two --hopefully they are both registered by now!+ nsendPeers "test-broadcast" "foundme"+ takeMVar one+ takeMVar two+ `shouldReturn` ()++ it "will run a single process with a global name" $ do+ testBoot $ do+ pid <- getSelfPid+ let racer mv =+ do void $ registerCandidate "test-global" $+ do send pid ()+ expect :: Process ()+ putMVar mv ()+ expect :: Process ()++ one <- forkSlave racer+ two <- forkSlave racer+ takeMVar one+ takeMVar two+ -- we should only get one+ () <- expect :: Process ()+ expectTimeout 100000+ `shouldReturn` (Nothing :: Maybe ())++ it "will elect a new global when one exits" $ do+ testBoot $ do+ pid <- getSelfPid++ let racer mv =+ do Right gpid <- registerCandidate "test-global2" $+ do send pid ()+ expect :: Process ()+ putMVar mv gpid+ expect :: Process ()++ one <- forkSlave racer+ two <- forkSlave racer+ three <- forkSlave racer+ first <- takeMVar one+ void $ takeMVar two+ void $ takeMVar three+ exit first "because"+ () <- expect :: Process ()+ () <- expect :: Process ()+ Just second <- whereisGlobal "test-global2"+ -- make sure only ONE candidate is elected on exit of+ -- first+ Nothing <- expectTimeout 100000 :: Process (Maybe ())+ return (first /= second)+ `shouldReturn` True++ it "will not leak processes when a candidate is attempted to register again" $ do+ testBoot $ do+ nodeid <- getSelfNode++ Right gpid <- registerCandidate "test-global3" $+ (expect :: Process ())++ Right stats <- getNodeStats nodeid+ let b4 = nodeStatsProcesses stats++ Right same <- registerCandidate "test-global3" $+ (expect :: Process ())++ Right stats' <- getNodeStats nodeid+ let after' = nodeStatsProcesses stats'++ return ((b4 + 1 == after') && (gpid == same))+ `shouldReturn` True++ it "will clear the name cache when a service exits" $ do+ testBoot $ do+ parent <- getSelfPid+ void $ spawnLocal $+ do self <- getSelfPid+ register "test-cache" self+ threadDelay 10000+ (expect :: Process ())+ send parent ()+ pid <- waitZKRegistration "test-cache"+ send pid ()+ (expect :: Process ())+ threadDelay 10000+ getCapable "test-cache"+ `shouldReturn` []++ it "will clear the name cache when a global process exits" $ do+ testBoot $ do+ Right gpid <- registerCandidate "test-cache2" $+ (expect :: Process ())+ send gpid ()+ threadDelay 100000+ whereisGlobal "test-cache2"+ `shouldReturn` Nothing++ it "will strictly evaluate command arguments in the client process" $ do+ testBoot $ do+ result <- getSelfPid >>= registerZK (error "asplode")+ return result+ -- if the Name field of the Register constructor were+ -- not strict, we'd get a ProcessLinkException here+ -- because it would be evaluated in the Controller+ -- process+ `shouldThrow` anyErrorCall++ describe "supports authorization" $ do++ it "can setup with restricted acl" $ do+ testBootAuth "user:rightpassword" $ do+ -- we just run this for the initialization with+ -- the "correct" credentials and restricted ACL+ return ()+ `shouldReturn` ()++ it "will fail if there is an ACL with different credentials" $ do+ testBootAuth "user:wrongpassword" $ do+ return ()+ `shouldThrow` linkException++linkException :: ProcessLinkException -> Bool+linkException _ = True++zookeepers = "localhost:2181"++forkSlave ma =+ do mv <- newEmptyMVar+ void $ fork $ testBoot (ma mv)+ return mv++testBoot :: (MonadBaseControl IO io, MonadIO io, Show a, NFData a)+ => Process a -> io a+testBoot = testBootWith defaultConfig zookeepers -- {logTrace = sayTrace}++testBootAuth :: (MonadBaseControl IO io, MonadIO io, Show a, NFData a)+ => String -> Process a -> io a+testBootAuth creds = testBootWith defaultConfig { credentials = Just ("digest", pack creds)+ , acl = ZK.CreatorAll+ }+ (zookeepers ++ "/testauth")++testBootWith :: (MonadBaseControl IO io, MonadIO io, Show a, NFData a)+ => Config -> String -> Process a -> io a+testBootWith config servers ma = testProcessTimeout 1000 (ma >>= waitReturn )+ (liftIO . bootstrapWith config+ "localhost"+ "0"+ servers+ initRemoteTable)++waitReturn a = do threadDelay 100000; return a++testProcessTimeout :: (Show a, NFData a, MonadBaseControl IO io)+ => Int -> Process a -> (Process () -> io ()) -> io a+testProcessTimeout timeout ma runProc = do+ resultMV <- newEmptyMVar+ void . fork $+ catchAny (runProc $+ do a <- ma+ a `deepseq` putMVar resultMV a)+ (\e -> putMVar resultMV (throw e))+ !result <- waitMVar resultMV+ return result+ where+ waitMVar mv = loop 0+ where loop lapsed+ | lapsed < timeout =+ tryTakeMVar mv >>= maybe (delay >> loop (lapsed + 1))+ return+ | otherwise = error "Execution of Process test timed-out."+ delay = threadDelay 2000++texpect :: Serializable a => Process a+texpect = do+ gotit <- expectTimeout 500000 -- 500ms may as well be never+ case gotit of+ Nothing -> error "Timed out in test expect"+ Just v -> return v++waitZKRegistration name = loop+ where loop = do found <- getCapable name+ case found of+ [] -> threadDelay 1000 >> loop+ pid : _ -> return pid