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

Merge remote-tracking branch 'origin/MAX-160' into MAX-237

Browse Source 
Conflicts:
	query_classifier/query_classifier.cc
	query_classifier/query_classifier.h
This commit is contained in:
Markus Makela
2014-08-29 11:02:03 +03:00
parent 531d8d7b47 cf5821d4ef
commit 7ea53f0141
21 changed files with 934 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
server/core/buffer.c
View File

@ -40,6 +40,11 @@
#include <atomic.h>
#include <skygw_debug.h>
static buffer_object_t* gwbuf_remove_buffer_object(
GWBUF* buf,
buffer_object_t* bufobj);
/**
* Allocate a new gateway buffer structure of size bytes.
*
@ -77,13 +82,15 @@ SHARED_BUF *sbuf;
free(sbuf);
return NULL;
}
spinlock_init(&rval->gwbuf_lock);
rval->start = sbuf->data;
rval->end = rval->start + size;
sbuf->refcount = 1;
rval->sbuf = sbuf;
rval->next = NULL;
rval->gwbuf_type = GWBUF_TYPE_UNDEFINED;
rval->command = 0;
rval->gwbuf_info = GWBUF_INFO_NONE;
rval->gwbuf_bufobj = NULL;
CHK_GWBUF(rval);
return rval;
}
@ -96,12 +103,20 @@ SHARED_BUF *sbuf;
void
gwbuf_free(GWBUF *buf)
{
buffer_object_t* bo;
CHK_GWBUF(buf);
if (atomic_add(&buf->sbuf->refcount, -1) == 1)
{
free(buf->sbuf->data);
free(buf->sbuf);
}
bo = buf->gwbuf_bufobj;
while (bo != NULL)
{
bo = gwbuf_remove_buffer_object(buf, bo);
}
}
free(buf);
}
@ -130,6 +145,8 @@ GWBUF *rval;
rval->start = buf->start;
rval->end = buf->end;
rval->gwbuf_type = buf->gwbuf_type;
rval->gwbuf_info = buf->gwbuf_info;
rval->gwbuf_bufobj = buf->gwbuf_bufobj;
rval->next = NULL;
CHK_GWBUF(rval);
return rval;
@ -156,6 +173,8 @@ GWBUF *gwbuf_clone_portion(
clonebuf->start = (void *)((char*)buf->start)+start_offset;
clonebuf->end = (void *)((char *)clonebuf->start)+length;
clonebuf->gwbuf_type = buf->gwbuf_type; /*< clone the type for now */
clonebuf->gwbuf_info = buf->gwbuf_info;
clonebuf->gwbuf_bufobj = buf->gwbuf_bufobj;
clonebuf->next = NULL;
CHK_GWBUF(clonebuf);
return clonebuf;
@ -338,5 +357,87 @@ void gwbuf_set_type(
}
}
/**
* Add a buffer object to GWBUF buffer.
*
* @param buf GWBUF where object is added
* @param id Type identifier for object
* @param data Object data
* @param donefun_dp Clean-up function to be executed before buffer is freed.
*/
void gwbuf_add_buffer_object(
GWBUF* buf,
bufobj_id_t id,
void* data,
void (*donefun_fp)(void *))
{
buffer_object_t** p_b;
buffer_object_t* newb;
CHK_GWBUF(buf);
newb = (buffer_object_t *)malloc(sizeof(buffer_object_t));
newb->bo_id = id;
newb->bo_data = data;
newb->bo_donefun_fp = donefun_fp;
newb->bo_next = NULL;
/** Lock */
spinlock_acquire(&buf->gwbuf_lock);
p_b = &buf->gwbuf_bufobj;
/** Search the end of the list and add there */
while (*p_b != NULL)
{
p_b = &(*p_b)->bo_next;
}
*p_b = newb;
/** Set flag */
buf->gwbuf_info |= GWBUF_INFO_PARSED;
/** Unlock */
spinlock_release(&buf->gwbuf_lock);
}
/**
* Search buffer object which matches with the id.
*
* @param buf GWBUF to be searched
* @param id Identifier for the object
*
* @return Searched buffer object or NULL if not found
*/
void* gwbuf_get_buffer_object_data(
GWBUF* buf,
bufobj_id_t id)
{
buffer_object_t* bo;
CHK_GWBUF(buf);
/** Lock */
spinlock_acquire(&buf->gwbuf_lock);
bo = buf->gwbuf_bufobj;
while (bo != NULL && bo->bo_id != id)
{
bo = bo->bo_next;
}
/** Unlock */
spinlock_release(&buf->gwbuf_lock);
return bo->bo_data;
}
/**
* @return pointer to next buffer object or NULL
*/
static buffer_object_t* gwbuf_remove_buffer_object(
GWBUF* buf,
buffer_object_t* bufobj)
{
buffer_object_t* next;
next = bufobj->bo_next;
/** Call corresponding clean-up function to clean buffer object's data */
bufobj->bo_donefun_fp(bufobj->bo_data);
free(bufobj);
return next;
}

