2
\$\begingroup\$

Could you please conduct code review for the code below and suggest some improvements?

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.
  4. Parser should count how many times each lexem appeared in text. This functionality is nice to have and if removing it could dramatically simplify or speed up the parser, it could be ignored.

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.

Design

  1. The parser is based on the natural language text parser.
  2. Being a proof of concept is utilizes the Microsoft concurrent_unordered_map which is not portable, but doesn’t require external libraries.
  3. The project is expected to have some TextBases which hold lexems_data and dict maps to store lexems information to allow access trough lexem id and lexem text.
  4. The proposed approach is to have two distinct types TextBaseSingleThreaded and TextBaseMultiThreaded to hold different types of maps, since std::maps is much faster in single-threaded execution.
  5. The main piece of code is method TextBaseMultiThreaded::tokenize which splits the work in chunks and responsible for managing lexems on borders and merging the compiled sequences of lexem ids.

Reservations

  1. Please don’t consider the tokenizer code, since it is covered under the natural language text parser; feel free to participate there, if you like.
  2. Another option would be to use OneAPI/tbb/concurrent_unordered_map to make the code portable, but this would require installing an additional library which I am in two minds about for proof of concept.
  3. Most of types implemented as structs instead of classes to avoid swamping current code with getters and setters. Production code will utilize classes.
  4. Don’t pay much attention to function main, except maybe usage of these objects from the client side.
  5. I can’t find other multithreading parsers to learn from, so if you aware of such, please, share.

Key questions

  1. The main question if the approach correct at all. First of all, it the idea of chunking the best one or something better (heaps?) could be used?
  2. Is the object model correct or it could be improved? For example, I would like tokenizer to return CompiledText, but I don’t want to rely on RVO/NRVO which could frame me.
  3. I am not sure if dynamic polymorphic types could help here.
  4. Please review the tokenize and tokenize_chunk functions source code if it could/need to be improved and LexemDataAtomic structure.
  5. One the long question about function extraction/decomposition, the tokenize (tokenize_chunk for multithreaded version) function here does two things, namely creating the dictionaries and compiling the text and this makes sense, since both are done simultaneously and we same execution time here. People often say that function must have one responsibility only. This works if the responsibilities could be split logically. Is there a chance to split these two responsibilities here without performance impact? Note, that data size expected to be huge, so even an additional pass could affect the performance.
  6. What could be done better?

The source code

Fully functional runnable demo (requires Microsoft C++ compiler).

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

Parser

using lexem_t = unsigned int;

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

struct LexemDataAtomic {
    lexem_t id;
    std::atomic<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 LexemsMapType, typename DictMapType>
struct TextBase {
    LexemsMapType   lexems_data;
    DictMapType     dict;

    enum {
        use_all_available_threads = 0,
    };
};

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

struct TextBaseMultiThreaded
    : public TextBase<Concurrency::concurrent_unordered_map<std::string, LexemDataAtomic>, 
                      Concurrency::concurrent_unordered_map<lexem_t, Concurrency::concurrent_unordered_map<std::string, LexemDataAtomic>::iterator>>
{
    void tokenize(CompiledText<TextBaseMultiThreaded>& compiled_text, const std::string_view& text, std::size_t threads = 0);
private:
    void tokenize_chunk(std::vector<lexem_t>& compiled, TextBaseMultiThreaded& text_base, const std::string_view& text, lexem_t start_id);
};

void TextBaseSingleThreaded::tokenize(CompiledText<TextBaseSingleThreaded>& compiled_text, const std::string_view& text)
{
    lexem_t id = 0;

    for (auto lexem : TokenRange(text)) {
        auto res = compiled_text.text_base.lexems_data.insert({ std::string(lexem), { id, 0 } });

        if (res.second) {
            compiled_text.text_base.dict[id] = res.first;
            ++id;
        }

        res.first->second.count++;

        const int code = res.first->second.id;
        compiled_text.data.push_back(code);
    }
}

void TextBaseMultiThreaded::tokenize_chunk(std::vector<lexem_t>& compiled, TextBaseMultiThreaded& text_base, const std::string_view& text, lexem_t start_id)
{
    auto id = start_id;

    for (auto lexem : TokenRange(text)) {

        auto res = text_base.lexems_data.insert({ std::string(lexem),{ id, 0} });

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

        ++(res.first->second.count);

        const int code = res.first->second.id;
        compiled.push_back(code);
    }
}

// If n_chunks == 0, use all available threads
void TextBaseMultiThreaded::tokenize(CompiledText<TextBaseMultiThreaded>& compiled_text, const 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::vector<std::thread> threads;

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

    std::vector<std::vector<lexem_t>> compiled(n_chunks);

    std::size_t carry = 0;

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

    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;

        auto text_for_chunk = text.substr(chunk_start, chunk_end - chunk_start);

        threads.emplace_back(&TextBaseMultiThreaded::tokenize_chunk, this, std::ref(compiled[chunk]), std::ref(compiled_text.text_base),
                             text.substr(chunk_start, chunk_end - chunk_start), (lexem_t)(chunk * chunk_size));
    }

