diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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) diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/echoworker.hs b/echoworker.hs new file mode 100644 --- /dev/null +++ b/echoworker.hs @@ -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 { } diff --git a/lib/System/Network/ZMQ/MDP/Client.hs b/lib/System/Network/ZMQ/MDP/Client.hs new file mode 100644 --- /dev/null +++ b/lib/System/Network/ZMQ/MDP/Client.hs @@ -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 diff --git a/lib/System/Network/ZMQ/MDP/Worker.hs b/lib/System/Network/ZMQ/MDP/Worker.hs new file mode 100644 --- /dev/null +++ b/lib/System/Network/ZMQ/MDP/Worker.hs @@ -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" + diff --git a/majordomo.cabal b/majordomo.cabal new file mode 100644 --- /dev/null +++ b/majordomo.cabal @@ -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 + diff --git a/mdp_client.hs b/mdp_client.hs new file mode 100644 --- /dev/null +++ b/mdp_client.hs @@ -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 +