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