From e3f85ddc8ea565bc81b538269298a9f97b055c8c Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Sun, 23 Mar 2014 14:10:45 +0100 Subject: Fix livelock in synchronous ZeroMQ scenario --- src/InputZeroMQReader.cpp | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) (limited to 'src/InputZeroMQReader.cpp') diff --git a/src/InputZeroMQReader.cpp b/src/InputZeroMQReader.cpp index afb6b7d..483429b 100644 --- a/src/InputZeroMQReader.cpp +++ b/src/InputZeroMQReader.cpp @@ -3,7 +3,7 @@ Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyrigth (C) 2013 + Copyrigth (C) 2013, 2014 Matthias P. Braendli, matthias.braendli@mpb.li */ /* @@ -117,12 +117,15 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata) } else { + workerdata->in_messages->notify(); + if (!buffer_full) { - workerdata->in_messages->notify(); fprintf(stderr, "ZeroMQ buffer overfull !\n"); buffer_full = true; } + + queue_size = workerdata->in_messages->size(); } if (queue_size < 5) { @@ -132,9 +135,11 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata) } } catch (zmq::error_t& err) { - printf("ZeroMQ error in RecvProcess: '%s'\n", err.what()); + fprintf(stderr, "ZeroMQ error in RecvProcess: '%s'\n", err.what()); } + fprintf(stderr, "ZeroMQ input worker terminated\n"); + subscriber.close(); } -- cgit v1.2.3