Merge pull request #12639 from michaelolbrich/job-order

make the run queue order deterministic

src/basic/unit-def.h

@@ -5,14 +5,17 @@
 
 #include "macro.h"
 
+/* The enum order is used to order unit jobs in the job queue
+ * when other criteria (cpu weight, nice level) are identical.
+ * In this case service units have the hightest priority. */
 typedef enum UnitType {
         UNIT_SERVICE = 0,
+        UNIT_MOUNT,
+        UNIT_SWAP,
         UNIT_SOCKET,
         UNIT_TARGET,
         UNIT_DEVICE,
-        UNIT_MOUNT,
         UNIT_AUTOMOUNT,
-        UNIT_SWAP,
         UNIT_TIMER,
         UNIT_PATH,
         UNIT_SLICE,

src/core/cgroup.c

@@ -3510,6 +3510,45 @@ void manager_invalidate_startup_units(Manager *m) {
                 unit_invalidate_cgroup(u, CGROUP_MASK_CPU|CGROUP_MASK_IO|CGROUP_MASK_BLKIO);
 }
 
+static int unit_get_nice(Unit *u) {
+        ExecContext *ec;
+
+        ec = unit_get_exec_context(u);
+        return ec ? ec->nice : 0;
+}
+
+static uint64_t unit_get_cpu_weight(Unit *u) {
+        ManagerState state = manager_state(u->manager);
+        CGroupContext *cc;
+
+        cc = unit_get_cgroup_context(u);
+        return cc ? cgroup_context_cpu_weight(cc, state) : CGROUP_WEIGHT_DEFAULT;
+}
+
+int compare_job_priority(const void *a, const void *b) {
+        const Job *x = a, *y = b;
+        int nice_x, nice_y;
+        uint64_t weight_x, weight_y;
+        int ret;
+
+        weight_x = unit_get_cpu_weight(x->unit);
+        weight_y = unit_get_cpu_weight(y->unit);
+
+        if ((ret = CMP(weight_y, weight_x)) != 0)
+                return ret;
+
+        nice_x = unit_get_nice(x->unit);
+        nice_y = unit_get_nice(y->unit);
+
+        if ((ret = CMP(nice_x, nice_y)) != 0)
+                return ret;
+
+        if ((ret = CMP(x->unit->type, y->unit->type)) != 0)
+                return ret;
+
+        return strcmp(x->unit->id, y->unit->id);
+}
+
 static const char* const cgroup_device_policy_table[_CGROUP_DEVICE_POLICY_MAX] = {
         [CGROUP_AUTO] = "auto",
         [CGROUP_CLOSED] = "closed",

src/core/cgroup.h

@@ -252,3 +252,5 @@ const char* cgroup_device_policy_to_string(CGroupDevicePolicy i) _const_;
 CGroupDevicePolicy cgroup_device_policy_from_string(const char *s) _pure_;
 
 bool unit_cgroup_delegate(Unit *u);
+
+int compare_job_priority(const void *a, const void *b);

src/core/job.c

@@ -7,6 +7,7 @@
 
 #include "alloc-util.h"
 #include "async.h"
+#include "cgroup.h"
 #include "dbus-job.h"
 #include "dbus.h"
 #include "escape.h"
@@ -73,7 +74,7 @@ void job_unlink(Job *j) {
         assert(!j->object_list);
 
         if (j->in_run_queue) {
-                LIST_REMOVE(run_queue, j->manager->run_queue, j);
+                prioq_remove(j->manager->run_queue, j, &j->run_queue_idx);
                 j->in_run_queue = false;
         }
 
@@ -647,7 +648,7 @@ int job_run_and_invalidate(Job *j) {
         assert(j->type < _JOB_TYPE_MAX_IN_TRANSACTION);
         assert(j->in_run_queue);
 
-        LIST_REMOVE(run_queue, j->manager->run_queue, j);
+        prioq_remove(j->manager->run_queue, j, &j->run_queue_idx);
         j->in_run_queue = false;
 
         if (j->state != JOB_WAITING)
@@ -1146,13 +1147,13 @@ void job_add_to_run_queue(Job *j) {
         if (j->in_run_queue)
                 return;
 
-        if (!j->manager->run_queue) {
+        if (prioq_isempty(j->manager->run_queue)) {
                 r = sd_event_source_set_enabled(j->manager->run_queue_event_source, SD_EVENT_ONESHOT);
                 if (r < 0)
                         log_warning_errno(r, "Failed to enable job run queue event source, ignoring: %m");
         }
 
-        LIST_PREPEND(run_queue, j->manager->run_queue, j);
+        prioq_put(j->manager->run_queue, j, &j->run_queue_idx);
         j->in_run_queue = true;
 }
 

src/core/job.h

@@ -115,7 +115,6 @@ struct Job {
         Unit *unit;
 
         LIST_FIELDS(Job, transaction);
-        LIST_FIELDS(Job, run_queue);
         LIST_FIELDS(Job, dbus_queue);
         LIST_FIELDS(Job, gc_queue);
 
@@ -147,6 +146,8 @@ struct Job {
 
         JobResult result;
 
+        unsigned run_queue_idx;
+
         bool installed:1;
         bool in_run_queue:1;
         bool matters_to_anchor:1;

src/core/manager.c

@@ -814,6 +814,10 @@ int manager_new(UnitFileScope scope, ManagerTestRunFlags test_run_flags, Manager
         if (r < 0)
                 return r;
 
+        r = prioq_ensure_allocated(&m->run_queue, compare_job_priority);
+        if (r < 0)
+                return r;
+
         r = manager_setup_prefix(m);
         if (r < 0)
                 return r;
@@ -1274,7 +1278,7 @@ static void manager_clear_jobs_and_units(Manager *m) {
         manager_dispatch_cleanup_queue(m);
 
         assert(!m->load_queue);
-        assert(!m->run_queue);
+        assert(prioq_isempty(m->run_queue));
         assert(!m->dbus_unit_queue);
         assert(!m->dbus_job_queue);
         assert(!m->cleanup_queue);
@@ -1323,6 +1327,8 @@ Manager* manager_free(Manager *m) {
         hashmap_free(m->watch_pids);
         hashmap_free(m->watch_bus);
 
+        prioq_free(m->run_queue);
+
         set_free(m->startup_units);
         set_free(m->failed_units);
 
@@ -2164,7 +2170,7 @@ static int manager_dispatch_run_queue(sd_event_source *source, void *userdata) {
         assert(source);
         assert(m);
 
-        while ((j = m->run_queue)) {
+        while ((j = prioq_peek(m->run_queue))) {
                 assert(j->installed);
                 assert(j->in_run_queue);
 

src/core/manager.h

@@ -13,6 +13,7 @@
 #include "hashmap.h"
 #include "ip-address-access.h"
 #include "list.h"
+#include "prioq.h"
 #include "ratelimit.h"
 
 struct libmnt_monitor;
@@ -145,7 +146,7 @@ struct Manager {
         LIST_HEAD(Unit, load_queue); /* this is actually more a stack than a queue, but uh. */
 
         /* Jobs that need to be run */
-        LIST_HEAD(Job, run_queue);   /* more a stack than a queue, too */
+        struct Prioq *run_queue;
 
         /* Units and jobs that have not yet been announced via
          * D-Bus. When something about a job changes it is added here