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

Merge branch 'master' of https://github.com/skysql/MaxScale

Browse Source
This commit is contained in:
VilhoRaatikka 2014-05-08 16:37:49 +03:00
commit 391ba19609
9 changed files with 221 additions and 282 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
log_manager/log_manager.cc
View File

@ -253,6 +253,7 @@ static int logmanager_write_log(
va_list valist);
static blockbuf_t* blockbuf_init(logfile_id_t id);
static void blockbuf_node_done(void* bb_data);
static char* blockbuf_get_writepos(
#if 0
int** refcount,
@ -996,8 +997,13 @@ static char* blockbuf_get_writepos(
simple_mutex_unlock(&bb->bb_mutex);
return pos;
}
static void blockbuf_node_done(
void* bb_data)
{
blockbuf_t* bb = (blockbuf_t *)bb_data;
simple_mutex_done(&bb->bb_mutex);
}
static blockbuf_t* blockbuf_init(
@ -2059,7 +2065,7 @@ static bool logfile_init(
if (mlist_init(&logfile->lf_blockbuf_list,
NULL,
strdup("logfile block buffer list"),
NULL,
blockbuf_node_done,
MAXNBLOCKBUFS) == NULL)
{
ss_dfprintf(stderr,

5
query_classifier/query_classifier.cc
View File

@ -706,6 +706,10 @@ static bool skygw_stmt_causes_implicit_commit(
{
succp = true;
}
else
{
succp =false;
}
break;
default:
succp = true;
@ -764,7 +768,6 @@ static int is_autocommit_stmt(
{
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)))
{

13
server/core/config.c
View File

@ -203,6 +203,19 @@ int error_count = 0;
char *enable_root_user =
config_get_value(obj->parameters, "enable_root_user");
if (obj->element == NULL) /*< if module load failed */
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Reading configuration "
"for router service '%s' failed. "
"Router %s is not loaded.",
obj->object,
obj->object)));
obj = obj->next;
continue; /*< process next obj */
}
if (enable_root_user)
serviceEnableRootUser(obj->element, atoi(enable_root_user));

23
server/core/load_utils.c
View File

@ -53,6 +53,17 @@ static void register_module(const char *module,
void *modobj);
static void unregister_module(const char *module);
char* get_maxscale_home(void)
{
char* home = getenv("MAXSCALE_HOME");
if (home == NULL)
{
home = "/usr/local/skysql/MaxScale";
}
return home;
}
/**
* Load the dynamic library related to a gateway module. The routine
* will look for library files in the current directory,
@ -82,10 +93,10 @@ MODULES *mod;
sprintf(fname, "./lib%s.so", module);
if (access(fname, F_OK) == -1)
{
if ((home = getenv("MAXSCALE_HOME")) == NULL)
home = "/usr/local/skysql/MaxScale";
home = get_maxscale_home ();
sprintf(fname, "%s/modules/lib%s.so", home, module);
if (access(fname, F_OK) == -1)
if (access(fname, F_OK) == -1)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
@ -100,7 +111,7 @@ MODULES *mod;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to load library for module: "
"%s, %s.",
"%s\n\t\t\t %s.",
module,
dlerror())));
return NULL;
@ -111,7 +122,7 @@ MODULES *mod;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Version interface not supported by "
"module: %s, %s.",
"module: %s\n\t\t\t %s.",
module,
dlerror())));
dlclose(dlhandle);
@ -134,7 +145,7 @@ MODULES *mod;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Expected entry point interface missing "
"from module: %s, %s.",
"from module: %s\n\t\t\t %s.",
module,
dlerror())));
dlclose(dlhandle);

16
server/core/service.c
View File

@ -34,6 +34,8 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <ctype.h>
#include <errno.h>
#include <session.h>
#include <service.h>
#include <server.h>
@ -70,6 +72,20 @@ SERVICE *service;
return NULL;
if ((service->router = load_module(router, MODULE_ROUTER)) == NULL)
{
char* home = get_maxscale_home();
char* ldpath = getenv("LD_LIBRARY_PATH");
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to load %s module \"%s\".\n\t\t\t"
" Ensure that lib%s.so exists in one of the "
"following directories :\n\t\t\t "
"- %s/modules\n\t\t\t - %s",
MODULE_ROUTER,
router,
router,
home,
ldpath)));
free(service);
return NULL;
}

126
server/include/gateway_mysql.h
View File

@ -1,126 +0,0 @@
/*
* This file is distributed as part of the SkySQL Gateway. It is free
* software: you can redistribute it and/or modify it under the terms of the
* GNU General Public License as published by the Free Software Foundation,
* version 2.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program; if not, write to the Free Software Foundation, Inc., 51
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright SkySQL Ab 2013
*
*/
/*
* MYSQL mysql protocol header file
* Revision History
*
* Date Who Description
* 10/06/13 Massimiliano Pinto Initial implementation
*
*/
#include <dcb.h>
/* Protocol packing macros. */
#define gw_mysql_set_byte2(__buffer, __int) do { \
(__buffer)[0]= (uint8_t)((__int) & 0xFF); \
(__buffer)[1]= (uint8_t)(((__int) >> 8) & 0xFF); } while (0)
#define gw_mysql_set_byte3(__buffer, __int) do { \
(__buffer)[0]= (uint8_t)((__int) & 0xFF); \
(__buffer)[1]= (uint8_t)(((__int) >> 8) & 0xFF); \
(__buffer)[2]= (uint8_t)(((__int) >> 16) & 0xFF); } while (0)
#define gw_mysql_set_byte4(__buffer, __int) do { \
(__buffer)[0]= (uint8_t)((__int) & 0xFF); \
(__buffer)[1]= (uint8_t)(((__int) >> 8) & 0xFF); \
(__buffer)[2]= (uint8_t)(((__int) >> 16) & 0xFF); \
(__buffer)[3]= (uint8_t)(((__int) >> 24) & 0xFF); } while (0)
/* Protocol unpacking macros. */
#define gw_mysql_get_byte2(__buffer) \
(uint16_t)((__buffer)[0] | \
((__buffer)[1] << 8))
#define gw_mysql_get_byte3(__buffer) \
(uint32_t)((__buffer)[0] | \
((__buffer)[1] << 8) | \
((__buffer)[2] << 16))
#define gw_mysql_get_byte4(__buffer) \
(uint32_t)((__buffer)[0] | \
((__buffer)[1] << 8) | \
((__buffer)[2] << 16) | \
((__buffer)[3] << 24))
#define gw_mysql_get_byte8(__buffer) \
((uint64_t)(__buffer)[0] | \
((uint64_t)(__buffer)[1] << 8) | \
((uint64_t)(__buffer)[2] << 16) | \
((uint64_t)(__buffer)[3] << 24) | \
((uint64_t)(__buffer)[4] << 32) | \
((uint64_t)(__buffer)[5] << 40) | \
((uint64_t)(__buffer)[6] << 48) | \
((uint64_t)(__buffer)[7] << 56))
typedef enum
{
GW_MYSQL_CAPABILITIES_NONE= 0,
GW_MYSQL_CAPABILITIES_LONG_PASSWORD= (1 << 0),
GW_MYSQL_CAPABILITIES_FOUND_ROWS= (1 << 1),
GW_MYSQL_CAPABILITIES_LONG_FLAG= (1 << 2),
GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB= (1 << 3),
GW_MYSQL_CAPABILITIES_NO_SCHEMA= (1 << 4),
GW_MYSQL_CAPABILITIES_COMPRESS= (1 << 5),
GW_MYSQL_CAPABILITIES_ODBC= (1 << 6),
GW_MYSQL_CAPABILITIES_LOCAL_FILES= (1 << 7),
GW_MYSQL_CAPABILITIES_IGNORE_SPACE= (1 << 8),
GW_MYSQL_CAPABILITIES_PROTOCOL_41= (1 << 9),
GW_MYSQL_CAPABILITIES_INTERACTIVE= (1 << 10),
GW_MYSQL_CAPABILITIES_SSL= (1 << 11),
GW_MYSQL_CAPABILITIES_IGNORE_SIGPIPE= (1 << 12),
GW_MYSQL_CAPABILITIES_TRANSACTIONS= (1 << 13),
GW_MYSQL_CAPABILITIES_RESERVED= (1 << 14),
GW_MYSQL_CAPABILITIES_SECURE_CONNECTION= (1 << 15),
GW_MYSQL_CAPABILITIES_MULTI_STATEMENTS= (1 << 16),
GW_MYSQL_CAPABILITIES_MULTI_RESULTS= (1 << 17),
GW_MYSQL_CAPABILITIES_PS_MULTI_RESULTS= (1 << 18),
GW_MYSQL_CAPABILITIES_PLUGIN_AUTH= (1 << 19),
GW_MYSQL_CAPABILITIES_SSL_VERIFY_SERVER_CERT= (1 << 30),
GW_MYSQL_CAPABILITIES_REMEMBER_OPTIONS= (1 << 31),
GW_MYSQL_CAPABILITIES_CLIENT= (GW_MYSQL_CAPABILITIES_LONG_PASSWORD |
GW_MYSQL_CAPABILITIES_FOUND_ROWS |
GW_MYSQL_CAPABILITIES_LONG_FLAG |
GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB |
GW_MYSQL_CAPABILITIES_LOCAL_FILES |
GW_MYSQL_CAPABILITIES_PLUGIN_AUTH |
GW_MYSQL_CAPABILITIES_TRANSACTIONS |
GW_MYSQL_CAPABILITIES_PROTOCOL_41 |
GW_MYSQL_CAPABILITIES_MULTI_STATEMENTS |
GW_MYSQL_CAPABILITIES_MULTI_RESULTS |
GW_MYSQL_CAPABILITIES_PS_MULTI_RESULTS |
GW_MYSQL_CAPABILITIES_SECURE_CONNECTION),
GW_MYSQL_CAPABILITIES_CLIENT_COMPRESS= (GW_MYSQL_CAPABILITIES_LONG_PASSWORD |
GW_MYSQL_CAPABILITIES_FOUND_ROWS |
GW_MYSQL_CAPABILITIES_LONG_FLAG |
GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB |
GW_MYSQL_CAPABILITIES_LOCAL_FILES |
GW_MYSQL_CAPABILITIES_PLUGIN_AUTH |
GW_MYSQL_CAPABILITIES_TRANSACTIONS |
GW_MYSQL_CAPABILITIES_PROTOCOL_41 |
GW_MYSQL_CAPABILITIES_MULTI_STATEMENTS |
GW_MYSQL_CAPABILITIES_MULTI_RESULTS |
GW_MYSQL_CAPABILITIES_PS_MULTI_RESULTS |
GW_MYSQL_CAPABILITIES_COMPRESS
),
} gw_mysql_capabilities_t;
#define SMALL_CHUNK 1024
#define MAX_CHUNK SMALL_CHUNK * 8 * 4
#define ToHex(Y) (Y>='0'&&Y<='9'?Y-'0':Y-'A'+10)
extern int mysql_send_ok(DCB *, int, int, const char *);
extern int MySQLSendHandshake(DCB *);

