projects / babeltrace.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
blame | history | raw | HEAD
Cleanup: src.ctf.lttng-live: coding style
[babeltrace.git] / src / plugins / ctf / lttng-live / viewer-connection.c
1 /*
2 * Copyright 2019 - Francis Deslauriers <francis.deslauriers@efficios.com>
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
24 #define BT_COMP_LOG_SELF_COMP (viewer_connection->self_comp)
25 #define BT_LOG_OUTPUT_LEVEL (viewer_connection->log_level)
26 #define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE/VIEWER"
27 #include "logging/comp-logging.h"
28
29 #include <stdio.h>
30 #include <stdint.h>
31 #include <stdlib.h>
32 #include <stdbool.h>
33 #include <unistd.h>
34 #include <glib.h>
35 #include <inttypes.h>
36 #include <sys/types.h>
37 #include <fcntl.h>
38
39 #include "compat/socket.h"
40 #include "compat/endian.h"
41 #include "compat/compiler.h"
42 #include "common/common.h"
43 #include <babeltrace2/babeltrace.h>
44
45 #include "lttng-live.h"
46 #include "viewer-connection.h"
47 #include "lttng-viewer-abi.h"
48 #include "data-stream.h"
49 #include "metadata.h"
50
51 static
52 ssize_t lttng_live_recv(struct live_viewer_connection *viewer_connection,
53 void *buf, size_t len)
54 {
55 ssize_t ret;
56 size_t copied = 0, to_copy = len;
57 struct lttng_live_msg_iter *lttng_live_msg_iter =
58 viewer_connection->lttng_live_msg_iter;
59 BT_SOCKET sock = viewer_connection->control_sock;
60
61 do {
62 ret = bt_socket_recv(sock, buf + copied, to_copy, 0);
63 if (ret > 0) {
64 BT_ASSERT(ret <= to_copy);
65 copied += ret;
66 to_copy -= ret;
67 }
68 if (ret == BT_SOCKET_ERROR && bt_socket_interrupted()) {
69 if (!viewer_connection->in_query &&
70 lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
71 break;
72 } else {
73 continue;
74 }
75 }
76 } while (ret > 0 && to_copy > 0);
77 if (ret > 0)
78 ret = copied;
79 /* ret = 0 means orderly shutdown, ret == BT_SOCKET_ERROR is error. */
80 return ret;
81 }
82
83 static
84 ssize_t lttng_live_send(struct live_viewer_connection *viewer_connection,
85 const void *buf, size_t len)
86 {
87 struct lttng_live_msg_iter *lttng_live_msg_iter =
88 viewer_connection->lttng_live_msg_iter;
89 BT_SOCKET sock = viewer_connection->control_sock;
90 ssize_t ret;
91
92 for (;;) {
93 ret = bt_socket_send_nosigpipe(sock, buf, len);
94 if (ret == BT_SOCKET_ERROR && bt_socket_interrupted()) {
95 if (!viewer_connection->in_query &&
96 lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
97 break;
98 } else {
99 continue;
100 }
101 } else {
102 break;
103 }
104 }
105 return ret;
106 }
107
108 static
109 int parse_url(struct live_viewer_connection *viewer_connection)
110 {
111 char error_buf[256] = { 0 };
112 struct bt_common_lttng_live_url_parts lttng_live_url_parts = { 0 };
113 int ret = -1;
114 const char *path = viewer_connection->url->str;
115
116 if (!path) {
117 goto end;
118 }
119
120 lttng_live_url_parts = bt_common_parse_lttng_live_url(path, error_buf,
121 sizeof(error_buf));
122 if (!lttng_live_url_parts.proto) {
123 BT_COMP_LOGW("Invalid LTTng live URL format: %s", error_buf);
124 goto end;
125 }
126
127 viewer_connection->relay_hostname = lttng_live_url_parts.hostname;
128 lttng_live_url_parts.hostname = NULL;
129
130 if (lttng_live_url_parts.port >= 0) {
131 viewer_connection->port = lttng_live_url_parts.port;
132 } else {
133 viewer_connection->port = LTTNG_DEFAULT_NETWORK_VIEWER_PORT;
134 }
135
136 viewer_connection->target_hostname = lttng_live_url_parts.target_hostname;
137 lttng_live_url_parts.target_hostname = NULL;
138
139 if (lttng_live_url_parts.session_name) {
140 viewer_connection->session_name = lttng_live_url_parts.session_name;
141 lttng_live_url_parts.session_name = NULL;
142 }
143
144 BT_COMP_LOGI("Connecting to hostname : %s, port : %d, "
145 "target hostname : %s, session name : %s, proto : %s",
146 viewer_connection->relay_hostname->str,
147 viewer_connection->port,
148 !viewer_connection->target_hostname ?
149 "<none>" : viewer_connection->target_hostname->str,
150 !viewer_connection->session_name ?
151 "<none>" : viewer_connection->session_name->str,
152 lttng_live_url_parts.proto->str);
153 ret = 0;
154
155 end:
156 bt_common_destroy_lttng_live_url_parts(&lttng_live_url_parts);
157 return ret;
158 }
159
160 static
161 int lttng_live_handshake(struct live_viewer_connection *viewer_connection)
162 {
163 struct lttng_viewer_cmd cmd;
164 struct lttng_viewer_connect connect;
165 const size_t cmd_buf_len = sizeof(cmd) + sizeof(connect);
166 char cmd_buf[cmd_buf_len];
167 int ret;
168 ssize_t ret_len;
169
170 cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
171 cmd.data_size = htobe64((uint64_t) sizeof(connect));
172 cmd.cmd_version = htobe32(0);
173
174 connect.viewer_session_id = -1ULL; /* will be set on recv */
175 connect.major = htobe32(LTTNG_LIVE_MAJOR);
176 connect.minor = htobe32(LTTNG_LIVE_MINOR);
177 connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
178
179 /*
180 * Merge the cmd and connection request to prevent a write-write
181 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
182 * second write to be performed quickly in presence of Nagle's algorithm
183 */
184 memcpy(cmd_buf, &cmd, sizeof(cmd));
185 memcpy(cmd_buf + sizeof(cmd), &connect, sizeof(connect));
186 ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
187 if (ret_len == BT_SOCKET_ERROR) {
188 BT_COMP_LOGE("Error sending version: %s", bt_socket_errormsg());
189 goto error;
190 }
191
192 BT_ASSERT(ret_len == cmd_buf_len);
193
194 ret_len = lttng_live_recv(viewer_connection, &connect, sizeof(connect));
195 if (ret_len == 0) {
196 BT_COMP_LOGI("Remote side has closed connection");
197 goto error;
198 }
199 if (ret_len == BT_SOCKET_ERROR) {
200 BT_COMP_LOGE("Error receiving version: %s", bt_socket_errormsg());
201 goto error;
202 }
203 BT_ASSERT(ret_len == sizeof(connect));
204
205 BT_COMP_LOGI("Received viewer session ID : %" PRIu64,
206 (uint64_t) be64toh(connect.viewer_session_id));
207 BT_COMP_LOGI("Relayd version : %u.%u", be32toh(connect.major),
208 be32toh(connect.minor));
209
210 if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) {
211 BT_COMP_LOGE("Incompatible lttng-relayd protocol");
212 goto error;
213 }
214 /* Use the smallest protocol version implemented. */
215 if (LTTNG_LIVE_MINOR > be32toh(connect.minor)) {
216 viewer_connection->minor = be32toh(connect.minor);
217 } else {
218 viewer_connection->minor = LTTNG_LIVE_MINOR;
219 }
220 viewer_connection->major = LTTNG_LIVE_MAJOR;
221 ret = 0;
222 return ret;
223
224 error:
225 BT_COMP_LOGE("Unable to establish connection");
226 return -1;
227 }
228
229 static
230 int lttng_live_connect_viewer(struct live_viewer_connection *viewer_connection)
231 {
232 struct hostent *host;
233 struct sockaddr_in server_addr;
234 int ret;
235
236 if (parse_url(viewer_connection)) {
237 goto error;
238 }
239
240 host = gethostbyname(viewer_connection->relay_hostname->str);
241 if (!host) {
242 BT_COMP_LOGE("Cannot lookup hostname %s",
243 viewer_connection->relay_hostname->str);
244 goto error;
245 }
246
247 if ((viewer_connection->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == BT_INVALID_SOCKET) {
248 BT_COMP_LOGE("Socket creation failed: %s", bt_socket_errormsg());
249 goto error;
250 }
251
252 server_addr.sin_family = AF_INET;
253 server_addr.sin_port = htons(viewer_connection->port);
254 server_addr.sin_addr = *((struct in_addr *) host->h_addr);
255 memset(&(server_addr.sin_zero), 0, 8);
256
257 if (connect(viewer_connection->control_sock, (struct sockaddr *) &server_addr,
258 sizeof(struct sockaddr)) == BT_SOCKET_ERROR) {
259 BT_COMP_LOGE("Connection failed: %s", bt_socket_errormsg());
260 goto error;
261 }
262 if (lttng_live_handshake(viewer_connection)) {
263 goto error;
264 }
265
266 ret = 0;
267
268 return ret;
269
270 error:
271 if (viewer_connection->control_sock != BT_INVALID_SOCKET) {
272 if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
273 BT_COMP_LOGE("Close: %s", bt_socket_errormsg());
274 }
275 }
276 viewer_connection->control_sock = BT_INVALID_SOCKET;
277 return -1;
278 }
279
280 static
281 void lttng_live_disconnect_viewer(
282 struct live_viewer_connection *viewer_connection)
283 {
284 if (viewer_connection->control_sock == BT_INVALID_SOCKET) {
285 return;
286 }
287 if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
288 BT_COMP_LOGE("Close: %s", bt_socket_errormsg());
289 viewer_connection->control_sock = BT_INVALID_SOCKET;
290 }
291 }
292
293 static
294 void connection_release(bt_object *obj)
295 {
296 struct live_viewer_connection *conn =
297 container_of(obj, struct live_viewer_connection, obj);
298
299 live_viewer_connection_destroy(conn);
300 }
301
302 static
303 int list_update_session(bt_value *results,
304 const struct lttng_viewer_session *session,
305 bool *_found, struct live_viewer_connection *viewer_connection)
306 {
307 int i, len, ret = 0;
308 bt_value *map = NULL;
309 bt_value *hostname = NULL;
310 bt_value *session_name = NULL;
311 bt_value *btval = NULL;
312 bool found = false;
313
314 len = bt_value_array_get_length(results);
315 if (len < 0) {
316 BT_COMP_LOGE_STR("Error getting size of array.");
317 ret = -1;
318 goto end;
319 }
320 for (i = 0; i < len; i++) {
321 const char *hostname_str = NULL;
322 const char *session_name_str = NULL;
323
324 map = bt_value_array_borrow_element_by_index(results, (size_t) i);
325 if (!map) {
326 BT_COMP_LOGE_STR("Error borrowing map.");
327 ret = -1;
328 goto end;
329 }
330 hostname = bt_value_map_borrow_entry_value(map, "target-hostname");
331 if (!hostname) {
332 BT_COMP_LOGE_STR("Error borrowing \"target-hostname\" entry.");
333 ret = -1;
334 goto end;
335 }
336 session_name = bt_value_map_borrow_entry_value(map, "session-name");
337 if (!session_name) {
338 BT_COMP_LOGE_STR("Error borrowing \"session-name\" entry.");
339 ret = -1;
340 goto end;
341 }
342 hostname_str = bt_value_string_get(hostname);
343 session_name_str = bt_value_string_get(session_name);
344
345 if (strcmp(session->hostname, hostname_str) == 0
346 && strcmp(session->session_name, session_name_str) == 0) {
347 int64_t val;
348 uint32_t streams = be32toh(session->streams);
349 uint32_t clients = be32toh(session->clients);
350
351 found = true;
352
353 btval = bt_value_map_borrow_entry_value(map, "stream-count");
354 if (!btval) {
355 BT_COMP_LOGE_STR("Error borrowing \"stream-count\" entry.");
356 ret = -1;
357 goto end;
358 }
359 val = bt_value_integer_unsigned_get(btval);
360 /* sum */
361 val += streams;
362 bt_value_integer_unsigned_set(btval, val);
363
364 btval = bt_value_map_borrow_entry_value(map, "client-count");
365 if (!btval) {
366 BT_COMP_LOGE_STR("Error borrowing \"client-count\" entry.");
367 ret = -1;
368 goto end;
369 }
370 val = bt_value_integer_unsigned_get(btval);
371 /* max */
372 val = bt_max_t(int64_t, clients, val);
373 bt_value_integer_unsigned_set(btval, val);
374 }
375
376 if (found) {
377 break;
378 }
379 }
380 end:
381 *_found = found;
382 return ret;
383 }
384
385 static
386 int list_append_session(bt_value *results,
387 GString *base_url,
388 const struct lttng_viewer_session *session,
389 struct live_viewer_connection *viewer_connection)
390 {
391 int ret = 0;
392 bt_value_map_insert_entry_status insert_status;
393 bt_value_array_append_element_status append_status;
394 bt_value *map = NULL;
395 GString *url = NULL;
396 bool found = false;
397
398 /*
399 * If the session already exists, add the stream count to it,
400 * and do max of client counts.
401 */
402 ret = list_update_session(results, session, &found, viewer_connection);
403 if (ret || found) {
404 goto end;
405 }
406
407 map = bt_value_map_create();
408 if (!map) {
409 BT_COMP_LOGE_STR("Error creating map value.");
410 ret = -1;
411 goto end;
412 }
413
414 if (base_url->len < 1) {
415 BT_COMP_LOGE_STR("Error: base_url length smaller than 1.");
416 ret = -1;
417 goto end;
418 }
419 /*
420 * key = "url",
421 * value = <string>,
422 */
423 url = g_string_new(base_url->str);
424 g_string_append(url, "/host/");
425 g_string_append(url, session->hostname);
426 g_string_append_c(url, '/');
427 g_string_append(url, session->session_name);
428
429 insert_status = bt_value_map_insert_string_entry(map, "url", url->str);
430 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
431 BT_COMP_LOGE_STR("Error inserting \"url\" entry.");
432 ret = -1;
433 goto end;
434 }
435
436 /*
437 * key = "target-hostname",
438 * value = <string>,
439 */
440 insert_status = bt_value_map_insert_string_entry(map, "target-hostname",
441 session->hostname);
442 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
443 BT_COMP_LOGE_STR("Error inserting \"target-hostname\" entry.");
444 ret = -1;
445 goto end;
446 }
447
448 /*
449 * key = "session-name",
450 * value = <string>,
451 */
452 insert_status = bt_value_map_insert_string_entry(map, "session-name",
453 session->session_name);
454 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
455 BT_COMP_LOGE_STR("Error inserting \"session-name\" entry.");
456 ret = -1;
457 goto end;
458 }
459
460 /*
461 * key = "timer-us",
462 * value = <integer>,
463 */
464 {
465 uint32_t live_timer = be32toh(session->live_timer);
466
467 insert_status = bt_value_map_insert_unsigned_integer_entry(
468 map, "timer-us", live_timer);
469 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
470 BT_COMP_LOGE_STR("Error inserting \"timer-us\" entry.");
471 ret = -1;
472 goto end;
473 }
474 }
475
476 /*
477 * key = "stream-count",
478 * value = <integer>,
479 */
480 {
481 uint32_t streams = be32toh(session->streams);
482
483 insert_status = bt_value_map_insert_unsigned_integer_entry(map,
484 "stream-count", streams);
485 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
486 BT_COMP_LOGE_STR("Error inserting \"stream-count\" entry.");
487 ret = -1;
488 goto end;
489 }
490 }
491
492 /*
493 * key = "client-count",
494 * value = <integer>,
495 */
496 {
497 uint32_t clients = be32toh(session->clients);
498
499 insert_status = bt_value_map_insert_unsigned_integer_entry(map,
500 "client-count", clients);
501 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
502 BT_COMP_LOGE_STR("Error inserting \"client-count\" entry.");
503 ret = -1;
504 goto end;
505 }
506 }
507
508 append_status = bt_value_array_append_element(results, map);
509 if (append_status != BT_VALUE_ARRAY_APPEND_ELEMENT_STATUS_OK) {
510 BT_COMP_LOGE_STR("Error appending map to results.");
511 ret = -1;
512 }
513
514 end:
515 if (url) {
516 g_string_free(url, true);
517 }
518 BT_VALUE_PUT_REF_AND_RESET(map);
519 return ret;
520 }
521
522 /*
523 * Data structure returned:
524 *
525 * {
526 * <array> = {
527 * [n] = {
528 * <map> = {
529 * {
530 * key = "url",
531 * value = <string>,
532 * },
533 * {
534 * key = "target-hostname",
535 * value = <string>,
536 * },
537 * {
538 * key = "session-name",
539 * value = <string>,
540 * },
541 * {
542 * key = "timer-us",
543 * value = <integer>,
544 * },
545 * {
546 * key = "stream-count",
547 * value = <integer>,
548 * },
549 * {
550 * key = "client-count",
551 * value = <integer>,
552 * },
553 * },
554 * }
555 * }
556 */
557
558 BT_HIDDEN
559 bt_component_class_query_method_status live_viewer_connection_list_sessions(
560 struct live_viewer_connection *viewer_connection,
561 const bt_value **user_result)
562 {
563 bt_component_class_query_method_status status =
564 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
565 bt_value *result = NULL;
566 struct lttng_viewer_cmd cmd;
567 struct lttng_viewer_list_sessions list;
568 uint32_t i, sessions_count;
569 ssize_t ret_len;
570
571 result = bt_value_array_create();
572 if (!result) {
573 BT_COMP_LOGE("Error creating array");
574 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
575 goto error;
576 }
577
578 cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
579 cmd.data_size = htobe64((uint64_t) 0);
580 cmd.cmd_version = htobe32(0);
581
582 ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
583 if (ret_len == BT_SOCKET_ERROR) {
584 BT_COMP_LOGE("Error sending cmd: %s", bt_socket_errormsg());
585 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
586 goto error;
587 }
588 BT_ASSERT(ret_len == sizeof(cmd));
589
590 ret_len = lttng_live_recv(viewer_connection, &list, sizeof(list));
591 if (ret_len == 0) {
592 BT_COMP_LOGI("Remote side has closed connection");
593 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
594 goto error;
595 }
596 if (ret_len == BT_SOCKET_ERROR) {
597 BT_COMP_LOGE("Error receiving session list: %s", bt_socket_errormsg());
598 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
599 goto error;
600 }
601 BT_ASSERT(ret_len == sizeof(list));
602
603 sessions_count = be32toh(list.sessions_count);
604 for (i = 0; i < sessions_count; i++) {
605 struct lttng_viewer_session lsession;
606
607 ret_len = lttng_live_recv(viewer_connection, &lsession,
608 sizeof(lsession));
609 if (ret_len == 0) {
610 BT_COMP_LOGI("Remote side has closed connection");
611 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
612 goto error;
613 }
614 if (ret_len == BT_SOCKET_ERROR) {
615 BT_COMP_LOGE("Error receiving session: %s", bt_socket_errormsg());
616 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
617 goto error;
618 }
619 BT_ASSERT(ret_len == sizeof(lsession));
620 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
621 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
622 if (list_append_session(result, viewer_connection->url,
623 &lsession, viewer_connection)) {
624 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
625 goto error;
626 }
627 }
628
629 *user_result = result;
630 goto end;
631 error:
632 BT_VALUE_PUT_REF_AND_RESET(result);
633 end:
634 return status;
635 }
636
637 static
638 int lttng_live_query_session_ids(struct lttng_live_msg_iter *lttng_live_msg_iter)
639 {
640 struct lttng_viewer_cmd cmd;
641 struct lttng_viewer_list_sessions list;
642 struct lttng_viewer_session lsession;
643 uint32_t i, sessions_count;
644 ssize_t ret_len;
645 uint64_t session_id;
646 struct live_viewer_connection *viewer_connection =
647 lttng_live_msg_iter->viewer_connection;
648
649 cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
650 cmd.data_size = htobe64((uint64_t) 0);
651 cmd.cmd_version = htobe32(0);
652
653 ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
654 if (ret_len == BT_SOCKET_ERROR) {
655 BT_COMP_LOGE("Error sending cmd: %s", bt_socket_errormsg());
656 goto error;
657 }
658 BT_ASSERT(ret_len == sizeof(cmd));
659
660 ret_len = lttng_live_recv(viewer_connection, &list, sizeof(list));
661 if (ret_len == 0) {
662 BT_COMP_LOGI("Remote side has closed connection");
663 goto error;
664 }
665 if (ret_len == BT_SOCKET_ERROR) {
666 BT_COMP_LOGE("Error receiving session list: %s", bt_socket_errormsg());
667 goto error;
668 }
669 BT_ASSERT(ret_len == sizeof(list));
670
671 sessions_count = be32toh(list.sessions_count);
672 for (i = 0; i < sessions_count; i++) {
673 ret_len = lttng_live_recv(viewer_connection,
674 &lsession, sizeof(lsession));
675 if (ret_len == 0) {
676 BT_COMP_LOGI("Remote side has closed connection");
677 goto error;
678 }
679 if (ret_len == BT_SOCKET_ERROR) {
680 BT_COMP_LOGE("Error receiving session: %s", bt_socket_errormsg());
681 goto error;
682 }
683 BT_ASSERT(ret_len == sizeof(lsession));
684 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
685 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
686 session_id = be64toh(lsession.id);
687
688 BT_COMP_LOGI("Adding session %" PRIu64 " hostname: %s session_name: %s",
689 session_id, lsession.hostname, lsession.session_name);
690
691 if ((strncmp(lsession.session_name,
692 viewer_connection->session_name->str,
693 LTTNG_VIEWER_NAME_MAX) == 0) && (strncmp(lsession.hostname,
694 viewer_connection->target_hostname->str,
695 LTTNG_VIEWER_HOST_NAME_MAX) == 0)) {
696 if (lttng_live_add_session(lttng_live_msg_iter, session_id,
697 lsession.hostname,
698 lsession.session_name)) {
699 goto error;
700 }
701 }
702 }
703
704 return 0;
705
706 error:
707 BT_COMP_LOGE("Unable to query session ids");
708 return -1;
709 }
710
711 BT_HIDDEN
712 int lttng_live_create_viewer_session(
713 struct lttng_live_msg_iter *lttng_live_msg_iter)
714 {
715 struct lttng_viewer_cmd cmd;
716 struct lttng_viewer_create_session_response resp;
717 ssize_t ret_len;
718 struct live_viewer_connection *viewer_connection =
719 lttng_live_msg_iter->viewer_connection;
720
721 cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
722 cmd.data_size = htobe64((uint64_t) 0);
723 cmd.cmd_version = htobe32(0);
724
725 ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
726 if (ret_len == BT_SOCKET_ERROR) {
727 BT_COMP_LOGE("Error sending cmd: %s", bt_socket_errormsg());
728 goto error;
729 }
730 BT_ASSERT(ret_len == sizeof(cmd));
731
732 ret_len = lttng_live_recv(viewer_connection, &resp, sizeof(resp));
733 if (ret_len == 0) {
734 BT_COMP_LOGI("Remote side has closed connection");
735 goto error;
736 }
737 if (ret_len == BT_SOCKET_ERROR) {
738 BT_COMP_LOGE("Error receiving create session reply: %s", bt_socket_errormsg());
739 goto error;
740 }
741 BT_ASSERT(ret_len == sizeof(resp));
742
743 if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
744 BT_COMP_LOGE("Error creating viewer session");
745 goto error;
746 }
747 if (lttng_live_query_session_ids(lttng_live_msg_iter)) {
748 goto error;
749 }
750
751 return 0;
752
753 error:
754 return -1;
755 }
756
757 static
758 int receive_streams(struct lttng_live_session *session,
759 uint32_t stream_count)
760 {
761 ssize_t ret_len;
762 uint32_t i;
763 struct lttng_live_msg_iter *lttng_live_msg_iter =
764 session->lttng_live_msg_iter;
765 struct live_viewer_connection *viewer_connection =
766 lttng_live_msg_iter->viewer_connection;
767
768 BT_COMP_LOGI("Getting %" PRIu32 " new streams:", stream_count);
769 for (i = 0; i < stream_count; i++) {
770 struct lttng_viewer_stream stream;
771 struct lttng_live_stream_iterator *live_stream;
772 uint64_t stream_id;
773 uint64_t ctf_trace_id;
774
775 ret_len = lttng_live_recv(viewer_connection, &stream, sizeof(stream));
776 if (ret_len == 0) {
777 BT_COMP_LOGI("Remote side has closed connection");
778 goto error;
779 }
780 if (ret_len == BT_SOCKET_ERROR) {
781 BT_COMP_LOGE("Error receiving stream");
782 goto error;
783 }
784 BT_ASSERT(ret_len == sizeof(stream));
785 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
786 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
787 stream_id = be64toh(stream.id);
788 ctf_trace_id = be64toh(stream.ctf_trace_id);
789
790 if (stream.metadata_flag) {
791 BT_COMP_LOGI(" metadata stream %" PRIu64 " : %s/%s",
792 stream_id, stream.path_name, stream.channel_name);
793 if (lttng_live_metadata_create_stream(session,
794 ctf_trace_id, stream_id,
795 stream.path_name)) {
796 BT_COMP_LOGE("Error creating metadata stream");
797
798 goto error;
799 }
800 session->lazy_stream_msg_init = true;
801 } else {
802 BT_COMP_LOGI(" stream %" PRIu64 " : %s/%s",
803 stream_id, stream.path_name, stream.channel_name);
804 live_stream = lttng_live_stream_iterator_create(session,
805 ctf_trace_id, stream_id);
806 if (!live_stream) {
807 BT_COMP_LOGE("Error creating streamn");
808 goto error;
809 }
810 }
811 }
812 return 0;
813
814 error:
815 return -1;
816 }
817
818 BT_HIDDEN
819 int lttng_live_attach_session(struct lttng_live_session *session)
820 {
821 struct lttng_viewer_cmd cmd;
822 struct lttng_viewer_attach_session_request rq;
823 struct lttng_viewer_attach_session_response rp;
824 struct lttng_live_msg_iter *lttng_live_msg_iter =
825 session->lttng_live_msg_iter;
826 struct live_viewer_connection *viewer_connection =
827 lttng_live_msg_iter->viewer_connection;
828 uint64_t session_id = session->id;
829 uint32_t streams_count;
830 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
831 char cmd_buf[cmd_buf_len];
832 ssize_t ret_len;
833
834 cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
835 cmd.data_size = htobe64((uint64_t) sizeof(rq));
836 cmd.cmd_version = htobe32(0);
837
838 memset(&rq, 0, sizeof(rq));
839 rq.session_id = htobe64(session_id);
840 // TODO: add cmd line parameter to select seek beginning
841 // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
842 rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
843
844 /*
845 * Merge the cmd and connection request to prevent a write-write
846 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
847 * second write to be performed quickly in presence of Nagle's algorithm.
848 */
849 memcpy(cmd_buf, &cmd, sizeof(cmd));
850 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
851 ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
852 if (ret_len == BT_SOCKET_ERROR) {
853 BT_COMP_LOGE("Error sending attach request: %s", bt_socket_errormsg());
854 goto error;
855 }
856
857 BT_ASSERT(ret_len == cmd_buf_len);
858 ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
859 if (ret_len == 0) {
860 BT_COMP_LOGI("Remote side has closed connection");
861 goto error;
862 }
863 if (ret_len == BT_SOCKET_ERROR) {
864 BT_COMP_LOGE("Error receiving attach response: %s", bt_socket_errormsg());
865 goto error;
866 }
867 BT_ASSERT(ret_len == sizeof(rp));
868
869 streams_count = be32toh(rp.streams_count);
870 switch(be32toh(rp.status)) {
871 case LTTNG_VIEWER_ATTACH_OK:
872 break;
873 case LTTNG_VIEWER_ATTACH_UNK:
874 BT_COMP_LOGW("Session id %" PRIu64 " is unknown", session_id);
875 goto error;
876 case LTTNG_VIEWER_ATTACH_ALREADY:
877 BT_COMP_LOGW("There is already a viewer attached to this session");
878 goto error;
879 case LTTNG_VIEWER_ATTACH_NOT_LIVE:
880 BT_COMP_LOGW("Not a live session");
881 goto error;
882 case LTTNG_VIEWER_ATTACH_SEEK_ERR:
883 BT_COMP_LOGE("Wrong seek parameter");
884 goto error;
885 default:
886 BT_COMP_LOGE("Unknown attach return code %u", be32toh(rp.status));
887 goto error;
888 }
889
890 /* We receive the initial list of streams. */
891 if (receive_streams(session, streams_count)) {
892 goto error;
893 }
894
895 session->attached = true;
896 session->new_streams_needed = false;
897
898 return 0;
899
900 error:
901 return -1;
902 }
903
904 BT_HIDDEN
905 int lttng_live_detach_session(struct lttng_live_session *session)
906 {
907 struct lttng_viewer_cmd cmd;
908 struct lttng_viewer_detach_session_request rq;
909 struct lttng_viewer_detach_session_response rp;
910 ssize_t ret_len;
911 struct lttng_live_msg_iter *lttng_live_msg_iter =
912 session->lttng_live_msg_iter;
913 struct live_viewer_connection *viewer_connection =
914 lttng_live_msg_iter->viewer_connection;
915 uint64_t session_id = session->id;
916 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
917 char cmd_buf[cmd_buf_len];
918
919 if (!session->attached) {
920 return 0;
921 }
922
923 cmd.cmd = htobe32(LTTNG_VIEWER_DETACH_SESSION);
924 cmd.data_size = htobe64((uint64_t) sizeof(rq));
925 cmd.cmd_version = htobe32(0);
926
927 memset(&rq, 0, sizeof(rq));
928 rq.session_id = htobe64(session_id);
929
930 /*
931 * Merge the cmd and connection request to prevent a write-write
932 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
933 * second write to be performed quickly in presence of Nagle's algorithm.
934 */
935 memcpy(cmd_buf, &cmd, sizeof(cmd));
936 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
937 ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
938 if (ret_len == BT_SOCKET_ERROR) {
939 BT_COMP_LOGE("Error sending detach request: %s", bt_socket_errormsg());
940 goto error;
941 }
942
943 BT_ASSERT(ret_len == cmd_buf_len);
944 ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
945 if (ret_len == 0) {
946 BT_COMP_LOGI("Remote side has closed connection");
947 goto error;
948 }
949 if (ret_len == BT_SOCKET_ERROR) {
950 BT_COMP_LOGE("Error receiving detach response: %s", bt_socket_errormsg());
951 goto error;
952 }
953 BT_ASSERT(ret_len == sizeof(rp));
954
955 switch(be32toh(rp.status)) {
956 case LTTNG_VIEWER_DETACH_SESSION_OK:
957 break;
958 case LTTNG_VIEWER_DETACH_SESSION_UNK:
959 BT_COMP_LOGW("Session id %" PRIu64 " is unknown", session_id);
960 goto error;
961 case LTTNG_VIEWER_DETACH_SESSION_ERR:
962 BT_COMP_LOGW("Error detaching session id %" PRIu64 "", session_id);
963 goto error;
964 default:
965 BT_COMP_LOGE("Unknown detach return code %u", be32toh(rp.status));
966 goto error;
967 }
968
969 session->attached = false;
970
971 return 0;
972
973 error:
974 return -1;
975 }
976
977 BT_HIDDEN
978 ssize_t lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace,
979 FILE *fp)
980 {
981 uint64_t len = 0;
982 int ret;
983 struct lttng_viewer_cmd cmd;
984 struct lttng_viewer_get_metadata rq;
985 struct lttng_viewer_metadata_packet rp;
986 char *data = NULL;
987 ssize_t ret_len;
988 struct lttng_live_session *session = trace->session;
989 struct lttng_live_msg_iter *lttng_live_msg_iter =
990 session->lttng_live_msg_iter;
991 struct lttng_live_metadata *metadata = trace->metadata;
992 struct live_viewer_connection *viewer_connection =
993 lttng_live_msg_iter->viewer_connection;
994 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
995 char cmd_buf[cmd_buf_len];
996
997 rq.stream_id = htobe64(metadata->stream_id);
998 cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
999 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1000 cmd.cmd_version = htobe32(0);
1001
1002 /*
1003 * Merge the cmd and connection request to prevent a write-write
1004 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1005 * second write to be performed quickly in presence of Nagle's algorithm.
1006 */
1007 memcpy(cmd_buf, &cmd, sizeof(cmd));
1008 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1009 ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1010 if (ret_len == BT_SOCKET_ERROR) {
1011 BT_COMP_LOGE("Error sending get_metadata request: %s", bt_socket_errormsg());
1012 goto error;
1013 }
1014
1015 BT_ASSERT(ret_len == cmd_buf_len);
1016 ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1017 if (ret_len == 0) {
1018 BT_COMP_LOGI("Remote side has closed connection");
1019 goto error;
1020 }
1021 if (ret_len == BT_SOCKET_ERROR) {
1022 BT_COMP_LOGE("Error receiving get_metadata response: %s", bt_socket_errormsg());
1023 goto error;
1024 }
1025 BT_ASSERT(ret_len == sizeof(rp));
1026
1027 switch (be32toh(rp.status)) {
1028 case LTTNG_VIEWER_METADATA_OK:
1029 BT_COMP_LOGD("get_metadata : OK");
1030 break;
1031 case LTTNG_VIEWER_NO_NEW_METADATA:
1032 BT_COMP_LOGD("get_metadata : NO NEW");
1033 ret = 0;
1034 goto end;
1035 case LTTNG_VIEWER_METADATA_ERR:
1036 BT_COMP_LOGD("get_metadata : ERR");
1037 goto error;
1038 default:
1039 BT_COMP_LOGD("get_metadata : UNKNOWN");
1040 goto error;
1041 }
1042
1043 len = be64toh(rp.len);
1044 BT_COMP_LOGD("Writing %" PRIu64" bytes to metadata", len);
1045 if (len <= 0) {
1046 goto error;
1047 }
1048
1049 data = calloc(1, len);
1050 if (!data) {
1051 BT_COMP_LOGE("relay data calloc: %s", strerror(errno));
1052 goto error;
1053 }
1054 ret_len = lttng_live_recv(viewer_connection, data, len);
1055 if (ret_len == 0) {
1056 BT_COMP_LOGI("Remote side has closed connection");
1057 goto error_free_data;
1058 }
1059 if (ret_len == BT_SOCKET_ERROR) {
1060 BT_COMP_LOGE("Error receiving trace packet: %s", bt_socket_errormsg());
1061 goto error_free_data;
1062 }
1063 BT_ASSERT(ret_len == len);
1064
1065 do {
1066 ret_len = fwrite(data, 1, len, fp);
1067 } while (ret_len < 0 && errno == EINTR);
1068 if (ret_len < 0) {
1069 BT_COMP_LOGE("Writing in the metadata fp");
1070 goto error_free_data;
1071 }
1072 BT_ASSERT(ret_len == len);
1073 free(data);
1074 ret = len;
1075 end:
1076 return ret;
1077
1078 error_free_data:
1079 free(data);
1080 error:
1081 return -1;
1082 }
1083
1084 /*
1085 * Assign the fields from a lttng_viewer_index to a packet_index.
1086 */
1087 static
1088 void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
1089 struct packet_index *pindex)
1090 {
1091 BT_ASSERT(lindex);
1092 BT_ASSERT(pindex);
1093
1094 pindex->offset = be64toh(lindex->offset);
1095 pindex->packet_size = be64toh(lindex->packet_size);
1096 pindex->content_size = be64toh(lindex->content_size);
1097 pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
1098 pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
1099 pindex->events_discarded = be64toh(lindex->events_discarded);
1100 }
1101
1102 BT_HIDDEN
1103 enum lttng_live_iterator_status lttng_live_get_next_index(
1104 struct lttng_live_msg_iter *lttng_live_msg_iter,
1105 struct lttng_live_stream_iterator *stream,
1106 struct packet_index *index)
1107 {
1108 struct lttng_viewer_cmd cmd;
1109 struct lttng_viewer_get_next_index rq;
1110 struct lttng_viewer_index rp;
1111 enum lttng_live_iterator_status retstatus = LTTNG_LIVE_ITERATOR_STATUS_OK;
1112 struct live_viewer_connection *viewer_connection =
1113 lttng_live_msg_iter->viewer_connection;
1114 struct lttng_live_trace *trace = stream->trace;
1115 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1116 char cmd_buf[cmd_buf_len];
1117 uint32_t flags, status;
1118 ssize_t ret_len;
1119
1120 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1121 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1122 cmd.cmd_version = htobe32(0);
1123
1124
1125 memset(&rq, 0, sizeof(rq));
1126 rq.stream_id = htobe64(stream->viewer_stream_id);
1127
1128 /*
1129 * Merge the cmd and connection request to prevent a write-write
1130 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1131 * second write to be performed quickly in presence of Nagle's algorithm.
1132 */
1133 memcpy(cmd_buf, &cmd, sizeof(cmd));
1134 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1135 ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1136 if (ret_len == BT_SOCKET_ERROR) {
1137 BT_COMP_LOGE("Error sending get_next_index request: %s",
1138 bt_socket_errormsg());
1139 goto error;
1140 }
1141
1142 BT_ASSERT(ret_len == cmd_buf_len);
1143 ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1144 if (ret_len == 0) {
1145 BT_COMP_LOGI("Remote side has closed connection");
1146 goto error;
1147 }
1148 if (ret_len == BT_SOCKET_ERROR) {
1149 BT_COMP_LOGE("Error receiving get_next_index response: %s",
1150 bt_socket_errormsg());
1151 goto error;
1152 }
1153 BT_ASSERT(ret_len == sizeof(rp));
1154
1155 flags = be32toh(rp.flags);
1156 status = be32toh(rp.status);
1157
1158 switch (status) {
1159 case LTTNG_VIEWER_INDEX_INACTIVE:
1160 {
1161 uint64_t ctf_stream_class_id;
1162
1163 BT_COMP_LOGD("get_next_index: inactive");
1164 memset(index, 0, sizeof(struct packet_index));
1165 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
1166 stream->current_inactivity_ts = index->ts_cycles.timestamp_end;
1167 ctf_stream_class_id = be64toh(rp.stream_id);
1168 if (stream->ctf_stream_class_id != -1ULL) {
1169 BT_ASSERT(stream->ctf_stream_class_id ==
1170 ctf_stream_class_id);
1171 } else {
1172 stream->ctf_stream_class_id = ctf_stream_class_id;
1173 }
1174 stream->state = LTTNG_LIVE_STREAM_QUIESCENT;
1175 break;
1176 }
1177 case LTTNG_VIEWER_INDEX_OK:
1178 {
1179 uint64_t ctf_stream_class_id;
1180
1181 BT_COMP_LOGD("get_next_index: OK");
1182 lttng_index_to_packet_index(&rp, index);
1183 ctf_stream_class_id = be64toh(rp.stream_id);
1184 if (stream->ctf_stream_class_id != -1ULL) {
1185 BT_ASSERT(stream->ctf_stream_class_id ==
1186 ctf_stream_class_id);
1187 } else {
1188 stream->ctf_stream_class_id = ctf_stream_class_id;
1189 }
1190
1191 stream->state = LTTNG_LIVE_STREAM_ACTIVE_DATA;
1192
1193 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1194 BT_COMP_LOGD("get_next_index: new metadata needed");
1195 trace->new_metadata_needed = true;
1196 }
1197 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1198 BT_COMP_LOGD("get_next_index: new streams needed");
1199 lttng_live_need_new_streams(lttng_live_msg_iter);
1200 }
1201 break;
1202 }
1203 case LTTNG_VIEWER_INDEX_RETRY:
1204 BT_COMP_LOGD("get_next_index: retry");
1205 memset(index, 0, sizeof(struct packet_index));
1206 retstatus = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1207 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1208 goto end;
1209 case LTTNG_VIEWER_INDEX_HUP:
1210 BT_COMP_LOGD("get_next_index: stream hung up");
1211 memset(index, 0, sizeof(struct packet_index));
1212 index->offset = EOF;
1213 retstatus = LTTNG_LIVE_ITERATOR_STATUS_END;
1214 stream->state = LTTNG_LIVE_STREAM_EOF;
1215 stream->has_stream_hung_up = true;
1216 break;
1217 case LTTNG_VIEWER_INDEX_ERR:
1218 BT_COMP_LOGE("get_next_index: error");
1219 memset(index, 0, sizeof(struct packet_index));
1220 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1221 goto error;
1222 default:
1223 BT_COMP_LOGE("get_next_index: unknown value");
1224 memset(index, 0, sizeof(struct packet_index));
1225 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1226 goto error;
1227 }
1228 end:
1229 return retstatus;
1230
1231 error:
1232 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
1233 retstatus = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1234 } else {
1235 retstatus = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1236 }
1237 return retstatus;
1238 }
1239
1240 BT_HIDDEN
1241 enum bt_msg_iter_medium_status lttng_live_get_stream_bytes(
1242 struct lttng_live_msg_iter *lttng_live_msg_iter,
1243 struct lttng_live_stream_iterator *stream, uint8_t *buf,
1244 uint64_t offset, uint64_t req_len, uint64_t *recv_len)
1245 {
1246 enum bt_msg_iter_medium_status retstatus = BT_MSG_ITER_MEDIUM_STATUS_OK;
1247 struct lttng_viewer_trace_packet rp;
1248 struct lttng_viewer_cmd cmd;
1249 struct lttng_viewer_get_packet rq;
1250 struct live_viewer_connection *viewer_connection =
1251 lttng_live_msg_iter->viewer_connection;
1252 struct lttng_live_trace *trace = stream->trace;
1253 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1254 char cmd_buf[cmd_buf_len];
1255 uint32_t flags, status;
1256 ssize_t ret_len;
1257
1258 BT_COMP_LOGD("lttng_live_get_stream_bytes: offset=%" PRIu64 ", req_len=%" PRIu64,
1259 offset, req_len);
1260 cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
1261 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1262 cmd.cmd_version = htobe32(0);
1263
1264 memset(&rq, 0, sizeof(rq));
1265 rq.stream_id = htobe64(stream->viewer_stream_id);
1266 rq.offset = htobe64(offset);
1267 rq.len = htobe32(req_len);
1268
1269 /*
1270 * Merge the cmd and connection request to prevent a write-write
1271 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1272 * second write to be performed quickly in presence of Nagle's algorithm.
1273 */
1274 memcpy(cmd_buf, &cmd, sizeof(cmd));
1275 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1276 ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1277 if (ret_len == BT_SOCKET_ERROR) {
1278 BT_COMP_LOGE("Error sending get_data request: %s", bt_socket_errormsg());
1279 goto error;
1280 }
1281
1282 BT_ASSERT(ret_len == cmd_buf_len);
1283 ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1284 if (ret_len == 0) {
1285 BT_COMP_LOGI("Remote side has closed connection");
1286 goto error;
1287 }
1288 if (ret_len == BT_SOCKET_ERROR) {
1289 BT_COMP_LOGE("Error receiving get_data response: %s", bt_socket_errormsg());
1290 goto error;
1291 }
1292 if (ret_len != sizeof(rp)) {
1293 BT_COMP_LOGE("get_data_packet: expected %zu"
1294 ", received %zd", sizeof(rp),
1295 ret_len);
1296 goto error;
1297 }
1298
1299 flags = be32toh(rp.flags);
1300 status = be32toh(rp.status);
1301
1302 switch (status) {
1303 case LTTNG_VIEWER_GET_PACKET_OK:
1304 req_len = be32toh(rp.len);
1305 BT_COMP_LOGD("get_data_packet: Ok, packet size : %" PRIu64 "", req_len);
1306 break;
1307 case LTTNG_VIEWER_GET_PACKET_RETRY:
1308 /* Unimplemented by relay daemon */
1309 BT_COMP_LOGD("get_data_packet: retry");
1310 retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1311 goto end;
1312 case LTTNG_VIEWER_GET_PACKET_ERR:
1313 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1314 BT_COMP_LOGD("get_data_packet: new metadata needed, try again later");
1315 trace->new_metadata_needed = true;
1316 }
1317 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1318 BT_COMP_LOGD("get_data_packet: new streams needed, try again later");
1319 lttng_live_need_new_streams(lttng_live_msg_iter);
1320 }
1321 if (flags & (LTTNG_VIEWER_FLAG_NEW_METADATA
1322 | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
1323 retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1324 goto end;
1325 }
1326 BT_COMP_LOGE("get_data_packet: error");
1327 goto error;
1328 case LTTNG_VIEWER_GET_PACKET_EOF:
1329 retstatus = BT_MSG_ITER_MEDIUM_STATUS_EOF;
1330 goto end;
1331 default:
1332 BT_COMP_LOGE("get_data_packet: unknown");
1333 goto error;
1334 }
1335
1336 if (req_len == 0) {
1337 goto error;
1338 }
1339
1340 ret_len = lttng_live_recv(viewer_connection, buf, req_len);
1341 if (ret_len == 0) {
1342 BT_COMP_LOGI("Remote side has closed connection");
1343 goto error;
1344 }
1345 if (ret_len == BT_SOCKET_ERROR) {
1346 BT_COMP_LOGE("Error receiving trace packet: %s", bt_socket_errormsg());
1347 goto error;
1348 }
1349 BT_ASSERT(ret_len == req_len);
1350 *recv_len = ret_len;
1351 end:
1352 return retstatus;
1353
1354 error:
1355 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
1356 retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1357 } else {
1358 retstatus = BT_MSG_ITER_MEDIUM_STATUS_ERROR;
1359 }
1360 return retstatus;
1361 }
1362
1363 /*
1364 * Request new streams for a session.
1365 */
1366 BT_HIDDEN
1367 enum lttng_live_iterator_status lttng_live_get_new_streams(
1368 struct lttng_live_session *session)
1369 {
1370 enum lttng_live_iterator_status status =
1371 LTTNG_LIVE_ITERATOR_STATUS_OK;
1372 struct lttng_viewer_cmd cmd;
1373 struct lttng_viewer_new_streams_request rq;
1374 struct lttng_viewer_new_streams_response rp;
1375 struct lttng_live_msg_iter *lttng_live_msg_iter =
1376 session->lttng_live_msg_iter;
1377 struct live_viewer_connection *viewer_connection =
1378 lttng_live_msg_iter->viewer_connection;
1379 uint32_t streams_count;
1380 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1381 char cmd_buf[cmd_buf_len];
1382 ssize_t ret_len;
1383
1384 if (!session->new_streams_needed) {
1385 return LTTNG_LIVE_ITERATOR_STATUS_OK;
1386 }
1387
1388 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1389 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1390 cmd.cmd_version = htobe32(0);
1391
1392 memset(&rq, 0, sizeof(rq));
1393 rq.session_id = htobe64(session->id);
1394
1395 /*
1396 * Merge the cmd and connection request to prevent a write-write
1397 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1398 * second write to be performed quickly in presence of Nagle's algorithm.
1399 */
1400 memcpy(cmd_buf, &cmd, sizeof(cmd));
1401 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1402 ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1403 if (ret_len == BT_SOCKET_ERROR) {
1404 BT_COMP_LOGE("Error sending get_new_streams request: %s",
1405 bt_socket_errormsg());
1406 goto error;
1407 }
1408
1409 BT_ASSERT(ret_len == cmd_buf_len);
1410 ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1411 if (ret_len == 0) {
1412 BT_COMP_LOGI("Remote side has closed connection");
1413 goto error;
1414 }
1415 if (ret_len == BT_SOCKET_ERROR) {
1416 BT_COMP_LOGE("Error receiving get_new_streams response");
1417 goto error;
1418 }
1419 BT_ASSERT(ret_len == sizeof(rp));
1420
1421 streams_count = be32toh(rp.streams_count);
1422
1423 switch(be32toh(rp.status)) {
1424 case LTTNG_VIEWER_NEW_STREAMS_OK:
1425 session->new_streams_needed = false;
1426 break;
1427 case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1428 session->new_streams_needed = false;
1429 goto end;
1430 case LTTNG_VIEWER_NEW_STREAMS_HUP:
1431 session->new_streams_needed = false;
1432 session->closed = true;
1433 status = LTTNG_LIVE_ITERATOR_STATUS_END;
1434 goto end;
1435 case LTTNG_VIEWER_NEW_STREAMS_ERR:
1436 BT_COMP_LOGE("get_new_streams error");
1437 goto error;
1438 default:
1439 BT_COMP_LOGE("Unknown return code %u", be32toh(rp.status));
1440 goto error;
1441 }
1442
1443 if (receive_streams(session, streams_count)) {
1444 goto error;
1445 }
1446 end:
1447 return status;
1448
1449 error:
1450 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
1451 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1452 } else {
1453 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1454 }
1455 return status;
1456 }
1457
1458 BT_HIDDEN
1459 struct live_viewer_connection *live_viewer_connection_create(
1460 const char *url, bool in_query,
1461 struct lttng_live_msg_iter *lttng_live_msg_iter,
1462 bt_logging_level log_level)
1463 {
1464 struct live_viewer_connection *viewer_connection;
1465
1466 viewer_connection = g_new0(struct live_viewer_connection, 1);
1467
1468 if (bt_socket_init(log_level) != 0) {
1469 goto error;
1470 }
1471
1472 viewer_connection->log_level = log_level;
1473
1474 if (lttng_live_msg_iter) {
1475 viewer_connection->self_comp = lttng_live_msg_iter->self_comp;
1476 }
1477
1478 bt_object_init_shared(&viewer_connection->obj, connection_release);
1479 viewer_connection->control_sock = BT_INVALID_SOCKET;
1480 viewer_connection->port = -1;
1481 viewer_connection->in_query = in_query;
1482 viewer_connection->lttng_live_msg_iter = lttng_live_msg_iter;
1483 viewer_connection->url = g_string_new(url);
1484 if (!viewer_connection->url) {
1485 goto error;
1486 }
1487
1488 BT_COMP_LOGI("Establishing connection to url \"%s\"...", url);
1489 if (lttng_live_connect_viewer(viewer_connection)) {
1490 goto error_report;
1491 }
1492 BT_COMP_LOGI("Connection to url \"%s\" is established", url);
1493 return viewer_connection;
1494
1495 error_report:
1496 BT_COMP_LOGW("Failure to establish connection to url \"%s\"", url);
1497 error:
1498 if (viewer_connection) {
1499 live_viewer_connection_destroy(viewer_connection);
1500 }
1501
1502 return NULL;
1503 }
1504
1505 BT_HIDDEN
1506 void live_viewer_connection_destroy(
1507 struct live_viewer_connection *viewer_connection)
1508 {
1509 BT_COMP_LOGI("Closing connection to url \"%s\"", viewer_connection->url->str);
1510 lttng_live_disconnect_viewer(viewer_connection);
1511 if (viewer_connection->url) {
1512 g_string_free(viewer_connection->url, true);
1513 }
1514 if (viewer_connection->relay_hostname) {
1515 g_string_free(viewer_connection->relay_hostname, true);
1516 }
1517 if (viewer_connection->target_hostname) {
1518 g_string_free(viewer_connection->target_hostname, true);
1519 }
1520 if (viewer_connection->session_name) {
1521 g_string_free(viewer_connection->session_name, true);
1522 }
1523 g_free(viewer_connection);
1524
1525 bt_socket_fini();
1526 }
BabelTrace - Trace Format Babel Tower
This page took 0.06059 seconds and 4 git commands to generate.