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