    for (auto& thread : threads) {
        thread.join();
    }

    for (auto comp : compiled) {
        compiled_text.data.insert(compiled_text.data.end(), comp.begin(), comp.end());
    }
}

Function main

int main()
{
    {
        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;
        CompiledText<TextBaseSingleThreaded> single_threaded_compiled_text(single_threaded_text_base);

        single_threaded_text_base.tokenize(single_threaded_compiled_text, 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.dict[single_threaded_compiled_text.data[i]]->first << " | ";
        }

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

        TextBaseMultiThreaded multi_threaded_text_base;
        CompiledText<TextBaseMultiThreaded> multi_threaded_compiled_text(multi_threaded_text_base);

        multi_threaded_text_base.tokenize(multi_threaded_compiled_text, 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.dict[multi_threaded_compiled_text.data[i]]->first << " | ";
        }

        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'00;

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

        {
            std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();

            single_threaded_text_base.tokenize(single_threaded_compiled_text, 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));
        }

        {
            std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();

            multi_threaded_text_base.tokenize(multi_threaded_compiled_text, 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));
        }

        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.dict[lex]->first; },
                [&](auto lex) {
                return multi_threaded_text_base.dict[lex]->first; });


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

    }
#endif //PERF
}

The expected results

On my PC (AMD Ryzen 9 3950X 16-Core Processor) it shows:

Simply parsed version:  Let's | consider | , | <other text>
Single-threaded version: Let's | consider | , | <other text>
Multi-threaded version: Let's | consider | , | <other text>

Single threaded test passed: OK
Multi threaded test passed : OK

Performance testing:
Duration of the single-threaded version = 7.53919s
Duration of the multi-threaded version = 1.85347s

Test passed: OK

So, the performance improvement is more than 4 times.

An alternative approach

The alternative approach could be to use execution policies in parallel for_each. Less control over the cores usage, but manipulations with threads are hidden:

void TextBaseMultiThreaded::tokenize_with_for(CompiledText<TextBaseMultiThreaded>& compiled_text, const std::string_view& text)
{
    std::size_t 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::vector<lexem_t>> compiled(n_chunks);

    struct Info {
        std::string_view text;
        lexem_t start_id = 0;
    };

    std::vector<Info> info(n_chunks);

    std::size_t carry = 0;

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

    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;

        auto text_for_chunk = text.substr(chunk_start, chunk_end - chunk_start);

        info[chunk].text = text.substr(chunk_start, chunk_end - chunk_start);
        info[chunk].start_id = (lexem_t)(chunk * chunk_size);
    }

    std::for_each(std::execution::par_unseq, info.begin(), info.end(), [&](auto& item) {
        std::size_t chunk = &item - info.data();
        tokenize_chunk(compiled[chunk], compiled_text.text_base, item.text, item.start_id);
        });

    for (auto comp : compiled) {
        compiled_text.data.insert(compiled_text.data.end(), comp.begin(), comp.end());
    }
}
\$\endgroup\$

1 Answer 1

2
\$\begingroup\$

Answers to your questions

  1. The main question if the approach correct at all. First of all, it the idea of chunking the best one or something better (heaps?) could be used?

It is a rather naive approach. There are several issues with it:

  • The total time depends on the slowest thread.
  • If you have more chunks than CPU cores, then you'll always have threads waiting for a timeslice. The OS has to perform context switching to make it seem they run concurrently, but this has an overhead.
  • If your chunks are too small, the overhead of spawning threads and having to merge the results of them at the end will dominate the total runtime.

Ideally, you have a thread pool that spawns only as many threads as there are CPU cores, and then have a queue of chunks that are not too small, and then each thread picks and processes chunks from the queue until it is empty.

  1. Is the object model correct or it could be improved? For example, I would like tokenizer to return CompiledText, but I don’t want to rely on RVO/NRVO which could frame me.

