lib/wsrep-lib
1
0
Fork 0
mirror of https://github.com/codership/wsrep-lib.git synced 2025-07-28 20:02:00 +03:00
Code Activity

Made transaction streaming context private and provided accessor method.

Browse Source
This commit is contained in:
Teemu Ollakka
2018-07-09 08:49:29 +03:00
parent 7c424d8337
commit 95dbab4c08
5 changed files with 62 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
include/wsrep/client_state.hpp
View File

@ -286,7 +286,7 @@ namespace wsrep
{ {
assert(mode_ == m_local); assert(mode_ == m_local);
assert(state_ == s_exec); assert(state_ == s_exec);
return (transaction_.streaming_context_.fragment_size() ? return (transaction_.streaming_context().fragment_size() ?
transaction_.after_row() : 0); transaction_.after_row() : 0);
} }
@ -307,11 +307,10 @@ namespace wsrep
fragment_unit, fragment_unit,
size_t fragment_size); size_t fragment_size);
void disable_streaming() /**
{ * Disable streaming for context.
assert(state_ == s_exec && mode_ == m_local); */
transaction_.streaming_context_.disable(); void disable_streaming();
}
/** /**
* Prepare write set meta data for fragment storage ordering. * Prepare write set meta data for fragment storage ordering.
@ -426,7 +425,7 @@ namespace wsrep
{ {
assert(mode_ == m_high_priority); assert(mode_ == m_high_priority);
transaction_.start_transaction(transaction.id()); transaction_.start_transaction(transaction.id());
transaction_.streaming_context_ = transaction.streaming_context_; transaction_.streaming_context() = transaction.streaming_context();
} }
/** @name Non-transactional operations */ /** @name Non-transactional operations */

5
include/wsrep/transaction.hpp
View File

@ -135,6 +135,10 @@ namespace wsrep
wsrep::ws_handle& ws_handle() { return ws_handle_; } wsrep::ws_handle& ws_handle() { return ws_handle_; }
const wsrep::ws_handle& ws_handle() const { return ws_handle_; } const wsrep::ws_handle& ws_handle() const { return ws_handle_; }
const wsrep::ws_meta& ws_meta() const { return ws_meta_; } const wsrep::ws_meta& ws_meta() const { return ws_meta_; }
const wsrep::streaming_context& streaming_context() const
{ return streaming_context_; }
wsrep::streaming_context& streaming_context()
{ return streaming_context_; }
private: private:
transaction(const transaction&); transaction(const transaction&);
transaction operator=(const transaction&); transaction operator=(const transaction&);
@ -162,7 +166,6 @@ namespace wsrep
int flags_; int flags_;
bool pa_unsafe_; bool pa_unsafe_;
bool certified_; bool certified_;
public:
wsrep::streaming_context streaming_context_; wsrep::streaming_context streaming_context_;
}; };

36
src/client_state.cpp
View File

