From 486a4471d860bc4e1f733fb7b4dda6e28d1ed8bc Mon Sep 17 00:00:00 2001 From: Martin Lucina Date: Thu, 16 Feb 2012 10:01:53 +0900 Subject: Fix pgm_receiver.cpp: zmq_assert (pending_bytes == 0) (LIBZMQ-205) This patch fixes the problem described in LIBZMQ-205. The assertion itself is probably caused by previously queued POLLIN events arriving after POLLIN has been disabled on the socket. The following additional bugs have been fixed as part of debugging this problem: - pgm_receiver_t does not flush messages written to the session in all cases which can lead to a stalled reader. Add calls to session->flush () in the appropriate places. - ensure to restart polling when a pending message is flushed in activate_in (). Signed-off-by: Martin Lucina --- src/pgm_receiver.cpp | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/pgm_receiver.cpp b/src/pgm_receiver.cpp index e87286d..ddf46b9 100644 --- a/src/pgm_receiver.cpp +++ b/src/pgm_receiver.cpp @@ -1,6 +1,7 @@ /* Copyright (c) 2009-2012 250bpm s.r.o. Copyright (c) 2007-2009 iMatix Corporation + Copyright (c) 2012 Lucina & Associates Copyright (c) 2010-2011 Miru Limited Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file @@ -117,8 +118,15 @@ void xs::pgm_receiver_t::activate_in () // processed the whole buffer but failed to write // the last message into the pipe. if (pending_bytes == 0) { - if (mru_decoder != NULL) + if (mru_decoder != NULL) { mru_decoder->process_buffer (NULL, 0); + session->flush (); + } + + // Resume polling. + set_pollin (pipe_handle); + set_pollin (socket_handle); + return; } @@ -128,6 +136,7 @@ void xs::pgm_receiver_t::activate_in () // Ask the decoder to process remaining data. size_t n = mru_decoder->process_buffer (pending_ptr, pending_bytes); pending_bytes -= n; + session->flush (); if (pending_bytes > 0) return; @@ -145,7 +154,8 @@ void xs::pgm_receiver_t::in_event () unsigned char *data = NULL; const pgm_tsi_t *tsi = NULL; - xs_assert (pending_bytes == 0); + if (pending_bytes > 0) + return; if (has_rx_timer) { cancel_timer (rx_timer_id); -- cgit v1.2.3