X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fplugins%2Fctf%2Ffs-src%2Ffs.c;h=0530f0152df1e542219310230068049731853006;hb=6d54260a276ca4284cafe05d45e84f6d6cc7bce4;hp=f03a363535b18933ae2d343a812b47d1e74c55b3;hpb=f30762e5afa0e61a609d2d7efe1d36264cf7acfa;p=babeltrace.git diff --git a/src/plugins/ctf/fs-src/fs.c b/src/plugins/ctf/fs-src/fs.c index f03a3635..0530f015 100644 --- a/src/plugins/ctf/fs-src/fs.c +++ b/src/plugins/ctf/fs-src/fs.c @@ -66,11 +66,13 @@ int msg_iter_data_set_current_ds_file(struct ctf_fs_msg_iter_data *msg_iter_data msg_iter_data->ds_file_group->ds_file_infos, msg_iter_data->ds_file_info_index); + /* Destroy the previous ds file. */ ctf_fs_ds_file_destroy(msg_iter_data->ds_file); + + /* Create the new ds file. */ msg_iter_data->ds_file = ctf_fs_ds_file_create( msg_iter_data->ds_file_group->ctf_fs_trace, msg_iter_data->self_msg_iter, - msg_iter_data->msg_iter, msg_iter_data->ds_file_group->stream, ds_file_info->path->str, msg_iter_data->log_level); @@ -78,6 +80,10 @@ int msg_iter_data_set_current_ds_file(struct ctf_fs_msg_iter_data *msg_iter_data ret = -1; } + /* Tell the ctf message iterator to iterate on the new ds file. */ + ctf_msg_iter_set_medops_data(msg_iter_data->msg_iter, + msg_iter_data->ds_file); + return ret; } @@ -92,7 +98,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,11 +108,11 @@ 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( - msg_iter_data->ds_file->msg_iter, + ctf_msg_iter_set_emit_stream_beginning_message( + msg_iter_data->msg_iter, msg_iter_data->ds_file_info_index == 0); - bt_msg_iter_set_emit_stream_end_message( - msg_iter_data->ds_file->msg_iter, + ctf_msg_iter_set_emit_stream_end_message( + msg_iter_data->msg_iter, msg_iter_data->ds_file_info_index == msg_iter_data->ds_file_group->ds_file_infos->len - 1); } @@ -123,7 +129,7 @@ bt_component_class_message_iterator_next_method_status ctf_fs_iterator_next_one( while (true) { bt_message *msg; - status = ctf_fs_ds_file_next(msg_iter_data->ds_file, &msg); + status = ctf_fs_ds_file_next(msg_iter_data->msg_iter, &msg); switch (status) { case BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK: *out_msg = msg; @@ -140,7 +146,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 +179,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 +215,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 +246,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: @@ -277,11 +306,11 @@ 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->self_msg_iter = self_msg_iter; - msg_iter_data->msg_iter = bt_msg_iter_create( + 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 +706,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 +782,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,28 +802,33 @@ 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, - bt_common_get_page_size(log_level) * 8, - ctf_fs_ds_file_medops, NULL, log_level, self_comp); - if (!msg_iter) { - BT_COMP_LOGE_STR("Cannot create a CTF message iterator."); + /* + * Create a temporary ds_file to read some properties about the data + * stream file. + */ + ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, NULL, path, + log_level); + if (!ds_file) { goto error; } - ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, msg_iter, - NULL, path, log_level); - if (!ds_file) { + /* Create a temporary iterator to read the ds_file. */ + msg_iter = ctf_msg_iter_create(ctf_fs_trace->metadata->tc, + bt_common_get_page_size(log_level) * 8, + ctf_fs_ds_file_medops, ds_file, log_level, self_comp, NULL); + if (!msg_iter) { + BT_COMP_LOGE_STR("Cannot create a CTF message iterator."); goto error; } - ret = bt_msg_iter_get_packet_properties(ds_file->msg_iter, &props); + ret = ctf_msg_iter_get_packet_properties(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`).", @@ -783,7 +861,7 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace, goto error; } - index = ctf_fs_ds_file_build_index(ds_file, ds_file_info); + index = ctf_fs_ds_file_build_index(ds_file, ds_file_info, msg_iter); if (!index) { BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE( self_comp, self_comp_class, @@ -875,7 +953,7 @@ end: 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); @@ -1429,30 +1507,18 @@ 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; BT_ASSERT(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, - bt_common_get_page_size(log_level) * 8, ctf_fs_ds_file_medops, - NULL, log_level, self_comp); - if (!msg_iter) { - /* bt_msg_iter_create() logs errors. */ - ret = -1; - goto end; - } - BT_ASSERT(index_entry); BT_ASSERT(index_entry->path); - ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, msg_iter, + ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, NULL, index_entry->path, log_level); if (!ds_file) { BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to create a ctf_fs_ds_file"); @@ -1460,16 +1526,28 @@ int decode_clock_snapshot_after_event(struct ctf_fs_trace *ctf_fs_trace, goto end; } + BT_ASSERT(ctf_fs_trace->metadata); + BT_ASSERT(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, + ds_file, log_level, self_comp, NULL); + if (!msg_iter) { + /* ctf_msg_iter_create() logs errors. */ + ret = -1; + goto end; + } + /* * 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(msg_iter, index_entry->offset); if (iter_status) { - /* bt_msg_iter_seek() logs errors. */ + /* ctf_msg_iter_seek() logs errors. */ ret = -1; goto end; } @@ -1481,13 +1559,13 @@ 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( - ds_file->msg_iter, cs); + iter_status = ctf_msg_iter_curr_packet_first_event_clock_snapshot( + 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( - ds_file->msg_iter, cs); + iter_status = ctf_msg_iter_curr_packet_last_event_clock_snapshot( + msg_iter, cs); break; default: bt_common_abort(); @@ -1512,7 +1590,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;