job: make the run queue order deterministic

Jobs are added to the run queue in random order. This happens because most
jobs are added by iterating over the transaction or dependency hash maps.

As a result, jobs that can be executed at the same time are started in a
different order each time.
On small embedded devices this can cause a measurable jitter for the point
in time when a job starts (~100ms jitter for 10 units that are started in
random order).
This results is a similar jitter for the boot time. This is undesirable in
general and make optimizing the boot time a lot harder.
Also, jobs that should have a higher priority because the unit has a higher
CPU weight might get executed later than others.

Fix this by turning the job run_queue into a Prioq and sort by the
following criteria (use the next if the values are equal):
- CPU weight
- nice level
- unit type
- unit name

The last one is just there for deterministic sorting to avoid any jitter.

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