liblttngkconsumerd/lttngkconsumerd.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
   6  * modify it under the terms of the GNU General Public License
   7  * as published by the Free Software Foundation; only version 2
   8  * of the License.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program; if not, write to the Free Software
  17  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  18  */
  19
  20 #define _GNU_SOURCE
  21 #include <assert.h>
  22 #include <fcntl.h>
  23 #include <poll.h>
  24 #include <pthread.h>
  25 #include <stdlib.h>
  26 #include <string.h>
  27 #include <sys/mman.h>
  28 #include <sys/socket.h>
  29 #include <sys/types.h>
  30 #include <unistd.h>
  31
  32 #include <lttng-kernel-ctl.h>
  33 #include <lttng-sessiond-comm.h>
  34 #include <lttng/lttng-kconsumerd.h>
  35 #include <lttngerr.h>
  36
  37 static struct lttng_kconsumerd_global_data {
  38         /*
  39          * kconsumerd_data.lock protects kconsumerd_data.fd_list,
  40          * kconsumerd_data.fds_count, and kconsumerd_data.need_update. It ensures
  41          * the count matches the number of items in the fd_list. It ensures the
  42          * list updates *always* trigger an fd_array update (therefore need to make
  43          * list update vs kconsumerd_data.need_update flag update atomic, and also
  44          * flag read, fd array and flag clear atomic).
  45          */
  46         pthread_mutex_t lock;
  47         /*
  48          * Number of element for the list below. Protected by kconsumerd_data.lock.
  49          */
  50         unsigned int fds_count;
  51         /*
  52          * List of FDs. Protected by kconsumerd_data.lock.
  53          */
  54         struct lttng_kconsumerd_fd_list fd_list;
  55         /*
  56          * Flag specifying if the local array of FDs needs update in the poll
  57          * function. Protected by kconsumerd_data.lock.
  58          */
  59         unsigned int need_update;
  60 } kconsumerd_data = {
  61         .fd_list.head = CDS_LIST_HEAD_INIT(kconsumerd_data.fd_list.head),
  62         .fds_count = 0,
  63         .need_update = 1,
  64 };
  65
  66 /* timeout parameter, to control the polling thread grace period. */
  67 static int kconsumerd_poll_timeout = -1;
  68
  69 /*
  70  * Flag to inform the polling thread to quit when all fd hung up. Updated by
  71  * the kconsumerd_thread_receive_fds when it notices that all fds has hung up.
  72  * Also updated by the signal handler (kconsumerd_should_exit()). Read by the
  73  * polling threads.
  74  */
  75 static volatile int kconsumerd_quit = 0;
  76
  77 /*
  78  * Find a session fd in the global list. The kconsumerd_data.lock must be
  79  * locked during this call.
  80  *
  81  * Return 1 if found else 0.
  82  */
  83 static int kconsumerd_find_session_fd(int fd)
  84 {
  85         struct lttng_kconsumerd_fd *iter;
  86
  87         cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
  88                 if (iter->sessiond_fd == fd) {
  89                         DBG("Duplicate session fd %d", fd);
  90                         return 1;
  91                 }
  92         }
  93
  94         return 0;
  95 }
  96
  97 /*
  98  * Remove a fd from the global list protected by a mutex.
  99  */
 100 static void kconsumerd_del_fd(struct lttng_kconsumerd_fd *lcf)
 101 {
 102         int ret;
 103         pthread_mutex_lock(&kconsumerd_data.lock);
 104         cds_list_del(&lcf->list);
 105         if (kconsumerd_data.fds_count > 0) {
 106                 kconsumerd_data.fds_count--;
 107                 if (lcf != NULL) {
 108                         if (lcf->mmap_base != NULL) {
 109                                 ret = munmap(lcf->mmap_base, lcf->mmap_len);
 110                                 if (ret != 0) {
 111                                         perror("munmap");
 112                                 }
 113                         }
 114                         if (lcf->out_fd != 0) {
 115                                 close(lcf->out_fd);
 116                         }
 117                         close(lcf->consumerd_fd);
 118                         free(lcf);
 119                         lcf = NULL;
 120                 }
 121         }
 122         kconsumerd_data.need_update = 1;
 123         pthread_mutex_unlock(&kconsumerd_data.lock);
 124 }
 125
 126 /*
 127  * Create a struct lttcomm_kconsumerd_msg from the
 128  * information received on the receiving socket
 129  */
 130 struct lttng_kconsumerd_fd *kconsumerd_allocate_fd(
 131                 struct lttcomm_kconsumerd_msg *buf,
 132                 int consumerd_fd)
 133 {
 134         struct lttng_kconsumerd_fd *tmp_fd;
 135
 136         tmp_fd = malloc(sizeof(struct lttng_kconsumerd_fd));
 137         if (tmp_fd == NULL) {
 138                 perror("malloc struct lttng_kconsumerd_fd");
 139                 goto end;
 140         }
 141
 142         tmp_fd->sessiond_fd = buf->fd;
 143         tmp_fd->consumerd_fd = consumerd_fd;
 144         tmp_fd->state = buf->state;
 145         tmp_fd->max_sb_size = buf->max_sb_size;
 146         tmp_fd->out_fd = 0;
 147         tmp_fd->out_fd_offset = 0;
 148         tmp_fd->mmap_len = 0;
 149         tmp_fd->mmap_base = NULL;
 150         tmp_fd->output = buf->output;
 151         strncpy(tmp_fd->path_name, buf->path_name, PATH_MAX);
 152         tmp_fd->path_name[PATH_MAX - 1] = '\0';
 153         DBG("Allocated %s (sessiond_fd %d, consumerd_fd %d, out_fd %d)",
 154                         tmp_fd->path_name, tmp_fd->sessiond_fd,
 155                         tmp_fd->consumerd_fd, tmp_fd->out_fd);
 156
 157 end:
 158         return tmp_fd;
 159 }
 160
 161 /*
 162  * Add a fd to the global list protected by a mutex.
 163  */
 164 static int kconsumerd_add_fd(struct lttng_kconsumerd_fd *tmp_fd)
 165 {
 166         int ret;
 167
 168         pthread_mutex_lock(&kconsumerd_data.lock);
 169         /* Check if already exist */
 170         ret = kconsumerd_find_session_fd(tmp_fd->sessiond_fd);
 171         if (ret == 1) {
 172                 goto end;
 173         }
 174         cds_list_add(&tmp_fd->list, &kconsumerd_data.fd_list.head);
 175         kconsumerd_data.fds_count++;
 176         kconsumerd_data.need_update = 1;
 177
 178 end:
 179         pthread_mutex_unlock(&kconsumerd_data.lock);
 180         return ret;
 181 }
 182
 183 /*
 184  * Update a fd according to what we just received.
 185  */
 186 static void kconsumerd_change_fd_state(int sessiond_fd,
 187                 enum lttng_kconsumerd_fd_state state)
 188 {
 189         struct lttng_kconsumerd_fd *iter;
 190
 191         pthread_mutex_lock(&kconsumerd_data.lock);
 192         cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
 193                 if (iter->sessiond_fd == sessiond_fd) {
 194                         iter->state = state;
 195                         break;
 196                 }
 197         }
 198         kconsumerd_data.need_update = 1;
 199         pthread_mutex_unlock(&kconsumerd_data.lock);
 200 }
 201
 202 /*
 203  * Allocate the pollfd structure and the local view of the out fds to avoid
 204  * doing a lookup in the linked list and concurrency issues when writing is
 205  * needed. Called with kconsumerd_data.lock held.
 206  *
 207  * Returns the number of fds in the structures.
 208  */
 209 static int kconsumerd_update_poll_array(
 210                 struct lttng_kconsumerd_local_data *ctx, struct pollfd **pollfd,
 211                 struct lttng_kconsumerd_fd **local_kconsumerd_fd)
 212 {
 213         struct lttng_kconsumerd_fd *iter;
 214         int i = 0;
 215
 216         DBG("Updating poll fd array");
 217         cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
 218                 if (iter->state == ACTIVE_FD) {
 219                         DBG("Active FD %d", iter->consumerd_fd);
 220                         (*pollfd)[i].fd = iter->consumerd_fd;
 221                         (*pollfd)[i].events = POLLIN | POLLPRI;
 222                         local_kconsumerd_fd[i] = iter;
 223                         i++;
 224                 }
 225         }
 226
 227         /*
 228          * Insert the kconsumerd_poll_pipe at the end of the array and don't
 229          * increment i so nb_fd is the number of real FD.
 230          */
 231         (*pollfd)[i].fd = ctx->kconsumerd_poll_pipe[0];
 232         (*pollfd)[i].events = POLLIN;
 233         return i;
 234 }
 235
 236 /*
 237  * Receives an array of file descriptors and the associated structures
 238  * describing each fd (path name).
 239  *
 240  * Returns the size of received data
 241  */
 242 static int kconsumerd_consumerd_recv_fd(
 243                 struct lttng_kconsumerd_local_data *ctx, int sfd,
 244                 struct pollfd *kconsumerd_sockpoll, int size,
 245                 enum lttng_kconsumerd_command cmd_type)
 246 {
 247         struct iovec iov[1];
 248         int ret = 0, i, j, tmp2;
 249         struct cmsghdr *cmsg;
 250         int nb_fd;
 251         char recv_fd[CMSG_SPACE(sizeof(int))];
 252         struct lttcomm_kconsumerd_msg lkm;
 253         struct lttng_kconsumerd_fd *new_fd;
 254         union {
 255                 unsigned char vc[4];
 256                 int vi;
 257         } tmp;
 258
 259         /* the number of fds we are about to receive */
 260         nb_fd = size / sizeof(struct lttcomm_kconsumerd_msg);
 261
 262         /*
 263          * nb_fd is the number of fds we receive. One fd per recvmsg.
 264          */
 265         for (i = 0; i < nb_fd; i++) {
 266                 struct msghdr msg = { 0 };
 267
 268                 /* Prepare to receive the structures */
 269                 iov[0].iov_base = &lkm;
 270                 iov[0].iov_len = sizeof(lkm);
 271                 msg.msg_iov = iov;
 272                 msg.msg_iovlen = 1;
 273
 274                 msg.msg_control = recv_fd;
 275                 msg.msg_controllen = sizeof(recv_fd);
 276
 277                 DBG("Waiting to receive fd");
 278                 if (lttng_kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
 279                         goto end;
 280                 }
 281
 282                 if ((ret = recvmsg(sfd, &msg, 0)) < 0) {
 283                         perror("recvmsg");
 284                         continue;
 285                 }
 286
 287                 if (ret != (size / nb_fd)) {
 288                         ERR("Received only %d, expected %d", ret, size);
 289                         lttng_kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD);
 290                         goto end;
 291                 }
 292
 293                 cmsg = CMSG_FIRSTHDR(&msg);
 294                 if (!cmsg) {
 295                         ERR("Invalid control message header");
 296                         ret = -1;
 297                         lttng_kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD);
 298                         goto end;
 299                 }
 300
 301                 /* if we received fds */
 302                 if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
 303                         switch (cmd_type) {
 304                                 case ADD_STREAM:
 305                                         for (j = 0; j < sizeof(int); j++)
 306                                                 tmp.vc[j] = CMSG_DATA(cmsg)[j];
 307                                         DBG("kconsumerd_add_fd %s (%d)", lkm.path_name, tmp.vi);
 308                                         new_fd = kconsumerd_allocate_fd(&lkm, tmp.vi);
 309                                         if (new_fd == NULL) {
 310                                                 lttng_kconsumerd_send_error(ctx, KCONSUMERD_OUTFD_ERROR);
 311                                                 goto end;
 312                                         }
 313
 314                                         if (ctx->on_recv_fd != NULL) {
 315                                                 ret = ctx->on_recv_fd(new_fd);
 316                                                 if (ret == 0) {
 317                                                         kconsumerd_add_fd(new_fd);
 318                                                 } else if (ret < 0) {
 319                                                         goto end;
 320                                                 }
 321                                         } else {
 322                                                 kconsumerd_add_fd(new_fd);
 323                                         }
 324                                         break;
 325                                 case UPDATE_STREAM:
 326                                         if (ctx->on_update_fd != NULL) {
 327                                                 ret = ctx->on_update_fd(lkm.fd, lkm.state);
 328                                                 if (ret == 0) {
 329                                                         kconsumerd_change_fd_state(lkm.fd, lkm.state);
 330                                                 } else if (ret < 0) {
 331                                                         goto end;
 332                                                 }
 333                                         } else {
 334                                                 kconsumerd_change_fd_state(lkm.fd, lkm.state);
 335                                         }
 336                                         break;
 337                                 default:
 338                                         break;
 339                         }
 340                         /* signal the poll thread */
 341                         tmp2 = write(ctx->kconsumerd_poll_pipe[1], "4", 1);
 342                         if (tmp2 < 0) {
 343                                 perror("write kconsumerd poll");
 344                         }
 345                 } else {
 346                         ERR("Didn't received any fd");
 347                         lttng_kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD);
 348                         ret = -1;
 349                         goto end;
 350                 }
 351         }
 352
 353 end:
 354         return ret;
 355 }
 356
 357 /*
 358  * Set the error socket.
 359  */
 360 void lttng_kconsumerd_set_error_sock(
 361                 struct lttng_kconsumerd_local_data *ctx, int sock)
 362 {
 363         ctx->kconsumerd_error_socket = sock;
 364 }
 365
 366 /*
 367  * Set the command socket path.
 368  */
 369
 370 void lttng_kconsumerd_set_command_sock_path(
 371                 struct lttng_kconsumerd_local_data *ctx, char *sock)
 372 {
 373         ctx->kconsumerd_command_sock_path = sock;
 374 }
 375
 376 static void lttng_kconsumerd_sync_trace_file(
 377                 struct lttng_kconsumerd_fd *kconsumerd_fd, off_t orig_offset)
 378 {
 379         int outfd = kconsumerd_fd->out_fd;
 380         /*
 381          * This does a blocking write-and-wait on any page that belongs to the
 382          * subbuffer prior to the one we just wrote.
 383          * Don't care about error values, as these are just hints and ways to
 384          * limit the amount of page cache used.
 385          */
 386         if (orig_offset >= kconsumerd_fd->max_sb_size) {
 387                 sync_file_range(outfd, orig_offset - kconsumerd_fd->max_sb_size,
 388                                 kconsumerd_fd->max_sb_size,
 389                                 SYNC_FILE_RANGE_WAIT_BEFORE
 390                                 | SYNC_FILE_RANGE_WRITE
 391                                 | SYNC_FILE_RANGE_WAIT_AFTER);
 392                 /*
 393                  * Give hints to the kernel about how we access the file:
 394                  * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
 395                  * we write it.
 396                  *
 397                  * We need to call fadvise again after the file grows because the
 398                  * kernel does not seem to apply fadvise to non-existing parts of the
 399                  * file.
 400                  *
 401                  * Call fadvise _after_ having waited for the page writeback to
 402                  * complete because the dirty page writeback semantic is not well
 403                  * defined. So it can be expected to lead to lower throughput in
 404                  * streaming.
 405                  */
 406                 posix_fadvise(outfd, orig_offset - kconsumerd_fd->max_sb_size,
 407                                 kconsumerd_fd->max_sb_size, POSIX_FADV_DONTNEED);
 408         }
 409 }
 410
 411
 412 /*
 413  * Mmap the ring buffer, read it and write the data to the tracefile.
 414  *
 415  * Returns the number of bytes written
 416  */
 417 int lttng_kconsumerd_on_read_subbuffer_mmap(
 418                 struct lttng_kconsumerd_local_data *ctx,
 419                 struct lttng_kconsumerd_fd *kconsumerd_fd, unsigned long len)
 420 {
 421         unsigned long mmap_offset;
 422         long ret = 0;
 423         off_t orig_offset = kconsumerd_fd->out_fd_offset;
 424         int fd = kconsumerd_fd->consumerd_fd;
 425         int outfd = kconsumerd_fd->out_fd;
 426
 427         /* get the offset inside the fd to mmap */
 428         ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
 429         if (ret != 0) {
 430                 ret = errno;
 431                 perror("kernctl_get_mmap_read_offset");
 432                 goto end;
 433         }
 434
 435         while (len > 0) {
 436                 ret = write(outfd, kconsumerd_fd->mmap_base + mmap_offset, len);
 437                 if (ret >= len) {
 438                         len = 0;
 439                 } else if (ret < 0) {
 440                         ret = errno;
 441                         perror("Error in file write");
 442                         goto end;
 443                 }
 444                 /* This won't block, but will start writeout asynchronously */
 445                 sync_file_range(outfd, kconsumerd_fd->out_fd_offset, ret,
 446                                 SYNC_FILE_RANGE_WRITE);
 447                 kconsumerd_fd->out_fd_offset += ret;
 448         }
 449
 450         lttng_kconsumerd_sync_trace_file(kconsumerd_fd, orig_offset);
 451
 452         goto end;
 453
 454 end:
 455         return ret;
 456 }
 457
 458 /*
 459  * Splice the data from the ring buffer to the tracefile.
 460  *
 461  * Returns the number of bytes spliced.
 462  */
 463 int lttng_kconsumerd_on_read_subbuffer_splice(
 464                 struct lttng_kconsumerd_local_data *ctx,
 465                 struct lttng_kconsumerd_fd *kconsumerd_fd, unsigned long len)
 466 {
 467         long ret = 0;
 468         loff_t offset = 0;
 469         off_t orig_offset = kconsumerd_fd->out_fd_offset;
 470         int fd = kconsumerd_fd->consumerd_fd;
 471         int outfd = kconsumerd_fd->out_fd;
 472
 473         while (len > 0) {
 474                 DBG("splice chan to pipe offset %lu (fd : %d)",
 475                                 (unsigned long)offset, fd);
 476                 ret = splice(fd, &offset, ctx->kconsumerd_thread_pipe[1], NULL, len,
 477                                 SPLICE_F_MOVE | SPLICE_F_MORE);
 478                 DBG("splice chan to pipe ret %ld", ret);
 479                 if (ret < 0) {
 480                         ret = errno;
 481                         perror("Error in relay splice");
 482                         goto splice_error;
 483                 }
 484
 485                 ret = splice(ctx->kconsumerd_thread_pipe[0], NULL, outfd, NULL, ret,
 486                                 SPLICE_F_MOVE | SPLICE_F_MORE);
 487                 DBG("splice pipe to file %ld", ret);
 488                 if (ret < 0) {
 489                         ret = errno;
 490                         perror("Error in file splice");
 491                         goto splice_error;
 492                 }
 493                 len -= ret;
 494                 /* This won't block, but will start writeout asynchronously */
 495                 sync_file_range(outfd, kconsumerd_fd->out_fd_offset, ret,
 496                                 SYNC_FILE_RANGE_WRITE);
 497                 kconsumerd_fd->out_fd_offset += ret;
 498         }
 499         lttng_kconsumerd_sync_trace_file(kconsumerd_fd, orig_offset);
 500
 501         goto end;
 502
 503 splice_error:
 504         /* send the appropriate error description to sessiond */
 505         switch(ret) {
 506         case EBADF:
 507                 lttng_kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_EBADF);
 508                 break;
 509         case EINVAL:
 510                 lttng_kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_EINVAL);
 511                 break;
 512         case ENOMEM:
 513                 lttng_kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_ENOMEM);
 514                 break;
 515         case ESPIPE:
 516                 lttng_kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_ESPIPE);
 517                 break;
 518         }
 519
 520 end:
 521         return ret;
 522 }
 523
 524 /*
 525  * Take a snapshot for a specific fd
 526  *
 527  * Returns 0 on success, < 0 on error
 528  */
 529 int lttng_kconsumerd_take_snapshot(struct lttng_kconsumerd_local_data *ctx,
 530                 struct lttng_kconsumerd_fd *kconsumerd_fd)
 531 {
 532         int ret = 0;
 533         int infd = kconsumerd_fd->consumerd_fd;
 534
 535         ret = kernctl_snapshot(infd);
 536         if (ret != 0) {
 537                 ret = errno;
 538                 perror("Getting sub-buffer snapshot.");
 539         }
 540
 541         return ret;
 542 }
 543
 544 /*
 545  * Get the produced position
 546  *
 547  * Returns 0 on success, < 0 on error
 548  */
 549 int lttng_kconsumerd_get_produced_snapshot(
 550                 struct lttng_kconsumerd_local_data *ctx,
 551                 struct lttng_kconsumerd_fd *kconsumerd_fd,
 552                 unsigned long *pos)
 553 {
 554         int ret;
 555         int infd = kconsumerd_fd->consumerd_fd;
 556
 557         ret = kernctl_snapshot_get_produced(infd, pos);
 558         if (ret != 0) {
 559                 ret = errno;
 560                 perror("kernctl_snapshot_get_produced");
 561         }
 562
 563         return ret;
 564 }
 565
 566 /*
 567  * Poll on the should_quit pipe and the command socket return -1 on error and
 568  * should exit, 0 if data is available on the command socket
 569  */
 570 int lttng_kconsumerd_poll_socket(struct pollfd *kconsumerd_sockpoll)
 571 {
 572         int num_rdy;
 573
 574         num_rdy = poll(kconsumerd_sockpoll, 2, -1);
 575         if (num_rdy == -1) {
 576                 perror("Poll error");
 577                 goto exit;
 578         }
 579         if (kconsumerd_sockpoll[0].revents == POLLIN) {
 580                 DBG("kconsumerd_should_quit wake up");
 581                 goto exit;
 582         }
 583         return 0;
 584
 585 exit:
 586         return -1;
 587 }
 588
 589 /*
 590  * This thread polls the fds in the ltt_fd_list to consume the data and write
 591  * it to tracefile if necessary.
 592  */
 593 void *lttng_kconsumerd_thread_poll_fds(void *data)
 594 {
 595         int num_rdy, num_hup, high_prio, ret, i;
 596         struct pollfd *pollfd = NULL;
 597         /* local view of the fds */
 598         struct lttng_kconsumerd_fd **local_kconsumerd_fd = NULL;
 599         /* local view of kconsumerd_data.fds_count */
 600         int nb_fd = 0;
 601         char tmp;
 602         int tmp2;
 603         struct lttng_kconsumerd_local_data *ctx = data;
 604
 605
 606         local_kconsumerd_fd = malloc(sizeof(struct lttng_kconsumerd_fd));
 607
 608         while (1) {
 609                 high_prio = 0;
 610                 num_hup = 0;
 611
 612                 /*
 613                  * the ltt_fd_list has been updated, we need to update our
 614                  * local array as well
 615                  */
 616                 pthread_mutex_lock(&kconsumerd_data.lock);
 617                 if (kconsumerd_data.need_update) {
 618                         if (pollfd != NULL) {
 619                                 free(pollfd);
 620                                 pollfd = NULL;
 621                         }
 622                         if (local_kconsumerd_fd != NULL) {
 623                                 free(local_kconsumerd_fd);
 624                                 local_kconsumerd_fd = NULL;
 625                         }
 626
 627                         /* allocate for all fds + 1 for the kconsumerd_poll_pipe */
 628                         pollfd = malloc((kconsumerd_data.fds_count + 1) * sizeof(struct pollfd));
 629                         if (pollfd == NULL) {
 630                                 perror("pollfd malloc");
 631                                 pthread_mutex_unlock(&kconsumerd_data.lock);
 632                                 goto end;
 633                         }
 634
 635                         /* allocate for all fds + 1 for the kconsumerd_poll_pipe */
 636                         local_kconsumerd_fd = malloc((kconsumerd_data.fds_count + 1) *
 637                                         sizeof(struct lttng_kconsumerd_fd));
 638                         if (local_kconsumerd_fd == NULL) {
 639                                 perror("local_kconsumerd_fd malloc");
 640                                 pthread_mutex_unlock(&kconsumerd_data.lock);
 641                                 goto end;
 642                         }
 643                         ret = kconsumerd_update_poll_array(ctx, &pollfd, local_kconsumerd_fd);
 644                         if (ret < 0) {
 645                                 ERR("Error in allocating pollfd or local_outfds");
 646                                 lttng_kconsumerd_send_error(ctx, KCONSUMERD_POLL_ERROR);
 647                                 pthread_mutex_unlock(&kconsumerd_data.lock);
 648                                 goto end;
 649                         }
 650                         nb_fd = ret;
 651                         kconsumerd_data.need_update = 0;
 652                 }
 653                 pthread_mutex_unlock(&kconsumerd_data.lock);
 654
 655                 /* poll on the array of fds */
 656                 DBG("polling on %d fd", nb_fd + 1);
 657                 num_rdy = poll(pollfd, nb_fd + 1, kconsumerd_poll_timeout);
 658                 DBG("poll num_rdy : %d", num_rdy);
 659                 if (num_rdy == -1) {
 660                         perror("Poll error");
 661                         lttng_kconsumerd_send_error(ctx, KCONSUMERD_POLL_ERROR);
 662                         goto end;
 663                 } else if (num_rdy == 0) {
 664                         DBG("Polling thread timed out");
 665                         goto end;
 666                 }
 667
 668                 /* No FDs and kconsumerd_quit, kconsumerd_cleanup the thread */
 669                 if (nb_fd == 0 && kconsumerd_quit == 1) {
 670                         goto end;
 671                 }
 672
 673                 /*
 674                  * If the kconsumerd_poll_pipe triggered poll go
 675                  * directly to the beginning of the loop to update the
 676                  * array. We want to prioritize array update over
 677                  * low-priority reads.
 678                  */
 679                 if (pollfd[nb_fd].revents == POLLIN) {
 680                         DBG("kconsumerd_poll_pipe wake up");
 681                         tmp2 = read(ctx->kconsumerd_poll_pipe[0], &tmp, 1);
 682                         if (tmp2 < 0) {
 683                                 perror("read kconsumerd poll");
 684                         }
 685                         continue;
 686                 }
 687
 688                 /* Take care of high priority channels first. */
 689                 for (i = 0; i < nb_fd; i++) {
 690                         switch(pollfd[i].revents) {
 691                         case POLLERR:
 692                                 ERR("Error returned in polling fd %d.", pollfd[i].fd);
 693                                 kconsumerd_del_fd(local_kconsumerd_fd[i]);
 694                                 num_hup++;
 695                                 break;
 696                         case POLLHUP:
 697                                 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
 698                                 kconsumerd_del_fd(local_kconsumerd_fd[i]);
 699                                 num_hup++;
 700                                 break;
 701                         case POLLNVAL:
 702                                 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
 703                                 kconsumerd_del_fd(local_kconsumerd_fd[i]);
 704                                 num_hup++;
 705                                 break;
 706                         case POLLPRI:
 707                                 DBG("Urgent read on fd %d", pollfd[i].fd);
 708                                 high_prio = 1;
 709                                 ret = ctx->on_buffer_ready(local_kconsumerd_fd[i]);
 710                                 /* it's ok to have an unavailable sub-buffer */
 711                                 if (ret == EAGAIN) {
 712                                         ret = 0;
 713                                 }
 714                                 break;
 715                         }
 716                 }
 717
 718                 /* If every buffer FD has hung up, we end the read loop here */
 719                 if (nb_fd > 0 && num_hup == nb_fd) {
 720                         DBG("every buffer FD has hung up\n");
 721                         if (kconsumerd_quit == 1) {
 722                                 goto end;
 723                         }
 724                         continue;
 725                 }
 726
 727                 /* Take care of low priority channels. */
 728                 if (high_prio == 0) {
 729                         for (i = 0; i < nb_fd; i++) {
 730                                 if (pollfd[i].revents == POLLIN) {
 731                                         DBG("Normal read on fd %d", pollfd[i].fd);
 732                                         ret = ctx->on_buffer_ready(local_kconsumerd_fd[i]);
 733                                         /* it's ok to have an unavailable subbuffer */
 734                                         if (ret == EAGAIN) {
 735                                                 ret = 0;
 736                                         }
 737                                 }
 738                         }
 739                 }
 740         }
 741 end:
 742         DBG("polling thread exiting");
 743         if (pollfd != NULL) {
 744                 free(pollfd);
 745                 pollfd = NULL;
 746         }
 747         if (local_kconsumerd_fd != NULL) {
 748                 free(local_kconsumerd_fd);
 749                 local_kconsumerd_fd = NULL;
 750         }
 751         return NULL;
 752 }
 753
 754 /*
 755  * Initialise the necessary environnement :
 756  * - create a new context
 757  * - create the poll_pipe
 758  * - create the should_quit pipe (for signal handler)
 759  * - create the thread pipe (for splice)
 760  *
 761  * Takes a function pointer as argument, this function is called when data is
 762  * available on a buffer. This function is responsible to do the
 763  * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
 764  * buffer configuration and then kernctl_put_next_subbuf at the end.
 765  *
 766  * Returns a pointer to the new context or NULL on error.
 767  */
 768 struct lttng_kconsumerd_local_data *lttng_kconsumerd_create(
 769                 int (*buffer_ready)(struct lttng_kconsumerd_fd *kconsumerd_fd),
 770                 int (*recv_fd)(struct lttng_kconsumerd_fd *kconsumerd_fd),
 771                 int (*update_fd)(int sessiond_fd, uint32_t state))
 772 {
 773         int ret, i;
 774         struct lttng_kconsumerd_local_data *ctx;
 775
 776         ctx = malloc(sizeof(struct lttng_kconsumerd_local_data));
 777         if (ctx == NULL) {
 778                 perror("allocating context");
 779                 goto error;
 780         }
 781
 782         ctx->kconsumerd_error_socket = -1;
 783         /* assign the callbacks */
 784         ctx->on_buffer_ready = buffer_ready;
 785         ctx->on_recv_fd = recv_fd;
 786         ctx->on_update_fd = update_fd;
 787
 788         ret = pipe(ctx->kconsumerd_poll_pipe);
 789         if (ret < 0) {
 790                 perror("Error creating poll pipe");
 791                 goto error_poll_pipe;
 792         }
 793
 794         ret = pipe(ctx->kconsumerd_should_quit);
 795         if (ret < 0) {
 796                 perror("Error creating recv pipe");
 797                 goto error_quit_pipe;
 798         }
 799
 800         ret = pipe(ctx->kconsumerd_thread_pipe);
 801         if (ret < 0) {
 802                 perror("Error creating thread pipe");
 803                 goto error_thread_pipe;
 804         }
 805
 806         return ctx;
 807
 808
 809 error_thread_pipe:
 810         for (i = 0; i < 2; i++) {
 811                 int err;
 812
 813                 err = close(ctx->kconsumerd_should_quit[i]);
 814                 assert(!err);
 815         }
 816 error_quit_pipe:
 817         for (i = 0; i < 2; i++) {
 818                 int err;
 819
 820                 err = close(ctx->kconsumerd_poll_pipe[i]);
 821                 assert(!err);
 822         }
 823 error_poll_pipe:
 824         free(ctx);
 825 error:
 826         return NULL;
 827 }
 828
 829 /*
 830  * Close all fds associated with the instance and free the context.
 831  */
 832 void lttng_kconsumerd_destroy(struct lttng_kconsumerd_local_data *ctx)
 833 {
 834         close(ctx->kconsumerd_error_socket);
 835         close(ctx->kconsumerd_thread_pipe[0]);
 836         close(ctx->kconsumerd_thread_pipe[1]);
 837         close(ctx->kconsumerd_poll_pipe[0]);
 838         close(ctx->kconsumerd_poll_pipe[1]);
 839         close(ctx->kconsumerd_should_quit[0]);
 840         close(ctx->kconsumerd_should_quit[1]);
 841         unlink(ctx->kconsumerd_command_sock_path);
 842         free(ctx);
 843         ctx = NULL;
 844 }
 845
 846 /*
 847  * This thread listens on the consumerd socket and receives the file
 848  * descriptors from the session daemon.
 849  */
 850 void *lttng_kconsumerd_thread_receive_fds(void *data)
 851 {
 852         int sock, client_socket, ret;
 853         struct lttcomm_kconsumerd_header tmp;
 854         /*
 855          * structure to poll for incoming data on communication socket avoids
 856          * making blocking sockets.
 857          */
 858         struct pollfd kconsumerd_sockpoll[2];
 859         struct lttng_kconsumerd_local_data *ctx = data;
 860
 861
 862         DBG("Creating command socket %s", ctx->kconsumerd_command_sock_path);
 863         unlink(ctx->kconsumerd_command_sock_path);
 864         client_socket = lttcomm_create_unix_sock(ctx->kconsumerd_command_sock_path);
 865         if (client_socket < 0) {
 866                 ERR("Cannot create command socket");
 867                 goto end;
 868         }
 869
 870         ret = lttcomm_listen_unix_sock(client_socket);
 871         if (ret < 0) {
 872                 goto end;
 873         }
 874
 875         DBG("Sending ready command to ltt-sessiond");
 876         ret = lttng_kconsumerd_send_error(ctx, KCONSUMERD_COMMAND_SOCK_READY);
 877         /* return < 0 on error, but == 0 is not fatal */
 878         if (ret < 0) {
 879                 ERR("Error sending ready command to ltt-sessiond");
 880                 goto end;
 881         }
 882
 883         ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
 884         if (ret < 0) {
 885                 perror("fcntl O_NONBLOCK");
 886                 goto end;
 887         }
 888
 889         /* prepare the FDs to poll : to client socket and the should_quit pipe */
 890         kconsumerd_sockpoll[0].fd = ctx->kconsumerd_should_quit[0];
 891         kconsumerd_sockpoll[0].events = POLLIN | POLLPRI;
 892         kconsumerd_sockpoll[1].fd = client_socket;
 893         kconsumerd_sockpoll[1].events = POLLIN | POLLPRI;
 894
 895         if (lttng_kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
 896                 goto end;
 897         }
 898         DBG("Connection on client_socket");
 899
 900         /* Blocking call, waiting for transmission */
 901         sock = lttcomm_accept_unix_sock(client_socket);
 902         if (sock <= 0) {
 903                 WARN("On accept");
 904                 goto end;
 905         }
 906         ret = fcntl(sock, F_SETFL, O_NONBLOCK);
 907         if (ret < 0) {
 908                 perror("fcntl O_NONBLOCK");
 909                 goto end;
 910         }
 911
 912         /* update the polling structure to poll on the established socket */
 913         kconsumerd_sockpoll[1].fd = sock;
 914         kconsumerd_sockpoll[1].events = POLLIN | POLLPRI;
 915
 916         while (1) {
 917                 if (lttng_kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
 918                         goto end;
 919                 }
 920                 DBG("Incoming fds on sock");
 921
 922                 /* We first get the number of fd we are about to receive */
 923                 ret = lttcomm_recv_unix_sock(sock, &tmp,
 924                                 sizeof(struct lttcomm_kconsumerd_header));
 925                 if (ret <= 0) {
 926                         ERR("Communication interrupted on command socket");
 927                         goto end;
 928                 }
 929                 if (tmp.cmd_type == STOP) {
 930                         DBG("Received STOP command");
 931                         goto end;
 932                 }
 933                 if (kconsumerd_quit) {
 934                         DBG("kconsumerd_thread_receive_fds received quit from signal");
 935                         goto end;
 936                 }
 937
 938                 /* we received a command to add or update fds */
 939                 ret = kconsumerd_consumerd_recv_fd(ctx, sock, kconsumerd_sockpoll,
 940                                 tmp.payload_size, tmp.cmd_type);
 941                 if (ret < 0) {
 942                         ERR("Receiving the FD, exiting");
 943                         goto end;
 944                 }
 945                 DBG("received fds on sock");
 946         }
 947
 948 end:
 949         DBG("kconsumerd_thread_receive_fds exiting");
 950
 951         /*
 952          * when all fds have hung up, the polling thread
 953          * can exit cleanly
 954          */
 955         kconsumerd_quit = 1;
 956
 957         /*
 958          * 2s of grace period, if no polling events occur during
 959          * this period, the polling thread will exit even if there
 960          * are still open FDs (should not happen, but safety mechanism).
 961          */
 962         kconsumerd_poll_timeout = LTTNG_KCONSUMERD_POLL_GRACE_PERIOD;
 963
 964         /* wake up the polling thread */
 965         ret = write(ctx->kconsumerd_poll_pipe[1], "4", 1);
 966         if (ret < 0) {
 967                 perror("poll pipe write");
 968         }
 969         return NULL;
 970 }
 971
 972 /*
 973  * Close all the tracefiles and stream fds, should be called when all instances
 974  * are destroyed.
 975  */
 976 void lttng_kconsumerd_cleanup(void)
 977 {
 978         struct lttng_kconsumerd_fd *iter, *tmp;
 979
 980         /*
 981          * close all outfd. Called when there are no more threads
 982          * running (after joining on the threads), no need to protect
 983          * list iteration with mutex.
 984          */
 985         cds_list_for_each_entry_safe(iter, tmp,
 986                         &kconsumerd_data.fd_list.head, list) {
 987                 kconsumerd_del_fd(iter);
 988         }
 989 }
 990
 991 /*
 992  * Called from signal handler.
 993  */
 994 void lttng_kconsumerd_should_exit(struct lttng_kconsumerd_local_data *ctx)
 995 {
 996         int ret;
 997         kconsumerd_quit = 1;
 998         ret = write(ctx->kconsumerd_should_quit[1], "4", 1);
 999         if (ret < 0) {
1000                 perror("write kconsumerd quit");
1001         }
1002 }
1003
1004 /*
1005  * Send return code to the session daemon.
1006  * If the socket is not defined, we return 0, it is not a fatal error
1007  */
1008 int lttng_kconsumerd_send_error(
1009                 struct lttng_kconsumerd_local_data *ctx, int cmd)
1010 {
1011         if (ctx->kconsumerd_error_socket > 0) {
1012                 return lttcomm_send_unix_sock(ctx->kconsumerd_error_socket, &cmd,
1013                                 sizeof(enum lttcomm_sessiond_command));
1014         }
1015
1016         return 0;
1017 }