imtcp: major multithreading and performance improvements

This commit significantly enhances imtcp by introducing a fully
functional worker thread pool, enabling true multi-threaded
processing for better scalability under high loads. This is
particularly beneficial when using TLS connections.

Notable changes:
- Implemented a complete worker pool for imtcp.
- Introduced the `workerthreads` config parameter for tuning
  concurrency.
- Improved epoll efficiency by enabling edge-triggered mode.
- Added starvation handling via `starvationProtection.maxReads`.
- Refactored session accept logic and optimized network object
  handling.
- Removed an obsolete network driver layer for event notification.
- Fixed multiple issues related to message timing, EPOLLERR
  handling, and tests.
- Improved performance in poll() mode by reducing redundant
  allocations.
- Introduced new CI tests for imtcp without epoll.
- Allowed disabling imtcp tests via a new configure switch.
- Added new impstats counters for worker thread pool statistics.

Details:
- The worker pool replaces an outdated experimental
  implementation.
- If `workerthreads=1`, no worker pool is created to minimize
  context switches.
- Moves worker pool variables inside `tcpsrv` instance to
  prevent conflicts.
- Extracts session `accept()` logic into a dedicated function
  for clarity.
- Fixes message ordering inconsistencies in multi-threaded
  scenarios.
- Properly handles `EPOLLERR` notifications to improve error
  resilience.
- Optimizes poll() mode by avoiding unnecessary reallocation
  of file descriptors.
- Replaces the old network driver layer for event notification
  with a streamlined solution.
  - Now uses **conditional compilation** to select the best
    method (epoll or poll) at build time.
  - This significantly reduces code complexity, improves
    maintainability, and boosts performance.
- The previous "thread pool" was a rough experiment that did
  not perform significantly better than single-threaded mode.
  - The **new implementation** allows multiple worker threads
    on platforms with `epoll`.
  - On non-epoll systems, an optimized **poll() based
    single-threaded approach** is used, which is expected to
    perform better than the old "thread pool."
- Adds `pthread_setname_np` only when available to improve
  portability.
- Fixes test cases that assumed strict message timing, which
  was unreliable.
- Reduces test parallelism for TSAN CI runs to prevent
  resource exhaustion.
- Moves a test case to `imdiag` to ensure stable execution.
- Provides a new CI environment to verify `imtcp` behavior
  without epoll.
- Introduces `--enable-imtcp-tests` configure switch for test
  flexibility.
- Improves debug logging and adds better error handling for
  worker pool startup.

New configuration parameters:
- `workerthreads`: Defines the number of worker threads for
  imtcp. If set to 1, no worker pool is created.
- `starvationProtection.maxReads`: Defines the maximum number
  of consecutive reads a worker can perform before being
  interrupted to allow other sessions to be processed.

New impstats counters (emitted only when `workerthreads > 1`):
- `runs`: Number of times the worker thread has been invoked.
- `read`: Number of read calls performed by the worker.
  For TLS, this includes read/write calls.
- `accept`: Number of `accept()` calls handled by the worker.
- `starvation_protect`: Number of times a socket was sent
  back to the queue due to reaching the maximum number of
  consecutive requests, ensuring fair scheduling of sessions.

These changes significantly enhance rsyslog’s TCP handling
performance and stability, particularly in high-volume
environments.

Closes rsyslog#5529, rsyslog#5532, rsyslog#5578, rsyslog#5580.

Author: rgerhards

.github/workflows/run_checks.yml