2
server/include/modules.h
View File

@ -55,4 +55,6 @@ extern void *load_module(const char *module, const char *type);
extern void unload_module(const char *module);
extern void printModules();
extern void dprintAllModules(DCB *);
char* get_maxscale_home(void);
#endif

6
server/modules/include/readwritesplit.h
View File

@ -122,6 +122,8 @@ struct router_client_session {
/*< cursor is pointer and status variable to current session command */
sescmd_cursor_t rses_cursor[BE_COUNT];
int rses_capabilities; /*< input type, for example */
bool rses_autocommit_enabled;
bool rses_transaction_active;
struct router_client_session* next;
#if defined(SS_DEBUG)
skygw_chk_t rses_chk_tail;
@ -155,4 +157,8 @@ typedef struct router_instance {
struct router_instance* next; /*< Next router on the list */
} ROUTER_INSTANCE;
#define BACKEND_TYPE(b) (SERVER_IS_MASTER((b)->backend_server) ? BE_MASTER : \
(SERVER_IS_SLAVE((b)->backend_server) ? BE_SLAVE : \
(SERVER_IS_JOINED((b)->backend_server) ? BE_JOINED : BE_UNDEFINED)));
#endif /*< _RWSPLITROUTER_H */

302
server/modules/routing/readwritesplit/readwritesplit.c
View File

