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