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 struct fs_writer
*insert_new_writer(
152 struct writer_component
*writer_component
,
153 struct bt_ctf_trace
*trace
)
155 struct bt_ctf_writer
*ctf_writer
= NULL
;
156 struct bt_ctf_trace
*writer_trace
= NULL
;
157 char trace_name
[PATH_MAX
];
158 enum bt_component_status ret
;
159 struct bt_ctf_stream
*stream
= NULL
;
160 struct fs_writer
*fs_writer
= NULL
;
163 /* FIXME: replace with trace name when it will work. */
164 snprintf(trace_name
, PATH_MAX
, "%s/%s_%03d",
165 writer_component
->base_path
->str
,
166 writer_component
->trace_name_base
->str
,
167 writer_component
->trace_id
++);
168 printf_verbose("CTF-Writer creating trace in %s\n", trace_name
);
170 ctf_writer
= bt_ctf_writer_create(trace_name
);
172 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
173 __func__
, __FILE__
, __LINE__
);
177 writer_trace
= bt_ctf_writer_get_trace(ctf_writer
);
179 fprintf(writer_component
->err
,
180 "[error] %s in %s:%d\n", __func__
, __FILE__
,
185 ret
= ctf_copy_trace(writer_component
->err
, trace
, writer_trace
);
186 if (ret
!= BT_COMPONENT_STATUS_OK
) {
187 fprintf(writer_component
->err
, "[error] Failed to copy trace\n");
188 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
189 __func__
, __FILE__
, __LINE__
);
194 fs_writer
= g_new0(struct fs_writer
, 1);
196 fprintf(writer_component
->err
,
197 "[error] %s in %s:%d\n", __func__
, __FILE__
,
201 fs_writer
->writer
= ctf_writer
;
202 fs_writer
->trace
= trace
;
203 fs_writer
->writer_trace
= writer_trace
;
204 BT_PUT(writer_trace
);
205 fs_writer
->stream_class_map
= g_hash_table_new_full(g_direct_hash
,
206 g_direct_equal
, NULL
, (GDestroyNotify
) unref_stream_class
);
207 fs_writer
->stream_map
= g_hash_table_new_full(g_direct_hash
,
208 g_direct_equal
, NULL
, (GDestroyNotify
) unref_stream
);
209 fs_writer
->stream_states
= g_hash_table_new_full(g_direct_hash
,
210 g_direct_equal
, NULL
, destroy_stream_state_key
);
212 /* Set all the existing streams in the unknown state. */
213 nr_stream
= bt_ctf_trace_get_stream_count(trace
);
214 for (i
= 0; i
< nr_stream
; i
++) {
215 stream
= bt_ctf_trace_get_stream_by_index(trace
, i
);
217 fprintf(writer_component
->err
,
218 "[error] %s in %s:%d\n", __func__
,
222 insert_new_stream_state(writer_component
, fs_writer
, stream
);
226 /* Check if the trace is already static or register a listener. */
227 if (bt_ctf_trace_is_static(trace
)) {
228 fs_writer
->trace_static
= 1;
229 fs_writer
->static_listener_id
= -1;
231 ret
= bt_ctf_trace_add_is_static_listener(trace
,
232 trace_is_static_listener
, &fs_writer
->trace_static
);
234 fprintf(writer_component
->err
,
235 "[error] %s in %s:%d\n", __func__
, __FILE__
,
239 fs_writer
->static_listener_id
= ret
;
242 g_hash_table_insert(writer_component
->trace_map
, (gpointer
) trace
,
250 bt_put(writer_trace
);
258 struct fs_writer
*get_fs_writer(struct writer_component
*writer_component
,
259 struct bt_ctf_stream_class
*stream_class
)
261 struct bt_ctf_trace
*trace
= NULL
;
262 struct fs_writer
*fs_writer
;
264 trace
= bt_ctf_stream_class_get_trace(stream_class
);
266 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
267 __func__
, __FILE__
, __LINE__
);
271 fs_writer
= g_hash_table_lookup(writer_component
->trace_map
,
274 fs_writer
= insert_new_writer(writer_component
, trace
);
286 struct fs_writer
*get_fs_writer_from_stream(
287 struct writer_component
*writer_component
,
288 struct bt_ctf_stream
*stream
)
290 struct bt_ctf_stream_class
*stream_class
= NULL
;
291 struct fs_writer
*fs_writer
;
293 stream_class
= bt_ctf_stream_get_class(stream
);
295 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
296 __func__
, __FILE__
, __LINE__
);
300 fs_writer
= get_fs_writer(writer_component
, stream_class
);
307 bt_put(stream_class
);
312 struct bt_ctf_stream_class
*lookup_stream_class(
313 struct writer_component
*writer_component
,
314 struct bt_ctf_stream_class
*stream_class
)
316 struct fs_writer
*fs_writer
= get_fs_writer(
317 writer_component
, stream_class
);
319 return (struct bt_ctf_stream_class
*) g_hash_table_lookup(
320 fs_writer
->stream_class_map
, (gpointer
) stream_class
);
324 struct bt_ctf_stream
*lookup_stream(struct writer_component
*writer_component
,
325 struct bt_ctf_stream
*stream
)
327 struct fs_writer
*fs_writer
= get_fs_writer_from_stream(
328 writer_component
, stream
);
330 return (struct bt_ctf_stream
*) g_hash_table_lookup(
331 fs_writer
->stream_map
, (gpointer
) stream
);
335 struct bt_ctf_stream
*insert_new_stream(
336 struct writer_component
*writer_component
,
337 struct fs_writer
*fs_writer
,
338 struct bt_ctf_stream_class
*stream_class
,
339 struct bt_ctf_stream
*stream
)
341 struct bt_ctf_stream
*writer_stream
= NULL
;
342 struct bt_ctf_stream_class
*writer_stream_class
= NULL
;
343 struct bt_ctf_writer
*ctf_writer
= bt_get(fs_writer
->writer
);
345 writer_stream_class
= lookup_stream_class(writer_component
,
347 if (!writer_stream_class
) {
348 writer_stream_class
= insert_new_stream_class(
349 writer_component
, fs_writer
, stream_class
);
350 if (!writer_stream_class
) {
351 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
352 __func__
, __FILE__
, __LINE__
);
356 bt_get(writer_stream_class
);
358 writer_stream
= bt_ctf_writer_create_stream(ctf_writer
,
359 writer_stream_class
);
360 if (!writer_stream
) {
361 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
362 __func__
, __FILE__
, __LINE__
);
366 g_hash_table_insert(fs_writer
->stream_map
, (gpointer
) stream
,
372 BT_PUT(writer_stream
);
375 bt_put(writer_stream_class
);
376 return writer_stream
;
380 struct bt_ctf_event_class
*get_event_class(struct writer_component
*writer_component
,
381 struct bt_ctf_stream_class
*writer_stream_class
,
382 struct bt_ctf_event_class
*event_class
)
384 return bt_ctf_stream_class_get_event_class_by_id(writer_stream_class
,
385 bt_ctf_event_class_get_id(event_class
));
389 struct bt_ctf_stream
*get_writer_stream(
390 struct writer_component
*writer_component
,
391 struct bt_ctf_packet
*packet
, struct bt_ctf_stream
*stream
)
393 struct bt_ctf_stream
*writer_stream
= NULL
;
395 writer_stream
= lookup_stream(writer_component
, stream
);
396 if (!writer_stream
) {
397 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
398 __func__
, __FILE__
, __LINE__
);
401 bt_get(writer_stream
);
406 BT_PUT(writer_stream
);
408 return writer_stream
;
412 void writer_close(struct writer_component
*writer_component
,
413 struct fs_writer
*fs_writer
)
415 if (fs_writer
->static_listener_id
> 0) {
416 bt_ctf_trace_remove_is_static_listener(fs_writer
->trace
,
417 fs_writer
->static_listener_id
);
420 /* Empty the stream class HT. */
421 g_hash_table_foreach_remove(fs_writer
->stream_class_map
,
423 g_hash_table_destroy(fs_writer
->stream_class_map
);
425 /* Empty the stream HT. */
426 g_hash_table_foreach_remove(fs_writer
->stream_map
,
428 g_hash_table_destroy(fs_writer
->stream_map
);
430 /* Empty the stream state HT. */
431 g_hash_table_foreach_remove(fs_writer
->stream_states
,
433 g_hash_table_destroy(fs_writer
->stream_states
);
437 enum bt_component_status
writer_stream_begin(
438 struct writer_component
*writer_component
,
439 struct bt_ctf_stream
*stream
)
441 struct bt_ctf_stream_class
*stream_class
= NULL
;
442 struct fs_writer
*fs_writer
;
443 struct bt_ctf_stream
*writer_stream
= NULL
;
444 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
445 enum fs_writer_stream_state
*state
;
447 stream_class
= bt_ctf_stream_get_class(stream
);
449 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
450 __func__
, __FILE__
, __LINE__
);
454 fs_writer
= get_fs_writer(writer_component
, stream_class
);
456 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
457 __func__
, __FILE__
, __LINE__
);
461 /* Set the stream as active */
462 state
= g_hash_table_lookup(fs_writer
->stream_states
, stream
);
464 if (fs_writer
->trace_static
) {
465 fprintf(writer_component
->err
, "[error] Adding a new "
466 "stream on a static trace\n");
469 state
= insert_new_stream_state(writer_component
, fs_writer
,
472 if (*state
!= FS_WRITER_UNKNOWN_STREAM
) {
473 fprintf(writer_component
->err
, "[error] Unexpected stream "
474 "state %d\n", *state
);
477 *state
= FS_WRITER_ACTIVE_STREAM
;
479 writer_stream
= insert_new_stream(writer_component
, fs_writer
,
480 stream_class
, stream
);
481 if (!writer_stream
) {
482 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
483 __func__
, __FILE__
, __LINE__
);
486 fs_writer
->active_streams
++;
491 ret
= BT_COMPONENT_STATUS_ERROR
;
493 bt_put(stream_class
);
497 void check_completed_trace(gpointer key
, gpointer value
, gpointer user_data
)
499 enum fs_writer_stream_state
*state
= value
;
500 int *trace_completed
= user_data
;
502 if (*state
!= FS_WRITER_COMPLETED_STREAM
) {
503 *trace_completed
= 0;
508 enum bt_component_status
writer_stream_end(
509 struct writer_component
*writer_component
,
510 struct bt_ctf_stream
*stream
)
512 struct bt_ctf_stream_class
*stream_class
= NULL
;
513 struct fs_writer
*fs_writer
;
514 struct bt_ctf_trace
*trace
= NULL
;
515 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
516 enum fs_writer_stream_state
*state
;
518 stream_class
= bt_ctf_stream_get_class(stream
);
520 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
521 __func__
, __FILE__
, __LINE__
);
525 fs_writer
= get_fs_writer(writer_component
, stream_class
);
527 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
528 __func__
, __FILE__
, __LINE__
);
532 state
= g_hash_table_lookup(fs_writer
->stream_states
, stream
);
533 if (*state
!= FS_WRITER_ACTIVE_STREAM
) {
534 fprintf(writer_component
->err
, "[error] Unexpected stream "
535 "state %d\n", *state
);
538 *state
= FS_WRITER_COMPLETED_STREAM
;
540 g_hash_table_remove(fs_writer
->stream_map
, stream
);
542 if (fs_writer
->trace_static
) {
543 int trace_completed
= 1;
545 g_hash_table_foreach(fs_writer
->stream_states
,
546 check_completed_trace
, &trace_completed
);
547 if (trace_completed
) {
548 writer_close(writer_component
, fs_writer
);
549 g_hash_table_remove(writer_component
->trace_map
,
557 ret
= BT_COMPONENT_STATUS_ERROR
;
560 BT_PUT(stream_class
);
565 enum bt_component_status
writer_new_packet(
566 struct writer_component
*writer_component
,
567 struct bt_ctf_packet
*packet
)
569 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
570 struct bt_ctf_field
*writer_packet_context
= NULL
;
571 enum bt_component_status ret
= BT_COMPONENT_STATUS_OK
;
574 stream
= bt_ctf_packet_get_stream(packet
);
576 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
577 __func__
, __FILE__
, __LINE__
);
581 writer_stream
= get_writer_stream(writer_component
, packet
, stream
);
582 if (!writer_stream
) {
583 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
584 __func__
, __FILE__
, __LINE__
);
589 writer_packet_context
= ctf_copy_packet_context(writer_component
->err
,
590 packet
, writer_stream
);
591 if (!writer_packet_context
) {
592 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
593 __func__
, __FILE__
, __LINE__
);
597 int_ret
= bt_ctf_stream_set_packet_context(writer_stream
,
598 writer_packet_context
);
600 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
604 BT_PUT(writer_stream
);
605 BT_PUT(writer_packet_context
);
610 ret
= BT_COMPONENT_STATUS_ERROR
;
612 bt_put(writer_stream
);
613 bt_put(writer_packet_context
);
619 enum bt_component_status
writer_close_packet(
620 struct writer_component
*writer_component
,
621 struct bt_ctf_packet
*packet
)
623 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
624 enum bt_component_status ret
;
626 stream
= bt_ctf_packet_get_stream(packet
);
628 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
629 __func__
, __FILE__
, __LINE__
);
633 writer_stream
= lookup_stream(writer_component
, stream
);
634 if (!writer_stream
) {
635 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
636 __func__
, __FILE__
, __LINE__
);
641 bt_get(writer_stream
);
643 ret
= bt_ctf_stream_flush(writer_stream
);
645 fprintf(writer_component
->err
,
646 "[error] Failed to flush packet\n");
649 BT_PUT(writer_stream
);
651 ret
= BT_COMPONENT_STATUS_OK
;
655 ret
= BT_COMPONENT_STATUS_ERROR
;
657 bt_put(writer_stream
);
663 enum bt_component_status
writer_output_event(
664 struct writer_component
*writer_component
,
665 struct bt_ctf_event
*event
)
667 enum bt_component_status ret
;
668 struct bt_ctf_event_class
*event_class
= NULL
, *writer_event_class
= NULL
;
669 struct bt_ctf_stream
*stream
= NULL
, *writer_stream
= NULL
;
670 struct bt_ctf_stream_class
*stream_class
= NULL
, *writer_stream_class
= NULL
;
671 struct bt_ctf_event
*writer_event
= NULL
;
672 const char *event_name
;
675 event_class
= bt_ctf_event_get_class(event
);
677 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
682 event_name
= bt_ctf_event_class_get_name(event_class
);
684 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
689 stream
= bt_ctf_event_get_stream(event
);
691 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
696 writer_stream
= lookup_stream(writer_component
, stream
);
697 if (!writer_stream
|| !bt_get(writer_stream
)) {
698 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
703 stream_class
= bt_ctf_event_class_get_stream_class(event_class
);
705 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
710 writer_stream_class
= lookup_stream_class(writer_component
, stream_class
);
711 if (!writer_stream_class
|| !bt_get(writer_stream_class
)) {
712 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
717 writer_event_class
= get_event_class(writer_component
,
718 writer_stream_class
, event_class
);
719 if (!writer_event_class
) {
720 writer_event_class
= ctf_copy_event_class(writer_component
->err
,
722 if (!writer_event_class
) {
723 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
724 __func__
, __FILE__
, __LINE__
);
727 int_ret
= bt_ctf_stream_class_add_event_class(
728 writer_stream_class
, writer_event_class
);
730 fprintf(writer_component
->err
, "[error] %s in %s:%d\n",
731 __func__
, __FILE__
, __LINE__
);
736 writer_event
= ctf_copy_event(writer_component
->err
, event
,
737 writer_event_class
, true);
739 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
741 fprintf(writer_component
->err
, "[error] Failed to copy event %s\n",
742 bt_ctf_event_class_get_name(writer_event_class
));
746 int_ret
= bt_ctf_stream_append_event(writer_stream
, writer_event
);
748 fprintf(writer_component
->err
, "[error] %s in %s:%d\n", __func__
,
750 fprintf(writer_component
->err
, "[error] Failed to append event %s\n",
751 bt_ctf_event_class_get_name(writer_event_class
));
755 ret
= BT_COMPONENT_STATUS_OK
;
759 ret
= BT_COMPONENT_STATUS_ERROR
;
761 bt_put(writer_event
);
762 bt_put(writer_event_class
);
763 bt_put(writer_stream_class
);
764 bt_put(stream_class
);
765 bt_put(writer_stream
);