On Thu, 2011-12-29 at 14:41 +0100, Gergely Nagy wrote:

There is a race condition between afmongodb_dd_queue() and afmongodb_worker_insert(), that can trigger an assertion when the rate of incoming messages is very high.

The scenario that triggers this is like this:

* We receive a message, put it in our internal queue, and notify the worker thread. * The worker thread resets the queue's parallel_push. * We receive another message, notice that the queue is empty, so we insert it, and set parallel_push at the same time. * The worker thread tries to pop the queue's head, but that triggers an assert, because parallel_push is set.

The root cause of the problem is that afmongodb_dd_queue() modifies the queue without locking the queue_mutex. This patch corrects that problem.

Reported-By: Costa Farber <costaf@wix.com> Signed-off-by: Gergely Nagy <algernon@balabit.hu> --- modules/afmongodb/afmongodb.c | 2 ++ 1 files changed, 2 insertions(+), 0 deletions(-)

diff --git a/modules/afmongodb/afmongodb.c b/modules/afmongodb/afmongodb.c index cbd0d16..a88aeaa 100644 --- a/modules/afmongodb/afmongodb.c +++ b/modules/afmongodb/afmongodb.c @@ -530,7 +530,9 @@ afmongodb_dd_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_optio g_mutex_lock(self->suspend_mutex); if (queue_was_empty && !self->writer_thread_suspended) { + g_mutex_lock(self->queue_mutex); log_queue_set_parallel_push(self->queue, 1, afmongodb_dd_queue_notify, self, NULL); + g_mutex_unlock(self->queue_mutex); } g_mutex_unlock(self->suspend_mutex); }

Thanks Gergely, applied to 3.3. -- Bazsi