formats/ctf/writer/stream.c

   1 /*
   2  * stream.c
   3  *
   4  * Babeltrace CTF Writer
   5  *
   6  * Copyright 2013 EfficiOS Inc.
   7  *
   8  * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
   9  *
  10  * Permission is hereby granted, free of charge, to any person obtaining a copy
  11  * of this software and associated documentation files (the "Software"), to deal
  12  * in the Software without restriction, including without limitation the rights
  13  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  14  * copies of the Software, and to permit persons to whom the Software is
  15  * furnished to do so, subject to the following conditions:
  16  *
  17  * The above copyright notice and this permission notice shall be included in
  18  * all copies or substantial portions of the Software.
  19  *
  20  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  21  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  22  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  23  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  24  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  25  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  26  * SOFTWARE.
  27  */
  28
  29 #include <babeltrace/ctf-writer/clock.h>
  30 #include <babeltrace/ctf-writer/clock-internal.h>
  31 #include <babeltrace/ctf-writer/event.h>
  32 #include <babeltrace/ctf-writer/event-internal.h>
  33 #include <babeltrace/ctf-writer/event-types-internal.h>
  34 #include <babeltrace/ctf-writer/event-fields-internal.h>
  35 #include <babeltrace/ctf-writer/stream.h>
  36 #include <babeltrace/ctf-writer/stream-internal.h>
  37 #include <babeltrace/ctf-writer/functor-internal.h>
  38 #include <babeltrace/compiler.h>
  39 #include <babeltrace/align.h>
  40
  41 static
  42 void bt_ctf_stream_destroy(struct bt_ctf_ref *ref);
  43 static
  44 void bt_ctf_stream_class_destroy(struct bt_ctf_ref *ref);
  45 static
  46 int init_event_header(struct bt_ctf_stream_class *stream_class,
  47                 enum bt_ctf_byte_order byte_order);
  48 static
  49 int init_packet_context(struct bt_ctf_stream_class *stream_class,
  50                 enum bt_ctf_byte_order byte_order);
  51 static
  52 int set_structure_field_integer(struct bt_ctf_field *, char *, uint64_t);
  53
  54 struct bt_ctf_stream_class *bt_ctf_stream_class_create(const char *name)
  55 {
  56         struct bt_ctf_stream_class *stream_class = NULL;
  57
  58         if (!name || !strlen(name)) {
  59                 goto error;
  60         }
  61
  62         stream_class = g_new0(struct bt_ctf_stream_class, 1);
  63         if (!stream_class) {
  64                 goto error;
  65         }
  66
  67         stream_class->name = g_string_new(name);
  68         stream_class->event_classes = g_ptr_array_new_with_free_func(
  69                 (GDestroyNotify)bt_ctf_event_class_put);
  70         if (!stream_class->event_classes) {
  71                 goto error_destroy;
  72         }
  73
  74         bt_ctf_ref_init(&stream_class->ref_count);
  75         return stream_class;
  76
  77 error_destroy:
  78         bt_ctf_stream_class_destroy(&stream_class->ref_count);
  79         stream_class = NULL;
  80 error:
  81         return stream_class;
  82 }
  83
  84 int bt_ctf_stream_class_set_clock(struct bt_ctf_stream_class *stream_class,
  85                 struct bt_ctf_clock *clock)
  86 {
  87         int ret = 0;
  88
  89         if (!stream_class || !clock || stream_class->frozen) {
  90                 ret = -1;
  91                 goto end;
  92         }
  93
  94         if (stream_class->clock) {
  95                 bt_ctf_clock_put(stream_class->clock);
  96         }
  97
  98         stream_class->clock = clock;
  99         bt_ctf_clock_get(clock);
 100 end:
 101         return ret;
 102 }
 103
 104 int bt_ctf_stream_class_add_event_class(
 105                 struct bt_ctf_stream_class *stream_class,
 106                 struct bt_ctf_event_class *event_class)
 107 {
 108         int ret = 0;
 109
 110         if (!stream_class || !event_class) {
 111                 ret = -1;
 112                 goto end;
 113         }
 114
 115         /* Check for duplicate event classes */
 116         struct search_query query = { .value = event_class, .found = 0 };
 117         g_ptr_array_foreach(stream_class->event_classes, value_exists, &query);
 118         if (query.found) {
 119                 ret = -1;
 120                 goto end;
 121         }
 122
 123         if (bt_ctf_event_class_set_id(event_class,
 124                 stream_class->next_event_id++)) {
 125                 /* The event is already associated to a stream class */
 126                 ret = -1;
 127                 goto end;
 128         }
 129
 130         bt_ctf_event_class_get(event_class);
 131         g_ptr_array_add(stream_class->event_classes, event_class);
 132 end:
 133         return ret;
 134 }
 135
 136 void bt_ctf_stream_class_get(struct bt_ctf_stream_class *stream_class)
 137 {
 138         if (!stream_class) {
 139                 return;
 140         }
 141
 142         bt_ctf_ref_get(&stream_class->ref_count);
 143 }
 144
 145 void bt_ctf_stream_class_put(struct bt_ctf_stream_class *stream_class)
 146 {
 147         if (!stream_class) {
 148                 return;
 149         }
 150
 151         bt_ctf_ref_put(&stream_class->ref_count, bt_ctf_stream_class_destroy);
 152 }
 153
 154 BT_HIDDEN
 155 void bt_ctf_stream_class_freeze(struct bt_ctf_stream_class *stream_class)
 156 {
 157         if (!stream_class) {
 158                 return;
 159         }
 160
 161         stream_class->frozen = 1;
 162         bt_ctf_clock_freeze(stream_class->clock);
 163         g_ptr_array_foreach(stream_class->event_classes,
 164                 (GFunc)bt_ctf_event_class_freeze, NULL);
 165 }
 166
 167 BT_HIDDEN
 168 int bt_ctf_stream_class_set_id(struct bt_ctf_stream_class *stream_class,
 169         uint32_t id)
 170 {
 171         int ret = 0;
 172
 173         if (!stream_class ||
 174                 (stream_class->id_set && (id != stream_class->id))) {
 175                 ret = -1;
 176                 goto end;
 177         }
 178
 179         stream_class->id = id;
 180         stream_class->id_set = 1;
 181 end:
 182         return ret;
 183 }
 184
 185 BT_HIDDEN
 186 int bt_ctf_stream_class_set_byte_order(struct bt_ctf_stream_class *stream_class,
 187         enum bt_ctf_byte_order byte_order)
 188 {
 189         int ret = 0;
 190
 191         ret = init_packet_context(stream_class, byte_order);
 192         if (ret) {
 193                 goto end;
 194         }
 195
 196         ret = init_event_header(stream_class, byte_order);
 197         if (ret) {
 198                 goto end;
 199         }
 200 end:
 201         return ret;
 202 }
 203
 204 BT_HIDDEN
 205 int bt_ctf_stream_class_serialize(struct bt_ctf_stream_class *stream_class,
 206                 struct metadata_context *context)
 207 {
 208         int ret = 0;
 209         size_t i;
 210
 211         g_string_assign(context->field_name, "");
 212         context->current_indentation_level = 1;
 213         if (!stream_class->id_set) {
 214                 ret = -1;
 215                 goto end;
 216         }
 217
 218         g_string_append_printf(context->string,
 219                 "stream {\n\tid = %" PRIu32 ";\n\tevent.header := ",
 220                 stream_class->id);
 221         ret = bt_ctf_field_type_serialize(stream_class->event_header_type,
 222                 context);
 223         if (ret) {
 224                 goto end;
 225         }
 226
 227         g_string_append(context->string, ";\n\n\tpacket.context := ");
 228         ret = bt_ctf_field_type_serialize(stream_class->packet_context_type,
 229                 context);
 230         if (ret) {
 231                 goto end;
 232         }
 233
 234         if (stream_class->event_context_type) {
 235                 g_string_append(context->string, ";\n\n\tevent.context := ");
 236                 ret = bt_ctf_field_type_serialize(
 237                         stream_class->event_context_type, context);
 238                 if (ret) {
 239                         goto end;
 240                 }
 241         }
 242
 243         g_string_append(context->string, ";\n};\n\n");
 244
 245         /* Assign this stream's ID to every event and serialize them */
 246         g_ptr_array_foreach(stream_class->event_classes,
 247                 (GFunc) bt_ctf_event_class_set_stream_id,
 248                 GUINT_TO_POINTER(stream_class->id));
 249         for (i = 0; i < stream_class->event_classes->len; i++) {
 250                 struct bt_ctf_event_class *event_class =
 251                         stream_class->event_classes->pdata[i];
 252
 253                 ret = bt_ctf_event_class_set_stream_id(event_class,
 254                         stream_class->id);
 255                 if (ret) {
 256                         goto end;
 257                 }
 258
 259                 ret = bt_ctf_event_class_serialize(event_class, context);
 260                 if (ret) {
 261                         goto end;
 262                 }
 263         }
 264 end:
 265         context->current_indentation_level = 0;
 266         return ret;
 267 }
 268
 269 BT_HIDDEN
 270 struct bt_ctf_stream *bt_ctf_stream_create(
 271                 struct bt_ctf_stream_class *stream_class)
 272 {
 273         struct bt_ctf_stream *stream = NULL;
 274
 275         if (!stream_class) {
 276                 goto end;
 277         }
 278
 279         stream = g_new0(struct bt_ctf_stream, 1);
 280         if (!stream) {
 281                 goto end;
 282         }
 283
 284         bt_ctf_ref_init(&stream->ref_count);
 285         stream->pos.fd = -1;
 286         stream->id = stream_class->next_stream_id++;
 287         stream->stream_class = stream_class;
 288         bt_ctf_stream_class_get(stream_class);
 289         bt_ctf_stream_class_freeze(stream_class);
 290         stream->events = g_ptr_array_new_with_free_func(
 291                 (GDestroyNotify)bt_ctf_event_put);
 292 end:
 293         return stream;
 294 }
 295
 296 BT_HIDDEN
 297 int bt_ctf_stream_set_flush_callback(struct bt_ctf_stream *stream,
 298         flush_func callback, void *data)
 299 {
 300         int ret = stream ? 0 : -1;
 301
 302         if (!stream) {
 303                 goto end;
 304         }
 305
 306         stream->flush.func = callback;
 307         stream->flush.data = data;
 308 end:
 309         return ret;
 310 }
 311
 312 BT_HIDDEN
 313 int bt_ctf_stream_set_fd(struct bt_ctf_stream *stream, int fd)
 314 {
 315         int ret = 0;
 316
 317         if (stream->pos.fd != -1) {
 318                 ret = -1;
 319                 goto end;
 320         }
 321
 322         ctf_init_pos(&stream->pos, NULL, fd, O_RDWR);
 323         stream->pos.fd = fd;
 324 end:
 325         return ret;
 326 }
 327
 328 void bt_ctf_stream_append_discarded_events(struct bt_ctf_stream *stream,
 329                 uint64_t event_count)
 330 {
 331         if (!stream) {
 332                 return;
 333         }
 334
 335         stream->events_discarded += event_count;
 336 }
 337
 338 int bt_ctf_stream_append_event(struct bt_ctf_stream *stream,
 339                 struct bt_ctf_event *event)
 340 {
 341         int ret = 0;
 342         uint64_t timestamp;
 343
 344         if (!stream || !event) {
 345                 ret = -1;
 346                 goto end;
 347         }
 348
 349         ret = bt_ctf_event_validate(event);
 350         if (ret) {
 351                 goto end;
 352         }
 353
 354         timestamp = bt_ctf_clock_get_time(stream->stream_class->clock);
 355         ret = bt_ctf_event_set_timestamp(event, timestamp);
 356         if (ret) {
 357                 goto end;
 358         }
 359
 360         bt_ctf_event_get(event);
 361         g_ptr_array_add(stream->events, event);
 362 end:
 363         return ret;
 364 }
 365
 366 int bt_ctf_stream_flush(struct bt_ctf_stream *stream)
 367 {
 368         int ret = 0;
 369         size_t i;
 370         uint64_t timestamp_begin, timestamp_end;
 371         struct bt_ctf_stream_class *stream_class;
 372         struct bt_ctf_field *integer = NULL;
 373         struct ctf_stream_pos packet_context_pos;
 374
 375         if (!stream) {
 376                 ret = -1;
 377                 goto end;
 378         }
 379
 380         if (!stream->events->len) {
 381                 goto end;
 382         }
 383
 384         if (stream->flush.func) {
 385                 stream->flush.func(stream, stream->flush.data);
 386         }
 387
 388         stream_class = stream->stream_class;
 389         timestamp_begin = ((struct bt_ctf_event *) g_ptr_array_index(
 390                 stream->events, 0))->timestamp;
 391         timestamp_end = ((struct bt_ctf_event *) g_ptr_array_index(
 392                 stream->events, stream->events->len - 1))->timestamp;
 393         ret = set_structure_field_integer(stream_class->packet_context,
 394                 "timestamp_begin", timestamp_begin);
 395         if (ret) {
 396                 goto end;
 397         }
 398
 399         ret = set_structure_field_integer(stream_class->packet_context,
 400                 "timestamp_end", timestamp_end);
 401         if (ret) {
 402                 goto end;
 403         }
 404
 405         ret = set_structure_field_integer(stream_class->packet_context,
 406                 "events_discarded", stream->events_discarded);
 407         if (ret) {
 408                 goto end;
 409         }
 410
 411         ret = set_structure_field_integer(stream_class->packet_context,
 412                 "content_size", UINT64_MAX);
 413         if (ret) {
 414                 goto end;
 415         }
 416
 417         ret = set_structure_field_integer(stream_class->packet_context,
 418                 "packet_size", UINT64_MAX);
 419         if (ret) {
 420                 goto end;
 421         }
 422
 423         /* Write packet context */
 424         memcpy(&packet_context_pos, &stream->pos,
 425                sizeof(struct ctf_stream_pos));
 426         ret = bt_ctf_field_serialize(stream_class->packet_context,
 427                 &stream->pos);
 428         if (ret) {
 429                 goto end;
 430         }
 431
 432         for (i = 0; i < stream->events->len; i++) {
 433                 struct bt_ctf_event *event = g_ptr_array_index(
 434                         stream->events, i);
 435                 uint32_t event_id = bt_ctf_event_class_get_id(
 436                         event->event_class);
 437                 uint64_t timestamp = bt_ctf_event_get_timestamp(event);
 438
 439                 ret = set_structure_field_integer(stream_class->event_header,
 440                         "id", event_id);
 441                 if (ret) {
 442                         goto end;
 443                 }
 444                 ret = set_structure_field_integer(stream_class->event_header,
 445                         "timestamp", timestamp);
 446                 if (ret) {
 447                         goto end;
 448                 }
 449
 450                 /* Write event header */
 451                 ret = bt_ctf_field_serialize(stream_class->event_header,
 452                         &stream->pos);
 453                 if (ret) {
 454                         goto end;
 455                 }
 456
 457                 /* Write event content */
 458                 ret = bt_ctf_event_serialize(event, &stream->pos);
 459                 if (ret) {
 460                         goto end;
 461                 }
 462         }
 463
 464         /*
 465          * Update the packet total size and content size and overwrite the
 466          * packet context.
 467          * Copy base_mma as the packet may have been remapped (e.g. when a
 468          * packet is resized).
 469          */
 470         packet_context_pos.base_mma = stream->pos.base_mma;
 471         ret = set_structure_field_integer(stream_class->packet_context,
 472                 "content_size", stream->pos.offset);
 473         if (ret) {
 474                 goto end;
 475         }
 476
 477         ret = set_structure_field_integer(stream_class->packet_context,
 478                 "packet_size", stream->pos.packet_size);
 479         if (ret) {
 480                 goto end;
 481         }
 482
 483         ret = bt_ctf_field_serialize(stream_class->packet_context,
 484                 &packet_context_pos);
 485         if (ret) {
 486                 goto end;
 487         }
 488
 489         g_ptr_array_set_size(stream->events, 0);
 490         stream->flushed_packet_count++;
 491 end:
 492         bt_ctf_field_put(integer);
 493         return ret;
 494 }
 495
 496 void bt_ctf_stream_get(struct bt_ctf_stream *stream)
 497 {
 498         if (!stream) {
 499                 return;
 500         }
 501
 502         bt_ctf_ref_get(&stream->ref_count);
 503 }
 504
 505 void bt_ctf_stream_put(struct bt_ctf_stream *stream)
 506 {
 507         if (!stream) {
 508                 return;
 509         }
 510
 511         bt_ctf_ref_put(&stream->ref_count, bt_ctf_stream_destroy);
 512 }
 513
 514 static
 515 void bt_ctf_stream_destroy(struct bt_ctf_ref *ref)
 516 {
 517         struct bt_ctf_stream *stream;
 518
 519         if (!ref) {
 520                 return;
 521         }
 522
 523         stream = container_of(ref, struct bt_ctf_stream, ref_count);
 524         ctf_fini_pos(&stream->pos);
 525         if (close(stream->pos.fd)) {
 526                 perror("close");
 527         }
 528         bt_ctf_stream_class_put(stream->stream_class);
 529         g_ptr_array_free(stream->events, TRUE);
 530         g_free(stream);
 531 }
 532
 533 static
 534 void bt_ctf_stream_class_destroy(struct bt_ctf_ref *ref)
 535 {
 536         struct bt_ctf_stream_class *stream_class;
 537
 538         if (!ref) {
 539                 return;
 540         }
 541
 542         stream_class = container_of(ref, struct bt_ctf_stream_class, ref_count);
 543         bt_ctf_clock_put(stream_class->clock);
 544
 545         if (stream_class->event_classes) {
 546                 g_ptr_array_free(stream_class->event_classes, TRUE);
 547         }
 548
 549         if (stream_class->name) {
 550                 g_string_free(stream_class->name, TRUE);
 551         }
 552
 553         bt_ctf_field_type_put(stream_class->event_header_type);
 554         bt_ctf_field_put(stream_class->event_header);
 555         bt_ctf_field_type_put(stream_class->packet_context_type);
 556         bt_ctf_field_put(stream_class->packet_context);
 557         bt_ctf_field_type_put(stream_class->event_context_type);
 558         bt_ctf_field_put(stream_class->event_context);
 559         g_free(stream_class);
 560 }
 561
 562 static
 563 int init_event_header(struct bt_ctf_stream_class *stream_class,
 564                 enum bt_ctf_byte_order byte_order)
 565 {
 566         int ret = 0;
 567         struct bt_ctf_field_type *event_header_type =
 568                 bt_ctf_field_type_structure_create();
 569         struct bt_ctf_field_type *_uint32_t =
 570                 get_field_type(FIELD_TYPE_ALIAS_UINT32_T);
 571         struct bt_ctf_field_type *_uint64_t =
 572                 get_field_type(FIELD_TYPE_ALIAS_UINT64_T);
 573
 574         if (!event_header_type) {
 575                 ret = -1;
 576                 goto end;
 577         }
 578
 579         ret = bt_ctf_field_type_set_byte_order(_uint32_t, byte_order);
 580         if (ret) {
 581                 goto end;
 582         }
 583
 584         ret = bt_ctf_field_type_set_byte_order(_uint64_t, byte_order);
 585         if (ret) {
 586                 goto end;
 587         }
 588
 589         ret = bt_ctf_field_type_structure_add_field(event_header_type,
 590                 _uint32_t, "id");
 591         if (ret) {
 592                 goto end;
 593         }
 594
 595         ret = bt_ctf_field_type_structure_add_field(event_header_type,
 596                 _uint64_t, "timestamp");
 597         if (ret) {
 598                 goto end;
 599         }
 600
 601         stream_class->event_header_type = event_header_type;
 602         stream_class->event_header = bt_ctf_field_create(
 603                 stream_class->event_header_type);
 604         if (!stream_class->event_header) {
 605                 ret = -1;
 606         }
 607 end:
 608         if (ret) {
 609                 bt_ctf_field_type_put(event_header_type);
 610         }
 611
 612         bt_ctf_field_type_put(_uint32_t);
 613         bt_ctf_field_type_put(_uint64_t);
 614         return ret;
 615 }
 616
 617 static
 618 int init_packet_context(struct bt_ctf_stream_class *stream_class,
 619                 enum bt_ctf_byte_order byte_order)
 620 {
 621         int ret = 0;
 622         struct bt_ctf_field_type *packet_context_type =
 623                 bt_ctf_field_type_structure_create();
 624         struct bt_ctf_field_type *_uint64_t =
 625                 get_field_type(FIELD_TYPE_ALIAS_UINT64_T);
 626
 627         if (!packet_context_type) {
 628                 ret = -1;
 629                 goto end;
 630         }
 631
 632         /*
 633          * We create a stream packet context as proposed in the CTF
 634          * specification.
 635          */
 636         ret = bt_ctf_field_type_set_byte_order(_uint64_t, byte_order);
 637         if (ret) {
 638                 goto end;
 639         }
 640
 641         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 642                 _uint64_t, "timestamp_begin");
 643         if (ret) {
 644                 goto end;
 645         }
 646
 647         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 648                 _uint64_t, "timestamp_end");
 649         if (ret) {
 650                 goto end;
 651         }
 652
 653         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 654                 _uint64_t, "content_size");
 655         if (ret) {
 656                 goto end;
 657         }
 658
 659         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 660                 _uint64_t, "packet_size");
 661         if (ret) {
 662                 goto end;
 663         }
 664
 665         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 666                 _uint64_t, "events_discarded");
 667         if (ret) {
 668                 goto end;
 669         }
 670
 671         stream_class->packet_context_type = packet_context_type;
 672         stream_class->packet_context = bt_ctf_field_create(packet_context_type);
 673         if (!stream_class->packet_context) {
 674                 ret = -1;
 675         }
 676 end:
 677         if (ret) {
 678                 bt_ctf_field_type_put(packet_context_type);
 679                 goto end;
 680         }
 681
 682         bt_ctf_field_type_put(_uint64_t);
 683         return ret;
 684 }
 685
 686 static
 687 int set_structure_field_integer(struct bt_ctf_field *structure, char *name,
 688                 uint64_t value)
 689 {
 690         int ret = 0;
 691
 692         struct bt_ctf_field *integer =
 693                 bt_ctf_field_structure_get_field(structure, name);
 694         if (!integer) {
 695                 ret = -1;
 696                 goto end;
 697         }
 698
 699         ret = bt_ctf_field_unsigned_integer_set_value(integer, value);
 700 end:
 701         bt_ctf_field_put(integer);
 702         return ret;
 703 }