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                 consumer_timer_live_start(new_channel, msg.u.channel.live_timer_interval);
 508
 509                 /* If we received an error in add_channel, we need to report it. */
 510                 if (ret < 0) {
 511                         ret = consumer_send_status_msg(sock, ret);
 512                         if (ret < 0) {
 513                                 goto error_fatal;
 514                         }
 515                         goto end_nosignal;
 516                 }
 517
 518                 goto end_nosignal;
 519         }
 520         case LTTNG_CONSUMER_ADD_STREAM:
 521         {
 522                 int fd;
 523                 struct lttng_pipe *stream_pipe;
 524                 struct lttng_consumer_stream *new_stream;
 525                 struct lttng_consumer_channel *channel;
 526                 int alloc_ret = 0;
 527
 528                 /*
 529                  * Get stream's channel reference. Needed when adding the stream to the
 530                  * global hash table.
 531                  */
 532                 channel = consumer_find_channel(msg.u.stream.channel_key);
 533                 if (!channel) {
 534                         /*
 535                          * We could not find the channel. Can happen if cpu hotplug
 536                          * happens while tearing down.
 537                          */
 538                         ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key);
 539                         ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
 540                 }
 541
 542                 /* First send a status message before receiving the fds. */
 543                 ret = consumer_send_status_msg(sock, ret_code);
 544                 if (ret < 0) {
 545                         /* Somehow, the session daemon is not responding anymore. */
 546                         goto error_fatal;
 547                 }
 548                 if (ret_code != LTTNG_OK) {
 549                         /* Channel was not found. */
 550                         goto end_nosignal;
 551                 }
 552
 553                 /* Blocking call */
 554                 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 555                         rcu_read_unlock();
 556                         return -EINTR;
 557                 }
 558
 559                 /* Get stream file descriptor from socket */
 560                 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 561                 if (ret != sizeof(fd)) {
 562                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 563                         rcu_read_unlock();
 564                         return ret;
 565                 }
 566
 567                 /*
 568                  * Send status code to session daemon only if the recv works. If the
 569                  * above recv() failed, the session daemon is notified through the
 570                  * error socket and the teardown is eventually done.
 571                  */
 572                 ret = consumer_send_status_msg(sock, ret_code);
 573                 if (ret < 0) {
 574                         /* Somehow, the session daemon is not responding anymore. */
 575                         goto end_nosignal;
 576                 }
 577
 578                 new_stream = consumer_allocate_stream(channel->key,
 579                                 fd,
 580                                 LTTNG_CONSUMER_ACTIVE_STREAM,
 581                                 channel->name,
 582                                 channel->uid,
 583                                 channel->gid,
 584                                 channel->relayd_id,
 585                                 channel->session_id,
 586                                 msg.u.stream.cpu,
 587                                 &alloc_ret,
 588                                 channel->type,
 589                                 channel->monitor);
 590                 if (new_stream == NULL) {
 591                         switch (alloc_ret) {
 592                         case -ENOMEM:
 593                         case -EINVAL:
 594                         default:
 595                                 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 596                                 break;
 597                         }
 598                         goto end_nosignal;
 599                 }
 600
 601                 new_stream->chan = channel;
 602                 new_stream->wait_fd = fd;
 603                 switch (channel->output) {
 604                 case CONSUMER_CHANNEL_SPLICE:
 605                         new_stream->output = LTTNG_EVENT_SPLICE;
 606                         break;
 607                 case CONSUMER_CHANNEL_MMAP:
 608                         new_stream->output = LTTNG_EVENT_MMAP;
 609                         break;
 610                 default:
 611                         ERR("Stream output unknown %d", channel->output);
 612                         goto end_nosignal;
 613                 }
 614
 615                 /*
 616                  * We've just assigned the channel to the stream so increment the
 617                  * refcount right now. We don't need to increment the refcount for
 618                  * streams in no monitor because we handle manually the cleanup of
 619                  * those. It is very important to make sure there is NO prior
 620                  * consumer_del_stream() calls or else the refcount will be unbalanced.
 621                  */
 622                 if (channel->monitor) {
 623                         uatomic_inc(&new_stream->chan->refcount);
 624                 }
 625
 626                 /*
 627                  * The buffer flush is done on the session daemon side for the kernel
 628                  * so no need for the stream "hangup_flush_done" variable to be
 629                  * tracked. This is important for a kernel stream since we don't rely
 630                  * on the flush state of the stream to read data. It's not the case for
 631                  * user space tracing.
 632                  */
 633                 new_stream->hangup_flush_done = 0;
 634
 635                 if (ctx->on_recv_stream) {
 636                         ret = ctx->on_recv_stream(new_stream);
 637                         if (ret < 0) {
 638                                 consumer_stream_free(new_stream);
 639                                 goto end_nosignal;
 640                         }
 641                 }
 642
 643                 if (new_stream->metadata_flag) {
 644                         channel->metadata_stream = new_stream;
 645                 }
 646
 647                 /* Do not monitor this stream. */
 648                 if (!channel->monitor) {
 649                         DBG("Kernel consumer add stream %s in no monitor mode with "
 650                                         "relayd id %" PRIu64, new_stream->name,
 651                                         new_stream->net_seq_idx);
 652                         cds_list_add(&new_stream->send_node, &channel->streams.head);
 653                         break;
 654                 }
 655
 656                 /* Send stream to relayd if the stream has an ID. */
 657                 if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
 658                         ret = consumer_send_relayd_stream(new_stream,
 659                                         new_stream->chan->pathname);
 660                         if (ret < 0) {
 661                                 consumer_stream_free(new_stream);
 662                                 goto end_nosignal;
 663                         }
 664                 }
 665
 666                 /* Get the right pipe where the stream will be sent. */
 667                 if (new_stream->metadata_flag) {
 668                         ret = consumer_add_metadata_stream(new_stream);
 669                         if (ret) {
 670                                 ERR("Consumer add metadata stream %" PRIu64 " failed. Continuing",
 671                                                 new_stream->key);
 672                                 consumer_stream_free(new_stream);
 673                                 goto end_nosignal;
 674                         }
 675                         stream_pipe = ctx->consumer_metadata_pipe;
 676                 } else {
 677                         ret = consumer_add_data_stream(new_stream);
 678                         if (ret) {
 679                                 ERR("Consumer add stream %" PRIu64 " failed. Continuing",
 680                                                 new_stream->key);
 681                                 consumer_stream_free(new_stream);
 682                                 goto end_nosignal;
 683                         }
 684                         stream_pipe = ctx->consumer_data_pipe;
 685                 }
 686
 687                 /* Vitible to other threads */
 688                 new_stream->globally_visible = 1;
 689
 690                 ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
 691                 if (ret < 0) {
 692                         ERR("Consumer write %s stream to pipe %d",
 693                                         new_stream->metadata_flag ? "metadata" : "data",
 694                                         lttng_pipe_get_writefd(stream_pipe));
 695                         if (new_stream->metadata_flag) {
 696                                 consumer_del_stream_for_metadata(new_stream);
 697                         } else {
 698                                 consumer_del_stream_for_data(new_stream);
 699                         }
 700                         goto end_nosignal;
 701                 }
 702
 703                 DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
 704                                 new_stream->name, fd, new_stream->relayd_stream_id);
 705                 break;
 706         }
 707         case LTTNG_CONSUMER_UPDATE_STREAM:
 708         {
 709                 rcu_read_unlock();
 710                 return -ENOSYS;
 711         }
 712         case LTTNG_CONSUMER_DESTROY_RELAYD:
 713         {
 714                 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
 715                 struct consumer_relayd_sock_pair *relayd;
 716
 717                 DBG("Kernel consumer destroying relayd %" PRIu64, index);
 718
 719                 /* Get relayd reference if exists. */
 720                 relayd = consumer_find_relayd(index);
 721                 if (relayd == NULL) {
 722                         DBG("Unable to find relayd %" PRIu64, index);
 723                         ret_code = LTTNG_ERR_NO_CONSUMER;
 724                 }
 725
 726                 /*
 727                  * Each relayd socket pair has a refcount of stream attached to it
 728                  * which tells if the relayd is still active or not depending on the
 729                  * refcount value.
 730                  *
 731                  * This will set the destroy flag of the relayd object and destroy it
 732                  * if the refcount reaches zero when called.
 733                  *
 734                  * The destroy can happen either here or when a stream fd hangs up.
 735                  */
 736                 if (relayd) {
 737                         consumer_flag_relayd_for_destroy(relayd);
 738                 }
 739
 740                 ret = consumer_send_status_msg(sock, ret_code);
 741                 if (ret < 0) {
 742                         /* Somehow, the session daemon is not responding anymore. */
 743                         goto error_fatal;
 744                 }
 745
 746                 goto end_nosignal;
 747         }
 748         case LTTNG_CONSUMER_DATA_PENDING:
 749         {
 750                 int32_t ret;
 751                 uint64_t id = msg.u.data_pending.session_id;
 752
 753                 DBG("Kernel consumer data pending command for id %" PRIu64, id);
 754
 755                 ret = consumer_data_pending(id);
 756
 757                 /* Send back returned value to session daemon */
 758                 ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
 759                 if (ret < 0) {
 760                         PERROR("send data pending ret code");
 761                         goto error_fatal;
 762                 }
 763
 764                 /*
 765                  * No need to send back a status message since the data pending
 766                  * returned value is the response.
 767                  */
 768                 break;
 769         }
 770         case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
 771         {
 772                 if (msg.u.snapshot_channel.metadata == 1) {
 773                         ret = lttng_kconsumer_snapshot_metadata(msg.u.snapshot_channel.key,
 774                                         msg.u.snapshot_channel.pathname,
 775                                         msg.u.snapshot_channel.relayd_id, ctx);
 776                         if (ret < 0) {
 777                                 ERR("Snapshot metadata failed");
 778                                 ret_code = LTTNG_ERR_KERN_META_FAIL;
 779                         }
 780                 } else {
 781                         ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
 782                                         msg.u.snapshot_channel.pathname,
 783                                         msg.u.snapshot_channel.relayd_id,
 784                                         msg.u.snapshot_channel.max_stream_size,
 785                                         ctx);
 786                         if (ret < 0) {
 787                                 ERR("Snapshot channel failed");
 788                                 ret_code = LTTNG_ERR_KERN_CHAN_FAIL;
 789                         }
 790                 }
 791
 792                 ret = consumer_send_status_msg(sock, ret_code);
 793                 if (ret < 0) {
 794                         /* Somehow, the session daemon is not responding anymore. */
 795                         goto end_nosignal;
 796                 }
 797                 break;
 798         }
 799         case LTTNG_CONSUMER_DESTROY_CHANNEL:
 800         {
 801                 uint64_t key = msg.u.destroy_channel.key;
 802                 struct lttng_consumer_channel *channel;
 803
 804                 channel = consumer_find_channel(key);
 805                 if (!channel) {
 806                         ERR("Kernel consumer destroy channel %" PRIu64 " not found", key);
 807                         ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
 808                 }
 809
 810                 ret = consumer_send_status_msg(sock, ret_code);
 811                 if (ret < 0) {
 812                         /* Somehow, the session daemon is not responding anymore. */
 813                         goto end_nosignal;
 814                 }
 815
 816                 /*
 817                  * This command should ONLY be issued for channel with streams set in
 818                  * no monitor mode.
 819                  */
 820                 assert(!channel->monitor);
 821
 822                 /*
 823                  * The refcount should ALWAYS be 0 in the case of a channel in no
 824                  * monitor mode.
 825                  */
 826                 assert(!uatomic_sub_return(&channel->refcount, 1));
 827
 828                 consumer_del_channel(channel);
 829
 830                 goto end_nosignal;
 831         }
 832         default:
 833                 goto end_nosignal;
 834         }
 835
 836 end_nosignal:
 837         rcu_read_unlock();
 838
 839         /*
 840          * Return 1 to indicate success since the 0 value can be a socket
 841          * shutdown during the recv() or send() call.
 842          */
 843         return 1;
 844
 845 error_fatal:
 846         rcu_read_unlock();
 847         /* This will issue a consumer stop. */
 848         return -1;
 849 }
 850
 851 /*
 852  * Populate index values of a kernel stream. Values are set in big endian order.
 853  *
 854  * Return 0 on success or else a negative value.
 855  */
 856 static int get_index_values(struct lttng_packet_index *index, int infd)
 857 {
 858         int ret;
 859
 860         ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin);
 861         if (ret < 0) {
 862                 PERROR("kernctl_get_timestamp_begin");
 863                 goto error;
 864         }
 865         index->timestamp_begin = htobe64(index->timestamp_begin);
 866
 867         ret = kernctl_get_timestamp_end(infd, &index->timestamp_end);
 868         if (ret < 0) {
 869                 PERROR("kernctl_get_timestamp_end");
 870                 goto error;
 871         }
 872         index->timestamp_end = htobe64(index->timestamp_end);
 873
 874         ret = kernctl_get_events_discarded(infd, &index->events_discarded);
 875         if (ret < 0) {
 876                 PERROR("kernctl_get_events_discarded");
 877                 goto error;
 878         }
 879         index->events_discarded = htobe64(index->events_discarded);
 880
 881         ret = kernctl_get_content_size(infd, &index->content_size);
 882         if (ret < 0) {
 883                 PERROR("kernctl_get_content_size");
 884                 goto error;
 885         }
 886         index->content_size = htobe64(index->content_size);
 887
 888         ret = kernctl_get_packet_size(infd, &index->packet_size);
 889         if (ret < 0) {
 890                 PERROR("kernctl_get_packet_size");
 891                 goto error;
 892         }
 893         index->packet_size = htobe64(index->packet_size);
 894
 895         ret = kernctl_get_stream_id(infd, &index->stream_id);
 896         if (ret < 0) {
 897                 PERROR("kernctl_get_stream_id");
 898                 goto error;
 899         }
 900         index->stream_id = htobe64(index->stream_id);
 901
 902 error:
 903         return ret;
 904 }
 905
 906 /*
 907  * Consume data on a file descriptor and write it on a trace file.
 908  */
 909 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 910                 struct lttng_consumer_local_data *ctx)
 911 {
 912         unsigned long len, subbuf_size, padding;
 913         int err, write_index = 1;
 914         ssize_t ret = 0;
 915         int infd = stream->wait_fd;
 916         struct lttng_packet_index index;
 917
 918         DBG("In read_subbuffer (infd : %d)", infd);
 919
 920         /* Get the next subbuffer */
 921         err = kernctl_get_next_subbuf(infd);
 922         if (err != 0) {
 923                 /*
 924                  * This is a debug message even for single-threaded consumer,
 925                  * because poll() have more relaxed criterions than get subbuf,
 926                  * so get_subbuf may fail for short race windows where poll()
 927                  * would issue wakeups.
 928                  */
 929                 DBG("Reserving sub buffer failed (everything is normal, "
 930                                 "it is due to concurrency)");
 931                 ret = -errno;
 932                 goto end;
 933         }
 934
 935         /* Get the full subbuffer size including padding */
 936         err = kernctl_get_padded_subbuf_size(infd, &len);
 937         if (err != 0) {
 938                 perror("Getting sub-buffer len failed.");
 939                 ret = -errno;
 940                 goto end;
 941         }
 942
 943         if (!stream->metadata_flag) {
 944                 ret = get_index_values(&index, infd);
 945                 if (ret < 0) {
 946                         goto end;
 947                 }
 948         } else {
 949                 write_index = 0;
 950         }
 951
 952         switch (stream->chan->output) {
 953         case CONSUMER_CHANNEL_SPLICE:
 954                 /*
 955                  * XXX: The lttng-modules splice "actor" does not handle copying
 956                  * partial pages hence only using the subbuffer size without the
 957                  * padding makes the splice fail.
 958                  */
 959                 subbuf_size = len;
 960                 padding = 0;
 961
 962                 /* splice the subbuffer to the tracefile */
 963                 ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
 964                                 padding, &index);
 965                 /*
 966                  * XXX: Splice does not support network streaming so the return value
 967                  * is simply checked against subbuf_size and not like the mmap() op.
 968                  */
 969                 if (ret != subbuf_size) {
 970                         /*
 971                          * display the error but continue processing to try
 972                          * to release the subbuffer
 973                          */
 974                         ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
 975                                         ret, subbuf_size);
 976                         write_index = 0;
 977                 }
 978                 break;
 979         case CONSUMER_CHANNEL_MMAP:
 980                 /* Get subbuffer size without padding */
 981                 err = kernctl_get_subbuf_size(infd, &subbuf_size);
 982                 if (err != 0) {
 983                         perror("Getting sub-buffer len failed.");
 984                         ret = -errno;
 985                         goto end;
 986                 }
 987
 988                 /* Make sure the tracer is not gone mad on us! */
 989                 assert(len >= subbuf_size);
 990
 991                 padding = len - subbuf_size;
 992
 993                 /* write the subbuffer to the tracefile */
 994                 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
 995                                 padding, &index);
 996                 /*
 997                  * The mmap operation should write subbuf_size amount of data when
 998                  * network streaming or the full padding (len) size when we are _not_
 999                  * streaming.
1000                  */
1001                 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
1002                                 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
1003                         /*
1004                          * Display the error but continue processing to try to release the
1005                          * subbuffer
1006                          */
1007                         ERR("Error writing to tracefile "
1008                                         "(ret: %zd != len: %lu != subbuf_size: %lu)",
1009                                         ret, len, subbuf_size);
1010                         write_index = 0;
1011                 }
1012                 break;
1013         default:
1014                 ERR("Unknown output method");
1015                 ret = -EPERM;
1016         }
1017
1018         err = kernctl_put_next_subbuf(infd);
1019         if (err != 0) {
1020                 if (errno == EFAULT) {
1021                         perror("Error in unreserving sub buffer\n");
1022                 } else if (errno == EIO) {
1023                         /* Should never happen with newer LTTng versions */
1024                         perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
1025                 }
1026                 ret = -errno;
1027                 goto end;
1028         }
1029
1030         /* Write index if needed. */
1031         if (!write_index) {
1032                 goto end;
1033         }
1034
1035         err = consumer_stream_write_index(stream, &index);
1036         if (err < 0) {
1037                 goto end;
1038         }
1039
1040 end:
1041         return ret;
1042 }
1043
1044 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1045 {
1046         int ret;
1047
1048         assert(stream);
1049
1050         /*
1051          * Don't create anything if this is set for streaming or should not be
1052          * monitored.
1053          */
1054         if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
1055                 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
1056                                 stream->chan->tracefile_size, stream->tracefile_count_current,
1057                                 stream->uid, stream->gid, NULL);
1058                 if (ret < 0) {
1059                         goto error;
1060                 }
1061                 stream->out_fd = ret;
1062                 stream->tracefile_size_current = 0;
1063
1064                 if (!stream->metadata_flag) {
1065                         ret = index_create_file(stream->chan->pathname,
1066                                         stream->name, stream->uid, stream->gid,
1067                                         stream->chan->tracefile_size,
1068                                         stream->tracefile_count_current);
1069                         if (ret < 0) {
1070                                 goto error;
1071                         }
1072                         stream->index_fd = ret;
1073                 }
1074         }
1075
1076         if (stream->output == LTTNG_EVENT_MMAP) {
1077                 /* get the len of the mmap region */
1078                 unsigned long mmap_len;
1079
1080                 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
1081                 if (ret != 0) {
1082                         PERROR("kernctl_get_mmap_len");
1083                         ret = -errno;
1084                         goto error_close_fd;
1085                 }
1086                 stream->mmap_len = (size_t) mmap_len;
1087
1088                 stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
1089                                 MAP_PRIVATE, stream->wait_fd, 0);
1090                 if (stream->mmap_base == MAP_FAILED) {
1091                         PERROR("Error mmaping");
1092                         ret = -1;
1093                         goto error_close_fd;
1094                 }
1095         }
1096
1097         /* we return 0 to let the library handle the FD internally */
1098         return 0;
1099
1100 error_close_fd:
1101         if (stream->out_fd >= 0) {
1102                 int err;
1103
1104                 err = close(stream->out_fd);
1105                 assert(!err);
1106                 stream->out_fd = -1;
1107         }
1108 error:
1109         return ret;
1110 }
1111
1112 /*
1113  * Check if data is still being extracted from the buffers for a specific
1114  * stream. Consumer data lock MUST be acquired before calling this function
1115  * and the stream lock.
1116  *
1117  * Return 1 if the traced data are still getting read else 0 meaning that the
1118  * data is available for trace viewer reading.
1119  */
1120 int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
1121 {
1122         int ret;
1123
1124         assert(stream);
1125
1126         if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1127                 ret = 0;
1128                 goto end;
1129         }
1130
1131         ret = kernctl_get_next_subbuf(stream->wait_fd);
1132         if (ret == 0) {
1133                 /* There is still data so let's put back this subbuffer. */
1134                 ret = kernctl_put_subbuf(stream->wait_fd);
1135                 assert(ret == 0);
1136                 ret = 1;   /* Data is pending */
1137                 goto end;
1138         }
1139
1140         /* Data is NOT pending and ready to be read. */
1141         ret = 0;
1142
1143 end:
1144         return ret;
1145 }