Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 29 additions & 9 deletions src/hdf5/hdf5_writer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -339,6 +339,28 @@ ArraystructContext* HDF5Writer::getDynamicAOS(Context * ctx) {
return NULL;
}

std::unique_ptr<HDF5DataSetHandler> HDF5Writer::openOrCreateNonSliceDataSet(
const std::string& tensorized_path, hid_t* dataset_id, int datatype,
hid_t location, int dim, int* size, int aos_rank,
int* aos_shapes, bool shapes_dataset,
DataEntryContext* dataentry_context) {
auto data_set = std::make_unique<HDF5DataSetHandler>(
true, dataentry_context->getURI());
data_set->setNonSliceMode();
constexpr bool create_chunk_cache = true;
if (H5Lexists(location, tensorized_path.c_str(), H5P_DEFAULT) == 0) {
data_set->create(tensorized_path.c_str(), dataset_id, datatype,
location, dim, size, aos_rank, aos_shapes,
shapes_dataset, create_chunk_cache);
} else {
data_set->open(tensorized_path.c_str(), location, dataset_id, dim,
size, datatype, shapes_dataset, create_chunk_cache,
dataentry_context->getURI(), aos_rank, aos_shapes);
data_set->setCurrentShapesAndExtend(size, aos_shapes);
}
return data_set;
}


