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

Merge branch 'release-1.0GA' into firewall

Browse Source
This commit is contained in:
Markus Makela
2014-12-16 11:10:02 +02:00
parent f2cec6e51e 82407ceaf4
commit 1e43102eea
16 changed files with 705 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
macros.cmake
View File

@ -11,7 +11,7 @@ macro(set_maxscale_version)
set(MAXSCALE_VERSION_MINOR "0") set(MAXSCALE_VERSION_MINOR "0")
set(MAXSCALE_VERSION_PATCH "2") set(MAXSCALE_VERSION_PATCH "2")
set(MAXSCALE_VERSION_NUMERIC "${MAXSCALE_VERSION_MAJOR}.${MAXSCALE_VERSION_MINOR}.${MAXSCALE_VERSION_PATCH}") set(MAXSCALE_VERSION_NUMERIC "${MAXSCALE_VERSION_MAJOR}.${MAXSCALE_VERSION_MINOR}.${MAXSCALE_VERSION_PATCH}")
set(MAXSCALE_VERSION "${MAXSCALE_VERSION_MAJOR}.${MAXSCALE_VERSION_MINOR}.${MAXSCALE_VERSION_PATCH}-beta") set(MAXSCALE_VERSION "${MAXSCALE_VERSION_MAJOR}.${MAXSCALE_VERSION_MINOR}.${MAXSCALE_VERSION_PATCH}-rc")
endmacro() endmacro()

6
query_classifier/query_classifier.cc
View File

