src/bin/lttng-sessiond/notification-thread-events.c

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