How to interrupt a QtConcurrent::blockingMap

Hello,

I am looking for a way to interrupt a QtConcurrent::blockingMap's job. I tried calling clear() on the QThreadPool's global instance, but it doesn't work as I expected. Here is a minimal example showing what I'm trying to do.

Code:

---

#include <QtConcurrent>
#include <QThreadPool>
 
#include <vector>
 
int main(int argc, char ** argv)
{
  std::vector<int> ones(100, 1);
  ones[10] = 123; // special value
 
  QThreadPool::globalInstance()->setMaxThreadCount(1);
 
  auto times_two = [](int & i)
  {
    if (i == 123) // when reaching the special value, I want to stop the other jobs here
      QThreadPool::globalInstance()->clear();
    else
      i *= 2;
  };
 
  QtConcurrent::blockingMap(ones.begin(), ones.end(), times_two);
 
  return 0;
}

---

Here, each value inside a vector is multiplied by 2, until a special value (123) is found. Then, the call to clear() seems to have no effect whatsoever, i.e. the lambda "times_two" is called on the rest of the vector. But according to the documentation, QThreadPool::clear() is supposed to "remove the runnables that are not yet started from the queue".

Is there something I am doing wrong here ?

Thanks.

Maybe use the non-blocking variant and then cancel the future?

Cheers,
_

Thanks for your feedback.

In my case however, I would like to prevent any further call to "times_two" from inside the function itself.
Any way to do that ?

Use a worker thread that checks the exit condition in its processing loop.

Basically a QThread derived class with run() implemented as a for-loop over the container with the loop's body what you currently have in two_times.

Cheers,
_

Hi anda_skoa,

I am working with StephanCat on this subject. The idea of using a blocking map was indeed to hide that top level for loop you are talking about.
What we are trying to do is to interrupt a blocking map in case one of the "instances" of the function that is being executed in parallel fails for some reason (exception or met condition). You see, if one fails, then the whole thing should fail with it. We understand that the other already running functions can't be stopped before they return but it's not the case of the ones that have not been started yet.

Is there no way to do that with a highlevel function like blocking map?

The reason I believed it was possible was that setting the number of threads to 1 on the global thread pool actually have an impact on the number of threads used by blocking map. Thus proving (maybe not) that blocking map actually uses the global instance thread pool under the hood to dispatch the work load. Since QThreadPool::clear() does exactly what we want, we hoped that calling it would be enough to do achieve the result. The minimal example provided proves this to be wrong though.

Yohann

If you want the whole operation to fail when a single element fails then "map" won't work, because that will have changed the container up until that point.

In any case, you are working with assumption on how that works internally and then trying to hack around how it works, which is never a good idea.

What if the algorithm, knowing the size of the input waits to N operations to finish? Removing scheduled operations before they finish would mean it will wait for something that will never happen.
What if the algorithm creates threadpool tasks in batches? Clearing the current batch would not impact any future batches.

Since you are calling a blocking method and only don't want it to process items in parallel, you don't need anything from QtConcurrent or threading in general, only a simple loop.
Which you can always just encapsulate in a function that takes a container and a functor/lambda.

Cheers,
_

Hi again,

I agree it's never a good idea to hack around something or even to rely on internal implementations to make something work. It was not our intent.

About setting the number of threads to 1 in our example, it was just to demonstrate without a doubt that changing the number of threads on the global thread pool was actually having an effect on the number of threads that blocking map would use (is it a hack? What is the right way to do that? The documentation doesn't say anything about it, stackoverflow says it is the right thing to do. They are wrong sometimes, agreed). We're not seriously considering multiplying each element of a vector by two using such an algorithm either you know ;) So don't take the example too seriously: it is just there so you have something simple to compile and play with, that reproduces what we are trying to do.

Don't get me wrong, we appreciate your help, and I'd like to thank you again.

Anyway, I understand there is no right way to do this with blocking map. So we are considering an implementation of our own blockingmap, based on the global QThreadPool + QRunnable, calling clear() from a job should it fail. Do you have any advice or warning about this approach ?

Kind regards

Yohann