@ -151,7 +151,7 @@ bool parse_query (
THD* thd; THD* thd;
uint8_t* data; uint8_t* data;
size_t len; size_t len;
char* query_str; char* query_str = NULL;
parsing_info_t* pi; parsing_info_t* pi;
CHK_GWBUF(querybuf); CHK_GWBUF(querybuf);
@ -173,9 +173,9 @@ bool parse_query (
/** Extract query and copy it to different buffer */ /** Extract query and copy it to different buffer */
data = (uint8_t*)GWBUF_DATA(querybuf); data = (uint8_t*)GWBUF_DATA(querybuf);
len = MYSQL_GET_PACKET_LEN(data)-1; /*< distract 1 for packet type byte */ len = MYSQL_GET_PACKET_LEN(data)-1; /*< distract 1 for packet type byte */
query_str = (char *)malloc(len+1);
if (query_str == NULL)
if (len < 1 || (query_str = (char *)malloc(len+1)) == NULL)
{ {
/** Free parsing info data */ /** Free parsing info data */
parsing_info_done(pi); parsing_info_done(pi);

5
server/core/buffer.c
View File

@ -179,7 +179,7 @@ gwbuf_clone(GWBUF *buf)
{ {
GWBUF *rval; GWBUF *rval;
if ((rval = (GWBUF *)malloc(sizeof(GWBUF))) == NULL) if ((rval = (GWBUF *)calloc(1,sizeof(GWBUF))) == NULL)
{ {
ss_dassert(rval != NULL); ss_dassert(rval != NULL);
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
@ -194,11 +194,8 @@ GWBUF *rval;
rval->start = buf->start; rval->start = buf->start;
rval->end = buf->end; rval->end = buf->end;
rval->gwbuf_type = buf->gwbuf_type; rval->gwbuf_type = buf->gwbuf_type;
rval->properties = NULL;
rval->hint = NULL;
rval->gwbuf_info = buf->gwbuf_info; rval->gwbuf_info = buf->gwbuf_info;
rval->gwbuf_bufobj = buf->gwbuf_bufobj; rval->gwbuf_bufobj = buf->gwbuf_bufobj;
rval->next = NULL;
rval->tail = rval; rval->tail = rval;
CHK_GWBUF(rval); CHK_GWBUF(rval);
return rval; return rval;

2
server/core/config.c
View File

@ -1814,7 +1814,7 @@ config_truth_value(char *str)
{ {
return 1; return 1;
} }
if (strcasecmp(str, "flase") == 0 || strcasecmp(str, "off") == 0) if (strcasecmp(str, "false") == 0 || strcasecmp(str, "off") == 0)
{ {
return 0; return 0;
} }

67
server/core/dcb.c
View File

@ -190,7 +190,7 @@ DCB *rval;
rval->readcheck = 0; rval->readcheck = 0;
rval->polloutbusy = 0; rval->polloutbusy = 0;
rval->writecheck = 0; rval->writecheck = 0;
rval->fd = -1; rval->fd = DCBFD_CLOSED;
rval->evq.next = NULL; rval->evq.next = NULL;
rval->evq.prev = NULL; rval->evq.prev = NULL;
@ -235,8 +235,10 @@ DCB *rval;
void void
dcb_free(DCB *dcb) dcb_free(DCB *dcb)
{ {
if (dcb->fd == -1) if (dcb->fd == DCBFD_CLOSED)
{
dcb_final_free(dcb); dcb_final_free(dcb);
}
else else
{ {
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
@ -308,7 +310,7 @@ DCB *clone;
return NULL; return NULL;
} }
clone->fd = -1; clone->fd = DCBFD_CLONED;;
clone->flags |= DCBF_CLONE; clone->flags |= DCBF_CLONE;
clone->state = orig->state; clone->state = orig->state;
clone->data = orig->data; clone->data = orig->data;
@ -551,12 +553,15 @@ bool succp = false;
DCB* dcb_next = NULL; DCB* dcb_next = NULL;
int rc = 0; int rc = 0;
if (dcb->fd > 0)
{
/*< /*<
* Close file descriptor and move to clean-up phase. * Close file descriptor and move to clean-up phase.
*/ */
rc = close(dcb->fd); rc = close(dcb->fd);
if (rc < 0) { if (rc < 0)
{
int eno = errno; int eno = errno;
errno = 0; errno = 0;
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
@ -568,8 +573,10 @@ bool succp = false;
eno, eno,
strerror(eno)))); strerror(eno))));
} }
#if defined(SS_DEBUG) else
else { {
dcb->fd = DCBFD_CLOSED;
LOGIF(LD, (skygw_log_write_flush( LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [dcb_process_zombies] Closed socket " "%lu [dcb_process_zombies] Closed socket "
@ -577,14 +584,11 @@ bool succp = false;
pthread_self(), pthread_self(),
dcb->fd, dcb->fd,
dcb))); dcb)));
#endif /* SS_DEBUG */
#if defined(FAKE_CODE) #if defined(FAKE_CODE)
conn_open[dcb->fd] = false; conn_open[dcb->fd] = false;
#endif /* FAKE_CODE */ #endif /* FAKE_CODE */
#if defined(SS_DEBUG)
ss_debug(dcb->fd = -1;)
} }
#endif /* SS_DEBUG */ }
LOGIF_MAYBE(LT, (dcb_get_ses_log_info( LOGIF_MAYBE(LT, (dcb_get_ses_log_info(
dcb, dcb,
&tls_log_info.li_sesid, &tls_log_info.li_sesid,
@ -657,7 +661,7 @@ int rc;
} }
fd = dcb->func.connect(dcb, server, session); fd = dcb->func.connect(dcb, server, session);
if (fd == -1) { if (fd == DCBFD_CLOSED) {
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
"%lu [dcb_connect] Failed to connect to server %s:%d, " "%lu [dcb_connect] Failed to connect to server %s:%d, "
@ -683,7 +687,7 @@ int rc;
session->client, session->client,
session->client->fd))); session->client->fd)));
} }
ss_dassert(dcb->fd == -1); /*< must be uninitialized at this point */ ss_dassert(dcb->fd == DCBFD_CLOSED); /*< must be uninitialized at this point */
/*< /*<
* Successfully connected to backend. Assign file descriptor to dcb * Successfully connected to backend. Assign file descriptor to dcb
*/ */
@ -704,7 +708,7 @@ int rc;
*/ */
rc = poll_add_dcb(dcb); rc = poll_add_dcb(dcb);
if (rc == -1) { if (rc == DCBFD_CLOSED) {
dcb_set_state(dcb, DCB_STATE_DISCONNECTED, NULL); dcb_set_state(dcb, DCB_STATE_DISCONNECTED, NULL);
dcb_final_free(dcb); dcb_final_free(dcb);
return NULL; return NULL;
@ -736,10 +740,21 @@ int dcb_read(
GWBUF *buffer = NULL; GWBUF *buffer = NULL;
int b; int b;
int rc; int rc;
int n ; int n;
int nread = 0; int nread = 0;
CHK_DCB(dcb); CHK_DCB(dcb);
if (dcb->fd <= 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Read failed, dcb is %s.",
dcb->fd == DCBFD_CLOSED ? "closed" : "cloned, not readable")));
n = 0;
goto return_n;
}
while (true) while (true)
{ {
int bufsize; int bufsize;
@ -864,6 +879,14 @@ int below_water;
below_water = (dcb->high_water && dcb->writeqlen < dcb->high_water) ? 1 : 0; below_water = (dcb->high_water && dcb->writeqlen < dcb->high_water) ? 1 : 0;
ss_dassert(queue != NULL); ss_dassert(queue != NULL);
if (dcb->fd <= 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Write failed, dcb is %s.",
dcb->fd == DCBFD_CLOSED ? "closed" : "cloned, not writable")));
return 0;
}
/** /**
* SESSION_STATE_STOPPING means that one of the backends is closing * SESSION_STATE_STOPPING means that one of the backends is closing
* the router session. Some backends may have not completed * the router session. Some backends may have not completed
@ -1208,8 +1231,6 @@ dcb_close(DCB *dcb)
* Stop dcb's listening and modify state accordingly. * Stop dcb's listening and modify state accordingly.
*/ */
if (dcb->state == DCB_STATE_POLLING) if (dcb->state == DCB_STATE_POLLING)
{
if (dcb->fd != -1)
{ {
rc = poll_remove_dcb(dcb); rc = poll_remove_dcb(dcb);
@ -1247,8 +1268,6 @@ dcb_close(DCB *dcb)
dcb_add_to_zombieslist(dcb); dcb_add_to_zombieslist(dcb);
} }
} }
}
ss_dassert(dcb->state == DCB_STATE_NOPOLLING || ss_dassert(dcb->state == DCB_STATE_NOPOLLING ||
dcb->state == DCB_STATE_ZOMBIE); dcb->state == DCB_STATE_ZOMBIE);
} }
@ -1757,14 +1776,16 @@ static bool dcb_set_state_nomutex(
* @param dcb The DCB to write buffer * @param dcb The DCB to write buffer
* @param buf Buffer to write * @param buf Buffer to write
* @param nbytes Number of bytes to write * @param nbytes Number of bytes to write
* @return Number of written bytes
*/ */
int int
gw_write(DCB *dcb, const void *buf, size_t nbytes) gw_write(DCB *dcb, const void *buf, size_t nbytes)
{ {
int w; int w = 0;
int fd = dcb->fd; int fd = dcb->fd;
#if defined(FAKE_CODE) #if defined(FAKE_CODE)
if (dcb_fake_write_errno[fd] != 0) { if (fd > 0 && dcb_fake_write_errno[fd] != 0)
{
ss_dassert(dcb_fake_write_ev[fd] != 0); ss_dassert(dcb_fake_write_ev[fd] != 0);
w = write(fd, buf, nbytes/2); /*< leave peer to read missing bytes */ w = write(fd, buf, nbytes/2); /*< leave peer to read missing bytes */
@ -1772,11 +1793,15 @@ gw_write(DCB *dcb, const void *buf, size_t nbytes)
w = -1; w = -1;
errno = dcb_fake_write_errno[fd]; errno = dcb_fake_write_errno[fd];
} }
} else { } else if (fd > 0)
{
w = write(fd, buf, nbytes); w = write(fd, buf, nbytes);
} }
#else #else
if (fd > 0)
{
w = write(fd, buf, nbytes); w = write(fd, buf, nbytes);
}
#endif /* FAKE_CODE */ #endif /* FAKE_CODE */
#if defined(SS_DEBUG_MYSQL) #if defined(SS_DEBUG_MYSQL)

2
server/core/gateway.c
View File

@ -572,7 +572,7 @@ return_succp:
static bool resolve_maxscale_homedir( static bool resolve_maxscale_homedir(
char** p_home_dir) char** p_home_dir)
{ {
bool succp; bool succp = false;
char* tmp; char* tmp;
char* tmp2; char* tmp2;
char* log_context = NULL; char* log_context = NULL;

75
server/core/modutil.c
View File

@ -234,7 +234,7 @@ modutil_get_query(GWBUF *buf)
uint8_t* packet; uint8_t* packet;
mysql_server_cmd_t packet_type; mysql_server_cmd_t packet_type;
size_t len; size_t len;
char* query_str; char* query_str = NULL;
packet = GWBUF_DATA(buf); packet = GWBUF_DATA(buf);
packet_type = packet[4]; packet_type = packet[4];
@ -252,7 +252,7 @@ modutil_get_query(GWBUF *buf)
case MYSQL_COM_QUERY: case MYSQL_COM_QUERY:
len = MYSQL_GET_PACKET_LEN(packet)-1; /*< distract 1 for packet type byte */ len = MYSQL_GET_PACKET_LEN(packet)-1; /*< distract 1 for packet type byte */
if ((query_str = (char *)malloc(len+1)) == NULL) if (len < 1 || (query_str = (char *)malloc(len+1)) == NULL)
{ {
goto retblock; goto retblock;
} }
@ -262,7 +262,7 @@ modutil_get_query(GWBUF *buf)
default: default:
len = strlen(STRPACKETTYPE(packet_type))+1; len = strlen(STRPACKETTYPE(packet_type))+1;
if ((query_str = (char *)malloc(len+1)) == NULL) if (len < 1 || (query_str = (char *)malloc(len+1)) == NULL)
{ {
goto retblock; goto retblock;
} }
@ -390,3 +390,72 @@ int modutil_send_mysql_err_packet (
return dcb->func.write(dcb, buf); return dcb->func.write(dcb, buf);
} }
/**
* Buffer contains at least one of the following:
* complete [complete] [partial] mysql packet
*
* return pointer to gwbuf containing a complete packet or
* NULL if no complete packet was found.
*/
GWBUF* modutil_get_next_MySQL_packet(
GWBUF** p_readbuf)
{
GWBUF* packetbuf;
GWBUF* readbuf;
size_t buflen;
size_t packetlen;
size_t totalbuflen;
uint8_t* data;
size_t nbytes_copied = 0;
uint8_t* target;
readbuf = *p_readbuf;
if (readbuf == NULL)
{
packetbuf = NULL;
goto return_packetbuf;
}
CHK_GWBUF(readbuf);
if (GWBUF_EMPTY(readbuf))
{
packetbuf = NULL;
goto return_packetbuf;
}
totalbuflen = gwbuf_length(readbuf);
data = (uint8_t *)GWBUF_DATA((readbuf));
packetlen = MYSQL_GET_PACKET_LEN(data)+4;
/** packet is incomplete */
if (packetlen > totalbuflen)
{
packetbuf = NULL;
goto return_packetbuf;
}
packetbuf = gwbuf_alloc(packetlen);
target = GWBUF_DATA(packetbuf);
packetbuf->gwbuf_type = readbuf->gwbuf_type; /*< Copy the type too */
/**
* Copy first MySQL packet to packetbuf and leave posible other
* packets to read buffer.
*/
while (nbytes_copied < packetlen && totalbuflen > 0)
{
uint8_t* src = GWBUF_DATA((*p_readbuf));
size_t bytestocopy;
buflen = GWBUF_LENGTH((*p_readbuf));
bytestocopy = MIN(buflen,packetlen-nbytes_copied);
memcpy(target+nbytes_copied, src, bytestocopy);
*p_readbuf = gwbuf_consume((*p_readbuf), bytestocopy);
totalbuflen = gwbuf_length((*p_readbuf));
nbytes_copied += bytestocopy;
}
ss_dassert(buflen == 0 || nbytes_copied == packetlen);
return_packetbuf:
return packetbuf;
}

9
server/core/poll.c
View File

@ -341,7 +341,13 @@ poll_remove_dcb(DCB *dcb)
/*< /*<
* Set state to NOPOLLING and remove dcb from poll set. * Set state to NOPOLLING and remove dcb from poll set.
*/ */
if (dcb_set_state(dcb, new_state, &old_state)) { if (dcb_set_state(dcb, new_state, &old_state))
{
/**
* Only positive fds can be removed from epoll set.
*/
if (dcb->fd > 0)
{
rc = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, dcb->fd, &ev); rc = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, dcb->fd, &ev);
if (rc != 0) { if (rc != 0) {
@ -355,6 +361,7 @@ poll_remove_dcb(DCB *dcb)
} }
ss_dassert(rc == 0); /*< trap in debug */ ss_dassert(rc == 0); /*< trap in debug */
} }
}
/*< /*<
* This call was redundant, but the end result is correct. * This call was redundant, but the end result is correct.
*/ */

