IExecutorService.h

/*
 * Copyright (c) 2008-2019, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifndef HAZELCAST_CLIENT_IEXECUTORSERVICE_H_
#define HAZELCAST_CLIENT_IEXECUTORSERVICE_H_

#include <vector>

#include "hazelcast/util/HazelcastDll.h"
#include "hazelcast/client/cluster/memberselector/MemberSelectors.h"
#include "hazelcast/client/ICompletableFuture.h"
#include "hazelcast/client/proxy/ProxyImpl.h"
#include "hazelcast/client/ExecutionCallback.h"
#include "hazelcast/client/MultiExecutionCallback.h"
#include "hazelcast/client/Member.h"
#include "hazelcast/client/spi/ClientClusterService.h"
#include "hazelcast/client/spi/ClientContext.h"
#include "hazelcast/util/UuidUtil.h"
#include "hazelcast/client/spi/impl/ClientInvocation.h"
#include "hazelcast/util/ExceptionUtil.h"
#include "hazelcast/client/spi/impl/ClientExecutionServiceImpl.h"
#include "hazelcast/client/internal/executor/CompletedFuture.h"
#include "hazelcast/client/proxy/IExecutorDelegatingFuture.h"

// CODECs
#include "hazelcast/client/protocol/codec/ExecutorServiceSubmitToPartitionCodec.h"
#include "hazelcast/client/protocol/codec/ExecutorServiceSubmitToAddressCodec.h"

#if  defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(push)
#pragma warning(disable: 4251) //for dll export
#endif

namespace hazelcast {
    namespace client {
        namespace executor {
            namespace impl {
                class ExecutorServiceProxyFactory;
            }
        }

        class HAZELCAST_API IExecutorService : public proxy::ProxyImpl {
            friend class executor::impl::ExecutorServiceProxyFactory;

        public:
            static const std::string SERVICE_NAME;

            template<typename HazelcastSerializable>
            void execute(const HazelcastSerializable &command) {
                submit<HazelcastSerializable, bool>(command);
            }

            template<typename HazelcastSerializable>
            void execute(const HazelcastSerializable &command,
                         const cluster::memberselector::MemberSelector &memberSelector) {
                std::vector<Member> members = selectMembers(memberSelector);
                int selectedMember = rand() % (int) members.size();
                executeOnMember<HazelcastSerializable>(command, members[selectedMember]);
            }

            template<typename HazelcastSerializable, typename K>
            void executeOnKeyOwner(const HazelcastSerializable &command, const K &key) {
                submitToKeyOwner<HazelcastSerializable, K>(command, key);
            }

            template<typename HazelcastSerializable>
            void executeOnMember(const HazelcastSerializable &command, const Member &member) {
                submitToMember<HazelcastSerializable, bool>(command, member);
            }

            template<typename HazelcastSerializable>
            void executeOnMembers(const HazelcastSerializable &command, const std::vector<Member> &members) {
                for (std::vector<Member>::const_iterator it = members.begin(); it != members.end(); ++it) {
                    submitToMember<HazelcastSerializable, bool>(command, *it);
                }
            }

            template<typename HazelcastSerializable>
            void executeOnMembers(const HazelcastSerializable &command,
                                  const cluster::memberselector::MemberSelector &memberSelector) {
                std::vector<Member> members = selectMembers(memberSelector);
                executeOnMembers<HazelcastSerializable>(command, members);
            }

            template<typename HazelcastSerializable>
            void executeOnAllMembers(const HazelcastSerializable &command) {
                std::vector<Member> memberList = getContext().getClientClusterService().getMemberList();
                for (std::vector<Member>::const_iterator it = memberList.begin(); it != memberList.end(); ++it) {
                    submitToMember<HazelcastSerializable, bool>(command, *it);
                }
            }

            template<typename HazelcastSerializable, typename T, typename K>
            boost::shared_ptr<ICompletableFuture<T> >
            submitToKeyOwner(const HazelcastSerializable &task, const K &key) {
                return submitToKeyOwnerInternal<HazelcastSerializable, T, K>(task, key, boost::shared_ptr<T>(), false);
            }

            template<typename HazelcastSerializable, typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            submitToMember(const HazelcastSerializable &task, const Member &member) {
                const Address memberAddress = getMemberAddress(member);
                return submitToTargetInternal<HazelcastSerializable, T>(task, memberAddress, boost::shared_ptr<T>(),
                                                                        false);
            }

            template<typename HazelcastSerializable, typename T>
            std::map<Member, boost::shared_ptr<ICompletableFuture<T> > >
            submitToMembers(const HazelcastSerializable &task, const std::vector<Member> &members) {
                std::map<Member, boost::shared_ptr<ICompletableFuture<T> > > futureMap;
                for (std::vector<Member>::const_iterator it = members.begin(); it != members.end(); ++it) {
                    Address memberAddress = getMemberAddress(*it);
                    boost::shared_ptr<ICompletableFuture<T> > f = submitToTargetInternal<HazelcastSerializable, T>(task,
                                                                                                                   memberAddress,
                                                                                                                   boost::shared_ptr<T>(),
                                                                                                                   true);
                    futureMap[*it] = f;
                }
                return futureMap;
            }

            template<typename HazelcastSerializable, typename T>
            std::map<Member, boost::shared_ptr<ICompletableFuture<T> > >
            submitToMembers(const HazelcastSerializable &task,
                            const cluster::memberselector::MemberSelector &memberSelector) {
                std::vector<Member> members = selectMembers(memberSelector);
                return submitToMembers<HazelcastSerializable, T>(task, members);
            }

            template<typename HazelcastSerializable, typename T>
            std::map<Member, boost::shared_ptr<ICompletableFuture<T> > >
            submitToAllMembers(const HazelcastSerializable &task) {
                std::vector<Member> members = getContext().getClientClusterService().getMemberList();
                std::map<Member, boost::shared_ptr<ICompletableFuture<T> > > futureMap;
                for (std::vector<Member>::const_iterator it = members.begin(); it != members.end(); ++it) {
                    Address memberAddress = getMemberAddress(*it);
                    boost::shared_ptr<ICompletableFuture<T> > f = submitToTargetInternal<HazelcastSerializable, T>(task,
                                                                                                                   memberAddress,
                                                                                                                   boost::shared_ptr<T>(),
                                                                                                                   true);
                    futureMap[*it] = f;
                }
                return futureMap;
            }

            template<typename HazelcastSerializable, typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            submit(const HazelcastSerializable &task, const boost::shared_ptr<T> &result) {
                Data taskData = toData<HazelcastSerializable>(task);

                if (taskData.hasPartitionHash()) {
                    int partitionId = getPartitionId(taskData);

                    return submitToPartitionInternal<T>(taskData, result, false, partitionId);
                } else {
                    return submitToRandomInternal<T>(taskData, result, false);
                }
            }

            template<typename HazelcastSerializable, typename T>
            boost::shared_ptr<ICompletableFuture<T> > submit(const HazelcastSerializable &task) {
                return submit<HazelcastSerializable, T>(task, boost::shared_ptr<T>());
            }

            template<typename HazelcastSerializable, typename T>
            void submit(const HazelcastSerializable &task, const boost::shared_ptr<ExecutionCallback<T> > &callback) {
                Data taskData = toData<HazelcastSerializable>(task);

                if (taskData.hasPartitionHash()) {
                    int partitionId = getPartitionId(taskData);

                    submitToPartitionInternal<T>(taskData, partitionId, callback);
                } else {
                    submitToRandomInternal<T>(taskData, callback);
                }
            }

            template<typename HazelcastSerializable, typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            submit(const HazelcastSerializable &task, const cluster::memberselector::MemberSelector &memberSelector) {
                std::vector<Member> members = selectMembers(memberSelector);
                int selectedMember = rand() % (int) members.size();
                return submitToMember<HazelcastSerializable, T>(task, members[selectedMember]);
            }

            template<typename HazelcastSerializable, typename T>
            void
            submit(const HazelcastSerializable &task, const cluster::memberselector::MemberSelector &memberSelector,
                   const boost::shared_ptr<ExecutionCallback<T> > &callback) {
                std::vector<Member> members = selectMembers(memberSelector);
                int selectedMember = rand() % (int) members.size();
                return submitToMember(task, members[selectedMember], callback);
            }

            template<typename HazelcastSerializable, typename T, typename K>
            void submitToKeyOwner(const HazelcastSerializable &task, const K &key,
                                  const boost::shared_ptr<ExecutionCallback<T> > &callback) {
                submitToKeyOwnerInternal<HazelcastSerializable, T, K>(task, key, callback);
            }

            template<typename HazelcastSerializable, typename T>
            void submitToMember(const HazelcastSerializable &task, const Member &member,
                                const boost::shared_ptr<ExecutionCallback<T> > &callback) {
                const Address memberAddress = getMemberAddress(member);
                return submitToTargetInternal<HazelcastSerializable, T>(task, memberAddress, callback);
            }

            template<typename HazelcastSerializable, typename T>
            void submitToMembers(const HazelcastSerializable &task, const std::vector<Member> &members,
                                 const boost::shared_ptr<MultiExecutionCallback<T> > &callback) {
                boost::shared_ptr<MultiExecutionCallbackWrapper < T> >
                multiExecutionCallbackWrapper(new MultiExecutionCallbackWrapper<T>((int) members.size(), callback));

                for (std::vector<Member>::const_iterator it = members.begin(); it != members.end(); ++it) {
                    boost::shared_ptr<ExecutionCallbackWrapper < T> >
                    executionCallback(new ExecutionCallbackWrapper<T>(multiExecutionCallbackWrapper, *it));
                    submitToMember<HazelcastSerializable, T>(task, *it, executionCallback);
                }
            }

            template<typename HazelcastSerializable, typename T>
            void submitToMembers(const HazelcastSerializable &task,
                                 const cluster::memberselector::MemberSelector &memberSelector,
                                 const boost::shared_ptr<MultiExecutionCallback<T> > &callback) {
                std::vector<Member> members = selectMembers(memberSelector);
                submitToMembers<HazelcastSerializable, T>(task, members, callback);
            }

            template<typename HazelcastSerializable, typename T>
            void submitToAllMembers(const HazelcastSerializable &task,
                                    const boost::shared_ptr<MultiExecutionCallback<T> > &callback) {
                std::vector<Member> memberList = getContext().getClientClusterService().getMemberList();
                submitToMembers<HazelcastSerializable, T>(task, memberList, callback);
                boost::shared_ptr<MultiExecutionCallbackWrapper < T> >
                multiExecutionCallbackWrapper(new MultiExecutionCallbackWrapper<T>((int) memberList.size(), callback));
                for (std::vector<Member>::const_iterator it = memberList.begin(); it != memberList.end(); ++it) {
                    boost::shared_ptr<ExecutionCallbackWrapper < T> >
                    executionCallback(new ExecutionCallbackWrapper<T>(multiExecutionCallbackWrapper, *it));
                    submitToMember<HazelcastSerializable, T>(task, *it, executionCallback);
                }
            }

            void shutdown();

            bool isShutdown();

            bool isTerminated();

        private:
            IExecutorService(const std::string &name, spi::ClientContext *context);

            template<typename T>
            class MultiExecutionCallbackWrapper : MultiExecutionCallback<T> {
            public:
                MultiExecutionCallbackWrapper(
                        int memberSize, const boost::shared_ptr<MultiExecutionCallback<T> > &multiExecutionCallback)
                        : multiExecutionCallback(multiExecutionCallback), members(memberSize) {
                }

            public:
                virtual void onResponse(const Member &member, const boost::shared_ptr<T> &value) {
                    multiExecutionCallback->onResponse(member, value);
                    values.put(member, value);

                    int waitingResponse = --members;
                    if (waitingResponse == 0) {
                        complete();
                    }
                }

                virtual void
                onFailure(const Member &member, const boost::shared_ptr<exception::IException> &exception) {
                    multiExecutionCallback->onFailure(member, exception);
                    exceptions.put(member, exception);

                    int waitingResponse = --members;
                    if (waitingResponse == 0) {
                        complete();
                    }
                }

                virtual void onComplete(const std::map<Member, boost::shared_ptr<T> > &values,
                                        const std::map<Member, boost::shared_ptr<exception::IException> > &exceptions) {
                    multiExecutionCallback->onComplete(values, exceptions);
                }

            private:
                void complete() {
                    std::map<Member, boost::shared_ptr<T> > completedValues;
                    typedef std::vector<std::pair<Member, boost::shared_ptr<T> > > ENTRYVECTOR;
                    ENTRYVECTOR entries = values.entrySet();
                    for (typename ENTRYVECTOR::const_iterator it = entries.begin();
                         it != entries.end(); ++it) {
                        completedValues[it->first] = it->second;
                    }

                    std::map<Member, boost::shared_ptr<exception::IException> > completedExceptions;
                    typedef std::vector<std::pair<Member, boost::shared_ptr<exception::IException> > > EXCEPTIONVECTOR;
                    EXCEPTIONVECTOR exceptionEntries = exceptions.entrySet();
                    for (typename EXCEPTIONVECTOR::const_iterator it = exceptionEntries.begin();
                         it != exceptionEntries.end(); ++it) {
                        completedExceptions[it->first] = it->second;
                    }

                    onComplete(completedValues, completedExceptions);
                }

                const boost::shared_ptr<MultiExecutionCallback<T> > multiExecutionCallback;
                util::SynchronizedMap<Member, T> values;
                util::SynchronizedMap<Member, exception::IException> exceptions;
                util::AtomicInt members;
            };

            template<typename T>
            class ExecutionCallbackWrapper : public ExecutionCallback<T> {
            public:
                ExecutionCallbackWrapper(
                        const boost::shared_ptr<MultiExecutionCallbackWrapper<T> > &multiExecutionCallbackWrapper,
                        const Member &member) : multiExecutionCallbackWrapper(multiExecutionCallbackWrapper),
                                                member(member) {}

                virtual void onResponse(const boost::shared_ptr<T> &response) {
                    multiExecutionCallbackWrapper->onResponse(member, response);
                }

                virtual void onFailure(const boost::shared_ptr<exception::IException> &e) {
                    multiExecutionCallbackWrapper->onFailure(member, e);
                }

            private:
                const boost::shared_ptr<MultiExecutionCallbackWrapper<T> > multiExecutionCallbackWrapper;
                const Member member;
            };

            std::vector<Member> selectMembers(const cluster::memberselector::MemberSelector &memberSelector);

            template<typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            submitToPartitionInternal(const serialization::pimpl::Data &taskData,
                                      const boost::shared_ptr<T> &defaultValue,
                                      bool preventSync, int partitionId) {
                std::string uuid = util::UuidUtil::newUnsecureUuidString();

                boost::shared_ptr<spi::impl::ClientInvocationFuture> f = invokeOnPartitionInternal(taskData,
                                                                                                   partitionId, uuid);

                return checkSync(f, uuid, partitionId, preventSync, defaultValue);
            }

            template<typename T>
            void submitToPartitionInternal(const serialization::pimpl::Data &taskData, int partitionId,
                                           const boost::shared_ptr<ExecutionCallback<T> > &callback) {
                std::string uuid = util::UuidUtil::newUnsecureUuidString();

                boost::shared_ptr<spi::impl::ClientInvocationFuture> f = invokeOnPartitionInternal(taskData,
                                                                                                   partitionId, uuid);


                boost::shared_ptr<ICompletableFuture<T> > delegatingFuture(
                        new internal::ClientDelegatingFuture<T>(f, getContext().getSerializationService(),
                                                                SUBMIT_TO_PARTITION_DECODER<T>(),
                                                                boost::shared_ptr<T>()));

                delegatingFuture->andThen(callback);
            }

            boost::shared_ptr<spi::impl::ClientInvocationFuture>
            invokeOnPartitionInternal(const Data &taskData, int partitionId, const std::string &uuid) {
                std::auto_ptr<protocol::ClientMessage> request =
                        protocol::codec::ExecutorServiceSubmitToPartitionCodec::encodeRequest(name, uuid, taskData,
                                                                                              partitionId);

                return invokeOnPartitionOwner(request, partitionId);
            }

            template<typename HazelcastSerializable, typename T, typename K>
            boost::shared_ptr<ICompletableFuture<T> >
            submitToKeyOwnerInternal(const HazelcastSerializable &task, const K &key,
                                     const boost::shared_ptr<T> &defaultValue, bool preventSync) {

                Data dataKey = toData<K>(key);

                int partitionId = getPartitionId(dataKey);

                return submitToPartitionInternal<T>(toData<HazelcastSerializable>(task), defaultValue, preventSync,
                                                    partitionId);
            }

            template<typename HazelcastSerializable, typename T, typename K>
            void submitToKeyOwnerInternal(const HazelcastSerializable &task, const K &key,
                                          const boost::shared_ptr<ExecutionCallback<T> > &callback) {

                Data dataKey = toData<K>(key);

                int partitionId = getPartitionId(dataKey);

                submitToPartitionInternal<T>(toData<HazelcastSerializable>(task), partitionId, callback);
            }

            template<typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            submitToRandomInternal(const serialization::pimpl::Data &taskData, const boost::shared_ptr<T> &defaultValue,
                                   bool preventSync) {

                int partitionId = randomPartitionId();

                return submitToPartitionInternal<T>(taskData, defaultValue, preventSync, partitionId);
            }

            template<typename T>
            void submitToRandomInternal(const serialization::pimpl::Data &taskData,
                                        const boost::shared_ptr<ExecutionCallback<T> > &callback) {

                int partitionId = randomPartitionId();

                submitToPartitionInternal<T>(taskData, partitionId, callback);
            }

            template<typename HazelcastSerializable, typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            submitToTargetInternal(const HazelcastSerializable &task, const Address &address,
                                   const boost::shared_ptr<T> &defaultValue, bool preventSync) {
                std::string uuid = util::UuidUtil::newUnsecureUuidString();

                boost::shared_ptr<spi::impl::ClientInvocationFuture> f = invokeOnAddressInternal<HazelcastSerializable>(
                        task, address, uuid);

                return checkSync<T>(f, uuid, address, preventSync, defaultValue);
            }

            template<typename HazelcastSerializable, typename T>
            void submitToTargetInternal(const HazelcastSerializable &task, const Address &address,
                                        const boost::shared_ptr<ExecutionCallback<T> > &callback) {
                std::string uuid = util::UuidUtil::newUnsecureUuidString();

                boost::shared_ptr<spi::impl::ClientInvocationFuture> f = invokeOnAddressInternal<HazelcastSerializable>(
                        task, address, uuid);

                boost::shared_ptr<ICompletableFuture<T> > delegatingFuture(
                        new internal::ClientDelegatingFuture<T>(f, getContext().getSerializationService(),
                                                                SUBMIT_TO_ADDRESS_DECODER<T>(),
                                                                boost::shared_ptr<T>()));

                delegatingFuture->andThen(callback);
            }

            template<typename HazelcastSerializable>
            boost::shared_ptr<spi::impl::ClientInvocationFuture>
            invokeOnAddressInternal(const HazelcastSerializable &task, const Address &address,
                                    const std::string &uuid) {
                std::auto_ptr<protocol::ClientMessage> request =
                        protocol::codec::ExecutorServiceSubmitToAddressCodec::encodeRequest(name, uuid,
                                                                                            toData(
                                                                                                    task), address);

                boost::shared_ptr<spi::impl::ClientInvocationFuture> f = invokeOnTarget(request, address);
                return f;
            }

            boost::shared_ptr<spi::impl::ClientInvocationFuture>
            invokeOnPartitionOwner(std::auto_ptr<protocol::ClientMessage> &request, int partitionId);

            boost::shared_ptr<spi::impl::ClientInvocationFuture>
            invokeOnTarget(std::auto_ptr<protocol::ClientMessage> &request, const Address &target);

            template<typename T>
            boost::shared_ptr<T>
            retrieveResultFromMessage(const boost::shared_ptr<spi::impl::ClientInvocationFuture> &f) {
                serialization::pimpl::SerializationService &serializationService = getContext().getSerializationService();
                std::auto_ptr<serialization::pimpl::Data> data = protocol::codec::ExecutorServiceSubmitToAddressCodec::ResponseParameters::decode(
                        *f->get()).response;
                return boost::shared_ptr<T>(serializationService.toObject<T>(data.get()));
            }

            template<typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            checkSync(const boost::shared_ptr<spi::impl::ClientInvocationFuture> &f, const std::string &uuid,
                      int partitionId, bool preventSync, boost::shared_ptr<T> defaultValue) {
                bool sync = isSyncComputation(preventSync);
                if (sync) {
                    return retrieveResultSync<T>(f);
                } else {
                    return boost::shared_ptr<ICompletableFuture<T> >(
                            new proxy::IExecutorDelegatingFuture<T>(f, getContext(), uuid, defaultValue,
                                                                    SUBMIT_TO_PARTITION_DECODER<T>(), name,
                                                                    partitionId));
                }

            }

            template<typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            checkSync(const boost::shared_ptr<spi::impl::ClientInvocationFuture> &f, const std::string &uuid,
                      const Address &address, bool preventSync, boost::shared_ptr<T> defaultValue) {
                bool sync = isSyncComputation(preventSync);
                if (sync) {
                    return retrieveResultSync<T>(f);
                } else {
                    return boost::shared_ptr<ICompletableFuture<T> >(
                            new proxy::IExecutorDelegatingFuture<T>(f, getContext(), uuid, defaultValue,
                                                                    SUBMIT_TO_ADDRESS_DECODER<T>(), name, address));
                }

            }

            template<typename T>
            boost::shared_ptr<ICompletableFuture<T> >
            retrieveResultSync(const boost::shared_ptr<spi::impl::ClientInvocationFuture> &f) {
                try {
                    boost::shared_ptr<T> response = retrieveResultFromMessage<T>(f);
                    boost::shared_ptr<ExecutorService> userExecutor = getContext().getClientExecutionService().getUserExecutor();
                    return boost::shared_ptr<ICompletableFuture<T> >(
                            new internal::executor::CompletedFuture<T>(response, userExecutor));
                } catch (exception::IException &e) {
                    boost::shared_ptr<ExecutorService> userExecutor = getContext().getClientExecutionService().getUserExecutor();
                    return boost::shared_ptr<ICompletableFuture<T> >(
                            new internal::executor::CompletedFuture<T>(
                                    boost::shared_ptr<exception::IException>(e.clone()), userExecutor));
                }
            }

            bool isSyncComputation(bool preventSync);

            Address getMemberAddress(const Member &member);

            int randomPartitionId();

            template<typename T>
            static const boost::shared_ptr<impl::ClientMessageDecoder<T> > SUBMIT_TO_PARTITION_DECODER() {
                return impl::DataMessageDecoder<protocol::codec::ExecutorServiceSubmitToPartitionCodec, T>::instance();
            }

            template<typename T>
            static const boost::shared_ptr<impl::ClientMessageDecoder<T> > SUBMIT_TO_ADDRESS_DECODER() {
                return impl::DataMessageDecoder<protocol::codec::ExecutorServiceSubmitToAddressCodec, T>::instance();
            }

            static const int32_t MIN_TIME_RESOLUTION_OF_CONSECUTIVE_SUBMITS = 10;
            static const int32_t MAX_CONSECUTIVE_SUBMITS = 100;

            util::Atomic<int32_t> consecutiveSubmits;
            util::Atomic<int64_t> lastSubmitTime;
        };
    }
}

#if  defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif

#endif /* HAZELCAST_CLIENT_IEXECUTORSERVICE_H_ */