udev: do not build full list of dependencies

Suppose the following dependent events are queued:

A(1000) -> C(1002)
                   \
                     --> E(1004)
                   /
B(1001) -> D(1003)

Here, each number is the seqnum of the event, and A -> B means that
event B depends on event A, that is, event B cannot be started until
event A is processed.

In this case, to know if event E can be started, we only need to check
if event C and D are processed, and not necessary to check the state of
event A or B.

However, the commit e1ae931064be9483aa98249294f6e195537f43d1 introduced
full dependency list (in the above example, all A, B, C, D are listed as
the blocker for E), but it is overkill and most information is not necessary.
Also, before the commit e1ae931064be9483aa98249294f6e195537f43d1, we
found blocker from the beginning of the queued events, but that's also
not necessary, as even A is processed, still C may be queued, and we
anyway need to check if C is processed or not.

This makes each event only stores the last blocker event for the event,
and finds the blocker from the end of the queue. With this change, the
memory cost can be reduced from O(n^2) to O(n). Also, as previously we
used Set for managing blockers, but now we only increment/decrement
the reference counter of events, so the speed should be also improved.

Fixes #39583 and #39817.

(cherry picked from commit cb16d47f30cc0d848ca6868c39b844810520d713)

src/udev/udev-manager.c

@@ -53,9 +53,14 @@ typedef enum EventState {
         EVENT_QUEUED,
         EVENT_RUNNING,
         EVENT_LOCKED,
+        EVENT_PROCESSED,
 } EventState;
 
 typedef struct Event {
+        /* All events that have not been processed (state != EVENT_PROCESSED) are referenced by the Manager.
+         * Additionally, an event may be referenced by events blocked by this event. See event_find_blocker(). */
+        unsigned n_ref;
+
         Manager *manager;
         Worker *worker;
         EventState state;
@@ -76,9 +81,8 @@ typedef struct Event {
         char *whole_disk;
         LIST_FIELDS(Event, same_disk);
 
-        bool dependencies_built;
-        Set *blocker_events;
-        Set *blocking_events;
+        /* The last blocker for this event. This event must not be processed before the blocker is processed. */
+        Event *blocker;
 
         LIST_FIELDS(Event, event);
 } Event;
@@ -101,21 +105,6 @@ typedef struct Worker {
         Event *event;
 } Worker;
 
-static void event_clear_dependencies(Event *event) {
-        assert(event);
-
-        Event *e;
-        while ((e = set_steal_first(event->blocker_events)))
-                assert_se(set_remove(e->blocking_events, event) == event);
-        event->blocker_events = set_free(event->blocker_events);
-
-        while ((e = set_steal_first(event->blocking_events)))
-                assert_se(set_remove(e->blocker_events, event) == event);
-        event->blocking_events = set_free(event->blocking_events);
-
-        event->dependencies_built = false;
-}
-
 static void event_unset_whole_disk(Event *event) {
         Manager *manager = ASSERT_PTR(ASSERT_PTR(event)->manager);
 
@@ -140,6 +129,10 @@ static void event_unset_whole_disk(Event *event) {
         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;
@@ -156,14 +149,25 @@ static Event* event_free(Event *event) {
         if (event->worker)
                 event->worker->event = NULL;
 
-        event_clear_dependencies(event);
+        event_unref(event->blocker);
 
         sd_device_unref(event->dev);
 
         return mfree(event);
 }
 
-DEFINE_TRIVIAL_CLEANUP_FUNC(Event*, event_free);
+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);
 
 static Worker* worker_free(Worker *worker) {
         if (!worker)
@@ -176,7 +180,6 @@ static Worker* worker_free(Worker *worker) {
         sd_event_source_unref(worker->timeout_warning_event_source);
         sd_event_source_unref(worker->timeout_kill_event_source);
         pidref_done(&worker->pidref);
-        event_free(worker->event);
 
         return mfree(worker);
 }
@@ -200,8 +203,8 @@ Manager* manager_free(Manager *manager) {
         udev_rules_free(manager->rules);
 
         hashmap_free(manager->workers);
-        while (manager->events)
-                event_free(manager->events);
+        LIST_FOREACH(event, event, manager->events)
+                event_enter_processed(event);
 
         prioq_free(manager->locked_events_by_time);
         hashmap_free(manager->locked_events_by_disk);
@@ -392,9 +395,85 @@ void manager_revert(Manager *manager) {
         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_sigchld(sd_event_source *s, const siginfo_t *si, void *userdata) {
         _cleanup_(worker_freep) Worker *worker = ASSERT_PTR(userdata);
-        sd_device *dev = worker->event ? ASSERT_PTR(worker->event->dev) : NULL;
+        _cleanup_(event_enter_processedp) Event *event = worker_detach_event(worker);
+        sd_device *dev = event ? ASSERT_PTR(event->dev) : NULL;
 
         assert(si);
 
@@ -472,81 +551,6 @@ static int worker_new(Worker **ret, Manager *manager, sd_device_monitor *worker_
         return 0;
 }
 
-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 worker_spawn(Manager *manager, Event *event) {
         int r;
 
@@ -649,17 +653,21 @@ bool devpath_conflict(const char *a, const char *b) {
         return *a == '/' || *b == '/' || *a == *b;
 }
 
-static int event_build_dependencies(Event *event) {
-        int r;
-
+static void event_find_blocker(Event *event) {
         assert(event);
 
         /* lookup event for identical, parent, child device */
 
-        if (event->dependencies_built)
-                return 0;
+        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;
 
-        LIST_FOREACH_BACKWARDS(event, e, event->event_prev) {
                 if (!streq_ptr(event->id, e->id) &&
                     !devpath_conflict(event->devpath, e->devpath) &&
                     !devpath_conflict(event->devpath, e->devpath_old) &&
@@ -667,22 +675,16 @@ static int event_build_dependencies(Event *event) {
                     !(event->devnode && streq_ptr(event->devnode, e->devnode)))
                         continue;
 
-                r = set_ensure_put(&event->blocker_events, NULL, e);
-                if (r < 0)
-                        return r;
-
-                r = set_ensure_put(&e->blocking_events, NULL, event);
-                if (r < 0) {
-                        assert_se(set_remove(event->blocker_events, e) == e);
-                        return r;
-                }
-
                 log_device_debug(event->dev, "SEQNUM=%" PRIu64 " blocked by SEQNUM=%" PRIu64,
                                  event->seqnum, e->seqnum);
+
+                unref_and_replace_full(event->blocker, e, event_ref, event_unref);
+                return;
         }
 
-        event->dependencies_built = true;
-        return 0;
+        /* 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) {
@@ -739,14 +741,10 @@ static int event_queue_start(Manager *manager) {
                 if (event->state != EVENT_QUEUED)
                         continue;
 
-                r = event_build_dependencies(event);
-                if (r < 0)
-                        log_device_warning_errno(event->dev, r,
-                                                 "Failed to check dependencies for event (SEQNUM=%"PRIu64", ACTION=%s), ignoring: %m",
-                                                 event->seqnum, strna(device_action_to_string(event->action)));
+                event_find_blocker(event);
 
                 /* do not start event if parent or child event is still running or queued */
-                if (!set_isempty(event->blocker_events))
+                if (event->blocker)
                         continue;
 
                 r = event_run(event);
@@ -927,11 +925,12 @@ static int event_queue_insert(Manager *manager, sd_device *dev) {
         if (r < 0 && r != -ENOENT)
                 return r;
 
-        _cleanup_(event_freep) Event *event = new(Event, 1);
+        _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,
@@ -943,6 +942,9 @@ static int event_queue_insert(Manager *manager, sd_device *dev) {
                 .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) {
@@ -954,8 +956,9 @@ static int event_queue_insert(Manager *manager, sd_device *dev) {
                                                         "The event (SEQNUM=%"PRIu64") has been already queued.",
                                                         event->seqnum);
 
-                /* Inserting an event in an earlier place may change dependency tree. Let's rebuild it later. */
-                event_clear_dependencies(e);
+                /* 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);
@@ -1211,7 +1214,7 @@ static int on_worker_notify(sd_event_source *s, int fd, uint32_t revents, void *
                 return 0;
         }
 
-        _cleanup_(event_freep) Event *event = worker_detach_event(worker);
+        _cleanup_(event_enter_processedp) Event *event = worker_detach_event(worker);
 
         if (strv_contains(l, "TRY_AGAIN=1")) {
                 /* Worker cannot lock the device. */