ThreadPool must execute all tasks posted to it

[snawaz]

Currently, ThreadPool's destructor does not wait for completion of all tasks posted to it, because Worker relies on m_running_flag to stay in the loop, as a result this flag is set to false as soon as ThreadPool destructor is invoked, as the threads exit without further processing the tasks from the queue.

One solution is to remove m_running_flag from the worker, and instead add a poison task to the queue and each worker reads this poison task and exit but before exiting enqueue it again, so that other workers can exit in similar mannner.

Also, ThreadPool may provide a function called wait() function which will add the poison task. Also, the destructor should call this wait() function.

[yvoinov]

On which use case ThreadPool must execute all tasks? Seems to me it must immediately stop workers on DTor call. Whatever queues states. Usuallly stopping ThreadPool means program exit, when no matter will execute remaining jobs in the queue.

If you use case requires independently stopping thread pool when other program part remains running, and you must execute all remains posts, ever if finish time can be infinite (in this case you destructor can hangs, is it?)....hmmmmmm.

So, logically and expected behaviour for DTor is just stop workers.

PS. If you need to stop posts just before pool destruction, to guarantee all queued posts will complete, you can always make it externally, before destructor fires.

PPS. Thread pool is primitive, not magic bullet for all and any corner cases. I hope.

[niXman]

I would prefer the same behavior as std::thread does: when destructing non-empty pool - call std::terminate().
also i would add detach() and join() memfns.

[yvoinov]

Sure, but keep in ming you have no method to check worker's queue is empty. And any different implementation will break KISS principe.

Also, you can, of course, write any member functions you want - analoques for terminate (What a bad idea terminating thread with std::terminate), detach, join :) As you whish ) Just make PR (or own fork ;)) - viola! - all done ;)

[niXman]

why? is there a lock-free queue used?
sorry, I haven't looked at the code :)

(What a bad idea terminating thread with std::terminate)

ah sure, also many people find it wrong to use assert() in irreparable situations =)

[yvoinov]

why? is there a lock-free queue used?
Exactly. Moreover, each worker here has own queue.

sorry, I haven't looked at the code :)
Getting into a discussion about code without looking at the code just to keep the conversation going isn't such a great idea either ))))))))))

[snawaz]

On which use case ThreadPool must execute all tasks? Seems to me it must immediately stop workers on DTor call. Whatever queues states. Usuallly stopping ThreadPool means program exit, when no matter will execute remaining jobs in the queue.

Why would you put tasks in the threadpool if you dont want to ensure their completion, in the general case?

If you use case requires independently stopping thread pool when other program part remains running, and you must execute all remains posts, ever if finish time can be infinite (in this case you destructor can hangs, is it?)....hmmmmmm.

What? and How?

So, logically and expected behaviour for DTor is just stop workers.

My main thought is not about the destructor itself, but an inability to wait for the completion of all tasks. Maybe, you dont want to call wait() from the destructor, fine. But there should be a wait() function on the threadpool, so that clients can use it to ensure that all tasks are done.

[yvoinov]

On which use case ThreadPool must execute all tasks? Seems to me it must immediately stop workers on DTor call. Whatever queues states. Usuallly stopping ThreadPool means program exit, when no matter will execute remaining jobs in the queue.

Why would you put tasks in the threadpool if you dont want to ensure their completion, in the general case?
Obviously, thread pool is code primitive, which is part of program. Correct? When you require to call pool DTor, it means that program is terminating, correctly or not. Still correct? Ok, let's go deeper. When you require mandatory finish all posted jobs, you always can write external (relatively thread pool) code, which will process this logic or waiting all jobs finish.

If you use case requires independently stopping thread pool when other program part remains running, and you must execute all remains posts, ever if finish time can be infinite (in this case you destructor can hangs, is it?)....hmmmmmm.

What? and How?
Trivially. Im using fork of this pool more than 2 years. All jobs in queue executes autonomously, without any control from pool side. Just imagine - how you can control executing posted job? For example, via thread pool gouverned timeout, in case of job can hang (I've often seen such cases in my code, independent from job's code quality. External dependencies for jobs can occur that).

So, logically and expected behaviour for DTor is just stop workers.

My main thought is not about the destructor itself, but an inability to wait for the completion of all tasks. Maybe, you dont want to call wait() from the destructor, fine. But there should be a wait() function on the threadpool, so that clients can use it to ensure that all tasks are done.

If you want it - just write it. :) And make PR :) Humanity will greatful :)

[tom-leys]

Use a std::packaged_task to see if the threadpool has finished executing a given task.

https://github.com/inkooboo/thread-pool-cpp/blob/master/tests/thread_pool.t.cpp

    tp::ThreadPool pool;

    std::packaged_task<int()> t([]()
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            return 42;
        });

    std::future<int> r = t.get_future();

    pool.post(t);

    ASSERT_EQ(42, r.get());

Note that r.get() is blocking here, until t() has been executed. Then it "returns" the value that was returned by t.

If you wait on all your tasks you care about in this way (you can collect tasks into a vector) then you shouldn't destroy the ThreadPool until all the tasks are done. This solves your issue.

