cp.h

/*
 * Copyright (c) 2008-2025, 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 client
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;
}
} // namespace internal
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 HAZELCAST_API 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 cp
} // namespace hazelcast
 
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 std
 
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;
};
} // namespace boost
 
#if defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif