src/common/ust-consumer/ust-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 <lttng/ust-ctl.h>
  22 #include <poll.h>
  23 #include <pthread.h>
  24 #include <stdlib.h>
  25 #include <string.h>
  26 #include <sys/mman.h>
  27 #include <sys/socket.h>
  28 #include <sys/stat.h>
  29 #include <sys/types.h>
  30 #include <inttypes.h>
  31 #include <unistd.h>
  32
  33 #include <common/common.h>
  34 #include <common/sessiond-comm/sessiond-comm.h>
  35 #include <common/relayd/relayd.h>
  36 #include <common/compat/fcntl.h>
  37
  38 #include "ust-consumer.h"
  39
  40 extern struct lttng_consumer_global_data consumer_data;
  41 extern int consumer_poll_timeout;
  42 extern volatile int consumer_quit;
  43
  44 /*
  45  * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
  46  * compiled out, we isolate it in this library.
  47  */
  48 int lttng_ustctl_get_mmap_read_offset(struct lttng_ust_shm_handle *handle,
  49                 struct lttng_ust_lib_ring_buffer *buf, unsigned long *off)
  50 {
  51         return ustctl_get_mmap_read_offset(handle, buf, off);
  52 };
  53
  54 /*
  55  * Take a snapshot for a specific fd
  56  *
  57  * Returns 0 on success, < 0 on error
  58  */
  59 int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
  60                 struct lttng_consumer_stream *stream)
  61 {
  62         int ret = 0;
  63
  64         ret = ustctl_snapshot(stream->chan->handle, stream->buf);
  65         if (ret != 0) {
  66                 errno = -ret;
  67                 PERROR("Getting sub-buffer snapshot.");
  68         }
  69
  70         return ret;
  71 }
  72
  73 /*
  74  * Get the produced position
  75  *
  76  * Returns 0 on success, < 0 on error
  77  */
  78 int lttng_ustconsumer_get_produced_snapshot(
  79                 struct lttng_consumer_local_data *ctx,
  80                 struct lttng_consumer_stream *stream,
  81                 unsigned long *pos)
  82 {
  83         int ret;
  84
  85         ret = ustctl_snapshot_get_produced(stream->chan->handle,
  86                         stream->buf, pos);
  87         if (ret != 0) {
  88                 errno = -ret;
  89                 PERROR("ustctl_snapshot_get_produced");
  90         }
  91
  92         return ret;
  93 }
  94
  95 /*
  96  * Receive command from session daemon and process it.
  97  *
  98  * Return 1 on success else a negative value or 0.
  99  */
 100 int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 101                 int sock, struct pollfd *consumer_sockpoll)
 102 {
 103         ssize_t ret;
 104         enum lttng_error_code ret_code = LTTNG_OK;
 105         struct lttcomm_consumer_msg msg;
 106
 107         ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
 108         if (ret != sizeof(msg)) {
 109                 DBG("Consumer received unexpected message size %zd (expects %zu)",
 110                         ret, sizeof(msg));
 111                 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 112                 /*
 113                  * The ret value might 0 meaning an orderly shutdown but this is ok
 114                  * since the caller handles this.
 115                  */
 116                 return ret;
 117         }
 118         if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
 119                 /*
 120                  * Notify the session daemon that the command is completed.
 121                  *
 122                  * On transport layer error, the function call will print an error
 123                  * message so handling the returned code is a bit useless since we
 124                  * return an error code anyway.
 125                  */
 126                 (void) consumer_send_status_msg(sock, ret_code);
 127                 return -ENOENT;
 128         }
 129
 130         /* relayd needs RCU read-side lock */
 131         rcu_read_lock();
 132
 133         switch (msg.cmd_type) {
 134         case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
 135         {
 136                 /* Session daemon status message are handled in the following call. */
 137                 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
 138                                 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
 139                                 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id);
 140                 goto end_nosignal;
 141         }
 142         case LTTNG_CONSUMER_ADD_CHANNEL:
 143         {
 144                 struct lttng_consumer_channel *new_channel;
 145                 int fds[1];
 146                 size_t nb_fd = 1;
 147
 148                 DBG("UST Consumer adding channel");
 149
 150                 /* First send a status message before receiving the fds. */
 151                 ret = consumer_send_status_msg(sock, ret_code);
 152                 if (ret < 0) {
 153                         /* Somehow, the session daemon is not responding anymore. */
 154                         goto end_nosignal;
 155                 }
 156
 157                 /* block */
 158                 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 159                         rcu_read_unlock();
 160                         return -EINTR;
 161                 }
 162                 ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
 163                 if (ret != sizeof(fds)) {
 164                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 165                         rcu_read_unlock();
 166                         /*
 167                          * The ret value might 0 meaning an orderly shutdown but this is ok
 168                          * since the caller handles this.
 169                          */
 170                         return ret;
 171                 }
 172
 173                 /*
 174                  * Send status code to session daemon only if the recv works. If the
 175                  * above recv() failed, the session daemon is notified through the
 176                  * error socket and the teardown is eventually done.
 177                  */
 178                 ret = consumer_send_status_msg(sock, ret_code);
 179                 if (ret < 0) {
 180                         /* Somehow, the session daemon is not responding anymore. */
 181                         goto end_nosignal;
 182                 }
 183
 184                 DBG("consumer_add_channel %d", msg.u.channel.channel_key);
 185
 186                 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
 187                                 fds[0], -1,
 188                                 msg.u.channel.mmap_len,
 189                                 msg.u.channel.max_sb_size,
 190                                 msg.u.channel.nb_init_streams);
 191                 if (new_channel == NULL) {
 192                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 193                         goto end_nosignal;
 194                 }
 195                 if (ctx->on_recv_channel != NULL) {
 196                         ret = ctx->on_recv_channel(new_channel);
 197                         if (ret == 0) {
 198                                 consumer_add_channel(new_channel);
 199                         } else if (ret < 0) {
 200                                 goto end_nosignal;
 201                         }
 202                 } else {
 203                         consumer_add_channel(new_channel);
 204                 }
 205                 goto end_nosignal;
 206         }
 207         case LTTNG_CONSUMER_ADD_STREAM:
 208         {
 209                 struct lttng_consumer_stream *new_stream;
 210                 int fds[2], stream_pipe;
 211                 size_t nb_fd = 2;
 212                 struct consumer_relayd_sock_pair *relayd = NULL;
 213                 int alloc_ret = 0;
 214
 215                 DBG("UST Consumer adding stream");
 216
 217                 /* First send a status message before receiving the fds. */
 218                 ret = consumer_send_status_msg(sock, ret_code);
 219                 if (ret < 0) {
 220                         /* Somehow, the session daemon is not responding anymore. */
 221                         goto end_nosignal;
 222                 }
 223
 224                 /* block */
 225                 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 226                         rcu_read_unlock();
 227                         return -EINTR;
 228                 }
 229                 ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
 230                 if (ret != sizeof(fds)) {
 231                         lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 232                         rcu_read_unlock();
 233                         /*
 234                          * The ret value might 0 meaning an orderly shutdown but this is ok
 235                          * since the caller handles this.
 236                          */
 237                         return ret;
 238                 }
 239
 240                 /*
 241                  * Send status code to session daemon only if the recv works. If the
 242                  * above recv() failed, the session daemon is notified through the
 243                  * error socket and the teardown is eventually done.
 244                  */
 245                 ret = consumer_send_status_msg(sock, ret_code);
 246                 if (ret < 0) {
 247                         /* Somehow, the session daemon is not responding anymore. */
 248                         goto end_nosignal;
 249                 }
 250
 251                 DBG("Consumer command ADD_STREAM chan %d stream %d",
 252                                 msg.u.stream.channel_key, msg.u.stream.stream_key);
 253
 254                 assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
 255                 new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
 256                                 msg.u.stream.stream_key,
 257                                 fds[0], fds[1],
 258                                 msg.u.stream.state,
 259                                 msg.u.stream.mmap_len,
 260                                 msg.u.stream.output,
 261                                 msg.u.stream.path_name,
 262                                 msg.u.stream.uid,
 263                                 msg.u.stream.gid,
 264                                 msg.u.stream.net_index,
 265                                 msg.u.stream.metadata_flag,
 266                                 msg.u.stream.session_id,
 267                                 &alloc_ret);
 268                 if (new_stream == NULL) {
 269                         switch (alloc_ret) {
 270                         case -ENOMEM:
 271                         case -EINVAL:
 272                         default:
 273                                 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 274                                 break;
 275                         case -ENOENT:
 276                                 /*
 277                                  * We could not find the channel. Can happen if cpu hotplug
 278                                  * happens while tearing down.
 279                                  */
 280                                 DBG3("Could not find channel");
 281                                 break;
 282                         }
 283                         goto end_nosignal;
 284                 }
 285
 286                 /* The stream is not metadata. Get relayd reference if exists. */
 287                 relayd = consumer_find_relayd(msg.u.stream.net_index);
 288                 if (relayd != NULL) {
 289                         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 290                         /* Add stream on the relayd */
 291                         ret = relayd_add_stream(&relayd->control_sock,
 292                                         msg.u.stream.name, msg.u.stream.path_name,
 293                                         &new_stream->relayd_stream_id);
 294                         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 295                         if (ret < 0) {
 296                                 consumer_del_stream(new_stream, NULL);
 297                                 goto end_nosignal;
 298                         }
 299                 } else if (msg.u.stream.net_index != -1) {
 300                         ERR("Network sequence index %d unknown. Not adding stream.",
 301                                         msg.u.stream.net_index);
 302                         consumer_del_stream(new_stream, NULL);
 303                         goto end_nosignal;
 304                 }
 305
 306                 /* Do actions once stream has been received. */
 307                 if (ctx->on_recv_stream) {
 308                         ret = ctx->on_recv_stream(new_stream);
 309                         if (ret < 0) {
 310                                 consumer_del_stream(new_stream, NULL);
 311                                 goto end_nosignal;
 312                         }
 313                 }
 314
 315                 /* Get the right pipe where the stream will be sent. */
 316                 if (new_stream->metadata_flag) {
 317                         stream_pipe = ctx->consumer_metadata_pipe[1];
 318                 } else {
 319                         stream_pipe = ctx->consumer_data_pipe[1];
 320                 }
 321
 322                 do {
 323                         ret = write(stream_pipe, &new_stream, sizeof(new_stream));
 324                 } while (ret < 0 && errno == EINTR);
 325                 if (ret < 0) {
 326                         PERROR("Consumer write %s stream to pipe %d",
 327                                         new_stream->metadata_flag ? "metadata" : "data",
 328                                         stream_pipe);
 329                         consumer_del_stream(new_stream, NULL);
 330                         goto end_nosignal;
 331                 }
 332
 333                 DBG("UST consumer ADD_STREAM %s (%d,%d) with relayd id %" PRIu64,
 334                                 msg.u.stream.path_name, fds[0], fds[1],
 335                                 new_stream->relayd_stream_id);
 336                 break;
 337         }
 338         case LTTNG_CONSUMER_DESTROY_RELAYD:
 339         {
 340                 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
 341                 struct consumer_relayd_sock_pair *relayd;
 342
 343                 DBG("UST consumer destroying relayd %" PRIu64, index);
 344
 345                 /* Get relayd reference if exists. */
 346                 relayd = consumer_find_relayd(index);
 347                 if (relayd == NULL) {
 348                         DBG("Unable to find relayd %" PRIu64, index);
 349                         ret_code = LTTNG_ERR_NO_CONSUMER;
 350                 }
 351
 352                 /*
 353                  * Each relayd socket pair has a refcount of stream attached to it
 354                  * which tells if the relayd is still active or not depending on the
 355                  * refcount value.
 356                  *
 357                  * This will set the destroy flag of the relayd object and destroy it
 358                  * if the refcount reaches zero when called.
 359                  *
 360                  * The destroy can happen either here or when a stream fd hangs up.
 361                  */
 362                 if (relayd) {
 363                         consumer_flag_relayd_for_destroy(relayd);
 364                 }
 365
 366                 ret = consumer_send_status_msg(sock, ret_code);
 367                 if (ret < 0) {
 368                         /* Somehow, the session daemon is not responding anymore. */
 369                         goto end_nosignal;
 370                 }
 371
 372                 goto end_nosignal;
 373         }
 374         case LTTNG_CONSUMER_UPDATE_STREAM:
 375         {
 376                 rcu_read_unlock();
 377                 return -ENOSYS;
 378         }
 379         case LTTNG_CONSUMER_DATA_PENDING:
 380         {
 381                 int ret, is_data_pending;
 382                 uint64_t id = msg.u.data_pending.session_id;
 383
 384                 DBG("UST consumer data pending command for id %" PRIu64, id);
 385
 386                 is_data_pending = consumer_data_pending(id);
 387
 388                 /* Send back returned value to session daemon */
 389                 ret = lttcomm_send_unix_sock(sock, &is_data_pending,
 390                                 sizeof(is_data_pending));
 391                 if (ret < 0) {
 392                         DBG("Error when sending the data pending ret code: %d", ret);
 393                 }
 394
 395                 /*
 396                  * No need to send back a status message since the data pending
 397                  * returned value is the response.
 398                  */
 399                 break;
 400         }
 401         default:
 402                 break;
 403         }
 404
 405 end_nosignal:
 406         rcu_read_unlock();
 407
 408         /*
 409          * Return 1 to indicate success since the 0 value can be a socket
 410          * shutdown during the recv() or send() call.
 411          */
 412         return 1;
 413 }
 414
 415 int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
 416 {
 417         struct lttng_ust_object_data obj;
 418
 419         obj.handle = -1;
 420         obj.shm_fd = chan->shm_fd;
 421         obj.wait_fd = chan->wait_fd;
 422         obj.memory_map_size = chan->mmap_len;
 423         chan->handle = ustctl_map_channel(&obj);
 424         if (!chan->handle) {
 425                 return -ENOMEM;
 426         }
 427         chan->wait_fd_is_copy = 1;
 428         chan->shm_fd = -1;
 429
 430         return 0;
 431 }
 432
 433 void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
 434 {
 435         ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
 436         stream->hangup_flush_done = 1;
 437 }
 438
 439 void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
 440 {
 441         ustctl_unmap_channel(chan->handle);
 442 }
 443
 444 int lttng_ustconsumer_add_stream(struct lttng_consumer_stream *stream)
 445 {
 446         struct lttng_ust_object_data obj;
 447         int ret;
 448
 449         obj.handle = -1;
 450         obj.shm_fd = stream->shm_fd;
 451         obj.wait_fd = stream->wait_fd;
 452         obj.memory_map_size = stream->mmap_len;
 453         ret = ustctl_add_stream(stream->chan->handle, &obj);
 454         if (ret) {
 455                 ERR("UST ctl add_stream failed with ret %d", ret);
 456                 goto error;
 457         }
 458
 459         stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
 460         if (!stream->buf) {
 461                 ERR("UST ctl open_stream_read failed");
 462                 ret = -EBUSY;
 463                 goto error;
 464         }
 465
 466         /* ustctl_open_stream_read has closed the shm fd. */
 467         stream->wait_fd_is_copy = 1;
 468         stream->shm_fd = -1;
 469
 470         stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
 471         if (!stream->mmap_base) {
 472                 ERR("UST ctl get_mmap_base failed");
 473                 ret = -EINVAL;
 474                 goto mmap_error;
 475         }
 476
 477         return 0;
 478
 479 mmap_error:
 480         ustctl_close_stream_read(stream->chan->handle, stream->buf);
 481 error:
 482         return ret;
 483 }
 484
 485 void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
 486 {
 487         ustctl_close_stream_read(stream->chan->handle, stream->buf);
 488 }
 489
 490
 491 int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 492                 struct lttng_consumer_local_data *ctx)
 493 {
 494         unsigned long len, subbuf_size, padding;
 495         int err;
 496         long ret = 0;
 497         struct lttng_ust_shm_handle *handle;
 498         struct lttng_ust_lib_ring_buffer *buf;
 499         char dummy;
 500
 501         DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
 502                 stream->wait_fd, stream->key);
 503
 504         /* We can consume the 1 byte written into the wait_fd by UST */
 505         if (!stream->hangup_flush_done) {
 506                 ssize_t readlen;
 507
 508                 do {
 509                         readlen = read(stream->wait_fd, &dummy, 1);
 510                 } while (readlen == -1 && errno == EINTR);
 511                 if (readlen == -1) {
 512                         ret = readlen;
 513                         goto end;
 514                 }
 515         }
 516
 517         buf = stream->buf;
 518         handle = stream->chan->handle;
 519         /* Get the next subbuffer */
 520         err = ustctl_get_next_subbuf(handle, buf);
 521         if (err != 0) {
 522                 ret = err;      /* ustctl_get_next_subbuf returns negative, caller expect positive. */
 523                 /*
 524                  * This is a debug message even for single-threaded consumer,
 525                  * because poll() have more relaxed criterions than get subbuf,
 526                  * so get_subbuf may fail for short race windows where poll()
 527                  * would issue wakeups.
 528                  */
 529                 DBG("Reserving sub buffer failed (everything is normal, "
 530                                 "it is due to concurrency)");
 531                 goto end;
 532         }
 533         assert(stream->output == LTTNG_EVENT_MMAP);
 534         /* Get the full padded subbuffer size */
 535         err = ustctl_get_padded_subbuf_size(handle, buf, &len);
 536         assert(err == 0);
 537
 538         /* Get subbuffer data size (without padding) */
 539         err = ustctl_get_subbuf_size(handle, buf, &subbuf_size);
 540         assert(err == 0);
 541
 542         /* Make sure we don't get a subbuffer size bigger than the padded */
 543         assert(len >= subbuf_size);
 544
 545         padding = len - subbuf_size;
 546         /* write the subbuffer to the tracefile */
 547         ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding);
 548         /*
 549          * The mmap operation should write subbuf_size amount of data when network
 550          * streaming or the full padding (len) size when we are _not_ streaming.
 551          */
 552         if ((ret != subbuf_size && stream->net_seq_idx != -1) ||
 553                         (ret != len && stream->net_seq_idx == -1)) {
 554                 /*
 555                  * Display the error but continue processing to try to release the
 556                  * subbuffer. This is a DBG statement since any unexpected kill or
 557                  * signal, the application gets unregistered, relayd gets closed or
 558                  * anything that affects the buffer lifetime will trigger this error.
 559                  * So, for the sake of the user, don't print this error since it can
 560                  * happen and it is OK with the code flow.
 561                  */
 562                 DBG("Error writing to tracefile "
 563                                 "(ret: %zd != len: %lu != subbuf_size: %lu)",
 564                                 ret, len, subbuf_size);
 565         }
 566         err = ustctl_put_next_subbuf(handle, buf);
 567         assert(err == 0);
 568 end:
 569         return ret;
 570 }
 571
 572 int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 573 {
 574         int ret;
 575
 576         /* Opening the tracefile in write mode */
 577         if (stream->path_name != NULL && stream->net_seq_idx == -1) {
 578                 ret = run_as_open(stream->path_name,
 579                                 O_WRONLY|O_CREAT|O_TRUNC,
 580                                 S_IRWXU|S_IRWXG|S_IRWXO,
 581                                 stream->uid, stream->gid);
 582                 if (ret < 0) {
 583                         ERR("Opening %s", stream->path_name);
 584                         PERROR("open");
 585                         goto error;
 586                 }
 587                 stream->out_fd = ret;
 588         }
 589
 590         ret = lttng_ustconsumer_add_stream(stream);
 591         if (ret) {
 592                 consumer_del_stream(stream, NULL);
 593                 ret = -1;
 594                 goto error;
 595         }
 596
 597         /* we return 0 to let the library handle the FD internally */
 598         return 0;
 599
 600 error:
 601         return ret;
 602 }
 603
 604 /*
 605  * Check if data is still being extracted from the buffers for a specific
 606  * stream. Consumer data lock MUST be acquired before calling this function
 607  * and the stream lock.
 608  *
 609  * Return 1 if the traced data are still getting read else 0 meaning that the
 610  * data is available for trace viewer reading.
 611  */
 612 int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
 613 {
 614         int ret;
 615
 616         assert(stream);
 617
 618         DBG("UST consumer checking data pending");
 619
 620         ret = ustctl_get_next_subbuf(stream->chan->handle, stream->buf);
 621         if (ret == 0) {
 622                 /* There is still data so let's put back this subbuffer. */
 623                 ret = ustctl_put_subbuf(stream->chan->handle, stream->buf);
 624                 assert(ret == 0);
 625                 ret = 1;  /* Data is pending */
 626                 goto end;
 627         }
 628
 629         /* Data is NOT pending so ready to be read. */
 630         ret = 0;
 631
 632 end:
 633         return ret;
 634 }