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

get session transation state from backend via session track mechanism

Browse Source
This commit is contained in:
Dapeng Huang
2018-01-14 12:23:38 +08:00
parent 27e563dcff
commit e1aeac8b07
16 changed files with 231 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/modules/protocol/MySQL/mariadbbackend/mysql_backend.c
View File

@ -977,6 +977,8 @@ gw_read_and_write(DCB *dcb)
gwbuf_set_type(stmt, GWBUF_TYPE_RESULT);
}
mxs_mysql_get_session_track_info(stmt, proto->server_capabilities);
session->service->router->clientReply(session->service->router_instance,
session->router_session,
stmt, dcb);

55
server/modules/protocol/MySQL/mariadbclient/mysql_client.cc
View File

@ -76,7 +76,7 @@ static spec_com_res_t process_special_commands(DCB *client_dcb, GWBUF *read_buff
static spec_com_res_t handle_query_kill(DCB* dcb, GWBUF* read_buffer, spec_com_res_t current,
bool is_complete, unsigned int packet_len);
static bool parse_kill_query(char *query, uint64_t *thread_id_out, kill_type_t *kt_out, std::string* user);
static void parse_and_set_trx_state(MXS_SESSION *ses, GWBUF *data);
/**
* The module entry point routine. It is this routine that
* must populate the structure that is referred to as the
@ -411,6 +411,11 @@ int MySQLSendHandshake(DCB* dcb)
*/
int gw_MySQLWrite_client(DCB *dcb, GWBUF *queue)
{
MySQLProtocol *protocol = DCB_PROTOCOL(dcb, MySQLProtocol);
if (GWBUF_IS_REPLY_OK(queue) && protocol->session_track_trx_state)
{
parse_and_set_trx_state(dcb->session, queue);
}
return dcb_write(dcb, queue);
}
@ -1266,7 +1271,8 @@ int gw_MySQLListener(DCB *listen_dcb, char *config_bind)
int gw_MySQLAccept(DCB *listener)
{
DCB *client_dcb;
MySQLProtocol *protocol;
MySQLProtocol *cli_proto;
SERV_LISTENER *listener_proto;
CHK_DCB(listener);
@ -1279,6 +1285,9 @@ int gw_MySQLAccept(DCB *listener)
while ((client_dcb = dcb_accept(listener)) != NULL)
{
gw_process_one_new_client(client_dcb);
cli_proto = DCB_PROTOCOL(client_dcb, MySQLProtocol);
listener_proto = (SERV_LISTENER *)listener->listener;
cli_proto->session_track_trx_state = listener_proto->session_track_trx_state;
} /**< while client_dcb != NULL */
}
@ -1510,7 +1519,8 @@ static int route_by_statement(MXS_SESSION* session, uint64_t capabilities, GWBUF
}
}
if (rcap_type_required(capabilities, RCAP_TYPE_TRANSACTION_TRACKING))
MySQLProtocol *protocol = DCB_PROTOCOL(session->client_dcb, MySQLProtocol);
if (rcap_type_required(capabilities, RCAP_TYPE_TRANSACTION_TRACKING) && !protocol->session_track_trx_state)
{
if (session_trx_is_ending(session))
{
@ -1889,3 +1899,42 @@ static bool parse_kill_query(char *query, uint64_t *thread_id_out, kill_type_t *
return true;
}
}
static void parse_and_set_trx_state(MXS_SESSION *ses, GWBUF *data)
{
char *autocommit = gwbuf_get_property(data, (char *)"autocommit");
if (autocommit)
{
if (strncasecmp(autocommit, "0", 1) == 0)
{
session_set_autocommit(ses, true);
}
if (strncasecmp(autocommit, "1", 1) == 0)
{
session_set_autocommit(ses, false);
}
}
char *trx_state = gwbuf_get_property(data, (char *)"trx_state");
if (trx_state)
{
mysql_tx_state_t s = parse_trx_state(trx_state);
MXS_DEBUG("parsed tx state:%d", s);
if (s == TX_EMPTY)
{
session_set_trx_state(ses, SESSION_TRX_INACTIVE);
}
else if(s & TX_READ_TRX)
{
session_set_trx_state(ses, SESSION_TRX_READ_ONLY);
}
else if(s & TX_WRITE_TRX)
{
session_set_trx_state(ses, SESSION_TRX_READ_WRITE);
}
else if((s & TX_EXPLICIT) | (s & TX_IMPLICIT))
{
session_set_trx_state(ses, SESSION_TRX_ACTIVE);
}
}
}

130
server/modules/protocol/MySQL/mysql_common.cc
View File

@ -1228,6 +1228,8 @@ create_capabilities(MySQLProtocol *conn, bool with_ssl, bool db_specified, bool
/* Maybe it should depend on whether CA certificate is provided */
/* final_capabilities |= (uint32_t)GW_MYSQL_CAPABILITIES_SSL_VERIFY_SERVER_CERT; */
}
/** add session track */
final_capabilities |= (uint32_t)GW_MYSQL_CAPABILITIES_SESSION_TRACK;
/* Compression is not currently supported */
ss_dassert(!compress);
@ -1459,10 +1461,7 @@ int gw_decode_mysql_server_handshake(MySQLProtocol *conn, uint8_t *payload)
mysql_server_capabilities_two = gw_mysql_get_byte2(payload);
memcpy(capab_ptr, &mysql_server_capabilities_one, 2);
// get capabilities part 2 (2 bytes)
memcpy(&capab_ptr[2], &mysql_server_capabilities_two, 2);
conn->server_capabilities = mysql_server_capabilities_one | mysql_server_capabilities_two << 16;
// 2 bytes shift
payload += 2;
@ -1768,3 +1767,126 @@ void mxs_mysql_execute_kill_user(MXS_SESSION* issuer, const char* user, kill_typ
mxs_mysql_send_ok(issuer->client_dcb, info.targets.size(), 0, NULL);
}
/**
* Decode ok packet and get session track info
*
* @param buffer Buffer contain ok packet
* @param server_capabilities Server capabilities
*
*/
void mxs_mysql_get_session_track_info(GWBUF *buff, uint32_t server_capabilities)
{
char *trx_info, *var_name, *var_value;
if (GWBUF_LENGTH(buff) < 9 || !mxs_mysql_is_ok_packet(buff))
{
return;
}
if (!GWBUF_IS_CONTIGUOUS(buff))
{
buff = gwbuf_make_contiguous(buff);
}
buff->gwbuf_type |= GWBUF_TYPE_REPLY_OK;
uint8_t* ptr = GWBUF_DATA(buff);
ptr += (MYSQL_COM_OFFSET + 1); // Header and Command type
mxs_leint_consume(&ptr); // Affected rows
mxs_leint_consume(&ptr); // Last insert-id
uint16_t server_status = gw_mysql_get_byte2(ptr);
ptr += 2; // status
ptr += 2; // number of warnings
if (server_capabilities & GW_MYSQL_CAPABILITIES_SESSION_TRACK)
{
if (ptr < buff->end) {
ptr += mxs_leint_consume(&ptr); // info
if (server_status & SERVER_SESSION_STATE_CHANGED)
{
mxs_leint_consume(&ptr); // total SERVER_SESSION_STATE_CHANGED length
while (ptr < buff->end)
{
enum_session_state_type type = (enum enum_session_state_type)mxs_leint_consume(&ptr);
switch (type)
{
case SESSION_TRACK_SYSTEM_VARIABLES:
case SESSION_TRACK_SCHEMA:
case SESSION_TRACK_GTIDS:
case SESSION_TRACK_TRANSACTION_CHARACTERISTICS:
ptr += mxs_leint_consume(&ptr);
break;
case SESSION_TRACK_STATE_CHANGE:
mxs_leint_consume(&ptr); //lenth
// system variables like autocommit, schema, charset ...
var_name = mxs_lestr_consume_dup(&ptr);
var_value = mxs_lestr_consume_dup(&ptr);
gwbuf_add_property(buff, var_name, var_value);
MXS_FREE(var_name);
MXS_FREE(var_value);
break;
case SESSION_TRACK_TRANSACTION_TYPE:
mxs_leint_consume(&ptr); // length
trx_info = mxs_lestr_consume_dup(&ptr);
MXS_INFO("get trx_info:%s", trx_info);
gwbuf_add_property(buff, (char *)"trx_state", trx_info);
MXS_FREE(trx_info);
break;
default:
ptr += mxs_leint_consume(&ptr);
MXS_WARNING("recieved unexpecting session track type:%d", type);
break;
}
}
}
}
}
}
mysql_tx_state_t parse_trx_state(char *str)
{
int s = TX_EMPTY;
assert(str);
do
{
switch (*str)
{
case 'T':
s |= TX_EXPLICIT;
break;
case 'I':
s |= TX_IMPLICIT;
break;
case 'r':
s |= TX_READ_UNSAFE;
break;
case 'R':
s |= TX_READ_TRX;
break;
case 'w':
s |= TX_WRITE_UNSAFE;
break;
case 'W':
s |= TX_WRITE_TRX;
break;
case 's':
s |= TX_STMT_UNSAFE;
break;
case 'S':
s |= TX_RESULT_SET;
break;
case 'L':
s |= TX_LOCKED_TABLES;
break;
default:
break;
}
} while(*(str++) != 0);
return (mysql_tx_state_t)s;
}