feat: better compression bandwidth

.gitignore

@@ -65,5 +65,10 @@ tests/test-io
 tests/test-misc
 tests/test-parse
 tests/test-private
+tests/test-zpool-cull
+tests/test-zpool-mt
+tests/test-zpool-order
+tests/testzmq.json
+tests/testzmq.sock
 tests/*/*.out
 tests/*/test.sh

Makefile.am

@@ -216,6 +216,7 @@ LIBNMSG_LIB_MODULES = \
 	nmsg/msgmodset.c \
 	nmsg/nmsg.c \
 	nmsg/output.c \
+	nmsg/output_async.c \
 	nmsg/output_json.c \
 	nmsg/output_nmsg.c \
 	nmsg/output_pres.c \
@@ -244,6 +245,7 @@ LIBNMSG_LIB_MODULES = \
 nmsg_libnmsg_la_SOURCES = \
 	libmy/crc32c.h \
 	libmy/list.h \
+	libmy/my_cpu.h \
 	libmy/my_time.h \
 	libmy/my_rate.h \
 	libmy/tree.h \
@@ -419,6 +421,7 @@ src_nmsgtool_SOURCES = \
 	libmy/argv.c \
 	libmy/argv.h \
 	libmy/argv_loc.h \
+	libmy/my_cpu.h \
 	src/daemon.c \
 	src/getsock.c \
 	src/io.c \
@@ -436,6 +439,11 @@ src_nmsgtool_SOURCES = \
 ##
 #
 
+# Tests that reach private symbols link the objects: libnmsg.la exports
+# nmsg_* only, and the dlopened base msgmod resolves its own from the
+# executable, which is what -rdynamic is for.
+PRIVATE_TEST_MODULES = $(LIBNMSG_LIB_MODULES:.c=.o)
+
 TESTS_ENVIRONMENT = NMSG_MSGMOD_DIR=$(abs_top_builddir)/nmsg/base/.libs
 TESTS_ENVIRONMENT += abs_top_builddir='$(abs_top_builddir)' abs_top_srcdir='$(abs_top_srcdir)'
 
@@ -537,7 +545,6 @@ TESTS += tests/test-private
 check_PROGRAMS += tests/test-private
 tests_test_private_LDFLAGS = -rdynamic
 tests_test_private_CPPFLAGS = -DSRCDIR="\"$(abs_srcdir)\"" $(AM_CPPFLAGS)
-PRIVATE_TEST_MODULES = $(LIBNMSG_LIB_MODULES:.c=.o)
 tests_test_private_LDADD = \
 	$(PRIVATE_TEST_MODULES) \
 	nmsg/nmsg.pb-c.o \
@@ -561,6 +568,33 @@ check_PROGRAMS += tests/test-nmsg_output_set_rate
 tests_test_nmsg_output_set_rate_SOURCES = tests/test-nmsg_output_set_rate.c
 tests_test_nmsg_output_set_rate_LDADD = nmsg/libnmsg.la
 
+TESTS += tests/test-zpool-order
+check_PROGRAMS += tests/test-zpool-order
+tests_test_zpool_order_LDFLAGS = -rdynamic
+tests_test_zpool_order_LDADD = \
+	$(PRIVATE_TEST_MODULES) \
+	nmsg/nmsg.pb-c.o \
+	$(LIBNMSG_LIB_DEPS)
+tests_test_zpool_order_SOURCES = tests/test-zpool-order.c
+
+TESTS += tests/test-zpool-mt
+check_PROGRAMS += tests/test-zpool-mt
+tests_test_zpool_mt_LDFLAGS = -rdynamic
+tests_test_zpool_mt_LDADD = \
+	$(PRIVATE_TEST_MODULES) \
+	nmsg/nmsg.pb-c.o \
+	$(LIBNMSG_LIB_DEPS)
+tests_test_zpool_mt_SOURCES = tests/test-zpool-mt.c
+
+TESTS += tests/test-zpool-cull
+check_PROGRAMS += tests/test-zpool-cull
+tests_test_zpool_cull_LDFLAGS = -rdynamic
+tests_test_zpool_cull_LDADD = \
+	$(PRIVATE_TEST_MODULES) \
+	nmsg/nmsg.pb-c.o \
+	$(LIBNMSG_LIB_DEPS)
+tests_test_zpool_cull_SOURCES = tests/test-zpool-cull.c
+
 DISTCLEANFILES += tests/group-operator-source-tests/test*.out
 DISTCLEANFILES += tests/nmsg-dns-tests/test*.out
 DISTCLEANFILES += tests/nmsg-dnsobs-tests/test*.out

README.md

@@ -60,9 +60,6 @@ to the `configure` script.
 Support for `librdkafka` can be disabled by passing the `--without-librdkafka` parameter
 to the `configure` script.
 
