Multithreaded natural language text parser (Rev.3)

Question

This is the third iteration of the Multithreaded natural language text parser code review. Special thanks goes to G. Sliepen who conducted the first two reviews.

Before reading this post, please read the previous iteration, since without it most of the text below doesn’t make sense.

Functional specification

Implement a multithreading parser based on existing natural language tokenizer.

1. Parser should compile text into sequences of lexem ids.
2. Parser should create structures to get lexem text by id and id by lexem text.
3. Lexems id must be unique for lexem, but no any other stronger requirements are provided.

Performance and memory footprint is critical, since the amount of data is huge (gigabytes).

Nice to have the approach which could be portable to GPU for further speed up.

Changes

1. The excessive refactoring leftover ‘tokenize_chunk’ has been removed.
2. The TextBaseSingleThreaded::tokenize and TextBaseMultiThreaded::tokenize_chunk merged to one template.
3. Functional decomposition to TextBase<LexemsMapType, DictMapType>::tokenize_chunk applied.

The code

Fully functional runnable demo.

Here is the updated code for the code review; could you please take a look and suggest further ways to improve or confirm that this is ready to go code?

Please note that godbolt.org doesn’t run code with MSVC, so don’t expect to see the output.

using lexem_t = unsigned int;

struct LexemData {
    lexem_t id;
    std::size_t count;
};

struct LexemDataAtomic {
    lexem_t id;
    std::atomic<std::size_t> count;

    LexemDataAtomic(unsigned int i, std::size_t count) :
        id{ i },
        count{ count }
    {
    }

    LexemDataAtomic(const LexemDataAtomic& rhs) :
        id{ rhs.id },
        count{ rhs.count.load() }
    {
    }

    LexemDataAtomic& operator=(const LexemDataAtomic& rhs)
    {
        if (&rhs != this)
        {
            id = rhs.id;
            count = rhs.count.load();
        }
    }

    LexemDataAtomic(LexemDataAtomic&&) = delete;  // atomic is not moveable
    LexemDataAtomic& operator=(LexemDataAtomic&&) = delete;
};

template <typename TextBaseType>
struct CompiledText {
    TextBaseType* text_base;
    std::vector<lexem_t> data;

    CompiledText(TextBaseType& text_base) : text_base(&text_base) {}
};

template <typename LexemDataType, typename LexemsMapType, typename DictMapType>
class TextBase {
protected:
    LexemsMapType   lexems_data;
    DictMapType     index;

    LexemDataType& add_to_dictionary(lexem_t& id, std::string_view lexem);
    void add_to_data(std::vector<lexem_t>& compiled, lexem_t code);

    std::vector<lexem_t> tokenize_chunk(std::string_view text, lexem_t start_id);
public:

    enum {
        use_all_available_threads = 0,
    };

    std::string_view lexem_by_id(lexem_t lex_id) {
        return index[lex_id]->first;
    }
};

template <typename LexemDataType, typename LexemsMapType, typename DictMapType>
LexemDataType& TextBase<LexemDataType, LexemsMapType, DictMapType>::add_to_dictionary(lexem_t& id, std::string_view lexem) {
    auto res = lexems_data.insert({ std::string(lexem),{ id, 0} });

    if (res.second) {
        index[id++] = res.first;
    }

    return res.first->second;
}

template <typename LexemDataType, typename LexemsMapType, typename DictMapType>
void TextBase<LexemDataType, LexemsMapType, DictMapType>::add_to_data(std::vector<lexem_t>& compiled, lexem_t code) {
    compiled.push_back(code);
}

template <typename LexemDataType, typename LexemsMapType, typename DictMapType>
std::vector<lexem_t> TextBase<LexemDataType, LexemsMapType, DictMapType>::tokenize_chunk(std::string_view text, lexem_t start_id)
{
    std::vector<lexem_t> compiled;
    auto id = start_id;

    for (auto lexem : TokenRange(text)) {
        auto lexem_data = add_to_dictionary(id, lexem);
        ++lexem_data.count;
        add_to_data(compiled, lexem_data.id);
    }

    return compiled;
}

struct TextBaseSingleThreaded
    : public TextBase<LexemData, std::unordered_map<std::string, LexemData>,
    std::unordered_map<lexem_t, std::unordered_map<std::string, LexemData>::iterator>>
{
    CompiledText<TextBaseSingleThreaded> tokenize(std::string_view text);
};

struct TextBaseMultiThreaded
    : public TextBase<LexemDataAtomic, Concurrency::concurrent_unordered_map<std::string, LexemDataAtomic>,
    Concurrency::concurrent_unordered_map<lexem_t, Concurrency::concurrent_unordered_map<std::string, LexemDataAtomic>::iterator>>
{
    CompiledText<TextBaseMultiThreaded> tokenize(std::string_view text, std::size_t threads = 0);
private:
};

