nanomsg-haskell 0.1.2 → 0.2.0
raw patch · 6 files changed
+133/−103 lines, 6 files
Files
- AUTHORS +0/−3
- README.md +1/−9
- benchmarks/Zmq.hs +40/−12
- nanomsg-haskell.cabal +7/−4
- src/Nanomsg.hsc +79/−74
- tests/Properties.hs +6/−1
− AUTHORS
@@ -1,3 +0,0 @@-Full list of copyright-holders:-Ivar Nymoen-
README.md view
@@ -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:
benchmarks/Zmq.hs view
@@ -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" ]
nanomsg-haskell.cabal view
@@ -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: <ivar.nymoen@gmail.com>-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
src/Nanomsg.hsc view
@@ -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. --
tests/Properties.hs view
@@ -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 ()