src/common/trace-chunk.c

   1 /*
   2  * Copyright (C) 2019 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  *
   4  * This library is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU Lesser General Public License, version 2.1 only,
   6  * as published by the Free Software Foundation.
   7  *
   8  * This library is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
  11  * for more details.
  12  *
  13  * You should have received a copy of the GNU Lesser General Public License
  14  * along with this library; if not, write to the Free Software Foundation,
  15  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  16  */
  17
  18 #include <lttng/constant.h>
  19 #include <common/string-utils/format.h>
  20 #include <common/trace-chunk.h>
  21 #include <common/trace-chunk-registry.h>
  22 #include <common/hashtable/utils.h>
  23 #include <common/hashtable/hashtable.h>
  24 #include <common/error.h>
  25 #include <common/utils.h>
  26 #include <common/time.h>
  27 #include <common/optional.h>
  28 #include <common/compat/directory-handle.h>
  29 #include <common/credentials.h>
  30 #include <common/defaults.h>
  31 #include <common/dynamic-array.h>
  32
  33 #include <urcu/ref.h>
  34 #include <urcu/rculfhash.h>
  35 #include <sys/stat.h>
  36 #include <inttypes.h>
  37 #include <pthread.h>
  38 #include <stdio.h>
  39
  40 /*
  41  * Two ISO 8601-compatible timestamps, separated by a hypen, followed an
  42  * index, i.e. <start-iso-8601>-<end-iso-8601>-<id-uint64_t>.
  43  */
  44 #define GENERATED_CHUNK_NAME_LEN (2 * sizeof("YYYYmmddTHHMMSS+HHMM") + MAX_INT_DEC_LEN(uint64_t))
  45 #define DIR_CREATION_MODE (S_IRWXU | S_IRWXG)
  46
  47 enum trace_chunk_mode {
  48         TRACE_CHUNK_MODE_USER,
  49         TRACE_CHUNK_MODE_OWNER,
  50 };
  51
  52 /*
  53  * Callback to invoke on release of a trace chunk. Note that there is no
  54  * need to 'lock' the trace chunk during the execution of these callbacks
  55  * since only one thread may access a chunk during its destruction (the last
  56  * to release its reference to the chunk).
  57  */
  58 typedef void (*chunk_close_command)(struct lttng_trace_chunk *trace_chunk);
  59
  60 /* Move a completed trace chunk to the 'completed' trace archive folder. */
  61 static
  62 void lttng_trace_chunk_move_to_completed(struct lttng_trace_chunk *trace_chunk);
  63
  64 struct chunk_credentials {
  65         bool use_current_user;
  66         struct lttng_credentials user;
  67 };
  68
  69 struct lttng_trace_chunk {
  70         pthread_mutex_t lock;
  71         struct urcu_ref ref;
  72         LTTNG_OPTIONAL(enum trace_chunk_mode) mode;
  73         /*
  74          * First-level directories created within the trace chunk.
  75          * Elements are of type 'char *'.
  76          */
  77         struct lttng_dynamic_pointer_array top_level_directories;
  78         /* Is contained within an lttng_trace_chunk_registry_element? */
  79         bool in_registry_element;
  80         bool name_overridden;
  81         char *name;
  82         /* An unset id means the chunk is anonymous. */
  83         LTTNG_OPTIONAL(uint64_t) id;
  84         LTTNG_OPTIONAL(time_t) timestamp_creation;
  85         LTTNG_OPTIONAL(time_t) timestamp_close;
  86         LTTNG_OPTIONAL(struct chunk_credentials) credentials;
  87         LTTNG_OPTIONAL(struct lttng_directory_handle) session_output_directory;
  88         LTTNG_OPTIONAL(struct lttng_directory_handle) chunk_directory;
  89         LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command;
  90 };
  91
  92 /* A trace chunk is uniquely identified by its (session id, chunk id) tuple. */
  93 struct lttng_trace_chunk_registry_element {
  94         uint64_t session_id;
  95         struct lttng_trace_chunk chunk;
  96         /* Weak and only set when added. */
  97         struct lttng_trace_chunk_registry *registry;
  98         struct cds_lfht_node trace_chunk_registry_ht_node;
  99         /* call_rcu delayed reclaim. */
 100         struct rcu_head rcu_node;
 101 };
 102
 103 struct lttng_trace_chunk_registry {
 104         struct cds_lfht *ht;
 105 };
 106
 107 const char *close_command_names[] = {
 108         [LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED] =
 109                 "move to completed chunk folder",
 110 };
 111
 112 chunk_close_command close_command_funcs[] = {
 113         [LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED] =
 114                         lttng_trace_chunk_move_to_completed,
 115 };
 116
 117 static
 118 bool lttng_trace_chunk_registry_element_equals(
 119                 const struct lttng_trace_chunk_registry_element *a,
 120                 const struct lttng_trace_chunk_registry_element *b)
 121 {
 122         if (a->session_id != b->session_id) {
 123                 goto not_equal;
 124         }
 125         if (a->chunk.id.is_set != b->chunk.id.is_set) {
 126                 goto not_equal;
 127         }
 128         if (a->chunk.id.is_set && a->chunk.id.value != b->chunk.id.value) {
 129                 goto not_equal;
 130         }
 131         return true;
 132 not_equal:
 133         return false;
 134 }
 135
 136 static
 137 int lttng_trace_chunk_registry_element_match(struct cds_lfht_node *node,
 138                 const void *key)
 139 {
 140         const struct lttng_trace_chunk_registry_element *element_a, *element_b;
 141
 142         element_a = (const struct lttng_trace_chunk_registry_element *) key;
 143         element_b = caa_container_of(node, typeof(*element_b),
 144                         trace_chunk_registry_ht_node);
 145         return lttng_trace_chunk_registry_element_equals(element_a, element_b);
 146 }
 147
 148 static
 149 unsigned long lttng_trace_chunk_registry_element_hash(
 150                 const struct lttng_trace_chunk_registry_element *element)
 151 {
 152         unsigned long hash = hash_key_u64(&element->session_id,
 153                         lttng_ht_seed);
 154
 155         if (element->chunk.id.is_set) {
 156                 hash ^= hash_key_u64(&element->chunk.id.value, lttng_ht_seed);
 157         }
 158
 159         return hash;
 160 }
 161
 162 static
 163 char *generate_chunk_name(uint64_t chunk_id, time_t creation_timestamp,
 164                 const time_t *close_timestamp)
 165 {
 166         int ret = 0;
 167         char *new_name= NULL;
 168         char start_datetime[sizeof("YYYYmmddTHHMMSS+HHMM")] = {};
 169         char end_datetime_suffix[sizeof("-YYYYmmddTHHMMSS+HHMM")] = {};
 170
 171         ret = time_to_iso8601_str(
 172                         creation_timestamp,
 173                         start_datetime, sizeof(start_datetime));
 174         if (ret) {
 175                 ERR("Failed to format trace chunk start date time");
 176                 goto error;
 177         }
 178         if (close_timestamp) {
 179                 *end_datetime_suffix = '-';
 180                 ret = time_to_iso8601_str(
 181                                 *close_timestamp,
 182                                 end_datetime_suffix + 1,
 183                                 sizeof(end_datetime_suffix));
 184                 if (ret) {
 185                         ERR("Failed to format trace chunk end date time");
 186                         goto error;
 187                 }
 188         }
 189         new_name = zmalloc(GENERATED_CHUNK_NAME_LEN);
 190         if (!new_name) {
 191                 ERR("Failed to allocate buffer for automatically-generated trace chunk name");
 192                 goto error;
 193         }
 194         ret = snprintf(new_name, GENERATED_CHUNK_NAME_LEN, "%s%s-%" PRIu64,
 195                         start_datetime, end_datetime_suffix, chunk_id);
 196         if (ret >= GENERATED_CHUNK_NAME_LEN || ret == -1) {
 197                 ERR("Failed to format trace chunk name");
 198                 goto error;
 199         }
 200
 201         return new_name;
 202 error:
 203         free(new_name);
 204         return NULL;
 205 }
 206
 207 static
 208 void lttng_trace_chunk_init(struct lttng_trace_chunk *chunk)
 209 {
 210         urcu_ref_init(&chunk->ref);
 211         pthread_mutex_init(&chunk->lock, NULL);
 212         lttng_dynamic_pointer_array_init(&chunk->top_level_directories, free);
 213 }
 214
 215 static
 216 void lttng_trace_chunk_fini(struct lttng_trace_chunk *chunk)
 217 {
 218         if (chunk->session_output_directory.is_set) {
 219                 lttng_directory_handle_fini(
 220                                 &chunk->session_output_directory.value);
 221         }
 222         if (chunk->chunk_directory.is_set) {
 223                 lttng_directory_handle_fini(&chunk->chunk_directory.value);
 224         }
 225         free(chunk->name);
 226         chunk->name = NULL;
 227         lttng_dynamic_pointer_array_reset(&chunk->top_level_directories);
 228         pthread_mutex_destroy(&chunk->lock);
 229 }
 230
 231 static
 232 struct lttng_trace_chunk *lttng_trace_chunk_allocate(void)
 233 {
 234         struct lttng_trace_chunk *chunk = NULL;
 235
 236         chunk = zmalloc(sizeof(*chunk));
 237         if (!chunk) {
 238                 ERR("Failed to allocate trace chunk");
 239                 goto end;
 240         }
 241         lttng_trace_chunk_init(chunk);
 242 end:
 243         return chunk;
 244 }
 245
 246 LTTNG_HIDDEN
 247 struct lttng_trace_chunk *lttng_trace_chunk_create_anonymous(void)
 248 {
 249         DBG("Creating anonymous trace chunk");
 250         return lttng_trace_chunk_allocate();
 251 }
 252
 253 LTTNG_HIDDEN
 254 struct lttng_trace_chunk *lttng_trace_chunk_create(
 255                 uint64_t chunk_id, time_t chunk_creation_time)
 256 {
 257         struct lttng_trace_chunk *chunk;
 258         char chunk_creation_datetime_buf[16] = {};
 259         const char *chunk_creation_datetime_str = "(formatting error)";
 260         struct tm timeinfo_buf, *timeinfo;
 261
 262         timeinfo = localtime_r(&chunk_creation_time, &timeinfo_buf);
 263         if (timeinfo) {
 264                 size_t strftime_ret;
 265
 266                 /* Don't fail because of this; it is only used for logging. */
 267                 strftime_ret = strftime(chunk_creation_datetime_buf,
 268                                 sizeof(chunk_creation_datetime_buf),
 269                                 "%Y%m%d-%H%M%S", timeinfo);
 270                 if (strftime_ret) {
 271                         chunk_creation_datetime_str =
 272                                         chunk_creation_datetime_buf;
 273                 }
 274         }
 275
 276         DBG("Creating trace chunk: chunk_id = %" PRIu64 ", creation time = %s",
 277                         chunk_id, chunk_creation_datetime_str);
 278         chunk = lttng_trace_chunk_allocate();
 279         if (!chunk) {
 280                 goto end;
 281         }
 282
 283         LTTNG_OPTIONAL_SET(&chunk->id, chunk_id);
 284         LTTNG_OPTIONAL_SET(&chunk->timestamp_creation, chunk_creation_time);
 285         if (chunk_id != 0) {
 286                 chunk->name = generate_chunk_name(chunk_id,
 287                                 chunk_creation_time, NULL);
 288                 if (!chunk->name) {
 289                         ERR("Failed to allocate trace chunk name storage");
 290                         goto error;
 291                 }
 292         }
 293
 294         DBG("Chunk name set to \"%s\"", chunk->name ? : "(none)");
 295 end:
 296         return chunk;
 297 error:
 298         lttng_trace_chunk_put(chunk);
 299         return NULL;
 300 }
 301
 302 LTTNG_HIDDEN
 303 enum lttng_trace_chunk_status lttng_trace_chunk_get_id(
 304                 struct lttng_trace_chunk *chunk, uint64_t *id)
 305 {
 306         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 307
 308         pthread_mutex_lock(&chunk->lock);
 309         if (chunk->id.is_set) {
 310                 *id = chunk->id.value;
 311         } else {
 312                 status = LTTNG_TRACE_CHUNK_STATUS_NONE;
 313         }
 314         pthread_mutex_unlock(&chunk->lock);
 315         return status;
 316 }
 317
 318 LTTNG_HIDDEN
 319 enum lttng_trace_chunk_status lttng_trace_chunk_get_creation_timestamp(
 320                 struct lttng_trace_chunk *chunk, time_t *creation_ts)
 321
 322 {
 323         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 324
 325         pthread_mutex_lock(&chunk->lock);
 326         if (chunk->timestamp_creation.is_set) {
 327                 *creation_ts = chunk->timestamp_creation.value;
 328         } else {
 329                 status = LTTNG_TRACE_CHUNK_STATUS_NONE;
 330         }
 331         pthread_mutex_unlock(&chunk->lock);
 332         return status;
 333 }
 334
 335 LTTNG_HIDDEN
 336 enum lttng_trace_chunk_status lttng_trace_chunk_get_close_timestamp(
 337                 struct lttng_trace_chunk *chunk, time_t *close_ts)
 338 {
 339         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 340
 341         pthread_mutex_lock(&chunk->lock);
 342         if (chunk->timestamp_close.is_set) {
 343                 *close_ts = chunk->timestamp_close.value;
 344         } else {
 345                 status = LTTNG_TRACE_CHUNK_STATUS_NONE;
 346         }
 347         pthread_mutex_unlock(&chunk->lock);
 348         return status;
 349 }
 350
 351 LTTNG_HIDDEN
 352 enum lttng_trace_chunk_status lttng_trace_chunk_set_close_timestamp(
 353                 struct lttng_trace_chunk *chunk, time_t close_ts)
 354 {
 355         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 356
 357         pthread_mutex_lock(&chunk->lock);
 358         if (!chunk->timestamp_creation.is_set) {
 359                 ERR("Failed to set trace chunk close timestamp: creation timestamp is unset");
 360                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_OPERATION;
 361                 goto end;
 362         }
 363         if (chunk->timestamp_creation.value > close_ts) {
 364                 ERR("Failed to set trace chunk close timestamp: close timestamp is before creation timestamp");
 365                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT;
 366                 goto end;
 367         }
 368         LTTNG_OPTIONAL_SET(&chunk->timestamp_close, close_ts);
 369         if (!chunk->name_overridden) {
 370                 free(chunk->name);
 371                 chunk->name = generate_chunk_name(LTTNG_OPTIONAL_GET(chunk->id),
 372                                 LTTNG_OPTIONAL_GET(chunk->timestamp_creation),
 373                                 &close_ts);
 374                 if (!chunk->name) {
 375                         status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 376                 }
 377         }
 378 end:
 379         pthread_mutex_unlock(&chunk->lock);
 380         return status;
 381 }
 382
 383 LTTNG_HIDDEN
 384 enum lttng_trace_chunk_status lttng_trace_chunk_get_name(
 385                 struct lttng_trace_chunk *chunk, const char **name,
 386                 bool *name_overridden)
 387 {
 388         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 389
 390         pthread_mutex_lock(&chunk->lock);
 391         if (name_overridden) {
 392                 *name_overridden = chunk->name_overridden;
 393         }
 394         if (!chunk->name) {
 395                 status = LTTNG_TRACE_CHUNK_STATUS_NONE;
 396                 goto end;
 397         }
 398         *name = chunk->name;
 399 end:
 400         pthread_mutex_unlock(&chunk->lock);
 401         return status;
 402 }
 403
 404 static
 405 bool is_valid_chunk_name(const char *name)
 406 {
 407         size_t len;
 408
 409         if (!name) {
 410                 return false;
 411         }
 412
 413         len = strnlen(name, LTTNG_NAME_MAX);
 414         if (len == 0 || len == LTTNG_NAME_MAX) {
 415                 return false;
 416         }
 417
 418         if (strchr(name, '/') || strchr(name, '.')) {
 419                 return false;
 420         }
 421
 422         return true;
 423 }
 424
 425 LTTNG_HIDDEN
 426 enum lttng_trace_chunk_status lttng_trace_chunk_override_name(
 427                 struct lttng_trace_chunk *chunk, const char *name)
 428
 429 {
 430         char *new_name;
 431         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 432
 433         if (!is_valid_chunk_name(name)) {
 434                 ERR("Attempted to set an invalid name on a trace chunk: name = %s",
 435                                 name ? : "NULL");
 436                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT;
 437                 goto end;
 438         }
 439
 440         pthread_mutex_lock(&chunk->lock);
 441         if (!chunk->id.is_set) {
 442                 ERR("Attempted to set an override name on an anonymous trace chunk: name = %s",
 443                                 name);
 444                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_OPERATION;
 445                 goto end_unlock;
 446         }
 447         new_name = strdup(name);
 448         if (!new_name) {
 449                 ERR("Failed to allocate new trace chunk name");
 450                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 451                 goto end_unlock;
 452         }
 453         free(chunk->name);
 454         chunk->name = new_name;
 455         chunk->name_overridden = true;
 456 end_unlock:
 457         pthread_mutex_unlock(&chunk->lock);
 458 end:
 459         return status;
 460 }
 461
 462 LTTNG_HIDDEN
 463 enum lttng_trace_chunk_status lttng_trace_chunk_get_credentials(
 464                 struct lttng_trace_chunk *chunk,
 465                 struct lttng_credentials *credentials)
 466 {
 467         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 468
 469         pthread_mutex_lock(&chunk->lock);
 470         if (chunk->credentials.is_set) {
 471                 if (chunk->credentials.value.use_current_user) {
 472                         credentials->uid = geteuid();
 473                         credentials->gid = getegid();
 474                 } else {
 475                         *credentials = chunk->credentials.value.user;
 476                 }
 477         } else {
 478                 status = LTTNG_TRACE_CHUNK_STATUS_NONE;
 479         }
 480         pthread_mutex_unlock(&chunk->lock);
 481         return status;
 482 }
 483
 484 LTTNG_HIDDEN
 485 enum lttng_trace_chunk_status lttng_trace_chunk_set_credentials(
 486                 struct lttng_trace_chunk *chunk,
 487                 const struct lttng_credentials *user_credentials)
 488 {
 489         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 490         const struct chunk_credentials credentials = {
 491                 .user = *user_credentials,
 492                 .use_current_user = false,
 493         };
 494
 495         pthread_mutex_lock(&chunk->lock);
 496         if (chunk->credentials.is_set) {
 497                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 498                 goto end;
 499         }
 500         LTTNG_OPTIONAL_SET(&chunk->credentials, credentials);
 501 end:
 502         pthread_mutex_unlock(&chunk->lock);
 503         return status;
 504 }
 505
 506 LTTNG_HIDDEN
 507 enum lttng_trace_chunk_status lttng_trace_chunk_set_credentials_current_user(
 508                 struct lttng_trace_chunk *chunk)
 509 {
 510         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 511         const struct chunk_credentials credentials = {
 512                 .use_current_user = true,
 513         };
 514
 515         pthread_mutex_lock(&chunk->lock);
 516         if (chunk->credentials.is_set) {
 517                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 518                 goto end;
 519         }
 520         LTTNG_OPTIONAL_SET(&chunk->credentials, credentials);
 521 end:
 522         pthread_mutex_unlock(&chunk->lock);
 523         return status;
 524 }
 525
 526
 527 LTTNG_HIDDEN
 528 enum lttng_trace_chunk_status lttng_trace_chunk_set_as_owner(
 529                 struct lttng_trace_chunk *chunk,
 530                 struct lttng_directory_handle *session_output_directory)
 531 {
 532         int ret;
 533         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 534         struct lttng_directory_handle chunk_directory_handle;
 535
 536         pthread_mutex_lock(&chunk->lock);
 537         if (chunk->mode.is_set) {
 538                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_OPERATION;
 539                 goto end;
 540         }
 541         if (!chunk->credentials.is_set) {
 542                 /*
 543                  * Fatal error, credentials must be set before a
 544                  * directory is created.
 545                  */
 546                 ERR("Credentials of trace chunk are unset: refusing to set session output directory");
 547                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 548                 goto end;
 549         }
 550
 551         if (chunk->name) {
 552                 /*
 553                  * A nameless chunk does not need its own output directory.
 554                  * The session's output directory will be used.
 555                  */
 556                 ret = lttng_directory_handle_create_subdirectory_as_user(
 557                                 session_output_directory,
 558                                 chunk->name,
 559                                 DIR_CREATION_MODE,
 560                                 !chunk->credentials.value.use_current_user ?
 561                                         &chunk->credentials.value.user : NULL);
 562                 if (ret) {
 563                         PERROR("Failed to create chunk output directory \"%s\"",
 564                                 chunk->name);
 565                         status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 566                         goto end;
 567                 }
 568         }
 569         ret = lttng_directory_handle_init_from_handle(&chunk_directory_handle,
 570                         chunk->name,
 571                         session_output_directory);
 572         if (ret) {
 573                 /* The function already logs on all error paths. */
 574                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 575                 goto end;
 576         }
 577         LTTNG_OPTIONAL_SET(&chunk->session_output_directory,
 578                         lttng_directory_handle_move(session_output_directory));
 579         LTTNG_OPTIONAL_SET(&chunk->chunk_directory,
 580                         lttng_directory_handle_move(&chunk_directory_handle));
 581         LTTNG_OPTIONAL_SET(&chunk->mode, TRACE_CHUNK_MODE_OWNER);
 582 end:
 583         pthread_mutex_unlock(&chunk->lock);
 584         return status;
 585 }
 586
 587 LTTNG_HIDDEN
 588 enum lttng_trace_chunk_status lttng_trace_chunk_set_as_user(
 589                 struct lttng_trace_chunk *chunk,
 590                 struct lttng_directory_handle *chunk_directory)
 591 {
 592         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 593
 594         pthread_mutex_lock(&chunk->lock);
 595         if (chunk->mode.is_set) {
 596                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_OPERATION;
 597                 goto end;
 598         }
 599         if (!chunk->credentials.is_set) {
 600                 ERR("Credentials of trace chunk are unset: refusing to set chunk output directory");
 601                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 602                 goto end;
 603         }
 604         LTTNG_OPTIONAL_SET(&chunk->chunk_directory,
 605                         lttng_directory_handle_move(chunk_directory));
 606         LTTNG_OPTIONAL_SET(&chunk->mode, TRACE_CHUNK_MODE_USER);
 607 end:
 608         pthread_mutex_unlock(&chunk->lock);
 609         return status;
 610 }
 611
 612 LTTNG_HIDDEN
 613 enum lttng_trace_chunk_status lttng_trace_chunk_get_chunk_directory_handle(
 614                 struct lttng_trace_chunk *chunk,
 615                 const struct lttng_directory_handle **handle)
 616 {
 617         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 618
 619         pthread_mutex_lock(&chunk->lock);
 620         if (!chunk->chunk_directory.is_set) {
 621                 status = LTTNG_TRACE_CHUNK_STATUS_NONE;
 622                 goto end;
 623         }
 624
 625         *handle = &chunk->chunk_directory.value;
 626 end:
 627         pthread_mutex_unlock(&chunk->lock);
 628         return status;
 629 }
 630
 631 /* Add a top-level directory to the trace chunk if it was previously unknown. */
 632 static
 633 int add_top_level_directory_unique(struct lttng_trace_chunk *chunk,
 634                 const char *new_path)
 635 {
 636         int ret = 0;
 637         bool found = false;
 638         size_t i, count = lttng_dynamic_pointer_array_get_count(
 639                         &chunk->top_level_directories);
 640         const char *new_path_separator_pos = strchr(new_path, '/');
 641         const ptrdiff_t new_path_top_level_len = new_path_separator_pos ?
 642                         new_path_separator_pos - new_path : strlen(new_path);
 643
 644         for (i = 0; i < count; i++) {
 645                 const char *path = lttng_dynamic_pointer_array_get_pointer(
 646                                 &chunk->top_level_directories, i);
 647                 const ptrdiff_t path_top_level_len = strlen(path);
 648
 649                 if (path_top_level_len != new_path_top_level_len) {
 650                         continue;
 651                 }
 652                 if (!strncmp(path, new_path, path_top_level_len)) {
 653                         found = true;
 654                         break;
 655                 }
 656         }
 657
 658         if (!found) {
 659                 char *copy = strndup(new_path, new_path_top_level_len);
 660
 661                 DBG("Adding new top-level directory \"%s\" to trace chunk \"%s\"",
 662                                 new_path, chunk->name ? : "(unnamed)");
 663                 if (!copy) {
 664                         PERROR("Failed to copy path");
 665                         ret = -1;
 666                         goto end;
 667                 }
 668                 ret = lttng_dynamic_pointer_array_add_pointer(
 669                                 &chunk->top_level_directories, copy);
 670                 if (ret) {
 671                         ERR("Allocation failure while adding top-level directory entry to a trace chunk");
 672                         free(copy);
 673                         goto end;
 674                 }
 675         }
 676 end:
 677         return ret;
 678 }
 679
 680 LTTNG_HIDDEN
 681 enum lttng_trace_chunk_status lttng_trace_chunk_create_subdirectory(
 682                 struct lttng_trace_chunk *chunk,
 683                 const char *path)
 684 {
 685         int ret;
 686         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 687
 688         DBG("Creating trace chunk subdirectory \"%s\"", path);
 689         pthread_mutex_lock(&chunk->lock);
 690         if (!chunk->credentials.is_set) {
 691                 /*
 692                  * Fatal error, credentials must be set before a
 693                  * directory is created.
 694                  */
 695                 ERR("Credentials of trace chunk are unset: refusing to create subdirectory \"%s\"",
 696                                 path);
 697                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 698                 goto end;
 699         }
 700         if (!chunk->mode.is_set ||
 701                         chunk->mode.value != TRACE_CHUNK_MODE_OWNER) {
 702                 ERR("Attempted to create trace chunk subdirectory \"%s\" through a non-owner chunk",
 703                                 path);
 704                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_OPERATION;
 705                 goto end;
 706         }
 707         if (!chunk->chunk_directory.is_set) {
 708                 ERR("Attempted to create trace chunk subdirectory \"%s\" before setting the chunk output directory",
 709                                 path);
 710                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 711                 goto end;
 712         }
 713         if (*path == '/') {
 714                 ERR("Refusing to create absolute trace chunk directory \"%s\"",
 715                                 path);
 716                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT;
 717                 goto end;
 718         }
 719         ret = lttng_directory_handle_create_subdirectory_recursive_as_user(
 720                         &chunk->chunk_directory.value, path,
 721                         DIR_CREATION_MODE,
 722                         chunk->credentials.value.use_current_user ?
 723                                         NULL : &chunk->credentials.value.user);
 724         if (ret) {
 725                 PERROR("Failed to create trace chunk subdirectory \"%s\"",
 726                                 path);
 727                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 728                 goto end;
 729         }
 730         ret = add_top_level_directory_unique(chunk, path);
 731         if (ret) {
 732                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 733                 goto end;
 734         }
 735 end:
 736         pthread_mutex_unlock(&chunk->lock);
 737         return status;
 738 }
 739
 740 LTTNG_HIDDEN
 741 enum lttng_trace_chunk_status lttng_trace_chunk_open_file(
 742                 struct lttng_trace_chunk *chunk, const char *file_path,
 743                 int flags, mode_t mode, int *out_fd)
 744 {
 745         int ret;
 746         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 747
 748         DBG("Opening trace chunk file \"%s\"", file_path);
 749         pthread_mutex_lock(&chunk->lock);
 750         if (!chunk->credentials.is_set) {
 751                 /*
 752                  * Fatal error, credentials must be set before a
 753                  * file is created.
 754                  */
 755                 ERR("Credentials of trace chunk are unset: refusing to open file \"%s\"",
 756                                 file_path);
 757                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 758                 goto end;
 759         }
 760         if (!chunk->chunk_directory.is_set) {
 761                 ERR("Attempted to open trace chunk file \"%s\" before setting the chunk output directory",
 762                                 file_path);
 763                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 764                 goto end;
 765         }
 766         ret = lttng_directory_handle_open_file_as_user(
 767                         &chunk->chunk_directory.value, file_path, flags, mode,
 768                         chunk->credentials.value.use_current_user ?
 769                                         NULL : &chunk->credentials.value.user);
 770         if (ret < 0) {
 771                 ERR("Failed to open file relative to trace chunk file_path = \"%s\", flags = %d, mode = %d",
 772                                 file_path, flags, (int) mode);
 773                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 774                 goto end;
 775         }
 776         *out_fd = ret;
 777 end:
 778         pthread_mutex_unlock(&chunk->lock);
 779         return status;
 780 }
 781
 782 LTTNG_HIDDEN
 783 int lttng_trace_chunk_unlink_file(struct lttng_trace_chunk *chunk,
 784                 const char *file_path)
 785 {
 786         int ret;
 787         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 788
 789         DBG("Unlinking trace chunk file \"%s\"", file_path);
 790         pthread_mutex_lock(&chunk->lock);
 791         if (!chunk->credentials.is_set) {
 792                 /*
 793                  * Fatal error, credentials must be set before a
 794                  * directory is created.
 795                  */
 796                 ERR("Credentials of trace chunk are unset: refusing to unlink file \"%s\"",
 797                                 file_path);
 798                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 799                 goto end;
 800         }
 801         if (!chunk->chunk_directory.is_set) {
 802                 ERR("Attempted to unlink trace chunk file \"%s\" before setting the chunk output directory",
 803                                 file_path);
 804                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 805                 goto end;
 806         }
 807         ret = lttng_directory_handle_unlink_file_as_user(
 808                         &chunk->chunk_directory.value, file_path,
 809                         chunk->credentials.value.use_current_user ?
 810                                         NULL : &chunk->credentials.value.user);
 811         if (ret < 0) {
 812                 status = LTTNG_TRACE_CHUNK_STATUS_ERROR;
 813                 goto end;
 814         }
 815 end:
 816         pthread_mutex_unlock(&chunk->lock);
 817         return status;
 818 }
 819
 820 static
 821 void lttng_trace_chunk_move_to_completed(struct lttng_trace_chunk *trace_chunk)
 822 {
 823         int ret;
 824         char *directory_to_rename = NULL;
 825         bool free_directory_to_rename = false;
 826         char *archived_chunk_name = NULL;
 827         const uint64_t chunk_id = LTTNG_OPTIONAL_GET(trace_chunk->id);
 828         const time_t creation_timestamp =
 829                         LTTNG_OPTIONAL_GET(trace_chunk->timestamp_creation);
 830         const time_t close_timestamp =
 831                         LTTNG_OPTIONAL_GET(trace_chunk->timestamp_close);
 832         LTTNG_OPTIONAL(struct lttng_directory_handle) archived_chunks_directory = {};
 833
 834         if (!trace_chunk->mode.is_set ||
 835                         trace_chunk->mode.value != TRACE_CHUNK_MODE_OWNER ||
 836                         !trace_chunk->session_output_directory.is_set) {
 837                 /*
 838                  * This command doesn't need to run if the output is remote
 839                  * or if the trace chunk is not owned by this process.
 840                  */
 841                 goto end;
 842         }
 843
 844         assert(trace_chunk->mode.value == TRACE_CHUNK_MODE_OWNER);
 845         assert(!trace_chunk->name_overridden);
 846
 847         /*
 848          * The fist trace chunk of a session is directly output to the
 849          * session's output folder. In this case, the top level directories
 850          * must be moved to a temporary folder before that temporary directory
 851          * is renamed to match the chunk's name.
 852          */
 853         if (chunk_id == 0) {
 854                 struct lttng_directory_handle temporary_rename_directory;
 855                 size_t i, count = lttng_dynamic_pointer_array_get_count(
 856                                 &trace_chunk->top_level_directories);
 857
 858                 ret = lttng_directory_handle_create_subdirectory_as_user(
 859                                 &trace_chunk->session_output_directory.value,
 860                                 DEFAULT_TEMPORARY_CHUNK_RENAME_DIRECTORY,
 861                                 DIR_CREATION_MODE,
 862                                 !trace_chunk->credentials.value.use_current_user ?
 863                                         &trace_chunk->credentials.value.user : NULL);
 864                 if (ret) {
 865                         PERROR("Failed to create temporary trace chunk rename directory \"%s\"",
 866                                         DEFAULT_TEMPORARY_CHUNK_RENAME_DIRECTORY);
 867                 }
 868
 869                 ret = lttng_directory_handle_init_from_handle(&temporary_rename_directory,
 870                                 DEFAULT_TEMPORARY_CHUNK_RENAME_DIRECTORY,
 871                                 &trace_chunk->session_output_directory.value);
 872                 if (ret) {
 873                         ERR("Failed to get handle to temporary trace chunk rename directory");
 874                         goto end;
 875                 }
 876
 877                 for (i = 0; i < count; i++) {
 878                         const char *top_level_name =
 879                                         lttng_dynamic_pointer_array_get_pointer(
 880                                                 &trace_chunk->top_level_directories, i);
 881
 882                         ret = lttng_directory_handle_rename_as_user(
 883                                         &trace_chunk->session_output_directory.value,
 884                                         top_level_name,
 885                                         &temporary_rename_directory,
 886                                         top_level_name,
 887                                         LTTNG_OPTIONAL_GET(trace_chunk->credentials).use_current_user ?
 888                                                 NULL :
 889                                                 &trace_chunk->credentials.value.user);
 890                         if (ret) {
 891                                 PERROR("Failed to move \"%s\" to temporary trace chunk rename directory",
 892                                                 top_level_name);
 893                                 lttng_directory_handle_fini(
 894                                                 &temporary_rename_directory);
 895                                 goto end;
 896                         }
 897                 }
 898                 lttng_directory_handle_fini(&temporary_rename_directory);
 899                 directory_to_rename = DEFAULT_TEMPORARY_CHUNK_RENAME_DIRECTORY;
 900                 free_directory_to_rename = false;
 901         } else {
 902                 directory_to_rename = generate_chunk_name(chunk_id,
 903                                 creation_timestamp, NULL);
 904                 if (!directory_to_rename) {
 905                         ERR("Failed to generate initial trace chunk name while renaming trace chunk");
 906                 }
 907                 free_directory_to_rename = true;
 908         }
 909
 910         archived_chunk_name = generate_chunk_name(chunk_id, creation_timestamp,
 911                         &close_timestamp);
 912         if (!archived_chunk_name) {
 913                 ERR("Failed to generate archived trace chunk name while renaming trace chunk");
 914                 goto end;
 915         }
 916
 917         ret = lttng_directory_handle_create_subdirectory_as_user(
 918                         &trace_chunk->session_output_directory.value,
 919                         DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY,
 920                         DIR_CREATION_MODE,
 921                         !trace_chunk->credentials.value.use_current_user ?
 922                                         &trace_chunk->credentials.value.user :
 923                                         NULL);
 924         if (ret) {
 925                 PERROR("Failed to create \"" DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY
 926                                 "\" directory for archived trace chunks");
 927                 goto end;
 928         }
 929
 930         ret = lttng_directory_handle_init_from_handle(
 931                         &archived_chunks_directory.value,
 932                         DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY,
 933                         &trace_chunk->session_output_directory.value);
 934         if (ret) {
 935                 PERROR("Failed to get handle to archived trace chunks directory");
 936                 goto end;
 937         }
 938         archived_chunks_directory.is_set = true;
 939
 940         ret = lttng_directory_handle_rename_as_user(
 941                         &trace_chunk->session_output_directory.value,
 942                         directory_to_rename,
 943                         &archived_chunks_directory.value,
 944                         archived_chunk_name,
 945                         LTTNG_OPTIONAL_GET(trace_chunk->credentials).use_current_user ?
 946                                 NULL :
 947                                 &trace_chunk->credentials.value.user);
 948         if (ret) {
 949                 PERROR("Failed to rename folder \"%s\" to \"%s\"",
 950                                 directory_to_rename, archived_chunk_name);
 951         }
 952
 953 end:
 954         if (archived_chunks_directory.is_set) {
 955                 lttng_directory_handle_fini(&archived_chunks_directory.value);
 956         }
 957         free(archived_chunk_name);
 958         if (free_directory_to_rename) {
 959                 free(directory_to_rename);
 960         }
 961 }
 962
 963 LTTNG_HIDDEN
 964 enum lttng_trace_chunk_status lttng_trace_chunk_get_close_command(
 965                 struct lttng_trace_chunk *chunk,
 966                 enum lttng_trace_chunk_command_type *command_type)
 967 {
 968         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 969
 970         pthread_mutex_lock(&chunk->lock);
 971         if (chunk->close_command.is_set) {
 972                 *command_type = chunk->close_command.value;
 973                 status = LTTNG_TRACE_CHUNK_STATUS_OK;
 974         } else {
 975                 status = LTTNG_TRACE_CHUNK_STATUS_NONE;
 976         }
 977         pthread_mutex_unlock(&chunk->lock);
 978         return status;
 979 }
 980
 981 LTTNG_HIDDEN
 982 enum lttng_trace_chunk_status lttng_trace_chunk_set_close_command(
 983                 struct lttng_trace_chunk *chunk,
 984                 enum lttng_trace_chunk_command_type close_command)
 985 {
 986         enum lttng_trace_chunk_status status = LTTNG_TRACE_CHUNK_STATUS_OK;
 987
 988         if (close_command < LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED ||
 989                         close_command >= LTTNG_TRACE_CHUNK_COMMAND_TYPE_MAX) {
 990                 status = LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT;
 991                 goto end_unlock;
 992         }
 993
 994         pthread_mutex_lock(&chunk->lock);
 995         if (chunk->close_command.is_set) {
 996                 DBG("Overriding trace chunk close command from \"%s\" to \"%s\"",
 997                                 close_command_names[chunk->close_command.value],
 998                                 close_command_names[close_command]);
 999         } else {
1000                 DBG("Setting trace chunk close command to \"%s\"",
1001                                 close_command_names[close_command]);
1002         }
1003         LTTNG_OPTIONAL_SET(&chunk->close_command, close_command);
1004         pthread_mutex_unlock(&chunk->lock);
1005 end_unlock:
1006         return status;
1007 }
1008
1009 LTTNG_HIDDEN
1010 const char *lttng_trace_chunk_command_type_get_name(
1011                 enum lttng_trace_chunk_command_type command)
1012 {
1013         switch (command) {
1014         case LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED:
1015                 return "move to completed trace chunk folder";
1016         default:
1017                 abort();
1018         }
1019 }
1020
1021 LTTNG_HIDDEN
1022 bool lttng_trace_chunk_get(struct lttng_trace_chunk *chunk)
1023 {
1024         return urcu_ref_get_unless_zero(&chunk->ref);
1025 }
1026
1027 static
1028 void free_lttng_trace_chunk_registry_element(struct rcu_head *node)
1029 {
1030         struct lttng_trace_chunk_registry_element *element =
1031                         container_of(node, typeof(*element), rcu_node);
1032
1033         lttng_trace_chunk_fini(&element->chunk);
1034         free(element);
1035 }
1036
1037 static
1038 void lttng_trace_chunk_release(struct urcu_ref *ref)
1039 {
1040         struct lttng_trace_chunk *chunk = container_of(ref, typeof(*chunk),
1041                         ref);
1042
1043         if (chunk->close_command.is_set) {
1044                 close_command_funcs[chunk->close_command.value](chunk);
1045         }
1046
1047         if (chunk->in_registry_element) {
1048                 struct lttng_trace_chunk_registry_element *element;
1049
1050                 element = container_of(chunk, typeof(*element), chunk);
1051                 if (element->registry) {
1052                         rcu_read_lock();
1053                         cds_lfht_del(element->registry->ht,
1054                                         &element->trace_chunk_registry_ht_node);
1055                         rcu_read_unlock();
1056                         call_rcu(&element->rcu_node,
1057                                         free_lttng_trace_chunk_registry_element);
1058                 } else {
1059                         /* Never published, can be free'd immediately. */
1060                         free_lttng_trace_chunk_registry_element(
1061                                         &element->rcu_node);
1062                 }
1063         } else {
1064                 /* Not RCU-protected, free immediately. */
1065                 lttng_trace_chunk_fini(chunk);
1066                 free(chunk);
1067         }
1068 }
1069
1070 LTTNG_HIDDEN
1071 void lttng_trace_chunk_put(struct lttng_trace_chunk *chunk)
1072 {
1073         if (!chunk) {
1074                 return;
1075         }
1076         assert(chunk->ref.refcount);
1077         urcu_ref_put(&chunk->ref, lttng_trace_chunk_release);
1078 }
1079
1080 LTTNG_HIDDEN
1081 struct lttng_trace_chunk_registry *lttng_trace_chunk_registry_create(void)
1082 {
1083         struct lttng_trace_chunk_registry *registry;
1084
1085         registry = zmalloc(sizeof(*registry));
1086         if (!registry) {
1087                 goto end;
1088         }
1089
1090         registry->ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
1091                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
1092         if (!registry->ht) {
1093                 goto error;
1094         }
1095 end:
1096         return registry;
1097 error:
1098         lttng_trace_chunk_registry_destroy(registry);
1099         goto end;
1100 }
1101
1102 LTTNG_HIDDEN
1103 void lttng_trace_chunk_registry_destroy(
1104                 struct lttng_trace_chunk_registry *registry)
1105 {
1106         if (!registry) {
1107                 return;
1108         }
1109         if (registry->ht) {
1110                 int ret = cds_lfht_destroy(registry->ht, NULL);
1111                 assert(!ret);
1112         }
1113         free(registry);
1114 }
1115
1116 static
1117 struct lttng_trace_chunk_registry_element *
1118 lttng_trace_chunk_registry_element_create_from_chunk(
1119                 struct lttng_trace_chunk *chunk, uint64_t session_id)
1120 {
1121         struct lttng_trace_chunk_registry_element *element =
1122                         zmalloc(sizeof(*element));
1123
1124         if (!element) {
1125                 goto end;
1126         }
1127         cds_lfht_node_init(&element->trace_chunk_registry_ht_node);
1128         element->session_id = session_id;
1129
1130         element->chunk = *chunk;
1131         lttng_trace_chunk_init(&element->chunk);
1132         if (chunk->session_output_directory.is_set) {
1133                 element->chunk.session_output_directory.value =
1134                                 lttng_directory_handle_move(
1135                                         &chunk->session_output_directory.value);
1136         }
1137         if (chunk->chunk_directory.is_set) {
1138                 element->chunk.chunk_directory.value =
1139                                 lttng_directory_handle_move(
1140                                         &chunk->chunk_directory.value);
1141         }
1142         /*
1143          * The original chunk becomes invalid; the name attribute is transferred
1144          * to the new chunk instance.
1145          */
1146         chunk->name = NULL;
1147         element->chunk.in_registry_element = true;
1148 end:
1149         return element;
1150 }
1151
1152 LTTNG_HIDDEN
1153 struct lttng_trace_chunk *
1154 lttng_trace_chunk_registry_publish_chunk(
1155                 struct lttng_trace_chunk_registry *registry,
1156                 uint64_t session_id, struct lttng_trace_chunk *chunk)
1157 {
1158         struct lttng_trace_chunk_registry_element *element;
1159         unsigned long element_hash;
1160
1161         pthread_mutex_lock(&chunk->lock);
1162         element = lttng_trace_chunk_registry_element_create_from_chunk(chunk,
1163                         session_id);
1164         pthread_mutex_unlock(&chunk->lock);
1165         if (!element) {
1166                 goto end;
1167         }
1168         /*
1169          * chunk is now invalid, the only valid operation is a 'put' from the
1170          * caller.
1171          */
1172         chunk = NULL;
1173         element_hash = lttng_trace_chunk_registry_element_hash(element);
1174
1175         rcu_read_lock();
1176         while (1) {
1177                 struct cds_lfht_node *published_node;
1178                 struct lttng_trace_chunk *published_chunk;
1179                 struct lttng_trace_chunk_registry_element *published_element;
1180
1181                 published_node = cds_lfht_add_unique(registry->ht,
1182                                 element_hash,
1183                                 lttng_trace_chunk_registry_element_match,
1184                                 element,
1185                                 &element->trace_chunk_registry_ht_node);
1186                 if (published_node == &element->trace_chunk_registry_ht_node) {
1187                         /* Successfully published the new element. */
1188                         element->registry = registry;
1189                         /* Acquire a reference for the caller. */
1190                         if (lttng_trace_chunk_get(&element->chunk)) {
1191                                 break;
1192                         } else {
1193                                 /*
1194                                  * Another thread concurrently unpublished the
1195                                  * trace chunk. This is currently unexpected.
1196                                  *
1197                                  * Re-attempt to publish.
1198                                  */
1199                                 ERR("Attemp to publish a trace chunk to the chunk registry raced with a trace chunk deletion");
1200                                 continue;
1201                         }
1202                 }
1203
1204                 /*
1205                  * An equivalent trace chunk was published before this trace
1206                  * chunk. Attempt to acquire a reference to the one that was
1207                  * already published and release the reference to the copy we
1208                  * created if successful.
1209                  */
1210                 published_element = container_of(published_node,
1211                                 typeof(*published_element),
1212                                 trace_chunk_registry_ht_node);
1213                 published_chunk = &published_element->chunk;
1214                 if (lttng_trace_chunk_get(published_chunk)) {
1215                         lttng_trace_chunk_put(&element->chunk);
1216                         element = published_element;
1217                         break;
1218                 }
1219                 /*
1220                  * A reference to the previously published trace chunk could not
1221                  * be acquired. Hence, retry to publish our copy of the trace
1222                  * chunk.
1223                  */
1224         }
1225         rcu_read_unlock();
1226 end:
1227         return element ? &element->chunk : NULL;
1228 }
1229
1230 /*
1231  * Note that the caller must be registered as an RCU thread.
1232  * However, it does not need to hold the RCU read lock. The RCU read lock is
1233  * acquired to perform the look-up in the registry's hash table and held until
1234  * after a reference to the "found" trace chunk is acquired.
1235  *
1236  * IOW, holding a reference guarantees the existence of the object for the
1237  * caller.
1238  */
1239 static
1240 struct lttng_trace_chunk *_lttng_trace_chunk_registry_find_chunk(
1241                 const struct lttng_trace_chunk_registry *registry,
1242                 uint64_t session_id, uint64_t *chunk_id)
1243 {
1244         const struct lttng_trace_chunk_registry_element target_element = {
1245                 .chunk.id.is_set = !!chunk_id,
1246                 .chunk.id.value = chunk_id ? *chunk_id : 0,
1247                 .session_id = session_id,
1248         };
1249         const unsigned long element_hash =
1250                         lttng_trace_chunk_registry_element_hash(
1251                                 &target_element);
1252         struct cds_lfht_node *published_node;
1253         struct lttng_trace_chunk_registry_element *published_element;
1254         struct lttng_trace_chunk *published_chunk = NULL;
1255         struct cds_lfht_iter iter;
1256
1257         rcu_read_lock();
1258         cds_lfht_lookup(registry->ht,
1259                         element_hash,
1260                         lttng_trace_chunk_registry_element_match,
1261                         &target_element,
1262                         &iter);
1263         published_node = cds_lfht_iter_get_node(&iter);
1264         if (!published_node) {
1265                 goto end;
1266         }
1267
1268         published_element = container_of(published_node,
1269                         typeof(*published_element),
1270                         trace_chunk_registry_ht_node);
1271         if (lttng_trace_chunk_get(&published_element->chunk)) {
1272                 published_chunk = &published_element->chunk;
1273         }
1274 end:
1275         rcu_read_unlock();
1276         return published_chunk;
1277 }
1278
1279 LTTNG_HIDDEN
1280 struct lttng_trace_chunk *
1281 lttng_trace_chunk_registry_find_chunk(
1282                 const struct lttng_trace_chunk_registry *registry,
1283                 uint64_t session_id, uint64_t chunk_id)
1284 {
1285         return _lttng_trace_chunk_registry_find_chunk(registry,
1286                         session_id, &chunk_id);
1287 }
1288
1289 LTTNG_HIDDEN
1290 struct lttng_trace_chunk *
1291 lttng_trace_chunk_registry_find_anonymous_chunk(
1292                 const struct lttng_trace_chunk_registry *registry,
1293                 uint64_t session_id)
1294 {
1295         return _lttng_trace_chunk_registry_find_chunk(registry,
1296                         session_id, NULL);
1297 }
1298
1299 unsigned int lttng_trace_chunk_registry_put_each_chunk(
1300                 struct lttng_trace_chunk_registry *registry)
1301 {
1302         struct cds_lfht_iter iter;
1303         struct lttng_trace_chunk_registry_element *chunk_element;
1304         unsigned int trace_chunks_left = 0;
1305
1306         DBG("Releasing trace chunk registry to all trace chunks");
1307         rcu_read_lock();
1308         cds_lfht_for_each_entry(registry->ht,
1309                         &iter, chunk_element, trace_chunk_registry_ht_node) {
1310                 const char *chunk_id_str = "none";
1311                 char chunk_id_buf[MAX_INT_DEC_LEN(uint64_t)];
1312
1313                 pthread_mutex_lock(&chunk_element->chunk.lock);
1314                 if (chunk_element->chunk.id.is_set) {
1315                         int fmt_ret;
1316
1317                         fmt_ret = snprintf(chunk_id_buf, sizeof(chunk_id_buf),
1318                                         "%" PRIu64,
1319                                         chunk_element->chunk.id.value);
1320                         if (fmt_ret < 0 || fmt_ret >= sizeof(chunk_id_buf)) {
1321                                 chunk_id_str = "formatting error";
1322                         } else {
1323                                 chunk_id_str = chunk_id_buf;
1324                         }
1325                 }
1326
1327                 DBG("Releasing reference to trace chunk: session_id = %" PRIu64
1328                                 "chunk_id = %s, name = \"%s\", status = %s",
1329                                 chunk_element->session_id,
1330                                 chunk_id_str,
1331                                 chunk_element->chunk.name ? : "none",
1332                                 chunk_element->chunk.close_command.is_set ?
1333                                                 "open" : "closed");
1334                 pthread_mutex_unlock(&chunk_element->chunk.lock);
1335                 lttng_trace_chunk_put(&chunk_element->chunk);
1336                 trace_chunks_left++;
1337         }
1338         rcu_read_unlock();
1339         DBG("Released reference to %u trace chunks in %s()", trace_chunks_left,
1340                         __FUNCTION__);
1341
1342         return trace_chunks_left;
1343 }