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