2
server/core/secrets.c
View File

@ -252,6 +252,7 @@ MAXKEYS key;
"Error : failed opening /dev/random. Error %d, %s.", "Error : failed opening /dev/random. Error %d, %s.",
errno, errno,
strerror(errno)))); strerror(errno))));
close(fd);
return 1; return 1;
} }
@ -260,6 +261,7 @@ MAXKEYS key;
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"Error : failed to read /dev/random."))); "Error : failed to read /dev/random.")));
close(fd);
close(randfd); close(randfd);
return 1; return 1;
} }

6
server/core/service.c
View File

@ -765,7 +765,7 @@ int n = 0;
if ((flist = (FILTER_DEF **)malloc(sizeof(FILTER_DEF *))) == NULL) if ((flist = (FILTER_DEF **)malloc(sizeof(FILTER_DEF *))) == NULL)
{ {
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR, LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
"Out of memory adding filters to service.\n"))); "Error : Out of memory adding filters to service.\n")));
return; return;
} }
ptr = strtok_r(filters, "|", &brkt); ptr = strtok_r(filters, "|", &brkt);
@ -776,14 +776,14 @@ int n = 0;
(n + 1) * sizeof(FILTER_DEF *))) == NULL) (n + 1) * sizeof(FILTER_DEF *))) == NULL)
{ {
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR, LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
"Out of memory adding filters to service.\n"))); "Error : Out of memory adding filters to service.\n")));
return; return;
} }
if ((flist[n-1] = filter_find(trim(ptr))) == NULL) if ((flist[n-1] = filter_find(trim(ptr))) == NULL)
{ {
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"Unable to find filter '%s' for service '%s'\n", "Warning : Unable to find filter '%s' for service '%s'\n",
trim(ptr), service->name trim(ptr), service->name
))); )));
n--; n--;

3
server/include/dcb.h
View File

