zed: allow limiting concurrent jobs

200ms time-out is relatively long, but if we already hit the cap,
then we'll likely be able to spawn multiple new jobs when we wake up

Reviewed-by: Brian Behlendorf <behlendorf1@llnl.gov>
Signed-off-by: Ahelenia Ziemiańska <nabijaczleweli@nabijaczleweli.xyz>
Closes openzfs#11807

cmd/zed/zed_conf.c

@@ -51,6 +51,7 @@ zed_conf_create(void)
 	zcp->state_fd = -1;		/* opened via zed_conf_open_state() */
 	zcp->zfs_hdl = NULL;		/* opened via zed_event_init() */
 	zcp->zevent_fd = -1;		/* opened via zed_event_init() */
+	zcp->max_jobs = 16;
 
 	if (!(zcp->conf_file = strdup(ZED_CONF_FILE)))
 		goto nomem;
@@ -172,6 +173,8 @@ _zed_conf_display_help(const char *prog, int got_err)
 	    "Write daemon's PID to FILE.", ZED_PID_FILE);
 	fprintf(fp, "%*c%*s %s [%s]\n", w1, 0x20, -w2, "-s FILE",
 	    "Write daemon's state to FILE.", ZED_STATE_FILE);
+	fprintf(fp, "%*c%*s %s [%d]\n", w1, 0x20, -w2, "-j JOBS",
+	    "Start at most JOBS at once.", 16);
 	fprintf(fp, "\n");
 
 	exit(got_err ? EXIT_FAILURE : EXIT_SUCCESS);
@@ -251,8 +254,9 @@ _zed_conf_parse_path(char **resultp, const char *path)
 void
 zed_conf_parse_opts(struct zed_conf *zcp, int argc, char **argv)
 {
-	const char * const opts = ":hLVc:d:p:P:s:vfFMZI";
+	const char * const opts = ":hLVc:d:p:P:s:vfFMZIj:";
 	int opt;
+	unsigned long raw;
 
 	if (!zcp || !argv || !argv[0])
 		zed_log_die("Failed to parse options: Internal error");
@@ -303,6 +307,17 @@ zed_conf_parse_opts(struct zed_conf *zcp, int argc, char **argv)
 		case 'Z':
 			zcp->do_zero = 1;
 			break;
+		case 'j':
+			errno = 0;
+			raw = strtoul(optarg, NULL, 0);
+			if (errno == ERANGE || raw > INT16_MAX) {
+				zed_log_die("%lu is too many jobs", raw);
+			} if (raw == 0) {
+				zed_log_die("0 jobs makes no sense");
+			} else {
+				zcp->max_jobs = raw;
+			}
+			break;
 		case '?':
 		default:
 			if (optopt == '?')

cmd/zed/zed_conf.h

@@ -39,6 +39,7 @@ struct zed_conf {
 	libzfs_handle_t	*zfs_hdl;		/* handle to libzfs */
 	int		zevent_fd;		/* fd for access to zevents */
 	char		*path;		/* custom $PATH for zedlets to use */
+	int16_t max_jobs;		/* max zedlets to run at one time */
 };
 
 struct zed_conf *zed_conf_create(void);

cmd/zed/zed_event.c

@@ -955,8 +955,7 @@ zed_event_service(struct zed_conf *zcp)
 
 		_zed_event_add_time_strings(eid, zsp, etime);
 
-		zed_exec_process(eid, class, subclass,
-		    zcp->zedlet_dir, zcp->zedlets, zsp, zcp->zevent_fd);
+		zed_exec_process(eid, class, subclass, zcp, zsp);
 
 		zed_conf_write_state(zcp, eid, etime);
 

cmd/zed/zed_exec.c

@@ -63,6 +63,7 @@ static pthread_t _reap_children_tid = (pthread_t)-1;
 static volatile boolean_t _reap_children_stop;
 static avl_tree_t _launched_processes;
 static pthread_mutex_t _launched_processes_lock = PTHREAD_MUTEX_INITIALIZER;
+static int16_t _launched_processes_limit;
 
 /*
  * Create an environment string array for passing to execve() using the
@@ -122,12 +123,18 @@ _zed_exec_fork_child(uint64_t eid, const char *dir, const char *prog,
 	int fd;
 	struct launched_process_node *node;
 	sigset_t mask;
+	struct timespec launch_timeout =
+		{ .tv_sec = 0, .tv_nsec = 200 * 1000 * 1000, };
 
 	assert(dir != NULL);
 	assert(prog != NULL);
 	assert(env != NULL);
 	assert(zfd >= 0);
 
+	while (__atomic_load_n(&_launched_processes_limit,
+	    __ATOMIC_SEQ_CST) <= 0)
+		(void) nanosleep(&launch_timeout, NULL);
+
 	n = snprintf(path, sizeof (path), "%s/%s", dir, prog);
 	if ((n < 0) || (n >= sizeof (path))) {
 		zed_log_msg(LOG_WARNING,
@@ -159,6 +166,7 @@ _zed_exec_fork_child(uint64_t eid, const char *dir, const char *prog,
 
 	/* parent process */
 
+	__atomic_sub_fetch(&_launched_processes_limit, 1, __ATOMIC_SEQ_CST);
 	zed_log_msg(LOG_INFO, "Invoking \"%s\" eid=%llu pid=%d",
 	    prog, eid, pid);
 
@@ -216,6 +224,8 @@ _reap_children(void *arg)
 				free(pnode);
 			}
 			(void) pthread_mutex_unlock(&_launched_processes_lock);
