trigger: implement is_equal
[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/notification/notification-internal.h>
25 #include <lttng/condition/condition-internal.h>
26 #include <lttng/condition/buffer-usage-internal.h>
27 #include <lttng/condition/session-consumed-size-internal.h>
28 #include <lttng/condition/session-rotation-internal.h>
29 #include <lttng/notification/channel-internal.h>
30 #include <lttng/trigger/trigger-internal.h>
31
32 #include <time.h>
33 #include <unistd.h>
34 #include <assert.h>
35 #include <inttypes.h>
36 #include <fcntl.h>
37
38 #include "notification-thread.h"
39 #include "notification-thread-events.h"
40 #include "notification-thread-commands.h"
41 #include "lttng-sessiond.h"
42 #include "kernel.h"
43
44 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
45 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
46
47 enum lttng_object_type {
48 LTTNG_OBJECT_TYPE_UNKNOWN,
49 LTTNG_OBJECT_TYPE_NONE,
50 LTTNG_OBJECT_TYPE_CHANNEL,
51 LTTNG_OBJECT_TYPE_SESSION,
52 };
53
54 struct lttng_trigger_list_element {
55 /* No ownership of the trigger object is assumed. */
56 struct lttng_trigger *trigger;
57 struct cds_list_head node;
58 };
59
60 struct lttng_channel_trigger_list {
61 struct channel_key channel_key;
62 /* List of struct lttng_trigger_list_element. */
63 struct cds_list_head list;
64 /* Node in the channel_triggers_ht */
65 struct cds_lfht_node channel_triggers_ht_node;
66 /* call_rcu delayed reclaim. */
67 struct rcu_head rcu_node;
68 };
69
70 /*
71 * List of triggers applying to a given session.
72 *
73 * See:
74 * - lttng_session_trigger_list_create()
75 * - lttng_session_trigger_list_build()
76 * - lttng_session_trigger_list_destroy()
77 * - lttng_session_trigger_list_add()
78 */
79 struct lttng_session_trigger_list {
80 /*
81 * Not owned by this; points to the session_info structure's
82 * session name.
83 */
84 const char *session_name;
85 /* List of struct lttng_trigger_list_element. */
86 struct cds_list_head list;
87 /* Node in the session_triggers_ht */
88 struct cds_lfht_node session_triggers_ht_node;
89 /*
90 * Weak reference to the notification system's session triggers
91 * hashtable.
92 *
93 * The session trigger list structure structure is owned by
94 * the session's session_info.
95 *
96 * The session_info is kept alive the the channel_infos holding a
97 * reference to it (reference counting). When those channels are
98 * destroyed (at runtime or on teardown), the reference they hold
99 * to the session_info are released. On destruction of session_info,
100 * session_info_destroy() will remove the list of triggers applying
101 * to this session from the notification system's state.
102 *
103 * This implies that the session_triggers_ht must be destroyed
104 * after the channels.
105 */
106 struct cds_lfht *session_triggers_ht;
107 /* Used for delayed RCU reclaim. */
108 struct rcu_head rcu_node;
109 };
110
111 struct lttng_trigger_ht_element {
112 struct lttng_trigger *trigger;
113 struct cds_lfht_node node;
114 /* call_rcu delayed reclaim. */
115 struct rcu_head rcu_node;
116 };
117
118 struct lttng_condition_list_element {
119 struct lttng_condition *condition;
120 struct cds_list_head node;
121 };
122
123 struct channel_state_sample {
124 struct channel_key key;
125 struct cds_lfht_node channel_state_ht_node;
126 uint64_t highest_usage;
127 uint64_t lowest_usage;
128 uint64_t channel_total_consumed;
129 /* call_rcu delayed reclaim. */
130 struct rcu_head rcu_node;
131 };
132
133 static unsigned long hash_channel_key(struct channel_key *key);
134 static int evaluate_buffer_condition(const struct lttng_condition *condition,
135 struct lttng_evaluation **evaluation,
136 const struct notification_thread_state *state,
137 const struct channel_state_sample *previous_sample,
138 const struct channel_state_sample *latest_sample,
139 uint64_t previous_session_consumed_total,
140 uint64_t latest_session_consumed_total,
141 struct channel_info *channel_info);
142 static
143 int send_evaluation_to_clients(const struct lttng_trigger *trigger,
144 const struct lttng_evaluation *evaluation,
145 struct notification_client_list *client_list,
146 struct notification_thread_state *state,
147 uid_t channel_uid, gid_t channel_gid);
148
149
150 /* session_info API */
151 static
152 void session_info_destroy(void *_data);
153 static
154 void session_info_get(struct session_info *session_info);
155 static
156 void session_info_put(struct session_info *session_info);
157 static
158 struct session_info *session_info_create(const char *name,
159 uid_t uid, gid_t gid,
160 struct lttng_session_trigger_list *trigger_list,
161 struct cds_lfht *sessions_ht);
162 static
163 void session_info_add_channel(struct session_info *session_info,
164 struct channel_info *channel_info);
165 static
166 void session_info_remove_channel(struct session_info *session_info,
167 struct channel_info *channel_info);
168
169 /* lttng_session_trigger_list API */
170 static
171 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
172 const char *session_name,
173 struct cds_lfht *session_triggers_ht);
174 static
175 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
176 const struct notification_thread_state *state,
177 const char *session_name);
178 static
179 void lttng_session_trigger_list_destroy(
180 struct lttng_session_trigger_list *list);
181 static
182 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
183 struct lttng_trigger *trigger);
184
185 static
186 int client_handle_transmission_status(
187 struct notification_client *client,
188 enum client_transmission_status transmission_status,
189 struct notification_thread_state *state);
190
191 static
192 int match_client_socket(struct cds_lfht_node *node, const void *key)
193 {
194 /* This double-cast is intended to supress pointer-to-cast warning. */
195 const int socket = (int) (intptr_t) key;
196 const struct notification_client *client = caa_container_of(node,
197 struct notification_client, client_socket_ht_node);
198
199 return client->socket == socket;
200 }
201
202 static
203 int match_client_id(struct cds_lfht_node *node, const void *key)
204 {
205 /* This double-cast is intended to supress pointer-to-cast warning. */
206 const notification_client_id id = *((notification_client_id *) key);
207 const struct notification_client *client = caa_container_of(
208 node, struct notification_client, client_id_ht_node);
209
210 return client->id == id;
211 }
212
213 static
214 int match_channel_trigger_list(struct cds_lfht_node *node, const void *key)
215 {
216 struct channel_key *channel_key = (struct channel_key *) key;
217 struct lttng_channel_trigger_list *trigger_list;
218
219 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
220 channel_triggers_ht_node);
221
222 return !!((channel_key->key == trigger_list->channel_key.key) &&
223 (channel_key->domain == trigger_list->channel_key.domain));
224 }
225
226 static
227 int match_session_trigger_list(struct cds_lfht_node *node, const void *key)
228 {
229 const char *session_name = (const char *) key;
230 struct lttng_session_trigger_list *trigger_list;
231
232 trigger_list = caa_container_of(node, struct lttng_session_trigger_list,
233 session_triggers_ht_node);
234
235 return !!(strcmp(trigger_list->session_name, session_name) == 0);
236 }
237
238 static
239 int match_channel_state_sample(struct cds_lfht_node *node, const void *key)
240 {
241 struct channel_key *channel_key = (struct channel_key *) key;
242 struct channel_state_sample *sample;
243
244 sample = caa_container_of(node, struct channel_state_sample,
245 channel_state_ht_node);
246
247 return !!((channel_key->key == sample->key.key) &&
248 (channel_key->domain == sample->key.domain));
249 }
250
251 static
252 int match_channel_info(struct cds_lfht_node *node, const void *key)
253 {
254 struct channel_key *channel_key = (struct channel_key *) key;
255 struct channel_info *channel_info;
256
257 channel_info = caa_container_of(node, struct channel_info,
258 channels_ht_node);
259
260 return !!((channel_key->key == channel_info->key.key) &&
261 (channel_key->domain == channel_info->key.domain));
262 }
263
264 static
265 int match_condition(struct cds_lfht_node *node, const void *key)
266 {
267 struct lttng_condition *condition_key = (struct lttng_condition *) key;
268 struct lttng_trigger_ht_element *trigger;
269 struct lttng_condition *condition;
270
271 trigger = caa_container_of(node, struct lttng_trigger_ht_element,
272 node);
273 condition = lttng_trigger_get_condition(trigger->trigger);
274 assert(condition);
275
276 return !!lttng_condition_is_equal(condition_key, condition);
277 }
278
279 static
280 int match_client_list_condition(struct cds_lfht_node *node, const void *key)
281 {
282 struct lttng_condition *condition_key = (struct lttng_condition *) key;
283 struct notification_client_list *client_list;
284 const struct lttng_condition *condition;
285
286 assert(condition_key);
287
288 client_list = caa_container_of(node, struct notification_client_list,
289 notification_trigger_clients_ht_node);
290 condition = lttng_trigger_get_const_condition(client_list->trigger);
291
292 return !!lttng_condition_is_equal(condition_key, condition);
293 }
294
295 static
296 int match_session(struct cds_lfht_node *node, const void *key)
297 {
298 const char *name = key;
299 struct session_info *session_info = caa_container_of(
300 node, struct session_info, sessions_ht_node);
301
302 return !strcmp(session_info->name, name);
303 }
304
305 static
306 unsigned long lttng_condition_buffer_usage_hash(
307 const struct lttng_condition *_condition)
308 {
309 unsigned long hash;
310 unsigned long condition_type;
311 struct lttng_condition_buffer_usage *condition;
312
313 condition = container_of(_condition,
314 struct lttng_condition_buffer_usage, parent);
315
316 condition_type = (unsigned long) condition->parent.type;
317 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
318 if (condition->session_name) {
319 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
320 }
321 if (condition->channel_name) {
322 hash ^= hash_key_str(condition->channel_name, lttng_ht_seed);
323 }
324 if (condition->domain.set) {
325 hash ^= hash_key_ulong(
326 (void *) condition->domain.type,
327 lttng_ht_seed);
328 }
329 if (condition->threshold_ratio.set) {
330 uint64_t val;
331
332 val = condition->threshold_ratio.value * (double) UINT32_MAX;
333 hash ^= hash_key_u64(&val, lttng_ht_seed);
334 } else if (condition->threshold_bytes.set) {
335 uint64_t val;
336
337 val = condition->threshold_bytes.value;
338 hash ^= hash_key_u64(&val, lttng_ht_seed);
339 }
340 return hash;
341 }
342
343 static
344 unsigned long lttng_condition_session_consumed_size_hash(
345 const struct lttng_condition *_condition)
346 {
347 unsigned long hash;
348 unsigned long condition_type =
349 (unsigned long) LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE;
350 struct lttng_condition_session_consumed_size *condition;
351 uint64_t val;
352
353 condition = container_of(_condition,
354 struct lttng_condition_session_consumed_size, parent);
355
356 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
357 if (condition->session_name) {
358 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
359 }
360 val = condition->consumed_threshold_bytes.value;
361 hash ^= hash_key_u64(&val, lttng_ht_seed);
362 return hash;
363 }
364
365 static
366 unsigned long lttng_condition_session_rotation_hash(
367 const struct lttng_condition *_condition)
368 {
369 unsigned long hash, condition_type;
370 struct lttng_condition_session_rotation *condition;
371
372 condition = container_of(_condition,
373 struct lttng_condition_session_rotation, parent);
374 condition_type = (unsigned long) condition->parent.type;
375 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
376 assert(condition->session_name);
377 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
378 return hash;
379 }
380
381 /*
382 * The lttng_condition hashing code is kept in this file (rather than
383 * condition.c) since it makes use of GPLv2 code (hashtable utils), which we
384 * don't want to link in liblttng-ctl.
385 */
386 static
387 unsigned long lttng_condition_hash(const struct lttng_condition *condition)
388 {
389 switch (condition->type) {
390 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
391 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
392 return lttng_condition_buffer_usage_hash(condition);
393 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
394 return lttng_condition_session_consumed_size_hash(condition);
395 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
396 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
397 return lttng_condition_session_rotation_hash(condition);
398 default:
399 ERR("[notification-thread] Unexpected condition type caught");
400 abort();
401 }
402 }
403
404 static
405 unsigned long hash_channel_key(struct channel_key *key)
406 {
407 unsigned long key_hash = hash_key_u64(&key->key, lttng_ht_seed);
408 unsigned long domain_hash = hash_key_ulong(
409 (void *) (unsigned long) key->domain, lttng_ht_seed);
410
411 return key_hash ^ domain_hash;
412 }
413
414 static
415 unsigned long hash_client_socket(int socket)
416 {
417 return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
418 }
419
420 static
421 unsigned long hash_client_id(notification_client_id id)
422 {
423 return hash_key_u64(&id, lttng_ht_seed);
424 }
425
426 /*
427 * Get the type of object to which a given condition applies. Bindings let
428 * the notification system evaluate a trigger's condition when a given
429 * object's state is updated.
430 *
431 * For instance, a condition bound to a channel will be evaluated everytime
432 * the channel's state is changed by a channel monitoring sample.
433 */
434 static
435 enum lttng_object_type get_condition_binding_object(
436 const struct lttng_condition *condition)
437 {
438 switch (lttng_condition_get_type(condition)) {
439 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
440 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
441 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
442 return LTTNG_OBJECT_TYPE_CHANNEL;
443 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
444 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
445 return LTTNG_OBJECT_TYPE_SESSION;
446 default:
447 return LTTNG_OBJECT_TYPE_UNKNOWN;
448 }
449 }
450
451 static
452 void free_channel_info_rcu(struct rcu_head *node)
453 {
454 free(caa_container_of(node, struct channel_info, rcu_node));
455 }
456
457 static
458 void channel_info_destroy(struct channel_info *channel_info)
459 {
460 if (!channel_info) {
461 return;
462 }
463
464 if (channel_info->session_info) {
465 session_info_remove_channel(channel_info->session_info,
466 channel_info);
467 session_info_put(channel_info->session_info);
468 }
469 if (channel_info->name) {
470 free(channel_info->name);
471 }
472 call_rcu(&channel_info->rcu_node, free_channel_info_rcu);
473 }
474
475 static
476 void free_session_info_rcu(struct rcu_head *node)
477 {
478 free(caa_container_of(node, struct session_info, rcu_node));
479 }
480
481 /* Don't call directly, use the ref-counting mechanism. */
482 static
483 void session_info_destroy(void *_data)
484 {
485 struct session_info *session_info = _data;
486 int ret;
487
488 assert(session_info);
489 if (session_info->channel_infos_ht) {
490 ret = cds_lfht_destroy(session_info->channel_infos_ht, NULL);
491 if (ret) {
492 ERR("[notification-thread] Failed to destroy channel information hash table");
493 }
494 }
495 lttng_session_trigger_list_destroy(session_info->trigger_list);
496
497 rcu_read_lock();
498 cds_lfht_del(session_info->sessions_ht,
499 &session_info->sessions_ht_node);
500 rcu_read_unlock();
501 free(session_info->name);
502 call_rcu(&session_info->rcu_node, free_session_info_rcu);
503 }
504
505 static
506 void session_info_get(struct session_info *session_info)
507 {
508 if (!session_info) {
509 return;
510 }
511 lttng_ref_get(&session_info->ref);
512 }
513
514 static
515 void session_info_put(struct session_info *session_info)
516 {
517 if (!session_info) {
518 return;
519 }
520 lttng_ref_put(&session_info->ref);
521 }
522
523 static
524 struct session_info *session_info_create(const char *name, uid_t uid, gid_t gid,
525 struct lttng_session_trigger_list *trigger_list,
526 struct cds_lfht *sessions_ht)
527 {
528 struct session_info *session_info;
529
530 assert(name);
531
532 session_info = zmalloc(sizeof(*session_info));
533 if (!session_info) {
534 goto end;
535 }
536 lttng_ref_init(&session_info->ref, session_info_destroy);
537
538 session_info->channel_infos_ht = cds_lfht_new(DEFAULT_HT_SIZE,
539 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
540 if (!session_info->channel_infos_ht) {
541 goto error;
542 }
543
544 cds_lfht_node_init(&session_info->sessions_ht_node);
545 session_info->name = strdup(name);
546 if (!session_info->name) {
547 goto error;
548 }
549 session_info->uid = uid;
550 session_info->gid = gid;
551 session_info->trigger_list = trigger_list;
552 session_info->sessions_ht = sessions_ht;
553 end:
554 return session_info;
555 error:
556 session_info_put(session_info);
557 return NULL;
558 }
559
560 static
561 void session_info_add_channel(struct session_info *session_info,
562 struct channel_info *channel_info)
563 {
564 rcu_read_lock();
565 cds_lfht_add(session_info->channel_infos_ht,
566 hash_channel_key(&channel_info->key),
567 &channel_info->session_info_channels_ht_node);
568 rcu_read_unlock();
569 }
570
571 static
572 void session_info_remove_channel(struct session_info *session_info,
573 struct channel_info *channel_info)
574 {
575 rcu_read_lock();
576 cds_lfht_del(session_info->channel_infos_ht,
577 &channel_info->session_info_channels_ht_node);
578 rcu_read_unlock();
579 }
580
581 static
582 struct channel_info *channel_info_create(const char *channel_name,
583 struct channel_key *channel_key, uint64_t channel_capacity,
584 struct session_info *session_info)
585 {
586 struct channel_info *channel_info = zmalloc(sizeof(*channel_info));
587
588 if (!channel_info) {
589 goto end;
590 }
591
592 cds_lfht_node_init(&channel_info->channels_ht_node);
593 cds_lfht_node_init(&channel_info->session_info_channels_ht_node);
594 memcpy(&channel_info->key, channel_key, sizeof(*channel_key));
595 channel_info->capacity = channel_capacity;
596
597 channel_info->name = strdup(channel_name);
598 if (!channel_info->name) {
599 goto error;
600 }
601
602 /*
603 * Set the references between session and channel infos:
604 * - channel_info holds a strong reference to session_info
605 * - session_info holds a weak reference to channel_info
606 */
607 session_info_get(session_info);
608 session_info_add_channel(session_info, channel_info);
609 channel_info->session_info = session_info;
610 end:
611 return channel_info;
612 error:
613 channel_info_destroy(channel_info);
614 return NULL;
615 }
616
617 LTTNG_HIDDEN
618 bool notification_client_list_get(struct notification_client_list *list)
619 {
620 return urcu_ref_get_unless_zero(&list->ref);
621 }
622
623 static
624 void free_notification_client_list_rcu(struct rcu_head *node)
625 {
626 free(caa_container_of(node, struct notification_client_list,
627 rcu_node));
628 }
629
630 static
631 void notification_client_list_release(struct urcu_ref *list_ref)
632 {
633 struct notification_client_list *list =
634 container_of(list_ref, typeof(*list), ref);
635 struct notification_client_list_element *client_list_element, *tmp;
636
637 if (list->notification_trigger_clients_ht) {
638 rcu_read_lock();
639 cds_lfht_del(list->notification_trigger_clients_ht,
640 &list->notification_trigger_clients_ht_node);
641 rcu_read_unlock();
642 list->notification_trigger_clients_ht = NULL;
643 }
644 cds_list_for_each_entry_safe(client_list_element, tmp,
645 &list->list, node) {
646 free(client_list_element);
647 }
648 pthread_mutex_destroy(&list->lock);
649 call_rcu(&list->rcu_node, free_notification_client_list_rcu);
650 }
651
652 static
653 struct notification_client_list *notification_client_list_create(
654 const struct lttng_trigger *trigger)
655 {
656 struct notification_client_list *client_list =
657 zmalloc(sizeof(*client_list));
658
659 if (!client_list) {
660 goto error;
661 }
662 pthread_mutex_init(&client_list->lock, NULL);
663 urcu_ref_init(&client_list->ref);
664 cds_lfht_node_init(&client_list->notification_trigger_clients_ht_node);
665 CDS_INIT_LIST_HEAD(&client_list->list);
666 client_list->trigger = trigger;
667 error:
668 return client_list;
669 }
670
671 static
672 void publish_notification_client_list(
673 struct notification_thread_state *state,
674 struct notification_client_list *list)
675 {
676 const struct lttng_condition *condition =
677 lttng_trigger_get_const_condition(list->trigger);
678
679 assert(!list->notification_trigger_clients_ht);
680 notification_client_list_get(list);
681
682 list->notification_trigger_clients_ht =
683 state->notification_trigger_clients_ht;
684
685 rcu_read_lock();
686 cds_lfht_add(state->notification_trigger_clients_ht,
687 lttng_condition_hash(condition),
688 &list->notification_trigger_clients_ht_node);
689 rcu_read_unlock();
690 }
691
692 LTTNG_HIDDEN
693 void notification_client_list_put(struct notification_client_list *list)
694 {
695 if (!list) {
696 return;
697 }
698 return urcu_ref_put(&list->ref, notification_client_list_release);
699 }
700
701 /* Provides a reference to the returned list. */
702 static
703 struct notification_client_list *get_client_list_from_condition(
704 struct notification_thread_state *state,
705 const struct lttng_condition *condition)
706 {
707 struct cds_lfht_node *node;
708 struct cds_lfht_iter iter;
709 struct notification_client_list *list = NULL;
710
711 rcu_read_lock();
712 cds_lfht_lookup(state->notification_trigger_clients_ht,
713 lttng_condition_hash(condition),
714 match_client_list_condition,
715 condition,
716 &iter);
717 node = cds_lfht_iter_get_node(&iter);
718 if (node) {
719 list = container_of(node, struct notification_client_list,
720 notification_trigger_clients_ht_node);
721 list = notification_client_list_get(list) ? list : NULL;
722 }
723
724 rcu_read_unlock();
725 return list;
726 }
727
728 static
729 int evaluate_channel_condition_for_client(
730 const struct lttng_condition *condition,
731 struct notification_thread_state *state,
732 struct lttng_evaluation **evaluation,
733 uid_t *session_uid, gid_t *session_gid)
734 {
735 int ret;
736 struct cds_lfht_iter iter;
737 struct cds_lfht_node *node;
738 struct channel_info *channel_info = NULL;
739 struct channel_key *channel_key = NULL;
740 struct channel_state_sample *last_sample = NULL;
741 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
742
743 rcu_read_lock();
744
745 /* Find the channel associated with the condition. */
746 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter,
747 channel_trigger_list, channel_triggers_ht_node) {
748 struct lttng_trigger_list_element *element;
749
750 cds_list_for_each_entry(element, &channel_trigger_list->list, node) {
751 const struct lttng_condition *current_condition =
752 lttng_trigger_get_const_condition(
753 element->trigger);
754
755 assert(current_condition);
756 if (!lttng_condition_is_equal(condition,
757 current_condition)) {
758 continue;
759 }
760
761 /* Found the trigger, save the channel key. */
762 channel_key = &channel_trigger_list->channel_key;
763 break;
764 }
765 if (channel_key) {
766 /* The channel key was found stop iteration. */
767 break;
768 }
769 }
770
771 if (!channel_key){
772 /* No channel found; normal exit. */
773 DBG("[notification-thread] No known channel associated with newly subscribed-to condition");
774 ret = 0;
775 goto end;
776 }
777
778 /* Fetch channel info for the matching channel. */
779 cds_lfht_lookup(state->channels_ht,
780 hash_channel_key(channel_key),
781 match_channel_info,
782 channel_key,
783 &iter);
784 node = cds_lfht_iter_get_node(&iter);
785 assert(node);
786 channel_info = caa_container_of(node, struct channel_info,
787 channels_ht_node);
788
789 /* Retrieve the channel's last sample, if it exists. */
790 cds_lfht_lookup(state->channel_state_ht,
791 hash_channel_key(channel_key),
792 match_channel_state_sample,
793 channel_key,
794 &iter);
795 node = cds_lfht_iter_get_node(&iter);
796 if (node) {
797 last_sample = caa_container_of(node,
798 struct channel_state_sample,
799 channel_state_ht_node);
800 } else {
801 /* Nothing to evaluate, no sample was ever taken. Normal exit */
802 DBG("[notification-thread] No channel sample associated with newly subscribed-to condition");
803 ret = 0;
804 goto end;
805 }
806
807 ret = evaluate_buffer_condition(condition, evaluation, state,
808 NULL, last_sample,
809 0, channel_info->session_info->consumed_data_size,
810 channel_info);
811 if (ret) {
812 WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition");
813 goto end;
814 }
815
816 *session_uid = channel_info->session_info->uid;
817 *session_gid = channel_info->session_info->gid;
818 end:
819 rcu_read_unlock();
820 return ret;
821 }
822
823 static
824 const char *get_condition_session_name(const struct lttng_condition *condition)
825 {
826 const char *session_name = NULL;
827 enum lttng_condition_status status;
828
829 switch (lttng_condition_get_type(condition)) {
830 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
831 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
832 status = lttng_condition_buffer_usage_get_session_name(
833 condition, &session_name);
834 break;
835 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
836 status = lttng_condition_session_consumed_size_get_session_name(
837 condition, &session_name);
838 break;
839 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
840 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
841 status = lttng_condition_session_rotation_get_session_name(
842 condition, &session_name);
843 break;
844 default:
845 abort();
846 }
847 if (status != LTTNG_CONDITION_STATUS_OK) {
848 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
849 goto end;
850 }
851 end:
852 return session_name;
853 }
854
855 static
856 int evaluate_session_condition_for_client(
857 const struct lttng_condition *condition,
858 struct notification_thread_state *state,
859 struct lttng_evaluation **evaluation,
860 uid_t *session_uid, gid_t *session_gid)
861 {
862 int ret;
863 struct cds_lfht_iter iter;
864 struct cds_lfht_node *node;
865 const char *session_name;
866 struct session_info *session_info = NULL;
867
868 rcu_read_lock();
869 session_name = get_condition_session_name(condition);
870
871 /* Find the session associated with the trigger. */
872 cds_lfht_lookup(state->sessions_ht,
873 hash_key_str(session_name, lttng_ht_seed),
874 match_session,
875 session_name,
876 &iter);
877 node = cds_lfht_iter_get_node(&iter);
878 if (!node) {
879 DBG("[notification-thread] No known session matching name \"%s\"",
880 session_name);
881 ret = 0;
882 goto end;
883 }
884
885 session_info = caa_container_of(node, struct session_info,
886 sessions_ht_node);
887 session_info_get(session_info);
888
889 /*
890 * Evaluation is performed in-line here since only one type of
891 * session-bound condition is handled for the moment.
892 */
893 switch (lttng_condition_get_type(condition)) {
894 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
895 if (!session_info->rotation.ongoing) {
896 ret = 0;
897 goto end_session_put;
898 }
899
900 *evaluation = lttng_evaluation_session_rotation_ongoing_create(
901 session_info->rotation.id);
902 if (!*evaluation) {
903 /* Fatal error. */
904 ERR("[notification-thread] Failed to create session rotation ongoing evaluation for session \"%s\"",
905 session_info->name);
906 ret = -1;
907 goto end_session_put;
908 }
909 ret = 0;
910 break;
911 default:
912 ret = 0;
913 goto end_session_put;
914 }
915
916 *session_uid = session_info->uid;
917 *session_gid = session_info->gid;
918
919 end_session_put:
920 session_info_put(session_info);
921 end:
922 rcu_read_unlock();
923 return ret;
924 }
925
926 static
927 int evaluate_condition_for_client(const struct lttng_trigger *trigger,
928 const struct lttng_condition *condition,
929 struct notification_client *client,
930 struct notification_thread_state *state)
931 {
932 int ret;
933 struct lttng_evaluation *evaluation = NULL;
934 struct notification_client_list client_list = {
935 .lock = PTHREAD_MUTEX_INITIALIZER,
936 };
937 struct notification_client_list_element client_list_element = { 0 };
938 uid_t object_uid = 0;
939 gid_t object_gid = 0;
940
941 assert(trigger);
942 assert(condition);
943 assert(client);
944 assert(state);
945
946 switch (get_condition_binding_object(condition)) {
947 case LTTNG_OBJECT_TYPE_SESSION:
948 ret = evaluate_session_condition_for_client(condition, state,
949 &evaluation, &object_uid, &object_gid);
950 break;
951 case LTTNG_OBJECT_TYPE_CHANNEL:
952 ret = evaluate_channel_condition_for_client(condition, state,
953 &evaluation, &object_uid, &object_gid);
954 break;
955 case LTTNG_OBJECT_TYPE_NONE:
956 ret = 0;
957 goto end;
958 case LTTNG_OBJECT_TYPE_UNKNOWN:
959 default:
960 ret = -1;
961 goto end;
962 }
963 if (ret) {
964 /* Fatal error. */
965 goto end;
966 }
967 if (!evaluation) {
968 /* Evaluation yielded nothing. Normal exit. */
969 DBG("[notification-thread] Newly subscribed-to condition evaluated to false, nothing to report to client");
970 ret = 0;
971 goto end;
972 }
973
974 /*
975 * Create a temporary client list with the client currently
976 * subscribing.
977 */
978 cds_lfht_node_init(&client_list.notification_trigger_clients_ht_node);
979 CDS_INIT_LIST_HEAD(&client_list.list);
980 client_list.trigger = trigger;
981
982 CDS_INIT_LIST_HEAD(&client_list_element.node);
983 client_list_element.client = client;
984 cds_list_add(&client_list_element.node, &client_list.list);
985
986 /* Send evaluation result to the newly-subscribed client. */
987 DBG("[notification-thread] Newly subscribed-to condition evaluated to true, notifying client");
988 ret = send_evaluation_to_clients(trigger, evaluation, &client_list,
989 state, object_uid, object_gid);
990
991 end:
992 return ret;
993 }
994
995 static
996 int notification_thread_client_subscribe(struct notification_client *client,
997 struct lttng_condition *condition,
998 struct notification_thread_state *state,
999 enum lttng_notification_channel_status *_status)
1000 {
1001 int ret = 0;
1002 struct notification_client_list *client_list = NULL;
1003 struct lttng_condition_list_element *condition_list_element = NULL;
1004 struct notification_client_list_element *client_list_element = NULL;
1005 enum lttng_notification_channel_status status =
1006 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1007
1008 /*
1009 * Ensure that the client has not already subscribed to this condition
1010 * before.
1011 */
1012 cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
1013 if (lttng_condition_is_equal(condition_list_element->condition,
1014 condition)) {
1015 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
1016 goto end;
1017 }
1018 }
1019
1020 condition_list_element = zmalloc(sizeof(*condition_list_element));
1021 if (!condition_list_element) {
1022 ret = -1;
1023 goto error;
1024 }
1025 client_list_element = zmalloc(sizeof(*client_list_element));
1026 if (!client_list_element) {
1027 ret = -1;
1028 goto error;
1029 }
1030
1031 /*
1032 * Add the newly-subscribed condition to the client's subscription list.
1033 */
1034 CDS_INIT_LIST_HEAD(&condition_list_element->node);
1035 condition_list_element->condition = condition;
1036 cds_list_add(&condition_list_element->node, &client->condition_list);
1037
1038 client_list = get_client_list_from_condition(state, condition);
1039 if (!client_list) {
1040 /*
1041 * No notification-emiting trigger registered with this
1042 * condition. We don't evaluate the condition right away
1043 * since this trigger is not registered yet.
1044 */
1045 free(client_list_element);
1046 goto end;
1047 }
1048
1049 /*
1050 * The condition to which the client just subscribed is evaluated
1051 * at this point so that conditions that are already TRUE result
1052 * in a notification being sent out.
1053 *
1054 * The client_list's trigger is used without locking the list itself.
1055 * This is correct since the list doesn't own the trigger and the
1056 * object is immutable.
1057 */
1058 if (evaluate_condition_for_client(client_list->trigger, condition,
1059 client, state)) {
1060 WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
1061 ret = -1;
1062 free(client_list_element);
1063 goto end;
1064 }
1065
1066 /*
1067 * Add the client to the list of clients interested in a given trigger
1068 * if a "notification" trigger with a corresponding condition was
1069 * added prior.
1070 */
1071 client_list_element->client = client;
1072 CDS_INIT_LIST_HEAD(&client_list_element->node);
1073
1074 pthread_mutex_lock(&client_list->lock);
1075 cds_list_add(&client_list_element->node, &client_list->list);
1076 pthread_mutex_unlock(&client_list->lock);
1077 end:
1078 if (_status) {
1079 *_status = status;
1080 }
1081 if (client_list) {
1082 notification_client_list_put(client_list);
1083 }
1084 return ret;
1085 error:
1086 free(condition_list_element);
1087 free(client_list_element);
1088 return ret;
1089 }
1090
1091 static
1092 int notification_thread_client_unsubscribe(
1093 struct notification_client *client,
1094 struct lttng_condition *condition,
1095 struct notification_thread_state *state,
1096 enum lttng_notification_channel_status *_status)
1097 {
1098 struct notification_client_list *client_list;
1099 struct lttng_condition_list_element *condition_list_element,
1100 *condition_tmp;
1101 struct notification_client_list_element *client_list_element,
1102 *client_tmp;
1103 bool condition_found = false;
1104 enum lttng_notification_channel_status status =
1105 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1106
1107 /* Remove the condition from the client's condition list. */
1108 cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
1109 &client->condition_list, node) {
1110 if (!lttng_condition_is_equal(condition_list_element->condition,
1111 condition)) {
1112 continue;
1113 }
1114
1115 cds_list_del(&condition_list_element->node);
1116 /*
1117 * The caller may be iterating on the client's conditions to
1118 * tear down a client's connection. In this case, the condition
1119 * will be destroyed at the end.
1120 */
1121 if (condition != condition_list_element->condition) {
1122 lttng_condition_destroy(
1123 condition_list_element->condition);
1124 }
1125 free(condition_list_element);
1126 condition_found = true;
1127 break;
1128 }
1129
1130 if (!condition_found) {
1131 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
1132 goto end;
1133 }
1134
1135 /*
1136 * Remove the client from the list of clients interested the trigger
1137 * matching the condition.
1138 */
1139 client_list = get_client_list_from_condition(state, condition);
1140 if (!client_list) {
1141 goto end;
1142 }
1143
1144 pthread_mutex_lock(&client_list->lock);
1145 cds_list_for_each_entry_safe(client_list_element, client_tmp,
1146 &client_list->list, node) {
1147 if (client_list_element->client->id != client->id) {
1148 continue;
1149 }
1150 cds_list_del(&client_list_element->node);
1151 free(client_list_element);
1152 break;
1153 }
1154 pthread_mutex_unlock(&client_list->lock);
1155 notification_client_list_put(client_list);
1156 client_list = NULL;
1157 end:
1158 lttng_condition_destroy(condition);
1159 if (_status) {
1160 *_status = status;
1161 }
1162 return 0;
1163 }
1164
1165 static
1166 void free_notification_client_rcu(struct rcu_head *node)
1167 {
1168 free(caa_container_of(node, struct notification_client, rcu_node));
1169 }
1170
1171 static
1172 void notification_client_destroy(struct notification_client *client,
1173 struct notification_thread_state *state)
1174 {
1175 if (!client) {
1176 return;
1177 }
1178
1179 /*
1180 * The client object is not reachable by other threads, no need to lock
1181 * the client here.
1182 */
1183 if (client->socket >= 0) {
1184 (void) lttcomm_close_unix_sock(client->socket);
1185 client->socket = -1;
1186 }
1187 client->communication.active = false;
1188 lttng_payload_reset(&client->communication.inbound.payload);
1189 lttng_payload_reset(&client->communication.outbound.payload);
1190 pthread_mutex_destroy(&client->lock);
1191 call_rcu(&client->rcu_node, free_notification_client_rcu);
1192 }
1193
1194 /*
1195 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1196 * client pointer).
1197 */
1198 static
1199 struct notification_client *get_client_from_socket(int socket,
1200 struct notification_thread_state *state)
1201 {
1202 struct cds_lfht_iter iter;
1203 struct cds_lfht_node *node;
1204 struct notification_client *client = NULL;
1205
1206 cds_lfht_lookup(state->client_socket_ht,
1207 hash_client_socket(socket),
1208 match_client_socket,
1209 (void *) (unsigned long) socket,
1210 &iter);
1211 node = cds_lfht_iter_get_node(&iter);
1212 if (!node) {
1213 goto end;
1214 }
1215
1216 client = caa_container_of(node, struct notification_client,
1217 client_socket_ht_node);
1218 end:
1219 return client;
1220 }
1221
1222 /*
1223 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1224 * client pointer).
1225 */
1226 static
1227 struct notification_client *get_client_from_id(notification_client_id id,
1228 struct notification_thread_state *state)
1229 {
1230 struct cds_lfht_iter iter;
1231 struct cds_lfht_node *node;
1232 struct notification_client *client = NULL;
1233
1234 cds_lfht_lookup(state->client_id_ht,
1235 hash_client_id(id),
1236 match_client_id,
1237 &id,
1238 &iter);
1239 node = cds_lfht_iter_get_node(&iter);
1240 if (!node) {
1241 goto end;
1242 }
1243
1244 client = caa_container_of(node, struct notification_client,
1245 client_id_ht_node);
1246 end:
1247 return client;
1248 }
1249
1250 static
1251 bool buffer_usage_condition_applies_to_channel(
1252 const struct lttng_condition *condition,
1253 const struct channel_info *channel_info)
1254 {
1255 enum lttng_condition_status status;
1256 enum lttng_domain_type condition_domain;
1257 const char *condition_session_name = NULL;
1258 const char *condition_channel_name = NULL;
1259
1260 status = lttng_condition_buffer_usage_get_domain_type(condition,
1261 &condition_domain);
1262 assert(status == LTTNG_CONDITION_STATUS_OK);
1263 if (channel_info->key.domain != condition_domain) {
1264 goto fail;
1265 }
1266
1267 status = lttng_condition_buffer_usage_get_session_name(
1268 condition, &condition_session_name);
1269 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1270
1271 status = lttng_condition_buffer_usage_get_channel_name(
1272 condition, &condition_channel_name);
1273 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_channel_name);
1274
1275 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1276 goto fail;
1277 }
1278 if (strcmp(channel_info->name, condition_channel_name)) {
1279 goto fail;
1280 }
1281
1282 return true;
1283 fail:
1284 return false;
1285 }
1286
1287 static
1288 bool session_consumed_size_condition_applies_to_channel(
1289 const struct lttng_condition *condition,
1290 const struct channel_info *channel_info)
1291 {
1292 enum lttng_condition_status status;
1293 const char *condition_session_name = NULL;
1294
1295 status = lttng_condition_session_consumed_size_get_session_name(
1296 condition, &condition_session_name);
1297 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1298
1299 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1300 goto fail;
1301 }
1302
1303 return true;
1304 fail:
1305 return false;
1306 }
1307
1308 static
1309 bool trigger_applies_to_channel(const struct lttng_trigger *trigger,
1310 const struct channel_info *channel_info)
1311 {
1312 const struct lttng_condition *condition;
1313 bool trigger_applies;
1314
1315 condition = lttng_trigger_get_const_condition(trigger);
1316 if (!condition) {
1317 goto fail;
1318 }
1319
1320 switch (lttng_condition_get_type(condition)) {
1321 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1322 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1323 trigger_applies = buffer_usage_condition_applies_to_channel(
1324 condition, channel_info);
1325 break;
1326 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
1327 trigger_applies = session_consumed_size_condition_applies_to_channel(
1328 condition, channel_info);
1329 break;
1330 default:
1331 goto fail;
1332 }
1333
1334 return trigger_applies;
1335 fail:
1336 return false;
1337 }
1338
1339 static
1340 bool trigger_applies_to_client(struct lttng_trigger *trigger,
1341 struct notification_client *client)
1342 {
1343 bool applies = false;
1344 struct lttng_condition_list_element *condition_list_element;
1345
1346 cds_list_for_each_entry(condition_list_element, &client->condition_list,
1347 node) {
1348 applies = lttng_condition_is_equal(
1349 condition_list_element->condition,
1350 lttng_trigger_get_condition(trigger));
1351 if (applies) {
1352 break;
1353 }
1354 }
1355 return applies;
1356 }
1357
1358 /* Must be called with RCU read lock held. */
1359 static
1360 struct lttng_session_trigger_list *get_session_trigger_list(
1361 struct notification_thread_state *state,
1362 const char *session_name)
1363 {
1364 struct lttng_session_trigger_list *list = NULL;
1365 struct cds_lfht_node *node;
1366 struct cds_lfht_iter iter;
1367
1368 cds_lfht_lookup(state->session_triggers_ht,
1369 hash_key_str(session_name, lttng_ht_seed),
1370 match_session_trigger_list,
1371 session_name,
1372 &iter);
1373 node = cds_lfht_iter_get_node(&iter);
1374 if (!node) {
1375 /*
1376 * Not an error, the list of triggers applying to that session
1377 * will be initialized when the session is created.
1378 */
1379 DBG("[notification-thread] No trigger list found for session \"%s\" as it is not yet known to the notification system",
1380 session_name);
1381 goto end;
1382 }
1383
1384 list = caa_container_of(node,
1385 struct lttng_session_trigger_list,
1386 session_triggers_ht_node);
1387 end:
1388 return list;
1389 }
1390
1391 /*
1392 * Allocate an empty lttng_session_trigger_list for the session named
1393 * 'session_name'.
1394 *
1395 * No ownership of 'session_name' is assumed by the session trigger list.
1396 * It is the caller's responsability to ensure the session name is alive
1397 * for as long as this list is.
1398 */
1399 static
1400 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
1401 const char *session_name,
1402 struct cds_lfht *session_triggers_ht)
1403 {
1404 struct lttng_session_trigger_list *list;
1405
1406 list = zmalloc(sizeof(*list));
1407 if (!list) {
1408 goto end;
1409 }
1410 list->session_name = session_name;
1411 CDS_INIT_LIST_HEAD(&list->list);
1412 cds_lfht_node_init(&list->session_triggers_ht_node);
1413 list->session_triggers_ht = session_triggers_ht;
1414
1415 rcu_read_lock();
1416 /* Publish the list through the session_triggers_ht. */
1417 cds_lfht_add(session_triggers_ht,
1418 hash_key_str(session_name, lttng_ht_seed),
1419 &list->session_triggers_ht_node);
1420 rcu_read_unlock();
1421 end:
1422 return list;
1423 }
1424
1425 static
1426 void free_session_trigger_list_rcu(struct rcu_head *node)
1427 {
1428 free(caa_container_of(node, struct lttng_session_trigger_list,
1429 rcu_node));
1430 }
1431
1432 static
1433 void lttng_session_trigger_list_destroy(struct lttng_session_trigger_list *list)
1434 {
1435 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1436
1437 /* Empty the list element by element, and then free the list itself. */
1438 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1439 &list->list, node) {
1440 cds_list_del(&trigger_list_element->node);
1441 free(trigger_list_element);
1442 }
1443 rcu_read_lock();
1444 /* Unpublish the list from the session_triggers_ht. */
1445 cds_lfht_del(list->session_triggers_ht,
1446 &list->session_triggers_ht_node);
1447 rcu_read_unlock();
1448 call_rcu(&list->rcu_node, free_session_trigger_list_rcu);
1449 }
1450
1451 static
1452 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
1453 struct lttng_trigger *trigger)
1454 {
1455 int ret = 0;
1456 struct lttng_trigger_list_element *new_element =
1457 zmalloc(sizeof(*new_element));
1458
1459 if (!new_element) {
1460 ret = -1;
1461 goto end;
1462 }
1463 CDS_INIT_LIST_HEAD(&new_element->node);
1464 new_element->trigger = trigger;
1465 cds_list_add(&new_element->node, &list->list);
1466 end:
1467 return ret;
1468 }
1469
1470 static
1471 bool trigger_applies_to_session(const struct lttng_trigger *trigger,
1472 const char *session_name)
1473 {
1474 bool applies = false;
1475 const struct lttng_condition *condition;
1476
1477 condition = lttng_trigger_get_const_condition(trigger);
1478 switch (lttng_condition_get_type(condition)) {
1479 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
1480 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
1481 {
1482 enum lttng_condition_status condition_status;
1483 const char *condition_session_name;
1484
1485 condition_status = lttng_condition_session_rotation_get_session_name(
1486 condition, &condition_session_name);
1487 if (condition_status != LTTNG_CONDITION_STATUS_OK) {
1488 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
1489 goto end;
1490 }
1491
1492 assert(condition_session_name);
1493 applies = !strcmp(condition_session_name, session_name);
1494 break;
1495 }
1496 default:
1497 goto end;
1498 }
1499 end:
1500 return applies;
1501 }
1502
1503 /*
1504 * Allocate and initialize an lttng_session_trigger_list which contains
1505 * all triggers that apply to the session named 'session_name'.
1506 *
1507 * No ownership of 'session_name' is assumed by the session trigger list.
1508 * It is the caller's responsability to ensure the session name is alive
1509 * for as long as this list is.
1510 */
1511 static
1512 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
1513 const struct notification_thread_state *state,
1514 const char *session_name)
1515 {
1516 int trigger_count = 0;
1517 struct lttng_session_trigger_list *session_trigger_list = NULL;
1518 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1519 struct cds_lfht_iter iter;
1520
1521 session_trigger_list = lttng_session_trigger_list_create(session_name,
1522 state->session_triggers_ht);
1523
1524 /* Add all triggers applying to the session named 'session_name'. */
1525 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1526 node) {
1527 int ret;
1528
1529 if (!trigger_applies_to_session(trigger_ht_element->trigger,
1530 session_name)) {
1531 continue;
1532 }
1533
1534 ret = lttng_session_trigger_list_add(session_trigger_list,
1535 trigger_ht_element->trigger);
1536 if (ret) {
1537 goto error;
1538 }
1539
1540 trigger_count++;
1541 }
1542
1543 DBG("[notification-thread] Found %i triggers that apply to newly created session",
1544 trigger_count);
1545 return session_trigger_list;
1546 error:
1547 lttng_session_trigger_list_destroy(session_trigger_list);
1548 return NULL;
1549 }
1550
1551 static
1552 struct session_info *find_or_create_session_info(
1553 struct notification_thread_state *state,
1554 const char *name, uid_t uid, gid_t gid)
1555 {
1556 struct session_info *session = NULL;
1557 struct cds_lfht_node *node;
1558 struct cds_lfht_iter iter;
1559 struct lttng_session_trigger_list *trigger_list;
1560
1561 rcu_read_lock();
1562 cds_lfht_lookup(state->sessions_ht,
1563 hash_key_str(name, lttng_ht_seed),
1564 match_session,
1565 name,
1566 &iter);
1567 node = cds_lfht_iter_get_node(&iter);
1568 if (node) {
1569 DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
1570 name, uid, gid);
1571 session = caa_container_of(node, struct session_info,
1572 sessions_ht_node);
1573 assert(session->uid == uid);
1574 assert(session->gid == gid);
1575 session_info_get(session);
1576 goto end;
1577 }
1578
1579 trigger_list = lttng_session_trigger_list_build(state, name);
1580 if (!trigger_list) {
1581 goto error;
1582 }
1583
1584 session = session_info_create(name, uid, gid, trigger_list,
1585 state->sessions_ht);
1586 if (!session) {
1587 ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1588 name, uid, gid);
1589 lttng_session_trigger_list_destroy(trigger_list);
1590 goto error;
1591 }
1592 trigger_list = NULL;
1593
1594 cds_lfht_add(state->sessions_ht, hash_key_str(name, lttng_ht_seed),
1595 &session->sessions_ht_node);
1596 end:
1597 rcu_read_unlock();
1598 return session;
1599 error:
1600 rcu_read_unlock();
1601 session_info_put(session);
1602 return NULL;
1603 }
1604
1605 static
1606 int handle_notification_thread_command_add_channel(
1607 struct notification_thread_state *state,
1608 const char *session_name, uid_t session_uid, gid_t session_gid,
1609 const char *channel_name, enum lttng_domain_type channel_domain,
1610 uint64_t channel_key_int, uint64_t channel_capacity,
1611 enum lttng_error_code *cmd_result)
1612 {
1613 struct cds_list_head trigger_list;
1614 struct channel_info *new_channel_info = NULL;
1615 struct channel_key channel_key = {
1616 .key = channel_key_int,
1617 .domain = channel_domain,
1618 };
1619 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
1620 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1621 int trigger_count = 0;
1622 struct cds_lfht_iter iter;
1623 struct session_info *session_info = NULL;
1624
1625 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
1626 channel_name, session_name, channel_key_int,
1627 channel_domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1628
1629 CDS_INIT_LIST_HEAD(&trigger_list);
1630
1631 session_info = find_or_create_session_info(state, session_name,
1632 session_uid, session_gid);
1633 if (!session_info) {
1634 /* Allocation error or an internal error occurred. */
1635 goto error;
1636 }
1637
1638 new_channel_info = channel_info_create(channel_name, &channel_key,
1639 channel_capacity, session_info);
1640 if (!new_channel_info) {
1641 goto error;
1642 }
1643
1644 rcu_read_lock();
1645 /* Build a list of all triggers applying to the new channel. */
1646 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1647 node) {
1648 struct lttng_trigger_list_element *new_element;
1649
1650 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
1651 new_channel_info)) {
1652 continue;
1653 }
1654
1655 new_element = zmalloc(sizeof(*new_element));
1656 if (!new_element) {
1657 rcu_read_unlock();
1658 goto error;
1659 }
1660 CDS_INIT_LIST_HEAD(&new_element->node);
1661 new_element->trigger = trigger_ht_element->trigger;
1662 cds_list_add(&new_element->node, &trigger_list);
1663 trigger_count++;
1664 }
1665 rcu_read_unlock();
1666
1667 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
1668 trigger_count);
1669 channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
1670 if (!channel_trigger_list) {
1671 goto error;
1672 }
1673 channel_trigger_list->channel_key = new_channel_info->key;
1674 CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
1675 cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
1676 cds_list_splice(&trigger_list, &channel_trigger_list->list);
1677
1678 rcu_read_lock();
1679 /* Add channel to the channel_ht which owns the channel_infos. */
1680 cds_lfht_add(state->channels_ht,
1681 hash_channel_key(&new_channel_info->key),
1682 &new_channel_info->channels_ht_node);
1683 /*
1684 * Add the list of triggers associated with this channel to the
1685 * channel_triggers_ht.
1686 */
1687 cds_lfht_add(state->channel_triggers_ht,
1688 hash_channel_key(&new_channel_info->key),
1689 &channel_trigger_list->channel_triggers_ht_node);
1690 rcu_read_unlock();
1691 session_info_put(session_info);
1692 *cmd_result = LTTNG_OK;
1693 return 0;
1694 error:
1695 channel_info_destroy(new_channel_info);
1696 session_info_put(session_info);
1697 return 1;
1698 }
1699
1700 static
1701 void free_channel_trigger_list_rcu(struct rcu_head *node)
1702 {
1703 free(caa_container_of(node, struct lttng_channel_trigger_list,
1704 rcu_node));
1705 }
1706
1707 static
1708 void free_channel_state_sample_rcu(struct rcu_head *node)
1709 {
1710 free(caa_container_of(node, struct channel_state_sample,
1711 rcu_node));
1712 }
1713
1714 static
1715 int handle_notification_thread_command_remove_channel(
1716 struct notification_thread_state *state,
1717 uint64_t channel_key, enum lttng_domain_type domain,
1718 enum lttng_error_code *cmd_result)
1719 {
1720 struct cds_lfht_node *node;
1721 struct cds_lfht_iter iter;
1722 struct lttng_channel_trigger_list *trigger_list;
1723 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1724 struct channel_key key = { .key = channel_key, .domain = domain };
1725 struct channel_info *channel_info;
1726
1727 DBG("[notification-thread] Removing channel key = %" PRIu64 " in %s domain",
1728 channel_key, domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1729
1730 rcu_read_lock();
1731
1732 cds_lfht_lookup(state->channel_triggers_ht,
1733 hash_channel_key(&key),
1734 match_channel_trigger_list,
1735 &key,
1736 &iter);
1737 node = cds_lfht_iter_get_node(&iter);
1738 /*
1739 * There is a severe internal error if we are being asked to remove a
1740 * channel that doesn't exist.
1741 */
1742 if (!node) {
1743 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
1744 goto end;
1745 }
1746
1747 /* Free the list of triggers associated with this channel. */
1748 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
1749 channel_triggers_ht_node);
1750 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1751 &trigger_list->list, node) {
1752 cds_list_del(&trigger_list_element->node);
1753 free(trigger_list_element);
1754 }
1755 cds_lfht_del(state->channel_triggers_ht, node);
1756 call_rcu(&trigger_list->rcu_node, free_channel_trigger_list_rcu);
1757
1758 /* Free sampled channel state. */
1759 cds_lfht_lookup(state->channel_state_ht,
1760 hash_channel_key(&key),
1761 match_channel_state_sample,
1762 &key,
1763 &iter);
1764 node = cds_lfht_iter_get_node(&iter);
1765 /*
1766 * This is expected to be NULL if the channel is destroyed before we
1767 * received a sample.
1768 */
1769 if (node) {
1770 struct channel_state_sample *sample = caa_container_of(node,
1771 struct channel_state_sample,
1772 channel_state_ht_node);
1773
1774 cds_lfht_del(state->channel_state_ht, node);
1775 call_rcu(&sample->rcu_node, free_channel_state_sample_rcu);
1776 }
1777
1778 /* Remove the channel from the channels_ht and free it. */
1779 cds_lfht_lookup(state->channels_ht,
1780 hash_channel_key(&key),
1781 match_channel_info,
1782 &key,
1783 &iter);
1784 node = cds_lfht_iter_get_node(&iter);
1785 assert(node);
1786 channel_info = caa_container_of(node, struct channel_info,
1787 channels_ht_node);
1788 cds_lfht_del(state->channels_ht, node);
1789 channel_info_destroy(channel_info);
1790 end:
1791 rcu_read_unlock();
1792 *cmd_result = LTTNG_OK;
1793 return 0;
1794 }
1795
1796 static
1797 int handle_notification_thread_command_session_rotation(
1798 struct notification_thread_state *state,
1799 enum notification_thread_command_type cmd_type,
1800 const char *session_name, uid_t session_uid, gid_t session_gid,
1801 uint64_t trace_archive_chunk_id,
1802 struct lttng_trace_archive_location *location,
1803 enum lttng_error_code *_cmd_result)
1804 {
1805 int ret = 0;
1806 enum lttng_error_code cmd_result = LTTNG_OK;
1807 struct lttng_session_trigger_list *trigger_list;
1808 struct lttng_trigger_list_element *trigger_list_element;
1809 struct session_info *session_info;
1810 const struct lttng_credentials session_creds = {
1811 .uid = LTTNG_OPTIONAL_INIT_VALUE(session_uid),
1812 .gid = LTTNG_OPTIONAL_INIT_VALUE(session_gid),
1813 };
1814
1815 rcu_read_lock();
1816
1817 session_info = find_or_create_session_info(state, session_name,
1818 session_uid, session_gid);
1819 if (!session_info) {
1820 /* Allocation error or an internal error occurred. */
1821 ret = -1;
1822 cmd_result = LTTNG_ERR_NOMEM;
1823 goto end;
1824 }
1825
1826 session_info->rotation.ongoing =
1827 cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING;
1828 session_info->rotation.id = trace_archive_chunk_id;
1829 trigger_list = get_session_trigger_list(state, session_name);
1830 if (!trigger_list) {
1831 DBG("[notification-thread] No triggers applying to session \"%s\" found",
1832 session_name);
1833 goto end;
1834 }
1835
1836 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
1837 node) {
1838 const struct lttng_condition *condition;
1839 struct lttng_trigger *trigger;
1840 struct notification_client_list *client_list;
1841 struct lttng_evaluation *evaluation = NULL;
1842 enum lttng_condition_type condition_type;
1843 enum action_executor_status executor_status;
1844
1845 trigger = trigger_list_element->trigger;
1846 condition = lttng_trigger_get_const_condition(trigger);
1847 assert(condition);
1848 condition_type = lttng_condition_get_type(condition);
1849
1850 if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING &&
1851 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1852 continue;
1853 } else if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED &&
1854 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED) {
1855 continue;
1856 }
1857
1858 client_list = get_client_list_from_condition(state, condition);
1859 if (cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1860 evaluation = lttng_evaluation_session_rotation_ongoing_create(
1861 trace_archive_chunk_id);
1862 } else {
1863 evaluation = lttng_evaluation_session_rotation_completed_create(
1864 trace_archive_chunk_id, location);
1865 }
1866
1867 if (!evaluation) {
1868 /* Internal error */
1869 ret = -1;
1870 cmd_result = LTTNG_ERR_UNK;
1871 goto put_list;
1872 }
1873
1874 /*
1875 * Ownership of `evaluation` transferred to the action executor
1876 * no matter the result.
1877 */
1878 executor_status = action_executor_enqueue(state->executor,
1879 trigger, evaluation, &session_creds,
1880 client_list);
1881 evaluation = NULL;
1882 switch (executor_status) {
1883 case ACTION_EXECUTOR_STATUS_OK:
1884 break;
1885 case ACTION_EXECUTOR_STATUS_ERROR:
1886 case ACTION_EXECUTOR_STATUS_INVALID:
1887 /*
1888 * TODO Add trigger identification (name/id) when
1889 * it is added to the API.
1890 */
1891 ERR("Fatal error occurred while enqueuing action associated with session rotation trigger");
1892 ret = -1;
1893 goto put_list;
1894 case ACTION_EXECUTOR_STATUS_OVERFLOW:
1895 /*
1896 * TODO Add trigger identification (name/id) when
1897 * it is added to the API.
1898 *
1899 * Not a fatal error.
1900 */
1901 WARN("No space left when enqueuing action associated with session rotation trigger");
1902 ret = 0;
1903 goto put_list;
1904 default:
1905 abort();
1906 }
1907
1908 put_list:
1909 notification_client_list_put(client_list);
1910 if (caa_unlikely(ret)) {
1911 break;
1912 }
1913 }
1914 end:
1915 session_info_put(session_info);
1916 *_cmd_result = cmd_result;
1917 rcu_read_unlock();
1918 return ret;
1919 }
1920
1921 static
1922 int condition_is_supported(struct lttng_condition *condition)
1923 {
1924 int ret;
1925
1926 switch (lttng_condition_get_type(condition)) {
1927 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1928 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1929 {
1930 enum lttng_domain_type domain;
1931
1932 ret = lttng_condition_buffer_usage_get_domain_type(condition,
1933 &domain);
1934 if (ret) {
1935 ret = -1;
1936 goto end;
1937 }
1938
1939 if (domain != LTTNG_DOMAIN_KERNEL) {
1940 ret = 1;
1941 goto end;
1942 }
1943
1944 /*
1945 * Older kernel tracers don't expose the API to monitor their
1946 * buffers. Therefore, we reject triggers that require that
1947 * mechanism to be available to be evaluated.
1948 */
1949 ret = kernel_supports_ring_buffer_snapshot_sample_positions();
1950 break;
1951 }
1952 default:
1953 ret = 1;
1954 }
1955 end:
1956 return ret;
1957 }
1958
1959 /* Must be called with RCU read lock held. */
1960 static
1961 int bind_trigger_to_matching_session(struct lttng_trigger *trigger,
1962 struct notification_thread_state *state)
1963 {
1964 int ret = 0;
1965 const struct lttng_condition *condition;
1966 const char *session_name;
1967 struct lttng_session_trigger_list *trigger_list;
1968
1969 condition = lttng_trigger_get_const_condition(trigger);
1970 switch (lttng_condition_get_type(condition)) {
1971 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
1972 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
1973 {
1974 enum lttng_condition_status status;
1975
1976 status = lttng_condition_session_rotation_get_session_name(
1977 condition, &session_name);
1978 if (status != LTTNG_CONDITION_STATUS_OK) {
1979 ERR("[notification-thread] Failed to bind trigger to session: unable to get 'session_rotation' condition's session name");
1980 ret = -1;
1981 goto end;
1982 }
1983 break;
1984 }
1985 default:
1986 ret = -1;
1987 goto end;
1988 }
1989
1990 trigger_list = get_session_trigger_list(state, session_name);
1991 if (!trigger_list) {
1992 DBG("[notification-thread] Unable to bind trigger applying to session \"%s\" as it is not yet known to the notification system",
1993 session_name);
1994 goto end;
1995
1996 }
1997
1998 DBG("[notification-thread] Newly registered trigger bound to session \"%s\"",
1999 session_name);
2000 ret = lttng_session_trigger_list_add(trigger_list, trigger);
2001 end:
2002 return ret;
2003 }
2004
2005 /* Must be called with RCU read lock held. */
2006 static
2007 int bind_trigger_to_matching_channels(struct lttng_trigger *trigger,
2008 struct notification_thread_state *state)
2009 {
2010 int ret = 0;
2011 struct cds_lfht_node *node;
2012 struct cds_lfht_iter iter;
2013 struct channel_info *channel;
2014
2015 cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
2016 channels_ht_node) {
2017 struct lttng_trigger_list_element *trigger_list_element;
2018 struct lttng_channel_trigger_list *trigger_list;
2019 struct cds_lfht_iter lookup_iter;
2020
2021 if (!trigger_applies_to_channel(trigger, channel)) {
2022 continue;
2023 }
2024
2025 cds_lfht_lookup(state->channel_triggers_ht,
2026 hash_channel_key(&channel->key),
2027 match_channel_trigger_list,
2028 &channel->key,
2029 &lookup_iter);
2030 node = cds_lfht_iter_get_node(&lookup_iter);
2031 assert(node);
2032 trigger_list = caa_container_of(node,
2033 struct lttng_channel_trigger_list,
2034 channel_triggers_ht_node);
2035
2036 trigger_list_element = zmalloc(sizeof(*trigger_list_element));
2037 if (!trigger_list_element) {
2038 ret = -1;
2039 goto end;
2040 }
2041 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
2042 trigger_list_element->trigger = trigger;
2043 cds_list_add(&trigger_list_element->node, &trigger_list->list);
2044 DBG("[notification-thread] Newly registered trigger bound to channel \"%s\"",
2045 channel->name);
2046 }
2047 end:
2048 return ret;
2049 }
2050
2051 static
2052 bool is_trigger_action_notify(const struct lttng_trigger *trigger)
2053 {
2054 bool is_notify = false;
2055 unsigned int i, count;
2056 enum lttng_action_status action_status;
2057 const struct lttng_action *action =
2058 lttng_trigger_get_const_action(trigger);
2059 enum lttng_action_type action_type;
2060
2061 assert(action);
2062 action_type = lttng_action_get_type(action);
2063 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2064 is_notify = true;
2065 goto end;
2066 } else if (action_type != LTTNG_ACTION_TYPE_GROUP) {
2067 goto end;
2068 }
2069
2070 action_status = lttng_action_group_get_count(action, &count);
2071 assert(action_status == LTTNG_ACTION_STATUS_OK);
2072
2073 for (i = 0; i < count; i++) {
2074 const struct lttng_action *inner_action =
2075 lttng_action_group_get_at_index(
2076 action, i);
2077
2078 action_type = lttng_action_get_type(inner_action);
2079 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2080 is_notify = true;
2081 goto end;
2082 }
2083 }
2084
2085 end:
2086 return is_notify;
2087 }
2088
2089 /*
2090 * FIXME A client's credentials are not checked when registering a trigger.
2091 *
2092 * The effects of this are benign since:
2093 * - The client will succeed in registering the trigger, as it is valid,
2094 * - The trigger will, internally, be bound to the channel/session,
2095 * - The notifications will not be sent since the client's credentials
2096 * are checked against the channel at that moment.
2097 *
2098 * If this function returns a non-zero value, it means something is
2099 * fundamentally broken and the whole subsystem/thread will be torn down.
2100 *
2101 * If a non-fatal error occurs, just set the cmd_result to the appropriate
2102 * error code.
2103 */
2104 static
2105 int handle_notification_thread_command_register_trigger(
2106 struct notification_thread_state *state,
2107 struct lttng_trigger *trigger,
2108 enum lttng_error_code *cmd_result)
2109 {
2110 int ret = 0;
2111 struct lttng_condition *condition;
2112 struct notification_client *client;
2113 struct notification_client_list *client_list = NULL;
2114 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2115 struct notification_client_list_element *client_list_element;
2116 struct cds_lfht_node *node;
2117 struct cds_lfht_iter iter;
2118 bool free_trigger = true;
2119 struct lttng_evaluation *evaluation = NULL;
2120 struct lttng_credentials object_creds;
2121 uid_t object_uid;
2122 gid_t object_gid;
2123 enum action_executor_status executor_status;
2124
2125 rcu_read_lock();
2126
2127 condition = lttng_trigger_get_condition(trigger);
2128 assert(condition);
2129
2130 ret = condition_is_supported(condition);
2131 if (ret < 0) {
2132 goto error;
2133 } else if (ret == 0) {
2134 *cmd_result = LTTNG_ERR_NOT_SUPPORTED;
2135 goto error;
2136 } else {
2137 /* Feature is supported, continue. */
2138 ret = 0;
2139 }
2140
2141 trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
2142 if (!trigger_ht_element) {
2143 ret = -1;
2144 goto error;
2145 }
2146
2147 /* Add trigger to the trigger_ht. */
2148 cds_lfht_node_init(&trigger_ht_element->node);
2149 trigger_ht_element->trigger = trigger;
2150
2151 node = cds_lfht_add_unique(state->triggers_ht,
2152 lttng_condition_hash(condition),
2153 match_condition,
2154 condition,
2155 &trigger_ht_element->node);
2156 if (node != &trigger_ht_element->node) {
2157 /* Not a fatal error, simply report it to the client. */
2158 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2159 goto error_free_ht_element;
2160 }
2161
2162 /*
2163 * Ownership of the trigger and of its wrapper was transfered to
2164 * the triggers_ht.
2165 */
2166 trigger_ht_element = NULL;
2167 free_trigger = false;
2168
2169 /*
2170 * The rest only applies to triggers that have a "notify" action.
2171 * It is not skipped as this is the only action type currently
2172 * supported.
2173 */
2174 if (is_trigger_action_notify(trigger)) {
2175 client_list = notification_client_list_create(trigger);
2176 if (!client_list) {
2177 ret = -1;
2178 goto error_free_ht_element;
2179 }
2180
2181 /* Build a list of clients to which this new trigger applies. */
2182 cds_lfht_for_each_entry (state->client_socket_ht, &iter, client,
2183 client_socket_ht_node) {
2184 if (!trigger_applies_to_client(trigger, client)) {
2185 continue;
2186 }
2187
2188 client_list_element =
2189 zmalloc(sizeof(*client_list_element));
2190 if (!client_list_element) {
2191 ret = -1;
2192 goto error_put_client_list;
2193 }
2194
2195 CDS_INIT_LIST_HEAD(&client_list_element->node);
2196 client_list_element->client = client;
2197 cds_list_add(&client_list_element->node,
2198 &client_list->list);
2199 }
2200
2201 /*
2202 * Client list ownership transferred to the
2203 * notification_trigger_clients_ht.
2204 */
2205 publish_notification_client_list(state, client_list);
2206 }
2207
2208 switch (get_condition_binding_object(condition)) {
2209 case LTTNG_OBJECT_TYPE_SESSION:
2210 /* Add the trigger to the list if it matches a known session. */
2211 ret = bind_trigger_to_matching_session(trigger, state);
2212 if (ret) {
2213 goto error_put_client_list;
2214 }
2215 break;
2216 case LTTNG_OBJECT_TYPE_CHANNEL:
2217 /*
2218 * Add the trigger to list of triggers bound to the channels
2219 * currently known.
2220 */
2221 ret = bind_trigger_to_matching_channels(trigger, state);
2222 if (ret) {
2223 goto error_put_client_list;
2224 }
2225 break;
2226 case LTTNG_OBJECT_TYPE_NONE:
2227 break;
2228 default:
2229 ERR("Unknown object type on which to bind a newly registered trigger was encountered");
2230 ret = -1;
2231 goto error_put_client_list;
2232 }
2233
2234 /*
2235 * The new trigger's condition must be evaluated against the current
2236 * state.
2237 *
2238 * In the case of `notify` action, nothing preventing clients from
2239 * subscribing to a condition before the corresponding trigger is
2240 * registered, we have to evaluate this new condition right away.
2241 *
2242 * At some point, we were waiting for the next "evaluation" (e.g. on
2243 * reception of a channel sample) to evaluate this new condition, but
2244 * that was broken.
2245 *
2246 * The reason it was broken is that waiting for the next sample
2247 * does not allow us to properly handle transitions for edge-triggered
2248 * conditions.
2249 *
2250 * Consider this example: when we handle a new channel sample, we
2251 * evaluate each conditions twice: once with the previous state, and
2252 * again with the newest state. We then use those two results to
2253 * determine whether a state change happened: a condition was false and
2254 * became true. If a state change happened, we have to notify clients.
2255 *
2256 * Now, if a client subscribes to a given notification and registers
2257 * a trigger *after* that subscription, we have to make sure the
2258 * condition is evaluated at this point while considering only the
2259 * current state. Otherwise, the next evaluation cycle may only see
2260 * that the evaluations remain the same (true for samples n-1 and n) and
2261 * the client will never know that the condition has been met.
2262 */
2263 switch (get_condition_binding_object(condition)) {
2264 case LTTNG_OBJECT_TYPE_SESSION:
2265 ret = evaluate_session_condition_for_client(condition, state,
2266 &evaluation, &object_uid,
2267 &object_gid);
2268 break;
2269 case LTTNG_OBJECT_TYPE_CHANNEL:
2270 ret = evaluate_channel_condition_for_client(condition, state,
2271 &evaluation, &object_uid,
2272 &object_gid);
2273 break;
2274 case LTTNG_OBJECT_TYPE_NONE:
2275 ret = 0;
2276 goto error_put_client_list;
2277 case LTTNG_OBJECT_TYPE_UNKNOWN:
2278 default:
2279 ret = -1;
2280 goto error_put_client_list;
2281 }
2282
2283 if (ret) {
2284 /* Fatal error. */
2285 goto error_put_client_list;
2286 }
2287
2288 LTTNG_OPTIONAL_SET(&object_creds.uid, object_uid);
2289 LTTNG_OPTIONAL_SET(&object_creds.gid, object_gid);
2290
2291 DBG("Newly registered trigger's condition evaluated to %s",
2292 evaluation ? "true" : "false");
2293 if (!evaluation) {
2294 /* Evaluation yielded nothing. Normal exit. */
2295 ret = 0;
2296 goto error_put_client_list;
2297 }
2298
2299 /*
2300 * Ownership of `evaluation` transferred to the action executor
2301 * no matter the result.
2302 */
2303 executor_status = action_executor_enqueue(state->executor, trigger,
2304 evaluation, &object_creds, client_list);
2305 evaluation = NULL;
2306 switch (executor_status) {
2307 case ACTION_EXECUTOR_STATUS_OK:
2308 break;
2309 case ACTION_EXECUTOR_STATUS_ERROR:
2310 case ACTION_EXECUTOR_STATUS_INVALID:
2311 /*
2312 * TODO Add trigger identification (name/id) when
2313 * it is added to the API.
2314 */
2315 ERR("Fatal error occurred while enqueuing action associated to newly registered trigger");
2316 ret = -1;
2317 goto error_put_client_list;
2318 case ACTION_EXECUTOR_STATUS_OVERFLOW:
2319 /*
2320 * TODO Add trigger identification (name/id) when
2321 * it is added to the API.
2322 *
2323 * Not a fatal error.
2324 */
2325 WARN("No space left when enqueuing action associated to newly registered trigger");
2326 ret = 0;
2327 goto error_put_client_list;
2328 default:
2329 abort();
2330 }
2331
2332 *cmd_result = LTTNG_OK;
2333
2334 error_put_client_list:
2335 notification_client_list_put(client_list);
2336
2337 error_free_ht_element:
2338 free(trigger_ht_element);
2339 error:
2340 if (free_trigger) {
2341 lttng_trigger_destroy(trigger);
2342 }
2343 rcu_read_unlock();
2344 return ret;
2345 }
2346
2347 static
2348 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node)
2349 {
2350 free(caa_container_of(node, struct lttng_trigger_ht_element,
2351 rcu_node));
2352 }
2353
2354 static
2355 int handle_notification_thread_command_unregister_trigger(
2356 struct notification_thread_state *state,
2357 struct lttng_trigger *trigger,
2358 enum lttng_error_code *_cmd_reply)
2359 {
2360 struct cds_lfht_iter iter;
2361 struct cds_lfht_node *triggers_ht_node;
2362 struct lttng_channel_trigger_list *trigger_list;
2363 struct notification_client_list *client_list;
2364 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2365 struct lttng_condition *condition = lttng_trigger_get_condition(
2366 trigger);
2367 enum lttng_error_code cmd_reply;
2368
2369 rcu_read_lock();
2370
2371 cds_lfht_lookup(state->triggers_ht,
2372 lttng_condition_hash(condition),
2373 match_condition,
2374 condition,
2375 &iter);
2376 triggers_ht_node = cds_lfht_iter_get_node(&iter);
2377 if (!triggers_ht_node) {
2378 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
2379 goto end;
2380 } else {
2381 cmd_reply = LTTNG_OK;
2382 }
2383
2384 /* Remove trigger from channel_triggers_ht. */
2385 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
2386 channel_triggers_ht_node) {
2387 struct lttng_trigger_list_element *trigger_element, *tmp;
2388
2389 cds_list_for_each_entry_safe(trigger_element, tmp,
2390 &trigger_list->list, node) {
2391 const struct lttng_condition *current_condition =
2392 lttng_trigger_get_const_condition(
2393 trigger_element->trigger);
2394
2395 assert(current_condition);
2396 if (!lttng_condition_is_equal(condition,
2397 current_condition)) {
2398 continue;
2399 }
2400
2401 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
2402 cds_list_del(&trigger_element->node);
2403 /* A trigger can only appear once per channel */
2404 break;
2405 }
2406 }
2407
2408 /*
2409 * Remove and release the client list from
2410 * notification_trigger_clients_ht.
2411 */
2412 client_list = get_client_list_from_condition(state, condition);
2413 assert(client_list);
2414
2415 /* Put new reference and the hashtable's reference. */
2416 notification_client_list_put(client_list);
2417 notification_client_list_put(client_list);
2418 client_list = NULL;
2419
2420 /* Remove trigger from triggers_ht. */
2421 trigger_ht_element = caa_container_of(triggers_ht_node,
2422 struct lttng_trigger_ht_element, node);
2423 cds_lfht_del(state->triggers_ht, triggers_ht_node);
2424
2425 /* Release the ownership of the trigger. */
2426 lttng_trigger_destroy(trigger_ht_element->trigger);
2427 call_rcu(&trigger_ht_element->rcu_node, free_lttng_trigger_ht_element_rcu);
2428 end:
2429 rcu_read_unlock();
2430 if (_cmd_reply) {
2431 *_cmd_reply = cmd_reply;
2432 }
2433 return 0;
2434 }
2435
2436 /* Returns 0 on success, 1 on exit requested, negative value on error. */
2437 int handle_notification_thread_command(
2438 struct notification_thread_handle *handle,
2439 struct notification_thread_state *state)
2440 {
2441 int ret;
2442 uint64_t counter;
2443 struct notification_thread_command *cmd;
2444
2445 /* Read the event pipe to put it back into a quiescent state. */
2446 ret = lttng_read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
2447 sizeof(counter));
2448 if (ret != sizeof(counter)) {
2449 goto error;
2450 }
2451
2452 pthread_mutex_lock(&handle->cmd_queue.lock);
2453 cmd = cds_list_first_entry(&handle->cmd_queue.list,
2454 struct notification_thread_command, cmd_list_node);
2455 cds_list_del(&cmd->cmd_list_node);
2456 pthread_mutex_unlock(&handle->cmd_queue.lock);
2457 switch (cmd->type) {
2458 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
2459 DBG("[notification-thread] Received register trigger command");
2460 ret = handle_notification_thread_command_register_trigger(
2461 state, cmd->parameters.trigger,
2462 &cmd->reply_code);
2463 break;
2464 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
2465 DBG("[notification-thread] Received unregister trigger command");
2466 ret = handle_notification_thread_command_unregister_trigger(
2467 state, cmd->parameters.trigger,
2468 &cmd->reply_code);
2469 break;
2470 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
2471 DBG("[notification-thread] Received add channel command");
2472 ret = handle_notification_thread_command_add_channel(
2473 state,
2474 cmd->parameters.add_channel.session.name,
2475 cmd->parameters.add_channel.session.uid,
2476 cmd->parameters.add_channel.session.gid,
2477 cmd->parameters.add_channel.channel.name,
2478 cmd->parameters.add_channel.channel.domain,
2479 cmd->parameters.add_channel.channel.key,
2480 cmd->parameters.add_channel.channel.capacity,
2481 &cmd->reply_code);
2482 break;
2483 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
2484 DBG("[notification-thread] Received remove channel command");
2485 ret = handle_notification_thread_command_remove_channel(
2486 state, cmd->parameters.remove_channel.key,
2487 cmd->parameters.remove_channel.domain,
2488 &cmd->reply_code);
2489 break;
2490 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
2491 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
2492 DBG("[notification-thread] Received session rotation %s command",
2493 cmd->type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING ?
2494 "ongoing" : "completed");
2495 ret = handle_notification_thread_command_session_rotation(
2496 state,
2497 cmd->type,
2498 cmd->parameters.session_rotation.session_name,
2499 cmd->parameters.session_rotation.uid,
2500 cmd->parameters.session_rotation.gid,
2501 cmd->parameters.session_rotation.trace_archive_chunk_id,
2502 cmd->parameters.session_rotation.location,
2503 &cmd->reply_code);
2504 break;
2505 case NOTIFICATION_COMMAND_TYPE_QUIT:
2506 DBG("[notification-thread] Received quit command");
2507 cmd->reply_code = LTTNG_OK;
2508 ret = 1;
2509 goto end;
2510 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE:
2511 {
2512 const enum client_transmission_status client_status =
2513 cmd->parameters.client_communication_update
2514 .status;
2515 const notification_client_id client_id =
2516 cmd->parameters.client_communication_update.id;
2517 struct notification_client *client;
2518
2519 rcu_read_lock();
2520 client = get_client_from_id(client_id, state);
2521
2522 if (!client) {
2523 /*
2524 * Client error was probably already picked-up by the
2525 * notification thread or it has disconnected
2526 * gracefully while this command was queued.
2527 */
2528 DBG("Failed to find notification client to update communication status, client id = %" PRIu64,
2529 client_id);
2530 ret = 0;
2531 } else {
2532 ret = client_handle_transmission_status(
2533 client, client_status, state);
2534 }
2535 rcu_read_unlock();
2536 break;
2537 }
2538 default:
2539 ERR("[notification-thread] Unknown internal command received");
2540 goto error_unlock;
2541 }
2542
2543 if (ret) {
2544 goto error_unlock;
2545 }
2546 end:
2547 if (cmd->is_async) {
2548 free(cmd);
2549 cmd = NULL;
2550 } else {
2551 lttng_waiter_wake_up(&cmd->reply_waiter);
2552 }
2553 return ret;
2554 error_unlock:
2555 /* Wake-up and return a fatal error to the calling thread. */
2556 lttng_waiter_wake_up(&cmd->reply_waiter);
2557 cmd->reply_code = LTTNG_ERR_FATAL;
2558 error:
2559 /* Indicate a fatal error to the caller. */
2560 return -1;
2561 }
2562
2563 static
2564 int socket_set_non_blocking(int socket)
2565 {
2566 int ret, flags;
2567
2568 /* Set the pipe as non-blocking. */
2569 ret = fcntl(socket, F_GETFL, 0);
2570 if (ret == -1) {
2571 PERROR("fcntl get socket flags");
2572 goto end;
2573 }
2574 flags = ret;
2575
2576 ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
2577 if (ret == -1) {
2578 PERROR("fcntl set O_NONBLOCK socket flag");
2579 goto end;
2580 }
2581 DBG("Client socket (fd = %i) set as non-blocking", socket);
2582 end:
2583 return ret;
2584 }
2585
2586 static
2587 int client_reset_inbound_state(struct notification_client *client)
2588 {
2589 int ret;
2590
2591
2592 lttng_payload_clear(&client->communication.inbound.payload);
2593
2594 client->communication.inbound.bytes_to_receive =
2595 sizeof(struct lttng_notification_channel_message);
2596 client->communication.inbound.msg_type =
2597 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
2598 LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
2599 LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
2600 ret = lttng_dynamic_buffer_set_size(
2601 &client->communication.inbound.payload.buffer,
2602 client->communication.inbound.bytes_to_receive);
2603
2604 return ret;
2605 }
2606
2607 int handle_notification_thread_client_connect(
2608 struct notification_thread_state *state)
2609 {
2610 int ret;
2611 struct notification_client *client;
2612
2613 DBG("[notification-thread] Handling new notification channel client connection");
2614
2615 client = zmalloc(sizeof(*client));
2616 if (!client) {
2617 /* Fatal error. */
2618 ret = -1;
2619 goto error;
2620 }
2621
2622 pthread_mutex_init(&client->lock, NULL);
2623 client->id = state->next_notification_client_id++;
2624 CDS_INIT_LIST_HEAD(&client->condition_list);
2625 lttng_payload_init(&client->communication.inbound.payload);
2626 lttng_payload_init(&client->communication.outbound.payload);
2627 client->communication.inbound.expect_creds = true;
2628
2629 ret = client_reset_inbound_state(client);
2630 if (ret) {
2631 ERR("[notification-thread] Failed to reset client communication's inbound state");
2632 ret = 0;
2633 goto error;
2634 }
2635
2636 ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
2637 if (ret < 0) {
2638 ERR("[notification-thread] Failed to accept new notification channel client connection");
2639 ret = 0;
2640 goto error;
2641 }
2642
2643 client->socket = ret;
2644
2645 ret = socket_set_non_blocking(client->socket);
2646 if (ret) {
2647 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
2648 goto error;
2649 }
2650
2651 ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
2652 if (ret < 0) {
2653 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
2654 ret = 0;
2655 goto error;
2656 }
2657
2658 ret = lttng_poll_add(&state->events, client->socket,
2659 LPOLLIN | LPOLLERR |
2660 LPOLLHUP | LPOLLRDHUP);
2661 if (ret < 0) {
2662 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
2663 ret = 0;
2664 goto error;
2665 }
2666 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
2667 client->socket);
2668
2669 rcu_read_lock();
2670 cds_lfht_add(state->client_socket_ht,
2671 hash_client_socket(client->socket),
2672 &client->client_socket_ht_node);
2673 cds_lfht_add(state->client_id_ht,
2674 hash_client_id(client->id),
2675 &client->client_id_ht_node);
2676 rcu_read_unlock();
2677
2678 return ret;
2679
2680 error:
2681 notification_client_destroy(client, state);
2682 return ret;
2683 }
2684
2685 /*
2686 * RCU read-lock must be held by the caller.
2687 * Client lock must _not_ be held by the caller.
2688 */
2689 static
2690 int notification_thread_client_disconnect(
2691 struct notification_client *client,
2692 struct notification_thread_state *state)
2693 {
2694 int ret;
2695 struct lttng_condition_list_element *condition_list_element, *tmp;
2696
2697 /* Acquire the client lock to disable its communication atomically. */
2698 pthread_mutex_lock(&client->lock);
2699 client->communication.active = false;
2700 cds_lfht_del(state->client_socket_ht, &client->client_socket_ht_node);
2701 cds_lfht_del(state->client_id_ht, &client->client_id_ht_node);
2702 pthread_mutex_unlock(&client->lock);
2703
2704 ret = lttng_poll_del(&state->events, client->socket);
2705 if (ret) {
2706 ERR("[notification-thread] Failed to remove client socket %d from poll set",
2707 client->socket);
2708 }
2709
2710 /* Release all conditions to which the client was subscribed. */
2711 cds_list_for_each_entry_safe(condition_list_element, tmp,
2712 &client->condition_list, node) {
2713 (void) notification_thread_client_unsubscribe(client,
2714 condition_list_element->condition, state, NULL);
2715 }
2716
2717 /*
2718 * Client no longer accessible to other threads (through the
2719 * client lists).
2720 */
2721 notification_client_destroy(client, state);
2722 return ret;
2723 }
2724
2725 int handle_notification_thread_client_disconnect(
2726 int client_socket, struct notification_thread_state *state)
2727 {
2728 int ret = 0;
2729 struct notification_client *client;
2730
2731 rcu_read_lock();
2732 DBG("[notification-thread] Closing client connection (socket fd = %i)",
2733 client_socket);
2734 client = get_client_from_socket(client_socket, state);
2735 if (!client) {
2736 /* Internal state corruption, fatal error. */
2737 ERR("[notification-thread] Unable to find client (socket fd = %i)",
2738 client_socket);
2739 ret = -1;
2740 goto end;
2741 }
2742
2743 ret = notification_thread_client_disconnect(client, state);
2744 end:
2745 rcu_read_unlock();
2746 return ret;
2747 }
2748
2749 int handle_notification_thread_client_disconnect_all(
2750 struct notification_thread_state *state)
2751 {
2752 struct cds_lfht_iter iter;
2753 struct notification_client *client;
2754 bool error_encoutered = false;
2755
2756 rcu_read_lock();
2757 DBG("[notification-thread] Closing all client connections");
2758 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
2759 client_socket_ht_node) {
2760 int ret;
2761
2762 ret = notification_thread_client_disconnect(
2763 client, state);
2764 if (ret) {
2765 error_encoutered = true;
2766 }
2767 }
2768 rcu_read_unlock();
2769 return error_encoutered ? 1 : 0;
2770 }
2771
2772 int handle_notification_thread_trigger_unregister_all(
2773 struct notification_thread_state *state)
2774 {
2775 bool error_occurred = false;
2776 struct cds_lfht_iter iter;
2777 struct lttng_trigger_ht_element *trigger_ht_element;
2778
2779 rcu_read_lock();
2780 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
2781 node) {
2782 int ret = handle_notification_thread_command_unregister_trigger(
2783 state, trigger_ht_element->trigger, NULL);
2784 if (ret) {
2785 error_occurred = true;
2786 }
2787 }
2788 rcu_read_unlock();
2789 return error_occurred ? -1 : 0;
2790 }
2791
2792 static
2793 int client_handle_transmission_status(
2794 struct notification_client *client,
2795 enum client_transmission_status transmission_status,
2796 struct notification_thread_state *state)
2797 {
2798 int ret = 0;
2799
2800 switch (transmission_status) {
2801 case CLIENT_TRANSMISSION_STATUS_COMPLETE:
2802 ret = lttng_poll_mod(&state->events, client->socket,
2803 CLIENT_POLL_MASK_IN);
2804 if (ret) {
2805 goto end;
2806 }
2807
2808 break;
2809 case CLIENT_TRANSMISSION_STATUS_QUEUED:
2810 /*
2811 * We want to be notified whenever there is buffer space
2812 * available to send the rest of the payload.
2813 */
2814 ret = lttng_poll_mod(&state->events, client->socket,
2815 CLIENT_POLL_MASK_IN_OUT);
2816 if (ret) {
2817 goto end;
2818 }
2819 break;
2820 case CLIENT_TRANSMISSION_STATUS_FAIL:
2821 ret = notification_thread_client_disconnect(client, state);
2822 if (ret) {
2823 goto end;
2824 }
2825 break;
2826 case CLIENT_TRANSMISSION_STATUS_ERROR:
2827 ret = -1;
2828 goto end;
2829 default:
2830 abort();
2831 }
2832 end:
2833 return ret;
2834 }
2835
2836 /* Client lock must be acquired by caller. */
2837 static
2838 enum client_transmission_status client_flush_outgoing_queue(
2839 struct notification_client *client)
2840 {
2841 ssize_t ret;
2842 size_t to_send_count;
2843 enum client_transmission_status status;
2844 struct lttng_payload_view pv = lttng_payload_view_from_payload(
2845 &client->communication.outbound.payload, 0, -1);
2846 const int fds_to_send_count =
2847 lttng_payload_view_get_fd_handle_count(&pv);
2848
2849 ASSERT_LOCKED(client->lock);
2850
2851 if (!client->communication.active) {
2852 status = CLIENT_TRANSMISSION_STATUS_FAIL;
2853 goto end;
2854 }
2855
2856 if (pv.buffer.size == 0) {
2857 /*
2858 * If both data and fds are equal to zero, we are in an invalid
2859 * state.
2860 */
2861 assert(fds_to_send_count != 0);
2862 goto send_fds;
2863 }
2864
2865 /* Send data. */
2866 to_send_count = pv.buffer.size;
2867 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
2868 client->socket);
2869
2870 ret = lttcomm_send_unix_sock_non_block(client->socket,
2871 pv.buffer.data,
2872 to_send_count);
2873 if ((ret >= 0 && ret < to_send_count)) {
2874 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
2875 client->socket);
2876 to_send_count -= max(ret, 0);
2877
2878 memmove(client->communication.outbound.payload.buffer.data,
2879 pv.buffer.data +
2880 pv.buffer.size - to_send_count,
2881 to_send_count);
2882 ret = lttng_dynamic_buffer_set_size(
2883 &client->communication.outbound.payload.buffer,
2884 to_send_count);
2885 if (ret) {
2886 goto error;
2887 }
2888
2889 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
2890 goto end;
2891 } else if (ret < 0) {
2892 /* Generic error, disable the client's communication. */
2893 ERR("[notification-thread] Failed to flush outgoing queue, disconnecting client (socket fd = %i)",
2894 client->socket);
2895 client->communication.active = false;
2896 status = CLIENT_TRANSMISSION_STATUS_FAIL;
2897 goto end;
2898 } else {
2899 /*
2900 * No error and flushed the queue completely.
2901 *
2902 * The payload buffer size is used later to
2903 * check if there is notifications queued. So albeit that the
2904 * direct caller knows that the transmission is complete, we
2905 * need to set the buffer size to zero.
2906 */
2907 ret = lttng_dynamic_buffer_set_size(
2908 &client->communication.outbound.payload.buffer, 0);
2909 if (ret) {
2910 goto error;
2911 }
2912 }
2913
2914 send_fds:
2915 /* No fds to send, transmission is complete. */
2916 if (fds_to_send_count == 0) {
2917 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
2918 goto end;
2919 }
2920
2921 ret = lttcomm_send_payload_view_fds_unix_sock_non_block(
2922 client->socket, &pv);
2923 if (ret < 0) {
2924 /* Generic error, disable the client's communication. */
2925 ERR("[notification-thread] Failed to flush outgoing fds queue, disconnecting client (socket fd = %i)",
2926 client->socket);
2927 client->communication.active = false;
2928 status = CLIENT_TRANSMISSION_STATUS_FAIL;
2929 goto end;
2930 } else if (ret == 0) {
2931 /* Nothing could be sent. */
2932 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
2933 } else {
2934 /* Fd passing is an all or nothing kind of thing. */
2935 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
2936 /*
2937 * The payload _fd_array count is used later to
2938 * check if there is notifications queued. So although the
2939 * direct caller knows that the transmission is complete, we
2940 * need to clear the _fd_array for the queuing check.
2941 */
2942 lttng_dynamic_pointer_array_clear(
2943 &client->communication.outbound.payload
2944 ._fd_handles);
2945 }
2946
2947 end:
2948 if (status == CLIENT_TRANSMISSION_STATUS_COMPLETE) {
2949 client->communication.outbound.queued_command_reply = false;
2950 client->communication.outbound.dropped_notification = false;
2951 lttng_payload_clear(&client->communication.outbound.payload);
2952 }
2953
2954 return status;
2955 error:
2956 return CLIENT_TRANSMISSION_STATUS_ERROR;
2957 }
2958
2959 static
2960 bool client_has_outbound_data_left(
2961 const struct notification_client *client)
2962 {
2963 const struct lttng_payload_view pv = lttng_payload_view_from_payload(
2964 &client->communication.outbound.payload, 0, -1);
2965 const bool has_data = pv.buffer.size != 0;
2966 const bool has_fds = lttng_payload_view_get_fd_handle_count(&pv);
2967
2968 return has_data || has_fds;
2969 }
2970
2971 /* Client lock must _not_ be held by the caller. */
2972 static
2973 int client_send_command_reply(struct notification_client *client,
2974 struct notification_thread_state *state,
2975 enum lttng_notification_channel_status status)
2976 {
2977 int ret;
2978 struct lttng_notification_channel_command_reply reply = {
2979 .status = (int8_t) status,
2980 };
2981 struct lttng_notification_channel_message msg = {
2982 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
2983 .size = sizeof(reply),
2984 };
2985 char buffer[sizeof(msg) + sizeof(reply)];
2986 enum client_transmission_status transmission_status;
2987
2988 memcpy(buffer, &msg, sizeof(msg));
2989 memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
2990 DBG("[notification-thread] Send command reply (%i)", (int) status);
2991
2992 pthread_mutex_lock(&client->lock);
2993 if (client->communication.outbound.queued_command_reply) {
2994 /* Protocol error. */
2995 goto error_unlock;
2996 }
2997
2998 /* Enqueue buffer to outgoing queue and flush it. */
2999 ret = lttng_dynamic_buffer_append(
3000 &client->communication.outbound.payload.buffer,
3001 buffer, sizeof(buffer));
3002 if (ret) {
3003 goto error_unlock;
3004 }
3005
3006 transmission_status = client_flush_outgoing_queue(client);
3007
3008 if (client_has_outbound_data_left(client)) {
3009 /* Queue could not be emptied. */
3010 client->communication.outbound.queued_command_reply = true;
3011 }
3012
3013 pthread_mutex_unlock(&client->lock);
3014 ret = client_handle_transmission_status(
3015 client, transmission_status, state);
3016 if (ret) {
3017 goto error;
3018 }
3019
3020 return 0;
3021 error_unlock:
3022 pthread_mutex_unlock(&client->lock);
3023 error:
3024 return -1;
3025 }
3026
3027 static
3028 int client_handle_message_unknown(struct notification_client *client,
3029 struct notification_thread_state *state)
3030 {
3031 int ret;
3032 /*
3033 * Receiving message header. The function will be called again
3034 * once the rest of the message as been received and can be
3035 * interpreted.
3036 */
3037 const struct lttng_notification_channel_message *msg;
3038
3039 assert(sizeof(*msg) == client->communication.inbound.payload.buffer.size);
3040 msg = (const struct lttng_notification_channel_message *)