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