55
server/core/modutil.c
View File

@ -29,6 +29,7 @@
*/
#include <buffer.h>
#include <string.h>
#include <mysql_client_server_protocol.h>
/**
* Check if a GWBUF structure is a MySQL COM_QUERY packet
@ -171,3 +172,57 @@ GWBUF *addition;
return orig;
}
/**
* Copy query string from GWBUF buffer to separate memory area.
*
* @param buf GWBUF buffer including the query
*
* @return Plaint text query if the packet type is COM_QUERY. Otherwise return
* a string including the packet type.
*/
char* modutil_get_query(
GWBUF* buf)
{
uint8_t* packet;
mysql_server_cmd_t packet_type;
size_t len;
char* query_str;
packet = GWBUF_DATA(buf);
packet_type = packet[4];
switch (packet_type) {
case MYSQL_COM_QUIT:
len = strlen("[Quit msg]")+1;
if ((query_str = (char *)malloc(len+1)) == NULL)
{
goto retblock;
}
memcpy(query_str, "[Quit msg]", len);
memset(&query_str[len], 0, 1);
break;
case MYSQL_COM_QUERY:
len = MYSQL_GET_PACKET_LEN(packet)-1; /*< distract 1 for packet type byte */
if ((query_str = (char *)malloc(len+1)) == NULL)
{
goto retblock;
}
memcpy(query_str, &packet[5], len);
memset(&query_str[len], 0, 1);
break;
default:
len = strlen(STRPACKETTYPE(packet_type))+1;
if ((query_str = (char *)malloc(len+1)) == NULL)
{
goto retblock;
}
memcpy(query_str, STRPACKETTYPE(packet_type), len);
memset(&query_str[len], 0, 1);
break;
} /*< switch */
retblock:
return query_str;
}

1
server/core/service.c
View File

@ -675,6 +675,7 @@ int n = 0;
"Unable to find filter '%s' for service '%s'\n",
trim(ptr), service->name
)));
n--;
}
flist[n] = NULL;
ptr = strtok_r(NULL, "|", &brkt);

46
server/include/buffer.h
View File

