Merge branch 'pr/async-logger' into 'master'

Draft: log: add async logging

See merge request pulseaudio/pulseaudio!459

src/daemon/cmdline.c

@@ -53,6 +53,7 @@ enum {
     ARG_LOG_TARGET,
     ARG_LOG_META,
     ARG_LOG_TIME,
+    ARG_LOG_ASYNC,
     ARG_LOG_BACKTRACE,
     ARG_LOAD,
     ARG_FILE,
@@ -89,6 +90,7 @@ static const struct option long_options[] = {
     {"log-target",                  1, 0, ARG_LOG_TARGET},
     {"log-meta",                    2, 0, ARG_LOG_META},
     {"log-time",                    2, 0, ARG_LOG_TIME},
+    {"log-async",                   2, 0, ARG_LOG_ASYNC},
     {"log-backtrace",               1, 0, ARG_LOG_BACKTRACE},
     {"load",                        1, 0, ARG_LOAD},
     {"file",                        1, 0, ARG_FILE},
@@ -143,6 +145,7 @@ void pa_cmdline_help(const char *argv0) {
            "  -v  --verbose                         Increase the verbosity level\n"
            "      --log-target={auto,syslog,stderr,file:PATH,newfile:PATH}\n"
            "                                        Specify the log target\n"
+           "      --log-async[=BOOL]                Asynchronously write to log file\n"
            "      --log-meta[=BOOL]                 Include code location in log messages\n"
            "      --log-time[=BOOL]                 Include timestamps in log messages\n"
            "      --log-backtrace=FRAMES            Include a backtrace in log messages\n"
@@ -341,6 +344,14 @@ int pa_cmdline_parse(pa_daemon_conf *conf, int argc, char *const argv [], int *d
                 conf->log_time = !!b;
                 break;
 
+            case ARG_LOG_ASYNC:
+                if ((b = optarg ? pa_parse_boolean(optarg) : 1) < 0) {
+                    pa_log(_("--log-async expects boolean argument"));
+                    goto fail;
+                }
+                conf->log_async = !!b;
+                break;
+
             case ARG_LOG_META:
                 if ((b = optarg ? pa_parse_boolean(optarg) : 1) < 0) {
                     pa_log(_("--log-meta expects boolean argument"));

src/daemon/daemon-conf.c

@@ -81,6 +81,7 @@ static const pa_daemon_conf default_conf = {
     .log_backtrace = 0,
     .log_meta = false,
     .log_time = false,
+    .log_async = false,
     .resample_method = PA_RESAMPLER_AUTO,
     .avoid_resampling = false,
     .disable_remixing = false,
@@ -619,6 +620,7 @@ int pa_daemon_conf_load(pa_daemon_conf *c, const char *filename) {
         { "shm-size-bytes",             pa_config_parse_size,     &c->shm_size, NULL },
         { "log-meta",                   pa_config_parse_bool,     &c->log_meta, NULL },
         { "log-time",                   pa_config_parse_bool,     &c->log_time, NULL },
+        { "log-async",                  pa_config_parse_bool,     &c->log_async, NULL },
         { "log-backtrace",              pa_config_parse_unsigned, &c->log_backtrace, NULL },
 #ifdef HAVE_SYS_RESOURCE_H
         { "rlimit-fsize",               parse_rlimit,             &c->rlimit_fsize, NULL },
@@ -837,6 +839,7 @@ char *pa_daemon_conf_dump(pa_daemon_conf *c) {
     pa_strbuf_printf(s, "shm-size-bytes = %lu\n", (unsigned long) c->shm_size);
     pa_strbuf_printf(s, "log-meta = %s\n", pa_yes_no(c->log_meta));
     pa_strbuf_printf(s, "log-time = %s\n", pa_yes_no(c->log_time));
+    pa_strbuf_printf(s, "log-async = %s\n", pa_yes_no(c->log_async));
     pa_strbuf_printf(s, "log-backtrace = %u\n", c->log_backtrace);
 #ifdef HAVE_SYS_RESOURCE_H
     pa_strbuf_printf(s, "rlimit-fsize = %li\n", c->rlimit_fsize.is_set ? (long int) c->rlimit_fsize.value : -1);

src/daemon/daemon-conf.h

@@ -76,6 +76,7 @@ typedef struct pa_daemon_conf {
         disallow_exit,
         log_meta,
         log_time,
+        log_async,
         flat_volumes,
         rescue_streams,
         lock_memory,

src/daemon/main.c

@@ -1243,6 +1243,9 @@ int main(int argc, char *argv[]) {
     if (!conf->no_cpu_limit)
         pa_assert_se(pa_cpu_limit_init(pa_mainloop_get_api(mainloop)) == 0);
 
+    if (conf->log_async)
+        pa_log_start_async();
+
     buf = pa_strbuf_new();
 
 #ifdef HAVE_DBUS
@@ -1356,6 +1359,8 @@ int main(int argc, char *argv[]) {
 #endif
 
 finish:
+    pa_log_stop_async();
+
 #ifdef HAVE_DBUS
     if (server_bus)
         pa_dbus_connection_unref(server_bus);

src/meson.build

@@ -13,6 +13,7 @@ libpulsecommon_sources = [
   'pulse/timeval.c',
   'pulse/rtclock.c',
   'pulse/volume.c',
+  'pulsecore/asyncq.c',
   'pulsecore/authkey.c',
   'pulsecore/conf-parser.c',
   'pulsecore/core-error.c',
@@ -83,6 +84,7 @@ libpulsecommon_headers = [
   'pulse/timeval.h',
   'pulse/rtclock.h',
   'pulse/volume.h',
+  'pulsecore/asyncq.h',
   'pulsecore/atomic.h',
   'pulsecore/authkey.h',
   'pulsecore/conf-parser.h',

src/pulsecore/log.c

@@ -38,6 +38,9 @@
 #include <syslog.h>
 #endif
 
+#include "flist.h"
+#include "asyncq.h"
+
 #ifdef HAVE_SYSTEMD_JOURNAL
 
 /* sd_journal_send() implicitly add fields for the source file,
@@ -94,6 +97,21 @@ static bool no_rate_limit = false;
 static int log_fd = -1;
 static int write_type = 0;
 
+#define ASYNC_LOG_BUFFER_SIZE 1024
+#define ASYNC_LOG_QUEUE_SIZE 1024
+
+struct log_item {
+    char buffer[ASYNC_LOG_BUFFER_SIZE];
+};
+static bool log_thread_do_exit = false;
+static pa_mutex *log_mutex = NULL; /* protects async_queue writer side and helps with log_cond */
+static pa_cond *log_cond = NULL;
+
+pa_thread *logger_thread = NULL;
+
+PA_STATIC_FLIST_DECLARE(log_items, ASYNC_LOG_QUEUE_SIZE, pa_xfree);
+pa_asyncq *async_queue = NULL;
+
 #ifdef HAVE_SYSLOG_H
 static const int level_to_syslog[] = {
     [PA_LOG_ERROR] = LOG_ERR,
@@ -146,6 +164,120 @@ void pa_log_set_level(pa_log_level_t l) {
     maximum_level = l;
 }
 
+static void log_file_write(const char *buf) {
+    int saved_errno = errno;
+    if (pa_write(log_fd, buf, strlen(buf), &write_type) < 0) {
+        pa_log_target new_target = { .type = PA_LOG_STDERR, .file = NULL };
+        saved_errno = errno;
+        fprintf(stderr, "%s\n", "Error writing logs to a file descriptor. Redirect log messages to console.");
+        fprintf(stderr, "%s", buf);
+        pa_log_set_target(&new_target);
+    }
+    errno = saved_errno;
+}
+
+static void log_thread_func(void *userdata) {
+    //struct userdata *u = userdata;
+
+    //pa_assert(u);
+
+    pa_log_debug("Thread starting up");
+
+    for (;;) {
+
+        struct log_item *l;
+
+        if ((l = pa_asyncq_pop(async_queue, false))) {
+            log_file_write(l->buffer);
+            if (pa_flist_push(PA_STATIC_FLIST_GET(log_items), l) < 0) {
+                pa_xfree(l);
+            }
+            if (!log_thread_do_exit)
+                continue;
+        }
+
+        /* if thread exit is not requested, wait for more items */
+
+        pa_mutex_lock(log_mutex);
+        if (!log_thread_do_exit)
+            pa_cond_wait(log_cond, log_mutex);
+        pa_mutex_unlock(log_mutex);
+
+        if (log_thread_do_exit)
+            break;
+    }
+
+finish:
+    pa_log_debug("Thread shutting down");
+}
+
+void pa_log_start_async(void) {
+    if (logger_thread)
+        return;
+
+    async_queue = pa_asyncq_new(ASYNC_LOG_QUEUE_SIZE);
+
+    log_mutex = pa_mutex_new(false, false);
+    log_cond = pa_cond_new();
+
+    if (!(logger_thread = pa_thread_new("async-log-writer", log_thread_func, NULL))) {
+        pa_log(_("Failed to start async logger thread."));
+
+        pa_cond_free(log_cond);
+        log_cond = NULL;
+
+        pa_mutex_free(log_mutex);
+        log_mutex = NULL;
+
+        return;
+    }
+
+    for (int i = 0; i < ASYNC_LOG_QUEUE_SIZE; ++i) {
+        struct log_item *l = pa_xnew(struct log_item, 1);
+        if (pa_flist_push(PA_STATIC_FLIST_GET(log_items), l) < 0) {
+            pa_xfree(l);
+            break;
+        }
+    }
+}
+
+static void log_thread_wakeup_unlocked(void) {
+    pa_cond_signal(log_cond, 0);
+}
+
+static void log_thread_wakeup(void) {
+    pa_mutex_lock(log_mutex);
+    log_thread_wakeup_unlocked();
+    pa_mutex_unlock(log_mutex);
+}
+
+void pa_log_stop_async(void) {
+    struct log_item *l;
+
+    if (!logger_thread)
+        return;
+
+    /* set exit flag and signal the condition to unblock thread */
+    log_thread_do_exit = true;
+    log_thread_wakeup();
+
+    pa_thread_free(logger_thread);
+    logger_thread = NULL;
+
+    pa_cond_free(log_cond);
+    log_cond = NULL;
+
+    pa_mutex_free(log_mutex);
+    log_mutex = NULL;
+
+    while((l = pa_asyncq_pop(async_queue, false))) {
+        log_file_write(l->buffer);
+    }
+
+    pa_asyncq_free(async_queue, NULL);
+    async_queue = NULL;
+}
+
 int pa_log_set_target(pa_log_target *t) {
     int fd = -1;
     int old_fd;
@@ -547,7 +679,36 @@ void pa_log_levelv_meta(
                     else
                         pa_snprintf(metadata, sizeof(metadata), "%s%s", timestamp, location);
 
-                    if ((pa_write(log_fd, metadata, strlen(metadata), &write_type) < 0)
+                    if (logger_thread) {
+                        struct log_item *l;
+                        bool allocated_new = false;
+
+                        if (!(l = pa_flist_pop(PA_STATIC_FLIST_GET(log_items)))) {
+                            allocated_new = true;
+                            l = pa_xnew(struct log_item, 1);
+                        }
+
+                        snprintf(l->buffer, sizeof(l->buffer), "[A]%s%s%s\n", metadata, t, pa_strempty(bt));
+
+                        if (allocated_new)
+                            l->buffer[1] = 'N';
+
+                        pa_mutex_lock(log_mutex);
+                        if (pa_asyncq_push(async_queue, l, false) == 0) {
+                            log_thread_wakeup_unlocked();
+                        } else {
+                            if (allocated_new)
+                                l->buffer[1] = 'S';
+                            else
+                                l->buffer[1] = 's';
+                            log_file_write(l->buffer);
+                            if (pa_flist_push(PA_STATIC_FLIST_GET(log_items), l) < 0) {
+                                pa_xfree(l);
+                            }
+                        }
+                        pa_mutex_unlock(log_mutex);
+
+                    } else if ((pa_write(log_fd, metadata, strlen(metadata), &write_type) < 0)
                             || (pa_write(log_fd, t, strlen(t), &write_type) < 0)
                             || (bt && pa_write(log_fd, bt, strlen(bt), &write_type) < 0)
                             || (pa_write(log_fd, "\n", 1, &write_type) < 0)) {

src/pulsecore/log.h

@@ -152,4 +152,7 @@ LOG_FUNC(error, PA_LOG_ERROR)
 
 bool pa_log_ratelimit(pa_log_level_t level);
 
+void pa_log_start_async();
+void pa_log_stop_async();
+
 #endif

src/pulsecore/meson.build

@@ -1,6 +1,5 @@
 libpulsecore_sources = [
   'asyncmsgq.c',
-  'asyncq.c',
   'auth-cookie.c',
   'card.c',
   'cli-command.c',
@@ -60,7 +59,6 @@ libpulsecore_sources = [
 
 libpulsecore_headers = [
   'asyncmsgq.h',
-  'asyncq.h',
   'auth-cookie.h',
   'card.h',
   'cli-command.h',