Bug #64836 for ZeroMQ: Message Corruption/Loss

I've coded a test using the ZeroMQ::Raw API and sadly the same behaviour is present. I'll look at a C version shortly to determine if it's ZeroMQ or the Perl binding.

use strict; use Test::More; use Test::Requires qw( Test::TCP ); use Data::Dumper; BEGIN { use_ok "ZeroMQ::Raw"; use_ok "ZeroMQ::Constants", ":all"; } my $port = empty_port(); test_tcp( client => sub { my $port = shift; my $ctxt = zmq_init(); my $sock = zmq_socket($ctxt, ZMQ_SUB); sleep(1); zmq_connect($sock,"tcp://127.0.0.1:$port" ); zmq_setsockopt($sock, ZMQ_SUBSCRIBE, ''); my $cnt=0; while($cnt<1000) { # expect to receive numbers 1..1000... my $rawmsg = undef; $rawmsg = zmq_recv($sock); my $msg = zmq_msg_data($rawmsg); is($msg, $cnt++, "Error -- messages not received in sequence or corrupt"); } note "OK"; }, server => sub { my $port = shift; my $ctxt = zmq_init(); my $sock = zmq_socket($ctxt, ZMQ_PUB); note "Server Binding to port $port\n"; zmq_bind($sock, "tcp://127.0.0.1:$port"); note "Waiting on client to bind..."; sleep 5; note "Server sending ordered data... (numbers 1..1000)"; for (my $c=0; $c<1000; $c++) { my $msg = zmq_msg_init_data($c); zmq_send($sock, $msg); zmq_msg_close($msg); } note "OK"; } ); done_testing;

calling it a day. I have attached a patch with the tests developed so far -- once I work out the basics for github I'll submit updates there to make life simpler, I get consistent results with these tests: t/999_messagecorruptions.t ........... Failed 100/102 subtests t/999_messagecorruptions_lowlevel.t .. ok The tests _should_ be identical. Note that increasing the number of messages in these tests will induce errors in the low level code -- nowhere near as badly as the higher level binding. Removing the 'threads->yield' guarantees a 100% failure rate in both tests -- I'm afraid I need to spend more time reviewing the code to understand why this would make any difference at all.

commit ded0374098e3014b2fe73e9c7cfbec4e618e6243 Author: Jason Ball <jason.ball@tibra.com> Date: Mon Jan 17 22:55:14 2011 +1100 Test cases for low level corruptions diff --git a/t/999_messagecorruptions.t b/t/999_messagecorruptions.t new file mode 100644 index 0000000..5a00340 --- /dev/null +++ b/t/999_messagecorruptions.t @@ -0,0 +1,57 @@ +use strict; +use Test::More; +use Test::Requires qw( Test::TCP AnyEvent ); +use ZeroMQ qw(ZMQ_PUB ZMQ_SUB ZMQ_SNDMORE); +use Time::HiRes qw(usleep); + +BEGIN { + use_ok "ZeroMQ"; + use_ok "ZeroMQ::Constants", ":all"; +} + + +my $port = empty_port(); + +test_tcp( + client => sub { + my $port = shift; + my $ctxt = ZeroMQ::Context->new(); + my $sock = $ctxt->socket(ZMQ_SUB); + + sleep(1); + $sock->connect("tcp://127.0.0.1:$port" ); + $sock->setsockopt(ZMQ_SUBSCRIBE, ''); + my $data = join '.', time(), $$, rand, {}; + + my $cnt=0; + + while($cnt<1000) { # expect to receive numbers 1..1000... + my $msg = undef; + $msg = $sock->recv(); + is($msg->data, $cnt++, "Error -- messages not received in sequence or corrupt"); + } + + note "OK"; + }, + server => sub { + my $port = shift; + my $ctxt = ZeroMQ::Context->new(); + my $sock = $ctxt->socket(ZMQ_PUB); + + note "Server Binding to port $port\n"; + $sock->bind("tcp://127.0.0.1:$port"); + + note "Waiting on client to bind..."; + sleep 5; + + note "Server sending ordered data... (numbers 1..1000)"; + for (my $c=0; $c<1000; $c++) { + $sock->send($c, ZMQ_SNDMORE); + #usleep(1000); + } + $sock->send("end"); # end of data stream... + note "OK"; + } +); + +done_testing; diff --git a/t/999_messagecorruptions_lowlevel.t b/t/999_messagecorruptions_lowlevel.t new file mode 100644 index 0000000..caf2bd9 --- /dev/null +++ b/t/999_messagecorruptions_lowlevel.t @@ -0,0 +1,56 @@ +use strict; +use Test::More; +use Test::Requires qw( Test::TCP ); +use Data::Dumper; + +BEGIN { + use_ok "ZeroMQ::Raw"; + use_ok "ZeroMQ::Constants", ":all"; +} + + +my $port = empty_port(); + +test_tcp( + client => sub { + my $port = shift; + my $ctxt = zmq_init(); + my $sock = zmq_socket($ctxt, ZMQ_SUB); + + sleep(1); + zmq_connect($sock,"tcp://127.0.0.1:$port" ); + zmq_setsockopt($sock, ZMQ_SUBSCRIBE, ''); + + my $cnt=0; + + while($cnt<1000) { # expect to receive numbers 1..1000... + my $rawmsg = undef; + $rawmsg = zmq_recv($sock); + my $msg = zmq_msg_data($rawmsg); + is($msg, $cnt++, "Error -- messages not received in sequence or corrupt"); + } + + note "OK"; + }, + server => sub { + my $port = shift; + my $ctxt = zmq_init(); + my $sock = zmq_socket($ctxt, ZMQ_PUB); + + note "Server Binding to port $port\n"; + zmq_bind($sock, "tcp://127.0.0.1:$port"); + + note "Waiting on client to bind..."; + sleep 5; + + note "Server sending ordered data... (numbers 1..1000)"; + for (my $c=0; $c<1000; $c++) { + my $msg = zmq_msg_init_data($c); + zmq_send($sock, $msg); + zmq_msg_close($msg); + } + note "OK"; + } +); + +done_testing;