Merge pull request #11530 from keszybz/journal-cache-trimming

Journal cache trimming

src/basic/prioq.c

@@ -259,15 +259,14 @@ int prioq_reshuffle(Prioq *q, void *data, unsigned *idx) {
         return 1;
 }
 
-void *prioq_peek(Prioq *q) {
-
+void *prioq_peek_by_index(Prioq *q, unsigned idx) {
         if (!q)
                 return NULL;
 
-        if (q->n_items <= 0)
+        if (idx >= q->n_items)
                 return NULL;
 
-        return q->items[0].data;
+        return q->items[idx].data;
 }
 
 void *prioq_pop(Prioq *q) {

src/basic/prioq.h

@@ -19,8 +19,14 @@ int prioq_put(Prioq *q, void *data, unsigned *idx);
 int prioq_remove(Prioq *q, void *data, unsigned *idx);
 int prioq_reshuffle(Prioq *q, void *data, unsigned *idx);
 
-void *prioq_peek(Prioq *q) _pure_;
+void *prioq_peek_by_index(Prioq *q, unsigned idx) _pure_;
+static inline void *prioq_peek(Prioq *q) {
+        return prioq_peek_by_index(q, 0);
+}
 void *prioq_pop(Prioq *q);
 
+#define PRIOQ_FOREACH_ITEM(q, p)                                \
+        for (unsigned _i = 0; (p = prioq_peek_by_index(q, _i)); _i++)
+
 unsigned prioq_size(Prioq *q) _pure_;
 bool prioq_isempty(Prioq *q) _pure_;

src/basic/procfs-util.c

@@ -201,13 +201,11 @@ int procfs_cpu_get_usage(nsec_t *ret) {
         return 0;
 }
 
-int procfs_memory_get_current(uint64_t *ret) {
+int procfs_memory_get(uint64_t *ret_total, uint64_t *ret_used) {
         uint64_t mem_total = UINT64_MAX, mem_free = UINT64_MAX;
         _cleanup_fclose_ FILE *f = NULL;
         int r;
 
-        assert(ret);
-
         f = fopen("/proc/meminfo", "re");
         if (!f)
                 return -errno;
@@ -262,6 +260,9 @@ int procfs_memory_get_current(uint64_t *ret) {
         if (mem_free > mem_total)
                 return -EINVAL;
 
-        *ret = (mem_total - mem_free) * 1024U;
+        if (ret_total)
+                *ret_total = mem_total * 1024U;
+        if (ret_used)
+                *ret_used = (mem_total - mem_free) * 1024U;
         return 0;
 }

src/basic/procfs-util.h

@@ -11,4 +11,7 @@ int procfs_tasks_get_current(uint64_t *ret);
 
 int procfs_cpu_get_usage(nsec_t *ret);
 
-int procfs_memory_get_current(uint64_t *ret);
+int procfs_memory_get(uint64_t *ret_total, uint64_t *ret_used);
+static inline int procfs_memory_get_used(uint64_t *ret) {
+        return procfs_memory_get(NULL, ret);
+}

src/cgtop/cgtop.c

@@ -291,7 +291,7 @@ static int process(
         } else if (streq(controller, "memory")) {
 
                 if (is_root_cgroup(path)) {
-                        r = procfs_memory_get_current(&g->memory);
+                        r = procfs_memory_get_used(&g->memory);
                         if (r < 0)
                                 return r;
                 } else {

src/core/cgroup.c

@@ -2780,7 +2780,7 @@ int unit_get_memory_current(Unit *u, uint64_t *ret) {
 
         /* The root cgroup doesn't expose this information, let's get it from /proc instead */
         if (unit_has_host_root_cgroup(u))
-                return procfs_memory_get_current(ret);
+                return procfs_memory_get_used(ret);
 
         if ((u->cgroup_realized_mask & CGROUP_MASK_MEMORY) == 0)
                 return -ENODATA;

src/journal/journald-context.c

@@ -16,6 +16,7 @@
 #include "parse-util.h"
 #include "path-util.h"
 #include "process-util.h"
+#include "procfs-util.h"
 #include "string-util.h"
 #include "syslog-util.h"
 #include "unaligned.h"
@@ -60,7 +61,37 @@
 /* Keep at most 16K entries in the cache. (Note though that this limit may be violated if enough streams pin entries in
  * the cache, in which case we *do* permit this limit to be breached. That's safe however, as the number of stream
  * clients itself is limited.) */
-#define CACHE_MAX (16*1024)
+#define CACHE_MAX_FALLBACK 128U
+#define CACHE_MAX_MAX (16*1024U)
+#define CACHE_MAX_MIN 64U
+
+static size_t cache_max(void) {
+        static size_t cached = -1;
+
+        if (cached == (size_t) -1) {
+                uint64_t mem_total;
+                int r;
+
+                r = procfs_memory_get(&mem_total, NULL);
+                if (r < 0) {
+                        log_warning_errno(r, "Cannot query /proc/meminfo for MemTotal: %m");
+                        cached = CACHE_MAX_FALLBACK;
+                } else {
+                        /* Cache entries are usually a few kB, but the process cmdline is controlled by the
+                         * user and can be up to _SC_ARG_MAX, usually 2MB. Let's say that approximately up to
+                         * 1/8th of memory may be used by the cache.
+                         *
+                         * In the common case, this formula gives 64 cache entries for each GB of RAM.
+                         */
+                        long l = sysconf(_SC_ARG_MAX);
+                        assert(l > 0);
+
+                        cached = CLAMP(mem_total / 8 / (uint64_t) l, CACHE_MAX_MIN, CACHE_MAX_MAX);
+                }
+        }
+
+        return cached;
+}
 
 static int client_context_compare(const void *a, const void *b) {
         const ClientContext *x = a, *y = b;
@@ -550,15 +581,39 @@ refresh:
 }
 
 static void client_context_try_shrink_to(Server *s, size_t limit) {
+        ClientContext *c;
+        usec_t t;
+
         assert(s);
 
+        /* Flush any cache entries for PIDs that have already moved on. Don't do this
+         * too often, since it's a slow process. */
+        t = now(CLOCK_MONOTONIC);
+        if (s->last_cache_pid_flush + MAX_USEC < t) {
+                unsigned n = prioq_size(s->client_contexts_lru), idx = 0;
+
+                /* We do a number of iterations based on the initial size of the prioq.  When we remove an
+                 * item, a new item is moved into its places, and items to the right might be reshuffled.
+                 */
+                for (unsigned i = 0; i < n; i++) {
+                        c = prioq_peek_by_index(s->client_contexts_lru, idx);
+
+                        assert(c->n_ref == 0);
+
+                        if (!pid_is_unwaited(c->pid))
+                                client_context_free(s, c);
+                        else
+                                idx ++;
+                }
+
+                s->last_cache_pid_flush = t;
+        }
+
         /* Bring the number of cache entries below the indicated limit, so that we can create a new entry without
          * breaching the limit. Note that we only flush out entries that aren't pinned here. This means the number of
          * cache entries may very well grow beyond the limit, if all entries stored remain pinned. */
 
         while (hashmap_size(s->client_contexts) > limit) {
-                ClientContext *c;
-
                 c = prioq_pop(s->client_contexts_lru);
                 if (!c)
                         break; /* All remaining entries are pinned, give up */
@@ -627,7 +682,7 @@ static int client_context_get_internal(
                 return 0;
         }
 
-        client_context_try_shrink_to(s, CACHE_MAX-1);
+        client_context_try_shrink_to(s, cache_max()-1);
 
         r = client_context_new(s, pid, &c);
         if (r < 0)

src/journal/journald-server.h

@@ -161,6 +161,8 @@ struct Server {
         Hashmap *client_contexts;
         Prioq *client_contexts_lru;
 
+        usec_t last_cache_pid_flush;
+
         ClientContext *my_context; /* the context of journald itself */
         ClientContext *pid1_context; /* the context of PID 1 */
 };

src/test/test-prioq.c

@@ -69,6 +69,11 @@ static void test_struct(void) {
         assert_se(q = prioq_new((compare_func_t) test_compare));
         assert_se(s = set_new(&test_hash_ops));
 
+        assert_se(prioq_peek(q) == NULL);
+        assert_se(prioq_peek_by_index(q, 0) == NULL);
+        assert_se(prioq_peek_by_index(q, 1) == NULL);
+        assert_se(prioq_peek_by_index(q, (unsigned) -1) == NULL);
+
         for (i = 0; i < SET_SIZE; i++) {
                 assert_se(t = new0(struct test, 1));
                 t->value = (unsigned) rand();
@@ -79,6 +84,17 @@ static void test_struct(void) {
                         assert_se(set_consume(s, t) >= 0);
         }
 
+        for (i = 0; i < SET_SIZE; i++)
+                assert_se(prioq_peek_by_index(q, i));
+        assert_se(prioq_peek_by_index(q, SET_SIZE) == NULL);
+
+        unsigned count = 0;
+        PRIOQ_FOREACH_ITEM(q, t) {
+                assert_se(t);
+                count++;
+        }
+        assert_se(count == SET_SIZE);
+
         while ((t = set_steal_first(s))) {
                 assert_se(prioq_remove(q, t, &t->idx) == 1);
                 assert_se(prioq_remove(q, t, &t->idx) == 0);

src/test/test-procfs-util.c

@@ -18,7 +18,7 @@ int main(int argc, char *argv[]) {
         assert_se(procfs_cpu_get_usage(&nsec) >= 0);
         log_info("Current system CPU time: %s", format_timespan(buf, sizeof(buf), nsec/NSEC_PER_USEC, 1));
 
-        assert_se(procfs_memory_get_current(&v) >= 0);
+        assert_se(procfs_memory_get_used(&v) >= 0);
         log_info("Current memory usage: %s", format_bytes(buf, sizeof(buf), v));
 
         assert_se(procfs_tasks_get_current(&v) >= 0);