General multithreaded file processing

Question

I'm trying to write a general multithreaded file processing facility. The idea is that some input file consists of a number of discrete records, each record needs to be processed in the same manner, and be written back out to another file. I would like to multithread this. I've tried to make this as simple as possible, so all the user needs to define is:

• How to get the next record from the input file.
• How to process each record.
• How to write a processed record to the output file.

I have a thread safe queue (taken from C++ Concurrency in action). My basic idea is to use two of these queues, one to buffer the records read by a 'reader' thread, and one to buffer the processed records produced by a number of 'worker' threads, which is then written to a file by a single 'writer' thread. Note this solution does not assume the record order will be maintained.

One problem is how to notify the worker and writer threads of the last record. I've solved this by using a wrapper class around the users input type. It's not pretty, but it works. Any other suggestions are welcome.

I'm looking for any runtime performance optimisations, and general design improvements.

#include <fstream>
#include <vector>
#include <thread>
#include "threadsafe_queue.h"

template<typename T>
struct QueueItem
{
    bool is_sentinel;
    T item;
    QueueItem() {}
    QueueItem(bool is_sentinel_) : is_sentinel(is_sentinel_) {}
    QueueItem(bool is_sentinel_, T item_) : is_sentinel(is_sentinel_), item(item_) {}
};

template<typename TIn, typename FGet>
void reader(threadsafe_queue<QueueItem<TIn>>& queue, const std::string& filename, FGet get_item, unsigned num_worker_threads)
{
    std::ifstream in_file(filename, std::ios::in | std::ios::binary);
    if (in_file) {
        TIn item;
        bool is_finished = false;
        while (!is_finished) {
            is_finished = get_item(in_file, item);
            queue.push(QueueItem<TIn>(false, item));
        }
        for (; num_worker_threads > 0; --num_worker_threads) {
            queue.push(QueueItem<TIn>(true));
        }
    } else {
        throw(errno);
    }
}

template<typename TIn, typename TOut, typename FProcess>
void worker(threadsafe_queue<QueueItem<TIn>>& in_queue, threadsafe_queue<QueueItem<TOut>>& out_queue,
        FProcess process_item)
{
    QueueItem<TIn> queue_item;
    while (true) {
        in_queue.wait_and_pop(queue_item);
        if (queue_item.is_sentinel) {
            out_queue.push(QueueItem<TOut>(true));
            break;
        }
        out_queue.push(QueueItem<TOut>(false, process_item(queue_item.item)));
    }
}

template<typename TOut, typename FWrite>
void writer(threadsafe_queue<QueueItem<TOut>>& queue, const std::string& filename, FWrite write_item)
{
    QueueItem<TOut> queue_item;
    std::ofstream out_file(filename, std::ios::out | std::ios::binary);
    if (out_file) {
        while (true) {
            queue.wait_and_pop(queue_item);
            if (queue_item.is_sentinel) {
                break;
            }
            write_item(out_file, queue_item.item);
        }
    } else {
        throw(errno);
    }
}

template<typename TIn, typename TOut, typename FGet, typename FProcess, typename FWrite>
void process_file(const std::string& in_file, const std::string& out_file, FGet get_item, FProcess process_item, FWrite write_item, const unsigned num_threads)
{
    threadsafe_queue<QueueItem<TIn>> in_queue;
    threadsafe_queue<QueueItem<TOut>> out_queue;

    const unsigned num_worker_threads = num_threads - 2;
    std::vector<std::thread> worker_threads;
    worker_threads.reserve(num_worker_threads);

    std::thread reader_thread(reader<TIn, FGet>, std::ref(in_queue),
                          in_file, get_item, num_worker_threads);
    for (unsigned i = 0; i < num_worker_threads; ++i) {
        worker_threads.push_back(std::thread(worker<TIn, TOut, FProcess>,
                        std::ref(in_queue), std::ref(out_queue), process_item));
    }
    std::thread writer_thread(writer<TOut, FWrite>, std::ref(out_queue), out_file, write_item);

    reader_thread.join();
    for (auto& worker_thread : worker_threads) worker_thread.join();
    writer_thread.join();
}

Here is an example user case:

#include "concurrent_file_process.h"

static char complements[85];
using FastaItem = std::pair<std::string, std::string>;

bool read_fasta_record(std::ifstream& in, FastaItem& record)
{
    return (!std::getline(in, record.first, '\n') || !std::getline(in, record.second, '\n'));
}