CompiledText<TextBaseSingleThreaded> TextBaseSingleThreaded::tokenize(std::string_view text)
{
    CompiledText<TextBaseSingleThreaded> compiled_text(*this);
    compiled_text.data = tokenize_chunk(text, 0);

    return compiled_text;
}

// If n_chunks == 0, use all available threads
CompiledText<TextBaseMultiThreaded> TextBaseMultiThreaded::tokenize(std::string_view text, std::size_t n_chunks)
{
    n_chunks = n_chunks != use_all_available_threads ? n_chunks : std::max((std::size_t)std::thread::hardware_concurrency(), (std::size_t)1);

    std::size_t chunk_size = text.size() / n_chunks;

    std::vector<std::future<std::vector<lexem_t>>> results;
    results.reserve(n_chunks);

    std::size_t carry = 0;

    const TokenRange empty({});
    auto delimiters = empty.begin().get_delimiters();

    CompiledText<TextBaseMultiThreaded> compiled_text(*this);

    for (std::size_t chunk = 0; chunk < n_chunks; ++chunk) {
        if (carry >= chunk_size) {
            carry -= chunk_size;
            continue;
        }

        std::size_t chunk_start = chunk * chunk_size + carry;
        std::size_t chunk_end = (chunk + 1) * chunk_size;

        std::string_view search_range = text.substr(chunk_end - 1);

        carry = search_range.find_first_of(delimiters);

        if (std::string_view::npos == carry) {
            carry = search_range.size();
        }

        chunk_end += carry;

        results.push_back(std::async(std::launch::async, &TextBaseMultiThreaded::tokenize_chunk, this,
            text.substr(chunk_start, chunk_end - chunk_start), (lexem_t)(chunk_start)));
    }

    for (auto& result : results) {
        compiled_text.data.append_range(result.get());
    }

    return compiled_text;
}

void generate_text(std::string& text, std::size_t length)
{
    std::vector<char> alphabet;

    for (char c = 'a'; c <= 'z'; ++c)
        alphabet.push_back(c);

    for (std::size_t i = 0; i < 5; ++i)
        alphabet.push_back(' ');

    for (std::size_t i = 0; i < 3; ++i)
        alphabet.push_back(',');

    for (std::size_t i = 0; i < 2; ++i)
        alphabet.push_back('.');

    text.resize(length);

    std::mt19937 generator(0);
    std::uniform_int_distribution<std::size_t> gen(0, alphabet.size() - 1);

    std::ranges::generate(text, [&]() { return alphabet[(std::size_t)gen(generator)]; });
}

int main()
{
    _CrtSetReportHook2(_CRT_RPTHOOK_INSTALL, [](int, char*, int*) { return 1; });

    {
        std::string sample = "Let's consider, this semi-simple sample, i.e. test data a+b with ints: 100, etc. For ... some testing...";

        std::stringstream simply_parsed;

        for (auto token : TokenRange(sample)) {
            simply_parsed << token << " | ";
        }

        std::cout << "Simply parsed version: " << simply_parsed.str() << "\n";

        TextBaseSingleThreaded single_threaded_text_base;
        auto single_threaded_compiled_text = single_threaded_text_base.tokenize(sample);

        std::stringstream single_threaded;

        for (std::size_t i = 0; i < single_threaded_compiled_text.data.size(); ++i) {
            single_threaded << single_threaded_text_base.lexem_by_id(single_threaded_compiled_text.data[i]) << " | ";
        }

        std::cout << "\nSingle-threaded version:" << single_threaded.str() << "\n";

        TextBaseMultiThreaded multi_threaded_text_base;
        CompiledText<TextBaseMultiThreaded> multi_threaded_compiled_text = multi_threaded_text_base.tokenize(sample);

        std::stringstream multi_threaded;

        for (std::size_t i = 0; i < multi_threaded_compiled_text.data.size(); ++i) {
            multi_threaded << multi_threaded_text_base.lexem_by_id(multi_threaded_compiled_text.data[i]) << " | ";
        }

        std::cout << "\nMulti-threaded version:" << multi_threaded.str() << "\n\n";

        std::cout << (simply_parsed.str() == single_threaded.str() ? "Single threaded test passed: OK" : "Single threaded test failed") << "\n";
        std::cout << (simply_parsed.str() == multi_threaded.str() ? "Multi threaded test passed : OK" : "Multi threaded test failed") << "\n";
    }

#define PERF
#ifdef PERF
    {
        std::cout << "\nPerformance testing:";

        const std::size_t text_length = 100'000'000;

        std::string text;
        generate_text(text, text_length);

        TextBaseSingleThreaded single_threaded_text_base;
        CompiledText<TextBaseSingleThreaded> single_threaded_compiled_text(single_threaded_text_base);
        TextBaseMultiThreaded multi_threaded_text_base;
        CompiledText<TextBaseMultiThreaded> multi_threaded_compiled_text(multi_threaded_text_base);

        decltype(std::chrono::duration<double>()) single_threaded_duration;
        {
            std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();

            single_threaded_compiled_text = single_threaded_text_base.tokenize(text);

            std::chrono::high_resolution_clock::time_point now = std::chrono::high_resolution_clock::now();
            std::cout << "\nDuration of the single-threaded version = " << (std::chrono::duration<double>(now - start));
            single_threaded_duration = (std::chrono::duration<double>(now - start));
        }

        decltype(std::chrono::duration<double>()) multi_threaded_duration;
        {
            std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();

            multi_threaded_compiled_text = multi_threaded_text_base.tokenize(text);

            std::chrono::high_resolution_clock::time_point now = std::chrono::high_resolution_clock::now();
            std::cout << "\nDuration of the multi-threaded version = " << (std::chrono::duration<double>(now - start));
            multi_threaded_duration = (std::chrono::duration<double>(now - start));
        }

        std::cout << "\nSpeed up (times): " << single_threaded_duration / multi_threaded_duration;

        auto is_compiled_texts_equal = std::ranges::equal(single_threaded_compiled_text.data, multi_threaded_compiled_text.data, std::equal_to{},
            [&](auto lex) {
                return single_threaded_text_base.lexem_by_id(lex); },
                [&](auto lex) {
                return multi_threaded_text_base.lexem_by_id(lex); });


        std::cout << "\n\n" << (is_compiled_texts_equal ? "Test passed: OK" : "Test failed") << "\n";

    }
#endif //PERF
} 

