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

Extracted streaming_context into separa teclass

Browse Source
This commit is contained in:
Teemu Ollakka
2018-06-13 16:59:25 +03:00
parent 37efaba3e5
commit 3cd7a11cb2
6 changed files with 108 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
include/wsrep/client_context.hpp
View File

@ -242,7 +242,7 @@ namespace wsrep
} }
int enable_streaming( int enable_streaming(
enum wsrep::transaction_context::streaming_context::fragment_unit enum wsrep::streaming_context::fragment_unit
fragment_unit, fragment_unit,
size_t fragment_size) size_t fragment_size)
{ {

97
include/wsrep/streaming_context.hpp Normal file
View File

@ -0,0 +1,97 @@
//
// Copyright (C) 2018 Codership Oy <info@codership.com>
//
#ifndef WSREP_STREAMING_CONTEXT_HPP
#define WSREP_STREAMING_CONTEXT_HPP
namespace wsrep
{
class streaming_context
{
public:
enum fragment_unit
{
row,
bytes,
statement
};
streaming_context()
: fragments_()
, rollback_replicated_for_()
, fragment_unit_()
, fragment_size_()
, bytes_certified_()
, unit_counter_()
{ }
void enable(enum fragment_unit fragment_unit, size_t fragment_size)
{
assert(fragment_size > 0);
fragment_unit_ = fragment_unit;
fragment_size_ = fragment_size;
}
enum fragment_unit fragment_unit() const { return fragment_unit_; }
size_t fragment_size() const { return fragment_size_; }
void disable()
{
fragment_size_ = 0;
}
void certified(wsrep::seqno seqno)
{
fragments_.push_back(seqno);
}
size_t fragments_certified() const
{
return fragments_.size();
}
size_t bytes_certified() const
{
return bytes_certified_;
}
void rolled_back(wsrep::transaction_id id)
{
assert(rollback_replicated_for_ == wsrep::transaction_id::invalid());
rollback_replicated_for_ = id;
}
bool rolled_back() const
{
return (rollback_replicated_for_ !=
wsrep::transaction_id::invalid());
}
size_t unit_counter() const { return unit_counter_; }
void increment_unit_counter(size_t inc)
{ unit_counter_ += inc; }
void reset_unit_counter() { unit_counter_ = 0; }
const std::vector<wsrep::seqno>& fragments() const
{
return fragments_;
}
void cleanup()
{
fragments_.clear();
rollback_replicated_for_ = wsrep::transaction_id::invalid();
bytes_certified_ = 0;
unit_counter_ = 0;
}
private:
std::vector<wsrep::seqno> fragments_;
wsrep::transaction_id rollback_replicated_for_;
enum fragment_unit fragment_unit_;
size_t fragment_size_;
size_t bytes_certified_;
size_t unit_counter_;
};
}
#endif // WSREP_STREAMING_CONTEXT_HPP

84
include/wsrep/transaction_context.hpp
View File

@ -8,6 +8,7 @@
#include "provider.hpp" #include "provider.hpp"
#include "server_context.hpp" #include "server_context.hpp"
#include "transaction_id.hpp" #include "transaction_id.hpp"
#include "streaming_context.hpp"
#include "lock.hpp" #include "lock.hpp"
#include <cassert> #include <cassert>
@ -154,89 +155,8 @@ namespace wsrep
int flags_; int flags_;
bool pa_unsafe_; bool pa_unsafe_;
bool certified_; bool certified_;
public: public:
class streaming_context wsrep::streaming_context streaming_context_;
{
public:
enum fragment_unit
{
row,
bytes,
statement
};
streaming_context()
: fragments_()
, rollback_replicated_for_()
, fragment_unit_()
, fragment_size_()
, bytes_certified_()
, unit_counter_()
{ }
void enable(enum fragment_unit fragment_unit, size_t fragment_size)
{
assert(fragment_size > 0);
fragment_unit_ = fragment_unit;
fragment_size_ = fragment_size;
}
enum fragment_unit fragment_unit() const { return fragment_unit_; }
size_t fragment_size() const { return fragment_size_; }
void disable()
{
fragment_size_ = 0;
}
void certified(wsrep::seqno seqno)
{
fragments_.push_back(seqno);
}
size_t fragments_certified() const
{
return fragments_.size();
}
size_t bytes_certified() const
{
return bytes_certified_;
}
void rolled_back(wsrep::transaction_id id)
{
assert(rollback_replicated_for_ == wsrep::transaction_id::invalid());
rollback_replicated_for_ = id;
}
bool rolled_back() const
{
return (rollback_replicated_for_ !=
wsrep::transaction_id::invalid());
}
size_t unit_counter() const { return unit_counter_; }
void increment_unit_counter(size_t inc)
{ unit_counter_ += inc; }
void reset_unit_counter() { unit_counter_ = 0; }
void cleanup()
{
fragments_.clear();
rollback_replicated_for_ = wsrep::transaction_id::invalid();
bytes_certified_ = 0;
unit_counter_ = 0;
}
private:
std::vector<wsrep::seqno> fragments_;
wsrep::transaction_id rollback_replicated_for_;
enum fragment_unit fragment_unit_;
size_t fragment_size_;
size_t bytes_certified_;
size_t unit_counter_;
} streaming_context_;
}; };
static inline std::string to_string(enum wsrep::transaction_context::state state) static inline std::string to_string(enum wsrep::transaction_context::state state)

