transactions.cpp

/*
 * Copyright (c) 2008-2022, 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.
 */
 
#include <boost/uuid/uuid_io.hpp>
 
#include <hazelcast/client/txn/client_transaction_util.h>
#include "hazelcast/client/txn/TransactionProxy.h"
#include "hazelcast/client/transaction_options.h"
#include "hazelcast/util/Util.h"
#include "hazelcast/client/spi/impl/ClientInvocation.h"
#include "hazelcast/client/connection/ClientConnectionManagerImpl.h"
#include "hazelcast/client/proxy/TransactionalMapImpl.h"
#include "hazelcast/client/proxy/TransactionalMultiMapImpl.h"
#include "hazelcast/client/proxy/TransactionalListImpl.h"
#include "hazelcast/client/proxy/TransactionalQueueImpl.h"
#include "hazelcast/client/proxy/TransactionalSetImpl.h"
#include "hazelcast/client/imap.h"
#include "hazelcast/client/multi_map.h"
#include "hazelcast/client/ilist.h"
#include "hazelcast/client/iqueue.h"
#include "hazelcast/client/iset.h"
#include "hazelcast/client/transaction_context.h"
#include "hazelcast/client/spi/impl/ClientTransactionManagerServiceImpl.h"
#include "hazelcast/client/protocol/codec/codecs.h"
 
namespace hazelcast {
    namespace client {
        namespace txn {
            TransactionProxy::TransactionProxy(transaction_options &txn_options, spi::ClientContext &client_context,
                                               std::shared_ptr<connection::Connection> connection)
                    : options_(txn_options), client_context_(client_context), connection_(connection),
                      thread_id_(util::get_current_thread_id()), state_(TxnState::NO_TXN) {}
 
            TransactionProxy::TransactionProxy(const TransactionProxy &rhs) : options_(rhs.options_),
                                                                              client_context_(rhs.client_context_),
                                                                              connection_(rhs.connection_),
                                                                              thread_id_(rhs.thread_id_), txn_id_(rhs.txn_id_),
                                                                              state_(rhs.state_),
                                                                              start_time_(rhs.start_time_) {
                transaction_exists_.store(rhs.transaction_exists_.load());
            }
 
            boost::uuids::uuid TransactionProxy::get_txn_id() const {
                return txn_id_;
            }
 
            TxnState TransactionProxy::get_state() const {
                return state_;
            }
 
            std::chrono::milliseconds TransactionProxy::get_timeout() const {
                return options_.get_timeout();
            }
 
            boost::future<void> TransactionProxy::begin() {
                try {
                    if (state_ == TxnState::ACTIVE) {
                        BOOST_THROW_EXCEPTION(exception::illegal_state("TransactionProxy::begin()",
                                                                               "Transaction is already active"));
                    }
                    check_thread();
                    if (transaction_exists_) {
                        BOOST_THROW_EXCEPTION(exception::illegal_state("TransactionProxy::begin()",
                                                                               "Nested transactions are not allowed!"));
                    }
                    transaction_exists_.store(true);
                    start_time_ = std::chrono::steady_clock::now();
                    auto request = protocol::codec::transaction_create_encode(
                            std::chrono::duration_cast<std::chrono::milliseconds>(get_timeout()).count(), options_.get_durability(),
                            static_cast<int32_t>(options_.get_transaction_type()), thread_id_);
                    return invoke(request).then(boost::launch::sync, [=](boost::future<protocol::ClientMessage> f) {
                        try {
                            auto msg = f.get();
                            // skip header
                            msg.rd_ptr(msg.RESPONSE_HEADER_LEN);
                            this->txn_id_ = msg.get<boost::uuids::uuid>();
                            this->state_ = TxnState::ACTIVE;
                        } catch (exception::iexception &) {
                            transaction_exists_.store(false);
                            throw;
                        }
                    });
                } catch (exception::iexception &) {
                    transaction_exists_.store(false);
                    throw;
                }
            }
 