-Support for `json-c` can be disabled by passing the `--without-json-c` parameter
-to the `configure` script.
-
 The documentation for the `libnmsg` API is located in the `doc/doxygen/html`
 directory. To rebuild the API documentation, run `make html`. This requires
 Doxygen to be installed.

configure.ac

@@ -96,6 +96,9 @@ AC_CHECK_FUNCS([clock_gettime])
 AC_SEARCH_LIBS([clock_nanosleep], [rt])
 AC_CHECK_FUNCS([clock_nanosleep])
 
+AC_SEARCH_LIBS([pthread_condattr_setclock], [pthread])
+AC_CHECK_FUNCS([pthread_condattr_setclock])
+
 AC_SEARCH_LIBS([dlopen], [dl])
 AC_CHECK_FUNCS([dlopen])
 
@@ -168,14 +171,7 @@ else
     use_libzmq="false"
 fi
 
-AC_ARG_WITH([json-c], AS_HELP_STRING([--without-json-c], [Disable json-c support]))
-if test "x$with_json_c" != "xno"; then
-    PKG_CHECK_MODULES([json_c], [json-c >= 0.13.0])
-    AC_DEFINE([HAVE_JSON_C], [1], [Define to 1 if json-c support is enabled.])
-    use_json_c="true"
-else
-    use_json_c="false"
-fi
+PKG_CHECK_MODULES([json_c], [json-c >= 0.13.0])
 
 AC_CHECK_HEADER([zlib.h], [], [ AC_MSG_ERROR([required header file not found]) ])
 AC_CHECK_LIB([z], [deflate], [], [ AC_MSG_ERROR([required library not found]) ])
