Move to kernel style SPDX license identifiers
[babeltrace.git] / 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 }
This page took 0.037352 seconds and 4 git commands to generate.