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