src/common/trace-chunk.c

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