3
\$\begingroup\$

This is a "minimal thread pool" I came up with to compare with a colleague's approach using Rust's Rayon's par_bridge() to automatically parallelize looping over an indexed container (a vector in his case).

This is basically a std::for_each with std::execution::unsequenced_policy (in this case), but you have access to the guts of the pool (you can decide on chunk size, number of threads, .. waiting for executors in C++26).

bool process_one(string filePath); // can be a lambda passed as argument too

void loopFilesVect(vector<string>& filesVect, int packSize, int nbThreads)
{
  int nTh = nbThreads;
  uint32_t N = filesVect.size()/* #items */, c = N/packSize /* #packs */;
  mutex mtx; vector< unique_ptr<std::latch> > sigs; // unique_ptr for automatic deletion of latches

  int k = 0; atomic_uint32_t success = 0;
  auto process_chunk = [&](const uint32_t start, const uint32_t end) {
    cerr<<std::format(" Worker {} : {} to {}\n", start/packSize, start,end-1);
    for(unsigned int i = start; i < end; ++i) {
      if(process_one(filesVect[i])) success++;
      else cerr<<" error processing "<<filesVect[i]<<"\n";
    }
    { scoped_lock<mutex> g{ mtx }; sigs[k++]->count_down(); }
    cout<<std::format(" Worker {} done\n", start/packSize);
  };
  
  std::vector<std::future<void>> workers; // std::async() -> future
  #define submitPack(start,end) { sigs.emplace_back(new std::latch{1}); \
    workers.push_back( std::async(std::launch::async, process_chunk, start, end )); }
  
  cout<<std::format(" Processing {} file{}..", N, N==1?"":"s");
  { if(nTh>c) nTh = c; auto it = sigs.cbegin(); int n = 0;
    for(n = 0; n < nTh; n++) // use up allowed task count
      submitPack( n*packSize, std::min((n+1)*packSize,N) );
    (*it)->wait(); it++; // wait for 1st task to finish, so we can start a new one without exceeding nTh
    n = nTh/* n*packSize already processed */; c-- /* now last pack makes n*packSize > c*packSize */;
    while(n*packSize < N) { // do nothing if all packs already submitted
      submitPack( n*packSize, n==c ? N/*last one*/ : (n+1)*packSize );
      if( n==c ) break; // previous line just submitted last pack
      (*it)->wait(); it++; // wait for oldest task to finish
    }
    sigs.back()->wait(); // wait for the last pack to be processed
  }

  cout<<std::format("Done ({} successfully processed)\n", success.load());
}

loopFilesVect(vFiles, thread::hardware_concurrency()/2, 200);

Performance was similar to Rust's, which was satisfying (given the power of Rayon).

It uses the lightest synchronization object available (a latch{1}), and it's all in front of you : you can make this as sophisticated as you want (from adding atomic counters for statistics like nProcessed, nFailed, .. to using an elaborate synch mechanism beyond scoped_lock - but then maybe you need a Pro-thread pool).

Do you see anything wrong/improvable with this implementation ?

\$\endgroup\$
2
  • 2
    \$\begingroup\$ It is certainly legal to write a single line that says a = x; /* foo */ b = y; /* bar */. But if you feel the need to repeatedly insert intra-line explanations, maybe it's time to take a step back. Why is this code tricky enough to need all those explanations? Would dedicating a full comment line above the code be an effective way to convey what the Reader needs to know? Would longer identifiers, or breaking out a helper function, lead to more self-explanatory code? As an exercise you might try copying out just the code (no comments), then edit that for clarity. \$\endgroup\$
    – J_H
    Feb 26 at 23:24
  • \$\begingroup\$ My comment is that. That code is really hard to read. Please space things out. This is not a code gold you don't want to use the minimal number of lines. Spread it out make it easy to read. \$\endgroup\$ Feb 29 at 20:26

1 Answer 1

2
\$\begingroup\$

Code style

This code is terrible to read; the formatting is inconsistent, sometimes there are multiple statements per line, sometimes not. There are comments in the middle of lines, making it easy to miss things.

I recommend you use an automatic code formatter to reformat your code.

Also avoid unnecessary abbreviations and one-letter variables. Exceptions are only things that are very commonly used, like i for a loop variable. However, you have N, n, k and c, and it's very hard to tell what those are from just their name alone.

This function does too many things