@@ -36,6 +36,7 @@ jobs:
       fail-fast: false
       matrix:
         config: [centos_7, centos_8, debian_10, debian_11,
+                 ubuntu_24_imtcp_no_epoll,
                  fedora_35, fedora_36,
                  ubuntu_18, ubuntu_20, ubuntu_24,
                  ubuntu_22_san, ubuntu_24_tsan, ubuntu_22_codecov, ubuntu_22_distcheck,
@@ -79,6 +80,21 @@ jobs:
               export RSYSLOG_CONFIGURE_OPTIONS_EXTRA="--disable-elasticsearch-tests --disable-kafka-tests \
                             --without-valgrind-testbench --enable-imdtls --enable-omdtls"
               ;;
+          'ubuntu_24_imtcp_no_epoll')
+              # This check tests if imtcp runs in poll (select) mode. We have only slow CI runners for
+              # platforms where this is the case, thus we do a quick run on ubuntu where we also have all
+              # thread and memory debuggers available.
+              export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_ubuntu:24.04'
+              export CI_VALGRIND_SUPPRESSIONS="ubuntu22.04.supp"
+              # Note: we completely override the container configure options here!
+              export RSYSLOG_CONFIGURE_OPTIONS_OVERRIDE="--enable-testbench --enable-omstdout \
+                     --enable-imdiag --disable-imtcp-epoll --enable-imtcp-tests\
+                     --disable-impstats --disable-imfile --disable-imfile-tests \
+                     --disable-fmhttp --enable-valgrind --enable-valgrind-testbench \
+                     --disable-helgrind --disable-default-tests --disable-kafka-tests \
+                     --disable-omkafka --disable-imkafka \
+                     --enable-gnutls --enable-openssl --enable-gnutls-tests"
+              ;;
           'fedora_35')
               export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_fedora:35'
               export RSYSLOG_CONFIGURE_OPTIONS_EXTRA="--disable-elasticsearch-tests \
@@ -130,7 +146,7 @@ jobs:
               ;;
           'ubuntu_24_tsan')
               export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_ubuntu:24.04'
-              export CI_MAKE_CHECK_OPT='-j3' # This is CPU-heavy due to tsan, so we need less concurrency to prevent flakes
+              export CI_MAKE_CHECK_OPT='-j2' # This is CPU-heavy due to tsan, so we need less concurrency to prevent flakes
               export CI_VALGRIND_SUPPRESSIONS="ubuntu22.04.supp"
               export CI_SANITIZE_BLACKLIST='tests/tsan.supp'
               export CC='clang'
@@ -169,7 +185,7 @@ jobs:
               export RSYSLOG_CONFIGURE_OPTIONS_OVERRIDE="--enable-testbench --enable-omstdout \
                      --enable-imdiag --disable-impstats --enable-imfile --disable-imfile-tests \
                      --disable-fmhttp --enable-valgrind --enable-valgrind-testbench \
-                     --disable-helgrind --disable-default-tests --enable-kafka-tests \
+                     --disable-helgrind --disable-default-tests --disable-imtcp-tests --enable-kafka-tests \
                      --enable-omkafka --enable-gnutls --disable-gnutls-tests --enable-imkafka"
               ;;
           'elasticsearch')
@@ -180,7 +196,7 @@ jobs:
               export ABORT_ALL_ON_TEST_FAIL='NO'
               export RSYSLOG_CONFIGURE_OPTIONS_OVERRIDE="--enable-testbench --enable-omstdout \
                      --enable-imdiag --enable-impstats --enable-imfile --disable-imfile-tests \
-                     --disable-fmhttp --enable-valgrind --disable-default-tests \
+                     --disable-fmhttp --enable-valgrind --disable-default-tests --disable-imtcp-tests \
                      --enable-elasticsearch-tests --enable-elasticsearch"
               export CI_MAKE_OPT='-j20'
               export CI_MAKE_CHECK_OPT='-j8'

configure.ac

@@ -233,7 +233,7 @@ AC_TYPE_SIGNAL
 AC_FUNC_STAT
 AC_FUNC_STRERROR_R
 AC_FUNC_VPRINTF
-AC_CHECK_FUNCS([flock recvmmsg basename alarm clock_gettime gethostbyname gethostname gettimeofday localtime_r memset mkdir regcomp select setsid socket strcasecmp strchr strdup strerror strndup strnlen strrchr strstr strtol strtoul uname ttyname_r getline malloc_trim prctl epoll_create epoll_create1 fdatasync syscall lseek64 asprintf close_range])
+AC_CHECK_FUNCS([flock recvmmsg basename alarm clock_gettime gethostbyname gethostname gettimeofday localtime_r memset mkdir regcomp select setsid socket strcasecmp strchr strdup strerror strndup strnlen strrchr strstr strtol strtoul uname ttyname_r getline malloc_trim prctl epoll_create epoll_create1 fdatasync syscall lseek64 asprintf close_range pthread_setname_np])
 AC_CHECK_FUNC([setns], [AC_DEFINE([HAVE_SETNS], [1], [Define if setns exists.])])
 AC_CHECK_TYPES([off64_t])
 
