src/plugins/ctf/lttng-live/data-stream.c

   1 /*
   2  * SPDX-License-Identifier: MIT
   3  *
   4  * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
   5  * Copyright 2016 Philippe Proulx <pproulx@efficios.com>
   6  * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   7  * Copyright 2010-2011 EfficiOS Inc. and Linux Foundation
   8  */
   9
  10 #define BT_COMP_LOG_SELF_COMP self_comp
  11 #define BT_LOG_OUTPUT_LEVEL log_level
  12 #define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE/DS"
  13 #include "logging/comp-logging.h"
  14
  15 #include <inttypes.h>
  16 #include <stdio.h>
  17 #include <stdlib.h>
  18
  19 #include <glib.h>
  20
  21 #include <babeltrace2/babeltrace.h>
  22
  23 #include "../common/msg-iter/msg-iter.h"
  24 #include "common/assert.h"
  25 #include "compat/mman.h"
  26 #include "data-stream.h"
  27
  28 #define STREAM_NAME_PREFIX      "stream-"
  29
  30 static
  31 enum ctf_msg_iter_medium_status medop_request_bytes(
  32                 size_t request_sz, uint8_t **buffer_addr,
  33                 size_t *buffer_sz, void *data)
  34 {
  35         enum ctf_msg_iter_medium_status status =
  36                 CTF_MSG_ITER_MEDIUM_STATUS_OK;
  37         struct lttng_live_stream_iterator *stream = data;
  38         struct lttng_live_trace *trace = stream->trace;
  39         struct lttng_live_session *session = trace->session;
  40         struct lttng_live_msg_iter *live_msg_iter = session->lttng_live_msg_iter;
  41         uint64_t recv_len = 0;
  42         uint64_t len_left;
  43         uint64_t read_len;
  44
  45         BT_ASSERT(request_sz);
  46
  47         if (stream->has_stream_hung_up) {
  48                 status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
  49                 goto end;
  50         }
  51
  52         len_left = stream->base_offset + stream->len - stream->offset;
  53         if (!len_left) {
  54                 lttng_live_stream_iterator_set_state(stream,
  55                                 LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
  56                 status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
  57                 goto end;
  58         }
  59
  60         read_len = MIN(request_sz, stream->buflen);
  61         read_len = MIN(read_len, len_left);
  62         status = lttng_live_get_stream_bytes(live_msg_iter,
  63                         stream, stream->buf, stream->offset,
  64                         read_len, &recv_len);
  65         *buffer_addr = stream->buf;
  66         *buffer_sz = recv_len;
  67         stream->offset += recv_len;
  68 end:
  69         return status;
  70 }
  71
  72 static
  73 bt_stream *medop_borrow_stream(bt_stream_class *stream_class,
  74                 int64_t stream_id, void *data)
  75 {
  76         struct lttng_live_stream_iterator *lttng_live_stream = data;
  77         bt_logging_level log_level = lttng_live_stream->log_level;
  78         bt_self_component *self_comp = lttng_live_stream->self_comp;
  79
  80         if (!lttng_live_stream->stream) {
  81                 uint64_t stream_class_id = bt_stream_class_get_id(stream_class);
  82
  83                 BT_COMP_LOGI("Creating stream %s (ID: %" PRIu64 ") out of stream "
  84                         "class %" PRId64, lttng_live_stream->name->str,
  85                         stream_id, stream_class_id);
  86
  87                 if (stream_id < 0) {
  88                         /*
  89                          * No stream instance ID in the stream. It's possible
  90                          * to encounter this situation with older version of
  91                          * LTTng. In these cases, use the viewer_stream_id that
  92                          * is unique for a live viewer session.
  93                          */
  94                         lttng_live_stream->stream = bt_stream_create_with_id(
  95                                 stream_class, lttng_live_stream->trace->trace,
  96                                 lttng_live_stream->viewer_stream_id);
  97                 } else {
  98                         lttng_live_stream->stream = bt_stream_create_with_id(
  99                                 stream_class, lttng_live_stream->trace->trace,
 100                                 (uint64_t) stream_id);
 101                 }
 102
 103                 if (!lttng_live_stream->stream) {
 104                         BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 105                                 "Cannot create stream %s (stream class ID "
 106                                 "%" PRId64 ", stream ID %" PRIu64 ")",
 107                                 lttng_live_stream->name->str,
 108                                 stream_class_id, stream_id);
 109                         goto end;
 110                 }
 111
 112                 bt_stream_set_name(lttng_live_stream->stream,
 113                         lttng_live_stream->name->str);
 114         }
 115
 116 end:
 117         return lttng_live_stream->stream;
 118 }
 119
 120 static struct ctf_msg_iter_medium_ops medops = {
 121         .request_bytes = medop_request_bytes,
 122         .seek = NULL,
 123         .borrow_stream = medop_borrow_stream,
 124 };
 125
 126 BT_HIDDEN
 127 enum lttng_live_iterator_status lttng_live_lazy_msg_init(
 128                 struct lttng_live_session *session,
 129                 bt_self_message_iterator *self_msg_iter)
 130 {
 131         struct lttng_live_component *lttng_live =
 132                 session->lttng_live_msg_iter->lttng_live_comp;
 133         uint64_t trace_idx, stream_iter_idx;
 134         bt_logging_level log_level = session->log_level;
 135         bt_self_component *self_comp = session->self_comp;
 136
 137         if (!session->lazy_stream_msg_init) {
 138                 return LTTNG_LIVE_ITERATOR_STATUS_OK;
 139         }
 140
 141         BT_COMP_LOGD("Lazily initializing self message iterator for live session: "
 142                 "session-id=%"PRIu64", self-msg-iter-addr=%p", session->id,
 143                 self_msg_iter);
 144
 145         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 146                 struct lttng_live_trace *trace =
 147                         g_ptr_array_index(session->traces, trace_idx);
 148
 149                 for (stream_iter_idx = 0;
 150                                 stream_iter_idx < trace->stream_iterators->len;
 151                                 stream_iter_idx++) {
 152                         struct ctf_trace_class *ctf_tc;
 153                         struct lttng_live_stream_iterator *stream_iter =
 154                                 g_ptr_array_index(trace->stream_iterators,
 155                                         stream_iter_idx);
 156
 157                         if (stream_iter->msg_iter) {
 158                                 continue;
 159                         }
 160                         ctf_tc = ctf_metadata_decoder_borrow_ctf_trace_class(
 161                                 trace->metadata->decoder);
 162                         BT_COMP_LOGD("Creating CTF message iterator: "
 163                                 "session-id=%"PRIu64", ctf-tc-addr=%p, "
 164                                 "stream-iter-name=%s, self-msg-iter-addr=%p",
 165                                 session->id, ctf_tc, stream_iter->name->str, self_msg_iter);
 166                         stream_iter->msg_iter = ctf_msg_iter_create(ctf_tc,
 167                                 lttng_live->max_query_size, medops, stream_iter,
 168                                 log_level, self_comp, self_msg_iter);
 169                         if (!stream_iter->msg_iter) {
 170                                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 171                                         "Failed to create CTF message iterator");
 172                                 goto error;
 173                         }
 174                 }
 175         }
 176
 177         session->lazy_stream_msg_init = false;
 178
 179         return LTTNG_LIVE_ITERATOR_STATUS_OK;
 180
 181 error:
 182         return LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 183 }
 184
 185 BT_HIDDEN
 186 struct lttng_live_stream_iterator *lttng_live_stream_iterator_create(
 187                 struct lttng_live_session *session,
 188                 uint64_t ctf_trace_id,
 189                 uint64_t stream_id,
 190                 bt_self_message_iterator *self_msg_iter)
 191 {
 192         struct lttng_live_stream_iterator *stream_iter;
 193         struct lttng_live_component *lttng_live;
 194         struct lttng_live_trace *trace;
 195         bt_logging_level log_level;
 196         bt_self_component *self_comp;
 197
 198         BT_ASSERT(session);
 199         BT_ASSERT(session->lttng_live_msg_iter);
 200         BT_ASSERT(session->lttng_live_msg_iter->lttng_live_comp);
 201         log_level = session->log_level;
 202         self_comp = session->self_comp;
 203
 204         lttng_live = session->lttng_live_msg_iter->lttng_live_comp;
 205
 206         stream_iter = g_new0(struct lttng_live_stream_iterator, 1);
 207         if (!stream_iter) {
 208                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 209                         "Failed to allocate struct lttng_live_stream_iterator");
 210                 goto error;
 211         }
 212
 213         stream_iter->log_level = log_level;
 214         stream_iter->self_comp = self_comp;
 215         trace = lttng_live_session_borrow_or_create_trace_by_id(session, ctf_trace_id);
 216         if (!trace) {
 217                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 218                         "Failed to borrow CTF trace.");
 219                 goto error;
 220         }
 221
 222         stream_iter->trace = trace;
 223         stream_iter->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
 224         stream_iter->viewer_stream_id = stream_id;
 225         stream_iter->ctf_stream_class_id = -1ULL;
 226         stream_iter->last_inactivity_ts = INT64_MIN;
 227
 228         if (trace->trace) {
 229                 struct ctf_trace_class *ctf_tc =
 230                         ctf_metadata_decoder_borrow_ctf_trace_class(
 231                                 trace->metadata->decoder);
 232                 BT_ASSERT(!stream_iter->msg_iter);
 233                 stream_iter->msg_iter = ctf_msg_iter_create(ctf_tc,
 234                         lttng_live->max_query_size, medops, stream_iter,
 235                         log_level, self_comp, self_msg_iter);
 236                 if (!stream_iter->msg_iter) {
 237                         BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 238                                 "Failed to create CTF message iterator");
 239                         goto error;
 240                 }
 241         }
 242         stream_iter->buf = g_new0(uint8_t, lttng_live->max_query_size);
 243         if (!stream_iter->buf) {
 244                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 245                         "Failed to allocate live stream iterator buffer");
 246                 goto error;
 247         }
 248
 249         stream_iter->buflen = lttng_live->max_query_size;
 250         stream_iter->name = g_string_new(NULL);
 251         if (!stream_iter->name) {
 252                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 253                         "Failed to allocate live stream iterator name buffer");
 254                 goto error;
 255         }
 256
 257         g_string_printf(stream_iter->name, STREAM_NAME_PREFIX "%" PRIu64,
 258                 stream_iter->viewer_stream_id);
 259         g_ptr_array_add(trace->stream_iterators, stream_iter);
 260
 261         /* Track the number of active stream iterator. */
 262         session->lttng_live_msg_iter->active_stream_iter++;
 263
 264         goto end;
 265 error:
 266         lttng_live_stream_iterator_destroy(stream_iter);
 267         stream_iter = NULL;
 268 end:
 269         return stream_iter;
 270 }
 271
 272 BT_HIDDEN
 273 void lttng_live_stream_iterator_destroy(
 274                 struct lttng_live_stream_iterator *stream_iter)
 275 {
 276         if (!stream_iter) {
 277                 return;
 278         }
 279
 280         if (stream_iter->stream) {
 281                 BT_STREAM_PUT_REF_AND_RESET(stream_iter->stream);
 282         }
 283
 284         if (stream_iter->msg_iter) {
 285                 ctf_msg_iter_destroy(stream_iter->msg_iter);
 286         }
 287         g_free(stream_iter->buf);
 288         if (stream_iter->name) {
 289                 g_string_free(stream_iter->name, TRUE);
 290         }
 291
 292         bt_message_put_ref(stream_iter->current_msg);
 293
 294         /* Track the number of active stream iterator. */
 295         stream_iter->trace->session->lttng_live_msg_iter->active_stream_iter--;
 296
 297         g_free(stream_iter);
 298 }