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