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

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])
 

debian/libnmsg8.symbols

@@ -148,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.docbook

@@ -656,6 +656,85 @@
         </listitem>
       </varlistentry>
 
+      <varlistentry>
+        <term><option>--zasync</option> <replaceable>n</replaceable></term>
+        <listitem>
+          <para>Compress written NMSG containers on
+          <replaceable>n</replaceable> separate threads rather than on
+          the thread that filled them. A thread that is compressing is
+          not reading its socket, and on a high volume channel that
+          pause is long enough for the kernel receive queue to
+          overflow.</para>
+
+          <para>Off by default. <replaceable>n</replaceable> is a
+          ceiling rather than a thread count: compressors start only as
+          load calls for them and are given back once idle, so an
+          output that never saturates never pays for them. See
+          <option>--zcull</option> and <option>--zmin</option>.
+          Compression is CPU-bound, so values above the cores available
+          to the process are rejected.</para>
+
+          <para>A value of <option>-1</option> chooses the ceiling: two
+          per input, divided across the file outputs but not under
+          <option>--mirror</option>, bounded by the cores the readers
+          leave spare, then never fewer than four nor more than the
+          core count. The cores counted are those in the process
+          affinity mask, which does not reflect a cgroup CPU quota.
+          Sizing it exactly does not matter, since a reader that finds
+          every compressor busy compresses the container itself. It
+          helps most where one input outruns a single core; a channel
+          spread over many ports already compresses on every reader
+          thread.</para>
+
+          <para>Compression itself still needs <option>-z</option>, and
+          a pool holds a serialized container per queued ticket, so it
+          can add tens of megabytes of buffering. Containers are always
+          written in the order they were filled; with a single input
+          the output is byte for byte identical to compressing inline,
+          whatever <replaceable>n</replaceable> is. File
+          (<option>-w</option>) outputs only, and cannot be combined
+          with <option>--unbuffered</option>.</para>
+        </listitem>
+      </varlistentry>
+
+      <varlistentry>
+        <term><option>--zcull</option> <replaceable>secs</replaceable></term>
+        <listitem>
+          <para>Give up a compressor thread that has been idle for
+          <replaceable>secs</replaceable> seconds, 300 by default. A
+          value of <option>0</option> never gives one up, leaving a
+          pool at the largest it ever needed to be.</para>
+
+          <para>Containers always go to the lowest-numbered free
+          compressor, so a pool that grew for a burst keeps the first
+          few busy and lets the rest fall quiet. A compressor only ever
+          leaves when it is holding nothing, and the pool grows again
+          on demand, so the output is unaffected either way. This
+          matters mainly for a long-running output that is not being
+          rotated: one closed by <option>-t</option> or
+          <option>-k</option> gives its whole pool back at every close
+          regardless.</para>
+        </listitem>
+      </varlistentry>
+
+      <varlistentry>
+        <term><option>--zmin</option> <replaceable>n</replaceable></term>
+        <listitem>
+          <para>Never let <option>--zcull</option> take an output
+          below <replaceable>n</replaceable> compressor threads, 1 by
+          default. A value of <option>0</option> lets the pool empty
+          completely.</para>
+
+          <para>A floor on what culling may take away, not a number of
+          threads to start: compressors are still started only on
+          demand, so an output that has never been busy is running none
+          of them whatever this is set to. A value above an explicit
+          <option>--zasync</option> is rejected; under <option>--zasync
+          -1</option> it is lowered to the ceiling chosen, which is
+          reported at <option>-dd</option>.</para>
+        </listitem>
+      </varlistentry>
+
       <varlistentry>
         <term><option>--mirror</option></term>
         <listitem>

libmy/my_cpu.h

@@ -0,0 +1,31 @@
+#ifndef MY_CPU_H
+#define MY_CPU_H
+
+#include <unistd.h>
+
+#ifdef __linux__
+# include <sched.h>
+#endif /* __linux__ */
+
+/* Cores this process may actually run on. Never less than one. */
+static inline long
+my_ncpu(void)
+{
+	long ncpu = -1;
+
+#ifdef __linux__
+	cpu_set_t set;
+
+	if (sched_getaffinity(0, sizeof(set), &set) == 0)
+		ncpu = CPU_COUNT(&set);
+#endif /* __linux__ */
+
+	if (ncpu < 1)
+		ncpu = sysconf(_SC_NPROCESSORS_ONLN);
+	if (ncpu < 1)
+		ncpu = 1;
+
+	return (ncpu);
+}
+
+#endif /* MY_CPU_H */

nmsg/output.c

