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