Problem: No direct support for setting socket priority (#4118)

* Problem: No direct support for setting socket priority

Solution: Add ZMQ_PRIORITY socket option, which sets the
SO_PRIORITY socket option on the underlying socket. This
socket option is not supported under Windows. Check option
and set socket option on creation of underlying socket.

CMakeLists.txt

@@ -723,6 +723,7 @@ if(NOT CMAKE_CROSSCOMPILING AND NOT MSVC)
   zmq_check_o_cloexec()
   zmq_check_so_bindtodevice()
   zmq_check_so_keepalive()
+  zmq_check_so_priority()
   zmq_check_tcp_keepcnt()
   zmq_check_tcp_keepidle()
   zmq_check_tcp_keepintvl()

RELICENSE/nokia.md

@@ -0,0 +1,13 @@
+# Permission to Relicense under MPLv2
+
+This is a statement by Nokia
+that grants permission to relicense its copyrights in the libzmq C++
+library (ZeroMQ) under the Mozilla Public License v2 (MPLv2).
+
+A portion of the commits made by the Github handle "aheroff1", with
+commit author "Andy Heroff", are copyright of Nokia.
+This document hereby grants the libzmq project team to relicense libzmq, 
+including all past, present and future contributions of the author listed above.
+
+Andy Heroff
+2021/01/05

acinclude.m4

@@ -870,6 +870,33 @@ int main (int argc, char *argv [])
     AS_IF([test "x$libzmq_cv_tcp_keepalive" = "xyes"], [$1], [$2])
 }])
 
+dnl ################################################################################
+dnl # LIBZMQ_CHECK_SO_PRIORITY([action-if-found], [action-if-not-found])           #
+dnl # Check if SO_PRIORITY is supported                                            #
+dnl ################################################################################
+AC_DEFUN([LIBZMQ_CHECK_SO_PRIORITY], [{
+    AC_CACHE_CHECK([whether SO_PRIORITY is supported], [libzmq_cv_so_priority],
+        [AC_TRY_RUN([/* SO_PRIORITY test */
+#include <sys/types.h>
+#include <sys/socket.h>
+
+int main (int argc, char *argv [])
+{
+    int s, rc, opt = 1;
+    return (
+        ((s = socket (PF_INET, SOCK_STREAM, 0)) == -1) ||
+        ((rc = setsockopt (s, SOL_SOCKET, SO_PRIORITY, (char*) &opt, sizeof (int))) == -1)
+    );
+}
+        ],
+        [libzmq_cv_so_priority="yes"],
+        [libzmq_cv_so_priority="no"],
+        [libzmq_cv_so_priority="not during cross-compile"]
+        )]
+    )
+    AS_IF([test "x$libzmq_cv_so_priority" = "xyes"], [$1], [$2])
+}])
+
 dnl ################################################################################
 dnl # LIBZMQ_CHECK_GETRANDOM([action-if-found], [action-if-not-found])  #
 dnl # Checks if getrandom is supported                                  #

builds/cmake/Modules/ZMQSourceRunChecks.cmake

@@ -311,3 +311,22 @@ int main(int argc, char *argv [])
 "
     ZMQ_HAVE_NOEXCEPT)
 endmacro()
+
+macro(zmq_check_so_priority)
+  message(STATUS "Checking whether SO_PRIORITY is supported")
+  check_c_source_runs(
+    "
+#include <sys/types.h>
+#include <sys/socket.h>
+
+int main (int argc, char *argv [])
+{
+    int s, rc, opt = 1;
+    return (
+        ((s = socket (PF_INET, SOCK_STREAM, 0)) == -1) ||
+        ((rc = setsockopt (s, SOL_SOCKET, SO_PRIORITY, (char*) &opt, sizeof (int))) == -1)
+    );
+}
+"
+    ZMQ_HAVE_SO_PRIORITY)
+endmacro()

builds/cmake/platform.hpp.in

@@ -40,6 +40,7 @@
 
 #cmakedefine ZMQ_HAVE_SOCK_CLOEXEC
 #cmakedefine ZMQ_HAVE_SO_KEEPALIVE
+#cmakedefine ZMQ_HAVE_SO_PRIORITY
 #cmakedefine ZMQ_HAVE_TCP_KEEPCNT
 #cmakedefine ZMQ_HAVE_TCP_KEEPIDLE
 #cmakedefine ZMQ_HAVE_TCP_KEEPINTVL