@ -129,6 +129,9 @@ typedef struct {
*/ */
#define GWPROTOCOL_VERSION {1, 0, 0} #define GWPROTOCOL_VERSION {1, 0, 0}
#define DCBFD_CLOSED -1
#define DCBFD_CLONED -2
/** /**
* The statitics gathered on a descriptor control block * The statitics gathered on a descriptor control block
*/ */

1
server/include/modutil.h
View File

@ -41,6 +41,7 @@ extern char *modutil_get_SQL(GWBUF *);
extern GWBUF *modutil_replace_SQL(GWBUF *, char *); extern GWBUF *modutil_replace_SQL(GWBUF *, char *);
extern char *modutil_get_query(GWBUF* buf); extern char *modutil_get_query(GWBUF* buf);
extern int modutil_send_mysql_err_packet(DCB *, int, int, int, const char *, const char *); extern int modutil_send_mysql_err_packet(DCB *, int, int, int, const char *, const char *);
GWBUF* modutil_get_next_MySQL_packet(GWBUF** p_readbuf);
GWBUF *modutil_create_mysql_err_msg( GWBUF *modutil_create_mysql_err_msg(
int packet_number, int packet_number,

132
server/modules/filter/tee.c
View File

@ -41,6 +41,7 @@
* Date Who Description * Date Who Description
* 20/06/2014 Mark Riddoch Initial implementation * 20/06/2014 Mark Riddoch Initial implementation
* 24/06/2014 Mark Riddoch Addition of support for multi-packet queries * 24/06/2014 Mark Riddoch Addition of support for multi-packet queries
* 12/12/2014 Mark Riddoch Add support for otehr packet types
* *
* @endverbatim * @endverbatim
*/ */
@ -58,6 +59,29 @@
#include <router.h> #include <router.h>
#include <dcb.h> #include <dcb.h>
#define MYSQL_COM_QUIT 0x01
#define MYSQL_COM_INITDB 0x02
#define MYSQL_COM_FIELD_LIST 0x04
#define MYSQL_COM_CHANGE_USER 0x11
#define MYSQL_COM_STMT_PREPARE 0x16
#define MYSQL_COM_STMT_EXECUTE 0x17
#define MYSQL_COM_STMT_SEND_LONG_DATA 0x18
#define MYSQL_COM_STMT_CLOSE 0x19
#define MYSQL_COM_STMT_RESET 0x1a
static unsigned char required_packets[] = {
MYSQL_COM_QUIT,
MYSQL_COM_INITDB,
MYSQL_COM_FIELD_LIST,
MYSQL_COM_CHANGE_USER,
MYSQL_COM_STMT_PREPARE,
MYSQL_COM_STMT_EXECUTE,
MYSQL_COM_STMT_SEND_LONG_DATA,
MYSQL_COM_STMT_CLOSE,
MYSQL_COM_STMT_RESET,
0 };
/** Defined in log_manager.cc */ /** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask; extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[]; extern size_t log_ses_count[];
@ -128,6 +152,7 @@ typedef struct {
int residual; /* Any outstanding SQL text */ int residual; /* Any outstanding SQL text */
} TEE_SESSION; } TEE_SESSION;
static int packet_is_required(GWBUF *queue);
/** /**
* Implementation of the mandatory version entry point * Implementation of the mandatory version entry point
* *
@ -280,27 +305,79 @@ TEE_INSTANCE *my_instance = (TEE_INSTANCE *)instance;
TEE_SESSION *my_session; TEE_SESSION *my_session;
char *remote, *userName; char *remote, *userName;
if (strcmp(my_instance->service->name, session->service->name) == 0)
{
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
"Error : %s: Recursive use of tee filter in service.",
session->service->name)));
my_session = NULL;
goto retblock;
}
if ((my_session = calloc(1, sizeof(TEE_SESSION))) != NULL) if ((my_session = calloc(1, sizeof(TEE_SESSION))) != NULL)
{ {
my_session->active = 1; my_session->active = 1;
my_session->residual = 0; my_session->residual = 0;
if (my_instance->source
&& (remote = session_get_remote(session)) != NULL) if (my_instance->source &&
(remote = session_get_remote(session)) != NULL)
{ {
if (strcmp(remote, my_instance->source)) if (strcmp(remote, my_instance->source))
{
my_session->active = 0; my_session->active = 0;
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Warning : Tee filter is not active.")));
}
} }
userName = session_getUser(session); userName = session_getUser(session);
if (my_instance->userName && userName && strcmp(userName,
my_instance->userName)) if (my_instance->userName &&
my_session->active = 0; userName &&
if (my_session->active) strcmp(userName, my_instance->userName))
{ {
my_session->branch_dcb = dcb_clone(session->client); my_session->active = 0;
my_session->branch_session = session_alloc(my_instance->service, my_session->branch_dcb);
} LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Warning : Tee filter is not active.")));
} }
if (my_session->active)
{
DCB* dcb;
SESSION* ses;
if ((dcb = dcb_clone(session->client)) == NULL)
{
freeSession(my_instance, (void *)my_session);
my_session = NULL;
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Error : Creating client DCB for Tee "
"filter failed. Terminating session.")));
goto retblock;
}
if ((ses = session_alloc(my_instance->service, dcb)) == NULL)
{
dcb_close(dcb);
freeSession(my_instance, (void *)my_session);
my_session = NULL;
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Error : Creating client session for Tee "
"filter failed. Terminating session.")));
goto retblock;
}
my_session->branch_session = ses;
my_session->branch_dcb = dcb;
}
}
retblock:
return my_session; return my_session;
} }
@ -325,17 +402,26 @@ SESSION *bsession;
{ {
if ((bsession = my_session->branch_session) != NULL) if ((bsession = my_session->branch_session) != NULL)
{ {
CHK_SESSION(bsession);
spinlock_acquire(&bsession->ses_lock);
if (bsession->state != SESSION_STATE_STOPPING)
{
bsession->state = SESSION_STATE_STOPPING;
}
router = bsession->service->router; router = bsession->service->router;
router_instance = bsession->service->router_instance; router_instance = bsession->service->router_instance;
rsession = bsession->router_session; rsession = bsession->router_session;
spinlock_release(&bsession->ses_lock);
/** Close router session and all its connections */ /** Close router session and all its connections */
router->closeSession(router_instance, rsession); router->closeSession(router_instance, rsession);
} }
dcb_free(my_session->branch_dcb);
/* No need to free the session, this is done as /* No need to free the session, this is done as
* a side effect of closing the client DCB of the * a side effect of closing the client DCB of the
* session. * session.
*/ */
my_session->active = 0;
} }
} }
@ -422,6 +508,10 @@ GWBUF *clone = NULL;
} }
free(ptr); free(ptr);
} }
else if (packet_is_required(queue))
{
clone = gwbuf_clone(queue);
}
/* Pass the query downstream */ /* Pass the query downstream */
rval = my_session->down.routeQuery(my_session->down.instance, rval = my_session->down.routeQuery(my_session->down.instance,
@ -477,3 +567,25 @@ TEE_SESSION *my_session = (TEE_SESSION *)fsession;
my_session->n_rejected); my_session->n_rejected);
} }
} }
/**
* Determine if the packet is a command that must be sent to the branch
* to maintain the session consistancy. These are COM_INIT_DB,
* COM_CHANGE_USER and COM_QUIT packets.
*
* @param queue The buffer to check
* @return non-zero if the packet should be sent to the branch
*/
static int
packet_is_required(GWBUF *queue)
{
uint8_t *ptr;
int i;
ptr = GWBUF_DATA(queue);
if (GWBUF_LENGTH(queue) > 4)
for (i = 0; required_packets[i]; i++)
if (ptr[4] == required_packets[i])
return 1;
return 0;
}

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

@ -323,4 +323,8 @@ typedef struct router_instance {
#define BACKEND_TYPE(b) (SERVER_IS_MASTER((b)->backend_server) ? BE_MASTER : \ #define BACKEND_TYPE(b) (SERVER_IS_MASTER((b)->backend_server) ? BE_MASTER : \
(SERVER_IS_SLAVE((b)->backend_server) ? BE_SLAVE : BE_UNDEFINED)); (SERVER_IS_SLAVE((b)->backend_server) ? BE_SLAVE : BE_UNDEFINED));
#define RSES_SESSION(r) (r->rses_backend_ref->bref_dcb->session)
#define RSES_CLIENT_DCB(r) (RSES_SESSION(r)->client)
#endif /*< _RWSPLITROUTER_H */ #endif /*< _RWSPLITROUTER_H */

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

