From 53a77198768a1b300aa6265a0f3c46c5c919e3e3 Mon Sep 17 00:00:00 2001 From: kadealicious Date: Wed, 2 Sep 2026 12:48:14 -0700 Subject: [PATCH 1/7] Release 1.4.0 --- debian/changelog | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/debian/changelog b/debian/changelog index 46e1509d..7fda6442 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,17 +1,30 @@ +<<<<<<< HEAD nmsg (1.4.0) debian-fsi; urgency=medium * Removed nmsgtool support for presentation inputs. * Properly handle fragmented containers in zmq/kafka inputs. +======= +nmsg (1.4.0-1) debian-fsi; urgency=medium + + * Handle fragmented containers in zmq/kafka inputs. +>>>>>>> 671744db (Release 1.4.0) * Support single nmsg payload kafka in/out sources. * Clean-up temp files after failed start using -w. * Improved warning for mismatched CRC. * Require json-c build dependency. +<<<<<<< HEAD * Fix various small memory leaks and mismanagement issues. * Fix strict aliasing violations. * Removal of various unused header files. * Various documentation fixes. -- DomainTools LLC Wed, 02 Sep 2026 12:41:40 -0700 +======= + * Various other small bugfixes. + * Various documentation fixes. + + -- Farsight Security Inc Wed, 02 Sep 2026 12:41:40 -0700 +>>>>>>> 671744db (Release 1.4.0) nmsg (1.3.3-1) debian-fsi; urgency=medium From f9efd46b77bb9da92171f1e0c42af31d3c12c5ed Mon Sep 17 00:00:00 2001 From: kadealicious Date: Wed, 2 Sep 2026 14:03:02 -0700 Subject: [PATCH 2/7] Adjust changelog publisher name --- debian/changelog | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/debian/changelog b/debian/changelog index 7fda6442..46e1509d 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,30 +1,17 @@ -<<<<<<< HEAD nmsg (1.4.0) debian-fsi; urgency=medium * Removed nmsgtool support for presentation inputs. * Properly handle fragmented containers in zmq/kafka inputs. -======= -nmsg (1.4.0-1) debian-fsi; urgency=medium - - * Handle fragmented containers in zmq/kafka inputs. ->>>>>>> 671744db (Release 1.4.0) * Support single nmsg payload kafka in/out sources. * Clean-up temp files after failed start using -w. * Improved warning for mismatched CRC. * Require json-c build dependency. -<<<<<<< HEAD * Fix various small memory leaks and mismanagement issues. * Fix strict aliasing violations. * Removal of various unused header files. * Various documentation fixes. -- DomainTools LLC Wed, 02 Sep 2026 12:41:40 -0700 -======= - * Various other small bugfixes. - * Various documentation fixes. - - -- Farsight Security Inc Wed, 02 Sep 2026 12:41:40 -0700 ->>>>>>> 671744db (Release 1.4.0) nmsg (1.3.3-1) debian-fsi; urgency=medium From cae581250d2693108fbe2da48f77f619cf71890e Mon Sep 17 00:00:00 2001 From: Juan Salazar <72028266+regalk13@users.noreply.github.com> Date: Thu, 5 Feb 2026 16:24:17 +0000 Subject: [PATCH 3/7] fix: --count exists without waiting for extra input message --- nmsg/io.c | 34 ++++++++++++---------------------- 1 file changed, 12 insertions(+), 22 deletions(-) diff --git a/nmsg/io.c b/nmsg/io.c index 9f13fb06..f29f3877 100644 --- a/nmsg/io.c +++ b/nmsg/io.c @@ -741,29 +741,24 @@ io_write(struct nmsg_io_thr *iothr, struct nmsg_io_output *io_output, nmsg_io_t io = iothr->io; nmsg_res res; - /* It's possible a set "count" has been reached. */ - check_close_event(iothr, io_output, 1); - - if (io->stop) { - reset_close_event(iothr, io_output); - nmsg_message_destroy(&msg); - return (nmsg_res_stop); - } res = nmsg_output_write(io_output->output, msg); if (io_output->output->type != nmsg_output_type_callback) nmsg_message_destroy(&msg); - /* - * Reset only after the write, in case another thread invokes - * check_close_event and makes changes to io_output in the meantime. - */ - reset_close_event(iothr, io_output); - if (res != nmsg_res_success) return (res); - + atomic_fetch_add_explicit(&io->io_count_nmsg_payload_out, 1, memory_order_relaxed); + + /* It's possible a set "count" has been reached. */ + check_close_event(iothr, io_output, 1); + reset_close_event(iothr, io_output); + + if (io->stop) { + // nmsg_message_destroy(&msg); + return (nmsg_res_stop); + } return (res); } @@ -797,6 +792,8 @@ check_close_event(struct nmsg_io_thr *iothr, struct nmsg_io_output *io_output, u if (io->count > 0 && io_output->count_next_close == 0) io_output->count_next_close = io->count; + io_output->count_nmsg_payload_out += count; + if (io->count > 0 && io_output->count_nmsg_payload_out == io_output->count_next_close) { @@ -856,13 +853,6 @@ check_close_event(struct nmsg_io_thr *iothr, struct nmsg_io_output *io_output, u } out: - /* - * This incr is implicitly locked IF it's used, and this counter is - * only used IF io->count > 0; that condition results in an acquired - * lock at the beginning of this function. - */ - io_output->count_nmsg_payload_out += count; - if (io->close_fp != NULL || io->count > 0) pthread_mutex_unlock(&io_output->lock); } From b5c863cea3b3ea4cfa3dc047584281422e004419 Mon Sep 17 00:00:00 2001 From: Juan Salazar <72028266+regalk13@users.noreply.github.com> Date: Tue, 16 Jun 2026 17:39:01 -0500 Subject: [PATCH 4/7] style: fix whitespace --- nmsg/io.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/nmsg/io.c b/nmsg/io.c index f29f3877..a445098b 100644 --- a/nmsg/io.c +++ b/nmsg/io.c @@ -748,9 +748,9 @@ io_write(struct nmsg_io_thr *iothr, struct nmsg_io_output *io_output, if (res != nmsg_res_success) return (res); - + atomic_fetch_add_explicit(&io->io_count_nmsg_payload_out, 1, memory_order_relaxed); - + /* It's possible a set "count" has been reached. */ check_close_event(iothr, io_output, 1); reset_close_event(iothr, io_output); From 7a7d206ea587f0d8716c4d87db056f9da7c91c9b Mon Sep 17 00:00:00 2001 From: Juan Salazar <72028266+regalk13@users.noreply.github.com> Date: Mon, 22 Jun 2026 10:57:19 -0500 Subject: [PATCH 5/7] clean destroy commented function --- nmsg/io.c | 1 - 1 file changed, 1 deletion(-) diff --git a/nmsg/io.c b/nmsg/io.c index a445098b..d8123b69 100644 --- a/nmsg/io.c +++ b/nmsg/io.c @@ -756,7 +756,6 @@ io_write(struct nmsg_io_thr *iothr, struct nmsg_io_output *io_output, reset_close_event(iothr, io_output); if (io->stop) { - // nmsg_message_destroy(&msg); return (nmsg_res_stop); } From cc0d917fd24ad47b85d5c0b68a4a4cde8bec71db Mon Sep 17 00:00:00 2001 From: Juan Salazar <72028266+regalk13@users.noreply.github.com> Date: Mon, 31 Aug 2026 10:41:39 -0500 Subject: [PATCH 6/7] test: multi-thread kicker test --- Makefile.am | 1 + tests/test-kicker-concurrent.sh | 22 ++++++++++++++++++++++ 2 files changed, 23 insertions(+) create mode 100644 tests/test-kicker-concurrent.sh diff --git a/Makefile.am b/Makefile.am index e12d51a0..3c3bddde 100644 --- a/Makefile.am +++ b/Makefile.am @@ -521,6 +521,7 @@ TESTS += tests/test-zmq-endpoints.sh TESTS += tests/test-socket-frag.sh TESTS += tests/test-msgmod.sh TESTS += tests/test-kicker.sh +TESTS += tests/test-kicker-concurrent.sh TESTS += tests/test-daemon.sh TESTS += tests/test-sample.sh diff --git a/tests/test-kicker-concurrent.sh b/tests/test-kicker-concurrent.sh new file mode 100644 index 00000000..cf688416 --- /dev/null +++ b/tests/test-kicker-concurrent.sh @@ -0,0 +1,22 @@ +#!/bin/sh + +nmsgtool=$abs_top_builddir/src/nmsgtool +sample=$abs_top_srcdir/tests/generic-tests/lorem.json +outdir=$abs_top_builddir/tests/test-kicker-concurrent/ +mkdir -p "$outdir"; cd "$outdir" + +for i in 1 2 3 4 5 6 7 8; do cp "$sample" "in$i.json"; done + +TSAN_OPTIONS="halt_on_error=1 exitcode=99" \ +"$nmsgtool" -ddd \ + -j in1.json -j in2.json -j in3.json -j in4.json \ + -j in5.json -j in6.json -j in7.json -j in8.json \ + -c 1 -k "echo" -w out > kicked.list 2>run.log +rc=$? + +[ $rc -ne 99 ] || { echo "FAIL: race condition "; cat run.log; exit 1; } +[ $rc -eq 0 ] || { echo "FAIL: nmsgtool exit $rc"; cat run.log; exit 1; } + +files=$(wc -l < kicked.list) +[ "$files" -ge 100 ] || { echo "FAIL: only $files rotations"; exit 1; } +echo PASS From 7133f574694e90725874e815223edae7fe1af31d Mon Sep 17 00:00:00 2001 From: Juan Salazar <72028266+regalk13@users.noreply.github.com> Date: Thu, 3 Sep 2026 15:02:07 -0500 Subject: [PATCH 7/7] fix candidate --- nmsg/io.c | 38 ++++++++++++++++++++++++++------------ 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/nmsg/io.c b/nmsg/io.c index d8123b69..c9ad40de 100644 --- a/nmsg/io.c +++ b/nmsg/io.c @@ -741,24 +741,30 @@ io_write(struct nmsg_io_thr *iothr, struct nmsg_io_output *io_output, nmsg_io_t io = iothr->io; nmsg_res res; + /* It's possible a set "count" has been reached. */ + check_close_event(iothr, io_output, 1); + + if (io->stop) { + reset_close_event(iothr, io_output); + nmsg_message_destroy(&msg); + return (nmsg_res_stop); + } res = nmsg_output_write(io_output->output, msg); if (io_output->output->type != nmsg_output_type_callback) nmsg_message_destroy(&msg); + /* + * Reset only after the write, in case another thread invokes + * check_close_event and makes changes to io_output in the meantime. + */ + reset_close_event(iothr, io_output); + if (res != nmsg_res_success) return (res); atomic_fetch_add_explicit(&io->io_count_nmsg_payload_out, 1, memory_order_relaxed); - /* It's possible a set "count" has been reached. */ - check_close_event(iothr, io_output, 1); - reset_close_event(iothr, io_output); - - if (io->stop) { - return (nmsg_res_stop); - } - return (res); } @@ -791,8 +797,6 @@ check_close_event(struct nmsg_io_thr *iothr, struct nmsg_io_output *io_output, u if (io->count > 0 && io_output->count_next_close == 0) io_output->count_next_close = io->count; - io_output->count_nmsg_payload_out += count; - if (io->count > 0 && io_output->count_nmsg_payload_out == io_output->count_next_close) { @@ -852,6 +856,13 @@ check_close_event(struct nmsg_io_thr *iothr, struct nmsg_io_output *io_output, u } out: + /* + * This incr is implicitly locked IF it's used, and this counter is + * only used IF io->count > 0; that condition results in an acquired + * lock at the beginning of this function. + */ + io_output->count_nmsg_payload_out += count; + if (io->close_fp != NULL || io->count > 0) pthread_mutex_unlock(&io_output->lock); } @@ -1042,22 +1053,25 @@ io_thr_input(void *user) { assert(msg != NULL); - io_input->count_nmsg_payload_in += 1; - if (iothr->filters != NULL) { res = io_run_filters(io, iothr->filters, &msg, &vres); if (res != nmsg_res_success) { nmsg_message_destroy(&msg); + io_input->count_nmsg_payload_in += 1; iothr->res = res; break; } if (vres == nmsg_filter_message_verdict_DROP) { nmsg_message_destroy(&msg); + io_input->count_nmsg_payload_in += 1; continue; } } + if (res != nmsg_res_stop) + io_input->count_nmsg_payload_in += 1; + if (io->output_mode == nmsg_io_output_mode_stripe) res = io_write(iothr, io_output, msg); else if (io->output_mode == nmsg_io_output_mode_mirror)