src/lib/lttng-ctl/channel.c

   1 /*
   2  * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  *
   4  * SPDX-License-Identifier: LGPL-2.1-only
   5  *
   6  */
   7
   8 #include <lttng/notification/notification-internal.h>
   9 #include <lttng/notification/channel-internal.h>
  10 #include <lttng/condition/condition-internal.h>
  11 #include <lttng/endpoint.h>
  12 #include <common/defaults.h>
  13 #include <common/error.h>
  14 #include <common/dynamic-buffer.h>
  15 #include <common/utils.h>
  16 #include <common/defaults.h>
  17 #include <common/payload.h>
  18 #include <common/payload-view.h>
  19 #include <common/unix.h>
  20 #include <assert.h>
  21 #include "lttng-ctl-helper.h"
  22 #include <common/compat/poll.h>
  23
  24 static
  25 int handshake(struct lttng_notification_channel *channel);
  26
  27 /*
  28  * Populates the reception buffer with the next complete message.
  29  * The caller must acquire the channel's lock.
  30  */
  31 static
  32 int receive_message(struct lttng_notification_channel *channel)
  33 {
  34         ssize_t ret;
  35         struct lttng_notification_channel_message msg;
  36
  37         lttng_payload_clear(&channel->reception_payload);
  38
  39         ret = lttcomm_recv_unix_sock(channel->socket, &msg, sizeof(msg));
  40         if (ret <= 0) {
  41                 ret = -1;
  42                 goto error;
  43         }
  44
  45         if (msg.size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
  46                 ret = -1;
  47                 goto error;
  48         }
  49
  50         /* Add message header at buffer's start. */
  51         ret = lttng_dynamic_buffer_append(&channel->reception_payload.buffer, &msg,
  52                         sizeof(msg));
  53         if (ret) {
  54                 goto error;
  55         }
  56
  57         /* Reserve space for the payload. */
  58         ret = lttng_dynamic_buffer_set_size(&channel->reception_payload.buffer,
  59                         channel->reception_payload.buffer.size + msg.size);
  60         if (ret) {
  61                 goto error;
  62         }
  63
  64         /* Receive message payload. */
  65         ret = lttcomm_recv_unix_sock(channel->socket,
  66                         channel->reception_payload.buffer.data + sizeof(msg), msg.size);
  67         if (ret < (ssize_t) msg.size) {
  68                 ret = -1;
  69                 goto error;
  70         }
  71
  72         /* Receive message fds. */
  73         if (msg.fds != 0) {
  74                 ret = lttcomm_recv_payload_fds_unix_sock(channel->socket,
  75                                 msg.fds, &channel->reception_payload);
  76                 if (ret < sizeof(int) * msg.fds) {
  77                         ret = -1;
  78                         goto error;
  79                 }
  80         }
  81         ret = 0;
  82 end:
  83         return ret;
  84 error:
  85         lttng_payload_clear(&channel->reception_payload);
  86         goto end;
  87 }
  88
  89 static
  90 enum lttng_notification_channel_message_type get_current_message_type(
  91                 struct lttng_notification_channel *channel)
  92 {
  93         struct lttng_notification_channel_message *msg;
  94
  95         assert(channel->reception_payload.buffer.size >= sizeof(*msg));
  96
  97         msg = (struct lttng_notification_channel_message *)
  98                         channel->reception_payload.buffer.data;
  99         return (enum lttng_notification_channel_message_type) msg->type;
 100 }
 101
 102 static
 103 struct lttng_notification *create_notification_from_current_message(
 104                 struct lttng_notification_channel *channel)
 105 {
 106         ssize_t ret;
 107         struct lttng_notification *notification = NULL;
 108
 109         if (channel->reception_payload.buffer.size <=
 110                         sizeof(struct lttng_notification_channel_message)) {
 111                 goto end;
 112         }
 113
 114         {
 115                 struct lttng_payload_view view = lttng_payload_view_from_payload(
 116                                 &channel->reception_payload,
 117                                 sizeof(struct lttng_notification_channel_message),
 118                                 -1);
 119
 120                 ret = lttng_notification_create_from_payload(
 121                                 &view, &notification);
 122         }
 123
 124         if (ret != channel->reception_payload.buffer.size -
 125                         sizeof(struct lttng_notification_channel_message)) {
 126                 lttng_notification_destroy(notification);
 127                 notification = NULL;
 128                 goto end;
 129         }
 130 end:
 131         return notification;
 132 }
 133
 134 struct lttng_notification_channel *lttng_notification_channel_create(
 135                 struct lttng_endpoint *endpoint)
 136 {
 137         int fd, ret;
 138         bool is_in_tracing_group = false, is_root = false;
 139         char *sock_path = NULL;
 140         struct lttng_notification_channel *channel = NULL;
 141
 142         if (!endpoint ||
 143                         endpoint != lttng_session_daemon_notification_endpoint) {
 144                 goto end;
 145         }
 146
 147         sock_path = zmalloc(LTTNG_PATH_MAX);
 148         if (!sock_path) {
 149                 goto end;
 150         }
 151
 152         channel = zmalloc(sizeof(struct lttng_notification_channel));
 153         if (!channel) {
 154                 goto end;
 155         }
 156         channel->socket = -1;
 157         pthread_mutex_init(&channel->lock, NULL);
 158         lttng_payload_init(&channel->reception_payload);
 159         CDS_INIT_LIST_HEAD(&channel->pending_notifications.list);
 160
 161         is_root = (getuid() == 0);
 162         if (!is_root) {
 163                 is_in_tracing_group = lttng_check_tracing_group();
 164         }
 165
 166         if (is_root || is_in_tracing_group) {
 167                 lttng_ctl_copy_string(sock_path,
 168                                 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK,
 169                                 LTTNG_PATH_MAX);
 170                 ret = lttcomm_connect_unix_sock(sock_path);
 171                 if (ret >= 0) {
 172                         fd = ret;
 173                         goto set_fd;
 174                 }
 175         }
 176
 177         /* Fallback to local session daemon. */
 178         ret = snprintf(sock_path, LTTNG_PATH_MAX,
 179                         DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
 180                         utils_get_home_dir());
 181         if (ret < 0 || ret >= LTTNG_PATH_MAX) {
 182                 goto error;
 183         }
 184
 185         ret = lttcomm_connect_unix_sock(sock_path);
 186         if (ret < 0) {
 187                 goto error;
 188         }
 189         fd = ret;
 190
 191 set_fd:
 192         channel->socket = fd;
 193
 194         ret = handshake(channel);
 195         if (ret) {
 196                 goto error;
 197         }
 198 end:
 199         free(sock_path);
 200         return channel;
 201 error:
 202         lttng_notification_channel_destroy(channel);
 203         channel = NULL;
 204         goto end;
 205 }
 206
 207 enum lttng_notification_channel_status
 208 lttng_notification_channel_get_next_notification(
 209                 struct lttng_notification_channel *channel,
 210                 struct lttng_notification **_notification)
 211 {
 212         int ret;
 213         struct lttng_notification *notification = NULL;
 214         enum lttng_notification_channel_status status =
 215                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 216         struct lttng_poll_event events;
 217
 218         if (!channel || !_notification) {
 219                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 220                 goto end;
 221         }
 222
 223         pthread_mutex_lock(&channel->lock);
 224
 225         if (channel->pending_notifications.count) {
 226                 struct pending_notification *pending_notification;
 227
 228                 assert(!cds_list_empty(&channel->pending_notifications.list));
 229
 230                 /* Deliver one of the pending notifications. */
 231                 pending_notification = cds_list_first_entry(
 232                                 &channel->pending_notifications.list,
 233                                 struct pending_notification,
 234                                 node);
 235                 notification = pending_notification->notification;
 236                 if (!notification) {
 237                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
 238                 }
 239                 cds_list_del(&pending_notification->node);
 240                 channel->pending_notifications.count--;
 241                 free(pending_notification);
 242                 goto end_unlock;
 243         }
 244
 245         /*
 246          * Block on interruptible epoll/poll() instead of the message reception
 247          * itself as the recvmsg() wrappers always restart on EINTR. We choose
 248          * to wait using interruptible epoll/poll() in order to:
 249          *   1) Return if a signal occurs,
 250          *   2) Not deal with partially received messages.
 251          *
 252          * The drawback to this approach is that we assume that messages
 253          * are complete/well formed. If a message is shorter than its
 254          * announced length, receive_message() will block on recvmsg()
 255          * and never return (even if a signal is received).
 256          */
 257         ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
 258         if (ret < 0) {
 259                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 260                 goto end_unlock;
 261         }
 262         ret = lttng_poll_add(&events, channel->socket, LPOLLIN | LPOLLERR);
 263         if (ret < 0) {
 264                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 265                 goto end_clean_poll;
 266         }
 267         ret = lttng_poll_wait_interruptible(&events, -1);
 268         if (ret <= 0) {
 269                 status = (ret == -1 && errno == EINTR) ?
 270                         LTTNG_NOTIFICATION_CHANNEL_STATUS_INTERRUPTED :
 271                         LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 272                 goto end_clean_poll;
 273         }
 274
 275         ret = receive_message(channel);
 276         if (ret) {
 277                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 278                 goto end_clean_poll;
 279         }
 280
 281         switch (get_current_message_type(channel)) {
 282         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 283                 notification = create_notification_from_current_message(
 284                                 channel);
 285                 if (!notification) {
 286                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 287                         goto end_clean_poll;
 288                 }
 289                 break;
 290         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 291                 /* No payload to consume. */
 292                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
 293                 break;
 294         default:
 295                 /* Protocol error. */
 296                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 297                 goto end_clean_poll;
 298         }
 299
 300 end_clean_poll:
 301         lttng_poll_clean(&events);
 302 end_unlock:
 303         pthread_mutex_unlock(&channel->lock);
 304         *_notification = notification;
 305 end:
 306         return status;
 307 }
 308
 309 static
 310 int enqueue_dropped_notification(
 311                 struct lttng_notification_channel *channel)
 312 {
 313         int ret = 0;
 314         struct pending_notification *pending_notification;
 315         struct cds_list_head *last_element =
 316                         channel->pending_notifications.list.prev;
 317
 318         pending_notification = caa_container_of(last_element,
 319                         struct pending_notification, node);
 320         if (!pending_notification->notification) {
 321                 /*
 322                  * The last enqueued notification indicates dropped
 323                  * notifications; there is nothing to do as we group
 324                  * dropped notifications together.
 325                  */
 326                 goto end;
 327         }
 328
 329         if (channel->pending_notifications.count >=
 330                         DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT &&
 331                         pending_notification->notification) {
 332                 /*
 333                  * Discard the last enqueued notification to indicate
 334                  * that notifications were dropped at this point.
 335                  */
 336                 lttng_notification_destroy(
 337                                 pending_notification->notification);
 338                 pending_notification->notification = NULL;
 339                 goto end;
 340         }
 341
 342         pending_notification = zmalloc(sizeof(*pending_notification));
 343         if (!pending_notification) {
 344                 ret = -1;
 345                 goto end;
 346         }
 347         CDS_INIT_LIST_HEAD(&pending_notification->node);
 348         cds_list_add(&pending_notification->node,
 349                         &channel->pending_notifications.list);
 350         channel->pending_notifications.count++;
 351 end:
 352         return ret;
 353 }
 354
 355 static
 356 int enqueue_notification_from_current_message(
 357                 struct lttng_notification_channel *channel)
 358 {
 359         int ret = 0;
 360         struct lttng_notification *notification;
 361         struct pending_notification *pending_notification;
 362
 363         if (channel->pending_notifications.count >=
 364                         DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT) {
 365                 /* Drop the notification. */
 366                 ret = enqueue_dropped_notification(channel);
 367                 goto end;
 368         }
 369
 370         pending_notification = zmalloc(sizeof(*pending_notification));
 371         if (!pending_notification) {
 372                 ret = -1;
 373                 goto error;
 374         }
 375         CDS_INIT_LIST_HEAD(&pending_notification->node);
 376
 377         notification = create_notification_from_current_message(channel);
 378         if (!notification) {
 379                 ret = -1;
 380                 goto error;
 381         }
 382
 383         pending_notification->notification = notification;
 384         cds_list_add(&pending_notification->node,
 385                         &channel->pending_notifications.list);
 386         channel->pending_notifications.count++;
 387 end:
 388         return ret;
 389 error:
 390         free(pending_notification);
 391         goto end;
 392 }
 393
 394 enum lttng_notification_channel_status
 395 lttng_notification_channel_has_pending_notification(
 396                 struct lttng_notification_channel *channel,
 397                 bool *_notification_pending)
 398 {
 399         int ret;
 400         enum lttng_notification_channel_status status =
 401                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 402         struct lttng_poll_event events;
 403
 404         if (!channel || !_notification_pending) {
 405                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 406                 goto end;
 407         }
 408
 409         pthread_mutex_lock(&channel->lock);
 410
 411         if (channel->pending_notifications.count) {
 412                 *_notification_pending = true;
 413                 goto end_unlock;
 414         }
 415
 416         if (channel->socket < 0) {
 417                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED;
 418                 goto end_unlock;
 419         }
 420
 421         /*
 422          * Check, without blocking, if data is available on the channel's
 423          * socket. If there is data available, it is safe to read (blocking)
 424          * on the socket for a message from the session daemon.
 425          *
 426          * Since all commands wait for the session daemon's reply before
 427          * releasing the channel's lock, the protocol only allows for
 428          * notifications and "notification dropped" messages to come
 429          * through. If we receive a different message type, it is
 430          * considered a protocol error.
 431          *
 432          * Note that this function is not guaranteed not to block. This
 433          * will block until our peer (the session daemon) has sent a complete
 434          * message if we see data available on the socket. If the peer does
 435          * not respect the protocol, this may block indefinitely.
 436          */
 437         ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
 438         if (ret < 0) {
 439                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 440                 goto end_unlock;
 441         }
 442         ret = lttng_poll_add(&events, channel->socket, LPOLLIN | LPOLLERR);
 443         if (ret < 0) {
 444                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 445                 goto end_clean_poll;
 446         }
 447         /* timeout = 0: return immediately. */
 448         ret = lttng_poll_wait_interruptible(&events, 0);
 449         if (ret == 0) {
 450                 /* No data available. */
 451                 *_notification_pending = false;
 452                 goto end_clean_poll;
 453         } else if (ret < 0) {
 454                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 455                 goto end_clean_poll;
 456         }
 457
 458         /* Data available on socket. */
 459         ret = receive_message(channel);
 460         if (ret) {
 461                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 462                 goto end_clean_poll;
 463         }
 464
 465         switch (get_current_message_type(channel)) {
 466         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 467                 ret = enqueue_notification_from_current_message(channel);
 468                 if (ret) {
 469                         goto end_clean_poll;
 470                 }
 471                 *_notification_pending = true;
 472                 break;
 473         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 474                 ret = enqueue_dropped_notification(channel);
 475                 if (ret) {
 476                         goto end_clean_poll;
 477                 }
 478                 *_notification_pending = true;
 479                 break;
 480         default:
 481                 /* Protocol error. */
 482                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 483                 goto end_clean_poll;
 484         }
 485
 486 end_clean_poll:
 487         lttng_poll_clean(&events);
 488 end_unlock:
 489         pthread_mutex_unlock(&channel->lock);
 490 end:
 491         return status;
 492 }
 493
 494 static
 495 int receive_command_reply(struct lttng_notification_channel *channel,
 496                 enum lttng_notification_channel_status *status)
 497 {
 498         int ret;
 499         struct lttng_notification_channel_command_reply *reply;
 500
 501         while (true) {
 502                 enum lttng_notification_channel_message_type msg_type;
 503
 504                 ret = receive_message(channel);
 505                 if (ret) {
 506                         goto end;
 507                 }
 508
 509                 msg_type = get_current_message_type(channel);
 510                 switch (msg_type) {
 511                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY:
 512                         goto exit_loop;
 513                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 514                         ret = enqueue_notification_from_current_message(
 515                                         channel);
 516                         if (ret) {
 517                                 goto end;
 518                         }
 519                         break;
 520                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 521                         ret = enqueue_dropped_notification(channel);
 522                         if (ret) {
 523                                 goto end;
 524                         }
 525                         break;
 526                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
 527                 {
 528                         struct lttng_notification_channel_command_handshake *handshake;
 529
 530                         handshake = (struct lttng_notification_channel_command_handshake *)
 531                                         (channel->reception_payload.buffer.data +
 532                                         sizeof(struct lttng_notification_channel_message));
 533                         channel->version.major = handshake->major;
 534                         channel->version.minor = handshake->minor;
 535                         channel->version.set = true;
 536                         break;
 537                 }
 538                 default:
 539                         ret = -1;
 540                         goto end;
 541                 }
 542         }
 543
 544 exit_loop:
 545         if (channel->reception_payload.buffer.size <
 546                         (sizeof(struct lttng_notification_channel_message) +
 547                         sizeof(*reply))) {
 548                 /* Invalid message received. */
 549                 ret = -1;
 550                 goto end;
 551         }
 552
 553         reply = (struct lttng_notification_channel_command_reply *)
 554                         (channel->reception_payload.buffer.data +
 555                         sizeof(struct lttng_notification_channel_message));
 556         *status = (enum lttng_notification_channel_status) reply->status;
 557 end:
 558         return ret;
 559 }
 560
 561 static
 562 int handshake(struct lttng_notification_channel *channel)
 563 {
 564         ssize_t ret;
 565         enum lttng_notification_channel_status status =
 566                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 567         struct lttng_notification_channel_command_handshake handshake = {
 568                 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
 569                 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
 570         };
 571         struct lttng_notification_channel_message msg_header = {
 572                 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
 573                 .size = sizeof(handshake),
 574         };
 575         char send_buffer[sizeof(msg_header) + sizeof(handshake)];
 576
 577         memcpy(send_buffer, &msg_header, sizeof(msg_header));
 578         memcpy(send_buffer + sizeof(msg_header), &handshake, sizeof(handshake));
 579
 580         pthread_mutex_lock(&channel->lock);
 581
 582         ret = lttcomm_send_creds_unix_sock(channel->socket, send_buffer,
 583                         sizeof(send_buffer));
 584         if (ret < 0) {
 585                 goto end_unlock;
 586         }
 587
 588         /* Receive handshake info from the sessiond. */
 589         ret = receive_command_reply(channel, &status);
 590         if (ret < 0) {
 591                 goto end_unlock;
 592         }
 593
 594         if (!channel->version.set) {
 595                 ret = -1;
 596                 goto end_unlock;
 597         }
 598
 599         if (channel->version.major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
 600                 ret = -1;
 601                 goto end_unlock;
 602         }
 603
 604 end_unlock:
 605         pthread_mutex_unlock(&channel->lock);
 606         return ret;
 607 }
 608
 609 static
 610 enum lttng_notification_channel_status send_condition_command(
 611                 struct lttng_notification_channel *channel,
 612                 enum lttng_notification_channel_message_type type,
 613                 const struct lttng_condition *condition)
 614 {
 615         int socket;
 616         ssize_t ret;
 617         enum lttng_notification_channel_status status =
 618                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 619         struct lttng_payload payload;
 620         struct lttng_notification_channel_message cmd_header = {
 621                 .type = (int8_t) type,
 622         };
 623
 624         lttng_payload_init(&payload);
 625
 626         if (!channel) {
 627                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 628                 goto end;
 629         }
 630
 631         assert(type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE ||
 632                 type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE);
 633
 634         pthread_mutex_lock(&channel->lock);
 635         socket = channel->socket;
 636
 637         if (!lttng_condition_validate(condition)) {
 638                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 639                 goto end_unlock;
 640         }
 641
 642         ret = lttng_dynamic_buffer_append(&payload.buffer, &cmd_header,
 643                         sizeof(cmd_header));
 644         if (ret) {
 645                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 646                 goto end_unlock;
 647         }
 648
 649         ret = lttng_condition_serialize(condition, &payload);
 650         if (ret) {
 651                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 652                 goto end_unlock;
 653         }
 654
 655         /* Update payload length. */
 656         ((struct lttng_notification_channel_message *) payload.buffer.data)->size =
 657                         (uint32_t) (payload.buffer.size - sizeof(cmd_header));
 658
 659         {
 660                 struct lttng_payload_view pv =
 661                                 lttng_payload_view_from_payload(
 662                                                 &payload, 0, -1);
 663                 const int fd_count =
 664                                 lttng_payload_view_get_fd_handle_count(&pv);
 665
 666                 /* Update fd count. */
 667                 ((struct lttng_notification_channel_message *) payload.buffer.data)->fds =
 668                         (uint32_t) fd_count;
 669
 670                 ret = lttcomm_send_unix_sock(
 671                         socket, pv.buffer.data, pv.buffer.size);
 672                 if (ret < 0) {
 673                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 674                         goto end_unlock;
 675                 }
 676
 677                 /* Pass fds if present. */
 678                 if (fd_count > 0) {
 679                         ret = lttcomm_send_payload_view_fds_unix_sock(socket,
 680                                         &pv);
 681                         if (ret < 0) {
 682                                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 683                                 goto end_unlock;
 684                         }
 685                 }
 686         }
 687
 688         ret = receive_command_reply(channel, &status);
 689         if (ret < 0) {
 690                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 691                 goto end_unlock;
 692         }
 693 end_unlock:
 694         pthread_mutex_unlock(&channel->lock);
 695 end:
 696         lttng_payload_reset(&payload);
 697         return status;
 698 }
 699
 700 enum lttng_notification_channel_status lttng_notification_channel_subscribe(
 701                 struct lttng_notification_channel *channel,
 702                 const struct lttng_condition *condition)
 703 {
 704         return send_condition_command(channel,
 705                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE,
 706                         condition);
 707 }
 708
 709 enum lttng_notification_channel_status lttng_notification_channel_unsubscribe(
 710                 struct lttng_notification_channel *channel,
 711                 const struct lttng_condition *condition)
 712 {
 713         return send_condition_command(channel,
 714                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE,
 715                         condition);
 716 }
 717
 718 void lttng_notification_channel_destroy(
 719                 struct lttng_notification_channel *channel)
 720 {
 721         if (!channel) {
 722                 return;
 723         }
 724
 725         if (channel->socket >= 0) {
 726                 (void) lttcomm_close_unix_sock(channel->socket);
 727         }
 728         pthread_mutex_destroy(&channel->lock);
 729         lttng_payload_reset(&channel->reception_payload);
 730         free(channel);
 731 }