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/notification/channel-internal.h>
37 #include "notification-thread.h"
38 #include "notification-thread-events.h"
39 #include "notification-thread-commands.h"
40 #include "lttng-sessiond.h"
43 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
44 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
46 enum lttng_object_type
{
47 LTTNG_OBJECT_TYPE_UNKNOWN
,
48 LTTNG_OBJECT_TYPE_NONE
,
49 LTTNG_OBJECT_TYPE_CHANNEL
,
50 LTTNG_OBJECT_TYPE_SESSION
,
53 struct lttng_trigger_list_element
{
54 /* No ownership of the trigger object is assumed. */
55 struct lttng_trigger
*trigger
;
56 struct cds_list_head node
;
59 struct lttng_channel_trigger_list
{
60 struct channel_key channel_key
;
61 /* List of struct lttng_trigger_list_element. */
62 struct cds_list_head list
;
63 /* Node in the channel_triggers_ht */
64 struct cds_lfht_node channel_triggers_ht_node
;
65 /* call_rcu delayed reclaim. */
66 struct rcu_head rcu_node
;
70 * List of triggers applying to a given session.
73 * - lttng_session_trigger_list_create()
74 * - lttng_session_trigger_list_build()
75 * - lttng_session_trigger_list_destroy()
76 * - lttng_session_trigger_list_add()
78 struct lttng_session_trigger_list
{
80 * Not owned by this; points to the session_info structure's
83 const char *session_name
;
84 /* List of struct lttng_trigger_list_element. */
85 struct cds_list_head list
;
86 /* Node in the session_triggers_ht */
87 struct cds_lfht_node session_triggers_ht_node
;
89 * Weak reference to the notification system's session triggers
92 * The session trigger list structure structure is owned by
93 * the session's session_info.
95 * The session_info is kept alive the the channel_infos holding a
96 * reference to it (reference counting). When those channels are
97 * destroyed (at runtime or on teardown), the reference they hold
98 * to the session_info are released. On destruction of session_info,
99 * session_info_destroy() will remove the list of triggers applying
100 * to this session from the notification system's state.
102 * This implies that the session_triggers_ht must be destroyed
103 * after the channels.
105 struct cds_lfht
*session_triggers_ht
;
106 /* Used for delayed RCU reclaim. */
107 struct rcu_head rcu_node
;
110 struct lttng_trigger_ht_element
{
111 struct lttng_trigger
*trigger
;
112 struct cds_lfht_node node
;
113 /* call_rcu delayed reclaim. */
114 struct rcu_head rcu_node
;
117 struct lttng_condition_list_element
{
118 struct lttng_condition
*condition
;
119 struct cds_list_head node
;
122 struct channel_state_sample
{
123 struct channel_key key
;
124 struct cds_lfht_node channel_state_ht_node
;
125 uint64_t highest_usage
;
126 uint64_t lowest_usage
;
127 uint64_t channel_total_consumed
;
128 /* call_rcu delayed reclaim. */
129 struct rcu_head rcu_node
;
132 static unsigned long hash_channel_key(struct channel_key
*key
);
133 static int evaluate_buffer_condition(const struct lttng_condition
*condition
,
134 struct lttng_evaluation
**evaluation
,
135 const struct notification_thread_state
*state
,
136 const struct channel_state_sample
*previous_sample
,
137 const struct channel_state_sample
*latest_sample
,
138 uint64_t previous_session_consumed_total
,
139 uint64_t latest_session_consumed_total
,
140 struct channel_info
*channel_info
);
142 int send_evaluation_to_clients(const struct lttng_trigger
*trigger
,
143 const struct lttng_evaluation
*evaluation
,
144 struct notification_client_list
*client_list
,
145 struct notification_thread_state
*state
,
146 uid_t channel_uid
, gid_t channel_gid
);
149 /* session_info API */
151 void session_info_destroy(void *_data
);
153 void session_info_get(struct session_info
*session_info
);
155 void session_info_put(struct session_info
*session_info
);
157 struct session_info
*session_info_create(const char *name
,
158 uid_t uid
, gid_t gid
,
159 struct lttng_session_trigger_list
*trigger_list
,
160 struct cds_lfht
*sessions_ht
);
162 void session_info_add_channel(struct session_info
*session_info
,
163 struct channel_info
*channel_info
);
165 void session_info_remove_channel(struct session_info
*session_info
,
166 struct channel_info
*channel_info
);
168 /* lttng_session_trigger_list API */
170 struct lttng_session_trigger_list
*lttng_session_trigger_list_create(
171 const char *session_name
,
172 struct cds_lfht
*session_triggers_ht
);
174 struct lttng_session_trigger_list
*lttng_session_trigger_list_build(
175 const struct notification_thread_state
*state
,
176 const char *session_name
);
178 void lttng_session_trigger_list_destroy(
179 struct lttng_session_trigger_list
*list
);
181 int lttng_session_trigger_list_add(struct lttng_session_trigger_list
*list
,
182 struct lttng_trigger
*trigger
);
185 int client_handle_transmission_status(
186 struct notification_client
*client
,
187 enum client_transmission_status transmission_status
,
188 struct notification_thread_state
*state
);
191 int match_client_socket(struct cds_lfht_node
*node
, const void *key
)
193 /* This double-cast is intended to supress pointer-to-cast warning. */
194 const int socket
= (int) (intptr_t) key
;
195 const struct notification_client
*client
= caa_container_of(node
,
196 struct notification_client
, client_socket_ht_node
);
198 return client
->socket
== socket
;
202 int match_client_id(struct cds_lfht_node
*node
, const void *key
)
204 /* This double-cast is intended to supress pointer-to-cast warning. */
205 const notification_client_id id
= *((notification_client_id
*) key
);
206 const struct notification_client
*client
= caa_container_of(
207 node
, struct notification_client
, client_id_ht_node
);
209 return client
->id
== id
;
213 int match_channel_trigger_list(struct cds_lfht_node
*node
, const void *key
)
215 struct channel_key
*channel_key
= (struct channel_key
*) key
;
216 struct lttng_channel_trigger_list
*trigger_list
;
218 trigger_list
= caa_container_of(node
, struct lttng_channel_trigger_list
,
219 channel_triggers_ht_node
);
221 return !!((channel_key
->key
== trigger_list
->channel_key
.key
) &&
222 (channel_key
->domain
== trigger_list
->channel_key
.domain
));
226 int match_session_trigger_list(struct cds_lfht_node
*node
, const void *key
)
228 const char *session_name
= (const char *) key
;
229 struct lttng_session_trigger_list
*trigger_list
;
231 trigger_list
= caa_container_of(node
, struct lttng_session_trigger_list
,
232 session_triggers_ht_node
);
234 return !!(strcmp(trigger_list
->session_name
, session_name
) == 0);
238 int match_channel_state_sample(struct cds_lfht_node
*node
, const void *key
)
240 struct channel_key
*channel_key
= (struct channel_key
*) key
;
241 struct channel_state_sample
*sample
;
243 sample
= caa_container_of(node
, struct channel_state_sample
,
244 channel_state_ht_node
);
246 return !!((channel_key
->key
== sample
->key
.key
) &&
247 (channel_key
->domain
== sample
->key
.domain
));
251 int match_channel_info(struct cds_lfht_node
*node
, const void *key
)
253 struct channel_key
*channel_key
= (struct channel_key
*) key
;
254 struct channel_info
*channel_info
;
256 channel_info
= caa_container_of(node
, struct channel_info
,
259 return !!((channel_key
->key
== channel_info
->key
.key
) &&
260 (channel_key
->domain
== channel_info
->key
.domain
));
264 int match_condition(struct cds_lfht_node
*node
, const void *key
)
266 struct lttng_condition
*condition_key
= (struct lttng_condition
*) key
;
267 struct lttng_trigger_ht_element
*trigger
;
268 struct lttng_condition
*condition
;
270 trigger
= caa_container_of(node
, struct lttng_trigger_ht_element
,
272 condition
= lttng_trigger_get_condition(trigger
->trigger
);
275 return !!lttng_condition_is_equal(condition_key
, condition
);
279 int match_client_list_condition(struct cds_lfht_node
*node
, const void *key
)
281 struct lttng_condition
*condition_key
= (struct lttng_condition
*) key
;
282 struct notification_client_list
*client_list
;
283 const struct lttng_condition
*condition
;
285 assert(condition_key
);
287 client_list
= caa_container_of(node
, struct notification_client_list
,
288 notification_trigger_clients_ht_node
);
289 condition
= lttng_trigger_get_const_condition(client_list
->trigger
);
291 return !!lttng_condition_is_equal(condition_key
, condition
);
295 int match_session(struct cds_lfht_node
*node
, const void *key
)
297 const char *name
= key
;
298 struct session_info
*session_info
= caa_container_of(
299 node
, struct session_info
, sessions_ht_node
);
301 return !strcmp(session_info
->name
, name
);
305 unsigned long lttng_condition_buffer_usage_hash(
306 const struct lttng_condition
*_condition
)
309 unsigned long condition_type
;
310 struct lttng_condition_buffer_usage
*condition
;
312 condition
= container_of(_condition
,
313 struct lttng_condition_buffer_usage
, parent
);
315 condition_type
= (unsigned long) condition
->parent
.type
;
316 hash
= hash_key_ulong((void *) condition_type
, lttng_ht_seed
);
317 if (condition
->session_name
) {
318 hash
^= hash_key_str(condition
->session_name
, lttng_ht_seed
);
320 if (condition
->channel_name
) {
321 hash
^= hash_key_str(condition
->channel_name
, lttng_ht_seed
);
323 if (condition
->domain
.set
) {
324 hash
^= hash_key_ulong(
325 (void *) condition
->domain
.type
,
328 if (condition
->threshold_ratio
.set
) {
331 val
= condition
->threshold_ratio
.value
* (double) UINT32_MAX
;
332 hash
^= hash_key_u64(&val
, lttng_ht_seed
);
333 } else if (condition
->threshold_bytes
.set
) {
336 val
= condition
->threshold_bytes
.value
;
337 hash
^= hash_key_u64(&val
, lttng_ht_seed
);
343 unsigned long lttng_condition_session_consumed_size_hash(
344 const struct lttng_condition
*_condition
)
347 unsigned long condition_type
=
348 (unsigned long) LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
;
349 struct lttng_condition_session_consumed_size
*condition
;
352 condition
= container_of(_condition
,
353 struct lttng_condition_session_consumed_size
, parent
);
355 hash
= hash_key_ulong((void *) condition_type
, lttng_ht_seed
);
356 if (condition
->session_name
) {
357 hash
^= hash_key_str(condition
->session_name
, lttng_ht_seed
);
359 val
= condition
->consumed_threshold_bytes
.value
;
360 hash
^= hash_key_u64(&val
, lttng_ht_seed
);
365 unsigned long lttng_condition_session_rotation_hash(
366 const struct lttng_condition
*_condition
)
368 unsigned long hash
, condition_type
;
369 struct lttng_condition_session_rotation
*condition
;
371 condition
= container_of(_condition
,
372 struct lttng_condition_session_rotation
, parent
);
373 condition_type
= (unsigned long) condition
->parent
.type
;
374 hash
= hash_key_ulong((void *) condition_type
, lttng_ht_seed
);
375 assert(condition
->session_name
);
376 hash
^= hash_key_str(condition
->session_name
, lttng_ht_seed
);
381 * The lttng_condition hashing code is kept in this file (rather than
382 * condition.c) since it makes use of GPLv2 code (hashtable utils), which we
383 * don't want to link in liblttng-ctl.
386 unsigned long lttng_condition_hash(const struct lttng_condition
*condition
)
388 switch (condition
->type
) {
389 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
390 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
391 return lttng_condition_buffer_usage_hash(condition
);
392 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
393 return lttng_condition_session_consumed_size_hash(condition
);
394 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
395 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
396 return lttng_condition_session_rotation_hash(condition
);
398 ERR("[notification-thread] Unexpected condition type caught");
404 unsigned long hash_channel_key(struct channel_key
*key
)
406 unsigned long key_hash
= hash_key_u64(&key
->key
, lttng_ht_seed
);
407 unsigned long domain_hash
= hash_key_ulong(
408 (void *) (unsigned long) key
->domain
, lttng_ht_seed
);
410 return key_hash
^ domain_hash
;
414 unsigned long hash_client_socket(int socket
)
416 return hash_key_ulong((void *) (unsigned long) socket
, lttng_ht_seed
);
420 unsigned long hash_client_id(notification_client_id id
)
422 return hash_key_u64(&id
, lttng_ht_seed
);
426 * Get the type of object to which a given condition applies. Bindings let
427 * the notification system evaluate a trigger's condition when a given
428 * object's state is updated.
430 * For instance, a condition bound to a channel will be evaluated everytime
431 * the channel's state is changed by a channel monitoring sample.
434 enum lttng_object_type
get_condition_binding_object(
435 const struct lttng_condition
*condition
)
437 switch (lttng_condition_get_type(condition
)) {
438 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
439 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
440 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
441 return LTTNG_OBJECT_TYPE_CHANNEL
;
442 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
443 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
444 return LTTNG_OBJECT_TYPE_SESSION
;
446 return LTTNG_OBJECT_TYPE_UNKNOWN
;
451 void free_channel_info_rcu(struct rcu_head
*node
)
453 free(caa_container_of(node
, struct channel_info
, rcu_node
));
457 void channel_info_destroy(struct channel_info
*channel_info
)
463 if (channel_info
->session_info
) {
464 session_info_remove_channel(channel_info
->session_info
,
466 session_info_put(channel_info
->session_info
);
468 if (channel_info
->name
) {
469 free(channel_info
->name
);
471 call_rcu(&channel_info
->rcu_node
, free_channel_info_rcu
);
475 void free_session_info_rcu(struct rcu_head
*node
)
477 free(caa_container_of(node
, struct session_info
, rcu_node
));
480 /* Don't call directly, use the ref-counting mechanism. */
482 void session_info_destroy(void *_data
)
484 struct session_info
*session_info
= _data
;
487 assert(session_info
);
488 if (session_info
->channel_infos_ht
) {
489 ret
= cds_lfht_destroy(session_info
->channel_infos_ht
, NULL
);
491 ERR("[notification-thread] Failed to destroy channel information hash table");
494 lttng_session_trigger_list_destroy(session_info
->trigger_list
);
497 cds_lfht_del(session_info
->sessions_ht
,
498 &session_info
->sessions_ht_node
);
500 free(session_info
->name
);
501 call_rcu(&session_info
->rcu_node
, free_session_info_rcu
);
505 void session_info_get(struct session_info
*session_info
)
510 lttng_ref_get(&session_info
->ref
);
514 void session_info_put(struct session_info
*session_info
)
519 lttng_ref_put(&session_info
->ref
);
523 struct session_info
*session_info_create(const char *name
, uid_t uid
, gid_t gid
,
524 struct lttng_session_trigger_list
*trigger_list
,
525 struct cds_lfht
*sessions_ht
)
527 struct session_info
*session_info
;
531 session_info
= zmalloc(sizeof(*session_info
));
535 lttng_ref_init(&session_info
->ref
, session_info_destroy
);
537 session_info
->channel_infos_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
538 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
539 if (!session_info
->channel_infos_ht
) {
543 cds_lfht_node_init(&session_info
->sessions_ht_node
);
544 session_info
->name
= strdup(name
);
545 if (!session_info
->name
) {
548 session_info
->uid
= uid
;
549 session_info
->gid
= gid
;
550 session_info
->trigger_list
= trigger_list
;
551 session_info
->sessions_ht
= sessions_ht
;
555 session_info_put(session_info
);
560 void session_info_add_channel(struct session_info
*session_info
,
561 struct channel_info
*channel_info
)
564 cds_lfht_add(session_info
->channel_infos_ht
,
565 hash_channel_key(&channel_info
->key
),
566 &channel_info
->session_info_channels_ht_node
);
571 void session_info_remove_channel(struct session_info
*session_info
,
572 struct channel_info
*channel_info
)
575 cds_lfht_del(session_info
->channel_infos_ht
,
576 &channel_info
->session_info_channels_ht_node
);
581 struct channel_info
*channel_info_create(const char *channel_name
,
582 struct channel_key
*channel_key
, uint64_t channel_capacity
,
583 struct session_info
*session_info
)
585 struct channel_info
*channel_info
= zmalloc(sizeof(*channel_info
));
591 cds_lfht_node_init(&channel_info
->channels_ht_node
);
592 cds_lfht_node_init(&channel_info
->session_info_channels_ht_node
);
593 memcpy(&channel_info
->key
, channel_key
, sizeof(*channel_key
));
594 channel_info
->capacity
= channel_capacity
;
596 channel_info
->name
= strdup(channel_name
);
597 if (!channel_info
->name
) {
602 * Set the references between session and channel infos:
603 * - channel_info holds a strong reference to session_info
604 * - session_info holds a weak reference to channel_info
606 session_info_get(session_info
);
607 session_info_add_channel(session_info
, channel_info
);
608 channel_info
->session_info
= session_info
;
612 channel_info_destroy(channel_info
);
617 bool notification_client_list_get(struct notification_client_list
*list
)
619 return urcu_ref_get_unless_zero(&list
->ref
);
623 void free_notification_client_list_rcu(struct rcu_head
*node
)
625 free(caa_container_of(node
, struct notification_client_list
,
630 void notification_client_list_release(struct urcu_ref
*list_ref
)
632 struct notification_client_list
*list
=
633 container_of(list_ref
, typeof(*list
), ref
);
634 struct notification_client_list_element
*client_list_element
, *tmp
;
636 if (list
->notification_trigger_clients_ht
) {
638 cds_lfht_del(list
->notification_trigger_clients_ht
,
639 &list
->notification_trigger_clients_ht_node
);
641 list
->notification_trigger_clients_ht
= NULL
;
643 cds_list_for_each_entry_safe(client_list_element
, tmp
,
645 free(client_list_element
);
647 pthread_mutex_destroy(&list
->lock
);
648 call_rcu(&list
->rcu_node
, free_notification_client_list_rcu
);
652 struct notification_client_list
*notification_client_list_create(
653 const struct lttng_trigger
*trigger
)
655 struct notification_client_list
*client_list
=
656 zmalloc(sizeof(*client_list
));
661 pthread_mutex_init(&client_list
->lock
, NULL
);
662 urcu_ref_init(&client_list
->ref
);
663 cds_lfht_node_init(&client_list
->notification_trigger_clients_ht_node
);
664 CDS_INIT_LIST_HEAD(&client_list
->list
);
665 client_list
->trigger
= trigger
;
671 void publish_notification_client_list(
672 struct notification_thread_state
*state
,
673 struct notification_client_list
*list
)
675 const struct lttng_condition
*condition
=
676 lttng_trigger_get_const_condition(list
->trigger
);
678 assert(!list
->notification_trigger_clients_ht
);
679 notification_client_list_get(list
);
681 list
->notification_trigger_clients_ht
=
682 state
->notification_trigger_clients_ht
;
685 cds_lfht_add(state
->notification_trigger_clients_ht
,
686 lttng_condition_hash(condition
),
687 &list
->notification_trigger_clients_ht_node
);
692 void notification_client_list_put(struct notification_client_list
*list
)
697 return urcu_ref_put(&list
->ref
, notification_client_list_release
);
700 /* Provides a reference to the returned list. */
702 struct notification_client_list
*get_client_list_from_condition(
703 struct notification_thread_state
*state
,
704 const struct lttng_condition
*condition
)
706 struct cds_lfht_node
*node
;
707 struct cds_lfht_iter iter
;
708 struct notification_client_list
*list
= NULL
;
711 cds_lfht_lookup(state
->notification_trigger_clients_ht
,
712 lttng_condition_hash(condition
),
713 match_client_list_condition
,
716 node
= cds_lfht_iter_get_node(&iter
);
718 list
= container_of(node
, struct notification_client_list
,
719 notification_trigger_clients_ht_node
);
720 list
= notification_client_list_get(list
) ? list
: NULL
;
728 int evaluate_channel_condition_for_client(
729 const struct lttng_condition
*condition
,
730 struct notification_thread_state
*state
,
731 struct lttng_evaluation
**evaluation
,
732 uid_t
*session_uid
, gid_t
*session_gid
)
735 struct cds_lfht_iter iter
;
736 struct cds_lfht_node
*node
;
737 struct channel_info
*channel_info
= NULL
;
738 struct channel_key
*channel_key
= NULL
;
739 struct channel_state_sample
*last_sample
= NULL
;
740 struct lttng_channel_trigger_list
*channel_trigger_list
= NULL
;
744 /* Find the channel associated with the condition. */
745 cds_lfht_for_each_entry(state
->channel_triggers_ht
, &iter
,
746 channel_trigger_list
, channel_triggers_ht_node
) {
747 struct lttng_trigger_list_element
*element
;
749 cds_list_for_each_entry(element
, &channel_trigger_list
->list
, node
) {
750 const struct lttng_condition
*current_condition
=
751 lttng_trigger_get_const_condition(
754 assert(current_condition
);
755 if (!lttng_condition_is_equal(condition
,
756 current_condition
)) {
760 /* Found the trigger, save the channel key. */
761 channel_key
= &channel_trigger_list
->channel_key
;
765 /* The channel key was found stop iteration. */
771 /* No channel found; normal exit. */
772 DBG("[notification-thread] No known channel associated with newly subscribed-to condition");
777 /* Fetch channel info for the matching channel. */
778 cds_lfht_lookup(state
->channels_ht
,
779 hash_channel_key(channel_key
),
783 node
= cds_lfht_iter_get_node(&iter
);
785 channel_info
= caa_container_of(node
, struct channel_info
,
788 /* Retrieve the channel's last sample, if it exists. */
789 cds_lfht_lookup(state
->channel_state_ht
,
790 hash_channel_key(channel_key
),
791 match_channel_state_sample
,
794 node
= cds_lfht_iter_get_node(&iter
);
796 last_sample
= caa_container_of(node
,
797 struct channel_state_sample
,
798 channel_state_ht_node
);
800 /* Nothing to evaluate, no sample was ever taken. Normal exit */
801 DBG("[notification-thread] No channel sample associated with newly subscribed-to condition");
806 ret
= evaluate_buffer_condition(condition
, evaluation
, state
,
808 0, channel_info
->session_info
->consumed_data_size
,
811 WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition");
815 *session_uid
= channel_info
->session_info
->uid
;
816 *session_gid
= channel_info
->session_info
->gid
;
823 const char *get_condition_session_name(const struct lttng_condition
*condition
)
825 const char *session_name
= NULL
;
826 enum lttng_condition_status status
;
828 switch (lttng_condition_get_type(condition
)) {
829 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
830 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
831 status
= lttng_condition_buffer_usage_get_session_name(
832 condition
, &session_name
);
834 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
835 status
= lttng_condition_session_consumed_size_get_session_name(
836 condition
, &session_name
);
838 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
839 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
840 status
= lttng_condition_session_rotation_get_session_name(
841 condition
, &session_name
);
846 if (status
!= LTTNG_CONDITION_STATUS_OK
) {
847 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
855 int evaluate_session_condition_for_client(
856 const struct lttng_condition
*condition
,
857 struct notification_thread_state
*state
,
858 struct lttng_evaluation
**evaluation
,
859 uid_t
*session_uid
, gid_t
*session_gid
)
862 struct cds_lfht_iter iter
;
863 struct cds_lfht_node
*node
;
864 const char *session_name
;
865 struct session_info
*session_info
= NULL
;
868 session_name
= get_condition_session_name(condition
);
870 /* Find the session associated with the trigger. */
871 cds_lfht_lookup(state
->sessions_ht
,
872 hash_key_str(session_name
, lttng_ht_seed
),
876 node
= cds_lfht_iter_get_node(&iter
);
878 DBG("[notification-thread] No known session matching name \"%s\"",
884 session_info
= caa_container_of(node
, struct session_info
,
886 session_info_get(session_info
);
889 * Evaluation is performed in-line here since only one type of
890 * session-bound condition is handled for the moment.
892 switch (lttng_condition_get_type(condition
)) {
893 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
894 if (!session_info
->rotation
.ongoing
) {
896 goto end_session_put
;
899 *evaluation
= lttng_evaluation_session_rotation_ongoing_create(
900 session_info
->rotation
.id
);
903 ERR("[notification-thread] Failed to create session rotation ongoing evaluation for session \"%s\"",
906 goto end_session_put
;
912 goto end_session_put
;
915 *session_uid
= session_info
->uid
;
916 *session_gid
= session_info
->gid
;
919 session_info_put(session_info
);
926 int evaluate_condition_for_client(const struct lttng_trigger
*trigger
,
927 const struct lttng_condition
*condition
,
928 struct notification_client
*client
,
929 struct notification_thread_state
*state
)
932 struct lttng_evaluation
*evaluation
= NULL
;
933 struct notification_client_list client_list
= {
934 .lock
= PTHREAD_MUTEX_INITIALIZER
,
936 struct notification_client_list_element client_list_element
= { 0 };
937 uid_t object_uid
= 0;
938 gid_t object_gid
= 0;
945 switch (get_condition_binding_object(condition
)) {
946 case LTTNG_OBJECT_TYPE_SESSION
:
947 ret
= evaluate_session_condition_for_client(condition
, state
,
948 &evaluation
, &object_uid
, &object_gid
);
950 case LTTNG_OBJECT_TYPE_CHANNEL
:
951 ret
= evaluate_channel_condition_for_client(condition
, state
,
952 &evaluation
, &object_uid
, &object_gid
);
954 case LTTNG_OBJECT_TYPE_NONE
:
957 case LTTNG_OBJECT_TYPE_UNKNOWN
:
967 /* Evaluation yielded nothing. Normal exit. */
968 DBG("[notification-thread] Newly subscribed-to condition evaluated to false, nothing to report to client");
974 * Create a temporary client list with the client currently
977 cds_lfht_node_init(&client_list
.notification_trigger_clients_ht_node
);
978 CDS_INIT_LIST_HEAD(&client_list
.list
);
979 client_list
.trigger
= trigger
;
981 CDS_INIT_LIST_HEAD(&client_list_element
.node
);
982 client_list_element
.client
= client
;
983 cds_list_add(&client_list_element
.node
, &client_list
.list
);
985 /* Send evaluation result to the newly-subscribed client. */
986 DBG("[notification-thread] Newly subscribed-to condition evaluated to true, notifying client");
987 ret
= send_evaluation_to_clients(trigger
, evaluation
, &client_list
,
988 state
, object_uid
, object_gid
);
995 int notification_thread_client_subscribe(struct notification_client
*client
,
996 struct lttng_condition
*condition
,
997 struct notification_thread_state
*state
,
998 enum lttng_notification_channel_status
*_status
)
1001 struct notification_client_list
*client_list
= NULL
;
1002 struct lttng_condition_list_element
*condition_list_element
= NULL
;
1003 struct notification_client_list_element
*client_list_element
= NULL
;
1004 enum lttng_notification_channel_status status
=
1005 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
1008 * Ensure that the client has not already subscribed to this condition
1011 cds_list_for_each_entry(condition_list_element
, &client
->condition_list
, node
) {
1012 if (lttng_condition_is_equal(condition_list_element
->condition
,
1014 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED
;
1019 condition_list_element
= zmalloc(sizeof(*condition_list_element
));
1020 if (!condition_list_element
) {
1024 client_list_element
= zmalloc(sizeof(*client_list_element
));
1025 if (!client_list_element
) {
1031 * Add the newly-subscribed condition to the client's subscription list.
1033 CDS_INIT_LIST_HEAD(&condition_list_element
->node
);
1034 condition_list_element
->condition
= condition
;
1035 cds_list_add(&condition_list_element
->node
, &client
->condition_list
);
1037 client_list
= get_client_list_from_condition(state
, condition
);
1040 * No notification-emiting trigger registered with this
1041 * condition. We don't evaluate the condition right away
1042 * since this trigger is not registered yet.
1044 free(client_list_element
);
1049 * The condition to which the client just subscribed is evaluated
1050 * at this point so that conditions that are already TRUE result
1051 * in a notification being sent out.
1053 * The client_list's trigger is used without locking the list itself.
1054 * This is correct since the list doesn't own the trigger and the
1055 * object is immutable.
1057 if (evaluate_condition_for_client(client_list
->trigger
, condition
,
1059 WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
1061 free(client_list_element
);
1066 * Add the client to the list of clients interested in a given trigger
1067 * if a "notification" trigger with a corresponding condition was
1070 client_list_element
->client
= client
;
1071 CDS_INIT_LIST_HEAD(&client_list_element
->node
);
1073 pthread_mutex_lock(&client_list
->lock
);
1074 cds_list_add(&client_list_element
->node
, &client_list
->list
);
1075 pthread_mutex_unlock(&client_list
->lock
);
1081 notification_client_list_put(client_list
);
1085 free(condition_list_element
);
1086 free(client_list_element
);
1091 int notification_thread_client_unsubscribe(
1092 struct notification_client
*client
,
1093 struct lttng_condition
*condition
,
1094 struct notification_thread_state
*state
,
1095 enum lttng_notification_channel_status
*_status
)
1097 struct notification_client_list
*client_list
;
1098 struct lttng_condition_list_element
*condition_list_element
,
1100 struct notification_client_list_element
*client_list_element
,
1102 bool condition_found
= false;
1103 enum lttng_notification_channel_status status
=
1104 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
1106 /* Remove the condition from the client's condition list. */
1107 cds_list_for_each_entry_safe(condition_list_element
, condition_tmp
,
1108 &client
->condition_list
, node
) {
1109 if (!lttng_condition_is_equal(condition_list_element
->condition
,
1114 cds_list_del(&condition_list_element
->node
);
1116 * The caller may be iterating on the client's conditions to
1117 * tear down a client's connection. In this case, the condition
1118 * will be destroyed at the end.
1120 if (condition
!= condition_list_element
->condition
) {
1121 lttng_condition_destroy(
1122 condition_list_element
->condition
);
1124 free(condition_list_element
);
1125 condition_found
= true;
1129 if (!condition_found
) {
1130 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION
;
1135 * Remove the client from the list of clients interested the trigger
1136 * matching the condition.
1138 client_list
= get_client_list_from_condition(state
, condition
);
1143 pthread_mutex_lock(&client_list
->lock
);
1144 cds_list_for_each_entry_safe(client_list_element
, client_tmp
,
1145 &client_list
->list
, node
) {
1146 if (client_list_element
->client
->id
!= client
->id
) {
1149 cds_list_del(&client_list_element
->node
);
1150 free(client_list_element
);
1153 pthread_mutex_unlock(&client_list
->lock
);
1154 notification_client_list_put(client_list
);
1157 lttng_condition_destroy(condition
);
1165 void free_notification_client_rcu(struct rcu_head
*node
)
1167 free(caa_container_of(node
, struct notification_client
, rcu_node
));
1171 void notification_client_destroy(struct notification_client
*client
,
1172 struct notification_thread_state
*state
)
1179 * The client object is not reachable by other threads, no need to lock
1182 if (client
->socket
>= 0) {
1183 (void) lttcomm_close_unix_sock(client
->socket
);
1184 client
->socket
= -1;
1186 client
->communication
.active
= false;
1187 lttng_dynamic_buffer_reset(&client
->communication
.inbound
.buffer
);
1188 lttng_dynamic_buffer_reset(&client
->communication
.outbound
.buffer
);
1189 pthread_mutex_destroy(&client
->lock
);
1190 call_rcu(&client
->rcu_node
, free_notification_client_rcu
);
1194 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1198 struct notification_client
*get_client_from_socket(int socket
,
1199 struct notification_thread_state
*state
)
1201 struct cds_lfht_iter iter
;
1202 struct cds_lfht_node
*node
;
1203 struct notification_client
*client
= NULL
;
1205 cds_lfht_lookup(state
->client_socket_ht
,
1206 hash_client_socket(socket
),
1207 match_client_socket
,
1208 (void *) (unsigned long) socket
,
1210 node
= cds_lfht_iter_get_node(&iter
);
1215 client
= caa_container_of(node
, struct notification_client
,
1216 client_socket_ht_node
);
1222 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1226 struct notification_client
*get_client_from_id(notification_client_id id
,
1227 struct notification_thread_state
*state
)
1229 struct cds_lfht_iter iter
;
1230 struct cds_lfht_node
*node
;
1231 struct notification_client
*client
= NULL
;
1233 cds_lfht_lookup(state
->client_id_ht
,
1238 node
= cds_lfht_iter_get_node(&iter
);
1243 client
= caa_container_of(node
, struct notification_client
,
1250 bool buffer_usage_condition_applies_to_channel(
1251 const struct lttng_condition
*condition
,
1252 const struct channel_info
*channel_info
)
1254 enum lttng_condition_status status
;
1255 enum lttng_domain_type condition_domain
;
1256 const char *condition_session_name
= NULL
;
1257 const char *condition_channel_name
= NULL
;
1259 status
= lttng_condition_buffer_usage_get_domain_type(condition
,
1261 assert(status
== LTTNG_CONDITION_STATUS_OK
);
1262 if (channel_info
->key
.domain
!= condition_domain
) {
1266 status
= lttng_condition_buffer_usage_get_session_name(
1267 condition
, &condition_session_name
);
1268 assert((status
== LTTNG_CONDITION_STATUS_OK
) && condition_session_name
);
1270 status
= lttng_condition_buffer_usage_get_channel_name(
1271 condition
, &condition_channel_name
);
1272 assert((status
== LTTNG_CONDITION_STATUS_OK
) && condition_channel_name
);
1274 if (strcmp(channel_info
->session_info
->name
, condition_session_name
)) {
1277 if (strcmp(channel_info
->name
, condition_channel_name
)) {
1287 bool session_consumed_size_condition_applies_to_channel(
1288 const struct lttng_condition
*condition
,
1289 const struct channel_info
*channel_info
)
1291 enum lttng_condition_status status
;
1292 const char *condition_session_name
= NULL
;
1294 status
= lttng_condition_session_consumed_size_get_session_name(
1295 condition
, &condition_session_name
);
1296 assert((status
== LTTNG_CONDITION_STATUS_OK
) && condition_session_name
);
1298 if (strcmp(channel_info
->session_info
->name
, condition_session_name
)) {
1308 bool trigger_applies_to_channel(const struct lttng_trigger
*trigger
,
1309 const struct channel_info
*channel_info
)
1311 const struct lttng_condition
*condition
;
1312 bool trigger_applies
;
1314 condition
= lttng_trigger_get_const_condition(trigger
);
1319 switch (lttng_condition_get_type(condition
)) {
1320 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
1321 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
1322 trigger_applies
= buffer_usage_condition_applies_to_channel(
1323 condition
, channel_info
);
1325 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
1326 trigger_applies
= session_consumed_size_condition_applies_to_channel(
1327 condition
, channel_info
);
1333 return trigger_applies
;
1339 bool trigger_applies_to_client(struct lttng_trigger
*trigger
,
1340 struct notification_client
*client
)
1342 bool applies
= false;
1343 struct lttng_condition_list_element
*condition_list_element
;
1345 cds_list_for_each_entry(condition_list_element
, &client
->condition_list
,
1347 applies
= lttng_condition_is_equal(
1348 condition_list_element
->condition
,
1349 lttng_trigger_get_condition(trigger
));
1357 /* Must be called with RCU read lock held. */
1359 struct lttng_session_trigger_list
*get_session_trigger_list(
1360 struct notification_thread_state
*state
,
1361 const char *session_name
)
1363 struct lttng_session_trigger_list
*list
= NULL
;
1364 struct cds_lfht_node
*node
;
1365 struct cds_lfht_iter iter
;
1367 cds_lfht_lookup(state
->session_triggers_ht
,
1368 hash_key_str(session_name
, lttng_ht_seed
),
1369 match_session_trigger_list
,
1372 node
= cds_lfht_iter_get_node(&iter
);
1375 * Not an error, the list of triggers applying to that session
1376 * will be initialized when the session is created.
1378 DBG("[notification-thread] No trigger list found for session \"%s\" as it is not yet known to the notification system",
1383 list
= caa_container_of(node
,
1384 struct lttng_session_trigger_list
,
1385 session_triggers_ht_node
);
1391 * Allocate an empty lttng_session_trigger_list for the session named
1394 * No ownership of 'session_name' is assumed by the session trigger list.
1395 * It is the caller's responsability to ensure the session name is alive
1396 * for as long as this list is.
1399 struct lttng_session_trigger_list
*lttng_session_trigger_list_create(
1400 const char *session_name
,
1401 struct cds_lfht
*session_triggers_ht
)
1403 struct lttng_session_trigger_list
*list
;
1405 list
= zmalloc(sizeof(*list
));
1409 list
->session_name
= session_name
;
1410 CDS_INIT_LIST_HEAD(&list
->list
);
1411 cds_lfht_node_init(&list
->session_triggers_ht_node
);
1412 list
->session_triggers_ht
= session_triggers_ht
;
1415 /* Publish the list through the session_triggers_ht. */
1416 cds_lfht_add(session_triggers_ht
,
1417 hash_key_str(session_name
, lttng_ht_seed
),
1418 &list
->session_triggers_ht_node
);
1425 void free_session_trigger_list_rcu(struct rcu_head
*node
)
1427 free(caa_container_of(node
, struct lttng_session_trigger_list
,
1432 void lttng_session_trigger_list_destroy(struct lttng_session_trigger_list
*list
)
1434 struct lttng_trigger_list_element
*trigger_list_element
, *tmp
;
1436 /* Empty the list element by element, and then free the list itself. */
1437 cds_list_for_each_entry_safe(trigger_list_element
, tmp
,
1438 &list
->list
, node
) {
1439 cds_list_del(&trigger_list_element
->node
);
1440 free(trigger_list_element
);
1443 /* Unpublish the list from the session_triggers_ht. */
1444 cds_lfht_del(list
->session_triggers_ht
,
1445 &list
->session_triggers_ht_node
);
1447 call_rcu(&list
->rcu_node
, free_session_trigger_list_rcu
);
1451 int lttng_session_trigger_list_add(struct lttng_session_trigger_list
*list
,
1452 struct lttng_trigger
*trigger
)
1455 struct lttng_trigger_list_element
*new_element
=
1456 zmalloc(sizeof(*new_element
));
1462 CDS_INIT_LIST_HEAD(&new_element
->node
);
1463 new_element
->trigger
= trigger
;
1464 cds_list_add(&new_element
->node
, &list
->list
);
1470 bool trigger_applies_to_session(const struct lttng_trigger
*trigger
,
1471 const char *session_name
)
1473 bool applies
= false;
1474 const struct lttng_condition
*condition
;
1476 condition
= lttng_trigger_get_const_condition(trigger
);
1477 switch (lttng_condition_get_type(condition
)) {
1478 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
1479 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
1481 enum lttng_condition_status condition_status
;
1482 const char *condition_session_name
;
1484 condition_status
= lttng_condition_session_rotation_get_session_name(
1485 condition
, &condition_session_name
);
1486 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
1487 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
1491 assert(condition_session_name
);
1492 applies
= !strcmp(condition_session_name
, session_name
);
1503 * Allocate and initialize an lttng_session_trigger_list which contains
1504 * all triggers that apply to the session named 'session_name'.
1506 * No ownership of 'session_name' is assumed by the session trigger list.
1507 * It is the caller's responsability to ensure the session name is alive
1508 * for as long as this list is.
1511 struct lttng_session_trigger_list
*lttng_session_trigger_list_build(
1512 const struct notification_thread_state
*state
,
1513 const char *session_name
)
1515 int trigger_count
= 0;
1516 struct lttng_session_trigger_list
*session_trigger_list
= NULL
;
1517 struct lttng_trigger_ht_element
*trigger_ht_element
= NULL
;
1518 struct cds_lfht_iter iter
;
1520 session_trigger_list
= lttng_session_trigger_list_create(session_name
,
1521 state
->session_triggers_ht
);
1523 /* Add all triggers applying to the session named 'session_name'. */
1524 cds_lfht_for_each_entry(state
->triggers_ht
, &iter
, trigger_ht_element
,
1528 if (!trigger_applies_to_session(trigger_ht_element
->trigger
,
1533 ret
= lttng_session_trigger_list_add(session_trigger_list
,
1534 trigger_ht_element
->trigger
);
1542 DBG("[notification-thread] Found %i triggers that apply to newly created session",
1544 return session_trigger_list
;
1546 lttng_session_trigger_list_destroy(session_trigger_list
);
1551 struct session_info
*find_or_create_session_info(
1552 struct notification_thread_state
*state
,
1553 const char *name
, uid_t uid
, gid_t gid
)
1555 struct session_info
*session
= NULL
;
1556 struct cds_lfht_node
*node
;
1557 struct cds_lfht_iter iter
;
1558 struct lttng_session_trigger_list
*trigger_list
;
1561 cds_lfht_lookup(state
->sessions_ht
,
1562 hash_key_str(name
, lttng_ht_seed
),
1566 node
= cds_lfht_iter_get_node(&iter
);
1568 DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
1570 session
= caa_container_of(node
, struct session_info
,
1572 assert(session
->uid
== uid
);
1573 assert(session
->gid
== gid
);
1574 session_info_get(session
);
1578 trigger_list
= lttng_session_trigger_list_build(state
, name
);
1579 if (!trigger_list
) {
1583 session
= session_info_create(name
, uid
, gid
, trigger_list
,
1584 state
->sessions_ht
);
1586 ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1588 lttng_session_trigger_list_destroy(trigger_list
);
1591 trigger_list
= NULL
;
1593 cds_lfht_add(state
->sessions_ht
, hash_key_str(name
, lttng_ht_seed
),
1594 &session
->sessions_ht_node
);
1600 session_info_put(session
);
1605 int handle_notification_thread_command_add_channel(
1606 struct notification_thread_state
*state
,
1607 const char *session_name
, uid_t session_uid
, gid_t session_gid
,
1608 const char *channel_name
, enum lttng_domain_type channel_domain
,
1609 uint64_t channel_key_int
, uint64_t channel_capacity
,
1610 enum lttng_error_code
*cmd_result
)
1612 struct cds_list_head trigger_list
;
1613 struct channel_info
*new_channel_info
= NULL
;
1614 struct channel_key channel_key
= {
1615 .key
= channel_key_int
,
1616 .domain
= channel_domain
,
1618 struct lttng_channel_trigger_list
*channel_trigger_list
= NULL
;
1619 struct lttng_trigger_ht_element
*trigger_ht_element
= NULL
;
1620 int trigger_count
= 0;
1621 struct cds_lfht_iter iter
;
1622 struct session_info
*session_info
= NULL
;
1624 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64
" in %s domain",
1625 channel_name
, session_name
, channel_key_int
,
1626 channel_domain
== LTTNG_DOMAIN_KERNEL
? "kernel" : "user space");
1628 CDS_INIT_LIST_HEAD(&trigger_list
);
1630 session_info
= find_or_create_session_info(state
, session_name
,
1631 session_uid
, session_gid
);
1632 if (!session_info
) {
1633 /* Allocation error or an internal error occurred. */
1637 new_channel_info
= channel_info_create(channel_name
, &channel_key
,
1638 channel_capacity
, session_info
);
1639 if (!new_channel_info
) {
1644 /* Build a list of all triggers applying to the new channel. */
1645 cds_lfht_for_each_entry(state
->triggers_ht
, &iter
, trigger_ht_element
,
1647 struct lttng_trigger_list_element
*new_element
;
1649 if (!trigger_applies_to_channel(trigger_ht_element
->trigger
,
1650 new_channel_info
)) {
1654 new_element
= zmalloc(sizeof(*new_element
));
1659 CDS_INIT_LIST_HEAD(&new_element
->node
);
1660 new_element
->trigger
= trigger_ht_element
->trigger
;
1661 cds_list_add(&new_element
->node
, &trigger_list
);
1666 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
1668 channel_trigger_list
= zmalloc(sizeof(*channel_trigger_list
));
1669 if (!channel_trigger_list
) {
1672 channel_trigger_list
->channel_key
= new_channel_info
->key
;
1673 CDS_INIT_LIST_HEAD(&channel_trigger_list
->list
);
1674 cds_lfht_node_init(&channel_trigger_list
->channel_triggers_ht_node
);
1675 cds_list_splice(&trigger_list
, &channel_trigger_list
->list
);
1678 /* Add channel to the channel_ht which owns the channel_infos. */
1679 cds_lfht_add(state
->channels_ht
,
1680 hash_channel_key(&new_channel_info
->key
),
1681 &new_channel_info
->channels_ht_node
);
1683 * Add the list of triggers associated with this channel to the
1684 * channel_triggers_ht.
1686 cds_lfht_add(state
->channel_triggers_ht
,
1687 hash_channel_key(&new_channel_info
->key
),
1688 &channel_trigger_list
->channel_triggers_ht_node
);
1690 session_info_put(session_info
);
1691 *cmd_result
= LTTNG_OK
;
1694 channel_info_destroy(new_channel_info
);
1695 session_info_put(session_info
);
1700 void free_channel_trigger_list_rcu(struct rcu_head
*node
)
1702 free(caa_container_of(node
, struct lttng_channel_trigger_list
,
1707 void free_channel_state_sample_rcu(struct rcu_head
*node
)
1709 free(caa_container_of(node
, struct channel_state_sample
,
1714 int handle_notification_thread_command_remove_channel(
1715 struct notification_thread_state
*state
,
1716 uint64_t channel_key
, enum lttng_domain_type domain
,
1717 enum lttng_error_code
*cmd_result
)
1719 struct cds_lfht_node
*node
;
1720 struct cds_lfht_iter iter
;
1721 struct lttng_channel_trigger_list
*trigger_list
;
1722 struct lttng_trigger_list_element
*trigger_list_element
, *tmp
;
1723 struct channel_key key
= { .key
= channel_key
, .domain
= domain
};
1724 struct channel_info
*channel_info
;
1726 DBG("[notification-thread] Removing channel key = %" PRIu64
" in %s domain",
1727 channel_key
, domain
== LTTNG_DOMAIN_KERNEL
? "kernel" : "user space");
1731 cds_lfht_lookup(state
->channel_triggers_ht
,
1732 hash_channel_key(&key
),
1733 match_channel_trigger_list
,
1736 node
= cds_lfht_iter_get_node(&iter
);
1738 * There is a severe internal error if we are being asked to remove a
1739 * channel that doesn't exist.
1742 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
1746 /* Free the list of triggers associated with this channel. */
1747 trigger_list
= caa_container_of(node
, struct lttng_channel_trigger_list
,
1748 channel_triggers_ht_node
);
1749 cds_list_for_each_entry_safe(trigger_list_element
, tmp
,
1750 &trigger_list
->list
, node
) {
1751 cds_list_del(&trigger_list_element
->node
);
1752 free(trigger_list_element
);
1754 cds_lfht_del(state
->channel_triggers_ht
, node
);
1755 call_rcu(&trigger_list
->rcu_node
, free_channel_trigger_list_rcu
);
1757 /* Free sampled channel state. */
1758 cds_lfht_lookup(state
->channel_state_ht
,
1759 hash_channel_key(&key
),
1760 match_channel_state_sample
,
1763 node
= cds_lfht_iter_get_node(&iter
);
1765 * This is expected to be NULL if the channel is destroyed before we
1766 * received a sample.
1769 struct channel_state_sample
*sample
= caa_container_of(node
,
1770 struct channel_state_sample
,
1771 channel_state_ht_node
);
1773 cds_lfht_del(state
->channel_state_ht
, node
);
1774 call_rcu(&sample
->rcu_node
, free_channel_state_sample_rcu
);
1777 /* Remove the channel from the channels_ht and free it. */
1778 cds_lfht_lookup(state
->channels_ht
,
1779 hash_channel_key(&key
),
1783 node
= cds_lfht_iter_get_node(&iter
);
1785 channel_info
= caa_container_of(node
, struct channel_info
,
1787 cds_lfht_del(state
->channels_ht
, node
);
1788 channel_info_destroy(channel_info
);
1791 *cmd_result
= LTTNG_OK
;
1796 int handle_notification_thread_command_session_rotation(
1797 struct notification_thread_state
*state
,
1798 enum notification_thread_command_type cmd_type
,
1799 const char *session_name
, uid_t session_uid
, gid_t session_gid
,
1800 uint64_t trace_archive_chunk_id
,
1801 struct lttng_trace_archive_location
*location
,
1802 enum lttng_error_code
*_cmd_result
)
1805 enum lttng_error_code cmd_result
= LTTNG_OK
;
1806 struct lttng_session_trigger_list
*trigger_list
;
1807 struct lttng_trigger_list_element
*trigger_list_element
;
1808 struct session_info
*session_info
;
1809 const struct lttng_credentials session_creds
= {
1816 session_info
= find_or_create_session_info(state
, session_name
,
1817 session_uid
, session_gid
);
1818 if (!session_info
) {
1819 /* Allocation error or an internal error occurred. */
1821 cmd_result
= LTTNG_ERR_NOMEM
;
1825 session_info
->rotation
.ongoing
=
1826 cmd_type
== NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
;
1827 session_info
->rotation
.id
= trace_archive_chunk_id
;
1828 trigger_list
= get_session_trigger_list(state
, session_name
);
1829 if (!trigger_list
) {
1830 DBG("[notification-thread] No triggers applying to session \"%s\" found",
1835 cds_list_for_each_entry(trigger_list_element
, &trigger_list
->list
,
1837 const struct lttng_condition
*condition
;
1838 struct lttng_trigger
*trigger
;
1839 struct notification_client_list
*client_list
;
1840 struct lttng_evaluation
*evaluation
= NULL
;
1841 enum lttng_condition_type condition_type
;
1842 enum action_executor_status executor_status
;
1844 trigger
= trigger_list_element
->trigger
;
1845 condition
= lttng_trigger_get_const_condition(trigger
);
1847 condition_type
= lttng_condition_get_type(condition
);
1849 if (condition_type
== LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
&&
1850 cmd_type
!= NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
) {
1852 } else if (condition_type
== LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
&&
1853 cmd_type
!= NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED
) {
1857 client_list
= get_client_list_from_condition(state
, condition
);
1858 if (cmd_type
== NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
) {
1859 evaluation
= lttng_evaluation_session_rotation_ongoing_create(
1860 trace_archive_chunk_id
);
1862 evaluation
= lttng_evaluation_session_rotation_completed_create(
1863 trace_archive_chunk_id
, location
);
1867 /* Internal error */
1869 cmd_result
= LTTNG_ERR_UNK
;
1874 * Ownership of `evaluation` transferred to the action executor
1875 * no matter the result.
1877 executor_status
= action_executor_enqueue(state
->executor
,
1878 trigger
, evaluation
, &session_creds
,
1881 switch (executor_status
) {
1882 case ACTION_EXECUTOR_STATUS_OK
:
1884 case ACTION_EXECUTOR_STATUS_ERROR
:
1885 case ACTION_EXECUTOR_STATUS_INVALID
:
1887 * TODO Add trigger identification (name/id) when
1888 * it is added to the API.
1890 ERR("Fatal error occurred while enqueuing action associated with session rotation trigger");
1893 case ACTION_EXECUTOR_STATUS_OVERFLOW
:
1895 * TODO Add trigger identification (name/id) when
1896 * it is added to the API.
1898 * Not a fatal error.
1900 WARN("No space left when enqueuing action associated with session rotation trigger");
1908 notification_client_list_put(client_list
);
1909 if (caa_unlikely(ret
)) {
1914 session_info_put(session_info
);
1915 *_cmd_result
= cmd_result
;
1921 int condition_is_supported(struct lttng_condition
*condition
)
1925 switch (lttng_condition_get_type(condition
)) {
1926 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
1927 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
1929 enum lttng_domain_type domain
;
1931 ret
= lttng_condition_buffer_usage_get_domain_type(condition
,
1938 if (domain
!= LTTNG_DOMAIN_KERNEL
) {
1944 * Older kernel tracers don't expose the API to monitor their
1945 * buffers. Therefore, we reject triggers that require that
1946 * mechanism to be available to be evaluated.
1948 ret
= kernel_supports_ring_buffer_snapshot_sample_positions();
1958 /* Must be called with RCU read lock held. */
1960 int bind_trigger_to_matching_session(struct lttng_trigger
*trigger
,
1961 struct notification_thread_state
*state
)
1964 const struct lttng_condition
*condition
;
1965 const char *session_name
;
1966 struct lttng_session_trigger_list
*trigger_list
;
1968 condition
= lttng_trigger_get_const_condition(trigger
);
1969 switch (lttng_condition_get_type(condition
)) {
1970 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING
:
1971 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED
:
1973 enum lttng_condition_status status
;
1975 status
= lttng_condition_session_rotation_get_session_name(
1976 condition
, &session_name
);
1977 if (status
!= LTTNG_CONDITION_STATUS_OK
) {
1978 ERR("[notification-thread] Failed to bind trigger to session: unable to get 'session_rotation' condition's session name");
1989 trigger_list
= get_session_trigger_list(state
, session_name
);
1990 if (!trigger_list
) {
1991 DBG("[notification-thread] Unable to bind trigger applying to session \"%s\" as it is not yet known to the notification system",
1997 DBG("[notification-thread] Newly registered trigger bound to session \"%s\"",
1999 ret
= lttng_session_trigger_list_add(trigger_list
, trigger
);
2004 /* Must be called with RCU read lock held. */
2006 int bind_trigger_to_matching_channels(struct lttng_trigger
*trigger
,
2007 struct notification_thread_state
*state
)
2010 struct cds_lfht_node
*node
;
2011 struct cds_lfht_iter iter
;
2012 struct channel_info
*channel
;
2014 cds_lfht_for_each_entry(state
->channels_ht
, &iter
, channel
,
2016 struct lttng_trigger_list_element
*trigger_list_element
;
2017 struct lttng_channel_trigger_list
*trigger_list
;
2018 struct cds_lfht_iter lookup_iter
;
2020 if (!trigger_applies_to_channel(trigger
, channel
)) {
2024 cds_lfht_lookup(state
->channel_triggers_ht
,
2025 hash_channel_key(&channel
->key
),
2026 match_channel_trigger_list
,
2029 node
= cds_lfht_iter_get_node(&lookup_iter
);
2031 trigger_list
= caa_container_of(node
,
2032 struct lttng_channel_trigger_list
,
2033 channel_triggers_ht_node
);
2035 trigger_list_element
= zmalloc(sizeof(*trigger_list_element
));
2036 if (!trigger_list_element
) {
2040 CDS_INIT_LIST_HEAD(&trigger_list_element
->node
);
2041 trigger_list_element
->trigger
= trigger
;
2042 cds_list_add(&trigger_list_element
->node
, &trigger_list
->list
);
2043 DBG("[notification-thread] Newly registered trigger bound to channel \"%s\"",
2051 bool is_trigger_action_notify(const struct lttng_trigger
*trigger
)
2053 bool is_notify
= false;
2054 unsigned int i
, count
;
2055 enum lttng_action_status action_status
;
2056 const struct lttng_action
*action
=
2057 lttng_trigger_get_const_action(trigger
);
2058 enum lttng_action_type action_type
;
2061 action_type
= lttng_action_get_type_const(action
);
2062 if (action_type
== LTTNG_ACTION_TYPE_NOTIFY
) {
2065 } else if (action_type
!= LTTNG_ACTION_TYPE_GROUP
) {
2069 action_status
= lttng_action_group_get_count(action
, &count
);
2070 assert(action_status
== LTTNG_ACTION_STATUS_OK
);
2072 for (i
= 0; i
< count
; i
++) {
2073 const struct lttng_action
*inner_action
=
2074 lttng_action_group_get_at_index(
2077 action_type
= lttng_action_get_type_const(inner_action
);
2078 if (action_type
== LTTNG_ACTION_TYPE_NOTIFY
) {
2089 * FIXME A client's credentials are not checked when registering a trigger.
2091 * The effects of this are benign since:
2092 * - The client will succeed in registering the trigger, as it is valid,
2093 * - The trigger will, internally, be bound to the channel/session,
2094 * - The notifications will not be sent since the client's credentials
2095 * are checked against the channel at that moment.
2097 * If this function returns a non-zero value, it means something is
2098 * fundamentally broken and the whole subsystem/thread will be torn down.
2100 * If a non-fatal error occurs, just set the cmd_result to the appropriate
2104 int handle_notification_thread_command_register_trigger(
2105 struct notification_thread_state
*state
,
2106 struct lttng_trigger
*trigger
,
2107 enum lttng_error_code
*cmd_result
)
2110 struct lttng_condition
*condition
;
2111 struct notification_client
*client
;
2112 struct notification_client_list
*client_list
= NULL
;
2113 struct lttng_trigger_ht_element
*trigger_ht_element
= NULL
;
2114 struct notification_client_list_element
*client_list_element
;
2115 struct cds_lfht_node
*node
;
2116 struct cds_lfht_iter iter
;
2117 bool free_trigger
= true;
2118 struct lttng_evaluation
*evaluation
= NULL
;
2119 struct lttng_credentials object_creds
;
2120 enum action_executor_status executor_status
;
2124 condition
= lttng_trigger_get_condition(trigger
);
2127 ret
= condition_is_supported(condition
);
2130 } else if (ret
== 0) {
2131 *cmd_result
= LTTNG_ERR_NOT_SUPPORTED
;
2134 /* Feature is supported, continue. */
2138 trigger_ht_element
= zmalloc(sizeof(*trigger_ht_element
));
2139 if (!trigger_ht_element
) {
2144 /* Add trigger to the trigger_ht. */
2145 cds_lfht_node_init(&trigger_ht_element
->node
);
2146 trigger_ht_element
->trigger
= trigger
;
2148 node
= cds_lfht_add_unique(state
->triggers_ht
,
2149 lttng_condition_hash(condition
),
2152 &trigger_ht_element
->node
);
2153 if (node
!= &trigger_ht_element
->node
) {
2154 /* Not a fatal error, simply report it to the client. */
2155 *cmd_result
= LTTNG_ERR_TRIGGER_EXISTS
;
2156 goto error_free_ht_element
;
2160 * Ownership of the trigger and of its wrapper was transfered to
2163 trigger_ht_element
= NULL
;
2164 free_trigger
= false;
2167 * The rest only applies to triggers that have a "notify" action.
2168 * It is not skipped as this is the only action type currently
2171 if (is_trigger_action_notify(trigger
)) {
2172 client_list
= notification_client_list_create(trigger
);
2175 goto error_free_ht_element
;
2178 /* Build a list of clients to which this new trigger applies. */
2179 cds_lfht_for_each_entry (state
->client_socket_ht
, &iter
, client
,
2180 client_socket_ht_node
) {
2181 if (!trigger_applies_to_client(trigger
, client
)) {
2185 client_list_element
=
2186 zmalloc(sizeof(*client_list_element
));
2187 if (!client_list_element
) {
2189 goto error_put_client_list
;
2192 CDS_INIT_LIST_HEAD(&client_list_element
->node
);
2193 client_list_element
->client
= client
;
2194 cds_list_add(&client_list_element
->node
,
2195 &client_list
->list
);
2199 * Client list ownership transferred to the
2200 * notification_trigger_clients_ht.
2202 publish_notification_client_list(state
, client_list
);
2205 switch (get_condition_binding_object(condition
)) {
2206 case LTTNG_OBJECT_TYPE_SESSION
:
2207 /* Add the trigger to the list if it matches a known session. */
2208 ret
= bind_trigger_to_matching_session(trigger
, state
);
2210 goto error_put_client_list
;
2213 case LTTNG_OBJECT_TYPE_CHANNEL
:
2215 * Add the trigger to list of triggers bound to the channels
2218 ret
= bind_trigger_to_matching_channels(trigger
, state
);
2220 goto error_put_client_list
;
2223 case LTTNG_OBJECT_TYPE_NONE
:
2226 ERR("Unknown object type on which to bind a newly registered trigger was encountered");
2228 goto error_put_client_list
;
2232 * The new trigger's condition must be evaluated against the current
2235 * In the case of `notify` action, nothing preventing clients from
2236 * subscribing to a condition before the corresponding trigger is
2237 * registered, we have to evaluate this new condition right away.
2239 * At some point, we were waiting for the next "evaluation" (e.g. on
2240 * reception of a channel sample) to evaluate this new condition, but
2243 * The reason it was broken is that waiting for the next sample
2244 * does not allow us to properly handle transitions for edge-triggered
2247 * Consider this example: when we handle a new channel sample, we
2248 * evaluate each conditions twice: once with the previous state, and
2249 * again with the newest state. We then use those two results to
2250 * determine whether a state change happened: a condition was false and
2251 * became true. If a state change happened, we have to notify clients.
2253 * Now, if a client subscribes to a given notification and registers
2254 * a trigger *after* that subscription, we have to make sure the
2255 * condition is evaluated at this point while considering only the
2256 * current state. Otherwise, the next evaluation cycle may only see
2257 * that the evaluations remain the same (true for samples n-1 and n) and
2258 * the client will never know that the condition has been met.
2260 switch (get_condition_binding_object(condition
)) {
2261 case LTTNG_OBJECT_TYPE_SESSION
:
2262 ret
= evaluate_session_condition_for_client(condition
, state
,
2263 &evaluation
, &object_creds
.uid
,
2266 case LTTNG_OBJECT_TYPE_CHANNEL
:
2267 ret
= evaluate_channel_condition_for_client(condition
, state
,
2268 &evaluation
, &object_creds
.uid
,
2271 case LTTNG_OBJECT_TYPE_NONE
:
2273 goto error_put_client_list
;
2274 case LTTNG_OBJECT_TYPE_UNKNOWN
:
2277 goto error_put_client_list
;
2282 goto error_put_client_list
;
2285 DBG("Newly registered trigger's condition evaluated to %s",
2286 evaluation
? "true" : "false");
2288 /* Evaluation yielded nothing. Normal exit. */
2290 goto error_put_client_list
;
2294 * Ownership of `evaluation` transferred to the action executor
2295 * no matter the result.
2297 executor_status
= action_executor_enqueue(state
->executor
, trigger
,
2298 evaluation
, &object_creds
, client_list
);
2300 switch (executor_status
) {
2301 case ACTION_EXECUTOR_STATUS_OK
:
2303 case ACTION_EXECUTOR_STATUS_ERROR
:
2304 case ACTION_EXECUTOR_STATUS_INVALID
:
2306 * TODO Add trigger identification (name/id) when
2307 * it is added to the API.
2309 ERR("Fatal error occurred while enqueuing action associated to newly registered trigger");
2311 goto error_put_client_list
;
2312 case ACTION_EXECUTOR_STATUS_OVERFLOW
:
2314 * TODO Add trigger identification (name/id) when
2315 * it is added to the API.
2317 * Not a fatal error.
2319 WARN("No space left when enqueuing action associated to newly registered trigger");
2321 goto error_put_client_list
;
2326 *cmd_result
= LTTNG_OK
;
2328 error_put_client_list
:
2329 notification_client_list_put(client_list
);
2331 error_free_ht_element
:
2332 free(trigger_ht_element
);
2335 lttng_trigger_destroy(trigger
);
2342 void free_lttng_trigger_ht_element_rcu(struct rcu_head
*node
)
2344 free(caa_container_of(node
, struct lttng_trigger_ht_element
,
2349 int handle_notification_thread_command_unregister_trigger(
2350 struct notification_thread_state
*state
,
2351 struct lttng_trigger
*trigger
,
2352 enum lttng_error_code
*_cmd_reply
)
2354 struct cds_lfht_iter iter
;
2355 struct cds_lfht_node
*triggers_ht_node
;
2356 struct lttng_channel_trigger_list
*trigger_list
;
2357 struct notification_client_list
*client_list
;
2358 struct lttng_trigger_ht_element
*trigger_ht_element
= NULL
;
2359 struct lttng_condition
*condition
= lttng_trigger_get_condition(
2361 enum lttng_error_code cmd_reply
;
2365 cds_lfht_lookup(state
->triggers_ht
,
2366 lttng_condition_hash(condition
),
2370 triggers_ht_node
= cds_lfht_iter_get_node(&iter
);
2371 if (!triggers_ht_node
) {
2372 cmd_reply
= LTTNG_ERR_TRIGGER_NOT_FOUND
;
2375 cmd_reply
= LTTNG_OK
;
2378 /* Remove trigger from channel_triggers_ht. */
2379 cds_lfht_for_each_entry(state
->channel_triggers_ht
, &iter
, trigger_list
,
2380 channel_triggers_ht_node
) {
2381 struct lttng_trigger_list_element
*trigger_element
, *tmp
;
2383 cds_list_for_each_entry_safe(trigger_element
, tmp
,
2384 &trigger_list
->list
, node
) {
2385 const struct lttng_condition
*current_condition
=
2386 lttng_trigger_get_const_condition(
2387 trigger_element
->trigger
);
2389 assert(current_condition
);
2390 if (!lttng_condition_is_equal(condition
,
2391 current_condition
)) {
2395 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
2396 cds_list_del(&trigger_element
->node
);
2397 /* A trigger can only appear once per channel */
2403 * Remove and release the client list from
2404 * notification_trigger_clients_ht.
2406 client_list
= get_client_list_from_condition(state
, condition
);
2407 assert(client_list
);
2409 /* Put new reference and the hashtable's reference. */
2410 notification_client_list_put(client_list
);
2411 notification_client_list_put(client_list
);
2414 /* Remove trigger from triggers_ht. */
2415 trigger_ht_element
= caa_container_of(triggers_ht_node
,
2416 struct lttng_trigger_ht_element
, node
);
2417 cds_lfht_del(state
->triggers_ht
, triggers_ht_node
);
2419 /* Release the ownership of the trigger. */
2420 lttng_trigger_destroy(trigger_ht_element
->trigger
);
2421 call_rcu(&trigger_ht_element
->rcu_node
, free_lttng_trigger_ht_element_rcu
);
2425 *_cmd_reply
= cmd_reply
;
2430 /* Returns 0 on success, 1 on exit requested, negative value on error. */
2431 int handle_notification_thread_command(
2432 struct notification_thread_handle
*handle
,
2433 struct notification_thread_state
*state
)
2437 struct notification_thread_command
*cmd
;
2439 /* Read the event pipe to put it back into a quiescent state. */
2440 ret
= lttng_read(lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
), &counter
,
2442 if (ret
!= sizeof(counter
)) {
2446 pthread_mutex_lock(&handle
->cmd_queue
.lock
);
2447 cmd
= cds_list_first_entry(&handle
->cmd_queue
.list
,
2448 struct notification_thread_command
, cmd_list_node
);
2449 cds_list_del(&cmd
->cmd_list_node
);
2450 pthread_mutex_unlock(&handle
->cmd_queue
.lock
);
2451 switch (cmd
->type
) {
2452 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER
:
2453 DBG("[notification-thread] Received register trigger command");
2454 ret
= handle_notification_thread_command_register_trigger(
2455 state
, cmd
->parameters
.trigger
,
2458 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER
:
2459 DBG("[notification-thread] Received unregister trigger command");
2460 ret
= handle_notification_thread_command_unregister_trigger(
2461 state
, cmd
->parameters
.trigger
,
2464 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL
:
2465 DBG("[notification-thread] Received add channel command");
2466 ret
= handle_notification_thread_command_add_channel(
2468 cmd
->parameters
.add_channel
.session
.name
,
2469 cmd
->parameters
.add_channel
.session
.uid
,
2470 cmd
->parameters
.add_channel
.session
.gid
,
2471 cmd
->parameters
.add_channel
.channel
.name
,
2472 cmd
->parameters
.add_channel
.channel
.domain
,
2473 cmd
->parameters
.add_channel
.channel
.key
,
2474 cmd
->parameters
.add_channel
.channel
.capacity
,
2477 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL
:
2478 DBG("[notification-thread] Received remove channel command");
2479 ret
= handle_notification_thread_command_remove_channel(
2480 state
, cmd
->parameters
.remove_channel
.key
,
2481 cmd
->parameters
.remove_channel
.domain
,
2484 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
:
2485 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED
:
2486 DBG("[notification-thread] Received session rotation %s command",
2487 cmd
->type
== NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING
?
2488 "ongoing" : "completed");
2489 ret
= handle_notification_thread_command_session_rotation(
2492 cmd
->parameters
.session_rotation
.session_name
,
2493 cmd
->parameters
.session_rotation
.uid
,
2494 cmd
->parameters
.session_rotation
.gid
,
2495 cmd
->parameters
.session_rotation
.trace_archive_chunk_id
,
2496 cmd
->parameters
.session_rotation
.location
,
2499 case NOTIFICATION_COMMAND_TYPE_QUIT
:
2500 DBG("[notification-thread] Received quit command");
2501 cmd
->reply_code
= LTTNG_OK
;
2504 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE
:
2506 const enum client_transmission_status client_status
=
2507 cmd
->parameters
.client_communication_update
2509 const notification_client_id client_id
=
2510 cmd
->parameters
.client_communication_update
.id
;
2511 struct notification_client
*client
;
2514 client
= get_client_from_id(client_id
, state
);
2518 * Client error was probably already picked-up by the
2519 * notification thread or it has disconnected
2520 * gracefully while this command was queued.
2522 DBG("Failed to find notification client to update communication status, client id = %" PRIu64
,
2526 pthread_mutex_lock(&client
->lock
);
2527 ret
= client_handle_transmission_status(
2528 client
, client_status
, state
);
2529 pthread_mutex_unlock(&client
->lock
);
2535 ERR("[notification-thread] Unknown internal command received");
2543 if (cmd
->is_async
) {
2547 lttng_waiter_wake_up(&cmd
->reply_waiter
);
2551 /* Wake-up and return a fatal error to the calling thread. */
2552 lttng_waiter_wake_up(&cmd
->reply_waiter
);
2553 cmd
->reply_code
= LTTNG_ERR_FATAL
;
2555 /* Indicate a fatal error to the caller. */
2560 int socket_set_non_blocking(int socket
)
2564 /* Set the pipe as non-blocking. */
2565 ret
= fcntl(socket
, F_GETFL
, 0);
2567 PERROR("fcntl get socket flags");
2572 ret
= fcntl(socket
, F_SETFL
, flags
| O_NONBLOCK
);
2574 PERROR("fcntl set O_NONBLOCK socket flag");
2577 DBG("Client socket (fd = %i) set as non-blocking", socket
);
2582 /* Client lock must be acquired by caller. */
2584 int client_reset_inbound_state(struct notification_client
*client
)
2588 ASSERT_LOCKED(client
->lock
);
2590 ret
= lttng_dynamic_buffer_set_size(
2591 &client
->communication
.inbound
.buffer
, 0);
2594 client
->communication
.inbound
.bytes_to_receive
=
2595 sizeof(struct lttng_notification_channel_message
);
2596 client
->communication
.inbound
.msg_type
=
2597 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN
;
2598 LTTNG_SOCK_SET_UID_CRED(&client
->communication
.inbound
.creds
, -1);
2599 LTTNG_SOCK_SET_GID_CRED(&client
->communication
.inbound
.creds
, -1);
2600 ret
= lttng_dynamic_buffer_set_size(
2601 &client
->communication
.inbound
.buffer
,
2602 client
->communication
.inbound
.bytes_to_receive
);
2606 int handle_notification_thread_client_connect(
2607 struct notification_thread_state
*state
)
2610 struct notification_client
*client
;
2612 DBG("[notification-thread] Handling new notification channel client connection");
2614 client
= zmalloc(sizeof(*client
));
2620 pthread_mutex_init(&client
->lock
, NULL
);
2621 client
->id
= state
->next_notification_client_id
++;
2622 CDS_INIT_LIST_HEAD(&client
->condition_list
);
2623 lttng_dynamic_buffer_init(&client
->communication
.inbound
.buffer
);
2624 lttng_dynamic_buffer_init(&client
->communication
.outbound
.buffer
);
2625 client
->communication
.inbound
.expect_creds
= true;
2627 pthread_mutex_lock(&client
->lock
);
2628 ret
= client_reset_inbound_state(client
);
2629 pthread_mutex_unlock(&client
->lock
);
2631 ERR("[notification-thread] Failed to reset client communication's inbound state");
2636 ret
= lttcomm_accept_unix_sock(state
->notification_channel_socket
);
2638 ERR("[notification-thread] Failed to accept new notification channel client connection");
2643 client
->socket
= ret
;
2645 ret
= socket_set_non_blocking(client
->socket
);
2647 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
2651 ret
= lttcomm_setsockopt_creds_unix_sock(client
->socket
);
2653 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
2658 ret
= lttng_poll_add(&state
->events
, client
->socket
,
2659 LPOLLIN
| LPOLLERR
|
2660 LPOLLHUP
| LPOLLRDHUP
);
2662 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
2666 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
2670 cds_lfht_add(state
->client_socket_ht
,
2671 hash_client_socket(client
->socket
),
2672 &client
->client_socket_ht_node
);
2673 cds_lfht_add(state
->client_id_ht
,
2674 hash_client_id(client
->id
),
2675 &client
->client_id_ht_node
);
2680 notification_client_destroy(client
, state
);
2684 /* RCU read-lock must be held by the caller. */
2685 /* Client lock must be held by the caller */
2687 int notification_thread_client_disconnect(
2688 struct notification_client
*client
,
2689 struct notification_thread_state
*state
)
2692 struct lttng_condition_list_element
*condition_list_element
, *tmp
;
2694 /* Acquire the client lock to disable its communication atomically. */
2695 client
->communication
.active
= false;
2696 ret
= lttng_poll_del(&state
->events
, client
->socket
);
2698 ERR("[notification-thread] Failed to remove client socket %d from poll set",
2702 cds_lfht_del(state
->client_socket_ht
, &client
->client_socket_ht_node
);
2703 cds_lfht_del(state
->client_id_ht
, &client
->client_id_ht_node
);
2705 /* Release all conditions to which the client was subscribed. */
2706 cds_list_for_each_entry_safe(condition_list_element
, tmp
,
2707 &client
->condition_list
, node
) {
2708 (void) notification_thread_client_unsubscribe(client
,
2709 condition_list_element
->condition
, state
, NULL
);
2713 * Client no longer accessible to other threads (through the
2716 notification_client_destroy(client
, state
);
2720 int handle_notification_thread_client_disconnect(
2721 int client_socket
, struct notification_thread_state
*state
)
2724 struct notification_client
*client
;
2727 DBG("[notification-thread] Closing client connection (socket fd = %i)",
2729 client
= get_client_from_socket(client_socket
, state
);
2731 /* Internal state corruption, fatal error. */
2732 ERR("[notification-thread] Unable to find client (socket fd = %i)",
2738 pthread_mutex_lock(&client
->lock
);
2739 ret
= notification_thread_client_disconnect(client
, state
);
2740 pthread_mutex_unlock(&client
->lock
);
2746 int handle_notification_thread_client_disconnect_all(
2747 struct notification_thread_state
*state
)
2749 struct cds_lfht_iter iter
;
2750 struct notification_client
*client
;
2751 bool error_encoutered
= false;
2754 DBG("[notification-thread] Closing all client connections");
2755 cds_lfht_for_each_entry(state
->client_socket_ht
, &iter
, client
,
2756 client_socket_ht_node
) {
2759 pthread_mutex_lock(&client
->lock
);
2760 ret
= notification_thread_client_disconnect(
2762 pthread_mutex_unlock(&client
->lock
);
2764 error_encoutered
= true;
2768 return error_encoutered
? 1 : 0;
2771 int handle_notification_thread_trigger_unregister_all(
2772 struct notification_thread_state
*state
)
2774 bool error_occurred
= false;
2775 struct cds_lfht_iter iter
;
2776 struct lttng_trigger_ht_element
*trigger_ht_element
;
2779 cds_lfht_for_each_entry(state
->triggers_ht
, &iter
, trigger_ht_element
,
2781 int ret
= handle_notification_thread_command_unregister_trigger(
2782 state
, trigger_ht_element
->trigger
, NULL
);
2784 error_occurred
= true;
2788 return error_occurred
? -1 : 0;
2792 int client_handle_transmission_status(
2793 struct notification_client
*client
,
2794 enum client_transmission_status transmission_status
,
2795 struct notification_thread_state
*state
)
2799 switch (transmission_status
) {
2800 case CLIENT_TRANSMISSION_STATUS_COMPLETE
:
2801 ret
= lttng_poll_mod(&state
->events
, client
->socket
,
2802 CLIENT_POLL_MASK_IN
);
2807 client
->communication
.outbound
.queued_command_reply
= false;
2808 client
->communication
.outbound
.dropped_notification
= false;
2810 case CLIENT_TRANSMISSION_STATUS_QUEUED
:
2812 * We want to be notified whenever there is buffer space
2813 * available to send the rest of the payload.
2815 ret
= lttng_poll_mod(&state
->events
, client
->socket
,
2816 CLIENT_POLL_MASK_IN_OUT
);
2821 case CLIENT_TRANSMISSION_STATUS_FAIL
:
2822 ret
= notification_thread_client_disconnect(client
, state
);
2827 case CLIENT_TRANSMISSION_STATUS_ERROR
:
2837 /* Client lock must be acquired by caller. */
2839 enum client_transmission_status
client_flush_outgoing_queue(
2840 struct notification_client
*client
)
2843 size_t to_send_count
;
2844 enum client_transmission_status status
;
2846 ASSERT_LOCKED(client
->lock
);
2848 if (!client
->communication
.active
) {
2849 status
= CLIENT_TRANSMISSION_STATUS_FAIL
;
2853 assert(client
->communication
.outbound
.buffer
.size
!= 0);
2854 to_send_count
= client
->communication
.outbound
.buffer
.size
;
2855 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
2858 ret
= lttcomm_send_unix_sock_non_block(client
->socket
,
2859 client
->communication
.outbound
.buffer
.data
,
2861 if ((ret
>= 0 && ret
< to_send_count
)) {
2862 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
2864 to_send_count
-= max(ret
, 0);
2866 memcpy(client
->communication
.outbound
.buffer
.data
,
2867 client
->communication
.outbound
.buffer
.data
+
2868 client
->communication
.outbound
.buffer
.size
- to_send_count
,
2870 ret
= lttng_dynamic_buffer_set_size(
2871 &client
->communication
.outbound
.buffer
,
2876 status
= CLIENT_TRANSMISSION_STATUS_QUEUED
;
2877 } else if (ret
< 0) {
2878 /* Generic error, disconnect the client. */
2879 ERR("[notification-thread] Failed to flush outgoing queue, disconnecting client (socket fd = %i)",
2881 status
= CLIENT_TRANSMISSION_STATUS_FAIL
;
2883 /* No error and flushed the queue completely. */
2884 ret
= lttng_dynamic_buffer_set_size(
2885 &client
->communication
.outbound
.buffer
, 0);
2889 status
= CLIENT_TRANSMISSION_STATUS_COMPLETE
;
2894 return CLIENT_TRANSMISSION_STATUS_ERROR
;
2897 /* Client lock must be acquired by caller. */
2899 int client_send_command_reply(struct notification_client
*client
,
2900 struct notification_thread_state
*state
,
2901 enum lttng_notification_channel_status status
)
2904 struct lttng_notification_channel_command_reply reply
= {
2905 .status
= (int8_t) status
,
2907 struct lttng_notification_channel_message msg
= {
2908 .type
= (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY
,
2909 .size
= sizeof(reply
),
2911 char buffer
[sizeof(msg
) + sizeof(reply
)];
2912 enum client_transmission_status transmission_status
;
2914 ASSERT_LOCKED(client
->lock
);
2916 if (client
->communication
.outbound
.queued_command_reply
) {
2917 /* Protocol error. */
2921 memcpy(buffer
, &msg
, sizeof(msg
));
2922 memcpy(buffer
+ sizeof(msg
), &reply
, sizeof(reply
));
2923 DBG("[notification-thread] Send command reply (%i)", (int) status
);
2925 /* Enqueue buffer to outgoing queue and flush it. */
2926 ret
= lttng_dynamic_buffer_append(
2927 &client
->communication
.outbound
.buffer
,
2928 buffer
, sizeof(buffer
));
2933 transmission_status
= client_flush_outgoing_queue(client
);
2934 ret
= client_handle_transmission_status(
2935 client
, transmission_status
, state
);
2940 if (client
->communication
.outbound
.buffer
.size
!= 0) {
2941 /* Queue could not be emptied. */
2942 client
->communication
.outbound
.queued_command_reply
= true;
2951 int client_handle_message_unknown(struct notification_client
*client
,
2952 struct notification_thread_state
*state
)
2956 pthread_mutex_lock(&client
->lock
);
2959 * Receiving message header. The function will be called again
2960 * once the rest of the message as been received and can be
2963 const struct lttng_notification_channel_message
*msg
;
2965 assert(sizeof(*msg
) == client
->communication
.inbound
.buffer
.size
);
2966 msg
= (const struct lttng_notification_channel_message
*)
2967 client
->communication
.inbound
.buffer
.data
;
2969 if (msg
->size
== 0 ||
2970 msg
->size
> DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE
) {
2971 ERR("[notification-thread] Invalid notification channel message: length = %u",
2977 switch (msg
->type
) {
2978 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE
:
2979 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE
:
2980 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
:
2984 ERR("[notification-thread] Invalid notification channel message: unexpected message type");
2988 client
->communication
.inbound
.bytes_to_receive
= msg
->size
;
2989 client
->communication
.inbound
.msg_type
=
2990 (enum lttng_notification_channel_message_type
) msg
->type
;
2991 ret
= lttng_dynamic_buffer_set_size(
2992 &client
->communication
.inbound
.buffer
, msg
->size
);
2994 pthread_mutex_unlock(&client
->lock
);
2999 int client_handle_message_handshake(struct notification_client
*client
,
3000 struct notification_thread_state
*state
)
3003 struct lttng_notification_channel_command_handshake
*handshake_client
;
3004 const struct lttng_notification_channel_command_handshake handshake_reply
= {
3005 .major
= LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR
,
3006 .minor
= LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR
,
3008 const struct lttng_notification_channel_message msg_header
= {
3009 .type
= LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
,
3010 .size
= sizeof(handshake_reply
),
3012 enum lttng_notification_channel_status status
=
3013 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
3014 char send_buffer
[sizeof(msg_header
) + sizeof(handshake_reply
)];
3015 enum client_transmission_status transmission_status
;
3017 pthread_mutex_lock(&client
->lock
);
3019 memcpy(send_buffer
, &msg_header
, sizeof(msg_header
));
3020 memcpy(send_buffer
+ sizeof(msg_header
), &handshake_reply
,
3021 sizeof(handshake_reply
));
3024 (struct lttng_notification_channel_command_handshake
*)
3025 client
->communication
.inbound
.buffer
3027 client
->major
= handshake_client
->major
;
3028 client
->minor
= handshake_client
->minor
;
3029 if (!client
->communication
.inbound
.creds_received
) {
3030 ERR("[notification-thread] No credentials received from client");
3035 client
->uid
= LTTNG_SOCK_GET_UID_CRED(
3036 &client
->communication
.inbound
.creds
);
3037 client
->gid
= LTTNG_SOCK_GET_GID_CRED(
3038 &client
->communication
.inbound
.creds
);
3039 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
3040 client
->uid
, client
->gid
, (int) client
->major
,
3041 (int) client
->minor
);
3043 if (handshake_client
->major
!=
3044 LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR
) {
3045 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION
;
3048 ret
= lttng_dynamic_buffer_append(
3049 &client
->communication
.outbound
.buffer
, send_buffer
,
3050 sizeof(send_buffer
));
3052 ERR("[notification-thread] Failed to send protocol version to notification channel client");
3056 client
->validated
= true;
3057 client
->communication
.active
= true;
3059 transmission_status
= client_flush_outgoing_queue(client
);
3060 ret
= client_handle_transmission_status(
3061 client
, transmission_status
, state
);
3066 ret
= client_send_command_reply(client
, state
, status
);
3068 ERR("[notification-thread] Failed to send reply to notification channel client");
3072 /* Set reception state to receive the next message header. */
3073 ret
= client_reset_inbound_state(client
);
3075 ERR("[notification-thread] Failed to reset client communication's inbound state");
3080 pthread_mutex_unlock(&client
->lock
);
3085 int client_handle_message_subscription(
3086 struct notification_client
*client
,
3087 enum lttng_notification_channel_message_type msg_type
,
3088 struct notification_thread_state
*state
)
3091 struct lttng_condition
*condition
;
3092 enum lttng_notification_channel_status status
=
3093 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
3094 struct lttng_payload_view condition_view
=
3095 lttng_payload_view_from_dynamic_buffer(
3096 &client
->communication
.inbound
.buffer
,
3098 size_t expected_condition_size
;
3100 pthread_mutex_lock(&client
->lock
);
3101 expected_condition_size
= client
->communication
.inbound
.buffer
.size
;
3102 pthread_mutex_unlock(&client
->lock
);
3104 ret
= lttng_condition_create_from_payload(&condition_view
, &condition
);
3105 if (ret
!= expected_condition_size
) {
3106 ERR("[notification-thread] Malformed condition received from client");
3110 if (msg_type
== LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE
) {
3111 ret
= notification_thread_client_subscribe(
3112 client
, condition
, state
, &status
);
3114 ret
= notification_thread_client_unsubscribe(
3115 client
, condition
, state
, &status
);
3121 pthread_mutex_lock(&client
->lock
);
3122 ret
= client_send_command_reply(client
, state
, status
);
3124 ERR("[notification-thread] Failed to send reply to notification channel client");
3128 /* Set reception state to receive the next message header. */
3129 ret
= client_reset_inbound_state(client
);
3131 ERR("[notification-thread] Failed to reset client communication's inbound state");
3136 pthread_mutex_unlock(&client
->lock
);
3142 int client_dispatch_message(struct notification_client
*client
,
3143 struct notification_thread_state
*state
)
3147 if (client
->communication
.inbound
.msg_type
!=
3148 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
&&
3149 client
->communication
.inbound
.msg_type
!=
3150 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN
&&
3151 !client
->validated
) {
3152 WARN("[notification-thread] client attempted a command before handshake");
3157 switch (client
->communication
.inbound
.msg_type
) {
3158 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN
:
3160 ret
= client_handle_message_unknown(client
, state
);
3163 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
:
3165 ret
= client_handle_message_handshake(client
, state
);
3168 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE
:
3169 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE
:
3171 ret
= client_handle_message_subscription(client
,
3172 client
->communication
.inbound
.msg_type
, state
);
3182 /* Incoming data from client. */
3183 int handle_notification_thread_client_in(
3184 struct notification_thread_state
*state
, int socket
)
3187 struct notification_client
*client
;
3190 bool message_is_complete
= false;
3193 client
= get_client_from_socket(socket
, state
);
3195 /* Internal error, abort. */
3200 pthread_mutex_lock(&client
->lock
);
3201 offset
= client
->communication
.inbound
.buffer
.size
-
3202 client
->communication
.inbound
.bytes_to_receive
;
3203 if (client
->communication
.inbound
.expect_creds
) {
3204 recv_ret
= lttcomm_recv_creds_unix_sock(socket
,
3205 client
->communication
.inbound
.buffer
.data
+ offset
,
3206 client
->communication
.inbound
.bytes_to_receive
,
3207 &client
->communication
.inbound
.creds
);
3209 client
->communication
.inbound
.expect_creds
= false;
3210 client
->communication
.inbound
.creds_received
= true;
3213 recv_ret
= lttcomm_recv_unix_sock_non_block(socket
,
3214 client
->communication
.inbound
.buffer
.data
+ offset
,
3215 client
->communication
.inbound
.bytes_to_receive
);
3217 if (recv_ret
>= 0) {
3218 client
->communication
.inbound
.bytes_to_receive
-= recv_ret
;
3219 message_is_complete
= client
->communication
.inbound
3220 .bytes_to_receive
== 0;
3222 pthread_mutex_unlock(&client
->lock
);
3224 goto error_disconnect_client
;
3227 if (message_is_complete
) {
3228 ret
= client_dispatch_message(client
, state
);
3231 * Only returns an error if this client must be
3234 goto error_disconnect_client
;
3240 error_disconnect_client
:
3241 pthread_mutex_lock(&client
->lock
);
3242 ret
= notification_thread_client_disconnect(client
, state
);
3243 pthread_mutex_unlock(&client
->lock
);
3247 /* Client ready to receive outgoing data. */
3248 int handle_notification_thread_client_out(
3249 struct notification_thread_state
*state
, int socket
)
3252 struct notification_client
*client
;
3253 enum client_transmission_status transmission_status
;
3256 client
= get_client_from_socket(socket
, state
);
3258 /* Internal error, abort. */
3263 pthread_mutex_lock(&client
->lock
);
3264 transmission_status
= client_flush_outgoing_queue(client
);
3265 ret
= client_handle_transmission_status(
3266 client
, transmission_status
, state
);
3267 pthread_mutex_unlock(&client
->lock
);
3277 bool evaluate_buffer_usage_condition(const struct lttng_condition
*condition
,
3278 const struct channel_state_sample
*sample
,
3279 uint64_t buffer_capacity
)
3281 bool result
= false;
3283 enum lttng_condition_type condition_type
;
3284 const struct lttng_condition_buffer_usage
*use_condition
= container_of(
3285 condition
, struct lttng_condition_buffer_usage
,
3288 if (use_condition
->threshold_bytes
.set
) {
3289 threshold
= use_condition
->threshold_bytes
.value
;
3292 * Threshold was expressed as a ratio.
3294 * TODO the threshold (in bytes) of conditions expressed
3295 * as a ratio of total buffer size could be cached to
3296 * forego this double-multiplication or it could be performed
3297 * as fixed-point math.
3299 * Note that caching should accommodates the case where the
3300 * condition applies to multiple channels (i.e. don't assume
3301 * that all channels matching my_chann* have the same size...)
3303 threshold
= (uint64_t) (use_condition
->threshold_ratio
.value
*
3304 (double) buffer_capacity
);
3307 condition_type
= lttng_condition_get_type(condition
);
3308 if (condition_type
== LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
) {
3309 DBG("[notification-thread] Low buffer usage condition being evaluated: threshold = %" PRIu64
", highest usage = %" PRIu64
,
3310 threshold
, sample
->highest_usage
);
3313 * The low condition should only be triggered once _all_ of the
3314 * streams in a channel have gone below the "low" threshold.
3316 if (sample
->highest_usage
<= threshold
) {
3320 DBG("[notification-thread] High buffer usage condition being evaluated: threshold = %" PRIu64
", highest usage = %" PRIu64
,
3321 threshold
, sample
->highest_usage
);
3324 * For high buffer usage scenarios, we want to trigger whenever
3325 * _any_ of the streams has reached the "high" threshold.
3327 if (sample
->highest_usage
>= threshold
) {
3336 bool evaluate_session_consumed_size_condition(
3337 const struct lttng_condition
*condition
,
3338 uint64_t session_consumed_size
)
3341 const struct lttng_condition_session_consumed_size
*size_condition
=
3342 container_of(condition
,
3343 struct lttng_condition_session_consumed_size
,
3346 threshold
= size_condition
->consumed_threshold_bytes
.value
;
3347 DBG("[notification-thread] Session consumed size condition being evaluated: threshold = %" PRIu64
", current size = %" PRIu64
,
3348 threshold
, session_consumed_size
);
3349 return session_consumed_size
>= threshold
;
3353 int evaluate_buffer_condition(const struct lttng_condition
*condition
,
3354 struct lttng_evaluation
**evaluation
,
3355 const struct notification_thread_state
*state
,
3356 const struct channel_state_sample
*previous_sample
,
3357 const struct channel_state_sample
*latest_sample
,
3358 uint64_t previous_session_consumed_total
,
3359 uint64_t latest_session_consumed_total
,
3360 struct channel_info
*channel_info
)
3363 enum lttng_condition_type condition_type
;
3364 const bool previous_sample_available
= !!previous_sample
;
3365 bool previous_sample_result
= false;
3366 bool latest_sample_result
;
3368 condition_type
= lttng_condition_get_type(condition
);
3370 switch (condition_type
) {
3371 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
3372 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
3373 if (caa_likely(previous_sample_available
)) {
3374 previous_sample_result
=
3375 evaluate_buffer_usage_condition(condition
,
3376 previous_sample
, channel_info
->capacity
);
3378 latest_sample_result
= evaluate_buffer_usage_condition(
3379 condition
, latest_sample
,
3380 channel_info
->capacity
);
3382 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
3383 if (caa_likely(previous_sample_available
)) {
3384 previous_sample_result
=
3385 evaluate_session_consumed_size_condition(
3387 previous_session_consumed_total
);
3389 latest_sample_result
=
3390 evaluate_session_consumed_size_condition(
3392 latest_session_consumed_total
);
3395 /* Unknown condition type; internal error. */
3399 if (!latest_sample_result
||
3400 (previous_sample_result
== latest_sample_result
)) {
3402 * Only trigger on a condition evaluation transition.
3404 * NOTE: This edge-triggered logic may not be appropriate for
3405 * future condition types.
3410 if (!evaluation
|| !latest_sample_result
) {
3414 switch (condition_type
) {
3415 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW
:
3416 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH
:
3417 *evaluation
= lttng_evaluation_buffer_usage_create(
3419 latest_sample
->highest_usage
,
3420 channel_info
->capacity
);
3422 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
:
3423 *evaluation
= lttng_evaluation_session_consumed_size_create(
3424 latest_session_consumed_total
);
3439 int client_notification_overflow(struct notification_client
*client
)
3442 const struct lttng_notification_channel_message msg
= {
3443 .type
= (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED
,
3446 ASSERT_LOCKED(client
->lock
);
3448 DBG("Dropping notification addressed to client (socket fd = %i)",
3450 if (client
->communication
.outbound
.dropped_notification
) {
3452 * The client already has a "notification dropped" message
3453 * in its outgoing queue. Nothing to do since all
3454 * of those messages are coalesced.
3459 client
->communication
.outbound
.dropped_notification
= true;
3460 ret
= lttng_dynamic_buffer_append(
3461 &client
->communication
.outbound
.buffer
, &msg
,
3464 PERROR("Failed to enqueue \"dropped notification\" message in client's (socket fd = %i) outgoing queue",
3471 static int client_handle_transmission_status_wrapper(
3472 struct notification_client
*client
,
3473 enum client_transmission_status status
,
3476 return client_handle_transmission_status(client
, status
,
3477 (struct notification_thread_state
*) user_data
);
3481 int send_evaluation_to_clients(const struct lttng_trigger
*trigger
,
3482 const struct lttng_evaluation
*evaluation
,
3483 struct notification_client_list
* client_list
,
3484 struct notification_thread_state
*state
,
3485 uid_t object_uid
, gid_t object_gid
)
3487 return notification_client_list_send_evaluation(client_list
,
3488 lttng_trigger_get_const_condition(trigger
), evaluation
,
3489 lttng_trigger_get_credentials(trigger
),
3490 &(struct lttng_credentials
){
3491 .uid
= object_uid
, .gid
= object_gid
},
3492 client_handle_transmission_status_wrapper
, state
);
3496 * Permission checks relative to notification channel clients are performed
3497 * here. Notice how object, client, and trigger credentials are involved in
3500 * The `object` credentials are the credentials associated with the "subject"
3501 * of a condition. For instance, a `rotation completed` condition applies
3502 * to a session. When that condition is met, it will produce an evaluation
3503 * against a session. Hence, in this case, the `object` credentials are the
3504 * credentials of the "subject" session.
3506 * The `trigger` credentials are the credentials of the user that registered the
3509 * The `client` credentials are the credentials of the user that created a given
3510 * notification channel.
3512 * In terms of visibility, it is expected that non-privilieged users can only
3513 * register triggers against "their" objects (their own sessions and
3514 * applications they are allowed to interact with). They can then open a
3515 * notification channel and subscribe to notifications associated with those
3518 * As for privilieged users, they can register triggers against the objects of
3519 * other users. They can then subscribe to the notifications associated to their
3520 * triggers. Privilieged users _can't_ subscribe to the notifications of
3521 * triggers owned by other users; they must create their own triggers.
3523 * This is more a concern of usability than security. It would be difficult for
3524 * a root user reliably subscribe to a specific set of conditions without
3525 * interference from external users (those could, for instance, unregister
3529 int notification_client_list_send_evaluation(
3530 struct notification_client_list
*client_list
,
3531 const struct lttng_condition
*condition
,
3532 const struct lttng_evaluation
*evaluation
,
3533 const struct lttng_credentials
*trigger_creds
,
3534 const struct lttng_credentials
*source_object_creds
,
3535 report_client_transmission_result_cb client_report
,
3539 struct lttng_payload msg_payload
;
3540 struct notification_client_list_element
*client_list_element
, *tmp
;
3541 const struct lttng_notification notification
= {
3542 .condition
= (struct lttng_condition
*) condition
,
3543 .evaluation
= (struct lttng_evaluation
*) evaluation
,
3545 struct lttng_notification_channel_message msg_header
= {
3546 .type
= (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION
,
3549 lttng_payload_init(&msg_payload
);
3551 ret
= lttng_dynamic_buffer_append(&msg_payload
.buffer
, &msg_header
,
3552 sizeof(msg_header
));
3557 ret
= lttng_notification_serialize(¬ification
, &msg_payload
);
3559 ERR("[notification-thread] Failed to serialize notification");
3564 /* Update payload size. */
3565 ((struct lttng_notification_channel_message
*) msg_payload
.buffer
.data
)
3566 ->size
= (uint32_t)(
3567 msg_payload
.buffer
.size
- sizeof(msg_header
));
3569 pthread_mutex_lock(&client_list
->lock
);
3570 cds_list_for_each_entry_safe(client_list_element
, tmp
,
3571 &client_list
->list
, node
) {
3572 enum client_transmission_status transmission_status
;
3573 struct notification_client
*client
=
3574 client_list_element
->client
;
3577 pthread_mutex_lock(&client
->lock
);
3578 if (source_object_creds
) {
3579 if (client
->uid
!= source_object_creds
->uid
&&
3580 client
->gid
!= source_object_creds
->gid
&&
3583 * Client is not allowed to monitor this
3586 DBG("[notification-thread] Skipping client at it does not have the object permission to receive notification for this trigger");
3591 if (client
->uid
!= trigger_creds
->uid
&& client
->gid
!= trigger_creds
->gid
) {
3592 DBG("[notification-thread] Skipping client at it does not have the permission to receive notification for this trigger");
3596 DBG("[notification-thread] Sending notification to client (fd = %i, %zu bytes)",
3597 client
->socket
, msg_payload
.buffer
.size
);
3598 if (client
->communication
.outbound
.buffer
.size
) {
3600 * Outgoing data is already buffered for this client;
3601 * drop the notification and enqueue a "dropped
3602 * notification" message if this is the first dropped
3603 * notification since the socket spilled-over to the
3606 ret
= client_notification_overflow(client
);
3612 ret
= lttng_dynamic_buffer_append_buffer(
3613 &client
->communication
.outbound
.buffer
,
3614 &msg_payload
.buffer
);
3619 transmission_status
= client_flush_outgoing_queue(client
);
3620 ret
= client_report(client
, transmission_status
, user_data
);
3625 pthread_mutex_unlock(&client
->lock
);
3627 goto end_unlock_list
;
3633 pthread_mutex_unlock(&client_list
->lock
);
3635 lttng_payload_reset(&msg_payload
);
3639 int handle_notification_thread_channel_sample(
3640 struct notification_thread_state
*state
, int pipe
,
3641 enum lttng_domain_type domain
)
3644 struct lttcomm_consumer_channel_monitor_msg sample_msg
;
3645 struct channel_info
*channel_info
;
3646 struct cds_lfht_node
*node
;
3647 struct cds_lfht_iter iter
;
3648 struct lttng_channel_trigger_list
*trigger_list
;
3649 struct lttng_trigger_list_element
*trigger_list_element
;
3650 bool previous_sample_available
= false;
3651 struct channel_state_sample previous_sample
, latest_sample
;
3652 uint64_t previous_session_consumed_total
, latest_session_consumed_total
;
3653 struct lttng_credentials channel_creds
;
3656 * The monitoring pipe only holds messages smaller than PIPE_BUF,
3657 * ensuring that read/write of sampling messages are atomic.
3659 ret
= lttng_read(pipe
, &sample_msg
, sizeof(sample_msg
));
3660 if (ret
!= sizeof(sample_msg
)) {
3661 ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
3668 latest_sample
.key
.key
= sample_msg
.key
;
3669 latest_sample
.key
.domain
= domain
;
3670 latest_sample
.highest_usage
= sample_msg
.highest
;
3671 latest_sample
.lowest_usage
= sample_msg
.lowest
;
3672 latest_sample
.channel_total_consumed
= sample_msg
.total_consumed
;
3676 /* Retrieve the channel's informations */
3677 cds_lfht_lookup(state
->channels_ht
,
3678 hash_channel_key(&latest_sample
.key
),
3682 node
= cds_lfht_iter_get_node(&iter
);
3683 if (caa_unlikely(!node
)) {
3685 * Not an error since the consumer can push a sample to the pipe
3686 * and the rest of the session daemon could notify us of the
3687 * channel's destruction before we get a chance to process that
3690 DBG("[notification-thread] Received a sample for an unknown channel from consumerd, key = %" PRIu64
" in %s domain",
3691 latest_sample
.key
.key
,
3692 domain
== LTTNG_DOMAIN_KERNEL
? "kernel" :
3696 channel_info
= caa_container_of(node
, struct channel_info
,
3698 DBG("[notification-thread] Handling channel sample for channel %s (key = %" PRIu64
") in session %s (highest usage = %" PRIu64
", lowest usage = %" PRIu64
", total consumed = %" PRIu64
")",
3700 latest_sample
.key
.key
,
3701 channel_info
->session_info
->name
,
3702 latest_sample
.highest_usage
,
3703 latest_sample
.lowest_usage
,
3704 latest_sample
.channel_total_consumed
);
3706 previous_session_consumed_total
=
3707 channel_info
->session_info
->consumed_data_size
;
3709 /* Retrieve the channel's last sample, if it exists, and update it. */
3710 cds_lfht_lookup(state
->channel_state_ht
,
3711 hash_channel_key(&latest_sample
.key
),
3712 match_channel_state_sample
,
3715 node
= cds_lfht_iter_get_node(&iter
);
3716 if (caa_likely(node
)) {
3717 struct channel_state_sample
*stored_sample
;
3719 /* Update the sample stored. */
3720 stored_sample
= caa_container_of(node
,
3721 struct channel_state_sample
,
3722 channel_state_ht_node
);
3724 memcpy(&previous_sample
, stored_sample
,
3725 sizeof(previous_sample
));
3726 stored_sample
->highest_usage
= latest_sample
.highest_usage
;
3727 stored_sample
->lowest_usage
= latest_sample
.lowest_usage
;
3728 stored_sample
->channel_total_consumed
= latest_sample
.channel_total_consumed
;
3729 previous_sample_available
= true;
3731 latest_session_consumed_total
=
3732 previous_session_consumed_total
+
3733 (latest_sample
.channel_total_consumed
- previous_sample
.channel_total_consumed
);
3736 * This is the channel's first sample, allocate space for and
3737 * store the new sample.
3739 struct channel_state_sample
*stored_sample
;
3741 stored_sample
= zmalloc(sizeof(*stored_sample
));
3742 if (!stored_sample
) {
3747 memcpy(stored_sample
, &latest_sample
, sizeof(*stored_sample
));
3748 cds_lfht_node_init(&stored_sample
->channel_state_ht_node
);
3749 cds_lfht_add(state
->channel_state_ht
,
3750 hash_channel_key(&stored_sample
->key
),
3751 &stored_sample
->channel_state_ht_node
);
3753 latest_session_consumed_total
=
3754 previous_session_consumed_total
+
3755 latest_sample
.channel_total_consumed
;
3758 channel_info
->session_info
->consumed_data_size
=
3759 latest_session_consumed_total
;
3761 /* Find triggers associated with this channel. */
3762 cds_lfht_lookup(state
->channel_triggers_ht
,
3763 hash_channel_key(&latest_sample
.key
),
3764 match_channel_trigger_list
,
3767 node
= cds_lfht_iter_get_node(&iter
);
3768 if (caa_likely(!node
)) {
3772 channel_creds
= (typeof(channel_creds
)) {
3773 .uid
= channel_info
->session_info
->uid
,
3774 .gid
= channel_info
->session_info
->gid
,
3777 trigger_list
= caa_container_of(node
, struct lttng_channel_trigger_list
,
3778 channel_triggers_ht_node
);
3779 cds_list_for_each_entry(trigger_list_element
, &trigger_list
->list
,
3781 const struct lttng_condition
*condition
;
3782 struct lttng_trigger
*trigger
;
3783 struct notification_client_list
*client_list
= NULL
;
3784 struct lttng_evaluation
*evaluation
= NULL
;
3785 enum action_executor_status executor_status
;
3788 trigger
= trigger_list_element
->trigger
;
3789 condition
= lttng_trigger_get_const_condition(trigger
);
3793 * Check if any client is subscribed to the result of this
3796 client_list
= get_client_list_from_condition(state
, condition
);
3798 ret
= evaluate_buffer_condition(condition
, &evaluation
, state
,
3799 previous_sample_available
? &previous_sample
: NULL
,
3801 previous_session_consumed_total
,
3802 latest_session_consumed_total
,
3804 if (caa_unlikely(ret
)) {
3808 if (caa_likely(!evaluation
)) {
3813 * Ownership of `evaluation` transferred to the action executor
3814 * no matter the result.
3816 executor_status
= action_executor_enqueue(state
->executor
,
3817 trigger
, evaluation
, &channel_creds
,
3820 switch (executor_status
) {
3821 case ACTION_EXECUTOR_STATUS_OK
:
3823 case ACTION_EXECUTOR_STATUS_ERROR
:
3824 case ACTION_EXECUTOR_STATUS_INVALID
:
3826 * TODO Add trigger identification (name/id) when
3827 * it is added to the API.
3829 ERR("Fatal error occurred while enqueuing action associated with buffer-condition trigger");
3832 case ACTION_EXECUTOR_STATUS_OVERFLOW
:
3834 * TODO Add trigger identification (name/id) when
3835 * it is added to the API.
3837 * Not a fatal error.
3839 WARN("No space left when enqueuing action associated with buffer-condition trigger");
3847 notification_client_list_put(client_list
);
3848 if (caa_unlikely(ret
)) {