my_application deadlocks when closing with ThreadPool threads running

[DaMilyutin]

I'm continuing to learn sixty fps with multiple workers example.

I've used thread pool to run worker threads.
(Actually, I tried two implementations from github.)

And it turns out application deadlocks (my_application.exe is seen in task manager) on close button pressed when I use

        pool.add(do_job, id, latency);
        //std::thread(do_job, id, latency).detach(); // insteaad of this 

I've managed to fix this using std::terminate

    ui->run();
    std::cout << "Exiting" << std::endl;
    std::terminate(); // this helps
    return 0;

But I'm not sure if it is right thing to do.
I'll attach code if you request.

[DaMilyutin]

Actually. had to do tricks with thread pool, because thread pool joins on destuction.
Since main window is closed and worker threads have blocking_invoke_from_event_loop it deadlocks, as I can guess.
Had first to detach all worker threads.
Then it terminates fine.
std::terminate didn't work as I expected and probably is not required.

Funny thing was that my_application.exe was not visible in task manager but not accessible as it still ran.
So my attempts to recompile lead to linker access errors and I could not delete my_application.exe.

I hope this research helps anyone. :-)

[ogoffart]

I guess the reason it deadlocks is because the thread are calling blocking_invoke_from_event_loop, but the event loop is already exited, so your worker thread are going to block forever. blocking_invoke_from_event_loop assumes that there is an event loop running in the main thread that will pick up its events.

This is a bit tricky to get right as you will ahve to do something like this (pseudo-code)

 // ...
ui->run();
 // this function exit when the window is closed, but you probably don't want that, you want to keep the 
 // event loop running until all the worker threads are finished.

 // send a message somehow to all the worker thread so they stop their work.
 pool.tell_the_threads_to_stop(); 

while (pool.still_working()) {
    sixtyfps::run_event_loop()
}
// ...


// ...
// in the worker thread, your code must be like this:
for( /* that's your busy loop */ ) {
   // .. do_some_small_work
   sixtyfps::blocking_invoke_from_event_loop([&]{ 
      // update the progress
   });
   std::unique_lock lock(pool.mutex);
   if (pool.must_stop()) {
       // tell the event loop to stop (NOT BLOCKING)
       sixtyfps::invoke_from_event_loop([&]{ 
          sixtyfps::quit_event_loop();
       }
       pool.notify_thread_finished(this);
       return; // this should exist the working thread
   }
}
pool.notify_thread_finished(this);

So the idea is to make sure that the event loop runs long enough so that no worker thread can be still calling blocked_invoke_from_event_loop. And you must be carefull to signal the event loop to close when all threads are finished.

I hope this helps.

[DaMilyutin]

@ogoffart, can you help me again with this?
I'm trying to reproduce your approach.
Maybe, I understood things not right...

Please, clarify. Is my understanding of logic right?

1. Let us have window closed when there are some worker threads calling blocking_invoke_from_event_loop
2. Event loop stops (?probably?)
3. (??) blocking_invoke_from_event_loop in each worker thread waits until event loop is ran again
4. I set stop flag and tell pool to drop jobs that are waiting in queue
   (jobs in worker threads are locked)
5. (??) I invoke run_event_loop
6. (??) All worker threads called blocking_invoke_from_event_loop wake up
   I wait jobs to finish:

    while(pool.activeJobs() > 0)
    {
        std::cout << "...left " << pool.activeJobs() << " jobs" << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds{1});
        sixtyfps::run_event_loop(); // not sure if I need to call it again
    }

6. (??) Worker threads read stop flag and finish their jobs as they can.
7. As worker threads finish their job and pool.activeJobs() tuns to 0 I call
   quit_event_loop()

Now it deadlocks after I re-run event loop

    ui->run();
    //pool.destroy_detach(); // older hack which worked
    std::cout << "exiting!" << std::endl;
    sixtyfps::run_event_loop(); // <--- it deadlocks here already!
    pool.clear();
    stop = true; // stop flag
    while(pool.activeJobs() > 0)
    {
        std::cout << "...left " << pool.activeJobs() << " jobs" << std::endl; 
        // this output was seem if `run_event_loop` above was commented
        
        std::this_thread::sleep_for(std::chrono::milliseconds{1});
        sixtyfps::run_event_loop(); // not sure if I need to call it again
    }
    sixtyfps::quit_event_loop(); 
    std::cout << "done!" << std::endl;

