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