Cleanup: Reduce scope of connections in main relayd thread
[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
18#define _GNU_SOURCE
6c1c0768 19#define _LGPL_SOURCE
48842b30
DG
20#include <errno.h>
21#include <stdio.h>
22#include <stdlib.h>
23#include <string.h>
24#include <unistd.h>
d88aee68 25#include <inttypes.h>
48842b30 26
990570ed 27#include <common/common.h>
db758600 28#include <common/consumer.h>
990570ed 29#include <common/defaults.h>
48842b30 30
00e2e675 31#include "consumer.h"
8782cc74 32#include "health-sessiond.h"
48842b30 33#include "ust-consumer.h"
331744e3
JD
34#include "buffer-registry.h"
35#include "session.h"
48842b30
DG
36
37/*
ffe60014
DG
38 * Return allocated full pathname of the session using the consumer trace path
39 * and subdir if available. On a successful allocation, the directory of the
40 * trace is created with the session credentials.
41 *
42 * The caller can safely free(3) the returned value. On error, NULL is
43 * returned.
48842b30 44 */
ffe60014
DG
45static char *setup_trace_path(struct consumer_output *consumer,
46 struct ust_app_session *ua_sess)
48842b30 47{
ffe60014
DG
48 int ret;
49 char *pathname;
37278a1e 50
ffe60014
DG
51 assert(consumer);
52 assert(ua_sess);
00e2e675 53
840cb59c 54 health_code_update();
ca03de58 55
ffe60014
DG
56 /* Allocate our self the string to make sure we never exceed PATH_MAX. */
57 pathname = zmalloc(PATH_MAX);
58 if (!pathname) {
48842b30
DG
59 goto error;
60 }
00e2e675 61
ffe60014
DG
62 /* Get correct path name destination */
63 if (consumer->type == CONSUMER_DST_LOCAL) {
64 /* Set application path to the destination path */
dec56f6c 65 ret = snprintf(pathname, PATH_MAX, "%s%s%s",
ffe60014
DG
66 consumer->dst.trace_path, consumer->subdir, ua_sess->path);
67 if (ret < 0) {
68 PERROR("snprintf channel path");
69 goto error;
70 }
ca03de58 71
ffe60014 72 /* Create directory. Ignore if exist. */
7972aab2
DG
73 ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG,
74 ua_sess->euid, ua_sess->egid);
ffe60014
DG
75 if (ret < 0) {
76 if (ret != -EEXIST) {
77 ERR("Trace directory creation error");
78 goto error;
79 }
80 }
81 } else {
dec56f6c 82 ret = snprintf(pathname, PATH_MAX, "%s%s", consumer->subdir,
ffe60014
DG
83 ua_sess->path);
84 if (ret < 0) {
85 PERROR("snprintf channel path");
86 goto error;
87 }
48842b30
DG
88 }
89
ffe60014 90 return pathname;
ca03de58 91
37278a1e 92error:
ffe60014
DG
93 free(pathname);
94 return NULL;
37278a1e
DG
95}
96
97/*
ffe60014
DG
98 * Send a single channel to the consumer using command ADD_CHANNEL.
99 *
7972aab2 100 * Consumer socket lock MUST be acquired before calling this.
37278a1e 101 */
ffe60014
DG
102static int ask_channel_creation(struct ust_app_session *ua_sess,
103 struct ust_app_channel *ua_chan, struct consumer_output *consumer,
7972aab2 104 struct consumer_socket *socket, struct ust_registry_session *registry)
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;
37278a1e 112
ffe60014
DG
113 assert(ua_sess);
114 assert(ua_chan);
115 assert(socket);
37278a1e 116 assert(consumer);
7972aab2 117 assert(registry);
ffe60014
DG
118
119 DBG2("Asking UST consumer for channel");
120
121 /* Get and create full trace path of session. */
10a50311
JD
122 if (ua_sess->output_traces) {
123 pathname = setup_trace_path(consumer, ua_sess);
124 if (!pathname) {
125 ret = -1;
126 goto error;
127 }
ffe60014
DG
128 }
129
7972aab2
DG
130 /* Depending on the buffer type, a different channel key is used. */
131 if (ua_sess->buffer_type == LTTNG_BUFFER_PER_UID) {
132 chan_reg_key = ua_chan->tracing_channel_id;
133 } else {
134 chan_reg_key = ua_chan->key;
135 }
136
137 if (ua_chan->attr.type == LTTNG_UST_CHAN_METADATA) {
138 chan_id = -1U;
139 } else {
140 chan_reg = ust_registry_channel_find(registry, chan_reg_key);
141 assert(chan_reg);
142 chan_id = chan_reg->chan_id;
143 }
144
0c759fc9
DG
145 switch (ua_chan->attr.output) {
146 case LTTNG_UST_MMAP:
147 default:
148 output = LTTNG_EVENT_MMAP;
149 break;
150 }
151
ffe60014
DG
152 consumer_init_ask_channel_comm_msg(&msg,
153 ua_chan->attr.subbuf_size,
154 ua_chan->attr.num_subbuf,
155 ua_chan->attr.overwrite,
156 ua_chan->attr.switch_timer_interval,
157 ua_chan->attr.read_timer_interval,
ecc48a90 158 ua_sess->live_timer_interval,
0c759fc9 159 output,
ffe60014 160 (int) ua_chan->attr.type,
7972aab2 161 ua_sess->tracing_id,
ca22feea 162 pathname,
ffe60014 163 ua_chan->name,
7972aab2
DG
164 ua_sess->euid,
165 ua_sess->egid,
ffe60014
DG
166 consumer->net_seq_index,
167 ua_chan->key,
7972aab2 168 registry->uuid,
1624d5b7
JD
169 chan_id,
170 ua_chan->tracefile_size,
2bba9e53 171 ua_chan->tracefile_count,
1950109e 172 ua_sess->id,
567eb353
DG
173 ua_sess->output_traces,
174 ua_sess->uid);
37278a1e 175
840cb59c 176 health_code_update();
ca03de58 177
52898cb1 178 ret = consumer_socket_send(socket, &msg, sizeof(msg));
37278a1e
DG
179 if (ret < 0) {
180 goto error;
181 }
182
ffe60014
DG
183 ret = consumer_recv_status_channel(socket, &key,
184 &ua_chan->expected_stream_count);
185 if (ret < 0) {
186 goto error;
187 }
188 /* Communication protocol error. */
189 assert(key == ua_chan->key);
190 /* We need at least one where 1 stream for 1 cpu. */
10a50311
JD
191 if (ua_sess->output_traces) {
192 assert(ua_chan->expected_stream_count > 0);
193 }
ffe60014 194
d88aee68 195 DBG2("UST ask channel %" PRIu64 " successfully done with %u stream(s)", key,
ffe60014 196 ua_chan->expected_stream_count);
ca03de58 197
37278a1e 198error:
ffe60014
DG
199 free(pathname);
200 health_code_update();
37278a1e
DG
201 return ret;
202}
203
204/*
ffe60014
DG
205 * Ask consumer to create a channel for a given session.
206 *
207 * Returns 0 on success else a negative value.
37278a1e 208 */
ffe60014
DG
209int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
210 struct ust_app_channel *ua_chan, struct consumer_output *consumer,
7972aab2 211 struct consumer_socket *socket, struct ust_registry_session *registry)
37278a1e
DG
212{
213 int ret;
37278a1e 214
ffe60014
DG
215 assert(ua_sess);
216 assert(ua_chan);
217 assert(consumer);
218 assert(socket);
7972aab2 219 assert(registry);
f50f23d9 220
d9078d0c
DG
221 if (!consumer->enabled) {
222 ret = -LTTNG_ERR_NO_CONSUMER;
223 DBG3("Consumer is disabled");
224 goto error;
225 }
226
ffe60014 227 pthread_mutex_lock(socket->lock);
37278a1e 228
7972aab2 229 ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket, registry);
37278a1e
DG
230 if (ret < 0) {
231 goto error;
232 }
233
48842b30 234error:
ffe60014 235 pthread_mutex_unlock(socket->lock);
48842b30
DG
236 return ret;
237}
238
239/*
ffe60014
DG
240 * Send a get channel command to consumer using the given channel key. The
241 * channel object is populated and the stream list.
242 *
243 * Return 0 on success else a negative value.
48842b30 244 */
ffe60014
DG
245int ust_consumer_get_channel(struct consumer_socket *socket,
246 struct ust_app_channel *ua_chan)
48842b30 247{
ffe60014 248 int ret;
37278a1e 249 struct lttcomm_consumer_msg msg;
48842b30 250
ffe60014
DG
251 assert(ua_chan);
252 assert(socket);
48842b30 253
53efb85a 254 memset(&msg, 0, sizeof(msg));
ffe60014
DG
255 msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL;
256 msg.u.get_channel.key = ua_chan->key;
37278a1e 257
ffe60014 258 pthread_mutex_lock(socket->lock);
840cb59c 259 health_code_update();
ca03de58 260
ffe60014
DG
261 /* Send command and wait for OK reply. */
262 ret = consumer_send_msg(socket, &msg);
37278a1e
DG
263 if (ret < 0) {
264 goto error;
265 }
266
ffe60014 267 /* First, get the channel from consumer. */
9363801e 268 ret = ustctl_recv_channel_from_consumer(*socket->fd_ptr, &ua_chan->obj);
37278a1e 269 if (ret < 0) {
ffe60014
DG
270 if (ret != -EPIPE) {
271 ERR("Error recv channel from consumer %d with ret %d",
9363801e 272 *socket->fd_ptr, ret);
ffe60014
DG
273 } else {
274 DBG3("UST app recv channel from consumer. Consumer is dead.");
275 }
37278a1e
DG
276 goto error;
277 }
00e2e675 278
ffe60014
DG
279 /* Next, get all streams. */
280 while (1) {
281 struct ust_app_stream *stream;
ca03de58 282
ffe60014
DG
283 /* Create UST stream */
284 stream = ust_app_alloc_stream();
285 if (stream == NULL) {
286 ret = -ENOMEM;
48842b30
DG
287 goto error;
288 }
289
ffe60014 290 /* Stream object is populated by this call if successful. */
9363801e 291 ret = ustctl_recv_stream_from_consumer(*socket->fd_ptr, &stream->obj);
37278a1e 292 if (ret < 0) {
ffe60014
DG
293 free(stream);
294 if (ret == -LTTNG_UST_ERR_NOENT) {
295 DBG3("UST app consumer has no more stream available");
296 ret = 0;
297 break;
298 }
299 if (ret != -EPIPE) {
300 ERR("Recv stream from consumer %d with ret %d",
9363801e 301 *socket->fd_ptr, ret);
ffe60014
DG
302 } else {
303 DBG3("UST app recv stream from consumer. Consumer is dead.");
00e2e675 304 }
48842b30
DG
305 goto error;
306 }
37278a1e 307
ffe60014
DG
308 /* Order is important this is why a list is used. */
309 cds_list_add_tail(&stream->list, &ua_chan->streams.head);
310 ua_chan->streams.count++;
37278a1e 311
5368d366 312 DBG2("UST app stream %d received successfully", ua_chan->streams.count);
ffe60014
DG
313 }
314
315 /* This MUST match or else we have a synchronization problem. */
316 assert(ua_chan->expected_stream_count == ua_chan->streams.count);
ca03de58 317
ffe60014
DG
318 /* Wait for confirmation that we can proceed with the streams. */
319 ret = consumer_recv_status_reply(socket);
37278a1e
DG
320 if (ret < 0) {
321 goto error;
322 }
323
324error:
ffe60014
DG
325 health_code_update();
326 pthread_mutex_unlock(socket->lock);
37278a1e
DG
327 return ret;
328}
329
330/*
ffe60014
DG
331 * Send a destroy channel command to consumer using the given channel key.
332 *
333 * Note that this command MUST be used prior to a successful
334 * LTTNG_CONSUMER_GET_CHANNEL because once this command is done successfully,
335 * the streams are dispatched to the consumer threads and MUST be teardown
336 * through the hang up process.
337 *
338 * Return 0 on success else a negative value.
37278a1e 339 */
ffe60014
DG
340int ust_consumer_destroy_channel(struct consumer_socket *socket,
341 struct ust_app_channel *ua_chan)
37278a1e 342{
ffe60014
DG
343 int ret;
344 struct lttcomm_consumer_msg msg;
a4b92340 345
ffe60014
DG
346 assert(ua_chan);
347 assert(socket);
37278a1e 348
53efb85a 349 memset(&msg, 0, sizeof(msg));
ffe60014
DG
350 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
351 msg.u.destroy_channel.key = ua_chan->key;
173af62f 352
ffe60014
DG
353 pthread_mutex_lock(socket->lock);
354 health_code_update();
37278a1e 355
ffe60014 356 ret = consumer_send_msg(socket, &msg);
37278a1e
DG
357 if (ret < 0) {
358 goto error;
48842b30
DG
359 }
360
ffe60014
DG
361error:
362 health_code_update();
363 pthread_mutex_unlock(socket->lock);
364 return ret;
365}
aeb96892 366
ffe60014
DG
367/*
368 * Send a given stream to UST tracer.
369 *
370 * On success return 0 else a negative value.
371 */
372int ust_consumer_send_stream_to_ust(struct ust_app *app,
373 struct ust_app_channel *channel, struct ust_app_stream *stream)
374{
375 int ret;
376
377 assert(app);
378 assert(stream);
379 assert(channel);
380
381 DBG2("UST consumer send stream to app %d", app->sock);
382
383 /* Relay stream to application. */
384 ret = ustctl_send_stream_to_ust(app->sock, channel->obj, stream->obj);
385 if (ret < 0) {
386 if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
8cf93def
DG
387 ERR("ustctl send stream handle %d to app pid: %d with ret %d",
388 stream->obj->handle, app->pid, ret);
ffe60014
DG
389 } else {
390 DBG3("UST app send stream to ust failed. Application is dead.");
48842b30 391 }
ffe60014 392 goto error;
48842b30 393 }
d0b96690 394 channel->handle = channel->obj->handle;
48842b30 395
ffe60014
DG
396error:
397 return ret;
398}
399
400/*
401 * Send channel previously received from the consumer to the UST tracer.
402 *
403 * On success return 0 else a negative value.
404 */
405int ust_consumer_send_channel_to_ust(struct ust_app *app,
406 struct ust_app_session *ua_sess, struct ust_app_channel *channel)
407{
408 int ret;
409
410 assert(app);
411 assert(ua_sess);
412 assert(channel);
413 assert(channel->obj);
414
7972aab2
DG
415 DBG2("UST app send channel to sock %d pid %d (name: %s, key: %" PRIu64 ")",
416 app->sock, app->pid, channel->name, channel->tracing_channel_id);
48842b30 417
ffe60014
DG
418 /* Send stream to application. */
419 ret = ustctl_send_channel_to_ust(app->sock, ua_sess->handle, channel->obj);
420 if (ret < 0) {
421 if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
422 ERR("Error ustctl send channel %s to app pid: %d with ret %d",
423 channel->name, app->pid, ret);
424 } else {
425 DBG3("UST app send channel to ust failed. Application is dead.");
426 }
427 goto error;
428 }
48842b30
DG
429
430error:
431 return ret;
432}
331744e3
JD
433
434/*
435 * Handle the metadata requests from the UST consumer
436 *
437 * Return 0 on success else a negative value.
438 */
439int ust_consumer_metadata_request(struct consumer_socket *socket)
440{
441 int ret;
442 ssize_t ret_push;
443 struct lttcomm_metadata_request_msg request;
444 struct buffer_reg_uid *reg_uid;
445 struct ust_registry_session *ust_reg;
446 struct lttcomm_consumer_msg msg;
447
448 assert(socket);
449
450 rcu_read_lock();
331744e3
JD
451 health_code_update();
452
453 /* Wait for a metadata request */
dc2bbdae 454 pthread_mutex_lock(socket->lock);
52898cb1 455 ret = consumer_socket_recv(socket, &request, sizeof(request));
dc2bbdae 456 pthread_mutex_unlock(socket->lock);
52898cb1 457 if (ret < 0) {
331744e3
JD
458 goto end;
459 }
460
1950109e 461 DBG("Metadata request received for session %" PRIu64 ", key %" PRIu64,
331744e3
JD
462 request.session_id, request.key);
463
464 reg_uid = buffer_reg_uid_find(request.session_id,
465 request.bits_per_long, request.uid);
466 if (reg_uid) {
467 ust_reg = reg_uid->registry->reg.ust;
468 } else {
469 struct buffer_reg_pid *reg_pid =
1950109e 470 buffer_reg_pid_find(request.session_id_per_pid);
331744e3 471 if (!reg_pid) {
1950109e
JD
472 DBG("PID registry not found for session id %" PRIu64,
473 request.session_id_per_pid);
331744e3 474
53efb85a 475 memset(&msg, 0, sizeof(msg));
331744e3
JD
476 msg.cmd_type = LTTNG_ERR_UND;
477 (void) consumer_send_msg(socket, &msg);
478 /*
479 * This is possible since the session might have been destroyed
480 * during a consumer metadata request. So here, return gracefully
481 * because the destroy session will push the remaining metadata to
482 * the consumer.
483 */
484 ret = 0;
485 goto end;
486 }
487 ust_reg = reg_pid->registry->reg.ust;
488 }
489 assert(ust_reg);
490
dc2bbdae 491 pthread_mutex_lock(&ust_reg->lock);
331744e3 492 ret_push = ust_app_push_metadata(ust_reg, socket, 1);
dc2bbdae 493 pthread_mutex_unlock(&ust_reg->lock);
331744e3
JD
494 if (ret_push < 0) {
495 ERR("Pushing metadata");
496 ret = -1;
497 goto end;
498 }
499 DBG("UST Consumer metadata pushed successfully");
500 ret = 0;
501
502end:
331744e3
JD
503 rcu_read_unlock();
504 return ret;
505}
This page took 0.105034 seconds and 4 git commands to generate.