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 struct fs_writer
*insert_new_writer(
132 struct writer_component
*writer_component
,
133 struct bt_ctf_trace
*trace
)
135 struct bt_ctf_writer
*ctf_writer
= NULL
;
136 struct bt_ctf_trace
*writer_trace
= NULL
;
137 char trace_name
[PATH_MAX
];
138 enum bt_component_status ret
;
139 struct bt_ctf_stream
*stream
= NULL
;
140 struct fs_writer
*fs_writer
= NULL
;
143 /* FIXME: replace with trace name when it will work. */
144 snprintf(trace_name
, PATH_MAX
, "%s/%s_%03d",
145 writer_component
->base_path
->str
,
146 writer_component
->trace_name_base
->str
,
147 writer_component
->trace_id
++);
148 printf_verbose("CTF-Writer creating trace in %s\n", trace_name
);
150 ctf_writer
= bt_ctf_writer_create(trace_name
);
152 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
153 __func__
, __FILE__
, __LINE__
);
157 writer_trace
= bt_ctf_writer_get_trace(ctf_writer
);
159 fprintf(writer_component
->err
,
160 "[error] %s in %s:%d\n", __func__
, __FILE__
,
165 ret
= ctf_copy_trace(writer_component
->err
, trace
, writer_trace
);
166 if (ret
!= BT_COMPONENT_STATUS_OK
) {
167 fprintf(writer_component
->err
, "[error] Failed to copy trace\n");
168 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
169 __func__
, __FILE__
, __LINE__
);
174 fs_writer
= g_new0(struct fs_writer
, 1);
176 fprintf(writer_component
->err
,
177 "[error] %s in %s:%d\n", __func__
, __FILE__
,
181 fs_writer
->writer
= ctf_writer
;
182 fs_writer
->trace
= trace
;
183 fs_writer
->writer_trace
= writer_trace
;
184 BT_PUT(writer_trace
);
185 fs_writer
->stream_class_map
= g_hash_table_new_full(g_direct_hash
,
186 g_direct_equal
, NULL
, (GDestroyNotify
) unref_stream_class
);
187 fs_writer
->stream_map
= g_hash_table_new_full(g_direct_hash
,
188 g_direct_equal
, NULL
, (GDestroyNotify
) unref_stream
);
189 fs_writer
->stream_states
= g_hash_table_new_full(g_direct_hash
,
190 g_direct_equal
, NULL
, destroy_stream_state_key
);
192 /* Set all the existing streams in the unknown state. */
193 nr_stream
= bt_ctf_trace_get_stream_count(trace
);
194 for (i
= 0; i
< nr_stream
; i
++) {
195 enum fs_writer_stream_state
*v
;
197 stream
= bt_ctf_trace_get_stream_by_index(trace
, i
);
199 fprintf(writer_component
->err
,
200 "[error] %s in %s:%d\n", __func__
,
205 v
= g_new0(enum fs_writer_stream_state
, 1);
207 fprintf(writer_component
->err
,
208 "[error] %s in %s:%d\n", __func__
,
212 *v
= FS_WRITER_UNKNOWN_STREAM
;
214 g_hash_table_insert(fs_writer
->stream_states
, stream
, v
);
218 /* Check if the trace is already static or register a listener. */
219 if (bt_ctf_trace_is_static(trace
)) {
220 fs_writer
->trace_static
= 1;
221 fs_writer
->static_listener_id
= -1;
223 ret
= bt_ctf_trace_add_is_static_listener(trace
,
224 trace_is_static_listener
, &fs_writer
->trace_static
);
226 fprintf(writer_component
->err
,
227 "[error] %s in %s:%d\n", __func__
, __FILE__
,
231 fs_writer
->static_listener_id
= ret
;
234 g_hash_table_insert(writer_component
->trace_map
, (gpointer
) trace
,
242 bt_put(writer_trace
);
250 struct fs_writer
*get_fs_writer(struct writer_component
*writer_component
,
251 struct bt_ctf_stream_class
*stream_class
)
253 struct bt_ctf_trace
*trace
= NULL
;
254 struct fs_writer
*fs_writer
;
256 trace
= bt_ctf_stream_class_get_trace(stream_class
);
258 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
259 __func__
, __FILE__
, __LINE__
);
263 fs_writer
= g_hash_table_lookup(writer_component
->trace_map
,
266 fs_writer
= insert_new_writer(writer_component
, trace
);
278 struct fs_writer
*get_fs_writer_from_stream(
279 struct writer_component
*writer_component
,
280 struct bt_ctf_stream
*stream
)
282 struct bt_ctf_stream_class
*stream_class
= NULL
;
283 struct fs_writer
*fs_writer
;
285 stream_class
= bt_ctf_stream_get_class(stream
);
287 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
288 __func__
, __FILE__
, __LINE__
);
292 fs_writer
= get_fs_writer(writer_component
, stream_class
);
299 bt_put(stream_class
);
304 struct bt_ctf_stream_class
*lookup_stream_class(
305 struct writer_component
*writer_component
,
306 struct bt_ctf_stream_class
*stream_class
)
308 struct fs_writer
*fs_writer
= get_fs_writer(
309 writer_component
, stream_class
);
311 return (struct bt_ctf_stream_class
*) g_hash_table_lookup(
312 fs_writer
->stream_class_map
, (gpointer
) stream_class
);
316 struct bt_ctf_stream
*lookup_stream(struct writer_component
*writer_component
,
317 struct bt_ctf_stream
*stream
)
319 struct fs_writer
*fs_writer
= get_fs_writer_from_stream(
320 writer_component
, stream
);
322 return (struct bt_ctf_stream
*) g_hash_table_lookup(
323 fs_writer
->stream_map
, (gpointer
) stream
);
327 struct bt_ctf_stream
*insert_new_stream(
328 struct writer_component
*writer_component
,
329 struct fs_writer
*fs_writer
,
330 struct bt_ctf_stream_class
*stream_class
,
331 struct bt_ctf_stream
*stream
)
333 struct bt_ctf_stream
*writer_stream
= NULL
;
334 struct bt_ctf_stream_class
*writer_stream_class
= NULL
;
335 struct bt_ctf_writer
*ctf_writer
= bt_get(fs_writer
->writer
);
337 writer_stream_class
= lookup_stream_class(writer_component
,
339 if (!writer_stream_class
) {
340 writer_stream_class
= insert_new_stream_class(
341 writer_component
, fs_writer
, stream_class
);
342 if (!writer_stream_class
) {
343 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
344 __func__
, __FILE__
, __LINE__
);
348 bt_get(writer_stream_class
);
350 writer_stream
= bt_ctf_writer_create_stream(ctf_writer
,
351 writer_stream_class
);
352 if (!writer_stream
) {
353 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
354 __func__
, __FILE__
, __LINE__
);
358 g_hash_table_insert(fs_writer
->stream_map
, (gpointer
) stream
,
364 BT_PUT(writer_stream
);
367 bt_put(writer_stream_class
);
368 return writer_stream
;
372 struct bt_ctf_event_class
*get_event_class(struct writer_component
*writer_component
,
373 struct bt_ctf_stream_class
*writer_stream_class
,
374 struct bt_ctf_event_class
*event_class
)
376 return bt_ctf_stream_class_get_event_class_by_id(writer_stream_class
,
377 bt_ctf_event_class_get_id(event_class
));
381 struct bt_ctf_stream
*get_writer_stream(
382 struct writer_component
*writer_component
,
383 struct bt_ctf_packet
*packet
, struct bt_ctf_stream
*stream
)
385 struct bt_ctf_stream
*writer_stream
= NULL
;
387 writer_stream
= lookup_stream(writer_component
, stream
);
388 if (!writer_stream
) {
389 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
390 __func__
, __FILE__
, __LINE__
);
393 bt_get(writer_stream
);
398 BT_PUT(writer_stream
);
400 return writer_stream
;
404 void writer_close(struct writer_component
*writer_component
,
405 struct fs_writer
*fs_writer
)
407 if (fs_writer
->static_listener_id
> 0) {
408 bt_ctf_trace_remove_is_static_listener(fs_writer
->trace
,
409 fs_writer
->static_listener_id
);
412 /* Empty the stream class HT. */
413 g_hash_table_foreach_remove(fs_writer
->stream_class_map
,
415 g_hash_table_destroy(fs_writer
->stream_class_map
);
417 /* Empty the stream HT. */
418 g_hash_table_foreach_remove(fs_writer
->stream_map
,
420 g_hash_table_destroy(fs_writer
->stream_map
);
422 /* Empty the stream state HT. */
423 g_hash_table_foreach_remove(fs_writer
->stream_states
,
425 g_hash_table_destroy(fs_writer
->stream_states
);
429 enum bt_component_status
writer_stream_begin(
430 struct writer_component
*writer_component
,
431 struct bt_ctf_stream
*stream
)
433 struct bt_ctf_stream_class
*stream_class
= NULL
;
434 struct fs_writer
*fs_writer
;
435 struct bt_ctf_stream
*writer_stream
= NULL
;
436 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
437 enum fs_writer_stream_state
*state
;
439 stream_class
= bt_ctf_stream_get_class(stream
);
441 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
442 __func__
, __FILE__
, __LINE__
);
446 fs_writer
= get_fs_writer(writer_component
, stream_class
);
448 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
449 __func__
, __FILE__
, __LINE__
);
453 /* Set the stream as active */
454 state
= g_hash_table_lookup(fs_writer
->stream_states
, stream
);
455 if (*state
!= FS_WRITER_UNKNOWN_STREAM
) {
456 fprintf(writer_component
->err
, "[error] Unexpected stream "
457 "state %d\n", *state
);
460 *state
= FS_WRITER_ACTIVE_STREAM
;
462 writer_stream
= insert_new_stream(writer_component
, fs_writer
,
463 stream_class
, stream
);
464 if (!writer_stream
) {
465 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
466 __func__
, __FILE__
, __LINE__
);
469 fs_writer
->active_streams
++;
474 ret
= BT_COMPONENT_STATUS_ERROR
;
476 bt_put(stream_class
);
480 void check_completed_trace(gpointer key
, gpointer value
, gpointer user_data
)
482 enum fs_writer_stream_state
*state
= value
;
483 int *trace_completed
= user_data
;
485 if (*state
!= FS_WRITER_COMPLETED_STREAM
) {
486 *trace_completed
= 0;
491 enum bt_component_status
writer_stream_end(
492 struct writer_component
*writer_component
,
493 struct bt_ctf_stream
*stream
)
495 struct bt_ctf_stream_class
*stream_class
= NULL
;
496 struct fs_writer
*fs_writer
;
497 struct bt_ctf_trace
*trace
= NULL
;
498 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
499 enum fs_writer_stream_state
*state
;
501 stream_class
= bt_ctf_stream_get_class(stream
);
503 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
504 __func__
, __FILE__
, __LINE__
);
508 fs_writer
= get_fs_writer(writer_component
, stream_class
);
510 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
511 __func__
, __FILE__
, __LINE__
);
515 state
= g_hash_table_lookup(fs_writer
->stream_states
, stream
);
516 if (*state
!= FS_WRITER_ACTIVE_STREAM
) {
517 fprintf(writer_component
->err
, "[error] Unexpected stream "
518 "state %d\n", *state
);
521 *state
= FS_WRITER_COMPLETED_STREAM
;
523 g_hash_table_remove(fs_writer
->stream_map
, stream
);
525 if (fs_writer
->trace_static
) {
526 int trace_completed
= 1;
528 g_hash_table_foreach(fs_writer
->stream_states
,
529 check_completed_trace
, &trace_completed
);
530 if (trace_completed
) {
531 writer_close(writer_component
, fs_writer
);
532 g_hash_table_remove(writer_component
->trace_map
,
540 ret
= BT_COMPONENT_STATUS_ERROR
;
543 BT_PUT(stream_class
);
548 enum bt_component_status
writer_new_packet(
549 struct writer_component
*writer_component
,
550 struct bt_ctf_packet
*packet
)
552 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
553 struct bt_ctf_field
*writer_packet_context
= NULL
;
554 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
557 stream
= bt_ctf_packet_get_stream(packet
);
559 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
560 __func__
, __FILE__
, __LINE__
);
564 writer_stream
= get_writer_stream(writer_component
, packet
, stream
);
565 if (!writer_stream
) {
566 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
567 __func__
, __FILE__
, __LINE__
);
572 writer_packet_context
= ctf_copy_packet_context(writer_component
->err
,
573 packet
, writer_stream
, 1);
574 if (!writer_packet_context
) {
575 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
576 __func__
, __FILE__
, __LINE__
);
580 int_ret
= bt_ctf_stream_set_packet_context(writer_stream
,
581 writer_packet_context
);
583 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
587 BT_PUT(writer_stream
);
588 BT_PUT(writer_packet_context
);
593 ret
= BT_COMPONENT_STATUS_ERROR
;
595 bt_put(writer_stream
);
596 bt_put(writer_packet_context
);
602 enum bt_component_status
writer_close_packet(
603 struct writer_component
*writer_component
,
604 struct bt_ctf_packet
*packet
)
606 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
607 enum bt_component_status ret
;
609 stream
= bt_ctf_packet_get_stream(packet
);
611 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
612 __func__
, __FILE__
, __LINE__
);
616 writer_stream
= lookup_stream(writer_component
, stream
);
617 if (!writer_stream
) {
618 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
619 __func__
, __FILE__
, __LINE__
);
624 bt_get(writer_stream
);
626 ret
= bt_ctf_stream_flush(writer_stream
);
628 fprintf(writer_component
->err
,
629 "[error] Failed to flush packet\n");
632 BT_PUT(writer_stream
);
634 ret
= BT_COMPONENT_STATUS_OK
;
638 ret
= BT_COMPONENT_STATUS_ERROR
;
640 bt_put(writer_stream
);
646 enum bt_component_status
writer_output_event(
647 struct writer_component
*writer_component
,
648 struct bt_ctf_event
*event
)
650 enum bt_component_status ret
;
651 struct bt_ctf_event_class
*event_class
= NULL
, *writer_event_class
= NULL
;
652 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
653 struct bt_ctf_stream_class
*stream_class
= NULL
, *writer_stream_class
= NULL
;
654 struct bt_ctf_event
*writer_event
= NULL
;
655 const char *event_name
;
658 event_class
= bt_ctf_event_get_class(event
);
660 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
665 event_name
= bt_ctf_event_class_get_name(event_class
);
667 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
672 stream
= bt_ctf_event_get_stream(event
);
674 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
679 writer_stream
= lookup_stream(writer_component
, stream
);
680 if (!writer_stream
|| !bt_get(writer_stream
)) {
681 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
686 stream_class
= bt_ctf_event_class_get_stream_class(event_class
);
688 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
693 writer_stream_class
= lookup_stream_class(writer_component
, stream_class
);
694 if (!writer_stream_class
|| !bt_get(writer_stream_class
)) {
695 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
700 writer_event_class
= get_event_class(writer_component
,
701 writer_stream_class
, event_class
);
702 if (!writer_event_class
) {
703 writer_event_class
= ctf_copy_event_class(writer_component
->err
,
705 if (!writer_event_class
) {
706 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
707 __func__
, __FILE__
, __LINE__
);
710 int_ret
= bt_ctf_stream_class_add_event_class(
711 writer_stream_class
, writer_event_class
);
713 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
714 __func__
, __FILE__
, __LINE__
);
719 writer_event
= ctf_copy_event(writer_component
->err
, event
,
720 writer_event_class
, true);
722 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
724 fprintf(writer_component
->err
, "[error] Failed to copy event %s\n",
725 bt_ctf_event_class_get_name(writer_event_class
));
729 int_ret
= bt_ctf_stream_append_event(writer_stream
, writer_event
);
731 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
733 fprintf(writer_component
->err
, "[error] Failed to append event %s\n",
734 bt_ctf_event_class_get_name(writer_event_class
));
738 ret
= BT_COMPONENT_STATUS_OK
;
742 ret
= BT_COMPONENT_STATUS_ERROR
;
744 bt_put(writer_event
);
745 bt_put(writer_event_class
);
746 bt_put(writer_stream_class
);
747 bt_put(stream_class
);
748 bt_put(writer_stream
);