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