Concerns

My main concerns again around this function decomposition I am not very happy with. I have decomposed the TextBase<LexemsMapType, DictMapType>::tokenize_chunk and now it looks better, but I still have at least two questions now.

The add_to_dictionary still has 3 different responsibilities if we adhere to function decomposition rules:

1. Add a value to the lexems_data
2. Update the index to lexems_data
3. Increment the id counter

I can somehow agree that first two could be considered as updating database with index in terms of a transaction. The third one is already suspicious, since doesn’t have any relation to the operation itself and to dictionary (as a database).

This leads to the second question if this is a good idea to pass id as a reference to add_to_dictionary. First of all, I hate the idea to keep client blind on the fact that the id could be changed with the call to add_to_dictionary. Well, one could see the function signature, but, anyway, pure code reading on client side hides the fact that id could be changed. I hate the idea pass the address to reveal this, as well, since it would force me to check it in the add_to_dictionary for nullptr.

I have two options:

1. Pass the id be value and get the updated id as a return value std::paired with the current result, but this would be weird.
2. I can raise a level of abstraction and return instead bool to show if the record was added to the dictionary which seems much better. But now I will have to duplicate the check if (added) which already exists in add_to_dictionary. Better high-level design, worse the performance on the critical path.

So, I am still in two minds with the better solution for this id which is in a wrong place in add_to_dictionary, but gives performance and avoiding code duplication benefits.

Don’t get me wrong, in case this was the problem, I would keep is as is and spend time for real tasks. I use this just as an example to show why I am concerned with functional “over”-decomposition which leads to the balance questions about performance/code duplication and nice functional decomposition. I will be thankful for any ideas except “programming is always searching trade-offs” which is correct, but obvious and, therefore, doesn’t help because this leads to holy wars where everybody stays on his position saying “this is my choice” and we stay with programming as an art, instead of moving to science.

With this I am still eager to hear your feedback on questions posted in form of comments to Natural language text fast tokenizer (Rev.3). Here is the summary of these questions, so the link above is just for the context.

Function extraction is one of the hardest problems for me and I am very interested in your help here. Here is my reasoning.

1. Yes, extracted functions are private, but they visible for all members of this class; let’s imagine a class with a function with complex calculations (find intersections of geometrical primitives); I will extract 50 sub-functions and this will be a mess of functions visible in this class.
2. I am not concerned with performance overhead, but I am concerned with source file size. Instead of 150 lines of large function I will have 400 lines of 50 small functions to be written, documented, learned, remembered and read by others. Well, I exaggerating to explain the idea, but I had the cases.
3. Most of these functions don’t have any meaning outside of the large function context, so it is hard to name and document them, since function find_left_upper_corner_of_frastum_and_horizon_intersection() is crazy and preconditions like “array must be sorted, all characters must be lowercase”, etc. could be deadly and function without them could be dangerous if used by another developer in support.
4. People always say, “you are a human, find the balance!”, but when it is suggested to extract 3 lines of code to a separate function, making it 7 lines of code, where is a balance? What about Occam’s razor? To put in a nutshell, I love functional decomposition, I see that most of good software uses it, but until some technique invented better than “just nest a ‘utils’ or ‘details’ structure or namespace”, I will be in twenty minds about over-extracting functions. Please help me, if you have good recommendation here.

