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 <unistd.h>
  29 #include <sys/stat.h>
  30
  31 #include <common/common.h>
  32 #include <common/kernel-ctl/kernel-ctl.h>
  33 #include <common/sessiond-comm/sessiond-comm.h>
  34 #include <common/compat/fcntl.h>
  35
  36 #include "kernel-consumer.h"
  37
  38 extern struct lttng_consumer_global_data consumer_data;
  39 extern int consumer_poll_timeout;
  40 extern volatile int consumer_quit;
  41
  42 /*
  43  * Mmap the ring buffer, read it and write the data to the tracefile.
  44  *
  45  * Returns the number of bytes written
  46  */
  47 ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
  48                 struct lttng_consumer_local_data *ctx,
  49                 struct lttng_consumer_stream *stream, unsigned long len)
  50 {
  51         unsigned long mmap_offset;
  52         ssize_t ret = 0, written = 0;
  53         off_t orig_offset = stream->out_fd_offset;
  54         int fd = stream->wait_fd;
  55         int outfd = stream->out_fd;
  56
  57         /* get the offset inside the fd to mmap */
  58         ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
  59         if (ret != 0) {
  60                 errno = -ret;
  61                 perror("kernctl_get_mmap_read_offset");
  62                 written = ret;
  63                 goto end;
  64         }
  65
  66         while (len > 0) {
  67                 ret = write(outfd, stream->mmap_base + mmap_offset, len);
  68                 if (ret < 0) {
  69                         if (errno == EINTR) {
  70                                 /* restart the interrupted system call */
  71                                 continue;
  72                         } else {
  73                                 perror("Error in file write");
  74                                 if (written == 0) {
  75                                         written = ret;
  76                                 }
  77                                 goto end;
  78                         }
  79                 } else if (ret > len) {
  80                         perror("Error in file write");
  81                         written += ret;
  82                         goto end;
  83                 } else {
  84                         len -= ret;
  85                         mmap_offset += ret;
  86                 }
  87                 /* This won't block, but will start writeout asynchronously */
  88                 lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
  89                                 SYNC_FILE_RANGE_WRITE);
  90                 stream->out_fd_offset += ret;
  91                 written += ret;
  92         }
  93         lttng_consumer_sync_trace_file(stream, orig_offset);
  94 end:
  95         return written;
  96 }
  97
  98 /*
  99  * Splice the data from the ring buffer to the tracefile.
 100  *
 101  * Returns the number of bytes spliced.
 102  */
 103 ssize_t lttng_kconsumer_on_read_subbuffer_splice(
 104                 struct lttng_consumer_local_data *ctx,
 105                 struct lttng_consumer_stream *stream, unsigned long len)
 106 {
 107         ssize_t ret = 0, written = 0;
 108         loff_t offset = 0;
 109         off_t orig_offset = stream->out_fd_offset;
 110         int fd = stream->wait_fd;
 111         int outfd = stream->out_fd;
 112
 113         while (len > 0) {
 114                 DBG("splice chan to pipe offset %lu (fd : %d)",
 115                                 (unsigned long)offset, fd);
 116                 ret = splice(fd, &offset, ctx->consumer_thread_pipe[1], NULL, len,
 117                                 SPLICE_F_MOVE | SPLICE_F_MORE);
 118                 DBG("splice chan to pipe ret %zd", ret);
 119                 if (ret < 0) {
 120                         perror("Error in relay splice");
 121                         if (written == 0) {
 122                                 written = ret;
 123                         }
 124                         ret = errno;
 125                         goto splice_error;
 126                 }
 127
 128                 ret = splice(ctx->consumer_thread_pipe[0], NULL, outfd, NULL, ret,
 129                                 SPLICE_F_MOVE | SPLICE_F_MORE);
 130                 DBG("splice pipe to file %zd", ret);
 131                 if (ret < 0) {
 132                         perror("Error in file splice");
 133                         if (written == 0) {
 134                                 written = ret;
 135                         }
 136                         ret = errno;
 137                         goto splice_error;
 138                 }
 139                 if (ret > len) {
 140                         errno = EINVAL;
 141                         perror("Wrote more data than requested");
 142                         written += ret;
 143                         ret = errno;
 144                         goto splice_error;
 145                 }
 146                 len -= ret;
 147                 /* This won't block, but will start writeout asynchronously */
 148                 lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
 149                                 SYNC_FILE_RANGE_WRITE);
 150                 stream->out_fd_offset += ret;
 151                 written += ret;
 152         }
 153         lttng_consumer_sync_trace_file(stream, orig_offset);
 154
 155         goto end;
 156
 157 splice_error:
 158         /* send the appropriate error description to sessiond */
 159         switch (ret) {
 160         case EBADF:
 161                 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EBADF);
 162                 break;
 163         case EINVAL:
 164                 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EINVAL);
 165                 break;
 166         case ENOMEM:
 167                 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ENOMEM);
 168                 break;
 169         case ESPIPE:
 170                 lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ESPIPE);
 171                 break;
 172         }
 173
 174 end:
 175         return written;
 176 }
 177
 178 /*
 179  * Take a snapshot for a specific fd
 180  *
 181  * Returns 0 on success, < 0 on error
 182  */
 183 int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
 184                 struct lttng_consumer_stream *stream)
 185 {
 186         int ret = 0;
 187         int infd = stream->wait_fd;
 188
 189         ret = kernctl_snapshot(infd);
 190         if (ret != 0) {
 191                 errno = -ret;
 192                 perror("Getting sub-buffer snapshot.");
 193         }
 194
 195         return ret;
 196 }
 197
 198 /*
 199  * Get the produced position
 200  *
 201  * Returns 0 on success, < 0 on error
 202  */
 203 int lttng_kconsumer_get_produced_snapshot(
 204                 struct lttng_consumer_local_data *ctx,
 205                 struct lttng_consumer_stream *stream,
 206                 unsigned long *pos)
 207 {
 208         int ret;
 209         int infd = stream->wait_fd;
 210
 211         ret = kernctl_snapshot_get_produced(infd, pos);
 212         if (ret != 0) {
 213                 errno = -ret;
 214                 perror("kernctl_snapshot_get_produced");
 215         }
 216
 217         return ret;
 218 }
 219
 220 int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 221                 int sock, struct pollfd *consumer_sockpoll)
 222 {
 223         ssize_t ret;
 224         struct lttcomm_consumer_msg msg;
 225
 226         ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
 227         if (ret != sizeof(msg)) {
 228                 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
 229                 return ret;
 230         }
 231         if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
 232                 return -ENOENT;
 233         }
 234
 235         switch (msg.cmd_type) {
 236         case LTTNG_CONSUMER_ADD_CHANNEL:
 237         {
 238                 struct lttng_consumer_channel *new_channel;
 239
 240                 DBG("consumer_add_channel %d", msg.u.channel.channel_key);
 241                 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
 242                                 -1, -1,
 243                                 msg.u.channel.mmap_len,
 244                                 msg.u.channel.max_sb_size);
 245                 if (new_channel == NULL) {
 246                         lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
 247                         goto end_nosignal;
 248                 }
 249                 if (ctx->on_recv_channel != NULL) {
 250                         ret = ctx->on_recv_channel(new_channel);
 251                         if (ret == 0) {
 252                                 consumer_add_channel(new_channel);
 253                         } else if (ret < 0) {
 254                                 goto end_nosignal;
 255                         }
 256                 } else {
 257                         consumer_add_channel(new_channel);
 258                 }
 259                 goto end_nosignal;
 260         }
 261         case LTTNG_CONSUMER_ADD_STREAM:
 262         {
 263                 struct lttng_consumer_stream *new_stream;
 264                 int fd;
 265
 266                 /* block */
 267                 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 268                         return -EINTR;
 269                 }
 270                 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 271                 if (ret != sizeof(fd)) {
 272                         lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
 273                         return ret;
 274                 }
 275
 276                 DBG("consumer_add_stream %s (%d)", msg.u.stream.path_name,
 277                         fd);
 278                 new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
 279                                 msg.u.stream.stream_key,
 280                                 fd, fd,
 281                                 msg.u.stream.state,
 282                                 msg.u.stream.mmap_len,
 283                                 msg.u.stream.output,
 284                                 msg.u.stream.path_name,
 285                                 msg.u.stream.uid,
 286                                 msg.u.stream.gid);
 287                 if (new_stream == NULL) {
 288                         lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
 289                         goto end;
 290                 }
 291                 if (ctx->on_recv_stream != NULL) {
 292                         ret = ctx->on_recv_stream(new_stream);
 293                         if (ret == 0) {
 294                                 consumer_add_stream(new_stream);
 295                         } else if (ret < 0) {
 296                                 goto end;
 297                         }
 298                 } else {
 299                         consumer_add_stream(new_stream);
 300                 }
 301                 break;
 302         }
 303         case LTTNG_CONSUMER_UPDATE_STREAM:
 304         {
 305                 if (ctx->on_update_stream != NULL) {
 306                         ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
 307                         if (ret == 0) {
 308                                 consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
 309                         } else if (ret < 0) {
 310                                 goto end;
 311                         }
 312                 } else {
 313                         consumer_change_stream_state(msg.u.stream.stream_key,
 314                                 msg.u.stream.state);
 315                 }
 316                 break;
 317         }
 318         default:
 319                 break;
 320         }
 321 end:
 322         /*
 323          * Wake-up the other end by writing a null byte in the pipe
 324          * (non-blocking). Important note: Because writing into the
 325          * pipe is non-blocking (and therefore we allow dropping wakeup
 326          * data, as long as there is wakeup data present in the pipe
 327          * buffer to wake up the other end), the other end should
 328          * perform the following sequence for waiting:
 329          * 1) empty the pipe (reads).
 330          * 2) perform update operation.
 331          * 3) wait on the pipe (poll).
 332          */
 333         do {
 334                 ret = write(ctx->consumer_poll_pipe[1], "", 1);
 335         } while (ret == -1UL && errno == EINTR);
 336 end_nosignal:
 337         return 0;
 338 }
 339
 340 /*
 341  * Consume data on a file descriptor and write it on a trace file.
 342  */
 343 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 344                 struct lttng_consumer_local_data *ctx)
 345 {
 346         unsigned long len;
 347         int err;
 348         ssize_t ret = 0;
 349         int infd = stream->wait_fd;
 350
 351         DBG("In read_subbuffer (infd : %d)", infd);
 352         /* Get the next subbuffer */
 353         err = kernctl_get_next_subbuf(infd);
 354         if (err != 0) {
 355                 /*
 356                  * This is a debug message even for single-threaded consumer,
 357                  * because poll() have more relaxed criterions than get subbuf,
 358                  * so get_subbuf may fail for short race windows where poll()
 359                  * would issue wakeups.
 360                  */
 361                 DBG("Reserving sub buffer failed (everything is normal, "
 362                                 "it is due to concurrency)");
 363                 goto end;
 364         }
 365
 366         switch (stream->output) {
 367                 case LTTNG_EVENT_SPLICE:
 368                         /* read the whole subbuffer */
 369                         err = kernctl_get_padded_subbuf_size(infd, &len);
 370                         if (err != 0) {
 371                                 errno = -ret;
 372                                 perror("Getting sub-buffer len failed.");
 373                                 goto end;
 374                         }
 375
 376                         /* splice the subbuffer to the tracefile */
 377                         ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
 378                         if (ret != len) {
 379                                 /*
 380                                  * display the error but continue processing to try
 381                                  * to release the subbuffer
 382                                  */
 383                                 ERR("Error splicing to tracefile");
 384                         }
 385
 386                         break;
 387                 case LTTNG_EVENT_MMAP:
 388                         /* read the used subbuffer size */
 389                         err = kernctl_get_padded_subbuf_size(infd, &len);
 390                         if (err != 0) {
 391                                 errno = -ret;
 392                                 perror("Getting sub-buffer len failed.");
 393                                 goto end;
 394                         }
 395                         /* write the subbuffer to the tracefile */
 396                         ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
 397                         if (ret != len) {
 398                                 /*
 399                                  * display the error but continue processing to try
 400                                  * to release the subbuffer
 401                                  */
 402                                 ERR("Error writing to tracefile");
 403                         }
 404                         break;
 405                 default:
 406                         ERR("Unknown output method");
 407                         ret = -1;
 408         }
 409
 410         err = kernctl_put_next_subbuf(infd);
 411         if (err != 0) {
 412                 errno = -ret;
 413                 if (errno == EFAULT) {
 414                         perror("Error in unreserving sub buffer\n");
 415                 } else if (errno == EIO) {
 416                         /* Should never happen with newer LTTng versions */
 417                         perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
 418                 }
 419                 goto end;
 420         }
 421
 422 end:
 423         return ret;
 424 }
 425
 426 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 427 {
 428         int ret;
 429
 430         /* Opening the tracefile in write mode */
 431         if (stream->path_name != NULL) {
 432                 ret = run_as_open(stream->path_name,
 433                                 O_WRONLY|O_CREAT|O_TRUNC,
 434                                 S_IRWXU|S_IRWXG|S_IRWXO,
 435                                 stream->uid, stream->gid);
 436                 if (ret < 0) {
 437                         ERR("Opening %s", stream->path_name);
 438                         perror("open");
 439                         goto error;
 440                 }
 441                 stream->out_fd = ret;
 442         }
 443
 444         if (stream->output == LTTNG_EVENT_MMAP) {
 445                 /* get the len of the mmap region */
 446                 unsigned long mmap_len;
 447
 448                 ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
 449                 if (ret != 0) {
 450                         errno = -ret;
 451                         perror("kernctl_get_mmap_len");
 452                         goto error_close_fd;
 453                 }
 454                 stream->mmap_len = (size_t) mmap_len;
 455
 456                 stream->mmap_base = mmap(NULL, stream->mmap_len,
 457                                 PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
 458                 if (stream->mmap_base == MAP_FAILED) {
 459                         perror("Error mmaping");
 460                         ret = -1;
 461                         goto error_close_fd;
 462                 }
 463         }
 464
 465         /* we return 0 to let the library handle the FD internally */
 466         return 0;
 467
 468 error_close_fd:
 469         {
 470                 int err;
 471
 472                 err = close(stream->out_fd);
 473                 assert(!err);
 474         }
 475 error:
 476         return ret;
 477 }
 478