fix: --count exists without waiting for extra input message

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
 

nmsg/io.c

@@ -1053,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)

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