Test for new metadata at each packet
[lttng-tools.git] / src / common / ust-consumer / ust-consumer.c
CommitLineData
3bd1e081
MD
1/*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 *
d14d33bf
AM
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
3bd1e081
MD
8 *
9 * This program is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU General Public License for more details.
13 *
d14d33bf
AM
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
3bd1e081
MD
17 */
18
19#define _GNU_SOURCE
20#include <assert.h>
f02e1e8a 21#include <lttng/ust-ctl.h>
3bd1e081
MD
22#include <poll.h>
23#include <pthread.h>
24#include <stdlib.h>
25#include <string.h>
26#include <sys/mman.h>
27#include <sys/socket.h>
dbb5dfe6 28#include <sys/stat.h>
3bd1e081 29#include <sys/types.h>
77c7c900 30#include <inttypes.h>
3bd1e081 31#include <unistd.h>
ffe60014 32#include <urcu/list.h>
331744e3 33#include <signal.h>
0857097f 34
990570ed 35#include <common/common.h>
10a8a223 36#include <common/sessiond-comm/sessiond-comm.h>
00e2e675 37#include <common/relayd/relayd.h>
dbb5dfe6 38#include <common/compat/fcntl.h>
331744e3 39#include <common/consumer-metadata-cache.h>
10a50311 40#include <common/consumer-stream.h>
331744e3 41#include <common/consumer-timer.h>
fe4477ee 42#include <common/utils.h>
309167d2 43#include <common/index/index.h>
10a8a223
DG
44
45#include "ust-consumer.h"
3bd1e081
MD
46
47extern struct lttng_consumer_global_data consumer_data;
48extern int consumer_poll_timeout;
49extern volatile int consumer_quit;
50
51/*
ffe60014
DG
52 * Free channel object and all streams associated with it. This MUST be used
53 * only and only if the channel has _NEVER_ been added to the global channel
54 * hash table.
3bd1e081 55 */
ffe60014 56static void destroy_channel(struct lttng_consumer_channel *channel)
3bd1e081 57{
ffe60014
DG
58 struct lttng_consumer_stream *stream, *stmp;
59
60 assert(channel);
61
62 DBG("UST consumer cleaning stream list");
63
64 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
65 send_node) {
66 cds_list_del(&stream->send_node);
67 ustctl_destroy_stream(stream->ustream);
68 free(stream);
69 }
70
71 /*
72 * If a channel is available meaning that was created before the streams
73 * were, delete it.
74 */
75 if (channel->uchan) {
76 lttng_ustconsumer_del_channel(channel);
77 }
78 free(channel);
79}
3bd1e081
MD
80
81/*
ffe60014 82 * Add channel to internal consumer state.
3bd1e081 83 *
ffe60014 84 * Returns 0 on success or else a negative value.
3bd1e081 85 */
ffe60014
DG
86static int add_channel(struct lttng_consumer_channel *channel,
87 struct lttng_consumer_local_data *ctx)
3bd1e081
MD
88{
89 int ret = 0;
90
ffe60014
DG
91 assert(channel);
92 assert(ctx);
93
94 if (ctx->on_recv_channel != NULL) {
95 ret = ctx->on_recv_channel(channel);
96 if (ret == 0) {
d8ef542d 97 ret = consumer_add_channel(channel, ctx);
ffe60014
DG
98 } else if (ret < 0) {
99 /* Most likely an ENOMEM. */
100 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
101 goto error;
102 }
103 } else {
d8ef542d 104 ret = consumer_add_channel(channel, ctx);
3bd1e081
MD
105 }
106
d88aee68 107 DBG("UST consumer channel added (key: %" PRIu64 ")", channel->key);
ffe60014
DG
108
109error:
3bd1e081
MD
110 return ret;
111}
112
113/*
ffe60014
DG
114 * Allocate and return a consumer channel object.
115 */
116static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
117 const char *pathname, const char *name, uid_t uid, gid_t gid,
da009f2c 118 uint64_t relayd_id, uint64_t key, enum lttng_event_output output,
2bba9e53 119 uint64_t tracefile_size, uint64_t tracefile_count,
ecc48a90
JD
120 uint64_t session_id_per_pid, unsigned int monitor,
121 unsigned int live_timer_interval)
ffe60014
DG
122{
123 assert(pathname);
124 assert(name);
125
1950109e
JD
126 return consumer_allocate_channel(key, session_id, pathname, name, uid,
127 gid, relayd_id, output, tracefile_size,
ecc48a90 128 tracefile_count, session_id_per_pid, monitor, live_timer_interval);
ffe60014
DG
129}
130
131/*
132 * Allocate and return a consumer stream object. If _alloc_ret is not NULL, the
133 * error value if applicable is set in it else it is kept untouched.
3bd1e081 134 *
ffe60014 135 * Return NULL on error else the newly allocated stream object.
3bd1e081 136 */
ffe60014
DG
137static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
138 struct lttng_consumer_channel *channel,
139 struct lttng_consumer_local_data *ctx, int *_alloc_ret)
140{
141 int alloc_ret;
142 struct lttng_consumer_stream *stream = NULL;
143
144 assert(channel);
145 assert(ctx);
146
147 stream = consumer_allocate_stream(channel->key,
148 key,
149 LTTNG_CONSUMER_ACTIVE_STREAM,
150 channel->name,
151 channel->uid,
152 channel->gid,
153 channel->relayd_id,
154 channel->session_id,
155 cpu,
156 &alloc_ret,
4891ece8
DG
157 channel->type,
158 channel->monitor);
ffe60014
DG
159 if (stream == NULL) {
160 switch (alloc_ret) {
161 case -ENOENT:
162 /*
163 * We could not find the channel. Can happen if cpu hotplug
164 * happens while tearing down.
165 */
166 DBG3("Could not find channel");
167 break;
168 case -ENOMEM:
169 case -EINVAL:
170 default:
171 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
172 break;
173 }
174 goto error;
175 }
176
177 stream->chan = channel;
178
179error:
180 if (_alloc_ret) {
181 *_alloc_ret = alloc_ret;
182 }
183 return stream;
184}
185
186/*
187 * Send the given stream pointer to the corresponding thread.
188 *
189 * Returns 0 on success else a negative value.
190 */
191static int send_stream_to_thread(struct lttng_consumer_stream *stream,
192 struct lttng_consumer_local_data *ctx)
193{
dae10966
DG
194 int ret;
195 struct lttng_pipe *stream_pipe;
ffe60014
DG
196
197 /* Get the right pipe where the stream will be sent. */
198 if (stream->metadata_flag) {
5ab66908
MD
199 ret = consumer_add_metadata_stream(stream);
200 if (ret) {
201 ERR("Consumer add metadata stream %" PRIu64 " failed.",
202 stream->key);
203 goto error;
204 }
dae10966 205 stream_pipe = ctx->consumer_metadata_pipe;
ffe60014 206 } else {
5ab66908
MD
207 ret = consumer_add_data_stream(stream);
208 if (ret) {
209 ERR("Consumer add stream %" PRIu64 " failed.",
210 stream->key);
211 goto error;
212 }
dae10966 213 stream_pipe = ctx->consumer_data_pipe;
ffe60014
DG
214 }
215
5ab66908
MD
216 /*
217 * From this point on, the stream's ownership has been moved away from
218 * the channel and becomes globally visible.
219 */
220 stream->globally_visible = 1;
221
dae10966 222 ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream));
ffe60014 223 if (ret < 0) {
dae10966
DG
224 ERR("Consumer write %s stream to pipe %d",
225 stream->metadata_flag ? "metadata" : "data",
226 lttng_pipe_get_writefd(stream_pipe));
5ab66908
MD
227 if (stream->metadata_flag) {
228 consumer_del_stream_for_metadata(stream);
229 } else {
230 consumer_del_stream_for_data(stream);
231 }
ffe60014 232 }
5ab66908 233error:
ffe60014
DG
234 return ret;
235}
236
d88aee68
DG
237/*
238 * Create streams for the given channel using liblttng-ust-ctl.
239 *
240 * Return 0 on success else a negative value.
241 */
ffe60014
DG
242static int create_ust_streams(struct lttng_consumer_channel *channel,
243 struct lttng_consumer_local_data *ctx)
244{
245 int ret, cpu = 0;
246 struct ustctl_consumer_stream *ustream;
247 struct lttng_consumer_stream *stream;
248
249 assert(channel);
250 assert(ctx);
251
252 /*
253 * While a stream is available from ustctl. When NULL is returned, we've
254 * reached the end of the possible stream for the channel.
255 */
256 while ((ustream = ustctl_create_stream(channel->uchan, cpu))) {
257 int wait_fd;
04ef1097 258 int ust_metadata_pipe[2];
ffe60014 259
04ef1097
MD
260 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && channel->monitor) {
261 ret = utils_create_pipe_cloexec_nonblock(ust_metadata_pipe);
262 if (ret < 0) {
263 ERR("Create ust metadata poll pipe");
264 goto error;
265 }
266 wait_fd = ust_metadata_pipe[0];
267 } else {
268 wait_fd = ustctl_stream_get_wait_fd(ustream);
269 }
ffe60014
DG
270
271 /* Allocate consumer stream object. */
272 stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret);
273 if (!stream) {
274 goto error_alloc;
275 }
276 stream->ustream = ustream;
277 /*
278 * Store it so we can save multiple function calls afterwards since
279 * this value is used heavily in the stream threads. This is UST
280 * specific so this is why it's done after allocation.
281 */
282 stream->wait_fd = wait_fd;
283
b31398bb
DG
284 /*
285 * Increment channel refcount since the channel reference has now been
286 * assigned in the allocation process above.
287 */
10a50311
JD
288 if (stream->chan->monitor) {
289 uatomic_inc(&stream->chan->refcount);
290 }
b31398bb 291
ffe60014
DG
292 /*
293 * Order is important this is why a list is used. On error, the caller
294 * should clean this list.
295 */
296 cds_list_add_tail(&stream->send_node, &channel->streams.head);
297
298 ret = ustctl_get_max_subbuf_size(stream->ustream,
299 &stream->max_sb_size);
300 if (ret < 0) {
301 ERR("ustctl_get_max_subbuf_size failed for stream %s",
302 stream->name);
303 goto error;
304 }
305
306 /* Do actions once stream has been received. */
307 if (ctx->on_recv_stream) {
308 ret = ctx->on_recv_stream(stream);
309 if (ret < 0) {
310 goto error;
311 }
312 }
313
d88aee68 314 DBG("UST consumer add stream %s (key: %" PRIu64 ") with relayd id %" PRIu64,
ffe60014
DG
315 stream->name, stream->key, stream->relayd_stream_id);
316
317 /* Set next CPU stream. */
318 channel->streams.count = ++cpu;
d88aee68
DG
319
320 /* Keep stream reference when creating metadata. */
321 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) {
322 channel->metadata_stream = stream;
04ef1097
MD
323 stream->ust_metadata_poll_pipe[0] = ust_metadata_pipe[0];
324 stream->ust_metadata_poll_pipe[1] = ust_metadata_pipe[1];
d88aee68 325 }
ffe60014
DG
326 }
327
328 return 0;
329
330error:
331error_alloc:
332 return ret;
333}
334
335/*
336 * Create an UST channel with the given attributes and send it to the session
337 * daemon using the ust ctl API.
338 *
339 * Return 0 on success or else a negative value.
340 */
341static int create_ust_channel(struct ustctl_consumer_channel_attr *attr,
342 struct ustctl_consumer_channel **chanp)
343{
344 int ret;
345 struct ustctl_consumer_channel *channel;
346
347 assert(attr);
348 assert(chanp);
349
350 DBG3("Creating channel to ustctl with attr: [overwrite: %d, "
351 "subbuf_size: %" PRIu64 ", num_subbuf: %" PRIu64 ", "
352 "switch_timer_interval: %u, read_timer_interval: %u, "
353 "output: %d, type: %d", attr->overwrite, attr->subbuf_size,
354 attr->num_subbuf, attr->switch_timer_interval,
355 attr->read_timer_interval, attr->output, attr->type);
356
357 channel = ustctl_create_channel(attr);
358 if (!channel) {
359 ret = -1;
360 goto error_create;
361 }
362
363 *chanp = channel;
364
365 return 0;
366
367error_create:
368 return ret;
369}
370
d88aee68
DG
371/*
372 * Send a single given stream to the session daemon using the sock.
373 *
374 * Return 0 on success else a negative value.
375 */
ffe60014
DG
376static int send_sessiond_stream(int sock, struct lttng_consumer_stream *stream)
377{
378 int ret;
379
380 assert(stream);
381 assert(sock >= 0);
382
3eb914c0 383 DBG("UST consumer sending stream %" PRIu64 " to sessiond", stream->key);
ffe60014
DG
384
385 /* Send stream to session daemon. */
386 ret = ustctl_send_stream_to_sessiond(sock, stream->ustream);
387 if (ret < 0) {
388 goto error;
389 }
390
ffe60014
DG
391error:
392 return ret;
393}
394
395/*
396 * Send channel to sessiond.
397 *
d88aee68 398 * Return 0 on success or else a negative value.
ffe60014
DG
399 */
400static int send_sessiond_channel(int sock,
401 struct lttng_consumer_channel *channel,
402 struct lttng_consumer_local_data *ctx, int *relayd_error)
403{
f2a444f1 404 int ret, ret_code = LTTNG_OK;
ffe60014
DG
405 struct lttng_consumer_stream *stream;
406
407 assert(channel);
408 assert(ctx);
409 assert(sock >= 0);
410
411 DBG("UST consumer sending channel %s to sessiond", channel->name);
412
62285ea4
DG
413 if (channel->relayd_id != (uint64_t) -1ULL) {
414 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
415 /* Try to send the stream to the relayd if one is available. */
416 ret = consumer_send_relayd_stream(stream, stream->chan->pathname);
417 if (ret < 0) {
418 /*
419 * Flag that the relayd was the problem here probably due to a
420 * communicaton error on the socket.
421 */
422 if (relayd_error) {
423 *relayd_error = 1;
424 }
425 ret_code = LTTNG_ERR_RELAYD_CONNECT_FAIL;
ffe60014 426 }
ffe60014 427 }
f2a444f1 428 }
ffe60014 429
f2a444f1
DG
430 /* Inform sessiond that we are about to send channel and streams. */
431 ret = consumer_send_status_msg(sock, ret_code);
432 if (ret < 0 || ret_code != LTTNG_OK) {
433 /*
434 * Either the session daemon is not responding or the relayd died so we
435 * stop now.
436 */
437 goto error;
438 }
439
440 /* Send channel to sessiond. */
441 ret = ustctl_send_channel_to_sessiond(sock, channel->uchan);
442 if (ret < 0) {
443 goto error;
444 }
445
446 ret = ustctl_channel_close_wakeup_fd(channel->uchan);
447 if (ret < 0) {
448 goto error;
449 }
450
451 /* The channel was sent successfully to the sessiond at this point. */
452 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
ffe60014
DG
453 /* Send stream to session daemon. */
454 ret = send_sessiond_stream(sock, stream);
455 if (ret < 0) {
456 goto error;
457 }
458 }
459
460 /* Tell sessiond there is no more stream. */
461 ret = ustctl_send_stream_to_sessiond(sock, NULL);
462 if (ret < 0) {
463 goto error;
464 }
465
466 DBG("UST consumer NULL stream sent to sessiond");
467
468 return 0;
469
470error:
f2a444f1
DG
471 if (ret_code != LTTNG_OK) {
472 ret = -1;
473 }
ffe60014
DG
474 return ret;
475}
476
477/*
478 * Creates a channel and streams and add the channel it to the channel internal
479 * state. The created stream must ONLY be sent once the GET_CHANNEL command is
480 * received.
481 *
482 * Return 0 on success or else, a negative value is returned and the channel
483 * MUST be destroyed by consumer_del_channel().
484 */
485static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
486 struct lttng_consumer_channel *channel,
487 struct ustctl_consumer_channel_attr *attr)
3bd1e081
MD
488{
489 int ret;
490
ffe60014
DG
491 assert(ctx);
492 assert(channel);
493 assert(attr);
494
495 /*
496 * This value is still used by the kernel consumer since for the kernel,
497 * the stream ownership is not IN the consumer so we need to have the
498 * number of left stream that needs to be initialized so we can know when
499 * to delete the channel (see consumer.c).
500 *
501 * As for the user space tracer now, the consumer creates and sends the
502 * stream to the session daemon which only sends them to the application
503 * once every stream of a channel is received making this value useless
504 * because we they will be added to the poll thread before the application
505 * receives them. This ensures that a stream can not hang up during
506 * initilization of a channel.
507 */
508 channel->nb_init_stream_left = 0;
509
510 /* The reply msg status is handled in the following call. */
511 ret = create_ust_channel(attr, &channel->uchan);
512 if (ret < 0) {
10a50311 513 goto end;
3bd1e081
MD
514 }
515
d8ef542d
MD
516 channel->wait_fd = ustctl_channel_get_wait_fd(channel->uchan);
517
10a50311
JD
518 /*
519 * For the snapshots (no monitor), we create the metadata streams
520 * on demand, not during the channel creation.
521 */
522 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && !channel->monitor) {
523 ret = 0;
524 goto end;
525 }
526
ffe60014
DG
527 /* Open all streams for this channel. */
528 ret = create_ust_streams(channel, ctx);
529 if (ret < 0) {
10a50311 530 goto end;
ffe60014
DG
531 }
532
10a50311 533end:
3bd1e081
MD
534 return ret;
535}
536
d88aee68
DG
537/*
538 * Send all stream of a channel to the right thread handling it.
539 *
540 * On error, return a negative value else 0 on success.
541 */
542static int send_streams_to_thread(struct lttng_consumer_channel *channel,
543 struct lttng_consumer_local_data *ctx)
544{
545 int ret = 0;
546 struct lttng_consumer_stream *stream, *stmp;
547
548 assert(channel);
549 assert(ctx);
550
551 /* Send streams to the corresponding thread. */
552 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
553 send_node) {
554 /* Sending the stream to the thread. */
555 ret = send_stream_to_thread(stream, ctx);
556 if (ret < 0) {
557 /*
558 * If we are unable to send the stream to the thread, there is
559 * a big problem so just stop everything.
560 */
5ab66908
MD
561 /* Remove node from the channel stream list. */
562 cds_list_del(&stream->send_node);
d88aee68
DG
563 goto error;
564 }
565
566 /* Remove node from the channel stream list. */
567 cds_list_del(&stream->send_node);
4891ece8 568
d88aee68
DG
569 }
570
571error:
572 return ret;
573}
574
7972aab2
DG
575/*
576 * Flush channel's streams using the given key to retrieve the channel.
577 *
578 * Return 0 on success else an LTTng error code.
579 */
580static int flush_channel(uint64_t chan_key)
581{
582 int ret = 0;
583 struct lttng_consumer_channel *channel;
584 struct lttng_consumer_stream *stream;
585 struct lttng_ht *ht;
586 struct lttng_ht_iter iter;
587
8fd623e0 588 DBG("UST consumer flush channel key %" PRIu64, chan_key);
7972aab2 589
a500c257 590 rcu_read_lock();
7972aab2
DG
591 channel = consumer_find_channel(chan_key);
592 if (!channel) {
8fd623e0 593 ERR("UST consumer flush channel %" PRIu64 " not found", chan_key);
7972aab2
DG
594 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
595 goto error;
596 }
597
598 ht = consumer_data.stream_per_chan_id_ht;
599
600 /* For each stream of the channel id, flush it. */
7972aab2
DG
601 cds_lfht_for_each_entry_duplicate(ht->ht,
602 ht->hash_fct(&channel->key, lttng_ht_seed), ht->match_fct,
603 &channel->key, &iter.iter, stream, node_channel_id.node) {
b8086166 604 ustctl_flush_buffer(stream->ustream, 1);
7972aab2 605 }
7972aab2 606error:
a500c257 607 rcu_read_unlock();
7972aab2
DG
608 return ret;
609}
be2b50c7
DG
610/*
611 * Close metadata stream wakeup_fd using the given key to retrieve the channel.
612 * RCU read side lock MUST be acquired before calling this function.
613 *
614 * NOTE: This function does NOT take any channel nor stream lock.
615 *
616 * Return 0 on success else LTTng error code.
617 */
618static int _close_metadata(struct lttng_consumer_channel *channel)
619{
620 int ret = LTTNG_OK;
621
622 assert(channel);
623 assert(channel->type == CONSUMER_CHANNEL_TYPE_METADATA);
624
625 if (channel->switch_timer_enabled == 1) {
626 DBG("Deleting timer on metadata channel");
627 consumer_timer_switch_stop(channel);
628 }
629
630 if (channel->metadata_stream) {
631 ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream);
632 if (ret < 0) {
633 ERR("UST consumer unable to close fd of metadata (ret: %d)", ret);
634 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
635 }
636
637 if (channel->monitor) {
638 /* Close the read-side in consumer_del_metadata_stream */
639 ret = close(channel->metadata_stream->ust_metadata_poll_pipe[1]);
640 if (ret < 0) {
641 PERROR("Close UST metadata write-side poll pipe");
642 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
643 }
644 }
645 }
646
647 return ret;
648}
7972aab2 649
d88aee68
DG
650/*
651 * Close metadata stream wakeup_fd using the given key to retrieve the channel.
a500c257 652 * RCU read side lock MUST be acquired before calling this function.
d88aee68
DG
653 *
654 * Return 0 on success else an LTTng error code.
655 */
656static int close_metadata(uint64_t chan_key)
657{
ea88ca2a 658 int ret = 0;
d88aee68
DG
659 struct lttng_consumer_channel *channel;
660
8fd623e0 661 DBG("UST consumer close metadata key %" PRIu64, chan_key);
d88aee68
DG
662
663 channel = consumer_find_channel(chan_key);
664 if (!channel) {
84cc9aa0
DG
665 /*
666 * This is possible if the metadata thread has issue a delete because
667 * the endpoint point of the stream hung up. There is no way the
668 * session daemon can know about it thus use a DBG instead of an actual
669 * error.
670 */
671 DBG("UST consumer close metadata %" PRIu64 " not found", chan_key);
d88aee68
DG
672 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
673 goto error;
674 }
675
ea88ca2a 676 pthread_mutex_lock(&consumer_data.lock);
a9838785 677 pthread_mutex_lock(&channel->lock);
73811ecc
DG
678
679 if (cds_lfht_is_node_deleted(&channel->node.node)) {
680 goto error_unlock;
681 }
682
be2b50c7 683 ret = _close_metadata(channel);
d88aee68 684
ea88ca2a 685error_unlock:
a9838785 686 pthread_mutex_unlock(&channel->lock);
ea88ca2a 687 pthread_mutex_unlock(&consumer_data.lock);
d88aee68
DG
688error:
689 return ret;
690}
691
692/*
693 * RCU read side lock MUST be acquired before calling this function.
694 *
695 * Return 0 on success else an LTTng error code.
696 */
697static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
698{
699 int ret;
700 struct lttng_consumer_channel *metadata;
701
8fd623e0 702 DBG("UST consumer setup metadata key %" PRIu64, key);
d88aee68
DG
703
704 metadata = consumer_find_channel(key);
705 if (!metadata) {
706 ERR("UST consumer push metadata %" PRIu64 " not found", key);
707 ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
10a50311
JD
708 goto end;
709 }
710
711 /*
712 * In no monitor mode, the metadata channel has no stream(s) so skip the
713 * ownership transfer to the metadata thread.
714 */
715 if (!metadata->monitor) {
716 DBG("Metadata channel in no monitor");
717 ret = 0;
718 goto end;
d88aee68
DG
719 }
720
721 /*
722 * Send metadata stream to relayd if one available. Availability is
723 * known if the stream is still in the list of the channel.
724 */
725 if (cds_list_empty(&metadata->streams.head)) {
726 ERR("Metadata channel key %" PRIu64 ", no stream available.", key);
727 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
f5a0c9cf 728 goto error_no_stream;
d88aee68
DG
729 }
730
731 /* Send metadata stream to relayd if needed. */
62285ea4
DG
732 if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) {
733 ret = consumer_send_relayd_stream(metadata->metadata_stream,
734 metadata->pathname);
735 if (ret < 0) {
736 ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
737 goto error;
738 }
d88aee68
DG
739 }
740
741 ret = send_streams_to_thread(metadata, ctx);
742 if (ret < 0) {
743 /*
744 * If we are unable to send the stream to the thread, there is
745 * a big problem so just stop everything.
746 */
747 ret = LTTCOMM_CONSUMERD_FATAL;
748 goto error;
749 }
750 /* List MUST be empty after or else it could be reused. */
751 assert(cds_list_empty(&metadata->streams.head));
752
10a50311
JD
753 ret = 0;
754 goto end;
d88aee68
DG
755
756error:
f2a444f1
DG
757 /*
758 * Delete metadata channel on error. At this point, the metadata stream can
759 * NOT be monitored by the metadata thread thus having the guarantee that
760 * the stream is still in the local stream list of the channel. This call
761 * will make sure to clean that list.
762 */
f5a0c9cf
DG
763 cds_list_del(&metadata->metadata_stream->send_node);
764 consumer_stream_destroy(metadata->metadata_stream, NULL);
765error_no_stream:
10a50311
JD
766end:
767 return ret;
768}
769
770/*
771 * Snapshot the whole metadata.
772 *
773 * Returns 0 on success, < 0 on error
774 */
775static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
776 struct lttng_consumer_local_data *ctx)
777{
778 int ret = 0;
10a50311
JD
779 struct lttng_consumer_channel *metadata_channel;
780 struct lttng_consumer_stream *metadata_stream;
781
782 assert(path);
783 assert(ctx);
784
785 DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s",
786 key, path);
787
788 rcu_read_lock();
789
790 metadata_channel = consumer_find_channel(key);
791 if (!metadata_channel) {
6a00837f
MD
792 ERR("UST snapshot metadata channel not found for key %" PRIu64,
793 key);
10a50311
JD
794 ret = -1;
795 goto error;
796 }
797 assert(!metadata_channel->monitor);
798
799 /*
800 * Ask the sessiond if we have new metadata waiting and update the
801 * consumer metadata cache.
802 */
94d49140 803 ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 1);
10a50311
JD
804 if (ret < 0) {
805 goto error;
806 }
807
808 /*
809 * The metadata stream is NOT created in no monitor mode when the channel
810 * is created on a sessiond ask channel command.
811 */
812 ret = create_ust_streams(metadata_channel, ctx);
813 if (ret < 0) {
814 goto error;
815 }
816
817 metadata_stream = metadata_channel->metadata_stream;
818 assert(metadata_stream);
819
820 if (relayd_id != (uint64_t) -1ULL) {
821 metadata_stream->net_seq_idx = relayd_id;
822 ret = consumer_send_relayd_stream(metadata_stream, path);
823 if (ret < 0) {
824 goto error_stream;
825 }
826 } else {
827 ret = utils_create_stream_file(path, metadata_stream->name,
828 metadata_stream->chan->tracefile_size,
829 metadata_stream->tracefile_count_current,
309167d2 830 metadata_stream->uid, metadata_stream->gid, NULL);
10a50311
JD
831 if (ret < 0) {
832 goto error_stream;
833 }
834 metadata_stream->out_fd = ret;
835 metadata_stream->tracefile_size_current = 0;
836 }
837
04ef1097 838 do {
10a50311
JD
839 ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
840 if (ret < 0) {
94d49140 841 goto error_stream;
10a50311 842 }
04ef1097 843 } while (ret > 0);
10a50311 844
10a50311
JD
845error_stream:
846 /*
847 * Clean up the stream completly because the next snapshot will use a new
848 * metadata stream.
849 */
850 cds_list_del(&metadata_stream->send_node);
851 consumer_stream_destroy(metadata_stream, NULL);
852 metadata_channel->metadata_stream = NULL;
853
854error:
855 rcu_read_unlock();
856 return ret;
857}
858
859/*
860 * Take a snapshot of all the stream of a channel.
861 *
862 * Returns 0 on success, < 0 on error
863 */
864static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
5c786ded 865 uint64_t max_stream_size, struct lttng_consumer_local_data *ctx)
10a50311
JD
866{
867 int ret;
868 unsigned use_relayd = 0;
869 unsigned long consumed_pos, produced_pos;
870 struct lttng_consumer_channel *channel;
871 struct lttng_consumer_stream *stream;
872
873 assert(path);
874 assert(ctx);
875
876 rcu_read_lock();
877
878 if (relayd_id != (uint64_t) -1ULL) {
879 use_relayd = 1;
880 }
881
882 channel = consumer_find_channel(key);
883 if (!channel) {
6a00837f 884 ERR("UST snapshot channel not found for key %" PRIu64, key);
10a50311
JD
885 ret = -1;
886 goto error;
887 }
888 assert(!channel->monitor);
6a00837f 889 DBG("UST consumer snapshot channel %" PRIu64, key);
10a50311
JD
890
891 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
892 /* Lock stream because we are about to change its state. */
893 pthread_mutex_lock(&stream->lock);
894 stream->net_seq_idx = relayd_id;
895
896 if (use_relayd) {
897 ret = consumer_send_relayd_stream(stream, path);
898 if (ret < 0) {
899 goto error_unlock;
900 }
901 } else {
902 ret = utils_create_stream_file(path, stream->name,
903 stream->chan->tracefile_size,
904 stream->tracefile_count_current,
309167d2 905 stream->uid, stream->gid, NULL);
10a50311
JD
906 if (ret < 0) {
907 goto error_unlock;
908 }
909 stream->out_fd = ret;
910 stream->tracefile_size_current = 0;
911
912 DBG("UST consumer snapshot stream %s/%s (%" PRIu64 ")", path,
913 stream->name, stream->key);
914 }
915
916 ustctl_flush_buffer(stream->ustream, 1);
917
918 ret = lttng_ustconsumer_take_snapshot(stream);
919 if (ret < 0) {
920 ERR("Taking UST snapshot");
921 goto error_unlock;
922 }
923
924 ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
925 if (ret < 0) {
926 ERR("Produced UST snapshot position");
927 goto error_unlock;
928 }
929
930 ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
931 if (ret < 0) {
932 ERR("Consumerd UST snapshot position");
933 goto error_unlock;
934 }
935
5c786ded
JD
936 /*
937 * The original value is sent back if max stream size is larger than
938 * the possible size of the snapshot. Also, we asume that the session
939 * daemon should never send a maximum stream size that is lower than
940 * subbuffer size.
941 */
942 consumed_pos = consumer_get_consumed_maxsize(consumed_pos,
943 produced_pos, max_stream_size);
944
10a50311
JD
945 while (consumed_pos < produced_pos) {
946 ssize_t read_len;
947 unsigned long len, padded_len;
948
949 DBG("UST consumer taking snapshot at pos %lu", consumed_pos);
950
951 ret = ustctl_get_subbuf(stream->ustream, &consumed_pos);
952 if (ret < 0) {
953 if (ret != -EAGAIN) {
954 PERROR("ustctl_get_subbuf snapshot");
955 goto error_close_stream;
956 }
957 DBG("UST consumer get subbuf failed. Skipping it.");
958 consumed_pos += stream->max_sb_size;
959 continue;
960 }
961
962 ret = ustctl_get_subbuf_size(stream->ustream, &len);
963 if (ret < 0) {
964 ERR("Snapshot ustctl_get_subbuf_size");
965 goto error_put_subbuf;
966 }
967
968 ret = ustctl_get_padded_subbuf_size(stream->ustream, &padded_len);
969 if (ret < 0) {
970 ERR("Snapshot ustctl_get_padded_subbuf_size");
971 goto error_put_subbuf;
972 }
973
974 read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
309167d2 975 padded_len - len, NULL);
10a50311
JD
976 if (use_relayd) {
977 if (read_len != len) {
56591bac 978 ret = -EPERM;
10a50311
JD
979 goto error_put_subbuf;
980 }
981 } else {
982 if (read_len != padded_len) {
56591bac 983 ret = -EPERM;
10a50311
JD
984 goto error_put_subbuf;
985 }
986 }
987
988 ret = ustctl_put_subbuf(stream->ustream);
989 if (ret < 0) {
990 ERR("Snapshot ustctl_put_subbuf");
991 goto error_close_stream;
992 }
993 consumed_pos += stream->max_sb_size;
994 }
995
996 /* Simply close the stream so we can use it on the next snapshot. */
997 consumer_stream_close(stream);
998 pthread_mutex_unlock(&stream->lock);
999 }
1000
1001 rcu_read_unlock();
1002 return 0;
1003
1004error_put_subbuf:
1005 if (ustctl_put_subbuf(stream->ustream) < 0) {
1006 ERR("Snapshot ustctl_put_subbuf");
1007 }
1008error_close_stream:
1009 consumer_stream_close(stream);
1010error_unlock:
1011 pthread_mutex_unlock(&stream->lock);
1012error:
1013 rcu_read_unlock();
d88aee68
DG
1014 return ret;
1015}
1016
331744e3
JD
1017/*
1018 * Receive the metadata updates from the sessiond.
1019 */
1020int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
5e41ebe1 1021 uint64_t len, struct lttng_consumer_channel *channel,
94d49140 1022 int timer, int wait)
331744e3
JD
1023{
1024 int ret, ret_code = LTTNG_OK;
1025 char *metadata_str;
1026
8fd623e0 1027 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len);
331744e3
JD
1028
1029 metadata_str = zmalloc(len * sizeof(char));
1030 if (!metadata_str) {
1031 PERROR("zmalloc metadata string");
1032 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
1033 goto end;
1034 }
1035
1036 /* Receive metadata string. */
1037 ret = lttcomm_recv_unix_sock(sock, metadata_str, len);
1038 if (ret < 0) {
1039 /* Session daemon is dead so return gracefully. */
1040 ret_code = ret;
1041 goto end_free;
1042 }
1043
1044 pthread_mutex_lock(&channel->metadata_cache->lock);
1045 ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
1046 if (ret < 0) {
1047 /* Unable to handle metadata. Notify session daemon. */
1048 ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
a32bd775
DG
1049 /*
1050 * Skip metadata flush on write error since the offset and len might
1051 * not have been updated which could create an infinite loop below when
1052 * waiting for the metadata cache to be flushed.
1053 */
1054 pthread_mutex_unlock(&channel->metadata_cache->lock);
a32bd775 1055 goto end_free;
331744e3
JD
1056 }
1057 pthread_mutex_unlock(&channel->metadata_cache->lock);
1058
94d49140
JD
1059 if (!wait) {
1060 goto end_free;
1061 }
5e41ebe1 1062 while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
331744e3
JD
1063 DBG("Waiting for metadata to be flushed");
1064 usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
1065 }
1066
1067end_free:
1068 free(metadata_str);
1069end:
1070 return ret_code;
1071}
1072
4cbc1a04
DG
1073/*
1074 * Receive command from session daemon and process it.
1075 *
1076 * Return 1 on success else a negative value or 0.
1077 */
3bd1e081
MD
1078int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1079 int sock, struct pollfd *consumer_sockpoll)
1080{
1081 ssize_t ret;
f50f23d9 1082 enum lttng_error_code ret_code = LTTNG_OK;
3bd1e081 1083 struct lttcomm_consumer_msg msg;
ffe60014 1084 struct lttng_consumer_channel *channel = NULL;
3bd1e081
MD
1085
1086 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
1087 if (ret != sizeof(msg)) {
173af62f
DG
1088 DBG("Consumer received unexpected message size %zd (expects %zu)",
1089 ret, sizeof(msg));
3be74084
DG
1090 /*
1091 * The ret value might 0 meaning an orderly shutdown but this is ok
1092 * since the caller handles this.
1093 */
489f70e9 1094 if (ret > 0) {
c6857fcf 1095 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
489f70e9
MD
1096 ret = -1;
1097 }
3bd1e081
MD
1098 return ret;
1099 }
1100 if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
f50f23d9
DG
1101 /*
1102 * Notify the session daemon that the command is completed.
1103 *
1104 * On transport layer error, the function call will print an error
1105 * message so handling the returned code is a bit useless since we
1106 * return an error code anyway.
1107 */
1108 (void) consumer_send_status_msg(sock, ret_code);
3bd1e081
MD
1109 return -ENOENT;
1110 }
1111
3f8e211f 1112 /* relayd needs RCU read-side lock */
b0b335c8
MD
1113 rcu_read_lock();
1114
3bd1e081 1115 switch (msg.cmd_type) {
00e2e675
DG
1116 case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
1117 {
f50f23d9 1118 /* Session daemon status message are handled in the following call. */
7735ef9e
DG
1119 ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
1120 msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
d3e2ba59
JD
1121 &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
1122 msg.u.relayd_sock.relayd_session_id);
00e2e675
DG
1123 goto end_nosignal;
1124 }
173af62f
DG
1125 case LTTNG_CONSUMER_DESTROY_RELAYD:
1126 {
a6ba4fe1 1127 uint64_t index = msg.u.destroy_relayd.net_seq_idx;
173af62f
DG
1128 struct consumer_relayd_sock_pair *relayd;
1129
a6ba4fe1 1130 DBG("UST consumer destroying relayd %" PRIu64, index);
173af62f
DG
1131
1132 /* Get relayd reference if exists. */
a6ba4fe1 1133 relayd = consumer_find_relayd(index);
173af62f 1134 if (relayd == NULL) {
3448e266 1135 DBG("Unable to find relayd %" PRIu64, index);
f50f23d9 1136 ret_code = LTTNG_ERR_NO_CONSUMER;
173af62f
DG
1137 }
1138
a6ba4fe1
DG
1139 /*
1140 * Each relayd socket pair has a refcount of stream attached to it
1141 * which tells if the relayd is still active or not depending on the
1142 * refcount value.
1143 *
1144 * This will set the destroy flag of the relayd object and destroy it
1145 * if the refcount reaches zero when called.
1146 *
1147 * The destroy can happen either here or when a stream fd hangs up.
1148 */
f50f23d9
DG
1149 if (relayd) {
1150 consumer_flag_relayd_for_destroy(relayd);
1151 }
1152
d88aee68 1153 goto end_msg_sessiond;
173af62f 1154 }
3bd1e081
MD
1155 case LTTNG_CONSUMER_UPDATE_STREAM:
1156 {
3f8e211f 1157 rcu_read_unlock();
7ad0a0cb 1158 return -ENOSYS;
3bd1e081 1159 }
6d805429 1160 case LTTNG_CONSUMER_DATA_PENDING:
53632229 1161 {
3be74084 1162 int ret, is_data_pending;
6d805429 1163 uint64_t id = msg.u.data_pending.session_id;
ca22feea 1164
6d805429 1165 DBG("UST consumer data pending command for id %" PRIu64, id);
ca22feea 1166
3be74084 1167 is_data_pending = consumer_data_pending(id);
ca22feea
DG
1168
1169 /* Send back returned value to session daemon */
3be74084
DG
1170 ret = lttcomm_send_unix_sock(sock, &is_data_pending,
1171 sizeof(is_data_pending));
ca22feea 1172 if (ret < 0) {
3be74084 1173 DBG("Error when sending the data pending ret code: %d", ret);
489f70e9 1174 goto error_fatal;
ca22feea 1175 }
f50f23d9
DG
1176
1177 /*
1178 * No need to send back a status message since the data pending
1179 * returned value is the response.
1180 */
ca22feea 1181 break;
53632229 1182 }
ffe60014
DG
1183 case LTTNG_CONSUMER_ASK_CHANNEL_CREATION:
1184 {
1185 int ret;
1186 struct ustctl_consumer_channel_attr attr;
1187
1188 /* Create a plain object and reserve a channel key. */
1189 channel = allocate_channel(msg.u.ask_channel.session_id,
1190 msg.u.ask_channel.pathname, msg.u.ask_channel.name,
1191 msg.u.ask_channel.uid, msg.u.ask_channel.gid,
1192 msg.u.ask_channel.relayd_id, msg.u.ask_channel.key,
1624d5b7
JD
1193 (enum lttng_event_output) msg.u.ask_channel.output,
1194 msg.u.ask_channel.tracefile_size,
2bba9e53 1195 msg.u.ask_channel.tracefile_count,
1950109e 1196 msg.u.ask_channel.session_id_per_pid,
ecc48a90
JD
1197 msg.u.ask_channel.monitor,
1198 msg.u.ask_channel.live_timer_interval);
ffe60014
DG
1199 if (!channel) {
1200 goto end_channel_error;
1201 }
1202
567eb353
DG
1203 /*
1204 * Assign UST application UID to the channel. This value is ignored for
1205 * per PID buffers. This is specific to UST thus setting this after the
1206 * allocation.
1207 */
1208 channel->ust_app_uid = msg.u.ask_channel.ust_app_uid;
1209
ffe60014
DG
1210 /* Build channel attributes from received message. */
1211 attr.subbuf_size = msg.u.ask_channel.subbuf_size;
1212 attr.num_subbuf = msg.u.ask_channel.num_subbuf;
1213 attr.overwrite = msg.u.ask_channel.overwrite;
1214 attr.switch_timer_interval = msg.u.ask_channel.switch_timer_interval;
1215 attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
7972aab2 1216 attr.chan_id = msg.u.ask_channel.chan_id;
10a50311 1217 attr.output = msg.u.ask_channel.output;
ffe60014
DG
1218 memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
1219
ffe60014
DG
1220 /* Translate and save channel type. */
1221 switch (msg.u.ask_channel.type) {
1222 case LTTNG_UST_CHAN_PER_CPU:
1223 channel->type = CONSUMER_CHANNEL_TYPE_DATA;
1224 attr.type = LTTNG_UST_CHAN_PER_CPU;
8633d6e3
MD
1225 /*
1226 * Set refcount to 1 for owner. Below, we will
1227 * pass ownership to the
1228 * consumer_thread_channel_poll() thread.
1229 */
1230 channel->refcount = 1;
ffe60014
DG
1231 break;
1232 case LTTNG_UST_CHAN_METADATA:
1233 channel->type = CONSUMER_CHANNEL_TYPE_METADATA;
1234 attr.type = LTTNG_UST_CHAN_METADATA;
1235 break;
1236 default:
1237 assert(0);
1238 goto error_fatal;
1239 };
1240
1241 ret = ask_channel(ctx, sock, channel, &attr);
1242 if (ret < 0) {
1243 goto end_channel_error;
1244 }
1245
fc643247
MD
1246 if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) {
1247 ret = consumer_metadata_cache_allocate(channel);
1248 if (ret < 0) {
1249 ERR("Allocating metadata cache");
1250 goto end_channel_error;
1251 }
1252 consumer_timer_switch_start(channel, attr.switch_timer_interval);
1253 attr.switch_timer_interval = 0;
94d49140
JD
1254 } else {
1255 consumer_timer_live_start(channel,
1256 msg.u.ask_channel.live_timer_interval);
fc643247
MD
1257 }
1258
ffe60014
DG
1259 /*
1260 * Add the channel to the internal state AFTER all streams were created
1261 * and successfully sent to session daemon. This way, all streams must
1262 * be ready before this channel is visible to the threads.
fc643247
MD
1263 * If add_channel succeeds, ownership of the channel is
1264 * passed to consumer_thread_channel_poll().
ffe60014
DG
1265 */
1266 ret = add_channel(channel, ctx);
1267 if (ret < 0) {
ea88ca2a
MD
1268 if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) {
1269 if (channel->switch_timer_enabled == 1) {
1270 consumer_timer_switch_stop(channel);
1271 }
1272 consumer_metadata_cache_destroy(channel);
1273 }
d3e2ba59
JD
1274 if (channel->live_timer_enabled == 1) {
1275 consumer_timer_live_stop(channel);
1276 }
ffe60014
DG
1277 goto end_channel_error;
1278 }
1279
1280 /*
1281 * Channel and streams are now created. Inform the session daemon that
1282 * everything went well and should wait to receive the channel and
1283 * streams with ustctl API.
1284 */
1285 ret = consumer_send_status_channel(sock, channel);
1286 if (ret < 0) {
1287 /*
489f70e9 1288 * There is probably a problem on the socket.
ffe60014 1289 */
489f70e9 1290 goto error_fatal;
ffe60014
DG
1291 }
1292
1293 break;
1294 }
1295 case LTTNG_CONSUMER_GET_CHANNEL:
1296 {
1297 int ret, relayd_err = 0;
d88aee68 1298 uint64_t key = msg.u.get_channel.key;
ffe60014 1299 struct lttng_consumer_channel *channel;
ffe60014
DG
1300
1301 channel = consumer_find_channel(key);
1302 if (!channel) {
8fd623e0 1303 ERR("UST consumer get channel key %" PRIu64 " not found", key);
ffe60014
DG
1304 ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND;
1305 goto end_msg_sessiond;
1306 }
1307
ffe60014
DG
1308 /* Send everything to sessiond. */
1309 ret = send_sessiond_channel(sock, channel, ctx, &relayd_err);
1310 if (ret < 0) {
1311 if (relayd_err) {
1312 /*
1313 * We were unable to send to the relayd the stream so avoid
1314 * sending back a fatal error to the thread since this is OK
f2a444f1
DG
1315 * and the consumer can continue its work. The above call
1316 * has sent the error status message to the sessiond.
ffe60014 1317 */
f2a444f1 1318 goto end_nosignal;
ffe60014
DG
1319 }
1320 /*
1321 * The communicaton was broken hence there is a bad state between
1322 * the consumer and sessiond so stop everything.
1323 */
1324 goto error_fatal;
1325 }
1326
10a50311
JD
1327 /*
1328 * In no monitor mode, the streams ownership is kept inside the channel
1329 * so don't send them to the data thread.
1330 */
1331 if (!channel->monitor) {
1332 goto end_msg_sessiond;
1333 }
1334
d88aee68
DG
1335 ret = send_streams_to_thread(channel, ctx);
1336 if (ret < 0) {
1337 /*
1338 * If we are unable to send the stream to the thread, there is
1339 * a big problem so just stop everything.
1340 */
1341 goto error_fatal;
ffe60014 1342 }
ffe60014
DG
1343 /* List MUST be empty after or else it could be reused. */
1344 assert(cds_list_empty(&channel->streams.head));
d88aee68
DG
1345 goto end_msg_sessiond;
1346 }
1347 case LTTNG_CONSUMER_DESTROY_CHANNEL:
1348 {
1349 uint64_t key = msg.u.destroy_channel.key;
d88aee68 1350
a0cbdd2e
MD
1351 /*
1352 * Only called if streams have not been sent to stream
1353 * manager thread. However, channel has been sent to
1354 * channel manager thread.
1355 */
1356 notify_thread_del_channel(ctx, key);
d88aee68 1357 goto end_msg_sessiond;
ffe60014 1358 }
d88aee68
DG
1359 case LTTNG_CONSUMER_CLOSE_METADATA:
1360 {
1361 int ret;
1362
1363 ret = close_metadata(msg.u.close_metadata.key);
1364 if (ret != 0) {
1365 ret_code = ret;
1366 }
1367
1368 goto end_msg_sessiond;
1369 }
7972aab2
DG
1370 case LTTNG_CONSUMER_FLUSH_CHANNEL:
1371 {
1372 int ret;
1373
1374 ret = flush_channel(msg.u.flush_channel.key);
1375 if (ret != 0) {
1376 ret_code = ret;
1377 }
1378
1379 goto end_msg_sessiond;
1380 }
d88aee68 1381 case LTTNG_CONSUMER_PUSH_METADATA:
ffe60014
DG
1382 {
1383 int ret;
d88aee68 1384 uint64_t len = msg.u.push_metadata.len;
d88aee68 1385 uint64_t key = msg.u.push_metadata.key;
331744e3 1386 uint64_t offset = msg.u.push_metadata.target_offset;
ffe60014
DG
1387 struct lttng_consumer_channel *channel;
1388
8fd623e0
DG
1389 DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key,
1390 len);
ffe60014
DG
1391
1392 channel = consumer_find_channel(key);
1393 if (!channel) {
8fd623e0 1394 ERR("UST consumer push metadata %" PRIu64 " not found", key);
ffe60014 1395 ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND;
4a2eb0ca 1396 goto end_msg_sessiond;
d88aee68
DG
1397 }
1398
1399 /* Tell session daemon we are ready to receive the metadata. */
ffe60014
DG
1400 ret = consumer_send_status_msg(sock, LTTNG_OK);
1401 if (ret < 0) {
1402 /* Somehow, the session daemon is not responding anymore. */
d88aee68
DG
1403 goto error_fatal;
1404 }
1405
1406 /* Wait for more data. */
1407 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
489f70e9 1408 goto error_fatal;
d88aee68
DG
1409 }
1410
331744e3 1411 ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
94d49140 1412 len, channel, 0, 1);
d88aee68 1413 if (ret < 0) {
331744e3 1414 /* error receiving from sessiond */
489f70e9 1415 goto error_fatal;
331744e3
JD
1416 } else {
1417 ret_code = ret;
d88aee68
DG
1418 goto end_msg_sessiond;
1419 }
d88aee68
DG
1420 }
1421 case LTTNG_CONSUMER_SETUP_METADATA:
1422 {
1423 int ret;
1424
1425 ret = setup_metadata(ctx, msg.u.setup_metadata.key);
1426 if (ret) {
1427 ret_code = ret;
1428 }
1429 goto end_msg_sessiond;
ffe60014 1430 }
6dc3064a
DG
1431 case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
1432 {
10a50311
JD
1433 if (msg.u.snapshot_channel.metadata) {
1434 ret = snapshot_metadata(msg.u.snapshot_channel.key,
1435 msg.u.snapshot_channel.pathname,
1436 msg.u.snapshot_channel.relayd_id,
1437 ctx);
1438 if (ret < 0) {
1439 ERR("Snapshot metadata failed");
1440 ret_code = LTTNG_ERR_UST_META_FAIL;
1441 }
1442 } else {
1443 ret = snapshot_channel(msg.u.snapshot_channel.key,
1444 msg.u.snapshot_channel.pathname,
1445 msg.u.snapshot_channel.relayd_id,
5c786ded 1446 msg.u.snapshot_channel.max_stream_size,
10a50311
JD
1447 ctx);
1448 if (ret < 0) {
1449 ERR("Snapshot channel failed");
1450 ret_code = LTTNG_ERR_UST_CHAN_FAIL;
1451 }
1452 }
1453
6dc3064a
DG
1454 ret = consumer_send_status_msg(sock, ret_code);
1455 if (ret < 0) {
1456 /* Somehow, the session daemon is not responding anymore. */
1457 goto end_nosignal;
1458 }
1459 break;
1460 }
3bd1e081
MD
1461 default:
1462 break;
1463 }
3f8e211f 1464
3bd1e081 1465end_nosignal:
b0b335c8 1466 rcu_read_unlock();
4cbc1a04
DG
1467
1468 /*
1469 * Return 1 to indicate success since the 0 value can be a socket
1470 * shutdown during the recv() or send() call.
1471 */
1472 return 1;
ffe60014
DG
1473
1474end_msg_sessiond:
1475 /*
1476 * The returned value here is not useful since either way we'll return 1 to
1477 * the caller because the session daemon socket management is done
1478 * elsewhere. Returning a negative code or 0 will shutdown the consumer.
1479 */
489f70e9
MD
1480 ret = consumer_send_status_msg(sock, ret_code);
1481 if (ret < 0) {
1482 goto error_fatal;
1483 }
ffe60014
DG
1484 rcu_read_unlock();
1485 return 1;
1486end_channel_error:
1487 if (channel) {
1488 /*
1489 * Free channel here since no one has a reference to it. We don't
1490 * free after that because a stream can store this pointer.
1491 */
1492 destroy_channel(channel);
1493 }
1494 /* We have to send a status channel message indicating an error. */
1495 ret = consumer_send_status_channel(sock, NULL);
1496 if (ret < 0) {
1497 /* Stop everything if session daemon can not be notified. */
1498 goto error_fatal;
1499 }
1500 rcu_read_unlock();
1501 return 1;
1502error_fatal:
1503 rcu_read_unlock();
1504 /* This will issue a consumer stop. */
1505 return -1;
3bd1e081
MD
1506}
1507
ffe60014
DG
1508/*
1509 * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
1510 * compiled out, we isolate it in this library.
1511 */
1512int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
1513 unsigned long *off)
3bd1e081 1514{
ffe60014
DG
1515 assert(stream);
1516 assert(stream->ustream);
b5c5fc29 1517
ffe60014 1518 return ustctl_get_mmap_read_offset(stream->ustream, off);
3bd1e081
MD
1519}
1520
ffe60014
DG
1521/*
1522 * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
1523 * compiled out, we isolate it in this library.
1524 */
1525void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
d056b477 1526{
ffe60014
DG
1527 assert(stream);
1528 assert(stream->ustream);
1529
1530 return ustctl_get_mmap_base(stream->ustream);
d056b477
MD
1531}
1532
ffe60014
DG
1533/*
1534 * Take a snapshot for a specific fd
1535 *
1536 * Returns 0 on success, < 0 on error
1537 */
1538int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081 1539{
ffe60014
DG
1540 assert(stream);
1541 assert(stream->ustream);
1542
1543 return ustctl_snapshot(stream->ustream);
3bd1e081
MD
1544}
1545
ffe60014
DG
1546/*
1547 * Get the produced position
1548 *
1549 * Returns 0 on success, < 0 on error
1550 */
1551int lttng_ustconsumer_get_produced_snapshot(
1552 struct lttng_consumer_stream *stream, unsigned long *pos)
3bd1e081 1553{
ffe60014
DG
1554 assert(stream);
1555 assert(stream->ustream);
1556 assert(pos);
7a57cf92 1557
ffe60014
DG
1558 return ustctl_snapshot_get_produced(stream->ustream, pos);
1559}
7a57cf92 1560
10a50311
JD
1561/*
1562 * Get the consumed position
1563 *
1564 * Returns 0 on success, < 0 on error
1565 */
1566int lttng_ustconsumer_get_consumed_snapshot(
1567 struct lttng_consumer_stream *stream, unsigned long *pos)
1568{
1569 assert(stream);
1570 assert(stream->ustream);
1571 assert(pos);
1572
1573 return ustctl_snapshot_get_consumed(stream->ustream, pos);
1574}
1575
ffe60014
DG
1576/*
1577 * Called when the stream signal the consumer that it has hang up.
1578 */
1579void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
1580{
1581 assert(stream);
1582 assert(stream->ustream);
2c1dd183 1583
ffe60014
DG
1584 ustctl_flush_buffer(stream->ustream, 0);
1585 stream->hangup_flush_done = 1;
1586}
ee77a7b0 1587
ffe60014
DG
1588void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
1589{
1590 assert(chan);
1591 assert(chan->uchan);
e316aad5 1592
ea88ca2a
MD
1593 if (chan->switch_timer_enabled == 1) {
1594 consumer_timer_switch_stop(chan);
1595 }
1596 consumer_metadata_cache_destroy(chan);
ffe60014 1597 ustctl_destroy_channel(chan->uchan);
3bd1e081
MD
1598}
1599
1600void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
1601{
ffe60014
DG
1602 assert(stream);
1603 assert(stream->ustream);
d41f73b7 1604
ea88ca2a
MD
1605 if (stream->chan->switch_timer_enabled == 1) {
1606 consumer_timer_switch_stop(stream->chan);
1607 }
ffe60014
DG
1608 ustctl_destroy_stream(stream->ustream);
1609}
d41f73b7 1610
309167d2
JD
1611/*
1612 * Populate index values of a UST stream. Values are set in big endian order.
1613 *
1614 * Return 0 on success or else a negative value.
1615 */
1616static int get_index_values(struct lttng_packet_index *index,
1617 struct ustctl_consumer_stream *ustream)
1618{
1619 int ret;
1620
1621 ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin);
1622 if (ret < 0) {
1623 PERROR("ustctl_get_timestamp_begin");
1624 goto error;
1625 }
1626 index->timestamp_begin = htobe64(index->timestamp_begin);
1627
1628 ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end);
1629 if (ret < 0) {
1630 PERROR("ustctl_get_timestamp_end");
1631 goto error;
1632 }
1633 index->timestamp_end = htobe64(index->timestamp_end);
1634
1635 ret = ustctl_get_events_discarded(ustream, &index->events_discarded);
1636 if (ret < 0) {
1637 PERROR("ustctl_get_events_discarded");
1638 goto error;
1639 }
1640 index->events_discarded = htobe64(index->events_discarded);
1641
1642 ret = ustctl_get_content_size(ustream, &index->content_size);
1643 if (ret < 0) {
1644 PERROR("ustctl_get_content_size");
1645 goto error;
1646 }
1647 index->content_size = htobe64(index->content_size);
1648
1649 ret = ustctl_get_packet_size(ustream, &index->packet_size);
1650 if (ret < 0) {
1651 PERROR("ustctl_get_packet_size");
1652 goto error;
1653 }
1654 index->packet_size = htobe64(index->packet_size);
1655
1656 ret = ustctl_get_stream_id(ustream, &index->stream_id);
1657 if (ret < 0) {
1658 PERROR("ustctl_get_stream_id");
1659 goto error;
1660 }
1661 index->stream_id = htobe64(index->stream_id);
1662
1663error:
1664 return ret;
1665}
1666
94d49140
JD
1667/*
1668 * Write up to one packet from the metadata cache to the channel.
1669 *
1670 * Returns the number of bytes pushed in the cache, or a negative value
1671 * on error.
1672 */
1673static
1674int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
1675{
1676 ssize_t write_len;
1677 int ret;
1678
1679 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
1680 if (stream->chan->metadata_cache->contiguous
1681 == stream->ust_metadata_pushed) {
1682 ret = 0;
1683 goto end;
1684 }
1685
1686 write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
1687 &stream->chan->metadata_cache->data[stream->ust_metadata_pushed],
1688 stream->chan->metadata_cache->contiguous
1689 - stream->ust_metadata_pushed);
1690 assert(write_len != 0);
1691 if (write_len < 0) {
1692 ERR("Writing one metadata packet");
1693 ret = -1;
1694 goto end;
1695 }
1696 stream->ust_metadata_pushed += write_len;
1697
1698 assert(stream->chan->metadata_cache->contiguous >=
1699 stream->ust_metadata_pushed);
1700 ret = write_len;
1701
1702end:
1703 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
1704 return ret;
1705}
1706
309167d2 1707
94d49140
JD
1708/*
1709 * Sync metadata meaning request them to the session daemon and snapshot to the
1710 * metadata thread can consumer them.
1711 *
1712 * Metadata stream lock MUST be acquired.
1713 *
1714 * Return 0 if new metadatda is available, EAGAIN if the metadata stream
1715 * is empty or a negative value on error.
1716 */
1717int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
1718 struct lttng_consumer_stream *metadata)
1719{
1720 int ret;
1721 int retry = 0;
1722
1723 assert(ctx);
1724 assert(metadata);
1725
1726 /*
1727 * Request metadata from the sessiond, but don't wait for the flush
1728 * because we locked the metadata thread.
1729 */
1730 ret = lttng_ustconsumer_request_metadata(ctx, metadata->chan, 0, 0);
1731 if (ret < 0) {
1732 goto end;
1733 }
1734
1735 ret = commit_one_metadata_packet(metadata);
1736 if (ret <= 0) {
1737 goto end;
1738 } else if (ret > 0) {
1739 retry = 1;
1740 }
1741
1742 ustctl_flush_buffer(metadata->ustream, 1);
1743 ret = ustctl_snapshot(metadata->ustream);
1744 if (ret < 0) {
1745 if (errno != EAGAIN) {
1746 ERR("Sync metadata, taking UST snapshot");
1747 goto end;
1748 }
1749 DBG("No new metadata when syncing them.");
1750 /* No new metadata, exit. */
1751 ret = ENODATA;
1752 goto end;
1753 }
1754
1755 /*
1756 * After this flush, we still need to extract metadata.
1757 */
1758 if (retry) {
1759 ret = EAGAIN;
1760 }
1761
1762end:
1763 return ret;
1764}
1765
1766/*
1767 * Read subbuffer from the given stream.
1768 *
1769 * Stream lock MUST be acquired.
1770 *
1771 * Return 0 on success else a negative value.
1772 */
d41f73b7
MD
1773int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
1774 struct lttng_consumer_local_data *ctx)
1775{
1d4dfdef 1776 unsigned long len, subbuf_size, padding;
1c20f0e2 1777 int err, write_index = 1;
d41f73b7 1778 long ret = 0;
d41f73b7 1779 char dummy;
ffe60014 1780 struct ustctl_consumer_stream *ustream;
309167d2 1781 struct lttng_packet_index index;
ffe60014
DG
1782
1783 assert(stream);
1784 assert(stream->ustream);
1785 assert(ctx);
d41f73b7 1786
3eb914c0 1787 DBG("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd,
ffe60014
DG
1788 stream->name);
1789
1790 /* Ease our life for what's next. */
1791 ustream = stream->ustream;
d41f73b7
MD
1792
1793 /* We can consume the 1 byte written into the wait_fd by UST */
04ef1097 1794 if (stream->monitor && !stream->hangup_flush_done) {
c617c0c6
MD
1795 ssize_t readlen;
1796
effcf122
MD
1797 do {
1798 readlen = read(stream->wait_fd, &dummy, 1);
87dc6a9c 1799 } while (readlen == -1 && errno == EINTR);
04ef1097 1800 if (readlen == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
effcf122
MD
1801 ret = readlen;
1802 goto end;
1803 }
d41f73b7
MD
1804 }
1805
04ef1097 1806retry:
d41f73b7 1807 /* Get the next subbuffer */
ffe60014 1808 err = ustctl_get_next_subbuf(ustream);
d41f73b7 1809 if (err != 0) {
04ef1097
MD
1810 /*
1811 * Populate metadata info if the existing info has
1812 * already been read.
1813 */
1814 if (stream->metadata_flag) {
94d49140
JD
1815 ret = commit_one_metadata_packet(stream);
1816 if (ret <= 0) {
04ef1097
MD
1817 goto end;
1818 }
04ef1097
MD
1819 ustctl_flush_buffer(stream->ustream, 1);
1820 goto retry;
1821 }
1822
1d4dfdef 1823 ret = err; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
d41f73b7
MD
1824 /*
1825 * This is a debug message even for single-threaded consumer,
1826 * because poll() have more relaxed criterions than get subbuf,
1827 * so get_subbuf may fail for short race windows where poll()
1828 * would issue wakeups.
1829 */
1830 DBG("Reserving sub buffer failed (everything is normal, "
ffe60014 1831 "it is due to concurrency) [ret: %d]", err);
d41f73b7
MD
1832 goto end;
1833 }
ffe60014 1834 assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
309167d2 1835
1c20f0e2 1836 if (!stream->metadata_flag) {
309167d2
JD
1837 index.offset = htobe64(stream->out_fd_offset);
1838 ret = get_index_values(&index, ustream);
1839 if (ret < 0) {
1840 goto end;
1841 }
1c20f0e2
JD
1842 } else {
1843 write_index = 0;
309167d2
JD
1844 }
1845
1d4dfdef 1846 /* Get the full padded subbuffer size */
ffe60014 1847 err = ustctl_get_padded_subbuf_size(ustream, &len);
effcf122 1848 assert(err == 0);
1d4dfdef
DG
1849
1850 /* Get subbuffer data size (without padding) */
ffe60014 1851 err = ustctl_get_subbuf_size(ustream, &subbuf_size);
1d4dfdef
DG
1852 assert(err == 0);
1853
1854 /* Make sure we don't get a subbuffer size bigger than the padded */
1855 assert(len >= subbuf_size);
1856
1857 padding = len - subbuf_size;
d41f73b7 1858 /* write the subbuffer to the tracefile */
309167d2 1859 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
91dfef6e
DG
1860 /*
1861 * The mmap operation should write subbuf_size amount of data when network
1862 * streaming or the full padding (len) size when we are _not_ streaming.
1863 */
d88aee68
DG
1864 if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
1865 (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
d41f73b7 1866 /*
91dfef6e 1867 * Display the error but continue processing to try to release the
c5c45efa
DG
1868 * subbuffer. This is a DBG statement since any unexpected kill or
1869 * signal, the application gets unregistered, relayd gets closed or
1870 * anything that affects the buffer lifetime will trigger this error.
1871 * So, for the sake of the user, don't print this error since it can
1872 * happen and it is OK with the code flow.
d41f73b7 1873 */
c5c45efa 1874 DBG("Error writing to tracefile "
8fd623e0 1875 "(ret: %ld != len: %lu != subbuf_size: %lu)",
91dfef6e 1876 ret, len, subbuf_size);
309167d2 1877 write_index = 0;
d41f73b7 1878 }
ffe60014 1879 err = ustctl_put_next_subbuf(ustream);
effcf122 1880 assert(err == 0);
331744e3 1881
309167d2 1882 /* Write index if needed. */
1c20f0e2
JD
1883 if (!write_index) {
1884 goto end;
1885 }
1886
94d49140
JD
1887 if (stream->chan->live_timer_interval && !stream->metadata_flag) {
1888 /*
1889 * In live, block until all the metadata is sent.
1890 */
1891 err = consumer_stream_sync_metadata(ctx, stream->session_id);
1892 if (err < 0) {
1893 goto end;
1894 }
1895 }
1896
1c20f0e2
JD
1897 assert(!stream->metadata_flag);
1898 err = consumer_stream_write_index(stream, &index);
1899 if (err < 0) {
1900 goto end;
309167d2
JD
1901 }
1902
d41f73b7
MD
1903end:
1904 return ret;
1905}
1906
ffe60014
DG
1907/*
1908 * Called when a stream is created.
fe4477ee
JD
1909 *
1910 * Return 0 on success or else a negative value.
ffe60014 1911 */
d41f73b7
MD
1912int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
1913{
fe4477ee
JD
1914 int ret;
1915
10a50311
JD
1916 assert(stream);
1917
fe4477ee 1918 /* Don't create anything if this is set for streaming. */
10a50311 1919 if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
fe4477ee
JD
1920 ret = utils_create_stream_file(stream->chan->pathname, stream->name,
1921 stream->chan->tracefile_size, stream->tracefile_count_current,
309167d2 1922 stream->uid, stream->gid, NULL);
fe4477ee
JD
1923 if (ret < 0) {
1924 goto error;
1925 }
1926 stream->out_fd = ret;
1927 stream->tracefile_size_current = 0;
309167d2
JD
1928
1929 if (!stream->metadata_flag) {
1930 ret = index_create_file(stream->chan->pathname,
1931 stream->name, stream->uid, stream->gid,
1932 stream->chan->tracefile_size,
1933 stream->tracefile_count_current);
1934 if (ret < 0) {
1935 goto error;
1936 }
1937 stream->index_fd = ret;
1938 }
fe4477ee
JD
1939 }
1940 ret = 0;
1941
1942error:
1943 return ret;
d41f73b7 1944}
ca22feea
DG
1945
1946/*
1947 * Check if data is still being extracted from the buffers for a specific
4e9a4686
DG
1948 * stream. Consumer data lock MUST be acquired before calling this function
1949 * and the stream lock.
ca22feea 1950 *
6d805429 1951 * Return 1 if the traced data are still getting read else 0 meaning that the
ca22feea
DG
1952 * data is available for trace viewer reading.
1953 */
6d805429 1954int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
ca22feea
DG
1955{
1956 int ret;
1957
1958 assert(stream);
ffe60014 1959 assert(stream->ustream);
ca22feea 1960
6d805429 1961 DBG("UST consumer checking data pending");
c8f59ee5 1962
ca6b395f
MD
1963 if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
1964 ret = 0;
1965 goto end;
1966 }
1967
04ef1097 1968 if (stream->chan->type == CONSUMER_CHANNEL_TYPE_METADATA) {
e6ee4eab
DG
1969 uint64_t contiguous, pushed;
1970
1971 /* Ease our life a bit. */
1972 contiguous = stream->chan->metadata_cache->contiguous;
1973 pushed = stream->ust_metadata_pushed;
1974
04ef1097
MD
1975 /*
1976 * We can simply check whether all contiguously available data
1977 * has been pushed to the ring buffer, since the push operation
1978 * is performed within get_next_subbuf(), and because both
1979 * get_next_subbuf() and put_next_subbuf() are issued atomically
1980 * thanks to the stream lock within
1981 * lttng_ustconsumer_read_subbuffer(). This basically means that
1982 * whetnever ust_metadata_pushed is incremented, the associated
1983 * metadata has been consumed from the metadata stream.
1984 */
1985 DBG("UST consumer metadata pending check: contiguous %" PRIu64 " vs pushed %" PRIu64,
e6ee4eab 1986 contiguous, pushed);
6acdf328 1987 assert(((int64_t) contiguous - pushed) >= 0);
e6ee4eab 1988 if ((contiguous != pushed) ||
6acdf328 1989 (((int64_t) contiguous - pushed) > 0 || contiguous == 0)) {
04ef1097
MD
1990 ret = 1; /* Data is pending */
1991 goto end;
1992 }
1993 } else {
1994 ret = ustctl_get_next_subbuf(stream->ustream);
1995 if (ret == 0) {
1996 /*
1997 * There is still data so let's put back this
1998 * subbuffer.
1999 */
2000 ret = ustctl_put_subbuf(stream->ustream);
2001 assert(ret == 0);
2002 ret = 1; /* Data is pending */
2003 goto end;
2004 }
ca22feea
DG
2005 }
2006
6d805429
DG
2007 /* Data is NOT pending so ready to be read. */
2008 ret = 0;
ca22feea 2009
6efae65e
DG
2010end:
2011 return ret;
ca22feea 2012}
d88aee68
DG
2013
2014/*
2015 * Close every metadata stream wait fd of the metadata hash table. This
2016 * function MUST be used very carefully so not to run into a race between the
2017 * metadata thread handling streams and this function closing their wait fd.
2018 *
2019 * For UST, this is used when the session daemon hangs up. Its the metadata
2020 * producer so calling this is safe because we are assured that no state change
2021 * can occur in the metadata thread for the streams in the hash table.
2022 */
2023void lttng_ustconsumer_close_metadata(struct lttng_ht *metadata_ht)
2024{
d88aee68
DG
2025 struct lttng_ht_iter iter;
2026 struct lttng_consumer_stream *stream;
2027
2028 assert(metadata_ht);
2029 assert(metadata_ht->ht);
2030
2031 DBG("UST consumer closing all metadata streams");
2032
2033 rcu_read_lock();
2034 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream,
2035 node.node) {
be2b50c7 2036 pthread_mutex_lock(&stream->chan->lock);
d88aee68 2037 /*
be2b50c7
DG
2038 * Whatever returned value, we must continue to try to close everything
2039 * so ignore it.
d88aee68 2040 */
be2b50c7
DG
2041 (void) _close_metadata(stream->chan);
2042 DBG("Metadata wait fd %d and poll pipe fd %d closed", stream->wait_fd,
2043 stream->ust_metadata_poll_pipe[1]);
2044 pthread_mutex_unlock(&stream->chan->lock);
2045
d88aee68
DG
2046 }
2047 rcu_read_unlock();
2048}
d8ef542d
MD
2049
2050void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
2051{
2052 int ret;
2053
2054 ret = ustctl_stream_close_wakeup_fd(stream->ustream);
2055 if (ret < 0) {
2056 ERR("Unable to close wakeup fd");
2057 }
2058}
331744e3 2059
f666ae70
MD
2060/*
2061 * Please refer to consumer-timer.c before adding any lock within this
2062 * function or any of its callees. Timers have a very strict locking
2063 * semantic with respect to teardown. Failure to respect this semantic
2064 * introduces deadlocks.
2065 */
331744e3 2066int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
94d49140 2067 struct lttng_consumer_channel *channel, int timer, int wait)
331744e3
JD
2068{
2069 struct lttcomm_metadata_request_msg request;
2070 struct lttcomm_consumer_msg msg;
2071 enum lttng_error_code ret_code = LTTNG_OK;
2072 uint64_t len, key, offset;
2073 int ret;
2074
2075 assert(channel);
2076 assert(channel->metadata_cache);
2077
2078 /* send the metadata request to sessiond */
2079 switch (consumer_data.type) {
2080 case LTTNG_CONSUMER64_UST:
2081 request.bits_per_long = 64;
2082 break;
2083 case LTTNG_CONSUMER32_UST:
2084 request.bits_per_long = 32;
2085 break;
2086 default:
2087 request.bits_per_long = 0;
2088 break;
2089 }
2090
2091 request.session_id = channel->session_id;
1950109e 2092 request.session_id_per_pid = channel->session_id_per_pid;
567eb353
DG
2093 /*
2094 * Request the application UID here so the metadata of that application can
2095 * be sent back. The channel UID corresponds to the user UID of the session
2096 * used for the rights on the stream file(s).
2097 */
2098 request.uid = channel->ust_app_uid;
331744e3 2099 request.key = channel->key;
567eb353 2100
1950109e 2101 DBG("Sending metadata request to sessiond, session id %" PRIu64
567eb353
DG
2102 ", per-pid %" PRIu64 ", app UID %u and channek key %" PRIu64,
2103 request.session_id, request.session_id_per_pid, request.uid,
2104 request.key);
331744e3 2105
75d83e50 2106 pthread_mutex_lock(&ctx->metadata_socket_lock);
331744e3
JD
2107 ret = lttcomm_send_unix_sock(ctx->consumer_metadata_socket, &request,
2108 sizeof(request));
2109 if (ret < 0) {
2110 ERR("Asking metadata to sessiond");
2111 goto end;
2112 }
2113
2114 /* Receive the metadata from sessiond */
2115 ret = lttcomm_recv_unix_sock(ctx->consumer_metadata_socket, &msg,
2116 sizeof(msg));
2117 if (ret != sizeof(msg)) {
8fd623e0 2118 DBG("Consumer received unexpected message size %d (expects %zu)",
331744e3
JD
2119 ret, sizeof(msg));
2120 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
2121 /*
2122 * The ret value might 0 meaning an orderly shutdown but this is ok
2123 * since the caller handles this.
2124 */
2125 goto end;
2126 }
2127
2128 if (msg.cmd_type == LTTNG_ERR_UND) {
2129 /* No registry found */
2130 (void) consumer_send_status_msg(ctx->consumer_metadata_socket,
2131 ret_code);
2132 ret = 0;
2133 goto end;
2134 } else if (msg.cmd_type != LTTNG_CONSUMER_PUSH_METADATA) {
2135 ERR("Unexpected cmd_type received %d", msg.cmd_type);
2136 ret = -1;
2137 goto end;
2138 }
2139
2140 len = msg.u.push_metadata.len;
2141 key = msg.u.push_metadata.key;
2142 offset = msg.u.push_metadata.target_offset;
2143
2144 assert(key == channel->key);
2145 if (len == 0) {
2146 DBG("No new metadata to receive for key %" PRIu64, key);
2147 }
2148
2149 /* Tell session daemon we are ready to receive the metadata. */
2150 ret = consumer_send_status_msg(ctx->consumer_metadata_socket,
2151 LTTNG_OK);
2152 if (ret < 0 || len == 0) {
2153 /*
2154 * Somehow, the session daemon is not responding anymore or there is
2155 * nothing to receive.
2156 */
2157 goto end;
2158 }
2159
2160 ret_code = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
94d49140 2161 key, offset, len, channel, timer, wait);
f2a444f1
DG
2162 if (ret_code >= 0) {
2163 /*
2164 * Only send the status msg if the sessiond is alive meaning a positive
2165 * ret code.
2166 */
2167 (void) consumer_send_status_msg(ctx->consumer_metadata_socket, ret_code);
2168 }
331744e3
JD
2169 ret = 0;
2170
2171end:
75d83e50 2172 pthread_mutex_unlock(&ctx->metadata_socket_lock);
331744e3
JD
2173 return ret;
2174}
This page took 0.243658 seconds and 4 git commands to generate.