2 * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 * This library is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU Lesser General Public License, version 2.1 only,
6 * as published by the Free Software Foundation.
8 * This library is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
13 * You should have received a copy of the GNU Lesser General Public License
14 * along with this library; if not, write to the Free Software Foundation,
15 * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18 #include <lttng/notification/notification-internal.h>
19 #include <lttng/notification/channel-internal.h>
20 #include <lttng/condition/condition-internal.h>
21 #include <lttng/endpoint.h>
22 #include <common/defaults.h>
23 #include <common/error.h>
24 #include <common/dynamic-buffer.h>
25 #include <common/utils.h>
26 #include <common/defaults.h>
28 #include "lttng-ctl-helper.h"
31 int handshake(struct lttng_notification_channel
*channel
);
34 * Populates the reception buffer with the next complete message.
35 * The caller must acquire the client's lock.
38 int receive_message(struct lttng_notification_channel
*channel
)
41 struct lttng_notification_channel_message msg
;
43 ret
= lttng_dynamic_buffer_set_size(&channel
->reception_buffer
, 0);
48 ret
= lttcomm_recv_unix_sock(channel
->socket
, &msg
, sizeof(msg
));
54 if (msg
.size
> DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE
) {
59 /* Add message header at buffer's start. */
60 ret
= lttng_dynamic_buffer_append(&channel
->reception_buffer
, &msg
,
66 /* Reserve space for the payload. */
67 ret
= lttng_dynamic_buffer_set_size(&channel
->reception_buffer
,
68 channel
->reception_buffer
.size
+ msg
.size
);
73 /* Receive message payload. */
74 ret
= lttcomm_recv_unix_sock(channel
->socket
,
75 channel
->reception_buffer
.data
+ sizeof(msg
), msg
.size
);
76 if (ret
< (ssize_t
) msg
.size
) {
84 if (lttng_dynamic_buffer_set_size(&channel
->reception_buffer
, 0)) {
91 enum lttng_notification_channel_message_type
get_current_message_type(
92 struct lttng_notification_channel
*channel
)
94 struct lttng_notification_channel_message
*msg
;
96 assert(channel
->reception_buffer
.size
>= sizeof(*msg
));
98 msg
= (struct lttng_notification_channel_message
*)
99 channel
->reception_buffer
.data
;
100 return (enum lttng_notification_channel_message_type
) msg
->type
;
104 struct lttng_notification
*create_notification_from_current_message(
105 struct lttng_notification_channel
*channel
)
108 struct lttng_notification
*notification
= NULL
;
109 struct lttng_buffer_view view
;
111 if (channel
->reception_buffer
.size
<=
112 sizeof(struct lttng_notification_channel_message
)) {
116 view
= lttng_buffer_view_from_dynamic_buffer(&channel
->reception_buffer
,
117 sizeof(struct lttng_notification_channel_message
), -1);
119 ret
= lttng_notification_create_from_buffer(&view
, ¬ification
);
120 if (ret
!= channel
->reception_buffer
.size
-
121 sizeof(struct lttng_notification_channel_message
)) {
122 lttng_notification_destroy(notification
);
130 struct lttng_notification_channel
*lttng_notification_channel_create(
131 struct lttng_endpoint
*endpoint
)
134 bool is_in_tracing_group
= false, is_root
= false;
135 char *sock_path
= NULL
;
136 struct lttng_notification_channel
*channel
= NULL
;
139 endpoint
!= lttng_session_daemon_notification_endpoint
) {
143 sock_path
= zmalloc(LTTNG_PATH_MAX
);
148 channel
= zmalloc(sizeof(struct lttng_notification_channel
));
152 channel
->socket
= -1;
153 pthread_mutex_init(&channel
->lock
, NULL
);
154 lttng_dynamic_buffer_init(&channel
->reception_buffer
);
155 CDS_INIT_LIST_HEAD(&channel
->pending_notifications
.list
);
157 is_root
= (getuid() == 0);
159 is_in_tracing_group
= lttng_check_tracing_group();
162 if (is_root
|| is_in_tracing_group
) {
163 lttng_ctl_copy_string(sock_path
,
164 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK
,
166 ret
= lttcomm_connect_unix_sock(sock_path
);
173 /* Fallback to local session daemon. */
174 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
175 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK
,
176 utils_get_home_dir());
177 if (ret
< 0 || ret
>= LTTNG_PATH_MAX
) {
181 ret
= lttcomm_connect_unix_sock(sock_path
);
188 channel
->socket
= fd
;
190 ret
= handshake(channel
);
198 lttng_notification_channel_destroy(channel
);
203 enum lttng_notification_channel_status
204 lttng_notification_channel_get_next_notification(
205 struct lttng_notification_channel
*channel
,
206 struct lttng_notification
**_notification
)
209 struct lttng_notification
*notification
= NULL
;
210 enum lttng_notification_channel_status status
=
211 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
213 if (!channel
|| !_notification
) {
214 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID
;
218 pthread_mutex_lock(&channel
->lock
);
220 if (channel
->pending_notifications
.count
) {
221 struct pending_notification
*pending_notification
;
223 assert(!cds_list_empty(&channel
->pending_notifications
.list
));
225 /* Deliver one of the pending notifications. */
226 pending_notification
= cds_list_first_entry(
227 &channel
->pending_notifications
.list
,
228 struct pending_notification
,
230 notification
= pending_notification
->notification
;
232 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
;
234 cds_list_del(&pending_notification
->node
);
235 channel
->pending_notifications
.count
--;
236 free(pending_notification
);
240 ret
= receive_message(channel
);
242 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR
;
246 switch (get_current_message_type(channel
)) {
247 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION
:
248 notification
= create_notification_from_current_message(
251 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR
;
255 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED
:
256 /* No payload to consume. */
257 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
;
260 /* Protocol error. */
261 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR
;
266 pthread_mutex_unlock(&channel
->lock
);
267 *_notification
= notification
;
273 int enqueue_dropped_notification(
274 struct lttng_notification_channel
*channel
)
277 struct pending_notification
*pending_notification
;
278 struct cds_list_head
*last_element
=
279 channel
->pending_notifications
.list
.prev
;
281 pending_notification
= caa_container_of(last_element
,
282 struct pending_notification
, node
);
283 if (!pending_notification
->notification
) {
285 * The last enqueued notification indicates dropped
286 * notifications; there is nothing to do as we group
287 * dropped notifications together.
292 if (channel
->pending_notifications
.count
>=
293 DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT
&&
294 pending_notification
->notification
) {
296 * Discard the last enqueued notification to indicate
297 * that notifications were dropped at this point.
299 lttng_notification_destroy(
300 pending_notification
->notification
);
301 pending_notification
->notification
= NULL
;
305 pending_notification
= zmalloc(sizeof(*pending_notification
));
306 if (!pending_notification
) {
310 CDS_INIT_LIST_HEAD(&pending_notification
->node
);
311 cds_list_add(&pending_notification
->node
,
312 &channel
->pending_notifications
.list
);
313 channel
->pending_notifications
.count
++;
319 int enqueue_notification_from_current_message(
320 struct lttng_notification_channel
*channel
)
323 struct lttng_notification
*notification
;
324 struct pending_notification
*pending_notification
;
326 if (channel
->pending_notifications
.count
>=
327 DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT
) {
328 /* Drop the notification. */
329 ret
= enqueue_dropped_notification(channel
);
333 pending_notification
= zmalloc(sizeof(*pending_notification
));
334 if (!pending_notification
) {
338 CDS_INIT_LIST_HEAD(&pending_notification
->node
);
340 notification
= create_notification_from_current_message(channel
);
346 pending_notification
->notification
= notification
;
347 cds_list_add(&pending_notification
->node
,
348 &channel
->pending_notifications
.list
);
349 channel
->pending_notifications
.count
++;
353 free(pending_notification
);
358 int receive_command_reply(struct lttng_notification_channel
*channel
,
359 enum lttng_notification_channel_status
*status
)
362 struct lttng_notification_channel_command_reply
*reply
;
365 enum lttng_notification_channel_message_type msg_type
;
367 ret
= receive_message(channel
);
372 msg_type
= get_current_message_type(channel
);
374 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY
:
376 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION
:
377 ret
= enqueue_notification_from_current_message(
383 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED
:
384 ret
= enqueue_dropped_notification(channel
);
389 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
:
391 struct lttng_notification_channel_command_handshake
*handshake
;
393 handshake
= (struct lttng_notification_channel_command_handshake
*)
394 (channel
->reception_buffer
.data
+
395 sizeof(struct lttng_notification_channel_message
));
396 channel
->version
.major
= handshake
->major
;
397 channel
->version
.minor
= handshake
->minor
;
398 channel
->version
.set
= true;
408 if (channel
->reception_buffer
.size
<
409 (sizeof(struct lttng_notification_channel_message
) +
411 /* Invalid message received. */
416 reply
= (struct lttng_notification_channel_command_reply
*)
417 (channel
->reception_buffer
.data
+
418 sizeof(struct lttng_notification_channel_message
));
419 *status
= (enum lttng_notification_channel_status
) reply
->status
;
425 int handshake(struct lttng_notification_channel
*channel
)
428 enum lttng_notification_channel_status status
=
429 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
430 struct lttng_notification_channel_command_handshake handshake
= {
431 .major
= LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR
,
432 .minor
= LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR
,
434 struct lttng_notification_channel_message msg_header
= {
435 .type
= LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE
,
436 .size
= sizeof(handshake
),
438 char send_buffer
[sizeof(msg_header
) + sizeof(handshake
)];
440 memcpy(send_buffer
, &msg_header
, sizeof(msg_header
));
441 memcpy(send_buffer
+ sizeof(msg_header
), &handshake
, sizeof(handshake
));
443 pthread_mutex_lock(&channel
->lock
);
445 ret
= lttcomm_send_creds_unix_sock(channel
->socket
, send_buffer
,
446 sizeof(send_buffer
));
451 /* Receive handshake info from the sessiond. */
452 ret
= receive_command_reply(channel
, &status
);
457 if (!channel
->version
.set
) {
462 if (channel
->version
.major
!= LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR
) {
468 pthread_mutex_unlock(&channel
->lock
);
473 enum lttng_notification_channel_status
send_condition_command(
474 struct lttng_notification_channel
*channel
,
475 enum lttng_notification_channel_message_type type
,
476 const struct lttng_condition
*condition
)
479 ssize_t command_size
, ret
;
480 enum lttng_notification_channel_status status
=
481 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
;
482 char *command_buffer
= NULL
;
483 struct lttng_notification_channel_message cmd_message
= {
488 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID
;
492 assert(type
== LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE
||
493 type
== LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE
);
495 pthread_mutex_lock(&channel
->lock
);
496 socket
= channel
->socket
;
497 if (!lttng_condition_validate(condition
)) {
498 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID
;
502 ret
= lttng_condition_serialize(condition
, NULL
);
504 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID
;
507 assert(ret
< UINT32_MAX
);
508 cmd_message
.size
= (uint32_t) ret
;
509 command_size
= ret
+ sizeof(
510 struct lttng_notification_channel_message
);
511 command_buffer
= zmalloc(command_size
);
512 if (!command_buffer
) {
516 memcpy(command_buffer
, &cmd_message
, sizeof(cmd_message
));
517 ret
= lttng_condition_serialize(condition
,
518 command_buffer
+ sizeof(cmd_message
));
523 ret
= lttcomm_send_unix_sock(socket
, command_buffer
, command_size
);
525 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR
;
529 ret
= receive_command_reply(channel
, &status
);
531 status
= LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR
;
535 pthread_mutex_unlock(&channel
->lock
);
537 free(command_buffer
);
541 enum lttng_notification_channel_status
lttng_notification_channel_subscribe(
542 struct lttng_notification_channel
*channel
,
543 const struct lttng_condition
*condition
)
545 return send_condition_command(channel
,
546 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE
,
550 enum lttng_notification_channel_status
lttng_notification_channel_unsubscribe(
551 struct lttng_notification_channel
*channel
,
552 const struct lttng_condition
*condition
)
554 return send_condition_command(channel
,
555 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE
,
559 void lttng_notification_channel_destroy(
560 struct lttng_notification_channel
*channel
)
566 if (channel
->socket
>= 0) {
567 (void) lttcomm_close_unix_sock(channel
->socket
);
569 pthread_mutex_destroy(&channel
->lock
);
570 lttng_dynamic_buffer_reset(&channel
->reception_buffer
);