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

Merge branch 'develop' into MAX-324

Browse Source 
Conflicts:
	query_classifier/query_classifier.h
This commit is contained in:
Markus Makela
2015-01-13 06:34:22 +02:00
parent 632190f9aa c7362cec23
commit 9177b6bcd3
19 changed files with 504 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
server/core/gateway.c
View File

@ -217,6 +217,7 @@ static void sigterm_handler (int i) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"MaxScale received signal SIGTERM. Exiting.")));
skygw_log_sync_all();
shutdown_server();
}
@ -228,6 +229,7 @@ sigint_handler (int i)
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"MaxScale received signal SIGINT. Shutting down.")));
skygw_log_sync_all();
shutdown_server();
fprintf(stderr, "\n\nShutting down MaxScale\n\n");
}
@ -270,6 +272,8 @@ sigfatal_handler (int i)
}
}
skygw_log_sync_all();
/* re-raise signal to enforce core dump */
fprintf(stderr, "\n\nWriting core dump\n");
signal_set(i, SIG_DFL);
@ -1527,7 +1531,7 @@ int main(int argc, char **argv)
free(log_context);
}
/*<
/**
* Init Log Manager for MaxScale.
* If $MAXSCALE_HOME is set then write the logs into $MAXSCALE_HOME/log.
* The skygw_logmanager_init expects to take arguments as passed to main
@ -1537,23 +1541,28 @@ int main(int argc, char **argv)
{
char buf[1024];
char *argv[8];
bool succp;
bool succp;
/** Set log directory under $MAXSCALE_HOME/log */
sprintf(buf, "%s/log", home_dir);
if(mkdir(buf, 0777) != 0){
if(errno != EEXIST){
fprintf(stderr,
"Error: Cannot create log directory: %s\n",buf);
goto return_main;
}
}
if(mkdir(buf, 0777) != 0)
{
if(errno != EEXIST)
{
fprintf(stderr,
"Error: Cannot create log directory: %s\n",
buf);
goto return_main;
}
}
argv[0] = "MaxScale";
argv[1] = "-j";
argv[2] = buf;
if (logtofile)
{
argv[3] = "-l"; /*< write to syslog */
/** Logs that should be syslogged */
argv[4] = "LOGFILE_MESSAGE,LOGFILE_ERROR"
"LOGFILE_DEBUG,LOGFILE_TRACE";
argv[5] = NULL;
@ -1562,9 +1571,9 @@ int main(int argc, char **argv)
else
{
argv[3] = "-s"; /*< store to shared memory */
argv[4] = "LOGFILE_DEBUG,LOGFILE_TRACE"; /*< ..these logs to shm */
argv[4] = "LOGFILE_DEBUG,LOGFILE_TRACE"; /*< to shm */
argv[5] = "-l"; /*< write to syslog */
argv[6] = "LOGFILE_MESSAGE,LOGFILE_ERROR"; /*< ..these logs to syslog */
argv[6] = "LOGFILE_MESSAGE,LOGFILE_ERROR"; /*< to syslog */
argv[7] = NULL;
succp = skygw_logmanager_init(7, argv);
}
@ -1575,8 +1584,7 @@ int main(int argc, char **argv)
goto return_main;
}
}
/*<
/**
* Resolve the full pathname for configuration file and check for
* read accessibility.
*/

15
server/core/poll.c
View File

