diff --git a/AUTHORS b/AUTHORS new file mode 100644 --- /dev/null +++ b/AUTHORS @@ -0,0 +1,5 @@ +Toralf Wittner original implementation +David Himmelstrup added send' +Nicolas Trangez added support for zmq_device and "queue" test app +Ville Tirronen added support for ZMG_SNDMORE + diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2010 Toralf Wittner, David Himmelstrup +Copyright (c) 2010 zeromq-haskell authors Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation diff --git a/README b/README --- a/README +++ b/README @@ -3,11 +3,11 @@ · Current status -Version 0.3.x - This software currently has *alpha* status, i.e. it had -only seen very limited testing and changes to its API might happen. +Version 0.4 - This software currently has *beta* status, i.e. it had +seen limited testing. Changes to its API may still happen. -This software was developed and tested on Linux 2.6.32 with GHC-6.12.3 -using the current git HEAD of zeromq2 (http://github.com/sustrik/zeromq2). +This software was developed and tested on Linux 2.6.35 with GHC-6.12.3 +using zeromq-2.0.8. · Installation @@ -36,7 +36,7 @@ One difference to 0MQ's API is that sockets are parameterized types, i.e. there is not one socket type but when creating a socket the desired socket type has -to be specified, e.g. 'P2P' and the resulting socket is of type 'Socket P2P'. +to be specified, e.g. 'Pair' and the resulting socket is of type 'Socket Pair'. This additional type information is used to ensure that only options applicable to the socket type can be set, hence ZMQ_SUBSCRIBE and ZMQ_UNSUBSCRIBE which only apply to ZMQ_SUB sockets have their own functions ('subscribe' and diff --git a/src/System/ZMQ.hs b/src/System/ZMQ.hs --- a/src/System/ZMQ.hs +++ b/src/System/ZMQ.hs @@ -21,6 +21,7 @@ SocketOption(..), Poll(..), PollEvent(..), + Device(..), SType, SubsType, @@ -31,6 +32,8 @@ Rep(..), XReq(..), XRep(..), + Pull(..), + Push(..), Up(..), Down(..), @@ -48,8 +51,10 @@ send', receive, - poll + poll, + device + ) where import Prelude hiding (init) @@ -144,6 +149,29 @@ instance SType XRep where zmqSocketType = const xresponse +-- | A socket of type ZMQ_PULL is used by a pipeline node to receive +-- messages from upstream pipeline nodes. Messages are fair-queued from +-- among all connected upstream nodes. The zmq_send() function is not +-- implemented for this socket type. +data Pull = Pull +instance SType Pull where + zmqSocketType = const pull + +-- | A socket of type ZMQ_PUSH is used by a pipeline node to send messages +-- to downstream pipeline nodes. Messages are load-balanced to all connected +-- downstream nodes. The zmq_recv() function is not implemented for this +-- socket type. +-- +-- When a ZMQ_PUSH socket enters an exceptional state due to having reached +-- the high water mark for all downstream nodes, or if there are no +-- downstream nodes at all, then any zmq_send(3) operations on the socket +-- shall block until the exceptional state ends or at least one downstream +-- node becomes available for sending; messages are not discarded. +data Push = Push +instance SType Push where + zmqSocketType = const push + +{-# DEPRECATED Up "Use Pull instead." #-} -- | Socket to receive messages from up the stream. Messages are -- fair-queued from among all the connected peers. Send function is not -- implemented for this socket type. /Compatible peer sockets/: 'Down'. @@ -151,6 +179,7 @@ instance SType Up where zmqSocketType = const upstream +{-# DEPRECATED Down "Use Push instead." #-} -- | Socket to send messages down stream. Messages are load-balanced -- among all the connected peers. Send function is not implemented for -- this socket type. /Compatible peer sockets/: 'Up'. @@ -247,6 +276,7 @@ -- If it cannot be performed immediatley an error will be thrown (errno -- is set to EAGAIN). data Flag = NoBlock -- ^ ZMQ_NOBLOCK + | SndMore -- ^ ZMQ_SNDMORE deriving (Eq, Ord, Show) -- | The events to wait for in poll (cf. man zmq_poll) @@ -263,6 +293,13 @@ forall a. S (Socket a) PollEvent | F Fd PollEvent +-- | Type representing ZeroMQ devices, as used with zmq_device +data Device = + Streamer -- ^ ZMQ_STREAMER + | Forwarder -- ^ ZMQ_FORWARDER + | Queue -- ^ ZMQ_QUEUE + deriving (Eq, Ord, Show) + -- | Initialize a 0MQ context (cf. zmq_init for details). init :: Size -> IO Context init ioThreads = do @@ -381,7 +418,20 @@ | e == (fromIntegral . pollVal $ pollInOut) = Just InOut | otherwise = Nothing +-- | Launch a ZeroMQ device (zmq_device). +-- +-- Please note that this call never returns. +device :: Device -> Socket a -> Socket b -> IO () +device device' (Socket insocket) (Socket outsocket) = + throwErrnoIfMinus1_ "device" $ + c_zmq_device (fromDevice device') insocket outsocket + where + fromDevice :: Device -> CInt + fromDevice Streamer = fromIntegral . deviceType $ deviceStreamer + fromDevice Forwarder = fromIntegral . deviceType $ deviceForwarder + fromDevice Queue = fromIntegral . deviceType $ deviceQueue + -- internal helpers: messageOf :: SB.ByteString -> IO Message @@ -435,6 +485,7 @@ toZMQFlag :: Flag -> ZMQFlag toZMQFlag NoBlock = noBlock +toZMQFlag SndMore = sndMore combine :: [Flag] -> CInt combine = fromIntegral . foldr ((.|.) . flagVal . toZMQFlag) 0 diff --git a/src/System/ZMQ/Base.hsc b/src/System/ZMQ/Base.hsc --- a/src/System/ZMQ/Base.hsc +++ b/src/System/ZMQ/Base.hsc @@ -64,6 +64,8 @@ response = ZMQ_REP, xrequest = ZMQ_XREQ, xresponse = ZMQ_XREP, + pull = ZMQ_PULL, + push = ZMQ_PUSH, upstream = ZMQ_UPSTREAM, downstream = ZMQ_DOWNSTREAM } @@ -88,6 +90,7 @@ #{enum ZMQFlag, ZMQFlag, noBlock = ZMQ_NOBLOCK + , sndMore = ZMQ_SNDMORE } newtype ZMQPollEvent = ZMQPollEvent { pollVal :: CShort } deriving (Eq, Ord) @@ -98,6 +101,14 @@ pollInOut = ZMQ_POLLIN | ZMQ_POLLOUT } +newtype ZMQDevice = ZMQDevice { deviceType :: CInt } deriving (Eq, Ord) + +#{enum ZMQDevice, ZMQDevice, + deviceStreamer = ZMQ_STREAMER, + deviceForwarder = ZMQ_FORWARDER, + deviceQueue = ZMQ_QUEUE +} + -- general initialization foreign import ccall unsafe "zmq.h zmq_init" @@ -154,4 +165,9 @@ foreign import ccall safe "zmq.h zmq_poll" c_zmq_poll :: ZMQPollPtr -> CInt -> CLong -> IO CInt + +-- device + +foreign import ccall safe "zmq.h zmq_device" + c_zmq_device :: CInt -> ZMQSocket -> ZMQSocket -> IO CInt diff --git a/test/queue.hs b/test/queue.hs new file mode 100644 --- /dev/null +++ b/test/queue.hs @@ -0,0 +1,103 @@ +-- Demo application for a ZeroMQ 'queue' device +-- +-- Compile using: +-- +-- ghc --make -threaded queue.hs + +import Control.Concurrent (forkOS, threadDelay) +import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar) + +import Control.Monad (forever, forM_, replicateM, replicateM_) + +import qualified Data.ByteString.Char8 as SB + +import qualified System.ZMQ as ZMQ + +main :: IO () +main = do + context <- ZMQ.init 1 + + lock <- newEmptyMVar + + _ <- forkOS $ launchQueue context lock + _ <- takeMVar lock + + forM_ [0..numWorkers] $ \i -> + forkOS $ launchWorker context i + + locks <- replicateM numClients newEmptyMVar + forM_ (zip [0..numClients] locks) $ \(i, lock') -> + forkOS $ launchClient context i lock' + + -- Wait untill all clients signal completion + forM_ locks takeMVar + + -- We can't clean up the context since our queue device is still running, + -- and can't be killed... + -- ZMQ.term context + + where + numWorkers :: Int + numWorkers = 5 + + numClients :: Int + numClients = 2 * numWorkers + + workersAddress :: String + workersAddress = "inproc://workers" + + clientsAddress :: String + clientsAddress = "tcp://127.0.0.1:5555" + + message :: SB.ByteString + message = SB.replicate 10 '\0' + + delay :: Int + delay = 1000000 + + launchQueue :: ZMQ.Context -> MVar () -> IO () + launchQueue context lock = do + workers <- ZMQ.socket context ZMQ.Xreq + ZMQ.bind workers workersAddress + + clients <- ZMQ.socket context ZMQ.Xrep + ZMQ.bind clients clientsAddress + + putMVar lock () + + ZMQ.device ZMQ.Queue clients workers + + -- This isn't reached + ZMQ.close workers + ZMQ.close clients + + launchWorker :: ZMQ.Context -> Int -> IO () + launchWorker context i = do + socket <- ZMQ.socket context ZMQ.Rep + ZMQ.connect socket workersAddress + + forever $ do + request <- ZMQ.receive socket [] + putStrLn $ + "Message received in worker " ++ show i ++ ": " ++ + SB.unpack request + threadDelay delay -- Do some 'work' + ZMQ.send socket message [] + putStrLn $ "Reply sent in worker " ++ show i + + -- This isn't reached + ZMQ.close socket + + launchClient :: ZMQ.Context -> Int -> MVar () -> IO () + launchClient context i lock = do + socket <- ZMQ.socket context ZMQ.Req + ZMQ.connect socket clientsAddress + + putStrLn $ "Sending message in client " ++ show i + ZMQ.send socket (SB.pack $ show i) [] + _ <- ZMQ.receive socket [] + putStrLn $ "Reply received in client " ++ show i + + ZMQ.close socket + + putMVar lock () diff --git a/zeromq-haskell.cabal b/zeromq-haskell.cabal --- a/zeromq-haskell.cabal +++ b/zeromq-haskell.cabal @@ -1,5 +1,5 @@ name: zeromq-haskell -version: 0.3.1 +version: 0.4 synopsis: bindings to zeromq description: Bindings to zeromq (http://zeromq.org) category: System, FFI @@ -7,13 +7,13 @@ license-file: LICENSE author: Toralf Wittner maintainer: toralf.wittner@gmail.com -copyright: Copyright (c) 2010 Toralf Wittner, David Himmelstrup +copyright: Copyright (c) 2010 zeromq-haskell authors homepage: http://github.com/twittner/zeromq-haskell/ stability: experimental -tested-With: GHC == 6.12.1 +tested-With: GHC == 6.12.3 cabal-version: >= 1.6.0 build-type: Simple -extra-source-files: README, test/*.hs, test/perf/*.hs +extra-source-files: README, AUTHORS, test/*.hs, test/perf/*.hs library exposed-modules: System.ZMQ