diff --git a/AUTHORS b/AUTHORS deleted file mode 100644 --- a/AUTHORS +++ /dev/null @@ -1,3 +0,0 @@ -Full list of copyright-holders: -Ivar Nymoen - diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -25,17 +25,9 @@ 1. [cabal test] -## Contributing - -Just submit a pull request for small stuff, but please get in touch -beforehand before sinking a lot of effort into major/API changes. - -Remember adding your name to the AUTHORS file. - - ## Usage -Simple Pub/sub example: +Simple pub/sub example: Server: diff --git a/benchmarks/Zmq.hs b/benchmarks/Zmq.hs --- a/benchmarks/Zmq.hs +++ b/benchmarks/Zmq.hs @@ -6,8 +6,8 @@ import qualified Data.ByteString.Char8 as C import Control.Monad (replicateM_) -nPair :: Int -> Int -> String -> String -> IO () -nPair size count bindString connString = do +nLat :: Int -> Int -> String -> String -> IO () +nLat size count bindString connString = do s1 <- N.socket N.Pair _ <- N.bind s1 bindString s2 <- N.socket N.Pair @@ -18,8 +18,8 @@ N.close s2 return () -zPair :: Int -> Int -> String -> String -> IO () -zPair size count bindString connString = Z.runZMQ $ do +zLat :: Int -> Int -> String -> String -> IO () +zLat size count bindString connString = Z.runZMQ $ do s1 <- Z.socket Z.Pair _ <- Z.bind s1 bindString s2 <- Z.socket Z.Pair @@ -30,15 +30,43 @@ Z.close s2 return () +nThr :: Int -> Int -> String -> String -> IO () +nThr size count bindString connString = do + s1 <- N.socket N.Pair + _ <- N.bind s1 bindString + s2 <- N.socket N.Pair + _ <- N.connect s2 connString + let msg = C.pack $ replicate size 'a' + replicateM_ count (replicateM_ 100 (N.send s1 msg) >> replicateM_ 100 (N.recv s2)) + N.close s1 + N.close s2 + return () + +zThr :: Int -> Int -> String -> String -> IO () +zThr size count bindString connString = Z.runZMQ $ do + s1 <- Z.socket Z.Pair + _ <- Z.bind s1 bindString + s2 <- Z.socket Z.Pair + _ <- Z.connect s2 connString + let msg = C.pack $ replicate size 'a' + replicateM_ count (replicateM_ 100 (Z.send s1 [] msg) >> replicateM_ 100 (Z.receive s2)) + Z.close s1 + Z.close s2 + return () + main :: IO () main = defaultMain - [ bench "nanomsg-haskell: 40 bytes x 1k messages, roundtrip, tcp" $ nfIO $ nPair 40 1000 "tcp://*:5566" "tcp://localhost:5566" - , bench "zeromq3-haskell: 40 bytes x 1k messages, roundtrip, tcp" $ nfIO $ zPair 40 1000 "tcp://*:5566" "tcp://localhost:5566" - , bench "nanomsg-haskell: 20k bytes x 20 messages, roundtrip, tcp" $ nfIO $ nPair 20000 20 "tcp://*:5566" "tcp://localhost:5566" - , bench "zeromq3-haskell: 20k bytes x 20 messages, roundtrip, tcp" $ nfIO $ zPair 20000 20 "tcp://*:5566" "tcp://localhost:5566" - , bench "nanomsg-haskell: 40 bytes x 1k messages, roundtrip, inproc" $ nfIO $ nPair 40 1000 "inproc://bench" "inproc://bench" - , bench "zeromq3-haskell: 40 bytes x 1k messages, roundtrip, inproc" $ nfIO $ zPair 40 1000 "inproc://bench" "inproc://bench" - , bench "nanomsg-haskell: 20k bytes x 20 messages, roundtrip, inproc" $ nfIO $ nPair 20000 20 "inproc://bench" "inproc://bench" - , bench "zeromq3-haskell: 20k bytes x 20 messages, roundtrip, inproc" $ nfIO $ zPair 20000 20 "inproc://bench" "inproc://bench" + [ bench "nanomsg-haskell: 40 bytes x 2k messages, lat, tcp" $ nfIO $ nLat 40 1000 "tcp://*:5566" "tcp://localhost:5566" + , bench "zeromq3-haskell: 40 bytes x 2k messages, lat, tcp" $ nfIO $ zLat 40 1000 "tcp://*:5566" "tcp://localhost:5566" + , bench "nanomsg-haskell: 20k bytes x 40 messages, lat, tcp" $ nfIO $ nLat 20000 20 "tcp://*:5566" "tcp://localhost:5566" + , bench "zeromq3-haskell: 20k bytes x 40 messages, lat, tcp" $ nfIO $ zLat 20000 20 "tcp://*:5566" "tcp://localhost:5566" + , bench "nanomsg-haskell: 40 bytes x 2k messages, lat, inproc" $ nfIO $ nLat 40 1000 "inproc://bench" "inproc://bench" + , bench "zeromq3-haskell: 40 bytes x 2k messages, lat, inproc" $ nfIO $ zLat 40 1000 "inproc://bench" "inproc://bench" + , bench "nanomsg-haskell: 20k bytes x 40 messages, lat, inproc" $ nfIO $ nLat 20000 20 "inproc://bench" "inproc://bench" + , bench "zeromq3-haskell: 20k bytes x 40 messages, lat, inproc" $ nfIO $ zLat 20000 20 "inproc://bench" "inproc://bench" + , bench "nanomsg-haskell: 40 bytes x 10k messages, throughput, tcp" $ nfIO $ nThr 40 100 "tcp://*:5566" "tcp://localhost:5566" + , bench "zeromq3-haskell: 40 bytes x 10k messages, throughput, tcp" $ nfIO $ zThr 40 100 "tcp://*:5566" "tcp://localhost:5566" + , bench "nanomsg-haskell: 40 bytes x 10k messages, throughput, inproc" $ nfIO $ nThr 40 100 "inproc://bench" "inproc://bench" + , bench "zeromq3-haskell: 40 bytes x 10k messages, throughput, inproc" $ nfIO $ zThr 40 100 "inproc://bench" "inproc://bench" ] diff --git a/nanomsg-haskell.cabal b/nanomsg-haskell.cabal --- a/nanomsg-haskell.cabal +++ b/nanomsg-haskell.cabal @@ -1,5 +1,5 @@ name: nanomsg-haskell -version: 0.1.2 +version: 0.2.0 synopsis: Bindings to the nanomsg library description: @@ -15,13 +15,16 @@ homepage: https://github.com/ivarnymoen/nanomsg-haskell license: MIT license-file: LICENSE -author: See AUTHORS file +author: Ivar Nymoen maintainer: -copyright: Copyright (c) 2013 the nanomsg-haskell authors +copyright: Copyright (c) 2013 Ivar Nymoen category: Network build-type: Simple cabal-version: >=1.10 -extra-source-files: AUTHORS, README.md +extra-source-files: + README.md + , tests/*.hs + , benchmarks/*.hs library hs-source-dirs: src diff --git a/src/Nanomsg.hsc b/src/Nanomsg.hsc --- a/src/Nanomsg.hsc +++ b/src/Nanomsg.hsc @@ -10,6 +10,7 @@ -- There's support for blocking send and recv, a non-blocking receive, -- and for all the socket types and the functions you need to wire -- them up and tear them down again. +-- -- Most socket options are available through accessor and mutator -- functions. Sockets are typed, transports are not. -- @@ -34,8 +35,11 @@ , Socket , Endpoint , NNException - -- * API - -- ** Operations + , SocketType + , Sender + , Receiver + -- * Operations + -- ** General operations , socket , withSocket , bind @@ -65,6 +69,8 @@ , setIpv4Only , requestResendInterval , setRequestResendInterval + , surveyorDeadline + , setSurveyorDeadline , tcpNoDelay , setTcpNoDelay ) where @@ -134,7 +140,7 @@ -- the connected respondents. Once the query is sent, the socket can be used -- to receive the responses. -- --- When the survey deadline expires, receive will return ETIMEDOUT error. +-- When the survey deadline expires, receive will throw an NNException. -- -- See also 'setSurveyorDeadline' data Surveyor = Surveyor @@ -169,75 +175,62 @@ data Socket a = Socket a CInt deriving (Eq, Show) --- | Typeclass used by all sockets, to extract their C type. -class Protocol a where - -- | Returns the C enum value for each type. E.g. Pair => #const NN_PAIR - protocolId :: a -> CInt +-- | Typeclass for all sockets +class SocketType a where + socketType :: a -> CInt -- ^ Returns the C enum value for each type. E.g. Pair => #const NN_PAIR -instance Protocol Pair where - protocolId Pair = #const NN_PAIR +instance SocketType Pair where + socketType Pair = #const NN_PAIR -instance Protocol Req where - protocolId Req = #const NN_REQ +instance SocketType Req where + socketType Req = #const NN_REQ -instance Protocol Rep where - protocolId Rep = #const NN_REP +instance SocketType Rep where + socketType Rep = #const NN_REP -instance Protocol Pub where - protocolId Pub = #const NN_PUB +instance SocketType Pub where + socketType Pub = #const NN_PUB -instance Protocol Sub where - protocolId Sub = #const NN_SUB +instance SocketType Sub where + socketType Sub = #const NN_SUB -instance Protocol Surveyor where - protocolId Surveyor = #const NN_SURVEYOR +instance SocketType Surveyor where + socketType Surveyor = #const NN_SURVEYOR -instance Protocol Respondent where - protocolId Respondent = #const NN_RESPONDENT +instance SocketType Respondent where + socketType Respondent = #const NN_RESPONDENT -instance Protocol Push where - protocolId Push = #const NN_PUSH +instance SocketType Push where + socketType Push = #const NN_PUSH -instance Protocol Pull where - protocolId Pull = #const NN_PULL +instance SocketType Pull where + socketType Pull = #const NN_PULL -instance Protocol Bus where - protocolId Bus = #const NN_BUS +instance SocketType Bus where + socketType Bus = #const NN_BUS -- | Typeclass restricting which sockets can use the send function. -class SendType a -instance SendType Pair -instance SendType Req -instance SendType Rep -instance SendType Pub -instance SendType Surveyor -instance SendType Respondent -instance SendType Push -instance SendType Bus +class (SocketType a) => Sender a +instance Sender Pair +instance Sender Req +instance Sender Rep +instance Sender Pub +instance Sender Surveyor +instance Sender Respondent +instance Sender Push +instance Sender Bus -- | Typeclass for sockets that implement recv -class RecvType a -instance RecvType Pair -instance RecvType Req -instance RecvType Rep -instance RecvType Sub -instance RecvType Surveyor -instance RecvType Respondent -instance RecvType Pull -instance RecvType Bus - --- | Sub socket functionality -class SubscriberType a -instance SubscriberType Sub - --- | Surveyor socket functionality -class SurvType a -instance SurvType Surveyor - --- | Req socket functionality -class ReqType a -instance ReqType Req +class (SocketType a) => Receiver a +instance Receiver Pair +instance Receiver Req +instance Receiver Rep +instance Receiver Sub +instance Receiver Surveyor +instance Receiver Respondent +instance Receiver Pull +instance Receiver Bus -- * Error handling @@ -291,8 +284,10 @@ throwErrnoIfRetry_ :: (a -> Bool) -> String -> IO a -> IO () throwErrnoIfRetry_ p loc f = void $ throwErrnoIfRetry p loc f +{- throwErrnoIfMinus1Retry :: (Eq a, Num a) => String -> IO a -> IO a throwErrnoIfMinus1Retry = throwErrnoIfRetry (== -1) +-} throwErrnoIfMinus1Retry_ :: (Eq a, Num a) => String -> IO a -> IO () throwErrnoIfMinus1Retry_ = throwErrnoIfRetry_ (== -1) @@ -344,7 +339,7 @@ -- NN_EXPORT int nn_recv (int s, void *buf, size_t len, int flags); foreign import ccall unsafe "nn.h nn_recv" - c_nn_recv_foreignbuf :: CInt -> Ptr CString -> CInt -> CInt -> IO CInt + c_nn_recv :: CInt -> Ptr CString -> CInt -> CInt -> IO CInt -- NN_EXPORT int nn_freemsg (void *msg); foreign import ccall unsafe "nn.h nn_freemsg" @@ -389,14 +384,14 @@ NN_EXPORT void *nn_allocmsg (size_t size, int type); -} --- * API +-- * Operations -- | Creates a socket. Connections are formed using 'bind' or 'connect'. -- -- See also: 'close'. -socket :: (Protocol a) => a -> IO (Socket a) +socket :: (SocketType a) => a -> IO (Socket a) socket t = do - sid <- throwErrnoIfMinus1 "socket" $ c_nn_socket (#const AF_SP) (protocolId t) + sid <- throwErrnoIfMinus1 "socket" $ c_nn_socket (#const AF_SP) (socketType t) return $ Socket t sid -- | Creates a socket and runs your action with it. @@ -409,7 +404,7 @@ -- > replicateM 10 (recv sub) -- -- Ensures the socket is closed when your action is done. -withSocket :: (Protocol a) => a -> (Socket a -> IO b) -> IO b +withSocket :: (SocketType a) => a -> (Socket a -> IO b) -> IO b withSocket t = bracket (socket t) close -- | Binds the socket to a local interface. @@ -457,7 +452,7 @@ -- | Blocking function for sending a message -- -- See also: 'recv', 'recv''. -send :: (SendType a, Protocol a) => Socket a -> ByteString -> IO () +send :: Sender a => Socket a -> ByteString -> IO () send (Socket t sid) string = U.unsafeUseAsCStringLen string $ \(ptr, len) -> throwErrnoIfMinus1RetryMayBlock_ @@ -466,12 +461,12 @@ (getOptionFd (Socket t sid) (#const NN_SNDFD) >>= threadWaitWrite) -- | Blocking receive. -recv :: (RecvType a, Protocol a) => Socket a -> IO ByteString +recv :: Receiver a => Socket a -> IO ByteString recv (Socket t sid) = alloca $ \ptr -> do len <- throwErrnoIfMinus1RetryMayBlock "recv" - (c_nn_recv_foreignbuf sid ptr (#const NN_MSG) (#const NN_DONTWAIT)) + (c_nn_recv sid ptr (#const NN_MSG) (#const NN_DONTWAIT)) (getOptionFd (Socket t sid) (#const NN_RCVFD) >>= threadWaitRead) buf <- peek ptr str <- C.packCStringLen (buf, fromIntegral len) @@ -479,10 +474,10 @@ return str -- | Nonblocking receive function. -recv' :: (RecvType a, Protocol a) => Socket a -> IO (Maybe ByteString) +recv' :: Receiver a => Socket a -> IO (Maybe ByteString) recv' (Socket _ sid) = alloca $ \ptr -> do - len <- c_nn_recv_foreignbuf sid ptr (#const NN_MSG) (#const NN_DONTWAIT) + len <- c_nn_recv sid ptr (#const NN_MSG) (#const NN_DONTWAIT) if len >= 0 then do buf <- peek ptr @@ -496,14 +491,14 @@ else throwErrno "recv'" -- | Subscribe to a given subject string. -subscribe :: (SubscriberType a, Protocol a) => Socket a -> ByteString -> IO () +subscribe :: Socket Sub -> ByteString -> IO () subscribe (Socket t sid) string = - setOption (Socket t sid) (protocolId t) (#const NN_SUB_SUBSCRIBE) (StringOption string) + setOption (Socket t sid) (socketType t) (#const NN_SUB_SUBSCRIBE) (StringOption string) -- | Unsubscribes from a subject. -unsubscribe :: (SubscriberType a, Protocol a) => Socket a -> ByteString -> IO () +unsubscribe :: Socket Sub -> ByteString -> IO () unsubscribe (Socket t sid) string = - setOption (Socket t sid) (protocolId t) (#const NN_SUB_UNSUBSCRIBE) (StringOption string) + setOption (Socket t sid) (socketType t) (#const NN_SUB_UNSUBSCRIBE) (StringOption string) -- | Closes the socket. Any buffered inbound messages that were not yet -- received by the application will be discarded. The library will try to @@ -564,7 +559,7 @@ size <- peek sizePtr if fdSize /= size then throwErrno "getOptionFd: output size not as expected" else return value --- | Specifies how long should the socket try to send pending outbound +-- | Specifies how long the socket should try to send pending outbound -- messages after close has been called, in milliseconds. -- -- Negative value means infinite linger. Default value is 1000 (1 second). @@ -714,7 +709,7 @@ -- resent. -- -- Default value is 60000 (1 minute). -requestResendInterval :: (ReqType a) => Socket a -> IO Int +requestResendInterval :: Socket Req -> IO Int requestResendInterval s = fromIntegral <$> getOption s (#const NN_REQ) (#const NN_REQ_RESEND_IVL) @@ -723,9 +718,19 @@ -- resent. -- -- Default value is 60000 (1 minute). -setRequestResendInterval :: (ReqType a) => Socket a -> Int -> IO () +setRequestResendInterval :: Socket Req -> Int -> IO () setRequestResendInterval s val = setOption s (#const NN_REQ) (#const NN_REQ_RESEND_IVL) (IntOption val) + +-- | Get timeout for Surveyor sockets +surveyorDeadline :: Socket Surveyor -> IO Int +surveyorDeadline s = + fromIntegral <$> getOption s (#const NN_SURVEYOR) (#const NN_SURVEYOR_DEADLINE) + +-- | Set timeout for Surveyor sockets +setSurveyorDeadline :: Socket Surveyor -> Int -> IO () +setSurveyorDeadline s val = + setOption s (#const NN_SURVEYOR) (#const NN_SURVEYOR_DEADLINE) (IntOption val) -- | This option, when set to 1, disables Nagle's algorithm. -- diff --git a/tests/Properties.hs b/tests/Properties.hs --- a/tests/Properties.hs +++ b/tests/Properties.hs @@ -172,6 +172,8 @@ res <- run $ do req <- socket Req _ <- bind req "tcp://*:5560" + surveyor <- socket Surveyor + _ <- bind surveyor "inproc://surveyor" threadDelay 1000 setTcpNoDelay req 1 v1 <- tcpNoDelay req @@ -195,10 +197,13 @@ v10 <- sndBuf req setLinger req 500 v11 <- linger req + setSurveyorDeadline surveyor 2000 + v12 <- surveyorDeadline surveyor close req + close surveyor threadDelay 1000 return [v1 == 1, v2 == 0, v3 == 30000, v4 == 0, v5 == 1, v6 == 7, - v7 == 50, v8 == 400, v9 == 200000, v10 == 150000, v11 == 500] + v7 == 50, v8 == 400, v9 == 200000, v10 == 150000, v11 == 500, v12 == 2000] assert $ and res main :: IO ()