daandemeyer / rpms / systemd

Forked from rpms/systemd 2 years ago
Clone
21255d
From 395eb7753a9772f505102fbbe3ba3261b57abbe9 Mon Sep 17 00:00:00 2001
21255d
From: Lennart Poettering <lennart@poettering.net>
21255d
Date: Mon, 23 Nov 2020 18:02:40 +0100
21255d
Subject: [PATCH] sd-event: add ability to ratelimit event sources
21255d
21255d
Let's a concept of "rate limiting" to event sources: if specific event
21255d
sources fire too often in some time interval temporarily take them
21255d
offline, and take them back online once the interval passed.
21255d
21255d
This is a simple scheme of avoiding starvation of event sources if some
21255d
event source fires too often.
21255d
21255d
This introduces the new conceptual states of "offline" and "online" for
21255d
event sources: an event source is "online" only when enabled *and* not
21255d
ratelimited, and offline in all other cases. An event source that is
21255d
online hence has its fds registered in the epoll, its signals in the
21255d
signalfd and so on.
21255d
21255d
(cherry picked from commit b6d5481b3d9f7c9b1198ab54b54326ec73e855bf)
21255d
21255d
Related: #1819868
21255d
---
21255d
 src/basic/ratelimit.h              |   8 +
21255d
 src/libsystemd/libsystemd.sym      |   7 +
21255d
 src/libsystemd/sd-event/sd-event.c | 433 +++++++++++++++++++++++------
21255d
 src/systemd/sd-event.h             |   3 +
21255d
 4 files changed, 369 insertions(+), 82 deletions(-)
21255d
21255d
diff --git a/src/basic/ratelimit.h b/src/basic/ratelimit.h
21255d
index de91def28d..0012b49935 100644
21255d
--- a/src/basic/ratelimit.h
21255d
+++ b/src/basic/ratelimit.h
21255d
@@ -38,3 +38,11 @@ typedef struct RateLimit {
21255d
         } while (false)
21255d
 
21255d
 bool ratelimit_below(RateLimit *r);
21255d
+
21255d
+static inline void ratelimit_reset(RateLimit *rl) {
21255d
+        rl->num = rl->begin = 0;
21255d
+}
21255d
+
21255d
+static inline bool ratelimit_configured(RateLimit *rl) {
21255d
+        return rl->interval > 0 && rl->burst > 0;
21255d
+}
21255d
diff --git a/src/libsystemd/libsystemd.sym b/src/libsystemd/libsystemd.sym
21255d
index 778e88a16c..149d2e7b82 100644
21255d
--- a/src/libsystemd/libsystemd.sym
21255d
+++ b/src/libsystemd/libsystemd.sym
21255d
@@ -572,3 +572,10 @@ global:
21255d
         sd_bus_enqueue_for_read;
21255d
         sd_event_source_disable_unref;
21255d
 } LIBSYSTEMD_238;
21255d
+
21255d
+LIBSYSTEMD_248 {
21255d
+global:
21255d
+        sd_event_source_set_ratelimit;
21255d
+        sd_event_source_get_ratelimit;
21255d
+        sd_event_source_is_ratelimited;
21255d
+} LIBSYSTEMD_239;
21255d
diff --git a/src/libsystemd/sd-event/sd-event.c b/src/libsystemd/sd-event/sd-event.c
21255d
index d18ce28a92..be912d94e3 100644
21255d
--- a/src/libsystemd/sd-event/sd-event.c
21255d
+++ b/src/libsystemd/sd-event/sd-event.c
21255d
@@ -19,6 +19,7 @@
21255d
 #include "missing.h"
21255d
 #include "prioq.h"
21255d
 #include "process-util.h"
21255d
+#include "ratelimit.h"
21255d
 #include "set.h"
21255d
 #include "signal-util.h"
21255d
 #include "string-table.h"
21255d
@@ -46,6 +47,7 @@ typedef enum EventSourceType {
21255d
         _SOURCE_EVENT_SOURCE_TYPE_INVALID = -1
21255d
 } EventSourceType;
21255d
 
