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

   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.hpp"
  24 #include "common/assert.h"
  25 #include "compat/mman.h"
  26 #include "data-stream.hpp"
  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         lttng_live_stream_iterator *stream = (lttng_live_stream_iterator *) 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         lttng_live_stream_iterator *lttng_live_stream = (lttng_live_stream_iterator *) 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         medop_request_bytes,
 122         nullptr,
 123         nullptr,
 124         medop_borrow_stream,
 125 };
 126
 127 BT_HIDDEN
 128 enum lttng_live_iterator_status lttng_live_lazy_msg_init(
 129                 struct lttng_live_session *session,
 130                 bt_self_message_iterator *self_msg_iter)
 131 {
 132         struct lttng_live_component *lttng_live =
 133                 session->lttng_live_msg_iter->lttng_live_comp;
 134         uint64_t trace_idx, stream_iter_idx;
 135         bt_logging_level log_level = session->log_level;
 136         bt_self_component *self_comp = session->self_comp;
 137
 138         if (!session->lazy_stream_msg_init) {
 139                 return LTTNG_LIVE_ITERATOR_STATUS_OK;
 140         }
 141
 142         BT_COMP_LOGD("Lazily initializing self message iterator for live session: "
 143                 "session-id=%" PRIu64 ", self-msg-iter-addr=%p", session->id,
 144                 self_msg_iter);
 145
 146         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 147                 struct lttng_live_trace *trace =
 148                         (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
 149
 150                 for (stream_iter_idx = 0;
 151                                 stream_iter_idx < trace->stream_iterators->len;
 152                                 stream_iter_idx++) {
 153                         struct ctf_trace_class *ctf_tc;
 154                         struct lttng_live_stream_iterator *stream_iter =
 155                                 (lttng_live_stream_iterator *) g_ptr_array_index(trace->stream_iterators,
 156                                         stream_iter_idx);
 157
 158                         if (stream_iter->msg_iter) {
 159                                 continue;
 160                         }
 161                         ctf_tc = ctf_metadata_decoder_borrow_ctf_trace_class(
 162                                 trace->metadata->decoder);
 163                         BT_COMP_LOGD("Creating CTF message iterator: "
 164                                 "session-id=%" PRIu64 ", ctf-tc-addr=%p, "
 165                                 "stream-iter-name=%s, self-msg-iter-addr=%p",
 166                                 session->id, ctf_tc, stream_iter->name->str, self_msg_iter);
 167                         stream_iter->msg_iter = ctf_msg_iter_create(ctf_tc,
 168                                 lttng_live->max_query_size, medops, stream_iter,
 169                                 log_level, self_comp, self_msg_iter);
 170                         if (!stream_iter->msg_iter) {
 171                                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 172                                         "Failed to create CTF message iterator");
 173                                 goto error;
 174                         }
 175                 }
 176         }
 177
 178         session->lazy_stream_msg_init = false;
 179
 180         return LTTNG_LIVE_ITERATOR_STATUS_OK;
 181
 182 error:
 183         return LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 184 }
 185
 186 BT_HIDDEN
 187 struct lttng_live_stream_iterator *lttng_live_stream_iterator_create(
 188                 struct lttng_live_session *session,
 189                 uint64_t ctf_trace_id,
 190                 uint64_t stream_id,
 191                 bt_self_message_iterator *self_msg_iter)
 192 {
 193         struct lttng_live_stream_iterator *stream_iter;
 194         struct lttng_live_component *lttng_live;
 195         struct lttng_live_trace *trace;
 196         bt_logging_level log_level;
 197         bt_self_component *self_comp;
 198
 199         BT_ASSERT(session);
 200         BT_ASSERT(session->lttng_live_msg_iter);
 201         BT_ASSERT(session->lttng_live_msg_iter->lttng_live_comp);
 202         log_level = session->log_level;
 203         self_comp = session->self_comp;
 204
 205         lttng_live = session->lttng_live_msg_iter->lttng_live_comp;
 206
 207         stream_iter = g_new0(struct lttng_live_stream_iterator, 1);
 208         if (!stream_iter) {
 209                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 210                         "Failed to allocate struct lttng_live_stream_iterator");
 211                 goto error;
 212         }
 213
 214         stream_iter->log_level = log_level;
 215         stream_iter->self_comp = self_comp;
 216         trace = lttng_live_session_borrow_or_create_trace_by_id(session, ctf_trace_id);
 217         if (!trace) {
 218                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 219                         "Failed to borrow CTF trace.");
 220                 goto error;
 221         }
 222
 223         stream_iter->trace = trace;
 224         stream_iter->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
 225         stream_iter->viewer_stream_id = stream_id;
 226
 227         stream_iter->ctf_stream_class_id.is_set = false;
 228         stream_iter->ctf_stream_class_id.value = UINT64_MAX;
 229
 230         stream_iter->last_inactivity_ts.is_set = false;
 231         stream_iter->last_inactivity_ts.value = 0;
 232
 233         if (trace->trace) {
 234                 struct ctf_trace_class *ctf_tc =
 235                         ctf_metadata_decoder_borrow_ctf_trace_class(
 236                                 trace->metadata->decoder);
 237                 BT_ASSERT(!stream_iter->msg_iter);
 238                 stream_iter->msg_iter = ctf_msg_iter_create(ctf_tc,
 239                         lttng_live->max_query_size, medops, stream_iter,
 240                         log_level, self_comp, self_msg_iter);
 241                 if (!stream_iter->msg_iter) {
 242                         BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 243                                 "Failed to create CTF message iterator");
 244                         goto error;
 245                 }
 246         }
 247         stream_iter->buf = g_new0(uint8_t, lttng_live->max_query_size);
 248         if (!stream_iter->buf) {
 249                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 250                         "Failed to allocate live stream iterator buffer");
 251                 goto error;
 252         }
 253
 254         stream_iter->buflen = lttng_live->max_query_size;
 255         stream_iter->name = g_string_new(NULL);
 256         if (!stream_iter->name) {
 257                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 258                         "Failed to allocate live stream iterator name buffer");
 259                 goto error;
 260         }
 261
 262         g_string_printf(stream_iter->name, STREAM_NAME_PREFIX "%" PRIu64,
 263                 stream_iter->viewer_stream_id);
 264         g_ptr_array_add(trace->stream_iterators, stream_iter);
 265
 266         /* Track the number of active stream iterator. */
 267         session->lttng_live_msg_iter->active_stream_iter++;
 268
 269         goto end;
 270 error:
 271         lttng_live_stream_iterator_destroy(stream_iter);
 272         stream_iter = NULL;
 273 end:
 274         return stream_iter;
 275 }
 276
 277 BT_HIDDEN
 278 void lttng_live_stream_iterator_destroy(
 279                 struct lttng_live_stream_iterator *stream_iter)
 280 {
 281         if (!stream_iter) {
 282                 return;
 283         }
 284
 285         if (stream_iter->stream) {
 286                 BT_STREAM_PUT_REF_AND_RESET(stream_iter->stream);
 287         }
 288
 289         if (stream_iter->msg_iter) {
 290                 ctf_msg_iter_destroy(stream_iter->msg_iter);
 291         }
 292         g_free(stream_iter->buf);
 293         if (stream_iter->name) {
 294                 g_string_free(stream_iter->name, TRUE);
 295         }
 296
 297         bt_message_put_ref(stream_iter->current_msg);
 298
 299         /* Track the number of active stream iterator. */
 300         stream_iter->trace->session->lttng_live_msg_iter->active_stream_iter--;
 301
 302         g_free(stream_iter);
 303 }