Merge branch '2.3' into develop
This commit is contained in:
@ -282,16 +282,20 @@ RWBackend* RWSplitSession::get_slave_backend(int max_rlag)
|
||||
auto counts = get_slave_counts(m_raw_backends, m_current_master);
|
||||
int best_priority {INT_MAX};
|
||||
auto current_rank = get_current_rank();
|
||||
// Slaves can be taken into use if we need more slave connections
|
||||
bool need_slaves = counts.second < m_router->max_slave_count();
|
||||
|
||||
// Create a list of backends valid for read operations
|
||||
for (auto& backend : m_raw_backends)
|
||||
{
|
||||
bool can_take_slave_into_use = !backend->in_use() && can_recover_servers()
|
||||
&& backend->can_connect() && counts.second < m_router->max_slave_count()
|
||||
&& (backend->is_slave() || backend->is_master());
|
||||
|
||||
// We can take the current master back into use even for reads
|
||||
bool my_master = backend == m_current_master;
|
||||
bool can_take_into_use = !backend->in_use() && can_recover_servers() && backend->can_connect();
|
||||
bool master_or_slave = backend->is_master() || backend->is_slave();
|
||||
bool is_usable = backend->in_use() || can_take_slave_into_use;
|
||||
|
||||
// The server is usable if it's already in use or it can be taken into use and we need either more
|
||||
// slaves or a master.
|
||||
bool is_usable = backend->in_use() || (can_take_into_use && (need_slaves || my_master));
|
||||
bool rlag_ok = rpl_lag_is_ok(backend, max_rlag);
|
||||
int priority = get_backend_priority(backend, m_config.master_accept_reads);
|
||||
auto rank = backend->server()->rank();
|
||||
|
@ -155,6 +155,7 @@ int32_t RWSplitSession::routeQuery(GWBUF* querybuf)
|
||||
m_query_queue.emplace_back(querybuf);
|
||||
querybuf = NULL;
|
||||
rval = 1;
|
||||
mxb_assert(m_expected_responses != 0);
|
||||
|
||||
if (m_expected_responses == 0 && !route_stored_query())
|
||||
{
|
||||
@ -806,6 +807,7 @@ void RWSplitSession::clientReply(GWBUF* writebuf, DCB* backend_dcb)
|
||||
// Backend is still in use and has more session commands to execute
|
||||
if (backend->execute_session_command() && backend->is_waiting_result())
|
||||
{
|
||||
MXS_INFO("%lu session commands left on '%s'", backend->session_command_count(), backend->name());
|
||||
m_expected_responses++;
|
||||
}
|
||||
}
|
||||
@ -948,6 +950,44 @@ bool RWSplitSession::start_trx_replay()
|
||||
return rval;
|
||||
}
|
||||
|
||||
bool RWSplitSession::retry_master_query(RWBackend* backend)
|
||||
{
|
||||
bool can_continue = false;
|
||||
|
||||
if (backend->has_session_commands())
|
||||
{
|
||||
// Try to route the session command again. If the master is not available, the response will be
|
||||
// returned from one of the slaves.
|
||||
|
||||
mxb_assert(!m_current_query.get());
|
||||
mxb_assert(!m_sescmd_list.empty());
|
||||
mxb_assert(m_sescmd_count >= 2);
|
||||
MXS_INFO("Retrying session command due to master failure: %s",
|
||||
backend->next_session_command()->to_string().c_str());
|
||||
|
||||
// Before routing it, pop the failed session command off the list and decrement the number of
|
||||
// executed session commands. This "overwrites" the existing command and prevents history duplication.
|
||||
m_sescmd_list.pop_back();
|
||||
--m_sescmd_count;
|
||||
|
||||
retry_query(backend->next_session_command()->deep_copy_buffer());
|
||||
can_continue = true;
|
||||
}
|
||||
else if (m_current_query.get())
|
||||
{
|
||||
retry_query(m_current_query.release());
|
||||
can_continue = true;
|
||||
}
|
||||
else
|
||||
{
|
||||
// This should never happen
|
||||
mxb_assert_message(!true, "m_current_query is empty and no session commands being executed");
|
||||
MXS_ERROR("Current query unexpectedly empty when trying to retry query on master");
|
||||
}
|
||||
|
||||
return can_continue;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Router error handling routine
|
||||
*
|
||||
@ -1014,8 +1054,7 @@ void RWSplitSession::handleError(GWBUF* errmsgbuf,
|
||||
|
||||
if (can_retry_query())
|
||||
{
|
||||
can_continue = true;
|
||||
retry_query(m_current_query.release());
|
||||
can_continue = retry_master_query(backend);
|
||||
}
|
||||
else if (m_config.master_failure_mode == RW_ERROR_ON_WRITE)
|
||||
{
|
||||
@ -1151,6 +1190,7 @@ bool RWSplitSession::handle_error_new_connection(DCB* backend_dcb, GWBUF* errmsg
|
||||
|
||||
if (stored && m_config.retry_failed_reads)
|
||||
{
|
||||
mxb_assert(m_expected_responses == 0);
|
||||
MXS_INFO("Re-routing failed read after server '%s' failed", backend->name());
|
||||
retry_query(stored, 0);
|
||||
}
|
||||
|
@ -186,6 +186,7 @@ private:
|
||||
int get_max_replication_lag();
|
||||
mxs::RWBackend* get_backend_from_dcb(DCB* dcb);
|
||||
|
||||
bool retry_master_query(mxs::RWBackend* backend);
|
||||
void handle_error_reply_client(DCB* backend_dcb, GWBUF* errmsg);
|
||||
bool handle_error_new_connection(DCB* backend_dcb, GWBUF* errmsg);
|
||||
void manage_transactions(mxs::RWBackend* backend, GWBUF* writebuf);
|
||||
|
Reference in New Issue
Block a user