Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Added issues 191 and 199

  • Loading branch information...
commit b7ea8e7c3ca44afb97c03c8cc2aadf5bcef96f7e 0 parents
@hintjens hintjens authored
7 191/.gitignore
@@ -0,0 +1,7 @@
+*.o
+core
+pub
+sub
+pub2
+sub2
+
19 191/2-1/crashes.txt
@@ -0,0 +1,19 @@
+Publisher
+=========
+
+Test scenario: start sub, then start pub.
+
+
+#0 0x00007fa7eb5be0c0 in zmq::array_item_t::set_array_index (this=0xec475abd67aa89d8, index_=0) at array.hpp:50
+50 array_index = index_;
+(gdb) bt
+#0 0x00007fa7eb5be0c0 in zmq::array_item_t::set_array_index (this=0xec475abd67aa89d8, index_=0) at array.hpp:50
+#1 0x00007fa7eb5c4bbf in zmq::array_t<zmq::writer_t>::swap (this=0x1525a58, index1_=0, index2_=1) at array.hpp:122
+#2 0x00007fa7eb5c4447 in zmq::dist_t::activated (this=0x1525a50, pipe_=0x1526200) at dist.cpp:103
+#3 0x00007fa7eb5d39d7 in zmq::writer_t::process_activate_writer (this=0x1526200, msgs_read_=0) at pipe.cpp:347
+#4 0x00007fa7eb5cd6a3 in zmq::object_t::process_command (this=0x1526200, cmd_=...) at object.cpp:67
+#5 0x00007fa7eb5e0d1b in zmq::socket_base_t::process_commands (this=0x15258f0, block_=false, throttle_=true) at socket_base.cpp:691
+#6 0x00007fa7eb5e03bb in zmq::socket_base_t::send (this=0x15258f0, msg_=0x7fff638e1ad0, flags_=2) at socket_base.cpp:474
+#7 0x00007fa7eb5ec110 in zmq_send (s_=0x15258f0, msg_=0x7fff638e1ad0, flags_=2) at zmq.cpp:375
+#8 0x0000000000400957 in main () at pub.c:18
+
152 191/2-1/issue191.patch
@@ -0,0 +1,152 @@
+diff --git a/src/dist.cpp b/src/dist.cpp
+index e447bc1..d6a5684 100644
+--- a/src/dist.cpp
++++ b/src/dist.cpp
+@@ -25,9 +25,11 @@
+ #include "err.hpp"
+ #include "own.hpp"
+ #include "msg_content.hpp"
++#include "likely.hpp"
+
+ zmq::dist_t::dist_t (own_t *sink_) :
+ active (0),
++ eligible (0),
+ more (false),
+ sink (sink_),
+ terminating (false)
+@@ -41,20 +43,24 @@ zmq::dist_t::~dist_t ()
+
+ void zmq::dist_t::attach (writer_t *pipe_)
+ {
+- // If we are in the middle of sending a message, let's postpone plugging
+- // in the pipe.
+- if (!terminating && more) {
+- new_pipes.push_back (pipe_);
+- return;
+- }
+-
+ pipe_->set_event_sink (this);
+
+- pipes.push_back (pipe_);
+- pipes.swap (active, pipes.size () - 1);
+- active++;
++ // If we are in the middle of sending a message, we'll add new pipe
++ // into the list of eligible pipes. Otherwise we add it to the list
++ // of active pipes.
++ if (more) {
++ pipes.push_back (pipe_);
++ pipes.swap (eligible, pipes.size () - 1);
++ eligible++;
++ }
++ else {
++ pipes.push_back (pipe_);
++ pipes.swap (active, pipes.size () - 1);
++ active++;
++ eligible++;
++ }
+
+- if (terminating) {
++ if (unlikely (terminating)) {
+ sink->register_term_acks (1);
+ pipe_->terminate ();
+ }
+@@ -72,21 +78,32 @@ void zmq::dist_t::terminate ()
+
+ void zmq::dist_t::terminated (writer_t *pipe_)
+ {
+- // Remove the pipe from the list; adjust number of active pipes
+- // accordingly.
++ // Remove the pipe from the list; adjust number of active and/or
++ // eligible pipes accordingly.
+ if (pipes.index (pipe_) < active)
+ active--;
++ if (pipes.index (pipe_) < eligible)
++ eligible--;
+ pipes.erase (pipe_);
+
+- if (terminating)
++ if (unlikely (terminating))
+ sink->unregister_term_ack ();
+ }
+
+ void zmq::dist_t::activated (writer_t *pipe_)
+ {
+- // Move the pipe to the list of active pipes.
+- pipes.swap (pipes.index (pipe_), active);
+- active++;
++ // If we are in the middle of sending a message, we'll add new pipe
++ // into the list of eligible pipes. Otherwise we add it to the list
++ // of active pipes.
++ if (more) {
++ pipes.swap (pipes.index (pipe_), eligible);
++ eligible++;
++ }
++ else {
++ pipes.swap (pipes.index (pipe_), active);
++ active++;
++ eligible++;
++ }
+ }
+
+ int zmq::dist_t::send (zmq_msg_t *msg_, int flags_)
+@@ -97,9 +114,9 @@ int zmq::dist_t::send (zmq_msg_t *msg_, int flags_)
+ // Push the message to active pipes.
+ distribute (msg_, flags_);
+
+- // If mutlipart message is fully sent, activate new pipes.
+- if (more && !msg_more)
+- clear_new_pipes ();
++ // If multipart message is fully sent, activate all the eligible pipes.
++ if (!msg_more)
++ active = eligible;
+
+ more = msg_more;
+
+@@ -181,16 +198,3 @@ bool zmq::dist_t::write (class writer_t *pipe_, zmq_msg_t *msg_)
+ pipe_->flush ();
+ return true;
+ }
+-
+-void zmq::dist_t::clear_new_pipes ()
+-{
+- for (new_pipes_t::iterator it = new_pipes.begin (); it != new_pipes.end ();
+- ++it) {
+- (*it)->set_event_sink (this);
+- pipes.push_back (*it);
+- pipes.swap (active, pipes.size () - 1);
+- active++;
+- }
+- new_pipes.clear ();
+-}
+-
+diff --git a/src/dist.hpp b/src/dist.hpp
+index ad9767a..6c73941 100644
+--- a/src/dist.hpp
++++ b/src/dist.hpp
+@@ -63,17 +63,18 @@ namespace zmq
+ typedef array_t <class writer_t> pipes_t;
+ pipes_t pipes;
+
+- // List of new pipes that were not yet inserted into 'pipes' list.
+- // These pipes are moves to 'pipes' list once the current multipart
+- // message is fully sent. This way we avoid sending incomplete messages
+- // to peers.
+- typedef std::vector <class writer_t*> new_pipes_t;
+- new_pipes_t new_pipes;
+-
+ // Number of active pipes. All the active pipes are located at the
+- // beginning of the pipes array.
++ // beginning of the pipes array. These are the pipes the messages
++ // can be sent to at the moment.
+ pipes_t::size_type active;
+
++ // Number of pipes eligible for sending messages to. This includes all
++ // the active pipes plus all the pipes that we can in theory send
++ // messages to (the HWM is not yet reached), but sending a message
++ // to them would result in partial message being delivered, ie. message
++ // with initial parts missing.
++ pipes_t::size_type eligible;
++
+ // True if last we are in the middle of a multipart message.
+ bool more;
+
35 191/2-1/pub.c
@@ -0,0 +1,35 @@
+#include "zmq.h"
+#include "assert.h"
+#include "stdint.h"
+#include "string.h"
+#include "stdio.h"
+
+int main ()
+{
+ void *ctx = zmq_init (1);
+ assert (ctx);
+ void *s = zmq_socket (ctx, ZMQ_PUB);
+ uint64_t sz = 100;
+ int rc = zmq_setsockopt (s, ZMQ_HWM, &sz, sizeof (sz));
+ assert (rc == 0);
+ rc = zmq_bind (s, "tcp://127.0.0.1:2211");
+ assert (rc == 0);
+
+ zmq_msg_t msg;
+ while (1) {
+ puts ("sending");
+ zmq_msg_init_size (&msg, 5);
+ rc = zmq_send (s, &msg, ZMQ_SNDMORE);
+ zmq_msg_close (&msg);
+
+ zmq_msg_init_size (&msg, 4);
+ rc = zmq_send (s, &msg, ZMQ_SNDMORE);
+ zmq_msg_close (&msg);
+
+ zmq_msg_init_size (&msg, 1000);
+ rc = zmq_send (s, &msg, 0);
+ zmq_msg_close (&msg);
+ }
+
+ return 0;
+}
55 191/2-1/sub.c
@@ -0,0 +1,55 @@
+#include "zmq.h"
+#include "assert.h"
+#include "stdint.h"
+#include "stdio.h"
+
+int
+zsockopt_rcvmore (void *socket)
+{
+ int64_t rcvmore;
+ size_t type_size = sizeof (int64_t);
+ zmq_getsockopt (socket, ZMQ_RCVMORE, &rcvmore, &type_size);
+ return (int) rcvmore;
+}
+
+
+int main ()
+{
+ void *ctx = zmq_init (1);
+ assert (ctx);
+ void *s = zmq_socket (ctx, ZMQ_SUB);
+ int rc = zmq_setsockopt (s, ZMQ_SUBSCRIBE, "", 0);
+ assert (rc == 0);
+ uint64_t sz = 100;
+ rc = zmq_setsockopt (s, ZMQ_HWM, &sz, sizeof (sz));
+ assert (rc == 0);
+ rc = zmq_connect (s, "tcp://127.0.0.1:2211");
+ assert (rc == 0);
+
+ while (1) {
+ puts ("receiving");
+ zmq_msg_t msg;
+ zmq_msg_init (&msg);
+ rc = zmq_recv (s, &msg, 0);
+ assert (zmq_msg_size (&msg) == 5);
+ zmq_msg_close (&msg);
+ int more = zsockopt_rcvmore (s);
+ assert (more);
+
+ zmq_msg_init (&msg);
+ rc = zmq_recv (s, &msg, 0);
+ assert (zmq_msg_size (&msg) == 4);
+ zmq_msg_close (&msg);
+ more = zsockopt_rcvmore (s);
+ assert (more);
+
+ zmq_msg_init (&msg);
+ rc = zmq_recv (s, &msg, 0);
+ assert (zmq_msg_size (&msg) == 1000);
+ zmq_msg_close (&msg);
+ more = zsockopt_rcvmore (s);
+ assert (!more);
+ }
+ return 0;
+}
+
169 191/3-0/0001-Message-atomicity-problem-in-PUB-socket-fixed.patch
@@ -0,0 +1,169 @@
+From dfb510b844f4a19f9405ae51846bf2dc8221c84d Mon Sep 17 00:00:00 2001
+From: Martin Sustrik <sustrik@250bpm.com>
+Date: Tue, 26 Apr 2011 08:25:46 +0200
+Subject: [PATCH] Message atomicity problem in PUB socket fixed.
+
+Reaching the HWM caused breaking message atomicity when the
+flow was reestablished - initial parts of multipart messages
+may have been lost.
+
+Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
+---
+ src/dist.cpp | 69 +++++++++++++++++++++++++++++++--------------------------
+ src/dist.hpp | 17 +++++++------
+ 2 files changed, 46 insertions(+), 40 deletions(-)
+
+diff --git a/src/dist.cpp b/src/dist.cpp
+index 093da79..be1d5af 100644
+--- a/src/dist.cpp
++++ b/src/dist.cpp
+@@ -23,9 +23,11 @@
+ #include "err.hpp"
+ #include "own.hpp"
+ #include "msg.hpp"
++#include "likely.hpp"
+
+ zmq::dist_t::dist_t (own_t *sink_) :
+ active (0),
++ eligible (0),
+ more (false),
+ sink (sink_),
+ terminating (false)
+@@ -39,20 +41,24 @@ zmq::dist_t::~dist_t ()
+
+ void zmq::dist_t::attach (writer_t *pipe_)
+ {
+- // If we are in the middle of sending a message, let's postpone plugging
+- // in the pipe.
+- if (!terminating && more) {
+- new_pipes.push_back (pipe_);
+- return;
+- }
+-
+ pipe_->set_event_sink (this);
+
+- pipes.push_back (pipe_);
+- pipes.swap (active, pipes.size () - 1);
+- active++;
++ // If we are in the middle of sending a message, we'll add new pipe
++ // into the list of eligible pipes. Otherwise we add it to the list
++ // of active pipes.
++ if (more) {
++ pipes.push_back (pipe_);
++ pipes.swap (eligible, pipes.size () - 1);
++ eligible++;
++ }
++ else {
++ pipes.push_back (pipe_);
++ pipes.swap (active, pipes.size () - 1);
++ active++;
++ eligible++;
++ }
+
+- if (terminating) {
++ if (unlikely (terminating)) {
+ sink->register_term_acks (1);
+ pipe_->terminate ();
+ }
+@@ -70,21 +76,32 @@ void zmq::dist_t::terminate ()
+
+ void zmq::dist_t::terminated (writer_t *pipe_)
+ {
+- // Remove the pipe from the list; adjust number of active pipes
+- // accordingly.
++ // Remove the pipe from the list; adjust number of active and/or
++ // eligible pipes accordingly.
+ if (pipes.index (pipe_) < active)
+ active--;
++ if (pipes.index (pipe_) < eligible)
++ eligible--;
+ pipes.erase (pipe_);
+
+- if (terminating)
++ if (unlikely (terminating))
+ sink->unregister_term_ack ();
+ }
+
+ void zmq::dist_t::activated (writer_t *pipe_)
+ {
+- // Move the pipe to the list of active pipes.
+- pipes.swap (pipes.index (pipe_), active);
+- active++;
++ // If we are in the middle of sending a message, we'll add new pipe
++ // into the list of eligible pipes. Otherwise we add it to the list
++ // of active pipes.
++ if (more) {
++ pipes.swap (pipes.index (pipe_), eligible);
++ eligible++;
++ }
++ else {
++ pipes.swap (pipes.index (pipe_), active);
++ active++;
++ eligible++;
++ }
+ }
+
+ int zmq::dist_t::send (msg_t *msg_, int flags_)
+@@ -95,9 +112,9 @@ int zmq::dist_t::send (msg_t *msg_, int flags_)
+ // Push the message to active pipes.
+ distribute (msg_, flags_);
+
+- // If mutlipart message is fully sent, activate new pipes.
+- if (more && !msg_more)
+- clear_new_pipes ();
++ // If mutlipart message is fully sent, activate all the eligible pipes.
++ if (!msg_more)
++ active = eligible;
+
+ more = msg_more;
+
+@@ -150,15 +167,3 @@ bool zmq::dist_t::write (class writer_t *pipe_, msg_t *msg_)
+ return true;
+ }
+
+-void zmq::dist_t::clear_new_pipes ()
+-{
+- for (new_pipes_t::iterator it = new_pipes.begin (); it != new_pipes.end ();
+- ++it) {
+- (*it)->set_event_sink (this);
+- pipes.push_back (*it);
+- pipes.swap (active, pipes.size () - 1);
+- active++;
+- }
+- new_pipes.clear ();
+-}
+-
+diff --git a/src/dist.hpp b/src/dist.hpp
+index ea05305..f510bd6 100644
+--- a/src/dist.hpp
++++ b/src/dist.hpp
+@@ -63,17 +63,18 @@ namespace zmq
+ typedef array_t <class writer_t> pipes_t;
+ pipes_t pipes;
+
+- // List of new pipes that were not yet inserted into 'pipes' list.
+- // These pipes are moves to 'pipes' list once the current multipart
+- // message is fully sent. This way we avoid sending incomplete messages
+- // to peers.
+- typedef std::vector <class writer_t*> new_pipes_t;
+- new_pipes_t new_pipes;
+-
+ // Number of active pipes. All the active pipes are located at the
+- // beginning of the pipes array.
++ // beginning of the pipes array. These are the pipes the messages
++ // can be sent to at the moment.
+ pipes_t::size_type active;
+
++ // Number of pipes eligible for sending messages to. This includes all
++ // the active pipes plus all the pipes that we can in theory send
++ // messages to (the HWM is not yet reached), but sending a message
++ // to them would result in partial message being delivered, ie. message
++ // with initial parts missing.
++ pipes_t::size_type eligible;
++
+ // True if last we are in the middle of a multipart message.
+ bool more;
+
+--
+1.7.0.4
+
94 191/3-0/crashes.txt
@@ -0,0 +1,94 @@
+Publisher
+=========
+
+Test scenario: start sub, then start pub.
+
+Assertion failed: size >= 2 (zmq_init.cpp:120)
+#0 0x00007f772651eba5 in raise (sig=<value optimized out>) at ../nptl/sysdeps/unix/sysv/linux/raise.c:64
+#1 0x00007f77265226b0 in abort () at abort.c:92
+#2 0x00007f77271b1cfc in zmq::zmq_init_t::write (this=0x696c70, msg_=0x696e10) at zmq_init.cpp:120
+#3 0x00007f7727186790 in zmq::decoder_t::message_ready (this=0x696dc8) at decoder.cpp:139
+#4 0x00007f77271b06c5 in zmq::decoder_base_t<zmq::decoder_t>::process_buffer (this=0x696dc8, data_=0x6986a0 "\001~i", size_=2) at decoder.hpp:122
+#5 0x00007f77271b0003 in zmq::zmq_engine_t::in_event (this=0x696d90) at zmq_engine.cpp:127
+#6 0x00007f77271afe25 in zmq::zmq_engine_t::plug (this=0x696d90, io_thread_=0x6963b0, inout_=0x696d28) at zmq_engine.cpp:78
+#7 0x00007f77271b1f91 in zmq::zmq_init_t::process_plug (this=0x696c70) at zmq_init.cpp:178
+#8 0x00007f7727190a7f in zmq::object_t::process_command (this=0x696c70, cmd_=...) at object.cpp:75
+#9 0x00007f772718d385 in zmq::io_thread_t::in_event (this=0x6963b0) at io_thread.cpp:81
+#10 0x00007f7727189c3f in zmq::epoll_t::loop (this=0x6963f0) at epoll.cpp:161
+#11 0x00007f7727189d16 in zmq::epoll_t::worker_routine (arg_=0x6963f0) at epoll.cpp:174
+#12 0x00007f77271a6b9f in thread_routine (arg_=0x696460) at thread.cpp:73
+#13 0x00007f7726af8971 in start_thread (arg=<value optimized out>) at pthread_create.c:304
+#14 0x00007f77265d192d in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:112
+#15 0x0000000000000000 in ?? ()
+
+Notes: did not reproduce after rebuilding libzmq with --enable-debug.
+
+
+Publisher
+=========
+
+Test scenario: start pub, then start sub.
+
+Program terminated with signal 11, Segmentation fault.
+#0 0x00007f1313b02520 in zmq::array_item_t::set_array_index (this=0x6f6f6f6f6f6f6f87, index_=0) at array.hpp:50
+50 array_index = index_;
+(gdb) bt
+#0 0x00007f1313b02520 in zmq::array_item_t::set_array_index (this=0x6f6f6f6f6f6f6f87, index_=0) at array.hpp:50
+#1 0x00007f1313b0970b in zmq::array_t<zmq::writer_t>::swap (this=0xd99a50, index1_=0, index2_=1) at array.hpp:122
+#2 0x00007f1313b09198 in zmq::dist_t::activated (this=0xd99a48, pipe_=0xd9a220) at dist.cpp:97
+#3 0x00007f1313b187db in zmq::writer_t::process_activate_writer (this=0xd9a220, msgs_read_=0) at pipe.cpp:262
+#4 0x00007f1313b12a47 in zmq::object_t::process_command (this=0xd9a220, cmd_=...) at object.cpp:67
+#5 0x00007f1313b25c69 in zmq::socket_base_t::process_commands (this=0xd998f0, block_=false, throttle_=true) at socket_base.cpp:700
+#6 0x00007f1313b252fd in zmq::socket_base_t::send (this=0xd998f0, msg_=0x7fff68119d20, flags_=0) at socket_base.cpp:481
+#7 0x00007f1313b2fe43 in zmq_sendmsg (s_=0xd998f0, msg_=0x7fff68119d20, flags_=0) at zmq.cpp:266
+#8 0x00007f1313b2fb53 in zmq_send (s_=0xd998f0, buf_=0x7fff68119d70, len_=1000, flags_=0) at zmq.cpp:219
+#9 0x00000000004008d8 in main () at pub.c:25
+
+Notes: this and following error happen arbitrarily, with both publishers.
+
+
+Publisher
+=========
+
+Test scenario: start pub2, then start sub2.
+
+Program terminated with signal 11, Segmentation fault.
+#0 0x00007f262dbd2468 in zmq::writer_t::check_write (this=0x0,
+msg_=0x7fff811835d0) at pipe.cpp:190
+190 if (unlikely (!active))
+(gdb) bt
+#0 0x00007f262dbd2468 in zmq::writer_t::check_write (this=0x0, msg_=0x7fff811835d0) at pipe.cpp:190
+#1 0x00007f262dbd24d1 in zmq::writer_t::write (this=0x0, msg_=0x7fff811835d0) at pipe.cpp:203
+#2 0x00007f262dbc3478 in zmq::dist_t::write (this=0x119ea48, pipe_=0x0, msg_=0x7fff811835d0) at dist.cpp:160
+#3 0x00007f262dbc33ac in zmq::dist_t::distribute (this=0x119ea48, msg_=0x7fff811835d0, flags_=2) at dist.cpp:141
+#4 0x00007f262dbc3250 in zmq::dist_t::send (this=0x119ea48, msg_=0x7fff811835d0, flags_=2) at dist.cpp:113
+#5 0x00007f262dbe4308 in zmq::xpub_t::xsend (this=0x119e8f0, msg_=0x7fff811835d0, flags_=2) at xpub.cpp:57
+#6 0x00007f262dbdf357 in zmq::socket_base_t::send (this=0x119e8f0, msg_=0x7fff811835d0, flags_=2) at socket_base.cpp:490
+#7 0x00007f262dbe9e43 in zmq_sendmsg (s_=0x119e8f0, msg_=0x7fff811835d0, flags_=2) at zmq.cpp:266
+#8 0x0000000000400957 in main () at pub2.c:22
+
+Notes: this and following error happen arbitrarily, with both publishers.
+
+
+Subscriber
+==========
+
+Test scenario: any, subscriber always crashes.
+
+sub: sub.c:25: main: Assertion `rc == 0' failed.
+Program terminated with signal 11, Segmentation fault.
+#0 0x00007f4ac8fc4468 in zmq::writer_t::check_write (this=0x0, msg_=0x7fff50f7d980) at pipe.cpp:190
+190 if (unlikely (!active))
+(gdb) bt
+#0 0x00007f4ac8fc4468 in zmq::writer_t::check_write (this=0x0, msg_=0x7fff50f7d980) at pipe.cpp:190
+#1 0x00007f4ac8fc44d1 in zmq::writer_t::write (this=0x0, msg_=0x7fff50f7d980) at pipe.cpp:203
+#2 0x00007f4ac8fb5478 in zmq::dist_t::write (this=0x1f96a48, pipe_=0x0, msg_=0x7fff50f7d980) at dist.cpp:160
+#3 0x00007f4ac8fb53ac in zmq::dist_t::distribute (this=0x1f96a48, msg_=0x7fff50f7d980, flags_=2) at dist.cpp:141
+#4 0x00007f4ac8fb5250 in zmq::dist_t::send (this=0x1f96a48, msg_=0x7fff50f7d980, flags_=2) at dist.cpp:113
+#5 0x00007f4ac8fd6308 in zmq::xpub_t::xsend (this=0x1f968f0, msg_=0x7fff50f7d980, flags_=2) at xpub.cpp:57
+#6 0x00007f4ac8fd1357 in zmq::socket_base_t::send (this=0x1f968f0, msg_=0x7fff50f7d980, flags_=2) at socket_base.cpp:490
+#7 0x00007f4ac8fdbe43 in zmq_sendmsg (s_=0x1f968f0, msg_=0x7fff50f7d980, flags_=2) at zmq.cpp:266
+#8 0x00007f4ac8fdbb53 in zmq_send (s_=0x1f968f0, buf_=0x400a73, len_=5, flags_=2) at zmq.cpp:219
+#9 0x00000000004008a7 in main () at sub.c:14
+
+Notes: this does NOT happen when using sub2.c.
30 191/3-0/pub.c
@@ -0,0 +1,30 @@
+#include "zmq.h"
+#include "assert.h"
+#include "stdint.h"
+#include "string.h"
+
+int main ()
+{
+ void *ctx = zmq_init (1);
+ assert (ctx);
+ void *s = zmq_socket (ctx, ZMQ_PUB);
+ int sz = 100;
+ int rc = zmq_setsockopt (s, ZMQ_SNDHWM, &sz, sizeof (sz));
+ assert (rc == 0);
+ rc = zmq_bind (s, "tcp://127.0.0.1:2211");
+ assert (rc == 0);
+
+ uint32_t seq = 0;
+ char buff [1000];
+ memset (buff, 111, 1000);
+ while (1) {
+ rc = zmq_send (s, "topic", 5, ZMQ_SNDMORE);
+ assert (rc == 5);
+ rc = zmq_send (s, &seq, 4, ZMQ_SNDMORE);
+ assert (rc == 4);
+ rc = zmq_send (s, buff, 1000, 0);
+ assert (rc == 1000);
+ }
+
+ return 0;
+}
35 191/3-0/pub2.c
@@ -0,0 +1,35 @@
+#include "zmq.h"
+#include "assert.h"
+#include "stdint.h"
+#include "string.h"
+#include "stdio.h"
+
+int main ()
+{
+ void *ctx = zmq_init (1);
+ assert (ctx);
+ void *s = zmq_socket (ctx, ZMQ_PUB);
+ int sz = 100;
+ int rc = zmq_setsockopt (s, ZMQ_SNDHWM, &sz, sizeof (sz));
+ assert (rc == 0);
+ rc = zmq_bind (s, "tcp://127.0.0.1:2211");
+ assert (rc == 0);
+
+ zmq_msg_t msg;
+ while (1) {
+ puts ("sending");
+ zmq_msg_init_size (&msg, 5);
+ rc = zmq_sendmsg (s, &msg, ZMQ_SNDMORE);
+ zmq_msg_close (&msg);
+
+ zmq_msg_init_size (&msg, 4);
+ rc = zmq_sendmsg (s, &msg, ZMQ_SNDMORE);
+ zmq_msg_close (&msg);
+
+ zmq_msg_init_size (&msg, 1000);
+ rc = zmq_sendmsg (s, &msg, 0);
+ zmq_msg_close (&msg);
+ }
+
+ return 0;
+}
42 191/3-0/sub.c
@@ -0,0 +1,42 @@
+#include "zmq.h"
+#include "assert.h"
+#include "stdint.h"
+
+int main ()
+{
+ void *ctx = zmq_init (1);
+ assert (ctx);
+ void *s = zmq_socket (ctx, ZMQ_SUB);
+ int rc = zmq_setsockopt (s, ZMQ_SUBSCRIBE, "", 0);
+ assert (rc == 0);
+ int sz = 100;
+ rc = zmq_setsockopt (s, ZMQ_RCVHWM, &sz, sizeof (sz));
+ assert (rc == 0);
+ rc = zmq_connect (s, "tcp://127.0.0.1:2211");
+ assert (rc == 0);
+
+ char buff [1000];
+ while (1) {
+ rc = zmq_recv (s, buff, 1000, 0);
+ assert (rc == 5);
+ int more;
+ size_t moresz;
+ rc = zmq_getsockopt (s, ZMQ_RCVMORE, &more, &moresz);
+ assert (rc == 0);
+ assert (more);
+
+ rc = zmq_recv (s, buff, 1000, 0);
+ assert (rc == 4);
+ rc = zmq_getsockopt (s, ZMQ_RCVMORE, &more, &moresz);
+ assert (rc == 0);
+ assert (more);
+
+ rc = zmq_recv (s, buff, 1000, 0);
+ assert (rc == 1000);
+ rc = zmq_getsockopt (s, ZMQ_RCVMORE, &more, &moresz);
+ assert (rc == 0);
+ assert (!more);
+ }
+
+ return 0;
+}
55 191/3-0/sub2.c
@@ -0,0 +1,55 @@
+#include "zmq.h"
+#include "assert.h"
+#include "stdint.h"
+#include "stdio.h"
+
+int
+zsockopt_rcvmore (void *socket)
+{
+ int rcvmore;
+ size_t type_size = sizeof (int);
+ zmq_getsockopt (socket, ZMQ_RCVMORE, &rcvmore, &type_size);
+ return rcvmore;
+}
+
+
+int main ()
+{
+ void *ctx = zmq_init (1);
+ assert (ctx);
+ void *s = zmq_socket (ctx, ZMQ_SUB);
+ int rc = zmq_setsockopt (s, ZMQ_SUBSCRIBE, "", 0);
+ assert (rc == 0);
+ int sz = 100;
+ rc = zmq_setsockopt (s, ZMQ_RCVHWM, &sz, sizeof (sz));
+ assert (rc == 0);
+ rc = zmq_connect (s, "tcp://127.0.0.1:2211");
+ assert (rc == 0);
+
+ while (1) {
+ puts ("receiving");
+ zmq_msg_t msg;
+ zmq_msg_init (&msg);
+ rc = zmq_recvmsg (s, &msg, 0);
+ assert (zmq_msg_size (&msg) == 5);
+ zmq_msg_close (&msg);
+ int more = zsockopt_rcvmore (s);
+ assert (more);
+
+ zmq_msg_init (&msg);
+ rc = zmq_recvmsg (s, &msg, 0);
+ assert (zmq_msg_size (&msg) == 4);
+ zmq_msg_close (&msg);
+ more = zsockopt_rcvmore (s);
+ assert (more);
+
+ zmq_msg_init (&msg);
+ rc = zmq_recvmsg (s, &msg, 0);
+ assert (zmq_msg_size (&msg) == 1000);
+ zmq_msg_close (&msg);
+ more = zsockopt_rcvmore (s);
+ assert (!more);
+ }
+ return 0;
+}
+
10 191/README
@@ -0,0 +1,10 @@
+This gist holds code and listings for https://github.com/zeromq/libzmq/issues/191
+
+.:
+2-1 3-0 README.md
+
+./2-1:
+crashes.txt pub.c sub.c
+
+./3-0:
+crashes.txt pub2.c pub.c sub2.c sub.c
192 199/2-1/server.c
@@ -0,0 +1,192 @@
+#include <zmq.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <assert.h>
+#include <string.h>
+#include <errno.h>
+
+
+#include <sys/types.h>
+#include <unistd.h>
+
+void to_hex(const char *data, size_t len, char *dest) {
+ int i;
+ for (i=0; i<len; ++i) {
+ snprintf(dest+2*i, 3, "%02x", (int)data[i]);
+ }
+}
+
+void stupid_recv(void *zsock) {
+ int64_t more;
+ size_t more_size = sizeof (more);
+
+ assert(zsock);
+
+ zmq_msg_t message;
+ assert(0 == zmq_msg_init(&message));
+
+ assert( 0 == zmq_recv(zsock, &message, 0) );
+ assert(zmq_msg_size(&message) > 0);
+ assert( 0 == zmq_getsockopt(zsock, ZMQ_RCVMORE, &more, &more_size) );
+
+ // check message size
+ fprintf(stderr, "STUPID RECV (JUST FOR SHOWING WHATS WAITING): mp part [%.*s] (%p) received. more==%li.\n",
+ (int)zmq_msg_size(&message), (char*)zmq_msg_data(&message),
+ zmq_msg_data(&message), more);
+ fflush(stderr);
+
+ assert( 0 == zmq_msg_close(&message) );
+}
+
+// * receive a message from socket
+// * message contents must equal `stress_id`+`expectation`
+// * check if ZMQ_RCVMORE is set, must match parameter `expect_more`
+void recv_assert(void *zsock, const char *identity,
+ const char *stress_id, const char *expectation, char expect_more) {
+ int64_t more;
+ size_t more_size = sizeof (more);
+
+ assert(zsock);
+
+ zmq_msg_t message;
+ assert(0 == zmq_msg_init(&message));
+
+ assert( 0 == zmq_recv(zsock, &message, 0) );
+ assert(zmq_msg_size(&message) > 0);
+ assert( 0 == zmq_getsockopt(zsock, ZMQ_RCVMORE, &more, &more_size) );
+
+ // check message size
+ if ((strlen(stress_id) + strlen(expectation)) != zmq_msg_size(&message)) {
+ fprintf(stderr, "%s: mp part [%.*s] (%p) but expected [%s%s] (size missmatch).\n",
+ identity, (int)zmq_msg_size(&message), (char*)zmq_msg_data(&message),
+ zmq_msg_data(&message),
+ stress_id, expectation); fflush(stderr);
+ while (1) stupid_recv(zsock);
+ assert(0);
+ }
+
+ // check message contents
+ size_t stress_id_len = strlen(stress_id);
+ if (strncmp(stress_id, (char*)zmq_msg_data(&message), stress_id_len) ||
+ strncmp(expectation, ((char*)zmq_msg_data(&message))+stress_id_len, strlen(expectation))) {
+ fprintf(stderr, "%s: mp part [%.*s] (%p) but expected [%s%s].\n",
+ identity, (int)zmq_msg_size(&message), (char*)zmq_msg_data(&message),
+ zmq_msg_data(&message),
+ stress_id, expectation); fflush(stderr);
+ while (1) stupid_recv(zsock);
+ assert(0);
+ }
+
+ // check if we expect more, but there is no more
+ if (expect_more && (!more)) {
+ fprintf(stderr, "%s: mp part [%.*s] ZMQ_RCVMORE not set (unexpected!)\n",
+ identity, (int)zmq_msg_size(&message), (char*)zmq_msg_data(&message)); fflush(stderr);
+ while (1) stupid_recv(zsock);
+ assert(0);
+ }
+
+ // check if we expect last message, but there are more
+ if ((!expect_more) && more) {
+ fprintf(stderr, "%s: mp part [%.*s] ZMQ_RCVMORE set (unexpected!)\n",
+ identity, (int)zmq_msg_size(&message), (char*)zmq_msg_data(&message)); fflush(stderr);
+ while (1) stupid_recv(zsock);
+ assert(0);
+ }
+
+ // we got what we expected - dont assert(0)
+ fprintf(stderr, "%s: mp part [%.*s]", identity, (int)zmq_msg_size(&message), (char*)zmq_msg_data(&message));
+ if (expect_more) {
+ fprintf(stderr, " and more.");
+ } else {
+ fprintf(stderr, " and no more.");
+ }
+ fprintf(stderr, " (as expected)\n"); fflush(stderr);
+ assert( 0 == zmq_msg_close(&message) );
+}
+
+void receive_mp(void *zsock) {
+ int64_t more;
+ size_t more_size = sizeof (more);
+
+ assert(zsock);
+
+ zmq_msg_t id_message;
+ assert(0 == zmq_msg_init(&id_message));
+
+ assert( 0 == zmq_recv(zsock, &id_message, 0) );
+ assert( 0 == zmq_getsockopt(zsock, ZMQ_RCVMORE, &more, &more_size) );
+
+ assert(more);
+ // we store the identity
+ assert(zmq_msg_size(&id_message) > 0);
+ char *identity = (char*)malloc(zmq_msg_size(&id_message)*2+1);
+ assert(identity);
+ to_hex((char*)zmq_msg_data(&id_message), zmq_msg_size(&id_message), identity);
+ assert( 0 == zmq_msg_close(&id_message) );
+
+ // receive first message
+ // this message starts with a number which is the
+ // prefix for all following message
+ // (to detect intermixed mp-messages)
+ zmq_msg_t first_message;
+ assert(0 == zmq_msg_init(&first_message));
+ more_size = sizeof (more);
+ assert( 0 == zmq_recv(zsock, &first_message, 0) );
+ assert( 0 == zmq_getsockopt(zsock, ZMQ_RCVMORE, &more, &more_size) );
+ assert(more);
+
+ assert(zmq_msg_data(&first_message));
+ // find length of this "prefix"
+ int stress_id_len = strspn((char*)zmq_msg_data(&first_message), "0123456789.");
+ assert(stress_id_len > 0);
+ char *stress_id = (char*)malloc(stress_id_len+1);
+ assert(stress_id);
+ memcpy(stress_id, zmq_msg_data(&first_message), stress_id_len);
+ stress_id[stress_id_len] = 0;
+
+ // check if the first message has the form PREFIX+"stress"
+ if (0 == strncmp((char*)zmq_msg_data(&first_message) + stress_id_len, "stress", 6)) {
+ fprintf(stderr, "%s: incoming stress mp [%.*s]\n", identity,
+ (int)zmq_msg_size(&first_message), (char*)zmq_msg_data(&first_message)); fflush(stderr);
+ } else {
+ // protocol violation (e.g. thats not what we sent)
+ fprintf(stderr, "%s: mp does not start with \"XXXXstress\": [%.*s] (%p)\n",
+ identity, (int)zmq_msg_size(&first_message), (char*)zmq_msg_data(&first_message),
+ zmq_msg_data(&first_message)); fflush(stderr);
+ while (1) stupid_recv(zsock);
+ assert(0);
+ }
+ assert( 0 == zmq_msg_close(&first_message) );
+
+ recv_assert(zsock, identity, stress_id, "STRESS A", 1);
+ recv_assert(zsock, identity, stress_id, "STRESS B", 1);
+ recv_assert(zsock, identity, stress_id, "STRESS C", 1);
+ recv_assert(zsock, identity, stress_id, "STRESS D", 1);
+ recv_assert(zsock, identity, stress_id, "STRESS E", 0);
+ free(stress_id);
+ free(identity);
+}
+
+int main(int argc, char **argv)
+{
+ // create context
+ void *context = zmq_init(1);
+ assert(context);
+
+ // create socket
+ void *zsock = zmq_socket(context, ZMQ_XREP);
+ assert(zsock);
+
+ // listen
+ int ret = zmq_bind(zsock, "tcp://127.0.0.1:5555");
+ assert(0 == ret);
+
+ // wait until protocol is violated...
+ while (1) receive_mp(zsock);
+ //while (1) stupid_recv(zsock);
+
+ // this wont be reached...
+ assert(0 == zmq_close(zsock));
+ assert(0 == zmq_term(context));
+ return 0;
+}
114 199/2-1/stress.c
@@ -0,0 +1,114 @@
+#include <zmq.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <assert.h>
+#include <string.h>
+#include <errno.h>
+
+
+#include <sys/types.h>
+#include <unistd.h>
+
+int zsend(void *socket, const char *buffer, const size_t len, char more);
+int zsend(void *socket, const char *buffer, const size_t len, char more) {
+ assert(socket);
+ assert(len > 0);
+ assert(buffer);
+
+ zmq_msg_t m;
+ assert(0 == zmq_msg_init_size(&m, len));
+ void *m_data = zmq_msg_data(&m);
+ assert(m_data);
+ memcpy(m_data, buffer, len);
+
+ if (zmq_send(socket, &m, more?ZMQ_SNDMORE:0)) {
+ fprintf(stderr, "failed to send. errno: %i\n", zmq_errno());
+ return 1;
+ }
+ assert(0 == zmq_msg_close(&m));
+ return 0;
+}
+
+int stress(void *s, const char *prefix) {
+ char buff[32];
+ int plen;
+
+ assert(prefix);
+ plen = strlen(prefix);
+ assert(plen < 20);
+
+ memcpy(buff, prefix, plen);
+
+ memcpy(buff+plen, "stress", 6);
+ if (zsend(s, buff, plen+6, 1)) {
+ fprintf(stderr, "failed to send initial stress\n");
+ return 1;
+ }
+
+ memcpy(buff+plen, "STRESS A", 8);
+ if (zsend(s, buff, plen+8, 1)) {
+ fprintf(stderr, "failed to send STRESS A\n");
+ return 1;
+ }
+
+ memcpy(buff+plen, "STRESS B", 8);
+ if (zsend(s, buff, plen+8, 1)) {
+ fprintf(stderr, "failed to send STRESS B\n");
+ return 1;
+ }
+
+ memcpy(buff+plen, "STRESS C", 8);
+ if (zsend(s, buff, plen+8, 1)) {
+ fprintf(stderr, "failed to send STRESS C\n");
+ return 1;
+ }
+
+ memcpy(buff+plen, "STRESS D", 8);
+ if (zsend(s, buff, plen+8, 1)) {
+ fprintf(stderr, "failed to send STRESS D\n");
+ return 1;
+ }
+
+ memcpy(buff+plen, "STRESS E", 8);
+ if (zsend(s, buff, plen+8, 0)) {
+ fprintf(stderr, "failed to send STRESS E\n");
+ return 1;
+ }
+
+ return 0;
+}
+
+
+int main(int argc, char **argv)
+{
+ if (argc != 1) return 1;
+
+ void *context;
+ context = zmq_init(1);
+ if (!context) return 2;
+
+ void *socket;
+ socket = zmq_socket(context, ZMQ_XREQ);
+ if (!socket) return 3;
+
+ int ret = zmq_connect(socket, "tcp://localhost:5555");
+ assert(0 == ret);
+
+ int pid = (int)getpid();
+ int i;
+ for (i=20; i--;) {
+ char id_prefix[21];
+ snprintf(id_prefix, 21, "%i.%i.", pid, i);
+ if (stress(socket, id_prefix)) {
+ fprintf(stderr, "failed to send stress chunk %i\n", i);
+ assert(0 == zmq_close(socket));
+ assert(0 == zmq_term(context));
+ return 5;
+ }
+ }
+
+ assert(0 == zmq_close(socket));
+ assert(0 == zmq_term(context));
+
+ return 0;
+}
1  199/README
@@ -0,0 +1 @@
+Run the server, then 'while ./stress; do; done'
8 README
@@ -0,0 +1,8 @@
+This is the git repo for issue test cases
+
+Each issue has a subdirectory named 'nnn' where that is the issue number.
+
+Within that subdirectory, there should directories 2-1 and 3-0 for version-specific test cases.
+
+Please add a README to each issue indicating how to run the tests and what results are expected.
+
Please sign in to comment.
Something went wrong with that request. Please try again.