@@ -1792,8 +1792,6 @@ AM_CONDITIONAL(ENABLE_LIBFAKETIME, test "x${enable_libfaketime}" = "xyes")
 # are those tests that do not need a special configure option. There are
 # some situations where we really want to turn them of so that we can
 # run tests only for a specific component (e.g. ElasticSearch).
-# This also enables us to do some parallel testing even while the
-# testbench is not yet able to support make -j check
 AC_ARG_ENABLE(default-tests,
         [AS_HELP_STRING([--enable-default-tests],[default-tests enabled @<:@default=yes@:>@])],
         [case "${enableval}" in
@@ -1811,6 +1809,17 @@ if test "x${IP}" = "xno"; then
 fi
 AM_CONDITIONAL(ENABLE_IP, test "x${IP}" = "xyes")
 
+AC_ARG_ENABLE(imtcp-tests,
+        [AS_HELP_STRING([--enable-imtcp-tests],[imtcp-tests enabled @<:@default=yes@:>@])],
+        [case "${enableval}" in
+         yes) enable_imtcp_tests="yes" ;;
+          no) enable_imtcp_tests="no" ;;
+           *) AC_MSG_ERROR(bad value ${enableval} for --enable-imtcp-tests) ;;
+         esac],
+        [enable_imtcp_tests=yes]
+)
+AM_CONDITIONAL(ENABLE_IMTCP_TESTS, test "x${enable_imtcp_tests}" = "xyes")
+
 
 # valgrind-testbench
 AC_ARG_WITH([valgrind_testbench],
@@ -2017,6 +2026,58 @@ AC_ARG_ENABLE(imsolaris,
 )
 AM_CONDITIONAL(ENABLE_IMSOLARIS, test x$enable_imsolaris = xyes)
 
+
+# settings for the imtcp input module
+AC_ARG_ENABLE(imtcp_epoll,
+        [AS_HELP_STRING([--disable-imtcp-epoll],[Disable epoll mode in the TCP input module (for testing purposes)])],
+        [case "${enableval}" in
+         yes) enable_imtcp_epoll="yes" ;;
+          no) enable_imtcp_epoll="no" ;;
+           *) AC_MSG_ERROR([bad value ${enableval} for --disable-imtcp-epoll]) ;;
+         esac],
+        [enable_imtcp_epoll=auto]
+)
+
+# Default: Enable imtcp_epoll if epoll is available, unless explicitly disabled
+if test "x$enable_imtcp_epoll" = "xauto"; then
+    if test "x$ac_cv_header_sys_epoll_h" = "xyes" && test "x$ac_cv_func_epoll_create" = "xyes"; then
+        enable_imtcp_epoll="yes"
+    else
+        enable_imtcp_epoll="no"
+    fi
+fi
+
+# Define ENABLE_IMTCP_EPOLL in config.h if epoll is enabled
+if test "x$enable_imtcp_epoll" = "xyes"; then
+    AC_DEFINE([ENABLE_IMTCP_EPOLL], [1], [Enable epoll mode for the imtcp input module])
+fi
+
+# Automake conditional
+AM_CONDITIONAL(ENABLE_IMTCP_EPOLL, test x$enable_imtcp_epoll = xyes)
+
+ Print result
+echo "    TCP input module epoll mode: $enable_imtcp_epoll"
+
+
+
+
+# settings for the imtcp input module
+AC_ARG_ENABLE(imtcp_epoll,
+        [AS_HELP_STRING([--enable-imtcp-epoll],[tcp input module epoll mode permitted @<:@default=yes@:>@])],
+        [case "${enableval}" in
+         yes) enable_imtcp_epoll="yes" ;;
+          no) enable_imtcp_epoll="no" ;;
+           *) AC_MSG_ERROR(bad value ${enableval} for --enable-imptcp-epoll) ;;
+         esac],
+        [enable_imptcp_epoll=yes]
+)
+AM_CONDITIONAL(ENABLE_IMPTCP_EPOLL, test x$enable_imptcp_epoll = xyes)
+# Define a preprocessor macro if enabled
+if test "x$enable_imtcp_epoll" = "xyes"; then
+    AC_DEFINE([ENABLE_IMTCP_EPOLL], [1], [Enable epoll mode for the imtcp input module])
+fi
+
+
 # settings for the ptcp input module
 AC_ARG_ENABLE(imptcp,
         [AS_HELP_STRING([--enable-imptcp],[plain tcp input module enabled @<:@default=no@:>@])],
@@ -3010,6 +3071,7 @@ if test "$unamestr" != "AIX"; then
 echo "    Klog functionality enabled:               $enable_klog ($os_type)"
 fi
 echo "    /dev/kmsg functionality enabled:          $enable_kmsg"
+echo "    tcp input module epoll permitted:         $enable_imtcp_epoll"
 echo "    plain tcp input module enabled:           $enable_imptcp"
 echo "    DTLS udp input module enabled:            $enable_imdtls"
 echo "    imdiag enabled:                           $enable_imdiag"
@@ -3124,6 +3186,7 @@ echo "    Kafka Tests enabled:                      $enable_kafka_tests"
 echo "    Redis Tests enabled:                      $enable_redis_tests"
 echo "    Omazureeventhubs Tests enabled:           $enable_omazureeventhubs_tests"
 echo "    Imdocker Tests enabled:                   $enable_imdocker_tests"
+echo "    imtcp tests enabled:                      $enable_imtcp_tests"
 echo "    gnutls tests enabled:                     $enable_gnutls_tests"
 echo "    imfile tests enabled:                     $enable_imfile_tests"
 echo "    systemd journal tests enabled:            $enable_journal_tests"

plugins/imdiag/imdiag.c

@@ -617,6 +617,7 @@ addTCPListener(void __attribute__((unused)) *pVal, uchar *pNewVal)
 	}
 	CHKmalloc(cnf_params = (tcpLstnParams_t*) calloc(1, sizeof(tcpLstnParams_t)));
 	CHKiRet(tcpsrv.Construct(&pOurTcpsrv));
