journal: implement parallel traversal in client

Makefile.am

@@ -965,6 +965,7 @@ test_id128_LDADD = \
 test_journal_SOURCES = \
 	src/journal/test-journal.c \
 	src/journal/sd-journal.c \
+        src/journal/journal-file.c \
         src/journal/lookup3.c \
         src/sd-id128.c
 
@@ -977,6 +978,7 @@ test_journal_LDADD = \
 systemd_journald_SOURCES = \
 	src/journal/journald.c \
 	src/journal/sd-journal.c \
+        src/journal/journal-file.c \
         src/journal/lookup3.c \
         src/sd-id128.c \
         src/acl-util.c
@@ -993,6 +995,7 @@ systemd_journald_LDADD = \
 systemd_journalctl_SOURCES = \
 	src/journal/journalctl.c \
 	src/journal/sd-journal.c \
+        src/journal/journal-file.c \
         src/journal/lookup3.c \
         src/sd-id128.c
 

src/journal/journal-def.h

@@ -48,7 +48,7 @@ enum {
 
 _packed_ struct ObjectHeader {
         uint8_t type;
-        uint8_t reserved[3];
+        uint8_t reserved[7];
         uint64_t size;
         uint8_t payload[];
 };
@@ -74,6 +74,7 @@ _packed_ struct EntryObject {
         uint64_t seqnum;
         uint64_t realtime;
         uint64_t monotonic;
+        sd_id128_t boot_id;
         uint64_t xor_hash;
         uint64_t prev_entry_offset;
         uint64_t next_entry_offset;
@@ -118,6 +119,7 @@ _packed_ struct Header {
         sd_id128_t file_id;
         sd_id128_t machine_id;
         sd_id128_t boot_id;
+        sd_id128_t seqnum_id;
         uint64_t arena_offset;
         uint64_t arena_size;
         uint64_t arena_max_size;
@@ -133,7 +135,6 @@ _packed_ struct Header {
         uint64_t tail_entry_offset;
         uint64_t last_bisect_offset;
         uint64_t n_objects;
-        uint64_t seqnum_base;
         uint64_t seqnum;
 };
 

src/journal/journal-file.h

@@ -1,7 +1,7 @@
 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
 
-#ifndef foojournalprivatehfoo
-#define foojournalprivatehfoo
+#ifndef foojournalfilehfoo
+#define foojournalfilehfoo
 
 /***
   This file is part of systemd.
@@ -24,7 +24,6 @@
 
 #include <inttypes.h>
 
-#include "sd-journal.h"
 #include "journal-def.h"
 #include "util.h"
 #include "sd-id128.h"
@@ -50,24 +49,25 @@ typedef struct JournalFile {
         uint64_t window_offset;
         uint64_t window_size;
 
-        Object *current;
         uint64_t current_offset;
 } JournalFile;
 
-typedef struct JournalCoursor {
-        sd_id128_t file_id;
-        sd_id128_t boot_id;
+typedef struct JournalCursor {
+        uint8_t version;
+        uint8_t reserved[7];
         uint64_t seqnum;
+        sd_id128_t seqnum_id;
+        sd_id128_t boot_id;
         uint64_t monotonic;
         uint64_t realtime;
         uint64_t xor_hash;
-} JournalCoursor;
+} JournalCursor;
 
 int journal_file_open(const char *fname, int flags, mode_t mode, JournalFile **ret);
 
 void journal_file_close(JournalFile *j);
 
-int journal_file_move_to_object(JournalFile *f, uint64_t offset, Object **ret);
+int journal_file_move_to_object(JournalFile *f, uint64_t offset, int type, Object **ret);
 
 uint64_t journal_file_entry_n_items(Object *o);
 

src/journal/journalctl.c

@@ -23,7 +23,7 @@
 #include <errno.h>
 #include <stddef.h>
 
-#include "journal-private.h"
+#include "journal-file.h"
 
 int main(int argc, char *argv[]) {
         int r;
@@ -62,21 +62,16 @@ int main(int argc, char *argv[]) {
                         uint64_t p, l;
 
                         p = le64toh(o->entry.items[i].object_offset);
-                        r = journal_file_move_to_object(f, p, &o);
+                        r = journal_file_move_to_object(f, p, OBJECT_DATA, &o);
                         if (r < 0) {
                                 log_error("Failed to move to data: %s", strerror(-r));
                                 goto finish;
                         }
 
-                        if (le64toh(o->object.type) != OBJECT_DATA) {
-                                log_error("Invalid file");
-                                goto finish;
-                        }
-
                         l = o->object.size - offsetof(Object, data.payload);
                         printf("\t[%.*s]\n", (int) l, o->data.payload);
 
-                        r = journal_file_move_to_object(f, offset, &o);
+                        r = journal_file_move_to_object(f, offset, OBJECT_ENTRY, &o);
                         if (r < 0) {
                                 log_error("Failed to move back to entry: %s", strerror(-r));
                                 goto finish;

src/journal/journald.c

@@ -29,7 +29,7 @@
 #include <acl/libacl.h>
 
 #include "hashmap.h"
-#include "journal-private.h"
+#include "journal-file.h"
 #include "sd-daemon.h"
 #include "socket-util.h"
 #include "acl-util.h"
@@ -282,7 +282,9 @@ static int process_event(Server *s, struct epoll_event *ev) {
                 log_debug("Received SIG%s", signal_to_string(sfsi.ssi_signo));
                 return 0;
 
-        } else {
+        }
+
+        if (ev->data.fd == s->syslog_fd) {
                 for (;;) {
                         char buf[LINE_MAX+1];
                         struct msghdr msghdr;
@@ -339,9 +341,12 @@ static int process_event(Server *s, struct epoll_event *ev) {
 
                         process_message(s, strstrip(buf), ucred, tv);
                 }
+
+                return 1;
         }
 
-        return 1;
+        log_error("Unknown event.");
+        return 0;
 }
 
 static int server_init(Server *s) {

src/journal/sd-journal.h

@@ -25,13 +25,12 @@
 #include <inttypes.h>
 #include <sys/types.h>
 
-#include "sd-id128.h"
-
 /* TODO:
  *
  *   - implement rotation
  *   - check LE/BE conversion for 8bit, 16bit, 32bit values
  *   - implement parallel traversal
+ *   - implement inotify usage on client
  *   - implement audit gateway
  *   - implement native gateway
  *   - extend hash table/bisect table as we go
@@ -45,12 +44,13 @@ void sd_journal_close(sd_journal *j);
 int sd_journal_previous(sd_journal *j);
 int sd_journal_next(sd_journal *j);
 
-void* sd_journal_get(sd_journal *j, const char *field, size_t *size);
-uint64_t sd_journal_get_seqnum(sd_journal *j);
-uint64_t sd_journal_get_realtime_usec(sd_journal *j);
-uint64_t sd_journal_get_monotonic_usec(sd_journal *j);
+int sd_journal_get(sd_journal *j, const char *field, const void **data, size_t *size);
+int sd_journal_get_seqnum(sd_journal *j, uint64_t *ret);
+int sd_journal_get_realtime_usec(sd_journal *j, uint64_t *ret);
+int sd_journal_get_monotonic_usec(sd_journal *j, uint64_t *ret);
 
-int sd_journal_add_match(sd_journal *j, const char *item, size_t *size);
+int sd_journal_add_match(sd_journal *j, const char *field, const void *data, size_t size);
+void sd_journal_flush_matches(sd_journal *j);
 
 int sd_journal_seek_head(sd_journal *j);
 int sd_journal_seek_tail(sd_journal *j);
@@ -59,16 +59,9 @@ int sd_journal_seek_seqnum(sd_journal *j, uint64_t seqnum);
 int sd_journal_seek_monotonic_usec(sd_journal *j, uint64_t usec);
 int sd_journal_seek_realtime_usec(sd_journal *j, uint64_t usec);
 
-uint64_t sd_journal_get_max_size(sd_journal *j);
-uint64_t sd_journal_get_min_size(sd_journal *j);
-uint64_t sd_journal_get_keep_free(sd_journal *j);
+int sd_journal_get_cursor(sd_journal *j, void **cursor, size_t *size);
+int sd_journal_set_cursor(sd_journal *j, const void *cursor, size_t size);
 
-int sd_journal_set_max_size(sd_journal *j, uint64_t size);
-int sd_journal_set_min_size(sd_journal *j, uint64_t size);
-int sd_journal_set_keep_free(sd_journal *j, uint64_t size);
-
-sd_id128_t sd_journal_get_file_id(sd_journal *j);
-sd_id128_t sd_journal_get_machine_id(sd_journal *j);
-sd_id128_t sd_journal_get_boot_id(sd_journal *j);
+int sd_journal_get_fd(sd_journal *j);
 
 #endif

src/journal/test-journal.c

@@ -21,7 +21,7 @@
 
 #include <fcntl.h>
 
-#include "journal-private.h"
+#include "journal-file.h"
 #include "log.h"
 
 int main(int argc, char *argv[]) {