+			__atomic_add_fetch(&_launched_processes_limit, 1,
+			    __ATOMIC_SEQ_CST);
 
 			if (WIFEXITED(status)) {
 				zed_log_msg(LOG_INFO,
@@ -270,20 +280,21 @@ zed_exec_fini(void)
  * Process the event [eid] by synchronously invoking all zedlets with a
  * matching class prefix.
  *
- * Each executable in [zedlets] from the directory [dir] is matched against
- * the event's [class], [subclass], and the "all" class (which matches
- * all events).  Every zedlet with a matching class prefix is invoked.
+ * Each executable in [zcp->zedlets] from the directory [zcp->zedlet_dir]
+ * is matched against the event's [class], [subclass], and the "all" class
+ * (which matches all events).
+ * Every zedlet with a matching class prefix is invoked.
  * The NAME=VALUE strings in [envs] will be passed to the zedlet as
  * environment variables.
  *
- * The file descriptor [zfd] is the zevent_fd used to track the
+ * The file descriptor [zcp->zevent_fd] is the zevent_fd used to track the
  * current cursor location within the zevent nvlist.
  *
  * Return 0 on success, -1 on error.
  */
 int
 zed_exec_process(uint64_t eid, const char *class, const char *subclass,
-    const char *dir, zed_strings_t *zedlets, zed_strings_t *envs, int zfd)
+    struct zed_conf *zcp, zed_strings_t *envs)
 {
 	const char *class_strings[4];
 	const char *allclass = "all";
@@ -292,10 +303,12 @@ zed_exec_process(uint64_t eid, const char *class, const char *subclass,
 	char **e;
 	int n;
 
-	if (!dir || !zedlets || !envs || zfd < 0)
+	if (!zcp->zedlet_dir || !zcp->zedlets || !envs || zcp->zevent_fd < 0)
 		return (-1);
 
 	if (_reap_children_tid == (pthread_t)-1) {
+		_launched_processes_limit = zcp->max_jobs;
+
 		if (pthread_create(&_reap_children_tid, NULL,
 		    _reap_children, NULL) != 0)
 			return (-1);
@@ -321,11 +334,13 @@ zed_exec_process(uint64_t eid, const char *class, const char *subclass,
 
 	e = _zed_exec_create_env(envs);
 
-	for (z = zed_strings_first(zedlets); z; z = zed_strings_next(zedlets)) {
+	for (z = zed_strings_first(zcp->zedlets); z;
+	    z = zed_strings_next(zcp->zedlets)) {
 		for (csp = class_strings; *csp; csp++) {
 			n = strlen(*csp);
 			if ((strncmp(z, *csp, n) == 0) && !isalpha(z[n]))
-				_zed_exec_fork_child(eid, dir, z, e, zfd);
+				_zed_exec_fork_child(eid, zcp->zedlet_dir,
+				    z, e, zcp->zevent_fd);
 		}
 	}
 	free(e);

cmd/zed/zed_exec.h

@@ -17,11 +17,11 @@
 
 #include <stdint.h>
 #include "zed_strings.h"
+#include "zed_conf.h"
 
 void zed_exec_fini(void);
 
 int zed_exec_process(uint64_t eid, const char *class, const char *subclass,
-    const char *dir, zed_strings_t *zedlets, zed_strings_t *envs,
-    int zevent_fd);
+    struct zed_conf *zcp, zed_strings_t *envs);
 
 #endif	/* !ZED_EXEC_H */

man/man8/zed.8.in

@@ -29,6 +29,7 @@ ZED \- ZFS Event Daemon
 [\fB\-p\fR \fIpidfile\fR]
 [\fB\-P\fR \fIpath\fR]
 [\fB\-s\fR \fIstatefile\fR]
+[\fB\-j\fR \fIjobs\fR]
 [\fB\-v\fR]
 [\fB\-V\fR]
 [\fB\-Z\fR]
@@ -95,6 +96,11 @@ it in production!
 .TP
 .BI \-s\  statefile
 Write the daemon's state to the specified file.
+.TP
+.BI \-j\  jobs
+Allow at most \fIjobs\fR ZEDLETs to run concurrently,
+delaying execution of new ones until they finish.
+Defaults to 16.
 .SH ZEVENTS
 .PP
 A zevent is comprised of a list of nvpairs (name/value pairs).  Each zevent

tests/zfs-tests/include/libtest.shlib

@@ -3705,8 +3705,8 @@ function zed_start
 		# run ZED in the background and redirect foreground logging
 		# output to $ZED_LOG.
 		log_must truncate -s 0 $ZED_DEBUG_LOG
-		log_must eval "zed -vF -d $ZEDLET_DIR -p $ZEDLET_DIR/zed.pid -P $PATH" \
-		    "-s $ZEDLET_DIR/state 2>$ZED_LOG &"
+		log_must eval "zed -vF -d $ZEDLET_DIR -P $PATH" \
+		    "-s $ZEDLET_DIR/state -j 1 2>$ZED_LOG &"
 	fi
 
 	return 0
@@ -3722,14 +3722,13 @@ function zed_stop
 	fi
 
 	log_note "Stopping ZED"
-	if [[ -f ${ZEDLET_DIR}/zed.pid ]]; then
-		zedpid=$(<${ZEDLET_DIR}/zed.pid)
-		kill $zedpid
-		while ps -p $zedpid > /dev/null; do
-			sleep 1
-		done
-		rm -f ${ZEDLET_DIR}/zed.pid
-	fi
+	while true; do
+		zedpids="$(pgrep -x zed)"
+		[ "$?" -ne 0 ] && break
+
+		log_must kill $zedpids
+		sleep 1
+	done
 	return 0
 }
 

tests/zfs-tests/tests/functional/events/zed_rc_filter.ksh

@@ -49,6 +49,7 @@ log_assert "Verify zpool sub-commands generate expected events"
 log_onexit cleanup
 
 log_must zpool events -c
+log_must zed_stop
 log_must zed_start
 
 # Backup our zed.rc