mctpd: timerfd based timers

Replace all sd-event timers with timerfd-like interfaces for
testability.

On release builds, those uses real timerfds.

On test builds, those are backed by MockClocks. autojump_clock is
ultilized to skip through all the Trio time related wait, reducing test
time.

Because we still have real timeouts (D-Bus call to mctpd subprocess,
mctpd SO_RCVTIMEO wait, ...), autojump_threshold (how much to wait
before skipping Trio timeouts) is set to a reasonable value to take that
into account.

Tested: Only tested with tests. Real timerfd backend is entirely
untested.

Signed-off-by: Khang D Nguyen <khangng@os.amperecomputing.com>

Author: khangng-ampere

src/mctp-ops.c

@@ -7,6 +7,8 @@
 
 #define _GNU_SOURCE
 
+#include <sys/time.h>
+#include <sys/timerfd.h>
 #include <unistd.h>
 #include <linux/netlink.h>
 #include <err.h>
@@ -57,6 +59,58 @@ static void mctp_bug_warn(const char *fmt, va_list args)
 	vwarnx(fmt, args);
 }
 
+static inline struct itimerspec itimespec_from_usec(uint64_t u)
+{
+	return (struct itimerspec){
+		.it_interval = { 0 },
+		.it_value.tv_sec = u / 1000000,
+		.it_value.tv_nsec = (u % 1000000) * 1000,
+	};
+}
+static inline uint64_t usec_from_itimespec(struct itimerspec spec)
+{
+	return (spec.it_value.tv_sec * 1000000) +
+	       (spec.it_value.tv_nsec / 1000);
+}
+
+static int mctp_op_timerfd_create()
+{
+	return timerfd_create(CLOCK_MONOTONIC, 0);
+}
+
+static int mctp_op_timerfd_settime(int fd, uint64_t new_value,
+				   uint64_t *old_value)
+{
+	struct itimerspec new = itimespec_from_usec(new_value);
+	struct itimerspec old;
+	int rc;
+
+	rc = timerfd_settime(fd, 0, &new, &old);
+	if (rc < 0)
+		return rc;
+
+	*old_value = usec_from_itimespec(old);
+
+	return 0;
+}
+
+static int mctp_op_timerfd_read(int fd, uint64_t *n)
+{
+	ssize_t rc;
+
+	rc = read(fd, n, sizeof(*n));
+	if (rc < 0) {
+		return rc;
+	}
+
+	if (rc != sizeof(*n)) {
+		warnx("timerfd read EPROTO");
+		return -1;
+	}
+
+	return 0;
+}
+
 const struct mctp_ops mctp_ops = {
 	.mctp = {
 		.socket = mctp_op_mctp_socket,
@@ -74,6 +128,11 @@ const struct mctp_ops mctp_ops = {
 		.recvfrom = mctp_op_recvfrom,
 		.close = mctp_op_close,
 	},
+	.timerfd = {
+        .create = mctp_op_timerfd_create,
+        .settime = mctp_op_timerfd_settime,
+        .read = mctp_op_timerfd_read,
+	},
 	.bug_warn = mctp_bug_warn,
 };
 

src/mctp-ops.h

@@ -7,8 +7,10 @@
  */
 #pragma once
 
+#include <stdint.h>
 #include <sys/socket.h>
 #include <stdarg.h>
+#include <sys/timerfd.h>
 
 #define _GNU_SOURCE
 
@@ -24,9 +26,16 @@ struct socket_ops {
 	int (*close)(int sd);
 };
 
+struct timerfd_ops {
+	int (*create)();
+	int (*settime)(int fd, uint64_t new_value, uint64_t *old_value);
+	int (*read)(int fd, uint64_t *n);
+};
+
 struct mctp_ops {
 	struct socket_ops mctp;
 	struct socket_ops nl;
+	struct timerfd_ops timerfd;
 	void (*bug_warn)(const char *fmt, va_list args);
 };
 

src/mctpd.c

@@ -3079,12 +3079,13 @@ static int method_endpoint_remove(sd_bus_message *call, void *data,
 	(MCTP_I2C_TSYM_MT1_MAX_US + 2 * MCTP_I2C_TSYM_MT3_MAX_US)
 #define MCTP_I2C_TSYM_MT2_MAX_MS MCTP_I2C_TSYM_MT4_MIN_US
 
-static int peer_endpoint_recover(sd_event_source *s, uint64_t usec,
+static int peer_endpoint_recover(sd_event_source *s, int timerfd, uint revents,
 				 void *userdata)
 {
 	struct peer *peer = userdata;
 	struct ctx *ctx = peer->ctx;
 	const char *peer_path;
+	uint64_t n_expireds;
 	int rc;
 
 	/*
@@ -3099,6 +3100,12 @@ static int peer_endpoint_recover(sd_event_source *s, uint64_t usec,
 
 	peer->recovery.npolls--;
 
+	// flush the timerfd
+	rc = mctp_ops.timerfd.read(timerfd, &n_expireds);
+	if (rc < 0) {
+		goto reschedule;
+	}
+
 	/*
 	 * Test if we still have connectivity to the endpoint. If we do, we will get a
 	 * response reporting the current EID. This is the test recommended by 8.17.6
@@ -3194,8 +3201,8 @@ static int peer_endpoint_recover(sd_event_source *s, uint64_t usec,
 
 reschedule:
 	if (peer->recovery.npolls > 0) {
-		rc = sd_event_source_set_time_relative(peer->recovery.source,
-						       peer->recovery.delay);
+		rc = mctp_ops.timerfd.settime(timerfd, peer->recovery.delay,
+					      NULL);
 		if (rc >= 0) {
 			rc = sd_event_source_set_enabled(peer->recovery.source,
 							 SD_EVENT_ONESHOT);
@@ -3217,11 +3224,13 @@ static int method_endpoint_recover(sd_bus_message *call, void *data,
 	struct peer *peer;
 	bool previously;
 	struct ctx *ctx;
+	int timerfd;
 	int rc;
 
 	peer = data;
 	ctx = peer->ctx;
 	previously = peer->degraded;
+	timerfd = -1;
 
 	if (!previously) {
 		assert(!peer->recovery.delay);
@@ -3230,13 +3239,38 @@ static int method_endpoint_recover(sd_bus_message *call, void *data,
 		peer->recovery.npolls = MCTP_I2C_TSYM_MN1_MIN + 1;
 		peer->recovery.delay =
 			(MCTP_I2C_TSYM_TRECLAIM_MIN_US / 2) - ctx->mctp_timeout;
-		rc = sd_event_add_time_relative(
-			ctx->event, &peer->recovery.source, CLOCK_MONOTONIC, 0,
-			ctx->mctp_timeout, peer_endpoint_recover, peer);
+
+		timerfd = mctp_ops.timerfd.create();
+		if (timerfd < 0) {
+			rc = -errno;
+			goto out;
+		}
+
+		rc = mctp_ops.timerfd.settime(timerfd, ctx->mctp_timeout, NULL);
+		if (rc < 0) {
+			goto out;
+		}
+
+		rc = sd_event_add_io(ctx->event, &peer->recovery.source,
+				     timerfd, EPOLLIN, peer_endpoint_recover,
+				     peer);
+		if (rc < 0) {
+			goto out;
+		}
+
+		rc = sd_event_source_set_enabled(peer->recovery.source,
+						 SD_EVENT_ONESHOT);
 		if (rc < 0) {
 			goto out;
 		}
 
+		rc = sd_event_source_set_io_fd_own(peer->recovery.source, true);
+		if (rc < 0) {
+			goto out;
+		}
+		// prevent double close, now that sd_event_source own the fd
+		timerfd = -1;
+
 		peer->degraded = true;
 
 		rc = sd_bus_emit_properties_changed(
@@ -3253,12 +3287,8 @@ static int method_endpoint_recover(sd_bus_message *call, void *data,
 
 out:
 	if (rc < 0 && !previously) {
-		if (peer->degraded) {
-			/* Cleanup the timer if it was setup successfully. */
-			sd_event_source_set_enabled(peer->recovery.source,
-						    SD_EVENT_OFF);
-			sd_event_source_unref(peer->recovery.source);
-		}
+		sd_event_source_disable_unref(peer->recovery.source);
+		close(timerfd);
 		peer->degraded = previously;
 		peer->recovery.delay = 0;
 		peer->recovery.source = NULL;

tests/conftest.py

@@ -1,5 +1,6 @@
 
 import sys
+import math
 
 import pytest
 import asyncdbus
@@ -25,7 +26,14 @@ def config():
     return None
 
 @pytest.fixture
-async def mctpd(nursery, dbus, sysnet, config):
+async def mctpd(nursery, dbus, sysnet, config, autojump_clock):
+    # some tests ultilize autojump_clock, which might interfere with
+    # the mctpd initialization NameOwnerChanged D-Bus call timeout
+    #
+    # initialize autojump_clock parameter to normal value here,
+    # then tests can override their own reasonable values
+    autojump_clock.rate = 1
+    autojump_clock.autojump_threshold = math.inf
     m = mctpenv.MctpdWrapper(dbus, sysnet, config = config)
     await m.start_mctpd(nursery)
     yield m

tests/mctp-ops-test.c

@@ -7,6 +7,7 @@
 
 #define _GNU_SOURCE
 
+#include <assert.h>
 #include <err.h>
 #include <errno.h>
 #include <stdlib.h>
@@ -38,10 +39,12 @@ static int mctp_op_socket(int type)
 	struct iovec iov;
 	int rc, var, sd;
 
-	if (type == AF_MCTP)
+	if (type == CONTROL_OP_SOCKET_MCTP)
 		req.type = CONTROL_OP_SOCKET_MCTP;
-	else if (type == AF_NETLINK)
+	else if (type == CONTROL_OP_SOCKET_NL)
 		req.type = CONTROL_OP_SOCKET_NL;
+	else if (type == CONTROL_OP_TIMER)
+		req.type = CONTROL_OP_TIMER;
 	else
 		errx(EXIT_FAILURE, "invalid socket type?");
 
@@ -72,12 +75,12 @@ static int mctp_op_socket(int type)
 
 static int mctp_op_mctp_socket(void)
 {
-	return mctp_op_socket(AF_MCTP);
+	return mctp_op_socket(CONTROL_OP_SOCKET_MCTP);
 }
 
 static int mctp_op_netlink_socket(void)
 {
-	return mctp_op_socket(AF_NETLINK);
+	return mctp_op_socket(CONTROL_OP_SOCKET_NL);
 }
 
 static int mctp_op_bind(int sd, struct sockaddr *addr, socklen_t addrlen)
@@ -227,6 +230,55 @@ static void mctp_bug_warn(const char *fmt, va_list args)
 	abort();
 }
 
+static int mctp_op_timerfd_create()
+{
+	return mctp_op_socket(CONTROL_OP_TIMER);
+}
+
+static int mctp_op_timerfd_settime(int fd, uint64_t new_value,
+				   uint64_t *old_value)
+{
+	uint64_t tmp;
+	ssize_t rc;
+
+	rc = write(fd, &new_value, sizeof(new_value));
+	if (rc < 0)
+		return rc;
+
+	// Maybe handle partial read?
+	if ((size_t)rc < sizeof(new_value))
+		errx(EXIT_FAILURE, "ops protocol error");
+
+	rc = read(fd, &tmp, sizeof(tmp));
+	if (rc < 0)
+		return rc;
+
+	if ((size_t)rc < sizeof(tmp))
+		errx(EXIT_FAILURE, "ops protocol error");
+
+	if (old_value)
+		*old_value = tmp;
+
+	return 0;
+}
+
+static int mctp_op_timerfd_read(int fd, uint64_t *n)
+{
+	ssize_t rc;
+	uint64_t v;
+
+	rc = read(fd, &v, sizeof(v));
+	if (rc < 0)
+		return rc;
+
+	if ((size_t)rc < sizeof(v) || v != 0)
+		errx(EXIT_FAILURE, "ops protocol error");
+
+	*n = 1;
+
+	return 0;
+}
+
 const struct mctp_ops mctp_ops = {
 	.mctp = {
 		.socket = mctp_op_mctp_socket,
@@ -244,6 +296,11 @@ const struct mctp_ops mctp_ops = {
 		.recvfrom = mctp_op_recvfrom,
 		.close = mctp_op_close,
 	},
+	.timerfd = {
+        .create = mctp_op_timerfd_create,
+        .settime = mctp_op_timerfd_settime,
+        .read = mctp_op_timerfd_read,
+	},
 	.bug_warn = mctp_bug_warn,
 };