@@ -280,12 +280,21 @@ nmsg_output_write(nmsg_output_t output, nmsg_message_t msg) {
 
 nmsg_res
 nmsg_output_close(nmsg_output_t *output) {
-	nmsg_res res;
+	nmsg_res res, async_res;
 
 	res = nmsg_res_success;
 	switch ((*output)->type) {
 	case nmsg_output_type_stream:
 		res = _output_nmsg_flush(*output);
+
+		/*
+		 * Before random, fd and the locks below, all of which the
+		 * compressor thread uses.
+		 */
+		async_res = _output_async_destroy(*output);
+		if (res == nmsg_res_success)
+			res = async_res;
+
 		if ((*output)->stream->random != NULL)
 			nmsg_random_destroy(&((*output)->stream->random));
 #ifdef HAVE_LIBRDKAFKA
@@ -307,6 +316,7 @@ nmsg_output_close(nmsg_output_t *output) {
 				close((*output)->stream->fd);
 		}
 		nmsg_container_destroy(&(*output)->stream->c);
+		pthread_cond_destroy(&(*output)->stream->c_drained);
 		pthread_mutex_destroy(&(*output)->stream->c_lock);
 		pthread_mutex_destroy(&(*output)->stream->w_lock);
 		free((*output)->stream);
@@ -409,6 +419,70 @@ nmsg_output_set_zlibout(nmsg_output_t output, bool zlibout) {
 	output->stream->do_zlib = zlibout;
 }
 
+nmsg_res
+nmsg_output_set_zlib_workers(nmsg_output_t output, unsigned workers)
+{
+	/*
+	 * Type test first: 'stream' is a union member, so reading stream->type
+	 * on a pres or json output would reinterpret another struct's bytes.
+	 */
+	if (output->type != nmsg_output_type_stream)
+		return (nmsg_res_success);
+	if (output->stream->type != nmsg_stream_type_file)
+		return (nmsg_res_success);
+
+	/*
+	 * Unbuffered flushes a container per message, so a pool would spend a
+	 * ticket and a wakeup per message to compress a single payload.
+	 */
+	if (workers > 0 && !output->stream->buffered)
+		return (nmsg_res_failure);
+
+	if (workers > 0)
+		return (_output_async_init(output, workers));
+
+	/* Turning a pool off can strand an error a worker recorded. */
+	return (_output_async_destroy(output));
+}
+
+void
+nmsg_output_set_zlib_cull(nmsg_output_t output, unsigned min_workers,
+			  unsigned idle_secs)
+{
+	struct nmsg_stream_output *ostr;
+	struct nmsg_ostr_async *pool;
+
+	/* Type test first, for the reason nmsg_output_set_zlib_workers() gives. */
+	if (output->type != nmsg_output_type_stream)
+		return;
+	if (output->stream->type != nmsg_stream_type_file)
+		return;
+
+	ostr = output->stream;
+
+	/*
+	 * Kept on the stream, not just in the pool: a ceiling change replaces
+	 * the pool, and this way the two setters may be called in any order.
+	 * Nothing is logged when there is no pool -- nmsgtool sets a policy on
+	 * every output, so it would fire on a plain '--unbuffered -w'.
+	 */
+	pthread_mutex_lock(&ostr->c_lock);
+	ostr->so_zmin = min_workers;
+	ostr->so_zcull = idle_secs;
+
+	/*
+	 * Taken under c_lock so a teardown cannot free the pool underneath.
+	 * Lock order is c_lock then pool->lock; nothing takes them the other way.
+	 */
+	pool = _output_async_ref(ostr);
+	pthread_mutex_unlock(&ostr->c_lock);
+
+	if (pool != NULL) {
+		_output_async_set_cull(pool, min_workers, idle_secs);
+		_output_async_unref(ostr);
+	}
+}
+
 void
 nmsg_output_set_endline(nmsg_output_t output, const char *endline) {
 	if (output->type == nmsg_output_type_pres) {
@@ -535,6 +609,8 @@ output_open_stream_base(nmsg_stream_type type, size_t bufsz) {
 	}
 	output->stream->type = type;
 	output->stream->buffered = true;
+	output->stream->so_zmin = NMSG_ZCULL_MIN_WORKERS_DEFAULT;
+	output->stream->so_zcull = NMSG_ZCULL_SECS_DEFAULT;
 
 	/* seed the rng, needed for fragment and sequence IDs */
 	output->stream->random = nmsg_random_init();
@@ -546,8 +622,21 @@ output_open_stream_base(nmsg_stream_type type, size_t bufsz) {
 
 	pthread_mutex_init(&output->stream->c_lock, NULL);
 	pthread_mutex_init(&output->stream->w_lock, NULL);
+	if (pthread_cond_init(&output->stream->c_drained, NULL) != 0) {
+		nmsg_random_destroy(&output->stream->random);
+		pthread_mutex_destroy(&output->stream->c_lock);
+		pthread_mutex_destroy(&output->stream->w_lock);
+		free(output->stream);
+		free(output);
+		return (NULL);
+	}
 
-	/* enable container sequencing */
+	/*
+	 * Enable container sequencing. Sock and zmq only, which is what lets
+	 * the async compressor number containers in compression order;
+	 * widening this test needs _output_nmsg_container_compress() looked
+	 * at.
+	 */
 	if (output->stream->type == nmsg_stream_type_sock ||
 	    output->stream->type == nmsg_stream_type_zmq)
 	{
@@ -570,6 +659,7 @@ output_open_stream_base(nmsg_stream_type type, size_t bufsz) {
 	output->stream->c = nmsg_container_init(bufsz);
 	if (output->stream->c == NULL) {
 		nmsg_random_destroy(&output->stream->random);
+		pthread_cond_destroy(&output->stream->c_drained);
 		pthread_mutex_destroy(&output->stream->c_lock);
 		pthread_mutex_destroy(&output->stream->w_lock);
 		free(output->stream);