Multithread reading and processing when working with HDF5 files

Question

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::moveing 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;
}

Answer 1

user673679’s answer covers pretty much all of the suggestions I thought to make about the actual code as presented. But it doesn’t answer the primary question: “Is there a more efficient way of implementing this multithreading?”

To answer that, I’m going to do a higher-level review, a review not of the actual code, but rather of the design. That’s going to require that I make some guesses and assumptions, because there’s so much code missing, and no real information about what the ultimate goals of the code are supposed be. So this design review is going to be necessarily vague.

A note about the C++ standard library (and particularly the threading stuff)

Before I begin, I want to give some guidance about the C++ standard library, and particularly the threading sub-library. The C++ standard library is very different from the standard libraries of most popular languages. Most languages try to give you a fully-complete, high-level, universally-usable set of libraries as part of their standard library—basically, most languages want you to use their standard libraries for everything that they cover, and only use third-party libraries for stuff that isn’t important enough to be worthy of being part of their standard library. The upshot of that is that you can do most things “out of the box” in those languages—it’s rare to need a third-party library. The downside is that their standard libraries tend to be HUGE… often there’s only a single implementation because it would be too big a task for anyone to try to re-implement the whole thing.

The C++ standard library goes another way. It does not try to be all things to all people. Instead, it’s quite spartan, focusing on providing only a small set of vocabulary types, and key facilities that you need to do many/most things, but are either too hard or simply impossible to roll your own portably. You can use the standard library directly if it happens to solve your problem, but the main goal of the C++ standard library is to give you the tools you need to write good, high-level libraries… not to be those libraries itself.

In other words, you shouldn’t look at the C++ standard library’s threading stuff and say, “okay, everything I need to solve my problem should be in here”. You should think about what high-level tools you need, and then see what the C++ standard library provides to help you build those tools. Or, even better, find a third-party library that’s already done that.

The reason why I explained all this will become clear in a moment.

So let’s get started.

The review

I’d say your intuition—that it’s horribly inefficient to make a half-million promises and futures—is spot on. Of course, if you need to do that, then, well, that’s that; if you need it, you need it, and you just have to accept that it’s gonna take time. But the million-object question here is: do you need those promises and futures?

Promises and futures are an excellent way to model values that will be coming from somewhere at any time, but that you don’t immediately need because you can do other stuff if the values aren’t available. That is, you have a task, and you need a value within that task… but not necessarily immediately—you can do other stuff while waiting for that value.

That… doesn’t really sound like what you’re doing, does it? You’re not really doing a task where you can do other stuff while waiting on a value. What you’re doing is generating values, and you want to be able to work with them as soon as they come available, even if all the values aren’t ready yet.

To me, that sounds like the tool you want is a concurrent queue.

And this is where we circle back to what I said about the C++ standard library. Unfortunately, there is no concurrent queue in the standard library. (Yet! There’s actually one proposed for the future.) You could write one yourself, using the threading stuff in the standard library, or, better, you can use a third-party library. There are a lot out there; Boost has one, for example. There are even lock-free concurrent queues out there, if you like (though they sometimes come with limitations on use, like only a single producer, or they’re less efficient because they do a lot of dynamic allocation).

So what would the code look like if you used a queue? Well, assuming a queue with an interface like the Boost one (which is similar to what’s proposed for future C++ standard library), it might look like this:

// The reader thread function.
//
// Pretty simple. Takes the path and a reference to the queue, and just
// iterates through the items in the path, reading the files, and pushing the
// data to the queue.
//
// Once it's done, closes the queue.
auto reader_thread_func(std::filesystem::path path, queue_t<DataEntry>& entries)
{
    for (auto&& p : std::filesystem::directory_iterator{path})
        entries.push(read(p.path()));

    entries.close();
}

// The main thread.
//
// First, we create the queue.
auto entries = queue_t<DataEntry>{};

// Then we create the reader thread.
auto reader_thread = std::thread{reader_thread_func, "/path/to/files", std::ref(entries)};

// At this point, the reader thread is busy reading the data files in the
// background. The data being read will be coming in entry-by-entry.
//
// So let's start using the entries as they come in:
while (true)
{
    auto entry = DataEntry{};

    if (auto result = entries.wait_pop(entry); result == queue_op_status::success)
    {
        // We just got another entry!
        //
        // Do whatever computation you want with it. I'll just use what you
        // wrote:
        std::this_thread::sleep_for(1s);//simulate some long computation
        //cout << e->energy << endl;
    }
    else if (result == queue_op_status::empty)
    {
        // The queue is empty, but not closed.
        //
        // This means we're processing entries faster than we're reading them.
        // So we have to wait until an entry becomes available. We'll just
        // have this thread give up its time, and then try again.
        std::this_thread::yield();
    }
    else if (result == queue_op_status::closed)
    {
        // The queue is empty, and closed.
        //
        // We're done.
        break;
    }
}

// Clean up.
reader_thread.join();

There’s no error handling above (it wouldn’t be too much to add), but basically, that’s everything you need.