configure.ac

@@ -959,6 +959,12 @@ LIBZMQ_CHECK_TCP_KEEPALIVE([
         [Whether TCP_KEEPALIVE is supported.])
     ])
 
+LIBZMQ_CHECK_SO_PRIORITY([
+    AC_DEFINE([ZMQ_HAVE_SO_PRIORITY],
+        [1],
+        [Whether SO_PRIORITY is supported.])
+    ])
+
 LIBZMQ_CHECK_GETRANDOM([
     AC_DEFINE([ZMQ_HAVE_GETRANDOM],
         [1],

doc/zmq_getsockopt.txt

@@ -513,6 +513,18 @@ Default value:: -1
 Applicable socket types:: all bound sockets, when using IPC or TCP transport
 
 
+ZMQ_PRIORITY: Retrieve the Priority on socket
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Gets the protocol-defined priority for all packets to be sent on this
+socket, where supported by the OS.
+
+[horizontal]
+Option value type:: int
+Option value unit:: >0
+Default value:: 0
+Applicable socket types:: all, only for connection-oriented transports
+
+
 ZMQ_RATE: Retrieve multicast data rate
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 The 'ZMQ_RATE' option shall retrieve the maximum send or receive data rate for

doc/zmq_setsockopt.txt

@@ -606,6 +606,19 @@ Default value:: -1
 Applicable socket types:: all bound sockets, when using IPC or TCP transport
 
 
+ZMQ_PRIORITY: Set the Priority on socket
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Sets the protocol-defined priority for all packets to be sent on this
+socket, where supported by the OS. In Linux, values greater than 6
+require admin capability (CAP_NET_ADMIN)
+
+[horizontal]
+Option value type:: int
+Option value unit:: >0
+Default value:: 0
+Applicable socket types:: all, only for connection-oriented transports
+
+
 ZMQ_PROBE_ROUTER: bootstrap connections to ROUTER sockets
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 When set to 1, the socket will automatically send an empty message when a

include/zmq.h

@@ -682,6 +682,7 @@ ZMQ_EXPORT void zmq_threadclose (void *thread_);
 #define ZMQ_RECONNECT_STOP 109
 #define ZMQ_HELLO_MSG 110
 #define ZMQ_DISCONNECT_MSG 111
+#define ZMQ_PRIORITY 112
 
 /*  DRAFT ZMQ_RECONNECT_STOP options                                          */
 #define ZMQ_RECONNECT_STOP_CONN_REFUSED 0x1

src/ip.cpp

@@ -230,6 +230,16 @@ void zmq::set_ip_type_of_service (fd_t s_, int iptos_)
 #endif
 }
 
+void zmq::set_socket_priority (fd_t s_, int priority_)
+{
+#ifdef ZMQ_HAVE_SO_PRIORITY
+    int rc =
+      setsockopt (s_, SOL_SOCKET, SO_PRIORITY,
+                  reinterpret_cast<char *> (&priority_), sizeof (priority_));
+    errno_assert (rc == 0);
+#endif
+}
+
 int zmq::set_nosigpipe (fd_t s_)
 {
 #ifdef SO_NOSIGPIPE

src/ip.hpp

@@ -51,6 +51,9 @@ int get_peer_ip_address (fd_t sockfd_, std::string &ip_addr_);
 // Sets the IP Type-Of-Service for the underlying socket
 void set_ip_type_of_service (fd_t s_, int iptos_);
 
+// Sets the protocol-defined priority for the underlying socket
+void set_socket_priority (fd_t s_, int priority_);
+
 // Sets the SO_NOSIGPIPE option for the underlying socket.
 // Return 0 on success, -1 if the connection has been closed by the peer
 int set_nosigpipe (fd_t s_);

src/options.cpp

@@ -206,6 +206,7 @@ zmq::options_t::options_t () :
     sndbuf (-1),
     rcvbuf (-1),
     tos (0),
+    priority (0),
     type (-1),
     linger (-1),
     connect_timeout (0),
@@ -844,6 +845,13 @@ int zmq::options_t::setsockopt (int option_,
 
             return 0;
 
+        case ZMQ_PRIORITY:
+            if (is_int && value >= 0) {
+                priority = value;
+                return 0;
+            }
+            break;
+
 #endif
 
         default:
@@ -1270,6 +1278,13 @@ int zmq::options_t::getsockopt (int option_,
                 return 0;
             }
             break;
+
+        case ZMQ_PRIORITY:
+            if (is_int) {
+                *value = priority;
+                return 0;
+            }
+            break;
 #endif
 
 

src/options.hpp

@@ -100,6 +100,9 @@ struct options_t
     // Type of service (containing DSCP and ECN socket options)
     int tos;
 
+    // Protocol-defined priority
+    int priority;
+
     //  Socket type.
     int8_t type;
 

src/tcp.cpp

@@ -379,6 +379,10 @@ zmq::fd_t zmq::tcp_open_socket (const char *address_,
     if (options_.tos != 0)
         set_ip_type_of_service (s, options_.tos);
 
+    // Set the protocol-defined priority for this socket
+    if (options_.priority != 0)
+        set_socket_priority (s, options_.priority);
+
     // Set the socket to loopback fastpath if configured.
     if (options_.loopback_fastpath)
         tcp_tune_loopback_fast_path (s);

src/tcp_listener.cpp

@@ -269,5 +269,9 @@ zmq::fd_t zmq::tcp_listener_t::accept ()
     if (options.tos != 0)
         set_ip_type_of_service (sock, options.tos);
 
+    // Set the protocol-defined priority for this client socket
+    if (options.priority != 0)
+        set_socket_priority (sock, options.priority);
+
     return sock;
 }

src/ws_listener.cpp

@@ -294,6 +294,10 @@ zmq::fd_t zmq::ws_listener_t::accept ()
     if (options.tos != 0)
         set_ip_type_of_service (sock, options.tos);
 
+    // Set the protocol-defined priority for this client socket
+    if (options.priority != 0)
+        set_socket_priority (sock, options.priority);
+
     return sock;
 }
 

src/zmq_draft.h

@@ -68,6 +68,7 @@
 #define ZMQ_RECONNECT_STOP 109
 #define ZMQ_HELLO_MSG 110
 #define ZMQ_DISCONNECT_MSG 111
+#define ZMQ_PRIORITY 112
 
 /*  DRAFT ZMQ_RECONNECT_STOP options                                          */
 #define ZMQ_RECONNECT_STOP_CONN_REFUSED 0x1

tests/test_setsockopt.cpp

@@ -137,6 +137,40 @@ void test_setsockopt_bindtodevice ()
     test_context_socket_close (socket);
 }
 
+void test_setsockopt_priority ()
+{
+#ifdef ZMQ_BUILD_DRAFT_API
+#ifdef ZMQ_HAVE_SO_PRIORITY
+    void *socket = test_context_socket (ZMQ_PUSH);
+
+    int val = 5;
+    size_t placeholder = sizeof (val);
+
+    TEST_ASSERT_SUCCESS_ERRNO (
+      zmq_getsockopt (socket, ZMQ_PRIORITY, &val, &placeholder));
+    TEST_ASSERT_EQUAL_INT (0, val);
+
+    val = 3;
+
+    TEST_ASSERT_SUCCESS_ERRNO (
+      zmq_setsockopt (socket, ZMQ_PRIORITY, &val, sizeof (val)));
+    TEST_ASSERT_EQUAL_INT (3, val);
+
+    TEST_ASSERT_SUCCESS_ERRNO (
+      zmq_getsockopt (socket, ZMQ_PRIORITY, &val, &placeholder));
+    TEST_ASSERT_EQUAL_INT (3, val);
+
+    test_context_socket_close (socket);
+#else
+    TEST_IGNORE_MESSAGE ("libzmq without ZMQ_PRIORITY support, "
+                         "ignoring setsockopt_priority test");
+#endif
+#else
+    TEST_IGNORE_MESSAGE ("libzmq without DRAFT support, ignoring "
+                         "setsockopt_priority test");
+#endif
+}
+
 int main ()
 {
     setup_test_environment ();
@@ -146,5 +180,6 @@ int main ()
     RUN_TEST (test_setsockopt_tcp_send_buffer);
     RUN_TEST (test_setsockopt_use_fd);
     RUN_TEST (test_setsockopt_bindtodevice);
+    RUN_TEST (test_setsockopt_priority);
     return UNITY_END ();
 }