lib: rename "notification" -> "message"
[babeltrace.git] / plugins / ctf / fs-src / fs.c
1 /*
2 * fs.c
3 *
4 * Babeltrace CTF file system Reader Component
5 *
6 * Copyright 2015-2017 Philippe Proulx <pproulx@efficios.com>
7 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
8 *
9 * Permission is hereby granted, free of charge, to any person obtaining a copy
10 * of this software and associated documentation files (the "Software"), to deal
11 * in the Software without restriction, including without limitation the rights
12 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
13 * copies of the Software, and to permit persons to whom the Software is
14 * furnished to do so, subject to the following conditions:
15 *
16 * The above copyright notice and this permission notice shall be included in
17 * all copies or substantial portions of the Software.
18 *
19 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
20 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
21 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
22 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
23 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
24 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
25 * SOFTWARE.
26 */
27
28 #include <babeltrace/common-internal.h>
29 #include <babeltrace/babeltrace.h>
30 #include <plugins-common.h>
31 #include <glib.h>
32 #include <babeltrace/assert-internal.h>
33 #include <inttypes.h>
34 #include <stdbool.h>
35 #include "fs.h"
36 #include "metadata.h"
37 #include "data-stream-file.h"
38 #include "file.h"
39 #include "../common/metadata/decoder.h"
40 #include "../common/msg-iter/msg-iter.h"
41 #include "../common/utils/utils.h"
42 #include "query.h"
43
44 #define BT_LOG_TAG "PLUGIN-CTF-FS-SRC"
45 #include "logging.h"
46
47 static
48 int msg_iter_data_set_current_ds_file(struct ctf_fs_msg_iter_data *msg_iter_data)
49 {
50 struct ctf_fs_ds_file_info *ds_file_info;
51 int ret = 0;
52
53 BT_ASSERT(msg_iter_data->ds_file_info_index <
54 msg_iter_data->ds_file_group->ds_file_infos->len);
55 ds_file_info = g_ptr_array_index(
56 msg_iter_data->ds_file_group->ds_file_infos,
57 msg_iter_data->ds_file_info_index);
58
59 ctf_fs_ds_file_destroy(msg_iter_data->ds_file);
60 msg_iter_data->ds_file = ctf_fs_ds_file_create(
61 msg_iter_data->ds_file_group->ctf_fs_trace,
62 msg_iter_data->pc_msg_iter,
63 msg_iter_data->msg_iter,
64 msg_iter_data->ds_file_group->stream,
65 ds_file_info->path->str);
66 if (!msg_iter_data->ds_file) {
67 ret = -1;
68 }
69
70 return ret;
71 }
72
73 static
74 void ctf_fs_msg_iter_data_destroy(
75 struct ctf_fs_msg_iter_data *msg_iter_data)
76 {
77 if (!msg_iter_data) {
78 return;
79 }
80
81 ctf_fs_ds_file_destroy(msg_iter_data->ds_file);
82
83 if (msg_iter_data->msg_iter) {
84 bt_msg_iter_destroy(msg_iter_data->msg_iter);
85 }
86
87 g_free(msg_iter_data);
88 }
89
90 static
91 enum bt_self_message_iterator_status ctf_fs_iterator_next_one(
92 struct ctf_fs_msg_iter_data *msg_iter_data,
93 const bt_message **msg)
94 {
95 enum bt_self_message_iterator_status status;
96 bt_message *priv_msg;
97 int ret;
98
99 BT_ASSERT(msg_iter_data->ds_file);
100 status = ctf_fs_ds_file_next(msg_iter_data->ds_file, &priv_msg);
101 *msg = priv_msg;
102
103 if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK &&
104 bt_message_get_type(*msg) ==
105 BT_MESSAGE_TYPE_STREAM_BEGINNING) {
106 if (msg_iter_data->skip_stream_begin_msgs) {
107 /*
108 * We already emitted a
109 * BT_MESSAGE_TYPE_STREAM_BEGINNING
110 * message: skip this one, get a new one.
111 */
112 BT_MESSAGE_PUT_REF_AND_RESET(*msg);
113 status = ctf_fs_ds_file_next(msg_iter_data->ds_file,
114 &priv_msg);
115 *msg = priv_msg;
116 BT_ASSERT(status != BT_SELF_MESSAGE_ITERATOR_STATUS_END);
117 goto end;
118 } else {
119 /*
120 * First BT_MESSAGE_TYPE_STREAM_BEGINNING
121 * message: skip all following.
122 */
123 msg_iter_data->skip_stream_begin_msgs = true;
124 goto end;
125 }
126 }
127
128 if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK &&
129 bt_message_get_type(*msg) ==
130 BT_MESSAGE_TYPE_STREAM_END) {
131 msg_iter_data->ds_file_info_index++;
132
133 if (msg_iter_data->ds_file_info_index ==
134 msg_iter_data->ds_file_group->ds_file_infos->len) {
135 /*
136 * No more stream files to read: we reached the
137 * real end. Emit this
138 * BT_MESSAGE_TYPE_STREAM_END message.
139 * The next time ctf_fs_iterator_next() is
140 * called for this message iterator,
141 * ctf_fs_ds_file_next() will return
142 * BT_SELF_MESSAGE_ITERATOR_STATUS_END().
143 */
144 goto end;
145 }
146
147 BT_MESSAGE_PUT_REF_AND_RESET(*msg);
148 bt_msg_iter_reset(msg_iter_data->msg_iter);
149
150 /*
151 * Open and start reading the next stream file within
152 * our stream file group.
153 */
154 ret = msg_iter_data_set_current_ds_file(msg_iter_data);
155 if (ret) {
156 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
157 goto end;
158 }
159
160 status = ctf_fs_ds_file_next(msg_iter_data->ds_file, &priv_msg);
161 *msg = priv_msg;
162
163 /*
164 * If we get a message, we expect to get a
165 * BT_MESSAGE_TYPE_STREAM_BEGINNING message
166 * because the iterator's state machine emits one before
167 * even requesting the first block of data from the
168 * medium. Skip this message because we're not
169 * really starting a new stream here, and try getting a
170 * new message (which, if it works, is a
171 * BT_MESSAGE_TYPE_PACKET_BEGINNING one). We're sure to
172 * get at least one pair of
173 * BT_MESSAGE_TYPE_PACKET_BEGINNING and
174 * BT_MESSAGE_TYPE_PACKET_END messages in the
175 * case of a single, empty packet. We know there's at
176 * least one packet because the stream file group does
177 * not contain empty stream files.
178 */
179 BT_ASSERT(msg_iter_data->skip_stream_begin_msgs);
180
181 if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
182 BT_ASSERT(bt_message_get_type(*msg) ==
183 BT_MESSAGE_TYPE_STREAM_BEGINNING);
184 BT_MESSAGE_PUT_REF_AND_RESET(*msg);
185 status = ctf_fs_ds_file_next(msg_iter_data->ds_file,
186 &priv_msg);
187 *msg = priv_msg;
188 BT_ASSERT(status != BT_SELF_MESSAGE_ITERATOR_STATUS_END);
189 }
190 }
191
192 end:
193 return status;
194 }
195
196 BT_HIDDEN
197 enum bt_self_message_iterator_status ctf_fs_iterator_next(
198 bt_self_message_iterator *iterator,
199 bt_message_array_const msgs, uint64_t capacity,
200 uint64_t *count)
201 {
202 enum bt_self_message_iterator_status status =
203 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
204 struct ctf_fs_msg_iter_data *msg_iter_data =
205 bt_self_message_iterator_get_data(iterator);
206 uint64_t i = 0;
207
208 while (i < capacity && status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
209 status = ctf_fs_iterator_next_one(msg_iter_data, &msgs[i]);
210 if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
211 i++;
212 }
213 }
214
215 if (i > 0) {
216 /*
217 * Even if ctf_fs_iterator_next_one() returned something
218 * else than BT_SELF_MESSAGE_ITERATOR_STATUS_OK, we
219 * accumulated message objects in the output
220 * message array, so we need to return
221 * BT_SELF_MESSAGE_ITERATOR_STATUS_OK so that they are
222 * transfered to downstream. This other status occurs
223 * again the next time muxer_msg_iter_do_next() is
224 * called, possibly without any accumulated
225 * message, in which case we'll return it.
226 */
227 *count = i;
228 status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
229 }
230
231 return status;
232 }
233
234 void ctf_fs_iterator_finalize(bt_self_message_iterator *it)
235 {
236 ctf_fs_msg_iter_data_destroy(
237 bt_self_message_iterator_get_data(it));
238 }
239
240 enum bt_self_message_iterator_status ctf_fs_iterator_init(
241 bt_self_message_iterator *self_msg_iter,
242 bt_self_component_source *self_comp,
243 bt_self_component_port_output *self_port)
244 {
245 struct ctf_fs_port_data *port_data;
246 struct ctf_fs_msg_iter_data *msg_iter_data = NULL;
247 enum bt_self_message_iterator_status ret =
248 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
249 int iret;
250
251 port_data = bt_self_component_port_get_data(
252 bt_self_component_port_output_as_self_component_port(
253 self_port));
254 BT_ASSERT(port_data);
255 msg_iter_data = g_new0(struct ctf_fs_msg_iter_data, 1);
256 if (!msg_iter_data) {
257 ret = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
258 goto error;
259 }
260
261 msg_iter_data->pc_msg_iter = self_msg_iter;
262 msg_iter_data->msg_iter = bt_msg_iter_create(
263 port_data->ds_file_group->ctf_fs_trace->metadata->tc,
264 bt_common_get_page_size() * 8,
265 ctf_fs_ds_file_medops, NULL);
266 if (!msg_iter_data->msg_iter) {
267 BT_LOGE_STR("Cannot create a CTF message iterator.");
268 ret = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
269 goto error;
270 }
271
272 msg_iter_data->ds_file_group = port_data->ds_file_group;
273 iret = msg_iter_data_set_current_ds_file(msg_iter_data);
274 if (iret) {
275 ret = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
276 goto error;
277 }
278
279 bt_self_message_iterator_set_data(self_msg_iter,
280 msg_iter_data);
281 if (ret != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
282 goto error;
283 }
284
285 msg_iter_data = NULL;
286 goto end;
287
288 error:
289 bt_self_message_iterator_set_data(self_msg_iter, NULL);
290
291 end:
292 ctf_fs_msg_iter_data_destroy(msg_iter_data);
293 return ret;
294 }
295
296 static
297 void ctf_fs_destroy(struct ctf_fs_component *ctf_fs)
298 {
299 if (!ctf_fs) {
300 return;
301 }
302
303 if (ctf_fs->traces) {
304 g_ptr_array_free(ctf_fs->traces, TRUE);
305 }
306
307 if (ctf_fs->port_data) {
308 g_ptr_array_free(ctf_fs->port_data, TRUE);
309 }
310
311 g_free(ctf_fs);
312 }
313
314 BT_HIDDEN
315 void ctf_fs_trace_destroy(struct ctf_fs_trace *ctf_fs_trace)
316 {
317 if (!ctf_fs_trace) {
318 return;
319 }
320
321 if (ctf_fs_trace->ds_file_groups) {
322 g_ptr_array_free(ctf_fs_trace->ds_file_groups, TRUE);
323 }
324
325 BT_TRACE_PUT_REF_AND_RESET(ctf_fs_trace->trace);
326
327 if (ctf_fs_trace->path) {
328 g_string_free(ctf_fs_trace->path, TRUE);
329 }
330
331 if (ctf_fs_trace->name) {
332 g_string_free(ctf_fs_trace->name, TRUE);
333 }
334
335 if (ctf_fs_trace->metadata) {
336 ctf_fs_metadata_fini(ctf_fs_trace->metadata);
337 g_free(ctf_fs_trace->metadata);
338 }
339
340 g_free(ctf_fs_trace);
341 }
342
343 static
344 void ctf_fs_trace_destroy_msgier(void *data)
345 {
346 struct ctf_fs_trace *trace = data;
347 ctf_fs_trace_destroy(trace);
348 }
349
350 void ctf_fs_finalize(bt_self_component_source *component)
351 {
352 ctf_fs_destroy(bt_self_component_get_data(
353 bt_self_component_source_as_self_component(component)));
354 }
355
356 static
357 void port_data_destroy(void *data) {
358 struct ctf_fs_port_data *port_data = data;
359
360 if (!port_data) {
361 return;
362 }
363
364 g_free(port_data);
365 }
366
367 static
368 GString *get_stream_instance_unique_name(
369 struct ctf_fs_ds_file_group *ds_file_group)
370 {
371 GString *name;
372 struct ctf_fs_ds_file_info *ds_file_info;
373
374 name = g_string_new(NULL);
375 if (!name) {
376 goto end;
377 }
378
379 /*
380 * If there's more than one stream file in the stream file
381 * group, the first (earliest) stream file's path is used as
382 * the stream's unique name.
383 */
384 BT_ASSERT(ds_file_group->ds_file_infos->len > 0);
385 ds_file_info = g_ptr_array_index(ds_file_group->ds_file_infos, 0);
386 g_string_assign(name, ds_file_info->path->str);
387
388 end:
389 return name;
390 }
391
392 static
393 int create_one_port_for_trace(struct ctf_fs_component *ctf_fs,
394 struct ctf_fs_trace *ctf_fs_trace,
395 struct ctf_fs_ds_file_group *ds_file_group)
396 {
397 int ret = 0;
398 struct ctf_fs_port_data *port_data = NULL;
399 GString *port_name = NULL;
400
401 port_name = get_stream_instance_unique_name(ds_file_group);
402 if (!port_name) {
403 goto error;
404 }
405
406 BT_LOGD("Creating one port named `%s`", port_name->str);
407
408 /* Create output port for this file */
409 port_data = g_new0(struct ctf_fs_port_data, 1);
410 if (!port_data) {
411 goto error;
412 }
413
414 port_data->ctf_fs = ctf_fs;
415 port_data->ds_file_group = ds_file_group;
416 ret = bt_self_component_source_add_output_port(
417 ctf_fs->self_comp, port_name->str, port_data, NULL);
418 if (ret) {
419 goto error;
420 }
421
422 g_ptr_array_add(ctf_fs->port_data, port_data);
423 port_data = NULL;
424 goto end;
425
426 error:
427 ret = -1;
428
429 end:
430 if (port_name) {
431 g_string_free(port_name, TRUE);
432 }
433
434 port_data_destroy(port_data);
435 return ret;
436 }
437
438 static
439 int create_ports_for_trace(struct ctf_fs_component *ctf_fs,
440 struct ctf_fs_trace *ctf_fs_trace)
441 {
442 int ret = 0;
443 size_t i;
444
445 /* Create one output port for each stream file group */
446 for (i = 0; i < ctf_fs_trace->ds_file_groups->len; i++) {
447 struct ctf_fs_ds_file_group *ds_file_group =
448 g_ptr_array_index(ctf_fs_trace->ds_file_groups, i);
449
450 ret = create_one_port_for_trace(ctf_fs, ctf_fs_trace,
451 ds_file_group);
452 if (ret) {
453 BT_LOGE("Cannot create output port.");
454 goto end;
455 }
456 }
457
458 end:
459 return ret;
460 }
461
462 static
463 void ctf_fs_ds_file_info_destroy(struct ctf_fs_ds_file_info *ds_file_info)
464 {
465 if (!ds_file_info) {
466 return;
467 }
468
469 if (ds_file_info->path) {
470 g_string_free(ds_file_info->path, TRUE);
471 }
472
473 ctf_fs_ds_index_destroy(ds_file_info->index);
474 g_free(ds_file_info);
475 }
476
477 static
478 struct ctf_fs_ds_file_info *ctf_fs_ds_file_info_create(const char *path,
479 int64_t begin_ns, struct ctf_fs_ds_index *index)
480 {
481 struct ctf_fs_ds_file_info *ds_file_info;
482
483 ds_file_info = g_new0(struct ctf_fs_ds_file_info, 1);
484 if (!ds_file_info) {
485 goto end;
486 }
487
488 ds_file_info->path = g_string_new(path);
489 if (!ds_file_info->path) {
490 ctf_fs_ds_file_info_destroy(ds_file_info);
491 ds_file_info = NULL;
492 goto end;
493 }
494
495 ds_file_info->begin_ns = begin_ns;
496 ds_file_info->index = index;
497 index = NULL;
498
499 end:
500 ctf_fs_ds_index_destroy(index);
501 return ds_file_info;
502 }
503
504 static
505 void ctf_fs_ds_file_group_destroy(struct ctf_fs_ds_file_group *ds_file_group)
506 {
507 if (!ds_file_group) {
508 return;
509 }
510
511 if (ds_file_group->ds_file_infos) {
512 g_ptr_array_free(ds_file_group->ds_file_infos, TRUE);
513 }
514
515 bt_stream_put_ref(ds_file_group->stream);
516 bt_stream_class_put_ref(ds_file_group->stream_class);
517 g_free(ds_file_group);
518 }
519
520 static
521 struct ctf_fs_ds_file_group *ctf_fs_ds_file_group_create(
522 struct ctf_fs_trace *ctf_fs_trace,
523 bt_stream_class *stream_class,
524 uint64_t stream_instance_id)
525 {
526 struct ctf_fs_ds_file_group *ds_file_group;
527
528 ds_file_group = g_new0(struct ctf_fs_ds_file_group, 1);
529 if (!ds_file_group) {
530 goto error;
531 }
532
533 ds_file_group->ds_file_infos = g_ptr_array_new_with_free_func(
534 (GDestroyNotify) ctf_fs_ds_file_info_destroy);
535 if (!ds_file_group->ds_file_infos) {
536 goto error;
537 }
538
539 ds_file_group->stream_id = stream_instance_id;
540 BT_ASSERT(stream_class);
541 ds_file_group->stream_class = stream_class;
542 bt_stream_class_get_ref(ds_file_group->stream_class);
543 ds_file_group->ctf_fs_trace = ctf_fs_trace;
544 goto end;
545
546 error:
547 ctf_fs_ds_file_group_destroy(ds_file_group);
548 ds_file_group = NULL;
549
550 end:
551 return ds_file_group;
552 }
553
554 /* Replace by g_ptr_array_insert when we depend on glib >= 2.40. */
555 static
556 void array_insert(GPtrArray *array, gpointer element, size_t pos)
557 {
558 size_t original_array_len = array->len;
559
560 /* Allocate an unused element at the end of the array. */
561 g_ptr_array_add(array, NULL);
562
563 /* If we are not inserting at the end, move the elements by one. */
564 if (pos < original_array_len) {
565 memmove(&(array->pdata[pos + 1]),
566 &(array->pdata[pos]),
567 (original_array_len - pos) * sizeof(gpointer));
568 }
569
570 /* Insert the value and bump the array len */
571 array->pdata[pos] = element;
572 }
573
574 static
575 int ctf_fs_ds_file_group_add_ds_file_info(
576 struct ctf_fs_ds_file_group *ds_file_group,
577 const char *path, int64_t begin_ns,
578 struct ctf_fs_ds_index *index)
579 {
580 struct ctf_fs_ds_file_info *ds_file_info;
581 gint i = 0;
582 int ret = 0;
583
584 /* Onwership of index is transferred. */
585 ds_file_info = ctf_fs_ds_file_info_create(path, begin_ns, index);
586 index = NULL;
587 if (!ds_file_info) {
588 goto error;
589 }
590
591 /* Find a spot to insert this one */
592 for (i = 0; i < ds_file_group->ds_file_infos->len; i++) {
593 struct ctf_fs_ds_file_info *other_ds_file_info =
594 g_ptr_array_index(ds_file_group->ds_file_infos, i);
595
596 if (begin_ns < other_ds_file_info->begin_ns) {
597 break;
598 }
599 }
600
601 array_insert(ds_file_group->ds_file_infos, ds_file_info, i);
602 ds_file_info = NULL;
603 goto end;
604
605 error:
606 ctf_fs_ds_file_info_destroy(ds_file_info);
607 ctf_fs_ds_index_destroy(index);
608 ret = -1;
609 end:
610 return ret;
611 }
612
613 static
614 int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
615 const char *path)
616 {
617 bt_stream_class *stream_class = NULL;
618 int64_t stream_instance_id = -1;
619 int64_t begin_ns = -1;
620 struct ctf_fs_ds_file_group *ds_file_group = NULL;
621 bool add_group = false;
622 int ret;
623 size_t i;
624 struct ctf_fs_ds_file *ds_file = NULL;
625 struct ctf_fs_ds_index *index = NULL;
626 struct bt_msg_iter *msg_iter = NULL;
627 struct ctf_stream_class *sc = NULL;
628 struct bt_msg_iter_packet_properties props;
629
630 msg_iter = bt_msg_iter_create(ctf_fs_trace->metadata->tc,
631 bt_common_get_page_size() * 8, ctf_fs_ds_file_medops, NULL);
632 if (!msg_iter) {
633 BT_LOGE_STR("Cannot create a CTF message iterator.");
634 goto error;
635 }
636
637 ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, msg_iter,
638 NULL, path);
639 if (!ds_file) {
640 goto error;
641 }
642
643 ret = ctf_fs_ds_file_borrow_packet_header_context_fields(ds_file,
644 NULL, NULL);
645 if (ret) {
646 BT_LOGE("Cannot get stream file's first packet's header and context fields (`%s`).",
647 path);
648 goto error;
649 }
650
651 ret = bt_msg_iter_get_packet_properties(ds_file->msg_iter, &props);
652 BT_ASSERT(ret == 0);
653 sc = ctf_trace_class_borrow_stream_class_by_id(ds_file->metadata->tc,
654 props.stream_class_id);
655 BT_ASSERT(sc);
656 stream_class = sc->ir_sc;
657 BT_ASSERT(stream_class);
658 stream_instance_id = props.data_stream_id;
659
660 if (props.snapshots.beginning_clock != UINT64_C(-1)) {
661 BT_ASSERT(sc->default_clock_class);
662 ret = bt_clock_class_cycles_to_ns_from_origin(
663 sc->default_clock_class,
664 props.snapshots.beginning_clock, &begin_ns);
665 if (ret) {
666 BT_LOGE("Cannot convert clock cycles to nanoseconds from origin (`%s`).",
667 path);
668 goto error;
669 }
670 }
671
672 index = ctf_fs_ds_file_build_index(ds_file);
673 if (!index) {
674 BT_LOGW("Failed to index CTF stream file \'%s\'",
675 ds_file->file->path->str);
676 }
677
678 if (begin_ns == -1) {
679 /*
680 * No beggining timestamp to sort the stream files
681 * within a stream file group, so consider that this
682 * file must be the only one within its group.
683 */
684 stream_instance_id = -1;
685 }
686
687 if (stream_instance_id == -1) {
688 /*
689 * No stream instance ID or no beginning timestamp:
690 * create a unique stream file group for this stream
691 * file because, even if there's a stream instance ID,
692 * there's no timestamp to order the file within its
693 * group.
694 */
695 ds_file_group = ctf_fs_ds_file_group_create(ctf_fs_trace,
696 stream_class, stream_instance_id);
697 if (!ds_file_group) {
698 goto error;
699 }
700
701 ret = ctf_fs_ds_file_group_add_ds_file_info(ds_file_group,
702 path, begin_ns, index);
703 /* Ownership of index is transferred. */
704 index = NULL;
705 if (ret) {
706 goto error;
707 }
708
709 add_group = true;
710 goto end;
711 }
712
713 BT_ASSERT(stream_instance_id != -1);
714 BT_ASSERT(begin_ns != -1);
715
716 /* Find an existing stream file group with this ID */
717 for (i = 0; i < ctf_fs_trace->ds_file_groups->len; i++) {
718 ds_file_group = g_ptr_array_index(
719 ctf_fs_trace->ds_file_groups, i);
720
721 if (ds_file_group->stream_class == stream_class &&
722 ds_file_group->stream_id ==
723 stream_instance_id) {
724 break;
725 }
726
727 ds_file_group = NULL;
728 }
729
730 if (!ds_file_group) {
731 ds_file_group = ctf_fs_ds_file_group_create(ctf_fs_trace,
732 stream_class, stream_instance_id);
733 if (!ds_file_group) {
734 goto error;
735 }
736
737 add_group = true;
738 }
739
740 ret = ctf_fs_ds_file_group_add_ds_file_info(ds_file_group, path,
741 begin_ns, index);
742 index = NULL;
743 if (ret) {
744 goto error;
745 }
746
747 goto end;
748
749 error:
750 ctf_fs_ds_file_group_destroy(ds_file_group);
751 ret = -1;
752
753 end:
754 if (add_group && ds_file_group) {
755 g_ptr_array_add(ctf_fs_trace->ds_file_groups, ds_file_group);
756 }
757
758 ctf_fs_ds_file_destroy(ds_file);
759
760 if (msg_iter) {
761 bt_msg_iter_destroy(msg_iter);
762 }
763
764 ctf_fs_ds_index_destroy(index);
765 return ret;
766 }
767
768 static
769 int create_ds_file_groups(struct ctf_fs_trace *ctf_fs_trace)
770 {
771 int ret = 0;
772 const char *basename;
773 GError *error = NULL;
774 GDir *dir = NULL;
775 size_t i;
776
777 /* Check each file in the path directory, except specific ones */
778 dir = g_dir_open(ctf_fs_trace->path->str, 0, &error);
779 if (!dir) {
780 BT_LOGE("Cannot open directory `%s`: %s (code %d)",
781 ctf_fs_trace->path->str, error->message,
782 error->code);
783 goto error;
784 }
785
786 while ((basename = g_dir_read_name(dir))) {
787 struct ctf_fs_file *file;
788
789 if (!strcmp(basename, CTF_FS_METADATA_FILENAME)) {
790 /* Ignore the metadata stream. */
791 BT_LOGD("Ignoring metadata file `%s" G_DIR_SEPARATOR_S "%s`",
792 ctf_fs_trace->path->str, basename);
793 continue;
794 }
795
796 if (basename[0] == '.') {
797 BT_LOGD("Ignoring hidden file `%s" G_DIR_SEPARATOR_S "%s`",
798 ctf_fs_trace->path->str, basename);
799 continue;
800 }
801
802 /* Create the file. */
803 file = ctf_fs_file_create();
804 if (!file) {
805 BT_LOGE("Cannot create stream file object for file `%s" G_DIR_SEPARATOR_S "%s`",
806 ctf_fs_trace->path->str, basename);
807 goto error;
808 }
809
810 /* Create full path string. */
811 g_string_append_printf(file->path, "%s" G_DIR_SEPARATOR_S "%s",
812 ctf_fs_trace->path->str, basename);
813 if (!g_file_test(file->path->str, G_FILE_TEST_IS_REGULAR)) {
814 BT_LOGD("Ignoring non-regular file `%s`",
815 file->path->str);
816 ctf_fs_file_destroy(file);
817 file = NULL;
818 continue;
819 }
820
821 ret = ctf_fs_file_open(file, "rb");
822 if (ret) {
823 BT_LOGE("Cannot open stream file `%s`", file->path->str);
824 goto error;
825 }
826
827 if (file->size == 0) {
828 /* Skip empty stream. */
829 BT_LOGD("Ignoring empty file `%s`", file->path->str);
830 ctf_fs_file_destroy(file);
831 continue;
832 }
833
834 ret = add_ds_file_to_ds_file_group(ctf_fs_trace,
835 file->path->str);
836 if (ret) {
837 BT_LOGE("Cannot add stream file `%s` to stream file group",
838 file->path->str);
839 ctf_fs_file_destroy(file);
840 goto error;
841 }
842
843 ctf_fs_file_destroy(file);
844 }
845
846 /*
847 * At this point, DS file groupes are created, but their
848 * associated stream objects do not exist yet. This is because
849 * we need to name the created stream object with the data
850 * stream file's path. We have everything we need here to do
851 * this.
852 */
853 for (i = 0; i < ctf_fs_trace->ds_file_groups->len; i++) {
854 struct ctf_fs_ds_file_group *ds_file_group =
855 g_ptr_array_index(ctf_fs_trace->ds_file_groups, i);
856 GString *name = get_stream_instance_unique_name(ds_file_group);
857
858 if (!name) {
859 goto error;
860 }
861
862 if (ds_file_group->stream_id == UINT64_C(-1)) {
863 /* No stream ID: use 0 */
864 ds_file_group->stream = bt_stream_create_with_id(
865 ds_file_group->stream_class,
866 ctf_fs_trace->trace,
867 ctf_fs_trace->next_stream_id);
868 ctf_fs_trace->next_stream_id++;
869 } else {
870 /* Specific stream ID */
871 ds_file_group->stream = bt_stream_create_with_id(
872 ds_file_group->stream_class,
873 ctf_fs_trace->trace,
874 (uint64_t) ds_file_group->stream_id);
875 }
876
877 if (!ds_file_group->stream) {
878 BT_LOGE("Cannot create stream for DS file group: "
879 "addr=%p, stream-name=\"%s\"",
880 ds_file_group, name->str);
881 g_string_free(name, TRUE);
882 goto error;
883 }
884
885 ret = bt_stream_set_name(ds_file_group->stream,
886 name->str);
887 if (ret) {
888 BT_LOGE("Cannot set stream's name: "
889 "addr=%p, stream-name=\"%s\"",
890 ds_file_group->stream, name->str);
891 g_string_free(name, TRUE);
892 goto error;
893 }
894
895 g_string_free(name, TRUE);
896 }
897
898 goto end;
899
900 error:
901 ret = -1;
902
903 end:
904 if (dir) {
905 g_dir_close(dir);
906 dir = NULL;
907 }
908
909 if (error) {
910 g_error_free(error);
911 }
912
913 return ret;
914 }
915
916 static
917 int set_trace_name(bt_trace *trace, const char *name_suffix)
918 {
919 int ret = 0;
920 const bt_trace_class *tc = bt_trace_borrow_class_const(trace);
921 const bt_value *val;
922 GString *name;
923
924 name = g_string_new(NULL);
925 if (!name) {
926 BT_LOGE_STR("Failed to allocate a GString.");
927 ret = -1;
928 goto end;
929 }
930
931 /*
932 * Check if we have a trace environment string value named `hostname`.
933 * If so, use it as the trace name's prefix.
934 */
935 val = bt_trace_class_borrow_environment_entry_value_by_name_const(
936 tc, "hostname");
937 if (val && bt_value_is_string(val)) {
938 g_string_append(name, bt_value_string_get(val));
939
940 if (name_suffix) {
941 g_string_append_c(name, G_DIR_SEPARATOR);
942 }
943 }
944
945 if (name_suffix) {
946 g_string_append(name, name_suffix);
947 }
948
949 ret = bt_trace_set_name(trace, name->str);
950 if (ret) {
951 goto end;
952 }
953
954 goto end;
955
956 end:
957 if (name) {
958 g_string_free(name, TRUE);
959 }
960
961 return ret;
962 }
963
964 BT_HIDDEN
965 struct ctf_fs_trace *ctf_fs_trace_create(const char *path, const char *name,
966 struct ctf_fs_metadata_config *metadata_config)
967 {
968 struct ctf_fs_trace *ctf_fs_trace;
969 int ret;
970
971 ctf_fs_trace = g_new0(struct ctf_fs_trace, 1);
972 if (!ctf_fs_trace) {
973 goto end;
974 }
975
976 ctf_fs_trace->path = g_string_new(path);
977 if (!ctf_fs_trace->path) {
978 goto error;
979 }
980
981 ctf_fs_trace->name = g_string_new(name);
982 if (!ctf_fs_trace->name) {
983 goto error;
984 }
985
986 ctf_fs_trace->metadata = g_new0(struct ctf_fs_metadata, 1);
987 if (!ctf_fs_trace->metadata) {
988 goto error;
989 }
990
991 ctf_fs_metadata_init(ctf_fs_trace->metadata);
992 ctf_fs_trace->ds_file_groups = g_ptr_array_new_with_free_func(
993 (GDestroyNotify) ctf_fs_ds_file_group_destroy);
994 if (!ctf_fs_trace->ds_file_groups) {
995 goto error;
996 }
997
998 ret = ctf_fs_metadata_set_trace_class(ctf_fs_trace, metadata_config);
999 if (ret) {
1000 goto error;
1001 }
1002
1003 ctf_fs_trace->trace =
1004 bt_trace_create(ctf_fs_trace->metadata->trace_class);
1005 if (!ctf_fs_trace->trace) {
1006 goto error;
1007 }
1008
1009 ret = set_trace_name(ctf_fs_trace->trace, name);
1010 if (ret) {
1011 goto error;
1012 }
1013
1014 ret = create_ds_file_groups(ctf_fs_trace);
1015 if (ret) {
1016 goto error;
1017 }
1018
1019 /*
1020 * create_ds_file_groups() created all the streams that this
1021 * trace needs. There won't be any more. Therefore it is safe to
1022 * make this trace static.
1023 */
1024 (void) bt_trace_make_static(ctf_fs_trace->trace);
1025
1026 goto end;
1027
1028 error:
1029 ctf_fs_trace_destroy(ctf_fs_trace);
1030 ctf_fs_trace = NULL;
1031
1032 end:
1033 return ctf_fs_trace;
1034 }
1035
1036 static
1037 int path_is_ctf_trace(const char *path)
1038 {
1039 GString *metadata_path = g_string_new(NULL);
1040 int ret = 0;
1041
1042 if (!metadata_path) {
1043 ret = -1;
1044 goto end;
1045 }
1046
1047 g_string_printf(metadata_path, "%s" G_DIR_SEPARATOR_S "%s", path, CTF_FS_METADATA_FILENAME);
1048
1049 if (g_file_test(metadata_path->str, G_FILE_TEST_IS_REGULAR)) {
1050 ret = 1;
1051 goto end;
1052 }
1053
1054 end:
1055 g_string_free(metadata_path, TRUE);
1056 return ret;
1057 }
1058
1059 static
1060 int add_trace_path(GList **trace_paths, const char *path)
1061 {
1062 GString *norm_path = NULL;
1063 int ret = 0;
1064
1065 norm_path = bt_common_normalize_path(path, NULL);
1066 if (!norm_path) {
1067 BT_LOGE("Failed to normalize path `%s`.", path);
1068 ret = -1;
1069 goto end;
1070 }
1071
1072 // FIXME: Remove or ifdef for __MINGW32__
1073 if (strcmp(norm_path->str, "/") == 0) {
1074 BT_LOGE("Opening a trace in `/` is not supported.");
1075 ret = -1;
1076 goto end;
1077 }
1078
1079 *trace_paths = g_list_prepend(*trace_paths, norm_path);
1080 BT_ASSERT(*trace_paths);
1081 norm_path = NULL;
1082
1083 end:
1084 if (norm_path) {
1085 g_string_free(norm_path, TRUE);
1086 }
1087
1088 return ret;
1089 }
1090
1091 BT_HIDDEN
1092 int ctf_fs_find_traces(GList **trace_paths, const char *start_path)
1093 {
1094 int ret;
1095 GError *error = NULL;
1096 GDir *dir = NULL;
1097 const char *basename = NULL;
1098
1099 /* Check if the starting path is a CTF trace itself */
1100 ret = path_is_ctf_trace(start_path);
1101 if (ret < 0) {
1102 goto end;
1103 }
1104
1105 if (ret) {
1106 /*
1107 * Stop recursion: a CTF trace cannot contain another
1108 * CTF trace.
1109 */
1110 ret = add_trace_path(trace_paths, start_path);
1111 goto end;
1112 }
1113
1114 /* Look for subdirectories */
1115 if (!g_file_test(start_path, G_FILE_TEST_IS_DIR)) {
1116 /* Starting path is not a directory: end of recursion */
1117 goto end;
1118 }
1119
1120 dir = g_dir_open(start_path, 0, &error);
1121 if (!dir) {
1122 if (error->code == G_FILE_ERROR_ACCES) {
1123 BT_LOGD("Cannot open directory `%s`: %s (code %d): continuing",
1124 start_path, error->message, error->code);
1125 goto end;
1126 }
1127
1128 BT_LOGE("Cannot open directory `%s`: %s (code %d)",
1129 start_path, error->message, error->code);
1130 ret = -1;
1131 goto end;
1132 }
1133
1134 while ((basename = g_dir_read_name(dir))) {
1135 GString *sub_path = g_string_new(NULL);
1136
1137 if (!sub_path) {
1138 ret = -1;
1139 goto end;
1140 }
1141
1142 g_string_printf(sub_path, "%s" G_DIR_SEPARATOR_S "%s", start_path, basename);
1143 ret = ctf_fs_find_traces(trace_paths, sub_path->str);
1144 g_string_free(sub_path, TRUE);
1145 if (ret) {
1146 goto end;
1147 }
1148 }
1149
1150 end:
1151 if (dir) {
1152 g_dir_close(dir);
1153 }
1154
1155 if (error) {
1156 g_error_free(error);
1157 }
1158
1159 return ret;
1160 }
1161
1162 BT_HIDDEN
1163 GList *ctf_fs_create_trace_names(GList *trace_paths, const char *base_path) {
1164 GList *trace_names = NULL;
1165 GList *node;
1166 const char *last_sep;
1167 size_t base_dist;
1168
1169 /*
1170 * At this point we know that all the trace paths are
1171 * normalized, and so is the base path. This means that
1172 * they are absolute and they don't end with a separator.
1173 * We can simply find the location of the last separator
1174 * in the base path, which gives us the name of the actual
1175 * directory to look into, and use this location as the
1176 * start of each trace name within each trace path.
1177 *
1178 * For example:
1179 *
1180 * Base path: /home/user/my-traces/some-trace
1181 * Trace paths:
1182 * - /home/user/my-traces/some-trace/host1/trace1
1183 * - /home/user/my-traces/some-trace/host1/trace2
1184 * - /home/user/my-traces/some-trace/host2/trace
1185 * - /home/user/my-traces/some-trace/other-trace
1186 *
1187 * In this case the trace names are:
1188 *
1189 * - some-trace/host1/trace1
1190 * - some-trace/host1/trace2
1191 * - some-trace/host2/trace
1192 * - some-trace/other-trace
1193 */
1194 last_sep = strrchr(base_path, G_DIR_SEPARATOR);
1195
1196 /* We know there's at least one separator */
1197 BT_ASSERT(last_sep);
1198
1199 /* Distance to base */
1200 base_dist = last_sep - base_path + 1;
1201
1202 /* Create the trace names */
1203 for (node = trace_paths; node; node = g_list_next(node)) {
1204 GString *trace_name = g_string_new(NULL);
1205 GString *trace_path = node->data;
1206
1207 BT_ASSERT(trace_name);
1208 g_string_assign(trace_name, &trace_path->str[base_dist]);
1209 trace_names = g_list_append(trace_names, trace_name);
1210 }
1211
1212 return trace_names;
1213 }
1214
1215 static
1216 int create_ctf_fs_traces(struct ctf_fs_component *ctf_fs,
1217 const char *path_param)
1218 {
1219 struct ctf_fs_trace *ctf_fs_trace = NULL;
1220 int ret = 0;
1221 GString *norm_path = NULL;
1222 GList *trace_paths = NULL;
1223 GList *trace_names = NULL;
1224 GList *tp_node;
1225 GList *tn_node;
1226
1227 norm_path = bt_common_normalize_path(path_param, NULL);
1228 if (!norm_path) {
1229 BT_LOGE("Failed to normalize path: `%s`.",
1230 path_param);
1231 goto error;
1232 }
1233
1234 ret = ctf_fs_find_traces(&trace_paths, norm_path->str);
1235 if (ret) {
1236 goto error;
1237 }
1238
1239 if (!trace_paths) {
1240 BT_LOGE("No CTF traces recursively found in `%s`.",
1241 path_param);
1242 goto error;
1243 }
1244
1245 trace_names = ctf_fs_create_trace_names(trace_paths, norm_path->str);
1246 if (!trace_names) {
1247 BT_LOGE("Cannot create trace names from trace paths.");
1248 goto error;
1249 }
1250
1251 for (tp_node = trace_paths, tn_node = trace_names; tp_node;
1252 tp_node = g_list_next(tp_node),
1253 tn_node = g_list_next(tn_node)) {
1254 GString *trace_path = tp_node->data;
1255 GString *trace_name = tn_node->data;
1256
1257 ctf_fs_trace = ctf_fs_trace_create(trace_path->str,
1258 trace_name->str, &ctf_fs->metadata_config);
1259 if (!ctf_fs_trace) {
1260 BT_LOGE("Cannot create trace for `%s`.",
1261 trace_path->str);
1262 goto error;
1263 }
1264
1265 ret = create_ports_for_trace(ctf_fs, ctf_fs_trace);
1266 if (ret) {
1267 goto error;
1268 }
1269
1270 g_ptr_array_add(ctf_fs->traces, ctf_fs_trace);
1271 ctf_fs_trace = NULL;
1272 }
1273
1274 goto end;
1275
1276 error:
1277 ret = -1;
1278 ctf_fs_trace_destroy(ctf_fs_trace);
1279
1280 end:
1281 for (tp_node = trace_paths; tp_node; tp_node = g_list_next(tp_node)) {
1282 if (tp_node->data) {
1283 g_string_free(tp_node->data, TRUE);
1284 }
1285 }
1286
1287 for (tn_node = trace_names; tn_node; tn_node = g_list_next(tn_node)) {
1288 if (tn_node->data) {
1289 g_string_free(tn_node->data, TRUE);
1290 }
1291 }
1292
1293 if (trace_paths) {
1294 g_list_free(trace_paths);
1295 }
1296
1297 if (trace_names) {
1298 g_list_free(trace_names);
1299 }
1300
1301 if (norm_path) {
1302 g_string_free(norm_path, TRUE);
1303 }
1304
1305 return ret;
1306 }
1307
1308 static
1309 struct ctf_fs_component *ctf_fs_create(
1310 bt_self_component_source *self_comp,
1311 const bt_value *params)
1312 {
1313 struct ctf_fs_component *ctf_fs;
1314 const bt_value *value = NULL;
1315 const char *path_param;
1316
1317 ctf_fs = g_new0(struct ctf_fs_component, 1);
1318 if (!ctf_fs) {
1319 goto end;
1320 }
1321
1322 bt_self_component_set_data(
1323 bt_self_component_source_as_self_component(self_comp),
1324 ctf_fs);
1325
1326 /*
1327 * We don't need to get a new reference here because as long as
1328 * our private ctf_fs_component object exists, the containing
1329 * private component should also exist.
1330 */
1331 ctf_fs->self_comp = self_comp;
1332 value = bt_value_map_borrow_entry_value_const(params, "path");
1333 if (value && !bt_value_is_string(value)) {
1334 goto error;
1335 }
1336
1337 path_param = bt_value_string_get(value);
1338 value = bt_value_map_borrow_entry_value_const(params,
1339 "clock-class-offset-s");
1340 if (value) {
1341 if (!bt_value_is_integer(value)) {
1342 BT_LOGE("clock-class-offset-s should be an integer");
1343 goto error;
1344 }
1345 ctf_fs->metadata_config.clock_class_offset_s = bt_value_integer_get(value);
1346 }
1347
1348 value = bt_value_map_borrow_entry_value_const(params,
1349 "clock-class-offset-ns");
1350 if (value) {
1351 if (!bt_value_is_integer(value)) {
1352 BT_LOGE("clock-class-offset-ns should be an integer");
1353 goto error;
1354 }
1355 ctf_fs->metadata_config.clock_class_offset_ns = bt_value_integer_get(value);
1356 }
1357
1358 ctf_fs->port_data = g_ptr_array_new_with_free_func(port_data_destroy);
1359 if (!ctf_fs->port_data) {
1360 goto error;
1361 }
1362
1363 ctf_fs->traces = g_ptr_array_new_with_free_func(
1364 ctf_fs_trace_destroy_msgier);
1365 if (!ctf_fs->traces) {
1366 goto error;
1367 }
1368
1369 if (create_ctf_fs_traces(ctf_fs, path_param)) {
1370 goto error;
1371 }
1372
1373 goto end;
1374
1375 error:
1376 ctf_fs_destroy(ctf_fs);
1377 ctf_fs = NULL;
1378 bt_self_component_set_data(
1379 bt_self_component_source_as_self_component(self_comp),
1380 NULL);
1381
1382 end:
1383 return ctf_fs;
1384 }
1385
1386 BT_HIDDEN
1387 enum bt_self_component_status ctf_fs_init(
1388 bt_self_component_source *self_comp,
1389 const bt_value *params, UNUSED_VAR void *init_method_data)
1390 {
1391 struct ctf_fs_component *ctf_fs;
1392 enum bt_self_component_status ret = BT_SELF_COMPONENT_STATUS_OK;
1393
1394 ctf_fs = ctf_fs_create(self_comp, params);
1395 if (!ctf_fs) {
1396 ret = BT_SELF_COMPONENT_STATUS_ERROR;
1397 }
1398
1399 return ret;
1400 }
1401
1402 BT_HIDDEN
1403 enum bt_query_status ctf_fs_query(
1404 bt_self_component_class_source *comp_class,
1405 const bt_query_executor *query_exec,
1406 const char *object, const bt_value *params,
1407 const bt_value **result)
1408 {
1409 enum bt_query_status status = BT_QUERY_STATUS_OK;
1410
1411 if (!strcmp(object, "metadata-info")) {
1412 status = metadata_info_query(comp_class, params, result);
1413 } else if (!strcmp(object, "trace-info")) {
1414 status = trace_info_query(comp_class, params, result);
1415 } else {
1416 BT_LOGE("Unknown query object `%s`", object);
1417 status = BT_QUERY_STATUS_INVALID_OBJECT;
1418 goto end;
1419 }
1420 end:
1421 return status;
1422 }
This page took 0.062056 seconds and 4 git commands to generate.