Again, this list is short, just issues “on a surface”.

Answers to the code review points

The total variation at the end will just be a small fraction of the total time your program runs, so you could ignore it, but you could also avoid mostly it.

I totally agree. The question is, is there a way to do this better or mitigate this? I mean, the statement is correct, but “so what?” question follows.

If the total time your program runs is not changed when you use 2 times less threads, then that means your program is not CPU-bound, and instead having more threads just makes it less efficient.

I am sorry, my wording was bad. Of course, I meant "nothing changed dramatically from the perspective of threads management"; the performance reduced with less threads number, but "as expected"; just wording. When I reduced the number of threads in 2 times, the performance degraded but this didn’t lead to much benefit from savings on threads start/stops as you pointed out. So, no such influence here.

You should decouple the size of the chunk from the number of cores. For example, if you have 8 cores and 16 chunks, you start 8 threads and have each thread process 2 chunks.

I can’t catch what you intend to achieve with this? For every core I will have two smaller chunks instead of one larger chunk, I will spend some (negligible, of course) effort to split two times more, but what I would be expected to win and in which way? I can implement this, but to implement this in correct way I need to understand the goal. Can you please help to catch it?

Another reason I mentioned this is that if parallelism doesn't give you the speedup you expected, then maybe you should focus your attention to the serial steps in your algorithm.

The code is on the table and it is quite simple. I am not sure this would be an easy take fruit to speed up this with linear execution in 4.4 times like I have on multithreaded CPU and I expect more on GPU. Do you have any clues what can be done to improve the linear version?

std::copy is the best case scenario for memory access. Your algorithm might not be able to achieve the same bandwidth.

Of course, it wouldn’t, I am just saying that I am not memory-bound with this task, as you suspected in code review iteration one. With only one reservation, although in the next quote:

I think it is likely you can get more performance. I would first identify where the bottleneck really is, and measure the time for setting up the chunks, for running the threads, and for creating the compiled_text vector afterwards. Using a profiler like perf might also help finding out where most of the time is spent.

Of course, I do this from the early days, since performance is critical. The main bottleneck which could be considered for optimization is this std::map. The 80% of time I work with it for searching and adding and this could be improved. I have some ideas with plan structures like std::vector with B-tree inside without expensive memory operations and cache-unfriendly std::map on large data, but this is another story. It wouldn’t be an easy taken fruit as multi-threading seems to be here.

Answer 1

To me this looks like a classic map-reduce problem. I would implement a map and reduce functions, of which mapping of chunks looks like

0. Setup thread local datastructures (no thread safety needed, because they are used by only one thread) and assign one to each worker

1. If the current chunk is bigger than threshold, divide it in half and search for closest end of a token, then generate two tasks from that (go to 1 again)

2. If the chunk is smaller or equal to threshold, perform normal tokenize

This allows a worker to concentrate on its own task and not contending on resources with other workers. With worker being threads, it will improve data locality thus leading to doing more useful things per CPU cycle.

After the mapping process finishes, the reduce part could be like the following:

1. Agree on common lexem ids

2. Post a task for each worker to establish lexem rename maps

3. Create a hashmap with value type being atomic (if false sharing is established to be a problem, pad entries to break false sharing)

4. Post task for every worker to sum the results of the index together

Although it looks like 4 might contend to update the same memory page, there will be at most number of worker contentions once, instead of number of workers multiplied by possible occurences. I do not know how long step 1 of reduce will take, but given sufficiently big text, I suppose it will be faster.

With the algorithm above, there will be multiple small taxes on performance (dividing the task, posting tasks for each worker, data structure allocations) and two big (establishing common lexem id map + rename maps for each worker). Number of unique lexems is likely much smaller than the text itself.

Static chunk scheduling suffers from variability in CPU time availability for the process. In most computing, the launched program is not the only running on the machine. Getting scheduled out might eat into CPU time of the process thus creating unevenness, stalling the other CPUs from progressing. Dynamic task scheduling with dividing the task into smaller chunks makes more work, but works a lot better in most situations.

Non-composable threading is not great example of executor. The code in question does not provide any means of cooperating with other threads in the program, nor does it allow the work chunks to be executed on something else. It would be great to be able to separate work division/scheduling from executor.