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) {
 330                 ret = -1;
 331                 goto end;
 332         }
 333
 334         if (!stream->events->len) {
 335                 goto end;
 336         }
 337
 338         if (stream->flush.func) {
 339                 stream->flush.func(stream, stream->flush.data);
 340         }
 341
 342         stream_class = stream->stream_class;
 343         timestamp_begin = ((struct bt_ctf_event *) g_ptr_array_index(
 344                 stream->events, 0))->timestamp;
 345         timestamp_end = ((struct bt_ctf_event *) g_ptr_array_index(
 346                 stream->events, stream->events->len - 1))->timestamp;
 347
 348         /* Set the default context attributes if present and unset. */
 349         ret = set_structure_field_integer(stream->packet_context,
 350                 "timestamp_begin", timestamp_begin);
 351         if (ret) {
 352                 goto end;
 353         }
 354
 355         ret = set_structure_field_integer(stream->packet_context,
 356                 "timestamp_end", timestamp_end);
 357         if (ret) {
 358                 goto end;
 359         }
 360
 361         ret = set_structure_field_integer(stream->packet_context,
 362                 "content_size", UINT64_MAX);
 363         if (ret) {
 364                 goto end;
 365         }
 366
 367         ret = set_structure_field_integer(stream->packet_context,
 368                 "packet_size", UINT64_MAX);
 369         if (ret) {
 370                 goto end;
 371         }
 372
 373         /* Write packet context */
 374         memcpy(&packet_context_pos, &stream->pos,
 375                sizeof(struct ctf_stream_pos));
 376         ret = bt_ctf_field_serialize(stream->packet_context,
 377                 &stream->pos);
 378         if (ret) {
 379                 goto end;
 380         }
 381
 382         ret = bt_ctf_stream_get_discarded_events_count(stream,
 383                 &events_discarded);
 384         if (ret) {
 385                 goto end;
 386         }
 387
 388         /* Unset the packet context's fields. */
 389         ret = bt_ctf_field_reset(stream->packet_context);
 390         if (ret) {
 391                 goto end;
 392         }
 393
 394         /* Set the previous number of discarded events. */
 395         ret = set_structure_field_integer(stream->packet_context,
 396                 "events_discarded", events_discarded);
 397         if (ret) {
 398                 goto end;
 399         }
 400
 401         for (i = 0; i < stream->events->len; i++) {
 402                 struct bt_ctf_event *event = g_ptr_array_index(
 403                         stream->events, i);
 404                 uint32_t event_id = bt_ctf_event_class_get_id(
 405                         event->event_class);
 406                 uint64_t timestamp = bt_ctf_event_get_timestamp(event);
 407
 408                 ret = bt_ctf_field_reset(stream_class->event_header);
 409                 if (ret) {
 410                         goto end;
 411                 }
 412
 413                 ret = set_structure_field_integer(stream_class->event_header,
 414                         "id", event_id);
 415                 if (ret) {
 416                         goto end;
 417                 }
 418                 ret = set_structure_field_integer(stream_class->event_header,
 419                         "timestamp", timestamp);
 420                 if (ret) {
 421                         goto end;
 422                 }
 423
 424                 /* Write event header */
 425                 ret = bt_ctf_field_serialize(stream_class->event_header,
 426                         &stream->pos);
 427                 if (ret) {
 428                         goto end;
 429                 }
 430
 431                 /* Write event content */
 432                 ret = bt_ctf_event_serialize(event, &stream->pos);
 433                 if (ret) {
 434                         goto end;
 435                 }
 436         }
 437
 438         /*
 439          * Update the packet total size and content size and overwrite the
 440          * packet context.
 441          * Copy base_mma as the packet may have been remapped (e.g. when a
 442          * packet is resized).
 443          */
 444         packet_context_pos.base_mma = stream->pos.base_mma;
 445         ret = set_structure_field_integer(stream->packet_context,
 446                 "content_size", stream->pos.offset);
 447         if (ret) {
 448                 goto end;
 449         }
 450
 451         ret = set_structure_field_integer(stream->packet_context,
 452                 "packet_size", stream->pos.packet_size);
 453         if (ret) {
 454                 goto end;
 455         }
 456
 457         ret = bt_ctf_field_serialize(stream->packet_context,
 458                 &packet_context_pos);
 459         if (ret) {
 460                 goto end;
 461         }
 462
 463         g_ptr_array_set_size(stream->events, 0);
 464         stream->flushed_packet_count++;
 465 end:
 466         bt_ctf_field_put(integer);
 467         return ret;
 468 }
 469
 470 void bt_ctf_stream_get(struct bt_ctf_stream *stream)
 471 {
 472         if (!stream) {
 473                 return;
 474         }
 475
 476         bt_ctf_ref_get(&stream->ref_count);
 477 }
 478
 479 void bt_ctf_stream_put(struct bt_ctf_stream *stream)
 480 {
 481         if (!stream) {
 482                 return;
 483         }
 484
 485         bt_ctf_ref_put(&stream->ref_count, bt_ctf_stream_destroy);
 486 }
 487
 488 static
 489 void bt_ctf_stream_destroy(struct bt_ctf_ref *ref)
 490 {
 491         struct bt_ctf_stream *stream;
 492
 493         if (!ref) {
 494                 return;
 495         }
 496
 497         stream = container_of(ref, struct bt_ctf_stream, ref_count);
 498         ctf_fini_pos(&stream->pos);
 499         if (close(stream->pos.fd)) {
 500                 perror("close");
 501         }
 502
 503         if (stream->stream_class) {
 504                 bt_ctf_stream_class_put(stream->stream_class);
 505         }
 506         if (stream->events) {
 507                 g_ptr_array_free(stream->events, TRUE);
 508         }
 509         if (stream->packet_context) {
 510                 bt_ctf_field_put(stream->packet_context);
 511         }
 512         g_free(stream);
 513 }
 514
 515 static
 516 int set_structure_field_integer(struct bt_ctf_field *structure, char *name,
 517                 uint64_t value)
 518 {
 519         int ret = 0;
 520         struct bt_ctf_field *integer =
 521                 bt_ctf_field_structure_get_field(structure, name);
 522
 523         if (!structure || !name) {
 524                 ret = -1;
 525                 goto end;
 526         }
 527
 528         if (!integer) {
 529                 /* Field not found, not an error. */
 530                 goto end;
 531         }
 532
 533         /* Make sure the payload has not already been set. */
 534         if (!bt_ctf_field_validate(integer)) {
 535                 /* Payload already set, not an error */
 536                 goto end;
 537         }
 538
 539         ret = bt_ctf_field_unsigned_integer_set_value(integer, value);
 540 end:
 541         bt_ctf_field_put(integer);
 542         return ret;
 543 }