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