src/logger/reader_pipe.c

   1 #include "reader_pipe.h"
   2
   3 #include "logger_internal.h"
   4
   5 void reader_pipe_free(struct reader_pipe *reader)
   6 {
   7         if (!reader)
   8                 return;
   9
  10         reader_deinit_common(&reader->common);
  11         logfile_free(&reader->file);
  12         if (reader->log_storage_reader_ptr)
  13                 log_storage_release_reader(reader->log_storage_reader_ptr);
  14         if (reader->filter)
  15                 log_filter_free(reader->filter);
  16
  17         free(reader);
  18 }
  19
  20 void reader_pipe_cleanup(struct reader_pipe *const *ptr)
  21 {
  22         assert(ptr);
  23         reader_pipe_free(*ptr);
  24 }
  25
  26 static void dispatch_event_reader_pipe(struct logger *server, struct epoll_event *event, void *userdata)
  27 {
  28         struct reader_pipe *const rp = (struct reader_pipe *) userdata;
  29         assert(rp);
  30
  31         if (event->events & (EPOLLHUP | EPOLLERR)) {
  32                 remove_reader_fd_entities(server, &rp->common);
  33                 if (rp->buf_ptr)
  34                         list_remove(&rp->buf_ptr->readers_pipe, rp);
  35                 reader_pipe_free(rp);
  36                 return;
  37         }
  38
  39         int r = print_out_logs(rp, server->time);
  40         if (r != 0) {
  41                 /* TODO: There is no reason not to free the reader in full. However, when I do so, some tests start to
  42                  * fail without any reasonable reason. You are welcome to *try* to figure out why does this happen. */
  43                 remove_reader_fd_entities(server, &rp->common);
  44         }
  45 }
  46
  47 static struct reader_pipe *reader_pipe_alloc(dlogutil_filter_options_s *filter, struct log_file *file, struct timespec ts,
  48         bool monitor, bool is_dumping)
  49 {
  50         struct reader_pipe *ret = calloc(1, sizeof(*ret));
  51         if (!ret)
  52                 return NULL;
  53
  54         ret->filter = log_filter_move(filter);
  55         ret->monitor = monitor;
  56         ret->is_dumping = is_dumping;
  57         logfile_move(&ret->file, file);
  58         ret->buf_ptr = NULL;
  59         ret->log_storage_reader_ptr = NULL;
  60         ret->last_read_time = ts;
  61         ret->partial_log_size = 0;
  62
  63         return ret;
  64 }
  65
  66 int reader_pipe_init(struct reader_pipe **reader, log_id_t buf_id, struct logger *server,
  67         dlogutil_filter_options_s *filter, struct log_file *file, bool monitor, bool is_dumping)
  68 {
  69         assert(reader);
  70         assert(buf_id > LOG_ID_INVALID);
  71         assert(buf_id < LOG_ID_MAX);
  72         assert(server);
  73         assert(filter);
  74
  75         __attribute__((cleanup(reader_pipe_cleanup))) struct reader_pipe *ret = reader_pipe_alloc(filter, file, server->time.mono, monitor, is_dumping);
  76         if (!ret)
  77                 return -ENOMEM;
  78
  79         ret->buf_ptr = server->buffers[buf_id];
  80         if (!ret->buf_ptr)
  81                 return -EINVAL;
  82
  83         init_fd_entity(&ret->common.fd_entity_sink, dispatch_event_reader_pipe, ret);
  84         init_fd_entity(&ret->common.fd_entity_source, dispatch_event_reader_pipe, ret);
  85
  86         *reader = ret;
  87         ret = NULL;
  88         return 0;
  89 }
  90
  91 int reader_pipe_init_with_writer(struct reader_pipe **reader, struct writer *writer, struct logger *server,
  92         dlogutil_filter_options_s *filter, struct log_file *file, bool monitor, bool is_dumping)
  93 {
  94         assert(reader);
  95         assert(writer);
  96         assert(writer->buf_ptr);
  97         assert(server);
  98         assert(filter);
  99
 100         __attribute__((cleanup(reader_pipe_cleanup))) struct reader_pipe *ret = reader_pipe_alloc(filter, file, server->time.mono, monitor, is_dumping);
 101         if (!ret)
 102                 return -ENOMEM;
 103
 104         ret->buf_ptr = writer->buf_ptr;
 105
 106         init_fd_entity(&ret->common.fd_entity_sink, dispatch_event_reader_pipe, ret);
 107         init_fd_entity(&ret->common.fd_entity_source, dispatch_event_reader_pipe, ret);
 108
 109         *reader = ret;
 110         ret = NULL;
 111         return 0;
 112 }
 113
 114 int reader_print_out_single_log(struct reader_pipe *reader, const dlogutil_entry_s *dlogutil_entry)
 115 {
 116         assert(reader);
 117         assert(reader->buf_ptr);
 118         assert(dlogutil_entry);
 119
 120         if (!log_should_print_line(reader->filter, dlogutil_entry))
 121                 return 0;
 122
 123         if (reader->file.path) {
 124                 logfile_write_with_rotation(dlogutil_entry, &reader->file, reader->buf_ptr->sort_by);
 125                 return 0;
 126         }
 127
 128         const char *tag = dlogutil_entry->msg + 1;
 129         if (!strlen(tag))
 130                 return 0;
 131
 132         int r = write(reader->file.path ? reader->file.fd : reader->common.fd_entity_sink.fd, dlogutil_entry, dlogutil_entry->len);
 133         if (r < 0) {
 134                 if (errno != EAGAIN)
 135                         return 1;
 136
 137                 /* The pipe is just clogged, this is not an actual error.
 138                  * We own the entry so it needs to be saved for later. */
 139                 r = 0;
 140         }
 141
 142         reader->file.size += r;
 143         if (r < dlogutil_entry->len) {
 144                 reader->partial_log_size = dlogutil_entry->len - r;
 145                 memcpy(reader->partial_log, ((char *)dlogutil_entry) + r, reader->partial_log_size);
 146                 return -1;
 147         } else if (logfile_rotate_needed(&reader->file) > 0) {
 148                 logfile_do_rotate(&reader->file);
 149         }
 150
 151         return 0;
 152 }
 153
 154 /**
 155  * @brief Print out logs
 156  * @details Make sure the reader is up to date on printed logs
 157  * @param[in] reader The reader to read the data
 158  * @param[in] _time Unused timestamps
 159  * @return 0 if data remains for the next iteration, 1 if the buffer is to be removed, else -1
 160  */
 161 int print_out_logs(struct reader_pipe *reader, struct now_t _time)
 162 {
 163         assert(reader);
 164
 165         assert(reader->buf_ptr);
 166
 167         if (reader->partial_log_size) {
 168                 int r = write(reader->common.fd_entity_sink.fd, reader->partial_log, reader->partial_log_size);
 169                 if (r <= 0)
 170                         return r != 0 && errno != EAGAIN;
 171
 172                 if (r < reader->partial_log_size) {
 173                         reader->partial_log_size -= r;
 174                         memmove(reader->partial_log, reader->partial_log + r, reader->partial_log_size);
 175                         return 0;
 176                 }
 177
 178                 reader->partial_log_size = 0;
 179         }
 180
 181         while (log_storage_reader_is_new_entry_available(reader->log_storage_reader_ptr)) {
 182                 const dlogutil_entry_s* ple = (const dlogutil_entry_s *)log_storage_reader_get_new_entry(reader->log_storage_reader_ptr);
 183
 184                 assert(ple);
 185
 186                 switch (reader_print_out_single_log(reader, ple)) {
 187                 case 0: /* nothing more to do, let's do next loop */
 188                         break;
 189
 190                 case 1: /* error after which we need to end the reader */
 191                         return 1;
 192
 193                 default: /* writing error, bounce out */
 194                         return -1;
 195                 }
 196         }
 197
 198         return reader->is_dumping ? 1 : -1;
 199 }
 200