@ -218,26 +218,40 @@ int wsrep::client_state::after_statement()
return 0; return 0;
} }
//////////////////////////////////////////////////////////////////////////////
// Streaming //
//////////////////////////////////////////////////////////////////////////////
int wsrep::client_state::enable_streaming( int wsrep::client_state::enable_streaming(
enum wsrep::streaming_context::fragment_unit enum wsrep::streaming_context::fragment_unit
fragment_unit, fragment_unit,
size_t fragment_size) size_t fragment_size)
{ {
assert(mode_ == m_local); assert(mode_ == m_local);
if (transaction_.active() && if (transaction_.streaming_context().fragments_certified() &&
transaction_.streaming_context_.fragment_unit() != transaction_.streaming_context().fragment_unit() !=
fragment_unit) fragment_unit)
{ {
wsrep::log_error() wsrep::log_error()
<< "Changing fragment unit for active transaction " << "Changing fragment unit for active streaming transaction "
<< "not allowed"; << "not allowed";
return 1; return 1;
} }
transaction_.streaming_context_.enable( transaction_.streaming_context().enable(fragment_unit, fragment_size);
fragment_unit, fragment_size);
return 0; return 0;
} }
void wsrep::client_state::disable_streaming()
{
assert(state_ == s_exec && mode_ == m_local);
transaction_.streaming_context().disable();
}
//////////////////////////////////////////////////////////////////////////////
// TOI //
//////////////////////////////////////////////////////////////////////////////
int wsrep::client_state::enter_toi(const wsrep::key_array& keys, int wsrep::client_state::enter_toi(const wsrep::key_array& keys,
const wsrep::const_buffer& buffer, const wsrep::const_buffer& buffer,
int flags) int flags)
@ -304,6 +318,10 @@ int wsrep::client_state::leave_toi()
return ret; return ret;
} }
///////////////////////////////////////////////////////////////////////////////
// RSU //
///////////////////////////////////////////////////////////////////////////////
int wsrep::client_state::begin_rsu(int timeout) int wsrep::client_state::begin_rsu(int timeout)
{ {
if (server_state_.desync()) if (server_state_.desync())
@ -349,6 +367,10 @@ int wsrep::client_state::end_rsu()
return ret; return ret;
} }
///////////////////////////////////////////////////////////////////////////////
// TOI //
///////////////////////////////////////////////////////////////////////////////
int wsrep::client_state::sync_wait(int timeout) int wsrep::client_state::sync_wait(int timeout)
{ {
std::pair<wsrep::gtid, enum wsrep::provider::status> result( std::pair<wsrep::gtid, enum wsrep::provider::status> result(
@ -370,7 +392,9 @@ int wsrep::client_state::sync_wait(int timeout)
return ret; return ret;
} }
// Private ///////////////////////////////////////////////////////////////////////////////
// Private //
///////////////////////////////////////////////////////////////////////////////
void wsrep::client_state::update_last_written_gtid(const wsrep::gtid& gtid) void wsrep::client_state::update_last_written_gtid(const wsrep::gtid& gtid)
{ {

38
test/transaction_test.cpp
View File

@ -982,7 +982,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_row_streaming_1pc_commit,
{ {
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
BOOST_REQUIRE(cc.ordered_commit() == 0); BOOST_REQUIRE(cc.ordered_commit() == 0);
BOOST_REQUIRE(cc.after_commit() == 0); BOOST_REQUIRE(cc.after_commit() == 0);
@ -1002,9 +1002,9 @@ BOOST_FIXTURE_TEST_CASE(transaction_row_batch_streaming_1pc_commit,
wsrep::streaming_context::row, 2) == 0); wsrep::streaming_context::row, 2) == 0);
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 0); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
BOOST_REQUIRE(cc.ordered_commit() == 0); BOOST_REQUIRE(cc.ordered_commit() == 0);
BOOST_REQUIRE(cc.after_commit() == 0); BOOST_REQUIRE(cc.after_commit() == 0);
@ -1023,11 +1023,11 @@ BOOST_FIXTURE_TEST_CASE(
{ {
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.after_statement() == 0); BOOST_REQUIRE(cc.after_statement() == 0);
BOOST_REQUIRE(cc.before_statement() == 0); BOOST_REQUIRE(cc.before_statement() == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 2); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 2);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
BOOST_REQUIRE(cc.ordered_commit() == 0); BOOST_REQUIRE(cc.ordered_commit() == 0);
BOOST_REQUIRE(cc.after_commit() == 0); BOOST_REQUIRE(cc.after_commit() == 0);
@ -1045,7 +1045,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_row_streaming_rollback,
{ {
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_rollback() == 0); BOOST_REQUIRE(cc.before_rollback() == 0);
BOOST_REQUIRE(cc.after_rollback() == 0); BOOST_REQUIRE(cc.after_rollback() == 0);
BOOST_REQUIRE(cc.after_statement() == 0); BOOST_REQUIRE(cc.after_statement() == 0);
@ -1062,7 +1062,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_row_streaming_cert_fail_non_commit,
{ {
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
sc.provider().certify_result_ = wsrep::provider::error_certification_failed; sc.provider().certify_result_ = wsrep::provider::error_certification_failed;
BOOST_REQUIRE(cc.after_row() == 1); BOOST_REQUIRE(cc.after_row() == 1);
sc.provider().certify_result_ = wsrep::provider::success; sc.provider().certify_result_ = wsrep::provider::success;
@ -1082,7 +1082,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_row_streaming_cert_fail_commit,
{ {
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
sc.provider().certify_result_ = wsrep::provider::error_certification_failed; sc.provider().certify_result_ = wsrep::provider::error_certification_failed;
BOOST_REQUIRE(cc.before_commit() == 1); BOOST_REQUIRE(cc.before_commit() == 1);
BOOST_REQUIRE(tc.state() == wsrep::transaction::s_cert_failed); BOOST_REQUIRE(tc.state() == wsrep::transaction::s_cert_failed);
@ -1104,7 +1104,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_row_streaming_bf_abort_committing,
{ {
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
BOOST_REQUIRE(tc.state() == wsrep::transaction::s_committing); BOOST_REQUIRE(tc.state() == wsrep::transaction::s_committing);
wsrep_test::bf_abort_ordered(cc); wsrep_test::bf_abort_ordered(cc);
@ -1127,7 +1127,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_byte_streaming_1pc_commit,
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
cc.bytes_generated_ = 1; cc.bytes_generated_ = 1;
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
BOOST_REQUIRE(cc.ordered_commit() == 0); BOOST_REQUIRE(cc.ordered_commit() == 0);
BOOST_REQUIRE(cc.after_commit() == 0); BOOST_REQUIRE(cc.after_commit() == 0);
@ -1146,10 +1146,10 @@ BOOST_FIXTURE_TEST_CASE(transaction_byte_batch_streaming_1pc_commit,
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
cc.bytes_generated_ = 1; cc.bytes_generated_ = 1;
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 0); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 0);
cc.bytes_generated_ = 2; cc.bytes_generated_ = 2;
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
BOOST_REQUIRE(cc.ordered_commit() == 0); BOOST_REQUIRE(cc.ordered_commit() == 0);
BOOST_REQUIRE(cc.after_commit() == 0); BOOST_REQUIRE(cc.after_commit() == 0);
@ -1168,9 +1168,9 @@ BOOST_FIXTURE_TEST_CASE(transaction_statement_streaming_1pc_commit,
wsrep::streaming_context::statement, 1) == 0); wsrep::streaming_context::statement, 1) == 0);
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 0); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 0);
BOOST_REQUIRE(cc.after_statement() == 0); BOOST_REQUIRE(cc.after_statement() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_statement() == 0); BOOST_REQUIRE(cc.before_statement() == 0);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
BOOST_REQUIRE(cc.ordered_commit() == 0); BOOST_REQUIRE(cc.ordered_commit() == 0);
@ -1189,14 +1189,14 @@ BOOST_FIXTURE_TEST_CASE(transaction_statement_batch_streaming_1pc_commit,
wsrep::streaming_context::statement, 2) == 0); wsrep::streaming_context::statement, 2) == 0);
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 0); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 0);
BOOST_REQUIRE(cc.after_statement() == 0); BOOST_REQUIRE(cc.after_statement() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 0); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 0);
BOOST_REQUIRE(cc.before_statement() == 0); BOOST_REQUIRE(cc.before_statement() == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 0); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 0);
BOOST_REQUIRE(cc.after_statement() == 0); BOOST_REQUIRE(cc.after_statement() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_statement() == 0); BOOST_REQUIRE(cc.before_statement() == 0);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
BOOST_REQUIRE(cc.ordered_commit() == 0); BOOST_REQUIRE(cc.ordered_commit() == 0);
@ -1215,7 +1215,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_statement_streaming_cert_fail,
wsrep::streaming_context::statement, 1) == 0); wsrep::streaming_context::statement, 1) == 0);
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 0); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 0);
sc.provider().certify_result_ = wsrep::provider::error_certification_failed; sc.provider().certify_result_ = wsrep::provider::error_certification_failed;
BOOST_REQUIRE(cc.after_statement()); BOOST_REQUIRE(cc.after_statement());
BOOST_REQUIRE(cc.current_error() == wsrep::e_deadlock_error); BOOST_REQUIRE(cc.current_error() == wsrep::e_deadlock_error);

