cp.h

/*
 * Copyright (c) 2008-2021, 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.
 */
 
#pragma once
 
#include <string>
#include <condition_variable>
#include <boost/thread/future.hpp>
 
#include "hazelcast/util/SynchronizedMap.h"
#include "hazelcast/client/proxy/ProxyImpl.h"
#include "hazelcast/client/exception/protocol_exceptions.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 spi {
            class ClientContext;
        }
    }
    namespace cp {
        class raft_proxy_factory;
 
        struct HAZELCAST_API raft_group_id {
            std::string name;
            int64_t seed;
            int64_t group_id;
 
            bool operator==(const raft_group_id &rhs) const;
 
            bool operator!=(const raft_group_id &rhs) const;
        };
 
        class HAZELCAST_API cp_proxy : public client::proxy::ProxyImpl {
        public:
            cp_proxy(const std::string &service_name, const std::string &proxy_name, client::spi::ClientContext *context,
                     const raft_group_id &group_id, const std::string &object_name);
 
            const raft_group_id &get_group_id() const;
 
        protected:
            raft_group_id group_id_;
            std::string object_name_;
 
            void on_destroy() ;
        };
 
        namespace internal {
            namespace session {
                class proxy_session_manager;
            }
        }
        class HAZELCAST_API session_aware_proxy : public cp_proxy {
        public:
            session_aware_proxy(const std::string &service_name, const std::string &proxy_name,
                                client::spi::ClientContext *context, const raft_group_id &group_id,
                                const std::string &object_name,
                                internal::session::proxy_session_manager &session_manager);
 
        protected:
            internal::session::proxy_session_manager &session_manager_;
 
            void release_session(int64_t session_id);
        };
 
        class HAZELCAST_API atomic_long : public cp_proxy {
        public:
            atomic_long(const std::string &name, client::spi::ClientContext &context,
                        const raft_group_id &group_id, const std::string &object_name);
 
            boost::future<int64_t> add_and_get(int64_t delta);
 
            boost::future<bool> compare_and_set(int64_t expect, int64_t update);
 
            boost::future<int64_t> get_and_decrement();
 
            boost::future<int64_t> decrement_and_get();
 
            boost::future<int64_t> get();
 
            boost::future<int64_t> get_and_add(int64_t delta);
 
            boost::future<int64_t> get_and_set(int64_t new_value);
 
            boost::future<int64_t> increment_and_get();
 
            boost::future<int64_t> get_and_increment();
 
            boost::future<void> set(int64_t new_value);
 
            template<typename F>
            boost::future<void> alter(const F &function) {
                return to_void_future(alter_and_get(function));
            }
 
            template<typename F>
            boost::future<int64_t> alter_and_get(const F &function) {
                auto f = to_data(function);
                return alter_data(f, alter_result_type::NEW_VALUE);
            }
 
            template<typename F>
            boost::future<int64_t> get_and_alter(const F &function) {
                auto f = to_data(function);
                return alter_data(f, alter_result_type::OLD_VALUE);
            }
 
            template<typename F, typename R>
            boost::future<boost::optional<R>> apply(const F &function) {
                auto f = to_data(function);
                return to_object<R>(apply_data(f));
            }
 
        private:
 
            static constexpr const char *SERVICE_NAME = "hz:raft:atomicLongService";
            enum alter_result_type {
                OLD_VALUE,
                NEW_VALUE
            };
 
            boost::future<int64_t>
            alter_data(client::serialization::pimpl::data &function_data, alter_result_type result_type);
 
            boost::future<boost::optional<client::serialization::pimpl::data>>
            apply_data(client::serialization::pimpl::data &function_data);
        };
 
        class HAZELCAST_API atomic_reference : public cp_proxy {
        public:
            atomic_reference(const std::string &name, client::spi::ClientContext &context,
                             const raft_group_id &group_id, const std::string &object_name);
 
            template<typename T>
            boost::future<boost::optional<typename std::remove_pointer<T>::type>> get() {
                return to_object<typename std::remove_pointer<T>::type>(get_data());
            }
 
            template<typename T>
            boost::future<boost::optional<typename std::remove_pointer<T>::type>> set(T new_value) {
                return to_object<typename std::remove_pointer<T>::type>(
                        set_data(to_data<typename std::remove_pointer<T>::type>(new_value)));
            }
 
            template<typename T>
            boost::future<boost::optional<typename std::remove_pointer<T>::type>> get_and_set(T new_value) {
                return to_object<typename std::remove_pointer<T>::type>(
                        get_and_set_data(to_data<typename std::remove_pointer<T>::type>(new_value)));
            }
 
            template<typename T, typename V>
            boost::future<bool> compare_and_set(T expect, V update) {
                return compare_and_set_data(to_data<typename std::remove_pointer<T>::type>(expect),
                                            to_data<typename std::remove_pointer<V>::type>(update));
            }
 
            boost::future<bool> is_null();
 
            boost::future<void> clear();
 
            template<typename T>
            boost::future<bool> contains(T value) {
                return contains_data(to_data<typename std::remove_pointer<T>::type>(value));
            }
 
            template<typename F>
            boost::future<void> alter(const F &function) {
                return alter_data(to_data(function));
            }
 
            template<typename T, typename F>
            boost::future<boost::optional<typename std::remove_pointer<T>::type>> alter_and_get(const F &function) {
                return to_object<typename std::remove_pointer<T>::type>(alter_and_get_data(to_data(function)));
            }
 
            template<typename T, typename F>
            boost::future<boost::optional<typename std::remove_pointer<T>::type>> get_and_alter(const F &function) {
                return to_object<typename std::remove_pointer<T>::type>(get_and_alter_data(to_data(function)));
            }
 
            template<typename R, typename F>
            boost::future<boost::optional<R>> apply(const F &function) {
                return to_object<R>(apply_data(to_data(function)));
            }
 
        private:
            static constexpr const char *SERVICE_NAME = "hz:raft:atomicRefService";
 
            enum struct return_value_type {
                NO_VALUE,
                OLD,
                NEW
            };
 
            boost::future<boost::optional<client::serialization::pimpl::data>> get_data();
 
            boost::future<boost::optional<client::serialization::pimpl::data>>
            set_data(const client::serialization::pimpl::data &new_value_data);
 
            boost::future<boost::optional<client::serialization::pimpl::data>>
            get_and_set_data(const client::serialization::pimpl::data &new_value_data);
 
            boost::future<bool> compare_and_set_data(const client::serialization::pimpl::data &expect_data,
                                                     const client::serialization::pimpl::data &update_data);
 
            boost::future<bool> contains_data(const client::serialization::pimpl::data &value_data);
 
            boost::future<void> alter_data(const client::serialization::pimpl::data &function_data);
 
            boost::future<boost::optional<client::serialization::pimpl::data>>
            alter_and_get_data(const client::serialization::pimpl::data &function_data);
 
            boost::future<boost::optional<client::serialization::pimpl::data>>
            get_and_alter_data(const client::serialization::pimpl::data &function_data);
 
            boost::future<boost::optional<client::serialization::pimpl::data>>
            apply_data(const client::serialization::pimpl::data &function_data);
 
            boost::future<boost::optional<client::serialization::pimpl::data>>
            invoke_apply(const client::serialization::pimpl::data function_data, return_value_type return_type,
                         bool alter);
        };
 
        class HAZELCAST_API latch : public cp_proxy {
        public:
            latch(const std::string &name, client::spi::ClientContext &context, const raft_group_id &group_id,
                  const std::string &object_name);
 
            boost::future<bool> try_set_count(int32_t count);
 
            boost::future<int32_t> get_count();
 
            boost::future<void> count_down();
 
            boost::future<bool> try_wait();
 
            boost::future<void> wait();
 
            boost::future<std::cv_status> wait_for(std::chrono::milliseconds timeout);
 
            template<typename Clock, typename Duration>
            boost::future<std::cv_status> wait_until(const std::chrono::time_point<Clock, Duration> &timeout_time) {
                return wait_for(std::chrono::duration_cast<std::chrono::milliseconds>(timeout_time - Clock::now()));
            }
 
        private:
            static constexpr const char *SERVICE_NAME = "hz:raft:countDownLatchService";
 
            boost::future<int32_t> get_round();
 
            void count_down(int round, boost::uuids::uuid invocation_uid);
 
        };
 
        class HAZELCAST_API fenced_lock : public session_aware_proxy {
        public:
            static constexpr int64_t INVALID_FENCE = 0L;
 
            fenced_lock(const std::string &name, client::spi::ClientContext &context,
                             const raft_group_id &group_id, const std::string &object_name);
 
            boost::future<void> lock();
 
            boost::future<int64_t> lock_and_get_fence();
 
            boost::future<bool> try_lock();
 
            boost::future<bool> try_lock(std::chrono::milliseconds timeout);
 
            boost::future<int64_t> try_lock_and_get_fence();
 
            boost::future<int64_t> try_lock_and_get_fence(std::chrono::milliseconds timeout);
 
            boost::future<void> unlock();
 
            boost::future<int64_t> get_fence();
 
            boost::future<bool> is_locked();
 
            boost::future<bool> is_locked_by_current_thread();
 
            boost::future<int32_t> get_lock_count();
 
            const raft_group_id &get_group_id();
 
            friend bool operator==(const fenced_lock &lhs, const fenced_lock &rhs);
 
        protected:
            void post_destroy() ;
 
        private:
            struct lock_ownership_state {
                int64_t fence;
                int32_t lock_count;
                int64_t session_id;
                int64_t thread_id;
 
                bool is_locked_by(int64_t session, int64_t thread);
 
                bool is_locked();
            };
 
            static constexpr const char *SERVICE_NAME = "hz:raft:lockService";
 
            // thread id -> id of the session that has acquired the lock
            util::SynchronizedMap<int64_t, int64_t> locked_session_ids_;
 
            void
            verify_locked_session_id_if_present(int64_t thread_id, int64_t session_id, bool should_release);
 
            void throw_lock_ownership_lost(int64_t session_id) const;
 
            void throw_illegal_monitor_state() const;
 
            boost::future<int64_t>
            do_lock(int64_t session_id, int64_t thread_id, boost::uuids::uuid invocation_uid);
 
            boost::future<int64_t>
            do_try_lock(int64_t session_id, int64_t thread_id, boost::uuids::uuid invocation_uid,
                        std::chrono::milliseconds timeout);
 
            boost::future<bool>
            do_unlock(int64_t session_id, int64_t thread_id, boost::uuids::uuid invocation_uid);
 
            boost::future<lock_ownership_state> do_get_lock_ownership_state();
 
            void invalidate_session(int64_t session_id);
 
            void verify_no_locked_session_id_present(int64_t thread_id);
        };
 
        class HAZELCAST_API counting_semaphore : public session_aware_proxy {
        public:
            //---- std::counting_semaphore method impl starts ---------
            virtual boost::future<void> acquire(int32_t permits = 1) = 0;
 
            virtual boost::future<void> release(int32_t permits = 1 ) = 0;
 
            boost::future<bool> try_acquire(int32_t permits = 1);
 
            boost::future<bool> try_acquire_for(std::chrono::milliseconds rel_time, int32_t permits = 1);
 
            template<class Clock, class Duration>
            boost::future<bool> try_acquire_until(const std::chrono::time_point<Clock, Duration>& abs_time, int32_t permits = 1) {
                auto now = Clock::now();
                return try_acquire_for(std::chrono::duration_cast<std::chrono::milliseconds>(abs_time - now), permits);
            }
 
            //---- std::counting_semaphore method impl ends ----------
 
            // --- extended methods
            boost::future<bool> init(int32_t permits);
 
            boost::future<int32_t> available_permits();
 
            virtual boost::future<int32_t> drain_permits() = 0;
 
            boost::future<void> reduce_permits(int32_t reduction);
 
            boost::future<void> increase_permits(int32_t increase);
 
        protected:
            static constexpr const char *SERVICE_NAME = "hz:raft:semaphoreService";
 
            counting_semaphore(const std::string &proxy_name, client::spi::ClientContext *context,
                               const raft_group_id &group_id, const std::string &object_name,
                               internal::session::proxy_session_manager &session_manager);
 
            virtual boost::future<bool> try_acquire_for_millis(int32_t permits, std::chrono::milliseconds timeout) = 0;
 
            virtual int64_t get_thread_id() = 0;
 
            virtual boost::future<void> do_change_permits(int32_t delta) = 0;
 
            boost::future<void> do_release(int32_t permits, int64_t thread_id, int64_t session_id);
        };
 
        class HAZELCAST_API sessionless_semaphore : public counting_semaphore {
        public:
            sessionless_semaphore(const std::string &proxy_name, client::spi::ClientContext *context,
                                  const raft_group_id &group_id, const std::string &object_name,
                                  internal::session::proxy_session_manager &session_manager);
 
            boost::future<void> acquire(int32_t permits) override;
 
            boost::future<void> release(int32_t permits) override;
 
            boost::future<int32_t> drain_permits() override;
 
        protected:
            boost::future<bool> try_acquire_for_millis(int32_t permits, std::chrono::milliseconds timeout) override;
 
            int64_t get_thread_id() override;
 
            boost::future<void> do_change_permits(int32_t delta) override;
 
        private:
            boost::future<bool> do_try_acquire(int32_t permits, std::chrono::milliseconds timeout_ms);
        };
 
        class HAZELCAST_API session_semaphore : public counting_semaphore {
        public:
            session_semaphore(const std::string &proxy_name, client::spi::ClientContext *context,
                                  const raft_group_id &group_id, const std::string &object_name,
                                  internal::session::proxy_session_manager &session_manager);
 
            boost::future<void> acquire(int32_t permits) override;
 
            boost::future<void> release(int32_t permits) override;
 
            boost::future<int32_t> drain_permits() override;
 
        protected:
            static constexpr int32_t DRAIN_SESSION_ACQ_COUNT = 1024;
 
            boost::future<bool> try_acquire_for_millis(int32_t permits, std::chrono::milliseconds timeout) override;
 
            void throw_illegal_state_exception(std::exception_ptr e);
 
            int64_t get_thread_id() override;
 
            boost::future<void> do_change_permits(int32_t delta) override;
        };
 
        class HAZELCAST_API raft_proxy_factory {
        public:
            raft_proxy_factory(client::spi::ClientContext &context);
 
            static std::string without_default_group_name(const std::string &n);
 
            static std::string object_name_for_proxy(const std::string &name);
 
            template<typename T>
            boost::future<std::shared_ptr<T>> create_proxy(const std::string &name) {
                auto proxy_name = without_default_group_name(name);
                auto object_name = object_name_for_proxy(proxy_name);
                return get_group_id(proxy_name, object_name).then([=] (boost::future<raft_group_id> f) {
                    auto group_id = f.get();
                    return create<T>(std::move(group_id), proxy_name, object_name);
                });
            }
 
        private:
            client::spi::ClientContext &context_;
            util::SynchronizedMap<std::string, fenced_lock> lock_proxies_;
 
            static constexpr const char *DEFAULT_GROUP_NAME = "default";
 
            template<typename T, typename = typename std::enable_if<std::is_same<atomic_long, T>::value ||
                                                                    std::is_same<atomic_reference, T>::value ||
                                                                    std::is_same<latch, T>::value>::type>
            std::shared_ptr<T>
            create(raft_group_id &&group_id, const std::string &proxy_name, const std::string &object_name) {
                return std::make_shared<T>(proxy_name, context_, std::move(group_id), object_name);
            }
 
            std::shared_ptr<fenced_lock>
            create_fenced_lock(raft_group_id &&group_id, const std::string &proxy_name, const std::string &object_name);
 
            template<typename T, typename = typename std::enable_if<std::is_same<fenced_lock, T>::value>::type>
            std::shared_ptr<fenced_lock>
            create(raft_group_id &&group_id, const std::string &proxy_name, const std::string &object_name) {
                return create_fenced_lock(std::move(group_id), proxy_name, object_name);
            }
 
            std::shared_ptr<counting_semaphore>
            create_semaphore(raft_group_id &&group_id, const std::string &proxy_name, const std::string &object_name);
 
            template<typename T, typename = typename std::enable_if<std::is_same<counting_semaphore, T>::value>::type>
            std::shared_ptr<counting_semaphore>
            create(raft_group_id &&group_id, const std::string &proxy_name, const std::string &object_name) {
                return create_semaphore(std::move(group_id), proxy_name, object_name);
            }
 
            boost::future<raft_group_id> get_group_id(const std::string &proxy_name, const std::string &object_name);
        };
 
        class HAZELCAST_API cp_subsystem {
        public:
            boost::future<std::shared_ptr<atomic_long>> get_atomic_long(const std::string &name);
 
            boost::future<std::shared_ptr<atomic_reference>> get_atomic_reference(const std::string &name);
 
            boost::future<std::shared_ptr<latch>> get_latch(const std::string &name);
 
            boost::future<std::shared_ptr<fenced_lock>> get_lock(const std::string &name);
 
            boost::future<std::shared_ptr<counting_semaphore>> get_semaphore(const std::string &name);
 
        private:
            friend client::impl::hazelcast_client_instance_impl;
            client::spi::ClientContext &context_;
            raft_proxy_factory proxy_factory_;
 
            cp_subsystem(client::spi::ClientContext &context);
        };
    }
}
 
namespace std {
    template<>
    struct HAZELCAST_API hash<hazelcast::cp::raft_group_id> {
        std::size_t operator()(const hazelcast::cp::raft_group_id &group_id) const noexcept;
    };
}
 
namespace boost {
    template<>
    struct HAZELCAST_API hash<hazelcast::cp::raft_group_id> {
        std::size_t operator()(const hazelcast::cp::raft_group_id &group_id) const noexcept;
    };
}
 
#if  defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif