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

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