1
src/transaction_context.cpp
View File

@ -764,7 +764,6 @@ int wsrep::transaction_context::certify_commit(
{ {
assert(lock.owns_lock()); assert(lock.owns_lock());
assert(active()); assert(active());
client_context_.wait_for_replayers(lock); client_context_.wait_for_replayers(lock);
assert(lock.owns_lock()); assert(lock.owns_lock());

6
test/client_context_fixture.hpp
View File

@ -162,7 +162,7 @@ namespace
// Verify initial state // Verify initial state
BOOST_REQUIRE(tc.active() == false); BOOST_REQUIRE(tc.active() == false);
BOOST_REQUIRE(tc.state() == wsrep::transaction_context::s_executing); BOOST_REQUIRE(tc.state() == wsrep::transaction_context::s_executing);
cc.enable_streaming(wsrep::transaction_context::streaming_context::row, 1); cc.enable_streaming(wsrep::streaming_context::row, 1);
} }
wsrep::fake_server_context sc; wsrep::fake_server_context sc;
wsrep::fake_client_context cc; wsrep::fake_client_context cc;
@ -182,7 +182,7 @@ namespace
// Verify initial state // Verify initial state
BOOST_REQUIRE(tc.active() == false); BOOST_REQUIRE(tc.active() == false);
BOOST_REQUIRE(tc.state() == wsrep::transaction_context::s_executing); BOOST_REQUIRE(tc.state() == wsrep::transaction_context::s_executing);
cc.enable_streaming(wsrep::transaction_context::streaming_context::bytes, 1); cc.enable_streaming(wsrep::streaming_context::bytes, 1);
} }
wsrep::fake_server_context sc; wsrep::fake_server_context sc;
wsrep::fake_client_context cc; wsrep::fake_client_context cc;
@ -202,7 +202,7 @@ namespace
// Verify initial state // Verify initial state
BOOST_REQUIRE(tc.active() == false); BOOST_REQUIRE(tc.active() == false);
BOOST_REQUIRE(tc.state() == wsrep::transaction_context::s_executing); BOOST_REQUIRE(tc.state() == wsrep::transaction_context::s_executing);
cc.enable_streaming(wsrep::transaction_context::streaming_context::row, 1); cc.enable_streaming(wsrep::streaming_context::row, 1);
} }
wsrep::fake_server_context sc; wsrep::fake_server_context sc;

10
test/transaction_context_test.cpp
View File

@ -924,7 +924,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_context_row_batch_streaming_1pc_commit,
streaming_client_fixture_row) streaming_client_fixture_row)
{ {
BOOST_REQUIRE(cc.enable_streaming( BOOST_REQUIRE(cc.enable_streaming(
wsrep::transaction_context::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);
@ -1067,7 +1067,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_context_byte_batch_streaming_1pc_commit,
{ {
BOOST_REQUIRE( BOOST_REQUIRE(
cc.enable_streaming( cc.enable_streaming(
wsrep::transaction_context::streaming_context::bytes, 2) == 0); wsrep::streaming_context::bytes, 2) == 0);
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);
@ -1090,7 +1090,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_context_statement_streaming_1pc_commit,
{ {
BOOST_REQUIRE( BOOST_REQUIRE(
cc.enable_streaming( cc.enable_streaming(
wsrep::transaction_context::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);
@ -1111,7 +1111,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_context_statement_batch_streaming_1pc_commit
{ {
BOOST_REQUIRE( BOOST_REQUIRE(
cc.enable_streaming( cc.enable_streaming(
wsrep::transaction_context::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);
@ -1137,7 +1137,7 @@ BOOST_FIXTURE_TEST_CASE(transaction_context_statement_streaming_cert_fail,
{ {
BOOST_REQUIRE( BOOST_REQUIRE(
cc.enable_streaming( cc.enable_streaming(
wsrep::transaction_context::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);