9bf146d3773ce6c4cb5a9ee894fbad6cdb481630
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread-events.c
1 /*
2 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 *
4 * SPDX-License-Identifier: GPL-2.0-only
5 *
6 */
7
8 #include "lttng/action/action.h"
9 #include "lttng/trigger/trigger-internal.h"
10 #define _LGPL_SOURCE
11 #include <urcu.h>
12 #include <urcu/rculfhash.h>
13
14 #include <common/defaults.h>
15 #include <common/error.h>
16 #include <common/futex.h>
17 #include <common/unix.h>
18 #include <common/dynamic-buffer.h>
19 #include <common/hashtable/utils.h>
20 #include <common/sessiond-comm/sessiond-comm.h>
21 #include <common/macros.h>
22 #include <lttng/condition/condition.h>
23 #include <lttng/action/action-internal.h>
24 #include <lttng/notification/notification-internal.h>
25 #include <lttng/condition/condition-internal.h>
26 #include <lttng/condition/buffer-usage-internal.h>
27 #include <lttng/condition/session-consumed-size-internal.h>
28 #include <lttng/condition/session-rotation-internal.h>
29 #include <lttng/condition/event-rule-internal.h>
30 #include <lttng/domain-internal.h>
31 #include <lttng/notification/channel-internal.h>
32 #include <lttng/trigger/trigger-internal.h>
33 #include <lttng/event-rule/event-rule-internal.h>
34
35 #include <time.h>
36 #include <unistd.h>
37 #include <assert.h>
38 #include <inttypes.h>
39 #include <fcntl.h>
40
41 #include "condition-internal.h"
42 #include "notification-thread.h"
43 #include "notification-thread-events.h"
44 #include "notification-thread-commands.h"
45 #include "lttng-sessiond.h"
46 #include "kernel.h"
47
48 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
49 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
50
51 enum lttng_object_type {
52 LTTNG_OBJECT_TYPE_UNKNOWN,
53 LTTNG_OBJECT_TYPE_NONE,
54 LTTNG_OBJECT_TYPE_CHANNEL,
55 LTTNG_OBJECT_TYPE_SESSION,
56 };
57
58 struct lttng_trigger_list_element {
59 /* No ownership of the trigger object is assumed. */
60 struct lttng_trigger *trigger;
61 struct cds_list_head node;
62 };
63
64 struct lttng_channel_trigger_list {
65 struct channel_key channel_key;
66 /* List of struct lttng_trigger_list_element. */
67 struct cds_list_head list;
68 /* Node in the channel_triggers_ht */
69 struct cds_lfht_node channel_triggers_ht_node;
70 /* call_rcu delayed reclaim. */
71 struct rcu_head rcu_node;
72 };
73
74 /*
75 * List of triggers applying to a given session.
76 *
77 * See:
78 * - lttng_session_trigger_list_create()
79 * - lttng_session_trigger_list_build()
80 * - lttng_session_trigger_list_destroy()
81 * - lttng_session_trigger_list_add()
82 */
83 struct lttng_session_trigger_list {
84 /*
85 * Not owned by this; points to the session_info structure's
86 * session name.
87 */
88 const char *session_name;
89 /* List of struct lttng_trigger_list_element. */
90 struct cds_list_head list;
91 /* Node in the session_triggers_ht */
92 struct cds_lfht_node session_triggers_ht_node;
93 /*
94 * Weak reference to the notification system's session triggers
95 * hashtable.
96 *
97 * The session trigger list structure structure is owned by
98 * the session's session_info.
99 *
100 * The session_info is kept alive the the channel_infos holding a
101 * reference to it (reference counting). When those channels are
102 * destroyed (at runtime or on teardown), the reference they hold
103 * to the session_info are released. On destruction of session_info,
104 * session_info_destroy() will remove the list of triggers applying
105 * to this session from the notification system's state.
106 *
107 * This implies that the session_triggers_ht must be destroyed
108 * after the channels.
109 */
110 struct cds_lfht *session_triggers_ht;
111 /* Used for delayed RCU reclaim. */
112 struct rcu_head rcu_node;
113 };
114
115 struct lttng_trigger_ht_element {
116 struct lttng_trigger *trigger;
117 struct cds_lfht_node node;
118 struct cds_lfht_node node_by_name_uid;
119 /* call_rcu delayed reclaim. */
120 struct rcu_head rcu_node;
121 };
122
123 struct lttng_condition_list_element {
124 struct lttng_condition *condition;
125 struct cds_list_head node;
126 };
127
128 struct channel_state_sample {
129 struct channel_key key;
130 struct cds_lfht_node channel_state_ht_node;
131 uint64_t highest_usage;
132 uint64_t lowest_usage;
133 uint64_t channel_total_consumed;
134 /* call_rcu delayed reclaim. */
135 struct rcu_head rcu_node;
136 };
137
138 static unsigned long hash_channel_key(struct channel_key *key);
139 static int evaluate_buffer_condition(const struct lttng_condition *condition,
140 struct lttng_evaluation **evaluation,
141 const struct notification_thread_state *state,
142 const struct channel_state_sample *previous_sample,
143 const struct channel_state_sample *latest_sample,
144 uint64_t previous_session_consumed_total,
145 uint64_t latest_session_consumed_total,
146 struct channel_info *channel_info);
147 static
148 int send_evaluation_to_clients(const struct lttng_trigger *trigger,
149 const struct lttng_evaluation *evaluation,
150 struct notification_client_list *client_list,
151 struct notification_thread_state *state,
152 uid_t channel_uid, gid_t channel_gid);
153
154
155 /* session_info API */
156 static
157 void session_info_destroy(void *_data);
158 static
159 void session_info_get(struct session_info *session_info);
160 static
161 void session_info_put(struct session_info *session_info);
162 static
163 struct session_info *session_info_create(const char *name,
164 uid_t uid, gid_t gid,
165 struct lttng_session_trigger_list *trigger_list,
166 struct cds_lfht *sessions_ht);
167 static
168 void session_info_add_channel(struct session_info *session_info,
169 struct channel_info *channel_info);
170 static
171 void session_info_remove_channel(struct session_info *session_info,
172 struct channel_info *channel_info);
173
174 /* lttng_session_trigger_list API */
175 static
176 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
177 const char *session_name,
178 struct cds_lfht *session_triggers_ht);
179 static
180 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
181 const struct notification_thread_state *state,
182 const char *session_name);
183 static
184 void lttng_session_trigger_list_destroy(
185 struct lttng_session_trigger_list *list);
186 static
187 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
188 struct lttng_trigger *trigger);
189
190 static
191 int client_handle_transmission_status(
192 struct notification_client *client,
193 enum client_transmission_status transmission_status,
194 struct notification_thread_state *state);
195
196 static
197 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node);
198
199 static
200 int match_client_socket(struct cds_lfht_node *node, const void *key)
201 {
202 /* This double-cast is intended to supress pointer-to-cast warning. */
203 const int socket = (int) (intptr_t) key;
204 const struct notification_client *client = caa_container_of(node,
205 struct notification_client, client_socket_ht_node);
206
207 return client->socket == socket;
208 }
209
210 static
211 int match_client_id(struct cds_lfht_node *node, const void *key)
212 {
213 /* This double-cast is intended to supress pointer-to-cast warning. */
214 const notification_client_id id = *((notification_client_id *) key);
215 const struct notification_client *client = caa_container_of(
216 node, struct notification_client, client_id_ht_node);
217
218 return client->id == id;
219 }
220
221 static
222 int match_channel_trigger_list(struct cds_lfht_node *node, const void *key)
223 {
224 struct channel_key *channel_key = (struct channel_key *) key;
225 struct lttng_channel_trigger_list *trigger_list;
226
227 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
228 channel_triggers_ht_node);
229
230 return !!((channel_key->key == trigger_list->channel_key.key) &&
231 (channel_key->domain == trigger_list->channel_key.domain));
232 }
233
234 static
235 int match_session_trigger_list(struct cds_lfht_node *node, const void *key)
236 {
237 const char *session_name = (const char *) key;
238 struct lttng_session_trigger_list *trigger_list;
239
240 trigger_list = caa_container_of(node, struct lttng_session_trigger_list,
241 session_triggers_ht_node);
242
243 return !!(strcmp(trigger_list->session_name, session_name) == 0);
244 }
245
246 static
247 int match_channel_state_sample(struct cds_lfht_node *node, const void *key)
248 {
249 struct channel_key *channel_key = (struct channel_key *) key;
250 struct channel_state_sample *sample;
251
252 sample = caa_container_of(node, struct channel_state_sample,
253 channel_state_ht_node);
254
255 return !!((channel_key->key == sample->key.key) &&
256 (channel_key->domain == sample->key.domain));
257 }
258
259 static
260 int match_channel_info(struct cds_lfht_node *node, const void *key)
261 {
262 struct channel_key *channel_key = (struct channel_key *) key;
263 struct channel_info *channel_info;
264
265 channel_info = caa_container_of(node, struct channel_info,
266 channels_ht_node);
267
268 return !!((channel_key->key == channel_info->key.key) &&
269 (channel_key->domain == channel_info->key.domain));
270 }
271
272 static
273 int match_trigger(struct cds_lfht_node *node, const void *key)
274 {
275 struct lttng_trigger *trigger_key = (struct lttng_trigger *) key;
276 struct lttng_trigger_ht_element *trigger_ht_element;
277
278 trigger_ht_element = caa_container_of(node, struct lttng_trigger_ht_element,
279 node);
280
281 return !!lttng_trigger_is_equal(trigger_key, trigger_ht_element->trigger);
282 }
283
284 static
285 int match_client_list_condition(struct cds_lfht_node *node, const void *key)
286 {
287 struct lttng_condition *condition_key = (struct lttng_condition *) key;
288 struct notification_client_list *client_list;
289 const struct lttng_condition *condition;
290
291 assert(condition_key);
292
293 client_list = caa_container_of(node, struct notification_client_list,
294 notification_trigger_clients_ht_node);
295 condition = lttng_trigger_get_const_condition(client_list->trigger);
296
297 return !!lttng_condition_is_equal(condition_key, condition);
298 }
299
300 static
301 int match_session(struct cds_lfht_node *node, const void *key)
302 {
303 const char *name = key;
304 struct session_info *session_info = caa_container_of(
305 node, struct session_info, sessions_ht_node);
306
307 return !strcmp(session_info->name, name);
308 }
309
310 /*
311 * Match trigger based on name and credentials only.
312 * Name duplication is NOT allowed for the same uid.
313 */
314 static
315 int match_trigger_by_name_uid(struct cds_lfht_node *node,
316 const void *key)
317 {
318 bool match = false;
319 const char *name;
320 const char *key_name;
321 enum lttng_trigger_status status;
322 const struct lttng_credentials *key_creds;
323 const struct lttng_credentials *node_creds;
324 const struct lttng_trigger *trigger_key =
325 (const struct lttng_trigger *) key;
326 const struct lttng_trigger_ht_element *trigger_ht_element =
327 caa_container_of(node,
328 struct lttng_trigger_ht_element,
329 node_by_name_uid);
330
331 status = lttng_trigger_get_name(trigger_ht_element->trigger, &name);
332 assert(status == LTTNG_TRIGGER_STATUS_OK);
333
334 status = lttng_trigger_get_name(trigger_key, &key_name);
335 assert(status == LTTNG_TRIGGER_STATUS_OK);
336
337 /* Compare the names. */
338 if (strcmp(name, key_name) != 0) {
339 goto end;
340 }
341
342 /* Compare the owners' UIDs. */
343 key_creds = lttng_trigger_get_credentials(trigger_key);
344 node_creds = lttng_trigger_get_credentials(trigger_ht_element->trigger);
345
346 match = lttng_credentials_is_equal_uid(key_creds, node_creds);
347
348 end:
349 return match;
350 }
351
352 /*
353 * Hash trigger based on name and credentials only.
354 */
355 static
356 unsigned long hash_trigger_by_name_uid(const struct lttng_trigger *trigger)
357 {
358 unsigned long hash = 0;
359 const struct lttng_credentials *trigger_creds;
360 const char *trigger_name;
361 enum lttng_trigger_status status;
362
363 status = lttng_trigger_get_name(trigger, &trigger_name);
364 if (status == LTTNG_TRIGGER_STATUS_OK) {
365 hash = hash_key_str(trigger_name, lttng_ht_seed);
366 }
367
368 trigger_creds = lttng_trigger_get_credentials(trigger);
369 hash ^= hash_key_ulong((void *) (unsigned long) LTTNG_OPTIONAL_GET(trigger_creds->uid),
370 lttng_ht_seed);
371
372 return hash;
373 }
374
375 static
376 unsigned long hash_channel_key(struct channel_key *key)
377 {
378 unsigned long key_hash = hash_key_u64(&key->key, lttng_ht_seed);
379 unsigned long domain_hash = hash_key_ulong(
380 (void *) (unsigned long) key->domain, lttng_ht_seed);
381
382 return key_hash ^ domain_hash;
383 }
384
385 static
386 unsigned long hash_client_socket(int socket)
387 {
388 return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
389 }
390
391 static
392 unsigned long hash_client_id(notification_client_id id)
393 {
394 return hash_key_u64(&id, lttng_ht_seed);
395 }
396
397 /*
398 * Get the type of object to which a given condition applies. Bindings let
399 * the notification system evaluate a trigger's condition when a given
400 * object's state is updated.
401 *
402 * For instance, a condition bound to a channel will be evaluated everytime
403 * the channel's state is changed by a channel monitoring sample.
404 */
405 static
406 enum lttng_object_type get_condition_binding_object(
407 const struct lttng_condition *condition)
408 {
409 switch (lttng_condition_get_type(condition)) {
410 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
411 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
412 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
413 return LTTNG_OBJECT_TYPE_CHANNEL;
414 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
415 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
416 return LTTNG_OBJECT_TYPE_SESSION;
417 case LTTNG_CONDITION_TYPE_EVENT_RULE_HIT:
418 return LTTNG_OBJECT_TYPE_NONE;
419 default:
420 return LTTNG_OBJECT_TYPE_UNKNOWN;
421 }
422 }
423
424 static
425 void free_channel_info_rcu(struct rcu_head *node)
426 {
427 free(caa_container_of(node, struct channel_info, rcu_node));
428 }
429
430 static
431 void channel_info_destroy(struct channel_info *channel_info)
432 {
433 if (!channel_info) {
434 return;
435 }
436
437 if (channel_info->session_info) {
438 session_info_remove_channel(channel_info->session_info,
439 channel_info);
440 session_info_put(channel_info->session_info);
441 }
442 if (channel_info->name) {
443 free(channel_info->name);
444 }
445 call_rcu(&channel_info->rcu_node, free_channel_info_rcu);
446 }
447
448 static
449 void free_session_info_rcu(struct rcu_head *node)
450 {
451 free(caa_container_of(node, struct session_info, rcu_node));
452 }
453
454 /* Don't call directly, use the ref-counting mechanism. */
455 static
456 void session_info_destroy(void *_data)
457 {
458 struct session_info *session_info = _data;
459 int ret;
460
461 assert(session_info);
462 if (session_info->channel_infos_ht) {
463 ret = cds_lfht_destroy(session_info->channel_infos_ht, NULL);
464 if (ret) {
465 ERR("[notification-thread] Failed to destroy channel information hash table");
466 }
467 }
468 lttng_session_trigger_list_destroy(session_info->trigger_list);
469
470 rcu_read_lock();
471 cds_lfht_del(session_info->sessions_ht,
472 &session_info->sessions_ht_node);
473 rcu_read_unlock();
474 free(session_info->name);
475 call_rcu(&session_info->rcu_node, free_session_info_rcu);
476 }
477
478 static
479 void session_info_get(struct session_info *session_info)
480 {
481 if (!session_info) {
482 return;
483 }
484 lttng_ref_get(&session_info->ref);
485 }
486
487 static
488 void session_info_put(struct session_info *session_info)
489 {
490 if (!session_info) {
491 return;
492 }
493 lttng_ref_put(&session_info->ref);
494 }
495
496 static
497 struct session_info *session_info_create(const char *name, uid_t uid, gid_t gid,
498 struct lttng_session_trigger_list *trigger_list,
499 struct cds_lfht *sessions_ht)
500 {
501 struct session_info *session_info;
502
503 assert(name);
504
505 session_info = zmalloc(sizeof(*session_info));
506 if (!session_info) {
507 goto end;
508 }
509 lttng_ref_init(&session_info->ref, session_info_destroy);
510
511 session_info->channel_infos_ht = cds_lfht_new(DEFAULT_HT_SIZE,
512 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
513 if (!session_info->channel_infos_ht) {
514 goto error;
515 }
516
517 cds_lfht_node_init(&session_info->sessions_ht_node);
518 session_info->name = strdup(name);
519 if (!session_info->name) {
520 goto error;
521 }
522 session_info->uid = uid;
523 session_info->gid = gid;
524 session_info->trigger_list = trigger_list;
525 session_info->sessions_ht = sessions_ht;
526 end:
527 return session_info;
528 error:
529 session_info_put(session_info);
530 return NULL;
531 }
532
533 static
534 void session_info_add_channel(struct session_info *session_info,
535 struct channel_info *channel_info)
536 {
537 rcu_read_lock();
538 cds_lfht_add(session_info->channel_infos_ht,
539 hash_channel_key(&channel_info->key),
540 &channel_info->session_info_channels_ht_node);
541 rcu_read_unlock();
542 }
543
544 static
545 void session_info_remove_channel(struct session_info *session_info,
546 struct channel_info *channel_info)
547 {
548 rcu_read_lock();
549 cds_lfht_del(session_info->channel_infos_ht,
550 &channel_info->session_info_channels_ht_node);
551 rcu_read_unlock();
552 }
553
554 static
555 struct channel_info *channel_info_create(const char *channel_name,
556 struct channel_key *channel_key, uint64_t channel_capacity,
557 struct session_info *session_info)
558 {
559 struct channel_info *channel_info = zmalloc(sizeof(*channel_info));
560
561 if (!channel_info) {
562 goto end;
563 }
564
565 cds_lfht_node_init(&channel_info->channels_ht_node);
566 cds_lfht_node_init(&channel_info->session_info_channels_ht_node);
567 memcpy(&channel_info->key, channel_key, sizeof(*channel_key));
568 channel_info->capacity = channel_capacity;
569
570 channel_info->name = strdup(channel_name);
571 if (!channel_info->name) {
572 goto error;
573 }
574
575 /*
576 * Set the references between session and channel infos:
577 * - channel_info holds a strong reference to session_info
578 * - session_info holds a weak reference to channel_info
579 */
580 session_info_get(session_info);
581 session_info_add_channel(session_info, channel_info);
582 channel_info->session_info = session_info;
583 end:
584 return channel_info;
585 error:
586 channel_info_destroy(channel_info);
587 return NULL;
588 }
589
590 LTTNG_HIDDEN
591 bool notification_client_list_get(struct notification_client_list *list)
592 {
593 return urcu_ref_get_unless_zero(&list->ref);
594 }
595
596 static
597 void free_notification_client_list_rcu(struct rcu_head *node)
598 {
599 free(caa_container_of(node, struct notification_client_list,
600 rcu_node));
601 }
602
603 static
604 void notification_client_list_release(struct urcu_ref *list_ref)
605 {
606 struct notification_client_list *list =
607 container_of(list_ref, typeof(*list), ref);
608 struct notification_client_list_element *client_list_element, *tmp;
609
610 if (list->notification_trigger_clients_ht) {
611 rcu_read_lock();
612 cds_lfht_del(list->notification_trigger_clients_ht,
613 &list->notification_trigger_clients_ht_node);
614 rcu_read_unlock();
615 list->notification_trigger_clients_ht = NULL;
616 }
617 cds_list_for_each_entry_safe(client_list_element, tmp,
618 &list->list, node) {
619 free(client_list_element);
620 }
621 pthread_mutex_destroy(&list->lock);
622 call_rcu(&list->rcu_node, free_notification_client_list_rcu);
623 }
624
625 static
626 struct notification_client_list *notification_client_list_create(
627 const struct lttng_trigger *trigger)
628 {
629 struct notification_client_list *client_list =
630 zmalloc(sizeof(*client_list));
631
632 if (!client_list) {
633 goto error;
634 }
635 pthread_mutex_init(&client_list->lock, NULL);
636 urcu_ref_init(&client_list->ref);
637 cds_lfht_node_init(&client_list->notification_trigger_clients_ht_node);
638 CDS_INIT_LIST_HEAD(&client_list->list);
639 client_list->trigger = trigger;
640 error:
641 return client_list;
642 }
643
644 static
645 void publish_notification_client_list(
646 struct notification_thread_state *state,
647 struct notification_client_list *list)
648 {
649 const struct lttng_condition *condition =
650 lttng_trigger_get_const_condition(list->trigger);
651
652 assert(!list->notification_trigger_clients_ht);
653 notification_client_list_get(list);
654
655 list->notification_trigger_clients_ht =
656 state->notification_trigger_clients_ht;
657
658 rcu_read_lock();
659 cds_lfht_add(state->notification_trigger_clients_ht,
660 lttng_condition_hash(condition),
661 &list->notification_trigger_clients_ht_node);
662 rcu_read_unlock();
663 }
664
665 LTTNG_HIDDEN
666 void notification_client_list_put(struct notification_client_list *list)
667 {
668 if (!list) {
669 return;
670 }
671 return urcu_ref_put(&list->ref, notification_client_list_release);
672 }
673
674 /* Provides a reference to the returned list. */
675 static
676 struct notification_client_list *get_client_list_from_condition(
677 struct notification_thread_state *state,
678 const struct lttng_condition *condition)
679 {
680 struct cds_lfht_node *node;
681 struct cds_lfht_iter iter;
682 struct notification_client_list *list = NULL;
683
684 rcu_read_lock();
685 cds_lfht_lookup(state->notification_trigger_clients_ht,
686 lttng_condition_hash(condition),
687 match_client_list_condition,
688 condition,
689 &iter);
690 node = cds_lfht_iter_get_node(&iter);
691 if (node) {
692 list = container_of(node, struct notification_client_list,
693 notification_trigger_clients_ht_node);
694 list = notification_client_list_get(list) ? list : NULL;
695 }
696
697 rcu_read_unlock();
698 return list;
699 }
700
701 static
702 int evaluate_channel_condition_for_client(
703 const struct lttng_condition *condition,
704 struct notification_thread_state *state,
705 struct lttng_evaluation **evaluation,
706 uid_t *session_uid, gid_t *session_gid)
707 {
708 int ret;
709 struct cds_lfht_iter iter;
710 struct cds_lfht_node *node;
711 struct channel_info *channel_info = NULL;
712 struct channel_key *channel_key = NULL;
713 struct channel_state_sample *last_sample = NULL;
714 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
715
716 rcu_read_lock();
717
718 /* Find the channel associated with the condition. */
719 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter,
720 channel_trigger_list, channel_triggers_ht_node) {
721 struct lttng_trigger_list_element *element;
722
723 cds_list_for_each_entry(element, &channel_trigger_list->list, node) {
724 const struct lttng_condition *current_condition =
725 lttng_trigger_get_const_condition(
726 element->trigger);
727
728 assert(current_condition);
729 if (!lttng_condition_is_equal(condition,
730 current_condition)) {
731 continue;
732 }
733
734 /* Found the trigger, save the channel key. */
735 channel_key = &channel_trigger_list->channel_key;
736 break;
737 }
738 if (channel_key) {
739 /* The channel key was found stop iteration. */
740 break;
741 }
742 }
743
744 if (!channel_key){
745 /* No channel found; normal exit. */
746 DBG("[notification-thread] No known channel associated with newly subscribed-to condition");
747 ret = 0;
748 goto end;
749 }
750
751 /* Fetch channel info for the matching channel. */
752 cds_lfht_lookup(state->channels_ht,
753 hash_channel_key(channel_key),
754 match_channel_info,
755 channel_key,
756 &iter);
757 node = cds_lfht_iter_get_node(&iter);
758 assert(node);
759 channel_info = caa_container_of(node, struct channel_info,
760 channels_ht_node);
761
762 /* Retrieve the channel's last sample, if it exists. */
763 cds_lfht_lookup(state->channel_state_ht,
764 hash_channel_key(channel_key),
765 match_channel_state_sample,
766 channel_key,
767 &iter);
768 node = cds_lfht_iter_get_node(&iter);
769 if (node) {
770 last_sample = caa_container_of(node,
771 struct channel_state_sample,
772 channel_state_ht_node);
773 } else {
774 /* Nothing to evaluate, no sample was ever taken. Normal exit */
775 DBG("[notification-thread] No channel sample associated with newly subscribed-to condition");
776 ret = 0;
777 goto end;
778 }
779
780 ret = evaluate_buffer_condition(condition, evaluation, state,
781 NULL, last_sample,
782 0, channel_info->session_info->consumed_data_size,
783 channel_info);
784 if (ret) {
785 WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition");
786 goto end;
787 }
788
789 *session_uid = channel_info->session_info->uid;
790 *session_gid = channel_info->session_info->gid;
791 end:
792 rcu_read_unlock();
793 return ret;
794 }
795
796 static
797 const char *get_condition_session_name(const struct lttng_condition *condition)
798 {
799 const char *session_name = NULL;
800 enum lttng_condition_status status;
801
802 switch (lttng_condition_get_type(condition)) {
803 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
804 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
805 status = lttng_condition_buffer_usage_get_session_name(
806 condition, &session_name);
807 break;
808 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
809 status = lttng_condition_session_consumed_size_get_session_name(
810 condition, &session_name);
811 break;
812 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
813 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
814 status = lttng_condition_session_rotation_get_session_name(
815 condition, &session_name);
816 break;
817 default:
818 abort();
819 }
820 if (status != LTTNG_CONDITION_STATUS_OK) {
821 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
822 goto end;
823 }
824 end:
825 return session_name;
826 }
827
828 static
829 int evaluate_session_condition_for_client(
830 const struct lttng_condition *condition,
831 struct notification_thread_state *state,
832 struct lttng_evaluation **evaluation,
833 uid_t *session_uid, gid_t *session_gid)
834 {
835 int ret;
836 struct cds_lfht_iter iter;
837 struct cds_lfht_node *node;
838 const char *session_name;
839 struct session_info *session_info = NULL;
840
841 rcu_read_lock();
842 session_name = get_condition_session_name(condition);
843
844 /* Find the session associated with the trigger. */
845 cds_lfht_lookup(state->sessions_ht,
846 hash_key_str(session_name, lttng_ht_seed),
847 match_session,
848 session_name,
849 &iter);
850 node = cds_lfht_iter_get_node(&iter);
851 if (!node) {
852 DBG("[notification-thread] No known session matching name \"%s\"",
853 session_name);
854 ret = 0;
855 goto end;
856 }
857
858 session_info = caa_container_of(node, struct session_info,
859 sessions_ht_node);
860 session_info_get(session_info);
861
862 /*
863 * Evaluation is performed in-line here since only one type of
864 * session-bound condition is handled for the moment.
865 */
866 switch (lttng_condition_get_type(condition)) {
867 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
868 if (!session_info->rotation.ongoing) {
869 ret = 0;
870 goto end_session_put;
871 }
872
873 *evaluation = lttng_evaluation_session_rotation_ongoing_create(
874 session_info->rotation.id);
875 if (!*evaluation) {
876 /* Fatal error. */
877 ERR("[notification-thread] Failed to create session rotation ongoing evaluation for session \"%s\"",
878 session_info->name);
879 ret = -1;
880 goto end_session_put;
881 }
882 ret = 0;
883 break;
884 default:
885 ret = 0;
886 goto end_session_put;
887 }
888
889 *session_uid = session_info->uid;
890 *session_gid = session_info->gid;
891
892 end_session_put:
893 session_info_put(session_info);
894 end:
895 rcu_read_unlock();
896 return ret;
897 }
898
899 static
900 int evaluate_condition_for_client(const struct lttng_trigger *trigger,
901 const struct lttng_condition *condition,
902 struct notification_client *client,
903 struct notification_thread_state *state)
904 {
905 int ret;
906 struct lttng_evaluation *evaluation = NULL;
907 struct notification_client_list client_list = {
908 .lock = PTHREAD_MUTEX_INITIALIZER,
909 };
910 struct notification_client_list_element client_list_element = { 0 };
911 uid_t object_uid = 0;
912 gid_t object_gid = 0;
913
914 assert(trigger);
915 assert(condition);
916 assert(client);
917 assert(state);
918
919 switch (get_condition_binding_object(condition)) {
920 case LTTNG_OBJECT_TYPE_SESSION:
921 ret = evaluate_session_condition_for_client(condition, state,
922 &evaluation, &object_uid, &object_gid);
923 break;
924 case LTTNG_OBJECT_TYPE_CHANNEL:
925 ret = evaluate_channel_condition_for_client(condition, state,
926 &evaluation, &object_uid, &object_gid);
927 break;
928 case LTTNG_OBJECT_TYPE_NONE:
929 DBG("[notification-thread] Newly subscribed-to condition not bound to object, nothing to evaluate");
930 ret = 0;
931 goto end;
932 case LTTNG_OBJECT_TYPE_UNKNOWN:
933 default:
934 ret = -1;
935 goto end;
936 }
937 if (ret) {
938 /* Fatal error. */
939 goto end;
940 }
941 if (!evaluation) {
942 /* Evaluation yielded nothing. Normal exit. */
943 DBG("[notification-thread] Newly subscribed-to condition evaluated to false, nothing to report to client");
944 ret = 0;
945 goto end;
946 }
947
948 /*
949 * Create a temporary client list with the client currently
950 * subscribing.
951 */
952 cds_lfht_node_init(&client_list.notification_trigger_clients_ht_node);
953 CDS_INIT_LIST_HEAD(&client_list.list);
954 client_list.trigger = trigger;
955
956 CDS_INIT_LIST_HEAD(&client_list_element.node);
957 client_list_element.client = client;
958 cds_list_add(&client_list_element.node, &client_list.list);
959
960 /* Send evaluation result to the newly-subscribed client. */
961 DBG("[notification-thread] Newly subscribed-to condition evaluated to true, notifying client");
962 ret = send_evaluation_to_clients(trigger, evaluation, &client_list,
963 state, object_uid, object_gid);
964
965 end:
966 return ret;
967 }
968
969 static
970 int notification_thread_client_subscribe(struct notification_client *client,
971 struct lttng_condition *condition,
972 struct notification_thread_state *state,
973 enum lttng_notification_channel_status *_status)
974 {
975 int ret = 0;
976 struct notification_client_list *client_list = NULL;
977 struct lttng_condition_list_element *condition_list_element = NULL;
978 struct notification_client_list_element *client_list_element = NULL;
979 enum lttng_notification_channel_status status =
980 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
981
982 /*
983 * Ensure that the client has not already subscribed to this condition
984 * before.
985 */
986 cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
987 if (lttng_condition_is_equal(condition_list_element->condition,
988 condition)) {
989 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
990 goto end;
991 }
992 }
993
994 condition_list_element = zmalloc(sizeof(*condition_list_element));
995 if (!condition_list_element) {
996 ret = -1;
997 goto error;
998 }
999 client_list_element = zmalloc(sizeof(*client_list_element));
1000 if (!client_list_element) {
1001 ret = -1;
1002 goto error;
1003 }
1004
1005 /*
1006 * Add the newly-subscribed condition to the client's subscription list.
1007 */
1008 CDS_INIT_LIST_HEAD(&condition_list_element->node);
1009 condition_list_element->condition = condition;
1010 cds_list_add(&condition_list_element->node, &client->condition_list);
1011
1012 client_list = get_client_list_from_condition(state, condition);
1013 if (!client_list) {
1014 /*
1015 * No notification-emiting trigger registered with this
1016 * condition. We don't evaluate the condition right away
1017 * since this trigger is not registered yet.
1018 */
1019 free(client_list_element);
1020 goto end;
1021 }
1022
1023 /*
1024 * The condition to which the client just subscribed is evaluated
1025 * at this point so that conditions that are already TRUE result
1026 * in a notification being sent out.
1027 *
1028 * The client_list's trigger is used without locking the list itself.
1029 * This is correct since the list doesn't own the trigger and the
1030 * object is immutable.
1031 */
1032 if (evaluate_condition_for_client(client_list->trigger, condition,
1033 client, state)) {
1034 WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
1035 ret = -1;
1036 free(client_list_element);
1037 goto end;
1038 }
1039
1040 /*
1041 * Add the client to the list of clients interested in a given trigger
1042 * if a "notification" trigger with a corresponding condition was
1043 * added prior.
1044 */
1045 client_list_element->client = client;
1046 CDS_INIT_LIST_HEAD(&client_list_element->node);
1047
1048 pthread_mutex_lock(&client_list->lock);
1049 cds_list_add(&client_list_element->node, &client_list->list);
1050 pthread_mutex_unlock(&client_list->lock);
1051 end:
1052 if (_status) {
1053 *_status = status;
1054 }
1055 if (client_list) {
1056 notification_client_list_put(client_list);
1057 }
1058 return ret;
1059 error:
1060 free(condition_list_element);
1061 free(client_list_element);
1062 return ret;
1063 }
1064
1065 static
1066 int notification_thread_client_unsubscribe(
1067 struct notification_client *client,
1068 struct lttng_condition *condition,
1069 struct notification_thread_state *state,
1070 enum lttng_notification_channel_status *_status)
1071 {
1072 struct notification_client_list *client_list;
1073 struct lttng_condition_list_element *condition_list_element,
1074 *condition_tmp;
1075 struct notification_client_list_element *client_list_element,
1076 *client_tmp;
1077 bool condition_found = false;
1078 enum lttng_notification_channel_status status =
1079 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1080
1081 /* Remove the condition from the client's condition list. */
1082 cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
1083 &client->condition_list, node) {
1084 if (!lttng_condition_is_equal(condition_list_element->condition,
1085 condition)) {
1086 continue;
1087 }
1088
1089 cds_list_del(&condition_list_element->node);
1090 /*
1091 * The caller may be iterating on the client's conditions to
1092 * tear down a client's connection. In this case, the condition
1093 * will be destroyed at the end.
1094 */
1095 if (condition != condition_list_element->condition) {
1096 lttng_condition_destroy(
1097 condition_list_element->condition);
1098 }
1099 free(condition_list_element);
1100 condition_found = true;
1101 break;
1102 }
1103
1104 if (!condition_found) {
1105 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
1106 goto end;
1107 }
1108
1109 /*
1110 * Remove the client from the list of clients interested the trigger
1111 * matching the condition.
1112 */
1113 client_list = get_client_list_from_condition(state, condition);
1114 if (!client_list) {
1115 goto end;
1116 }
1117
1118 pthread_mutex_lock(&client_list->lock);
1119 cds_list_for_each_entry_safe(client_list_element, client_tmp,
1120 &client_list->list, node) {
1121 if (client_list_element->client->id != client->id) {
1122 continue;
1123 }
1124 cds_list_del(&client_list_element->node);
1125 free(client_list_element);
1126 break;
1127 }
1128 pthread_mutex_unlock(&client_list->lock);
1129 notification_client_list_put(client_list);
1130 client_list = NULL;
1131 end:
1132 lttng_condition_destroy(condition);
1133 if (_status) {
1134 *_status = status;
1135 }
1136 return 0;
1137 }
1138
1139 static
1140 void free_notification_client_rcu(struct rcu_head *node)
1141 {
1142 free(caa_container_of(node, struct notification_client, rcu_node));
1143 }
1144
1145 static
1146 void notification_client_destroy(struct notification_client *client,
1147 struct notification_thread_state *state)
1148 {
1149 if (!client) {
1150 return;
1151 }
1152
1153 /*
1154 * The client object is not reachable by other threads, no need to lock
1155 * the client here.
1156 */
1157 if (client->socket >= 0) {
1158 (void) lttcomm_close_unix_sock(client->socket);
1159 client->socket = -1;
1160 }
1161 client->communication.active = false;
1162 lttng_payload_reset(&client->communication.inbound.payload);
1163 lttng_payload_reset(&client->communication.outbound.payload);
1164 pthread_mutex_destroy(&client->lock);
1165 call_rcu(&client->rcu_node, free_notification_client_rcu);
1166 }
1167
1168 /*
1169 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1170 * client pointer).
1171 */
1172 static
1173 struct notification_client *get_client_from_socket(int socket,
1174 struct notification_thread_state *state)
1175 {
1176 struct cds_lfht_iter iter;
1177 struct cds_lfht_node *node;
1178 struct notification_client *client = NULL;
1179
1180 cds_lfht_lookup(state->client_socket_ht,
1181 hash_client_socket(socket),
1182 match_client_socket,
1183 (void *) (unsigned long) socket,
1184 &iter);
1185 node = cds_lfht_iter_get_node(&iter);
1186 if (!node) {
1187 goto end;
1188 }
1189
1190 client = caa_container_of(node, struct notification_client,
1191 client_socket_ht_node);
1192 end:
1193 return client;
1194 }
1195
1196 /*
1197 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1198 * client pointer).
1199 */
1200 static
1201 struct notification_client *get_client_from_id(notification_client_id id,
1202 struct notification_thread_state *state)
1203 {
1204 struct cds_lfht_iter iter;
1205 struct cds_lfht_node *node;
1206 struct notification_client *client = NULL;
1207
1208 cds_lfht_lookup(state->client_id_ht,
1209 hash_client_id(id),
1210 match_client_id,
1211 &id,
1212 &iter);
1213 node = cds_lfht_iter_get_node(&iter);
1214 if (!node) {
1215 goto end;
1216 }
1217
1218 client = caa_container_of(node, struct notification_client,
1219 client_id_ht_node);
1220 end:
1221 return client;
1222 }
1223
1224 static
1225 bool buffer_usage_condition_applies_to_channel(
1226 const struct lttng_condition *condition,
1227 const struct channel_info *channel_info)
1228 {
1229 enum lttng_condition_status status;
1230 enum lttng_domain_type condition_domain;
1231 const char *condition_session_name = NULL;
1232 const char *condition_channel_name = NULL;
1233
1234 status = lttng_condition_buffer_usage_get_domain_type(condition,
1235 &condition_domain);
1236 assert(status == LTTNG_CONDITION_STATUS_OK);
1237 if (channel_info->key.domain != condition_domain) {
1238 goto fail;
1239 }
1240
1241 status = lttng_condition_buffer_usage_get_session_name(
1242 condition, &condition_session_name);
1243 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1244
1245 status = lttng_condition_buffer_usage_get_channel_name(
1246 condition, &condition_channel_name);
1247 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_channel_name);
1248
1249 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1250 goto fail;
1251 }
1252 if (strcmp(channel_info->name, condition_channel_name)) {
1253 goto fail;
1254 }
1255
1256 return true;
1257 fail:
1258 return false;
1259 }
1260
1261 static
1262 bool session_consumed_size_condition_applies_to_channel(
1263 const struct lttng_condition *condition,
1264 const struct channel_info *channel_info)
1265 {
1266 enum lttng_condition_status status;
1267 const char *condition_session_name = NULL;
1268
1269 status = lttng_condition_session_consumed_size_get_session_name(
1270 condition, &condition_session_name);
1271 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1272
1273 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1274 goto fail;
1275 }
1276
1277 return true;
1278 fail:
1279 return false;
1280 }
1281
1282 static
1283 bool trigger_applies_to_channel(const struct lttng_trigger *trigger,
1284 const struct channel_info *channel_info)
1285 {
1286 const struct lttng_condition *condition;
1287 bool trigger_applies;
1288
1289 condition = lttng_trigger_get_const_condition(trigger);
1290 if (!condition) {
1291 goto fail;
1292 }
1293
1294 switch (lttng_condition_get_type(condition)) {
1295 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1296 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1297 trigger_applies = buffer_usage_condition_applies_to_channel(
1298 condition, channel_info);
1299 break;
1300 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
1301 trigger_applies = session_consumed_size_condition_applies_to_channel(
1302 condition, channel_info);
1303 break;
1304 default:
1305 goto fail;
1306 }
1307
1308 return trigger_applies;
1309 fail:
1310 return false;
1311 }
1312
1313 static
1314 bool trigger_applies_to_client(struct lttng_trigger *trigger,
1315 struct notification_client *client)
1316 {
1317 bool applies = false;
1318 struct lttng_condition_list_element *condition_list_element;
1319
1320 cds_list_for_each_entry(condition_list_element, &client->condition_list,
1321 node) {
1322 applies = lttng_condition_is_equal(
1323 condition_list_element->condition,
1324 lttng_trigger_get_condition(trigger));
1325 if (applies) {
1326 break;
1327 }
1328 }
1329 return applies;
1330 }
1331
1332 /* Must be called with RCU read lock held. */
1333 static
1334 struct lttng_session_trigger_list *get_session_trigger_list(
1335 struct notification_thread_state *state,
1336 const char *session_name)
1337 {
1338 struct lttng_session_trigger_list *list = NULL;
1339 struct cds_lfht_node *node;
1340 struct cds_lfht_iter iter;
1341
1342 cds_lfht_lookup(state->session_triggers_ht,
1343 hash_key_str(session_name, lttng_ht_seed),
1344 match_session_trigger_list,
1345 session_name,
1346 &iter);
1347 node = cds_lfht_iter_get_node(&iter);
1348 if (!node) {
1349 /*
1350 * Not an error, the list of triggers applying to that session
1351 * will be initialized when the session is created.
1352 */
1353 DBG("[notification-thread] No trigger list found for session \"%s\" as it is not yet known to the notification system",
1354 session_name);
1355 goto end;
1356 }
1357
1358 list = caa_container_of(node,
1359 struct lttng_session_trigger_list,
1360 session_triggers_ht_node);
1361 end:
1362 return list;
1363 }
1364
1365 /*
1366 * Allocate an empty lttng_session_trigger_list for the session named
1367 * 'session_name'.
1368 *
1369 * No ownership of 'session_name' is assumed by the session trigger list.
1370 * It is the caller's responsability to ensure the session name is alive
1371 * for as long as this list is.
1372 */
1373 static
1374 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
1375 const char *session_name,
1376 struct cds_lfht *session_triggers_ht)
1377 {
1378 struct lttng_session_trigger_list *list;
1379
1380 list = zmalloc(sizeof(*list));
1381 if (!list) {
1382 goto end;
1383 }
1384 list->session_name = session_name;
1385 CDS_INIT_LIST_HEAD(&list->list);
1386 cds_lfht_node_init(&list->session_triggers_ht_node);
1387 list->session_triggers_ht = session_triggers_ht;
1388
1389 rcu_read_lock();
1390 /* Publish the list through the session_triggers_ht. */
1391 cds_lfht_add(session_triggers_ht,
1392 hash_key_str(session_name, lttng_ht_seed),
1393 &list->session_triggers_ht_node);
1394 rcu_read_unlock();
1395 end:
1396 return list;
1397 }
1398
1399 static
1400 void free_session_trigger_list_rcu(struct rcu_head *node)
1401 {
1402 free(caa_container_of(node, struct lttng_session_trigger_list,
1403 rcu_node));
1404 }
1405
1406 static
1407 void lttng_session_trigger_list_destroy(struct lttng_session_trigger_list *list)
1408 {
1409 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1410
1411 /* Empty the list element by element, and then free the list itself. */
1412 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1413 &list->list, node) {
1414 cds_list_del(&trigger_list_element->node);
1415 free(trigger_list_element);
1416 }
1417 rcu_read_lock();
1418 /* Unpublish the list from the session_triggers_ht. */
1419 cds_lfht_del(list->session_triggers_ht,
1420 &list->session_triggers_ht_node);
1421 rcu_read_unlock();
1422 call_rcu(&list->rcu_node, free_session_trigger_list_rcu);
1423 }
1424
1425 static
1426 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
1427 struct lttng_trigger *trigger)
1428 {
1429 int ret = 0;
1430 struct lttng_trigger_list_element *new_element =
1431 zmalloc(sizeof(*new_element));
1432
1433 if (!new_element) {
1434 ret = -1;
1435 goto end;
1436 }
1437 CDS_INIT_LIST_HEAD(&new_element->node);
1438 new_element->trigger = trigger;
1439 cds_list_add(&new_element->node, &list->list);
1440 end:
1441 return ret;
1442 }
1443
1444 static
1445 bool trigger_applies_to_session(const struct lttng_trigger *trigger,
1446 const char *session_name)
1447 {
1448 bool applies = false;
1449 const struct lttng_condition *condition;
1450
1451 condition = lttng_trigger_get_const_condition(trigger);
1452 switch (lttng_condition_get_type(condition)) {
1453 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
1454 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
1455 {
1456 enum lttng_condition_status condition_status;
1457 const char *condition_session_name;
1458
1459 condition_status = lttng_condition_session_rotation_get_session_name(
1460 condition, &condition_session_name);
1461 if (condition_status != LTTNG_CONDITION_STATUS_OK) {
1462 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
1463 goto end;
1464 }
1465
1466 assert(condition_session_name);
1467 applies = !strcmp(condition_session_name, session_name);
1468 break;
1469 }
1470 default:
1471 goto end;
1472 }
1473 end:
1474 return applies;
1475 }
1476
1477 /*
1478 * Allocate and initialize an lttng_session_trigger_list which contains
1479 * all triggers that apply to the session named 'session_name'.
1480 *
1481 * No ownership of 'session_name' is assumed by the session trigger list.
1482 * It is the caller's responsability to ensure the session name is alive
1483 * for as long as this list is.
1484 */
1485 static
1486 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
1487 const struct notification_thread_state *state,
1488 const char *session_name)
1489 {
1490 int trigger_count = 0;
1491 struct lttng_session_trigger_list *session_trigger_list = NULL;
1492 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1493 struct cds_lfht_iter iter;
1494
1495 session_trigger_list = lttng_session_trigger_list_create(session_name,
1496 state->session_triggers_ht);
1497
1498 /* Add all triggers applying to the session named 'session_name'. */
1499 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1500 node) {
1501 int ret;
1502
1503 if (!trigger_applies_to_session(trigger_ht_element->trigger,
1504 session_name)) {
1505 continue;
1506 }
1507
1508 ret = lttng_session_trigger_list_add(session_trigger_list,
1509 trigger_ht_element->trigger);
1510 if (ret) {
1511 goto error;
1512 }
1513
1514 trigger_count++;
1515 }
1516
1517 DBG("[notification-thread] Found %i triggers that apply to newly created session",
1518 trigger_count);
1519 return session_trigger_list;
1520 error:
1521 lttng_session_trigger_list_destroy(session_trigger_list);
1522 return NULL;
1523 }
1524
1525 static
1526 struct session_info *find_or_create_session_info(
1527 struct notification_thread_state *state,
1528 const char *name, uid_t uid, gid_t gid)
1529 {
1530 struct session_info *session = NULL;
1531 struct cds_lfht_node *node;
1532 struct cds_lfht_iter iter;
1533 struct lttng_session_trigger_list *trigger_list;
1534
1535 rcu_read_lock();
1536 cds_lfht_lookup(state->sessions_ht,
1537 hash_key_str(name, lttng_ht_seed),
1538 match_session,
1539 name,
1540 &iter);
1541 node = cds_lfht_iter_get_node(&iter);
1542 if (node) {
1543 DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
1544 name, uid, gid);
1545 session = caa_container_of(node, struct session_info,
1546 sessions_ht_node);
1547 assert(session->uid == uid);
1548 assert(session->gid == gid);
1549 session_info_get(session);
1550 goto end;
1551 }
1552
1553 trigger_list = lttng_session_trigger_list_build(state, name);
1554 if (!trigger_list) {
1555 goto error;
1556 }
1557
1558 session = session_info_create(name, uid, gid, trigger_list,
1559 state->sessions_ht);
1560 if (!session) {
1561 ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1562 name, uid, gid);
1563 lttng_session_trigger_list_destroy(trigger_list);
1564 goto error;
1565 }
1566 trigger_list = NULL;
1567
1568 cds_lfht_add(state->sessions_ht, hash_key_str(name, lttng_ht_seed),
1569 &session->sessions_ht_node);
1570 end:
1571 rcu_read_unlock();
1572 return session;
1573 error:
1574 rcu_read_unlock();
1575 session_info_put(session);
1576 return NULL;
1577 }
1578
1579 static
1580 int handle_notification_thread_command_add_channel(
1581 struct notification_thread_state *state,
1582 const char *session_name, uid_t session_uid, gid_t session_gid,
1583 const char *channel_name, enum lttng_domain_type channel_domain,
1584 uint64_t channel_key_int, uint64_t channel_capacity,
1585 enum lttng_error_code *cmd_result)
1586 {
1587 struct cds_list_head trigger_list;
1588 struct channel_info *new_channel_info = NULL;
1589 struct channel_key channel_key = {
1590 .key = channel_key_int,
1591 .domain = channel_domain,
1592 };
1593 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
1594 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1595 int trigger_count = 0;
1596 struct cds_lfht_iter iter;
1597 struct session_info *session_info = NULL;
1598
1599 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
1600 channel_name, session_name, channel_key_int,
1601 channel_domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1602
1603 CDS_INIT_LIST_HEAD(&trigger_list);
1604
1605 session_info = find_or_create_session_info(state, session_name,
1606 session_uid, session_gid);
1607 if (!session_info) {
1608 /* Allocation error or an internal error occurred. */
1609 goto error;
1610 }
1611
1612 new_channel_info = channel_info_create(channel_name, &channel_key,
1613 channel_capacity, session_info);
1614 if (!new_channel_info) {
1615 goto error;
1616 }
1617
1618 rcu_read_lock();
1619 /* Build a list of all triggers applying to the new channel. */
1620 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1621 node) {
1622 struct lttng_trigger_list_element *new_element;
1623
1624 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
1625 new_channel_info)) {
1626 continue;
1627 }
1628
1629 new_element = zmalloc(sizeof(*new_element));
1630 if (!new_element) {
1631 rcu_read_unlock();
1632 goto error;
1633 }
1634 CDS_INIT_LIST_HEAD(&new_element->node);
1635 new_element->trigger = trigger_ht_element->trigger;
1636 cds_list_add(&new_element->node, &trigger_list);
1637 trigger_count++;
1638 }
1639 rcu_read_unlock();
1640
1641 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
1642 trigger_count);
1643 channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
1644 if (!channel_trigger_list) {
1645 goto error;
1646 }
1647 channel_trigger_list->channel_key = new_channel_info->key;
1648 CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
1649 cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
1650 cds_list_splice(&trigger_list, &channel_trigger_list->list);
1651
1652 rcu_read_lock();
1653 /* Add channel to the channel_ht which owns the channel_infos. */
1654 cds_lfht_add(state->channels_ht,
1655 hash_channel_key(&new_channel_info->key),
1656 &new_channel_info->channels_ht_node);
1657 /*
1658 * Add the list of triggers associated with this channel to the
1659 * channel_triggers_ht.
1660 */
1661 cds_lfht_add(state->channel_triggers_ht,
1662 hash_channel_key(&new_channel_info->key),
1663 &channel_trigger_list->channel_triggers_ht_node);
1664 rcu_read_unlock();
1665 session_info_put(session_info);
1666 *cmd_result = LTTNG_OK;
1667 return 0;
1668 error:
1669 channel_info_destroy(new_channel_info);
1670 session_info_put(session_info);
1671 return 1;
1672 }
1673
1674 static
1675 void free_channel_trigger_list_rcu(struct rcu_head *node)
1676 {
1677 free(caa_container_of(node, struct lttng_channel_trigger_list,
1678 rcu_node));
1679 }
1680
1681 static
1682 void free_channel_state_sample_rcu(struct rcu_head *node)
1683 {
1684 free(caa_container_of(node, struct channel_state_sample,
1685 rcu_node));
1686 }
1687
1688 static
1689 int handle_notification_thread_command_remove_channel(
1690 struct notification_thread_state *state,
1691 uint64_t channel_key, enum lttng_domain_type domain,
1692 enum lttng_error_code *cmd_result)
1693 {
1694 struct cds_lfht_node *node;
1695 struct cds_lfht_iter iter;
1696 struct lttng_channel_trigger_list *trigger_list;
1697 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1698 struct channel_key key = { .key = channel_key, .domain = domain };
1699 struct channel_info *channel_info;
1700
1701 DBG("[notification-thread] Removing channel key = %" PRIu64 " in %s domain",
1702 channel_key, domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1703
1704 rcu_read_lock();
1705
1706 cds_lfht_lookup(state->channel_triggers_ht,
1707 hash_channel_key(&key),
1708 match_channel_trigger_list,
1709 &key,
1710 &iter);
1711 node = cds_lfht_iter_get_node(&iter);
1712 /*
1713 * There is a severe internal error if we are being asked to remove a
1714 * channel that doesn't exist.
1715 */
1716 if (!node) {
1717 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
1718 goto end;
1719 }
1720
1721 /* Free the list of triggers associated with this channel. */
1722 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
1723 channel_triggers_ht_node);
1724 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1725 &trigger_list->list, node) {
1726 cds_list_del(&trigger_list_element->node);
1727 free(trigger_list_element);
1728 }
1729 cds_lfht_del(state->channel_triggers_ht, node);
1730 call_rcu(&trigger_list->rcu_node, free_channel_trigger_list_rcu);
1731
1732 /* Free sampled channel state. */
1733 cds_lfht_lookup(state->channel_state_ht,
1734 hash_channel_key(&key),
1735 match_channel_state_sample,
1736 &key,
1737 &iter);
1738 node = cds_lfht_iter_get_node(&iter);
1739 /*
1740 * This is expected to be NULL if the channel is destroyed before we
1741 * received a sample.
1742 */
1743 if (node) {
1744 struct channel_state_sample *sample = caa_container_of(node,
1745 struct channel_state_sample,
1746 channel_state_ht_node);
1747
1748 cds_lfht_del(state->channel_state_ht, node);
1749 call_rcu(&sample->rcu_node, free_channel_state_sample_rcu);
1750 }
1751
1752 /* Remove the channel from the channels_ht and free it. */
1753 cds_lfht_lookup(state->channels_ht,
1754 hash_channel_key(&key),
1755 match_channel_info,
1756 &key,
1757 &iter);
1758 node = cds_lfht_iter_get_node(&iter);
1759 assert(node);
1760 channel_info = caa_container_of(node, struct channel_info,
1761 channels_ht_node);
1762 cds_lfht_del(state->channels_ht, node);
1763 channel_info_destroy(channel_info);
1764 end:
1765 rcu_read_unlock();
1766 *cmd_result = LTTNG_OK;
1767 return 0;
1768 }
1769
1770 static
1771 int handle_notification_thread_command_session_rotation(
1772 struct notification_thread_state *state,
1773 enum notification_thread_command_type cmd_type,
1774 const char *session_name, uid_t session_uid, gid_t session_gid,
1775 uint64_t trace_archive_chunk_id,
1776 struct lttng_trace_archive_location *location,
1777 enum lttng_error_code *_cmd_result)
1778 {
1779 int ret = 0;
1780 enum lttng_error_code cmd_result = LTTNG_OK;
1781 struct lttng_session_trigger_list *trigger_list;
1782 struct lttng_trigger_list_element *trigger_list_element;
1783 struct session_info *session_info;
1784 const struct lttng_credentials session_creds = {
1785 .uid = LTTNG_OPTIONAL_INIT_VALUE(session_uid),
1786 .gid = LTTNG_OPTIONAL_INIT_VALUE(session_gid),
1787 };
1788
1789 rcu_read_lock();
1790
1791 session_info = find_or_create_session_info(state, session_name,
1792 session_uid, session_gid);
1793 if (!session_info) {
1794 /* Allocation error or an internal error occurred. */
1795 ret = -1;
1796 cmd_result = LTTNG_ERR_NOMEM;
1797 goto end;
1798 }
1799
1800 session_info->rotation.ongoing =
1801 cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING;
1802 session_info->rotation.id = trace_archive_chunk_id;
1803 trigger_list = get_session_trigger_list(state, session_name);
1804 if (!trigger_list) {
1805 DBG("[notification-thread] No triggers applying to session \"%s\" found",
1806 session_name);
1807 goto end;
1808 }
1809
1810 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
1811 node) {
1812 const struct lttng_condition *condition;
1813 struct lttng_trigger *trigger;
1814 struct notification_client_list *client_list;
1815 struct lttng_evaluation *evaluation = NULL;
1816 enum lttng_condition_type condition_type;
1817 enum action_executor_status executor_status;
1818
1819 trigger = trigger_list_element->trigger;
1820 condition = lttng_trigger_get_const_condition(trigger);
1821 assert(condition);
1822 condition_type = lttng_condition_get_type(condition);
1823
1824 if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING &&
1825 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1826 continue;
1827 } else if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED &&
1828 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED) {
1829 continue;
1830 }
1831
1832 client_list = get_client_list_from_condition(state, condition);
1833 if (cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1834 evaluation = lttng_evaluation_session_rotation_ongoing_create(
1835 trace_archive_chunk_id);
1836 } else {
1837 evaluation = lttng_evaluation_session_rotation_completed_create(
1838 trace_archive_chunk_id, location);
1839 }
1840
1841 if (!evaluation) {
1842 /* Internal error */
1843 ret = -1;
1844 cmd_result = LTTNG_ERR_UNK;
1845 goto put_list;
1846 }
1847
1848 /*
1849 * Ownership of `evaluation` transferred to the action executor
1850 * no matter the result.
1851 */
1852 executor_status = action_executor_enqueue(state->executor,
1853 trigger, evaluation, &session_creds,
1854 client_list);
1855 evaluation = NULL;
1856 switch (executor_status) {
1857 case ACTION_EXECUTOR_STATUS_OK:
1858 break;
1859 case ACTION_EXECUTOR_STATUS_ERROR:
1860 case ACTION_EXECUTOR_STATUS_INVALID:
1861 /*
1862 * TODO Add trigger identification (name/id) when
1863 * it is added to the API.
1864 */
1865 ERR("Fatal error occurred while enqueuing action associated with session rotation trigger");
1866 ret = -1;
1867 goto put_list;
1868 case ACTION_EXECUTOR_STATUS_OVERFLOW:
1869 /*
1870 * TODO Add trigger identification (name/id) when
1871 * it is added to the API.
1872 *
1873 * Not a fatal error.
1874 */
1875 WARN("No space left when enqueuing action associated with session rotation trigger");
1876 ret = 0;
1877 goto put_list;
1878 default:
1879 abort();
1880 }
1881
1882 put_list:
1883 notification_client_list_put(client_list);
1884 if (caa_unlikely(ret)) {
1885 break;
1886 }
1887 }
1888 end:
1889 session_info_put(session_info);
1890 *_cmd_result = cmd_result;
1891 rcu_read_unlock();
1892 return ret;
1893 }
1894
1895 static
1896 int handle_notification_thread_command_add_tracer_event_source(
1897 struct notification_thread_state *state,
1898 int tracer_event_source_fd,
1899 enum lttng_domain_type domain_type,
1900 enum lttng_error_code *_cmd_result)
1901 {
1902 int ret = 0;
1903 enum lttng_error_code cmd_result = LTTNG_OK;
1904 struct notification_event_tracer_event_source_element *element = NULL;
1905
1906 element = zmalloc(sizeof(*element));
1907 if (!element) {
1908 cmd_result = LTTNG_ERR_NOMEM;
1909 ret = -1;
1910 goto end;
1911 }
1912
1913 CDS_INIT_LIST_HEAD(&element->node);
1914 element->fd = tracer_event_source_fd;
1915 element->domain = domain_type;
1916
1917 cds_list_add(&element->node, &state->tracer_event_sources_list);
1918
1919 DBG3("[notification-thread] Adding tracer event source fd to poll set: tracer_event_source_fd = %d, domain = '%s'",
1920 tracer_event_source_fd,
1921 lttng_domain_type_str(domain_type));
1922
1923 /* Adding the read side pipe to the event poll. */
1924 ret = lttng_poll_add(&state->events, tracer_event_source_fd, LPOLLIN | LPOLLERR);
1925 if (ret < 0) {
1926 ERR("[notification-thread] Failed to add tracer event source to poll set: tracer_event_source_fd = %d, domain = '%s'",
1927 tracer_event_source_fd,
1928 lttng_domain_type_str(element->domain));
1929 cds_list_del(&element->node);
1930 free(element);
1931 goto end;
1932 }
1933
1934 element->is_fd_in_poll_set = true;
1935
1936 end:
1937 *_cmd_result = cmd_result;
1938 return ret;
1939 }
1940
1941 static
1942 int handle_notification_thread_command_remove_tracer_event_source(
1943 struct notification_thread_state *state,
1944 int tracer_event_source_fd,
1945 enum lttng_error_code *_cmd_result)
1946 {
1947 int ret = 0;
1948 enum lttng_error_code cmd_result = LTTNG_OK;
1949 struct notification_event_tracer_event_source_element *source_element = NULL, *tmp;
1950
1951 cds_list_for_each_entry_safe(source_element, tmp,
1952 &state->tracer_event_sources_list, node) {
1953 if (source_element->fd != tracer_event_source_fd) {
1954 continue;
1955 }
1956
1957 DBG("[notification-thread] Removed tracer event source from poll set: tracer_event_source_fd = %d, domain = '%s'",
1958 tracer_event_source_fd,
1959 lttng_domain_type_str(source_element->domain));
1960 cds_list_del(&source_element->node);
1961 break;
1962 }
1963
1964 /* It should always be found. */
1965 assert(source_element);
1966
1967 if (!source_element->is_fd_in_poll_set) {
1968 /* Skip the poll set removal. */
1969 goto end;
1970 }
1971
1972 DBG3("[notification-thread] Removing tracer event source from poll set: tracer_event_source_fd = %d, domain = '%s'",
1973 tracer_event_source_fd,
1974 lttng_domain_type_str(source_element->domain));
1975
1976 /* Removing the fd from the event poll set. */
1977 ret = lttng_poll_del(&state->events, tracer_event_source_fd);
1978 if (ret < 0) {
1979 ERR("[notification-thread] Failed to remove tracer event source from poll set: tracer_event_source_fd = %d, domain = '%s'",
1980 tracer_event_source_fd,
1981 lttng_domain_type_str(source_element->domain));
1982 cmd_result = LTTNG_ERR_FATAL;
1983 goto end;
1984 }
1985
1986 end:
1987 free(source_element);
1988 *_cmd_result = cmd_result;
1989 return ret;
1990 }
1991
1992 int handle_notification_thread_remove_tracer_event_source_no_result(
1993 struct notification_thread_state *state,
1994 int tracer_event_source_fd)
1995 {
1996 int ret;
1997 enum lttng_error_code cmd_result;
1998
1999 ret = handle_notification_thread_command_remove_tracer_event_source(
2000 state, tracer_event_source_fd, &cmd_result);
2001 (void) cmd_result;
2002 return ret;
2003 }
2004
2005 static int handle_notification_thread_command_list_triggers(
2006 struct notification_thread_handle *handle,
2007 struct notification_thread_state *state,
2008 uid_t client_uid,
2009 struct lttng_triggers **triggers,
2010 enum lttng_error_code *_cmd_result)
2011 {
2012 int ret = 0;
2013 enum lttng_error_code cmd_result = LTTNG_OK;
2014 struct cds_lfht_iter iter;
2015 struct lttng_trigger_ht_element *trigger_ht_element;
2016 struct lttng_triggers *local_triggers = NULL;
2017 const struct lttng_credentials *creds;
2018
2019 rcu_read_lock();
2020
2021 local_triggers = lttng_triggers_create();
2022 if (!local_triggers) {
2023 /* Not a fatal error. */
2024 cmd_result = LTTNG_ERR_NOMEM;
2025 goto end;
2026 }
2027
2028 cds_lfht_for_each_entry(state->triggers_ht, &iter,
2029 trigger_ht_element, node) {
2030 /*
2031 * Only return the triggers to which the client has access.
2032 * The root user has visibility over all triggers.
2033 */
2034 creds = lttng_trigger_get_credentials(trigger_ht_element->trigger);
2035 if (client_uid != lttng_credentials_get_uid(creds) && client_uid != 0) {
2036 continue;
2037 }
2038
2039 ret = lttng_triggers_add(local_triggers,
2040 trigger_ht_element->trigger);
2041 if (ret < 0) {
2042 /* Not a fatal error. */
2043 ret = 0;
2044 cmd_result = LTTNG_ERR_NOMEM;
2045 goto end;
2046 }
2047 }
2048
2049 /* Transferring ownership to the caller. */
2050 *triggers = local_triggers;
2051 local_triggers = NULL;
2052
2053 end:
2054 rcu_read_unlock();
2055 lttng_triggers_destroy(local_triggers);
2056 *_cmd_result = cmd_result;
2057 return ret;
2058 }
2059
2060 static
2061 bool condition_is_supported(struct lttng_condition *condition)
2062 {
2063 bool is_supported;
2064
2065 switch (lttng_condition_get_type(condition)) {
2066 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
2067 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
2068 {
2069 int ret;
2070 enum lttng_domain_type domain;
2071
2072 ret = lttng_condition_buffer_usage_get_domain_type(condition,
2073 &domain);
2074 assert(ret == 0);
2075
2076 if (domain != LTTNG_DOMAIN_KERNEL) {
2077 is_supported = true;
2078 goto end;
2079 }
2080
2081 /*
2082 * Older kernel tracers don't expose the API to monitor their
2083 * buffers. Therefore, we reject triggers that require that
2084 * mechanism to be available to be evaluated.
2085 *
2086 * Assume unsupported on error.
2087 */
2088 is_supported = kernel_supports_ring_buffer_snapshot_sample_positions() == 1;
2089 break;
2090 }
2091 case LTTNG_CONDITION_TYPE_EVENT_RULE_HIT:
2092 {
2093 const struct lttng_event_rule *event_rule;
2094 enum lttng_domain_type domain;
2095 const enum lttng_condition_status status =
2096 lttng_condition_event_rule_get_rule(
2097 condition, &event_rule);
2098
2099 assert(status == LTTNG_CONDITION_STATUS_OK);
2100
2101 domain = lttng_event_rule_get_domain_type(event_rule);
2102 if (domain != LTTNG_DOMAIN_KERNEL) {
2103 is_supported = true;
2104 goto end;
2105 }
2106
2107 /*
2108 * Older kernel tracers can't emit notification. Therefore, we
2109 * reject triggers that require that mechanism to be available
2110 * to be evaluated.
2111 *
2112 * Assume unsupported on error.
2113 */
2114 is_supported = kernel_supports_event_notifiers() == 1;
2115 break;
2116 }
2117 default:
2118 is_supported = true;
2119 }
2120 end:
2121 return is_supported;
2122 }
2123
2124 /* Must be called with RCU read lock held. */
2125 static
2126 int bind_trigger_to_matching_session(struct lttng_trigger *trigger,
2127 struct notification_thread_state *state)
2128 {
2129 int ret = 0;
2130 const struct lttng_condition *condition;
2131 const char *session_name;
2132 struct lttng_session_trigger_list *trigger_list;
2133
2134 condition = lttng_trigger_get_const_condition(trigger);
2135 switch (lttng_condition_get_type(condition)) {
2136 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
2137 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
2138 {
2139 enum lttng_condition_status status;
2140
2141 status = lttng_condition_session_rotation_get_session_name(
2142 condition, &session_name);
2143 if (status != LTTNG_CONDITION_STATUS_OK) {
2144 ERR("[notification-thread] Failed to bind trigger to session: unable to get 'session_rotation' condition's session name");
2145 ret = -1;
2146 goto end;
2147 }
2148 break;
2149 }
2150 default:
2151 ret = -1;
2152 goto end;
2153 }
2154
2155 trigger_list = get_session_trigger_list(state, session_name);
2156 if (!trigger_list) {
2157 DBG("[notification-thread] Unable to bind trigger applying to session \"%s\" as it is not yet known to the notification system",
2158 session_name);
2159 goto end;
2160
2161 }
2162
2163 DBG("[notification-thread] Newly registered trigger bound to session \"%s\"",
2164 session_name);
2165 ret = lttng_session_trigger_list_add(trigger_list, trigger);
2166 end:
2167 return ret;
2168 }
2169
2170 /* Must be called with RCU read lock held. */
2171 static
2172 int bind_trigger_to_matching_channels(struct lttng_trigger *trigger,
2173 struct notification_thread_state *state)
2174 {
2175 int ret = 0;
2176 struct cds_lfht_node *node;
2177 struct cds_lfht_iter iter;
2178 struct channel_info *channel;
2179
2180 cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
2181 channels_ht_node) {
2182 struct lttng_trigger_list_element *trigger_list_element;
2183 struct lttng_channel_trigger_list *trigger_list;
2184 struct cds_lfht_iter lookup_iter;
2185
2186 if (!trigger_applies_to_channel(trigger, channel)) {
2187 continue;
2188 }
2189
2190 cds_lfht_lookup(state->channel_triggers_ht,
2191 hash_channel_key(&channel->key),
2192 match_channel_trigger_list,
2193 &channel->key,
2194 &lookup_iter);
2195 node = cds_lfht_iter_get_node(&lookup_iter);
2196 assert(node);
2197 trigger_list = caa_container_of(node,
2198 struct lttng_channel_trigger_list,
2199 channel_triggers_ht_node);
2200
2201 trigger_list_element = zmalloc(sizeof(*trigger_list_element));
2202 if (!trigger_list_element) {
2203 ret = -1;
2204 goto end;
2205 }
2206 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
2207 trigger_list_element->trigger = trigger;
2208 cds_list_add(&trigger_list_element->node, &trigger_list->list);
2209 DBG("[notification-thread] Newly registered trigger bound to channel \"%s\"",
2210 channel->name);
2211 }
2212 end:
2213 return ret;
2214 }
2215
2216 static
2217 bool is_trigger_action_notify(const struct lttng_trigger *trigger)
2218 {
2219 bool is_notify = false;
2220 unsigned int i, count;
2221 enum lttng_action_status action_status;
2222 const struct lttng_action *action =
2223 lttng_trigger_get_const_action(trigger);
2224 enum lttng_action_type action_type;
2225
2226 assert(action);
2227 action_type = lttng_action_get_type(action);
2228 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2229 is_notify = true;
2230 goto end;
2231 } else if (action_type != LTTNG_ACTION_TYPE_GROUP) {
2232 goto end;
2233 }
2234
2235 action_status = lttng_action_group_get_count(action, &count);
2236 assert(action_status == LTTNG_ACTION_STATUS_OK);
2237
2238 for (i = 0; i < count; i++) {
2239 const struct lttng_action *inner_action =
2240 lttng_action_group_get_at_index(
2241 action, i);
2242
2243 action_type = lttng_action_get_type(inner_action);
2244 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2245 is_notify = true;
2246 goto end;
2247 }
2248 }
2249
2250 end:
2251 return is_notify;
2252 }
2253
2254 static bool trigger_name_taken(struct notification_thread_state *state,
2255 const struct lttng_trigger *trigger)
2256 {
2257 struct cds_lfht_iter iter;
2258
2259 /*
2260 * No duplicata is allowed in the triggers_by_name_uid_ht.
2261 * The match is done against the trigger name and uid.
2262 */
2263 cds_lfht_lookup(state->triggers_by_name_uid_ht,
2264 hash_trigger_by_name_uid(trigger),
2265 match_trigger_by_name_uid,
2266 trigger,
2267 &iter);
2268 return !!cds_lfht_iter_get_node(&iter);
2269 }
2270
2271 static
2272 enum lttng_error_code generate_trigger_name(
2273 struct notification_thread_state *state,
2274 struct lttng_trigger *trigger, const char **name)
2275 {
2276 enum lttng_error_code ret_code = LTTNG_OK;
2277 bool taken = false;
2278 enum lttng_trigger_status status;
2279
2280 do {
2281 const int ret = lttng_trigger_generate_name(trigger,
2282 state->trigger_id.name_offset++);
2283 if (ret) {
2284 /* The only reason this can fail right now. */
2285 ret_code = LTTNG_ERR_NOMEM;
2286 break;
2287 }
2288
2289 status = lttng_trigger_get_name(trigger, name);
2290 assert(status == LTTNG_TRIGGER_STATUS_OK);
2291
2292 taken = trigger_name_taken(state, trigger);
2293 } while (taken || state->trigger_id.name_offset == UINT64_MAX);
2294
2295 return ret_code;
2296 }
2297
2298 /*
2299 * FIXME A client's credentials are not checked when registering a trigger.
2300 *
2301 * The effects of this are benign since:
2302 * - The client will succeed in registering the trigger, as it is valid,
2303 * - The trigger will, internally, be bound to the channel/session,
2304 * - The notifications will not be sent since the client's credentials
2305 * are checked against the channel at that moment.
2306 *
2307 * If this function returns a non-zero value, it means something is
2308 * fundamentally broken and the whole subsystem/thread will be torn down.
2309 *
2310 * If a non-fatal error occurs, just set the cmd_result to the appropriate
2311 * error code.
2312 */
2313 static
2314 int handle_notification_thread_command_register_trigger(
2315 struct notification_thread_state *state,
2316 struct lttng_trigger *trigger,
2317 enum lttng_error_code *cmd_result)
2318 {
2319 int ret = 0;
2320 struct lttng_condition *condition;
2321 struct notification_client *client;
2322 struct notification_client_list *client_list = NULL;
2323 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2324 struct notification_client_list_element *client_list_element;
2325 struct cds_lfht_node *node;
2326 struct cds_lfht_iter iter;
2327 const char* trigger_name;
2328 bool free_trigger = true;
2329 struct lttng_evaluation *evaluation = NULL;
2330 struct lttng_credentials object_creds;
2331 uid_t object_uid;
2332 gid_t object_gid;
2333 enum action_executor_status executor_status;
2334 const uint64_t trigger_tracer_token =
2335 state->trigger_id.next_tracer_token++;
2336
2337 rcu_read_lock();
2338
2339 /* Set the trigger's tracer token. */
2340 lttng_trigger_set_tracer_token(trigger, trigger_tracer_token);
2341
2342 if (lttng_trigger_get_name(trigger, &trigger_name) ==
2343 LTTNG_TRIGGER_STATUS_UNSET) {
2344 const enum lttng_error_code ret_code = generate_trigger_name(
2345 state, trigger, &trigger_name);
2346
2347 if (ret_code != LTTNG_OK) {
2348 /* Fatal error. */
2349 ret = -1;
2350 *cmd_result = ret_code;
2351 goto error;
2352 }
2353 } else if (trigger_name_taken(state, trigger)) {
2354 /* Not a fatal error. */
2355 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2356 ret = 0;
2357 goto error;
2358 }
2359
2360 condition = lttng_trigger_get_condition(trigger);
2361 assert(condition);
2362
2363 /* Some conditions require tracers to implement a minimal ABI version. */
2364 if (!condition_is_supported(condition)) {
2365 *cmd_result = LTTNG_ERR_NOT_SUPPORTED;
2366 goto error;
2367 }
2368
2369 trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
2370 if (!trigger_ht_element) {
2371 ret = -1;
2372 goto error;
2373 }
2374
2375 /* Add trigger to the trigger_ht. */
2376 cds_lfht_node_init(&trigger_ht_element->node);
2377 cds_lfht_node_init(&trigger_ht_element->node_by_name_uid);
2378 trigger_ht_element->trigger = trigger;
2379
2380 node = cds_lfht_add_unique(state->triggers_ht,
2381 lttng_condition_hash(condition),
2382 match_trigger,
2383 trigger,
2384 &trigger_ht_element->node);
2385 if (node != &trigger_ht_element->node) {
2386 /* Not a fatal error, simply report it to the client. */
2387 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2388 goto error_free_ht_element;
2389 }
2390
2391 node = cds_lfht_add_unique(state->triggers_by_name_uid_ht,
2392 hash_trigger_by_name_uid(trigger),
2393 match_trigger_by_name_uid,
2394 trigger,
2395 &trigger_ht_element->node_by_name_uid);
2396 if (node != &trigger_ht_element->node_by_name_uid) {
2397 /* Not a fatal error, simply report it to the client. */
2398 cds_lfht_del(state->triggers_ht, &trigger_ht_element->node);
2399 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2400 goto error_free_ht_element;
2401 }
2402
2403 /*
2404 * Ownership of the trigger and of its wrapper was transfered to
2405 * the triggers_ht.
2406 */
2407 trigger_ht_element = NULL;
2408 free_trigger = false;
2409
2410 /*
2411 * The rest only applies to triggers that have a "notify" action.
2412 * It is not skipped as this is the only action type currently
2413 * supported.
2414 */
2415 if (is_trigger_action_notify(trigger)) {
2416 client_list = notification_client_list_create(trigger);
2417 if (!client_list) {
2418 ret = -1;
2419 goto error_free_ht_element;
2420 }
2421
2422 /* Build a list of clients to which this new trigger applies. */
2423 cds_lfht_for_each_entry (state->client_socket_ht, &iter, client,
2424 client_socket_ht_node) {
2425 if (!trigger_applies_to_client(trigger, client)) {
2426 continue;
2427 }
2428
2429 client_list_element =
2430 zmalloc(sizeof(*client_list_element));
2431 if (!client_list_element) {
2432 ret = -1;
2433 goto error_put_client_list;
2434 }
2435
2436 CDS_INIT_LIST_HEAD(&client_list_element->node);
2437 client_list_element->client = client;
2438 cds_list_add(&client_list_element->node,
2439 &client_list->list);
2440 }
2441
2442 /*
2443 * Client list ownership transferred to the
2444 * notification_trigger_clients_ht.
2445 */
2446 publish_notification_client_list(state, client_list);
2447 }
2448
2449 switch (get_condition_binding_object(condition)) {
2450 case LTTNG_OBJECT_TYPE_SESSION:
2451 /* Add the trigger to the list if it matches a known session. */
2452 ret = bind_trigger_to_matching_session(trigger, state);
2453 if (ret) {
2454 goto error_put_client_list;
2455 }
2456 break;
2457 case LTTNG_OBJECT_TYPE_CHANNEL:
2458 /*
2459 * Add the trigger to list of triggers bound to the channels
2460 * currently known.
2461 */
2462 ret = bind_trigger_to_matching_channels(trigger, state);
2463 if (ret) {
2464 goto error_put_client_list;
2465 }
2466 break;
2467 case LTTNG_OBJECT_TYPE_NONE:
2468 break;
2469 default:
2470 ERR("Unknown object type on which to bind a newly registered trigger was encountered");
2471 ret = -1;
2472 goto error_put_client_list;
2473 }
2474
2475 /*
2476 * The new trigger's condition must be evaluated against the current
2477 * state.
2478 *
2479 * In the case of `notify` action, nothing preventing clients from
2480 * subscribing to a condition before the corresponding trigger is
2481 * registered, we have to evaluate this new condition right away.
2482 *
2483 * At some point, we were waiting for the next "evaluation" (e.g. on
2484 * reception of a channel sample) to evaluate this new condition, but
2485 * that was broken.
2486 *
2487 * The reason it was broken is that waiting for the next sample
2488 * does not allow us to properly handle transitions for edge-triggered
2489 * conditions.
2490 *
2491 * Consider this example: when we handle a new channel sample, we
2492 * evaluate each conditions twice: once with the previous state, and
2493 * again with the newest state. We then use those two results to
2494 * determine whether a state change happened: a condition was false and
2495 * became true. If a state change happened, we have to notify clients.
2496 *
2497 * Now, if a client subscribes to a given notification and registers
2498 * a trigger *after* that subscription, we have to make sure the
2499 * condition is evaluated at this point while considering only the
2500 * current state. Otherwise, the next evaluation cycle may only see
2501 * that the evaluations remain the same (true for samples n-1 and n) and
2502 * the client will never know that the condition has been met.
2503 */
2504 switch (get_condition_binding_object(condition)) {
2505 case LTTNG_OBJECT_TYPE_SESSION:
2506 ret = evaluate_session_condition_for_client(condition, state,
2507 &evaluation, &object_uid,
2508 &object_gid);
2509 break;
2510 case LTTNG_OBJECT_TYPE_CHANNEL:
2511 ret = evaluate_channel_condition_for_client(condition, state,
2512 &evaluation, &object_uid,
2513 &object_gid);
2514 break;
2515 case LTTNG_OBJECT_TYPE_NONE:
2516 ret = 0;
2517 break;
2518 case LTTNG_OBJECT_TYPE_UNKNOWN:
2519 default:
2520 ret = -1;
2521 break;
2522 }
2523
2524 if (ret) {
2525 /* Fatal error. */
2526 goto error_put_client_list;
2527 }
2528
2529 LTTNG_OPTIONAL_SET(&object_creds.uid, object_uid);
2530 LTTNG_OPTIONAL_SET(&object_creds.gid, object_gid);
2531
2532 DBG("Newly registered trigger's condition evaluated to %s",
2533 evaluation ? "true" : "false");
2534 if (!evaluation) {
2535 /* Evaluation yielded nothing. Normal exit. */
2536 ret = 0;
2537 goto end;
2538 }
2539
2540 /*
2541 * Ownership of `evaluation` transferred to the action executor
2542 * no matter the result.
2543 */
2544 executor_status = action_executor_enqueue(state->executor, trigger,
2545 evaluation, &object_creds, client_list);
2546 evaluation = NULL;
2547 switch (executor_status) {
2548 case ACTION_EXECUTOR_STATUS_OK:
2549 break;
2550 case ACTION_EXECUTOR_STATUS_ERROR:
2551 case ACTION_EXECUTOR_STATUS_INVALID:
2552 /*
2553 * TODO Add trigger identification (name/id) when
2554 * it is added to the API.
2555 */
2556 ERR("Fatal error occurred while enqueuing action associated to newly registered trigger");
2557 ret = -1;
2558 goto error_put_client_list;
2559 case ACTION_EXECUTOR_STATUS_OVERFLOW:
2560 /*
2561 * TODO Add trigger identification (name/id) when
2562 * it is added to the API.
2563 *
2564 * Not a fatal error.
2565 */
2566 WARN("No space left when enqueuing action associated to newly registered trigger");
2567 ret = 0;
2568 goto end;
2569 default:
2570 abort();
2571 }
2572
2573 end:
2574 *cmd_result = LTTNG_OK;
2575 DBG("Registered trigger: name = `%s`, tracer token = %" PRIu64,
2576 trigger_name, trigger_tracer_token);
2577
2578 error_put_client_list:
2579 notification_client_list_put(client_list);
2580
2581 error_free_ht_element:
2582 if (trigger_ht_element) {
2583 /* Delayed removal due to RCU constraint on delete. */
2584 call_rcu(&trigger_ht_element->rcu_node,
2585 free_lttng_trigger_ht_element_rcu);
2586 }
2587
2588 error:
2589 if (free_trigger) {
2590 lttng_trigger_destroy(trigger);
2591 }
2592 rcu_read_unlock();
2593 return ret;
2594 }
2595
2596 static
2597 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node)
2598 {
2599 free(caa_container_of(node, struct lttng_trigger_ht_element,
2600 rcu_node));
2601 }
2602
2603 static
2604 int handle_notification_thread_command_unregister_trigger(
2605 struct notification_thread_state *state,
2606 struct lttng_trigger *trigger,
2607 enum lttng_error_code *_cmd_reply)
2608 {
2609 struct cds_lfht_iter iter;
2610 struct cds_lfht_node *triggers_ht_node;
2611 struct lttng_channel_trigger_list *trigger_list;
2612 struct notification_client_list *client_list;
2613 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2614 struct lttng_condition *condition = lttng_trigger_get_condition(
2615 trigger);
2616 enum lttng_error_code cmd_reply;
2617
2618 rcu_read_lock();
2619
2620 cds_lfht_lookup(state->triggers_ht,
2621 lttng_condition_hash(condition),
2622 match_trigger,
2623 trigger,
2624 &iter);
2625 triggers_ht_node = cds_lfht_iter_get_node(&iter);
2626 if (!triggers_ht_node) {
2627 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
2628 goto end;
2629 } else {
2630 cmd_reply = LTTNG_OK;
2631 }
2632
2633 /* Remove trigger from channel_triggers_ht. */
2634 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
2635 channel_triggers_ht_node) {
2636 struct lttng_trigger_list_element *trigger_element, *tmp;
2637
2638 cds_list_for_each_entry_safe(trigger_element, tmp,
2639 &trigger_list->list, node) {
2640 if (!lttng_trigger_is_equal(trigger, trigger_element->trigger)) {
2641 continue;
2642 }
2643
2644 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
2645 cds_list_del(&trigger_element->node);
2646 /* A trigger can only appear once per channel */
2647 break;
2648 }
2649 }
2650
2651 if (is_trigger_action_notify(trigger)) {
2652 /*
2653 * Remove and release the client list from
2654 * notification_trigger_clients_ht.
2655 */
2656 client_list = get_client_list_from_condition(state, condition);
2657 assert(client_list);
2658
2659 /* Put new reference and the hashtable's reference. */
2660 notification_client_list_put(client_list);
2661 notification_client_list_put(client_list);
2662 client_list = NULL;
2663 }
2664
2665 /* Remove trigger from triggers_ht. */
2666 trigger_ht_element = caa_container_of(triggers_ht_node,
2667 struct lttng_trigger_ht_element, node);
2668 cds_lfht_del(state->triggers_by_name_uid_ht, &trigger_ht_element->node_by_name_uid);
2669 cds_lfht_del(state->triggers_ht, triggers_ht_node);
2670
2671 /* Release the ownership of the trigger. */
2672 lttng_trigger_destroy(trigger_ht_element->trigger);
2673 call_rcu(&trigger_ht_element->rcu_node, free_lttng_trigger_ht_element_rcu);
2674 end:
2675 rcu_read_unlock();
2676 if (_cmd_reply) {
2677 *_cmd_reply = cmd_reply;
2678 }
2679 return 0;
2680 }
2681
2682 /* Returns 0 on success, 1 on exit requested, negative value on error. */
2683 int handle_notification_thread_command(
2684 struct notification_thread_handle *handle,
2685 struct notification_thread_state *state)
2686 {
2687 int ret;
2688 uint64_t counter;
2689 struct notification_thread_command *cmd;
2690
2691 /* Read the event pipe to put it back into a quiescent state. */
2692 ret = lttng_read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
2693 sizeof(counter));
2694 if (ret != sizeof(counter)) {
2695 goto error;
2696 }
2697
2698 pthread_mutex_lock(&handle->cmd_queue.lock);
2699 cmd = cds_list_first_entry(&handle->cmd_queue.list,
2700 struct notification_thread_command, cmd_list_node);
2701 cds_list_del(&cmd->cmd_list_node);
2702 pthread_mutex_unlock(&handle->cmd_queue.lock);
2703 switch (cmd->type) {
2704 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
2705 DBG("[notification-thread] Received register trigger command");
2706 ret = handle_notification_thread_command_register_trigger(
2707 state, cmd->parameters.trigger,
2708 &cmd->reply_code);
2709 break;
2710 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
2711 DBG("[notification-thread] Received unregister trigger command");
2712 ret = handle_notification_thread_command_unregister_trigger(
2713 state, cmd->parameters.trigger,
2714 &cmd->reply_code);
2715 break;
2716 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
2717 DBG("[notification-thread] Received add channel command");
2718 ret = handle_notification_thread_command_add_channel(
2719 state,
2720 cmd->parameters.add_channel.session.name,
2721 cmd->parameters.add_channel.session.uid,
2722 cmd->parameters.add_channel.session.gid,
2723 cmd->parameters.add_channel.channel.name,
2724 cmd->parameters.add_channel.channel.domain,
2725 cmd->parameters.add_channel.channel.key,
2726 cmd->parameters.add_channel.channel.capacity,
2727 &cmd->reply_code);
2728 break;
2729 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
2730 DBG("[notification-thread] Received remove channel command");
2731 ret = handle_notification_thread_command_remove_channel(
2732 state, cmd->parameters.remove_channel.key,
2733 cmd->parameters.remove_channel.domain,
2734 &cmd->reply_code);
2735 break;
2736 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
2737 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
2738 DBG("[notification-thread] Received session rotation %s command",
2739 cmd->type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING ?
2740 "ongoing" : "completed");
2741 ret = handle_notification_thread_command_session_rotation(
2742 state,
2743 cmd->type,
2744 cmd->parameters.session_rotation.session_name,
2745 cmd->parameters.session_rotation.uid,
2746 cmd->parameters.session_rotation.gid,
2747 cmd->parameters.session_rotation.trace_archive_chunk_id,
2748 cmd->parameters.session_rotation.location,
2749 &cmd->reply_code);
2750 break;
2751 case NOTIFICATION_COMMAND_TYPE_ADD_TRACER_EVENT_SOURCE:
2752 ret = handle_notification_thread_command_add_tracer_event_source(
2753 state,
2754 cmd->parameters.tracer_event_source.tracer_event_source_fd,
2755 cmd->parameters.tracer_event_source.domain,
2756 &cmd->reply_code);
2757 break;
2758 case NOTIFICATION_COMMAND_TYPE_REMOVE_TRACER_EVENT_SOURCE:
2759 ret = handle_notification_thread_command_remove_tracer_event_source(
2760 state,
2761 cmd->parameters.tracer_event_source.tracer_event_source_fd,
2762 &cmd->reply_code);
2763 break;
2764 case NOTIFICATION_COMMAND_TYPE_LIST_TRIGGERS:
2765 {
2766 struct lttng_triggers *triggers = NULL;
2767
2768 ret = handle_notification_thread_command_list_triggers(
2769 handle,
2770 state,
2771 cmd->parameters.list_triggers.uid,
2772 &triggers,
2773 &cmd->reply_code);
2774 cmd->reply.list_triggers.triggers = triggers;
2775 ret = 0;
2776 break;
2777 }
2778 case NOTIFICATION_COMMAND_TYPE_QUIT:
2779 DBG("[notification-thread] Received quit command");
2780 cmd->reply_code = LTTNG_OK;
2781 ret = 1;
2782 goto end;
2783 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE:
2784 {
2785 const enum client_transmission_status client_status =
2786 cmd->parameters.client_communication_update
2787 .status;
2788 const notification_client_id client_id =
2789 cmd->parameters.client_communication_update.id;
2790 struct notification_client *client;
2791
2792 rcu_read_lock();
2793 client = get_client_from_id(client_id, state);
2794
2795 if (!client) {
2796 /*
2797 * Client error was probably already picked-up by the
2798 * notification thread or it has disconnected
2799 * gracefully while this command was queued.
2800 */
2801 DBG("Failed to find notification client to update communication status, client id = %" PRIu64,
2802 client_id);
2803 ret = 0;
2804 } else {
2805 ret = client_handle_transmission_status(
2806 client, client_status, state);
2807 }
2808 rcu_read_unlock();
2809 break;
2810 }
2811 default:
2812 ERR("[notification-thread] Unknown internal command received");
2813 goto error_unlock;
2814 }
2815
2816 if (ret) {
2817 goto error_unlock;
2818 }
2819 end:
2820 if (cmd->is_async) {
2821 free(cmd);
2822 cmd = NULL;
2823 } else {
2824 lttng_waiter_wake_up(&cmd->reply_waiter);
2825 }
2826 return ret;
2827 error_unlock:
2828 /* Wake-up and return a fatal error to the calling thread. */
2829 lttng_waiter_wake_up(&cmd->reply_waiter);
2830 cmd->reply_code = LTTNG_ERR_FATAL;
2831 error:
2832 /* Indicate a fatal error to the caller. */
2833 return -1;
2834 }
2835
2836 static
2837 int socket_set_non_blocking(int socket)
2838 {
2839 int ret, flags;
2840
2841 /* Set the pipe as non-blocking. */
2842 ret = fcntl(socket, F_GETFL, 0);
2843 if (ret == -1) {
2844 PERROR("fcntl get socket flags");
2845 goto end;
2846 }
2847 flags = ret;
2848
2849 ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
2850 if (ret == -1) {
2851 PERROR("fcntl set O_NONBLOCK socket flag");
2852 goto end;
2853 }
2854 DBG("Client socket (fd = %i) set as non-blocking", socket);
2855 end:
2856 return ret;
2857 }
2858
2859 static
2860 int client_reset_inbound_state(struct notification_client *client)
2861 {
2862 int ret;
2863
2864
2865 lttng_payload_clear(&client->communication.inbound.payload);
2866
2867 client->communication.inbound.bytes_to_receive =
2868 sizeof(struct lttng_notification_channel_message);
2869 client->communication.inbound.msg_type =
2870 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
2871 LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
2872 LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
2873 ret = lttng_dynamic_buffer_set_size(
2874 &client->communication.inbound.payload.buffer,
2875 client->communication.inbound.bytes_to_receive);
2876
2877 return ret;
2878 }
2879
2880 int handle_notification_thread_client_connect(
2881 struct notification_thread_state *state)
2882 {
2883 int ret;
2884 struct notification_client *client;
2885
2886 DBG("[notification-thread] Handling new notification channel client connection");
2887
2888 client = zmalloc(sizeof(*client));
2889 if (!client) {
2890 /* Fatal error. */
2891 ret = -1;
2892 goto error;
2893 }
2894
2895 pthread_mutex_init(&client->lock, NULL);
2896 client->id = state->next_notification_client_id++;
2897 CDS_INIT_LIST_HEAD(&client->condition_list);
2898 lttng_payload_init(&client->communication.inbound.payload);
2899 lttng_payload_init(&client->communication.outbound.payload);
2900 client->communication.inbound.expect_creds = true;
2901
2902 ret = client_reset_inbound_state(client);
2903 if (ret) {
2904 ERR("[notification-thread] Failed to reset client communication's inbound state");
2905 ret = 0;
2906 goto error;
2907 }
2908
2909 ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
2910 if (ret < 0) {
2911 ERR("[notification-thread] Failed to accept new notification channel client connection");
2912 ret = 0;
2913 goto error;
2914 }
2915
2916 client->socket = ret;
2917
2918 ret = socket_set_non_blocking(client->socket);
2919 if (ret) {
2920 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
2921 goto error;
2922 }
2923
2924 ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
2925 if (ret < 0) {
2926 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
2927 ret = 0;
2928 goto error;
2929 }
2930
2931 ret = lttng_poll_add(&state->events, client->socket,
2932 LPOLLIN | LPOLLERR |
2933 LPOLLHUP | LPOLLRDHUP);
2934 if (ret < 0) {
2935 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
2936 ret = 0;
2937 goto error;
2938 }
2939 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
2940 client->socket);
2941
2942 rcu_read_lock();
2943 cds_lfht_add(state->client_socket_ht,
2944 hash_client_socket(client->socket),
2945 &client->client_socket_ht_node);
2946 cds_lfht_add(state->client_id_ht,
2947 hash_client_id(client->id),
2948 &client->client_id_ht_node);
2949 rcu_read_unlock();
2950
2951 return ret;
2952
2953 error:
2954 notification_client_destroy(client, state);
2955 return ret;
2956 }
2957
2958 /*
2959 * RCU read-lock must be held by the caller.
2960 * Client lock must _not_ be held by the caller.
2961 */
2962 static
2963 int notification_thread_client_disconnect(
2964 struct notification_client *client,
2965 struct notification_thread_state *state)
2966 {
2967 int ret;
2968 struct lttng_condition_list_element *condition_list_element, *tmp;
2969
2970 /* Acquire the client lock to disable its communication atomically. */
2971 pthread_mutex_lock(&client->lock);
2972 client->communication.active = false;
2973 cds_lfht_del(state->client_socket_ht, &client->client_socket_ht_node);
2974 cds_lfht_del(state->client_id_ht, &client->client_id_ht_node);
2975 pthread_mutex_unlock(&client->lock);
2976
2977 ret = lttng_poll_del(&state->events, client->socket);
2978 if (ret) {
2979 ERR("[notification-thread] Failed to remove client socket %d from poll set",
2980 client->socket);
2981 }
2982
2983 /* Release all conditions to which the client was subscribed. */
2984 cds_list_for_each_entry_safe(condition_list_element, tmp,
2985 &client->condition_list, node) {
2986 (void) notification_thread_client_unsubscribe(client,
2987 condition_list_element->condition, state, NULL);
2988 }
2989
2990 /*
2991 * Client no longer accessible to other threads (through the
2992 * client lists).
2993 */
2994 notification_client_destroy(client, state);
2995 return ret;
2996 }
2997
2998 int handle_notification_thread_client_disconnect(
2999 int client_socket, struct notification_thread_state *state)
3000 {
3001 int ret = 0;
3002 struct notification_client *client;
3003
3004 rcu_read_lock();
3005 DBG("[notification-thread] Closing client connection (socket fd = %i)",
3006 client_socket);
3007 client = get_client_from_socket(client_socket, state);
3008 if (!client) {
3009 /* Internal state corruption, fatal error. */
3010 ERR("[notification-thread] Unable to find client (socket fd = %i)",
3011 client_socket);
3012 ret = -1;
3013 goto end;
3014 }
3015
3016 ret = notification_thread_client_disconnect(client, state);
3017 end:
3018 rcu_read_unlock();
3019 return ret;
3020 }
3021
3022 int handle_notification_thread_client_disconnect_all(
3023 struct notification_thread_state *state)
3024 {
3025 struct cds_lfht_iter iter;
3026 struct notification_client *client;
3027 bool error_encoutered = false;
3028
3029 rcu_read_lock();
3030 DBG("[notification-thread] Closing all client connections");
3031 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
3032 client_socket_ht_node) {
3033 int ret;
3034
3035 ret = notification_thread_client_disconnect(
3036 client, state);
3037 if (ret) {
3038 error_encoutered = true;
3039 }
3040 }
3041 rcu_read_unlock();
3042 return error_encoutered ? 1 : 0;
3043 }
3044
3045 int handle_notification_thread_trigger_unregister_all(
3046 struct notification_thread_state *state)
3047 {
3048 bool error_occurred = false;
3049 struct cds_lfht_iter iter;
3050 struct lttng_trigger_ht_element *trigger_ht_element;
3051
3052 rcu_read_lock();
3053 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
3054 node) {
3055 int ret = handle_notification_thread_command_unregister_trigger(
3056 state, trigger_ht_element->trigger, NULL);
3057 if (ret) {
3058 error_occurred = true;
3059 }
3060 }
3061 rcu_read_unlock();
3062 return error_occurred ? -1 : 0;
3063 }
3064
3065 static
3066 int client_handle_transmission_status(
3067 struct notification_client *client,
3068 enum client_transmission_status transmission_status,
3069 struct notification_thread_state *state)
3070 {
3071 int ret = 0;
3072
3073 switch (transmission_status) {
3074 case CLIENT_TRANSMISSION_STATUS_COMPLETE:
3075 ret = lttng_poll_mod(&state->events, client->socket,
3076 CLIENT_POLL_MASK_IN);
3077 if (ret) {
3078 goto end;
3079 }
3080
3081 break;
3082 case CLIENT_TRANSMISSION_STATUS_QUEUED:
3083 /*
3084 * We want to be notified whenever there is buffer space
3085 * available to send the rest of the payload.
3086 */
3087 ret = lttng_poll_mod(&state->events, client->socket,
3088 CLIENT_POLL_MASK_IN_OUT);
3089 if (ret) {
3090 goto end;
3091 }
3092 break;
3093 case CLIENT_TRANSMISSION_STATUS_FAIL:
3094 ret = notification_thread_client_disconnect(client, state);
3095 if (ret) {
3096 goto end;
3097 }
3098 break;
3099 case CLIENT_TRANSMISSION_STATUS_ERROR:
3100 ret = -1;
3101 goto end;
3102 default:
3103 abort();
3104 }
3105 end:
3106 return ret;
3107 }
3108
3109 /* Client lock must be acquired by caller. */
3110 static
3111 enum client_transmission_status client_flush_outgoing_queue(
3112 struct notification_client *client)
3113 {
3114 ssize_t ret;
3115 size_t to_send_count;
3116 enum client_transmission_status status;
3117 struct lttng_payload_view pv = lttng_payload_view_from_payload(
3118 &client->communication.outbound.payload, 0, -1);
3119 const int fds_to_send_count =
3120 lttng_payload_view_get_fd_handle_count(&pv);
3121
3122 ASSERT_LOCKED(client->lock);
3123
3124 if (!client->communication.active) {
3125 status = CLIENT_TRANSMISSION_STATUS_FAIL;
3126 goto end;
3127 }
3128
3129 if (pv.buffer.size == 0) {
3130 /*
3131 * If both data and fds are equal to zero, we are in an invalid
3132 * state.
3133 */
3134 assert(fds_to_send_count != 0);
3135 goto send_fds;
3136 }
3137
3138 /* Send data. */
3139 to_send_count = pv.buffer.size;
3140 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
3141 client->socket);
3142
3143 ret = lttcomm_send_unix_sock_non_block(client->socket,
3144 pv.buffer.data,
3145 to_send_count);
3146 if ((ret >= 0 && ret < to_send_count)) {
3147 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
3148 client->socket);
3149 to_send_count -= max(ret, 0);
3150
3151 memmove(client->communication.outbound.payload.buffer.data,
3152 pv.buffer.data +
3153 pv.buffer.size - to_send_count,
3154 to_send_count);
3155 ret = lttng_dynamic_buffer_set_size(
3156 &client->communication.outbound.payload.buffer,
3157 to_send_count);
3158 if (ret) {
3159 goto error;
3160 }
3161
3162 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
3163 goto end;
3164 } else if (ret < 0) {
3165 /* Generic error, disable the client's communication. */
3166 ERR("[notification-thread] Failed to flush outgoing queue, disconnecting client (socket fd = %i)",
3167 client->socket);
3168 client->communication.active = false;
3169 status = CLIENT_TRANSMISSION_STATUS_FAIL;
3170 goto end;
3171 } else {
3172 /*
3173 * No error and flushed the queue completely.
3174 *
3175 * The payload buffer size is used later to
3176 * check if there is notifications queued. So albeit that the
3177 * direct caller knows that the transmission is complete, we
3178 * need to set the buffer size to zero.
3179 */
3180 ret = lttng_dynamic_buffer_set_size(
3181 &client->communication.outbound.payload.buffer, 0);
3182 if (ret) {
3183 goto error;
3184 }
3185 }
3186
3187 send_fds:
3188 /* No fds to send, transmission is complete. */
3189 if (fds_to_send_count == 0) {
3190 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
3191 goto end;
3192 }
3193
3194 ret = lttcomm_send_payload_view_fds_unix_sock_non_block(
3195 client->socket, &pv);
3196 if (ret < 0) {
3197 /* Generic error, disable the client's communication. */
3198 ERR("[notification-thread] Failed to flush outgoing fds queue, disconnecting client (socket fd = %i)",
3199 client->socket);
3200 client->communication.active = false;
3201 status = CLIENT_TRANSMISSION_STATUS_FAIL;
3202 goto end;
3203 } else if (ret == 0) {
3204 /* Nothing could be sent. */
3205 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
3206 } else {
3207 /* Fd passing is an all or nothing kind of thing. */
3208 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
3209 /*
3210 * The payload _fd_array count is used later to
3211 * check if there is notifications queued. So although the
3212 * direct caller knows that the transmission is complete, we
3213 * need to clear the _fd_array for the queuing check.
3214 */
3215 lttng_dynamic_pointer_array_clear(
3216 &client->communication.outbound.payload
3217 ._fd_handles);
3218 }
3219
3220 end:
3221 if (status == CLIENT_TRANSMISSION_STATUS_COMPLETE) {
3222 client->communication.outbound.queued_command_reply = false;
3223 client->communication.outbound.dropped_notification = false;
3224 lttng_payload_clear(&client->communication.outbound.payload);
3225 }
3226
3227 return status;
3228 error:
3229 return CLIENT_TRANSMISSION_STATUS_ERROR;
3230 }
3231
3232 static
3233 bool client_has_outbound_data_left(
3234 const struct notification_client *client)
3235 {
3236 const struct lttng_payload_view pv = lttng_payload_view_from_payload(
3237 &client->communication.outbound.payload, 0, -1);
3238 const bool has_data = pv.buffer.size != 0;
3239 const bool has_fds = lttng_payload_view_get_fd_handle_count(&pv);
3240
3241 return has_data || has_fds;
3242 }
3243
3244 /* Client lock must _not_ be held by the caller. */
3245 static
3246 int client_send_command_reply(struct notification_client *client,
3247 struct notification_thread_state *state,
3248 enum lttng_notification_channel_status status)
3249 {
3250 int ret;
3251 struct lttng_notification_channel_command_reply reply = {
3252 .status = (int8_t) status,
3253 };
3254 struct lttng_notification_channel_message msg = {
3255 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
3256 .size = sizeof(reply),
3257 };
3258 char buffer[sizeof(msg) + sizeof(reply)];
3259 enum client_transmission_status transmission_status;
3260
3261 memcpy(buffer, &msg, sizeof(msg));
3262 memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
3263 DBG("[notification-thread] Send command reply (%i)", (int) status);
3264
3265 pthread_mutex_lock(&client->lock);
3266 if (client->communication.outbound.queued_command_reply) {
3267 /* Protocol error. */
3268 goto error_unlock;
3269 }
3270
3271 /* Enqueue buffer to outgoing queue and flush it. */
3272 ret = lttng_dynamic_buffer_append(
3273 &client->communication.outbound.payload.buffer,
3274 buffer, sizeof(buffer));
3275 if (ret) {
3276 goto error_unlock;
3277 }
3278
3279 transmission_status = client_flush_outgoing_queue(client);
3280
3281 if (client_has_outbound_data_left(client)) {
3282 /* Queue could not be emptied. */
3283 client->communication.outbound.queued_command_reply = true;
3284 }
3285
3286 pthread_mutex_unlock(&client->lock);
3287 ret = client_handle_transmission_status(
3288 client, transmission_status, state);
3289 if (ret) {
3290 goto error;
3291 }
3292
3293 return 0;
3294 error_unlock:
3295 pthread_mutex_unlock(&client->lock);
3296 error:
3297 return -1;
3298 }
3299
3300 static
3301 int client_handle_message_unknown(struct notification_client *client,
3302 struct notification_thread_state *state)
3303 {
3304 int ret;
3305 /*
3306 * Receiving message header. The function will be called again
3307 * once the rest of the message as been received and can be
3308 * interpreted.
3309 */
3310 const struct lttng_notification_channel_message *msg;
3311
3312 assert(sizeof(*msg) == client->communication.inbound.payload.buffer.size);
3313 msg = (const struct lttng_notification_channel_message *)
3314 client->communication.inbound.payload.buffer.data;
3315
3316 if (msg->size == 0 ||
3317 msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
3318 ERR("[notification-thread] Invalid notification channel message: length = %u",
3319 msg->size);
3320 ret = -1;
3321 goto end;
3322 }
3323
3324 switch (msg->type) {
3325 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
3326 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
3327 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
3328 break;
3329 default:
3330 ret = -1;
3331 ERR("[notification-thread] Invalid notification channel message: unexpected message type");
3332 goto end;
3333 }
3334
3335 client->communication.inbound.bytes_to_receive = msg->size;
3336 client->communication.inbound.fds_to_receive = msg->fds;
3337 client->communication.inbound.msg_type =
3338 (enum lttng_notification_channel_message_type) msg->type;
3339 ret = lttng_dynamic_buffer_set_size(
3340 &client->communication.inbound.payload.buffer, msg->size);
3341
3342 /* msg is not valid anymore due to lttng_dynamic_buffer_set_size. */
3343 msg = NULL;
3344 end:
3345 return ret;
3346 }
3347
3348 static
3349 int client_handle_message_handshake(struct notification_client *client,
3350 struct notification_thread_state *state)
3351 {
3352 int ret;
3353 struct lttng_notification_channel_command_handshake *handshake_client;
3354 const struct lttng_notification_channel_command_handshake handshake_reply = {
3355 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
3356 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
3357 };
3358 const struct lttng_notification_channel_message msg_header = {
3359 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
3360 .size = sizeof(handshake_reply),
3361 };
3362 enum lttng_notification_channel_status status =
3363 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
3364 char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
3365
3366 memcpy(send_buffer, &msg_header, sizeof(msg_header));
3367 memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
3368 sizeof(handshake_reply));
3369
3370 handshake_client =
3371 (struct lttng_notification_channel_command_handshake *)
3372 client->communication.inbound.payload.buffer
3373 .data;
3374 client->major = handshake_client->major;
3375 client->minor = handshake_client->minor;
3376 if (!client->communication.inbound.creds_received) {
3377 ERR("[notification-thread] No credentials received from client");
3378 ret = -1;
3379 goto end;
3380 }
3381
3382 client->uid = LTTNG_SOCK_GET_UID_CRED(
3383 &client->communication.inbound.creds);
3384 client->gid = LTTNG_SOCK_GET_GID_CRED(
3385 &client->communication.inbound.creds);
3386 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
3387 client->uid, client->gid, (int) client->major,
3388 (int) client->minor);
3389
3390 if (handshake_client->major !=
3391 LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
3392 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION;
3393 }
3394
3395 pthread_mutex_lock(&client->lock);
3396 /* Outgoing queue will be flushed when the command reply is sent. */
3397 ret = lttng_dynamic_buffer_append(
3398 &client->communication.outbound.payload.buffer, send_buffer,
3399 sizeof(send_buffer));
3400 if (ret) {
3401 ERR("[notification-thread] Failed to send protocol version to notification channel client");
3402 goto end_unlock;
3403 }
3404
3405 client->validated = true;
3406 client->communication.active = true;
3407 pthread_mutex_unlock(&client->lock);
3408
3409 /* Set reception state to receive the next message header. */
3410 ret = client_reset_inbound_state(client);
3411 if (ret) {
3412 ERR("[notification-thread] Failed to reset client communication's inbound state");
3413 goto end;
3414 }
3415
3416 /* Flushes the outgoing queue. */
3417 ret = client_send_command_reply(client, state, status);
3418 if (ret) {
3419 ERR("[notification-thread] Failed to send reply to notification channel client");
3420 goto end;
3421 }
3422
3423 goto end;
3424 end_unlock:
3425 pthread_mutex_unlock(&client->lock);
3426 end:
3427 return ret;
3428 }
3429
3430 static
3431 int client_handle_message_subscription(
3432 struct notification_client *client,
3433 enum lttng_notification_channel_message_type msg_type,
3434 struct notification_thread_state *state)
3435 {
3436 int ret;
3437 struct lttng_condition *condition;
3438 enum lttng_notification_channel_status status =
3439 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
3440 struct lttng_payload_view condition_view =
3441 lttng_payload_view_from_payload(
3442 &client->communication.inbound.payload,
3443 0, -1);
3444 size_t expected_condition_size;
3445
3446 /*
3447 * No need to lock client to sample the inbound state as the only
3448 * other thread accessing clients (action executor) only uses the
3449 * outbound state.
3450 */
3451 expected_condition_size = client->communication.inbound.payload.buffer.size;
3452 ret = lttng_condition_create_from_payload(&condition_view, &condition);
3453 if (ret != expected_condition_size) {
3454 ERR("[notification-thread] Malformed condition received from client");
3455 goto end;
3456 }
3457
3458 if (msg_type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE) {
3459 ret = notification_thread_client_subscribe(
3460 client, condition, state, &status);
3461 } else {
3462 ret = notification_thread_client_unsubscribe(
3463 client, condition, state, &status);
3464 }
3465
3466 if (ret) {
3467 goto end;
3468 }
3469
3470 /* Set reception state to receive the next message header. */
3471 ret = client_reset_inbound_state(client);
3472 if (ret) {
3473 ERR("[notification-thread] Failed to reset client communication's inbound state");
3474 goto end;
3475 }
3476
3477 ret = client_send_command_reply(client, state, status);
3478 if (ret) {
3479 ERR("[notification-thread] Failed to send reply to notification channel client");
3480 goto end;
3481 }
3482
3483 end:
3484 return ret;
3485 }
3486
3487 static
3488 int client_dispatch_message(struct notification_client *client,