2e53f9a2dc00b5a123641e0a2314ec0418c78c87
[lttng-tools.git] / 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, 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
255 /* the number of fds we are about to receive */
256 nb_fd = size / sizeof(struct lttcomm_kconsumerd_msg);
257
258 /*
259 * nb_fd is the number of fds we receive. One fd per recvmsg.
260 */
261 for (i = 0; i < nb_fd; i++) {
262 struct msghdr msg = { 0 };
263
264 /* Prepare to receive the structures */
265 iov[0].iov_base = &lkm;
266 iov[0].iov_len = sizeof(lkm);
267 msg.msg_iov = iov;
268 msg.msg_iovlen = 1;
269
270 msg.msg_control = recv_fd;
271 msg.msg_controllen = sizeof(recv_fd);
272
273 DBG("Waiting to receive fd");
274 if (lttng_kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
275 goto end;
276 }
277
278 if ((ret = recvmsg(sfd, &msg, 0)) < 0) {
279 perror("recvmsg");
280 continue;
281 }
282
283 if (ret != (size / nb_fd)) {
284 ERR("Received only %d, expected %d", ret, size);
285 lttng_kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD);
286 goto end;
287 }
288
289 cmsg = CMSG_FIRSTHDR(&msg);
290 if (!cmsg) {
291 ERR("Invalid control message header");
292 ret = -1;
293 lttng_kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD);
294 goto end;
295 }
296
297 /* if we received fds */
298 if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
299 switch (cmd_type) {
300 case ADD_STREAM:
301 DBG("kconsumerd_add_fd %s (%d)", lkm.path_name,
302 ((int *) CMSG_DATA(cmsg))[0]);
303
304 new_fd = kconsumerd_allocate_fd(&lkm, ((int *) CMSG_DATA(cmsg))[0]);
305 if (new_fd == NULL) {
306 lttng_kconsumerd_send_error(ctx, KCONSUMERD_OUTFD_ERROR);
307 goto end;
308 }
309
310 if (ctx->on_recv_fd != NULL) {
311 ret = ctx->on_recv_fd(new_fd);
312 if (ret == 0) {
313 kconsumerd_add_fd(new_fd);
314 } else if (ret < 0) {
315 goto end;
316 }
317 } else {
318 kconsumerd_add_fd(new_fd);
319 }
320 break;
321 case UPDATE_STREAM:
322 if (ctx->on_update_fd != NULL) {
323 ret = ctx->on_update_fd(lkm.fd, lkm.state);
324 if (ret == 0) {
325 kconsumerd_change_fd_state(lkm.fd, lkm.state);
326 } else if (ret < 0) {
327 goto end;
328 }
329 } else {
330 kconsumerd_change_fd_state(lkm.fd, lkm.state);
331 }
332 break;
333 default:
334 break;
335 }
336 /* signal the poll thread */
337 tmp2 = write(ctx->kconsumerd_poll_pipe[1], "4", 1);
338 if (tmp2 < 0) {
339 perror("write kconsumerd poll");
340 }
341 } else {
342 ERR("Didn't received any fd");
343 lttng_kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD);
344 ret = -1;
345 goto end;
346 }
347 }
348
349 end:
350 return ret;
351 }
352
353 /*
354 * Set the error socket.
355 */
356 void lttng_kconsumerd_set_error_sock(
357 struct lttng_kconsumerd_local_data *ctx, int sock)
358 {
359 ctx->kconsumerd_error_socket = sock;
360 }
361
362 /*
363 * Set the command socket path.
364 */
365
366 void lttng_kconsumerd_set_command_sock_path(
367 struct lttng_kconsumerd_local_data *ctx, char *sock)
368 {
369 ctx->kconsumerd_command_sock_path = sock;
370 }
371
372 static void lttng_kconsumerd_sync_trace_file(
373 struct lttng_kconsumerd_fd *kconsumerd_fd, off_t orig_offset)
374 {
375 int outfd = kconsumerd_fd->out_fd;
376 /*
377 * This does a blocking write-and-wait on any page that belongs to the
378 * subbuffer prior to the one we just wrote.
379 * Don't care about error values, as these are just hints and ways to
380 * limit the amount of page cache used.
381 */
382 if (orig_offset >= kconsumerd_fd->max_sb_size) {
383 sync_file_range(outfd, orig_offset - kconsumerd_fd->max_sb_size,
384 kconsumerd_fd->max_sb_size,
385 SYNC_FILE_RANGE_WAIT_BEFORE
386 | SYNC_FILE_RANGE_WRITE
387 | SYNC_FILE_RANGE_WAIT_AFTER);
388 /*
389 * Give hints to the kernel about how we access the file:
390 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
391 * we write it.
392 *
393 * We need to call fadvise again after the file grows because the
394 * kernel does not seem to apply fadvise to non-existing parts of the
395 * file.
396 *
397 * Call fadvise _after_ having waited for the page writeback to
398 * complete because the dirty page writeback semantic is not well
399 * defined. So it can be expected to lead to lower throughput in
400 * streaming.
401 */
402 posix_fadvise(outfd, orig_offset - kconsumerd_fd->max_sb_size,
403 kconsumerd_fd->max_sb_size, POSIX_FADV_DONTNEED);
404 }
405 }
406
407
408 /*
409 * Mmap the ring buffer, read it and write the data to the tracefile.
410 *
411 * Returns the number of bytes written
412 */
413 int lttng_kconsumerd_on_read_subbuffer_mmap(
414 struct lttng_kconsumerd_local_data *ctx,
415 struct lttng_kconsumerd_fd *kconsumerd_fd, unsigned long len)
416 {
417 unsigned long mmap_offset;
418 long ret = 0;
419 off_t orig_offset = kconsumerd_fd->out_fd_offset;
420 int fd = kconsumerd_fd->consumerd_fd;
421 int outfd = kconsumerd_fd->out_fd;
422
423 /* get the offset inside the fd to mmap */
424 ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
425 if (ret != 0) {
426 ret = errno;
427 perror("kernctl_get_mmap_read_offset");
428 goto end;
429 }
430
431 while (len > 0) {
432 ret = write(outfd, kconsumerd_fd->mmap_base + mmap_offset, len);
433 if (ret >= len) {
434 len = 0;
435 } else if (ret < 0) {
436 ret = errno;
437 perror("Error in file write");
438 goto end;
439 }
440 /* This won't block, but will start writeout asynchronously */
441 sync_file_range(outfd, kconsumerd_fd->out_fd_offset, ret,
442 SYNC_FILE_RANGE_WRITE);
443 kconsumerd_fd->out_fd_offset += ret;
444 }
445
446 lttng_kconsumerd_sync_trace_file(kconsumerd_fd, orig_offset);
447
448 goto end;
449
450 end:
451 return ret;
452 }
453
454 /*
455 * Splice the data from the ring buffer to the tracefile.
456 *
457 * Returns the number of bytes spliced.
458 */
459 int lttng_kconsumerd_on_read_subbuffer_splice(
460 struct lttng_kconsumerd_local_data *ctx,
461 struct lttng_kconsumerd_fd *kconsumerd_fd, unsigned long len)
462 {
463 long ret = 0;
464 loff_t offset = 0;
465 off_t orig_offset = kconsumerd_fd->out_fd_offset;
466 int fd = kconsumerd_fd->consumerd_fd;
467 int outfd = kconsumerd_fd->out_fd;
468
469 while (len > 0) {
470 DBG("splice chan to pipe offset %lu (fd : %d)",
471 (unsigned long)offset, fd);
472 ret = splice(fd, &offset, ctx->kconsumerd_thread_pipe[1], NULL, len,
473 SPLICE_F_MOVE | SPLICE_F_MORE);
474 DBG("splice chan to pipe ret %ld", ret);
475 if (ret < 0) {
476 ret = errno;
477 perror("Error in relay splice");
478 goto splice_error;
479 }
480
481 ret = splice(ctx->kconsumerd_thread_pipe[0], NULL, outfd, NULL, ret,
482 SPLICE_F_MOVE | SPLICE_F_MORE);
483 DBG("splice pipe to file %ld", ret);
484 if (ret < 0) {
485 ret = errno;
486 perror("Error in file splice");
487 goto splice_error;
488 }
489 len -= ret;
490 /* This won't block, but will start writeout asynchronously */
491 sync_file_range(outfd, kconsumerd_fd->out_fd_offset, ret,
492 SYNC_FILE_RANGE_WRITE);
493 kconsumerd_fd->out_fd_offset += ret;
494 }
495 lttng_kconsumerd_sync_trace_file(kconsumerd_fd, orig_offset);
496
497 goto end;
498
499 splice_error:
500 /* send the appropriate error description to sessiond */
501 switch(ret) {
502 case EBADF:
503 lttng_kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_EBADF);
504 break;
505 case EINVAL:
506 lttng_kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_EINVAL);
507 break;
508 case ENOMEM:
509 lttng_kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_ENOMEM);
510 break;
511 case ESPIPE:
512 lttng_kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_ESPIPE);
513 break;
514 }
515
516 end:
517 return ret;
518 }
519
520 /*
521 * Take a snapshot for a specific fd
522 *
523 * Returns 0 on success, < 0 on error
524 */
525 int lttng_kconsumerd_take_snapshot(struct lttng_kconsumerd_local_data *ctx,
526 struct lttng_kconsumerd_fd *kconsumerd_fd)
527 {
528 int ret = 0;
529 int infd = kconsumerd_fd->consumerd_fd;
530
531 ret = kernctl_snapshot(infd);
532 if (ret != 0) {
533 ret = errno;
534 perror("Getting sub-buffer snapshot.");
535 }
536
537 return ret;
538 }
539
540 /*
541 * Get the produced position
542 *
543 * Returns 0 on success, < 0 on error
544 */
545 int lttng_kconsumerd_get_produced_snapshot(
546 struct lttng_kconsumerd_local_data *ctx,
547 struct lttng_kconsumerd_fd *kconsumerd_fd,
548 unsigned long *pos)
549 {
550 int ret;
551 int infd = kconsumerd_fd->consumerd_fd;
552
553 ret = kernctl_snapshot_get_produced(infd, pos);
554 if (ret != 0) {
555 ret = errno;
556 perror("kernctl_snapshot_get_produced");
557 }
558
559 return ret;
560 }
561
562 /*
563 * Poll on the should_quit pipe and the command socket return -1 on error and
564 * should exit, 0 if data is available on the command socket
565 */
566 int lttng_kconsumerd_poll_socket(struct pollfd *kconsumerd_sockpoll)
567 {
568 int num_rdy;
569
570 num_rdy = poll(kconsumerd_sockpoll, 2, -1);
571 if (num_rdy == -1) {
572 perror("Poll error");
573 goto exit;
574 }
575 if (kconsumerd_sockpoll[0].revents == POLLIN) {
576 DBG("kconsumerd_should_quit wake up");
577 goto exit;
578 }
579 return 0;
580
581 exit:
582 return -1;
583 }
584
585 /*
586 * This thread polls the fds in the ltt_fd_list to consume the data and write
587 * it to tracefile if necessary.
588 */
589 void *lttng_kconsumerd_thread_poll_fds(void *data)
590 {
591 int num_rdy, num_hup, high_prio, ret, i;
592 struct pollfd *pollfd = NULL;
593 /* local view of the fds */
594 struct lttng_kconsumerd_fd **local_kconsumerd_fd = NULL;
595 /* local view of kconsumerd_data.fds_count */
596 int nb_fd = 0;
597 char tmp;
598 int tmp2;
599 struct lttng_kconsumerd_local_data *ctx = data;
600
601
602 local_kconsumerd_fd = malloc(sizeof(struct lttng_kconsumerd_fd));
603
604 while (1) {
605 high_prio = 0;
606 num_hup = 0;
607
608 /*
609 * the ltt_fd_list has been updated, we need to update our
610 * local array as well
611 */
612 pthread_mutex_lock(&kconsumerd_data.lock);
613 if (kconsumerd_data.need_update) {
614 if (pollfd != NULL) {
615 free(pollfd);
616 pollfd = NULL;
617 }
618 if (local_kconsumerd_fd != NULL) {
619 free(local_kconsumerd_fd);
620 local_kconsumerd_fd = NULL;
621 }
622
623 /* allocate for all fds + 1 for the kconsumerd_poll_pipe */
624 pollfd = malloc((kconsumerd_data.fds_count + 1) * sizeof(struct pollfd));
625 if (pollfd == NULL) {
626 perror("pollfd malloc");
627 pthread_mutex_unlock(&kconsumerd_data.lock);
628 goto end;
629 }
630
631 /* allocate for all fds + 1 for the kconsumerd_poll_pipe */
632 local_kconsumerd_fd = malloc((kconsumerd_data.fds_count + 1) *
633 sizeof(struct lttng_kconsumerd_fd));
634 if (local_kconsumerd_fd == NULL) {
635 perror("local_kconsumerd_fd malloc");
636 pthread_mutex_unlock(&kconsumerd_data.lock);
637 goto end;
638 }
639 ret = kconsumerd_update_poll_array(ctx, &pollfd, local_kconsumerd_fd);
640 if (ret < 0) {
641 ERR("Error in allocating pollfd or local_outfds");
642 lttng_kconsumerd_send_error(ctx, KCONSUMERD_POLL_ERROR);
643 pthread_mutex_unlock(&kconsumerd_data.lock);
644 goto end;
645 }
646 nb_fd = ret;
647 kconsumerd_data.need_update = 0;
648 }
649 pthread_mutex_unlock(&kconsumerd_data.lock);
650
651 /* poll on the array of fds */
652 DBG("polling on %d fd", nb_fd + 1);
653 num_rdy = poll(pollfd, nb_fd + 1, kconsumerd_poll_timeout);
654 DBG("poll num_rdy : %d", num_rdy);
655 if (num_rdy == -1) {
656 perror("Poll error");
657 lttng_kconsumerd_send_error(ctx, KCONSUMERD_POLL_ERROR);
658 goto end;
659 } else if (num_rdy == 0) {
660 DBG("Polling thread timed out");
661 goto end;
662 }
663
664 /* No FDs and kconsumerd_quit, kconsumerd_cleanup the thread */
665 if (nb_fd == 0 && kconsumerd_quit == 1) {
666 goto end;
667 }
668
669 /*
670 * If the kconsumerd_poll_pipe triggered poll go
671 * directly to the beginning of the loop to update the
672 * array. We want to prioritize array update over
673 * low-priority reads.
674 */
675 if (pollfd[nb_fd].revents == POLLIN) {
676 DBG("kconsumerd_poll_pipe wake up");
677 tmp2 = read(ctx->kconsumerd_poll_pipe[0], &tmp, 1);
678 if (tmp2 < 0) {
679 perror("read kconsumerd poll");
680 }
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 kconsumerd_del_fd(local_kconsumerd_fd[i]);
690 num_hup++;
691 break;
692 case POLLHUP:
693 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
694 kconsumerd_del_fd(local_kconsumerd_fd[i]);
695 num_hup++;
696 break;
697 case POLLNVAL:
698 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
699 kconsumerd_del_fd(local_kconsumerd_fd[i]);
700 num_hup++;
701 break;
702 case POLLPRI:
703 DBG("Urgent read on fd %d", pollfd[i].fd);
704 high_prio = 1;
705 ret = ctx->on_buffer_ready(local_kconsumerd_fd[i]);
706 /* it's ok to have an unavailable sub-buffer */
707 if (ret == EAGAIN) {
708 ret = 0;
709 }
710 break;
711 }
712 }
713
714 /* If every buffer FD has hung up, we end the read loop here */
715 if (nb_fd > 0 && num_hup == nb_fd) {
716 DBG("every buffer FD has hung up\n");
717 if (kconsumerd_quit == 1) {
718 goto end;
719 }
720 continue;
721 }
722
723 /* Take care of low priority channels. */
724 if (high_prio == 0) {
725 for (i = 0; i < nb_fd; i++) {
726 if (pollfd[i].revents == POLLIN) {
727 DBG("Normal read on fd %d", pollfd[i].fd);
728 ret = ctx->on_buffer_ready(local_kconsumerd_fd[i]);
729 /* it's ok to have an unavailable subbuffer */
730 if (ret == EAGAIN) {
731 ret = 0;
732 }
733 }
734 }
735 }
736 }
737 end:
738 DBG("polling thread exiting");
739 if (pollfd != NULL) {
740 free(pollfd);
741 pollfd = NULL;
742 }
743 if (local_kconsumerd_fd != NULL) {
744 free(local_kconsumerd_fd);
745 local_kconsumerd_fd = NULL;
746 }
747 return NULL;
748 }
749
750 /*
751 * Initialise the necessary environnement :
752 * - create a new context
753 * - create the poll_pipe
754 * - create the should_quit pipe (for signal handler)
755 * - create the thread pipe (for splice)
756 *
757 * Takes a function pointer as argument, this function is called when data is
758 * available on a buffer. This function is responsible to do the
759 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
760 * buffer configuration and then kernctl_put_next_subbuf at the end.
761 *
762 * Returns a pointer to the new context or NULL on error.
763 */
764 struct lttng_kconsumerd_local_data *lttng_kconsumerd_create(
765 int (*buffer_ready)(struct lttng_kconsumerd_fd *kconsumerd_fd),
766 int (*recv_fd)(struct lttng_kconsumerd_fd *kconsumerd_fd),
767 int (*update_fd)(int sessiond_fd, uint32_t state))
768 {
769 int ret, i;
770 struct lttng_kconsumerd_local_data *ctx;
771
772 ctx = malloc(sizeof(struct lttng_kconsumerd_local_data));
773 if (ctx == NULL) {
774 perror("allocating context");
775 goto error;
776 }
777
778 ctx->kconsumerd_error_socket = -1;
779 /* assign the callbacks */
780 ctx->on_buffer_ready = buffer_ready;
781 ctx->on_recv_fd = recv_fd;
782 ctx->on_update_fd = update_fd;
783
784 ret = pipe(ctx->kconsumerd_poll_pipe);
785 if (ret < 0) {
786 perror("Error creating poll pipe");
787 goto error_poll_pipe;
788 }
789
790 ret = pipe(ctx->kconsumerd_should_quit);
791 if (ret < 0) {
792 perror("Error creating recv pipe");
793 goto error_quit_pipe;
794 }
795
796 ret = pipe(ctx->kconsumerd_thread_pipe);
797 if (ret < 0) {
798 perror("Error creating thread pipe");
799 goto error_thread_pipe;
800 }
801
802 return ctx;
803
804
805 error_thread_pipe:
806 for (i = 0; i < 2; i++) {
807 int err;
808
809 err = close(ctx->kconsumerd_should_quit[i]);
810 assert(!err);
811 }
812 error_quit_pipe:
813 for (i = 0; i < 2; i++) {
814 int err;
815
816 err = close(ctx->kconsumerd_poll_pipe[i]);
817 assert(!err);
818 }
819 error_poll_pipe:
820 free(ctx);
821 error:
822 return NULL;
823 }
824
825 /*
826 * Close all fds associated with the instance and free the context.
827 */
828 void lttng_kconsumerd_destroy(struct lttng_kconsumerd_local_data *ctx)
829 {
830 close(ctx->kconsumerd_error_socket);
831 close(ctx->kconsumerd_thread_pipe[0]);
832 close(ctx->kconsumerd_thread_pipe[1]);
833 close(ctx->kconsumerd_poll_pipe[0]);
834 close(ctx->kconsumerd_poll_pipe[1]);
835 close(ctx->kconsumerd_should_quit[0]);
836 close(ctx->kconsumerd_should_quit[1]);
837 unlink(ctx->kconsumerd_command_sock_path);
838 free(ctx);
839 ctx = NULL;
840 }
841
842 /*
843 * This thread listens on the consumerd socket and receives the file
844 * descriptors from the session daemon.
845 */
846 void *lttng_kconsumerd_thread_receive_fds(void *data)
847 {
848 int sock, client_socket, ret;
849 struct lttcomm_kconsumerd_header tmp;
850 /*
851 * structure to poll for incoming data on communication socket avoids
852 * making blocking sockets.
853 */
854 struct pollfd kconsumerd_sockpoll[2];
855 struct lttng_kconsumerd_local_data *ctx = data;
856
857
858 DBG("Creating command socket %s", ctx->kconsumerd_command_sock_path);
859 unlink(ctx->kconsumerd_command_sock_path);
860 client_socket = lttcomm_create_unix_sock(ctx->kconsumerd_command_sock_path);
861 if (client_socket < 0) {
862 ERR("Cannot create command socket");
863 goto end;
864 }
865
866 ret = lttcomm_listen_unix_sock(client_socket);
867 if (ret < 0) {
868 goto end;
869 }
870
871 DBG("Sending ready command to ltt-sessiond");
872 ret = lttng_kconsumerd_send_error(ctx, KCONSUMERD_COMMAND_SOCK_READY);
873 /* return < 0 on error, but == 0 is not fatal */
874 if (ret < 0) {
875 ERR("Error sending ready command to ltt-sessiond");
876 goto end;
877 }
878
879 ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
880 if (ret < 0) {
881 perror("fcntl O_NONBLOCK");
882 goto end;
883 }
884
885 /* prepare the FDs to poll : to client socket and the should_quit pipe */
886 kconsumerd_sockpoll[0].fd = ctx->kconsumerd_should_quit[0];
887 kconsumerd_sockpoll[0].events = POLLIN | POLLPRI;
888 kconsumerd_sockpoll[1].fd = client_socket;
889 kconsumerd_sockpoll[1].events = POLLIN | POLLPRI;
890
891 if (lttng_kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
892 goto end;
893 }
894 DBG("Connection on client_socket");
895
896 /* Blocking call, waiting for transmission */
897 sock = lttcomm_accept_unix_sock(client_socket);
898 if (sock <= 0) {
899 WARN("On accept");
900 goto end;
901 }
902 ret = fcntl(sock, F_SETFL, O_NONBLOCK);
903 if (ret < 0) {
904 perror("fcntl O_NONBLOCK");
905 goto end;
906 }
907
908 /* update the polling structure to poll on the established socket */
909 kconsumerd_sockpoll[1].fd = sock;
910 kconsumerd_sockpoll[1].events = POLLIN | POLLPRI;
911
912 while (1) {
913 if (lttng_kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
914 goto end;
915 }
916 DBG("Incoming fds on sock");
917
918 /* We first get the number of fd we are about to receive */
919 ret = lttcomm_recv_unix_sock(sock, &tmp,
920 sizeof(struct lttcomm_kconsumerd_header));
921 if (ret <= 0) {
922 ERR("Communication interrupted on command socket");
923 goto end;
924 }
925 if (tmp.cmd_type == STOP) {
926 DBG("Received STOP command");
927 goto end;
928 }
929 if (kconsumerd_quit) {
930 DBG("kconsumerd_thread_receive_fds received quit from signal");
931 goto end;
932 }
933
934 /* we received a command to add or update fds */
935 ret = kconsumerd_consumerd_recv_fd(ctx, sock, kconsumerd_sockpoll,
936 tmp.payload_size, tmp.cmd_type);
937 if (ret < 0) {
938 ERR("Receiving the FD, exiting");
939 goto end;
940 }
941 DBG("received fds on sock");
942 }
943
944 end:
945 DBG("kconsumerd_thread_receive_fds exiting");
946
947 /*
948 * when all fds have hung up, the polling thread
949 * can exit cleanly
950 */
951 kconsumerd_quit = 1;
952
953 /*
954 * 2s of grace period, if no polling events occur during
955 * this period, the polling thread will exit even if there
956 * are still open FDs (should not happen, but safety mechanism).
957 */
958 kconsumerd_poll_timeout = LTTNG_KCONSUMERD_POLL_GRACE_PERIOD;
959
960 /* wake up the polling thread */
961 ret = write(ctx->kconsumerd_poll_pipe[1], "4", 1);
962 if (ret < 0) {
963 perror("poll pipe write");
964 }
965 return NULL;
966 }
967
968 /*
969 * Close all the tracefiles and stream fds, should be called when all instances
970 * are destroyed.
971 */
972 void lttng_kconsumerd_cleanup(void)
973 {
974 struct lttng_kconsumerd_fd *iter, *tmp;
975
976 /*
977 * close all outfd. Called when there are no more threads
978 * running (after joining on the threads), no need to protect
979 * list iteration with mutex.
980 */
981 cds_list_for_each_entry_safe(iter, tmp,
982 &kconsumerd_data.fd_list.head, list) {
983 kconsumerd_del_fd(iter);
984 }
985 }
986
987 /*
988 * Called from signal handler.
989 */
990 void lttng_kconsumerd_should_exit(struct lttng_kconsumerd_local_data *ctx)
991 {
992 int ret;
993 kconsumerd_quit = 1;
994 ret = write(ctx->kconsumerd_should_quit[1], "4", 1);
995 if (ret < 0) {
996 perror("write kconsumerd quit");
997 }
998 }
999
1000 /*
1001 * Send return code to the session daemon.
1002 * If the socket is not defined, we return 0, it is not a fatal error
1003 */
1004 int lttng_kconsumerd_send_error(
1005 struct lttng_kconsumerd_local_data *ctx, int cmd)
1006 {
1007 if (ctx->kconsumerd_error_socket > 0) {
1008 return lttcomm_send_unix_sock(ctx->kconsumerd_error_socket, &cmd,
1009 sizeof(enum lttcomm_sessiond_command));
1010 }
1011
1012 return 0;
1013 }
This page took 0.081179 seconds and 3 git commands to generate.