You claim "this is a minimal thread pool", but it's not. It's a function that only works on a vector of strings, and calls process_one() on each of them in parallel. What if you have something that is not a vector of strings? What if you want to call some other function than process_one()?

I would try to create something that's just a generic thread pool, without hardcoding what kind of container to work on or what function to call on each of the elements. That will result in much more reusable code.

Avoid macros

submitPack() does not need to be a macro, just make it a lambda.

Use std::size_t for sizes, counts and indices

You are mixing int, unsigned int and uint32_t. Be consistent, and prefer to use std::size_t for sizes, counts and indices.

It's more complicated than necessary

There's a mutex, latches and futures. You shouldn't need all three. Furthermore, you are starting a new thread for each chunk, which has some unnecessary overhead.

I would just start nTh tasks that each pick chunks themselves until all chunks have been processed. Something like:

auto process_chunks = [&]() {
    while (true) { 
        std::size_t start;
        std::size_t end;
        {
            std::lock_guard lock(mtx);
            if (k > c)
                return;
            start = k * packSize;
            end = std::min(start + packSize, N);
            k++;
        }
        for (std::size_t i = start; i < end; ++i)
            process_one(filesVect[i]);
    }
}

std::vector<std::jthread> workers;
workers.reserve(n);

for (std::size_t n = 0; n < nTh; ++n) {
    workers.emplace_back(process_chunks);
}

// Wait for all work to finish:
workers.clear();

I've used std::jthread here for simplicity, but you can do the same with std::async(). And instead of using a mutex, you could even use a std::atomic<std::size_t> to track the next chunk to process.

loopFilesVect() is called with parameters in the wrong order

When you call loopFilesVect() at the end of your code, you passed the number of threads and the chunk size in the wrong order.

This is a common problem when passing two parameters of the same type; the compiler cannot tell that there is a problem here. You can avoid this in several ways. Maybe the simplest is to not have a parameter for the number of threads, but just the chunk size, and always use up to std::thread::hardware_concurrency() threads. Another way is to define a struct that contains the parameters, so you can use designated initializers:

struct thread_pool_parameters {
    std::size_t packSize;
    std::size_t nbThreads;
};

void loopFilesVect(vector<string>& filesVect, thread_pool_parameters parameters) {
    …
}

loopFilesVect(vFiles, {.packSize = 200,
                       .nbThreads = std::thread::hardware_concurrency() / 2});
\$\endgroup\$
11
  • \$\begingroup\$ Thank you for the review ! I agree in general with the comments about style, the real thing looks nothing like this, be assured. I am not a fan, however, under the pretense of "clarity", of stretching a 1-active-line (ususally a syscall)-function to fill up half a page. Being able to visually dominate a slice of code helps my understanding, scrolling a mile.. doesn't. Here the "style" I chose went with my boasting of an "inline thread pool" to go, which, admittedly, went a bit far (I did invert args in function call during the hair cut). \$\endgroup\$ Feb 27 at 17:02
  • \$\begingroup\$ As per the strategy, you're absolutely right, restarting threads is costly, the eager-worker polling for meals to eat is of course better (that's exactly what I do in my main program), this was an experiment to compare Rust & C++ (I'm still very impressed by Rayon). I tried jthread and async : async startup is like 3 times faster on Windows 10 (we're talking pennies). As per your size_t comment, I know it's the wise choice, but I have no sense of range with size_t. Somehow I'm repeatedly surprised when I look it up. I like knowing my program can handle 4 billions files in a single folder.. \$\endgroup\$ Feb 27 at 17:12
  • \$\begingroup\$ some other places where 4 billions looked like a (remote) risk (a lock-free queue with ever-increasing pop/push cursors - for perf.) I chose uint64_t and I can sleep better.. \$\endgroup\$ Feb 27 at 17:14
  • \$\begingroup\$ Regarding the macro use, until C++ takes reflection seriously, I have no shame coding with macros. A two-liner is no issue at all, but lambdas are indeed a must for longer/more complex logic. \$\endgroup\$ Feb 27 at 17:27
  • 1
    \$\begingroup\$ With std::function neither Clang nor GCC seem able to inline recursive calls. But in C++23 you can write recursive lambdas, and those will be inlined (of course, as long as the compiler can see at compile time if the recursive call will happen). Anyway, you mention lots of decisions you made because of code not mentioned in your question. We can of course not review based on things we cannot see, so our answers will only reflect the posted code in isolation. \$\endgroup\$
    – G. Sliepen
    Feb 28 at 18:12

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.