3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-29 00:32:34 +08:00
Code Issues Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/zeromq/libzmq

Browse Source
This commit is contained in:
Ian Barber 2012-06-08 23:55:58 +01:00
commit 0bf1d0262e
12 changed files with 180 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/zmq.h
View File

@ -59,8 +59,8 @@ extern "C" {
/* Version macros for compile-time API version detection */
#define ZMQ_VERSION_MAJOR 3
#define ZMQ_VERSION_MINOR 1
#define ZMQ_VERSION_PATCH 1
#define ZMQ_VERSION_MINOR 3
#define ZMQ_VERSION_PATCH 0
#define ZMQ_MAKE_VERSION(major, minor, patch) \
((major) * 10000 + (minor) * 100 + (patch))

2
src/Makefile.am
View File

@ -65,6 +65,7 @@ libzmq_la_SOURCES = \
stdint.hpp \
stream_engine.hpp \
sub.hpp \
tcp.hpp \
tcp_address.hpp \
tcp_connecter.hpp \
tcp_listener.hpp \
@ -123,6 +124,7 @@ libzmq_la_SOURCES = \
socket_base.cpp \
stream_engine.cpp \
sub.cpp \
tcp.cpp \
tcp_address.cpp \
tcp_connecter.cpp \
tcp_listener.cpp \

83
src/ip.cpp
View File

@ -66,89 +66,6 @@ zmq::fd_t zmq::open_socket (int domain_, int type_, int protocol_)
return s;
}
void zmq::tune_tcp_socket (fd_t s_)
{
// Disable Nagle's algorithm. We are doing data batching on 0MQ level,
// so using Nagle wouldn't improve throughput in anyway, but it would
// hurt latency.
int nodelay = 1;
int rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELAY, (char*) &nodelay,
sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
#ifdef ZMQ_HAVE_OPENVMS
// Disable delayed acknowledgements as they hurt latency is serious manner.
int nodelack = 1;
rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELACK, (char*) &nodelack,
sizeof (int));
errno_assert (rc != SOCKET_ERROR);
#endif
}
void zmq::tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_)
{
// Tuning TCP keep-alives if platform allows it
// All values = -1 means skip and leave it for OS
#ifdef ZMQ_HAVE_SO_KEEPALIVE
if (keepalive_ != -1) {
int rc = setsockopt (s_, SOL_SOCKET, SO_KEEPALIVE, (char*) &keepalive_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
#ifdef ZMQ_HAVE_TCP_KEEPCNT
if (keepalive_cnt_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPCNT, &keepalive_cnt_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
}
#endif // ZMQ_HAVE_TCP_KEEPCNT
#ifdef ZMQ_HAVE_TCP_KEEPIDLE
if (keepalive_idle_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPIDLE, &keepalive_idle_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
}
#else // ZMQ_HAVE_TCP_KEEPIDLE
#ifdef ZMQ_HAVE_TCP_KEEPALIVE
if (keepalive_idle_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPALIVE, &keepalive_idle_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
}
#endif // ZMQ_HAVE_TCP_KEEPALIVE
#endif // ZMQ_HAVE_TCP_KEEPIDLE
#ifdef ZMQ_HAVE_TCP_KEEPINTVL
if (keepalive_intvl_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPINTVL, &keepalive_intvl_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
}
#endif // ZMQ_HAVE_TCP_KEEPINTVL
}
#endif // ZMQ_HAVE_SO_KEEPALIVE
}
void zmq::unblock_socket (fd_t s_)
{
#ifdef ZMQ_HAVE_WINDOWS

6
src/ip.hpp
View File

@ -30,12 +30,6 @@ namespace zmq
// Same as socket(2), but allows for transparent tweaking the options.
fd_t open_socket (int domain_, int type_, int protocol_);
// Tunes the supplied TCP socket for the best latency.
void tune_tcp_socket (fd_t s_);
// Tunes TCP keep-alives
void tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_);
// Sets the socket into non-blocking mode.
void unblock_socket (fd_t s_);

4
src/ipc_listener.cpp
View File

@ -103,7 +103,11 @@ void zmq::ipc_listener_t::in_event ()
int zmq::ipc_listener_t::get_address (std::string &addr_)
{
struct sockaddr_storage ss;
#ifdef ZMQ_HAVE_HPUX
int sl = sizeof (ss);
#else
socklen_t sl = sizeof (ss);
#endif
int rc = getsockname (s, (sockaddr *) &ss, &sl);
if (rc != 0) {
addr_.clear ();

1
src/pgm_socket.cpp
View File

@ -86,7 +86,6 @@ int zmq::pgm_socket_t::init_address (const char *network_,
pgm_error_t *pgm_error = NULL;
struct pgm_addrinfo_t hints, *res = NULL;
sa_family_t sa_family;
memset (&hints, 0, sizeof (hints));
hints.ai_family = AF_UNSPEC;

2
src/stream_engine.cpp
View File

@ -78,7 +78,7 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_) :
#endif
}
#if defined ZMQ_HAVE_OSX || defined ZMQ_HAVE_FREEBSD
#ifdef SO_NOSIGPIPE
// Make sure that SIGPIPE signal is not generated when writing to a
// connection that was already closed by the peer.
int set = 1;

122
src/tcp.cpp Normal file
View File

@ -0,0 +1,122 @@
/*
Copyright (c) 2010-2011 250bpm s.r.o.
Copyright (c) 2007-2009 iMatix Corporation
Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "ip.hpp"
#include "tcp.hpp"
#include "err.hpp"
#include "platform.hpp"
#if defined ZMQ_HAVE_WINDOWS
#include "windows.hpp"
#else
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#endif
#if defined ZMQ_HAVE_OPENVMS
#include <ioctl.h>
#endif
void zmq::tune_tcp_socket (fd_t s_)
{
// Disable Nagle's algorithm. We are doing data batching on 0MQ level,
// so using Nagle wouldn't improve throughput in anyway, but it would
// hurt latency.
int nodelay = 1;
int rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELAY, (char*) &nodelay,
sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
#ifdef ZMQ_HAVE_OPENVMS
// Disable delayed acknowledgements as they hurt latency is serious manner.
int nodelack = 1;
rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELACK, (char*) &nodelack,
sizeof (int));
errno_assert (rc != SOCKET_ERROR);
#endif
}
void zmq::tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_)
{
// Tuning TCP keep-alives if platform allows it
// All values = -1 means skip and leave it for OS
#ifdef ZMQ_HAVE_SO_KEEPALIVE
if (keepalive_ != -1) {
int rc = setsockopt (s_, SOL_SOCKET, SO_KEEPALIVE, (char*) &keepalive_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
#ifdef ZMQ_HAVE_TCP_KEEPCNT
if (keepalive_cnt_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPCNT, &keepalive_cnt_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
}
#endif // ZMQ_HAVE_TCP_KEEPCNT
#ifdef ZMQ_HAVE_TCP_KEEPIDLE
if (keepalive_idle_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPIDLE, &keepalive_idle_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
}
#else // ZMQ_HAVE_TCP_KEEPIDLE
#ifdef ZMQ_HAVE_TCP_KEEPALIVE
if (keepalive_idle_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPALIVE, &keepalive_idle_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
}
#endif // ZMQ_HAVE_TCP_KEEPALIVE
#endif // ZMQ_HAVE_TCP_KEEPIDLE
#ifdef ZMQ_HAVE_TCP_KEEPINTVL
if (keepalive_intvl_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPINTVL, &keepalive_intvl_, sizeof (int));
#ifdef ZMQ_HAVE_WINDOWS
wsa_assert (rc != SOCKET_ERROR);
#else
errno_assert (rc == 0);
#endif
}
#endif // ZMQ_HAVE_TCP_KEEPINTVL
}
#endif // ZMQ_HAVE_SO_KEEPALIVE
}

38
src/tcp.hpp Normal file
View File

@ -0,0 +1,38 @@
/*
Copyright (c) 2010-2011 250bpm s.r.o.
Copyright (c) 2007-2009 iMatix Corporation
Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_TCP_HPP_INCLUDED__
#define __ZMQ_TCP_HPP_INCLUDED__
#include "fd.hpp"
namespace zmq
{
// Tunes the supplied TCP socket for the best latency.
void tune_tcp_socket (fd_t s_);
// Tunes TCP keep-alives
void tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_);
}
#endif

1
src/tcp_connecter.cpp
View File

@ -29,6 +29,7 @@
#include "random.hpp"
#include "err.hpp"
#include "ip.hpp"
#include "tcp.hpp"
#include "address.hpp"
#include "tcp_address.hpp"
#include "session_base.hpp"

9
src/tcp_listener.cpp
View File

@ -31,6 +31,7 @@
#include "config.hpp"
#include "err.hpp"
#include "ip.hpp"
#include "tcp.hpp"
#include "socket_base.hpp"
#ifdef ZMQ_HAVE_WINDOWS
@ -133,7 +134,11 @@ int zmq::tcp_listener_t::get_address (std::string &addr_)
{
// Get the details of the TCP socket
struct sockaddr_storage ss;
#ifdef ZMQ_HAVE_HPUX
int sl = sizeof (ss);
#else
socklen_t sl = sizeof (ss);
#endif
int rc = getsockname (s, (struct sockaddr *) &ss, &sl);
if (rc != 0) {
@ -233,7 +238,11 @@ zmq::fd_t zmq::tcp_listener_t::accept ()
zmq_assert (s != retired_fd);
struct sockaddr_storage ss = {0};
#ifdef ZMQ_HAVE_HPUX
int ss_len = sizeof (ss);
#else
socklen_t ss_len = sizeof (ss);
#endif
fd_t sock = ::accept (s, (struct sockaddr *) &ss, &ss_len);
#ifdef ZMQ_HAVE_WINDOWS

2
tests/test_connect_resolve.cpp
View File

@ -40,7 +40,7 @@ int main (int argc, char *argv [])
int rc = zmq_connect (sock, "tcp://localhost:1234");
assert (rc == 0);
rc = zmq_connect (sock, "tcp://foobar123xyz:1234");
rc = zmq_connect (sock, "tcp://0mq.is.teh.best:1234");
assert (rc == -1);
assert (errno == EINVAL);