            boost::future<void> TransactionProxy::commit() {
                try {
                    if (state_ != TxnState::ACTIVE) {
                        BOOST_THROW_EXCEPTION(exception::illegal_state("TransactionProxy::commit()",
                                                                               "Transaction is not active"));
                    }
                    state_ = TxnState::COMMITTING;
                    check_thread();
                    check_timeout();
 
                    auto request = protocol::codec::transaction_commit_encode(txn_id_, thread_id_);
                    return invoke(request).then(boost::launch::sync, [=](boost::future<protocol::ClientMessage> f) {
                        try {
                            f.get();
                            state_ = TxnState::COMMITTED;
                        } catch (exception::iexception &) {
                            transaction_exists_.store(false);
                            client_transaction_util::transaction_exception_factory()->rethrow(std::current_exception(),
                                                                                              "TransactionProxy::commit() failed");
                        }
                    });
                } catch (...) {
                    state_ = TxnState::COMMIT_FAILED;
                    transaction_exists_.store(false);
                    client_transaction_util::transaction_exception_factory()->rethrow(std::current_exception(),
                                                                                      "TransactionProxy::commit() failed");
                    return boost::make_ready_future();
                }
            }
 
            boost::future<void> TransactionProxy::rollback() {
                try {
                    if (state_ == TxnState::NO_TXN || state_ == TxnState::ROLLED_BACK) {
                        BOOST_THROW_EXCEPTION(exception::illegal_state("TransactionProxy::rollback()",
                                                                               "Transaction is not active"));
                    }
                    state_ = TxnState::ROLLING_BACK;
                    check_thread();
                    try {
                        auto request = protocol::codec::transaction_rollback_encode(txn_id_, thread_id_);
                        return invoke(request).then(boost::launch::sync,
                                                    [=](boost::future<protocol::ClientMessage> f) {
                                                        try {
                                                            state_ = TxnState::ROLLED_BACK;
                                                            transaction_exists_.store(false);
                                                            f.get();
                                                        } catch (exception::iexception &e) {
                                                            HZ_LOG(client_context_.get_logger(), warning,
                                                                   boost::str(boost::format(
                                                                           "Exception while rolling back the transaction. "
                                                                           "Exception: %1%")
                                                                              % e)
                                                            );
                            }
                        });
                    } catch (exception::iexception &exception) {
                        HZ_LOG(client_context_.get_logger(), warning,
                            boost::str(boost::format("Exception while rolling back the transaction. "
                                                     "Exception: %1%")
                                                     % exception)
                        );
                    }
                    state_ = TxnState::ROLLED_BACK;
                    transaction_exists_.store(false);
                } catch (exception::iexception &) {
                    transaction_exists_.store(false);
                    client_transaction_util::transaction_exception_factory()->rethrow(std::current_exception(),
                                                                                      "TransactionProxy::rollback() failed");
                }
                return boost::make_ready_future();
            }
 
            serialization::pimpl::SerializationService &TransactionProxy::get_serialization_service() {
                return client_context_.get_serialization_service();
            }
 
            std::shared_ptr<connection::Connection> TransactionProxy::get_connection() {
                return connection_;
            }
 
            void TransactionProxy::check_thread() {
                if (thread_id_ != util::get_current_thread_id()) {
                    BOOST_THROW_EXCEPTION(exception::illegal_state("TransactionProxy::checkThread()",
                                                                           "Transaction cannot span multiple threads!"));
                }
            }
 
            void TransactionProxy::check_timeout() {
                if (start_time_ + options_.get_timeout() < std::chrono::steady_clock::now()) {
                    BOOST_THROW_EXCEPTION(exception::transaction("TransactionProxy::checkTimeout()",
                                                                          "Transaction is timed-out!"));
                }
            }
 
            TxnState::TxnState(state value)
                    : value(value) {
                values.resize(9);
                values[0] = NO_TXN;
                values[1] = ACTIVE;
                values[2] = PREPARING;
                values[3] = PREPARED;
                values[4] = COMMITTING;
                values[5] = COMMITTED;
                values[6] = COMMIT_FAILED;
                values[7] = ROLLING_BACK;
                values[8] = ROLLED_BACK;
            }
 
            TxnState::operator int() const {
                return value;
            }
 
            void TxnState::operator=(int i) {
                value = values[i];
            }
 
            boost::future<protocol::ClientMessage> TransactionProxy::invoke(protocol::ClientMessage &request) {
                return client_transaction_util::invoke(request, boost::uuids::to_string(get_txn_id()), client_context_, connection_);
            }
 
            spi::ClientContext &TransactionProxy::get_client_context() const {
                return client_context_;
            }
 
