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

acinclude.m4

@@ -581,6 +581,28 @@ AC_DEFUN([LIBZMQ_CHECK_LANG_VISIBILITY], [{
           [AC_MSG_RESULT(yes) ; $1], [AC_MSG_RESULT(no) ; $2])
 }])
 
+dnl ################################################################################
+dnl # LIBZMQ_CHECK_SOCK_CLOEXEC([action-if-found], [action-if-not-found])          #
+dnl # Check if SOCK_CLOEXEC is supported                                           #
+dnl ################################################################################
+AC_DEFUN([LIBZMQ_CHECK_SOCK_CLOEXEC], [{
+    AC_MSG_CHECKING(whether SOCK_CLOEXEC is supported)
+    AC_TRY_RUN([/* SOCK_CLOEXEC test */
+#include <sys/types.h>
+#include <sys/socket.h>
+
+int main (int argc, char *argv [])
+{
+    int s = socket (PF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
+    return (s == -1);
+}
+    ],
+    [AC_MSG_RESULT(yes) ; libzmq_cv_sock_cloexec="yes" ; $1],
+    [AC_MSG_RESULT(no)  ; libzmq_cv_sock_cloexec="no"  ; $2],
+    [AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_sock_cloexec="no"]
+    )
+}])
+
 dnl ################################################################################
 dnl # LIBZMQ_CHECK_POLLER_KQUEUE([action-if-found], [action-if-not-found])         #
 dnl # Checks kqueue polling system                                                 #

configure.in

@@ -274,7 +274,7 @@ AC_C_VOLATILE
 #  PGM extension
 libzmq_pgm_ext="no"
 
-pgm_basename="libpgm-5.1.115~dfsg"
+pgm_basename="libpgm-5.1.118~dfsg"
 
 AC_ARG_WITH([pgm], [AS_HELP_STRING([--with-pgm], 
     [build libzmq with PGM extension [default=no]])], 
@@ -383,6 +383,11 @@ AM_CONDITIONAL(ON_MINGW, test "x$libzmq_on_mingw32" = "xyes")
 AC_TYPE_SIGNAL
 AC_CHECK_FUNCS(perror gettimeofday memset socket getifaddrs freeifaddrs)
 AC_CHECK_HEADERS([alloca.h])
+LIBZMQ_CHECK_SOCK_CLOEXEC([AC_DEFINE(
+                              [ZMQ_HAVE_SOCK_CLOEXEC],
+                              [1],
+                              [Whether SOCK_CLOEXEC is defined and functioning.])
+                          ])
 
 # Subst LIBZMQ_EXTRA_CFLAGS & CXXFLAGS & LDFLAGS
 AC_SUBST(LIBZMQ_EXTRA_CFLAGS)

doc/zmq_getsockopt.txt

@@ -81,12 +81,10 @@ blocking or dropping sent messages. Refer to the individual socket descriptions
 in linkzmq:zmq_socket[3] for details on the exact action taken for each socket
 type.
 
-The default 'ZMQ_SNDHWM' value of zero means "no limit".
-
 [horizontal]
 Option value type:: int
 Option value unit:: messages
-Default value:: 0
+Default value:: 1000
 Applicable socket types:: all
 
 
@@ -103,12 +101,10 @@ blocking or dropping sent messages. Refer to the individual socket descriptions
 in linkzmq:zmq_socket[3] for details on the exact action taken for each socket
 type.
 
-The default 'ZMQ_RCVHWM' value of zero means "no limit".
-
 [horizontal]
 Option value type:: int
 Option value unit:: messages
-Default value:: 0
+Default value:: 1000
 Applicable socket types:: all
 
 

doc/zmq_setsockopt.txt

@@ -38,12 +38,10 @@ blocking or dropping sent messages. Refer to the individual socket descriptions
 in linkzmq:zmq_socket[3] for details on the exact action taken for each socket
 type.
 
-The default 'ZMQ_SNDHWM' value of zero means "no limit".
-
 [horizontal]
 Option value type:: int
 Option value unit:: messages
-Default value:: 0
+Default value:: 1000
 Applicable socket types:: all
 
 
@@ -60,12 +58,10 @@ blocking or dropping sent messages. Refer to the individual socket descriptions
 in linkzmq:zmq_socket[3] for details on the exact action taken for each socket
 type.
 
-The default 'ZMQ_RCVHWM' value of zero means "no limit".
-
 [horizontal]
 Option value type:: int
 Option value unit:: messages
-Default value:: 0
+Default value:: 1000
 Applicable socket types:: all
 
 

src/ip.cpp

@@ -40,7 +40,7 @@ zmq::fd_t zmq::open_socket (int domain_, int type_, int protocol_)
 {
     //  Setting this option result in sane behaviour when exec() functions
     //  are used. Old sockets are closed and don't block TCP ports etc.
-#if defined SOCK_CLOEXEC
+#if defined ZMQ_HAVE_SOCK_CLOEXEC
     type_ |= SOCK_CLOEXEC;
 #endif
 
@@ -51,7 +51,7 @@ zmq::fd_t zmq::open_socket (int domain_, int type_, int protocol_)
     //  If there's no SOCK_CLOEXEC, let's try the second best option. Note that
     //  race condition can cause socket not to be closed (if fork happens
     //  between socket creation and this point).
-#if !defined SOCK_CLOEXEC && defined FD_CLOEXEC
+#if !defined ZMQ_HAVE_SOCK_CLOEXEC && defined FD_CLOEXEC
     int rc = fcntl (s, F_SETFD, FD_CLOEXEC);
     errno_assert (rc != -1);
 #endif

src/kqueue.cpp

@@ -33,6 +33,7 @@
 #include "err.hpp"
 #include "config.hpp"
 #include "i_poll_events.hpp"
+#include "likely.hpp"
 
 //  NetBSD defines (struct kevent).udata as intptr_t, everyone else
 //  as void *.
@@ -106,30 +107,38 @@ void zmq::kqueue_t::rm_fd (handle_t handle_)
 void zmq::kqueue_t::set_pollin (handle_t handle_)
 {
     poll_entry_t *pe = (poll_entry_t*) handle_;
+    if (likely (!pe->flag_pollin)) {
         pe->flag_pollin = true;
         kevent_add (pe->fd, EVFILT_READ, pe);
     }
+}
 
 void zmq::kqueue_t::reset_pollin (handle_t handle_)
 {
     poll_entry_t *pe = (poll_entry_t*) handle_;
+    if (likely (pe->flag_pollin)) {
         pe->flag_pollin = false;
         kevent_delete (pe->fd, EVFILT_READ);
     }
+}
 
 void zmq::kqueue_t::set_pollout (handle_t handle_)
 {
     poll_entry_t *pe = (poll_entry_t*) handle_;
+    if (likely (!pe->flag_pollout)) {
         pe->flag_pollout = true;
         kevent_add (pe->fd, EVFILT_WRITE, pe);
     }
+}
 
 void zmq::kqueue_t::reset_pollout (handle_t handle_)
 {
     poll_entry_t *pe = (poll_entry_t*) handle_;
+    if (likely (pe->flag_pollout)) {
         pe->flag_pollout = false;
         kevent_delete (pe->fd, EVFILT_WRITE);
    }
+}
 
 void zmq::kqueue_t::start ()
 {

src/options.cpp

@@ -24,8 +24,8 @@
 #include "err.hpp"
 
 zmq::options_t::options_t () :
-    sndhwm (0),
+    sndhwm (1000),
-    rcvhwm (0),
+    rcvhwm (1000),
     affinity (0),
     rate (100),
     recovery_ivl (10000),

src/pgm_socket.cpp

@@ -222,6 +222,7 @@ int zmq::pgm_socket_t::init (bool udp_encapsulation_, const char *network_)
             goto err_abort;
     } else {
         const int send_only        = 1,
+                  max_rte      = (int) ((options.rate * 1000) / 8),
                   txw_max_tpdu     = (int) pgm_max_tpdu,
                   txw_sqns         = compute_sqns (txw_max_tpdu),
                   ambient_spm      = pgm_secs (30),
@@ -237,6 +238,8 @@ int zmq::pgm_socket_t::init (bool udp_encapsulation_, const char *network_)
 
         if (!pgm_setsockopt (sock, IPPROTO_PGM, PGM_SEND_ONLY,
                 &send_only, sizeof (send_only)) ||
+            !pgm_setsockopt (sock, IPPROTO_PGM, PGM_ODATA_MAX_RTE,
+                &max_rte, sizeof (max_rte)) ||
             !pgm_setsockopt (sock, IPPROTO_PGM, PGM_TXW_SQNS,
                 &txw_sqns, sizeof (txw_sqns)) ||
             !pgm_setsockopt (sock, IPPROTO_PGM, PGM_AMBIENT_SPM,