Adding a test which do a sequence of seek BEGIN, LAST, BEGIN, LAST
[babeltrace.git] / lib / iterator.c
CommitLineData
6f3077a2
JD
1/*
2 * iterator.c
3 *
4 * Babeltrace Library
5 *
6 * Copyright 2010-2011 EfficiOS Inc. and Linux Foundation
7 *
8 * Author: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
9 *
10 * Permission is hereby granted, free of charge, to any person obtaining a copy
11 * of this software and associated documentation files (the "Software"), to deal
12 * in the Software without restriction, including without limitation the rights
13 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
14 * copies of the Software, and to permit persons to whom the Software is
15 * furnished to do so, subject to the following conditions:
16 *
17 * The above copyright notice and this permission notice shall be included in
18 * all copies or substantial portions of the Software.
19 */
20
21#include <stdlib.h>
22#include <babeltrace/babeltrace.h>
95d36295 23#include <babeltrace/context.h>
08c22d05 24#include <babeltrace/context-internal.h>
6f3077a2 25#include <babeltrace/iterator-internal.h>
6204d33c 26#include <babeltrace/iterator.h>
6f3077a2 27#include <babeltrace/prio_heap.h>
e4195791 28#include <babeltrace/ctf/metadata.h>
9843982d 29#include <babeltrace/ctf/events.h>
90fcbacc 30#include <inttypes.h>
6f3077a2 31
55c21ea5
JD
32static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
33 const struct bt_iter_pos *begin_pos,
34 unsigned long stream_id);
35
6f3077a2
JD
36struct stream_saved_pos {
37 /*
38 * Use file_stream pointer to check if the trace collection we
39 * restore to match the one we saved from, for each stream.
40 */
41 struct ctf_file_stream *file_stream;
42 size_t cur_index; /* current index in packet index */
43 ssize_t offset; /* offset from base, in bits. EOF for end of file. */
03798a93
JD
44 uint64_t current_real_timestamp;
45 uint64_t current_cycles_timestamp;
6f3077a2
JD
46};
47
e8c92a62 48struct bt_saved_pos {
6f3077a2
JD
49 struct trace_collection *tc;
50 GArray *stream_saved_pos; /* Contains struct stream_saved_pos */
51};
52
53static int stream_read_event(struct ctf_file_stream *sin)
54{
55 int ret;
56
57 ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
58 if (ret == EOF)
59 return EOF;
60 else if (ret) {
3394d22e 61 fprintf(stderr, "[error] Reading event failed.\n");
6f3077a2
JD
62 return ret;
63 }
64 return 0;
65}
66
67/*
68 * returns true if a < b, false otherwise.
69 */
2002e48a 70static int stream_compare(void *a, void *b)
6f3077a2
JD
71{
72 struct ctf_file_stream *s_a = a, *s_b = b;
73
03798a93 74 if (s_a->parent.real_timestamp < s_b->parent.real_timestamp)
6f3077a2
JD
75 return 1;
76 else
77 return 0;
78}
79
90fcbacc
JD
80void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
81{
6cba487f
MD
82 if (!iter_pos)
83 return;
84
5acdc773 85 if (iter_pos->type == BT_SEEK_RESTORE && iter_pos->u.restore) {
6cba487f
MD
86 if (iter_pos->u.restore->stream_saved_pos) {
87 g_array_free(
88 iter_pos->u.restore->stream_saved_pos,
89 TRUE);
90fcbacc 90 }
6cba487f 91 g_free(iter_pos->u.restore);
90fcbacc 92 }
6cba487f 93 g_free(iter_pos);
90fcbacc
JD
94}
95
dc81689e
JD
96/*
97 * seek_file_stream_by_timestamp
98 *
99 * Browse a filestream by index, if an index contains the timestamp passed in
100 * argument, seek inside the corresponding packet it until we find the event we
101 * are looking for (either the exact timestamp or the event just after the
102 * timestamp).
103 *
e32cb1e4
MD
104 * Return 0 if the seek succeded, EOF if we didn't find any packet
105 * containing the timestamp, or a positive integer for error.
f7ed6563
MD
106 *
107 * TODO: this should be turned into a binary search! It is currently
108 * doing a linear search in the packets. This is a O(n) operation on a
109 * very frequent code path.
dc81689e
JD
110 */
111static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
112 uint64_t timestamp)
113{
114 struct ctf_stream_pos *stream_pos;
115 struct packet_index *index;
116 int i, ret;
117
118 stream_pos = &cfs->pos;
03798a93
JD
119 for (i = 0; i < stream_pos->packet_real_index->len; i++) {
120 index = &g_array_index(stream_pos->packet_real_index,
dc81689e 121 struct packet_index, i);
e32cb1e4 122 if (index->timestamp_end < timestamp)
dc81689e
JD
123 continue;
124
20d0dcf9 125 stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
5d2a5af2 126 do {
dc81689e 127 ret = stream_read_event(cfs);
03798a93 128 } while (cfs->parent.real_timestamp < timestamp && ret == 0);
5d2a5af2 129
e32cb1e4
MD
130 /* Can return either EOF, 0, or error (> 0). */
131 return ret;
dc81689e 132 }
e32cb1e4
MD
133 /*
134 * Cannot find the timestamp within the stream packets, return
135 * EOF.
136 */
dc81689e
JD
137 return EOF;
138}
139
140/*
141 * seek_ctf_trace_by_timestamp : for each file stream, seek to the event with
142 * the corresponding timestamp
143 *
144 * Return 0 on success.
145 * If the timestamp is not part of any file stream, return EOF to inform the
e32cb1e4
MD
146 * user the timestamp is out of the scope.
147 * On other errors, return positive value.
dc81689e
JD
148 */
149static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin,
150 uint64_t timestamp, struct ptr_heap *stream_heap)
151{
152 int i, j, ret;
e32cb1e4 153 int found = 0;
dc81689e
JD
154
155 /* for each stream_class */
156 for (i = 0; i < tin->streams->len; i++) {
f380e105 157 struct ctf_stream_declaration *stream_class;
dc81689e
JD
158
159 stream_class = g_ptr_array_index(tin->streams, i);
e32cb1e4
MD
160 if (!stream_class)
161 continue;
dc81689e
JD
162 /* for each file_stream */
163 for (j = 0; j < stream_class->streams->len; j++) {
9e88d150 164 struct ctf_stream_definition *stream;
dc81689e
JD
165 struct ctf_file_stream *cfs;
166
167 stream = g_ptr_array_index(stream_class->streams, j);
e32cb1e4
MD
168 if (!stream)
169 continue;
dc81689e
JD
170 cfs = container_of(stream, struct ctf_file_stream,
171 parent);
172 ret = seek_file_stream_by_timestamp(cfs, timestamp);
173 if (ret == 0) {
174 /* Add to heap */
175 ret = heap_insert(stream_heap, cfs);
e32cb1e4
MD
176 if (ret) {
177 /* Return positive error. */
178 return -ret;
179 }
180 found = 1;
181 } else if (ret > 0) {
182 /*
183 * Error in seek (not EOF), failure.
184 */
185 return ret;
dc81689e 186 }
e32cb1e4 187 /* on EOF just do not put stream into heap. */
dc81689e
JD
188 }
189 }
190
e32cb1e4 191 return found ? 0 : EOF;
dc81689e
JD
192}
193
f7ed6563
MD
194/*
195 * Find timestamp of last event in the stream.
196 *
197 * Return value: 0 if OK, positive error value on error, EOF if no
198 * events were found.
199 */
200static int find_max_timestamp_ctf_file_stream(struct ctf_file_stream *cfs,
201 uint64_t *timestamp_end)
202{
203 int ret, count = 0, i;
204 uint64_t timestamp = 0;
205 struct ctf_stream_pos *stream_pos;
206
207 stream_pos = &cfs->pos;
208 /*
209 * We start by the last packet, and iterate backwards until we
210 * either find at least one event, or we reach the first packet
211 * (some packets can be empty).
212 */
213 for (i = stream_pos->packet_real_index->len - 1; i >= 0; i--) {
214 stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
215 count = 0;
216 /* read each event until we reach the end of the stream */
217 do {
218 ret = stream_read_event(cfs);
219 if (ret == 0) {
220 count++;
221 timestamp = cfs->parent.real_timestamp;
222 }
223 } while (ret == 0);
224
225 /* Error */
226 if (ret > 0)
227 goto end;
228 assert(ret == EOF);
229 if (count)
230 break;
231 }
232
233 if (count) {
234 *timestamp_end = timestamp;
235 ret = 0;
236 } else {
237 /* Return EOF if no events were found */
238 ret = EOF;
239 }
240end:
241 return ret;
242}
243
244/*
245 * Find the stream within a stream class that contains the event with
246 * the largest timestamp, and save that timestamp.
247 *
248 * Return 0 if OK, EOF if no events were found in the streams, or
249 * positive value on error.
250 */
251static int find_max_timestamp_ctf_stream_class(
252 struct ctf_stream_declaration *stream_class,
253 struct ctf_file_stream **cfsp,
254 uint64_t *max_timestamp)
255{
38380786 256 int ret = EOF, i, found = 0;
f7ed6563
MD
257
258 for (i = 0; i < stream_class->streams->len; i++) {
259 struct ctf_stream_definition *stream;
260 struct ctf_file_stream *cfs;
261 uint64_t current_max_ts = 0;
262
263 stream = g_ptr_array_index(stream_class->streams, i);
264 if (!stream)
265 continue;
266 cfs = container_of(stream, struct ctf_file_stream, parent);
267 ret = find_max_timestamp_ctf_file_stream(cfs, &current_max_ts);
268 if (ret == EOF)
269 continue;
270 if (ret != 0)
271 break;
272 if (current_max_ts >= *max_timestamp) {
273 *max_timestamp = current_max_ts;
274 *cfsp = cfs;
38380786 275 found = 1;
f7ed6563
MD
276 }
277 }
278 assert(ret >= 0 || ret == EOF);
38380786
YB
279 if (found) {
280 return 0;
281 }
f7ed6563
MD
282 return ret;
283}
284
285/*
286 * seek_last_ctf_trace_collection: seek trace collection to last event.
287 *
288 * Return 0 if OK, EOF if no events were found, or positive error value
289 * on error.
290 */
291static int seek_last_ctf_trace_collection(struct trace_collection *tc,
292 struct ctf_file_stream **cfsp)
293{
294 int i, j, ret;
295 int found = 0;
296 uint64_t max_timestamp = 0;
297
298 if (!tc)
299 return 1;
300
301 /* For each trace in the trace_collection */
302 for (i = 0; i < tc->array->len; i++) {
303 struct ctf_trace *tin;
304 struct trace_descriptor *td_read;
305
306 td_read = g_ptr_array_index(tc->array, i);
307 if (!td_read)
308 continue;
309 tin = container_of(td_read, struct ctf_trace, parent);
310 /* For each stream_class in the trace */
311 for (j = 0; j < tin->streams->len; j++) {
312 struct ctf_stream_declaration *stream_class;
313
314 stream_class = g_ptr_array_index(tin->streams, j);
315 if (!stream_class)
316 continue;
317 ret = find_max_timestamp_ctf_stream_class(stream_class,
318 cfsp, &max_timestamp);
319 if (ret > 0)
320 goto end;
321 if (ret == 0)
322 found = 1;
323 assert(ret == EOF || ret == 0);
324 }
325 }
326 /*
327 * Now we know in which file stream the last event is located,
328 * and we know its timestamp.
329 */
330 if (!found) {
331 ret = EOF;
332 } else {
333 ret = seek_file_stream_by_timestamp(*cfsp, max_timestamp);
334 assert(ret == 0);
335 }
336end:
337 return ret;
338}
339
90fcbacc
JD
340int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
341{
dc81689e 342 struct trace_collection *tc;
90fcbacc
JD
343 int i, ret;
344
7f89ddce
MD
345 if (!iter || !iter_pos)
346 return -EINVAL;
347
dc81689e
JD
348 switch (iter_pos->type) {
349 case BT_SEEK_RESTORE:
350 if (!iter_pos->u.restore)
7344374e 351 return -EINVAL;
dc81689e 352
90fcbacc 353 heap_free(iter->stream_heap);
90fcbacc
JD
354 ret = heap_init(iter->stream_heap, 0, stream_compare);
355 if (ret < 0)
bed54c92 356 goto error_heap_init;
90fcbacc
JD
357
358 for (i = 0; i < iter_pos->u.restore->stream_saved_pos->len;
359 i++) {
360 struct stream_saved_pos *saved_pos;
361 struct ctf_stream_pos *stream_pos;
9e88d150 362 struct ctf_stream_definition *stream;
90fcbacc
JD
363
364 saved_pos = &g_array_index(
365 iter_pos->u.restore->stream_saved_pos,
366 struct stream_saved_pos, i);
367 stream = &saved_pos->file_stream->parent;
368 stream_pos = &saved_pos->file_stream->pos;
90fcbacc 369
06789ffd 370 stream_pos->packet_seek(&stream_pos->parent,
20d0dcf9 371 saved_pos->cur_index, SEEK_SET);
90fcbacc
JD
372
373 /*
374 * the timestamp needs to be restored after
06789ffd 375 * packet_seek, because this function resets
90fcbacc
JD
376 * the timestamp to the beginning of the packet
377 */
03798a93
JD
378 stream->real_timestamp = saved_pos->current_real_timestamp;
379 stream->cycles_timestamp = saved_pos->current_cycles_timestamp;
90fcbacc 380 stream_pos->offset = saved_pos->offset;
3a25e036 381 stream_pos->last_offset = LAST_OFFSET_POISON;
90fcbacc 382
03798a93
JD
383 stream->prev_real_timestamp = 0;
384 stream->prev_real_timestamp_end = 0;
385 stream->prev_cycles_timestamp = 0;
386 stream->prev_cycles_timestamp_end = 0;
90fcbacc 387
fef3bf22
MD
388 printf_debug("restored to cur_index = %" PRId64 " and "
389 "offset = %" PRId64 ", timestamp = %" PRIu64 "\n",
90fcbacc 390 stream_pos->cur_index,
03798a93 391 stream_pos->offset, stream->real_timestamp);
90fcbacc 392
28fcbaca
MD
393 ret = stream_read_event(saved_pos->file_stream);
394 if (ret != 0) {
395 goto error;
396 }
90fcbacc
JD
397
398 /* Add to heap */
399 ret = heap_insert(iter->stream_heap,
400 saved_pos->file_stream);
401 if (ret)
402 goto error;
403 }
5acdc773 404 return 0;
dc81689e
JD
405 case BT_SEEK_TIME:
406 tc = iter->ctx->tc;
407
dc81689e
JD
408 heap_free(iter->stream_heap);
409 ret = heap_init(iter->stream_heap, 0, stream_compare);
410 if (ret < 0)
bed54c92 411 goto error_heap_init;
dc81689e
JD
412
413 /* for each trace in the trace_collection */
414 for (i = 0; i < tc->array->len; i++) {
415 struct ctf_trace *tin;
416 struct trace_descriptor *td_read;
417
418 td_read = g_ptr_array_index(tc->array, i);
e32cb1e4
MD
419 if (!td_read)
420 continue;
dc81689e
JD
421 tin = container_of(td_read, struct ctf_trace, parent);
422
423 ret = seek_ctf_trace_by_timestamp(tin,
424 iter_pos->u.seek_time,
425 iter->stream_heap);
e32cb1e4
MD
426 /*
427 * Positive errors are failure. Negative value
428 * is EOF (for which we continue with other
429 * traces). 0 is success. Note: on EOF, it just
430 * means that no stream has been added to the
431 * iterator for that trace, which is fine.
432 */
433 if (ret != 0 && ret != EOF)
dc81689e
JD
434 goto error;
435 }
436 return 0;
55c21ea5
JD
437 case BT_SEEK_BEGIN:
438 tc = iter->ctx->tc;
db8a4511
JD
439 heap_free(iter->stream_heap);
440 ret = heap_init(iter->stream_heap, 0, stream_compare);
441 if (ret < 0)
bed54c92 442 goto error_heap_init;
db8a4511 443
55c21ea5
JD
444 for (i = 0; i < tc->array->len; i++) {
445 struct ctf_trace *tin;
446 struct trace_descriptor *td_read;
447 int stream_id;
448
449 td_read = g_ptr_array_index(tc->array, i);
e32cb1e4
MD
450 if (!td_read)
451 continue;
55c21ea5
JD
452 tin = container_of(td_read, struct ctf_trace, parent);
453
454 /* Populate heap with each stream */
455 for (stream_id = 0; stream_id < tin->streams->len;
456 stream_id++) {
f380e105 457 struct ctf_stream_declaration *stream;
55c21ea5
JD
458 int filenr;
459
460 stream = g_ptr_array_index(tin->streams,
461 stream_id);
462 if (!stream)
463 continue;
464 for (filenr = 0; filenr < stream->streams->len;
465 filenr++) {
466 struct ctf_file_stream *file_stream;
467 file_stream = g_ptr_array_index(
468 stream->streams,
469 filenr);
e32cb1e4
MD
470 if (!file_stream)
471 continue;
55c21ea5
JD
472 ret = babeltrace_filestream_seek(
473 file_stream, iter_pos,
474 stream_id);
e32cb1e4
MD
475 if (ret != 0 && ret != EOF) {
476 goto error;
477 }
08ac0e08
MD
478 if (ret == EOF) {
479 /* Do not add EOF streams */
480 continue;
481 }
db8a4511
JD
482 ret = heap_insert(iter->stream_heap, file_stream);
483 if (ret)
484 goto error;
55c21ea5
JD
485 }
486 }
487 }
488 break;
f7ed6563
MD
489 case BT_SEEK_LAST:
490 {
4b8de153 491 struct ctf_file_stream *cfs = NULL;
f7ed6563
MD
492
493 tc = iter->ctx->tc;
494 ret = seek_last_ctf_trace_collection(tc, &cfs);
4b8de153 495 if (ret != 0 || !cfs)
f7ed6563
MD
496 goto error;
497 /* remove all streams from the heap */
498 heap_free(iter->stream_heap);
499 /* Create a new empty heap */
500 ret = heap_init(iter->stream_heap, 0, stream_compare);
501 if (ret < 0)
502 goto error;
503 /* Insert the stream that contains the last event */
504 ret = heap_insert(iter->stream_heap, cfs);
505 if (ret)
506 goto error;
507 break;
508 }
dc81689e
JD
509 default:
510 /* not implemented */
7344374e 511 return -EINVAL;
90fcbacc
JD
512 }
513
514 return 0;
dc81689e 515
90fcbacc
JD
516error:
517 heap_free(iter->stream_heap);
bed54c92 518error_heap_init:
dc81689e
JD
519 if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
520 heap_free(iter->stream_heap);
521 g_free(iter->stream_heap);
522 iter->stream_heap = NULL;
523 ret = -ENOMEM;
524 }
bed54c92 525
dc81689e 526 return ret;
90fcbacc
JD
527}
528
529struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
530{
531 struct bt_iter_pos *pos;
7f89ddce 532 struct trace_collection *tc;
6fabd7af
JD
533 struct ctf_file_stream *file_stream = NULL, *removed;
534 struct ptr_heap iter_heap_copy;
535 int ret;
90fcbacc 536
7f89ddce
MD
537 if (!iter)
538 return NULL;
539
540 tc = iter->ctx->tc;
90fcbacc 541 pos = g_new0(struct bt_iter_pos, 1);
5acdc773 542 pos->type = BT_SEEK_RESTORE;
90fcbacc 543 pos->u.restore = g_new0(struct bt_saved_pos, 1);
90fcbacc
JD
544 pos->u.restore->tc = tc;
545 pos->u.restore->stream_saved_pos = g_array_new(FALSE, TRUE,
546 sizeof(struct stream_saved_pos));
547 if (!pos->u.restore->stream_saved_pos)
548 goto error;
549
6fabd7af
JD
550 ret = heap_copy(&iter_heap_copy, iter->stream_heap);
551 if (ret < 0)
552 goto error_heap;
90fcbacc 553
6fabd7af
JD
554 /* iterate over each stream in the heap */
555 file_stream = heap_maximum(&iter_heap_copy);
556 while (file_stream != NULL) {
557 struct stream_saved_pos saved_pos;
90fcbacc 558
6fabd7af
JD
559 assert(file_stream->pos.last_offset != LAST_OFFSET_POISON);
560 saved_pos.offset = file_stream->pos.last_offset;
561 saved_pos.file_stream = file_stream;
562 saved_pos.cur_index = file_stream->pos.cur_index;
90fcbacc 563
03798a93
JD
564 saved_pos.current_real_timestamp = file_stream->parent.real_timestamp;
565 saved_pos.current_cycles_timestamp = file_stream->parent.cycles_timestamp;
90fcbacc 566
6fabd7af
JD
567 g_array_append_val(
568 pos->u.restore->stream_saved_pos,
569 saved_pos);
90fcbacc 570
6fabd7af
JD
571 printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
572 "offset : %zd, "
573 "timestamp = %" PRIu64 "\n",
574 file_stream->parent.stream_id,
575 saved_pos.cur_index, saved_pos.offset,
03798a93 576 saved_pos.current_real_timestamp);
6fabd7af
JD
577
578 /* remove the stream from the heap copy */
579 removed = heap_remove(&iter_heap_copy);
580 assert(removed == file_stream);
581
582 file_stream = heap_maximum(&iter_heap_copy);
583 }
584 heap_free(&iter_heap_copy);
90fcbacc
JD
585 return pos;
586
6fabd7af
JD
587error_heap:
588 g_array_free(pos->u.restore->stream_saved_pos, TRUE);
90fcbacc 589error:
6fabd7af 590 g_free(pos);
90fcbacc
JD
591 return NULL;
592}
593
dc81689e
JD
594struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *iter,
595 uint64_t timestamp)
596{
597 struct bt_iter_pos *pos;
598
7f89ddce
MD
599 if (!iter)
600 return NULL;
601
dc81689e
JD
602 pos = g_new0(struct bt_iter_pos, 1);
603 pos->type = BT_SEEK_TIME;
604 pos->u.seek_time = timestamp;
605 return pos;
606}
607
6f3077a2
JD
608/*
609 * babeltrace_filestream_seek: seek a filestream to given position.
610 *
611 * The stream_id parameter is only useful for BT_SEEK_RESTORE.
612 */
613static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
e8c92a62 614 const struct bt_iter_pos *begin_pos,
6f3077a2
JD
615 unsigned long stream_id)
616{
617 int ret = 0;
618
7f89ddce
MD
619 if (!file_stream || !begin_pos)
620 return -EINVAL;
621
6f3077a2
JD
622 switch (begin_pos->type) {
623 case BT_SEEK_CUR:
624 /*
625 * just insert into the heap we should already know
626 * the timestamps
627 */
628 break;
629 case BT_SEEK_BEGIN:
06789ffd 630 file_stream->pos.packet_seek(&file_stream->pos.parent,
d6425aaf 631 0, SEEK_SET);
6f3077a2
JD
632 ret = stream_read_event(file_stream);
633 break;
634 case BT_SEEK_TIME:
635 case BT_SEEK_RESTORE:
6f3077a2
JD
636 default:
637 assert(0); /* Not yet defined */
638 }
639
640 return ret;
641}
642
e4195791
MD
643int bt_iter_init(struct bt_iter *iter,
644 struct bt_context *ctx,
04ae3991
MD
645 const struct bt_iter_pos *begin_pos,
646 const struct bt_iter_pos *end_pos)
6f3077a2
JD
647{
648 int i, stream_id;
649 int ret = 0;
6f3077a2 650
7f89ddce
MD
651 if (!iter || !ctx)
652 return -EINVAL;
653
e003e871
JD
654 if (ctx->current_iterator) {
655 ret = -1;
656 goto error_ctx;
657 }
658
6f3077a2 659 iter->stream_heap = g_new(struct ptr_heap, 1);
6f3077a2 660 iter->end_pos = end_pos;
6cba487f 661 bt_context_get(ctx);
95d36295 662 iter->ctx = ctx;
6f3077a2
JD
663
664 ret = heap_init(iter->stream_heap, 0, stream_compare);
665 if (ret < 0)
666 goto error_heap_init;
667
95d36295 668 for (i = 0; i < ctx->tc->array->len; i++) {
6f3077a2
JD
669 struct ctf_trace *tin;
670 struct trace_descriptor *td_read;
671
95d36295 672 td_read = g_ptr_array_index(ctx->tc->array, i);
e32cb1e4
MD
673 if (!td_read)
674 continue;
6f3077a2
JD
675 tin = container_of(td_read, struct ctf_trace, parent);
676
677 /* Populate heap with each stream */
678 for (stream_id = 0; stream_id < tin->streams->len;
679 stream_id++) {
f380e105 680 struct ctf_stream_declaration *stream;
6f3077a2
JD
681 int filenr;
682
683 stream = g_ptr_array_index(tin->streams, stream_id);
684 if (!stream)
685 continue;
686 for (filenr = 0; filenr < stream->streams->len;
687 filenr++) {
688 struct ctf_file_stream *file_stream;
689
690 file_stream = g_ptr_array_index(stream->streams,
691 filenr);
e32cb1e4
MD
692 if (!file_stream)
693 continue;
6f3077a2 694 if (begin_pos) {
b42d4e4e
JD
695 ret = babeltrace_filestream_seek(
696 file_stream,
697 begin_pos,
6f3077a2 698 stream_id);
b42d4e4e
JD
699 } else {
700 struct bt_iter_pos pos;
701 pos.type = BT_SEEK_BEGIN;
702 ret = babeltrace_filestream_seek(
703 file_stream, &pos,
704 stream_id);
705 }
706 if (ret == EOF) {
707 ret = 0;
708 continue;
709 } else if (ret) {
710 goto error;
6f3077a2
JD
711 }
712 /* Add to heap */
713 ret = heap_insert(iter->stream_heap, file_stream);
714 if (ret)
715 goto error;
716 }
717 }
718 }
719
e003e871 720 ctx->current_iterator = iter;
e4195791 721 return 0;
6f3077a2
JD
722
723error:
724 heap_free(iter->stream_heap);
725error_heap_init:
726 g_free(iter->stream_heap);
bed54c92 727 iter->stream_heap = NULL;
e003e871 728error_ctx:
e4195791 729 return ret;
6f3077a2
JD
730}
731
e4195791 732struct bt_iter *bt_iter_create(struct bt_context *ctx,
04ae3991
MD
733 const struct bt_iter_pos *begin_pos,
734 const struct bt_iter_pos *end_pos)
e4195791
MD
735{
736 struct bt_iter *iter;
737 int ret;
738
7f89ddce
MD
739 if (!ctx)
740 return NULL;
741
e4195791
MD
742 iter = g_new0(struct bt_iter, 1);
743 ret = bt_iter_init(iter, ctx, begin_pos, end_pos);
744 if (ret) {
745 g_free(iter);
746 return NULL;
747 }
748 return iter;
749}
750
751void bt_iter_fini(struct bt_iter *iter)
6f3077a2 752{
7f89ddce 753 assert(iter);
dc81689e
JD
754 if (iter->stream_heap) {
755 heap_free(iter->stream_heap);
756 g_free(iter->stream_heap);
757 }
e003e871 758 iter->ctx->current_iterator = NULL;
95d36295 759 bt_context_put(iter->ctx);
e4195791 760}
95d36295 761
e4195791
MD
762void bt_iter_destroy(struct bt_iter *iter)
763{
7f89ddce 764 assert(iter);
e4195791 765 bt_iter_fini(iter);
90fcbacc 766 g_free(iter);
6f3077a2
JD
767}
768
e8c92a62 769int bt_iter_next(struct bt_iter *iter)
6f3077a2
JD
770{
771 struct ctf_file_stream *file_stream, *removed;
772 int ret;
773
7f89ddce
MD
774 if (!iter)
775 return -EINVAL;
776
6f3077a2
JD
777 file_stream = heap_maximum(iter->stream_heap);
778 if (!file_stream) {
779 /* end of file for all streams */
780 ret = 0;
781 goto end;
782 }
783
784 ret = stream_read_event(file_stream);
785 if (ret == EOF) {
786 removed = heap_remove(iter->stream_heap);
787 assert(removed == file_stream);
788 ret = 0;
789 goto end;
790 } else if (ret) {
791 goto end;
792 }
793 /* Reinsert the file stream into the heap, and rebalance. */
794 removed = heap_replace_max(iter->stream_heap, file_stream);
795 assert(removed == file_stream);
796
797end:
798 return ret;
799}
This page took 0.06003 seconds and 4 git commands to generate.