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                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
  55                 status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
  56                 goto end;
  57         }
  58
  59         read_len = MIN(request_sz, stream->buflen);
  60         read_len = MIN(read_len, len_left);
  61         status = lttng_live_get_stream_bytes(live_msg_iter,
  62                         stream, stream->buf, stream->offset,
  63                         read_len, &recv_len);
  64         *buffer_addr = stream->buf;
  65         *buffer_sz = recv_len;
  66         stream->offset += recv_len;
  67 end:
  68         return status;
  69 }
  70
  71 static
  72 bt_stream *medop_borrow_stream(bt_stream_class *stream_class,
  73                 int64_t stream_id, void *data)
  74 {
  75         struct lttng_live_stream_iterator *lttng_live_stream = 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, lttng_live_stream->name->str,
  84                         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,
  95                                 lttng_live_stream->viewer_stream_id);
  96                 } else {
  97                         lttng_live_stream->stream = bt_stream_create_with_id(
  98                                 stream_class, lttng_live_stream->trace->trace,
  99                                 (uint64_t) stream_id);
 100                 }
 101
 102                 if (!lttng_live_stream->stream) {
 103                         BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 104                                 "Cannot create stream %s (stream class ID "
 105                                 "%" PRId64 ", stream ID %" PRIu64 ")",
 106                                 lttng_live_stream->name->str,
 107                                 stream_class_id, stream_id);
 108                         goto end;
 109                 }
 110
 111                 bt_stream_set_name(lttng_live_stream->stream,
 112                         lttng_live_stream->name->str);
 113         }
 114
 115 end:
 116         return lttng_live_stream->stream;
 117 }
 118
 119 static struct ctf_msg_iter_medium_ops medops = {
 120         .request_bytes = medop_request_bytes,
 121         .seek = NULL,
 122         .borrow_stream = medop_borrow_stream,
 123 };
 124
 125 BT_HIDDEN
 126 enum lttng_live_iterator_status lttng_live_lazy_msg_init(
 127                 struct lttng_live_session *session,
 128                 bt_self_message_iterator *self_msg_iter)
 129 {
 130         struct lttng_live_component *lttng_live =
 131                 session->lttng_live_msg_iter->lttng_live_comp;
 132         uint64_t trace_idx, stream_iter_idx;
 133         bt_logging_level log_level = session->log_level;
 134         bt_self_component *self_comp = session->self_comp;
 135
 136         if (!session->lazy_stream_msg_init) {
 137                 return LTTNG_LIVE_ITERATOR_STATUS_OK;
 138         }
 139
 140         BT_COMP_LOGD("Lazily initializing self message iterator for live session: "
 141                 "session-id=%"PRIu64", self-msg-iter-addr=%p", session->id,
 142                 self_msg_iter);
 143
 144         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 145                 struct lttng_live_trace *trace =
 146                         g_ptr_array_index(session->traces, trace_idx);
 147
 148                 for (stream_iter_idx = 0;
 149                                 stream_iter_idx < trace->stream_iterators->len;
 150                                 stream_iter_idx++) {
 151                         struct ctf_trace_class *ctf_tc;
 152                         struct lttng_live_stream_iterator *stream_iter =
 153                                 g_ptr_array_index(trace->stream_iterators,
 154                                         stream_iter_idx);
 155
 156                         if (stream_iter->msg_iter) {
 157                                 continue;
 158                         }
 159                         ctf_tc = ctf_metadata_decoder_borrow_ctf_trace_class(
 160                                 trace->metadata->decoder);
 161                         BT_COMP_LOGD("Creating CTF message iterator: "
 162                                 "session-id=%"PRIu64", ctf-tc-addr=%p, "
 163                                 "stream-iter-name=%s, self-msg-iter-addr=%p",
 164                                 session->id, ctf_tc, stream_iter->name->str, self_msg_iter);
 165                         stream_iter->msg_iter = ctf_msg_iter_create(ctf_tc,
 166                                 lttng_live->max_query_size, medops, stream_iter,
 167                                 log_level, self_comp, self_msg_iter);
 168                         if (!stream_iter->msg_iter) {
 169                                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 170                                         "Failed to create CTF message iterator");
 171                                 goto error;
 172                         }
 173                 }
 174         }
 175
 176         session->lazy_stream_msg_init = false;
 177
 178         return LTTNG_LIVE_ITERATOR_STATUS_OK;
 179
 180 error:
 181         return LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 182 }
 183
 184 BT_HIDDEN
 185 struct lttng_live_stream_iterator *lttng_live_stream_iterator_create(
 186                 struct lttng_live_session *session,
 187                 uint64_t ctf_trace_id,
 188                 uint64_t stream_id,
 189                 bt_self_message_iterator *self_msg_iter)
 190 {
 191         struct lttng_live_stream_iterator *stream_iter;
 192         struct lttng_live_component *lttng_live;
 193         struct lttng_live_trace *trace;
 194         bt_logging_level log_level;
 195         bt_self_component *self_comp;
 196
 197         BT_ASSERT(session);
 198         BT_ASSERT(session->lttng_live_msg_iter);
 199         BT_ASSERT(session->lttng_live_msg_iter->lttng_live_comp);
 200         log_level = session->log_level;
 201         self_comp = session->self_comp;
 202
 203         lttng_live = session->lttng_live_msg_iter->lttng_live_comp;
 204
 205         stream_iter = g_new0(struct lttng_live_stream_iterator, 1);
 206         if (!stream_iter) {
 207                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 208                         "Failed to allocate struct lttng_live_stream_iterator");
 209                 goto error;
 210         }
 211
 212         stream_iter->log_level = log_level;
 213         stream_iter->self_comp = self_comp;
 214         trace = lttng_live_session_borrow_or_create_trace_by_id(session, ctf_trace_id);
 215         if (!trace) {
 216                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 217                         "Failed to borrow CTF trace.");
 218                 goto error;
 219         }
 220
 221         stream_iter->trace = trace;
 222         stream_iter->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
 223         stream_iter->viewer_stream_id = stream_id;
 224         stream_iter->ctf_stream_class_id = -1ULL;
 225         stream_iter->last_inactivity_ts = INT64_MIN;
 226
 227         if (trace->trace) {
 228                 struct ctf_trace_class *ctf_tc =
 229                         ctf_metadata_decoder_borrow_ctf_trace_class(
 230                                 trace->metadata->decoder);
 231                 BT_ASSERT(!stream_iter->msg_iter);
 232                 stream_iter->msg_iter = ctf_msg_iter_create(ctf_tc,
 233                         lttng_live->max_query_size, medops, stream_iter,
 234                         log_level, self_comp, self_msg_iter);
 235                 if (!stream_iter->msg_iter) {
 236                         BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 237                                 "Failed to create CTF message iterator");
 238                         goto error;
 239                 }
 240         }
 241         stream_iter->buf = g_new0(uint8_t, lttng_live->max_query_size);
 242         if (!stream_iter->buf) {
 243                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 244                         "Failed to allocate live stream iterator buffer");
 245                 goto error;
 246         }
 247
 248         stream_iter->buflen = lttng_live->max_query_size;
 249         stream_iter->name = g_string_new(NULL);
 250         if (!stream_iter->name) {
 251                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 252                         "Failed to allocate live stream iterator name buffer");
 253                 goto error;
 254         }
 255
 256         g_string_printf(stream_iter->name, STREAM_NAME_PREFIX "%" PRIu64,
 257                 stream_iter->viewer_stream_id);
 258         g_ptr_array_add(trace->stream_iterators, stream_iter);
 259
 260         /* Track the number of active stream iterator. */
 261         session->lttng_live_msg_iter->active_stream_iter++;
 262
 263         goto end;
 264 error:
 265         lttng_live_stream_iterator_destroy(stream_iter);
 266         stream_iter = NULL;
 267 end:
 268         return stream_iter;
 269 }
 270
 271 BT_HIDDEN
 272 void lttng_live_stream_iterator_destroy(
 273                 struct lttng_live_stream_iterator *stream_iter)
 274 {
 275         if (!stream_iter) {
 276                 return;
 277         }
 278
 279         if (stream_iter->stream) {
 280                 BT_STREAM_PUT_REF_AND_RESET(stream_iter->stream);
 281         }
 282
 283         if (stream_iter->msg_iter) {
 284                 ctf_msg_iter_destroy(stream_iter->msg_iter);
 285         }
 286         g_free(stream_iter->buf);
 287         if (stream_iter->name) {
 288                 g_string_free(stream_iter->name, TRUE);
 289         }
 290
 291         bt_message_put_ref(stream_iter->current_msg);
 292
 293         /* Track the number of active stream iterator. */
 294         stream_iter->trace->session->lttng_live_msg_iter->active_stream_iter--;
 295
 296         g_free(stream_iter);
 297 }