Merge branch 'develop'
Completion of fix to bug #418 http://bugs.skysql.com/show_bug.cgi?id=418
This commit is contained in:
commit
b20ce05ee3
@ -5,6 +5,7 @@
|
||||
|
||||
include ../../build_gateway.inc
|
||||
include ../../makefile.inc
|
||||
include ../../test.inc
|
||||
|
||||
CC = gcc
|
||||
CPP = g++
|
||||
@ -43,7 +44,7 @@ buildtests:
|
||||
|
||||
|
||||
runtests:
|
||||
@echo "" >> $(TESTLOG)
|
||||
@echo "" > $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo $(shell date) >> $(TESTLOG)
|
||||
@echo "Test Log Manager" >> $(TESTLOG)
|
||||
@ -62,3 +63,5 @@ runtests:
|
||||
@-$(LAUNCH_DEBUGGER) $(TESTAPP) "-t 16" 2>>$(TESTLOG)
|
||||
@echo "Log Manager PASSED" >> $(TESTLOG)
|
||||
@echo "" >> $(TESTLOG)
|
||||
|
||||
# @cat $(TESTLOG) >> $(TEST_MAXSCALE_LOG)
|
||||
|
@ -50,11 +50,13 @@
|
||||
#include <sql_parse.h>
|
||||
#include <errmsg.h>
|
||||
#include <client_settings.h>
|
||||
|
||||
#include <set_var.h>
|
||||
#include <strfunc.h>
|
||||
#include <item_func.h>
|
||||
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <stdarg.h>
|
||||
|
||||
extern int lm_enabled_logfiles_bitmask;
|
||||
@ -73,9 +75,13 @@ static bool create_parse_tree(
|
||||
|
||||
static skygw_query_type_t resolve_query_type(
|
||||
THD* thd);
|
||||
|
||||
static bool skygw_stmt_causes_implicit_commit(
|
||||
LEX* lex,
|
||||
uint mask);
|
||||
LEX* lex,
|
||||
int* autocommit_stmt);
|
||||
|
||||
static int is_autocommit_stmt(
|
||||
LEX* lex);
|
||||
|
||||
/**
|
||||
* @node (write brief function description here)
|
||||
@ -346,9 +352,9 @@ return_here:
|
||||
* restrictive, for example, QUERY_TYPE_READ is smaller than QUERY_TYPE_WRITE.
|
||||
*
|
||||
*/
|
||||
static u_int8_t set_query_type(
|
||||
u_int8_t* qtype,
|
||||
u_int8_t new_type)
|
||||
static u_int16_t set_query_type(
|
||||
u_int16_t* qtype,
|
||||
u_int16_t new_type)
|
||||
{
|
||||
*qtype = MAX(*qtype, new_type);
|
||||
return *qtype;
|
||||
@ -374,7 +380,8 @@ static skygw_query_type_t resolve_query_type(
|
||||
THD* thd)
|
||||
{
|
||||
skygw_query_type_t qtype = QUERY_TYPE_UNKNOWN;
|
||||
u_int8_t type = QUERY_TYPE_UNKNOWN;
|
||||
u_int16_t type = QUERY_TYPE_UNKNOWN;
|
||||
int set_autocommit_stmt = -1; /*< -1 no, 0 disable, 1 enable */
|
||||
LEX* lex;
|
||||
Item* item;
|
||||
/**
|
||||
@ -397,7 +404,9 @@ static skygw_query_type_t resolve_query_type(
|
||||
goto return_qtype;
|
||||
}
|
||||
|
||||
if (skygw_stmt_causes_implicit_commit(lex, CF_AUTO_COMMIT_TRANS))
|
||||
if (skygw_stmt_causes_implicit_commit(
|
||||
lex,
|
||||
&set_autocommit_stmt))
|
||||
{
|
||||
if (LOG_IS_ENABLED(LOGFILE_TRACE))
|
||||
{
|
||||
@ -418,7 +427,25 @@ static skygw_query_type_t resolve_query_type(
|
||||
"next command.");
|
||||
}
|
||||
}
|
||||
|
||||
if (set_autocommit_stmt == 1)
|
||||
{
|
||||
type |= QUERY_TYPE_ENABLE_AUTOCOMMIT;
|
||||
}
|
||||
type |= QUERY_TYPE_COMMIT;
|
||||
}
|
||||
|
||||
if (set_autocommit_stmt == 0)
|
||||
{
|
||||
if (LOG_IS_ENABLED(LOGFILE_TRACE))
|
||||
{
|
||||
skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Disable autocommit : implicit START TRANSACTION"
|
||||
" before executing the next command.");
|
||||
}
|
||||
type |= QUERY_TYPE_DISABLE_AUTOCOMMIT;
|
||||
type |= QUERY_TYPE_BEGIN_TRX;
|
||||
}
|
||||
/**
|
||||
* REVOKE ALL, ASSIGN_TO_KEYCACHE,
|
||||
@ -648,11 +675,18 @@ return_qtype:
|
||||
return qtype;
|
||||
}
|
||||
|
||||
static bool skygw_stmt_causes_implicit_commit(LEX* lex, uint mask)
|
||||
/**
|
||||
* Checks if statement causes implicit COMMIT.
|
||||
* autocommit_stmt gets values 1, 0 or -1 if stmt is enable, disable or
|
||||
* something else than autocommit.
|
||||
*/
|
||||
static bool skygw_stmt_causes_implicit_commit(
|
||||
LEX* lex,
|
||||
int* autocommit_stmt)
|
||||
{
|
||||
bool succp;
|
||||
|
||||
if (!(sql_command_flags[lex->sql_command] & mask))
|
||||
|
||||
if (!(sql_command_flags[lex->sql_command] & CF_AUTO_COMMIT_TRANS))
|
||||
{
|
||||
succp = false;
|
||||
goto return_succp;
|
||||
@ -668,13 +702,83 @@ static bool skygw_stmt_causes_implicit_commit(LEX* lex, uint mask)
|
||||
succp = !(lex->create_info.options & HA_LEX_CREATE_TMP_TABLE);
|
||||
break;
|
||||
case SQLCOM_SET_OPTION:
|
||||
succp = lex->autocommit ? true : false;
|
||||
if ((*autocommit_stmt = is_autocommit_stmt(lex)) == 1)
|
||||
{
|
||||
succp = true;
|
||||
}
|
||||
break;
|
||||
default:
|
||||
succp = true;
|
||||
break;
|
||||
}
|
||||
|
||||
return_succp:
|
||||
return succp;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Finds out if stmt is SET autocommit
|
||||
* and if the new value matches with the enable_cmd argument.
|
||||
*
|
||||
* Returns 1, 0, or -1 if command was:
|
||||
* enable, disable, or not autocommit, respectively.
|
||||
*/
|
||||
static int is_autocommit_stmt(
|
||||
LEX* lex)
|
||||
{
|
||||
struct list_node* node;
|
||||
set_var* setvar;
|
||||
int rc = -1;
|
||||
static char target[8]; /*< for converted string */
|
||||
Item* item = NULL;
|
||||
|
||||
node = lex->var_list.first_node();
|
||||
setvar=(set_var*)node->info;
|
||||
|
||||
if (setvar == NULL)
|
||||
{
|
||||
goto return_rc;
|
||||
}
|
||||
|
||||
do /*< Search for the last occurrence of 'autocommit' */
|
||||
{
|
||||
if ((sys_var*)setvar->var == Sys_autocommit_ptr)
|
||||
{
|
||||
item = setvar->value;
|
||||
}
|
||||
node = node->next;
|
||||
} while ((setvar = (set_var*)node->info) != NULL);
|
||||
|
||||
if (item != NULL) /*< found autocommit command */
|
||||
{
|
||||
if (item->type() == Item::INT_ITEM) /*< '0' or '1' */
|
||||
{
|
||||
rc = item->val_int();
|
||||
|
||||
if (rc > 1 || rc < 0)
|
||||
{
|
||||
rc = -1;
|
||||
}
|
||||
}
|
||||
else if (item->type() == Item::STRING_ITEM) /*< 'on' or 'off' */
|
||||
{
|
||||
String str(target, sizeof(target), system_charset_info);
|
||||
String* res = item->val_str(&str);
|
||||
int rc;
|
||||
|
||||
if ((rc = find_type(&bool_typelib, res->ptr(), res->length(), false)))
|
||||
{
|
||||
ss_dassert(rc >= 0 && rc <= 2);
|
||||
/**
|
||||
* rc is the position of matchin string in
|
||||
* typelib's value array.
|
||||
* 1=OFF, 2=ON.
|
||||
*/
|
||||
rc -= 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return_rc:
|
||||
return rc;
|
||||
}
|
||||
|
@ -29,15 +29,17 @@ EXTERN_C_BLOCK_BEGIN
|
||||
* is modified
|
||||
*/
|
||||
typedef enum {
|
||||
QUERY_TYPE_UNKNOWN = 0, /*< Couln't find out or parse error */
|
||||
QUERY_TYPE_LOCAL_READ = (1<<0), /*< Read non-database data, execute in MaxScale */
|
||||
QUERY_TYPE_READ = (1<<1), /*< No updates */
|
||||
QUERY_TYPE_WRITE = (1<<2), /*< Master data will be modified */
|
||||
QUERY_TYPE_SESSION_WRITE = (1<<3), /*< Session data will be modified */
|
||||
QUERY_TYPE_GLOBAL_WRITE = (1<<4), /*< Global system variable modification */
|
||||
QUERY_TYPE_BEGIN_TRX = (1<<5), /*< BEGIN or START TRANSACTION */
|
||||
QUERY_TYPE_ROLLBACK = (1<<6), /*< ROLLBACK */
|
||||
QUERY_TYPE_COMMIT = (1<<7), /*< COMMIT */
|
||||
QUERY_TYPE_UNKNOWN = 0x000, /*< Initial value, can't be tested bitwisely */
|
||||
QUERY_TYPE_LOCAL_READ = 0x001, /*< Read non-database data, execute in MaxScale */
|
||||
QUERY_TYPE_READ = 0x002, /*< No updates */
|
||||
QUERY_TYPE_WRITE = 0x004, /*< Master data will be modified */
|
||||
QUERY_TYPE_SESSION_WRITE = 0x008, /*< Session data will be modified */
|
||||
QUERY_TYPE_GLOBAL_WRITE = 0x010, /*< Global system variable modification */
|
||||
QUERY_TYPE_BEGIN_TRX = 0x020, /*< BEGIN or START TRANSACTION */
|
||||
QUERY_TYPE_ENABLE_AUTOCOMMIT = 0x040,/*< SET autocommit=1 */
|
||||
QUERY_TYPE_DISABLE_AUTOCOMMIT = 0x080,/*< SET autocommit=0 */
|
||||
QUERY_TYPE_ROLLBACK = 0x100, /*< ROLLBACK */
|
||||
QUERY_TYPE_COMMIT = 0x200 /*< COMMIT */
|
||||
} skygw_query_type_t;
|
||||
|
||||
#define QUERY_IS_TYPE(mask,type) ((mask & type) == type)
|
||||
|
@ -5,6 +5,7 @@
|
||||
|
||||
include ../../build_gateway.inc
|
||||
include ../../makefile.inc
|
||||
include ../../test.inc
|
||||
|
||||
CC = gcc
|
||||
CPP = g++
|
||||
@ -48,7 +49,7 @@ buildtests:
|
||||
$(LDLIBS) $(LDMYSQL)
|
||||
|
||||
runtests:
|
||||
@echo "" >> $(TESTLOG)
|
||||
@echo "" > $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo $(shell date) >> $(TESTLOG)
|
||||
@echo "Test Query Classifier" >> $(TESTLOG)
|
||||
@ -59,3 +60,4 @@ ifeq ($?, 0)
|
||||
else
|
||||
@echo "Query Classifier FAILED" >> $(TESTLOG)
|
||||
endif
|
||||
@cat $(TESTLOG) >> $(TEST_MAXSCALE_LOG)
|
@ -86,7 +86,7 @@ static char* server_options[] = {
|
||||
"--no-defaults",
|
||||
"--datadir=",
|
||||
"--language=",
|
||||
"--skip-innodb",
|
||||
// "--skip-innodb",
|
||||
"--default-storage-engine=myisam",
|
||||
NULL
|
||||
};
|
||||
@ -100,7 +100,7 @@ static char* server_groups[] = {
|
||||
"server",
|
||||
"server",
|
||||
"embedded",
|
||||
"server",
|
||||
// "server",
|
||||
"server",
|
||||
NULL
|
||||
};
|
||||
|
@ -5,6 +5,7 @@
|
||||
|
||||
include ../../../build_gateway.inc
|
||||
include ../../../makefile.inc
|
||||
include ../../../test.inc
|
||||
|
||||
CC=cc
|
||||
TESTLOG := $(shell pwd)/testhash.log
|
||||
@ -26,7 +27,7 @@ buildtests :
|
||||
testhash.c ../hashtable.o ../atomic.o ../spinlock.o -o testhash
|
||||
|
||||
runtests:
|
||||
@echo "" >> $(TESTLOG)
|
||||
@echo "" > $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo $(shell date) >> $(TESTLOG)
|
||||
@echo "Test MaxScale core" >> $(TESTLOG)
|
||||
@ -38,3 +39,4 @@ else
|
||||
@echo "MaxScale core FAILED" >> $(TESTLOG)
|
||||
endif
|
||||
@echo "" >> $(TESTLOG)
|
||||
@cat $(TESTLOG) >> $(TEST_MAXSCALE_LOG)
|
@ -82,4 +82,20 @@ install: $(MODULES)
|
||||
install -D $(MODULES) $(DEST)/MaxScale/modules
|
||||
(cd readwritesplit; make DEST=$(DEST) install)
|
||||
|
||||
cleantests:
|
||||
$(MAKE) -C readwritesplit/test cleantests
|
||||
$(MAKE) -C test cleantests
|
||||
|
||||
buildtests:
|
||||
$(MAKE) -C readwritesplit/test DEBUG=Y buildtests
|
||||
$(MAKE) -C test DEBUG=Y buildtests
|
||||
|
||||
runtests:
|
||||
$(MAKE) -C test runtests
|
||||
$(MAKE) -C readwritesplit runtests
|
||||
|
||||
testall:
|
||||
$(MAKE) -C test testall
|
||||
$(MAKE) -C readwritesplit testall
|
||||
|
||||
include depend.mk
|
||||
|
@ -20,9 +20,9 @@
|
||||
|
||||
include ../../../../build_gateway.inc
|
||||
|
||||
LOGPATH := $(ROOT_PATH)/log_manager
|
||||
UTILSPATH := $(ROOT_PATH)/utils
|
||||
QCLASSPATH := $(ROOT_PATH)/query_classifier
|
||||
LOGPATH := $(ROOT_PATH)/log_manager
|
||||
UTILSPATH := $(ROOT_PATH)/utils
|
||||
QCLASSPATH := $(ROOT_PATH)/query_classifier
|
||||
|
||||
CC=cc
|
||||
CFLAGS=-c -fPIC -I/usr/include -I../../include -I../../../include \
|
||||
@ -61,5 +61,18 @@ depend:
|
||||
|
||||
install: $(MODULES)
|
||||
install -D $(MODULES) $(DEST)/MaxScale/modules
|
||||
|
||||
cleantests:
|
||||
$(MAKE) -C test cleantest
|
||||
|
||||
testall:
|
||||
$(MAKE) -C test testall
|
||||
|
||||
buildtests:
|
||||
$(MAKE) -C test buildtests
|
||||
|
||||
runtests:
|
||||
$(MAKE) -C runtests
|
||||
|
||||
|
||||
include depend.mk
|
||||
|
@ -157,6 +157,13 @@ static void tracelog_routed_query(
|
||||
DCB* dcb,
|
||||
GWBUF* buf);
|
||||
|
||||
static bool route_session_write(
|
||||
ROUTER_CLIENT_SES* router_client_ses,
|
||||
GWBUF* querybuf,
|
||||
ROUTER_INSTANCE* inst,
|
||||
unsigned char packet_type,
|
||||
skygw_query_type_t qtype);
|
||||
|
||||
static SPINLOCK instlock;
|
||||
static ROUTER_INSTANCE* instances;
|
||||
|
||||
@ -574,10 +581,12 @@ static int routeQuery(
|
||||
bool rses_is_closed;
|
||||
rses_property_t* prop;
|
||||
size_t len;
|
||||
static bool transaction_active;
|
||||
/** if false everything goes to master and session commands to slave too */
|
||||
static bool autocommit_enabled = true;
|
||||
/** if true everything goes to master and session commands to slave too */
|
||||
static bool transaction_active = false;
|
||||
|
||||
CHK_CLIENT_RSES(router_cli_ses);
|
||||
|
||||
|
||||
/** Dirty read for quick check if router is closed. */
|
||||
if (router_cli_ses->rses_closed)
|
||||
@ -669,207 +678,96 @@ static int routeQuery(
|
||||
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE,
|
||||
"Packet type\t%s",
|
||||
STRPACKETTYPE(packet_type))));
|
||||
|
||||
if (QUERY_IS_TYPE(qtype,QUERY_TYPE_COMMIT) &&
|
||||
transaction_active)
|
||||
/**
|
||||
* If autocommit is disabled or transaction is explicitly started
|
||||
* transaction becomes active and master gets all statements until
|
||||
* transaction is committed and autocommit is enabled again.
|
||||
*/
|
||||
if (autocommit_enabled &&
|
||||
QUERY_IS_TYPE(qtype, QUERY_TYPE_DISABLE_AUTOCOMMIT))
|
||||
{
|
||||
autocommit_enabled = false;
|
||||
|
||||
if (!transaction_active)
|
||||
{
|
||||
transaction_active = true;
|
||||
}
|
||||
}
|
||||
else if (!transaction_active &&
|
||||
QUERY_IS_TYPE(qtype, QUERY_TYPE_BEGIN_TRX))
|
||||
{
|
||||
transaction_active = true;
|
||||
}
|
||||
/**
|
||||
* Explicit COMMIT and ROLLBACK, implicit COMMIT.
|
||||
*/
|
||||
if (autocommit_enabled &&
|
||||
transaction_active &&
|
||||
(QUERY_IS_TYPE(qtype,QUERY_TYPE_COMMIT) ||
|
||||
QUERY_IS_TYPE(qtype,QUERY_TYPE_ROLLBACK)))
|
||||
{
|
||||
transaction_active = false;
|
||||
}
|
||||
else if (!autocommit_enabled &&
|
||||
QUERY_IS_TYPE(qtype, QUERY_TYPE_ENABLE_AUTOCOMMIT))
|
||||
{
|
||||
autocommit_enabled = true;
|
||||
transaction_active = false;
|
||||
}
|
||||
|
||||
|
||||
switch (qtype) {
|
||||
case QUERY_TYPE_WRITE:
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"%lu [routeQuery:rwsplit] Query type\t%s, "
|
||||
"routing to Master.",
|
||||
pthread_self(),
|
||||
STRQTYPE(qtype))));
|
||||
|
||||
LOGIF(LT, tracelog_routed_query(router_cli_ses,
|
||||
"routeQuery",
|
||||
master_dcb,
|
||||
gwbuf_clone(querybuf)));
|
||||
|
||||
ret = master_dcb->func.write(master_dcb, querybuf);
|
||||
atomic_add(&inst->stats.n_master, 1);
|
||||
|
||||
goto return_ret;
|
||||
break;
|
||||
|
||||
case QUERY_TYPE_READ:
|
||||
LOGIF(LT, (skygw_log_write_flush(
|
||||
LOGFILE_TRACE,
|
||||
"%lu [routeQuery:rwsplit] Query type\t%s, "
|
||||
"routing to %s.",
|
||||
pthread_self(),
|
||||
STRQTYPE(qtype),
|
||||
(transaction_active ? "Master" : "Slave"))));
|
||||
|
||||
LOGIF(LT, tracelog_routed_query(
|
||||
router_cli_ses,
|
||||
"routeQuery",
|
||||
(transaction_active ? master_dcb : slave_dcb),
|
||||
gwbuf_clone(querybuf)));
|
||||
|
||||
if (transaction_active)
|
||||
{
|
||||
ret = master_dcb->func.write(master_dcb, querybuf);
|
||||
}
|
||||
else
|
||||
{
|
||||
ret = slave_dcb->func.write(slave_dcb, querybuf);
|
||||
}
|
||||
LOGIF(LT, (skygw_log_write_flush(
|
||||
LOGFILE_TRACE,
|
||||
"%lu [routeQuery:rwsplit] Routed.",
|
||||
pthread_self())));
|
||||
|
||||
|
||||
atomic_add(&inst->stats.n_slave, 1);
|
||||
goto return_ret;
|
||||
break;
|
||||
|
||||
case QUERY_TYPE_SESSION_WRITE:
|
||||
/**
|
||||
* Execute in backends used by current router session.
|
||||
* Save session variable commands to router session property
|
||||
* struct. Thus, they can be replayed in backends which are
|
||||
* started and joined later.
|
||||
*
|
||||
* Suppress redundant OK packets sent by backends.
|
||||
*
|
||||
* DOES THIS ALL APPLY TO COM_QUIT AS WELL??
|
||||
*
|
||||
* The first OK packet is replied to the client.
|
||||
*
|
||||
*/
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"%lu [routeQuery:rwsplit] DCB M:%p s:%p, "
|
||||
"Query type\t%s, "
|
||||
"packet type %s, routing to all servers.",
|
||||
pthread_self(),
|
||||
master_dcb,
|
||||
slave_dcb,
|
||||
STRQTYPE(qtype),
|
||||
STRPACKETTYPE(packet_type))));
|
||||
/**
|
||||
* COM_QUIT is one-way message. Server doesn't respond to that.
|
||||
* Therefore reply processing is unnecessary and session
|
||||
* command property is not needed. It is just routed to both
|
||||
* backends.
|
||||
*/
|
||||
if (packet_type == COM_QUIT)
|
||||
{
|
||||
int rc;
|
||||
int rc2;
|
||||
|
||||
rc = master_dcb->func.write(master_dcb, gwbuf_clone(querybuf));
|
||||
rc2 = slave_dcb->func.write(slave_dcb, querybuf);
|
||||
|
||||
if (rc == 1 && rc == rc2)
|
||||
{
|
||||
ret = 1;
|
||||
}
|
||||
goto return_ret;
|
||||
}
|
||||
prop = rses_property_init(RSES_PROP_TYPE_SESCMD);
|
||||
/**
|
||||
* Additional reference is created to querybuf to
|
||||
* prevent it from being released before properties
|
||||
* are cleaned up as a part of router sessionclean-up.
|
||||
*/
|
||||
mysql_sescmd_init(prop, querybuf, packet_type, router_cli_ses);
|
||||
|
||||
/** Lock router session */
|
||||
if (!rses_begin_locked_router_action(router_cli_ses))
|
||||
{
|
||||
rses_property_done(prop);
|
||||
goto return_ret;
|
||||
}
|
||||
/** Add sescmd property to router client session */
|
||||
rses_property_add(router_cli_ses, prop);
|
||||
|
||||
/** Execute session command in master */
|
||||
if (execute_sescmd_in_backend(router_cli_ses, BE_MASTER))
|
||||
{
|
||||
ret = 1;
|
||||
}
|
||||
else
|
||||
{
|
||||
/** Log error */
|
||||
}
|
||||
/** Execute session command in slave */
|
||||
if (execute_sescmd_in_backend(router_cli_ses, BE_SLAVE))
|
||||
/**
|
||||
* Session update is always routed in the same way.
|
||||
*/
|
||||
if (QUERY_IS_TYPE(qtype, QUERY_TYPE_SESSION_WRITE))
|
||||
{
|
||||
if (route_session_write(
|
||||
router_cli_ses,
|
||||
querybuf,
|
||||
inst,
|
||||
packet_type,
|
||||
qtype))
|
||||
{
|
||||
ret = 1;
|
||||
}
|
||||
else
|
||||
else
|
||||
{
|
||||
/** Log error */
|
||||
ret = 0;
|
||||
}
|
||||
|
||||
/** Unlock router session */
|
||||
rses_end_locked_router_action(router_cli_ses);
|
||||
|
||||
atomic_add(&inst->stats.n_all, 1);
|
||||
goto return_ret;
|
||||
break;
|
||||
|
||||
case QUERY_TYPE_BEGIN_TRX:
|
||||
transaction_active = true;
|
||||
LOGIF(LT, tracelog_routed_query(router_cli_ses,
|
||||
"routeQuery",
|
||||
master_dcb,
|
||||
gwbuf_clone(querybuf)));
|
||||
ret = master_dcb->func.write(master_dcb, querybuf);
|
||||
LOGIF(LT, (skygw_log_write_flush(
|
||||
LOGFILE_TRACE,
|
||||
"%lu [routeQuery:rwsplit] Routed.",
|
||||
pthread_self())));
|
||||
atomic_add(&inst->stats.n_master, 1);
|
||||
goto return_ret;
|
||||
break;
|
||||
|
||||
case QUERY_TYPE_COMMIT:
|
||||
case QUERY_TYPE_ROLLBACK:
|
||||
transaction_active = false;
|
||||
LOGIF(LT, tracelog_routed_query(router_cli_ses,
|
||||
"routeQuery",
|
||||
master_dcb,
|
||||
gwbuf_clone(querybuf)));
|
||||
ret = master_dcb->func.write(master_dcb, querybuf);
|
||||
LOGIF(LT, (skygw_log_write_flush(
|
||||
LOGFILE_TRACE,
|
||||
"%lu [routeQuery:rwsplit] Routed.",
|
||||
pthread_self())));
|
||||
atomic_add(&inst->stats.n_master, 1);
|
||||
goto return_ret;
|
||||
break;
|
||||
|
||||
default:
|
||||
}
|
||||
else if (transaction_active) /*< all to master */
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"%lu [routeQuery:rwsplit] Query type\t%s, "
|
||||
"routing to Master by default.",
|
||||
pthread_self(),
|
||||
STRQTYPE(qtype))));
|
||||
|
||||
/**
|
||||
* Is this really ok?
|
||||
* What is not known is routed to master.
|
||||
*/
|
||||
LOGIF(LT, tracelog_routed_query(router_cli_ses,
|
||||
"routeQuery",
|
||||
master_dcb,
|
||||
gwbuf_clone(querybuf)));
|
||||
|
||||
LOGFILE_TRACE,
|
||||
"Transaction is active, routing to Master.")));
|
||||
|
||||
ret = master_dcb->func.write(master_dcb, querybuf);
|
||||
atomic_add(&inst->stats.n_master, 1);
|
||||
|
||||
goto return_ret;
|
||||
break;
|
||||
} /*< switch by query type */
|
||||
}
|
||||
else if (QUERY_IS_TYPE(qtype, QUERY_TYPE_READ))
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Read-only query, routing to Slave.")));
|
||||
|
||||
ss_dassert(QUERY_IS_TYPE(qtype, QUERY_TYPE_READ));
|
||||
ret = slave_dcb->func.write(slave_dcb, querybuf);
|
||||
atomic_add(&inst->stats.n_slave, 1);
|
||||
|
||||
goto return_ret;
|
||||
}
|
||||
else
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Begin transaction, write or unspecified type, "
|
||||
"routing to Master.")));
|
||||
ret = master_dcb->func.write(master_dcb, querybuf);
|
||||
atomic_add(&inst->stats.n_master, 1);
|
||||
|
||||
goto return_ret;
|
||||
}
|
||||
|
||||
return_ret:
|
||||
if (plainsqlbuf != NULL)
|
||||
@ -1851,4 +1749,102 @@ static uint8_t getCapabilities (
|
||||
|
||||
return_rc:
|
||||
return rc;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute in backends used by current router session.
|
||||
* Save session variable commands to router session property
|
||||
* struct. Thus, they can be replayed in backends which are
|
||||
* started and joined later.
|
||||
*
|
||||
* Suppress redundant OK packets sent by backends.
|
||||
*
|
||||
* The first OK packet is replied to the client.
|
||||
* Return true if succeed, false is returned if router session was closed or
|
||||
* if execute_sescmd_in_backend failed.
|
||||
*/
|
||||
static bool route_session_write(
|
||||
ROUTER_CLIENT_SES* router_cli_ses,
|
||||
GWBUF* querybuf,
|
||||
ROUTER_INSTANCE* inst,
|
||||
unsigned char packet_type,
|
||||
skygw_query_type_t qtype)
|
||||
{
|
||||
bool succp;
|
||||
DCB* master_dcb;
|
||||
DCB* slave_dcb;
|
||||
rses_property_t* prop;
|
||||
|
||||
master_dcb = router_cli_ses->rses_dcb[BE_MASTER];
|
||||
slave_dcb = router_cli_ses->rses_dcb[BE_SLAVE];
|
||||
CHK_DCB(master_dcb);
|
||||
CHK_DCB(slave_dcb);
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Session write, query type\t%s, packet type %s, "
|
||||
"routing to all servers.",
|
||||
STRQTYPE(qtype),
|
||||
STRPACKETTYPE(packet_type))));
|
||||
/**
|
||||
* COM_QUIT is one-way message. Server doesn't respond to that.
|
||||
* Therefore reply processing is unnecessary and session
|
||||
* command property is not needed. It is just routed to both
|
||||
* backends.
|
||||
*/
|
||||
if (packet_type == COM_QUIT)
|
||||
{
|
||||
int rc;
|
||||
int rc2;
|
||||
|
||||
rc = master_dcb->func.write(master_dcb, gwbuf_clone(querybuf));
|
||||
rc2 = slave_dcb->func.write(slave_dcb, querybuf);
|
||||
|
||||
if (rc == 1 && rc == rc2)
|
||||
{
|
||||
succp = true;
|
||||
}
|
||||
goto return_succp;
|
||||
}
|
||||
prop = rses_property_init(RSES_PROP_TYPE_SESCMD);
|
||||
/**
|
||||
* Additional reference is created to querybuf to
|
||||
* prevent it from being released before properties
|
||||
* are cleaned up as a part of router sessionclean-up.
|
||||
*/
|
||||
mysql_sescmd_init(prop, querybuf, packet_type, router_cli_ses);
|
||||
|
||||
/** Lock router session */
|
||||
if (!rses_begin_locked_router_action(router_cli_ses))
|
||||
{
|
||||
rses_property_done(prop);
|
||||
succp = false;
|
||||
goto return_succp;
|
||||
}
|
||||
/** Add sescmd property to router client session */
|
||||
rses_property_add(router_cli_ses, prop);
|
||||
|
||||
/** Execute session command in master */
|
||||
succp = execute_sescmd_in_backend(router_cli_ses, BE_MASTER);
|
||||
|
||||
if (!succp)
|
||||
{
|
||||
/** Unlock router session */
|
||||
rses_end_locked_router_action(router_cli_ses);
|
||||
goto return_succp;
|
||||
}
|
||||
/** Execute session command in slave */
|
||||
succp = execute_sescmd_in_backend(router_cli_ses, BE_SLAVE);
|
||||
if (!succp)
|
||||
{
|
||||
/** Unlock router session */
|
||||
rses_end_locked_router_action(router_cli_ses);
|
||||
goto return_succp;
|
||||
}
|
||||
/** Unlock router session */
|
||||
rses_end_locked_router_action(router_cli_ses);
|
||||
|
||||
atomic_add(&inst->stats.n_all, 1);
|
||||
|
||||
return_succp:
|
||||
return succp;
|
||||
}
|
@ -8,7 +8,7 @@ include $(ROOT_PATH)/makefile.inc
|
||||
include $(ROOT_PATH)/test.inc
|
||||
|
||||
CC=cc
|
||||
TESTLOG := $(shell pwd)/testrouters.log
|
||||
TESTLOG := $(shell pwd)/testrwsplit.log
|
||||
RET := -1
|
||||
|
||||
cleantests:
|
||||
@ -25,17 +25,11 @@ buildtests:
|
||||
|
||||
|
||||
runtests:
|
||||
@echo "" >> $(TESTLOG)
|
||||
@echo "" > $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo $(shell date) >> $(TESTLOG)
|
||||
@echo "Test MaxScale R/W Split" >> $(TESTLOG)
|
||||
@echo "Test Read/Write split router" >> $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
./rwsplit.sh $(TESTLOG) $(THOST) $(TPORT_RW) $(TMASTER_ID) $(TUSER) $(TPWD)
|
||||
@echo "" >> $(TESTLOG)
|
||||
|
||||
|
||||
pesce:
|
||||
@echo "fine"
|
||||
|
||||
|
||||
|
||||
@cat $(TESTLOG) >> $(TEST_MAXSCALE_LOG)
|
@ -30,9 +30,46 @@ else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
TINPUT=test_transaction_routing2b.sql
|
||||
TRETVAL=0
|
||||
a=`$RUNCMD < ./$TINPUT`
|
||||
if [ "$a" != "$TRETVAL" ]; then
|
||||
echo "$TINPUT FAILED, return value $a when $TRETVAL was expected">>$TLOG;
|
||||
else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
TINPUT=test_transaction_routing3.sql
|
||||
TRETVAL=2
|
||||
a=`$RUNCMD < ./$TINPUT`
|
||||
if [ "$a" == "$TMASTER_ID" ]; then
|
||||
echo "$TINPUT FAILED, return value $a when one of the slave IDs was expected">>$TLOG;
|
||||
else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
TINPUT=test_transaction_routing3b.sql
|
||||
TRETVAL=2
|
||||
a=`$RUNCMD < ./$TINPUT`
|
||||
if [ "$a" == "$TMASTER_ID" ]; then
|
||||
echo "$TINPUT FAILED, return value $a when one of the slave IDs was expected">>$TLOG;
|
||||
else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
# test implicit transaction, that is, not started explicitly, autocommit=0
|
||||
TINPUT=test_transaction_routing4.sql
|
||||
TRETVAL=0
|
||||
a=`$RUNCMD < ./$TINPUT`
|
||||
if [ "$a" != "$TRETVAL" ]; then
|
||||
echo "$TINPUT FAILED, return value $a when $TRETVAL was expected">>$TLOG;
|
||||
else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
TINPUT=test_transaction_routing4b.sql
|
||||
TRETVAL=0
|
||||
a=`$RUNCMD < ./$TINPUT`
|
||||
if [ "$a" != "$TRETVAL" ]; then
|
||||
echo "$TINPUT FAILED, return value $a when $TRETVAL was expected">>$TLOG;
|
||||
else
|
||||
@ -113,3 +150,31 @@ if [ "$a" == "$TRETVAL" ]; then
|
||||
else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
TINPUT=test_autocommit_disabled1.sql
|
||||
TRETVAL=1
|
||||
a=`$RUNCMD < ./$TINPUT`
|
||||
if [ "$a" != "$TRETVAL" ]; then
|
||||
echo "$TINPUT FAILED, return value $a when $TRETVAL was expected">>$TLOG;
|
||||
else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
TINPUT=test_autocommit_disabled1b.sql
|
||||
TRETVAL=1
|
||||
a=`$RUNCMD < ./$TINPUT`
|
||||
if [ "$a" != "$TRETVAL" ]; then
|
||||
echo "$TINPUT FAILED, return value $a when $TRETVAL was expected">>$TLOG;
|
||||
else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
TINPUT=test_autocommit_disabled2.sql
|
||||
TRETVAL=1
|
||||
a=`$RUNCMD < ./$TINPUT`
|
||||
if [ "$a" != "$TRETVAL" ]; then
|
||||
echo "$TINPUT FAILED, return value $a when $TRETVAL was expected">>$TLOG;
|
||||
else
|
||||
echo "$TINPUT PASSED">>$TLOG ;
|
||||
fi
|
||||
|
||||
|
@ -0,0 +1,7 @@
|
||||
use test;
|
||||
drop table if exists t1;
|
||||
create table t1 (id integer);
|
||||
set autocommit=0; -- open transaction
|
||||
insert into t1 values(1); -- write to master
|
||||
select count(*) from t1; -- read from master
|
||||
drop table t1;
|
@ -0,0 +1,9 @@
|
||||
use test;
|
||||
drop table if exists t1;
|
||||
create table t1 (id integer);
|
||||
set autocommit=0; -- open transaction
|
||||
begin;
|
||||
insert into t1 values(1); -- write to master
|
||||
commit;
|
||||
select count(*) from t1; -- read from master since autocommit is disabled
|
||||
drop table t1;
|
@ -1,5 +1,5 @@
|
||||
DROP DATABASE If EXISTS FOO;
|
||||
SET autocommit=0;
|
||||
SET autocommit=1;
|
||||
BEGIN;
|
||||
CREATE DATABASE FOO; -- implicit commit
|
||||
SELECT (@@server_id) INTO @a;
|
||||
|
@ -1,7 +1,7 @@
|
||||
USE test;
|
||||
DROP TABLE IF EXISTS T1;
|
||||
DROP EVENT IF EXISTS myevent;
|
||||
SET autocommit=0;
|
||||
SET autocommit=1;
|
||||
BEGIN;
|
||||
CREATE TABLE T1 (id integer);
|
||||
CREATE EVENT myevent
|
||||
|
@ -1,6 +1,6 @@
|
||||
USE test;
|
||||
DROP TABLE IF EXISTS T1;
|
||||
SET autocommit=0;
|
||||
SET autocommit=1;
|
||||
BEGIN;
|
||||
CREATE TABLE T1 (id integer); -- implicit commit
|
||||
SELECT (@@server_id) INTO @a;
|
||||
|
@ -1,6 +1,6 @@
|
||||
USE test;
|
||||
DROP PROCEDURE IF EXISTS simpleproc;
|
||||
SET autocommit=0;
|
||||
SET autocommit=1;
|
||||
BEGIN;
|
||||
DELIMITER //
|
||||
CREATE PROCEDURE simpleproc (OUT param1 INT)
|
||||
|
@ -1,6 +1,6 @@
|
||||
USE test;
|
||||
DROP FUNCTION IF EXISTS hello;
|
||||
SET autocommit=0;
|
||||
SET autocommit=1;
|
||||
BEGIN;
|
||||
CREATE FUNCTION hello (s CHAR(20))
|
||||
RETURNS CHAR(50) DETERMINISTIC
|
||||
|
@ -1,7 +1,7 @@
|
||||
USE test;
|
||||
DROP TABLE IF EXISTS T1;
|
||||
CREATE TABLE T1 (id integer); -- implicit commit
|
||||
SET autocommit=0;
|
||||
SET autocommit=1;
|
||||
BEGIN;
|
||||
CREATE INDEX foo_t1 on T1 (id); -- implicit commit
|
||||
SELECT (@@server_id) INTO @a;
|
||||
|
@ -1,19 +1,6 @@
|
||||
USE test;
|
||||
SET autocommit = 0;
|
||||
SET @a= -1;
|
||||
SET @b= -2;
|
||||
START TRANSACTION;
|
||||
CREATE TABLE IF NOT EXISTS myCity (a int, b char(20));
|
||||
INSERT INTO myCity VALUES (1, 'Milan');
|
||||
INSERT INTO myCity VALUES (2, 'London');
|
||||
COMMIT;
|
||||
START TRANSACTION;
|
||||
DELETE FROM myCity;
|
||||
SET @a = (SELECT COUNT(*) FROM myCity);
|
||||
ROLLBACK;
|
||||
START TRANSACTION;
|
||||
SET @b = (SELECT COUNT(*) FROM myCity);
|
||||
START TRANSACTION;
|
||||
DROP TABLE myCity;
|
||||
SELECT (@a+@b) AS res;
|
||||
COMMIT;
|
||||
use test; -- in both
|
||||
drop table if exists t1;
|
||||
create table t1 (id integer);
|
||||
insert into t1 values(1); -- in master
|
||||
commit;
|
||||
select count(*) from t1; -- in slave
|
||||
|
@ -1,10 +1,7 @@
|
||||
-- Read from slave after implicit COMMIT
|
||||
USE test;
|
||||
SET autocommit = 0;
|
||||
START TRANSACTION;
|
||||
CREATE TABLE IF NOT EXISTS myCity (a int, b char(20));
|
||||
INSERT INTO myCity VALUES (1, 'Milan');
|
||||
INSERT INTO myCity VALUES (2, 'London');
|
||||
COMMIT;
|
||||
DELETE FROM myCity;
|
||||
SELECT COUNT(*) FROM myCity; -- read transaction's modifications from slave
|
||||
COMMIT;
|
||||
CREATE TABLE IF NOT EXISTS T2 (id integer);
|
||||
INSERT INTO T2 VALUES (@@server_id);
|
||||
SET AUTOCOMMIT=1;
|
||||
SELECT id from T2; -- read transaction's modifications from slave
|
||||
|
@ -0,0 +1,9 @@
|
||||
USE test;
|
||||
SET autocommit = 0;
|
||||
CREATE TABLE IF NOT EXISTS myCity (a int, b char(20));
|
||||
INSERT INTO myCity VALUES (1, 'Milan');
|
||||
INSERT INTO myCity VALUES (2, 'London');
|
||||
COMMIT;
|
||||
DELETE FROM myCity; -- implicit transaction started
|
||||
SELECT COUNT(*) FROM myCity; -- read transaction's modifications from master
|
||||
COMMIT;
|
@ -5,6 +5,7 @@
|
||||
|
||||
include ../../build_gateway.inc
|
||||
include ../../makefile.inc
|
||||
include ../../test.inc
|
||||
|
||||
TEST_ROOT := $(ROOT_PATH)/test
|
||||
PARENT_DIR := $(ROOT_PATH)/server
|
||||
@ -39,8 +40,15 @@ testall:
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo $(shell date) >> $(TESTLOG)
|
||||
@echo "Test Server Core" >> $(TESTLOG)
|
||||
$(MAKE) -C $(ROOT_PATH)/server/core testall
|
||||
@echo "Query Classifier PASSED" >> $(TESTLOG)
|
||||
$(MAKE) -C $(ROOT_PATH)/server/core testall
|
||||
@echo "Core PASSED" >> $(TESTLOG)
|
||||
@echo "" >> $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo $(shell date) >> $(TESTLOG)
|
||||
@echo "Test Modules" >> $(TESTLOG)
|
||||
$(MAKE) -C $(ROOT_PATH)/server/modules/routing testall
|
||||
@killall maxscale
|
||||
@echo "Routing modules PASSED" >> $(TESTLOG)
|
||||
|
||||
|
||||
buildtests:
|
||||
@ -51,7 +59,7 @@ else
|
||||
endif
|
||||
|
||||
runtests:
|
||||
@echo "" >> $(TESTLOG)
|
||||
@echo "" > $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo $(shell date) >> $(TESTLOG)
|
||||
@echo "Test MaxScale server" >> $(TESTLOG)
|
||||
@ -63,5 +71,4 @@ runtests:
|
||||
@sleep 5
|
||||
@echo "MaxScale server PASSED" >> $(TESTLOG)
|
||||
@echo "" >> $(TESTLOG)
|
||||
@killall maxscale
|
||||
|
||||
@cat $(TESTLOG) >> $(TEST_MAXSCALE_LOG)
|
||||
|
8
test.inc
8
test.inc
@ -32,4 +32,10 @@ TPWD :=
|
||||
# master's server_id, for example:
|
||||
# TMASTER_ID := 2
|
||||
#
|
||||
TMASTER_ID :=
|
||||
TMASTER_ID :=
|
||||
|
||||
#
|
||||
# Global test log where all log is gathered
|
||||
# TEST_MAXSCALE_LOG := $(ROOT_PATH)/test/test_maxscale.log
|
||||
#
|
||||
TEST_MAXSCALE_LOG :=
|
||||
|
@ -57,7 +57,8 @@ MAXSCALE_HOME
|
||||
log_manager,
|
||||
query_classifier,
|
||||
server,
|
||||
utils
|
||||
utils,
|
||||
modules
|
||||
|
|
||||
|- utils cleantests, buildtests, runtests, testall
|
||||
| |
|
||||
|
@ -5,11 +5,12 @@
|
||||
|
||||
include ../build_gateway.inc
|
||||
include ../makefile.inc
|
||||
include ../test.inc
|
||||
|
||||
export MAXSCALE_HOME=$(shell pwd)/MaxScale
|
||||
|
||||
CC=cc
|
||||
TESTLOG := $(shell pwd)/testmaxscale.log
|
||||
TESTLOG := $(TEST_MAXSCALE_LOG)
|
||||
|
||||
testall:
|
||||
$(MAKE) cleantests
|
||||
@ -39,7 +40,7 @@ testall:
|
||||
@echo "Test utils" >> $(TESTLOG)
|
||||
$(MAKE) -C $(ROOT_PATH)/utils testall
|
||||
@echo "Utils PASSED" >> $(TESTLOG)
|
||||
|
||||
|
||||
cleantests:
|
||||
$(DEL) *~
|
||||
$(MAKE) -C $(ROOT_PATH)/log_manager cleantests
|
||||
@ -55,9 +56,10 @@ buildtests:
|
||||
$(MAKE) -C $(ROOT_PATH)/utils buildtests
|
||||
|
||||
runtests:
|
||||
@echo "" >> $(TESTLOG)
|
||||
@echo "" > $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo $(shell date) >> $(TESTLOG)
|
||||
@echo "Test MaxScale root" >> $(TESTLOG)
|
||||
@echo "-------------------------------" >> $(TESTLOG)
|
||||
@echo "Nothing to run here so far" >> $(TESTLOG)
|
||||
|
Loading…
x
Reference in New Issue
Block a user