nanomsg-haskell 0.2.1.1 → 0.2.2
raw patch · 8 files changed
+435/−82 lines, 8 filesdep +binarydep +zeromq4-haskelldep −zeromq3-haskelldep ~base
Dependencies added: binary, zeromq4-haskell
Dependencies removed: zeromq3-haskell
Dependency ranges changed: base
Files
- CONTRIBUTORS +3/−0
- README.md +50/−48
- benchmarks/Zmq.hs +7/−7
- changelog +4/−0
- nanomsg-haskell.cabal +32/−8
- src/Nanomsg.hsc +28/−19
- src/Nanomsg/Binary.hs +94/−0
- tests/BinaryProperties.hs +217/−0
+ CONTRIBUTORS view
@@ -0,0 +1,3 @@+Ivar Nymoen <ivar.nymoen@gmail.com>+João Cristóvão <jmacristovao@gmail.com>+Jakub Stasiak <jakub@stasiak.at>
README.md view
@@ -16,7 +16,7 @@ install from Hackage using "cabal update && cabal install nanomsg-haskell", but can build from the repository following these steps: - 1. Build and install nanomsg (and zmq3, if you are building benchmarks)+ 1. Build and install nanomsg (and zeromq, if you are building benchmarks) 1. git clone https://github.com/ivarnymoen/nanomsg-haskell 1. cd nanomsg-haskell && cabal sandbox init 1. cabal install --dependencies-only [--enable-tests] [--enable-benchmarks]@@ -30,60 +30,62 @@ Simple pub/sub example: Server:-- module Main where+```haskell+module Main where - import Nanomsg- import qualified Data.ByteString.Char8 as C- import Control.Monad (mapM_)- import Control.Concurrent (threadDelay)+import Nanomsg+import qualified Data.ByteString.Char8 as C+import Control.Monad (mapM_)+import Control.Concurrent (threadDelay) - main :: IO ()- main = do- withSocket Pub $ \s -> do- _ <- bind s "tcp://*:5560"- mapM_ (\num -> sendNumber s num) (cycle [1..1000000])- where- sendNumber s number = do- threadDelay 1000 -- let's conserve some cycles- let numAsString = show number- send s (C.pack numAsString)+main :: IO ()+main = do+ withSocket Pub $ \s -> do+ _ <- bind s "tcp://*:5560"+ mapM_ (\num -> sendNumber s num) (cycle [1..1000000])+ where+ sendNumber s number = do+ threadDelay 1000 -- let's conserve some cycles+ let numAsString = show number+ send s (C.pack numAsString)+``` Client:-- module Main where+```haskell+module Main where - import Nanomsg- import qualified Data.ByteString.Char8 as C- import Control.Monad (forever)+import Nanomsg+import qualified Data.ByteString.Char8 as C+import Control.Monad (forever) - main :: IO ()- main = do- withSocket Sub $ \s -> do- _ <- connect s "tcp://localhost:5560"- subscribe s $ C.pack ""- forever $ do- msg <- recv s- C.putStrLn msg+main :: IO ()+main = do+ withSocket Sub $ \s -> do+ _ <- connect s "tcp://localhost:5560"+ subscribe s $ C.pack ""+ forever $ do+ msg <- recv s+ C.putStrLn msg+``` Nonblocking client:-- module Main where-- import Nanomsg- import qualified Data.ByteString.Char8 as C- import Control.Monad (forever)- import Control.Concurrent (threadDelay)+```haskell+module Main where - main :: IO ()- main =- withSocket Sub $ \s -> do- _ <- connect s "tcp://localhost:5560"- subscribe s $ C.pack ""- forever $ do- threadDelay 700 -- let's conserve some cycles- msg <- recv' s- C.putStrLn $ case msg of- Nothing -> C.pack "No message"- Just s -> s+import Nanomsg+import qualified Data.ByteString.Char8 as C+import Control.Monad (forever)+import Control.Concurrent (threadDelay) +main :: IO ()+main =+ withSocket Sub $ \s -> do+ _ <- connect s "tcp://localhost:5560"+ subscribe s $ C.pack ""+ forever $ do+ threadDelay 700 -- let's conserve some cycles+ msg <- recv' s+ C.putStrLn $ case msg of+ Nothing -> C.pack "No message"+ Just s -> s+```
benchmarks/Zmq.hs view
@@ -1,7 +1,7 @@ module Main where import qualified Nanomsg as N-import qualified System.ZMQ3.Monadic as Z+import qualified System.ZMQ4.Monadic as Z import Criterion.Main import qualified Data.ByteString.Char8 as C import Control.Monad (replicateM_)@@ -57,16 +57,16 @@ main :: IO () main = defaultMain [ 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 "zeromq4-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 "zeromq4-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 "zeromq4-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 "zeromq4-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 "zeromq4-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"+ , bench "zeromq4-haskell: 40 bytes x 10k messages, throughput, inproc" $ nfIO $ zThr 40 100 "inproc://bench" "inproc://bench" ]
+ changelog view
@@ -0,0 +1,4 @@+0.2.2+ * Added a thin binary based serialization layer+ * Benchmarks now depend on ZMQ4+
nanomsg-haskell.cabal view
@@ -1,17 +1,20 @@ name: nanomsg-haskell-version: 0.2.1.1+version: 0.2.2 synopsis: Bindings to the nanomsg library description: This is a Haskell binding for the nanomsg library: <http://nanomsg.org/>. - There's support for blocking send and recv, a non-blocking receive,+ There's support for (evented) 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 sockets options are available through accessor and mutator functions. Sockets are typed, transports are not. + The module Nanomsg.Binary provides a simple Binary based serialization+ layer over send and receive.+ homepage: https://github.com/ivarnymoen/nanomsg-haskell license: MIT license-file: LICENSE@@ -23,6 +26,8 @@ cabal-version: >=1.10 extra-source-files: README.md+ , CONTRIBUTORS+ , changelog , tests/*.hs , benchmarks/*.hs @@ -31,12 +36,14 @@ ghc-options: -O2 -Wall default-language: Haskell2010 exposed-modules: Nanomsg+ Nanomsg.Binary default-extensions: ForeignFunctionInterface, DeriveDataTypeable includes: nanomsg/nn.h extra-libraries: nanomsg build-depends:- base >= 4.5 && < 4.8,- bytestring >= 0.9.0 && < 0.11+ base >= 4.5 && < 5,+ bytestring >= 0.9.0 && < 0.11,+ binary >= 0.7 && < 0.8 test-suite tests type: exitcode-stdio-1.0@@ -45,7 +52,7 @@ ghc-options: -O2 -Wall -threaded default-language: Haskell2010 build-depends:- base >= 4.5 && < 4.8,+ base >= 4.5 && < 5, bytestring >= 0.9.0 && < 0.11, nanomsg-haskell, QuickCheck,@@ -53,6 +60,23 @@ test-framework-quickcheck2, test-framework-th +test-suite tests-binary+ type: exitcode-stdio-1.0+ hs-source-dirs: tests+ main-is: BinaryProperties.hs+ ghc-options: -O2 -Wall -threaded+ default-language: Haskell2010+ build-depends:+ base >= 4.5 && < 5,+ bytestring >= 0.9.0 && < 0.11,+ nanomsg-haskell,+ binary,+ QuickCheck,+ test-framework,+ test-framework-quickcheck2,+ test-framework-th++ source-repository head type: git location: https://github.com/ivarnymoen/nanomsg-haskell@@ -64,7 +88,7 @@ default-language: Haskell2010 hs-source-dirs: benchmarks build-depends:- base >= 4.5 && < 4.8,+ base >= 4.5 && < 5, bytestring >= 0.9.0 && < 0.11, nanomsg-haskell, criterion@@ -76,9 +100,9 @@ default-language: Haskell2010 hs-source-dirs: benchmarks build-depends:- base >= 4.5 && < 4.8,+ base >= 4.5 && < 5, bytestring >= 0.9.0 && < 0.11, nanomsg-haskell,- zeromq3-haskell,+ zeromq4-haskell, criterion
src/Nanomsg.hsc view
@@ -7,16 +7,16 @@ -- -- This is a Haskell binding for the nanomsg library: <http://nanomsg.org/>. ----- There's support for blocking send and recv, a non-blocking receive,+-- There's support for (evented) 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. ----- Socket type documentation is adapted or quoted verbatim from the--- nanomsg manual. Please refer to nanomsg.org for information on--- how to use the library.+-- The documentation is adapted or quoted verbatim from the nanomsg manual,+-- please refer to nanomsg.org for authoritative info.+-- There's a simple code example in <https://github.com/ivarnymoen/nanomsg-haskell#usage README.md>. module Nanomsg ( -- * Types@@ -111,52 +111,61 @@ -- will block until it’s possible to send the message. data Pair = Pair --- | Used to implement a client application that sends requests--- and receives replies. The socket will resend requests automatically+-- | Request socket. Pairs with 'Rep' sockets.+--+-- The socket will resend requests automatically -- if there's no reply within a given time. The default timeout -- is 1 minute. ----- See also 'setRequestResendInterval'.+-- See also 'Rep', 'setRequestResendInterval'. data Req = Req --- | Used to implement a stateless worker that receives requests--- and sends replies.+-- | Reply socket.+--+-- See also 'Req'. data Rep = Rep --- | This socket is used to distribute messages to multiple destinations.+-- | Publish socket. Pairs with subscribe sockets.+--+-- See also 'Sub'. data Pub = Pub --- | Receives messages from the publisher. Only messages that the socket is--- subscribed to are received. When the socket is created there are no--- subscriptions and thus no messages will be received.+-- | Subscribe socket. ----- See also 'subscribe' and 'unsubscribe'.+-- Only messages that the socket is subscribed to are received. When the socket+-- is created there are no subscriptions and thus no messages will be received.+--+-- See also 'Pub', 'subscribe' and 'unsubscribe'. data Sub = Sub -- | Surveyor and respondent are used to broadcast a survey to multiple -- locations and gather the responses. ----- This socket is used to send the survey. The survey is delivered to all--- the connected respondents. Once the query is sent, the socket can be used+-- This socket is used to send a survey. The survey is delivered to all+-- onnected respondents. Once the query is sent, the socket can be used -- to receive the responses. -- -- When the survey deadline expires, receive will throw an NNException. ----- See also 'setSurveyorDeadline'+-- See also 'Respondent', 'setSurveyorDeadline'. data Surveyor = Surveyor -- | Used to respond to a survey. Survey is received using receive, -- response is sent using send. This socket can be connected to -- at most one peer.+--+-- See also 'Surveyor'. data Respondent = Respondent -- | Push and Pull sockets fair queue messages from one processing step, load -- balancing them among instances of the next processing step. ----- This socket is used to send messages to a cluster of load-balanced nodes.+-- See also 'Pull'. data Push = Push --- | This socket is used to receive a message from a cluster of nodes.+-- | Pull socket.+--+-- See also 'Push'. data Pull = Pull -- | Broadcasts messages from any node to all other nodes in the topology.
+ src/Nanomsg/Binary.hs view
@@ -0,0 +1,94 @@+{-# LANGUAGE ForeignFunctionInterface, DeriveDataTypeable #-}+-- |+-- Module: Nanomsg.Binary+--+-- This module offers a thin serialization layer ("Binary" based)+-- over @'send'@ and @'receive'@. You just need to import +-- @Nanomsg.Binary@ instead of @Nanomsg@.++module Nanomsg.Binary+ (+ -- * Types+ -- ** Socket types+ Pair(..)+ , Req(..)+ , Rep(..)+ , Pub(..)+ , Sub(..)+ , Surveyor(..)+ , Respondent(..)+ , Push(..)+ , Pull(..)+ , Bus(..)+ -- ** Other+ , Socket+ , Endpoint+ , NNException+-- , eTERM+-- , eFSM+ , SocketType+ , Sender+ , Receiver+ -- * Operations+ -- ** General operations+ , socket+ , withSocket+ , bind+ , connect+ , send+ , recv+ , recv'+ , subscribe+ , unsubscribe+ , shutdown+ , close+ , term+ -- ** Socket option settings+ , linger+ , setLinger+ , sndBuf+ , setSndBuf+ , rcvBuf+ , setRcvBuf+ , reconnectInterval+ , setReconnectInterval+ , reconnectIntervalMax+ , setReconnectIntervalMax+ , sndPrio+ , setSndPrio+ , ipv4Only+ , setIpv4Only+ , requestResendInterval+ , setRequestResendInterval+ , surveyorDeadline+ , setSurveyorDeadline+ , tcpNoDelay+ , setTcpNoDelay++ ) where++import Control.Applicative+import Nanomsg hiding (send,recv,recv')+import qualified Nanomsg as NM+import Data.Binary+import Data.ByteString.Lazy++send+ :: (Sender s, Binary dat)+ => Socket s+ -> dat+ -> IO ()+send s d = NM.send s (toStrict . encode $ d)++recv+ :: (Receiver s, Binary dat)+ => Socket s+ -> IO dat+recv s = decode . fromStrict <$> NM.recv s++recv'+ :: (Receiver s, Binary dat)+ => Socket s+ -> IO (Maybe dat)+recv' s = fmap (decode . fromStrict) <$> NM.recv' s+
+ tests/BinaryProperties.hs view
@@ -0,0 +1,217 @@+{-# LANGUAGE TemplateHaskell #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Main where++import Nanomsg.Binary+import Test.Framework.TH (defaultMainGenerator)+import Test.Framework.Providers.QuickCheck2 (testProperty)+import Test.QuickCheck+import Test.QuickCheck.Monadic+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as C+import Control.Concurrent (threadDelay)+import Control.Applicative ( (<$>) )+import Data.Maybe (catMaybes)++instance Arbitrary ByteString where+ arbitrary = C.pack <$> arbitrary++-- dummy test+prop_reverse :: [Int] -> Bool+prop_reverse xs =+ xs == reverse (reverse xs)++type MsgType = PropertyM IO [String]++-- test Pub and Sub sockets+prop_PubSub :: Property+prop_PubSub = monadicIO $ do+ msgs <- pick arbitrary :: MsgType+ pre $ not (null msgs)+ res <- run $ do+ pub <- socket Pub+ ep1 <- bind pub "inproc://pubsub"+ sub1 <- socket Sub+ ep2 <- connect sub1 "inproc://pubsub"+ subscribe sub1 $ C.pack ""+ sub2 <- socket Sub+ ep3 <- connect sub2 "inproc://pubsub"+ subscribe sub2 $ C.pack ""+ threadDelay 1000+ r <- mapM (sendMsg pub sub1 sub2) msgs+ unsubscribe sub2 $ C.pack ""+ unsubscribe sub1 $ C.pack ""+ shutdown sub2 ep3+ shutdown sub1 ep2+ shutdown pub ep1+ close pub+ close sub1+ close sub2+ threadDelay 1000+ return r+ assert $ and res+ where+ sendMsg pub sub1 sub2 msg = do+ send pub msg+ send pub msg+ a <- recv sub1+ b <- recv sub1+ c <- recv sub2+ d <- recv sub2+ return $ a == msg && b == msg && c == msg && d == msg++-- test Pair sockets+prop_Pair :: Property+prop_Pair = monadicIO $ do+ msgs <- pick arbitrary :: MsgType+ let recvS :: (Receiver a) => Socket a -> IO String+ recvS = recv+ pre $ not (null msgs)+ res <- run $ do+ s1 <- socket Pair+ _ <- bind s1 "inproc://pair"+ s2 <- socket Pair+ _ <- connect s2 "inproc://pair"+ threadDelay 1000+ -- Send message from s1 to s2, then back from s2 to s1, then make sure it hasn't changed+ r <- mapM (\m -> send s1 m >> recvS s2 >>= send s2 >> recv s1 >>= return . (== m)) msgs+ close s1+ close s2+ threadDelay 1000+ return r+ assert $ and res++-- test Pipeline (Push & Pull) sockets+prop_Pipeline :: Property+prop_Pipeline = monadicIO $ do+ msgs <- pick arbitrary :: MsgType+ pre $ not (null msgs)+ res <- run $ do+ push <- socket Push+ _ <- bind push "inproc://pipeline"+ pull1 <- socket Pull+ pull2 <- socket Pull+ _ <- connect pull1 "inproc://pipeline"+ _ <- connect pull2 "inproc://pipeline"+ threadDelay 1000+ r <- mapM (testSockets push pull1 pull2) msgs+ close push+ close pull1+ close pull2+ threadDelay 1000+ return r+ assert $ and res+ where+ testSockets push pull1 pull2 msg = do+ send push msg+ send push msg+ send push msg+ threadDelay 1000+ a <- recv' pull1+ b <- recv' pull1+ c <- recv' pull1+ d <- recv' pull2+ e <- recv' pull2+ f <- recv' pull2+ let xs = catMaybes [a, b, c, d, e, f]+ return $ all (== msg) xs && (length xs == 3)++-- test Req and Rep sockets+prop_ReqRep :: Property+prop_ReqRep = monadicIO $ do+ msgs <- pick arbitrary :: MsgType+ let recvS :: (Receiver a) => Socket a -> IO String+ recvS = recv+ pre $ not (null msgs)+ res <- run $ do+ req <- socket Req+ _ <- bind req "inproc://reqrep"+ rep <- socket Rep+ _ <- connect rep "inproc://reqrep"+ threadDelay 1000+ r <- mapM (\m -> send req m >> recvS rep >>= send rep >> recv req >>= return . (== m)) msgs+ close req+ close rep+ threadDelay 1000+ return r+ assert $ and res++-- test Bus socket+prop_Bus :: Property+prop_Bus = monadicIO $ do+ msgs <- pick arbitrary :: MsgType+ pre $ not (null msgs)+ res <- run $ do+ -- Probably not how you're supposed to connect Bus nodes..+ b1 <- socket Bus+ _ <- bind b1 "inproc://bus1"+ b2 <- socket Bus+ _ <- connect b2 "inproc://bus1"+ _ <- bind b2 "inproc://bus2"+ b3 <- socket Bus+ _ <- connect b3 "inproc://bus2"+ _ <- bind b3 "inproc://bus3"+ _ <- connect b1 "inproc://bus3"+ threadDelay 1000+ r <- mapM (testSockets b1 b2 b3) msgs+ close b1+ close b2+ close b3+ threadDelay 1000+ return r+ assert $ and res+ where+ testSockets b1 b2 b3 msg = do+ send b1 msg+ a <- recv b2+ b <- recv b3+ send b2 msg+ c <- recv b1+ d <- recv b3+ send b3 msg+ e <- recv b1+ f <- recv b2+ return $ all (== msg) [a, b, c, d, e, f]++prop_TestOptions :: Property+prop_TestOptions = monadicIO $ do+ 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+ setTcpNoDelay req 0+ v2 <- tcpNoDelay req+ setRequestResendInterval req 30000+ v3 <- requestResendInterval req+ setIpv4Only req 0+ v4 <- ipv4Only req+ setIpv4Only req 1+ v5 <- ipv4Only req+ setSndPrio req 7+ v6 <- sndPrio req+ setReconnectInterval req 50+ v7 <- reconnectInterval req+ setReconnectIntervalMax req 400+ v8 <- reconnectIntervalMax req+ setRcvBuf req 200000+ v9 <- rcvBuf req+ setSndBuf req 150000+ 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, v12 == 2000]+ assert $ and res++main :: IO ()+main = $defaultMainGenerator+