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

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