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