Skip to content

Commit

Permalink
zed: allow limiting concurrent jobs
Browse files Browse the repository at this point in the history
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

Signed-off-by: Ahelenia Ziemiańska <[email protected]>
  • Loading branch information
nabijaczleweli committed Mar 29, 2021
1 parent 4b3de48 commit 3abccd6
Show file tree
Hide file tree
Showing 6 changed files with 50 additions and 13 deletions.
18 changes: 17 additions & 1 deletion cmd/zed/zed_conf.c
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -251,7 +254,7 @@ _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;

if (!zcp || !argv || !argv[0])
Expand Down Expand Up @@ -303,6 +306,19 @@ zed_conf_parse_opts(struct zed_conf *zcp, int argc, char **argv)
case 'Z':
zcp->do_zero = 1;
break;
case 'j': {
unsigned long raw;

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 == '?')
Expand Down
1 change: 1 addition & 0 deletions cmd/zed/zed_conf.h
Original file line number Diff line number Diff line change
Expand Up @@ -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 amount of zedlets to start at a time */
};

struct zed_conf *zed_conf_create(void);
Expand Down
3 changes: 1 addition & 2 deletions cmd/zed/zed_event.c
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down
31 changes: 23 additions & 8 deletions cmd/zed/zed_exec.c
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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";
Expand 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);
Expand All @@ -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);
Expand Down
4 changes: 2 additions & 2 deletions cmd/zed/zed_exec.h
Original file line number Diff line number Diff line change
Expand Up @@ -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 */
6 changes: 6 additions & 0 deletions man/man8/zed.8.in
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down Expand Up @@ -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
Expand Down

0 comments on commit 3abccd6

Please sign in to comment.