[Issue 4070][pulsar-client-cpp] Fix for possible deadlock when closing Pulsar client

[heronr]

Fixes #4070

Motivation

This change is to fix a possible deadlock that can occur when closing the Pulsar client that is caused by the ExecutorService worker thread attempting to join itself.

Modifications

The close() method on the ExecutorService will now not join the worker_ thread if its thread id is the same as the calling thread. The type of worker_ was changed to std::thread to allow for the check since the thread id is not exposed by boost::asio::detail::thread.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API: no
• The schema: no
• The default values of configurations: no
• The wire protocol: no
• The rest endpoints: no
• The admin cli options: no
• Anything that affects deployment: no

Documentation

• Does this pull request introduce a new feature? no

[merlimat]

@heronr I think there might be a possible issue with this change.

I'm getting a segfault when running tests and tests failing (for unrelated issues):

#0  __GI_raise (sig=sig@entry=6) at ../sysdeps/unix/sysv/linux/raise.c:50
#1  0x00007f351d252899 in __GI_abort () at abort.c:79
#2  0x00007f351d4d994a in ?? () from /lib/x86_64-linux-gnu/libstdc++.so.6
#3  0x00007f351d4e535c in ?? () from /lib/x86_64-linux-gnu/libstdc++.so.6
#4  0x00007f351d4e53c7 in std::terminate() () from /lib/x86_64-linux-gnu/libstdc++.so.6
#5  0x000055708bcafbbf in std::thread::~thread (this=0x55708c947268, __in_chrg=<optimized out>) at /usr/include/c++/9/thread:139
#6  0x00007f351da183f2 in pulsar::ExecutorService::~ExecutorService (this=0x55708c947250, __in_chrg=<optimized out>) at /pulsar/pulsar-client-cpp/lib/ExecutorService.cc:52

where we have:

    ~thread()
    {
      if (joinable())
        std::terminate();
    }

So I think that in this case we have not called join() before the thread was actually destroyed.

[heronr]

This implies that there is thread leak in certain cases, probably caused by the Executor service being destroyed in the context of the worker thread. I will try to run the tests locally and track this down, but it may prove difficult to resolve.

[merlimat]

@heronr I don't think there's a thread leak. Rather, the error seems to be that the std::thread object is destroyed before the actual thread was stopped.

The main issue I see here is that the ~std::thread() is being invoked from the thread itself and therefore it's not "joined" and it cannot be "joined" under any circumstance.

I believe the only solution is that we should ensure, that ~ExecutorService() is never called from the same thread.

[heronr]

Agreed on your assessment. I tracked one instance of a non-joined thread to the HTTPLookupService owning the ExecutorServiceProvider that it posts its own work onto. Now the provider it uses is owned by the ClientImpl, but this just exposed a different non-joined thread caused by the ClientImpl itself being destroyed on the ExecutorService worker thread. Here is the callstack

Ultimately because shared_ptrs are being used to manage the lifetime of objects that then post work to ExecutorServices owned by those same objects, we can encounter a non-joinable thread since there is not fine control over when the ref count goes to 0.

If we can designate the pulsar::Client as non-copyable then its destructor can force a shutdown() on the ClientImpl and ensure that all outstanding worker threads are joined at that point as a result. The latest exception that I linked stems from the fact that if a pulsar::Client is destructed without first calling shutdown, any outstanding work on an ExecutorService thread will likely result in the destruction of the ClientImpl on that thread and make it impossible to join.

[merlimat]

@heronr I'm not sure that fixes the underlying issue. It still will be triggering (another) std::thread object being destroyed from itself.

Actually, I think the solution should be easy: just use std::sthread::detach() (http://www.cplusplus.com/reference/thread/thread/detach/) on top of your original commit:

if (std::this_thread::get_id() != worker_.get_id() && worker_.joinable()) {
    worker_.join();
} else {
    worker_.detach();
}

[heronr]

Yes, I also considered using detach() as it is the path of least resistance. I just can't help but feel like it's a workaround solution that sweeps the underlying problem under the rug.
That being said, for the purposes of this PR I will go ahead and use std::thread::detach() to remove the deadlock. I will revisit this later with a (hopefully) better solution.