@ -42,8 +42,11 @@
* @endverbatim
*/
#include <skygw_debug.h>
#include <spinlock.h>
EXTERN_C_BLOCK_BEGIN
typedef enum
{
GWBUF_TYPE_UNDEFINED = 0x00,
@ -73,6 +76,35 @@ typedef struct {
int refcount; /*< Reference count on the buffer */
} SHARED_BUF;
typedef enum
{
GWBUF_INFO_NONE = 0x0,
GWBUF_INFO_PARSED = 0x1
} gwbuf_info_t;
#define GWBUF_IS_PARSED(b) (b->gwbuf_info & GWBUF_INFO_PARSED)
/**
* A structure for cleaning up memory allocations of structures which are
* referred to by GWBUF and deallocated in gwbuf_free but GWBUF doesn't
* know what they are.
* All functions on the list are executed before freeing memory of GWBUF struct.
*/
typedef enum
{
GWBUF_PARSING_INFO
} bufobj_id_t;
typedef struct buffer_object_st buffer_object_t;
struct buffer_object_st {
bufobj_id_t bo_id;
void* bo_data;
void (*bo_donefun_fp)(void *);
buffer_object_t* bo_next;
};
/**
* The buffer structure used by the descriptor control blocks.
*
@ -82,11 +114,13 @@ typedef struct {
* be copied within the gateway.
*/
typedef struct gwbuf {
SPINLOCK gwbuf_lock;
struct gwbuf *next; /*< Next buffer in a linked chain of buffers */
void *start; /*< Start of the valid data */
void *end; /*< First byte after the valid data */
SHARED_BUF *sbuf; /*< The shared buffer with the real data */
int command;/*< The command type for the queue */
buffer_object_t *gwbuf_bufobj; /*< List of objects referred to by GWBUF */
gwbuf_info_t gwbuf_info; /*< Info bits */
gwbuf_type_t gwbuf_type; /*< buffer's data type information */
} GWBUF;
@ -121,4 +155,14 @@ extern unsigned int gwbuf_length(GWBUF *head);
extern GWBUF *gwbuf_clone_portion(GWBUF *head, size_t offset, size_t len);
extern GWBUF *gwbuf_clone_transform(GWBUF *head, gwbuf_type_t type);
extern void gwbuf_set_type(GWBUF *head, gwbuf_type_t type);
void gwbuf_add_buffer_object(GWBUF* buf,
bufobj_id_t id,
void* data,
void (*donefun_fp)(void *));
void* gwbuf_get_buffer_object_data(GWBUF* buf, bufobj_id_t id);
EXTERN_C_BLOCK_END
#endif

2
server/include/modutil.h
View File

@ -36,4 +36,6 @@ extern int modutil_is_SQL(GWBUF *);
extern int modutil_extract_SQL(GWBUF *, char **, int *);
extern int modutil_MySQL_Query(GWBUF *, char **, int *, int *);
extern GWBUF *modutil_replace_SQL(GWBUF *, char *);
char* modutil_get_query(GWBUF* buf);
#endif

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

@ -31,6 +31,7 @@
#include <dcb.h>
#include <spinlock.h>
#include <modinfo.h>
#include <modutil.h>
#include <mysql_client_server_protocol.h>
MODULE_INFO info = {
@ -703,7 +704,7 @@ static void* newSession(
backend_ref[i].bref_sescmd_cur.scmd_cur_ptr_property =
&client_rses->rses_properties[RSES_PROP_TYPE_SESCMD];
backend_ref[i].bref_sescmd_cur.scmd_cur_cmd = NULL;
}
}
max_nslaves = rses_get_max_slavecount(client_rses, router_nservers);
max_slave_rlag = rses_get_max_replication_lag(client_rses);
@ -1031,9 +1032,6 @@ static int routeQuery(
GWBUF* querybuf)
{
skygw_query_type_t qtype = QUERY_TYPE_UNKNOWN;
GWBUF* plainsqlbuf = NULL;
char* querystr = NULL;
char* startpos;
mysql_server_cmd_t packet_type;
uint8_t* packet;
int ret = 0;
@ -1042,8 +1040,6 @@ static int routeQuery(
ROUTER_INSTANCE* inst = (ROUTER_INSTANCE *)instance;
ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
bool rses_is_closed = false;
size_t len;
MYSQL* mysql = NULL;
CHK_CLIENT_RSES(router_cli_ses);
@ -1066,6 +1062,8 @@ static int routeQuery(
*/
if (packet_type != MYSQL_COM_QUIT)
{
char* query_str = modutil_get_query(querybuf);
LOGIF(LE,
(skygw_log_write_flush(
LOGFILE_ERROR,
@ -1073,15 +1071,15 @@ static int routeQuery(
"backend server. %s.",
STRPACKETTYPE(packet_type),
STRQTYPE(qtype),
(querystr == NULL ? "(empty)" : querystr),
(query_str == NULL ? "(empty)" : query_str),
(rses_is_closed ? "Router was closed" :
"Router has no backend servers where to "
"route to"))));
free(querybuf);
}
goto return_ret;
}
inst->stats.n_queries++;
startpos = (char *)&packet[5];
master_dcb = router_cli_ses->rses_master_ref->bref_dcb;
CHK_DCB(master_dcb);
@ -1105,44 +1103,16 @@ static int routeQuery(
break;
case MYSQL_COM_QUERY:
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);
/**
* Use mysql handle to query information from parse tree.
* call skygw_query_classifier_free before exit!
*/
qtype = skygw_query_classifier_get_type(querystr, 0, &mysql);
qtype = query_classifier_get_type(querybuf);
break;
case MYSQL_COM_STMT_PREPARE:
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_classifier_get_type(querybuf);
qtype |= QUERY_TYPE_PREPARE_STMT;
break;
case MYSQL_COM_STMT_EXECUTE:
/** Parsing is not needed for this type of packet */
#if defined(NOT_USED)
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);
#endif
qtype = QUERY_TYPE_EXEC_STMT;
break;
@ -1207,7 +1177,7 @@ static int routeQuery(
*/
bool succp = route_session_write(
router_cli_ses,
querybuf,
gwbuf_clone(querybuf),
inst,
packet_type,
qtype);
@ -1238,7 +1208,7 @@ static int routeQuery(
if (succp)
{
if ((ret = slave_dcb->func.write(slave_dcb, querybuf)) == 1)
if ((ret = slave_dcb->func.write(slave_dcb, gwbuf_clone(querybuf))) == 1)
{
backend_ref_t* bref;
@ -1252,10 +1222,12 @@ static int routeQuery(
}
else
{
char* query_str = modutil_get_query(querybuf);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Routing query \"%s\" failed.",
querystr)));
(query_str == NULL ? "not available" : query_str))));
free(query_str);
}
}
rses_end_locked_router_action(router_cli_ses);
@ -1307,7 +1279,7 @@ static int routeQuery(
if (succp)
{
if ((ret = master_dcb->func.write(master_dcb, querybuf)) == 1)
if ((ret = master_dcb->func.write(master_dcb, gwbuf_clone(querybuf))) == 1)
{
backend_ref_t* bref;
@ -1333,18 +1305,23 @@ static int routeQuery(
}
}
return_ret:
if (plainsqlbuf != NULL)
#if defined(SS_DEBUG)
{
gwbuf_free(plainsqlbuf);
}
if (querystr != NULL)
{
free(querystr);
}
if (mysql != NULL)
{
skygw_query_classifier_free(mysql);
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
gwbuf_free(querybuf);
return ret;
}
@ -3663,4 +3640,3 @@ static BACKEND *get_root_master(backend_ref_t *servers, int router_nservers) {
}
return master_host;
}