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

   1 /*
   2  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
   3  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License, version 2 only,
   7  * as published by the Free Software Foundation.
   8  *
   9  * This program is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  * GNU General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  17  */
  18
  19 #define _GNU_SOURCE
  20 #include <assert.h>
  21 #include <poll.h>
  22 #include <pthread.h>
  23 #include <stdlib.h>
  24 #include <string.h>
  25 #include <sys/mman.h>
  26 #include <sys/socket.h>
  27 #include <sys/types.h>
  28 #include <inttypes.h>
  29 #include <unistd.h>
  30 #include <sys/stat.h>
  31
  32 #include <common/common.h>
  33 #include <common/kernel-ctl/kernel-ctl.h>
  34 #include <common/sessiond-comm/sessiond-comm.h>
  35 #include <common/sessiond-comm/relayd.h>
  36 #include <common/compat/fcntl.h>
  37 #include <common/pipe.h>
  38 #include <common/relayd/relayd.h>
  39 #include <common/utils.h>
  40 #include <common/consumer-stream.h>
  41 #include <common/index/index.h>
  42 #include <common/consumer-timer.h>
  43
  44 #include "kernel-consumer.h"
  45
  46 extern struct lttng_consumer_global_data consumer_data;
  47 extern int consumer_poll_timeout;
  48 extern volatile int consumer_quit;
  49
  50 /*
  51  * Take a snapshot for a specific fd
  52  *
  53  * Returns 0 on success, < 0 on error
  54  */
  55 int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream)
  56 {
  57         int ret = 0;
  58         int infd = stream->wait_fd;
  59
  60         ret = kernctl_snapshot(infd);
  61         if (ret != 0) {
  62                 perror("Getting sub-buffer snapshot.");
  63                 ret = -errno;
  64         }
  65
  66         return ret;
  67 }
  68
  69 /*
  70  * Get the produced position
  71  *
  72  * Returns 0 on success, < 0 on error
  73  */
  74 int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
  75                 unsigned long *pos)
  76 {
  77         int ret;
  78         int infd = stream->wait_fd;
  79
  80         ret = kernctl_snapshot_get_produced(infd, pos);
  81         if (ret != 0) {
  82                 perror("kernctl_snapshot_get_produced");
  83                 ret = -errno;
  84         }
  85
  86         return ret;
  87 }
  88
  89 /*
  90  * Get the consumerd position
  91  *
  92  * Returns 0 on success, < 0 on error
  93  */
  94 int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
  95                 unsigned long *pos)
  96 {
  97         int ret;
  98         int infd = stream->wait_fd;
  99
 100         ret = kernctl_snapshot_get_consumed(infd, pos);
 101         if (ret != 0) {
 102                 perror("kernctl_snapshot_get_consumed");
 103                 ret = -errno;
 104         }
 105
 106         return ret;
 107 }
 108
 109 /*
 110  * Take a snapshot of all the stream of a channel
 111  *
 112  * Returns 0 on success, < 0 on error
 113  */
 114 int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
 115                 uint64_t relayd_id, uint64_t max_stream_size,
 116                 struct lttng_consumer_local_data *ctx)
 117 {
 118         int ret;
 119         unsigned long consumed_pos, produced_pos;
 120         struct lttng_consumer_channel *channel;
 121         struct lttng_consumer_stream *stream;
 122
 123         DBG("Kernel consumer snapshot channel %" PRIu64, key);
 124
 125         rcu_read_lock();
 126
 127         channel = consumer_find_channel(key);
 128         if (!channel) {
 129                 ERR("No channel found for key %" PRIu64, key);
 130                 ret = -1;
 131                 goto end;
 132         }
 133
 134         /* Splice is not supported yet for channel snapshot. */
 135         if (channel->output != CONSUMER_CHANNEL_MMAP) {
 136                 ERR("Unsupported output %d", channel->output);
 137                 ret = -1;
 138                 goto end;
 139         }
 140
 141         cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
 142                 /*
 143                  * Lock stream because we are about to change its state.
 144                  */
 145                 pthread_mutex_lock(&stream->lock);
 146
 147                 /*
 148                  * Assign the received relayd ID so we can use it for streaming. The streams
 149                  * are not visible to anyone so this is OK to change it.
 150                  */
 151                 stream->net_seq_idx = relayd_id;
 152                 channel->relayd_id = relayd_id;
 153                 if (relayd_id != (uint64_t) -1ULL) {
 154                         ret = consumer_send_relayd_stream(stream, path);
 155                         if (ret < 0) {
 156                                 ERR("sending stream to relayd");
 157                                 goto end_unlock;
 158                         }
 159                 } else {
 160                         ret = utils_create_stream_file(path, stream->name,
 161                                         stream->chan->tracefile_size,
 162                                         stream->tracefile_count_current,
 163                                         stream->uid, stream->gid, NULL);
 164                         if (ret < 0) {
 165                                 ERR("utils_create_stream_file");
 166                                 goto end_unlock;
 167                         }
 168
 169                         stream->out_fd = ret;
 170                         stream->tracefile_size_current = 0;
 171
 172                         DBG("Kernel consumer snapshot stream %s/%s (%" PRIu64 ")",
 173                                         path, stream->name, stream->key);
 174                 }
 175
 176                 ret = kernctl_buffer_flush(stream->wait_fd);
 177                 if (ret < 0) {
 178                         ERR("Failed to flush kernel stream");
 179                         ret = -errno;
 180                         goto end_unlock;
 181                 }
 182
 183                 ret = lttng_kconsumer_take_snapshot(stream);
 184                 if (ret < 0) {
 185                         ERR("Taking kernel snapshot");
 186                         goto end_unlock;
 187                 }
 188
 189                 ret = lttng_kconsumer_get_produced_snapshot(stream, &produced_pos);
 190                 if (ret < 0) {
 191                         ERR("Produced kernel snapshot position");
 192                         goto end_unlock;
 193                 }
 194
 195                 ret = lttng_kconsumer_get_consumed_snapshot(stream, &consumed_pos);
 196                 if (ret < 0) {
 197                         ERR("Consumerd kernel snapshot position");
 198                         goto end_unlock;
 199                 }
 200
 201                 if (stream->max_sb_size == 0) {
 202                         ret = kernctl_get_max_subbuf_size(stream->wait_fd,
 203                                         &stream->max_sb_size);
 204                         if (ret < 0) {
 205                                 ERR("Getting kernel max_sb_size");
 206                                 ret = -errno;
 207                                 goto end_unlock;
 208                         }
 209                 }
 210
 211                 /*
 212                  * The original value is sent back if max stream size is larger than
 213                  * the possible size of the snapshot. Also, we asume that the session
 214                  * daemon should never send a maximum stream size that is lower than
 215                  * subbuffer size.
 216                  */
 217                 consumed_pos = consumer_get_consumed_maxsize(consumed_pos,
 218                                 produced_pos, max_stream_size);
 219
 220                 while (consumed_pos < produced_pos) {
 221                         ssize_t read_len;
 222                         unsigned long len, padded_len;
 223
 224                         DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos);
 225
 226                         ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos);
 227                         if (ret < 0) {
 228                                 if (errno != EAGAIN) {
 229                                         PERROR("kernctl_get_subbuf snapshot");
 230                                         ret = -errno;
 231                                         goto end_unlock;
 232                                 }
 233                                 DBG("Kernel consumer get subbuf failed. Skipping it.");
 234                                 consumed_pos += stream->max_sb_size;
 235                                 continue;
 236                         }
 237
 238                         ret = kernctl_get_subbuf_size(stream->wait_fd, &len);
 239                         if (ret < 0) {
 240                                 ERR("Snapshot kernctl_get_subbuf_size");
 241                                 ret = -errno;
 242                                 goto error_put_subbuf;
 243                         }
 244
 245                         ret = kernctl_get_padded_subbuf_size(stream->wait_fd, &padded_len);
 246                         if (ret < 0) {
 247                                 ERR("Snapshot kernctl_get_padded_subbuf_size");
 248                                 ret = -errno;
 249                                 goto error_put_subbuf;
 250                         }
 251
 252                         read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
 253                                         padded_len - len, NULL);
 254                         /*
 255                          * We write the padded len in local tracefiles but the data len
 256                          * when using a relay. Display the error but continue processing
 257                          * to try to release the subbuffer.
 258                          */
 259                         if (relayd_id != (uint64_t) -1ULL) {
 260                                 if (read_len != len) {
 261                                         ERR("Error sending to the relay (ret: %zd != len: %lu)",
 262                                                         read_len, len);
 263                                 }
 264                         } else {
 265                                 if (read_len != padded_len) {
 266                                         ERR("Error writing to tracefile (ret: %zd != len: %lu)",
 267                                                         read_len, padded_len);
 268                                 }
 269                         }
 270
 271                         ret = kernctl_put_subbuf(stream->wait_fd);
 272                         if (ret < 0) {
 273                                 ERR("Snapshot kernctl_put_subbuf");
 274                                 ret = -errno;
 275                                 goto end_unlock;
 276                         }
 277                         consumed_pos += stream->max_sb_size;
 278                 }
 279
 280                 if (relayd_id == (uint64_t) -1ULL) {
 281                         if (stream->out_fd >= 0) {
 282                                 ret = close(stream->out_fd);
 283                                 if (ret < 0) {
 284                                         PERROR("Kernel consumer snapshot close out_fd");
 285                                         goto end_unlock;
 286                                 }
 287                                 stream->out_fd = -1;
 288                         }
 289                 } else {
 290                         close_relayd_stream(stream);
 291                         stream->net_seq_idx = (uint64_t) -1ULL;
 292                 }
 293                 pthread_mutex_unlock(&stream->lock);
 294         }
 295
 296         /* All good! */
 297         ret = 0;
 298         goto end;
 299
 300 error_put_subbuf:
 301         ret = kernctl_put_subbuf(stream->wait_fd);
 302         if (ret < 0) {
 303                 ret = -errno;
 304                 ERR("Snapshot kernctl_put_subbuf error path");
 305         }
 306 end_unlock:
 307         pthread_mutex_unlock(&stream->lock);
 308 end:
 309         rcu_read_unlock();
 310         return ret;
 311 }
 312
 313 /*
 314  * Read the whole metadata available for a snapshot.
 315  *
 316  * Returns 0 on success, < 0 on error
 317  */
 318 int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
 319                 uint64_t relayd_id, struct lttng_consumer_local_data *ctx)
 320 {
 321         int ret, use_relayd = 0;
 322         ssize_t ret_read;
 323         struct lttng_consumer_channel *metadata_channel;
 324         struct lttng_consumer_stream *metadata_stream;
 325
 326         assert(ctx);
 327
 328         DBG("Kernel consumer snapshot metadata with key %" PRIu64 " at path %s",
 329                         key, path);
 330
 331         rcu_read_lock();
 332
 333         metadata_channel = consumer_find_channel(key);
 334         if (!metadata_channel) {
 335                 ERR("Kernel snapshot metadata not found for key %" PRIu64, key);
 336                 ret = -1;
 337                 goto error;
 338         }
 339
 340         metadata_stream = metadata_channel->metadata_stream;
 341         assert(metadata_stream);
 342
 343         /* Flag once that we have a valid relayd for the stream. */
 344         if (relayd_id != (uint64_t) -1ULL) {
 345                 use_relayd = 1;
 346         }
 347
 348         if (use_relayd) {
 349                 ret = consumer_send_relayd_stream(metadata_stream, path);
 350                 if (ret < 0) {
 351                         goto error;
 352                 }
 353         } else {
 354                 ret = utils_create_stream_file(path, metadata_stream->name,
 355                                 metadata_stream->chan->tracefile_size,
 356                                 metadata_stream->tracefile_count_current,
 357                                 metadata_stream->uid, metadata_stream->gid, NULL);
 358                 if (ret < 0) {
 359                         goto error;
 360                 }
 361                 metadata_stream->out_fd = ret;
 362         }
 363
 364         do {
 365                 ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx);
 366                 if (ret_read < 0) {
 367                         if (ret_read != -EAGAIN) {
 368                                 ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)",
 369                                                 ret_read);
 370                                 goto error;
 371                         }
 372                         /* ret_read is negative at this point so we will exit the loop. */
 373                         continue;
 374                 }
 375         } while (ret_read >= 0);
 376
 377         if (use_relayd) {
 378                 close_relayd_stream(metadata_stream);
 379                 metadata_stream->net_seq_idx = (uint64_t) -1ULL;
 380         } else {
 381                 if (metadata_stream->out_fd >= 0) {
 382                         ret = close(metadata_stream->out_fd);
 383                         if (ret < 0) {
 384                                 PERROR("Kernel consumer snapshot metadata close out_fd");
 385                                 /*
 386                                  * Don't go on error here since the snapshot was successful at this
 387                                  * point but somehow the close failed.
 388                                  */
 389                         }
 390                         metadata_stream->out_fd = -1;
 391                 }
 392         }
 393
 394         ret = 0;
 395
 396         cds_list_del(&metadata_stream->send_node);
 397         consumer_stream_destroy(metadata_stream, NULL);
 398         metadata_channel->metadata_stream = NULL;
 399 error:
 400         rcu_read_unlock();
 401         return ret;
 402 }
 403
 404 /*
 405  * Receive command from session daemon and process it.
 406  *
 407  * Return 1 on success else a negative value or 0.
 408  */
 409 int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 410                 int sock, struct pollfd *consumer_sockpoll)
 411 {
 412         ssize_t ret;
 413         enum lttng_error_code ret_code = LTTNG_OK;
 414         struct lttcomm_consumer_msg msg;
 415
 416         ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
 417         if (ret != sizeof(msg)) {
 418                 if (ret > 0) {
 419                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
 420                         ret = -1;
 421                 }
 422                 return ret;
 423         }
 424         if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
 425                 /*
 426                  * Notify the session daemon that the command is completed.
 427                  *
 428                  * On transport layer error, the function call will print an error
 429                  * message so handling the returned code is a bit useless since we
 430                  * return an error code anyway.
 431                  */
 432                 (void) consumer_send_status_msg(sock, ret_code);
 433                 return -ENOENT;
 434         }
 435
 436         /* relayd needs RCU read-side protection */
 437         rcu_read_lock();
 438
 439         switch (msg.cmd_type) {
 440         case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
 441         {
 442                 /* Session daemon status message are handled in the following call. */
 443                 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
 444                                 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
 445                                 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
 446                                  msg.u.relayd_sock.relayd_session_id);
 447                 goto end_nosignal;
 448         }
 449         case LTTNG_CONSUMER_ADD_CHANNEL:
 450         {
 451                 struct lttng_consumer_channel *new_channel;
 452                 int ret_recv;
 453
 454                 /* First send a status message before receiving the fds. */
 455                 ret = consumer_send_status_msg(sock, ret_code);
 456                 if (ret < 0) {
 457                         /* Somehow, the session daemon is not responding anymore. */
 458                         goto error_fatal;
 459                 }
 460                 DBG("consumer_add_channel %" PRIu64, msg.u.channel.channel_key);
 461                 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
 462                                 msg.u.channel.session_id, msg.u.channel.pathname,
 463                                 msg.u.channel.name, msg.u.channel.uid, msg.u.channel.gid,
 464                                 msg.u.channel.relayd_id, msg.u.channel.output,
 465                                 msg.u.channel.tracefile_size,
 466                                 msg.u.channel.tracefile_count, 0,
 467                                 msg.u.channel.monitor,
 468                                 msg.u.channel.live_timer_interval);
 469                 if (new_channel == NULL) {
 470                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 471                         goto end_nosignal;
 472                 }
 473                 new_channel->nb_init_stream_left = msg.u.channel.nb_init_streams;
 474                 switch (msg.u.channel.output) {
 475                 case LTTNG_EVENT_SPLICE:
 476                         new_channel->output = CONSUMER_CHANNEL_SPLICE;
 477                         break;
 478                 case LTTNG_EVENT_MMAP:
 479                         new_channel->output = CONSUMER_CHANNEL_MMAP;
 480                         break;
 481                 default:
 482                         ERR("Channel output unknown %d", msg.u.channel.output);
 483                         goto end_nosignal;
 484                 }
 485
 486                 /* Translate and save channel type. */
 487                 switch (msg.u.channel.type) {
 488                 case CONSUMER_CHANNEL_TYPE_DATA:
 489                 case CONSUMER_CHANNEL_TYPE_METADATA:
 490                         new_channel->type = msg.u.channel.type;
 491                         break;
 492                 default:
 493                         assert(0);
 494                         goto end_nosignal;
 495                 };
 496
 497                 if (ctx->on_recv_channel != NULL) {
 498                         ret_recv = ctx->on_recv_channel(new_channel);
 499                         if (ret_recv == 0) {
 500                                 ret = consumer_add_channel(new_channel, ctx);
 501                         } else if (ret_recv < 0) {
 502                                 goto end_nosignal;
 503                         }
 504                 } else {
 505                         ret = consumer_add_channel(new_channel, ctx);
 506                 }
 507                 if (CONSUMER_CHANNEL_TYPE_DATA) {
 508                         consumer_timer_live_start(new_channel,
 509                                         msg.u.channel.live_timer_interval);
 510                 }
 511
 512                 /* If we received an error in add_channel, we need to report it. */
 513                 if (ret < 0) {
 514                         ret = consumer_send_status_msg(sock, ret);
 515                         if (ret < 0) {
 516                                 goto error_fatal;
 517                         }
 518                         goto end_nosignal;
 519                 }
 520
 521                 goto end_nosignal;
 522         }
 523         case LTTNG_CONSUMER_ADD_STREAM:
 524         {
 525                 int fd;
 526                 struct lttng_pipe *stream_pipe;
 527                 struct lttng_consumer_stream *new_stream;
 528                 struct lttng_consumer_channel *channel;
 529                 int alloc_ret = 0;
 530
 531                 /*
 532                  * Get stream's channel reference. Needed when adding the stream to the
 533                  * global hash table.
 534                  */
 535                 channel = consumer_find_channel(msg.u.stream.channel_key);
 536                 if (!channel) {
 537                         /*
 538                          * We could not find the channel. Can happen if cpu hotplug
 539                          * happens while tearing down.
 540                          */
 541                         ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key);
 542                         ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
 543                 }
 544
 545                 /* First send a status message before receiving the fds. */
 546                 ret = consumer_send_status_msg(sock, ret_code);
 547                 if (ret < 0) {
 548                         /* Somehow, the session daemon is not responding anymore. */
 549                         goto error_fatal;
 550                 }
 551                 if (ret_code != LTTNG_OK) {
 552                         /* Channel was not found. */
 553                         goto end_nosignal;
 554                 }
 555
 556                 /* Blocking call */
 557                 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 558                         rcu_read_unlock();
 559                         return -EINTR;
 560                 }
 561
 562                 /* Get stream file descriptor from socket */
 563                 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 564                 if (ret != sizeof(fd)) {
 565                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 566                         rcu_read_unlock();
 567                         return ret;
 568                 }
 569
 570                 /*
 571                  * Send status code to session daemon only if the recv works. If the
 572                  * above recv() failed, the session daemon is notified through the
 573                  * error socket and the teardown is eventually done.
 574                  */
 575                 ret = consumer_send_status_msg(sock, ret_code);
 576                 if (ret < 0) {
 577                         /* Somehow, the session daemon is not responding anymore. */
 578                         goto end_nosignal;
 579                 }
 580
 581                 new_stream = consumer_allocate_stream(channel->key,
 582                                 fd,
 583                                 LTTNG_CONSUMER_ACTIVE_STREAM,
 584                                 channel->name,
 585                                 channel->uid,
 586                                 channel->gid,
 587                                 channel->relayd_id,
 588                                 channel->session_id,
 589                                 msg.u.stream.cpu,
 590                                 &alloc_ret,
 591                                 channel->type,
 592                                 channel->monitor);
 593                 if (new_stream == NULL) {
 594                         switch (alloc_ret) {
 595                         case -ENOMEM:
 596                         case -EINVAL:
 597                         default:
 598                                 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 599                                 break;
 600                         }
 601                         goto end_nosignal;
 602                 }
 603
 604                 new_stream->chan = channel;
 605                 new_stream->wait_fd = fd;
 606                 switch (channel->output) {
 607                 case CONSUMER_CHANNEL_SPLICE:
 608                         new_stream->output = LTTNG_EVENT_SPLICE;
 609                         break;
 610                 case CONSUMER_CHANNEL_MMAP:
 611                         new_stream->output = LTTNG_EVENT_MMAP;
 612                         break;
 613                 default:
 614                         ERR("Stream output unknown %d", channel->output);
 615                         goto end_nosignal;
 616                 }
 617
 618                 /*
 619                  * We've just assigned the channel to the stream so increment the
 620                  * refcount right now. We don't need to increment the refcount for
 621                  * streams in no monitor because we handle manually the cleanup of
 622                  * those. It is very important to make sure there is NO prior
 623                  * consumer_del_stream() calls or else the refcount will be unbalanced.
 624                  */
 625                 if (channel->monitor) {
 626                         uatomic_inc(&new_stream->chan->refcount);
 627                 }
 628
 629                 /*
 630                  * The buffer flush is done on the session daemon side for the kernel
 631                  * so no need for the stream "hangup_flush_done" variable to be
 632                  * tracked. This is important for a kernel stream since we don't rely
 633                  * on the flush state of the stream to read data. It's not the case for
 634                  * user space tracing.
 635                  */
 636                 new_stream->hangup_flush_done = 0;
 637
 638                 if (ctx->on_recv_stream) {
 639                         ret = ctx->on_recv_stream(new_stream);
 640                         if (ret < 0) {
 641                                 consumer_stream_free(new_stream);
 642                                 goto end_nosignal;
 643                         }
 644                 }
 645
 646                 if (new_stream->metadata_flag) {
 647                         channel->metadata_stream = new_stream;
 648                 }
 649
 650                 /* Do not monitor this stream. */
 651                 if (!channel->monitor) {
 652                         DBG("Kernel consumer add stream %s in no monitor mode with "
 653                                         "relayd id %" PRIu64, new_stream->name,
 654                                         new_stream->net_seq_idx);
 655                         cds_list_add(&new_stream->send_node, &channel->streams.head);
 656                         break;
 657                 }
 658
 659                 /* Send stream to relayd if the stream has an ID. */
 660                 if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
 661                         ret = consumer_send_relayd_stream(new_stream,
 662                                         new_stream->chan->pathname);
 663                         if (ret < 0) {
 664                                 consumer_stream_free(new_stream);
 665                                 goto end_nosignal;
 666                         }
 667                 }
 668
 669                 /* Get the right pipe where the stream will be sent. */
 670                 if (new_stream->metadata_flag) {
 671                         ret = consumer_add_metadata_stream(new_stream);
 672                         if (ret) {
 673                                 ERR("Consumer add metadata stream %" PRIu64 " failed. Continuing",
 674                                                 new_stream->key);
 675                                 consumer_stream_free(new_stream);
 676                                 goto end_nosignal;
 677                         }
 678                         stream_pipe = ctx->consumer_metadata_pipe;
 679                 } else {
 680                         ret = consumer_add_data_stream(new_stream);
 681                         if (ret) {
 682                                 ERR("Consumer add stream %" PRIu64 " failed. Continuing",
 683                                                 new_stream->key);
 684                                 consumer_stream_free(new_stream);
 685                                 goto end_nosignal;
 686                         }
 687                         stream_pipe = ctx->consumer_data_pipe;
 688                 }
 689
 690                 /* Vitible to other threads */
 691                 new_stream->globally_visible = 1;
 692
 693                 ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
 694                 if (ret < 0) {
 695                         ERR("Consumer write %s stream to pipe %d",
 696                                         new_stream->metadata_flag ? "metadata" : "data",
 697                                         lttng_pipe_get_writefd(stream_pipe));
 698                         if (new_stream->metadata_flag) {
 699                                 consumer_del_stream_for_metadata(new_stream);
 700                         } else {
 701                                 consumer_del_stream_for_data(new_stream);
 702                         }
 703                         goto end_nosignal;
 704                 }
 705
 706                 DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
 707                                 new_stream->name, fd, new_stream->relayd_stream_id);
 708                 break;
 709         }
 710         case LTTNG_CONSUMER_UPDATE_STREAM:
 711         {
 712                 rcu_read_unlock();
 713                 return -ENOSYS;
 714         }
 715         case LTTNG_CONSUMER_DESTROY_RELAYD:
 716         {
 717                 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
 718                 struct consumer_relayd_sock_pair *relayd;
 719
 720                 DBG("Kernel consumer destroying relayd %" PRIu64, index);
 721
 722                 /* Get relayd reference if exists. */
 723                 relayd = consumer_find_relayd(index);
 724                 if (relayd == NULL) {
 725                         DBG("Unable to find relayd %" PRIu64, index);
 726                         ret_code = LTTNG_ERR_NO_CONSUMER;
 727                 }
 728
 729                 /*
 730                  * Each relayd socket pair has a refcount of stream attached to it
 731                  * which tells if the relayd is still active or not depending on the
 732                  * refcount value.
 733                  *
 734                  * This will set the destroy flag of the relayd object and destroy it
 735                  * if the refcount reaches zero when called.
 736                  *
 737                  * The destroy can happen either here or when a stream fd hangs up.
 738                  */
 739                 if (relayd) {
 740                         consumer_flag_relayd_for_destroy(relayd);
 741                 }
 742
 743                 ret = consumer_send_status_msg(sock, ret_code);
 744                 if (ret < 0) {
 745                         /* Somehow, the session daemon is not responding anymore. */
 746                         goto error_fatal;
 747                 }
 748
 749                 goto end_nosignal;
 750         }
 751         case LTTNG_CONSUMER_DATA_PENDING:
 752         {
 753                 int32_t ret;
 754                 uint64_t id = msg.u.data_pending.session_id;
 755
 756                 DBG("Kernel consumer data pending command for id %" PRIu64, id);
 757
 758                 ret = consumer_data_pending(id);
 759
 760                 /* Send back returned value to session daemon */
 761                 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
 762                 if (ret < 0) {
 763                         PERROR("send data pending ret code");
 764                         goto error_fatal;
 765                 }
 766
 767                 /*
 768                  * No need to send back a status message since the data pending
 769                  * returned value is the response.
 770                  */
 771                 break;
 772         }
 773         case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
 774         {
 775                 if (msg.u.snapshot_channel.metadata == 1) {
 776                         ret = lttng_kconsumer_snapshot_metadata(msg.u.snapshot_channel.key,
 777                                         msg.u.snapshot_channel.pathname,
 778                                         msg.u.snapshot_channel.relayd_id, ctx);
 779                         if (ret < 0) {
 780                                 ERR("Snapshot metadata failed");
 781                                 ret_code = LTTNG_ERR_KERN_META_FAIL;
 782                         }
 783                 } else {
 784                         ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
 785                                         msg.u.snapshot_channel.pathname,
 786                                         msg.u.snapshot_channel.relayd_id,
 787                                         msg.u.snapshot_channel.max_stream_size,
 788                                         ctx);
 789                         if (ret < 0) {
 790                                 ERR("Snapshot channel failed");
 791                                 ret_code = LTTNG_ERR_KERN_CHAN_FAIL;
 792                         }
 793                 }
 794
 795                 ret = consumer_send_status_msg(sock, ret_code);
 796                 if (ret < 0) {
 797                         /* Somehow, the session daemon is not responding anymore. */
 798                         goto end_nosignal;
 799                 }
 800                 break;
 801         }
 802         case LTTNG_CONSUMER_DESTROY_CHANNEL:
 803         {
 804                 uint64_t key = msg.u.destroy_channel.key;
 805                 struct lttng_consumer_channel *channel;
 806
 807                 channel = consumer_find_channel(key);
 808                 if (!channel) {
 809                         ERR("Kernel consumer destroy channel %" PRIu64 " not found", key);
 810                         ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
 811                 }
 812
 813                 ret = consumer_send_status_msg(sock, ret_code);
 814                 if (ret < 0) {
 815                         /* Somehow, the session daemon is not responding anymore. */
 816                         goto end_nosignal;
 817                 }
 818
 819                 /*
 820                  * This command should ONLY be issued for channel with streams set in
 821                  * no monitor mode.
 822                  */
 823                 assert(!channel->monitor);
 824
 825                 /*
 826                  * The refcount should ALWAYS be 0 in the case of a channel in no
 827                  * monitor mode.
 828                  */
 829                 assert(!uatomic_sub_return(&channel->refcount, 1));
 830
 831                 consumer_del_channel(channel);
 832
 833                 goto end_nosignal;
 834         }
 835         default:
 836                 goto end_nosignal;
 837         }
 838
 839 end_nosignal:
 840         rcu_read_unlock();
 841
 842         /*
 843          * Return 1 to indicate success since the 0 value can be a socket
 844          * shutdown during the recv() or send() call.
 845          */
 846         return 1;
 847
 848 error_fatal:
 849         rcu_read_unlock();
 850         /* This will issue a consumer stop. */
 851         return -1;
 852 }
 853
 854 /*
 855  * Populate index values of a kernel stream. Values are set in big endian order.
 856  *
 857  * Return 0 on success or else a negative value.
 858  */
 859 static int get_index_values(struct lttng_packet_index *index, int infd)
 860 {
 861         int ret;
 862
 863         ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin);
 864         if (ret < 0) {
 865                 PERROR("kernctl_get_timestamp_begin");
 866                 goto error;
 867         }
 868         index->timestamp_begin = htobe64(index->timestamp_begin);
 869
 870         ret = kernctl_get_timestamp_end(infd, &index->timestamp_end);
 871         if (ret < 0) {
 872                 PERROR("kernctl_get_timestamp_end");
 873                 goto error;
 874         }
 875         index->timestamp_end = htobe64(index->timestamp_end);
 876
 877         ret = kernctl_get_events_discarded(infd, &index->events_discarded);
 878         if (ret < 0) {
 879                 PERROR("kernctl_get_events_discarded");
 880                 goto error;
 881         }
 882         index->events_discarded = htobe64(index->events_discarded);
 883
 884         ret = kernctl_get_content_size(infd, &index->content_size);
 885         if (ret < 0) {
 886                 PERROR("kernctl_get_content_size");
 887                 goto error;
 888         }
 889         index->content_size = htobe64(index->content_size);
 890
 891         ret = kernctl_get_packet_size(infd, &index->packet_size);
 892         if (ret < 0) {
 893                 PERROR("kernctl_get_packet_size");
 894                 goto error;
 895         }
 896         index->packet_size = htobe64(index->packet_size);
 897
 898         ret = kernctl_get_stream_id(infd, &index->stream_id);
 899         if (ret < 0) {
 900                 PERROR("kernctl_get_stream_id");
 901                 goto error;
 902         }
 903         index->stream_id = htobe64(index->stream_id);
 904
 905 error:
 906         return ret;
 907 }
 908 /*
 909  * Sync metadata meaning request them to the session daemon and snapshot to the
 910  * metadata thread can consumer them.
 911  *
 912  * Metadata stream lock MUST be acquired.
 913  *
 914  * Return 0 if new metadatda is available, EAGAIN if the metadata stream
 915  * is empty or a negative value on error.
 916  */
 917 int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata)
 918 {
 919         int ret;
 920
 921         assert(metadata);
 922
 923         ret = kernctl_buffer_flush(metadata->wait_fd);
 924         if (ret < 0) {
 925                 ERR("Failed to flush kernel stream");
 926                 goto end;
 927         }
 928
 929         ret = kernctl_snapshot(metadata->wait_fd);
 930         if (ret < 0) {
 931                 if (errno != EAGAIN) {
 932                         ERR("Sync metadata, taking kernel snapshot failed.");
 933                         goto end;
 934                 }
 935                 DBG("Sync metadata, no new kernel metadata");
 936                 /* No new metadata, exit. */
 937                 ret = ENODATA;
 938                 goto end;
 939         }
 940
 941 end:
 942         return ret;
 943 }
 944
 945 /*
 946  * Consume data on a file descriptor and write it on a trace file.
 947  */
 948 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 949                 struct lttng_consumer_local_data *ctx)
 950 {
 951         unsigned long len, subbuf_size, padding;
 952         int err, write_index = 1;
 953         ssize_t ret = 0;
 954         int infd = stream->wait_fd;
 955         struct lttng_packet_index index;
 956
 957         DBG("In read_subbuffer (infd : %d)", infd);
 958
 959         /* Get the next subbuffer */
 960         err = kernctl_get_next_subbuf(infd);
 961         if (err != 0) {
 962                 /*
 963                  * This is a debug message even for single-threaded consumer,
 964                  * because poll() have more relaxed criterions than get subbuf,
 965                  * so get_subbuf may fail for short race windows where poll()
 966                  * would issue wakeups.
 967                  */
 968                 DBG("Reserving sub buffer failed (everything is normal, "
 969                                 "it is due to concurrency)");
 970                 ret = -errno;
 971                 goto end;
 972         }
 973
 974         /* Get the full subbuffer size including padding */
 975         err = kernctl_get_padded_subbuf_size(infd, &len);
 976         if (err != 0) {
 977                 perror("Getting sub-buffer len failed.");
 978                 ret = -errno;
 979                 goto end;
 980         }
 981
 982         if (!stream->metadata_flag) {
 983                 ret = get_index_values(&index, infd);
 984                 if (ret < 0) {
 985                         goto end;
 986                 }
 987         } else {
 988                 write_index = 0;
 989         }
 990
 991         switch (stream->chan->output) {
 992         case CONSUMER_CHANNEL_SPLICE:
 993                 /*
 994                  * XXX: The lttng-modules splice "actor" does not handle copying
 995                  * partial pages hence only using the subbuffer size without the
 996                  * padding makes the splice fail.
 997                  */
 998                 subbuf_size = len;
 999                 padding = 0;
1000
1001                 /* splice the subbuffer to the tracefile */
1002                 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
1003                                 padding, &index);
1004                 /*
1005                  * XXX: Splice does not support network streaming so the return value
1006                  * is simply checked against subbuf_size and not like the mmap() op.
1007                  */
1008                 if (ret != subbuf_size) {
1009                         /*
1010                          * display the error but continue processing to try
1011                          * to release the subbuffer
1012                          */
1013                         ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
1014                                         ret, subbuf_size);
1015                         write_index = 0;
1016                 }
1017                 break;
1018         case CONSUMER_CHANNEL_MMAP:
1019                 /* Get subbuffer size without padding */
1020                 err = kernctl_get_subbuf_size(infd, &subbuf_size);
1021                 if (err != 0) {
1022                         perror("Getting sub-buffer len failed.");
1023                         ret = -errno;
1024                         goto end;
1025                 }
1026
1027                 /* Make sure the tracer is not gone mad on us! */
1028                 assert(len >= subbuf_size);
1029
1030                 padding = len - subbuf_size;
1031
1032                 /* write the subbuffer to the tracefile */
1033                 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
1034                                 padding, &index);
1035                 /*
1036                  * The mmap operation should write subbuf_size amount of data when
1037                  * network streaming or the full padding (len) size when we are _not_
1038                  * streaming.
1039                  */
1040                 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
1041                                 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
1042                         /*
1043                          * Display the error but continue processing to try to release the
1044                          * subbuffer
1045                          */
1046                         ERR("Error writing to tracefile "
1047                                         "(ret: %zd != len: %lu != subbuf_size: %lu)",
1048                                         ret, len, subbuf_size);
1049                         write_index = 0;
1050                 }
1051                 break;
1052         default:
1053                 ERR("Unknown output method");
1054                 ret = -EPERM;
1055         }
1056
1057         err = kernctl_put_next_subbuf(infd);
1058         if (err != 0) {
1059                 if (errno == EFAULT) {
1060                         perror("Error in unreserving sub buffer\n");
1061                 } else if (errno == EIO) {
1062                         /* Should never happen with newer LTTng versions */
1063                         perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
1064                 }
1065                 ret = -errno;
1066                 goto end;
1067         }
1068
1069         /* Write index if needed. */
1070         if (!write_index) {
1071                 goto end;
1072         }
1073
1074         if (stream->chan->live_timer_interval && !stream->metadata_flag) {
1075                 /*
1076                  * In live, block until all the metadata is sent.
1077                  */
1078                 err = consumer_stream_sync_metadata(ctx, stream->session_id);
1079                 if (err < 0) {
1080                         goto end;
1081                 }
1082         }
1083
1084         err = consumer_stream_write_index(stream, &index);
1085         if (err < 0) {
1086                 goto end;
1087         }
1088
1089 end:
1090         return ret;
1091 }
1092
1093 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1094 {
1095         int ret;
1096
1097         assert(stream);
1098
1099         /*
1100          * Don't create anything if this is set for streaming or should not be
1101          * monitored.
1102          */
1103         if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
1104                 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
1105                                 stream->chan->tracefile_size, stream->tracefile_count_current,
1106                                 stream->uid, stream->gid, NULL);
1107                 if (ret < 0) {
1108                         goto error;
1109                 }
1110                 stream->out_fd = ret;
1111                 stream->tracefile_size_current = 0;
1112
1113                 if (!stream->metadata_flag) {
1114                         ret = index_create_file(stream->chan->pathname,
1115                                         stream->name, stream->uid, stream->gid,
1116                                         stream->chan->tracefile_size,
1117                                         stream->tracefile_count_current);
1118                         if (ret < 0) {
1119                                 goto error;
1120                         }
1121                         stream->index_fd = ret;
1122                 }
1123         }
1124
1125         if (stream->output == LTTNG_EVENT_MMAP) {
1126                 /* get the len of the mmap region */
1127                 unsigned long mmap_len;
1128
1129                 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
1130                 if (ret != 0) {
1131                         PERROR("kernctl_get_mmap_len");
1132                         ret = -errno;
1133                         goto error_close_fd;
1134                 }
1135                 stream->mmap_len = (size_t) mmap_len;
1136
1137                 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
1138                                 MAP_PRIVATE, stream->wait_fd, 0);
1139                 if (stream->mmap_base == MAP_FAILED) {
1140                         PERROR("Error mmaping");
1141                         ret = -1;
1142                         goto error_close_fd;
1143                 }
1144         }
1145
1146         /* we return 0 to let the library handle the FD internally */
1147         return 0;
1148
1149 error_close_fd:
1150         if (stream->out_fd >= 0) {
1151                 int err;
1152
1153                 err = close(stream->out_fd);
1154                 assert(!err);
1155                 stream->out_fd = -1;
1156         }
1157 error:
1158         return ret;
1159 }
1160
1161 /*
1162  * Check if data is still being extracted from the buffers for a specific
1163  * stream. Consumer data lock MUST be acquired before calling this function
1164  * and the stream lock.
1165  *
1166  * Return 1 if the traced data are still getting read else 0 meaning that the
1167  * data is available for trace viewer reading.
1168  */
1169 int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
1170 {
1171         int ret;
1172
1173         assert(stream);
1174
1175         if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1176                 ret = 0;
1177                 goto end;
1178         }
1179
1180         ret = kernctl_get_next_subbuf(stream->wait_fd);
1181         if (ret == 0) {
1182                 /* There is still data so let's put back this subbuffer. */
1183                 ret = kernctl_put_subbuf(stream->wait_fd);
1184                 assert(ret == 0);
1185                 ret = 1;   /* Data is pending */
1186                 goto end;
1187         }
1188
1189         /* Data is NOT pending and ready to be read. */
1190         ret = 0;
1191
1192 end:
1193         return ret;
1194 }