lttng-ctl: notifications: use epoll()/poll() instead of select()
[lttng-tools.git] / src / bin / lttng-sessiond / kernel-consumer.c
1 /*
2 * Copyright (C) 2012 - 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 <stdio.h>
20 #include <stdlib.h>
21 #include <sys/stat.h>
22 #include <unistd.h>
23 #include <inttypes.h>
24
25 #include <common/common.h>
26 #include <common/defaults.h>
27 #include <common/compat/string.h>
28
29 #include "consumer.h"
30 #include "health-sessiond.h"
31 #include "kernel-consumer.h"
32 #include "notification-thread-commands.h"
33 #include "session.h"
34 #include "lttng-sessiond.h"
35
36 static char *create_channel_path(struct consumer_output *consumer)
37 {
38 int ret;
39 char tmp_path[PATH_MAX];
40 char *pathname = NULL;
41
42 assert(consumer);
43
44 /* Get the right path name destination */
45 if (consumer->type == CONSUMER_DST_LOCAL ||
46 (consumer->type == CONSUMER_DST_NET &&
47 consumer->relay_major_version == 2 &&
48 consumer->relay_minor_version >= 11)) {
49 pathname = strdup(consumer->domain_subdir);
50 if (!pathname) {
51 PERROR("Failed to copy domain subdirectory string %s",
52 consumer->domain_subdir);
53 goto error;
54 }
55 DBG3("Kernel local consumer trace path relative to current trace chunk: \"%s\"",
56 pathname);
57 } else {
58 /* Network output. */
59 ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s",
60 consumer->dst.net.base_dir,
61 consumer->domain_subdir);
62 if (ret < 0) {
63 PERROR("snprintf kernel metadata path");
64 goto error;
65 } else if (ret >= sizeof(tmp_path)) {
66 ERR("Kernel channel path exceeds the maximal allowed length of of %zu bytes (%i bytes required) with path \"%s%s\"",
67 sizeof(tmp_path), ret,
68 consumer->dst.net.base_dir,
69 consumer->domain_subdir);
70 goto error;
71 }
72 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
73 if (!pathname) {
74 PERROR("lttng_strndup");
75 goto error;
76 }
77 DBG3("Kernel network consumer subdir path: %s", pathname);
78 }
79
80 return pathname;
81
82 error:
83 free(pathname);
84 return NULL;
85 }
86
87 /*
88 * Sending a single channel to the consumer with command ADD_CHANNEL.
89 */
90 static
91 int kernel_consumer_add_channel(struct consumer_socket *sock,
92 struct ltt_kernel_channel *channel,
93 struct ltt_kernel_session *ksession,
94 unsigned int monitor)
95 {
96 int ret;
97 char *pathname = NULL;
98 struct lttcomm_consumer_msg lkm;
99 struct consumer_output *consumer;
100 enum lttng_error_code status;
101 struct ltt_session *session = NULL;
102 struct lttng_channel_extended *channel_attr_extended;
103 bool is_local_trace;
104
105 /* Safety net */
106 assert(channel);
107 assert(ksession);
108 assert(ksession->consumer);
109
110 consumer = ksession->consumer;
111 channel_attr_extended = (struct lttng_channel_extended *)
112 channel->channel->attr.extended.ptr;
113
114 DBG("Kernel consumer adding channel %s to kernel consumer",
115 channel->channel->name);
116 is_local_trace = consumer->net_seq_index == -1ULL;
117
118 pathname = create_channel_path(consumer);
119 if (!pathname) {
120 ret = -1;
121 goto error;
122 }
123
124 if (is_local_trace && ksession->current_trace_chunk) {
125 enum lttng_trace_chunk_status chunk_status;
126 char *pathname_index;
127
128 ret = asprintf(&pathname_index, "%s/" DEFAULT_INDEX_DIR,
129 pathname);
130 if (ret < 0) {
131 ERR("Failed to format channel index directory");
132 ret = -1;
133 goto error;
134 }
135
136 /*
137 * Create the index subdirectory which will take care
138 * of implicitly creating the channel's path.
139 */
140 chunk_status = lttng_trace_chunk_create_subdirectory(
141 ksession->current_trace_chunk, pathname_index);
142 free(pathname_index);
143 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
144 ret = -1;
145 goto error;
146 }
147 }
148
149 /* Prep channel message structure */
150 consumer_init_add_channel_comm_msg(&lkm,
151 channel->key,
152 ksession->id,
153 pathname,
154 ksession->uid,
155 ksession->gid,
156 consumer->net_seq_index,
157 channel->channel->name,
158 channel->stream_count,
159 channel->channel->attr.output,
160 CONSUMER_CHANNEL_TYPE_DATA,
161 channel->channel->attr.tracefile_size,
162 channel->channel->attr.tracefile_count,
163 monitor,
164 channel->channel->attr.live_timer_interval,
165 channel_attr_extended->monitor_timer_interval,
166 ksession->current_trace_chunk);
167
168 health_code_update();
169
170 ret = consumer_send_channel(sock, &lkm);
171 if (ret < 0) {
172 goto error;
173 }
174
175 health_code_update();
176 rcu_read_lock();
177 session = session_find_by_id(ksession->id);
178 assert(session);
179 assert(pthread_mutex_trylock(&session->lock));
180 assert(session_trylock_list());
181
182 status = notification_thread_command_add_channel(
183 notification_thread_handle, session->name,
184 ksession->uid, ksession->gid,
185 channel->channel->name, channel->key,
186 LTTNG_DOMAIN_KERNEL,
187 channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf);
188 rcu_read_unlock();
189 if (status != LTTNG_OK) {
190 ret = -1;
191 goto error;
192 }
193
194 channel->published_to_notification_thread = true;
195
196 error:
197 if (session) {
198 session_put(session);
199 }
200 free(pathname);
201 return ret;
202 }
203
204 /*
205 * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
206 *
207 * The consumer socket lock must be held by the caller.
208 */
209 int kernel_consumer_add_metadata(struct consumer_socket *sock,
210 struct ltt_kernel_session *ksession, unsigned int monitor)
211 {
212 int ret;
213 struct lttcomm_consumer_msg lkm;
214 struct consumer_output *consumer;
215
216 rcu_read_lock();
217
218 /* Safety net */
219 assert(ksession);
220 assert(ksession->consumer);
221 assert(sock);
222
223 DBG("Sending metadata %d to kernel consumer",
224 ksession->metadata_stream_fd);
225
226 /* Get consumer output pointer */
227 consumer = ksession->consumer;
228
229 /* Prep channel message structure */
230 consumer_init_add_channel_comm_msg(&lkm,
231 ksession->metadata->key,
232 ksession->id,
233 DEFAULT_KERNEL_TRACE_DIR,
234 ksession->uid,
235 ksession->gid,
236 consumer->net_seq_index,
237 DEFAULT_METADATA_NAME,
238 1,
239 DEFAULT_KERNEL_CHANNEL_OUTPUT,
240 CONSUMER_CHANNEL_TYPE_METADATA,
241 0, 0,
242 monitor, 0, 0, ksession->current_trace_chunk);
243
244 health_code_update();
245
246 ret = consumer_send_channel(sock, &lkm);
247 if (ret < 0) {
248 goto error;
249 }
250
251 health_code_update();
252
253 /* Prep stream message structure */
254 consumer_init_add_stream_comm_msg(&lkm,
255 ksession->metadata->key,
256 ksession->metadata_stream_fd,
257 0 /* CPU: 0 for metadata. */);
258
259 health_code_update();
260
261 /* Send stream and file descriptor */
262 ret = consumer_send_stream(sock, consumer, &lkm,
263 &ksession->metadata_stream_fd, 1);
264 if (ret < 0) {
265 goto error;
266 }
267
268 health_code_update();
269
270 error:
271 rcu_read_unlock();
272 return ret;
273 }
274
275 /*
276 * Sending a single stream to the consumer with command ADD_STREAM.
277 */
278 static
279 int kernel_consumer_add_stream(struct consumer_socket *sock,
280 struct ltt_kernel_channel *channel,
281 struct ltt_kernel_stream *stream,
282 struct ltt_kernel_session *session, unsigned int monitor)
283 {
284 int ret;
285 struct lttcomm_consumer_msg lkm;
286 struct consumer_output *consumer;
287
288 assert(channel);
289 assert(stream);
290 assert(session);
291 assert(session->consumer);
292 assert(sock);
293
294 DBG("Sending stream %d of channel %s to kernel consumer",
295 stream->fd, channel->channel->name);
296
297 /* Get consumer output pointer */
298 consumer = session->consumer;
299
300 /* Prep stream consumer message */
301 consumer_init_add_stream_comm_msg(&lkm,
302 channel->key,
303 stream->fd,
304 stream->cpu);
305
306 health_code_update();
307
308 /* Send stream and file descriptor */
309 ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
310 if (ret < 0) {
311 goto error;
312 }
313
314 health_code_update();
315
316 error:
317 return ret;
318 }
319
320 /*
321 * Sending the notification that all streams were sent with STREAMS_SENT.
322 */
323 int kernel_consumer_streams_sent(struct consumer_socket *sock,
324 struct ltt_kernel_session *session, uint64_t channel_key)
325 {
326 int ret;
327 struct lttcomm_consumer_msg lkm;
328 struct consumer_output *consumer;
329
330 assert(sock);
331 assert(session);
332
333 DBG("Sending streams_sent");
334 /* Get consumer output pointer */
335 consumer = session->consumer;
336
337 /* Prep stream consumer message */
338 consumer_init_streams_sent_comm_msg(&lkm,
339 LTTNG_CONSUMER_STREAMS_SENT,
340 channel_key, consumer->net_seq_index);
341
342 health_code_update();
343
344 /* Send stream and file descriptor */
345 ret = consumer_send_msg(sock, &lkm);
346 if (ret < 0) {
347 goto error;
348 }
349
350 error:
351 return ret;
352 }
353
354 /*
355 * Send all stream fds of kernel channel to the consumer.
356 *
357 * The consumer socket lock must be held by the caller.
358 */
359 int kernel_consumer_send_channel_streams(struct consumer_socket *sock,
360 struct ltt_kernel_channel *channel, struct ltt_kernel_session *ksession,
361 unsigned int monitor)
362 {
363 int ret = LTTNG_OK;
364 struct ltt_kernel_stream *stream;
365
366 /* Safety net */
367 assert(channel);
368 assert(ksession);
369 assert(ksession->consumer);
370 assert(sock);
371
372 rcu_read_lock();
373
374 /* Bail out if consumer is disabled */
375 if (!ksession->consumer->enabled) {
376 ret = LTTNG_OK;
377 goto error;
378 }
379
380 DBG("Sending streams of channel %s to kernel consumer",
381 channel->channel->name);
382
383 if (!channel->sent_to_consumer) {
384 ret = kernel_consumer_add_channel(sock, channel, ksession, monitor);
385 if (ret < 0) {
386 goto error;
387 }
388 channel->sent_to_consumer = true;
389 }
390
391 /* Send streams */
392 cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
393 if (!stream->fd || stream->sent_to_consumer) {
394 continue;
395 }
396
397 /* Add stream on the kernel consumer side. */
398 ret = kernel_consumer_add_stream(sock, channel, stream,
399 ksession, monitor);
400 if (ret < 0) {
401 goto error;
402 }
403 stream->sent_to_consumer = true;
404 }
405
406 error:
407 rcu_read_unlock();
408 return ret;
409 }
410
411 /*
412 * Send all stream fds of the kernel session to the consumer.
413 *
414 * The consumer socket lock must be held by the caller.
415 */
416 int kernel_consumer_send_session(struct consumer_socket *sock,
417 struct ltt_kernel_session *session)
418 {
419 int ret, monitor = 0;
420 struct ltt_kernel_channel *chan;
421
422 /* Safety net */
423 assert(session);
424 assert(session->consumer);
425 assert(sock);
426
427 /* Bail out if consumer is disabled */
428 if (!session->consumer->enabled) {
429 ret = LTTNG_OK;
430 goto error;
431 }
432
433 /* Don't monitor the streams on the consumer if in flight recorder. */
434 if (session->output_traces) {
435 monitor = 1;
436 }
437
438 DBG("Sending session stream to kernel consumer");
439
440 if (session->metadata_stream_fd >= 0 && session->metadata) {
441 ret = kernel_consumer_add_metadata(sock, session, monitor);
442 if (ret < 0) {
443 goto error;
444 }
445 }
446
447 /* Send channel and streams of it */
448 cds_list_for_each_entry(chan, &session->channel_list.head, list) {
449 ret = kernel_consumer_send_channel_streams(sock, chan, session,
450 monitor);
451 if (ret < 0) {
452 goto error;
453 }
454 if (monitor) {
455 /*
456 * Inform the relay that all the streams for the
457 * channel were sent.
458 */
459 ret = kernel_consumer_streams_sent(sock, session, chan->key);
460 if (ret < 0) {
461 goto error;
462 }
463 }
464 }
465
466 DBG("Kernel consumer FDs of metadata and channel streams sent");
467
468 session->consumer_fds_sent = 1;
469 return 0;
470
471 error:
472 return ret;
473 }
474
475 int kernel_consumer_destroy_channel(struct consumer_socket *socket,
476 struct ltt_kernel_channel *channel)
477 {
478 int ret;
479 struct lttcomm_consumer_msg msg;
480
481 assert(channel);
482 assert(socket);
483
484 DBG("Sending kernel consumer destroy channel key %" PRIu64, channel->key);
485
486 memset(&msg, 0, sizeof(msg));
487 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
488 msg.u.destroy_channel.key = channel->key;
489
490 pthread_mutex_lock(socket->lock);
491 health_code_update();
492
493 ret = consumer_send_msg(socket, &msg);
494 if (ret < 0) {
495 goto error;
496 }
497
498 error:
499 health_code_update();
500 pthread_mutex_unlock(socket->lock);
501 return ret;
502 }
503
504 int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
505 struct ltt_kernel_metadata *metadata)
506 {
507 int ret;
508 struct lttcomm_consumer_msg msg;
509
510 assert(metadata);
511 assert(socket);
512
513 DBG("Sending kernel consumer destroy channel key %" PRIu64, metadata->key);
514
515 memset(&msg, 0, sizeof(msg));
516 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
517 msg.u.destroy_channel.key = metadata->key;
518
519 pthread_mutex_lock(socket->lock);
520 health_code_update();
521
522 ret = consumer_send_msg(socket, &msg);
523 if (ret < 0) {
524 goto error;
525 }
526
527 error:
528 health_code_update();
529 pthread_mutex_unlock(socket->lock);
530 return ret;
531 }
This page took 0.039576 seconds and 4 git commands to generate.