            const std::shared_ptr<util::exception_util::runtime_exception_factory> client_transaction_util::exceptionFactory(
                    new class transaction_exception_factory());
 
            boost::future<protocol::ClientMessage>
            client_transaction_util::invoke(protocol::ClientMessage &request,
                                            const std::string &object_name,
                                            spi::ClientContext &client,
                                            const std::shared_ptr<connection::Connection> &connection) {
                try {
                    std::shared_ptr<spi::impl::ClientInvocation> clientInvocation = spi::impl::ClientInvocation::create(
                            client, request, object_name, connection);
                    return clientInvocation->invoke();
                } catch (exception::iexception &) {
                    transaction_exception_factory()->rethrow(std::current_exception(),
                                                             "ClientTransactionUtil::invoke failed");
                    return boost::make_ready_future(protocol::ClientMessage(0));
                }
            }
 
            const std::shared_ptr<util::exception_util::runtime_exception_factory> &
            client_transaction_util::transaction_exception_factory() {
                return exceptionFactory;
            }
 
            void
            client_transaction_util::transaction_exception_factory::rethrow(std::exception_ptr throwable,
                                                                            const std::string &message) {
                try {
                    std::rethrow_exception(throwable);
                } catch (...) {
                    std::throw_with_nested(
                            boost::enable_current_exception(
                                    exception::transaction("transaction_exceptionFactory::create", message)));
                }
            }
        }
 
        namespace proxy {
            TransactionalMapImpl::TransactionalMapImpl(const std::string &name, txn::TransactionProxy &transaction_proxy)
                    : TransactionalObject(imap::SERVICE_NAME, name, transaction_proxy) {}
 
            boost::future<bool> TransactionalMapImpl::contains_key_data(const serialization::pimpl::data &key) {
                auto request = protocol::codec::transactionalmap_containskey_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key);
 
                return invoke_and_get_future<bool>(request);
            }
 
            boost::future<boost::optional<serialization::pimpl::data>>
            TransactionalMapImpl::get_data(const serialization::pimpl::data &key) {
                auto request = protocol::codec::transactionalmap_get_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key);
 
                return invoke_and_get_future<boost::optional<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<int> TransactionalMapImpl::size() {
                auto request = protocol::codec::transactionalmap_size_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id());
 
                return invoke_and_get_future<int>(request);
            }
 
            boost::future<bool> TransactionalMapImpl::is_empty() {
                auto request = protocol::codec::transactionalmap_isempty_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id());
 
