plugins/ctf/fs-sink/write.c

   1 /*
   2  * writer.c
   3  *
   4  * Babeltrace CTF Writer Output Plugin Event Handling
   5  *
   6  * Copyright 2016 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/event.h>
  30 #include <babeltrace/ctf-ir/packet.h>
  31 #include <babeltrace/ctf-ir/event-class.h>
  32 #include <babeltrace/ctf-ir/stream.h>
  33 #include <babeltrace/ctf-ir/stream-class.h>
  34 #include <babeltrace/ctf-ir/clock-class.h>
  35 #include <babeltrace/ctf-ir/fields.h>
  36 #include <babeltrace/ctf-writer/stream-class.h>
  37 #include <babeltrace/ctf-writer/stream.h>
  38 #include <assert.h>
  39 #include <glib.h>
  40
  41 #include <ctfcopytrace.h>
  42
  43 #include "writer.h"
  44
  45 static
  46 void unref_stream_class(struct bt_ctf_stream_class *writer_stream_class)
  47 {
  48         bt_put(writer_stream_class);
  49 }
  50
  51 static
  52 void unref_stream(struct bt_ctf_stream_class *writer_stream)
  53 {
  54         bt_put(writer_stream);
  55 }
  56
  57 static
  58 gboolean empty_ht(gpointer key, gpointer value, gpointer user_data)
  59 {
  60         return TRUE;
  61 }
  62
  63 static
  64 gboolean empty_streams_ht(gpointer key, gpointer value, gpointer user_data)
  65 {
  66         struct bt_ctf_stream *writer_stream = value;
  67
  68         bt_ctf_stream_flush(writer_stream);
  69
  70         return TRUE;
  71 }
  72
  73 static
  74 void destroy_stream_state_key(gpointer key)
  75 {
  76         g_free((enum fs_writer_stream_state *) key);
  77 }
  78
  79 static
  80 void check_completed_trace(gpointer key, gpointer value, gpointer user_data)
  81 {
  82         enum fs_writer_stream_state *state = value;
  83         int *trace_completed = user_data;
  84
  85         if (*state != FS_WRITER_COMPLETED_STREAM) {
  86                 *trace_completed = 0;
  87         }
  88 }
  89
  90 static
  91 void trace_is_static_listener(struct bt_ctf_trace *trace, void *data)
  92 {
  93         struct fs_writer *fs_writer = data;
  94         int trace_completed = 1;
  95
  96         fs_writer->trace_static = 1;
  97
  98         g_hash_table_foreach(fs_writer->stream_states,
  99                         check_completed_trace, &trace_completed);
 100         if (trace_completed) {
 101                 writer_close(fs_writer->writer_component, fs_writer);
 102                 g_hash_table_remove(fs_writer->writer_component->trace_map,
 103                                 fs_writer->trace);
 104         }
 105 }
 106
 107 static
 108 struct bt_ctf_stream_class *insert_new_stream_class(
 109                 struct writer_component *writer_component,
 110                 struct fs_writer *fs_writer,
 111                 struct bt_ctf_stream_class *stream_class)
 112 {
 113         struct bt_ctf_stream_class *writer_stream_class = NULL;
 114         struct bt_ctf_trace *trace = NULL, *writer_trace = NULL;
 115         struct bt_ctf_writer *ctf_writer = fs_writer->writer;
 116         enum bt_component_status ret;
 117
 118         trace = bt_ctf_stream_class_get_trace(stream_class);
 119         if (!trace) {
 120                 fprintf(writer_component->err,
 121                                 "[error] %s in %s:%d\n", __func__, __FILE__,
 122                                 __LINE__);
 123                 goto error;
 124         }
 125
 126         writer_trace = bt_ctf_writer_get_trace(ctf_writer);
 127         if (!writer_trace) {
 128                 fprintf(writer_component->err,
 129                                 "[error] %s in %s:%d\n", __func__, __FILE__,
 130                                 __LINE__);
 131                 goto error;
 132         }
 133
 134         ret = ctf_copy_clock_classes(writer_component->err, writer_trace,
 135                         writer_stream_class, trace);
 136         if (ret != BT_COMPONENT_STATUS_OK) {
 137                 fprintf(writer_component->err,
 138                                 "[error] %s in %s:%d\n", __func__, __FILE__,
 139                                 __LINE__);
 140                 goto error;
 141         }
 142
 143         writer_stream_class = ctf_copy_stream_class(writer_component->err,
 144                         stream_class, writer_trace, true);
 145         if (!writer_stream_class) {
 146                 fprintf(writer_component->err, "[error] Failed to copy stream class\n");
 147                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 148                                 __func__, __FILE__, __LINE__);
 149                 goto error;
 150         }
 151
 152         ret = bt_ctf_trace_add_stream_class(writer_trace, writer_stream_class);
 153         if (ret) {
 154                 fprintf(writer_component->err,
 155                                 "[error] %s in %s:%d\n", __func__, __FILE__,
 156                                 __LINE__);
 157                 goto error;
 158         }
 159
 160         g_hash_table_insert(fs_writer->stream_class_map,
 161                         (gpointer) stream_class, writer_stream_class);
 162
 163         goto end;
 164
 165 error:
 166         BT_PUT(writer_stream_class);
 167 end:
 168         bt_put(writer_trace);
 169         bt_put(trace);
 170         return writer_stream_class;
 171 }
 172
 173 static
 174 enum fs_writer_stream_state *insert_new_stream_state(
 175                 struct writer_component *writer_component,
 176                 struct fs_writer *fs_writer, struct bt_ctf_stream *stream)
 177 {
 178         enum fs_writer_stream_state *v = NULL;
 179
 180         v = g_new0(enum fs_writer_stream_state, 1);
 181         if (!v) {
 182                 fprintf(writer_component->err,
 183                                 "[error] %s in %s:%d\n", __func__,
 184                                 __FILE__, __LINE__);
 185         }
 186         *v = FS_WRITER_UNKNOWN_STREAM;
 187
 188         g_hash_table_insert(fs_writer->stream_states, stream, v);
 189
 190         return v;
 191 }
 192
 193 /*
 194  * Make sure the output path is valid for a single trace: either it does
 195  * not exists or it is empty.
 196  *
 197  * Return 0 if the path is valid, -1 otherwise.
 198  */
 199 static
 200 bool valid_single_trace_path(const char *path)
 201 {
 202         GError *error = NULL;
 203         GDir *dir = NULL;
 204         int ret = 0;
 205
 206         dir = g_dir_open(path, 0, &error);
 207
 208         /* Non-existent directory. */
 209         if (!dir) {
 210                 /* For any other error, return an error */
 211                 if (error->code != G_FILE_ERROR_NOENT) {
 212                         ret = -1;
 213                 }
 214                 goto end;
 215         }
 216
 217         /* g_dir_read_name skips "." and "..", error out on first result */
 218         while (g_dir_read_name(dir) != NULL) {
 219                 ret = -1;
 220                 break;
 221         }
 222
 223 end:
 224         if (dir) {
 225                 g_dir_close(dir);
 226         }
 227         if (error) {
 228                 g_error_free(error);
 229         }
 230
 231         return ret;
 232 }
 233
 234 static
 235 int make_trace_path(struct writer_component *writer_component,
 236                 struct bt_ctf_trace *trace, char *trace_path)
 237 {
 238         int ret;
 239         const char *trace_name;
 240
 241         if (writer_component->single_trace) {
 242                 trace_name = "\0";
 243         } else {
 244                 trace_name = bt_ctf_trace_get_name(trace);
 245                 if (!trace_name) {
 246                         trace_name = writer_component->trace_name_base->str;
 247                 }
 248         }
 249
 250         /* Sanitize the trace name. */
 251         if (strlen(trace_name) == 2 && !strcmp(trace_name, "..")) {
 252                 fprintf(writer_component->err, "[error] Trace name cannot "
 253                                 "be \"..\"\n");
 254                 goto error;
 255         }
 256
 257         if (strstr(trace_name, "../")) {
 258                 fprintf(writer_component->err, "[error] Trace name cannot "
 259                                 "contain \"../\", received \"%s\"\n",
 260                                 trace_name);
 261                 goto error;
 262
 263         }
 264
 265         snprintf(trace_path, PATH_MAX, "%s/%s",
 266                         writer_component->base_path->str,
 267                         trace_name);
 268         /*
 269          * Append a suffix if the trace_path exists and we are not in
 270          * single-trace mode.
 271          */
 272         if (writer_component->single_trace) {
 273                 if (valid_single_trace_path(trace_path) != 0) {
 274                         fprintf(writer_component->err,
 275                                         "[error] Invalid output directory\n");
 276                         goto error;
 277                 }
 278         } else {
 279                 if (g_file_test(trace_path, G_FILE_TEST_EXISTS)) {
 280                         int i = 0;
 281
 282                         do {
 283                                 snprintf(trace_path, PATH_MAX, "%s/%s-%d",
 284                                                 writer_component->base_path->str,
 285                                                 trace_name, ++i);
 286                         } while (g_file_test(trace_path, G_FILE_TEST_EXISTS) && i < INT_MAX);
 287                         if (i == INT_MAX) {
 288                                 fprintf(writer_component->err, "[error] Unable to find "
 289                                                 "a unique trace path\n");
 290                                 goto error;
 291                         }
 292                 }
 293         }
 294
 295         ret = 0;
 296         goto end;
 297
 298 error:
 299         ret = -1;
 300 end:
 301         return ret;
 302 }
 303
 304 static
 305 struct fs_writer *insert_new_writer(
 306                 struct writer_component *writer_component,
 307                 struct bt_ctf_trace *trace)
 308 {
 309         struct bt_ctf_writer *ctf_writer = NULL;
 310         struct bt_ctf_trace *writer_trace = NULL;
 311         char trace_path[PATH_MAX];
 312         enum bt_component_status ret;
 313         struct bt_ctf_stream *stream = NULL;
 314         struct fs_writer *fs_writer = NULL;
 315         int nr_stream, i;
 316
 317         if (writer_component->single_trace && writer_component->nr_traces > 0) {
 318                 fprintf(writer_component->err, "[error] Trying to process more "
 319                                 "than one trace but --single-trace mode "
 320                                 "enabled\n");
 321                 goto error;
 322         }
 323
 324         ret = make_trace_path(writer_component, trace, trace_path);
 325         if (ret) {
 326                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 327                                 __func__, __FILE__, __LINE__);
 328                 goto error;
 329         }
 330
 331         printf("ctf.fs sink creating trace in %s\n", trace_path);
 332
 333         ctf_writer = bt_ctf_writer_create(trace_path);
 334         if (!ctf_writer) {
 335                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 336                                 __func__, __FILE__, __LINE__);
 337                 goto error;
 338         }
 339
 340         writer_trace = bt_ctf_writer_get_trace(ctf_writer);
 341         if (!writer_trace) {
 342                 fprintf(writer_component->err,
 343                                 "[error] %s in %s:%d\n", __func__, __FILE__,
 344                                 __LINE__);
 345                 goto error;
 346         }
 347
 348         ret = ctf_copy_trace(writer_component->err, trace, writer_trace);
 349         if (ret != BT_COMPONENT_STATUS_OK) {
 350                 fprintf(writer_component->err, "[error] Failed to copy trace\n");
 351                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 352                                 __func__, __FILE__, __LINE__);
 353                 BT_PUT(ctf_writer);
 354                 goto error;
 355         }
 356
 357         fs_writer = g_new0(struct fs_writer, 1);
 358         if (!fs_writer) {
 359                 fprintf(writer_component->err,
 360                                 "[error] %s in %s:%d\n", __func__, __FILE__,
 361                                 __LINE__);
 362                 goto error;
 363         }
 364         fs_writer->writer = ctf_writer;
 365         fs_writer->trace = trace;
 366         fs_writer->writer_trace = writer_trace;
 367         fs_writer->writer_component = writer_component;
 368         BT_PUT(writer_trace);
 369         fs_writer->stream_class_map = g_hash_table_new_full(g_direct_hash,
 370                         g_direct_equal, NULL, (GDestroyNotify) unref_stream_class);
 371         fs_writer->stream_map = g_hash_table_new_full(g_direct_hash,
 372                         g_direct_equal, NULL, (GDestroyNotify) unref_stream);
 373         fs_writer->stream_states = g_hash_table_new_full(g_direct_hash,
 374                         g_direct_equal, NULL, destroy_stream_state_key);
 375
 376         /* Set all the existing streams in the unknown state. */
 377         nr_stream = bt_ctf_trace_get_stream_count(trace);
 378         for (i = 0; i < nr_stream; i++) {
 379                 stream = bt_ctf_trace_get_stream_by_index(trace, i);
 380                 if (!stream) {
 381                         fprintf(writer_component->err,
 382                                         "[error] %s in %s:%d\n", __func__,
 383                                         __FILE__, __LINE__);
 384                         goto error;
 385                 }
 386                 insert_new_stream_state(writer_component, fs_writer, stream);
 387                 BT_PUT(stream);
 388         }
 389
 390         /* Check if the trace is already static or register a listener. */
 391         if (bt_ctf_trace_is_static(trace)) {
 392                 fs_writer->trace_static = 1;
 393                 fs_writer->static_listener_id = -1;
 394         } else {
 395                 ret = bt_ctf_trace_add_is_static_listener(trace,
 396                                 trace_is_static_listener, fs_writer);
 397                 if (ret < 0) {
 398                         fprintf(writer_component->err,
 399                                         "[error] %s in %s:%d\n", __func__, __FILE__,
 400                                         __LINE__);
 401                         goto error;
 402                 }
 403                 fs_writer->static_listener_id = ret;
 404         }
 405
 406         writer_component->nr_traces++;
 407         g_hash_table_insert(writer_component->trace_map, (gpointer) trace,
 408                         fs_writer);
 409
 410         goto end;
 411
 412 error:
 413         g_free(fs_writer);
 414         fs_writer = NULL;
 415         bt_put(writer_trace);
 416         bt_put(stream);
 417         BT_PUT(ctf_writer);
 418 end:
 419         return fs_writer;
 420 }
 421
 422 static
 423 struct fs_writer *get_fs_writer(struct writer_component *writer_component,
 424                 struct bt_ctf_stream_class *stream_class)
 425 {
 426         struct bt_ctf_trace *trace = NULL;
 427         struct fs_writer *fs_writer;
 428
 429         trace = bt_ctf_stream_class_get_trace(stream_class);
 430         if (!trace) {
 431                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 432                                 __func__, __FILE__, __LINE__);
 433                 goto error;
 434         }
 435
 436         fs_writer = g_hash_table_lookup(writer_component->trace_map,
 437                         (gpointer) trace);
 438         if (!fs_writer) {
 439                 fs_writer = insert_new_writer(writer_component, trace);
 440         }
 441         BT_PUT(trace);
 442         goto end;
 443
 444 error:
 445         fs_writer = NULL;
 446 end:
 447         return fs_writer;
 448 }
 449
 450 static
 451 struct fs_writer *get_fs_writer_from_stream(
 452                 struct writer_component *writer_component,
 453                 struct bt_ctf_stream *stream)
 454 {
 455         struct bt_ctf_stream_class *stream_class = NULL;
 456         struct fs_writer *fs_writer;
 457
 458         stream_class = bt_ctf_stream_get_class(stream);
 459         if (!stream_class) {
 460                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 461                                 __func__, __FILE__, __LINE__);
 462                 goto error;
 463         }
 464
 465         fs_writer = get_fs_writer(writer_component, stream_class);
 466         goto end;
 467
 468 error:
 469         fs_writer = NULL;
 470
 471 end:
 472         bt_put(stream_class);
 473         return fs_writer;
 474 }
 475
 476 static
 477 struct bt_ctf_stream_class *lookup_stream_class(
 478                 struct writer_component *writer_component,
 479                 struct bt_ctf_stream_class *stream_class)
 480 {
 481         struct fs_writer *fs_writer = get_fs_writer(
 482                         writer_component, stream_class);
 483         assert(fs_writer);
 484         return (struct bt_ctf_stream_class *) g_hash_table_lookup(
 485                         fs_writer->stream_class_map, (gpointer) stream_class);
 486 }
 487
 488 static
 489 struct bt_ctf_stream *lookup_stream(struct writer_component *writer_component,
 490                 struct bt_ctf_stream *stream)
 491 {
 492         struct fs_writer *fs_writer = get_fs_writer_from_stream(
 493                         writer_component, stream);
 494         assert(fs_writer);
 495         return (struct bt_ctf_stream *) g_hash_table_lookup(
 496                         fs_writer->stream_map, (gpointer) stream);
 497 }
 498
 499 static
 500 struct bt_ctf_stream *insert_new_stream(
 501                 struct writer_component *writer_component,
 502                 struct fs_writer *fs_writer,
 503                 struct bt_ctf_stream_class *stream_class,
 504                 struct bt_ctf_stream *stream)
 505 {
 506         struct bt_ctf_stream *writer_stream = NULL;
 507         struct bt_ctf_stream_class *writer_stream_class = NULL;
 508         struct bt_ctf_writer *ctf_writer = bt_get(fs_writer->writer);
 509
 510         writer_stream_class = lookup_stream_class(writer_component,
 511                         stream_class);
 512         if (!writer_stream_class) {
 513                 writer_stream_class = insert_new_stream_class(
 514                                 writer_component, fs_writer, stream_class);
 515                 if (!writer_stream_class) {
 516                         fprintf(writer_component->err, "[error] %s in %s:%d\n",
 517                                         __func__, __FILE__, __LINE__);
 518                         goto error;
 519                 }
 520         }
 521         bt_get(writer_stream_class);
 522
 523         writer_stream = bt_ctf_stream_create(writer_stream_class,
 524                 bt_ctf_stream_get_name(stream));
 525         if (!writer_stream) {
 526                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 527                                 __func__, __FILE__, __LINE__);
 528                 goto error;
 529         }
 530
 531         g_hash_table_insert(fs_writer->stream_map, (gpointer) stream,
 532                         writer_stream);
 533
 534         goto end;
 535
 536 error:
 537         BT_PUT(writer_stream);
 538 end:
 539         bt_put(ctf_writer);
 540         bt_put(writer_stream_class);
 541         return writer_stream;
 542 }
 543
 544 static
 545 struct bt_ctf_event_class *get_event_class(struct writer_component *writer_component,
 546                 struct bt_ctf_stream_class *writer_stream_class,
 547                 struct bt_ctf_event_class *event_class)
 548 {
 549         return bt_ctf_stream_class_get_event_class_by_id(writer_stream_class,
 550                         bt_ctf_event_class_get_id(event_class));
 551 }
 552
 553 static
 554 struct bt_ctf_stream *get_writer_stream(
 555                 struct writer_component *writer_component,
 556                 struct bt_ctf_packet *packet, struct bt_ctf_stream *stream)
 557 {
 558         struct bt_ctf_stream *writer_stream = NULL;
 559
 560         writer_stream = lookup_stream(writer_component, stream);
 561         if (!writer_stream) {
 562                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 563                                 __func__, __FILE__, __LINE__);
 564                 goto error;
 565         }
 566         bt_get(writer_stream);
 567
 568         goto end;
 569
 570 error:
 571         BT_PUT(writer_stream);
 572 end:
 573         return writer_stream;
 574 }
 575
 576 BT_HIDDEN
 577 void writer_close(struct writer_component *writer_component,
 578                 struct fs_writer *fs_writer)
 579 {
 580         if (fs_writer->static_listener_id >= 0) {
 581                 bt_ctf_trace_remove_is_static_listener(fs_writer->trace,
 582                                 fs_writer->static_listener_id);
 583         }
 584
 585         /* Empty the stream class HT. */
 586         g_hash_table_foreach_remove(fs_writer->stream_class_map,
 587                         empty_ht, NULL);
 588         g_hash_table_destroy(fs_writer->stream_class_map);
 589
 590         /* Empty the stream HT. */
 591         g_hash_table_foreach_remove(fs_writer->stream_map,
 592                         empty_streams_ht, NULL);
 593         g_hash_table_destroy(fs_writer->stream_map);
 594
 595         /* Empty the stream state HT. */
 596         g_hash_table_foreach_remove(fs_writer->stream_states,
 597                         empty_ht, NULL);
 598         g_hash_table_destroy(fs_writer->stream_states);
 599 }
 600
 601 BT_HIDDEN
 602 enum bt_component_status writer_stream_begin(
 603                 struct writer_component *writer_component,
 604                 struct bt_ctf_stream *stream)
 605 {
 606         struct bt_ctf_stream_class *stream_class = NULL;
 607         struct fs_writer *fs_writer;
 608         struct bt_ctf_stream *writer_stream = NULL;
 609         enum bt_component_status ret = BT_COMPONENT_STATUS_OK;
 610         enum fs_writer_stream_state *state;
 611
 612         stream_class = bt_ctf_stream_get_class(stream);
 613         if (!stream_class) {
 614                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 615                                 __func__, __FILE__, __LINE__);
 616                 goto error;
 617         }
 618
 619         fs_writer = get_fs_writer(writer_component, stream_class);
 620         if (!fs_writer) {
 621                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 622                                 __func__, __FILE__, __LINE__);
 623                 goto error;
 624         }
 625
 626         /* Set the stream as active */
 627         state = g_hash_table_lookup(fs_writer->stream_states, stream);
 628         if (!state) {
 629                 if (fs_writer->trace_static) {
 630                         fprintf(writer_component->err, "[error] Adding a new "
 631                                         "stream on a static trace\n");
 632                         goto error;
 633                 }
 634                 state = insert_new_stream_state(writer_component, fs_writer,
 635                                 stream);
 636         }
 637         if (*state != FS_WRITER_UNKNOWN_STREAM) {
 638                 fprintf(writer_component->err, "[error] Unexpected stream "
 639                                 "state %d\n", *state);
 640                 goto error;
 641         }
 642         *state = FS_WRITER_ACTIVE_STREAM;
 643
 644         writer_stream = insert_new_stream(writer_component, fs_writer,
 645                         stream_class, stream);
 646         if (!writer_stream) {
 647                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 648                                 __func__, __FILE__, __LINE__);
 649                 goto error;
 650         }
 651
 652         goto end;
 653
 654 error:
 655         ret = BT_COMPONENT_STATUS_ERROR;
 656 end:
 657         bt_put(stream_class);
 658         return ret;
 659 }
 660
 661 BT_HIDDEN
 662 enum bt_component_status writer_stream_end(
 663                 struct writer_component *writer_component,
 664                 struct bt_ctf_stream *stream)
 665 {
 666         struct bt_ctf_stream_class *stream_class = NULL;
 667         struct fs_writer *fs_writer;
 668         struct bt_ctf_trace *trace = NULL;
 669         enum bt_component_status ret = BT_COMPONENT_STATUS_OK;
 670         enum fs_writer_stream_state *state;
 671
 672         stream_class = bt_ctf_stream_get_class(stream);
 673         if (!stream_class) {
 674                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 675                                 __func__, __FILE__, __LINE__);
 676                 goto error;
 677         }
 678
 679         fs_writer = get_fs_writer(writer_component, stream_class);
 680         if (!fs_writer) {
 681                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 682                                 __func__, __FILE__, __LINE__);
 683                 goto error;
 684         }
 685
 686         state = g_hash_table_lookup(fs_writer->stream_states, stream);
 687         if (*state != FS_WRITER_ACTIVE_STREAM) {
 688                 fprintf(writer_component->err, "[error] Unexpected stream "
 689                                 "state %d\n", *state);
 690                 goto error;
 691         }
 692         *state = FS_WRITER_COMPLETED_STREAM;
 693
 694         g_hash_table_remove(fs_writer->stream_map, stream);
 695
 696         if (fs_writer->trace_static) {
 697                 int trace_completed = 1;
 698
 699                 g_hash_table_foreach(fs_writer->stream_states,
 700                                 check_completed_trace, &trace_completed);
 701                 if (trace_completed) {
 702                         writer_close(writer_component, fs_writer);
 703                         g_hash_table_remove(writer_component->trace_map,
 704                                         fs_writer->trace);
 705                 }
 706         }
 707
 708         goto end;
 709
 710 error:
 711         ret = BT_COMPONENT_STATUS_ERROR;
 712 end:
 713         BT_PUT(trace);
 714         BT_PUT(stream_class);
 715         return ret;
 716 }
 717
 718 BT_HIDDEN
 719 enum bt_component_status writer_new_packet(
 720                 struct writer_component *writer_component,
 721                 struct bt_ctf_packet *packet)
 722 {
 723         struct bt_ctf_stream *stream = NULL, *writer_stream = NULL;
 724         enum bt_component_status ret = BT_COMPONENT_STATUS_OK;
 725         int int_ret;
 726
 727         stream = bt_ctf_packet_get_stream(packet);
 728         if (!stream) {
 729                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 730                                 __func__, __FILE__, __LINE__);
 731                 goto error;
 732         }
 733
 734         writer_stream = get_writer_stream(writer_component, packet, stream);
 735         if (!writer_stream) {
 736                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 737                                 __func__, __FILE__, __LINE__);
 738                 goto error;
 739         }
 740         BT_PUT(stream);
 741
 742         int_ret = ctf_stream_copy_packet_context(
 743                         writer_component->err, packet, writer_stream);
 744         if (int_ret < 0) {
 745                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 746                                 __func__, __FILE__, __LINE__);
 747                 goto error;
 748         }
 749
 750         ret = ctf_stream_copy_packet_header(writer_component->err,
 751                         packet, writer_stream);
 752         if (ret != 0) {
 753                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 754                                 __func__, __FILE__, __LINE__);
 755                 goto error;
 756         }
 757
 758         goto end;
 759
 760 error:
 761         ret = BT_COMPONENT_STATUS_ERROR;
 762 end:
 763         bt_put(writer_stream);
 764         bt_put(stream);
 765         return ret;
 766 }
 767
 768 BT_HIDDEN
 769 enum bt_component_status writer_close_packet(
 770                 struct writer_component *writer_component,
 771                 struct bt_ctf_packet *packet)
 772 {
 773         struct bt_ctf_stream *stream = NULL, *writer_stream = NULL;
 774         enum bt_component_status ret;
 775
 776         stream = bt_ctf_packet_get_stream(packet);
 777         if (!stream) {
 778                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 779                                 __func__, __FILE__, __LINE__);
 780                 goto error;
 781         }
 782
 783         writer_stream = lookup_stream(writer_component, stream);
 784         if (!writer_stream) {
 785                 fprintf(writer_component->err, "[error] %s in %s:%d\n",
 786                                 __func__, __FILE__, __LINE__);
 787                 goto error;
 788         }
 789         BT_PUT(stream);
 790
 791         bt_get(writer_stream);
 792
 793         ret = bt_ctf_stream_flush(writer_stream);
 794         if (ret < 0) {
 795                 fprintf(writer_component->err,
 796                                 "[error] Failed to flush packet\n");
 797                 goto error;
 798         }
 799         BT_PUT(writer_stream);
 800
 801         ret = BT_COMPONENT_STATUS_OK;
 802         goto end;
 803
 804 error:
 805         ret = BT_COMPONENT_STATUS_ERROR;
 806 end:
 807         bt_put(writer_stream);
 808         bt_put(stream);
 809         return ret;
 810 }
 811
 812 BT_HIDDEN
 813 enum bt_component_status writer_output_event(
 814                 struct writer_component *writer_component,
 815                 struct bt_ctf_event *event)
 816 {
 817         enum bt_component_status ret;
 818         struct bt_ctf_event_class *event_class = NULL, *writer_event_class = NULL;
 819         struct bt_ctf_stream *stream = NULL, *writer_stream = NULL;
 820         struct bt_ctf_stream_class *stream_class = NULL, *writer_stream_class = NULL;
 821         struct bt_ctf_event *writer_event = NULL;
 822         const char *event_name;
 823         int int_ret;
 824
 825         event_class = bt_ctf_event_get_class(event);
 826         if (!event_class) {
 827                 fprintf(writer_component->err, "[error] %s in %s:%d\n", __func__,
 828                                 __FILE__, __LINE__);
 829                 goto error;
 830         }
 831
 832         event_name = bt_ctf_event_class_get_name(event_class);
 833         if (!event_name) {
 834                 fprintf(writer_component->err, "[error] %s in %s:%d\n", __func__,
 835                                 __FILE__, __LINE__);
 836                 goto error;
 837         }
 838
 839         stream = bt_ctf_event_get_stream(event);
 840         if (!stream) {
 841                 fprintf(writer_component->err, "[error] %s in %s:%d\n", __func__,
 842                                 __FILE__, __LINE__);
 843                 goto error;
 844         }
 845
 846         writer_stream = lookup_stream(writer_component, stream);
 847         if (!writer_stream || !bt_get(writer_stream)) {
 848                 fprintf(writer_component->err, "[error] %s in %s:%d\n", __func__,
 849                                 __FILE__, __LINE__);
 850                 goto error;
 851         }
 852
 853         stream_class = bt_ctf_event_class_get_stream_class(event_class);
 854         if (!stream_class) {
 855                 fprintf(writer_component->err, "[error] %s in %s:%d\n", __func__,
 856                                 __FILE__, __LINE__);
 857                 goto error;
 858         }
 859
 860         writer_stream_class = lookup_stream_class(writer_component, stream_class);
 861         if (!writer_stream_class || !bt_get(writer_stream_class)) {
 862                 fprintf(writer_component->err, "[error] %s in %s:%d\n", __func__,
 863                                 __FILE__, __LINE__);
 864                 goto error;
 865         }
 866
 867         writer_event_class = get_event_class(writer_component,
 868                         writer_stream_class, event_class);
 869         if (!writer_event_class) {
 870                 writer_event_class = ctf_copy_event_class(writer_component->err,
 871                                 event_class);
 872                 if (!writer_event_class) {
 873                         fprintf(writer_component->err, "[error] %s in %s:%d\n",
 874                                         __func__, __FILE__, __LINE__);
 875                         goto error;
 876                 }
 877                 int_ret = bt_ctf_stream_class_add_event_class(
 878                                 writer_stream_class, writer_event_class);
 879                 if (int_ret) {
 880                         fprintf(writer_component->err, "[error] %s in %s:%d\n",
 881                                         __func__, __FILE__, __LINE__);
 882                         goto error;
 883                 }
 884         }
 885
 886         writer_event = ctf_copy_event(writer_component->err, event,
 887                         writer_event_class, true);
 888         if (!writer_event) {
 889                 fprintf(writer_component->err, "[error] %s in %s:%d\n", __func__,
 890                                 __FILE__, __LINE__);
 891                 fprintf(writer_component->err, "[error] Failed to copy event %s\n",
 892                                 bt_ctf_event_class_get_name(writer_event_class));
 893                 goto error;
 894         }
 895
 896         int_ret = bt_ctf_stream_append_event(writer_stream, writer_event);
 897         if (int_ret < 0) {
 898                 fprintf(writer_component->err, "[error] %s in %s:%d\n", __func__,
 899                                 __FILE__, __LINE__);
 900                 fprintf(writer_component->err, "[error] Failed to append event %s\n",
 901                                 bt_ctf_event_class_get_name(writer_event_class));
 902                 goto error;
 903         }
 904
 905         ret = BT_COMPONENT_STATUS_OK;
 906         goto end;
 907
 908 error:
 909         ret = BT_COMPONENT_STATUS_ERROR;
 910 end:
 911         bt_put(writer_event);
 912         bt_put(writer_event_class);
 913         bt_put(writer_stream_class);
 914         bt_put(stream_class);
 915         bt_put(writer_stream);
 916         bt_put(stream);
 917         bt_put(event_class);
 918         return ret;
 919 }