src/bin/lttng-relayd/viewer-stream.c

   1 /*
   2  * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
   3  *                      David Goulet <dgoulet@efficios.com>
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License, version 2 only, as
   7  * published by the Free Software Foundation.
   8  *
   9  * This program is distributed in the hope that it will be useful, but WITHOUT
  10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  12  * more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along with
  15  * this program; if not, write to the Free Software Foundation, Inc., 51
  16  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  17  */
  18
  19 #define _GNU_SOURCE
  20 #define _LGPL_SOURCE
  21 #include <common/common.h>
  22 #include <common/index/index.h>
  23
  24 #include "lttng-relayd.h"
  25 #include "viewer-stream.h"
  26
  27 static void free_stream(struct relay_viewer_stream *stream)
  28 {
  29         assert(stream);
  30
  31         free(stream->path_name);
  32         free(stream->channel_name);
  33         free(stream);
  34 }
  35
  36 static void deferred_free_viewer_stream(struct rcu_head *head)
  37 {
  38         struct relay_viewer_stream *stream =
  39                 caa_container_of(head, struct relay_viewer_stream, rcu_node);
  40
  41         free_stream(stream);
  42 }
  43
  44 struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
  45                 enum lttng_viewer_seek seek_t, struct ctf_trace *ctf_trace)
  46 {
  47         struct relay_viewer_stream *vstream;
  48
  49         assert(stream);
  50         assert(ctf_trace);
  51
  52         vstream = zmalloc(sizeof(*vstream));
  53         if (!vstream) {
  54                 PERROR("relay viewer stream zmalloc");
  55                 goto error;
  56         }
  57
  58         vstream->session_id = stream->session_id;
  59         vstream->stream_handle = stream->stream_handle;
  60         vstream->path_name = strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
  61         if (vstream->path_name == NULL) {
  62                 PERROR("relay viewer path_name alloc");
  63                 goto error;
  64         }
  65         vstream->channel_name = strndup(stream->channel_name,
  66                         LTTNG_VIEWER_NAME_MAX);
  67         if (vstream->channel_name == NULL) {
  68                 PERROR("relay viewer channel_name alloc");
  69                 goto error;
  70         }
  71         vstream->tracefile_count = stream->tracefile_count;
  72         vstream->metadata_flag = stream->metadata_flag;
  73         vstream->tracefile_count_last = -1ULL;
  74
  75         switch (seek_t) {
  76         case LTTNG_VIEWER_SEEK_BEGINNING:
  77                 vstream->tracefile_count_current = stream->oldest_tracefile_id;
  78                 break;
  79         case LTTNG_VIEWER_SEEK_LAST:
  80                 vstream->tracefile_count_current = stream->tracefile_count_current;
  81                 break;
  82         default:
  83                 assert(0);
  84                 goto error;
  85         }
  86
  87         if (vstream->metadata_flag) {
  88                 ctf_trace->viewer_metadata_stream = vstream;
  89         }
  90
  91         /* Globally visible after the add unique. */
  92         lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
  93         lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
  94
  95         vstream->index_read_fd = -1;
  96         vstream->read_fd = -1;
  97
  98         /*
  99          * This is to avoid a race between the initialization of this object and
 100          * the close of the given stream. If the stream is unable to find this
 101          * viewer stream when closing, this copy will at least take the latest
 102          * value. We also need that for the seek_last.
 103          */
 104         vstream->total_index_received = stream->total_index_received;
 105
 106         /*
 107          * If we never received an index for the current stream, delay the opening
 108          * of the index, otherwise open it right now.
 109          */
 110         if (vstream->tracefile_count_current == stream->tracefile_count_current
 111                         && vstream->total_index_received == 0) {
 112                 vstream->index_read_fd = -1;
 113         } else {
 114                 int read_fd;
 115
 116                 read_fd = index_open(vstream->path_name, vstream->channel_name,
 117                                 vstream->tracefile_count, vstream->tracefile_count_current);
 118                 if (read_fd < 0) {
 119                         goto error;
 120                 }
 121                 vstream->index_read_fd = read_fd;
 122         }
 123
 124         if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_read_fd >= 0) {
 125                 off_t lseek_ret;
 126
 127                 lseek_ret = lseek(vstream->index_read_fd,
 128                                 vstream->total_index_received * sizeof(struct ctf_packet_index),
 129                                 SEEK_CUR);
 130                 if (lseek_ret < 0) {
 131                         goto error;
 132                 }
 133                 vstream->last_sent_index = vstream->total_index_received;
 134         }
 135
 136         return vstream;
 137
 138 error:
 139         if (vstream) {
 140                 free_stream(vstream);
 141         }
 142         return NULL;
 143 }
 144
 145 void viewer_stream_delete(struct relay_viewer_stream *stream)
 146 {
 147         int ret;
 148         struct lttng_ht_iter iter;
 149
 150         iter.iter.node = &stream->stream_n.node;
 151         ret = lttng_ht_del(viewer_streams_ht, &iter);
 152         assert(!ret);
 153 }
 154
 155 void viewer_stream_destroy(struct ctf_trace *ctf_trace,
 156                 struct relay_viewer_stream *stream)
 157 {
 158         int ret;
 159
 160         assert(stream);
 161
 162         if (ctf_trace) {
 163                 ctf_trace_put_ref(ctf_trace);
 164         }
 165
 166         if (stream->read_fd >= 0) {
 167                 ret = close(stream->read_fd);
 168                 if (ret < 0) {
 169                         PERROR("close read_fd");
 170                 }
 171         }
 172         if (stream->index_read_fd >= 0) {
 173                 ret = close(stream->index_read_fd);
 174                 if (ret < 0) {
 175                         PERROR("close index_read_fd");
 176                 }
 177         }
 178
 179         call_rcu(&stream->rcu_node, deferred_free_viewer_stream);
 180 }
 181
 182 /*
 183  * Find viewer stream by id. RCU read side lock MUST be acquired.
 184  *
 185  * Return stream if found else NULL.
 186  */
 187 struct relay_viewer_stream *viewer_stream_find_by_id(uint64_t id)
 188 {
 189         struct lttng_ht_node_u64 *node;
 190         struct lttng_ht_iter iter;
 191         struct relay_viewer_stream *stream = NULL;
 192
 193         lttng_ht_lookup(viewer_streams_ht, &id, &iter);
 194         node = lttng_ht_iter_get_node_u64(&iter);
 195         if (!node) {
 196                 DBG("Relay viewer stream %" PRIu64 " not found", id);
 197                 goto end;
 198         }
 199         stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
 200
 201 end:
 202         return stream;
 203 }
 204
 205 /*
 206  * Rotate a stream to the next tracefile.
 207  *
 208  * Must be called with viewer_stream_rotation_lock held.
 209  * Returns 0 on success, 1 on EOF, a negative value on error.
 210  */
 211 int viewer_stream_rotate(struct relay_viewer_stream *vstream,
 212                 struct relay_stream *stream)
 213 {
 214         int ret;
 215         uint64_t tracefile_id;
 216
 217         assert(vstream);
 218         assert(stream);
 219
 220         if (vstream->tracefile_count == 0) {
 221                 /* Ignore rotation, there is none to do. */
 222                 ret = 0;
 223                 goto end;
 224         }
 225
 226         tracefile_id = (vstream->tracefile_count_current + 1) %
 227                 vstream->tracefile_count;
 228
 229         /* Detect the last tracefile to open. */
 230         if (vstream->tracefile_count_last != -1ULL &&
 231                         vstream->tracefile_count_last ==
 232                         vstream->tracefile_count_current) {
 233                 ret = 1;
 234                 goto end;
 235         }
 236
 237         /*
 238          * The writer and the reader are not working in the same tracefile, we can
 239          * read up to EOF, we don't care about the total_index_received.
 240          */
 241         if (stream->close_flag || (stream->tracefile_count_current != tracefile_id)) {
 242                 vstream->close_write_flag = 1;
 243         } else {
 244                 /*
 245                  * We are opening a file that is still open in write, make sure we
 246                  * limit our reading to the number of indexes received.
 247                  */
 248                 vstream->close_write_flag = 0;
 249                 if (stream->close_flag) {
 250                         vstream->total_index_received = stream->total_index_received;
 251                 }
 252         }
 253         vstream->tracefile_count_current = tracefile_id;
 254
 255         ret = close(vstream->index_read_fd);
 256         if (ret < 0) {
 257                 PERROR("close index file %d", vstream->index_read_fd);
 258         }
 259         vstream->index_read_fd = -1;
 260
 261         ret = close(vstream->read_fd);
 262         if (ret < 0) {
 263                 PERROR("close tracefile %d", vstream->read_fd);
 264         }
 265         vstream->read_fd = -1;
 266
 267         pthread_mutex_lock(&vstream->overwrite_lock);
 268         vstream->abort_flag = 0;
 269         pthread_mutex_unlock(&vstream->overwrite_lock);
 270
 271         ret = index_open(vstream->path_name, vstream->channel_name,
 272                         vstream->tracefile_count, vstream->tracefile_count_current);
 273         if (ret < 0) {
 274                 goto error;
 275         }
 276         vstream->index_read_fd = ret;
 277
 278         ret = 0;
 279
 280 end:
 281 error:
 282         return ret;
 283 }