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 = session_uid,
1811                 .gid = 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_const(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_const(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         enum action_executor_status executor_status;
2121
2122         rcu_read_lock();
2123
2124         condition = lttng_trigger_get_condition(trigger);
2125         assert(condition);
2126
2127         ret = condition_is_supported(condition);
2128         if (ret < 0) {
2129                 goto error;
2130         } else if (ret == 0) {
2131                 *cmd_result = LTTNG_ERR_NOT_SUPPORTED;
2132                 goto error;
2133         } else {
2134                 /* Feature is supported, continue. */
2135                 ret = 0;
2136         }
2137
2138         trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
2139         if (!trigger_ht_element) {
2140                 ret = -1;
2141                 goto error;
2142         }
2143
2144         /* Add trigger to the trigger_ht. */
2145         cds_lfht_node_init(&trigger_ht_element->node);
2146         trigger_ht_element->trigger = trigger;
2147
2148         node = cds_lfht_add_unique(state->triggers_ht,
2149                         lttng_condition_hash(condition),
2150                         match_condition,
2151                         condition,
2152                         &trigger_ht_element->node);
2153         if (node != &trigger_ht_element->node) {
2154                 /* Not a fatal error, simply report it to the client. */
2155                 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2156                 goto error_free_ht_element;
2157         }
2158
2159         /*
2160          * Ownership of the trigger and of its wrapper was transfered to
2161          * the triggers_ht.
2162          */
2163         trigger_ht_element = NULL;
2164         free_trigger = false;
2165
2166         /*
2167          * The rest only applies to triggers that have a "notify" action.
2168          * It is not skipped as this is the only action type currently
2169          * supported.
2170          */
2171         if (is_trigger_action_notify(trigger)) {
2172                 client_list = notification_client_list_create(trigger);
2173                 if (!client_list) {
2174                         ret = -1;
2175                         goto error_free_ht_element;
2176                 }
2177
2178                 /* Build a list of clients to which this new trigger applies. */
2179                 cds_lfht_for_each_entry (state->client_socket_ht, &iter, client,
2180                                 client_socket_ht_node) {
2181                         if (!trigger_applies_to_client(trigger, client)) {
2182                                 continue;
2183                         }
2184
2185                         client_list_element =
2186                                         zmalloc(sizeof(*client_list_element));
2187                         if (!client_list_element) {
2188                                 ret = -1;
2189                                 goto error_put_client_list;
2190                         }
2191
2192                         CDS_INIT_LIST_HEAD(&client_list_element->node);
2193                         client_list_element->client = client;
2194                         cds_list_add(&client_list_element->node,
2195                                         &client_list->list);
2196                 }
2197
2198                 /*
2199                  * Client list ownership transferred to the
2200                  * notification_trigger_clients_ht.
2201                  */
2202                 publish_notification_client_list(state, client_list);
2203         }
2204
2205         switch (get_condition_binding_object(condition)) {
2206         case LTTNG_OBJECT_TYPE_SESSION:
2207                 /* Add the trigger to the list if it matches a known session. */
2208                 ret = bind_trigger_to_matching_session(trigger, state);
2209                 if (ret) {
2210                         goto error_put_client_list;
2211                 }
2212                 break;
2213         case LTTNG_OBJECT_TYPE_CHANNEL:
2214                 /*
2215                  * Add the trigger to list of triggers bound to the channels
2216                  * currently known.
2217                  */
2218                 ret = bind_trigger_to_matching_channels(trigger, state);
2219                 if (ret) {
2220                         goto error_put_client_list;
2221                 }
2222                 break;
2223         case LTTNG_OBJECT_TYPE_NONE:
2224                 break;
2225         default:
2226                 ERR("Unknown object type on which to bind a newly registered trigger was encountered");
2227                 ret = -1;
2228                 goto error_put_client_list;
2229         }
2230
2231         /*
2232          * The new trigger's condition must be evaluated against the current
2233          * state.
2234          *
2235          * In the case of `notify` action, nothing preventing clients from
2236          * subscribing to a condition before the corresponding trigger is
2237          * registered, we have to evaluate this new condition right away.
2238          *
2239          * At some point, we were waiting for the next "evaluation" (e.g. on
2240          * reception of a channel sample) to evaluate this new condition, but
2241          * that was broken.
2242          *
2243          * The reason it was broken is that waiting for the next sample
2244          * does not allow us to properly handle transitions for edge-triggered
2245          * conditions.
2246          *
2247          * Consider this example: when we handle a new channel sample, we
2248          * evaluate each conditions twice: once with the previous state, and
2249          * again with the newest state. We then use those two results to
2250          * determine whether a state change happened: a condition was false and
2251          * became true. If a state change happened, we have to notify clients.
2252          *
2253          * Now, if a client subscribes to a given notification and registers
2254          * a trigger *after* that subscription, we have to make sure the
2255          * condition is evaluated at this point while considering only the
2256          * current state. Otherwise, the next evaluation cycle may only see
2257          * that the evaluations remain the same (true for samples n-1 and n) and
2258          * the client will never know that the condition has been met.
2259          */
2260         switch (get_condition_binding_object(condition)) {
2261         case LTTNG_OBJECT_TYPE_SESSION:
2262                 ret = evaluate_session_condition_for_client(condition, state,
2263                                 &evaluation, &object_creds.uid,
2264                                 &object_creds.gid);
2265                 break;
2266         case LTTNG_OBJECT_TYPE_CHANNEL:
2267                 ret = evaluate_channel_condition_for_client(condition, state,
2268                                 &evaluation, &object_creds.uid,
2269                                 &object_creds.gid);
2270                 break;
2271         case LTTNG_OBJECT_TYPE_NONE:
2272                 ret = 0;
2273                 goto error_put_client_list;
2274         case LTTNG_OBJECT_TYPE_UNKNOWN:
2275         default:
2276                 ret = -1;
2277                 goto error_put_client_list;
2278         }
2279
2280         if (ret) {
2281                 /* Fatal error. */
2282                 goto error_put_client_list;
2283         }
2284
2285         DBG("Newly registered trigger's condition evaluated to %s",
2286                         evaluation ? "true" : "false");
2287         if (!evaluation) {
2288                 /* Evaluation yielded nothing. Normal exit. */
2289                 ret = 0;
2290                 goto error_put_client_list;
2291         }
2292
2293         /*
2294          * Ownership of `evaluation` transferred to the action executor
2295          * no matter the result.
2296          */
2297         executor_status = action_executor_enqueue(state->executor, trigger,
2298                         evaluation, &object_creds, client_list);
2299         evaluation = NULL;
2300         switch (executor_status) {
2301         case ACTION_EXECUTOR_STATUS_OK:
2302                 break;
2303         case ACTION_EXECUTOR_STATUS_ERROR:
2304         case ACTION_EXECUTOR_STATUS_INVALID:
2305                 /*
2306                  * TODO Add trigger identification (name/id) when
2307                  * it is added to the API.
2308                  */
2309                 ERR("Fatal error occurred while enqueuing action associated to newly registered trigger");
2310                 ret = -1;
2311                 goto error_put_client_list;
2312         case ACTION_EXECUTOR_STATUS_OVERFLOW:
2313                 /*
2314                  * TODO Add trigger identification (name/id) when
2315                  * it is added to the API.
2316                  *
2317                  * Not a fatal error.
2318                  */
2319                 WARN("No space left when enqueuing action associated to newly registered trigger");
2320                 ret = 0;
2321                 goto error_put_client_list;
2322         default:
2323                 abort();
2324         }
2325
2326         *cmd_result = LTTNG_OK;
2327
2328 error_put_client_list:
2329         notification_client_list_put(client_list);
2330
2331 error_free_ht_element:
2332         free(trigger_ht_element);
2333 error:
2334         if (free_trigger) {
2335                 lttng_trigger_destroy(trigger);
2336         }
2337         rcu_read_unlock();
2338         return ret;
2339 }
2340
2341 static
2342 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node)
2343 {
2344         free(caa_container_of(node, struct lttng_trigger_ht_element,
2345                         rcu_node));
2346 }
2347
2348 static
2349 int handle_notification_thread_command_unregister_trigger(
2350                 struct notification_thread_state *state,
2351                 struct lttng_trigger *trigger,
2352                 enum lttng_error_code *_cmd_reply)
2353 {
2354         struct cds_lfht_iter iter;
2355         struct cds_lfht_node *triggers_ht_node;
2356         struct lttng_channel_trigger_list *trigger_list;
2357         struct notification_client_list *client_list;
2358         struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2359         struct lttng_condition *condition = lttng_trigger_get_condition(
2360                         trigger);
2361         enum lttng_error_code cmd_reply;
2362
2363         rcu_read_lock();
2364
2365         cds_lfht_lookup(state->triggers_ht,
2366                         lttng_condition_hash(condition),
2367                         match_condition,
2368                         condition,
2369                         &iter);
2370         triggers_ht_node = cds_lfht_iter_get_node(&iter);
2371         if (!triggers_ht_node) {
2372                 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
2373                 goto end;
2374         } else {
2375                 cmd_reply = LTTNG_OK;
2376         }
2377
2378         /* Remove trigger from channel_triggers_ht. */
2379         cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
2380                         channel_triggers_ht_node) {
2381                 struct lttng_trigger_list_element *trigger_element, *tmp;
2382
2383                 cds_list_for_each_entry_safe(trigger_element, tmp,
2384                                 &trigger_list->list, node) {
2385                         const struct lttng_condition *current_condition =
2386                                         lttng_trigger_get_const_condition(
2387                                                 trigger_element->trigger);
2388
2389                         assert(current_condition);
2390                         if (!lttng_condition_is_equal(condition,
2391                                         current_condition)) {
2392                                 continue;
2393                         }
2394
2395                         DBG("[notification-thread] Removed trigger from channel_triggers_ht");
2396                         cds_list_del(&trigger_element->node);
2397                         /* A trigger can only appear once per channel */
2398                         break;
2399                 }
2400         }
2401
2402         /*
2403          * Remove and release the client list from
2404          * notification_trigger_clients_ht.
2405          */
2406         client_list = get_client_list_from_condition(state, condition);
2407         assert(client_list);
2408
2409         /* Put new reference and the hashtable's reference. */
2410         notification_client_list_put(client_list);
2411         notification_client_list_put(client_list);
2412         client_list = NULL;
2413
2414         /* Remove trigger from triggers_ht. */
2415         trigger_ht_element = caa_container_of(triggers_ht_node,
2416                         struct lttng_trigger_ht_element, node);
2417         cds_lfht_del(state->triggers_ht, triggers_ht_node);
2418
2419         /* Release the ownership of the trigger. */
2420         lttng_trigger_destroy(trigger_ht_element->trigger);
2421         call_rcu(&trigger_ht_element->rcu_node, free_lttng_trigger_ht_element_rcu);
2422 end:
2423         rcu_read_unlock();
2424         if (_cmd_reply) {
2425                 *_cmd_reply = cmd_reply;
2426         }
2427         return 0;
2428 }
2429
2430 /* Returns 0 on success, 1 on exit requested, negative value on error. */
2431 int handle_notification_thread_command(
2432                 struct notification_thread_handle *handle,
2433                 struct notification_thread_state *state)
2434 {
2435         int ret;
2436         uint64_t counter;
2437         struct notification_thread_command *cmd;
2438
2439         /* Read the event pipe to put it back into a quiescent state. */
2440         ret = lttng_read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
2441                         sizeof(counter));
2442         if (ret != sizeof(counter)) {
2443                 goto error;
2444         }
2445
2446         pthread_mutex_lock(&handle->cmd_queue.lock);
2447         cmd = cds_list_first_entry(&handle->cmd_queue.list,
2448                         struct notification_thread_command, cmd_list_node);
2449         cds_list_del(&cmd->cmd_list_node);
2450         pthread_mutex_unlock(&handle->cmd_queue.lock);
2451         switch (cmd->type) {
2452         case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
2453                 DBG("[notification-thread] Received register trigger command");
2454                 ret = handle_notification_thread_command_register_trigger(
2455                                 state, cmd->parameters.trigger,
2456                                 &cmd->reply_code);
2457                 break;
2458         case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
2459                 DBG("[notification-thread] Received unregister trigger command");
2460                 ret = handle_notification_thread_command_unregister_trigger(
2461                                 state, cmd->parameters.trigger,
2462                                 &cmd->reply_code);
2463                 break;
2464         case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
2465                 DBG("[notification-thread] Received add channel command");
2466                 ret = handle_notification_thread_command_add_channel(
2467                                 state,
2468                                 cmd->parameters.add_channel.session.name,
2469                                 cmd->parameters.add_channel.session.uid,
2470                                 cmd->parameters.add_channel.session.gid,
2471                                 cmd->parameters.add_channel.channel.name,
2472                                 cmd->parameters.add_channel.channel.domain,
2473                                 cmd->parameters.add_channel.channel.key,
2474                                 cmd->parameters.add_channel.channel.capacity,
2475                                 &cmd->reply_code);
2476                 break;
2477         case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
2478                 DBG("[notification-thread] Received remove channel command");
2479                 ret = handle_notification_thread_command_remove_channel(
2480                                 state, cmd->parameters.remove_channel.key,
2481                                 cmd->parameters.remove_channel.domain,
2482                                 &cmd->reply_code);
2483                 break;
2484         case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
2485         case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
2486                 DBG("[notification-thread] Received session rotation %s command",
2487                                 cmd->type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING ?
2488                                 "ongoing" : "completed");
2489                 ret = handle_notification_thread_command_session_rotation(
2490                                 state,
2491                                 cmd->type,
2492                                 cmd->parameters.session_rotation.session_name,
2493                                 cmd->parameters.session_rotation.uid,
2494                                 cmd->parameters.session_rotation.gid,
2495                                 cmd->parameters.session_rotation.trace_archive_chunk_id,
2496                                 cmd->parameters.session_rotation.location,
2497                                 &cmd->reply_code);
2498                 break;
2499         case NOTIFICATION_COMMAND_TYPE_QUIT:
2500                 DBG("[notification-thread] Received quit command");
2501                 cmd->reply_code = LTTNG_OK;
2502                 ret = 1;
2503                 goto end;
2504         case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE:
2505         {
2506                 const enum client_transmission_status client_status =
2507                                 cmd->parameters.client_communication_update
2508                                                 .status;
2509                 const notification_client_id client_id =
2510                                 cmd->parameters.client_communication_update.id;
2511                 struct notification_client *client;
2512
2513                 rcu_read_lock();
2514                 client = get_client_from_id(client_id, state);
2515
2516                 if (!client) {
2517                         /*
2518                          * Client error was probably already picked-up by the
2519                          * notification thread or it has disconnected
2520                          * gracefully while this command was queued.
2521                          */
2522                         DBG("Failed to find notification client to update communication status, client id = %" PRIu64,
2523                                         client_id);
2524                         ret = 0;
2525                 } else {
2526                         ret = client_handle_transmission_status(
2527                                         client, client_status, state);
2528                 }
2529                 rcu_read_unlock();
2530                 break;
2531         }
2532         default:
2533                 ERR("[notification-thread] Unknown internal command received");
2534                 goto error_unlock;
2535         }
2536
2537         if (ret) {
2538                 goto error_unlock;
2539         }
2540 end:
2541         if (cmd->is_async) {
2542                 free(cmd);
2543                 cmd = NULL;
2544         } else {
2545                 lttng_waiter_wake_up(&cmd->reply_waiter);
2546         }
2547         return ret;
2548 error_unlock:
2549         /* Wake-up and return a fatal error to the calling thread. */
2550         lttng_waiter_wake_up(&cmd->reply_waiter);
2551         cmd->reply_code = LTTNG_ERR_FATAL;
2552 error:
2553         /* Indicate a fatal error to the caller. */
2554         return -1;
2555 }
2556
2557 static
2558 int socket_set_non_blocking(int socket)
2559 {
2560         int ret, flags;
2561
2562         /* Set the pipe as non-blocking. */
2563         ret = fcntl(socket, F_GETFL, 0);
2564         if (ret == -1) {
2565                 PERROR("fcntl get socket flags");
2566                 goto end;
2567         }
2568         flags = ret;
2569
2570         ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
2571         if (ret == -1) {
2572                 PERROR("fcntl set O_NONBLOCK socket flag");
2573                 goto end;
2574         }
2575         DBG("Client socket (fd = %i) set as non-blocking", socket);
2576 end:
2577         return ret;
2578 }
2579
2580 static
2581 int client_reset_inbound_state(struct notification_client *client)
2582 {
2583         int ret;
2584
2585
2586         lttng_payload_clear(&client->communication.inbound.payload);
2587
2588         client->communication.inbound.bytes_to_receive =
2589                         sizeof(struct lttng_notification_channel_message);
2590         client->communication.inbound.msg_type =
2591                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
2592         LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
2593         LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
2594         ret = lttng_dynamic_buffer_set_size(
2595                         &client->communication.inbound.payload.buffer,
2596                         client->communication.inbound.bytes_to_receive);
2597
2598         return ret;
2599 }
2600
2601 int handle_notification_thread_client_connect(
2602                 struct notification_thread_state *state)
2603 {
2604         int ret;
2605         struct notification_client *client;
2606
2607         DBG("[notification-thread] Handling new notification channel client connection");
2608
2609         client = zmalloc(sizeof(*client));
2610         if (!client) {
2611                 /* Fatal error. */
2612                 ret = -1;
2613                 goto error;
2614         }
2615
2616         pthread_mutex_init(&client->lock, NULL);
2617         client->id = state->next_notification_client_id++;
2618         CDS_INIT_LIST_HEAD(&client->condition_list);
2619         lttng_payload_init(&client->communication.inbound.payload);
2620         lttng_payload_init(&client->communication.outbound.payload);
2621         client->communication.inbound.expect_creds = true;
2622
2623         ret = client_reset_inbound_state(client);
2624         if (ret) {
2625                 ERR("[notification-thread] Failed to reset client communication's inbound state");
2626                 ret = 0;
2627                 goto error;
2628         }
2629
2630         ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
2631         if (ret < 0) {
2632                 ERR("[notification-thread] Failed to accept new notification channel client connection");
2633                 ret = 0;
2634                 goto error;
2635         }
2636
2637         client->socket = ret;
2638
2639         ret = socket_set_non_blocking(client->socket);
2640         if (ret) {
2641                 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
2642                 goto error;
2643         }
2644
2645         ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
2646         if (ret < 0) {
2647                 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
2648                 ret = 0;
2649                 goto error;
2650         }
2651
2652         ret = lttng_poll_add(&state->events, client->socket,
2653                         LPOLLIN | LPOLLERR |
2654                         LPOLLHUP | LPOLLRDHUP);
2655         if (ret < 0) {
2656                 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
2657                 ret = 0;
2658                 goto error;
2659         }
2660         DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
2661                         client->socket);
2662
2663         rcu_read_lock();
2664         cds_lfht_add(state->client_socket_ht,
2665                         hash_client_socket(client->socket),
2666                         &client->client_socket_ht_node);
2667         cds_lfht_add(state->client_id_ht,
2668                         hash_client_id(client->id),
2669                         &client->client_id_ht_node);
2670         rcu_read_unlock();
2671
2672         return ret;
2673
2674 error:
2675         notification_client_destroy(client, state);
2676         return ret;
2677 }
2678
2679 /*
2680  * RCU read-lock must be held by the caller.
2681  * Client lock must _not_ be held by the caller.
2682  */
2683 static
2684 int notification_thread_client_disconnect(
2685                 struct notification_client *client,
2686                 struct notification_thread_state *state)
2687 {
2688         int ret;
2689         struct lttng_condition_list_element *condition_list_element, *tmp;
2690
2691         /* Acquire the client lock to disable its communication atomically. */
2692         pthread_mutex_lock(&client->lock);
2693         client->communication.active = false;
2694         cds_lfht_del(state->client_socket_ht, &client->client_socket_ht_node);
2695         cds_lfht_del(state->client_id_ht, &client->client_id_ht_node);
2696         pthread_mutex_unlock(&client->lock);
2697
2698         ret = lttng_poll_del(&state->events, client->socket);
2699         if (ret) {
2700                 ERR("[notification-thread] Failed to remove client socket %d from poll set",
2701                                 client->socket);
2702         }
2703
2704         /* Release all conditions to which the client was subscribed. */
2705         cds_list_for_each_entry_safe(condition_list_element, tmp,
2706                         &client->condition_list, node) {
2707                 (void) notification_thread_client_unsubscribe(client,
2708                                 condition_list_element->condition, state, NULL);
2709         }
2710
2711         /*
2712          * Client no longer accessible to other threads (through the
2713          * client lists).
2714          */
2715         notification_client_destroy(client, state);
2716         return ret;
2717 }
2718
2719 int handle_notification_thread_client_disconnect(
2720                 int client_socket, struct notification_thread_state *state)
2721 {
2722         int ret = 0;
2723         struct notification_client *client;
2724
2725         rcu_read_lock();
2726         DBG("[notification-thread] Closing client connection (socket fd = %i)",
2727                         client_socket);
2728         client = get_client_from_socket(client_socket, state);
2729         if (!client) {
2730                 /* Internal state corruption, fatal error. */
2731                 ERR("[notification-thread] Unable to find client (socket fd = %i)",
2732                                 client_socket);
2733                 ret = -1;
2734                 goto end;
2735         }
2736
2737         ret = notification_thread_client_disconnect(client, state);
2738 end:
2739         rcu_read_unlock();
2740         return ret;
2741 }
2742
2743 int handle_notification_thread_client_disconnect_all(
2744                 struct notification_thread_state *state)
2745 {
2746         struct cds_lfht_iter iter;
2747         struct notification_client *client;
2748         bool error_encoutered = false;
2749
2750         rcu_read_lock();
2751         DBG("[notification-thread] Closing all client connections");
2752         cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
2753                         client_socket_ht_node) {
2754                 int ret;
2755
2756                 ret = notification_thread_client_disconnect(
2757                                 client, state);
2758                 if (ret) {
2759                         error_encoutered = true;
2760                 }
2761         }
2762         rcu_read_unlock();
2763         return error_encoutered ? 1 : 0;
2764 }
2765
2766 int handle_notification_thread_trigger_unregister_all(
2767                 struct notification_thread_state *state)
2768 {
2769         bool error_occurred = false;
2770         struct cds_lfht_iter iter;
2771         struct lttng_trigger_ht_element *trigger_ht_element;
2772
2773         rcu_read_lock();
2774         cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
2775                         node) {
2776                 int ret = handle_notification_thread_command_unregister_trigger(
2777                                 state, trigger_ht_element->trigger, NULL);
2778                 if (ret) {
2779                         error_occurred = true;
2780                 }
2781         }
2782         rcu_read_unlock();
2783         return error_occurred ? -1 : 0;
2784 }
2785
2786 static
2787 int client_handle_transmission_status(
2788                 struct notification_client *client,
2789                 enum client_transmission_status transmission_status,
2790                 struct notification_thread_state *state)
2791 {
2792         int ret = 0;
2793
2794         switch (transmission_status) {
2795         case CLIENT_TRANSMISSION_STATUS_COMPLETE:
2796                 ret = lttng_poll_mod(&state->events, client->socket,
2797                                 CLIENT_POLL_MASK_IN);
2798                 if (ret) {
2799                         goto end;
2800                 }
2801
2802                 break;
2803         case CLIENT_TRANSMISSION_STATUS_QUEUED:
2804                 /*
2805                  * We want to be notified whenever there is buffer space
2806                  * available to send the rest of the payload.
2807                  */
2808                 ret = lttng_poll_mod(&state->events, client->socket,
2809                                 CLIENT_POLL_MASK_IN_OUT);
2810                 if (ret) {
2811                         goto end;
2812                 }
2813                 break;
2814         case CLIENT_TRANSMISSION_STATUS_FAIL:
2815                 ret = notification_thread_client_disconnect(client, state);
2816                 if (ret) {
2817                         goto end;
2818                 }
2819                 break;
2820         case CLIENT_TRANSMISSION_STATUS_ERROR:
2821                 ret = -1;
2822                 goto end;
2823         default:
2824                 abort();
2825         }
2826 end:
2827         return ret;
2828 }
2829
2830 /* Client lock must be acquired by caller. */
2831 static
2832 enum client_transmission_status client_flush_outgoing_queue(
2833                 struct notification_client *client)
2834 {
2835         ssize_t ret;
2836         size_t to_send_count;
2837         enum client_transmission_status status;
2838         struct lttng_payload_view pv = lttng_payload_view_from_payload(
2839                         &client->communication.outbound.payload, 0, -1);
2840         const int fds_to_send_count =
2841                         lttng_payload_view_get_fd_handle_count(&pv);
2842
2843         ASSERT_LOCKED(client->lock);
2844
2845         if (!client->communication.active) {
2846                 status = CLIENT_TRANSMISSION_STATUS_FAIL;
2847                 goto end;
2848         }
2849
2850         if (pv.buffer.size == 0) {
2851                 /*
2852                  * If both data and fds are equal to zero, we are in an invalid
2853                  * state.
2854                  */
2855                 assert(fds_to_send_count != 0);
2856                 goto send_fds;
2857         }
2858
2859         /* Send data. */
2860         to_send_count = pv.buffer.size;
2861         DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
2862                         client->socket);
2863
2864         ret = lttcomm_send_unix_sock_non_block(client->socket,
2865                         pv.buffer.data,
2866                         to_send_count);
2867         if ((ret >= 0 && ret < to_send_count)) {
2868                 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
2869                                 client->socket);
2870                 to_send_count -= max(ret, 0);
2871
2872                 memcpy(client->communication.outbound.payload.buffer.data,
2873                                 pv.buffer.data +
2874                                 pv.buffer.size - to_send_count,
2875                                 to_send_count);
2876                 ret = lttng_dynamic_buffer_set_size(
2877                                 &client->communication.outbound.payload.buffer,
2878                                 to_send_count);
2879                 if (ret) {
2880                         goto error;
2881                 }
2882
2883                 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
2884                 goto end;
2885         } else if (ret < 0) {
2886                 /* Generic error, disable the client's communication. */
2887                 ERR("[notification-thread] Failed to flush outgoing queue, disconnecting client (socket fd = %i)",
2888                                 client->socket);
2889                 client->communication.active = false;
2890                 status = CLIENT_TRANSMISSION_STATUS_FAIL;
2891                 goto end;
2892         } else {
2893                 /*
2894                  * No error and flushed the queue completely.
2895                  *
2896                  * The payload buffer size is used later to
2897                  * check if there is notifications queued. So albeit that the
2898                  * direct caller knows that the transmission is complete, we
2899                  * need to set the buffer size to zero.
2900                  */
2901                 ret = lttng_dynamic_buffer_set_size(
2902                                 &client->communication.outbound.payload.buffer, 0);
2903                 if (ret) {
2904                         goto error;
2905                 }
2906         }
2907
2908 send_fds:
2909         /* No fds to send, transmission is complete. */
2910         if (fds_to_send_count == 0) {
2911                 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
2912                 goto end;
2913         }
2914
2915         ret = lttcomm_send_payload_view_fds_unix_sock_non_block(
2916                         client->socket, &pv);
2917         if (ret < 0) {
2918                 /* Generic error, disable the client's communication. */
2919                 ERR("[notification-thread] Failed to flush outgoing fds queue, disconnecting client (socket fd = %i)",
2920                                 client->socket);
2921                 client->communication.active = false;
2922                 status = CLIENT_TRANSMISSION_STATUS_FAIL;
2923                 goto end;
2924         } else if (ret == 0) {
2925                 /* Nothing could be sent. */
2926                 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
2927         } else {
2928                 /* Fd passing is an all or nothing kind of thing. */
2929                 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
2930                 /*
2931                  * The payload _fd_array count is used later to
2932                  * check if there is notifications queued. So although the
2933                  * direct caller knows that the transmission is complete, we
2934                  * need to clear the _fd_array for the queuing check.
2935                  */
2936                 lttng_dynamic_pointer_array_clear(
2937                                 &client->communication.outbound.payload
2938                                                  ._fd_handles);
2939         }
2940
2941 end:
2942         if (status == CLIENT_TRANSMISSION_STATUS_COMPLETE) {
2943                 client->communication.outbound.queued_command_reply = false;
2944                 client->communication.outbound.dropped_notification = false;
2945                 lttng_payload_clear(&client->communication.outbound.payload);
2946         }
2947
2948         return status;
2949 error:
2950         return CLIENT_TRANSMISSION_STATUS_ERROR;
2951 }
2952
2953 static
2954 bool client_has_outbound_data_left(
2955                 const struct notification_client *client)
2956 {
2957         const struct lttng_payload_view pv = lttng_payload_view_from_payload(
2958                         &client->communication.outbound.payload, 0, -1);
2959         const bool has_data = pv.buffer.size != 0;
2960         const bool has_fds = lttng_payload_view_get_fd_handle_count(&pv);
2961
2962         return has_data || has_fds;
2963 }
2964
2965 /* Client lock must _not_ be held by the caller. */
2966 static
2967 int client_send_command_reply(struct notification_client *client,
2968                 struct notification_thread_state *state,
2969                 enum lttng_notification_channel_status status)
2970 {
2971         int ret;
2972         struct lttng_notification_channel_command_reply reply = {
2973                 .status = (int8_t) status,
2974         };
2975         struct lttng_notification_channel_message msg = {
2976                 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
2977                 .size = sizeof(reply),
2978         };
2979         char buffer[sizeof(msg) + sizeof(reply)];
2980         enum client_transmission_status transmission_status;
2981
2982         memcpy(buffer, &msg, sizeof(msg));
2983         memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
2984         DBG("[notification-thread] Send command reply (%i)", (int) status);
2985
2986         pthread_mutex_lock(&client->lock);
2987         if (client->communication.outbound.queued_command_reply) {
2988                 /* Protocol error. */
2989                 goto error_unlock;
2990         }
2991
2992         /* Enqueue buffer to outgoing queue and flush it. */
2993         ret = lttng_dynamic_buffer_append(
2994                         &client->communication.outbound.payload.buffer,
2995                         buffer, sizeof(buffer));
2996         if (ret) {
2997                 goto error_unlock;
2998         }
2999
3000         transmission_status = client_flush_outgoing_queue(client);
3001
3002         if (client_has_outbound_data_left(client)) {
3003                 /* Queue could not be emptied. */
3004                 client->communication.outbound.queued_command_reply = true;
3005         }
3006
3007         pthread_mutex_unlock(&client->lock);
3008         ret = client_handle_transmission_status(
3009                         client, transmission_status, state);
3010         if (ret) {
3011                 goto error;
3012         }
3013
3014         return 0;
3015 error_unlock:
3016         pthread_mutex_unlock(&client->lock);
3017 error:
3018         return -1;
3019 }
3020
3021 static
3022 int client_handle_message_unknown(struct notification_client *client,
3023                 struct notification_thread_state *state)
3024 {
3025         int ret;
3026         /*
3027          * Receiving message header. The function will be called again
3028          * once the rest of the message as been received and can be
3029          * interpreted.
3030          */
3031         const struct lttng_notification_channel_message *msg;
3032
3033         assert(sizeof(*msg) == client->communication.inbound.payload.buffer.size);
3034         msg = (const struct lttng_notification_channel_message *)
3035                               client->communication.inbound.payload.buffer.data;
3036
3037         if (msg->size == 0 ||
3038                         msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
3039                 ERR("[notification-thread] Invalid notification channel message: length = %u",