Better fix for ThreadPool/ORequestThread life cycle

This is a follow up to d015384e1d98fe77fd59339044f58efb1ab9fb25 "Fixed
ThreadPool (and dependent ORequestThread) life cycle" that still had some
problems:

* First, if Bridge::terminate was first entered from the reader or writer
thread, it would not join on that thread, so that thread could still be running
during exit.

That has been addressed by giving Bridge::dispose new semantics:  It waits until
both Bridge::terminate has completed (even if that was called from a different
thread) and all spawned threads (reader, writer, ORequestThread workers) have
been joined.  (This implies that Bridge::dispose must not be called from such a
thread, to avoid deadlock.)

* Second, if Bridge::terminate was first entered from an ORequestThread, the
call to uno_threadpool_dispose(0) to join on all such worker threads could
deadlock.

That has been addressed by making the last call to uno_threadpool_destroy wait
to join on all worker threads, and by calling uno_threadpool_destroy only from
the final Bridge::terminate (from Bridge::dispose), to avoid deadlock.  (The
special semantics of uno_threadpool_dispose(0) are no longer needed and have
been removed, as they conflicted with the fix for the third problem below.)

* Third, once uno_threadpool_destroy had called uno_threadpool_dispose(0), the
ThreadAdmin singleton had been disposed, so no new remote bridges could
successfully be created afterwards.

That has been addressed by making ThreadAdmin a member of ThreadPool, and making
(only) those uno_ThreadPool handles with overlapping life spans share one
ThreadPool instance (which thus is no longer a singleton, either).
Additionally, ORequestThread has been made more robust (in the style of
salhelper::Thread) to avoid races.

Change-Id: I2cbd1b3f9aecc1bf4649e482d2c22b33b471788f

6 files changed, 161 insertions, 77 deletions

diff --git a/binaryurp/source/bridge.cxx b/binaryurp/source/bridge.cxx
index 2d1f622369a3..fbb70a40c835 100644
--- a/binaryurp/source/bridge.cxx
+++ b/binaryurp/source/bridge.cxx
@@ -106,11 +106,9 @@ extern "C" void SAL_CALL freeProxyCallback(
     static_cast< Proxy * >(pProxy)->do_free();
 }
 
