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