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 = stream->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         timestamp_begin = ((struct bt_ctf_event *) g_ptr_array_index(
 389                 stream->events, 0))->timestamp;
 390         timestamp_end = ((struct bt_ctf_event *) g_ptr_array_index(
 391                 stream->events, stream->events->len - 1))->timestamp;
 392         ret = set_structure_field_integer(stream_class->packet_context,
 393                 "timestamp_begin", timestamp_begin);
 394         if (ret) {
 395                 goto end;
 396         }
 397
 398         ret = set_structure_field_integer(stream_class->packet_context,
 399                 "timestamp_end", timestamp_end);
 400         if (ret) {
 401                 goto end;
 402         }
 403
 404         ret = set_structure_field_integer(stream_class->packet_context,
 405                 "events_discarded", stream->events_discarded);
 406         if (ret) {
 407                 goto end;
 408         }
 409
 410         ret = set_structure_field_integer(stream_class->packet_context,
 411                 "content_size", UINT64_MAX);
 412         if (ret) {
 413                 goto end;
 414         }
 415
 416         ret = set_structure_field_integer(stream_class->packet_context,
 417                 "packet_size", UINT64_MAX);
 418         if (ret) {
 419                 goto end;
 420         }
 421
 422         /* Write packet context */
 423         memcpy(&packet_context_pos, &stream->pos,
 424                sizeof(struct ctf_stream_pos));
 425         ret = bt_ctf_field_serialize(stream_class->packet_context,
 426                 &stream->pos);
 427         if (ret) {
 428                 goto end;
 429         }
 430
 431         for (i = 0; i < stream->events->len; i++) {
 432                 struct bt_ctf_event *event = g_ptr_array_index(
 433                         stream->events, i);
 434                 uint32_t event_id = bt_ctf_event_class_get_id(
 435                         event->event_class);
 436                 uint64_t timestamp = bt_ctf_event_get_timestamp(event);
 437
 438                 ret = set_structure_field_integer(stream_class->event_header,
 439                         "id", event_id);
 440                 if (ret) {
 441                         goto end;
 442                 }
 443                 ret = set_structure_field_integer(stream_class->event_header,
 444                         "timestamp", timestamp);
 445                 if (ret) {
 446                         goto end;
 447                 }
 448
 449                 /* Write event header */
 450                 ret = bt_ctf_field_serialize(stream_class->event_header,
 451                         &stream->pos);
 452                 if (ret) {
 453                         goto end;
 454                 }
 455
 456                 /* Write event content */
 457                 ret = bt_ctf_event_serialize(event, &stream->pos);
 458                 if (ret) {
 459                         goto end;
 460                 }
 461         }
 462
 463         /*
 464          * Update the packet total size and content size and overwrite the
 465          * packet context.
 466          */
 467         ret = set_structure_field_integer(stream_class->packet_context,
 468                 "content_size", stream->pos.offset);
 469         if (ret) {
 470                 goto end;
 471         }
 472
 473         ret = set_structure_field_integer(stream_class->packet_context,
 474                 "packet_size", stream->pos.packet_size);
 475         if (ret) {
 476                 goto end;
 477         }
 478
 479         ret = bt_ctf_field_serialize(stream_class->packet_context,
 480                 &packet_context_pos);
 481         if (ret) {
 482                 goto end;
 483         }
 484
 485         g_ptr_array_set_size(stream->events, 0);
 486         stream->flushed_packet_count++;
 487 end:
 488         bt_ctf_field_put(integer);
 489         return ret;
 490 }
 491
 492 void bt_ctf_stream_get(struct bt_ctf_stream *stream)
 493 {
 494         if (!stream) {
 495                 return;
 496         }
 497
 498         bt_ctf_ref_get(&stream->ref_count);
 499 }
 500
 501 void bt_ctf_stream_put(struct bt_ctf_stream *stream)
 502 {
 503         if (!stream) {
 504                 return;
 505         }
 506
 507         bt_ctf_ref_put(&stream->ref_count, bt_ctf_stream_destroy);
 508 }
 509
 510 static
 511 void bt_ctf_stream_destroy(struct bt_ctf_ref *ref)
 512 {
 513         struct bt_ctf_stream *stream;
 514
 515         if (!ref) {
 516                 return;
 517         }
 518
 519         stream = container_of(ref, struct bt_ctf_stream, ref_count);
 520         ctf_fini_pos(&stream->pos);
 521         close(stream->pos.fd);
 522         bt_ctf_stream_class_put(stream->stream_class);
 523         g_ptr_array_free(stream->events, TRUE);
 524         g_free(stream);
 525 }
 526
 527 static
 528 void bt_ctf_stream_class_destroy(struct bt_ctf_ref *ref)
 529 {
 530         struct bt_ctf_stream_class *stream_class;
 531
 532         if (!ref) {
 533                 return;
 534         }
 535
 536         stream_class = container_of(ref, struct bt_ctf_stream_class, ref_count);
 537         bt_ctf_clock_put(stream_class->clock);
 538
 539         if (stream_class->event_classes) {
 540                 g_ptr_array_free(stream_class->event_classes, TRUE);
 541         }
 542
 543         if (stream_class->name) {
 544                 g_string_free(stream_class->name, TRUE);
 545         }
 546
 547         bt_ctf_field_type_put(stream_class->event_header_type);
 548         bt_ctf_field_put(stream_class->event_header);
 549         bt_ctf_field_type_put(stream_class->packet_context_type);
 550         bt_ctf_field_put(stream_class->packet_context);
 551         bt_ctf_field_type_put(stream_class->event_context_type);
 552         bt_ctf_field_put(stream_class->event_context);
 553         g_free(stream_class);
 554 }
 555
 556 static
 557 int init_event_header(struct bt_ctf_stream_class *stream_class,
 558                 enum bt_ctf_byte_order byte_order)
 559 {
 560         int ret = 0;
 561         struct bt_ctf_field_type *event_header_type =
 562                 bt_ctf_field_type_structure_create();
 563         struct bt_ctf_field_type *_uint32_t =
 564                 get_field_type(FIELD_TYPE_ALIAS_UINT32_T);
 565         struct bt_ctf_field_type *_uint64_t =
 566                 get_field_type(FIELD_TYPE_ALIAS_UINT64_T);
 567
 568         if (!event_header_type) {
 569                 ret = -1;
 570                 goto end;
 571         }
 572
 573         ret = bt_ctf_field_type_set_byte_order(_uint32_t, byte_order);
 574         if (ret) {
 575                 goto end;
 576         }
 577
 578         ret = bt_ctf_field_type_set_byte_order(_uint64_t, byte_order);
 579         if (ret) {
 580                 goto end;
 581         }
 582
 583         ret = bt_ctf_field_type_structure_add_field(event_header_type,
 584                 _uint32_t, "id");
 585         if (ret) {
 586                 goto end;
 587         }
 588
 589         ret = bt_ctf_field_type_structure_add_field(event_header_type,
 590                 _uint64_t, "timestamp");
 591         if (ret) {
 592                 goto end;
 593         }
 594
 595         stream_class->event_header_type = event_header_type;
 596         stream_class->event_header = bt_ctf_field_create(
 597                 stream_class->event_header_type);
 598         if (!stream_class->event_header) {
 599                 ret = -1;
 600         }
 601 end:
 602         if (ret) {
 603                 bt_ctf_field_type_put(event_header_type);
 604         }
 605
 606         bt_ctf_field_type_put(_uint32_t);
 607         bt_ctf_field_type_put(_uint64_t);
 608         return ret;
 609 }
 610
 611 static
 612 int init_packet_context(struct bt_ctf_stream_class *stream_class,
 613                 enum bt_ctf_byte_order byte_order)
 614 {
 615         int ret = 0;
 616         struct bt_ctf_field_type *packet_context_type =
 617                 bt_ctf_field_type_structure_create();
 618         struct bt_ctf_field_type *_uint64_t =
 619                 get_field_type(FIELD_TYPE_ALIAS_UINT64_T);
 620
 621         if (!packet_context_type) {
 622                 ret = -1;
 623                 goto end;
 624         }
 625
 626         /*
 627          * We create a stream packet context as proposed in the CTF
 628          * specification.
 629          */
 630         ret = bt_ctf_field_type_set_byte_order(_uint64_t, byte_order);
 631         if (ret) {
 632                 goto end;
 633         }
 634
 635         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 636                 _uint64_t, "timestamp_begin");
 637         if (ret) {
 638                 goto end;
 639         }
 640
 641         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 642                 _uint64_t, "timestamp_end");
 643         if (ret) {
 644                 goto end;
 645         }
 646
 647         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 648                 _uint64_t, "content_size");
 649         if (ret) {
 650                 goto end;
 651         }
 652
 653         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 654                 _uint64_t, "packet_size");
 655         if (ret) {
 656                 goto end;
 657         }
 658
 659         ret = bt_ctf_field_type_structure_add_field(packet_context_type,
 660                 _uint64_t, "events_discarded");
 661         if (ret) {
 662                 goto end;
 663         }
 664
 665         stream_class->packet_context_type = packet_context_type;
 666         stream_class->packet_context = bt_ctf_field_create(packet_context_type);
 667         if (!stream_class->packet_context) {
 668                 ret = -1;
 669         }
 670 end:
 671         if (ret) {
 672                 bt_ctf_field_type_put(packet_context_type);
 673                 goto end;
 674         }
 675
 676         bt_ctf_field_type_put(_uint64_t);
 677         return ret;
 678 }
 679
 680 static
 681 int set_structure_field_integer(struct bt_ctf_field *structure, char *name,
 682                 uint64_t value)
 683 {
 684         int ret = 0;
 685
 686         struct bt_ctf_field *integer =
 687                 bt_ctf_field_structure_get_field(structure, name);
 688         if (!integer) {
 689                 ret = -1;
 690                 goto end;
 691         }
 692
 693         ret = bt_ctf_field_unsigned_integer_set_value(integer, value);
 694 end:
 695         bt_ctf_field_put(integer);
 696         return ret;
 697 }