void HDF5Writer::write_ND_Data(Context * ctx, std::string & att_name, std::string & timebasename, int datatype, int dim, int *size, void *data)
{
Expand Down Expand Up @@ -414,13 +436,11 @@ void HDF5Writer::write_ND_Data(Context * ctx, std::string & att_name, std::strin
if (slice_mode != SLICE_OP) {

//std::cout << "WRITER NOT IN SLICE MODE!!! " << std::endl;
if (dataset_id < 0) //in global mode (put), data set not yet created
if (dataset_id < 0) //not in this action's opened_data_sets cache -- may still exist on disk from a prior action on the same open pulse
{
std::unique_ptr < HDF5DataSetHandler > dataSetHandler(new HDF5DataSetHandler(true, dec->getURI()));
dataSetHandler->setNonSliceMode();
bool create_chunk_cache = true;
dataSetHandler->create(tensorized_path.c_str(), &dataset_id, datatype, gid, dim, size, AOSRank, arrctx_shapes.data(), false, create_chunk_cache);
data_set = std::move(dataSetHandler);
data_set = openOrCreateNonSliceDataSet(
tensorized_path, &dataset_id, datatype, gid, dim, size,
AOSRank, arrctx_shapes.data(), false, dec);
} else {
data_set = std::move(got->second);
opened_data_sets.erase(got);
Expand Down Expand Up @@ -606,9 +626,9 @@ std::string & timebasename, int timed_AOS_index, const std::vector < int > &curr
}

} else {
data_set->setNonSliceMode();
bool create_chunk_cache = true;
data_set->create(tensorized_path.c_str(), &dataset_id, alconst::integer_data, loc_id, dim, size, AOSRank, aos_shapes.data(), shapes_dataset, create_chunk_cache);
data_set = openOrCreateNonSliceDataSet(
tensorized_path, &dataset_id, alconst::integer_data, loc_id,
dim, size, AOSRank, aos_shapes.data(), shapes_dataset, dec);
}

if (data_set->useBuffering && slice_mode != SLICE_OP) {
Expand Down
5 changes: 5 additions & 0 deletions src/hdf5/hdf5_writer.h
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,11 @@ class HDF5Writer {
int getDynamic_AOS_slices_extension(Context *ctx);
int getDynamic_slices_extension(Context *ctx, int timed_AOS_index, int time_vector_length);
ArraystructContext* getDynamicAOS(Context * ctx);
std::unique_ptr<HDF5DataSetHandler> openOrCreateNonSliceDataSet(
const std::string& tensorized_path, hid_t* dataset_id, int datatype,
hid_t location, int dim, int* size, int aos_rank,
int* aos_shapes, bool shapes_dataset,
DataEntryContext* dataentry_context);

public:

Expand Down
86 changes: 86 additions & 0 deletions tests/contract/test_structured_data.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -367,6 +367,92 @@ TEST_P(EquilibriumSeedMatrix, RoundTripHashMatches) {
EXPECT_EQ(al_close_pulse(pctx, CLOSE_PULSE).code, 0);
}

// A second whole-IDS write on the *same still-open* pulse must succeed and
// overwrite in place, not fail as if the underlying storage were being
// created for the first time (issue #60 follow-up: HDF5Writer::write_ND_Data
// and ::createOrUpdateShapesDataSet decided create-vs-reuse purely from their
// own per-action opened_data_sets cache -- which is cleared at the end of
// every al_begin_global_action/al_end_action pair -- instead of asking the
// file itself via H5Lexists, so any second global WRITE_OP on an already-
// populated open pulse re-attempted H5Dcreate2 on a dataset that already
// existed and threw "Unable to create HDF5 dataset"). Two independent
// al_begin_global_action(WRITE_OP) calls on one pctx reproduce the exact
// shape of the failure regardless of what benchmark or caller triggers it.
TEST_P(EquilibriumSeedMatrix, RepeatedWholeIdsWriteOnSameOpenPulseSucceeds) {
const BackendCase b = GetParam();
if (b.on_disk) base_.make_legacy_tree(pulse_);
const std::string uri = al_contract::build_uri(b.id, base_.str(), pulse_);
ASSERT_FALSE(uri.empty());

int pctx = -1;
AL_ASSERT_OK(al_begin_dataentry_action(uri.c_str(), FORCE_CREATE_PULSE, &pctx));
AL_ASSERT_OK(equilibrium_seed::write(pctx));
AL_ASSERT_OK(equilibrium_seed::write(pctx))
<< "a second whole-IDS write on the same still-open pulse must "
"overwrite in place, not fail as if the dataset didn't exist yet";

std::vector<equilibrium_seed::Obs> obs;
AL_ASSERT_OK(equilibrium_seed::read_back(pctx, &obs));
EXPECT_EQ(equilibrium_seed::canonical_hash(obs),
equilibrium_seed::expected_hash())
<< "repeated writes must never accumulate stale data";

EXPECT_EQ(al_close_pulse(pctx, CLOSE_PULSE).code, 0);
}

// Reproduce the shared benchmark process ordering from issue #60: a complete
// whole-IDS read, followed by the node-by-node read of one time_slice element.
// This keeps the smoke-only HDF5 failure covered without relying on benchmark
// registration order or a standalone filtered process.
TEST_P(EquilibriumSeedMatrix, WholeIdsReadThenTimeSliceElementReadSucceeds) {
const BackendCase b = GetParam();
// ASCII's READ_OP setup exhausts its sequential stream before
// beginReadArraystructAction can recover the AOS size. This is the
// documented AOS read known defect pinned by AosKnownDefects.* above;
// this regression specifically covers the HDF5 benchmark ordering.
if (b.id == ASCII_BACKEND) {
GTEST_SKIP() << "known defect for " << b.name
<< " AOS read — see AosKnownDefects.*";
}
if (b.on_disk) base_.make_legacy_tree(pulse_);
const std::string uri = al_contract::build_uri(b.id, base_.str(), pulse_);
ASSERT_FALSE(uri.empty());

int pctx = -1;
AL_ASSERT_OK(al_begin_dataentry_action(uri.c_str(), FORCE_CREATE_PULSE, &pctx));
AL_ASSERT_OK(equilibrium_seed::write(pctx));

std::vector<equilibrium_seed::Obs> whole_ids;
AL_ASSERT_OK(equilibrium_seed::read_back(pctx, &whole_ids));

int op = -1;
AL_ASSERT_OK(al_begin_global_action(pctx, equilibrium_seed::kIds, "",
READ_OP, &op));
int size = 0;
int aos_ctx = -1;
AL_ASSERT_OK(al_begin_arraystruct_action(op, equilibrium_seed::kAos,
equilibrium_seed::kAosTime,
&size, &aos_ctx));
ASSERT_EQ(size, equilibrium_seed::kNSlices);
AL_ASSERT_OK(al_iterate_over_arraystruct(aos_ctx, size - 1));

std::vector<int> shape;
std::vector<double> data;
auto read_leaf = [&](const char* path, int rank) {
AL_ASSERT_OK(al_contract::read_data<double>(aos_ctx, path, rank,
&shape, &data));
};
read_leaf(equilibrium_seed::kSliceTime, 0);
read_leaf(equilibrium_seed::kPsi, 1);
read_leaf(equilibrium_seed::kIp, 0);
read_leaf(equilibrium_seed::kMeasured, 0);
read_leaf(equilibrium_seed::kWeight, 0);

AL_ASSERT_OK(al_end_action(aos_ctx));
AL_ASSERT_OK(al_end_action(op));
EXPECT_EQ(al_close_pulse(pctx, CLOSE_PULSE).code, 0);
}

// --- structural-hash sensitivity (issue #33 acceptance criteria 2 & 3) ------
// Hermetic proofs that the canonical hash catches the two structural
// corruptions a values-only hash would miss. They operate on the production
Expand Down
Loading