Move to kernel style SPDX license identifiers
[babeltrace.git] / src / plugins / ctf / lttng-live / data-stream.c
CommitLineData
7cdc2bab 1/*
0235b0db 2 * SPDX-License-Identifier: MIT
7cdc2bab 3 *
0235b0db
MJ
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
7cdc2bab
MD
8 */
9
2ece7dd0 10#define BT_COMP_LOG_SELF_COMP self_comp
c01594de 11#define BT_LOG_OUTPUT_LEVEL log_level
350ad6c1 12#define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE/DS"
d9c39b0a 13#include "logging/comp-logging.h"
020bc26f 14
3c22a242 15#include <inttypes.h>
7cdc2bab 16#include <stdio.h>
7cdc2bab 17#include <stdlib.h>
3c22a242 18
7cdc2bab 19#include <glib.h>
3c22a242 20
3fadfbc0 21#include <babeltrace2/babeltrace.h>
3c22a242 22
d6e69534 23#include "../common/msg-iter/msg-iter.h"
578e048b 24#include "common/assert.h"
3c22a242 25#include "compat/mman.h"
087bc060 26#include "data-stream.h"
7cdc2bab 27
14f28187
FD
28#define STREAM_NAME_PREFIX "stream-"
29
7cdc2bab 30static
18a1979b 31enum ctf_msg_iter_medium_status medop_request_bytes(
7cdc2bab
MD
32 size_t request_sz, uint8_t **buffer_addr,
33 size_t *buffer_sz, void *data)
34{
18a1979b
SM
35 enum ctf_msg_iter_medium_status status =
36 CTF_MSG_ITER_MEDIUM_STATUS_OK;
7cdc2bab
MD
37 struct lttng_live_stream_iterator *stream = data;
38 struct lttng_live_trace *trace = stream->trace;
39 struct lttng_live_session *session = trace->session;
14f28187 40 struct lttng_live_msg_iter *live_msg_iter = session->lttng_live_msg_iter;
7cdc2bab
MD
41 uint64_t recv_len = 0;
42 uint64_t len_left;
43 uint64_t read_len;
7cdc2bab 44
c9694de4
SM
45 BT_ASSERT(request_sz);
46
4a39caef 47 if (stream->has_stream_hung_up) {
18a1979b 48 status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
4a39caef
FD
49 goto end;
50 }
51
7cdc2bab
MD
52 len_left = stream->base_offset + stream->len - stream->offset;
53 if (!len_left) {
54 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
18a1979b 55 status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
4a39caef 56 goto end;
7cdc2bab 57 }
4a39caef 58
7cdc2bab
MD
59 read_len = MIN(request_sz, stream->buflen);
60 read_len = MIN(read_len, len_left);
14f28187 61 status = lttng_live_get_stream_bytes(live_msg_iter,
7cdc2bab
MD
62 stream, stream->buf, stream->offset,
63 read_len, &recv_len);
7cdc2bab
MD
64 *buffer_addr = stream->buf;
65 *buffer_sz = recv_len;
66 stream->offset += recv_len;
4a39caef 67end:
7cdc2bab
MD
68 return status;
69}
70
71static
14f28187
FD
72bt_stream *medop_borrow_stream(bt_stream_class *stream_class,
73 int64_t stream_id, void *data)
7cdc2bab
MD
74{
75 struct lttng_live_stream_iterator *lttng_live_stream = data;
c01594de 76 bt_logging_level log_level = lttng_live_stream->log_level;
2ece7dd0 77 bt_self_component *self_comp = lttng_live_stream->self_comp;
7cdc2bab
MD
78
79 if (!lttng_live_stream->stream) {
14f28187 80 uint64_t stream_class_id = bt_stream_class_get_id(stream_class);
b92735af 81
2ece7dd0 82 BT_COMP_LOGI("Creating stream %s (ID: %" PRIu64 ") out of stream "
14f28187
FD
83 "class %" PRId64, lttng_live_stream->name->str,
84 stream_id, stream_class_id);
b92735af 85
14f28187
FD
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);
b92735af 96 } else {
14f28187
FD
97 lttng_live_stream->stream = bt_stream_create_with_id(
98 stream_class, lttng_live_stream->trace->trace,
99 (uint64_t) stream_id);
b92735af 100 }
7cdc2bab 101
7cdc2bab 102 if (!lttng_live_stream->stream) {
1419db2b
FD
103 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
104 "Cannot create stream %s (stream class ID "
14f28187
FD
105 "%" PRId64 ", stream ID %" PRIu64 ")",
106 lttng_live_stream->name->str,
107 stream_class_id, stream_id);
b80991f6 108 goto end;
7cdc2bab 109 }
b80991f6 110
14f28187
FD
111 bt_stream_set_name(lttng_live_stream->stream,
112 lttng_live_stream->name->str);
7cdc2bab
MD
113 }
114
b80991f6 115end:
7cdc2bab
MD
116 return lttng_live_stream->stream;
117}
118
18a1979b 119static struct ctf_msg_iter_medium_ops medops = {
7cdc2bab 120 .request_bytes = medop_request_bytes,
14f28187
FD
121 .seek = NULL,
122 .borrow_stream = medop_borrow_stream,
7cdc2bab
MD
123};
124
125BT_HIDDEN
14f28187 126enum lttng_live_iterator_status lttng_live_lazy_msg_init(
851de941
SM
127 struct lttng_live_session *session,
128 bt_self_message_iterator *self_msg_iter)
7cdc2bab 129{
14f28187
FD
130 struct lttng_live_component *lttng_live =
131 session->lttng_live_msg_iter->lttng_live_comp;
132 uint64_t trace_idx, stream_iter_idx;
c01594de 133 bt_logging_level log_level = session->log_level;
2ece7dd0 134 bt_self_component *self_comp = session->self_comp;
7cdc2bab 135
d6e69534 136 if (!session->lazy_stream_msg_init) {
14f28187 137 return LTTNG_LIVE_ITERATOR_STATUS_OK;
7cdc2bab
MD
138 }
139
ecb4ba8a
FD
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
14f28187
FD
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,
0f1979c3 154 stream_iter_idx);
7cdc2bab 155
14f28187 156 if (stream_iter->msg_iter) {
7cdc2bab
MD
157 continue;
158 }
14f28187 159 ctf_tc = ctf_metadata_decoder_borrow_ctf_trace_class(
0f1979c3 160 trace->metadata->decoder);
ecb4ba8a
FD
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);
18a1979b 165 stream_iter->msg_iter = ctf_msg_iter_create(ctf_tc,
0f1979c3 166 lttng_live->max_query_size, medops, stream_iter,
851de941 167 log_level, self_comp, self_msg_iter);
14f28187 168 if (!stream_iter->msg_iter) {
1419db2b
FD
169 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
170 "Failed to create CTF message iterator");
7cdc2bab
MD
171 goto error;
172 }
173 }
174 }
175
d6e69534 176 session->lazy_stream_msg_init = false;
7cdc2bab 177
14f28187 178 return LTTNG_LIVE_ITERATOR_STATUS_OK;
7cdc2bab
MD
179
180error:
14f28187 181 return LTTNG_LIVE_ITERATOR_STATUS_ERROR;
7cdc2bab
MD
182}
183
184BT_HIDDEN
185struct lttng_live_stream_iterator *lttng_live_stream_iterator_create(
186 struct lttng_live_session *session,
187 uint64_t ctf_trace_id,
851de941
SM
188 uint64_t stream_id,
189 bt_self_message_iterator *self_msg_iter)
7cdc2bab 190{
14f28187
FD
191 struct lttng_live_stream_iterator *stream_iter;
192 struct lttng_live_component *lttng_live;
7cdc2bab 193 struct lttng_live_trace *trace;
c01594de 194 bt_logging_level log_level;
2ece7dd0 195 bt_self_component *self_comp;
7cdc2bab 196
14f28187
FD
197 BT_ASSERT(session);
198 BT_ASSERT(session->lttng_live_msg_iter);
199 BT_ASSERT(session->lttng_live_msg_iter->lttng_live_comp);
c01594de 200 log_level = session->log_level;
2ece7dd0 201 self_comp = session->self_comp;
14f28187
FD
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) {
1419db2b
FD
207 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
208 "Failed to allocate struct lttng_live_stream_iterator");
14f28187
FD
209 goto error;
210 }
211
c01594de 212 stream_iter->log_level = log_level;
2ece7dd0 213 stream_iter->self_comp = self_comp;
36e94ad6 214 trace = lttng_live_session_borrow_or_create_trace_by_id(session, ctf_trace_id);
7cdc2bab 215 if (!trace) {
1419db2b
FD
216 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
217 "Failed to borrow CTF trace.");
7cdc2bab
MD
218 goto error;
219 }
220
14f28187
FD
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;
7cdc2bab
MD
226
227 if (trace->trace) {
14f28187
FD
228 struct ctf_trace_class *ctf_tc =
229 ctf_metadata_decoder_borrow_ctf_trace_class(
0f1979c3 230 trace->metadata->decoder);
14f28187 231 BT_ASSERT(!stream_iter->msg_iter);
18a1979b 232 stream_iter->msg_iter = ctf_msg_iter_create(ctf_tc,
0f1979c3 233 lttng_live->max_query_size, medops, stream_iter,
851de941 234 log_level, self_comp, self_msg_iter);
14f28187 235 if (!stream_iter->msg_iter) {
1419db2b
FD
236 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
237 "Failed to create CTF message iterator");
7cdc2bab
MD
238 goto error;
239 }
240 }
14f28187
FD
241 stream_iter->buf = g_new0(uint8_t, lttng_live->max_query_size);
242 if (!stream_iter->buf) {
1419db2b
FD
243 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
244 "Failed to allocate live stream iterator buffer");
14f28187
FD
245 goto error;
246 }
7cdc2bab 247
14f28187
FD
248 stream_iter->buflen = lttng_live->max_query_size;
249 stream_iter->name = g_string_new(NULL);
250 if (!stream_iter->name) {
1419db2b
FD
251 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
252 "Failed to allocate live stream iterator name buffer");
14f28187
FD
253 goto error;
254 }
255
256 g_string_printf(stream_iter->name, STREAM_NAME_PREFIX "%" PRIu64,
0f1979c3 257 stream_iter->viewer_stream_id);
14f28187 258 g_ptr_array_add(trace->stream_iterators, stream_iter);
7cdc2bab 259
14f28187
FD
260 /* Track the number of active stream iterator. */
261 session->lttng_live_msg_iter->active_stream_iter++;
7cdc2bab
MD
262
263 goto end;
264error:
14f28187
FD
265 lttng_live_stream_iterator_destroy(stream_iter);
266 stream_iter = NULL;
7cdc2bab 267end:
14f28187 268 return stream_iter;
7cdc2bab
MD
269}
270
271BT_HIDDEN
14f28187
FD
272void lttng_live_stream_iterator_destroy(
273 struct lttng_live_stream_iterator *stream_iter)
7cdc2bab 274{
14f28187 275 if (!stream_iter) {
7cdc2bab
MD
276 return;
277 }
278
14f28187
FD
279 if (stream_iter->stream) {
280 BT_STREAM_PUT_REF_AND_RESET(stream_iter->stream);
7cdc2bab
MD
281 }
282
14f28187 283 if (stream_iter->msg_iter) {
18a1979b 284 ctf_msg_iter_destroy(stream_iter->msg_iter);
7cdc2bab 285 }
19bbdc9b 286 g_free(stream_iter->buf);
14f28187
FD
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
14f28187 296 g_free(stream_iter);
7cdc2bab 297}
This page took 0.077468 seconds and 4 git commands to generate.