SymbX/systemd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a70dbe88d465e7852a57755030a61704c8b4e335
systemd/src/udev/udev-manager.c
T
Yu Watanabe a70dbe88d4 udev: close inotify fd after disabling the corresponding event source 
The event source should be already disabled in manager_exit(),
but for safety.

(cherry picked from commit 02c1be8faf601ec374572446c55c8c5152810fe3)
2026-10-16 17:24:51 +01:00

1542 lines
56 KiB
C
Raw Blame History

/* SPDX-License-Identifier: GPL-2.0-or-later */
#include <poll.h>
#include <stdlib.h>
#include <unistd.h>
#include "sd-varlink.h"
#include "cgroup-setup.h"
#include "cgroup-util.h"
#include "common-signal.h"
#include "daemon-util.h"
#include "device-monitor-private.h"
#include "device-private.h"
#include "device-util.h"
#include "errno-util.h"
#include "event-util.h"
#include "fd-util.h"
#include "format-util.h"
#include "fs-util.h"
#include "hashmap.h"
#include "io-util.h"
#include "list.h"
#include "notify-recv.h"
#include "path-util.h"
#include "pidref.h"
#include "prioq.h"
#include "process-util.h"
#include "selinux-util.h"
#include "set.h"
#include "signal-util.h"
#include "socket-util.h"
#include "string-util.h"
#include "strv.h"
#include "time-util.h"
#include "udev-builtin.h"
#include "udev-config.h"
#include "udev-error.h"
#include "udev-manager.h"
#include "udev-rules.h"
#include "udev-spawn.h"
#include "udev-trace.h"
#include "udev-util.h"
#include "udev-varlink.h"
#include "udev-watch.h"
#include "udev-worker.h"
#define EVENT_REQUEUE_INTERVAL_USEC (200 * USEC_PER_MSEC)
#define EVENT_REQUEUE_TIMEOUT_USEC (3 * USEC_PER_MINUTE)
static void event_unset_whole_disk(Event *event) {
Manager *manager = ASSERT_PTR(ASSERT_PTR(event)->manager);
if (!event->whole_disk)
return;
if (event->same_disk_prev)
/* If this is not the first event, then simply remove this event. */
event->same_disk_prev->same_disk_next = event->same_disk_next;
else if (event->same_disk_next)
/* If this is the first event, replace with the next event. */
assert_se(hashmap_replace(manager->locked_events_by_disk, event->same_disk_next->whole_disk, event->same_disk_next) >= 0);
else
/* Otherwise, remove the entry. */
assert_se(hashmap_remove(manager->locked_events_by_disk, event->whole_disk) == event);
if (event->same_disk_next)
event->same_disk_next->same_disk_prev = event->same_disk_prev;
event->same_disk_prev = event->same_disk_next = NULL;
event->whole_disk = mfree(event->whole_disk);
}
static Event* event_free(Event *event);
DEFINE_PRIVATE_TRIVIAL_REF_UNREF_FUNC(Event, event, event_free);
DEFINE_TRIVIAL_CLEANUP_FUNC(Event*, event_unref);
static Event* event_free(Event *event) {
if (!event)
return NULL;
if (event->manager) {
event_unset_whole_disk(event);
prioq_remove(event->manager->locked_events_by_time, event, &event->locked_event_prioq_index);
if (event->manager->last_event == event)
event->manager->last_event = event->event_prev;
LIST_REMOVE(event, event->manager->events, event);
}
if (event->worker)
event->worker->event = NULL;
event_unref(event->blocker);
sd_device_unref(event->dev);
return mfree(event);
}
static Event* event_enter_processed(Event *event) {
if (!event)
return NULL;
if (event->state == EVENT_PROCESSED)
return NULL;
event->state = EVENT_PROCESSED;
return event_unref(event);
}
DEFINE_TRIVIAL_CLEANUP_FUNC(Event*, event_enter_processed);
Worker* worker_free(Worker *worker) {
if (!worker)
return NULL;
if (worker->manager)
hashmap_remove(worker->manager->workers, &worker->pidref);
/* If an event is still attached, clear its back-pointer so it doesn't dangle (mirrors event_free(),
* which clears the worker's pointer). */
if (worker->event)
worker->event->worker = NULL;
sd_event_source_disable_unref(worker->child_event_source);
sd_event_source_unref(worker->timeout_warning_event_source);
sd_event_source_unref(worker->timeout_kill_event_source);
pidref_done(&worker->pidref);
return mfree(worker);
}
DEFINE_TRIVIAL_CLEANUP_FUNC(Worker*, worker_free);
DEFINE_PRIVATE_HASH_OPS_WITH_VALUE_DESTRUCTOR(
worker_hash_op,
PidRef,
pidref_hash_func,
pidref_compare_func,
Worker,
worker_free);
Manager* manager_free(Manager *manager) {
if (!manager)
return NULL;
udev_builtin_exit();
hashmap_free(manager->properties);
udev_rules_free(manager->rules);
hashmap_free(manager->workers);
LIST_FOREACH(event, event, manager->events)
event_enter_processed(event);
prioq_free(manager->locked_events_by_time);
hashmap_free(manager->locked_events_by_disk);
sd_event_source_unref(manager->requeue_locked_events_timer_event_source);
free(manager->worker_notify_socket_path);
sd_device_monitor_unref(manager->monitor);
sd_varlink_server_unref(manager->varlink_server);
/* udev watch */
sd_event_source_unref(manager->inotify_event);
set_free(manager->synthesize_change_child_event_sources);
safe_close(manager->inotify_fd);
set_free(manager->synthesized_events);
sd_event_source_unref(manager->synthesized_events_clear_event_source);
sd_event_source_unref(manager->kill_workers_event);
sd_event_unref(manager->event);
free(manager->workers_cgroup);
return mfree(manager);
}
Manager* manager_new(void) {
Manager *manager;
manager = new(Manager, 1);
if (!manager)
return NULL;
*manager = (Manager) {
.inotify_fd = -EBADF,
.config_by_udev_conf = UDEV_CONFIG_INIT,
.config_by_command = UDEV_CONFIG_INIT,
.config_by_kernel = UDEV_CONFIG_INIT,
.config_by_control = UDEV_CONFIG_INIT,
.config = UDEV_CONFIG_INIT,
};
return manager;
}
void manager_kill_workers(Manager *manager, int signo) {
assert(manager);
Worker *worker;
HASHMAP_FOREACH(worker, manager->workers) {
worker->state = WORKER_KILLED;
(void) pidref_kill(&worker->pidref, signo);
}
}
static int on_kill_workers_event(sd_event_source *s, uint64_t usec, void *userdata) {
Manager *manager = ASSERT_PTR(userdata);
log_debug("Cleaning up idle workers.");
manager_kill_workers(manager, SIGTERM);
return 0;
}
int manager_reset_kill_workers_timer(Manager *manager) {
int r;
assert(manager);
if (hashmap_isempty(manager->workers)) {
/* There are no workers. Disabling unnecessary timer event source. */
r = sd_event_source_set_enabled(manager->kill_workers_event, SD_EVENT_OFF);
if (r < 0)
return log_warning_errno(r, "Failed to disable timer event source for cleaning up workers: %m");
} else {
r = event_reset_time_relative(
manager->event,
&manager->kill_workers_event,
CLOCK_MONOTONIC,
3 * USEC_PER_SEC,
USEC_PER_SEC,
on_kill_workers_event,
manager,
EVENT_PRIORITY_WORKER_TIMER,
"kill-workers-event",
/* force_reset= */ false);
if (r < 0)
return log_warning_errno(r, "Failed to enable timer event source for cleaning up workers: %m");
}
return 0;
}
void manager_exit(Manager *manager) {
assert(manager);
if (manager->exit)
return;
manager->exit = true;
(void) sd_notify(/* unset_environment= */ false, NOTIFY_STOPPING_MESSAGE);
/* close sources of new events and discard buffered events */
manager->varlink_server = sd_varlink_server_unref(manager->varlink_server);
(void) manager_serialize_config(manager);
/* Disable the event source, but do not close the inotify fd here, as we may still receive
* notification messages about requests to add or remove inotify watches. */
manager->inotify_event = sd_event_source_disable_unref(manager->inotify_event);
/* Disable the device monitor but do not free device monitor, as it may be used when a worker failed,
* and the manager needs to broadcast the kernel event assigned to the worker to libudev listeners.
* Note, here we cannot use sd_device_monitor_stop(), as it changes the multicast group of the socket. */
(void) sd_event_source_set_enabled(sd_device_monitor_get_event_source(manager->monitor), SD_EVENT_OFF);
(void) sd_device_monitor_detach_event(manager->monitor);
/* Kill all workers with SIGTERM, and disable unnecessary timer event source. */
manager_kill_workers(manager, SIGTERM);
manager->kill_workers_event = sd_event_source_disable_unref(manager->kill_workers_event);
(void) event_source_disable(manager->requeue_locked_events_timer_event_source);
}
void notify_ready(Manager *manager) {
int r;
assert(manager);
if (manager->exit)
return;
r = sd_notifyf(/* unset_environment= */ false,
"READY=1\n"
"STATUS=Processing with %u children at max", manager->config.children_max);
if (r < 0)
log_warning_errno(r, "Failed to send readiness notification, ignoring: %m");
}
/* reload requested, HUP signal received, rules changed, builtin changed */
void manager_reload(Manager *manager, bool force) {
_cleanup_(udev_rules_freep) UdevRules *rules = NULL;
usec_t now_usec;
int r;
assert(manager);
if (manager->exit)
return;
assert_se(sd_event_now(manager->event, CLOCK_MONOTONIC, &now_usec) >= 0);
if (!force && now_usec < usec_add(manager->last_usec, 3 * USEC_PER_SEC))
/* check for changed config, every 3 seconds at most */
return;
manager->last_usec = now_usec;
/* Reload SELinux label database, to make the child inherit the up-to-date database. */
mac_selinux_maybe_reload();
UdevReloadFlags flags = udev_builtin_should_reload();
if (udev_rules_should_reload(manager->rules))
flags |= UDEV_RELOAD_RULES | UDEV_RELOAD_KILL_WORKERS;
if (flags == 0 && !force)
/* Neither .rules files nor config files for builtins e.g. .link files changed. It is not
* necessary to reload configs. Note, udev.conf is not checked in the above, hence reloaded
* when explicitly requested or at least one .rules file or friend is updated. */
return;
(void) notify_reloading();
flags |= manager_reload_config(manager);
if (FLAGS_SET(flags, UDEV_RELOAD_KILL_WORKERS))
manager_kill_workers(manager, SIGTERM);
udev_builtin_reload(flags);
if (FLAGS_SET(flags, UDEV_RELOAD_RULES)) {
r = udev_rules_load(&rules, manager->config.resolve_name_timing, /* extra= */ NULL);
if (r < 0)
log_warning_errno(r, "Failed to read udev rules, using the previously loaded rules, ignoring: %m");
else
udev_rules_free_and_replace(manager->rules, rules);
}
notify_ready(manager);
}
void manager_revert(Manager *manager) {
assert(manager);
UdevReloadFlags flags = manager_revert_config(manager);
if (flags == 0)
return;
assert(flags == UDEV_RELOAD_KILL_WORKERS);
manager_kill_workers(manager, SIGTERM);
}
static int on_worker_timeout_kill(sd_event_source *s, uint64_t usec, void *userdata) {
Worker *worker = ASSERT_PTR(userdata);
Manager *manager = ASSERT_PTR(worker->manager);
Event *event = ASSERT_PTR(worker->event);
(void) pidref_kill_and_sigcont(&worker->pidref, manager->config.timeout_signal);
worker->state = WORKER_KILLED;
log_device_error(event->dev, "Worker ["PID_FMT"] processing SEQNUM=%"PRIu64" killed.", worker->pidref.pid, event->seqnum);
return 0;
}
static int on_worker_timeout_warning(sd_event_source *s, uint64_t usec, void *userdata) {
Worker *worker = ASSERT_PTR(userdata);
Event *event = ASSERT_PTR(worker->event);
log_device_warning(event->dev, "Worker ["PID_FMT"] processing SEQNUM=%"PRIu64" is taking a long time.", worker->pidref.pid, event->seqnum);
return 0;
}
static void worker_attach_event(Worker *worker, Event *event) {
Manager *manager = ASSERT_PTR(ASSERT_PTR(worker)->manager);
assert(event);
assert(event->state == EVENT_QUEUED);
assert(!event->worker);
assert(IN_SET(worker->state, WORKER_UNDEF, WORKER_IDLE));
assert(!worker->event);
worker->state = WORKER_RUNNING;
worker->event = event;
event->state = EVENT_RUNNING;
event->worker = worker;
(void) event_reset_time_relative(
manager->event,
&worker->timeout_warning_event_source,
CLOCK_MONOTONIC,
udev_warn_timeout(manager->config.timeout_usec),
USEC_PER_SEC,
on_worker_timeout_warning,
worker,
EVENT_PRIORITY_WORKER_TIMER,
"worker-timeout-warn",
/* force_reset= */ true);
(void) event_reset_time_relative(
manager->event,
&worker->timeout_kill_event_source,
CLOCK_MONOTONIC,
manager_kill_worker_timeout(manager),
USEC_PER_SEC,
on_worker_timeout_kill,
worker,
EVENT_PRIORITY_WORKER_TIMER,
"worker-timeout-kill",
/* force_reset= */ true);
}
static Event* worker_detach_event(Worker *worker) {
assert(worker);
Event *event = TAKE_PTR(worker->event);
if (event)
assert_se(TAKE_PTR(event->worker) == worker);
if (worker->state != WORKER_KILLED)
worker->state = WORKER_IDLE;
(void) event_source_disable(worker->timeout_warning_event_source);
(void) event_source_disable(worker->timeout_kill_event_source);
return event;
}
static int on_worker_exit(sd_event_source *s, const siginfo_t *si, void *userdata) {
_cleanup_(worker_freep) Worker *worker = ASSERT_PTR(userdata);
_cleanup_(event_enter_processedp) Event *event = worker_detach_event(worker);
sd_device *dev = event ? ASSERT_PTR(event->dev) : NULL;
assert(si);
switch (si->si_code) {
case CLD_EXITED:
if (si->si_status == 0) {
log_device_debug(dev, "Worker ["PID_FMT"] exited.", si->si_pid);
return 0;
}
log_device_warning(dev, "Worker ["PID_FMT"] exited with return code %i.",
si->si_pid, si->si_status);
if (!dev)
return 0;
(void) device_add_exit_status(dev, si->si_status);
break;
case CLD_KILLED:
case CLD_DUMPED:
log_device_warning(dev, "Worker ["PID_FMT"] terminated by signal %i (%s).",
si->si_pid, si->si_status, signal_to_string(si->si_status));
if (!dev)
return 0;
(void) device_add_signal(dev, si->si_status);
break;
default:
assert_not_reached();
}
(void) device_broadcast_on_error(dev, worker->manager->monitor);
return 0;
}
static int worker_new(Worker **ret, Manager *manager, sd_device_monitor *worker_monitor, PidRef *pidref) {
_cleanup_(worker_freep) Worker *worker = NULL;
int r;
assert(ret);
assert(manager);
assert(worker_monitor);
assert(pidref);
/* This takes and invalidates pidref even on some error cases. */
worker = new(Worker, 1);
if (!worker)
return -ENOMEM;
*worker = (Worker) {
.pidref = TAKE_PIDREF(*pidref),
};
r = device_monitor_get_address(worker_monitor, &worker->address);
if (r < 0)
return r;
r = event_add_child_pidref(manager->event, &worker->child_event_source, &worker->pidref, WEXITED, on_worker_exit, worker);
if (r < 0)
return r;
r = sd_event_source_set_priority(worker->child_event_source, EVENT_PRIORITY_WORKER_EXIT);
if (r < 0)
return r;
r = hashmap_ensure_put(&manager->workers, &worker_hash_op, &worker->pidref, worker);
if (r < 0)
return r;
worker->manager = manager;
*ret = TAKE_PTR(worker);
return 0;
}
static int worker_spawn(Manager *manager, Event *event) {
int r;
assert(manager);
assert(event);
/* listen for new events */
_cleanup_(sd_device_monitor_unrefp) sd_device_monitor *worker_monitor = NULL;
r = device_monitor_new_full(&worker_monitor, MONITOR_GROUP_NONE, -EBADF);
if (r < 0)
return r;
(void) sd_device_monitor_set_description(worker_monitor, "worker");
/* allow the main daemon netlink address to send devices to the worker */
r = device_monitor_allow_unicast_sender(worker_monitor, manager->monitor);
if (r < 0)
return log_error_errno(r, "Worker: Failed to set unicast sender: %m");
pid_t manager_pid = getpid_cached();
_cleanup_(pidref_done) PidRef pidref = PIDREF_NULL;
r = pidref_safe_fork("(udev-worker)", FORK_DEATHSIG_SIGTERM, &pidref);
if (r < 0) {
event->state = EVENT_QUEUED;
return log_error_errno(r, "Failed to fork() worker: %m");
}
if (r == 0) {
_cleanup_(udev_worker_done) UdevWorker w = {
.monitor = TAKE_PTR(worker_monitor),
.properties = TAKE_PTR(manager->properties),
.rules = TAKE_PTR(manager->rules),
.config = manager->config,
.manager_pid = manager_pid,
};
if (manager->workers_cgroup) {
r = cg_attach(manager->workers_cgroup, /* pid= */ 0);
if (r < 0) {
log_error_errno(r, "Failed to move worker into cgroup '%s': %m", manager->workers_cgroup);
_exit(EXIT_FAILURE);
}
}
if (setenv("NOTIFY_SOCKET", manager->worker_notify_socket_path, /* overwrite= */ true) < 0) {
log_error_errno(errno, "Failed to set $NOTIFY_SOCKET: %m");
_exit(EXIT_FAILURE);
}
/* Worker process */
r = udev_worker_main(&w, event->dev);
log_close();
_exit(r < 0 ? EXIT_FAILURE : EXIT_SUCCESS);
}
Worker *worker;
r = worker_new(&worker, manager, worker_monitor, &pidref);
if (r < 0)
return log_error_errno(r, "Failed to create worker object: %m");
worker_attach_event(worker, event);
log_device_debug(event->dev, "Worker ["PID_FMT"] is forked for processing SEQNUM=%"PRIu64".", worker->pidref.pid, event->seqnum);
return 0;
}
static int event_run(Event *event) {
Manager *manager = ASSERT_PTR(ASSERT_PTR(event)->manager);
int r;
log_device_uevent(event->dev, "Device ready for processing");
Worker *worker;
HASHMAP_FOREACH(worker, manager->workers) {
if (worker->state != WORKER_IDLE)
continue;
r = device_monitor_send(manager->monitor, &worker->address, event->dev);
if (r < 0) {
log_device_error_errno(event->dev, r, "Worker ["PID_FMT"] did not accept message, killing the worker: %m",
worker->pidref.pid);
(void) pidref_kill(&worker->pidref, SIGKILL);
worker->state = WORKER_KILLED;
continue;
}
worker_attach_event(worker, event);
return 0;
}
/* No idle worker could accept the event. If we already reached the worker limit, e.g. because
* we just killed the only idle worker above, leave the event queued and wait for SIGCHLD of an
* exiting worker to free up a slot. on_post() will retry processing the queue. */
if (hashmap_size(manager->workers) >= manager->config.children_max)
return 0;
/* start new worker and pass initial device */
r = worker_spawn(manager, event);
if (r < 0)
return r;
return 0;
}
bool devpath_conflict(const char *a, const char *b) {
/* This returns true when two paths are equivalent, or one is a child of another. */
if (!a || !b)
return false;
for (; *a != '\0' && *b != '\0'; a++, b++)
if (*a != *b)
return false;
return *a == '/' || *b == '/' || *a == *b;
}
static void event_find_blocker(Event *event) {
assert(event);
/* lookup event for identical, parent, child device */
if (event->blocker && event->blocker->state != EVENT_PROCESSED)
/* Previously found blocker is not processed yet. */
return;
/* If we have not found blocker yet, or the previously found blocker has been processed, let's find
* (another) blocker for this event. */
LIST_FOREACH_BACKWARDS(event, e, (event->blocker ?: event)->event_prev) {
if (e->state == EVENT_PROCESSED)
continue;
if (!streq_ptr(event->id, e->id) &&
!devpath_conflict(event->devpath, e->devpath) &&
!devpath_conflict(event->devpath, e->devpath_old) &&
!devpath_conflict(event->devpath_old, e->devpath) &&
!(event->devnode && streq_ptr(event->devnode, e->devnode)))
continue;
log_device_debug(event->dev, "SEQNUM=%" PRIu64 " blocked by SEQNUM=%" PRIu64,
event->seqnum, e->seqnum);
unref_and_replace_new_ref(event->blocker, e, event_ref, event_unref);
return;
}
/* No new blocker is found, and if set, previously found blocker has been processed. Clear the
* previous blocker if set. */
event->blocker = event_unref(event->blocker);
}
static bool manager_can_process_event(Manager *manager) {
static bool children_max_reached_logged = false;
assert(manager);
/* Check if there is a free room for processing an event. */
if (hashmap_size(manager->workers) < manager->config.children_max)
goto yes_we_can; /* new worker can be spawned */
Worker *worker;
HASHMAP_FOREACH(worker, manager->workers)
if (worker->state == WORKER_IDLE)
goto yes_we_can; /* found an idle worker */
/* Avoid spamming the debug logs if the limit is already reached and
* many events still need to be processed */
if (!children_max_reached_logged) {
log_debug("Maximum number (%u) of children reached.", hashmap_size(manager->workers));
children_max_reached_logged = true;
}
return false;
yes_we_can:
/* Re-enable the debug message for the next batch of events */
children_max_reached_logged = false;
return true;
}
static int event_queue_start(Manager *manager) {
int r;
assert(manager);
assert(!manager->exit);
if (!manager->events || manager->stop_exec_queue)
return 0;
r = event_source_disable(manager->kill_workers_event);
if (r < 0)
log_warning_errno(r, "Failed to disable event source for cleaning up idle workers, ignoring: %m");
manager_reload(manager, /* force= */ false);
/* manager_reload() may kill idle workers, hence we may not be possible to start processing an event.
* Let's check that and return earlier if we cannot. */
if (!manager_can_process_event(manager))
return 0;
LIST_FOREACH(event, event, manager->events) {
if (event->state != EVENT_QUEUED)
continue;
event_find_blocker(event);
/* do not start event if parent or child event is still running or queued */
if (event->blocker)
continue;
r = event_run(event);
if (r < 0)
return r;
/* A worker is activated now. Let's check if we can process more events. */
if (!manager_can_process_event(manager))
break;
}
return 0;
}
static int on_requeue_locked_events(sd_event_source *s, uint64_t usec, void *userdata) {
/* This does nothing. The on_post() callback will requeue locked events. */
return 1;
}
static int manager_requeue_locked_events(Manager *manager) {
usec_t now_usec = 0;
int r;
assert(manager);
for (;;) {
Event *event = prioq_peek(manager->locked_events_by_time);
if (!event)
return event_source_disable(manager->requeue_locked_events_timer_event_source);
if (now_usec == 0) {
r = sd_event_now(manager->event, CLOCK_MONOTONIC, &now_usec);
if (r < 0)
return r;
}
if (event->requeue_next_usec > now_usec)
return event_reset_time(
manager->event,
&manager->requeue_locked_events_timer_event_source,
CLOCK_MONOTONIC,
event->requeue_next_usec,
USEC_PER_SEC,
on_requeue_locked_events,
/* userdata= */ NULL,
EVENT_PRIORITY_REQUEUE_EVENT,
"requeue-locked-events",
/* force_reset= */ true);
assert_se(prioq_pop(manager->locked_events_by_time) == event);
event_unset_whole_disk(event);
event->state = EVENT_QUEUED;
}
}
int manager_requeue_locked_events_by_device(Manager *manager, sd_device *dev) {
int r;
/* When a new event for a block device is queued or we get an inotify event, assume that the
* device is not locked anymore. The assumption may not be true, but that should not cause any
* issues, as in that case events will be requeued soon. */
if (hashmap_isempty(manager->locked_events_by_disk))
return 0;
const char *devname;
r = udev_get_whole_disk(dev, NULL, &devname);
if (r <= 0)
return r;
Event *first = hashmap_remove(manager->locked_events_by_disk, devname);
if (!first)
return 0;
Event *event;
while ((event = LIST_POP(same_disk, first))) {
assert_se(prioq_remove(manager->locked_events_by_time, event, &event->locked_event_prioq_index) > 0);
event->whole_disk = mfree(event->whole_disk);
event->state = EVENT_QUEUED;
}
return 0;
}
static int locked_event_compare(const Event *x, const Event *y) {
return CMP(x->requeue_next_usec, y->requeue_next_usec);
}
static int event_enter_locked(Event *event, const char *whole_disk) {
Manager *manager = ASSERT_PTR(ASSERT_PTR(event)->manager);
sd_device *dev = ASSERT_PTR(event->dev);
usec_t now_usec;
int r;
/* add a short delay to suppress busy loop */
r = sd_event_now(manager->event, CLOCK_MONOTONIC, &now_usec);
if (r < 0)
return log_device_warning_errno(
dev, r,
"Failed to get current time, skipping event (SEQNUM=%"PRIu64", ACTION=%s): %m",
event->seqnum, strna(device_action_to_string(event->action)));
if (event->requeue_timeout_usec > 0 && event->requeue_timeout_usec <= now_usec)
return log_device_warning_errno(
dev, SYNTHETIC_ERRNO(ETIMEDOUT),
"The underlying block device is locked by a process more than %s, skipping event (SEQNUM=%"PRIu64", ACTION=%s).",
FORMAT_TIMESPAN(EVENT_REQUEUE_TIMEOUT_USEC, USEC_PER_MINUTE),
event->seqnum, strna(device_action_to_string(event->action)));
event->requeue_next_usec = usec_add(now_usec, EVENT_REQUEUE_INTERVAL_USEC);
if (event->requeue_timeout_usec == 0)
event->requeue_timeout_usec = usec_add(now_usec, EVENT_REQUEUE_TIMEOUT_USEC);
if (isempty(whole_disk))
return log_device_warning_errno(
dev, SYNTHETIC_ERRNO(EBADMSG),
"Unexpected notify message received, skipping event (SEQNUM=%"PRIu64", ACTION=%s).",
event->seqnum, strna(device_action_to_string(event->action)));
_cleanup_free_ char *whole_disk_copy = strdup(whole_disk);
if (!whole_disk_copy)
return log_oom();
Event *first = hashmap_get(manager->locked_events_by_disk, whole_disk_copy);
LIST_PREPEND(same_disk, first, event);
r = hashmap_ensure_replace(&manager->locked_events_by_disk, &path_hash_ops, whole_disk_copy, first);
if (r < 0) {
LIST_REMOVE(same_disk, first, event);
return log_oom();
}
event->whole_disk = TAKE_PTR(whole_disk_copy);
r = prioq_ensure_put(&manager->locked_events_by_time, locked_event_compare, event, &event->locked_event_prioq_index);
if (r < 0)
return log_oom();
event->state = EVENT_LOCKED;
return 0;
}
static int event_queue_insert(Manager *manager, sd_device *dev) {
const char *devpath, *devpath_old = NULL, *id = NULL, *devnode = NULL;
sd_device_action_t action;
uint64_t seqnum;
int r;
assert(manager);
assert(dev);
/* If the event is the one we triggered, remove the UUID from the list. */
sd_id128_t uuid;
if (sd_device_get_trigger_uuid(dev, &uuid) >= 0)
free(set_remove(manager->synthesized_events, &uuid));
/* We only accepts devices received by device monitor. */
r = sd_device_get_seqnum(dev, &seqnum);
if (r < 0)
return r;
r = sd_device_get_action(dev, &action);
if (r < 0)
return r;
r = sd_device_get_devpath(dev, &devpath);
if (r < 0)
return r;
r = sd_device_get_property_value(dev, "DEVPATH_OLD", &devpath_old);
if (r < 0 && r != -ENOENT)
return r;
r = sd_device_get_device_id(dev, &id);
if (r < 0 && r != -ENOENT)
return r;
r = sd_device_get_devname(dev, &devnode);
if (r < 0 && r != -ENOENT)
return r;
_cleanup_(event_unrefp) Event *event = new(Event, 1);
if (!event)
return -ENOMEM;
*event = (Event) {
.n_ref = 1,
.dev = sd_device_ref(dev),
.seqnum = seqnum,
.action = action,
.id = id,
.devpath = devpath,
.devpath_old = devpath_old,
.devnode = devnode,
.state = EVENT_QUEUED,
.locked_event_prioq_index = PRIOQ_IDX_NULL,
};
/* The kernel sometimes sends events in a wrong order, and we may receive an event with smaller
* SEQNUM after one with larger SEQNUM. To workaround the issue, let's reorder events if necessary. */
Event *prev = NULL;
LIST_FOREACH_BACKWARDS(event, e, manager->last_event) {
if (e->seqnum < event->seqnum) {
prev = e;
break;
}
if (e->seqnum == event->seqnum)
return log_device_warning_errno(dev, SYNTHETIC_ERRNO(EALREADY),
"The event (SEQNUM=%"PRIu64") has been already queued.",
event->seqnum);
/* The inserted event may be a blocker of an already queued event, hence the already found
* blocker may not be the last one. Let's find the last blocker again later. */
e->blocker = event_unref(e->blocker);
}
LIST_INSERT_AFTER(event, manager->events, prev, event);
if (prev == manager->last_event)
manager->last_event = event;
else
log_device_debug(dev, "Unordered event is received (last queued event seqnum=%"PRIu64", newly received event seqnum=%"PRIu64"), reordering.",
manager->last_event->seqnum, event->seqnum);
event->manager = manager;
TAKE_PTR(event);
log_device_uevent(dev, "Device is queued");
return 0;
}
static int manager_serialize_events(Manager *manager) {
int r;
assert(manager);
_cleanup_(sd_device_monitor_unrefp) sd_device_monitor *storage = NULL;
r = device_monitor_new_full(&storage, MONITOR_GROUP_NONE, -EBADF);
if (r < 0)
return log_warning_errno(r, "Failed to create new device monitor instance: %m");
union sockaddr_union a;
r = device_monitor_get_address(storage, &a);
if (r < 0)
return log_warning_errno(r, "Failed to get address of device monitor socket: %m");
uint64_t n = 0;
LIST_FOREACH(event, event, manager->events) {
if (event->state != EVENT_QUEUED)
continue;
r = device_monitor_send(storage, &a, event->dev);
if (r < 0) {
log_device_warning_errno(event->dev, r, "Failed to save event to socket storage, ignoring: %m");
continue;
}
n++;
}
if (n == 0)
return 0;
r = notify_push_fd(sd_device_monitor_get_fd(storage), "event-serialization");
if (r < 0)
return log_warning_errno(r, "Failed to push event serialization fd to service manager: %m");
log_debug("Serialized %"PRIu64" events.", n);
return 0;
}
static int manager_deserialize_events(Manager *manager, int *fd) {
int r;
assert(manager);
assert(fd);
assert(*fd >= 0);
/* This may take and invalidate passed file descriptor even on failure. */
/* At this stage, we have not receive any events from the kernel, hence should be empty. */
if (manager->events)
return log_warning_errno(SYNTHETIC_ERRNO(EALREADY), "Received multiple event storage socket (%i).", *fd);
r = sd_is_socket(*fd, AF_NETLINK, SOCK_RAW, /* listening= */ -1);
if (r < 0)
return log_warning_errno(r, "Failed to verify type of event storage socket (%i): %m", *fd);
if (r == 0)
return log_warning_errno(SYNTHETIC_ERRNO(EINVAL), "Received invalid event storage socket (%i).", *fd);
_cleanup_(sd_device_monitor_unrefp) sd_device_monitor *storage = NULL;
r = device_monitor_new_full(&storage, MONITOR_GROUP_NONE, *fd);
if (r < 0)
return log_warning_errno(r, "Failed to initialize event storage: %m");
TAKE_FD(*fd);
r = device_monitor_allow_unicast_sender(storage, storage);
if (r < 0)
return log_warning_errno(r, "Failed to set trusted sender for event storage: %m");
uint64_t n = 0;
for (;;) {
r = fd_wait_for_event(sd_device_monitor_get_fd(storage), POLLIN, 0);
if (r == -EINTR)
continue;
if (r < 0)
return log_warning_errno(r, "Failed to wait for event from event storage: %m");
if (r == 0)
break;
_cleanup_(sd_device_unrefp) sd_device *dev = NULL;
r = sd_device_monitor_receive(storage, &dev);
if (r < 0) {
log_warning_errno(r, "Failed to receive device from event storage, ignoring: %m");
continue;
}
r = event_queue_insert(manager, dev);
if (r < 0) {
log_device_warning_errno(dev, r, "Failed to insert device into event queue, ignoring: %m");
continue;
}
n++;
}
if (n > 0)
(void) manager_create_queue_file(manager);
log_debug("Deserialized %"PRIu64" events.", n);
return 0;
}
static int on_uevent(sd_device_monitor *monitor, sd_device *dev, void *userdata) {
Manager *manager = ASSERT_PTR(userdata);
int r;
DEVICE_TRACE_POINT(kernel_uevent_received, dev);
(void) manager_create_queue_file(manager);
device_ensure_usec_initialized(dev, NULL);
r = event_queue_insert(manager, dev);
if (r < 0) {
log_device_error_errno(dev, r, "Failed to insert device into event queue: %m");
return 1;
}
(void) manager_requeue_locked_events_by_device(manager, dev);
return 1;
}
static int manager_init_device_monitor(Manager *manager, int fd) {
int r;
assert(manager);
/* This takes passed file descriptor on success. */
if (fd >= 0) {
if (manager->monitor)
return log_warning_errno(SYNTHETIC_ERRNO(EALREADY), "Received multiple netlink socket (%i), ignoring.", fd);
r = sd_is_socket(fd, AF_NETLINK, SOCK_RAW, /* listening= */ -1);
if (r < 0)
return log_warning_errno(r, "Failed to verify socket type of %i, ignoring: %m", fd);
if (r == 0)
return log_warning_errno(SYNTHETIC_ERRNO(EINVAL), "Received invalid netlink socket (%i), ignoring.", fd);
} else {
if (manager->monitor)
return 0;
}
r = device_monitor_new_full(&manager->monitor, MONITOR_GROUP_KERNEL, fd);
if (r < 0)
return log_error_errno(r, "Failed to initialize device monitor: %m");
return 0;
}
static int manager_start_device_monitor(Manager *manager) {
int r;
assert(manager);
r = manager_init_device_monitor(manager, -EBADF);
if (r < 0)
return r;
(void) sd_device_monitor_set_description(manager->monitor, "manager");
r = sd_device_monitor_attach_event(manager->monitor, manager->event);
if (r < 0)
return log_error_errno(r, "Failed to attach event to device monitor: %m");
r = sd_device_monitor_start(manager->monitor, on_uevent, manager);
if (r < 0)
return log_error_errno(r, "Failed to start device monitor: %m");
r = sd_event_source_set_priority(sd_device_monitor_get_event_source(manager->monitor), EVENT_PRIORITY_DEVICE_MONITOR);
if (r < 0)
return log_error_errno(r, "Failed to set priority to device monitor: %m");
return 0;
}
static int on_worker_notify(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
Manager *manager = ASSERT_PTR(userdata);
int r;
assert(fd >= 0);
_cleanup_(pidref_done) PidRef sender = PIDREF_NULL;
_cleanup_strv_free_ char **l = NULL;
r = notify_recv_strv(fd, &l, /* ret_ucred= */ NULL, &sender);
if (r == -EAGAIN)
return 0;
if (r < 0)
return r;
/* lookup worker who sent the signal */
Worker *worker = hashmap_get(manager->workers, &sender);
if (!worker) {
log_warning("Received notify datagram of unknown process ["PID_FMT"], ignoring.", sender.pid);
return 0;
}
if (strv_contains(l, "INOTIFY_WATCH_ADD=1")) {
assert(worker->event);
r = manager_add_watch(manager, worker->event->dev);
if (ERRNO_IS_NEG_DEVICE_ABSENT(r))
r = 0;
if (r < 0)
log_device_warning_errno(worker->event->dev, r, "Failed to add inotify watch, ignoring: %m");
/* Send the result back to the worker process. */
r = pidref_sigqueue(&sender, SIGUSR1, r);
if (r < 0) {
log_device_warning_errno(worker->event->dev, r,
"Failed to send signal to worker process ["PID_FMT"], killing the worker process: %m",
sender.pid);
(void) pidref_kill(&sender, SIGTERM);
worker->state = WORKER_KILLED;
}
return 0;
}
if (strv_contains(l, "INOTIFY_WATCH_REMOVE=1")) {
assert(worker->event);
r = manager_remove_watch(manager, worker->event->dev);
if (r < 0)
log_device_warning_errno(worker->event->dev, r, "Failed to remove inotify watch, ignoring: %m");
/* Send the result back to the worker process. */
r = pidref_sigqueue(&sender, SIGUSR1, r);
if (r < 0) {
log_device_warning_errno(worker->event->dev, r,
"Failed to send signal to worker process ["PID_FMT"], killing the worker process: %m",
sender.pid);
(void) pidref_kill(&sender, SIGTERM);
worker->state = WORKER_KILLED;
}
return 0;
}
_cleanup_(event_enter_processedp) Event *event = worker_detach_event(worker);
if (strv_contains(l, "TRY_AGAIN=1")) {
/* Worker cannot lock the device. */
r = event_enter_locked(event, strv_find_startswith(l, "WHOLE_DISK="));
if (r < 0) {
(void) device_add_errno(event->dev, r);
(void) device_broadcast_on_error(event->dev, manager->monitor);
} else
TAKE_PTR(event);
}
return 0;
}
static int manager_start_worker_notify(Manager *manager) {
int r;
assert(manager);
assert(manager->event);
r = notify_socket_prepare(
manager->event,
EVENT_PRIORITY_WORKER_NOTIFY,
on_worker_notify,
manager,
&manager->worker_notify_socket_path);
if (r < 0)
return log_error_errno(r, "Failed to prepare worker notification socket: %m");
return 0;
}
static int on_sigterm(sd_event_source *s, const struct signalfd_siginfo *si, void *userdata) {
Manager *manager = ASSERT_PTR(userdata);
manager_exit(manager);
return 1;
}
static int on_sighup(sd_event_source *s, const struct signalfd_siginfo *si, void *userdata) {
Manager *manager = ASSERT_PTR(userdata);
manager_reload(manager, /* force= */ true);
return 1;
}
int manager_create_queue_file(Manager *manager) {
int r;
assert(manager);
if (manager->queue_file_created)
return 0;
r = touch("/run/udev/queue");
if (r < 0)
return log_warning_errno(r, "Failed to touch /run/udev/queue: %m");
manager->queue_file_created = true;
return 0;
}
static int manager_unlink_queue_file(Manager *manager) {
assert(manager);
if (manager->events)
return 0; /* There are queued events. */
if (!set_isempty(manager->synthesize_change_child_event_sources))
return 0; /* There are child processes that should trigger synthetic events. */
if (!set_isempty(manager->synthesized_events))
return 0; /* We have triggered synthesized change events. */
/* There are no queued events. Let's remove /run/udev/queue and clean up the idle processes. */
if (unlink("/run/udev/queue") < 0) {
if (errno != ENOENT)
return log_warning_errno(errno, "Failed to unlink /run/udev/queue: %m");
} else
log_debug("No events are queued, removed /run/udev/queue.");
manager->queue_file_created = false;
return 0;
}
static int on_post_exit(Manager *manager) {
assert(manager);
assert(manager->exit);
LIST_FOREACH(event, event, manager->events)
if (event->state == EVENT_RUNNING)
return 0; /* There still exist events being processed. */
(void) manager_unlink_queue_file(manager);
if (!hashmap_isempty(manager->workers))
return 0; /* There still exist running workers. */
(void) manager_serialize_events(manager);
udev_watch_dump();
return sd_event_exit(manager->event, 0);
}
static int on_post(sd_event_source *s, void *userdata) {
Manager *manager = ASSERT_PTR(userdata);
if (manager->exit)
return on_post_exit(manager);
if (manager->events) {
(void) manager_requeue_locked_events(manager);
/* Try to process pending events if idle workers exist. Why is this necessary?
* When a worker finished an event and became idle, even if there was a pending event,
* the corresponding device might have been locked and the processing of the event
* delayed for a while, preventing the worker from processing the event immediately.
* Now, the device may be unlocked. Let's try again! */
(void) event_queue_start(manager);
return 0;
}
/* Check whether there are uevents not received yet. If so, we will soon receive them. Do not remove
* the queue file and do not kill workers. */
if (sd_event_source_get_pending(sd_device_monitor_get_event_source(manager->monitor)) != 0)
return 0;
/* We are currently synthesizing change uevents, hence new uevents may come soon. */
if (!set_isempty(manager->synthesize_change_child_event_sources))
return 0;
(void) manager_unlink_queue_file(manager);
(void) manager_reset_kill_workers_timer(manager);
if (!hashmap_isempty(manager->workers))
return 0; /* There still exist idle workers. */
if (manager->workers_cgroup)
/* cleanup possible left-over processes in the workers cgroup */
if (cg_kill_kernel_sigkill(manager->workers_cgroup, /* ret_n_pids_killed= */ NULL) == -EOPNOTSUPP)
(void) cg_kill(manager->workers_cgroup, SIGKILL, CGROUP_IGNORE_SELF, /* killed_pids= */ NULL, /* log_kill= */ NULL, /* userdata= */ NULL);
return 0;
}
static int manager_setup_signal(
Manager *manager,
sd_event *event,
int signal,
sd_event_signal_handler_t handler,
int64_t priority,
const char *description) {
_cleanup_(sd_event_source_unrefp) sd_event_source *s = NULL;
int r;
assert(manager);
assert(event);
r = sd_event_add_signal(event, &s, signal | SD_EVENT_SIGNAL_PROCMASK, handler, manager);
if (r < 0)
return r;
r = sd_event_source_set_priority(s, priority);
if (r < 0)
return r;
(void) sd_event_source_set_description(s, description);
r = sd_event_source_set_floating(s, true);
if (r < 0)
return r;
return 0;
}
static int manager_setup_event(Manager *manager) {
_cleanup_(sd_event_unrefp) sd_event *e = NULL;
int r;
assert(manager);
r = sd_event_default(&e);
if (r < 0)
return log_error_errno(r, "Failed to allocate event loop: %m");
r = manager_setup_signal(manager, e, SIGINT, on_sigterm, EVENT_PRIORITY_SIGTERM, "sigint-event-source");
if (r < 0)
return log_error_errno(r, "Failed to create SIGINT event source: %m");
r = manager_setup_signal(manager, e, SIGTERM, on_sigterm, EVENT_PRIORITY_SIGTERM, "sigterm-event-source");
if (r < 0)
return log_error_errno(r, "Failed to create SIGTERM event source: %m");
r = manager_setup_signal(manager, e, SIGHUP, on_sighup, EVENT_PRIORITY_SIGHUP, "sighup-event-source");
if (r < 0)
return log_error_errno(r, "Failed to create SIGHUP event source: %m");
_cleanup_(sd_event_source_unrefp) sd_event_source *s = NULL;
r = sd_event_add_post(e, &s, on_post, manager);
if (r < 0)
return log_error_errno(r, "Failed to create post event source: %m");
r = sd_event_source_set_priority(s, EVENT_PRIORITY_POST);
if (r < 0)
return log_error_errno(r, "Failed to set priority to post event source: %m");
(void) sd_event_source_set_description(s, "post-event-source");
r = sd_event_source_set_floating(s, true);
if (r < 0)
return log_error_errno(r, "Failed to set floating to post event source: %m");
/* Eventually, we probably want to do more here on memory pressure, for example, kill idle workers immediately */
r = sd_event_add_memory_pressure(e, /* ret= */ NULL, /* callback= */ NULL, /* userdata= */ NULL);
if (r < 0)
log_full_errno(ERRNO_IS_NOT_SUPPORTED(r) || ERRNO_IS_PRIVILEGE(r) || (r == -EHOSTDOWN) ? LOG_DEBUG : LOG_WARNING, r,
"Failed to allocate memory pressure watch, ignoring: %m");
r = sd_event_add_signal(e, /* ret= */ NULL,
(SIGRTMIN+18) | SD_EVENT_SIGNAL_PROCMASK, sigrtmin18_handler, /* userdata= */ NULL);
if (r < 0)
return log_error_errno(r, "Failed to allocate SIGRTMIN+18 event source, ignoring: %m");
r = sd_event_set_watchdog(e, true);
if (r < 0)
return log_error_errno(r, "Failed to create watchdog event source: %m");
manager->event = TAKE_PTR(e);
return 0;
}
static int manager_listen_fds(Manager *manager, int *ret_varlink_fd) {
_cleanup_strv_free_ char **names = NULL;
int varlink_fd = -EBADF;
int r;
assert(manager);
assert(ret_varlink_fd);
int n = sd_listen_fds_with_names(/* unset_environment= */ true, &names);
if (n < 0)
return log_error_errno(n, "Failed to listen on fds: %m");
for (int i = 0; i < n; i++) {
int fd = SD_LISTEN_FDS_START + i;
if (streq(names[i], "varlink")) {
varlink_fd = fd;
r = 0;
} else if (streq(names[i], "systemd-udevd-kernel.socket"))
r = manager_init_device_monitor(manager, fd);
else if (streq(names[i], "inotify"))
r = manager_init_inotify(manager, fd);
else if (streq(names[i], "config-serialization"))
r = manager_deserialize_config(manager, &fd);
else if (streq(names[i], "event-serialization"))
r = manager_deserialize_events(manager, &fd);
else
r = log_debug_errno(SYNTHETIC_ERRNO(EINVAL),
"Received unexpected fd (%s), ignoring.", names[i]);
if (r < 0)
close_and_notify_warn(fd, names[i]);
}
*ret_varlink_fd = varlink_fd;
return 0;
}
int manager_main(Manager *manager) {
_cleanup_close_ int varlink_fd = -EBADF;
int r;
assert(manager);
_cleanup_free_ char *cgroup = NULL;
r = cg_pid_get_path(/* pid= */ 0, &cgroup);
if (r < 0)
log_debug_errno(r, "Failed to get cgroup, ignoring: %m");
else if (endswith(cgroup, "/udev")) { /* If we are in a subcgroup /udev/ we assume it was delegated to us */
log_debug("Running in delegated subcgroup '%s'.", cgroup);
/* Try to create a sibling 'workers' cgroup and spawn all workers inside it, so that we can
* use cgroup.kill to atomically clear all workers. */
_cleanup_free_ char *workers_cgroup = NULL;
r = path_extract_directory(cgroup, &workers_cgroup);
if (r < 0)
return log_warning_errno(r, "Failed to extract parent of cgroup '%s': %m", cgroup);
if (!path_extend(&workers_cgroup, "workers"))
return log_oom();
r = cg_create(workers_cgroup);
if (r < 0)
log_warning_errno(r, "Failed to create workers cgroup '%s', ignoring: %m", workers_cgroup);
else {
log_debug("Running workers in delegated subcgroup '%s'.", workers_cgroup);
manager->workers_cgroup = TAKE_PTR(workers_cgroup);
}
}
r = manager_setup_event(manager);
if (r < 0)
return r;
r = manager_listen_fds(manager, &varlink_fd);
if (r < 0)
return r;
r = manager_start_varlink_server(manager, TAKE_FD(varlink_fd));
if (r < 0)
return r;
r = manager_start_device_monitor(manager);
if (r < 0)
return r;
r = manager_start_inotify(manager);
if (r < 0)
return r;
r = manager_start_worker_notify(manager);
if (r < 0)
return r;
manager->last_usec = now(CLOCK_MONOTONIC);
udev_builtin_init();
r = udev_rules_load(&manager->rules, manager->config.resolve_name_timing, /* extra= */ NULL);
if (r < 0)
return log_error_errno(r, "Failed to read udev rules: %m");
r = udev_rules_apply_static_dev_perms(manager->rules);
if (r < 0)
log_warning_errno(r, "Failed to apply permissions on static device nodes, ignoring: %m");
_unused_ _cleanup_(notify_on_cleanup) const char *notify_message =
notify_start(NOTIFY_READY_MESSAGE, NOTIFY_STOPPING_MESSAGE);
/* We will start processing events in the loop below. Before starting processing, let's remove the
* event serialization fd from the fdstore, to avoid retrieving the serialized events again in future
* invocations. Otherwise, the serialized events may be processed multiple times. */
(void) notify_remove_fd_warn("event-serialization");
r = sd_event_loop(manager->event);
if (r < 0)
return log_error_errno(r, "Event loop failed: %m");
return 0;
}
Reference in New Issue View Git Blame Copy Permalink