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                 ret = lttng_strncpy(sock_path,
 168                                 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK,
 169                                 LTTNG_PATH_MAX);
 170                 if (ret) {
 171                         ret = -LTTNG_ERR_INVALID;
 172                         goto error;
 173                 }
 174
 175                 ret = lttcomm_connect_unix_sock(sock_path);
 176                 if (ret >= 0) {
 177                         fd = ret;
 178                         goto set_fd;
 179                 }
 180         }
 181
 182         /* Fallback to local session daemon. */
 183         ret = snprintf(sock_path, LTTNG_PATH_MAX,
 184                         DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
 185                         utils_get_home_dir());
 186         if (ret < 0 || ret >= LTTNG_PATH_MAX) {
 187                 goto error;
 188         }
 189
 190         ret = lttcomm_connect_unix_sock(sock_path);
 191         if (ret < 0) {
 192                 goto error;
 193         }
 194         fd = ret;
 195
 196 set_fd:
 197         channel->socket = fd;
 198
 199         ret = handshake(channel);
 200         if (ret) {
 201                 goto error;
 202         }
 203 end:
 204         free(sock_path);
 205         return channel;
 206 error:
 207         lttng_notification_channel_destroy(channel);
 208         channel = NULL;
 209         goto end;
 210 }
 211
 212 enum lttng_notification_channel_status
 213 lttng_notification_channel_get_next_notification(
 214                 struct lttng_notification_channel *channel,
 215                 struct lttng_notification **_notification)
 216 {
 217         int ret;
 218         struct lttng_notification *notification = NULL;
 219         enum lttng_notification_channel_status status =
 220                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 221         struct lttng_poll_event events;
 222
 223         if (!channel || !_notification) {
 224                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 225                 goto end;
 226         }
 227
 228         pthread_mutex_lock(&channel->lock);
 229
 230         if (channel->pending_notifications.count) {
 231                 struct pending_notification *pending_notification;
 232
 233                 assert(!cds_list_empty(&channel->pending_notifications.list));
 234
 235                 /* Deliver one of the pending notifications. */
 236                 pending_notification = cds_list_first_entry(
 237                                 &channel->pending_notifications.list,
 238                                 struct pending_notification,
 239                                 node);
 240                 notification = pending_notification->notification;
 241                 if (!notification) {
 242                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
 243                 }
 244                 cds_list_del(&pending_notification->node);
 245                 channel->pending_notifications.count--;
 246                 free(pending_notification);
 247                 goto end_unlock;
 248         }
 249
 250         /*
 251          * Block on interruptible epoll/poll() instead of the message reception
 252          * itself as the recvmsg() wrappers always restart on EINTR. We choose
 253          * to wait using interruptible epoll/poll() in order to:
 254          *   1) Return if a signal occurs,
 255          *   2) Not deal with partially received messages.
 256          *
 257          * The drawback to this approach is that we assume that messages
 258          * are complete/well formed. If a message is shorter than its
 259          * announced length, receive_message() will block on recvmsg()
 260          * and never return (even if a signal is received).
 261          */
 262         ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
 263         if (ret < 0) {
 264                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 265                 goto end_unlock;
 266         }
 267         ret = lttng_poll_add(&events, channel->socket, LPOLLIN | LPOLLERR);
 268         if (ret < 0) {
 269                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 270                 goto end_clean_poll;
 271         }
 272         ret = lttng_poll_wait_interruptible(&events, -1);
 273         if (ret <= 0) {
 274                 status = (ret == -1 && errno == EINTR) ?
 275                         LTTNG_NOTIFICATION_CHANNEL_STATUS_INTERRUPTED :
 276                         LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 277                 goto end_clean_poll;
 278         }
 279
 280         ret = receive_message(channel);
 281         if (ret) {
 282                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 283                 goto end_clean_poll;
 284         }
 285
 286         switch (get_current_message_type(channel)) {
 287         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 288                 notification = create_notification_from_current_message(
 289                                 channel);
 290                 if (!notification) {
 291                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 292                         goto end_clean_poll;
 293                 }
 294                 break;
 295         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 296                 /* No payload to consume. */
 297                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
 298                 break;
 299         default:
 300                 /* Protocol error. */
 301                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 302                 goto end_clean_poll;
 303         }
 304
 305 end_clean_poll:
 306         lttng_poll_clean(&events);
 307 end_unlock:
 308         pthread_mutex_unlock(&channel->lock);
 309         *_notification = notification;
 310 end:
 311         return status;
 312 }
 313
 314 static
 315 int enqueue_dropped_notification(
 316                 struct lttng_notification_channel *channel)
 317 {
 318         int ret = 0;
 319         struct pending_notification *pending_notification;
 320         struct cds_list_head *last_element =
 321                         channel->pending_notifications.list.prev;
 322
 323         pending_notification = caa_container_of(last_element,
 324                         struct pending_notification, node);
 325         if (!pending_notification->notification) {
 326                 /*
 327                  * The last enqueued notification indicates dropped
 328                  * notifications; there is nothing to do as we group
 329                  * dropped notifications together.
 330                  */
 331                 goto end;
 332         }
 333
 334         if (channel->pending_notifications.count >=
 335                         DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT &&
 336                         pending_notification->notification) {
 337                 /*
 338                  * Discard the last enqueued notification to indicate
 339                  * that notifications were dropped at this point.
 340                  */
 341                 lttng_notification_destroy(
 342                                 pending_notification->notification);
 343                 pending_notification->notification = NULL;
 344                 goto end;
 345         }
 346
 347         pending_notification = zmalloc(sizeof(*pending_notification));
 348         if (!pending_notification) {
 349                 ret = -1;
 350                 goto end;
 351         }
 352         CDS_INIT_LIST_HEAD(&pending_notification->node);
 353         cds_list_add(&pending_notification->node,
 354                         &channel->pending_notifications.list);
 355         channel->pending_notifications.count++;
 356 end:
 357         return ret;
 358 }
 359
 360 static
 361 int enqueue_notification_from_current_message(
 362                 struct lttng_notification_channel *channel)
 363 {
 364         int ret = 0;
 365         struct lttng_notification *notification;
 366         struct pending_notification *pending_notification;
 367
 368         if (channel->pending_notifications.count >=
 369                         DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT) {
 370                 /* Drop the notification. */
 371                 ret = enqueue_dropped_notification(channel);
 372                 goto end;
 373         }
 374
 375         pending_notification = zmalloc(sizeof(*pending_notification));
 376         if (!pending_notification) {
 377                 ret = -1;
 378                 goto error;
 379         }
 380         CDS_INIT_LIST_HEAD(&pending_notification->node);
 381
 382         notification = create_notification_from_current_message(channel);
 383         if (!notification) {
 384                 ret = -1;
 385                 goto error;
 386         }
 387
 388         pending_notification->notification = notification;
 389         cds_list_add(&pending_notification->node,
 390                         &channel->pending_notifications.list);
 391         channel->pending_notifications.count++;
 392 end:
 393         return ret;
 394 error:
 395         free(pending_notification);
 396         goto end;
 397 }
 398
 399 enum lttng_notification_channel_status
 400 lttng_notification_channel_has_pending_notification(
 401                 struct lttng_notification_channel *channel,
 402                 bool *_notification_pending)
 403 {
 404         int ret;
 405         enum lttng_notification_channel_status status =
 406                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 407         struct lttng_poll_event events;
 408
 409         if (!channel || !_notification_pending) {
 410                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 411                 goto end;
 412         }
 413
 414         pthread_mutex_lock(&channel->lock);
 415
 416         if (channel->pending_notifications.count) {
 417                 *_notification_pending = true;
 418                 goto end_unlock;
 419         }
 420
 421         if (channel->socket < 0) {
 422                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED;
 423                 goto end_unlock;
 424         }
 425
 426         /*
 427          * Check, without blocking, if data is available on the channel's
 428          * socket. If there is data available, it is safe to read (blocking)
 429          * on the socket for a message from the session daemon.
 430          *
 431          * Since all commands wait for the session daemon's reply before
 432          * releasing the channel's lock, the protocol only allows for
 433          * notifications and "notification dropped" messages to come
 434          * through. If we receive a different message type, it is
 435          * considered a protocol error.
 436          *
 437          * Note that this function is not guaranteed not to block. This
 438          * will block until our peer (the session daemon) has sent a complete
 439          * message if we see data available on the socket. If the peer does
 440          * not respect the protocol, this may block indefinitely.
 441          */
 442         ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
 443         if (ret < 0) {
 444                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 445                 goto end_unlock;
 446         }
 447         ret = lttng_poll_add(&events, channel->socket, LPOLLIN | LPOLLERR);
 448         if (ret < 0) {
 449                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 450                 goto end_clean_poll;
 451         }
 452         /* timeout = 0: return immediately. */
 453         ret = lttng_poll_wait_interruptible(&events, 0);
 454         if (ret == 0) {
 455                 /* No data available. */
 456                 *_notification_pending = false;
 457                 goto end_clean_poll;
 458         } else if (ret < 0) {
 459                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 460                 goto end_clean_poll;
 461         }
 462
 463         /* Data available on socket. */
 464         ret = receive_message(channel);
 465         if (ret) {
 466                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 467                 goto end_clean_poll;
 468         }
 469
 470         switch (get_current_message_type(channel)) {
 471         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 472                 ret = enqueue_notification_from_current_message(channel);
 473                 if (ret) {
 474                         goto end_clean_poll;
 475                 }
 476                 *_notification_pending = true;
 477                 break;
 478         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 479                 ret = enqueue_dropped_notification(channel);
 480                 if (ret) {
 481                         goto end_clean_poll;
 482                 }
 483                 *_notification_pending = true;
 484                 break;
 485         default:
 486                 /* Protocol error. */
 487                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 488                 goto end_clean_poll;
 489         }
 490
 491 end_clean_poll:
 492         lttng_poll_clean(&events);
 493 end_unlock:
 494         pthread_mutex_unlock(&channel->lock);
 495 end:
 496         return status;
 497 }
 498
 499 static
 500 int receive_command_reply(struct lttng_notification_channel *channel,
 501                 enum lttng_notification_channel_status *status)
 502 {
 503         int ret;
 504         struct lttng_notification_channel_command_reply *reply;
 505
 506         while (true) {
 507                 enum lttng_notification_channel_message_type msg_type;
 508
 509                 ret = receive_message(channel);
 510                 if (ret) {
 511                         goto end;
 512                 }
 513
 514                 msg_type = get_current_message_type(channel);
 515                 switch (msg_type) {
 516                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY:
 517                         goto exit_loop;
 518                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 519                         ret = enqueue_notification_from_current_message(
 520                                         channel);
 521                         if (ret) {
 522                                 goto end;
 523                         }
 524                         break;
 525                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 526                         ret = enqueue_dropped_notification(channel);
 527                         if (ret) {
 528                                 goto end;
 529                         }
 530                         break;
 531                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
 532                 {
 533                         struct lttng_notification_channel_command_handshake *handshake;
 534
 535                         handshake = (struct lttng_notification_channel_command_handshake *)
 536                                         (channel->reception_payload.buffer.data +
 537                                         sizeof(struct lttng_notification_channel_message));
 538                         channel->version.major = handshake->major;
 539                         channel->version.minor = handshake->minor;
 540                         channel->version.set = true;
 541                         break;
 542                 }
 543                 default:
 544                         ret = -1;
 545                         goto end;
 546                 }
 547         }
 548
 549 exit_loop:
 550         if (channel->reception_payload.buffer.size <
 551                         (sizeof(struct lttng_notification_channel_message) +
 552                         sizeof(*reply))) {
 553                 /* Invalid message received. */
 554                 ret = -1;
 555                 goto end;
 556         }
 557
 558         reply = (struct lttng_notification_channel_command_reply *)
 559                         (channel->reception_payload.buffer.data +
 560                         sizeof(struct lttng_notification_channel_message));
 561         *status = (enum lttng_notification_channel_status) reply->status;
 562 end:
 563         return ret;
 564 }
 565
 566 static
 567 int handshake(struct lttng_notification_channel *channel)
 568 {
 569         ssize_t ret;
 570         enum lttng_notification_channel_status status =
 571                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 572         struct lttng_notification_channel_command_handshake handshake = {
 573                 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
 574                 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
 575         };
 576         struct lttng_notification_channel_message msg_header = {
 577                 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
 578                 .size = sizeof(handshake),
 579         };
 580         char send_buffer[sizeof(msg_header) + sizeof(handshake)];
 581
 582         memcpy(send_buffer, &msg_header, sizeof(msg_header));
 583         memcpy(send_buffer + sizeof(msg_header), &handshake, sizeof(handshake));
 584
 585         pthread_mutex_lock(&channel->lock);
 586
 587         ret = lttcomm_send_creds_unix_sock(channel->socket, send_buffer,
 588                         sizeof(send_buffer));
 589         if (ret < 0) {
 590                 goto end_unlock;
 591         }
 592
 593         /* Receive handshake info from the sessiond. */
 594         ret = receive_command_reply(channel, &status);
 595         if (ret < 0) {
 596                 goto end_unlock;
 597         }
 598
 599         if (!channel->version.set) {
 600                 ret = -1;
 601                 goto end_unlock;
 602         }
 603
 604         if (channel->version.major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
 605                 ret = -1;
 606                 goto end_unlock;
 607         }
 608
 609 end_unlock:
 610         pthread_mutex_unlock(&channel->lock);
 611         return ret;
 612 }
 613
 614 static
 615 enum lttng_notification_channel_status send_condition_command(
 616                 struct lttng_notification_channel *channel,
 617                 enum lttng_notification_channel_message_type type,
 618                 const struct lttng_condition *condition)
 619 {
 620         int socket;
 621         ssize_t ret;
 622         enum lttng_notification_channel_status status =
 623                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 624         struct lttng_payload payload;
 625         struct lttng_notification_channel_message cmd_header = {
 626                 .type = (int8_t) type,
 627         };
 628
 629         lttng_payload_init(&payload);
 630
 631         if (!channel) {
 632                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 633                 goto end;
 634         }
 635
 636         assert(type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE ||
 637                 type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE);
 638
 639         pthread_mutex_lock(&channel->lock);
 640         socket = channel->socket;
 641
 642         if (!lttng_condition_validate(condition)) {
 643                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 644                 goto end_unlock;
 645         }
 646
 647         ret = lttng_dynamic_buffer_append(&payload.buffer, &cmd_header,
 648                         sizeof(cmd_header));
 649         if (ret) {
 650                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 651                 goto end_unlock;
 652         }
 653
 654         ret = lttng_condition_serialize(condition, &payload);
 655         if (ret) {
 656                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 657                 goto end_unlock;
 658         }
 659
 660         /* Update payload length. */
 661         ((struct lttng_notification_channel_message *) payload.buffer.data)->size =
 662                         (uint32_t) (payload.buffer.size - sizeof(cmd_header));
 663
 664         {
 665                 struct lttng_payload_view pv =
 666                                 lttng_payload_view_from_payload(
 667                                                 &payload, 0, -1);
 668                 const int fd_count =
 669                                 lttng_payload_view_get_fd_handle_count(&pv);
 670
 671                 /* Update fd count. */
 672                 ((struct lttng_notification_channel_message *) payload.buffer.data)->fds =
 673                         (uint32_t) fd_count;
 674
 675                 ret = lttcomm_send_unix_sock(
 676                         socket, pv.buffer.data, pv.buffer.size);
 677                 if (ret < 0) {
 678                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 679                         goto end_unlock;
 680                 }
 681
 682                 /* Pass fds if present. */
 683                 if (fd_count > 0) {
 684                         ret = lttcomm_send_payload_view_fds_unix_sock(socket,
 685                                         &pv);
 686                         if (ret < 0) {
 687                                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 688                                 goto end_unlock;
 689                         }
 690                 }
 691         }
 692
 693         ret = receive_command_reply(channel, &status);
 694         if (ret < 0) {
 695                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 696                 goto end_unlock;
 697         }
 698 end_unlock:
 699         pthread_mutex_unlock(&channel->lock);
 700 end:
 701         lttng_payload_reset(&payload);
 702         return status;
 703 }
 704
 705 enum lttng_notification_channel_status lttng_notification_channel_subscribe(
 706                 struct lttng_notification_channel *channel,
 707                 const struct lttng_condition *condition)
 708 {
 709         return send_condition_command(channel,
 710                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE,
 711                         condition);
 712 }
 713
 714 enum lttng_notification_channel_status lttng_notification_channel_unsubscribe(
 715                 struct lttng_notification_channel *channel,
 716                 const struct lttng_condition *condition)
 717 {
 718         return send_condition_command(channel,
 719                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE,
 720                         condition);
 721 }
 722
 723 void lttng_notification_channel_destroy(
 724                 struct lttng_notification_channel *channel)
 725 {
 726         if (!channel) {
 727                 return;
 728         }
 729
 730         if (channel->socket >= 0) {
 731                 (void) lttcomm_close_unix_sock(channel->socket);
 732         }
 733         pthread_mutex_destroy(&channel->lock);
 734         lttng_payload_reset(&channel->reception_payload);
 735         free(channel);
 736 }