RVO is guaranteed since C++17, but even before that many compilers would do that. I recommend you return a value instead of using an output parameter. See below for some possible benefits.

  1. I am not sure if dynamic polymorphic types could help here.

I don't see any reason for it in your example code. I would avoid dynamic polymorphism unless you do have a good reason.

  1. One the long question about function extraction/decomposition, the tokenize (tokenize_chunk for multithreaded version) function here does two things, namely creating the dictionaries and compiling the text […]. Is there a chance to split these two responsibilities here without performance impact?

I would start by writing the function in this way:

auto tokenize(…)
{
    …
    for (auto lexem: TokenRange(text)) {
        add_to_dictionary(…, lexem);
        add_to_data(…, lexem);
    }
}

Now the responsibility of this function is just looping over the tokens and delegating further processing to two other functions.

Of course, now you might have to pass references to lots of things to those two new functions, which might clutter the code a bit. Still, each function now is much simpler. Also, the compiler will inline them so there should be no performance impact. The functions have names so the code is more self-documenting. I would consider this worth doing.

Avoid copying strings

Your LexemsMapType is always some kind of map from std::string to lexem data. But why not use std::string_view here as well? That avoids having to make actual copies of the strings.

Consider using std::async

If you are not going for the thread pool approach, then consider using std::async. This is like a thread, but it also captures the return value from the function, and automically joins (just like std::jthread). So combined with tokenize_chunk() returning the vector of lexems, you can write:

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

for (std::size_t chunk = 0; chunk < n_chunks; ++chunk) {
    …
    results.push_back(std::async(std::launch::async, tokenize_chunk(…)));
}

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

An added benefit of this approach is that it can already start processing the result of the first thread without having to wait for all the others to finish, although you can easily do that as well with your version by combining the last two for loops into one.

Pass std::string_views by value

A std::string_view is already kind of a reference type, and it's very small (just a pointer and a size). So passing a reference to a std::string_view is almost always unnecessary. I recommend passing them by value.

What performance improvement do you expect?

I already pointed out some possible performance issues caused by using multiple threads. In general, if you have \$N\$ threads, you will almost never get a speedup of a factor \$N\$, it's always going to be somewhat less, because virtually nothing will parallelize perfectly. Consider:

  • Starting and stopping threads has its own overhead.
  • You are doing a serial step after the multi-threaded part (combining the results into one std::vector). Even if the serial step is just a small fraction of the total work, it means the maximum speedup is a fixed number, regardless of how many threads you use (see Amdahl's law).
  • CPU processing power is not the only resource each thread consumes; memory bandwidth is another. If your tasks are doing a lot of memory reads and writes, memory will become the bottleneck.
  • Accessing shared resources is not free. Even accessing a std::atomic<int> is potentially much slower than a regular int. Your concurrent map might use a std::mutex under the hood to serialize access to the map.

So how much speedup do you expect to get from parallelizing your code? Now measure it with varying number of threads, and plot a graph showing the speedup you get. This will tell you if it's worth parallelizing, and after how many threads there will not be a gain anymore.

\$\endgroup\$
3
  • \$\begingroup\$ Thank you so much for the review. I will apply the recommendations. On the question on the amount of threads and CPU cores; isn't it what I do with the std::size_t n_chunks = std::max((std::size_t)std::thread::hardware_concurrency(), (std::size_t)1); ? And answer on "Avoid copying strings"; the dictionary size much smaller than the text size, so I will release the text buffer once it is compiled in production code; this I can't keep references to the strings in dict (more comments on this decision will follow in Rev2). \$\endgroup\$ Feb 15 at 17:19
  • \$\begingroup\$ std::thread::hardware_concurrency() might seem a good guess for the number of chunks, but I mentioned several issues that parallelizing this code has, so it might not be the optimum number. I agree that if you want to release the text buffer then you need to make copies. However, it might be more efficient to make copies at the end, instead of doing lots of conversions from std::string_view to std::string during the parsing; even if lexems_data.insert() does not insert, you still do that conversion. \$\endgroup\$
    – G. Sliepen
    Feb 15 at 19:26
  • \$\begingroup\$ Thank you again for the review. Please take a look at the Rev.2 which is mostly focused on my thoughts, answers, and request for help with function decomposition questions, but has the updated code, as well. \$\endgroup\$ Feb 18 at 20:19

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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