kconsumerd/kconsumerd.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; either version 2
   8  * of the License, or (at your option) any later version.
   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 <fcntl.h>
  22 #include <getopt.h>
  23 #include <grp.h>
  24 #include <limits.h>
  25 #include <pthread.h>
  26 #include <signal.h>
  27 #include <stdio.h>
  28 #include <stdlib.h>
  29 #include <string.h>
  30 #include <sys/ipc.h>
  31 #include <sys/shm.h>
  32 #include <sys/socket.h>
  33 #include <sys/stat.h>
  34 #include <sys/types.h>
  35 #include <urcu/list.h>
  36 #include <poll.h>
  37 #include <unistd.h>
  38
  39 #include "lttngerr.h"
  40 #include "libkernelctl.h"
  41 #include "liblttsessiondcomm.h"
  42 #include "kconsumerd.h"
  43
  44 /* Init the list of FDs */
  45 static struct ltt_kconsumerd_fd_list kconsumerd_fd_list = {
  46         .head = CDS_LIST_HEAD_INIT(kconsumerd_fd_list.head),
  47 };
  48
  49 /* Number of element for the list below. */
  50 static unsigned int fds_count;
  51
  52 /* If the local array of FDs needs update in the poll function */
  53 static unsigned int update_fd_array = 1;
  54
  55 /* lock the fd array and structures */
  56 static pthread_mutex_t kconsumerd_lock_fds;
  57
  58 /* the two threads (receive fd and poll) */
  59 static pthread_t threads[2];
  60
  61 /* communication with splice */
  62 static int thread_pipe[2];
  63
  64 /* socket to communicate errors with sessiond */
  65 static int error_socket = -1;
  66
  67 /* Argument variables */
  68 int opt_quiet;
  69 int opt_verbose;
  70 static int opt_daemon;
  71 static const char *progname;
  72 static char command_sock_path[PATH_MAX]; /* Global command socket path */
  73 static char error_sock_path[PATH_MAX]; /* Global error path */
  74
  75 /*
  76  *  cleanup
  77  *
  78  *  Cleanup the daemon's socket on exit
  79  */
  80 static void cleanup()
  81 {
  82         unlink(command_sock_path);
  83 }
  84
  85 /* send_error
  86  *
  87  * send return code to ltt-sessiond
  88  */
  89 static int send_error(enum lttcomm_return_code cmd)
  90 {
  91         if (error_socket > 0) {
  92                 return lttcomm_send_unix_sock(error_socket, &cmd,
  93                                 sizeof(enum lttcomm_sessiond_command));
  94         } else {
  95                 return 0;
  96         }
  97 }
  98
  99 /*
 100  * cleanup_kconsumerd_fd
 101  *
 102  * Close the FDs and frees a ltt_kconsumerd_fd struct
 103  */
 104 static void cleanup_kconsumerd_fd(struct ltt_kconsumerd_fd *lcf)
 105 {
 106         if (lcf != NULL) {
 107                 close(lcf->out_fd);
 108                 close(lcf->consumerd_fd);
 109                 free(lcf);
 110                 lcf = NULL;
 111         }
 112 }
 113
 114 /*
 115  * add_fd
 116  *
 117  * Add a fd to the global list protected by a mutex
 118  */
 119 static int add_fd(struct lttcomm_kconsumerd_msg *buf, int consumerd_fd)
 120 {
 121         struct ltt_kconsumerd_fd *tmp_fd;
 122         int ret;
 123
 124         tmp_fd = malloc(sizeof(struct ltt_kconsumerd_fd));
 125         tmp_fd->sessiond_fd = buf->fd;
 126         tmp_fd->consumerd_fd = consumerd_fd;
 127         tmp_fd->state = buf->state;
 128         tmp_fd->max_sb_size = buf->max_sb_size;
 129         strncpy(tmp_fd->path_name, buf->path_name, PATH_MAX);
 130
 131         /* Opening the tracefile in write mode */
 132         DBG("Opening %s for writing", tmp_fd->path_name);
 133         ret = open(tmp_fd->path_name,
 134                         O_WRONLY|O_CREAT, S_IRWXU|S_IRWXG|S_IRWXO);
 135         if (ret < 0) {
 136                 ERR("Opening %s", tmp_fd->path_name);
 137                 perror("open");
 138                 goto end;
 139         }
 140         tmp_fd->out_fd = ret;
 141         tmp_fd->out_fd_offset = 0;
 142
 143         DBG("Adding %s (%d, %d, %d)", tmp_fd->path_name,
 144                         tmp_fd->sessiond_fd, tmp_fd->consumerd_fd, tmp_fd->out_fd);
 145
 146         pthread_mutex_lock(&kconsumerd_lock_fds);
 147         cds_list_add(&tmp_fd->list, &kconsumerd_fd_list.head);
 148         fds_count++;
 149         pthread_mutex_unlock(&kconsumerd_lock_fds);
 150
 151 end:
 152         return ret;
 153 }
 154
 155 /*
 156  * del_fd
 157  *
 158  * Remove a fd from the global list protected by a mutex
 159  */
 160 static void del_fd(struct ltt_kconsumerd_fd *lcf)
 161 {
 162         pthread_mutex_lock(&kconsumerd_lock_fds);
 163         cds_list_del(&lcf->list);
 164         if (fds_count > 0) {
 165                 fds_count--;
 166                 DBG("Removed ltt_kconsumerd_fd");
 167                 cleanup_kconsumerd_fd(lcf);
 168         }
 169         pthread_mutex_unlock(&kconsumerd_lock_fds);
 170 }
 171
 172 /*
 173  * close_outfds
 174  *
 175  * Close all fds in the previous fd_list
 176  * Must be used with kconsumerd_lock_fds lock held
 177  */
 178 static void close_outfds()
 179 {
 180         struct ltt_kconsumerd_fd *iter;
 181         cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
 182                 del_fd(iter);
 183         }
 184 }
 185
 186 /*
 187  *  sighandler
 188  *
 189  *  Signal handler for the daemon
 190  */
 191 static void sighandler(int sig)
 192 {
 193         /* unblock the threads */
 194         pthread_cancel(threads[0]);
 195         pthread_cancel(threads[1]);
 196
 197         close_outfds();
 198         cleanup();
 199
 200         return;
 201 }
 202
 203 /*
 204  *  set_signal_handler
 205  *
 206  *  Setup signal handler for :
 207  *      SIGINT, SIGTERM, SIGPIPE
 208  */
 209 static int set_signal_handler(void)
 210 {
 211         int ret = 0;
 212         struct sigaction sa;
 213         sigset_t sigset;
 214
 215         if ((ret = sigemptyset(&sigset)) < 0) {
 216                 perror("sigemptyset");
 217                 return ret;
 218         }
 219
 220         sa.sa_handler = sighandler;
 221         sa.sa_mask = sigset;
 222         sa.sa_flags = 0;
 223         if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
 224                 perror("sigaction");
 225                 return ret;
 226         }
 227
 228         if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
 229                 perror("sigaction");
 230                 return ret;
 231         }
 232
 233         if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
 234                 perror("sigaction");
 235                 return ret;
 236         }
 237
 238         return ret;
 239 }
 240
 241 /*
 242  * on_read_subbuffer
 243  *
 244  * Splice the data from the ring buffer to the tracefile.
 245  * Returns the number of bytes spliced
 246  */
 247 static int on_read_subbuffer(struct ltt_kconsumerd_fd *kconsumerd_fd,
 248                 unsigned long len)
 249 {
 250         long ret = 0;
 251         loff_t offset = 0;
 252         off_t orig_offset = kconsumerd_fd->out_fd_offset;
 253         int fd = kconsumerd_fd->consumerd_fd;
 254         int outfd = kconsumerd_fd->out_fd;
 255
 256         while (len > 0) {
 257                 DBG("splice chan to pipe offset %lu (fd : %d)",
 258                                 (unsigned long)offset, fd);
 259                 ret = splice(fd, &offset, thread_pipe[1], NULL, len,
 260                                 SPLICE_F_MOVE | SPLICE_F_MORE);
 261                 DBG("splice chan to pipe ret %ld", ret);
 262                 if (ret < 0) {
 263                         ret = errno;
 264                         perror("Error in relay splice");
 265                         goto splice_error;
 266                 }
 267
 268                 ret = splice(thread_pipe[0], NULL, outfd, NULL, ret,
 269                                 SPLICE_F_MOVE | SPLICE_F_MORE);
 270                 DBG("splice pipe to file %ld", ret);
 271                 if (ret < 0) {
 272                         ret = errno;
 273                         perror("Error in file splice");
 274                         goto splice_error;
 275                 }
 276                 if (ret >= len) {
 277                         len = 0;
 278                 }
 279                 /* This won't block, but will start writeout asynchronously */
 280                 sync_file_range(outfd, kconsumerd_fd->out_fd_offset, ret,
 281                                 SYNC_FILE_RANGE_WRITE);
 282                 kconsumerd_fd->out_fd_offset += ret;
 283         }
 284
 285         /*
 286          * This does a blocking write-and-wait on any page that belongs to the
 287          * subbuffer prior to the one we just wrote.
 288          * Don't care about error values, as these are just hints and ways to
 289          * limit the amount of page cache used.
 290          */
 291         if (orig_offset >= kconsumerd_fd->max_sb_size) {
 292                 sync_file_range(outfd, orig_offset - kconsumerd_fd->max_sb_size,
 293                                 kconsumerd_fd->max_sb_size,
 294                                 SYNC_FILE_RANGE_WAIT_BEFORE
 295                                 | SYNC_FILE_RANGE_WRITE
 296                                 | SYNC_FILE_RANGE_WAIT_AFTER);
 297                 /*
 298                  * Give hints to the kernel about how we access the file:
 299                  * POSIX_FADV_DONTNEED : we won't re-access data in a near
 300                  * future after we write it.
 301                  * We need to call fadvise again after the file grows because
 302                  * the kernel does not seem to apply fadvise to non-existing
 303                  * parts of the file.
 304                  * Call fadvise _after_ having waited for the page writeback to
 305                  * complete because the dirty page writeback semantic is not
 306                  * well defined. So it can be expected to lead to lower
 307                  * throughput in streaming.
 308                  */
 309                 posix_fadvise(outfd, orig_offset - kconsumerd_fd->max_sb_size,
 310                                 kconsumerd_fd->max_sb_size, POSIX_FADV_DONTNEED);
 311         }
 312         goto end;
 313
 314 splice_error:
 315         /* send the appropriate error description to sessiond */
 316         switch(ret) {
 317                 case EBADF:
 318                         send_error(KCONSUMERD_SPLICE_EBADF);
 319                         break;
 320                 case EINVAL:
 321                         send_error(KCONSUMERD_SPLICE_EINVAL);
 322                         break;
 323                 case ENOMEM:
 324                         send_error(KCONSUMERD_SPLICE_ENOMEM);
 325                         break;
 326                 case ESPIPE:
 327                         send_error(KCONSUMERD_SPLICE_ESPIPE);
 328                         break;
 329         }
 330
 331 end:
 332         return ret;
 333 }
 334
 335 /*
 336  * read_subbuffer
 337  *
 338  * Consume data on a file descriptor and write it on a trace file
 339  */
 340 static int read_subbuffer(struct ltt_kconsumerd_fd *kconsumerd_fd)
 341 {
 342         unsigned long len;
 343         int err;
 344         long ret = 0;
 345         int infd = kconsumerd_fd->consumerd_fd;
 346
 347         DBG("In read_subbuffer");
 348         /* Get the next subbuffer */
 349         err = kernctl_get_next_subbuf(infd);
 350         if (err != 0) {
 351                 ret = errno;
 352                 perror("Reserving sub buffer failed (everything is normal, "
 353                                 "it is due to concurrency)");
 354                 goto end;
 355         }
 356
 357         /* read the whole subbuffer */
 358         err = kernctl_get_padded_subbuf_size(infd, &len);
 359         if (err != 0) {
 360                 ret = errno;
 361                 perror("Getting sub-buffer len failed.");
 362                 goto end;
 363         }
 364
 365         /* splice the subbuffer to the tracefile */
 366         ret = on_read_subbuffer(kconsumerd_fd, len);
 367         if (ret < 0) {
 368                 /*
 369                  * display the error but continue processing to try
 370                  * to release the subbuffer
 371                  */
 372                 ERR("Error splicing to tracefile");
 373         }
 374
 375         err = kernctl_put_next_subbuf(infd);
 376         if (err != 0) {
 377                 ret = errno;
 378                 if (errno == EFAULT) {
 379                         perror("Error in unreserving sub buffer\n");
 380                 } else if (errno == EIO) {
 381                         /* Should never happen with newer LTTng versions */
 382                         perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
 383                 }
 384                 goto end;
 385         }
 386
 387 end:
 388         return ret;
 389 }
 390
 391 /*
 392  * change_fd_state
 393  *
 394  * Update a fd according to what we just received
 395  */
 396 static void change_fd_state(int sessiond_fd,
 397                 enum lttcomm_kconsumerd_fd_state state)
 398 {
 399         struct ltt_kconsumerd_fd *iter;
 400         cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
 401                 if (iter->sessiond_fd == sessiond_fd) {
 402                         iter->state = state;
 403                         break;
 404                 }
 405         }
 406 }
 407
 408 /*
 409  * consumerd_recv_fd
 410  *
 411  * Receives an array of file descriptors and the associated
 412  * structures describing each fd (path name).
 413  * Returns the size of received data
 414  */
 415 static int consumerd_recv_fd(int sfd, int size,
 416                 enum lttcomm_consumerd_command cmd_type)
 417 {
 418         struct msghdr msg;
 419         struct iovec iov[1];
 420         int ret, i;
 421         struct cmsghdr *cmsg;
 422         int nb_fd;
 423         char tmp[CMSG_SPACE(size)];
 424         struct lttcomm_kconsumerd_msg *buf;
 425         /* the number of fds we are about to receive */
 426         nb_fd = size/sizeof(struct lttcomm_kconsumerd_msg);
 427
 428         buf = malloc(size);
 429
 430         memset(&msg, 0, sizeof(msg));
 431
 432         /* Prepare to receive the structures */
 433         iov[0].iov_base = buf;
 434         iov[0].iov_len = size;
 435         msg.msg_iov = iov;
 436         msg.msg_iovlen = 1;
 437
 438         msg.msg_control = tmp;
 439         msg.msg_controllen = sizeof(tmp);
 440
 441         DBG("Waiting to receive fds");
 442         if ((ret = recvmsg(sfd, &msg, 0)) < 0) {
 443                 perror("recvmsg");
 444         }
 445         if (ret != size) {
 446                 ERR("Received only %d, expected %d", ret, size);
 447                 send_error(KCONSUMERD_ERROR_RECV_FD);
 448                 goto end;
 449         }
 450
 451         cmsg = CMSG_FIRSTHDR(&msg);
 452         if (!cmsg) {
 453                 ERR("Invalid control message header");
 454                 ret = -1;
 455                 send_error(KCONSUMERD_ERROR_RECV_FD);
 456                 goto end;
 457         }
 458
 459         /* if we received fds */
 460         if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
 461                 DBG("Receive : expecting %d fds", nb_fd);
 462                 for (i = 0; i < nb_fd; i++) {
 463                         switch (cmd_type) {
 464                                 case LTTCOMM_ADD_STREAM:
 465                                         DBG("add_fd %s (%d)", buf[i].path_name, ((int *)CMSG_DATA(cmsg))[i]);
 466                                         ret = add_fd(&buf[i], ((int *)CMSG_DATA(cmsg))[i]);
 467                                         if (ret < 0) {
 468                                                 send_error(KCONSUMERD_OUTFD_ERROR);
 469                                                 goto end;
 470                                         }
 471                                         break;
 472                                 case LTTCOMM_UPDATE_STREAM:
 473                                         change_fd_state(buf[i].fd, buf[i].state);
 474                                         break;
 475                                 default:
 476                                         break;
 477                         }
 478                 }
 479                 /* flag to tell the polling thread to update its fd array */
 480                 update_fd_array = 1;
 481                 send_error(KCONSUMERD_SUCCESS_RECV_FD);
 482         } else {
 483                 ERR("Didn't received any fd");
 484                 send_error(KCONSUMERD_ERROR_RECV_FD);
 485                 ret = -1;
 486                 goto end;
 487         }
 488
 489 end:
 490         if (buf != NULL) {
 491                 free(buf);
 492                 buf = NULL;
 493         }
 494         return ret;
 495 }
 496
 497 /*
 498  *  thread_receive_fds
 499  *
 500  *  This thread listens on the consumerd socket and
 501  *  receives the file descriptors from ltt-sessiond
 502  */
 503 static void *thread_receive_fds(void *data)
 504 {
 505         int sock, client_socket, ret;
 506         struct lttcomm_kconsumerd_header tmp;
 507
 508         DBG("Creating command socket %s", command_sock_path);
 509         unlink(command_sock_path);
 510         client_socket = lttcomm_create_unix_sock(command_sock_path);
 511         if (client_socket < 0) {
 512                 ERR("Cannot create command socket");
 513                 goto error;
 514         }
 515
 516         ret = lttcomm_listen_unix_sock(client_socket);
 517         if (ret < 0) {
 518                 goto error;
 519         }
 520
 521         DBG("Sending ready command to ltt-sessiond");
 522         ret = send_error(KCONSUMERD_COMMAND_SOCK_READY);
 523         if (ret < 0) {
 524                 ERR("Error sending ready command to ltt-sessiond");
 525                 goto error;
 526         }
 527
 528         while (1) {
 529                 /* Blocking call, waiting for transmission */
 530                 sock = lttcomm_accept_unix_sock(client_socket);
 531                 if (sock <= 0) {
 532                         continue;
 533                 }
 534
 535                 /* We first get the number of fd we are about to receive */
 536                 ret = lttcomm_recv_unix_sock(sock, &tmp,
 537                                 sizeof(struct lttcomm_kconsumerd_header));
 538                 if (ret < 0) {
 539                         ERR("Receiving the lttcomm_kconsumerd_header");
 540                         continue;
 541                 }
 542                 ret = consumerd_recv_fd(sock, tmp.payload_size, tmp.cmd_type);
 543                 if (ret < 0) {
 544                         continue;
 545                 }
 546         }
 547
 548 error:
 549         return NULL;
 550 }
 551
 552 /*
 553  * update_poll_array
 554  *
 555  * Allocate the pollfd structure and the local view of the out fds
 556  * to avoid doing a lookup in the linked list and concurrency issues
 557  * when writing is needed.
 558  * Returns the number of fds in the structures
 559  */
 560 static int update_poll_array(struct pollfd **pollfd,
 561                 struct ltt_kconsumerd_fd **local_kconsumerd_fd)
 562 {
 563         struct ltt_kconsumerd_fd *iter;
 564         int i = 0;
 565
 566         if (*pollfd != NULL) {
 567                 free(*pollfd);
 568                 *pollfd = NULL;
 569         }
 570
 571         if (*local_kconsumerd_fd != NULL) {
 572                 free(*local_kconsumerd_fd);
 573                 *local_kconsumerd_fd = NULL;
 574         }
 575
 576         *pollfd = malloc(fds_count * sizeof(struct pollfd));
 577         if (*pollfd == NULL) {
 578                 perror("pollfd malloc");
 579                 goto error_mem;
 580         }
 581
 582         *local_kconsumerd_fd = malloc(fds_count * sizeof(struct ltt_kconsumerd_fd));
 583         if (*local_kconsumerd_fd == NULL) {
 584                 perror("local_kconsumerd_fd malloc");
 585                 goto error_mem;
 586         }
 587
 588         DBG("Updating poll fd array");
 589         pthread_mutex_lock(&kconsumerd_lock_fds);
 590
 591         cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
 592                 DBG("Inside for each");
 593                 if (iter->state == ACTIVE_FD) {
 594                         DBG("Active FD %d", iter->consumerd_fd);
 595                         pollfd[i]->fd = iter->consumerd_fd;
 596                         pollfd[i]->events = POLLIN | POLLPRI;
 597                         local_kconsumerd_fd[i] = iter;
 598                         i++;
 599                 } else if (iter->state == DELETE_FD) {
 600                         del_fd(iter);
 601                 }
 602         }
 603         update_fd_array = 0;
 604         pthread_mutex_unlock(&kconsumerd_lock_fds);
 605         return i;
 606
 607 error_mem:
 608         return -ENOMEM;
 609 }
 610
 611 /*
 612  *  thread_poll_fds
 613  *
 614  *  This thread polls the fds in the ltt_fd_list to consume the data
 615  *  and write it to tracefile if necessary.
 616  */
 617 static void *thread_poll_fds(void *data)
 618 {
 619         int num_rdy, num_hup, high_prio, ret, i;
 620         struct pollfd *pollfd = NULL;
 621         /* local view of the fds */
 622         struct ltt_kconsumerd_fd *local_kconsumerd_fd = NULL;
 623         /* local view of fds_count */
 624         int nb_fd = 0;
 625
 626         ret = pipe(thread_pipe);
 627         if (ret < 0) {
 628                 perror("Error creating pipe");
 629                 goto end;
 630         }
 631
 632         while (1) {
 633                 high_prio = 0;
 634                 num_hup = 0;
 635
 636                 /*
 637                  * the ltt_fd_list has been updated, we need to update our
 638                  * local array as well
 639                  */
 640                 if (update_fd_array) {
 641                         ret = update_poll_array(&pollfd, &local_kconsumerd_fd);
 642                         if (ret < 0) {
 643                                 ERR("Error in allocating pollfd or local_outfds");
 644                                 send_error(KCONSUMERD_POLL_ERROR);
 645                                 goto end;
 646                         }
 647                         nb_fd = ret;
 648                 }
 649
 650                 /* poll on the array of fds */
 651                 DBG("polling on %d fd", nb_fd);
 652                 num_rdy = poll(pollfd, nb_fd, POLL_TIMEOUT);
 653                 DBG("poll num_rdy : %d", num_rdy);
 654                 if (num_rdy == -1) {
 655                         perror("Poll error");
 656                         send_error(KCONSUMERD_POLL_ERROR);
 657                         goto end;
 658                 }
 659
 660                 /* Take care of high priority channels first. */
 661                 for (i = 0; i < nb_fd; i++) {
 662                         switch(pollfd[i].revents) {
 663                                 case POLLERR:
 664                                         ERR("Error returned in polling fd %d.", pollfd[i].fd);
 665                                         num_hup++;
 666                                         send_error(KCONSUMERD_POLL_ERROR);
 667                                         break;
 668                                 case POLLHUP:
 669                                         ERR("Polling fd %d tells it has hung up.", pollfd[i].fd);
 670                                         num_hup++;
 671                                         break;
 672                                 case POLLNVAL:
 673                                         ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
 674                                         send_error(KCONSUMERD_POLL_NVAL);
 675                                         num_hup++;
 676                                         break;
 677                                 case POLLPRI:
 678                                         DBG("Urgent read on fd %d", pollfd[i].fd);
 679                                         high_prio = 1;
 680                                         ret = read_subbuffer(&local_kconsumerd_fd[i]);
 681                                         /* it's ok to have an unavailable sub-buffer (FIXME : is it ?) */
 682                                         if (ret == EAGAIN) {
 683                                                 ret = 0;
 684                                         }
 685                                         break;
 686                         }
 687                 }
 688
 689                 /* If every buffer FD has hung up, we end the read loop here */
 690                 if (nb_fd > 0 && num_hup == nb_fd) {
 691                         DBG("every buffer FD has hung up\n");
 692                         send_error(KCONSUMERD_POLL_HUP);
 693                         continue;
 694                 }
 695
 696                 /* Take care of low priority channels. */
 697                 if (!high_prio) {
 698                         for (i = 0; i < nb_fd; i++) {
 699                                 switch(pollfd[i].revents) {
 700                                         case POLLIN:
 701                                                 DBG("Normal read on fd %d", pollfd[i].fd);
 702                                                 ret = read_subbuffer(&local_kconsumerd_fd[i]);
 703                                                 /* it's ok to have an unavailable subbuffer (FIXME : is it ?) */
 704                                                 if (ret == EAGAIN) {
 705                                                         ret = 0;
 706                                                 }
 707                                                 break;
 708                                 }
 709                         }
 710                 }
 711         }
 712 end:
 713         if (pollfd != NULL) {
 714                 free(pollfd);
 715                 pollfd = NULL;
 716         }
 717         if (local_kconsumerd_fd != NULL) {
 718                 free(local_kconsumerd_fd);
 719                 local_kconsumerd_fd = NULL;
 720         }
 721         return NULL;
 722 }
 723
 724 /*
 725  * usage function on stderr
 726  */
 727 static void usage(void)
 728 {
 729         fprintf(stderr, "Usage: %s OPTIONS\n\nOptions:\n", progname);
 730         fprintf(stderr, "  -h, --help                         "
 731                         "Display this usage.\n");
 732         fprintf(stderr, "  -c, --kconsumerd-cmd-sock PATH     "
 733                         "Specify path for the command socket\n");
 734         fprintf(stderr, "  -e, --kconsumerd-err-sock PATH     "
 735                         "Specify path for the error socket\n");
 736         fprintf(stderr, "  -d, --daemonize                    "
 737                         "Start as a daemon.\n");
 738         fprintf(stderr, "  -q, --quiet                        "
 739                         "No output at all.\n");
 740         fprintf(stderr, "  -v, --verbose                      "
 741                         "Verbose mode. Activate DBG() macro.\n");
 742         fprintf(stderr, "  -V, --version                      "
 743                         "Show version number.\n");
 744 }
 745
 746 /*
 747  * daemon argument parsing
 748  */
 749 static void parse_args(int argc, char **argv)
 750 {
 751         int c;
 752
 753         static struct option long_options[] = {
 754                 { "kconsumerd-cmd-sock", 1, 0, 'c' },
 755                 { "kconsumerd-err-sock", 1, 0, 'e' },
 756                 { "daemonize", 0, 0, 'd' },
 757                 { "help", 0, 0, 'h' },
 758                 { "quiet", 0, 0, 'q' },
 759                 { "verbose", 0, 0, 'v' },
 760                 { "version", 0, 0, 'V' },
 761                 { NULL, 0, 0, 0 }
 762         };
 763
 764         while (1) {
 765                 int option_index = 0;
 766                 c = getopt_long(argc, argv, "dhqvV" "c:e:", long_options, &option_index);
 767                 if (c == -1) {
 768                         break;
 769                 }
 770
 771                 switch (c) {
 772                         case 0:
 773                                 fprintf(stderr, "option %s", long_options[option_index].name);
 774                                 if (optarg) {
 775                                         fprintf(stderr, " with arg %s\n", optarg);
 776                                 }
 777                                 break;
 778                         case 'c':
 779                                 snprintf(command_sock_path, PATH_MAX, "%s", optarg);
 780                                 break;
 781                         case 'e':
 782                                 snprintf(error_sock_path, PATH_MAX, "%s", optarg);
 783                                 break;
 784                         case 'd':
 785                                 opt_daemon = 1;
 786                                 break;
 787                         case 'h':
 788                                 usage();
 789                                 exit(EXIT_FAILURE);
 790                         case 'q':
 791                                 opt_quiet = 1;
 792                                 break;
 793                         case 'v':
 794                                 opt_verbose = 1;
 795                                 break;
 796                         case 'V':
 797                                 fprintf(stdout, "%s\n", VERSION);
 798                                 exit(EXIT_SUCCESS);
 799                         default:
 800                                 usage();
 801                                 exit(EXIT_FAILURE);
 802                 }
 803         }
 804 }
 805
 806
 807 /*
 808  * main
 809  */
 810 int main(int argc, char **argv)
 811 {
 812         int i;
 813         int ret = 0;
 814         void *status;
 815
 816         /* Parse arguments */
 817         progname = argv[0];
 818         parse_args(argc, argv);
 819
 820         /* Daemonize */
 821         if (opt_daemon) {
 822                 ret = daemon(0, 0);
 823                 if (ret < 0) {
 824                         perror("daemon");
 825                         goto error;
 826                 }
 827         }
 828
 829         if (strlen(command_sock_path) == 0) {
 830                 snprintf(command_sock_path, PATH_MAX,
 831                                 KCONSUMERD_CMD_SOCK_PATH);
 832         }
 833         if (strlen(error_sock_path) == 0) {
 834                 snprintf(error_sock_path, PATH_MAX,
 835                                 KCONSUMERD_ERR_SOCK_PATH);
 836         }
 837
 838         if (set_signal_handler() < 0) {
 839                 goto error;
 840         }
 841
 842         /* Connect to the socket created by ltt-sessiond to report errors */
 843         DBG("Connecting to error socket %s", error_sock_path);
 844         error_socket = lttcomm_connect_unix_sock(error_sock_path);
 845         /* not a fatal error, but all communication with ltt-sessiond will fail */
 846         if (error_socket < 0) {
 847                 WARN("Cannot connect to error socket, is ltt-sessiond started ?");
 848         }
 849
 850         /* Create the thread to manage the receive of fd */
 851         ret = pthread_create(&threads[0], NULL, thread_receive_fds, (void *) NULL);
 852         if (ret != 0) {
 853                 perror("pthread_create");
 854                 goto error;
 855         }
 856
 857         /* Create thread to manage the polling/writing of traces */
 858         ret = pthread_create(&threads[1], NULL, thread_poll_fds, (void *) NULL);
 859         if (ret != 0) {
 860                 perror("pthread_create");
 861                 goto error;
 862         }
 863
 864         for (i = 0; i < 2; i++) {
 865                 ret = pthread_join(threads[i], &status);
 866                 if (ret != 0) {
 867                         perror("pthread_join");
 868                         goto error;
 869                 }
 870         }
 871         ret = EXIT_SUCCESS;
 872         send_error(KCONSUMERD_EXIT_SUCCESS);
 873         goto end;
 874
 875 error:
 876         ret = EXIT_FAILURE;
 877         send_error(KCONSUMERD_EXIT_FAILURE);
 878
 879 end:
 880         cleanup();
 881
 882         return ret;
 883 }