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

Merge branch 'develop' into MAX-111

Browse Source
This commit is contained in:
Mark Riddoch
2014-06-26 16:52:57 +01:00
parent 031294f655 10ac2cb6e6
commit ace2550d93
17 changed files with 1273 additions and 484 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
maxscale.spec
View File

@ -39,9 +39,11 @@ ln -s /lib64/libaio.so.1 /lib64/libaio.so
%install %install
mkdir -p $RPM_BUILD_ROOT/etc/ld.so.conf.d/ mkdir -p $RPM_BUILD_ROOT/etc/ld.so.conf.d/
mkdir -p $RPM_BUILD_ROOT/etc/init.d/
mkdir -p $RPM_BUILD_ROOT%{install_path} mkdir -p $RPM_BUILD_ROOT%{install_path}
cp -r binaries/* $RPM_BUILD_ROOT%{install_path} cp -r binaries/* $RPM_BUILD_ROOT%{install_path}
cp maxscale.conf $RPM_BUILD_ROOT/etc/ld.so.conf.d/ cp maxscale.conf $RPM_BUILD_ROOT/etc/ld.so.conf.d/
cp -r etc/init.d/maxscale $RPM_BUILD_ROOT/etc/init.d/
%clean %clean
@ -49,5 +51,6 @@ cp maxscale.conf $RPM_BUILD_ROOT/etc/ld.so.conf.d/
%defattr(-,root,root) %defattr(-,root,root)
%{install_path} %{install_path}
/etc/ld.so.conf.d/maxscale.conf /etc/ld.so.conf.d/maxscale.conf
/etc/inid.d/maxscale
%changelog %changelog

55
query_classifier/query_classifier.cc
View File

@ -101,7 +101,8 @@ static int is_autocommit_stmt(
*/ */
skygw_query_type_t skygw_query_classifier_get_type( skygw_query_type_t skygw_query_classifier_get_type(
const char* query, const char* query,
unsigned long client_flags) unsigned long client_flags,
MYSQL** p_mysql)
{ {
MYSQL* mysql; MYSQL* mysql;
char* query_str; char* query_str;
@ -129,9 +130,13 @@ skygw_query_type_t skygw_query_classifier_get_type(
mysql_error(mysql)))); mysql_error(mysql))));
mysql_library_end(); mysql_library_end();
goto return_without_server; goto return_qtype;
} }
if (p_mysql != NULL)
{
*p_mysql = mysql;
}
/** Set methods and authentication to mysql */ /** Set methods and authentication to mysql */
mysql_options(mysql, MYSQL_READ_DEFAULT_GROUP, "libmysqld_skygw"); mysql_options(mysql, MYSQL_READ_DEFAULT_GROUP, "libmysqld_skygw");
mysql_options(mysql, MYSQL_OPT_USE_EMBEDDED_CONNECTION, NULL); mysql_options(mysql, MYSQL_OPT_USE_EMBEDDED_CONNECTION, NULL);
@ -143,25 +148,39 @@ skygw_query_type_t skygw_query_classifier_get_type(
/** Get one or create new THD object to be use in parsing */ /** Get one or create new THD object to be use in parsing */
thd = get_or_create_thd_for_parsing(mysql, query_str); thd = get_or_create_thd_for_parsing(mysql, query_str);
if (thd == NULL) { if (thd == NULL)
goto return_with_server_handle; {
skygw_query_classifier_free(mysql);
} }
/** Create parse_tree inside thd */ /** Create parse_tree inside thd */
failp = create_parse_tree(thd); failp = create_parse_tree(thd);
if (failp) { if (failp)
goto return_with_thd; {
skygw_query_classifier_free(mysql);
*p_mysql = NULL;
} }
qtype = resolve_query_type(thd); qtype = resolve_query_type(thd);
return_with_thd: if (p_mysql == NULL)
{
skygw_query_classifier_free(mysql);
}
return_qtype:
return qtype;
}
void skygw_query_classifier_free(
MYSQL* mysql)
{
if (mysql->thd != NULL)
{
(*mysql->methods->free_embedded_thd)(mysql); (*mysql->methods->free_embedded_thd)(mysql);
mysql->thd = 0; mysql->thd = NULL;
return_with_server_handle: }
mysql_close(mysql); mysql_close(mysql);
mysql_thread_end(); mysql_thread_end();
return_without_server:
return qtype;
} }
@ -492,6 +511,7 @@ static skygw_query_type_t resolve_query_type(
} }
/**<! fall through */ /**<! fall through */
case SQLCOM_CHANGE_DB: case SQLCOM_CHANGE_DB:
case SQLCOM_DEALLOCATE_PREPARE:
type |= QUERY_TYPE_SESSION_WRITE; type |= QUERY_TYPE_SESSION_WRITE;
break; break;
@ -518,6 +538,11 @@ static skygw_query_type_t resolve_query_type(
goto return_qtype; goto return_qtype;
break; break;
case SQLCOM_PREPARE:
type |= QUERY_TYPE_PREPARE_NAMED_STMT;
goto return_qtype;
break;
default: default:
break; break;
} }
@ -783,3 +808,11 @@ static int is_autocommit_stmt(
return_rc: return_rc:
return rc; return rc;
} }
char* skygw_query_classifier_get_stmtname(
MYSQL* mysql)
{
return ((THD *)(mysql->thd))->lex->prepared_stmt_name.str;
}

36
query_classifier/query_classifier.h
View File

@ -19,6 +19,7 @@ Copyright SkySQL Ab
/** getpid */ /** getpid */
#include <unistd.h> #include <unistd.h>
#include <mysql.h>
#include "../utils/skygw_utils.h" #include "../utils/skygw_utils.h"
EXTERN_C_BLOCK_BEGIN EXTERN_C_BLOCK_BEGIN
@ -29,25 +30,36 @@ EXTERN_C_BLOCK_BEGIN
* is modified * is modified
*/ */
typedef enum { typedef enum {
QUERY_TYPE_UNKNOWN = 0x000, /*< Initial value, can't be tested bitwisely */ QUERY_TYPE_UNKNOWN = 0x0000, /*< Initial value, can't be tested bitwisely */
QUERY_TYPE_LOCAL_READ = 0x001, /*< Read non-database data, execute in MaxScale */ QUERY_TYPE_LOCAL_READ = 0x0001, /*< Read non-database data, execute in MaxScale */
QUERY_TYPE_READ = 0x002, /*< No updates */ QUERY_TYPE_READ = 0x0002, /*< No updates */
QUERY_TYPE_WRITE = 0x004, /*< Master data will be modified */ QUERY_TYPE_WRITE = 0x0004, /*< Master data will be modified */
QUERY_TYPE_SESSION_WRITE = 0x008, /*< Session data will be modified */ QUERY_TYPE_SESSION_WRITE = 0x0008, /*< Session data will be modified */
QUERY_TYPE_GLOBAL_WRITE = 0x010, /*< Global system variable modification */ QUERY_TYPE_GLOBAL_WRITE = 0x0010, /*< Global system variable modification */
QUERY_TYPE_BEGIN_TRX = 0x020, /*< BEGIN or START TRANSACTION */ QUERY_TYPE_BEGIN_TRX = 0x0020, /*< BEGIN or START TRANSACTION */
QUERY_TYPE_ENABLE_AUTOCOMMIT = 0x040,/*< SET autocommit=1 */ QUERY_TYPE_ENABLE_AUTOCOMMIT = 0x0040, /*< SET autocommit=1 */
QUERY_TYPE_DISABLE_AUTOCOMMIT = 0x080,/*< SET autocommit=0 */ QUERY_TYPE_DISABLE_AUTOCOMMIT = 0x0080, /*< SET autocommit=0 */
QUERY_TYPE_ROLLBACK = 0x100, /*< ROLLBACK */ QUERY_TYPE_ROLLBACK = 0x0100, /*< ROLLBACK */
QUERY_TYPE_COMMIT = 0x200 /*< COMMIT */ QUERY_TYPE_COMMIT = 0x0200, /*< COMMIT */
QUERY_TYPE_PREPARE_NAMED_STMT = 0x0400, /*< Prepared stmt with name from user */
QUERY_TYPE_PREPARE_STMT = 0x0800, /*< Prepared stmt with id provided by server */
QUERY_TYPE_EXEC_STMT = 0x1000 /*< Execute prepared statement */
} skygw_query_type_t; } skygw_query_type_t;
#define QUERY_IS_TYPE(mask,type) ((mask & type) == type) #define QUERY_IS_TYPE(mask,type) ((mask & type) == type)
/**
* Create THD and use it for creating parse tree. Examine parse tree and
* classify the query.
*/
skygw_query_type_t skygw_query_classifier_get_type( skygw_query_type_t skygw_query_classifier_get_type(
const char* query_str, const char* query_str,
unsigned long client_flags); unsigned long client_flags,
MYSQL** mysql);
/** Free THD context and close MYSQL */
void skygw_query_classifier_free(MYSQL* mysql);
char* skygw_query_classifier_get_stmtname(MYSQL* mysql);
EXTERN_C_BLOCK_END EXTERN_C_BLOCK_END

17
server/core/buffer.c
View File

@ -186,10 +186,9 @@ GWBUF *gwbuf_clone_transform(
goto return_clonebuf; goto return_clonebuf;
} }
switch (src_type) if (GWBUF_IS_TYPE_MYSQL(head))
{ {
case GWBUF_TYPE_MYSQL: if (GWBUF_TYPE_PLAINSQL == targettype)
if (targettype == GWBUF_TYPE_PLAINSQL)
{ {
/** Crete reference to string part of buffer */ /** Crete reference to string part of buffer */
clonebuf = gwbuf_clone_portion( clonebuf = gwbuf_clone_portion(
@ -198,18 +197,17 @@ GWBUF *gwbuf_clone_transform(
GWBUF_LENGTH(head)-5); GWBUF_LENGTH(head)-5);
ss_dassert(clonebuf != NULL); ss_dassert(clonebuf != NULL);
/** Overwrite the type with new format */ /** Overwrite the type with new format */
clonebuf->gwbuf_type = targettype; gwbuf_set_type(clonebuf, targettype);
} }
else else
{ {
clonebuf = NULL; clonebuf = NULL;
} }
break; }
else
default: {
clonebuf = NULL; clonebuf = NULL;
break; }
} /*< switch (src_type) */
return_clonebuf: return_clonebuf:
return clonebuf; return clonebuf;
@ -329,6 +327,7 @@ bool gwbuf_set_type(
case GWBUF_TYPE_MYSQL: case GWBUF_TYPE_MYSQL:
case GWBUF_TYPE_PLAINSQL: case GWBUF_TYPE_PLAINSQL:
case GWBUF_TYPE_UNDEFINED: case GWBUF_TYPE_UNDEFINED:
case GWBUF_TYPE_SINGLE_STMT: /*< buffer contains one stmt */
buf->gwbuf_type |= type; buf->gwbuf_type |= type;
succp = true; succp = true;
break; break;

10
server/core/dcb.c
View File

@ -301,8 +301,9 @@ dcb_final_free(DCB *dcb)
DCB_CALLBACK *cb; DCB_CALLBACK *cb;
CHK_DCB(dcb); CHK_DCB(dcb);
ss_info_dassert(dcb->state == DCB_STATE_DISCONNECTED, ss_info_dassert(dcb->state == DCB_STATE_DISCONNECTED ||
"dcb not in DCB_STATE_DISCONNECTED state."); dcb->state == DCB_STATE_ALLOC,
"dcb not in DCB_STATE_DISCONNECTED not in DCB_STATE_ALLOC state.");
/*< First remove this DCB from the chain */ /*< First remove this DCB from the chain */
spinlock_acquire(&dcbspin); spinlock_acquire(&dcbspin);
@ -701,6 +702,11 @@ int dcb_read(
n = 0; n = 0;
goto return_n; goto return_n;
} }
else if (b == 0)
{
n = 0;
goto return_n;
}
bufsize = MIN(b, MAX_BUFFER_SIZE); bufsize = MIN(b, MAX_BUFFER_SIZE);
if ((buffer = gwbuf_alloc(bufsize)) == NULL) if ((buffer = gwbuf_alloc(bufsize)) == NULL)

3
server/core/load_utils.c
View File

@ -116,7 +116,8 @@ MODULE_INFO *mod_info = NULL;
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"Error : Unable to load library for module: " "Error : Unable to load library for module: "
"%s\n\t\t\t %s.", "%s\n\n\t\t %s."
"\n\n",
module, module,
dlerror()))); dlerror())));
return NULL; return NULL;

22
server/core/session.c
View File

