src/plugins/ctf/fs-sink/fs-sink.c

   1 /*
   2  * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
   3  *
   4  * Permission is hereby granted, free of charge, to any person obtaining a copy
   5  * of this software and associated documentation files (the "Software"), to deal
   6  * in the Software without restriction, including without limitation the rights
   7  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   8  * copies of the Software, and to permit persons to whom the Software is
   9  * furnished to do so, subject to the following conditions:
  10  *
  11  * The above copyright notice and this permission notice shall be included in
  12  * all copies or substantial portions of the Software.
  13  *
  14  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  19  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  20  * SOFTWARE.
  21  */
  22
  23 #define BT_COMP_LOG_SELF_COMP (fs_sink->self_comp)
  24 #define BT_LOG_OUTPUT_LEVEL (fs_sink->log_level)
  25 #define BT_LOG_TAG "PLUGIN/SINK.CTF.FS"
  26 #include "plugins/comp-logging.h"
  27
  28 #include <babeltrace2/babeltrace.h>
  29 #include <stdio.h>
  30 #include <stdbool.h>
  31 #include <glib.h>
  32 #include "common/assert.h"
  33 #include "ctfser/ctfser.h"
  34
  35 #include "fs-sink.h"
  36 #include "fs-sink-trace.h"
  37 #include "fs-sink-stream.h"
  38 #include "fs-sink-ctf-meta.h"
  39 #include "translate-trace-ir-to-ctf-ir.h"
  40 #include "translate-ctf-ir-to-tsdl.h"
  41
  42 static
  43 const char * const in_port_name = "in";
  44
  45 static
  46 bt_self_component_status ensure_output_dir_exists(
  47                 struct fs_sink_comp *fs_sink)
  48 {
  49         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
  50         int ret;
  51
  52         ret = g_mkdir_with_parents(fs_sink->output_dir_path->str, 0755);
  53         if (ret) {
  54                 BT_COMP_LOGE_ERRNO(
  55                         "Cannot create directories for output directory",
  56                         ": output-dir-path=\"%s\"",
  57                         fs_sink->output_dir_path->str);
  58                 status = BT_SELF_COMPONENT_STATUS_ERROR;
  59                 goto end;
  60         }
  61
  62 end:
  63         return status;
  64 }
  65
  66 static
  67 bt_self_component_status configure_component(struct fs_sink_comp *fs_sink,
  68                 const bt_value *params)
  69 {
  70         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
  71         const bt_value *value;
  72
  73         value = bt_value_map_borrow_entry_value_const(params, "path");
  74         if (!value) {
  75                 BT_COMP_LOGE_STR("Missing mandatory `path` parameter.");
  76                 status = BT_SELF_COMPONENT_STATUS_ERROR;
  77                 goto end;
  78         }
  79
  80         if (!bt_value_is_string(value)) {
  81                 BT_COMP_LOGE_STR("`path` parameter: expecting a string.");
  82                 status = BT_SELF_COMPONENT_STATUS_ERROR;
  83                 goto end;
  84         }
  85
  86         g_string_assign(fs_sink->output_dir_path,
  87                 bt_value_string_get(value));
  88         value = bt_value_map_borrow_entry_value_const(params,
  89                 "assume-single-trace");
  90         if (value) {
  91                 if (!bt_value_is_bool(value)) {
  92                         BT_COMP_LOGE_STR("`assume-single-trace` parameter: expecting a boolean.");
  93                         status = BT_SELF_COMPONENT_STATUS_ERROR;
  94                         goto end;
  95                 }
  96
  97                 fs_sink->assume_single_trace = (bool) bt_value_bool_get(value);
  98         }
  99
 100         value = bt_value_map_borrow_entry_value_const(params,
 101                 "ignore-discarded-events");
 102         if (value) {
 103                 if (!bt_value_is_bool(value)) {
 104                         BT_COMP_LOGE_STR("`ignore-discarded-events` parameter: expecting a boolean.");
 105                         status = BT_SELF_COMPONENT_STATUS_ERROR;
 106                         goto end;
 107                 }
 108
 109                 fs_sink->ignore_discarded_events =
 110                         (bool) bt_value_bool_get(value);
 111         }
 112
 113         value = bt_value_map_borrow_entry_value_const(params,
 114                 "ignore-discarded-packets");
 115         if (value) {
 116                 if (!bt_value_is_bool(value)) {
 117                         BT_COMP_LOGE_STR("`ignore-discarded-packets` parameter: expecting a boolean.");
 118                         status = BT_SELF_COMPONENT_STATUS_ERROR;
 119                         goto end;
 120                 }
 121
 122                 fs_sink->ignore_discarded_packets =
 123                         (bool) bt_value_bool_get(value);
 124         }
 125
 126         value = bt_value_map_borrow_entry_value_const(params,
 127                 "quiet");
 128         if (value) {
 129                 if (!bt_value_is_bool(value)) {
 130                         BT_COMP_LOGE_STR("`quiet` parameter: expecting a boolean.");
 131                         status = BT_SELF_COMPONENT_STATUS_ERROR;
 132                         goto end;
 133                 }
 134
 135                 fs_sink->quiet = (bool) bt_value_bool_get(value);
 136         }
 137
 138 end:
 139         return status;
 140 }
 141
 142 static
 143 void destroy_fs_sink_comp(struct fs_sink_comp *fs_sink)
 144 {
 145         if (!fs_sink) {
 146                 goto end;
 147         }
 148
 149         if (fs_sink->output_dir_path) {
 150                 g_string_free(fs_sink->output_dir_path, TRUE);
 151                 fs_sink->output_dir_path = NULL;
 152         }
 153
 154         if (fs_sink->traces) {
 155                 g_hash_table_destroy(fs_sink->traces);
 156                 fs_sink->traces = NULL;
 157         }
 158
 159         BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_PUT_REF_AND_RESET(
 160                 fs_sink->upstream_iter);
 161         g_free(fs_sink);
 162
 163 end:
 164         return;
 165 }
 166
 167 BT_HIDDEN
 168 bt_self_component_status ctf_fs_sink_init(
 169                 bt_self_component_sink *self_comp_sink, const bt_value *params,
 170                 void *init_method_data)
 171 {
 172         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 173         struct fs_sink_comp *fs_sink = NULL;
 174         bt_self_component *self_comp =
 175                 bt_self_component_sink_as_self_component(self_comp_sink);
 176         bt_logging_level log_level = bt_component_get_logging_level(
 177                 bt_self_component_as_component(self_comp));
 178
 179         fs_sink = g_new0(struct fs_sink_comp, 1);
 180         if (!fs_sink) {
 181                 BT_COMP_LOG_CUR_LVL(BT_LOG_ERROR, log_level, self_comp,
 182                         "Failed to allocate one CTF FS sink structure.");
 183                 status = BT_SELF_COMPONENT_STATUS_NOMEM;
 184                 goto end;
 185         }
 186
 187         fs_sink->log_level = log_level;
 188         fs_sink->self_comp = self_comp;
 189         fs_sink->output_dir_path = g_string_new(NULL);
 190         status = configure_component(fs_sink, params);
 191         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 192                 /* configure_component() logs errors */
 193                 goto end;
 194         }
 195
 196         if (fs_sink->assume_single_trace &&
 197                         g_file_test(fs_sink->output_dir_path->str,
 198                                 G_FILE_TEST_EXISTS)) {
 199                 BT_COMP_LOGE("Single trace mode, but output path exists: "
 200                         "output-path=\"%s\"", fs_sink->output_dir_path->str);
 201                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 202                 goto end;
 203         }
 204
 205         status = ensure_output_dir_exists(fs_sink);
 206         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 207                 /* ensure_output_dir_exists() logs errors */
 208                 goto end;
 209         }
 210
 211         fs_sink->traces = g_hash_table_new_full(g_direct_hash, g_direct_equal,
 212                 NULL, (GDestroyNotify) fs_sink_trace_destroy);
 213         if (!fs_sink->traces) {
 214                 BT_COMP_LOGE_STR("Failed to allocate one GHashTable.");
 215                 status = BT_SELF_COMPONENT_STATUS_NOMEM;
 216                 goto end;
 217         }
 218
 219         status = bt_self_component_sink_add_input_port(self_comp_sink,
 220                 in_port_name, NULL, NULL);
 221         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 222                 goto end;
 223         }
 224
 225         bt_self_component_set_data(self_comp, fs_sink);
 226
 227 end:
 228         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 229                 destroy_fs_sink_comp(fs_sink);
 230         }
 231
 232         return status;
 233 }
 234
 235 static inline
 236 struct fs_sink_stream *borrow_stream(struct fs_sink_comp *fs_sink,
 237                 const bt_stream *ir_stream)
 238 {
 239         const bt_trace *ir_trace = bt_stream_borrow_trace_const(ir_stream);
 240         struct fs_sink_trace *trace;
 241         struct fs_sink_stream *stream = NULL;
 242
 243         trace = g_hash_table_lookup(fs_sink->traces, ir_trace);
 244         if (G_UNLIKELY(!trace)) {
 245                 if (fs_sink->assume_single_trace &&
 246                                 g_hash_table_size(fs_sink->traces) > 0) {
 247                         BT_COMP_LOGE("Single trace mode, but getting more than one trace: "
 248                                 "stream-name=\"%s\"",
 249                                 bt_stream_get_name(ir_stream));
 250                         goto end;
 251                 }
 252
 253                 trace = fs_sink_trace_create(fs_sink, ir_trace);
 254                 if (!trace) {
 255                         goto end;
 256                 }
 257         }
 258
 259         stream = g_hash_table_lookup(trace->streams, ir_stream);
 260         if (G_UNLIKELY(!stream)) {
 261                 stream = fs_sink_stream_create(trace, ir_stream);
 262                 if (!stream) {
 263                         goto end;
 264                 }
 265         }
 266
 267 end:
 268         return stream;
 269 }
 270
 271 static inline
 272 bt_self_component_status handle_event_msg(struct fs_sink_comp *fs_sink,
 273                 const bt_message *msg)
 274 {
 275         int ret;
 276         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 277         const bt_event *ir_event = bt_message_event_borrow_event_const(msg);
 278         const bt_stream *ir_stream = bt_event_borrow_stream_const(ir_event);
 279         struct fs_sink_stream *stream;
 280         struct fs_sink_ctf_event_class *ec = NULL;
 281         const bt_clock_snapshot *cs = NULL;
 282
 283         stream = borrow_stream(fs_sink, ir_stream);
 284         if (G_UNLIKELY(!stream)) {
 285                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 286                 goto end;
 287         }
 288
 289         ret = try_translate_event_class_trace_ir_to_ctf_ir(fs_sink,
 290                 stream->sc, bt_event_borrow_class_const(ir_event), &ec);
 291         if (ret) {
 292                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 293                 goto end;
 294         }
 295
 296         BT_ASSERT(ec);
 297
 298         if (stream->sc->default_clock_class) {
 299                 cs = bt_message_event_borrow_default_clock_snapshot_const(
 300                         msg);
 301         }
 302
 303         ret = fs_sink_stream_write_event(stream, cs, ir_event, ec);
 304         if (G_UNLIKELY(ret)) {
 305                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 306                 goto end;
 307         }
 308
 309 end:
 310         return status;
 311 }
 312
 313 static inline
 314 bt_self_component_status handle_packet_beginning_msg(
 315                 struct fs_sink_comp *fs_sink, const bt_message *msg)
 316 {
 317         int ret;
 318         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 319         const bt_packet *ir_packet =
 320                 bt_message_packet_beginning_borrow_packet_const(msg);
 321         const bt_stream *ir_stream = bt_packet_borrow_stream_const(ir_packet);
 322         struct fs_sink_stream *stream;
 323         const bt_clock_snapshot *cs = NULL;
 324
 325         stream = borrow_stream(fs_sink, ir_stream);
 326         if (G_UNLIKELY(!stream)) {
 327                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 328                 goto end;
 329         }
 330
 331         if (stream->sc->packets_have_ts_begin) {
 332                 cs = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
 333                         msg);
 334                 BT_ASSERT(cs);
 335         }
 336
 337         /*
 338          * If we previously received a discarded events message with
 339          * a time range, make sure that its beginning time matches what's
 340          * expected for CTF 1.8, that is:
 341          *
 342          * * Its beginning time is the previous packet's end
 343          *   time (or the current packet's beginning time if
 344          *   this is the first packet).
 345          *
 346          * We check this here instead of in handle_packet_end_msg()
 347          * because we want to catch any incompatible message as early as
 348          * possible to report the error.
 349          *
 350          * Validation of the discarded events message's end time is
 351          * performed in handle_packet_end_msg().
 352          */
 353         if (stream->discarded_events_state.in_range) {
 354                 uint64_t expected_cs;
 355
 356                 /*
 357                  * `stream->discarded_events_state.in_range` is only set
 358                  * when the stream class's discarded events have a time
 359                  * range.
 360                  *
 361                  * It is required that the packet beginning and end
 362                  * messages for this stream class have times when
 363                  * discarded events have a time range.
 364                  */
 365                 BT_ASSERT(stream->sc->discarded_events_has_ts);
 366                 BT_ASSERT(stream->sc->packets_have_ts_begin);
 367                 BT_ASSERT(stream->sc->packets_have_ts_end);
 368
 369                 if (stream->prev_packet_state.end_cs == UINT64_C(-1)) {
 370                         /* We're opening the first packet */
 371                         expected_cs = bt_clock_snapshot_get_value(cs);
 372                 } else {
 373                         expected_cs = stream->prev_packet_state.end_cs;
 374                 }
 375
 376                 if (stream->discarded_events_state.beginning_cs !=
 377                                 expected_cs) {
 378                         BT_COMP_LOGE("Incompatible discarded events message: "
 379                                 "unexpected beginning time: "
 380                                 "beginning-cs-val=%" PRIu64 ", "
 381                                 "expected-beginning-cs-val=%" PRIu64 ", "
 382                                 "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 383                                 "trace-name=\"%s\", path=\"%s/%s\"",
 384                                 stream->discarded_events_state.beginning_cs,
 385                                 expected_cs,
 386                                 bt_stream_get_id(ir_stream),
 387                                 bt_stream_get_name(ir_stream),
 388                                 bt_trace_get_name(
 389                                         bt_stream_borrow_trace_const(ir_stream)),
 390                                 stream->trace->path->str, stream->file_name->str);
 391                         status = BT_SELF_COMPONENT_STATUS_ERROR;
 392                         goto end;
 393                 }
 394         }
 395
 396         /*
 397          * If we previously received a discarded packets message with a
 398          * time range, make sure that its beginning and end times match
 399          * what's expected for CTF 1.8, that is:
 400          *
 401          * * Its beginning time is the previous packet's end time.
 402          *
 403          * * Its end time is the current packet's beginning time.
 404          */
 405         if (stream->discarded_packets_state.in_range) {
 406                 uint64_t expected_end_cs;
 407
 408                 /*
 409                  * `stream->discarded_packets_state.in_range` is only
 410                  * set when the stream class's discarded packets have a
 411                  * time range.
 412                  *
 413                  * It is required that the packet beginning and end
 414                  * messages for this stream class have times when
 415                  * discarded packets have a time range.
 416                  */
 417                 BT_ASSERT(stream->sc->discarded_packets_has_ts);
 418                 BT_ASSERT(stream->sc->packets_have_ts_begin);
 419                 BT_ASSERT(stream->sc->packets_have_ts_end);
 420
 421                 /*
 422                  * It is not supported to have a discarded packets
 423                  * message _before_ the first packet: we cannot validate
 424                  * that its beginning time is compatible with CTF 1.8 in
 425                  * this case.
 426                  */
 427                 if (stream->prev_packet_state.end_cs == UINT64_C(-1)) {
 428                         BT_COMP_LOGE("Incompatible discarded packets message "
 429                                 "occuring before the stream's first packet: "
 430                                 "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 431                                 "trace-name=\"%s\", path=\"%s/%s\"",
 432                                 bt_stream_get_id(ir_stream),
 433                                 bt_stream_get_name(ir_stream),
 434                                 bt_trace_get_name(
 435                                         bt_stream_borrow_trace_const(ir_stream)),
 436                                 stream->trace->path->str, stream->file_name->str);
 437                         status = BT_SELF_COMPONENT_STATUS_ERROR;
 438                         goto end;
 439                 }
 440
 441                 if (stream->discarded_packets_state.beginning_cs !=
 442                                 stream->prev_packet_state.end_cs) {
 443                         BT_COMP_LOGE("Incompatible discarded packets message: "
 444                                 "unexpected beginning time: "
 445                                 "beginning-cs-val=%" PRIu64 ", "
 446                                 "expected-beginning-cs-val=%" PRIu64 ", "
 447                                 "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 448                                 "trace-name=\"%s\", path=\"%s/%s\"",
 449                                 stream->discarded_packets_state.beginning_cs,
 450                                 stream->prev_packet_state.end_cs,
 451                                 bt_stream_get_id(ir_stream),
 452                                 bt_stream_get_name(ir_stream),
 453                                 bt_trace_get_name(
 454                                         bt_stream_borrow_trace_const(ir_stream)),
 455                                 stream->trace->path->str, stream->file_name->str);
 456                         status = BT_SELF_COMPONENT_STATUS_ERROR;
 457                         goto end;
 458                 }
 459
 460                 expected_end_cs = bt_clock_snapshot_get_value(cs);
 461
 462                 if (stream->discarded_packets_state.end_cs !=
 463                                 expected_end_cs) {
 464                         BT_COMP_LOGE("Incompatible discarded packets message: "
 465                                 "unexpected end time: "
 466                                 "end-cs-val=%" PRIu64 ", "
 467                                 "expected-end-cs-val=%" PRIu64 ", "
 468                                 "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 469                                 "trace-name=\"%s\", path=\"%s/%s\"",
 470                                 stream->discarded_packets_state.end_cs,
 471                                 expected_end_cs,
 472                                 bt_stream_get_id(ir_stream),
 473                                 bt_stream_get_name(ir_stream),
 474                                 bt_trace_get_name(
 475                                         bt_stream_borrow_trace_const(ir_stream)),
 476                                 stream->trace->path->str, stream->file_name->str);
 477                         status = BT_SELF_COMPONENT_STATUS_ERROR;
 478                         goto end;
 479                 }
 480         }
 481
 482         /*
 483          * We're not in a discarded packets time range anymore since we
 484          * require that the discarded packets time ranges go from one
 485          * packet's end time to the next packet's beginning time, and
 486          * we're handling a packet beginning message here.
 487          */
 488         stream->discarded_packets_state.in_range = false;
 489
 490         ret = fs_sink_stream_open_packet(stream, cs, ir_packet);
 491         if (ret) {
 492                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 493                 goto end;
 494         }
 495
 496 end:
 497         return status;
 498 }
 499
 500 static inline
 501 bt_self_component_status handle_packet_end_msg(
 502                 struct fs_sink_comp *fs_sink, const bt_message *msg)
 503 {
 504         int ret;
 505         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 506         const bt_packet *ir_packet =
 507                 bt_message_packet_end_borrow_packet_const(msg);
 508         const bt_stream *ir_stream = bt_packet_borrow_stream_const(ir_packet);
 509         struct fs_sink_stream *stream;
 510         const bt_clock_snapshot *cs = NULL;
 511
 512         stream = borrow_stream(fs_sink, ir_stream);
 513         if (G_UNLIKELY(!stream)) {
 514                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 515                 goto end;
 516         }
 517
 518         if (stream->sc->packets_have_ts_end) {
 519                 cs = bt_message_packet_end_borrow_default_clock_snapshot_const(
 520                         msg);
 521                 BT_ASSERT(cs);
 522         }
 523
 524         /*
 525          * If we previously received a discarded events message with
 526          * a time range, make sure that its end time matches what's
 527          * expected for CTF 1.8, that is:
 528          *
 529          * * Its end time is the current packet's end time.
 530          *
 531          * Validation of the discarded events message's beginning time
 532          * is performed in handle_packet_beginning_msg().
 533          */
 534         if (stream->discarded_events_state.in_range) {
 535                 uint64_t expected_cs;
 536
 537                 /*
 538                  * `stream->discarded_events_state.in_range` is only set
 539                  * when the stream class's discarded events have a time
 540                  * range.
 541                  *
 542                  * It is required that the packet beginning and end
 543                  * messages for this stream class have times when
 544                  * discarded events have a time range.
 545                  */
 546                 BT_ASSERT(stream->sc->discarded_events_has_ts);
 547                 BT_ASSERT(stream->sc->packets_have_ts_begin);
 548                 BT_ASSERT(stream->sc->packets_have_ts_end);
 549
 550                 expected_cs = bt_clock_snapshot_get_value(cs);
 551
 552                 if (stream->discarded_events_state.end_cs != expected_cs) {
 553                         BT_COMP_LOGE("Incompatible discarded events message: "
 554                                 "unexpected end time: "
 555                                 "end-cs-val=%" PRIu64 ", "
 556                                 "expected-end-cs-val=%" PRIu64 ", "
 557                                 "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 558                                 "trace-name=\"%s\", path=\"%s/%s\"",
 559                                 stream->discarded_events_state.end_cs,
 560                                 expected_cs,
 561                                 bt_stream_get_id(ir_stream),
 562                                 bt_stream_get_name(ir_stream),
 563                                 bt_trace_get_name(
 564                                         bt_stream_borrow_trace_const(ir_stream)),
 565                                 stream->trace->path->str, stream->file_name->str);
 566                         status = BT_SELF_COMPONENT_STATUS_ERROR;
 567                         goto end;
 568                 }
 569         }
 570
 571         ret = fs_sink_stream_close_packet(stream, cs);
 572         if (ret) {
 573                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 574                 goto end;
 575         }
 576
 577         /*
 578          * We're not in a discarded events time range anymore since we
 579          * require that the discarded events time ranges go from one
 580          * packet's end time to the next packet's end time, and we're
 581          * handling a packet end message here.
 582          */
 583         stream->discarded_events_state.in_range = false;
 584
 585 end:
 586         return status;
 587 }
 588
 589 static inline
 590 bt_self_component_status handle_stream_beginning_msg(
 591                 struct fs_sink_comp *fs_sink, const bt_message *msg)
 592 {
 593         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 594         const bt_stream *ir_stream =
 595                 bt_message_stream_beginning_borrow_stream_const(msg);
 596         const bt_stream_class *ir_sc =
 597                 bt_stream_borrow_class_const(ir_stream);
 598         struct fs_sink_stream *stream;
 599         bool packets_have_beginning_end_cs =
 600                 bt_stream_class_packets_have_beginning_default_clock_snapshot(ir_sc) &&
 601                 bt_stream_class_packets_have_end_default_clock_snapshot(ir_sc);
 602
 603         /*
 604          * Not supported: discarded events with default clock snapshots,
 605          * but packet beginning/end without default clock snapshot.
 606          */
 607         if (!fs_sink->ignore_discarded_events &&
 608                         bt_stream_class_discarded_events_have_default_clock_snapshots(ir_sc) &&
 609                         !packets_have_beginning_end_cs) {
 610                 BT_COMP_LOGE("Unsupported stream: discarded events have "
 611                         "default clock snapshots, but packets have no "
 612                         "beginning and/or end default clock snapshots: "
 613                         "stream-addr=%p, "
 614                         "stream-id=%" PRIu64 ", "
 615                         "stream-name=\"%s\"",
 616                         ir_stream, bt_stream_get_id(ir_stream),
 617                         bt_stream_get_name(ir_stream));
 618                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 619                 goto end;
 620         }
 621
 622         /*
 623          * Not supported: discarded packets with default clock
 624          * snapshots, but packet beginning/end without default clock
 625          * snapshot.
 626          */
 627         if (!fs_sink->ignore_discarded_packets &&
 628                         bt_stream_class_discarded_packets_have_default_clock_snapshots(ir_sc) &&
 629                         !packets_have_beginning_end_cs) {
 630                 BT_COMP_LOGE("Unsupported stream: discarded packets have "
 631                         "default clock snapshots, but packets have no "
 632                         "beginning and/or end default clock snapshots: "
 633                         "stream-addr=%p, "
 634                         "stream-id=%" PRIu64 ", "
 635                         "stream-name=\"%s\"",
 636                         ir_stream, bt_stream_get_id(ir_stream),
 637                         bt_stream_get_name(ir_stream));
 638                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 639                 goto end;
 640         }
 641
 642         stream = borrow_stream(fs_sink, ir_stream);
 643         if (!stream) {
 644                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 645                 goto end;
 646         }
 647
 648         BT_COMP_LOGI("Created new, empty stream file: "
 649                 "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 650                 "trace-name=\"%s\", path=\"%s/%s\"",
 651                 bt_stream_get_id(ir_stream), bt_stream_get_name(ir_stream),
 652                 bt_trace_get_name(bt_stream_borrow_trace_const(ir_stream)),
 653                 stream->trace->path->str, stream->file_name->str);
 654
 655 end:
 656         return status;
 657 }
 658
 659 static inline
 660 bt_self_component_status handle_stream_end_msg(struct fs_sink_comp *fs_sink,
 661                 const bt_message *msg)
 662 {
 663         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 664         const bt_stream *ir_stream =
 665                 bt_message_stream_end_borrow_stream_const(msg);
 666         struct fs_sink_stream *stream;
 667
 668         stream = borrow_stream(fs_sink, ir_stream);
 669         if (!stream) {
 670                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 671                 goto end;
 672         }
 673
 674         BT_COMP_LOGI("Closing stream file: "
 675                 "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 676                 "trace-name=\"%s\", path=\"%s/%s\"",
 677                 bt_stream_get_id(ir_stream), bt_stream_get_name(ir_stream),
 678                 bt_trace_get_name(bt_stream_borrow_trace_const(ir_stream)),
 679                 stream->trace->path->str, stream->file_name->str);
 680
 681         /*
 682          * This destroys the stream object and frees all its resources,
 683          * closing the stream file.
 684          */
 685         g_hash_table_remove(stream->trace->streams, ir_stream);
 686
 687 end:
 688         return status;
 689 }
 690
 691 static inline
 692 bt_self_component_status handle_discarded_events_msg(
 693                 struct fs_sink_comp *fs_sink, const bt_message *msg)
 694 {
 695         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 696         const bt_stream *ir_stream =
 697                 bt_message_discarded_events_borrow_stream_const(msg);
 698         struct fs_sink_stream *stream;
 699         const bt_clock_snapshot *cs = NULL;
 700         bt_property_availability avail;
 701         uint64_t count;
 702
 703         stream = borrow_stream(fs_sink, ir_stream);
 704         if (!stream) {
 705                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 706                 goto end;
 707         }
 708
 709         if (fs_sink->ignore_discarded_events) {
 710                 BT_COMP_LOGI("Ignoring discarded events message: "
 711                         "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 712                         "trace-name=\"%s\", path=\"%s/%s\"",
 713                         bt_stream_get_id(ir_stream),
 714                         bt_stream_get_name(ir_stream),
 715                         bt_trace_get_name(
 716                                 bt_stream_borrow_trace_const(ir_stream)),
 717                         stream->trace->path->str, stream->file_name->str);
 718                 goto end;
 719         }
 720
 721         if (stream->discarded_events_state.in_range) {
 722                 BT_COMP_LOGE("Unsupported contiguous discarded events message: "
 723                         "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 724                         "trace-name=\"%s\", path=\"%s/%s\"",
 725                         bt_stream_get_id(ir_stream),
 726                         bt_stream_get_name(ir_stream),
 727                         bt_trace_get_name(
 728                                 bt_stream_borrow_trace_const(ir_stream)),
 729                         stream->trace->path->str, stream->file_name->str);
 730                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 731                 goto end;
 732         }
 733
 734         /*
 735          * If we're currently in an opened packet (got a packet
 736          * beginning message, but no packet end message yet), we do not
 737          * support having a discarded events message with a time range
 738          * because we require that the discarded events message's time
 739          * range go from a packet's end time to the next packet's end
 740          * time.
 741          */
 742         if (stream->packet_state.is_open &&
 743                         stream->sc->discarded_events_has_ts) {
 744                 BT_COMP_LOGE("Unsupported discarded events message with "
 745                         "default clock snapshots occuring within a packet: "
 746                         "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 747                         "trace-name=\"%s\", path=\"%s/%s\"",
 748                         bt_stream_get_id(ir_stream),
 749                         bt_stream_get_name(ir_stream),
 750                         bt_trace_get_name(
 751                                 bt_stream_borrow_trace_const(ir_stream)),
 752                         stream->trace->path->str, stream->file_name->str);
 753                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 754                 goto end;
 755         }
 756
 757         if (stream->sc->discarded_events_has_ts) {
 758                 /*
 759                  * Make the stream's state be in the time range of a
 760                  * discarded events message since we have the message's
 761                  * time range (`stream->sc->discarded_events_has_ts`).
 762                  */
 763                 stream->discarded_events_state.in_range = true;
 764
 765                 /*
 766                  * The clock snapshot values will be validated when
 767                  * handling the next packet beginning and end messages
 768                  * (next calls to handle_packet_beginning_msg() and
 769                  * handle_packet_end_msg()).
 770                  */
 771                 cs = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
 772                         msg);
 773                 BT_ASSERT(cs);
 774                 stream->discarded_events_state.beginning_cs =
 775                         bt_clock_snapshot_get_value(cs);
 776                 cs = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
 777                         msg);
 778                 BT_ASSERT(cs);
 779                 stream->discarded_events_state.end_cs = bt_clock_snapshot_get_value(cs);
 780         }
 781
 782         avail = bt_message_discarded_events_get_count(msg, &count);
 783         if (avail != BT_PROPERTY_AVAILABILITY_AVAILABLE) {
 784                 /*
 785                  * There's no specific count of discarded events: set it
 786                  * to 1 so that we know that we at least discarded
 787                  * something.
 788                  */
 789                 count = 1;
 790         }
 791
 792         stream->packet_state.discarded_events_counter += count;
 793
 794 end:
 795         return status;
 796 }
 797
 798 static inline
 799 bt_self_component_status handle_discarded_packets_msg(
 800                 struct fs_sink_comp *fs_sink, const bt_message *msg)
 801 {
 802         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 803         const bt_stream *ir_stream =
 804                 bt_message_discarded_packets_borrow_stream_const(msg);
 805         struct fs_sink_stream *stream;
 806         const bt_clock_snapshot *cs = NULL;
 807         bt_property_availability avail;
 808         uint64_t count;
 809
 810         stream = borrow_stream(fs_sink, ir_stream);
 811         if (!stream) {
 812                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 813                 goto end;
 814         }
 815
 816         if (fs_sink->ignore_discarded_packets) {
 817                 BT_COMP_LOGI("Ignoring discarded packets message: "
 818                         "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 819                         "trace-name=\"%s\", path=\"%s/%s\"",
 820                         bt_stream_get_id(ir_stream),
 821                         bt_stream_get_name(ir_stream),
 822                         bt_trace_get_name(
 823                                 bt_stream_borrow_trace_const(ir_stream)),
 824                         stream->trace->path->str, stream->file_name->str);
 825                 goto end;
 826         }
 827
 828         if (stream->discarded_packets_state.in_range) {
 829                 BT_COMP_LOGE("Unsupported contiguous discarded packets message: "
 830                         "stream-id=%" PRIu64 ", stream-name=\"%s\", "
 831                         "trace-name=\"%s\", path=\"%s/%s\"",
 832                         bt_stream_get_id(ir_stream),
 833                         bt_stream_get_name(ir_stream),
 834                         bt_trace_get_name(
 835                                 bt_stream_borrow_trace_const(ir_stream)),
 836                         stream->trace->path->str, stream->file_name->str);
 837                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 838                 goto end;
 839         }
 840
 841         /*
 842          * Discarded packets messages are guaranteed to occur between
 843          * packets.
 844          */
 845         BT_ASSERT(!stream->packet_state.is_open);
 846
 847         if (stream->sc->discarded_packets_has_ts) {
 848                 /*
 849                  * Make the stream's state be in the time range of a
 850                  * discarded packets message since we have the message's
 851                  * time range (`stream->sc->discarded_packets_has_ts`).
 852                  */
 853                 stream->discarded_packets_state.in_range = true;
 854
 855                 /*
 856                  * The clock snapshot values will be validated when
 857                  * handling the next packet beginning message (next call
 858                  * to handle_packet_beginning_msg()).
 859                  */
 860                 cs = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
 861                         msg);
 862                 BT_ASSERT(cs);
 863                 stream->discarded_packets_state.beginning_cs =
 864                         bt_clock_snapshot_get_value(cs);
 865                 cs = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
 866                         msg);
 867                 BT_ASSERT(cs);
 868                 stream->discarded_packets_state.end_cs =
 869                         bt_clock_snapshot_get_value(cs);
 870         }
 871
 872         avail = bt_message_discarded_packets_get_count(msg, &count);
 873         if (avail != BT_PROPERTY_AVAILABILITY_AVAILABLE) {
 874                 /*
 875                  * There's no specific count of discarded packets: set
 876                  * it to 1 so that we know that we at least discarded
 877                  * something.
 878                  */
 879                 count = 1;
 880         }
 881
 882         stream->packet_state.seq_num += count;
 883
 884 end:
 885         return status;
 886 }
 887
 888 static inline
 889 void put_messages(bt_message_array_const msgs, uint64_t count)
 890 {
 891         uint64_t i;
 892
 893         for (i = 0; i < count; i++) {
 894                 BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
 895         }
 896 }
 897
 898 BT_HIDDEN
 899 bt_self_component_status ctf_fs_sink_consume(bt_self_component_sink *self_comp)
 900 {
 901         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 902         struct fs_sink_comp *fs_sink;
 903         bt_message_iterator_status it_status;
 904         uint64_t msg_count = 0;
 905         bt_message_array_const msgs;
 906
 907         fs_sink = bt_self_component_get_data(
 908                         bt_self_component_sink_as_self_component(self_comp));
 909         BT_ASSERT(fs_sink);
 910         BT_ASSERT(fs_sink->upstream_iter);
 911
 912         /* Consume messages */
 913         it_status = bt_self_component_port_input_message_iterator_next(
 914                 fs_sink->upstream_iter, &msgs, &msg_count);
 915         if (it_status < 0) {
 916                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 917                 goto end;
 918         }
 919
 920         switch (it_status) {
 921         case BT_MESSAGE_ITERATOR_STATUS_OK:
 922         {
 923                 uint64_t i;
 924
 925                 for (i = 0; i < msg_count; i++) {
 926                         const bt_message *msg = msgs[i];
 927
 928                         BT_ASSERT(msg);
 929
 930                         switch (bt_message_get_type(msg)) {
 931                         case BT_MESSAGE_TYPE_EVENT:
 932                                 status = handle_event_msg(fs_sink, msg);
 933                                 break;
 934                         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 935                                 status = handle_packet_beginning_msg(
 936                                         fs_sink, msg);
 937                                 break;
 938                         case BT_MESSAGE_TYPE_PACKET_END:
 939                                 status = handle_packet_end_msg(
 940                                         fs_sink, msg);
 941                                 break;
 942                         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 943                                 /* Ignore */
 944                                 BT_COMP_LOGD_STR("Ignoring message iterator inactivity message.");
 945                                 break;
 946                         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 947                                 status = handle_stream_beginning_msg(
 948                                         fs_sink, msg);
 949                                 break;
 950                         case BT_MESSAGE_TYPE_STREAM_END:
 951                                 status = handle_stream_end_msg(
 952                                         fs_sink, msg);
 953                                 break;
 954                         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
 955                         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
 956                                 /* Not supported by CTF 1.8 */
 957                                 BT_COMP_LOGD_STR("Ignoring stream activity message.");
 958                                 break;
 959                         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 960                                 status = handle_discarded_events_msg(
 961                                         fs_sink, msg);
 962                                 break;
 963                         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 964                                 status = handle_discarded_packets_msg(
 965                                         fs_sink, msg);
 966                                 break;
 967                         default:
 968                                 abort();
 969                         }
 970
 971                         BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
 972
 973                         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 974                                 BT_COMP_LOGE("Failed to handle message: "
 975                                         "generated CTF traces could be incomplete: "
 976                                         "output-dir-path=\"%s\"",
 977                                         fs_sink->output_dir_path->str);
 978                                 goto error;
 979                         }
 980                 }
 981
 982                 break;
 983         }
 984         case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
 985                 status = BT_SELF_COMPONENT_STATUS_AGAIN;
 986                 break;
 987         case BT_MESSAGE_ITERATOR_STATUS_END:
 988                 /* TODO: Finalize all traces (should already be done?) */
 989                 status = BT_SELF_COMPONENT_STATUS_END;
 990                 break;
 991         case BT_MESSAGE_ITERATOR_STATUS_NOMEM:
 992                 status = BT_SELF_COMPONENT_STATUS_NOMEM;
 993                 break;
 994         case BT_MESSAGE_ITERATOR_STATUS_ERROR:
 995                 status = BT_SELF_COMPONENT_STATUS_NOMEM;
 996                 break;
 997         default:
 998                 break;
 999         }