If you want to call clear on a threadpool, it might be better to use your own instance, so you are sure you are only removing your runnables.

But despite the multiply by 2 being just an example, it doesn't change what you have said that you are trying to do.

If you want to exit when you encounter a certain element, you can only have one thread working through the container => not a use case for a threadpool.
If you want this to block, why use threads at all?

Cheers,
_

That's not what I meant to say. I am sorry if our example is too dumb to illustrate our case. I am not sure a copy/paste of our actual code base would have helped more.

Let me rephrase again (and for the last time, because we started implementing our own blocking map(*) and need to move on):
1. we have fairly independant jobs to do in parallel,
2. doing them in parallel saves us a lot of time (we have hundred thousands of them and we actually achieve n threads => n times faster up to 48 cores, could not test with a bigger machine),
3. most of the time, they all succeed, and we are fine and happy because we have our result and it went fast (well n times faster),
4. occasionnally some may fail (1) depending on the input not being what is expected, or whatever, and the overall result (after all jobs have executed) becomes useless to us (too bad). As a result, we don't want to bother waiting for the pending (2) jobs to start and return, should there be a few, because they won't make the overall result any more useful.
5. the main thread is blocked because it has nothing else to do but to wait for the parallel jobs to return.

Does it make sense? Are we really that way off of multithreading purposes?

Kind regards

(*) we will consider your advice about using a private threadpool instead of the global one.
(1) reaching an element is a way to simulate failure in our example, maybe that's what you missed.
(2) what ThreadPool::clear does, no more, no less.

Here is a suggestion: just add a "fail" atomic boolean flag. This flag starts cleared and is set by times_two when it decides that the computation has failed. In times_two, check the flag and return immediately if it is set.

There is a good chance that this is all you need. Of course, after failure happens, times_two is still called on the remaining elements of the sequence, but:

• those invocations are really fast since all they do is check a flag and return;
• as you said, failure is quite exceptional, and perhaps you can afford waiting for those quick invocations to finish.

If you want to prevent times_two from being called on the remaining elements of the sequence, then you can still use QtConcurrent, just not the blocking variant: call QtConcurrent::map() instead, then set up a QFutureWatcher to watch the QFuture that it returns, then call QFutureWatcher::waitForFinished(). In times_two, upon detecting failure, use QMetaObject::invokeMethod() to asynchronously invoke the cancel() slot of your QFutureWatcher.

Note: if this does not work, then it could be that QFutureWatcher::waitForFinished() does not internally run an event loop, in which case you can just run a QEventLoop that exits as soon as your QFutureWatcher emits finished().

That's it, by using these various Qt technologies, you solve your problem in less than 15 lines of code, and you avoid reinventing the wheel. Seriously, stay with QtConcurrent.

From the description I would say that you're better off implementing your own thread pool, one that can deal with these kind of dependencies.

QThreadPool is more a fire-and-forget type solution, not so well suited for things that need to be controlled while things execute.

One library that could be interesting is https://inqlude.org/libraries/threadweaver.html
It has support for communicating between jobs and threadpool on whether to run a job at all, if to abort queueing it, etc., it can organize jobs in collections and so on.

Cheers,
_

Hi yeye_olive

Don't worry, we did not reinvent the wheel. We simply wrote our own BlockingMap function from the building blocks of QtConcurrent, ie.
- a QThreadPool that defaults to the global instance
- a for loop that iterates the elements and enqueues a special kind of QRunnable (see below)
- a special kind of QRunnable that wraps the function to be launched concurrently, with exception support: it calls clear on the threadpool and makes sure no further job is enqueued by the for loop if an uncaught exception is thrown from the function, and finally transfers exception to the main thread.

We tried your second suggestion but it had a problem that we did not understand: it looked like it had "inertia", since it seemed to keep starting further jobs (a lot) after the one that failed before it eventually stopped (with only a few jobs that did not execute at all).

Thanks a lot for your feedback though, we really appreciated it.

Thanks again to anda_skoa, even though we don't seem to agree on everything, your comment that we might be playing into undefined behavior was a significant step forward to us.

I have a similar use case.
I have presented some solutions here.
Thank you yeye_olive, your advices works.