Add ActivatingConcurrencyMax for slice startup pacing

Introduce Slice.ActivatingConcurrencyMax to limit how many units
within a slice hierarchy may be in activating state concurrently.

Expose the setting over D-Bus, support transient/property parsing,
enforce it during unit start dispatch, and re-check queued starts when
units leave activating state.

Document the new slice option and add a PID1 concurrency test covering
queued startup behavior.

src/core/dbus-slice.c

@@ -34,6 +34,7 @@ const sd_bus_vtable bus_slice_vtable[] = {
          * systemctl set-property), hence they aren't marked as constant */
         SD_BUS_PROPERTY("ConcurrencyHardMax", "u", bus_property_get_unsigned, offsetof(Slice, concurrency_hard_max), 0),
         SD_BUS_PROPERTY("ConcurrencySoftMax", "u", bus_property_get_unsigned, offsetof(Slice, concurrency_soft_max), 0),
+        SD_BUS_PROPERTY("ActivatingConcurrencyMax", "u", bus_property_get_unsigned, offsetof(Slice, activating_concurrency_max), 0),
         SD_BUS_PROPERTY("NCurrentlyActive", "u", property_get_currently_active, 0, 0),
         SD_BUS_VTABLE_END
 };
@@ -59,6 +60,9 @@ static int bus_slice_set_transient_property(
         if (streq(name, "ConcurrencySoftMax"))
                 return bus_set_transient_unsigned(u, name, &s->concurrency_soft_max, message, flags, reterr_error);
 
+        if (streq(name, "ActivatingConcurrencyMax"))
+                return bus_set_transient_unsigned(u, name, &s->activating_concurrency_max, message, flags, reterr_error);
+
         return 0;
 }
 

src/core/load-fragment-gperf.gperf.in

@@ -634,6 +634,7 @@ Path.TriggerLimitIntervalSec,                 config_parse_sec,
 Path.TriggerLimitBurst,                       config_parse_unsigned,                              0,                                  offsetof(Path, trigger_limit.burst)
 Slice.ConcurrencySoftMax,                     config_parse_concurrency_max,                       0,                                  offsetof(Slice, concurrency_soft_max)
 Slice.ConcurrencyHardMax,                     config_parse_concurrency_max,                       0,                                  offsetof(Slice, concurrency_hard_max)
+Slice.ActivatingConcurrencyMax,               config_parse_concurrency_max,                       0,                                  offsetof(Slice, activating_concurrency_max)
 {{ CGROUP_CONTEXT_CONFIG_ITEMS('Slice') }}
 {{ CGROUP_CONTEXT_CONFIG_ITEMS('Scope') }}
 {{ KILL_CONTEXT_CONFIG_ITEMS('Scope') }}

src/core/slice.c

