plugins/ctf/fs-sink/fs-sink-stream.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_LOG_TAG "PLUGIN-CTF-FS-SINK-STREAM"
  24 #include "logging.h"
  25
  26 #include <babeltrace/babeltrace.h>
  27 #include <stdio.h>
  28 #include <stdbool.h>
  29 #include <glib.h>
  30 #include <babeltrace/assert-internal.h>
  31 #include <babeltrace/ctfser-internal.h>
  32 #include <babeltrace/endian-internal.h>
  33
  34 #include "fs-sink-trace.h"
  35 #include "fs-sink-stream.h"
  36 #include "translate-trace-ir-to-ctf-ir.h"
  37
  38 BT_HIDDEN
  39 void fs_sink_stream_destroy(struct fs_sink_stream *stream)
  40 {
  41         if (!stream) {
  42                 goto end;
  43         }
  44
  45         bt_ctfser_fini(&stream->ctfser);
  46
  47         if (stream->file_name) {
  48                 g_string_free(stream->file_name, TRUE);
  49                 stream->file_name = NULL;
  50         }
  51
  52         bt_packet_put_ref(stream->packet_state.packet);
  53         g_free(stream);
  54
  55 end:
  56         return;
  57 }
  58
  59 static
  60 bool stream_file_name_exists(struct fs_sink_trace *trace, const char *name)
  61 {
  62         bool exists = false;
  63         GHashTableIter iter;
  64         gpointer key, value;
  65
  66         g_hash_table_iter_init(&iter, trace->streams);
  67
  68         while (g_hash_table_iter_next(&iter, &key, &value)) {
  69                 struct fs_sink_stream *stream = value;
  70
  71                 if (strcmp(name, stream->file_name->str) == 0) {
  72                         exists = true;
  73                         goto end;
  74                 }
  75         }
  76
  77 end:
  78         return exists;
  79 }
  80
  81 static
  82 GString *sanitize_stream_file_name(const char *file_name)
  83 {
  84         GString *san_file_name = g_string_new(NULL);
  85         const char *ch;
  86         gchar *basename;
  87
  88         BT_ASSERT(san_file_name);
  89         BT_ASSERT(file_name);
  90         basename = g_path_get_basename(file_name);
  91
  92         for (ch = basename; *ch != '\0'; ch++) {
  93                 if (*ch == '/') {
  94                         g_string_append_c(san_file_name, '_');
  95                 } else {
  96                         g_string_append_c(san_file_name, *ch);
  97                 }
  98         }
  99
 100         /* Do not allow `.` and `..` either */
 101         if (strcmp(san_file_name->str, ".") == 0 ||
 102                         strcmp(san_file_name->str, "..") == 0) {
 103                 g_string_assign(san_file_name, "stream");
 104         }
 105
 106         g_free(basename);
 107         return san_file_name;
 108 }
 109
 110 static
 111 GString *make_unique_stream_file_name(struct fs_sink_trace *trace,
 112                 const char *base)
 113 {
 114         GString *san_base = sanitize_stream_file_name(base);
 115         GString *name = g_string_new(san_base->str);
 116         unsigned int suffix = 0;
 117
 118         BT_ASSERT(name);
 119
 120         while (stream_file_name_exists(trace, name->str) &&
 121                         strcmp(name->str, "metadata") == 0) {
 122                 g_string_assign(name, san_base->str);
 123                 g_string_append_printf(name, "%u", suffix);
 124                 suffix++;
 125         }
 126
 127         g_string_free(san_base, TRUE);
 128         return name;
 129 }
 130
 131 static
 132 void set_stream_file_name(struct fs_sink_stream *stream)
 133 {
 134         const char *base_name = bt_stream_get_name(stream->ir_stream);
 135
 136         if (!base_name) {
 137                 base_name = "stream";
 138         }
 139
 140         BT_ASSERT(!stream->file_name);
 141         stream->file_name = make_unique_stream_file_name(stream->trace,
 142                 base_name);
 143 }
 144
 145 BT_HIDDEN
 146 struct fs_sink_stream *fs_sink_stream_create(struct fs_sink_trace *trace,
 147                 const bt_stream *ir_stream)
 148 {
 149         struct fs_sink_stream *stream = g_new0(struct fs_sink_stream, 1);
 150         int ret;
 151         GString *path = g_string_new(trace->path->str);
 152
 153         if (!stream) {
 154                 goto end;
 155         }
 156
 157         stream->trace = trace;
 158         stream->ir_stream = ir_stream;
 159         stream->packet_state.beginning_cs = UINT64_C(-1);
 160         stream->packet_state.end_cs = UINT64_C(-1);
 161         stream->prev_packet_state.end_cs = UINT64_C(-1);
 162         stream->prev_packet_state.discarded_events_counter = UINT64_C(-1);
 163         stream->prev_packet_state.seq_num = UINT64_C(-1);
 164         ret = try_translate_stream_class_trace_ir_to_ctf_ir(trace->tc,
 165                 bt_stream_borrow_class_const(ir_stream), &stream->sc);
 166         if (ret) {
 167                 goto error;
 168         }
 169
 170         set_stream_file_name(stream);
 171         g_string_append_printf(path, "/%s", stream->file_name->str);
 172         ret = bt_ctfser_init(&stream->ctfser, path->str);
 173         if (ret) {
 174                 goto error;
 175         }
 176
 177         g_hash_table_insert(trace->streams, (gpointer) ir_stream, stream);
 178         goto end;
 179
 180 error:
 181         fs_sink_stream_destroy(stream);
 182         stream = NULL;
 183
 184 end:
 185         if (path) {
 186                 g_string_free(path, TRUE);
 187         }
 188
 189         return stream;
 190 }
 191
 192 static
 193 int write_field(struct fs_sink_stream *stream,
 194                 struct fs_sink_ctf_field_class *fc, const bt_field *field);
 195
 196 static inline
 197 int write_int_field(struct fs_sink_stream *stream,
 198                 struct fs_sink_ctf_field_class_int *fc, const bt_field *field)
 199 {
 200         int ret;
 201
 202         if (fc->is_signed) {
 203                 ret = bt_ctfser_write_signed_int(&stream->ctfser,
 204                         bt_field_signed_integer_get_value(field),
 205                         fc->base.base.alignment, fc->base.size, BYTE_ORDER);
 206         } else {
 207                 ret = bt_ctfser_write_unsigned_int(&stream->ctfser,
 208                         bt_field_unsigned_integer_get_value(field),
 209                         fc->base.base.alignment, fc->base.size, BYTE_ORDER);
 210         }
 211
 212         return ret;
 213 }
 214
 215 static inline
 216 int write_float_field(struct fs_sink_stream *stream,
 217                 struct fs_sink_ctf_field_class_float *fc, const bt_field *field)
 218 {
 219         int ret;
 220         double val = bt_field_real_get_value(field);
 221
 222         if (fc->base.size == 32) {
 223                 ret = bt_ctfser_write_float32(&stream->ctfser, val,
 224                         fc->base.base.alignment, BYTE_ORDER);
 225         } else {
 226                 ret = bt_ctfser_write_float32(&stream->ctfser, val,
 227                         fc->base.base.alignment, BYTE_ORDER);
 228         }
 229
 230         return ret;
 231 }
 232
 233 static inline
 234 int write_string_field(struct fs_sink_stream *stream,
 235                 struct fs_sink_ctf_field_class_string *fc, const bt_field *field)
 236 {
 237         return bt_ctfser_write_string(&stream->ctfser,
 238                 bt_field_string_get_value(field));
 239 }
 240
 241 static inline
 242 int write_array_field_elements(struct fs_sink_stream *stream,
 243                 struct fs_sink_ctf_field_class_array_base *fc,
 244                 const bt_field *field)
 245 {
 246         uint64_t i;
 247         uint64_t len = bt_field_array_get_length(field);
 248         int ret = 0;
 249
 250         for (i = 0; i < len; i++) {
 251                 const bt_field *elem_field =
 252                         bt_field_array_borrow_element_field_by_index_const(
 253                                 field, i);
 254                 ret = write_field(stream, fc->elem_fc, elem_field);
 255                 if (unlikely(ret)) {
 256                         goto end;
 257                 }
 258         }
 259
 260 end:
 261         return ret;
 262 }
 263
 264 static inline
 265 int write_sequence_field(struct fs_sink_stream *stream,
 266                 struct fs_sink_ctf_field_class_sequence *fc,
 267                 const bt_field *field)
 268 {
 269         int ret;
 270
 271         if (fc->length_is_before) {
 272                 ret = bt_ctfser_write_unsigned_int(&stream->ctfser,
 273                         bt_field_array_get_length(field), 8, 32, BYTE_ORDER);
 274                 if (unlikely(ret)) {
 275                         goto end;
 276                 }
 277         }
 278
 279         ret = write_array_field_elements(stream, (void *) fc, field);
 280
 281 end:
 282         return ret;
 283 }
 284
 285 static inline
 286 int write_struct_field(struct fs_sink_stream *stream,
 287                 struct fs_sink_ctf_field_class_struct *fc,
 288                 const bt_field *field, bool align_struct)
 289 {
 290         int ret = 0;
 291         uint64_t i;
 292
 293         if (likely(align_struct)) {
 294                 ret = bt_ctfser_align_offset_in_current_packet(&stream->ctfser,
 295                         fc->base.alignment);
 296                 if (unlikely(ret)) {
 297                         goto end;
 298                 }
 299         }
 300
 301         for (i = 0; i < fc->members->len; i++) {
 302                 const bt_field *memb_field =
 303                         bt_field_structure_borrow_member_field_by_index_const(
 304                                 field, i);
 305                 struct fs_sink_ctf_field_class *member_fc =
 306                         fs_sink_ctf_field_class_struct_borrow_member_by_index(
 307                                 fc, i)->fc;
 308
 309                 ret = write_field(stream, member_fc, memb_field);
 310                 if (unlikely(ret)) {
 311                         goto end;
 312                 }
 313         }
 314
 315 end:
 316         return ret;
 317 }
 318
 319 static inline
 320 int write_variant_field(struct fs_sink_stream *stream,
 321                 struct fs_sink_ctf_field_class_variant *fc,
 322                 const bt_field *field)
 323 {
 324         uint64_t opt_index =
 325                 bt_field_variant_get_selected_option_field_index(field);
 326         int ret;
 327
 328         if (fc->tag_is_before) {
 329                 ret = bt_ctfser_write_unsigned_int(&stream->ctfser,
 330                         opt_index, 8, 16, BYTE_ORDER);
 331                 if (unlikely(ret)) {
 332                         goto end;
 333                 }
 334         }
 335
 336         ret = write_field(stream,
 337                 fs_sink_ctf_field_class_variant_borrow_option_by_index(fc,
 338                         opt_index)->fc,
 339                 bt_field_variant_borrow_selected_option_field_const(field));
 340
 341 end:
 342         return ret;
 343 }
 344
 345 static
 346 int write_field(struct fs_sink_stream *stream,
 347                 struct fs_sink_ctf_field_class *fc, const bt_field *field)
 348 {
 349         int ret;
 350
 351         switch (fc->type) {
 352         case FS_SINK_CTF_FIELD_CLASS_TYPE_INT:
 353                 ret = write_int_field(stream, (void *) fc, field);
 354                 break;
 355         case FS_SINK_CTF_FIELD_CLASS_TYPE_FLOAT:
 356                 ret = write_float_field(stream, (void *) fc, field);
 357                 break;
 358         case FS_SINK_CTF_FIELD_CLASS_TYPE_STRING:
 359                 ret = write_string_field(stream, (void *) fc, field);
 360                 break;
 361         case FS_SINK_CTF_FIELD_CLASS_TYPE_STRUCT:
 362                 ret = write_struct_field(stream, (void *) fc, field, true);
 363                 break;
 364         case FS_SINK_CTF_FIELD_CLASS_TYPE_ARRAY:
 365                 ret = write_array_field_elements(stream, (void *) fc, field);
 366                 break;
 367         case FS_SINK_CTF_FIELD_CLASS_TYPE_SEQUENCE:
 368                 ret = write_sequence_field(stream, (void *) fc, field);
 369                 break;
 370         case FS_SINK_CTF_FIELD_CLASS_TYPE_VARIANT:
 371                 ret = write_variant_field(stream, (void *) fc, field);
 372                 break;
 373         default:
 374                 abort();
 375         }
 376
 377         return ret;
 378 }
 379
 380 static inline
 381 int write_event_header(struct fs_sink_stream *stream,
 382                 const bt_clock_snapshot *cs, struct fs_sink_ctf_event_class *ec)
 383 {
 384         int ret;
 385
 386         /* Event class ID */
 387         ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 388                 bt_event_class_get_id(ec->ir_ec), 8, 64, BYTE_ORDER);
 389         if (unlikely(ret)) {
 390                 goto end;
 391         }
 392
 393         /* Time */
 394         if (stream->sc->default_clock_class) {
 395                 BT_ASSERT(cs);
 396                 ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 397                         bt_clock_snapshot_get_value(cs), 8, 64, BYTE_ORDER);
 398                 if (unlikely(ret)) {
 399                         goto end;
 400                 }
 401         }
 402
 403 end:
 404         return ret;
 405 }
 406
 407 BT_HIDDEN
 408 int fs_sink_stream_write_event(struct fs_sink_stream *stream,
 409                 const bt_clock_snapshot *cs, const bt_event *event,
 410                 struct fs_sink_ctf_event_class *ec)
 411 {
 412         int ret;
 413         const bt_field *field;
 414
 415         /* Header */
 416         ret = write_event_header(stream, cs, ec);
 417         if (unlikely(ret)) {
 418                 goto end;
 419         }
 420
 421         /* Common context */
 422         if (stream->sc->event_common_context_fc) {
 423                 field = bt_event_borrow_common_context_field_const(event);
 424                 BT_ASSERT(field);
 425                 ret = write_struct_field(stream,
 426                         (void *) stream->sc->event_common_context_fc,
 427                         field, true);
 428                 if (unlikely(ret)) {
 429                         goto end;
 430                 }
 431         }
 432
 433         /* Specific context */
 434         if (ec->spec_context_fc) {
 435                 field = bt_event_borrow_specific_context_field_const(event);
 436                 BT_ASSERT(field);
 437                 ret = write_struct_field(stream, (void *) ec->spec_context_fc,
 438                         field, true);
 439                 if (unlikely(ret)) {
 440                         goto end;
 441                 }
 442         }
 443
 444         /* Specific context */
 445         if (ec->payload_fc) {
 446                 field = bt_event_borrow_payload_field_const(event);
 447                 BT_ASSERT(field);
 448                 ret = write_struct_field(stream, (void *) ec->payload_fc,
 449                         field, true);
 450                 if (unlikely(ret)) {
 451                         goto end;
 452                 }
 453         }
 454
 455 end:
 456         return ret;
 457 }
 458
 459 static
 460 int write_packet_context(struct fs_sink_stream *stream)
 461 {
 462         int ret;
 463
 464         /* Packet total size */
 465         ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 466                 stream->packet_state.total_size, 8, 64, BYTE_ORDER);
 467         if (ret) {
 468                 goto end;
 469         }
 470
 471         /* Packet content size */
 472         ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 473                 stream->packet_state.content_size, 8, 64, BYTE_ORDER);
 474         if (ret) {
 475                 goto end;
 476         }
 477
 478         if (stream->sc->default_clock_class) {
 479                 /* Beginning time */
 480                 ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 481                         stream->packet_state.beginning_cs, 8, 64, BYTE_ORDER);
 482                 if (ret) {
 483                         goto end;
 484                 }
 485
 486                 /* End time */
 487                 ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 488                         stream->packet_state.end_cs, 8, 64, BYTE_ORDER);
 489                 if (ret) {
 490                         goto end;
 491                 }
 492         }
 493
 494         /* Discarded event counter */
 495         ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 496                 stream->packet_state.discarded_events_counter, 8, 64,
 497                 BYTE_ORDER);
 498         if (ret) {
 499                 goto end;
 500         }
 501
 502         /* Sequence number */
 503         ret = bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 504                 stream->packet_state.seq_num, 8, 64, BYTE_ORDER);
 505         if (ret) {
 506                 goto end;
 507         }
 508
 509         /* Other members */
 510         if (stream->sc->packet_context_fc) {
 511                 const bt_field *packet_context_field =
 512                         bt_packet_borrow_context_field_const(
 513                                 stream->packet_state.packet);
 514
 515                 BT_ASSERT(packet_context_field);
 516                 ret = write_struct_field(stream,
 517                         (void *) stream->sc->packet_context_fc,
 518                         packet_context_field, false);
 519                 if (ret) {
 520                         goto end;
 521                 }
 522         }
 523
 524 end:
 525         return ret;
 526 }
 527
 528 BT_HIDDEN
 529 int fs_sink_stream_open_packet(struct fs_sink_stream *stream,
 530                 const bt_clock_snapshot *cs, const bt_packet *packet)
 531 {
 532         int ret;
 533         uint64_t i;
 534
 535         BT_ASSERT(!stream->packet_state.is_open);
 536         bt_packet_put_ref(stream->packet_state.packet);
 537         stream->packet_state.packet = packet;
 538         bt_packet_get_ref(stream->packet_state.packet);
 539         if (cs) {
 540                 stream->packet_state.beginning_cs =
 541                         bt_clock_snapshot_get_value(cs);
 542         }
 543
 544         /* Open packet */
 545         ret = bt_ctfser_open_packet(&stream->ctfser);
 546         if (ret) {
 547                 /* bt_ctfser_open_packet() logs errors */
 548                 goto end;
 549         }
 550
 551         /* Packet header: magic */
 552         bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 553                 UINT64_C(0xc1fc1fc1), 8, 32, BYTE_ORDER);
 554
 555         /* Packet header: UUID */
 556         for (i = 0; i < BABELTRACE_UUID_LEN; i++) {
 557                 bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 558                         (uint64_t) stream->sc->tc->uuid[i], 8, 8, BYTE_ORDER);
 559         }
 560
 561         /* Packet header: stream class ID */
 562         bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 563                 bt_stream_class_get_id(stream->sc->ir_sc), 8, 64, BYTE_ORDER);
 564
 565         /* Packet header: stream ID */
 566         bt_ctfser_write_byte_aligned_unsigned_int(&stream->ctfser,
 567                 bt_stream_get_id(stream->ir_stream), 8, 64, BYTE_ORDER);
 568
 569         /* Save packet context's offset to rewrite it later */
 570         stream->packet_state.context_offset_bits =
 571                 bt_ctfser_get_offset_in_current_packet_bits(&stream->ctfser);
 572
 573         /* Write packet context just to advance to content (first event) */
 574         ret = write_packet_context(stream);
 575         if (ret) {
 576                 goto end;
 577         }
 578
 579         stream->packet_state.is_open = true;
 580
 581 end:
 582         return ret;
 583 }
 584
 585 BT_HIDDEN
 586 int fs_sink_stream_close_packet(struct fs_sink_stream *stream,
 587                 const bt_clock_snapshot *cs)
 588 {
 589         int ret;
 590
 591         BT_ASSERT(stream->packet_state.is_open);
 592
 593         if (cs) {
 594                 stream->packet_state.end_cs = bt_clock_snapshot_get_value(cs);
 595         }
 596
 597         stream->packet_state.content_size =
 598                 bt_ctfser_get_offset_in_current_packet_bits(&stream->ctfser);
 599         stream->packet_state.total_size =
 600                 (stream->packet_state.content_size + 7) & ~UINT64_C(7);
 601
 602         /* Rewrite packet context */
 603         bt_ctfser_set_offset_in_current_packet_bits(&stream->ctfser,
 604                 stream->packet_state.context_offset_bits);
 605         ret = write_packet_context(stream);
 606         if (ret) {
 607                 goto end;
 608         }
 609
 610         /* Close packet */
 611         bt_ctfser_close_current_packet(&stream->ctfser,
 612                 stream->packet_state.total_size / 8);
 613
 614         /* Partially copy current packet state to previous packet state */
 615         stream->prev_packet_state.end_cs = stream->packet_state.end_cs;
 616         stream->prev_packet_state.discarded_events_counter =
 617                 stream->packet_state.discarded_events_counter;
 618         stream->prev_packet_state.seq_num =
 619                 stream->packet_state.seq_num;
 620
 621         /* Reset current packet state */
 622         stream->packet_state.beginning_cs = UINT64_C(-1);
 623         stream->packet_state.end_cs = UINT64_C(-1);
 624         stream->packet_state.content_size = 0;
 625         stream->packet_state.total_size = 0;
 626         stream->packet_state.seq_num += 1;
 627         stream->packet_state.context_offset_bits = 0;
 628         stream->packet_state.is_open = false;
 629         BT_PACKET_PUT_REF_AND_RESET(stream->packet_state.packet);
 630
 631 end:
 632         return ret;
 633 }