Client connection freezes and resends are missing when servers restart on heavy load

CMakeLists.txt

@@ -130,7 +130,7 @@ ENDIF(${CMAKE_SYSTEM_NAME} MATCHES "Windows")
 
 IF(${HZ_BUILD_TESTS} MATCHES "ON")
 	SET(BUILD_GTEST "ON")
-	SET(DBUILD_GMOCK "OFF")
+	SET(BUILD_GMOCK "OFF")
 	ADD_SUBDIRECTORY(hazelcast/test)
 ENDIF(${HZ_BUILD_TESTS} MATCHES "ON")
 

hazelcast/include/hazelcast/client/HazelcastClient.h

@@ -414,6 +414,7 @@ namespace hazelcast {
             /**
             * Constructs a hazelcastClient with given ClientConfig.
             * Note: ClientConfig will be copied.
+            * @param config client configuration to start the client with
             */
             HazelcastClient(ClientConfig&);
 

hazelcast/include/hazelcast/client/IMap.h

@@ -508,7 +508,7 @@ namespace hazelcast {
             */
             std::map<K, V> getAll(const std::set<K> &keys) {
                 std::vector<serialization::pimpl::Data> keySet(keys.size());
-                int i = 0;
+                size_t i = 0;
                 for (typename std::set<K>::iterator it = keys.begin(); it != keys.end(); ++it) {
                     keySet[i++] = toData(*it);
                 }
@@ -533,9 +533,9 @@ namespace hazelcast {
             */
             std::vector<K> keySet() {
                 std::vector<serialization::pimpl::Data> dataResult = proxy::IMapImpl::keySet();
-                int size = dataResult.size();
+                size_t size = dataResult.size();
                 std::vector<K> keys(size);
-                for (int i = 0; i < size; ++i) {
+                for (size_t i = 0; i < size; ++i) {
                     boost::shared_ptr<K> key = toObject<K>(dataResult[i]);
                     keys[i] = *key;
                 }
@@ -554,9 +554,9 @@ namespace hazelcast {
               */
             std::vector<K> keySet(const serialization::IdentifiedDataSerializable &predicate) {
                 std::vector<serialization::pimpl::Data> dataResult = proxy::IMapImpl::keySet(predicate);
-                int size = dataResult.size();
+                size_t size = dataResult.size();
                 std::vector<K> keys(size);
-                for (int i = 0; i < size; ++i) {
+                for (size_t i = 0; i < size; ++i) {
                     boost::shared_ptr<K> key = toObject<K>(dataResult[i]);
                     keys[i] = *key;
                 }
@@ -572,9 +572,9 @@ namespace hazelcast {
             */
             std::vector<V> values() {
                 std::vector<serialization::pimpl::Data> dataResult = proxy::IMapImpl::values();
-                int size = dataResult.size();
+                size_t size = dataResult.size();
                 std::vector<K> values(size);
-                for (int i = 0; i < size; ++i) {
+                for (size_t i = 0; i < size; ++i) {
                     boost::shared_ptr<K> value = toObject<K>(dataResult[i]);
                     values[i] = *value;
                 }
@@ -591,9 +591,9 @@ namespace hazelcast {
             */
             std::vector<V> values(const serialization::IdentifiedDataSerializable &predicate) {
                 std::vector<serialization::pimpl::Data> dataResult = proxy::IMapImpl::values(predicate);
-                int size = dataResult.size();
+                size_t size = dataResult.size();
                 std::vector<V> values(size);
-                for (int i = 0; i < size; ++i) {
+                for (size_t i = 0; i < size; ++i) {
                     boost::shared_ptr<V> value = toObject<V>(dataResult[i]);
                     values[i] = *value;
                 }

hazelcast/include/hazelcast/client/IQueue.h

@@ -151,7 +151,7 @@ namespace hazelcast {
             * @param elements the vector that elements will be drained to.
             * @return number of elements drained.
             */
-            int drainTo(std::vector<E>& elements) {
+            size_t drainTo(std::vector<E>& elements) {
                 return drainTo(elements, -1);
             }
 
@@ -162,7 +162,7 @@ namespace hazelcast {
             * @param elements vector that elements will be drained to.
             * @return number of elements drained.
             */
-            int drainTo(std::vector<E>& elements, int maxElements) {
+            size_t drainTo(std::vector<E>& elements, size_t maxElements) {
                 std::vector<serialization::pimpl::Data> coll = proxy::IQueueImpl::drainTo(maxElements);
                 for (std::vector<serialization::pimpl::Data>::const_iterator it = coll.begin(); it != coll.end(); ++it) {
                     boost::shared_ptr<E> e = context->getSerializationService().template toObject<E>(*it);
@@ -266,8 +266,7 @@ namespace hazelcast {
             }
         private:
             IQueue(const std::string& instanceName, spi::ClientContext *context)
-            : proxy::IQueueImpl(instanceName, context){
-
+            : proxy::IQueueImpl(instanceName, context) {
             }
         };
     }

hazelcast/include/hazelcast/client/connection/CallPromise.h

@@ -16,19 +16,13 @@
 //
 // Created by sancar koyunlu on 03/01/14.
 //
-
-
-
-
 #ifndef HAZELCAST_ClientCallPromise
 #define HAZELCAST_ClientCallPromise
 
-
 #include "hazelcast/util/HazelcastDll.h"
 #include "hazelcast/util/Future.h"
 #include "hazelcast/util/AtomicInt.h"
 
-#include "hazelcast/client/impl/BaseEventHandler.h"
 #include <memory>
 
 namespace hazelcast {

hazelcast/include/hazelcast/client/connection/ClusterListenerThread.h

@@ -28,6 +28,7 @@
 #include "hazelcast/client/MembershipEvent.h"
 
 #include <boost/shared_ptr.hpp>
+#include <set>
 
 #if  defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
 #pragma warning(push)
@@ -69,7 +70,7 @@ namespace hazelcast {
                                                          const int32_t &operationType,
                                                          std::auto_ptr<std::string> value);
 
-                std::vector<Address> getSocketAddresses();
+                std::set<Address, addressComparator> getSocketAddresses();
 
                 util::CountDownLatch startLatch;
 

hazelcast/include/hazelcast/client/connection/Connection.h

@@ -85,6 +85,10 @@ namespace hazelcast {
 
                 virtual void handleMessage(connection::Connection &connection, std::auto_ptr<protocol::ClientMessage> message);
 
+                int getConnectionId() const;
+
+                void setConnectionId(int connectionId);
+
                 util::AtomicInt lastRead;
                 util::AtomicBoolean live;
             private:
@@ -101,6 +105,8 @@ namespace hazelcast {
                 protocol::ClientMessageBuilder messageBuilder;
                 protocol::ClientMessage wrapperMessage;
                 std::auto_ptr<protocol::ClientMessage> responseMessage;
+
+                int connectionId;
             };
 
         }

hazelcast/include/hazelcast/client/connection/ConnectionManager.h

@@ -96,18 +96,35 @@ namespace hazelcast {
                 /**
                 * Tries to connect to an address in member list.
                 *
+                * @param tryCount The number of times it shall try during connection establishment if not connected
                 * @return authenticated connection
                 * @throws Exception authentication failed or no connection found
                 */
                 boost::shared_ptr<Connection> getRandomConnection(int tryCount);
 
+                /**
+                * Tries to connect to an address in member list.
+                *
+                * This check is to guarantee that the same server is not retried. This may happen since the
+                * cluster member update and the partition update may be received a little later and/or the load
+                * balancer may produce the same address.
+                *
+                * @param tryCount The number of times it shall try during connection establishment if not connected
+                * @param retryWaitTime The number of seconds to wait if the found random address is the same as the last
+                *              tried address comes out to be the same as the randomly selected server address.
+                * @return authenticated connection
+                * @throws Exception authentication failed or no connection found
+                */
+                boost::shared_ptr<Connection> getRandomConnection(int tryCount, const std::string &lastTriedAddress,
+                                                                  int retryWaitTime);
+
                 /**
                 * Called when an connection is closed.
                 * Clears related resources of given clientConnection.
                 *
                 * @param clientConnection closed connection
                 */
-                void onConnectionClose(const Address &address);
+                void onConnectionClose(const Address &address, int socketId);
 
                 /**
                 * Shutdown clientConnectionManager
@@ -168,6 +185,8 @@ namespace hazelcast {
                                                            std::auto_ptr<std::string> uuid,
                                                            std::auto_ptr<std::string> ownerUuid);
 
+                boost::shared_ptr<Connection> getOwnerConnection();
+
                 std::vector<byte> PROTOCOL;
                 util::SynchronizedMap<Address, Connection, addressComparator> connections;
                 util::SynchronizedMap<int, Connection> socketConnections;
@@ -188,6 +207,7 @@ namespace hazelcast {
                 OwnerConnectionFuture ownerConnectionFuture;
 
                 util::Atomic<int64_t> callIdGenerator;
+                util::Atomic<int> connectionIdCounter;
             };
         }
     }

hazelcast/include/hazelcast/client/connection/IOSelector.h

@@ -60,6 +60,8 @@ namespace hazelcast {
 
                 void addTask(ListenerTask *listenerTask);
 
+                void cancelTask(ListenerTask *listenerTask);
+
                 void wakeUp();
 
                 void shutdown();

hazelcast/include/hazelcast/client/exception/pimpl/ExceptionHandler.h

@@ -44,6 +44,8 @@ namespace hazelcast {
                     */
 
                     static std::string INSTANCE_NOT_ACTIVE;
+
+                    static std::string ILLEGAL_STATE;
                     /**
                     * InternalAPI rethrows the exception with appropriate type
                     *

hazelcast/include/hazelcast/client/protocol/ClientMessageBuilder.h

@@ -47,7 +47,7 @@ namespace hazelcast {
             class ClientMessageBuilder {
 
             public:
-                ClientMessageBuilder(IMessageHandler *service, connection::Connection &connection);
+                ClientMessageBuilder(IMessageHandler &service, connection::Connection &connection);
 
                 virtual ~ClientMessageBuilder();
 
@@ -77,7 +77,7 @@ namespace hazelcast {
 
                 std::auto_ptr<ClientMessage> message;
 
-                IMessageHandler *messageHandler;
+                IMessageHandler &messageHandler;
                 connection::Connection &connection;
 
                 int32_t frameLen;

hazelcast/include/hazelcast/client/proxy/IQueueImpl.h

@@ -46,7 +46,7 @@ namespace hazelcast {
 
                 bool contains(const serialization::pimpl::Data& element);
 
-                std::vector<serialization::pimpl::Data>  drainTo(int maxElements);
+                std::vector<serialization::pimpl::Data>  drainTo(size_t maxElements);
 
                 std::auto_ptr<serialization::pimpl::Data> peek();
 

hazelcast/include/hazelcast/client/proxy/TransactionalObject.h

@@ -88,9 +88,9 @@ namespace hazelcast {
 
                 template<typename K>
                 std::vector<K> toObjectCollection(const std::vector<serialization::pimpl::Data> &keyDataSet) {
-                    int size = keyDataSet.size();
+                    size_t size = keyDataSet.size();
                     std::vector<K> keys(size);
-                    for (int i = 0; i < size; i++) {
+                    for (size_t i = 0; i < size; i++) {
                         boost::shared_ptr<K> v = toObject<K>(keyDataSet[i]);
                         keys[i] = *v;
                     }

hazelcast/include/hazelcast/client/spi/ClusterService.h

@@ -105,7 +105,7 @@ namespace hazelcast {
 
                 void setMembers(std::auto_ptr<std::map<Address, Member, addressComparator> > map);
 
-                boost::shared_ptr<connection::Connection> connectToOne();
+                boost::shared_ptr<connection::Connection> connectToOne(const Address *previousConnectionAddr);
                 // ------------------------------------------------------
 
             };