Client does not shut down properly when cluster connection attempts all fail (#285)

ihsandemir · web-flow · commit 217484bb6ac4 · 2017-06-05T12:20:17.000+03:00
* Fix for shutting down the client properly if the cluster connection attempts all fail. The Thread join now will detect if it is being called from running thread and avoid deadlock.

* Corrected the shutdown order of services, changed to be the same as Java client.

* Corrected what is being tested at IssueTest.issue221.

* Prevented client destruction before shutdown. 

* Fixes when IOSelector sets isAlive to true and prevents wakeup if the wakeup socket is not initialized yet.

* Added connection closing when ConnectionManager is shutting down. 

* Corrected the error message printing for protocol messages.
diff --git a/hazelcast/include/hazelcast/client/connection/Connection.h b/hazelcast/include/hazelcast/client/connection/Connection.h
@@ -68,7 +68,7 @@ namespace hazelcast {
 
                 void connect(int timeoutInMillis);
 
-                void close();
+                void close(const char *closeReason = NULL);
 
                 void write(protocol::ClientMessage *message);
 
diff --git a/hazelcast/include/hazelcast/client/exception/ProtocolExceptions.h b/hazelcast/include/hazelcast/client/exception/ProtocolExceptions.h
@@ -33,8 +33,9 @@ namespace hazelcast {
         namespace exception {
             class HAZELCAST_API ProtocolException : public IException {
             public:
-                ProtocolException(const std::string& source, const std::string& message, int32_t errorNo, int32_t causeCode)
-                        : IException(source, message), errorCode(errorNo), causeErrorCode(causeCode) {
+                ProtocolException(const std::string& message, const std::string& details, int32_t errorNo,
+                                  int32_t causeCode)
+                        : IException("Cluster", details), errorCode(errorNo), causeErrorCode(causeCode) {
                 }
 
                 ProtocolException(const std::string& source, const std::string& message)
@@ -56,8 +57,8 @@ namespace hazelcast {
 #define DEFINE_PROTOCOL_EXCEPTION(ClassName) \
             class HAZELCAST_API ClassName : public ProtocolException {\
             public:\
-                ClassName(const std::string& source, const std::string& message, int32_t errorCode, int32_t causeCode) \
-                    : ProtocolException(source, message, errorCode, causeCode) {\
+                ClassName(const std::string& message, const std::string& details, int32_t errorNo, int32_t causeCode) \
+                    : ProtocolException(message, details, errorNo, causeCode) {\
                 }\
                 ClassName(const std::string& source, const std::string& message) \
                     : ProtocolException(source, message) {\
diff --git a/hazelcast/include/hazelcast/client/spi/LifecycleService.h b/hazelcast/include/hazelcast/client/spi/LifecycleService.h
@@ -16,9 +16,6 @@
 //
 // Created by sancar koyunlu on 6/17/13.
 
-
-
-
 #ifndef HAZELCAST_LIFECYCLE_SERVICE
 #define HAZELCAST_LIFECYCLE_SERVICE
 
@@ -67,7 +64,7 @@ namespace hazelcast {
                 std::set<LifecycleListener *> listeners;
                 util::Mutex listenerLock;
                 util::AtomicBoolean active;
-
+                util::Mutex shutdownLock;
             };
 
         }
diff --git a/hazelcast/include/hazelcast/util/Closeable.h b/hazelcast/include/hazelcast/util/Closeable.h
@@ -27,7 +27,7 @@ namespace hazelcast {
         public:
             virtual ~Closeable();
 
-            virtual void close() = 0;
+            virtual void close(const char *closeReason) = 0;
         };
     }
 }
diff --git a/hazelcast/include/hazelcast/util/IOUtil.h b/hazelcast/include/hazelcast/util/IOUtil.h
@@ -46,7 +46,7 @@ namespace hazelcast {
                 return value;
             }
 
-            static void closeResource(Closeable *closable);
+            static void closeResource(Closeable *closable, const char *closeReason = NULL);
 
         };
     }
diff --git a/hazelcast/src/hazelcast/client/connection/ClusterListenerThread.cpp b/hazelcast/src/hazelcast/client/connection/ClusterListenerThread.cpp
@@ -84,6 +84,7 @@ namespace hazelcast {
                                 util::ILogger::getLogger().severe(
                                         std::string("Error while connecting to cluster! =>") + e.what());
                                 isStartedSuccessfully = false;
+                                clientContext.getLifecycleService().shutdown();
                                 startLatch.countDown();
                                 return;
                             }
@@ -105,7 +106,7 @@ namespace hazelcast {
 
                         clientContext.getConnectionManager().onCloseOwnerConnection();
                         if (deletingConnection.compareAndSet(false, true)) {
-                            util::IOUtil::closeResource(conn.get());
+                            util::IOUtil::closeResource(conn.get(), "Error while listening cluster events");
                             conn.reset();
                             deletingConnection = false;
                             clientContext.getLifecycleService().fireLifecycleEvent(LifecycleEvent::CLIENT_DISCONNECTED);
@@ -117,7 +118,7 @@ namespace hazelcast {
 
             void ClusterListenerThread::stop() {
                 if (deletingConnection.compareAndSet(false, true)) {
-                    util::IOUtil::closeResource(conn.get());
+                    util::IOUtil::closeResource(conn.get(), "Cluster listener thread is stopping");
                     conn.reset();
                     deletingConnection = false;
                 }
diff --git a/hazelcast/src/hazelcast/client/connection/Connection.cpp b/hazelcast/src/hazelcast/client/connection/Connection.cpp
@@ -89,7 +89,7 @@ namespace hazelcast {
                 outputSocketStream.write(PROTOCOL);
             }
 
-            void Connection::close() {
+            void Connection::close(const char *closeReason) {
                 if (!live.compareAndSet(true, false)) {
                     return;
                 }
@@ -98,8 +98,10 @@ namespace hazelcast {
                 int socketId = socket->getSocketId();
                 
                 std::stringstream message;
-                message << "Closing connection (id:" << connectionId << ") to " << serverAddr << " with socket id " << socketId <<
-                        (_isOwnerConnection ? " as the owner connection." : ".");
+                message << "Closing connection (id:" << connectionId << ") to " << serverAddr <<
+                        " with socket id " << socketId <<
+                        (_isOwnerConnection ? " as the owner connection." : ". ") <<
+                        (NULL != closeReason ? closeReason : "");
                 util::ILogger::getLogger().warning(message.str());
                 if (!_isOwnerConnection) {
                     readHandler.deRegisterSocket();
diff --git a/hazelcast/src/hazelcast/client/connection/ConnectionManager.cpp b/hazelcast/src/hazelcast/client/connection/ConnectionManager.cpp
@@ -16,6 +16,7 @@
 //
 // Created by sancar koyunlu on 8/21/13.
 
+#include <boost/foreach.hpp>
 #include "hazelcast/util/Util.h"
 #include "hazelcast/client/protocol/AuthenticationStatus.h"
 #include "hazelcast/client/exception/AuthenticationException.h"
@@ -69,6 +70,10 @@ namespace hazelcast {
 
             void ConnectionManager::shutdown() {
                 live = false;
+                // close connections
+                BOOST_FOREACH(boost::shared_ptr<Connection> connection ,  connections.values()) {
+                                connection->close("Hazelcast client is shutting down");
+                            }
                 heartBeater.shutdown();
                 if (heartBeatThread.get() != NULL) {
                     heartBeatThread->cancel();
@@ -350,7 +355,7 @@ namespace hazelcast {
 
             void ConnectionManager::checkLive() {
                 if (!live) {
-                    throw exception::HazelcastException("ConnectionManager is not active!");
+                    throw exception::HazelcastException("ConnectionManager", "ConnectionManager is not active!");
                 }
             }
 
@@ -384,7 +389,7 @@ namespace hazelcast {
                 } else {
                     ownerConnectionFuture.close();
                 }
-                util::IOUtil::closeResource(&connection);
+                util::IOUtil::closeResource(&connection, "Heartbeat failed");
             }
 
             void ConnectionManager::removeEndpoint(const Address &address) {
diff --git a/hazelcast/src/hazelcast/client/connection/IOSelector.cpp b/hazelcast/src/hazelcast/client/connection/IOSelector.cpp
@@ -42,7 +42,6 @@ namespace hazelcast {
             :connectionManager(connectionManager) {
                 t.tv_sec = 5;
                 t.tv_usec = 0;
-                isAlive = true;
             }
 
             void IOSelector::staticListen(util::ThreadArgs &args) {
@@ -55,6 +54,10 @@ namespace hazelcast {
             }
 
             void IOSelector::wakeUp() {
+                if (!wakeUpSocket.get()) {
+                    return;
+                }
+
                 int wakeUpSignal = 9;
                 try {
                     wakeUpSocket->send(&wakeUpSignal, sizeof(int));
@@ -91,6 +94,7 @@ namespace hazelcast {
                     sleepingSocket->setBlocking(false);
                     wakeUpSocketSet.insertSocket(sleepingSocket.get());
                     wakeUpListenerSocketId = sleepingSocket->getSocketId();
+                    isAlive = true;
                     return true;
                 } else {
                     util::ILogger::getLogger().severe("IOSelector::initListenSocket " + std::string(strerror(errno)));
@@ -99,7 +103,14 @@ namespace hazelcast {
             }
 
             void IOSelector::shutdown() {
-                isAlive = false;
+                if (!isAlive.compareAndSet(true, false)) {
+                    return;
+                }
+                try {
+                    wakeUp();
+                } catch (exception::IOException &) {
+                    // suppress io exception
+                }
             }
 
             void IOSelector::addTask(ListenerTask *listenerTask) {
diff --git a/hazelcast/src/hazelcast/client/connection/OwnerConnectionFuture.cpp b/hazelcast/src/hazelcast/client/connection/OwnerConnectionFuture.cpp
@@ -91,7 +91,7 @@ namespace hazelcast {
                 std::stringstream message;
                 message << "Closing owner connection to " << currentOwnerConnection->getRemoteEndpoint();
                 util::ILogger::getLogger().finest(message.str());
-                util::IOUtil::closeResource(currentOwnerConnection.get());
+                util::IOUtil::closeResource(currentOwnerConnection.get(), message.str().c_str());
                 markAsClosed();
             }
         }
diff --git a/hazelcast/src/hazelcast/client/spi/ClusterService.cpp b/hazelcast/src/hazelcast/client/spi/ClusterService.cpp
@@ -69,7 +69,9 @@ namespace hazelcast {
             }
 
             void ClusterService::shutdown() {
-                active = false;
+                if (!active.compareAndSet(true, false)) {
+                    return;
+                }
                 if (NULL != clusterThread.getThread()) {
                     // avoid anyone waiting on the start latch to get stuck
                     clusterThread.startLatch.countDown();
diff --git a/hazelcast/src/hazelcast/client/spi/LifecycleService.cpp b/hazelcast/src/hazelcast/client/spi/LifecycleService.cpp
@@ -64,13 +64,16 @@ namespace hazelcast {
             }
 
             void LifecycleService::shutdown() {
-                if (!active.compareAndSet(true, false))
+                util::LockGuard guard(shutdownLock);
+
+                if (!active.compareAndSet(true, false)) {
                     return;
+                }
                 fireLifecycleEvent(LifecycleEvent::SHUTTING_DOWN);
-                clientContext.getInvocationService().shutdown();
-                clientContext.getPartitionService().shutdown();
-                clientContext.getClusterService().shutdown();
                 clientContext.getConnectionManager().shutdown();
+                clientContext.getClusterService().shutdown();
+                clientContext.getPartitionService().shutdown();
+                clientContext.getInvocationService().shutdown();
                 clientContext.getNearCacheManager().destroyAllNearCaches();
                 fireLifecycleEvent(LifecycleEvent::SHUTDOWN);
             }
diff --git a/hazelcast/src/hazelcast/util/IOUtil.cpp b/hazelcast/src/hazelcast/util/IOUtil.cpp
@@ -22,10 +22,10 @@
 
 namespace hazelcast {
     namespace util {
-        void IOUtil::closeResource(Closeable *closable) {
+        void IOUtil::closeResource(Closeable *closable, const char *closeReason) {
             if (closable != NULL) {
                 try {
-                    closable->close();
+                    closable->close(closeReason);
                 } catch (client::exception::IException& e) {
                     std::stringstream message;
                     message << "closeResource failed" << e.what();
diff --git a/hazelcast/src/hazelcast/util/Thread.cpp b/hazelcast/src/hazelcast/util/Thread.cpp
@@ -50,6 +50,7 @@ namespace hazelcast {
         , isJoined(false)
 		, isInterrupted(false){
             init(func, arg0, arg1, arg2, arg3);
+
         }
 
         long Thread::getThreadID() {
@@ -92,6 +93,11 @@ namespace hazelcast {
             if (!isJoined.compareAndSet(false, true)) {
                 return true;
             }
+            if (id == getThreadID()) {
+                // called from inside the thread, deadlock possibility
+                return false;
+            }
+
             DWORD err = WaitForSingleObject(thread, INFINITE);
             if (err != WAIT_OBJECT_0) {
                 return false;
@@ -197,6 +203,12 @@ namespace hazelcast {
             if (!isJoined.compareAndSet(false, true)) {
                 return true;
             }
+
+            if (pthread_equal(thread, pthread_self())) {
+                // called from inside the thread, deadlock possibility
+                return false;
+            }
+
             int err = pthread_join(thread, NULL);
             if (EINVAL == err || ESRCH == err || EDEADLK == err) {
                 isJoined = false;
diff --git a/hazelcast/test/src/cluster/ClientConnectionTest.cpp b/hazelcast/test/src/cluster/ClientConnectionTest.cpp
@@ -48,13 +48,6 @@ namespace hazelcast {
                 ASSERT_THROW(HazelcastClient client(config), exception::IllegalStateException);
             }
 
-            TEST_F(ClientConnectionTest, testTcpSocketConnectionTimeout_withIntMax) {
-                HazelcastServer instance(*g_srvFactory, true);
-                ClientConfig config;
-                config.addAddress(Address("8.8.8.8", 5701));
-                ASSERT_THROW(HazelcastClient client(config), exception::IllegalStateException);
-            }
-
             #ifdef HZ_BUILD_WITH_SSL
             TEST_F(ClientConnectionTest, testSslSocketTimeoutToOutsideNetwork) {
                 HazelcastServer instance(*g_srvFactory, true);
diff --git a/hazelcast/test/src/cluster/ClusterTest.cpp b/hazelcast/test/src/cluster/ClusterTest.cpp
diff --git a/hazelcast/test/src/issues/IssueTest.cpp b/hazelcast/test/src/issues/IssueTest.cpp
diff --git a/hazelcast/test/src/util/ClientUtilTest.cpp b/hazelcast/test/src/util/ClientUtilTest.cpp

Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ namespace hazelcast {`
`27`	`27`	`public:`
`28`	`28`	`virtual ~Closeable();`
`29`	`29`
`30`		`- virtual void close() = 0;`
	`30`	`+ virtual void close(const char *closeReason) = 0;`
`31`	`31`	`};`
`32`	`32`	`}`
`33`	`33`	`}`
Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ namespace hazelcast {`
`46`	`46`	`return value;`
`47`	`47`	`}`
`48`	`48`
`49`		`- static void closeResource(Closeable *closable);`
	`49`	`+ static void closeResource(Closeable closable, const char closeReason = NULL);`
`50`	`50`
`51`	`51`	`};`
`52`	`52`	`}`
Original file line number	Diff line number	Diff line change
`@@ -91,7 +91,7 @@ namespace hazelcast {`
`91`	`91`	`std::stringstream message;`
`92`	`92`	`message << "Closing owner connection to " << currentOwnerConnection->getRemoteEndpoint();`
`93`	`93`	`util::ILogger::getLogger().finest(message.str());`
`94`		`- util::IOUtil::closeResource(currentOwnerConnection.get());`
	`94`	`+ util::IOUtil::closeResource(currentOwnerConnection.get(), message.str().c_str());`
`95`	`95`	`markAsClosed();`
`96`	`96`	`}`
`97`	`97`	`}`