Multithreading worker issue

[SoftwareMechanic]

V.3.1.39

I am compiling and running a program which use 3rd party C++ library.

the 3rd party library uses multithreading. theoretically uses the number of processors as maximum number of threads, (that number is an input I give using std::thread::hardware_concurrency();)

but when this program is compiled with Emscripten and run in the browser, seems that it creates/uses a different web worker for each iteration, instead of "reusing" already created threads/workers, as it should.

this is where multithreading startsinit_future_ = std::async(std::launch::async, [this]() { process_concurrently(); });:

bool initialize(){
...
if (num_threads_ != 1) {
    collect();

    init_future_ = std::async(std::launch::async, [this]() { process_concurrently(); });

    // wait for the first element, because after init(), get() can be called.
    // so the element conversion must succeed
    initialization_outcome_ = wait_for_element();
} else {
    initialization_outcome_ = create();
}
...
}

this is processConcurrently function

void process_concurrently() {
    size_t conc_threads = num_threads_;
    if (conc_threads > tasks_.size()) {
        conc_threads = tasks_.size();
    }

    kernel_pool.reserve(conc_threads);
    for (unsigned i = 0; i < conc_threads; ++i) {
        kernel_pool.push_back(new MAKE_TYPE_NAME(Kernel)(kernel));
    }

    std::vector<std::future<geometry_conversion_task*>> threadpool;         

    for (auto& rep : tasks_) {
        MAKE_TYPE_NAME(Kernel)* K = nullptr;
        if (threadpool.size() < kernel_pool.size()) {
            K = kernel_pool[threadpool.size()];
        }

        while (threadpool.size() == conc_threads) {
            for (int i = 0; i < (int)threadpool.size(); i++) {
                auto& fu = threadpool[i];
                std::future_status status;
                status = fu.wait_for(std::chrono::seconds(0));
                if (status == std::future_status::ready) {
                    process_finished_rep(fu.get());                         

                    std::swap(threadpool[i], threadpool.back());
                    threadpool.pop_back();
                    std::swap(kernel_pool[i], kernel_pool.back());
                    K = kernel_pool.back();
                    break;
                } // if
            }   // for
        }     // while

        std::future<geometry_conversion_task*> fu = std::async(
            std::launch::async, [this](
                IfcGeom::MAKE_TYPE_NAME(Kernel)* kernel,
                const IfcGeom::IteratorSettings& settings,
                geometry_conversion_task* rep) {
                    this->create_element_(kernel, settings, rep); 
                    return rep;
                },
            K,
            std::ref(settings),
            &rep);

        if (terminating_) {
            break;
        }

        threadpool.emplace_back(std::move(fu));
    }

    for (auto& fu : threadpool) {
        process_finished_rep(fu.get());
    }

    finished_ = true;

    if (!terminating_) {
        Logger::Status("\rDone creating geometry (" + boost::lexical_cast<std::string>(all_processed_elements_.size()) +
            " objects)                                ");
    }
}

I think this could be somehow related to https://github.com/emscripten-core/emscripten/issues/8201

ps. I tried setting THREAD_POOL_SIZE to 490 (just as a test) and the iteration goes for more or less 490 objects.

this is an example on how I use the iterator

if (geom_iterator_serializer.initialize()) {
        do {

               // Do my stuff
               //Here seems that every iteration uses a new thread/worker
               //leading to errors
         } while (geom_iterator_serializer.next());
}

errors:

Tried to spawn a new thread, but the thread pool is exhausted.
This might result in a deadlock unless some threads eventually exit or the code explicitly breaks out to the event loop.
If you want to increase the pool size, use setting `-sPTHREAD_POOL_SIZE=...`.

worker.js onmessage() captured an uncaught exception: std::__2::system_error: thread constructor failed: Resource temporarily unavailable

Uncaught exception from main loop: ffbd51c6-1c0a-44cc-b833-e08ba3ca66c0:9:99729

unwind

Halting program

Pthread 0x4c147998 sent an error! blob:http://localhost:58010/ffbd51c6-1c0a-44cc-b833-e08ba3ca66c0:9: Error: thread constructor failed: Resource temporarily unavailable

any tip will be much appreciated!!

[sbc100]

Emscripten will only only create new workers when there are no unused workers in the pool.

Workers are returned to the "unused" pool each time a C++ thread exits.

If you are seeing the pool continue to grow then that means that your C++ threads probably are not exiting correctly and emscripten things that they are still running. Perhaps there is some issue with the way C++ threads are built on top of pthreads? Perhaps libc++ is pooling pthreads at some level?

I wonder if you could share a simple example of a program that continues to grow the worker pool even though the number of active C++ threads is not going up?

[SoftwareMechanic]

thank you for the response @sbc100 !

I found that even with the following code, at some point, will try to create workers:

std::cout << "Test trheads!" << std::endl

while (true) {
    std::thread test1(func);
    test1.detach();
    test1.~thread();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}

void func()
{
    std::cout << "I'm a thread!\n";
}

I can see the output printed, but after some iterations, it fails because of the same error. shouldn't detachfree the worker?

I tried also not allowing the memory to grow but it doesn't change the result.

I am sorry if this is a noob error, I am new to Emscripten, so please bear with me.

EDIT: also by setting -sPTHREAD_POOL_SIZE_STRICT=0 I face the same behaviour, but errors will not be printed

EDIT2: Using std::thread::join() instead of detach just works with that simple example, and every thread created has the same ID.

if you need more info don't hesitate to ask!

[xrui94]

You can see this explanation, and I believe you will get a good solution.