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