*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
*/
#include <stdlib.h>
#include <babeltrace/babeltrace.h>
+#include <babeltrace/babeltrace-internal.h>
#include <babeltrace/context.h>
#include <babeltrace/context-internal.h>
#include <babeltrace/iterator-internal.h>
ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
if (ret == EOF)
return EOF;
+ else if (ret == EAGAIN)
+ /* Stream is inactive for now (live reading). */
+ return EAGAIN;
else if (ret) {
fprintf(stderr, "[error] Reading event failed.\n");
return ret;
}
+
return 0;
}
/*
- * returns true if a < b, false otherwise.
+ * Return true if a < b, false otherwise.
+ * If time stamps are exactly the same, compare by stream path. This
+ * ensures we get the same result between runs on the same trace
+ * collection on different environments.
+ * The result will be random for memory-mapped traces since there is no
+ * fixed path leading to those (they have empty path string).
*/
static int stream_compare(void *a, void *b)
{
struct ctf_file_stream *s_a = a, *s_b = b;
- if (s_a->parent.real_timestamp < s_b->parent.real_timestamp)
+ if (s_a->parent.real_timestamp < s_b->parent.real_timestamp) {
return 1;
- else
+ } else if (likely(s_a->parent.real_timestamp > s_b->parent.real_timestamp)) {
return 0;
+ } else {
+ return strcmp(s_a->parent.path, s_b->parent.path);
+ }
}
void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
int i, ret;
stream_pos = &cfs->pos;
- for (i = 0; i < stream_pos->packet_real_index->len; i++) {
- index = &g_array_index(stream_pos->packet_real_index,
+ for (i = 0; i < stream_pos->packet_index->len; i++) {
+ index = &g_array_index(stream_pos->packet_index,
struct packet_index, i);
- if (index->timestamp_end < timestamp)
+ if (index->ts_real.timestamp_end < timestamp)
continue;
stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
{
int i, j, ret;
int found = 0;
+ struct bt_trace_descriptor *td = &tin->parent;
+
+ if (td->interval_set) {
+ /*
+ * If this trace has an interval selected, don't allow seeks
+ * before the selected interval. We seek to the start of the
+ * interval, thereby presenting a shorter "virtual" trace.
+ */
+ timestamp = max(timestamp, td->interval_real.timestamp_begin);
+ }
/* for each stream_class */
for (i = 0; i < tin->streams->len; i++) {
ret = seek_file_stream_by_timestamp(cfs, timestamp);
if (ret == 0) {
/* Add to heap */
- ret = heap_insert(stream_heap, cfs);
+ ret = bt_heap_insert(stream_heap, cfs);
if (ret) {
/* Return positive error. */
return -ret;
* either find at least one event, or we reach the first packet
* (some packets can be empty).
*/
- for (i = stream_pos->packet_real_index->len - 1; i >= 0; i--) {
+ for (i = stream_pos->packet_index->len - 1; i >= 0; i--) {
stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
count = 0;
/* read each event until we reach the end of the stream */
struct ctf_file_stream **cfsp,
uint64_t *max_timestamp)
{
- int ret = EOF, i;
+ int ret = EOF, i, found = 0;
for (i = 0; i < stream_class->streams->len; i++) {
struct ctf_stream_definition *stream;
if (current_max_ts >= *max_timestamp) {
*max_timestamp = current_max_ts;
*cfsp = cfs;
+ found = 1;
}
}
assert(ret >= 0 || ret == EOF);
+ if (found) {
+ return 0;
+ }
return ret;
}
/* For each trace in the trace_collection */
for (i = 0; i < tc->array->len; i++) {
struct ctf_trace *tin;
- struct trace_descriptor *td_read;
+ struct bt_trace_descriptor *td_read;
td_read = g_ptr_array_index(tc->array, i);
if (!td_read)
if (!iter_pos->u.restore)
return -EINVAL;
- heap_free(iter->stream_heap);
- ret = heap_init(iter->stream_heap, 0, stream_compare);
+ bt_heap_free(iter->stream_heap);
+ ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
if (ret < 0)
goto error_heap_init;
stream_pos->offset = saved_pos->offset;
stream_pos->last_offset = LAST_OFFSET_POISON;
- stream->prev_real_timestamp = 0;
- stream->prev_real_timestamp_end = 0;
- stream->prev_cycles_timestamp = 0;
- stream->prev_cycles_timestamp_end = 0;
+ stream->current.real.begin = 0;
+ stream->current.real.end = 0;
+ stream->current.cycles.begin = 0;
+ stream->current.cycles.end = 0;
+
+ stream->prev.real.begin = 0;
+ stream->prev.real.end = 0;
+ stream->prev.cycles.begin = 0;
+ stream->prev.cycles.end = 0;
- printf_debug("restored to cur_index = %zd and "
- "offset = %zd, timestamp = %" PRIu64 "\n",
+ printf_debug("restored to cur_index = %" PRId64 " and "
+ "offset = %" PRId64 ", timestamp = %" PRIu64 "\n",
stream_pos->cur_index,
stream_pos->offset, stream->real_timestamp);
}
/* Add to heap */
- ret = heap_insert(iter->stream_heap,
+ ret = bt_heap_insert(iter->stream_heap,
saved_pos->file_stream);
if (ret)
goto error;
case BT_SEEK_TIME:
tc = iter->ctx->tc;
- heap_free(iter->stream_heap);
- ret = heap_init(iter->stream_heap, 0, stream_compare);
+ bt_heap_free(iter->stream_heap);
+ ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
if (ret < 0)
goto error_heap_init;
/* for each trace in the trace_collection */
for (i = 0; i < tc->array->len; i++) {
struct ctf_trace *tin;
- struct trace_descriptor *td_read;
+ struct bt_trace_descriptor *td_read;
td_read = g_ptr_array_index(tc->array, i);
if (!td_read)
return 0;
case BT_SEEK_BEGIN:
tc = iter->ctx->tc;
- heap_free(iter->stream_heap);
- ret = heap_init(iter->stream_heap, 0, stream_compare);
+ bt_heap_free(iter->stream_heap);
+ ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
if (ret < 0)
goto error_heap_init;
for (i = 0; i < tc->array->len; i++) {
struct ctf_trace *tin;
- struct trace_descriptor *td_read;
+ struct bt_trace_descriptor *td_read;
int stream_id;
td_read = g_ptr_array_index(tc->array, i);
if (ret != 0 && ret != EOF) {
goto error;
}
- ret = heap_insert(iter->stream_heap, file_stream);
+ if (ret == EOF) {
+ /* Do not add EOF streams */
+ continue;
+ }
+ ret = bt_heap_insert(iter->stream_heap, file_stream);
if (ret)
goto error;
}
if (ret != 0 || !cfs)
goto error;
/* remove all streams from the heap */
- heap_free(iter->stream_heap);
+ bt_heap_free(iter->stream_heap);
/* Create a new empty heap */
- ret = heap_init(iter->stream_heap, 0, stream_compare);
+ ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
if (ret < 0)
goto error;
/* Insert the stream that contains the last event */
- ret = heap_insert(iter->stream_heap, cfs);
+ ret = bt_heap_insert(iter->stream_heap, cfs);
if (ret)
goto error;
break;
return 0;
error:
- heap_free(iter->stream_heap);
+ bt_heap_free(iter->stream_heap);
error_heap_init:
- if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
- heap_free(iter->stream_heap);
+ if (bt_heap_init(iter->stream_heap, 0, stream_compare) < 0) {
+ bt_heap_free(iter->stream_heap);
g_free(iter->stream_heap);
iter->stream_heap = NULL;
ret = -ENOMEM;
if (!pos->u.restore->stream_saved_pos)
goto error;
- ret = heap_copy(&iter_heap_copy, iter->stream_heap);
+ ret = bt_heap_copy(&iter_heap_copy, iter->stream_heap);
if (ret < 0)
goto error_heap;
/* iterate over each stream in the heap */
- file_stream = heap_maximum(&iter_heap_copy);
+ file_stream = bt_heap_maximum(&iter_heap_copy);
while (file_stream != NULL) {
struct stream_saved_pos saved_pos;
saved_pos.current_real_timestamp);
/* remove the stream from the heap copy */
- removed = heap_remove(&iter_heap_copy);
+ removed = bt_heap_remove(&iter_heap_copy);
assert(removed == file_stream);
- file_stream = heap_maximum(&iter_heap_copy);
+ file_stream = bt_heap_maximum(&iter_heap_copy);
}
- heap_free(&iter_heap_copy);
+ bt_heap_free(&iter_heap_copy);
return pos;
error_heap:
return NULL;
}
-struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *iter,
+struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *unused,
uint64_t timestamp)
{
struct bt_iter_pos *pos;
- if (!iter)
- return NULL;
-
pos = g_new0(struct bt_iter_pos, 1);
pos->type = BT_SEEK_TIME;
pos->u.seek_time = timestamp;
return ret;
}
+int bt_iter_add_trace(struct bt_iter *iter,
+ struct bt_trace_descriptor *td_read)
+{
+ struct ctf_trace *tin;
+ int stream_id, ret = 0;
+
+ tin = container_of(td_read, struct ctf_trace, parent);
+
+ /* Populate heap with each stream */
+ for (stream_id = 0; stream_id < tin->streams->len;
+ stream_id++) {
+ struct ctf_stream_declaration *stream;
+ int filenr;
+
+ stream = g_ptr_array_index(tin->streams, stream_id);
+ if (!stream)
+ continue;
+ for (filenr = 0; filenr < stream->streams->len;
+ filenr++) {
+ struct ctf_file_stream *file_stream;
+ struct bt_iter_pos pos;
+
+ file_stream = g_ptr_array_index(stream->streams,
+ filenr);
+ if (!file_stream)
+ continue;
+
+ pos.type = BT_SEEK_BEGIN;
+ ret = babeltrace_filestream_seek(file_stream,
+ &pos, stream_id);
+
+ if (ret == EOF) {
+ ret = 0;
+ continue;
+ } else if (ret != 0 && ret != EAGAIN) {
+ goto error;
+ }
+ /* Add to heap */
+ ret = bt_heap_insert(iter->stream_heap, file_stream);
+ if (ret)
+ goto error;
+ }
+ }
+
+error:
+ return ret;
+}
+
int bt_iter_init(struct bt_iter *iter,
struct bt_context *ctx,
const struct bt_iter_pos *begin_pos,
const struct bt_iter_pos *end_pos)
{
- int i, stream_id;
+ int i;
int ret = 0;
- if (!iter || !ctx)
+ if (!iter || !ctx || !ctx->tc || !ctx->tc->array)
return -EINVAL;
if (ctx->current_iterator) {
bt_context_get(ctx);
iter->ctx = ctx;
- ret = heap_init(iter->stream_heap, 0, stream_compare);
+ ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
if (ret < 0)
goto error_heap_init;
for (i = 0; i < ctx->tc->array->len; i++) {
- struct ctf_trace *tin;
- struct trace_descriptor *td_read;
+ struct bt_trace_descriptor *td_read;
td_read = g_ptr_array_index(ctx->tc->array, i);
if (!td_read)
continue;
- tin = container_of(td_read, struct ctf_trace, parent);
-
- /* Populate heap with each stream */
- for (stream_id = 0; stream_id < tin->streams->len;
- stream_id++) {
- struct ctf_stream_declaration *stream;
- int filenr;
-
- stream = g_ptr_array_index(tin->streams, stream_id);
- if (!stream)
- continue;
- for (filenr = 0; filenr < stream->streams->len;
- filenr++) {
- struct ctf_file_stream *file_stream;
+ ret = bt_iter_add_trace(iter, td_read);
+ if (ret < 0)
+ goto error;
+ }
- file_stream = g_ptr_array_index(stream->streams,
- filenr);
- if (!file_stream)
- continue;
- if (begin_pos) {
- ret = babeltrace_filestream_seek(
- file_stream,
- begin_pos,
- stream_id);
- } else {
- struct bt_iter_pos pos;
- pos.type = BT_SEEK_BEGIN;
- ret = babeltrace_filestream_seek(
- file_stream, &pos,
- stream_id);
- }
- if (ret == EOF) {
- ret = 0;
- continue;
- } else if (ret) {
- goto error;
- }
- /* Add to heap */
- ret = heap_insert(iter->stream_heap, file_stream);
- if (ret)
- goto error;
- }
+ ctx->current_iterator = iter;
+ if (begin_pos && begin_pos->type != BT_SEEK_BEGIN) {
+ ret = bt_iter_set_pos(iter, begin_pos);
+ if (ret) {
+ goto error;
}
}
- ctx->current_iterator = iter;
- return 0;
+ return ret;
error:
- heap_free(iter->stream_heap);
+ bt_heap_free(iter->stream_heap);
error_heap_init:
g_free(iter->stream_heap);
iter->stream_heap = NULL;
{
assert(iter);
if (iter->stream_heap) {
- heap_free(iter->stream_heap);
+ bt_heap_free(iter->stream_heap);
g_free(iter->stream_heap);
}
iter->ctx->current_iterator = NULL;
{
struct ctf_file_stream *file_stream, *removed;
int ret;
+ bool event_outside_interval = false;
if (!iter)
return -EINVAL;
- file_stream = heap_maximum(iter->stream_heap);
+ file_stream = bt_heap_maximum(iter->stream_heap);
if (!file_stream) {
/* end of file for all streams */
ret = 0;
}
ret = stream_read_event(file_stream);
- if (ret == EOF) {
- removed = heap_remove(iter->stream_heap);
+ if (file_stream->pos.parent.trace->interval_set) {
+ event_outside_interval =
+ file_stream->parent.real_timestamp >
+ file_stream->pos.parent.trace->interval_real.timestamp_end;
+ }
+ if (ret == EOF || event_outside_interval) {
+ removed = bt_heap_remove(iter->stream_heap);
assert(removed == file_stream);
ret = 0;
goto end;
+ } else if (ret == EAGAIN) {
+ /*
+ * Live streaming: the stream is inactive for now, we
+ * just updated the timestamp_end to skip over this
+ * stream up to a certain point in time.
+ *
+ * Since we can't guarantee that a stream will ever have
+ * any activity, we can't rely on the fact that
+ * bt_iter_next will be called for each stream and deal
+ * with inactive streams. So instead, we return 0 here
+ * to the caller and let the read API handle the
+ * retry case.
+ */
+ ret = 0;
+ goto reinsert;
} else if (ret) {
goto end;
}
+
+reinsert:
/* Reinsert the file stream into the heap, and rebalance. */
- removed = heap_replace_max(iter->stream_heap, file_stream);
+ removed = bt_heap_replace_max(iter->stream_heap, file_stream);
assert(removed == file_stream);
-
end:
return ret;
}
projects / babeltrace.git / blobdiff