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