@ -74,7 +74,7 @@ static uint8_t getCapabilities (ROUTER* inst, void* router_session);
static bool search_backend_servers(
BACKEND** p_master,
BACKEND** p_slave,
ROUTER_INSTANCE* router);
ROUTER_INSTANCE* router);
static ROUTER_OBJECT MyObject = {
createInstance,
@ -218,10 +218,10 @@ static ROUTER* createInstance(
SERVICE* service,
char** options)
{
ROUTER_INSTANCE* router;
SERVER* server;
ROUTER_INSTANCE* router;
SERVER* server;
int n;
int i;
int i;
if ((router = calloc(1, sizeof(ROUTER_INSTANCE))) == NULL) {
return NULL;
@ -251,7 +251,6 @@ static ROUTER* createInstance(
"module but none are supported. The options will be "
"ignored.")));
}
/**
* Create an array of the backend servers in the router structure to
* maintain a count of the number of connections to each
@ -273,7 +272,7 @@ static ROUTER* createInstance(
router->servers[n]->backend_conn_count = 0;
n += 1;
server = server->nextdb;
}
}
router->servers[n] = NULL;
/**
@ -307,7 +306,7 @@ static ROUTER* createInstance(
}
}
}
/**
/**
* We have completed the creation of the router data, so now
* insert this router into the linked list of routers
* that have been created with this module.
@ -336,12 +335,12 @@ static void* newSession(
{
BACKEND* local_backend[BE_COUNT];
ROUTER_CLIENT_SES* client_rses;
ROUTER_INSTANCE* router = (ROUTER_INSTANCE *)router_inst;
ROUTER_INSTANCE* router = (ROUTER_INSTANCE *)router_inst;
bool succp;
client_rses =
(ROUTER_CLIENT_SES *)calloc(1, sizeof(ROUTER_CLIENT_SES));
if (client_rses == NULL)
{
ss_dassert(false);
@ -352,7 +351,7 @@ static void* newSession(
#if defined(SS_DEBUG)
client_rses->rses_chk_top = CHK_NUM_ROUTER_SES;
client_rses->rses_chk_tail = CHK_NUM_ROUTER_SES;
#endif
#endif
/** store pointers to sescmd list to both cursors */
client_rses->rses_cursor[BE_MASTER].scmd_cur_rses = client_rses;
client_rses->rses_cursor[BE_MASTER].scmd_cur_active = false;
@ -367,8 +366,8 @@ static void* newSession(
&client_rses->rses_properties[RSES_PROP_TYPE_SESCMD];
client_rses->rses_cursor[BE_SLAVE].scmd_cur_cmd = NULL;
client_rses->rses_cursor[BE_SLAVE].scmd_cur_be_type = BE_SLAVE;
/**
/**
* Find a backend server to connect to. This is the extent of the
* load balancing algorithm we need to implement for this simple
* connection router.
@ -376,7 +375,7 @@ static void* newSession(
succp = search_backend_servers(&local_backend[BE_MASTER],
&local_backend[BE_SLAVE],
router);
/** Both Master and Slave must be found */
if (!succp) {
free(client_rses);
@ -392,23 +391,23 @@ static void* newSession(
if (client_rses->rses_dcb[BE_SLAVE] == NULL) {
ss_dassert(session->refcount == 1);
free(client_rses);
free(client_rses);
return NULL;
}
/**
}
/**
* Open the master connection.
*/
*/
client_rses->rses_dcb[BE_MASTER] = dcb_connect(
local_backend[BE_MASTER]->backend_server,
session,
local_backend[BE_MASTER]->backend_server->protocol);
if (client_rses->rses_dcb[BE_MASTER] == NULL)
{
{
/** Close slave connection first. */
client_rses->rses_dcb[BE_SLAVE]->func.close(client_rses->rses_dcb[BE_SLAVE]);
free(client_rses);
return NULL;
}
}
/**
* We now have a master and a slave server with the least connections.
* Bump the connection counts for these servers.
@ -418,7 +417,7 @@ static void* newSession(
client_rses->rses_backend[BE_SLAVE] = local_backend[BE_SLAVE];
client_rses->rses_backend[BE_MASTER] = local_backend[BE_MASTER];
router->stats.n_sessions += 1;
router->stats.n_sessions += 1;
client_rses->rses_capabilities = RCAP_TYPE_STMT_INPUT;
/**
@ -430,15 +429,17 @@ static void* newSession(
* Add this session to end of the list of active sessions in router.
*/
spinlock_acquire(&router->lock);
client_rses->next = router->connections;
client_rses->next = router->connections;
router->connections = client_rses;
spinlock_release(&router->lock);
CHK_CLIENT_RSES(client_rses);
CHK_CLIENT_RSES(client_rses);
return (void *)client_rses;
}
/**
* Close a session with the router, this is the mechanism
* by which a router may cleanup data structure etc.
@ -453,7 +454,7 @@ static void closeSession(
ROUTER_CLIENT_SES* router_cli_ses;
DCB* slave_dcb;
DCB* master_dcb;
router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
CHK_CLIENT_RSES(router_cli_ses);
/**
@ -464,7 +465,7 @@ static void closeSession(
/** decrease server current connection counters */
atomic_add(&router_cli_ses->rses_backend[BE_SLAVE]->backend_server->stats.n_current, -1);
atomic_add(&router_cli_ses->rses_backend[BE_MASTER]->backend_server->stats.n_current, -1);
slave_dcb = router_cli_ses->rses_dcb[BE_SLAVE];
router_cli_ses->rses_dcb[BE_SLAVE] = NULL;
master_dcb = router_cli_ses->rses_dcb[BE_MASTER];
@ -473,14 +474,14 @@ static void closeSession(
router_cli_ses->rses_closed = true;
/** Unlock */
rses_end_locked_router_action(router_cli_ses);
/**
/**
* Close the backend server connections
*/
if (slave_dcb != NULL) {
CHK_DCB(slave_dcb);
slave_dcb->func.close(slave_dcb);
}
}
if (master_dcb != NULL) {
master_dcb->func.close(master_dcb);
@ -498,8 +499,8 @@ static void freeSession(
int i;
router_cli_ses = (ROUTER_CLIENT_SES *)router_client_session;
router = (ROUTER_INSTANCE *)router_instance;
router = (ROUTER_INSTANCE *)router_instance;
atomic_add(&router_cli_ses->rses_backend[BE_SLAVE]->backend_conn_count, -1);
atomic_add(&router_cli_ses->rses_backend[BE_MASTER]->backend_conn_count, -1);
@ -567,15 +568,15 @@ static int routeQuery(
void* router_session,
GWBUF* querybuf)
{
skygw_query_type_t qtype = QUERY_TYPE_UNKNOWN;
GWBUF* plainsqlbuf = NULL;
char* querystr = NULL;
skygw_query_type_t qtype = QUERY_TYPE_UNKNOWN;
GWBUF* plainsqlbuf = NULL;
char* querystr = NULL;
char* startpos;
unsigned char packet_type;
uint8_t* packet;
int ret = 0;
DCB* master_dcb = NULL;
DCB* slave_dcb = NULL;
DCB* master_dcb = NULL;
DCB* slave_dcb = NULL;
ROUTER_INSTANCE* inst = (ROUTER_INSTANCE *)instance;
ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
bool rses_is_closed;
@ -618,16 +619,16 @@ static int routeQuery(
"Error: Failed to route %s:%s:\"%s\" to "
"backend server. %s.",
STRPACKETTYPE(packet_type),
STRQTYPE(qtype),
(querystr == NULL ? "(empty)" : querystr),
(rses_is_closed ? "Router was closed" :
"Router has no backend servers where to "
"route to"))));
STRQTYPE(qtype),
(querystr == NULL ? "(empty)" : querystr),
(rses_is_closed ? "Router was closed" :
"Router has no backend servers where to "
"route to"))));
goto return_ret;
}
inst->stats.n_queries++;
startpos = (char *)&packet[5];
switch(packet_type) {
case COM_QUIT: /**< 1 QUIT will close all sessions */
case COM_INIT_DB: /**< 2 DDL must go to the master */
@ -656,7 +657,7 @@ static int routeQuery(
querystr = master_dcb->func.getquerystr(
(void *) gwbuf_clone(querybuf),
&querystr_is_copy);
*/
*/
qtype = skygw_query_classifier_get_type(querystr, 0);
break;
@ -683,88 +684,94 @@ static int routeQuery(
* transaction becomes active and master gets all statements until
* transaction is committed and autocommit is enabled again.
*/
if (autocommit_enabled &&
if (router_cli_ses->rses_autocommit_enabled &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_DISABLE_AUTOCOMMIT))
{
autocommit_enabled = false;
router_cli_ses->rses_autocommit_enabled = false;
if (!transaction_active)
if (!router_cli_ses->rses_transaction_active)
{
transaction_active = true;
router_cli_ses->rses_transaction_active = true;
}
}
else if (!transaction_active &&
else if (!router_cli_ses->rses_transaction_active &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_BEGIN_TRX))
{
transaction_active = true;
router_cli_ses->rses_transaction_active = true;
}
/**
* Explicit COMMIT and ROLLBACK, implicit COMMIT.
*/
if (autocommit_enabled &&
transaction_active &&
if (router_cli_ses->rses_autocommit_enabled &&
router_cli_ses->rses_transaction_active &&
(QUERY_IS_TYPE(qtype,QUERY_TYPE_COMMIT) ||
QUERY_IS_TYPE(qtype,QUERY_TYPE_ROLLBACK)))
{
transaction_active = false;
router_cli_ses->rses_transaction_active = false;
}
else if (!autocommit_enabled &&
else if (!router_cli_ses->rses_autocommit_enabled &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_ENABLE_AUTOCOMMIT))
{
autocommit_enabled = true;
transaction_active = false;
router_cli_ses->rses_autocommit_enabled = true;
router_cli_ses->rses_transaction_active = false;
}
/**
* 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))
bool succp = route_session_write(
router_cli_ses,
querybuf,
inst,
packet_type,
qtype);
if (succp)
{
ret = 1;
}
else
{
ret = 0;
}
ss_dassert(succp);
ss_dassert(ret == 1);
goto return_ret;
}
else if (transaction_active) /*< all to master */
else if (QUERY_IS_TYPE(qtype, QUERY_TYPE_READ) &&
!router_cli_ses->rses_transaction_active)
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Transaction is active, routing to Master.")));
ret = master_dcb->func.write(master_dcb, querybuf);
atomic_add(&inst->stats.n_master, 1);
bool succp;
goto return_ret;
}
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);
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.")));
bool succp = true;
if (LOG_IS_ENABLED(LOGFILE_TRACE))
{
if (router_cli_ses->rses_transaction_active) /*< all to master */
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Transaction is active, routing to Master.")));
}
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);
atomic_add(&inst->stats.n_master, 1);
goto return_ret;
}
@ -934,8 +941,7 @@ static void clientReply(
client_dcb = backend_dcb->session->client;
/** Unlock */
rses_end_locked_router_action(router_cli_ses);
rses_end_locked_router_action(router_cli_ses);
/**
* 1. Check if backend received reply to sescmd.
* 2. Check sescmd's state whether OK_PACKET has been
@ -955,14 +961,14 @@ static void clientReply(
}
if (backend_dcb == master_dcb)
{
{
be_type = BE_MASTER;
}
}
else if (backend_dcb == slave_dcb)
{
{
be_type = BE_SLAVE;
}
LOGIF(LT, tracelog_routed_query(router_cli_ses,
}
LOGIF(LT, tracelog_routed_query(router_cli_ses,
"reply_by_statement",
backend_dcb,
gwbuf_clone(writebuf)));
@ -983,8 +989,7 @@ static void clientReply(
{
writebuf = sescmd_cursor_process_replies(client_dcb,
writebuf,
scur);
scur);
}
/** Unlock router session */
rses_end_locked_router_action(router_cli_ses);
@ -1021,7 +1026,7 @@ lock_failed:
*
* @param inst - in, use
* Pointer to router instance
*
*
* @return true, if all what what requested found, false if the request
* was not satisfied or was partially satisfied.
*
@ -1032,12 +1037,12 @@ lock_failed:
static bool search_backend_servers(
BACKEND** p_master,
BACKEND** p_slave,
ROUTER_INSTANCE* router)
{
ROUTER_INSTANCE* router)
{
BACKEND* local_backend[BE_COUNT] = {NULL,NULL};
int i;
int i;
bool succp = true;
/*
* Loop over all the servers and find any that have fewer connections
* than current candidate server.
@ -1052,32 +1057,32 @@ static bool search_backend_servers(
* very low load.
*
* If master is searched for, the first master found is chosen.
*/
*/
for (i = 0; router->servers[i] != NULL; i++) {
BACKEND* be = router->servers[i];
if (be != NULL) {
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"%lu [search_backend_servers] Examine server "
"%s:%d with %d connections. Status is %d, "
"router->bitvalue is %d",
pthread_self(),
"%s:%d with %d connections. Status is %d, "
"router->bitvalue is %d",
pthread_self(),
be->backend_server->name,
be->backend_server->port,
be->backend_conn_count,
be->backend_server->status,
router->bitmask)));
router->bitmask)));
}
if (be != NULL &&
SERVER_IS_RUNNING(be->backend_server) &&
(be->backend_server->status & router->bitmask) ==
router->bitvalue)
{
{
if (SERVER_IS_SLAVE(be->backend_server) &&
p_slave != NULL)
{
{
/**
* If no candidate set, set first running
* server as an initial candidate server.
@ -1085,7 +1090,7 @@ static bool search_backend_servers(
if (local_backend[BE_SLAVE] == NULL)
{
local_backend[BE_SLAVE] = be;
}
}
else if (be->backend_conn_count <
local_backend[BE_SLAVE]->backend_conn_count)
{
@ -1101,74 +1106,74 @@ static bool search_backend_servers(
be->backend_server->stats.n_connections <
local_backend[BE_SLAVE]->backend_server->stats.n_connections)
{
/**
/**
* This running server has the same
* number of connections currently
* as the candidate but has had
* fewer connections over time
* than candidate, set this server
* to candidate.
*/
*/
local_backend[BE_SLAVE] = be;
}
}
}
}
else if (p_master != NULL &&
local_backend[BE_MASTER] == NULL &&
SERVER_IS_MASTER(be->backend_server))
{
{
local_backend[BE_MASTER] = be;
}
}
else if (p_master != NULL &&
local_backend[BE_JOINED] == NULL &&
SERVER_IS_JOINED(be->backend_server))
{
{
local_backend[BE_JOINED] = be;
}
}
}
}
}
}
if (router->bitvalue != 0 &&
p_master != NULL &&
local_backend[BE_JOINED] == NULL)
{
{
succp = false;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
LOGFILE_ERROR,
"Error : Couldn't find a Joined Galera node from %d "
"candidates.",
"candidates.",
i)));
goto return_succp;
}
if (p_slave != NULL && local_backend[BE_SLAVE] == NULL) {
succp = false;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Couldn't find suitable Slave from %d "
"candidates.",
"candidates.",
i)));
}
if (p_master != NULL && local_backend[BE_MASTER] == NULL) {
succp = false;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Couldn't find suitable Master from %d "
"candidates.",
"candidates.",
i)));
}
}
if (local_backend[BE_SLAVE] != NULL) {
*p_slave = local_backend[BE_SLAVE];
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"%lu [readwritesplit:search_backend_servers] Selected "
"Slave %s:%d from %d candidates.",
pthread_self(),
local_backend[BE_SLAVE]->backend_server->name,
local_backend[BE_SLAVE]->backend_server->port,
i)));
}
}
if (local_backend[BE_MASTER] != NULL) {
*p_master = local_backend[BE_MASTER];
LOGIF(LT, (skygw_log_write(
@ -1528,7 +1533,7 @@ static bool execute_sescmd_in_backend(
sescmd_cursor_t* scur;
dcb = rses->rses_dcb[be_type];
CHK_DCB(dcb);
CHK_CLIENT_RSES(rses);
ss_dassert(SPINLOCK_IS_LOCKED(&rses->rses_lock));
@ -1563,7 +1568,7 @@ static bool execute_sescmd_in_backend(
dcb->session,
sescmd_cursor_clone_querybuf(scur));
break;
case COM_QUIT:
case COM_QUERY:
case COM_INIT_DB:
@ -1588,6 +1593,7 @@ return_succp:
return succp;
}
/**
* Moves cursor to next property and copied address of its sescmd to cursor.
* Current propery must be non-null.
@ -1665,7 +1671,6 @@ static rses_property_t* mysql_sescmd_get_property(
return scmd->my_sescmd_prop;
}
static void tracelog_routed_query(
ROUTER_CLIENT_SES* rses,
char* funcname,
@ -1679,7 +1684,7 @@ static void tracelog_routed_query(
char* querystr;
char* startpos = (char *)&packet[5];
backend_type_t be_type;
if (rses->rses_dcb[BE_MASTER] == dcb)
{
be_type = BE_MASTER;
@ -1722,11 +1727,13 @@ static void tracelog_routed_query(
-1)),
STRBETYPE(be_type),
dcb)));
free(querystr);
}
}
gwbuf_free(buf);
}
/**
* Return rc, rc < 0 if router session is closed. rc == 0 if there are no
* capabilities specified, rc > 0 when there are capabilities.
@ -1774,7 +1781,7 @@ static bool route_session_write(
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);
@ -1795,12 +1802,12 @@ static bool route_session_write(
{
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;
@ -1822,16 +1829,16 @@ static bool route_session_write(
}
/** 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;
}
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)
@ -1839,7 +1846,7 @@ static bool route_session_write(
/** 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);
@ -1847,4 +1854,5 @@ static bool route_session_write(
return_succp:
return succp;
}
}