                return invoke_and_get_future<bool>(
                        request);
            }
 
            boost::future<boost::optional<serialization::pimpl::data>> TransactionalMapImpl::put_data(
                    const serialization::pimpl::data &key, const serialization::pimpl::data &value) {
 
                auto request = protocol::codec::transactionalmap_put_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key, value,
                                std::chrono::duration_cast<std::chrono::milliseconds>(get_timeout()).count());
 
                return invoke_and_get_future<boost::optional<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<void>
            TransactionalMapImpl::set_data(const serialization::pimpl::data &key, const serialization::pimpl::data &value) {
                auto request = protocol::codec::transactionalmap_set_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key, value);
 
                return to_void_future(invoke(request));
            }
 
            boost::future<boost::optional<serialization::pimpl::data>>
            TransactionalMapImpl::put_if_absent_data(const serialization::pimpl::data &key,
                                                  const serialization::pimpl::data &value) {
                auto request = protocol::codec::transactionalmap_putifabsent_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key, value);
 
                return invoke_and_get_future<boost::optional<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<boost::optional<serialization::pimpl::data>>
            TransactionalMapImpl::replace_data(const serialization::pimpl::data &key,
                                              const serialization::pimpl::data &value) {
                auto request = protocol::codec::transactionalmap_replace_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key, value);
 
                return invoke_and_get_future<boost::optional<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<bool> TransactionalMapImpl::replace_data(const serialization::pimpl::data &key,
                                               const serialization::pimpl::data &old_value,
                                               const serialization::pimpl::data &new_value) {
                auto request = protocol::codec::transactionalmap_replaceifsame_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key, old_value, new_value);
 
                return invoke_and_get_future<bool>(
                        request);
            }
 
            boost::future<boost::optional<serialization::pimpl::data>>
            TransactionalMapImpl::remove_data(const serialization::pimpl::data &key) {
                auto request = protocol::codec::transactionalmap_remove_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key);
 
                return invoke_and_get_future<boost::optional<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<void> TransactionalMapImpl::delete_entry_data(const serialization::pimpl::data &key) {
                auto request = protocol::codec::transactionalmap_delete_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key);
 
                return to_void_future(invoke(request));
            }
 
            boost::future<bool> TransactionalMapImpl::remove_data(const serialization::pimpl::data &key,
                                              const serialization::pimpl::data &value) {
                auto request = protocol::codec::transactionalmap_removeifsame_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key, value);
 
                return invoke_and_get_future<bool>(
                        request);
            }
 
            boost::future<std::vector<serialization::pimpl::data>> TransactionalMapImpl::key_set_data() {
                auto request = protocol::codec::transactionalmap_keyset_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id());
 
                return invoke_and_get_future<std::vector<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<std::vector<serialization::pimpl::data>>
            TransactionalMapImpl::key_set_data(const serialization::pimpl::data &predicate) {
                auto request = protocol::codec::transactionalmap_keysetwithpredicate_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), predicate);
 
                return invoke_and_get_future<std::vector<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<std::vector<serialization::pimpl::data>> TransactionalMapImpl::values_data() {
                auto request = protocol::codec::transactionalmap_values_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id());
 
                return invoke_and_get_future<std::vector<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<std::vector<serialization::pimpl::data>>
            TransactionalMapImpl::values_data(const serialization::pimpl::data &predicate) {
                auto request = protocol::codec::transactionalmap_valueswithpredicate_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), predicate);
 
                return invoke_and_get_future<std::vector<serialization::pimpl::data>>(
                        request);
            }
 
            TransactionalMultiMapImpl::TransactionalMultiMapImpl(const std::string &name,
                                                                 txn::TransactionProxy &transaction_proxy)
                    : TransactionalObject(multi_map::SERVICE_NAME, name, transaction_proxy) {}
 
            boost::future<bool> TransactionalMultiMapImpl::put_data(const serialization::pimpl::data &key,
                                                const serialization::pimpl::data &value) {
                auto request = protocol::codec::transactionalmultimap_put_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key, value);
 
                return invoke_and_get_future<bool>(
                        request);
            }
 
            boost::future<std::vector<serialization::pimpl::data>> TransactionalMultiMapImpl::get_data(
                    const serialization::pimpl::data &key) {
                auto request = protocol::codec::transactionalmultimap_get_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key);
 
                return invoke_and_get_future<std::vector<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<bool> TransactionalMultiMapImpl::remove(const serialization::pimpl::data &key,
                                                   const serialization::pimpl::data &value) {
                auto request = protocol::codec::transactionalmultimap_removeentry_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key, value);
 
                return invoke_and_get_future<bool>(
                        request);
            }
 
            boost::future<std::vector<serialization::pimpl::data>> TransactionalMultiMapImpl::remove_data(
                    const serialization::pimpl::data &key) {
                auto request = protocol::codec::transactionalmultimap_remove_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key);
 
                return invoke_and_get_future<std::vector<serialization::pimpl::data>>(
                        request);
            }
 
            boost::future<int> TransactionalMultiMapImpl::value_count(const serialization::pimpl::data &key) {
                auto request = protocol::codec::transactionalmultimap_valuecount_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), key);
 
                return invoke_and_get_future<int>(
                        request);
            }
 
            boost::future<int> TransactionalMultiMapImpl::size() {
                auto request = protocol::codec::transactionalmultimap_size_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id());
 
                return invoke_and_get_future<int>(
                        request);
            }
 
            TransactionalListImpl::TransactionalListImpl(const std::string &object_name, txn::TransactionProxy &context)
                    : TransactionalObject(ilist::SERVICE_NAME, object_name, context) {}
 
            boost::future<bool> TransactionalListImpl::add(const serialization::pimpl::data &e) {
                auto request = protocol::codec::transactionallist_add_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), e);
 
                return invoke_and_get_future<bool>(
                        request);
            }
 
            boost::future<bool> TransactionalListImpl::remove(const serialization::pimpl::data &e) {
                auto request = protocol::codec::transactionallist_remove_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), e);
 
                return invoke_and_get_future<bool>(
                        request);
            }
 
            boost::future<int> TransactionalListImpl::size() {
                auto request = protocol::codec::transactionallist_size_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id());
 
                return invoke_and_get_future<int>(
                        request);
            }
 
            TransactionalSetImpl::TransactionalSetImpl(const std::string &name, txn::TransactionProxy &transaction_proxy)
                    : TransactionalObject(iset::SERVICE_NAME, name, transaction_proxy) {}
 
            boost::future<bool> TransactionalSetImpl::add_data(const serialization::pimpl::data &e) {
                auto request = protocol::codec::transactionalset_add_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), e);
 
                return invoke_and_get_future<bool>(request);
            }
 
            boost::future<bool> TransactionalSetImpl::remove_data(const serialization::pimpl::data &e) {
                auto request = protocol::codec::transactionalset_remove_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id(), e);
 
                return invoke_and_get_future<bool>(
                        request);
            }
 
            boost::future<int> TransactionalSetImpl::size() {
                auto request = protocol::codec::transactionalset_size_encode(
                                get_name(), get_transaction_id(), util::get_current_thread_id());
 
                return invoke_and_get_future<int>(request);
            }
            
            TransactionalObject::TransactionalObject(const std::string &service_name, const std::string &object_name,
                                                     txn::TransactionProxy &context)
                    : proxy::SerializingProxy(context.get_client_context(), object_name), service_name_(service_name),
                      name_(object_name), context_(context) {}
 
            TransactionalObject::~TransactionalObject() = default;
 
            const std::string &TransactionalObject::get_service_name() {
                return service_name_;
            }
 
            const std::string &TransactionalObject::get_name() {
                return name_;
            }
 
            boost::future<void> TransactionalObject::destroy() {
                on_destroy();
                auto request = protocol::codec::client_destroyproxy_encode(name_, service_name_);
                return to_void_future(invoke_on_connection(request, context_.get_connection()));
            }
 
            void TransactionalObject::on_destroy() {}
 
            boost::uuids::uuid TransactionalObject::get_transaction_id() const {
                return context_.get_txn_id();
            }
 
            std::chrono::milliseconds TransactionalObject::get_timeout() const {
                return context_.get_timeout();
            }
        }
 
        transaction_context::transaction_context(spi::impl::ClientTransactionManagerServiceImpl &transaction_manager,
                                                 const transaction_options &txn_options) : options_(txn_options),
                                                                                           txn_connection_(
                                                                                               transaction_manager.connect()),
                                                                                           transaction_(options_,
                                                                                                   transaction_manager.get_client(),
                                                                                                   txn_connection_) {
        }
 
        boost::uuids::uuid  transaction_context::get_txn_id() const {
            return transaction_.get_txn_id();
        }
 
        boost::future<void> transaction_context::begin_transaction() {
            return transaction_.begin();
        }
 
        boost::future<void> transaction_context::commit_transaction() {
            return transaction_.commit();
        }
 
        boost::future<void> transaction_context::rollback_transaction() {
            return transaction_.rollback();
        }
 
        transaction_options::transaction_options() : timeout_(std::chrono::minutes(2)), durability_(1),
                                                     transaction_type_(transaction_type::TWO_PHASE) {}
 
        transaction_options::transaction_type transaction_options::get_transaction_type() const {
            return transaction_type_;
        }
 
        transaction_options &transaction_options::set_transaction_type(transaction_type type) {
            transaction_type_ = type;
            return *this;
        }
 
        std::chrono::milliseconds transaction_options::get_timeout() const {
            return timeout_;
        }
 
        transaction_options &transaction_options::set_timeout(std::chrono::milliseconds duration) {
            if (duration.count() <= 0) {
                BOOST_THROW_EXCEPTION(exception::illegal_state("TransactionOptions::setTimeout",
                                                                       "Timeout must be positive!"));
            }
            timeout_ = duration;
            return *this;
        }
 
        int transaction_options::get_durability() const {
            return durability_;
        }
 
        transaction_options &transaction_options::set_durability(int num_machines) {
            if (num_machines < 0) {
                BOOST_THROW_EXCEPTION(exception::illegal_state("TransactionOptions::setDurability",
                                                                       "Durability cannot be negative!"));
            }
            this->durability_ = num_machines;
            return *this;
        }
    }
}
 
namespace std {
    std::size_t hash<std::pair<std::string, std::string>>::operator()(
            const std::pair<std::string, std::string> &val) const noexcept {
        return std::hash<std::string>{}(val.first + val.second);
    }
}