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