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