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