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

   1 /*
   2  * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
   3  *                      David Goulet <dgoulet@efficios.com>
   4  *               2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  *
   6  * This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU General Public License, version 2 only, as
   8  * published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but WITHOUT
  11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  13  * more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along with
  16  * this program; if not, write to the Free Software Foundation, Inc., 51
  17  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18  */
  19
  20 #define _GNU_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 viewer_stream_destroy(struct relay_viewer_stream *vstream)
  28 {
  29         free(vstream->path_name);
  30         free(vstream->channel_name);
  31         free(vstream);
  32 }
  33
  34 static void viewer_stream_destroy_rcu(struct rcu_head *head)
  35 {
  36         struct relay_viewer_stream *vstream =
  37                 caa_container_of(head, struct relay_viewer_stream, rcu_node);
  38
  39         viewer_stream_destroy(vstream);
  40 }
  41
  42 struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
  43                 enum lttng_viewer_seek seek_t)
  44 {
  45         struct relay_viewer_stream *vstream;
  46
  47         vstream = zmalloc(sizeof(*vstream));
  48         if (!vstream) {
  49                 PERROR("relay viewer stream zmalloc");
  50                 goto error;
  51         }
  52
  53         vstream->path_name = strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
  54         if (vstream->path_name == NULL) {
  55                 PERROR("relay viewer path_name alloc");
  56                 goto error;
  57         }
  58         vstream->channel_name = strndup(stream->channel_name,
  59                         LTTNG_VIEWER_NAME_MAX);
  60         if (vstream->channel_name == NULL) {
  61                 PERROR("relay viewer channel_name alloc");
  62                 goto error;
  63         }
  64
  65         if (!stream_get(stream)) {
  66                 ERR("Cannot get stream");
  67                 goto error;
  68         }
  69         vstream->stream = stream;
  70
  71         pthread_mutex_lock(&stream->lock);
  72
  73         if (stream->is_metadata && stream->trace->viewer_metadata_stream) {
  74                 ERR("Cannot attach viewer metadata stream to trace (busy).");
  75                 goto error_unlock;
  76         }
  77
  78         switch (seek_t) {
  79         case LTTNG_VIEWER_SEEK_BEGINNING:
  80         {
  81                 uint64_t seq_tail = tracefile_array_get_seq_tail(stream->tfa);
  82
  83                 if (seq_tail == -1ULL) {
  84                         /*
  85                          * Tail may not be initialized yet. Nonetheless, we know
  86                          * we want to send the first index once it becomes
  87                          * available.
  88                          */
  89                         seq_tail = 0;
  90                 }
  91                 vstream->current_tracefile_id =
  92                         tracefile_array_get_file_index_tail(stream->tfa);
  93                 vstream->index_sent_seqcount = seq_tail;
  94                 break;
  95         }
  96         case LTTNG_VIEWER_SEEK_LAST:
  97                 vstream->current_tracefile_id =
  98                         tracefile_array_get_file_index_head(stream->tfa);
  99                 /*
 100                  * We seek at the very end of each stream, awaiting for
 101                  * a future packet to eventually come in.
 102                  *
 103                  * We don't need to check the head position for -1ULL since the
 104                  * increment will set it to 0.
 105                  */
 106                 vstream->index_sent_seqcount =
 107                                 tracefile_array_get_seq_head(stream->tfa) + 1;
 108                 break;
 109         default:
 110                 goto error_unlock;
 111         }
 112
 113         /*
 114          * If we never received an index for the current stream, delay
 115          * the opening of the index, otherwise open it right now.
 116          */
 117         if (stream->index_received_seqcount == 0) {
 118                 vstream->index_fd = NULL;
 119         } else {
 120                 int read_fd;
 121
 122                 read_fd = index_open(vstream->path_name, vstream->channel_name,
 123                                 stream->tracefile_count,
 124                                 vstream->current_tracefile_id);
 125                 if (read_fd < 0) {
 126                         goto error_unlock;
 127                 }
 128                 vstream->index_fd = stream_fd_create(read_fd);
 129                 if (!vstream->index_fd) {
 130                         if (close(read_fd)) {
 131                                 PERROR("close");
 132                         }
 133                         goto error_unlock;
 134                 }
 135         }
 136
 137         if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_fd) {
 138                 off_t lseek_ret;
 139
 140                 lseek_ret = lseek(vstream->index_fd->fd, 0, SEEK_END);
 141                 if (lseek_ret < 0) {
 142                         goto error_unlock;
 143                 }
 144         }
 145         if (stream->is_metadata) {
 146                 rcu_assign_pointer(stream->trace->viewer_metadata_stream,
 147                                 vstream);
 148         }
 149         pthread_mutex_unlock(&stream->lock);
 150
 151         /* Globally visible after the add unique. */
 152         lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
 153         lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
 154
 155         pthread_mutex_init(&vstream->reflock, NULL);
 156         urcu_ref_init(&vstream->ref);
 157
 158         return vstream;
 159
 160 error_unlock:
 161         pthread_mutex_unlock(&stream->lock);
 162 error:
 163         if (vstream) {
 164                 viewer_stream_destroy(vstream);
 165         }
 166         return NULL;
 167 }
 168
 169 static void viewer_stream_unpublish(struct relay_viewer_stream *vstream)
 170 {
 171         int ret;
 172         struct lttng_ht_iter iter;
 173
 174         iter.iter.node = &vstream->stream_n.node;
 175         ret = lttng_ht_del(viewer_streams_ht, &iter);
 176         assert(!ret);
 177 }
 178
 179 static void viewer_stream_release(struct urcu_ref *ref)
 180 {
 181         struct relay_viewer_stream *vstream = caa_container_of(ref,
 182                         struct relay_viewer_stream, ref);
 183
 184         if (vstream->stream->is_metadata) {
 185                 rcu_assign_pointer(vstream->stream->trace->viewer_metadata_stream, NULL);
 186         }
 187
 188         viewer_stream_unpublish(vstream);
 189
 190         if (vstream->stream_fd) {
 191                 stream_fd_put(vstream->stream_fd);
 192                 vstream->stream_fd = NULL;
 193         }
 194         if (vstream->index_fd) {
 195                 stream_fd_put(vstream->index_fd);
 196                 vstream->index_fd = NULL;
 197         }
 198         if (vstream->stream) {
 199                 stream_put(vstream->stream);
 200                 vstream->stream = NULL;
 201         }
 202         call_rcu(&vstream->rcu_node, viewer_stream_destroy_rcu);
 203 }
 204
 205 /* Must be called with RCU read-side lock held. */
 206 bool viewer_stream_get(struct relay_viewer_stream *vstream)
 207 {
 208         bool has_ref = false;
 209
 210         pthread_mutex_lock(&vstream->reflock);
 211         if (vstream->ref.refcount != 0) {
 212                 has_ref = true;
 213                 urcu_ref_get(&vstream->ref);
 214         }
 215         pthread_mutex_unlock(&vstream->reflock);
 216
 217         return has_ref;
 218 }
 219
 220 /*
 221  * Get viewer stream by id.
 222  *
 223  * Return viewer stream if found else NULL.
 224  */
 225 struct relay_viewer_stream *viewer_stream_get_by_id(uint64_t id)
 226 {
 227         struct lttng_ht_node_u64 *node;
 228         struct lttng_ht_iter iter;
 229         struct relay_viewer_stream *vstream = NULL;
 230
 231         rcu_read_lock();
 232         lttng_ht_lookup(viewer_streams_ht, &id, &iter);
 233         node = lttng_ht_iter_get_node_u64(&iter);
 234         if (!node) {
 235                 DBG("Relay viewer stream %" PRIu64 " not found", id);
 236                 goto end;
 237         }
 238         vstream = caa_container_of(node, struct relay_viewer_stream, stream_n);
 239         if (!viewer_stream_get(vstream)) {
 240                 vstream = NULL;
 241         }
 242 end:
 243         rcu_read_unlock();
 244         return vstream;
 245 }
 246
 247 void viewer_stream_put(struct relay_viewer_stream *vstream)
 248 {
 249         rcu_read_lock();
 250         pthread_mutex_lock(&vstream->reflock);
 251         urcu_ref_put(&vstream->ref, viewer_stream_release);
 252         pthread_mutex_unlock(&vstream->reflock);
 253         rcu_read_unlock();
 254 }
 255
 256 /*
 257  * Rotate a stream to the next tracefile.
 258  *
 259  * Must be called with the rstream lock held.
 260  * Returns 0 on success, 1 on EOF, a negative value on error.
 261  */
 262 int viewer_stream_rotate(struct relay_viewer_stream *vstream)
 263 {
 264         int ret;
 265         struct relay_stream *stream = vstream->stream;
 266         uint64_t new_id;
 267
 268         /* Detect the last tracefile to open. */
 269         if (stream->index_received_seqcount
 270                         == vstream->index_sent_seqcount
 271                         && stream->trace->session->connection_closed) {
 272                 ret = 1;
 273                 goto end;
 274         }
 275
 276         if (stream->tracefile_count == 0) {
 277                 /* Ignore rotation, there is none to do. */
 278                 ret = 0;
 279                 goto end;
 280         }
 281
 282         /*
 283          * Try to move to the next file.
 284          */
 285         new_id = (vstream->current_tracefile_id + 1)
 286                         % stream->tracefile_count;
 287         if (tracefile_array_seq_in_file(stream->tfa, new_id,
 288                         vstream->index_sent_seqcount)) {
 289                 vstream->current_tracefile_id = new_id;
 290         } else {
 291                 uint64_t seq_tail = tracefile_array_get_seq_tail(stream->tfa);
 292
 293                 /*
 294                  * This can only be reached on overwrite, which implies there
 295                  * has been data written at some point, which will have set the
 296                  * tail.
 297                  */
 298                 assert(seq_tail != -1ULL);
 299                 /*
 300                  * We need to resync because we lag behind tail.
 301                  */
 302                 vstream->current_tracefile_id =
 303                         tracefile_array_get_file_index_tail(stream->tfa);
 304                 vstream->index_sent_seqcount = seq_tail;
 305         }
 306
 307         if (vstream->index_fd) {
 308                 stream_fd_put(vstream->index_fd);
 309                 vstream->index_fd = NULL;
 310         }
 311         if (vstream->stream_fd) {
 312                 stream_fd_put(vstream->stream_fd);
 313                 vstream->stream_fd = NULL;
 314         }
 315
 316         ret = index_open(vstream->path_name, vstream->channel_name,
 317                         stream->tracefile_count,
 318                         vstream->current_tracefile_id);
 319         if (ret < 0) {
 320                 goto end;
 321         }
 322         vstream->index_fd = stream_fd_create(ret);
 323         if (vstream->index_fd) {
 324                 ret = 0;
 325         } else {
 326                 if (close(ret)) {
 327                         PERROR("close");
 328                 }
 329                 ret = -1;
 330         }
 331 end:
 332         return ret;
 333 }
 334
 335 void print_viewer_streams(void)
 336 {
 337         struct lttng_ht_iter iter;
 338         struct relay_viewer_stream *vstream;
 339
 340         rcu_read_lock();
 341         cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, vstream,
 342                         stream_n.node) {
 343                 if (!viewer_stream_get(vstream)) {
 344                         continue;
 345                 }
 346                 DBG("vstream %p refcount %ld stream %" PRIu64 " trace %" PRIu64
 347                         " session %" PRIu64,
 348                         vstream,
 349                         vstream->ref.refcount,
 350                         vstream->stream->stream_handle,
 351                         vstream->stream->trace->id,
 352                         vstream->stream->trace->session->id);
 353                 viewer_stream_put(vstream);
 354         }
 355         rcu_read_unlock();
 356 }