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