I suggest the snippet above is put on the front page of the repo, it is more relevant than the benchmark that proves this library doesn't copy 100mb Heavy instances around RAM.

[snawaz]

@tom-leys

If you wait on all your tasks you care about in this way (you can collect tasks into a vector) then you shouldn't destroy the ThreadPool until all the tasks are done.

I don't destroy the threadpool; it gets destroyed automatically when it goes out of scope. To prevent its destruction means somehow I have to ensure that it does not go out of scope. How do I ensure that... and for exactly how long? Do you see the problem here?

To solve this problem in a simple and easy way, threadpool should have at least wait() function to wait on all tasks (if waiting implicitly in the destructor isn't a good idea).

[tom-leys]

How do I ensure that... and for exactly how long? Do you see the problem here?

Yes, I understand. In the code above the line ASSERT_EQ(42, r.get()); blocks until the task is done. That is what the r.get() does.

You'd have to stay inside the scope of the threadpool until all tasks are done like this.

I like the solution you suggested last year (first comment in this thread), it would certainly help avoid confusion for most simple use cases. An automatically easy to use task manager which always exits correctly would be very nice.

[snawaz]

You'd have to stay inside the scope of the threadpool until all tasks are done like this.

Please show me how you stay inside the scope of the thread pool! Show me the code! You cannot do that easily!

[tom-leys]

Well, my initial example does. Run it.

…

On Wed, 9 Jun 2021, 16:54 Sarfaraz Nawaz, ***@***.***> wrote: You'd have to stay inside the scope of the threadpool until all tasks are done like this. Please show me how you stay inside the scope of the thread pool! Show me the code! You cannot do that easily! — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#32 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABIKNVLT7OMPX6L4V2ZNSL3TR3XYFANCNFSM4PE3CCIQ> .

[snawaz]

@tom-leys

Well, my initial example does. Run it.

You mean this code (quoting from your post):

    tp::ThreadPool pool;

    std::packaged_task<int()> t([]()
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            return 42;
        });

    std::future<int> r = t.get_future();

    pool.post(t);

But this example makes an arbitrary assumption that you need to wait for 1 milliseconds only? How exactly do you know that by that time all tasks will be done? How do you know how long should you wait?

[tom-leys]

Sleeping for 1 ms is done in a subthread, which is run inside the pool.

r.get() will block however long that subtask takes.

You cut the last line off the example, the line which actually contains r.get() inside an assert.

I've annotated each line below with the thread which runs it. Either A (foreground) or B (a thread in the pool). I also put a sequence number. It is unclear if the sleep begins before or after r.get() is entered but it is guaranteed that r.get() will wait "the right time" and will NOT RETURN (blocking thread A) until the task is done.

    tp::ThreadPool pool;        // A Thread < 1--- Make a pool. Lasts until example is finished.

    std::packaged_task<int()> t([]()  // A 2 <--- Declare a task. It doesn't run right now.
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(1)); // B some time after 4 -- Sleep for 1 ms 
            return 42; // B 6 <--- Return value for the task
        });

    std::future<int> r = t.get_future(); // A 3 <--- Get the future which will block and return the task's value

    pool.post(t); // A 4 <---- Tell the thread pool to start the task

    ASSERT_EQ(42, r.get()); // A 5 <--- Begin sleeping inside r.get() waiting for thread to finish
   // A 7 <--- r.get() has stopped blocking thread A since the task was done.

---

What we are waiting on is a future r, provided by a packaged_task t
https://en.cppreference.com/w/cpp/thread/packaged_task
The class template std::packaged_task wraps any Callable target (function, lambda expression, bind expression, or another function object) so that it can be invoked asynchronously. Its return value or exception thrown is stored in a shared state which can be accessed through std::future objects.

https://en.cppreference.com/w/cpp/thread/future
An asynchronous operation (created via std::async, std::packaged_task, or std::promise) can provide a std::future object to the creator of that asynchronous operation.

The creator of the asynchronous operation can then use a variety of methods to query, wait for, or extract a value from the std::future. These methods may block if the asynchronous operation has not yet provided a value

[snawaz]

@tom-leys

Sleeping for 1 ms is done in a subthread, which is run inside the pool.

r.get() will block however long that subtask takes.

Yes. It'll block as long as that particular task takes to complete. Not the other tasks. So once this particular task is done, the pool goes out of scope, without completing the other tasks.

You cut the last line off the example, the line which actually contains r.get() inside an assert.

That does not help. Remember, it does not ensure other tasks to complete.

Here is what I meant (I've added add_more_tasks_from_a_different_thread(pool) line to your example):

    tp::ThreadPool pool;

    std::packaged_task<int()> t([]()
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            return 42;
        });

    std::future<int> r = t.get_future();

    pool.post(t);

    add_more_tasks_from_a_different_thread(pool); // pool is taken by reference

    ASSERT_EQ(42, r.get());

Now since add_more_tasks_from_a_different_thread() runs in a different thread and it adds more tasks to the pool. Those tasks might not be executed by the pool at all if your t finishes first.