hazelcast::cp::session_semaphore Class Reference

Inheritance diagram for hazelcast::cp::session_semaphore:

Public Member Functions
	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 Member Functions
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

Static Protected Attributes
static constexpr int32_t	DRAIN_SESSION_ACQ_COUNT = 1024
	Since a proxy does not know how many permits will be drained on the Raft group, it uses this constant to increment its local session acquire count.

Detailed Description

Definition at line 1278 of file cp.h.

Constructor & Destructor Documentation

session_semaphore()

hazelcast::cp::session_semaphore::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 )

Definition at line 1172 of file cp.cpp.

  : counting_semaphore(proxy_name,
                       context,
                       group_id,
                       object_name,
                       session_manager)
{}

Member Function Documentation

acquire()

boost::future< void > hazelcast::cp::session_semaphore::acquire ( int32_t permits )

override

Definition at line 1186 of file cp.cpp.

{
    return to_void_future(
      try_acquire_for_millis(permits, std::chrono::milliseconds(-1)));
}

do_change_permits()

boost::future< void > hazelcast::cp::session_semaphore::do_change_permits ( int32_t delta )

overrideprotected

Definition at line 1355 of file cp.cpp.

{
    auto session_id =
      session_manager_.acquire_session(group_id_, DRAIN_SESSION_ACQ_COUNT);
    auto thread_id = get_thread_id();
    auto invocation_uid =
      get_context().get_hazelcast_client_implementation()->random_uuid();
 
    auto request = client::protocol::codec::semaphore_change_encode(
      group_id_, object_name_, session_id, thread_id, invocation_uid, delta);
    return client::spi::impl::ClientInvocation::create(
             context_, request, object_name_)
      ->invoke()
      .then(boost::launch::sync,
            [=](boost::future<client::protocol::ClientMessage> f) {
                try {
                    f.get();
                    session_manager_.release_session(group_id_, session_id);
                } catch (client::exception::session_expired&) {
                    session_manager_.invalidate_session(group_id_, session_id);
                    throw_illegal_state_exception(std::current_exception());
                }
            });
}

drain_permits()

boost::future< int32_t > hazelcast::cp::session_semaphore::drain_permits ( )

override

Definition at line 1313 of file cp.cpp.

{
    auto thread_id = get_thread_id();
    auto invocation_uid =
      get_context().get_hazelcast_client_implementation()->random_uuid();
 
    auto do_drain_once = ([=]() {
        auto session_id =
          session_manager_.acquire_session(group_id_, DRAIN_SESSION_ACQ_COUNT);
        auto request = client::protocol::codec::semaphore_drain_encode(
          group_id_, object_name_, session_id, thread_id, invocation_uid);
        return client::spi::impl::ClientInvocation::create(
                 context_, request, object_name_)
          ->invoke()
          .then(
            boost::launch::sync,
            [=](boost::future<client::protocol::ClientMessage> f) {
                try {
                    auto count = f.get().get_first_fixed_sized_field<int32_t>();
                    session_manager_.release_session(
                      group_id_, session_id, DRAIN_SESSION_ACQ_COUNT - count);
                    return count;
                } catch (client::exception::session_expired&) {
                    session_manager_.invalidate_session(group_id_, session_id);
                    return -1;
                }
            });
    });
 
    return do_drain_once().then(
      boost::launch::sync, [=](boost::future<int32_t> f) {
          int32_t count = f.get();
          if (count != -1) {
              return count;
          }
          while ((count = do_drain_once().get()) == -1) {
          }
          return count;
      });
}

get_thread_id()

int64_t hazelcast::cp::session_semaphore::get_thread_id ( )

overrideprotected

Definition at line 1307 of file cp.cpp.

{
    return util::get_current_thread_id();
}

release()

boost::future< void > hazelcast::cp::session_semaphore::release ( int32_t permits )

override

Definition at line 1269 of file cp.cpp.