21255d
+
21255d
 static const char* const event_source_type_table[_SOURCE_EVENT_SOURCE_TYPE_MAX] = {
21255d
         [SOURCE_IO] = "io",
21255d
         [SOURCE_TIME_REALTIME] = "realtime",
21255d
@@ -76,7 +78,25 @@ typedef enum WakeupType {
21255d
         _WAKEUP_TYPE_INVALID = -1,
21255d
 } WakeupType;
21255d
 
21255d
-#define EVENT_SOURCE_IS_TIME(t) IN_SET((t), SOURCE_TIME_REALTIME, SOURCE_TIME_BOOTTIME, SOURCE_TIME_MONOTONIC, SOURCE_TIME_REALTIME_ALARM, SOURCE_TIME_BOOTTIME_ALARM)
21255d
+#define EVENT_SOURCE_IS_TIME(t)                 \
21255d
+        IN_SET((t),                             \
21255d
+               SOURCE_TIME_REALTIME,            \
21255d
+               SOURCE_TIME_BOOTTIME,            \
21255d
+               SOURCE_TIME_MONOTONIC,           \
21255d
+               SOURCE_TIME_REALTIME_ALARM,      \
21255d
+               SOURCE_TIME_BOOTTIME_ALARM)
21255d
+
21255d
+#define EVENT_SOURCE_CAN_RATE_LIMIT(t)          \
21255d
+        IN_SET((t),                             \
21255d
+               SOURCE_IO,                       \
21255d
+               SOURCE_TIME_REALTIME,            \
21255d
+               SOURCE_TIME_BOOTTIME,            \
21255d
+               SOURCE_TIME_MONOTONIC,           \
21255d
+               SOURCE_TIME_REALTIME_ALARM,      \
21255d
+               SOURCE_TIME_BOOTTIME_ALARM,      \
21255d
+               SOURCE_SIGNAL,                   \
21255d
+               SOURCE_DEFER,                    \
21255d
+               SOURCE_INOTIFY)
21255d
 
21255d
 struct inode_data;
21255d
 
21255d
@@ -96,6 +116,7 @@ struct sd_event_source {
21255d
         bool pending:1;
21255d
         bool dispatching:1;
21255d
         bool floating:1;
21255d
+        bool ratelimited:1;
21255d
 
21255d
         int64_t priority;
21255d
         unsigned pending_index;
21255d
@@ -107,6 +128,10 @@ struct sd_event_source {
21255d
 
21255d
         LIST_FIELDS(sd_event_source, sources);
21255d
 
21255d
+        RateLimit rate_limit;
21255d
+
21255d
+        /* These are primarily fields relevant for time event sources, but since any event source can
21255d
+         * effectively become one when rate-limited, this is part of the common fields. */
21255d
         unsigned earliest_index;
21255d
         unsigned latest_index;
21255d
 
21255d
@@ -266,7 +291,7 @@ struct sd_event {
21255d
         Hashmap *signal_data; /* indexed by priority */
21255d
 
21255d
         Hashmap *child_sources;
21255d
-        unsigned n_enabled_child_sources;
21255d
+        unsigned n_online_child_sources;
21255d
 
21255d
         Set *post_sources;
21255d
 
21255d
@@ -311,12 +336,23 @@ static thread_local sd_event *default_event = NULL;
21255d
 static void source_disconnect(sd_event_source *s);
21255d
 static void event_gc_inode_data(sd_event *e, struct inode_data *d);
21255d
 
21255d
+static bool event_source_is_online(sd_event_source *s) {
21255d
+        assert(s);
21255d
+        return s->enabled != SD_EVENT_OFF && !s->ratelimited;
21255d
+}
21255d
+
21255d
+static bool event_source_is_offline(sd_event_source *s) {
21255d
+        assert(s);
21255d
+        return s->enabled == SD_EVENT_OFF || s->ratelimited;
21255d
+}
21255d
+
21255d
 static sd_event *event_resolve(sd_event *e) {
21255d
         return e == SD_EVENT_DEFAULT ? default_event : e;
21255d
 }
21255d
 
21255d
 static int pending_prioq_compare(const void *a, const void *b) {
21255d
         const sd_event_source *x = a, *y = b;
21255d
+        int r;
21255d
 
21255d
         assert(x->pending);
21255d
         assert(y->pending);
21255d
@@ -327,23 +363,23 @@ static int pending_prioq_compare(const void *a, const void *b) {
21255d
         if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
21255d
                 return 1;
21255d
 
21255d
+        /* Non rate-limited ones first. */
21255d
+        r = CMP(!!x->ratelimited, !!y->ratelimited);
21255d
+        if (r != 0)
21255d
+                return r;
21255d
+
21255d
         /* Lower priority values first */
21255d
-        if (x->priority < y->priority)
21255d
-                return -1;
21255d
-        if (x->priority > y->priority)
21255d
-                return 1;
21255d
+        r = CMP(x->priority, y->priority);
21255d
+        if (r != 0)
21255d
+                return r;
21255d
 
21255d
         /* Older entries first */
21255d
-        if (x->pending_iteration < y->pending_iteration)
21255d
-                return -1;
21255d
-        if (x->pending_iteration > y->pending_iteration)
21255d
-                return 1;
21255d
-
21255d
-        return 0;
21255d
+        return CMP(x->pending_iteration, y->pending_iteration);
21255d
 }
21255d
 
21255d
 static int prepare_prioq_compare(const void *a, const void *b) {
21255d
         const sd_event_source *x = a, *y = b;
21255d
+        int r;
21255d
 
21255d
         assert(x->prepare);
21255d
         assert(y->prepare);
21255d
@@ -354,29 +390,46 @@ static int prepare_prioq_compare(const void *a, const void *b) {
21255d
         if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
21255d
                 return 1;
21255d
 
21255d
+        /* Non rate-limited ones first. */
21255d
+        r = CMP(!!x->ratelimited, !!y->ratelimited);
21255d
+        if (r != 0)
21255d
+                return r;
21255d
+
21255d
         /* Move most recently prepared ones last, so that we can stop
21255d
          * preparing as soon as we hit one that has already been
21255d
          * prepared in the current iteration */
21255d
-        if (x->prepare_iteration < y->prepare_iteration)
21255d
-                return -1;
21255d
-        if (x->prepare_iteration > y->prepare_iteration)
21255d
-                return 1;
21255d
+        r = CMP(x->prepare_iteration, y->prepare_iteration);
21255d
+        if (r != 0)
21255d
+                return r;
21255d
 
21255d
         /* Lower priority values first */
21255d
-        if (x->priority < y->priority)
21255d
-                return -1;
21255d
-        if (x->priority > y->priority)
21255d
-                return 1;
21255d
+        return CMP(x->priority, y->priority);
21255d
+}
21255d
 
21255d
-        return 0;
21255d
+static usec_t time_event_source_next(const sd_event_source *s) {
21255d
+        assert(s);
21255d
+
21255d
+        /* We have two kinds of event sources that have elapsation times associated with them: the actual
21255d
+         * time based ones and the ones for which a ratelimit can be in effect (where we want to be notified
21255d
+         * once the ratelimit time window ends). Let's return the next elapsing time depending on what we are
21255d
+         * looking at here. */
21255d
+
21255d
+        if (s->ratelimited) { /* If rate-limited the next elapsation is when the ratelimit time window ends */
21255d
+                assert(s->rate_limit.begin != 0);
21255d
+                assert(s->rate_limit.interval != 0);
21255d
+                return usec_add(s->rate_limit.begin, s->rate_limit.interval);
21255d
+        }
21255d
+
21255d
+        /* Otherwise this must be a time event source, if not ratelimited */
21255d
+        if (EVENT_SOURCE_IS_TIME(s->type))
21255d
+                return s->time.next;
21255d
+
21255d
+        return USEC_INFINITY;
21255d
 }
21255d
 
21255d
 static int earliest_time_prioq_compare(const void *a, const void *b) {
21255d
         const sd_event_source *x = a, *y = b;
21255d
 
21255d
-        assert(EVENT_SOURCE_IS_TIME(x->type));
21255d
-        assert(x->type == y->type);
21255d
-
21255d
         /* Enabled ones first */
21255d
         if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
21255d
                 return -1;
21255d
@@ -390,24 +443,30 @@ static int earliest_time_prioq_compare(const void *a, const void *b) {
21255d
                 return 1;
21255d
 
21255d
         /* Order by time */
21255d
-        if (x->time.next < y->time.next)
21255d
-                return -1;
21255d
-        if (x->time.next > y->time.next)
21255d
-                return 1;
21255d
-
21255d
-        return 0;
21255d
+        return CMP(time_event_source_next(x), time_event_source_next(y));
21255d
 }
21255d
 
21255d
 static usec_t time_event_source_latest(const sd_event_source *s) {
21255d
-        return usec_add(s->time.next, s->time.accuracy);
21255d
+        assert(s);
21255d
+
21255d
+        if (s->ratelimited) { /* For ratelimited stuff the earliest and the latest time shall actually be the
21255d
+                               * same, as we should avoid adding additional inaccuracy on an inaccuracy time
21255d
+                               * window */
21255d
+                assert(s->rate_limit.begin != 0);
21255d
+                assert(s->rate_limit.interval != 0);
21255d
+                return usec_add(s->rate_limit.begin, s->rate_limit.interval);
21255d
+        }
21255d
+
21255d
+        /* Must be a time event source, if not ratelimited */
21255d
+        if (EVENT_SOURCE_IS_TIME(s->type))
21255d
+                return usec_add(s->time.next, s->time.accuracy);
21255d
+
21255d
+        return USEC_INFINITY;
21255d
 }
21255d
 
21255d
 static int latest_time_prioq_compare(const void *a, const void *b) {
21255d
         const sd_event_source *x = a, *y = b;
21255d
 
21255d
-        assert(EVENT_SOURCE_IS_TIME(x->type));
21255d
-        assert(x->type == y->type);
21255d
-
21255d
         /* Enabled ones first */
21255d
         if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
21255d
                 return -1;
21255d
@@ -852,12 +911,12 @@ static void event_gc_signal_data(sd_event *e, const int64_t *priority, int sig)
21255d
          * the signalfd for it. */
21255d
 
21255d
         if (sig == SIGCHLD &&
21255d
-            e->n_enabled_child_sources > 0)
21255d
+            e->n_online_child_sources > 0)
21255d
                 return;
21255d
 
21255d
         if (e->signal_sources &&
21255d
             e->signal_sources[sig] &&
21255d
-            e->signal_sources[sig]->enabled != SD_EVENT_OFF)
21255d
+            event_source_is_online(e->signal_sources[sig]))
21255d
                 return;
21255d
 
21255d
         /*
21255d
@@ -904,11 +963,17 @@ static void event_source_time_prioq_reshuffle(sd_event_source *s) {
21255d
         struct clock_data *d;
21255d
 
21255d
         assert(s);
21255d
-        assert(EVENT_SOURCE_IS_TIME(s->type));
21255d
 
21255d
         /* Called whenever the event source's timer ordering properties changed, i.e. time, accuracy,
21255d
          * pending, enable state. Makes sure the two prioq's are ordered properly again. */
21255d
-        assert_se(d = event_get_clock_data(s->event, s->type));
21255d
+
21255d
+        if (s->ratelimited)
21255d
+                d = &s->event->monotonic;
21255d
+        else {
21255d
+                assert(EVENT_SOURCE_IS_TIME(s->type));
21255d
+                assert_se(d = event_get_clock_data(s->event, s->type));
21255d
+        }
21255d
+
21255d
         prioq_reshuffle(d->earliest, s, &s->earliest_index);
21255d
         prioq_reshuffle(d->latest, s, &s->latest_index);
21255d
         d->needs_rearm = true;
21255d
@@ -949,12 +1014,18 @@ static void source_disconnect(sd_event_source *s) {
21255d
         case SOURCE_TIME_BOOTTIME:
21255d
         case SOURCE_TIME_MONOTONIC:
21255d
         case SOURCE_TIME_REALTIME_ALARM:
21255d
-        case SOURCE_TIME_BOOTTIME_ALARM: {
21255d
-                struct clock_data *d;
21255d
-                assert_se(d = event_get_clock_data(s->event, s->type));
21255d
-                event_source_time_prioq_remove(s, d);
21255d
+        case SOURCE_TIME_BOOTTIME_ALARM:
21255d
+                /* Only remove this event source from the time event source here if it is not ratelimited. If
21255d
+                 * it is ratelimited, we'll remove it below, separately. Why? Because the clock used might
21255d
+                 * differ: ratelimiting always uses CLOCK_MONOTONIC, but timer events might use any clock */
21255d
+
21255d
+                if (!s->ratelimited) {
21255d
+                        struct clock_data *d;
21255d
+                        assert_se(d = event_get_clock_data(s->event, s->type));
21255d
+                        event_source_time_prioq_remove(s, d);
21255d
+                }
21255d
+
21255d
                 break;
21255d
-        }
21255d
 
21255d
         case SOURCE_SIGNAL:
21255d
                 if (s->signal.sig > 0) {
21255d
@@ -969,9 +1040,9 @@ static void source_disconnect(sd_event_source *s) {
21255d
 
21255d
         case SOURCE_CHILD:
21255d
                 if (s->child.pid > 0) {
21255d
-                        if (s->enabled != SD_EVENT_OFF) {
21255d
-                                assert(s->event->n_enabled_child_sources > 0);
21255d
-                                s->event->n_enabled_child_sources--;
21255d
+                        if (event_source_is_online(s)) {
21255d
+                                assert(s->event->n_online_child_sources > 0);
21255d
+                                s->event->n_online_child_sources--;
21255d
                         }
21255d
 
21255d
                         (void) hashmap_remove(s->event->child_sources, PID_TO_PTR(s->child.pid));
21255d
@@ -1037,6 +1108,9 @@ static void source_disconnect(sd_event_source *s) {
21255d
         if (s->prepare)
21255d
                 prioq_remove(s->event->prepare, s, &s->prepare_index);
21255d
 
21255d
+        if (s->ratelimited)
21255d
+                event_source_time_prioq_remove(s, &s->event->monotonic);
21255d
+
21255d
         event = s->event;
21255d
 
21255d
         s->type = _SOURCE_EVENT_SOURCE_TYPE_INVALID;
21255d
@@ -1458,11 +1532,11 @@ _public_ int sd_event_add_child(
21255d
                 return r;
21255d
         }
21255d
 
21255d
-        e->n_enabled_child_sources++;
21255d
+        e->n_online_child_sources++;
21255d
 
21255d
         r = event_make_signal_data(e, SIGCHLD, NULL);
21255d
         if (r < 0) {
21255d
-                e->n_enabled_child_sources--;
21255d
+                e->n_online_child_sources--;
21255d
                 source_free(s);
21255d
                 return r;
21255d
         }
21255d
@@ -2079,7 +2153,7 @@ _public_ int sd_event_source_set_io_fd(sd_event_source *s, int fd) {
21255d
         if (s->io.fd == fd)
21255d
                 return 0;
21255d
 
21255d
-        if (s->enabled == SD_EVENT_OFF) {
21255d
+        if (event_source_is_offline(s)) {
21255d
                 s->io.fd = fd;
21255d
                 s->io.registered = false;
21255d
         } else {
21255d
@@ -2146,7 +2220,7 @@ _public_ int sd_event_source_set_io_events(sd_event_source *s, uint32_t events)
21255d
         if (r < 0)
21255d
                 return r;
21255d
 
21255d
-        if (s->enabled != SD_EVENT_OFF) {
21255d
+        if (event_source_is_online(s)) {
21255d
                 r = source_io_register(s, s->enabled, events);
21255d
                 if (r < 0)
21255d
                         return r;
21255d
@@ -2249,7 +2323,7 @@ _public_ int sd_event_source_set_priority(sd_event_source *s, int64_t priority)
21255d
 
21255d
                 event_gc_inode_data(s->event, old_inode_data);
21255d
 
21255d
-        } else if (s->type == SOURCE_SIGNAL && s->enabled != SD_EVENT_OFF) {
21255d
+        } else if (s->type == SOURCE_SIGNAL && event_source_is_online(s)) {
21255d
                 struct signal_data *old, *d;
21255d
 
21255d
                 /* Move us from the signalfd belonging to the old
21255d
@@ -2296,20 +2370,29 @@ _public_ int sd_event_source_get_enabled(sd_event_source *s, int *ret) {
21255d
         return s->enabled != SD_EVENT_OFF;
21255d
 }
21255d
 
21255d
-static int event_source_disable(sd_event_source *s) {
21255d
+static int event_source_offline(
21255d
+                sd_event_source *s,
21255d
+                int enabled,
21255d
+                bool ratelimited) {
21255d
+
21255d
+        bool was_offline;
21255d
         int r;
21255d
 
21255d
         assert(s);
21255d
-        assert(s->enabled != SD_EVENT_OFF);
21255d
+        assert(enabled == SD_EVENT_OFF || ratelimited);
21255d
 
21255d
         /* Unset the pending flag when this event source is disabled */
21255d
-        if (!IN_SET(s->type, SOURCE_DEFER, SOURCE_EXIT)) {
21255d
+        if (s->enabled != SD_EVENT_OFF &&
21255d
+            enabled == SD_EVENT_OFF &&
21255d
+            !IN_SET(s->type, SOURCE_DEFER, SOURCE_EXIT)) {
21255d
                 r = source_set_pending(s, false);
21255d
                 if (r < 0)
21255d
                         return r;
21255d
         }
21255d
 
21255d
-        s->enabled = SD_EVENT_OFF;
21255d
+        was_offline = event_source_is_offline(s);
21255d
+        s->enabled = enabled;
21255d
+        s->ratelimited = ratelimited;
21255d
 
21255d
         switch (s->type) {
21255d
 
21255d
@@ -2330,8 +2413,10 @@ static int event_source_disable(sd_event_source *s) {
21255d
                 break;
21255d
 
21255d
         case SOURCE_CHILD:
21255d
-                assert(s->event->n_enabled_child_sources > 0);
21255d
-                s->event->n_enabled_child_sources--;
21255d
+                if (!was_offline) {
21255d
+                        assert(s->event->n_online_child_sources > 0);
21255d
+                        s->event->n_online_child_sources--;
21255d
+                }
21255d
 
21255d
                 event_gc_signal_data(s->event, &s->priority, SIGCHLD);
21255d
                 break;
21255d
@@ -2349,26 +2434,42 @@ static int event_source_disable(sd_event_source *s) {
21255d
                 assert_not_reached("Wut? I shouldn't exist.");
21255d
         }
21255d
 
21255d
-        return 0;
21255d
+        return 1;
21255d
 }
21255d
 
21255d
-static int event_source_enable(sd_event_source *s, int enable) {
21255d
+static int event_source_online(
21255d
+                sd_event_source *s,
21255d
+                int enabled,
21255d
+                bool ratelimited) {
21255d
+
21255d
+        bool was_online;
21255d
         int r;
21255d
 
21255d
         assert(s);
21255d
-        assert(IN_SET(enable, SD_EVENT_ON, SD_EVENT_ONESHOT));
21255d
-        assert(s->enabled == SD_EVENT_OFF);
21255d
+        assert(enabled != SD_EVENT_OFF || !ratelimited);
21255d
 
21255d
         /* Unset the pending flag when this event source is enabled */
21255d
-        if (!IN_SET(s->type, SOURCE_DEFER, SOURCE_EXIT)) {
21255d
+        if (s->enabled == SD_EVENT_OFF &&
21255d
+            enabled != SD_EVENT_OFF &&
21255d
+            !IN_SET(s->type, SOURCE_DEFER, SOURCE_EXIT)) {
21255d
                 r = source_set_pending(s, false);
21255d
                 if (r < 0)
21255d
                         return r;
21255d
         }
21255d
 
21255d
+        /* Are we really ready for onlining? */
21255d
+        if (enabled == SD_EVENT_OFF || ratelimited) {
21255d
+                /* Nope, we are not ready for onlining, then just update the precise state and exit */
21255d
+                s->enabled = enabled;
21255d
+                s->ratelimited = ratelimited;
21255d
+                return 0;
21255d
+        }
21255d
+
21255d
+        was_online = event_source_is_online(s);
21255d
+
21255d
         switch (s->type) {
21255d
         case SOURCE_IO:
21255d
-                r = source_io_register(s, enable, s->io.events);
21255d
+                r = source_io_register(s, enabled, s->io.events);
21255d
                 if (r < 0)
21255d
                         return r;
21255d
                 break;
21255d
@@ -2386,13 +2487,13 @@ static int event_source_enable(sd_event_source *s, int enable) {
21255d
                 r = event_make_signal_data(s->event, SIGCHLD, NULL);
21255d
                 if (r < 0) {
21255d
                         s->enabled = SD_EVENT_OFF;
21255d
-                        s->event->n_enabled_child_sources--;
21255d
+                        s->event->n_online_child_sources--;
21255d
                         event_gc_signal_data(s->event, &s->priority, SIGCHLD);
21255d
                         return r;
21255d
                 }
21255d
 
21255d
-                s->event->n_enabled_child_sources++;
21255d
-
21255d
+                if (!was_online)
21255d
+                        s->event->n_online_child_sources++;
21255d
                 break;
21255d
 
21255d
         case SOURCE_TIME_REALTIME:
21255d
@@ -2410,7 +2511,8 @@ static int event_source_enable(sd_event_source *s, int enable) {
21255d
                 assert_not_reached("Wut? I shouldn't exist.");
21255d
         }
21255d
 
21255d
-        s->enabled = enable;
21255d
+        s->enabled = enabled;
21255d
+        s->ratelimited = ratelimited;
21255d
 
21255d
         /* Non-failing operations below */
21255d
         switch (s->type) {
21255d
@@ -2430,7 +2532,7 @@ static int event_source_enable(sd_event_source *s, int enable) {
21255d
                 break;
21255d
         }
21255d
 
21255d
-        return 0;
21255d
+        return 1;
21255d
 }
21255d
 
21255d
 _public_ int sd_event_source_set_enabled(sd_event_source *s, int m) {
21255d
@@ -2448,7 +2550,7 @@ _public_ int sd_event_source_set_enabled(sd_event_source *s, int m) {
21255d
                 return 0;
21255d
 
21255d
         if (m == SD_EVENT_OFF)
21255d
-                r = event_source_disable(s);
21255d
+                r = event_source_offline(s, m, s->ratelimited);
21255d
         else {
21255d
                 if (s->enabled != SD_EVENT_OFF) {
21255d
                         /* Switching from "on" to "oneshot" or back? If that's the case, we can take a shortcut, the
21255d
@@ -2457,7 +2559,7 @@ _public_ int sd_event_source_set_enabled(sd_event_source *s, int m) {
21255d
                         return 0;
21255d
                 }
21255d
 
21255d
-                r = event_source_enable(s, m);
21255d
+                r = event_source_online(s, m, s->ratelimited);
21255d
         }
21255d
         if (r < 0)
21255d
                 return r;
21255d
@@ -2605,6 +2707,96 @@ _public_ void *sd_event_source_set_userdata(sd_event_source *s, void *userdata)
21255d
         return ret;
21255d
 }
21255d
 
21255d
+static int event_source_enter_ratelimited(sd_event_source *s) {
21255d
+        int r;
21255d
+
21255d
+        assert(s);
21255d
+
21255d
+        /* When an event source becomes ratelimited, we place it in the CLOCK_MONOTONIC priority queue, with
21255d
+         * the end of the rate limit time window, much as if it was a timer event source. */
21255d
+
21255d
+        if (s->ratelimited)
21255d
+                return 0; /* Already ratelimited, this is a NOP hence */
21255d
+
21255d
+        /* Make sure we can install a CLOCK_MONOTONIC event further down. */
21255d
+        r = setup_clock_data(s->event, &s->event->monotonic, CLOCK_MONOTONIC);
21255d
+        if (r < 0)
21255d
+                return r;
21255d
+
21255d
+        /* Timer event sources are already using the earliest/latest queues for the timer scheduling. Let's
21255d
+         * first remove them from the prioq appropriate for their own clock, so that we can use the prioq
21255d
+         * fields of the event source then for adding it to the CLOCK_MONOTONIC prioq instead. */
21255d
+        if (EVENT_SOURCE_IS_TIME(s->type))
21255d
+                event_source_time_prioq_remove(s, event_get_clock_data(s->event, s->type));
21255d
+
21255d
+        /* Now, let's add the event source to the monotonic clock instead */
21255d
+        r = event_source_time_prioq_put(s, &s->event->monotonic);
21255d
+        if (r < 0)
21255d
+                goto fail;
21255d
+
21255d
+        /* And let's take the event source officially offline */
21255d
+        r = event_source_offline(s, s->enabled, /* ratelimited= */ true);
21255d
+        if (r < 0) {
21255d
+                event_source_time_prioq_remove(s, &s->event->monotonic);
21255d
+                goto fail;
21255d
+        }
21255d
+
21255d
+        event_source_pp_prioq_reshuffle(s);
21255d
+
21255d
+        log_debug("Event source %p (%s) entered rate limit state.", s, strna(s->description));
21255d
+        return 0;
21255d
+
21255d
+fail:
21255d
+        /* Reinstall time event sources in the priority queue as before. This shouldn't fail, since the queue
21255d
+         * space for it should already be allocated. */
21255d
+        if (EVENT_SOURCE_IS_TIME(s->type))
21255d
+                assert_se(event_source_time_prioq_put(s, event_get_clock_data(s->event, s->type)) >= 0);
21255d
+
21255d
+        return r;
21255d
+}
21255d
+
21255d
+static int event_source_leave_ratelimit(sd_event_source *s) {
21255d
+        int r;
21255d
+
21255d
+        assert(s);
21255d
+
21255d
+        if (!s->ratelimited)
21255d
+                return 0;
21255d
+
21255d
+        /* Let's take the event source out of the monotonic prioq first. */
21255d
+        event_source_time_prioq_remove(s, &s->event->monotonic);
21255d
+
21255d
+        /* Let's then add the event source to its native clock prioq again — if this is a timer event source */
21255d
+        if (EVENT_SOURCE_IS_TIME(s->type)) {
21255d
+                r = event_source_time_prioq_put(s, event_get_clock_data(s->event, s->type));
21255d
+                if (r < 0)
21255d
+                        goto fail;
21255d
+        }
21255d
+
21255d
+        /* Let's try to take it online again.  */
21255d
+        r = event_source_online(s, s->enabled, /* ratelimited= */ false);
21255d
+        if (r < 0) {
21255d
+                /* Do something roughly sensible when this failed: undo the two prioq ops above */
21255d
+                if (EVENT_SOURCE_IS_TIME(s->type))
21255d
+                        event_source_time_prioq_remove(s, event_get_clock_data(s->event, s->type));
21255d
+
21255d
+                goto fail;
21255d
+        }
21255d
+
21255d
+        event_source_pp_prioq_reshuffle(s);
21255d
+        ratelimit_reset(&s->rate_limit);
21255d
+
21255d
+        log_debug("Event source %p (%s) left rate limit state.", s, strna(s->description));
21255d
+        return 0;
21255d
+
21255d
+fail:
21255d
+        /* Do something somewhat reasonable when we cannot move an event sources out of ratelimited mode:
21255d
+         * simply put it back in it, maybe we can then process it more successfully next iteration. */
21255d
+        assert_se(event_source_time_prioq_put(s, &s->event->monotonic) >= 0);
21255d
+
21255d
+        return r;
21255d
+}
21255d
+
21255d
 static usec_t sleep_between(sd_event *e, usec_t a, usec_t b) {
21255d
         usec_t c;
21255d
         assert(e);
21255d
@@ -2703,7 +2895,7 @@ static int event_arm_timer(
21255d
                 d->needs_rearm = false;
21255d
 
21255d
         a = prioq_peek(d->earliest);
21255d
-        if (!a || a->enabled == SD_EVENT_OFF || a->time.next == USEC_INFINITY) {
21255d
+        if (!a || a->enabled == SD_EVENT_OFF || time_event_source_next(a) == USEC_INFINITY) {
21255d
 
21255d
                 if (d->fd < 0)
21255d
                         return 0;
21255d
@@ -2723,7 +2915,7 @@ static int event_arm_timer(
21255d
         b = prioq_peek(d->latest);
21255d
         assert_se(b && b->enabled != SD_EVENT_OFF);
21255d
 
21255d
-        t = sleep_between(e, a->time.next, time_event_source_latest(b));
21255d
+        t = sleep_between(e, time_event_source_next(a), time_event_source_latest(b));
21255d
         if (d->next == t)
21255d
                 return 0;
21255d
 
21255d
@@ -2802,10 +2994,22 @@ static int process_timer(
21255d
 
21255d
         for (;;) {
21255d
                 s = prioq_peek(d->earliest);
21255d
-                if (!s ||
21255d
-                    s->time.next > n ||
21255d
-                    s->enabled == SD_EVENT_OFF ||
21255d
-                    s->pending)
21255d
+                if (!s || time_event_source_next(s) > n)
21255d
+                        break;
21255d
+
21255d
+                if (s->ratelimited) {
21255d
+                        /* This is an event sources whose ratelimit window has ended. Let's turn it on
21255d
+                         * again. */
21255d
+                        assert(s->ratelimited);
21255d
+
21255d
+                        r = event_source_leave_ratelimit(s);
21255d
+                        if (r < 0)
21255d
+                                return r;
21255d
+
21255d
+                        continue;
21255d
+                }
21255d
+
21255d
+                if (s->enabled == SD_EVENT_OFF || s->pending)
21255d
                         break;
21255d
 
21255d
                 r = source_set_pending(s, true);
21255d
@@ -2851,7 +3055,7 @@ static int process_child(sd_event *e) {
21255d
                 if (s->pending)
21255d
                         continue;
21255d
 
21255d
-                if (s->enabled == SD_EVENT_OFF)
21255d
+                if (event_source_is_offline(s))
21255d
                         continue;
21255d
 
21255d
                 zero(s->child.siginfo);
21255d
@@ -3024,7 +3228,7 @@ static int event_inotify_data_process(sd_event *e, struct inotify_data *d) {
21255d
 
21255d
                                 LIST_FOREACH(inotify.by_inode_data, s, inode_data->event_sources) {
21255d
 
21255d
-                                        if (s->enabled == SD_EVENT_OFF)
21255d
+                                        if (event_source_is_offline(s))
21255d
                                                 continue;
21255d
 
21255d
                                         r = source_set_pending(s, true);
21255d
@@ -3060,7 +3264,7 @@ static int event_inotify_data_process(sd_event *e, struct inotify_data *d) {
21255d
                          * sources if IN_IGNORED or IN_UNMOUNT is set. */
21255d
                         LIST_FOREACH(inotify.by_inode_data, s, inode_data->event_sources) {
21255d
 
21255d
-                                if (s->enabled == SD_EVENT_OFF)
21255d
+                                if (event_source_is_offline(s))
21255d
                                         continue;
21255d
 
21255d
                                 if ((d->buffer.ev.mask & (IN_IGNORED|IN_UNMOUNT)) == 0 &&
21255d
@@ -3099,6 +3303,7 @@ static int process_inotify(sd_event *e) {
21255d
 }
21255d
 
21255d
 static int source_dispatch(sd_event_source *s) {
21255d
+        _cleanup_(sd_event_unrefp) sd_event *saved_event = NULL;
21255d
         EventSourceType saved_type;
21255d
         int r = 0;
21255d
 
21255d
@@ -3109,6 +3314,20 @@ static int source_dispatch(sd_event_source *s) {
21255d
          * the event. */
21255d
         saved_type = s->type;
21255d
 
21255d
+        /* Similar, store a reference to the event loop object, so that we can still access it after the
21255d
+         * callback might have invalidated/disconnected the event source. */
21255d
+        saved_event = sd_event_ref(s->event);
21255d
+
21255d
+        /* Check if we hit the ratelimit for this event source, if so, let's disable it. */
21255d
+        assert(!s->ratelimited);
21255d
+        if (!ratelimit_below(&s->rate_limit)) {
21255d
+                r = event_source_enter_ratelimited(s);
21255d
+                if (r < 0)
21255d
+                        return r;
21255d
+
21255d
+                return 1;
21255d
+        }
21255d
+
21255d
         if (!IN_SET(s->type, SOURCE_DEFER, SOURCE_EXIT)) {
21255d
                 r = source_set_pending(s, false);
21255d
                 if (r < 0)
21255d
@@ -3235,7 +3454,7 @@ static int event_prepare(sd_event *e) {
21255d
                 sd_event_source *s;
21255d
 
21255d
                 s = prioq_peek(e->prepare);
21255d
-                if (!s || s->prepare_iteration == e->iteration || s->enabled == SD_EVENT_OFF)
21255d
+                if (!s || s->prepare_iteration == e->iteration || event_source_is_offline(s))
21255d
                         break;
21255d
 
21255d
                 s->prepare_iteration = e->iteration;
21255d
@@ -3269,7 +3488,7 @@ static int dispatch_exit(sd_event *e) {
21255d
         assert(e);
21255d
 
21255d
         p = prioq_peek(e->exit);
21255d
-        if (!p || p->enabled == SD_EVENT_OFF) {
21255d
+        if (!p || event_source_is_offline(p)) {
21255d
                 e->state = SD_EVENT_FINISHED;
21255d
                 return 0;
21255d
         }
21255d
@@ -3291,7 +3510,7 @@ static sd_event_source* event_next_pending(sd_event *e) {
21255d
         if (!p)
21255d
                 return NULL;
21255d
 
21255d
-        if (p->enabled == SD_EVENT_OFF)
21255d
+        if (event_source_is_offline(p))
21255d
                 return NULL;
21255d
 
21255d
         return p;
21255d
@@ -3844,3 +4063,53 @@ _public_ int sd_event_source_get_destroy_callback(sd_event_source *s, sd_event_d
21255d
 
21255d
         return !!s->destroy_callback;
21255d
 }
21255d
+
21255d
+_public_ int sd_event_source_set_ratelimit(sd_event_source *s, uint64_t interval, unsigned burst) {
21255d
+        int r;
21255d
+
21255d
+        assert_return(s, -EINVAL);
21255d
+
21255d
+        /* Turning on ratelimiting on event source types that don't support it, is a loggable offense. Doing
21255d
+         * so is a programming error. */
21255d
+        assert_return(EVENT_SOURCE_CAN_RATE_LIMIT(s->type), -EDOM);
21255d
+
21255d
+        /* When ratelimiting is configured we'll always reset the rate limit state first and start fresh,
21255d
+         * non-ratelimited. */
21255d
+        r = event_source_leave_ratelimit(s);
21255d
+        if (r < 0)
21255d
+                return r;
21255d
+
21255d
+        RATELIMIT_INIT(s->rate_limit, interval, burst);
21255d
+        return 0;
21255d
+}
21255d
+
21255d
+_public_ int sd_event_source_get_ratelimit(sd_event_source *s, uint64_t *ret_interval, unsigned *ret_burst) {
21255d
+        assert_return(s, -EINVAL);
21255d
+
21255d
+        /* Querying whether an event source has ratelimiting configured is not a loggable offsense, hence
21255d
+         * don't use assert_return(). Unlike turning on ratelimiting it's not really a programming error */
21255d
+        if (!EVENT_SOURCE_CAN_RATE_LIMIT(s->type))
21255d
+                return -EDOM;
21255d
+
21255d
+        if (!ratelimit_configured(&s->rate_limit))
21255d
+                return -ENOEXEC;
21255d
+
21255d
+        if (ret_interval)
21255d
+                *ret_interval = s->rate_limit.interval;
21255d
+        if (ret_burst)
21255d
+                *ret_burst = s->rate_limit.burst;
21255d
+
21255d
+        return 0;
21255d
+}
21255d
+
21255d
+_public_ int sd_event_source_is_ratelimited(sd_event_source *s) {
21255d
+        assert_return(s, -EINVAL);
21255d
+
21255d
+        if (!EVENT_SOURCE_CAN_RATE_LIMIT(s->type))
21255d
+                return false;
21255d
+
21255d
+        if (!ratelimit_configured(&s->rate_limit))
21255d
+                return false;
21255d
+
21255d
+        return s->ratelimited;
21255d
+}
21255d
diff --git a/src/systemd/sd-event.h b/src/systemd/sd-event.h
21255d
index 9876be01c6..a17a9b3488 100644
21255d
--- a/src/systemd/sd-event.h
21255d
+++ b/src/systemd/sd-event.h
21255d
@@ -144,6 +144,9 @@ int sd_event_source_get_child_pid(sd_event_source *s, pid_t *pid);
21255d
 int sd_event_source_get_inotify_mask(sd_event_source *s, uint32_t *ret);
21255d
 int sd_event_source_set_destroy_callback(sd_event_source *s, sd_event_destroy_t callback);
21255d
 int sd_event_source_get_destroy_callback(sd_event_source *s, sd_event_destroy_t *ret);
21255d
+int sd_event_source_set_ratelimit(sd_event_source *s, uint64_t interval_usec, unsigned burst);
21255d
+int sd_event_source_get_ratelimit(sd_event_source *s, uint64_t *ret_interval_usec, unsigned *ret_burst);
21255d
+int sd_event_source_is_ratelimited(sd_event_source *s);
21255d
 
21255d
 /* Define helpers so that __attribute__((cleanup(sd_event_unrefp))) and similar may be used. */
21255d
 _SD_DEFINE_POINTER_CLEANUP_FUNC(sd_event, sd_event_unref);