generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-07-03 17:17:14 +02:00
Code Issues Projects Releases Wiki Activity

Fix pgm_receiver.cpp: zmq_assert (pending_bytes == 0) (LIBZMQ-205)

Browse Source 
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 <martin@lucina.net>
This commit is contained in:
Martin Lucina 2012-01-04 11:48:41 +01:00
parent b3fbe0113d
commit c34a144365
1 changed files with 11 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/pgm_receiver.cpp
View File

@ -117,8 +117,15 @@ void zmq::pgm_receiver_t::activate_in ()
// processed the whole buffer but failed to write // processed the whole buffer but failed to write
// the last message into the pipe. // the last message into the pipe.
if (pending_bytes == 0) { if (pending_bytes == 0) {
if (mru_decoder != NULL) if (mru_decoder != NULL) {
mru_decoder->process_buffer (NULL, 0); mru_decoder->process_buffer (NULL, 0);
session->flush ();
}
// Resume polling.
set_pollin (pipe_handle);
set_pollin (socket_handle);
return; return;
} }
@ -128,6 +135,7 @@ void zmq::pgm_receiver_t::activate_in ()
// Ask the decoder to process remaining data. // Ask the decoder to process remaining data.
size_t n = mru_decoder->process_buffer (pending_ptr, pending_bytes); size_t n = mru_decoder->process_buffer (pending_ptr, pending_bytes);
pending_bytes -= n; pending_bytes -= n;
session->flush ();
if (pending_bytes > 0) if (pending_bytes > 0)
return; return;
@ -145,7 +153,8 @@ void zmq::pgm_receiver_t::in_event ()
unsigned char *data = NULL; unsigned char *data = NULL;
const pgm_tsi_t *tsi = NULL; const pgm_tsi_t *tsi = NULL;
zmq_assert (pending_bytes == 0); if (pending_bytes > 0)
return;
if (has_rx_timer) { if (has_rx_timer) {
cancel_timer (rx_timer_id); cancel_timer (rx_timer_id);