majordomo (empty) → 0.1
raw patch · 8 files changed
+447/−0 lines, 8 filesdep +basedep +bytestringdep +cmdargssetup-changed
Dependencies added: base, bytestring, cmdargs, majordomo, monad-loops, old-locale, threads, time, unix, zeromq-haskell
Files
- LICENSE +30/−0
- README.md +12/−0
- Setup.hs +2/−0
- echoworker.hs +31/−0
- lib/System/Network/ZMQ/MDP/Client.hs +72/−0
- lib/System/Network/ZMQ/MDP/Worker.hs +217/−0
- majordomo.cabal +49/−0
- mdp_client.hs +34/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c)2012, Mark Wotton++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 Mark Wotton 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.
+ README.md view
@@ -0,0 +1,12 @@+++from http://rfc.zeromq.org/spec:7+"The Majordomo Protocol (MDP) defines a reliable service-oriented+request-reply dialog between a set of client applications, a broker+and a set of worker applications. MDP covers presence, heartbeating,+and service-oriented request-reply processing. It originated from the+Majordomo pattern defined in Chapter 4 of the Guide."++This is an implementation for Haskell.+Examples of use can be found in echo_worker.hs and mdp_client.hs+(which doubles as a helpful command line tool for issuing MDP commands)
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ echoworker.hs view
@@ -0,0 +1,31 @@+{-# LANGUAGE OverloadedStrings #-}+module Main where+import Data.ByteString hiding (map)++import qualified System.Network.ZMQ.MDP.Worker as W+import Data.Foldable+import Control.Concurrent.Thread.Group as TG+import System.Posix.Signals+import qualified Control.Concurrent as CC++threaded :: [IO ()] -> IO ()+threaded actions = do+ tg <- TG.new+ tids <- mapM (TG.forkIO tg) actions+ _ <- installHandler sigINT (CatchOnce $ do+ Prelude.putStrLn "worker caught an interrupt"+ forM_ tids ((\x -> print x >> CC.killThread x) . fst)+ ) Nothing+ Prelude.putStrLn "waiting..."+ TG.wait tg+ Prelude.putStrLn "all dead"++main :: IO ()+main = threaded $ flip map [1..4] $ \tid ->+ W.withWorker "tcp://127.0.0.1:5555" "echo"+ (\x -> return ("hi there, " `append` x))++ + -- withContext 1 $ \c ->+ -- threaded $ flip map [1..4] $ \tid ->+ -- W.start W.defaultWorker { }
+ lib/System/Network/ZMQ/MDP/Client.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE OverloadedStrings #-}+module System.Network.ZMQ.MDP.Client (+ -- | Types+ Response(..),+ ClientSocket, -- opaque datatype + ClientError(..),+ -- | Functions+ sendAndReceive,+ withClientSocket+) where++import Data.ByteString.Char8+import qualified System.ZMQ as Z+import System.ZMQ hiding(send)+import Control.Applicative+import System.Timeout++data Protocol = MDCP01++data Response = Response { protocol :: Protocol,+ service :: ByteString,+ response :: [ByteString] }+++-- this can either be XReq or Req...+data ClientSocket = ClientSocket { clientSocket :: Socket Req }+++data ClientError = ClientTimedOut+ | ClientBadProtocol++withClientSocket :: String -> (ClientSocket -> IO a) -> IO a+withClientSocket socketAddress io =+ withContext 1 $ \c -> withSocket c Req $ \s -> do+ connect s socketAddress+ io (ClientSocket s)+++sendAndReceive :: ClientSocket -> ByteString -> [ByteString] -> IO (Either ClientError Response)+sendAndReceive mdpcs svc msgs =+ do -- Z.send sock "" [SndMore]+ Z.send sock "MDPC01" [SndMore]+ Z.send sock svc [SndMore]+ sendAll sock msgs+ maybeprot <- timeout (1000000 * 3) $ receive sock []+ case maybeprot of+ Nothing -> return $ Left ClientTimedOut+ Just "MDPC01" -> do+ res <- Response MDCP01 <$> receive sock [] <*> receiveTillEmpty sock+ return $ Right res+ _ -> return $ Left ClientBadProtocol+ where+ sock = clientSocket mdpcs++--+-- Helper functions+--++receiveTillEmpty :: Socket a -> IO [ByteString]+receiveTillEmpty sock = do+ more <- Z.moreToReceive sock+ if more+ then (:) <$> receive sock [] <*> receiveTillEmpty sock+ else return []++--+-- Sends a multipart message+--+sendAll :: Socket a -> [ByteString] -> IO ()+sendAll _sock [] = return ()+sendAll sock (m:[]) = Z.send sock m []+sendAll sock (m:ms) = Z.send sock m [SndMore] >> sendAll sock ms
+ lib/System/Network/ZMQ/MDP/Worker.hs view
@@ -0,0 +1,217 @@+{-# LANGUAGE OverloadedStrings #-}+module System.Network.ZMQ.MDP.Worker where+import Data.ByteString.Char8+import Data.Int+import Prelude hiding (putStr, putStrLn)+import qualified Prelude+import qualified System.ZMQ as Z+import System.ZMQ hiding(receive)+import Control.Applicative++import Control.Exception+import Control.Monad+import Data.Time.Clock+import Data.Time.Format+import System.Locale+import Control.Concurrent+import System.Timeout++data Protocol = WORKER_PROTOCOL+instance Show Protocol where+ show WORKER_PROTOCOL = "MDPW01"++type Address = ByteString -- cheaty.++data Response = Response { envelope :: ! [Address],+ body :: ! [ByteString] }++data ResponseCode = REPLY | READY | WORKER_HEARTBEAT+instance Show ResponseCode where+ show REPLY = "\003"+ show READY = "\001"+ show WORKER_HEARTBEAT = "\004"++data CommandCode = REQUEST | HEARTBEAT | DISCONNECT+instance Show CommandCode where+ show REQUEST = "\002"+ show HEARTBEAT = "\004"+ show DISCONNECT = "\005"++parseCommand :: ByteString -> Maybe CommandCode+parseCommand "\002" = Just REQUEST+parseCommand "\004" = Just HEARTBEAT+parseCommand "\005" = Just DISCONNECT+parseCommand _ = Nothing+++type MDError = ByteString++ +read_till_empty :: Socket a -> IO [ByteString]+read_till_empty sock = do+ frame <- Z.receive sock []+ if frame == ""+ then return []+ else (frame:) <$> read_till_empty sock+++send_all :: Socket a -> [ByteString] -> IO ()+send_all sock = go+ where go [l] = Z.send sock l []+ go (x:xs) = Z.send sock x [SndMore] >> go xs+ go [] = error "empty send not allowed"+++send_to_broker :: Socket a -> ResponseCode -> [ByteString] ->+ [ByteString] -> IO ()+send_to_broker sock cmd option message =+ send_all sock $ ["",+ pack $ show WORKER_PROTOCOL,+ pack $ show cmd] ++ option ++ message++{-+Frame 0: Empty frame+Frame 1: "MDPW01" (six bytes, representing MDP/Worker v0.1)+Frame 2: 0x03 (one byte, representing REPLY)+Frame 3: Client address (envelope stack)+Frame 4: Empty (zero bytes, envelope delimiter)+Frames 5+: Reply body (opaque binary)+-}++-- FIXME might we ever want to send a multi-part body?+--+send_response :: Socket a -> Response -> IO ()+send_response sock resp = send_to_broker sock REPLY (envelope resp) (body resp)++-- send_to_broker sock REPLY Nothing [reply])++whileJust :: Monad m => (b -> m (Maybe b)) -> b -> m b+whileJust action seed = action seed >>= maybe (return seed) (whileJust action)++start :: WorkerState a -> IO ()+start worker = forever (withBroker readTillDrop worker)++readTillDrop :: Socket a -> WorkerState a1 -> IO (WorkerState a1)+readTillDrop sock worker = whileJust (receive sock) worker+++data WorkerState a = WorkerState { heartbeat_at :: ! UTCTime,+ liveness :: ! Int,+ heartbeat :: ! Int64,+ reconnect :: ! Int,+ broker :: String,+ context :: System.ZMQ.Context,+ svc :: ByteString,+ handler :: ByteString -> IO ByteString+ }++epoch :: UTCTime+epoch = buildTime defaultTimeLocale []++lIVENESS :: Int+lIVENESS = 3+++withWorker :: String -> ByteString -> (ByteString -> IO ByteString) -> IO ()+withWorker broker_ service_ io =+ withContext 1 $ \c -> + start WorkerState { broker = broker_,+ context = c,+ svc = service_,+ handler = io,+ liveness = 1,+ heartbeat_at = epoch,+ heartbeat = 2,+ reconnect = 2+ }+ +++withBroker :: (Socket XReq -> WorkerState a -> IO b) -> WorkerState t -> IO b+withBroker go worker =+ withSocket (context worker) XReq $ \sock -> do+ loggedPut ( "connecting to broker " ++ broker worker)+ connect sock (broker worker)+ send_to_broker sock READY [svc worker] []+ now <- getCurrentTime+ let time = addUTCTime (fromIntegral $ heartbeat worker) now+ loggedPut ("beat at:" ++ show time)+ go sock worker { liveness = lIVENESS,+ heartbeat_at = time+ }++loggedPut :: String -> IO ()+loggedPut _res = return () -- do+-- Prelude.putStr . show =<< getCurrentTime+-- Prelude.putStrLn (": " ++ res)++receive :: Socket a -> WorkerState a1 -> IO (Maybe (WorkerState a2))+receive sock worker = do loggedPut "polling"+ next <- getMessage+ case next of+ Nothing -> loggedPut "no message" >> return Nothing+ Just w -> loggedPut "message!" >> postCheck w+ where++ getMessage = do+ -- this timeout is different in 3.1+ -- loggedPut $ "Polling socket: should finish in " ++ (show (heartbeat worker)) ++ "seconds"++ [S _ polled] <- poll [S sock In] $ 1000000 * heartbeat worker+ -- use this in 3.1+ -- polled <- timeout (1000000 * fromIntegral (heartbeat worker)) $ Z.receive sock []++ case polled of+ None -> noMessage+ In -> Z.receive sock [] >>= handleEvent+ -- case polled of+ -- Nothing -> noMessage+ -- Just s -> handleEvent s++ noMessage :: IO (Maybe (WorkerState b))+ noMessage = do+ let live = liveness worker - 1+ if liveness worker == 0+ then loggedPut "reconnecting" >> threadDelay (1000000 * reconnect worker) >> return Nothing+ else return $ Just worker { liveness = live }+ postCheck :: WorkerState a -> IO (Maybe (WorkerState a))+ postCheck worker = do+ --loggedPut "postcheck"+ time <- {-# SCC "getCurrentTime" #-} getCurrentTime+ -- loggedPut $ "beat at " ++ show (heartbeat_at worker)+ if {-# SCC "time_comparison" #-} time > heartbeat_at worker+ then do --loggedPut "sending heartbeat"+ {-# SCC "postcheck_send" #-} send_to_broker sock WORKER_HEARTBEAT [] []+ --loggedPut "sent heartbeat!"+ {-# SCC "postcheck_return" #-} return $ Just $ updateWorkerTime worker time+ else return $ Just worker++ updateWorkerTime w time =+ w { heartbeat_at = {-# SCC "addtime" #-} addUTCTime (fromIntegral $! heartbeat w) time}++ handleEvent header = do+ let zrecv = Z.receive sock []+ -- loggedPut "handling"+ assert (header == "") (return ())+ prot <- zrecv+ assert (prot == "MDPW01") (return ())+ -- ideally, we'd encapsulate the process of reading+ -- the whole thing in in the parser. this will do for now though.+ command <- parseCommand <$> zrecv+ let new_worker = worker { liveness = lIVENESS }+ case command of+ Just REQUEST -> do+ addresses <- read_till_empty sock+ msg <- zrecv+ reply_string <- (handler worker) msg+ send_response sock Response { envelope = addresses,+ body = [reply_string] }+ return $ Just new_worker+ Just HEARTBEAT -> do+ -- loggedPut "handling a heartbeat"+ return $ Just new_worker+ Just DISCONNECT -> do+ -- loggedPut "handling a disconnect"+ return Nothing+ Nothing -> error "borked"+
+ majordomo.cabal view
@@ -0,0 +1,49 @@+Name: majordomo+Version: 0.1+Synopsis: Majordomo protocol for ZeroMQ+Description: The Majordomo Protocol (MDP) defines a reliable+ service-oriented request-reply dialog between a+ set of client applications, a broker and a set of+ worker applications. MDP covers presence,+ heartbeating, and service-oriented request-reply+ processing. It originated from the Majordomo+ pattern defined in Chapter 4 of the Guide. ++ http://rfc.zeromq.org/spec:7++License: BSD3+License-file: LICENSE+Extra-Source-Files: README.md+Author: Mark Wotton, Sean Seefried+Maintainer: mark@ninjablocks.com, sean@ninjablocks.com++Category: Network+Build-type: Simple+Cabal-version: >= 1.8+Source-Repository head+ type: git+ location: git://github.com/ninjablocks/majordomo.git++Executable mdp_client+ Main-is: mdp_client.hs+ Build-depends: majordomo,base, bytestring, cmdargs+ ghc-options: -O2 -threaded++Executable echoworker+ Main-is: echoworker.hs+ ghc-options: -O2 -threaded+ ghc-prof-options: -auto-all+ Build-depends: majordomo,base, bytestring, threads, unix++Library+ -- Modules exported by the library.+ Exposed-modules: System.Network.ZMQ.MDP.Worker,+ System.Network.ZMQ.MDP.Client+ -- Packages needed in order to build this package.+ hs-source-dirs: lib+ ghc-prof-options: -auto-all+ ghc-options: -O2+ Build-depends: base >= 2 && <= 4.5,+ zeromq-haskell >= 0.8.4,+ bytestring, monad-loops, time, old-locale+
+ mdp_client.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE OverloadedStrings, DeriveDataTypeable #-}+module Main where++import Prelude hiding(getContents, putStr, putStrLn)+import Data.ByteString.Char8 as BS+import qualified System.Network.ZMQ.MDP.Client as C+import Data.List as L+import System.Console.CmdArgs.Implicit++data Client = Client {broker :: String,+ service::String, + message_parts::[String]} deriving (Show, Data, Typeable)+++client :: Client+client = Client{ broker = def &= argPos 0 ,+ service = def &= argPos 1 ,+ message_parts = def &= args+ } &= summary "connect to a 0mq server"+++main :: IO ()+main = do+ s <- cmdArgs client+ -- print s+ if (L.null $ message_parts s)+ then putStrLn "must send at least one argument"+ else C.withClientSocket (broker s) $ \sock -> do+ res <- C.sendAndReceive sock (pack $ service s) (Prelude.map pack $ message_parts s)+ putStr $ case res of+ Left C.ClientTimedOut -> "Timed out!"+ Left C.ClientBadProtocol -> "Bad protocol!"+ Right l -> BS.concat . L.intersperse "\n" $ C.response l+