@ -127,7 +127,8 @@ session_alloc(SERVICE *service, DCB *client_dcb)
* session, therefore it is important that the session lock is * session, therefore it is important that the session lock is
* relinquished beforethe router call. * relinquished beforethe router call.
*/ */
if (client_dcb->state != DCB_STATE_LISTENING && client_dcb->dcb_role != DCB_ROLE_INTERNAL) if (client_dcb->state != DCB_STATE_LISTENING &&
client_dcb->dcb_role != DCB_ROLE_INTERNAL)
{ {
session->router_session = session->router_session =
service->router->newSession(service->router_instance, service->router->newSession(service->router_instance,
@ -196,6 +197,20 @@ session_alloc(SERVICE *service, DCB *client_dcb)
} }
spinlock_acquire(&session_spin); spinlock_acquire(&session_spin);
if (session->state != SESSION_STATE_READY)
{
session_free(session);
client_dcb->session = NULL;
session = NULL;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Failed to create %s session.",
service->name)));
spinlock_release(&session_spin);
}
else
{
session->state = SESSION_STATE_ROUTER_READY; session->state = SESSION_STATE_ROUTER_READY;
session->next = allSessions; session->next = allSessions;
allSessions = session; allSessions = session;
@ -203,7 +218,7 @@ session_alloc(SERVICE *service, DCB *client_dcb)
atomic_add(&service->stats.n_sessions, 1); atomic_add(&service->stats.n_sessions, 1);
atomic_add(&service->stats.n_current, 1); atomic_add(&service->stats.n_current, 1);
CHK_SESSION(session); CHK_SESSION(session);
}
return_session: return_session:
return session; return session;
} }
@ -310,9 +325,6 @@ bool session_free(
/* Free router_session and session */ /* Free router_session and session */
if (session->router_session) { if (session->router_session) {
session->service->router->closeSession(
session->service->router_instance,
session->router_session);
session->service->router->freeSession( session->service->router->freeSession(
session->service->router_instance, session->service->router_instance,
session->router_session); session->router_session);

5
server/include/buffer.h
View File

@ -48,11 +48,14 @@ typedef enum
{ {
GWBUF_TYPE_UNDEFINED = 0x00, GWBUF_TYPE_UNDEFINED = 0x00,
GWBUF_TYPE_PLAINSQL = 0x01, GWBUF_TYPE_PLAINSQL = 0x01,
GWBUF_TYPE_MYSQL = 0x02 GWBUF_TYPE_MYSQL = 0x02,
GWBUF_TYPE_SINGLE_STMT = 0x04
} gwbuf_type_t; } gwbuf_type_t;
#define GWBUF_IS_TYPE_UNDEFINED(b) (b->gwbuf_type == 0)
#define GWBUF_IS_TYPE_PLAINSQL(b) (b->gwbuf_type & GWBUF_TYPE_PLAINSQL) #define GWBUF_IS_TYPE_PLAINSQL(b) (b->gwbuf_type & GWBUF_TYPE_PLAINSQL)
#define GWBUF_IS_TYPE_MYSQL(b) (b->gwbuf_type & GWBUF_TYPE_MYSQL) #define GWBUF_IS_TYPE_MYSQL(b) (b->gwbuf_type & GWBUF_TYPE_MYSQL)
#define GWBUF_IS_TYPE_SINGLE_STMT(b) (b->gwbuf_type & GWBUF_TYPE_SINGLE_STMT)
/** /**
* A structure to encapsulate the data in a form that the data itself can be * A structure to encapsulate the data in a form that the data itself can be

1
server/include/dcb.h
View File

@ -229,7 +229,6 @@ typedef struct dcb {
struct service *service; /**< The related service */ struct service *service; /**< The related service */
void *data; /**< Specific client data */ void *data; /**< Specific client data */
DCBMM memdata; /**< The data related to DCB memory management */ DCBMM memdata; /**< The data related to DCB memory management */
int command; /**< Specific client command type */
SPINLOCK cb_lock; /**< The lock for the callbacks linked list */ SPINLOCK cb_lock; /**< The lock for the callbacks linked list */
DCB_CALLBACK *callbacks; /**< The list of callbacks for the DCB */ DCB_CALLBACK *callbacks; /**< The list of callbacks for the DCB */

125
server/modules/include/mysql_client_server_protocol.h
View File

@ -98,35 +98,9 @@ typedef enum {
MYSQL_AUTH_SENT, MYSQL_AUTH_SENT,
MYSQL_AUTH_RECV, MYSQL_AUTH_RECV,
MYSQL_AUTH_FAILED, MYSQL_AUTH_FAILED,
MYSQL_IDLE, MYSQL_IDLE
MYSQL_ROUTING, } mysql_auth_state_t;
MYSQL_WAITING_RESULT,
MYSQL_SESSION_CHANGE
} mysql_pstate_t;
/*
* MySQL Protocol specific state data
*/
typedef struct {
#if defined(SS_DEBUG)
skygw_chk_t protocol_chk_top;
#endif
int fd; /*< The socket descriptor */
struct dcb *owner_dcb; /*< The DCB of the socket
* we are running on */
mysql_pstate_t state; /*< Current protocol state */
uint8_t scramble[MYSQL_SCRAMBLE_LEN]; /*< server scramble,
* created or received */
uint32_t server_capabilities; /*< server capabilities,
* created or received */
uint32_t client_capabilities; /*< client capabilities,
* created or received */
unsigned long tid; /*< MySQL Thread ID, in
* handshake */
#if defined(SS_DEBUG)
skygw_chk_t protocol_chk_tail;
#endif
} MySQLProtocol;
/* /*
* MySQL session specific data * MySQL session specific data
@ -139,7 +113,6 @@ typedef struct mysql_session {
} MYSQL_session; } MYSQL_session;
/** Protocol packing macros. */ /** Protocol packing macros. */
#define gw_mysql_set_byte2(__buffer, __int) do { \ #define gw_mysql_set_byte2(__buffer, __int) do { \
(__buffer)[0]= (uint8_t)((__int) & 0xFF); \ (__buffer)[0]= (uint8_t)((__int) & 0xFF); \
@ -230,18 +203,90 @@ typedef enum
), ),
} gw_mysql_capabilities_t; } gw_mysql_capabilities_t;
/** Basic mysql commands */ /** Copy from enum in mariadb-5.5 mysql_com.h */
#define MYSQL_COM_CHANGE_USER 0x11 typedef enum mysql_server_cmd {
#define MYSQL_COM_QUIT 0x1 MYSQL_COM_UNDEFINED = -1,
#define MYSQL_COM_INIT_DB 0x2 MYSQL_COM_SLEEP = 0,
#define MYSQL_COM_QUERY 0x3 MYSQL_COM_QUIT,
MYSQL_COM_INIT_DB,
MYSQL_COM_QUERY,
MYSQL_COM_FIELD_LIST,
MYSQL_COM_CREATE_DB,
MYSQL_COM_DROP_DB,
MYSQL_COM_REFRESH,
MYSQL_COM_SHUTDOWN,
MYSQL_COM_STATISTICS,
MYSQL_COM_PROCESS_INFO,
MYSQL_COM_CONNECT,
MYSQL_COM_PROCESS_KILL,
MYSQL_COM_DEBUG,
MYSQL_COM_PING,
MYSQL_COM_TIME,
MYSQL_COM_DELAYED_INSERT,
MYSQL_COM_CHANGE_USER,
MYSQL_COM_BINLOG_DUMP,
MYSQL_COM_TABLE_DUMP,
MYSQL_COM_CONNECT_OUT,
MYSQL_COM_REGISTER_SLAVE,
MYSQL_COM_STMT_PREPARE,
MYSQL_COM_STMT_EXECUTE,
MYSQL_COM_STMT_SEND_LONG_DATA,
MYSQL_COM_STMT_CLOSE,
MYSQL_COM_STMT_RESET,
MYSQL_COM_SET_OPTION,
MYSQL_COM_STMT_FETCH,
MYSQL_COM_DAEMON
} mysql_server_cmd_t;
/**
* List of server commands, and number of response packets are stored here.
* server_command_t is used in MySQLProtocol structure, so for each DCB there is
* one MySQLProtocol and one server command list.
*/
typedef struct server_command_st {
mysql_server_cmd_t cmd;
int nresponse_packets; /** filled when reply arrives */
struct server_command_st* next;
} server_command_t;
/*
* MySQL Protocol specific state data
*/
typedef struct {
#if defined(SS_DEBUG)
skygw_chk_t protocol_chk_top;
#endif
int fd; /*< The socket descriptor */
struct dcb *owner_dcb; /*< The DCB of the socket
* we are running on */
SPINLOCK protocol_lock;
server_command_t protocol_command; /*< list of active commands */
mysql_auth_state_t protocol_auth_state; /*< Authentication status */
uint8_t scramble[MYSQL_SCRAMBLE_LEN]; /*< server scramble,
* created or received */
uint32_t server_capabilities; /*< server capabilities,
* created or received */
uint32_t client_capabilities; /*< client capabilities,
* created or received */
unsigned long tid; /*< MySQL Thread ID, in
* handshake */
#if defined(SS_DEBUG)
skygw_chk_t protocol_chk_tail;
#endif
} MySQLProtocol;
#define MYSQL_GET_COMMAND(payload) (payload[4]) #define MYSQL_GET_COMMAND(payload) (payload[4])
#define MYSQL_GET_PACKET_NO(payload) (payload[3]) #define MYSQL_GET_PACKET_NO(payload) (payload[3])
#define MYSQL_GET_PACKET_LEN(payload) (gw_mysql_get_byte3(payload)) #define MYSQL_GET_PACKET_LEN(payload) (gw_mysql_get_byte3(payload))
#define MYSQL_GET_ERRCODE(payload) (gw_mysql_get_byte2(&payload[5])) #define MYSQL_GET_ERRCODE(payload) (gw_mysql_get_byte2(&payload[5]))
#define MYSQL_GET_STMTOK_NPARAM(payload) (gw_mysql_get_byte2(&payload[9]))
#define MYSQL_GET_STMTOK_NATTR(payload) (gw_mysql_get_byte2(&payload[11]))
#define MYSQL_IS_ERROR_PACKET(payload) (MYSQL_GET_COMMAND(payload)==0xff)
#endif #endif /** _MYSQL_PROTOCOL_H */
void gw_mysql_close(MySQLProtocol **ptr); void gw_mysql_close(MySQLProtocol **ptr);
MySQLProtocol* mysql_protocol_init(DCB* dcb, int fd); MySQLProtocol* mysql_protocol_init(DCB* dcb, int fd);
@ -314,4 +359,14 @@ char *gw_strend(register const char *s);
int setnonblocking(int fd); int setnonblocking(int fd);
int setipaddress(struct in_addr *a, char *p); int setipaddress(struct in_addr *a, char *p);
GWBUF* gw_MySQL_get_next_packet(GWBUF** p_readbuf); GWBUF* gw_MySQL_get_next_packet(GWBUF** p_readbuf);
GWBUF* gw_MySQL_get_packets(GWBUF** p_readbuf, int* npackets);
GWBUF* gw_MySQL_discard_packets(GWBUF* buf, int npackets);
void protocol_add_srv_command(MySQLProtocol* p, mysql_server_cmd_t cmd);
void protocol_remove_srv_command(MySQLProtocol* p);
bool protocol_waits_response(MySQLProtocol* p);
mysql_server_cmd_t protocol_get_srv_command(MySQLProtocol* p,bool removep);
int get_stmt_nresponse_packets(GWBUF* buf, mysql_server_cmd_t cmd);
int protocol_get_nresponse_packets (MySQLProtocol* p);
bool protocol_set_nresponse_packets (MySQLProtocol* p, int nresponse_packets);

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

