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

Take SessionCommand into use in readwritesplit

Browse Source 
Readwritesplit now uses the SessionCommand class as a "master list" of
executed session commands. This allows the session commands to be easily
copied over to slaves that are taken into use after session commands have
already been executed.

Currently, the code doesn't execute the session command history when a
mid-session reconnection occurs. A method to cleanly copy the session
commands needs to be exposed by the Backend class.
This commit is contained in:
Markus Mäkelä
2017-06-15 20:42:20 +03:00
parent b48bb4fc5e
commit 0c45eb5d17
6 changed files with 100 additions and 828 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -775,6 +775,7 @@ static MXS_ROUTER_SESSION *newSession(MXS_ROUTER *router_inst, MXS_SESSION *sess
client_rses->load_data_state = LOAD_DATA_INACTIVE;
client_rses->sent_sescmd = 0;
client_rses->recv_sescmd = 0;
client_rses->sescmd_count = 1; // Needs to be a positive number to work
memcpy(&client_rses->rses_config, &router->rwsplit_config, sizeof(client_rses->rses_config));
int router_nservers = router->service->n_dbref;
@ -977,7 +978,7 @@ static void diagnostics(MXS_ROUTER *instance, DCB *dcb)
router->rwsplit_config.strict_multi_stmt ? "true" : "false");
dcb_printf(dcb, "\tdisable_sescmd_history: %s\n",
router->rwsplit_config.disable_sescmd_history ? "true" : "false");
dcb_printf(dcb, "\tmax_sescmd_history: %d\n",
dcb_printf(dcb, "\tmax_sescmd_history: %lu\n",
router->rwsplit_config.max_sescmd_history);
dcb_printf(dcb, "\tmaster_accept_reads: %s\n",
router->rwsplit_config.master_accept_reads ? "true" : "false");

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

@ -21,6 +21,7 @@
#include <maxscale/cppdefs.hh>
#include <tr1/unordered_set>
#include <map>
#include <string>
#include <maxscale/dcb.h>
@ -139,7 +140,7 @@ struct rwsplit_config_t
int max_slave_replication_lag; /**< Maximum replication lag */
mxs_target_t use_sql_variables_in; /**< Whether to send user variables
* to master or all nodes */
int max_sescmd_history; /**< Maximum amount of session commands to store */
uint64_t max_sescmd_history; /**< Maximum amount of session commands to store */
bool disable_sescmd_history; /**< Disable session command history */
bool master_accept_reads; /**< Use master for reads */
bool strict_multi_stmt; /**< Force non-multistatement queries to be routed
@ -185,6 +186,7 @@ typedef std::tr1::shared_ptr<RWBackend> SRWBackend;
typedef std::list<SRWBackend> SRWBackendList;
typedef std::tr1::unordered_set<std::string> TableSet;
typedef std::map<uint64_t, uint8_t> ResponseMap;
/**
* The client session structure used within this router.
@ -198,18 +200,20 @@ struct ROUTER_CLIENT_SES
SRWBackend target_node; /**< The currently locked target node */
rwsplit_config_t rses_config; /**< copied config info from router instance */
int rses_nbackends;
int rses_nsescmd; /**< Number of executed session commands */
enum ld_state load_data_state; /**< Current load data state */
bool have_tmp_tables;
uint64_t rses_load_data_sent; /**< How much data has been sent */
DCB* client_dcb;
uint64_t pos_generator;
uint64_t sescmd_count;
int expected_responses; /**< Number of expected responses to the current query */
GWBUF* query_queue; /**< Queued commands waiting to be executed */
struct ROUTER_INSTANCE *router; /**< The router instance */
struct ROUTER_CLIENT_SES *next;
TableSet temp_tables; /**< Set of temporary tables */
mxs::SessionCommandList sescmd_list; /**< List of executed session commands */
ResponseMap sescmd_responses; /**< Response to each session command */
uint64_t sent_sescmd; /**< ID of the last sent session command*/
uint64_t recv_sescmd; /**< ID of the most recently completed session command */
skygw_chk_t rses_chk_tail;
};

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

@ -43,7 +43,7 @@ bool handle_target_is_all(route_target_t route_target,
GWBUF *querybuf, int packet_type, uint32_t qtype);
uint8_t determine_packet_type(GWBUF *querybuf, bool *non_empty_packet);
void log_transaction_status(ROUTER_CLIENT_SES *rses, GWBUF *querybuf, uint32_t qtype);
bool is_packet_a_one_way_message(int packet_type);
bool command_will_respond(uint8_t packet_type);
bool is_packet_a_query(int packet_type);
bool send_readonly_error(DCB *dcb);
@ -75,10 +75,7 @@ bool handle_master_is_target(ROUTER_INSTANCE *inst, ROUTER_CLIENT_SES *rses,
SRWBackend& target);
bool handle_got_target(ROUTER_INSTANCE *inst, ROUTER_CLIENT_SES *rses,
GWBUF *querybuf, SRWBackend& target, bool store);
bool route_session_write(ROUTER_CLIENT_SES *router_cli_ses,
GWBUF *querybuf, ROUTER_INSTANCE *inst,
int packet_type,
uint32_t qtype);
bool route_session_write(ROUTER_CLIENT_SES *rses, GWBUF *querybuf, uint8_t command);
void process_sescmd_response(ROUTER_CLIENT_SES* rses, SRWBackend& bref,
GWBUF** ppPacket, bool* reconnect);

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