{
    util::Preconditions::check_positive(permits, "Permits must be positive!");
    auto session_id = session_manager_.get_session(group_id_);
    if (session_id == internal::session::proxy_session_manager::NO_SESSION_ID) {
        throw_illegal_state_exception(nullptr);
    }
 
    auto thread_id = get_thread_id();
    return do_release(permits, thread_id, session_id)
      .then([=](boost::future<void> f) {
          try {
              f.get();
              session_manager_.release_session(group_id_, session_id, permits);
          } catch (client::exception::session_expired&) {
              session_manager_.invalidate_session(group_id_, session_id);
              session_manager_.release_session(group_id_, session_id, permits);
              throw_illegal_state_exception(std::current_exception());
          }
      });
}

throw_illegal_state_exception()

void hazelcast::cp::session_semaphore::throw_illegal_state_exception ( std::exception_ptr e )

protected

Definition at line 1292 of file cp.cpp.

{
    auto ise = boost::enable_current_exception(client::exception::illegal_state(
      "session_semaphore::illegal_state", "No valid session!"));
    if (!e) {
        throw ise;
    }
    try {
        std::rethrow_exception(e);
    } catch (...) {
        std::throw_with_nested(ise);
    }
}

try_acquire_for_millis()

boost::future< bool > hazelcast::cp::session_semaphore::try_acquire_for_millis ( int32_t permits, std::chrono::milliseconds timeout )

overrideprotected

Definition at line 1193 of file cp.cpp.

{
    util::Preconditions::check_not_negative(
      permits, "permits must not be negative number.");
 
    auto thread_id = get_thread_id();
    auto invocation_uid =
      get_context().get_hazelcast_client_implementation()->random_uuid();
 
    auto do_try_acquire_once = ([=]() {
        auto start = std::chrono::steady_clock::now();
        auto use_timeout = timeout >= std::chrono::milliseconds::zero();
        auto session_id = session_manager_.acquire_session(group_id_, permits);
        auto request =
          client::protocol::codec::semaphore_acquire_encode(group_id_,
                                                            object_name_,
                                                            session_id,
                                                            thread_id,
                                                            invocation_uid,
                                                            permits,
                                                            timeout.count());
        return client::spi::impl::ClientInvocation::create(
                 context_, request, object_name_)
          ->invoke()
          .then(
            boost::launch::sync,
            [=](boost::future<client::protocol::ClientMessage> f) {
                try {
                    auto acquired = f.get().get_first_fixed_sized_field<bool>();
                    if (!acquired) {
                        session_manager_.release_session(group_id_, session_id);
                    }
                    // first bool means acquired or not, second bool means if
                    // should try again
                    return std::make_pair(acquired, false);
                } catch (client::exception::session_expired&) {
                    session_manager_.invalidate_session(group_id_, session_id);
                    if (use_timeout &&
                        (timeout - (std::chrono::steady_clock::now() - start) <=
                         std::chrono::milliseconds::zero())) {
                        return std::make_pair(false, false);
                    }
                    return std::make_pair(false, true);
                } catch (client::exception::wait_key_cancelled&) {
                    session_manager_.release_session(
                      group_id_, session_id, permits);
                    if (!use_timeout) {
                        BOOST_THROW_EXCEPTION(client::exception::illegal_state(
                          "session_semaphore::try_acquire_for_millis",
                          (boost::format(
                             "Semaphore[%1%] not acquired because the acquire "
                             "call on the CP group is cancelled, possibly "
                             "because of another indeterminate call from the "
                             "same thread.") %
                           object_name_)
                            .str()));
                    }
                    return std::make_pair(false, false);
                }
            });
    });
 
    return do_try_acquire_once().then(
      boost::launch::sync, [=](boost::future<std::pair<bool, bool>> f) {
          auto result = f.get();
          if (!result.second) {
              return result.first;
          }
          for (; result.second; result = do_try_acquire_once().get())
              ;
          return result.first;
      });
}

Member Data Documentation

DRAIN_SESSION_ACQ_COUNT

int32_t hazelcast::cp::session_semaphore::DRAIN_SESSION_ACQ_COUNT = 1024

staticconstexprprotected

Since a proxy does not know how many permits will be drained on the Raft group, it uses this constant to increment its local session acquire count.

Then, it adjusts the local session acquire count after the drain response is returned.

Definition at line 1301 of file cp.h.

---

The documentation for this class was generated from the following files:

• hazelcast/include/hazelcast/cp/cp.h
• hazelcast/src/hazelcast/cp/cp.cpp