@ -111,6 +111,21 @@ static backend_ref_t* check_candidate_bref(
backend_ref_t* new_bref, backend_ref_t* new_bref,
select_criteria_t sc); select_criteria_t sc);
static skygw_query_type_t is_read_tmp_table(
ROUTER_CLIENT_SES* router_cli_ses,
GWBUF* querybuf,
skygw_query_type_t type);
static void check_create_tmp_table(
ROUTER_CLIENT_SES* router_cli_ses,
GWBUF* querybuf,
skygw_query_type_t type);
static bool route_single_stmt(
ROUTER_INSTANCE* inst,
ROUTER_CLIENT_SES* rses,
GWBUF* querybuf);
static uint8_t getCapabilities (ROUTER* inst, void* router_session); static uint8_t getCapabilities (ROUTER* inst, void* router_session);
@ -1505,14 +1520,12 @@ static route_target_t get_route_target (
/** /**
* Check if the query is a DROP TABLE... query and * Check if the query is a DROP TABLE... query and
* if it targets a temporary table, remove it from the hashtable. * if it targets a temporary table, remove it from the hashtable.
* @param instance Router instance * @param router_cli_ses Router client session
* @param router_session Router client session
* @param querybuf GWBUF containing the query * @param querybuf GWBUF containing the query
* @param type The type of the query resolved so far * @param type The type of the query resolved so far
*/ */
void check_drop_tmp_table( void check_drop_tmp_table(
ROUTER* instance, ROUTER_CLIENT_SES* router_cli_ses,
void* router_session,
GWBUF* querybuf, GWBUF* querybuf,
skygw_query_type_t type) skygw_query_type_t type)
{ {
@ -1522,7 +1535,6 @@ void check_drop_tmp_table(
char *hkey,*dbname; char *hkey,*dbname;
MYSQL_session* data; MYSQL_session* data;
ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
DCB* master_dcb = NULL; DCB* master_dcb = NULL;
rses_property_t* rses_prop_tmp; rses_property_t* rses_prop_tmp;
@ -1567,15 +1579,13 @@ void check_drop_tmp_table(
/** /**
* Check if the query targets a temporary table. * Check if the query targets a temporary table.
* @param instance Router instance * @param router_cli_ses Router client session
* @param router_session Router client session
* @param querybuf GWBUF containing the query * @param querybuf GWBUF containing the query
* @param type The type of the query resolved so far * @param type The type of the query resolved so far
* @return The type of the query * @return The type of the query
*/ */
skygw_query_type_t is_read_tmp_table( static skygw_query_type_t is_read_tmp_table(
ROUTER* instance, ROUTER_CLIENT_SES* router_cli_ses,
void* router_session,
GWBUF* querybuf, GWBUF* querybuf,
skygw_query_type_t type) skygw_query_type_t type)
{ {
@ -1586,7 +1596,6 @@ skygw_query_type_t is_read_tmp_table(
char *hkey,*dbname; char *hkey,*dbname;
MYSQL_session* data; MYSQL_session* data;
ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
DCB* master_dcb = NULL; DCB* master_dcb = NULL;
skygw_query_type_t qtype = type; skygw_query_type_t qtype = type;
rses_property_t* rses_prop_tmp; rses_property_t* rses_prop_tmp;
@ -1656,14 +1665,12 @@ skygw_query_type_t is_read_tmp_table(
* the database and table name, create a hashvalue and * the database and table name, create a hashvalue and
* add it to the router client session's property. If property * add it to the router client session's property. If property
* doesn't exist then create it first. * doesn't exist then create it first.
* @param instance Router instance * @param router_cli_ses Router client session
* @param router_session Router client session
* @param querybuf GWBUF containing the query * @param querybuf GWBUF containing the query
* @param type The type of the query resolved so far * @param type The type of the query resolved so far
*/ */
void check_create_tmp_table( static void check_create_tmp_table(
ROUTER* instance, ROUTER_CLIENT_SES* router_cli_ses,
void* router_session,
GWBUF* querybuf, GWBUF* querybuf,
skygw_query_type_t type) skygw_query_type_t type)
{ {
@ -1673,7 +1680,6 @@ void check_create_tmp_table(
char *hkey,*dbname; char *hkey,*dbname;
MYSQL_session* data; MYSQL_session* data;
ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
DCB* master_dcb = NULL; DCB* master_dcb = NULL;
rses_property_t* rses_prop_tmp; rses_property_t* rses_prop_tmp;
HASHTABLE* h; HASHTABLE* h;
@ -1800,6 +1806,134 @@ static int routeQuery(
ROUTER* instance, ROUTER* instance,
void* router_session, void* router_session,
GWBUF* querybuf) GWBUF* querybuf)
{
int ret = 0;
ROUTER_INSTANCE* inst = (ROUTER_INSTANCE *)instance;
ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
bool succp = false;
CHK_CLIENT_RSES(router_cli_ses);
/**
* Untyped GWBUF means that it can consist of incomplete and/or multiple
* MySQL packets.
* Read and route found MySQL packets one by one and store potential
* incomplete packet to DCB's dcb_readqueue.
*/
if (GWBUF_IS_TYPE_UNDEFINED(querybuf))
{
GWBUF* tmpbuf = querybuf;
do
{
/**
* Try to read complete MySQL packet from tmpbuf.
* Append leftover to client's read queue.
*/
if ((querybuf = modutil_get_next_MySQL_packet(&tmpbuf)) == NULL)
{
if (GWBUF_LENGTH(tmpbuf) > 0)
{
DCB* dcb = RSES_CLIENT_DCB(router_cli_ses);
dcb->dcb_readqueue = gwbuf_append(dcb->dcb_readqueue, tmpbuf);
}
succp = true;
goto retblock;
}
/** Mark buffer to as MySQL type */
gwbuf_set_type(querybuf, GWBUF_TYPE_MYSQL);
gwbuf_set_type(querybuf, GWBUF_TYPE_SINGLE_STMT);
/**
* If router is closed, discard the packet
*/
if (router_cli_ses->rses_closed)
{
uint8_t* packet;
mysql_server_cmd_t packet_type;
packet = GWBUF_DATA(querybuf);
packet_type = packet[4];
if (packet_type != MYSQL_COM_QUIT)
{
char* query_str = modutil_get_query(querybuf);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Can't route %s:\"%s\" to "
"backend server. Router is closed.",
STRPACKETTYPE(packet_type),
(query_str == NULL ? "(empty)" : query_str))));
free(query_str);
}
}
else
{
succp = route_single_stmt(inst, router_cli_ses, querybuf);
}
}
while (tmpbuf != NULL);
}
/**
* If router is closed, discard the packet
*/
else if (router_cli_ses->rses_closed)
{
uint8_t* packet;
mysql_server_cmd_t packet_type;
packet = GWBUF_DATA(querybuf);
packet_type = packet[4];
if (packet_type != MYSQL_COM_QUIT)
{
char* query_str = modutil_get_query(querybuf);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Can't route %s:\"%s\" to "
"backend server. Router is closed.",
STRPACKETTYPE(packet_type),
(query_str == NULL ? "(empty)" : query_str))));
free(query_str);
}
}
else
{
succp = route_single_stmt(inst, router_cli_ses, querybuf);
}
retblock:
#if defined(SS_DEBUG2)
if (querybuf != NULL)
{
char* canonical_query_str;
canonical_query_str = skygw_get_canonical(querybuf);
if (canonical_query_str != NULL)
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Canonical version: %s",
canonical_query_str)));
free(canonical_query_str);
}
}
#endif
if (querybuf != NULL) gwbuf_free(querybuf);
if (succp) ret = 1;
return ret;
}
static bool route_single_stmt(
ROUTER_INSTANCE* inst,
ROUTER_CLIENT_SES* rses,
GWBUF* querybuf)
{ {
skygw_query_type_t qtype = QUERY_TYPE_UNKNOWN; skygw_query_type_t qtype = QUERY_TYPE_UNKNOWN;
mysql_server_cmd_t packet_type; mysql_server_cmd_t packet_type;
@ -1807,55 +1941,21 @@ static int routeQuery(
int ret = 0; int ret = 0;
DCB* master_dcb = NULL; DCB* master_dcb = NULL;
DCB* target_dcb = NULL; DCB* target_dcb = NULL;
ROUTER_INSTANCE* inst = (ROUTER_INSTANCE *)instance;
ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
bool rses_is_closed = false;
route_target_t route_target; route_target_t route_target;
bool succp = false; bool succp = false;
int rlag_max = MAX_RLAG_UNDEFINED; int rlag_max = MAX_RLAG_UNDEFINED;
backend_type_t btype; /*< target backend type */ backend_type_t btype; /*< target backend type */
CHK_CLIENT_RSES(router_cli_ses);
/** Dirty read for quick check if router is closed. */
if (router_cli_ses->rses_closed)
{
rses_is_closed = true;
}
ss_dassert(!GWBUF_IS_TYPE_UNDEFINED(querybuf)); ss_dassert(!GWBUF_IS_TYPE_UNDEFINED(querybuf));
packet = GWBUF_DATA(querybuf); packet = GWBUF_DATA(querybuf);
packet_type = packet[4]; packet_type = packet[4];
if (rses_is_closed)
{
/**
* MYSQL_COM_QUIT may have sent by client and as a part of backend
* closing procedure.
*/
if (packet_type != MYSQL_COM_QUIT)
{
char* query_str = modutil_get_query(querybuf);
LOGIF(LE,
(skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Can't route %s:%s:\"%s\" to "
"backend server. Router is closed.",
STRPACKETTYPE(packet_type),
STRQTYPE(qtype),
(query_str == NULL ? "(empty)" : query_str))));
free(query_str);
}
ret = 0;
goto retblock;
}
/** /**
* Read stored master DCB pointer. If master is not set, routing must * Read stored master DCB pointer. If master is not set, routing must
* be aborted * be aborted
*/ */
if ((master_dcb = router_cli_ses->rses_master_ref->bref_dcb) == NULL) if ((master_dcb = rses->rses_master_ref->bref_dcb) == NULL)
{ {
char* query_str = modutil_get_query(querybuf); char* query_str = modutil_get_query(querybuf);
CHK_DCB(master_dcb); CHK_DCB(master_dcb);
@ -1868,7 +1968,7 @@ static int routeQuery(
STRQTYPE(qtype), STRQTYPE(qtype),
(query_str == NULL ? "(empty)" : query_str)))); (query_str == NULL ? "(empty)" : query_str))));
free(query_str); free(query_str);
ret = 0; succp = false;
goto retblock; goto retblock;
} }
@ -1925,45 +2025,45 @@ static int routeQuery(
/** /**
* Check if the query has anything to do with temporary tables. * Check if the query has anything to do with temporary tables.
*/ */
qtype = is_read_tmp_table(instance,router_session,querybuf,qtype); qtype = is_read_tmp_table(rses, querybuf, qtype);
check_create_tmp_table(instance,router_session,querybuf,qtype); check_create_tmp_table(rses, querybuf, qtype);
check_drop_tmp_table(instance,router_session,querybuf,qtype); check_drop_tmp_table(rses, querybuf,qtype);
/** /**
* If autocommit is disabled or transaction is explicitly started * If autocommit is disabled or transaction is explicitly started
* transaction becomes active and master gets all statements until * transaction becomes active and master gets all statements until
* transaction is committed and autocommit is enabled again. * transaction is committed and autocommit is enabled again.
*/ */
if (router_cli_ses->rses_autocommit_enabled && if (rses->rses_autocommit_enabled &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_DISABLE_AUTOCOMMIT)) QUERY_IS_TYPE(qtype, QUERY_TYPE_DISABLE_AUTOCOMMIT))
{ {
router_cli_ses->rses_autocommit_enabled = false; rses->rses_autocommit_enabled = false;
if (!router_cli_ses->rses_transaction_active) if (!rses->rses_transaction_active)
{ {
router_cli_ses->rses_transaction_active = true; rses->rses_transaction_active = true;
} }
} }
else if (!router_cli_ses->rses_transaction_active && else if (!rses->rses_transaction_active &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_BEGIN_TRX)) QUERY_IS_TYPE(qtype, QUERY_TYPE_BEGIN_TRX))
{ {
router_cli_ses->rses_transaction_active = true; rses->rses_transaction_active = true;
} }
/** /**
* Explicit COMMIT and ROLLBACK, implicit COMMIT. * Explicit COMMIT and ROLLBACK, implicit COMMIT.
*/ */
if (router_cli_ses->rses_autocommit_enabled && if (rses->rses_autocommit_enabled &&
router_cli_ses->rses_transaction_active && rses->rses_transaction_active &&
(QUERY_IS_TYPE(qtype,QUERY_TYPE_COMMIT) || (QUERY_IS_TYPE(qtype,QUERY_TYPE_COMMIT) ||
QUERY_IS_TYPE(qtype,QUERY_TYPE_ROLLBACK))) QUERY_IS_TYPE(qtype,QUERY_TYPE_ROLLBACK)))
{ {
router_cli_ses->rses_transaction_active = false; rses->rses_transaction_active = false;
} }
else if (!router_cli_ses->rses_autocommit_enabled && else if (!rses->rses_autocommit_enabled &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_ENABLE_AUTOCOMMIT)) QUERY_IS_TYPE(qtype, QUERY_TYPE_ENABLE_AUTOCOMMIT))
{ {
router_cli_ses->rses_autocommit_enabled = true; rses->rses_autocommit_enabled = true;
router_cli_ses->rses_transaction_active = false; rses->rses_transaction_active = false;
} }
if (LOG_IS_ENABLED(LOGFILE_TRACE)) if (LOG_IS_ENABLED(LOGFILE_TRACE))
@ -1980,8 +2080,8 @@ static int routeQuery(
LOGFILE_TRACE, LOGFILE_TRACE,
"> Autocommit: %s, trx is %s, cmd: %s, type: %s, " "> Autocommit: %s, trx is %s, cmd: %s, type: %s, "
"stmt: %s%s %s", "stmt: %s%s %s",
(router_cli_ses->rses_autocommit_enabled ? "[enabled]" : "[disabled]"), (rses->rses_autocommit_enabled ? "[enabled]" : "[disabled]"),
(router_cli_ses->rses_transaction_active ? "[open]" : "[not open]"), (rses->rses_transaction_active ? "[open]" : "[not open]"),
STRPACKETTYPE(ptype), STRPACKETTYPE(ptype),
(qtypestr==NULL ? "N/A" : qtypestr), (qtypestr==NULL ? "N/A" : qtypestr),
contentstr, contentstr,
@ -2009,8 +2109,8 @@ static int routeQuery(
* eventually to master * eventually to master
*/ */
route_target = get_route_target(qtype, route_target = get_route_target(qtype,
router_cli_ses->rses_transaction_active, rses->rses_transaction_active,
router_cli_ses->rses_config.rw_use_sql_variables_in, rses->rses_config.rw_use_sql_variables_in,
querybuf->hint); querybuf->hint);
if (TARGET_IS_ALL(route_target)) if (TARGET_IS_ALL(route_target))
@ -2020,7 +2120,8 @@ static int routeQuery(
* response. Statement is examined in route_session_write. * response. Statement is examined in route_session_write.
* Router locking is done inside the function. * Router locking is done inside the function.
*/ */
succp = route_session_write(router_cli_ses, succp = route_session_write(
rses,
gwbuf_clone(querybuf), gwbuf_clone(querybuf),
inst, inst,
packet_type, packet_type,
@ -2029,18 +2130,27 @@ static int routeQuery(
if (succp) if (succp)
{ {
atomic_add(&inst->stats.n_all, 1); atomic_add(&inst->stats.n_all, 1);
ret = 1;
} }
goto retblock; goto retblock;
} }
/** Lock router session */ /** Lock router session */
if (!rses_begin_locked_router_action(router_cli_ses)) if (!rses_begin_locked_router_action(rses))
{ {
LOGIF(LT, (skygw_log_write( if (packet_type != MYSQL_COM_QUIT)
LOGFILE_TRACE, {
"Route query aborted! Routing session is closed <"))); char* query_str = modutil_get_query(querybuf);
ret = 0;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Can't route %s:%s:\"%s\" to "
"backend server. Router is closed.",
STRPACKETTYPE(packet_type),
STRQTYPE(qtype),
(query_str == NULL ? "(empty)" : query_str))));
free(query_str);
}
succp = false;
goto retblock; goto retblock;
} }
/** /**
@ -2099,18 +2209,15 @@ static int routeQuery(
if (rlag_max == MAX_RLAG_UNDEFINED) /*< no rlag max hint, use config */ if (rlag_max == MAX_RLAG_UNDEFINED) /*< no rlag max hint, use config */
{ {
rlag_max = rses_get_max_replication_lag(router_cli_ses); rlag_max = rses_get_max_replication_lag(rses);
} }
btype = BE_UNDEFINED; /*< target may be master or slave */ btype = BE_UNDEFINED; /*< target may be master or slave */
/** /**
* Search backend server by name or replication lag. * Search backend server by name or replication lag.
* If it fails, then try to find valid slave or master. * If it fails, then try to find valid slave or master.
*/ */
succp = get_dcb(&target_dcb, succp = get_dcb(&target_dcb, rses, btype, named_server,rlag_max);
router_cli_ses,
btype,
named_server,
rlag_max);
if (!succp) if (!succp)
{ {
if (TARGET_IS_NAMED_SERVER(route_target)) if (TARGET_IS_NAMED_SERVER(route_target))
@ -2139,26 +2246,23 @@ static int routeQuery(
if (rlag_max == MAX_RLAG_UNDEFINED) /*< no rlag max hint, use config */ if (rlag_max == MAX_RLAG_UNDEFINED) /*< no rlag max hint, use config */
{ {
rlag_max = rses_get_max_replication_lag(router_cli_ses); rlag_max = rses_get_max_replication_lag(rses);
} }
/** /**
* Search suitable backend server, get DCB in target_dcb * Search suitable backend server, get DCB in target_dcb
*/ */
succp = get_dcb(&target_dcb, succp = get_dcb(&target_dcb, rses, BE_SLAVE, NULL,rlag_max);
router_cli_ses,
BE_SLAVE,
NULL,
rlag_max);
if (succp) if (succp)
{ {
#if defined(SS_EXTRA_DEBUG) #if defined(SS_EXTRA_DEBUG)
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE, LOGIF(LT, (skygw_log_write(LOGFILE_TRACE,
"Found DCB for slave."))); "Found DCB for slave.")));
#endif #endif
ss_dassert(get_bref_from_dcb(router_cli_ses, target_dcb) != ss_dassert(get_bref_from_dcb(rses, target_dcb) !=
router_cli_ses->rses_master_ref); rses->rses_master_ref);
ss_dassert(get_root_master_bref(router_cli_ses) == ss_dassert(get_root_master_bref(rses) ==
router_cli_ses->rses_master_ref); rses->rses_master_ref);
atomic_add(&inst->stats.n_slave, 1); atomic_add(&inst->stats.n_slave, 1);
} }
else else
@ -2174,7 +2278,7 @@ static int routeQuery(
DCB* curr_master_dcb = NULL; DCB* curr_master_dcb = NULL;
succp = get_dcb(&curr_master_dcb, succp = get_dcb(&curr_master_dcb,
router_cli_ses, rses,
BE_MASTER, BE_MASTER,
NULL, NULL,
MAX_RLAG_UNDEFINED); MAX_RLAG_UNDEFINED);
@ -2206,9 +2310,8 @@ static int routeQuery(
/** /**
* Master has changed. Return with error indicator. * Master has changed. Return with error indicator.
*/ */
rses_end_locked_router_action(router_cli_ses); rses_end_locked_router_action(rses);
succp = false; succp = false;
ret = 0;
goto retblock; goto retblock;
} }
} }
@ -2218,9 +2321,11 @@ static int routeQuery(
backend_ref_t* bref; backend_ref_t* bref;
sescmd_cursor_t* scur; sescmd_cursor_t* scur;
bref = get_bref_from_dcb(router_cli_ses, target_dcb); bref = get_bref_from_dcb(rses, target_dcb);
scur = &bref->bref_sescmd_cur; scur = &bref->bref_sescmd_cur;
ss_dassert(target_dcb != NULL);
LOGIF(LT, (skygw_log_write( LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE, LOGFILE_TRACE,
"Route query to %s \t%s:%d <", "Route query to %s \t%s:%d <",
@ -2230,16 +2335,21 @@ static int routeQuery(
bref->bref_backend->backend_server->port))); bref->bref_backend->backend_server->port)));
/** /**
* Store current stmt if execution of previous session command * Store current stmt if execution of previous session command
* haven't completed yet. Note that according to MySQL protocol * haven't completed yet.
*
* !!! Note that according to MySQL protocol
* there can only be one such non-sescmd stmt at the time. * there can only be one such non-sescmd stmt at the time.
*
* If the assertion below traps, pending queries are treated
* somehow wrong, or client is sending more queries before
* previous is received.
*/ */
if (sescmd_cursor_is_active(scur)) if (sescmd_cursor_is_active(scur))
{ {
ss_dassert(bref->bref_pending_cmd == NULL); ss_dassert(bref->bref_pending_cmd == NULL);
bref->bref_pending_cmd = gwbuf_clone(querybuf); bref->bref_pending_cmd = gwbuf_clone(querybuf);
rses_end_locked_router_action(router_cli_ses); rses_end_locked_router_action(rses);
ret = 1;
goto retblock; goto retblock;
} }
@ -2251,7 +2361,7 @@ static int routeQuery(
/** /**
* Add one query response waiter to backend reference * Add one query response waiter to backend reference
*/ */
bref = get_bref_from_dcb(router_cli_ses, target_dcb); bref = get_bref_from_dcb(rses, target_dcb);
bref_set_state(bref, BREF_QUERY_ACTIVE); bref_set_state(bref, BREF_QUERY_ACTIVE);
bref_set_state(bref, BREF_WAITING_RESULT); bref_set_state(bref, BREF_WAITING_RESULT);
} }
@ -2260,9 +2370,11 @@ static int routeQuery(
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"Error : Routing query failed."))); "Error : Routing query failed.")));
succp = false;
} }
} }
rses_end_locked_router_action(router_cli_ses); rses_end_locked_router_action(rses);
retblock: retblock:
#if defined(SS_DEBUG2) #if defined(SS_DEBUG2)
{ {
@ -2280,12 +2392,10 @@ retblock:
} }
} }
#endif #endif
gwbuf_free(querybuf); return succp;
return ret;
} }
/** /**
* @node Acquires lock to router client session if it is not closed. * @node Acquires lock to router client session if it is not closed.
* *
@ -2412,7 +2522,6 @@ char *weightby;
} }
} }
} }
/** /**
@ -3703,6 +3812,7 @@ static bool execute_sescmd_in_backend(
tmpbuf = scur->scmd_cur_cmd->my_sescmd_buf; tmpbuf = scur->scmd_cur_cmd->my_sescmd_buf;
qlen = MYSQL_GET_PACKET_LEN((unsigned char*)tmpbuf->start); qlen = MYSQL_GET_PACKET_LEN((unsigned char*)tmpbuf->start);
memset(data->db,0,MYSQL_DATABASE_MAXLEN+1); memset(data->db,0,MYSQL_DATABASE_MAXLEN+1);
if(qlen > 0)
strncpy(data->db,tmpbuf->start+5,qlen - 1); strncpy(data->db,tmpbuf->start+5,qlen - 1);
} }
/** Fallthrough */ /** Fallthrough */