The neat thing about using the right tool for the job is that you usually get additional benefits. In this case, a concurrent queue is really the right tool for the job, and it comes with benefits aplenty. For starters, you no longer have to allocate all the data entries—you can just use them more or less as they come in, and then discard them. That means no need for a half-million element vector. In fact, you can even restrict the maximum number of data entries to read to keep memory use bounded; the reader thread can simply wait until you’ve finished with a data entry and are ready for more. Also, you could even have multiple reader threads—perhaps even a thread pool of reader threads, so you can read like 4 or 8 or more files at a time.

Okay, but I’ve made an assumption here (remember, there’s a lot of guesswork here for me, because you haven’t given enough information). I’m assuming that you don’t actually need all half-million data entry objects at the same time. I’m assuming you can read each entry, then discard it. But if that assumption is wrong—if you have to read the data entries, do some computation on them one-by-one as they come in, and then later do another computation on them altogether, then you can’t simply use a queue (because with a queue, you’d be discarding each entry after you processed it).

But the basic idea still applies. You could use a vector instead of a queue, and do all the concurrency stuff manually. Or, perhaps better, you could make a “concurrent queue view” wrapped around a vector, and use the queue view while the data is coming in, and still have the vector of all entries at the end. What’s “best” depends entirely on the details, and I don’t know the details.

Summary

Promises and futures are the wrong tool for this job. They’re for situations where you’re waiting on a value, but you can do other stuff while you wait. Your situation is that you have a bunch of values coming in, and you want to start using them as soon as possible, without waiting for all the values to finish reading it. You’re not really doing anything else while waiting on each value… you’re just waiting on the next value. That situation usually means you want to use a queue.

Unfortunately, the standard library doesn’t have a concurrent queue (yet). So you either have to roll your own, or use a third-party library. That’s pretty normal when it comes to C++ and its standard library. It doesn’t come with everything built-in, and all the bells and whistles you could ever need. It’s pretty spartan, only providing the bare minimum necessary to that actually useful libraries can be built on top of it.

So get/make a concurrent queue, and then just use one thread (or multiple threads!) to read in the data, while another thread (or, again, multiple threads!) keeps pumping the queue for whatever data is available. You’ll spare yourself from having to allocate a half-million promises and a half-million futures, and get a more flexible and powerful abstraction to boot.

Answer 2

struct DataEntry {
    double energy;
    int n_points;
    double2* position;
    DataEntry(int n):n_points(n) {
        position = new double2[n_points];
    }
    ~DataEntry() {
        delete[] position;
    }
};

Use a std::vector for the positions:

struct DataEntry {
    double energy;
    std::vector<double2> position;
    DataEntry(int n):
        position(n) { }
};

If we need a pointer to the data, we can use position.data().

---

using namespace std;
using namespace porous;

It's best to avoid using whole namespaces like this to avoid name collisions. Since the things being defined are inside the porous namespace, we don't actually need that using statement at all.

---

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;
}   

Is there anything that we particularly want to catch here? I think we could just let any exception escape for the user to handle, rather than carrying on with only a message to cout.

It might be a good idea to move this code out of the constructor and into read_all (and take the path as an argument there). I don't think there's any reason to separate iterating through the directory from the rest of the work that the Reader does.

---

DataEntry* Reader::read(string path) {
    ...
    DataEntry* entry = new DataEntry(coord_dimension);

If we need the dynamic memory allocation, we should be using a smart pointer to make ownership simpler and easier.

Or we could just pass the DataEntry around using move semantics (which is easy if we use a std::vector internally for the positions).

---

void Reader::detach() {
    reader_thread.detach();
}

Since all the work is already complete when this is called, we can join instead of detaching. The join can be done in the Reader destructor so it happens automatically.

---

We should be able to reduce the amount of state and number of functions in the Reader class significantly:

class Reader
{
public:
    
    Reader() = default;
    ~Reader() { if (m_thread.joinable()) m_thread.join(); }
    
    future_entries_t read_all(std::string const& dir_path);
    
private:
    
    std::thread m_thread;
};

namespace
{
    
    std::vector<std::string> list_files_in_dir(std::string const& dir_path);
    
    DataEntry read_file(std::string const& path);
    
    void read_files(promised_entries_t entries, std::vector<std::string> files); // threaded_read...
    
} // unnamed

future_entries_t Reader::read_all(std::string const& dir_path)
{
    auto files = list_files_in_dir(dir_path);
    
    future_entries_t futures;
    promised_entries_t promises;
    for (auto i = std::size_t{ 0 }; i != files.size(); ++i)
    {
        // ... as before
    }
    
    m_thread = std::thread(&read_files, std::move(promises), std::move(files));
    return futures;
}

int main()
{
    Reader r;
    for (auto& future : r.read_all("../../big-graphene/test"))
    {
        auto e = future.get();
        // ...
    }
}

This fetches the list of files in read_all, and passes it to read_files, along with the promises.

An anonymous namespace is used for functions that don't need access to any data inside the Reader class.