@ -30,15 +30,33 @@
*/ */
#include <dcb.h> #include <dcb.h>
#include <hashtable.h>
#undef PREP_STMT_CACHING
#if defined(PREP_STMT_CACHING)
typedef enum prep_stmt_type {
PREP_STMT_NAME,
PREP_STMT_ID
} prep_stmt_type_t;
typedef enum prep_stmt_state {
PREP_STMT_ALLOC,
PREP_STMT_SENT,
PREP_STMT_RECV,
PREP_STMT_DROPPED
} prep_stmt_state_t;
#endif /*< PREP_STMT_CACHING */
typedef enum bref_state { typedef enum bref_state {
BREF_NOT_USED = 0x00,
BREF_IN_USE = 0x01, BREF_IN_USE = 0x01,
BREF_WAITING_RESULT = 0x02, /*< for anything that responds */ BREF_WAITING_RESULT = 0x02, /*< for anything that responds */
BREF_CLOSED = 0x04 BREF_CLOSED = 0x04
} bref_state_t; } bref_state_t;
#define BREF_IS_NOT_USED(s) (s->bref_state & BREF_NOT_USED) #define BREF_IS_NOT_USED(s) (s->bref_state & ~BREF_IN_USE)
#define BREF_IS_IN_USE(s) (s->bref_state & BREF_IN_USE) #define BREF_IS_IN_USE(s) (s->bref_state & BREF_IN_USE)
#define BREF_IS_WAITING_RESULT(s) (s->bref_state & BREF_WAITING_RESULT) #define BREF_IS_WAITING_RESULT(s) (s->bref_state & BREF_WAITING_RESULT)
#define BREF_IS_CLOSED(s) (s->bref_state & BREF_CLOSED) #define BREF_IS_CLOSED(s) (s->bref_state & BREF_CLOSED)
@ -186,6 +204,25 @@ typedef struct rwsplit_config_st {
} rwsplit_config_t; } rwsplit_config_t;
#if defined(PREP_STMT_CACHING)
typedef struct prep_stmt_st {
#if defined(SS_DEBUG)
skygw_chk_t pstmt_chk_top;
#endif
union id {
int seq;
char* name;
} pstmt_id;
prep_stmt_state_t pstmt_state;
prep_stmt_type_t pstmt_type;
#if defined(SS_DEBUG)
skygw_chk_t pstmt_chk_tail;
#endif
} prep_stmt_t;
#endif /*< PREP_STMT_CACHING */
/** /**
* The client session structure used within this router. * The client session structure used within this router.
*/ */
@ -205,7 +242,9 @@ struct router_client_session {
int rses_capabilities; /*< input type, for example */ int rses_capabilities; /*< input type, for example */
bool rses_autocommit_enabled; bool rses_autocommit_enabled;
bool rses_transaction_active; bool rses_transaction_active;
uint64_t rses_id; /*< ID for router client session */ #if defined(PREP_STMT_CACHING)
HASHTABLE* rses_prep_stmt[2];
#endif
struct router_client_session* next; struct router_client_session* next;
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
skygw_chk_t rses_chk_tail; skygw_chk_t rses_chk_tail;

270
server/modules/protocol/mysql_backend.c
View File

@ -163,7 +163,7 @@ static int gw_read_backend_event(DCB *dcb) {
backend_protocol = (MySQLProtocol *) dcb->protocol; backend_protocol = (MySQLProtocol *) dcb->protocol;
CHK_PROTOCOL(backend_protocol); CHK_PROTOCOL(backend_protocol);
#if 1
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_backend_event] Read dcb %p fd %d protocol " "%lu [gw_read_backend_event] Read dcb %p fd %d protocol "
@ -171,8 +171,9 @@ static int gw_read_backend_event(DCB *dcb) {
pthread_self(), pthread_self(),
dcb, dcb,
dcb->fd, dcb->fd,
backend_protocol->state, backend_protocol->protocol_auth_state,
STRPROTOCOLSTATE(backend_protocol->state)))); STRPROTOCOLSTATE(backend_protocol->protocol_auth_state))));
#endif
/* backend is connected: /* backend is connected:
@ -186,17 +187,18 @@ static int gw_read_backend_event(DCB *dcb) {
* If starting to auhenticate with backend server, lock dcb * If starting to auhenticate with backend server, lock dcb
* to prevent overlapping processing of auth messages. * to prevent overlapping processing of auth messages.
*/ */
if (backend_protocol->state == MYSQL_CONNECTED) { if (backend_protocol->protocol_auth_state == MYSQL_CONNECTED)
{
spinlock_acquire(&dcb->authlock); spinlock_acquire(&dcb->authlock);
backend_protocol = (MySQLProtocol *) dcb->protocol; backend_protocol = (MySQLProtocol *) dcb->protocol;
CHK_PROTOCOL(backend_protocol); CHK_PROTOCOL(backend_protocol);
if (backend_protocol->state == MYSQL_CONNECTED) { if (backend_protocol->protocol_auth_state == MYSQL_CONNECTED)
{
if (gw_read_backend_handshake(backend_protocol) != 0) { if (gw_read_backend_handshake(backend_protocol) != 0)
backend_protocol->state = MYSQL_AUTH_FAILED; {
backend_protocol->protocol_auth_state = MYSQL_AUTH_FAILED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_backend_event] after " "%lu [gw_read_backend_event] after "
@ -205,7 +207,9 @@ static int gw_read_backend_event(DCB *dcb) {
pthread_self(), pthread_self(),
backend_protocol->owner_dcb->fd))); backend_protocol->owner_dcb->fd)));
} else { }
else
{
/* handshake decoded, send the auth credentials */ /* handshake decoded, send the auth credentials */
if (gw_send_authentication_to_backend( if (gw_send_authentication_to_backend(
current_session->db, current_session->db,
@ -213,7 +217,7 @@ static int gw_read_backend_event(DCB *dcb) {
current_session->client_sha1, current_session->client_sha1,
backend_protocol) != 0) backend_protocol) != 0)
{ {
backend_protocol->state = MYSQL_AUTH_FAILED; backend_protocol->protocol_auth_state = MYSQL_AUTH_FAILED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_backend_event] after " "%lu [gw_read_backend_event] after "
@ -221,31 +225,31 @@ static int gw_read_backend_event(DCB *dcb) {
"fd %d, state = MYSQL_AUTH_FAILED.", "fd %d, state = MYSQL_AUTH_FAILED.",
pthread_self(), pthread_self(),
backend_protocol->owner_dcb->fd))); backend_protocol->owner_dcb->fd)));
} else { }
backend_protocol->state = MYSQL_AUTH_RECV; else
{
backend_protocol->protocol_auth_state = MYSQL_AUTH_RECV;
} }
} }
} }
spinlock_release(&dcb->authlock); spinlock_release(&dcb->authlock);
} }
/* /*
* Now: * Now:
* -- check the authentication reply from backend * -- check the authentication reply from backend
* OR * OR
* -- handle a previous handshake error * -- handle a previous handshake error
*/ */
if (backend_protocol->state == MYSQL_AUTH_RECV || if (backend_protocol->protocol_auth_state == MYSQL_AUTH_RECV ||
backend_protocol->state == MYSQL_AUTH_FAILED) backend_protocol->protocol_auth_state == MYSQL_AUTH_FAILED)
{ {
spinlock_acquire(&dcb->authlock); spinlock_acquire(&dcb->authlock);
backend_protocol = (MySQLProtocol *) dcb->protocol; backend_protocol = (MySQLProtocol *) dcb->protocol;
CHK_PROTOCOL(backend_protocol); CHK_PROTOCOL(backend_protocol);
if (backend_protocol->state == MYSQL_AUTH_RECV || if (backend_protocol->protocol_auth_state == MYSQL_AUTH_RECV ||
backend_protocol->state == MYSQL_AUTH_FAILED) backend_protocol->protocol_auth_state == MYSQL_AUTH_FAILED)
{ {
ROUTER_OBJECT *router = NULL; ROUTER_OBJECT *router = NULL;
ROUTER *router_instance = NULL; ROUTER *router_instance = NULL;
@ -259,7 +263,7 @@ static int gw_read_backend_event(DCB *dcb) {
router_instance = session->service->router_instance; router_instance = session->service->router_instance;
rsession = session->router_session; rsession = session->router_session;
if (backend_protocol->state == MYSQL_AUTH_RECV) { if (backend_protocol->protocol_auth_state == MYSQL_AUTH_RECV) {
/*< /*<
* Read backed auth reply * Read backed auth reply
*/ */
@ -268,14 +272,14 @@ static int gw_read_backend_event(DCB *dcb) {
switch (receive_rc) { switch (receive_rc) {
case -1: case -1:
backend_protocol->state = MYSQL_AUTH_FAILED; backend_protocol->protocol_auth_state = MYSQL_AUTH_FAILED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_backend_event] after " "%lu [gw_read_backend_event] after "
"gw_receive_backend_authentication " "gw_receive_backend_authentication "
"fd %d, state = MYSQL_AUTH_FAILED.", "fd %d, state = MYSQL_AUTH_FAILED.",
backend_protocol->owner_dcb->fd, pthread_self(),
pthread_self()))); backend_protocol->owner_dcb->fd)));
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
@ -286,7 +290,7 @@ static int gw_read_backend_event(DCB *dcb) {
current_session->user))); current_session->user)));
break; break;
case 1: case 1:
backend_protocol->state = MYSQL_IDLE; backend_protocol->protocol_auth_state = MYSQL_IDLE;
LOGIF(LD, (skygw_log_write_flush( LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG, LOGFILE_DEBUG,
@ -316,7 +320,7 @@ static int gw_read_backend_event(DCB *dcb) {
} /* switch */ } /* switch */
} }
if (backend_protocol->state == MYSQL_AUTH_FAILED) if (backend_protocol->protocol_auth_state == MYSQL_AUTH_FAILED)
{ {
/** /**
* protocol state won't change anymore, * protocol state won't change anymore,
@ -340,9 +344,14 @@ static int gw_read_backend_event(DCB *dcb) {
/* try reload users' table for next connection */ /* try reload users' table for next connection */
service_refresh_users(dcb->session->service); service_refresh_users(dcb->session->service);
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
LOGIF(LE, (skygw_log_write_flush( LOGIF(LD, (skygw_log_write(
LOGFILE_ERROR, LOGFILE_DEBUG,
"Backend read error handling."))); "%lu [gw_read_backend_event] "
"calling handleError. Backend "
"DCB %p, session %p",
pthread_self(),
dcb,
dcb->session)));
#endif #endif
errbuf = mysql_create_custom_error( errbuf = mysql_create_custom_error(
@ -360,6 +369,15 @@ static int gw_read_backend_event(DCB *dcb) {
ss_dassert(!succp); ss_dassert(!succp);
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_read_backend_event] "
"after calling handleError. Backend "
"DCB %p, session %p",
pthread_self(),
dcb,
dcb->session)));
if (session != NULL) if (session != NULL)
{ {
spinlock_acquire(&session->ses_lock); spinlock_acquire(&session->ses_lock);
@ -373,7 +391,7 @@ static int gw_read_backend_event(DCB *dcb) {
} }
else else
{ {
ss_dassert(backend_protocol->state == MYSQL_IDLE); ss_dassert(backend_protocol->protocol_auth_state == MYSQL_IDLE);
LOGIF(LD, (skygw_log_write_flush( LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_backend_event] " "%lu [gw_read_backend_event] "
@ -398,11 +416,13 @@ static int gw_read_backend_event(DCB *dcb) {
/* reading MySQL command output from backend and writing to the client */ /* reading MySQL command output from backend and writing to the client */
{ {
GWBUF *writebuf = NULL; GWBUF *readbuf = NULL;
ROUTER_OBJECT *router = NULL; ROUTER_OBJECT *router = NULL;
ROUTER *router_instance = NULL; ROUTER *router_instance = NULL;
void *rsession = NULL; void *rsession = NULL;
SESSION *session = dcb->session; SESSION *session = dcb->session;
int nbytes_read = 0;
mysql_server_cmd_t srvcmd = MYSQL_COM_UNDEFINED;
CHK_SESSION(session); CHK_SESSION(session);
router = session->service->router; router = session->service->router;
@ -410,32 +430,17 @@ static int gw_read_backend_event(DCB *dcb) {
rsession = session->router_session; rsession = session->router_session;
/* read available backend data */ /* read available backend data */
rc = dcb_read(dcb, &writebuf); rc = dcb_read(dcb, &readbuf);
if (rc < 0) if (rc < 0)
{ {
/*< vraa : errorHandle */
/*<
* Backend generated EPOLLIN event and if backend has
* failed, connection must be closed to avoid backend
* dcb from getting hanged.
*/
GWBUF* errbuf; GWBUF* errbuf;
bool succp; bool succp;
/**
* - send error for client
* - mark failed backend BREF_NOT_USED
* - go through all servers and select one according to
* the criteria that user specified in the beginning.
*/
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"Backend read error handling #2."))); "Backend read error handling #2.")));
#endif #endif
errbuf = mysql_create_custom_error( errbuf = mysql_create_custom_error(
1, 1,
0, 0,
@ -458,33 +463,111 @@ static int gw_read_backend_event(DCB *dcb) {
rc = 0; rc = 0;
goto return_rc; goto return_rc;
} }
nbytes_read = gwbuf_length(readbuf);
if (writebuf == NULL) { if (nbytes_read == 0)
rc = 0; {
goto return_rc; goto return_rc;
} }
else
{
ss_dassert(readbuf != NULL);
}
/**
* ask for next response (1 or more packets) like in
* gw_MySQL_get_next_packet but gw_MySQL_get_next_response
*/
srvcmd = protocol_get_srv_command((MySQLProtocol *)dcb->protocol,
false);
/**
* If backend DCB is waiting for response to COM_STMT_PREPARE,
* it, then only that must be passed to clientReply.
*
* If response consists of ses cmd response and response to
* COM_STMT_PREPARE, there can't be anything after
* COM_STMT_PREPARE response because whole buffer may be
* discarded since router doesn't know the borderlines of MySQL
* packets.
*/
/**
* Read all packets from <readbuf> which belong to STMT PREPARE
* response.
* Move packets not belonging to STMT PREPARE response to
* dcb_readqueue.
* When whole response is read, pass <readbuf> forward to
* clientReply.
*/
if (srvcmd == MYSQL_COM_STMT_PREPARE)
{
MySQLProtocol* p;
int nresponse_packets;
GWBUF* tmpbuf;
p = (MySQLProtocol *)dcb->protocol;
nresponse_packets = protocol_get_nresponse_packets(p);
/** count only once per response */
if (nresponse_packets == 0)
{
nresponse_packets = get_stmt_nresponse_packets(
readbuf,
srvcmd);
}
tmpbuf = gw_MySQL_get_packets(&readbuf, &nresponse_packets);
gwbuf_append(dcb->dcb_readqueue, readbuf);
readbuf = tmpbuf;
/** <readbuf> contains incomplete response to STMT PREPARE */
if (nresponse_packets != 0)
{
rc = 0;
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Backend fd %d read incomplete response packet. "
"Waiting %d more, cmd %s.",
dcb->fd,
nresponse_packets,
STRPACKETTYPE(srvcmd))));
/**
* store the number of how many packets the
* reponse consists of to backend's protocol.
*/
protocol_set_nresponse_packets(p, nresponse_packets);
goto return_rc;
}
protocol_remove_srv_command((MySQLProtocol *)dcb->protocol);
}
/*< /*<
* If dcb->session->client is freed already it may be NULL. * If dcb->session->client is freed already it may be NULL.
*/ */
if (dcb->session->client != NULL) { if (dcb->session->client != NULL)
{
client_protocol = SESSION_PROTOCOL(dcb->session, client_protocol = SESSION_PROTOCOL(dcb->session,
MySQLProtocol); MySQLProtocol);
if (client_protocol != NULL) { if (client_protocol != NULL)
{
CHK_PROTOCOL(client_protocol); CHK_PROTOCOL(client_protocol);
if (client_protocol->state == MYSQL_IDLE) if (client_protocol->protocol_auth_state ==
MYSQL_IDLE)
{ {
gwbuf_set_type(writebuf, GWBUF_TYPE_MYSQL); gwbuf_set_type(readbuf, GWBUF_TYPE_MYSQL);
router->clientReply(router_instance, router->clientReply(router_instance,
rsession, rsession,
writebuf, readbuf,
dcb); dcb);
rc = 1; rc = 1;
} }
goto return_rc; goto return_rc;
} else if (dcb->session->client->dcb_role == DCB_ROLE_INTERNAL) { }
gwbuf_set_type(writebuf, GWBUF_TYPE_MYSQL); else if (dcb->session->client->dcb_role == DCB_ROLE_INTERNAL)
router->clientReply(router_instance, rsession, writebuf, dcb); {
gwbuf_set_type(readbuf, GWBUF_TYPE_MYSQL);
router->clientReply(router_instance, rsession, readbuf, dcb);
rc = 1; rc = 1;
} }
} }
@ -550,8 +633,8 @@ static int gw_write_backend_event(DCB *dcb) {
goto return_rc; goto return_rc;
} }
if (backend_protocol->state == MYSQL_PENDING_CONNECT) { if (backend_protocol->protocol_auth_state == MYSQL_PENDING_CONNECT) {
backend_protocol->state = MYSQL_CONNECTED; backend_protocol->protocol_auth_state = MYSQL_CONNECTED;
rc = 1; rc = 1;
goto return_rc; goto return_rc;
} }
@ -571,7 +654,7 @@ return_rc:
} }
/* /*
* Write function for backend DCB * Write function for backend DCB. Store command to protocol.
* *
* @param dcb The DCB of the backend * @param dcb The DCB of the backend
* @param queue Queue of buffers to write * @param queue Queue of buffers to write
@ -589,7 +672,7 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
* If auth failed, return value is 0, write and buffered write * If auth failed, return value is 0, write and buffered write
* return 1. * return 1.
*/ */
switch(backend_protocol->state) { switch(backend_protocol->protocol_auth_state) {
case MYSQL_AUTH_FAILED: case MYSQL_AUTH_FAILED:
{ {
size_t len; size_t len;
@ -617,6 +700,10 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
} }
case MYSQL_IDLE: case MYSQL_IDLE:
{
uint8_t* ptr = GWBUF_DATA(queue);
int cmd = MYSQL_GET_COMMAND(ptr);
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_MySQLWrite_backend] write to dcb %p " "%lu [gw_MySQLWrite_backend] write to dcb %p "
@ -624,18 +711,37 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
pthread_self(), pthread_self(),
dcb, dcb,
dcb->fd, dcb->fd,
STRPROTOCOLSTATE(backend_protocol->state)))); STRPROTOCOLSTATE(backend_protocol->protocol_auth_state))));
spinlock_release(&dcb->authlock); spinlock_release(&dcb->authlock);
/**
* Server commands are stored to MySQLProtocol structure
* if buffer always includes a single statement. That
* information is stored in GWBUF type field
* (GWBUF_TYPE_SINGLE_STMT bit).
*/
if (GWBUF_IS_TYPE_SINGLE_STMT(queue))
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Write to backend's DCB fd %d "
"cmd %s protocol state %s.",
dcb->fd,
STRPACKETTYPE(cmd),
STRPROTOCOLSTATE(backend_protocol->protocol_auth_state))));
/** Record the command to backend's protocol */
protocol_add_srv_command(backend_protocol, cmd);
}
/** Write to backend */
rc = dcb_write(dcb, queue); rc = dcb_write(dcb, queue);
goto return_rc; goto return_rc;
break; break;
}
default: default:
/*< {
* Now put the incoming data to the delay queue unless backend is uint8_t* ptr = GWBUF_DATA(queue);
* connected with auth ok int cmd = MYSQL_GET_COMMAND(ptr);
*/
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_MySQLWrite_backend] delayed write to " "%lu [gw_MySQLWrite_backend] delayed write to "
@ -643,13 +749,37 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
pthread_self(), pthread_self(),
dcb, dcb,
dcb->fd, dcb->fd,
STRPROTOCOLSTATE(backend_protocol->state)))); STRPROTOCOLSTATE(backend_protocol->protocol_auth_state))));
/**
* Since it is known that buffer contains one complete
* command, store the command to backend's protocol. When
* backend server responses the command determines how
* response needs to be processed. This is mainly due to
* MYSQL_COM_STMT_PREPARE whose response consists of
* arbitrary number of packets.
*/
if (GWBUF_IS_TYPE_SINGLE_STMT(queue))
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Write to backend's delayqueue fd %d "
"protocol state %s.",
dcb->fd,
STRPROTOCOLSTATE(backend_protocol->protocol_auth_state))));
/** Record the command to backend's protocol */
protocol_add_srv_command(backend_protocol, cmd);
}
/*<
* Now put the incoming data to the delay queue unless backend is
* connected with auth ok
*/
backend_set_delayqueue(dcb, queue); backend_set_delayqueue(dcb, queue);
spinlock_release(&dcb->authlock); spinlock_release(&dcb->authlock);
rc = 1; rc = 1;
goto return_rc; goto return_rc;
break; break;
} }
}
return_rc: return_rc:
return rc; return rc;
} }
@ -757,7 +887,7 @@ static int gw_create_backend_connection(
case 0: case 0:
ss_dassert(fd > 0); ss_dassert(fd > 0);
protocol->fd = fd; protocol->fd = fd;
protocol->state = MYSQL_CONNECTED; protocol->protocol_auth_state = MYSQL_CONNECTED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_create_backend_connection] Established " "%lu [gw_create_backend_connection] Established "
@ -772,7 +902,7 @@ static int gw_create_backend_connection(
case 1: case 1:
ss_dassert(fd > 0); ss_dassert(fd > 0);
protocol->state = MYSQL_PENDING_CONNECT; protocol->protocol_auth_state = MYSQL_PENDING_CONNECT;
protocol->fd = fd; protocol->fd = fd;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
@ -787,7 +917,7 @@ static int gw_create_backend_connection(
default: default:
ss_dassert(fd == -1); ss_dassert(fd == -1);
ss_dassert(protocol->state == MYSQL_ALLOC); ss_dassert(protocol->protocol_auth_state == MYSQL_ALLOC);
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_create_backend_connection] Connection " "%lu [gw_create_backend_connection] Connection "

42
server/modules/protocol/mysql_client.c
View File

@ -587,7 +587,7 @@ int gw_read_client_event(
/** /**
* Now there should be at least one complete mysql packet in read_buffer. * Now there should be at least one complete mysql packet in read_buffer.
*/ */
switch (protocol->state) { switch (protocol->protocol_auth_state) {
case MYSQL_AUTH_SENT: case MYSQL_AUTH_SENT:
{ {
@ -600,7 +600,7 @@ int gw_read_client_event(
if (auth_val == 0) if (auth_val == 0)
{ {
SESSION *session = NULL; SESSION *session = NULL;
protocol->state = MYSQL_AUTH_RECV; protocol->protocol_auth_state = MYSQL_AUTH_RECV;
/** /**
* Create session, and a router session for it. * Create session, and a router session for it.
* If successful, there will be backend connection(s) * If successful, there will be backend connection(s)
@ -612,7 +612,8 @@ int gw_read_client_event(
{ {
CHK_SESSION(session); CHK_SESSION(session);
ss_dassert(session->state != SESSION_STATE_ALLOC); ss_dassert(session->state != SESSION_STATE_ALLOC);
protocol->state = MYSQL_IDLE;
protocol->protocol_auth_state = MYSQL_IDLE;
/** /**
* Send an AUTH_OK packet to the client, * Send an AUTH_OK packet to the client,
* packet sequence is # 2 * packet sequence is # 2
@ -621,7 +622,7 @@ int gw_read_client_event(
} }
else else
{ {
protocol->state = MYSQL_AUTH_FAILED; protocol->protocol_auth_state = MYSQL_AUTH_FAILED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_client_event] session " "%lu [gw_read_client_event] session "
@ -642,7 +643,7 @@ int gw_read_client_event(
} }
else else
{ {
protocol->state = MYSQL_AUTH_FAILED; protocol->protocol_auth_state = MYSQL_AUTH_FAILED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_client_event] after " "%lu [gw_read_client_event] after "
@ -894,7 +895,7 @@ int gw_write_client_event(DCB *dcb)
protocol = (MySQLProtocol *)dcb->protocol; protocol = (MySQLProtocol *)dcb->protocol;
CHK_PROTOCOL(protocol); CHK_PROTOCOL(protocol);
if (protocol->state == MYSQL_IDLE) if (protocol->protocol_auth_state == MYSQL_IDLE)
{ {
dcb_drain_writeq(dcb); dcb_drain_writeq(dcb);
goto return_1; goto return_1;
@ -1236,7 +1237,7 @@ int gw_MySQLAccept(DCB *listener)
MySQLSendHandshake(client_dcb); MySQLSendHandshake(client_dcb);
// client protocol state change // client protocol state change
protocol->state = MYSQL_AUTH_SENT; protocol->protocol_auth_state = MYSQL_AUTH_SENT;
/** /**
* Set new descriptor to event set. At the same time, * Set new descriptor to event set. At the same time,
@ -1294,8 +1295,15 @@ static int gw_error_client_event(
SESSION* session; SESSION* session;
CHK_DCB(dcb); CHK_DCB(dcb);
session = dcb->session;
CHK_SESSION(session); LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_error_client_event] Error event handling for DCB %p "
"in state %s, session %p.",
pthread_self(),
dcb,
STRDCBSTATE(dcb->state),
(session != NULL ? session : NULL))));
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
@ -1373,7 +1381,7 @@ gw_client_hangup_event(DCB *dcb)
/** /**
* Detect if buffer includes partial mysql packet or multiple packets. * Detect if buffer includes partial mysql packet or multiple packets.
* Store partial packet to pendingqueue. Send complete packets one by one * Store partial packet to dcb_readqueue. Send complete packets one by one
* to router. * to router.
* *
* It is assumed readbuf includes at least one complete packet. * It is assumed readbuf includes at least one complete packet.
@ -1392,6 +1400,20 @@ static int route_by_statement(SESSION *session, GWBUF *readbuf)
if (packetbuf != NULL) if (packetbuf != NULL)
{ {
CHK_GWBUF(packetbuf); CHK_GWBUF(packetbuf);
/**
* This means that buffer includes exactly one MySQL
* statement.
* backend func.write uses the information. MySQL backend
* protocol, for example, stores the command identifier
* to protocol structure. When some other thread reads
* the corresponding response the command tells how to
* handle response.
*
* Set it here instead of gw_read_client_event to make
* sure it is set to each (MySQL) packet.
*/
gwbuf_set_type(packetbuf, GWBUF_TYPE_SINGLE_STMT);
/** Route query */
rc = SESSION_ROUTE_QUERY(session, packetbuf); rc = SESSION_ROUTE_QUERY(session, packetbuf);
} }
else else

235
server/modules/protocol/mysql_common.c
View File

@ -76,7 +76,8 @@ MySQLProtocol* mysql_protocol_init(
strerror(eno)))); strerror(eno))));
goto return_p; goto return_p;
} }
p->state = MYSQL_ALLOC; p->protocol_auth_state = MYSQL_ALLOC;
p->protocol_command.cmd = MYSQL_COM_UNDEFINED;
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
p->protocol_chk_top = CHK_NUM_PROTOCOL; p->protocol_chk_top = CHK_NUM_PROTOCOL;
p->protocol_chk_tail = CHK_NUM_PROTOCOL; p->protocol_chk_tail = CHK_NUM_PROTOCOL;
@ -151,7 +152,7 @@ int gw_read_backend_handshake(
if (h_len <= 4) { if (h_len <= 4) {
/* log error this exit point */ /* log error this exit point */
conn->state = MYSQL_AUTH_FAILED; conn->protocol_auth_state = MYSQL_AUTH_FAILED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_backend_handshake] after " "%lu [gw_read_backend_handshake] after "
@ -198,7 +199,7 @@ int gw_read_backend_handshake(
* data in buffer less than expected in the * data in buffer less than expected in the
* packet. Log error this exit point * packet. Log error this exit point
*/ */
conn->state = MYSQL_AUTH_FAILED; conn->protocol_auth_state = MYSQL_AUTH_FAILED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_backend_handshake] after " "%lu [gw_read_backend_handshake] after "
@ -223,7 +224,7 @@ int gw_read_backend_handshake(
* we cannot continue * we cannot continue
* log error this exit point * log error this exit point
*/ */
conn->state = MYSQL_AUTH_FAILED; conn->protocol_auth_state = MYSQL_AUTH_FAILED;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [gw_read_backend_handshake] after " "%lu [gw_read_backend_handshake] after "
@ -236,7 +237,7 @@ int gw_read_backend_handshake(
return 1; return 1;
} }
conn->state = MYSQL_AUTH_SENT; conn->protocol_auth_state = MYSQL_AUTH_SENT;
// consume all the data here // consume all the data here
head = gwbuf_consume(head, GWBUF_LENGTH(head)); head = gwbuf_consume(head, GWBUF_LENGTH(head));
@ -789,13 +790,7 @@ gw_mysql_protocol_state2string (int state) {
case MYSQL_AUTH_FAILED: case MYSQL_AUTH_FAILED:
return "MySQL Authentication failed"; return "MySQL Authentication failed";
case MYSQL_IDLE: case MYSQL_IDLE:
return "MySQL Auth done. Protocol is idle, waiting for statements"; return "MySQL authentication is succesfully done.";
case MYSQL_ROUTING:
return "MySQL received command has been routed to backend(s)";
case MYSQL_WAITING_RESULT:
return "MySQL Waiting for result set";
case MYSQL_SESSION_CHANGE:
return "MySQL change session";
default: default:
return "MySQL (unknown protocol state)"; return "MySQL (unknown protocol state)";
} }
@ -960,11 +955,9 @@ int mysql_send_custom_error (
const char *mysql_message) const char *mysql_message)
{ {
GWBUF* buf; GWBUF* buf;
int nbytes;
buf = mysql_create_custom_error(dcb, in_affected_rows, mysql_message); buf = mysql_create_custom_error(packet_number, in_affected_rows, mysql_message);
nbytes = GWBUF_LENGTH(buf);
dcb->func.write(dcb, buf); dcb->func.write(dcb, buf);
return GWBUF_LENGTH(buf); return GWBUF_LENGTH(buf);
@ -1500,7 +1493,7 @@ GWBUF* gw_MySQL_get_next_packet(
packetbuf = NULL; packetbuf = NULL;
goto return_packetbuf; goto return_packetbuf;
} }
/** there is one complete packet in the buffer */
if (packetlen == buflen) if (packetlen == buflen)
{ {
packetbuf = gwbuf_clone_portion(readbuf, 0, packetlen); packetbuf = gwbuf_clone_portion(readbuf, 0, packetlen);
@ -1541,3 +1534,213 @@ return_packetbuf:
return packetbuf; return packetbuf;
} }
/**
* Move <npackets> from buffer pointed to by <*p_readbuf>.
*/
GWBUF* gw_MySQL_get_packets(
GWBUF** p_srcbuf,
int* npackets)
{
GWBUF* packetbuf;
GWBUF* targetbuf = NULL;
while (*npackets > 0 && (packetbuf = gw_MySQL_get_next_packet(p_srcbuf)) != NULL)
{
targetbuf = gwbuf_append(targetbuf, packetbuf);
*npackets -= 1;
}
ss_dassert(*npackets < 128);
ss_dassert(*npackets >= 0);
return targetbuf;
}
/**
* If router expects to get separate, complete statements, add MySQL command
* to MySQLProtocol structure. It is removed when response has arrived.
*/
void protocol_add_srv_command(
MySQLProtocol* p,
mysql_server_cmd_t cmd)
{
spinlock_acquire(&p->protocol_lock);
if (p->protocol_command.cmd == MYSQL_COM_UNDEFINED)
{
p->protocol_command.cmd = cmd;
p->protocol_command.nresponse_packets = 0;
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Added command %s to fd %d.",
STRPACKETTYPE(cmd),
p->owner_dcb->fd)));
}
else
{
server_command_t* c =
(server_command_t *)malloc(sizeof(server_command_t));
c->cmd = cmd;
c->nresponse_packets = 0;
c->next = NULL;
p->protocol_command.next = c;
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Added another command %s to fd %d.",
STRPACKETTYPE(cmd),
p->owner_dcb->fd)));
#if defined(SS_DEBUG)
c = &p->protocol_command;
while (c != NULL && c->cmd != MYSQL_COM_UNDEFINED)
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"fd %d : %d %s",
p->owner_dcb->fd,
c->cmd,
STRPACKETTYPE(c->cmd))));
c = c->next;
}
#endif
}
spinlock_release(&p->protocol_lock);
}
/**
* If router processes separate statements, every stmt has corresponding MySQL
* command stored in MySQLProtocol structure.
*
* Remove current (=oldest) command.
*/
void protocol_remove_srv_command(
MySQLProtocol* p)
{
server_command_t* s;
spinlock_acquire(&p->protocol_lock);
s = &p->protocol_command;
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Removed command %s from fd %d.",
STRPACKETTYPE(s->cmd),
p->owner_dcb->fd)));
if (s->next == NULL)
{
p->protocol_command.cmd = MYSQL_COM_UNDEFINED;
}
else
{
p->protocol_command = *(s->next);
free(s->next);
}
spinlock_release(&p->protocol_lock);
}
mysql_server_cmd_t protocol_get_srv_command(
MySQLProtocol* p,
bool removep)
{
mysql_server_cmd_t cmd;
cmd = p->protocol_command.cmd;
if (removep)
{
protocol_remove_srv_command(p);
}
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Read command %s for fd %d.",
STRPACKETTYPE(cmd),
p->owner_dcb->fd)));
return cmd;
}
/**
* Return how many packets are included in the server's response.
*/
int get_stmt_nresponse_packets(
GWBUF* buf,
mysql_server_cmd_t cmd)
{
int npackets;
uint8_t* packet;
int nparam;
int nattr;
uint8_t* data;
switch (cmd) {
case MYSQL_COM_STMT_PREPARE:
data = (uint8_t *)buf->start;
if (data[4] == 0xff)
{
npackets = 1; /*< error packet */
}
else
{
packet = (uint8_t *)GWBUF_DATA(buf);
/** ok + nparam + eof + nattr + eof */
nparam = MYSQL_GET_STMTOK_NPARAM(packet);
nattr = MYSQL_GET_STMTOK_NATTR(packet);
npackets = 1 + nparam + MIN(1, nparam) +
nattr + MIN(nattr, 1);
ss_dassert(npackets<128);
}
break;
default:
npackets = 1;
break;
}
ss_dassert(npackets<128);
return npackets;
}
int protocol_get_nresponse_packets (
MySQLProtocol* p)
{
int rval;
CHK_PROTOCOL(p);
spinlock_acquire(&p->protocol_lock);
rval = p->protocol_command.nresponse_packets;
spinlock_release(&p->protocol_lock);
ss_dassert(rval<128);
return rval;
}
bool protocol_set_nresponse_packets (
MySQLProtocol* p,
int nresponse_packets)
{
bool succp;
CHK_PROTOCOL(p);
spinlock_acquire(&p->protocol_lock);
if (p->protocol_command.nresponse_packets > 0 &&
nresponse_packets > p->protocol_command.nresponse_packets)
{
succp = false;
}
else
{
p->protocol_command.nresponse_packets = nresponse_packets;
ss_dassert(nresponse_packets<128);
succp = true;
}
spinlock_release(&p->protocol_lock);
return succp;
}