@ -674,6 +674,14 @@ poll_set_maxwait(unsigned int maxwait)
* to process the DCB. If there are pending events the DCB will be moved to the
* back of the queue so that other DCB's will have a share of the threads to
* execute events for them.
*
* Including session id to log entries depends on this function. Assumption is
* that when maxscale thread starts processing of an event it processes one
* and only one session until it returns from this function. Session id is
* read to thread's local storage in macro LOGIF_MAYBE(...) and reset back
* to zero just before returning in LOGIF(...) macro.
* Thread local storage (tls_log_info_t) follows thread and is accessed every
* time log is written to particular log.
*
* @param thread_id The thread ID of the calling thread
* @return 0 if no DCB's have been processed
@ -798,7 +806,7 @@ unsigned long qtime;
simple_mutex_unlock(&dcb->dcb_write_lock);
#else
atomic_add(&pollStats.n_write, 1);
/** Read session id to thread's local storage */
LOGIF_MAYBE(LT, (dcb_get_ses_log_info(
dcb,
&tls_log_info.li_sesid,
@ -852,6 +860,7 @@ unsigned long qtime;
dcb,
dcb->fd)));
atomic_add(&pollStats.n_read, 1);
/** Read session id to thread's local storage */
LOGIF_MAYBE(LT, (dcb_get_ses_log_info(
dcb,
&tls_log_info.li_sesid,
@ -891,6 +900,7 @@ unsigned long qtime;
strerror(eno))));
}
atomic_add(&pollStats.n_error, 1);
/** Read session id to thread's local storage */
LOGIF_MAYBE(LT, (dcb_get_ses_log_info(
dcb,
&tls_log_info.li_sesid,
@ -919,6 +929,7 @@ unsigned long qtime;
{
dcb->flags |= DCBF_HUNG;
spinlock_release(&dcb->dcb_initlock);
/** Read session id to thread's local storage */
LOGIF_MAYBE(LT, (dcb_get_ses_log_info(
dcb,
&tls_log_info.li_sesid,
@ -951,6 +962,7 @@ unsigned long qtime;
{
dcb->flags |= DCBF_HUNG;
spinlock_release(&dcb->dcb_initlock);
/** Read session id to thread's local storage */
LOGIF_MAYBE(LT, (dcb_get_ses_log_info(
dcb,
&tls_log_info.li_sesid,
@ -1016,6 +1028,7 @@ unsigned long qtime;
}
}
dcb->evq.processing = 0;
/** Reset session id from thread's local storage */
LOGIF(LT, tls_log_info.li_sesid = 0);
spinlock_release(&pollqlock);

3
server/core/test/testadminusers.c
View File

@ -283,6 +283,9 @@ char *home, buf[1024];
result += test4();
result += test5();
/* Add the default user back so other tests can use it */
admin_add_user("admin", "skysql");
exit(result);
}

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

@ -60,6 +60,8 @@
#include <dcb.h>
#include <sys/time.h>
#include <poll.h>
#include <mysql_client_server_protocol.h>
#include <housekeeper.h>
#define MYSQL_COM_QUIT 0x01
#define MYSQL_COM_INITDB 0x02
@ -73,6 +75,11 @@
#define REPLY_TIMEOUT_SECOND 5
#define REPLY_TIMEOUT_MILLISECOND 1
#define PARENT 0
#define CHILD 1
#define PTR_IS_RESULTSET(b) (b[0] == 0x01 && b[1] == 0x0 && b[2] == 0x0 && b[3] == 0x01)
#define PTR_IS_EOF(b) (b[4] == 0xfe)
static unsigned char required_packets[] = {
MYSQL_COM_QUIT,
@ -153,10 +160,9 @@ typedef struct {
FILTER_DEF* dummy_filterdef;
int active; /* filter is active? */
int waiting; /* if the client is waiting for a reply */
int replies; /* Number of queries received */
int min_replies; /* Minimum number of replies to receive
* before forwarding the packet to the client*/
bool waiting[2]; /* if the client is waiting for a reply */
int eof[2];
int replies[2]; /* Number of queries received */
DCB *branch_dcb; /* Client DCB for "branch" service */
SESSION *branch_session;/* The branch service session */
int n_duped; /* Number of duplicated queries */
@ -202,6 +208,83 @@ static int hcfn(
return strcmp(i1,i2);
}
static void
orphan_free(void* data)
{
spinlock_acquire(&orphanLock);
orphan_session_t *ptr = allOrphans, *finished = NULL, *tmp = NULL;
#ifdef SS_DEBUG
int o_stopping = 0, o_ready = 0, o_freed = 0;
#endif
while(ptr)
{
if(ptr->session->state == SESSION_STATE_TO_BE_FREED)
{
if(ptr == allOrphans)
{
tmp = ptr;
allOrphans = ptr->next;
}
else
{
tmp = allOrphans;
while(tmp && tmp->next != ptr)
tmp = tmp->next;
if(tmp)
{
tmp->next = ptr->next;
tmp = ptr;
}
}
}
#ifdef SS_DEBUG
else if(ptr->session->state == SESSION_STATE_STOPPING)
{
o_stopping++;
}
else if(ptr->session->state == SESSION_STATE_ROUTER_READY)
{
o_ready++;
}
#endif
ptr = ptr->next;
if(tmp)
{
tmp->next = finished;
finished = tmp;
tmp = NULL;
}
}
spinlock_release(&orphanLock);
#ifdef SS_DEBUG
if(o_stopping + o_ready > 0)
skygw_log_write(LOGFILE_DEBUG, "tee.c: %d orphans in "
"SESSION_STATE_STOPPING, %d orphans in "
"SESSION_STATE_ROUTER_READY. ", o_stopping, o_ready);
#endif
while(finished)
{
o_freed++;
tmp = finished;
finished = finished->next;
tmp->session->service->router->freeSession(
tmp->session->service->router_instance,
tmp->session->router_session);
tmp->session->state = SESSION_STATE_FREE;
free(tmp->session);
free(tmp);
}
#ifdef SS_DEBUG
skygw_log_write(LOGFILE_DEBUG, "tee.c: %d orphans freed.", o_freed);
#endif
}
/**
* Implementation of the mandatory version entry point
*
@ -221,6 +304,7 @@ void
ModuleInit()
{
spinlock_init(&orphanLock);
hktask_add("tee orphan cleanup",orphan_free,NULL,15);
}
/**
@ -490,7 +574,6 @@ char *remote, *userName;
}
ses->tail = *dummy_upstream;
my_session->min_replies = 2;
my_session->branch_session = ses;
my_session->branch_dcb = dcb;
my_session->dummy_filterdef = dummy;
@ -605,78 +688,6 @@ SESSION* ses = my_session->branch_session;
}
free(session);
spinlock_acquire(&orphanLock);
orphan_session_t *ptr = allOrphans, *finished = NULL,*tmp = NULL;
#ifdef SS_DEBUG
int o_stopping = 0, o_ready = 0,o_freed = 0;
#endif
while(ptr)
{
if(ptr->session->state == SESSION_STATE_TO_BE_FREED)
{
if(ptr == allOrphans)
{
tmp = ptr;
allOrphans = ptr->next;
}
else
{
tmp = allOrphans;
while(tmp && tmp->next != ptr)
tmp = tmp->next;
if(tmp)
{
tmp->next = ptr->next;
tmp = ptr;
}
}
}
#ifdef SS_DEBUG
else if(ptr->session->state == SESSION_STATE_STOPPING)
{
o_stopping++;
}
else if(ptr->session->state == SESSION_STATE_ROUTER_READY)
{
o_ready++;
}
#endif
ptr = ptr->next;
if(tmp)
{
tmp->next = finished;
finished = tmp;
tmp = NULL;
}
}
spinlock_release(&orphanLock);
#ifdef SS_DEBUG
if(o_stopping + o_ready > 0)
skygw_log_write(LOGFILE_DEBUG,"tee.c: %d orphans in "
"SESSION_STATE_STOPPING, %d orphans in "
"SESSION_STATE_ROUTER_READY. ",o_stopping,o_ready);
#endif
while(finished)
{
#ifdef SS_DEBUG
skygw_log_write(LOGFILE_DEBUG,"tee.c: %d orphans freed.",++o_freed);
#endif
tmp = finished;
finished = finished->next;
tmp->session->service->router->freeSession(
tmp->session->service->router_instance,
tmp->session->router_session);
tmp->session->state = SESSION_STATE_FREE;
free(tmp->session);
free(tmp);
}
return;
}
/**
@ -761,8 +772,6 @@ GWBUF *clone = NULL;
(my_instance->nomatch == NULL ||
regexec(&my_instance->nore,ptr,0,NULL, 0) != 0))
{
char *dummy;
length = modutil_MySQL_query_len(queue, &residual);
clone = gwbuf_clone_all(queue);
my_session->residual = residual;
@ -775,9 +784,14 @@ GWBUF *clone = NULL;
}
}
/* Pass the query downstream */
my_session->replies = 0;
rval = my_session->down.routeQuery(my_session->down.instance,
ss_dassert(my_session->tee_replybuf == NULL);
memset(my_session->replies,0,2*sizeof(int));
memset(my_session->eof,0,2*sizeof(int));
memset(my_session->waiting,0,2*sizeof(bool));
rval = my_session->down.routeQuery(my_session->down.instance,
my_session->down.session,
queue);
if (clone)
@ -809,9 +823,42 @@ GWBUF *clone = NULL;
}
my_session->n_rejected++;
}
return rval;
}
/**
* Scans the GWBUF for EOF packets. If two packets for this session have been found
* from either the parent or the child branch, mark the response set from that branch as over.
* @param session The Tee filter session
* @param branch Parent or child branch
* @param reply Buffer to scan
*/
void
scan_resultset(TEE_SESSION *session, int branch, GWBUF *reply)
{
unsigned char* ptr = (unsigned char*) reply->start;
unsigned char* end = (unsigned char*) reply->end;
int pktlen = 0;
while(ptr < end)
{
pktlen = gw_mysql_get_byte3(ptr) + 4;
if(PTR_IS_EOF(ptr))
{
session->eof[branch]++;
if(session->eof[branch] == 2)
{
session->waiting[branch] = false;
session->eof[branch] = 0;
return;
}
}
ptr += pktlen;
}
}
/**
* The clientReply entry point. This is passed the response buffer
@ -826,33 +873,51 @@ GWBUF *clone = NULL;
static int
clientReply (FILTER* instance, void *session, GWBUF *reply)
{
int rc;
int rc, branch;
TEE_SESSION *my_session = (TEE_SESSION *) session;
spinlock_acquire(&my_session->tee_lock);
ss_dassert(my_session->active);
my_session->replies++;
if (my_session->tee_replybuf == NULL &&
instance != NULL)
{
my_session->tee_replybuf = reply;
}
else
{
gwbuf_free(reply);
}
if((my_session->branch_session == NULL ||
my_session->replies >= my_session->min_replies) &&
my_session->tee_replybuf != NULL)
branch = instance == NULL ? CHILD : PARENT;
unsigned char *ptr = (unsigned char*)reply->start;
if(my_session->replies[branch] == 0)
{
if(PTR_IS_RESULTSET(ptr))
{
my_session->waiting[branch] = true;
my_session->eof[branch] = 0;
}
}
if(my_session->waiting[branch])
{
scan_resultset(my_session,branch,reply);
}
if(branch == PARENT)
{
ss_dassert(my_session->tee_replybuf == NULL)
my_session->tee_replybuf = reply;
}
else
{
gwbuf_free(reply);
}
my_session->replies[branch]++;
if(my_session->tee_replybuf != NULL &&
(my_session->branch_session == NULL ||
my_session->waiting[PARENT] ||
(!my_session->waiting[CHILD] && !my_session->waiting[PARENT])))
{
rc = my_session->up.clientReply (
my_session->up.instance,
my_session->up.session,
my_session->tee_replybuf);
my_session->replies = 0;
my_session->tee_replybuf = NULL;
}
else

2
server/test/MaxScale_test.cnf
View File

@ -15,6 +15,7 @@ router=readwritesplit
servers=server1,server2,server3,server4
user=maxuser
passwd=maxpwd
max_slave_connections=100%
[DBShard Router]
type=service
@ -29,6 +30,7 @@ router=readwritesplit
servers=server1,server2,server3,server4
user=maxuser
passwd=maxpwd
max_slave_connections=100%
filters=Hint