I have a program that should read and process about 500,000 files in the format hdf5
, each of them containing about 400 data points representing the coordinates of carbon atoms in a sheet of graphene. Since I have an HDD, the process of reading is slow and as such I don't want to delay the reading process as it waits for the computation to finish. My idea was moving the reading process onto a new thread so that new files are read all the time and processed whenever they are available in the main thread.
My implementation of this was spawning an std::thread
for reading and std::move
ing a std::vector<std::promise<DataEntry*>>
of structs I have defined that are then used in the main thread through their respective futures.
Is there a more efficient way of implementing this multithreading? It feels horribly inefficient to have to make 500,000 promises and futures. I suppose it would be easy to delete
the futures once used on the main thread but I don't know how I would go about freeing the promises that are still on the reading thread.
Apart from that, are there any other ways in general of improving my code, especially in terms of performance?
utility.h
#pragma once
#include <vector>
#include <future>
#include <chrono>
#include <string>
#include <iostream>
namespace porous {
struct Timer {
high_resolution_clock::time_point start;
high_resolution_clock::time_point end;
std::string message;
Timer(std::string msg) {
message=msg;
start = high_resolution_clock::now();
}
void now() {
end = high_resolution_clock::now();
int64_t duration = std::chrono::duration_cast<std::chrono::milliseconds>(end-start).count();
std::cout << message << " took " << duration << "ms\n";
}
~Timer() {
end = high_resolution_clock::now();
int64_t duration = std::chrono::duration_cast<std::chrono::milliseconds>(end-start).count();
std::cout << message << " took " << duration << "ms\n";
}
};
struct double2 {
double x,y;
};
struct DataEntry {
double energy;
int n_points;
double2* position;
DataEntry(int n):n_points(n) {
position = new double2[n_points];
}
~DataEntry() {
delete[] position;
}
};
}
typedef std::vector<std::promise<porous::DataEntry*>> promised_entries_t;
typedef std::vector<std::future<porous::DataEntry*>> future_entries_t;
reader.h
#pragma once
#include <string>
#include <filesystem>
#include <chrono>
#include <mutex>
#include <future>
#include <vector>
#include <H5Cpp.h>
#include "utility.h"
using namespace porous;
using namespace H5;
using std::filesystem::directory_iterator;
using std::filesystem::directory_entry;
namespace porous {
class Reader {
private:
std::vector<std::string> files;
std::thread reader_thread;
void threaded_read(promised_entries_t entries);
promised_entries_t entry_promises;
const char* next();
public:
Reader(std::string basedir);
future_entries_t read_all();
DataEntry* read(std::string path); //called from worker thread
void detach();
};
}
reader.cpp
#include "reader.h"
#include "utility.h"
#include <iostream>
namespace fs = std::filesystem;
using namespace std;
using namespace porous;
Reader::Reader(string path) {
try {
Timer t("[Reader] init");
for (auto& entry : fs::directory_iterator(path)) {
files.push_back(entry.path().string());
}
cout << "There are " << files.size() << " to read\n";
} catch (std::exception& e) {
cout << e.what() << endl;
}
}
DataEntry* Reader::read(string path) {
H5File f;
try{
f = H5File(path, H5F_ACC_RDONLY);
} catch(...) {
cout << "this file did not read:\n" << path << endl;
return NULL;
}
DataSet coords_ds = f.openDataSet("coordinates");
DataSpace coords_space = coords_ds.getSpace();
hsize_t dims1[2];
coords_space.getSimpleExtentDims(dims1,NULL);
const int coord_dimension = (int)dims1[0];
DataEntry* entry = new DataEntry(coord_dimension);
const DataSet energy_ds = f.openDataSet("energy");
energy_ds.read(&entry->energy, PredType::NATIVE_DOUBLE);
hsize_t offset[2] = {0,1};
hsize_t count[2] = {coord_dimension,2};
coords_space.selectHyperslab(H5S_SELECT_SET,count,offset);
hsize_t offset_output[2] = {0,0};
hsize_t dim_output[2] = {coord_dimension,2};
DataSpace output_space(2, dim_output);
output_space.selectHyperslab(H5S_SELECT_SET,dim_output,offset_output);
coords_ds.read(entry->position, PredType::NATIVE_DOUBLE, output_space,coords_space);
f.close();
return entry;
}
void Reader::threaded_read(promised_entries_t entries) {
for (int i = 0; i < files.size(); i++) {
entries[i].set_value(this->read(files[i]));
}
}
future_entries_t Reader::read_all() {
future_entries_t zukunft;
for(int i = 0; i < files.size(); i++) {
std::promise<DataEntry*> pr;
zukunft.push_back(pr.get_future());
entry_promises.push_back(std::move(pr));
}
reader_thread = std::thread(&Reader::threaded_read, this, std::move(entry_promises));
return zukunft;
}
void Reader::detach() {
reader_thread.detach();
}
main.cpp
#include <iostream>
#include <string>
#include <chrono>
#include "utility.h"
#include "reader.h"
using namespace porous;
using namespace std;
using namespace std::chrono_literals;
int main() {
using namespace porous;
Reader r("../../big-graphene/test");
future_entries_t dat = r.read_all();
for (auto& fut_dat : dat) {
DataEntry* e = fut_dat.get();
std::this_thread::sleep_for(1s);//simulate some long computation
//cout << e->energy << endl;
delete e;
}
r.detach();
return 0;
}
reader_thread
? How does it work? Does it run each task sequentially? Or does it spawn threads? Use a thread pool? No idea. Try to put together the minimal amount of code that compiles, and maybe someone could help. \$\endgroup\$reader_thread
is just a member variable of the class, but I will add some stubs to make this compile \$\endgroup\$