4
server/modules/routing/readconnroute.c
View File

@ -113,7 +113,7 @@ static void clientReply(
static void handleError( static void handleError(
ROUTER *instance, ROUTER *instance,
void *router_session, void *router_session,
char *message, GWBUF *errbuf,
DCB *backend_dcb, DCB *backend_dcb,
int action, int action,
bool *succp); bool *succp);
@ -708,7 +708,7 @@ static void
handleError( handleError(
ROUTER *instance, ROUTER *instance,
void *router_session, void *router_session,
char *message, GWBUF *errbuf,
DCB *backend_dcb, DCB *backend_dcb,
int action, int action,
bool *succp) bool *succp)

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

@ -31,6 +31,7 @@
#include <dcb.h> #include <dcb.h>
#include <spinlock.h> #include <spinlock.h>
#include <modinfo.h> #include <modinfo.h>
#include <mysql_client_server_protocol.h>
MODULE_INFO info = { MODULE_INFO info = {
MODULE_API_ROUTER, MODULE_API_ROUTER,
@ -97,6 +98,11 @@ static backend_ref_t* get_bref_from_dcb(ROUTER_CLIENT_SES* rses, DCB* dcb);
static uint8_t getCapabilities (ROUTER* inst, void* router_session); static uint8_t getCapabilities (ROUTER* inst, void* router_session);
#if defined(PREP_STMT_CACHING)
static prep_stmt_t* prep_stmt_init(prep_stmt_type_t type, void* id);
static void prep_stmt_done(prep_stmt_t* pstmt);
#endif /*< PREP_STMT_CACHING */
int bref_cmp_global_conn( int bref_cmp_global_conn(
const void* bref1, const void* bref1,
const void* bref2); const void* bref2);
@ -207,8 +213,7 @@ static bool sescmd_cursor_next(
static GWBUF* sescmd_cursor_process_replies( static GWBUF* sescmd_cursor_process_replies(
DCB* client_dcb, DCB* client_dcb,
GWBUF* replybuf, GWBUF* replybuf,
sescmd_cursor_t* scur, backend_ref_t* bref);
bool* has_query);
static void tracelog_routed_query( static void tracelog_routed_query(
ROUTER_CLIENT_SES* rses, ROUTER_CLIENT_SES* rses,
@ -229,8 +234,15 @@ static void refreshInstance(
static void bref_clear_state(backend_ref_t* bref, bref_state_t state); static void bref_clear_state(backend_ref_t* bref, bref_state_t state);
static void bref_set_state(backend_ref_t* bref, bref_state_t state); static void bref_set_state(backend_ref_t* bref, bref_state_t state);
static sescmd_cursor_t* backend_ref_get_sescmd_cursor (backend_ref_t* bref);
static int router_handle_state_switch(DCB* dcb, DCB_REASON reason, void* data); static int router_handle_state_switch(DCB* dcb, DCB_REASON reason, void* data);
static bool handle_error_new_connection(
ROUTER_INSTANCE* inst,
ROUTER_CLIENT_SES* rses,
DCB* backend_dcb,
GWBUF* errmsg);
static bool handle_error_reply_client(SESSION* ses, GWBUF* errmsg);
static SPINLOCK instlock; static SPINLOCK instlock;
static ROUTER_INSTANCE* instances; static ROUTER_INSTANCE* instances;
@ -482,8 +494,6 @@ static void* newSession(
} }
/** Copy config struct from router instance */ /** Copy config struct from router instance */
client_rses->rses_config = router->rwsplit_config; client_rses->rses_config = router->rwsplit_config;
/** Create ID for the new client (router_client_ses) session */
client_rses->rses_id = router_client_ses_seq += 1;
spinlock_release(&router->lock); spinlock_release(&router->lock);
/** /**
@ -571,7 +581,6 @@ static void* newSession(
backend_ref[i].bref_sescmd_cur.scmd_cur_chk_tail = CHK_NUM_SESCMD_CUR; backend_ref[i].bref_sescmd_cur.scmd_cur_chk_tail = CHK_NUM_SESCMD_CUR;
#endif #endif
backend_ref[i].bref_state = 0; backend_ref[i].bref_state = 0;
bref_set_state(&backend_ref[i], BREF_NOT_USED);
backend_ref[i].bref_backend = router->servers[i]; backend_ref[i].bref_backend = router->servers[i];
/** store pointers to sescmd list to both cursors */ /** store pointers to sescmd list to both cursors */
backend_ref[i].bref_sescmd_cur.scmd_cur_rses = client_rses; backend_ref[i].bref_sescmd_cur.scmd_cur_rses = client_rses;
@ -814,6 +823,7 @@ static bool get_dcb(
*p_dcb = backend_ref[i].bref_dcb; *p_dcb = backend_ref[i].bref_dcb;
smallest_nconn = b->backend_conn_count; smallest_nconn = b->backend_conn_count;
succp = true; succp = true;
ss_dassert(backend_ref[i].bref_dcb->state != DCB_STATE_ZOMBIE);
} }
} }
@ -825,6 +835,7 @@ static bool get_dcb(
{ {
*p_dcb = backend_ref->bref_dcb; *p_dcb = backend_ref->bref_dcb;
succp = true; succp = true;
ss_dassert(backend_ref->bref_dcb->state != DCB_STATE_ZOMBIE);
ss_dassert( ss_dassert(
SERVER_IS_MASTER(backend_ref->bref_backend->backend_server) && SERVER_IS_MASTER(backend_ref->bref_backend->backend_server) &&
@ -892,7 +903,7 @@ static int routeQuery(
GWBUF* plainsqlbuf = NULL; GWBUF* plainsqlbuf = NULL;
char* querystr = NULL; char* querystr = NULL;
char* startpos; char* startpos;
unsigned char packet_type; mysql_server_cmd_t packet_type;
uint8_t* packet; uint8_t* packet;
int ret = 0; int ret = 0;
DCB* master_dcb = NULL; DCB* master_dcb = NULL;
@ -901,6 +912,7 @@ static int routeQuery(
ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session; ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
bool rses_is_closed = false; bool rses_is_closed = false;
size_t len; size_t len;
MYSQL* mysql = NULL;
CHK_CLIENT_RSES(router_cli_ses); CHK_CLIENT_RSES(router_cli_ses);
@ -915,10 +927,10 @@ static int routeQuery(
if (rses_is_closed) if (rses_is_closed)
{ {
/** /**
* COM_QUIT may have sent by client and as a part of backend * MYSQL_COM_QUIT may have sent by client and as a part of backend
* closing procedure. * closing procedure.
*/ */
if (packet_type != COM_QUIT) if (packet_type != MYSQL_COM_QUIT)
{ {
LOGIF(LE, LOGIF(LE,
(skygw_log_write_flush( (skygw_log_write_flush(
@ -941,21 +953,24 @@ static int routeQuery(
CHK_DCB(master_dcb); CHK_DCB(master_dcb);
switch(packet_type) { switch(packet_type) {
case COM_QUIT: /**< 1 QUIT will close all sessions */ case MYSQL_COM_QUIT: /*< 1 QUIT will close all sessions */
case COM_INIT_DB: /**< 2 DDL must go to the master */ case MYSQL_COM_INIT_DB: /*< 2 DDL must go to the master */
case COM_REFRESH: /**< 7 - I guess this is session but not sure */ case MYSQL_COM_REFRESH: /*< 7 - I guess this is session but not sure */
case COM_DEBUG: /**< 0d all servers dump debug info to stdout */ case MYSQL_COM_DEBUG: /*< 0d all servers dump debug info to stdout */
case COM_PING: /**< 0e all servers are pinged */ case MYSQL_COM_PING: /*< 0e all servers are pinged */
case COM_CHANGE_USER: /**< 11 all servers change it accordingly */ case MYSQL_COM_CHANGE_USER: /*< 11 all servers change it accordingly */
case MYSQL_COM_STMT_CLOSE: /*< free prepared statement */
case MYSQL_COM_STMT_SEND_LONG_DATA: /*< send data to column */
case MYSQL_COM_STMT_RESET: /*< resets the data of a prepared statement */
qtype = QUERY_TYPE_SESSION_WRITE; qtype = QUERY_TYPE_SESSION_WRITE;
break; break;
case COM_CREATE_DB: /**< 5 DDL must go to the master */ case MYSQL_COM_CREATE_DB: /**< 5 DDL must go to the master */
case COM_DROP_DB: /**< 6 DDL must go to the master */ case MYSQL_COM_DROP_DB: /**< 6 DDL must go to the master */
qtype = QUERY_TYPE_WRITE; qtype = QUERY_TYPE_WRITE;
break; break;
case COM_QUERY: case MYSQL_COM_QUERY:
plainsqlbuf = gwbuf_clone_transform(querybuf, plainsqlbuf = gwbuf_clone_transform(querybuf,
GWBUF_TYPE_PLAINSQL); GWBUF_TYPE_PLAINSQL);
len = GWBUF_LENGTH(plainsqlbuf); len = GWBUF_LENGTH(plainsqlbuf);
@ -963,24 +978,45 @@ static int routeQuery(
querystr = (char *)malloc(len+1); querystr = (char *)malloc(len+1);
memcpy(querystr, startpos, len); memcpy(querystr, startpos, len);
memset(&querystr[len], 0, 1); memset(&querystr[len], 0, 1);
// querystr = (char *)GWBUF_DATA(plainsqlbuf); /**
/* * Use mysql handle to query information from parse tree.
* querystr = master_dcb->func.getquerystr( * call skygw_query_classifier_free before exit!
* (void *) gwbuf_clone(querybuf),
* &querystr_is_copy);
*/ */
qtype = skygw_query_classifier_get_type(querystr, 0, &mysql);
qtype = skygw_query_classifier_get_type(querystr, 0);
break; break;
case COM_SHUTDOWN: /**< 8 where should shutdown be routed ? */ case MYSQL_COM_STMT_PREPARE:
case COM_STATISTICS: /**< 9 ? */ plainsqlbuf = gwbuf_clone_transform(querybuf,
case COM_PROCESS_INFO: /**< 0a ? */ GWBUF_TYPE_PLAINSQL);
case COM_CONNECT: /**< 0b ? */ len = GWBUF_LENGTH(plainsqlbuf);
case COM_PROCESS_KILL: /**< 0c ? */ /** unnecessary if buffer includes additional terminating null */
case COM_TIME: /**< 0f should this be run in gateway ? */ querystr = (char *)malloc(len+1);
case COM_DELAYED_INSERT: /**< 10 ? */ memcpy(querystr, startpos, len);
case COM_DAEMON: /**< 1d ? */ memset(&querystr[len], 0, 1);
qtype = skygw_query_classifier_get_type(querystr, 0, &mysql);
qtype |= QUERY_TYPE_PREPARE_STMT;
break;
case MYSQL_COM_STMT_EXECUTE:
plainsqlbuf = gwbuf_clone_transform(querybuf,
GWBUF_TYPE_PLAINSQL);
len = GWBUF_LENGTH(plainsqlbuf);
/** unnecessary if buffer includes additional terminating null */
querystr = (char *)malloc(len+1);
memcpy(querystr, startpos, len);
memset(&querystr[len], 0, 1);
qtype = skygw_query_classifier_get_type(querystr, 0, &mysql);
qtype |= QUERY_TYPE_EXEC_STMT;
break;
case MYSQL_COM_SHUTDOWN: /**< 8 where should shutdown be routed ? */
case MYSQL_COM_STATISTICS: /**< 9 ? */
case MYSQL_COM_PROCESS_INFO: /**< 0a ? */
case MYSQL_COM_CONNECT: /**< 0b ? */
case MYSQL_COM_PROCESS_KILL: /**< 0c ? */
case MYSQL_COM_TIME: /**< 0f should this be run in gateway ? */
case MYSQL_COM_DELAYED_INSERT: /**< 10 ? */
case MYSQL_COM_DAEMON: /**< 1d ? */
default: default:
break; break;
} /**< switch by packet type */ } /**< switch by packet type */
@ -1024,11 +1060,13 @@ static int routeQuery(
/** /**
* Session update is always routed in the same way. * Session update is always routed in the same way.
*/ */
if (QUERY_IS_TYPE(qtype, QUERY_TYPE_SESSION_WRITE)) if (QUERY_IS_TYPE(qtype, QUERY_TYPE_SESSION_WRITE) ||
QUERY_IS_TYPE(qtype, QUERY_TYPE_PREPARE_STMT) ||
QUERY_IS_TYPE(qtype, QUERY_TYPE_PREPARE_NAMED_STMT))
{ {
/** /**
* It is not sure if the session command in question requires * It is not sure if the session command in question requires
* response. Statement must be examined in route_session_write. * response. Statement is examined in route_session_write.
*/ */
bool succp = route_session_write( bool succp = route_session_write(
router_cli_ses, router_cli_ses,
@ -1050,9 +1088,8 @@ static int routeQuery(
LOGIF(LT, (skygw_log_write( LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE, LOGFILE_TRACE,
"[%s.%d]\tRead-only query, routing to Slave.", "[%s]\tRead-only query, routing to Slave.",
inst->service->name, inst->service->name)));
router_cli_ses->rses_id)));
ss_dassert(QUERY_IS_TYPE(qtype, QUERY_TYPE_READ)); ss_dassert(QUERY_IS_TYPE(qtype, QUERY_TYPE_READ));
/** Lock router session */ /** Lock router session */
@ -1060,7 +1097,6 @@ static int routeQuery(
{ {
goto return_ret; goto return_ret;
} }
succp = get_dcb(&slave_dcb, router_cli_ses, BE_SLAVE); succp = get_dcb(&slave_dcb, router_cli_ses, BE_SLAVE);
if (succp) if (succp)
@ -1152,6 +1188,10 @@ return_ret:
{ {
free(querystr); free(querystr);
} }
if (mysql != NULL)
{
skygw_query_classifier_free(mysql);
}
return ret; return ret;
} }
@ -1279,8 +1319,7 @@ static void clientReply(
DCB* client_dcb; DCB* client_dcb;
ROUTER_CLIENT_SES* router_cli_ses; ROUTER_CLIENT_SES* router_cli_ses;
sescmd_cursor_t* scur = NULL; sescmd_cursor_t* scur = NULL;
backend_ref_t* backend_ref; backend_ref_t* bref;
int i;
router_cli_ses = (ROUTER_CLIENT_SES *)router_session; router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
CHK_CLIENT_RSES(router_cli_ses); CHK_CLIENT_RSES(router_cli_ses);
@ -1323,23 +1362,10 @@ static void clientReply(
/** Log to debug that router was closed */ /** Log to debug that router was closed */
goto lock_failed; goto lock_failed;
} }
backend_ref = router_cli_ses->rses_backend_ref; bref = get_bref_from_dcb(router_cli_ses, backend_dcb);
/** find backend_dcb's corresponding BACKEND */ CHK_BACKEND_REF(bref);
i = 0; scur = &bref->bref_sescmd_cur;
while (i<router_cli_ses->rses_nbackends &&
backend_ref[i].bref_dcb != backend_dcb)
{
i++;
}
ss_dassert(backend_ref[i].bref_dcb == backend_dcb);
LOGIF(LT, tracelog_routed_query(router_cli_ses,
"reply_by_statement",
&backend_ref[i],
gwbuf_clone(writebuf)));
scur = &backend_ref[i].bref_sescmd_cur;
/** /**
* Active cursor means that reply is from session command * Active cursor means that reply is from session command
* execution. Majority of the time there are no session commands * execution. Majority of the time there are no session commands
@ -1347,34 +1373,76 @@ static void clientReply(
*/ */
if (sescmd_cursor_is_active(scur)) if (sescmd_cursor_is_active(scur))
{ {
bool has_query; if (MYSQL_IS_ERROR_PACKET(((uint8_t *)GWBUF_DATA(writebuf))))
writebuf = sescmd_cursor_process_replies(client_dcb,
writebuf,
scur,
&has_query);
if (has_query)
{ {
bref_clear_state(backend_ref, BREF_WAITING_RESULT); SESSION* ses = backend_dcb->session;
} uint8_t* buf =
} (uint8_t *)GWBUF_DATA((scur->scmd_cur_cmd->my_sescmd_buf));
size_t len = MYSQL_GET_PACKET_LEN(buf);
char* cmdstr = (char *)malloc(len+1);
ROUTER_INSTANCE* inst = (ROUTER_INSTANCE *)instance;
snprintf(cmdstr, len+1, "%s", &buf[5]);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Failed to execute %s in %s:%d.",
cmdstr,
bref->bref_backend->backend_server->name,
bref->bref_backend->backend_server->port)));
free(cmdstr);
/** Inform the client */
handle_error_reply_client(ses,writebuf);
/** Unlock router session */ /** Unlock router session */
rses_end_locked_router_action(router_cli_ses); rses_end_locked_router_action(router_cli_ses);
goto lock_failed;
}
else
{
/**
* Discard all those responses that have already been sent to
* the client. Return with buffer including response that
* needs to be sent to client or NULL.
*/
writebuf = sescmd_cursor_process_replies(client_dcb,
writebuf,
bref);
}
}
if (writebuf != NULL && client_dcb != NULL) if (writebuf != NULL && client_dcb != NULL)
{ {
/** Write reply to client DCB */ /** Write reply to client DCB */
SESSION_ROUTE_REPLY(backend_dcb->session, writebuf); SESSION_ROUTE_REPLY(backend_dcb->session, writebuf);
bref_clear_state(bref, BREF_WAITING_RESULT);
LOGIF(LT, (skygw_log_write_flush(
LOGFILE_TRACE,
"%lu [clientReply:rwsplit] client dcb %p, "
"backend dcb %p. End of normal reply.",
pthread_self(),
client_dcb,
backend_dcb)));
bref_clear_state(backend_ref, BREF_WAITING_RESULT);
} }
/** Unlock router session */
rses_end_locked_router_action(router_cli_ses);
/** Lock router session */
if (!rses_begin_locked_router_action(router_cli_ses))
{
/** Log to debug that router was closed */
goto lock_failed;
}
/** There is one pending session command to be xexecuted. */
if (sescmd_cursor_is_active(scur))
{
bool succp;
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Backend %s:%d processed reply and starts to execute "
"active cursor.",
bref->bref_backend->backend_server->name,
bref->bref_backend->backend_server->port)));
succp = execute_sescmd_in_backend(bref);
}
/** Unlock router session */
rses_end_locked_router_action(router_cli_ses);
lock_failed: lock_failed:
return; return;
@ -1423,6 +1491,13 @@ static void bref_set_state(
bref_state_t state) bref_state_t state)
{ {
bref->bref_state |= state; bref->bref_state |= state;
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Set state %d for %s:%d fd %d",
bref->bref_state,
bref->bref_backend->backend_server->name,
bref->bref_backend->backend_server->port,
bref->bref_dcb->fd)));
} }
/** /**
@ -1603,7 +1678,7 @@ static bool select_connect_backend_servers(
{ {
BACKEND* b = backend_ref[i].bref_backend; BACKEND* b = backend_ref[i].bref_backend;
LOGIF(LT, (skygw_log_write( LOGIF(LT, (skygw_log_write_flush(
LOGFILE_TRACE, LOGFILE_TRACE,
"Examine server " "Examine server "
"%s:%d %s with %d connections. " "%s:%d %s with %d connections. "
@ -1645,18 +1720,15 @@ static bool select_connect_backend_servers(
*/ */
execute_sescmd_history(&backend_ref[i]); execute_sescmd_history(&backend_ref[i]);
/** /**
* Callback which is called when * When server fails, this callback
* node fails. * is called.
*/ */
dcb_add_callback( dcb_add_callback(
backend_ref[i].bref_dcb, backend_ref[i].bref_dcb,
DCB_REASON_NOT_RESPONDING, DCB_REASON_NOT_RESPONDING,
&router_handle_state_switch, &router_handle_state_switch,
(void *)&backend_ref[i]); (void *)&backend_ref[i]);
bref_clear_state(&backend_ref[i], backend_ref[i].bref_state = 0;
BREF_CLOSED);
bref_clear_state(&backend_ref[i],
BREF_NOT_USED);
bref_set_state(&backend_ref[i], bref_set_state(&backend_ref[i],
BREF_IN_USE); BREF_IN_USE);
/** /**
@ -1698,20 +1770,20 @@ static bool select_connect_backend_servers(
if (backend_ref[i].bref_dcb != NULL) if (backend_ref[i].bref_dcb != NULL)
{ {
master_connected = true; master_connected = true;
/**
* When server fails, this callback
* is called.
*/
dcb_add_callback( dcb_add_callback(
backend_ref[i].bref_dcb, backend_ref[i].bref_dcb,
DCB_REASON_NOT_RESPONDING, DCB_REASON_NOT_RESPONDING,
&router_handle_state_switch, &router_handle_state_switch,
(void *)&backend_ref[i]); (void *)&backend_ref[i]);
bref_clear_state(&backend_ref[i], backend_ref[i].bref_state = 0;
BREF_NOT_USED);
bref_set_state(&backend_ref[i], bref_set_state(&backend_ref[i],
BREF_IN_USE); BREF_IN_USE);
/** Increase backend connection counters */
/** Increase backend connection counter */
/** Increase backend connection counter */
atomic_add(&b->backend_server->stats.n_current, 1); atomic_add(&b->backend_server->stats.n_current, 1);
atomic_add(&b->backend_server->stats.n_connections, 1); atomic_add(&b->backend_server->stats.n_connections, 1);
atomic_add(&b->backend_conn_count, 1); atomic_add(&b->backend_conn_count, 1);
@ -1725,7 +1797,7 @@ static bool select_connect_backend_servers(
"connection with master %s:%d", "connection with master %s:%d",
b->backend_server->name, b->backend_server->name,
b->backend_server->port))); b->backend_server->port)));
/* handle connect error */ /** handle connect error */
} }
} }
} }
@ -1905,8 +1977,7 @@ static bool select_connect_backend_servers(
/** disconnect opened connections */ /** disconnect opened connections */
dcb_close(backend_ref[i].bref_dcb); dcb_close(backend_ref[i].bref_dcb);
bref_clear_state(&backend_ref[i], BREF_IN_USE); bref_clear_state(&backend_ref[i], BREF_IN_USE);
bref_set_state(&backend_ref[i], BREF_NOT_USED); /** Decrease backend's connection counter. */
atomic_add(&backend_ref[i].bref_backend->backend_conn_count, -1); atomic_add(&backend_ref[i].bref_backend->backend_conn_count, -1);
} }
} }
@ -1956,7 +2027,7 @@ static void rses_property_done(
mysql_sescmd_done(&prop->rses_prop_data.sescmd); mysql_sescmd_done(&prop->rses_prop_data.sescmd);
break; break;
default: default:
LOGIF(LD, (skygw_log_write_flush( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [rses_property_done] Unknown property type %d " "%lu [rses_property_done] Unknown property type %d "
"in property %p", "in property %p",
@ -2100,16 +2171,16 @@ static void mysql_sescmd_done(
static GWBUF* sescmd_cursor_process_replies( static GWBUF* sescmd_cursor_process_replies(
DCB* client_dcb, DCB* client_dcb,
GWBUF* replybuf, GWBUF* replybuf,
sescmd_cursor_t* scur, backend_ref_t* bref)
bool* has_query)
{ {
const size_t headerlen = 4; /*< mysql packet header */ const size_t headerlen = 4; /*< mysql packet header */
uint8_t* packet;
size_t packetlen; size_t packetlen;
uint8_t* packet;
mysql_sescmd_t* scmd; mysql_sescmd_t* scmd;
sescmd_cursor_t* scur;
scur = &bref->bref_sescmd_cur;
ss_dassert(SPINLOCK_IS_LOCKED(&(scur->scmd_cur_rses->rses_lock))); ss_dassert(SPINLOCK_IS_LOCKED(&(scur->scmd_cur_rses->rses_lock)));
scmd = sescmd_cursor_get_command(scur); scmd = sescmd_cursor_get_command(scur);
CHK_DCB(client_dcb); CHK_DCB(client_dcb);
@ -2121,18 +2192,41 @@ static GWBUF* sescmd_cursor_process_replies(
*/ */
while (scmd != NULL && replybuf != NULL) while (scmd != NULL && replybuf != NULL)
{ {
/** Faster backend has already responded to client : discard */
if (scmd->my_sescmd_is_replied) if (scmd->my_sescmd_is_replied)
{ {
/**
* Discard heading packets if their related command is
* already replied.
*/
CHK_GWBUF(replybuf); CHK_GWBUF(replybuf);
packet = (uint8_t *)GWBUF_DATA(replybuf); packet = (uint8_t *)GWBUF_DATA(replybuf);
/**
* If it is response to MYSQL_COM_STMT_PREPARE, then buffer
* only includes the response.
*/
if (scmd->my_sescmd_prop->rses_prop_data.sescmd.my_sescmd_packet_type ==
MYSQL_COM_STMT_PREPARE)
{
while (replybuf != NULL)
{
#if defined(SS_DEBUG)
int buflen;
buflen = GWBUF_LENGTH(replybuf);
replybuf = gwbuf_consume(replybuf, buflen);
#else
replybuf = gwbuf_consume(
replybuf,
GWBUF_LENGTH(replybuf));
#endif
}
}
/** Only consume the leading packet */
else
{
packetlen = packet[0]+packet[1]*256+packet[2]*256*256; packetlen = packet[0]+packet[1]*256+packet[2]*256*256;
replybuf = gwbuf_consume(replybuf, packetlen+headerlen); replybuf = gwbuf_consume(replybuf, packetlen+headerlen);
} }
else }
/** Response is in the buffer and it will be sent to client. */
else if (replybuf != NULL)
{ {
/** Mark the rest session commands as replied */ /** Mark the rest session commands as replied */
scmd->my_sescmd_is_replied = true; scmd->my_sescmd_is_replied = true;
@ -2149,8 +2243,6 @@ static GWBUF* sescmd_cursor_process_replies(
scur->scmd_cur_active = false; scur->scmd_cur_active = false;
} }
} }
/** vraa:this is set but only because there's not yet way to find out */
*has_query = false;
ss_dassert(replybuf == NULL || *scur->scmd_cur_ptr_property == NULL); ss_dassert(replybuf == NULL || *scur->scmd_cur_ptr_property == NULL);
return replybuf; return replybuf;
@ -2287,12 +2379,13 @@ static bool execute_sescmd_in_backend(
backend_ref_t* backend_ref) backend_ref_t* backend_ref)
{ {
DCB* dcb; DCB* dcb;
bool succp = true; bool succp;
int rc = 0; int rc = 0;
sescmd_cursor_t* scur; sescmd_cursor_t* scur;
if (BREF_IS_CLOSED(backend_ref)) if (BREF_IS_CLOSED(backend_ref))
{ {
succp = false;
goto return_succp; goto return_succp;
} }
dcb = backend_ref->bref_dcb; dcb = backend_ref->bref_dcb;
@ -2309,6 +2402,10 @@ static bool execute_sescmd_in_backend(
if (sescmd_cursor_get_command(scur) == NULL) if (sescmd_cursor_get_command(scur) == NULL)
{ {
succp = false; succp = false;
LOGIF(LT, (skygw_log_write_flush(
LOGFILE_TRACE,
"Cursor had no pending session commands.")));
goto return_succp; goto return_succp;
} }
@ -2317,14 +2414,26 @@ static bool execute_sescmd_in_backend(
/** Cursor is left active when function returns. */ /** Cursor is left active when function returns. */
sescmd_cursor_set_active(scur, true); sescmd_cursor_set_active(scur, true);
} }
#if defined(SS_DEBUG)
LOGIF(LT, tracelog_routed_query(scur->scmd_cur_rses, LOGIF(LT, tracelog_routed_query(scur->scmd_cur_rses,
"execute_sescmd_in_backend", "execute_sescmd_in_backend",
backend_ref, backend_ref,
sescmd_cursor_clone_querybuf(scur))); sescmd_cursor_clone_querybuf(scur)));
{
GWBUF* tmpbuf = sescmd_cursor_clone_querybuf(scur);
uint8_t* ptr = GWBUF_DATA(tmpbuf);
unsigned char cmd = MYSQL_GET_COMMAND(ptr);
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Just before write, fd %d : cmd %s.",
dcb->fd,
STRPACKETTYPE(cmd))));
}
#endif
switch (scur->scmd_cur_cmd->my_sescmd_packet_type) { switch (scur->scmd_cur_cmd->my_sescmd_packet_type) {
case COM_CHANGE_USER: case MYSQL_COM_CHANGE_USER:
rc = dcb->func.auth( rc = dcb->func.auth(
dcb, dcb,
NULL, NULL,
@ -2332,8 +2441,8 @@ static bool execute_sescmd_in_backend(
sescmd_cursor_clone_querybuf(scur)); sescmd_cursor_clone_querybuf(scur));
break; break;
case COM_QUERY: case MYSQL_COM_QUERY:
case COM_INIT_DB: case MYSQL_COM_INIT_DB:
default: default:
rc = dcb->func.write( rc = dcb->func.write(
dcb, dcb,
@ -2349,10 +2458,7 @@ static bool execute_sescmd_in_backend(
if (rc == 1) if (rc == 1)
{ {
/** succp = true;
* All but COM_QUIT cause backend to send reply. flag backend_ref.
*/
bref_set_state(backend_ref, BREF_WAITING_RESULT);
} }
else else
{ {
@ -2464,7 +2570,7 @@ static void tracelog_routed_query(
be_type = BACKEND_TYPE(b); be_type = BACKEND_TYPE(b);
if (GWBUF_TYPE(buf) == GWBUF_TYPE_MYSQL) if (GWBUF_IS_TYPE_MYSQL(buf))
{ {
len = packet[0]; len = packet[0];
len += 256*packet[1]; len += 256*packet[1];
@ -2488,6 +2594,28 @@ static void tracelog_routed_query(
dcb))); dcb)));
free(querystr); free(querystr);
} }
else if (packet_type == '\x22' ||
packet_type == 0x22 ||
packet_type == '\x26' ||
packet_type == 0x26 ||
true)
{
querystr = (char *)malloc(len);
memcpy(querystr, startpos, len-1);
querystr[len-1] = '\0';
LOGIF(LT, (skygw_log_write_flush(
LOGFILE_TRACE,
"%lu [%s] %d bytes long buf, \"%s\" -> %s:%d %s dcb %p",
pthread_self(),
funcname,
buflen,
querystr,
b->backend_server->name,
b->backend_server->port,
STRBETYPE(be_type),
dcb)));
free(querystr);
}
} }
gwbuf_free(buf); gwbuf_free(buf);
} }
@ -2551,12 +2679,14 @@ static bool route_session_write(
backend_ref = router_cli_ses->rses_backend_ref; backend_ref = router_cli_ses->rses_backend_ref;
/** /**
* COM_QUIT is one-way message. Server doesn't respond to that. * These are one-way messages and server doesn't respond to them.
* Therefore reply processing is unnecessary and session * Therefore reply processing is unnecessary and session
* command property is not needed. It is just routed to both * command property is not needed. It is just routed to all available
* backends. * backends.
*/ */
if (packet_type == COM_QUIT) if (packet_type == MYSQL_COM_STMT_SEND_LONG_DATA ||
packet_type == MYSQL_COM_QUIT ||
packet_type == MYSQL_COM_STMT_CLOSE)
{ {
int rc; int rc;
@ -2587,6 +2717,14 @@ static bool route_session_write(
gwbuf_free(querybuf); gwbuf_free(querybuf);
goto return_succp; goto return_succp;
} }
/** Lock router session */
if (!rses_begin_locked_router_action(router_cli_ses))
{
rses_property_done(prop);
succp = false;
goto return_succp;
}
prop = rses_property_init(RSES_PROP_TYPE_SESCMD); prop = rses_property_init(RSES_PROP_TYPE_SESCMD);
/** /**
* Additional reference is created to querybuf to * Additional reference is created to querybuf to
@ -2595,19 +2733,40 @@ static bool route_session_write(
*/ */
mysql_sescmd_init(prop, querybuf, packet_type, router_cli_ses); 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 */ /** Add sescmd property to router client session */
rses_property_add(router_cli_ses, prop); rses_property_add(router_cli_ses, prop);
for (i=0; i<router_cli_ses->rses_nbackends; i++) for (i=0; i<router_cli_ses->rses_nbackends; i++)
{ {
if (BREF_IS_IN_USE((&backend_ref[i]))) if (BREF_IS_IN_USE((&backend_ref[i])))
{
sescmd_cursor_t* scur;
scur = backend_ref_get_sescmd_cursor(&backend_ref[i]);
/**
* This backend_ref waits reply, flag it.
*/
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.
* Otherwise, cursor will execute pending commands
* when it completes with previous commands.
*/
if (sescmd_cursor_is_active(scur))
{
succp = true;
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Backend %s:%d already executing sescmd.",
backend_ref[i].bref_backend->backend_server->name,
backend_ref[i].bref_backend->backend_server->port)));
}
else
{ {
succp = execute_sescmd_in_backend(&backend_ref[i]); succp = execute_sescmd_in_backend(&backend_ref[i]);
@ -2622,6 +2781,7 @@ static bool route_session_write(
} }
} }
} }
}
/** Unlock router session */ /** Unlock router session */
rses_end_locked_router_action(router_cli_ses); rses_end_locked_router_action(router_cli_ses);
@ -2696,6 +2856,7 @@ static void rwsplit_process_options(
* Even if succp == true connecting to new slave may have failed. succp is to * Even if succp == true connecting to new slave may have failed. succp is to
* tell whether router has enough master/slave connections to continue work. * tell whether router has enough master/slave connections to continue work.
*/ */
static void handleError ( static void handleError (
ROUTER* instance, ROUTER* instance,
void* router_session, void* router_session,
@ -2719,10 +2880,6 @@ static void handleError (
switch (action) { switch (action) {
case ERRACT_NEW_CONNECTION: case ERRACT_NEW_CONNECTION:
{ {
int router_nservers;
int max_nslaves;
backend_ref_t* bref;
CHK_CLIENT_RSES(rses); CHK_CLIENT_RSES(rses);
if (!rses_begin_locked_router_action(rses)) if (!rses_begin_locked_router_action(rses))
@ -2731,14 +2888,78 @@ static void handleError (
return; return;
} }
*succp = handle_error_new_connection(inst,
rses,
backend_dcb,
errmsgbuf);
rses_end_locked_router_action(rses);
break;
}
case ERRACT_REPLY_CLIENT:
{
*succp = handle_error_reply_client(session, errmsgbuf);
break;
}
default:
*succp = false;
break;
}
}
static bool handle_error_reply_client(
SESSION* ses,
GWBUF* errmsg)
{
session_state_t sesstate;
DCB* client_dcb;
bool succp;
spinlock_acquire(&ses->ses_lock);
sesstate = ses->state;
client_dcb = ses->client;
spinlock_release(&ses->ses_lock);
if (sesstate == SESSION_STATE_ROUTER_READY)
{
CHK_DCB(client_dcb);
client_dcb->func.write(client_dcb, errmsg);
}
succp = false; /** false because new servers aren's selected. */
return succp;
}
/**
* This must be called with router lock
*/
static bool handle_error_new_connection(
ROUTER_INSTANCE* inst,
ROUTER_CLIENT_SES* rses,
DCB* backend_dcb,
GWBUF* errmsg)
{
SESSION* ses;
int router_nservers;
int max_nslaves;
backend_ref_t* bref;
bool succp;
ss_dassert(SPINLOCK_IS_LOCKED(&rses->rses_lock));
ses = backend_dcb->session;
CHK_SESSION(ses);
bref = get_bref_from_dcb(rses, backend_dcb); bref = get_bref_from_dcb(rses, backend_dcb);
/** failed DCB has already been replaced */ /** failed DCB has already been replaced */
if (bref == NULL) if (bref == NULL)
{ {
rses_end_locked_router_action(rses); rses_end_locked_router_action(rses);
*succp = true; succp = true;
return; goto return_succp;
} }
/** /**
* Error handler is already called for this DCB because * Error handler is already called for this DCB because
@ -2748,8 +2969,8 @@ static void handleError (
if (backend_dcb->state != DCB_STATE_POLLING) if (backend_dcb->state != DCB_STATE_POLLING)
{ {
rses_end_locked_router_action(rses); rses_end_locked_router_action(rses);
*succp = true; succp = true;
return; goto return_succp;
} }
CHK_BACKEND_REF(bref); CHK_BACKEND_REF(bref);
@ -2757,12 +2978,11 @@ static void handleError (
if (BREF_IS_WAITING_RESULT(bref)) if (BREF_IS_WAITING_RESULT(bref))
{ {
DCB* client_dcb; DCB* client_dcb;
client_dcb = session->client; client_dcb = ses->client;
client_dcb->func.write(client_dcb, errmsgbuf); client_dcb->func.write(client_dcb, errmsg);
bref_clear_state(bref, BREF_WAITING_RESULT); bref_clear_state(bref, BREF_WAITING_RESULT);
} }
bref_clear_state(bref, BREF_IN_USE); bref_clear_state(bref, BREF_IN_USE);
bref_set_state(bref, BREF_NOT_USED);
bref_set_state(bref, BREF_CLOSED); bref_set_state(bref, BREF_CLOSED);
/** /**
* Remove callback because this DCB won't be used * Remove callback because this DCB won't be used
@ -2780,42 +3000,19 @@ static void handleError (
* Try to get replacement slave or at least the minimum * Try to get replacement slave or at least the minimum
* number of slave connections for router session. * number of slave connections for router session.
*/ */
*succp = select_connect_backend_servers( succp = select_connect_backend_servers(
&rses->rses_master_ref, &rses->rses_master_ref,
rses->rses_backend_ref, rses->rses_backend_ref,
router_nservers, router_nservers,
max_nslaves, max_nslaves,
rses->rses_config.rw_slave_select_criteria, rses->rses_config.rw_slave_select_criteria,
session, ses,
inst); inst);
rses_end_locked_router_action(rses); return_succp:
break; return succp;
} }
case ERRACT_REPLY_CLIENT:
{
session_state_t sesstate;
spinlock_acquire(&session->ses_lock);
sesstate = session->state;
client_dcb = session->client;
spinlock_release(&session->ses_lock);
if (sesstate == SESSION_STATE_ROUTER_READY)
{
CHK_DCB(client_dcb);
client_dcb->func.write(client_dcb, errmsgbuf);
}
succp = false; /** false because new servers aren's selected. */
break;
}
default:
*succp = false;
break;
}
}
static void print_error_packet( static void print_error_packet(
ROUTER_CLIENT_SES* rses, ROUTER_CLIENT_SES* rses,
@ -2823,7 +3020,7 @@ static void print_error_packet(
DCB* dcb) DCB* dcb)
{ {
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
if (buf->gwbuf_type == GWBUF_TYPE_MYSQL) if (GWBUF_IS_TYPE_MYSQL(buf))
{ {
while (gwbuf_length(buf) > 0) while (gwbuf_length(buf) > 0)
{ {
@ -2977,3 +3174,71 @@ static int router_handle_state_switch(
return_rc: return_rc:
return rc; return rc;
} }
static 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;
}
#if defined(PREP_STMT_CACHING)
#define MAX_STMT_LEN 1024
static prep_stmt_t* prep_stmt_init(
prep_stmt_type_t type,
void* id)
{
prep_stmt_t* pstmt;
pstmt = (prep_stmt_t *)calloc(1, sizeof(prep_stmt_t));
if (pstmt != NULL)
{
#if defined(SS_DEBUG)
pstmt->pstmt_chk_top = CHK_NUM_PREP_STMT;
pstmt->pstmt_chk_tail = CHK_NUM_PREP_STMT;
#endif
pstmt->pstmt_state = PREP_STMT_ALLOC;
pstmt->pstmt_type = type;
if (type == PREP_STMT_NAME)
{
pstmt->pstmt_id.name = strndup((char *)id, MAX_STMT_LEN);
}
else
{
pstmt->pstmt_id.seq = 0;
}
}
CHK_PREP_STMT(pstmt);
return pstmt;
}
static void prep_stmt_done(
prep_stmt_t* pstmt)
{
CHK_PREP_STMT(pstmt);
if (pstmt->pstmt_type == PREP_STMT_NAME)
{
free(pstmt->pstmt_id.name);
}
free(pstmt);
}
static bool prep_stmt_drop(
prep_stmt_t* pstmt)
{
CHK_PREP_STMT(pstmt);
pstmt->pstmt_state = PREP_STMT_DROPPED;
return true;
}
#endif /*< PREP_STMT_CACHING */

51
utils/skygw_debug.h
View File

@ -122,7 +122,8 @@ typedef enum skygw_chk_t {
CHK_NUM_ROUTER_PROPERTY, CHK_NUM_ROUTER_PROPERTY,
CHK_NUM_SESCMD_CUR, CHK_NUM_SESCMD_CUR,
CHK_NUM_BACKEND, CHK_NUM_BACKEND,
CHK_NUM_BACKEND_REF CHK_NUM_BACKEND_REF,
CHK_NUM_PREP_STMT
} skygw_chk_t; } skygw_chk_t;
# define STRBOOL(b) ((b) ? "true" : "false") # define STRBOOL(b) ((b) ? "true" : "false")
@ -140,23 +141,26 @@ typedef enum skygw_chk_t {
((i) == LOGFILE_DEBUG ? "LOGFILE_DEBUG" : \ ((i) == LOGFILE_DEBUG ? "LOGFILE_DEBUG" : \
"Unknown logfile type")))) "Unknown logfile type"))))
#define STRPACKETTYPE(p) ((p) == COM_INIT_DB ? "COM_INIT_DB" : \ #define STRPACKETTYPE(p) ((p) == MYSQL_COM_INIT_DB ? "COM_INIT_DB" : \
((p) == COM_CREATE_DB ? "COM_CREATE_DB" : \ ((p) == MYSQL_COM_CREATE_DB ? "COM_CREATE_DB" : \
((p) == COM_DROP_DB ? "COM_DROP_DB" : \ ((p) == MYSQL_COM_DROP_DB ? "COM_DROP_DB" : \
((p) == COM_REFRESH ? "COM_REFRESH" : \ ((p) == MYSQL_COM_REFRESH ? "COM_REFRESH" : \
((p) == COM_DEBUG ? "COM_DEBUG" : \ ((p) == MYSQL_COM_DEBUG ? "COM_DEBUG" : \
((p) == COM_PING ? "COM_PING" : \ ((p) == MYSQL_COM_PING ? "COM_PING" : \
((p) == COM_CHANGE_USER ? "COM_CHANGE_USER" : \ ((p) == MYSQL_COM_CHANGE_USER ? "COM_CHANGE_USER" : \
((p) == COM_QUERY ? "COM_QUERY" : \ ((p) == MYSQL_COM_QUERY ? "COM_QUERY" : \
((p) == COM_SHUTDOWN ? "COM_SHUTDOWN" : \ ((p) == MYSQL_COM_SHUTDOWN ? "COM_SHUTDOWN" : \
((p) == COM_PROCESS_INFO ? "COM_PROCESS_INFO" : \ ((p) == MYSQL_COM_PROCESS_INFO ? "COM_PROCESS_INFO" : \
((p) == COM_CONNECT ? "COM_CONNECT" : \ ((p) == MYSQL_COM_CONNECT ? "COM_CONNECT" : \
((p) == COM_PROCESS_KILL ? "COM_PROCESS_KILL" : \ ((p) == MYSQL_COM_PROCESS_KILL ? "COM_PROCESS_KILL" : \
((p) == COM_TIME ? "COM_TIME" : \ ((p) == MYSQL_COM_TIME ? "COM_TIME" : \
((p) == COM_DELAYED_INSERT ? "COM_DELAYED_INSERT" : \ ((p) == MYSQL_COM_DELAYED_INSERT ? "COM_DELAYED_INSERT" : \
((p) == COM_DAEMON ? "COM_DAEMON" : \ ((p) == MYSQL_COM_DAEMON ? "COM_DAEMON" : \
((p) == COM_QUIT ? "COM_QUIT" : \ ((p) == MYSQL_COM_QUIT ? "COM_QUIT" : \
"UNKNOWN MYSQL PACKET TYPE")))))))))))))))) ((p) == MYSQL_COM_STMT_PREPARE ? "MYSQL_COM_STMT_PREPARE" : \
((p) == MYSQL_COM_STMT_EXECUTE ? "MYSQL_COM_STMT_EXECUTE" : \
((p) == MYSQL_COM_UNDEFINED ? "MYSQL_COM_UNDEFINED" : \
"UNKNOWN MYSQL PACKET TYPE")))))))))))))))))))
#define STRDCBSTATE(s) ((s) == DCB_STATE_ALLOC ? "DCB_STATE_ALLOC" : \ #define STRDCBSTATE(s) ((s) == DCB_STATE_ALLOC ? "DCB_STATE_ALLOC" : \
((s) == DCB_STATE_POLLING ? "DCB_STATE_POLLING" : \ ((s) == DCB_STATE_POLLING ? "DCB_STATE_POLLING" : \
@ -180,10 +184,7 @@ typedef enum skygw_chk_t {
((s) == MYSQL_AUTH_RECV ? "MYSQL_AUTH_RECV" : \ ((s) == MYSQL_AUTH_RECV ? "MYSQL_AUTH_RECV" : \
((s) == MYSQL_AUTH_FAILED ? "MYSQL_AUTH_FAILED" : \ ((s) == MYSQL_AUTH_FAILED ? "MYSQL_AUTH_FAILED" : \
((s) == MYSQL_IDLE ? "MYSQL_IDLE" : \ ((s) == MYSQL_IDLE ? "MYSQL_IDLE" : \
((s) == MYSQL_ROUTING ? "MYSQL_ROUTING" : \ "UNKNOWN MYSQL STATE")))))))
((s) == MYSQL_WAITING_RESULT ? "MYSQL_WAITING_RESULT" : \
((s) == MYSQL_SESSION_CHANGE ? "MYSQL_SESSION_CHANGE" : \
"UNKNOWN MYSQL STATE"))))))))))
#define STRITEMTYPE(t) ((t) == Item::FIELD_ITEM ? "FIELD_ITEM" : \ #define STRITEMTYPE(t) ((t) == Item::FIELD_ITEM ? "FIELD_ITEM" : \
((t) == Item::FUNC_ITEM ? "FUNC_ITEM" : \ ((t) == Item::FUNC_ITEM ? "FUNC_ITEM" : \
@ -478,6 +479,12 @@ typedef enum skygw_chk_t {
"Backend reference has invalid check fields"); \ "Backend reference has invalid check fields"); \
} }
#define CHK_PREP_STMT(p) { \
ss_info_dassert((p)->pstmt_chk_top == CHK_NUM_PREP_STMT && \
(p)->pstmt_chk_tail == CHK_NUM_PREP_STMT, \
"Prepared statement struct has invalid check fields"); \
}
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
bool conn_open[10240]; bool conn_open[10240];