X-Git-Url: https://www.chiark.greenend.org.uk/ucgi/~ianmdlvl/git?a=blobdiff_plain;f=src%2Fjournal%2Fjournald-stream.c;h=89da150a605bacdb87e79a29a28eb787bc02dbcf;hb=d12b8cad40aa78fc948362340204c3fde778082d;hp=5c15a56bacf61e038ed9c9d8320f548e7c772cc7;hpb=e3bfb7be07d9b1f4ebb12eb22c4c8bcd2a988d51;p=elogind.git diff --git a/src/journal/journald-stream.c b/src/journal/journald-stream.c index 5c15a56ba..89da150a6 100644 --- a/src/journal/journald-stream.c +++ b/src/journal/journald-stream.c @@ -22,12 +22,12 @@ #include #include #include -#include #ifdef HAVE_SELINUX #include #endif +#include "sd-event.h" #include "socket-util.h" #include "selinux-util.h" #include "journald-server.h" @@ -35,6 +35,7 @@ #include "journald-syslog.h" #include "journald-kmsg.h" #include "journald-console.h" +#include "journald-wall.h" #define STDOUT_STREAMS_MAX 4096 @@ -71,6 +72,8 @@ struct StdoutStream { char buffer[LINE_MAX+1]; size_t length; + sd_event_source *event_source; + LIST_FIELDS(StdoutStream, stdout_stream); }; @@ -104,6 +107,9 @@ static int stdout_stream_log(StdoutStream *s, const char *p) { if (s->forward_to_console || s->server->forward_to_console) server_forward_console(s->server, priority, s->identifier, p, &s->ucred); + if (s->server->forward_to_wall) + server_forward_wall(s->server, priority, s->identifier, p, &s->ucred); + IOVEC_SET_STRING(iovec[n++], "_TRANSPORT=stdout"); syslog_priority[strlen("PRIORITY=")] = '0' + LOG_PRI(priority); @@ -282,12 +288,18 @@ static int stdout_stream_scan(StdoutStream *s, bool force_flush) { return 0; } -int stdout_stream_process(StdoutStream *s) { +static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents, void *userdata) { + StdoutStream *s = userdata; ssize_t l; int r; assert(s); + if ((revents|EPOLLIN|EPOLLHUP) != (EPOLLIN|EPOLLHUP)) { + log_error("Got invalid event from epoll for stdout stream: %"PRIx32, revents); + goto terminate; + } + l = read(s->fd, s->buffer+s->length, sizeof(s->buffer)-1-s->length); if (l < 0) { @@ -295,24 +307,24 @@ int stdout_stream_process(StdoutStream *s) { return 0; log_warning("Failed to read from stream: %m"); - return -errno; + goto terminate; } if (l == 0) { - r = stdout_stream_scan(s, true); - if (r < 0) - return r; - - return 0; + stdout_stream_scan(s, true); + goto terminate; } s->length += l; r = stdout_stream_scan(s, false); if (r < 0) - return r; + goto terminate; return 1; +terminate: + stdout_stream_free(s); + return 0; } void stdout_stream_free(StdoutStream *s) { @@ -324,13 +336,13 @@ void stdout_stream_free(StdoutStream *s) { LIST_REMOVE(stdout_stream, s->server->stdout_streams, s); } - if (s->fd >= 0) { - if (s->server) - epoll_ctl(s->server->epoll_fd, EPOLL_CTL_DEL, s->fd, NULL); - - close_nointr_nofail(s->fd); + if (s->event_source) { + sd_event_source_set_enabled(s->event_source, SD_EVENT_OFF); + s->event_source = sd_event_source_unref(s->event_source); } + safe_close(s->fd); + #ifdef HAVE_SELINUX if (s->security_context) freecon(s->security_context); @@ -341,14 +353,18 @@ void stdout_stream_free(StdoutStream *s) { free(s); } -int stdout_stream_new(Server *s) { +static int stdout_stream_new(sd_event_source *es, int listen_fd, uint32_t revents, void *userdata) { + Server *s = userdata; StdoutStream *stream; int fd, r; - socklen_t len; - struct epoll_event ev; assert(s); + if (revents != EPOLLIN) { + log_error("Got invalid event from epoll for stdout server fd: %"PRIx32, revents); + return -EIO; + } + fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC); if (fd < 0) { if (errno == EAGAIN) @@ -360,22 +376,21 @@ int stdout_stream_new(Server *s) { if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) { log_warning("Too many stdout streams, refusing connection."); - close_nointr_nofail(fd); + safe_close(fd); return 0; } stream = new0(StdoutStream, 1); if (!stream) { - close_nointr_nofail(fd); + safe_close(fd); return log_oom(); } stream->fd = fd; - len = sizeof(stream->ucred); - if (getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &stream->ucred, &len) < 0) { + r = getpeercred(fd, &stream->ucred); + if (r < 0) { log_error("Failed to determine peer credentials: %m"); - r = -errno; goto fail; } @@ -388,16 +403,18 @@ int stdout_stream_new(Server *s) { if (shutdown(fd, SHUT_WR) < 0) { log_error("Failed to shutdown writing side of socket: %m"); - r = -errno; goto fail; } - zero(ev); - ev.data.ptr = stream; - ev.events = EPOLLIN; - if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, fd, &ev) < 0) { - log_error("Failed to add stream to event loop: %m"); - r = -errno; + r = sd_event_add_io(s->event, &stream->event_source, fd, EPOLLIN, stdout_stream_process, stream); + if (r < 0) { + log_error("Failed to add stream to event loop: %s", strerror(-r)); + goto fail; + } + + r = sd_event_source_set_priority(stream->event_source, SD_EVENT_PRIORITY_NORMAL+5); + if (r < 0) { + log_error("Failed to adjust stdout event source priority: %s", strerror(-r)); goto fail; } @@ -409,12 +426,11 @@ int stdout_stream_new(Server *s) { fail: stdout_stream_free(stream); - return r; + return 0; } int server_open_stdout_socket(Server *s) { int r; - struct epoll_event ev = { .events = EPOLLIN }; assert(s); @@ -447,10 +463,16 @@ int server_open_stdout_socket(Server *s) { } else fd_nonblock(s->stdout_fd, 1); - ev.data.fd = s->stdout_fd; - if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->stdout_fd, &ev) < 0) { - log_error("Failed to add stdout server fd to epoll object: %m"); - return -errno; + r = sd_event_add_io(s->event, &s->stdout_event_source, s->stdout_fd, EPOLLIN, stdout_stream_new, s); + if (r < 0) { + log_error("Failed to add stdout server fd to event source: %s", strerror(-r)); + return r; + } + + r = sd_event_source_set_priority(s->stdout_event_source, SD_EVENT_PRIORITY_NORMAL+10); + if (r < 0) { + log_error("Failed to adjust priority of stdout server event source: %s", strerror(-r)); + return r; } return 0;