Remove stdbool.h includes from C++ files
[babeltrace.git] / src / plugins / ctf / lttng-live / viewer-connection.cpp
CommitLineData
7cdc2bab 1/*
0235b0db 2 * SPDX-License-Identifier: MIT
7cdc2bab 3 *
0235b0db
MJ
4 * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
5 * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
7cdc2bab
MD
6 */
7
3c22a242 8#include <fcntl.h>
c802cacb 9#include <glib.h>
7cdc2bab 10#include <stdint.h>
3c22a242 11#include <stdio.h>
7cdc2bab 12#include <stdlib.h>
3c22a242 13#include <sys/types.h>
7cdc2bab 14#include <unistd.h>
3c22a242 15
c802cacb
SM
16#include <babeltrace2/babeltrace.h>
17
18#define BT_COMP_LOG_SELF_COMP (viewer_connection->self_comp)
19#define BT_LOG_OUTPUT_LEVEL (viewer_connection->log_level)
20#define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE/VIEWER"
21#include "logging/comp-logging.h"
7cdc2bab 22
578e048b 23#include "common/common.h"
c802cacb
SM
24#include "compat/compiler.h"
25#include "compat/endian.h"
26#include "compat/socket.h"
7cdc2bab 27
c802cacb 28#include "data-stream.hpp"
087cd0f5 29#include "lttng-live.hpp"
087cd0f5 30#include "lttng-viewer-abi.hpp"
087cd0f5 31#include "metadata.hpp"
c802cacb 32#include "viewer-connection.hpp"
7cdc2bab 33
4164020e
SM
34#define viewer_handle_send_recv_status(_self_comp, _self_comp_class, _status, _action, _msg_str) \
35 do { \
36 switch (_status) { \
37 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED: \
38 break; \
39 case LTTNG_LIVE_VIEWER_STATUS_ERROR: \
40 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(_self_comp, _self_comp_class, \
41 "Error " _action " " _msg_str); \
42 break; \
43 default: \
44 bt_common_abort(); \
45 } \
46 } while (0)
47
48#define viewer_handle_send_status(_self_comp, _self_comp_class, _status, _msg_str) \
49 viewer_handle_send_recv_status(_self_comp, _self_comp_class, _status, "sending", _msg_str)
50
51#define viewer_handle_recv_status(_self_comp, _self_comp_class, _status, _msg_str) \
52 viewer_handle_send_recv_status(_self_comp, _self_comp_class, _status, "receiving", _msg_str)
53
54#define LTTNG_LIVE_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE_ERRNO(_self_comp, _self_comp_class, _msg, \
55 _fmt, ...) \
56 do { \
57 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(_self_comp, _self_comp_class, _msg ": %s" _fmt, \
58 bt_socket_errormsg(), ##__VA_ARGS__); \
59 } while (0)
60
61static const char *lttng_viewer_command_string(enum lttng_viewer_command cmd)
f93afbf9 62{
4164020e
SM
63 switch (cmd) {
64 case LTTNG_VIEWER_CONNECT:
65 return "CONNECT";
66 case LTTNG_VIEWER_LIST_SESSIONS:
67 return "LIST_SESSIONS";
68 case LTTNG_VIEWER_ATTACH_SESSION:
69 return "ATTACH_SESSION";
70 case LTTNG_VIEWER_GET_NEXT_INDEX:
71 return "GET_NEXT_INDEX";
72 case LTTNG_VIEWER_GET_PACKET:
73 return "GET_PACKET";
74 case LTTNG_VIEWER_GET_METADATA:
75 return "GET_METADATA";
76 case LTTNG_VIEWER_GET_NEW_STREAMS:
77 return "GET_NEW_STREAMS";
78 case LTTNG_VIEWER_CREATE_SESSION:
79 return "CREATE_SESSION";
80 case LTTNG_VIEWER_DETACH_SESSION:
81 return "DETACH_SESSION";
82 }
83
84 bt_common_abort();
f93afbf9
FD
85}
86
4164020e
SM
87static const char *
88lttng_viewer_next_index_return_code_string(enum lttng_viewer_next_index_return_code code)
f93afbf9 89{
4164020e
SM
90 switch (code) {
91 case LTTNG_VIEWER_INDEX_OK:
92 return "INDEX_OK";
93 case LTTNG_VIEWER_INDEX_RETRY:
94 return "INDEX_RETRY";
95 case LTTNG_VIEWER_INDEX_HUP:
96 return "INDEX_HUP";
97 case LTTNG_VIEWER_INDEX_ERR:
98 return "INDEX_ERR";
99 case LTTNG_VIEWER_INDEX_INACTIVE:
100 return "INDEX_INACTIVE";
101 case LTTNG_VIEWER_INDEX_EOF:
102 return "INDEX_EOF";
103 }
104
105 bt_common_abort();
f93afbf9
FD
106}
107
4164020e 108static const char *lttng_viewer_next_index_return_code_string(uint32_t code)
087cd0f5 109{
4164020e 110 return lttng_viewer_next_index_return_code_string((lttng_viewer_next_index_return_code) code);
087cd0f5
SM
111}
112
4164020e
SM
113static const char *
114lttng_viewer_get_packet_return_code_string(enum lttng_viewer_get_packet_return_code code)
f93afbf9 115{
4164020e
SM
116 switch (code) {
117 case LTTNG_VIEWER_GET_PACKET_OK:
118 return "GET_PACKET_OK";
119 case LTTNG_VIEWER_GET_PACKET_RETRY:
120 return "GET_PACKET_RETRY";
121 case LTTNG_VIEWER_GET_PACKET_ERR:
122 return "GET_PACKET_ERR";
123 case LTTNG_VIEWER_GET_PACKET_EOF:
124 return "GET_PACKET_EOF";
125 }
126
127 bt_common_abort();
f93afbf9
FD
128};
129
4164020e 130static const char *lttng_viewer_get_packet_return_code_string(uint32_t code)
087cd0f5 131{
4164020e 132 return lttng_viewer_get_packet_return_code_string((lttng_viewer_get_packet_return_code) code);
087cd0f5
SM
133}
134
4164020e 135static const char *lttng_viewer_seek_string(enum lttng_viewer_seek seek)
f93afbf9 136{
4164020e
SM
137 switch (seek) {
138 case LTTNG_VIEWER_SEEK_BEGINNING:
139 return "SEEK_BEGINNING";
140 case LTTNG_VIEWER_SEEK_LAST:
141 return "SEEK_LAST";
142 }
143
144 bt_common_abort();
f93afbf9
FD
145}
146
4164020e
SM
147static inline enum lttng_live_iterator_status
148viewer_status_to_live_iterator_status(enum lttng_live_viewer_status viewer_status)
f79c2d7a 149{
4164020e
SM
150 switch (viewer_status) {
151 case LTTNG_LIVE_VIEWER_STATUS_OK:
152 return LTTNG_LIVE_ITERATOR_STATUS_OK;
153 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
154 return LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
155 case LTTNG_LIVE_VIEWER_STATUS_ERROR:
156 return LTTNG_LIVE_ITERATOR_STATUS_ERROR;
157 }
158
159 bt_common_abort();
f79c2d7a
FD
160}
161
4164020e
SM
162static inline enum ctf_msg_iter_medium_status
163viewer_status_to_ctf_msg_iter_medium_status(enum lttng_live_viewer_status viewer_status)
f79c2d7a 164{
4164020e
SM
165 switch (viewer_status) {
166 case LTTNG_LIVE_VIEWER_STATUS_OK:
167 return CTF_MSG_ITER_MEDIUM_STATUS_OK;
168 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
169 return CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
170 case LTTNG_LIVE_VIEWER_STATUS_ERROR:
171 return CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
172 }
173
174 bt_common_abort();
f79c2d7a
FD
175}
176
4164020e 177static inline void viewer_connection_close_socket(struct live_viewer_connection *viewer_connection)
b197ca37 178{
4164020e
SM
179 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
180 bt_self_component *self_comp = viewer_connection->self_comp;
181 int ret = bt_socket_close(viewer_connection->control_sock);
182 if (ret == -1) {
183 BT_COMP_OR_COMP_CLASS_LOGW_ERRNO(self_comp, self_comp_class,
184 "Error closing viewer connection socket: ", ".");
185 }
186
187 viewer_connection->control_sock = BT_INVALID_SOCKET;
b197ca37
FD
188}
189
f79c2d7a
FD
190/*
191 * This function receives a message from the Relay daemon.
192 * If it received the entire message, it returns _OK,
193 * If it's interrupted, it returns _INTERRUPTED,
194 * otherwise, it returns _ERROR.
195 */
4164020e
SM
196static enum lttng_live_viewer_status
197lttng_live_recv(struct live_viewer_connection *viewer_connection, void *buf, size_t len)
7cdc2bab 198{
4164020e
SM
199 ssize_t received;
200 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
201 bt_self_component *self_comp = viewer_connection->self_comp;
202 size_t total_received = 0, to_receive = len;
203 struct lttng_live_msg_iter *lttng_live_msg_iter = viewer_connection->lttng_live_msg_iter;
204 enum lttng_live_viewer_status status;
205 BT_SOCKET sock = viewer_connection->control_sock;
206
207 /*
208 * Receive a message from the Relay.
209 */
210 do {
211 received = bt_socket_recv(sock, (char *) buf + total_received, to_receive, 0);
212 if (received == BT_SOCKET_ERROR) {
213 if (bt_socket_interrupted()) {
214 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
215 /*
216 * This interruption was due to a
217 * SIGINT and the graph is being torn
218 * down.
219 */
220 status = LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED;
221 lttng_live_msg_iter->was_interrupted = true;
222 goto end;
223 } else {
224 /*
225 * A signal was received, but the graph
226 * is not being torn down. Carry on.
227 */
228 continue;
229 }
230 } else {
231 /*
232 * For any other types of socket error, close
233 * the socket and return an error.
234 */
235 LTTNG_LIVE_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE_ERRNO(
236 self_comp, self_comp_class, "Error receiving from Relay", ".");
237
238 viewer_connection_close_socket(viewer_connection);
239 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
240 goto end;
241 }
242 } else if (received == 0) {
243 /*
244 * The recv() call returned 0. This means the
245 * connection was orderly shutdown from the other peer.
246 * If that happens when we are trying to receive
247 * a message from it, it means something when wrong.
248 * Close the socket and return an error.
249 */
250 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
251 "Remote side has closed connection");
252 viewer_connection_close_socket(viewer_connection);
253 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
254 goto end;
255 }
256
257 BT_ASSERT(received <= to_receive);
258 total_received += received;
259 to_receive -= received;
260
261 } while (to_receive > 0);
262
263 BT_ASSERT(total_received == len);
264 status = LTTNG_LIVE_VIEWER_STATUS_OK;
f79c2d7a
FD
265
266end:
4164020e 267 return status;
7cdc2bab
MD
268}
269
f79c2d7a
FD
270/*
271 * This function sends a message to the Relay daemon.
272 * If it send the message, it returns _OK,
273 * If it's interrupted, it returns _INTERRUPTED,
274 * otherwise, it returns _ERROR.
275 */
4164020e
SM
276static enum lttng_live_viewer_status
277lttng_live_send(struct live_viewer_connection *viewer_connection, const void *buf, size_t len)
7cdc2bab 278{
4164020e
SM
279 enum lttng_live_viewer_status status;
280 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
281 bt_self_component *self_comp = viewer_connection->self_comp;
282 struct lttng_live_msg_iter *lttng_live_msg_iter = viewer_connection->lttng_live_msg_iter;
283 BT_SOCKET sock = viewer_connection->control_sock;
284 size_t to_send = len;
285 ssize_t total_sent = 0;
286
287 do {
288 ssize_t sent = bt_socket_send_nosigpipe(sock, (char *) buf + total_sent, to_send);
289 if (sent == BT_SOCKET_ERROR) {
290 if (bt_socket_interrupted()) {
291 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
292 /*
293 * This interruption was a SIGINT and
294 * the graph is being teared down.
295 */
296 status = LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED;
297 lttng_live_msg_iter->was_interrupted = true;
298 goto end;
299 } else {
300 /*
301 * A signal was received, but the graph
302 * is not being teared down. Carry on.
303 */
304 continue;
305 }
306 } else {
307 /*
308 * For any other types of socket error, close
309 * the socket and return an error.
310 */
311 LTTNG_LIVE_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE_ERRNO(
312 self_comp, self_comp_class, "Error sending to Relay", ".");
313
314 viewer_connection_close_socket(viewer_connection);
315 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
316 goto end;
317 }
318 }
319
320 BT_ASSERT(sent <= to_send);
321 total_sent += sent;
322 to_send -= sent;
323
324 } while (to_send > 0);
325
326 BT_ASSERT(total_sent == len);
327 status = LTTNG_LIVE_VIEWER_STATUS_OK;
f79c2d7a
FD
328
329end:
4164020e 330 return status;
7cdc2bab
MD
331}
332
4164020e 333static int parse_url(struct live_viewer_connection *viewer_connection)
7cdc2bab 334{
4164020e
SM
335 char error_buf[256] = {0};
336 bt_self_component *self_comp = viewer_connection->self_comp;
337 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
dd420a9b 338 struct bt_common_lttng_live_url_parts lttng_live_url_parts = {};
4164020e
SM
339 int ret = -1;
340 const char *path = viewer_connection->url->str;
341
342 if (!path) {
343 goto end;
344 }
345
346 lttng_live_url_parts = bt_common_parse_lttng_live_url(path, error_buf, sizeof(error_buf));
347 if (!lttng_live_url_parts.proto) {
348 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
349 "Invalid LTTng live URL format: %s", error_buf);
350 goto end;
351 }
352 viewer_connection->proto = lttng_live_url_parts.proto;
353 lttng_live_url_parts.proto = NULL;
354
355 viewer_connection->relay_hostname = lttng_live_url_parts.hostname;
356 lttng_live_url_parts.hostname = NULL;
357
358 if (lttng_live_url_parts.port >= 0) {
359 viewer_connection->port = lttng_live_url_parts.port;
360 } else {
361 viewer_connection->port = LTTNG_DEFAULT_NETWORK_VIEWER_PORT;
362 }
363
364 viewer_connection->target_hostname = lttng_live_url_parts.target_hostname;
365 lttng_live_url_parts.target_hostname = NULL;
366
367 if (lttng_live_url_parts.session_name) {
368 viewer_connection->session_name = lttng_live_url_parts.session_name;
369 lttng_live_url_parts.session_name = NULL;
370 }
371
372 ret = 0;
7cdc2bab
MD
373
374end:
4164020e
SM
375 bt_common_destroy_lttng_live_url_parts(&lttng_live_url_parts);
376 return ret;
7cdc2bab
MD
377}
378
4164020e
SM
379static enum lttng_live_viewer_status
380lttng_live_handshake(struct live_viewer_connection *viewer_connection)
7cdc2bab 381{
4164020e
SM
382 struct lttng_viewer_cmd cmd;
383 struct lttng_viewer_connect connect;
384 enum lttng_live_viewer_status status;
385 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
386 bt_self_component *self_comp = viewer_connection->self_comp;
387 const size_t cmd_buf_len = sizeof(cmd) + sizeof(connect);
388 char cmd_buf[cmd_buf_len];
389
390 BT_COMP_OR_COMP_CLASS_LOGD(
391 self_comp, self_comp_class,
392 "Handshaking with the relay daemon: cmd=%s, major-version=%u, minor-version=%u",
393 lttng_viewer_command_string(LTTNG_VIEWER_CONNECT), LTTNG_LIVE_MAJOR, LTTNG_LIVE_MINOR);
394
395 cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
396 cmd.data_size = htobe64((uint64_t) sizeof(connect));
397 cmd.cmd_version = htobe32(0);
398
399 connect.viewer_session_id = -1ULL; /* will be set on recv */
400 connect.major = htobe32(LTTNG_LIVE_MAJOR);
401 connect.minor = htobe32(LTTNG_LIVE_MINOR);
402 connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
403
404 /*
405 * Merge the cmd and connection request to prevent a write-write
406 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
407 * second write to be performed quickly in presence of Nagle's algorithm
408 */
409 memcpy(cmd_buf, &cmd, sizeof(cmd));
410 memcpy(cmd_buf + sizeof(cmd), &connect, sizeof(connect));
411
412 status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
413 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
414 viewer_handle_send_status(self_comp, self_comp_class, status, "viewer connect command");
415 goto end;
416 }
417
418 status = lttng_live_recv(viewer_connection, &connect, sizeof(connect));
419 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
420 viewer_handle_recv_status(self_comp, self_comp_class, status, "viewer connect reply");
421 goto end;
422 }
423
424 BT_COMP_OR_COMP_CLASS_LOGI(self_comp, self_comp_class, "Received viewer session ID : %" PRIu64,
425 (uint64_t) be64toh(connect.viewer_session_id));
426 BT_COMP_OR_COMP_CLASS_LOGI(self_comp, self_comp_class, "Relayd version : %u.%u",
427 be32toh(connect.major), be32toh(connect.minor));
428
429 if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) {
430 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
431 "Incompatible lttng-relayd protocol");
432 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
433 goto end;
434 }
435 /* Use the smallest protocol version implemented. */
436 if (LTTNG_LIVE_MINOR > be32toh(connect.minor)) {
437 viewer_connection->minor = be32toh(connect.minor);
438 } else {
439 viewer_connection->minor = LTTNG_LIVE_MINOR;
440 }
441 viewer_connection->major = LTTNG_LIVE_MAJOR;
442
443 status = LTTNG_LIVE_VIEWER_STATUS_OK;
444
445 goto end;
f79c2d7a
FD
446
447end:
4164020e 448 return status;
7cdc2bab
MD
449}
450
4164020e
SM
451static enum lttng_live_viewer_status
452lttng_live_connect_viewer(struct live_viewer_connection *viewer_connection)
7cdc2bab 453{
4164020e
SM
454 struct hostent *host;
455 struct sockaddr_in server_addr;
456 enum lttng_live_viewer_status status;
457 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
458 bt_self_component *self_comp = viewer_connection->self_comp;
459
460 if (parse_url(viewer_connection)) {
461 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class, "Failed to parse URL");
462 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
463 goto error;
464 }
465
466 BT_COMP_OR_COMP_CLASS_LOGD(
467 self_comp, self_comp_class,
468 "Connecting to hostname : %s, port : %d, "
469 "target hostname : %s, session name : %s, proto : %s",
470 viewer_connection->relay_hostname->str, viewer_connection->port,
471 !viewer_connection->target_hostname ? "<none>" : viewer_connection->target_hostname->str,
472 !viewer_connection->session_name ? "<none>" : viewer_connection->session_name->str,
473 viewer_connection->proto->str);
474
475 host = gethostbyname(viewer_connection->relay_hostname->str);
476 if (!host) {
477 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
478 "Cannot lookup hostname: hostname=\"%s\"",
479 viewer_connection->relay_hostname->str);
480 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
481 goto error;
482 }
483
484 if ((viewer_connection->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == BT_INVALID_SOCKET) {
485 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
486 "Socket creation failed: %s", bt_socket_errormsg());
487 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
488 goto error;
489 }
490
491 server_addr.sin_family = AF_INET;
492 server_addr.sin_port = htons(viewer_connection->port);
493 server_addr.sin_addr = *((struct in_addr *) host->h_addr);
494 memset(&(server_addr.sin_zero), 0, 8);
495
496 if (connect(viewer_connection->control_sock, (struct sockaddr *) &server_addr,
497 sizeof(struct sockaddr)) == BT_SOCKET_ERROR) {
498 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class, "Connection failed: %s",
499 bt_socket_errormsg());
500 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
501 goto error;
502 }
503
504 status = lttng_live_handshake(viewer_connection);
505
506 /*
507 * Only print error and append cause in case of error. not in case of
508 * interruption.
509 */
510 if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
511 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
512 "Viewer handshake failed");
513 goto error;
514 } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
515 goto end;
516 }
517
518 goto end;
7cdc2bab
MD
519
520error:
4164020e
SM
521 if (viewer_connection->control_sock != BT_INVALID_SOCKET) {
522 if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
523 BT_COMP_OR_COMP_CLASS_LOGW(self_comp, self_comp_class, "Error closing socket: %s.",
524 bt_socket_errormsg());
525 }
526 }
527 viewer_connection->control_sock = BT_INVALID_SOCKET;
f79c2d7a 528end:
4164020e 529 return status;
7cdc2bab
MD
530}
531
4164020e 532static void lttng_live_disconnect_viewer(struct live_viewer_connection *viewer_connection)
7cdc2bab 533{
4164020e
SM
534 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
535 bt_self_component *self_comp = viewer_connection->self_comp;
536
537 if (viewer_connection->control_sock == BT_INVALID_SOCKET) {
538 return;
539 }
540 if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
541 BT_COMP_OR_COMP_CLASS_LOGW(self_comp, self_comp_class, "Error closing socket: %s",
542 bt_socket_errormsg());
543 viewer_connection->control_sock = BT_INVALID_SOCKET;
544 }
7cdc2bab
MD
545}
546
4164020e
SM
547static int list_update_session(bt_value *results, const struct lttng_viewer_session *session,
548 bool *_found, struct live_viewer_connection *viewer_connection)
7cdc2bab 549{
4164020e
SM
550 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
551 bt_self_component *self_comp = viewer_connection->self_comp;
552 int ret = 0;
553 uint64_t i, len;
554 bt_value *map = NULL;
555 bt_value *hostname = NULL;
556 bt_value *session_name = NULL;
557 bt_value *btval = NULL;
558 bool found = false;
559
560 len = bt_value_array_get_length(results);
561 for (i = 0; i < len; i++) {
562 const char *hostname_str = NULL;
563 const char *session_name_str = NULL;
564
565 map = bt_value_array_borrow_element_by_index(results, i);
566 hostname = bt_value_map_borrow_entry_value(map, "target-hostname");
567 if (!hostname) {
568 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
569 "Error borrowing \"target-hostname\" entry.");
570 ret = -1;
571 goto end;
572 }
573 session_name = bt_value_map_borrow_entry_value(map, "session-name");
574 if (!session_name) {
575 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
576 "Error borrowing \"session-name\" entry.");
577 ret = -1;
578 goto end;
579 }
580 hostname_str = bt_value_string_get(hostname);
581 session_name_str = bt_value_string_get(session_name);
582
583 if (strcmp(session->hostname, hostname_str) == 0 &&
584 strcmp(session->session_name, session_name_str) == 0) {
585 int64_t val;
586 uint32_t streams = be32toh(session->streams);
587 uint32_t clients = be32toh(session->clients);
588
589 found = true;
590
591 btval = bt_value_map_borrow_entry_value(map, "stream-count");
592 if (!btval) {
593 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
594 "Error borrowing \"stream-count\" entry.");
595 ret = -1;
596 goto end;
597 }
598 val = bt_value_integer_unsigned_get(btval);
599 /* sum */
600 val += streams;
601 bt_value_integer_unsigned_set(btval, val);
602
603 btval = bt_value_map_borrow_entry_value(map, "client-count");
604 if (!btval) {
605 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
606 "Error borrowing \"client-count\" entry.");
607 ret = -1;
608 goto end;
609 }
610 val = bt_value_integer_unsigned_get(btval);
611 /* max */
612 val = bt_max_t(int64_t, clients, val);
613 bt_value_integer_unsigned_set(btval, val);
614 }
615
616 if (found) {
617 break;
618 }
619 }
7cdc2bab 620end:
4164020e
SM
621 *_found = found;
622 return ret;
7cdc2bab
MD
623}
624
4164020e
SM
625static int list_append_session(bt_value *results, GString *base_url,
626 const struct lttng_viewer_session *session,
627 struct live_viewer_connection *viewer_connection)
7cdc2bab 628{
4164020e
SM
629 int ret = 0;
630 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
631 bt_value_map_insert_entry_status insert_status;
632 bt_value_array_append_element_status append_status;
633 bt_value *map = NULL;
634 GString *url = NULL;
635 bool found = false;
636
637 /*
638 * If the session already exists, add the stream count to it,
639 * and do max of client counts.
640 */
641 ret = list_update_session(results, session, &found, viewer_connection);
642 if (ret || found) {
643 goto end;
644 }
645
646 map = bt_value_map_create();
647 if (!map) {
648 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error creating map value.");
649 ret = -1;
650 goto end;
651 }
652
653 if (base_url->len < 1) {
654 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error: base_url length smaller than 1.");
655 ret = -1;
656 goto end;
657 }
658 /*
659 * key = "url",
660 * value = <string>,
661 */
662 url = g_string_new(base_url->str);
663 g_string_append(url, "/host/");
664 g_string_append(url, session->hostname);
665 g_string_append_c(url, '/');
666 g_string_append(url, session->session_name);
667
668 insert_status = bt_value_map_insert_string_entry(map, "url", url->str);
669 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
670 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error inserting \"url\" entry.");
671 ret = -1;
672 goto end;
673 }
674
675 /*
676 * key = "target-hostname",
677 * value = <string>,
678 */
679 insert_status = bt_value_map_insert_string_entry(map, "target-hostname", session->hostname);
680 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
681 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class,
682 "Error inserting \"target-hostname\" entry.");
683 ret = -1;
684 goto end;
685 }
686
687 /*
688 * key = "session-name",
689 * value = <string>,
690 */
691 insert_status = bt_value_map_insert_string_entry(map, "session-name", session->session_name);
692 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
693 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error inserting \"session-name\" entry.");
694 ret = -1;
695 goto end;
696 }
697
698 /*
699 * key = "timer-us",
700 * value = <integer>,
701 */
702 {
703 uint32_t live_timer = be32toh(session->live_timer);
704
705 insert_status = bt_value_map_insert_unsigned_integer_entry(map, "timer-us", live_timer);
706 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
707 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error inserting \"timer-us\" entry.");
708 ret = -1;
709 goto end;
710 }
711 }
712
713 /*
714 * key = "stream-count",
715 * value = <integer>,
716 */
717 {
718 uint32_t streams = be32toh(session->streams);
719
720 insert_status = bt_value_map_insert_unsigned_integer_entry(map, "stream-count", streams);
721 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
722 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class,
723 "Error inserting \"stream-count\" entry.");
724 ret = -1;
725 goto end;
726 }
727 }
728
729 /*
730 * key = "client-count",
731 * value = <integer>,
732 */
733 {
734 uint32_t clients = be32toh(session->clients);
735
736 insert_status = bt_value_map_insert_unsigned_integer_entry(map, "client-count", clients);
737 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
738 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class,
739 "Error inserting \"client-count\" entry.");
740 ret = -1;
741 goto end;
742 }
743 }
744
745 append_status = bt_value_array_append_element(results, map);
746 if (append_status != BT_VALUE_ARRAY_APPEND_ELEMENT_STATUS_OK) {
747 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error appending map to results.");
748 ret = -1;
749 }
14f28187 750
7cdc2bab 751end:
4164020e
SM
752 if (url) {
753 g_string_free(url, true);
754 }
755 BT_VALUE_PUT_REF_AND_RESET(map);
756 return ret;
7cdc2bab
MD
757}
758
759/*
760 * Data structure returned:
761 *
762 * {
763 * <array> = {
764 * [n] = {
765 * <map> = {
766 * {
767 * key = "url",
768 * value = <string>,
769 * },
770 * {
771 * key = "target-hostname",
772 * value = <string>,
773 * },
774 * {
775 * key = "session-name",
776 * value = <string>,
777 * },
778 * {
779 * key = "timer-us",
780 * value = <integer>,
781 * },
782 * {
783 * key = "stream-count",
784 * value = <integer>,
785 * },
786 * {
787 * key = "client-count",
788 * value = <integer>,
789 * },
790 * },
791 * }
792 * }
793 */
794
4164020e
SM
795bt_component_class_query_method_status
796live_viewer_connection_list_sessions(struct live_viewer_connection *viewer_connection,
797 const bt_value **user_result)
7cdc2bab 798{
4164020e
SM
799 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
800 bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
801 bt_value *result = NULL;
802 enum lttng_live_viewer_status viewer_status;
803 struct lttng_viewer_cmd cmd;
804 struct lttng_viewer_list_sessions list;
805 uint32_t i, sessions_count;
806
807 result = bt_value_array_create();
808 if (!result) {
809 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error creating array");
810 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
811 goto error;
812 }
813
814 BT_LOGD("Requesting list of sessions: cmd=%s",
815 lttng_viewer_command_string(LTTNG_VIEWER_LIST_SESSIONS));
816
817 cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
818 cmd.data_size = htobe64((uint64_t) 0);
819 cmd.cmd_version = htobe32(0);
820
821 viewer_status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
822 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
823 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error sending list sessions command");
824 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
825 goto error;
826 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
827 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
828 goto error;
829 }
830
831 viewer_status = lttng_live_recv(viewer_connection, &list, sizeof(list));
832 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
833 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error receiving session list");
834 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
835 goto error;
836 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
837 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
838 goto error;
839 }
840
841 sessions_count = be32toh(list.sessions_count);
842 for (i = 0; i < sessions_count; i++) {
843 struct lttng_viewer_session lsession;
844
845 viewer_status = lttng_live_recv(viewer_connection, &lsession, sizeof(lsession));
846 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
847 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error receiving session:");
848 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
849 goto error;
850 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
851 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
852 goto error;
853 }
854
855 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
856 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
857 if (list_append_session(result, viewer_connection->url, &lsession, viewer_connection)) {
858 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class, "Error appending session");
859 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
860 goto error;
861 }
862 }
863
864 *user_result = result;
865 goto end;
7cdc2bab 866error:
4164020e 867 BT_VALUE_PUT_REF_AND_RESET(result);
7cdc2bab 868end:
4164020e 869 return status;
7cdc2bab
MD
870}
871
4164020e
SM
872static enum lttng_live_viewer_status
873lttng_live_query_session_ids(struct lttng_live_msg_iter *lttng_live_msg_iter)
7cdc2bab 874{
4164020e
SM
875 struct lttng_viewer_cmd cmd;
876 struct lttng_viewer_list_sessions list;
877 struct lttng_viewer_session lsession;
878 uint32_t i, sessions_count;
879 uint64_t session_id;
880 enum lttng_live_viewer_status status;
881 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
882 bt_self_component *self_comp = viewer_connection->self_comp;
883 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
884
885 BT_COMP_LOGD("Asking the relay daemon for the list of sessions: cmd=%s",
886 lttng_viewer_command_string(LTTNG_VIEWER_LIST_SESSIONS));
887
888 cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
889 cmd.data_size = htobe64((uint64_t) 0);
890 cmd.cmd_version = htobe32(0);
891
892 status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
893 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
894 viewer_handle_send_status(self_comp, self_comp_class, status, "list sessions command");
895 goto end;
896 }
897
898 status = lttng_live_recv(viewer_connection, &list, sizeof(list));
899 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
900 viewer_handle_recv_status(self_comp, self_comp_class, status, "session list reply");
901 goto end;
902 }
903
904 sessions_count = be32toh(list.sessions_count);
905 for (i = 0; i < sessions_count; i++) {
906 status = lttng_live_recv(viewer_connection, &lsession, sizeof(lsession));
907 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
908 viewer_handle_recv_status(self_comp, self_comp_class, status, "session reply");
909 goto end;
910 }
911 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
912 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
913 session_id = be64toh(lsession.id);
914
915 BT_COMP_LOGI("Adding session to internal list: "
916 "session-id=%" PRIu64 ", hostname=\"%s\", session-name=\"%s\"",
917 session_id, lsession.hostname, lsession.session_name);
918
919 if ((strncmp(lsession.session_name, viewer_connection->session_name->str,
920 LTTNG_VIEWER_NAME_MAX) == 0) &&
921 (strncmp(lsession.hostname, viewer_connection->target_hostname->str,
922 LTTNG_VIEWER_HOST_NAME_MAX) == 0)) {
923 if (lttng_live_add_session(lttng_live_msg_iter, session_id, lsession.hostname,
924 lsession.session_name)) {
925 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to add live session");
926 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
927 goto end;
928 }
929 }
930 }
931
932 status = LTTNG_LIVE_VIEWER_STATUS_OK;
7cdc2bab 933
f79c2d7a 934end:
4164020e 935 return status;
7cdc2bab
MD
936}
937
4164020e
SM
938enum lttng_live_viewer_status
939lttng_live_create_viewer_session(struct lttng_live_msg_iter *lttng_live_msg_iter)
7cdc2bab 940{
4164020e
SM
941 struct lttng_viewer_cmd cmd;
942 struct lttng_viewer_create_session_response resp;
943 enum lttng_live_viewer_status status;
944 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
945 bt_self_component *self_comp = viewer_connection->self_comp;
946 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
947
948 BT_COMP_OR_COMP_CLASS_LOGD(self_comp, self_comp_class, "Creating a viewer session: cmd=%s",
949 lttng_viewer_command_string(LTTNG_VIEWER_CREATE_SESSION));
950
951 cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
952 cmd.data_size = htobe64((uint64_t) 0);
953 cmd.cmd_version = htobe32(0);
954
955 status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
956 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
957 viewer_handle_send_status(self_comp, self_comp_class, status, "create session command");
958 goto end;
959 }
960
961 status = lttng_live_recv(viewer_connection, &resp, sizeof(resp));
962 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
963 viewer_handle_recv_status(self_comp, self_comp_class, status, "create session reply");
964 goto end;
965 }
966
967 if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
968 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Error creating viewer session");
969 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
970 goto end;
971 }
972
973 status = lttng_live_query_session_ids(lttng_live_msg_iter);
974 if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
975 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to query live viewer session ids");
976 goto end;
977 } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
978 goto end;
979 }
7cdc2bab 980
f79c2d7a 981end:
4164020e 982 return status;
7cdc2bab
MD
983}
984
4164020e
SM
985static enum lttng_live_viewer_status receive_streams(struct lttng_live_session *session,
986 uint32_t stream_count,
987 bt_self_message_iterator *self_msg_iter)
7cdc2bab 988{
4164020e
SM
989 uint32_t i;
990 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
991 enum lttng_live_viewer_status status;
992 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
993 bt_self_component *self_comp = viewer_connection->self_comp;
994
995 BT_COMP_LOGI("Getting %" PRIu32 " new streams", stream_count);
996 for (i = 0; i < stream_count; i++) {
997 struct lttng_viewer_stream stream;
998 struct lttng_live_stream_iterator *live_stream;
999 uint64_t stream_id;
1000 uint64_t ctf_trace_id;
1001
1002 status = lttng_live_recv(viewer_connection, &stream, sizeof(stream));
1003 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1004 viewer_handle_recv_status(self_comp, NULL, status, "stream reply");
1005 goto end;
1006 }
1007 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
1008 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
1009 stream_id = be64toh(stream.id);
1010 ctf_trace_id = be64toh(stream.ctf_trace_id);
1011
1012 if (stream.metadata_flag) {
1013 BT_COMP_LOGI(" metadata stream %" PRIu64 " : %s/%s", stream_id, stream.path_name,
1014 stream.channel_name);
44bd6303 1015 if (lttng_live_metadata_create_stream(session, ctf_trace_id, stream_id)) {
4164020e
SM
1016 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Error creating metadata stream");
1017 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1018 goto end;
1019 }
1020 session->lazy_stream_msg_init = true;
1021 } else {
1022 BT_COMP_LOGI(" stream %" PRIu64 " : %s/%s", stream_id, stream.path_name,
1023 stream.channel_name);
1024 live_stream =
1025 lttng_live_stream_iterator_create(session, ctf_trace_id, stream_id, self_msg_iter);
1026 if (!live_stream) {
1027 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Error creating stream");
1028 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1029 goto end;
1030 }
1031 }
1032 }
1033 status = LTTNG_LIVE_VIEWER_STATUS_OK;
7cdc2bab 1034
f79c2d7a 1035end:
4164020e 1036 return status;
7cdc2bab
MD
1037}
1038
4164020e
SM
1039enum lttng_live_viewer_status lttng_live_session_attach(struct lttng_live_session *session,
1040 bt_self_message_iterator *self_msg_iter)
7cdc2bab 1041{
4164020e
SM
1042 struct lttng_viewer_cmd cmd;
1043 enum lttng_live_viewer_status status;
1044 struct lttng_viewer_attach_session_request rq;
1045 struct lttng_viewer_attach_session_response rp;
1046 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1047 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1048 bt_self_component *self_comp = viewer_connection->self_comp;
1049 uint64_t session_id = session->id;
1050 uint32_t streams_count;
1051 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1052 char cmd_buf[cmd_buf_len];
1053
1054 BT_COMP_LOGD("Attaching to session: cmd=%s, session-id=%" PRIu64 ", seek=%s",
1055 lttng_viewer_command_string(LTTNG_VIEWER_ATTACH_SESSION), session_id,
1056 lttng_viewer_seek_string(LTTNG_VIEWER_SEEK_LAST));
1057
1058 cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
1059 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1060 cmd.cmd_version = htobe32(0);
1061
1062 memset(&rq, 0, sizeof(rq));
1063 rq.session_id = htobe64(session_id);
1064 // TODO: add cmd line parameter to select seek beginning
1065 // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
1066 rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
1067
1068 /*
1069 * Merge the cmd and connection request to prevent a write-write
1070 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1071 * second write to be performed quickly in presence of Nagle's algorithm.
1072 */
1073 memcpy(cmd_buf, &cmd, sizeof(cmd));
1074 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1075 status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1076 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1077 viewer_handle_send_status(self_comp, NULL, status, "attach session command");
1078 goto end;
1079 }
1080
1081 status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1082 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1083 viewer_handle_recv_status(self_comp, NULL, status, "attach session reply");
1084 goto end;
1085 }
1086
1087 streams_count = be32toh(rp.streams_count);
1088 switch (be32toh(rp.status)) {
1089 case LTTNG_VIEWER_ATTACH_OK:
1090 break;
1091 case LTTNG_VIEWER_ATTACH_UNK:
1092 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Session id %" PRIu64 " is unknown", session_id);
1093 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1094 goto end;
1095 case LTTNG_VIEWER_ATTACH_ALREADY:
1096 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "There is already a viewer attached to this session");
1097 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1098 goto end;
1099 case LTTNG_VIEWER_ATTACH_NOT_LIVE:
1100 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Not a live session");
1101 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1102 goto end;
1103 case LTTNG_VIEWER_ATTACH_SEEK_ERR:
1104 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Wrong seek parameter");
1105 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1106 goto end;
1107 default:
1108 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Unknown attach return code %u", be32toh(rp.status));
1109 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1110 goto end;
1111 }
1112
1113 /* We receive the initial list of streams. */
1114 status = receive_streams(session, streams_count, self_msg_iter);
1115 switch (status) {
1116 case LTTNG_LIVE_VIEWER_STATUS_OK:
1117 break;
1118 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
1119 goto end;
1120 case LTTNG_LIVE_VIEWER_STATUS_ERROR:
1121 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Error receiving streams");
1122 goto end;
1123 default:
1124 bt_common_abort();
1125 }
1126
1127 session->attached = true;
1128 session->new_streams_needed = false;
7cdc2bab 1129
eee8e741 1130end:
4164020e 1131 return status;
7cdc2bab
MD
1132}
1133
4164020e 1134enum lttng_live_viewer_status lttng_live_session_detach(struct lttng_live_session *session)
7cdc2bab 1135{
4164020e
SM
1136 struct lttng_viewer_cmd cmd;
1137 enum lttng_live_viewer_status status;
1138 struct lttng_viewer_detach_session_request rq;
1139 struct lttng_viewer_detach_session_response rp;
1140 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1141 bt_self_component *self_comp = session->self_comp;
1142 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1143 uint64_t session_id = session->id;
1144 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1145 char cmd_buf[cmd_buf_len];
1146
1147 /*
1148 * The session might already be detached and the viewer socket might
1149 * already been closed. This happens when calling this function when
1150 * tearing down the graph after an error.
1151 */
1152 if (!session->attached || viewer_connection->control_sock == BT_INVALID_SOCKET) {
1153 return LTTNG_LIVE_VIEWER_STATUS_OK;
1154 }
1155
1156 BT_COMP_LOGD("Detaching from session: cmd=%s, session-id=%" PRIu64,
1157 lttng_viewer_command_string(LTTNG_VIEWER_DETACH_SESSION), session_id);
1158
1159 cmd.cmd = htobe32(LTTNG_VIEWER_DETACH_SESSION);
1160 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1161 cmd.cmd_version = htobe32(0);
1162
1163 memset(&rq, 0, sizeof(rq));
1164 rq.session_id = htobe64(session_id);
1165
1166 /*
1167 * Merge the cmd and connection request to prevent a write-write
1168 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1169 * second write to be performed quickly in presence of Nagle's algorithm.
1170 */
1171 memcpy(cmd_buf, &cmd, sizeof(cmd));
1172 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1173 status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1174 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1175 viewer_handle_send_status(self_comp, NULL, status, "detach session command");
1176 goto end;
1177 }
1178
1179 status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1180 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1181 viewer_handle_recv_status(self_comp, NULL, status, "detach session reply");
1182 goto end;
1183 }
1184
1185 switch (be32toh(rp.status)) {
1186 case LTTNG_VIEWER_DETACH_SESSION_OK:
1187 break;
1188 case LTTNG_VIEWER_DETACH_SESSION_UNK:
1189 BT_COMP_LOGW("Session id %" PRIu64 " is unknown", session_id);
1190 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1191 goto end;
1192 case LTTNG_VIEWER_DETACH_SESSION_ERR:
1193 BT_COMP_LOGW("Error detaching session id %" PRIu64 "", session_id);
1194 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1195 goto end;
1196 default:
1197 BT_COMP_LOGE("Unknown detach return code %u", be32toh(rp.status));
1198 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1199 goto end;
1200 }
1201
1202 session->attached = false;
1203
1204 status = LTTNG_LIVE_VIEWER_STATUS_OK;
7cdc2bab 1205
f79c2d7a 1206end:
4164020e 1207 return status;
7cdc2bab
MD
1208}
1209
4164020e
SM
1210enum lttng_live_get_one_metadata_status
1211lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, FILE *fp, size_t *reply_len)
7cdc2bab 1212{
4164020e
SM
1213 uint64_t len = 0;
1214 enum lttng_live_get_one_metadata_status status;
1215 enum lttng_live_viewer_status viewer_status;
1216 struct lttng_viewer_cmd cmd;
1217 struct lttng_viewer_get_metadata rq;
1218 struct lttng_viewer_metadata_packet rp;
1219 gchar *data = NULL;
1220 ssize_t writelen;
1221 struct lttng_live_session *session = trace->session;
1222 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1223 struct lttng_live_metadata *metadata = trace->metadata;
1224 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1225 bt_self_component *self_comp = viewer_connection->self_comp;
1226 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1227 char cmd_buf[cmd_buf_len];
1228
1229 BT_COMP_LOGD("Requesting new metadata for trace:"
1230 "cmd=%s, trace-id=%" PRIu64 ", metadata-stream-id=%" PRIu64,
1231 lttng_viewer_command_string(LTTNG_VIEWER_GET_METADATA), trace->id,
1232 metadata->stream_id);
1233
1234 rq.stream_id = htobe64(metadata->stream_id);
1235 cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
1236 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1237 cmd.cmd_version = htobe32(0);
1238
1239 /*
1240 * Merge the cmd and connection request to prevent a write-write
1241 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1242 * second write to be performed quickly in presence of Nagle's algorithm.
1243 */
1244 memcpy(cmd_buf, &cmd, sizeof(cmd));
1245 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1246 viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1247 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1248 viewer_handle_send_status(self_comp, NULL, viewer_status, "get metadata command");
1249 status = (enum lttng_live_get_one_metadata_status) viewer_status;
1250 goto end;
1251 }
1252
1253 viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1254 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1255 viewer_handle_recv_status(self_comp, NULL, viewer_status, "get metadata reply");
1256 status = (enum lttng_live_get_one_metadata_status) viewer_status;
1257 goto end;
1258 }
1259
1260 switch (be32toh(rp.status)) {
1261 case LTTNG_VIEWER_METADATA_OK:
1262 BT_COMP_LOGD("Received get_metadata response: ok");
1263 break;
1264 case LTTNG_VIEWER_NO_NEW_METADATA:
1265 BT_COMP_LOGD("Received get_metadata response: no new");
1266 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_END;
1267 goto end;
1268 case LTTNG_VIEWER_METADATA_ERR:
1269 /*
1270 * The Relayd cannot find this stream id. Maybe its
1271 * gone already. This can happen in short lived UST app
1272 * in a per-pid session.
1273 */
1274 BT_COMP_LOGD("Received get_metadata response: error");
1275 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_CLOSED;
1276 goto end;
1277 default:
1278 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Received get_metadata response: unknown");
1279 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1280 goto end;
1281 }
1282
1283 len = be64toh(rp.len);
c5ce3927
FD
1284 if (len == 0) {
1285 /*
1286 * We received a `LTTNG_VIEWER_METADATA_OK` with a packet
1287 * length of 0. This means we must try again. This scenario
1288 * arises when a clear command is performed on an lttng session.
1289 */
1290 BT_COMP_LOGD(
1291 "Expecting a metadata packet of size 0. Retry to get a packet from the relay.");
1292 goto empty_metadata_packet_retry;
1293 }
1294
4164020e
SM
1295 BT_COMP_LOGD("Writing %" PRIu64 " bytes to metadata", len);
1296 if (len <= 0) {
1297 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Erroneous response length");
1298 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1299 goto end;
1300 }
1301
1302 data = g_new0(gchar, len);
1303 if (!data) {
1304 BT_COMP_LOGE_APPEND_CAUSE_ERRNO(self_comp, "Failed to allocate data buffer", ".");
1305 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1306 goto end;
1307 }
1308
1309 viewer_status = lttng_live_recv(viewer_connection, data, len);
1310 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1311 viewer_handle_recv_status(self_comp, NULL, viewer_status, "get metadata packet");
1312 status = (enum lttng_live_get_one_metadata_status) viewer_status;
1313 goto end;
1314 }
1315
1316 /*
1317 * Write the metadata to the file handle.
1318 */
1319 writelen = fwrite(data, sizeof(uint8_t), len, fp);
1320 if (writelen != len) {
1321 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Writing in the metadata file stream");
1322 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1323 goto end;
1324 }
1325
c5ce3927 1326empty_metadata_packet_retry:
4164020e
SM
1327 *reply_len = len;
1328 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_OK;
c28512ab 1329
c28512ab 1330end:
4164020e
SM
1331 g_free(data);
1332 return status;
7cdc2bab
MD
1333}
1334
1335/*
1336 * Assign the fields from a lttng_viewer_index to a packet_index.
1337 */
4164020e
SM
1338static void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
1339 struct packet_index *pindex)
7cdc2bab 1340{
4164020e
SM
1341 BT_ASSERT(lindex);
1342 BT_ASSERT(pindex);
1343
1344 pindex->offset = be64toh(lindex->offset);
1345 pindex->packet_size = be64toh(lindex->packet_size);
1346 pindex->content_size = be64toh(lindex->content_size);
1347 pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
1348 pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
1349 pindex->events_discarded = be64toh(lindex->events_discarded);
7cdc2bab
MD
1350}
1351
4164020e 1352static void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter)
36e94ad6 1353{
4164020e
SM
1354 uint64_t session_idx;
1355 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1356
1357 for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
1358 struct lttng_live_session *session =
1359 (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
1360 BT_COMP_LOGD("Marking session as needing new streams: "
1361 "session-id=%" PRIu64,
1362 session->id);
1363 session->new_streams_needed = true;
1364 }
36e94ad6
FD
1365}
1366
4164020e
SM
1367enum lttng_live_iterator_status
1368lttng_live_get_next_index(struct lttng_live_msg_iter *lttng_live_msg_iter,
1369 struct lttng_live_stream_iterator *stream, struct packet_index *index)
7cdc2bab 1370{
4164020e
SM
1371 struct lttng_viewer_cmd cmd;
1372 struct lttng_viewer_get_next_index rq;
1373 enum lttng_live_viewer_status viewer_status;
1374 struct lttng_viewer_index rp;
1375 enum lttng_live_iterator_status status;
1376 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1377 bt_self_component *self_comp = viewer_connection->self_comp;
1378 struct lttng_live_trace *trace = stream->trace;
1379 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1380 char cmd_buf[cmd_buf_len];
1381 uint32_t flags, rp_status;
1382
1383 BT_COMP_LOGD("Requesting next index for stream: cmd=%s, "
1384 "viewer-stream-id=%" PRIu64,
1385 lttng_viewer_command_string(LTTNG_VIEWER_GET_NEXT_INDEX),
1386 stream->viewer_stream_id);
1387 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1388 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1389 cmd.cmd_version = htobe32(0);
1390
1391 memset(&rq, 0, sizeof(rq));
1392 rq.stream_id = htobe64(stream->viewer_stream_id);
1393
1394 /*
1395 * Merge the cmd and connection request to prevent a write-write
1396 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1397 * second write to be performed quickly in presence of Nagle's algorithm.
1398 */
1399 memcpy(cmd_buf, &cmd, sizeof(cmd));
1400 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1401
1402 viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1403 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1404 viewer_handle_send_status(self_comp, NULL, viewer_status, "get next index command");
1405 goto error;
1406 }
1407
1408 viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1409 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1410 viewer_handle_recv_status(self_comp, NULL, viewer_status, "get next index reply");
1411 goto error;
1412 }
1413
1414 flags = be32toh(rp.flags);
1415 rp_status = be32toh(rp.status);
1416
1417 BT_COMP_LOGD("Received response from relay daemon: cmd=%s, response=%s",
1418 lttng_viewer_command_string(LTTNG_VIEWER_GET_NEXT_INDEX),
1419 lttng_viewer_next_index_return_code_string(rp_status));
1420 switch (rp_status) {
1421 case LTTNG_VIEWER_INDEX_INACTIVE:
1422 {
1423 uint64_t ctf_stream_class_id;
1424
1425 memset(index, 0, sizeof(struct packet_index));
1426 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
1427 stream->current_inactivity_ts = index->ts_cycles.timestamp_end;
1428 ctf_stream_class_id = be64toh(rp.stream_id);
1429 if (stream->ctf_stream_class_id.is_set) {
1430 BT_ASSERT(stream->ctf_stream_class_id.value == ctf_stream_class_id);
1431 } else {
1432 stream->ctf_stream_class_id.value = ctf_stream_class_id;
1433 stream->ctf_stream_class_id.is_set = true;
1434 }
1435 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_QUIESCENT);
1436 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1437 break;
1438 }
1439 case LTTNG_VIEWER_INDEX_OK:
1440 {
1441 uint64_t ctf_stream_class_id;
1442
1443 lttng_index_to_packet_index(&rp, index);
1444 ctf_stream_class_id = be64toh(rp.stream_id);
1445 if (stream->ctf_stream_class_id.is_set) {
1446 BT_ASSERT(stream->ctf_stream_class_id.value == ctf_stream_class_id);
1447 } else {
1448 stream->ctf_stream_class_id.value = ctf_stream_class_id;
1449 stream->ctf_stream_class_id.is_set = true;
1450 }
1451
1452 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_DATA);
1453
1454 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1455 BT_COMP_LOGD("Marking trace as needing new metadata: "
1456 "response=%s, response-flag=NEW_METADATA, trace-id=%" PRIu64,
1457 lttng_viewer_next_index_return_code_string(rp_status), trace->id);
1458 trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
1459 }
1460 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1461 BT_COMP_LOGD("Marking all sessions as possibly needing new streams: "
1462 "response=%s, response-flag=NEW_STREAM",
1463 lttng_viewer_next_index_return_code_string(rp_status));
1464 lttng_live_need_new_streams(lttng_live_msg_iter);
1465 }
1466 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1467 break;
1468 }
1469 case LTTNG_VIEWER_INDEX_RETRY:
1470 memset(index, 0, sizeof(struct packet_index));
1471 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1472 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1473 goto end;
1474 case LTTNG_VIEWER_INDEX_HUP:
1475 memset(index, 0, sizeof(struct packet_index));
1476 index->offset = EOF;
1477 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_EOF);
1478 stream->has_stream_hung_up = true;
1479 status = LTTNG_LIVE_ITERATOR_STATUS_END;
1480 break;
1481 case LTTNG_VIEWER_INDEX_ERR:
1482 memset(index, 0, sizeof(struct packet_index));
1483 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1484 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1485 goto end;
1486 default:
1487 BT_COMP_LOGD("Received get_next_index response: unknown value");
1488 memset(index, 0, sizeof(struct packet_index));
1489 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1490 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1491 goto end;
1492 }
1493 goto end;
7cdc2bab
MD
1494
1495error:
4164020e 1496 status = viewer_status_to_live_iterator_status(viewer_status);
f79c2d7a 1497end:
4164020e 1498 return status;
7cdc2bab
MD
1499}
1500
4164020e
SM
1501enum ctf_msg_iter_medium_status
1502lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
1503 struct lttng_live_stream_iterator *stream, uint8_t *buf,
1504 uint64_t offset, uint64_t req_len, uint64_t *recv_len)
7cdc2bab 1505{
4164020e
SM
1506 enum ctf_msg_iter_medium_status status;
1507 enum lttng_live_viewer_status viewer_status;
1508 struct lttng_viewer_trace_packet rp;
1509 struct lttng_viewer_cmd cmd;
1510 struct lttng_viewer_get_packet rq;
1511 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1512 bt_self_component *self_comp = viewer_connection->self_comp;
1513 struct lttng_live_trace *trace = stream->trace;
1514 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1515 char cmd_buf[cmd_buf_len];
1516 uint32_t flags, rp_status;
1517
1518 BT_COMP_LOGD("Requesting data from stream: cmd=%s, "
1519 "offset=%" PRIu64 ", request-len=%" PRIu64,
1520 lttng_viewer_command_string(LTTNG_VIEWER_GET_PACKET), offset, req_len);
1521
1522 cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
1523 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1524 cmd.cmd_version = htobe32(0);
1525
1526 memset(&rq, 0, sizeof(rq));
1527 rq.stream_id = htobe64(stream->viewer_stream_id);
1528 rq.offset = htobe64(offset);
1529 rq.len = htobe32(req_len);
1530
1531 /*
1532 * Merge the cmd and connection request to prevent a write-write
1533 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1534 * second write to be performed quickly in presence of Nagle's algorithm.
1535 */
1536 memcpy(cmd_buf, &cmd, sizeof(cmd));
1537 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1538
1539 viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1540 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1541 viewer_handle_send_status(self_comp, NULL, viewer_status, "get data packet command");
1542 goto error_convert_status;
1543 }
1544
1545 viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1546 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1547 viewer_handle_recv_status(self_comp, NULL, viewer_status, "get data packet reply");
1548 goto error_convert_status;
1549 }
1550
1551 flags = be32toh(rp.flags);
1552 rp_status = be32toh(rp.status);
1553
1554 BT_COMP_LOGD("Received response from relay daemon: cmd=%s, response=%s",
1555 lttng_viewer_command_string(LTTNG_VIEWER_GET_PACKET),
1556 lttng_viewer_get_packet_return_code_string(rp_status));
1557 switch (rp_status) {
1558 case LTTNG_VIEWER_GET_PACKET_OK:
1559 req_len = be32toh(rp.len);
1560 BT_COMP_LOGD("Got packet from relay daemon: response=%s, packet-len=%" PRIu64 "",
1561 lttng_viewer_get_packet_return_code_string(rp_status), req_len);
4164020e
SM
1562 break;
1563 case LTTNG_VIEWER_GET_PACKET_RETRY:
1564 /* Unimplemented by relay daemon */
1565 status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
1566 goto end;
1567 case LTTNG_VIEWER_GET_PACKET_ERR:
1568 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1569 BT_COMP_LOGD("Marking trace as needing new metadata: "
1570 "response=%s, response-flag=NEW_METADATA, trace-id=%" PRIu64,
1571 lttng_viewer_next_index_return_code_string(rp_status), trace->id);
1572 trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
1573 }
1574 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1575 BT_COMP_LOGD("Marking all sessions as possibly needing new streams: "
1576 "response=%s, response-flag=NEW_STREAM",
1577 lttng_viewer_next_index_return_code_string(rp_status));
1578 lttng_live_need_new_streams(lttng_live_msg_iter);
1579 }
1580 if (flags & (LTTNG_VIEWER_FLAG_NEW_METADATA | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
1581 status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
1582 BT_COMP_LOGD("Reply with any one flags set means we should retry: response=%s",
1583 lttng_viewer_get_packet_return_code_string(rp_status));
1584 goto end;
1585 }
1586 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Received get_data_packet response: error");
1587 status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1588 goto end;
1589 case LTTNG_VIEWER_GET_PACKET_EOF:
1590 status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
1591 goto end;
1592 default:
1593 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Received get_data_packet response: unknown (%d)",
1594 rp_status);
1595 status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1596 goto end;
1597 }
1598
1599 if (req_len == 0) {
1600 status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1601 goto end;
1602 }
1603
1604 viewer_status = lttng_live_recv(viewer_connection, buf, req_len);
1605 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1606 viewer_handle_recv_status(self_comp, NULL, viewer_status, "get data packet");
1607 goto error_convert_status;
1608 }
1609 *recv_len = req_len;
1610
1611 status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
1612 goto end;
f79c2d7a 1613
535fb48a 1614error_convert_status:
4164020e 1615 status = viewer_status_to_ctf_msg_iter_medium_status(viewer_status);
f79c2d7a 1616end:
4164020e 1617 return status;
7cdc2bab
MD
1618}
1619
1620/*
1621 * Request new streams for a session.
1622 */
4164020e
SM
1623enum lttng_live_iterator_status
1624lttng_live_session_get_new_streams(struct lttng_live_session *session,
1625 bt_self_message_iterator *self_msg_iter)
7cdc2bab 1626{
4164020e
SM
1627 enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1628 struct lttng_viewer_cmd cmd;
1629 struct lttng_viewer_new_streams_request rq;
1630 struct lttng_viewer_new_streams_response rp;
1631 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1632 enum lttng_live_viewer_status viewer_status;
1633 struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1634 bt_self_component *self_comp = viewer_connection->self_comp;
1635 uint32_t streams_count;
1636 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1637 char cmd_buf[cmd_buf_len];
1638
1639 if (!session->new_streams_needed) {
1640 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1641 goto end;
1642 }
1643
1644 BT_COMP_LOGD("Requesting new streams for session: cmd=%s, "
1645 "session-id=%" PRIu64,
1646 lttng_viewer_command_string(LTTNG_VIEWER_GET_NEW_STREAMS), session->id);
1647
1648 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1649 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1650 cmd.cmd_version = htobe32(0);
1651
1652 memset(&rq, 0, sizeof(rq));
1653 rq.session_id = htobe64(session->id);
1654
1655 /*
1656 * Merge the cmd and connection request to prevent a write-write
1657 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1658 * second write to be performed quickly in presence of Nagle's algorithm.
1659 */
1660 memcpy(cmd_buf, &cmd, sizeof(cmd));
1661 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1662
1663 viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1664 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1665 viewer_handle_send_status(self_comp, NULL, viewer_status, "get new streams command");
1666 status = viewer_status_to_live_iterator_status(viewer_status);
1667 goto end;
1668 }
1669
1670 viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1671 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1672 viewer_handle_recv_status(self_comp, NULL, viewer_status, "get new streams reply");
1673 status = viewer_status_to_live_iterator_status(viewer_status);
1674 goto end;
1675 }
1676
1677 streams_count = be32toh(rp.streams_count);
1678
1679 switch (be32toh(rp.status)) {
1680 case LTTNG_VIEWER_NEW_STREAMS_OK:
1681 session->new_streams_needed = false;
1682 break;
1683 case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1684 session->new_streams_needed = false;
1685 goto end;
1686 case LTTNG_VIEWER_NEW_STREAMS_HUP:
1687 session->new_streams_needed = false;
1688 session->closed = true;
1689 status = LTTNG_LIVE_ITERATOR_STATUS_END;
1690 goto end;
1691 case LTTNG_VIEWER_NEW_STREAMS_ERR:
1692 BT_COMP_LOGD("Received get_new_streams response: error");
1693 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1694 goto end;
1695 default:
1696 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1697 "Received get_new_streams response: Unknown:"
1698 "return code %u",
1699 be32toh(rp.status));
1700 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1701 goto end;
1702 }
1703
1704 viewer_status = receive_streams(session, streams_count, self_msg_iter);
1705 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1706 viewer_handle_recv_status(self_comp, NULL, viewer_status, "new streams");
1707 status = viewer_status_to_live_iterator_status(viewer_status);
1708 goto end;
1709 }
1710
1711 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
f79c2d7a 1712end:
4164020e 1713 return status;
7cdc2bab
MD
1714}
1715
f79c2d7a 1716enum lttng_live_viewer_status live_viewer_connection_create(
4164020e
SM
1717 bt_self_component *self_comp, bt_self_component_class *self_comp_class,
1718 bt_logging_level log_level, const char *url, bool in_query,
1719 struct lttng_live_msg_iter *lttng_live_msg_iter, struct live_viewer_connection **viewer)
7cdc2bab 1720{
4164020e
SM
1721 struct live_viewer_connection *viewer_connection;
1722 enum lttng_live_viewer_status status;
1723
1724 viewer_connection = g_new0(struct live_viewer_connection, 1);
1725
1726 if (bt_socket_init(log_level) != 0) {
1727 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
1728 "Failed to init socket");
1729 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1730 goto error;
1731 }
1732
1733 viewer_connection->log_level = log_level;
1734
1735 viewer_connection->self_comp = self_comp;
1736 viewer_connection->self_comp_class = self_comp_class;
1737
1738 viewer_connection->control_sock = BT_INVALID_SOCKET;
1739 viewer_connection->port = -1;
1740 viewer_connection->in_query = in_query;
1741 viewer_connection->lttng_live_msg_iter = lttng_live_msg_iter;
1742 viewer_connection->url = g_string_new(url);
1743 if (!viewer_connection->url) {
1744 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
1745 "Failed to allocate URL buffer");
1746 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1747 goto error;
1748 }
1749
1750 BT_COMP_OR_COMP_CLASS_LOGD(self_comp, self_comp_class,
1751 "Establishing connection to url \"%s\"...", url);
1752 status = lttng_live_connect_viewer(viewer_connection);
1753 /*
1754 * Only print error and append cause in case of error. not in case of
1755 * interruption.
1756 */
1757 if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1758 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
1759 "Failed to establish connection: "
1760 "url=\"%s\"",
1761 url);
1762 goto error;
1763 } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1764 goto error;
1765 }
1766 BT_COMP_OR_COMP_CLASS_LOGD(self_comp, self_comp_class,
1767 "Connection to url \"%s\" is established", url);
1768
1769 *viewer = viewer_connection;
1770 status = LTTNG_LIVE_VIEWER_STATUS_OK;
1771 goto end;
7cdc2bab 1772
7cdc2bab 1773error:
4164020e
SM
1774 if (viewer_connection) {
1775 live_viewer_connection_destroy(viewer_connection);
1776 }
f79c2d7a 1777end:
4164020e 1778 return status;
7cdc2bab
MD
1779}
1780
4164020e 1781void live_viewer_connection_destroy(struct live_viewer_connection *viewer_connection)
7cdc2bab 1782{
4164020e
SM
1783 bt_self_component *self_comp = viewer_connection->self_comp;
1784 bt_self_component_class *self_comp_class = viewer_connection->self_comp_class;
b9e6ec43 1785
4164020e
SM
1786 if (!viewer_connection) {
1787 goto end;
1788 }
b9e6ec43 1789
4164020e
SM
1790 BT_COMP_OR_COMP_CLASS_LOGD(self_comp, self_comp_class,
1791 "Closing connection to relay:"
1792 "relay-url=\"%s\"",
1793 viewer_connection->url->str);
ecb4ba8a 1794
4164020e 1795 lttng_live_disconnect_viewer(viewer_connection);
b9e6ec43 1796
4164020e
SM
1797 if (viewer_connection->url) {
1798 g_string_free(viewer_connection->url, true);
1799 }
b9e6ec43 1800
4164020e
SM
1801 if (viewer_connection->relay_hostname) {
1802 g_string_free(viewer_connection->relay_hostname, true);
1803 }
b9e6ec43 1804
4164020e
SM
1805 if (viewer_connection->target_hostname) {
1806 g_string_free(viewer_connection->target_hostname, true);
1807 }
b9e6ec43 1808
4164020e
SM
1809 if (viewer_connection->session_name) {
1810 g_string_free(viewer_connection->session_name, true);
1811 }
b9e6ec43 1812
4164020e
SM
1813 if (viewer_connection->proto) {
1814 g_string_free(viewer_connection->proto, true);
1815 }
ecb4ba8a 1816
4164020e 1817 g_free(viewer_connection);
1cb3cdd7 1818
4164020e 1819 bt_socket_fini();
b9e6ec43
FD
1820
1821end:
4164020e 1822 return;
7cdc2bab 1823}
This page took 0.187007 seconds and 4 git commands to generate.