1000
1001         goto end;
1002
1003 error:
1004         BT_ASSERT(status != BT_SELF_COMPONENT_STATUS_OK);
1005         put_messages(msgs, msg_count);
1006
1007 end:
1008         return status;
1009 }
1010
1011 BT_HIDDEN
1012 bt_self_component_status ctf_fs_sink_graph_is_configured(
1013                 bt_self_component_sink *self_comp)
1014 {
1015         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
1016         struct fs_sink_comp *fs_sink = bt_self_component_get_data(
1017                         bt_self_component_sink_as_self_component(self_comp));
1018
1019         fs_sink->upstream_iter =
1020                 bt_self_component_port_input_message_iterator_create(
1021                         bt_self_component_sink_borrow_input_port_by_name(
1022                                 self_comp, in_port_name));
1023         if (!fs_sink->upstream_iter) {
1024                 status = BT_SELF_COMPONENT_STATUS_NOMEM;
1025                 goto end;
1026         }
1027
1028 end:
1029         return status;
1030 }
1031
1032 BT_HIDDEN
1033 void ctf_fs_sink_finalize(bt_self_component_sink *self_comp)
1034 {
1035         struct fs_sink_comp *fs_sink = bt_self_component_get_data(
1036                         bt_self_component_sink_as_self_component(self_comp));
1037
1038         destroy_fs_sink_comp(fs_sink);
1039 }