221c348df055722918793e32617b6812dcfbc680
[lttng-tools.git] / src / common / consumer.c
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
46 struct lttng_consumer_global_data consumer_data = {
47 .stream_count = 0,
48 .need_update = 1,
49 .type = LTTNG_CONSUMER_UNKNOWN,
50 };
51
52 enum consumer_channel_action {
53 CONSUMER_CHANNEL_ADD,
54 CONSUMER_CHANNEL_DEL,
55 CONSUMER_CHANNEL_QUIT,
56 };
57
58 struct 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 */
70 volatile 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 */
77 static struct lttng_ht *metadata_ht;
78 static 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 */
85 static 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
94 static 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
112 void 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
118 static 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 */
141 static 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
168 static 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 */
192 struct 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
212 static 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
222 static 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 */
235 static 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 */
259 void 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 */
285 void 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
296 /* Delete streams that might have been left in the stream list. */
297 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
298 send_node) {
299 cds_list_del(&stream->send_node);
300 /*
301 * Once a stream is added to this list, the buffers were created so
302 * we have a guarantee that this call will succeed.
303 */
304 consumer_stream_destroy(stream, NULL);
305 }
306
307 switch (consumer_data.type) {
308 case LTTNG_CONSUMER_KERNEL:
309 break;
310 case LTTNG_CONSUMER32_UST:
311 case LTTNG_CONSUMER64_UST:
312 lttng_ustconsumer_del_channel(channel);
313 break;
314 default:
315 ERR("Unknown consumer_data type");
316 assert(0);
317 goto end;
318 }
319
320 rcu_read_lock();
321 iter.iter.node = &channel->node.node;
322 ret = lttng_ht_del(consumer_data.channel_ht, &iter);
323 assert(!ret);
324 rcu_read_unlock();
325
326 call_rcu(&channel->node.head, free_channel_rcu);
327 end:
328 pthread_mutex_unlock(&channel->lock);
329 pthread_mutex_unlock(&consumer_data.lock);
330 }
331
332 /*
333 * Iterate over the relayd hash table and destroy each element. Finally,
334 * destroy the whole hash table.
335 */
336 static void cleanup_relayd_ht(void)
337 {
338 struct lttng_ht_iter iter;
339 struct consumer_relayd_sock_pair *relayd;
340
341 rcu_read_lock();
342
343 cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
344 node.node) {
345 consumer_destroy_relayd(relayd);
346 }
347
348 rcu_read_unlock();
349
350 lttng_ht_destroy(consumer_data.relayd_ht);
351 }
352
353 /*
354 * Update the end point status of all streams having the given network sequence
355 * index (relayd index).
356 *
357 * It's atomically set without having the stream mutex locked which is fine
358 * because we handle the write/read race with a pipe wakeup for each thread.
359 */
360 static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
361 enum consumer_endpoint_status status)
362 {
363 struct lttng_ht_iter iter;
364 struct lttng_consumer_stream *stream;
365
366 DBG("Consumer set delete flag on stream by idx %" PRIu64, net_seq_idx);
367
368 rcu_read_lock();
369
370 /* Let's begin with metadata */
371 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
372 if (stream->net_seq_idx == net_seq_idx) {
373 uatomic_set(&stream->endpoint_status, status);
374 DBG("Delete flag set to metadata stream %d", stream->wait_fd);
375 }
376 }
377
378 /* Follow up by the data streams */
379 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
380 if (stream->net_seq_idx == net_seq_idx) {
381 uatomic_set(&stream->endpoint_status, status);
382 DBG("Delete flag set to data stream %d", stream->wait_fd);
383 }
384 }
385 rcu_read_unlock();
386 }
387
388 /*
389 * Cleanup a relayd object by flagging every associated streams for deletion,
390 * destroying the object meaning removing it from the relayd hash table,
391 * closing the sockets and freeing the memory in a RCU call.
392 *
393 * If a local data context is available, notify the threads that the streams'
394 * state have changed.
395 */
396 static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
397 struct lttng_consumer_local_data *ctx)
398 {
399 uint64_t netidx;
400
401 assert(relayd);
402
403 DBG("Cleaning up relayd sockets");
404
405 /* Save the net sequence index before destroying the object */
406 netidx = relayd->net_seq_idx;
407
408 /*
409 * Delete the relayd from the relayd hash table, close the sockets and free
410 * the object in a RCU call.
411 */
412 consumer_destroy_relayd(relayd);
413
414 /* Set inactive endpoint to all streams */
415 update_endpoint_status_by_netidx(netidx, CONSUMER_ENDPOINT_INACTIVE);
416
417 /*
418 * With a local data context, notify the threads that the streams' state
419 * have changed. The write() action on the pipe acts as an "implicit"
420 * memory barrier ordering the updates of the end point status from the
421 * read of this status which happens AFTER receiving this notify.
422 */
423 if (ctx) {
424 notify_thread_lttng_pipe(ctx->consumer_data_pipe);
425 notify_thread_lttng_pipe(ctx->consumer_metadata_pipe);
426 }
427 }
428
429 /*
430 * Flag a relayd socket pair for destruction. Destroy it if the refcount
431 * reaches zero.
432 *
433 * RCU read side lock MUST be aquired before calling this function.
434 */
435 void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
436 {
437 assert(relayd);
438
439 /* Set destroy flag for this object */
440 uatomic_set(&relayd->destroy_flag, 1);
441
442 /* Destroy the relayd if refcount is 0 */
443 if (uatomic_read(&relayd->refcount) == 0) {
444 consumer_destroy_relayd(relayd);
445 }
446 }
447
448 /*
449 * Completly destroy stream from every visiable data structure and the given
450 * hash table if one.
451 *
452 * One this call returns, the stream object is not longer usable nor visible.
453 */
454 void consumer_del_stream(struct lttng_consumer_stream *stream,
455 struct lttng_ht *ht)
456 {
457 consumer_stream_destroy(stream, ht);
458 }
459
460 /*
461 * XXX naming of del vs destroy is all mixed up.
462 */
463 void consumer_del_stream_for_data(struct lttng_consumer_stream *stream)
464 {
465 consumer_stream_destroy(stream, data_ht);
466 }
467
468 void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream)
469 {
470 consumer_stream_destroy(stream, metadata_ht);
471 }
472
473 struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
474 uint64_t stream_key,
475 enum lttng_consumer_stream_state state,
476 const char *channel_name,
477 uid_t uid,
478 gid_t gid,
479 uint64_t relayd_id,
480 uint64_t session_id,
481 int cpu,
482 int *alloc_ret,
483 enum consumer_channel_type type,
484 unsigned int monitor)
485 {
486 int ret;
487 struct lttng_consumer_stream *stream;
488
489 stream = zmalloc(sizeof(*stream));
490 if (stream == NULL) {
491 PERROR("malloc struct lttng_consumer_stream");
492 ret = -ENOMEM;
493 goto end;
494 }
495
496 rcu_read_lock();
497
498 stream->key = stream_key;
499 stream->out_fd = -1;
500 stream->out_fd_offset = 0;
501 stream->state = state;
502 stream->uid = uid;
503 stream->gid = gid;
504 stream->net_seq_idx = relayd_id;
505 stream->session_id = session_id;
506 stream->monitor = monitor;
507 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
508 pthread_mutex_init(&stream->lock, NULL);
509
510 /* If channel is the metadata, flag this stream as metadata. */
511 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
512 stream->metadata_flag = 1;
513 /* Metadata is flat out. */
514 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
515 } else {
516 /* Format stream name to <channel_name>_<cpu_number> */
517 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
518 channel_name, cpu);
519 if (ret < 0) {
520 PERROR("snprintf stream name");
521 goto error;
522 }
523 }
524
525 /* Key is always the wait_fd for streams. */
526 lttng_ht_node_init_u64(&stream->node, stream->key);
527
528 /* Init node per channel id key */
529 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
530
531 /* Init session id node with the stream session id */
532 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
533
534 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
535 " relayd_id %" PRIu64 ", session_id %" PRIu64,
536 stream->name, stream->key, channel_key,
537 stream->net_seq_idx, stream->session_id);
538
539 rcu_read_unlock();
540 return stream;
541
542 error:
543 rcu_read_unlock();
544 free(stream);
545 end:
546 if (alloc_ret) {
547 *alloc_ret = ret;
548 }
549 return NULL;
550 }
551
552 /*
553 * Add a stream to the global list protected by a mutex.
554 */
555 int consumer_add_data_stream(struct lttng_consumer_stream *stream)
556 {
557 struct lttng_ht *ht = data_ht;
558 int ret = 0;
559
560 assert(stream);
561 assert(ht);
562
563 DBG3("Adding consumer stream %" PRIu64, stream->key);
564
565 pthread_mutex_lock(&consumer_data.lock);
566 pthread_mutex_lock(&stream->chan->lock);
567 pthread_mutex_lock(&stream->chan->timer_lock);
568 pthread_mutex_lock(&stream->lock);
569 rcu_read_lock();
570
571 /* Steal stream identifier to avoid having streams with the same key */
572 steal_stream_key(stream->key, ht);
573
574 lttng_ht_add_unique_u64(ht, &stream->node);
575
576 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
577 &stream->node_channel_id);
578
579 /*
580 * Add stream to the stream_list_ht of the consumer data. No need to steal
581 * the key since the HT does not use it and we allow to add redundant keys
582 * into this table.
583 */
584 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
585
586 /*
587 * When nb_init_stream_left reaches 0, we don't need to trigger any action
588 * in terms of destroying the associated channel, because the action that
589 * causes the count to become 0 also causes a stream to be added. The
590 * channel deletion will thus be triggered by the following removal of this
591 * stream.
592 */
593 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
594 /* Increment refcount before decrementing nb_init_stream_left */
595 cmm_smp_wmb();
596 uatomic_dec(&stream->chan->nb_init_stream_left);
597 }
598
599 /* Update consumer data once the node is inserted. */
600 consumer_data.stream_count++;
601 consumer_data.need_update = 1;
602
603 rcu_read_unlock();
604 pthread_mutex_unlock(&stream->lock);
605 pthread_mutex_unlock(&stream->chan->timer_lock);
606 pthread_mutex_unlock(&stream->chan->lock);
607 pthread_mutex_unlock(&consumer_data.lock);
608
609 return ret;
610 }
611
612 void consumer_del_data_stream(struct lttng_consumer_stream *stream)
613 {
614 consumer_del_stream(stream, data_ht);
615 }
616
617 /*
618 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
619 * be acquired before calling this.
620 */
621 static int add_relayd(struct consumer_relayd_sock_pair *relayd)
622 {
623 int ret = 0;
624 struct lttng_ht_node_u64 *node;
625 struct lttng_ht_iter iter;
626
627 assert(relayd);
628
629 lttng_ht_lookup(consumer_data.relayd_ht,
630 &relayd->net_seq_idx, &iter);
631 node = lttng_ht_iter_get_node_u64(&iter);
632 if (node != NULL) {
633 goto end;
634 }
635 lttng_ht_add_unique_u64(consumer_data.relayd_ht, &relayd->node);
636
637 end:
638 return ret;
639 }
640
641 /*
642 * Allocate and return a consumer relayd socket.
643 */
644 struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
645 uint64_t net_seq_idx)
646 {
647 struct consumer_relayd_sock_pair *obj = NULL;
648
649 /* net sequence index of -1 is a failure */
650 if (net_seq_idx == (uint64_t) -1ULL) {
651 goto error;
652 }
653
654 obj = zmalloc(sizeof(struct consumer_relayd_sock_pair));
655 if (obj == NULL) {
656 PERROR("zmalloc relayd sock");
657 goto error;
658 }
659
660 obj->net_seq_idx = net_seq_idx;
661 obj->refcount = 0;
662 obj->destroy_flag = 0;
663 obj->control_sock.sock.fd = -1;
664 obj->data_sock.sock.fd = -1;
665 lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
666 pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
667
668 error:
669 return obj;
670 }
671
672 /*
673 * Find a relayd socket pair in the global consumer data.
674 *
675 * Return the object if found else NULL.
676 * RCU read-side lock must be held across this call and while using the
677 * returned object.
678 */
679 struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
680 {
681 struct lttng_ht_iter iter;
682 struct lttng_ht_node_u64 *node;
683 struct consumer_relayd_sock_pair *relayd = NULL;
684
685 /* Negative keys are lookup failures */
686 if (key == (uint64_t) -1ULL) {
687 goto error;
688 }
689
690 lttng_ht_lookup(consumer_data.relayd_ht, &key,
691 &iter);
692 node = lttng_ht_iter_get_node_u64(&iter);
693 if (node != NULL) {
694 relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
695 }
696
697 error:
698 return relayd;
699 }
700
701 /*
702 * Find a relayd and send the stream
703 *
704 * Returns 0 on success, < 0 on error
705 */
706 int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
707 char *path)
708 {
709 int ret = 0;
710 struct consumer_relayd_sock_pair *relayd;
711
712 assert(stream);
713 assert(stream->net_seq_idx != -1ULL);
714 assert(path);
715
716 /* The stream is not metadata. Get relayd reference if exists. */
717 rcu_read_lock();
718 relayd = consumer_find_relayd(stream->net_seq_idx);
719 if (relayd != NULL) {
720 /* Add stream on the relayd */
721 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
722 ret = relayd_add_stream(&relayd->control_sock, stream->name,
723 path, &stream->relayd_stream_id,
724 stream->chan->tracefile_size, stream->chan->tracefile_count);
725 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
726 if (ret < 0) {
727 goto end;
728 }
729 uatomic_inc(&relayd->refcount);
730 stream->sent_to_relayd = 1;
731 } else {
732 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
733 stream->key, stream->net_seq_idx);
734 ret = -1;
735 goto end;
736 }
737
738 DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
739 stream->name, stream->key, stream->net_seq_idx);
740
741 end:
742 rcu_read_unlock();
743 return ret;
744 }
745
746 /*
747 * Find a relayd and close the stream
748 */
749 void close_relayd_stream(struct lttng_consumer_stream *stream)
750 {
751 struct consumer_relayd_sock_pair *relayd;
752
753 /* The stream is not metadata. Get relayd reference if exists. */
754 rcu_read_lock();
755 relayd = consumer_find_relayd(stream->net_seq_idx);
756 if (relayd) {
757 consumer_stream_relayd_close(stream, relayd);
758 }
759 rcu_read_unlock();
760 }
761
762 /*
763 * Handle stream for relayd transmission if the stream applies for network
764 * streaming where the net sequence index is set.
765 *
766 * Return destination file descriptor or negative value on error.
767 */
768 static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
769 size_t data_size, unsigned long padding,
770 struct consumer_relayd_sock_pair *relayd)
771 {
772 int outfd = -1, ret;
773 struct lttcomm_relayd_data_hdr data_hdr;
774
775 /* Safety net */
776 assert(stream);
777 assert(relayd);
778
779 /* Reset data header */
780 memset(&data_hdr, 0, sizeof(data_hdr));
781
782 if (stream->metadata_flag) {
783 /* Caller MUST acquire the relayd control socket lock */
784 ret = relayd_send_metadata(&relayd->control_sock, data_size);
785 if (ret < 0) {
786 goto error;
787 }
788
789 /* Metadata are always sent on the control socket. */
790 outfd = relayd->control_sock.sock.fd;
791 } else {
792 /* Set header with stream information */
793 data_hdr.stream_id = htobe64(stream->relayd_stream_id);
794 data_hdr.data_size = htobe32(data_size);
795 data_hdr.padding_size = htobe32(padding);
796 /*
797 * Note that net_seq_num below is assigned with the *current* value of
798 * next_net_seq_num and only after that the next_net_seq_num will be
799 * increment. This is why when issuing a command on the relayd using
800 * this next value, 1 should always be substracted in order to compare
801 * the last seen sequence number on the relayd side to the last sent.
802 */
803 data_hdr.net_seq_num = htobe64(stream->next_net_seq_num);
804 /* Other fields are zeroed previously */
805
806 ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
807 sizeof(data_hdr));
808 if (ret < 0) {
809 goto error;
810 }
811
812 ++stream->next_net_seq_num;
813
814 /* Set to go on data socket */
815 outfd = relayd->data_sock.sock.fd;
816 }
817
818 error:
819 return outfd;
820 }
821
822 /*
823 * Allocate and return a new lttng_consumer_channel object using the given key
824 * to initialize the hash table node.
825 *
826 * On error, return NULL.
827 */
828 struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
829 uint64_t session_id,
830 const char *pathname,
831 const char *name,
832 uid_t uid,
833 gid_t gid,
834 uint64_t relayd_id,
835 enum lttng_event_output output,
836 uint64_t tracefile_size,
837 uint64_t tracefile_count,
838 uint64_t session_id_per_pid,
839 unsigned int monitor)
840 {
841 struct lttng_consumer_channel *channel;
842
843 channel = zmalloc(sizeof(*channel));
844 if (channel == NULL) {
845 PERROR("malloc struct lttng_consumer_channel");
846 goto end;
847 }
848
849 channel->key = key;
850 channel->refcount = 0;
851 channel->session_id = session_id;
852 channel->session_id_per_pid = session_id_per_pid;
853 channel->uid = uid;
854 channel->gid = gid;
855 channel->relayd_id = relayd_id;
856 channel->output = output;
857 channel->tracefile_size = tracefile_size;
858 channel->tracefile_count = tracefile_count;
859 channel->monitor = monitor;
860 pthread_mutex_init(&channel->lock, NULL);
861 pthread_mutex_init(&channel->timer_lock, NULL);
862
863 /*
864 * In monitor mode, the streams associated with the channel will be put in
865 * a special list ONLY owned by this channel. So, the refcount is set to 1
866 * here meaning that the channel itself has streams that are referenced.
867 *
868 * On a channel deletion, once the channel is no longer visible, the
869 * refcount is decremented and checked for a zero value to delete it. With
870 * streams in no monitor mode, it will now be safe to destroy the channel.
871 */
872 if (!channel->monitor) {
873 channel->refcount = 1;
874 }
875
876 strncpy(channel->pathname, pathname, sizeof(channel->pathname));
877 channel->pathname[sizeof(channel->pathname) - 1] = '\0';
878
879 strncpy(channel->name, name, sizeof(channel->name));
880 channel->name[sizeof(channel->name) - 1] = '\0';
881
882 lttng_ht_node_init_u64(&channel->node, channel->key);
883
884 channel->wait_fd = -1;
885
886 CDS_INIT_LIST_HEAD(&channel->streams.head);
887
888 DBG("Allocated channel (key %" PRIu64 ")", channel->key)
889
890 end:
891 return channel;
892 }
893
894 /*
895 * Add a channel to the global list protected by a mutex.
896 *
897 * On success 0 is returned else a negative value.
898 */
899 int consumer_add_channel(struct lttng_consumer_channel *channel,
900 struct lttng_consumer_local_data *ctx)
901 {
902 int ret = 0;
903 struct lttng_ht_node_u64 *node;
904 struct lttng_ht_iter iter;
905
906 pthread_mutex_lock(&consumer_data.lock);
907 pthread_mutex_lock(&channel->lock);
908 pthread_mutex_lock(&channel->timer_lock);
909 rcu_read_lock();
910
911 lttng_ht_lookup(consumer_data.channel_ht, &channel->key, &iter);
912 node = lttng_ht_iter_get_node_u64(&iter);
913 if (node != NULL) {
914 /* Channel already exist. Ignore the insertion */
915 ERR("Consumer add channel key %" PRIu64 " already exists!",
916 channel->key);
917 ret = -EEXIST;
918 goto end;
919 }
920
921 lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
922
923 end:
924 rcu_read_unlock();
925 pthread_mutex_unlock(&channel->timer_lock);
926 pthread_mutex_unlock(&channel->lock);
927 pthread_mutex_unlock(&consumer_data.lock);
928
929 if (!ret && channel->wait_fd != -1 &&
930 channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
931 notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
932 }
933 return ret;
934 }
935
936 /*
937 * Allocate the pollfd structure and the local view of the out fds to avoid
938 * doing a lookup in the linked list and concurrency issues when writing is
939 * needed. Called with consumer_data.lock held.
940 *
941 * Returns the number of fds in the structures.
942 */
943 static int update_poll_array(struct lttng_consumer_local_data *ctx,
944 struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
945 struct lttng_ht *ht)
946 {
947 int i = 0;
948 struct lttng_ht_iter iter;
949 struct lttng_consumer_stream *stream;
950
951 assert(ctx);
952 assert(ht);
953 assert(pollfd);
954 assert(local_stream);
955
956 DBG("Updating poll fd array");
957 rcu_read_lock();
958 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
959 /*
960 * Only active streams with an active end point can be added to the
961 * poll set and local stream storage of the thread.
962 *
963 * There is a potential race here for endpoint_status to be updated
964 * just after the check. However, this is OK since the stream(s) will
965 * be deleted once the thread is notified that the end point state has
966 * changed where this function will be called back again.
967 */
968 if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM ||
969 stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
970 continue;
971 }
972 /*
973 * This clobbers way too much the debug output. Uncomment that if you
974 * need it for debugging purposes.
975 *
976 * DBG("Active FD %d", stream->wait_fd);
977 */
978 (*pollfd)[i].fd = stream->wait_fd;
979 (*pollfd)[i].events = POLLIN | POLLPRI;
980 local_stream[i] = stream;
981 i++;
982 }
983 rcu_read_unlock();
984
985 /*
986 * Insert the consumer_data_pipe at the end of the array and don't
987 * increment i so nb_fd is the number of real FD.
988 */
989 (*pollfd)[i].fd = lttng_pipe_get_readfd(ctx->consumer_data_pipe);
990 (*pollfd)[i].events = POLLIN | POLLPRI;
991 return i;
992 }
993
994 /*
995 * Poll on the should_quit pipe and the command socket return -1 on error and
996 * should exit, 0 if data is available on the command socket
997 */
998 int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
999 {
1000 int num_rdy;
1001
1002 restart:
1003 num_rdy = poll(consumer_sockpoll, 2, -1);
1004 if (num_rdy == -1) {
1005 /*
1006 * Restart interrupted system call.
1007 */
1008 if (errno == EINTR) {
1009 goto restart;
1010 }
1011 PERROR("Poll error");
1012 goto exit;
1013 }
1014 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
1015 DBG("consumer_should_quit wake up");
1016 goto exit;
1017 }
1018 return 0;
1019
1020 exit:
1021 return -1;
1022 }
1023
1024 /*
1025 * Set the error socket.
1026 */
1027 void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
1028 int sock)
1029 {
1030 ctx->consumer_error_socket = sock;
1031 }
1032
1033 /*
1034 * Set the command socket path.
1035 */
1036 void lttng_consumer_set_command_sock_path(
1037 struct lttng_consumer_local_data *ctx, char *sock)
1038 {
1039 ctx->consumer_command_sock_path = sock;
1040 }
1041
1042 /*
1043 * Send return code to the session daemon.
1044 * If the socket is not defined, we return 0, it is not a fatal error
1045 */
1046 int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
1047 {
1048 if (ctx->consumer_error_socket > 0) {
1049 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
1050 sizeof(enum lttcomm_sessiond_command));
1051 }
1052
1053 return 0;
1054 }
1055
1056 /*
1057 * Close all the tracefiles and stream fds and MUST be called when all
1058 * instances are destroyed i.e. when all threads were joined and are ended.
1059 */
1060 void lttng_consumer_cleanup(void)
1061 {
1062 struct lttng_ht_iter iter;
1063 struct lttng_consumer_channel *channel;
1064
1065 rcu_read_lock();
1066
1067 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
1068 node.node) {
1069 consumer_del_channel(channel);
1070 }
1071
1072 rcu_read_unlock();
1073
1074 lttng_ht_destroy(consumer_data.channel_ht);
1075
1076 cleanup_relayd_ht();
1077
1078 lttng_ht_destroy(consumer_data.stream_per_chan_id_ht);
1079
1080 /*
1081 * This HT contains streams that are freed by either the metadata thread or
1082 * the data thread so we do *nothing* on the hash table and simply destroy
1083 * it.
1084 */
1085 lttng_ht_destroy(consumer_data.stream_list_ht);
1086 }
1087
1088 /*
1089 * Called from signal handler.
1090 */
1091 void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
1092 {
1093 int ret;
1094 consumer_quit = 1;
1095 do {
1096 ret = write(ctx->consumer_should_quit[1], "4", 1);
1097 } while (ret < 0 && errno == EINTR);
1098 if (ret < 0 || ret != 1) {
1099 PERROR("write consumer quit");
1100 }
1101
1102 DBG("Consumer flag that it should quit");
1103 }
1104
1105 void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
1106 off_t orig_offset)
1107 {
1108 int outfd = stream->out_fd;
1109
1110 /*
1111 * This does a blocking write-and-wait on any page that belongs to the
1112 * subbuffer prior to the one we just wrote.
1113 * Don't care about error values, as these are just hints and ways to
1114 * limit the amount of page cache used.
1115 */
1116 if (orig_offset < stream->max_sb_size) {
1117 return;
1118 }
1119 lttng_sync_file_range(outfd, orig_offset - stream->max_sb_size,
1120 stream->max_sb_size,
1121 SYNC_FILE_RANGE_WAIT_BEFORE
1122 | SYNC_FILE_RANGE_WRITE
1123 | SYNC_FILE_RANGE_WAIT_AFTER);
1124 /*
1125 * Give hints to the kernel about how we access the file:
1126 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
1127 * we write it.
1128 *
1129 * We need to call fadvise again after the file grows because the
1130 * kernel does not seem to apply fadvise to non-existing parts of the
1131 * file.
1132 *
1133 * Call fadvise _after_ having waited for the page writeback to
1134 * complete because the dirty page writeback semantic is not well
1135 * defined. So it can be expected to lead to lower throughput in
1136 * streaming.
1137 */
1138 posix_fadvise(outfd, orig_offset - stream->max_sb_size,
1139 stream->max_sb_size, POSIX_FADV_DONTNEED);
1140 }
1141
1142 /*
1143 * Initialise the necessary environnement :
1144 * - create a new context
1145 * - create the poll_pipe
1146 * - create the should_quit pipe (for signal handler)
1147 * - create the thread pipe (for splice)
1148 *
1149 * Takes a function pointer as argument, this function is called when data is
1150 * available on a buffer. This function is responsible to do the
1151 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
1152 * buffer configuration and then kernctl_put_next_subbuf at the end.
1153 *
1154 * Returns a pointer to the new context or NULL on error.
1155 */
1156 struct lttng_consumer_local_data *lttng_consumer_create(
1157 enum lttng_consumer_type type,
1158 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
1159 struct lttng_consumer_local_data *ctx),
1160 int (*recv_channel)(struct lttng_consumer_channel *channel),
1161 int (*recv_stream)(struct lttng_consumer_stream *stream),
1162 int (*update_stream)(uint64_t stream_key, uint32_t state))
1163 {
1164 int ret;
1165 struct lttng_consumer_local_data *ctx;
1166
1167 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
1168 consumer_data.type == type);
1169 consumer_data.type = type;
1170
1171 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
1172 if (ctx == NULL) {
1173 PERROR("allocating context");
1174 goto error;
1175 }
1176
1177 ctx->consumer_error_socket = -1;
1178 ctx->consumer_metadata_socket = -1;
1179 pthread_mutex_init(&ctx->metadata_socket_lock, NULL);
1180 /* assign the callbacks */
1181 ctx->on_buffer_ready = buffer_ready;
1182 ctx->on_recv_channel = recv_channel;
1183 ctx->on_recv_stream = recv_stream;
1184 ctx->on_update_stream = update_stream;
1185
1186 ctx->consumer_data_pipe = lttng_pipe_open(0);
1187 if (!ctx->consumer_data_pipe) {
1188 goto error_poll_pipe;
1189 }
1190
1191 ret = pipe(ctx->consumer_should_quit);
1192 if (ret < 0) {
1193 PERROR("Error creating recv pipe");
1194 goto error_quit_pipe;
1195 }
1196
1197 ret = pipe(ctx->consumer_thread_pipe);
1198 if (ret < 0) {
1199 PERROR("Error creating thread pipe");
1200 goto error_thread_pipe;
1201 }
1202
1203 ret = pipe(ctx->consumer_channel_pipe);
1204 if (ret < 0) {
1205 PERROR("Error creating channel pipe");
1206 goto error_channel_pipe;
1207 }
1208
1209 ctx->consumer_metadata_pipe = lttng_pipe_open(0);
1210 if (!ctx->consumer_metadata_pipe) {
1211 goto error_metadata_pipe;
1212 }
1213
1214 ret = utils_create_pipe(ctx->consumer_splice_metadata_pipe);
1215 if (ret < 0) {
1216 goto error_splice_pipe;
1217 }
1218
1219 return ctx;
1220
1221 error_splice_pipe:
1222 lttng_pipe_destroy(ctx->consumer_metadata_pipe);
1223 error_metadata_pipe:
1224 utils_close_pipe(ctx->consumer_channel_pipe);
1225 error_channel_pipe:
1226 utils_close_pipe(ctx->consumer_thread_pipe);
1227 error_thread_pipe:
1228 utils_close_pipe(ctx->consumer_should_quit);
1229 error_quit_pipe:
1230 lttng_pipe_destroy(ctx->consumer_data_pipe);
1231 error_poll_pipe:
1232 free(ctx);
1233 error:
1234 return NULL;
1235 }
1236
1237 /*
1238 * Close all fds associated with the instance and free the context.
1239 */
1240 void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
1241 {
1242 int ret;
1243
1244 DBG("Consumer destroying it. Closing everything.");
1245
1246 ret = close(ctx->consumer_error_socket);
1247 if (ret) {
1248 PERROR("close");
1249 }
1250 ret = close(ctx->consumer_metadata_socket);
1251 if (ret) {
1252 PERROR("close");
1253 }
1254 utils_close_pipe(ctx->consumer_thread_pipe);
1255 utils_close_pipe(ctx->consumer_channel_pipe);
1256 lttng_pipe_destroy(ctx->consumer_data_pipe);
1257 lttng_pipe_destroy(ctx->consumer_metadata_pipe);
1258 utils_close_pipe(ctx->consumer_should_quit);
1259 utils_close_pipe(ctx->consumer_splice_metadata_pipe);
1260
1261 unlink(ctx->consumer_command_sock_path);
1262 free(ctx);
1263 }
1264
1265 /*
1266 * Write the metadata stream id on the specified file descriptor.
1267 */
1268 static int write_relayd_metadata_id(int fd,
1269 struct lttng_consumer_stream *stream,
1270 struct consumer_relayd_sock_pair *relayd, unsigned long padding)
1271 {
1272 int ret;
1273 struct lttcomm_relayd_metadata_payload hdr;
1274
1275 hdr.stream_id = htobe64(stream->relayd_stream_id);
1276 hdr.padding_size = htobe32(padding);
1277 do {
1278 ret = write(fd, (void *) &hdr, sizeof(hdr));
1279 } while (ret < 0 && errno == EINTR);
1280 if (ret < 0 || ret != sizeof(hdr)) {
1281 /*
1282 * This error means that the fd's end is closed so ignore the perror
1283 * not to clubber the error output since this can happen in a normal
1284 * code path.
1285 */
1286 if (errno != EPIPE) {
1287 PERROR("write metadata stream id");
1288 }
1289 DBG3("Consumer failed to write relayd metadata id (errno: %d)", errno);
1290 /*
1291 * Set ret to a negative value because if ret != sizeof(hdr), we don't
1292 * handle writting the missing part so report that as an error and
1293 * don't lie to the caller.
1294 */
1295 ret = -1;
1296 goto end;
1297 }
1298 DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
1299 stream->relayd_stream_id, padding);
1300
1301 end:
1302 return ret;
1303 }
1304
1305 /*
1306 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1307 * core function for writing trace buffers to either the local filesystem or
1308 * the network.
1309 *
1310 * It must be called with the stream lock held.
1311 *
1312 * Careful review MUST be put if any changes occur!
1313 *
1314 * Returns the number of bytes written
1315 */
1316 ssize_t lttng_consumer_on_read_subbuffer_mmap(
1317 struct lttng_consumer_local_data *ctx,
1318 struct lttng_consumer_stream *stream, unsigned long len,
1319 unsigned long padding)
1320 {
1321 unsigned long mmap_offset;
1322 void *mmap_base;
1323 ssize_t ret = 0, written = 0;
1324 off_t orig_offset = stream->out_fd_offset;
1325 /* Default is on the disk */
1326 int outfd = stream->out_fd;
1327 struct consumer_relayd_sock_pair *relayd = NULL;
1328 unsigned int relayd_hang_up = 0;
1329
1330 /* RCU lock for the relayd pointer */
1331 rcu_read_lock();
1332
1333 /* Flag that the current stream if set for network streaming. */
1334 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1335 relayd = consumer_find_relayd(stream->net_seq_idx);
1336 if (relayd == NULL) {
1337 ret = -EPIPE;
1338 goto end;
1339 }
1340 }
1341
1342 /* get the offset inside the fd to mmap */
1343 switch (consumer_data.type) {
1344 case LTTNG_CONSUMER_KERNEL:
1345 mmap_base = stream->mmap_base;
1346 ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
1347 if (ret != 0) {
1348 PERROR("tracer ctl get_mmap_read_offset");
1349 written = -errno;
1350 goto end;
1351 }
1352 break;
1353 case LTTNG_CONSUMER32_UST:
1354 case LTTNG_CONSUMER64_UST:
1355 mmap_base = lttng_ustctl_get_mmap_base(stream);
1356 if (!mmap_base) {
1357 ERR("read mmap get mmap base for stream %s", stream->name);
1358 written = -EPERM;
1359 goto end;
1360 }
1361 ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
1362 if (ret != 0) {
1363 PERROR("tracer ctl get_mmap_read_offset");
1364 written = ret;
1365 goto end;
1366 }
1367 break;
1368 default:
1369 ERR("Unknown consumer_data type");
1370 assert(0);
1371 }
1372
1373 /* Handle stream on the relayd if the output is on the network */
1374 if (relayd) {
1375 unsigned long netlen = len;
1376
1377 /*
1378 * Lock the control socket for the complete duration of the function
1379 * since from this point on we will use the socket.
1380 */
1381 if (stream->metadata_flag) {
1382 /* Metadata requires the control socket. */
1383 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1384 netlen += sizeof(struct lttcomm_relayd_metadata_payload);
1385 }
1386
1387 ret = write_relayd_stream_header(stream, netlen, padding, relayd);
1388 if (ret >= 0) {
1389 /* Use the returned socket. */
1390 outfd = ret;
1391
1392 /* Write metadata stream id before payload */
1393 if (stream->metadata_flag) {
1394 ret = write_relayd_metadata_id(outfd, stream, relayd, padding);
1395 if (ret < 0) {
1396 written = ret;
1397 /* Socket operation failed. We consider the relayd dead */
1398 if (ret == -EPIPE || ret == -EINVAL) {
1399 relayd_hang_up = 1;
1400 goto write_error;
1401 }
1402 goto end;
1403 }
1404 }
1405 } else {
1406 /* Socket operation failed. We consider the relayd dead */
1407 if (ret == -EPIPE || ret == -EINVAL) {
1408 relayd_hang_up = 1;
1409 goto write_error;
1410 }
1411 /* Else, use the default set before which is the filesystem. */
1412 }
1413 } else {
1414 /* No streaming, we have to set the len with the full padding */
1415 len += padding;
1416
1417 /*
1418 * Check if we need to change the tracefile before writing the packet.
1419 */
1420 if (stream->chan->tracefile_size > 0 &&
1421 (stream->tracefile_size_current + len) >
1422 stream->chan->tracefile_size) {
1423 ret = utils_rotate_stream_file(stream->chan->pathname,
1424 stream->name, stream->chan->tracefile_size,
1425 stream->chan->tracefile_count, stream->uid, stream->gid,
1426 stream->out_fd, &(stream->tracefile_count_current));
1427 if (ret < 0) {
1428 ERR("Rotating output file");
1429 goto end;
1430 }
1431 outfd = stream->out_fd = ret;
1432 /* Reset current size because we just perform a rotation. */
1433 stream->tracefile_size_current = 0;
1434 }
1435 stream->tracefile_size_current += len;
1436 }
1437
1438 while (len > 0) {
1439 do {
1440 ret = write(outfd, mmap_base + mmap_offset, len);
1441 } while (ret < 0 && errno == EINTR);
1442 DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
1443 if (ret < 0) {
1444 /*
1445 * This is possible if the fd is closed on the other side (outfd)
1446 * or any write problem. It can be verbose a bit for a normal
1447 * execution if for instance the relayd is stopped abruptly. This
1448 * can happen so set this to a DBG statement.
1449 */
1450 DBG("Error in file write mmap");
1451 if (written == 0) {
1452 written = -errno;
1453 }
1454 /* Socket operation failed. We consider the relayd dead */
1455 if (errno == EPIPE || errno == EINVAL) {
1456 relayd_hang_up = 1;
1457 goto write_error;
1458 }
1459 goto end;
1460 } else if (ret > len) {
1461 PERROR("Error in file write (ret %zd > len %lu)", ret, len);
1462 written += ret;
1463 goto end;
1464 } else {
1465 len -= ret;
1466 mmap_offset += ret;
1467 }
1468
1469 /* This call is useless on a socket so better save a syscall. */
1470 if (!relayd) {
1471 /* This won't block, but will start writeout asynchronously */
1472 lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
1473 SYNC_FILE_RANGE_WRITE);
1474 stream->out_fd_offset += ret;
1475 }
1476 written += ret;
1477 }
1478 lttng_consumer_sync_trace_file(stream, orig_offset);
1479
1480 write_error:
1481 /*
1482 * This is a special case that the relayd has closed its socket. Let's
1483 * cleanup the relayd object and all associated streams.
1484 */
1485 if (relayd && relayd_hang_up) {
1486 cleanup_relayd(relayd, ctx);
1487 }
1488
1489 end:
1490 /* Unlock only if ctrl socket used */
1491 if (relayd && stream->metadata_flag) {
1492 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1493 }
1494
1495 rcu_read_unlock();
1496 return written;
1497 }
1498
1499 /*
1500 * Splice the data from the ring buffer to the tracefile.
1501 *
1502 * It must be called with the stream lock held.
1503 *
1504 * Returns the number of bytes spliced.
1505 */
1506 ssize_t lttng_consumer_on_read_subbuffer_splice(
1507 struct lttng_consumer_local_data *ctx,
1508 struct lttng_consumer_stream *stream, unsigned long len,
1509 unsigned long padding)
1510 {
1511 ssize_t ret = 0, written = 0, ret_splice = 0;
1512 loff_t offset = 0;
1513 off_t orig_offset = stream->out_fd_offset;
1514 int fd = stream->wait_fd;
1515 /* Default is on the disk */
1516 int outfd = stream->out_fd;
1517 struct consumer_relayd_sock_pair *relayd = NULL;
1518 int *splice_pipe;
1519 unsigned int relayd_hang_up = 0;
1520
1521 switch (consumer_data.type) {
1522 case LTTNG_CONSUMER_KERNEL:
1523 break;
1524 case LTTNG_CONSUMER32_UST:
1525 case LTTNG_CONSUMER64_UST:
1526 /* Not supported for user space tracing */
1527 return -ENOSYS;
1528 default:
1529 ERR("Unknown consumer_data type");
1530 assert(0);
1531 }
1532
1533 /* RCU lock for the relayd pointer */
1534 rcu_read_lock();
1535
1536 /* Flag that the current stream if set for network streaming. */
1537 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1538 relayd = consumer_find_relayd(stream->net_seq_idx);
1539 if (relayd == NULL) {
1540 ret = -EPIPE;
1541 goto end;
1542 }
1543 }
1544
1545 /*
1546 * Choose right pipe for splice. Metadata and trace data are handled by
1547 * different threads hence the use of two pipes in order not to race or
1548 * corrupt the written data.
1549 */
1550 if (stream->metadata_flag) {
1551 splice_pipe = ctx->consumer_splice_metadata_pipe;
1552 } else {
1553 splice_pipe = ctx->consumer_thread_pipe;
1554 }
1555
1556 /* Write metadata stream id before payload */
1557 if (relayd) {
1558 int total_len = len;
1559
1560 if (stream->metadata_flag) {
1561 /*
1562 * Lock the control socket for the complete duration of the function
1563 * since from this point on we will use the socket.
1564 */
1565 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1566
1567 ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
1568 padding);
1569 if (ret < 0) {
1570 written = ret;
1571 /* Socket operation failed. We consider the relayd dead */
1572 if (ret == -EBADF) {
1573 WARN("Remote relayd disconnected. Stopping");
1574 relayd_hang_up = 1;
1575 goto write_error;
1576 }
1577 goto end;
1578 }
1579
1580 total_len += sizeof(struct lttcomm_relayd_metadata_payload);
1581 }
1582
1583 ret = write_relayd_stream_header(stream, total_len, padding, relayd);
1584 if (ret >= 0) {
1585 /* Use the returned socket. */
1586 outfd = ret;
1587 } else {
1588 /* Socket operation failed. We consider the relayd dead */
1589 if (ret == -EBADF) {
1590 WARN("Remote relayd disconnected. Stopping");
1591 relayd_hang_up = 1;
1592 goto write_error;
1593 }
1594 goto end;
1595 }
1596 } else {
1597 /* No streaming, we have to set the len with the full padding */
1598 len += padding;
1599
1600 /*
1601 * Check if we need to change the tracefile before writing the packet.
1602 */
1603 if (stream->chan->tracefile_size > 0 &&
1604 (stream->tracefile_size_current + len) >
1605 stream->chan->tracefile_size) {
1606 ret = utils_rotate_stream_file(stream->chan->pathname,
1607 stream->name, stream->chan->tracefile_size,
1608 stream->chan->tracefile_count, stream->uid, stream->gid,
1609 stream->out_fd, &(stream->tracefile_count_current));
1610 if (ret < 0) {
1611 ERR("Rotating output file");
1612 goto end;
1613 }
1614 outfd = stream->out_fd = ret;
1615 /* Reset current size because we just perform a rotation. */
1616 stream->tracefile_size_current = 0;
1617 }
1618 stream->tracefile_size_current += len;
1619 }
1620
1621 while (len > 0) {
1622 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
1623 (unsigned long)offset, len, fd, splice_pipe[1]);
1624 ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
1625 SPLICE_F_MOVE | SPLICE_F_MORE);
1626 DBG("splice chan to pipe, ret %zd", ret_splice);
1627 if (ret_splice < 0) {
1628 PERROR("Error in relay splice");
1629 if (written == 0) {
1630 written = ret_splice;
1631 }
1632 ret = errno;
1633 goto splice_error;
1634 }
1635
1636 /* Handle stream on the relayd if the output is on the network */
1637 if (relayd) {
1638 if (stream->metadata_flag) {
1639 size_t metadata_payload_size =
1640 sizeof(struct lttcomm_relayd_metadata_payload);
1641
1642 /* Update counter to fit the spliced data */
1643 ret_splice += metadata_payload_size;
1644 len += metadata_payload_size;
1645 /*
1646 * We do this so the return value can match the len passed as
1647 * argument to this function.
1648 */
1649 written -= metadata_payload_size;
1650 }
1651 }
1652
1653 /* Splice data out */
1654 ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
1655 ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
1656 DBG("Consumer splice pipe to file, ret %zd", ret_splice);
1657 if (ret_splice < 0) {
1658 PERROR("Error in file splice");
1659 if (written == 0) {
1660 written = ret_splice;
1661 }
1662 /* Socket operation failed. We consider the relayd dead */
1663 if (errno == EBADF || errno == EPIPE) {
1664 WARN("Remote relayd disconnected. Stopping");
1665 relayd_hang_up = 1;
1666 goto write_error;
1667 }
1668 ret = errno;
1669 goto splice_error;
1670 } else if (ret_splice > len) {
1671 errno = EINVAL;
1672 PERROR("Wrote more data than requested %zd (len: %lu)",
1673 ret_splice, len);
1674 written += ret_splice;
1675 ret = errno;
1676 goto splice_error;
1677 }
1678 len -= ret_splice;
1679
1680 /* This call is useless on a socket so better save a syscall. */
1681 if (!relayd) {
1682 /* This won't block, but will start writeout asynchronously */
1683 lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
1684 SYNC_FILE_RANGE_WRITE);
1685 stream->out_fd_offset += ret_splice;
1686 }
1687 written += ret_splice;
1688 }
1689 lttng_consumer_sync_trace_file(stream, orig_offset);
1690
1691 ret = ret_splice;
1692
1693 goto end;
1694
1695 write_error:
1696 /*
1697 * This is a special case that the relayd has closed its socket. Let's
1698 * cleanup the relayd object and all associated streams.
1699 */
1700 if (relayd && relayd_hang_up) {
1701 cleanup_relayd(relayd, ctx);
1702 /* Skip splice error so the consumer does not fail */
1703 goto end;
1704 }
1705
1706 splice_error:
1707 /* send the appropriate error description to sessiond */
1708 switch (ret) {
1709 case EINVAL:
1710 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
1711 break;
1712 case ENOMEM:
1713 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
1714 break;
1715 case ESPIPE:
1716 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
1717 break;
1718 }
1719
1720 end:
1721 if (relayd && stream->metadata_flag) {
1722 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1723 }
1724
1725 rcu_read_unlock();
1726 return written;
1727 }
1728
1729 /*
1730 * Take a snapshot for a specific fd
1731 *
1732 * Returns 0 on success, < 0 on error
1733 */
1734 int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
1735 {
1736 switch (consumer_data.type) {
1737 case LTTNG_CONSUMER_KERNEL:
1738 return lttng_kconsumer_take_snapshot(stream);
1739 case LTTNG_CONSUMER32_UST:
1740 case LTTNG_CONSUMER64_UST:
1741 return lttng_ustconsumer_take_snapshot(stream);
1742 default:
1743 ERR("Unknown consumer_data type");
1744 assert(0);
1745 return -ENOSYS;
1746 }
1747 }
1748
1749 /*
1750 * Get the produced position
1751 *
1752 * Returns 0 on success, < 0 on error
1753 */
1754 int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
1755 unsigned long *pos)
1756 {
1757 switch (consumer_data.type) {
1758 case LTTNG_CONSUMER_KERNEL:
1759 return lttng_kconsumer_get_produced_snapshot(stream, pos);
1760 case LTTNG_CONSUMER32_UST:
1761 case LTTNG_CONSUMER64_UST:
1762 return lttng_ustconsumer_get_produced_snapshot(stream, pos);
1763 default:
1764 ERR("Unknown consumer_data type");
1765 assert(0);
1766 return -ENOSYS;
1767 }
1768 }
1769
1770 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1771 int sock, struct pollfd *consumer_sockpoll)
1772 {
1773 switch (consumer_data.type) {
1774 case LTTNG_CONSUMER_KERNEL:
1775 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
1776 case LTTNG_CONSUMER32_UST:
1777 case LTTNG_CONSUMER64_UST:
1778 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
1779 default:
1780 ERR("Unknown consumer_data type");
1781 assert(0);
1782 return -ENOSYS;
1783 }
1784 }
1785
1786 /*
1787 * Iterate over all streams of the hashtable and free them properly.
1788 *
1789 * WARNING: *MUST* be used with data stream only.
1790 */
1791 static void destroy_data_stream_ht(struct lttng_ht *ht)
1792 {
1793 struct lttng_ht_iter iter;
1794 struct lttng_consumer_stream *stream;
1795
1796 if (ht == NULL) {
1797 return;
1798 }
1799
1800 rcu_read_lock();
1801 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1802 /*
1803 * Ignore return value since we are currently cleaning up so any error
1804 * can't be handled.
1805 */
1806 (void) consumer_del_stream(stream, ht);
1807 }
1808 rcu_read_unlock();
1809
1810 lttng_ht_destroy(ht);
1811 }
1812
1813 /*
1814 * Iterate over all streams of the hashtable and free them properly.
1815 *
1816 * XXX: Should not be only for metadata stream or else use an other name.
1817 */
1818 static void destroy_stream_ht(struct lttng_ht *ht)
1819 {
1820 struct lttng_ht_iter iter;
1821 struct lttng_consumer_stream *stream;
1822
1823 if (ht == NULL) {
1824 return;
1825 }
1826
1827 rcu_read_lock();
1828 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1829 /*
1830 * Ignore return value since we are currently cleaning up so any error
1831 * can't be handled.
1832 */
1833 (void) consumer_del_metadata_stream(stream, ht);
1834 }
1835 rcu_read_unlock();
1836
1837 lttng_ht_destroy(ht);
1838 }
1839
1840 void lttng_consumer_close_metadata(void)
1841 {
1842 switch (consumer_data.type) {
1843 case LTTNG_CONSUMER_KERNEL:
1844 /*
1845 * The Kernel consumer has a different metadata scheme so we don't
1846 * close anything because the stream will be closed by the session
1847 * daemon.
1848 */
1849 break;
1850 case LTTNG_CONSUMER32_UST:
1851 case LTTNG_CONSUMER64_UST:
1852 /*
1853 * Close all metadata streams. The metadata hash table is passed and
1854 * this call iterates over it by closing all wakeup fd. This is safe
1855 * because at this point we are sure that the metadata producer is
1856 * either dead or blocked.
1857 */
1858 lttng_ustconsumer_close_metadata(metadata_ht);
1859 break;
1860 default:
1861 ERR("Unknown consumer_data type");
1862 assert(0);
1863 }
1864 }
1865
1866 /*
1867 * Clean up a metadata stream and free its memory.
1868 */
1869 void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
1870 struct lttng_ht *ht)
1871 {
1872 int ret;
1873 struct lttng_ht_iter iter;
1874 struct lttng_consumer_channel *free_chan = NULL;
1875 struct consumer_relayd_sock_pair *relayd;
1876
1877 assert(stream);
1878 /*
1879 * This call should NEVER receive regular stream. It must always be
1880 * metadata stream and this is crucial for data structure synchronization.
1881 */
1882 assert(stream->metadata_flag);
1883
1884 DBG3("Consumer delete metadata stream %d", stream->wait_fd);
1885
1886 if (ht == NULL) {
1887 /* Means the stream was allocated but not successfully added */
1888 goto free_stream_rcu;
1889 }
1890
1891 pthread_mutex_lock(&consumer_data.lock);
1892 pthread_mutex_lock(&stream->chan->lock);
1893 pthread_mutex_lock(&stream->lock);
1894
1895 switch (consumer_data.type) {
1896 case LTTNG_CONSUMER_KERNEL:
1897 if (stream->mmap_base != NULL) {
1898 ret = munmap(stream->mmap_base, stream->mmap_len);
1899 if (ret != 0) {
1900 PERROR("munmap metadata stream");
1901 }
1902 }
1903 if (stream->wait_fd >= 0) {
1904 ret = close(stream->wait_fd);
1905 if (ret < 0) {
1906 PERROR("close kernel metadata wait_fd");
1907 }
1908 }
1909 break;
1910 case LTTNG_CONSUMER32_UST:
1911 case LTTNG_CONSUMER64_UST:
1912 if (stream->monitor) {
1913 /* close the write-side in close_metadata */
1914 ret = close(stream->ust_metadata_poll_pipe[0]);
1915 if (ret < 0) {
1916 PERROR("Close UST metadata read-side poll pipe");
1917 }
1918 }
1919 lttng_ustconsumer_del_stream(stream);
1920 break;
1921 default:
1922 ERR("Unknown consumer_data type");
1923 assert(0);
1924 goto end;
1925 }
1926
1927 rcu_read_lock();
1928 iter.iter.node = &stream->node.node;
1929 ret = lttng_ht_del(ht, &iter);
1930 assert(!ret);
1931
1932 iter.iter.node = &stream->node_channel_id.node;
1933 ret = lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
1934 assert(!ret);
1935
1936 iter.iter.node = &stream->node_session_id.node;
1937 ret = lttng_ht_del(consumer_data.stream_list_ht, &iter);
1938 assert(!ret);
1939 rcu_read_unlock();
1940
1941 if (stream->out_fd >= 0) {
1942 ret = close(stream->out_fd);
1943 if (ret) {
1944 PERROR("close");
1945 }
1946 }
1947
1948 /* Check and cleanup relayd */
1949 rcu_read_lock();
1950 relayd = consumer_find_relayd(stream->net_seq_idx);
1951 if (relayd != NULL) {
1952 uatomic_dec(&relayd->refcount);
1953 assert(uatomic_read(&relayd->refcount) >= 0);
1954
1955 /* Closing streams requires to lock the control socket. */
1956 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1957 ret = relayd_send_close_stream(&relayd->control_sock,
1958 stream->relayd_stream_id, stream->next_net_seq_num - 1);
1959 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1960 if (ret < 0) {
1961 DBG("Unable to close stream on the relayd. Continuing");
1962 /*
1963 * Continue here. There is nothing we can do for the relayd.
1964 * Chances are that the relayd has closed the socket so we just
1965 * continue cleaning up.
1966 */
1967 }
1968
1969 /* Both conditions are met, we destroy the relayd. */
1970 if (uatomic_read(&relayd->refcount) == 0 &&
1971 uatomic_read(&relayd->destroy_flag)) {
1972 consumer_destroy_relayd(relayd);
1973 }
1974 }
1975 rcu_read_unlock();
1976
1977 /* Atomically decrement channel refcount since other threads can use it. */
1978 if (!uatomic_sub_return(&stream->chan->refcount, 1)
1979 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
1980 /* Go for channel deletion! */
1981 free_chan = stream->chan;
1982 }
1983
1984 end:
1985 /*
1986 * Nullify the stream reference so it is not used after deletion. The
1987 * channel lock MUST be acquired before being able to check for
1988 * a NULL pointer value.
1989 */
1990 stream->chan->metadata_stream = NULL;
1991
1992 pthread_mutex_unlock(&stream->lock);
1993 pthread_mutex_unlock(&stream->chan->lock);
1994 pthread_mutex_unlock(&consumer_data.lock);
1995
1996 if (free_chan) {
1997 consumer_del_channel(free_chan);
1998 }
1999
2000 free_stream_rcu:
2001 call_rcu(&stream->node.head, free_stream_rcu);
2002 }
2003
2004 /*
2005 * Action done with the metadata stream when adding it to the consumer internal
2006 * data structures to handle it.
2007 */
2008 int consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
2009 {
2010 struct lttng_ht *ht = metadata_ht;
2011 int ret = 0;
2012 struct lttng_ht_iter iter;
2013 struct lttng_ht_node_u64 *node;
2014
2015 assert(stream);
2016 assert(ht);
2017
2018 DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
2019
2020 pthread_mutex_lock(&consumer_data.lock);
2021 pthread_mutex_lock(&stream->chan->lock);
2022 pthread_mutex_lock(&stream->chan->timer_lock);
2023 pthread_mutex_lock(&stream->lock);
2024
2025 /*
2026 * From here, refcounts are updated so be _careful_ when returning an error
2027 * after this point.
2028 */
2029
2030 rcu_read_lock();
2031
2032 /*
2033 * Lookup the stream just to make sure it does not exist in our internal
2034 * state. This should NEVER happen.
2035 */
2036 lttng_ht_lookup(ht, &stream->key, &iter);
2037 node = lttng_ht_iter_get_node_u64(&iter);
2038 assert(!node);
2039
2040 /*
2041 * When nb_init_stream_left reaches 0, we don't need to trigger any action
2042 * in terms of destroying the associated channel, because the action that
2043 * causes the count to become 0 also causes a stream to be added. The
2044 * channel deletion will thus be triggered by the following removal of this
2045 * stream.
2046 */
2047 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
2048 /* Increment refcount before decrementing nb_init_stream_left */
2049 cmm_smp_wmb();
2050 uatomic_dec(&stream->chan->nb_init_stream_left);
2051 }
2052
2053 lttng_ht_add_unique_u64(ht, &stream->node);
2054
2055 lttng_ht_add_unique_u64(consumer_data.stream_per_chan_id_ht,
2056 &stream->node_channel_id);
2057
2058 /*
2059 * Add stream to the stream_list_ht of the consumer data. No need to steal
2060 * the key since the HT does not use it and we allow to add redundant keys
2061 * into this table.
2062 */
2063 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
2064
2065 rcu_read_unlock();
2066
2067 pthread_mutex_unlock(&stream->lock);
2068 pthread_mutex_unlock(&stream->chan->lock);
2069 pthread_mutex_unlock(&stream->chan->timer_lock);
2070 pthread_mutex_unlock(&consumer_data.lock);
2071 return ret;
2072 }
2073
2074 /*
2075 * Delete data stream that are flagged for deletion (endpoint_status).
2076 */
2077 static void validate_endpoint_status_data_stream(void)
2078 {
2079 struct lttng_ht_iter iter;
2080 struct lttng_consumer_stream *stream;
2081
2082 DBG("Consumer delete flagged data stream");
2083
2084 rcu_read_lock();
2085 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
2086 /* Validate delete flag of the stream */
2087 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
2088 continue;
2089 }
2090 /* Delete it right now */
2091 consumer_del_stream(stream, data_ht);
2092 }
2093 rcu_read_unlock();
2094 }
2095
2096 /*
2097 * Delete metadata stream that are flagged for deletion (endpoint_status).
2098 */
2099 static void validate_endpoint_status_metadata_stream(
2100 struct lttng_poll_event *pollset)
2101 {
2102 struct lttng_ht_iter iter;
2103 struct lttng_consumer_stream *stream;
2104
2105 DBG("Consumer delete flagged metadata stream");
2106
2107 assert(pollset);
2108
2109 rcu_read_lock();
2110 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
2111 /* Validate delete flag of the stream */
2112 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
2113 continue;
2114 }
2115 /*
2116 * Remove from pollset so the metadata thread can continue without
2117 * blocking on a deleted stream.
2118 */
2119 lttng_poll_del(pollset, stream->wait_fd);
2120
2121 /* Delete it right now */
2122 consumer_del_metadata_stream(stream, metadata_ht);
2123 }
2124 rcu_read_unlock();
2125 }
2126
2127 /*
2128 * Thread polls on metadata file descriptor and write them on disk or on the
2129 * network.
2130 */
2131 void *consumer_thread_metadata_poll(void *data)
2132 {
2133 int ret, i, pollfd;
2134 uint32_t revents, nb_fd;
2135 struct lttng_consumer_stream *stream = NULL;
2136 struct lttng_ht_iter iter;
2137 struct lttng_ht_node_u64 *node;
2138 struct lttng_poll_event events;
2139 struct lttng_consumer_local_data *ctx = data;
2140 ssize_t len;
2141
2142 rcu_register_thread();
2143
2144 metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2145 if (!metadata_ht) {
2146 /* ENOMEM at this point. Better to bail out. */
2147 goto end_ht;
2148 }
2149
2150 DBG("Thread metadata poll started");
2151
2152 /* Size is set to 1 for the consumer_metadata pipe */
2153 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2154 if (ret < 0) {
2155 ERR("Poll set creation failed");
2156 goto end_poll;
2157 }
2158
2159 ret = lttng_poll_add(&events,
2160 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
2161 if (ret < 0) {
2162 goto end;
2163 }
2164
2165 /* Main loop */
2166 DBG("Metadata main loop started");
2167
2168 while (1) {
2169 /* Only the metadata pipe is set */
2170 if (LTTNG_POLL_GETNB(&events) == 0 && consumer_quit == 1) {
2171 goto end;
2172 }
2173
2174 restart:
2175 DBG("Metadata poll wait with %d fd(s)", LTTNG_POLL_GETNB(&events));
2176 ret = lttng_poll_wait(&events, -1);
2177 DBG("Metadata event catched in thread");
2178 if (ret < 0) {
2179 if (errno == EINTR) {
2180 ERR("Poll EINTR catched");
2181 goto restart;
2182 }
2183 goto error;
2184 }
2185
2186 nb_fd = ret;
2187
2188 /* From here, the event is a metadata wait fd */
2189 for (i = 0; i < nb_fd; i++) {
2190 revents = LTTNG_POLL_GETEV(&events, i);
2191 pollfd = LTTNG_POLL_GETFD(&events, i);
2192
2193 /* Just don't waste time if no returned events for the fd */
2194 if (!revents) {
2195 continue;
2196 }
2197
2198 if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
2199 if (revents & (LPOLLERR | LPOLLHUP )) {
2200 DBG("Metadata thread pipe hung up");
2201 /*
2202 * Remove the pipe from the poll set and continue the loop
2203 * since their might be data to consume.
2204 */
2205 lttng_poll_del(&events,
2206 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2207 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
2208 continue;
2209 } else if (revents & LPOLLIN) {
2210 ssize_t pipe_len;
2211
2212 pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
2213 &stream, sizeof(stream));
2214 if (pipe_len < 0) {
2215 ERR("read metadata stream, ret: %zd", pipe_len);
2216 /*
2217 * Continue here to handle the rest of the streams.
2218 */
2219 continue;
2220 }
2221
2222 /* A NULL stream means that the state has changed. */
2223 if (stream == NULL) {
2224 /* Check for deleted streams. */
2225 validate_endpoint_status_metadata_stream(&events);
2226 goto restart;
2227 }
2228
2229 DBG("Adding metadata stream %d to poll set",
2230 stream->wait_fd);
2231
2232 /* Add metadata stream to the global poll events list */
2233 lttng_poll_add(&events, stream->wait_fd,
2234 LPOLLIN | LPOLLPRI);
2235 }
2236
2237 /* Handle other stream */
2238 continue;
2239 }
2240
2241 rcu_read_lock();
2242 {
2243 uint64_t tmp_id = (uint64_t) pollfd;
2244
2245 lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
2246 }
2247 node = lttng_ht_iter_get_node_u64(&iter);
2248 assert(node);
2249
2250 stream = caa_container_of(node, struct lttng_consumer_stream,
2251 node);
2252
2253 /* Check for error event */
2254 if (revents & (LPOLLERR | LPOLLHUP)) {
2255 DBG("Metadata fd %d is hup|err.", pollfd);
2256 if (!stream->hangup_flush_done
2257 && (consumer_data.type == LTTNG_CONSUMER32_UST
2258 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2259 DBG("Attempting to flush and consume the UST buffers");
2260 lttng_ustconsumer_on_stream_hangup(stream);
2261
2262 /* We just flushed the stream now read it. */
2263 do {
2264 len = ctx->on_buffer_ready(stream, ctx);
2265 /*
2266 * We don't check the return value here since if we get
2267 * a negative len, it means an error occured thus we
2268 * simply remove it from the poll set and free the
2269 * stream.
2270 */
2271 } while (len > 0);
2272 }
2273
2274 lttng_poll_del(&events, stream->wait_fd);
2275 /*
2276 * This call update the channel states, closes file descriptors
2277 * and securely free the stream.
2278 */
2279 consumer_del_metadata_stream(stream, metadata_ht);
2280 } else if (revents & (LPOLLIN | LPOLLPRI)) {
2281 /* Get the data out of the metadata file descriptor */
2282 DBG("Metadata available on fd %d", pollfd);
2283 assert(stream->wait_fd == pollfd);
2284
2285 do {
2286 len = ctx->on_buffer_ready(stream, ctx);
2287 /*
2288 * We don't check the return value here since if we get
2289 * a negative len, it means an error occured thus we
2290 * simply remove it from the poll set and free the
2291 * stream.
2292 */
2293 } while (len > 0);
2294
2295 /* It's ok to have an unavailable sub-buffer */
2296 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2297 /* Clean up stream from consumer and free it. */
2298 lttng_poll_del(&events, stream->wait_fd);
2299 consumer_del_metadata_stream(stream, metadata_ht);
2300 }
2301 }
2302
2303 /* Release RCU lock for the stream looked up */
2304 rcu_read_unlock();
2305 }
2306 }
2307
2308 error:
2309 end:
2310 DBG("Metadata poll thread exiting");
2311
2312 lttng_poll_clean(&events);
2313 end_poll:
2314 destroy_stream_ht(metadata_ht);
2315 end_ht:
2316 rcu_unregister_thread();
2317 return NULL;
2318 }
2319
2320 /*
2321 * This thread polls the fds in the set to consume the data and write
2322 * it to tracefile if necessary.
2323 */
2324 void *consumer_thread_data_poll(void *data)
2325 {
2326 int num_rdy, num_hup, high_prio, ret, i;
2327 struct pollfd *pollfd = NULL;
2328 /* local view of the streams */
2329 struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
2330 /* local view of consumer_data.fds_count */
2331 int nb_fd = 0;
2332 struct lttng_consumer_local_data *ctx = data;
2333 ssize_t len;
2334
2335 rcu_register_thread();
2336
2337 data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2338 if (data_ht == NULL) {
2339 /* ENOMEM at this point. Better to bail out. */
2340 goto end;
2341 }
2342
2343 local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
2344 if (local_stream == NULL) {
2345 PERROR("local_stream malloc");
2346 goto end;
2347 }
2348
2349 while (1) {
2350 high_prio = 0;
2351 num_hup = 0;
2352
2353 /*
2354 * the fds set has been updated, we need to update our
2355 * local array as well
2356 */
2357 pthread_mutex_lock(&consumer_data.lock);
2358 if (consumer_data.need_update) {
2359 free(pollfd);
2360 pollfd = NULL;
2361
2362 free(local_stream);
2363 local_stream = NULL;
2364
2365 /* allocate for all fds + 1 for the consumer_data_pipe */
2366 pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
2367 if (pollfd == NULL) {
2368 PERROR("pollfd malloc");
2369 pthread_mutex_unlock(&consumer_data.lock);
2370 goto end;
2371 }
2372
2373 /* allocate for all fds + 1 for the consumer_data_pipe */
2374 local_stream = zmalloc((consumer_data.stream_count + 1) *
2375 sizeof(struct lttng_consumer_stream *));
2376 if (local_stream == NULL) {
2377 PERROR("local_stream malloc");
2378 pthread_mutex_unlock(&consumer_data.lock);
2379 goto end;
2380 }
2381 ret = update_poll_array(ctx, &pollfd, local_stream,
2382 data_ht);
2383 if (ret < 0) {
2384 ERR("Error in allocating pollfd or local_outfds");
2385 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
2386 pthread_mutex_unlock(&consumer_data.lock);
2387 goto end;
2388 }
2389 nb_fd = ret;
2390 consumer_data.need_update = 0;
2391 }
2392 pthread_mutex_unlock(&consumer_data.lock);
2393
2394 /* No FDs and consumer_quit, consumer_cleanup the thread */
2395 if (nb_fd == 0 && consumer_quit == 1) {
2396 goto end;
2397 }
2398 /* poll on the array of fds */
2399 restart:
2400 DBG("polling on %d fd", nb_fd + 1);
2401 num_rdy = poll(pollfd, nb_fd + 1, -1);
2402 DBG("poll num_rdy : %d", num_rdy);
2403 if (num_rdy == -1) {
2404 /*
2405 * Restart interrupted system call.
2406 */
2407 if (errno == EINTR) {
2408 goto restart;
2409 }
2410 PERROR("Poll error");
2411 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
2412 goto end;
2413 } else if (num_rdy == 0) {
2414 DBG("Polling thread timed out");
2415 goto end;
2416 }
2417
2418 /*
2419 * If the consumer_data_pipe triggered poll go directly to the
2420 * beginning of the loop to update the array. We want to prioritize
2421 * array update over low-priority reads.
2422 */
2423 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
2424 ssize_t pipe_readlen;
2425
2426 DBG("consumer_data_pipe wake up");
2427 pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
2428 &new_stream, sizeof(new_stream));
2429 if (pipe_readlen < 0) {
2430 ERR("Consumer data pipe ret %zd", pipe_readlen);
2431 /* Continue so we can at least handle the current stream(s). */
2432 continue;
2433 }
2434
2435 /*
2436 * If the stream is NULL, just ignore it. It's also possible that
2437 * the sessiond poll thread changed the consumer_quit state and is
2438 * waking us up to test it.
2439 */
2440 if (new_stream == NULL) {
2441 validate_endpoint_status_data_stream();
2442 continue;
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 }
2544 end:
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 */
2570 static
2571 void 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
2613 static 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 */
2640 void *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
2682 restart:
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
2836 end:
2837 lttng_poll_clean(&events);
2838 end_poll:
2839 destroy_channel_ht(channel_ht);
2840 end_ht:
2841 DBG("Channel poll thread exiting");
2842 rcu_unregister_thread();
2843 return NULL;
2844 }
2845
2846 static 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
2869 error:
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 */
2877 void *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 }
2972 end:
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
3015 ssize_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
3041 int 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 */
3059 void 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 */
3073 int 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
3254 error:
3255 if (consumer_send_status_msg(sock, ret_code) < 0) {
3256 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3257 }
3258
3259 error_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 */
3279 static 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
3298 end:
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 */
3308 static 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