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 enum ctf_msg_iter_medium_status medop_request_bytes(size_t request_sz, uint8_t **buffer_addr,
  31                                                            size_t *buffer_sz, void *data)
  32 {
  33     enum ctf_msg_iter_medium_status status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
  34     lttng_live_stream_iterator *stream = (lttng_live_stream_iterator *) data;
  35     struct lttng_live_trace *trace = stream->trace;
  36     struct lttng_live_session *session = trace->session;
  37     struct lttng_live_msg_iter *live_msg_iter = session->lttng_live_msg_iter;
  38     uint64_t recv_len = 0;
  39     uint64_t len_left;
  40     uint64_t read_len;
  41
  42     BT_ASSERT(request_sz);
  43
  44     if (stream->has_stream_hung_up) {
  45         status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
  46         goto end;
  47     }
  48
  49     len_left = stream->base_offset + stream->len - stream->offset;
  50     if (!len_left) {
  51         lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
  52         status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
  53         goto end;
  54     }
  55
  56     read_len = MIN(request_sz, stream->buflen);
  57     read_len = MIN(read_len, len_left);
  58     status = lttng_live_get_stream_bytes(live_msg_iter, stream, stream->buf, stream->offset,
  59                                          read_len, &recv_len);
  60     *buffer_addr = stream->buf;
  61     *buffer_sz = recv_len;
  62     stream->offset += recv_len;
  63 end:
  64     return status;
  65 }
  66
  67 static bt_stream *medop_borrow_stream(bt_stream_class *stream_class, int64_t stream_id, void *data)
  68 {
  69     lttng_live_stream_iterator *lttng_live_stream = (lttng_live_stream_iterator *) data;
  70     bt_logging_level log_level = lttng_live_stream->log_level;
  71     bt_self_component *self_comp = lttng_live_stream->self_comp;
  72
  73     if (!lttng_live_stream->stream) {
  74         uint64_t stream_class_id = bt_stream_class_get_id(stream_class);
  75
  76         BT_COMP_LOGI("Creating stream %s (ID: %" PRIu64 ") out of stream "
  77                      "class %" PRId64,
  78                      lttng_live_stream->name->str, stream_id, stream_class_id);
  79
  80         if (stream_id < 0) {
  81             /*
  82              * No stream instance ID in the stream. It's possible
  83              * to encounter this situation with older version of
  84              * LTTng. In these cases, use the viewer_stream_id that
  85              * is unique for a live viewer session.
  86              */
  87             lttng_live_stream->stream = bt_stream_create_with_id(
  88                 stream_class, lttng_live_stream->trace->trace, lttng_live_stream->viewer_stream_id);
  89         } else {
  90             lttng_live_stream->stream = bt_stream_create_with_id(
  91                 stream_class, lttng_live_stream->trace->trace, (uint64_t) stream_id);
  92         }
  93
  94         if (!lttng_live_stream->stream) {
  95             BT_COMP_LOGE_APPEND_CAUSE(self_comp,
  96                                       "Cannot create stream %s (stream class ID "
  97                                       "%" PRId64 ", stream ID %" PRIu64 ")",
  98                                       lttng_live_stream->name->str, stream_class_id, stream_id);
  99             goto end;
 100         }
 101
 102         bt_stream_set_name(lttng_live_stream->stream, lttng_live_stream->name->str);
 103     }
 104
 105 end:
 106     return lttng_live_stream->stream;
 107 }
 108
 109 static struct ctf_msg_iter_medium_ops medops = {
 110     medop_request_bytes,
 111     nullptr,
 112     nullptr,
 113     medop_borrow_stream,
 114 };
 115
 116 BT_HIDDEN
 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 BT_HIDDEN
 171 struct lttng_live_stream_iterator *
 172 lttng_live_stream_iterator_create(struct lttng_live_session *session, uint64_t ctf_trace_id,
 173                                   uint64_t stream_id, bt_self_message_iterator *self_msg_iter)
 174 {
 175     struct lttng_live_stream_iterator *stream_iter;
 176     struct lttng_live_component *lttng_live;
 177     struct lttng_live_trace *trace;
 178     bt_logging_level log_level;
 179     bt_self_component *self_comp;
 180
 181     BT_ASSERT(session);
 182     BT_ASSERT(session->lttng_live_msg_iter);
 183     BT_ASSERT(session->lttng_live_msg_iter->lttng_live_comp);
 184     log_level = session->log_level;
 185     self_comp = session->self_comp;
 186
 187     lttng_live = session->lttng_live_msg_iter->lttng_live_comp;
 188
 189     stream_iter = g_new0(struct lttng_live_stream_iterator, 1);
 190     if (!stream_iter) {
 191         BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 192                                   "Failed to allocate struct lttng_live_stream_iterator");
 193         goto error;
 194     }
 195
 196     stream_iter->log_level = log_level;
 197     stream_iter->self_comp = self_comp;
 198     trace = lttng_live_session_borrow_or_create_trace_by_id(session, ctf_trace_id);
 199     if (!trace) {
 200         BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to borrow CTF trace.");
 201         goto error;
 202     }
 203
 204     stream_iter->trace = trace;
 205     stream_iter->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
 206     stream_iter->viewer_stream_id = stream_id;
 207
 208     stream_iter->ctf_stream_class_id.is_set = false;
 209     stream_iter->ctf_stream_class_id.value = UINT64_MAX;
 210
 211     stream_iter->last_inactivity_ts.is_set = false;
 212     stream_iter->last_inactivity_ts.value = 0;
 213
 214     if (trace->trace) {
 215         struct ctf_trace_class *ctf_tc =
 216             ctf_metadata_decoder_borrow_ctf_trace_class(trace->metadata->decoder);
 217         BT_ASSERT(!stream_iter->msg_iter);
 218         stream_iter->msg_iter =
 219             ctf_msg_iter_create(ctf_tc, lttng_live->max_query_size, medops, stream_iter, log_level,
 220                                 self_comp, self_msg_iter);
 221         if (!stream_iter->msg_iter) {
 222             BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to create CTF message iterator");
 223             goto error;
 224         }
 225     }
 226     stream_iter->buf = g_new0(uint8_t, lttng_live->max_query_size);
 227     if (!stream_iter->buf) {
 228         BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to allocate live stream iterator buffer");
 229         goto error;
 230     }
 231
 232     stream_iter->buflen = lttng_live->max_query_size;
 233     stream_iter->name = g_string_new(NULL);
 234     if (!stream_iter->name) {
 235         BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to allocate live stream iterator name buffer");
 236         goto error;
 237     }
 238
 239     g_string_printf(stream_iter->name, STREAM_NAME_PREFIX "%" PRIu64,
 240                     stream_iter->viewer_stream_id);
 241     g_ptr_array_add(trace->stream_iterators, stream_iter);
 242
 243     /* Track the number of active stream iterator. */
 244     session->lttng_live_msg_iter->active_stream_iter++;
 245
 246     goto end;
 247 error:
 248     lttng_live_stream_iterator_destroy(stream_iter);
 249     stream_iter = NULL;
 250 end:
 251     return stream_iter;
 252 }
 253
 254 BT_HIDDEN
 255 void lttng_live_stream_iterator_destroy(struct lttng_live_stream_iterator *stream_iter)
 256 {
 257     if (!stream_iter) {
 258         return;
 259     }
 260
 261     if (stream_iter->stream) {
 262         BT_STREAM_PUT_REF_AND_RESET(stream_iter->stream);
 263     }
 264
 265     if (stream_iter->msg_iter) {
 266         ctf_msg_iter_destroy(stream_iter->msg_iter);
 267     }
 268     g_free(stream_iter->buf);
 269     if (stream_iter->name) {
 270         g_string_free(stream_iter->name, TRUE);
 271     }
 272
 273     bt_message_put_ref(stream_iter->current_msg);
 274
 275     /* Track the number of active stream iterator. */
 276     stream_iter->trace->session->lttng_live_msg_iter->active_stream_iter--;
 277
 278     g_free(stream_iter);
 279 }