projects / babeltrace.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
get_new_streams and multi-session
[babeltrace.git] / formats / lttng-live / lttng-live-functions.c
CommitLineData
4a744367
JD		 1/*
2 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
3 * 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#include <sys/socket.h>
25#include <sys/types.h>
26#include <netinet/in.h>
27#include <netdb.h>
28#include <stdio.h>
29#include <string.h>
30#include <stdlib.h>
31#include <unistd.h>
32#include <errno.h>
33#include <inttypes.h>
34#include <fcntl.h>
35#include <sys/mman.h>
36
37#include <babeltrace/ctf/ctf-index.h>
38
39#include <babeltrace/babeltrace.h>
40#include <babeltrace/ctf/events.h>
41#include <babeltrace/ctf/callbacks.h>
42#include <babeltrace/ctf/iterator.h>
43
44/* for packet_index */
45#include <babeltrace/ctf/types.h>
46
47#include <babeltrace/ctf/metadata.h>
48#include <babeltrace/ctf-text/types.h>
49#include <babeltrace/ctf/events-internal.h>
50#include <formats/ctf/events-private.h>
51
52#include "lttng-live-functions.h"
158eea6e 53#include "lttng-viewer-abi.h"
4a744367
JD		 54
55/*
56 * Memory allocation zeroed
57 */
58#define zmalloc(x) calloc(1, x)
59
60#ifndef max_t
61#define max_t(type, a, b) \
62 ((type) (a) > (type) (b) ? (type) (a) : (type) (b))
63#endif
64
2acdc547
JD		 65void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
66 int whence);
67static void add_traces(gpointer key, gpointer value, gpointer user_data);
68static int del_traces(gpointer key, gpointer value, gpointer user_data);
69
4a744367
JD		 70int lttng_live_connect_viewer(struct lttng_live_ctx *ctx, char *hostname,
71 int port)
72{
73 struct hostent *host;
74 struct sockaddr_in server_addr;
75 int ret;
76
77 host = gethostbyname(hostname);
78 if (!host) {
79 ret = -1;
80 goto end;
81 }
82
83 if ((ctx->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
84 perror("Socket");
85 ret = -1;
86 goto end;
87 }
88
89 server_addr.sin_family = AF_INET;
90 server_addr.sin_port = htons(port);
91 server_addr.sin_addr = *((struct in_addr *) host->h_addr);
92 bzero(&(server_addr.sin_zero), 8);
93
94 if (connect(ctx->control_sock, (struct sockaddr *) &server_addr,
95 sizeof(struct sockaddr)) == -1) {
96 perror("Connect");
97 ret = -1;
98 goto end;
99 }
100
101 ret = 0;
102
103end:
104 return ret;
105}
106
107int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
108{
109 struct lttng_viewer_cmd cmd;
110 struct lttng_viewer_connect connect;
111 int ret;
112 ssize_t ret_len;
113
114 cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
115 cmd.data_size = sizeof(connect);
116 cmd.cmd_version = 0;
117
0c3cd7e1 118 connect.viewer_session_id = -1ULL; /* will be set on recv */
4a744367
JD		 119 connect.major = htobe32(LTTNG_LIVE_MAJOR);
120 connect.minor = htobe32(LTTNG_LIVE_MINOR);
121 connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
122
123 do {
124 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
125 } while (ret_len < 0 && errno == EINTR);
126 if (ret_len < 0) {
127 fprintf(stderr, "[error] Error sending cmd\n");
128 ret = ret_len;
129 goto error;
130 }
131 assert(ret_len == sizeof(cmd));
132
133 do {
134 ret_len = send(ctx->control_sock, &connect, sizeof(connect), 0);
135 } while (ret_len < 0 && errno == EINTR);
136 if (ret_len < 0) {
137 fprintf(stderr, "[error] Error sending version\n");
138 ret = ret_len;
139 goto error;
140 }
141 assert(ret_len == sizeof(connect));
142
143 do {
144 ret_len = recv(ctx->control_sock, &connect, sizeof(connect), 0);
145 } while (ret_len < 0 && errno == EINTR);
146 if (ret_len < 0) {
147 fprintf(stderr, "[error] Error receiving version\n");
148 ret = ret_len;
149 goto error;
150 }
151 assert(ret_len == sizeof(connect));
152
153 printf_verbose("Received viewer session ID : %" PRIu64 "\n",
154 be64toh(connect.viewer_session_id));
155 printf_verbose("Relayd version : %u.%u\n", be32toh(connect.major),
156 be32toh(connect.minor));
157
158 ret = 0;
159
160error:
161 return ret;
162}
163
164int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
165{
166 struct lttng_viewer_cmd cmd;
167 struct lttng_viewer_list_sessions list;
168 struct lttng_viewer_session lsession;
169 int i, ret;
170 ssize_t ret_len;
171 int sessions_count;
172
173 cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
174 cmd.data_size = 0;
175 cmd.cmd_version = 0;
176
177 do {
178 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
179 } while (ret_len < 0 && errno == EINTR);
180 if (ret_len < 0) {
181 fprintf(stderr, "[error] Error sending cmd\n");
182 ret = ret_len;
183 goto error;
184 }
185 assert(ret_len == sizeof(cmd));
186
187 do {
188 ret_len = recv(ctx->control_sock, &list, sizeof(list), 0);
189 } while (ret_len < 0 && errno == EINTR);
190 if (ret_len < 0) {
191 fprintf(stderr, "[error] Error receiving session list\n");
192 ret = ret_len;
193 goto error;
194 }
195 assert(ret_len == sizeof(list));
196
197 sessions_count = be32toh(list.sessions_count);
198 fprintf(stdout, "%u active session(s)%c\n", sessions_count,
199 sessions_count > 0 ? ':' : ' ');
200 for (i = 0; i < sessions_count; i++) {
201 do {
202 ret_len = recv(ctx->control_sock, &lsession, sizeof(lsession), 0);
203 } while (ret_len < 0 && errno == EINTR);
204 if (ret_len < 0) {
205 fprintf(stderr, "[error] Error receiving session\n");
206 ret = ret_len;
207 goto error;
208 }
209 assert(ret_len == sizeof(lsession));
9ed31e53
MD		 210 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
211 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
4a744367
JD		 212
213 fprintf(stdout, "%s/%" PRIu64 " : %s on host %s (timer = %u, "
214 "%u stream(s), %u client(s) connected)\n",
215 path, be64toh(lsession.id),
216 lsession.session_name, lsession.hostname,
217 be32toh(lsession.live_timer),
218 be32toh(lsession.streams),
219 be32toh(lsession.clients));
220 }
221
222 ret = 0;
223
224error:
225 return ret;
226}
227
228int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream,
229 uint64_t ctf_trace_id)
230{
231 struct lttng_live_ctf_trace *trace;
232 int ret = 0;
233
234 trace = g_hash_table_lookup(stream->session->ctf_traces,
235 (gpointer) ctf_trace_id);
236 if (!trace) {
237 trace = g_new0(struct lttng_live_ctf_trace, 1);
4a744367
JD		 238 trace->ctf_trace_id = ctf_trace_id;
239 trace->streams = g_ptr_array_new();
240 g_hash_table_insert(stream->session->ctf_traces,
241 (gpointer) ctf_trace_id,
242 trace);
243 }
244 if (stream->metadata_flag)
245 trace->metadata_stream = stream;
246
247 stream->ctf_trace = trace;
248 g_ptr_array_add(trace->streams, stream);
249
4a744367
JD		 250 return ret;
251}
252
253int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
254{
255 struct lttng_viewer_cmd cmd;
256 struct lttng_viewer_attach_session_request rq;
257 struct lttng_viewer_attach_session_response rp;
258 struct lttng_viewer_stream stream;
259 int ret, i;
260 ssize_t ret_len;
261
262 cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
263 cmd.data_size = sizeof(rq);
264 cmd.cmd_version = 0;
265
266 memset(&rq, 0, sizeof(rq));
267 rq.session_id = htobe64(id);
268 // TODO: add cmd line parameter to select seek beginning
269 // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
270 rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
271
272 do {
273 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
274 } while (ret_len < 0 && errno == EINTR);
275 if (ret_len < 0) {
276 fprintf(stderr, "[error] Error sending cmd\n");
277 ret = ret_len;
278 goto error;
279 }
280 assert(ret_len == sizeof(cmd));
281
282 do {
283 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
284 } while (ret_len < 0 && errno == EINTR);
285 if (ret_len < 0) {
286 fprintf(stderr, "[error] Error sending attach request\n");
287 ret = ret_len;
288 goto error;
289 }
290 assert(ret_len == sizeof(rq));
291
292 do {
293 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
294 } while (ret_len < 0 && errno == EINTR);
295 if (ret_len < 0) {
296 fprintf(stderr, "[error] Error receiving attach response\n");
297 ret = ret_len;
298 goto error;
299 }
300 assert(ret_len == sizeof(rp));
301
302 switch(be32toh(rp.status)) {
303 case LTTNG_VIEWER_ATTACH_OK:
304 break;
305 case LTTNG_VIEWER_ATTACH_UNK:
306 ret = -LTTNG_VIEWER_ATTACH_UNK;
307 goto end;
308 case LTTNG_VIEWER_ATTACH_ALREADY:
309 fprintf(stderr, "[error] Already a viewer attached\n");
310 ret = -1;
311 goto end;
312 case LTTNG_VIEWER_ATTACH_NOT_LIVE:
313 fprintf(stderr, "[error] Not a live session\n");
314 ret = -1;
315 goto end;
316 case LTTNG_VIEWER_ATTACH_SEEK_ERR:
317 fprintf(stderr, "[error] Wrong seek parameter\n");
318 ret = -1;
319 goto end;
320 default:
321 fprintf(stderr, "[error] Unknown attach return code %u\n",
322 be32toh(rp.status));
323 ret = -1;
324 goto end;
325 }
326 if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
327 ret = -1;
328 goto end;
329 }
330
2acdc547 331 ctx->session->stream_count += be32toh(rp.streams_count);
4a744367
JD		 332 /*
333 * When the session is created but not started, we do an active wait
334 * until it starts. It allows the viewer to start processing the trace
335 * as soon as the session starts.
336 */
337 if (ctx->session->stream_count == 0) {
338 ret = 0;
339 goto end;
340 }
341 printf_verbose("Waiting for %" PRIu64 " streams:\n",
342 ctx->session->stream_count);
343 ctx->session->streams = g_new0(struct lttng_live_viewer_stream,
344 ctx->session->stream_count);
4a744367
JD		 345 for (i = 0; i < be32toh(rp.streams_count); i++) {
346 do {
347 ret_len = recv(ctx->control_sock, &stream, sizeof(stream), 0);
348 } while (ret_len < 0 && errno == EINTR);
349 if (ret_len < 0) {
350 fprintf(stderr, "[error] Error receiving stream\n");
351 ret = ret_len;
352 goto error;
353 }
354 assert(ret_len == sizeof(stream));
9ed31e53
MD		 355 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
356 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
4a744367
JD		 357
358 printf_verbose(" stream %" PRIu64 " : %s/%s\n",
359 be64toh(stream.id), stream.path_name,
360 stream.channel_name);
361 ctx->session->streams[i].id = be64toh(stream.id);
362 ctx->session->streams[i].session = ctx->session;
363
364 ctx->session->streams[i].first_read = 1;
365 ctx->session->streams[i].mmap_size = 0;
366
367 if (be32toh(stream.metadata_flag)) {
368 char *path;
369
370 path = strdup(LTTNG_METADATA_PATH_TEMPLATE);
6a4d2b5a
MD		 371 if (!path) {
372 perror("strdup");
373 ret = -1;
374 goto error;
375 }
376 if (!mkdtemp(path)) {
377 perror("mkdtemp");
378 free(path);
379 ret = -1;
380 goto error;
381 }
4a744367 382 ctx->session->streams[i].metadata_flag = 1;
4a744367
JD		 383 snprintf(ctx->session->streams[i].path,
384 sizeof(ctx->session->streams[i].path),
385 "%s/%s", path,
386 stream.channel_name);
387 ret = open(ctx->session->streams[i].path,
388 O_WRONLY | O_CREAT | O_TRUNC,
389 S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
390 if (ret < 0) {
ab9d07f0 391 perror("open");
6a4d2b5a 392 free(path);
4a744367
JD		 393 goto error;
394 }
395 ctx->session->streams[i].fd = ret;
6a4d2b5a 396 free(path);
4a744367
JD		 397 }
398 ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i],
399 be64toh(stream.ctf_trace_id));
400 if (ret < 0) {
401 goto error;
402 }
403
404 }
405 ret = 0;
406
407end:
408error:
409 return ret;
410}
411
2acdc547
JD		 412static
413int ask_new_streams(struct lttng_live_ctx *ctx)
414{
415 int i, ret = 0;
416 uint64_t id;
417
418restart:
419 for (i = 0; i < ctx->session_ids->len; i++) {
420 id = g_array_index(ctx->session_ids, uint64_t, i);
421 ret = lttng_live_get_new_streams(ctx, id);
422 printf_verbose("Asking for new streams returns %d\n",
423 ret);
424 if (ret < 0) {
425 if (ret == -LTTNG_VIEWER_NEW_STREAMS_HUP) {
426 printf_verbose("Session %" PRIu64 " closed\n",
427 id);
428 g_array_remove_index(ctx->session_ids, i);
429 /*
430 * We can't continue iterating on the g_array
431 * after a remove, we have to start again.
432 */
433 goto restart;
434 } else {
435 ret = -1;
436 goto end;
437 }
438 }
439 }
440
441end:
442 return ret;
443}
444
4a744367
JD		 445static
446int get_data_packet(struct lttng_live_ctx *ctx,
447 struct ctf_stream_pos *pos,
448 struct lttng_live_viewer_stream *stream, uint64_t offset,
449 uint64_t len)
450{
451 struct lttng_viewer_cmd cmd;
452 struct lttng_viewer_get_packet rq;
453 struct lttng_viewer_trace_packet rp;
454 ssize_t ret_len;
455 int ret;
456
457 cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
458 cmd.data_size = sizeof(rq);
459 cmd.cmd_version = 0;
460
461 memset(&rq, 0, sizeof(rq));
462 rq.stream_id = htobe64(stream->id);
463 /* Already in big endian. */
464 rq.offset = offset;
465 rq.len = htobe32(len);
466
467 do {
468 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
469 } while (ret_len < 0 && errno == EINTR);
470 if (ret_len < 0) {
471 fprintf(stderr, "[error] Error sending cmd\n");
472 ret = ret_len;
473 goto error;
474 }
475 assert(ret_len == sizeof(cmd));
476
477 do {
478 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
479 } while (ret_len < 0 && errno == EINTR);
480 if (ret_len < 0) {
481 fprintf(stderr, "[error] Error sending get_data_packet request\n");
482 ret = ret_len;
483 goto error;
484 }
485 assert(ret_len == sizeof(rq));
486
487 do {
488 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
489 } while (ret_len < 0 && errno == EINTR);
490 if (ret_len < 0) {
491 fprintf(stderr, "[error] Error receiving data response\n");
492 ret = ret_len;
493 goto error;
494 }
57b9d843
JD		 495 if (ret_len != sizeof(rp)) {
496 fprintf(stderr, "[error] get_data_packet: expected %" PRId64
497 ", received %" PRId64 "\n", ret_len,
498 sizeof(rp));
499 ret = -1;
500 goto error;
501 }
4a744367
JD		 502
503 rp.flags = be32toh(rp.flags);
504
505 switch (be32toh(rp.status)) {
506 case LTTNG_VIEWER_GET_PACKET_OK:
507 len = be32toh(rp.len);
508 printf_verbose("get_data_packet: Ok, packet size : %" PRIu64
509 "\n", len);
510 break;
511 case LTTNG_VIEWER_GET_PACKET_RETRY:
512 printf_verbose("get_data_packet: retry\n");
513 ret = -1;
514 goto end;
515 case LTTNG_VIEWER_GET_PACKET_ERR:
516 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
517 printf_verbose("get_data_packet: new metadata needed\n");
518 ret = 0;
519 goto end;
520 }
2acdc547
JD		 521 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
522 ret = ask_new_streams(ctx);
523 if (ret < 0)
524 goto error;
525 g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
526 ctx->bt_ctx);
527 }
4a744367
JD		 528 fprintf(stderr, "[error] get_data_packet: error\n");
529 ret = -1;
530 goto end;
531 case LTTNG_VIEWER_GET_PACKET_EOF:
532 ret = -2;
533 goto error;
534 default:
535 printf_verbose("get_data_packet: unknown\n");
536 assert(0);
537 ret = -1;
538 goto end;
539 }
540
541 if (len <= 0) {
542 ret = -1;
543 goto end;
544 }
545
546 if (len > stream->mmap_size) {
547 uint64_t new_size;
548
549 new_size = max_t(uint64_t, len, stream->mmap_size << 1);
550 if (pos->base_mma) {
551 /* unmap old base */
552 ret = munmap_align(pos->base_mma);
553 if (ret) {
554 fprintf(stderr, "[error] Unable to unmap old base: %s.\n",
555 strerror(errno));
556 ret = -1;
557 goto error;
558 }
559 pos->base_mma = NULL;
560 }
561 pos->base_mma = mmap_align(new_size,
562 PROT_READ | PROT_WRITE,
563 MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
564 if (pos->base_mma == MAP_FAILED) {
565 fprintf(stderr, "[error] mmap error %s.\n",
566 strerror(errno));
567 pos->base_mma = NULL;
568 ret = -1;
569 goto error;
570 }
571
572 stream->mmap_size = new_size;
573 printf_verbose("Expanding stream mmap size to %" PRIu64 " bytes\n",
574 stream->mmap_size);
575 }
576
577 do {
578 ret_len = recv(ctx->control_sock,
579 mmap_align_addr(pos->base_mma), len,
580 MSG_WAITALL);
581 } while (ret_len < 0 && errno == EINTR);
582 if (ret_len < 0) {
583 fprintf(stderr, "[error] Error receiving trace packet\n");
584 ret = ret_len;
585 goto error;
586 }
587 assert(ret_len == len);
588
589end:
590error:
591 return ret;
592}
593
594/*
595 * Return number of metadata bytes written or a negative value on error.
596 */
597static
598int get_new_metadata(struct lttng_live_ctx *ctx,
599 struct lttng_live_viewer_stream *viewer_stream,
600 uint64_t *metadata_len)
601{
602 uint64_t len = 0;
603 int ret;
604 struct lttng_viewer_cmd cmd;
605 struct lttng_viewer_get_metadata rq;
606 struct lttng_viewer_metadata_packet rp;
607 struct lttng_live_viewer_stream *metadata_stream;
608 char *data = NULL;
609 ssize_t ret_len;
610
611 cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
612 cmd.data_size = sizeof(rq);
613 cmd.cmd_version = 0;
614
615 metadata_stream = viewer_stream->ctf_trace->metadata_stream;
616 rq.stream_id = htobe64(metadata_stream->id);
617
618 do {
619 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
620 } while (ret_len < 0 && errno == EINTR);
621 if (ret_len < 0) {
622 fprintf(stderr, "[error] Error sending cmd\n");
623 ret = ret_len;
624 goto error;
625 }
626 assert(ret_len == sizeof(cmd));
627
628 do {
629 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
630 } while (ret_len < 0 && errno == EINTR);
631 if (ret_len < 0) {
632 fprintf(stderr, "[error] Error sending get_metadata request\n");
633 ret = ret_len;
634 goto error;
635 }
636 assert(ret_len == sizeof(rq));
637
638 do {
639 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
640 } while (ret_len < 0 && errno == EINTR);
641 if (ret_len < 0) {
642 fprintf(stderr, "[error] Error receiving metadata response\n");
643 ret = ret_len;
644 goto error;
645 }
646 assert(ret_len == sizeof(rp));
647
648 switch (be32toh(rp.status)) {
649 case LTTNG_VIEWER_METADATA_OK:
650 printf_verbose("get_metadata : OK\n");
651 break;
652 case LTTNG_VIEWER_NO_NEW_METADATA:
653 printf_verbose("get_metadata : NO NEW\n");
654 ret = -1;
655 goto end;
656 case LTTNG_VIEWER_METADATA_ERR:
657 printf_verbose("get_metadata : ERR\n");
658 ret = -1;
659 goto end;
660 default:
661 printf_verbose("get_metadata : UNKNOWN\n");
662 ret = -1;
663 goto end;
664 }
665
666 len = be64toh(rp.len);
667 printf_verbose("Writing %" PRIu64" bytes to metadata\n", len);
668 if (len <= 0) {
669 ret = -1;
670 goto end;
671 }
672
673 data = zmalloc(len);
674 if (!data) {
675 perror("relay data zmalloc");
676 ret = -1;
677 goto error;
678 }
679 do {
680 ret_len = recv(ctx->control_sock, data, len, MSG_WAITALL);
681 } while (ret_len < 0 && errno == EINTR);
682 if (ret_len < 0) {
683 fprintf(stderr, "[error] Error receiving trace packet\n");
684 ret = ret_len;
685 free(data);
686 goto error;
687 }
688 assert(ret_len == len);
689
690 do {
691 ret_len = write(metadata_stream->fd, data, len);
692 } while (ret_len < 0 && errno == EINTR);
693 if (ret_len < 0) {
694 free(data);
695 ret = ret_len;
696 goto error;
697 }
698 assert(ret_len == len);
699
700 free(data);
701
702 *metadata_len = len;
703 ret = 0;
704end:
705error:
706 return ret;
707}
708
709/*
710 * Get one index for a stream.
711 *
712 * Returns 0 on success or a negative value on error.
713 */
714static
715int get_next_index(struct lttng_live_ctx *ctx,
716 struct lttng_live_viewer_stream *viewer_stream,
717 struct packet_index *index)
718{
719 struct lttng_viewer_cmd cmd;
720 struct lttng_viewer_get_next_index rq;
721 struct lttng_viewer_index rp;
722 int ret;
723 uint64_t metadata_len;
724 ssize_t ret_len;
725
726 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
727 cmd.data_size = sizeof(rq);
728 cmd.cmd_version = 0;
729
730 memset(&rq, 0, sizeof(rq));
731 rq.stream_id = htobe64(viewer_stream->id);
732
733retry:
734 do {
735 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
736 } while (ret_len < 0 && errno == EINTR);
737 if (ret_len < 0) {
738 fprintf(stderr, "[error] Error sending cmd\n");
739 ret = ret_len;
740 goto error;
741 }
742 assert(ret_len == sizeof(cmd));
743
744 do {
745 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
746 } while (ret_len < 0 && errno == EINTR);
747 if (ret_len < 0) {
748 fprintf(stderr, "[error] Error sending get_next_index request\n");
749 ret = ret_len;
750 goto error;
751 }
752 assert(ret_len == sizeof(rq));
753
754 do {
755 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
756 } while (ret_len < 0 && errno == EINTR);
757 if (ret_len < 0) {
758 fprintf(stderr, "[error] Error receiving index response\n");
759 ret = ret_len;
760 goto error;
761 }
762 assert(ret_len == sizeof(rp));
763
764 rp.flags = be32toh(rp.flags);
765
766 switch (be32toh(rp.status)) {
767 case LTTNG_VIEWER_INDEX_INACTIVE:
768 printf_verbose("get_next_index: inactive\n");
769 memset(index, 0, sizeof(struct packet_index));
992e8cc0 770 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
4a744367
JD		 771 break;
772 case LTTNG_VIEWER_INDEX_OK:
773 printf_verbose("get_next_index: Ok, need metadata update : %u\n",
774 rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA);
775 index->offset = be64toh(rp.offset);
776 index->packet_size = be64toh(rp.packet_size);
777 index->content_size = be64toh(rp.content_size);
992e8cc0
MD		 778 index->ts_cycles.timestamp_begin = be64toh(rp.timestamp_begin);
779 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
4a744367
JD		 780 index->events_discarded = be64toh(rp.events_discarded);
781
782 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
783 printf_verbose("get_next_index: new metadata needed\n");
784 ret = get_new_metadata(ctx, viewer_stream,
785 &metadata_len);
786 if (ret < 0) {
787 goto error;
788 }
789 }
2acdc547
JD		 790 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
791 ret = ask_new_streams(ctx);
792 if (ret < 0)
793 goto error;
794 g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
795 ctx->bt_ctx);
796 }
4a744367
JD		 797 break;
798 case LTTNG_VIEWER_INDEX_RETRY:
799 printf_verbose("get_next_index: retry\n");
800 sleep(1);
801 goto retry;
802 case LTTNG_VIEWER_INDEX_HUP:
803 printf_verbose("get_next_index: stream hung up\n");
804 viewer_stream->id = -1ULL;
805 viewer_stream->fd = -1;
806 index->offset = EOF;
2acdc547 807 ctx->session->stream_count--;
4a744367
JD		 808 break;
809 case LTTNG_VIEWER_INDEX_ERR:
810 fprintf(stderr, "[error] get_next_index: error\n");
811 ret = -1;
812 goto error;
813 default:
814 fprintf(stderr, "[error] get_next_index: unkwown value\n");
815 ret = -1;
816 goto error;
817 }
818
819 ret = 0;
820
821error:
822 return ret;
823}
824
825void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
826 int whence)
827{
828 struct ctf_stream_pos *pos;
829 struct ctf_file_stream *file_stream;
f1f52630 830 struct packet_index *prev_index = NULL, *cur_index;
4a744367
JD		 831 struct lttng_live_viewer_stream *viewer_stream;
832 struct lttng_live_session *session;
833 int ret;
834
835retry:
836 pos = ctf_pos(stream_pos);
837 file_stream = container_of(pos, struct ctf_file_stream, pos);
838 viewer_stream = (struct lttng_live_viewer_stream *) pos->priv;
839 session = viewer_stream->session;
840
f1f52630
MD		 841 switch (pos->packet_index->len) {
842 case 0:
843 g_array_set_size(pos->packet_index, 1);
844 cur_index = &g_array_index(pos->packet_index,
845 struct packet_index, 0);
846 break;
847 case 1:
848 g_array_set_size(pos->packet_index, 2);
849 prev_index = &g_array_index(pos->packet_index,
850 struct packet_index, 0);
851 cur_index = &g_array_index(pos->packet_index,
852 struct packet_index, 1);
853 break;
854 case 2:
855 g_array_index(pos->packet_index,
856 struct packet_index, 0) =
857 g_array_index(pos->packet_index,
858 struct packet_index, 1);
859 prev_index = &g_array_index(pos->packet_index,
860 struct packet_index, 0);
861 cur_index = &g_array_index(pos->packet_index,
862 struct packet_index, 1);
863 break;
864 default:
865 abort();
866 break;
867 }
4a744367 868 printf_verbose("get_next_index for stream %" PRIu64 "\n", viewer_stream->id);
f1f52630 869 ret = get_next_index(session->ctx, viewer_stream, cur_index);
4a744367
JD		 870 if (ret < 0) {
871 pos->offset = EOF;
872 fprintf(stderr, "[error] get_next_index failed\n");
873 return;
874 }
875
f1f52630
MD		 876 pos->packet_size = cur_index->packet_size;
877 pos->content_size = cur_index->content_size;
4a744367 878 pos->mmap_base_offset = 0;
f1f52630 879 if (cur_index->offset == EOF) {
4a744367
JD		 880 pos->offset = EOF;
881 } else {
882 pos->offset = 0;
883 }
884
f1f52630
MD		 885 if (cur_index->content_size == 0) {
886 file_stream->parent.cycles_timestamp =
887 cur_index->ts_cycles.timestamp_end;
4a744367 888 file_stream->parent.real_timestamp = ctf_get_real_timestamp(
f1f52630
MD		 889 &file_stream->parent,
890 cur_index->ts_cycles.timestamp_end);
4a744367 891 } else {
f1f52630
MD		 892 /* Convert the timestamps and append to the real_index. */
893 cur_index->ts_real.timestamp_begin = ctf_get_real_timestamp(
894 &file_stream->parent,
895 cur_index->ts_cycles.timestamp_begin);
896 cur_index->ts_real.timestamp_end = ctf_get_real_timestamp(
897 &file_stream->parent,
898 cur_index->ts_cycles.timestamp_end);
899
900 ctf_update_current_packet_index(&file_stream->parent,
901 prev_index, cur_index);
902
903 file_stream->parent.cycles_timestamp =
904 cur_index->ts_cycles.timestamp_begin;
905 file_stream->parent.real_timestamp =
906 cur_index->ts_real.timestamp_begin;
4a744367
JD		 907 }
908
909 if (pos->packet_size == 0 || pos->offset == EOF) {
910 goto end;
911 }
912
913 printf_verbose("get_data_packet for stream %" PRIu64 "\n",
914 viewer_stream->id);
915 ret = get_data_packet(session->ctx, pos, viewer_stream,
f1f52630
MD		 916 be64toh(cur_index->offset),
917 cur_index->packet_size / CHAR_BIT);
4a744367
JD		 918 if (ret == -2) {
919 goto retry;
920 } else if (ret < 0) {
921 pos->offset = EOF;
922 fprintf(stderr, "[error] get_data_packet failed\n");
923 return;
924 }
925
926 printf_verbose("Index received : packet_size : %" PRIu64
927 ", offset %" PRIu64 ", content_size %" PRIu64
928 ", timestamp_end : %" PRIu64 "\n",
f1f52630
MD		 929 cur_index->packet_size, cur_index->offset,
930 cur_index->content_size,
931 cur_index->ts_cycles.timestamp_end);
4a744367
JD		 932
933 /* update trace_packet_header and stream_packet_context */
934 if (pos->prot != PROT_WRITE && file_stream->parent.trace_packet_header) {
935 /* Read packet header */
936 ret = generic_rw(&pos->parent, &file_stream->parent.trace_packet_header->p);
937 if (ret) {
938 pos->offset = EOF;
939 fprintf(stderr, "[error] trace packet header read failed\n");
940 goto end;
941 }
942 }
943 if (pos->prot != PROT_WRITE && file_stream->parent.stream_packet_context) {
944 /* Read packet context */
945 ret = generic_rw(&pos->parent, &file_stream->parent.stream_packet_context->p);
946 if (ret) {
947 pos->offset = EOF;
948 fprintf(stderr, "[error] stream packet context read failed\n");
949 goto end;
950 }
951 }
952 pos->data_offset = pos->offset;
953
954end:
955 return;
956}
957
2acdc547
JD		 958int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx)
959{
960 struct lttng_viewer_cmd cmd;
961 struct lttng_viewer_create_session_response resp;
962 int ret;
963 ssize_t ret_len;
964
965 cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
966 cmd.data_size = 0;
967 cmd.cmd_version = 0;
968
969 do {
970 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
971 } while (ret_len < 0 && errno == EINTR);
972 if (ret_len < 0) {
973 fprintf(stderr, "[error] Error sending cmd\n");
974 ret = ret_len;
975 goto error;
976 }
977 assert(ret_len == sizeof(cmd));
978
979 do {
980 ret_len = recv(ctx->control_sock, &resp, sizeof(resp), 0);
981 } while (ret_len < 0 && errno == EINTR);
982 if (ret_len < 0) {
983 fprintf(stderr, "[error] Error receiving create session reply\n");
984 ret = ret_len;
985 goto error;
986 }
987 assert(ret_len == sizeof(resp));
988
989 if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
990 fprintf(stderr, "[error] Error creating viewer session\n");
991 ret = -1;
992 goto error;
993 }
994 ret = 0;
995
996error:
997 return ret;
998}
999
1000static
1001int del_traces(gpointer key, gpointer value, gpointer user_data)
4a744367
JD		 1002{
1003 struct bt_context *bt_ctx = user_data;
1004 struct lttng_live_ctf_trace *trace = value;
1005 int ret;
1006
1007 ret = bt_context_remove_trace(bt_ctx, trace->trace_id);
1008 if (ret < 0)
1009 fprintf(stderr, "[error] removing trace from context\n");
1010
1011 /* remove the key/value pair from the HT. */
1012 return 1;
1013}
1014
2acdc547
JD		 1015static
1016void add_traces(gpointer key, gpointer value, gpointer user_data)
4a744367
JD		 1017{
1018 int i, ret, total_metadata = 0;
1019 uint64_t metadata_len;
1020 struct bt_context *bt_ctx = user_data;
1021 struct lttng_live_ctf_trace *trace = value;
1022 struct lttng_live_viewer_stream *stream;
1023 struct bt_mmap_stream *new_mmap_stream;
1024 struct bt_mmap_stream_list mmap_list;
1025 struct lttng_live_ctx *ctx = NULL;
1026
2acdc547
JD		 1027 if (trace->in_use)
1028 return;
1029
4a744367
JD		 1030 BT_INIT_LIST_HEAD(&mmap_list.head);
1031
1032 for (i = 0; i < trace->streams->len; i++) {
1033 stream = g_ptr_array_index(trace->streams, i);
1034 ctx = stream->session->ctx;
1035
1036 if (!stream->metadata_flag) {
1037 new_mmap_stream = zmalloc(sizeof(struct bt_mmap_stream));
1038 new_mmap_stream->priv = (void *) stream;
1039 new_mmap_stream->fd = -1;
1040 bt_list_add(&new_mmap_stream->list, &mmap_list.head);
1041 } else {
1042 /* Get all possible metadata before starting */
1043 do {
1044 ret = get_new_metadata(ctx, stream,
1045 &metadata_len);
1046 if (ret == 0) {
1047 total_metadata += metadata_len;
1048 }
1049 } while (ret == 0 || total_metadata == 0);
1050 trace->metadata_fp = fopen(stream->path, "r");
1051 }
1052 }
1053
1054 if (!trace->metadata_fp) {
1055 fprintf(stderr, "[error] No metadata stream opened\n");
1056 goto end_free;
1057 }
1058
1059 ret = bt_context_add_trace(bt_ctx, NULL, "ctf",
1060 ctf_live_packet_seek, &mmap_list, trace->metadata_fp);
1061 if (ret < 0) {
1062 fprintf(stderr, "[error] Error adding trace\n");
1063 goto end_free;
1064 }
2acdc547
JD		 1065
1066 if (bt_ctx->current_iterator) {
1067 struct bt_trace_descriptor *td;
1068 struct bt_trace_handle *handle;
1069
1070 handle = (struct bt_trace_handle *) g_hash_table_lookup(
1071 bt_ctx->trace_handles,
1072 (gpointer) (unsigned long) ret);
1073 td = handle->td;
1074 fprintf(stderr, "Handle : %d, td : %p\n", ret, td);
1075 bt_iter_add_trace(bt_ctx->current_iterator, td);
1076 }
1077
4a744367 1078 trace->trace_id = ret;
2acdc547 1079 trace->in_use = 1;
4a744367
JD		 1080
1081 goto end;
1082
1083end_free:
1084 bt_context_put(bt_ctx);
1085end:
1086 return;
1087}
1088
2acdc547 1089int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
6192b988
JD		 1090{
1091 struct lttng_viewer_cmd cmd;
2acdc547
JD		 1092 struct lttng_viewer_new_streams_request rq;
1093 struct lttng_viewer_new_streams_response rp;
1094 struct lttng_viewer_stream stream;
1095 int ret, i;
6192b988
JD		 1096 ssize_t ret_len;
1097
2acdc547
JD		 1098 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1099 cmd.data_size = sizeof(rq);
6192b988
JD		 1100 cmd.cmd_version = 0;
1101
2acdc547
JD		 1102 memset(&rq, 0, sizeof(rq));
1103 rq.session_id = htobe64(id);
1104
6192b988
JD		 1105 do {
1106 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
1107 } while (ret_len < 0 && errno == EINTR);
1108 if (ret_len < 0) {
1109 fprintf(stderr, "[error] Error sending cmd\n");
1110 ret = ret_len;
1111 goto error;
1112 }
1113 assert(ret_len == sizeof(cmd));
1114
1115 do {
2acdc547 1116 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
6192b988
JD		 1117 } while (ret_len < 0 && errno == EINTR);
1118 if (ret_len < 0) {
2acdc547 1119 fprintf(stderr, "[error] Error sending attach request\n");
6192b988
JD		 1120 ret = ret_len;
1121 goto error;
1122 }
2acdc547 1123 assert(ret_len == sizeof(rq));
6192b988 1124
2acdc547
JD		 1125 do {
1126 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
1127 } while (ret_len < 0 && errno == EINTR);
1128 if (ret_len < 0) {
1129 fprintf(stderr, "[error] Error receiving get_new_streams response\n");
1130 ret = ret_len;
6192b988
JD		 1131 goto error;
1132 }
2acdc547
JD		 1133 assert(ret_len == sizeof(rp));
1134
1135 switch(be32toh(rp.status)) {
1136 case LTTNG_VIEWER_NEW_STREAMS_OK:
1137 break;
1138 case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1139 ret = 0;
1140 goto end;
1141 case LTTNG_VIEWER_NEW_STREAMS_HUP:
1142 ret = -LTTNG_VIEWER_NEW_STREAMS_HUP;
1143 goto end;
1144 case LTTNG_VIEWER_NEW_STREAMS_ERR:
1145 fprintf(stderr, "[error] get_new_streams error\n");
1146 ret = -1;
1147 goto end;
1148 default:
1149 fprintf(stderr, "[error] Unknown return code %u\n",
1150 be32toh(rp.status));
1151 ret = -1;
1152 goto end;
1153 }
1154
1155 ctx->session->stream_count += be32toh(rp.streams_count);
1156 /*
1157 * When the session is created but not started, we do an active wait
1158 * until it starts. It allows the viewer to start processing the trace
1159 * as soon as the session starts.
1160 */
1161 if (ctx->session->stream_count == 0) {
1162 ret = 0;
1163 goto end;
1164 }
1165 printf_verbose("Waiting for %" PRIu64 " streams:\n",
1166 ctx->session->stream_count);
1167 ctx->session->streams = g_new0(struct lttng_live_viewer_stream,
1168 ctx->session->stream_count);
1169 for (i = 0; i < be32toh(rp.streams_count); i++) {
1170 do {
1171 ret_len = recv(ctx->control_sock, &stream, sizeof(stream), 0);
1172 } while (ret_len < 0 && errno == EINTR);
1173 if (ret_len < 0) {
1174 fprintf(stderr, "[error] Error receiving stream\n");
1175 ret = ret_len;
1176 goto error;
1177 }
1178 assert(ret_len == sizeof(stream));
1179 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
1180 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
1181
1182 printf_verbose(" stream %" PRIu64 " : %s/%s\n",
1183 be64toh(stream.id), stream.path_name,
1184 stream.channel_name);
1185 ctx->session->streams[i].id = be64toh(stream.id);
1186 ctx->session->streams[i].session = ctx->session;
1187
1188 ctx->session->streams[i].first_read = 1;
1189 ctx->session->streams[i].mmap_size = 0;
1190
1191 if (be32toh(stream.metadata_flag)) {
1192 char *path;
1193
1194 path = strdup(LTTNG_METADATA_PATH_TEMPLATE);
1195 if (!path) {
1196 perror("strdup");
1197 ret = -1;
1198 goto error;
1199 }
1200 if (!mkdtemp(path)) {
1201 perror("mkdtemp");
1202 free(path);
1203 ret = -1;
1204 goto error;
1205 }
1206 ctx->session->streams[i].metadata_flag = 1;
1207 snprintf(ctx->session->streams[i].path,
1208 sizeof(ctx->session->streams[i].path),
1209 "%s/%s", path,
1210 stream.channel_name);
1211 ret = open(ctx->session->streams[i].path,
1212 O_WRONLY | O_CREAT | O_TRUNC,
1213 S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
1214 if (ret < 0) {
1215 perror("open");
1216 free(path);
1217 goto error;
1218 }
1219 ctx->session->streams[i].fd = ret;
1220 free(path);
1221 }
1222 ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i],
1223 be64toh(stream.ctf_trace_id));
1224 if (ret < 0) {
1225 goto error;
1226 }
1227
1228 }
6192b988
JD		 1229 ret = 0;
1230
2acdc547 1231end:
6192b988
JD		 1232error:
1233 return ret;
1234}
1235
2acdc547 1236void lttng_live_read(struct lttng_live_ctx *ctx)
4a744367 1237{
2acdc547 1238 int ret, i;
4a744367
JD		 1239 struct bt_ctf_iter *iter;
1240 const struct bt_ctf_event *event;
1241 struct bt_iter_pos begin_pos;
1242 struct bt_trace_descriptor *td_write;
1243 struct bt_format *fmt_write;
1244 struct ctf_text_stream_pos *sout;
2acdc547 1245 uint64_t id;
4a744367 1246
2acdc547
JD		 1247 ctx->bt_ctx = bt_context_create();
1248 if (!ctx->bt_ctx) {
4a744367
JD		 1249 fprintf(stderr, "[error] bt_context_create allocation\n");
1250 goto end;
1251 }
1252
1253 fmt_write = bt_lookup_format(g_quark_from_static_string("text"));
1254 if (!fmt_write) {
1255 fprintf(stderr, "[error] ctf-text error\n");
1256 goto end;
1257 }
1258
1259 td_write = fmt_write->open_trace(NULL, O_RDWR, NULL, NULL);
1260 if (!td_write) {
1261 fprintf(stderr, "[error] Error opening output trace\n");
1262 goto end_free;
1263 }
1264
1265 sout = container_of(td_write, struct ctf_text_stream_pos,
1266 trace_descriptor);
1267 if (!sout->parent.event_cb)
1268 goto end_free;
1269
6192b988
JD		 1270 ret = lttng_live_create_viewer_session(ctx);
1271 if (ret < 0) {
1272 goto end_free;
1273 }
1274
2acdc547
JD		 1275 for (i = 0; i < ctx->session_ids->len; i++) {
1276 id = g_array_index(ctx->session_ids, uint64_t, i);
1277 printf_verbose("Attaching to session %lu\n", id);
1278 ret = lttng_live_attach_session(ctx, id);
1279 printf_verbose("Attaching session returns %d\n", ret);
1280 if (ret < 0) {
1281 if (ret == -LTTNG_VIEWER_ATTACH_UNK) {
1282 fprintf(stderr, "[error] Unknown session ID\n");
1283 }
1284 goto end_free;
1285 }
1286 }
1287
4a744367 1288 /*
2acdc547 1289 * As long as the session is active, we try to get new streams.
4a744367 1290 */
2acdc547 1291 for (;;) {
4a744367
JD		 1292 int flags;
1293
2acdc547
JD		 1294 while (!ctx->session->stream_count) {
1295 if (ctx->session_ids->len == 0)
4a744367 1296 goto end_free;
2acdc547
JD		 1297 ret = ask_new_streams(ctx);
1298 if (ret < 0)
1299 goto end_free;
1300 }
4a744367 1301
2acdc547
JD		 1302 g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
1303 ctx->bt_ctx);
4a744367
JD		 1304
1305 begin_pos.type = BT_SEEK_BEGIN;
2acdc547 1306 iter = bt_ctf_iter_create(ctx->bt_ctx, &begin_pos, NULL);
4a744367
JD		 1307 if (!iter) {
1308 fprintf(stderr, "[error] Iterator creation error\n");
1309 goto end;
1310 }
1311 for (;;) {
1312 event = bt_ctf_iter_read_event_flags(iter, &flags);
1313 if (!(flags & BT_ITER_FLAG_RETRY)) {
1314 if (!event) {
1315 /* End of trace */
1316 break;
1317 }
1318 ret = sout->parent.event_cb(&sout->parent,
1319 event->parent->stream);
1320 if (ret) {
1321 fprintf(stderr, "[error] Writing "
1322 "event failed.\n");
1323 goto end_free;
1324 }
1325 }
1326 ret = bt_iter_next(bt_ctf_get_iter(iter));
1327 if (ret < 0) {
1328 goto end_free;
1329 }
1330 }
1331 bt_ctf_iter_destroy(iter);
2acdc547
JD		 1332 g_hash_table_foreach_remove(ctx->session->ctf_traces,
1333 del_traces, ctx->bt_ctx);
1334 ctx->session->stream_count = 0;
1335 }
4a744367
JD		 1336
1337end_free:
2acdc547 1338 bt_context_put(ctx->bt_ctx);
4a744367
JD		 1339end:
1340 return;
1341}
BabelTrace - Trace Format Babel Tower
This page took 0.074048 seconds and 4 git commands to generate.