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
 115 uint64_t reader_buffered_space(const struct reader_pipe *reader)
 116 {
 117         assert(reader);
 118         assert(reader->buf_ptr);
 119         return log_storage_reader_get_ready_bytes(reader->log_storage_reader_ptr);
 120 }
 121
 122 int reader_is_bufferable(const struct reader_pipe *reader)
 123 {
 124         assert(reader);
 125         return reader->buf_ptr && reader->file.path != NULL;
 126 }
 127
 128 int reader_ms_since(const struct reader_pipe *reader, struct timespec *ts)
 129 {
 130         return (ts->tv_sec - reader->last_read_time.tv_sec) * 1000 + (ts->tv_nsec - reader->last_read_time.tv_nsec) / 1000000;
 131 }
 132
 133 int reader_should_buffer(struct reader_pipe *reader, const struct buf_params *buf_params, struct timespec now)
 134 {
 135         assert(reader);
 136         assert(buf_params);
 137
 138         if (!reader_is_bufferable(reader))
 139                 return 0;
 140
 141         if (reader_buffered_space(reader) < (uint64_t)buf_params->bytes && reader_ms_since(reader, &now) < (buf_params->time * 1000))
 142                 return 1;
 143
 144         reader->last_read_time = now;
 145         return 0;
 146 }
 147
 148 int reader_print_out_single_log(struct reader_pipe *reader, const dlogutil_entry_s *dlogutil_entry)
 149 {
 150         assert(reader);
 151         assert(reader->buf_ptr);
 152         assert(dlogutil_entry);
 153
 154         if (!log_should_print_line(reader->filter, dlogutil_entry))
 155                 return 0;
 156
 157         if (reader->file.path) {
 158                 logfile_write_with_rotation(dlogutil_entry, &reader->file, reader->buf_ptr->sort_by);
 159                 return 0;
 160         }
 161
 162         const char *tag = dlogutil_entry->msg + 1;
 163         if (!strlen(tag))
 164                 return 0;
 165
 166         int r = write(reader->file.path ? reader->file.fd : reader->common.fd_entity_sink.fd, dlogutil_entry, dlogutil_entry->len);
 167         if (r < 0) {
 168                 if (errno != EAGAIN)
 169                         return 1;
 170
 171                 /* The pipe is just clogged, this is not an actual error.
 172                  * We own the entry so it needs to be saved for later. */
 173                 r = 0;
 174         }
 175
 176         reader->file.size += r;
 177         if (r < dlogutil_entry->len) {
 178                 reader->partial_log_size = dlogutil_entry->len - r;
 179                 memcpy(reader->partial_log, ((char *)dlogutil_entry) + r, reader->partial_log_size);
 180                 return -1;
 181         } else if (logfile_rotate_needed(&reader->file) > 0) {
 182                 logfile_do_rotate(&reader->file);
 183         }
 184
 185         return 0;
 186 }
 187
 188 /**
 189  * @brief Print out logs
 190  * @details Make sure the reader is up to date on printed logs
 191  * @param[in] reader The reader to read the data
 192  * @param[in] _time Unused timestamps
 193  * @return 0 if data remains for the next iteration, 1 if the buffer is to be removed, else -1
 194  */
 195 int print_out_logs(struct reader_pipe *reader, struct now_t _time)
 196 {
 197         assert(reader);
 198
 199         assert(reader->buf_ptr);
 200
 201         if (reader->partial_log_size) {
 202                 int r = write(reader->common.fd_entity_sink.fd, reader->partial_log, reader->partial_log_size);
 203                 if (r <= 0)
 204                         return r != 0 && errno != EAGAIN;
 205
 206                 if (r < reader->partial_log_size) {
 207                         reader->partial_log_size -= r;
 208                         memmove(reader->partial_log, reader->partial_log + r, reader->partial_log_size);
 209                         return 0;
 210                 }
 211
 212                 reader->partial_log_size = 0;
 213         }
 214
 215         while (log_storage_reader_is_new_entry_available(reader->log_storage_reader_ptr)) {
 216                 const dlogutil_entry_s* ple = (const dlogutil_entry_s *)log_storage_reader_get_new_entry(reader->log_storage_reader_ptr);
 217
 218                 assert(ple);
 219
 220                 switch (reader_print_out_single_log(reader, ple)) {
 221                 case 0: /* nothing more to do, let's do next loop */
 222                         break;
 223
 224                 case 1: /* error after which we need to end the reader */
 225                         return 1;
 226
 227                 default: /* writing error, bounce out */
 228                         return -1;
 229                 }
 230         }
 231
 232         return reader->is_dumping ? 1 : -1;
 233 }
 234