@ -118,11 +118,11 @@ is_packet_a_query(int packet_type)
* @param packet_type Type of packet (integer)
* @return bool indicating whether packet contains a one way message
*/
bool
is_packet_a_one_way_message(int packet_type)
bool command_will_respond(uint8_t packet_type)
{
return (packet_type == MYSQL_COM_STMT_SEND_LONG_DATA ||
packet_type == MYSQL_COM_QUIT || packet_type == MYSQL_COM_STMT_CLOSE);
return packet_type != MYSQL_COM_STMT_SEND_LONG_DATA &&
packet_type != MYSQL_COM_QUIT &&
packet_type != MYSQL_COM_STMT_CLOSE;
}
/*
@ -234,7 +234,7 @@ bool handle_target_is_all(route_target_t route_target, ROUTER_INSTANCE *inst,
MXS_FREE(query_str);
MXS_FREE(qtype_str);
}
else if (route_session_write(rses, gwbuf_clone(querybuf), inst, packet_type, qtype))
else if (route_session_write(rses, gwbuf_clone(querybuf), packet_type))
{
result = true;
@ -281,194 +281,24 @@ void closed_session_reply(GWBUF *querybuf)
* @param scur Session cursor
* @param bref Router session data for a backend server
*/
void check_session_command_reply(GWBUF *writebuf, sescmd_cursor_t *scur, backend_ref_t *bref)
void check_session_command_reply(GWBUF *writebuf, SRWBackend bref)
{
if (MXS_LOG_PRIORITY_IS_ENABLED(LOG_ERR) &&
MYSQL_IS_ERROR_PACKET(((uint8_t *)GWBUF_DATA(writebuf))))
if (MYSQL_IS_ERROR_PACKET(((uint8_t *)GWBUF_DATA(writebuf))))
{
uint8_t *buf = (uint8_t *)GWBUF_DATA((scur->scmd_cur_cmd->my_sescmd_buf));
uint8_t *replybuf = (uint8_t *)GWBUF_DATA(writebuf);
size_t len = MYSQL_GET_PAYLOAD_LEN(buf);
size_t replylen = MYSQL_GET_PAYLOAD_LEN(replybuf);
char *err = strndup(&((char *)replybuf)[8], 5);
char *replystr = strndup(&((char *)replybuf)[13], replylen - 4 - 5);
ss_dassert(len + 4 == GWBUF_LENGTH(scur->scmd_cur_cmd->my_sescmd_buf));
size_t replylen = MYSQL_GET_PAYLOAD_LEN(GWBUF_DATA(writebuf));
char replybuf[replylen];
gwbuf_copy_data(writebuf, 0, gwbuf_length(writebuf), (uint8_t*)replybuf);
std::string err;
std::string msg;
err.append(replybuf + 8, 5);
msg.append(replybuf + 13, replylen - 4 - 5);
MXS_ERROR("Failed to execute session command in [%s]:%d. Error was: %s %s",
bref->ref->server->name,
bref->ref->server->port, err, replystr);
MXS_FREE(err);
MXS_FREE(replystr);
bref->server()->name, bref->server()->port,
err.c_str(), msg.c_str());
}
}
/**
* @brief If session command cursor is passive, sends the command to backend for
* execution.
*
* Returns true if command was sent or added successfully to the queue.
* Returns false if command sending failed or if there are no pending session
* commands.
*
* Router session must be locked.
*
* @param backend_ref Router session backend database data
* @return bool - true for success, false for failure
*/
/*
* Uses MySQL specific values in the large switch statement, although it
* may be possible to generalize them.
*/
bool execute_sescmd_in_backend(backend_ref_t *backend_ref)
{
ss_dassert(backend_ref);
CHK_BACKEND_REF(backend_ref);
bool succp = false;
if (BREF_IS_CLOSED(backend_ref))
{
return succp;
}
DCB *dcb = backend_ref->bref_dcb;
CHK_DCB(dcb);
/**
* Get cursor pointer and copy of command buffer to cursor.
*/
sescmd_cursor_t *scur = &backend_ref->bref_sescmd_cur;
/** Return if there are no pending ses commands */
if (sescmd_cursor_get_command(scur) == NULL)
{
succp = true;
MXS_INFO("Cursor had no pending session commands.");
return succp;
}
if (!sescmd_cursor_is_active(scur))
{
/** Cursor is left active when function returns. */
sescmd_cursor_set_active(scur, true);
}
int rc = 0;
GWBUF *buf;
switch (scur->scmd_cur_cmd->my_sescmd_packet_type)
{
case MYSQL_COM_CHANGE_USER:
/** This makes it possible to handle replies correctly */
gwbuf_set_type(scur->scmd_cur_cmd->my_sescmd_buf, GWBUF_TYPE_SESCMD);
buf = sescmd_cursor_clone_querybuf(scur);
rc = dcb->func.auth(dcb, NULL, dcb->session, buf);
break;
case MYSQL_COM_INIT_DB:
{
/**
* Record database name and store to session.
*
* TODO: Do this in the client protocol module
*/
GWBUF *tmpbuf;
MYSQL_session* data;
unsigned int qlen;
mxs_mysql_set_current_db(dcb->session, "");
tmpbuf = scur->scmd_cur_cmd->my_sescmd_buf;
qlen = MYSQL_GET_PAYLOAD_LEN((unsigned char *) GWBUF_DATA(tmpbuf));
if (qlen)
{
--qlen; // The COM_INIT_DB byte
if (qlen > MYSQL_DATABASE_MAXLEN)
{
MXS_ERROR("Too long a database name received in COM_INIT_DB, "
"trailing data will be cut.");
qlen = MYSQL_DATABASE_MAXLEN;
}
char db[qlen + 1];
memcpy(db, (char*)GWBUF_DATA(tmpbuf) + 5, qlen);
db[qlen] = 0;
mxs_mysql_set_current_db(dcb->session, db);
}
}
/** Fallthrough */
case MYSQL_COM_QUERY:
default:
/**
* Mark session command buffer, it triggers writing
* MySQL command to protocol
*/
gwbuf_set_type(scur->scmd_cur_cmd->my_sescmd_buf, GWBUF_TYPE_SESCMD);
buf = sescmd_cursor_clone_querybuf(scur);
rc = dcb->func.write(dcb, buf);
break;
}
if (rc == 1)
{
succp = true;
ss_dassert(backend_ref->reply_state == REPLY_STATE_DONE);
LOG_RS(backend_ref, REPLY_STATE_START);
backend_ref->reply_state = REPLY_STATE_START;
}
return succp;
}
/*
* End of functions called from other router modules; start of functions that
* are internal to this module
*/
/**
* Get client DCB pointer of the router client session.
* This routine must be protected by Router client session lock.
*
* APPEARS TO NEVER BE USED!!
*
* @param rses Router client session pointer
*
* @return Pointer to client DCB
*/
static DCB *rses_get_client_dcb(ROUTER_CLIENT_SES *rses)
{
DCB *dcb = NULL;
int i;
for (i = 0; i < rses->rses_nbackends; i++)
{
if ((dcb = rses->rses_backend_ref[i].bref_dcb) != NULL &&
BREF_IS_IN_USE(&rses->rses_backend_ref[i]) && dcb->session != NULL &&
dcb->session->client_dcb != NULL)
{
return dcb->session->client_dcb;
}
}
return NULL;
}
/*
* The following are internal (directly or indirectly) to routing a statement
* and should be moved to rwsplit_route_cmd.c if the MySQL specific code can
* be removed.
*/
sescmd_cursor_t *backend_ref_get_sescmd_cursor(backend_ref_t *bref)
{
sescmd_cursor_t *scur;
CHK_BACKEND_REF(bref);
scur = &bref->bref_sescmd_cur;
CHK_SESCMD_CUR(scur);
return scur;
}
/**
* Send an error message to the client telling that the server is in read only mode
* @param dcb Client DCB

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

@ -230,212 +230,88 @@ bool route_single_stmt(ROUTER_INSTANCE *inst, ROUTER_CLIENT_SES *rses,
* backends being used, otherwise false.
*
*/
bool route_session_write(ROUTER_CLIENT_SES *router_cli_ses,
GWBUF *querybuf, ROUTER_INSTANCE *inst,
int packet_type,
uint32_t qtype)
bool route_session_write(ROUTER_CLIENT_SES *rses, GWBUF *querybuf, uint8_t command)
{
bool succp;
rses_property_t *prop;
backend_ref_t *backend_ref;
int i;
int max_nslaves;
int nbackends;
int nsucc;
/** The SessionCommand takes ownership of the buffer */
uint64_t id = rses->sescmd_count++;
mxs::SSessionCommand sescmd(new mxs::SessionCommand(querybuf, id));
bool expecting_response = command_will_respond(command);
int nsucc = 0;
uint64_t lowest_pos = id;
MXS_INFO("Session write, routing to all servers.");
/** Maximum number of slaves in this router client session */
max_nslaves =
rses_get_max_slavecount(router_cli_ses, router_cli_ses->rses_nbackends);
nsucc = 0;
nbackends = 0;
backend_ref = router_cli_ses->rses_backend_ref;
/**
* These are one-way messages and server doesn't respond to them.
* Therefore reply processing is unnecessary and session
* command property is not needed. It is just routed to all available
* backends.
*/
if (is_packet_a_one_way_message(packet_type))
for (SRWBackendList::iterator it = rses->backends.begin();
it != rses->backends.end(); it++)
{
int rc;
SRWBackend& bref = *it;
for (i = 0; i < router_cli_ses->rses_nbackends; i++)
if (bref->in_use())
{
DCB *dcb = backend_ref[i].bref_dcb;
bref->add_session_command(sescmd);
if (MXS_LOG_PRIORITY_IS_ENABLED(LOG_INFO) &&
BREF_IS_IN_USE((&backend_ref[i])))
uint64_t current_pos = bref->next_session_command()->get_position();
if (current_pos < lowest_pos)
{
MXS_INFO("Route query to %s \t[%s]:%d%s",
(SERVER_IS_MASTER(backend_ref[i].ref->server)
? "master" : "slave"),
backend_ref[i].ref->server->name,
backend_ref[i].ref->server->port,
(i + 1 == router_cli_ses->rses_nbackends ? " <" : " "));
lowest_pos = current_pos;
}
if (BREF_IS_IN_USE((&backend_ref[i])))
if (bref->execute_session_command())
{
nbackends += 1;
if ((rc = dcb->func.write(dcb, gwbuf_clone(querybuf))) == 1)
nsucc += 1;
if (expecting_response)
{
nsucc += 1;
bref->set_reply_state(REPLY_STATE_START);
rses->expected_responses++;
}
MXS_INFO("Route query to %s \t[%s]:%d",
SERVER_IS_MASTER(bref->server()) ? "master" : "slave",
bref->server()->name, bref->server()->port);
}
else
{
MXS_ERROR("Failed to execute session command in [%s]:%d",
bref->server()->name, bref->server()->port);
}
}
gwbuf_free(querybuf);
goto return_succp;
}
if (router_cli_ses->rses_nbackends <= 0)
{
MXS_INFO("Router session doesn't have any backends in use. Routing failed. <");
goto return_succp;
}
if (router_cli_ses->rses_config.max_sescmd_history > 0 &&
router_cli_ses->rses_nsescmd >=
router_cli_ses->rses_config.max_sescmd_history)
if (rses->rses_config.max_sescmd_history > 0 &&
rses->sescmd_count >= rses->rses_config.max_sescmd_history)
{
MXS_WARNING("Router session exceeded session command history limit. "
"Slave recovery is disabled and only slave servers with "
"consistent session state are used "
"for the duration of the session.");
router_cli_ses->rses_config.disable_sescmd_history = true;
router_cli_ses->rses_config.max_sescmd_history = 0;
rses->rses_config.disable_sescmd_history = true;
rses->rses_config.max_sescmd_history = 0;
rses->sescmd_list.clear();
}
if (router_cli_ses->rses_config.disable_sescmd_history)
if (rses->rses_config.disable_sescmd_history)
{
rses_property_t *prop, *tmp;
backend_ref_t *bref;
bool conflict;
/** Prune stored responses */
ResponseMap::iterator it = rses->sescmd_responses.lower_bound(lowest_pos);
prop = router_cli_ses->rses_properties[RSES_PROP_TYPE_SESCMD];
while (prop)
if (it != rses->sescmd_responses.end())
{
conflict = false;
for (i = 0; i < router_cli_ses->rses_nbackends; i++)
{
bref = &backend_ref[i];
if (BREF_IS_IN_USE(bref))
{
if (bref->bref_sescmd_cur.position <=
prop->rses_prop_data.sescmd.position + 1)
{
conflict = true;
break;
}
}
}
if (conflict)
{
break;
}
tmp = prop;
router_cli_ses->rses_properties[RSES_PROP_TYPE_SESCMD] = prop->rses_prop_next;
rses_property_done(tmp);
prop = router_cli_ses->rses_properties[RSES_PROP_TYPE_SESCMD];
rses->sescmd_responses.erase(rses->sescmd_responses.begin(), it);
}
}
/**
* Additional reference is created to querybuf to
* prevent it from being released before properties
* are cleaned up as a part of router sessionclean-up.
*/
if ((prop = rses_property_init(RSES_PROP_TYPE_SESCMD)) == NULL)
else
{
MXS_ERROR("Router session property initialization failed");
return false;
rses->sescmd_list.push_back(sescmd);
}
mysql_sescmd_init(prop, querybuf, packet_type, router_cli_ses);
/** Add sescmd property to router client session */
if (rses_property_add(router_cli_ses, prop) != 0)
if (nsucc)
{
MXS_ERROR("Session property addition failed.");
return false;
rses->sent_sescmd = id;
}
if (!router_cli_ses->rses_config.disable_sescmd_history)
{
/** The stored buffer points to the one in the session command
* which is freed in freeSession. */
uint64_t id = prop->rses_prop_data.sescmd.position;
router_cli_ses->sescmd_list.push_front(mxs::SSessionCommand(new mxs::SessionCommand(querybuf, id)));
}
for (i = 0; i < router_cli_ses->rses_nbackends; i++)
{
if (BREF_IS_IN_USE((&backend_ref[i])))
{
sescmd_cursor_t *scur;
nbackends += 1;
if (MXS_LOG_PRIORITY_IS_ENABLED(LOG_INFO))
{
MXS_INFO("Route query to %s \t[%s]:%d%s",
(SERVER_IS_MASTER(backend_ref[i].ref->server)
? "master" : "slave"),
backend_ref[i].ref->server->name,
backend_ref[i].ref->server->port,
(i + 1 == router_cli_ses->rses_nbackends ? " <" : " "));
}
scur = backend_ref_get_sescmd_cursor(&backend_ref[i]);
/**
* Add one waiter to backend reference.
*/
bref_set_state(get_bref_from_dcb(router_cli_ses, backend_ref[i].bref_dcb),
BREF_WAITING_RESULT);
/**
* Start execution if cursor is not already executing or this is the
* master server. Otherwise, cursor will execute pending commands
* when it completes the previous command.
*/
if (sescmd_cursor_is_active(scur) && &backend_ref[i] != router_cli_ses->rses_master_ref)
{
nsucc += 1;
MXS_INFO("Backend [%s]:%d already executing sescmd.",
backend_ref[i].ref->server->name,
backend_ref[i].ref->server->port);
}
else
{
if (execute_sescmd_in_backend(&backend_ref[i]))
{
router_cli_ses->expected_responses++;
nsucc += 1;
}
else
{
MXS_ERROR("Failed to execute session command in [%s]:%d",
backend_ref[i].ref->server->name,
backend_ref[i].ref->server->port);
}
}
}
}
atomic_add(&router_cli_ses->rses_nsescmd, 1);
return_succp:
/**
* Routing must succeed to all backends that are used.
* There must be at least one and at most max_nslaves+1 backends.
*/
succp = (nbackends > 0 && nsucc == nbackends && nbackends <= max_nslaves + 1);
return succp;
return nsucc;
}
/**
@ -1245,55 +1121,6 @@ handle_got_target(ROUTER_INSTANCE *inst, ROUTER_CLIENT_SES *rses,
}
}
/**
* @brief Add property to the router client session
*
* Add property to the router_client_ses structure's rses_properties
* array. The slot is determined by the type of property.
* In each slot there is a list of properties of similar type.
*
* Router client session must be locked.
*
* @param rses Router session
* @param prop Router session property to be added
*
* @return -1 on failure, 0 on success
*/
int rses_property_add(ROUTER_CLIENT_SES *rses, rses_property_t *prop)
{
if (rses == NULL)
{
MXS_ERROR("Router client session is NULL. (%s:%d)", __FILE__, __LINE__);
return -1;
}
if (prop == NULL)
{
MXS_ERROR("Router client session property is NULL. (%s:%d)", __FILE__, __LINE__);
return -1;
}
rses_property_t *p;
CHK_CLIENT_RSES(rses);
CHK_RSES_PROP(prop);
prop->rses_prop_rsession = rses;
p = rses->rses_properties[prop->rses_prop_type];
if (p == NULL)
{
rses->rses_properties[prop->rses_prop_type] = prop;
}
else
{
while (p->rses_prop_next != NULL)
{
p = p->rses_prop_next;
}
p->rses_prop_next = prop;
}
return 0;
}
/********************************
* This routine returns the root master server from MySQL replication tree
* Get the root Master rule:

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

@ -26,432 +26,45 @@
* Functions for session command handling
*/
static bool sescmd_cursor_history_empty(sescmd_cursor_t *scur);
static void sescmd_cursor_reset(sescmd_cursor_t *scur);
static bool sescmd_cursor_next(sescmd_cursor_t *scur);
static rses_property_t *mysql_sescmd_get_property(mysql_sescmd_t *scmd);
/*
* The following functions, all to do with the handling of session commands,
* are called from other modules of the read write split router:
*/
/**
* Router session must be locked.
* Return session command pointer if succeed, NULL if failed.
*/
mysql_sescmd_t *rses_property_get_sescmd(rses_property_t *prop)
void process_sescmd_response(ROUTER_CLIENT_SES* rses, SRWBackend& bref, GWBUF** ppPacket, bool* reconnect)
{
mysql_sescmd_t *sescmd;
if (prop == NULL)
if (bref->session_command_count())
{
MXS_ERROR("[%s] Error: NULL parameter.", __FUNCTION__);
return NULL;
}
CHK_RSES_PROP(prop);
sescmd = &prop->rses_prop_data.sescmd;
if (sescmd != NULL)
{
CHK_MYSQL_SESCMD(sescmd);
}
return sescmd;
}
/**
* Create session command property.
*/
mysql_sescmd_t *mysql_sescmd_init(rses_property_t *rses_prop,
GWBUF *sescmd_buf,
unsigned char packet_type,
ROUTER_CLIENT_SES *rses)
{
mysql_sescmd_t *sescmd;
CHK_RSES_PROP(rses_prop);
/** Can't call rses_property_get_sescmd with uninitialized sescmd */
sescmd = &rses_prop->rses_prop_data.sescmd;
sescmd->my_sescmd_prop = rses_prop; /*< reference to owning property */
sescmd->my_sescmd_chk_top = CHK_NUM_MY_SESCMD;
sescmd->my_sescmd_chk_tail = CHK_NUM_MY_SESCMD;
/** Set session command buffer */
sescmd->my_sescmd_buf = sescmd_buf;
sescmd->my_sescmd_packet_type = packet_type;
sescmd->position = atomic_add_uint64(&rses->pos_generator, 1);
sescmd->my_sescmd_is_replied = false;
sescmd->reply_cmd = 0;
return sescmd;
}
void mysql_sescmd_done(mysql_sescmd_t *sescmd)
{
if (sescmd == NULL)
{
MXS_ERROR("[%s] Error: NULL parameter.", __FUNCTION__);
return;
}
CHK_RSES_PROP(sescmd->my_sescmd_prop);
gwbuf_free(sescmd->my_sescmd_buf);
memset(sescmd, 0, sizeof(mysql_sescmd_t));
}
/**
* All cases where backend message starts at least with one response to session
* command are handled here.
* Read session commands from property list. If command is already replied,
* discard packet. Else send reply to client. In both cases move cursor forward
* until all session command replies are handled.
*
* Cases that are expected to happen and which are handled:
* s = response not yet replied to client, S = already replied response,
* q = query
* 1. q+ for example : select * from mysql.user
* 2. s+ for example : set autocommit=1
* 3. S+
* 4. sq+
* 5. Sq+
* 6. Ss+
* 7. Ss+q+
* 8. S+q+
* 9. s+q+
*/
GWBUF *sescmd_cursor_process_replies(GWBUF *replybuf,
backend_ref_t *bref,
bool *reconnect)
{
sescmd_cursor_t *scur = &bref->bref_sescmd_cur;
mysql_sescmd_t *scmd = sescmd_cursor_get_command(scur);
ROUTER_CLIENT_SES *ses = (*scur->scmd_cur_ptr_property)->rses_prop_rsession;
CHK_GWBUF(replybuf);
/**
* Walk through packets in the message and the list of session
* commands.
*/
while (scmd != NULL && replybuf != NULL)
{
bref->reply_cmd = *((unsigned char *)replybuf->start + 4);
scur->position = scmd->position;
/** Faster backend has already responded to client : discard */
if (scmd->my_sescmd_is_replied)
/** We are executing a session command */
if (GWBUF_IS_TYPE_SESCMD_RESPONSE((*ppPacket)))
{
bool last_packet = false;
uint8_t cmd;
gwbuf_copy_data(*ppPacket, MYSQL_HEADER_LEN, 1, &cmd);
uint64_t id = bref->complete_session_command();
CHK_GWBUF(replybuf);
while (!last_packet)
if (rses->recv_sescmd < rses->sent_sescmd &&
id == rses->recv_sescmd + 1 &&
(!rses->current_master || // Session doesn't have a master
rses->current_master == bref)) // This is the master's response
{
int buflen;
/** First reply to this session command, route it to the client */
++rses->recv_sescmd;
buflen = GWBUF_LENGTH(replybuf);
last_packet = GWBUF_IS_TYPE_RESPONSE_END(replybuf);
/** discard packet */
replybuf = gwbuf_consume(replybuf, buflen);
/** Store the master's response so that the slave responses can
* be compared to it */
rses->sescmd_responses[id] = cmd;
}
/** Set response status received */
bref_clear_state(bref, BREF_WAITING_RESULT);
if (bref->reply_cmd != scmd->reply_cmd && BREF_IS_IN_USE(bref))
else
{
MXS_ERROR("Slave server '%s': response differs from master's response. "
"Closing connection due to inconsistent session state.",
bref->ref->server->unique_name);
close_failed_bref(bref, true);
/** The reply to this session command has already been sent to
* the client, discard it */
gwbuf_free(*ppPacket);
*ppPacket = NULL;
RW_CHK_DCB(bref, bref->bref_dcb);
dcb_close(bref->bref_dcb);
RW_CLOSE_BREF(bref);
*reconnect = true;
gwbuf_free(replybuf);
replybuf = NULL;
}
}
/** This is a response from the master and it is the "right" one.
* A slave server's response will be compared to this and if
* their response differs from the master server's response, they
* are dropped from the valid list of backend servers.
* Response is in the buffer and it will be sent to client.
*
* If we have no master server, the first slave's response is considered
* the "right" one. */
else if (ses->rses_master_ref == NULL ||
!BREF_IS_IN_USE(ses->rses_master_ref) ||
ses->rses_master_ref->bref_dcb == bref->bref_dcb)
{
/** Mark the rest session commands as replied */
scmd->my_sescmd_is_replied = true;
scmd->reply_cmd = *((unsigned char *)replybuf->start + 4);
MXS_INFO("Server '%s' responded to a session command, sending the response "
"to the client.", bref->ref->server->unique_name);
for (int i = 0; i < ses->rses_nbackends; i++)
{
if (!BREF_IS_WAITING_RESULT(&ses->rses_backend_ref[i]))
if (rses->sescmd_responses[id] != cmd)
{
/** This backend has already received a response */
if (ses->rses_backend_ref[i].reply_cmd != scmd->reply_cmd &&
!BREF_IS_CLOSED(&ses->rses_backend_ref[i]) &&
BREF_IS_IN_USE(&ses->rses_backend_ref[i]))
{
close_failed_bref(&ses->rses_backend_ref[i], true);
if (ses->rses_backend_ref[i].bref_dcb)
{
RW_CHK_DCB(&ses->rses_backend_ref[i], ses->rses_backend_ref[i].bref_dcb);
dcb_close(ses->rses_backend_ref[i].bref_dcb);
RW_CLOSE_BREF(&ses->rses_backend_ref[i]);
}
*reconnect = true;
MXS_INFO("Disabling slave [%s]:%d, result differs from "
"master's result. Master: %d Slave: %d",
ses->rses_backend_ref[i].ref->server->name,
ses->rses_backend_ref[i].ref->server->port,
bref->reply_cmd, ses->rses_backend_ref[i].reply_cmd);
}
MXS_ERROR("Slave server '%s': response differs from master's response. "
"Closing connection due to inconsistent session state.",
bref->server()->unique_name);
bref->close(mxs::Backend::CLOSE_FATAL);
*reconnect = true;
}
}
}
else
{
MXS_INFO("Slave '%s' responded before master to a session command. Result: %d",
bref->ref->server->unique_name,
(int)bref->reply_cmd);
if (bref->reply_cmd == 0xff)
{
SERVER *serv = bref->ref->server;
MXS_ERROR("Slave '%s' (%s:%u) failed to execute session command.",
serv->unique_name, serv->name, serv->port);
}
gwbuf_free(replybuf);
replybuf = NULL;
}
if (sescmd_cursor_next(scur))
{
scmd = sescmd_cursor_get_command(scur);
}
else
{
scmd = NULL;
/** All session commands are replied */
scur->scmd_cur_active = false;
}
}
ss_dassert(replybuf == NULL || *scur->scmd_cur_ptr_property == NULL);
return replybuf;
}
/**
* Get the address of current session command.
*
* Router session must be locked */
mysql_sescmd_t *sescmd_cursor_get_command(sescmd_cursor_t *scur)
{
mysql_sescmd_t *scmd;
scur->scmd_cur_cmd = rses_property_get_sescmd(*scur->scmd_cur_ptr_property);
CHK_MYSQL_SESCMD(scur->scmd_cur_cmd);
scmd = scur->scmd_cur_cmd;
return scmd;
}
/** router must be locked */
bool sescmd_cursor_is_active(sescmd_cursor_t *sescmd_cursor)
{
bool succp;
if (sescmd_cursor == NULL)
{
MXS_ERROR("[%s] Error: NULL parameter.", __FUNCTION__);
return false;
}
succp = sescmd_cursor->scmd_cur_active;
return succp;
}
/** router must be locked */
void sescmd_cursor_set_active(sescmd_cursor_t *sescmd_cursor,
bool value)
{
/** avoid calling unnecessarily */
ss_dassert(sescmd_cursor->scmd_cur_active != value);
sescmd_cursor->scmd_cur_active = value;
}
/**
* Clone session command's command buffer.
* Router session must be locked
*/
GWBUF *sescmd_cursor_clone_querybuf(sescmd_cursor_t *scur)
{
GWBUF *buf;
if (scur == NULL)
{
MXS_ERROR("[%s] Error: NULL parameter.", __FUNCTION__);
return NULL;
}
ss_dassert(scur->scmd_cur_cmd != NULL);
buf = gwbuf_clone(scur->scmd_cur_cmd->my_sescmd_buf);
CHK_GWBUF(buf);
return buf;
}
bool execute_sescmd_history(backend_ref_t *bref)
{
ss_dassert(bref);
CHK_BACKEND_REF(bref);
bool succp = true;
sescmd_cursor_t *scur = &bref->bref_sescmd_cur;
CHK_SESCMD_CUR(scur);
if (!sescmd_cursor_history_empty(scur))
{
sescmd_cursor_reset(scur);
if ((succp = execute_sescmd_in_backend(bref)))
{
scur->scmd_cur_rses->expected_responses++;
}
}
return succp;
}
static bool sescmd_cursor_history_empty(sescmd_cursor_t *scur)
{
bool succp;
if (scur == NULL)
{
MXS_ERROR("[%s] Error: NULL parameter.", __FUNCTION__);
return true;
}
CHK_SESCMD_CUR(scur);
if (scur->scmd_cur_rses->rses_properties[RSES_PROP_TYPE_SESCMD] == NULL)
{
succp = true;
}
else
{
succp = false;
}
return succp;
}
/*
* End of functions called from other modules of the read write split router;
* start of functions that are internal to this module.
*/
static void sescmd_cursor_reset(sescmd_cursor_t *scur)
{
ROUTER_CLIENT_SES *rses;
if (scur == NULL)
{
MXS_ERROR("[%s] Error: NULL parameter.", __FUNCTION__);
return;
}
CHK_SESCMD_CUR(scur);
CHK_CLIENT_RSES(scur->scmd_cur_rses);
rses = scur->scmd_cur_rses;
scur->scmd_cur_ptr_property = &rses->rses_properties[RSES_PROP_TYPE_SESCMD];
CHK_RSES_PROP((*scur->scmd_cur_ptr_property));
scur->scmd_cur_active = false;
scur->scmd_cur_cmd = &(*scur->scmd_cur_ptr_property)->rses_prop_data.sescmd;
}
/**
* Moves cursor to next property and copied address of its sescmd to cursor.
* Current propery must be non-null.
* If current property is the last on the list, *scur->scmd_ptr_property == NULL
*
* Router session must be locked
*/
static bool sescmd_cursor_next(sescmd_cursor_t *scur)
{
bool succp = false;
rses_property_t *prop_curr;
rses_property_t *prop_next;
if (scur == NULL)
{
MXS_ERROR("[%s] Error: NULL parameter.", __FUNCTION__);
return false;
}
ss_dassert(scur != NULL);
ss_dassert(*(scur->scmd_cur_ptr_property) != NULL);
/** Illegal situation */
if (scur == NULL || *scur->scmd_cur_ptr_property == NULL ||
scur->scmd_cur_cmd == NULL)
{
/** Log error */
goto return_succp;
}
prop_curr = *(scur->scmd_cur_ptr_property);
CHK_MYSQL_SESCMD(scur->scmd_cur_cmd);
ss_dassert(prop_curr == mysql_sescmd_get_property(scur->scmd_cur_cmd));
CHK_RSES_PROP(prop_curr);
/** Copy address of pointer to next property */
scur->scmd_cur_ptr_property = &(prop_curr->rses_prop_next);
prop_next = *scur->scmd_cur_ptr_property;
ss_dassert(prop_next == *(scur->scmd_cur_ptr_property));
/** If there is a next property move forward */
if (prop_next != NULL)
{
CHK_RSES_PROP(prop_next);
CHK_RSES_PROP((*(scur->scmd_cur_ptr_property)));
/** Get pointer to next property's sescmd */
scur->scmd_cur_cmd = rses_property_get_sescmd(prop_next);
ss_dassert(prop_next == scur->scmd_cur_cmd->my_sescmd_prop);
CHK_MYSQL_SESCMD(scur->scmd_cur_cmd);
CHK_RSES_PROP(scur->scmd_cur_cmd->my_sescmd_prop);
}
else
{
/** No more properties, can't proceed. */
goto return_succp;
}
if (scur->scmd_cur_cmd != NULL)
{
succp = true;
}
else
{
ss_dassert(false); /*< Log error, sescmd shouldn't be NULL */
}
return_succp:
return succp;
}
static rses_property_t *mysql_sescmd_get_property(mysql_sescmd_t *scmd)
{
CHK_MYSQL_SESCMD(scmd);
return scmd->my_sescmd_prop;
}