6
test/transaction_test_2pc.cpp
View File

@ -189,7 +189,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_streaming_2pc_commit,
{ {
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.before_prepare() == 0); BOOST_REQUIRE(cc.before_prepare() == 0);
BOOST_REQUIRE(cc.after_prepare() == 0); BOOST_REQUIRE(cc.after_prepare() == 0);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);
@ -206,11 +206,11 @@ BOOST_FIXTURE_TEST_CASE(transaction_streaming_2pc_commit_two_statements,
{ {
BOOST_REQUIRE(cc.start_transaction(1) == 0); BOOST_REQUIRE(cc.start_transaction(1) == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 1); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
BOOST_REQUIRE(cc.after_statement() == 0); BOOST_REQUIRE(cc.after_statement() == 0);
BOOST_REQUIRE(cc.before_statement() == 0); BOOST_REQUIRE(cc.before_statement() == 0);
BOOST_REQUIRE(cc.after_row() == 0); BOOST_REQUIRE(cc.after_row() == 0);
BOOST_REQUIRE(tc.streaming_context_.fragments_certified() == 2); BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 2);
BOOST_REQUIRE(cc.before_prepare() == 0); BOOST_REQUIRE(cc.before_prepare() == 0);
BOOST_REQUIRE(cc.after_prepare() == 0); BOOST_REQUIRE(cc.after_prepare() == 0);
BOOST_REQUIRE(cc.before_commit() == 0); BOOST_REQUIRE(cc.before_commit() == 0);