+	CHKiRet(tcpsrv.SetNumWrkr(pOurTcpsrv, 1)); // We always need only one worker for this diag module!
 	CHKiRet(tcpsrv.SetSessMax(pOurTcpsrv, iTCPSessMax));
 	CHKiRet(tcpsrv.SetCBIsPermittedHost(pOurTcpsrv, isPermittedHost));
 	CHKiRet(tcpsrv.SetCBRcvData(pOurTcpsrv, doRcvData));

plugins/imtcp/imtcp.c

@@ -91,6 +91,12 @@ static int n_tcpsrv = 0;
 
 static permittedPeers_t *pPermPeersRoot = NULL;
 
+/* default number of workes to configure. We choose 2, as this is probably good for
+ * many installations. High-Volume ones may need much higher number!
+ */
+#define DEFAULT_NUMWRKR 2
+#define DEFAULT_STARVATIONMAXREADS 500
+
 #define FRAMING_UNSET -1
 
 /* config settings */
@@ -123,6 +129,7 @@ static struct configSettings_s {
 struct instanceConf_s {
 	int iTCPSessMax;
 	int iTCPLstnMax;
+	unsigned numWrkr;
 	tcpLstnParams_t *cnf_params;	/**< listener config parameters */
 	uchar *pszBindRuleset;		/* name of ruleset to bind to */
 	ruleset_t *pBindRuleset;	/* ruleset to bind listener to (use system default if unspecified) */
@@ -157,6 +164,7 @@ struct instanceConf_s {
 	int iKeepAliveIntvl;
 	int iKeepAliveProbes;
 	int iKeepAliveTime;
+	unsigned starvationMaxReads;
 	struct instanceConf_s *next;
 };
 
@@ -166,6 +174,7 @@ struct modConfData_s {
 	instanceConf_t *root, *tail;
 	int iTCPSessMax; /* max number of sessions */
 	int iTCPLstnMax; /* max number of sessions */
+	unsigned numWrkr;
 	int iStrmDrvrMode; /* mode for stream driver, driver-dependent (0 mostly means plain tcp) */
 	int iStrmDrvrExtendedCertCheck; /* verify also purpose OID in certificate extended field */
 	int iStrmDrvrSANPreference; /* ignore CN when any SAN set */
@@ -193,6 +202,7 @@ struct modConfData_s {
 	permittedPeers_t *pPermPeersRoot;
 	sbool configSetViaV2Method;
 	sbool bPreserveCase; /* preserve case of fromhost; true by default */
+	unsigned starvationMaxReads;
 };
 
 static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */
@@ -211,6 +221,8 @@ static struct cnfparamdescr modpdescr[] = {
 	{ "maxsessions", eCmdHdlrPositiveInt, 0 },
 	{ "maxlistners", eCmdHdlrPositiveInt, 0 },
 	{ "maxlisteners", eCmdHdlrPositiveInt, 0 },
+	{ "workerthreads", eCmdHdlrPositiveInt, 0 },
+	{ "starvationprotection.maxreads", eCmdHdlrNonNegInt, 0 },
 	{ "streamdriver.mode", eCmdHdlrNonNegInt, 0 },
 	{ "streamdriver.authmode", eCmdHdlrString, 0 },
 	{ "streamdriver.permitexpiredcerts", eCmdHdlrString, 0 },
@@ -241,6 +253,7 @@ static struct cnfparamdescr inppdescr[] = {
 	{ "port", eCmdHdlrString, CNFPARAM_REQUIRED }, /* legacy: InputTCPServerRun */
 	{ "maxsessions", eCmdHdlrPositiveInt, 0 },
 	{ "maxlisteners", eCmdHdlrPositiveInt, 0 },
+	{ "workerthreads", eCmdHdlrPositiveInt, 0 },
 	{ "flowcontrol", eCmdHdlrBinary, 0 },
 	{ "disablelfdelimiter", eCmdHdlrBinary, 0 },
 	{ "discardtruncatedmsg", eCmdHdlrBinary, 0 },
@@ -254,6 +267,7 @@ static struct cnfparamdescr inppdescr[] = {
 	{ "name", eCmdHdlrString, 0 },
 	{ "defaulttz", eCmdHdlrString, 0 },
 	{ "ruleset", eCmdHdlrString, 0 },
+	{ "starvationprotection.maxreads", eCmdHdlrNonNegInt, 0 },
 	{ "streamdriver.mode", eCmdHdlrNonNegInt, 0 },
 	{ "streamdriver.authmode", eCmdHdlrString, 0 },
 	{ "streamdriver.permitexpiredcerts", eCmdHdlrString, 0 },
@@ -403,6 +417,8 @@ createInstance(instanceConf_t **pinst)
 	inst->iSynBacklog = 0; /* default: unset */
 	inst->iTCPLstnMax = loadModConf->iTCPLstnMax;
 	inst->iTCPSessMax = loadModConf->iTCPSessMax;
+	inst->numWrkr = loadModConf->numWrkr;
+	inst->starvationMaxReads = loadModConf->starvationMaxReads;
 
 	inst->cnf_params->pszLstnPortFileName = NULL;
 
@@ -473,6 +489,8 @@ static rsRetVal addInstance(void __attribute__((unused)) *pVal, uchar *pNewVal)
 	inst->iAddtlFrameDelim = cs.iAddtlFrameDelim;
 	inst->iTCPLstnMax = cs.iTCPLstnMax;
 	inst->iTCPSessMax = cs.iTCPSessMax;
+	inst->numWrkr = DEFAULT_NUMWRKR;
+	inst->starvationMaxReads = DEFAULT_STARVATIONMAXREADS;
 	inst->iStrmDrvrMode = cs.iStrmDrvrMode;
 
 finalize_it:
@@ -497,6 +515,8 @@ addListner(modConfData_t *modConf, instanceConf_t *inst)
 	CHKiRet(tcpsrv.SetCBOnRegularClose(pOurTcpsrv, onRegularClose));
 	CHKiRet(tcpsrv.SetCBOnErrClose(pOurTcpsrv, onErrClose));
 	/* params */
+	CHKiRet(tcpsrv.SetNumWrkr(pOurTcpsrv, inst->numWrkr));
+	CHKiRet(tcpsrv.SetStarvationMaxReads(pOurTcpsrv, inst->starvationMaxReads));
 	CHKiRet(tcpsrv.SetKeepAlive(pOurTcpsrv, inst->bKeepAlive));
 	CHKiRet(tcpsrv.SetKeepAliveIntvl(pOurTcpsrv, inst->iKeepAliveIntvl));
 	CHKiRet(tcpsrv.SetKeepAliveProbes(pOurTcpsrv, inst->iKeepAliveProbes));
@@ -652,6 +672,8 @@ CODESTARTnewInpInst
 			inst->pszStrmDrvrCertFile = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
 		} else if(!strcmp(inppblk.descr[i].name, "streamdriver.name")) {
 			inst->pszStrmDrvrName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
+		} else if(!strcmp(inppblk.descr[i].name, "starvationprotection.maxreads")) {
+			inst->starvationMaxReads = (unsigned) pvals[i].val.d.n;
 		} else if(!strcmp(inppblk.descr[i].name, "gnutlsprioritystring")) {
 			inst->gnutlsPriorityString = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
 		} else if(!strcmp(inppblk.descr[i].name, "permittedpeer")) {
@@ -685,6 +707,8 @@ CODESTARTnewInpInst
 			inst->iTCPSessMax = (int) pvals[i].val.d.n;
 		} else if(!strcmp(inppblk.descr[i].name, "maxlisteners")) {
 			inst->iTCPLstnMax = (int) pvals[i].val.d.n;
+		} else if(!strcmp(inppblk.descr[i].name, "workerthreads")) {
+			inst->numWrkr = (int) pvals[i].val.d.n;
 		} else if(!strcmp(inppblk.descr[i].name, "supportoctetcountedframing")) {
 			inst->cnf_params->bSuppOctetFram = (int) pvals[i].val.d.n;
 		} else if(!strcmp(inppblk.descr[i].name, "keepalive")) {
@@ -723,6 +747,8 @@ CODESTARTbeginCnfLoad
 	/* init our settings */
 	loadModConf->iTCPSessMax = 200;
 	loadModConf->iTCPLstnMax = 20;
+	loadModConf->numWrkr = DEFAULT_NUMWRKR;
+	loadModConf->starvationMaxReads = DEFAULT_STARVATIONMAXREADS;
 	loadModConf->bSuppOctetFram = 1;
 	loadModConf->iStrmDrvrMode = 0;
 	loadModConf->iStrmDrvrExtendedCertCheck = 0;
@@ -800,9 +826,13 @@ CODESTARTsetModCnf
 			}
 		} else if(!strcmp(modpblk.descr[i].name, "maxsessions")) {
 			loadModConf->iTCPSessMax = (int) pvals[i].val.d.n;
+		} else if(!strcmp(modpblk.descr[i].name, "starvationprotection.maxreads")) {
+			loadModConf->starvationMaxReads = (unsigned) pvals[i].val.d.n;
 		} else if(!strcmp(modpblk.descr[i].name, "maxlisteners") ||
 			  !strcmp(modpblk.descr[i].name, "maxlistners")) { /* keep old name for a while */
 			loadModConf->iTCPLstnMax = (int) pvals[i].val.d.n;
+		} else if(!strcmp(modpblk.descr[i].name, "workerthreads")) {
+			loadModConf->numWrkr = (int) pvals[i].val.d.n;
 		} else if(!strcmp(modpblk.descr[i].name, "keepalive")) {
 			loadModConf->bKeepAlive = (int) pvals[i].val.d.n;
 		} else if(!strcmp(modpblk.descr[i].name, "keepalive.probes")) {
@@ -1078,7 +1108,6 @@ CODESTARTafterRun
 	tcpsrv_etry_t *del;
 	while(etry != NULL) {
 		iRet = tcpsrv.Destruct(&etry->tcpsrv);
-		// TODO: check iRet, reprot error
 		del = etry;
 		etry = etry->next;
 		free(del);

runtime/Makefile.am

@@ -179,9 +179,7 @@ endif
 
 # network stream master class and stream factory
 lmnetstrms_la_SOURCES = netstrms.c netstrms.h \
-			netstrm.c netstrm.h \
-			nssel.c nssel.h \
-			nspoll.c nspoll.h
+			netstrm.c netstrm.h
 lmnetstrms_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS)
 lmnetstrms_la_LDFLAGS = -module -avoid-version
 lmnetstrms_la_LIBADD =
@@ -195,9 +193,7 @@ endif
 
 # plain tcp driver - main driver
 pkglib_LTLIBRARIES += lmnsd_ptcp.la
-lmnsd_ptcp_la_SOURCES = nsd_ptcp.c nsd_ptcp.h \
-		  	nsdsel_ptcp.c nsdsel_ptcp.h \
-			nsdpoll_ptcp.c nsdpoll_ptcp.h
+lmnsd_ptcp_la_SOURCES = nsd_ptcp.c nsd_ptcp.h
 lmnsd_ptcp_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS)
 lmnsd_ptcp_la_LDFLAGS = -module -avoid-version
 lmnsd_ptcp_la_LIBADD =
@@ -215,7 +211,7 @@ endif # if ENABLE_INET
 if ENABLE_OPENSSL
 # noinst_LTLIBRARIES += lmnsd_ossl.la
 pkglib_LTLIBRARIES += lmnsd_ossl.la
-lmnsd_ossl_la_SOURCES = net_ossl.c net_ossl.h nsd_ossl.c nsd_ossl.h nsdsel_ossl.c  nsdsel_ossl.h
+lmnsd_ossl_la_SOURCES = net_ossl.c net_ossl.h nsd_ossl.c nsd_ossl.h
 lmnsd_ossl_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS) $(OPENSSL_CFLAGS)
 lmnsd_ossl_la_LDFLAGS = -module -avoid-version
 lmnsd_ossl_la_LIBADD = $(OPENSSL_LIBS)
@@ -227,7 +223,7 @@ endif
 #
 if ENABLE_GNUTLS
 pkglib_LTLIBRARIES += lmnsd_gtls.la
-lmnsd_gtls_la_SOURCES = nsd_gtls.c nsd_gtls.h nsdsel_gtls.c  nsdsel_gtls.h
+lmnsd_gtls_la_SOURCES = nsd_gtls.c nsd_gtls.h
 lmnsd_gtls_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS) $(GNUTLS_CFLAGS)
 lmnsd_gtls_la_LDFLAGS = -module -avoid-version
 lmnsd_gtls_la_LIBADD = $(GNUTLS_LIBS)

runtime/netstrms.c

@@ -34,8 +34,6 @@
 #include "obj.h"
 #include "nsd.h"
 #include "netstrm.h"
-#include "nssel.h"
-#include "nspoll.h"
 #include "netstrms.h"
 #include "rsconf.h"
 
@@ -539,8 +537,6 @@ ENDObjClassInit(netstrms)
 
 BEGINmodExit
 CODESTARTmodExit
-	nsselClassExit();
-	nspollClassExit();
 	netstrmsClassExit();
 	netstrmClassExit(); /* we use this object, so we must exit it after we are finished */
 ENDmodExit
@@ -558,7 +554,5 @@ CODESTARTmodInit
 
 	/* Initialize all classes that are in our module - this includes ourselfs */
 	CHKiRet(netstrmClassInit(pModInfo));
-	CHKiRet(nsselClassInit(pModInfo));
-	CHKiRet(nspollClassInit(pModInfo));
 	CHKiRet(netstrmsClassInit(pModInfo));
 ENDmodInit