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