-void joinThread(salhelper::Thread * thread) {
+bool isThread(salhelper::Thread * thread) {
     assert(thread != 0);
-    if (thread->getIdentifier() != osl::Thread::getCurrentIdentifier()) {
-        thread->join();
-    }
+    return osl::Thread::getCurrentIdentifier() == thread->getIdentifier();
 }
 
 class AttachThread: private boost::noncopyable {
@@ -214,8 +212,8 @@ Bridge::Bridge(
         rtl::OUString(
             RTL_CONSTASCII_USTRINGPARAM(
                 "com.sun.star.bridge.XProtocolProperties::commitChange"))),
-    threadPool_(0), currentContextMode_(false), proxies_(0), calls_(0),
-    normalCall_(false), activeCalls_(0), terminated_(false),
+    state_(STATE_INITIAL), threadPool_(0), currentContextMode_(false),
+    proxies_(0), calls_(0), normalCall_(false), activeCalls_(0),
     mode_(MODE_REQUESTED)
 {
     assert(factory.is() && connection.is());
@@ -239,11 +237,14 @@ void Bridge::start() {
     rtl::Reference< Writer > w(new Writer(this));
     {
         osl::MutexGuard g(mutex_);
-        assert(threadPool_ == 0 && !writer_.is() && !reader_.is());
+        assert(
+            state_ == STATE_INITIAL && threadPool_ == 0 && !writer_.is() &&
+            !reader_.is());
         threadPool_ = uno_threadpool_create();
         assert(threadPool_ != 0);
         reader_ = r;
         writer_ = w;
+        state_ = STATE_STARTED;
     }
     // It is important to call reader_->launch() last here; both
     // Writer::execute and Reader::execute can call Bridge::terminate, but
@@ -255,60 +256,117 @@ void Bridge::start() {
     r->launch();
 }
 
-void Bridge::terminate() {
+void Bridge::terminate(bool final) {
     uno_ThreadPool tp;
-    rtl::Reference< Reader > r;
-    rtl::Reference< Writer > w;
-    Listeners ls;
+    // Make sure function-local variables (Stubs s, etc.) are destroyed before
+    // the final uno_threadpool_destroy/threadPool_ = 0:
     {
-        osl::MutexGuard g(mutex_);
-        if (terminated_) {
-            return;
+        rtl::Reference< Reader > r;
+        rtl::Reference< Writer > w;
+        bool joinW;
+        Listeners ls;
+        {
+            osl::ClearableMutexGuard g(mutex_);
+            switch (state_) {
+            case STATE_INITIAL: // via ~Bridge -> dispose -> terminate
+            case STATE_FINAL:
+                return;
+            case STATE_STARTED:
+                break;
+            case STATE_TERMINATED:
+                if (final) {
+                    g.clear();
+                    terminated_.wait();
+                    {
+                        osl::MutexGuard g2(mutex_);
+                        tp = threadPool_;
+                        threadPool_ = 0;
+                        assert(!(reader_.is() && isThread(reader_.get())));
+                        std::swap(reader_, r);
+                        assert(!(writer_.is() && isThread(writer_.get())));
+                        std::swap(writer_, w);
+                        state_ = STATE_FINAL;
+                    }
+                    assert(!(r.is() && w.is()));
+                    if (r.is()) {
+                        r->join();
+                    } else if (w.is()) {
+                        w->join();
+                    }
+                    if (tp != 0) {
+                        uno_threadpool_destroy(tp);
+                    }
+                }
+                return;
+            }
+            tp = threadPool_;
+            assert(!(final && isThread(reader_.get())));
+            if (!isThread(reader_.get())) {
+                std::swap(reader_, r);
+            }
+            w = writer_;
+            joinW = !isThread(writer_.get());
+            assert(!final || joinW);
+            if (joinW) {
+                writer_.clear();
+            }
+            ls.swap(listeners_);
+            state_ = final ? STATE_FINAL : STATE_TERMINATED;
+        }
+        try {
+            connection_->close();
+        } catch (const css::io::IOException & e) {
+            SAL_INFO("binaryurp", "caught IO exception '" << e.Message << '\'');
+        }
+        assert(w.is());
+        w->stop();
+        if (r.is()) {
+            r->join();
+        }
+        if (joinW) {
+            w->join();
+        }
+        assert(tp != 0);
+        uno_threadpool_dispose(tp);
+        Stubs s;
+        {
+            osl::MutexGuard g(mutex_);
+            s.swap(stubs_);
+        }
+        for (Stubs::iterator i(s.begin()); i != s.end(); ++i) {
+            for (Stub::iterator j(i->second.begin()); j != i->second.end(); ++j)
+            {
+                SAL_INFO(
+                    "binaryurp",
+                    "stub '" << i->first << "', '" << toString(j->first)
+                        << "' still mapped at Bridge::terminate");
+                binaryUno_.get()->pExtEnv->revokeInterface(
+                    binaryUno_.get()->pExtEnv, j->second.object.get());
+            }
+        }
+        factory_->removeBridge(this);
+        for (Listeners::iterator i(ls.begin()); i != ls.end(); ++i) {
+            try {
+                (*i)->disposing(
+                    css::lang::EventObject(
+                        static_cast< cppu::OWeakObject * >(this)));
+            } catch (const css::uno::RuntimeException & e) {
+                SAL_WARN(
+                    "binaryurp",
+                    "caught runtime exception '" << e.Message << '\'');
+            }
         }
-        tp = threadPool_;
-        std::swap(reader_, r);
-        std::swap(writer_, w);
-        ls.swap(listeners_);
-        terminated_ = true;
     }
-    try {
-        connection_->close();
-    } catch (const css::io::IOException & e) {
-        SAL_INFO("binaryurp", "caught IO exception '" << e.Message << '\'');
-    }
-    assert(w.is());
-    w->stop();
-    joinThread(r.get());
-    joinThread(w.get());
-    assert(tp != 0);
-    uno_threadpool_dispose(tp);
-    Stubs s;
+    if (final) {
+        uno_threadpool_destroy(tp);
+    }
     {
         osl::MutexGuard g(mutex_);
-        s.swap(stubs_);
-    }
-    for (Stubs::iterator i(s.begin()); i != s.end(); ++i) {
-        for (Stub::iterator j(i->second.begin()); j != i->second.end(); ++j) {
-            SAL_INFO(
-                "binaryurp",
-                "stub '" << i->first << "', '" << toString(j->first)
-                    << "' still mapped at Bridge::terminate");
-            binaryUno_.get()->pExtEnv->revokeInterface(
-                binaryUno_.get()->pExtEnv, j->second.object.get());
+        if (final) {
+            threadPool_ = 0;
         }
     }
-    factory_->removeBridge(this);
-    for (Listeners::iterator i(ls.begin()); i != ls.end(); ++i) {
-        try {
-            (*i)->disposing(
-                css::lang::EventObject(
-                    static_cast< cppu::OWeakObject * >(this)));
-        } catch (const css::uno::RuntimeException & e) {
-            SAL_WARN(
-                "binaryurp", "caught runtime exception '" << e.Message << '\'');
-        }
-    }
-    uno_threadpool_destroy(tp);
+    terminated_.set();
 }
 
 css::uno::Reference< css::connection::XConnection > Bridge::getConnection()
@@ -340,19 +398,14 @@ BinaryAny Bridge::mapCppToBinaryAny(css::uno::Any const & cppAny) {
 
 uno_ThreadPool Bridge::getThreadPool() {
     osl::MutexGuard g(mutex_);
+    checkDisposed();
     assert(threadPool_ != 0);
     return threadPool_;
 }
 
 rtl::Reference< Writer > Bridge::getWriter() {
     osl::MutexGuard g(mutex_);
-    if (terminated_) {
-        throw css::lang::DisposedException(
-            rtl::OUString(
-                RTL_CONSTASCII_USTRINGPARAM(
-                    "Binary URP bridge already disposed")),
-            static_cast< cppu::OWeakObject * >(this));
-    }
+    checkDisposed();
     assert(writer_.is());
     return writer_;
 }
@@ -822,9 +875,15 @@ bool Bridge::isCurrentContextMode() {
 }
 
 Bridge::~Bridge() {
-    if (getThreadPool() != 0) {
-        terminate();
+#if OSL_DEBUG_LEVEL > 0
+    {
+        osl::MutexGuard g(mutex_);
+        SAL_WARN_IF(
+            state_ == STATE_STARTED || state_ == STATE_TERMINATED, "binaryurp",
+            "undisposed bridge, potential deadlock ahead");
     }
+#endif
+    dispose();
 }
 
 css::uno::Reference< css::uno::XInterface > Bridge::getInstance(
@@ -885,7 +944,11 @@ rtl::OUString Bridge::getDescription() throw (css::uno::RuntimeException) {
 }
 
 void Bridge::dispose() throw (css::uno::RuntimeException) {
-    terminate();
+    // For terminate(true) not to deadlock, an external protocol must ensure
+    // that dispose is not called from a thread pool worker thread (that dispose
+    // is never called from the reader or writer thread is already ensured
+    // internally):
+    terminate(true);
     // OOo expects dispose to not return while there are still remote calls in
     // progress; an external protocol must ensure that dispose is not called
     // from within an incoming or outgoing remote call, as passive_.wait() would
@@ -900,7 +963,8 @@ void Bridge::addEventListener(
     assert(xListener.is());
     {
         osl::MutexGuard g(mutex_);
-        if (!terminated_) {
+        assert(state_ != STATE_INITIAL);
+        if (state_ == STATE_STARTED) {
             listeners_.push_back(xListener);
             return;
         }
@@ -995,7 +1059,18 @@ void Bridge::terminateWhenUnused(bool unused) {
         // That the current thread considers the bridge unused implies that it
         // is not within an incoming or outgoing remote call (so calling
         // terminate cannot lead to deadlock):
-        terminate();
+        terminate(false);
+    }
+}
+
+void Bridge::checkDisposed() {
+    assert(state_ != STATE_INITIAL);
+    if (state_ != STATE_STARTED) {
+        throw css::lang::DisposedException(
+            rtl::OUString(
+                RTL_CONSTASCII_USTRINGPARAM(
+                    "Binary URP bridge already disposed")),
+            static_cast< cppu::OWeakObject * >(this));
     }
 }
 
diff --git a/binaryurp/source/bridge.hxx b/binaryurp/source/bridge.hxx
index 8d667897d253..3ffbfbaeb43b 100644
--- a/binaryurp/source/bridge.hxx
+++ b/binaryurp/source/bridge.hxx
@@ -93,8 +93,11 @@ public:
     void start();
 
     // Internally waits for all incoming and outgoing remote calls to terminate,
-    // so must not be called from within such a call:
-    void terminate();
+    // so must not be called from within such a call; when final is true, also
+    // joins all remaining threads (reader, writer, and worker threads from the
+    // thread pool), so must not be called with final set to true from such a
+    // thread:
+    void terminate(bool final);
 
     com::sun::star::uno::Reference< com::sun::star::connection::XConnection >
     getConnection() const;
@@ -228,6 +231,9 @@ private:
 
     void terminateWhenUnused(bool unused);
 
+    // Must only be called with mutex_ locked:
+    void checkDisposed();
+
     typedef
         std::list<
             com::sun::star::uno::Reference<
@@ -240,6 +246,8 @@ private:
 
     typedef std::map< rtl::OUString, Stub > Stubs;
 
+    enum State { STATE_INITIAL, STATE_STARTED, STATE_TERMINATED, STATE_FINAL };
+
     enum Mode {
         MODE_REQUESTED, MODE_REPLY_MINUS1, MODE_REPLY_0, MODE_REPLY_1,
         MODE_WAIT, MODE_NORMAL, MODE_NORMAL_WAIT };
@@ -259,8 +267,15 @@ private:
     com::sun::star::uno::TypeDescription protPropRequest_;
     com::sun::star::uno::TypeDescription protPropCommit_;
     OutgoingRequests outgoingRequests_;
+    osl::Condition passive_;
+        // to guarantee that passive_ is eventually set (to avoid deadlock, see
+        // dispose), activeCalls_ only counts those calls for which it can be
+        // guaranteed that incrementActiveCalls is indeed followed by
+        // decrementActiveCalls, without an intervening exception
+    osl::Condition terminated_;
 
     osl::Mutex mutex_;
+    State state_;
     Listeners listeners_;
     uno_ThreadPool threadPool_;
     rtl::Reference< Writer > writer_;
@@ -271,12 +286,6 @@ private:
     std::size_t calls_;
     bool normalCall_;
     std::size_t activeCalls_;
-    osl::Condition passive_;
-        // to guarantee that passive_ is eventually set (to avoid deadlock, see
-        // dispose), activeCalls_ only counts those calls for which it can be
-        // guaranteed that incrementActiveCalls is indeed followed by
-        // decrementActiveCalls, without an intervening exception
-    bool terminated_;
 
     // Only accessed from reader_ thread:
     Mode mode_;
diff --git a/binaryurp/source/bridgefactory.cxx b/binaryurp/source/bridgefactory.cxx
index de7762f3d6ed..55a9a78ace25 100644
--- a/binaryurp/source/bridgefactory.cxx
+++ b/binaryurp/source/bridgefactory.cxx
@@ -210,7 +210,7 @@ static cppu::ImplementationEntry const services[] = {
     { &binaryurp::BridgeFactory::static_create,
       &binaryurp::BridgeFactory::static_getImplementationName,
       &binaryurp::BridgeFactory::static_getSupportedServiceNames,
-      &cppu::createSingleComponentFactory, 0, 0 },
+      &cppu::createOneInstanceComponentFactory, 0, 0 },
     { 0, 0, 0, 0, 0, 0 }
 };
 
diff --git a/binaryurp/source/incomingrequest.cxx b/binaryurp/source/incomingrequest.cxx
index 431c88505ad1..83b0030623e7 100644
--- a/binaryurp/source/incomingrequest.cxx
+++ b/binaryurp/source/incomingrequest.cxx
@@ -123,7 +123,7 @@ void IncomingRequest::execute() const {
         } catch (const std::exception & e) {
             OSL_TRACE(OSL_LOG_PREFIX "caught C++ exception '%s'", e.what());
         }
-        bridge_->terminate();
+        bridge_->terminate(false);
     } else {
         if (isExc) {
             OSL_TRACE(OSL_LOG_PREFIX "oneway method raised exception");
diff --git a/binaryurp/source/reader.cxx b/binaryurp/source/reader.cxx
index 5a4491efb61b..c59d92ae3dd9 100644
--- a/binaryurp/source/reader.cxx
+++ b/binaryurp/source/reader.cxx
@@ -149,7 +149,7 @@ void Reader::execute() {
     } catch (const std::exception & e) {
         SAL_WARN("binaryurp", "caught C++ exception '" << e.what() << '\'');
     }
-    bridge_->terminate();
+    bridge_->terminate(false);
 }
 
 void Reader::readMessage(Unmarshal & unmarshal) {
diff --git a/binaryurp/source/writer.cxx b/binaryurp/source/writer.cxx
index e1b2291ff28a..0273fa65f4a3 100644
--- a/binaryurp/source/writer.cxx
+++ b/binaryurp/source/writer.cxx
@@ -194,7 +194,7 @@ void Writer::execute() {
     } catch (const std::exception & e) {
         OSL_TRACE(OSL_LOG_PREFIX "caught C++ exception '%s'", e.what());
     }
-    bridge_->terminate();
+    bridge_->terminate(false);
 }
 
 void Writer::sendRequest(