hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move most readwritesplit functions into classes

Browse Source 
Most of the funtionality is now a member function of either the RWSplit or
RWSplitSession class. This removes the need to pass the router and session
parameters to all functions.
This commit is contained in:
Markus Mäkelä
2018-04-03 23:18:57 +03:00
parent ce500d782a
commit 15f15be49d
11 changed files with 360 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files

254
server/modules/routing/readwritesplit/readwritesplit.cc
View File

@ -32,7 +32,6 @@
#include <maxscale/spinlock.h> #include <maxscale/spinlock.h>
#include <maxscale/mysql_utils.h> #include <maxscale/mysql_utils.h>
#include "rwsplit_internal.hh"
#include "rwsplitsession.hh" #include "rwsplitsession.hh"
#include "routeinfo.hh" #include "routeinfo.hh"
@ -50,40 +49,20 @@ using namespace maxscale;
/** Maximum number of slaves */ /** Maximum number of slaves */
#define MAX_SLAVE_COUNT "255" #define MAX_SLAVE_COUNT "255"
/*
* The functions that implement the router module API
*/
static bool rwsplit_process_router_options(RWSplit *router,
char **options);
static void handle_error_reply_client(MXS_SESSION *ses, RWSplitSession *rses,
DCB *backend_dcb, GWBUF *errmsg);
static bool handle_error_new_connection(RWSplit *inst,
RWSplitSession **rses,
DCB *backend_dcb, GWBUF *errmsg);
static bool route_stored_query(RWSplitSession *rses);
/**
* Internal functions
*/
/* /*
* @brief Get the maximum replication lag for this router * @brief Get the maximum replication lag for this router
* *
* @param rses Router client session * @param rses Router client session
* @return Replication lag from configuration or very large number * @return Replication lag from configuration or very large number
*/ */
int rses_get_max_replication_lag(RWSplitSession *rses) int RWSplitSession::get_max_replication_lag()
{ {
int conf_max_rlag; int conf_max_rlag;
CHK_CLIENT_RSES(rses);
/** if there is no configured value, then longest possible int is used */ /** if there is no configured value, then longest possible int is used */
if (rses->rses_config.max_slave_replication_lag > 0) if (rses_config.max_slave_replication_lag > 0)
{ {
conf_max_rlag = rses->rses_config.max_slave_replication_lag; conf_max_rlag = rses_config.max_slave_replication_lag;
} }
else else
{ {
@ -94,25 +73,18 @@ int rses_get_max_replication_lag(RWSplitSession *rses)
} }
/** /**
* @brief Find a back end reference that matches the given DCB * @brief Find the backend reference that matches the given DCB
* *
* Finds out if there is a backend reference pointing at the DCB given as * @param dcb DCB to match
* parameter.
* *
* @param rses router client session * @return The correct reference
* @param dcb DCB
*
* @return backend reference pointer if succeed or NULL
*/ */
SRWBackend& RWSplitSession::get_backend_from_dcb(DCB *dcb)
static inline SRWBackend& get_backend_from_dcb(RWSplitSession *rses, DCB *dcb)
{ {
ss_dassert(dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER); ss_dassert(dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER);
CHK_DCB(dcb); CHK_DCB(dcb);
CHK_CLIENT_RSES(rses);
for (SRWBackendList::iterator it = rses->backends.begin(); for (auto it = backends.begin(); it != backends.end(); it++)
it != rses->backends.end(); it++)
{ {
SRWBackend& backend = *it; SRWBackend& backend = *it;
@ -277,14 +249,10 @@ static bool handle_max_slaves(Config& config, const char *str)
* @param backend_dcb DCB for the backend server that has failed * @param backend_dcb DCB for the backend server that has failed
* @param errmsg GWBUF containing the error message * @param errmsg GWBUF containing the error message
*/ */
static void handle_error_reply_client(MXS_SESSION *ses, RWSplitSession *rses, void RWSplitSession::handle_error_reply_client(DCB *backend_dcb, GWBUF *errmsg)
DCB *backend_dcb, GWBUF *errmsg)
{ {
mxs_session_state_t sesstate = client_dcb->session->state;
mxs_session_state_t sesstate = ses->state; SRWBackend& backend = get_backend_from_dcb(backend_dcb);
DCB *client_dcb = ses->client_dcb;
SRWBackend& backend = get_backend_from_dcb(rses, backend_dcb);
backend->close(); backend->close();
@ -295,24 +263,22 @@ static void handle_error_reply_client(MXS_SESSION *ses, RWSplitSession *rses,
} }
} }
static bool reroute_stored_statement(RWSplitSession *rses, const SRWBackend& old, GWBUF *stored) bool RWSplitSession::reroute_stored_statement(const SRWBackend& old, GWBUF *stored)
{ {
bool success = false; bool success = false;
if (!session_trx_is_active(rses->client_dcb->session)) if (!session_trx_is_active(client_dcb->session))
{ {
/** /**
* Only try to retry the read if autocommit is enabled and we are * Only try to retry the read if autocommit is enabled and we are
* outside of a transaction * outside of a transaction
*/ */
for (SRWBackendList::iterator it = rses->backends.begin(); for (auto it = backends.begin(); it != backends.end(); it++)
it != rses->backends.end(); it++)
{ {
SRWBackend& backend = *it; SRWBackend& backend = *it;
if (backend->in_use() && backend != old && if (backend->in_use() && backend != old &&
!backend->is_master() && !backend->is_master() && backend->is_slave())
backend->is_slave())
{ {
/** Found a valid candidate; a non-master slave that's in use */ /** Found a valid candidate; a non-master slave that's in use */
if (backend->write(stored)) if (backend->write(stored))
@ -320,25 +286,25 @@ static bool reroute_stored_statement(RWSplitSession *rses, const SRWBackend& old
MXS_INFO("Retrying failed read at '%s'.", backend->name()); MXS_INFO("Retrying failed read at '%s'.", backend->name());
ss_dassert(backend->get_reply_state() == REPLY_STATE_DONE); ss_dassert(backend->get_reply_state() == REPLY_STATE_DONE);
backend->set_reply_state(REPLY_STATE_START); backend->set_reply_state(REPLY_STATE_START);
rses->expected_responses++; expected_responses++;
success = true; success = true;
break; break;
} }
} }
} }
if (!success && rses->current_master && rses->current_master->in_use()) if (!success && current_master && current_master->in_use())
{ {
/** /**
* Either we failed to write to the slave or no valid slave was found. * Either we failed to write to the slave or no valid slave was found.
* Try to retry the read on the master. * Try to retry the read on the master.
*/ */
if (rses->current_master->write(stored)) if (current_master->write(stored))
{ {
MXS_INFO("Retrying failed read at '%s'.", rses->current_master->name()); MXS_INFO("Retrying failed read at '%s'.", current_master->name());
ss_dassert(rses->current_master->get_reply_state() == REPLY_STATE_DONE); ss_dassert(current_master->get_reply_state() == REPLY_STATE_DONE);
rses->current_master->set_reply_state(REPLY_STATE_START); current_master->set_reply_state(REPLY_STATE_START);
rses->expected_responses++; expected_responses++;
success = true; success = true;
} }
} }
@ -362,21 +328,17 @@ static bool reroute_stored_statement(RWSplitSession *rses, const SRWBackend& old
* @return true if there are enough backend connections to continue, false if * @return true if there are enough backend connections to continue, false if
* not * not
*/ */
static bool handle_error_new_connection(RWSplit *inst, bool RWSplitSession::handle_error_new_connection(DCB *backend_dcb, GWBUF *errmsg)
RWSplitSession **rses,
DCB *backend_dcb, GWBUF *errmsg)
{ {
RWSplitSession *myrses = *rses; SRWBackend& backend = get_backend_from_dcb(backend_dcb);
SRWBackend& backend = get_backend_from_dcb(myrses, backend_dcb);
MXS_SESSION* ses = backend_dcb->session; MXS_SESSION* ses = backend_dcb->session;
bool route_stored = false; bool route_stored = false;
CHK_SESSION(ses); CHK_SESSION(ses);
if (backend->is_waiting_result()) if (backend->is_waiting_result())
{ {
ss_dassert(myrses->expected_responses > 0); ss_dassert(expected_responses > 0);
myrses->expected_responses--; expected_responses--;
/** /**
* A query was sent through the backend and it is waiting for a reply. * A query was sent through the backend and it is waiting for a reply.
@ -387,7 +349,7 @@ static bool handle_error_new_connection(RWSplit *inst,
const SERVER *target = NULL; const SERVER *target = NULL;
if (!session_take_stmt(backend_dcb->session, &stored, &target) || if (!session_take_stmt(backend_dcb->session, &stored, &target) ||
target != backend->backend()->server || target != backend->backend()->server ||
!reroute_stored_statement(*rses, backend, stored)) !reroute_stored_statement(backend, stored))
{ {
/** /**
* We failed to route the stored statement or no statement was * We failed to route the stored statement or no statement was
@ -407,7 +369,7 @@ static bool handle_error_new_connection(RWSplit *inst,
client_dcb->func.write(client_dcb, gwbuf_clone(errmsg)); client_dcb->func.write(client_dcb, gwbuf_clone(errmsg));
} }
if (myrses->expected_responses == 0) if (expected_responses == 0)
{ {
/** The response from this server was the last one, try to /** The response from this server was the last one, try to
* route all queued queries */ * route all queued queries */
@ -424,25 +386,26 @@ static bool handle_error_new_connection(RWSplit *inst,
if (route_stored) if (route_stored)
{ {
route_stored_query(myrses); route_stored_query();
} }
int max_nslaves = inst->max_slave_count(); int max_nslaves = router->max_slave_count();
bool succp; bool succp;
/** /**
* Try to get replacement slave or at least the minimum * Try to get replacement slave or at least the minimum
* number of slave connections for router session. * number of slave connections for router session.
*/ */
if (myrses->recv_sescmd > 0 && inst->config().disable_sescmd_history) if (recv_sescmd > 0 && rses_config.disable_sescmd_history)
{ {
succp = inst->have_enough_servers(); succp = router->have_enough_servers();
} }
else else
{ {
succp = select_connect_backend_servers(inst, ses, myrses->backends, succp = router->select_connect_backend_servers(ses, backends,
myrses->current_master, &myrses->sescmd_list, current_master,
&myrses->expected_responses, &sescmd_list,
connection_type::SLAVE); &expected_responses,
connection_type::SLAVE);
} }
return succp; return succp;
@ -458,43 +421,43 @@ static bool handle_error_new_connection(RWSplit *inst,
* @param rses Router client session * @param rses Router client session
* @return True if a stored query was routed successfully * @return True if a stored query was routed successfully
*/ */
static bool route_stored_query(RWSplitSession *rses) bool RWSplitSession::route_stored_query()
{ {
bool rval = true; bool rval = true;
/** Loop over the stored statements as long as the routeQuery call doesn't /** Loop over the stored statements as long as the routeQuery call doesn't
* append more data to the queue. If it appends data to the queue, we need * append more data to the queue. If it appends data to the queue, we need
* to wait for a response before attempting another reroute */ * to wait for a response before attempting another reroute */
while (rses->query_queue) while (query_queue)
{ {
GWBUF* query_queue = modutil_get_next_MySQL_packet(&rses->query_queue); GWBUF* query_queue = modutil_get_next_MySQL_packet(&query_queue);
query_queue = gwbuf_make_contiguous(query_queue); query_queue = gwbuf_make_contiguous(query_queue);
/** Store the query queue locally for the duration of the routeQuery call. /** Store the query queue locally for the duration of the routeQuery call.
* This prevents recursive calls into this function. */ * This prevents recursive calls into this function. */
GWBUF *temp_storage = rses->query_queue; GWBUF *temp_storage = query_queue;
rses->query_queue = NULL; query_queue = NULL;
// TODO: Move the handling of queued queries to the client protocol // TODO: Move the handling of queued queries to the client protocol
// TODO: module where the command tracking is done automatically. // TODO: module where the command tracking is done automatically.
uint8_t cmd = mxs_mysql_get_command(query_queue); uint8_t cmd = mxs_mysql_get_command(query_queue);
mysql_protocol_set_current_command(rses->client_dcb, (mxs_mysql_cmd_t)cmd); mysql_protocol_set_current_command(client_dcb, (mxs_mysql_cmd_t)cmd);
if (!rses->routeQuery(query_queue)) if (!routeQuery(query_queue))
{ {
rval = false; rval = false;
MXS_ERROR("Failed to route queued query."); MXS_ERROR("Failed to route queued query.");
} }
if (rses->query_queue == NULL) if (query_queue == NULL)
{ {
/** Query successfully routed and no responses are expected */ /** Query successfully routed and no responses are expected */
rses->query_queue = temp_storage; query_queue = temp_storage;
} }
else else
{ {
/** Routing was stopped, we need to wait for a response before retrying */ /** Routing was stopped, we need to wait for a response before retrying */
rses->query_queue = gwbuf_append(temp_storage, rses->query_queue); query_queue = gwbuf_append(temp_storage, query_queue);
break; break;
} }
} }
@ -669,21 +632,18 @@ RWSplitSession* RWSplit::newSession(MXS_SESSION *session)
*/ */
void RWSplitSession::close() void RWSplitSession::close()
{ {
RWSplitSession *router_cli_ses = this; if (!rses_closed)
CHK_CLIENT_RSES(router_cli_ses);
if (!router_cli_ses->rses_closed)
{ {
router_cli_ses->rses_closed = true; rses_closed = true;
close_all_connections(router_cli_ses->backends); close_all_connections(backends);
if (MXS_LOG_PRIORITY_IS_ENABLED(LOG_INFO) && if (MXS_LOG_PRIORITY_IS_ENABLED(LOG_INFO) &&
router_cli_ses->sescmd_list.size()) sescmd_list.size())
{ {
std::string sescmdstr; std::string sescmdstr;
for (mxs::SessionCommandList::iterator it = router_cli_ses->sescmd_list.begin(); for (mxs::SessionCommandList::iterator it = sescmd_list.begin();
it != router_cli_ses->sescmd_list.end(); it++) it != sescmd_list.end(); it++)
{ {
mxs::SSessionCommand& scmd = *it; mxs::SSessionCommand& scmd = *it;
sescmdstr += scmd->to_string(); sescmdstr += scmd->to_string();
@ -697,29 +657,25 @@ void RWSplitSession::close()
int32_t RWSplitSession::routeQuery(GWBUF* querybuf) int32_t RWSplitSession::routeQuery(GWBUF* querybuf)
{ {
RWSplit *inst = router;
RWSplitSession *rses = this;
int rval = 0; int rval = 0;
CHK_CLIENT_RSES(rses); if (rses_closed)
if (rses->rses_closed)
{ {
closed_session_reply(querybuf); closed_session_reply(querybuf);
} }
else else
{ {
if (rses->query_queue == NULL && if (query_queue == NULL &&
(rses->expected_responses == 0 || (expected_responses == 0 ||
mxs_mysql_get_command(querybuf) == MXS_COM_STMT_FETCH || mxs_mysql_get_command(querybuf) == MXS_COM_STMT_FETCH ||
rses->load_data_state == LOAD_DATA_ACTIVE || load_data_state == LOAD_DATA_ACTIVE ||
rses->large_query)) large_query))
{ {
/** Gather the information required to make routing decisions */ /** Gather the information required to make routing decisions */
RouteInfo info(rses, querybuf); RouteInfo info(this, querybuf);
/** No active or pending queries */ /** No active or pending queries */
if (route_single_stmt(inst, rses, querybuf, info)) if (route_single_stmt(querybuf, info))
{ {
rval = 1; rval = 1;
} }
@ -730,16 +686,15 @@ int32_t RWSplitSession::routeQuery(GWBUF* querybuf)
* We are already processing a request from the client. Store the * We are already processing a request from the client. Store the
* new query and wait for the previous one to complete. * new query and wait for the previous one to complete.
*/ */
ss_dassert(rses->expected_responses || rses->query_queue); ss_dassert(expected_responses || query_queue);
MXS_INFO("Storing query (len: %d cmd: %0x), expecting %d replies to current command", MXS_INFO("Storing query (len: %d cmd: %0x), expecting %d replies to current command",
gwbuf_length(querybuf), GWBUF_DATA(querybuf)[4], gwbuf_length(querybuf), GWBUF_DATA(querybuf)[4], expected_responses);
rses->expected_responses); query_queue = gwbuf_append(query_queue, querybuf);
rses->query_queue = gwbuf_append(rses->query_queue, querybuf);
querybuf = NULL; querybuf = NULL;
rval = 1; rval = 1;
ss_dassert(rses->expected_responses > 0); ss_dassert(expected_responses > 0);
if (rses->expected_responses == 0 && !route_stored_query(rses)) if (expected_responses == 0 && !route_stored_query())
{ {
rval = 0; rval = 0;
} }
@ -906,7 +861,7 @@ static void log_unexpected_response(DCB* dcb, GWBUF* buffer)
* @param proto MySQLProtocol * @param proto MySQLProtocol
* @return reset buffer * @return reset buffer
*/ */
GWBUF *discard_master_wait_gtid_result(GWBUF *buffer, RWSplitSession *rses) GWBUF* RWSplitSession::discard_master_wait_gtid_result(GWBUF *buffer)
{ {
uint8_t header_and_command[MYSQL_HEADER_LEN + 1]; uint8_t header_and_command[MYSQL_HEADER_LEN + 1];
uint8_t packet_len = 0; uint8_t packet_len = 0;
@ -917,15 +872,15 @@ GWBUF *discard_master_wait_gtid_result(GWBUF *buffer, RWSplitSession *rses)
/* ignore error packet */ /* ignore error packet */
if (MYSQL_GET_COMMAND(header_and_command) == MYSQL_REPLY_ERR) if (MYSQL_GET_COMMAND(header_and_command) == MYSQL_REPLY_ERR)
{ {
rses->wait_gtid_state = EXPECTING_NOTHING; wait_gtid_state = EXPECTING_NOTHING;
return buffer; return buffer;
} }
/* this packet must be an ok packet now */ /* this packet must be an ok packet now */
ss_dassert(MYSQL_GET_COMMAND(header_and_command) == MYSQL_REPLY_OK); ss_dassert(MYSQL_GET_COMMAND(header_and_command) == MYSQL_REPLY_OK);
packet_len = MYSQL_GET_PAYLOAD_LEN(header_and_command) + MYSQL_HEADER_LEN; packet_len = MYSQL_GET_PAYLOAD_LEN(header_and_command) + MYSQL_HEADER_LEN;
rses->wait_gtid_state = EXPECTING_REAL_RESULT; wait_gtid_state = EXPECTING_REAL_RESULT;
rses->next_seq = 1; next_seq = 1;
return gwbuf_consume(buffer, packet_len); return gwbuf_consume(buffer, packet_len);
} }
@ -937,19 +892,19 @@ GWBUF *discard_master_wait_gtid_result(GWBUF *buffer, RWSplitSession *rses)
* @param proto MySQLProtocol * @param proto MySQLProtocol
* *
*/ */
void correct_packet_sequence(GWBUF *buffer, RWSplitSession *rses) void RWSplitSession::correct_packet_sequence(GWBUF *buffer)
{ {
uint8_t header[3]; uint8_t header[3];
uint32_t offset = 0; uint32_t offset = 0;
uint32_t packet_len = 0; uint32_t packet_len = 0;
if (rses->wait_gtid_state == EXPECTING_REAL_RESULT) if (wait_gtid_state == EXPECTING_REAL_RESULT)
{ {
while (gwbuf_copy_data(buffer, offset, 3, header) == 3) while (gwbuf_copy_data(buffer, offset, 3, header) == 3)
{ {
packet_len = MYSQL_GET_PAYLOAD_LEN(header) + MYSQL_HEADER_LEN; packet_len = MYSQL_GET_PAYLOAD_LEN(header) + MYSQL_HEADER_LEN;
uint8_t *seq = gwbuf_byte_pointer(buffer, offset + MYSQL_SEQ_OFFSET); uint8_t *seq = gwbuf_byte_pointer(buffer, offset + MYSQL_SEQ_OFFSET);
*seq = rses->next_seq; *seq = next_seq;
rses->next_seq++; next_seq++;
offset += packet_len; offset += packet_len;
} }
} }
@ -957,38 +912,35 @@ void correct_packet_sequence(GWBUF *buffer, RWSplitSession *rses)
void RWSplitSession::clientReply(GWBUF *writebuf, DCB *backend_dcb) void RWSplitSession::clientReply(GWBUF *writebuf, DCB *backend_dcb)
{ {
RWSplitSession *rses = this;
RWSplit *inst = router;
DCB *client_dcb = backend_dcb->session->client_dcb; DCB *client_dcb = backend_dcb->session->client_dcb;
CHK_CLIENT_RSES(rses); ss_dassert(!rses_closed);
ss_dassert(!rses->rses_closed);
SRWBackend& backend = get_backend_from_dcb(rses, backend_dcb); SRWBackend& backend = get_backend_from_dcb(backend_dcb);
if (inst->config().enable_causal_read && if (rses_config.enable_causal_read &&
GWBUF_IS_REPLY_OK(writebuf) && GWBUF_IS_REPLY_OK(writebuf) &&
backend == rses->current_master) backend == current_master)
{ {
/** Save gtid position */ /** Save gtid position */
char *tmp = gwbuf_get_property(writebuf, (char *)"gtid"); char *tmp = gwbuf_get_property(writebuf, (char *)"gtid");
if (tmp) if (tmp)
{ {
rses->gtid_pos = std::string(tmp); gtid_pos = std::string(tmp);
} }
} }
if (rses->wait_gtid_state == EXPECTING_WAIT_GTID_RESULT) if (wait_gtid_state == EXPECTING_WAIT_GTID_RESULT)
{ {
ss_dassert(rses->rses_config.enable_causal_read); ss_dassert(rses_config.enable_causal_read);
if ((writebuf = discard_master_wait_gtid_result(writebuf, rses)) == NULL) if ((writebuf = discard_master_wait_gtid_result(writebuf)) == NULL)
{ {
// Nothing to route, return // Nothing to route, return
return; return;
} }
} }
if (rses->wait_gtid_state == EXPECTING_REAL_RESULT) if (wait_gtid_state == EXPECTING_REAL_RESULT)
{ {
correct_packet_sequence(writebuf, rses); correct_packet_sequence(writebuf);
} }
if (backend->get_reply_state() == REPLY_STATE_DONE) if (backend->get_reply_state() == REPLY_STATE_DONE)
@ -1011,33 +963,33 @@ void RWSplitSession::clientReply(GWBUF *writebuf, DCB *backend_dcb)
{ {
/** Got a complete reply, acknowledge the write and decrement expected response count */ /** Got a complete reply, acknowledge the write and decrement expected response count */
backend->ack_write(); backend->ack_write();
rses->expected_responses--; expected_responses--;
ss_dassert(rses->expected_responses >= 0); ss_dassert(expected_responses >= 0);
ss_dassert(backend->get_reply_state() == REPLY_STATE_DONE); ss_dassert(backend->get_reply_state() == REPLY_STATE_DONE);
MXS_INFO("Reply complete, last reply from %s", backend->name()); MXS_INFO("Reply complete, last reply from %s", backend->name());
} }
else else
{ {
MXS_INFO("Reply not yet complete. Waiting for %d replies, got one from %s", MXS_INFO("Reply not yet complete. Waiting for %d replies, got one from %s",
rses->expected_responses, backend->name()); expected_responses, backend->name());
} }
if (backend->have_session_commands()) if (backend->have_session_commands())
{ {
/** Reply to an executed session command */ /** Reply to an executed session command */
process_sescmd_response(rses, backend, &writebuf); process_sescmd_response(backend, &writebuf);
} }
if (backend->have_session_commands()) if (backend->have_session_commands())
{ {
if (backend->execute_session_command()) if (backend->execute_session_command())
{ {
rses->expected_responses++; expected_responses++;
} }
} }
else if (rses->expected_responses == 0 && rses->query_queue) else if (expected_responses == 0 && query_queue)
{ {
route_stored_query(rses); route_stored_query();
} }
if (writebuf) if (writebuf)
@ -1073,12 +1025,9 @@ void RWSplitSession::handleError(GWBUF *errmsgbuf, DCB *problem_dcb,
mxs_error_action_t action, bool *succp) mxs_error_action_t action, bool *succp)
{ {
ss_dassert(problem_dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER); ss_dassert(problem_dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER);
RWSplit *inst = router;
RWSplitSession *rses = this;
CHK_CLIENT_RSES(rses);
CHK_DCB(problem_dcb); CHK_DCB(problem_dcb);
if (rses->rses_closed) if (rses_closed)
{ {
*succp = false; *succp = false;
return; return;
@ -1087,18 +1036,17 @@ void RWSplitSession::handleError(GWBUF *errmsgbuf, DCB *problem_dcb,
MXS_SESSION *session = problem_dcb->session; MXS_SESSION *session = problem_dcb->session;
ss_dassert(session); ss_dassert(session);
SRWBackend& backend = get_backend_from_dcb(rses, problem_dcb); SRWBackend& backend = get_backend_from_dcb(problem_dcb);
ss_dassert(backend->in_use()); ss_dassert(backend->in_use());
switch (action) switch (action)
{ {
case ERRACT_NEW_CONNECTION: case ERRACT_NEW_CONNECTION:
{ {
if (rses->current_master && rses->current_master->in_use() && if (current_master && current_master->in_use() && current_master == backend)
rses->current_master == backend)
{ {
/** The connection to the master has failed */ /** The connection to the master has failed */
SERVER *srv = rses->current_master->server(); SERVER *srv = current_master->server();
bool can_continue = false; bool can_continue = false;
if (!backend->is_waiting_result()) if (!backend->is_waiting_result())
@ -1113,7 +1061,7 @@ void RWSplitSession::handleError(GWBUF *errmsgbuf, DCB *problem_dcb,
* can't be sure whether it was executed or not. In this * can't be sure whether it was executed or not. In this
* case the safest thing to do is to close the client * case the safest thing to do is to close the client
* connection. */ * connection. */
if (rses->rses_config.master_failure_mode != RW_FAIL_INSTANTLY) if (rses_config.master_failure_mode != RW_FAIL_INSTANTLY)
{ {
can_continue = true; can_continue = true;
} }
@ -1121,15 +1069,15 @@ void RWSplitSession::handleError(GWBUF *errmsgbuf, DCB *problem_dcb,
else else
{ {
// We were expecting a response but we aren't going to get one // We were expecting a response but we aren't going to get one
rses->expected_responses--; expected_responses--;
if (rses->rses_config.master_failure_mode == RW_ERROR_ON_WRITE) if (rses_config.master_failure_mode == RW_ERROR_ON_WRITE)
{ {
/** In error_on_write mode, the session can continue even /** In error_on_write mode, the session can continue even
* if the master is lost. Send a read-only error to * if the master is lost. Send a read-only error to
* the client to let it know that the query failed. */ * the client to let it know that the query failed. */
can_continue = true; can_continue = true;
send_readonly_error(rses->client_dcb); send_readonly_error(client_dcb);
} }
if (!SERVER_IS_MASTER(srv) && !srv->master_err_is_logged) if (!SERVER_IS_MASTER(srv) && !srv->master_err_is_logged)
@ -1153,7 +1101,7 @@ void RWSplitSession::handleError(GWBUF *errmsgbuf, DCB *problem_dcb,
} }
else else
{ {
if (rses->target_node && rses->target_node == backend && if (target_node && target_node == backend &&
session_trx_is_read_only(problem_dcb->session)) session_trx_is_read_only(problem_dcb->session))
{ {
/** /**
@ -1167,7 +1115,7 @@ void RWSplitSession::handleError(GWBUF *errmsgbuf, DCB *problem_dcb,
else else
{ {
/** Try to replace the failed connection with a new one */ /** Try to replace the failed connection with a new one */
*succp = handle_error_new_connection(inst, &rses, problem_dcb, errmsgbuf); *succp = handle_error_new_connection(problem_dcb, errmsgbuf);
} }
} }
@ -1177,7 +1125,7 @@ void RWSplitSession::handleError(GWBUF *errmsgbuf, DCB *problem_dcb,
case ERRACT_REPLY_CLIENT: case ERRACT_REPLY_CLIENT:
{ {
handle_error_reply_client(session, rses, problem_dcb, errmsgbuf); handle_error_reply_client(problem_dcb, errmsgbuf);
*succp = false; /*< no new backend servers were made available */ *succp = false; /*< no new backend servers were made available */
break; break;
} }

36
server/modules/routing/readwritesplit/readwritesplit.hh
View File

@ -30,10 +30,11 @@
#include <maxscale/log_manager.h> #include <maxscale/log_manager.h>
#include <maxscale/router.hh> #include <maxscale/router.hh>
#include <maxscale/service.h> #include <maxscale/service.h>
#include <maxscale/backend.hh>
#include <maxscale/session_command.hh> #include <maxscale/session_command.hh>
#include <maxscale/protocol/mysql.h> #include <maxscale/protocol/mysql.h>
#include "rwbackend.hh"
enum backend_type_t enum backend_type_t
{ {
BE_UNDEFINED = -1, BE_UNDEFINED = -1,
@ -43,6 +44,12 @@ enum backend_type_t
BE_COUNT BE_COUNT
}; };
enum connection_type
{
ALL,
SLAVE
};
enum route_target_t enum route_target_t
{ {
TARGET_UNDEFINED = 0x00, TARGET_UNDEFINED = 0x00,
@ -248,7 +255,12 @@ public:
const Stats& stats() const; const Stats& stats() const;
int max_slave_count() const; int max_slave_count() const;
bool have_enough_servers() const; bool have_enough_servers() const;
bool select_connect_backend_servers(MXS_SESSION *session,
mxs::SRWBackendList& backends,
mxs::SRWBackend& current_master,
mxs::SessionCommandList* sescmd_list,
int* expected_responses,
connection_type type);
// API functions // API functions
/** /**
@ -349,3 +361,23 @@ static inline const char* failure_mode_to_str(enum failure_mode type)
return "UNDEFINED_MODE"; return "UNDEFINED_MODE";
} }
} }
void closed_session_reply(GWBUF *querybuf);
bool send_readonly_error(DCB *dcb);
mxs::SRWBackend get_root_master(const mxs::SRWBackendList& backends);
/**
* Get total slave count and connected slave count
*
* @param backends List of backend servers
* @param master Current master
*
* @return Total number of slaves and number of slaves we are connected to
*/
std::pair<int, int> get_slave_counts(mxs::SRWBackendList& backends, mxs::SRWBackend& master);
/*
* The following are implemented in rwsplit_tmp_table_multi.c
*/
void close_all_connections(mxs::SRWBackendList& backends);

2
server/modules/routing/readwritesplit/routeinfo.cc
View File

@ -15,8 +15,6 @@
#include <maxscale/alloc.h> #include <maxscale/alloc.h>
#include <maxscale/queryclassifier.hh> #include <maxscale/queryclassifier.hh>
#include "rwsplitsession.hh" #include "rwsplitsession.hh"
// Only for is_ps_command
#include "rwsplit_internal.hh"
#define RWSPLIT_TRACE_MSG_LEN 1000 #define RWSPLIT_TRACE_MSG_LEN 1000

109
server/modules/routing/readwritesplit/rwsplit_internal.hh
View File

@ -1,109 +0,0 @@
#pragma once
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2020-01-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#include "readwritesplit.hh"
#include <string>
#include <utility>
#include <maxscale/query_classifier.h>
#include <maxscale/protocol/mysql.h>
#include "rwsplitsession.hh"
class RouteInfo;
#define RW_CHK_DCB(b, d) \
do{ \
if(d->state == DCB_STATE_DISCONNECTED){ \
MXS_NOTICE("DCB was closed on line %d and another attempt to close it is made on line %d." , \
(b) ? (b)->closed_at : -1, __LINE__); \
} \
}while (false)
#define RW_CLOSE_BREF(b) do{ if (b){ (b)->closed_at = __LINE__; } } while (false)
/*
* The following are implemented in rwsplit_mysql.c
*/
bool route_single_stmt(RWSplit *inst, RWSplitSession *rses,
GWBUF *querybuf);
void closed_session_reply(GWBUF *querybuf);
void print_error_packet(RWSplitSession *rses, GWBUF *buf, DCB *dcb);
void check_session_command_reply(GWBUF *buffer, mxs::SRWBackend& backend);
bool execute_sescmd_in_backend(mxs::SRWBackend& backend_ref);
bool handle_target_is_all(route_target_t route_target,
RWSplit *inst, RWSplitSession *rses,
GWBUF *querybuf, int packet_type, uint32_t qtype);
bool send_readonly_error(DCB *dcb);
/*
* The following are implemented in readwritesplit.c
*/
int router_handle_state_switch(DCB *dcb, DCB_REASON reason, void *data);
int rses_get_max_replication_lag(RWSplitSession *rses);
/*
* The following are implemented in rwsplit_route_stmt.c
*/
bool route_single_stmt(RWSplit *inst, RWSplitSession *rses,
GWBUF *querybuf, const RouteInfo& info);
mxs::SRWBackend get_target_backend(RWSplitSession *rses, backend_type_t btype,
char *name, int max_rlag);
mxs::SRWBackend handle_hinted_target(RWSplitSession *rses, GWBUF *querybuf,
route_target_t route_target);
mxs::SRWBackend handle_slave_is_target(RWSplit *inst, RWSplitSession *rses,
uint8_t cmd, uint32_t id);
bool handle_master_is_target(RWSplit *inst, RWSplitSession *rses,
mxs::SRWBackend* dest);
bool handle_got_target(RWSplit *inst, RWSplitSession *rses,
GWBUF *querybuf, mxs::SRWBackend& target, bool store);
bool route_session_write(RWSplitSession *rses, GWBUF *querybuf,
uint8_t command, uint32_t type);
void process_sescmd_response(RWSplitSession* rses, mxs::SRWBackend& bref, GWBUF** ppPacket);
/*
* The following are implemented in rwsplit_select_backends.c
*/
/** What sort of connections should be create */
enum connection_type
{
ALL,
SLAVE
};
bool select_connect_backend_servers(RWSplit *inst, MXS_SESSION *session,
mxs::SRWBackendList& backends,
mxs::SRWBackend& current_master,
mxs::SessionCommandList* sescmd,
int* expected_responses,
connection_type type);
mxs::SRWBackend get_root_master(const mxs::SRWBackendList& backends);
/**
* Get total slave count and connected slave count
*
* @param backends List of backend servers
* @param master Current master
*
* @return Total number of slaves and number of slaves we are connected to
*/
std::pair<int, int> get_slave_counts(mxs::SRWBackendList& backends, mxs::SRWBackend& master);
/*
* The following are implemented in rwsplit_tmp_table_multi.c
*/
void close_all_connections(mxs::SRWBackendList& backends);

13
server/modules/routing/readwritesplit/rwsplit_mysql.cc
View File

@ -12,7 +12,7 @@
*/ */
#include "readwritesplit.hh" #include "readwritesplit.hh"
#include "rwsplit_internal.hh" #include "rwsplitsession.hh"
#include <strings.h> #include <strings.h>
#include <string.h> #include <string.h>
@ -79,9 +79,8 @@
* @param qtype Query type * @param qtype Query type
* @return bool indicating whether the session can continue * @return bool indicating whether the session can continue
*/ */
bool handle_target_is_all(route_target_t route_target, RWSplit *inst, bool RWSplitSession::handle_target_is_all(route_target_t route_target, GWBUF *querybuf,
RWSplitSession *rses, GWBUF *querybuf, int packet_type, uint32_t qtype)
int packet_type, uint32_t qtype)
{ {
bool result = false; bool result = false;
@ -105,18 +104,18 @@ bool handle_target_is_all(route_target_t route_target, RWSplit *inst,
if (errbuf) if (errbuf)
{ {
rses->client_dcb->func.write(rses->client_dcb, errbuf); client_dcb->func.write(client_dcb, errbuf);
result = true; result = true;
} }
MXS_FREE(query_str); MXS_FREE(query_str);
MXS_FREE(qtype_str); MXS_FREE(qtype_str);
} }
else if (route_session_write(rses, gwbuf_clone(querybuf), packet_type, qtype)) else if (route_session_write(gwbuf_clone(querybuf), packet_type, qtype))
{ {
result = true; result = true;
atomic_add_uint64(&inst->stats().n_all, 1); atomic_add_uint64(&router->stats().n_all, 1);
} }
return result; return result;

3
server/modules/routing/readwritesplit/rwsplit_ps.cc
View File

@ -12,13 +12,12 @@
*/ */
#include "readwritesplit.hh" #include "readwritesplit.hh"
#include "rwsplit_ps.hh"
#include <maxscale/alloc.h> #include <maxscale/alloc.h>
#include <maxscale/query_classifier.h> #include <maxscale/query_classifier.h>
#include <maxscale/protocol/mysql.h> #include <maxscale/protocol/mysql.h>
#include "rwsplit_internal.hh"
uint32_t get_prepare_type(GWBUF* buffer) uint32_t get_prepare_type(GWBUF* buffer)
{ {
uint32_t type; uint32_t type;

315
server/modules/routing/readwritesplit/rwsplit_route_stmt.cc
View File

@ -12,6 +12,7 @@
*/ */
#include "readwritesplit.hh" #include "readwritesplit.hh"
#include "rwsplitsession.hh"
#include <stdint.h> #include <stdint.h>
#include <stdlib.h> #include <stdlib.h>
@ -27,7 +28,6 @@
#include <maxscale/utils.hh> #include <maxscale/utils.hh>
#include "routeinfo.hh" #include "routeinfo.hh"
#include "rwsplit_internal.hh"
using namespace maxscale; using namespace maxscale;
@ -67,16 +67,14 @@ static SRWBackend compare_backends(SRWBackend a, SRWBackend b, select_criteria_t
return p(a, b) <= 0 ? a : b; return p(a, b) <= 0 ? a : b;
} }
void handle_connection_keepalive(RWSplit *inst, RWSplitSession *rses, void RWSplitSession::handle_connection_keepalive(SRWBackend& target)
SRWBackend& target)
{ {
ss_dassert(target); ss_dassert(target);
ss_debug(int nserv = 0); ss_debug(int nserv = 0);
/** Each heartbeat is 1/10th of a second */ /** Each heartbeat is 1/10th of a second */
int keepalive = inst->config().connection_keepalive * 10; int keepalive = rses_config.connection_keepalive * 10;
for (SRWBackendList::iterator it = rses->backends.begin(); for (auto it = backends.begin(); it != backends.end(); it++)
it != rses->backends.end(); it++)
{ {
SRWBackend backend = *it; SRWBackend backend = *it;
@ -94,15 +92,10 @@ void handle_connection_keepalive(RWSplit *inst, RWSplitSession *rses,
} }
} }
ss_dassert(nserv < rses->rses_nbackends); ss_dassert(nserv < rses_nbackends);
} }
static inline bool locked_to_master(RWSplitSession *rses) bool RWSplitSession::prepare_target(SRWBackend& target, route_target_t route_target)
{
return rses->large_query || (rses->current_master && rses->target_node == rses->current_master);
}
static bool prepare_target(RWSplitSession* rses, SRWBackend& target, route_target_t route_target)
{ {
bool rval = true; bool rval = true;
@ -110,17 +103,17 @@ static bool prepare_target(RWSplitSession* rses, SRWBackend& target, route_targe
if (!target->in_use() && target->can_connect()) if (!target->in_use() && target->can_connect())
{ {
if (TARGET_IS_SLAVE(route_target) || if (TARGET_IS_SLAVE(route_target) ||
(rses->rses_config.master_reconnection && TARGET_IS_MASTER(route_target))) (rses_config.master_reconnection && TARGET_IS_MASTER(route_target)))
{ {
if ((!rses->rses_config.disable_sescmd_history || rses->recv_sescmd == 0)) if ((!rses_config.disable_sescmd_history || recv_sescmd == 0))
{ {
rval = target->connect(rses->client_dcb->session, &rses->sescmd_list); rval = target->connect(client_dcb->session, &sescmd_list);
} }
else else
{ {
MXS_ERROR("Cannot reconnect to server '%s', session command" MXS_ERROR("Cannot reconnect to server '%s', session command"
" history is disabled (session has executed" " history is disabled (session has executed"
" %lu session commands).", target->name(), rses->recv_sescmd); " %lu session commands).", target->name(), recv_sescmd);
} }
} }
else if (TARGET_IS_MASTER(route_target)) else if (TARGET_IS_MASTER(route_target))
@ -144,14 +137,14 @@ static bool prepare_target(RWSplitSession* rses, SRWBackend& target, route_targe
* @return true if routing succeed or if it failed due to unsupported query. * @return true if routing succeed or if it failed due to unsupported query.
* false if backend failure was encountered. * false if backend failure was encountered.
*/ */
bool route_single_stmt(RWSplit *inst, RWSplitSession *rses, GWBUF *querybuf, const RouteInfo& info) bool RWSplitSession::route_single_stmt(GWBUF *querybuf, const RouteInfo& info)
{ {
bool succp = false; bool succp = false;
uint32_t stmt_id = info.stmt_id; uint32_t stmt_id = info.stmt_id;
uint8_t command = info.command; uint8_t command = info.command;
uint32_t qtype = info.type; uint32_t qtype = info.type;
route_target_t route_target = info.target; route_target_t route_target = info.target;
bool not_locked_to_master = !locked_to_master(rses); bool not_locked_to_master = !locked_to_master();
if (not_locked_to_master && mxs_mysql_is_ps_command(command)) if (not_locked_to_master && mxs_mysql_is_ps_command(command))
{ {
@ -165,18 +158,18 @@ bool route_single_stmt(RWSplit *inst, RWSplitSession *rses, GWBUF *querybuf, con
if (TARGET_IS_ALL(route_target)) if (TARGET_IS_ALL(route_target))
{ {
// TODO: Handle payloads larger than (2^24 - 1) bytes that are routed to all servers // TODO: Handle payloads larger than (2^24 - 1) bytes that are routed to all servers
succp = handle_target_is_all(route_target, inst, rses, querybuf, command, qtype); succp = handle_target_is_all(route_target, querybuf, command, qtype);
} }
else else
{ {
bool store_stmt = false; bool store_stmt = false;
if (rses->large_query) if (large_query)
{ {
/** We're processing a large query that's split across multiple packets. /** We're processing a large query that's split across multiple packets.
* Route it to the same backend where we routed the previous packet. */ * Route it to the same backend where we routed the previous packet. */
ss_dassert(rses->prev_target); ss_dassert(prev_target);
target = rses->prev_target; target = prev_target;
succp = true; succp = true;
} }
else if (TARGET_IS_NAMED_SERVER(route_target) || TARGET_IS_RLAG_MAX(route_target)) else if (TARGET_IS_NAMED_SERVER(route_target) || TARGET_IS_RLAG_MAX(route_target))
@ -186,35 +179,35 @@ bool route_single_stmt(RWSplit *inst, RWSplitSession *rses, GWBUF *querybuf, con
* hint which sets maximum allowed replication lag for the * hint which sets maximum allowed replication lag for the
* backend. * backend.
*/ */
if ((target = handle_hinted_target(rses, querybuf, route_target))) if ((target = handle_hinted_target(querybuf, route_target)))
{ {
succp = true; succp = true;
} }
} }
else if (TARGET_IS_SLAVE(route_target)) else if (TARGET_IS_SLAVE(route_target))
{ {
if ((target = handle_slave_is_target(inst, rses, command, stmt_id))) if ((target = handle_slave_is_target(command, stmt_id)))
{ {
succp = true; succp = true;
store_stmt = rses->rses_config.retry_failed_reads; store_stmt = rses_config.retry_failed_reads;
} }
} }
else if (TARGET_IS_MASTER(route_target)) else if (TARGET_IS_MASTER(route_target))
{ {
succp = handle_master_is_target(inst, rses, &target); succp = handle_master_is_target(&target);
if (!rses->rses_config.strict_multi_stmt && if (!rses_config.strict_multi_stmt &&
!rses->rses_config.strict_sp_calls && !rses_config.strict_sp_calls &&
rses->target_node == rses->current_master) target_node == current_master)
{ {
/** Reset the forced node as we're in relaxed multi-statement mode */ /** Reset the forced node as we're in relaxed multi-statement mode */
rses->target_node.reset(); target_node.reset();
} }
} }
if (succp && target) if (succp && target)
{ {
if (!prepare_target(rses, target, route_target)) if (!prepare_target(target, route_target))
{ {
// The connection to target was down and we failed to reconnect // The connection to target was down and we failed to reconnect
succp = false; succp = false;
@ -222,14 +215,14 @@ bool route_single_stmt(RWSplit *inst, RWSplitSession *rses, GWBUF *querybuf, con
else if (target->have_session_commands()) else if (target->have_session_commands())
{ {
// We need to wait until the session commands are executed // We need to wait until the session commands are executed
rses->expected_responses++; expected_responses++;
rses->query_queue = gwbuf_append(rses->query_queue, gwbuf_clone(querybuf)); query_queue = gwbuf_append(query_queue, gwbuf_clone(querybuf));
} }
else else
{ {
// Target server was found and is in the correct state // Target server was found and is in the correct state
ss_dassert(!store_stmt || TARGET_IS_SLAVE(route_target)); ss_dassert(!store_stmt || TARGET_IS_SLAVE(route_target));
succp = handle_got_target(inst, rses, querybuf, target, store_stmt); succp = handle_got_target(querybuf, target, store_stmt);
if (succp && command == MXS_COM_STMT_EXECUTE && not_locked_to_master) if (succp && command == MXS_COM_STMT_EXECUTE && not_locked_to_master)
{ {
@ -237,17 +230,17 @@ bool route_single_stmt(RWSplit *inst, RWSplitSession *rses, GWBUF *querybuf, con
* information is used to route all COM_STMT_FETCH commands * information is used to route all COM_STMT_FETCH commands
* to the same server where the COM_STMT_EXECUTE was done. */ * to the same server where the COM_STMT_EXECUTE was done. */
ss_dassert(stmt_id > 0); ss_dassert(stmt_id > 0);
rses->exec_map[stmt_id] = target; exec_map[stmt_id] = target;
MXS_INFO("COM_STMT_EXECUTE on %s", target->uri()); MXS_INFO("COM_STMT_EXECUTE on %s", target->uri());
} }
} }
} }
} }
if (succp && inst->config().connection_keepalive && if (succp && router->config().connection_keepalive &&
(TARGET_IS_SLAVE(route_target) || TARGET_IS_MASTER(route_target))) (TARGET_IS_SLAVE(route_target) || TARGET_IS_MASTER(route_target)))
{ {
handle_connection_keepalive(inst, rses, target); handle_connection_keepalive(target);
} }
return succp; return succp;
@ -256,10 +249,9 @@ bool route_single_stmt(RWSplit *inst, RWSplitSession *rses, GWBUF *querybuf, con
/** /**
* Purge session command history * Purge session command history
* *
* @param rses Router session
* @param sescmd Executed session command * @param sescmd Executed session command
*/ */
static void purge_history(RWSplitSession* rses, mxs::SSessionCommand& sescmd) void RWSplitSession::purge_history(mxs::SSessionCommand& sescmd)
{ {
/** /**
* We can try to purge duplicate text protocol session commands. This * We can try to purge duplicate text protocol session commands. This
@ -281,21 +273,21 @@ static void purge_history(RWSplitSession* rses, mxs::SSessionCommand& sescmd)
// As the PS handles map to explicit IDs, we must retain all COM_STMT_PREPARE commands // As the PS handles map to explicit IDs, we must retain all COM_STMT_PREPARE commands
if (sescmd->get_command() != MXS_COM_STMT_PREPARE) if (sescmd->get_command() != MXS_COM_STMT_PREPARE)
{ {
auto first = std::find_if(rses->sescmd_list.begin(), rses->sescmd_list.end(), auto first = std::find_if(sescmd_list.begin(), sescmd_list.end(),
mxs::equal_pointees(sescmd)); mxs::equal_pointees(sescmd));
if (first != rses->sescmd_list.end()) if (first != sescmd_list.end())
{ {
// We have at least one of these commands. See if we have a second one // We have at least one of these commands. See if we have a second one
auto second = std::find_if(std::next(first), rses->sescmd_list.end(), auto second = std::find_if(std::next(first), sescmd_list.end(),
mxs::equal_pointees(sescmd)); mxs::equal_pointees(sescmd));
if (second != rses->sescmd_list.end()) if (second != sescmd_list.end())
{ {
// We have a total of three commands, remove the middle one // We have a total of three commands, remove the middle one
auto old_cmd = *second; auto old_cmd = *second;
rses->sescmd_responses.erase(old_cmd->get_position()); sescmd_responses.erase(old_cmd->get_position());
rses->sescmd_list.erase(second); sescmd_list.erase(second);
} }
} }
} }
@ -311,7 +303,6 @@ static void purge_history(RWSplitSession* rses, mxs::SSessionCommand& sescmd)
* *
* The first OK packet is replied to the client. * The first OK packet is replied to the client.
* *
* @param router_cli_ses Client's router session pointer
* @param querybuf GWBUF including the query to be routed * @param querybuf GWBUF including the query to be routed
* @param inst Router instance * @param inst Router instance
* @param packet_type Type of MySQL packet * @param packet_type Type of MySQL packet
@ -321,11 +312,10 @@ static void purge_history(RWSplitSession* rses, mxs::SSessionCommand& sescmd)
* backends being used, otherwise false. * backends being used, otherwise false.
* *
*/ */
bool route_session_write(RWSplitSession *rses, GWBUF *querybuf, bool RWSplitSession::route_session_write(GWBUF *querybuf, uint8_t command, uint32_t type)
uint8_t command, uint32_t type)
{ {
/** The SessionCommand takes ownership of the buffer */ /** The SessionCommand takes ownership of the buffer */
uint64_t id = rses->sescmd_count++; uint64_t id = sescmd_count++;
mxs::SSessionCommand sescmd(new mxs::SessionCommand(querybuf, id)); mxs::SSessionCommand sescmd(new mxs::SessionCommand(querybuf, id));
bool expecting_response = mxs_mysql_command_will_respond(command); bool expecting_response = mxs_mysql_command_will_respond(command);
int nsucc = 0; int nsucc = 0;
@ -335,13 +325,12 @@ bool route_session_write(RWSplitSession *rses, GWBUF *querybuf,
qc_query_is_type(type, QUERY_TYPE_PREPARE_STMT)) qc_query_is_type(type, QUERY_TYPE_PREPARE_STMT))
{ {
gwbuf_set_type(querybuf, GWBUF_TYPE_COLLECT_RESULT); gwbuf_set_type(querybuf, GWBUF_TYPE_COLLECT_RESULT);
rses->ps_manager.store(querybuf, id); ps_manager.store(querybuf, id);
} }
MXS_INFO("Session write, routing to all servers."); MXS_INFO("Session write, routing to all servers.");
for (SRWBackendList::iterator it = rses->backends.begin(); for (auto it = backends.begin(); it != backends.end(); it++)
it != rses->backends.end(); it++)
{ {
SRWBackend& backend = *it; SRWBackend& backend = *it;
@ -362,7 +351,7 @@ bool route_session_write(RWSplitSession *rses, GWBUF *querybuf,
if (expecting_response) if (expecting_response)
{ {
rses->expected_responses++; expected_responses++;
} }
MXS_INFO("Route query to %s \t%s", MXS_INFO("Route query to %s \t%s",
@ -376,8 +365,7 @@ bool route_session_write(RWSplitSession *rses, GWBUF *querybuf,
} }
} }
if (rses->rses_config.max_sescmd_history > 0 && if (rses_config.max_sescmd_history > 0 && sescmd_list.size() >= rses_config.max_sescmd_history)
rses->sescmd_list.size() >= rses->rses_config.max_sescmd_history)
{ {
static bool warn_history_exceeded = true; static bool warn_history_exceeded = true;
if (warn_history_exceeded) if (warn_history_exceeded)
@ -389,40 +377,40 @@ bool route_session_write(RWSplitSession *rses, GWBUF *querybuf,
"command history, add `disable_sescmd_history=true` to " "command history, add `disable_sescmd_history=true` to "
"service '%s'. To increase the limit (currently %lu), add " "service '%s'. To increase the limit (currently %lu), add "
"`max_sescmd_history` to the same service and increase the value.", "`max_sescmd_history` to the same service and increase the value.",
rses->router->service()->name, rses->rses_config.max_sescmd_history); router->service()->name, rses_config.max_sescmd_history);
warn_history_exceeded = false; warn_history_exceeded = false;
} }
rses->rses_config.disable_sescmd_history = true; rses_config.disable_sescmd_history = true;
rses->rses_config.max_sescmd_history = 0; rses_config.max_sescmd_history = 0;
rses->sescmd_list.clear(); sescmd_list.clear();
} }
if (rses->rses_config.disable_sescmd_history) if (rses_config.disable_sescmd_history)
{ {
/** Prune stored responses */ /** Prune stored responses */
ResponseMap::iterator it = rses->sescmd_responses.lower_bound(lowest_pos); ResponseMap::iterator it = sescmd_responses.lower_bound(lowest_pos);
if (it != rses->sescmd_responses.end()) if (it != sescmd_responses.end())
{ {
rses->sescmd_responses.erase(rses->sescmd_responses.begin(), it); sescmd_responses.erase(sescmd_responses.begin(), it);
} }
} }
else else
{ {
purge_history(rses, sescmd); purge_history(sescmd);
rses->sescmd_list.push_back(sescmd); sescmd_list.push_back(sescmd);
} }
if (nsucc) if (nsucc)
{ {
rses->sent_sescmd = id; sent_sescmd = id;
if (!expecting_response) if (!expecting_response)
{ {
/** The command doesn't generate a response so we increment the /** The command doesn't generate a response so we increment the
* completed session command count */ * completed session command count */
rses->recv_sescmd++; recv_sescmd++;
} }
} }
@ -439,11 +427,11 @@ static inline bool rpl_lag_is_ok(SRWBackend& backend, int max_rlag)
backend->server()->rlag <= max_rlag); backend->server()->rlag <= max_rlag);
} }
SRWBackend get_hinted_backend(RWSplitSession *rses, char *name) SRWBackend RWSplitSession::get_hinted_backend(char *name)
{ {
SRWBackend rval; SRWBackend rval;
for (auto it = rses->backends.begin(); it != rses->backends.end(); it++) for (auto it = backends.begin(); it != backends.end(); it++)
{ {
auto& backend = *it; auto& backend = *it;
@ -459,12 +447,12 @@ SRWBackend get_hinted_backend(RWSplitSession *rses, char *name)
return rval; return rval;
} }
SRWBackend get_slave_backend(RWSplitSession *rses, int max_rlag) SRWBackend RWSplitSession::get_slave_backend(int max_rlag)
{ {
SRWBackend rval; SRWBackend rval;
auto counts = get_slave_counts(rses->backends, rses->current_master); auto counts = get_slave_counts(backends, current_master);
for (auto it = rses->backends.begin(); it != rses->backends.end(); it++) for (auto it = backends.begin(); it != backends.end(); it++)
{ {
auto& backend = *it; auto& backend = *it;
@ -474,15 +462,15 @@ SRWBackend get_slave_backend(RWSplitSession *rses, int max_rlag)
if (!rval) if (!rval)
{ {
// No previous candidate, accept any valid server (includes master) // No previous candidate, accept any valid server (includes master)
if ((backend->is_master() && backend == rses->current_master) || if ((backend->is_master() && backend == current_master) ||
backend->is_slave()) backend->is_slave())
{ {
rval = backend; rval = backend;
} }
} }
else if (backend->in_use() || counts.second < rses->router->max_slave_count()) else if (backend->in_use() || counts.second < router->max_slave_count())
{ {
if (!rses->rses_config.master_accept_reads && rval->is_master()) if (!rses_config.master_accept_reads && rval->is_master())
{ {
// Pick slaves over masters with master_accept_reads=false // Pick slaves over masters with master_accept_reads=false
rval = backend; rval = backend;
@ -490,7 +478,7 @@ SRWBackend get_slave_backend(RWSplitSession *rses, int max_rlag)
else else
{ {
// Compare the two servers and pick the best one // Compare the two servers and pick the best one
rval = compare_backends(rval, backend, rses->rses_config.slave_selection_criteria); rval = compare_backends(rval, backend, rses_config.slave_selection_criteria);
} }
} }
} }
@ -499,11 +487,11 @@ SRWBackend get_slave_backend(RWSplitSession *rses, int max_rlag)
return rval; return rval;
} }
SRWBackend get_master_backend(RWSplitSession *rses) SRWBackend RWSplitSession::get_master_backend()
{ {
SRWBackend rval; SRWBackend rval;
/** get root master from available servers */ /** get root master from available servers */
SRWBackend master = get_root_master(rses->backends); SRWBackend master = get_root_master(backends);
if (master) if (master)
{ {
@ -540,16 +528,14 @@ SRWBackend get_master_backend(RWSplitSession *rses)
* *
* @return True if a backend was found * @return True if a backend was found
*/ */
SRWBackend get_target_backend(RWSplitSession *rses, backend_type_t btype, SRWBackend RWSplitSession::get_target_backend(backend_type_t btype,
char *name, int max_rlag) char *name, int max_rlag)
{ {
CHK_CLIENT_RSES(rses); /** Check whether using target_node as target SLAVE */
if (target_node && session_trx_is_read_only(client_dcb->session))
/** Check whether using rses->target_node as target SLAVE */
if (rses->target_node && session_trx_is_read_only(rses->client_dcb->session))
{ {
MXS_DEBUG("In READ ONLY transaction, using server '%s'", rses->target_node->name()); MXS_DEBUG("In READ ONLY transaction, using server '%s'", target_node->name());
return rses->target_node; return target_node;
} }
SRWBackend rval; SRWBackend rval;
@ -558,16 +544,16 @@ SRWBackend get_target_backend(RWSplitSession *rses, backend_type_t btype,
{ {
ss_dassert(btype != BE_MASTER); ss_dassert(btype != BE_MASTER);
btype = BE_SLAVE; btype = BE_SLAVE;
rval = get_hinted_backend(rses, name); rval = get_hinted_backend(name);
} }
else if (btype == BE_SLAVE) else if (btype == BE_SLAVE)
{ {
rval = get_slave_backend(rses, max_rlag); rval = get_slave_backend(max_rlag);
} }
else if (btype == BE_MASTER) else if (btype == BE_MASTER)
{ {
rval = get_master_backend(rses); rval = get_master_backend();
} }
return rval; return rval;
@ -585,8 +571,7 @@ SRWBackend get_target_backend(RWSplitSession *rses, backend_type_t btype,
* *
* @return bool - true if succeeded, false otherwise * @return bool - true if succeeded, false otherwise
*/ */
SRWBackend handle_hinted_target(RWSplitSession *rses, GWBUF *querybuf, SRWBackend RWSplitSession::handle_hinted_target(GWBUF *querybuf, route_target_t route_target)
route_target_t route_target)
{ {
char *named_server = NULL; char *named_server = NULL;
int rlag_max = MAX_RLAG_UNDEFINED; int rlag_max = MAX_RLAG_UNDEFINED;
@ -622,7 +607,7 @@ SRWBackend handle_hinted_target(RWSplitSession *rses, GWBUF *querybuf,
if (rlag_max == MAX_RLAG_UNDEFINED) /*< no rlag max hint, use config */ if (rlag_max == MAX_RLAG_UNDEFINED) /*< no rlag max hint, use config */
{ {
rlag_max = rses_get_max_replication_lag(rses); rlag_max = get_max_replication_lag();
} }
/** target may be master or slave */ /** target may be master or slave */
@ -632,7 +617,7 @@ SRWBackend handle_hinted_target(RWSplitSession *rses, GWBUF *querybuf,
* Search backend server by name or replication lag. * Search backend server by name or replication lag.
* If it fails, then try to find valid slave or master. * If it fails, then try to find valid slave or master.
*/ */
SRWBackend target = get_target_backend(rses, btype, named_server, rlag_max); SRWBackend target = get_target_backend(btype, named_server, rlag_max);
if (!target) if (!target)
{ {
@ -654,29 +639,25 @@ SRWBackend handle_hinted_target(RWSplitSession *rses, GWBUF *querybuf,
} }
/** /**
* @brief Handle slave is the target * Handle slave target type
* *
* One of the possible types of handling required when a request is routed * @param cmd Command being executed
* @param stmt_id Prepared statement ID
* *
* @param inst Router instance * @return The target backend if one was found
* @param ses Router session
* @param target_dcb DCB for the target server
*
* @return bool - true if succeeded, false otherwise
*/ */
SRWBackend handle_slave_is_target(RWSplit *inst, RWSplitSession *rses, SRWBackend RWSplitSession::handle_slave_is_target(uint8_t cmd, uint32_t stmt_id)
uint8_t cmd, uint32_t stmt_id)
{ {
int rlag_max = rses_get_max_replication_lag(rses); int rlag_max = get_max_replication_lag();
SRWBackend target; SRWBackend target;
if (cmd == MXS_COM_STMT_FETCH) if (cmd == MXS_COM_STMT_FETCH)
{ {
/** The COM_STMT_FETCH must be executed on the same server as the /** The COM_STMT_FETCH must be executed on the same server as the
* COM_STMT_EXECUTE was executed on */ * COM_STMT_EXECUTE was executed on */
ExecMap::iterator it = rses->exec_map.find(stmt_id); ExecMap::iterator it = exec_map.find(stmt_id);
if (it != rses->exec_map.end()) if (it != exec_map.end())
{ {
target = it->second; target = it->second;
MXS_INFO("COM_STMT_FETCH on %s", target->uri()); MXS_INFO("COM_STMT_FETCH on %s", target->uri());
@ -689,12 +670,12 @@ SRWBackend handle_slave_is_target(RWSplit *inst, RWSplitSession *rses,
if (!target) if (!target)
{ {
target = get_target_backend(rses, BE_SLAVE, NULL, rlag_max); target = get_target_backend(BE_SLAVE, NULL, rlag_max);
} }
if (target) if (target)
{ {
atomic_add_uint64(&inst->stats().n_slave, 1); atomic_add_uint64(&router->stats().n_slave, 1);
} }
else else
{ {
@ -706,11 +687,10 @@ SRWBackend handle_slave_is_target(RWSplit *inst, RWSplitSession *rses,
/** /**
* @brief Log master write failure * @brief Log master write failure
*
* @param rses Router session
*/ */
static void log_master_routing_failure(RWSplitSession *rses, bool found, void RWSplitSession::log_master_routing_failure(bool found,
SRWBackend& old_master, SRWBackend& curr_master) SRWBackend& old_master,
SRWBackend& curr_master)
{ {
/** Both backends should either be empty, not connected or the DCB should /** Both backends should either be empty, not connected or the DCB should
* be a backend (the last check is slightly redundant). */ * be a backend (the last check is slightly redundant). */
@ -725,7 +705,7 @@ static void log_master_routing_failure(RWSplitSession *rses, bool found,
else if (old_master && curr_master && old_master->in_use()) else if (old_master && curr_master && old_master->in_use())
{ {
/** We found a master but it's not the same connection */ /** We found a master but it's not the same connection */
ss_dassert(!rses->rses_config.master_reconnection); ss_dassert(!rses_config.master_reconnection);
ss_dassert(old_master != curr_master); ss_dassert(old_master != curr_master);
sprintf(errmsg, "Master server changed from '%s' to '%s'", sprintf(errmsg, "Master server changed from '%s' to '%s'",
old_master->name(), curr_master->name()); old_master->name(), curr_master->name());
@ -741,7 +721,7 @@ static void log_master_routing_failure(RWSplitSession *rses, bool found,
else else
{ {
/** We never had a master connection, the session must be in read-only mode */ /** We never had a master connection, the session must be in read-only mode */
if (rses->rses_config.master_failure_mode != RW_FAIL_INSTANTLY) if (rses_config.master_failure_mode != RW_FAIL_INSTANTLY)
{ {
sprintf(errmsg, "Session is in read-only mode because it was created " sprintf(errmsg, "Session is in read-only mode because it was created "
"when no master was available"); "when no master was available");
@ -756,28 +736,28 @@ static void log_master_routing_failure(RWSplitSession *rses, bool found,
} }
MXS_WARNING("[%s] Write query received from %s@%s. %s. Closing client connection.", MXS_WARNING("[%s] Write query received from %s@%s. %s. Closing client connection.",
rses->router->service()->name, rses->client_dcb->user, router->service()->name, client_dcb->user,
rses->client_dcb->remote, errmsg); client_dcb->remote, errmsg);
} }
bool should_replace_master(RWSplitSession *rses, SRWBackend& target) bool RWSplitSession::should_replace_master(SRWBackend& target)
{ {
return rses->rses_config.master_reconnection && return rses_config.master_reconnection &&
// We have a target server and it's not the current master // We have a target server and it's not the current master
target && target != rses->current_master && target && target != current_master &&
// We are not inside a transaction (also checks for autocommit=1) // We are not inside a transaction (also checks for autocommit=1)
!session_trx_is_active(rses->client_dcb->session) && !session_trx_is_active(client_dcb->session) &&
// We are not locked to the old master // We are not locked to the old master
!locked_to_master(rses); !locked_to_master();
} }
void replace_master(RWSplitSession *rses, SRWBackend& target) void RWSplitSession::replace_master(SRWBackend& target)
{ {
rses->current_master = target; current_master = target;
// As the master has changed, we can reset the temporary table information // As the master has changed, we can reset the temporary table information
rses->have_tmp_tables = false; have_tmp_tables = false;
rses->temp_tables.clear(); temp_tables.clear();
} }
/** /**
@ -791,38 +771,37 @@ void replace_master(RWSplitSession *rses, SRWBackend& target)
* *
* @return bool - true if succeeded, false otherwise * @return bool - true if succeeded, false otherwise
*/ */
bool handle_master_is_target(RWSplit *inst, RWSplitSession *rses, bool RWSplitSession::handle_master_is_target(SRWBackend* dest)
SRWBackend* dest)
{ {
SRWBackend target = get_target_backend(rses, BE_MASTER, NULL, MAX_RLAG_UNDEFINED); SRWBackend target = get_target_backend(BE_MASTER, NULL, MAX_RLAG_UNDEFINED);
bool succp = true; bool succp = true;
if (should_replace_master(rses, target)) if (should_replace_master(target))
{ {
MXS_INFO("Replacing old master '%s' with new master '%s'", rses->current_master ? MXS_INFO("Replacing old master '%s' with new master '%s'", current_master ?
rses->current_master->name() : "<no previous master>", target->name()); current_master->name() : "<no previous master>", target->name());
replace_master(rses, target); replace_master(target);
} }
if (target && target == rses->current_master) if (target && target == current_master)
{ {
atomic_add_uint64(&inst->stats().n_master, 1); atomic_add_uint64(&router->stats().n_master, 1);
} }
else else
{ {
/** The original master is not available, we can't route the write */ /** The original master is not available, we can't route the write */
if (rses->rses_config.master_failure_mode == RW_ERROR_ON_WRITE) if (rses_config.master_failure_mode == RW_ERROR_ON_WRITE)
{ {
succp = send_readonly_error(rses->client_dcb); succp = send_readonly_error(client_dcb);
if (rses->current_master && rses->current_master->in_use()) if (current_master && current_master->in_use())
{ {
rses->current_master->close(); current_master->close();
} }
} }
else else
{ {
log_master_routing_failure(rses, succp, rses->current_master, target); log_master_routing_failure(succp, current_master, target);
succp = false; succp = false;
} }
} }
@ -852,7 +831,7 @@ static inline bool is_large_query(GWBUF* buf)
* @param origin origin send buffer * @param origin origin send buffer
* @return A new buffer contains wait statement and origin query * @return A new buffer contains wait statement and origin query
*/ */
GWBUF *add_prefix_wait_gtid(RWSplit *inst, RWSplitSession *rses, SERVER *server, GWBUF *origin) GWBUF* RWSplitSession::add_prefix_wait_gtid(SERVER *server, GWBUF *origin)
{ {
/** /**
@ -869,14 +848,14 @@ GWBUF *add_prefix_wait_gtid(RWSplit *inst, RWSplitSession *rses, SERVER *server,
GWBUF *rval; GWBUF *rval;
const char* wait_func = (server->server_type == SERVER_TYPE_MARIADB) ? const char* wait_func = (server->server_type == SERVER_TYPE_MARIADB) ?
MARIADB_WAIT_GTID_FUNC : MYSQL_WAIT_GTID_FUNC; MARIADB_WAIT_GTID_FUNC : MYSQL_WAIT_GTID_FUNC;
const char *gtid_wait_timeout = inst->config().causal_read_timeout.c_str(); const char *gtid_wait_timeout = router->config().causal_read_timeout.c_str();
const char *gtid_pos = rses->gtid_pos.c_str(); const char *gtid_position = gtid_pos.c_str();
/* Create a new buffer to store prefix sql */ /* Create a new buffer to store prefix sql */
size_t prefix_len = strlen(gtid_wait_stmt) + strlen(gtid_pos) + size_t prefix_len = strlen(gtid_wait_stmt) + strlen(gtid_position) +
strlen(gtid_wait_timeout) + strlen(wait_func); strlen(gtid_wait_timeout) + strlen(wait_func);
char prefix_sql[prefix_len]; char prefix_sql[prefix_len];
snprintf(prefix_sql, prefix_len, gtid_wait_stmt, wait_func, gtid_pos, gtid_wait_timeout); snprintf(prefix_sql, prefix_len, gtid_wait_stmt, wait_func, gtid_position, gtid_wait_timeout);
GWBUF *prefix_buff = modutil_create_query(prefix_sql); GWBUF *prefix_buff = modutil_create_query(prefix_sql);
/* Trim origin to sql, Append origin buffer to the prefix buffer */ /* Trim origin to sql, Append origin buffer to the prefix buffer */
@ -900,17 +879,15 @@ GWBUF *add_prefix_wait_gtid(RWSplit *inst, RWSplitSession *rses, SERVER *server,
* *
* @return True on success * @return True on success
*/ */
bool handle_got_target(RWSplit *inst, RWSplitSession *rses, bool RWSplitSession::handle_got_target(GWBUF* querybuf, SRWBackend& target, bool store)
GWBUF *querybuf, SRWBackend& target, bool store)
{ {
/** /**
* If the transaction is READ ONLY set forced_node to this backend. * If the transaction is READ ONLY set forced_node to this backend.
* This SLAVE backend will be used until the COMMIT is seen. * This SLAVE backend will be used until the COMMIT is seen.
*/ */
if (!rses->target_node && if (!target_node && session_trx_is_read_only(client_dcb->session))
session_trx_is_read_only(rses->client_dcb->session))
{ {
rses->target_node = target; target_node = target;
MXS_DEBUG("Setting forced_node SLAVE to %s within an opened READ ONLY transaction", MXS_DEBUG("Setting forced_node SLAVE to %s within an opened READ ONLY transaction",
target->name()); target->name());
} }
@ -922,16 +899,16 @@ bool handle_got_target(RWSplit *inst, RWSplitSession *rses,
ss_dassert(!target->have_session_commands()); ss_dassert(!target->have_session_commands());
mxs::Backend::response_type response = mxs::Backend::NO_RESPONSE; mxs::Backend::response_type response = mxs::Backend::NO_RESPONSE;
rses->wait_gtid_state = EXPECTING_NOTHING; wait_gtid_state = EXPECTING_NOTHING;
uint8_t cmd = mxs_mysql_get_command(querybuf); uint8_t cmd = mxs_mysql_get_command(querybuf);
GWBUF *send_buf = gwbuf_clone(querybuf); GWBUF *send_buf = gwbuf_clone(querybuf);
if (cmd == COM_QUERY && inst->config().enable_causal_read && rses->gtid_pos != "") if (cmd == COM_QUERY && router->config().enable_causal_read && gtid_pos != "")
{ {
send_buf = add_prefix_wait_gtid(inst, rses, target->server(), send_buf); send_buf = add_prefix_wait_gtid(target->server(), send_buf);
rses->wait_gtid_state = EXPECTING_WAIT_GTID_RESULT; wait_gtid_state = EXPECTING_WAIT_GTID_RESULT;
} }
if (rses->load_data_state != LOAD_DATA_ACTIVE && if (load_data_state != LOAD_DATA_ACTIVE &&
mxs_mysql_command_will_respond(cmd)) mxs_mysql_command_will_respond(cmd))
{ {
response = mxs::Backend::EXPECT_RESPONSE; response = mxs::Backend::EXPECT_RESPONSE;
@ -941,55 +918,55 @@ bool handle_got_target(RWSplit *inst, RWSplitSession *rses,
if (target->write(send_buf, response)) if (target->write(send_buf, response))
{ {
if (store && !session_store_stmt(rses->client_dcb->session, querybuf, target->server())) if (store && !session_store_stmt(client_dcb->session, querybuf, target->server()))
{ {
MXS_ERROR("Failed to store current statement, it won't be retried if it fails."); MXS_ERROR("Failed to store current statement, it won't be retried if it fails.");
} }
atomic_add_uint64(&inst->stats().n_queries, 1); atomic_add_uint64(&router->stats().n_queries, 1);
if (!rses->large_query && response == mxs::Backend::EXPECT_RESPONSE) if (!large_query && response == mxs::Backend::EXPECT_RESPONSE)
{ {
/** The server will reply to this command */ /** The server will reply to this command */
ss_dassert(target->get_reply_state() == REPLY_STATE_DONE); ss_dassert(target->get_reply_state() == REPLY_STATE_DONE);
target->set_reply_state(REPLY_STATE_START); target->set_reply_state(REPLY_STATE_START);
rses->expected_responses++; expected_responses++;
if (rses->load_data_state == LOAD_DATA_START) if (load_data_state == LOAD_DATA_START)
{ {
/** The first packet contains the actual query and the server /** The first packet contains the actual query and the server
* will respond to it */ * will respond to it */
rses->load_data_state = LOAD_DATA_ACTIVE; load_data_state = LOAD_DATA_ACTIVE;
} }
else if (rses->load_data_state == LOAD_DATA_END) else if (load_data_state == LOAD_DATA_END)
{ {
/** The final packet in a LOAD DATA LOCAL INFILE is an empty packet /** The final packet in a LOAD DATA LOCAL INFILE is an empty packet
* to which the server responds with an OK or an ERR packet */ * to which the server responds with an OK or an ERR packet */
ss_dassert(gwbuf_length(querybuf) == 4); ss_dassert(gwbuf_length(querybuf) == 4);
rses->load_data_state = LOAD_DATA_INACTIVE; load_data_state = LOAD_DATA_INACTIVE;
} }
} }
if ((rses->large_query = large_query)) if ((large_query = large_query))
{ {
/** Store the previous target as we're processing a multi-packet query */ /** Store the previous target as we're processing a multi-packet query */
rses->prev_target = target; prev_target = target;
} }
else else
{ {
/** Otherwise reset it so we know the query is complete */ /** Otherwise reset it so we know the query is complete */
rses->prev_target.reset(); prev_target.reset();
} }
/** /**
* If a READ ONLY transaction is ending set forced_node to NULL * If a READ ONLY transaction is ending set forced_node to NULL
*/ */
if (rses->target_node && if (target_node &&
session_trx_is_read_only(rses->client_dcb->session) && session_trx_is_read_only(client_dcb->session) &&
session_trx_is_ending(rses->client_dcb->session)) session_trx_is_ending(client_dcb->session))
{ {
MXS_DEBUG("An opened READ ONLY transaction ends: forced_node is set to NULL"); MXS_DEBUG("An opened READ ONLY transaction ends: forced_node is set to NULL");
rses->target_node.reset(); target_node.reset();
} }
return true; return true;
} }

19
server/modules/routing/readwritesplit/rwsplit_select_backends.cc
View File

@ -12,7 +12,6 @@
*/ */
#include "readwritesplit.hh" #include "readwritesplit.hh"
#include "rwsplit_internal.hh"
#include <stdio.h> #include <stdio.h>
#include <strings.h> #include <strings.h>
@ -296,23 +295,23 @@ std::pair<int, int> get_slave_counts(SRWBackendList& backends, SRWBackend& maste
* *
* @return True if session can continue * @return True if session can continue
*/ */
bool select_connect_backend_servers(RWSplit *inst, MXS_SESSION *session, bool RWSplit::select_connect_backend_servers(MXS_SESSION *session,
SRWBackendList& backends, SRWBackendList& backends,
SRWBackend& current_master, SRWBackend& current_master,
mxs::SessionCommandList* sescmd_list, SessionCommandList* sescmd_list,
int* expected_responses, int* expected_responses,
connection_type type) connection_type type)
{ {
SRWBackend master = get_root_master(backends); SRWBackend master = get_root_master(backends);
if (!master && inst->config().master_failure_mode == RW_FAIL_INSTANTLY) if (!master && config().master_failure_mode == RW_FAIL_INSTANTLY)
{ {
MXS_ERROR("Couldn't find suitable Master from %lu candidates.", backends.size()); MXS_ERROR("Couldn't find suitable Master from %lu candidates.", backends.size());
return false; return false;
} }
/** Check slave selection criteria and set compare function */ /** Check slave selection criteria and set compare function */
select_criteria_t select_criteria = inst->config().slave_selection_criteria; select_criteria_t select_criteria = config().slave_selection_criteria;
int (*cmpfun)(const SRWBackend&, const SRWBackend&) = criteria_cmpfun[select_criteria]; int (*cmpfun)(const SRWBackend&, const SRWBackend&) = criteria_cmpfun[select_criteria];
ss_dassert(cmpfun); ss_dassert(cmpfun);
@ -342,7 +341,7 @@ bool select_connect_backend_servers(RWSplit *inst, MXS_SESSION *session,
auto counts = get_slave_counts(backends, master); auto counts = get_slave_counts(backends, master);
int slaves_found = counts.first; int slaves_found = counts.first;
int slaves_connected = counts.second; int slaves_connected = counts.second;
int max_nslaves = inst->max_slave_count(); int max_nslaves = max_slave_count();
ss_dassert(slaves_connected < max_nslaves || max_nslaves == 0); ss_dassert(slaves_connected < max_nslaves || max_nslaves == 0);

26
server/modules/routing/readwritesplit/rwsplit_session_cmd.cc
View File

@ -12,7 +12,7 @@
*/ */
#include "readwritesplit.hh" #include "readwritesplit.hh"
#include "rwsplit_internal.hh" #include "rwsplitsession.hh"
#include <stdio.h> #include <stdio.h>
#include <strings.h> #include <strings.h>
@ -67,7 +67,7 @@ static void discard_if_response_differs(SRWBackend backend, uint8_t master_cmd,
} }
} }
void process_sescmd_response(RWSplitSession* rses, SRWBackend& backend, GWBUF** ppPacket) void RWSplitSession::process_sescmd_response(SRWBackend& backend, GWBUF** ppPacket)
{ {
if (backend->have_session_commands()) if (backend->have_session_commands())
{ {
@ -89,18 +89,18 @@ void process_sescmd_response(RWSplitSession* rses, SRWBackend& backend, GWBUF**
backend->add_ps_handle(id, resp.id); backend->add_ps_handle(id, resp.id);
} }
if (rses->recv_sescmd < rses->sent_sescmd && id == rses->recv_sescmd + 1) if (recv_sescmd < sent_sescmd && id == recv_sescmd + 1)
{ {
if (!rses->current_master || !rses->current_master->in_use() || // Session doesn't have a master if (!current_master || !current_master->in_use() || // Session doesn't have a master
rses->current_master == backend) // This is the master's response current_master == backend) // This is the master's response
{ {
/** First reply to this session command, route it to the client */ /** First reply to this session command, route it to the client */
++rses->recv_sescmd; ++recv_sescmd;
discard = false; discard = false;
/** Store the master's response so that the slave responses can /** Store the master's response so that the slave responses can
* be compared to it */ * be compared to it */
rses->sescmd_responses[id] = cmd; sescmd_responses[id] = cmd;
if (cmd == MYSQL_REPLY_ERR) if (cmd == MYSQL_REPLY_ERR)
{ {
@ -111,28 +111,28 @@ void process_sescmd_response(RWSplitSession* rses, SRWBackend& backend, GWBUF**
{ {
/** Map the returned response to the internal ID */ /** Map the returned response to the internal ID */
MXS_INFO("PS ID %u maps to internal ID %lu", resp.id, id); MXS_INFO("PS ID %u maps to internal ID %lu", resp.id, id);
rses->ps_handles[resp.id] = id; ps_handles[resp.id] = id;
} }
// Discard any slave connections that did not return the same result // Discard any slave connections that did not return the same result
for (SlaveResponseList::iterator it = rses->slave_responses.begin(); for (SlaveResponseList::iterator it = slave_responses.begin();
it != rses->slave_responses.end(); it++) it != slave_responses.end(); it++)
{ {
discard_if_response_differs(it->first, cmd, it->second); discard_if_response_differs(it->first, cmd, it->second);
} }
rses->slave_responses.clear(); slave_responses.clear();
} }
else else
{ {
/** Record slave command so that the response can be validated /** Record slave command so that the response can be validated
* against the master's response when it arrives. */ * against the master's response when it arrives. */
rses->slave_responses.push_back(std::make_pair(backend, cmd)); slave_responses.push_back(std::make_pair(backend, cmd));
} }
} }
else else
{ {
discard_if_response_differs(backend, rses->sescmd_responses[id], cmd); discard_if_response_differs(backend, sescmd_responses[id], cmd);
} }
if (discard) if (discard)

5
server/modules/routing/readwritesplit/rwsplitsession.cc
View File

@ -12,7 +12,6 @@
*/ */
#include "rwsplitsession.hh" #include "rwsplitsession.hh"
#include "rwsplit_internal.hh"
#include "routeinfo.hh" #include "routeinfo.hh"
#include <cmath> #include <cmath>
@ -69,8 +68,8 @@ RWSplitSession* RWSplitSession::create(RWSplit* router, MXS_SESSION* session)
SRWBackend master; SRWBackend master;
if (select_connect_backend_servers(router, session, backends, master, if (router->select_connect_backend_servers(session, backends, master, NULL,
NULL, NULL, connection_type::ALL)) NULL, connection_type::ALL))
{ {
if ((rses = new RWSplitSession(router, session, backends, master))) if ((rses = new RWSplitSession(router, session, backends, master)))
{ {

48
server/modules/routing/readwritesplit/rwsplitsession.hh
View File

@ -15,6 +15,7 @@
#include "readwritesplit.hh" #include "readwritesplit.hh"
#include "rwsplit_ps.hh" #include "rwsplit_ps.hh"
#include "rwbackend.hh" #include "rwbackend.hh"
#include "routeinfo.hh"
#include <string> #include <string>
@ -127,6 +128,53 @@ public:
private: private:
RWSplitSession(RWSplit* instance, MXS_SESSION* session, RWSplitSession(RWSplit* instance, MXS_SESSION* session,
const mxs::SRWBackendList& backends, const mxs::SRWBackend& master); const mxs::SRWBackendList& backends, const mxs::SRWBackend& master);
void process_sescmd_response(mxs::SRWBackend& backend, GWBUF** ppPacket);
void purge_history(mxs::SSessionCommand& sescmd);
bool route_session_write(GWBUF *querybuf, uint8_t command, uint32_t type);
bool route_single_stmt(GWBUF *querybuf, const RouteInfo& info);
bool route_stored_query();
bool reroute_stored_statement(const mxs::SRWBackend& old, GWBUF *stored);
mxs::SRWBackend get_hinted_backend(char *name);
mxs::SRWBackend get_slave_backend(int max_rlag);
mxs::SRWBackend get_master_backend();
mxs::SRWBackend get_target_backend(backend_type_t btype, char *name, int max_rlag);
bool handle_target_is_all(route_target_t route_target, GWBUF *querybuf,
int packet_type, uint32_t qtype);
mxs::SRWBackend handle_hinted_target(GWBUF *querybuf, route_target_t route_target);
mxs::SRWBackend handle_slave_is_target(uint8_t cmd, uint32_t stmt_id);
bool handle_master_is_target(mxs::SRWBackend* dest);
bool handle_got_target(GWBUF* querybuf, mxs::SRWBackend& target, bool store);
void handle_connection_keepalive(mxs::SRWBackend& target);
bool prepare_target(mxs::SRWBackend& target, route_target_t route_target);
bool should_replace_master(mxs::SRWBackend& target);
void replace_master(mxs::SRWBackend& target);
void log_master_routing_failure(bool found, mxs::SRWBackend& old_master,
mxs::SRWBackend& curr_master);
GWBUF* add_prefix_wait_gtid(SERVER *server, GWBUF *origin);
void correct_packet_sequence(GWBUF *buffer);
GWBUF* discard_master_wait_gtid_result(GWBUF *buffer);
int get_max_replication_lag();
mxs::SRWBackend& get_backend_from_dcb(DCB *dcb);
void handle_error_reply_client(DCB *backend_dcb, GWBUF *errmsg);
bool handle_error_new_connection(DCB *backend_dcb, GWBUF *errmsg);
/**
* Check if the session is locked to the master
*
* @return Whether the session is locked to the master
*/
inline bool locked_to_master() const
{
return large_query || (current_master && target_node == current_master);
}
}; };
/** /**