4 * Babeltrace CTF Writer Output Plugin Event Handling
6 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
8 * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
10 * Permission is hereby granted, free of charge, to any person obtaining a copy
11 * of this software and associated documentation files (the "Software"), to deal
12 * in the Software without restriction, including without limitation the rights
13 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
14 * copies of the Software, and to permit persons to whom the Software is
15 * furnished to do so, subject to the following conditions:
17 * The above copyright notice and this permission notice shall be included in
18 * all copies or substantial portions of the Software.
20 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
21 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
22 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
23 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
24 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
25 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
29 #include <babeltrace/ctf-ir/event.h>
30 #include <babeltrace/ctf-ir/packet.h>
31 #include <babeltrace/ctf-ir/event-class.h>
32 #include <babeltrace/ctf-ir/stream.h>
33 #include <babeltrace/ctf-ir/stream-class.h>
34 #include <babeltrace/ctf-ir/clock-class.h>
35 #include <babeltrace/ctf-ir/fields.h>
36 #include <babeltrace/ctf-writer/stream-class.h>
37 #include <babeltrace/ctf-writer/stream.h>
40 #include <ctfcopytrace.h>
45 void unref_stream_class(struct bt_ctf_stream_class
*writer_stream_class
)
47 bt_put(writer_stream_class
);
51 void unref_stream(struct bt_ctf_stream_class
*writer_stream
)
53 bt_put(writer_stream
);
56 gboolean
empty_ht(gpointer key
, gpointer value
, gpointer user_data
)
61 void destroy_stream_state_key(gpointer key
)
63 g_free((enum fs_writer_stream_state
*) key
);
67 void trace_is_static_listener(struct bt_ctf_trace
*trace
, void *data
)
73 struct bt_ctf_stream_class
*insert_new_stream_class(
74 struct writer_component
*writer_component
,
75 struct fs_writer
*fs_writer
,
76 struct bt_ctf_stream_class
*stream_class
)
78 struct bt_ctf_stream_class
*writer_stream_class
= NULL
;
79 struct bt_ctf_trace
*trace
= NULL
, *writer_trace
= NULL
;
80 struct bt_ctf_writer
*ctf_writer
= fs_writer
->writer
;
81 enum bt_component_status ret
;
83 trace
= bt_ctf_stream_class_get_trace(stream_class
);
85 fprintf(writer_component
->err
,
86 "[error] %s in %s:%d\n", __func__
, __FILE__
,
91 writer_trace
= bt_ctf_writer_get_trace(ctf_writer
);
93 fprintf(writer_component
->err
,
94 "[error] %s in %s:%d\n", __func__
, __FILE__
,
99 ret
= ctf_copy_clock_classes(writer_component
->err
, writer_trace
,
100 writer_stream_class
, trace
);
101 if (ret
!= BT_COMPONENT_STATUS_OK
) {
102 fprintf(writer_component
->err
,
103 "[error] %s in %s:%d\n", __func__
, __FILE__
,
108 writer_stream_class
= ctf_copy_stream_class(writer_component
->err
,
109 stream_class
, writer_trace
, true);
110 if (!writer_stream_class
) {
111 fprintf(writer_component
->err
, "[error] Failed to copy stream class\n");
112 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
113 __func__
, __FILE__
, __LINE__
);
117 g_hash_table_insert(fs_writer
->stream_class_map
,
118 (gpointer
) stream_class
, writer_stream_class
);
123 BT_PUT(writer_stream_class
);
125 bt_put(writer_trace
);
127 return writer_stream_class
;
131 enum fs_writer_stream_state
*insert_new_stream_state(
132 struct writer_component
*writer_component
,
133 struct fs_writer
*fs_writer
, struct bt_ctf_stream
*stream
)
135 enum fs_writer_stream_state
*v
= NULL
;
137 v
= g_new0(enum fs_writer_stream_state
, 1);
139 fprintf(writer_component
->err
,
140 "[error] %s in %s:%d\n", __func__
,
143 *v
= FS_WRITER_UNKNOWN_STREAM
;
145 g_hash_table_insert(fs_writer
->stream_states
, stream
, v
);
151 int make_trace_path(struct writer_component
*writer_component
,
152 struct bt_ctf_trace
*trace
, char *trace_path
)
155 const char *trace_name
;
157 trace_name
= bt_ctf_trace_get_name(trace
);
159 trace_name
= writer_component
->trace_name_base
->str
;
161 /* XXX: we might have to skip the first level, TBD. */
163 /* Sanitize the trace name. */
164 if (strlen(trace_name
) == 2 && !strcmp(trace_name
, "..")) {
165 fprintf(writer_component
->err
, "[error] Trace name cannot "
170 if (strstr(trace_name
, "../")) {
171 fprintf(writer_component
->err
, "[error] Trace name cannot "
172 "contain \"../\", received \"%s\"\n",
178 snprintf(trace_path
, PATH_MAX
, "%s/%s",
179 writer_component
->base_path
->str
,
181 if (g_file_test(trace_path
, G_FILE_TEST_EXISTS
)) {
184 snprintf(trace_path
, PATH_MAX
, "%s/%s-%d",
185 writer_component
->base_path
->str
,
187 } while (g_file_test(trace_path
, G_FILE_TEST_EXISTS
) && i
< INT_MAX
);
189 fprintf(writer_component
->err
, "[error] Unable to find "
190 "a unique trace path\n");
206 struct fs_writer
*insert_new_writer(
207 struct writer_component
*writer_component
,
208 struct bt_ctf_trace
*trace
)
210 struct bt_ctf_writer
*ctf_writer
= NULL
;
211 struct bt_ctf_trace
*writer_trace
= NULL
;
212 char trace_path
[PATH_MAX
];
213 enum bt_component_status ret
;
214 struct bt_ctf_stream
*stream
= NULL
;
215 struct fs_writer
*fs_writer
= NULL
;
218 ret
= make_trace_path(writer_component
, trace
, trace_path
);
220 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
221 __func__
, __FILE__
, __LINE__
);
225 printf("ctf.fs sink creating trace in %s\n", trace_path
);
227 ctf_writer
= bt_ctf_writer_create(trace_path
);
229 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
230 __func__
, __FILE__
, __LINE__
);
234 writer_trace
= bt_ctf_writer_get_trace(ctf_writer
);
236 fprintf(writer_component
->err
,
237 "[error] %s in %s:%d\n", __func__
, __FILE__
,
242 ret
= ctf_copy_trace(writer_component
->err
, trace
, writer_trace
);
243 if (ret
!= BT_COMPONENT_STATUS_OK
) {
244 fprintf(writer_component
->err
, "[error] Failed to copy trace\n");
245 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
246 __func__
, __FILE__
, __LINE__
);
251 fs_writer
= g_new0(struct fs_writer
, 1);
253 fprintf(writer_component
->err
,
254 "[error] %s in %s:%d\n", __func__
, __FILE__
,
258 fs_writer
->writer
= ctf_writer
;
259 fs_writer
->trace
= trace
;
260 fs_writer
->writer_trace
= writer_trace
;
261 BT_PUT(writer_trace
);
262 fs_writer
->stream_class_map
= g_hash_table_new_full(g_direct_hash
,
263 g_direct_equal
, NULL
, (GDestroyNotify
) unref_stream_class
);
264 fs_writer
->stream_map
= g_hash_table_new_full(g_direct_hash
,
265 g_direct_equal
, NULL
, (GDestroyNotify
) unref_stream
);
266 fs_writer
->stream_states
= g_hash_table_new_full(g_direct_hash
,
267 g_direct_equal
, NULL
, destroy_stream_state_key
);
269 /* Set all the existing streams in the unknown state. */
270 nr_stream
= bt_ctf_trace_get_stream_count(trace
);
271 for (i
= 0; i
< nr_stream
; i
++) {
272 stream
= bt_ctf_trace_get_stream_by_index(trace
, i
);
274 fprintf(writer_component
->err
,
275 "[error] %s in %s:%d\n", __func__
,
279 insert_new_stream_state(writer_component
, fs_writer
, stream
);
283 /* Check if the trace is already static or register a listener. */
284 if (bt_ctf_trace_is_static(trace
)) {
285 fs_writer
->trace_static
= 1;
286 fs_writer
->static_listener_id
= -1;
288 ret
= bt_ctf_trace_add_is_static_listener(trace
,
289 trace_is_static_listener
, &fs_writer
->trace_static
);
291 fprintf(writer_component
->err
,
292 "[error] %s in %s:%d\n", __func__
, __FILE__
,
296 fs_writer
->static_listener_id
= ret
;
299 g_hash_table_insert(writer_component
->trace_map
, (gpointer
) trace
,
307 bt_put(writer_trace
);
315 struct fs_writer
*get_fs_writer(struct writer_component
*writer_component
,
316 struct bt_ctf_stream_class
*stream_class
)
318 struct bt_ctf_trace
*trace
= NULL
;
319 struct fs_writer
*fs_writer
;
321 trace
= bt_ctf_stream_class_get_trace(stream_class
);
323 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
324 __func__
, __FILE__
, __LINE__
);
328 fs_writer
= g_hash_table_lookup(writer_component
->trace_map
,
331 fs_writer
= insert_new_writer(writer_component
, trace
);
343 struct fs_writer
*get_fs_writer_from_stream(
344 struct writer_component
*writer_component
,
345 struct bt_ctf_stream
*stream
)
347 struct bt_ctf_stream_class
*stream_class
= NULL
;
348 struct fs_writer
*fs_writer
;
350 stream_class
= bt_ctf_stream_get_class(stream
);
352 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
353 __func__
, __FILE__
, __LINE__
);
357 fs_writer
= get_fs_writer(writer_component
, stream_class
);
364 bt_put(stream_class
);
369 struct bt_ctf_stream_class
*lookup_stream_class(
370 struct writer_component
*writer_component
,
371 struct bt_ctf_stream_class
*stream_class
)
373 struct fs_writer
*fs_writer
= get_fs_writer(
374 writer_component
, stream_class
);
376 return (struct bt_ctf_stream_class
*) g_hash_table_lookup(
377 fs_writer
->stream_class_map
, (gpointer
) stream_class
);
381 struct bt_ctf_stream
*lookup_stream(struct writer_component
*writer_component
,
382 struct bt_ctf_stream
*stream
)
384 struct fs_writer
*fs_writer
= get_fs_writer_from_stream(
385 writer_component
, stream
);
387 return (struct bt_ctf_stream
*) g_hash_table_lookup(
388 fs_writer
->stream_map
, (gpointer
) stream
);
392 struct bt_ctf_stream
*insert_new_stream(
393 struct writer_component
*writer_component
,
394 struct fs_writer
*fs_writer
,
395 struct bt_ctf_stream_class
*stream_class
,
396 struct bt_ctf_stream
*stream
)
398 struct bt_ctf_stream
*writer_stream
= NULL
;
399 struct bt_ctf_stream_class
*writer_stream_class
= NULL
;
400 struct bt_ctf_writer
*ctf_writer
= bt_get(fs_writer
->writer
);
402 writer_stream_class
= lookup_stream_class(writer_component
,
404 if (!writer_stream_class
) {
405 writer_stream_class
= insert_new_stream_class(
406 writer_component
, fs_writer
, stream_class
);
407 if (!writer_stream_class
) {
408 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
409 __func__
, __FILE__
, __LINE__
);
413 bt_get(writer_stream_class
);
415 writer_stream
= bt_ctf_writer_create_stream(ctf_writer
,
416 writer_stream_class
);
417 if (!writer_stream
) {
418 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
419 __func__
, __FILE__
, __LINE__
);
423 g_hash_table_insert(fs_writer
->stream_map
, (gpointer
) stream
,
429 BT_PUT(writer_stream
);
432 bt_put(writer_stream_class
);
433 return writer_stream
;
437 struct bt_ctf_event_class
*get_event_class(struct writer_component
*writer_component
,
438 struct bt_ctf_stream_class
*writer_stream_class
,
439 struct bt_ctf_event_class
*event_class
)
441 return bt_ctf_stream_class_get_event_class_by_id(writer_stream_class
,
442 bt_ctf_event_class_get_id(event_class
));
446 struct bt_ctf_stream
*get_writer_stream(
447 struct writer_component
*writer_component
,
448 struct bt_ctf_packet
*packet
, struct bt_ctf_stream
*stream
)
450 struct bt_ctf_stream
*writer_stream
= NULL
;
452 writer_stream
= lookup_stream(writer_component
, stream
);
453 if (!writer_stream
) {
454 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
455 __func__
, __FILE__
, __LINE__
);
458 bt_get(writer_stream
);
463 BT_PUT(writer_stream
);
465 return writer_stream
;
469 void writer_close(struct writer_component
*writer_component
,
470 struct fs_writer
*fs_writer
)
472 if (fs_writer
->static_listener_id
> 0) {
473 bt_ctf_trace_remove_is_static_listener(fs_writer
->trace
,
474 fs_writer
->static_listener_id
);
477 /* Empty the stream class HT. */
478 g_hash_table_foreach_remove(fs_writer
->stream_class_map
,
480 g_hash_table_destroy(fs_writer
->stream_class_map
);
482 /* Empty the stream HT. */
483 g_hash_table_foreach_remove(fs_writer
->stream_map
,
485 g_hash_table_destroy(fs_writer
->stream_map
);
487 /* Empty the stream state HT. */
488 g_hash_table_foreach_remove(fs_writer
->stream_states
,
490 g_hash_table_destroy(fs_writer
->stream_states
);
494 enum bt_component_status
writer_stream_begin(
495 struct writer_component
*writer_component
,
496 struct bt_ctf_stream
*stream
)
498 struct bt_ctf_stream_class
*stream_class
= NULL
;
499 struct fs_writer
*fs_writer
;
500 struct bt_ctf_stream
*writer_stream
= NULL
;
501 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
502 enum fs_writer_stream_state
*state
;
504 stream_class
= bt_ctf_stream_get_class(stream
);
506 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
507 __func__
, __FILE__
, __LINE__
);
511 fs_writer
= get_fs_writer(writer_component
, stream_class
);
513 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
514 __func__
, __FILE__
, __LINE__
);
518 /* Set the stream as active */
519 state
= g_hash_table_lookup(fs_writer
->stream_states
, stream
);
521 if (fs_writer
->trace_static
) {
522 fprintf(writer_component
->err
, "[error] Adding a new "
523 "stream on a static trace\n");
526 state
= insert_new_stream_state(writer_component
, fs_writer
,
529 if (*state
!= FS_WRITER_UNKNOWN_STREAM
) {
530 fprintf(writer_component
->err
, "[error] Unexpected stream "
531 "state %d\n", *state
);
534 *state
= FS_WRITER_ACTIVE_STREAM
;
536 writer_stream
= insert_new_stream(writer_component
, fs_writer
,
537 stream_class
, stream
);
538 if (!writer_stream
) {
539 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
540 __func__
, __FILE__
, __LINE__
);
543 fs_writer
->active_streams
++;
548 ret
= BT_COMPONENT_STATUS_ERROR
;
550 bt_put(stream_class
);
554 void check_completed_trace(gpointer key
, gpointer value
, gpointer user_data
)
556 enum fs_writer_stream_state
*state
= value
;
557 int *trace_completed
= user_data
;
559 if (*state
!= FS_WRITER_COMPLETED_STREAM
) {
560 *trace_completed
= 0;
565 enum bt_component_status
writer_stream_end(
566 struct writer_component
*writer_component
,
567 struct bt_ctf_stream
*stream
)
569 struct bt_ctf_stream_class
*stream_class
= NULL
;
570 struct fs_writer
*fs_writer
;
571 struct bt_ctf_trace
*trace
= NULL
;
572 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
573 enum fs_writer_stream_state
*state
;
575 stream_class
= bt_ctf_stream_get_class(stream
);
577 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
578 __func__
, __FILE__
, __LINE__
);
582 fs_writer
= get_fs_writer(writer_component
, stream_class
);
584 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
585 __func__
, __FILE__
, __LINE__
);
589 state
= g_hash_table_lookup(fs_writer
->stream_states
, stream
);
590 if (*state
!= FS_WRITER_ACTIVE_STREAM
) {
591 fprintf(writer_component
->err
, "[error] Unexpected stream "
592 "state %d\n", *state
);
595 *state
= FS_WRITER_COMPLETED_STREAM
;
597 g_hash_table_remove(fs_writer
->stream_map
, stream
);
599 if (fs_writer
->trace_static
) {
600 int trace_completed
= 1;
602 g_hash_table_foreach(fs_writer
->stream_states
,
603 check_completed_trace
, &trace_completed
);
604 if (trace_completed
) {
605 writer_close(writer_component
, fs_writer
);
606 g_hash_table_remove(writer_component
->trace_map
,
614 ret
= BT_COMPONENT_STATUS_ERROR
;
617 BT_PUT(stream_class
);
622 enum bt_component_status
writer_new_packet(
623 struct writer_component
*writer_component
,
624 struct bt_ctf_packet
*packet
)
626 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
627 struct bt_ctf_field
*writer_packet_context
= NULL
;
628 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
631 stream
= bt_ctf_packet_get_stream(packet
);
633 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
634 __func__
, __FILE__
, __LINE__
);
638 writer_stream
= get_writer_stream(writer_component
, packet
, stream
);
639 if (!writer_stream
) {
640 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
641 __func__
, __FILE__
, __LINE__
);
646 writer_packet_context
= ctf_copy_packet_context(writer_component
->err
,
647 packet
, writer_stream
);
648 if (!writer_packet_context
) {
649 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
650 __func__
, __FILE__
, __LINE__
);
654 int_ret
= bt_ctf_stream_set_packet_context(writer_stream
,
655 writer_packet_context
);
657 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
661 BT_PUT(writer_stream
);
662 BT_PUT(writer_packet_context
);
667 ret
= BT_COMPONENT_STATUS_ERROR
;
669 bt_put(writer_stream
);
670 bt_put(writer_packet_context
);
676 enum bt_component_status
writer_close_packet(
677 struct writer_component
*writer_component
,
678 struct bt_ctf_packet
*packet
)
680 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
681 enum bt_component_status ret
;
683 stream
= bt_ctf_packet_get_stream(packet
);
685 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
686 __func__
, __FILE__
, __LINE__
);
690 writer_stream
= lookup_stream(writer_component
, stream
);
691 if (!writer_stream
) {
692 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
693 __func__
, __FILE__
, __LINE__
);
698 bt_get(writer_stream
);
700 ret
= bt_ctf_stream_flush(writer_stream
);
702 fprintf(writer_component
->err
,
703 "[error] Failed to flush packet\n");
706 BT_PUT(writer_stream
);
708 ret
= BT_COMPONENT_STATUS_OK
;
712 ret
= BT_COMPONENT_STATUS_ERROR
;
714 bt_put(writer_stream
);
720 enum bt_component_status
writer_output_event(
721 struct writer_component
*writer_component
,
722 struct bt_ctf_event
*event
)
724 enum bt_component_status ret
;
725 struct bt_ctf_event_class
*event_class
= NULL
, *writer_event_class
= NULL
;
726 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
727 struct bt_ctf_stream_class
*stream_class
= NULL
, *writer_stream_class
= NULL
;
728 struct bt_ctf_event
*writer_event
= NULL
;
729 const char *event_name
;
732 event_class
= bt_ctf_event_get_class(event
);
734 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
739 event_name
= bt_ctf_event_class_get_name(event_class
);
741 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
746 stream
= bt_ctf_event_get_stream(event
);
748 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
753 writer_stream
= lookup_stream(writer_component
, stream
);
754 if (!writer_stream
|| !bt_get(writer_stream
)) {
755 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
760 stream_class
= bt_ctf_event_class_get_stream_class(event_class
);
762 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
767 writer_stream_class
= lookup_stream_class(writer_component
, stream_class
);
768 if (!writer_stream_class
|| !bt_get(writer_stream_class
)) {
769 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
774 writer_event_class
= get_event_class(writer_component
,
775 writer_stream_class
, event_class
);
776 if (!writer_event_class
) {
777 writer_event_class
= ctf_copy_event_class(writer_component
->err
,
779 if (!writer_event_class
) {
780 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
781 __func__
, __FILE__
, __LINE__
);
784 int_ret
= bt_ctf_stream_class_add_event_class(
785 writer_stream_class
, writer_event_class
);
787 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
788 __func__
, __FILE__
, __LINE__
);
793 writer_event
= ctf_copy_event(writer_component
->err
, event
,
794 writer_event_class
, true);
796 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
798 fprintf(writer_component
->err
, "[error] Failed to copy event %s\n",
799 bt_ctf_event_class_get_name(writer_event_class
));
803 int_ret
= bt_ctf_stream_append_event(writer_stream
, writer_event
);
805 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
807 fprintf(writer_component
->err
, "[error] Failed to append event %s\n",
808 bt_ctf_event_class_get_name(writer_event_class
));
812 ret
= BT_COMPONENT_STATUS_OK
;
816 ret
= BT_COMPONENT_STATUS_ERROR
;
818 bt_put(writer_event
);
819 bt_put(writer_event_class
);
820 bt_put(writer_stream_class
);
821 bt_put(stream_class
);
822 bt_put(writer_stream
);