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