2 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 * SPDX-License-Identifier: GPL-2.0-only
8 #include "lttng/action/action.h"
9 #include "lttng/trigger/trigger-internal.h"
12 #include <urcu/rculfhash.h>
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>
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"
48 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
49 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
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
,
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
;
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
;
75 * List of triggers applying to a given session.
78 * - lttng_session_trigger_list_create()
79 * - lttng_session_trigger_list_build()
80 * - lttng_session_trigger_list_destroy()
81 * - lttng_session_trigger_list_add()
83 struct lttng_session_trigger_list
{
85 * Not owned by this; points to the session_info structure's
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
;
94 * Weak reference to the notification system's session triggers
97 * The session trigger list structure structure is owned by
98 * the session's session_info.
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.
107 * This implies that the session_triggers_ht must be destroyed
108 * after the channels.
110 struct cds_lfht
*session_triggers_ht
;
111 /* Used for delayed RCU reclaim. */
112 struct rcu_head rcu_node
;
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
;
123 struct lttng_condition_list_element
{
124 struct lttng_condition
*condition
;
125 struct cds_list_head node
;
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
;
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
);
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
);
155 /* session_info API */
157 void session_info_destroy(void *_data
);
159 void session_info_get(struct session_info
*session_info
);
161 void session_info_put(struct session_info
*session_info
);
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
);
168 void session_info_add_channel(struct session_info
*session_info
,
169 struct channel_info
*channel_info
);
171 void session_info_remove_channel(struct session_info
*session_info
,
172 struct channel_info
*channel_info
);
174 /* lttng_session_trigger_list API */
176 struct lttng_session_trigger_list
*lttng_session_trigger_list_create(
177 const char *session_name
,
178 struct cds_lfht
*session_triggers_ht
);
180 struct lttng_session_trigger_list
*lttng_session_trigger_list_build(
181 const struct notification_thread_state
*state
,
182 const char *session_name
);
184 void lttng_session_trigger_list_destroy(
185 struct lttng_session_trigger_list
*list
);
187 int lttng_session_trigger_list_add(struct lttng_session_trigger_list
*list
,
188 struct lttng_trigger
*trigger
);
191 int client_handle_transmission_status(
192 struct notification_client
*client
,
193 enum client_transmission_status transmission_status
,
194 struct notification_thread_state
*state
);
197 void free_lttng_trigger_ht_element_rcu(struct rcu_head
*node
);
200 int match_client_socket(struct cds_lfht_node
*node
, const void *key
)
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
);
207 return client
->socket
== socket
;
211 int match_client_id(struct cds_lfht_node
*node
, const void *key
)
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
);
218 return client
->id
== id
;
222 int match_channel_trigger_list(struct cds_lfht_node
*node
, const void *key
)
224 struct channel_key
*channel_key
= (struct channel_key
*) key
;
225 struct lttng_channel_trigger_list
*trigger_list
;
227 trigger_list
= caa_container_of(node
, struct lttng_channel_trigger_list
,
228 channel_triggers_ht_node
);
230 return !!((channel_key
->key
== trigger_list
->channel_key
.key
) &&
231 (channel_key
->domain
== trigger_list
->channel_key
.domain
));
235 int match_session_trigger_list(struct cds_lfht_node
*node
, const void *key
)
237 const char *session_name
= (const char *) key
;
238 struct lttng_session_trigger_list
*trigger_list
;
240 trigger_list
= caa_container_of(node
, struct lttng_session_trigger_list
,
241 session_triggers_ht_node
);
243 return !!(strcmp(trigger_list
->session_name
, session_name
) == 0);
247 int match_channel_state_sample(struct cds_lfht_node
*node
, const void *key
)
249 struct channel_key
*channel_key
= (struct channel_key
*) key
;
250 struct channel_state_sample
*sample
;
252 sample
= caa_container_of(node
, struct channel_state_sample
,
253 channel_state_ht_node
);
255 return !!((channel_key
->key
== sample
->key
.key
) &&
256 (channel_key
->domain
== sample
->key
.domain
));
260 int match_channel_info(struct cds_lfht_node
*node
, const void *key
)
262 struct channel_key
*channel_key
= (struct channel_key
*) key
;
263 struct channel_info
*channel_info
;
265 channel_info
= caa_container_of(node
, struct channel_info
,
268 return !!((channel_key
->key
== channel_info
->key
.key
) &&
269 (channel_key
->domain
== channel_info
->key
.domain
));
273 int match_trigger(struct cds_lfht_node
*node
, const void *key
)
275 struct lttng_trigger
*trigger_key
= (struct lttng_trigger
*) key
;
276 struct lttng_trigger_ht_element
*trigger_ht_element
;
278 trigger_ht_element
= caa_container_of(node
, struct lttng_trigger_ht_element
,
281 return !!lttng_trigger_is_equal(trigger_key
, trigger_ht_element
->trigger
);
285 int match_trigger_token(struct cds_lfht_node
*node
, const void *key
)
287 const uint64_t *_key
= key
;
288 struct notification_trigger_tokens_ht_element
*element
;
290 element
= caa_container_of(node
,
291 struct notification_trigger_tokens_ht_element
, node
);
292 return *_key
== element
->token
;
296 int match_client_list_condition(struct cds_lfht_node
*node
, const void *key
)
298 struct lttng_condition
*condition_key
= (struct lttng_condition
*) key
;
299 struct notification_client_list
*client_list
;
300 const struct lttng_condition
*condition
;
302 assert(condition_key
);
304 client_list
= caa_container_of(node
, struct notification_client_list
,
305 notification_trigger_clients_ht_node
);
306 condition
= lttng_trigger_get_const_condition(client_list
->trigger
);
308 return !!lttng_condition_is_equal(condition_key
, condition
);
312 int match_session(struct cds_lfht_node
*node
, const void *key
)
314 const char *name
= key
;
315 struct session_info
*session_info
= caa_container_of(
316 node
, struct session_info
, sessions_ht_node
);
318 return !strcmp(session_info
->name
, name
);
322 const char *notification_command_type_str(
323 enum notification_thread_command_type type
)
326 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER
:
327 return "REGISTER_TRIGGER";
328 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER
:
329 return "UNREGISTER_TRIGGER";
330 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL
:
331 return "ADD_CHANNEL";
332 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL
:
333 return "REMOVE_CHANNEL";
334 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
:
335 return "SESSION_ROTATION_ONGOING";
336 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED
:
337 return "SESSION_ROTATION_COMPLETED";
338 case NOTIFICATION_COMMAND_TYPE_ADD_TRACER_EVENT_SOURCE
:
339 return "ADD_TRACER_EVENT_SOURCE";
340 case NOTIFICATION_COMMAND_TYPE_REMOVE_TRACER_EVENT_SOURCE
:
341 return "REMOVE_TRACER_EVENT_SOURCE";
342 case NOTIFICATION_COMMAND_TYPE_LIST_TRIGGERS
:
343 return "LIST_TRIGGERS";
344 case NOTIFICATION_COMMAND_TYPE_QUIT
:
346 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE
:
347 return "CLIENT_COMMUNICATION_UPDATE";
354 * Match trigger based on name and credentials only.
355 * Name duplication is NOT allowed for the same uid.
358 int match_trigger_by_name_uid(struct cds_lfht_node
*node
,
363 const char *key_name
;
364 enum lttng_trigger_status status
;
365 const struct lttng_credentials
*key_creds
;
366 const struct lttng_credentials
*node_creds
;
367 const struct lttng_trigger
*trigger_key
=
368 (const struct lttng_trigger
*) key
;
369 const struct lttng_trigger_ht_element
*trigger_ht_element
=
370 caa_container_of(node
,
371 struct lttng_trigger_ht_element
,
374 status
= lttng_trigger_get_name(trigger_ht_element
->trigger
, &name
);
375 assert(status
== LTTNG_TRIGGER_STATUS_OK
);
377 status
= lttng_trigger_get_name(trigger_key
, &key_name
);
378 assert(status
== LTTNG_TRIGGER_STATUS_OK
);
380 /* Compare the names. */
381 if (strcmp(name
, key_name
) != 0) {
385 /* Compare the owners' UIDs. */
386 key_creds
= lttng_trigger_get_credentials(trigger_key
);
387 node_creds
= lttng_trigger_get_credentials(trigger_ht_element
->trigger
);
389 match
= lttng_credentials_is_equal_uid(key_creds
, node_creds
);
396 * Hash trigger based on name and credentials only.
399 unsigned long hash_trigger_by_name_uid(const struct lttng_trigger
*trigger
)
401 unsigned long hash
= 0;
402 const struct lttng_credentials
*trigger_creds
;
403 const char *trigger_name
;
404 enum lttng_trigger_status status
;
406 status
= lttng_trigger_get_name(trigger
, &trigger_name
);
407 if (status
== LTTNG_TRIGGER_STATUS_OK
) {
408 hash
= hash_key_str(trigger_name
, lttng_ht_seed
);
411 trigger_creds
= lttng_trigger_get_credentials(trigger
);
412 hash
^= hash_key_ulong((void *) (unsigned long) LTTNG_OPTIONAL_GET(trigger_creds
->uid
),
419 unsigned long hash_channel_key(struct channel_key
*key
)
421 unsigned long key_hash
= hash_key_u64(&key
->key
, lttng_ht_seed
);
422 unsigned long domain_hash
= hash_key_ulong(
423 (void *) (unsigned long) key
->domain
, lttng_ht_seed
);
425 return key_hash
^ domain_hash
;
429 unsigned long hash_client_socket(int socket
)
431 return hash_key_ulong((void *) (unsigned long) socket
, lttng_ht_seed
);
435 unsigned long hash_client_id(notification_client_id id
)
437 return hash_key_u64(&id
, lttng_ht_seed
);
441 * Get the type of object to which a given condition applies. Bindings let
442 * the notification system evaluate a trigger's condition when a given
443 * object's state is updated.
445 * For instance, a condition bound to a channel will be evaluated everytime
446 * the channel's state is changed by a channel monitoring sample.
449 enum lttng_object_type
get_condition_binding_object(
450 const struct lttng_condition
*condition
)
452 switch (lttng_condition_get_type(condition
)) {
453 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
454 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
455 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
456 return LTTNG_OBJECT_TYPE_CHANNEL
;
457 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
458 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
459 return LTTNG_OBJECT_TYPE_SESSION
;
460 case LTTNG_CONDITION_TYPE_EVENT_RULE_HIT
:
461 return LTTNG_OBJECT_TYPE_NONE
;
463 return LTTNG_OBJECT_TYPE_UNKNOWN
;
468 void free_channel_info_rcu(struct rcu_head
*node
)
470 free(caa_container_of(node
, struct channel_info
, rcu_node
));
474 void channel_info_destroy(struct channel_info
*channel_info
)
480 if (channel_info
->session_info
) {
481 session_info_remove_channel(channel_info
->session_info
,
483 session_info_put(channel_info
->session_info
);
485 if (channel_info
->name
) {
486 free(channel_info
->name
);
488 call_rcu(&channel_info
->rcu_node
, free_channel_info_rcu
);
492 void free_session_info_rcu(struct rcu_head
*node
)
494 free(caa_container_of(node
, struct session_info
, rcu_node
));
497 /* Don't call directly, use the ref-counting mechanism. */
499 void session_info_destroy(void *_data
)
501 struct session_info
*session_info
= _data
;
504 assert(session_info
);
505 if (session_info
->channel_infos_ht
) {
506 ret
= cds_lfht_destroy(session_info
->channel_infos_ht
, NULL
);
508 ERR("[notification-thread] Failed to destroy channel information hash table");
511 lttng_session_trigger_list_destroy(session_info
->trigger_list
);
514 cds_lfht_del(session_info
->sessions_ht
,
515 &session_info
->sessions_ht_node
);
517 free(session_info
->name
);
518 call_rcu(&session_info
->rcu_node
, free_session_info_rcu
);
522 void session_info_get(struct session_info
*session_info
)
527 lttng_ref_get(&session_info
->ref
);
531 void session_info_put(struct session_info
*session_info
)
536 lttng_ref_put(&session_info
->ref
);
540 struct session_info
*session_info_create(const char *name
, uid_t uid
, gid_t gid
,
541 struct lttng_session_trigger_list
*trigger_list
,
542 struct cds_lfht
*sessions_ht
)
544 struct session_info
*session_info
;
548 session_info
= zmalloc(sizeof(*session_info
));
552 lttng_ref_init(&session_info
->ref
, session_info_destroy
);
554 session_info
->channel_infos_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
555 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
556 if (!session_info
->channel_infos_ht
) {
560 cds_lfht_node_init(&session_info
->sessions_ht_node
);
561 session_info
->name
= strdup(name
);
562 if (!session_info
->name
) {
565 session_info
->uid
= uid
;
566 session_info
->gid
= gid
;
567 session_info
->trigger_list
= trigger_list
;
568 session_info
->sessions_ht
= sessions_ht
;
572 session_info_put(session_info
);
577 void session_info_add_channel(struct session_info
*session_info
,
578 struct channel_info
*channel_info
)
581 cds_lfht_add(session_info
->channel_infos_ht
,
582 hash_channel_key(&channel_info
->key
),
583 &channel_info
->session_info_channels_ht_node
);
588 void session_info_remove_channel(struct session_info
*session_info
,
589 struct channel_info
*channel_info
)
592 cds_lfht_del(session_info
->channel_infos_ht
,
593 &channel_info
->session_info_channels_ht_node
);
598 struct channel_info
*channel_info_create(const char *channel_name
,
599 struct channel_key
*channel_key
, uint64_t channel_capacity
,
600 struct session_info
*session_info
)
602 struct channel_info
*channel_info
= zmalloc(sizeof(*channel_info
));
608 cds_lfht_node_init(&channel_info
->channels_ht_node
);
609 cds_lfht_node_init(&channel_info
->session_info_channels_ht_node
);
610 memcpy(&channel_info
->key
, channel_key
, sizeof(*channel_key
));
611 channel_info
->capacity
= channel_capacity
;
613 channel_info
->name
= strdup(channel_name
);
614 if (!channel_info
->name
) {
619 * Set the references between session and channel infos:
620 * - channel_info holds a strong reference to session_info
621 * - session_info holds a weak reference to channel_info
623 session_info_get(session_info
);
624 session_info_add_channel(session_info
, channel_info
);
625 channel_info
->session_info
= session_info
;
629 channel_info_destroy(channel_info
);
634 bool notification_client_list_get(struct notification_client_list
*list
)
636 return urcu_ref_get_unless_zero(&list
->ref
);
640 void free_notification_client_list_rcu(struct rcu_head
*node
)
642 free(caa_container_of(node
, struct notification_client_list
,
647 void notification_client_list_release(struct urcu_ref
*list_ref
)
649 struct notification_client_list
*list
=
650 container_of(list_ref
, typeof(*list
), ref
);
651 struct notification_client_list_element
*client_list_element
, *tmp
;
653 if (list
->notification_trigger_clients_ht
) {
655 cds_lfht_del(list
->notification_trigger_clients_ht
,
656 &list
->notification_trigger_clients_ht_node
);
658 list
->notification_trigger_clients_ht
= NULL
;
660 cds_list_for_each_entry_safe(client_list_element
, tmp
,
662 free(client_list_element
);
664 pthread_mutex_destroy(&list
->lock
);
665 call_rcu(&list
->rcu_node
, free_notification_client_list_rcu
);
669 struct notification_client_list
*notification_client_list_create(
670 const struct lttng_trigger
*trigger
)
672 struct notification_client_list
*client_list
=
673 zmalloc(sizeof(*client_list
));
678 pthread_mutex_init(&client_list
->lock
, NULL
);
679 urcu_ref_init(&client_list
->ref
);
680 cds_lfht_node_init(&client_list
->notification_trigger_clients_ht_node
);
681 CDS_INIT_LIST_HEAD(&client_list
->list
);
682 client_list
->trigger
= trigger
;
688 void publish_notification_client_list(
689 struct notification_thread_state
*state
,
690 struct notification_client_list
*list
)
692 const struct lttng_condition
*condition
=
693 lttng_trigger_get_const_condition(list
->trigger
);
695 assert(!list
->notification_trigger_clients_ht
);
696 notification_client_list_get(list
);
698 list
->notification_trigger_clients_ht
=
699 state
->notification_trigger_clients_ht
;
702 cds_lfht_add(state
->notification_trigger_clients_ht
,
703 lttng_condition_hash(condition
),
704 &list
->notification_trigger_clients_ht_node
);
709 void notification_client_list_put(struct notification_client_list
*list
)
714 return urcu_ref_put(&list
->ref
, notification_client_list_release
);
717 /* Provides a reference to the returned list. */
719 struct notification_client_list
*get_client_list_from_condition(
720 struct notification_thread_state
*state
,
721 const struct lttng_condition
*condition
)
723 struct cds_lfht_node
*node
;
724 struct cds_lfht_iter iter
;
725 struct notification_client_list
*list
= NULL
;
728 cds_lfht_lookup(state
->notification_trigger_clients_ht
,
729 lttng_condition_hash(condition
),
730 match_client_list_condition
,
733 node
= cds_lfht_iter_get_node(&iter
);
735 list
= container_of(node
, struct notification_client_list
,
736 notification_trigger_clients_ht_node
);
737 list
= notification_client_list_get(list
) ? list
: NULL
;
745 int evaluate_channel_condition_for_client(
746 const struct lttng_condition
*condition
,
747 struct notification_thread_state
*state
,
748 struct lttng_evaluation
**evaluation
,
749 uid_t
*session_uid
, gid_t
*session_gid
)
752 struct cds_lfht_iter iter
;
753 struct cds_lfht_node
*node
;
754 struct channel_info
*channel_info
= NULL
;
755 struct channel_key
*channel_key
= NULL
;
756 struct channel_state_sample
*last_sample
= NULL
;
757 struct lttng_channel_trigger_list
*channel_trigger_list
= NULL
;
761 /* Find the channel associated with the condition. */
762 cds_lfht_for_each_entry(state
->channel_triggers_ht
, &iter
,
763 channel_trigger_list
, channel_triggers_ht_node
) {
764 struct lttng_trigger_list_element
*element
;
766 cds_list_for_each_entry(element
, &channel_trigger_list
->list
, node
) {
767 const struct lttng_condition
*current_condition
=
768 lttng_trigger_get_const_condition(
771 assert(current_condition
);
772 if (!lttng_condition_is_equal(condition
,
773 current_condition
)) {
777 /* Found the trigger, save the channel key. */
778 channel_key
= &channel_trigger_list
->channel_key
;
782 /* The channel key was found stop iteration. */
788 /* No channel found; normal exit. */
789 DBG("[notification-thread] No known channel associated with newly subscribed-to condition");
794 /* Fetch channel info for the matching channel. */
795 cds_lfht_lookup(state
->channels_ht
,
796 hash_channel_key(channel_key
),
800 node
= cds_lfht_iter_get_node(&iter
);
802 channel_info
= caa_container_of(node
, struct channel_info
,
805 /* Retrieve the channel's last sample, if it exists. */
806 cds_lfht_lookup(state
->channel_state_ht
,
807 hash_channel_key(channel_key
),
808 match_channel_state_sample
,
811 node
= cds_lfht_iter_get_node(&iter
);
813 last_sample
= caa_container_of(node
,
814 struct channel_state_sample
,
815 channel_state_ht_node
);
817 /* Nothing to evaluate, no sample was ever taken. Normal exit */
818 DBG("[notification-thread] No channel sample associated with newly subscribed-to condition");
823 ret
= evaluate_buffer_condition(condition
, evaluation
, state
,
825 0, channel_info
->session_info
->consumed_data_size
,
828 WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition");
832 *session_uid
= channel_info
->session_info
->uid
;
833 *session_gid
= channel_info
->session_info
->gid
;
840 const char *get_condition_session_name(const struct lttng_condition
*condition
)
842 const char *session_name
= NULL
;
843 enum lttng_condition_status status
;
845 switch (lttng_condition_get_type(condition
)) {
846 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
847 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
848 status
= lttng_condition_buffer_usage_get_session_name(
849 condition
, &session_name
);
851 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
852 status
= lttng_condition_session_consumed_size_get_session_name(
853 condition
, &session_name
);
855 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
856 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
857 status
= lttng_condition_session_rotation_get_session_name(
858 condition
, &session_name
);
863 if (status
!= LTTNG_CONDITION_STATUS_OK
) {
864 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
872 int evaluate_session_condition_for_client(
873 const struct lttng_condition
*condition
,
874 struct notification_thread_state
*state
,
875 struct lttng_evaluation
**evaluation
,
876 uid_t
*session_uid
, gid_t
*session_gid
)
879 struct cds_lfht_iter iter
;
880 struct cds_lfht_node
*node
;
881 const char *session_name
;
882 struct session_info
*session_info
= NULL
;
885 session_name
= get_condition_session_name(condition
);
887 /* Find the session associated with the trigger. */
888 cds_lfht_lookup(state
->sessions_ht
,
889 hash_key_str(session_name
, lttng_ht_seed
),
893 node
= cds_lfht_iter_get_node(&iter
);
895 DBG("[notification-thread] No known session matching name \"%s\"",
901 session_info
= caa_container_of(node
, struct session_info
,
903 session_info_get(session_info
);
906 * Evaluation is performed in-line here since only one type of
907 * session-bound condition is handled for the moment.
909 switch (lttng_condition_get_type(condition
)) {
910 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
911 if (!session_info
->rotation
.ongoing
) {
913 goto end_session_put
;
916 *evaluation
= lttng_evaluation_session_rotation_ongoing_create(
917 session_info
->rotation
.id
);
920 ERR("[notification-thread] Failed to create session rotation ongoing evaluation for session \"%s\"",
923 goto end_session_put
;
929 goto end_session_put
;
932 *session_uid
= session_info
->uid
;
933 *session_gid
= session_info
->gid
;
936 session_info_put(session_info
);
943 int evaluate_condition_for_client(const struct lttng_trigger
*trigger
,
944 const struct lttng_condition
*condition
,
945 struct notification_client
*client
,
946 struct notification_thread_state
*state
)
949 struct lttng_evaluation
*evaluation
= NULL
;
950 struct notification_client_list client_list
= {
951 .lock
= PTHREAD_MUTEX_INITIALIZER
,
953 struct notification_client_list_element client_list_element
= { 0 };
954 uid_t object_uid
= 0;
955 gid_t object_gid
= 0;
962 switch (get_condition_binding_object(condition
)) {
963 case LTTNG_OBJECT_TYPE_SESSION
:
964 ret
= evaluate_session_condition_for_client(condition
, state
,
965 &evaluation
, &object_uid
, &object_gid
);
967 case LTTNG_OBJECT_TYPE_CHANNEL
:
968 ret
= evaluate_channel_condition_for_client(condition
, state
,
969 &evaluation
, &object_uid
, &object_gid
);
971 case LTTNG_OBJECT_TYPE_NONE
:
972 DBG("[notification-thread] Newly subscribed-to condition not bound to object, nothing to evaluate");
975 case LTTNG_OBJECT_TYPE_UNKNOWN
:
985 /* Evaluation yielded nothing. Normal exit. */
986 DBG("[notification-thread] Newly subscribed-to condition evaluated to false, nothing to report to client");
992 * Create a temporary client list with the client currently
995 cds_lfht_node_init(&client_list
.notification_trigger_clients_ht_node
);
996 CDS_INIT_LIST_HEAD(&client_list
.list
);
997 client_list
.trigger
= trigger
;
999 CDS_INIT_LIST_HEAD(&client_list_element
.node
);
1000 client_list_element
.client
= client
;
1001 cds_list_add(&client_list_element
.node
, &client_list
.list
);
1003 /* Send evaluation result to the newly-subscribed client. */
1004 DBG("[notification-thread] Newly subscribed-to condition evaluated to true, notifying client");
1005 ret
= send_evaluation_to_clients(trigger
, evaluation
, &client_list
,
1006 state
, object_uid
, object_gid
);
1013 int notification_thread_client_subscribe(struct notification_client
*client
,
1014 struct lttng_condition
*condition
,
1015 struct notification_thread_state
*state
,
1016 enum lttng_notification_channel_status
*_status
)
1019 struct notification_client_list
*client_list
= NULL
;
1020 struct lttng_condition_list_element
*condition_list_element
= NULL
;
1021 struct notification_client_list_element
*client_list_element
= NULL
;
1022 enum lttng_notification_channel_status status
=
1023 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
1026 * Ensure that the client has not already subscribed to this condition
1029 cds_list_for_each_entry(condition_list_element
, &client
->condition_list
, node
) {
1030 if (lttng_condition_is_equal(condition_list_element
->condition
,
1032 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED
;
1037 condition_list_element
= zmalloc(sizeof(*condition_list_element
));
1038 if (!condition_list_element
) {
1042 client_list_element
= zmalloc(sizeof(*client_list_element
));
1043 if (!client_list_element
) {
1049 * Add the newly-subscribed condition to the client's subscription list.
1051 CDS_INIT_LIST_HEAD(&condition_list_element
->node
);
1052 condition_list_element
->condition
= condition
;
1053 cds_list_add(&condition_list_element
->node
, &client
->condition_list
);
1055 client_list
= get_client_list_from_condition(state
, condition
);
1058 * No notification-emiting trigger registered with this
1059 * condition. We don't evaluate the condition right away
1060 * since this trigger is not registered yet.
1062 free(client_list_element
);
1067 * The condition to which the client just subscribed is evaluated
1068 * at this point so that conditions that are already TRUE result
1069 * in a notification being sent out.
1071 * The client_list's trigger is used without locking the list itself.
1072 * This is correct since the list doesn't own the trigger and the
1073 * object is immutable.
1075 if (evaluate_condition_for_client(client_list
->trigger
, condition
,
1077 WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
1079 free(client_list_element
);
1084 * Add the client to the list of clients interested in a given trigger
1085 * if a "notification" trigger with a corresponding condition was
1088 client_list_element
->client
= client
;
1089 CDS_INIT_LIST_HEAD(&client_list_element
->node
);
1091 pthread_mutex_lock(&client_list
->lock
);
1092 cds_list_add(&client_list_element
->node
, &client_list
->list
);
1093 pthread_mutex_unlock(&client_list
->lock
);
1099 notification_client_list_put(client_list
);
1103 free(condition_list_element
);
1104 free(client_list_element
);
1109 int notification_thread_client_unsubscribe(
1110 struct notification_client
*client
,
1111 struct lttng_condition
*condition
,
1112 struct notification_thread_state
*state
,
1113 enum lttng_notification_channel_status
*_status
)
1115 struct notification_client_list
*client_list
;
1116 struct lttng_condition_list_element
*condition_list_element
,
1118 struct notification_client_list_element
*client_list_element
,
1120 bool condition_found
= false;
1121 enum lttng_notification_channel_status status
=
1122 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
1124 /* Remove the condition from the client's condition list. */
1125 cds_list_for_each_entry_safe(condition_list_element
, condition_tmp
,
1126 &client
->condition_list
, node
) {
1127 if (!lttng_condition_is_equal(condition_list_element
->condition
,
1132 cds_list_del(&condition_list_element
->node
);
1134 * The caller may be iterating on the client's conditions to
1135 * tear down a client's connection. In this case, the condition
1136 * will be destroyed at the end.
1138 if (condition
!= condition_list_element
->condition
) {
1139 lttng_condition_destroy(
1140 condition_list_element
->condition
);
1142 free(condition_list_element
);
1143 condition_found
= true;
1147 if (!condition_found
) {
1148 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION
;
1153 * Remove the client from the list of clients interested the trigger
1154 * matching the condition.
1156 client_list
= get_client_list_from_condition(state
, condition
);
1161 pthread_mutex_lock(&client_list
->lock
);
1162 cds_list_for_each_entry_safe(client_list_element
, client_tmp
,
1163 &client_list
->list
, node
) {
1164 if (client_list_element
->client
->id
!= client
->id
) {
1167 cds_list_del(&client_list_element
->node
);
1168 free(client_list_element
);
1171 pthread_mutex_unlock(&client_list
->lock
);
1172 notification_client_list_put(client_list
);
1175 lttng_condition_destroy(condition
);
1183 void free_notification_client_rcu(struct rcu_head
*node
)
1185 free(caa_container_of(node
, struct notification_client
, rcu_node
));
1189 void notification_client_destroy(struct notification_client
*client
,
1190 struct notification_thread_state
*state
)
1197 * The client object is not reachable by other threads, no need to lock
1200 if (client
->socket
>= 0) {
1201 (void) lttcomm_close_unix_sock(client
->socket
);
1202 client
->socket
= -1;
1204 client
->communication
.active
= false;
1205 lttng_payload_reset(&client
->communication
.inbound
.payload
);
1206 lttng_payload_reset(&client
->communication
.outbound
.payload
);
1207 pthread_mutex_destroy(&client
->lock
);
1208 call_rcu(&client
->rcu_node
, free_notification_client_rcu
);
1212 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1216 struct notification_client
*get_client_from_socket(int socket
,
1217 struct notification_thread_state
*state
)
1219 struct cds_lfht_iter iter
;
1220 struct cds_lfht_node
*node
;
1221 struct notification_client
*client
= NULL
;
1223 cds_lfht_lookup(state
->client_socket_ht
,
1224 hash_client_socket(socket
),
1225 match_client_socket
,
1226 (void *) (unsigned long) socket
,
1228 node
= cds_lfht_iter_get_node(&iter
);
1233 client
= caa_container_of(node
, struct notification_client
,
1234 client_socket_ht_node
);
1240 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1244 struct notification_client
*get_client_from_id(notification_client_id id
,
1245 struct notification_thread_state
*state
)
1247 struct cds_lfht_iter iter
;
1248 struct cds_lfht_node
*node
;
1249 struct notification_client
*client
= NULL
;
1251 cds_lfht_lookup(state
->client_id_ht
,
1256 node
= cds_lfht_iter_get_node(&iter
);
1261 client
= caa_container_of(node
, struct notification_client
,
1268 bool buffer_usage_condition_applies_to_channel(
1269 const struct lttng_condition
*condition
,
1270 const struct channel_info
*channel_info
)
1272 enum lttng_condition_status status
;
1273 enum lttng_domain_type condition_domain
;
1274 const char *condition_session_name
= NULL
;
1275 const char *condition_channel_name
= NULL
;
1277 status
= lttng_condition_buffer_usage_get_domain_type(condition
,
1279 assert(status
== LTTNG_CONDITION_STATUS_OK
);
1280 if (channel_info
->key
.domain
!= condition_domain
) {
1284 status
= lttng_condition_buffer_usage_get_session_name(
1285 condition
, &condition_session_name
);
1286 assert((status
== LTTNG_CONDITION_STATUS_OK
) && condition_session_name
);
1288 status
= lttng_condition_buffer_usage_get_channel_name(
1289 condition
, &condition_channel_name
);
1290 assert((status
== LTTNG_CONDITION_STATUS_OK
) && condition_channel_name
);
1292 if (strcmp(channel_info
->session_info
->name
, condition_session_name
)) {
1295 if (strcmp(channel_info
->name
, condition_channel_name
)) {
1305 bool session_consumed_size_condition_applies_to_channel(
1306 const struct lttng_condition
*condition
,
1307 const struct channel_info
*channel_info
)
1309 enum lttng_condition_status status
;
1310 const char *condition_session_name
= NULL
;
1312 status
= lttng_condition_session_consumed_size_get_session_name(
1313 condition
, &condition_session_name
);
1314 assert((status
== LTTNG_CONDITION_STATUS_OK
) && condition_session_name
);
1316 if (strcmp(channel_info
->session_info
->name
, condition_session_name
)) {
1326 bool trigger_applies_to_channel(const struct lttng_trigger
*trigger
,
1327 const struct channel_info
*channel_info
)
1329 const struct lttng_condition
*condition
;
1330 bool trigger_applies
;
1332 condition
= lttng_trigger_get_const_condition(trigger
);
1337 switch (lttng_condition_get_type(condition
)) {
1338 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
1339 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
1340 trigger_applies
= buffer_usage_condition_applies_to_channel(
1341 condition
, channel_info
);
1343 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
1344 trigger_applies
= session_consumed_size_condition_applies_to_channel(
1345 condition
, channel_info
);
1351 return trigger_applies
;
1357 bool trigger_applies_to_client(struct lttng_trigger
*trigger
,
1358 struct notification_client
*client
)
1360 bool applies
= false;
1361 struct lttng_condition_list_element
*condition_list_element
;
1363 cds_list_for_each_entry(condition_list_element
, &client
->condition_list
,
1365 applies
= lttng_condition_is_equal(
1366 condition_list_element
->condition
,
1367 lttng_trigger_get_condition(trigger
));
1375 /* Must be called with RCU read lock held. */
1377 struct lttng_session_trigger_list
*get_session_trigger_list(
1378 struct notification_thread_state
*state
,
1379 const char *session_name
)
1381 struct lttng_session_trigger_list
*list
= NULL
;
1382 struct cds_lfht_node
*node
;
1383 struct cds_lfht_iter iter
;
1385 cds_lfht_lookup(state
->session_triggers_ht
,
1386 hash_key_str(session_name
, lttng_ht_seed
),
1387 match_session_trigger_list
,
1390 node
= cds_lfht_iter_get_node(&iter
);
1393 * Not an error, the list of triggers applying to that session
1394 * will be initialized when the session is created.
1396 DBG("[notification-thread] No trigger list found for session \"%s\" as it is not yet known to the notification system",
1401 list
= caa_container_of(node
,
1402 struct lttng_session_trigger_list
,
1403 session_triggers_ht_node
);
1409 * Allocate an empty lttng_session_trigger_list for the session named
1412 * No ownership of 'session_name' is assumed by the session trigger list.
1413 * It is the caller's responsability to ensure the session name is alive
1414 * for as long as this list is.
1417 struct lttng_session_trigger_list
*lttng_session_trigger_list_create(
1418 const char *session_name
,
1419 struct cds_lfht
*session_triggers_ht
)
1421 struct lttng_session_trigger_list
*list
;
1423 list
= zmalloc(sizeof(*list
));
1427 list
->session_name
= session_name
;
1428 CDS_INIT_LIST_HEAD(&list
->list
);
1429 cds_lfht_node_init(&list
->session_triggers_ht_node
);
1430 list
->session_triggers_ht
= session_triggers_ht
;
1433 /* Publish the list through the session_triggers_ht. */
1434 cds_lfht_add(session_triggers_ht
,
1435 hash_key_str(session_name
, lttng_ht_seed
),
1436 &list
->session_triggers_ht_node
);
1443 void free_session_trigger_list_rcu(struct rcu_head
*node
)
1445 free(caa_container_of(node
, struct lttng_session_trigger_list
,
1450 void lttng_session_trigger_list_destroy(struct lttng_session_trigger_list
*list
)
1452 struct lttng_trigger_list_element
*trigger_list_element
, *tmp
;
1454 /* Empty the list element by element, and then free the list itself. */
1455 cds_list_for_each_entry_safe(trigger_list_element
, tmp
,
1456 &list
->list
, node
) {
1457 cds_list_del(&trigger_list_element
->node
);
1458 free(trigger_list_element
);
1461 /* Unpublish the list from the session_triggers_ht. */
1462 cds_lfht_del(list
->session_triggers_ht
,
1463 &list
->session_triggers_ht_node
);
1465 call_rcu(&list
->rcu_node
, free_session_trigger_list_rcu
);
1469 int lttng_session_trigger_list_add(struct lttng_session_trigger_list
*list
,
1470 struct lttng_trigger
*trigger
)
1473 struct lttng_trigger_list_element
*new_element
=
1474 zmalloc(sizeof(*new_element
));
1480 CDS_INIT_LIST_HEAD(&new_element
->node
);
1481 new_element
->trigger
= trigger
;
1482 cds_list_add(&new_element
->node
, &list
->list
);
1488 bool trigger_applies_to_session(const struct lttng_trigger
*trigger
,
1489 const char *session_name
)
1491 bool applies
= false;
1492 const struct lttng_condition
*condition
;
1494 condition
= lttng_trigger_get_const_condition(trigger
);
1495 switch (lttng_condition_get_type(condition
)) {
1496 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
1497 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
1499 enum lttng_condition_status condition_status
;
1500 const char *condition_session_name
;
1502 condition_status
= lttng_condition_session_rotation_get_session_name(
1503 condition
, &condition_session_name
);
1504 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
1505 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
1509 assert(condition_session_name
);
1510 applies
= !strcmp(condition_session_name
, session_name
);
1521 * Allocate and initialize an lttng_session_trigger_list which contains
1522 * all triggers that apply to the session named 'session_name'.
1524 * No ownership of 'session_name' is assumed by the session trigger list.
1525 * It is the caller's responsability to ensure the session name is alive
1526 * for as long as this list is.
1529 struct lttng_session_trigger_list
*lttng_session_trigger_list_build(
1530 const struct notification_thread_state
*state
,
1531 const char *session_name
)
1533 int trigger_count
= 0;
1534 struct lttng_session_trigger_list
*session_trigger_list
= NULL
;
1535 struct lttng_trigger_ht_element
*trigger_ht_element
= NULL
;
1536 struct cds_lfht_iter iter
;
1538 session_trigger_list
= lttng_session_trigger_list_create(session_name
,
1539 state
->session_triggers_ht
);
1541 /* Add all triggers applying to the session named 'session_name'. */
1542 cds_lfht_for_each_entry(state
->triggers_ht
, &iter
, trigger_ht_element
,
1546 if (!trigger_applies_to_session(trigger_ht_element
->trigger
,
1551 ret
= lttng_session_trigger_list_add(session_trigger_list
,
1552 trigger_ht_element
->trigger
);
1560 DBG("[notification-thread] Found %i triggers that apply to newly created session",
1562 return session_trigger_list
;
1564 lttng_session_trigger_list_destroy(session_trigger_list
);
1569 struct session_info
*find_or_create_session_info(
1570 struct notification_thread_state
*state
,
1571 const char *name
, uid_t uid
, gid_t gid
)
1573 struct session_info
*session
= NULL
;
1574 struct cds_lfht_node
*node
;
1575 struct cds_lfht_iter iter
;
1576 struct lttng_session_trigger_list
*trigger_list
;
1579 cds_lfht_lookup(state
->sessions_ht
,
1580 hash_key_str(name
, lttng_ht_seed
),
1584 node
= cds_lfht_iter_get_node(&iter
);
1586 DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
1588 session
= caa_container_of(node
, struct session_info
,
1590 assert(session
->uid
== uid
);
1591 assert(session
->gid
== gid
);
1592 session_info_get(session
);
1596 trigger_list
= lttng_session_trigger_list_build(state
, name
);
1597 if (!trigger_list
) {
1601 session
= session_info_create(name
, uid
, gid
, trigger_list
,
1602 state
->sessions_ht
);
1604 ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1606 lttng_session_trigger_list_destroy(trigger_list
);
1609 trigger_list
= NULL
;
1611 cds_lfht_add(state
->sessions_ht
, hash_key_str(name
, lttng_ht_seed
),
1612 &session
->sessions_ht_node
);
1618 session_info_put(session
);
1623 int handle_notification_thread_command_add_channel(
1624 struct notification_thread_state
*state
,
1625 const char *session_name
, uid_t session_uid
, gid_t session_gid
,
1626 const char *channel_name
, enum lttng_domain_type channel_domain
,
1627 uint64_t channel_key_int
, uint64_t channel_capacity
,
1628 enum lttng_error_code
*cmd_result
)
1630 struct cds_list_head trigger_list
;
1631 struct channel_info
*new_channel_info
= NULL
;
1632 struct channel_key channel_key
= {
1633 .key
= channel_key_int
,
1634 .domain
= channel_domain
,
1636 struct lttng_channel_trigger_list
*channel_trigger_list
= NULL
;
1637 struct lttng_trigger_ht_element
*trigger_ht_element
= NULL
;
1638 int trigger_count
= 0;
1639 struct cds_lfht_iter iter
;
1640 struct session_info
*session_info
= NULL
;
1642 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64
" in %s domain",
1643 channel_name
, session_name
, channel_key_int
,
1644 lttng_domain_type_str(channel_domain
));
1646 CDS_INIT_LIST_HEAD(&trigger_list
);
1648 session_info
= find_or_create_session_info(state
, session_name
,
1649 session_uid
, session_gid
);
1650 if (!session_info
) {
1651 /* Allocation error or an internal error occurred. */
1655 new_channel_info
= channel_info_create(channel_name
, &channel_key
,
1656 channel_capacity
, session_info
);
1657 if (!new_channel_info
) {
1662 /* Build a list of all triggers applying to the new channel. */
1663 cds_lfht_for_each_entry(state
->triggers_ht
, &iter
, trigger_ht_element
,
1665 struct lttng_trigger_list_element
*new_element
;
1667 if (!trigger_applies_to_channel(trigger_ht_element
->trigger
,
1668 new_channel_info
)) {
1672 new_element
= zmalloc(sizeof(*new_element
));
1677 CDS_INIT_LIST_HEAD(&new_element
->node
);
1678 new_element
->trigger
= trigger_ht_element
->trigger
;
1679 cds_list_add(&new_element
->node
, &trigger_list
);
1684 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
1686 channel_trigger_list
= zmalloc(sizeof(*channel_trigger_list
));
1687 if (!channel_trigger_list
) {
1690 channel_trigger_list
->channel_key
= new_channel_info
->key
;
1691 CDS_INIT_LIST_HEAD(&channel_trigger_list
->list
);
1692 cds_lfht_node_init(&channel_trigger_list
->channel_triggers_ht_node
);
1693 cds_list_splice(&trigger_list
, &channel_trigger_list
->list
);
1696 /* Add channel to the channel_ht which owns the channel_infos. */
1697 cds_lfht_add(state
->channels_ht
,
1698 hash_channel_key(&new_channel_info
->key
),
1699 &new_channel_info
->channels_ht_node
);
1701 * Add the list of triggers associated with this channel to the
1702 * channel_triggers_ht.
1704 cds_lfht_add(state
->channel_triggers_ht
,
1705 hash_channel_key(&new_channel_info
->key
),
1706 &channel_trigger_list
->channel_triggers_ht_node
);
1708 session_info_put(session_info
);
1709 *cmd_result
= LTTNG_OK
;
1712 channel_info_destroy(new_channel_info
);
1713 session_info_put(session_info
);
1718 void free_channel_trigger_list_rcu(struct rcu_head
*node
)
1720 free(caa_container_of(node
, struct lttng_channel_trigger_list
,
1725 void free_channel_state_sample_rcu(struct rcu_head
*node
)
1727 free(caa_container_of(node
, struct channel_state_sample
,
1732 int handle_notification_thread_command_remove_channel(
1733 struct notification_thread_state
*state
,
1734 uint64_t channel_key
, enum lttng_domain_type domain
,
1735 enum lttng_error_code
*cmd_result
)
1737 struct cds_lfht_node
*node
;
1738 struct cds_lfht_iter iter
;
1739 struct lttng_channel_trigger_list
*trigger_list
;
1740 struct lttng_trigger_list_element
*trigger_list_element
, *tmp
;
1741 struct channel_key key
= { .key
= channel_key
, .domain
= domain
};
1742 struct channel_info
*channel_info
;
1744 DBG("[notification-thread] Removing channel key = %" PRIu64
" in %s domain",
1745 channel_key
, lttng_domain_type_str(domain
));
1749 cds_lfht_lookup(state
->channel_triggers_ht
,
1750 hash_channel_key(&key
),
1751 match_channel_trigger_list
,
1754 node
= cds_lfht_iter_get_node(&iter
);
1756 * There is a severe internal error if we are being asked to remove a
1757 * channel that doesn't exist.
1760 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
1764 /* Free the list of triggers associated with this channel. */
1765 trigger_list
= caa_container_of(node
, struct lttng_channel_trigger_list
,
1766 channel_triggers_ht_node
);
1767 cds_list_for_each_entry_safe(trigger_list_element
, tmp
,
1768 &trigger_list
->list
, node
) {
1769 cds_list_del(&trigger_list_element
->node
);
1770 free(trigger_list_element
);
1772 cds_lfht_del(state
->channel_triggers_ht
, node
);
1773 call_rcu(&trigger_list
->rcu_node
, free_channel_trigger_list_rcu
);
1775 /* Free sampled channel state. */
1776 cds_lfht_lookup(state
->channel_state_ht
,
1777 hash_channel_key(&key
),
1778 match_channel_state_sample
,
1781 node
= cds_lfht_iter_get_node(&iter
);
1783 * This is expected to be NULL if the channel is destroyed before we
1784 * received a sample.
1787 struct channel_state_sample
*sample
= caa_container_of(node
,
1788 struct channel_state_sample
,
1789 channel_state_ht_node
);
1791 cds_lfht_del(state
->channel_state_ht
, node
);
1792 call_rcu(&sample
->rcu_node
, free_channel_state_sample_rcu
);
1795 /* Remove the channel from the channels_ht and free it. */
1796 cds_lfht_lookup(state
->channels_ht
,
1797 hash_channel_key(&key
),
1801 node
= cds_lfht_iter_get_node(&iter
);
1803 channel_info
= caa_container_of(node
, struct channel_info
,
1805 cds_lfht_del(state
->channels_ht
, node
);
1806 channel_info_destroy(channel_info
);
1809 *cmd_result
= LTTNG_OK
;
1814 int handle_notification_thread_command_session_rotation(
1815 struct notification_thread_state
*state
,
1816 enum notification_thread_command_type cmd_type
,
1817 const char *session_name
, uid_t session_uid
, gid_t session_gid
,
1818 uint64_t trace_archive_chunk_id
,
1819 struct lttng_trace_archive_location
*location
,
1820 enum lttng_error_code
*_cmd_result
)
1823 enum lttng_error_code cmd_result
= LTTNG_OK
;
1824 struct lttng_session_trigger_list
*trigger_list
;
1825 struct lttng_trigger_list_element
*trigger_list_element
;
1826 struct session_info
*session_info
;
1827 const struct lttng_credentials session_creds
= {
1828 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session_uid
),
1829 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session_gid
),
1834 session_info
= find_or_create_session_info(state
, session_name
,
1835 session_uid
, session_gid
);
1836 if (!session_info
) {
1837 /* Allocation error or an internal error occurred. */
1839 cmd_result
= LTTNG_ERR_NOMEM
;
1843 session_info
->rotation
.ongoing
=
1844 cmd_type
== NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
;
1845 session_info
->rotation
.id
= trace_archive_chunk_id
;
1846 trigger_list
= get_session_trigger_list(state
, session_name
);
1847 if (!trigger_list
) {
1848 DBG("[notification-thread] No triggers applying to session \"%s\" found",
1853 cds_list_for_each_entry(trigger_list_element
, &trigger_list
->list
,
1855 const struct lttng_condition
*condition
;
1856 struct lttng_trigger
*trigger
;
1857 struct notification_client_list
*client_list
;
1858 struct lttng_evaluation
*evaluation
= NULL
;
1859 enum lttng_condition_type condition_type
;
1860 enum action_executor_status executor_status
;
1862 trigger
= trigger_list_element
->trigger
;
1863 condition
= lttng_trigger_get_const_condition(trigger
);
1865 condition_type
= lttng_condition_get_type(condition
);
1867 if (condition_type
== LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
&&
1868 cmd_type
!= NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
) {
1870 } else if (condition_type
== LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
&&
1871 cmd_type
!= NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED
) {
1875 client_list
= get_client_list_from_condition(state
, condition
);
1876 if (cmd_type
== NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
) {
1877 evaluation
= lttng_evaluation_session_rotation_ongoing_create(
1878 trace_archive_chunk_id
);
1880 evaluation
= lttng_evaluation_session_rotation_completed_create(
1881 trace_archive_chunk_id
, location
);
1885 /* Internal error */
1887 cmd_result
= LTTNG_ERR_UNK
;
1892 * Ownership of `evaluation` transferred to the action executor
1893 * no matter the result.
1895 executor_status
= action_executor_enqueue(state
->executor
,
1896 trigger
, evaluation
, &session_creds
,
1899 switch (executor_status
) {
1900 case ACTION_EXECUTOR_STATUS_OK
:
1902 case ACTION_EXECUTOR_STATUS_ERROR
:
1903 case ACTION_EXECUTOR_STATUS_INVALID
:
1905 * TODO Add trigger identification (name/id) when
1906 * it is added to the API.
1908 ERR("Fatal error occurred while enqueuing action associated with session rotation trigger");
1911 case ACTION_EXECUTOR_STATUS_OVERFLOW
:
1913 * TODO Add trigger identification (name/id) when
1914 * it is added to the API.
1916 * Not a fatal error.
1918 WARN("No space left when enqueuing action associated with session rotation trigger");
1926 notification_client_list_put(client_list
);
1927 if (caa_unlikely(ret
)) {
1932 session_info_put(session_info
);
1933 *_cmd_result
= cmd_result
;
1939 int handle_notification_thread_command_add_tracer_event_source(
1940 struct notification_thread_state
*state
,
1941 int tracer_event_source_fd
,
1942 enum lttng_domain_type domain_type
,
1943 enum lttng_error_code
*_cmd_result
)
1946 enum lttng_error_code cmd_result
= LTTNG_OK
;
1947 struct notification_event_tracer_event_source_element
*element
= NULL
;
1949 element
= zmalloc(sizeof(*element
));
1951 cmd_result
= LTTNG_ERR_NOMEM
;
1956 element
->fd
= tracer_event_source_fd
;
1957 element
->domain
= domain_type
;
1959 cds_list_add(&element
->node
, &state
->tracer_event_sources_list
);
1961 DBG3("[notification-thread] Adding tracer event source fd to poll set: tracer_event_source_fd = %d, domain = '%s'",
1962 tracer_event_source_fd
,
1963 lttng_domain_type_str(domain_type
));
1965 /* Adding the read side pipe to the event poll. */
1966 ret
= lttng_poll_add(&state
->events
, tracer_event_source_fd
, LPOLLIN
| LPOLLERR
);
1968 ERR("[notification-thread] Failed to add tracer event source to poll set: tracer_event_source_fd = %d, domain = '%s'",
1969 tracer_event_source_fd
,
1970 lttng_domain_type_str(element
->domain
));
1971 cds_list_del(&element
->node
);
1976 element
->is_fd_in_poll_set
= true;
1979 *_cmd_result
= cmd_result
;
1984 int handle_notification_thread_command_remove_tracer_event_source(
1985 struct notification_thread_state
*state
,
1986 int tracer_event_source_fd
,
1987 enum lttng_error_code
*_cmd_result
)
1991 enum lttng_error_code cmd_result
= LTTNG_OK
;
1992 struct notification_event_tracer_event_source_element
*source_element
= NULL
, *tmp
;
1994 cds_list_for_each_entry_safe(source_element
, tmp
,
1995 &state
->tracer_event_sources_list
, node
) {
1996 if (source_element
->fd
!= tracer_event_source_fd
) {
2000 DBG("[notification-thread] Removed tracer event source from poll set: tracer_event_source_fd = %d, domain = '%s'",
2001 tracer_event_source_fd
,
2002 lttng_domain_type_str(source_element
->domain
));
2003 cds_list_del(&source_element
->node
);
2010 * This is temporarily allowed since the poll activity set is
2011 * not properly cleaned-up for the moment. This is adressed in
2014 source_element
= NULL
;
2018 if (!source_element
->is_fd_in_poll_set
) {
2019 /* Skip the poll set removal. */
2023 DBG3("[notification-thread] Removing tracer event source from poll set: tracer_event_source_fd = %d, domain = '%s'",
2024 tracer_event_source_fd
,
2025 lttng_domain_type_str(source_element
->domain
));
2027 /* Removing the fd from the event poll set. */
2028 ret
= lttng_poll_del(&state
->events
, tracer_event_source_fd
);
2030 ERR("[notification-thread] Failed to remove tracer event source from poll set: tracer_event_source_fd = %d, domain = '%s'",
2031 tracer_event_source_fd
,
2032 lttng_domain_type_str(source_element
->domain
));
2033 cmd_result
= LTTNG_ERR_FATAL
;
2037 source_element
->is_fd_in_poll_set
= false;
2040 free(source_element
);
2041 *_cmd_result
= cmd_result
;
2045 int handle_notification_thread_remove_tracer_event_source_no_result(
2046 struct notification_thread_state
*state
,
2047 int tracer_event_source_fd
)
2050 enum lttng_error_code cmd_result
;
2052 ret
= handle_notification_thread_command_remove_tracer_event_source(
2053 state
, tracer_event_source_fd
, &cmd_result
);
2058 static int handle_notification_thread_command_list_triggers(
2059 struct notification_thread_handle
*handle
,
2060 struct notification_thread_state
*state
,
2062 struct lttng_triggers
**triggers
,
2063 enum lttng_error_code
*_cmd_result
)
2066 enum lttng_error_code cmd_result
= LTTNG_OK
;
2067 struct cds_lfht_iter iter
;
2068 struct lttng_trigger_ht_element
*trigger_ht_element
;
2069 struct lttng_triggers
*local_triggers
= NULL
;
2070 const struct lttng_credentials
*creds
;
2074 local_triggers
= lttng_triggers_create();
2075 if (!local_triggers
) {
2076 /* Not a fatal error. */
2077 cmd_result
= LTTNG_ERR_NOMEM
;
2081 cds_lfht_for_each_entry(state
->triggers_ht
, &iter
,
2082 trigger_ht_element
, node
) {
2084 * Only return the triggers to which the client has access.
2085 * The root user has visibility over all triggers.
2087 creds
= lttng_trigger_get_credentials(trigger_ht_element
->trigger
);
2088 if (client_uid
!= lttng_credentials_get_uid(creds
) && client_uid
!= 0) {
2092 ret
= lttng_triggers_add(local_triggers
,
2093 trigger_ht_element
->trigger
);
2095 /* Not a fatal error. */
2097 cmd_result
= LTTNG_ERR_NOMEM
;
2102 /* Transferring ownership to the caller. */
2103 *triggers
= local_triggers
;
2104 local_triggers
= NULL
;
2108 lttng_triggers_destroy(local_triggers
);
2109 *_cmd_result
= cmd_result
;
2114 bool condition_is_supported(struct lttng_condition
*condition
)
2118 switch (lttng_condition_get_type(condition
)) {
2119 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
2120 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
2123 enum lttng_domain_type domain
;
2125 ret
= lttng_condition_buffer_usage_get_domain_type(condition
,
2129 if (domain
!= LTTNG_DOMAIN_KERNEL
) {
2130 is_supported
= true;
2135 * Older kernel tracers don't expose the API to monitor their
2136 * buffers. Therefore, we reject triggers that require that
2137 * mechanism to be available to be evaluated.
2139 * Assume unsupported on error.
2141 is_supported
= kernel_supports_ring_buffer_snapshot_sample_positions() == 1;
2144 case LTTNG_CONDITION_TYPE_EVENT_RULE_HIT
:
2146 const struct lttng_event_rule
*event_rule
;
2147 enum lttng_domain_type domain
;
2148 const enum lttng_condition_status status
=
2149 lttng_condition_event_rule_get_rule(
2150 condition
, &event_rule
);
2152 assert(status
== LTTNG_CONDITION_STATUS_OK
);
2154 domain
= lttng_event_rule_get_domain_type(event_rule
);
2155 if (domain
!= LTTNG_DOMAIN_KERNEL
) {
2156 is_supported
= true;
2161 * Older kernel tracers can't emit notification. Therefore, we
2162 * reject triggers that require that mechanism to be available
2165 * Assume unsupported on error.
2167 is_supported
= kernel_supports_event_notifiers() == 1;
2171 is_supported
= true;
2174 return is_supported
;
2177 /* Must be called with RCU read lock held. */
2179 int bind_trigger_to_matching_session(struct lttng_trigger
*trigger
,
2180 struct notification_thread_state
*state
)
2183 const struct lttng_condition
*condition
;
2184 const char *session_name
;
2185 struct lttng_session_trigger_list
*trigger_list
;
2187 condition
= lttng_trigger_get_const_condition(trigger
);
2188 switch (lttng_condition_get_type(condition
)) {
2189 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
2190 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
2192 enum lttng_condition_status status
;
2194 status
= lttng_condition_session_rotation_get_session_name(
2195 condition
, &session_name
);
2196 if (status
!= LTTNG_CONDITION_STATUS_OK
) {
2197 ERR("[notification-thread] Failed to bind trigger to session: unable to get 'session_rotation' condition's session name");
2208 trigger_list
= get_session_trigger_list(state
, session_name
);
2209 if (!trigger_list
) {
2210 DBG("[notification-thread] Unable to bind trigger applying to session \"%s\" as it is not yet known to the notification system",
2216 DBG("[notification-thread] Newly registered trigger bound to session \"%s\"",
2218 ret
= lttng_session_trigger_list_add(trigger_list
, trigger
);
2223 /* Must be called with RCU read lock held. */
2225 int bind_trigger_to_matching_channels(struct lttng_trigger
*trigger
,
2226 struct notification_thread_state
*state
)
2229 struct cds_lfht_node
*node
;
2230 struct cds_lfht_iter iter
;
2231 struct channel_info
*channel
;
2233 cds_lfht_for_each_entry(state
->channels_ht
, &iter
, channel
,
2235 struct lttng_trigger_list_element
*trigger_list_element
;
2236 struct lttng_channel_trigger_list
*trigger_list
;
2237 struct cds_lfht_iter lookup_iter
;
2239 if (!trigger_applies_to_channel(trigger
, channel
)) {
2243 cds_lfht_lookup(state
->channel_triggers_ht
,
2244 hash_channel_key(&channel
->key
),
2245 match_channel_trigger_list
,
2248 node
= cds_lfht_iter_get_node(&lookup_iter
);
2250 trigger_list
= caa_container_of(node
,
2251 struct lttng_channel_trigger_list
,
2252 channel_triggers_ht_node
);
2254 trigger_list_element
= zmalloc(sizeof(*trigger_list_element
));
2255 if (!trigger_list_element
) {
2259 CDS_INIT_LIST_HEAD(&trigger_list_element
->node
);
2260 trigger_list_element
->trigger
= trigger
;
2261 cds_list_add(&trigger_list_element
->node
, &trigger_list
->list
);
2262 DBG("[notification-thread] Newly registered trigger bound to channel \"%s\"",
2270 bool is_trigger_action_notify(const struct lttng_trigger
*trigger
)
2272 bool is_notify
= false;
2273 unsigned int i
, count
;
2274 enum lttng_action_status action_status
;
2275 const struct lttng_action
*action
=
2276 lttng_trigger_get_const_action(trigger
);
2277 enum lttng_action_type action_type
;
2280 action_type
= lttng_action_get_type(action
);
2281 if (action_type
== LTTNG_ACTION_TYPE_NOTIFY
) {
2284 } else if (action_type
!= LTTNG_ACTION_TYPE_GROUP
) {
2288 action_status
= lttng_action_group_get_count(action
, &count
);
2289 assert(action_status
== LTTNG_ACTION_STATUS_OK
);
2291 for (i
= 0; i
< count
; i
++) {
2292 const struct lttng_action
*inner_action
=
2293 lttng_action_group_get_at_index(
2296 action_type
= lttng_action_get_type(inner_action
);
2297 if (action_type
== LTTNG_ACTION_TYPE_NOTIFY
) {
2307 static bool trigger_name_taken(struct notification_thread_state
*state
,
2308 const struct lttng_trigger
*trigger
)
2310 struct cds_lfht_iter iter
;
2313 * No duplicata is allowed in the triggers_by_name_uid_ht.
2314 * The match is done against the trigger name and uid.
2316 cds_lfht_lookup(state
->triggers_by_name_uid_ht
,
2317 hash_trigger_by_name_uid(trigger
),
2318 match_trigger_by_name_uid
,
2321 return !!cds_lfht_iter_get_node(&iter
);
2325 enum lttng_error_code
generate_trigger_name(
2326 struct notification_thread_state
*state
,
2327 struct lttng_trigger
*trigger
, const char **name
)
2329 enum lttng_error_code ret_code
= LTTNG_OK
;
2331 enum lttng_trigger_status status
;
2334 const int ret
= lttng_trigger_generate_name(trigger
,
2335 state
->trigger_id
.name_offset
++);
2337 /* The only reason this can fail right now. */
2338 ret_code
= LTTNG_ERR_NOMEM
;
2342 status
= lttng_trigger_get_name(trigger
, name
);
2343 assert(status
== LTTNG_TRIGGER_STATUS_OK
);
2345 taken
= trigger_name_taken(state
, trigger
);
2346 } while (taken
|| state
->trigger_id
.name_offset
== UINT64_MAX
);
2352 * FIXME A client's credentials are not checked when registering a trigger.
2354 * The effects of this are benign since:
2355 * - The client will succeed in registering the trigger, as it is valid,
2356 * - The trigger will, internally, be bound to the channel/session,
2357 * - The notifications will not be sent since the client's credentials
2358 * are checked against the channel at that moment.
2360 * If this function returns a non-zero value, it means something is
2361 * fundamentally broken and the whole subsystem/thread will be torn down.
2363 * If a non-fatal error occurs, just set the cmd_result to the appropriate
2367 int handle_notification_thread_command_register_trigger(
2368 struct notification_thread_state
*state
,
2369 struct lttng_trigger
*trigger
,
2370 enum lttng_error_code
*cmd_result
)
2373 struct lttng_condition
*condition
;
2374 struct notification_client
*client
;
2375 struct notification_client_list
*client_list
= NULL
;
2376 struct lttng_trigger_ht_element
*trigger_ht_element
= NULL
;
2377 struct notification_client_list_element
*client_list_element
;
2378 struct notification_trigger_tokens_ht_element
*trigger_tokens_ht_element
= NULL
;
2379 struct cds_lfht_node
*node
;
2380 struct cds_lfht_iter iter
;
2381 const char* trigger_name
;
2382 bool free_trigger
= true;
2383 struct lttng_evaluation
*evaluation
= NULL
;
2384 struct lttng_credentials object_creds
;
2387 enum action_executor_status executor_status
;
2388 const uint64_t trigger_tracer_token
=
2389 state
->trigger_id
.next_tracer_token
++;
2393 /* Set the trigger's tracer token. */
2394 lttng_trigger_set_tracer_token(trigger
, trigger_tracer_token
);
2396 if (lttng_trigger_get_name(trigger
, &trigger_name
) ==
2397 LTTNG_TRIGGER_STATUS_UNSET
) {
2398 const enum lttng_error_code ret_code
= generate_trigger_name(
2399 state
, trigger
, &trigger_name
);
2401 if (ret_code
!= LTTNG_OK
) {
2404 *cmd_result
= ret_code
;
2407 } else if (trigger_name_taken(state
, trigger
)) {
2408 /* Not a fatal error. */
2409 *cmd_result
= LTTNG_ERR_TRIGGER_EXISTS
;
2414 condition
= lttng_trigger_get_condition(trigger
);
2417 /* Some conditions require tracers to implement a minimal ABI version. */
2418 if (!condition_is_supported(condition
)) {
2419 *cmd_result
= LTTNG_ERR_NOT_SUPPORTED
;
2423 trigger_ht_element
= zmalloc(sizeof(*trigger_ht_element
));
2424 if (!trigger_ht_element
) {
2429 /* Add trigger to the trigger_ht. */
2430 cds_lfht_node_init(&trigger_ht_element
->node
);
2431 cds_lfht_node_init(&trigger_ht_element
->node_by_name_uid
);
2432 trigger_ht_element
->trigger
= trigger
;
2434 node
= cds_lfht_add_unique(state
->triggers_ht
,
2435 lttng_condition_hash(condition
),
2438 &trigger_ht_element
->node
);
2439 if (node
!= &trigger_ht_element
->node
) {
2440 /* Not a fatal error, simply report it to the client. */
2441 *cmd_result
= LTTNG_ERR_TRIGGER_EXISTS
;
2442 goto error_free_ht_element
;
2445 node
= cds_lfht_add_unique(state
->triggers_by_name_uid_ht
,
2446 hash_trigger_by_name_uid(trigger
),
2447 match_trigger_by_name_uid
,
2449 &trigger_ht_element
->node_by_name_uid
);
2450 if (node
!= &trigger_ht_element
->node_by_name_uid
) {
2451 /* Not a fatal error, simply report it to the client. */
2452 cds_lfht_del(state
->triggers_ht
, &trigger_ht_element
->node
);
2453 *cmd_result
= LTTNG_ERR_TRIGGER_EXISTS
;
2454 goto error_free_ht_element
;
2457 if (lttng_condition_get_type(condition
) == LTTNG_CONDITION_TYPE_EVENT_RULE_HIT
) {
2458 trigger_tokens_ht_element
= zmalloc(sizeof(*trigger_tokens_ht_element
));
2459 if (!trigger_tokens_ht_element
) {
2462 cds_lfht_del(state
->triggers_ht
,
2463 &trigger_ht_element
->node
);
2464 cds_lfht_del(state
->triggers_by_name_uid_ht
,
2465 &trigger_ht_element
->node_by_name_uid
);
2466 goto error_free_ht_element
;
2469 /* Add trigger token to the trigger_tokens_ht. */
2470 cds_lfht_node_init(&trigger_tokens_ht_element
->node
);
2471 trigger_tokens_ht_element
->token
=
2472 LTTNG_OPTIONAL_GET(trigger
->tracer_token
);
2473 trigger_tokens_ht_element
->trigger
= trigger
;
2475 node
= cds_lfht_add_unique(state
->trigger_tokens_ht
,
2476 hash_key_u64(&trigger_tokens_ht_element
->token
,
2478 match_trigger_token
,
2479 &trigger_tokens_ht_element
->token
,
2480 &trigger_tokens_ht_element
->node
);
2481 if (node
!= &trigger_tokens_ht_element
->node
) {
2482 /* Internal corruption, fatal error. */
2484 *cmd_result
= LTTNG_ERR_TRIGGER_EXISTS
;
2485 cds_lfht_del(state
->triggers_ht
,
2486 &trigger_ht_element
->node
);
2487 cds_lfht_del(state
->triggers_by_name_uid_ht
,
2488 &trigger_ht_element
->node_by_name_uid
);
2489 goto error_free_ht_element
;
2494 * Ownership of the trigger and of its wrapper was transfered to
2495 * the triggers_ht. Same for token ht element if necessary.
2497 trigger_tokens_ht_element
= NULL
;
2498 trigger_ht_element
= NULL
;
2499 free_trigger
= false;
2502 * The rest only applies to triggers that have a "notify" action.
2503 * It is not skipped as this is the only action type currently
2506 if (is_trigger_action_notify(trigger
)) {
2507 client_list
= notification_client_list_create(trigger
);
2510 goto error_free_ht_element
;
2513 /* Build a list of clients to which this new trigger applies. */
2514 cds_lfht_for_each_entry (state
->client_socket_ht
, &iter
, client
,
2515 client_socket_ht_node
) {
2516 if (!trigger_applies_to_client(trigger
, client
)) {
2520 client_list_element
=
2521 zmalloc(sizeof(*client_list_element
));
2522 if (!client_list_element
) {
2524 goto error_put_client_list
;
2527 CDS_INIT_LIST_HEAD(&client_list_element
->node
);
2528 client_list_element
->client
= client
;
2529 cds_list_add(&client_list_element
->node
,
2530 &client_list
->list
);
2534 * Client list ownership transferred to the
2535 * notification_trigger_clients_ht.
2537 publish_notification_client_list(state
, client_list
);
2540 switch (get_condition_binding_object(condition
)) {
2541 case LTTNG_OBJECT_TYPE_SESSION
:
2542 /* Add the trigger to the list if it matches a known session. */
2543 ret
= bind_trigger_to_matching_session(trigger
, state
);
2545 goto error_put_client_list
;
2548 case LTTNG_OBJECT_TYPE_CHANNEL
:
2550 * Add the trigger to list of triggers bound to the channels
2553 ret
= bind_trigger_to_matching_channels(trigger
, state
);
2555 goto error_put_client_list
;
2558 case LTTNG_OBJECT_TYPE_NONE
:
2561 ERR("Unknown object type on which to bind a newly registered trigger was encountered");
2563 goto error_put_client_list
;
2567 * The new trigger's condition must be evaluated against the current
2570 * In the case of `notify` action, nothing preventing clients from
2571 * subscribing to a condition before the corresponding trigger is
2572 * registered, we have to evaluate this new condition right away.
2574 * At some point, we were waiting for the next "evaluation" (e.g. on
2575 * reception of a channel sample) to evaluate this new condition, but
2578 * The reason it was broken is that waiting for the next sample
2579 * does not allow us to properly handle transitions for edge-triggered
2582 * Consider this example: when we handle a new channel sample, we
2583 * evaluate each conditions twice: once with the previous state, and
2584 * again with the newest state. We then use those two results to
2585 * determine whether a state change happened: a condition was false and
2586 * became true. If a state change happened, we have to notify clients.
2588 * Now, if a client subscribes to a given notification and registers
2589 * a trigger *after* that subscription, we have to make sure the
2590 * condition is evaluated at this point while considering only the
2591 * current state. Otherwise, the next evaluation cycle may only see
2592 * that the evaluations remain the same (true for samples n-1 and n) and
2593 * the client will never know that the condition has been met.
2595 switch (get_condition_binding_object(condition
)) {
2596 case LTTNG_OBJECT_TYPE_SESSION
:
2597 ret
= evaluate_session_condition_for_client(condition
, state
,
2598 &evaluation
, &object_uid
,
2600 LTTNG_OPTIONAL_SET(&object_creds
.uid
, object_uid
);
2601 LTTNG_OPTIONAL_SET(&object_creds
.gid
, object_gid
);
2603 case LTTNG_OBJECT_TYPE_CHANNEL
:
2604 ret
= evaluate_channel_condition_for_client(condition
, state
,
2605 &evaluation
, &object_uid
,
2607 LTTNG_OPTIONAL_SET(&object_creds
.uid
, object_uid
);
2608 LTTNG_OPTIONAL_SET(&object_creds
.gid
, object_gid
);
2610 case LTTNG_OBJECT_TYPE_NONE
:
2613 case LTTNG_OBJECT_TYPE_UNKNOWN
:
2621 goto error_put_client_list
;
2624 DBG("Newly registered trigger's condition evaluated to %s",
2625 evaluation
? "true" : "false");
2627 /* Evaluation yielded nothing. Normal exit. */
2633 * Ownership of `evaluation` transferred to the action executor
2634 * no matter the result.
2636 executor_status
= action_executor_enqueue(state
->executor
, trigger
,
2637 evaluation
, &object_creds
, client_list
);
2639 switch (executor_status
) {
2640 case ACTION_EXECUTOR_STATUS_OK
:
2642 case ACTION_EXECUTOR_STATUS_ERROR
:
2643 case ACTION_EXECUTOR_STATUS_INVALID
:
2645 * TODO Add trigger identification (name/id) when
2646 * it is added to the API.
2648 ERR("Fatal error occurred while enqueuing action associated to newly registered trigger");
2650 goto error_put_client_list
;
2651 case ACTION_EXECUTOR_STATUS_OVERFLOW
:
2653 * TODO Add trigger identification (name/id) when
2654 * it is added to the API.
2656 * Not a fatal error.
2658 WARN("No space left when enqueuing action associated to newly registered trigger");
2666 *cmd_result
= LTTNG_OK
;
2667 DBG("Registered trigger: name = `%s`, tracer token = %" PRIu64
,
2668 trigger_name
, trigger_tracer_token
);
2670 error_put_client_list
:
2671 notification_client_list_put(client_list
);
2673 error_free_ht_element
:
2674 if (trigger_ht_element
) {
2675 /* Delayed removal due to RCU constraint on delete. */
2676 call_rcu(&trigger_ht_element
->rcu_node
,
2677 free_lttng_trigger_ht_element_rcu
);
2680 free(trigger_tokens_ht_element
);
2683 lttng_trigger_destroy(trigger
);
2690 void free_lttng_trigger_ht_element_rcu(struct rcu_head
*node
)
2692 free(caa_container_of(node
, struct lttng_trigger_ht_element
,
2697 void free_notification_trigger_tokens_ht_element_rcu(struct rcu_head
*node
)
2699 free(caa_container_of(node
, struct notification_trigger_tokens_ht_element
,
2704 int handle_notification_thread_command_unregister_trigger(
2705 struct notification_thread_state
*state
,
2706 const struct lttng_trigger
*trigger
,
2707 enum lttng_error_code
*_cmd_reply
)
2709 struct cds_lfht_iter iter
;
2710 struct cds_lfht_node
*triggers_ht_node
;
2711 struct lttng_channel_trigger_list
*trigger_list
;
2712 struct notification_client_list
*client_list
;
2713 struct lttng_trigger_ht_element
*trigger_ht_element
= NULL
;
2714 const struct lttng_condition
*condition
= lttng_trigger_get_const_condition(
2716 enum lttng_error_code cmd_reply
;
2720 cds_lfht_lookup(state
->triggers_ht
,
2721 lttng_condition_hash(condition
),
2725 triggers_ht_node
= cds_lfht_iter_get_node(&iter
);
2726 if (!triggers_ht_node
) {
2727 cmd_reply
= LTTNG_ERR_TRIGGER_NOT_FOUND
;
2730 cmd_reply
= LTTNG_OK
;
2733 /* Remove trigger from channel_triggers_ht. */
2734 cds_lfht_for_each_entry(state
->channel_triggers_ht
, &iter
, trigger_list
,
2735 channel_triggers_ht_node
) {
2736 struct lttng_trigger_list_element
*trigger_element
, *tmp
;
2738 cds_list_for_each_entry_safe(trigger_element
, tmp
,
2739 &trigger_list
->list
, node
) {
2740 if (!lttng_trigger_is_equal(trigger
, trigger_element
->trigger
)) {
2744 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
2745 cds_list_del(&trigger_element
->node
);
2746 /* A trigger can only appear once per channel */
2751 if (lttng_condition_get_type(condition
) ==
2752 LTTNG_CONDITION_TYPE_EVENT_RULE_HIT
) {
2753 struct notification_trigger_tokens_ht_element
2754 *trigger_tokens_ht_element
;
2756 cds_lfht_for_each_entry (state
->trigger_tokens_ht
, &iter
,
2757 trigger_tokens_ht_element
, node
) {
2758 if (!lttng_trigger_is_equal(trigger
,
2759 trigger_tokens_ht_element
->trigger
)) {
2763 DBG("[notification-thread] Removed trigger from tokens_ht");
2764 cds_lfht_del(state
->trigger_tokens_ht
,
2765 &trigger_tokens_ht_element
->node
);
2766 call_rcu(&trigger_tokens_ht_element
->rcu_node
,
2767 free_notification_trigger_tokens_ht_element_rcu
);
2773 if (is_trigger_action_notify(trigger
)) {
2775 * Remove and release the client list from
2776 * notification_trigger_clients_ht.
2778 client_list
= get_client_list_from_condition(state
, condition
);
2779 assert(client_list
);
2781 /* Put new reference and the hashtable's reference. */
2782 notification_client_list_put(client_list
);
2783 notification_client_list_put(client_list
);
2787 /* Remove trigger from triggers_ht. */
2788 trigger_ht_element
= caa_container_of(triggers_ht_node
,
2789 struct lttng_trigger_ht_element
, node
);
2790 cds_lfht_del(state
->triggers_by_name_uid_ht
, &trigger_ht_element
->node_by_name_uid
);
2791 cds_lfht_del(state
->triggers_ht
, triggers_ht_node
);
2793 /* Release the ownership of the trigger. */
2794 lttng_trigger_destroy(trigger_ht_element
->trigger
);
2795 call_rcu(&trigger_ht_element
->rcu_node
, free_lttng_trigger_ht_element_rcu
);
2799 *_cmd_reply
= cmd_reply
;
2804 /* Returns 0 on success, 1 on exit requested, negative value on error. */
2805 int handle_notification_thread_command(
2806 struct notification_thread_handle
*handle
,
2807 struct notification_thread_state
*state
)
2811 struct notification_thread_command
*cmd
;
2813 /* Read the event pipe to put it back into a quiescent state. */
2814 ret
= lttng_read(lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
), &counter
,
2816 if (ret
!= sizeof(counter
)) {
2820 pthread_mutex_lock(&handle
->cmd_queue
.lock
);
2821 cmd
= cds_list_first_entry(&handle
->cmd_queue
.list
,
2822 struct notification_thread_command
, cmd_list_node
);
2823 cds_list_del(&cmd
->cmd_list_node
);
2824 pthread_mutex_unlock(&handle
->cmd_queue
.lock
);
2826 DBG("[notification-thread] Received `%s` command",
2827 notification_command_type_str(cmd
->type
));
2828 switch (cmd
->type
) {
2829 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER
:
2830 ret
= handle_notification_thread_command_register_trigger(state
,
2831 cmd
->parameters
.register_trigger
.trigger
,
2834 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER
:
2835 ret
= handle_notification_thread_command_unregister_trigger(
2837 cmd
->parameters
.unregister_trigger
.trigger
,
2840 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL
:
2841 ret
= handle_notification_thread_command_add_channel(
2843 cmd
->parameters
.add_channel
.session
.name
,
2844 cmd
->parameters
.add_channel
.session
.uid
,
2845 cmd
->parameters
.add_channel
.session
.gid
,
2846 cmd
->parameters
.add_channel
.channel
.name
,
2847 cmd
->parameters
.add_channel
.channel
.domain
,
2848 cmd
->parameters
.add_channel
.channel
.key
,
2849 cmd
->parameters
.add_channel
.channel
.capacity
,
2852 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL
:
2853 ret
= handle_notification_thread_command_remove_channel(
2854 state
, cmd
->parameters
.remove_channel
.key
,
2855 cmd
->parameters
.remove_channel
.domain
,
2858 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
:
2859 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED
:
2860 ret
= handle_notification_thread_command_session_rotation(
2863 cmd
->parameters
.session_rotation
.session_name
,
2864 cmd
->parameters
.session_rotation
.uid
,
2865 cmd
->parameters
.session_rotation
.gid
,
2866 cmd
->parameters
.session_rotation
.trace_archive_chunk_id
,
2867 cmd
->parameters
.session_rotation
.location
,
2870 case NOTIFICATION_COMMAND_TYPE_ADD_TRACER_EVENT_SOURCE
:
2871 ret
= handle_notification_thread_command_add_tracer_event_source(
2873 cmd
->parameters
.tracer_event_source
.tracer_event_source_fd
,
2874 cmd
->parameters
.tracer_event_source
.domain
,
2877 case NOTIFICATION_COMMAND_TYPE_REMOVE_TRACER_EVENT_SOURCE
:
2878 ret
= handle_notification_thread_command_remove_tracer_event_source(
2880 cmd
->parameters
.tracer_event_source
.tracer_event_source_fd
,
2883 case NOTIFICATION_COMMAND_TYPE_LIST_TRIGGERS
:
2885 struct lttng_triggers
*triggers
= NULL
;
2887 ret
= handle_notification_thread_command_list_triggers(
2890 cmd
->parameters
.list_triggers
.uid
,
2893 cmd
->reply
.list_triggers
.triggers
= triggers
;
2897 case NOTIFICATION_COMMAND_TYPE_QUIT
:
2898 cmd
->reply_code
= LTTNG_OK
;
2901 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE
:
2903 const enum client_transmission_status client_status
=
2904 cmd
->parameters
.client_communication_update
2906 const notification_client_id client_id
=
2907 cmd
->parameters
.client_communication_update
.id
;
2908 struct notification_client
*client
;
2911 client
= get_client_from_id(client_id
, state
);
2915 * Client error was probably already picked-up by the
2916 * notification thread or it has disconnected
2917 * gracefully while this command was queued.
2919 DBG("Failed to find notification client to update communication status, client id = %" PRIu64
,
2923 ret
= client_handle_transmission_status(
2924 client
, client_status
, state
);
2930 ERR("[notification-thread] Unknown internal command received");
2938 if (cmd
->is_async
) {
2942 lttng_waiter_wake_up(&cmd
->reply_waiter
);
2946 /* Wake-up and return a fatal error to the calling thread. */
2947 lttng_waiter_wake_up(&cmd
->reply_waiter
);
2948 cmd
->reply_code
= LTTNG_ERR_FATAL
;
2950 /* Indicate a fatal error to the caller. */
2955 int socket_set_non_blocking(int socket
)
2959 /* Set the pipe as non-blocking. */
2960 ret
= fcntl(socket
, F_GETFL
, 0);
2962 PERROR("fcntl get socket flags");
2967 ret
= fcntl(socket
, F_SETFL
, flags
| O_NONBLOCK
);
2969 PERROR("fcntl set O_NONBLOCK socket flag");
2972 DBG("Client socket (fd = %i) set as non-blocking", socket
);
2978 int client_reset_inbound_state(struct notification_client
*client
)
2983 lttng_payload_clear(&client
->communication
.inbound
.payload
);
2985 client
->communication
.inbound
.bytes_to_receive
=
2986 sizeof(struct lttng_notification_channel_message
);
2987 client
->communication
.inbound
.msg_type
=
2988 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN
;
2989 LTTNG_SOCK_SET_UID_CRED(&client
->communication
.inbound
.creds
, -1);
2990 LTTNG_SOCK_SET_GID_CRED(&client
->communication
.inbound
.creds
, -1);
2991 ret
= lttng_dynamic_buffer_set_size(
2992 &client
->communication
.inbound
.payload
.buffer
,
2993 client
->communication
.inbound
.bytes_to_receive
);
2998 int handle_notification_thread_client_connect(
2999 struct notification_thread_state
*state
)
3002 struct notification_client
*client
;
3004 DBG("[notification-thread] Handling new notification channel client connection");
3006 client
= zmalloc(sizeof(*client
));
3013 pthread_mutex_init(&client
->lock
, NULL
);
3014 client
->id
= state
->next_notification_client_id
++;
3015 CDS_INIT_LIST_HEAD(&client
->condition_list
);
3016 lttng_payload_init(&client
->communication
.inbound
.payload
);
3017 lttng_payload_init(&client
->communication
.outbound
.payload
);
3018 client
->communication
.inbound
.expect_creds
= true;
3020 ret
= client_reset_inbound_state(client
);
3022 ERR("[notification-thread] Failed to reset client communication's inbound state");
3027 ret
= lttcomm_accept_unix_sock(state
->notification_channel_socket
);
3029 ERR("[notification-thread] Failed to accept new notification channel client connection");
3034 client
->socket
= ret
;
3036 ret
= socket_set_non_blocking(client
->socket
);
3038 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
3042 ret
= lttcomm_setsockopt_creds_unix_sock(client
->socket
);
3044 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
3049 ret
= lttng_poll_add(&state
->events
, client
->socket
,
3050 LPOLLIN
| LPOLLERR
|
3051 LPOLLHUP
| LPOLLRDHUP
);
3053 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
3057 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
3061 cds_lfht_add(state
->client_socket_ht
,
3062 hash_client_socket(client
->socket
),
3063 &client
->client_socket_ht_node
);
3064 cds_lfht_add(state
->client_id_ht
,
3065 hash_client_id(client
->id
),
3066 &client
->client_id_ht_node
);
3072 notification_client_destroy(client
, state
);
3077 * RCU read-lock must be held by the caller.
3078 * Client lock must _not_ be held by the caller.
3081 int notification_thread_client_disconnect(
3082 struct notification_client
*client
,
3083 struct notification_thread_state
*state
)
3086 struct lttng_condition_list_element
*condition_list_element
, *tmp
;
3088 /* Acquire the client lock to disable its communication atomically. */
3089 pthread_mutex_lock(&client
->lock
);
3090 client
->communication
.active
= false;
3091 cds_lfht_del(state
->client_socket_ht
, &client
->client_socket_ht_node
);
3092 cds_lfht_del(state
->client_id_ht
, &client
->client_id_ht_node
);
3093 pthread_mutex_unlock(&client
->lock
);
3095 ret
= lttng_poll_del(&state
->events
, client
->socket
);
3097 ERR("[notification-thread] Failed to remove client socket %d from poll set",
3101 /* Release all conditions to which the client was subscribed. */
3102 cds_list_for_each_entry_safe(condition_list_element
, tmp
,
3103 &client
->condition_list
, node
) {
3104 (void) notification_thread_client_unsubscribe(client
,
3105 condition_list_element
->condition
, state
, NULL
);
3109 * Client no longer accessible to other threads (through the
3112 notification_client_destroy(client
, state
);
3116 int handle_notification_thread_client_disconnect(
3117 int client_socket
, struct notification_thread_state
*state
)
3120 struct notification_client
*client
;
3123 DBG("[notification-thread] Closing client connection (socket fd = %i)",
3125 client
= get_client_from_socket(client_socket
, state
);
3127 /* Internal state corruption, fatal error. */
3128 ERR("[notification-thread] Unable to find client (socket fd = %i)",
3134 ret
= notification_thread_client_disconnect(client
, state
);
3140 int handle_notification_thread_client_disconnect_all(
3141 struct notification_thread_state
*state
)
3143 struct cds_lfht_iter iter
;
3144 struct notification_client
*client
;
3145 bool error_encoutered
= false;
3148 DBG("[notification-thread] Closing all client connections");
3149 cds_lfht_for_each_entry(state
->client_socket_ht
, &iter
, client
,
3150 client_socket_ht_node
) {
3153 ret
= notification_thread_client_disconnect(
3156 error_encoutered
= true;
3160 return error_encoutered
? 1 : 0;
3163 int handle_notification_thread_trigger_unregister_all(
3164 struct notification_thread_state
*state
)
3166 bool error_occurred
= false;
3167 struct cds_lfht_iter iter
;
3168 struct lttng_trigger_ht_element
*trigger_ht_element
;
3171 cds_lfht_for_each_entry(state
->triggers_ht
, &iter
, trigger_ht_element
,
3173 int ret
= handle_notification_thread_command_unregister_trigger(
3174 state
, trigger_ht_element
->trigger
, NULL
);
3176 error_occurred
= true;
3180 return error_occurred
? -1 : 0;
3184 int client_handle_transmission_status(
3185 struct notification_client
*client
,
3186 enum client_transmission_status transmission_status
,
3187 struct notification_thread_state
*state
)
3191 switch (transmission_status
) {
3192 case CLIENT_TRANSMISSION_STATUS_COMPLETE
:
3193 ret
= lttng_poll_mod(&state
->events
, client
->socket
,
3194 CLIENT_POLL_MASK_IN
);
3200 case CLIENT_TRANSMISSION_STATUS_QUEUED
:
3202 * We want to be notified whenever there is buffer space
3203 * available to send the rest of the payload.
3205 ret
= lttng_poll_mod(&state
->events
, client
->socket
,
3206 CLIENT_POLL_MASK_IN_OUT
);
3211 case CLIENT_TRANSMISSION_STATUS_FAIL
:
3212 ret
= notification_thread_client_disconnect(client
, state
);
3217 case CLIENT_TRANSMISSION_STATUS_ERROR
:
3227 /* Client lock must be acquired by caller. */
3229 enum client_transmission_status
client_flush_outgoing_queue(
3230 struct notification_client
*client
)
3233 size_t to_send_count
;
3234 enum client_transmission_status status
;
3235 struct lttng_payload_view pv
= lttng_payload_view_from_payload(
3236 &client
->communication
.outbound
.payload
, 0, -1);
3237 const int fds_to_send_count
=
3238 lttng_payload_view_get_fd_handle_count(&pv
);
3240 ASSERT_LOCKED(client
->lock
);
3242 if (!client
->communication
.active
) {
3243 status
= CLIENT_TRANSMISSION_STATUS_FAIL
;
3247 if (pv
.buffer
.size
== 0) {
3249 * If both data and fds are equal to zero, we are in an invalid
3252 assert(fds_to_send_count
!= 0);
3257 to_send_count
= pv
.buffer
.size
;
3258 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
3261 ret
= lttcomm_send_unix_sock_non_block(client
->socket
,
3264 if ((ret
>= 0 && ret
< to_send_count
)) {
3265 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
3267 to_send_count
-= max(ret
, 0);
3269 memmove(client
->communication
.outbound
.payload
.buffer
.data
,
3271 pv
.buffer
.size
- to_send_count
,
3273 ret
= lttng_dynamic_buffer_set_size(
3274 &client
->communication
.outbound
.payload
.buffer
,
3280 status
= CLIENT_TRANSMISSION_STATUS_QUEUED
;
3282 } else if (ret
< 0) {
3283 /* Generic error, disable the client's communication. */
3284 ERR("[notification-thread] Failed to flush outgoing queue, disconnecting client (socket fd = %i)",
3286 client
->communication
.active
= false;
3287 status
= CLIENT_TRANSMISSION_STATUS_FAIL
;
3291 * No error and flushed the queue completely.
3293 * The payload buffer size is used later to
3294 * check if there is notifications queued. So albeit that the
3295 * direct caller knows that the transmission is complete, we
3296 * need to set the buffer size to zero.
3298 ret
= lttng_dynamic_buffer_set_size(
3299 &client
->communication
.outbound
.payload
.buffer
, 0);
3306 /* No fds to send, transmission is complete. */
3307 if (fds_to_send_count
== 0) {
3308 status
= CLIENT_TRANSMISSION_STATUS_COMPLETE
;
3312 ret
= lttcomm_send_payload_view_fds_unix_sock_non_block(
3313 client
->socket
, &pv
);
3315 /* Generic error, disable the client's communication. */
3316 ERR("[notification-thread] Failed to flush outgoing fds queue, disconnecting client (socket fd = %i)",
3318 client
->communication
.active
= false;
3319 status
= CLIENT_TRANSMISSION_STATUS_FAIL
;
3321 } else if (ret
== 0) {
3322 /* Nothing could be sent. */
3323 status
= CLIENT_TRANSMISSION_STATUS_QUEUED
;
3325 /* Fd passing is an all or nothing kind of thing. */
3326 status
= CLIENT_TRANSMISSION_STATUS_COMPLETE
;
3328 * The payload _fd_array count is used later to
3329 * check if there is notifications queued. So although the
3330 * direct caller knows that the transmission is complete, we
3331 * need to clear the _fd_array for the queuing check.
3333 lttng_dynamic_pointer_array_clear(
3334 &client
->communication
.outbound
.payload
3339 if (status
== CLIENT_TRANSMISSION_STATUS_COMPLETE
) {
3340 client
->communication
.outbound
.queued_command_reply
= false;
3341 client
->communication
.outbound
.dropped_notification
= false;
3342 lttng_payload_clear(&client
->communication
.outbound
.payload
);
3347 return CLIENT_TRANSMISSION_STATUS_ERROR
;
3351 bool client_has_outbound_data_left(
3352 const struct notification_client
*client
)
3354 const struct lttng_payload_view pv
= lttng_payload_view_from_payload(
3355 &client
->communication
.outbound
.payload
, 0, -1);
3356 const bool has_data
= pv
.buffer
.size
!= 0;
3357 const bool has_fds
= lttng_payload_view_get_fd_handle_count(&pv
);
3359 return has_data
|| has_fds
;
3362 /* Client lock must _not_ be held by the caller. */
3364 int client_send_command_reply(struct notification_client
*client
,
3365 struct notification_thread_state
*state
,
3366 enum lttng_notification_channel_status status
)
3369 struct lttng_notification_channel_command_reply reply
= {
3370 .status
= (int8_t) status
,
3372 struct lttng_notification_channel_message msg
= {
3373 .type
= (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY
,
3374 .size
= sizeof(reply
),
3376 char buffer
[sizeof(msg
) + sizeof(reply
)];
3377 enum client_transmission_status transmission_status
;
3379 memcpy(buffer
, &msg
, sizeof(msg
));
3380 memcpy(buffer
+ sizeof(msg
), &reply
, sizeof(reply
));
3381 DBG("[notification-thread] Send command reply (%i)", (int) status
);
3383 pthread_mutex_lock(&client
->lock
);
3384 if (client
->communication
.outbound
.queued_command_reply
) {
3385 /* Protocol error. */
3389 /* Enqueue buffer to outgoing queue and flush it. */
3390 ret
= lttng_dynamic_buffer_append(
3391 &client
->communication
.outbound
.payload
.buffer
,
3392 buffer
, sizeof(buffer
));
3397 transmission_status
= client_flush_outgoing_queue(client
);
3399 if (client_has_outbound_data_left(client
)) {
3400 /* Queue could not be emptied. */
3401 client
->communication
.outbound
.queued_command_reply
= true;
3404 pthread_mutex_unlock(&client
->lock
);
3405 ret
= client_handle_transmission_status(
3406 client
, transmission_status
, state
);
3413 pthread_mutex_unlock(&client
->lock
);
3419 int client_handle_message_unknown(struct notification_client
*client
,
3420 struct notification_thread_state
*state
)
3424 * Receiving message header. The function will be called again
3425 * once the rest of the message as been received and can be
3428 const struct lttng_notification_channel_message
*msg
;
3430 assert(sizeof(*msg
) == client
->communication
.inbound
.payload
.buffer
.size
);
3431 msg
= (const struct lttng_notification_channel_message
*)
3432 client
->communication
.inbound
.payload
.buffer
.data
;
3434 if (msg
->size
== 0 ||
3435 msg
->size
> DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE
) {
3436 ERR("[notification-thread] Invalid notification channel message: length = %u",
3442 switch (msg
->type
) {
3443 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE
:
3444 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE
:
3445 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
:
3449 ERR("[notification-thread] Invalid notification channel message: unexpected message type");
3453 client
->communication
.inbound
.bytes_to_receive
= msg
->size
;
3454 client
->communication
.inbound
.fds_to_receive
= msg
->fds
;
3455 client
->communication
.inbound
.msg_type
=
3456 (enum lttng_notification_channel_message_type
) msg
->type
;
3457 ret
= lttng_dynamic_buffer_set_size(
3458 &client
->communication
.inbound
.payload
.buffer
, msg
->size
);
3460 /* msg is not valid anymore due to lttng_dynamic_buffer_set_size. */
3467 int client_handle_message_handshake(struct notification_client
*client
,
3468 struct notification_thread_state
*state
)
3471 struct lttng_notification_channel_command_handshake
*handshake_client
;
3472 const struct lttng_notification_channel_command_handshake handshake_reply
= {
3473 .major
= LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR
,
3474 .minor
= LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR
,
3476 const struct lttng_notification_channel_message msg_header
= {
3477 .type
= LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
,
3478 .size
= sizeof(handshake_reply
),
3480 enum lttng_notification_channel_status status
=
3481 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
3482 char send_buffer
[sizeof(msg_header
) + sizeof(handshake_reply
)];
3484 memcpy(send_buffer
, &msg_header
, sizeof(msg_header
));
3485 memcpy(send_buffer
+ sizeof(msg_header
), &handshake_reply
,
3486 sizeof(handshake_reply
));
3489 (struct lttng_notification_channel_command_handshake
*)
3490 client
->communication
.inbound
.payload
.buffer
3492 client
->major
= handshake_client
->major
;
3493 client
->minor
= handshake_client
->minor
;
3494 if (!client
->communication
.inbound
.creds_received
) {
3495 ERR("[notification-thread] No credentials received from client");
3500 client
->uid
= LTTNG_SOCK_GET_UID_CRED(
3501 &client
->communication
.inbound
.creds
);
3502 client
->gid
= LTTNG_SOCK_GET_GID_CRED(
3503 &client
->communication
.inbound
.creds
);
3504 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
3505 client
->uid
, client
->gid
, (int) client
->major
,
3506 (int) client
->minor
);
3508 if (handshake_client
->major
!=
3509 LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR
) {
3510 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION
;
3513 pthread_mutex_lock(&client
->lock
);
3514 /* Outgoing queue will be flushed when the command reply is sent. */
3515 ret
= lttng_dynamic_buffer_append(
3516 &client
->communication
.outbound
.payload
.buffer
, send_buffer
,
3517 sizeof(send_buffer
));
3519 ERR("[notification-thread] Failed to send protocol version to notification channel client");
3523 client
->validated
= true;
3524 client
->communication
.active
= true;
3525 pthread_mutex_unlock(&client
->lock
);
3527 /* Set reception state to receive the next message header. */
3528 ret
= client_reset_inbound_state(client
);
3530 ERR("[notification-thread] Failed to reset client communication's inbound state");
3534 /* Flushes the outgoing queue. */
3535 ret
= client_send_command_reply(client
, state
, status
);
3537 ERR("[notification-thread] Failed to send reply to notification channel client");
3543 pthread_mutex_unlock(&client
->lock
);
3549 int client_handle_message_subscription(
3550 struct notification_client
*client
,
3551 enum lttng_notification_channel_message_type msg_type
,
3552 struct notification_thread_state
*state
)
3555 struct lttng_condition
*condition
;
3556 enum lttng_notification_channel_status status
=
3557 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
3558 struct lttng_payload_view condition_view
=
3559 lttng_payload_view_from_payload(
3560 &client
->communication
.inbound
.payload
,
3562 size_t expected_condition_size
;
3565 * No need to lock client to sample the inbound state as the only
3566 * other thread accessing clients (action executor) only uses the
3569 expected_condition_size
= client
->communication
.inbound
.payload
.buffer
.size
;
3570 ret
= lttng_condition_create_from_payload(&condition_view
, &condition
);
3571 if (ret
!= expected_condition_size
) {
3572 ERR("[notification-thread] Malformed condition received from client");
3576 if (msg_type
== LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE
) {
3577 ret
= notification_thread_client_subscribe(
3578 client
, condition
, state
, &status
);
3580 ret
= notification_thread_client_unsubscribe(
3581 client
, condition
, state
, &status
);
3588 /* Set reception state to receive the next message header. */
3589 ret
= client_reset_inbound_state(client
);
3591 ERR("[notification-thread] Failed to reset client communication's inbound state");
3595 ret
= client_send_command_reply(client
, state
, status
);
3597 ERR("[notification-thread] Failed to send reply to notification channel client");
3606 int client_dispatch_message(struct notification_client
*client
,
3607 struct notification_thread_state
*state
)
3611 if (client
->communication
.inbound
.msg_type
!=
3612 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
&&
3613 client
->communication
.inbound
.msg_type
!=
3614 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN
&&
3615 !client
->validated
) {
3616 WARN("[notification-thread] client attempted a command before handshake");
3621 switch (client
->communication
.inbound
.msg_type
) {
3622 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN
:
3624 ret
= client_handle_message_unknown(client
, state
);
3627 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
:
3629 ret
= client_handle_message_handshake(client
, state
);
3632 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE
:
3633 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE
:
3635 ret
= client_handle_message_subscription(client
,
3636 client
->communication
.inbound
.msg_type
, state
);
3646 /* Incoming data from client. */
3647 int handle_notification_thread_client_in(
3648 struct notification_thread_state
*state
, int socket
)
3651 struct notification_client
*client
;
3656 client
= get_client_from_socket(socket
, state
);
3658 /* Internal error, abort. */
3663 offset
= client
->communication
.inbound
.payload
.buffer
.size
-
3664 client
->communication
.inbound
.bytes_to_receive
;
3665 if (client
->communication
.inbound
.expect_creds
) {
3666 recv_ret
= lttcomm_recv_creds_unix_sock(socket
,
3667 client
->communication
.inbound
.payload
.buffer
.data
+ offset
,
3668 client
->communication
.inbound
.bytes_to_receive
,
3669 &client
->communication
.inbound
.creds
);
3671 client
->communication
.inbound
.expect_creds
= false;
3672 client
->communication
.inbound
.creds_received
= true;
3675 recv_ret
= lttcomm_recv_unix_sock_non_block(socket
,
3676 client
->communication
.inbound
.payload
.buffer
.data
+ offset
,
3677 client
->communication
.inbound
.bytes_to_receive
);
3679 if (recv_ret
>= 0) {
3680 client
->communication
.inbound
.bytes_to_receive
-= recv_ret
;
3682 goto error_disconnect_client
;
3685 if (client
->communication
.inbound
.bytes_to_receive
!= 0) {
3686 /* Message incomplete wait for more data. */
3691 assert(client
->communication
.inbound
.bytes_to_receive
== 0);
3694 if (client
->communication
.inbound
.fds_to_receive
!= 0) {
3695 ret
= lttcomm_recv_payload_fds_unix_sock_non_block(
3697 client
->communication
.inbound
.fds_to_receive
,
3698 &client
->communication
.inbound
.payload
);
3701 * Fds received. non blocking fds passing is all
3704 ssize_t expected_size
;
3706 expected_size
= sizeof(int) *
3707 client
->communication
.inbound
3709 assert(ret
== expected_size
);
3710 client
->communication
.inbound
.fds_to_receive
= 0;
3711 } else if (ret
== 0) {
3712 /* Received nothing. */
3716 goto error_disconnect_client
;
3720 /* At this point the message is complete.*/
3721 assert(client
->communication
.inbound
.bytes_to_receive
== 0 &&
3722 client
->communication
.inbound
.fds_to_receive
== 0);
3723 ret
= client_dispatch_message(client
, state
);
3726 * Only returns an error if this client must be
3729 goto error_disconnect_client
;
3736 error_disconnect_client
:
3737 ret
= notification_thread_client_disconnect(client
, state
);
3741 /* Client ready to receive outgoing data. */
3742 int handle_notification_thread_client_out(
3743 struct notification_thread_state
*state
, int socket
)
3746 struct notification_client
*client
;
3747 enum client_transmission_status transmission_status
;
3750 client
= get_client_from_socket(socket
, state
);
3752 /* Internal error, abort. */
3757 pthread_mutex_lock(&client
->lock
);
3758 transmission_status
= client_flush_outgoing_queue(client
);
3759 pthread_mutex_unlock(&client
->lock
);
3761 ret
= client_handle_transmission_status(
3762 client
, transmission_status
, state
);
3772 bool evaluate_buffer_usage_condition(const struct lttng_condition
*condition
,
3773 const struct channel_state_sample
*sample
,
3774 uint64_t buffer_capacity
)
3776 bool result
= false;
3778 enum lttng_condition_type condition_type
;
3779 const struct lttng_condition_buffer_usage
*use_condition
= container_of(
3780 condition
, struct lttng_condition_buffer_usage
,
3783 if (use_condition
->threshold_bytes
.set
) {
3784 threshold
= use_condition
->threshold_bytes
.value
;
3787 * Threshold was expressed as a ratio.
3789 * TODO the threshold (in bytes) of conditions expressed
3790 * as a ratio of total buffer size could be cached to
3791 * forego this double-multiplication or it could be performed
3792 * as fixed-point math.
3794 * Note that caching should accommodates the case where the
3795 * condition applies to multiple channels (i.e. don't assume
3796 * that all channels matching my_chann* have the same size...)
3798 threshold
= (uint64_t) (use_condition
->threshold_ratio
.value
*
3799 (double) buffer_capacity
);
3802 condition_type
= lttng_condition_get_type(condition
);
3803 if (condition_type
== LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
) {
3804 DBG("[notification-thread] Low buffer usage condition being evaluated: threshold = %" PRIu64
", highest usage = %" PRIu64
,
3805 threshold
, sample
->highest_usage
);
3808 * The low condition should only be triggered once _all_ of the
3809 * streams in a channel have gone below the "low" threshold.
3811 if (sample
->highest_usage
<= threshold
) {
3815 DBG("[notification-thread] High buffer usage condition being evaluated: threshold = %" PRIu64
", highest usage = %" PRIu64
,
3816 threshold
, sample
->highest_usage
);
3819 * For high buffer usage scenarios, we want to trigger whenever
3820 * _any_ of the streams has reached the "high" threshold.
3822 if (sample
->highest_usage
>= threshold
) {
3831 bool evaluate_session_consumed_size_condition(
3832 const struct lttng_condition
*condition
,
3833 uint64_t session_consumed_size
)
3836 const struct lttng_condition_session_consumed_size
*size_condition
=
3837 container_of(condition
,
3838 struct lttng_condition_session_consumed_size
,
3841 threshold
= size_condition
->consumed_threshold_bytes
.value
;
3842 DBG("[notification-thread] Session consumed size condition being evaluated: threshold = %" PRIu64
", current size = %" PRIu64
,
3843 threshold
, session_consumed_size
);
3844 return session_consumed_size
>= threshold
;
3848 int evaluate_buffer_condition(const struct lttng_condition
*condition
,
3849 struct lttng_evaluation
**evaluation
,
3850 const struct notification_thread_state
*state
,
3851 const struct channel_state_sample
*previous_sample
,
3852 const struct channel_state_sample
*latest_sample
,
3853 uint64_t previous_session_consumed_total
,
3854 uint64_t latest_session_consumed_total
,
3855 struct channel_info
*channel_info
)
3858 enum lttng_condition_type condition_type
;
3859 const bool previous_sample_available
= !!previous_sample
;
3860 bool previous_sample_result
= false;
3861 bool latest_sample_result
;
3863 condition_type
= lttng_condition_get_type(condition
);
3865 switch (condition_type
) {
3866 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
3867 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
3868 if (caa_likely(previous_sample_available
)) {
3869 previous_sample_result
=
3870 evaluate_buffer_usage_condition(condition
,
3871 previous_sample
, channel_info
->capacity
);
3873 latest_sample_result
= evaluate_buffer_usage_condition(
3874 condition
, latest_sample
,
3875 channel_info
->capacity
);
3877 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
3878 if (caa_likely(previous_sample_available
)) {
3879 previous_sample_result
=
3880 evaluate_session_consumed_size_condition(
3882 previous_session_consumed_total
);
3884 latest_sample_result
=
3885 evaluate_session_consumed_size_condition(
3887 latest_session_consumed_total
);
3890 /* Unknown condition type; internal error. */
3894 if (!latest_sample_result
||
3895 (previous_sample_result
== latest_sample_result
)) {
3897 * Only trigger on a condition evaluation transition.
3899 * NOTE: This edge-triggered logic may not be appropriate for
3900 * future condition types.
3905 if (!evaluation
|| !latest_sample_result
) {
3909 switch (condition_type
) {
3910 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
3911 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
3912 *evaluation
= lttng_evaluation_buffer_usage_create(
3914 latest_sample
->highest_usage
,
3915 channel_info
->capacity
);
3917 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
3918 *evaluation
= lttng_evaluation_session_consumed_size_create(
3919 latest_session_consumed_total
);
3934 int client_notification_overflow(struct notification_client
*client
)
3937 const struct lttng_notification_channel_message msg
= {
3938 .type
= (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED
,
3941 ASSERT_LOCKED(client
->lock
);
3943 DBG("Dropping notification addressed to client (socket fd = %i)",
3945 if (client
->communication
.outbound
.dropped_notification
) {
3947 * The client already has a "notification dropped" message
3948 * in its outgoing queue. Nothing to do since all
3949 * of those messages are coalesced.
3954 client
->communication
.outbound
.dropped_notification
= true;
3955 ret
= lttng_dynamic_buffer_append(
3956 &client
->communication
.outbound
.payload
.buffer
, &msg
,
3959 PERROR("Failed to enqueue \"dropped notification\" message in client's (socket fd = %i) outgoing queue",
3966 static int client_handle_transmission_status_wrapper(
3967 struct notification_client
*client
,
3968 enum client_transmission_status status
,
3971 return client_handle_transmission_status(client
, status
,
3972 (struct notification_thread_state
*) user_data
);
3976 int send_evaluation_to_clients(const struct lttng_trigger
*trigger
,
3977 const struct lttng_evaluation
*evaluation
,
3978 struct notification_client_list
* client_list
,
3979 struct notification_thread_state
*state
,
3980 uid_t object_uid
, gid_t object_gid
)
3982 const struct lttng_credentials creds
= {
3983 .uid
= LTTNG_OPTIONAL_INIT_VALUE(object_uid
),
3984 .gid
= LTTNG_OPTIONAL_INIT_VALUE(object_gid
),
3987 return notification_client_list_send_evaluation(client_list
,
3988 lttng_trigger_get_const_condition(trigger
), evaluation
,
3989 lttng_trigger_get_credentials(trigger
),
3991 client_handle_transmission_status_wrapper
, state
);
3995 * Permission checks relative to notification channel clients are performed
3996 * here. Notice how object, client, and trigger credentials are involved in
3999 * The `object` credentials are the credentials associated with the "subject"
4000 * of a condition. For instance, a `rotation completed` condition applies
4001 * to a session. When that condition is met, it will produce an evaluation
4002 * against a session. Hence, in this case, the `object` credentials are the
4003 * credentials of the "subject" session.
4005 * The `trigger` credentials are the credentials of the user that registered the
4008 * The `client` credentials are the credentials of the user that created a given
4009 * notification channel.
4011 * In terms of visibility, it is expected that non-privilieged users can only
4012 * register triggers against "their" objects (their own sessions and
4013 * applications they are allowed to interact with). They can then open a
4014 * notification channel and subscribe to notifications associated with those
4017 * As for privilieged users, they can register triggers against the objects of
4018 * other users. They can then subscribe to the notifications associated to their
4019 * triggers. Privilieged users _can't_ subscribe to the notifications of
4020 * triggers owned by other users; they must create their own triggers.
4022 * This is more a concern of usability than security. It would be difficult for
4023 * a root user reliably subscribe to a specific set of conditions without
4024 * interference from external users (those could, for instance, unregister
4028 int notification_client_list_send_evaluation(
4029 struct notification_client_list
*client_list
,
4030 const struct lttng_condition
*condition
,
4031 const struct lttng_evaluation
*evaluation
,
4032 const struct lttng_credentials
*trigger_creds
,
4033 const struct lttng_credentials
*source_object_creds
,
4034 report_client_transmission_result_cb client_report
,
4038 struct lttng_payload msg_payload
;
4039 struct notification_client_list_element
*client_list_element
, *tmp
;
4040 const struct lttng_notification notification
= {
4041 .condition
= (struct lttng_condition
*) condition
,
4042 .evaluation
= (struct lttng_evaluation
*) evaluation
,
4044 struct lttng_notification_channel_message msg_header
= {
4045 .type
= (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION
,
4048 lttng_payload_init(&msg_payload
);
4050 ret
= lttng_dynamic_buffer_append(&msg_payload
.buffer
, &msg_header
,
4051 sizeof(msg_header
));
4056 ret
= lttng_notification_serialize(¬ification
, &msg_payload
);
4058 ERR("[notification-thread] Failed to serialize notification");
4063 /* Update payload size. */
4064 ((struct lttng_notification_channel_message
*) msg_payload
.buffer
.data
)
4065 ->size
= (uint32_t)(
4066 msg_payload
.buffer
.size
- sizeof(msg_header
));
4068 /* Update the payload number of fds. */
4070 const struct lttng_payload_view pv
= lttng_payload_view_from_payload(
4071 &msg_payload
, 0, -1);
4073 ((struct lttng_notification_channel_message
*)
4074 msg_payload
.buffer
.data
)->fds
= (uint32_t)
4075 lttng_payload_view_get_fd_handle_count(&pv
);
4078 pthread_mutex_lock(&client_list
->lock
);
4079 cds_list_for_each_entry_safe(client_list_element
, tmp
,
4080 &client_list
->list
, node
) {
4081 enum client_transmission_status transmission_status
;
4082 struct notification_client
*client
=
4083 client_list_element
->client
;
4086 pthread_mutex_lock(&client
->lock
);
4087 if (!client
->communication
.active
) {
4089 * Skip inactive client (protocol error or
4092 DBG("Skipping client at it is marked as inactive");
4096 if (source_object_creds
) {
4097 if (client
->uid
!= lttng_credentials_get_uid(source_object_creds
) &&
4098 client
->gid
!= lttng_credentials_get_gid(source_object_creds
) &&
4101 * Client is not allowed to monitor this
4104 DBG("[notification-thread] Skipping client at it does not have the object permission to receive notification for this trigger");
4109 if (client
->uid
!= lttng_credentials_get_uid(trigger_creds
) && client
->gid
!= lttng_credentials_get_gid(trigger_creds
)) {
4110 DBG("[notification-thread] Skipping client at it does not have the permission to receive notification for this trigger");
4114 DBG("[notification-thread] Sending notification to client (fd = %i, %zu bytes)",
4115 client
->socket
, msg_payload
.buffer
.size
);
4117 if (client_has_outbound_data_left(client
)) {
4119 * Outgoing data is already buffered for this client;
4120 * drop the notification and enqueue a "dropped
4121 * notification" message if this is the first dropped
4122 * notification since the socket spilled-over to the
4125 ret
= client_notification_overflow(client
);
4132 ret
= lttng_payload_copy(&msg_payload
, &client
->communication
.outbound
.payload
);
4138 transmission_status
= client_flush_outgoing_queue(client
);
4139 pthread_mutex_unlock(&client
->lock
);
4140 ret
= client_report(client
, transmission_status
, user_data
);
4143 goto end_unlock_list
;
4149 pthread_mutex_unlock(&client
->lock
);
4152 goto end_unlock_list
;
4158 pthread_mutex_unlock(&client_list
->lock
);
4160 lttng_payload_reset(&msg_payload
);
4164 static struct lttng_event_notifier_notification
*receive_notification(
4165 int notification_pipe_read_fd
, enum lttng_domain_type domain
)
4169 struct lttng_event_notifier_notification
*notification
= NULL
;
4170 void *reception_buffer
;
4171 size_t reception_size
;
4173 struct lttng_ust_event_notifier_notification ust_notification
;
4174 struct lttng_kernel_event_notifier_notification kernel_notification
;
4176 /* Init lttng_event_notifier_notification */
4178 case LTTNG_DOMAIN_UST
:
4179 reception_buffer
= (void *) &ust_notification
;
4180 reception_size
= sizeof(ust_notification
);
4182 case LTTNG_DOMAIN_KERNEL
:
4183 reception_buffer
= (void *) &kernel_notification
;
4184 reception_size
= sizeof(kernel_notification
);
4191 * The monitoring pipe only holds messages smaller than PIPE_BUF,
4192 * ensuring that read/write of tracer notifications are atomic.
4194 ret
= lttng_read(notification_pipe_read_fd
, reception_buffer
,
4196 if (ret
!= reception_size
) {
4197 PERROR("Failed to read from event source notification pipe: fd = %d, size to read = %zu, ret = %d",
4198 notification_pipe_read_fd
, reception_size
, ret
);
4204 case LTTNG_DOMAIN_UST
:
4205 token
= ust_notification
.token
;
4207 case LTTNG_DOMAIN_KERNEL
:
4208 token
= kernel_notification
.token
;
4214 notification
= lttng_event_notifier_notification_create(
4217 return notification
;
4220 int handle_notification_thread_event_notification(struct notification_thread_state
*state
,
4222 enum lttng_domain_type domain
)
4225 enum lttng_trigger_status trigger_status
;
4226 struct cds_lfht_node
*node
;
4227 struct cds_lfht_iter iter
;
4228 struct notification_trigger_tokens_ht_element
*element
;
4229 struct lttng_evaluation
*evaluation
= NULL
;
4230 struct lttng_event_notifier_notification
*notification
= NULL
;
4231 enum action_executor_status executor_status
;
4232 struct notification_client_list
*client_list
= NULL
;
4233 const char *trigger_name
;
4235 notification
= receive_notification(pipe
, domain
);
4236 if (notification
== NULL
) {
4237 ERR("[notification-thread] Error receiving notification from tracer (fd = %i, domain = %s)",
4238 pipe
, lttng_domain_type_str(domain
));
4243 /* Find triggers associated with this token. */
4245 cds_lfht_lookup(state
->trigger_tokens_ht
,
4246 hash_key_u64(¬ification
->tracer_token
, lttng_ht_seed
),
4247 match_trigger_token
, ¬ification
->tracer_token
, &iter
);
4248 node
= cds_lfht_iter_get_node(&iter
);
4249 if (caa_unlikely(!node
)) {
4251 * This is not an error, slow consumption of the tracer
4252 * notifications can lead to situations where a trigger is
4253 * removed but we still get tracer notifications matching a
4254 * trigger that no longer exists.
4260 element
= caa_container_of(node
,
4261 struct notification_trigger_tokens_ht_element
,
4264 if (!lttng_trigger_should_fire(element
->trigger
)) {
4269 lttng_trigger_fire(element
->trigger
);
4271 trigger_status
= lttng_trigger_get_name(element
->trigger
, &trigger_name
);
4272 assert(trigger_status
== LTTNG_TRIGGER_STATUS_OK
);
4274 evaluation
= lttng_evaluation_event_rule_create(
4276 if (evaluation
== NULL
) {
4277 ERR("[notification-thread] Failed to create event rule hit evaluation while creating and enqueuing action executor job");
4282 client_list
= get_client_list_from_condition(state
,
4283 lttng_trigger_get_const_condition(element
->trigger
));
4284 executor_status
= action_executor_enqueue(state
->executor
,
4285 element
->trigger
, evaluation
, NULL
, client_list
);
4286 switch (executor_status
) {
4287 case ACTION_EXECUTOR_STATUS_OK
:
4290 case ACTION_EXECUTOR_STATUS_OVERFLOW
:
4292 struct notification_client_list_element
*client_list_element
,
4296 * Not a fatal error; this is expected and simply means the
4297 * executor has too much work queued already.
4301 /* No clients subscribed to notifications for this trigger. */
4306 /* Warn clients that a notification (or more) was dropped. */
4307 pthread_mutex_lock(&client_list
->lock
);
4308 cds_list_for_each_entry_safe(client_list_element
, tmp
,
4309 &client_list
->list
, node
) {
4310 enum client_transmission_status transmission_status
;
4311 struct notification_client
*client
=
4312 client_list_element
->client
;
4314 pthread_mutex_lock(&client
->lock
);
4315 ret
= client_notification_overflow(client
);
4321 transmission_status
=
4322 client_flush_outgoing_queue(client
);
4323 ret
= client_handle_transmission_status(
4324 client
, transmission_status
, state
);
4330 pthread_mutex_unlock(&client
->lock
);
4336 pthread_mutex_unlock(&client_list
->lock
);
4339 case ACTION_EXECUTOR_STATUS_ERROR
:
4340 /* Fatal error, shut down everything. */
4341 ERR("Fatal error encoutered while enqueuing action to the action executor");
4345 /* Unhandled error. */
4350 lttng_event_notifier_notification_destroy(notification
);
4351 notification_client_list_put(client_list
);
4357 int handle_notification_thread_channel_sample(
4358 struct notification_thread_state
*state
, int pipe
,
4359 enum lttng_domain_type domain
)
4362 struct lttcomm_consumer_channel_monitor_msg sample_msg
;
4363 struct channel_info
*channel_info
;
4364 struct cds_lfht_node
*node
;
4365 struct cds_lfht_iter iter
;
4366 struct lttng_channel_trigger_list
*trigger_list
;
4367 struct lttng_trigger_list_element
*trigger_list_element
;
4368 bool previous_sample_available
= false;
4369 struct channel_state_sample previous_sample
, latest_sample
;
4370 uint64_t previous_session_consumed_total
, latest_session_consumed_total
;
4371 struct lttng_credentials channel_creds
;
4374 * The monitoring pipe only holds messages smaller than PIPE_BUF,
4375 * ensuring that read/write of sampling messages are atomic.
4377 ret
= lttng_read(pipe
, &sample_msg
, sizeof(sample_msg
));
4378 if (ret
!= sizeof(sample_msg
)) {
4379 ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
4386 latest_sample
.key
.key
= sample_msg
.key
;
4387 latest_sample
.key
.domain
= domain
;
4388 latest_sample
.highest_usage
= sample_msg
.highest
;
4389 latest_sample
.lowest_usage
= sample_msg
.lowest
;
4390 latest_sample
.channel_total_consumed
= sample_msg
.total_consumed
;
4394 /* Retrieve the channel's informations */
4395 cds_lfht_lookup(state
->channels_ht
,
4396 hash_channel_key(&latest_sample
.key
),
4400 node
= cds_lfht_iter_get_node(&iter
);
4401 if (caa_unlikely(!node
)) {
4403 * Not an error since the consumer can push a sample to the pipe
4404 * and the rest of the session daemon could notify us of the
4405 * channel's destruction before we get a chance to process that
4408 DBG("[notification-thread] Received a sample for an unknown channel from consumerd, key = %" PRIu64
" in %s domain",
4409 latest_sample
.key
.key
,
4410 lttng_domain_type_str(domain
));
4413 channel_info
= caa_container_of(node
, struct channel_info
,
4415 DBG("[notification-thread] Handling channel sample for channel %s (key = %" PRIu64
") in session %s (highest usage = %" PRIu64
", lowest usage = %" PRIu64
", total consumed = %" PRIu64
")",
4417 latest_sample
.key
.key
,
4418 channel_info
->session_info
->name
,
4419 latest_sample
.highest_usage
,
4420 latest_sample
.lowest_usage
,
4421 latest_sample
.channel_total_consumed
);
4423 previous_session_consumed_total
=
4424 channel_info
->session_info
->consumed_data_size
;
4426 /* Retrieve the channel's last sample, if it exists, and update it. */
4427 cds_lfht_lookup(state
->channel_state_ht
,
4428 hash_channel_key(&latest_sample
.key
),
4429 match_channel_state_sample
,
4432 node
= cds_lfht_iter_get_node(&iter
);
4433 if (caa_likely(node
)) {
4434 struct channel_state_sample
*stored_sample
;
4436 /* Update the sample stored. */
4437 stored_sample
= caa_container_of(node
,
4438 struct channel_state_sample
,
4439 channel_state_ht_node
);
4441 memcpy(&previous_sample
, stored_sample
,
4442 sizeof(previous_sample
));
4443 stored_sample
->highest_usage
= latest_sample
.highest_usage
;
4444 stored_sample
->lowest_usage
= latest_sample
.lowest_usage
;
4445 stored_sample
->channel_total_consumed
= latest_sample
.channel_total_consumed
;
4446 previous_sample_available
= true;
4448 latest_session_consumed_total
=
4449 previous_session_consumed_total
+
4450 (latest_sample
.channel_total_consumed
- previous_sample
.channel_total_consumed
);
4453 * This is the channel's first sample, allocate space for and
4454 * store the new sample.
4456 struct channel_state_sample
*stored_sample
;
4458 stored_sample
= zmalloc(sizeof(*stored_sample
));
4459 if (!stored_sample
) {
4464 memcpy(stored_sample
, &latest_sample
, sizeof(*stored_sample
));
4465 cds_lfht_node_init(&stored_sample
->channel_state_ht_node
);
4466 cds_lfht_add(state
->channel_state_ht
,
4467 hash_channel_key(&stored_sample
->key
),
4468 &stored_sample
->channel_state_ht_node
);
4470 latest_session_consumed_total
=
4471 previous_session_consumed_total
+
4472 latest_sample
.channel_total_consumed
;
4475 channel_info
->session_info
->consumed_data_size
=
4476 latest_session_consumed_total
;
4478 /* Find triggers associated with this channel. */
4479 cds_lfht_lookup(state
->channel_triggers_ht
,
4480 hash_channel_key(&latest_sample
.key
),
4481 match_channel_trigger_list
,
4484 node
= cds_lfht_iter_get_node(&iter
);
4485 if (caa_likely(!node
)) {
4489 channel_creds
= (typeof(channel_creds
)) {
4490 .uid
= LTTNG_OPTIONAL_INIT_VALUE(channel_info
->session_info
->uid
),
4491 .gid
= LTTNG_OPTIONAL_INIT_VALUE(channel_info
->session_info
->gid
),
4494 trigger_list
= caa_container_of(node
, struct lttng_channel_trigger_list
,
4495 channel_triggers_ht_node
);
4496 cds_list_for_each_entry(trigger_list_element
, &trigger_list
->list
,
4498 const struct lttng_condition
*condition
;
4499 struct lttng_trigger
*trigger
;
4500 struct notification_client_list
*client_list
= NULL
;
4501 struct lttng_evaluation
*evaluation
= NULL
;
4502 enum action_executor_status executor_status
;
4505 trigger
= trigger_list_element
->trigger
;
4506 condition
= lttng_trigger_get_const_condition(trigger
);
4510 * Check if any client is subscribed to the result of this
4513 client_list
= get_client_list_from_condition(state
, condition
);
4515 ret
= evaluate_buffer_condition(condition
, &evaluation
, state
,
4516 previous_sample_available
? &previous_sample
: NULL
,
4518 previous_session_consumed_total
,
4519 latest_session_consumed_total
,
4521 if (caa_unlikely(ret
)) {
4525 if (caa_likely(!evaluation
)) {
4529 if (!lttng_trigger_should_fire(trigger
)) {
4533 lttng_trigger_fire(trigger
);
4536 * Ownership of `evaluation` transferred to the action executor
4537 * no matter the result.
4539 executor_status
= action_executor_enqueue(state
->executor
,
4540 trigger
, evaluation
, &channel_creds
,
4543 switch (executor_status
) {
4544 case ACTION_EXECUTOR_STATUS_OK
:
4546 case ACTION_EXECUTOR_STATUS_ERROR
:
4547 case ACTION_EXECUTOR_STATUS_INVALID
:
4549 * TODO Add trigger identification (name/id) when
4550 * it is added to the API.
4552 ERR("Fatal error occurred while enqueuing action associated with buffer-condition trigger");
4555 case ACTION_EXECUTOR_STATUS_OVERFLOW
:
4557 * TODO Add trigger identification (name/id) when
4558 * it is added to the API.
4560 * Not a fatal error.
4562 WARN("No space left when enqueuing action associated with buffer-condition trigger");
4570 notification_client_list_put(client_list
);
4571 if (caa_unlikely(ret
)) {