Block the process until the threads / child processes are complete

Question

There is such code:

if(pid_t pid = fork()) // spawn child process { // parent process LOG("Try to execute smth. in child process"); return; } else { // child process doWork(); std::terminate(); }

Here a child process is generated in which some work is performed. Somewhere else (namely, at the time of completion of the process) there is this:

 // wait childern(some code may call fork()) const pid_t pid{::wait(nullptr)};

Here the process waits for the completion of the child processes (which have been hoisted earlier). So, all this fun does not work together with the -fsanintize=thread key. The code hangs on wait , because the compiler adds some child processes that do not terminate themselves.

Question. How to achieve similar behavior without wait ? So that tasks are started asynchronously, and the main process waited for the completion of these tasks in case of termination of work. Maybe with the help of threads, not processes. If I had Qt at my disposal, I would throw these doWork in QThreadPool and he would wait for the completion of all tasks in the destructor. But I don't have Qt, but I have C ++ 14 and pthread. Can someone suggest any solution?

Write on the usual C and not fool (above all to yourself) your head.
I will not answer, but simply ask: does generating and handling exceptions make it easier to solve the problem? ..
You just need to run asynchronous tasks that are guaranteed to be completed.
You remember the pid all the processes that you want to wait for completion in the array, then at the waiting point you are looking for the pid returned by wait() in this array and if you find "strike out" it.
As soon as the array became empty you waited for all the desired ones.

Croessmah croessmah 4,197 eight 18 · Accepted Answer · 2018-01-06T14:33:29

Throw in a hurry, but I think the idea is clear.

 #include <sys/types.h> #include <sys/wait.h> #include <unistd.h> #include <vector> #include <chrono> #include <thread> #include <random> #include <iostream> class fork_storage { public: fork_storage() noexcept { m_pid = fork(); } fork_storage(const fork_storage&) = delete; fork_storage& operator=(const fork_storage&) = delete; fork_storage& operator=(fork_storage&& rhv) = delete; fork_storage(fork_storage&& src) noexcept { m_pid = src.pid(); src.reset(); } ~fork_storage() { if (is_parent()) { wait(); } } bool fail() const noexcept { return m_pid < 0; } bool is_parent() const noexcept { return m_pid > 0; } void reset() noexcept { m_pid = -1; } pid_t pid() const noexcept { return m_pid; } int wait() noexcept { int status = -1; if (is_parent()) { waitpid(m_pid, &status, WUNTRACED); } reset(); return status; } private: pid_t m_pid; }; std::mt19937 &random_gen() { static std::mt19937 gen(std::chrono::system_clock::now().time_since_epoch().count()); return gen; } int main() { std::vector<fork_storage> processes; for (unsigned i = 0; i < 10; ++i) { fork_storage fork_obj; if (fork_obj.fail()) { std::cerr << "fork error\n"; } else { if (fork_obj.is_parent()) { std::string message = "process created: " + std::to_string(fork_obj.pid()) + "\n"; processes.emplace_back(std::move(fork_obj)); } else { for (auto &process: processes) { process.reset(); } processes.clear(); std::chrono::seconds sec(random_gen()() % 5 + 1); std::string message = "child " + std::to_string(getpid()) + " paused: " + std::to_string(sec.count()) + " sec\n"; std::cout << message; std::this_thread::sleep_for(sec); break; } } } while (!processes.empty()) { fork_storage &fork_obj = processes.back(); if(fork_obj.is_parent()) { std::string message = "wait process: " + std::to_string(fork_obj.pid()) + "\n"; std::cout << message; } processes.pop_back(); } }

http://rextester.com/LUZI37730

Answer 2 · 2018-01-03T06:43:53

You can use std::thread :

 #include <iostream> #include <thread> void func1() //Функция для запуска в потоке { std::cout << "Executing func1" << std::endl; } void func2(int x) //Еще одна функция с аргументом { std::cout << "Executing func2 with x=" << x << std::endl; } int main() { //Запускаем функции в отдельных потоках std::thread thread1 (func1); std::thread thread2 (func2,1); //Параллельно можем исполнять код в главном процессе std::cout << "Execute some commands in main process" << std::endl; //Ждем завершения кода в потоках thread1.join(); thread2.join(); std::cout << "Thread execution completed" << std::endl; return 0; }

In addition, you can use std::async to parallelize the process:

 #include <cstdlib> #include <iostream> #include <vector> #include <algorithm> #include <numeric> #include <future> #include <chrono> //Некоторая задача которую мы будем выполнять асинхронно template <typename RandomIt> int parallel_sum(RandomIt beg, RandomIt end) { auto len = end - beg; //определяем длину вектора if (len < 1000) return std::accumulate(beg, end, 0); //короткий вектор суммируем сразу //Рекурсивно запускаем parallel_sum для получения большего числа потоков RandomIt mid = beg + len/2; auto handle = std::async(std::launch::async, parallel_sum<RandomIt>, mid, end); int sum = parallel_sum(beg, mid); return sum + handle.get(); } int main() { //Для примера будем суммировать большой vector случайных чисел std::vector<int> v(100000); for (auto& e: v){ e = std::rand(); } //Для интереса сравним время выполнения суммирования вектора с помощью async: auto t0 = std::chrono::high_resolution_clock::now(); std::cout << "The sum is " << parallel_sum(v.begin(), v.end()) << '\n'; auto t1 = std::chrono::high_resolution_clock::now(); std::cout << std::chrono::duration_cast<std::chrono::milliseconds>(t1-t0).count() << "msec\n"; //и суммирование в одном главном потоке: t0 = std::chrono::high_resolution_clock::now(); std::cout << std::accumulate(v.begin(), v.end(), 0); t1 = std::chrono::high_resolution_clock::now(); std::cout << std::chrono::duration_cast<std::chrono::milliseconds>(t1-t0).count() << "msec\n"; }

My result is the following:

 The sum is 1746632618 9msec 17466326181msec

std::async is a high-level tool, it does not guarantee that the code will be executed in a separate thread.

Further, I give the simplest example of flow control, although this is a completely different story, I emphasize that this is the simplest example, there is still a lot to be solved, for example, sharing cout with all threads:

 #include <iostream> #include <thread> #include <vector> using namespace std; void func(const int i) { // Функция, исполняемая в потоках cout << "Thread N " << i << " created" << endl; } int main() { vector<thread> thread_pool; //создаем пул потоков на основе вектора int threadNum = 10; //количество потоков for (int i=0; i < threadNum; i++) { thread_pool.push_back(thread(func,i)); //запускаем потоки } for (auto& t : thread_pool) { t.join(); //ожидаем окончание выполнения потоков } }

That is, objects std::thread must somehow be accumulated and stored.
Well, here you can create an array, a list, a vector or whatever, create a loop, attach it in a loop, delete it in a loop
@koresh, do you understand the difference between the processes generated by the fork (in the question) and the threads (in your answer)?
Well, as far as I understand it is necessary "So that tasks are started asynchronously, and the main process waits for these tasks to be completed in case of termination. Maybe with the help of threads, not processes" for mine what you need

Block the process until the threads / child processes are complete

2 answers 2

More articles: