Fix: futex wait: handle spurious futex wakeups
[lttng-tools.git] / tests / regression / tools / live / live_test.cpp
1 /*
2 * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
3 *
4 * SPDX-License-Identifier: GPL-2.0-only
5 *
6 */
7
8 #include <stdio.h>
9 #include <stdlib.h>
10 #include <string.h>
11 #include <unistd.h>
12 #include <common/compat/time.hpp>
13 #include <sys/types.h>
14 #include <inttypes.h>
15 #include <stdlib.h>
16 #include <sys/socket.h>
17 #include <netinet/in.h>
18 #include <netdb.h>
19 #include <fcntl.h>
20 #include <sys/mman.h>
21 #include <sys/stat.h>
22
23 #include <tap/tap.h>
24 #include <lttng/lttng.h>
25
26 #include <urcu/list.h>
27 #include <common/common.hpp>
28
29 #include <bin/lttng-relayd/lttng-viewer-abi.hpp>
30 #include <common/index/ctf-index.hpp>
31
32 #include <common/compat/errno.hpp>
33 #include <common/compat/endian.hpp>
34
35 #define SESSION1 "test1"
36 #define RELAYD_URL "net://localhost"
37 #define LIVE_TIMER 2000000
38
39 /* Number of TAP tests in this file */
40 #define NUM_TESTS 11
41 #define mmap_size 524288
42
43 #ifdef HAVE_LIBLTTNG_UST_CTL
44 #include <lttng/lttng-export.h>
45 #include <lttng/ust-sigbus.h>
46 LTTNG_EXPORT DEFINE_LTTNG_UST_SIGBUS_STATE();
47 #endif
48
49 namespace {
50 struct live_session *session;
51 int control_sock;
52
53 int first_packet_offset;
54 int first_packet_len;
55 int first_packet_stream_id = -1;
56
57 struct viewer_stream {
58 uint64_t id;
59 uint64_t ctf_trace_id;
60 void *mmap_base;
61 int fd;
62 int metadata_flag;
63 int first_read;
64 char path[PATH_MAX];
65 };
66
67 struct live_session {
68 struct viewer_stream *streams;
69 uint64_t live_timer_interval;
70 uint64_t stream_count;
71 };
72 } /* namespace */
73
74 static
75 ssize_t lttng_live_recv(int fd, void *buf, size_t len)
76 {
77 ssize_t ret;
78 size_t copied = 0, to_copy = len;
79
80 do {
81 ret = recv(fd, (char *) buf + copied, to_copy, 0);
82 if (ret > 0) {
83 LTTNG_ASSERT(ret <= to_copy);
84 copied += ret;
85 to_copy -= ret;
86 }
87 } while ((ret > 0 && to_copy > 0)
88 || (ret < 0 && errno == EINTR));
89 if (ret > 0)
90 ret = copied;
91 /* ret = 0 means orderly shutdown, ret < 0 is error. */
92 return ret;
93 }
94
95 static
96 ssize_t lttng_live_send(int fd, const void *buf, size_t len)
97 {
98 ssize_t ret;
99
100 do {
101 ret = send(fd, buf, len, MSG_NOSIGNAL);
102 } while (ret < 0 && errno == EINTR);
103 return ret;
104 }
105
106 static
107 int connect_viewer(const char *hostname)
108 {
109 struct hostent *host;
110 struct sockaddr_in server_addr;
111 int ret;
112
113 host = gethostbyname(hostname);
114 if (!host) {
115 ret = -1;
116 goto end;
117 }
118
119 if ((control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
120 PERROR("Socket");
121 ret = -1;
122 goto end;
123 }
124
125 server_addr.sin_family = AF_INET;
126 server_addr.sin_port = htons(5344);
127 server_addr.sin_addr = *((struct in_addr *) host->h_addr);
128 bzero(&(server_addr.sin_zero), 8);
129
130 if (connect(control_sock, (struct sockaddr *) &server_addr,
131 sizeof(struct sockaddr)) == -1) {
132 PERROR("Connect");
133 ret = -1;
134 goto end;
135 }
136
137 server_addr.sin_family = AF_INET;
138 server_addr.sin_port = htons(5345);
139 server_addr.sin_addr = *((struct in_addr *) host->h_addr);
140 bzero(&(server_addr.sin_zero), 8);
141
142 ret = 0;
143
144 end:
145 return ret;
146 }
147
148 static
149 int establish_connection(void)
150 {
151 struct lttng_viewer_cmd cmd;
152 struct lttng_viewer_connect connect;
153 ssize_t ret_len;
154
155 cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
156 cmd.data_size = htobe64(sizeof(connect));
157 cmd.cmd_version = htobe32(0);
158
159 memset(&connect, 0, sizeof(connect));
160 connect.major = htobe32(VERSION_MAJOR);
161 connect.minor = htobe32(VERSION_MINOR);
162 connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
163
164 ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
165 if (ret_len < 0) {
166 diag("Error sending cmd");
167 goto error;
168 }
169 ret_len = lttng_live_send(control_sock, &connect, sizeof(connect));
170 if (ret_len < 0) {
171 diag("Error sending version");
172 goto error;
173 }
174
175 ret_len = lttng_live_recv(control_sock, &connect, sizeof(connect));
176 if (ret_len == 0) {
177 diag("[error] Remote side has closed connection");
178 goto error;
179 }
180 if (ret_len < 0) {
181 diag("Error receiving version");
182 goto error;
183 }
184 return 0;
185
186 error:
187 return -1;
188 }
189
190 /*
191 * Returns the number of sessions, should be 1 during the unit test.
192 */
193 static
194 int list_sessions(uint64_t *session_id)
195 {
196 struct lttng_viewer_cmd cmd;
197 struct lttng_viewer_list_sessions list;
198 struct lttng_viewer_session lsession;
199 int i;
200 ssize_t ret_len;
201 int first_session = 0;
202
203 cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
204 cmd.data_size = htobe64(0);
205 cmd.cmd_version = htobe32(0);
206
207 ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
208 if (ret_len < 0) {
209 diag("Error sending cmd");
210 goto error;
211 }
212
213 ret_len = lttng_live_recv(control_sock, &list, sizeof(list));
214 if (ret_len == 0) {
215 diag("[error] Remote side has closed connection");
216 goto error;
217 }
218 if (ret_len < 0) {
219 diag("Error receiving session list");
220 goto error;
221 }
222
223 for (i = 0; i < be32toh(list.sessions_count); i++) {
224 ret_len = lttng_live_recv(control_sock, &lsession, sizeof(lsession));
225 if (ret_len < 0) {
226 diag("Error receiving session");
227 goto error;
228 }
229 if (lsession.streams > 0 && first_session <= 0) {
230 first_session = be64toh(lsession.id);
231 *session_id = first_session;
232 }
233 }
234
235 return be32toh(list.sessions_count);
236
237 error:
238 return -1;
239 }
240
241 static
242 int create_viewer_session(void)
243 {
244 struct lttng_viewer_cmd cmd;
245 struct lttng_viewer_create_session_response resp;
246 ssize_t ret_len;
247
248 cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
249 cmd.data_size = htobe64(0);
250 cmd.cmd_version = htobe32(0);
251
252 ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
253 if (ret_len < 0) {
254 diag("[error] Error sending cmd");
255 goto error;
256 }
257 LTTNG_ASSERT(ret_len == sizeof(cmd));
258
259 ret_len = lttng_live_recv(control_sock, &resp, sizeof(resp));
260 if (ret_len == 0) {
261 diag("[error] Remote side has closed connection");
262 goto error;
263 }
264 if (ret_len < 0) {
265 diag("[error] Error receiving create session reply");
266 goto error;
267 }
268 LTTNG_ASSERT(ret_len == sizeof(resp));
269
270 if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
271 diag("[error] Error creating viewer session");
272 goto error;
273 }
274 return 0;
275
276 error:
277 return -1;
278 }
279
280 static
281 int attach_session(uint64_t id)
282 {
283 struct lttng_viewer_cmd cmd;
284 struct lttng_viewer_attach_session_request rq;
285 struct lttng_viewer_attach_session_response rp;
286 struct lttng_viewer_stream stream;
287 int i;
288 ssize_t ret_len;
289
290 session = zmalloc<live_session>();
291 if (!session) {
292 goto error;
293 }
294
295 cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
296 cmd.data_size = htobe64(sizeof(rq));
297 cmd.cmd_version = htobe32(0);
298
299 memset(&rq, 0, sizeof(rq));
300 rq.session_id = htobe64(id);
301 rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
302
303 ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
304 if (ret_len < 0) {
305 diag("Error sending cmd LTTNG_VIEWER_ATTACH_SESSION");
306 goto error;
307 }
308 ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
309 if (ret_len < 0) {
310 diag("Error sending attach request");
311 goto error;
312 }
313
314 ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
315 if (ret_len == 0) {
316 diag("[error] Remote side has closed connection");
317 goto error;
318 }
319 if (ret_len < 0) {
320 diag("Error receiving attach response");
321 goto error;
322 }
323 if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
324 goto error;
325 }
326
327 session->stream_count = be32toh(rp.streams_count);
328 if (session->stream_count == 0) {
329 diag("Got session stream count == 0");
330 goto error;
331 }
332 session->streams = calloc<viewer_stream>(session->stream_count);
333 if (!session->streams) {
334 goto error;
335 }
336
337 for (i = 0; i < be32toh(rp.streams_count); i++) {
338 ret_len = lttng_live_recv(control_sock, &stream, sizeof(stream));
339 if (ret_len == 0) {
340 diag("[error] Remote side has closed connection");
341 goto error;
342 }
343 if (ret_len < 0) {
344 diag("Error receiving stream");
345 goto error;
346 }
347 session->streams[i].id = be64toh(stream.id);
348
349 session->streams[i].ctf_trace_id = be64toh(stream.ctf_trace_id);
350 session->streams[i].first_read = 1;
351 session->streams[i].mmap_base = mmap(NULL, mmap_size,
352 PROT_READ | PROT_WRITE,
353 MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
354 if (session->streams[i].mmap_base == MAP_FAILED) {
355 diag("mmap error");
356 goto error;
357 }
358
359 if (be32toh(stream.metadata_flag)) {
360 session->streams[i].metadata_flag = 1;
361 }
362 }
363 return session->stream_count;
364
365 error:
366 return -1;
367 }
368
369 static
370 int get_metadata(void)
371 {
372 struct lttng_viewer_cmd cmd;
373 struct lttng_viewer_get_metadata rq;
374 struct lttng_viewer_metadata_packet rp;
375 ssize_t ret_len;
376 int ret;
377 uint64_t i;
378 char *data = NULL;
379 uint64_t len = 0;
380 int metadata_stream_id = -1;
381
382 cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
383 cmd.data_size = htobe64(sizeof(rq));
384 cmd.cmd_version = htobe32(0);
385
386 for (i = 0; i < session->stream_count; i++) {
387 if (session->streams[i].metadata_flag) {
388 metadata_stream_id = i;
389 break;
390 }
391 }
392
393 if (metadata_stream_id < 0) {
394 diag("No metadata stream found");
395 goto error;
396 }
397
398 rq.stream_id = htobe64(session->streams[metadata_stream_id].id);
399
400 retry:
401 ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
402 if (ret_len < 0) {
403 diag("Error sending cmd");
404 goto error;
405 }
406 ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
407 if (ret_len < 0) {
408 diag("Error sending get_metadata request");
409 goto error;
410 }
411 ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
412 if (ret_len == 0) {
413 diag("[error] Remote side has closed connection");
414 goto error;
415 }
416 if (ret_len < 0) {
417 diag("Error receiving metadata response");
418 goto error;
419 }
420 switch (be32toh(rp.status)) {
421 case LTTNG_VIEWER_METADATA_OK:
422 break;
423 case LTTNG_VIEWER_NO_NEW_METADATA:
424 diag("Got LTTNG_VIEWER_NO_NEW_METADATA:");
425 usleep(50);
426 goto retry;
427 case LTTNG_VIEWER_METADATA_ERR:
428 diag("Got LTTNG_VIEWER_METADATA_ERR:");
429 goto error;
430 default:
431 diag("Got unknown status during LTTNG_VIEWER_GET_METADATA");
432 goto error;
433 }
434
435 len = be64toh(rp.len);
436 if (len <= 0) {
437 goto error;
438 }
439
440 data = calloc<char>(len);
441 if (!data) {
442 PERROR("relay data zmalloc");
443 goto error;
444 }
445 ret_len = lttng_live_recv(control_sock, data, len);
446 if (ret_len == 0) {
447 diag("[error] Remote side has closed connection");
448 goto error_free_data;
449 }
450 if (ret_len < 0) {
451 diag("Error receiving trace packet");
452 goto error_free_data;
453 }
454 free(data);
455 ret = len;
456
457 return ret;
458
459 error_free_data:
460 free(data);
461 error:
462 return -1;
463 }
464
465 static
466 int get_next_index(void)
467 {
468 struct lttng_viewer_cmd cmd;
469 struct lttng_viewer_get_next_index rq;
470 struct lttng_viewer_index rp;
471 ssize_t ret_len;
472 int id;
473
474 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
475 cmd.data_size = htobe64(sizeof(rq));
476 cmd.cmd_version = htobe32(0);
477
478 for (id = 0; id < session->stream_count; id++) {
479 if (session->streams[id].metadata_flag) {
480 continue;
481 }
482 memset(&rq, 0, sizeof(rq));
483 rq.stream_id = htobe64(session->streams[id].id);
484
485 retry:
486 ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
487 if (ret_len < 0) {
488 diag("Error sending cmd");
489 goto error;
490 }
491 ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
492 if (ret_len < 0) {
493 diag("Error sending get_next_index request");
494 goto error;
495 }
496 ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
497 if (ret_len == 0) {
498 diag("[error] Remote side has closed connection");
499 goto error;
500 }
501 if (ret_len < 0) {
502 diag("Error receiving index response");
503 goto error;
504 }
505
506 rp.flags = be32toh(rp.flags);
507
508 switch (be32toh(rp.status)) {
509 case LTTNG_VIEWER_INDEX_INACTIVE:
510 /* Skip this stream. */
511 diag("Got LTTNG_VIEWER_INDEX_INACTIVE");
512 continue;
513 case LTTNG_VIEWER_INDEX_OK:
514 break;
515 case LTTNG_VIEWER_INDEX_RETRY:
516 sleep(1);
517 goto retry;
518 case LTTNG_VIEWER_INDEX_HUP:
519 diag("Got LTTNG_VIEWER_INDEX_HUP");
520 session->streams[id].id = -1ULL;
521 session->streams[id].fd = -1;
522 goto error;
523 case LTTNG_VIEWER_INDEX_ERR:
524 diag("Got LTTNG_VIEWER_INDEX_ERR");
525 goto error;
526 default:
527 diag("Unknown reply status during LTTNG_VIEWER_GET_NEXT_INDEX (%d)", be32toh(rp.status));
528 goto error;
529 }
530 if (first_packet_stream_id < 0) {
531 /*
532 * Initialize the first packet stream id. That is,
533 * the first active stream encoutered.
534 */
535 first_packet_offset = be64toh(rp.offset);
536 first_packet_len = be64toh(rp.packet_size) / CHAR_BIT;
537 first_packet_stream_id = id;
538 diag("Got first packet index with offset %d and len %d",
539 first_packet_offset, first_packet_len);
540 }
541 }
542 return 0;
543
544 error:
545 return -1;
546 }
547
548 static
549 int get_data_packet(int id, uint64_t offset,
550 uint64_t len)
551 {
552 struct lttng_viewer_cmd cmd;
553 struct lttng_viewer_get_packet rq;
554 struct lttng_viewer_trace_packet rp;
555 ssize_t ret_len;
556
557 cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
558 cmd.data_size = htobe64(sizeof(rq));
559 cmd.cmd_version = htobe32(0);
560
561 memset(&rq, 0, sizeof(rq));
562 rq.stream_id = htobe64(session->streams[id].id);
563 /* Already in big endian. */
564 rq.offset = offset;
565 rq.len = htobe32(len);
566
567 ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
568 if (ret_len < 0) {
569 diag("Error sending cmd");
570 goto error;
571 }
572 ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
573 if (ret_len < 0) {
574 diag("Error sending get_data_packet request");
575 goto error;
576 }
577 ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
578 if (ret_len == 0) {
579 diag("[error] Remote side has closed connection");
580 goto error;
581 }
582 if (ret_len < 0) {
583 diag("Error receiving data response");
584 goto error;
585 }
586 rp.flags = be32toh(rp.flags);
587
588 switch (be32toh(rp.status)) {
589 case LTTNG_VIEWER_GET_PACKET_OK:
590 len = be32toh(rp.len);
591 if (len == 0) {
592 diag("Got LTTNG_VIEWER_GET_PACKET_OK, but len == 0");
593 goto error;
594 }
595 break;
596 case LTTNG_VIEWER_GET_PACKET_RETRY:
597 diag("Got LTTNG_VIEWER_GET_PACKET_RETRY:");
598 goto error;
599 case LTTNG_VIEWER_GET_PACKET_ERR:
600 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
601 diag("Got LTTNG_VIEWER_GET_PACKET_ERR with NEW_METADATA flag");
602 goto end;
603 }
604 diag("Got LTTNG_VIEWER_GET_PACKET_ERR:");
605 goto error;
606 default:
607 diag("Got unknown status code during LTTNG_VIEWER_GET_PACKET");
608 goto error;
609 }
610
611 if (len > mmap_size) {
612 diag("mmap_size not big enough");
613 goto error;
614 }
615
616 ret_len = lttng_live_recv(control_sock, session->streams[id].mmap_base, len);
617 if (ret_len == 0) {
618 diag("[error] Remote side has closed connection");
619 goto error;
620 }
621 if (ret_len < 0) {
622 diag("Error receiving trace packet");
623 goto error;
624 }
625 end:
626 return 0;
627 error:
628 return -1;
629 }
630
631 static
632 int detach_viewer_session(uint64_t id)
633 {
634 struct lttng_viewer_cmd cmd;
635 struct lttng_viewer_detach_session_response resp;
636 struct lttng_viewer_detach_session_request rq;
637 int ret;
638 ssize_t ret_len;
639
640 cmd.cmd = htobe32(LTTNG_VIEWER_DETACH_SESSION);
641 cmd.data_size = htobe64(sizeof(rq));
642 cmd.cmd_version = htobe32(0);
643
644 memset(&rq, 0, sizeof(rq));
645 rq.session_id = htobe64(id);
646
647 ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
648 if (ret_len < 0) {
649 fprintf(stderr, "[error] Error sending cmd\n");
650 ret = ret_len;
651 goto error;
652 }
653
654 ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
655 if (ret_len < 0) {
656 fprintf(stderr, "Error sending attach request\n");
657 ret = ret_len;
658 goto error;
659 }
660
661 ret_len = lttng_live_recv(control_sock, &resp, sizeof(resp));
662 if (ret_len < 0) {
663 fprintf(stderr, "[error] Error receiving detach session reply\n");
664 ret = ret_len;
665 goto error;
666 }
667
668 if (be32toh(resp.status) != LTTNG_VIEWER_DETACH_SESSION_OK) {
669 fprintf(stderr, "[error] Error detaching viewer session\n");
670 ret = -1;
671 goto error;
672 }
673 ret = 0;
674
675 error:
676 return ret;
677 }
678
679 int main(void)
680 {
681 int ret;
682 uint64_t session_id;
683
684 plan_tests(NUM_TESTS);
685
686 diag("Live unit tests");
687
688 ret = connect_viewer("localhost");
689 ok(ret == 0, "Connect viewer to relayd");
690
691 ret = establish_connection();
692 ok(ret == 0, "Established connection and version check with %d.%d",
693 VERSION_MAJOR, VERSION_MINOR);
694
695 ret = list_sessions(&session_id);
696 ok(ret > 0, "List sessions : %d session(s)", ret);
697 if (ret < 0) {
698 goto end;
699 }
700
701 ret = create_viewer_session();
702 ok(ret == 0, "Create viewer session");
703
704 ret = attach_session(session_id);
705 ok(ret > 0, "Attach to session, %d stream(s) received", ret);
706
707 ret = get_metadata();
708 ok(ret > 0, "Get metadata, received %d bytes", ret);
709
710 ret = get_next_index();
711 ok(ret == 0, "Get one index per stream");
712
713 ret = get_data_packet(first_packet_stream_id, first_packet_offset,
714 first_packet_len);
715 ok(ret == 0,
716 "Get one data packet for stream %d, offset %d, len %d",
717 first_packet_stream_id, first_packet_offset,
718 first_packet_len);
719
720 ret = detach_viewer_session(session_id);
721 ok(ret == 0, "Detach viewer session");
722
723 ret = list_sessions(&session_id);
724 ok(ret > 0, "List sessions : %d session(s)", ret);
725
726 ret = attach_session(session_id);
727 ok(ret > 0, "Attach to session, %d streams received", ret);
728 end:
729 return exit_status();
730 }
This page took 0.04416 seconds and 4 git commands to generate.