5
\$\begingroup\$

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;
};
\$\endgroup\$
  • \$\begingroup\$ Is the implementation of process_item CPU intensive? If not, then it's unlikely to see big improvements by multithreading, because I/O will be the bottleneck. The more CPU intensive process_item, the more visible the improvement. \$\endgroup\$ – janos Sep 6 '14 at 13:31
  • \$\begingroup\$ Depends on the use case. I mention this at the end of my question. In the example I provide (reverse complementing a Fasta file), the processing step isn't particularly complex, but I still get a small runtime performance gain over a single threaded approach on my 4 core machine. But I imagine this would indeed be more useful in attacking more complex tasks (optimal local alignment comes to mind), and for use on clusters etc. \$\endgroup\$ – Daniel Sep 6 '14 at 18:05
  • \$\begingroup\$ I tested on something a bit more taxing (Smith-Waterman local alignment) on the same smallish Fasta input (aligned against a pre-defined sequence). Single-threaded ~ 64secs, multi-threaded ~ 40secs. Using 4 cores, so only 2 cores actually doing the processing. I would have hoped for a little better (less than half the runtime of the single threaded version), but I suppose there's always going to be a penalty for concurrency mechanisms. \$\endgroup\$ – Daniel Sep 6 '14 at 18:43
  • \$\begingroup\$ Its probably not overhead from concurrency. You have two serial processes that have not been parallelized. Reading/Writing are still both sequential. Also overlapping Reading/Writing operations does not necessarily make the application faster as the device may not be able to handle parallel operations (try raiding some drives together). So you should measurements in the improvement are flawed. Personally I would not bother to put Read/Write in separate threads. Do all the reading/writing from the main thread. \$\endgroup\$ – Martin York Sep 6 '14 at 21:51
  • 2
    \$\begingroup\$ if num_threads in process_file is less than or equal to 2 your program won't work properly because either there will be 0 worker threads, or the unsigned will wrap around and it will try to create billions of threads. \$\endgroup\$ – programmerjake Sep 8 '14 at 8:59
4
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ +1 for the emplace suggestion - I've added that in. I'm not really concerned with the correctness of the example (it was just a rough-and-ready example). I had thought about the exceptions, but not too sure how best to go about dealing with them in an efficient manner. \$\endgroup\$ – Daniel Sep 11 '14 at 20:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.