97
server/test/MaxScale_test.cnf
View File

@ -1,27 +1,5 @@
#
# Example MaxScale.cnf configuration file
#
#
#
# Number of server threads
# Valid options are:
# threads=<number of threads>
[maxscale] [maxscale]
threads=1 threads=4
# Define a monitor that can be used to determine the state and role of
# the servers.
#
# Valid options are:
#
# module=<name of module to load>
# servers=<server name>,<server name>,...
# user =<user name - must have slave replication and
# slave client privileges>
# passwd=<password of the above user, plain text currently>
# monitor_interval=<sampling interval in milliseconds,
# default value is 10000>
[MySQL Monitor] [MySQL Monitor]
type=monitor type=monitor
@ -29,41 +7,19 @@ module=mysqlmon
servers=server1,server2,server3,server4 servers=server1,server2,server3,server4
user=maxuser user=maxuser
passwd=maxpwd passwd=maxpwd
monitor_interval=10000
# A series of service definition
#
# Valid options are:
#
# router=<name of router module>
# servers=<server name>,<server name>,...
# user=<User to fetch password inforamtion with>
# passwd=<Password of the user, plain text currently>
# enable_root_user=<0 or 1, default is 0>
# version_string=<specific string for server handshake,
# default is the MariaDB embedded library version>
#
# Valid router modules currently are:
# readwritesplit, readconnroute and debugcli
[RW Split Router] [RW Split Router]
type=service type=service
router=readwritesplit router=readwritesplit
servers=server1,server2,server3,server4 servers=server1,server2,server3,server4
max_slave_connections=90%
write_ses_variables_to_all=Yes
read_ses_variables_from_slaves=Yes
user=maxuser user=maxuser
passwd=maxpwd passwd=maxpwd
filters=Hint
[RW Split Hint Router] [RW Split Hint Router]
type=service type=service
router=readwritesplit router=readwritesplit
servers=server1,server2,server3,server4 servers=server1,server2,server3,server4
max_slave_connections=90%
write_ses_variables_to_all=Yes
read_ses_variables_from_slaves=Yes
user=maxuser user=maxuser
passwd=maxpwd passwd=maxpwd
filters=Hint filters=Hint
@ -77,31 +33,23 @@ servers=server1
user=maxuser user=maxuser
passwd=maxpwd passwd=maxpwd
[Hint]
[HTTPD Router] type=filter
type=service module=hintfilter
router=testroute
servers=server1,server2,server3
[Debug Interface] [Debug Interface]
type=service type=service
router=debugcli router=debugcli
[CLI]
type=service
router=cli
[Hint] [Read Connection Listener]
type=filter type=listener
module=hintfilter service=Read Connection Router
protocol=MySQLClient
port=4008
# Listener definitions for the services
#
# Valid options are:
#
# service=<name of service defined elsewhere>
# protocol=<name of protocol module with which to listen>
# port=<Listening port>
# address=<Address to bind to>
# socket=<Listening socket>
[RW Split Listener] [RW Split Listener]
type=listener type=listener
@ -115,28 +63,17 @@ service=RW Split Hint Router
protocol=MySQLClient protocol=MySQLClient
port=4009 port=4009
[Read Connection Listener]
type=listener
service=Read Connection Router
protocol=MySQLClient
port=4008
#socket=/tmp/readconn.sock
[Debug Listener] [Debug Listener]
type=listener type=listener
service=Debug Interface service=Debug Interface
protocol=telnetd protocol=telnetd
port=4442 port=4442
#address=127.0.0.1
[HTTPD Listener] [CLI Listener]
type=listener type=listener
service=HTTPD Router service=CLI
protocol=HTTPD protocol=maxscaled
port=6444 port=6603
# Definition of the servers
[server1] [server1]
type=server type=server
address=127.0.0.1 address=127.0.0.1