@@ -28,6 +28,7 @@ static void slice_init(Unit *u) {
         u->ignore_on_isolate = true;
         s->concurrency_hard_max = UINT_MAX;
         s->concurrency_soft_max = UINT_MAX;
+        s->activating_concurrency_max = UINT_MAX;
 }
 
 static void slice_set_state(Slice *s, SliceState state) {
@@ -410,6 +411,25 @@ unsigned slice_get_currently_active(Slice *slice, Unit *ignore, bool with_pendin
         return n;
 }
 
+static unsigned slice_get_currently_activating(Slice *slice, Unit *ignore) {
+        Unit *u = ASSERT_PTR(UNIT(slice));
+        unsigned n = 0;
+        Unit *member;
+
+        UNIT_FOREACH_DEPENDENCY(member, u, UNIT_ATOM_SLICE_OF) {
+                if (member == ignore)
+                        continue;
+
+                if (unit_active_state(member) == UNIT_ACTIVATING)
+                        n++;
+
+                if (member->type == UNIT_SLICE)
+                        n += slice_get_currently_activating(SLICE(member), ignore);
+        }
+
+        return n;
+}
+
 bool slice_concurrency_soft_max_reached(Slice *slice, Unit *ignore) {
         assert(slice);
 
@@ -438,6 +458,20 @@ bool slice_concurrency_hard_max_reached(Slice *slice, Unit *ignore) {
         return false;
 }
 
+bool slice_activating_concurrency_max_reached(Slice *slice, Unit *ignore) {
+        assert(slice);
+
+        if (slice->activating_concurrency_max != UINT_MAX &&
+            slice_get_currently_activating(slice, ignore) >= slice->activating_concurrency_max)
+                return true;
+
+        Unit *parent = UNIT_GET_SLICE(UNIT(slice));
+        if (parent)
+                return slice_activating_concurrency_max_reached(SLICE(parent), ignore);
+
+        return false;
+}
+
 const UnitVTable slice_vtable = {
         .object_size = sizeof(Slice),
         .cgroup_context_offset = offsetof(Slice, cgroup_context),

src/core/slice.h

@@ -12,6 +12,7 @@ typedef struct Slice {
 
         unsigned concurrency_soft_max;
         unsigned concurrency_hard_max;
+        unsigned activating_concurrency_max;
 
         CGroupContext cgroup_context;
 
@@ -26,3 +27,4 @@ unsigned slice_get_currently_active(Slice *slice, Unit *ignore, bool with_pendin
 
 bool slice_concurrency_hard_max_reached(Slice *slice, Unit *ignore);
 bool slice_concurrency_soft_max_reached(Slice *slice, Unit *ignore);
+bool slice_activating_concurrency_max_reached(Slice *slice, Unit *ignore);

src/core/unit.c

@@ -2033,6 +2033,10 @@ int unit_start(Unit *u, ActivationDetails *details) {
                          * the queue */
                         if (slice_concurrency_soft_max_reached(slice, u))
                                 return -EAGAIN; /* Try again, keep in queue */
+
+                        /* Check activating concurrency limit to pace concurrent startups */
+                        if (slice_activating_concurrency_max_reached(slice, u))
+                                return -EAGAIN; /* Try again, keep in queue */
                 }
         }
 
@@ -2713,20 +2717,21 @@ static void unit_check_concurrency_limit(Unit *u) {
         if (!slice)
                 return;
 
-        /* If a unit was stopped, maybe it has pending siblings (or children thereof) that can be started now */
+        /* If a unit was stopped, maybe it has pending siblings (or children thereof) that can be started now.
+         * Walk up the slice hierarchy and re-dispatch for each ancestor that has a limit configured. */
 
-        if (SLICE(slice)->concurrency_soft_max != UINT_MAX) {
-                Unit *sibling;
-                UNIT_FOREACH_DEPENDENCY(sibling, slice, UNIT_ATOM_SLICE_OF) {
-                        if (sibling == u)
-                                continue;
+        for (Unit *s = slice; s; s = UNIT_GET_SLICE(s)) {
+                if (SLICE(s)->concurrency_soft_max != UINT_MAX ||
+                    SLICE(s)->activating_concurrency_max != UINT_MAX) {
+                        Unit *member;
+                        UNIT_FOREACH_DEPENDENCY(member, s, UNIT_ATOM_SLICE_OF) {
+                                if (member == u)
+                                        continue;
 
-                        unit_recursive_add_to_run_queue(sibling);
+                                unit_recursive_add_to_run_queue(member);
+                        }
                 }
         }
-
-        /* Also go up the tree. */
-        unit_check_concurrency_limit(slice);
 }
 
 void unit_notify(Unit *u, UnitActiveState os, UnitActiveState ns, bool reload_success) {
@@ -2882,7 +2887,15 @@ void unit_notify(Unit *u, UnitActiveState os, UnitActiveState ns, bool reload_su
                  * when something BindsTo= to a Type=oneshot unit, as these units go directly from starting to
                  * inactive, without ever entering started.) */
                 unit_submit_to_stop_when_bound_queue(u);
+
+                /* Maybe the activating concurrency limits now allow dispatching of another start job in this slice? */
+                unit_check_concurrency_limit(u);
         }
+
+        /* When a unit leaves the activating state (to deactivating, inactive, or active), it frees up a slot for
+         * ActivatingConcurrencyMax. Re-dispatch queued starts. */
+        if (os == UNIT_ACTIVATING && ns != UNIT_ACTIVATING)
+                unit_check_concurrency_limit(u);
 }
 
 int unit_watch_pidref(Unit *u, const PidRef *pid, bool exclusive) {