Visibility hidden by default
[babeltrace.git] / src / plugins / ctf / fs-sink / fs-sink-stream.cpp
1 /*
2 * SPDX-License-Identifier: MIT
3 *
4 * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
5 */
6
7 #define BT_COMP_LOG_SELF_COMP (stream->trace->fs_sink->self_comp)
8 #define BT_LOG_OUTPUT_LEVEL (stream->log_level)
9 #define BT_LOG_TAG "PLUGIN/SINK.CTF.FS/STREAM"
10 #include "logging/comp-logging.h"
11
12 #include <babeltrace2/babeltrace.h>
13 #include <stdio.h>
14 #include <stdbool.h>
15 #include <glib.h>
16 #include "common/assert.h"
17 #include "ctfser/ctfser.h"
18 #include "compat/endian.h"
19
20 #include "fs-sink.hpp"
21 #include "fs-sink-trace.hpp"
22 #include "fs-sink-stream.hpp"
23 #include "translate-trace-ir-to-ctf-ir.hpp"
24
25 void fs_sink_stream_destroy(struct fs_sink_stream *stream)
26 {
27 if (!stream) {
28 goto end;
29 }
30
31 bt_ctfser_fini(&stream->ctfser);
32
33 if (stream->file_name) {
34 g_string_free(stream->file_name, TRUE);
35 stream->file_name = NULL;
36 }
37
38 bt_packet_put_ref(stream->packet_state.packet);
39 g_free(stream);
40
41 end:
42 return;
43 }
44
45 static bool stream_file_name_exists(struct fs_sink_trace *trace, const char *name)
46 {
47 bool exists = false;
48 GHashTableIter iter;
49 gpointer key, value;
50
51 g_hash_table_iter_init(&iter, trace->streams);
52
53 while (g_hash_table_iter_next(&iter, &key, &value)) {
54 struct fs_sink_stream *stream = (fs_sink_stream *) value;
55
56 if (strcmp(name, stream->file_name->str) == 0) {
57 exists = true;
58 goto end;
59 }
60 }
61
62 end:
63 return exists;
64 }
65
66 static GString *sanitize_stream_file_name(const char *file_name)
67 {
68 GString *san_file_name = g_string_new(NULL);
69 const char *ch;
70 gchar *basename;
71
72 BT_ASSERT(san_file_name);
73 BT_ASSERT(file_name);
74 basename = g_path_get_basename(file_name);
75
76 for (ch = basename; *ch != '\0'; ch++) {
77 if (*ch == '/') {
78 g_string_append_c(san_file_name, '_');
79 } else {
80 g_string_append_c(san_file_name, *ch);
81 }
82 }
83
84 /* Do not allow `.` and `..` either */
85 if (strcmp(san_file_name->str, ".") == 0 || strcmp(san_file_name->str, "..") == 0) {
86 g_string_assign(san_file_name, "stream");
87 }
88
89 g_free(basename);
90 return san_file_name;
91 }
92
93 static GString *make_unique_stream_file_name(struct fs_sink_trace *trace, const char *base)
94 {
95 GString *san_base = sanitize_stream_file_name(base);
96 GString *name = g_string_new(san_base->str);
97 unsigned int suffix = 0;
98
99 BT_ASSERT(name);
100
101 while (stream_file_name_exists(trace, name->str) || strcmp(name->str, "metadata") == 0) {
102 g_string_printf(name, "%s-%u", san_base->str, suffix);
103 suffix++;
104 }
105
106 g_string_free(san_base, TRUE);
107 return name;
108 }
109
110 static void set_stream_file_name(struct fs_sink_stream *stream)
111 {
112 const char *base_name = bt_stream_get_name(stream->ir_stream);
113
114 if (!base_name) {
115 base_name = "stream";
116 }
117
118 BT_ASSERT(!stream->file_name);
119 stream->file_name = make_unique_stream_file_name(stream->trace, base_name);
120 }
121
122 struct fs_sink_stream *fs_sink_stream_create(struct fs_sink_trace *trace,
123 const bt_stream *ir_stream)
124 {
125 struct fs_sink_stream *stream = g_new0(struct fs_sink_stream, 1);
126 int ret;
127 GString *path = g_string_new(trace->path->str);
128
129 if (!stream) {
130 goto end;
131 }
132
133 stream->log_level = trace->log_level;
134 stream->trace = trace;
135 stream->ir_stream = ir_stream;
136 stream->packet_state.beginning_cs = UINT64_C(-1);
137 stream->packet_state.end_cs = UINT64_C(-1);
138 stream->prev_packet_state.end_cs = UINT64_C(-1);
139 stream->prev_packet_state.discarded_events_counter = UINT64_C(-1);
140 stream->prev_packet_state.seq_num = UINT64_C(-1);
141 ret = try_translate_stream_class_trace_ir_to_ctf_ir(
142 trace->fs_sink, trace->trace, bt_stream_borrow_class_const(ir_stream), &stream->sc);
143 if (ret) {
144 goto error;
145 }
146
147 set_stream_file_name(stream);
148 g_string_append_printf(path, "/%s", stream->file_name->str);
149 ret = bt_ctfser_init(&stream->ctfser, path->str, stream->log_level);
150 if (ret) {
151 goto error;
152 }
153
154 g_hash_table_insert(trace->streams, (gpointer) ir_stream, stream);
155 goto end;
156
157 error:
158 fs_sink_stream_destroy(stream);
159 stream = NULL;
160
161 end:
162 if (path) {
163 g_string_free(path, TRUE);
164 }
165
166 return stream;
167 }
168
169 static int write_field(struct fs_sink_stream *stream, struct fs_sink_ctf_field_class *fc,
170 const bt_field *field);
171
172 static inline int write_bool_field(struct fs_sink_stream *stream,
173 struct fs_sink_ctf_field_class_bool *fc, const bt_field *field)
174 {
175 /*
176 * CTF 1.8 has no boolean field class type, so this component
177 * translates this boolean field to an 8-bit unsigned integer
178 * field which has the value 0 (false) or 1 (true).
179 */
180 return bt_ctfser_write_unsigned_int(&stream->ctfser, bt_field_bool_get_value(field) ? 1 : 0,
181 fc->base.base.alignment, fc->base.size, BYTE_ORDER);
182 }
183
184 static inline int write_bit_array_field(struct fs_sink_stream *stream,
185 struct fs_sink_ctf_field_class_bit_array *fc,
186 const bt_field *field)
187 {
188 /*
189 * CTF 1.8 has no bit array field class type, so this component
190 * translates this bit array field to an unsigned integer field.
191 */
192 return bt_ctfser_write_unsigned_int(&stream->ctfser,
193 bt_field_bit_array_get_value_as_integer(field),
194 fc->base.alignment, fc->size, BYTE_ORDER);
195 }
196
197 static inline int write_int_field(struct fs_sink_stream *stream,
198 struct fs_sink_ctf_field_class_int *fc, const bt_field *field)
199 {
200 int ret;
201
202 if (fc->is_signed) {
203 ret = bt_ctfser_write_signed_int(&stream->ctfser, bt_field_integer_signed_get_value(field),
204 fc->base.base.alignment, fc->base.size, BYTE_ORDER);
205 } else {
206 ret = bt_ctfser_write_unsigned_int(&stream->ctfser,
207 bt_field_integer_unsigned_get_value(field),
208 fc->base.base.alignment, fc->base.size, BYTE_ORDER);
209 }
210
211 return ret;
212 }
213
214 static inline int write_float_field(struct fs_sink_stream *stream,
215 struct fs_sink_ctf_field_class_float *fc, const bt_field *field)
216 {
217 int ret;
218 double val;
219
220 if (fc->base.size == 32) {
221 val = (double) bt_field_real_single_precision_get_value(field);
222 ret = bt_ctfser_write_float32(&stream->ctfser, val, fc->base.base.alignment, BYTE_ORDER);
223 } else {
224 val = bt_field_real_double_precision_get_value(field);
225 ret = bt_ctfser_write_float64(&stream->ctfser, val, fc->base.base.alignment, BYTE_ORDER);
226 }
227
228 return ret;
229 }
230
231 static inline int write_string_field(struct fs_sink_stream *stream,
232 struct fs_sink_ctf_field_class_string *fc,
233 const bt_field *field)
234 {
235 return bt_ctfser_write_string(&stream->ctfser, bt_field_string_get_value(field));
236 }
237
238 static inline int write_array_base_field_elements(struct fs_sink_stream *stream,
239 struct fs_sink_ctf_field_class_array_base *fc,
240 const bt_field *field)
241 {
242 uint64_t i;
243 uint64_t len = bt_field_array_get_length(field);
244 int ret = 0;
245
246 for (i = 0; i < len; i++) {
247 const bt_field *elem_field = bt_field_array_borrow_element_field_by_index_const(field, i);
248 ret = write_field(stream, fc->elem_fc, elem_field);
249 if (G_UNLIKELY(ret)) {
250 goto end;
251 }
252 }
253
254 end:
255 return ret;
256 }
257
258 static inline int write_sequence_field(struct fs_sink_stream *stream,
259 struct fs_sink_ctf_field_class_sequence *fc,
260 const bt_field *field)
261 {
262 int ret;
263
264 if (fc->length_is_before) {
265 ret = bt_ctfser_write_unsigned_int(&stream->ctfser, bt_field_array_get_length(field), 8, 32,
266 BYTE_ORDER);
267 if (G_UNLIKELY(ret)) {
268 goto end;
269 }
270 }
271
272 ret = write_array_base_field_elements(stream, &fc->base, field);
273
274 end:
275 return ret;
276 }
277
278 static inline int write_struct_field(struct fs_sink_stream *stream,
279 struct fs_sink_ctf_field_class_struct *fc,
280 const bt_field *field, bool align_struct)
281 {
282 int ret = 0;
283 uint64_t i;
284
285 if (G_LIKELY(align_struct)) {
286 ret = bt_ctfser_align_offset_in_current_packet(&stream->ctfser, fc->base.alignment);
287 if (G_UNLIKELY(ret)) {
288 goto end;
289 }
290 }
291
292 for (i = 0; i < fc->members->len; i++) {
293 const bt_field *memb_field =
294 bt_field_structure_borrow_member_field_by_index_const(field, i);
295 struct fs_sink_ctf_field_class *member_fc =
296 fs_sink_ctf_field_class_struct_borrow_member_by_index(fc, i)->fc;
297
298 ret = write_field(stream, member_fc, memb_field);
299 if (G_UNLIKELY(ret)) {
300 goto end;
301 }
302 }
303
304 end:
305 return ret;
306 }
307
308 static inline int write_option_field(struct fs_sink_stream *stream,
309 struct fs_sink_ctf_field_class_option *fc,
310 const bt_field *field)
311 {
312 int ret;
313 const bt_field *content_field = bt_field_option_borrow_field_const(field);
314
315 ret = bt_ctfser_write_unsigned_int(&stream->ctfser, content_field ? 1 : 0, 8, 8, BYTE_ORDER);
316 if (G_UNLIKELY(ret)) {
317 goto end;
318 }
319
320 /*
321 * CTF 1.8 has no option field class type, so this component
322 * translates the option field class to a variant field class
323 * where the options are:
324 *
325 * * An empty structure field class (field occupies 0 bits).
326 * * The optional field class itself.
327 *
328 * If `content_field` is `NULL`, do not write anything (empty
329 * structure).
330 */
331 if (content_field) {
332 ret = write_field(stream, fc->content_fc, content_field);
333 }
334
335 end:
336 return ret;
337 }
338
339 static inline int write_variant_field(struct fs_sink_stream *stream,
340 struct fs_sink_ctf_field_class_variant *fc,
341 const bt_field *field)
342 {
343 uint64_t opt_index = bt_field_variant_get_selected_option_index(field);
344 int ret;
345
346 if (fc->tag_is_before) {
347 ret = bt_ctfser_write_unsigned_int(&stream->ctfser, opt_index, 8, 16, BYTE_ORDER);
348 if (G_UNLIKELY(ret)) {
349 goto end;
350 }
351 }
352
353 ret = write_field(stream,
354 fs_sink_ctf_field_class_variant_borrow_option_by_index(fc, opt_index)->fc,
355 bt_field_variant_borrow_selected_option_field_const(field));
356
357 end:
358 return ret;
359 }
360
361 static int write_field(struct fs_sink_stream *stream, struct fs_sink_ctf_field_class *fc,
362 const bt_field *field)
363 {
364 int ret;
365
366 switch (fc->type) {
367 case FS_SINK_CTF_FIELD_CLASS_TYPE_BOOL:
368 ret = write_bool_field(stream, fs_sink_ctf_field_class_as_bool(fc), field);
369 break;
370 case FS_SINK_CTF_FIELD_CLASS_TYPE_BIT_ARRAY:
371 ret = write_bit_array_field(stream, fs_sink_ctf_field_class_as_bit_array(fc), field);
372 break;
373 case FS_SINK_CTF_FIELD_CLASS_TYPE_INT:
374 ret = write_int_field(stream, fs_sink_ctf_field_class_as_int(fc), field);
375 break;
376 case FS_SINK_CTF_FIELD_CLASS_TYPE_FLOAT:
377 ret = write_float_field(stream, fs_sink_ctf_field_class_as_float(fc), field);
378 break;
379 case FS_SINK_CTF_FIELD_CLASS_TYPE_STRING:
380 ret = write_string_field(stream, fs_sink_ctf_field_class_as_string(fc), field);
381 break;
382 case FS_SINK_CTF_FIELD_CLASS_TYPE_STRUCT:
383 ret = write_struct_field(stream, fs_sink_ctf_field_class_as_struct(fc), field, true);
384 break;
385 case FS_SINK_CTF_FIELD_CLASS_TYPE_ARRAY:
386 ret = write_array_base_field_elements(stream, fs_sink_ctf_field_class_as_array_base(fc),
387 field);
388 break;
389 case FS_SINK_CTF_FIELD_CLASS_TYPE_SEQUENCE:
390 ret = write_sequence_field(stream, fs_sink_ctf_field_class_as_sequence(fc), field);
391 break;
392 case FS_SINK_CTF_FIELD_CLASS_TYPE_OPTION:
393 ret = write_option_field(stream, fs_sink_ctf_field_class_as_option(fc), field);
394 break;
395 case FS_SINK_CTF_FIELD_CLASS_TYPE_VARIANT:
396 ret = write_variant_field(stream, fs_sink_ctf_field_class_as_variant(fc), field);
397 break;
398 default:
399 bt_common_abort();
400 }
401
402 return ret;
403 }
404
405 static inline int write_event_header(struct fs_sink_stream *stream, const bt_clock_snapshot *cs,
406 struct fs_sink_ctf_event_class *ec)
407 {
408 int ret;
409
410 /* Event class ID */
411 ret = bt_ctfser_write_byte_aligned_unsigned_int(
412 &stream->ctfser, bt_event_class_get_id(ec->ir_ec), 8, 64, BYTE_ORDER);
413 if (G_UNLIKELY(ret)) {
414 goto end;
415 }
416
417 /* Time */
418 if (stream->sc->default_clock_class) {
419 BT_ASSERT_DBG(cs);
420 ret = bt_ctfser_write_byte_aligned_unsigned_int(
421 &stream->ctfser, bt_clock_snapshot_get_value(cs), 8, 64, BYTE_ORDER);
422 if (G_UNLIKELY(ret)) {
423 goto end;
424 }
425 }
426
427 end:
428 return ret;
429 }
430
431 int fs_sink_stream_write_event(struct fs_sink_stream *stream, const bt_clock_snapshot *cs,
432 const bt_event *event, struct fs_sink_ctf_event_class *ec)
433 {
434 int ret;
435 const bt_field *field;
436
437 /* Header */
438 ret = write_event_header(stream, cs, ec);
439 if (G_UNLIKELY(ret)) {
440 goto end;
441 }
442
443 /* Common context */
444 if (stream->sc->event_common_context_fc) {
445 field = bt_event_borrow_common_context_field_const(event);
446 BT_ASSERT_DBG(field);
447 ret = write_struct_field(
448 stream, fs_sink_ctf_field_class_as_struct(stream->sc->event_common_context_fc), field,
449 true);
450 if (G_UNLIKELY(ret)) {
451 goto end;
452 }
453 }
454
455 /* Specific context */
456 if (ec->spec_context_fc) {
457 field = bt_event_borrow_specific_context_field_const(event);
458 BT_ASSERT_DBG(field);
459 ret = write_struct_field(stream, fs_sink_ctf_field_class_as_struct(ec->spec_context_fc),
460 field, true);
461 if (G_UNLIKELY(ret)) {
462 goto end;
463 }
464 }
465
466 /* Specific context */
467 if (ec->payload_fc) {
468 field = bt_event_borrow_payload_field_const(event);
469 BT_ASSERT_DBG(field);
470 ret = write_struct_field(stream, fs_sink_ctf_field_class_as_struct(ec->payload_fc), field,
471 true);
472 if (G_UNLIKELY(ret)) {
473 goto end;
474 }
475 }
476
477 end:
478 return ret;
479 }
480
481 static int write_packet_context(struct fs_sink_stream *stream)
482 {
483 int ret;
484
485 /* Packet total size */
486 ret = bt_ctfser_write_byte_aligned_unsigned_int(
487 &stream->ctfser, stream->packet_state.total_size, 8, 64, BYTE_ORDER);
488 if (ret) {
489 goto end;
490 }
491
492 /* Packet content size */
493 ret = bt_ctfser_write_byte_aligned_unsigned_int(
494 &stream->ctfser, stream->packet_state.content_size, 8, 64, BYTE_ORDER);
495 if (ret) {
496 goto end;
497 }
498
499 if (stream->sc->packets_have_ts_begin) {
500 /* Beginning time */
501 ret = bt_ctfser_write_byte_aligned_unsigned_int(
502 &stream->ctfser, stream->packet_state.beginning_cs, 8, 64, BYTE_ORDER);
503 if (ret) {
504 goto end;
505 }
506 }
507
508 if (stream->sc->packets_have_ts_end) {
509 /* End time */
510 ret = bt_ctfser_write_byte_aligned_unsigned_int(
511 &stream->ctfser, stream->packet_state.end_cs, 8, 64, BYTE_ORDER);
512 if (ret) {
513 goto end;
514 }
515 }
516
517 if (stream->sc->has_discarded_events) {
518 /* Discarded event counter */
519 ret = bt_ctfser_write_byte_aligned_unsigned_int(
520 &stream->ctfser, stream->packet_state.discarded_events_counter, 8, 64, BYTE_ORDER);
521 if (ret) {
522 goto end;
523 }
524 }
525
526 /* Sequence number */
527 ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser, stream->packet_state.seq_num,
528 8, 64, BYTE_ORDER);
529 if (ret) {
530 goto end;
531 }
532
533 /* Other members */
534 if (stream->sc->packet_context_fc) {
535 const bt_field *packet_context_field;
536
537 BT_ASSERT(stream->packet_state.packet);
538 packet_context_field = bt_packet_borrow_context_field_const(stream->packet_state.packet);
539 BT_ASSERT(packet_context_field);
540 ret = write_struct_field(stream,
541 fs_sink_ctf_field_class_as_struct(stream->sc->packet_context_fc),
542 packet_context_field, false);
543 if (ret) {
544 goto end;
545 }
546 }
547
548 end:
549 return ret;
550 }
551
552 int fs_sink_stream_open_packet(struct fs_sink_stream *stream, const bt_clock_snapshot *cs,
553 const bt_packet *packet)
554 {
555 int ret;
556 uint64_t i;
557
558 BT_ASSERT(!stream->packet_state.is_open);
559 bt_packet_put_ref(stream->packet_state.packet);
560 stream->packet_state.packet = packet;
561 bt_packet_get_ref(stream->packet_state.packet);
562 if (cs) {
563 stream->packet_state.beginning_cs = bt_clock_snapshot_get_value(cs);
564 }
565
566 /* Open packet */
567 ret = bt_ctfser_open_packet(&stream->ctfser);
568 if (ret) {
569 /* bt_ctfser_open_packet() logs errors */
570 goto end;
571 }
572
573 /* Packet header: magic */
574 ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser, UINT64_C(0xc1fc1fc1), 8, 32,
575 BYTE_ORDER);
576 if (ret) {
577 BT_COMP_LOGE("Error writing packet header magic: stream-file-name=%s",
578 stream->file_name->str);
579 goto end;
580 }
581
582 /* Packet header: UUID */
583 for (i = 0; i < BT_UUID_LEN; i++) {
584 ret = bt_ctfser_write_byte_aligned_unsigned_int(
585 &stream->ctfser, (uint64_t) stream->sc->trace->uuid[i], 8, 8, BYTE_ORDER);
586 if (ret) {
587 BT_COMP_LOGE("Error writing packet header UUID: stream-file-name=%s",
588 stream->file_name->str);
589 goto end;
590 }
591 }
592
593 /* Packet header: stream class ID */
594 ret = bt_ctfser_write_byte_aligned_unsigned_int(
595 &stream->ctfser, bt_stream_class_get_id(stream->sc->ir_sc), 8, 64, BYTE_ORDER);
596 if (ret) {
597 BT_COMP_LOGE("Error writing packet header stream class id: "
598 "stream-file-name=%s, stream-class-id=%" PRIu64,
599 stream->file_name->str, bt_stream_class_get_id(stream->sc->ir_sc));
600 goto end;
601 }
602
603 /* Packet header: stream ID */
604 ret = bt_ctfser_write_byte_aligned_unsigned_int(
605 &stream->ctfser, bt_stream_get_id(stream->ir_stream), 8, 64, BYTE_ORDER);
606 if (ret) {
607 BT_COMP_LOGE("Error writing packet header stream id: "
608 "stream-file-name=%s, stream-id=%" PRIu64,
609 stream->file_name->str, bt_stream_get_id(stream->ir_stream));
610 goto end;
611 }
612
613 /* Save packet context's offset to rewrite it later */
614 stream->packet_state.context_offset_bits =
615 bt_ctfser_get_offset_in_current_packet_bits(&stream->ctfser);
616
617 /* Write packet context just to advance to content (first event) */
618 ret = write_packet_context(stream);
619 if (ret) {
620 goto end;
621 }
622
623 stream->packet_state.is_open = true;
624
625 end:
626 return ret;
627 }
628
629 int fs_sink_stream_close_packet(struct fs_sink_stream *stream, const bt_clock_snapshot *cs)
630 {
631 int ret;
632
633 BT_ASSERT(stream->packet_state.is_open);
634
635 if (cs) {
636 stream->packet_state.end_cs = bt_clock_snapshot_get_value(cs);
637 }
638
639 stream->packet_state.content_size =
640 bt_ctfser_get_offset_in_current_packet_bits(&stream->ctfser);
641 stream->packet_state.total_size = (stream->packet_state.content_size + 7) & ~UINT64_C(7);
642
643 /* Rewrite packet context */
644 bt_ctfser_set_offset_in_current_packet_bits(&stream->ctfser,
645 stream->packet_state.context_offset_bits);
646 ret = write_packet_context(stream);
647 if (ret) {
648 goto end;
649 }
650
651 /* Close packet */
652 bt_ctfser_close_current_packet(&stream->ctfser, stream->packet_state.total_size / 8);
653
654 /* Partially copy current packet state to previous packet state */
655 stream->prev_packet_state.end_cs = stream->packet_state.end_cs;
656 stream->prev_packet_state.discarded_events_counter =
657 stream->packet_state.discarded_events_counter;
658 stream->prev_packet_state.seq_num = stream->packet_state.seq_num;
659
660 /* Reset current packet state */
661 stream->packet_state.beginning_cs = UINT64_C(-1);
662 stream->packet_state.end_cs = UINT64_C(-1);
663 stream->packet_state.content_size = 0;
664 stream->packet_state.total_size = 0;
665 stream->packet_state.seq_num += 1;
666 stream->packet_state.context_offset_bits = 0;
667 stream->packet_state.is_open = false;
668 BT_PACKET_PUT_REF_AND_RESET(stream->packet_state.packet);
669
670 end:
671 return ret;
672 }
This page took 0.043117 seconds and 4 git commands to generate.