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