Add new option ZMQ_MULTICAST_MAXTPDU to set PGM_MTU.

Fixes #1646

doc/zmq_getsockopt.txt

@@ -410,6 +410,21 @@ Default value:: 1
 Applicable socket types:: all, when using multicast transports
 
 
+ZMQ_MULTICAST_MAXTPDU: Maximum transport data unit size for multicast packets
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+The 'ZMQ_MULTICAST_MAXTPDU' option shall retrieve the maximum transport
+data unit size used for outbound multicast packets.
+
+This must be set at or below the minimum Maximum Transmission Unit (MTU) for
+all network paths over which multicast reception is required.
+
+[horizontal]
+Option value type:: int
+Option value unit:: bytes
+Default value:: 1500
+Applicable socket types:: all, when using multicast transports
+
+
 ZMQ_PLAIN_PASSWORD: Retrieve current password
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 The 'ZMQ_PLAIN_PASSWORD' option shall retrieve the last password set for

doc/zmq_setsockopt.txt

@@ -434,6 +434,21 @@ Default value:: 1
 Applicable socket types:: all, when using multicast transports
 
 
+ZMQ_MULTICAST_MAXTPDU: Maximum transport data unit size for multicast packets
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Sets the maximum transport data unit size used for outbound multicast
+packets.
+
+This must be set at or below the minimum Maximum Transmission Unit (MTU) for
+all network paths over which multicast reception is required.
+
+[horizontal]
+Option value type:: int
+Option value unit:: bytes
+Default value:: 1500
+Applicable socket types:: all, when using multicast transports
+
+
 ZMQ_PLAIN_PASSWORD: Set PLAIN security password
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Sets the password for outgoing connections over TCP or IPC. If you set this

include/zmq.h

@@ -325,6 +325,7 @@ ZMQ_EXPORT uint32_t zmq_msg_routing_id (zmq_msg_t *msg);
 #define ZMQ_THREAD_SAFE 81
 #define ZMQ_TCP_RECV_BUFFER 82
 #define ZMQ_TCP_SEND_BUFFER 83
+#define ZMQ_MULTICAST_MAXTPDU 84
 
 /*  Message options                                                           */
 #define ZMQ_MORE 1

src/config.hpp

@@ -84,9 +84,6 @@ namespace zmq
         //  possible latencies.
         clock_precision = 1000000,
 
-        //  Maximum transport data unit size for PGM (TPDU).
-        pgm_max_tpdu = 1500,
-
         //  On some OSes the signaler has to be emulated using a TCP
         //  connection. In such cases following port is used.
         //  If 0, it lets the OS choose a free port without requiring use of a 

src/options.cpp

@@ -41,6 +41,7 @@ zmq::options_t::options_t () :
     rate (100),
     recovery_ivl (10000),
     multicast_hops (1),
+    multicast_maxtpdu (1500),
     sndbuf (-1),
     rcvbuf (-1),
     tos (0),
@@ -211,6 +212,13 @@ int zmq::options_t::setsockopt (int option_, const void *optval_,
             }
             break;
 
+        case ZMQ_MULTICAST_MAXTPDU:
+            if (is_int && value > 0) {
+                multicast_maxtpdu = value;
+                return 0;
+            }
+            break;
+
         case ZMQ_RCVTIMEO:
             if (is_int && value >= -1) {
                 rcvtimeo = value;
@@ -735,6 +743,13 @@ int zmq::options_t::getsockopt (int option_, void *optval_, size_t *optvallen_)
             }
             break;
 
+        case ZMQ_MULTICAST_MAXTPDU:
+            if (is_int) {
+                *value = multicast_maxtpdu;
+                return 0;
+            }
+            break;
+
         case ZMQ_RCVTIMEO:
             if (is_int) {
                 *value = rcvtimeo;

src/options.hpp

@@ -79,6 +79,10 @@ namespace zmq
         // Sets the time-to-live field in every multicast packet sent.
         int multicast_hops;
 
+        // Sets the maximum transport data unit size in every multicast
+        // packet sent.
+        int multicast_maxtpdu;
+
         // SO_SNDBUF and SO_RCVBUF to be passed to underlying transport sockets.
         int sndbuf;
         int rcvbuf;

src/pgm_socket.cpp

@@ -209,7 +209,7 @@ int zmq::pgm_socket_t::init (bool udp_encapsulation_, const char *network_)
                 goto err_abort;
         }
 
-        const int max_tpdu = (int) pgm_max_tpdu;
+        const int max_tpdu = (int) options.multicast_maxtpdu;
         if (!pgm_setsockopt (sock, IPPROTO_PGM, PGM_MTU, &max_tpdu,
               sizeof (max_tpdu)))
             goto err_abort;
@@ -217,7 +217,7 @@ int zmq::pgm_socket_t::init (bool udp_encapsulation_, const char *network_)
 
     if (receiver) {
         const int recv_only        = 1,
-                  rxw_max_tpdu     = (int) pgm_max_tpdu,
+                  rxw_max_tpdu     = (int) options.multicast_maxtpdu,
                   rxw_sqns         = compute_sqns (rxw_max_tpdu),
                   peer_expiry      = pgm_secs (300),
                   spmr_expiry      = pgm_msecs (25),
@@ -250,7 +250,7 @@ int zmq::pgm_socket_t::init (bool udp_encapsulation_, const char *network_)
     else {
         const int send_only        = 1,
                   max_rte      = (int) ((options.rate * 1000) / 8),
-                  txw_max_tpdu     = (int) pgm_max_tpdu,
+                  txw_max_tpdu     = (int) options.multicast_maxtpdu,
                   txw_sqns         = compute_sqns (txw_max_tpdu),
                   ambient_spm      = pgm_secs (30),
                   heartbeat_spm[]  = { pgm_msecs (100),