@@ -270,7 +266,6 @@ AC_MSG_RESULT([
         bigendian:              ${ac_cv_c_bigendian}
         libzmq support:         ${use_libzmq}
         librdkafka support:     ${use_librdkafka}
-        json-c support:         ${use_json_c}
 
         building html docs:     ${DOC_HTML_MSG}
         building manpage docs:  ${DOC_MAN_MSG}

debian/libnmsg8.symbols

@@ -30,6 +30,7 @@ libnmsg.so.8 libnmsg8 #MINVER#
  nmsg_input_open_json@Base 0.10.0
  nmsg_input_open_kafka_endpoint@Base 1.2.0
  nmsg_input_open_kafka_json@Base 1.2.0
+ nmsg_input_open_kafka_payload@Base 1.4.0
  nmsg_input_open_null@Base 0.7.0
  nmsg_input_open_pcap@Base 0.5.0
  nmsg_input_open_pres@Base 0.5.0
@@ -138,6 +139,7 @@ libnmsg.so.8 libnmsg8 #MINVER#
  nmsg_output_open_sock@Base 0.5.0
  nmsg_output_open_zmq@Base 0.14.0
  nmsg_output_open_zmq_endpoint@Base 0.14.0
+ nmsg_output_open_kafka_payload@Base 1.4.0
  nmsg_output_set_buffered@Base 0.5.0
  nmsg_output_set_endline@Base 0.5.0
  nmsg_output_set_filter_msgtype@Base 0.5.0
@@ -146,6 +148,8 @@ libnmsg.so.8 libnmsg8 #MINVER#
  nmsg_output_set_operator@Base 0.5.0
  nmsg_output_set_rate@Base 0.5.0
  nmsg_output_set_source@Base 0.5.0
+ nmsg_output_set_zlib_cull@Base 1.4.0
+ nmsg_output_set_zlib_workers@Base 1.4.0
  nmsg_output_set_zlibout@Base 0.5.0
  nmsg_output_write@Base 0.11.1
  nmsg_pcap_filter@Base 0.6.5

doc/docbook/nmsgtool.1

@@ -2,12 +2,12 @@
 .\"     Title: nmsgtool
 .\"    Author: [FIXME: author] [see http://www.docbook.org/tdg5/en/html/author]
 .\" Generator: DocBook XSL Stylesheets v1.79.2 <http://docbook.sf.net/>
-.\"      Date: 10/02/2024
+.\"      Date: 04/22/2026
 .\"    Manual:  
 .\"    Source:    
 .\"  Language: English
 .\"
-.TH "NMSGTOOL" "1" "10/02/2024" "" ""
+.TH "NMSGTOOL" "1" "04/22/2026" "" ""
 .\" -----------------------------------------------------------------
 .\" * Define some portability stuff
 .\" -----------------------------------------------------------------
@@ -118,6 +118,8 @@ Writing ASCII presentation form data to a file\&. See the
 option\&.
 .RE
 .PP
+There are more input and output types supported, depending upon the installation\&. See below in the options list\&.
+.PP
 Reading or writing data in a non\-NMSG format requires the use of an external module (called an "nmsgpb module") to convert to or from NMSG format\&.
 \fBnmsgtool\fR
 selects an nmsgpb module based on a vendor ID and message type\&. For input data, these fields must be set with the
@@ -142,15 +144,15 @@ Increment debugging level\&.
 \fB\-dd\fR
 is verbose and
 \fB\-dddd\fR
-is very verbose\&.
-If the
+is very verbose\&. If the
 \fBNMSG_KAFKA_LOG_LEVEL\fR
-environment variable is set and a Kafka input/output is used, Kafka will log non-errors at the specified logging level\&.
+environment variable is set and a Kafka input/output is used, Kafka will log non\-errors at the specified logging level\&.
 .RE
 .PP
-\fB\-v\fR \fIversion\fR
+\fB\-v\fR, \fB\-\-version\fR
 .RS 4
-Print the version number of \fBnmsgtool\fR.
+Print
+\fBnmsgtool\fR\*(Aqs version number\&.
 .RE
 .PP
 \fB\-V\fR \fIvendor\fR, \fB\-\-vendor\fR \fIvendor\fR
@@ -167,24 +169,34 @@ Set the message type field of generated NMSG payloads to the message type identi
 .PP
 \fB\-B\fR \fIbyterate\fR, \fB\-\-byterate\fR \fIbyterate\fR
 .RS 4
-Set the maximum bytes-per-second that libnmsg will process from file inputs.
+Set the maximum bytes\-per\-second that libnmsg will process from file inputs\&.
 .RE
 .PP
-\fB\-D\fR \fIdaemon\fR
+\fB\-D\fR, \fB\-\-daemon\fR
 .RS 4
-Fork \fBnmsgtool\fR into the background as a daemon process.
+Fork
+\fBnmsgtool\fR
+into the background as a daemon process\&.
 .RE
 .PP
 \fB\-P\fR \fIpidfile\fR, \fB\-\-pidfile\fR \fIpidfile\fR
 .RS 4
-Write \fBnmsgtool\fR's process ID into a file identified by \fIpidfile\fR.  When 
-\fBnmsgtool\fR exits, this file's contents will be erased.
+Write
+\fBnmsgtool\fR\*(Aqs process ID into a file identified by
+\fIpidfile\fR\&. When
+\fB nmsgtool\fR
+exits, this file\*(Aqs contents will be erased\&.
 .RE
 .PP
 \fB\-U\fR \fIusername\fR, \fB\-\-username\fR \fIusername\fR
 .RS 4
-Attempt to drop root privileges and run \fBnmsgtool\fR as user \fIusername\fR.  
-If the current user is \fIusername\fR, \fBnmsgtool\fR will exit.
+Attempt to drop root privileges and run
+\fBnmsgtool \fR
+as user
+\fIusername\fR\&. If the current user is
+\fIusername\fR,
+\fB nmsgtool\fR
+will exit\&.
 .RE
 .PP
 \fB\-e\fR \fIendline\fR, \fB\-\-endline\fR \fIendline\fR
@@ -264,9 +276,48 @@ environment variable\&.
 Read NMSG payloads in either binary or JSON format from a Kafka endpoint\&. The address
 \fIkafka\fR
 has format
-\fBproto:topic[#partition|%group_id]@broker[:port][,offset]\fR\&. Either a partition number or a consumer group ID may be optionally supplied\&. Also optional is an offset consisting of either a numerical value or the string \*(Aqoldest\*(Aq or \*(Aqnewest\*(Aq in order to start retrieval at the oldest/newest messages in the Kafka topic\&. An example of a possible
+\fBproto:topic[#partition|%group_id]@broker[:port][,offset]\fR\&. Either a partition number or a consumer group ID may be optionally supplied\&. Also optional is an offset consisting of either a numerical value or the string \*(Aqoldest\*(Aq or \*(Aqnewest\*(Aq in order to start retrieval at the oldest/newest messages in the Kafka topic\&.
+.sp
+There are three
 \fIkafka\fR
-endpoint is "nmsg:ch202#0@kafka\&.example\&.com:9092,3000" to indicate that nmsgtool shall read nmsg containers from topic "ch202" on partition 0 at offset 3000 from the Kafka broker at kafka\&.example\&.com, port 9092\&. Configuration for Kafka can be supplied through the
+endpoint proto types defined:
+.sp
+.RS 4
+.ie n \{\
+\h'-04'\(bu\h'+03'\c
+.\}
+.el \{\
+.sp -1
+.IP \(bu 2.3
+.\}
+json: to read json payloads from each kafka record
+.RE
+.sp
+.RS 4
+.ie n \{\
+\h'-04'\(bu\h'+03'\c
+.\}
+.el \{\
+.sp -1
+.IP \(bu 2.3
+.\}
+nmsg: to read a full binary NMSG container from each kafka record
+.RE
+.sp
+.RS 4
+.ie n \{\
+\h'-04'\(bu\h'+03'\c
+.\}
+.el \{\
+.sp -1
+.IP \(bu 2.3
+.\}
+nmsgp: to read a single binary NMSG payload from each kafka record
+.RE
+.sp
+Example: "nmsg:ch202#0@kafka\&.example\&.com:9092,3000" to indicate that nmsgtool shall read nmsg containers from topic "ch202" on partition 0 at offset 3000 from the Kafka broker at kafka\&.example\&.com, port 9092\&.
+.sp
+Configuration for Kafka can be supplied through the
 \fBNMSG_KAFKA_CONFIG\fR
 environment variable\&. This variable accepts either key=value pairs separated by colons, or a path to a configuration file in INI format\&.
 .RE
@@ -355,17 +406,6 @@ environment variable\&.
 Read NMSG payloads from a file\&.
 .RE
 .PP
-\fB\-f\fR \fIfile\fR, \fB\-\-readpres\fR \fIfile\fR
-.RS 4
-Read presentation format data from a file and convert to NMSG payloads\&. This option is dependent on the
-\fB\-V\fR
-and
-\fB\-T\fR
-options being set in order to select a specific nmsgpb module to perform presentation format to NMSG payload conversion\&. Not all nmsgpb modules necessarily support this conversion method, in which case
-\fBnmsgtool\fR
-will print a "function not implemented" message\&.
-.RE
-.PP
 \fB\-j\fR \fIfile\fR, \fB\-\-readjson\fR \fIfile\fR
 .RS 4
 Read JSON format data from a file\&. See documentation for
@@ -514,9 +554,50 @@ Write NMSG payloads to a file\&.
 Write NMSG payloads in either binary or JSON format to a Kafka endpoint\&. The address
 \fIkafka\fR
 has format
-\fBproto:topic[#partition|%group_id]@broker[:port]\fR\&. Either a partition number or a consumer group ID may be optionally supplied\&. An example of a possible
+\fBproto:topic[#partition|%group_id]@broker[:port]\fR\&. Either a partition number or a consumer group ID may be optionally supplied\&.
+.sp
+There are three
 \fIkafka\fR
-endpoint is "nmsg:ch202#0@kafka\&.example\&.com:9092" to indicate that nmsgtool shall write nmsg containers to topic "ch202" on partition 0 to Kafka broker kafka\&.example\&.com, port 9092\&. Note that nmsgtool ignores offsets for Kafka producers\&. Configuration for Kafka can be supplied through the
+endpoint proto types defined:
+.sp
+.RS 4
+.ie n \{\
+\h'-04'\(bu\h'+03'\c
+.\}
+.el \{\
+.sp -1
+.IP \(bu 2.3
+.\}
+json: to write JSON payloads to Kafka\&.
+.RE
+.sp
+.RS 4
+.ie n \{\
+\h'-04'\(bu\h'+03'\c
+.\}
+.el \{\
+.sp -1
+.IP \(bu 2.3
+.\}
+nmsg: to write a full binary NMSG container in each kafka record\&.
+.RE
+.sp
+.RS 4
+.ie n \{\
+\h'-04'\(bu\h'+03'\c
+.\}
+.el \{\
+.sp -1
+.IP \(bu 2.3
+.\}
+nmsgp: to write a single binary NMSG payload in each kafka record\&.
+.RE
+.sp
+Example: "nmsg:ch202#0@kafka\&.example\&.com:9092" to indicate that nmsgtool shall write nmsg containers to topic "ch202" on partition 0 to Kafka broker kafka\&.example\&.com, port 9092\&.
+.sp
+Note that nmsgtool ignores offsets for Kafka producers\&.
+.sp
+Configuration for Kafka can be supplied through the
 \fBNMSG_KAFKA_CONFIG\fR
 environment variable\&. This variable accepts either key=value pairs separated by colons, or a path to a configuration file in INI format\&.
 .RE
@@ -800,7 +881,7 @@ To read NMSG payloads from a file and write them to a ZeroMQ "PUSH" socket over
 .PP
 This attempts to connect to a TCP reader on 127\&.0\&.0\&.1:5555, such as the nmsgtool command in the previous example\&.
 .PP
-To read NMSG payloads from an SIE channel named "ch222" and write them to stdout while writing IO stats every 3 seconds to a file name "ch222.stats":
+To read NMSG payloads from an SIE channel named "ch222" and write them to stdout while writing IO stats every 3 seconds to a file name "ch222\&.stats":
 .sp
 .if n \{\
 .RS 4