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                                 NULL, NULL);
 524                 if (new_channel == NULL) {
 525                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 526                         goto end_nosignal;
 527                 }
 528                 new_channel->nb_init_stream_left = msg.u.channel.nb_init_streams;
 529                 switch (msg.u.channel.output) {
 530                 case LTTNG_EVENT_SPLICE:
 531                         new_channel->output = CONSUMER_CHANNEL_SPLICE;
 532                         break;
 533                 case LTTNG_EVENT_MMAP:
 534                         new_channel->output = CONSUMER_CHANNEL_MMAP;
 535                         break;
 536                 default:
 537                         ERR("Channel output unknown %d", msg.u.channel.output);
 538                         goto end_nosignal;
 539                 }
 540
 541                 /* Translate and save channel type. */
 542                 switch (msg.u.channel.type) {
 543                 case CONSUMER_CHANNEL_TYPE_DATA:
 544                 case CONSUMER_CHANNEL_TYPE_METADATA:
 545                         new_channel->type = msg.u.channel.type;
 546                         break;
 547                 default:
 548                         assert(0);
 549                         goto end_nosignal;
 550                 };
 551
 552                 health_code_update();
 553
 554                 if (ctx->on_recv_channel != NULL) {
 555                         ret_recv = ctx->on_recv_channel(new_channel);
 556                         if (ret_recv == 0) {
 557                                 ret = consumer_add_channel(new_channel, ctx);
 558                         } else if (ret_recv < 0) {
 559                                 goto end_nosignal;
 560                         }
 561                 } else {
 562                         ret = consumer_add_channel(new_channel, ctx);
 563                 }
 564                 if (CONSUMER_CHANNEL_TYPE_DATA) {
 565                         consumer_timer_live_start(new_channel,
 566                                         msg.u.channel.live_timer_interval);
 567                 }
 568
 569                 health_code_update();
 570
 571                 /* If we received an error in add_channel, we need to report it. */
 572                 if (ret < 0) {
 573                         ret = consumer_send_status_msg(sock, ret);
 574                         if (ret < 0) {
 575                                 goto error_fatal;
 576                         }
 577                         goto end_nosignal;
 578                 }
 579
 580                 goto end_nosignal;
 581         }
 582         case LTTNG_CONSUMER_ADD_STREAM:
 583         {
 584                 int fd;
 585                 struct lttng_pipe *stream_pipe;
 586                 struct lttng_consumer_stream *new_stream;
 587                 struct lttng_consumer_channel *channel;
 588                 int alloc_ret = 0;
 589
 590                 /*
 591                  * Get stream's channel reference. Needed when adding the stream to the
 592                  * global hash table.
 593                  */
 594                 channel = consumer_find_channel(msg.u.stream.channel_key);
 595                 if (!channel) {
 596                         /*
 597                          * We could not find the channel. Can happen if cpu hotplug
 598                          * happens while tearing down.
 599                          */
 600                         ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key);
 601                         ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
 602                 }
 603
 604                 health_code_update();
 605
 606                 /* First send a status message before receiving the fds. */
 607                 ret = consumer_send_status_msg(sock, ret_code);
 608                 if (ret < 0) {
 609                         /* Somehow, the session daemon is not responding anymore. */
 610                         goto error_fatal;
 611                 }
 612
 613                 health_code_update();
 614
 615                 if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
 616                         /* Channel was not found. */
 617                         goto end_nosignal;
 618                 }
 619
 620                 /* Blocking call */
 621                 health_poll_entry();
 622                 ret = lttng_consumer_poll_socket(consumer_sockpoll);
 623                 health_poll_exit();
 624                 if (ret) {
 625                         goto error_fatal;
 626                 }
 627
 628                 health_code_update();
 629
 630                 /* Get stream file descriptor from socket */
 631                 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 632                 if (ret != sizeof(fd)) {
 633                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 634                         rcu_read_unlock();
 635                         return ret;
 636                 }
 637
 638                 health_code_update();
 639
 640                 /*
 641                  * Send status code to session daemon only if the recv works. If the
 642                  * above recv() failed, the session daemon is notified through the
 643                  * error socket and the teardown is eventually done.
 644                  */
 645                 ret = consumer_send_status_msg(sock, ret_code);
 646                 if (ret < 0) {
 647                         /* Somehow, the session daemon is not responding anymore. */
 648                         goto end_nosignal;
 649                 }
 650
 651                 health_code_update();
 652
 653                 new_stream = consumer_allocate_stream(channel->key,
 654                                 fd,
 655                                 LTTNG_CONSUMER_ACTIVE_STREAM,
 656                                 channel->name,
 657                                 channel->uid,
 658                                 channel->gid,
 659                                 channel->relayd_id,
 660                                 channel->session_id,
 661                                 msg.u.stream.cpu,
 662                                 &alloc_ret,
 663                                 channel->type,
 664                                 channel->monitor);
 665                 if (new_stream == NULL) {
 666                         switch (alloc_ret) {
 667                         case -ENOMEM:
 668                         case -EINVAL:
 669                         default:
 670                                 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 671                                 break;
 672                         }
 673                         goto end_nosignal;
 674                 }
 675
 676                 new_stream->chan = channel;
 677                 new_stream->wait_fd = fd;
 678                 switch (channel->output) {
 679                 case CONSUMER_CHANNEL_SPLICE:
 680                         new_stream->output = LTTNG_EVENT_SPLICE;
 681                         ret = utils_create_pipe(new_stream->splice_pipe);
 682                         if (ret < 0) {
 683                                 goto end_nosignal;
 684                         }
 685                         break;
 686                 case CONSUMER_CHANNEL_MMAP:
 687                         new_stream->output = LTTNG_EVENT_MMAP;
 688                         break;
 689                 default:
 690                         ERR("Stream output unknown %d", channel->output);
 691                         goto end_nosignal;
 692                 }
 693
 694                 /*
 695                  * We've just assigned the channel to the stream so increment the
 696                  * refcount right now. We don't need to increment the refcount for
 697                  * streams in no monitor because we handle manually the cleanup of
 698                  * those. It is very important to make sure there is NO prior
 699                  * consumer_del_stream() calls or else the refcount will be unbalanced.
 700                  */
 701                 if (channel->monitor) {
 702                         uatomic_inc(&new_stream->chan->refcount);
 703                 }
 704
 705                 /*
 706                  * The buffer flush is done on the session daemon side for the kernel
 707                  * so no need for the stream "hangup_flush_done" variable to be
 708                  * tracked. This is important for a kernel stream since we don't rely
 709                  * on the flush state of the stream to read data. It's not the case for
 710                  * user space tracing.
 711                  */
 712                 new_stream->hangup_flush_done = 0;
 713
 714                 health_code_update();
 715
 716                 if (ctx->on_recv_stream) {
 717                         ret = ctx->on_recv_stream(new_stream);
 718                         if (ret < 0) {
 719                                 consumer_stream_free(new_stream);
 720                                 goto end_nosignal;
 721                         }
 722                 }
 723
 724                 health_code_update();
 725
 726                 if (new_stream->metadata_flag) {
 727                         channel->metadata_stream = new_stream;
 728                 }
 729
 730                 /* Do not monitor this stream. */
 731                 if (!channel->monitor) {
 732                         DBG("Kernel consumer add stream %s in no monitor mode with "
 733                                         "relayd id %" PRIu64, new_stream->name,
 734                                         new_stream->relayd_id);
 735                         cds_list_add(&new_stream->send_node, &channel->streams.head);
 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                                 consumer_stream_free(new_stream);
 745                                 goto end_nosignal;
 746                         }
 747
 748                         /*
 749                          * If adding an extra stream to an already
 750                          * existing channel (e.g. cpu hotplug), we need
 751                          * to send the "streams_sent" command to relayd.
 752                          */
 753                         if (channel->streams_sent_to_relayd) {
 754                                 ret = consumer_send_relayd_streams_sent(
 755                                                 new_stream->relayd_id);
 756                                 if (ret < 0) {
 757                                         goto end_nosignal;
 758                                 }
 759                         }
 760                 }
 761
 762                 /* Get the right pipe where the stream will be sent. */
 763                 if (new_stream->metadata_flag) {
 764                         ret = consumer_add_metadata_stream(new_stream);
 765                         if (ret) {
 766                                 ERR("Consumer add metadata stream %" PRIu64 " failed. Continuing",
 767                                                 new_stream->key);
 768                                 consumer_stream_free(new_stream);
 769                                 goto end_nosignal;
 770                         }
 771                         stream_pipe = ctx->consumer_metadata_pipe;
 772                 } else {
 773                         ret = consumer_add_data_stream(new_stream);
 774                         if (ret) {
 775                                 ERR("Consumer add stream %" PRIu64 " failed. Continuing",
 776                                                 new_stream->key);
 777                                 consumer_stream_free(new_stream);
 778                                 goto end_nosignal;
 779                         }
 780                         stream_pipe = ctx->consumer_data_pipe;
 781                 }
 782
 783                 /* Vitible to other threads */
 784                 new_stream->globally_visible = 1;
 785
 786                 health_code_update();
 787
 788                 ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
 789                 if (ret < 0) {
 790                         ERR("Consumer write %s stream to pipe %d",
 791                                         new_stream->metadata_flag ? "metadata" : "data",
 792                                         lttng_pipe_get_writefd(stream_pipe));
 793                         if (new_stream->metadata_flag) {
 794                                 consumer_del_stream_for_metadata(new_stream);
 795                         } else {
 796                                 consumer_del_stream_for_data(new_stream);
 797                         }
 798                         goto end_nosignal;
 799                 }
 800
 801                 DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
 802                                 new_stream->name, fd, new_stream->relayd_stream_id);
 803                 break;
 804         }
 805         case LTTNG_CONSUMER_STREAMS_SENT:
 806         {
 807                 struct lttng_consumer_channel *channel;
 808
 809                 /*
 810                  * Get stream's channel reference. Needed when adding the stream to the
 811                  * global hash table.
 812                  */
 813                 channel = consumer_find_channel(msg.u.sent_streams.channel_key);
 814                 if (!channel) {
 815                         /*
 816                          * We could not find the channel. Can happen if cpu hotplug
 817                          * happens while tearing down.
 818                          */
 819                         ERR("Unable to find channel key %" PRIu64,
 820                                         msg.u.sent_streams.channel_key);
 821                         ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
 822                 }
 823
 824                 health_code_update();
 825
 826                 /*
 827                  * Send status code to session daemon.
 828                  */
 829                 ret = consumer_send_status_msg(sock, ret_code);
 830                 if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
 831                         /* Somehow, the session daemon is not responding anymore. */
 832                         goto end_nosignal;
 833                 }
 834
 835                 health_code_update();
 836
 837                 /*
 838                  * We should not send this message if we don't monitor the
 839                  * streams in this channel.
 840                  */
 841                 if (!channel->monitor) {
 842                         break;
 843                 }
 844
 845                 health_code_update();
 846                 /* Send stream to relayd if the stream has an ID. */
 847                 if (msg.u.sent_streams.net_seq_idx != (uint64_t) -1ULL) {
 848                         ret = consumer_send_relayd_streams_sent(
 849                                         msg.u.sent_streams.net_seq_idx);
 850                         if (ret < 0) {
 851                                 goto end_nosignal;
 852                         }
 853                         channel->streams_sent_to_relayd = true;
 854                 }
 855                 break;
 856         }
 857         case LTTNG_CONSUMER_UPDATE_STREAM:
 858         {
 859                 rcu_read_unlock();
 860                 return -ENOSYS;
 861         }
 862         case LTTNG_CONSUMER_DESTROY_RELAYD:
 863         {
 864                 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
 865                 struct consumer_relayd_sock_pair *relayd;
 866
 867                 DBG("Kernel consumer destroying relayd %" PRIu64, index);
 868
 869                 /* Get relayd reference if exists. */
 870                 relayd = consumer_find_relayd(index);
 871                 if (relayd == NULL) {
 872                         DBG("Unable to find relayd %" PRIu64, index);
 873                         ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
 874                 }
 875
 876                 /*
 877                  * Each relayd socket pair has a refcount of stream attached to it
 878                  * which tells if the relayd is still active or not depending on the
 879                  * refcount value.
 880                  *
 881                  * This will set the destroy flag of the relayd object and destroy it
 882                  * if the refcount reaches zero when called.
 883                  *
 884                  * The destroy can happen either here or when a stream fd hangs up.
 885                  */
 886                 if (relayd) {
 887                         consumer_flag_relayd_for_destroy(relayd);
 888                 }
 889
 890                 health_code_update();
 891
 892                 ret = consumer_send_status_msg(sock, ret_code);
 893                 if (ret < 0) {
 894                         /* Somehow, the session daemon is not responding anymore. */
 895                         goto error_fatal;
 896                 }
 897
 898                 goto end_nosignal;
 899         }
 900         case LTTNG_CONSUMER_DATA_PENDING:
 901         {
 902                 int32_t ret;
 903                 uint64_t id = msg.u.data_pending.session_id;
 904
 905                 DBG("Kernel consumer data pending command for id %" PRIu64, id);
 906
 907                 ret = consumer_data_pending(id);
 908
 909                 health_code_update();
 910
 911                 /* Send back returned value to session daemon */
 912                 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
 913                 if (ret < 0) {
 914                         PERROR("send data pending ret code");
 915                         goto error_fatal;
 916                 }
 917
 918                 /*
 919                  * No need to send back a status message since the data pending
 920                  * returned value is the response.
 921                  */
 922                 break;
 923         }
 924         case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
 925         {
 926                 if (msg.u.snapshot_channel.metadata == 1) {
 927                         ret = lttng_kconsumer_snapshot_metadata(msg.u.snapshot_channel.key,
 928                                         msg.u.snapshot_channel.pathname,
 929                                         msg.u.snapshot_channel.relayd_id, ctx);
 930                         if (ret < 0) {
 931                                 ERR("Snapshot metadata failed");
 932                                 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
 933                         }
 934                 } else {
 935                         ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
 936                                         msg.u.snapshot_channel.pathname,
 937                                         msg.u.snapshot_channel.relayd_id,
 938                                         msg.u.snapshot_channel.nb_packets_per_stream,
 939                                         ctx);
 940                         if (ret < 0) {
 941                                 ERR("Snapshot channel failed");
 942                                 ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
 943                         }
 944                 }
 945
 946                 health_code_update();
 947
 948                 ret = consumer_send_status_msg(sock, ret_code);
 949                 if (ret < 0) {
 950                         /* Somehow, the session daemon is not responding anymore. */
 951                         goto end_nosignal;
 952                 }
 953                 break;
 954         }
 955         case LTTNG_CONSUMER_DESTROY_CHANNEL:
 956         {
 957                 uint64_t key = msg.u.destroy_channel.key;
 958                 struct lttng_consumer_channel *channel;
 959
 960                 channel = consumer_find_channel(key);
 961                 if (!channel) {
 962                         ERR("Kernel consumer destroy channel %" PRIu64 " not found", key);
 963                         ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
 964                 }
 965
 966                 health_code_update();
 967
 968                 ret = consumer_send_status_msg(sock, ret_code);
 969                 if (ret < 0) {
 970                         /* Somehow, the session daemon is not responding anymore. */
 971                         goto end_nosignal;
 972                 }
 973
 974                 health_code_update();
 975
 976                 /* Stop right now if no channel was found. */
 977                 if (!channel) {
 978                         goto end_nosignal;
 979                 }
 980
 981                 /*
 982                  * This command should ONLY be issued for channel with streams set in
 983                  * no monitor mode.
 984                  */
 985                 assert(!channel->monitor);
 986
 987                 /*
 988                  * The refcount should ALWAYS be 0 in the case of a channel in no
 989                  * monitor mode.
 990                  */
 991                 assert(!uatomic_sub_return(&channel->refcount, 1));
 992
 993                 consumer_del_channel(channel);
 994
 995                 goto end_nosignal;
 996         }
 997         case LTTNG_CONSUMER_DISCARDED_EVENTS:
 998         {
 999                 uint64_t ret;
1000                 struct lttng_consumer_channel *channel;
1001                 uint64_t id = msg.u.discarded_events.session_id;
1002                 uint64_t key = msg.u.discarded_events.channel_key;
1003
1004                 DBG("Kernel consumer discarded events command for session id %"
1005                                 PRIu64 ", channel key %" PRIu64, id, key);
1006
1007                 channel = consumer_find_channel(key);
1008                 if (!channel) {
1009                         ERR("Kernel consumer discarded events channel %"
1010                                         PRIu64 " not found", key);
1011                         ret = 0;
1012                 } else {
1013                         ret = channel->discarded_events;
1014                 }
1015
1016                 health_code_update();
1017
1018                 /* Send back returned value to session daemon */
1019                 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
1020                 if (ret < 0) {
1021                         PERROR("send discarded events");
1022                         goto error_fatal;
1023                 }
1024
1025                 break;
1026         }
1027         case LTTNG_CONSUMER_LOST_PACKETS:
1028         {
1029                 uint64_t ret;
1030                 struct lttng_consumer_channel *channel;
1031                 uint64_t id = msg.u.lost_packets.session_id;
1032                 uint64_t key = msg.u.lost_packets.channel_key;
1033
1034                 DBG("Kernel consumer lost packets command for session id %"
1035                                 PRIu64 ", channel key %" PRIu64, id, key);
1036
1037                 channel = consumer_find_channel(key);
1038                 if (!channel) {
1039                         ERR("Kernel consumer lost packets channel %"
1040                                         PRIu64 " not found", key);
1041                         ret = 0;
1042                 } else {
1043                         ret = channel->lost_packets;
1044                 }
1045
1046                 health_code_update();
1047
1048                 /* Send back returned value to session daemon */
1049                 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
1050                 if (ret < 0) {
1051                         PERROR("send lost packets");
1052                         goto error_fatal;
1053                 }
1054
1055                 break;
1056         }
1057         default:
1058                 goto end_nosignal;
1059         }
1060
1061 end_nosignal:
1062         rcu_read_unlock();
1063
1064         /*
1065          * Return 1 to indicate success since the 0 value can be a socket
1066          * shutdown during the recv() or send() call.
1067          */
1068         health_code_update();
1069         return 1;
1070
1071 error_fatal:
1072         rcu_read_unlock();
1073         /* This will issue a consumer stop. */
1074         return -1;
1075 }
1076
1077 /*
1078  * Populate index values of a kernel stream. Values are set in big endian order.
1079  *
1080  * Return 0 on success or else a negative value.
1081  */
1082 static int get_index_values(struct ctf_packet_index *index, int infd)
1083 {
1084         int ret;
1085
1086         ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin);
1087         if (ret < 0) {
1088                 PERROR("kernctl_get_timestamp_begin");
1089                 goto error;
1090         }
1091         index->timestamp_begin = htobe64(index->timestamp_begin);
1092
1093         ret = kernctl_get_timestamp_end(infd, &index->timestamp_end);
1094         if (ret < 0) {
1095                 PERROR("kernctl_get_timestamp_end");
1096                 goto error;
1097         }
1098         index->timestamp_end = htobe64(index->timestamp_end);
1099
1100         ret = kernctl_get_events_discarded(infd, &index->events_discarded);
1101         if (ret < 0) {
1102                 PERROR("kernctl_get_events_discarded");
1103                 goto error;
1104         }
1105         index->events_discarded = htobe64(index->events_discarded);
1106
1107         ret = kernctl_get_content_size(infd, &index->content_size);
1108         if (ret < 0) {
1109                 PERROR("kernctl_get_content_size");
1110                 goto error;
1111         }
1112         index->content_size = htobe64(index->content_size);
1113
1114         ret = kernctl_get_packet_size(infd, &index->packet_size);
1115         if (ret < 0) {
1116                 PERROR("kernctl_get_packet_size");
1117                 goto error;
1118         }
1119         index->packet_size = htobe64(index->packet_size);
1120
1121         ret = kernctl_get_stream_id(infd, &index->stream_id);
1122         if (ret < 0) {
1123                 PERROR("kernctl_get_stream_id");
1124                 goto error;
1125         }
1126         index->stream_id = htobe64(index->stream_id);
1127
1128         ret = kernctl_get_instance_id(infd, &index->stream_instance_id);
1129         if (ret < 0) {
1130                 if (ret == -ENOTTY) {
1131                         /* Command not implemented by lttng-modules. */
1132                         index->stream_instance_id = -1ULL;
1133                         ret = 0;
1134                 } else {
1135                         PERROR("kernctl_get_instance_id");
1136                         goto error;
1137                 }
1138         }
1139         index->stream_instance_id = htobe64(index->stream_instance_id);
1140
1141         ret = kernctl_get_sequence_number(infd, &index->packet_seq_num);
1142         if (ret < 0) {
1143                 if (ret == -ENOTTY) {
1144                         /* Command not implemented by lttng-modules. */
1145                         index->packet_seq_num = -1ULL;
1146                         ret = 0;
1147                 } else {
1148                         PERROR("kernctl_get_sequence_number");
1149                         goto error;
1150                 }
1151         }
1152         index->packet_seq_num = htobe64(index->packet_seq_num);
1153
1154 error:
1155         return ret;
1156 }
1157 /*
1158  * Sync metadata meaning request them to the session daemon and snapshot to the
1159  * metadata thread can consumer them.
1160  *
1161  * Metadata stream lock MUST be acquired.
1162  *
1163  * Return 0 if new metadatda is available, EAGAIN if the metadata stream
1164  * is empty or a negative value on error.
1165  */
1166 int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata)
1167 {
1168         int ret;
1169
1170         assert(metadata);
1171
1172         ret = kernctl_buffer_flush(metadata->wait_fd);
1173         if (ret < 0) {
1174                 ERR("Failed to flush kernel stream");
1175                 goto end;
1176         }
1177
1178         ret = kernctl_snapshot(metadata->wait_fd);
1179         if (ret < 0) {
1180                 if (ret != -EAGAIN) {
1181                         ERR("Sync metadata, taking kernel snapshot failed.");
1182                         goto end;
1183                 }
1184                 DBG("Sync metadata, no new kernel metadata");
1185                 /* No new metadata, exit. */
1186                 ret = ENODATA;
1187                 goto end;
1188         }
1189
1190 end:
1191         return ret;
1192 }
1193
1194 static
1195 int update_stream_stats(struct lttng_consumer_stream *stream)
1196 {
1197         int ret;
1198         uint64_t seq, discarded;
1199
1200         ret = kernctl_get_sequence_number(stream->wait_fd, &seq);
1201         if (ret < 0) {
1202                 if (ret == -ENOTTY) {
1203                         /* Command not implemented by lttng-modules. */
1204                         seq = -1ULL;
1205                         ret = 0;
1206                 } else {
1207                         PERROR("kernctl_get_sequence_number");
1208                         goto end;
1209                 }
1210         }
1211
1212         /*
1213          * Start the sequence when we extract the first packet in case we don't
1214          * start at 0 (for example if a consumer is not connected to the
1215          * session immediately after the beginning).
1216          */
1217         if (stream->last_sequence_number == -1ULL) {
1218                 stream->last_sequence_number = seq;
1219         } else if (seq > stream->last_sequence_number) {
1220                 stream->chan->lost_packets += seq -
1221                                 stream->last_sequence_number - 1;
1222         } else {
1223                 /* seq <= last_sequence_number */
1224                 ERR("Sequence number inconsistent : prev = %" PRIu64
1225                                 ", current = %" PRIu64,
1226                                 stream->last_sequence_number, seq);
1227                 ret = -1;
1228                 goto end;
1229         }
1230         stream->last_sequence_number = seq;
1231
1232         ret = kernctl_get_events_discarded(stream->wait_fd, &discarded);
1233         if (ret < 0) {
1234                 PERROR("kernctl_get_events_discarded");
1235                 goto end;
1236         }
1237         if (discarded < stream->last_discarded_events) {
1238                 /*
1239                  * Overflow has occurred. We assume only one wrap-around
1240                  * has occurred.
1241                  */
1242                 stream->chan->discarded_events += (1ULL << (CAA_BITS_PER_LONG - 1)) -
1243                         stream->last_discarded_events + discarded;
1244         } else {
1245                 stream->chan->discarded_events += discarded -
1246                         stream->last_discarded_events;
1247         }
1248         stream->last_discarded_events = discarded;
1249         ret = 0;
1250
1251 end:
1252         return ret;
1253 }
1254
1255 /*
1256  * Check if the local version of the metadata stream matches with the version
1257  * of the metadata stream in the kernel. If it was updated, set the reset flag
1258  * on the stream.
1259  */
1260 static
1261 int metadata_stream_check_version(int infd, struct lttng_consumer_stream *stream)
1262 {
1263         int ret;
1264         uint64_t cur_version;
1265
1266         ret = kernctl_get_metadata_version(infd, &cur_version);
1267         if (ret < 0) {
1268                 if (ret == -ENOTTY) {
1269                         /*
1270                          * LTTng-modules does not implement this
1271                          * command.
1272                          */
1273                         ret = 0;
1274                         goto end;
1275                 }
1276                 ERR("Failed to get the metadata version");
1277                 goto end;
1278         }
1279
1280         if (stream->metadata_version == cur_version) {
1281                 ret = 0;
1282                 goto end;
1283         }
1284
1285         DBG("New metadata version detected");
1286         stream->metadata_version = cur_version;
1287         stream->reset_metadata_flag = 1;
1288         ret = 0;
1289
1290 end:
1291         return ret;
1292 }
1293
1294 /*
1295  * Consume data on a file descriptor and write it on a trace file.
1296  */
1297 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
1298                 struct lttng_consumer_local_data *ctx)
1299 {
1300         unsigned long len, subbuf_size, padding;
1301         int err, write_index = 1;
1302         ssize_t ret = 0;
1303         int infd = stream->wait_fd;
1304         struct ctf_packet_index index;
1305
1306         DBG("In read_subbuffer (infd : %d)", infd);
1307
1308         /* Get the next subbuffer */
1309         err = kernctl_get_next_subbuf(infd);
1310         if (err != 0) {
1311                 /*
1312                  * This is a debug message even for single-threaded consumer,
1313                  * because poll() have more relaxed criterions than get subbuf,
1314                  * so get_subbuf may fail for short race windows where poll()
1315                  * would issue wakeups.
1316                  */
1317                 DBG("Reserving sub buffer failed (everything is normal, "
1318                                 "it is due to concurrency)");
1319                 ret = err;
1320                 goto end;
1321         }
1322
1323         /* Get the full subbuffer size including padding */
1324         err = kernctl_get_padded_subbuf_size(infd, &len);
1325         if (err != 0) {
1326                 PERROR("Getting sub-buffer len failed.");
1327                 err = kernctl_put_subbuf(infd);
1328                 if (err != 0) {
1329                         if (err == -EFAULT) {
1330                                 PERROR("Error in unreserving sub buffer\n");
1331                         } else if (err == -EIO) {
1332                                 /* Should never happen with newer LTTng versions */
1333                                 PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1334                         }
1335                         ret = err;
1336                         goto end;
1337                 }
1338                 ret = err;
1339                 goto end;
1340         }
1341
1342         if (!stream->metadata_flag) {
1343                 ret = get_index_values(&index, infd);
1344                 if (ret < 0) {
1345                         err = kernctl_put_subbuf(infd);
1346                         if (err != 0) {
1347                                 if (err == -EFAULT) {
1348                                         PERROR("Error in unreserving sub buffer\n");
1349                                 } else if (err == -EIO) {
1350                                         /* Should never happen with newer LTTng versions */
1351                                         PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1352                                 }
1353                                 ret = err;
1354                                 goto end;
1355                         }
1356                         goto end;
1357                 }
1358                 ret = update_stream_stats(stream);
1359                 if (ret < 0) {
1360                         err = kernctl_put_subbuf(infd);
1361                         if (err != 0) {
1362                                 if (err == -EFAULT) {
1363                                         PERROR("Error in unreserving sub buffer\n");
1364                                 } else if (err == -EIO) {
1365                                         /* Should never happen with newer LTTng versions */
1366                                         PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1367                                 }
1368                                 ret = err;
1369                                 goto end;
1370                         }
1371                         goto end;
1372                 }
1373         } else {
1374                 write_index = 0;
1375                 ret = metadata_stream_check_version(infd, stream);
1376                 if (ret < 0) {
1377                         err = kernctl_put_subbuf(infd);
1378                         if (err != 0) {
1379                                 if (err == -EFAULT) {
1380                                         PERROR("Error in unreserving sub buffer\n");
1381                                 } else if (err == -EIO) {
1382                                         /* Should never happen with newer LTTng versions */
1383                                         PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1384                                 }
1385                                 ret = err;
1386                                 goto end;
1387                         }
1388                         goto end;
1389                 }
1390         }
1391
1392         switch (stream->chan->output) {
1393         case CONSUMER_CHANNEL_SPLICE:
1394                 /*
1395                  * XXX: The lttng-modules splice "actor" does not handle copying
1396                  * partial pages hence only using the subbuffer size without the
1397                  * padding makes the splice fail.
1398                  */
1399                 subbuf_size = len;
1400                 padding = 0;
1401
1402                 /* splice the subbuffer to the tracefile */
1403                 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
1404                                 padding, &index);
1405                 /*
1406                  * XXX: Splice does not support network streaming so the return value
1407                  * is simply checked against subbuf_size and not like the mmap() op.
1408                  */
1409                 if (ret != subbuf_size) {
1410                         /*
1411                          * display the error but continue processing to try
1412                          * to release the subbuffer
1413                          */
1414                         ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
1415                                         ret, subbuf_size);
1416                         write_index = 0;
1417                 }
1418                 break;
1419         case CONSUMER_CHANNEL_MMAP:
1420         {
1421                 const char *subbuf_addr;
1422                 struct lttng_buffer_view subbuf_view;
1423
1424                 /* Get subbuffer size without padding */
1425                 err = kernctl_get_subbuf_size(infd, &subbuf_size);
1426                 if (err != 0) {
1427                         PERROR("Getting sub-buffer len failed.");
1428                         err = kernctl_put_subbuf(infd);
1429                         if (err != 0) {
1430                                 if (err == -EFAULT) {
1431                                         PERROR("Error in unreserving sub buffer\n");
1432                                 } else if (err == -EIO) {
1433                                         /* Should never happen with newer LTTng versions */
1434                                         PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1435                                 }
1436                                 ret = err;
1437                                 goto end;
1438                         }
1439                         ret = err;
1440                         goto end;
1441                 }
1442
1443                 ret = get_current_subbuf_addr(stream, &subbuf_addr);
1444                 if (ret) {
1445                         goto error_put_subbuf;
1446                 }
1447
1448                 /* Make sure the tracer is not gone mad on us! */
1449                 assert(len >= subbuf_size);
1450
1451                 padding = len - subbuf_size;
1452
1453                 subbuf_view = lttng_buffer_view_init(subbuf_addr, 0, len);
1454
1455                 /* write the subbuffer to the tracefile */
1456                 ret = lttng_consumer_on_read_subbuffer_mmap(
1457                                 ctx, stream, &subbuf_view, padding, &index);
1458                 /*
1459                  * The mmap operation should write subbuf_size amount of data
1460                  * when network streaming or the full padding (len) size when we
1461                  * are _not_ streaming.
1462                  */
1463                 if ((ret != subbuf_size && stream->relayd_id != (uint64_t) -1ULL) ||
1464                                 (ret != len && stream->relayd_id == (uint64_t) -1ULL)) {
1465                         /*
1466                          * Display the error but continue processing to try to release the
1467                          * subbuffer. This is a DBG statement since this is possible to
1468                          * happen without being a critical error.
1469                          */
1470                         DBG("Error writing to tracefile "
1471                                         "(ret: %zd != len: %lu != subbuf_size: %lu)",
1472                                         ret, len, subbuf_size);
1473                         write_index = 0;
1474                 }
1475                 break;
1476         }
1477         default:
1478                 ERR("Unknown output method");
1479                 ret = -EPERM;
1480         }
1481 error_put_subbuf:
1482         err = kernctl_put_next_subbuf(infd);
1483         if (err != 0) {
1484                 if (err == -EFAULT) {
1485                         PERROR("Error in unreserving sub buffer\n");
1486                 } else if (err == -EIO) {
1487                         /* Should never happen with newer LTTng versions */
1488                         PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
1489                 }
1490                 ret = err;
1491                 goto end;
1492         }
1493
1494         /* Write index if needed. */
1495         if (!write_index) {
1496                 goto end;
1497         }
1498
1499         if (stream->chan->live_timer_interval && !stream->metadata_flag) {
1500                 /*
1501                  * In live, block until all the metadata is sent.
1502                  */
1503                 pthread_mutex_lock(&stream->metadata_timer_lock);
1504                 assert(!stream->missed_metadata_flush);
1505                 stream->waiting_on_metadata = true;
1506                 pthread_mutex_unlock(&stream->metadata_timer_lock);
1507
1508                 err = consumer_stream_sync_metadata(ctx, stream->session_id);
1509
1510                 pthread_mutex_lock(&stream->metadata_timer_lock);
1511                 stream->waiting_on_metadata = false;
1512                 if (stream->missed_metadata_flush) {
1513                         stream->missed_metadata_flush = false;
1514                         pthread_mutex_unlock(&stream->metadata_timer_lock);
1515                         (void) consumer_flush_kernel_index(stream);
1516                 } else {
1517                         pthread_mutex_unlock(&stream->metadata_timer_lock);
1518                 }
1519                 if (err < 0) {
1520                         goto end;
1521                 }
1522         }
1523
1524         err = consumer_stream_write_index(stream, &index);
1525         if (err < 0) {
1526                 goto end;
1527         }
1528
1529 end:
1530         return ret;
1531 }
1532
1533 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1534 {
1535         int ret;
1536
1537         assert(stream);
1538
1539         /*
1540          * Don't create anything if this is set for streaming or should not be
1541          * monitored.
1542          */
1543         if (stream->relayd_id == (uint64_t) -1ULL && stream->chan->monitor) {
1544                 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
1545                                 stream->chan->tracefile_size, stream->tracefile_count_current,
1546                                 stream->uid, stream->gid, NULL);
1547                 if (ret < 0) {
1548                         goto error;
1549                 }
1550                 stream->out_fd = ret;
1551                 stream->tracefile_size_current = 0;
1552
1553                 if (!stream->metadata_flag) {
1554                         struct lttng_index_file *index_file;
1555
1556                         index_file = lttng_index_file_create(stream->chan->pathname,
1557                                         stream->name, stream->uid, stream->gid,
1558                                         stream->chan->tracefile_size,
1559                                         stream->tracefile_count_current,
1560                                         CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
1561                         if (!index_file) {
1562                                 goto error;
1563                         }
1564                         stream->index_file = index_file;
1565                 }
1566         }
1567
1568         if (stream->output == LTTNG_EVENT_MMAP) {
1569                 /* get the len of the mmap region */
1570                 unsigned long mmap_len;
1571
1572                 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
1573                 if (ret != 0) {
1574                         PERROR("kernctl_get_mmap_len");
1575                         goto error_close_fd;
1576                 }
1577                 stream->mmap_len = (size_t) mmap_len;
1578
1579                 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
1580                                 MAP_PRIVATE, stream->wait_fd, 0);
1581                 if (stream->mmap_base == MAP_FAILED) {
1582                         PERROR("Error mmaping");
1583                         ret = -1;
1584                         goto error_close_fd;
1585                 }
1586         }
1587
1588         /* we return 0 to let the library handle the FD internally */
1589         return 0;
1590
1591 error_close_fd:
1592         if (stream->out_fd >= 0) {
1593                 int err;
1594
1595                 err = close(stream->out_fd);
1596                 assert(!err);
1597                 stream->out_fd = -1;
1598         }
1599 error:
1600         return ret;
1601 }
1602
1603 /*
1604  * Check if data is still being extracted from the buffers for a specific
1605  * stream. Consumer data lock MUST be acquired before calling this function
1606  * and the stream lock.
1607  *
1608  * Return 1 if the traced data are still getting read else 0 meaning that the
1609  * data is available for trace viewer reading.
1610  */
1611 int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
1612 {
1613         int ret;
1614
1615         assert(stream);
1616
1617         if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1618                 ret = 0;
1619                 goto end;
1620         }
1621
1622         ret = kernctl_get_next_subbuf(stream->wait_fd);
1623         if (ret == 0) {
1624                 /* There is still data so let's put back this subbuffer. */
1625                 ret = kernctl_put_subbuf(stream->wait_fd);
1626                 assert(ret == 0);
1627                 ret = 1;   /* Data is pending */
1628                 goto end;
1629         }
1630
1631         /* Data is NOT pending and ready to be read. */
1632         ret = 0;
1633
1634 end:
1635         return ret;
1636 }