Fix: consumer: put each chunk on teardown
[lttng-tools.git] / src / bin / lttng-sessiond / notify-apps.c
1 /*
2 * Copyright (C) 2013 - David Goulet <dgoulet@efficios.com>
3 *
4 * This program is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License, version 2 only, as
6 * published by the Free Software Foundation.
7 *
8 * This program is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
11 * more details.
12 *
13 * You should have received a copy of the GNU General Public License along with
14 * this program; if not, write to the Free Software Foundation, Inc., 51
15 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
16 */
17
18 #define _LGPL_SOURCE
19 #include <assert.h>
20
21 #include <common/common.h>
22 #include <common/utils.h>
23
24 #include "fd-limit.h"
25 #include "lttng-sessiond.h"
26 #include "notify-apps.h"
27 #include "health-sessiond.h"
28 #include "testpoint.h"
29 #include "utils.h"
30 #include "thread.h"
31
32 struct thread_notifiers {
33 struct lttng_pipe *quit_pipe;
34 int apps_cmd_notify_pipe_read_fd;
35 };
36
37 /*
38 * This thread manage application notify communication.
39 */
40 static void *thread_application_notification(void *data)
41 {
42 int i, ret, pollfd, err = -1;
43 ssize_t size_ret;
44 uint32_t revents, nb_fd;
45 struct lttng_poll_event events;
46 struct thread_notifiers *notifiers = data;
47 const int quit_pipe_read_fd = lttng_pipe_get_readfd(notifiers->quit_pipe);
48
49 DBG("[ust-thread] Manage application notify command");
50
51 rcu_register_thread();
52 rcu_thread_online();
53
54 health_register(health_sessiond,
55 HEALTH_SESSIOND_TYPE_APP_MANAGE_NOTIFY);
56
57 if (testpoint(sessiond_thread_app_manage_notify)) {
58 goto error_testpoint;
59 }
60
61 health_code_update();
62
63 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
64 if (ret < 0) {
65 goto error_poll_create;
66 }
67
68 /* Add notify pipe to the pollset. */
69 ret = lttng_poll_add(&events, notifiers->apps_cmd_notify_pipe_read_fd,
70 LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
71 if (ret < 0) {
72 goto error;
73 }
74
75 ret = lttng_poll_add(&events, quit_pipe_read_fd,
76 LPOLLIN | LPOLLERR);
77 if (ret < 0) {
78 goto error;
79 }
80
81 health_code_update();
82
83 while (1) {
84 DBG3("[ust-thread] Manage notify polling");
85
86 /* Inifinite blocking call, waiting for transmission */
87 restart:
88 health_poll_entry();
89 ret = lttng_poll_wait(&events, -1);
90 DBG3("[ust-thread] Manage notify return from poll on %d fds",
91 LTTNG_POLL_GETNB(&events));
92 health_poll_exit();
93 if (ret < 0) {
94 /*
95 * Restart interrupted system call.
96 */
97 if (errno == EINTR) {
98 goto restart;
99 }
100 goto error;
101 }
102
103 nb_fd = ret;
104
105 for (i = 0; i < nb_fd; i++) {
106 health_code_update();
107
108 /* Fetch once the poll data */
109 revents = LTTNG_POLL_GETEV(&events, i);
110 pollfd = LTTNG_POLL_GETFD(&events, i);
111
112 /* Thread quit pipe has been closed. Killing thread. */
113 if (pollfd == quit_pipe_read_fd) {
114 err = 0;
115 goto exit;
116 } else if (pollfd == notifiers->apps_cmd_notify_pipe_read_fd) {
117 /* Inspect the apps cmd pipe */
118 int sock;
119
120 if (revents & LPOLLIN) {
121 /* Get socket from dispatch thread. */
122 size_ret = lttng_read(notifiers->apps_cmd_notify_pipe_read_fd,
123 &sock, sizeof(sock));
124 if (size_ret < sizeof(sock)) {
125 PERROR("read apps notify pipe");
126 goto error;
127 }
128 health_code_update();
129
130 ret = lttng_poll_add(&events, sock,
131 LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
132 if (ret < 0) {
133 /*
134 * It's possible we've reached the max poll fd allowed.
135 * Let's close the socket but continue normal execution.
136 */
137 ret = close(sock);
138 if (ret) {
139 PERROR("close notify socket %d", sock);
140 }
141 lttng_fd_put(LTTNG_FD_APPS, 1);
142 continue;
143 }
144 DBG3("UST thread notify added sock %d to pollset", sock);
145 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
146 ERR("Apps notify command pipe error");
147 goto error;
148 } else {
149 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
150 goto error;
151 }
152 } else {
153 /*
154 * At this point, we know that a registered application
155 * triggered the event.
156 */
157 if (revents & (LPOLLIN | LPOLLPRI)) {
158 ret = ust_app_recv_notify(pollfd);
159 if (ret < 0) {
160 /* Removing from the poll set */
161 ret = lttng_poll_del(&events, pollfd);
162 if (ret < 0) {
163 goto error;
164 }
165
166 /* The socket is closed after a grace period here. */
167 ust_app_notify_sock_unregister(pollfd);
168 }
169 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
170 /* Removing from the poll set */
171 ret = lttng_poll_del(&events, pollfd);
172 if (ret < 0) {
173 goto error;
174 }
175
176 /* The socket is closed after a grace period here. */
177 ust_app_notify_sock_unregister(pollfd);
178 } else {
179 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
180 goto error;
181 }
182 health_code_update();
183 }
184 }
185 }
186
187 exit:
188 error:
189 lttng_poll_clean(&events);
190 error_poll_create:
191 error_testpoint:
192
193 DBG("Application notify communication apps thread cleanup complete");
194 if (err) {
195 health_error();
196 ERR("Health error occurred in %s", __func__);
197 }
198 health_unregister(health_sessiond);
199 rcu_thread_offline();
200 rcu_unregister_thread();
201 return NULL;
202 }
203
204 static bool shutdown_application_notification_thread(void *data)
205 {
206 struct thread_notifiers *notifiers = data;
207 const int write_fd = lttng_pipe_get_writefd(notifiers->quit_pipe);
208
209 return notify_thread_pipe(write_fd) == 1;
210 }
211
212 static void cleanup_application_notification_thread(void *data)
213 {
214 struct thread_notifiers *notifiers = data;
215
216 lttng_pipe_destroy(notifiers->quit_pipe);
217 free(notifiers);
218 }
219
220 bool launch_application_notification_thread(int apps_cmd_notify_pipe_read_fd)
221 {
222 struct lttng_thread *thread;
223 struct thread_notifiers *notifiers;
224 struct lttng_pipe *quit_pipe;
225
226 notifiers = zmalloc(sizeof(*notifiers));
227 if (!notifiers) {
228 goto error_alloc;
229 }
230 notifiers->apps_cmd_notify_pipe_read_fd = apps_cmd_notify_pipe_read_fd;
231
232 quit_pipe = lttng_pipe_open(FD_CLOEXEC);
233 if (!quit_pipe) {
234 goto error;
235 }
236 notifiers->quit_pipe = quit_pipe;
237
238 thread = lttng_thread_create("Application notification",
239 thread_application_notification,
240 shutdown_application_notification_thread,
241 cleanup_application_notification_thread,
242 notifiers);
243 if (!thread) {
244 goto error;
245 }
246 lttng_thread_put(thread);
247 return true;
248 error:
249 cleanup_application_notification_thread(notifiers);
250 error_alloc:
251 return false;
252 }
This page took 0.03656 seconds and 5 git commands to generate.