formats/ctf/ir/stream.c

   1 /*
   2  * stream.c
   3  *
   4  * Babeltrace CTF IR - Stream
   5  *
   6  * Copyright 2013, 2014 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   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-ir/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-ir/stream.h>
  36 #include <babeltrace/ctf-ir/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 int set_structure_field_integer(struct bt_ctf_field *, char *, uint64_t);
  46
  47 BT_HIDDEN
  48 struct bt_ctf_stream *bt_ctf_stream_create(
  49                 struct bt_ctf_stream_class *stream_class)
  50 {
  51         int ret;
  52         struct bt_ctf_stream *stream = NULL;
  53
  54         if (!stream_class) {
  55                 goto end;
  56         }
  57
  58         stream = g_new0(struct bt_ctf_stream, 1);
  59         if (!stream) {
  60                 goto end;
  61         }
  62
  63         bt_ctf_ref_init(&stream->ref_count);
  64         stream->packet_context = bt_ctf_field_create(
  65                 stream_class->packet_context_type);
  66         if (!stream->packet_context) {
  67                 goto error_destroy;
  68         }
  69
  70         /* Initialize events_discarded*/
  71         ret = set_structure_field_integer(stream->packet_context,
  72                 "events_discarded", 0);
  73         if (ret) {
  74                 goto error_destroy;
  75         }
  76
  77         stream->pos.fd = -1;
  78         stream->id = stream_class->next_stream_id++;
  79         stream->stream_class = stream_class;
  80         bt_ctf_stream_class_get(stream_class);
  81         bt_ctf_stream_class_freeze(stream_class);
  82         stream->events = g_ptr_array_new_with_free_func(
  83                 (GDestroyNotify)bt_ctf_event_put);
  84 end:
  85         return stream;
  86 error_destroy:
  87         bt_ctf_stream_destroy(&stream->ref_count);
  88         return NULL;
  89 }
  90
  91 BT_HIDDEN
  92 int bt_ctf_stream_set_flush_callback(struct bt_ctf_stream *stream,
  93         flush_func callback, void *data)
  94 {
  95         int ret = stream ? 0 : -1;
  96
  97         if (!stream) {
  98                 goto end;
  99         }
 100
 101         stream->flush.func = callback;
 102         stream->flush.data = data;
 103 end:
 104         return ret;
 105 }
 106
 107 BT_HIDDEN
 108 int bt_ctf_stream_set_fd(struct bt_ctf_stream *stream, int fd)
 109 {
 110         int ret = 0;
 111
 112         if (stream->pos.fd != -1) {
 113                 ret = -1;
 114                 goto end;
 115         }
 116
 117         ctf_init_pos(&stream->pos, NULL, fd, O_RDWR);
 118         stream->pos.fd = fd;
 119 end:
 120         return ret;
 121 }
 122
 123 int bt_ctf_stream_get_discarded_events_count(
 124                 struct bt_ctf_stream *stream, uint64_t *count)
 125 {
 126         int64_t ret = 0;
 127         int field_signed;
 128         struct bt_ctf_field *events_discarded_field = NULL;
 129         struct bt_ctf_field_type *events_discarded_field_type = NULL;
 130
 131         if (!stream || !count || !stream->packet_context) {
 132                 ret = -1;
 133                 goto end;
 134         }
 135
 136         events_discarded_field = bt_ctf_field_structure_get_field(
 137                 stream->packet_context, "events_discarded");
 138         if (!events_discarded_field) {
 139                 ret = -1;
 140                 goto end;
 141         }
 142
 143         events_discarded_field_type = bt_ctf_field_get_type(
 144                 events_discarded_field);
 145         if (!events_discarded_field_type) {
 146                 ret = -1;
 147                 goto end;
 148         }
 149
 150         field_signed = bt_ctf_field_type_integer_get_signed(
 151                 events_discarded_field_type);
 152         if (field_signed < 0) {
 153                 ret = field_signed;
 154                 goto end;
 155         }
 156
 157         if (field_signed) {
 158                 int64_t signed_count;
 159
 160                 ret = bt_ctf_field_signed_integer_get_value(
 161                         events_discarded_field, &signed_count);
 162                 if (ret) {
 163                         goto end;
 164                 }
 165                 if (signed_count < 0) {
 166                         /* Invalid value */
 167                         ret = -1;
 168                         goto end;
 169                 }
 170                 *count = (uint64_t) signed_count;
 171         } else {
 172                 ret = bt_ctf_field_unsigned_integer_get_value(
 173                         events_discarded_field, count);
 174                 if (ret) {
 175                         goto end;
 176                 }
 177         }
 178 end:
 179         if (events_discarded_field) {
 180                 bt_ctf_field_put(events_discarded_field);
 181         }
 182         if (events_discarded_field_type) {
 183                 bt_ctf_field_type_put(events_discarded_field_type);
 184         }
 185         return ret;
 186 }
 187
 188 void bt_ctf_stream_append_discarded_events(struct bt_ctf_stream *stream,
 189                 uint64_t event_count)
 190 {
 191         int ret;
 192         int field_signed;
 193         uint64_t previous_count;
 194         uint64_t new_count;
 195         struct bt_ctf_field *events_discarded_field = NULL;
 196         struct bt_ctf_field_type *events_discarded_field_type = NULL;
 197
 198         if (!stream || !stream->packet_context) {
 199                 goto end;
 200         }
 201
 202         ret = bt_ctf_stream_get_discarded_events_count(stream,
 203                 &previous_count);
 204         if (ret) {
 205                 goto end;
 206         }
 207
 208         events_discarded_field = bt_ctf_field_structure_get_field(
 209                 stream->packet_context, "events_discarded");
 210         if (!events_discarded_field) {
 211                 goto end;
 212         }
 213
 214         events_discarded_field_type = bt_ctf_field_get_type(
 215                 events_discarded_field);
 216         if (!events_discarded_field_type) {
 217                 goto end;
 218         }
 219
 220         field_signed = bt_ctf_field_type_integer_get_signed(
 221                 events_discarded_field_type);
 222         if (field_signed < 0) {
 223                 goto end;
 224         }
 225
 226         new_count = previous_count + event_count;
 227         if (field_signed) {
 228                 ret = bt_ctf_field_signed_integer_set_value(
 229                         events_discarded_field, (int64_t) new_count);
 230                 if (ret) {
 231                         goto end;
 232                 }
 233         } else {
 234                 ret = bt_ctf_field_unsigned_integer_set_value(
 235                         events_discarded_field, new_count);
 236                 if (ret) {
 237                         goto end;
 238                 }
 239         }
 240
 241 end:
 242         if (events_discarded_field) {
 243                 bt_ctf_field_put(events_discarded_field);
 244         }
 245         if (events_discarded_field_type) {
 246                 bt_ctf_field_type_put(events_discarded_field_type);
 247         }
 248 }
 249
 250 int bt_ctf_stream_append_event(struct bt_ctf_stream *stream,
 251                 struct bt_ctf_event *event)
 252 {
 253         int ret = 0;
 254         uint64_t timestamp;
 255
 256         if (!stream || !event) {
 257                 ret = -1;
 258                 goto end;
 259         }
 260
 261         ret = bt_ctf_event_validate(event);
 262         if (ret) {
 263                 goto end;
 264         }
 265
 266         timestamp = bt_ctf_clock_get_time(stream->stream_class->clock);
 267         ret = bt_ctf_event_set_timestamp(event, timestamp);
 268         if (ret) {
 269                 goto end;
 270         }
 271
 272         bt_ctf_event_get(event);
 273         g_ptr_array_add(stream->events, event);
 274 end:
 275         return ret;
 276 }
 277
 278 struct bt_ctf_field *bt_ctf_stream_get_packet_context(
 279                 struct bt_ctf_stream *stream)
 280 {
 281         struct bt_ctf_field *packet_context = NULL;
 282
 283         if (!stream) {
 284                 goto end;
 285         }
 286
 287         packet_context = stream->packet_context;
 288 end:
 289         if (packet_context) {
 290                 bt_ctf_field_get(packet_context);
 291         }
 292         return packet_context;
 293 }
 294
 295 int bt_ctf_stream_set_packet_context(struct bt_ctf_stream *stream,
 296                 struct bt_ctf_field *field)
 297 {
 298         int ret = 0;
 299         struct bt_ctf_field_type *field_type;
 300
 301         if (!stream || !field) {
 302                 ret = -1;
 303                 goto end;
 304         }
 305
 306         field_type = bt_ctf_field_get_type(field);
 307         if (field_type != stream->stream_class->packet_context_type) {
 308                 ret = -1;
 309                 goto end;
 310         }
 311
 312         bt_ctf_field_type_put(field_type);
 313         bt_ctf_field_get(field);
 314         bt_ctf_field_put(stream->packet_context);
 315         stream->packet_context = field;
 316 end:
 317         return ret;
 318 }
 319
 320 int bt_ctf_stream_flush(struct bt_ctf_stream *stream)
 321 {
 322         int ret = 0;
 323         size_t i;
 324         uint64_t timestamp_begin, timestamp_end, events_discarded;
 325         struct bt_ctf_stream_class *stream_class;
 326         struct bt_ctf_field *integer = NULL;
 327         struct ctf_stream_pos packet_context_pos;
 328
 329         if (!stream || stream->pos.fd < 0) {
 330                 /*
 331                  * Stream does not have an associated fd. It is,
 332                  * therefore, not a stream being used to write events.
 333                  */
 334                 ret = -1;
 335                 goto end;
 336         }
 337
 338         if (!stream->events->len) {
 339                 goto end;
 340         }
 341
 342         if (stream->flush.func) {
 343                 stream->flush.func(stream, stream->flush.data);
 344         }
 345
 346         stream_class = stream->stream_class;
 347         timestamp_begin = ((struct bt_ctf_event *) g_ptr_array_index(
 348                 stream->events, 0))->timestamp;
 349         timestamp_end = ((struct bt_ctf_event *) g_ptr_array_index(
 350                 stream->events, stream->events->len - 1))->timestamp;
 351
 352         /* Set the default context attributes if present and unset. */
 353         ret = set_structure_field_integer(stream->packet_context,
 354                 "timestamp_begin", timestamp_begin);
 355         if (ret) {
 356                 goto end;
 357         }
 358
 359         ret = set_structure_field_integer(stream->packet_context,
 360                 "timestamp_end", timestamp_end);
 361         if (ret) {
 362                 goto end;
 363         }
 364
 365         ret = set_structure_field_integer(stream->packet_context,
 366                 "content_size", UINT64_MAX);
 367         if (ret) {
 368                 goto end;
 369         }
 370
 371         ret = set_structure_field_integer(stream->packet_context,
 372                 "packet_size", UINT64_MAX);
 373         if (ret) {
 374                 goto end;
 375         }
 376
 377         /* Write packet context */
 378         memcpy(&packet_context_pos, &stream->pos,
 379                sizeof(struct ctf_stream_pos));
 380         ret = bt_ctf_field_serialize(stream->packet_context,
 381                 &stream->pos);
 382         if (ret) {
 383                 goto end;
 384         }
 385
 386         ret = bt_ctf_stream_get_discarded_events_count(stream,
 387                 &events_discarded);
 388         if (ret) {
 389                 goto end;
 390         }
 391
 392         /* Unset the packet context's fields. */
 393         ret = bt_ctf_field_reset(stream->packet_context);
 394         if (ret) {
 395                 goto end;
 396         }
 397
 398         /* Set the previous number of discarded events. */
 399         ret = set_structure_field_integer(stream->packet_context,
 400                 "events_discarded", events_discarded);
 401         if (ret) {
 402                 goto end;
 403         }
 404
 405         for (i = 0; i < stream->events->len; i++) {
 406                 struct bt_ctf_event *event = g_ptr_array_index(
 407                         stream->events, i);
 408                 uint32_t event_id = bt_ctf_event_class_get_id(
 409                         event->event_class);
 410                 uint64_t timestamp = bt_ctf_event_get_timestamp(event);
 411
 412                 ret = bt_ctf_field_reset(stream_class->event_header);
 413                 if (ret) {
 414                         goto end;
 415                 }
 416
 417                 ret = set_structure_field_integer(stream_class->event_header,
 418                         "id", event_id);
 419                 if (ret) {
 420                         goto end;
 421                 }
 422                 ret = set_structure_field_integer(stream_class->event_header,
 423                         "timestamp", timestamp);
 424                 if (ret) {
 425                         goto end;
 426                 }
 427
 428                 /* Write event header */
 429                 ret = bt_ctf_field_serialize(stream_class->event_header,
 430                         &stream->pos);
 431                 if (ret) {
 432                         goto end;
 433                 }
 434
 435                 /* Write event content */
 436                 ret = bt_ctf_event_serialize(event, &stream->pos);
 437                 if (ret) {
 438                         goto end;
 439                 }
 440         }
 441
 442         /*
 443          * Update the packet total size and content size and overwrite the
 444          * packet context.
 445          * Copy base_mma as the packet may have been remapped (e.g. when a
 446          * packet is resized).
 447          */
 448         packet_context_pos.base_mma = stream->pos.base_mma;
 449         ret = set_structure_field_integer(stream->packet_context,
 450                 "content_size", stream->pos.offset);
 451         if (ret) {
 452                 goto end;
 453         }
 454
 455         ret = set_structure_field_integer(stream->packet_context,
 456                 "packet_size", stream->pos.packet_size);
 457         if (ret) {
 458                 goto end;
 459         }
 460
 461         ret = bt_ctf_field_serialize(stream->packet_context,
 462                 &packet_context_pos);
 463         if (ret) {
 464                 goto end;
 465         }
 466
 467         g_ptr_array_set_size(stream->events, 0);
 468         stream->flushed_packet_count++;
 469 end:
 470         bt_ctf_field_put(integer);
 471         return ret;
 472 }
 473
 474 void bt_ctf_stream_get(struct bt_ctf_stream *stream)
 475 {
 476         if (!stream) {
 477                 return;
 478         }
 479
 480         bt_ctf_ref_get(&stream->ref_count);
 481 }
 482
 483 void bt_ctf_stream_put(struct bt_ctf_stream *stream)
 484 {
 485         if (!stream) {
 486                 return;
 487         }
 488
 489         bt_ctf_ref_put(&stream->ref_count, bt_ctf_stream_destroy);
 490 }
 491
 492 static
 493 void bt_ctf_stream_destroy(struct bt_ctf_ref *ref)
 494 {
 495         struct bt_ctf_stream *stream;
 496
 497         if (!ref) {
 498                 return;
 499         }
 500
 501         stream = container_of(ref, struct bt_ctf_stream, ref_count);
 502         ctf_fini_pos(&stream->pos);
 503         if (close(stream->pos.fd)) {
 504                 perror("close");
 505         }
 506
 507         if (stream->stream_class) {
 508                 bt_ctf_stream_class_put(stream->stream_class);
 509         }
 510         if (stream->events) {
 511                 g_ptr_array_free(stream->events, TRUE);
 512         }
 513         if (stream->packet_context) {
 514                 bt_ctf_field_put(stream->packet_context);
 515         }
 516         g_free(stream);
 517 }
 518
 519 static
 520 int set_structure_field_integer(struct bt_ctf_field *structure, char *name,
 521                 uint64_t value)
 522 {
 523         int ret = 0;
 524         struct bt_ctf_field *integer =
 525                 bt_ctf_field_structure_get_field(structure, name);
 526
 527         if (!structure || !name) {
 528                 ret = -1;
 529                 goto end;
 530         }
 531
 532         if (!integer) {
 533                 /* Field not found, not an error. */
 534                 goto end;
 535         }
 536
 537         /* Make sure the payload has not already been set. */
 538         if (!bt_ctf_field_validate(integer)) {
 539                 /* Payload already set, not an error */
 540                 goto end;
 541         }
 542
 543         ret = bt_ctf_field_unsigned_integer_set_value(integer, value);
 544 end:
 545         bt_ctf_field_put(integer);
 546         return ret;
 547 }