FastaItem reverse_complement(FastaItem record)
{
    std::size_t len = record.second.length();
    char c;
    for (std::size_t i = 0, j = len - 1; i < j; ++i, --j) {
        c = complements[record.second[i]];
        record.second[i] = complements[record.second[j]];
        record.second[j] = c;
    }
    return record;
}

void write_fasta_record(std::ofstream& out, const FastaItem& record)
{
    out << record.first << "\n" << record.second << "\n";
}

void reverse_complement_fasta(const std::string& infile, const std::string& outfile)
{
    process_file<FastaItem, FastaItem>(infile, outfile, read_fasta_record, reverse_complement, write_fasta_record, std::thread::hardware_concurrency());
}

int main(int argc, const char * argv[])
{
    complements[65] = 'T';
    complements[67] = 'G';
    complements[71] = 'C';
    complements[84] = 'A';

    if (argc < 3) {
        std::cerr << "Usage: " << argv[0] << " SOURCE DESTINATION" << std::endl;
        return 1;
    }
    reverse_complement_fasta(argv[1], argv[2]);
}

I've tested this against a naive single threaded implementation and I do see some performance gains. However I imagine this method would be better suited to tasks where the processing step was harder (and with a computer with more cores!).

Edit

To make this question more complete, here is the code for the thread safe queue that I'm using. Note this is shamelessly copied from C++ Concurrency in action (minus the emplace method - which I now use instead of push).

#include <queue>
#include <memory>
#include <mutex>
#include <condition_variable>

template<typename T>
class threadsafe_queue
{
public:
    threadsafe_queue() {}
    threadsafe_queue(const threadsafe_queue& other)
    {
        std::lock_guard<std::mutex> lock(other.mut);
        data_queue = other.data_queue;
    }

    void push(T new_value)
    {
        std::lock_guard<std::mutex> lock(mut);
        data_queue.push(new_value);
        data_cond.notify_one();
    }

    template<typename... Args>
    void emplace(Args... args)
    {
        std::lock_guard<std::mutex> lock(mut);
        data_queue.emplace(args...);
        data_cond.notify_one();
    }

    void wait_and_pop(T& value)
    {
        std::unique_lock<std::mutex> lock(mut);
        data_cond.wait(lock, [this]{return !data_queue.empty();});
        value = data_queue.front();
        data_queue.pop();
    }

    std::shared_ptr<T> wait_and_pop()
    {
        std::unique_lock<std::mutex> lock(mut);
        data_cond.wait(lock, [this]{return !data_queue.empty();});
        std::shared_ptr<T> result(std::make_shared<T>(data_queue.front()));
        return result;
    }

    bool try_pop(T& value)
    {
        std::lock_guard<std::mutex> lock(mut);
        if (data_queue.empty()) { return false; }
        value = data_queue.front();
        data_queue.pop();
        return true;
    }

    std::shared_ptr<T> try_pop()
    {
        std::lock_guard<std::mutex> lock(mut);
        if (data_queue.empty()) {
            return std::shared_ptr<T>();
        }
        std::shared_ptr<T> result(std::make_shared<T>(data_queue.front()));
        data_queue.pop();
        return result;
    }

    bool empty() const
    {
        std::lock_guard<std::mutex> lock(mut);
        return data_queue.empty();
    }
private:
    mutable std::mutex mut;
    std::queue<T> data_queue;
    std::condition_variable data_cond;
};

Answer 1

A potential disaster waiting to happen.

static char complements[85];

because

c = complements[record.second[i]];

we don't know that the filesystem hasn't given us some garbage which is outside the range 0-84.

Use std::array for a nice throw if it fails as a basic error handling:

static std::array<char, 85> complements;

This brings us to the next point: where do you handle the throws? Is it safe that the files are not checked for different names?

Regarding performance, the reader and writer should insure that the files have a large buffer associated with them as it improves disk performance. More exotic techniques of reading a block and doing your own search in it can be considered as the repeated calls to std::getline() might not be optimal.

queue.push(QueueItem<TIn>(false, item));

Is the queue full C++11? If so, you should use std::emplace_back() to construct the object directly in the queue, saving some copying.

Which brings us to QueueItem:

template<typename T>
struct QueueItem

The thread safe queue may make too many copies and assigns, and too few moves. The same with QueueItem; move is often better than copy construct or assign.

Process file and friends nearly screams to be made into a class; it is nearly there.