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