f9a29e49ecd2625f8e6503a8a66a812d1bb72957
[lttng-tools.git] / src / bin / lttng-sessiond / ust-consumer.c
1 /*
2 * Copyright (C) 2011 - David Goulet <david.goulet@polymtl.ca>
3 *
4 * This program is free software; you can redistribute it and/or modify
5 * it under the terms of the GNU General Public License, version 2 only,
6 * as published by the Free Software Foundation.
7 *
8 * This program is distributed in the hope that it will be useful,
9 * but WITHOUT ANY WARRANTY; without even the implied warranty of
10 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11 * GNU General Public License for more details.
12 *
13 * You should have received a copy of the GNU General Public License along
14 * with this program; if not, write to the Free Software Foundation, Inc.,
15 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
16 */
17
18 #define _LGPL_SOURCE
19 #include <errno.h>
20 #include <stdio.h>
21 #include <stdlib.h>
22 #include <string.h>
23 #include <unistd.h>
24 #include <inttypes.h>
25
26 #include <common/common.h>
27 #include <common/consumer/consumer.h>
28 #include <common/defaults.h>
29
30 #include "consumer.h"
31 #include "health-sessiond.h"
32 #include "ust-consumer.h"
33 #include "buffer-registry.h"
34 #include "session.h"
35 #include "lttng-sessiond.h"
36
37 /*
38 * Return allocated full pathname of the session using the consumer trace path
39 * and subdir if available.
40 *
41 * The caller can safely free(3) the returned value. On error, NULL is
42 * returned.
43 */
44 static char *setup_channel_trace_path(struct consumer_output *consumer,
45 struct ust_app_session *ua_sess)
46 {
47 int ret;
48 char *pathname;
49
50 assert(consumer);
51 assert(ua_sess);
52
53 health_code_update();
54
55 /*
56 * Allocate the string ourself to make sure we never exceed
57 * LTTNG_PATH_MAX.
58 */
59 pathname = zmalloc(LTTNG_PATH_MAX);
60 if (!pathname) {
61 goto error;
62 }
63
64 /* Get correct path name destination */
65 if (consumer->type == CONSUMER_DST_NET &&
66 consumer->relay_major_version == 2 &&
67 consumer->relay_minor_version < 11) {
68 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s/%s%s",
69 consumer->dst.net.base_dir,
70 consumer->chunk_path, consumer->domain_subdir,
71 ua_sess->path);
72 } else {
73 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s",
74 consumer->domain_subdir, ua_sess->path);
75 }
76 DBG3("Userspace consumer trace path relative to current trace chunk: \"%s\"",
77 pathname);
78 if (ret < 0) {
79 PERROR("Failed to format channel path");
80 goto error;
81 } else if (ret >= LTTNG_PATH_MAX) {
82 ERR("Truncation occurred while formatting channel path");
83 ret = -1;
84 goto error;
85 }
86
87 return pathname;
88 error:
89 free(pathname);
90 return NULL;
91 }
92
93 /*
94 * Send a single channel to the consumer using command ASK_CHANNEL_CREATION.
95 *
96 * Consumer socket lock MUST be acquired before calling this.
97 */
98 static int ask_channel_creation(struct ust_app_session *ua_sess,
99 struct ust_app_channel *ua_chan,
100 struct consumer_output *consumer,
101 struct consumer_socket *socket,
102 struct ust_registry_session *registry,
103 struct lttng_trace_chunk *trace_chunk)
104 {
105 int ret, output;
106 uint32_t chan_id;
107 uint64_t key, chan_reg_key;
108 char *pathname = NULL;
109 struct lttcomm_consumer_msg msg;
110 struct ust_registry_channel *chan_reg;
111 char shm_path[PATH_MAX] = "";
112 char root_shm_path[PATH_MAX] = "";
113 bool is_local_trace;
114
115 assert(ua_sess);
116 assert(ua_chan);
117 assert(socket);
118 assert(consumer);
119 assert(registry);
120
121 DBG2("Asking UST consumer for channel");
122
123 is_local_trace = consumer->net_seq_index == -1ULL;
124 /* Format the channel's path (relative to the current trace chunk). */
125 pathname = setup_channel_trace_path(consumer, ua_sess);
126 if (!pathname) {
127 ret = -1;
128 goto error;
129 }
130
131 if (is_local_trace && trace_chunk) {
132 enum lttng_trace_chunk_status chunk_status;
133 char *pathname_index;
134
135 ret = asprintf(&pathname_index, "%s/" DEFAULT_INDEX_DIR,
136 pathname);
137 if (ret < 0) {
138 ERR("Failed to format channel index directory");
139 ret = -1;
140 goto error;
141 }
142
143 /*
144 * Create the index subdirectory which will take care
145 * of implicitly creating the channel's path.
146 */
147 chunk_status = lttng_trace_chunk_create_subdirectory(
148 trace_chunk, pathname_index);
149 free(pathname_index);
150 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
151 ret = -1;
152 goto error;
153 }
154 }
155
156 /* Depending on the buffer type, a different channel key is used. */
157 if (ua_sess->buffer_type == LTTNG_BUFFER_PER_UID) {
158 chan_reg_key = ua_chan->tracing_channel_id;
159 } else {
160 chan_reg_key = ua_chan->key;
161 }
162
163 if (ua_chan->attr.type == LTTNG_UST_CHAN_METADATA) {
164 chan_id = -1U;
165 /*
166 * Metadata channels shm_path (buffers) are handled within
167 * session daemon. Consumer daemon should not try to create
168 * those buffer files.
169 */
170 } else {
171 chan_reg = ust_registry_channel_find(registry, chan_reg_key);
172 assert(chan_reg);
173 chan_id = chan_reg->chan_id;
174 if (ua_sess->shm_path[0]) {
175 strncpy(shm_path, ua_sess->shm_path, sizeof(shm_path));
176 shm_path[sizeof(shm_path) - 1] = '\0';
177 strncat(shm_path, "/",
178 sizeof(shm_path) - strlen(shm_path) - 1);
179 strncat(shm_path, ua_chan->name,
180 sizeof(shm_path) - strlen(shm_path) - 1);
181 strncat(shm_path, "_",
182 sizeof(shm_path) - strlen(shm_path) - 1);
183 }
184 strncpy(root_shm_path, ua_sess->root_shm_path, sizeof(root_shm_path));
185 root_shm_path[sizeof(root_shm_path) - 1] = '\0';
186 }
187
188 switch (ua_chan->attr.output) {
189 case LTTNG_UST_MMAP:
190 default:
191 output = LTTNG_EVENT_MMAP;
192 break;
193 }
194
195 consumer_init_ask_channel_comm_msg(&msg,
196 ua_chan->attr.subbuf_size,
197 ua_chan->attr.num_subbuf,
198 ua_chan->attr.overwrite,
199 ua_chan->attr.switch_timer_interval,
200 ua_chan->attr.read_timer_interval,
201 ua_sess->live_timer_interval,
202 ua_chan->monitor_timer_interval,
203 output,
204 (int) ua_chan->attr.type,
205 ua_sess->tracing_id,
206 pathname,
207 ua_chan->name,
208 consumer->net_seq_index,
209 ua_chan->key,
210 registry->uuid,
211 chan_id,
212 ua_chan->tracefile_size,
213 ua_chan->tracefile_count,
214 ua_sess->id,
215 ua_sess->output_traces,
216 ua_sess->real_credentials.uid,
217 ua_chan->attr.blocking_timeout,
218 root_shm_path, shm_path,
219 trace_chunk,
220 &ua_sess->effective_credentials);
221
222 health_code_update();
223
224 ret = consumer_socket_send(socket, &msg, sizeof(msg));
225 if (ret < 0) {
226 goto error;
227 }
228
229 ret = consumer_recv_status_channel(socket, &key,
230 &ua_chan->expected_stream_count);
231 if (ret < 0) {
232 goto error;
233 }
234 /* Communication protocol error. */
235 assert(key == ua_chan->key);
236 /* We need at least one where 1 stream for 1 cpu. */
237 if (ua_sess->output_traces) {
238 assert(ua_chan->expected_stream_count > 0);
239 }
240
241 DBG2("UST ask channel %" PRIu64 " successfully done with %u stream(s)", key,
242 ua_chan->expected_stream_count);
243
244 error:
245 free(pathname);
246 health_code_update();
247 return ret;
248 }
249
250 /*
251 * Ask consumer to create a channel for a given session.
252 *
253 * Session list and rcu read side locks must be held by the caller.
254 *
255 * Returns 0 on success else a negative value.
256 */
257 int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
258 struct ust_app_channel *ua_chan,
259 struct consumer_output *consumer,
260 struct consumer_socket *socket,
261 struct ust_registry_session *registry,
262 struct lttng_trace_chunk * trace_chunk)
263 {
264 int ret;
265
266 assert(ua_sess);
267 assert(ua_chan);
268 assert(consumer);
269 assert(socket);
270 assert(registry);
271
272 if (!consumer->enabled) {
273 ret = -LTTNG_ERR_NO_CONSUMER;
274 DBG3("Consumer is disabled");
275 goto error;
276 }
277
278 pthread_mutex_lock(socket->lock);
279 ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket, registry,
280 trace_chunk);
281 pthread_mutex_unlock(socket->lock);
282 if (ret < 0) {
283 ERR("ask_channel_creation consumer command failed");
284 goto error;
285 }
286
287 error:
288 return ret;
289 }
290
291 /*
292 * Send a get channel command to consumer using the given channel key. The
293 * channel object is populated and the stream list.
294 *
295 * Return 0 on success else a negative value.
296 */
297 int ust_consumer_get_channel(struct consumer_socket *socket,
298 struct ust_app_channel *ua_chan)
299 {
300 int ret;
301 struct lttcomm_consumer_msg msg;
302
303 assert(ua_chan);
304 assert(socket);
305
306 memset(&msg, 0, sizeof(msg));
307 msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL;
308 msg.u.get_channel.key = ua_chan->key;
309
310 pthread_mutex_lock(socket->lock);
311 health_code_update();
312
313 /* Send command and wait for OK reply. */
314 ret = consumer_send_msg(socket, &msg);
315 if (ret < 0) {
316 goto error;
317 }
318
319 /* First, get the channel from consumer. */
320 ret = ustctl_recv_channel_from_consumer(*socket->fd_ptr, &ua_chan->obj);
321 if (ret < 0) {
322 if (ret != -EPIPE) {
323 ERR("Error recv channel from consumer %d with ret %d",
324 *socket->fd_ptr, ret);
325 } else {
326 DBG3("UST app recv channel from consumer. Consumer is dead.");
327 }
328 goto error;
329 }
330
331 /* Next, get all streams. */
332 while (1) {
333 struct ust_app_stream *stream;
334
335 /* Create UST stream */
336 stream = ust_app_alloc_stream();
337 if (stream == NULL) {
338 ret = -ENOMEM;
339 goto error;
340 }
341
342 /* Stream object is populated by this call if successful. */
343 ret = ustctl_recv_stream_from_consumer(*socket->fd_ptr, &stream->obj);
344 if (ret < 0) {
345 free(stream);
346 if (ret == -LTTNG_UST_ERR_NOENT) {
347 DBG3("UST app consumer has no more stream available");
348 break;
349 }
350 if (ret != -EPIPE) {
351 ERR("Recv stream from consumer %d with ret %d",
352 *socket->fd_ptr, ret);
353 } else {
354 DBG3("UST app recv stream from consumer. Consumer is dead.");
355 }
356 goto error;
357 }
358
359 /* Order is important this is why a list is used. */
360 cds_list_add_tail(&stream->list, &ua_chan->streams.head);
361 ua_chan->streams.count++;
362
363 DBG2("UST app stream %d received successfully", ua_chan->streams.count);
364 }
365
366 /* This MUST match or else we have a synchronization problem. */
367 assert(ua_chan->expected_stream_count == ua_chan->streams.count);
368
369 /* Wait for confirmation that we can proceed with the streams. */
370 ret = consumer_recv_status_reply(socket);
371 if (ret < 0) {
372 goto error;
373 }
374
375 error:
376 health_code_update();
377 pthread_mutex_unlock(socket->lock);
378 return ret;
379 }
380
381 /*
382 * Send a destroy channel command to consumer using the given channel key.
383 *
384 * Note that this command MUST be used prior to a successful
385 * LTTNG_CONSUMER_GET_CHANNEL because once this command is done successfully,
386 * the streams are dispatched to the consumer threads and MUST be teardown
387 * through the hang up process.
388 *
389 * Return 0 on success else a negative value.
390 */
391 int ust_consumer_destroy_channel(struct consumer_socket *socket,
392 struct ust_app_channel *ua_chan)
393 {
394 int ret;
395 struct lttcomm_consumer_msg msg;
396
397 assert(ua_chan);
398 assert(socket);
399
400 memset(&msg, 0, sizeof(msg));
401 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
402 msg.u.destroy_channel.key = ua_chan->key;
403
404 pthread_mutex_lock(socket->lock);
405 health_code_update();
406
407 ret = consumer_send_msg(socket, &msg);
408 if (ret < 0) {
409 goto error;
410 }
411
412 error:
413 health_code_update();
414 pthread_mutex_unlock(socket->lock);
415 return ret;
416 }
417
418 /*
419 * Send a given stream to UST tracer.
420 *
421 * On success return 0 else a negative value.
422 */
423 int ust_consumer_send_stream_to_ust(struct ust_app *app,
424 struct ust_app_channel *channel, struct ust_app_stream *stream)
425 {
426 int ret;
427
428 assert(app);
429 assert(stream);
430 assert(channel);
431
432 DBG2("UST consumer send stream to app %d", app->sock);
433
434 /* Relay stream to application. */
435 pthread_mutex_lock(&app->sock_lock);
436 ret = ustctl_send_stream_to_ust(app->sock, channel->obj, stream->obj);
437 pthread_mutex_unlock(&app->sock_lock);
438 if (ret < 0) {
439 if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
440 ERR("ustctl send stream handle %d to app pid: %d with ret %d",
441 stream->obj->handle, app->pid, ret);
442 } else {
443 DBG3("UST app send stream to ust failed. Application is dead.");
444 }
445 goto error;
446 }
447 channel->handle = channel->obj->handle;
448
449 error:
450 return ret;
451 }
452
453 /*
454 * Send channel previously received from the consumer to the UST tracer.
455 *
456 * On success return 0 else a negative value.
457 */
458 int ust_consumer_send_channel_to_ust(struct ust_app *app,
459 struct ust_app_session *ua_sess, struct ust_app_channel *channel)
460 {
461 int ret;
462
463 assert(app);
464 assert(ua_sess);
465 assert(channel);
466 assert(channel->obj);
467
468 DBG2("UST app send channel to sock %d pid %d (name: %s, key: %" PRIu64 ")",
469 app->sock, app->pid, channel->name, channel->tracing_channel_id);
470
471 /* Send stream to application. */
472 pthread_mutex_lock(&app->sock_lock);
473 ret = ustctl_send_channel_to_ust(app->sock, ua_sess->handle, channel->obj);
474 pthread_mutex_unlock(&app->sock_lock);
475 if (ret < 0) {
476 if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
477 ERR("Error ustctl send channel %s to app pid: %d with ret %d",
478 channel->name, app->pid, ret);
479 } else {
480 DBG3("UST app send channel to ust failed. Application is dead.");
481 }
482 goto error;
483 }
484
485 error:
486 return ret;
487 }
488
489 /*
490 * Handle the metadata requests from the UST consumer
491 *
492 * Return 0 on success else a negative value.
493 */
494 int ust_consumer_metadata_request(struct consumer_socket *socket)
495 {
496 int ret;
497 ssize_t ret_push;
498 struct lttcomm_metadata_request_msg request;
499 struct buffer_reg_uid *reg_uid;
500 struct ust_registry_session *ust_reg;
501 struct lttcomm_consumer_msg msg;
502
503 assert(socket);
504
505 rcu_read_lock();
506 health_code_update();
507
508 /* Wait for a metadata request */
509 pthread_mutex_lock(socket->lock);
510 ret = consumer_socket_recv(socket, &request, sizeof(request));
511 pthread_mutex_unlock(socket->lock);
512 if (ret < 0) {
513 goto end;
514 }
515
516 DBG("Metadata request received for session %" PRIu64 ", key %" PRIu64,
517 request.session_id, request.key);
518
519 reg_uid = buffer_reg_uid_find(request.session_id,
520 request.bits_per_long, request.uid);
521 if (reg_uid) {
522 ust_reg = reg_uid->registry->reg.ust;
523 } else {
524 struct buffer_reg_pid *reg_pid =
525 buffer_reg_pid_find(request.session_id_per_pid);
526 if (!reg_pid) {
527 DBG("PID registry not found for session id %" PRIu64,
528 request.session_id_per_pid);
529
530 memset(&msg, 0, sizeof(msg));
531 msg.cmd_type = LTTNG_ERR_UND;
532 pthread_mutex_lock(socket->lock);
533 (void) consumer_send_msg(socket, &msg);
534 pthread_mutex_unlock(socket->lock);
535 /*
536 * This is possible since the session might have been destroyed
537 * during a consumer metadata request. So here, return gracefully
538 * because the destroy session will push the remaining metadata to
539 * the consumer.
540 */
541 ret = 0;
542 goto end;
543 }
544 ust_reg = reg_pid->registry->reg.ust;
545 }
546 assert(ust_reg);
547
548 pthread_mutex_lock(&ust_reg->lock);
549 ret_push = ust_app_push_metadata(ust_reg, socket, 1);
550 pthread_mutex_unlock(&ust_reg->lock);
551 if (ret_push == -EPIPE) {
552 DBG("Application or relay closed while pushing metadata");
553 } else if (ret_push < 0) {
554 ERR("Pushing metadata");
555 ret = -1;
556 goto end;
557 } else {
558 DBG("UST Consumer metadata pushed successfully");
559 }
560 ret = 0;
561
562 end:
563 rcu_read_unlock();
564 return ret;
565 }
This page took 0.058543 seconds and 4 git commands to generate.