Introduce channel timer lock
[lttng-tools.git] / src / common / consumer.c
... / ...
CommitLineData
1/*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * 2012 - David Goulet <dgoulet@efficios.com>
5 *
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License, version 2 only,
8 * as published by the Free Software Foundation.
9 *
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
13 * more details.
14 *
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18 */
19
20#define _GNU_SOURCE
21#include <assert.h>
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>
28#include <sys/types.h>
29#include <unistd.h>
30#include <inttypes.h>
31#include <signal.h>
32
33#include <common/common.h>
34#include <common/utils.h>
35#include <common/compat/poll.h>
36#include <common/kernel-ctl/kernel-ctl.h>
37#include <common/sessiond-comm/relayd.h>
38#include <common/sessiond-comm/sessiond-comm.h>
39#include <common/kernel-consumer/kernel-consumer.h>
40#include <common/relayd/relayd.h>
41#include <common/ust-consumer/ust-consumer.h>
42
43#include "consumer.h"
44#include "consumer-stream.h"
45
46struct lttng_consumer_global_data consumer_data = {
47 .stream_count = 0,
48 .need_update = 1,
49 .type = LTTNG_CONSUMER_UNKNOWN,
50};
51
52enum consumer_channel_action {
53 CONSUMER_CHANNEL_ADD,
54 CONSUMER_CHANNEL_DEL,
55 CONSUMER_CHANNEL_QUIT,
56};
57
58struct consumer_channel_msg {
59 enum consumer_channel_action action;
60 struct lttng_consumer_channel *chan; /* add */
61 uint64_t key; /* del */
62};
63
64/*
65 * Flag to inform the polling thread to quit when all fd hung up. Updated by
66 * the consumer_thread_receive_fds when it notices that all fds has hung up.
67 * Also updated by the signal handler (consumer_should_exit()). Read by the
68 * polling threads.
69 */
70volatile int consumer_quit;
71
72/*
73 * Global hash table containing respectively metadata and data streams. The
74 * stream element in this ht should only be updated by the metadata poll thread
75 * for the metadata and the data poll thread for the data.
76 */
77static struct lttng_ht *metadata_ht;
78static struct lttng_ht *data_ht;
79
80/*
81 * Notify a thread lttng pipe to poll back again. This usually means that some
82 * global state has changed so we just send back the thread in a poll wait
83 * call.
84 */
85static void notify_thread_lttng_pipe(struct lttng_pipe *pipe)
86{
87 struct lttng_consumer_stream *null_stream = NULL;
88
89 assert(pipe);
90
91 (void) lttng_pipe_write(pipe, &null_stream, sizeof(null_stream));
92}
93
94static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
95 struct lttng_consumer_channel *chan,
96 uint64_t key,
97 enum consumer_channel_action action)
98{
99 struct consumer_channel_msg msg;
100 int ret;
101
102 memset(&msg, 0, sizeof(msg));
103
104 msg.action = action;
105 msg.chan = chan;
106 msg.key = key;
107 do {
108 ret = write(ctx->consumer_channel_pipe[1], &msg, sizeof(msg));
109 } while (ret < 0 && errno == EINTR);
110}
111
112void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
113 uint64_t key)
114{
115 notify_channel_pipe(ctx, NULL, key, CONSUMER_CHANNEL_DEL);
116}
117
118static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
119 struct lttng_consumer_channel **chan,
120 uint64_t *key,
121 enum consumer_channel_action *action)
122{
123 struct consumer_channel_msg msg;
124 int ret;
125
126 do {
127 ret = read(ctx->consumer_channel_pipe[0], &msg, sizeof(msg));
128 } while (ret < 0 && errno == EINTR);
129 if (ret > 0) {
130 *action = msg.action;
131 *chan = msg.chan;
132 *key = msg.key;
133 }
134 return ret;
135}
136
137/*
138 * Find a stream. The consumer_data.lock must be locked during this
139 * call.
140 */
141static struct lttng_consumer_stream *find_stream(uint64_t key,
142 struct lttng_ht *ht)
143{
144 struct lttng_ht_iter iter;
145 struct lttng_ht_node_u64 *node;
146 struct lttng_consumer_stream *stream = NULL;
147
148 assert(ht);
149
150 /* -1ULL keys are lookup failures */
151 if (key == (uint64_t) -1ULL) {
152 return NULL;
153 }
154
155 rcu_read_lock();
156
157 lttng_ht_lookup(ht, &key, &iter);
158 node = lttng_ht_iter_get_node_u64(&iter);
159 if (node != NULL) {
160 stream = caa_container_of(node, struct lttng_consumer_stream, node);
161 }
162
163 rcu_read_unlock();
164
165 return stream;
166}
167
168static void steal_stream_key(uint64_t key, struct lttng_ht *ht)
169{
170 struct lttng_consumer_stream *stream;
171
172 rcu_read_lock();
173 stream = find_stream(key, ht);
174 if (stream) {
175 stream->key = (uint64_t) -1ULL;
176 /*
177 * We don't want the lookup to match, but we still need
178 * to iterate on this stream when iterating over the hash table. Just
179 * change the node key.
180 */
181 stream->node.key = (uint64_t) -1ULL;
182 }
183 rcu_read_unlock();
184}
185
186/*
187 * Return a channel object for the given key.
188 *
189 * RCU read side lock MUST be acquired before calling this function and
190 * protects the channel ptr.
191 */
192struct lttng_consumer_channel *consumer_find_channel(uint64_t key)
193{
194 struct lttng_ht_iter iter;
195 struct lttng_ht_node_u64 *node;
196 struct lttng_consumer_channel *channel = NULL;
197
198 /* -1ULL keys are lookup failures */
199 if (key == (uint64_t) -1ULL) {
200 return NULL;
201 }
202
203 lttng_ht_lookup(consumer_data.channel_ht, &key, &iter);
204 node = lttng_ht_iter_get_node_u64(&iter);
205 if (node != NULL) {
206 channel = caa_container_of(node, struct lttng_consumer_channel, node);
207 }
208
209 return channel;
210}
211
212static void free_stream_rcu(struct rcu_head *head)
213{
214 struct lttng_ht_node_u64 *node =
215 caa_container_of(head, struct lttng_ht_node_u64, head);
216 struct lttng_consumer_stream *stream =
217 caa_container_of(node, struct lttng_consumer_stream, node);
218
219 free(stream);
220}
221
222static void free_channel_rcu(struct rcu_head *head)
223{
224 struct lttng_ht_node_u64 *node =
225 caa_container_of(head, struct lttng_ht_node_u64, head);
226 struct lttng_consumer_channel *channel =
227 caa_container_of(node, struct lttng_consumer_channel, node);
228
229 free(channel);
230}
231
232/*
233 * RCU protected relayd socket pair free.
234 */
235static void free_relayd_rcu(struct rcu_head *head)
236{
237 struct lttng_ht_node_u64 *node =
238 caa_container_of(head, struct lttng_ht_node_u64, head);
239 struct consumer_relayd_sock_pair *relayd =
240 caa_container_of(node, struct consumer_relayd_sock_pair, node);
241
242 /*
243 * Close all sockets. This is done in the call RCU since we don't want the
244 * socket fds to be reassigned thus potentially creating bad state of the
245 * relayd object.
246 *
247 * We do not have to lock the control socket mutex here since at this stage
248 * there is no one referencing to this relayd object.
249 */
250 (void) relayd_close(&relayd->control_sock);
251 (void) relayd_close(&relayd->data_sock);
252
253 free(relayd);
254}
255
256/*
257 * Destroy and free relayd socket pair object.
258 */
259void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd)
260{
261 int ret;
262 struct lttng_ht_iter iter;
263
264 if (relayd == NULL) {
265 return;
266 }
267
268 DBG("Consumer destroy and close relayd socket pair");
269
270 iter.iter.node = &relayd->node.node;
271 ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
272 if (ret != 0) {
273 /* We assume the relayd is being or is destroyed */
274 return;
275 }
276
277 /* RCU free() call */
278 call_rcu(&relayd->node.head, free_relayd_rcu);
279}
280
281/*
282 * Remove a channel from the global list protected by a mutex. This function is
283 * also responsible for freeing its data structures.
284 */
285void consumer_del_channel(struct lttng_consumer_channel *channel)
286{
287 int ret;
288 struct lttng_ht_iter iter;
289 struct lttng_consumer_stream *stream, *stmp;
290
291 DBG("Consumer delete channel key %" PRIu64, channel->key);
292
293 pthread_mutex_lock(&consumer_data.lock);
294 pthread_mutex_lock(&channel->lock);
295 pthread_mutex_lock(&channel->timer_lock);
296
297 /* Delete streams that might have been left in the stream list. */
298 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
299 send_node) {
300 cds_list_del(&stream->send_node);
301 /*
302 * Once a stream is added to this list, the buffers were created so
303 * we have a guarantee that this call will succeed.
304 */
305 consumer_stream_destroy(stream, NULL);
306 }
307
308 switch (consumer_data.type) {
309 case LTTNG_CONSUMER_KERNEL:
310 break;
311 case LTTNG_CONSUMER32_UST:
312 case LTTNG_CONSUMER64_UST:
313 lttng_ustconsumer_del_channel(channel);
314 break;
315 default:
316 ERR("Unknown consumer_data type");
317 assert(0);
318 goto end;
319 }
320
321 rcu_read_lock();
322 iter.iter.node = &channel->node.node;
323 ret = lttng_ht_del(consumer_data.channel_ht, &iter);
324 assert(!ret);
325 rcu_read_unlock();
326
327 call_rcu(&channel->node.head, free_channel_rcu);
328end:
329 pthread_mutex_unlock(&channel->timer_lock);
330 pthread_mutex_unlock(&channel->lock);
331 pthread_mutex_unlock(&consumer_data.lock);
332}
333
334/*
335 * Iterate over the relayd hash table and destroy each element. Finally,
336 * destroy the whole hash table.
337 */
338static void cleanup_relayd_ht(void)
339{
340 struct lttng_ht_iter iter;
341 struct consumer_relayd_sock_pair *relayd;
342
343 rcu_read_lock();
344
345 cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
346 node.node) {
347 consumer_destroy_relayd(relayd);
348 }
349
350 rcu_read_unlock();
351
352 lttng_ht_destroy(consumer_data.relayd_ht);
353}
354
355/*
356 * Update the end point status of all streams having the given network sequence
357 * index (relayd index).
358 *
359 * It's atomically set without having the stream mutex locked which is fine
360 * because we handle the write/read race with a pipe wakeup for each thread.
361 */
362static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
363 enum consumer_endpoint_status status)
364{
365 struct lttng_ht_iter iter;
366 struct lttng_consumer_stream *stream;
367
368 DBG("Consumer set delete flag on stream by idx %" PRIu64, net_seq_idx);
369
370 rcu_read_lock();
371
372 /* Let's begin with metadata */
373 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
374 if (stream->net_seq_idx == net_seq_idx) {
375 uatomic_set(&stream->endpoint_status, status);
376 DBG("Delete flag set to metadata stream %d", stream->wait_fd);
377 }
378 }
379
380 /* Follow up by the data streams */
381 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
382 if (stream->net_seq_idx == net_seq_idx) {
383 uatomic_set(&stream->endpoint_status, status);
384 DBG("Delete flag set to data stream %d", stream->wait_fd);
385 }
386 }
387 rcu_read_unlock();
388}
389
390/*
391 * Cleanup a relayd object by flagging every associated streams for deletion,
392 * destroying the object meaning removing it from the relayd hash table,
393 * closing the sockets and freeing the memory in a RCU call.
394 *
395 * If a local data context is available, notify the threads that the streams'
396 * state have changed.
397 */
398static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
399 struct lttng_consumer_local_data *ctx)
400{
401 uint64_t netidx;
402
403 assert(relayd);
404
405 DBG("Cleaning up relayd sockets");
406
407 /* Save the net sequence index before destroying the object */
408 netidx = relayd->net_seq_idx;
409
410 /*
411 * Delete the relayd from the relayd hash table, close the sockets and free
412 * the object in a RCU call.
413 */
414 consumer_destroy_relayd(relayd);
415
416 /* Set inactive endpoint to all streams */
417 update_endpoint_status_by_netidx(netidx, CONSUMER_ENDPOINT_INACTIVE);
418
419 /*
420 * With a local data context, notify the threads that the streams' state
421 * have changed. The write() action on the pipe acts as an "implicit"
422 * memory barrier ordering the updates of the end point status from the
423 * read of this status which happens AFTER receiving this notify.
424 */
425 if (ctx) {
426 notify_thread_lttng_pipe(ctx->consumer_data_pipe);
427 notify_thread_lttng_pipe(ctx->consumer_metadata_pipe);
428 }
429}
430
431/*
432 * Flag a relayd socket pair for destruction. Destroy it if the refcount
433 * reaches zero.
434 *
435 * RCU read side lock MUST be aquired before calling this function.
436 */
437void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
438{
439 assert(relayd);
440
441 /* Set destroy flag for this object */
442 uatomic_set(&relayd->destroy_flag, 1);
443
444 /* Destroy the relayd if refcount is 0 */
445 if (uatomic_read(&relayd->refcount) == 0) {
446 consumer_destroy_relayd(relayd);
447 }
448}
449
450/*
451 * Completly destroy stream from every visiable data structure and the given
452 * hash table if one.
453 *
454 * One this call returns, the stream object is not longer usable nor visible.
455 */
456void consumer_del_stream(struct lttng_consumer_stream *stream,
457 struct lttng_ht *ht)
458{
459 consumer_stream_destroy(stream, ht);
460}
461
462struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
463 uint64_t stream_key,
464 enum lttng_consumer_stream_state state,
465 const char *channel_name,
466 uid_t uid,
467 gid_t gid,
468 uint64_t relayd_id,
469 uint64_t session_id,
470 int cpu,
471 int *alloc_ret,
472 enum consumer_channel_type type,
473 unsigned int monitor)
474{
475 int ret;
476 struct lttng_consumer_stream *stream;
477
478 stream = zmalloc(sizeof(*stream));
479 if (stream == NULL) {
480 PERROR("malloc struct lttng_consumer_stream");
481 ret = -ENOMEM;
482 goto end;
483 }
484
485 rcu_read_lock();
486
487 stream->key = stream_key;
488 stream->out_fd = -1;
489 stream->out_fd_offset = 0;
490 stream->state = state;
491 stream->uid = uid;
492 stream->gid = gid;
493 stream->net_seq_idx = relayd_id;
494 stream->session_id = session_id;
495 stream->monitor = monitor;
496 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
497 pthread_mutex_init(&stream->lock, NULL);
498
499 /* If channel is the metadata, flag this stream as metadata. */
500 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
501 stream->metadata_flag = 1;
502 /* Metadata is flat out. */
503 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
504 } else {
505 /* Format stream name to <channel_name>_<cpu_number> */
506 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
507 channel_name, cpu);
508 if (ret < 0) {
509 PERROR("snprintf stream name");
510 goto error;
511 }
512 }
513
514 /* Key is always the wait_fd for streams. */
515 lttng_ht_node_init_u64(&stream->node, stream->key);
516
517 /* Init node per channel id key */
518 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
519
520 /* Init session id node with the stream session id */
521 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
522
523 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
524 " relayd_id %" PRIu64 ", session_id %" PRIu64,
525 stream->name, stream->key, channel_key,
526 stream->net_seq_idx, stream->session_id);
527
528 rcu_read_unlock();
529 return stream;
530
531error:
532 rcu_read_unlock();
533 free(stream);
534end:
535 if (alloc_ret) {
536 *alloc_ret = ret;
537 }
538 return NULL;
539}
540
541/*
542 * Add a stream to the global list protected by a mutex.
543 */
544static int add_stream(struct lttng_consumer_stream *stream,
545 struct lttng_ht *ht)
546{
547 int ret = 0;
548
549 assert(stream);
550 assert(ht);
551
552 DBG3("Adding consumer stream %" PRIu64, stream->key);
553
554 pthread_mutex_lock(&consumer_data.lock);
555 pthread_mutex_lock(&stream->chan->lock);
556 pthread_mutex_lock(&stream->chan->timer_lock);
557 pthread_mutex_lock(&stream->lock);
558 rcu_read_lock();
559
560 /* Steal stream identifier to avoid having streams with the same key */
561 steal_stream_key(stream->key, ht);
562
563 lttng_ht_add_unique_u64(ht, &stream->node);
564
565 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
566 &stream->node_channel_id);
567
568 /*
569 * Add stream to the stream_list_ht of the consumer data. No need to steal
570 * the key since the HT does not use it and we allow to add redundant keys
571 * into this table.
572 */
573 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
574
575 /*
576 * When nb_init_stream_left reaches 0, we don't need to trigger any action
577 * in terms of destroying the associated channel, because the action that
578 * causes the count to become 0 also causes a stream to be added. The
579 * channel deletion will thus be triggered by the following removal of this
580 * stream.
581 */
582 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
583 /* Increment refcount before decrementing nb_init_stream_left */
584 cmm_smp_wmb();
585 uatomic_dec(&stream->chan->nb_init_stream_left);
586 }
587
588 /* Update consumer data once the node is inserted. */
589 consumer_data.stream_count++;
590 consumer_data.need_update = 1;
591
592 rcu_read_unlock();
593 pthread_mutex_unlock(&stream->lock);
594 pthread_mutex_unlock(&stream->chan->timer_lock);
595 pthread_mutex_unlock(&stream->chan->lock);
596 pthread_mutex_unlock(&consumer_data.lock);
597
598 return ret;
599}
600
601/*
602 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
603 * be acquired before calling this.
604 */
605static int add_relayd(struct consumer_relayd_sock_pair *relayd)
606{
607 int ret = 0;
608 struct lttng_ht_node_u64 *node;
609 struct lttng_ht_iter iter;
610
611 assert(relayd);
612
613 lttng_ht_lookup(consumer_data.relayd_ht,
614 &relayd->net_seq_idx, &iter);
615 node = lttng_ht_iter_get_node_u64(&iter);
616 if (node != NULL) {
617 goto end;
618 }
619 lttng_ht_add_unique_u64(consumer_data.relayd_ht, &relayd->node);
620
621end:
622 return ret;
623}
624
625/*
626 * Allocate and return a consumer relayd socket.
627 */
628struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
629 uint64_t net_seq_idx)
630{
631 struct consumer_relayd_sock_pair *obj = NULL;
632
633 /* net sequence index of -1 is a failure */
634 if (net_seq_idx == (uint64_t) -1ULL) {
635 goto error;
636 }
637
638 obj = zmalloc(sizeof(struct consumer_relayd_sock_pair));
639 if (obj == NULL) {
640 PERROR("zmalloc relayd sock");
641 goto error;
642 }
643
644 obj->net_seq_idx = net_seq_idx;
645 obj->refcount = 0;
646 obj->destroy_flag = 0;
647 obj->control_sock.sock.fd = -1;
648 obj->data_sock.sock.fd = -1;
649 lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
650 pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
651
652error:
653 return obj;
654}
655
656/*
657 * Find a relayd socket pair in the global consumer data.
658 *
659 * Return the object if found else NULL.
660 * RCU read-side lock must be held across this call and while using the
661 * returned object.
662 */
663struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
664{
665 struct lttng_ht_iter iter;
666 struct lttng_ht_node_u64 *node;
667 struct consumer_relayd_sock_pair *relayd = NULL;
668
669 /* Negative keys are lookup failures */
670 if (key == (uint64_t) -1ULL) {
671 goto error;
672 }
673
674 lttng_ht_lookup(consumer_data.relayd_ht, &key,
675 &iter);
676 node = lttng_ht_iter_get_node_u64(&iter);
677 if (node != NULL) {
678 relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
679 }
680
681error:
682 return relayd;
683}
684
685/*
686 * Find a relayd and send the stream
687 *
688 * Returns 0 on success, < 0 on error
689 */
690int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
691 char *path)
692{
693 int ret = 0;
694 struct consumer_relayd_sock_pair *relayd;
695
696 assert(stream);
697 assert(stream->net_seq_idx != -1ULL);
698 assert(path);
699
700 /* The stream is not metadata. Get relayd reference if exists. */
701 rcu_read_lock();
702 relayd = consumer_find_relayd(stream->net_seq_idx);
703 if (relayd != NULL) {
704 /* Add stream on the relayd */
705 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
706 ret = relayd_add_stream(&relayd->control_sock, stream->name,
707 path, &stream->relayd_stream_id,
708 stream->chan->tracefile_size, stream->chan->tracefile_count);
709 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
710 if (ret < 0) {
711 goto end;
712 }
713 uatomic_inc(&relayd->refcount);
714 stream->sent_to_relayd = 1;
715 } else {
716 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
717 stream->key, stream->net_seq_idx);
718 ret = -1;
719 goto end;
720 }
721
722 DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
723 stream->name, stream->key, stream->net_seq_idx);
724
725end:
726 rcu_read_unlock();
727 return ret;
728}
729
730/*
731 * Find a relayd and close the stream
732 */
733void close_relayd_stream(struct lttng_consumer_stream *stream)
734{
735 struct consumer_relayd_sock_pair *relayd;
736
737 /* The stream is not metadata. Get relayd reference if exists. */
738 rcu_read_lock();
739 relayd = consumer_find_relayd(stream->net_seq_idx);
740 if (relayd) {
741 consumer_stream_relayd_close(stream, relayd);
742 }
743 rcu_read_unlock();
744}
745
746/*
747 * Handle stream for relayd transmission if the stream applies for network
748 * streaming where the net sequence index is set.
749 *
750 * Return destination file descriptor or negative value on error.
751 */
752static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
753 size_t data_size, unsigned long padding,
754 struct consumer_relayd_sock_pair *relayd)
755{
756 int outfd = -1, ret;
757 struct lttcomm_relayd_data_hdr data_hdr;
758
759 /* Safety net */
760 assert(stream);
761 assert(relayd);
762
763 /* Reset data header */
764 memset(&data_hdr, 0, sizeof(data_hdr));
765
766 if (stream->metadata_flag) {
767 /* Caller MUST acquire the relayd control socket lock */
768 ret = relayd_send_metadata(&relayd->control_sock, data_size);
769 if (ret < 0) {
770 goto error;
771 }
772
773 /* Metadata are always sent on the control socket. */
774 outfd = relayd->control_sock.sock.fd;
775 } else {
776 /* Set header with stream information */
777 data_hdr.stream_id = htobe64(stream->relayd_stream_id);
778 data_hdr.data_size = htobe32(data_size);
779 data_hdr.padding_size = htobe32(padding);
780 /*
781 * Note that net_seq_num below is assigned with the *current* value of
782 * next_net_seq_num and only after that the next_net_seq_num will be
783 * increment. This is why when issuing a command on the relayd using
784 * this next value, 1 should always be substracted in order to compare
785 * the last seen sequence number on the relayd side to the last sent.
786 */
787 data_hdr.net_seq_num = htobe64(stream->next_net_seq_num);
788 /* Other fields are zeroed previously */
789
790 ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
791 sizeof(data_hdr));
792 if (ret < 0) {
793 goto error;
794 }
795
796 ++stream->next_net_seq_num;
797
798 /* Set to go on data socket */
799 outfd = relayd->data_sock.sock.fd;
800 }
801
802error:
803 return outfd;
804}
805
806/*
807 * Allocate and return a new lttng_consumer_channel object using the given key
808 * to initialize the hash table node.
809 *
810 * On error, return NULL.
811 */
812struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
813 uint64_t session_id,
814 const char *pathname,
815 const char *name,
816 uid_t uid,
817 gid_t gid,
818 uint64_t relayd_id,
819 enum lttng_event_output output,
820 uint64_t tracefile_size,
821 uint64_t tracefile_count,
822 uint64_t session_id_per_pid,
823 unsigned int monitor)
824{
825 struct lttng_consumer_channel *channel;
826
827 channel = zmalloc(sizeof(*channel));
828 if (channel == NULL) {
829 PERROR("malloc struct lttng_consumer_channel");
830 goto end;
831 }
832
833 channel->key = key;
834 channel->refcount = 0;
835 channel->session_id = session_id;
836 channel->session_id_per_pid = session_id_per_pid;
837 channel->uid = uid;
838 channel->gid = gid;
839 channel->relayd_id = relayd_id;
840 channel->output = output;
841 channel->tracefile_size = tracefile_size;
842 channel->tracefile_count = tracefile_count;
843 channel->monitor = monitor;
844 pthread_mutex_init(&channel->lock, NULL);
845 pthread_mutex_init(&channel->timer_lock, NULL);
846
847 /*
848 * In monitor mode, the streams associated with the channel will be put in
849 * a special list ONLY owned by this channel. So, the refcount is set to 1
850 * here meaning that the channel itself has streams that are referenced.
851 *
852 * On a channel deletion, once the channel is no longer visible, the
853 * refcount is decremented and checked for a zero value to delete it. With
854 * streams in no monitor mode, it will now be safe to destroy the channel.
855 */
856 if (!channel->monitor) {
857 channel->refcount = 1;
858 }
859
860 strncpy(channel->pathname, pathname, sizeof(channel->pathname));
861 channel->pathname[sizeof(channel->pathname) - 1] = '\0';
862
863 strncpy(channel->name, name, sizeof(channel->name));
864 channel->name[sizeof(channel->name) - 1] = '\0';
865
866 lttng_ht_node_init_u64(&channel->node, channel->key);
867
868 channel->wait_fd = -1;
869
870 CDS_INIT_LIST_HEAD(&channel->streams.head);
871
872 DBG("Allocated channel (key %" PRIu64 ")", channel->key)
873
874end:
875 return channel;
876}
877
878/*
879 * Add a channel to the global list protected by a mutex.
880 *
881 * On success 0 is returned else a negative value.
882 */
883int consumer_add_channel(struct lttng_consumer_channel *channel,
884 struct lttng_consumer_local_data *ctx)
885{
886 int ret = 0;
887 struct lttng_ht_node_u64 *node;
888 struct lttng_ht_iter iter;
889
890 pthread_mutex_lock(&consumer_data.lock);
891 pthread_mutex_lock(&channel->lock);
892 pthread_mutex_lock(&channel->timer_lock);
893 rcu_read_lock();
894
895 lttng_ht_lookup(consumer_data.channel_ht, &channel->key, &iter);
896 node = lttng_ht_iter_get_node_u64(&iter);
897 if (node != NULL) {
898 /* Channel already exist. Ignore the insertion */
899 ERR("Consumer add channel key %" PRIu64 " already exists!",
900 channel->key);
901 ret = -EEXIST;
902 goto end;
903 }
904
905 lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
906
907end:
908 rcu_read_unlock();
909 pthread_mutex_unlock(&channel->timer_lock);
910 pthread_mutex_unlock(&channel->lock);
911 pthread_mutex_unlock(&consumer_data.lock);
912
913 if (!ret && channel->wait_fd != -1 &&
914 channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
915 notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
916 }
917 return ret;
918}
919
920/*
921 * Allocate the pollfd structure and the local view of the out fds to avoid
922 * doing a lookup in the linked list and concurrency issues when writing is
923 * needed. Called with consumer_data.lock held.
924 *
925 * Returns the number of fds in the structures.
926 */
927static int update_poll_array(struct lttng_consumer_local_data *ctx,
928 struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
929 struct lttng_ht *ht)
930{
931 int i = 0;
932 struct lttng_ht_iter iter;
933 struct lttng_consumer_stream *stream;
934
935 assert(ctx);
936 assert(ht);
937 assert(pollfd);
938 assert(local_stream);
939
940 DBG("Updating poll fd array");
941 rcu_read_lock();
942 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
943 /*
944 * Only active streams with an active end point can be added to the
945 * poll set and local stream storage of the thread.
946 *
947 * There is a potential race here for endpoint_status to be updated
948 * just after the check. However, this is OK since the stream(s) will
949 * be deleted once the thread is notified that the end point state has
950 * changed where this function will be called back again.
951 */
952 if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM ||
953 stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
954 continue;
955 }
956 /*
957 * This clobbers way too much the debug output. Uncomment that if you
958 * need it for debugging purposes.
959 *
960 * DBG("Active FD %d", stream->wait_fd);
961 */
962 (*pollfd)[i].fd = stream->wait_fd;
963 (*pollfd)[i].events = POLLIN | POLLPRI;
964 local_stream[i] = stream;
965 i++;
966 }
967 rcu_read_unlock();
968
969 /*
970 * Insert the consumer_data_pipe at the end of the array and don't
971 * increment i so nb_fd is the number of real FD.
972 */
973 (*pollfd)[i].fd = lttng_pipe_get_readfd(ctx->consumer_data_pipe);
974 (*pollfd)[i].events = POLLIN | POLLPRI;
975 return i;
976}
977
978/*
979 * Poll on the should_quit pipe and the command socket return -1 on error and
980 * should exit, 0 if data is available on the command socket
981 */
982int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
983{
984 int num_rdy;
985
986restart:
987 num_rdy = poll(consumer_sockpoll, 2, -1);
988 if (num_rdy == -1) {
989 /*
990 * Restart interrupted system call.
991 */
992 if (errno == EINTR) {
993 goto restart;
994 }
995 PERROR("Poll error");
996 goto exit;
997 }
998 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
999 DBG("consumer_should_quit wake up");
1000 goto exit;
1001 }
1002 return 0;
1003
1004exit:
1005 return -1;
1006}
1007
1008/*
1009 * Set the error socket.
1010 */
1011void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
1012 int sock)
1013{
1014 ctx->consumer_error_socket = sock;
1015}
1016
1017/*
1018 * Set the command socket path.
1019 */
1020void lttng_consumer_set_command_sock_path(
1021 struct lttng_consumer_local_data *ctx, char *sock)
1022{
1023 ctx->consumer_command_sock_path = sock;
1024}
1025
1026/*
1027 * Send return code to the session daemon.
1028 * If the socket is not defined, we return 0, it is not a fatal error
1029 */
1030int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
1031{
1032 if (ctx->consumer_error_socket > 0) {
1033 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
1034 sizeof(enum lttcomm_sessiond_command));
1035 }
1036
1037 return 0;
1038}
1039
1040/*
1041 * Close all the tracefiles and stream fds and MUST be called when all
1042 * instances are destroyed i.e. when all threads were joined and are ended.
1043 */
1044void lttng_consumer_cleanup(void)
1045{
1046 struct lttng_ht_iter iter;
1047 struct lttng_consumer_channel *channel;
1048
1049 rcu_read_lock();
1050
1051 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
1052 node.node) {
1053 consumer_del_channel(channel);
1054 }
1055
1056 rcu_read_unlock();
1057
1058 lttng_ht_destroy(consumer_data.channel_ht);
1059
1060 cleanup_relayd_ht();
1061
1062 lttng_ht_destroy(consumer_data.stream_per_chan_id_ht);
1063
1064 /*
1065 * This HT contains streams that are freed by either the metadata thread or
1066 * the data thread so we do *nothing* on the hash table and simply destroy
1067 * it.
1068 */
1069 lttng_ht_destroy(consumer_data.stream_list_ht);
1070}
1071
1072/*
1073 * Called from signal handler.
1074 */
1075void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
1076{
1077 int ret;
1078 consumer_quit = 1;
1079 do {
1080 ret = write(ctx->consumer_should_quit[1], "4", 1);
1081 } while (ret < 0 && errno == EINTR);
1082 if (ret < 0 || ret != 1) {
1083 PERROR("write consumer quit");
1084 }
1085
1086 DBG("Consumer flag that it should quit");
1087}
1088
1089void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
1090 off_t orig_offset)
1091{
1092 int outfd = stream->out_fd;
1093
1094 /*
1095 * This does a blocking write-and-wait on any page that belongs to the
1096 * subbuffer prior to the one we just wrote.
1097 * Don't care about error values, as these are just hints and ways to
1098 * limit the amount of page cache used.
1099 */
1100 if (orig_offset < stream->max_sb_size) {
1101 return;
1102 }
1103 lttng_sync_file_range(outfd, orig_offset - stream->max_sb_size,
1104 stream->max_sb_size,
1105 SYNC_FILE_RANGE_WAIT_BEFORE
1106 | SYNC_FILE_RANGE_WRITE
1107 | SYNC_FILE_RANGE_WAIT_AFTER);
1108 /*
1109 * Give hints to the kernel about how we access the file:
1110 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
1111 * we write it.
1112 *
1113 * We need to call fadvise again after the file grows because the
1114 * kernel does not seem to apply fadvise to non-existing parts of the
1115 * file.
1116 *
1117 * Call fadvise _after_ having waited for the page writeback to
1118 * complete because the dirty page writeback semantic is not well
1119 * defined. So it can be expected to lead to lower throughput in
1120 * streaming.
1121 */
1122 posix_fadvise(outfd, orig_offset - stream->max_sb_size,
1123 stream->max_sb_size, POSIX_FADV_DONTNEED);
1124}
1125
1126/*
1127 * Initialise the necessary environnement :
1128 * - create a new context
1129 * - create the poll_pipe
1130 * - create the should_quit pipe (for signal handler)
1131 * - create the thread pipe (for splice)
1132 *
1133 * Takes a function pointer as argument, this function is called when data is
1134 * available on a buffer. This function is responsible to do the
1135 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
1136 * buffer configuration and then kernctl_put_next_subbuf at the end.
1137 *
1138 * Returns a pointer to the new context or NULL on error.
1139 */
1140struct lttng_consumer_local_data *lttng_consumer_create(
1141 enum lttng_consumer_type type,
1142 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
1143 struct lttng_consumer_local_data *ctx),
1144 int (*recv_channel)(struct lttng_consumer_channel *channel),
1145 int (*recv_stream)(struct lttng_consumer_stream *stream),
1146 int (*update_stream)(uint64_t stream_key, uint32_t state))
1147{
1148 int ret;
1149 struct lttng_consumer_local_data *ctx;
1150
1151 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
1152 consumer_data.type == type);
1153 consumer_data.type = type;
1154
1155 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
1156 if (ctx == NULL) {
1157 PERROR("allocating context");
1158 goto error;
1159 }
1160
1161 ctx->consumer_error_socket = -1;
1162 ctx->consumer_metadata_socket = -1;
1163 pthread_mutex_init(&ctx->metadata_socket_lock, NULL);
1164 /* assign the callbacks */
1165 ctx->on_buffer_ready = buffer_ready;
1166 ctx->on_recv_channel = recv_channel;
1167 ctx->on_recv_stream = recv_stream;
1168 ctx->on_update_stream = update_stream;
1169
1170 ctx->consumer_data_pipe = lttng_pipe_open(0);
1171 if (!ctx->consumer_data_pipe) {
1172 goto error_poll_pipe;
1173 }
1174
1175 ret = pipe(ctx->consumer_should_quit);
1176 if (ret < 0) {
1177 PERROR("Error creating recv pipe");
1178 goto error_quit_pipe;
1179 }
1180
1181 ret = pipe(ctx->consumer_thread_pipe);
1182 if (ret < 0) {
1183 PERROR("Error creating thread pipe");
1184 goto error_thread_pipe;
1185 }
1186
1187 ret = pipe(ctx->consumer_channel_pipe);
1188 if (ret < 0) {
1189 PERROR("Error creating channel pipe");
1190 goto error_channel_pipe;
1191 }
1192
1193 ctx->consumer_metadata_pipe = lttng_pipe_open(0);
1194 if (!ctx->consumer_metadata_pipe) {
1195 goto error_metadata_pipe;
1196 }
1197
1198 ret = utils_create_pipe(ctx->consumer_splice_metadata_pipe);
1199 if (ret < 0) {
1200 goto error_splice_pipe;
1201 }
1202
1203 return ctx;
1204
1205error_splice_pipe:
1206 lttng_pipe_destroy(ctx->consumer_metadata_pipe);
1207error_metadata_pipe:
1208 utils_close_pipe(ctx->consumer_channel_pipe);
1209error_channel_pipe:
1210 utils_close_pipe(ctx->consumer_thread_pipe);
1211error_thread_pipe:
1212 utils_close_pipe(ctx->consumer_should_quit);
1213error_quit_pipe:
1214 lttng_pipe_destroy(ctx->consumer_data_pipe);
1215error_poll_pipe:
1216 free(ctx);
1217error:
1218 return NULL;
1219}
1220
1221/*
1222 * Close all fds associated with the instance and free the context.
1223 */
1224void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
1225{
1226 int ret;
1227
1228 DBG("Consumer destroying it. Closing everything.");
1229
1230 ret = close(ctx->consumer_error_socket);
1231 if (ret) {
1232 PERROR("close");
1233 }
1234 ret = close(ctx->consumer_metadata_socket);
1235 if (ret) {
1236 PERROR("close");
1237 }
1238 utils_close_pipe(ctx->consumer_thread_pipe);
1239 utils_close_pipe(ctx->consumer_channel_pipe);
1240 lttng_pipe_destroy(ctx->consumer_data_pipe);
1241 lttng_pipe_destroy(ctx->consumer_metadata_pipe);
1242 utils_close_pipe(ctx->consumer_should_quit);
1243 utils_close_pipe(ctx->consumer_splice_metadata_pipe);
1244
1245 unlink(ctx->consumer_command_sock_path);
1246 free(ctx);
1247}
1248
1249/*
1250 * Write the metadata stream id on the specified file descriptor.
1251 */
1252static int write_relayd_metadata_id(int fd,
1253 struct lttng_consumer_stream *stream,
1254 struct consumer_relayd_sock_pair *relayd, unsigned long padding)
1255{
1256 int ret;
1257 struct lttcomm_relayd_metadata_payload hdr;
1258
1259 hdr.stream_id = htobe64(stream->relayd_stream_id);
1260 hdr.padding_size = htobe32(padding);
1261 do {
1262 ret = write(fd, (void *) &hdr, sizeof(hdr));
1263 } while (ret < 0 && errno == EINTR);
1264 if (ret < 0 || ret != sizeof(hdr)) {
1265 /*
1266 * This error means that the fd's end is closed so ignore the perror
1267 * not to clubber the error output since this can happen in a normal
1268 * code path.
1269 */
1270 if (errno != EPIPE) {
1271 PERROR("write metadata stream id");
1272 }
1273 DBG3("Consumer failed to write relayd metadata id (errno: %d)", errno);
1274 /*
1275 * Set ret to a negative value because if ret != sizeof(hdr), we don't
1276 * handle writting the missing part so report that as an error and
1277 * don't lie to the caller.
1278 */
1279 ret = -1;
1280 goto end;
1281 }
1282 DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
1283 stream->relayd_stream_id, padding);
1284
1285end:
1286 return ret;
1287}
1288
1289/*
1290 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1291 * core function for writing trace buffers to either the local filesystem or
1292 * the network.
1293 *
1294 * It must be called with the stream lock held.
1295 *
1296 * Careful review MUST be put if any changes occur!
1297 *
1298 * Returns the number of bytes written
1299 */
1300ssize_t lttng_consumer_on_read_subbuffer_mmap(
1301 struct lttng_consumer_local_data *ctx,
1302 struct lttng_consumer_stream *stream, unsigned long len,
1303 unsigned long padding)
1304{
1305 unsigned long mmap_offset;
1306 void *mmap_base;
1307 ssize_t ret = 0, written = 0;
1308 off_t orig_offset = stream->out_fd_offset;
1309 /* Default is on the disk */
1310 int outfd = stream->out_fd;
1311 struct consumer_relayd_sock_pair *relayd = NULL;
1312 unsigned int relayd_hang_up = 0;
1313
1314 /* RCU lock for the relayd pointer */
1315 rcu_read_lock();
1316
1317 /* Flag that the current stream if set for network streaming. */
1318 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1319 relayd = consumer_find_relayd(stream->net_seq_idx);
1320 if (relayd == NULL) {
1321 goto end;
1322 }
1323 }
1324
1325 /* get the offset inside the fd to mmap */
1326 switch (consumer_data.type) {
1327 case LTTNG_CONSUMER_KERNEL:
1328 mmap_base = stream->mmap_base;
1329 ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
1330 break;
1331 case LTTNG_CONSUMER32_UST:
1332 case LTTNG_CONSUMER64_UST:
1333 mmap_base = lttng_ustctl_get_mmap_base(stream);
1334 if (!mmap_base) {
1335 ERR("read mmap get mmap base for stream %s", stream->name);
1336 written = -1;
1337 goto end;
1338 }
1339 ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
1340
1341 break;
1342 default:
1343 ERR("Unknown consumer_data type");
1344 assert(0);
1345 }
1346 if (ret != 0) {
1347 errno = -ret;
1348 PERROR("tracer ctl get_mmap_read_offset");
1349 written = ret;
1350 goto end;
1351 }
1352
1353 /* Handle stream on the relayd if the output is on the network */
1354 if (relayd) {
1355 unsigned long netlen = len;
1356
1357 /*
1358 * Lock the control socket for the complete duration of the function
1359 * since from this point on we will use the socket.
1360 */
1361 if (stream->metadata_flag) {
1362 /* Metadata requires the control socket. */
1363 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1364 netlen += sizeof(struct lttcomm_relayd_metadata_payload);
1365 }
1366
1367 ret = write_relayd_stream_header(stream, netlen, padding, relayd);
1368 if (ret >= 0) {
1369 /* Use the returned socket. */
1370 outfd = ret;
1371
1372 /* Write metadata stream id before payload */
1373 if (stream->metadata_flag) {
1374 ret = write_relayd_metadata_id(outfd, stream, relayd, padding);
1375 if (ret < 0) {
1376 written = ret;
1377 /* Socket operation failed. We consider the relayd dead */
1378 if (ret == -EPIPE || ret == -EINVAL) {
1379 relayd_hang_up = 1;
1380 goto write_error;
1381 }
1382 goto end;
1383 }
1384 }
1385 } else {
1386 /* Socket operation failed. We consider the relayd dead */
1387 if (ret == -EPIPE || ret == -EINVAL) {
1388 relayd_hang_up = 1;
1389 goto write_error;
1390 }
1391 /* Else, use the default set before which is the filesystem. */
1392 }
1393 } else {
1394 /* No streaming, we have to set the len with the full padding */
1395 len += padding;
1396
1397 /*
1398 * Check if we need to change the tracefile before writing the packet.
1399 */
1400 if (stream->chan->tracefile_size > 0 &&
1401 (stream->tracefile_size_current + len) >
1402 stream->chan->tracefile_size) {
1403 ret = utils_rotate_stream_file(stream->chan->pathname,
1404 stream->name, stream->chan->tracefile_size,
1405 stream->chan->tracefile_count, stream->uid, stream->gid,
1406 stream->out_fd, &(stream->tracefile_count_current));
1407 if (ret < 0) {
1408 ERR("Rotating output file");
1409 goto end;
1410 }
1411 outfd = stream->out_fd = ret;
1412 /* Reset current size because we just perform a rotation. */
1413 stream->tracefile_size_current = 0;
1414 }
1415 stream->tracefile_size_current += len;
1416 }
1417
1418 while (len > 0) {
1419 do {
1420 ret = write(outfd, mmap_base + mmap_offset, len);
1421 } while (ret < 0 && errno == EINTR);
1422 DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
1423 if (ret < 0) {
1424 /*
1425 * This is possible if the fd is closed on the other side (outfd)
1426 * or any write problem. It can be verbose a bit for a normal
1427 * execution if for instance the relayd is stopped abruptly. This
1428 * can happen so set this to a DBG statement.
1429 */
1430 DBG("Error in file write mmap");
1431 if (written == 0) {
1432 written = ret;
1433 }
1434 /* Socket operation failed. We consider the relayd dead */
1435 if (errno == EPIPE || errno == EINVAL) {
1436 relayd_hang_up = 1;
1437 goto write_error;
1438 }
1439 goto end;
1440 } else if (ret > len) {
1441 PERROR("Error in file write (ret %zd > len %lu)", ret, len);
1442 written += ret;
1443 goto end;
1444 } else {
1445 len -= ret;
1446 mmap_offset += ret;
1447 }
1448
1449 /* This call is useless on a socket so better save a syscall. */
1450 if (!relayd) {
1451 /* This won't block, but will start writeout asynchronously */
1452 lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
1453 SYNC_FILE_RANGE_WRITE);
1454 stream->out_fd_offset += ret;
1455 }
1456 written += ret;
1457 }
1458 lttng_consumer_sync_trace_file(stream, orig_offset);
1459
1460write_error:
1461 /*
1462 * This is a special case that the relayd has closed its socket. Let's
1463 * cleanup the relayd object and all associated streams.
1464 */
1465 if (relayd && relayd_hang_up) {
1466 cleanup_relayd(relayd, ctx);
1467 }
1468
1469end:
1470 /* Unlock only if ctrl socket used */
1471 if (relayd && stream->metadata_flag) {
1472 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1473 }
1474
1475 rcu_read_unlock();
1476 return written;
1477}
1478
1479/*
1480 * Splice the data from the ring buffer to the tracefile.
1481 *
1482 * It must be called with the stream lock held.
1483 *
1484 * Returns the number of bytes spliced.
1485 */
1486ssize_t lttng_consumer_on_read_subbuffer_splice(
1487 struct lttng_consumer_local_data *ctx,
1488 struct lttng_consumer_stream *stream, unsigned long len,
1489 unsigned long padding)
1490{
1491 ssize_t ret = 0, written = 0, ret_splice = 0;
1492 loff_t offset = 0;
1493 off_t orig_offset = stream->out_fd_offset;
1494 int fd = stream->wait_fd;
1495 /* Default is on the disk */
1496 int outfd = stream->out_fd;
1497 struct consumer_relayd_sock_pair *relayd = NULL;
1498 int *splice_pipe;
1499 unsigned int relayd_hang_up = 0;
1500
1501 switch (consumer_data.type) {
1502 case LTTNG_CONSUMER_KERNEL:
1503 break;
1504 case LTTNG_CONSUMER32_UST:
1505 case LTTNG_CONSUMER64_UST:
1506 /* Not supported for user space tracing */
1507 return -ENOSYS;
1508 default:
1509 ERR("Unknown consumer_data type");
1510 assert(0);
1511 }
1512
1513 /* RCU lock for the relayd pointer */
1514 rcu_read_lock();
1515
1516 /* Flag that the current stream if set for network streaming. */
1517 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1518 relayd = consumer_find_relayd(stream->net_seq_idx);
1519 if (relayd == NULL) {
1520 goto end;
1521 }
1522 }
1523
1524 /*
1525 * Choose right pipe for splice. Metadata and trace data are handled by
1526 * different threads hence the use of two pipes in order not to race or
1527 * corrupt the written data.
1528 */
1529 if (stream->metadata_flag) {
1530 splice_pipe = ctx->consumer_splice_metadata_pipe;
1531 } else {
1532 splice_pipe = ctx->consumer_thread_pipe;
1533 }
1534
1535 /* Write metadata stream id before payload */
1536 if (relayd) {
1537 int total_len = len;
1538
1539 if (stream->metadata_flag) {
1540 /*
1541 * Lock the control socket for the complete duration of the function
1542 * since from this point on we will use the socket.
1543 */
1544 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1545
1546 ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
1547 padding);
1548 if (ret < 0) {
1549 written = ret;
1550 /* Socket operation failed. We consider the relayd dead */
1551 if (ret == -EBADF) {
1552 WARN("Remote relayd disconnected. Stopping");
1553 relayd_hang_up = 1;
1554 goto write_error;
1555 }
1556 goto end;
1557 }
1558
1559 total_len += sizeof(struct lttcomm_relayd_metadata_payload);
1560 }
1561
1562 ret = write_relayd_stream_header(stream, total_len, padding, relayd);
1563 if (ret >= 0) {
1564 /* Use the returned socket. */
1565 outfd = ret;
1566 } else {
1567 /* Socket operation failed. We consider the relayd dead */
1568 if (ret == -EBADF) {
1569 WARN("Remote relayd disconnected. Stopping");
1570 relayd_hang_up = 1;
1571 goto write_error;
1572 }
1573 goto end;
1574 }
1575 } else {
1576 /* No streaming, we have to set the len with the full padding */
1577 len += padding;
1578
1579 /*
1580 * Check if we need to change the tracefile before writing the packet.
1581 */
1582 if (stream->chan->tracefile_size > 0 &&
1583 (stream->tracefile_size_current + len) >
1584 stream->chan->tracefile_size) {
1585 ret = utils_rotate_stream_file(stream->chan->pathname,
1586 stream->name, stream->chan->tracefile_size,
1587 stream->chan->tracefile_count, stream->uid, stream->gid,
1588 stream->out_fd, &(stream->tracefile_count_current));
1589 if (ret < 0) {
1590 ERR("Rotating output file");
1591 goto end;
1592 }
1593 outfd = stream->out_fd = ret;
1594 /* Reset current size because we just perform a rotation. */
1595 stream->tracefile_size_current = 0;
1596 }
1597 stream->tracefile_size_current += len;
1598 }
1599
1600 while (len > 0) {
1601 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
1602 (unsigned long)offset, len, fd, splice_pipe[1]);
1603 ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
1604 SPLICE_F_MOVE | SPLICE_F_MORE);
1605 DBG("splice chan to pipe, ret %zd", ret_splice);
1606 if (ret_splice < 0) {
1607 PERROR("Error in relay splice");
1608 if (written == 0) {
1609 written = ret_splice;
1610 }
1611 ret = errno;
1612 goto splice_error;
1613 }
1614
1615 /* Handle stream on the relayd if the output is on the network */
1616 if (relayd) {
1617 if (stream->metadata_flag) {
1618 size_t metadata_payload_size =
1619 sizeof(struct lttcomm_relayd_metadata_payload);
1620
1621 /* Update counter to fit the spliced data */
1622 ret_splice += metadata_payload_size;
1623 len += metadata_payload_size;
1624 /*
1625 * We do this so the return value can match the len passed as
1626 * argument to this function.
1627 */
1628 written -= metadata_payload_size;
1629 }
1630 }
1631
1632 /* Splice data out */
1633 ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
1634 ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
1635 DBG("Consumer splice pipe to file, ret %zd", ret_splice);
1636 if (ret_splice < 0) {
1637 PERROR("Error in file splice");
1638 if (written == 0) {
1639 written = ret_splice;
1640 }
1641 /* Socket operation failed. We consider the relayd dead */
1642 if (errno == EBADF || errno == EPIPE) {
1643 WARN("Remote relayd disconnected. Stopping");
1644 relayd_hang_up = 1;
1645 goto write_error;
1646 }
1647 ret = errno;
1648 goto splice_error;
1649 } else if (ret_splice > len) {
1650 errno = EINVAL;
1651 PERROR("Wrote more data than requested %zd (len: %lu)",
1652 ret_splice, len);
1653 written += ret_splice;
1654 ret = errno;
1655 goto splice_error;
1656 }
1657 len -= ret_splice;
1658
1659 /* This call is useless on a socket so better save a syscall. */
1660 if (!relayd) {
1661 /* This won't block, but will start writeout asynchronously */
1662 lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
1663 SYNC_FILE_RANGE_WRITE);
1664 stream->out_fd_offset += ret_splice;
1665 }
1666 written += ret_splice;
1667 }
1668 lttng_consumer_sync_trace_file(stream, orig_offset);
1669
1670 ret = ret_splice;
1671
1672 goto end;
1673
1674write_error:
1675 /*
1676 * This is a special case that the relayd has closed its socket. Let's
1677 * cleanup the relayd object and all associated streams.
1678 */
1679 if (relayd && relayd_hang_up) {
1680 cleanup_relayd(relayd, ctx);
1681 /* Skip splice error so the consumer does not fail */
1682 goto end;
1683 }
1684
1685splice_error:
1686 /* send the appropriate error description to sessiond */
1687 switch (ret) {
1688 case EINVAL:
1689 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
1690 break;
1691 case ENOMEM:
1692 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
1693 break;
1694 case ESPIPE:
1695 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
1696 break;
1697 }
1698
1699end:
1700 if (relayd && stream->metadata_flag) {
1701 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1702 }
1703
1704 rcu_read_unlock();
1705 return written;
1706}
1707
1708/*
1709 * Take a snapshot for a specific fd
1710 *
1711 * Returns 0 on success, < 0 on error
1712 */
1713int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
1714{
1715 switch (consumer_data.type) {
1716 case LTTNG_CONSUMER_KERNEL:
1717 return lttng_kconsumer_take_snapshot(stream);
1718 case LTTNG_CONSUMER32_UST:
1719 case LTTNG_CONSUMER64_UST:
1720 return lttng_ustconsumer_take_snapshot(stream);
1721 default:
1722 ERR("Unknown consumer_data type");
1723 assert(0);
1724 return -ENOSYS;
1725 }
1726}
1727
1728/*
1729 * Get the produced position
1730 *
1731 * Returns 0 on success, < 0 on error
1732 */
1733int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
1734 unsigned long *pos)
1735{
1736 switch (consumer_data.type) {
1737 case LTTNG_CONSUMER_KERNEL:
1738 return lttng_kconsumer_get_produced_snapshot(stream, pos);
1739 case LTTNG_CONSUMER32_UST:
1740 case LTTNG_CONSUMER64_UST:
1741 return lttng_ustconsumer_get_produced_snapshot(stream, pos);
1742 default:
1743 ERR("Unknown consumer_data type");
1744 assert(0);
1745 return -ENOSYS;
1746 }
1747}
1748
1749int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1750 int sock, struct pollfd *consumer_sockpoll)
1751{
1752 switch (consumer_data.type) {
1753 case LTTNG_CONSUMER_KERNEL:
1754 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
1755 case LTTNG_CONSUMER32_UST:
1756 case LTTNG_CONSUMER64_UST:
1757 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
1758 default:
1759 ERR("Unknown consumer_data type");
1760 assert(0);
1761 return -ENOSYS;
1762 }
1763}
1764
1765/*
1766 * Iterate over all streams of the hashtable and free them properly.
1767 *
1768 * WARNING: *MUST* be used with data stream only.
1769 */
1770static void destroy_data_stream_ht(struct lttng_ht *ht)
1771{
1772 struct lttng_ht_iter iter;
1773 struct lttng_consumer_stream *stream;
1774
1775 if (ht == NULL) {
1776 return;
1777 }
1778
1779 rcu_read_lock();
1780 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1781 /*
1782 * Ignore return value since we are currently cleaning up so any error
1783 * can't be handled.
1784 */
1785 (void) consumer_del_stream(stream, ht);
1786 }
1787 rcu_read_unlock();
1788
1789 lttng_ht_destroy(ht);
1790}
1791
1792/*
1793 * Iterate over all streams of the hashtable and free them properly.
1794 *
1795 * XXX: Should not be only for metadata stream or else use an other name.
1796 */
1797static void destroy_stream_ht(struct lttng_ht *ht)
1798{
1799 struct lttng_ht_iter iter;
1800 struct lttng_consumer_stream *stream;
1801
1802 if (ht == NULL) {
1803 return;
1804 }
1805
1806 rcu_read_lock();
1807 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1808 /*
1809 * Ignore return value since we are currently cleaning up so any error
1810 * can't be handled.
1811 */
1812 (void) consumer_del_metadata_stream(stream, ht);
1813 }
1814 rcu_read_unlock();
1815
1816 lttng_ht_destroy(ht);
1817}
1818
1819void lttng_consumer_close_metadata(void)
1820{
1821 switch (consumer_data.type) {
1822 case LTTNG_CONSUMER_KERNEL:
1823 /*
1824 * The Kernel consumer has a different metadata scheme so we don't
1825 * close anything because the stream will be closed by the session
1826 * daemon.
1827 */
1828 break;
1829 case LTTNG_CONSUMER32_UST:
1830 case LTTNG_CONSUMER64_UST:
1831 /*
1832 * Close all metadata streams. The metadata hash table is passed and
1833 * this call iterates over it by closing all wakeup fd. This is safe
1834 * because at this point we are sure that the metadata producer is
1835 * either dead or blocked.
1836 */
1837 lttng_ustconsumer_close_metadata(metadata_ht);
1838 break;
1839 default:
1840 ERR("Unknown consumer_data type");
1841 assert(0);
1842 }
1843}
1844
1845/*
1846 * Clean up a metadata stream and free its memory.
1847 */
1848void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
1849 struct lttng_ht *ht)
1850{
1851 int ret;
1852 struct lttng_ht_iter iter;
1853 struct lttng_consumer_channel *free_chan = NULL;
1854 struct consumer_relayd_sock_pair *relayd;
1855
1856 assert(stream);
1857 /*
1858 * This call should NEVER receive regular stream. It must always be
1859 * metadata stream and this is crucial for data structure synchronization.
1860 */
1861 assert(stream->metadata_flag);
1862
1863 DBG3("Consumer delete metadata stream %d", stream->wait_fd);
1864
1865 if (ht == NULL) {
1866 /* Means the stream was allocated but not successfully added */
1867 goto free_stream_rcu;
1868 }
1869
1870 pthread_mutex_lock(&consumer_data.lock);
1871 pthread_mutex_lock(&stream->chan->lock);
1872 pthread_mutex_lock(&stream->chan->timer_lock);
1873 pthread_mutex_lock(&stream->lock);
1874
1875 switch (consumer_data.type) {
1876 case LTTNG_CONSUMER_KERNEL:
1877 if (stream->mmap_base != NULL) {
1878 ret = munmap(stream->mmap_base, stream->mmap_len);
1879 if (ret != 0) {
1880 PERROR("munmap metadata stream");
1881 }
1882 }
1883 if (stream->wait_fd >= 0) {
1884 ret = close(stream->wait_fd);
1885 if (ret < 0) {
1886 PERROR("close kernel metadata wait_fd");
1887 }
1888 }
1889 break;
1890 case LTTNG_CONSUMER32_UST:
1891 case LTTNG_CONSUMER64_UST:
1892 if (stream->monitor) {
1893 /* close the write-side in close_metadata */
1894 ret = close(stream->ust_metadata_poll_pipe[0]);
1895 if (ret < 0) {
1896 PERROR("Close UST metadata read-side poll pipe");
1897 }
1898 }
1899 lttng_ustconsumer_del_stream(stream);
1900 break;
1901 default:
1902 ERR("Unknown consumer_data type");
1903 assert(0);
1904 goto end;
1905 }
1906
1907 rcu_read_lock();
1908 iter.iter.node = &stream->node.node;
1909 ret = lttng_ht_del(ht, &iter);
1910 assert(!ret);
1911
1912 iter.iter.node = &stream->node_channel_id.node;
1913 ret = lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
1914 assert(!ret);
1915
1916 iter.iter.node = &stream->node_session_id.node;
1917 ret = lttng_ht_del(consumer_data.stream_list_ht, &iter);
1918 assert(!ret);
1919 rcu_read_unlock();
1920
1921 if (stream->out_fd >= 0) {
1922 ret = close(stream->out_fd);
1923 if (ret) {
1924 PERROR("close");
1925 }
1926 }
1927
1928 /* Check and cleanup relayd */
1929 rcu_read_lock();
1930 relayd = consumer_find_relayd(stream->net_seq_idx);
1931 if (relayd != NULL) {
1932 uatomic_dec(&relayd->refcount);
1933 assert(uatomic_read(&relayd->refcount) >= 0);
1934
1935 /* Closing streams requires to lock the control socket. */
1936 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1937 ret = relayd_send_close_stream(&relayd->control_sock,
1938 stream->relayd_stream_id, stream->next_net_seq_num - 1);
1939 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1940 if (ret < 0) {
1941 DBG("Unable to close stream on the relayd. Continuing");
1942 /*
1943 * Continue here. There is nothing we can do for the relayd.
1944 * Chances are that the relayd has closed the socket so we just
1945 * continue cleaning up.
1946 */
1947 }
1948
1949 /* Both conditions are met, we destroy the relayd. */
1950 if (uatomic_read(&relayd->refcount) == 0 &&
1951 uatomic_read(&relayd->destroy_flag)) {
1952 consumer_destroy_relayd(relayd);
1953 }
1954 }
1955 rcu_read_unlock();
1956
1957 /* Atomically decrement channel refcount since other threads can use it. */
1958 if (!uatomic_sub_return(&stream->chan->refcount, 1)
1959 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
1960 /* Go for channel deletion! */
1961 free_chan = stream->chan;
1962 }
1963
1964end:
1965 /*
1966 * Nullify the stream reference so it is not used after deletion. The
1967 * consumer data lock MUST be acquired before being able to check for a
1968 * NULL pointer value.
1969 */
1970 stream->chan->metadata_stream = NULL;
1971
1972 pthread_mutex_unlock(&stream->lock);
1973 pthread_mutex_unlock(&stream->chan->timer_lock);
1974 pthread_mutex_unlock(&stream->chan->lock);
1975 pthread_mutex_unlock(&consumer_data.lock);
1976
1977 if (free_chan) {
1978 consumer_del_channel(free_chan);
1979 }
1980
1981free_stream_rcu:
1982 call_rcu(&stream->node.head, free_stream_rcu);
1983}
1984
1985/*
1986 * Action done with the metadata stream when adding it to the consumer internal
1987 * data structures to handle it.
1988 */
1989static int add_metadata_stream(struct lttng_consumer_stream *stream,
1990 struct lttng_ht *ht)
1991{
1992 int ret = 0;
1993 struct lttng_ht_iter iter;
1994 struct lttng_ht_node_u64 *node;
1995
1996 assert(stream);
1997 assert(ht);
1998
1999 DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
2000
2001 pthread_mutex_lock(&consumer_data.lock);
2002 pthread_mutex_lock(&stream->chan->lock);
2003 pthread_mutex_lock(&stream->chan->timer_lock);
2004 pthread_mutex_lock(&stream->lock);
2005
2006 /*
2007 * From here, refcounts are updated so be _careful_ when returning an error
2008 * after this point.
2009 */
2010
2011 rcu_read_lock();
2012
2013 /*
2014 * Lookup the stream just to make sure it does not exist in our internal
2015 * state. This should NEVER happen.
2016 */
2017 lttng_ht_lookup(ht, &stream->key, &iter);
2018 node = lttng_ht_iter_get_node_u64(&iter);
2019 assert(!node);
2020
2021 /*
2022 * When nb_init_stream_left reaches 0, we don't need to trigger any action
2023 * in terms of destroying the associated channel, because the action that
2024 * causes the count to become 0 also causes a stream to be added. The
2025 * channel deletion will thus be triggered by the following removal of this
2026 * stream.
2027 */
2028 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
2029 /* Increment refcount before decrementing nb_init_stream_left */
2030 cmm_smp_wmb();
2031 uatomic_dec(&stream->chan->nb_init_stream_left);
2032 }
2033
2034 lttng_ht_add_unique_u64(ht, &stream->node);
2035
2036 lttng_ht_add_unique_u64(consumer_data.stream_per_chan_id_ht,
2037 &stream->node_channel_id);
2038
2039 /*
2040 * Add stream to the stream_list_ht of the consumer data. No need to steal
2041 * the key since the HT does not use it and we allow to add redundant keys
2042 * into this table.
2043 */
2044 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
2045
2046 rcu_read_unlock();
2047
2048 pthread_mutex_unlock(&stream->lock);
2049 pthread_mutex_unlock(&stream->chan->lock);
2050 pthread_mutex_unlock(&stream->chan->timer_lock);
2051 pthread_mutex_unlock(&consumer_data.lock);
2052 return ret;
2053}
2054
2055/*
2056 * Delete data stream that are flagged for deletion (endpoint_status).
2057 */
2058static void validate_endpoint_status_data_stream(void)
2059{
2060 struct lttng_ht_iter iter;
2061 struct lttng_consumer_stream *stream;
2062
2063 DBG("Consumer delete flagged data stream");
2064
2065 rcu_read_lock();
2066 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
2067 /* Validate delete flag of the stream */
2068 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
2069 continue;
2070 }
2071 /* Delete it right now */
2072 consumer_del_stream(stream, data_ht);
2073 }
2074 rcu_read_unlock();
2075}
2076
2077/*
2078 * Delete metadata stream that are flagged for deletion (endpoint_status).
2079 */
2080static void validate_endpoint_status_metadata_stream(
2081 struct lttng_poll_event *pollset)
2082{
2083 struct lttng_ht_iter iter;
2084 struct lttng_consumer_stream *stream;
2085
2086 DBG("Consumer delete flagged metadata stream");
2087
2088 assert(pollset);
2089
2090 rcu_read_lock();
2091 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
2092 /* Validate delete flag of the stream */
2093 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
2094 continue;
2095 }
2096 /*
2097 * Remove from pollset so the metadata thread can continue without
2098 * blocking on a deleted stream.
2099 */
2100 lttng_poll_del(pollset, stream->wait_fd);
2101
2102 /* Delete it right now */
2103 consumer_del_metadata_stream(stream, metadata_ht);
2104 }
2105 rcu_read_unlock();
2106}
2107
2108/*
2109 * Thread polls on metadata file descriptor and write them on disk or on the
2110 * network.
2111 */
2112void *consumer_thread_metadata_poll(void *data)
2113{
2114 int ret, i, pollfd;
2115 uint32_t revents, nb_fd;
2116 struct lttng_consumer_stream *stream = NULL;
2117 struct lttng_ht_iter iter;
2118 struct lttng_ht_node_u64 *node;
2119 struct lttng_poll_event events;
2120 struct lttng_consumer_local_data *ctx = data;
2121 ssize_t len;
2122
2123 rcu_register_thread();
2124
2125 metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2126 if (!metadata_ht) {
2127 /* ENOMEM at this point. Better to bail out. */
2128 goto end_ht;
2129 }
2130
2131 DBG("Thread metadata poll started");
2132
2133 /* Size is set to 1 for the consumer_metadata pipe */
2134 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2135 if (ret < 0) {
2136 ERR("Poll set creation failed");
2137 goto end_poll;
2138 }
2139
2140 ret = lttng_poll_add(&events,
2141 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
2142 if (ret < 0) {
2143 goto end;
2144 }
2145
2146 /* Main loop */
2147 DBG("Metadata main loop started");
2148
2149 while (1) {
2150 /* Only the metadata pipe is set */
2151 if (LTTNG_POLL_GETNB(&events) == 0 && consumer_quit == 1) {
2152 goto end;
2153 }
2154
2155restart:
2156 DBG("Metadata poll wait with %d fd(s)", LTTNG_POLL_GETNB(&events));
2157 ret = lttng_poll_wait(&events, -1);
2158 DBG("Metadata event catched in thread");
2159 if (ret < 0) {
2160 if (errno == EINTR) {
2161 ERR("Poll EINTR catched");
2162 goto restart;
2163 }
2164 goto error;
2165 }
2166
2167 nb_fd = ret;
2168
2169 /* From here, the event is a metadata wait fd */
2170 for (i = 0; i < nb_fd; i++) {
2171 revents = LTTNG_POLL_GETEV(&events, i);
2172 pollfd = LTTNG_POLL_GETFD(&events, i);
2173
2174 /* Just don't waste time if no returned events for the fd */
2175 if (!revents) {
2176 continue;
2177 }
2178
2179 if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
2180 if (revents & (LPOLLERR | LPOLLHUP )) {
2181 DBG("Metadata thread pipe hung up");
2182 /*
2183 * Remove the pipe from the poll set and continue the loop
2184 * since their might be data to consume.
2185 */
2186 lttng_poll_del(&events,
2187 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2188 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
2189 continue;
2190 } else if (revents & LPOLLIN) {
2191 ssize_t pipe_len;
2192
2193 pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
2194 &stream, sizeof(stream));
2195 if (pipe_len < 0) {
2196 ERR("read metadata stream, ret: %ld", pipe_len);
2197 /*
2198 * Continue here to handle the rest of the streams.
2199 */
2200 continue;
2201 }
2202
2203 /* A NULL stream means that the state has changed. */
2204 if (stream == NULL) {
2205 /* Check for deleted streams. */
2206 validate_endpoint_status_metadata_stream(&events);
2207 goto restart;
2208 }
2209
2210 DBG("Adding metadata stream %d to poll set",
2211 stream->wait_fd);
2212
2213 ret = add_metadata_stream(stream, metadata_ht);
2214 if (ret) {
2215 ERR("Unable to add metadata stream");
2216 /* Stream was not setup properly. Continuing. */
2217 consumer_del_metadata_stream(stream, NULL);
2218 continue;
2219 }
2220
2221 /* Add metadata stream to the global poll events list */
2222 lttng_poll_add(&events, stream->wait_fd,
2223 LPOLLIN | LPOLLPRI);
2224 }
2225
2226 /* Handle other stream */
2227 continue;
2228 }
2229
2230 rcu_read_lock();
2231 {
2232 uint64_t tmp_id = (uint64_t) pollfd;
2233
2234 lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
2235 }
2236 node = lttng_ht_iter_get_node_u64(&iter);
2237 assert(node);
2238
2239 stream = caa_container_of(node, struct lttng_consumer_stream,
2240 node);
2241
2242 /* Check for error event */
2243 if (revents & (LPOLLERR | LPOLLHUP)) {
2244 DBG("Metadata fd %d is hup|err.", pollfd);
2245 if (!stream->hangup_flush_done
2246 && (consumer_data.type == LTTNG_CONSUMER32_UST
2247 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2248 DBG("Attempting to flush and consume the UST buffers");
2249 lttng_ustconsumer_on_stream_hangup(stream);
2250
2251 /* We just flushed the stream now read it. */
2252 do {
2253 len = ctx->on_buffer_ready(stream, ctx);
2254 /*
2255 * We don't check the return value here since if we get
2256 * a negative len, it means an error occured thus we
2257 * simply remove it from the poll set and free the
2258 * stream.
2259 */
2260 } while (len > 0);
2261 }
2262
2263 lttng_poll_del(&events, stream->wait_fd);
2264 /*
2265 * This call update the channel states, closes file descriptors
2266 * and securely free the stream.
2267 */
2268 consumer_del_metadata_stream(stream, metadata_ht);
2269 } else if (revents & (LPOLLIN | LPOLLPRI)) {
2270 /* Get the data out of the metadata file descriptor */
2271 DBG("Metadata available on fd %d", pollfd);
2272 assert(stream->wait_fd == pollfd);
2273
2274 do {
2275 len = ctx->on_buffer_ready(stream, ctx);
2276 /*
2277 * We don't check the return value here since if we get
2278 * a negative len, it means an error occured thus we
2279 * simply remove it from the poll set and free the
2280 * stream.
2281 */
2282 } while (len > 0);
2283
2284 /* It's ok to have an unavailable sub-buffer */
2285 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2286 /* Clean up stream from consumer and free it. */
2287 lttng_poll_del(&events, stream->wait_fd);
2288 consumer_del_metadata_stream(stream, metadata_ht);
2289 }
2290 }
2291
2292 /* Release RCU lock for the stream looked up */
2293 rcu_read_unlock();
2294 }
2295 }
2296
2297error:
2298end:
2299 DBG("Metadata poll thread exiting");
2300
2301 lttng_poll_clean(&events);
2302end_poll:
2303 destroy_stream_ht(metadata_ht);
2304end_ht:
2305 rcu_unregister_thread();
2306 return NULL;
2307}
2308
2309/*
2310 * This thread polls the fds in the set to consume the data and write
2311 * it to tracefile if necessary.
2312 */
2313void *consumer_thread_data_poll(void *data)
2314{
2315 int num_rdy, num_hup, high_prio, ret, i;
2316 struct pollfd *pollfd = NULL;
2317 /* local view of the streams */
2318 struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
2319 /* local view of consumer_data.fds_count */
2320 int nb_fd = 0;
2321 struct lttng_consumer_local_data *ctx = data;
2322 ssize_t len;
2323
2324 rcu_register_thread();
2325
2326 data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2327 if (data_ht == NULL) {
2328 /* ENOMEM at this point. Better to bail out. */
2329 goto end;
2330 }
2331
2332 local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
2333 if (local_stream == NULL) {
2334 PERROR("local_stream malloc");
2335 goto end;
2336 }
2337
2338 while (1) {
2339 high_prio = 0;
2340 num_hup = 0;
2341
2342 /*
2343 * the fds set has been updated, we need to update our
2344 * local array as well
2345 */
2346 pthread_mutex_lock(&consumer_data.lock);
2347 if (consumer_data.need_update) {
2348 free(pollfd);
2349 pollfd = NULL;
2350
2351 free(local_stream);
2352 local_stream = NULL;
2353
2354 /* allocate for all fds + 1 for the consumer_data_pipe */
2355 pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
2356 if (pollfd == NULL) {
2357 PERROR("pollfd malloc");
2358 pthread_mutex_unlock(&consumer_data.lock);
2359 goto end;
2360 }
2361
2362 /* allocate for all fds + 1 for the consumer_data_pipe */
2363 local_stream = zmalloc((consumer_data.stream_count + 1) *
2364 sizeof(struct lttng_consumer_stream *));
2365 if (local_stream == NULL) {
2366 PERROR("local_stream malloc");
2367 pthread_mutex_unlock(&consumer_data.lock);
2368 goto end;
2369 }
2370 ret = update_poll_array(ctx, &pollfd, local_stream,
2371 data_ht);
2372 if (ret < 0) {
2373 ERR("Error in allocating pollfd or local_outfds");
2374 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
2375 pthread_mutex_unlock(&consumer_data.lock);
2376 goto end;
2377 }
2378 nb_fd = ret;
2379 consumer_data.need_update = 0;
2380 }
2381 pthread_mutex_unlock(&consumer_data.lock);
2382
2383 /* No FDs and consumer_quit, consumer_cleanup the thread */
2384 if (nb_fd == 0 && consumer_quit == 1) {
2385 goto end;
2386 }
2387 /* poll on the array of fds */
2388 restart:
2389 DBG("polling on %d fd", nb_fd + 1);
2390 num_rdy = poll(pollfd, nb_fd + 1, -1);
2391 DBG("poll num_rdy : %d", num_rdy);
2392 if (num_rdy == -1) {
2393 /*
2394 * Restart interrupted system call.
2395 */
2396 if (errno == EINTR) {
2397 goto restart;
2398 }
2399 PERROR("Poll error");
2400 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
2401 goto end;
2402 } else if (num_rdy == 0) {
2403 DBG("Polling thread timed out");
2404 goto end;
2405 }
2406
2407 /*
2408 * If the consumer_data_pipe triggered poll go directly to the
2409 * beginning of the loop to update the array. We want to prioritize
2410 * array update over low-priority reads.
2411 */
2412 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
2413 ssize_t pipe_readlen;
2414
2415 DBG("consumer_data_pipe wake up");
2416 pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
2417 &new_stream, sizeof(new_stream));
2418 if (pipe_readlen < 0) {
2419 ERR("Consumer data pipe ret %ld", pipe_readlen);
2420 /* Continue so we can at least handle the current stream(s). */
2421 continue;
2422 }
2423
2424 /*
2425 * If the stream is NULL, just ignore it. It's also possible that
2426 * the sessiond poll thread changed the consumer_quit state and is
2427 * waking us up to test it.
2428 */
2429 if (new_stream == NULL) {
2430 validate_endpoint_status_data_stream();
2431 continue;
2432 }
2433
2434 ret = add_stream(new_stream, data_ht);
2435 if (ret) {
2436 ERR("Consumer add stream %" PRIu64 " failed. Continuing",
2437 new_stream->key);
2438 /*
2439 * At this point, if the add_stream fails, it is not in the
2440 * hash table thus passing the NULL value here.
2441 */
2442 consumer_del_stream(new_stream, NULL);
2443 }
2444
2445 /* Continue to update the local streams and handle prio ones */
2446 continue;
2447 }
2448
2449 /* Take care of high priority channels first. */
2450 for (i = 0; i < nb_fd; i++) {
2451 if (local_stream[i] == NULL) {
2452 continue;
2453 }
2454 if (pollfd[i].revents & POLLPRI) {
2455 DBG("Urgent read on fd %d", pollfd[i].fd);
2456 high_prio = 1;
2457 len = ctx->on_buffer_ready(local_stream[i], ctx);
2458 /* it's ok to have an unavailable sub-buffer */
2459 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2460 /* Clean the stream and free it. */
2461 consumer_del_stream(local_stream[i], data_ht);
2462 local_stream[i] = NULL;
2463 } else if (len > 0) {
2464 local_stream[i]->data_read = 1;
2465 }
2466 }
2467 }
2468
2469 /*
2470 * If we read high prio channel in this loop, try again
2471 * for more high prio data.
2472 */
2473 if (high_prio) {
2474 continue;
2475 }
2476
2477 /* Take care of low priority channels. */
2478 for (i = 0; i < nb_fd; i++) {
2479 if (local_stream[i] == NULL) {
2480 continue;
2481 }
2482 if ((pollfd[i].revents & POLLIN) ||
2483 local_stream[i]->hangup_flush_done) {
2484 DBG("Normal read on fd %d", pollfd[i].fd);
2485 len = ctx->on_buffer_ready(local_stream[i], ctx);
2486 /* it's ok to have an unavailable sub-buffer */
2487 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2488 /* Clean the stream and free it. */
2489 consumer_del_stream(local_stream[i], data_ht);
2490 local_stream[i] = NULL;
2491 } else if (len > 0) {
2492 local_stream[i]->data_read = 1;
2493 }
2494 }
2495 }
2496
2497 /* Handle hangup and errors */
2498 for (i = 0; i < nb_fd; i++) {
2499 if (local_stream[i] == NULL) {
2500 continue;
2501 }
2502 if (!local_stream[i]->hangup_flush_done
2503 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
2504 && (consumer_data.type == LTTNG_CONSUMER32_UST
2505 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2506 DBG("fd %d is hup|err|nval. Attempting flush and read.",
2507 pollfd[i].fd);
2508 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
2509 /* Attempt read again, for the data we just flushed. */
2510 local_stream[i]->data_read = 1;
2511 }
2512 /*
2513 * If the poll flag is HUP/ERR/NVAL and we have
2514 * read no data in this pass, we can remove the
2515 * stream from its hash table.
2516 */
2517 if ((pollfd[i].revents & POLLHUP)) {
2518 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
2519 if (!local_stream[i]->data_read) {
2520 consumer_del_stream(local_stream[i], data_ht);
2521 local_stream[i] = NULL;
2522 num_hup++;
2523 }
2524 } else if (pollfd[i].revents & POLLERR) {
2525 ERR("Error returned in polling fd %d.", pollfd[i].fd);
2526 if (!local_stream[i]->data_read) {
2527 consumer_del_stream(local_stream[i], data_ht);
2528 local_stream[i] = NULL;
2529 num_hup++;
2530 }
2531 } else if (pollfd[i].revents & POLLNVAL) {
2532 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
2533 if (!local_stream[i]->data_read) {
2534 consumer_del_stream(local_stream[i], data_ht);
2535 local_stream[i] = NULL;
2536 num_hup++;
2537 }
2538 }
2539 if (local_stream[i] != NULL) {
2540 local_stream[i]->data_read = 0;
2541 }
2542 }
2543 }
2544end:
2545 DBG("polling thread exiting");
2546 free(pollfd);
2547 free(local_stream);
2548
2549 /*
2550 * Close the write side of the pipe so epoll_wait() in
2551 * consumer_thread_metadata_poll can catch it. The thread is monitoring the
2552 * read side of the pipe. If we close them both, epoll_wait strangely does
2553 * not return and could create a endless wait period if the pipe is the
2554 * only tracked fd in the poll set. The thread will take care of closing
2555 * the read side.
2556 */
2557 (void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
2558
2559 destroy_data_stream_ht(data_ht);
2560
2561 rcu_unregister_thread();
2562 return NULL;
2563}
2564
2565/*
2566 * Close wake-up end of each stream belonging to the channel. This will
2567 * allow the poll() on the stream read-side to detect when the
2568 * write-side (application) finally closes them.
2569 */
2570static
2571void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
2572{
2573 struct lttng_ht *ht;
2574 struct lttng_consumer_stream *stream;
2575 struct lttng_ht_iter iter;
2576
2577 ht = consumer_data.stream_per_chan_id_ht;
2578
2579 rcu_read_lock();
2580 cds_lfht_for_each_entry_duplicate(ht->ht,
2581 ht->hash_fct(&channel->key, lttng_ht_seed),
2582 ht->match_fct, &channel->key,
2583 &iter.iter, stream, node_channel_id.node) {
2584 /*
2585 * Protect against teardown with mutex.
2586 */
2587 pthread_mutex_lock(&stream->lock);
2588 if (cds_lfht_is_node_deleted(&stream->node.node)) {
2589 goto next;
2590 }
2591 switch (consumer_data.type) {
2592 case LTTNG_CONSUMER_KERNEL:
2593 break;
2594 case LTTNG_CONSUMER32_UST:
2595 case LTTNG_CONSUMER64_UST:
2596 /*
2597 * Note: a mutex is taken internally within
2598 * liblttng-ust-ctl to protect timer wakeup_fd
2599 * use from concurrent close.
2600 */
2601 lttng_ustconsumer_close_stream_wakeup(stream);
2602 break;
2603 default:
2604 ERR("Unknown consumer_data type");
2605 assert(0);
2606 }
2607 next:
2608 pthread_mutex_unlock(&stream->lock);
2609 }
2610 rcu_read_unlock();
2611}
2612
2613static void destroy_channel_ht(struct lttng_ht *ht)
2614{
2615 struct lttng_ht_iter iter;
2616 struct lttng_consumer_channel *channel;
2617 int ret;
2618
2619 if (ht == NULL) {
2620 return;
2621 }
2622
2623 rcu_read_lock();
2624 cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
2625 ret = lttng_ht_del(ht, &iter);
2626 assert(ret != 0);
2627 }
2628 rcu_read_unlock();
2629
2630 lttng_ht_destroy(ht);
2631}
2632
2633/*
2634 * This thread polls the channel fds to detect when they are being
2635 * closed. It closes all related streams if the channel is detected as
2636 * closed. It is currently only used as a shim layer for UST because the
2637 * consumerd needs to keep the per-stream wakeup end of pipes open for
2638 * periodical flush.
2639 */
2640void *consumer_thread_channel_poll(void *data)
2641{
2642 int ret, i, pollfd;
2643 uint32_t revents, nb_fd;
2644 struct lttng_consumer_channel *chan = NULL;
2645 struct lttng_ht_iter iter;
2646 struct lttng_ht_node_u64 *node;
2647 struct lttng_poll_event events;
2648 struct lttng_consumer_local_data *ctx = data;
2649 struct lttng_ht *channel_ht;
2650
2651 rcu_register_thread();
2652
2653 channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2654 if (!channel_ht) {
2655 /* ENOMEM at this point. Better to bail out. */
2656 goto end_ht;
2657 }
2658
2659 DBG("Thread channel poll started");
2660
2661 /* Size is set to 1 for the consumer_channel pipe */
2662 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2663 if (ret < 0) {
2664 ERR("Poll set creation failed");
2665 goto end_poll;
2666 }
2667
2668 ret = lttng_poll_add(&events, ctx->consumer_channel_pipe[0], LPOLLIN);
2669 if (ret < 0) {
2670 goto end;
2671 }
2672
2673 /* Main loop */
2674 DBG("Channel main loop started");
2675
2676 while (1) {
2677 /* Only the channel pipe is set */
2678 if (LTTNG_POLL_GETNB(&events) == 0 && consumer_quit == 1) {
2679 goto end;
2680 }
2681
2682restart:
2683 DBG("Channel poll wait with %d fd(s)", LTTNG_POLL_GETNB(&events));
2684 ret = lttng_poll_wait(&events, -1);
2685 DBG("Channel event catched in thread");
2686 if (ret < 0) {
2687 if (errno == EINTR) {
2688 ERR("Poll EINTR catched");
2689 goto restart;
2690 }
2691 goto end;
2692 }
2693
2694 nb_fd = ret;
2695
2696 /* From here, the event is a channel wait fd */
2697 for (i = 0; i < nb_fd; i++) {
2698 revents = LTTNG_POLL_GETEV(&events, i);
2699 pollfd = LTTNG_POLL_GETFD(&events, i);
2700
2701 /* Just don't waste time if no returned events for the fd */
2702 if (!revents) {
2703 continue;
2704 }
2705 if (pollfd == ctx->consumer_channel_pipe[0]) {
2706 if (revents & (LPOLLERR | LPOLLHUP)) {
2707 DBG("Channel thread pipe hung up");
2708 /*
2709 * Remove the pipe from the poll set and continue the loop
2710 * since their might be data to consume.
2711 */
2712 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
2713 continue;
2714 } else if (revents & LPOLLIN) {
2715 enum consumer_channel_action action;
2716 uint64_t key;
2717
2718 ret = read_channel_pipe(ctx, &chan, &key, &action);
2719 if (ret <= 0) {
2720 ERR("Error reading channel pipe");
2721 continue;
2722 }
2723
2724 switch (action) {
2725 case CONSUMER_CHANNEL_ADD:
2726 DBG("Adding channel %d to poll set",
2727 chan->wait_fd);
2728
2729 lttng_ht_node_init_u64(&chan->wait_fd_node,
2730 chan->wait_fd);
2731 rcu_read_lock();
2732 lttng_ht_add_unique_u64(channel_ht,
2733 &chan->wait_fd_node);
2734 rcu_read_unlock();
2735 /* Add channel to the global poll events list */
2736 lttng_poll_add(&events, chan->wait_fd,
2737 LPOLLIN | LPOLLPRI);
2738 break;
2739 case CONSUMER_CHANNEL_DEL:
2740 {
2741 struct lttng_consumer_stream *stream, *stmp;
2742
2743 rcu_read_lock();
2744 chan = consumer_find_channel(key);
2745 if (!chan) {
2746 rcu_read_unlock();
2747 ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
2748 break;
2749 }
2750 lttng_poll_del(&events, chan->wait_fd);
2751 iter.iter.node = &chan->wait_fd_node.node;
2752 ret = lttng_ht_del(channel_ht, &iter);
2753 assert(ret == 0);
2754 consumer_close_channel_streams(chan);
2755
2756 switch (consumer_data.type) {
2757 case LTTNG_CONSUMER_KERNEL:
2758 break;
2759 case LTTNG_CONSUMER32_UST:
2760 case LTTNG_CONSUMER64_UST:
2761 /* Delete streams that might have been left in the stream list. */
2762 cds_list_for_each_entry_safe(stream, stmp, &chan->streams.head,
2763 send_node) {
2764 cds_list_del(&stream->send_node);
2765 lttng_ustconsumer_del_stream(stream);
2766 uatomic_sub(&stream->chan->refcount, 1);
2767 assert(&chan->refcount);
2768 free(stream);
2769 }
2770 break;
2771 default:
2772 ERR("Unknown consumer_data type");
2773 assert(0);
2774 }
2775
2776 /*
2777 * Release our own refcount. Force channel deletion even if
2778 * streams were not initialized.
2779 */
2780 if (!uatomic_sub_return(&chan->refcount, 1)) {
2781 consumer_del_channel(chan);
2782 }
2783 rcu_read_unlock();
2784 goto restart;
2785 }
2786 case CONSUMER_CHANNEL_QUIT:
2787 /*
2788 * Remove the pipe from the poll set and continue the loop
2789 * since their might be data to consume.
2790 */
2791 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
2792 continue;
2793 default:
2794 ERR("Unknown action");
2795 break;
2796 }
2797 }
2798
2799 /* Handle other stream */
2800 continue;
2801 }
2802
2803 rcu_read_lock();
2804 {
2805 uint64_t tmp_id = (uint64_t) pollfd;
2806
2807 lttng_ht_lookup(channel_ht, &tmp_id, &iter);
2808 }
2809 node = lttng_ht_iter_get_node_u64(&iter);
2810 assert(node);
2811
2812 chan = caa_container_of(node, struct lttng_consumer_channel,
2813 wait_fd_node);
2814
2815 /* Check for error event */
2816 if (revents & (LPOLLERR | LPOLLHUP)) {
2817 DBG("Channel fd %d is hup|err.", pollfd);
2818
2819 lttng_poll_del(&events, chan->wait_fd);
2820 ret = lttng_ht_del(channel_ht, &iter);
2821 assert(ret == 0);
2822 consumer_close_channel_streams(chan);
2823
2824 /* Release our own refcount */
2825 if (!uatomic_sub_return(&chan->refcount, 1)
2826 && !uatomic_read(&chan->nb_init_stream_left)) {
2827 consumer_del_channel(chan);
2828 }
2829 }
2830
2831 /* Release RCU lock for the channel looked up */
2832 rcu_read_unlock();
2833 }
2834 }
2835
2836end:
2837 lttng_poll_clean(&events);
2838end_poll:
2839 destroy_channel_ht(channel_ht);
2840end_ht:
2841 DBG("Channel poll thread exiting");
2842 rcu_unregister_thread();
2843 return NULL;
2844}
2845
2846static int set_metadata_socket(struct lttng_consumer_local_data *ctx,
2847 struct pollfd *sockpoll, int client_socket)
2848{
2849 int ret;
2850
2851 assert(ctx);
2852 assert(sockpoll);
2853
2854 if (lttng_consumer_poll_socket(sockpoll) < 0) {
2855 ret = -1;
2856 goto error;
2857 }
2858 DBG("Metadata connection on client_socket");
2859
2860 /* Blocking call, waiting for transmission */
2861 ctx->consumer_metadata_socket = lttcomm_accept_unix_sock(client_socket);
2862 if (ctx->consumer_metadata_socket < 0) {
2863 WARN("On accept metadata");
2864 ret = -1;
2865 goto error;
2866 }
2867 ret = 0;
2868
2869error:
2870 return ret;
2871}
2872
2873/*
2874 * This thread listens on the consumerd socket and receives the file
2875 * descriptors from the session daemon.
2876 */
2877void *consumer_thread_sessiond_poll(void *data)
2878{
2879 int sock = -1, client_socket, ret;
2880 /*
2881 * structure to poll for incoming data on communication socket avoids
2882 * making blocking sockets.
2883 */
2884 struct pollfd consumer_sockpoll[2];
2885 struct lttng_consumer_local_data *ctx = data;
2886
2887 rcu_register_thread();
2888
2889 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
2890 unlink(ctx->consumer_command_sock_path);
2891 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
2892 if (client_socket < 0) {
2893 ERR("Cannot create command socket");
2894 goto end;
2895 }
2896
2897 ret = lttcomm_listen_unix_sock(client_socket);
2898 if (ret < 0) {
2899 goto end;
2900 }
2901
2902 DBG("Sending ready command to lttng-sessiond");
2903 ret = lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY);
2904 /* return < 0 on error, but == 0 is not fatal */
2905 if (ret < 0) {
2906 ERR("Error sending ready command to lttng-sessiond");
2907 goto end;
2908 }
2909
2910 /* prepare the FDs to poll : to client socket and the should_quit pipe */
2911 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
2912 consumer_sockpoll[0].events = POLLIN | POLLPRI;
2913 consumer_sockpoll[1].fd = client_socket;
2914 consumer_sockpoll[1].events = POLLIN | POLLPRI;
2915
2916 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
2917 goto end;
2918 }
2919 DBG("Connection on client_socket");
2920
2921 /* Blocking call, waiting for transmission */
2922 sock = lttcomm_accept_unix_sock(client_socket);
2923 if (sock < 0) {
2924 WARN("On accept");
2925 goto end;
2926 }
2927
2928 /*
2929 * Setup metadata socket which is the second socket connection on the
2930 * command unix socket.
2931 */
2932 ret = set_metadata_socket(ctx, consumer_sockpoll, client_socket);
2933 if (ret < 0) {
2934 goto end;
2935 }
2936
2937 /* This socket is not useful anymore. */
2938 ret = close(client_socket);
2939 if (ret < 0) {
2940 PERROR("close client_socket");
2941 }
2942 client_socket = -1;
2943
2944 /* update the polling structure to poll on the established socket */
2945 consumer_sockpoll[1].fd = sock;
2946 consumer_sockpoll[1].events = POLLIN | POLLPRI;
2947
2948 while (1) {
2949 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
2950 goto end;
2951 }
2952 DBG("Incoming command on sock");
2953 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
2954 if (ret == -ENOENT) {
2955 DBG("Received STOP command");
2956 goto end;
2957 }
2958 if (ret <= 0) {
2959 /*
2960 * This could simply be a session daemon quitting. Don't output
2961 * ERR() here.
2962 */
2963 DBG("Communication interrupted on command socket");
2964 goto end;
2965 }
2966 if (consumer_quit) {
2967 DBG("consumer_thread_receive_fds received quit from signal");
2968 goto end;
2969 }
2970 DBG("received command on sock");
2971 }
2972end:
2973 DBG("Consumer thread sessiond poll exiting");
2974
2975 /*
2976 * Close metadata streams since the producer is the session daemon which
2977 * just died.
2978 *
2979 * NOTE: for now, this only applies to the UST tracer.
2980 */
2981 lttng_consumer_close_metadata();
2982
2983 /*
2984 * when all fds have hung up, the polling thread
2985 * can exit cleanly
2986 */
2987 consumer_quit = 1;
2988
2989 /*
2990 * Notify the data poll thread to poll back again and test the
2991 * consumer_quit state that we just set so to quit gracefully.
2992 */
2993 notify_thread_lttng_pipe(ctx->consumer_data_pipe);
2994
2995 notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
2996
2997 /* Cleaning up possibly open sockets. */
2998 if (sock >= 0) {
2999 ret = close(sock);
3000 if (ret < 0) {
3001 PERROR("close sock sessiond poll");
3002 }
3003 }
3004 if (client_socket >= 0) {
3005 ret = close(client_socket);
3006 if (ret < 0) {
3007 PERROR("close client_socket sessiond poll");
3008 }
3009 }
3010
3011 rcu_unregister_thread();
3012 return NULL;
3013}
3014
3015ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
3016 struct lttng_consumer_local_data *ctx)
3017{
3018 ssize_t ret;
3019
3020 pthread_mutex_lock(&stream->lock);
3021
3022 switch (consumer_data.type) {
3023 case LTTNG_CONSUMER_KERNEL:
3024 ret = lttng_kconsumer_read_subbuffer(stream, ctx);
3025 break;
3026 case LTTNG_CONSUMER32_UST:
3027 case LTTNG_CONSUMER64_UST:
3028 ret = lttng_ustconsumer_read_subbuffer(stream, ctx);
3029 break;
3030 default:
3031 ERR("Unknown consumer_data type");
3032 assert(0);
3033 ret = -ENOSYS;
3034 break;
3035 }
3036
3037 pthread_mutex_unlock(&stream->lock);
3038 return ret;
3039}
3040
3041int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
3042{
3043 switch (consumer_data.type) {
3044 case LTTNG_CONSUMER_KERNEL:
3045 return lttng_kconsumer_on_recv_stream(stream);
3046 case LTTNG_CONSUMER32_UST:
3047 case LTTNG_CONSUMER64_UST:
3048 return lttng_ustconsumer_on_recv_stream(stream);
3049 default:
3050 ERR("Unknown consumer_data type");
3051 assert(0);
3052 return -ENOSYS;
3053 }
3054}
3055
3056/*
3057 * Allocate and set consumer data hash tables.
3058 */
3059void lttng_consumer_init(void)
3060{
3061 consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3062 consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3063 consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3064 consumer_data.stream_per_chan_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3065}
3066
3067/*
3068 * Process the ADD_RELAYD command receive by a consumer.
3069 *
3070 * This will create a relayd socket pair and add it to the relayd hash table.
3071 * The caller MUST acquire a RCU read side lock before calling it.
3072 */
3073int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
3074 struct lttng_consumer_local_data *ctx, int sock,
3075 struct pollfd *consumer_sockpoll,
3076 struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id)
3077{
3078 int fd = -1, ret = -1, relayd_created = 0;
3079 enum lttng_error_code ret_code = LTTNG_OK;
3080 struct consumer_relayd_sock_pair *relayd = NULL;
3081
3082 assert(ctx);
3083 assert(relayd_sock);
3084
3085 DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
3086
3087 /* Get relayd reference if exists. */
3088 relayd = consumer_find_relayd(net_seq_idx);
3089 if (relayd == NULL) {
3090 assert(sock_type == LTTNG_STREAM_CONTROL);
3091 /* Not found. Allocate one. */
3092 relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
3093 if (relayd == NULL) {
3094 ret = -ENOMEM;
3095 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
3096 goto error;
3097 } else {
3098 relayd->sessiond_session_id = sessiond_id;
3099 relayd_created = 1;
3100 }
3101
3102 /*
3103 * This code path MUST continue to the consumer send status message to
3104 * we can notify the session daemon and continue our work without
3105 * killing everything.
3106 */
3107 } else {
3108 /*
3109 * relayd key should never be found for control socket.
3110 */
3111 assert(sock_type != LTTNG_STREAM_CONTROL);
3112 }
3113
3114 /* First send a status message before receiving the fds. */
3115 ret = consumer_send_status_msg(sock, LTTNG_OK);
3116 if (ret < 0) {
3117 /* Somehow, the session daemon is not responding anymore. */
3118 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3119 goto error_nosignal;
3120 }
3121
3122 /* Poll on consumer socket. */
3123 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
3124 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3125 ret = -EINTR;
3126 goto error_nosignal;
3127 }
3128
3129 /* Get relayd socket from session daemon */
3130 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
3131 if (ret != sizeof(fd)) {
3132 ret = -1;
3133 fd = -1; /* Just in case it gets set with an invalid value. */
3134
3135 /*
3136 * Failing to receive FDs might indicate a major problem such as
3137 * reaching a fd limit during the receive where the kernel returns a
3138 * MSG_CTRUNC and fails to cleanup the fd in the queue. Any case, we
3139 * don't take any chances and stop everything.
3140 *
3141 * XXX: Feature request #558 will fix that and avoid this possible
3142 * issue when reaching the fd limit.
3143 */
3144 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
3145 ret_code = LTTCOMM_CONSUMERD_ERROR_RECV_FD;
3146 goto error;
3147 }
3148
3149 /* Copy socket information and received FD */
3150 switch (sock_type) {
3151 case LTTNG_STREAM_CONTROL:
3152 /* Copy received lttcomm socket */
3153 lttcomm_copy_sock(&relayd->control_sock.sock, &relayd_sock->sock);
3154 ret = lttcomm_create_sock(&relayd->control_sock.sock);
3155 /* Handle create_sock error. */
3156 if (ret < 0) {
3157 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
3158 goto error;
3159 }
3160 /*
3161 * Close the socket created internally by
3162 * lttcomm_create_sock, so we can replace it by the one
3163 * received from sessiond.
3164 */
3165 if (close(relayd->control_sock.sock.fd)) {
3166 PERROR("close");
3167 }
3168
3169 /* Assign new file descriptor */
3170 relayd->control_sock.sock.fd = fd;
3171 fd = -1; /* For error path */
3172 /* Assign version values. */
3173 relayd->control_sock.major = relayd_sock->major;
3174 relayd->control_sock.minor = relayd_sock->minor;
3175
3176 /*
3177 * Create a session on the relayd and store the returned id. Lock the
3178 * control socket mutex if the relayd was NOT created before.
3179 */
3180 if (!relayd_created) {
3181 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3182 }
3183 ret = relayd_create_session(&relayd->control_sock,
3184 &relayd->relayd_session_id);
3185 if (!relayd_created) {
3186 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3187 }
3188 if (ret < 0) {
3189 /*
3190 * Close all sockets of a relayd object. It will be freed if it was
3191 * created at the error code path or else it will be garbage
3192 * collect.
3193 */
3194 (void) relayd_close(&relayd->control_sock);
3195 (void) relayd_close(&relayd->data_sock);
3196 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
3197 goto error;
3198 }
3199
3200 break;
3201 case LTTNG_STREAM_DATA:
3202 /* Copy received lttcomm socket */
3203 lttcomm_copy_sock(&relayd->data_sock.sock, &relayd_sock->sock);
3204 ret = lttcomm_create_sock(&relayd->data_sock.sock);
3205 /* Handle create_sock error. */
3206 if (ret < 0) {
3207 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
3208 goto error;
3209 }
3210 /*
3211 * Close the socket created internally by
3212 * lttcomm_create_sock, so we can replace it by the one
3213 * received from sessiond.
3214 */
3215 if (close(relayd->data_sock.sock.fd)) {
3216 PERROR("close");
3217 }
3218
3219 /* Assign new file descriptor */
3220 relayd->data_sock.sock.fd = fd;
3221 fd = -1; /* for eventual error paths */
3222 /* Assign version values. */
3223 relayd->data_sock.major = relayd_sock->major;
3224 relayd->data_sock.minor = relayd_sock->minor;
3225 break;
3226 default:
3227 ERR("Unknown relayd socket type (%d)", sock_type);
3228 ret = -1;
3229 ret_code = LTTCOMM_CONSUMERD_FATAL;
3230 goto error;
3231 }
3232
3233 DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
3234 sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
3235 relayd->net_seq_idx, fd);
3236
3237 /* We successfully added the socket. Send status back. */
3238 ret = consumer_send_status_msg(sock, ret_code);
3239 if (ret < 0) {
3240 /* Somehow, the session daemon is not responding anymore. */
3241 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3242 goto error_nosignal;
3243 }
3244
3245 /*
3246 * Add relayd socket pair to consumer data hashtable. If object already
3247 * exists or on error, the function gracefully returns.
3248 */
3249 add_relayd(relayd);
3250
3251 /* All good! */
3252 return 0;
3253
3254error:
3255 if (consumer_send_status_msg(sock, ret_code) < 0) {
3256 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3257 }
3258
3259error_nosignal:
3260 /* Close received socket if valid. */
3261 if (fd >= 0) {
3262 if (close(fd)) {
3263 PERROR("close received socket");
3264 }
3265 }
3266
3267 if (relayd_created) {
3268 free(relayd);
3269 }
3270
3271 return ret;
3272}
3273
3274/*
3275 * Try to lock the stream mutex.
3276 *
3277 * On success, 1 is returned else 0 indicating that the mutex is NOT lock.
3278 */
3279static int stream_try_lock(struct lttng_consumer_stream *stream)
3280{
3281 int ret;
3282
3283 assert(stream);
3284
3285 /*
3286 * Try to lock the stream mutex. On failure, we know that the stream is
3287 * being used else where hence there is data still being extracted.
3288 */
3289 ret = pthread_mutex_trylock(&stream->lock);
3290 if (ret) {
3291 /* For both EBUSY and EINVAL error, the mutex is NOT locked. */
3292 ret = 0;
3293 goto end;
3294 }
3295
3296 ret = 1;
3297
3298end:
3299 return ret;
3300}
3301
3302/*
3303 * Search for a relayd associated to the session id and return the reference.
3304 *
3305 * A rcu read side lock MUST be acquire before calling this function and locked
3306 * until the relayd object is no longer necessary.
3307 */
3308static struct consumer_relayd_sock_pair *find_relayd_by_session_id(uint64_t id)
3309{
3310 struct lttng_ht_iter iter;
3311 struct consumer_relayd_sock_pair *relayd = NULL;
3312
3313 /* Iterate over all relayd since they are indexed by net_seq_idx. */
3314 cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
3315 node.node) {
3316 /*
3317 * Check by sessiond id which is unique here where the relayd session
3318 * id might not be when having multiple relayd.
3319 */
3320 if (relayd->sessiond_session_id == id) {
3321 /* Found the relayd. There can be only one per id. */
3322 goto found;
3323 }
3324 }
3325
3326 return NULL;
3327
3328found:
3329 return relayd;
3330}
3331
3332/*
3333 * Check if for a given session id there is still data needed to be extract
3334 * from the buffers.
3335 *
3336 * Return 1 if data is pending or else 0 meaning ready to be read.
3337 */
3338int consumer_data_pending(uint64_t id)
3339{
3340 int ret;
3341 struct lttng_ht_iter iter;
3342 struct lttng_ht *ht;
3343 struct lttng_consumer_stream *stream;
3344 struct consumer_relayd_sock_pair *relayd = NULL;
3345 int (*data_pending)(struct lttng_consumer_stream *);
3346
3347 DBG("Consumer data pending command on session id %" PRIu64, id);
3348
3349 rcu_read_lock();
3350 pthread_mutex_lock(&consumer_data.lock);
3351
3352 switch (consumer_data.type) {
3353 case LTTNG_CONSUMER_KERNEL:
3354 data_pending = lttng_kconsumer_data_pending;
3355 break;
3356 case LTTNG_CONSUMER32_UST:
3357 case LTTNG_CONSUMER64_UST:
3358 data_pending = lttng_ustconsumer_data_pending;
3359 break;
3360 default:
3361 ERR("Unknown consumer data type");
3362 assert(0);
3363 }
3364
3365 /* Ease our life a bit */
3366 ht = consumer_data.stream_list_ht;
3367
3368 relayd = find_relayd_by_session_id(id);
3369 if (relayd) {
3370 /* Send init command for data pending. */
3371 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3372 ret = relayd_begin_data_pending(&relayd->control_sock,
3373 relayd->relayd_session_id);
3374 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3375 if (ret < 0) {
3376 /* Communication error thus the relayd so no data pending. */
3377 goto data_not_pending;
3378 }
3379 }
3380
3381 cds_lfht_for_each_entry_duplicate(ht->ht,
3382 ht->hash_fct(&id, lttng_ht_seed),
3383 ht->match_fct, &id,
3384 &iter.iter, stream, node_session_id.node) {
3385 /* If this call fails, the stream is being used hence data pending. */
3386 ret = stream_try_lock(stream);
3387 if (!ret) {
3388 goto data_pending;
3389 }
3390
3391 /*
3392 * A removed node from the hash table indicates that the stream has
3393 * been deleted thus having a guarantee that the buffers are closed
3394 * on the consumer side. However, data can still be transmitted
3395 * over the network so don't skip the relayd check.
3396 */
3397 ret = cds_lfht_is_node_deleted(&stream->node.node);
3398 if (!ret) {
3399 /* Check the stream if there is data in the buffers. */
3400 ret = data_pending(stream);
3401 if (ret == 1) {
3402 pthread_mutex_unlock(&stream->lock);
3403 goto data_pending;
3404 }
3405 }
3406
3407 /* Relayd check */
3408 if (relayd) {
3409 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3410 if (stream->metadata_flag) {
3411 ret = relayd_quiescent_control(&relayd->control_sock,
3412 stream->relayd_stream_id);
3413 } else {
3414 ret = relayd_data_pending(&relayd->control_sock,
3415 stream->relayd_stream_id,
3416 stream->next_net_seq_num - 1);
3417 }
3418 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3419 if (ret == 1) {
3420 pthread_mutex_unlock(&stream->lock);
3421 goto data_pending;
3422 }
3423 }
3424 pthread_mutex_unlock(&stream->lock);
3425 }
3426
3427 if (relayd) {
3428 unsigned int is_data_inflight = 0;
3429
3430 /* Send init command for data pending. */
3431 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3432 ret = relayd_end_data_pending(&relayd->control_sock,
3433 relayd->relayd_session_id, &is_data_inflight);
3434 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3435 if (ret < 0) {
3436 goto data_not_pending;
3437 }
3438 if (is_data_inflight) {
3439 goto data_pending;
3440 }
3441 }
3442
3443 /*
3444 * Finding _no_ node in the hash table and no inflight data means that the
3445 * stream(s) have been removed thus data is guaranteed to be available for
3446 * analysis from the trace files.
3447 */
3448
3449data_not_pending:
3450 /* Data is available to be read by a viewer. */
3451 pthread_mutex_unlock(&consumer_data.lock);
3452 rcu_read_unlock();
3453 return 0;
3454
3455data_pending:
3456 /* Data is still being extracted from buffers. */
3457 pthread_mutex_unlock(&consumer_data.lock);
3458 rcu_read_unlock();
3459 return 1;
3460}
3461
3462/*
3463 * Send a ret code status message to the sessiond daemon.
3464 *
3465 * Return the sendmsg() return value.
3466 */
3467int consumer_send_status_msg(int sock, int ret_code)
3468{
3469 struct lttcomm_consumer_status_msg msg;
3470
3471 msg.ret_code = ret_code;
3472
3473 return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
3474}
3475
3476/*
3477 * Send a channel status message to the sessiond daemon.
3478 *
3479 * Return the sendmsg() return value.
3480 */
3481int consumer_send_status_channel(int sock,
3482 struct lttng_consumer_channel *channel)
3483{
3484 struct lttcomm_consumer_status_channel msg;
3485
3486 assert(sock >= 0);
3487
3488 if (!channel) {
3489 msg.ret_code = -LTTNG_ERR_UST_CHAN_FAIL;
3490 } else {
3491 msg.ret_code = LTTNG_OK;
3492 msg.key = channel->key;
3493 msg.stream_count = channel->streams.count;
3494 }
3495
3496 return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
3497}
3498
3499/*
3500 * Using a maximum stream size with the produced and consumed position of a
3501 * stream, computes the new consumed position to be as close as possible to the
3502 * maximum possible stream size.
3503 *
3504 * If maximum stream size is lower than the possible buffer size (produced -
3505 * consumed), the consumed_pos given is returned untouched else the new value
3506 * is returned.
3507 */
3508unsigned long consumer_get_consumed_maxsize(unsigned long consumed_pos,
3509 unsigned long produced_pos, uint64_t max_stream_size)
3510{
3511 if (max_stream_size && max_stream_size < (produced_pos - consumed_pos)) {
3512 /* Offset from the produced position to get the latest buffers. */
3513 return produced_pos - max_stream_size;
3514 }
3515
3516 return consumed_pos;
3517}
This page took 0.03491 seconds and 4 git commands to generate.