Rename C++ header files to .hpp
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread-events.cpp
1 /*
2 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 *
4 * SPDX-License-Identifier: GPL-2.0-only
5 *
6 */
7
8 #include "lttng/action/action.h"
9 #include "lttng/trigger/trigger-internal.hpp"
10 #define _LGPL_SOURCE
11 #include <urcu.h>
12 #include <urcu/rculfhash.h>
13
14 #include <common/defaults.hpp>
15 #include <common/error.hpp>
16 #include <common/futex.hpp>
17 #include <common/unix.hpp>
18 #include <common/dynamic-buffer.hpp>
19 #include <common/hashtable/utils.hpp>
20 #include <common/sessiond-comm/sessiond-comm.hpp>
21 #include <common/macros.hpp>
22 #include <lttng/condition/condition.h>
23 #include <lttng/action/action-internal.hpp>
24 #include <lttng/action/list-internal.hpp>
25 #include <lttng/domain-internal.hpp>
26 #include <lttng/notification/notification-internal.hpp>
27 #include <lttng/condition/condition-internal.hpp>
28 #include <lttng/condition/buffer-usage-internal.hpp>
29 #include <lttng/condition/session-consumed-size-internal.hpp>
30 #include <lttng/condition/session-rotation-internal.hpp>
31 #include <lttng/condition/event-rule-matches-internal.hpp>
32 #include <lttng/domain-internal.hpp>
33 #include <lttng/notification/channel-internal.hpp>
34 #include <lttng/trigger/trigger-internal.hpp>
35 #include <lttng/event-rule/event-rule-internal.hpp>
36
37 #include <time.h>
38 #include <unistd.h>
39 #include <inttypes.h>
40 #include <fcntl.h>
41
42 #include "condition-internal.hpp"
43 #include "event-notifier-error-accounting.hpp"
44 #include "notification-thread.hpp"
45 #include "notification-thread-events.hpp"
46 #include "notification-thread-commands.hpp"
47 #include "lttng-sessiond.hpp"
48 #include "kernel.hpp"
49
50 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
51 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
52
53 /* The tracers currently limit the capture size to PIPE_BUF (4kb on linux). */
54 #define MAX_CAPTURE_SIZE (PIPE_BUF)
55
56 enum lttng_object_type {
57 LTTNG_OBJECT_TYPE_UNKNOWN,
58 LTTNG_OBJECT_TYPE_NONE,
59 LTTNG_OBJECT_TYPE_CHANNEL,
60 LTTNG_OBJECT_TYPE_SESSION,
61 };
62
63 struct lttng_trigger_list_element {
64 /* No ownership of the trigger object is assumed. */
65 struct lttng_trigger *trigger;
66 struct cds_list_head node;
67 };
68
69 struct lttng_channel_trigger_list {
70 struct channel_key channel_key;
71 /* List of struct lttng_trigger_list_element. */
72 struct cds_list_head list;
73 /* Node in the channel_triggers_ht */
74 struct cds_lfht_node channel_triggers_ht_node;
75 /* call_rcu delayed reclaim. */
76 struct rcu_head rcu_node;
77 };
78
79 /*
80 * List of triggers applying to a given session.
81 *
82 * See:
83 * - lttng_session_trigger_list_create()
84 * - lttng_session_trigger_list_build()
85 * - lttng_session_trigger_list_destroy()
86 * - lttng_session_trigger_list_add()
87 */
88 struct lttng_session_trigger_list {
89 /*
90 * Not owned by this; points to the session_info structure's
91 * session name.
92 */
93 const char *session_name;
94 /* List of struct lttng_trigger_list_element. */
95 struct cds_list_head list;
96 /* Node in the session_triggers_ht */
97 struct cds_lfht_node session_triggers_ht_node;
98 /*
99 * Weak reference to the notification system's session triggers
100 * hashtable.
101 *
102 * The session trigger list structure structure is owned by
103 * the session's session_info.
104 *
105 * The session_info is kept alive the the channel_infos holding a
106 * reference to it (reference counting). When those channels are
107 * destroyed (at runtime or on teardown), the reference they hold
108 * to the session_info are released. On destruction of session_info,
109 * session_info_destroy() will remove the list of triggers applying
110 * to this session from the notification system's state.
111 *
112 * This implies that the session_triggers_ht must be destroyed
113 * after the channels.
114 */
115 struct cds_lfht *session_triggers_ht;
116 /* Used for delayed RCU reclaim. */
117 struct rcu_head rcu_node;
118 };
119
120 struct lttng_trigger_ht_element {
121 struct lttng_trigger *trigger;
122 struct cds_lfht_node node;
123 struct cds_lfht_node node_by_name_uid;
124 struct cds_list_head client_list_trigger_node;
125 /* call_rcu delayed reclaim. */
126 struct rcu_head rcu_node;
127 };
128
129 struct lttng_condition_list_element {
130 struct lttng_condition *condition;
131 struct cds_list_head node;
132 };
133
134 struct channel_state_sample {
135 struct channel_key key;
136 struct cds_lfht_node channel_state_ht_node;
137 uint64_t highest_usage;
138 uint64_t lowest_usage;
139 uint64_t channel_total_consumed;
140 /* call_rcu delayed reclaim. */
141 struct rcu_head rcu_node;
142 };
143
144 static unsigned long hash_channel_key(struct channel_key *key);
145 static int evaluate_buffer_condition(const struct lttng_condition *condition,
146 struct lttng_evaluation **evaluation,
147 const struct notification_thread_state *state,
148 const struct channel_state_sample *previous_sample,
149 const struct channel_state_sample *latest_sample,
150 uint64_t previous_session_consumed_total,
151 uint64_t latest_session_consumed_total,
152 struct channel_info *channel_info);
153 static
154 int send_evaluation_to_clients(const struct lttng_trigger *trigger,
155 const struct lttng_evaluation *evaluation,
156 struct notification_client_list *client_list,
157 struct notification_thread_state *state,
158 uid_t channel_uid, gid_t channel_gid);
159
160
161 /* session_info API */
162 static
163 void session_info_destroy(void *_data);
164 static
165 void session_info_get(struct session_info *session_info);
166 static
167 void session_info_put(struct session_info *session_info);
168 static
169 struct session_info *session_info_create(const char *name,
170 uid_t uid, gid_t gid,
171 struct lttng_session_trigger_list *trigger_list,
172 struct cds_lfht *sessions_ht);
173 static
174 void session_info_add_channel(struct session_info *session_info,
175 struct channel_info *channel_info);
176 static
177 void session_info_remove_channel(struct session_info *session_info,
178 struct channel_info *channel_info);
179
180 /* lttng_session_trigger_list API */
181 static
182 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
183 const char *session_name,
184 struct cds_lfht *session_triggers_ht);
185 static
186 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
187 const struct notification_thread_state *state,
188 const char *session_name);
189 static
190 void lttng_session_trigger_list_destroy(
191 struct lttng_session_trigger_list *list);
192 static
193 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
194 struct lttng_trigger *trigger);
195
196 static
197 int client_handle_transmission_status(
198 struct notification_client *client,
199 enum client_transmission_status transmission_status,
200 struct notification_thread_state *state);
201
202 static
203 int handle_one_event_notifier_notification(
204 struct notification_thread_state *state,
205 int pipe, enum lttng_domain_type domain);
206
207 static
208 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node);
209
210 static
211 int match_client_socket(struct cds_lfht_node *node, const void *key)
212 {
213 /* This double-cast is intended to supress pointer-to-cast warning. */
214 const int socket = (int) (intptr_t) key;
215 const struct notification_client *client = caa_container_of(node,
216 struct notification_client, client_socket_ht_node);
217
218 return client->socket == socket;
219 }
220
221 static
222 int match_client_id(struct cds_lfht_node *node, const void *key)
223 {
224 /* This double-cast is intended to supress pointer-to-cast warning. */
225 const notification_client_id id = *((notification_client_id *) key);
226 const struct notification_client *client = caa_container_of(
227 node, struct notification_client, client_id_ht_node);
228
229 return client->id == id;
230 }
231
232 static
233 int match_channel_trigger_list(struct cds_lfht_node *node, const void *key)
234 {
235 struct channel_key *channel_key = (struct channel_key *) key;
236 struct lttng_channel_trigger_list *trigger_list;
237
238 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
239 channel_triggers_ht_node);
240
241 return !!((channel_key->key == trigger_list->channel_key.key) &&
242 (channel_key->domain == trigger_list->channel_key.domain));
243 }
244
245 static
246 int match_session_trigger_list(struct cds_lfht_node *node, const void *key)
247 {
248 const char *session_name = (const char *) key;
249 struct lttng_session_trigger_list *trigger_list;
250
251 trigger_list = caa_container_of(node, struct lttng_session_trigger_list,
252 session_triggers_ht_node);
253
254 return !!(strcmp(trigger_list->session_name, session_name) == 0);
255 }
256
257 static
258 int match_channel_state_sample(struct cds_lfht_node *node, const void *key)
259 {
260 struct channel_key *channel_key = (struct channel_key *) key;
261 struct channel_state_sample *sample;
262
263 sample = caa_container_of(node, struct channel_state_sample,
264 channel_state_ht_node);
265
266 return !!((channel_key->key == sample->key.key) &&
267 (channel_key->domain == sample->key.domain));
268 }
269
270 static
271 int match_channel_info(struct cds_lfht_node *node, const void *key)
272 {
273 struct channel_key *channel_key = (struct channel_key *) key;
274 struct channel_info *channel_info;
275
276 channel_info = caa_container_of(node, struct channel_info,
277 channels_ht_node);
278
279 return !!((channel_key->key == channel_info->key.key) &&
280 (channel_key->domain == channel_info->key.domain));
281 }
282
283 static
284 int match_trigger(struct cds_lfht_node *node, const void *key)
285 {
286 struct lttng_trigger *trigger_key = (struct lttng_trigger *) key;
287 struct lttng_trigger_ht_element *trigger_ht_element;
288
289 trigger_ht_element = caa_container_of(node, struct lttng_trigger_ht_element,
290 node);
291
292 return !!lttng_trigger_is_equal(trigger_key, trigger_ht_element->trigger);
293 }
294
295 static
296 int match_trigger_token(struct cds_lfht_node *node, const void *key)
297 {
298 const uint64_t *_key = (uint64_t *) key;
299 struct notification_trigger_tokens_ht_element *element;
300
301 element = caa_container_of(node,
302 struct notification_trigger_tokens_ht_element, node);
303 return *_key == element->token;
304 }
305
306 static
307 int match_client_list_condition(struct cds_lfht_node *node, const void *key)
308 {
309 struct lttng_condition *condition_key = (struct lttng_condition *) key;
310 struct notification_client_list *client_list;
311 const struct lttng_condition *condition;
312
313 LTTNG_ASSERT(condition_key);
314
315 client_list = caa_container_of(node, struct notification_client_list,
316 notification_trigger_clients_ht_node);
317 condition = client_list->condition;
318
319 return !!lttng_condition_is_equal(condition_key, condition);
320 }
321
322 static
323 int match_session(struct cds_lfht_node *node, const void *key)
324 {
325 const char *name = (const char *) key;
326 struct session_info *session_info = caa_container_of(
327 node, struct session_info, sessions_ht_node);
328
329 return !strcmp(session_info->name, name);
330 }
331
332 static
333 const char *notification_command_type_str(
334 enum notification_thread_command_type type)
335 {
336 switch (type) {
337 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
338 return "REGISTER_TRIGGER";
339 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
340 return "UNREGISTER_TRIGGER";
341 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
342 return "ADD_CHANNEL";
343 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
344 return "REMOVE_CHANNEL";
345 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
346 return "SESSION_ROTATION_ONGOING";
347 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
348 return "SESSION_ROTATION_COMPLETED";
349 case NOTIFICATION_COMMAND_TYPE_ADD_TRACER_EVENT_SOURCE:
350 return "ADD_TRACER_EVENT_SOURCE";
351 case NOTIFICATION_COMMAND_TYPE_REMOVE_TRACER_EVENT_SOURCE:
352 return "REMOVE_TRACER_EVENT_SOURCE";
353 case NOTIFICATION_COMMAND_TYPE_LIST_TRIGGERS:
354 return "LIST_TRIGGERS";
355 case NOTIFICATION_COMMAND_TYPE_GET_TRIGGER:
356 return "GET_TRIGGER";
357 case NOTIFICATION_COMMAND_TYPE_QUIT:
358 return "QUIT";
359 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE:
360 return "CLIENT_COMMUNICATION_UPDATE";
361 default:
362 abort();
363 }
364 }
365
366 /*
367 * Match trigger based on name and credentials only.
368 * Name duplication is NOT allowed for the same uid.
369 */
370 static
371 int match_trigger_by_name_uid(struct cds_lfht_node *node,
372 const void *key)
373 {
374 bool match = false;
375 const char *element_trigger_name;
376 const char *key_name;
377 enum lttng_trigger_status status;
378 const struct lttng_credentials *key_creds;
379 const struct lttng_credentials *node_creds;
380 const struct lttng_trigger *trigger_key =
381 (const struct lttng_trigger *) key;
382 const struct lttng_trigger_ht_element *trigger_ht_element =
383 caa_container_of(node,
384 struct lttng_trigger_ht_element,
385 node_by_name_uid);
386
387 status = lttng_trigger_get_name(trigger_ht_element->trigger,
388 &element_trigger_name);
389 element_trigger_name = status == LTTNG_TRIGGER_STATUS_OK ?
390 element_trigger_name : NULL;
391
392 status = lttng_trigger_get_name(trigger_key, &key_name);
393 key_name = status == LTTNG_TRIGGER_STATUS_OK ? key_name : NULL;
394
395 /*
396 * Compare the names.
397 * Consider null names as not equal. This is to maintain backwards
398 * compatibility with pre-2.13 anonymous triggers. Multiples anonymous
399 * triggers are allowed for a given user.
400 */
401 if (!element_trigger_name || !key_name) {
402 goto end;
403 }
404
405 if (strcmp(element_trigger_name, key_name) != 0) {
406 goto end;
407 }
408
409 /* Compare the owners' UIDs. */
410 key_creds = lttng_trigger_get_credentials(trigger_key);
411 node_creds = lttng_trigger_get_credentials(trigger_ht_element->trigger);
412
413 match = lttng_credentials_is_equal_uid(key_creds, node_creds);
414
415 end:
416 return match;
417 }
418
419 /*
420 * Hash trigger based on name and credentials only.
421 */
422 static
423 unsigned long hash_trigger_by_name_uid(const struct lttng_trigger *trigger)
424 {
425 unsigned long hash = 0;
426 const struct lttng_credentials *trigger_creds;
427 const char *trigger_name;
428 enum lttng_trigger_status status;
429
430 status = lttng_trigger_get_name(trigger, &trigger_name);
431 if (status == LTTNG_TRIGGER_STATUS_OK) {
432 hash = hash_key_str(trigger_name, lttng_ht_seed);
433 }
434
435 trigger_creds = lttng_trigger_get_credentials(trigger);
436 hash ^= hash_key_ulong((void *) (unsigned long) LTTNG_OPTIONAL_GET(trigger_creds->uid),
437 lttng_ht_seed);
438
439 return hash;
440 }
441
442 static
443 unsigned long hash_channel_key(struct channel_key *key)
444 {
445 unsigned long key_hash = hash_key_u64(&key->key, lttng_ht_seed);
446 unsigned long domain_hash = hash_key_ulong(
447 (void *) (unsigned long) key->domain, lttng_ht_seed);
448
449 return key_hash ^ domain_hash;
450 }
451
452 static
453 unsigned long hash_client_socket(int socket)
454 {
455 return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
456 }
457
458 static
459 unsigned long hash_client_id(notification_client_id id)
460 {
461 return hash_key_u64(&id, lttng_ht_seed);
462 }
463
464 /*
465 * Get the type of object to which a given condition applies. Bindings let
466 * the notification system evaluate a trigger's condition when a given
467 * object's state is updated.
468 *
469 * For instance, a condition bound to a channel will be evaluated everytime
470 * the channel's state is changed by a channel monitoring sample.
471 */
472 static
473 enum lttng_object_type get_condition_binding_object(
474 const struct lttng_condition *condition)
475 {
476 switch (lttng_condition_get_type(condition)) {
477 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
478 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
479 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
480 return LTTNG_OBJECT_TYPE_CHANNEL;
481 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
482 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
483 return LTTNG_OBJECT_TYPE_SESSION;
484 case LTTNG_CONDITION_TYPE_EVENT_RULE_MATCHES:
485 return LTTNG_OBJECT_TYPE_NONE;
486 default:
487 return LTTNG_OBJECT_TYPE_UNKNOWN;
488 }
489 }
490
491 static
492 void free_channel_info_rcu(struct rcu_head *node)
493 {
494 free(caa_container_of(node, struct channel_info, rcu_node));
495 }
496
497 static
498 void channel_info_destroy(struct channel_info *channel_info)
499 {
500 if (!channel_info) {
501 return;
502 }
503
504 if (channel_info->session_info) {
505 session_info_remove_channel(channel_info->session_info,
506 channel_info);
507 session_info_put(channel_info->session_info);
508 }
509 if (channel_info->name) {
510 free(channel_info->name);
511 }
512 call_rcu(&channel_info->rcu_node, free_channel_info_rcu);
513 }
514
515 static
516 void free_session_info_rcu(struct rcu_head *node)
517 {
518 free(caa_container_of(node, struct session_info, rcu_node));
519 }
520
521 /* Don't call directly, use the ref-counting mechanism. */
522 static
523 void session_info_destroy(void *_data)
524 {
525 struct session_info *session_info = (struct session_info *) _data;
526 int ret;
527
528 LTTNG_ASSERT(session_info);
529 if (session_info->channel_infos_ht) {
530 ret = cds_lfht_destroy(session_info->channel_infos_ht, NULL);
531 if (ret) {
532 ERR("Failed to destroy channel information hash table");
533 }
534 }
535 lttng_session_trigger_list_destroy(session_info->trigger_list);
536
537 rcu_read_lock();
538 cds_lfht_del(session_info->sessions_ht,
539 &session_info->sessions_ht_node);
540 rcu_read_unlock();
541 free(session_info->name);
542 call_rcu(&session_info->rcu_node, free_session_info_rcu);
543 }
544
545 static
546 void session_info_get(struct session_info *session_info)
547 {
548 if (!session_info) {
549 return;
550 }
551 lttng_ref_get(&session_info->ref);
552 }
553
554 static
555 void session_info_put(struct session_info *session_info)
556 {
557 if (!session_info) {
558 return;
559 }
560 lttng_ref_put(&session_info->ref);
561 }
562
563 static
564 struct session_info *session_info_create(const char *name, uid_t uid, gid_t gid,
565 struct lttng_session_trigger_list *trigger_list,
566 struct cds_lfht *sessions_ht)
567 {
568 struct session_info *session_info;
569
570 LTTNG_ASSERT(name);
571
572 session_info = (struct session_info *) zmalloc(sizeof(*session_info));
573 if (!session_info) {
574 goto end;
575 }
576 lttng_ref_init(&session_info->ref, session_info_destroy);
577
578 session_info->channel_infos_ht = cds_lfht_new(DEFAULT_HT_SIZE,
579 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
580 if (!session_info->channel_infos_ht) {
581 goto error;
582 }
583
584 cds_lfht_node_init(&session_info->sessions_ht_node);
585 session_info->name = strdup(name);
586 if (!session_info->name) {
587 goto error;
588 }
589 session_info->uid = uid;
590 session_info->gid = gid;
591 session_info->trigger_list = trigger_list;
592 session_info->sessions_ht = sessions_ht;
593 end:
594 return session_info;
595 error:
596 session_info_put(session_info);
597 return NULL;
598 }
599
600 static
601 void session_info_add_channel(struct session_info *session_info,
602 struct channel_info *channel_info)
603 {
604 rcu_read_lock();
605 cds_lfht_add(session_info->channel_infos_ht,
606 hash_channel_key(&channel_info->key),
607 &channel_info->session_info_channels_ht_node);
608 rcu_read_unlock();
609 }
610
611 static
612 void session_info_remove_channel(struct session_info *session_info,
613 struct channel_info *channel_info)
614 {
615 rcu_read_lock();
616 cds_lfht_del(session_info->channel_infos_ht,
617 &channel_info->session_info_channels_ht_node);
618 rcu_read_unlock();
619 }
620
621 static
622 struct channel_info *channel_info_create(const char *channel_name,
623 struct channel_key *channel_key, uint64_t channel_capacity,
624 struct session_info *session_info)
625 {
626 struct channel_info *channel_info = (struct channel_info *) zmalloc(sizeof(*channel_info));
627
628 if (!channel_info) {
629 goto end;
630 }
631
632 cds_lfht_node_init(&channel_info->channels_ht_node);
633 cds_lfht_node_init(&channel_info->session_info_channels_ht_node);
634 memcpy(&channel_info->key, channel_key, sizeof(*channel_key));
635 channel_info->capacity = channel_capacity;
636
637 channel_info->name = strdup(channel_name);
638 if (!channel_info->name) {
639 goto error;
640 }
641
642 /*
643 * Set the references between session and channel infos:
644 * - channel_info holds a strong reference to session_info
645 * - session_info holds a weak reference to channel_info
646 */
647 session_info_get(session_info);
648 session_info_add_channel(session_info, channel_info);
649 channel_info->session_info = session_info;
650 end:
651 return channel_info;
652 error:
653 channel_info_destroy(channel_info);
654 return NULL;
655 }
656
657 bool notification_client_list_get(struct notification_client_list *list)
658 {
659 return urcu_ref_get_unless_zero(&list->ref);
660 }
661
662 static
663 void free_notification_client_list_rcu(struct rcu_head *node)
664 {
665 free(caa_container_of(node, struct notification_client_list,
666 rcu_node));
667 }
668
669 static
670 void notification_client_list_release(struct urcu_ref *list_ref)
671 {
672 struct notification_client_list *list =
673 container_of(list_ref, typeof(*list), ref);
674 struct notification_client_list_element *client_list_element, *tmp;
675
676 lttng_condition_put(list->condition);
677
678 if (list->notification_trigger_clients_ht) {
679 rcu_read_lock();
680
681 cds_lfht_del(list->notification_trigger_clients_ht,
682 &list->notification_trigger_clients_ht_node);
683 rcu_read_unlock();
684 list->notification_trigger_clients_ht = NULL;
685 }
686 cds_list_for_each_entry_safe(client_list_element, tmp,
687 &list->clients_list, node) {
688 free(client_list_element);
689 }
690
691 LTTNG_ASSERT(cds_list_empty(&list->triggers_list));
692
693 pthread_mutex_destroy(&list->lock);
694 call_rcu(&list->rcu_node, free_notification_client_list_rcu);
695 }
696
697 static
698 bool condition_applies_to_client(const struct lttng_condition *condition,
699 struct notification_client *client)
700 {
701 bool applies = false;
702 struct lttng_condition_list_element *condition_list_element;
703
704 cds_list_for_each_entry(condition_list_element, &client->condition_list,
705 node) {
706 applies = lttng_condition_is_equal(
707 condition_list_element->condition,
708 condition);
709 if (applies) {
710 break;
711 }
712 }
713
714 return applies;
715 }
716
717 static
718 struct notification_client_list *notification_client_list_create(
719 struct notification_thread_state *state,
720 const struct lttng_condition *condition)
721 {
722 struct notification_client *client;
723 struct cds_lfht_iter iter;
724 struct notification_client_list *client_list;
725
726 client_list = (notification_client_list *) zmalloc(sizeof(*client_list));
727 if (!client_list) {
728 PERROR("Failed to allocate notification client list");
729 goto end;
730 }
731
732 pthread_mutex_init(&client_list->lock, NULL);
733 /*
734 * The trigger that owns the condition has the first reference to this
735 * client list.
736 */
737 urcu_ref_init(&client_list->ref);
738 cds_lfht_node_init(&client_list->notification_trigger_clients_ht_node);
739 CDS_INIT_LIST_HEAD(&client_list->clients_list);
740 CDS_INIT_LIST_HEAD(&client_list->triggers_list);
741
742 /*
743 * Create a copy of the condition so that it's independent of any
744 * trigger. The client list may outlive the trigger object (which owns
745 * the condition) that is used to create it.
746 */
747 client_list->condition = lttng_condition_copy(condition);
748
749 /* Build a list of clients to which this new condition applies. */
750 cds_lfht_for_each_entry (state->client_socket_ht, &iter, client,
751 client_socket_ht_node) {
752 struct notification_client_list_element *client_list_element;
753
754 if (!condition_applies_to_client(condition, client)) {
755 continue;
756 }
757
758 client_list_element = (notification_client_list_element *) zmalloc(sizeof(*client_list_element));
759 if (!client_list_element) {
760 goto error_put_client_list;
761 }
762
763 CDS_INIT_LIST_HEAD(&client_list_element->node);
764 client_list_element->client = client;
765 cds_list_add(&client_list_element->node, &client_list->clients_list);
766 }
767
768 client_list->notification_trigger_clients_ht =
769 state->notification_trigger_clients_ht;
770
771 rcu_read_lock();
772 /*
773 * Add the client list to the global list of client list.
774 */
775 cds_lfht_add_unique(state->notification_trigger_clients_ht,
776 lttng_condition_hash(client_list->condition),
777 match_client_list_condition,
778 client_list->condition,
779 &client_list->notification_trigger_clients_ht_node);
780 rcu_read_unlock();
781 goto end;
782
783 error_put_client_list:
784 notification_client_list_put(client_list);
785 client_list = NULL;
786
787 end:
788 return client_list;
789 }
790
791 void notification_client_list_put(struct notification_client_list *list)
792 {
793 if (!list) {
794 return;
795 }
796 return urcu_ref_put(&list->ref, notification_client_list_release);
797 }
798
799 /* Provides a reference to the returned list. */
800 static
801 struct notification_client_list *get_client_list_from_condition(
802 struct notification_thread_state *state,
803 const struct lttng_condition *condition)
804 {
805 struct cds_lfht_node *node;
806 struct cds_lfht_iter iter;
807 struct notification_client_list *list = NULL;
808
809 rcu_read_lock();
810 cds_lfht_lookup(state->notification_trigger_clients_ht,
811 lttng_condition_hash(condition),
812 match_client_list_condition,
813 condition,
814 &iter);
815 node = cds_lfht_iter_get_node(&iter);
816 if (node) {
817 list = container_of(node, struct notification_client_list,
818 notification_trigger_clients_ht_node);
819 list = notification_client_list_get(list) ? list : NULL;
820 }
821
822 rcu_read_unlock();
823 return list;
824 }
825
826 static
827 int evaluate_channel_condition_for_client(
828 const struct lttng_condition *condition,
829 struct notification_thread_state *state,
830 struct lttng_evaluation **evaluation,
831 uid_t *session_uid, gid_t *session_gid)
832 {
833 int ret;
834 struct cds_lfht_iter iter;
835 struct cds_lfht_node *node;
836 struct channel_info *channel_info = NULL;
837 struct channel_key *channel_key = NULL;
838 struct channel_state_sample *last_sample = NULL;
839 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
840
841 rcu_read_lock();
842
843 /* Find the channel associated with the condition. */
844 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter,
845 channel_trigger_list, channel_triggers_ht_node) {
846 struct lttng_trigger_list_element *element;
847
848 cds_list_for_each_entry(element, &channel_trigger_list->list, node) {
849 const struct lttng_condition *current_condition =
850 lttng_trigger_get_const_condition(
851 element->trigger);
852
853 LTTNG_ASSERT(current_condition);
854 if (!lttng_condition_is_equal(condition,
855 current_condition)) {
856 continue;
857 }
858
859 /* Found the trigger, save the channel key. */
860 channel_key = &channel_trigger_list->channel_key;
861 break;
862 }
863 if (channel_key) {
864 /* The channel key was found stop iteration. */
865 break;
866 }
867 }
868
869 if (!channel_key){
870 /* No channel found; normal exit. */
871 DBG("No known channel associated with newly subscribed-to condition");
872 ret = 0;
873 goto end;
874 }
875
876 /* Fetch channel info for the matching channel. */
877 cds_lfht_lookup(state->channels_ht,
878 hash_channel_key(channel_key),
879 match_channel_info,
880 channel_key,
881 &iter);
882 node = cds_lfht_iter_get_node(&iter);
883 LTTNG_ASSERT(node);
884 channel_info = caa_container_of(node, struct channel_info,
885 channels_ht_node);
886
887 /* Retrieve the channel's last sample, if it exists. */
888 cds_lfht_lookup(state->channel_state_ht,
889 hash_channel_key(channel_key),
890 match_channel_state_sample,
891 channel_key,
892 &iter);
893 node = cds_lfht_iter_get_node(&iter);
894 if (node) {
895 last_sample = caa_container_of(node,
896 struct channel_state_sample,
897 channel_state_ht_node);
898 } else {
899 /* Nothing to evaluate, no sample was ever taken. Normal exit */
900 DBG("No channel sample associated with newly subscribed-to condition");
901 ret = 0;
902 goto end;
903 }
904
905 ret = evaluate_buffer_condition(condition, evaluation, state,
906 NULL, last_sample,
907 0, channel_info->session_info->consumed_data_size,
908 channel_info);
909 if (ret) {
910 WARN("Fatal error occurred while evaluating a newly subscribed-to condition");
911 goto end;
912 }
913
914 *session_uid = channel_info->session_info->uid;
915 *session_gid = channel_info->session_info->gid;
916 end:
917 rcu_read_unlock();
918 return ret;
919 }
920
921 static
922 const char *get_condition_session_name(const struct lttng_condition *condition)
923 {
924 const char *session_name = NULL;
925 enum lttng_condition_status status;
926
927 switch (lttng_condition_get_type(condition)) {
928 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
929 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
930 status = lttng_condition_buffer_usage_get_session_name(
931 condition, &session_name);
932 break;
933 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
934 status = lttng_condition_session_consumed_size_get_session_name(
935 condition, &session_name);
936 break;
937 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
938 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
939 status = lttng_condition_session_rotation_get_session_name(
940 condition, &session_name);
941 break;
942 default:
943 abort();
944 }
945 if (status != LTTNG_CONDITION_STATUS_OK) {
946 ERR("Failed to retrieve session rotation condition's session name");
947 goto end;
948 }
949 end:
950 return session_name;
951 }
952
953 static
954 int evaluate_session_condition_for_client(
955 const struct lttng_condition *condition,
956 struct notification_thread_state *state,
957 struct lttng_evaluation **evaluation,
958 uid_t *session_uid, gid_t *session_gid)
959 {
960 int ret;
961 struct cds_lfht_iter iter;
962 struct cds_lfht_node *node;
963 const char *session_name;
964 struct session_info *session_info = NULL;
965
966 rcu_read_lock();
967 session_name = get_condition_session_name(condition);
968
969 /* Find the session associated with the trigger. */
970 cds_lfht_lookup(state->sessions_ht,
971 hash_key_str(session_name, lttng_ht_seed),
972 match_session,
973 session_name,
974 &iter);
975 node = cds_lfht_iter_get_node(&iter);
976 if (!node) {
977 DBG("No known session matching name \"%s\"",
978 session_name);
979 ret = 0;
980 goto end;
981 }
982
983 session_info = caa_container_of(node, struct session_info,
984 sessions_ht_node);
985 session_info_get(session_info);
986
987 /*
988 * Evaluation is performed in-line here since only one type of
989 * session-bound condition is handled for the moment.
990 */
991 switch (lttng_condition_get_type(condition)) {
992 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
993 if (!session_info->rotation.ongoing) {
994 ret = 0;
995 goto end_session_put;
996 }
997
998 *evaluation = lttng_evaluation_session_rotation_ongoing_create(
999 session_info->rotation.id);
1000 if (!*evaluation) {
1001 /* Fatal error. */
1002 ERR("Failed to create session rotation ongoing evaluation for session \"%s\"",
1003 session_info->name);
1004 ret = -1;
1005 goto end_session_put;
1006 }
1007 ret = 0;
1008 break;
1009 default:
1010 ret = 0;
1011 goto end_session_put;
1012 }
1013
1014 *session_uid = session_info->uid;
1015 *session_gid = session_info->gid;
1016
1017 end_session_put:
1018 session_info_put(session_info);
1019 end:
1020 rcu_read_unlock();
1021 return ret;
1022 }
1023
1024 static
1025 int evaluate_condition_for_client(const struct lttng_trigger *trigger,
1026 const struct lttng_condition *condition,
1027 struct notification_client *client,
1028 struct notification_thread_state *state)
1029 {
1030 int ret;
1031 struct lttng_evaluation *evaluation = NULL;
1032 struct notification_client_list client_list = {
1033 .lock = PTHREAD_MUTEX_INITIALIZER,
1034 .ref = {},
1035 .condition = NULL,
1036 .triggers_list = {},
1037 .clients_list = {},
1038 .notification_trigger_clients_ht = NULL,
1039 .notification_trigger_clients_ht_node = {},
1040 .rcu_node = {},
1041 };
1042 struct notification_client_list_element client_list_element = {};
1043 uid_t object_uid = 0;
1044 gid_t object_gid = 0;
1045
1046 LTTNG_ASSERT(trigger);
1047 LTTNG_ASSERT(condition);
1048 LTTNG_ASSERT(client);
1049 LTTNG_ASSERT(state);
1050
1051 switch (get_condition_binding_object(condition)) {
1052 case LTTNG_OBJECT_TYPE_SESSION:
1053 ret = evaluate_session_condition_for_client(condition, state,
1054 &evaluation, &object_uid, &object_gid);
1055 break;
1056 case LTTNG_OBJECT_TYPE_CHANNEL:
1057 ret = evaluate_channel_condition_for_client(condition, state,
1058 &evaluation, &object_uid, &object_gid);
1059 break;
1060 case LTTNG_OBJECT_TYPE_NONE:
1061 DBG("Newly subscribed-to condition not bound to object, nothing to evaluate");
1062 ret = 0;
1063 goto end;
1064 case LTTNG_OBJECT_TYPE_UNKNOWN:
1065 default:
1066 ret = -1;
1067 goto end;
1068 }
1069 if (ret) {
1070 /* Fatal error. */
1071 goto end;
1072 }
1073 if (!evaluation) {
1074 /* Evaluation yielded nothing. Normal exit. */
1075 DBG("Newly subscribed-to condition evaluated to false, nothing to report to client");
1076 ret = 0;
1077 goto end;
1078 }
1079
1080 /*
1081 * Create a temporary client list with the client currently
1082 * subscribing.
1083 */
1084 cds_lfht_node_init(&client_list.notification_trigger_clients_ht_node);
1085 CDS_INIT_LIST_HEAD(&client_list.clients_list);
1086
1087 CDS_INIT_LIST_HEAD(&client_list_element.node);
1088 client_list_element.client = client;
1089 cds_list_add(&client_list_element.node, &client_list.clients_list);
1090
1091 /* Send evaluation result to the newly-subscribed client. */
1092 DBG("Newly subscribed-to condition evaluated to true, notifying client");
1093 ret = send_evaluation_to_clients(trigger, evaluation, &client_list,
1094 state, object_uid, object_gid);
1095
1096 end:
1097 return ret;
1098 }
1099
1100 static
1101 int notification_thread_client_subscribe(struct notification_client *client,
1102 struct lttng_condition *condition,
1103 struct notification_thread_state *state,
1104 enum lttng_notification_channel_status *_status)
1105 {
1106 int ret = 0;
1107 struct notification_client_list *client_list = NULL;
1108 struct lttng_condition_list_element *condition_list_element = NULL;
1109 struct notification_client_list_element *client_list_element = NULL;
1110 struct lttng_trigger_ht_element *trigger_ht_element;
1111 enum lttng_notification_channel_status status =
1112 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1113
1114 /*
1115 * Ensure that the client has not already subscribed to this condition
1116 * before.
1117 */
1118 cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
1119 if (lttng_condition_is_equal(condition_list_element->condition,
1120 condition)) {
1121 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
1122 goto end;
1123 }
1124 }
1125
1126 condition_list_element = (lttng_condition_list_element *) zmalloc(sizeof(*condition_list_element));
1127 if (!condition_list_element) {
1128 ret = -1;
1129 goto error;
1130 }
1131 client_list_element = (notification_client_list_element *) zmalloc(sizeof(*client_list_element));
1132 if (!client_list_element) {
1133 ret = -1;
1134 goto error;
1135 }
1136
1137 /*
1138 * Add the newly-subscribed condition to the client's subscription list.
1139 */
1140 CDS_INIT_LIST_HEAD(&condition_list_element->node);
1141 condition_list_element->condition = condition;
1142 condition = NULL;
1143 cds_list_add(&condition_list_element->node, &client->condition_list);
1144
1145 client_list = get_client_list_from_condition(
1146 state, condition_list_element->condition);
1147 if (!client_list) {
1148 /*
1149 * No notification-emiting trigger registered with this
1150 * condition. We don't evaluate the condition right away
1151 * since this trigger is not registered yet.
1152 */
1153 free(client_list_element);
1154 goto end;
1155 }
1156
1157 /*
1158 * The condition to which the client just subscribed is evaluated
1159 * at this point so that conditions that are already TRUE result
1160 * in a notification being sent out.
1161 *
1162 * Note the iteration on all triggers which share an identical
1163 * `condition` than the one to which the client is registering. This is
1164 * done to ensure that the client receives a distinct notification for
1165 * all triggers that have a `notify` action that have this condition.
1166 */
1167 pthread_mutex_lock(&client_list->lock);
1168 cds_list_for_each_entry(trigger_ht_element,
1169 &client_list->triggers_list, client_list_trigger_node) {
1170 if (evaluate_condition_for_client(trigger_ht_element->trigger, condition_list_element->condition,
1171 client, state)) {
1172 WARN("Evaluation of a condition on client subscription failed, aborting.");
1173 ret = -1;
1174 free(client_list_element);
1175 pthread_mutex_unlock(&client_list->lock);
1176 goto end;
1177 }
1178 }
1179 pthread_mutex_unlock(&client_list->lock);
1180
1181 /*
1182 * Add the client to the list of clients interested in a given trigger
1183 * if a "notification" trigger with a corresponding condition was
1184 * added prior.
1185 */
1186 client_list_element->client = client;
1187 CDS_INIT_LIST_HEAD(&client_list_element->node);
1188
1189 pthread_mutex_lock(&client_list->lock);
1190 cds_list_add(&client_list_element->node, &client_list->clients_list);
1191 pthread_mutex_unlock(&client_list->lock);
1192 end:
1193 if (_status) {
1194 *_status = status;
1195 }
1196 if (client_list) {
1197 notification_client_list_put(client_list);
1198 }
1199 lttng_condition_destroy(condition);
1200 return ret;
1201 error:
1202 free(condition_list_element);
1203 free(client_list_element);
1204 lttng_condition_destroy(condition);
1205 return ret;
1206 }
1207
1208 static
1209 int notification_thread_client_unsubscribe(
1210 struct notification_client *client,
1211 struct lttng_condition *condition,
1212 struct notification_thread_state *state,
1213 enum lttng_notification_channel_status *_status)
1214 {
1215 struct notification_client_list *client_list;
1216 struct lttng_condition_list_element *condition_list_element,
1217 *condition_tmp;
1218 struct notification_client_list_element *client_list_element,
1219 *client_tmp;
1220 bool condition_found = false;
1221 enum lttng_notification_channel_status status =
1222 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1223
1224 /* Remove the condition from the client's condition list. */
1225 cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
1226 &client->condition_list, node) {
1227 if (!lttng_condition_is_equal(condition_list_element->condition,
1228 condition)) {
1229 continue;
1230 }
1231
1232 cds_list_del(&condition_list_element->node);
1233 /*
1234 * The caller may be iterating on the client's conditions to
1235 * tear down a client's connection. In this case, the condition
1236 * will be destroyed at the end.
1237 */
1238 if (condition != condition_list_element->condition) {
1239 lttng_condition_destroy(
1240 condition_list_element->condition);
1241 }
1242 free(condition_list_element);
1243 condition_found = true;
1244 break;
1245 }
1246
1247 if (!condition_found) {
1248 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
1249 goto end;
1250 }
1251
1252 /*
1253 * Remove the client from the list of clients interested the trigger
1254 * matching the condition.
1255 */
1256 client_list = get_client_list_from_condition(state, condition);
1257 if (!client_list) {
1258 goto end;
1259 }
1260
1261 pthread_mutex_lock(&client_list->lock);
1262 cds_list_for_each_entry_safe(client_list_element, client_tmp,
1263 &client_list->clients_list, node) {
1264 if (client_list_element->client->id != client->id) {
1265 continue;
1266 }
1267 cds_list_del(&client_list_element->node);
1268 free(client_list_element);
1269 break;
1270 }
1271 pthread_mutex_unlock(&client_list->lock);
1272 notification_client_list_put(client_list);
1273 client_list = NULL;
1274 end:
1275 lttng_condition_destroy(condition);
1276 if (_status) {
1277 *_status = status;
1278 }
1279 return 0;
1280 }
1281
1282 static
1283 void free_notification_client_rcu(struct rcu_head *node)
1284 {
1285 free(caa_container_of(node, struct notification_client, rcu_node));
1286 }
1287
1288 static
1289 void notification_client_destroy(struct notification_client *client)
1290 {
1291 if (!client) {
1292 return;
1293 }
1294
1295 /*
1296 * The client object is not reachable by other threads, no need to lock
1297 * the client here.
1298 */
1299 if (client->socket >= 0) {
1300 (void) lttcomm_close_unix_sock(client->socket);
1301 client->socket = -1;
1302 }
1303 client->communication.active = false;
1304 lttng_payload_reset(&client->communication.inbound.payload);
1305 lttng_payload_reset(&client->communication.outbound.payload);
1306 pthread_mutex_destroy(&client->lock);
1307 call_rcu(&client->rcu_node, free_notification_client_rcu);
1308 }
1309
1310 /*
1311 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1312 * client pointer).
1313 */
1314 static
1315 struct notification_client *get_client_from_socket(int socket,
1316 struct notification_thread_state *state)
1317 {
1318 struct cds_lfht_iter iter;
1319 struct cds_lfht_node *node;
1320 struct notification_client *client = NULL;
1321
1322 ASSERT_RCU_READ_LOCKED();
1323
1324 cds_lfht_lookup(state->client_socket_ht,
1325 hash_client_socket(socket),
1326 match_client_socket,
1327 (void *) (unsigned long) socket,
1328 &iter);
1329 node = cds_lfht_iter_get_node(&iter);
1330 if (!node) {
1331 goto end;
1332 }
1333
1334 client = caa_container_of(node, struct notification_client,
1335 client_socket_ht_node);
1336 end:
1337 return client;
1338 }
1339
1340 /*
1341 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1342 * client pointer).
1343 */
1344 static
1345 struct notification_client *get_client_from_id(notification_client_id id,
1346 struct notification_thread_state *state)
1347 {
1348 struct cds_lfht_iter iter;
1349 struct cds_lfht_node *node;
1350 struct notification_client *client = NULL;
1351
1352 ASSERT_RCU_READ_LOCKED();
1353
1354 cds_lfht_lookup(state->client_id_ht,
1355 hash_client_id(id),
1356 match_client_id,
1357 &id,
1358 &iter);
1359 node = cds_lfht_iter_get_node(&iter);
1360 if (!node) {
1361 goto end;
1362 }
1363
1364 client = caa_container_of(node, struct notification_client,
1365 client_id_ht_node);
1366 end:
1367 return client;
1368 }
1369
1370 static
1371 bool buffer_usage_condition_applies_to_channel(
1372 const struct lttng_condition *condition,
1373 const struct channel_info *channel_info)
1374 {
1375 enum lttng_condition_status status;
1376 enum lttng_domain_type condition_domain;
1377 const char *condition_session_name = NULL;
1378 const char *condition_channel_name = NULL;
1379
1380 status = lttng_condition_buffer_usage_get_domain_type(condition,
1381 &condition_domain);
1382 LTTNG_ASSERT(status == LTTNG_CONDITION_STATUS_OK);
1383 if (channel_info->key.domain != condition_domain) {
1384 goto fail;
1385 }
1386
1387 status = lttng_condition_buffer_usage_get_session_name(
1388 condition, &condition_session_name);
1389 LTTNG_ASSERT((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1390
1391 status = lttng_condition_buffer_usage_get_channel_name(
1392 condition, &condition_channel_name);
1393 LTTNG_ASSERT((status == LTTNG_CONDITION_STATUS_OK) && condition_channel_name);
1394
1395 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1396 goto fail;
1397 }
1398 if (strcmp(channel_info->name, condition_channel_name)) {
1399 goto fail;
1400 }
1401
1402 return true;
1403 fail:
1404 return false;
1405 }
1406
1407 static
1408 bool session_consumed_size_condition_applies_to_channel(
1409 const struct lttng_condition *condition,
1410 const struct channel_info *channel_info)
1411 {
1412 enum lttng_condition_status status;
1413 const char *condition_session_name = NULL;
1414
1415 status = lttng_condition_session_consumed_size_get_session_name(
1416 condition, &condition_session_name);
1417 LTTNG_ASSERT((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1418
1419 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1420 goto fail;
1421 }
1422
1423 return true;
1424 fail:
1425 return false;
1426 }
1427
1428 static
1429 bool trigger_applies_to_channel(const struct lttng_trigger *trigger,
1430 const struct channel_info *channel_info)
1431 {
1432 const struct lttng_condition *condition;
1433 bool trigger_applies;
1434
1435 condition = lttng_trigger_get_const_condition(trigger);
1436 if (!condition) {
1437 goto fail;
1438 }
1439
1440 switch (lttng_condition_get_type(condition)) {
1441 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1442 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1443 trigger_applies = buffer_usage_condition_applies_to_channel(
1444 condition, channel_info);
1445 break;
1446 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
1447 trigger_applies = session_consumed_size_condition_applies_to_channel(
1448 condition, channel_info);
1449 break;
1450 default:
1451 goto fail;
1452 }
1453
1454 return trigger_applies;
1455 fail:
1456 return false;
1457 }
1458
1459 /* Must be called with RCU read lock held. */
1460 static
1461 struct lttng_session_trigger_list *get_session_trigger_list(
1462 struct notification_thread_state *state,
1463 const char *session_name)
1464 {
1465 struct lttng_session_trigger_list *list = NULL;
1466 struct cds_lfht_node *node;
1467 struct cds_lfht_iter iter;
1468
1469 ASSERT_RCU_READ_LOCKED();
1470
1471 cds_lfht_lookup(state->session_triggers_ht,
1472 hash_key_str(session_name, lttng_ht_seed),
1473 match_session_trigger_list,
1474 session_name,
1475 &iter);
1476 node = cds_lfht_iter_get_node(&iter);
1477 if (!node) {
1478 /*
1479 * Not an error, the list of triggers applying to that session
1480 * will be initialized when the session is created.
1481 */
1482 DBG("No trigger list found for session \"%s\" as it is not yet known to the notification system",
1483 session_name);
1484 goto end;
1485 }
1486
1487 list = caa_container_of(node,
1488 struct lttng_session_trigger_list,
1489 session_triggers_ht_node);
1490 end:
1491 return list;
1492 }
1493
1494 /*
1495 * Allocate an empty lttng_session_trigger_list for the session named
1496 * 'session_name'.
1497 *
1498 * No ownership of 'session_name' is assumed by the session trigger list.
1499 * It is the caller's responsability to ensure the session name is alive
1500 * for as long as this list is.
1501 */
1502 static
1503 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
1504 const char *session_name,
1505 struct cds_lfht *session_triggers_ht)
1506 {
1507 struct lttng_session_trigger_list *list;
1508
1509 list = (lttng_session_trigger_list *) zmalloc(sizeof(*list));
1510 if (!list) {
1511 goto end;
1512 }
1513 list->session_name = session_name;
1514 CDS_INIT_LIST_HEAD(&list->list);
1515 cds_lfht_node_init(&list->session_triggers_ht_node);
1516 list->session_triggers_ht = session_triggers_ht;
1517
1518 rcu_read_lock();
1519 /* Publish the list through the session_triggers_ht. */
1520 cds_lfht_add(session_triggers_ht,
1521 hash_key_str(session_name, lttng_ht_seed),
1522 &list->session_triggers_ht_node);
1523 rcu_read_unlock();
1524 end:
1525 return list;
1526 }
1527
1528 static
1529 void free_session_trigger_list_rcu(struct rcu_head *node)
1530 {
1531 free(caa_container_of(node, struct lttng_session_trigger_list,
1532 rcu_node));
1533 }
1534
1535 static
1536 void lttng_session_trigger_list_destroy(struct lttng_session_trigger_list *list)
1537 {
1538 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1539
1540 /* Empty the list element by element, and then free the list itself. */
1541 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1542 &list->list, node) {
1543 cds_list_del(&trigger_list_element->node);
1544 free(trigger_list_element);
1545 }
1546 rcu_read_lock();
1547 /* Unpublish the list from the session_triggers_ht. */
1548 cds_lfht_del(list->session_triggers_ht,
1549 &list->session_triggers_ht_node);
1550 rcu_read_unlock();
1551 call_rcu(&list->rcu_node, free_session_trigger_list_rcu);
1552 }
1553
1554 static
1555 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
1556 struct lttng_trigger *trigger)
1557 {
1558 int ret = 0;
1559 struct lttng_trigger_list_element *new_element =
1560 (lttng_trigger_list_element *) zmalloc(sizeof(*new_element));
1561
1562 if (!new_element) {
1563 ret = -1;
1564 goto end;
1565 }
1566 CDS_INIT_LIST_HEAD(&new_element->node);
1567 new_element->trigger = trigger;
1568 cds_list_add(&new_element->node, &list->list);
1569 end:
1570 return ret;
1571 }
1572
1573 static
1574 bool trigger_applies_to_session(const struct lttng_trigger *trigger,
1575 const char *session_name)
1576 {
1577 bool applies = false;
1578 const struct lttng_condition *condition;
1579
1580 condition = lttng_trigger_get_const_condition(trigger);
1581 switch (lttng_condition_get_type(condition)) {
1582 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
1583 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
1584 {
1585 enum lttng_condition_status condition_status;
1586 const char *condition_session_name;
1587
1588 condition_status = lttng_condition_session_rotation_get_session_name(
1589 condition, &condition_session_name);
1590 if (condition_status != LTTNG_CONDITION_STATUS_OK) {
1591 ERR("Failed to retrieve session rotation condition's session name");
1592 goto end;
1593 }
1594
1595 LTTNG_ASSERT(condition_session_name);
1596 applies = !strcmp(condition_session_name, session_name);
1597 break;
1598 }
1599 default:
1600 goto end;
1601 }
1602 end:
1603 return applies;
1604 }
1605
1606 /*
1607 * Allocate and initialize an lttng_session_trigger_list which contains
1608 * all triggers that apply to the session named 'session_name'.
1609 *
1610 * No ownership of 'session_name' is assumed by the session trigger list.
1611 * It is the caller's responsability to ensure the session name is alive
1612 * for as long as this list is.
1613 */
1614 static
1615 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
1616 const struct notification_thread_state *state,
1617 const char *session_name)
1618 {
1619 int trigger_count = 0;
1620 struct lttng_session_trigger_list *session_trigger_list = NULL;
1621 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1622 struct cds_lfht_iter iter;
1623
1624 session_trigger_list = lttng_session_trigger_list_create(session_name,
1625 state->session_triggers_ht);
1626
1627 /* Add all triggers applying to the session named 'session_name'. */
1628 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1629 node) {
1630 int ret;
1631
1632 if (!trigger_applies_to_session(trigger_ht_element->trigger,
1633 session_name)) {
1634 continue;
1635 }
1636
1637 ret = lttng_session_trigger_list_add(session_trigger_list,
1638 trigger_ht_element->trigger);
1639 if (ret) {
1640 goto error;
1641 }
1642
1643 trigger_count++;
1644 }
1645
1646 DBG("Found %i triggers that apply to newly created session",
1647 trigger_count);
1648 return session_trigger_list;
1649 error:
1650 lttng_session_trigger_list_destroy(session_trigger_list);
1651 return NULL;
1652 }
1653
1654 static
1655 struct session_info *find_or_create_session_info(
1656 struct notification_thread_state *state,
1657 const char *name, uid_t uid, gid_t gid)
1658 {
1659 struct session_info *session = NULL;
1660 struct cds_lfht_node *node;
1661 struct cds_lfht_iter iter;
1662 struct lttng_session_trigger_list *trigger_list;
1663
1664 rcu_read_lock();
1665 cds_lfht_lookup(state->sessions_ht,
1666 hash_key_str(name, lttng_ht_seed),
1667 match_session,
1668 name,
1669 &iter);
1670 node = cds_lfht_iter_get_node(&iter);
1671 if (node) {
1672 DBG("Found session info of session \"%s\" (uid = %i, gid = %i)",
1673 name, uid, gid);
1674 session = caa_container_of(node, struct session_info,
1675 sessions_ht_node);
1676 LTTNG_ASSERT(session->uid == uid);
1677 LTTNG_ASSERT(session->gid == gid);
1678 session_info_get(session);
1679 goto end;
1680 }
1681
1682 trigger_list = lttng_session_trigger_list_build(state, name);
1683 if (!trigger_list) {
1684 goto error;
1685 }
1686
1687 session = session_info_create(name, uid, gid, trigger_list,
1688 state->sessions_ht);
1689 if (!session) {
1690 ERR("Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1691 name, uid, gid);
1692 lttng_session_trigger_list_destroy(trigger_list);
1693 goto error;
1694 }
1695 trigger_list = NULL;
1696
1697 cds_lfht_add(state->sessions_ht, hash_key_str(name, lttng_ht_seed),
1698 &session->sessions_ht_node);
1699 end:
1700 rcu_read_unlock();
1701 return session;
1702 error:
1703 rcu_read_unlock();
1704 session_info_put(session);
1705 return NULL;
1706 }
1707
1708 static
1709 int handle_notification_thread_command_add_channel(
1710 struct notification_thread_state *state,
1711 const char *session_name, uid_t session_uid, gid_t session_gid,
1712 const char *channel_name, enum lttng_domain_type channel_domain,
1713 uint64_t channel_key_int, uint64_t channel_capacity,
1714 enum lttng_error_code *cmd_result)
1715 {
1716 struct cds_list_head trigger_list;
1717 struct channel_info *new_channel_info = NULL;
1718 struct channel_key channel_key = {
1719 .key = channel_key_int,
1720 .domain = channel_domain,
1721 };
1722 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
1723 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1724 int trigger_count = 0;
1725 struct cds_lfht_iter iter;
1726 struct session_info *session_info = NULL;
1727
1728 DBG("Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
1729 channel_name, session_name, channel_key_int,
1730 lttng_domain_type_str(channel_domain));
1731
1732 CDS_INIT_LIST_HEAD(&trigger_list);
1733
1734 session_info = find_or_create_session_info(state, session_name,
1735 session_uid, session_gid);
1736 if (!session_info) {
1737 /* Allocation error or an internal error occurred. */
1738 goto error;
1739 }
1740
1741 new_channel_info = channel_info_create(channel_name, &channel_key,
1742 channel_capacity, session_info);
1743 if (!new_channel_info) {
1744 goto error;
1745 }
1746
1747 rcu_read_lock();
1748 /* Build a list of all triggers applying to the new channel. */
1749 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1750 node) {
1751 struct lttng_trigger_list_element *new_element;
1752
1753 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
1754 new_channel_info)) {
1755 continue;
1756 }
1757
1758 new_element = (lttng_trigger_list_element *) zmalloc(sizeof(*new_element));
1759 if (!new_element) {
1760 rcu_read_unlock();
1761 goto error;
1762 }
1763 CDS_INIT_LIST_HEAD(&new_element->node);
1764 new_element->trigger = trigger_ht_element->trigger;
1765 cds_list_add(&new_element->node, &trigger_list);
1766 trigger_count++;
1767 }
1768 rcu_read_unlock();
1769
1770 DBG("Found %i triggers that apply to newly added channel",
1771 trigger_count);
1772 channel_trigger_list = (lttng_channel_trigger_list *) zmalloc(sizeof(*channel_trigger_list));
1773 if (!channel_trigger_list) {
1774 goto error;
1775 }
1776 channel_trigger_list->channel_key = new_channel_info->key;
1777 CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
1778 cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
1779 cds_list_splice(&trigger_list, &channel_trigger_list->list);
1780
1781 rcu_read_lock();
1782 /* Add channel to the channel_ht which owns the channel_infos. */
1783 cds_lfht_add(state->channels_ht,
1784 hash_channel_key(&new_channel_info->key),
1785 &new_channel_info->channels_ht_node);
1786 /*
1787 * Add the list of triggers associated with this channel to the
1788 * channel_triggers_ht.
1789 */
1790 cds_lfht_add(state->channel_triggers_ht,
1791 hash_channel_key(&new_channel_info->key),
1792 &channel_trigger_list->channel_triggers_ht_node);
1793 rcu_read_unlock();
1794 session_info_put(session_info);
1795 *cmd_result = LTTNG_OK;
1796 return 0;
1797 error:
1798 channel_info_destroy(new_channel_info);
1799 session_info_put(session_info);
1800 return 1;
1801 }
1802
1803 static
1804 void free_channel_trigger_list_rcu(struct rcu_head *node)
1805 {
1806 free(caa_container_of(node, struct lttng_channel_trigger_list,
1807 rcu_node));
1808 }
1809
1810 static
1811 void free_channel_state_sample_rcu(struct rcu_head *node)
1812 {
1813 free(caa_container_of(node, struct channel_state_sample,
1814 rcu_node));
1815 }
1816
1817 static
1818 int handle_notification_thread_command_remove_channel(
1819 struct notification_thread_state *state,
1820 uint64_t channel_key, enum lttng_domain_type domain,
1821 enum lttng_error_code *cmd_result)
1822 {
1823 struct cds_lfht_node *node;
1824 struct cds_lfht_iter iter;
1825 struct lttng_channel_trigger_list *trigger_list;
1826 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1827 struct channel_key key = { .key = channel_key, .domain = domain };
1828 struct channel_info *channel_info;
1829
1830 DBG("Removing channel key = %" PRIu64 " in %s domain",
1831 channel_key, lttng_domain_type_str(domain));
1832
1833 rcu_read_lock();
1834
1835 cds_lfht_lookup(state->channel_triggers_ht,
1836 hash_channel_key(&key),
1837 match_channel_trigger_list,
1838 &key,
1839 &iter);
1840 node = cds_lfht_iter_get_node(&iter);
1841 /*
1842 * There is a severe internal error if we are being asked to remove a
1843 * channel that doesn't exist.
1844 */
1845 if (!node) {
1846 ERR("Channel being removed is unknown to the notification thread");
1847 goto end;
1848 }
1849
1850 /* Free the list of triggers associated with this channel. */
1851 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
1852 channel_triggers_ht_node);
1853 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1854 &trigger_list->list, node) {
1855 cds_list_del(&trigger_list_element->node);
1856 free(trigger_list_element);
1857 }
1858 cds_lfht_del(state->channel_triggers_ht, node);
1859 call_rcu(&trigger_list->rcu_node, free_channel_trigger_list_rcu);
1860
1861 /* Free sampled channel state. */
1862 cds_lfht_lookup(state->channel_state_ht,
1863 hash_channel_key(&key),
1864 match_channel_state_sample,
1865 &key,
1866 &iter);
1867 node = cds_lfht_iter_get_node(&iter);
1868 /*
1869 * This is expected to be NULL if the channel is destroyed before we
1870 * received a sample.
1871 */
1872 if (node) {
1873 struct channel_state_sample *sample = caa_container_of(node,
1874 struct channel_state_sample,
1875 channel_state_ht_node);
1876
1877 cds_lfht_del(state->channel_state_ht, node);
1878 call_rcu(&sample->rcu_node, free_channel_state_sample_rcu);
1879 }
1880
1881 /* Remove the channel from the channels_ht and free it. */
1882 cds_lfht_lookup(state->channels_ht,
1883 hash_channel_key(&key),
1884 match_channel_info,
1885 &key,
1886 &iter);
1887 node = cds_lfht_iter_get_node(&iter);
1888 LTTNG_ASSERT(node);
1889 channel_info = caa_container_of(node, struct channel_info,
1890 channels_ht_node);
1891 cds_lfht_del(state->channels_ht, node);
1892 channel_info_destroy(channel_info);
1893 end:
1894 rcu_read_unlock();
1895 *cmd_result = LTTNG_OK;
1896 return 0;
1897 }
1898
1899 static
1900 int handle_notification_thread_command_session_rotation(
1901 struct notification_thread_state *state,
1902 enum notification_thread_command_type cmd_type,
1903 const char *session_name, uid_t session_uid, gid_t session_gid,
1904 uint64_t trace_archive_chunk_id,
1905 struct lttng_trace_archive_location *location,
1906 enum lttng_error_code *_cmd_result)
1907 {
1908 int ret = 0;
1909 enum lttng_error_code cmd_result = LTTNG_OK;
1910 struct lttng_session_trigger_list *trigger_list;
1911 struct lttng_trigger_list_element *trigger_list_element;
1912 struct session_info *session_info;
1913 const struct lttng_credentials session_creds = {
1914 .uid = LTTNG_OPTIONAL_INIT_VALUE(session_uid),
1915 .gid = LTTNG_OPTIONAL_INIT_VALUE(session_gid),
1916 };
1917
1918 rcu_read_lock();
1919
1920 session_info = find_or_create_session_info(state, session_name,
1921 session_uid, session_gid);
1922 if (!session_info) {
1923 /* Allocation error or an internal error occurred. */
1924 ret = -1;
1925 cmd_result = LTTNG_ERR_NOMEM;
1926 goto end;
1927 }
1928
1929 session_info->rotation.ongoing =
1930 cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING;
1931 session_info->rotation.id = trace_archive_chunk_id;
1932 trigger_list = get_session_trigger_list(state, session_name);
1933 if (!trigger_list) {
1934 DBG("No triggers applying to session \"%s\" found",
1935 session_name);
1936 goto end;
1937 }
1938
1939 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
1940 node) {
1941 const struct lttng_condition *condition;
1942 struct lttng_trigger *trigger;
1943 struct notification_client_list *client_list;
1944 struct lttng_evaluation *evaluation = NULL;
1945 enum lttng_condition_type condition_type;
1946 enum action_executor_status executor_status;
1947
1948 trigger = trigger_list_element->trigger;
1949 condition = lttng_trigger_get_const_condition(trigger);
1950 LTTNG_ASSERT(condition);
1951 condition_type = lttng_condition_get_type(condition);
1952
1953 if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING &&
1954 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1955 continue;
1956 } else if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED &&
1957 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED) {
1958 continue;
1959 }
1960
1961 client_list = get_client_list_from_condition(state, condition);
1962 if (cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1963 evaluation = lttng_evaluation_session_rotation_ongoing_create(
1964 trace_archive_chunk_id);
1965 } else {
1966 evaluation = lttng_evaluation_session_rotation_completed_create(
1967 trace_archive_chunk_id, location);
1968 }
1969
1970 if (!evaluation) {
1971 /* Internal error */
1972 ret = -1;
1973 cmd_result = LTTNG_ERR_UNK;
1974 goto put_list;
1975 }
1976
1977 /*
1978 * Ownership of `evaluation` transferred to the action executor
1979 * no matter the result.
1980 */
1981 executor_status = action_executor_enqueue_trigger(
1982 state->executor, trigger, evaluation,
1983 &session_creds, client_list);
1984 evaluation = NULL;
1985 switch (executor_status) {
1986 case ACTION_EXECUTOR_STATUS_OK:
1987 break;
1988 case ACTION_EXECUTOR_STATUS_ERROR:
1989 case ACTION_EXECUTOR_STATUS_INVALID:
1990 /*
1991 * TODO Add trigger identification (name/id) when
1992 * it is added to the API.
1993 */
1994 ERR("Fatal error occurred while enqueuing action associated with session rotation trigger");
1995 ret = -1;
1996 goto put_list;
1997 case ACTION_EXECUTOR_STATUS_OVERFLOW:
1998 /*
1999 * TODO Add trigger identification (name/id) when
2000 * it is added to the API.
2001 *
2002 * Not a fatal error.
2003 */
2004 WARN("No space left when enqueuing action associated with session rotation trigger");
2005 ret = 0;
2006 goto put_list;
2007 default:
2008 abort();
2009 }
2010
2011 put_list:
2012 notification_client_list_put(client_list);
2013 if (caa_unlikely(ret)) {
2014 break;
2015 }
2016 }
2017 end:
2018 session_info_put(session_info);
2019 *_cmd_result = cmd_result;
2020 rcu_read_unlock();
2021 return ret;
2022 }
2023
2024 static
2025 int handle_notification_thread_command_add_tracer_event_source(
2026 struct notification_thread_state *state,
2027 int tracer_event_source_fd,
2028 enum lttng_domain_type domain_type,
2029 enum lttng_error_code *_cmd_result)
2030 {
2031 int ret = 0;
2032 enum lttng_error_code cmd_result = LTTNG_OK;
2033 struct notification_event_tracer_event_source_element *element = NULL;
2034
2035 element = (notification_event_tracer_event_source_element *) zmalloc(sizeof(*element));
2036 if (!element) {
2037 cmd_result = LTTNG_ERR_NOMEM;
2038 ret = -1;
2039 goto end;
2040 }
2041
2042 element->fd = tracer_event_source_fd;
2043 element->domain = domain_type;
2044
2045 cds_list_add(&element->node, &state->tracer_event_sources_list);
2046
2047 DBG3("Adding tracer event source fd to poll set: tracer_event_source_fd = %d, domain = '%s'",
2048 tracer_event_source_fd,
2049 lttng_domain_type_str(domain_type));
2050
2051 /* Adding the read side pipe to the event poll. */
2052 ret = lttng_poll_add(&state->events, tracer_event_source_fd, LPOLLIN | LPOLLERR);
2053 if (ret < 0) {
2054 ERR("Failed to add tracer event source to poll set: tracer_event_source_fd = %d, domain = '%s'",
2055 tracer_event_source_fd,
2056 lttng_domain_type_str(element->domain));
2057 cds_list_del(&element->node);
2058 free(element);
2059 goto end;
2060 }
2061
2062 element->is_fd_in_poll_set = true;
2063
2064 end:
2065 *_cmd_result = cmd_result;
2066 return ret;
2067 }
2068
2069 static
2070 int drain_event_notifier_notification_pipe(
2071 struct notification_thread_state *state,
2072 int pipe, enum lttng_domain_type domain)
2073 {
2074 struct lttng_poll_event events = {};
2075 int ret;
2076
2077 ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
2078 if (ret < 0) {
2079 ERR("Error creating lttng_poll_event");
2080 goto end;
2081 }
2082
2083 ret = lttng_poll_add(&events, pipe, LPOLLIN);
2084 if (ret < 0) {
2085 ERR("Error adding fd event notifier notification pipe to lttng_poll_event: fd = %d",
2086 pipe);
2087 goto end;
2088 }
2089
2090 while (true) {
2091 /*
2092 * Continue to consume notifications as long as there are new
2093 * ones coming in. The tracer has been asked to stop producing
2094 * them.
2095 *
2096 * LPOLLIN is explicitly checked since LPOLLHUP is implicitly
2097 * monitored (on Linux, at least) and will be returned when
2098 * the pipe is closed but empty.
2099 */
2100 ret = lttng_poll_wait_interruptible(&events, 0);
2101 if (ret == 0 || (LTTNG_POLL_GETEV(&events, 0) & LPOLLIN) == 0) {
2102 /* No more notification to be read on this pipe. */
2103 ret = 0;
2104 goto end;
2105 } else if (ret < 0) {
2106 PERROR("Failed on lttng_poll_wait_interruptible() call");
2107 ret = -1;
2108 goto end;
2109 }
2110
2111 ret = handle_one_event_notifier_notification(state, pipe, domain);
2112 if (ret) {
2113 ERR("Error consuming an event notifier notification from pipe: fd = %d",
2114 pipe);
2115 }
2116 }
2117 end:
2118 lttng_poll_clean(&events);
2119 return ret;
2120 }
2121
2122 static
2123 struct notification_event_tracer_event_source_element *
2124 find_tracer_event_source_element(struct notification_thread_state *state,
2125 int tracer_event_source_fd)
2126 {
2127 struct notification_event_tracer_event_source_element *source_element;
2128
2129 cds_list_for_each_entry(source_element,
2130 &state->tracer_event_sources_list, node) {
2131 if (source_element->fd == tracer_event_source_fd) {
2132 goto end;
2133 }
2134 }
2135
2136 source_element = NULL;
2137 end:
2138 return source_element;
2139 }
2140
2141 static
2142 int remove_tracer_event_source_from_pollset(
2143 struct notification_thread_state *state,
2144 struct notification_event_tracer_event_source_element *source_element)
2145 {
2146 int ret = 0;
2147
2148 LTTNG_ASSERT(source_element->is_fd_in_poll_set);
2149
2150 DBG3("Removing tracer event source from poll set: tracer_event_source_fd = %d, domain = '%s'",
2151 source_element->fd,
2152 lttng_domain_type_str(source_element->domain));
2153
2154 /* Removing the fd from the event poll set. */
2155 ret = lttng_poll_del(&state->events, source_element->fd);
2156 if (ret < 0) {
2157 ERR("Failed to remove tracer event source from poll set: tracer_event_source_fd = %d, domain = '%s'",
2158 source_element->fd,
2159 lttng_domain_type_str(source_element->domain));
2160 ret = -1;
2161 goto end;
2162 }
2163
2164 source_element->is_fd_in_poll_set = false;
2165
2166 /*
2167 * Force the notification thread to restart the poll() loop to ensure
2168 * that any events from the removed fd are removed.
2169 */
2170 state->restart_poll = true;
2171
2172 ret = drain_event_notifier_notification_pipe(state, source_element->fd,
2173 source_element->domain);
2174 if (ret) {
2175 ERR("Error draining event notifier notification: tracer_event_source_fd = %d, domain = %s",
2176 source_element->fd,
2177 lttng_domain_type_str(source_element->domain));
2178 ret = -1;
2179 goto end;
2180 }
2181
2182 end:
2183 return ret;
2184 }
2185
2186 int handle_notification_thread_tracer_event_source_died(
2187 struct notification_thread_state *state,
2188 int tracer_event_source_fd)
2189 {
2190 int ret = 0;
2191 struct notification_event_tracer_event_source_element *source_element;
2192
2193 source_element = find_tracer_event_source_element(state,
2194 tracer_event_source_fd);
2195
2196 LTTNG_ASSERT(source_element);
2197
2198 ret = remove_tracer_event_source_from_pollset(state, source_element);
2199 if (ret) {
2200 ERR("Failed to remove dead tracer event source from poll set");
2201 }
2202
2203 return ret;
2204 }
2205
2206 static
2207 int handle_notification_thread_command_remove_tracer_event_source(
2208 struct notification_thread_state *state,
2209 int tracer_event_source_fd,
2210 enum lttng_error_code *_cmd_result)
2211 {
2212 int ret = 0;
2213 enum lttng_error_code cmd_result = LTTNG_OK;
2214 struct notification_event_tracer_event_source_element *source_element = NULL;
2215
2216 source_element = find_tracer_event_source_element(state,
2217 tracer_event_source_fd);
2218
2219 LTTNG_ASSERT(source_element);
2220
2221 /* Remove the tracer source from the list. */
2222 cds_list_del(&source_element->node);
2223
2224 if (!source_element->is_fd_in_poll_set) {
2225 /* Skip the poll set removal. */
2226 goto end;
2227 }
2228
2229 ret = remove_tracer_event_source_from_pollset(state, source_element);
2230 if (ret) {
2231 ERR("Failed to remove tracer event source from poll set");
2232 cmd_result = LTTNG_ERR_FATAL;
2233 }
2234
2235 end:
2236 free(source_element);
2237 *_cmd_result = cmd_result;
2238 return ret;
2239 }
2240
2241 static int handle_notification_thread_command_list_triggers(
2242 struct notification_thread_handle *handle __attribute__((unused)),
2243 struct notification_thread_state *state,
2244 uid_t client_uid,
2245 struct lttng_triggers **triggers,
2246 enum lttng_error_code *_cmd_result)
2247 {
2248 int ret = 0;
2249 enum lttng_error_code cmd_result = LTTNG_OK;
2250 struct cds_lfht_iter iter;
2251 struct lttng_trigger_ht_element *trigger_ht_element;
2252 struct lttng_triggers *local_triggers = NULL;
2253 const struct lttng_credentials *creds;
2254
2255 rcu_read_lock();
2256
2257 local_triggers = lttng_triggers_create();
2258 if (!local_triggers) {
2259 /* Not a fatal error. */
2260 cmd_result = LTTNG_ERR_NOMEM;
2261 goto end;
2262 }
2263
2264 cds_lfht_for_each_entry(state->triggers_ht, &iter,
2265 trigger_ht_element, node) {
2266 /*
2267 * Only return the triggers to which the client has access.
2268 * The root user has visibility over all triggers.
2269 */
2270 creds = lttng_trigger_get_credentials(trigger_ht_element->trigger);
2271 if (client_uid != lttng_credentials_get_uid(creds) && client_uid != 0) {
2272 continue;
2273 }
2274
2275 ret = lttng_triggers_add(local_triggers,
2276 trigger_ht_element->trigger);
2277 if (ret < 0) {
2278 /* Not a fatal error. */
2279 ret = 0;
2280 cmd_result = LTTNG_ERR_NOMEM;
2281 goto end;
2282 }
2283 }
2284
2285 /* Transferring ownership to the caller. */
2286 *triggers = local_triggers;
2287 local_triggers = NULL;
2288
2289 end:
2290 rcu_read_unlock();
2291 lttng_triggers_destroy(local_triggers);
2292 *_cmd_result = cmd_result;
2293 return ret;
2294 }
2295
2296 static inline void get_trigger_info_for_log(const struct lttng_trigger *trigger,
2297 const char **trigger_name,
2298 uid_t *trigger_owner_uid)
2299 {
2300 enum lttng_trigger_status trigger_status;
2301
2302 trigger_status = lttng_trigger_get_name(trigger, trigger_name);
2303 switch (trigger_status) {
2304 case LTTNG_TRIGGER_STATUS_OK:
2305 break;
2306 case LTTNG_TRIGGER_STATUS_UNSET:
2307 *trigger_name = "(anonymous)";
2308 break;
2309 default:
2310 abort();
2311 }
2312
2313 trigger_status = lttng_trigger_get_owner_uid(trigger,
2314 trigger_owner_uid);
2315 LTTNG_ASSERT(trigger_status == LTTNG_TRIGGER_STATUS_OK);
2316 }
2317
2318 static int handle_notification_thread_command_get_trigger(
2319 struct notification_thread_state *state,
2320 const struct lttng_trigger *trigger,
2321 struct lttng_trigger **registered_trigger,
2322 enum lttng_error_code *_cmd_result)
2323 {
2324 int ret = -1;
2325 struct cds_lfht_iter iter;
2326 struct lttng_trigger_ht_element *trigger_ht_element;
2327 enum lttng_error_code cmd_result = LTTNG_ERR_TRIGGER_NOT_FOUND;
2328 const char *trigger_name;
2329 uid_t trigger_owner_uid;
2330
2331 rcu_read_lock();
2332
2333 cds_lfht_for_each_entry(
2334 state->triggers_ht, &iter, trigger_ht_element, node) {
2335 if (lttng_trigger_is_equal(
2336 trigger, trigger_ht_element->trigger)) {
2337 /* Take one reference on the return trigger. */
2338 *registered_trigger = trigger_ht_element->trigger;
2339 lttng_trigger_get(*registered_trigger);
2340 ret = 0;
2341 cmd_result = LTTNG_OK;
2342 goto end;
2343 }
2344 }
2345
2346 /* Not a fatal error if the trigger is not found. */
2347 get_trigger_info_for_log(trigger, &trigger_name, &trigger_owner_uid);
2348 DBG("Failed to retrieve registered version of trigger: trigger name = '%s', trigger owner uid = %d",
2349 trigger_name, (int) trigger_owner_uid);
2350
2351 ret = 0;
2352
2353 end:
2354 rcu_read_unlock();
2355 *_cmd_result = cmd_result;
2356 return ret;
2357 }
2358
2359 static
2360 bool condition_is_supported(struct lttng_condition *condition)
2361 {
2362 bool is_supported;
2363
2364 switch (lttng_condition_get_type(condition)) {
2365 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
2366 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
2367 {
2368 int ret;
2369 enum lttng_domain_type domain;
2370
2371 ret = lttng_condition_buffer_usage_get_domain_type(condition,
2372 &domain);
2373 LTTNG_ASSERT(ret == 0);
2374
2375 if (domain != LTTNG_DOMAIN_KERNEL) {
2376 is_supported = true;
2377 goto end;
2378 }
2379
2380 /*
2381 * Older kernel tracers don't expose the API to monitor their
2382 * buffers. Therefore, we reject triggers that require that
2383 * mechanism to be available to be evaluated.
2384 *
2385 * Assume unsupported on error.
2386 */
2387 is_supported = kernel_supports_ring_buffer_snapshot_sample_positions() == 1;
2388 break;
2389 }
2390 case LTTNG_CONDITION_TYPE_EVENT_RULE_MATCHES:
2391 {
2392 const struct lttng_event_rule *event_rule;
2393 enum lttng_domain_type domain;
2394 const enum lttng_condition_status status =
2395 lttng_condition_event_rule_matches_get_rule(
2396 condition, &event_rule);
2397
2398 LTTNG_ASSERT(status == LTTNG_CONDITION_STATUS_OK);
2399
2400 domain = lttng_event_rule_get_domain_type(event_rule);
2401 if (domain != LTTNG_DOMAIN_KERNEL) {
2402 is_supported = true;
2403 goto end;
2404 }
2405
2406 /*
2407 * Older kernel tracers can't emit notification. Therefore, we
2408 * reject triggers that require that mechanism to be available
2409 * to be evaluated.
2410 *
2411 * Assume unsupported on error.
2412 */
2413 is_supported = kernel_supports_event_notifiers() == 1;
2414 break;
2415 }
2416 default:
2417 is_supported = true;
2418 }
2419 end:
2420 return is_supported;
2421 }
2422
2423 /* Must be called with RCU read lock held. */
2424 static
2425 int bind_trigger_to_matching_session(struct lttng_trigger *trigger,
2426 struct notification_thread_state *state)
2427 {
2428 int ret = 0;
2429 const struct lttng_condition *condition;
2430 const char *session_name;
2431 struct lttng_session_trigger_list *trigger_list;
2432
2433 ASSERT_RCU_READ_LOCKED();
2434
2435 condition = lttng_trigger_get_const_condition(trigger);
2436 switch (lttng_condition_get_type(condition)) {
2437 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
2438 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
2439 {
2440 enum lttng_condition_status status;
2441
2442 status = lttng_condition_session_rotation_get_session_name(
2443 condition, &session_name);
2444 if (status != LTTNG_CONDITION_STATUS_OK) {
2445 ERR("Failed to bind trigger to session: unable to get 'session_rotation' condition's session name");
2446 ret = -1;
2447 goto end;
2448 }
2449 break;
2450 }
2451 default:
2452 ret = -1;
2453 goto end;
2454 }
2455
2456 trigger_list = get_session_trigger_list(state, session_name);
2457 if (!trigger_list) {
2458 DBG("Unable to bind trigger applying to session \"%s\" as it is not yet known to the notification system",
2459 session_name);
2460 goto end;
2461
2462 }
2463
2464 DBG("Newly registered trigger bound to session \"%s\"",
2465 session_name);
2466 ret = lttng_session_trigger_list_add(trigger_list, trigger);
2467 end:
2468 return ret;
2469 }
2470
2471 /* Must be called with RCU read lock held. */
2472 static
2473 int bind_trigger_to_matching_channels(struct lttng_trigger *trigger,
2474 struct notification_thread_state *state)
2475 {
2476 int ret = 0;
2477 struct cds_lfht_node *node;
2478 struct cds_lfht_iter iter;
2479 struct channel_info *channel;
2480
2481 ASSERT_RCU_READ_LOCKED();
2482
2483 cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
2484 channels_ht_node) {
2485 struct lttng_trigger_list_element *trigger_list_element;
2486 struct lttng_channel_trigger_list *trigger_list;
2487 struct cds_lfht_iter lookup_iter;
2488
2489 if (!trigger_applies_to_channel(trigger, channel)) {
2490 continue;
2491 }
2492
2493 cds_lfht_lookup(state->channel_triggers_ht,
2494 hash_channel_key(&channel->key),
2495 match_channel_trigger_list,
2496 &channel->key,
2497 &lookup_iter);
2498 node = cds_lfht_iter_get_node(&lookup_iter);
2499 LTTNG_ASSERT(node);
2500 trigger_list = caa_container_of(node,
2501 struct lttng_channel_trigger_list,
2502 channel_triggers_ht_node);
2503
2504 trigger_list_element = (lttng_trigger_list_element *) zmalloc(sizeof(*trigger_list_element));
2505 if (!trigger_list_element) {
2506 ret = -1;
2507 goto end;
2508 }
2509 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
2510 trigger_list_element->trigger = trigger;
2511 cds_list_add(&trigger_list_element->node, &trigger_list->list);
2512 DBG("Newly registered trigger bound to channel \"%s\"",
2513 channel->name);
2514 }
2515 end:
2516 return ret;
2517 }
2518
2519 static
2520 bool is_trigger_action_notify(const struct lttng_trigger *trigger)
2521 {
2522 bool is_notify = false;
2523 unsigned int i, count;
2524 enum lttng_action_status action_status;
2525 const struct lttng_action *action =
2526 lttng_trigger_get_const_action(trigger);
2527 enum lttng_action_type action_type;
2528
2529 LTTNG_ASSERT(action);
2530 action_type = lttng_action_get_type(action);
2531 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2532 is_notify = true;
2533 goto end;
2534 } else if (action_type != LTTNG_ACTION_TYPE_LIST) {
2535 goto end;
2536 }
2537
2538 action_status = lttng_action_list_get_count(action, &count);
2539 LTTNG_ASSERT(action_status == LTTNG_ACTION_STATUS_OK);
2540
2541 for (i = 0; i < count; i++) {
2542 const struct lttng_action *inner_action =
2543 lttng_action_list_get_at_index(
2544 action, i);
2545
2546 action_type = lttng_action_get_type(inner_action);
2547 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2548 is_notify = true;
2549 goto end;
2550 }
2551 }
2552
2553 end:
2554 return is_notify;
2555 }
2556
2557 static bool trigger_name_taken(struct notification_thread_state *state,
2558 const struct lttng_trigger *trigger)
2559 {
2560 struct cds_lfht_iter iter;
2561
2562 /*
2563 * No duplicata is allowed in the triggers_by_name_uid_ht.
2564 * The match is done against the trigger name and uid.
2565 */
2566 cds_lfht_lookup(state->triggers_by_name_uid_ht,
2567 hash_trigger_by_name_uid(trigger),
2568 match_trigger_by_name_uid,
2569 trigger,
2570 &iter);
2571 return !!cds_lfht_iter_get_node(&iter);
2572 }
2573
2574 static
2575 enum lttng_error_code generate_trigger_name(
2576 struct notification_thread_state *state,
2577 struct lttng_trigger *trigger, const char **name)
2578 {
2579 enum lttng_error_code ret_code = LTTNG_OK;
2580 bool taken = false;
2581 enum lttng_trigger_status status;
2582
2583 do {
2584 const int ret = lttng_trigger_generate_name(trigger,
2585 state->trigger_id.name_offset++);
2586 if (ret) {
2587 /* The only reason this can fail right now. */
2588 ret_code = LTTNG_ERR_NOMEM;
2589 break;
2590 }
2591
2592 status = lttng_trigger_get_name(trigger, name);
2593 LTTNG_ASSERT(status == LTTNG_TRIGGER_STATUS_OK);
2594
2595 taken = trigger_name_taken(state, trigger);
2596 } while (taken || state->trigger_id.name_offset == UINT64_MAX);
2597
2598 return ret_code;
2599 }
2600
2601 static inline
2602 void notif_thread_state_remove_trigger_ht_elem(
2603 struct notification_thread_state *state,
2604 struct lttng_trigger_ht_element *trigger_ht_element)
2605 {
2606 LTTNG_ASSERT(state);
2607 LTTNG_ASSERT(trigger_ht_element);
2608
2609 cds_lfht_del(state->triggers_ht, &trigger_ht_element->node);
2610 cds_lfht_del(state->triggers_by_name_uid_ht, &trigger_ht_element->node_by_name_uid);
2611 }
2612
2613 static
2614 enum lttng_error_code setup_tracer_notifier(
2615 struct notification_thread_state *state,
2616 struct lttng_trigger *trigger)
2617 {
2618 enum lttng_error_code ret;
2619 enum event_notifier_error_accounting_status error_accounting_status;
2620 struct cds_lfht_node *node;
2621 uint64_t error_counter_index = 0;
2622 struct lttng_condition *condition = lttng_trigger_get_condition(trigger);
2623 struct notification_trigger_tokens_ht_element *trigger_tokens_ht_element = NULL;
2624
2625 trigger_tokens_ht_element = (notification_trigger_tokens_ht_element *) zmalloc(sizeof(*trigger_tokens_ht_element));
2626 if (!trigger_tokens_ht_element) {
2627 ret = LTTNG_ERR_NOMEM;
2628 goto end;
2629 }
2630
2631 /* Add trigger token to the trigger_tokens_ht. */
2632 cds_lfht_node_init(&trigger_tokens_ht_element->node);
2633 trigger_tokens_ht_element->token = LTTNG_OPTIONAL_GET(trigger->tracer_token);
2634 trigger_tokens_ht_element->trigger = trigger;
2635
2636 node = cds_lfht_add_unique(state->trigger_tokens_ht,
2637 hash_key_u64(&trigger_tokens_ht_element->token, lttng_ht_seed),
2638 match_trigger_token,
2639 &trigger_tokens_ht_element->token,
2640 &trigger_tokens_ht_element->node);
2641 if (node != &trigger_tokens_ht_element->node) {
2642 ret = LTTNG_ERR_TRIGGER_EXISTS;
2643 goto error_free_ht_element;
2644 }
2645
2646 error_accounting_status = event_notifier_error_accounting_register_event_notifier(
2647 trigger, &error_counter_index);
2648 if (error_accounting_status != EVENT_NOTIFIER_ERROR_ACCOUNTING_STATUS_OK) {
2649 if (error_accounting_status == EVENT_NOTIFIER_ERROR_ACCOUNTING_STATUS_NO_INDEX_AVAILABLE) {
2650 DBG("Trigger list error accounting counter full.");
2651 ret = LTTNG_ERR_EVENT_NOTIFIER_ERROR_ACCOUNTING_FULL;
2652 } else {
2653 ERR("Error registering trigger for error accounting");
2654 ret = LTTNG_ERR_EVENT_NOTIFIER_REGISTRATION;
2655 }
2656
2657 goto error_remove_ht_element;
2658 }
2659
2660 lttng_condition_event_rule_matches_set_error_counter_index(
2661 condition, error_counter_index);
2662
2663 ret = LTTNG_OK;
2664 goto end;
2665
2666 error_remove_ht_element:
2667 cds_lfht_del(state->trigger_tokens_ht, &trigger_tokens_ht_element->node);
2668 error_free_ht_element:
2669 free(trigger_tokens_ht_element);
2670 end:
2671 return ret;
2672 }
2673
2674 /*
2675 * FIXME A client's credentials are not checked when registering a trigger.
2676 *
2677 * The effects of this are benign since:
2678 * - The client will succeed in registering the trigger, as it is valid,
2679 * - The trigger will, internally, be bound to the channel/session,
2680 * - The notifications will not be sent since the client's credentials
2681 * are checked against the channel at that moment.
2682 *
2683 * If this function returns a non-zero value, it means something is
2684 * fundamentally broken and the whole subsystem/thread will be torn down.
2685 *
2686 * If a non-fatal error occurs, just set the cmd_result to the appropriate
2687 * error code.
2688 */
2689 static
2690 int handle_notification_thread_command_register_trigger(
2691 struct notification_thread_state *state,
2692 struct lttng_trigger *trigger,
2693 bool is_trigger_anonymous,
2694 enum lttng_error_code *cmd_result)
2695 {
2696 int ret = 0;
2697 struct lttng_condition *condition;
2698 struct notification_client_list *client_list = NULL;
2699 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2700 struct cds_lfht_node *node;
2701 const char* trigger_name;
2702 bool free_trigger = true;
2703 struct lttng_evaluation *evaluation = NULL;
2704 struct lttng_credentials object_creds;
2705 uid_t object_uid;
2706 gid_t object_gid;
2707 enum action_executor_status executor_status;
2708 const uint64_t trigger_tracer_token =
2709 state->trigger_id.next_tracer_token++;
2710
2711 rcu_read_lock();
2712
2713 /* Set the trigger's tracer token. */
2714 lttng_trigger_set_tracer_token(trigger, trigger_tracer_token);
2715
2716 if (!is_trigger_anonymous) {
2717 if (lttng_trigger_get_name(trigger, &trigger_name) ==
2718 LTTNG_TRIGGER_STATUS_UNSET) {
2719 const enum lttng_error_code ret_code =
2720 generate_trigger_name(state, trigger,
2721 &trigger_name);
2722
2723 if (ret_code != LTTNG_OK) {
2724 /* Fatal error. */
2725 ret = -1;
2726 *cmd_result = ret_code;
2727 goto error;
2728 }
2729 } else if (trigger_name_taken(state, trigger)) {
2730 /* Not a fatal error. */
2731 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2732 ret = 0;
2733 goto error;
2734 }
2735 } else {
2736 trigger_name = "(anonymous)";
2737 }
2738
2739 condition = lttng_trigger_get_condition(trigger);
2740 LTTNG_ASSERT(condition);
2741
2742 /* Some conditions require tracers to implement a minimal ABI version. */
2743 if (!condition_is_supported(condition)) {
2744 *cmd_result = LTTNG_ERR_NOT_SUPPORTED;
2745 goto error;
2746 }
2747
2748 trigger_ht_element = (lttng_trigger_ht_element *) zmalloc(sizeof(*trigger_ht_element));
2749 if (!trigger_ht_element) {
2750 ret = -1;
2751 goto error;
2752 }
2753
2754 /* Add trigger to the trigger_ht. */
2755 cds_lfht_node_init(&trigger_ht_element->node);
2756 cds_lfht_node_init(&trigger_ht_element->node_by_name_uid);
2757 trigger_ht_element->trigger = trigger;
2758
2759 node = cds_lfht_add_unique(state->triggers_ht,
2760 lttng_condition_hash(condition),
2761 match_trigger,
2762 trigger,
2763 &trigger_ht_element->node);
2764 if (node != &trigger_ht_element->node) {
2765 /* Not a fatal error, simply report it to the client. */
2766 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2767 goto error_free_ht_element;
2768 }
2769
2770 node = cds_lfht_add_unique(state->triggers_by_name_uid_ht,
2771 hash_trigger_by_name_uid(trigger),
2772 match_trigger_by_name_uid,
2773 trigger,
2774 &trigger_ht_element->node_by_name_uid);
2775 if (node != &trigger_ht_element->node_by_name_uid) {
2776 /* Internal error: add to triggers_ht should have failed. */
2777 ret = -1;
2778 goto error_free_ht_element;
2779 }
2780
2781 /* From this point consider the trigger registered. */
2782 lttng_trigger_set_as_registered(trigger);
2783
2784 /*
2785 * Some triggers might need a tracer notifier depending on its
2786 * condition and actions.
2787 */
2788 if (lttng_trigger_needs_tracer_notifier(trigger)) {
2789 enum lttng_error_code error_code;
2790
2791 error_code = setup_tracer_notifier(state, trigger);
2792 if (error_code != LTTNG_OK) {
2793 notif_thread_state_remove_trigger_ht_elem(state,
2794 trigger_ht_element);
2795 if (error_code == LTTNG_ERR_NOMEM) {
2796 ret = -1;
2797 } else {
2798 *cmd_result = error_code;
2799 ret = 0;
2800 }
2801
2802 goto error_free_ht_element;
2803 }
2804 }
2805
2806 /*
2807 * The rest only applies to triggers that have a "notify" action.
2808 * It is not skipped as this is the only action type currently
2809 * supported.
2810 */
2811 if (is_trigger_action_notify(trigger)) {
2812 /*
2813 * Find or create the client list of this condition. It may
2814 * already be present if another trigger is already registered
2815 * with the same condition.
2816 */
2817 client_list = get_client_list_from_condition(state, condition);
2818 if (!client_list) {
2819 /*
2820 * No client list for this condition yet. We create new
2821 * one and build it up.
2822 */
2823 client_list = notification_client_list_create(state, condition);
2824 if (!client_list) {
2825 ERR("Error creating notification client list for trigger %s", trigger->name);
2826 goto error_free_ht_element;
2827 }
2828 }
2829
2830 CDS_INIT_LIST_HEAD(&trigger_ht_element->client_list_trigger_node);
2831
2832 pthread_mutex_lock(&client_list->lock);
2833 cds_list_add(&trigger_ht_element->client_list_trigger_node, &client_list->triggers_list);
2834 pthread_mutex_unlock(&client_list->lock);
2835 }
2836
2837 /*
2838 * Ownership of the trigger and of its wrapper was transfered to
2839 * the triggers_ht. Same for token ht element if necessary.
2840 */
2841 trigger_ht_element = NULL;
2842 free_trigger = false;
2843
2844 switch (get_condition_binding_object(condition)) {
2845 case LTTNG_OBJECT_TYPE_SESSION:
2846 /* Add the trigger to the list if it matches a known session. */
2847 ret = bind_trigger_to_matching_session(trigger, state);
2848 if (ret) {
2849 goto error_free_ht_element;
2850 }
2851 break;
2852 case LTTNG_OBJECT_TYPE_CHANNEL:
2853 /*
2854 * Add the trigger to list of triggers bound to the channels
2855 * currently known.
2856 */
2857 ret = bind_trigger_to_matching_channels(trigger, state);
2858 if (ret) {
2859 goto error_free_ht_element;
2860 }
2861 break;
2862 case LTTNG_OBJECT_TYPE_NONE:
2863 break;
2864 default:
2865 ERR("Unknown object type on which to bind a newly registered trigger was encountered");
2866 ret = -1;
2867 goto error_free_ht_element;
2868 }
2869
2870 /*
2871 * The new trigger's condition must be evaluated against the current
2872 * state.
2873 *
2874 * In the case of `notify` action, nothing preventing clients from
2875 * subscribing to a condition before the corresponding trigger is
2876 * registered, we have to evaluate this new condition right away.
2877 *
2878 * At some point, we were waiting for the next "evaluation" (e.g. on
2879 * reception of a channel sample) to evaluate this new condition, but
2880 * that was broken.
2881 *
2882 * The reason it was broken is that waiting for the next sample
2883 * does not allow us to properly handle transitions for edge-triggered
2884 * conditions.
2885 *
2886 * Consider this example: when we handle a new channel sample, we
2887 * evaluate each conditions twice: once with the previous state, and
2888 * again with the newest state. We then use those two results to
2889 * determine whether a state change happened: a condition was false and
2890 * became true. If a state change happened, we have to notify clients.
2891 *
2892 * Now, if a client subscribes to a given notification and registers
2893 * a trigger *after* that subscription, we have to make sure the
2894 * condition is evaluated at this point while considering only the
2895 * current state. Otherwise, the next evaluation cycle may only see
2896 * that the evaluations remain the same (true for samples n-1 and n) and
2897 * the client will never know that the condition has been met.
2898 */
2899 switch (get_condition_binding_object(condition)) {
2900 case LTTNG_OBJECT_TYPE_SESSION:
2901 ret = evaluate_session_condition_for_client(condition, state,
2902 &evaluation, &object_uid,
2903 &object_gid);
2904 LTTNG_OPTIONAL_SET(&object_creds.uid, object_uid);
2905 LTTNG_OPTIONAL_SET(&object_creds.gid, object_gid);
2906 break;
2907 case LTTNG_OBJECT_TYPE_CHANNEL:
2908 ret = evaluate_channel_condition_for_client(condition, state,
2909 &evaluation, &object_uid,
2910 &object_gid);
2911 LTTNG_OPTIONAL_SET(&object_creds.uid, object_uid);
2912 LTTNG_OPTIONAL_SET(&object_creds.gid, object_gid);
2913 break;
2914 case LTTNG_OBJECT_TYPE_NONE:
2915 ret = 0;
2916 break;
2917 case LTTNG_OBJECT_TYPE_UNKNOWN:
2918 default:
2919 ret = -1;
2920 break;
2921 }
2922
2923 if (ret) {
2924 /* Fatal error. */
2925 goto error_free_ht_element;
2926 }
2927
2928 DBG("Newly registered trigger's condition evaluated to %s",
2929 evaluation ? "true" : "false");
2930 if (!evaluation) {
2931 /* Evaluation yielded nothing. Normal exit. */
2932 ret = 0;
2933 goto success;
2934 }
2935
2936 /*
2937 * Ownership of `evaluation` transferred to the action executor
2938 * no matter the result.
2939 */
2940 executor_status = action_executor_enqueue_trigger(state->executor,
2941 trigger, evaluation, &object_creds, client_list);
2942 evaluation = NULL;
2943 switch (executor_status) {
2944 case ACTION_EXECUTOR_STATUS_OK:
2945 break;
2946 case ACTION_EXECUTOR_STATUS_ERROR:
2947 case ACTION_EXECUTOR_STATUS_INVALID:
2948 /*
2949 * TODO Add trigger identification (name/id) when
2950 * it is added to the API.
2951 */
2952 ERR("Fatal error occurred while enqueuing action associated to newly registered trigger");
2953 ret = -1;
2954 goto error_free_ht_element;
2955 case ACTION_EXECUTOR_STATUS_OVERFLOW:
2956 /*
2957 * TODO Add trigger identification (name/id) when
2958 * it is added to the API.
2959 *
2960 * Not a fatal error.
2961 */
2962 WARN("No space left when enqueuing action associated to newly registered trigger");
2963 ret = 0;
2964 goto success;
2965 default:
2966 abort();
2967 }
2968
2969 success:
2970 *cmd_result = LTTNG_OK;
2971 DBG("Registered trigger: name = `%s`, tracer token = %" PRIu64,
2972 trigger_name, trigger_tracer_token);
2973 goto end;
2974
2975 error_free_ht_element:
2976 if (trigger_ht_element) {
2977 /* Delayed removal due to RCU constraint on delete. */
2978 call_rcu(&trigger_ht_element->rcu_node,
2979 free_lttng_trigger_ht_element_rcu);
2980 }
2981 error:
2982 if (free_trigger) {
2983 /*
2984 * Other objects might have a reference to the trigger, mark it
2985 * as unregistered.
2986 */
2987 lttng_trigger_set_as_unregistered(trigger);
2988 lttng_trigger_destroy(trigger);
2989 }
2990 end:
2991 rcu_read_unlock();
2992 return ret;
2993 }
2994
2995 static
2996 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node)
2997 {
2998 free(caa_container_of(node, struct lttng_trigger_ht_element,
2999 rcu_node));
3000 }
3001
3002 static
3003 void free_notification_trigger_tokens_ht_element_rcu(struct rcu_head *node)
3004 {
3005 free(caa_container_of(node, struct notification_trigger_tokens_ht_element,
3006 rcu_node));
3007 }
3008
3009 static
3010 void teardown_tracer_notifier(struct notification_thread_state *state,
3011 const struct lttng_trigger *trigger)
3012 {
3013 struct cds_lfht_iter iter;
3014 struct notification_trigger_tokens_ht_element *trigger_tokens_ht_element;
3015
3016 cds_lfht_for_each_entry(state->trigger_tokens_ht, &iter,
3017 trigger_tokens_ht_element, node) {
3018
3019 if (!lttng_trigger_is_equal(trigger,
3020 trigger_tokens_ht_element->trigger)) {
3021 continue;
3022 }
3023
3024 event_notifier_error_accounting_unregister_event_notifier(
3025 trigger_tokens_ht_element->trigger);
3026
3027 /* TODO talk to all app and remove it */
3028 DBG("Removed trigger from tokens_ht");
3029 cds_lfht_del(state->trigger_tokens_ht,
3030 &trigger_tokens_ht_element->node);
3031
3032 call_rcu(&trigger_tokens_ht_element->rcu_node,
3033 free_notification_trigger_tokens_ht_element_rcu);
3034
3035 break;
3036 }
3037 }
3038
3039 static
3040 int handle_notification_thread_command_unregister_trigger(
3041 struct notification_thread_state *state,
3042 const struct lttng_trigger *trigger,
3043 enum lttng_error_code *_cmd_reply)
3044 {
3045 struct cds_lfht_iter iter;
3046 struct cds_lfht_node *triggers_ht_node;
3047 struct lttng_channel_trigger_list *trigger_list;
3048 struct notification_client_list *client_list;
3049 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
3050 const struct lttng_condition *condition = lttng_trigger_get_const_condition(
3051 trigger);
3052 enum lttng_error_code cmd_reply;
3053
3054 rcu_read_lock();
3055
3056 cds_lfht_lookup(state->triggers_ht,
3057 lttng_condition_hash(condition),
3058 match_trigger,
3059 trigger,
3060 &iter);
3061 triggers_ht_node = cds_lfht_iter_get_node(&iter);
3062 if (!triggers_ht_node) {
3063 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
3064 goto end;
3065 } else {
3066 cmd_reply = LTTNG_OK;
3067 }
3068
3069 trigger_ht_element = caa_container_of(triggers_ht_node,
3070 struct lttng_trigger_ht_element, node);
3071
3072 /* Remove trigger from channel_triggers_ht. */
3073 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
3074 channel_triggers_ht_node) {
3075 struct lttng_trigger_list_element *trigger_element, *tmp;
3076
3077 cds_list_for_each_entry_safe(trigger_element, tmp,
3078 &trigger_list->list, node) {
3079 if (!lttng_trigger_is_equal(trigger, trigger_element->trigger)) {
3080 continue;
3081 }
3082
3083 DBG("Removed trigger from channel_triggers_ht");
3084 cds_list_del(&trigger_element->node);
3085 /* A trigger can only appear once per channel */
3086 break;
3087 }
3088 }
3089
3090 if (lttng_trigger_needs_tracer_notifier(trigger)) {
3091 teardown_tracer_notifier(state, trigger);
3092 }
3093
3094 if (is_trigger_action_notify(trigger)) {
3095 /*
3096 * Remove and release the client list from
3097 * notification_trigger_clients_ht.
3098 */
3099 client_list = get_client_list_from_condition(state, condition);
3100 LTTNG_ASSERT(client_list);
3101
3102 pthread_mutex_lock(&client_list->lock);
3103 cds_list_del(&trigger_ht_element->client_list_trigger_node);
3104 pthread_mutex_unlock(&client_list->lock);
3105
3106 /* Put new reference and the hashtable's reference. */
3107 notification_client_list_put(client_list);
3108 notification_client_list_put(client_list);
3109 client_list = NULL;
3110 }
3111
3112 /* Remove trigger from triggers_ht. */
3113 notif_thread_state_remove_trigger_ht_elem(state, trigger_ht_element);
3114
3115 /* Release the ownership of the trigger. */
3116 lttng_trigger_destroy(trigger_ht_element->trigger);
3117 call_rcu(&trigger_ht_element->rcu_node, free_lttng_trigger_ht_element_rcu);
3118 end:
3119 rcu_read_unlock();
3120 if (_cmd_reply) {
3121 *_cmd_reply = cmd_reply;
3122 }
3123 return 0;
3124 }
3125
3126 /* Returns 0 on success, 1 on exit requested, negative value on error. */
3127 int handle_notification_thread_command(
3128 struct notification_thread_handle *handle,
3129 struct notification_thread_state *state)
3130 {
3131 int ret;
3132 uint64_t counter;
3133 struct notification_thread_command *cmd;
3134
3135 /* Read the event pipe to put it back into a quiescent state. */
3136 ret = lttng_read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
3137 sizeof(counter));
3138 if (ret != sizeof(counter)) {
3139 goto error;
3140 }
3141
3142 pthread_mutex_lock(&handle->cmd_queue.lock);
3143 cmd = cds_list_first_entry(&handle->cmd_queue.list,
3144 struct notification_thread_command, cmd_list_node);
3145 cds_list_del(&cmd->cmd_list_node);
3146 pthread_mutex_unlock(&handle->cmd_queue.lock);
3147
3148 DBG("Received `%s` command",
3149 notification_command_type_str(cmd->type));
3150 switch (cmd->type) {
3151 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
3152 ret = handle_notification_thread_command_register_trigger(state,
3153 cmd->parameters.register_trigger.trigger,
3154 cmd->parameters.register_trigger.is_trigger_anonymous,
3155 &cmd->reply_code);
3156 break;
3157 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
3158 ret = handle_notification_thread_command_unregister_trigger(
3159 state,
3160 cmd->parameters.unregister_trigger.trigger,
3161 &cmd->reply_code);
3162 break;
3163 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
3164 ret = handle_notification_thread_command_add_channel(
3165 state,
3166 cmd->parameters.add_channel.session.name,
3167 cmd->parameters.add_channel.session.uid,
3168 cmd->parameters.add_channel.session.gid,
3169 cmd->parameters.add_channel.channel.name,
3170 cmd->parameters.add_channel.channel.domain,
3171 cmd->parameters.add_channel.channel.key,
3172 cmd->parameters.add_channel.channel.capacity,
3173 &cmd->reply_code);
3174 break;
3175 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
3176 ret = handle_notification_thread_command_remove_channel(
3177 state, cmd->parameters.remove_channel.key,
3178 cmd->parameters.remove_channel.domain,
3179 &cmd->reply_code);
3180 break;
3181 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
3182 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
3183 ret = handle_notification_thread_command_session_rotation(
3184 state,
3185 cmd->type,
3186 cmd->parameters.session_rotation.session_name,
3187 cmd->parameters.session_rotation.uid,
3188 cmd->parameters.session_rotation.gid,
3189 cmd->parameters.session_rotation.trace_archive_chunk_id,
3190 cmd->parameters.session_rotation.location,
3191 &cmd->reply_code);
3192 break;
3193 case NOTIFICATION_COMMAND_TYPE_ADD_TRACER_EVENT_SOURCE:
3194 ret = handle_notification_thread_command_add_tracer_event_source(
3195 state,
3196 cmd->parameters.tracer_event_source.tracer_event_source_fd,
3197 cmd->parameters.tracer_event_source.domain,
3198 &cmd->reply_code);
3199 break;
3200 case NOTIFICATION_COMMAND_TYPE_REMOVE_TRACER_EVENT_SOURCE:
3201 ret = handle_notification_thread_command_remove_tracer_event_source(
3202 state,
3203 cmd->parameters.tracer_event_source.tracer_event_source_fd,
3204 &cmd->reply_code);
3205 break;
3206 case NOTIFICATION_COMMAND_TYPE_LIST_TRIGGERS:
3207 {
3208 struct lttng_triggers *triggers = NULL;
3209
3210 ret = handle_notification_thread_command_list_triggers(
3211 handle,
3212 state,
3213 cmd->parameters.list_triggers.uid,
3214 &triggers,
3215 &cmd->reply_code);
3216 cmd->reply.list_triggers.triggers = triggers;
3217 ret = 0;
3218 break;
3219 }
3220 case NOTIFICATION_COMMAND_TYPE_QUIT:
3221 cmd->reply_code = LTTNG_OK;
3222 ret = 1;
3223 goto end;
3224 case NOTIFICATION_COMMAND_TYPE_GET_TRIGGER:
3225 {
3226 struct lttng_trigger *trigger = NULL;
3227
3228 ret = handle_notification_thread_command_get_trigger(state,
3229 cmd->parameters.get_trigger.trigger, &trigger,
3230 &cmd->reply_code);
3231 cmd->reply.get_trigger.trigger = trigger;
3232 break;
3233 }
3234 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE:
3235 {
3236 const enum client_transmission_status client_status =
3237 cmd->parameters.client_communication_update
3238 .status;
3239 const notification_client_id client_id =
3240 cmd->parameters.client_communication_update.id;
3241 struct notification_client *client;
3242
3243 rcu_read_lock();
3244 client = get_client_from_id(client_id, state);
3245
3246 if (!client) {
3247 /*
3248 * Client error was probably already picked-up by the
3249 * notification thread or it has disconnected
3250 * gracefully while this command was queued.
3251 */
3252 DBG("Failed to find notification client to update communication status, client id = %" PRIu64,
3253 client_id);
3254 ret = 0;
3255 } else {
3256 ret = client_handle_transmission_status(
3257 client, client_status, state);
3258 }
3259 rcu_read_unlock();
3260 break;
3261 }
3262 default:
3263 ERR("Unknown internal command received");
3264 goto error_unlock;
3265 }
3266
3267 if (ret) {
3268 goto error_unlock;
3269 }
3270 end:
3271 if (cmd->is_async) {
3272 free(cmd);
3273 cmd = NULL;
3274 } else {
3275 lttng_waiter_wake_up(&cmd->reply_waiter);
3276 }
3277 return ret;
3278 error_unlock:
3279 /* Wake-up and return a fatal error to the calling thread. */
3280 lttng_waiter_wake_up(&cmd->reply_waiter);
3281 cmd->reply_code = LTTNG_ERR_FATAL;
3282 error:
3283 /* Indicate a fatal error to the caller. */
3284 return -1;
3285 }
3286
3287 static
3288 int socket_set_non_blocking(int socket)
3289 {
3290 int ret, flags;
3291
3292 /* Set the pipe as non-blocking. */
3293 ret = fcntl(socket, F_GETFL, 0);
3294 if (ret == -1) {
3295 PERROR("fcntl get socket flags");
3296 goto end;
3297 }
3298 flags = ret;
3299
3300 ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
3301 if (ret == -1) {
3302 PERROR("fcntl set O_NONBLOCK socket flag");
3303 goto end;
3304 }
3305 DBG("Client socket (fd = %i) set as non-blocking", socket);
3306 end:
3307 return ret;
3308 }
3309
3310 static
3311 int client_reset_inbound_state(struct notification_client *client)
3312 {
3313 int ret;
3314
3315
3316 lttng_payload_clear(&client->communication.inbound.payload);
3317
3318 client->communication.inbound.bytes_to_receive =
3319 sizeof(struct lttng_notification_channel_message);
3320 client->communication.inbound.msg_type =
3321 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
3322 LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
3323 LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
3324 ret = lttng_dynamic_buffer_set_size(
3325 &client->communication.inbound.payload.buffer,
3326 client->communication.inbound.bytes_to_receive);
3327
3328 return ret;
3329 }
3330
3331 int handle_notification_thread_client_connect(
3332 struct notification_thread_state *state)
3333 {
3334 int ret;
3335 struct notification_client *client;
3336
3337 DBG("Handling new notification channel client connection");
3338
3339 client = (notification_client *) zmalloc(sizeof(*client));
3340 if (!client) {
3341 /* Fatal error. */
3342 ret = -1;
3343 goto error;
3344 }
3345
3346 pthread_mutex_init(&client->lock, NULL);
3347 client->id = state->next_notification_client_id++;
3348 CDS_INIT_LIST_HEAD(&client->condition_list);
3349 lttng_payload_init(&client->communication.inbound.payload);
3350 lttng_payload_init(&client->communication.outbound.payload);
3351 client->communication.inbound.expect_creds = true;
3352
3353 ret = client_reset_inbound_state(client);
3354 if (ret) {
3355 ERR("Failed to reset client communication's inbound state");
3356 ret = 0;
3357 goto error;
3358 }
3359
3360 ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
3361 if (ret < 0) {
3362 ERR("Failed to accept new notification channel client connection");
3363 ret = 0;
3364 goto error;
3365 }
3366
3367 client->socket = ret;
3368
3369 ret = socket_set_non_blocking(client->socket);
3370 if (ret) {
3371 ERR("Failed to set new notification channel client connection socket as non-blocking");
3372 goto error;
3373 }
3374
3375 ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
3376 if (ret < 0) {
3377 ERR("Failed to set socket options on new notification channel client socket");
3378 ret = 0;
3379 goto error;
3380 }
3381
3382 ret = lttng_poll_add(&state->events, client->socket,
3383 LPOLLIN | LPOLLERR |
3384 LPOLLHUP | LPOLLRDHUP);
3385 if (ret < 0) {
3386 ERR("Failed to add notification channel client socket to poll set");
3387 ret = 0;
3388 goto error;
3389 }
3390 DBG("Added new notification channel client socket (%i) to poll set",
3391 client->socket);
3392
3393 rcu_read_lock();
3394 cds_lfht_add(state->client_socket_ht,
3395 hash_client_socket(client->socket),
3396 &client->client_socket_ht_node);
3397 cds_lfht_add(state->client_id_ht,
3398 hash_client_id(client->id),
3399 &client->client_id_ht_node);
3400 rcu_read_unlock();
3401
3402 return ret;
3403
3404 error:
3405 notification_client_destroy(client);
3406 return ret;
3407 }
3408
3409 /*
3410 * RCU read-lock must be held by the caller.
3411 * Client lock must _not_ be held by the caller.
3412 */
3413 static
3414 int notification_thread_client_disconnect(
3415 struct notification_client *client,
3416 struct notification_thread_state *state)
3417 {
3418 int ret;
3419 struct lttng_condition_list_element *condition_list_element, *tmp;
3420
3421 ASSERT_RCU_READ_LOCKED();
3422
3423 /* Acquire the client lock to disable its communication atomically. */
3424 pthread_mutex_lock(&client->lock);
3425 client->communication.active = false;
3426 cds_lfht_del(state->client_socket_ht, &client->client_socket_ht_node);
3427 cds_lfht_del(state->client_id_ht, &client->client_id_ht_node);
3428 pthread_mutex_unlock(&client->lock);
3429
3430 ret = lttng_poll_del(&state->events, client->socket);
3431 if (ret) {
3432 ERR("Failed to remove client socket %d from poll set",
3433 client->socket);
3434 }
3435
3436 /* Release all conditions to which the client was subscribed. */
3437 cds_list_for_each_entry_safe(condition_list_element, tmp,
3438 &client->condition_list, node) {
3439 (void) notification_thread_client_unsubscribe(client,
3440 condition_list_element->condition, state, NULL);
3441 }
3442
3443 /*
3444 * Client no longer accessible to other threads (through the
3445 * client lists).
3446 */
3447 notification_client_destroy(client);
3448 return ret;
3449 }
3450
3451 int handle_notification_thread_client_disconnect(
3452 int client_socket, struct notification_thread_state *state)
3453 {
3454 int ret = 0;
3455 struct notification_client *client;
3456
3457 rcu_read_lock();
3458 DBG("Closing client connection (socket fd = %i)",
3459 client_socket);
3460 client = get_client_from_socket(client_socket, state);
3461 if (!client) {
3462 /* Internal state corruption, fatal error. */
3463 ERR("Unable to find client (socket fd = %i)",
3464 client_socket);
3465 ret = -1;
3466 goto end;
3467 }
3468
3469 ret = notification_thread_client_disconnect(client, state);
3470 end:
3471 rcu_read_unlock();
3472 return ret;
3473 }
3474
3475 int handle_notification_thread_client_disconnect_all(
3476 struct notification_thread_state *state)
3477 {
3478 struct cds_lfht_iter iter;
3479 struct notification_client *client;
3480 bool error_encoutered = false;
3481
3482 rcu_read_lock();
3483 DBG("Closing all client connections");
3484 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
3485 client_socket_ht_node) {
3486 int ret;
3487
3488 ret = notification_thread_client_disconnect(
3489 client, state);
3490 if (ret) {
3491 error_encoutered = true;
3492 }
3493 }
3494 rcu_read_unlock();
3495 return error_encoutered ? 1 : 0;
3496 }
3497
3498 int handle_notification_thread_trigger_unregister_all(
3499 struct notification_thread_state *state)
3500 {
3501 bool error_occurred = false;
3502 struct cds_lfht_iter iter;
3503 struct lttng_trigger_ht_element *trigger_ht_element;
3504
3505 rcu_read_lock();
3506 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
3507 node) {
3508 int ret = handle_notification_thread_command_unregister_trigger(
3509 state, trigger_ht_element->trigger, NULL);
3510 if (ret) {
3511 error_occurred = true;
3512 }
3513 }
3514 rcu_read_unlock();
3515 return error_occurred ? -1 : 0;
3516 }
3517
3518 static
3519 int client_handle_transmission_status(
3520 struct notification_client *client,
3521 enum client_transmission_status transmission_status,
3522 struct notification_thread_state *state)
3523 {
3524 int ret = 0;
3525
3526 switch (transmission_status) {
3527 case CLIENT_TRANSMISSION_STATUS_COMPLETE:
3528 ret = lttng_poll_mod(&state->events, client->socket,
3529 CLIENT_POLL_MASK_IN);
3530 if (ret) {
3531 goto end;
3532 }
3533
3534 break;
3535 case CLIENT_TRANSMISSION_STATUS_QUEUED:
3536 /*
3537 * We want to be notified whenever there is buffer space
3538 * available to send the rest of the payload.
3539 */
3540 ret = lttng_poll_mod(&state->events, client->socket,
3541 CLIENT_POLL_MASK_IN_OUT);
3542 if (ret) {
3543 goto end;
3544 }
3545 break;
3546 case CLIENT_TRANSMISSION_STATUS_FAIL:
3547 ret = notification_thread_client_disconnect(client, state);
3548 if (ret) {
3549 goto end;
3550 }
3551 break;
3552 case CLIENT_TRANSMISSION_STATUS_ERROR:
3553 ret = -1;
3554 goto end;
3555 default:
3556 abort();
3557 }
3558 end:
3559 return ret;
3560 }
3561
3562 /* Client lock must be acquired by caller. */
3563 static
3564 enum client_transmission_status client_flush_outgoing_queue(
3565 struct notification_client *client)
3566 {
3567 ssize_t ret;
3568 size_t to_send_count;
3569 enum client_transmission_status status;
3570 struct lttng_payload_view pv = lttng_payload_view_from_payload(
3571 &client->communication.outbound.payload, 0, -1);
3572 const int fds_to_send_count =
3573 lttng_payload_view_get_fd_handle_count(&pv);
3574
3575 ASSERT_LOCKED(client->lock);
3576
3577 if (!client->communication.active) {
3578 status = CLIENT_TRANSMISSION_STATUS_FAIL;
3579 goto end;
3580 }
3581
3582 if (pv.buffer.size == 0) {
3583 /*
3584 * If both data and fds are equal to zero, we are in an invalid
3585 * state.
3586 */
3587 LTTNG_ASSERT(fds_to_send_count != 0);
3588 goto send_fds;
3589 }
3590
3591 /* Send data. */
3592 to_send_count = pv.buffer.size;
3593 DBG("Flushing client (socket fd = %i) outgoing queue",
3594 client->socket);
3595
3596 ret = lttcomm_send_unix_sock_non_block(client->socket,
3597 pv.buffer.data,
3598 to_send_count);
3599 if ((ret >= 0 && ret < to_send_count)) {
3600 DBG("Client (socket fd = %i) outgoing queue could not be completely flushed",
3601 client->socket);
3602 to_send_count -= std::max(ret, (ssize_t) 0);
3603
3604 memmove(client->communication.outbound.payload.buffer.data,
3605 pv.buffer.data +
3606 pv.buffer.size - to_send_count,
3607 to_send_count);
3608 ret = lttng_dynamic_buffer_set_size(
3609 &client->communication.outbound.payload.buffer,
3610 to_send_count);
3611 if (ret) {
3612 goto error;
3613 }
3614
3615 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
3616 goto end;
3617 } else if (ret < 0) {
3618 /* Generic error, disable the client's communication. */
3619 ERR("Failed to flush outgoing queue, disconnecting client (socket fd = %i)",
3620 client->socket);
3621 client->communication.active = false;
3622 status = CLIENT_TRANSMISSION_STATUS_FAIL;
3623 goto end;
3624 } else {
3625 /*
3626 * No error and flushed the queue completely.
3627 *
3628 * The payload buffer size is used later to
3629 * check if there is notifications queued. So albeit that the
3630 * direct caller knows that the transmission is complete, we
3631 * need to set the buffer size to zero.
3632 */
3633 ret = lttng_dynamic_buffer_set_size(
3634 &client->communication.outbound.payload.buffer, 0);
3635 if (ret) {
3636 goto error;
3637 }
3638 }
3639
3640 send_fds:
3641 /* No fds to send, transmission is complete. */
3642 if (fds_to_send_count == 0) {
3643 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
3644 goto end;
3645 }
3646
3647 ret = lttcomm_send_payload_view_fds_unix_sock_non_block(
3648 client->socket, &pv);
3649 if (ret < 0) {
3650 /* Generic error, disable the client's communication. */
3651 ERR("Failed to flush outgoing fds queue, disconnecting client (socket fd = %i)",
3652 client->socket);
3653 client->communication.active = false;
3654 status = CLIENT_TRANSMISSION_STATUS_FAIL;
3655 goto end;
3656 } else if (ret == 0) {
3657 /* Nothing could be sent. */
3658 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
3659 } else {
3660 /* Fd passing is an all or nothing kind of thing. */
3661 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
3662 /*
3663 * The payload _fd_array count is used later to
3664 * check if there is notifications queued. So although the
3665 * direct caller knows that the transmission is complete, we
3666 * need to clear the _fd_array for the queuing check.
3667 */
3668 lttng_dynamic_pointer_array_clear(
3669 &client->communication.outbound.payload
3670 ._fd_handles);
3671 }
3672
3673 end:
3674 if (status == CLIENT_TRANSMISSION_STATUS_COMPLETE) {
3675 client->communication.outbound.queued_command_reply = false;
3676 client->communication.outbound.dropped_notification = false;
3677 lttng_payload_clear(&client->communication.outbound.payload);
3678 }
3679
3680 return status;
3681 error:
3682 return CLIENT_TRANSMISSION_STATUS_ERROR;
3683 }
3684
3685 static
3686 bool client_has_outbound_data_left(
3687 const struct notification_client *client)
3688 {
3689 const struct lttng_payload_view pv = lttng_payload_view_from_payload(
3690 &client->communication.outbound.payload, 0, -1);
3691 const bool has_data = pv.buffer.size != 0;
3692 const bool has_fds = lttng_payload_view_get_fd_handle_count(&pv);
3693
3694 return has_data || has_fds;
3695 }
3696
3697 /* Client lock must _not_ be held by the caller. */
3698 static
3699 int client_send_command_reply(struct notification_client *client,
3700 struct notification_thread_state *state,
3701 enum lttng_notification_channel_status status)
3702 {
3703 int ret;
3704 struct lttng_notification_channel_command_reply reply = {
3705 .status = (int8_t) status,
3706 };
3707 struct lttng_notification_channel_message msg = {
3708 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
3709 .size = sizeof(reply),
3710 .fds = 0,
3711 };
3712 char buffer[sizeof(msg) + sizeof(reply)];
3713 enum client_transmission_status transmission_status;
3714
3715 memcpy(buffer, &msg, sizeof(msg));
3716 memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
3717 DBG("Send command reply (%i)", (int) status);
3718
3719 pthread_mutex_lock(&client->lock);
3720 if (client->communication.outbound.queued_command_reply) {
3721 /* Protocol error. */
3722 goto error_unlock;
3723 }
3724
3725 /* Enqueue buffer to outgoing queue and flush it. */
3726 ret = lttng_dynamic_buffer_append(
3727 &client->communication.outbound.payload.buffer,
3728 buffer, sizeof(buffer));
3729 if (ret) {
3730 goto error_unlock;
3731 }
3732
3733 transmission_status = client_flush_outgoing_queue(client);
3734
3735 if (client_has_outbound_data_left(client)) {
3736 /* Queue could not be emptied. */
3737 client->communication.outbound.queued_command_reply = true;
3738 }
3739
3740 pthread_mutex_unlock(&client->lock);
3741 ret = client_handle_transmission_status(
3742 client, transmission_status, state);
3743 if (ret) {
3744 goto error;
3745 }
3746
3747 return 0;
3748 error_unlock:
3749 pthread_mutex_unlock(&client->lock);
3750 error:
3751 return -1;
3752 }
3753
3754 static
3755 int client_handle_message_unknown(struct notification_client *client,
3756 struct notification_thread_state *state __attribute__((unused)))
3757 {
3758 int ret;
3759 /*
3760 * Receiving message header. The function will be called again
3761 * once the rest of the message as been received and can be
3762 * interpreted.
3763 */
3764 const struct lttng_notification_channel_message *msg;
3765
3766 LTTNG_ASSERT(sizeof(*msg) == client->communication.inbound.payload.buffer.size);
3767 msg = (const struct lttng_notification_channel_message *)
3768 client->communication.inbound.payload.buffer.data;
3769
3770 if (msg->size == 0 ||
3771 msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
3772 ERR("Invalid notification channel message: length = %u",
3773 msg->size);
3774 ret = -1;
3775 goto end;
3776 }
3777
3778 switch (msg->type) {
3779 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
3780 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
3781 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
3782 break;
3783 default:
3784 ret = -1;
3785 ERR("Invalid notification channel message: unexpected message type");
3786 goto end;
3787 }
3788
3789 client->communication.inbound.bytes_to_receive = msg->size;
3790 client->communication.inbound.fds_to_receive = msg->fds;
3791 client->communication.inbound.msg_type =
3792 (enum lttng_notification_channel_message_type) msg->type;
3793 ret = lttng_dynamic_buffer_set_size(
3794 &client->communication.inbound.payload.buffer, msg->size);
3795
3796 /* msg is not valid anymore due to lttng_dynamic_buffer_set_size. */
3797 msg = NULL;
3798 end:
3799 return ret;
3800 }
3801
3802 static
3803 int client_handle_message_handshake(struct notification_client *client,
3804 struct notification_thread_state *state)
3805 {
3806 int ret;
3807 struct lttng_notification_channel_command_handshake *handshake_client;
3808 const struct lttng_notification_channel_command_handshake handshake_reply = {
3809 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
3810 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
3811 };
3812 const struct lttng_notification_channel_message msg_header = {
3813 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
3814 .size = sizeof(handshake_reply),
3815 .fds = 0,
3816 };
3817 enum lttng_notification_channel_status status =
3818 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
3819 char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
3820
3821 memcpy(send_buffer, &msg_header, sizeof(msg_header));
3822 memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
3823 sizeof(handshake_reply));
3824
3825 handshake_client =
3826 (struct lttng_notification_channel_command_handshake *)
3827 client->communication.inbound.payload.buffer
3828 .data;
3829 client->major = handshake_client->major;
3830 client->minor = handshake_client->minor;
3831 if (!client->communication.inbound.creds_received) {
3832 ERR("No credentials received from client");
3833 ret = -1;
3834 goto end;
3835 }
3836
3837 client->uid = LTTNG_SOCK_GET_UID_CRED(
3838 &client->communication.inbound.creds);
3839 client->gid = LTTNG_SOCK_GET_GID_CRED(
3840 &client->communication.inbound.creds);
3841 client->is_sessiond = LTTNG_SOCK_GET_PID_CRED(&client->communication.inbound.creds) == getpid();
3842 DBG("Received handshake from client: uid = %u, gid = %u, protocol version = %i.%i, client is sessiond = %s",
3843 client->uid, client->gid, (int) client->major,
3844 (int) client->minor,
3845 client->is_sessiond ? "true" : "false");
3846
3847 if (handshake_client->major !=
3848 LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
3849 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION;
3850 }
3851
3852 pthread_mutex_lock(&client->lock);
3853 /* Outgoing queue will be flushed when the command reply is sent. */
3854 ret = lttng_dynamic_buffer_append(
3855 &client->communication.outbound.payload.buffer, send_buffer,
3856 sizeof(send_buffer));
3857 if (ret) {
3858 ERR("Failed to send protocol version to notification channel client");
3859 goto end_unlock;
3860 }
3861
3862 client->validated = true;
3863 client->communication.active = true;
3864 pthread_mutex_unlock(&client->lock);
3865
3866 /* Set reception state to receive the next message header. */
3867 ret = client_reset_inbound_state(client);
3868 if (ret) {
3869 ERR("Failed to reset client communication's inbound state");
3870 goto end;
3871 }
3872
3873 /* Flushes the outgoing queue. */
3874 ret = client_send_command_reply(client, state, status);
3875 if (ret) {
3876 ERR("Failed to send reply to notification channel client");
3877 goto end;
3878 }
3879
3880 goto end;
3881 end_unlock:
3882 pthread_mutex_unlock(&client->lock);
3883 end:
3884 return ret;
3885 }
3886
3887 static
3888 int client_handle_message_subscription(
3889 struct notification_client *client,
3890 enum lttng_notification_channel_message_type msg_type,
3891 struct notification_thread_state *state)
3892 {
3893 int ret;
3894 struct lttng_condition *condition;
3895 enum lttng_notification_channel_status status =
3896 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
3897 struct lttng_payload_view condition_view =
3898 lttng_payload_view_from_payload(
3899 &client->communication.inbound.payload,
3900 0, -1);
3901 size_t expected_condition_size;
3902
3903 /*
3904 * No need to lock client to sample the inbound state as the only
3905 * other thread accessing clients (action executor) only uses the
3906 * outbound state.
3907 */
3908 expected_condition_size = client->communication.inbound.payload.buffer.size;
3909 ret = lttng_condition_create_from_payload(&condition_view, &condition);
3910 if (ret != expected_condition_size) {
3911 ERR("Malformed condition received from client");
3912 goto end;
3913 }
3914
3915 /* Ownership of condition is always transferred. */
3916 if (msg_type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE) {
3917 ret = notification_thread_client_subscribe(
3918 client, condition, state, &status);
3919 } else {
3920 ret = notification_thread_client_unsubscribe(
3921 client, condition, state, &status);
3922 }
3923
3924 if (ret) {
3925 goto end;
3926 }
3927
3928 /* Set reception state to receive the next message header. */
3929 ret = client_reset_inbound_state(client);
3930 if (ret) {
3931 ERR("Failed to reset client communication's inbound state");
3932 goto end;
3933 }
3934
3935 ret = client_send_command_reply(client, state, status);
3936 if (ret) {
3937 ERR("Failed to send reply to notification channel client");
3938 goto end;
3939 }
3940
3941 end:
3942 return ret;
3943 }
3944
3945 static
3946 int client_dispatch_message(struct notification_client *client,
3947 struct notification_thread_state *state)
3948 {
3949 int ret = 0;
3950
3951 if (client->communication.inbound.msg_type !=
3952 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE &&
3953 client->communication.inbound.msg_type !=
3954 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN &&
3955 !client->validated) {
3956 WARN("client attempted a command before handshake");
3957 ret = -1;
3958 goto end;
3959 }
3960
3961 switch (client->communication.inbound.msg_type) {
3962 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN:
3963 {
3964 ret = client_handle_message_unknown(client, state);
3965 break;
3966 }
3967 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
3968 {
3969 ret = client_handle_message_handshake(client, state);
3970 break;
3971 }
3972 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
3973 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
3974 {
3975 ret = client_handle_message_subscription(client,
3976 client->communication.inbound.msg_type, state);
3977 break;
3978 }
3979 default:
3980 abort();
3981 }
3982 end:
3983 return ret;
3984 }
3985
3986 /* Incoming data from client. */
3987 int handle_notification_thread_client_in(
3988 struct notification_thread_state *state, int socket)
3989 {
3990 int ret = 0;
3991 struct notification_client *client;
3992 ssize_t recv_ret;
3993 size_t offset;
3994
3995 rcu_read_lock();
3996 client = get_client_from_socket(socket, state);
3997 if (!client) {
3998 /* Internal error, abort. */
3999 ret = -1;
4000 goto end;
4001 }
4002
4003 if (client->communication.inbound.bytes_to_receive == 0 &&
4004 client->communication.inbound.fds_to_receive != 0) {
4005 /* Only FDs left to receive. */
4006 goto receive_fds;
4007 }
4008
4009 offset = client->communication.inbound.payload.buffer.size -
4010 client->communication.inbound.bytes_to_receive;
4011 if (client->communication.inbound.expect_creds) {
4012 recv_ret = lttcomm_recv_creds_unix_sock(socket,
4013 client->communication.inbound.payload.buffer.data + offset,
4014 client->communication.inbound.bytes_to_receive,
4015 &client->communication.inbound.creds);
4016 if (recv_ret > 0) {
4017 client->communication.inbound.expect_creds = false;
4018 client->communication.inbound.creds_received = true;
4019 }
4020 } else {
4021 recv_ret = lttcomm_recv_unix_sock_non_block(socket,
4022 client->communication.inbound.payload.buffer.data + offset,
4023 client->communication.inbound.bytes_to_receive);
4024 }
4025 if (recv_ret >= 0) {
4026 client->communication.inbound.bytes_to_receive -= recv_ret;
4027 } else {
4028 goto error_disconnect_client;
4029 }
4030
4031 if (client->communication.inbound.bytes_to_receive != 0) {
4032 /* Message incomplete wait for more data. */
4033 ret = 0;
4034 goto end;
4035 }
4036
4037 receive_fds:
4038 LTTNG_ASSERT(client->communication.inbound.bytes_to_receive == 0);
4039
4040 /* Receive fds. */
4041 if (client->communication.inbound.fds_to_receive != 0) {
4042 ret = lttcomm_recv_payload_fds_unix_sock_non_block(
4043 client->socket,
4044 client->communication.inbound.fds_to_receive,
4045 &client->communication.inbound.payload);
4046 if (ret > 0) {
4047 /*
4048 * Fds received. non blocking fds passing is all
4049 * or nothing.
4050 */
4051 ssize_t expected_size;
4052
4053 expected_size = sizeof(int) *
4054 client->communication.inbound
4055 .fds_to_receive;
4056 LTTNG_ASSERT(ret == expected_size);
4057 client->communication.inbound.fds_to_receive = 0;
4058 } else if (ret == 0) {
4059 /* Received nothing. */
4060 ret = 0;
4061 goto end;
4062 } else {
4063 goto error_disconnect_client;
4064 }
4065 }
4066
4067 /* At this point the message is complete.*/
4068 LTTNG_ASSERT(client->communication.inbound.bytes_to_receive == 0 &&
4069 client->communication.inbound.fds_to_receive == 0);
4070 ret = client_dispatch_message(client, state);
4071 if (ret) {
4072 /*
4073 * Only returns an error if this client must be
4074 * disconnected.
4075 */
4076 goto error_disconnect_client;
4077 }
4078
4079 end:
4080 rcu_read_unlock();
4081 return ret;
4082
4083 error_disconnect_client:
4084 ret = notification_thread_client_disconnect(client, state);
4085 goto end;
4086 }
4087
4088 /* Client ready to receive outgoing data. */
4089 int handle_notification_thread_client_out(
4090 struct notification_thread_state *state, int socket)
4091 {
4092 int ret;
4093 struct notification_client *client;
4094 enum client_transmission_status transmission_status;
4095
4096 rcu_read_lock();
4097 client = get_client_from_socket(socket, state);
4098 if (!client) {
4099 /* Internal error, abort. */
4100 ret = -1;
4101 goto end;
4102 }
4103
4104 pthread_mutex_lock(&client->lock);
4105 transmission_status = client_flush_outgoing_queue(client);
4106 pthread_mutex_unlock(&client->lock);
4107
4108 ret = client_handle_transmission_status(
4109 client, transmission_status, state);
4110 if (ret) {
4111 goto end;
4112 }
4113 end:
4114 rcu_read_unlock();
4115 return ret;
4116 }
4117
4118 static
4119 bool evaluate_buffer_usage_condition(const struct lttng_condition *condition,
4120 const struct channel_state_sample *sample,
4121 uint64_t buffer_capacity)
4122 {
4123 bool result = false;
4124 uint64_t threshold;
4125 enum lttng_condition_type condition_type;
4126 const struct lttng_condition_buffer_usage *use_condition = container_of(
4127 condition, struct lttng_condition_buffer_usage,
4128 parent);
4129
4130 if (use_condition->threshold_bytes.set) {
4131 threshold = use_condition->threshold_bytes.value;
4132 } else {
4133 /*
4134 * Threshold was expressed as a ratio.
4135 *
4136 * TODO the threshold (in bytes) of conditions expressed
4137 * as a ratio of total buffer size could be cached to
4138 * forego this double-multiplication or it could be performed
4139 * as fixed-point math.
4140 *
4141 * Note that caching should accommodates the case where the
4142 * condition applies to multiple channels (i.e. don't assume
4143 * that all channels matching my_chann* have the same size...)
4144 */
4145 threshold = (uint64_t) (use_condition->threshold_ratio.value *
4146 (double) buffer_capacity);
4147 }
4148
4149 condition_type = lttng_condition_get_type(condition);
4150 if (condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW) {
4151 DBG("Low buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
4152 threshold, sample->highest_usage);
4153
4154 /*
4155 * The low condition should only be triggered once _all_ of the
4156 * streams in a channel have gone below the "low" threshold.
4157 */
4158 if (sample->highest_usage <= threshold) {
4159 result = true;
4160 }
4161 } else {
4162 DBG("High buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
4163 threshold, sample->highest_usage);
4164
4165 /*
4166 * For high buffer usage scenarios, we want to trigger whenever
4167 * _any_ of the streams has reached the "high" threshold.
4168 */
4169 if (sample->highest_usage >= threshold) {
4170 result = true;
4171 }
4172 }
4173
4174 return result;
4175 }
4176
4177 static
4178 bool evaluate_session_consumed_size_condition(
4179 const struct lttng_condition *condition,
4180 uint64_t session_consumed_size)
4181 {
4182 uint64_t threshold;
4183 const struct lttng_condition_session_consumed_size *size_condition =
4184 container_of(condition,
4185 struct lttng_condition_session_consumed_size,
4186 parent);
4187
4188 threshold = size_condition->consumed_threshold_bytes.value;
4189 DBG("Session consumed size condition being evaluated: threshold = %" PRIu64 ", current size = %" PRIu64,
4190 threshold, session_consumed_size);
4191 return session_consumed_size >= threshold;
4192 }
4193
4194 static
4195 int evaluate_buffer_condition(const struct lttng_condition *condition,
4196 struct lttng_evaluation **evaluation,
4197 const struct notification_thread_state *state __attribute__((unused)),
4198 const struct channel_state_sample *previous_sample,
4199 const struct channel_state_sample *latest_sample,
4200 uint64_t previous_session_consumed_total,
4201 uint64_t latest_session_consumed_total,
4202 struct channel_info *channel_info)
4203 {
4204 int ret = 0;
4205 enum lttng_condition_type condition_type;
4206 const bool previous_sample_available = !!previous_sample;
4207 bool previous_sample_result = false;
4208 bool latest_sample_result;
4209
4210 condition_type = lttng_condition_get_type(condition);
4211
4212 switch (condition_type) {
4213 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
4214 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
4215 if (caa_likely(previous_sample_available)) {
4216 previous_sample_result =
4217 evaluate_buffer_usage_condition(condition,
4218 previous_sample, channel_info->capacity);
4219 }
4220 latest_sample_result = evaluate_buffer_usage_condition(
4221 condition, latest_sample,
4222 channel_info->capacity);
4223 break;
4224 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
4225 if (caa_likely(previous_sample_available)) {
4226 previous_sample_result =
4227 evaluate_session_consumed_size_condition(
4228 condition,
4229 previous_session_consumed_total);
4230 }
4231 latest_sample_result =
4232 evaluate_session_consumed_size_condition(
4233 condition,
4234 latest_session_consumed_total);
4235 break;
4236 default:
4237 /* Unknown condition type; internal error. */
4238 abort();
4239 }
4240
4241 if (!latest_sample_result ||
4242 (previous_sample_result == latest_sample_result)) {
4243 /*
4244 * Only trigger on a condition evaluation transition.
4245 *
4246 * NOTE: This edge-triggered logic may not be appropriate for
4247 * future condition types.
4248 */
4249 goto end;
4250 }
4251
4252 if (!evaluation || !latest_sample_result) {
4253 goto end;
4254 }
4255
4256 switch (condition_type) {
4257 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
4258 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
4259 *evaluation = lttng_evaluation_buffer_usage_create(
4260 condition_type,
4261 latest_sample->highest_usage,
4262 channel_info->capacity);
4263 break;
4264 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
4265 *evaluation = lttng_evaluation_session_consumed_size_create(
4266 latest_session_consumed_total);
4267 break;
4268 default:
4269 abort();
4270 }
4271
4272 if (!*evaluation) {
4273 ret = -1;
4274 goto end;
4275 }
4276 end:
4277 return ret;
4278 }
4279
4280 static
4281 int client_notification_overflow(struct notification_client *client)
4282 {
4283 int ret = 0;
4284 const struct lttng_notification_channel_message msg = {
4285 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED,
4286 .size = 0,
4287 .fds = 0,
4288 };
4289
4290 ASSERT_LOCKED(client->lock);
4291
4292 DBG("Dropping notification addressed to client (socket fd = %i)",
4293 client->socket);
4294 if (client->communication.outbound.dropped_notification) {
4295 /*
4296 * The client already has a "notification dropped" message
4297 * in its outgoing queue. Nothing to do since all
4298 * of those messages are coalesced.
4299 */
4300 goto end;
4301 }
4302
4303 client->communication.outbound.dropped_notification = true;
4304 ret = lttng_dynamic_buffer_append(
4305 &client->communication.outbound.payload.buffer, &msg,
4306 sizeof(msg));
4307 if (ret) {
4308 PERROR("Failed to enqueue \"dropped notification\" message in client's (socket fd = %i) outgoing queue",
4309 client->socket);
4310 }
4311 end:
4312 return ret;
4313 }
4314
4315 static int client_handle_transmission_status_wrapper(
4316 struct notification_client *client,
4317 enum client_transmission_status status,
4318 void *user_data)
4319 {
4320 return client_handle_transmission_status(client, status,
4321 (struct notification_thread_state *) user_data);
4322 }
4323
4324 static
4325 int send_evaluation_to_clients(const struct lttng_trigger *trigger,
4326 const struct lttng_evaluation *evaluation,
4327 struct notification_client_list* client_list,
4328 struct notification_thread_state *state,
4329 uid_t object_uid, gid_t object_gid)
4330 {
4331 const struct lttng_credentials creds = {
4332 .uid = LTTNG_OPTIONAL_INIT_VALUE(object_uid),
4333 .gid = LTTNG_OPTIONAL_INIT_VALUE(object_gid),
4334 };
4335
4336 return notification_client_list_send_evaluation(client_list,
4337 trigger, evaluation,
4338 &creds,
4339 client_handle_transmission_status_wrapper, state);
4340 }
4341
4342 /*
4343 * Permission checks relative to notification channel clients are performed
4344 * here. Notice how object, client, and trigger credentials are involved in
4345 * this check.
4346 *
4347 * The `object` credentials are the credentials associated with the "subject"
4348 * of a condition. For instance, a `rotation completed` condition applies
4349 * to a session. When that condition is met, it will produce an evaluation
4350 * against a session. Hence, in this case, the `object` credentials are the
4351 * credentials of the "subject" session.
4352 *
4353 * The `trigger` credentials are the credentials of the user that registered the
4354 * trigger.
4355 *
4356 * The `client` credentials are the credentials of the user that created a given
4357 * notification channel.
4358 *
4359 * In terms of visibility, it is expected that non-privilieged users can only
4360 * register triggers against "their" objects (their own sessions and
4361 * applications they are allowed to interact with). They can then open a
4362 * notification channel and subscribe to notifications associated with those
4363 * triggers.
4364 *
4365 * As for privilieged users, they can register triggers against the objects of
4366 * other users. They can then subscribe to the notifications associated to their
4367 * triggers. Privilieged users _can't_ subscribe to the notifications of
4368 * triggers owned by other users; they must create their own triggers.
4369 *
4370 * This is more a concern of usability than security. It would be difficult for
4371 * a root user reliably subscribe to a specific set of conditions without
4372 * interference from external users (those could, for instance, unregister
4373 * their triggers).
4374 */
4375 int notification_client_list_send_evaluation(
4376 struct notification_client_list *client_list,
4377 const struct lttng_trigger *trigger,
4378 const struct lttng_evaluation *evaluation,
4379 const struct lttng_credentials *source_object_creds,
4380 report_client_transmission_result_cb client_report,
4381 void *user_data)
4382 {
4383 int ret = 0;
4384 struct lttng_payload msg_payload;
4385 struct notification_client_list_element *client_list_element, *tmp;
4386 const struct lttng_notification notification = {
4387 .trigger = (struct lttng_trigger *) trigger,
4388 .evaluation = (struct lttng_evaluation *) evaluation,
4389 };
4390 struct lttng_notification_channel_message msg_header = {
4391 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION,
4392 .size = 0,
4393 .fds = 0,
4394 };
4395 const struct lttng_credentials *trigger_creds =
4396 lttng_trigger_get_credentials(trigger);
4397
4398 lttng_payload_init(&msg_payload);
4399
4400 ret = lttng_dynamic_buffer_append(&msg_payload.buffer, &msg_header,
4401 sizeof(msg_header));
4402 if (ret) {
4403 goto end;
4404 }
4405
4406 ret = lttng_notification_serialize(&notification, &msg_payload);
4407 if (ret) {
4408 ERR("Failed to serialize notification");
4409 ret = -1;
4410 goto end;
4411 }
4412
4413 /* Update payload size. */
4414 ((struct lttng_notification_channel_message *) msg_payload.buffer.data)
4415 ->size = (uint32_t)(
4416 msg_payload.buffer.size - sizeof(msg_header));
4417
4418 /* Update the payload number of fds. */
4419 {
4420 const struct lttng_payload_view pv = lttng_payload_view_from_payload(
4421 &msg_payload, 0, -1);
4422
4423 ((struct lttng_notification_channel_message *)
4424 msg_payload.buffer.data)->fds = (uint32_t)
4425 lttng_payload_view_get_fd_handle_count(&pv);
4426 }
4427
4428 pthread_mutex_lock(&client_list->lock);
4429 cds_list_for_each_entry_safe(client_list_element, tmp,
4430 &client_list->clients_list, node) {
4431 enum client_transmission_status transmission_status;
4432 struct notification_client *client =
4433 client_list_element->client;
4434
4435 ret = 0;
4436 pthread_mutex_lock(&client->lock);
4437 if (!client->communication.active) {
4438 /*
4439 * Skip inactive client (protocol error or
4440 * disconnecting).
4441 */
4442 DBG("Skipping client at it is marked as inactive");
4443 goto skip_client;
4444 }
4445
4446 if (lttng_trigger_is_hidden(trigger) && !client->is_sessiond) {
4447 /*
4448 * Notifications resulting from an hidden trigger are
4449 * only sent to the session daemon.
4450 */
4451 DBG("Skipping client as the trigger is hidden and the client is not the session daemon");
4452 goto skip_client;
4453 }
4454
4455 if (source_object_creds) {
4456 if (client->uid != lttng_credentials_get_uid(source_object_creds) &&
4457 client->gid != lttng_credentials_get_gid(source_object_creds) &&
4458 client->uid != 0) {
4459 /*
4460 * Client is not allowed to monitor this
4461 * object.
4462 */
4463 DBG("Skipping client at it does not have the object permission to receive notification for this trigger");
4464 goto skip_client;
4465 }
4466 }
4467
4468 if (client->uid != lttng_credentials_get_uid(trigger_creds)) {
4469 DBG("Skipping client at it does not have the permission to receive notification for this trigger");
4470 goto skip_client;
4471 }
4472
4473 DBG("Sending notification to client (fd = %i, %zu bytes)",
4474 client->socket, msg_payload.buffer.size);
4475
4476 if (client_has_outbound_data_left(client)) {
4477 /*
4478 * Outgoing data is already buffered for this client;
4479 * drop the notification and enqueue a "dropped
4480 * notification" message if this is the first dropped
4481 * notification since the socket spilled-over to the
4482 * queue.
4483 */
4484 ret = client_notification_overflow(client);
4485 if (ret) {
4486 /* Fatal error. */
4487 goto skip_client;
4488 }
4489 }
4490
4491 ret = lttng_payload_copy(&msg_payload, &client->communication.outbound.payload);
4492 if (ret) {
4493 /* Fatal error. */
4494 goto skip_client;
4495 }
4496
4497 transmission_status = client_flush_outgoing_queue(client);
4498 pthread_mutex_unlock(&client->lock);
4499 ret = client_report(client, transmission_status, user_data);
4500 if (ret) {
4501 /* Fatal error. */
4502 goto end_unlock_list;
4503 }
4504
4505 continue;
4506
4507 skip_client:
4508 pthread_mutex_unlock(&client->lock);
4509 if (ret) {
4510 /* Fatal error. */
4511 goto end_unlock_list;
4512 }
4513 }
4514 ret = 0;
4515
4516 end_unlock_list:
4517 pthread_mutex_unlock(&client_list->lock);
4518 end:
4519 lttng_payload_reset(&msg_payload);
4520 return ret;
4521 }
4522
4523 static
4524 struct lttng_event_notifier_notification *recv_one_event_notifier_notification(
4525 int notification_pipe_read_fd, enum lttng_domain_type domain)
4526 {
4527 int ret;
4528 uint64_t token;
4529 struct lttng_event_notifier_notification *notification = NULL;
4530 char *capture_buffer = NULL;
4531 size_t capture_buffer_size;
4532 void *reception_buffer;
4533 size_t reception_size;
4534
4535 struct lttng_ust_abi_event_notifier_notification ust_notification;
4536 struct lttng_kernel_abi_event_notifier_notification kernel_notification;
4537
4538 /* Init lttng_event_notifier_notification */
4539 switch(domain) {
4540 case LTTNG_DOMAIN_UST:
4541 reception_buffer = (void *) &ust_notification;
4542 reception_size = sizeof(ust_notification);
4543 break;
4544 case LTTNG_DOMAIN_KERNEL:
4545 reception_buffer = (void *) &kernel_notification;
4546 reception_size = sizeof(kernel_notification);
4547 break;
4548 default:
4549 abort();
4550 }
4551
4552 /*
4553 * The monitoring pipe only holds messages smaller than PIPE_BUF,
4554 * ensuring that read/write of tracer notifications are atomic.
4555 */
4556 ret = lttng_read(notification_pipe_read_fd, reception_buffer,
4557 reception_size);
4558 if (ret != reception_size) {
4559 PERROR("Failed to read from event source notification pipe: fd = %d, size to read = %zu, ret = %d",
4560 notification_pipe_read_fd, reception_size, ret);
4561 ret = -1;
4562 goto end;
4563 }
4564
4565 switch(domain) {
4566 case LTTNG_DOMAIN_UST:
4567 token = ust_notification.token;
4568 capture_buffer_size = ust_notification.capture_buf_size;
4569 break;
4570 case LTTNG_DOMAIN_KERNEL:
4571 token = kernel_notification.token;
4572 capture_buffer_size = kernel_notification.capture_buf_size;
4573 break;
4574 default:
4575 abort();
4576 }
4577
4578 if (capture_buffer_size == 0) {
4579 capture_buffer = NULL;
4580 goto skip_capture;
4581 }
4582
4583 if (capture_buffer_size > MAX_CAPTURE_SIZE) {
4584 ERR("Event notifier has a capture payload size which exceeds the maximum allowed size: capture_payload_size = %zu bytes, max allowed size = %d bytes",
4585 capture_buffer_size, MAX_CAPTURE_SIZE);
4586 goto end;
4587 }
4588
4589 capture_buffer = (char *) zmalloc(capture_buffer_size);
4590 if (!capture_buffer) {
4591 ERR("Failed to allocate capture buffer");
4592 goto end;
4593 }
4594
4595 /* Fetch additional payload (capture). */
4596 ret = lttng_read(notification_pipe_read_fd, capture_buffer, capture_buffer_size);
4597 if (ret != capture_buffer_size) {
4598 ERR("Failed to read from event source pipe (fd = %i)",
4599 notification_pipe_read_fd);
4600 goto end;
4601 }
4602
4603 skip_capture:
4604 notification = lttng_event_notifier_notification_create(token, domain,
4605 capture_buffer, capture_buffer_size);
4606 if (notification == NULL) {
4607 goto end;
4608 }
4609
4610 /*
4611 * Ownership transfered to the lttng_event_notifier_notification object.
4612 */
4613 capture_buffer = NULL;
4614
4615 end:
4616 free(capture_buffer);
4617 return notification;
4618 }
4619
4620 static
4621 int dispatch_one_event_notifier_notification(struct notification_thread_state *state,
4622 struct lttng_event_notifier_notification *notification)
4623 {
4624 struct cds_lfht_node *node;
4625 struct cds_lfht_iter iter;
4626 struct notification_trigger_tokens_ht_element *element;
4627 struct lttng_evaluation *evaluation = NULL;
4628 enum action_executor_status executor_status;
4629 struct notification_client_list *client_list = NULL;
4630 int ret;
4631 unsigned int capture_count = 0;
4632
4633 /* Find triggers associated with this token. */
4634 rcu_read_lock();
4635 cds_lfht_lookup(state->trigger_tokens_ht,
4636 hash_key_u64(&notification->tracer_token, lttng_ht_seed),
4637 match_trigger_token, &notification->tracer_token, &iter);
4638 node = cds_lfht_iter_get_node(&iter);
4639 if (caa_unlikely(!node)) {
4640 /*
4641 * This is not an error, slow consumption of the tracer
4642 * notifications can lead to situations where a trigger is
4643 * removed but we still get tracer notifications matching a
4644 * trigger that no longer exists.
4645 */
4646 ret = 0;
4647 goto end_unlock;
4648 }
4649
4650 element = caa_container_of(node,
4651 struct notification_trigger_tokens_ht_element,
4652 node);
4653
4654 if (lttng_condition_event_rule_matches_get_capture_descriptor_count(
4655 lttng_trigger_get_const_condition(element->trigger),
4656 &capture_count) != LTTNG_CONDITION_STATUS_OK) {
4657 ERR("Failed to get capture count");
4658 ret = -1;
4659 goto end;
4660 }
4661
4662 if (!notification->capture_buffer && capture_count != 0) {
4663 ERR("Expected capture but capture buffer is null");
4664 ret = -1;
4665 goto end;
4666 }
4667
4668 evaluation = lttng_evaluation_event_rule_matches_create(
4669 container_of(lttng_trigger_get_const_condition(
4670 element->trigger),
4671 struct lttng_condition_event_rule_matches,
4672 parent),
4673 notification->capture_buffer,
4674 notification->capture_buf_size, false);
4675
4676 if (evaluation == NULL) {
4677 ERR("Failed to create event rule matches evaluation while creating and enqueuing action executor job");
4678 ret = -1;
4679 goto end_unlock;
4680 }
4681
4682 client_list = get_client_list_from_condition(state,
4683 lttng_trigger_get_const_condition(element->trigger));
4684 executor_status = action_executor_enqueue_trigger(state->executor,
4685 element->trigger, evaluation, NULL, client_list);
4686 switch (executor_status) {
4687 case ACTION_EXECUTOR_STATUS_OK:
4688 ret = 0;
4689 break;
4690 case ACTION_EXECUTOR_STATUS_OVERFLOW:
4691 {
4692 struct notification_client_list_element *client_list_element,
4693 *tmp;
4694
4695 /*
4696 * Not a fatal error; this is expected and simply means the
4697 * executor has too much work queued already.
4698 */
4699 ret = 0;
4700
4701 /* No clients subscribed to notifications for this trigger. */
4702 if (!client_list) {
4703 break;
4704 }
4705
4706 /* Warn clients that a notification (or more) was dropped. */
4707 pthread_mutex_lock(&client_list->lock);
4708 cds_list_for_each_entry_safe(client_list_element, tmp,
4709 &client_list->clients_list, node) {
4710 enum client_transmission_status transmission_status;
4711 struct notification_client *client =
4712 client_list_element->client;
4713
4714 pthread_mutex_lock(&client->lock);
4715 ret = client_notification_overflow(client);
4716 if (ret) {
4717 /* Fatal error. */
4718 goto next_client;
4719 }
4720
4721 transmission_status =
4722 client_flush_outgoing_queue(client);
4723 ret = client_handle_transmission_status(
4724 client, transmission_status, state);
4725 if (ret) {
4726 /* Fatal error. */
4727 goto next_client;
4728 }
4729 next_client:
4730 pthread_mutex_unlock(&client->lock);
4731 if (ret) {
4732 break;
4733 }
4734 }
4735
4736 pthread_mutex_unlock(&client_list->lock);
4737 break;
4738 }
4739 case ACTION_EXECUTOR_STATUS_INVALID:
4740 case ACTION_EXECUTOR_STATUS_ERROR:
4741 /* Fatal error, shut down everything. */
4742 ERR("Fatal error encoutered while enqueuing action to the action executor");
4743 ret = -1;
4744 goto end_unlock;
4745 default:
4746 /* Unhandled error. */
4747 abort();
4748 }
4749
4750 end_unlock:
4751 notification_client_list_put(client_list);
4752 rcu_read_unlock();
4753 end:
4754 return ret;
4755 }
4756
4757 static
4758 int handle_one_event_notifier_notification(
4759 struct notification_thread_state *state,
4760 int pipe, enum lttng_domain_type domain)
4761 {
4762 int ret = 0;
4763 struct lttng_event_notifier_notification *notification = NULL;
4764
4765 notification = recv_one_event_notifier_notification(pipe, domain);
4766 if (notification == NULL) {
4767 /* Reception failed, don't consider it fatal. */
4768 ERR("Error receiving an event notifier notification from tracer: fd = %i, domain = %s",
4769 pipe, lttng_domain_type_str(domain));
4770 goto end;
4771 }
4772
4773 ret = dispatch_one_event_notifier_notification(state, notification);
4774 if (ret) {
4775 ERR("Error dispatching an event notifier notification from tracer: fd = %i, domain = %s",
4776 pipe, lttng_domain_type_str(domain));
4777 goto end;
4778 }
4779
4780 end:
4781 lttng_event_notifier_notification_destroy(notification);
4782 return ret;
4783 }
4784
4785 int handle_notification_thread_event_notification(struct notification_thread_state *state,
4786 int pipe, enum lttng_domain_type domain)
4787 {
4788 return handle_one_event_notifier_notification(state, pipe, domain);
4789 }
4790
4791 int handle_notification_thread_channel_sample(
4792 struct notification_thread_state *state, int pipe,
4793 enum lttng_domain_type domain)
4794 {
4795 int ret = 0;
4796 struct lttcomm_consumer_channel_monitor_msg sample_msg;
4797 struct channel_info *channel_info;
4798 struct cds_lfht_node *node;
4799 struct cds_lfht_iter iter;
4800 struct lttng_channel_trigger_list *trigger_list;
4801 struct lttng_trigger_list_element *trigger_list_element;
4802 bool previous_sample_available = false;
4803 struct channel_state_sample previous_sample, latest_sample;
4804 uint64_t previous_session_consumed_total, latest_session_consumed_total;
4805 struct lttng_credentials channel_creds;
4806
4807 /*
4808 * The monitoring pipe only holds messages smaller than PIPE_BUF,
4809 * ensuring that read/write of sampling messages are atomic.
4810 */
4811 ret = lttng_read(pipe, &sample_msg, sizeof(sample_msg));
4812 if (ret != sizeof(sample_msg)) {
4813 ERR("Failed to read from monitoring pipe (fd = %i)",
4814 pipe);
4815 ret = -1;
4816 goto end;
4817 }
4818
4819 ret = 0;
4820 latest_sample.key.key = sample_msg.key;
4821 latest_sample.key.domain = domain;
4822 latest_sample.highest_usage = sample_msg.highest;
4823 latest_sample.lowest_usage = sample_msg.lowest;
4824 latest_sample.channel_total_consumed = sample_msg.total_consumed;
4825
4826 rcu_read_lock();
4827
4828 /* Retrieve the channel's informations */
4829 cds_lfht_lookup(state->channels_ht,
4830 hash_channel_key(&latest_sample.key),
4831 match_channel_info,
4832 &latest_sample.key,
4833 &iter);
4834 node = cds_lfht_iter_get_node(&iter);
4835 if (caa_unlikely(!node)) {
4836 /*
4837 * Not an error since the consumer can push a sample to the pipe
4838 * and the rest of the session daemon could notify us of the
4839 * channel's destruction before we get a chance to process that
4840 * sample.
4841 */
4842 DBG("Received a sample for an unknown channel from consumerd, key = %" PRIu64 " in %s domain",
4843 latest_sample.key.key,
4844 lttng_domain_type_str(domain));
4845 goto end_unlock;
4846 }
4847 channel_info = caa_container_of(node, struct channel_info,
4848 channels_ht_node);
4849 DBG("Handling channel sample for channel %s (key = %" PRIu64 ") in session %s (highest usage = %" PRIu64 ", lowest usage = %" PRIu64", total consumed = %" PRIu64")",
4850 channel_info->name,
4851 latest_sample.key.key,
4852 channel_info->session_info->name,
4853 latest_sample.highest_usage,
4854 latest_sample.lowest_usage,
4855 latest_sample.channel_total_consumed);
4856
4857 previous_session_consumed_total =
4858 channel_info->session_info->consumed_data_size;
4859
4860 /* Retrieve the channel's last sample, if it exists, and update it. */
4861 cds_lfht_lookup(state->channel_state_ht,
4862 hash_channel_key(&latest_sample.key),
4863 match_channel_state_sample,
4864 &latest_sample.key,
4865 &iter);
4866 node = cds_lfht_iter_get_node(&iter);
4867 if (caa_likely(node)) {
4868 struct channel_state_sample *stored_sample;
4869
4870 /* Update the sample stored. */
4871 stored_sample = caa_container_of(node,
4872 struct channel_state_sample,
4873 channel_state_ht_node);
4874
4875 memcpy(&previous_sample, stored_sample,
4876 sizeof(previous_sample));
4877 stored_sample->highest_usage = latest_sample.highest_usage;
4878 stored_sample->lowest_usage = latest_sample.lowest_usage;
4879 stored_sample->channel_total_consumed = latest_sample.channel_total_consumed;
4880 previous_sample_available = true;
4881
4882 latest_session_consumed_total =
4883 previous_session_consumed_total +
4884 (latest_sample.channel_total_consumed - previous_sample.channel_total_consumed);
4885 } else {
4886 /*
4887 * This is the channel's first sample, allocate space for and
4888 * store the new sample.
4889 */
4890 struct channel_state_sample *stored_sample;
4891
4892 stored_sample = (channel_state_sample *) zmalloc(sizeof(*stored_sample));
4893 if (!stored_sample) {
4894 ret = -1;
4895 goto end_unlock;
4896 }
4897
4898 memcpy(stored_sample, &latest_sample, sizeof(*stored_sample));
4899 cds_lfht_node_init(&stored_sample->channel_state_ht_node);
4900 cds_lfht_add(state->channel_state_ht,
4901 hash_channel_key(&stored_sample->key),
4902 &stored_sample->channel_state_ht_node);
4903
4904 latest_session_consumed_total =
4905 previous_session_consumed_total +
4906 latest_sample.channel_total_consumed;
4907 }
4908
4909 channel_info->session_info->consumed_data_size =
4910 latest_session_consumed_total;
4911
4912 /* Find triggers associated with this channel. */
4913 cds_lfht_lookup(state->channel_triggers_ht,
4914 hash_channel_key(&latest_sample.key),
4915 match_channel_trigger_list,
4916 &latest_sample.key,
4917 &iter);
4918 node = cds_lfht_iter_get_node(&iter);
4919 if (caa_likely(!node)) {
4920 goto end_unlock;
4921 }
4922
4923 channel_creds = (typeof(channel_creds)) {
4924 .uid = LTTNG_OPTIONAL_INIT_VALUE(channel_info->session_info->uid),
4925 .gid = LTTNG_OPTIONAL_INIT_VALUE(channel_info->session_info->gid),
4926 };
4927
4928 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
4929 channel_triggers_ht_node);
4930 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
4931 node) {
4932 const struct lttng_condition *condition;
4933 struct lttng_trigger *trigger;
4934 struct notification_client_list *client_list = NULL;
4935 struct lttng_evaluation *evaluation = NULL;
4936 enum action_executor_status executor_status;
4937
4938 ret = 0;
4939 trigger = trigger_list_element->trigger;
4940 condition = lttng_trigger_get_const_condition(trigger);
4941 LTTNG_ASSERT(condition);
4942
4943 /*
4944 * Check if any client is subscribed to the result of this
4945 * evaluation.
4946 */
4947 client_list = get_client_list_from_condition(state, condition);
4948
4949 ret = evaluate_buffer_condition(condition, &evaluation, state,
4950 previous_sample_available ? &previous_sample : NULL,
4951 &latest_sample,
4952 previous_session_consumed_total,
4953 latest_session_consumed_total,
4954 channel_info);
4955 if (caa_unlikely(ret)) {
4956 goto put_list;
4957 }
4958
4959 if (caa_likely(!evaluation)) {
4960 goto put_list;
4961 }
4962
4963 /*
4964 * Ownership of `evaluation` transferred to the action executor
4965 * no matter the result.
4966 */
4967 executor_status = action_executor_enqueue_trigger(
4968 state->executor, trigger, evaluation,
4969 &channel_creds, client_list);
4970 evaluation = NULL;
4971 switch (executor_status) {
4972 case ACTION_EXECUTOR_STATUS_OK:
4973 break;
4974 case ACTION_EXECUTOR_STATUS_ERROR:
4975 case ACTION_EXECUTOR_STATUS_INVALID:
4976 /*
4977 * TODO Add trigger identification (name/id) when
4978 * it is added to the API.
4979 */
4980 ERR("Fatal error occurred while enqueuing action associated with buffer-condition trigger");
4981 ret = -1;
4982 goto put_list;
4983 case ACTION_EXECUTOR_STATUS_OVERFLOW:
4984 /*
4985 * TODO Add trigger identification (name/id) when
4986 * it is added to the API.
4987 *
4988 * Not a fatal error.
4989 */
4990 WARN("No space left when enqueuing action associated with buffer-condition trigger");
4991 ret = 0;
4992 goto put_list;
4993 default:
4994 abort();
4995 }
4996
4997 put_list:
4998 notification_client_list_put(client_list);
4999 if (caa_unlikely(ret)) {
5000 break;
5001 }
5002 }
5003 end_unlock:
5004 rcu_read_unlock();
5005 end:
5006 return ret;
5007 }
This page took 0.13124 seconds and 5 git commands to generate.