X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fplugins%2Fctf%2Ffs-src%2Ffs.c;h=64fff672705f68ee4e2304c5739ab392a8721745;hb=1505f33a1a42c691890dff61aea9762b43d1bca2;hp=e92de4bca52a0b4db2593f25862668a6388e0c7a;hpb=98b15851a941e7342b8bb19e265cdc3a40fabfb8;p=babeltrace.git diff --git a/src/plugins/ctf/fs-src/fs.c b/src/plugins/ctf/fs-src/fs.c index e92de4bc..64fff672 100644 --- a/src/plugins/ctf/fs-src/fs.c +++ b/src/plugins/ctf/fs-src/fs.c @@ -69,7 +69,7 @@ int msg_iter_data_set_current_ds_file(struct ctf_fs_msg_iter_data *msg_iter_data ctf_fs_ds_file_destroy(msg_iter_data->ds_file); msg_iter_data->ds_file = ctf_fs_ds_file_create( msg_iter_data->ds_file_group->ctf_fs_trace, - msg_iter_data->pc_msg_iter, + msg_iter_data->self_msg_iter, msg_iter_data->msg_iter, msg_iter_data->ds_file_group->stream, ds_file_info->path->str, @@ -92,7 +92,7 @@ void ctf_fs_msg_iter_data_destroy( ctf_fs_ds_file_destroy(msg_iter_data->ds_file); if (msg_iter_data->msg_iter) { - bt_msg_iter_destroy(msg_iter_data->msg_iter); + ctf_msg_iter_destroy(msg_iter_data->msg_iter); } g_free(msg_iter_data); @@ -102,10 +102,10 @@ static void set_msg_iter_emits_stream_beginning_end_messages( struct ctf_fs_msg_iter_data *msg_iter_data) { - bt_msg_iter_set_emit_stream_beginning_message( + ctf_msg_iter_set_emit_stream_beginning_message( msg_iter_data->ds_file->msg_iter, msg_iter_data->ds_file_info_index == 0); - bt_msg_iter_set_emit_stream_end_message( + ctf_msg_iter_set_emit_stream_end_message( msg_iter_data->ds_file->msg_iter, msg_iter_data->ds_file_info_index == msg_iter_data->ds_file_group->ds_file_infos->len - 1); @@ -140,7 +140,7 @@ bt_component_class_message_iterator_next_method_status ctf_fs_iterator_next_one( } msg_iter_data->ds_file_info_index++; - bt_msg_iter_reset_for_next_stream_file( + ctf_msg_iter_reset_for_next_stream_file( msg_iter_data->msg_iter); set_msg_iter_emits_stream_beginning_end_messages( msg_iter_data); @@ -173,19 +173,29 @@ bt_component_class_message_iterator_next_method_status ctf_fs_iterator_next( bt_message_array_const msgs, uint64_t capacity, uint64_t *count) { - bt_component_class_message_iterator_next_method_status status = - BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK; + bt_component_class_message_iterator_next_method_status status; struct ctf_fs_msg_iter_data *msg_iter_data = bt_self_message_iterator_get_data(iterator); uint64_t i = 0; - while (i < capacity && - status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) { + if (G_UNLIKELY(msg_iter_data->next_saved_error)) { + /* + * Last time we were called, we hit an error but had some + * messages to deliver, so we stashed the error here. Return + * it now. + */ + BT_CURRENT_THREAD_MOVE_ERROR_AND_RESET(msg_iter_data->next_saved_error); + status = msg_iter_data->next_saved_status; + goto end; + } + + do { status = ctf_fs_iterator_next_one(msg_iter_data, &msgs[i]); if (status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) { i++; } - } + } while (i < capacity && + status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK); if (i > 0) { /* @@ -199,10 +209,23 @@ bt_component_class_message_iterator_next_method_status ctf_fs_iterator_next( * called, possibly without any accumulated * message, in which case we'll return it. */ + if (status < 0) { + /* + * Save this error for the next _next call. Assume that + * this component always appends error causes when + * returning an error status code, which will cause the + * current thread error to be non-NULL. + */ + msg_iter_data->next_saved_error = bt_current_thread_take_error(); + BT_ASSERT(msg_iter_data->next_saved_error); + msg_iter_data->next_saved_status = status; + } + *count = i; status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK; } +end: return status; } @@ -217,7 +240,7 @@ int ctf_fs_iterator_reset(struct ctf_fs_msg_iter_data *msg_iter_data) goto end; } - bt_msg_iter_reset(msg_iter_data->msg_iter); + ctf_msg_iter_reset(msg_iter_data->msg_iter); set_msg_iter_emits_stream_beginning_end_messages(msg_iter_data); end: @@ -276,12 +299,12 @@ bt_component_class_message_iterator_initialize_method_status ctf_fs_iterator_ini msg_iter_data->log_level = log_level; msg_iter_data->self_comp = self_comp; - msg_iter_data->pc_msg_iter = self_msg_iter; - msg_iter_data->msg_iter = bt_msg_iter_create( + msg_iter_data->self_msg_iter = self_msg_iter; + msg_iter_data->msg_iter = ctf_msg_iter_create( port_data->ds_file_group->ctf_fs_trace->metadata->tc, bt_common_get_page_size(msg_iter_data->log_level) * 8, ctf_fs_ds_file_medops, NULL, msg_iter_data->log_level, - self_comp); + self_comp, self_msg_iter); if (!msg_iter_data->msg_iter) { BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Cannot create a CTF message iterator."); ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR; @@ -677,24 +700,69 @@ void ds_file_group_insert_ds_file_info_sorted( array_insert(ds_file_group->ds_file_infos, ds_file_info, i); } +static +bool ds_index_entries_equal( + const struct ctf_fs_ds_index_entry *left, + const struct ctf_fs_ds_index_entry *right) +{ + if (left->packet_size != right->packet_size) { + return false; + } + + if (left->timestamp_begin != right->timestamp_begin) { + return false; + } + + if (left->timestamp_end != right->timestamp_end) { + return false; + } + + if (left->packet_seq_num != right->packet_seq_num) { + return false; + } + + return true; +} + +/* + * Insert `entry` into `index`, without duplication. + * + * The entry is inserted only if there isn't an identical entry already. + * + * In any case, the ownership of `entry` is transferred to this function. So if + * the entry is not inserted, it is freed. + */ + static void ds_index_insert_ds_index_entry_sorted( struct ctf_fs_ds_index *index, struct ctf_fs_ds_index_entry *entry) { guint i; + struct ctf_fs_ds_index_entry *other_entry; /* Find the spot where to insert this index entry. */ for (i = 0; i < index->entries->len; i++) { - struct ctf_fs_ds_index_entry *other_entry = - g_ptr_array_index(index->entries, i); + other_entry = g_ptr_array_index(index->entries, i); - if (entry->timestamp_begin_ns < other_entry->timestamp_begin_ns) { + if (entry->timestamp_begin_ns <= other_entry->timestamp_begin_ns) { break; } } - array_insert(index->entries, entry, i); + /* + * Insert the entry only if a duplicate doesn't already exist. + * + * There can be duplicate packets if reading multiple overlapping + * snapshots of the same trace. We then want the index to contain + * a reference to only one copy of that packet. + */ + if (i == index->entries->len || + !ds_index_entries_equal(entry, other_entry)) { + array_insert(index->entries, entry, i); + } else { + g_free(entry); + } } static @@ -708,10 +776,9 @@ void merge_ctf_fs_ds_indexes(struct ctf_fs_ds_index *dest, struct ctf_fs_ds_inde /* * Ownership of the ctf_fs_ds_index_entry is transferred to - * dest. + * ds_index_insert_ds_index_entry_sorted. */ g_ptr_array_index(src->entries, i) = NULL; - ds_index_insert_ds_index_entry_sorted(dest, entry); } } @@ -729,16 +796,16 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace, struct ctf_fs_ds_file *ds_file = NULL; struct ctf_fs_ds_file_info *ds_file_info = NULL; struct ctf_fs_ds_index *index = NULL; - struct bt_msg_iter *msg_iter = NULL; + struct ctf_msg_iter *msg_iter = NULL; struct ctf_stream_class *sc = NULL; - struct bt_msg_iter_packet_properties props; + struct ctf_msg_iter_packet_properties props; bt_logging_level log_level = ctf_fs_trace->log_level; bt_self_component *self_comp = ctf_fs_trace->self_comp; bt_self_component_class *self_comp_class = ctf_fs_trace->self_comp_class; - msg_iter = bt_msg_iter_create(ctf_fs_trace->metadata->tc, + msg_iter = ctf_msg_iter_create(ctf_fs_trace->metadata->tc, bt_common_get_page_size(log_level) * 8, - ctf_fs_ds_file_medops, NULL, log_level, self_comp); + ctf_fs_ds_file_medops, NULL, log_level, self_comp, NULL); if (!msg_iter) { BT_COMP_LOGE_STR("Cannot create a CTF message iterator."); goto error; @@ -750,7 +817,7 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace, goto error; } - ret = bt_msg_iter_get_packet_properties(ds_file->msg_iter, &props); + ret = ctf_msg_iter_get_packet_properties(ds_file->msg_iter, &props); if (ret) { BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class, "Cannot get stream file's first packet's header and context fields (`%s`).", @@ -818,7 +885,8 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace, goto error; } - ds_file_group_insert_ds_file_info_sorted(ds_file_group, ds_file_info); + ds_file_group_insert_ds_file_info_sorted(ds_file_group, + BT_MOVE_REF(ds_file_info)); add_group = true; goto end; @@ -855,7 +923,8 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace, merge_ctf_fs_ds_indexes(ds_file_group->index, index); } - ds_file_group_insert_ds_file_info_sorted(ds_file_group, ds_file_info); + ds_file_group_insert_ds_file_info_sorted(ds_file_group, + BT_MOVE_REF(ds_file_info)); goto end; @@ -870,9 +939,10 @@ end: } ctf_fs_ds_file_destroy(ds_file); + ctf_fs_ds_file_info_destroy(ds_file_info); if (msg_iter) { - bt_msg_iter_destroy(msg_iter); + ctf_msg_iter_destroy(msg_iter); } ctf_fs_ds_index_destroy(index); @@ -1426,9 +1496,9 @@ int decode_clock_snapshot_after_event(struct ctf_fs_trace *ctf_fs_trace, struct ctf_fs_ds_index_entry *index_entry, enum target_event target_event, uint64_t *cs, int64_t *ts_ns) { - enum bt_msg_iter_status iter_status = BT_MSG_ITER_STATUS_OK; + enum ctf_msg_iter_status iter_status = CTF_MSG_ITER_STATUS_OK; struct ctf_fs_ds_file *ds_file = NULL; - struct bt_msg_iter *msg_iter = NULL; + struct ctf_msg_iter *msg_iter = NULL; bt_logging_level log_level = ctf_fs_trace->log_level; bt_self_component *self_comp = ctf_fs_trace->self_comp; int ret = 0; @@ -1437,11 +1507,11 @@ int decode_clock_snapshot_after_event(struct ctf_fs_trace *ctf_fs_trace, BT_ASSERT(ctf_fs_trace->metadata); BT_ASSERT(ctf_fs_trace->metadata->tc); - msg_iter = bt_msg_iter_create(ctf_fs_trace->metadata->tc, + msg_iter = ctf_msg_iter_create(ctf_fs_trace->metadata->tc, bt_common_get_page_size(log_level) * 8, ctf_fs_ds_file_medops, - NULL, log_level, self_comp); + NULL, log_level, self_comp, NULL); if (!msg_iter) { - /* bt_msg_iter_create() logs errors. */ + /* ctf_msg_iter_create() logs errors. */ ret = -1; goto end; } @@ -1461,12 +1531,12 @@ int decode_clock_snapshot_after_event(struct ctf_fs_trace *ctf_fs_trace, * Turn on dry run mode to prevent the creation and usage of Babeltrace * library objects (bt_field, bt_message_*, etc.). */ - bt_msg_iter_set_dry_run(msg_iter, true); + ctf_msg_iter_set_dry_run(msg_iter, true); /* Seek to the beginning of the target packet. */ - iter_status = bt_msg_iter_seek(ds_file->msg_iter, index_entry->offset); + iter_status = ctf_msg_iter_seek(ds_file->msg_iter, index_entry->offset); if (iter_status) { - /* bt_msg_iter_seek() logs errors. */ + /* ctf_msg_iter_seek() logs errors. */ ret = -1; goto end; } @@ -1478,16 +1548,16 @@ int decode_clock_snapshot_after_event(struct ctf_fs_trace *ctf_fs_trace, * the first event. To extract the first event's clock * snapshot. */ - iter_status = bt_msg_iter_curr_packet_first_event_clock_snapshot( + iter_status = ctf_msg_iter_curr_packet_first_event_clock_snapshot( ds_file->msg_iter, cs); break; case LAST_EVENT: /* Decode the packet to extract the last event's clock snapshot. */ - iter_status = bt_msg_iter_curr_packet_last_event_clock_snapshot( + iter_status = ctf_msg_iter_curr_packet_last_event_clock_snapshot( ds_file->msg_iter, cs); break; default: - abort(); + bt_common_abort(); } if (iter_status) { ret = -1; @@ -1509,7 +1579,7 @@ end: ctf_fs_ds_file_destroy(ds_file); } if (msg_iter) { - bt_msg_iter_destroy(msg_iter); + ctf_msg_iter_destroy(msg_iter); } return ret;