My all pile of code is below. I can not insist you to look this through.
But maybe we will figure out something useful from it.
Or at least I'll understand why it is not working as I expect.

    auto ui = AppWindow::create();

    auto task_data_model = std::make_shared<sixtyfps::VectorModel<ListItemData>>();
    auto task_id2index = std::make_shared<IdMap>();

    ui->set_task_data_model(task_data_model);

    auto counter = F60_PROPERTY(counter, int, ui);// make_property<int>([ui](){ return ui->get_counter();}, [ui](int i){ui->set_counter(i);});

    dbr::cc::ThreadPool pool;
    std::atomic<bool>   stop = false; // stop flag

    auto do_job = [&](int id, float latency){
            auto const period = std::chrono::milliseconds{int(10.*latency)};
            auto onExit = [&]()
            {
                auto r = task_id2index->id2row[id];
                task_data_model->erase(r);
                task_id2index->erase(r);
            };

            for(int i = 1; i <= 100; ++i)
            {
                if(stop)
                {
                    std::cout << "terminate task" << id << std::endl;
                    sixtyfps::blocking_invoke_from_event_loop(onExit);
                    return;
                }
                std::this_thread::sleep_for(period);
                sixtyfps::blocking_invoke_from_event_loop([&]{
                    auto r = task_id2index->id2row[id];
                    task_data_model->set_row_data(r, ListItemData{id, float(i)});
                });
            }

            std::this_thread::sleep_for(std::chrono::seconds{1});
            sixtyfps::blocking_invoke_from_event_loop([&] {                
                counter += 1;
                onExit();
            });
        };

    ui->on_request_increase_value([&]{
        int id = nextId();
        float latency = ui->get_latency();
        int r = task_data_model->row_count(); 
        task_data_model->push_back(ListItemData{id, 0.});
        task_id2index->push(id, r);
        pool.add(do_job, id, latency);
        //std::thread(do_job, id, latency).detach();
    });
    
    ui->run();
    //pool.destroy_detach(); // older hack which worked
    std::cout << "exiting!" << std::endl;
    sixtyfps::run_event_loop(); // <--- it deadlocks here already!
    pool.clear();
    stop = true;

    while(pool.activeJobs() > 0)
    {
        std::cout << "...left " << pool.activeJobs() << " jobs" << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds{1});
        sixtyfps::run_event_loop();
    }
    sixtyfps::quit_event_loop();
    std::cout << "done!" << std::endl;
    return 0;

[ogoffart]

I think your understanding is correct.

     ui->run();
    std::cout << "exiting!" << std::endl;
   // yes, of course this would deadlock there because nobody will try to exit the event loop.
  // you only must start the event loop if you know that another thread will exit it.
    // sixtyfps::run_event_loop(); // <--- it deadlocks here already!
    pool.clear();
    stop = true;

    while(pool.activeJobs() > 0)
    {
        std::cout << "...left " << pool.activeJobs() << " jobs" << std::endl;

// what is that for, there should be no need to wait
        std::this_thread::sleep_for(std::chrono::milliseconds{1});
        sixtyfps::run_event_loop();
    }

In my code example, i added a

       sixtyfps::invoke_from_event_loop([&]{ 
          sixtyfps::quit_event_loop();
       }

from the main thread to tell the event loop to only process one event.

Maybe what's missing is a sixtyfps::process_events or something similar that only process the queued event and return, so that one does not need the quit_event_loop hack.

[DaMilyutin]

Can you explain once more why I should add

       sixtyfps::invoke_from_event_loop([&]{ 
          sixtyfps::quit_event_loop();
       }

in worker?

What I can't get is... When I close, event loop stops anyway. So worker threads will probably sleep on blocking.
And main thread will try to join so it will all sleep until event loop is re-started.
However, calling this code above in worker may not run if main loop stopped earlier.

Does event loop run in main or separate thread?

[ogoffart]

The event loop runs in the main thread.
When all windows are closed, the eventloop stops.

If there are still threads blocked, we need to do something about that otherwise this is a deadlock. For example, we can restart the event loop. But then we somehow need a way to stop it.

So there is what should happen:

1. the event loop exits because the windows are closed
2. If threads are still runing, somehow tell them they need to stop.
3. in the main thread: sixtyfps::invoke_from_event_loop([&]{ sixtyfps::quit_event_loop(); } tell the event loop to only run once.
4. run the event loop (but that will be only one iteration)
5. if there are still thread running, go to 3

the step 3 is just going to make the event loop run once, this is not ideal because the steps 3 - 4 - 5 are in fact a busy wait.
Doing this correctly is a bit more involved:

// global variable
std::mutex  global_quit_mutex;
int global_waiting_threads;


// ...
// in main()
 ui->run();
 pool.tell_the_threads_to_stop(); 
 while (pool.still_working()) {
   global_quit_mutex.lock(); // because we don't want to start the event loop if all the threads are done
   if (!global_waiting_threads) {
       //  no thread are waiting, we're done.
       global_quit_mutex.unlock();
       return;
   }
     // We will unlock the mutex only when we are sure our loop is started
     sixtyfps::invoke_from_event_loop([&]{ global_quit_mutex.unlock();  }
     sixtyfps::run_event_loop();
  }
  // ...
  pool.join();


// ...
// At the end of the worker thread
   {
       std::unique_lock lock(global_quit_mutex);
       global_waiting_threads--;
   }
   sixtyfps::invoke_from_event_loop([&]{ sixtyfps::quit_event_loop();  }
}

Please note that i haven't tested this code and i might have made mistake :-s

I realize this is difficult and we might want to add some helper functions in sixtyfps to simplify that

Maybe @tronical will have some idea of a good API

[tronical]

We discussed this a bit and think that it might make sense to add API to SixtyFPS to make it easier to create a robust interaction between worker threads and the event loop, especially in the shutdown phase. I'll summarise our thinking in a ticket (will update here, of course).

[tronical]

I've tried to summarise this in #658 - please feel free to comment there, too :)