src/common/kernel-consumer/kernel-consumer.c

   1 /*
   2  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
   3  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   4  * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License, version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18  */
  19
  20 #define _LGPL_SOURCE
  21 #include <assert.h>
  22 #include <poll.h>
  23 #include <pthread.h>
  24 #include <stdlib.h>
  25 #include <string.h>
  26 #include <sys/mman.h>
  27 #include <sys/socket.h>
  28 #include <sys/types.h>
  29 #include <inttypes.h>
  30 #include <unistd.h>
  31 #include <sys/stat.h>
  32 #include <stdint.h>
  33
  34 #include <bin/lttng-consumerd/health-consumerd.h>
  35 #include <common/common.h>
  36 #include <common/kernel-ctl/kernel-ctl.h>
  37 #include <common/sessiond-comm/sessiond-comm.h>
  38 #include <common/sessiond-comm/relayd.h>
  39 #include <common/compat/fcntl.h>
  40 #include <common/compat/endian.h>
  41 #include <common/pipe.h>
  42 #include <common/relayd/relayd.h>
  43 #include <common/utils.h>
  44 #include <common/consumer/consumer-stream.h>
  45 #include <common/index/index.h>
  46 #include <common/consumer/consumer-timer.h>
  47 #include <common/optional.h>
  48 #include <common/buffer-view.h>
  49 #include <common/consumer/consumer.h>
  50 #include <common/consumer/metadata-bucket.h>
  51
  52 #include "kernel-consumer.h"
  53
  54 extern struct lttng_consumer_global_data consumer_data;
  55 extern int consumer_poll_timeout;
  56
  57 /*
  58  * Take a snapshot for a specific fd
  59  *
  60  * Returns 0 on success, < 0 on error
  61  */
  62 int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream)
  63 {
  64         int ret = 0;
  65         int infd = stream->wait_fd;
  66
  67         ret = kernctl_snapshot(infd);
  68         /*
  69          * -EAGAIN is not an error, it just means that there is no data to
  70          *  be read.
  71          */
  72         if (ret != 0 && ret != -EAGAIN) {
  73                 PERROR("Getting sub-buffer snapshot.");
  74         }
  75
  76         return ret;
  77 }
  78
  79 /*
  80  * Sample consumed and produced positions for a specific fd.
  81  *
  82  * Returns 0 on success, < 0 on error.
  83  */
  84 int lttng_kconsumer_sample_snapshot_positions(
  85                 struct lttng_consumer_stream *stream)
  86 {
  87         assert(stream);
  88
  89         return kernctl_snapshot_sample_positions(stream->wait_fd);
  90 }
  91
  92 /*
  93  * Get the produced position
  94  *
  95  * Returns 0 on success, < 0 on error
  96  */
  97 int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
  98                 unsigned long *pos)
  99 {
 100         int ret;
 101         int infd = stream->wait_fd;
 102
 103         ret = kernctl_snapshot_get_produced(infd, pos);
 104         if (ret != 0) {
 105                 PERROR("kernctl_snapshot_get_produced");
 106         }
 107
 108         return ret;
 109 }
 110
 111 /*
 112  * Get the consumerd position
 113  *
 114  * Returns 0 on success, < 0 on error
 115  */
 116 int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
 117                 unsigned long *pos)
 118 {
 119         int ret;
 120         int infd = stream->wait_fd;
 121
 122         ret = kernctl_snapshot_get_consumed(infd, pos);
 123         if (ret != 0) {
 124                 PERROR("kernctl_snapshot_get_consumed");
 125         }
 126
 127         return ret;
 128 }
 129
 130 static
 131 int get_current_subbuf_addr(struct lttng_consumer_stream *stream,
 132                 const char **addr)
 133 {
 134         int ret;
 135         unsigned long mmap_offset;
 136         const char *mmap_base = stream->mmap_base;
 137
 138         ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
 139         if (ret < 0) {
 140                 PERROR("Failed to get mmap read offset");
 141                 goto error;
 142         }
 143
 144         *addr = mmap_base + mmap_offset;
 145 error:
 146         return ret;
 147 }
 148
 149 /*
 150  * Take a snapshot of all the stream of a channel
 151  * RCU read-side lock must be held across this function to ensure existence of
 152  * channel. The channel lock must be held by the caller.
 153  *
 154  * Returns 0 on success, < 0 on error
 155  */
 156 static int lttng_kconsumer_snapshot_channel(
 157                 struct lttng_consumer_channel *channel,
 158                 uint64_t key, char *path, uint64_t relayd_id,
 159                 uint64_t nb_packets_per_stream,
 160                 struct lttng_consumer_local_data *ctx)
 161 {
 162         int ret;
 163         struct lttng_consumer_stream *stream;
 164
 165         DBG("Kernel consumer snapshot channel %" PRIu64, key);
 166
 167         rcu_read_lock();
 168
 169         /* Splice is not supported yet for channel snapshot. */
 170         if (channel->output != CONSUMER_CHANNEL_MMAP) {
 171                 ERR("Unsupported output type for channel \"%s\": mmap output is required to record a snapshot",
 172                                 channel->name);
 173                 ret = -1;
 174                 goto end;
 175         }
 176
 177         cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
 178                 unsigned long consumed_pos, produced_pos;
 179
 180                 health_code_update();
 181
 182                 /*
 183                  * Lock stream because we are about to change its state.
 184                  */
 185                 pthread_mutex_lock(&stream->lock);
 186
 187                 assert(channel->trace_chunk);
 188                 if (!lttng_trace_chunk_get(channel->trace_chunk)) {
 189                         /*
 190                          * Can't happen barring an internal error as the channel
 191                          * holds a reference to the trace chunk.
 192                          */
 193                         ERR("Failed to acquire reference to channel's trace chunk");
 194                         ret = -1;
 195                         goto end_unlock;
 196                 }
 197                 assert(!stream->trace_chunk);
 198                 stream->trace_chunk = channel->trace_chunk;
 199
 200                 /*
 201                  * Assign the received relayd ID so we can use it for streaming. The streams
 202                  * are not visible to anyone so this is OK to change it.
 203                  */
 204                 stream->net_seq_idx = relayd_id;
 205                 channel->relayd_id = relayd_id;
 206                 if (relayd_id != (uint64_t) -1ULL) {
 207                         ret = consumer_send_relayd_stream(stream, path);
 208                         if (ret < 0) {
 209                                 ERR("sending stream to relayd");
 210                                 goto end_unlock;
 211                         }
 212                 } else {
 213                         ret = consumer_stream_create_output_files(stream,
 214                                         false);
 215                         if (ret < 0) {
 216                                 goto end_unlock;
 217                         }
 218                         DBG("Kernel consumer snapshot stream (%" PRIu64 ")",
 219                                         stream->key);
 220                 }
 221
 222                 ret = kernctl_buffer_flush_empty(stream->wait_fd);
 223                 if (ret < 0) {
 224                         /*
 225                          * Doing a buffer flush which does not take into
 226                          * account empty packets. This is not perfect
 227                          * for stream intersection, but required as a
 228                          * fall-back when "flush_empty" is not
 229                          * implemented by lttng-modules.
 230                          */
 231                         ret = kernctl_buffer_flush(stream->wait_fd);
 232                         if (ret < 0) {
 233                                 ERR("Failed to flush kernel stream");
 234                                 goto end_unlock;
 235                         }
 236                         goto end_unlock;
 237                 }
 238
 239                 ret = lttng_kconsumer_take_snapshot(stream);
 240                 if (ret < 0) {
 241                         ERR("Taking kernel snapshot");
 242                         goto end_unlock;
 243                 }
 244
 245                 ret = lttng_kconsumer_get_produced_snapshot(stream, &produced_pos);
 246                 if (ret < 0) {
 247                         ERR("Produced kernel snapshot position");
 248                         goto end_unlock;
 249                 }
 250
 251                 ret = lttng_kconsumer_get_consumed_snapshot(stream, &consumed_pos);
 252                 if (ret < 0) {
 253                         ERR("Consumerd kernel snapshot position");
 254                         goto end_unlock;
 255                 }
 256
 257                 consumed_pos = consumer_get_consume_start_pos(consumed_pos,
 258                                 produced_pos, nb_packets_per_stream,
 259                                 stream->max_sb_size);
 260
 261                 while ((long) (consumed_pos - produced_pos) < 0) {
 262                         ssize_t read_len;
 263                         unsigned long len, padded_len;
 264                         const char *subbuf_addr;
 265                         struct lttng_buffer_view subbuf_view;
 266
 267                         health_code_update();
 268                         DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos);
 269
 270                         ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos);
 271                         if (ret < 0) {
 272                                 if (ret != -EAGAIN) {
 273                                         PERROR("kernctl_get_subbuf snapshot");
 274                                         goto end_unlock;
 275                                 }
 276                                 DBG("Kernel consumer get subbuf failed. Skipping it.");
 277                                 consumed_pos += stream->max_sb_size;
 278                                 stream->chan->lost_packets++;
 279                                 continue;
 280                         }
 281
 282                         ret = kernctl_get_subbuf_size(stream->wait_fd, &len);
 283                         if (ret < 0) {
 284                                 ERR("Snapshot kernctl_get_subbuf_size");
 285                                 goto error_put_subbuf;
 286                         }
 287
 288                         ret = kernctl_get_padded_subbuf_size(stream->wait_fd, &padded_len);
 289                         if (ret < 0) {
 290                                 ERR("Snapshot kernctl_get_padded_subbuf_size");
 291                                 goto error_put_subbuf;
 292                         }
 293
 294                         ret = get_current_subbuf_addr(stream, &subbuf_addr);
 295                         if (ret) {
 296                                 goto error_put_subbuf;
 297                         }
 298
 299                         subbuf_view = lttng_buffer_view_init(
 300                                         subbuf_addr, 0, padded_len);
 301                         read_len = lttng_consumer_on_read_subbuffer_mmap(
 302                                         stream, &subbuf_view,
 303                                         padded_len - len);
 304                         /*
 305                          * We write the padded len in local tracefiles but the data len
 306                          * when using a relay. Display the error but continue processing
 307                          * to try to release the subbuffer.
 308                          */
 309                         if (relayd_id != (uint64_t) -1ULL) {
 310                                 if (read_len != len) {
 311                                         ERR("Error sending to the relay (ret: %zd != len: %lu)",
 312                                                         read_len, len);
 313                                 }
 314                         } else {
 315                                 if (read_len != padded_len) {
 316                                         ERR("Error writing to tracefile (ret: %zd != len: %lu)",
 317                                                         read_len, padded_len);
 318                                 }
 319                         }
 320
 321                         ret = kernctl_put_subbuf(stream->wait_fd);
 322                         if (ret < 0) {
 323                                 ERR("Snapshot kernctl_put_subbuf");
 324                                 goto end_unlock;
 325                         }
 326                         consumed_pos += stream->max_sb_size;
 327                 }
 328
 329                 if (relayd_id == (uint64_t) -1ULL) {
 330                         if (stream->out_fd >= 0) {
 331                                 ret = close(stream->out_fd);
 332                                 if (ret < 0) {
 333                                         PERROR("Kernel consumer snapshot close out_fd");
 334                                         goto end_unlock;
 335                                 }
 336                                 stream->out_fd = -1;
 337                         }
 338                 } else {
 339                         close_relayd_stream(stream);
 340                         stream->net_seq_idx = (uint64_t) -1ULL;
 341                 }
 342                 lttng_trace_chunk_put(stream->trace_chunk);
 343                 stream->trace_chunk = NULL;
 344                 pthread_mutex_unlock(&stream->lock);
 345         }
 346
 347         /* All good! */
 348         ret = 0;
 349         goto end;
 350
 351 error_put_subbuf:
 352         ret = kernctl_put_subbuf(stream->wait_fd);
 353         if (ret < 0) {
 354                 ERR("Snapshot kernctl_put_subbuf error path");
 355         }
 356 end_unlock:
 357         pthread_mutex_unlock(&stream->lock);
 358 end:
 359         rcu_read_unlock();
 360         return ret;
 361 }
 362
 363 /*
 364  * Read the whole metadata available for a snapshot.
 365  * RCU read-side lock must be held across this function to ensure existence of
 366  * metadata_channel. The channel lock must be held by the caller.
 367  *
 368  * Returns 0 on success, < 0 on error
 369  */
 370 static int lttng_kconsumer_snapshot_metadata(
 371                 struct lttng_consumer_channel *metadata_channel,
 372                 uint64_t key, char *path, uint64_t relayd_id,
 373                 struct lttng_consumer_local_data *ctx)
 374 {
 375         int ret, use_relayd = 0;
 376         ssize_t ret_read;
 377         struct lttng_consumer_stream *metadata_stream;
 378
 379         assert(ctx);
 380
 381         DBG("Kernel consumer snapshot metadata with key %" PRIu64 " at path %s",
 382                         key, path);
 383
 384         rcu_read_lock();
 385
 386         metadata_stream = metadata_channel->metadata_stream;
 387         assert(metadata_stream);
 388
 389         pthread_mutex_lock(&metadata_stream->lock);
 390         assert(metadata_channel->trace_chunk);
 391         assert(metadata_stream->trace_chunk);
 392
 393         /* Flag once that we have a valid relayd for the stream. */
 394         if (relayd_id != (uint64_t) -1ULL) {
 395                 use_relayd = 1;
 396         }
 397
 398         if (use_relayd) {
 399                 ret = consumer_send_relayd_stream(metadata_stream, path);
 400                 if (ret < 0) {
 401                         goto error_snapshot;
 402                 }
 403         } else {
 404                 ret = consumer_stream_create_output_files(metadata_stream,
 405                                 false);
 406                 if (ret < 0) {
 407                         goto error_snapshot;
 408                 }
 409         }
 410
 411         do {
 412                 health_code_update();
 413
 414                 ret_read = lttng_consumer_read_subbuffer(metadata_stream, ctx, true);
 415                 if (ret_read < 0) {
 416                         if (ret_read != -EAGAIN) {
 417                                 ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)",
 418                                                 ret_read);
 419                                 ret = ret_read;
 420                                 goto error_snapshot;
 421                         }
 422                         /* ret_read is negative at this point so we will exit the loop. */
 423                         continue;
 424                 }
 425         } while (ret_read >= 0);
 426
 427         if (use_relayd) {
 428                 close_relayd_stream(metadata_stream);
 429                 metadata_stream->net_seq_idx = (uint64_t) -1ULL;
 430         } else {
 431                 if (metadata_stream->out_fd >= 0) {
 432                         ret = close(metadata_stream->out_fd);
 433                         if (ret < 0) {
 434                                 PERROR("Kernel consumer snapshot metadata close out_fd");
 435                                 /*
 436                                  * Don't go on error here since the snapshot was successful at this
 437                                  * point but somehow the close failed.
 438                                  */
 439                         }
 440                         metadata_stream->out_fd = -1;
 441                         lttng_trace_chunk_put(metadata_stream->trace_chunk);
 442                         metadata_stream->trace_chunk = NULL;
 443                 }
 444         }
 445
 446         ret = 0;
 447 error_snapshot:
 448         pthread_mutex_unlock(&metadata_stream->lock);
 449         cds_list_del(&metadata_stream->send_node);
 450         consumer_stream_destroy(metadata_stream, NULL);
 451         metadata_channel->metadata_stream = NULL;
 452         rcu_read_unlock();
 453         return ret;
 454 }
 455
 456 /*
 457  * Receive command from session daemon and process it.
 458  *
 459  * Return 1 on success else a negative value or 0.
 460  */
 461 int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 462                 int sock, struct pollfd *consumer_sockpoll)
 463 {
 464         ssize_t ret;
 465         enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
 466         struct lttcomm_consumer_msg msg;
 467
 468         health_code_update();
 469
 470         ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
 471         if (ret != sizeof(msg)) {
 472                 if (ret > 0) {
 473                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
 474                         ret = -1;
 475                 }
 476                 return ret;
 477         }
 478
 479         health_code_update();
 480
 481         /* Deprecated command */
 482         assert(msg.cmd_type != LTTNG_CONSUMER_STOP);
 483
 484         health_code_update();
 485
 486         /* relayd needs RCU read-side protection */
 487         rcu_read_lock();
 488
 489         switch (msg.cmd_type) {
 490         case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
 491         {
 492                 /* Session daemon status message are handled in the following call. */
 493                 consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
 494                                 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
 495                                 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
 496                                 msg.u.relayd_sock.relayd_session_id);
 497                 goto end_nosignal;
 498         }
 499         case LTTNG_CONSUMER_ADD_CHANNEL:
 500         {
 501                 struct lttng_consumer_channel *new_channel;
 502                 int ret_recv;
 503                 const uint64_t chunk_id = msg.u.channel.chunk_id.value;
 504
 505                 health_code_update();
 506
 507                 /* First send a status message before receiving the fds. */
 508                 ret = consumer_send_status_msg(sock, ret_code);
 509                 if (ret < 0) {
 510                         /* Somehow, the session daemon is not responding anymore. */
 511                         goto error_fatal;
 512                 }
 513
 514                 health_code_update();
 515
 516                 DBG("consumer_add_channel %" PRIu64, msg.u.channel.channel_key);
 517                 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
 518                                 msg.u.channel.session_id,
 519                                 msg.u.channel.chunk_id.is_set ?
 520                                                 &chunk_id : NULL,
 521                                 msg.u.channel.pathname,
 522                                 msg.u.channel.name,
 523                                 msg.u.channel.relayd_id, msg.u.channel.output,
 524                                 msg.u.channel.tracefile_size,
 525                                 msg.u.channel.tracefile_count, 0,
 526                                 msg.u.channel.monitor,
 527                                 msg.u.channel.live_timer_interval,
 528                                 msg.u.channel.is_live,
 529                                 NULL, NULL);
 530                 if (new_channel == NULL) {
 531                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 532                         goto end_nosignal;
 533                 }
 534                 new_channel->nb_init_stream_left = msg.u.channel.nb_init_streams;
 535                 switch (msg.u.channel.output) {
 536                 case LTTNG_EVENT_SPLICE:
 537                         new_channel->output = CONSUMER_CHANNEL_SPLICE;
 538                         break;
 539                 case LTTNG_EVENT_MMAP:
 540                         new_channel->output = CONSUMER_CHANNEL_MMAP;
 541                         break;
 542                 default:
 543                         ERR("Channel output unknown %d", msg.u.channel.output);
 544                         goto end_nosignal;
 545                 }
 546
 547                 /* Translate and save channel type. */
 548                 switch (msg.u.channel.type) {
 549                 case CONSUMER_CHANNEL_TYPE_DATA:
 550                 case CONSUMER_CHANNEL_TYPE_METADATA:
 551                         new_channel->type = msg.u.channel.type;
 552                         break;
 553                 default:
 554                         assert(0);
 555                         goto end_nosignal;
 556                 };
 557
 558                 health_code_update();
 559
 560                 if (ctx->on_recv_channel != NULL) {
 561                         ret_recv = ctx->on_recv_channel(new_channel);
 562                         if (ret_recv == 0) {
 563                                 ret = consumer_add_channel(new_channel, ctx);
 564                         } else if (ret_recv < 0) {
 565                                 goto end_nosignal;
 566                         }
 567                 } else {
 568                         ret = consumer_add_channel(new_channel, ctx);
 569                 }
 570                 if (msg.u.channel.type == CONSUMER_CHANNEL_TYPE_DATA && !ret) {
 571                         int monitor_start_ret;
 572
 573                         DBG("Consumer starting monitor timer");
 574                         consumer_timer_live_start(new_channel,
 575                                         msg.u.channel.live_timer_interval);
 576                         monitor_start_ret = consumer_timer_monitor_start(
 577                                         new_channel,
 578                                         msg.u.channel.monitor_timer_interval);
 579                         if (monitor_start_ret < 0) {
 580                                 ERR("Starting channel monitoring timer failed");
 581                                 goto end_nosignal;
 582                         }
 583
 584                 }
 585
 586                 health_code_update();
 587
 588                 /* If we received an error in add_channel, we need to report it. */
 589                 if (ret < 0) {
 590                         ret = consumer_send_status_msg(sock, ret);
 591                         if (ret < 0) {
 592                                 goto error_fatal;
 593                         }
 594                         goto end_nosignal;
 595                 }
 596
 597                 goto end_nosignal;
 598         }
 599         case LTTNG_CONSUMER_ADD_STREAM:
 600         {
 601                 int fd;
 602                 struct lttng_pipe *stream_pipe;
 603                 struct lttng_consumer_stream *new_stream;
 604                 struct lttng_consumer_channel *channel;
 605                 int alloc_ret = 0;
 606
 607                 /*
 608                  * Get stream's channel reference. Needed when adding the stream to the
 609                  * global hash table.
 610                  */
 611                 channel = consumer_find_channel(msg.u.stream.channel_key);
 612                 if (!channel) {
 613                         /*
 614                          * We could not find the channel. Can happen if cpu hotplug
 615                          * happens while tearing down.
 616                          */
 617                         ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key);
 618                         ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
 619                 }
 620
 621                 health_code_update();
 622
 623                 /* First send a status message before receiving the fds. */
 624                 ret = consumer_send_status_msg(sock, ret_code);
 625                 if (ret < 0) {
 626                         /* Somehow, the session daemon is not responding anymore. */
 627                         goto error_add_stream_fatal;
 628                 }
 629
 630                 health_code_update();
 631
 632                 if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
 633                         /* Channel was not found. */
 634                         goto error_add_stream_nosignal;
 635                 }
 636
 637                 /* Blocking call */
 638                 health_poll_entry();
 639                 ret = lttng_consumer_poll_socket(consumer_sockpoll);
 640                 health_poll_exit();
 641                 if (ret) {
 642                         goto error_add_stream_fatal;
 643                 }
 644
 645                 health_code_update();
 646
 647                 /* Get stream file descriptor from socket */
 648                 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 649                 if (ret != sizeof(fd)) {
 650                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 651                         goto end;
 652                 }
 653
 654                 health_code_update();
 655
 656                 /*
 657                  * Send status code to session daemon only if the recv works. If the
 658                  * above recv() failed, the session daemon is notified through the
 659                  * error socket and the teardown is eventually done.
 660                  */
 661                 ret = consumer_send_status_msg(sock, ret_code);
 662                 if (ret < 0) {
 663                         /* Somehow, the session daemon is not responding anymore. */
 664                         goto error_add_stream_nosignal;
 665                 }
 666
 667                 health_code_update();
 668
 669                 pthread_mutex_lock(&channel->lock);
 670                 new_stream = consumer_stream_create(
 671                                 channel,
 672                                 channel->key,
 673                                 fd,
 674                                 channel->name,
 675                                 channel->relayd_id,
 676                                 channel->session_id,
 677                                 channel->trace_chunk,
 678                                 msg.u.stream.cpu,
 679                                 &alloc_ret,
 680                                 channel->type,
 681                                 channel->monitor);
 682                 if (new_stream == NULL) {
 683                         switch (alloc_ret) {
 684                         case -ENOMEM:
 685                         case -EINVAL:
 686                         default:
 687                                 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 688                                 break;
 689                         }
 690                         pthread_mutex_unlock(&channel->lock);
 691                         goto error_add_stream_nosignal;
 692                 }
 693
 694                 new_stream->wait_fd = fd;
 695                 ret = kernctl_get_max_subbuf_size(new_stream->wait_fd,
 696                                 &new_stream->max_sb_size);
 697                 if (ret < 0) {
 698                         pthread_mutex_unlock(&channel->lock);
 699                         ERR("Failed to get kernel maximal subbuffer size");
 700                         goto error_add_stream_nosignal;
 701                 }
 702
 703                 consumer_stream_update_channel_attributes(new_stream,
 704                                 channel);
 705
 706                 /*
 707                  * We've just assigned the channel to the stream so increment the
 708                  * refcount right now. We don't need to increment the refcount for
 709                  * streams in no monitor because we handle manually the cleanup of
 710                  * those. It is very important to make sure there is NO prior
 711                  * consumer_del_stream() calls or else the refcount will be unbalanced.
 712                  */
 713                 if (channel->monitor) {
 714                         uatomic_inc(&new_stream->chan->refcount);
 715                 }
 716
 717                 /*
 718                  * The buffer flush is done on the session daemon side for the kernel
 719                  * so no need for the stream "hangup_flush_done" variable to be
 720                  * tracked. This is important for a kernel stream since we don't rely
 721                  * on the flush state of the stream to read data. It's not the case for
 722                  * user space tracing.
 723                  */
 724                 new_stream->hangup_flush_done = 0;
 725
 726                 health_code_update();
 727
 728                 pthread_mutex_lock(&new_stream->lock);
 729                 if (ctx->on_recv_stream) {
 730                         ret = ctx->on_recv_stream(new_stream);
 731                         if (ret < 0) {
 732                                 pthread_mutex_unlock(&new_stream->lock);
 733                                 pthread_mutex_unlock(&channel->lock);
 734                                 consumer_stream_free(new_stream);
 735                                 goto error_add_stream_nosignal;
 736                         }
 737                 }
 738                 health_code_update();
 739
 740                 if (new_stream->metadata_flag) {
 741                         channel->metadata_stream = new_stream;
 742                 }
 743
 744                 /* Do not monitor this stream. */
 745                 if (!channel->monitor) {
 746                         DBG("Kernel consumer add stream %s in no monitor mode with "
 747                                         "relayd id %" PRIu64, new_stream->name,
 748                                         new_stream->net_seq_idx);
 749                         cds_list_add(&new_stream->send_node, &channel->streams.head);
 750                         pthread_mutex_unlock(&new_stream->lock);
 751                         pthread_mutex_unlock(&channel->lock);
 752                         goto end_add_stream;
 753                 }
 754
 755                 /* Send stream to relayd if the stream has an ID. */
 756                 if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
 757                         ret = consumer_send_relayd_stream(new_stream,
 758                                         new_stream->chan->pathname);
 759                         if (ret < 0) {
 760                                 pthread_mutex_unlock(&new_stream->lock);
 761                                 pthread_mutex_unlock(&channel->lock);
 762                                 consumer_stream_free(new_stream);
 763                                 goto error_add_stream_nosignal;
 764                         }
 765
 766                         /*
 767                          * If adding an extra stream to an already
 768                          * existing channel (e.g. cpu hotplug), we need
 769                          * to send the "streams_sent" command to relayd.
 770                          */
 771                         if (channel->streams_sent_to_relayd) {
 772                                 ret = consumer_send_relayd_streams_sent(
 773                                                 new_stream->net_seq_idx);
 774                                 if (ret < 0) {
 775                                         pthread_mutex_unlock(&new_stream->lock);
 776                                         pthread_mutex_unlock(&channel->lock);
 777                                         goto error_add_stream_nosignal;
 778                                 }
 779                         }
 780                 }
 781                 pthread_mutex_unlock(&new_stream->lock);
 782                 pthread_mutex_unlock(&channel->lock);
 783
 784                 /* Get the right pipe where the stream will be sent. */
 785                 if (new_stream->metadata_flag) {
 786                         consumer_add_metadata_stream(new_stream);
 787                         stream_pipe = ctx->consumer_metadata_pipe;
 788                 } else {
 789                         consumer_add_data_stream(new_stream);
 790                         stream_pipe = ctx->consumer_data_pipe;
 791                 }
 792
 793                 /* Visible to other threads */
 794                 new_stream->globally_visible = 1;
 795
 796                 health_code_update();
 797
 798                 ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
 799                 if (ret < 0) {
 800                         ERR("Consumer write %s stream to pipe %d",
 801                                         new_stream->metadata_flag ? "metadata" : "data",
 802                                         lttng_pipe_get_writefd(stream_pipe));
 803                         if (new_stream->metadata_flag) {
 804                                 consumer_del_stream_for_metadata(new_stream);
 805                         } else {
 806                                 consumer_del_stream_for_data(new_stream);
 807                         }
 808                         goto error_add_stream_nosignal;
 809                 }
 810
 811                 DBG("Kernel consumer ADD_STREAM %s (fd: %d) %s with relayd id %" PRIu64,
 812                                 new_stream->name, fd, new_stream->chan->pathname, new_stream->relayd_stream_id);
 813 end_add_stream:
 814                 break;
 815 error_add_stream_nosignal:
 816                 goto end_nosignal;
 817 error_add_stream_fatal:
 818                 goto error_fatal;
 819         }
 820         case LTTNG_CONSUMER_STREAMS_SENT:
 821         {
 822                 struct lttng_consumer_channel *channel;
 823
 824                 /*
 825                  * Get stream's channel reference. Needed when adding the stream to the
 826                  * global hash table.
 827                  */
 828                 channel = consumer_find_channel(msg.u.sent_streams.channel_key);
 829                 if (!channel) {
 830                         /*
 831                          * We could not find the channel. Can happen if cpu hotplug
 832                          * happens while tearing down.
 833                          */
 834                         ERR("Unable to find channel key %" PRIu64,
 835                                         msg.u.sent_streams.channel_key);
 836                         ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
 837                 }
 838
 839                 health_code_update();
 840
 841                 /*
 842                  * Send status code to session daemon.
 843                  */
 844                 ret = consumer_send_status_msg(sock, ret_code);
 845                 if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
 846                         /* Somehow, the session daemon is not responding anymore. */
 847                         goto error_streams_sent_nosignal;
 848                 }
 849
 850                 health_code_update();
 851
 852                 /*
 853                  * We should not send this message if we don't monitor the
 854                  * streams in this channel.
 855                  */
 856                 if (!channel->monitor) {
 857                         goto end_error_streams_sent;
 858                 }
 859
 860                 health_code_update();
 861                 /* Send stream to relayd if the stream has an ID. */
 862                 if (msg.u.sent_streams.net_seq_idx != (uint64_t) -1ULL) {
 863                         ret = consumer_send_relayd_streams_sent(
 864                                         msg.u.sent_streams.net_seq_idx);
 865                         if (ret < 0) {
 866                                 goto error_streams_sent_nosignal;
 867                         }
 868                         channel->streams_sent_to_relayd = true;
 869                 }
 870 end_error_streams_sent:
 871                 break;
 872 error_streams_sent_nosignal:
 873                 goto end_nosignal;
 874         }
 875         case LTTNG_CONSUMER_UPDATE_STREAM:
 876         {
 877                 rcu_read_unlock();
 878                 return -ENOSYS;
 879         }
 880         case LTTNG_CONSUMER_DESTROY_RELAYD:
 881         {
 882                 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
 883                 struct consumer_relayd_sock_pair *relayd;
 884
 885                 DBG("Kernel consumer destroying relayd %" PRIu64, index);
 886
 887                 /* Get relayd reference if exists. */
 888                 relayd = consumer_find_relayd(index);
 889                 if (relayd == NULL) {
 890                         DBG("Unable to find relayd %" PRIu64, index);
 891                         ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
 892                 }
 893
 894                 /*
 895                  * Each relayd socket pair has a refcount of stream attached to it
 896                  * which tells if the relayd is still active or not depending on the
 897                  * refcount value.
 898                  *
 899                  * This will set the destroy flag of the relayd object and destroy it
 900                  * if the refcount reaches zero when called.
 901                  *
 902                  * The destroy can happen either here or when a stream fd hangs up.
 903                  */
 904                 if (relayd) {
 905                         consumer_flag_relayd_for_destroy(relayd);
 906                 }
 907
 908                 health_code_update();
 909
 910                 ret = consumer_send_status_msg(sock, ret_code);
 911                 if (ret < 0) {
 912                         /* Somehow, the session daemon is not responding anymore. */
 913                         goto error_fatal;
 914                 }
 915
 916                 goto end_nosignal;
 917         }
 918         case LTTNG_CONSUMER_DATA_PENDING:
 919         {
 920                 int32_t ret;
 921                 uint64_t id = msg.u.data_pending.session_id;
 922
 923                 DBG("Kernel consumer data pending command for id %" PRIu64, id);
 924
 925                 ret = consumer_data_pending(id);
 926
 927                 health_code_update();
 928
 929                 /* Send back returned value to session daemon */
 930                 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
 931                 if (ret < 0) {
 932                         PERROR("send data pending ret code");
 933                         goto error_fatal;
 934                 }
 935
 936                 /*
 937                  * No need to send back a status message since the data pending
 938                  * returned value is the response.
 939                  */
 940                 break;
 941         }
 942         case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
 943         {
 944                 struct lttng_consumer_channel *channel;
 945                 uint64_t key = msg.u.snapshot_channel.key;
 946
 947                 channel = consumer_find_channel(key);
 948                 if (!channel) {
 949                         ERR("Channel %" PRIu64 " not found", key);
 950                         ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
 951                 } else {
 952                         pthread_mutex_lock(&channel->lock);
 953                         if (msg.u.snapshot_channel.metadata == 1) {
 954                                 ret = lttng_kconsumer_snapshot_metadata(channel, key,
 955                                                 msg.u.snapshot_channel.pathname,
 956                                                 msg.u.snapshot_channel.relayd_id, ctx);
 957                                 if (ret < 0) {
 958                                         ERR("Snapshot metadata failed");
 959                                         ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
 960                                 }
 961                         } else {
 962                                 ret = lttng_kconsumer_snapshot_channel(channel, key,
 963                                                 msg.u.snapshot_channel.pathname,
 964                                                 msg.u.snapshot_channel.relayd_id,
 965                                                 msg.u.snapshot_channel.nb_packets_per_stream,
 966                                                 ctx);
 967                                 if (ret < 0) {
 968                                         ERR("Snapshot channel failed");
 969                                         ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
 970                                 }
 971                         }
 972                         pthread_mutex_unlock(&channel->lock);
 973                 }
 974                 health_code_update();
 975
 976                 ret = consumer_send_status_msg(sock, ret_code);
 977                 if (ret < 0) {
 978                         /* Somehow, the session daemon is not responding anymore. */
 979                         goto end_nosignal;
 980                 }
 981                 break;
 982         }
 983         case LTTNG_CONSUMER_DESTROY_CHANNEL:
 984         {
 985                 uint64_t key = msg.u.destroy_channel.key;
 986                 struct lttng_consumer_channel *channel;
 987
 988                 channel = consumer_find_channel(key);
 989                 if (!channel) {
 990                         ERR("Kernel consumer destroy channel %" PRIu64 " not found", key);
 991                         ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
 992                 }
 993
 994                 health_code_update();
 995
 996                 ret = consumer_send_status_msg(sock, ret_code);
 997                 if (ret < 0) {
 998                         /* Somehow, the session daemon is not responding anymore. */
 999                         goto end_destroy_channel;
1000                 }
1001
1002                 health_code_update();
1003
1004                 /* Stop right now if no channel was found. */
1005                 if (!channel) {
1006                         goto end_destroy_channel;
1007                 }
1008
1009                 /*
1010                  * This command should ONLY be issued for channel with streams set in
1011                  * no monitor mode.
1012                  */
1013                 assert(!channel->monitor);
1014
1015                 /*
1016                  * The refcount should ALWAYS be 0 in the case of a channel in no
1017                  * monitor mode.
1018                  */
1019                 assert(!uatomic_sub_return(&channel->refcount, 1));
1020
1021                 consumer_del_channel(channel);
1022 end_destroy_channel:
1023                 goto end_nosignal;
1024         }
1025         case LTTNG_CONSUMER_DISCARDED_EVENTS:
1026         {
1027                 ssize_t ret;
1028                 uint64_t count;
1029                 struct lttng_consumer_channel *channel;
1030                 uint64_t id = msg.u.discarded_events.session_id;
1031                 uint64_t key = msg.u.discarded_events.channel_key;
1032
1033                 DBG("Kernel consumer discarded events command for session id %"
1034                                 PRIu64 ", channel key %" PRIu64, id, key);
1035
1036                 channel = consumer_find_channel(key);
1037                 if (!channel) {
1038                         ERR("Kernel consumer discarded events channel %"
1039                                         PRIu64 " not found", key);
1040                         count = 0;
1041                 } else {
1042                         count = channel->discarded_events;
1043                 }
1044
1045                 health_code_update();
1046
1047                 /* Send back returned value to session daemon */
1048                 ret = lttcomm_send_unix_sock(sock, &count, sizeof(count));
1049                 if (ret < 0) {
1050                         PERROR("send discarded events");
1051                         goto error_fatal;
1052                 }
1053
1054                 break;
1055         }
1056         case LTTNG_CONSUMER_LOST_PACKETS:
1057         {
1058                 ssize_t ret;
1059                 uint64_t count;
1060                 struct lttng_consumer_channel *channel;
1061                 uint64_t id = msg.u.lost_packets.session_id;
1062                 uint64_t key = msg.u.lost_packets.channel_key;
1063
1064                 DBG("Kernel consumer lost packets command for session id %"
1065                                 PRIu64 ", channel key %" PRIu64, id, key);
1066
1067                 channel = consumer_find_channel(key);
1068                 if (!channel) {
1069                         ERR("Kernel consumer lost packets channel %"
1070                                         PRIu64 " not found", key);
1071                         count = 0;
1072                 } else {
1073                         count = channel->lost_packets;
1074                 }
1075
1076                 health_code_update();
1077
1078                 /* Send back returned value to session daemon */
1079                 ret = lttcomm_send_unix_sock(sock, &count, sizeof(count));
1080                 if (ret < 0) {
1081                         PERROR("send lost packets");
1082                         goto error_fatal;
1083                 }
1084
1085                 break;
1086         }
1087         case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1088         {
1089                 int channel_monitor_pipe;
1090
1091                 ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1092                 /* Successfully received the command's type. */
1093                 ret = consumer_send_status_msg(sock, ret_code);
1094                 if (ret < 0) {
1095                         goto error_fatal;
1096                 }
1097
1098                 ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
1099                                 1);
1100                 if (ret != sizeof(channel_monitor_pipe)) {
1101                         ERR("Failed to receive channel monitor pipe");
1102                         goto error_fatal;
1103                 }
1104
1105                 DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
1106                 ret = consumer_timer_thread_set_channel_monitor_pipe(
1107                                 channel_monitor_pipe);
1108                 if (!ret) {
1109                         int flags;
1110
1111                         ret_code = LTTCOMM_CONSUMERD_SUCCESS;
1112                         /* Set the pipe as non-blocking. */
1113                         ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
1114                         if (ret == -1) {
1115                                 PERROR("fcntl get flags of the channel monitoring pipe");
1116                                 goto error_fatal;
1117                         }
1118                         flags = ret;
1119
1120                         ret = fcntl(channel_monitor_pipe, F_SETFL,
1121                                         flags | O_NONBLOCK);
1122                         if (ret == -1) {
1123                                 PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
1124                                 goto error_fatal;
1125                         }
1126                         DBG("Channel monitor pipe set as non-blocking");
1127                 } else {
1128                         ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
1129                 }
1130                 ret = consumer_send_status_msg(sock, ret_code);
1131                 if (ret < 0) {
1132                         goto error_fatal;
1133                 }
1134                 break;
1135         }
1136         case LTTNG_CONSUMER_ROTATE_CHANNEL:
1137         {
1138                 struct lttng_consumer_channel *channel;
1139                 uint64_t key = msg.u.rotate_channel.key;
1140
1141                 DBG("Consumer rotate channel %" PRIu64, key);
1142
1143                 channel = consumer_find_channel(key);
1144                 if (!channel) {
1145                         ERR("Channel %" PRIu64 " not found", key);
1146                         ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
1147                 } else {
1148                         /*
1149                          * Sample the rotate position of all the streams in this channel.
1150                          */
1151                         ret = lttng_consumer_rotate_channel(channel, key,
1152                                         msg.u.rotate_channel.relayd_id,
1153                                         msg.u.rotate_channel.metadata,
1154                                         ctx);
1155                         if (ret < 0) {
1156                                 ERR("Rotate channel failed");
1157                                 ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL;
1158                         }
1159
1160                         health_code_update();
1161                 }
1162                 ret = consumer_send_status_msg(sock, ret_code);
1163                 if (ret < 0) {
1164                         /* Somehow, the session daemon is not responding anymore. */
1165                         goto error_rotate_channel;
1166                 }
1167                 if (channel) {
1168                         /* Rotate the streams that are ready right now. */
1169                         ret = lttng_consumer_rotate_ready_streams(
1170                                         channel, key, ctx);
1171                         if (ret < 0) {
1172                                 ERR("Rotate ready streams failed");
1173                         }
1174                 }
1175                 break;
1176 error_rotate_channel:
1177                 goto end_nosignal;
1178         }
1179         case LTTNG_CONSUMER_INIT:
1180         {
1181                 ret_code = lttng_consumer_init_command(ctx,
1182                                 msg.u.init.sessiond_uuid);
1183                 health_code_update();
1184                 ret = consumer_send_status_msg(sock, ret_code);
1185                 if (ret < 0) {
1186                         /* Somehow, the session daemon is not responding anymore. */
1187                         goto end_nosignal;
1188                 }
1189                 break;
1190         }
1191         case LTTNG_CONSUMER_CREATE_TRACE_CHUNK:
1192         {
1193                 const struct lttng_credentials credentials = {
1194                         .uid = msg.u.create_trace_chunk.credentials.value.uid,
1195                         .gid = msg.u.create_trace_chunk.credentials.value.gid,
1196                 };
1197                 const bool is_local_trace =
1198                                 !msg.u.create_trace_chunk.relayd_id.is_set;
1199                 const uint64_t relayd_id =
1200                                 msg.u.create_trace_chunk.relayd_id.value;
1201                 const char *chunk_override_name =
1202                                 *msg.u.create_trace_chunk.override_name ?
1203                                         msg.u.create_trace_chunk.override_name :
1204                                         NULL;
1205                 LTTNG_OPTIONAL(struct lttng_directory_handle) chunk_directory_handle =
1206                                 LTTNG_OPTIONAL_INIT;
1207
1208                 /*
1209                  * The session daemon will only provide a chunk directory file
1210                  * descriptor for local traces.
1211                  */
1212                 if (is_local_trace) {
1213                         int chunk_dirfd;
1214
1215                         /* Acnowledge the reception of the command. */
1216                         ret = consumer_send_status_msg(sock,
1217                                         LTTCOMM_CONSUMERD_SUCCESS);
1218                         if (ret < 0) {
1219                                 /* Somehow, the session daemon is not responding anymore. */
1220                                 goto end_nosignal;
1221                         }
1222
1223                         ret = lttcomm_recv_fds_unix_sock(sock, &chunk_dirfd, 1);
1224                         if (ret != sizeof(chunk_dirfd)) {
1225                                 ERR("Failed to receive trace chunk directory file descriptor");
1226                                 goto error_fatal;
1227                         }
1228
1229                         DBG("Received trace chunk directory fd (%d)",
1230                                         chunk_dirfd);
1231                         ret = lttng_directory_handle_init_from_dirfd(
1232                                         &chunk_directory_handle.value,
1233                                         chunk_dirfd);
1234                         if (ret) {
1235                                 ERR("Failed to initialize chunk directory handle from directory file descriptor");
1236                                 if (close(chunk_dirfd)) {
1237                                         PERROR("Failed to close chunk directory file descriptor");
1238                                 }
1239                                 goto error_fatal;
1240                         }
1241                         chunk_directory_handle.is_set = true;
1242                 }
1243
1244                 ret_code = lttng_consumer_create_trace_chunk(
1245                                 !is_local_trace ? &relayd_id : NULL,
1246                                 msg.u.create_trace_chunk.session_id,
1247                                 msg.u.create_trace_chunk.chunk_id,
1248                                 (time_t) msg.u.create_trace_chunk
1249                                                 .creation_timestamp,
1250                                 chunk_override_name,
1251                                 msg.u.create_trace_chunk.credentials.is_set ?
1252                                                 &credentials :
1253                                                 NULL,
1254                                 chunk_directory_handle.is_set ?
1255                                                 &chunk_directory_handle.value :
1256                                                 NULL);
1257
1258                 if (chunk_directory_handle.is_set) {
1259                         lttng_directory_handle_fini(
1260                                         &chunk_directory_handle.value);
1261                 }
1262                 goto end_msg_sessiond;
1263         }
1264         case LTTNG_CONSUMER_CLOSE_TRACE_CHUNK:
1265         {
1266                 enum lttng_trace_chunk_command_type close_command =
1267                                 msg.u.close_trace_chunk.close_command.value;
1268                 const uint64_t relayd_id =
1269                                 msg.u.close_trace_chunk.relayd_id.value;
1270                 struct lttcomm_consumer_close_trace_chunk_reply reply;
1271                 char path[LTTNG_PATH_MAX];
1272
1273                 ret_code = lttng_consumer_close_trace_chunk(
1274                                 msg.u.close_trace_chunk.relayd_id.is_set ?
1275                                                 &relayd_id :
1276                                                 NULL,
1277                                 msg.u.close_trace_chunk.session_id,
1278                                 msg.u.close_trace_chunk.chunk_id,
1279                                 (time_t) msg.u.close_trace_chunk.close_timestamp,
1280                                 msg.u.close_trace_chunk.close_command.is_set ?
1281                                                 &close_command :
1282                                                 NULL, path);
1283                 reply.ret_code = ret_code;
1284                 reply.path_length = strlen(path) + 1;
1285                 ret = lttcomm_send_unix_sock(sock, &reply, sizeof(reply));
1286                 if (ret != sizeof(reply)) {
1287                         goto error_fatal;
1288                 }
1289                 ret = lttcomm_send_unix_sock(sock, path, reply.path_length);
1290                 if (ret != reply.path_length) {
1291                         goto error_fatal;
1292                 }
1293                 goto end_nosignal;
1294         }
1295         case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS:
1296         {
1297                 const uint64_t relayd_id =
1298                                 msg.u.trace_chunk_exists.relayd_id.value;
1299
1300                 ret_code = lttng_consumer_trace_chunk_exists(
1301                                 msg.u.trace_chunk_exists.relayd_id.is_set ?
1302                                                 &relayd_id : NULL,
1303                                 msg.u.trace_chunk_exists.session_id,
1304                                 msg.u.trace_chunk_exists.chunk_id);
1305                 goto end_msg_sessiond;
1306         }
1307         default:
1308                 goto end_nosignal;
1309         }
1310
1311 end_nosignal:
1312         /*
1313          * Return 1 to indicate success since the 0 value can be a socket
1314          * shutdown during the recv() or send() call.
1315          */
1316         ret = 1;
1317         goto end;
1318 error_fatal:
1319         /* This will issue a consumer stop. */
1320         ret = -1;
1321         goto end;
1322 end_msg_sessiond:
1323         /*
1324          * The returned value here is not useful since either way we'll return 1 to
1325          * the caller because the session daemon socket management is done
1326          * elsewhere. Returning a negative code or 0 will shutdown the consumer.
1327          */
1328         ret = consumer_send_status_msg(sock, ret_code);
1329         if (ret < 0) {
1330                 goto error_fatal;
1331         }
1332         ret = 1;
1333 end:
1334         health_code_update();
1335         rcu_read_unlock();
1336         return ret;
1337 }
1338
1339 /*
1340  * Sync metadata meaning request them to the session daemon and snapshot to the
1341  * metadata thread can consumer them.
1342  *
1343  * Metadata stream lock MUST be acquired.
1344  */
1345 enum sync_metadata_status lttng_kconsumer_sync_metadata(
1346                 struct lttng_consumer_stream *metadata)
1347 {
1348         int ret;
1349         enum sync_metadata_status status;
1350
1351         assert(metadata);
1352
1353         ret = kernctl_buffer_flush(metadata->wait_fd);
1354         if (ret < 0) {
1355                 ERR("Failed to flush kernel stream");
1356                 status = SYNC_METADATA_STATUS_ERROR;
1357                 goto end;
1358         }
1359
1360         ret = kernctl_snapshot(metadata->wait_fd);
1361         if (ret < 0) {
1362                 if (errno == EAGAIN) {
1363                         /* No new metadata, exit. */
1364                         DBG("Sync metadata, no new kernel metadata");
1365                         status = SYNC_METADATA_STATUS_NO_DATA;
1366                 } else {
1367                         ERR("Sync metadata, taking kernel snapshot failed.");
1368                         status = SYNC_METADATA_STATUS_ERROR;
1369                 }
1370         } else {
1371                 status = SYNC_METADATA_STATUS_NEW_DATA;
1372         }
1373
1374 end:
1375         return status;
1376 }
1377
1378 static
1379 int extract_common_subbuffer_info(struct lttng_consumer_stream *stream,
1380                 struct stream_subbuffer *subbuf)
1381 {
1382         int ret;
1383
1384         ret = kernctl_get_subbuf_size(
1385                         stream->wait_fd, &subbuf->info.data.subbuf_size);
1386         if (ret) {
1387                 goto end;
1388         }
1389
1390         ret = kernctl_get_padded_subbuf_size(
1391                         stream->wait_fd, &subbuf->info.data.padded_subbuf_size);
1392         if (ret) {
1393                 goto end;
1394         }
1395
1396 end:
1397         return ret;
1398 }
1399
1400 static
1401 int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream,
1402                 struct stream_subbuffer *subbuf)
1403 {
1404         int ret;
1405
1406         ret = extract_common_subbuffer_info(stream, subbuf);
1407         if (ret) {
1408                 goto end;
1409         }
1410
1411         ret = kernctl_get_metadata_version(
1412                         stream->wait_fd, &subbuf->info.metadata.version);
1413         if (ret) {
1414                 goto end;
1415         }
1416
1417 end:
1418         return ret;
1419 }
1420
1421 static
1422 int extract_data_subbuffer_info(struct lttng_consumer_stream *stream,
1423                 struct stream_subbuffer *subbuf)
1424 {
1425         int ret;
1426
1427         ret = extract_common_subbuffer_info(stream, subbuf);
1428         if (ret) {
1429                 goto end;
1430         }
1431
1432         ret = kernctl_get_packet_size(
1433                         stream->wait_fd, &subbuf->info.data.packet_size);
1434         if (ret < 0) {
1435                 PERROR("Failed to get sub-buffer packet size");
1436                 goto end;
1437         }
1438
1439         ret = kernctl_get_content_size(
1440                         stream->wait_fd, &subbuf->info.data.content_size);
1441         if (ret < 0) {
1442                 PERROR("Failed to get sub-buffer content size");
1443                 goto end;
1444         }
1445
1446         ret = kernctl_get_timestamp_begin(
1447                         stream->wait_fd, &subbuf->info.data.timestamp_begin);
1448         if (ret < 0) {
1449                 PERROR("Failed to get sub-buffer begin timestamp");
1450                 goto end;
1451         }
1452
1453         ret = kernctl_get_timestamp_end(
1454                         stream->wait_fd, &subbuf->info.data.timestamp_end);
1455         if (ret < 0) {
1456                 PERROR("Failed to get sub-buffer end timestamp");
1457                 goto end;
1458         }
1459
1460         ret = kernctl_get_events_discarded(
1461                         stream->wait_fd, &subbuf->info.data.events_discarded);
1462         if (ret) {
1463                 PERROR("Failed to get sub-buffer events discarded count");
1464                 goto end;
1465         }
1466
1467         ret = kernctl_get_sequence_number(stream->wait_fd,
1468                         &subbuf->info.data.sequence_number.value);
1469         if (ret) {
1470                 /* May not be supported by older LTTng-modules. */
1471                 if (ret != -ENOTTY) {
1472                         PERROR("Failed to get sub-buffer sequence number");
1473                         goto end;
1474                 }
1475         } else {
1476                 subbuf->info.data.sequence_number.is_set = true;
1477         }
1478
1479         ret = kernctl_get_stream_id(
1480                         stream->wait_fd, &subbuf->info.data.stream_id);
1481         if (ret < 0) {
1482                 PERROR("Failed to get stream id");
1483                 goto end;
1484         }
1485
1486         ret = kernctl_get_instance_id(stream->wait_fd,
1487                         &subbuf->info.data.stream_instance_id.value);
1488         if (ret) {
1489                 /* May not be supported by older LTTng-modules. */
1490                 if (ret != -ENOTTY) {
1491                         PERROR("Failed to get stream instance id");
1492                         goto end;
1493                 }
1494         } else {
1495                 subbuf->info.data.stream_instance_id.is_set = true;
1496         }
1497 end:
1498         return ret;
1499 }
1500
1501 static
1502 int get_subbuffer_common(struct lttng_consumer_stream *stream,
1503                 struct stream_subbuffer *subbuffer)
1504 {
1505         int ret;
1506
1507         ret = kernctl_get_next_subbuf(stream->wait_fd);
1508         if (ret) {
1509                 goto end;
1510         }
1511
1512         ret = stream->read_subbuffer_ops.extract_subbuffer_info(
1513                         stream, subbuffer);
1514 end:
1515         return ret;
1516 }
1517
1518 static
1519 int get_next_subbuffer_splice(struct lttng_consumer_stream *stream,
1520                 struct stream_subbuffer *subbuffer)
1521 {
1522         int ret;
1523
1524         ret = get_subbuffer_common(stream, subbuffer);
1525         if (ret) {
1526                 goto end;
1527         }
1528
1529         subbuffer->buffer.fd = stream->wait_fd;
1530 end:
1531         return ret;
1532 }
1533
1534 static
1535 int get_next_subbuffer_mmap(struct lttng_consumer_stream *stream,
1536                 struct stream_subbuffer *subbuffer)
1537 {
1538         int ret;
1539         const char *addr;
1540
1541         ret = get_subbuffer_common(stream, subbuffer);
1542         if (ret) {
1543                 goto end;
1544         }
1545
1546         ret = get_current_subbuf_addr(stream, &addr);
1547         if (ret) {
1548                 goto end;
1549         }
1550
1551         subbuffer->buffer.buffer = lttng_buffer_view_init(
1552                         addr, 0, subbuffer->info.data.padded_subbuf_size);
1553 end:
1554         return ret;
1555 }
1556
1557 static
1558 int get_next_subbuffer_metadata_check(struct lttng_consumer_stream *stream,
1559                 struct stream_subbuffer *subbuffer)
1560 {
1561         int ret;
1562         const char *addr;
1563         bool coherent;
1564
1565         ret = kernctl_get_next_subbuf_metadata_check(stream->wait_fd,
1566                         &coherent);
1567         if (ret) {
1568                 goto end;
1569         }
1570
1571         ret = stream->read_subbuffer_ops.extract_subbuffer_info(
1572                         stream, subbuffer);
1573         if (ret) {
1574                 goto end;
1575         }
1576
1577         LTTNG_OPTIONAL_SET(&subbuffer->info.metadata.coherent, coherent);
1578
1579         ret = get_current_subbuf_addr(stream, &addr);
1580         if (ret) {
1581                 goto end;
1582         }
1583
1584         subbuffer->buffer.buffer = lttng_buffer_view_init(
1585                         addr, 0, subbuffer->info.data.padded_subbuf_size);
1586         DBG("Got metadata packet with padded_subbuf_size = %lu, coherent = %s",
1587                         subbuffer->info.metadata.padded_subbuf_size,
1588                         coherent ? "true" : "false");
1589 end:
1590         return ret;
1591 }
1592
1593 static
1594 int put_next_subbuffer(struct lttng_consumer_stream *stream,
1595                 struct stream_subbuffer *subbuffer)
1596 {
1597         const int ret = kernctl_put_next_subbuf(stream->wait_fd);
1598
1599         if (ret) {
1600                 if (ret == -EFAULT) {
1601                         PERROR("Error in unreserving sub buffer");
1602                 } else if (ret == -EIO) {
1603                         /* Should never happen with newer LTTng versions */
1604                         PERROR("Reader has been pushed by the writer, last sub-buffer corrupted");
1605                 }
1606         }
1607
1608         return ret;
1609 }
1610
1611 static
1612 bool is_get_next_check_metadata_available(int tracer_fd)
1613 {
1614         const int ret = kernctl_get_next_subbuf_metadata_check(tracer_fd, NULL);
1615         const bool available = ret != -ENOTTY;
1616
1617         if (ret == 0) {
1618                 /* get succeeded, make sure to put the subbuffer. */
1619                 kernctl_put_subbuf(tracer_fd);
1620         }
1621
1622         return available;
1623 }
1624
1625 static
1626 int lttng_kconsumer_set_stream_ops(
1627                 struct lttng_consumer_stream *stream)
1628 {
1629         int ret = 0;
1630
1631         if (stream->metadata_flag && stream->chan->is_live) {
1632                 DBG("Attempting to enable metadata bucketization for live consumers");
1633                 if (is_get_next_check_metadata_available(stream->wait_fd)) {
1634                         DBG("Kernel tracer supports get_next_subbuffer_metadata_check, metadata will be accumulated until a coherent state is reached");
1635                         stream->read_subbuffer_ops.get_next_subbuffer =
1636                                         get_next_subbuffer_metadata_check;
1637                         ret = consumer_stream_enable_metadata_bucketization(
1638                                         stream);
1639                         if (ret) {
1640                                 goto end;
1641                         }
1642                 } else {
1643                         /*
1644                          * The kernel tracer version is too old to indicate
1645                          * when the metadata stream has reached a "coherent"
1646                          * (parseable) point.
1647                          *
1648                          * This means that a live viewer may see an incoherent
1649                          * sequence of metadata and fail to parse it.
1650                          */
1651                         WARN("Kernel tracer does not support get_next_subbuffer_metadata_check which may cause live clients to fail to parse the metadata stream");
1652                         metadata_bucket_destroy(stream->metadata_bucket);
1653                         stream->metadata_bucket = NULL;
1654                 }
1655         }
1656
1657         if (!stream->read_subbuffer_ops.get_next_subbuffer) {
1658                 if (stream->chan->output == CONSUMER_CHANNEL_MMAP) {
1659                         stream->read_subbuffer_ops.get_next_subbuffer =
1660                                         get_next_subbuffer_mmap;
1661                 } else {
1662                         stream->read_subbuffer_ops.get_next_subbuffer =
1663                                         get_next_subbuffer_splice;
1664                 }
1665         }
1666
1667         if (stream->metadata_flag) {
1668                 stream->read_subbuffer_ops.extract_subbuffer_info =
1669                                 extract_metadata_subbuffer_info;
1670         } else {
1671                 stream->read_subbuffer_ops.extract_subbuffer_info =
1672                                 extract_data_subbuffer_info;
1673                 if (stream->chan->is_live) {
1674                         stream->read_subbuffer_ops.send_live_beacon =
1675                                         consumer_flush_kernel_index;
1676                 }
1677         }
1678
1679         stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer;
1680 end:
1681         return ret;
1682 }
1683
1684 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1685 {
1686         int ret;
1687
1688         assert(stream);
1689
1690         /*
1691          * Don't create anything if this is set for streaming or if there is
1692          * no current trace chunk on the parent channel.
1693          */
1694         if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor &&
1695                         stream->chan->trace_chunk) {
1696                 ret = consumer_stream_create_output_files(stream, true);
1697                 if (ret) {
1698                         goto error;
1699                 }
1700         }
1701
1702         if (stream->output == LTTNG_EVENT_MMAP) {
1703                 /* get the len of the mmap region */
1704                 unsigned long mmap_len;
1705
1706                 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
1707                 if (ret != 0) {
1708                         PERROR("kernctl_get_mmap_len");
1709                         goto error_close_fd;
1710                 }
1711                 stream->mmap_len = (size_t) mmap_len;
1712
1713                 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
1714                                 MAP_PRIVATE, stream->wait_fd, 0);
1715                 if (stream->mmap_base == MAP_FAILED) {
1716                         PERROR("Error mmaping");
1717                         ret = -1;
1718                         goto error_close_fd;
1719                 }
1720         }
1721
1722         ret = lttng_kconsumer_set_stream_ops(stream);
1723         if (ret) {
1724                 goto error_close_fd;
1725         }
1726
1727         /* we return 0 to let the library handle the FD internally */
1728         return 0;
1729
1730 error_close_fd:
1731         if (stream->out_fd >= 0) {
1732                 int err;
1733
1734                 err = close(stream->out_fd);
1735                 assert(!err);
1736                 stream->out_fd = -1;
1737         }
1738 error:
1739         return ret;
1740 }
1741
1742 /*
1743  * Check if data is still being extracted from the buffers for a specific
1744  * stream. Consumer data lock MUST be acquired before calling this function
1745  * and the stream lock.
1746  *
1747  * Return 1 if the traced data are still getting read else 0 meaning that the
1748  * data is available for trace viewer reading.
1749  */
1750 int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
1751 {
1752         int ret;
1753
1754         assert(stream);
1755
1756         if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1757                 ret = 0;
1758                 goto end;
1759         }
1760
1761         ret = kernctl_get_next_subbuf(stream->wait_fd);
1762         if (ret == 0) {
1763                 /* There is still data so let's put back this subbuffer. */
1764                 ret = kernctl_put_subbuf(stream->wait_fd);
1765                 assert(ret == 0);
1766                 ret = 1;   /* Data is pending */
1767                 goto end;
1768         }
1769
1770         /* Data is NOT pending and ready to be read. */
1771         ret = 0;
1772
1773 end:
1774         return ret;
1775 }