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

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

Browse Source
This commit is contained in:
VilhoRaatikka
2014-05-28 10:10:57 +03:00
parent 056b5d5f37 70115d35c9
commit 1e1a78f535
30 changed files with 2404 additions and 653 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
Documentation/Debug And Diagnostic Support.pdf
View File

Binary file not shown.

BIN
Documentation/MaxScale 0.6 Release Notes.pdf Normal file
View File

Binary file not shown.

225
server/core/config.c
View File

@ -30,6 +30,7 @@
* 14/02/14 Massimiliano Pinto Added enable_root_user in the service_params list * 14/02/14 Massimiliano Pinto Added enable_root_user in the service_params list
* 11/03/14 Massimiliano Pinto Added Unix socket support * 11/03/14 Massimiliano Pinto Added Unix socket support
* 11/05/14 Massimiliano Pinto Added version_string support to service * 11/05/14 Massimiliano Pinto Added version_string support to service
* 19/05/14 Mark Riddoch Added unique names from section headers
* *
* @endverbatim * @endverbatim
*/ */
@ -129,7 +130,6 @@ int rval;
if (ptr) { if (ptr) {
*ptr = '\0'; *ptr = '\0';
} }
} }
mysql_close(conn); mysql_close(conn);
} }
@ -165,7 +165,6 @@ int rval;
if (!config_file) if (!config_file)
return 0; return 0;
if (gateway.version_string) if (gateway.version_string)
free(gateway.version_string); free(gateway.version_string);
@ -218,6 +217,8 @@ int error_count = 0;
"router"); "router");
if (router) if (router)
{ {
char* max_slave_conn_str;
obj->element = service_alloc(obj->object, router); obj->element = service_alloc(obj->object, router);
char *user = char *user =
config_get_value(obj->parameters, "user"); config_get_value(obj->parameters, "user");
@ -228,13 +229,6 @@ int error_count = 0;
char *version_string = config_get_value(obj->parameters, "version_string"); char *version_string = config_get_value(obj->parameters, "version_string");
if (version_string) {
((SERVICE *)(obj->element))->version_string = strdup(version_string);
} else {
if (gateway.version_string)
((SERVICE *)(obj->element))->version_string = strdup(gateway.version_string);
}
if (obj->element == NULL) /*< if module load failed */ if (obj->element == NULL) /*< if module load failed */
{ {
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
@ -248,6 +242,17 @@ int error_count = 0;
continue; /*< process next obj */ continue; /*< process next obj */
} }
if (version_string) {
((SERVICE *)(obj->element))->version_string = strdup(version_string);
} else {
if (gateway.version_string)
((SERVICE *)(obj->element))->version_string = strdup(gateway.version_string);
}
max_slave_conn_str =
config_get_value(obj->parameters,
"max_slave_connections");
if (enable_root_user) if (enable_root_user)
serviceEnableRootUser(obj->element, atoi(enable_root_user)); serviceEnableRootUser(obj->element, atoi(enable_root_user));
@ -267,6 +272,35 @@ int error_count = 0;
"corresponding password.", "corresponding password.",
obj->object))); obj->object)));
} }
if (max_slave_conn_str != NULL)
{
CONFIG_PARAMETER* param;
bool succp;
param = config_get_param(obj->parameters,
"max_slave_connections");
succp = service_set_slave_conn_limit(
obj->element,
param,
max_slave_conn_str,
COUNT_ATMOST);
if (!succp)
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"* Warning : invalid value type "
"for parameter \'%s.%s = %s\'\n\tExpected "
"type is either <int> for slave connection "
"count or\n\t<int>%% for specifying the "
"maximum percentage of available the "
"slaves that will be connected.",
((SERVICE*)obj->element)->name,
param->name,
param->value)));
}
}
} }
else else
{ {
@ -299,6 +333,7 @@ int error_count = 0;
obj->element = server_alloc(address, obj->element = server_alloc(address,
protocol, protocol,
atoi(port)); atoi(port));
server_set_unique_name(obj->element, obj->object);
} }
else else
{ {
@ -560,6 +595,89 @@ config_get_value(CONFIG_PARAMETER *params, const char *name)
return NULL; return NULL;
} }
CONFIG_PARAMETER* config_get_param(
CONFIG_PARAMETER* params,
const char* name)
{
while (params)
{
if (!strcmp(params->name, name))
return params;
params = params->next;
}
return NULL;
}
config_param_type_t config_get_paramtype(
CONFIG_PARAMETER* param)
{
return param->qfd_param_type;
}
int config_get_valint(
CONFIG_PARAMETER* param,
const char* name, /*< if NULL examine current param only */
config_param_type_t ptype)
{
int val = -1; /*< -1 indicates failure */
while (param)
{
if (name == NULL || !strncmp(param->name, name, MAX_PARAM_LEN))
{
switch (ptype) {
case COUNT_TYPE:
val = param->qfd.valcount;
goto return_val;
case PERCENT_TYPE:
val = param->qfd.valpercent;
goto return_val;
case BOOL_TYPE:
val = param->qfd.valbool;
goto return_val;
default:
goto return_val;
}
}
else if (name == NULL)
{
goto return_val;
}
param = param->next;
}
return_val:
return val;
}
CONFIG_PARAMETER* config_clone_param(
CONFIG_PARAMETER* param)
{
CONFIG_PARAMETER* p2;
p2 = (CONFIG_PARAMETER*) malloc(sizeof(CONFIG_PARAMETER));
if (p2 == NULL)
{
goto return_p2;
}
memcpy(p2, param, sizeof(CONFIG_PARAMETER));
p2->name = strndup(param->name, MAX_PARAM_LEN);
p2->value = strndup(param->value, MAX_PARAM_LEN);
if (param->qfd_param_type == STRING_TYPE)
{
p2->qfd.valstr = strndup(param->qfd.valstr, MAX_PARAM_LEN);
}
return_p2:
return p2;
}
/** /**
* Free a config tree * Free a config tree
* *
@ -671,6 +789,7 @@ SERVER *server;
char *user; char *user;
char *auth; char *auth;
char *enable_root_user; char *enable_root_user;
char* max_slave_conn_str;
char *version_string; char *version_string;
enable_root_user = config_get_value(obj->parameters, "enable_root_user"); enable_root_user = config_get_value(obj->parameters, "enable_root_user");
@ -683,8 +802,9 @@ SERVER *server;
version_string = config_get_value(obj->parameters, "version_string"); version_string = config_get_value(obj->parameters, "version_string");
if (version_string) { if (version_string) {
if (service->version_string) if (service->version_string) {
free(service->version_string); free(service->version_string);
}
service->version_string = strdup(version_string); service->version_string = strdup(version_string);
} }
@ -694,6 +814,42 @@ SERVER *server;
auth); auth);
if (enable_root_user) if (enable_root_user)
serviceEnableRootUser(service, atoi(enable_root_user)); serviceEnableRootUser(service, atoi(enable_root_user));
max_slave_conn_str =
config_get_value(
obj->parameters,
"max_slave_connections");
if (max_slave_conn_str != NULL)
{
CONFIG_PARAMETER* param;
bool succp;
param = config_get_param(obj->parameters,
"max_slave_connections");
succp = service_set_slave_conn_limit(
service,
param,
max_slave_conn_str,
COUNT_ATMOST);
if (!succp)
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"* Warning : invalid value type "
"for parameter \'%s.%s = %s\'\n\tExpected "
"type is either <int> for slave connection "
"count or\n\t<int>%% for specifying the "
"maximum percentage of available the "
"slaves that will be connected.",
((SERVICE*)obj->element)->name,
param->name,
param->value)));
}
}
} }
obj->element = service; obj->element = service;
@ -704,7 +860,9 @@ SERVER *server;
char *auth; char *auth;
char *enable_root_user; char *enable_root_user;
enable_root_user = config_get_value(obj->parameters, "enable_root_user"); enable_root_user =
config_get_value(obj->parameters,
"enable_root_user");
user = config_get_value(obj->parameters, user = config_get_value(obj->parameters,
"user"); "user");
@ -920,6 +1078,7 @@ static char *service_params[] =
"user", "user",
"passwd", "passwd",
"enable_root_user", "enable_root_user",
"max_slave_connections",
"version_string", "version_string",
NULL NULL
}; };
@ -1010,3 +1169,47 @@ int i;
obj = obj->next; obj = obj->next;
} }
} }
/**
* Set qualified parameter value to CONFIG_PARAMETER struct.
*/
bool config_set_qualified_param(
CONFIG_PARAMETER* param,
void* val,
config_param_type_t type)
{
bool succp;
switch (type) {
case STRING_TYPE:
param->qfd.valstr = strndup((const char *)val, MAX_PARAM_LEN);
succp = true;
break;
case COUNT_TYPE:
param->qfd.valcount = *(int *)val;
succp = true;
break;
case PERCENT_TYPE:
param->qfd.valpercent = *(int *)val;
succp = true;
break;
case BOOL_TYPE:
param->qfd.valbool = *(bool *)val;
succp = true;
break;
default:
succp = false;
break;
}
if (succp)
{
param->qfd_param_type = type;
}
return succp;
}

17
server/core/dbusers.c
View File

@ -184,6 +184,8 @@ getUsers(SERVICE *service, struct users *users)
} }
serviceGetUser(service, &service_user, &service_passwd); serviceGetUser(service, &service_user, &service_passwd);
if (service_user == NULL || service_passwd == NULL)
return -1;
/** multi-thread environment requires that thread init succeeds. */ /** multi-thread environment requires that thread init succeeds. */
if (mysql_thread_init()) { if (mysql_thread_init()) {
@ -218,15 +220,26 @@ getUsers(SERVICE *service, struct users *users)
*/ */
server = service->databases; server = service->databases;
dpwd = decryptPassword(service_passwd); dpwd = decryptPassword(service_passwd);
while (server != NULL && mysql_real_connect(con, while (server != NULL && (mysql_real_connect(con,
server->name, server->name,
service_user, service_user,
dpwd, dpwd,
NULL, NULL,
server->port, server->port,
NULL, NULL,
0) == NULL) 0) == NULL))
{ {
if (server == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to connect to %s:%d, \"%s\"",
server->name,
server->port,
mysql_error(con))));
mysql_close(con);
return -1;
}
server = server->nextdb; server = server->nextdb;
} }
free(dpwd); free(dpwd);

237
server/core/dcb.c
View File

@ -47,6 +47,7 @@
* error and 0 bytes to read. * error and 0 bytes to read.
* This fixes a bug with many reads from * This fixes a bug with many reads from
* backend * backend
* 07/05/2014 Mark Riddoch Addition of callback mechanism
* *
* @endverbatim * @endverbatim
*/ */
@ -81,6 +82,7 @@ static bool dcb_set_state_nomutex(
DCB* dcb, DCB* dcb,
const dcb_state_t new_state, const dcb_state_t new_state,
dcb_state_t* old_state); dcb_state_t* old_state);
static void dcb_call_callback(DCB *dcb, DCB_REASON reason);
DCB* dcb_get_zombies(void) DCB* dcb_get_zombies(void)
{ {
@ -95,8 +97,8 @@ DCB* dcb_get_zombies(void)
* *
* @return A newly allocated DCB or NULL if non could be allocated. * @return A newly allocated DCB or NULL if non could be allocated.
*/ */
DCB * dcb_alloc( DCB *
dcb_role_t role) dcb_alloc(dcb_role_t role)
{ {
DCB *rval; DCB *rval;
@ -118,13 +120,17 @@ DCB *rval;
spinlock_init(&rval->dcb_initlock); spinlock_init(&rval->dcb_initlock);
spinlock_init(&rval->writeqlock); spinlock_init(&rval->writeqlock);
spinlock_init(&rval->delayqlock); spinlock_init(&rval->delayqlock);
spinlock_init(&rval->dcb_readqlock);
spinlock_init(&rval->authlock); spinlock_init(&rval->authlock);
spinlock_init(&rval->cb_lock);
rval->fd = -1; rval->fd = -1;
memset(&rval->stats, 0, sizeof(DCBSTATS)); // Zero the statistics memset(&rval->stats, 0, sizeof(DCBSTATS)); // Zero the statistics
rval->state = DCB_STATE_ALLOC; rval->state = DCB_STATE_ALLOC;
bitmask_init(&rval->memdata.bitmask); bitmask_init(&rval->memdata.bitmask);
rval->writeqlen = 0;
rval->high_water = 0;
rval->low_water = 0;
rval->next = NULL; rval->next = NULL;
rval->callbacks = NULL;
spinlock_acquire(&dcbspin); spinlock_acquire(&dcbspin);
if (allDCBs == NULL) if (allDCBs == NULL)
@ -250,6 +256,8 @@ dcb_add_to_zombieslist(DCB *dcb)
static void static void
dcb_final_free(DCB *dcb) dcb_final_free(DCB *dcb)
{ {
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 not in DCB_STATE_DISCONNECTED state.");
@ -309,6 +317,19 @@ dcb_final_free(DCB *dcb)
GWBUF *queue = dcb->delayq; GWBUF *queue = dcb->delayq;
while ((queue = gwbuf_consume(queue, GWBUF_LENGTH(queue))) != NULL); while ((queue = gwbuf_consume(queue, GWBUF_LENGTH(queue))) != NULL);
} }
if (dcb->dcb_readqueue)
{
GWBUF* queue = dcb->dcb_readqueue;
while ((queue = gwbuf_consume(queue, GWBUF_LENGTH(queue))) != NULL);
}
spinlock_acquire(&dcb->cb_lock);
while ((cb = dcb->callbacks) != NULL)
{
dcb->callbacks = cb->next;
free(cb);
}
spinlock_release(&dcb->cb_lock);
if (dcb->dcb_readqueue) if (dcb->dcb_readqueue)
{ {
@ -690,16 +711,27 @@ return_n:
int int
dcb_write(DCB *dcb, GWBUF *queue) dcb_write(DCB *dcb, GWBUF *queue)
{ {
int w; int w, qlen;
int saved_errno = 0; int saved_errno = 0;
int below_water;
below_water = (dcb->high_water && dcb->writeqlen < dcb->high_water) ? 1 : 0;
ss_dassert(queue != NULL); ss_dassert(queue != NULL);
/**
* SESSION_STATE_STOPPING means that one of the backends is closing
* the router session. Some backends may have not completed
* authentication yet and thus they have no information about router
* being closed. Session state is changed to SESSION_STATE_STOPPING
* before router's closeSession is called and that tells that DCB may
* still be writable.
*/
if (queue == NULL || if (queue == NULL ||
(dcb->state != DCB_STATE_ALLOC && (dcb->state != DCB_STATE_ALLOC &&
dcb->state != DCB_STATE_POLLING && dcb->state != DCB_STATE_POLLING &&
dcb->state != DCB_STATE_LISTENING && dcb->state != DCB_STATE_LISTENING &&
dcb->state != DCB_STATE_NOPOLLING)) dcb->state != DCB_STATE_NOPOLLING &&
dcb->session->state != SESSION_STATE_STOPPING))
{ {
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
@ -710,6 +742,7 @@ dcb_write(DCB *dcb, GWBUF *queue)
dcb, dcb,
STRDCBSTATE(dcb->state), STRDCBSTATE(dcb->state),
dcb->fd))); dcb->fd)));
ss_dassert(false);
return 0; return 0;
} }
@ -726,6 +759,8 @@ dcb_write(DCB *dcb, GWBUF *queue)
* the routine that drains the queue data, so we should * the routine that drains the queue data, so we should
* not have a race condition on the event. * not have a race condition on the event.
*/ */
qlen = gwbuf_length(queue);
atomic_add(&dcb->writeqlen, qlen);
dcb->writeq = gwbuf_append(dcb->writeq, queue); dcb->writeq = gwbuf_append(dcb->writeq, queue);
dcb->stats.n_buffered++; dcb->stats.n_buffered++;
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
@ -838,6 +873,8 @@ dcb_write(DCB *dcb, GWBUF *queue)
* for suspended write. * for suspended write.
*/ */
dcb->writeq = queue; dcb->writeq = queue;
qlen = gwbuf_length(queue);
atomic_add(&dcb->writeqlen, qlen);
if (queue != NULL) if (queue != NULL)
{ {
@ -861,6 +898,13 @@ dcb_write(DCB *dcb, GWBUF *queue)
return 0; return 0;
} }
spinlock_release(&dcb->writeqlock); spinlock_release(&dcb->writeqlock);
if (dcb->high_water && dcb->writeqlen > dcb->high_water && below_water)
{
atomic_add(&dcb->stats.n_high_water, 1);
dcb_call_callback(dcb, DCB_REASON_HIGH_WATER);
}
return 1; return 1;
} }
@ -878,6 +922,9 @@ dcb_drain_writeq(DCB *dcb)
int n = 0; int n = 0;
int w; int w;
int saved_errno = 0; int saved_errno = 0;
int above_water;
above_water = (dcb->low_water && dcb->writeqlen > dcb->low_water) ? 1 : 0;
spinlock_acquire(&dcb->writeqlock); spinlock_acquire(&dcb->writeqlock);
if (dcb->writeq) if (dcb->writeq)
@ -938,6 +985,17 @@ int saved_errno = 0;
} }
} }
spinlock_release(&dcb->writeqlock); spinlock_release(&dcb->writeqlock);
atomic_add(&dcb->writeqlen, -n);
/* The write queue has drained, potentially need to call a callback function */
if (dcb->writeq == NULL)
dcb_call_callback(dcb, DCB_REASON_DRAINED);
if (above_water && dcb->writeqlen < dcb->low_water)
{
atomic_add(&dcb->stats.n_low_water, 1);
dcb_call_callback(dcb, DCB_REASON_LOW_WATER);
}
return n; return n;
} }
@ -980,6 +1038,8 @@ dcb_close(DCB *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);
dcb_call_callback(dcb, DCB_REASON_CLOSE);
if (rc == 0) { if (rc == 0) {
LOGIF(LD, (skygw_log_write( LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG, LOGFILE_DEBUG,
@ -1022,6 +1082,8 @@ printDCB(DCB *dcb)
printf("\t\tNo. of Writes: %d\n", dcb->stats.n_writes); printf("\t\tNo. of Writes: %d\n", dcb->stats.n_writes);
printf("\t\tNo. of Buffered Writes: %d\n", dcb->stats.n_buffered); printf("\t\tNo. of Buffered Writes: %d\n", dcb->stats.n_buffered);
printf("\t\tNo. of Accepts: %d\n", dcb->stats.n_accepts); printf("\t\tNo. of Accepts: %d\n", dcb->stats.n_accepts);
printf("\t\tNo. of High Water Events: %d\n", dcb->stats.n_high_water);
printf("\t\tNo. of Low Water Events: %d\n", dcb->stats.n_low_water);
} }
/** /**
@ -1068,6 +1130,8 @@ DCB *dcb;
dcb_printf(pdcb, "\t\tNo. of Writes: %d\n", dcb->stats.n_writes); dcb_printf(pdcb, "\t\tNo. of Writes: %d\n", dcb->stats.n_writes);
dcb_printf(pdcb, "\t\tNo. of Buffered Writes: %d\n", dcb->stats.n_buffered); dcb_printf(pdcb, "\t\tNo. of Buffered Writes: %d\n", dcb->stats.n_buffered);
dcb_printf(pdcb, "\t\tNo. of Accepts: %d\n", dcb->stats.n_accepts); dcb_printf(pdcb, "\t\tNo. of Accepts: %d\n", dcb->stats.n_accepts);
dcb_printf(pdcb, "\t\tNo. of High Water Events: %d\n", dcb->stats.n_high_water);
dcb_printf(pdcb, "\t\tNo. of Low Water Events: %d\n", dcb->stats.n_low_water);
dcb = dcb->next; dcb = dcb->next;
} }
spinlock_release(&dcbspin); spinlock_release(&dcbspin);
@ -1093,6 +1157,8 @@ dprintDCB(DCB *pdcb, DCB *dcb)
dcb_printf(pdcb, "\t\tNo. of Writes: %d\n", dcb->stats.n_writes); dcb_printf(pdcb, "\t\tNo. of Writes: %d\n", dcb->stats.n_writes);
dcb_printf(pdcb, "\t\tNo. of Buffered Writes: %d\n", dcb->stats.n_buffered); dcb_printf(pdcb, "\t\tNo. of Buffered Writes: %d\n", dcb->stats.n_buffered);
dcb_printf(pdcb, "\t\tNo. of Accepts: %d\n", dcb->stats.n_accepts); dcb_printf(pdcb, "\t\tNo. of Accepts: %d\n", dcb->stats.n_accepts);
dcb_printf(pdcb, "\t\tNo. of High Water Events: %d\n", dcb->stats.n_high_water);
dcb_printf(pdcb, "\t\tNo. of Low Water Events: %d\n", dcb->stats.n_low_water);
} }
/** /**
@ -1429,4 +1495,163 @@ int gw_write(
return w; return w;
} }
/**
* Add a callback
*
* Duplicate registrations are not allowed, therefore an error will be returned if
* the specific function, reason and userdata triple are already registered.
* An error will also be returned if the is insufficient memeory available to
* create the registration.
*
* @param dcb The DCB to add the callback to
* @param reason The callback reason
* @param cb The callback function to call
* @param userdata User data to send in the call
* @return Non-zero (true) if the callback was added
*/
int
dcb_add_callback(DCB *dcb, DCB_REASON reason, int (*callback)(struct dcb *, DCB_REASON, void *), void *userdata)
{
DCB_CALLBACK *cb, *ptr;
int rval = 1;
if ((ptr = (DCB_CALLBACK *)malloc(sizeof(DCB_CALLBACK))) == NULL)
{
return 0;
}
ptr->reason = reason;
ptr->cb = callback;
ptr->userdata = userdata;
ptr->next = NULL;
spinlock_acquire(&dcb->cb_lock);
cb = dcb->callbacks;
if (cb == NULL)
{
dcb->callbacks = ptr;
spinlock_release(&dcb->cb_lock);
}
else
{
while (cb)
{
if (cb->reason == reason && cb->cb == callback &&
cb->userdata == userdata)
{
free(ptr);
spinlock_release(&dcb->cb_lock);
return 0;
}
if (cb->next == NULL)
cb->next = ptr;
cb = cb->next;
}
spinlock_release(&dcb->cb_lock);
}
return rval;
}
/**
* Remove a callback from the callback list for the DCB
*
* Searches down the linked list to find he callback with a matching reason, function
* and userdata.
*
* @param dcb The DCB to add the callback to
* @param reason The callback reason
* @param cb The callback function to call
* @param userdata User data to send in the call
* @return Non-zero (true) if the callback was removed
*/
int
dcb_remove_callback(DCB *dcb, DCB_REASON reason, int (*callback)(struct dcb *, DCB_REASON), void *userdata)
{
DCB_CALLBACK *cb, *pcb = NULL;
int rval = 0;
spinlock_acquire(&dcb->cb_lock);
cb = dcb->callbacks;
if (cb == NULL)
{
rval = 0;
}
else
{
while (cb)
{
if (cb->reason == reason && cb->cb == callback
&& cb->userdata == userdata)
{
if (pcb == NULL)
pcb->next = cb->next;
else
dcb->callbacks = cb->next;
spinlock_release(&dcb->cb_lock);
free(cb);
rval = 1;
break;
}
pcb = cb;
cb = cb->next;
}
}
if (!rval)
spinlock_release(&dcb->cb_lock);
return rval;
}
/**
* Call the set of callbacks registered for a particular reason.
*
* @param dcb The DCB to call the callbacks regarding
* @param reason The reason that has triggered the call
*/
static void
dcb_call_callback(DCB *dcb, DCB_REASON reason)
{
DCB_CALLBACK *cb, *nextcb;
spinlock_acquire(&dcb->cb_lock);
cb = dcb->callbacks;
while (cb)
{
if (cb->reason == reason)
{
nextcb = cb->next;
spinlock_release(&dcb->cb_lock);
cb->cb(dcb, reason, cb->userdata);
spinlock_acquire(&dcb->cb_lock);
cb = nextcb;
}
else
cb = cb->next;
}
spinlock_release(&dcb->cb_lock);
}
/**
* Check the passed DCB to ensure it is in the list of allDCBS
*
* @param DCB The DCB to check
* @return 1 if the DCB is in the list, otherwise 0
*/
int
dcb_isvalid(DCB *dcb)
{
DCB *ptr;
int rval = 0;
spinlock_acquire(&dcbspin);
ptr = allDCBs;
while (ptr)
{
if (ptr == dcb)
{
rval = 1;
break;
}
ptr = ptr->next;
}
spinlock_release(&dcbspin);
return rval;
}

4
server/core/gateway.c
View File

@ -136,7 +136,7 @@ static void usage(void);
static char* get_expanded_pathname( static char* get_expanded_pathname(
char** abs_path, char** abs_path,
char* input_path, char* input_path,
char* fname); const char* fname);
static void print_log_n_stderr( static void print_log_n_stderr(
bool do_log, bool do_log,
bool do_stderr, bool do_stderr,
@ -725,7 +725,7 @@ static bool file_is_writable(
static char* get_expanded_pathname( static char* get_expanded_pathname(
char** output_path, char** output_path,
char* relative_path, char* relative_path,
char* fname) const char* fname)
{ {
char* cnf_file_buf = NULL; char* cnf_file_buf = NULL;
char* expanded_path; char* expanded_path;

23
server/core/monitor.c
View File

@ -197,3 +197,26 @@ MONITOR *ptr;
} }
spinlock_release(&monLock); spinlock_release(&monLock);
} }
/**
* Find a monitor by name
*
* @param name The name of the monitor
* @return Pointer to the monitor or NULL
*/
MONITOR *
monitor_find(char *name)
{
MONITOR *ptr;
spinlock_acquire(&monLock);
ptr = allMonitors;
while (ptr)
{
if (!strcmp(ptr->name, name))
break;
ptr = ptr->next;
}
spinlock_release(&monLock);
return ptr;
}

53
server/core/server.c
View File

@ -24,6 +24,8 @@
* *
* Date Who Description * Date Who Description
* 18/06/13 Mark Riddoch Initial implementation * 18/06/13 Mark Riddoch Initial implementation
* 17/05/14 Mark Riddoch Addition of unique_name
* 20/05/14 Massimiliano Pinto Addition of server_string
* *
* @endverbatim * @endverbatim
*/ */
@ -67,6 +69,8 @@ SERVER *server;
server->nextdb = NULL; server->nextdb = NULL;
server->monuser = NULL; server->monuser = NULL;
server->monpw = NULL; server->monpw = NULL;
server->unique_name = NULL;
server->server_string = NULL;
spinlock_acquire(&server_spin); spinlock_acquire(&server_spin);
server->next = allServers; server->next = allServers;
@ -109,10 +113,51 @@ SERVER *ptr;
/* Clean up session and free the memory */ /* Clean up session and free the memory */
free(server->name); free(server->name);
free(server->protocol); free(server->protocol);
if (server->unique_name)
free(server->unique_name);
if (server->server_string)
free(server->server_string);
free(server); free(server);
return 1; return 1;
} }
/**
* Set a unique name for the server
*
* @param server The server to ste the name on
* @param name The unique name for the server
*/
void
server_set_unique_name(SERVER *server, char *name)
{
server->unique_name = strdup(name);
}
/**
* Find an existing server using the unique section name in
* configuration file
*
* @param servname The Server name or address
* @param port The server port
* @return The server or NULL if not found
*/
SERVER *
server_find_by_unique_name(char *name)
{
SERVER *server;
spinlock_acquire(&server_spin);
server = allServers;
while (server)
{
if (strcmp(server->unique_name, name) == 0)
break;
server = server->next;
}
spinlock_release(&server_spin);
return server;
}
/** /**
* Find an existing server * Find an existing server
* *
@ -190,13 +235,15 @@ char *stat;
ptr = allServers; ptr = allServers;
while (ptr) while (ptr)
{ {
dcb_printf(dcb, "Server %p\n", ptr); dcb_printf(dcb, "Server %p (%s)\n", ptr, ptr->unique_name);
dcb_printf(dcb, "\tServer: %s\n", ptr->name); dcb_printf(dcb, "\tServer: %s\n", ptr->name);
stat = server_status(ptr); stat = server_status(ptr);
dcb_printf(dcb, "\tStatus: %s\n", stat); dcb_printf(dcb, "\tStatus: %s\n", stat);
free(stat); free(stat);
dcb_printf(dcb, "\tProtocol: %s\n", ptr->protocol); dcb_printf(dcb, "\tProtocol: %s\n", ptr->protocol);
dcb_printf(dcb, "\tPort: %d\n", ptr->port); dcb_printf(dcb, "\tPort: %d\n", ptr->port);
if (ptr->server_string)
dcb_printf(dcb, "\tServer Version:\t\t%s\n", ptr->server_string);
dcb_printf(dcb, "\tNumber of connections: %d\n", ptr->stats.n_connections); dcb_printf(dcb, "\tNumber of connections: %d\n", ptr->stats.n_connections);
dcb_printf(dcb, "\tCurrent no. of connections: %d\n", ptr->stats.n_current); dcb_printf(dcb, "\tCurrent no. of connections: %d\n", ptr->stats.n_current);
ptr = ptr->next; ptr = ptr->next;
@ -215,13 +262,15 @@ dprintServer(DCB *dcb, SERVER *server)
{ {
char *stat; char *stat;
dcb_printf(dcb, "Server %p\n", server); dcb_printf(dcb, "Server %p (%s)\n", server, server->unique_name);
dcb_printf(dcb, "\tServer: %s\n", server->name); dcb_printf(dcb, "\tServer: %s\n", server->name);
stat = server_status(server); stat = server_status(server);
dcb_printf(dcb, "\tStatus: %s\n", stat); dcb_printf(dcb, "\tStatus: %s\n", stat);
free(stat); free(stat);
dcb_printf(dcb, "\tProtocol: %s\n", server->protocol); dcb_printf(dcb, "\tProtocol: %s\n", server->protocol);
dcb_printf(dcb, "\tPort: %d\n", server->port); dcb_printf(dcb, "\tPort: %d\n", server->port);
if (server->server_string)
dcb_printf(dcb, "\tServer Version:\t\t%s\n", server->server_string);
dcb_printf(dcb, "\tNumber of connections: %d\n", server->stats.n_connections); dcb_printf(dcb, "\tNumber of connections: %d\n", server->stats.n_connections);
dcb_printf(dcb, "\tCurrent No. of connections: %d\n", server->stats.n_current); dcb_printf(dcb, "\tCurrent No. of connections: %d\n", server->stats.n_current);
} }

143
server/core/service.c
View File

@ -29,6 +29,7 @@
* 25/02/14 Massimiliano Pinto Added: service refresh limit feature * 25/02/14 Massimiliano Pinto Added: service refresh limit feature
* 28/02/14 Massimiliano Pinto users_alloc moved from service_alloc to serviceStartPort (generic hashable for services) * 28/02/14 Massimiliano Pinto users_alloc moved from service_alloc to serviceStartPort (generic hashable for services)
* 07/05/14 Massimiliano Pinto Added: version_string initialized to NULL * 07/05/14 Massimiliano Pinto Added: version_string initialized to NULL
* 23/05/14 Mark Riddoch Addition of service validation call
* *
* @endverbatim * @endverbatim
*/ */
@ -55,6 +56,11 @@ extern int lm_enabled_logfiles_bitmask;
static SPINLOCK service_spin = SPINLOCK_INIT; static SPINLOCK service_spin = SPINLOCK_INIT;
static SERVICE *allServices = NULL; static SERVICE *allServices = NULL;
static void service_add_qualified_param(
SERVICE* svc,
CONFIG_PARAMETER* param);
/** /**
* Allocate a new service for the gateway to support * Allocate a new service for the gateway to support
* *
@ -102,6 +108,8 @@ SERVICE *service;
service->enable_root = 0; service->enable_root = 0;
service->routerOptions = NULL; service->routerOptions = NULL;
service->databases = NULL; service->databases = NULL;
service->svc_config_param = NULL;
service->svc_config_version = 0;
spinlock_init(&service->spin); spinlock_init(&service->spin);
spinlock_init(&service->users_table_spin); spinlock_init(&service->users_table_spin);
memset(&service->rate_limit, 0, sizeof(SERVICE_REFRESH_RATE)); memset(&service->rate_limit, 0, sizeof(SERVICE_REFRESH_RATE));
@ -114,6 +122,33 @@ SERVICE *service;
return service; return service;
} }
/**
* Check to see if a service pointer is valid
*
* @param service The poitner to check
* @return 1 if the service is in the list of all services
*/
int
service_isvalid(SERVICE *service)
{
SERVICE *ptr;
int rval = 0;
spinlock_acquire(&service_spin);
ptr = allServices;
while (ptr)
{
if (ptr == service)
{
rval = 1;
break;
}
ptr = ptr->next;
}
spinlock_release(&service_spin);
return rval;
}
/** /**
* Start an individual port/protocol pair * Start an individual port/protocol pair
* *
@ -782,3 +817,111 @@ int service_refresh_users(SERVICE *service) {
else else
return 1; return 1;
} }
bool service_set_slave_conn_limit (
SERVICE* service,
CONFIG_PARAMETER* param,
char* valstr,
count_spec_t count_spec)
{
char* p;
int valint;
bool percent = false;
bool succp;
/**
* Find out whether the value is numeric and ends with '%' or '\0'
*/
p = valstr;
while(isdigit(*p)) p++;
errno = 0;
if (p == valstr || (*p != '%' && *p != '\0'))
{
succp = false;
}
else if (*p == '%')
{
if (*(p+1) == '\0')
{
*p = '\0';
valint = (int) strtol(valstr, (char **)NULL, 10);
if (valint == 0 && errno != 0)
{
succp = false;
}
else
{
succp = true;
config_set_qualified_param(param, (void *)&valint, PERCENT_TYPE);
}
}
else
{
succp = false;
}
}
else if (*p == '\0')
{
valint = (int) strtol(valstr, (char **)NULL, 10);
if (valint == 0 && errno != 0)
{
succp = false;
}
else
{
succp = true;
config_set_qualified_param(param, (void *)&valint, COUNT_TYPE);
}
}
if (succp)
{
service_add_qualified_param(service, param); /*< add param to svc */
}
return succp;
}
/**
* Add qualified config parameter to SERVICE struct.
*/
static void service_add_qualified_param(
SERVICE* svc,
CONFIG_PARAMETER* param)
{
CONFIG_PARAMETER** p;
spinlock_acquire(&svc->spin);
p = &svc->svc_config_param;
if ((*p) != NULL)
{
do
{
/** If duplicate is found, latter remains */
if (strncasecmp(param->name,
(*p)->name,
strlen(param->name)) == 0)
{
*p = config_clone_param(param);
break;
}
}
while ((*p)->next != NULL);
(*p)->next = config_clone_param(param);
}
else
{
(*p) = config_clone_param(param);
}
/** Increment service's configuration version */
atomic_add(&svc->svc_config_version, 1);
(*p)->next = NULL;
spinlock_release(&svc->spin);
}

55
server/core/session.c
View File

@ -114,7 +114,7 @@ session_alloc(SERVICE *service, DCB *client_dcb)
/* /*
* Only create a router session if we are not the listening * Only create a router session if we are not the listening
* DCB. Creating a router session may create a connection to a * DCB or an internal DCB. Creating a router session may create a connection to a
* backend server, depending upon the router module implementation * backend server, depending upon the router module implementation
* and should be avoided for the listener session * and should be avoided for the listener session
* *
@ -122,13 +122,17 @@ 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) 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,
session); session);
if (session->router_session == NULL) { if (session->router_session == NULL) {
/**
* Inform other threads that session is closing.
*/
session->state == SESSION_STATE_STOPPING;
/*< /*<
* Decrease refcount, set dcb's session pointer NULL * Decrease refcount, set dcb's session pointer NULL
* and set session pointer to NULL. * and set session pointer to NULL.
@ -138,8 +142,8 @@ session_alloc(SERVICE *service, DCB *client_dcb)
session = NULL; session = NULL;
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"Error : Failed to create router " "Error : Failed to create %s session.",
"client session. Freeing allocated resources."))); service->name)));
goto return_session; goto return_session;
} }
@ -269,6 +273,34 @@ return_succp :
return succp; return succp;
} }
/**
* Check to see if a session is valid, i.e. in the list of all sessions
*
* @param session Session to check
* @return 1 if the session is valid otherwise 0
*/
int
session_isvalid(SESSION *session)
{
SESSION *ptr;
int rval = 0;
spinlock_acquire(&session_spin);
ptr = allSessions;
while (ptr)
{
if (ptr == session)
{
rval = 1;
break;
}
ptr = ptr->next;
}
spinlock_release(&session_spin);
return rval;
}
/** /**
* Print details of an individual session * Print details of an individual session
* *
@ -441,3 +473,18 @@ session_state(int state)
return "Invalid State"; return "Invalid State";
} }
} }
SESSION* get_session_by_router_ses(
void* rses)
{
SESSION* ses = allSessions;
while (ses->router_session != rses && ses->next != NULL)
ses = ses->next;
if (ses->router_session != rses)
{
ses = NULL;
}
return ses;
}

36
server/include/config.h
View File

@ -17,6 +17,7 @@
* *
* Copyright SkySQL Ab 2013 * Copyright SkySQL Ab 2013
*/ */
#include <skygw_utils.h>
/** /**
* @file config.h The configuration handling elements * @file config.h The configuration handling elements
@ -31,12 +32,32 @@
* @endverbatim * @endverbatim
*/ */
/**
* Maximum length for configuration parameter value.
*/
enum {MAX_PARAM_LEN=256};
typedef enum {
UNDEFINED_TYPE=0,
STRING_TYPE,
COUNT_TYPE,
PERCENT_TYPE,
BOOL_TYPE
} config_param_type_t;
/** /**
* The config parameter * The config parameter
*/ */
typedef struct config_parameter { typedef struct config_parameter {
char *name; /**< The name of the parameter */ char *name; /**< The name of the parameter */
char *value; /**< The value of the parameter */ char *value; /**< The value of the parameter */
union { /*< qualified parameter value by type */
char* valstr; /*< terminated char* array */
int valcount; /*< int */
int valpercent; /*< int */
bool valbool; /*< bool */
} qfd;
config_param_type_t qfd_param_type;
struct config_parameter *next; /**< Next pointer in the linked list */ struct config_parameter *next; /**< Next pointer in the linked list */
} CONFIG_PARAMETER; } CONFIG_PARAMETER;
@ -62,4 +83,19 @@ typedef struct {
extern int config_load(char *); extern int config_load(char *);
extern int config_reload(); extern int config_reload();
extern int config_threadcount(); extern int config_threadcount();
CONFIG_PARAMETER* config_get_param(CONFIG_PARAMETER* params, const char* name);
config_param_type_t config_get_paramtype(CONFIG_PARAMETER* param);
CONFIG_PARAMETER* config_clone_param(CONFIG_PARAMETER* param);
bool config_set_qualified_param(
CONFIG_PARAMETER* param,
void* val,
config_param_type_t type);
int config_get_valint(
CONFIG_PARAMETER* param,
const char* name, /*< if NULL examine current param only */
config_param_type_t ptype);
#endif #endif

48
server/include/dcb.h
View File

@ -48,6 +48,8 @@ struct service;
* 15/07/2013 Massimiliano Pinto Added session entry point * 15/07/2013 Massimiliano Pinto Added session entry point
* 16/07/2013 Massimiliano Pinto Added command type for dcb * 16/07/2013 Massimiliano Pinto Added command type for dcb
* 07/02/2014 Massimiliano Pinto Added ipv4 data struct into for dcb * 07/02/2014 Massimiliano Pinto Added ipv4 data struct into for dcb
* 07/05/2014 Mark Riddoch Addition of callback mechanism
* 08/05/2014 Mark Riddoch Addition of writeq high and low watermarks
* *
* @endverbatim * @endverbatim
*/ */
@ -99,6 +101,8 @@ typedef struct dcbstats {
int n_writes; /*< Number of writes on this descriptor */ int n_writes; /*< Number of writes on this descriptor */
int n_accepts; /*< Number of accepts on this descriptor */ int n_accepts; /*< Number of accepts on this descriptor */
int n_buffered; /*< Number of buffered writes */ int n_buffered; /*< Number of buffered writes */
int n_high_water; /*< Number of crosses of high water mark */
int n_low_water; /*< Number of crosses of low water mark */
} DCBSTATS; } DCBSTATS;
/** /**
@ -138,9 +142,34 @@ typedef enum {
typedef enum { typedef enum {
DCB_ROLE_SERVICE_LISTENER, /*< Receives initial connect requests from clients */ DCB_ROLE_SERVICE_LISTENER, /*< Receives initial connect requests from clients */
DCB_ROLE_REQUEST_HANDLER /*< Serves dedicated client */ DCB_ROLE_REQUEST_HANDLER, /*< Serves dedicated client */
DCB_ROLE_INTERNAL /*< Internal DCB not connected to the outside */
} dcb_role_t; } dcb_role_t;
/**
* Callback reasons for the DCB callback mechanism.
*/
typedef enum {
DCB_REASON_CLOSE, /*< The DCB is closing */
DCB_REASON_DRAINED, /*< The write delay queue has drained */
DCB_REASON_HIGH_WATER, /*< Cross high water mark */
DCB_REASON_LOW_WATER, /*< Cross low water mark */
DCB_REASON_ERROR, /*< An error was flagged on the connection */
DCB_REASON_HUP /*< A hangup was detected */
} DCB_REASON;
/**
* Callback structure - used to track callbacks registered on a DCB
*/
typedef struct dcb_callback {
DCB_REASON reason; /*< The reason for the callback */
int (*cb)(struct dcb *dcb, DCB_REASON reason, void *userdata);
void *userdata; /*< User data to be sent in the callback */
struct dcb_callback
*next; /*< Next callback for this DCB */
} DCB_CALLBACK;
/** /**
* Descriptor Control Block * Descriptor Control Block
* *
@ -172,12 +201,12 @@ typedef struct dcb {
struct session *session; /**< The owning session */ struct session *session; /**< The owning session */
GWPROTOCOL func; /**< The functions for this descriptor */ GWPROTOCOL func; /**< The functions for this descriptor */
unsigned int writeqlen; /**< Current number of byes in the write queue */
SPINLOCK writeqlock; /**< Write Queue spinlock */ SPINLOCK writeqlock; /**< Write Queue spinlock */
GWBUF *writeq; /**< Write Data Queue */ GWBUF *writeq; /**< Write Data Queue */
SPINLOCK delayqlock; /**< Delay Backend Write Queue spinlock */ SPINLOCK delayqlock; /**< Delay Backend Write Queue spinlock */
GWBUF *delayq; /**< Delay Backend Write Data Queue */ GWBUF *delayq; /**< Delay Backend Write Data Queue */
GWBUF *dcb_readqueue; /**< read queue for storing incomplete reads */ GWBUF *dcb_readqueue; /**< read queue for storing incomplete reads */
SPINLOCK dcb_readqlock; /**< read/write access protection to read queue */
SPINLOCK authlock; /**< Generic Authorization spinlock */ SPINLOCK authlock; /**< Generic Authorization spinlock */
DCBSTATS stats; /**< DCB related statistics */ DCBSTATS stats; /**< DCB related statistics */
@ -187,6 +216,11 @@ typedef struct dcb {
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 */ int command; /**< Specific client command type */
SPINLOCK cb_lock; /**< The lock for the callbacks linked list */
DCB_CALLBACK *callbacks; /**< The list of callbacks for the DCB */
unsigned int high_water; /**< High water mark */
unsigned int low_water; /**< Low water mark */
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
skygw_chk_t dcb_chk_tail; skygw_chk_t dcb_chk_tail;
#endif #endif
@ -205,6 +239,11 @@ int fail_accept_errno;
#define DCB_SESSION(x) (x)->session #define DCB_SESSION(x) (x)->session
#define DCB_PROTOCOL(x, type) (type *)((x)->protocol) #define DCB_PROTOCOL(x, type) (type *)((x)->protocol)
#define DCB_ISZOMBIE(x) ((x)->state == DCB_STATE_ZOMBIE) #define DCB_ISZOMBIE(x) ((x)->state == DCB_STATE_ZOMBIE)
#define DCB_WRITEQLEN(x) (x)->writeqlen
#define DCB_SET_LOW_WATER(x, lo) (x)->low_water = (lo);
#define DCB_SET_HIGH_WATER(x, hi) (x)->low_water = (hi);
#define DCB_BELOW_LOW_WATER(x) ((x)->low_water && (x)->writeqlen < (x)->low_water)
#define DCB_ABOVE_HIGH_WATER(x) ((x)->high_water && (x)->writeqlen > (x)->high_water)
DCB *dcb_get_zombies(void); DCB *dcb_get_zombies(void);
int gw_write( int gw_write(
@ -231,6 +270,11 @@ void dcb_printf(DCB *, const char *, ...); /* DCB version of printf */
int dcb_isclient(DCB *); /* the DCB is the client of the session */ int dcb_isclient(DCB *); /* the DCB is the client of the session */
void dcb_hashtable_stats(DCB *, void *); /**< Print statisitics */ void dcb_hashtable_stats(DCB *, void *); /**< Print statisitics */
void dcb_add_to_zombieslist(DCB* dcb); void dcb_add_to_zombieslist(DCB* dcb);
int dcb_add_callback(DCB *, DCB_REASON, int (*)(struct dcb *, DCB_REASON, void *),
void *);
int dcb_remove_callback(DCB *, DCB_REASON, int (*)(struct dcb *, DCB_REASON),
void *);
int dcb_isvalid(DCB *); /* Check the DCB is in the linked list */
bool dcb_set_state( bool dcb_set_state(
DCB* dcb, DCB* dcb,

2
server/include/monitor.h
View File

@ -29,6 +29,7 @@
* Date Who Description * Date Who Description
* 07/07/13 Mark Riddoch Initial implementation * 07/07/13 Mark Riddoch Initial implementation
* 25/07/13 Mark Riddoch Addition of diagnotics * 25/07/13 Mark Riddoch Addition of diagnotics
* 23/05/14 Mark Riddoch Addition of routine to find monitors by name
* *
* @endverbatim * @endverbatim
*/ */
@ -79,6 +80,7 @@ typedef struct monitor {
extern MONITOR *monitor_alloc(char *, char *); extern MONITOR *monitor_alloc(char *, char *);
extern void monitor_free(MONITOR *); extern void monitor_free(MONITOR *);
extern MONITOR *monitor_find(char *);
extern void monitorAddServer(MONITOR *, SERVER *); extern void monitorAddServer(MONITOR *, SERVER *);
extern void monitorAddUser(MONITOR *, char *, char *); extern void monitorAddUser(MONITOR *, char *, char *);
extern void monitorStop(MONITOR *); extern void monitorStop(MONITOR *);

5
server/include/server.h
View File

@ -31,6 +31,8 @@
* 14/06/13 Mark Riddoch Initial implementation * 14/06/13 Mark Riddoch Initial implementation
* 21/06/13 Mark Riddoch Addition of server status flags * 21/06/13 Mark Riddoch Addition of server status flags
* 22/07/13 Mark Riddoch Addition of JOINED status for Galera * 22/07/13 Mark Riddoch Addition of JOINED status for Galera
* 18/05/14 Mark Riddoch Addition of unique_name field
* 20/05/14 Massimiliano Pinto Addition of server_string field
* *
* @endverbatim * @endverbatim
*/ */
@ -51,6 +53,7 @@ typedef struct {
* between the gateway and the server. * between the gateway and the server.
*/ */
typedef struct server { typedef struct server {
char *unique_name; /**< Unique name for the server */
char *name; /**< Server name/IP address*/ char *name; /**< Server name/IP address*/
unsigned short port; /**< Port to listen on */ unsigned short port; /**< Port to listen on */
char *protocol; /**< Protocol module to use */ char *protocol; /**< Protocol module to use */
@ -60,6 +63,7 @@ typedef struct server {
SERVER_STATS stats; /**< The server statistics */ SERVER_STATS stats; /**< The server statistics */
struct server *next; /**< Next server */ struct server *next; /**< Next server */
struct server *nextdb; /**< Next server in list attached to a service */ struct server *nextdb; /**< Next server in list attached to a service */
char *server_string; /**< Server version string, i.e. MySQL server version */
} SERVER; } SERVER;
/** /**
@ -103,6 +107,7 @@ typedef struct server {
extern SERVER *server_alloc(char *, char *, unsigned short); extern SERVER *server_alloc(char *, char *, unsigned short);
extern int server_free(SERVER *); extern int server_free(SERVER *);
extern SERVER *server_find_by_unique_name(char *);
extern SERVER *server_find(char *, unsigned short); extern SERVER *server_find(char *, unsigned short);
extern void printServer(SERVER *); extern void printServer(SERVER *);
extern void printAllServers(); extern void printAllServers();

11
server/include/service.h
View File

@ -22,6 +22,7 @@
#include <spinlock.h> #include <spinlock.h>
#include <dcb.h> #include <dcb.h>
#include <server.h> #include <server.h>
#include "config.h"
/** /**
* @file service.h * @file service.h
@ -116,6 +117,8 @@ typedef struct service {
SERVICE_STATS stats; /**< The service statistics */ SERVICE_STATS stats; /**< The service statistics */
struct users *users; /**< The user data for this service */ struct users *users; /**< The user data for this service */
int enable_root; /**< Allow root user access */ int enable_root; /**< Allow root user access */
CONFIG_PARAMETER* svc_config_param; /*< list of config params and values */
int svc_config_version; /*< Version number of configuration */
SPINLOCK SPINLOCK
users_table_spin; /**< The spinlock for users data refresh */ users_table_spin; /**< The spinlock for users data refresh */
SERVICE_REFRESH_RATE SERVICE_REFRESH_RATE
@ -123,12 +126,15 @@ typedef struct service {
struct service *next; /**< The next service in the linked list */ struct service *next; /**< The next service in the linked list */
} SERVICE; } SERVICE;
typedef enum count_spec_t {COUNT_ATLEAST=0, COUNT_EXACT, COUNT_ATMOST} count_spec_t;
#define SERVICE_STATE_ALLOC 1 /**< The service has been allocated */ #define SERVICE_STATE_ALLOC 1 /**< The service has been allocated */
#define SERVICE_STATE_STARTED 2 /**< The service has been started */ #define SERVICE_STATE_STARTED 2 /**< The service has been started */
extern SERVICE *service_alloc(char *, char *); extern SERVICE *service_alloc(char *, char *);
extern int service_free(SERVICE *); extern int service_free(SERVICE *);
extern SERVICE *service_find(char *); extern SERVICE *service_find(char *);
extern int service_isvalid(SERVICE *);
extern int serviceAddProtocol(SERVICE *, char *, char *, unsigned short); extern int serviceAddProtocol(SERVICE *, char *, char *, unsigned short);
extern int serviceHasProtocol(SERVICE *, char *, unsigned short); extern int serviceHasProtocol(SERVICE *, char *, unsigned short);
extern void serviceAddBackend(SERVICE *, SERVER *); extern void serviceAddBackend(SERVICE *, SERVER *);
@ -148,5 +154,10 @@ extern int service_refresh_users(SERVICE *);
extern void printService(SERVICE *); extern void printService(SERVICE *);
extern void printAllServices(); extern void printAllServices();
extern void dprintAllServices(DCB *); extern void dprintAllServices(DCB *);
bool service_set_slave_conn_limit (
SERVICE* service,
CONFIG_PARAMETER* param,
char* valstr,
count_spec_t count_spec);
extern void dprintService(DCB *, SERVICE *); extern void dprintService(DCB *, SERVICE *);
#endif #endif

3
server/include/session.h
View File

@ -53,6 +53,7 @@ typedef enum {
SESSION_STATE_ALLOC, /*< for all sessions */ SESSION_STATE_ALLOC, /*< for all sessions */
SESSION_STATE_READY, /*< for router session */ SESSION_STATE_READY, /*< for router session */
SESSION_STATE_ROUTER_READY, /*< for router session */ SESSION_STATE_ROUTER_READY, /*< for router session */
SESSION_STATE_STOPPING, /*< router is being closed */
SESSION_STATE_LISTENER, /*< for listener session */ SESSION_STATE_LISTENER, /*< for listener session */
SESSION_STATE_LISTENER_STOPPED, /*< for listener session */ SESSION_STATE_LISTENER_STOPPED, /*< for listener session */
SESSION_STATE_FREE /*< for all sessions */ SESSION_STATE_FREE /*< for all sessions */
@ -87,10 +88,12 @@ typedef struct session {
SESSION *session_alloc(struct service *, struct dcb *); SESSION *session_alloc(struct service *, struct dcb *);
bool session_free(SESSION *); bool session_free(SESSION *);
int session_isvalid(SESSION *);
void printAllSessions(); void printAllSessions();
void printSession(SESSION *); void printSession(SESSION *);
void dprintAllSessions(struct dcb *); void dprintAllSessions(struct dcb *);
void dprintSession(struct dcb *, SESSION *); void dprintSession(struct dcb *, SESSION *);
char *session_state(int); char *session_state(int);
bool session_link_dcb(SESSION *, struct dcb *); bool session_link_dcb(SESSION *, struct dcb *);
SESSION* get_session_by_router_ses(void* rses);
#endif #endif

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

@ -41,6 +41,7 @@ struct cli_session;
typedef struct cli_instance { typedef struct cli_instance {
SPINLOCK lock; /*< The instance spinlock */ SPINLOCK lock; /*< The instance spinlock */
SERVICE *service; /*< The debug cli service */ SERVICE *service; /*< The debug cli service */
int mode; /*< CLI interface mode */
struct cli_session struct cli_session
*sessions; /*< Linked list of sessions within this instance */ *sessions; /*< Linked list of sessions within this instance */
struct cli_instance struct cli_instance
@ -53,8 +54,13 @@ typedef struct cli_instance {
*/ */
typedef struct cli_session { typedef struct cli_session {
char cmdbuf[80]; /*< The command buffer used to build up user commands */ char cmdbuf[80]; /*< The command buffer used to build up user commands */
int mode; /*< The CLI Mode for this session */
SESSION *session; /*< The gateway session */ SESSION *session; /*< The gateway session */
struct cli_session struct cli_session
*next; /*< The next pointer for the list of sessions */ *next; /*< The next pointer for the list of sessions */
} CLI_SESSION; } CLI_SESSION;
/* Command line interface modes */
#define CLIM_USER 1
#define CLIM_DEVELOPER 2
#endif #endif

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

@ -31,15 +31,13 @@
#include <dcb.h> #include <dcb.h>
/** typedef enum backend_type_t {
* Internal structure used to define the set of backend servers we are routing BE_UNDEFINED=-1,
* connections to. This provides the storage for routing module specific data BE_MASTER,
* that is required for each of the backend servers. BE_JOINED = BE_MASTER,
*/ BE_SLAVE,
typedef struct backend { BE_COUNT
SERVER* backend_server; /*< The server itself */ } backend_type_t;
int backend_conn_count; /*< Number of connections to the server */
} BACKEND;
typedef struct rses_property_st rses_property_t; typedef struct rses_property_st rses_property_t;
typedef struct router_client_session ROUTER_CLIENT_SES; typedef struct router_client_session ROUTER_CLIENT_SES;
@ -52,13 +50,33 @@ typedef enum rses_property_type_t {
RSES_PROP_TYPE_COUNT=RSES_PROP_TYPE_LAST+1 RSES_PROP_TYPE_COUNT=RSES_PROP_TYPE_LAST+1
} rses_property_type_t; } rses_property_type_t;
typedef enum backend_type_t {
BE_UNDEFINED=-1,
BE_MASTER, /**
BE_JOINED = BE_MASTER, * This criteria is used when backends are chosen for a router session's use.
BE_SLAVE, * Backend servers are sorted to ascending order according to the criteria
BE_COUNT * and top N are chosen.
} backend_type_t; */
typedef enum select_criteria {
UNDEFINED_CRITERIA=0,
LEAST_GLOBAL_CONNECTIONS, /*< all connections established by MaxScale */
DEFAULT_CRITERIA=LEAST_GLOBAL_CONNECTIONS,
LEAST_ROUTER_CONNECTIONS, /*< connections established by this router */
LEAST_BEHIND_MASTER,
LAST_CRITERIA /*< not used except for an index */
} select_criteria_t;
/** default values for rwsplit configuration parameters */
#define CONFIG_MAX_SLAVE_CONN 1
#define GET_SELECT_CRITERIA(s) \
(strncmp(s,"LEAST_GLOBAL_CONNECTIONS", strlen("LEAST_GLOBAL_CONNECTIONS")) == 0 ? \
LEAST_GLOBAL_CONNECTIONS : ( \
strncmp(s,"LEAST_BEHIND_MASTER", strlen("LEAST_BEHIND_MASTER")) == 0 ? \
LEAST_BEHIND_MASTER : ( \
strncmp(s,"LEAST_ROUTER_CONNECTIONS", strlen("LEAST_ROUTER_CONNECTIONS")) == 0 ? \
LEAST_ROUTER_CONNECTIONS : UNDEFINED_CRITERIA)))
/** /**
* Session variable command * Session variable command
@ -98,13 +116,63 @@ struct rses_property_st {
}; };
typedef struct sescmd_cursor_st { typedef struct sescmd_cursor_st {
#if defined(SS_DEBUG)
skygw_chk_t scmd_cur_chk_top;
#endif
ROUTER_CLIENT_SES* scmd_cur_rses; /*< pointer to owning router session */ ROUTER_CLIENT_SES* scmd_cur_rses; /*< pointer to owning router session */
rses_property_t** scmd_cur_ptr_property; /*< address of pointer to owner property */ rses_property_t** scmd_cur_ptr_property; /*< address of pointer to owner property */
mysql_sescmd_t* scmd_cur_cmd; /*< pointer to current session command */ mysql_sescmd_t* scmd_cur_cmd; /*< pointer to current session command */
bool scmd_cur_active; /*< true if command is being executed */ bool scmd_cur_active; /*< true if command is being executed */
backend_type_t scmd_cur_be_type; /*< BE_MASTER or BE_SLAVE */ #if defined(SS_DEBUG)
skygw_chk_t scmd_cur_chk_tail;
#endif
} sescmd_cursor_t; } sescmd_cursor_t;
/**
* Internal structure used to define the set of backend servers we are routing
* connections to. This provides the storage for routing module specific data
* that is required for each of the backend servers.
*
* Owned by router_instance, referenced by each routing session.
*/
typedef struct backend_st {
#if defined(SS_DEBUG)
skygw_chk_t be_chk_top;
#endif
SERVER* backend_server; /*< The server itself */
int backend_conn_count; /*< Number of connections to the server */
bool be_valid; /*< valid when belongs to the router's configuration */
#if defined(SS_DEBUG)
skygw_chk_t be_chk_tail;
#endif
} BACKEND;
/**
* Reference to BACKEND.
*
* Owned by router client session.
*/
typedef struct backend_ref_st {
#if defined(SS_DEBUG)
skygw_chk_t bref_chk_top;
#endif
BACKEND* bref_backend;
DCB* bref_dcb;
sescmd_cursor_t bref_sescmd_cur;
#if defined(SS_DEBUG)
skygw_chk_t bref_chk_tail;
#endif
} backend_ref_t;
typedef struct rwsplit_config_st {
int rw_max_slave_conn_percent;
int rw_max_slave_conn_count;
select_criteria_t rw_slave_select_criteria;
} rwsplit_config_t;
/** /**
* The client session structure used within this router. * The client session structure used within this router.
*/ */
@ -113,17 +181,18 @@ struct router_client_session {
skygw_chk_t rses_chk_top; skygw_chk_t rses_chk_top;
#endif #endif
SPINLOCK rses_lock; /*< protects rses_deleted */ SPINLOCK rses_lock; /*< protects rses_deleted */
int rses_versno; /*< even = no active update, else odd */ int rses_versno; /*< even = no active update, else odd. not used 4/14 */
bool rses_closed; /*< true when closeSession is called */ bool rses_closed; /*< true when closeSession is called */
/** Properties listed by their type */ /** Properties listed by their type */
rses_property_t* rses_properties[RSES_PROP_TYPE_COUNT]; rses_property_t* rses_properties[RSES_PROP_TYPE_COUNT];
BACKEND* rses_backend[BE_COUNT];/*< Backends used by client session */ backend_ref_t* rses_master_ref;
DCB* rses_dcb[BE_COUNT]; backend_ref_t* rses_backend_ref; /*< Pointer to backend reference array */
/*< cursor is pointer and status variable to current session command */ rwsplit_config_t rses_config; /*< copied config info from router instance */
sescmd_cursor_t rses_cursor[BE_COUNT]; int rses_nbackends;
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 */
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;
@ -151,6 +220,8 @@ typedef struct router_instance {
SPINLOCK lock; /*< Lock for the instance data */ SPINLOCK lock; /*< Lock for the instance data */
BACKEND** servers; /*< Backend servers */ BACKEND** servers; /*< Backend servers */
BACKEND* master; /*< NULL or pointer */ BACKEND* master; /*< NULL or pointer */
rwsplit_config_t rwsplit_config; /*< expanded config info from SERVICE */
int rwsplit_version;/*< version number for router's config */
unsigned int bitmask; /*< Bitmask to apply to server->status */ unsigned int bitmask; /*< Bitmask to apply to server->status */
unsigned int bitvalue; /*< Required value of server->status */ unsigned int bitvalue; /*< Required value of server->status */
ROUTER_STATS stats; /*< Statistics for this router */ ROUTER_STATS stats; /*< Statistics for this router */

2
server/modules/monitor/galera_mon.c
View File

@ -360,6 +360,6 @@ MONITOR_SERVERS *ptr;
monitorDatabase(ptr, handle->defaultUser, handle->defaultPasswd); monitorDatabase(ptr, handle->defaultUser, handle->defaultPasswd);
ptr = ptr->next; ptr = ptr->next;
} }
thread_millisleep(10000); thread_millisleep(MONITOR_INTERVAL);
} }
} }

41
server/modules/monitor/mysql_mon.c
View File

@ -28,6 +28,8 @@
* status * status
* 25/07/13 Mark Riddoch Addition of decrypt for passwords and * 25/07/13 Mark Riddoch Addition of decrypt for passwords and
* diagnostic interface * diagnostic interface
* 20/05/14 Massimiliano Pinto Addition of support for MariadDB multimaster replication setup.
* New server field version_string is updated.
* *
* @endverbatim * @endverbatim
*/ */
@ -49,7 +51,7 @@ extern int lm_enabled_logfiles_bitmask;
static void monitorMain(void *); static void monitorMain(void *);
static char *version_str = "V1.0.0"; static char *version_str = "V1.1.0";
static void *startMonitor(void *); static void *startMonitor(void *);
static void stopMonitor(void *); static void stopMonitor(void *);
@ -290,6 +292,8 @@ MYSQL_RES *result;
int num_fields; int num_fields;
int ismaster = 0, isslave = 0; int ismaster = 0, isslave = 0;
char *uname = defaultUser, *passwd = defaultPasswd; char *uname = defaultUser, *passwd = defaultPasswd;
unsigned long int server_version = 0;
char *server_string;
if (database->server->monuser != NULL) if (database->server->monuser != NULL)
{ {
@ -321,6 +325,15 @@ char *uname = defaultUser, *passwd = defaultPasswd;
/* If we get this far then we have a working connection */ /* If we get this far then we have a working connection */
server_set_status(database->server, SERVER_RUNNING); server_set_status(database->server, SERVER_RUNNING);
/* get server version from current server */
server_version = mysql_get_server_version(database->con);
/* get server version string */
server_string = (char *)mysql_get_server_info(database->con);
if (server_string) {
database->server->server_string = strdup(server_string);
}
/* Check SHOW SLAVE HOSTS - if we get rows then we are a master */ /* Check SHOW SLAVE HOSTS - if we get rows then we are a master */
if (mysql_query(database->con, "SHOW SLAVE HOSTS")) if (mysql_query(database->con, "SHOW SLAVE HOSTS"))
{ {
@ -342,6 +355,31 @@ char *uname = defaultUser, *passwd = defaultPasswd;
/* Check if the Slave_SQL_Running and Slave_IO_Running status is /* Check if the Slave_SQL_Running and Slave_IO_Running status is
* set to Yes * set to Yes
*/ */
/* Check first for MariaDB 10.x.x and get status for multimaster replication */
if (server_version >= 100000) {
if (mysql_query(database->con, "SHOW ALL SLAVES STATUS") == 0
&& (result = mysql_store_result(database->con)) != NULL)
{
int i = 0;
num_fields = mysql_num_fields(result);
while ((row = mysql_fetch_row(result)))
{
if (strncmp(row[12], "Yes", 3) == 0
&& strncmp(row[13], "Yes", 3) == 0) {
isslave += 1;
}
i++;
}
mysql_free_result(result);
if (isslave == i)
isslave = 1;
else
isslave = 0;
}
} else {
if (mysql_query(database->con, "SHOW SLAVE STATUS") == 0 if (mysql_query(database->con, "SHOW SLAVE STATUS") == 0
&& (result = mysql_store_result(database->con)) != NULL) && (result = mysql_store_result(database->con)) != NULL)
{ {
@ -354,6 +392,7 @@ char *uname = defaultUser, *passwd = defaultPasswd;
} }
mysql_free_result(result); mysql_free_result(result);
} }
}
if (ismaster) if (ismaster)
{ {

3
server/modules/monitor/mysqlmon.h
View File

@ -29,6 +29,7 @@
* *
* Date Who Description * Date Who Description
* 08/07/13 Mark Riddoch Initial implementation * 08/07/13 Mark Riddoch Initial implementation
* 26/05/14 Massimiliano Pinto Default values for MONITOR_INTERVAL
* *
* @endverbatim * @endverbatim
*/ */
@ -61,4 +62,6 @@ typedef struct {
#define MONITOR_STOPPING 2 #define MONITOR_STOPPING 2
#define MONITOR_STOPPED 3 #define MONITOR_STOPPED 3
#define MONITOR_INTERVAL 10000 // in milliseconds
#endif #endif

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

@ -311,12 +311,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->client->service); service_refresh_users(dcb->session->client->service);
while (session->state != SESSION_STATE_ROUTER_READY) while (session->state != SESSION_STATE_ROUTER_READY &&
session->state != SESSION_STATE_STOPPING)
{ {
ss_dassert( ss_dassert(
session->state == SESSION_STATE_READY || session->state == SESSION_STATE_READY ||
session->state == session->state ==
SESSION_STATE_ROUTER_READY); SESSION_STATE_ROUTER_READY ||
session->state == SESSION_STATE_STOPPING);
/** /**
* Session shouldn't be NULL at this point * Session shouldn't be NULL at this point
* anymore. Just checking.. * anymore. Just checking..
@ -328,6 +330,15 @@ static int gw_read_backend_event(DCB *dcb) {
} }
usleep(1); usleep(1);
} }
if (session->state == SESSION_STATE_STOPPING)
{
goto return_with_lock;
}
spinlock_acquire(&session->ses_lock);
session->state = SESSION_STATE_STOPPING;
spinlock_release(&session->ses_lock);
/** /**
* rsession shouldn't be NULL since session * rsession shouldn't be NULL since session
* state indicates that it was initialized * state indicates that it was initialized
@ -362,8 +373,7 @@ static int gw_read_backend_event(DCB *dcb) {
/* check the delay queue and flush the data */ /* check the delay queue and flush the data */
if (dcb->delayq) if (dcb->delayq)
{ {
backend_write_delayqueue(dcb); rc = backend_write_delayqueue(dcb);
rc = 1;
goto return_with_lock; goto return_with_lock;
} }
} }
@ -419,8 +429,6 @@ static int gw_read_backend_event(DCB *dcb) {
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);
@ -433,6 +441,10 @@ static int gw_read_backend_event(DCB *dcb) {
rc = 1; rc = 1;
} }
goto return_rc; goto return_rc;
} else if (dcb->session->client->dcb_role == DCB_ROLE_INTERNAL) {
router->clientReply(router_instance, rsession, writebuf, dcb);
rc = 1;
}
} }
} }
@ -572,9 +584,7 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
snprintf(str, len+1, "%s", startpoint); snprintf(str, len+1, "%s", startpoint);
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"Error : Routing query \"%s\" failed due to " "Error : Authentication to backend failed.")));
"authentication failure.",
str)));
/** Consume query buffer */ /** Consume query buffer */
while ((queue = gwbuf_consume( while ((queue = gwbuf_consume(
queue, queue,
@ -672,6 +682,10 @@ static int gw_error_backend_event(DCB *dcb) {
if (session->state == SESSION_STATE_ROUTER_READY) if (session->state == SESSION_STATE_ROUTER_READY)
{ {
spinlock_acquire(&session->ses_lock);
session->state = SESSION_STATE_STOPPING;
spinlock_release(&session->ses_lock);
rsession = session->router_session; rsession = session->router_session;
/*< /*<
* rsession should never be NULL here. * rsession should never be NULL here.
@ -852,34 +866,36 @@ static int backend_write_delayqueue(DCB *dcb)
spinlock_acquire(&dcb->delayqlock); spinlock_acquire(&dcb->delayqlock);
if (dcb->delayq == NULL)
{
spinlock_release(&dcb->delayqlock);
rc = 1;
}
else
{
localq = dcb->delayq; localq = dcb->delayq;
dcb->delayq = NULL; dcb->delayq = NULL;
spinlock_release(&dcb->delayqlock); spinlock_release(&dcb->delayqlock);
rc = dcb_write(dcb, localq); rc = dcb_write(dcb, localq);
}
if (rc == 0) { if (rc == 0) {
/*< vraa : errorHandle */
/**
* This error can be muted because it is often due
* unexpected dcb state which means that concurrent thread
* already wrote the queue and closed dcb.
*/
#if 0
LOGIF(LE, (skygw_log_write_flush( LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"%lu [backend_write_delayqueue] Some error occurred in " "Error : failed to write buffered data to back-end "
"backend.", "server. Buffer was empty of back-end was disconnected "
pthread_self()))); "during operation.")));
#endif
mysql_send_custom_error( mysql_send_custom_error(
dcb->session->client, dcb->session->client,
1, 1,
0, 0,
"Unable to write to backend server. Connection was " "Failed to write buffered data to back-end server. "
"closed."); "Buffer was empty or back-end was disconnected during "
"operation.");
dcb_close(dcb); dcb_close(dcb);
} }
return rc; return rc;
} }

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

@ -552,7 +552,7 @@ int gw_read_client_event(DCB* dcb) {
goto return_rc; goto return_rc;
} }
// close client socket and the sessioA too // close client socket and the session too
dcb->func.close(dcb); dcb->func.close(dcb);
} else { } else {
// do nothing if reading 1 byte // do nothing if reading 1 byte
@ -723,7 +723,7 @@ int gw_read_client_event(DCB* dcb) {
dcb, dcb,
1, 1,
0, 0,
"Query routing failed. Connection to " "Can't route query. Connection to "
"backend lost"); "backend lost");
protocol->state = MYSQL_IDLE; protocol->state = MYSQL_IDLE;
} }
@ -772,12 +772,8 @@ int gw_read_client_event(DCB* dcb) {
"backend. Close client dcb %p", "backend. Close client dcb %p",
pthread_self(), pthread_self(),
dcb))); dcb)));
/** close client connection, closes router session too */
/** close client connection */ rc = dcb->func.close(dcb);
(dcb->func).close(dcb);
/** close backends connection */
router->closeSession(router_instance, rsession);
rc = 1;
} }
else else
{ {
@ -1263,37 +1259,16 @@ return_rc:
return rc; return rc;
} }
static int gw_error_client_event(DCB *dcb) { static int gw_error_client_event(
SESSION* session; DCB* dcb)
ROUTER_OBJECT* router;
void* router_instance;
void* rsession;
#if defined(SS_DEBUG)
MySQLProtocol* protocol = (MySQLProtocol *)dcb->protocol;
if (dcb->state == DCB_STATE_POLLING ||
dcb->state == DCB_STATE_NOPOLLING ||
dcb->state == DCB_STATE_ZOMBIE)
{ {
CHK_PROTOCOL(protocol); int rc;
}
#endif
session = dcb->session; CHK_DCB(dcb);
/** rc = dcb->func.close(dcb);
* session may be NULL if session_alloc failed.
* In that case router session was not created. return rc;
*/
if (session != NULL) {
CHK_SESSION(session);
router = session->service->router;
router_instance = session->service->router_instance;
rsession = session->router_session;
router->closeSession(router_instance, rsession);
}
dcb_close(dcb);
return 1;
} }
static int static int
@ -1320,6 +1295,10 @@ gw_client_close(DCB *dcb)
*/ */
if (session != NULL) { if (session != NULL) {
CHK_SESSION(session); CHK_SESSION(session);
spinlock_acquire(&session->ses_lock);
session->state = SESSION_STATE_STOPPING;
spinlock_release(&session->ses_lock);
router = session->service->router; router = session->service->router;
router_instance = session->service->router_instance; router_instance = session->service->router_instance;
rsession = session->router_session; rsession = session->router_session;
@ -1342,42 +1321,11 @@ gw_client_close(DCB *dcb)
static int static int
gw_client_hangup_event(DCB *dcb) gw_client_hangup_event(DCB *dcb)
{ {
SESSION* session; int rc;
ROUTER_OBJECT* router;
void* router_instance;
void* rsession;
int rc = 1;
#if defined(SS_DEBUG)
MySQLProtocol* protocol = (MySQLProtocol *)dcb->protocol;
if (dcb->state == DCB_STATE_POLLING ||
dcb->state == DCB_STATE_NOPOLLING ||
dcb->state == DCB_STATE_ZOMBIE)
{
CHK_PROTOCOL(protocol);
}
#endif
CHK_DCB(dcb); CHK_DCB(dcb);
rc = dcb->func.close(dcb);
if (dcb->state != DCB_STATE_POLLING) {
goto return_rc;
}
session = dcb->session;
/**
* session may be NULL if session_alloc failed.
* In that case router session was not created.
*/
if (session != NULL) {
CHK_SESSION(session);
router = session->service->router;
router_instance = session->service->router_instance;
rsession = session->router_session;
router->closeSession(router_instance, rsession);
}
dcb_close(dcb);
return_rc:
return rc; return rc;
} }

29
server/modules/routing/debugcli.c
View File

@ -45,7 +45,7 @@
extern int lm_enabled_logfiles_bitmask; extern int lm_enabled_logfiles_bitmask;
static char *version_str = "V1.0.1"; static char *version_str = "V1.1.0";
/* The router entry points */ /* The router entry points */
static ROUTER *createInstance(SERVICE *service, char **options); static ROUTER *createInstance(SERVICE *service, char **options);
@ -127,6 +127,7 @@ static ROUTER *
createInstance(SERVICE *service, char **options) createInstance(SERVICE *service, char **options)
{ {
CLI_INSTANCE *inst; CLI_INSTANCE *inst;
int i;
if ((inst = malloc(sizeof(CLI_INSTANCE))) == NULL) if ((inst = malloc(sizeof(CLI_INSTANCE))) == NULL)
return NULL; return NULL;
@ -134,7 +135,29 @@ CLI_INSTANCE *inst;
inst->service = service; inst->service = service;
spinlock_init(&inst->lock); spinlock_init(&inst->lock);
inst->sessions = NULL; inst->sessions = NULL;
inst->mode = CLIM_USER;
if (options)
{
for (i = 0; options[i]; i++)
{
if (!strcasecmp(options[i], "developer"))
{
inst->mode = CLIM_DEVELOPER;
}
else if (!strcasecmp(options[i], "user"))
{
inst->mode = CLIM_USER;
}
else
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Unknown option for CLI '%s'\n",
options[i])));
}
}
}
/* /*
* We have completed the creation of the instance data, so now * We have completed the creation of the instance data, so now
@ -176,11 +199,15 @@ CLI_SESSION *client;
spinlock_release(&inst->lock); spinlock_release(&inst->lock);
session->state = SESSION_STATE_READY; session->state = SESSION_STATE_READY;
client->mode = inst->mode;
dcb_printf(session->client, "Welcome the SkySQL MaxScale Debug Interface (%s).\n", dcb_printf(session->client, "Welcome the SkySQL MaxScale Debug Interface (%s).\n",
version_str); version_str);
if (client->mode == CLIM_DEVELOPER)
{
dcb_printf(session->client, "WARNING: This interface is meant for developer usage,\n"); dcb_printf(session->client, "WARNING: This interface is meant for developer usage,\n");
dcb_printf(session->client, "passing incorrect addresses to commands can endanger your MaxScale server.\n\n"); dcb_printf(session->client, "passing incorrect addresses to commands can endanger your MaxScale server.\n\n");
}
dcb_printf(session->client, "Type help for a list of available commands.\n\n"); dcb_printf(session->client, "Type help for a list of available commands.\n\n");
return (void *)client; return (void *)client;

277
server/modules/routing/debugcmd.c
View File

@ -36,7 +36,10 @@
* Date Who Description * Date Who Description
* 20/06/13 Mark Riddoch Initial implementation * 20/06/13 Mark Riddoch Initial implementation
* 17/07/13 Mark Riddoch Additional commands * 17/07/13 Mark Riddoch Additional commands
* 09/08/2013 Massimiliano Pinto Addes enable/disable commands (now only for log) * 09/08/2013 Massimiliano Pinto Added enable/disable commands (now only for log)
* 20/05/14 Mark Riddoch Added ability to give server and service names rather
* than simply addresses
* 23/05/14 Mark Riddoch Added support for developer and user modes
* *
* @endverbatim * @endverbatim
*/ */
@ -69,6 +72,12 @@
#define ARG_TYPE_ADDRESS 1 #define ARG_TYPE_ADDRESS 1
#define ARG_TYPE_STRING 2 #define ARG_TYPE_STRING 2
#define ARG_TYPE_SERVICE 3 #define ARG_TYPE_SERVICE 3
#define ARG_TYPE_SERVER 4
#define ARG_TYPE_DBUSERS 5
#define ARG_TYPE_SESSION 6
#define ARG_TYPE_DCB 7
#define ARG_TYPE_MONITOR 8
/** /**
* The subcommand structure * The subcommand structure
* *
@ -79,6 +88,7 @@ struct subcommand {
int n_args; int n_args;
void (*fn)(); void (*fn)();
char *help; char *help;
char *devhelp;
int arg_types[3]; int arg_types[3];
}; };
@ -87,33 +97,59 @@ static void telnetdShowUsers(DCB *);
* The subcommands of the show command * The subcommands of the show command
*/ */
struct subcommand showoptions[] = { struct subcommand showoptions[] = {
{ "dcbs", 0, dprintAllDCBs, "Show all descriptor control blocks (network connections)", { "dcbs", 0, dprintAllDCBs,
"Show all descriptor control blocks (network connections)",
"Show all descriptor control blocks (network connections)",
{0, 0, 0} }, {0, 0, 0} },
{ "dcb", 1, dprintDCB, "Show a single descriptor control block e.g. show dcb 0x493340", { "dcb", 1, dprintDCB,
{ARG_TYPE_ADDRESS, 0, 0} }, "Show a single descriptor control block e.g. show dcb 0x493340",
{ "dbusers", 1, dcb_usersPrint, "Show statistics and user names for a service's user table.\n\t\tExample : show dbusers <ptr of 'User's data' from services list>", "Show a single descriptor control block e.g. show dcb 0x493340",
{ARG_TYPE_ADDRESS, 0, 0} }, {ARG_TYPE_DCB, 0, 0} },
{ "epoll", 0, dprintPollStats, "Show the poll statistics", { "dbusers", 1, dcb_usersPrint,
"Show statistics and user names for a service's user table.\n\t\tExample : show dbusers <service name>",
"Show statistics and user names for a service's user table.\n\t\tExample : show dbusers <ptr of 'User's data' from services list>|<service name>",
{ARG_TYPE_DBUSERS, 0, 0} },
{ "epoll", 0, dprintPollStats,
"Show the poll statistics",
"Show the poll statistics",
{0, 0, 0} }, {0, 0, 0} },
{ "modules", 0, dprintAllModules, "Show all currently loaded modules", { "modules", 0, dprintAllModules,
"Show all currently loaded modules",
"Show all currently loaded modules",
{0, 0, 0} }, {0, 0, 0} },
{ "monitors", 0, monitorShowAll, "Show the monitors that are configured", { "monitors", 0, monitorShowAll,
"Show the monitors that are configured",
"Show the monitors that are configured",
{0, 0, 0} }, {0, 0, 0} },
{ "server", 1, dprintServer, "Show details for a server, e.g. show server 0x485390", { "server", 1, dprintServer,
{ARG_TYPE_ADDRESS, 0, 0} }, "Show details for a named server, e.g. show server dbnode1",
{ "servers", 0, dprintAllServers, "Show all configured servers", "Show details for a server, e.g. show server 0x485390. The address may also be repalced with the server name from the configuration file",
{ARG_TYPE_SERVER, 0, 0} },
{ "servers", 0, dprintAllServers,
"Show all configured servers",
"Show all configured servers",
{0, 0, 0} }, {0, 0, 0} },
{ "services", 0, dprintAllServices, "Show all configured services in MaxScale", { "services", 0, dprintAllServices,
"Show all configured services in MaxScale",
"Show all configured services in MaxScale",
{0, 0, 0} }, {0, 0, 0} },
{ "service", 1, dprintService, "Show single service in MaxScale", { "service", 1, dprintService,
"Show a single service in MaxScale, may be passed a service name",
"Show a single service in MaxScale, may be passed a service name or address of a service object",
{ARG_TYPE_SERVICE, 0, 0} }, {ARG_TYPE_SERVICE, 0, 0} },
{ "session", 1, dprintSession, "Show a single session in MaxScale, e.g. show session 0x284830", { "session", 1, dprintSession,
{ARG_TYPE_ADDRESS, 0, 0} }, "Show a single session in MaxScale, e.g. show session 0x284830",
{ "sessions", 0, dprintAllSessions, "Show all active sessions in MaxScale", "Show a single session in MaxScale, e.g. show session 0x284830",
{ARG_TYPE_SESSION, 0, 0} },
{ "sessions", 0, dprintAllSessions,
"Show all active sessions in MaxScale",
"Show all active sessions in MaxScale",
{0, 0, 0} }, {0, 0, 0} },
{ "users", 0, telnetdShowUsers, "Show statistics and user names for the debug interface", { "users", 0, telnetdShowUsers,
{ARG_TYPE_ADDRESS, 0, 0} }, "Show statistics and user names for the debug interface",
{ NULL, 0, NULL, NULL, "Show statistics and user names for the debug interface",
{0, 0, 0} },
{ NULL, 0, NULL, NULL, NULL,
{0, 0, 0} } {0, 0, 0} }
}; };
@ -129,7 +165,7 @@ struct subcommand shutdownoptions[] = {
0, 0,
shutdown_server, shutdown_server,
"Shutdown MaxScale", "Shutdown MaxScale",
"Shutdown MaxScale",
{0, 0, 0} {0, 0, 0}
}, },
{ {
@ -137,13 +173,15 @@ struct subcommand shutdownoptions[] = {
1, 1,
shutdown_monitor, shutdown_monitor,
"Shutdown a monitor, e.g. shutdown monitor 0x48381e0", "Shutdown a monitor, e.g. shutdown monitor 0x48381e0",
{ARG_TYPE_ADDRESS, 0, 0} "Shutdown a monitor, e.g. shutdown monitor 0x48381e0",
{ARG_TYPE_MONITOR, 0, 0}
}, },
{ {
"service", "service",
1, 1,
shutdown_service, shutdown_service,
"Shutdown a service, e.g. shutdown service 0x4838320", "Shutdown a service, e.g. shutdown service \"Sales Database\"",
"Shutdown a service, e.g. shutdown service 0x4838320 or shutdown service \"Sales Database\"",
{ARG_TYPE_SERVICE, 0, 0} {ARG_TYPE_SERVICE, 0, 0}
}, },
{ {
@ -151,6 +189,7 @@ struct subcommand shutdownoptions[] = {
0, 0,
NULL, NULL,
NULL, NULL,
NULL,
{0, 0, 0} {0, 0, 0}
} }
}; };
@ -162,11 +201,15 @@ static void restart_monitor(DCB *dcb, MONITOR *monitor);
* The subcommands of the restart command * The subcommands of the restart command
*/ */
struct subcommand restartoptions[] = { struct subcommand restartoptions[] = {
{ "monitor", 1, restart_monitor, "Restart a monitor, e.g. restart monitor 0x48181e0", { "monitor", 1, restart_monitor,
{ARG_TYPE_ADDRESS, 0, 0} }, "Restart a monitor, e.g. restart monitor 0x48181e0",
{ "service", 1, restart_service, "Restart a service, e.g. restart service name", "Restart a monitor, e.g. restart monitor 0x48181e0",
{ARG_TYPE_MONITOR, 0, 0} },
{ "service", 1, restart_service,
"Restart a service, e.g. restart service \"Test Service\"",
"Restart a service, e.g. restart service 0x4838320",
{ARG_TYPE_SERVICE, 0, 0} }, {ARG_TYPE_SERVICE, 0, 0} },
{ NULL, 0, NULL, NULL, { NULL, 0, NULL, NULL, NULL,
{0, 0, 0} } {0, 0, 0} }
}; };
@ -175,9 +218,11 @@ static void set_server(DCB *dcb, SERVER *server, char *bit);
* The subcommands of the set command * The subcommands of the set command
*/ */
struct subcommand setoptions[] = { struct subcommand setoptions[] = {
{ "server", 2, set_server, "Set the status of a server. E.g. set server 0x4838320 master", { "server", 2, set_server,
{ARG_TYPE_ADDRESS, ARG_TYPE_STRING, 0} }, "Set the status of a server. E.g. set server dbnode4 master",
{ NULL, 0, NULL, NULL, "Set the status of a server. E.g. set server 0x4838320 master",
{ARG_TYPE_SERVER, ARG_TYPE_STRING, 0} },
{ NULL, 0, NULL, NULL, NULL,
{0, 0, 0} } {0, 0, 0} }
}; };
@ -186,9 +231,11 @@ static void clear_server(DCB *dcb, SERVER *server, char *bit);
* The subcommands of the clear command * The subcommands of the clear command
*/ */
struct subcommand clearoptions[] = { struct subcommand clearoptions[] = {
{ "server", 2, clear_server, "Clear the status of a server. E.g. clear server 0x4838320 master", { "server", 2, clear_server,
{ARG_TYPE_ADDRESS, ARG_TYPE_STRING, 0} }, "Clear the status of a server. E.g. clear server dbnode2 master",
{ NULL, 0, NULL, NULL, "Clear the status of a server. E.g. clear server 0x4838320 master",
{ARG_TYPE_SERVER, ARG_TYPE_STRING, 0} },
{ NULL, 0, NULL, NULL, NULL,
{0, 0, 0} } {0, 0, 0} }
}; };
@ -199,11 +246,15 @@ static void reload_config(DCB *dcb);
* The subcommands of the reload command * The subcommands of the reload command
*/ */
struct subcommand reloadoptions[] = { struct subcommand reloadoptions[] = {
{ "config", 0, reload_config, "Reload the configuration data for MaxScale.", { "config", 0, reload_config,
{ARG_TYPE_ADDRESS, 0, 0} }, "Reload the configuration data for MaxScale.",
{ "dbusers", 1, reload_dbusers, "Reload the dbuser data for a service. E.g. reload dbusers 0x849420", "Reload the configuration data for MaxScale.",
{ARG_TYPE_ADDRESS, 0, 0} }, {0, 0, 0} },
{ NULL, 0, NULL, NULL, { "dbusers", 1, reload_dbusers,
"Reload the dbuser data for a service. E.g. reload dbusers \"splitter service\"",
"Reload the dbuser data for a service. E.g. reload dbusers 0x849420",
{ARG_TYPE_DBUSERS, 0, 0} },
{ NULL, 0, NULL, NULL, NULL,
{0, 0, 0} } {0, 0, 0} }
}; };
@ -220,6 +271,8 @@ struct subcommand enableoptions[] = {
enable_log_action, enable_log_action,
"Enable Log options for MaxScale, options trace | error | " "Enable Log options for MaxScale, options trace | error | "
"message E.g. enable log message.", "message E.g. enable log message.",
"Enable Log options for MaxScale, options trace | error | "
"message E.g. enable log message.",
{ARG_TYPE_STRING, 0, 0} {ARG_TYPE_STRING, 0, 0}
}, },
{ {
@ -227,6 +280,7 @@ struct subcommand enableoptions[] = {
0, 0,
NULL, NULL,
NULL, NULL,
NULL,
{0, 0, 0} {0, 0, 0}
} }
}; };
@ -242,6 +296,8 @@ struct subcommand disableoptions[] = {
disable_log_action, disable_log_action,
"Disable Log for MaxScale, Options: debug | trace | error | message " "Disable Log for MaxScale, Options: debug | trace | error | message "
"E.g. disable log debug", "E.g. disable log debug",
"Disable Log for MaxScale, Options: debug | trace | error | message "
"E.g. disable log debug",
{ARG_TYPE_STRING, 0, 0} {ARG_TYPE_STRING, 0, 0}
}, },
{ {
@ -249,6 +305,7 @@ struct subcommand disableoptions[] = {
0, 0,
NULL, NULL,
NULL, NULL,
NULL,
{0, 0, 0} {0, 0, 0}
} }
}; };
@ -267,6 +324,7 @@ struct subcommand failoptions[] = {
0, 0,
fail_backendfd, fail_backendfd,
"Fail backend socket for next operation.", "Fail backend socket for next operation.",
"Fail backend socket for next operation.",
{ARG_TYPE_STRING, 0, 0} {ARG_TYPE_STRING, 0, 0}
}, },
{ {
@ -274,6 +332,7 @@ struct subcommand failoptions[] = {
0, 0,
fail_clientfd, fail_clientfd,
"Fail client socket for next operation.", "Fail client socket for next operation.",
"Fail client socket for next operation.",
{ARG_TYPE_STRING, 0, 0} {ARG_TYPE_STRING, 0, 0}
}, },
{ {
@ -281,6 +340,7 @@ struct subcommand failoptions[] = {
2, 2,
fail_accept, fail_accept,
"Fail to accept next client connection.", "Fail to accept next client connection.",
"Fail to accept next client connection.",
{ARG_TYPE_STRING, ARG_TYPE_STRING, 0} {ARG_TYPE_STRING, ARG_TYPE_STRING, 0}
}, },
{ {
@ -288,6 +348,7 @@ struct subcommand failoptions[] = {
0, 0,
NULL, NULL,
NULL, NULL,
NULL,
{0, 0, 0} {0, 0, 0}
} }
}; };
@ -298,9 +359,11 @@ static void telnetdAddUser(DCB *, char *, char *);
* The subcommands of the add command * The subcommands of the add command
*/ */
struct subcommand addoptions[] = { struct subcommand addoptions[] = {
{ "user", 2, telnetdAddUser, "Add a new user for the debug interface. E.g. add user john today", { "user", 2, telnetdAddUser,
"Add a new user for the debug interface. E.g. add user john today",
"Add a new user for the debug interface. E.g. add user john today",
{ARG_TYPE_STRING, ARG_TYPE_STRING, 0} }, {ARG_TYPE_STRING, ARG_TYPE_STRING, 0} },
{ NULL, 0, NULL, NULL, { NULL, 0, NULL, NULL, NULL,
{0, 0, 0} } {0, 0, 0} }
}; };
@ -315,10 +378,11 @@ struct subcommand removeoptions[] = {
2, 2,
telnetdRemoveUser, telnetdRemoveUser,
"Remove existing maxscale user. Example : remove user john johnpwd", "Remove existing maxscale user. Example : remove user john johnpwd",
"Remove existing maxscale user. Example : remove user john johnpwd",
{ARG_TYPE_STRING, ARG_TYPE_STRING, 0} {ARG_TYPE_STRING, ARG_TYPE_STRING, 0}
}, },
{ {
NULL, 0, NULL, NULL, {0, 0, 0} NULL, 0, NULL, NULL, NULL, {0, 0, 0}
} }
}; };
@ -351,25 +415,55 @@ static struct {
* Convert a string argument to a numeric, observing prefixes * Convert a string argument to a numeric, observing prefixes
* for number bases, e.g. 0x for hex, 0 for octal * for number bases, e.g. 0x for hex, 0 for octal
* *
* @param mode The CLI mode
* @param arg The string representation of the argument * @param arg The string representation of the argument
* @param arg_type The target type for the argument * @param arg_type The target type for the argument
* @return The argument as a long integer * @return The argument as a long integer
*/ */
static unsigned long static unsigned long
convert_arg(char *arg, int arg_type) convert_arg(int mode, char *arg, int arg_type)
{ {
unsigned long rval; unsigned long rval;
SERVICE *service;
switch (arg_type) switch (arg_type)
{ {
case ARG_TYPE_SERVICE:
if ((rval = (unsigned long)strtol(arg, NULL, 0)) == 0)
rval = (unsigned long)service_find(arg);
return rval;
case ARG_TYPE_ADDRESS: case ARG_TYPE_ADDRESS:
return (unsigned long)strtol(arg, NULL, 0); return (unsigned long)strtol(arg, NULL, 0);
case ARG_TYPE_STRING: case ARG_TYPE_STRING:
return (unsigned long)arg; return (unsigned long)arg;
case ARG_TYPE_SERVICE:
if (mode == CLIM_USER || (rval = (unsigned long)strtol(arg, NULL, 0)) == 0)
rval = (unsigned long)service_find(arg);
return rval;
case ARG_TYPE_SERVER:
if (mode == CLIM_USER || (rval = (unsigned long)strtol(arg, NULL, 0)) == 0)
rval = (unsigned long)server_find_by_unique_name(arg);
return rval;
case ARG_TYPE_DBUSERS:
if (mode == CLIM_USER || (rval = (unsigned long)strtol(arg, NULL, 0)) == 0)
{
service = service_find(arg);
if (service)
return (unsigned long)(service->users);
else
return 0;
}
return rval;
case ARG_TYPE_DCB:
rval = (unsigned long)strtol(arg, NULL, 0);
if (mode == CLIM_USER && dcb_isvalid((DCB *)rval) == 0)
rval = 0;
return rval;
case ARG_TYPE_SESSION:
rval = (unsigned long)strtol(arg, NULL, 0);
if (mode == CLIM_USER && session_isvalid((SESSION *)rval) == 0)
rval = 0;
return rval;
case ARG_TYPE_MONITOR:
if (mode == CLIM_USER || (rval = (unsigned long)strtol(arg, NULL, 0)) == 0)
rval = (unsigned long)monitor_find(arg);
return rval;
} }
return 0; return 0;
} }
@ -396,33 +490,94 @@ int argc, i, j, found = 0;
char *args[MAXARGS]; char *args[MAXARGS];
char *saveptr, *delim = " \t\r\n"; char *saveptr, *delim = " \t\r\n";
unsigned long arg1, arg2, arg3; unsigned long arg1, arg2, arg3;
int in_quotes = 0, escape_next = 0;
char *ptr, *lptr;
/* Tokenize the input string */ args[0] = cli->cmdbuf;
args[0] = strtok_r(cli->cmdbuf, delim, &saveptr); ptr = args[0];
lptr = ptr;
i = 0; i = 0;
do { /*
* Break the command line into a number of words. Whitespace is used
* to delimit words and may be escaped by use of the \ character or
* the use of double quotes.
* The array args contains the broken down words, one per index.
*/
while (*ptr)
{
if (escape_next)
{
*lptr++ = *ptr++;
escape_next = 0;
}
else if (*ptr == '\\')
{
escape_next = 1;
ptr++;
}
else if (in_quotes == 0 && (*ptr == ' ' || *ptr == '\t' || *ptr == '\r' || *ptr == '\n'))
{
*lptr = 0;
if (args[i] == ptr)
args[i] = ptr + 1;
else
{
i++; i++;
args[i] = strtok_r(NULL, delim, &saveptr); if (i >= MAXARGS)
} while (args[i] != NULL && i < MAXARGS); break;
args[i] = ptr + 1;
}
ptr++;
lptr++;
}
else if (*ptr == '\"' && in_quotes == 0)
{
in_quotes = 1;
ptr++;
}
else if (*ptr == '\"' && in_quotes == 1)
{
in_quotes = 0;
ptr++;
}
else
{
*lptr++ = *ptr++;
}
}
*lptr = 0;
args[i+1] = NULL;
if (args[0] == NULL) if (args[0] == NULL || *args[0] == 0)
return 1; return 1;
argc = i - 2; /* The number of extra arguments to commands */ argc = i - 2; /* The number of extra arguments to commands */
if (!strcasecmp(args[0], "help")) if (!strcasecmp(args[0], "help"))
{ {
if (args[1] == NULL) if (args[1] == NULL || *args[1] == 0)
{ {
found = 1; found = 1;
dcb_printf(dcb, "Available commands:\n"); dcb_printf(dcb, "Available commands:\n");
for (i = 0; cmds[i].cmd; i++) for (i = 0; cmds[i].cmd; i++)
{ {
if (cmds[i].options[1].arg1 == NULL)
dcb_printf(dcb, " %s %s\n", cmds[i].cmd, cmds[i].options[0].arg1);
else
{
dcb_printf(dcb, " %s [", cmds[i].cmd);
for (j = 0; cmds[i].options[j].arg1; j++) for (j = 0; cmds[i].options[j].arg1; j++)
{ {
dcb_printf(dcb, " %s %s\n", cmds[i].cmd, cmds[i].options[j].arg1); dcb_printf(dcb, "%s%s", cmds[i].options[j].arg1,
cmds[i].options[j+1].arg1 ? "|" : "");
}
dcb_printf(dcb, "]\n");
} }
} }
dcb_printf(dcb, "\nType help command to see details of each command.\n");
dcb_printf(dcb, "Where commands require names as arguments and these names contain\n");
dcb_printf(dcb, "whitespace either the \\ character may be used to escape the whitespace\n");
dcb_printf(dcb, "or the name may be enclosed in double quotes \".\n\n");
} }
else else
{ {
@ -458,9 +613,9 @@ unsigned long arg1, arg2, arg3;
{ {
for (j = 0; cmds[i].options[j].arg1; j++) for (j = 0; cmds[i].options[j].arg1; j++)
{ {
found = 1; /**< command and sub-command match */
if (strcasecmp(args[1], cmds[i].options[j].arg1) == 0) if (strcasecmp(args[1], cmds[i].options[j].arg1) == 0)
{ {
found = 1; /**< command and sub-command match */
if (argc != cmds[i].options[j].n_args) if (argc != cmds[i].options[j].n_args)
{ {
dcb_printf(dcb, "Incorrect number of arguments: %s %s expects %d arguments\n", dcb_printf(dcb, "Incorrect number of arguments: %s %s expects %d arguments\n",
@ -476,7 +631,7 @@ unsigned long arg1, arg2, arg3;
cmds[i].options[j].fn(dcb); cmds[i].options[j].fn(dcb);
break; break;
case 1: case 1:
arg1 = convert_arg(args[2],cmds[i].options[j].arg_types[0]); arg1 = convert_arg(cli->mode, args[2],cmds[i].options[j].arg_types[0]);
if (arg1) if (arg1)
cmds[i].options[j].fn(dcb, arg1); cmds[i].options[j].fn(dcb, arg1);
else else
@ -484,8 +639,8 @@ unsigned long arg1, arg2, arg3;
args[2]); args[2]);
break; break;
case 2: case 2:
arg1 = convert_arg(args[2],cmds[i].options[j].arg_types[0]); arg1 = convert_arg(cli->mode, args[2],cmds[i].options[j].arg_types[0]);
arg2 = convert_arg(args[3],cmds[i].options[j].arg_types[1]); arg2 = convert_arg(cli->mode, args[3],cmds[i].options[j].arg_types[1]);
if (arg1 && arg2) if (arg1 && arg2)
cmds[i].options[j].fn(dcb, arg1, arg2); cmds[i].options[j].fn(dcb, arg1, arg2);
else if (arg1 == 0) else if (arg1 == 0)
@ -496,9 +651,9 @@ unsigned long arg1, arg2, arg3;
args[3]); args[3]);
break; break;
case 3: case 3:
arg1 = convert_arg(args[2],cmds[i].options[j].arg_types[0]); arg1 = convert_arg(cli->mode, args[2],cmds[i].options[j].arg_types[0]);
arg2 = convert_arg(args[3],cmds[i].options[j].arg_types[1]); arg2 = convert_arg(cli->mode, args[3],cmds[i].options[j].arg_types[1]);
arg3 = convert_arg(args[4],cmds[i].options[j].arg_types[2]); arg3 = convert_arg(cli->mode, args[4],cmds[i].options[j].arg_types[2]);
if (arg1 && arg2 && arg3) if (arg1 && arg2 && arg3)
cmds[i].options[j].fn(dcb, arg1, arg2, arg3); cmds[i].options[j].fn(dcb, arg1, arg2, arg3);
else if (arg1 == 0) else if (arg1 == 0)

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

@ -252,10 +252,12 @@ int i, n;
} }
else else
{ {
LOGIF(LE, (skygw_log_write( LOGIF(LM, (skygw_log_write(
LOGFILE_ERROR, LOGFILE_MESSAGE,
"Warning : Unsupported router " "* Warning : Unsupported router "
"option %s for readconnroute.", "option \'%s\' for readconnroute. "
"Expected router options are "
"[slave|master|synced]",
options[i]))); options[i])));
} }
} }

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

File diff suppressed because it is too large Load Diff

9
server/modules/routing/readwritesplit/test/test_autocommit_disabled3.sql Normal file
View File

@ -0,0 +1,9 @@
use test;
drop table if exists t1;
create table t1 (id integer);
set autocommit=0; -- open transaction
begin;
insert into t1 values(1); -- write to master
commit;
select count(*) from t1; -- read from master since autocommit is disabled
drop table t1;

41
server/modules/routing/test/makefile Normal file
View File

@ -0,0 +1,41 @@
# cleantests - clean local and subdirectories' tests
# buildtests - build all local and subdirectories' tests
# runtests - run all local tests
# testall - clean, build and run local and subdirectories' tests
include ../../../../build_gateway.inc
include $(ROOT_PATH)/makefile.inc
include $(ROOT_PATH)/test.inc
CC=cc
TESTLOG := $(shell pwd)/testrouting.log
RET := -1
cleantests:
- $(DEL) *.o
- $(DEL) *~
testall:
-$(MAKE) cleantests
-$(MAKE) DEBUG=Y buildtests
-$(MAKE) runtests
@echo "" >> $(TESTLOG)
@echo "-------------------------------" >> $(TESTLOG)
@echo $(shell date) >> $(TESTLOG)
@echo "Test Read/Write Split Router" >> $(TESTLOG)
$(MAKE) -C $(ROOT_PATH)/server/modules/routing/readwritesplit testall
buildtests:
$(MAKE) -C $(ROOT_PATH)/server/modules/routing/readwritesplit buildtests
runtests:
@echo "" > $(TESTLOG)
@echo "-------------------------------" >> $(TESTLOG)
@echo $(shell date) >> $(TESTLOG)
@echo "Test routing" >> $(TESTLOG)
@echo "-------------------------------" >> $(TESTLOG)
@echo "Nothing to run here so far" >> $(TESTLOG)
@cat $(TESTLOG) >> $(TEST_MAXSCALE_LOG)

28
utils/skygw_debug.h
View File

@ -119,7 +119,10 @@ typedef enum skygw_chk_t {
CHK_NUM_SESSION, CHK_NUM_SESSION,
CHK_NUM_ROUTER_SES, CHK_NUM_ROUTER_SES,
CHK_NUM_MY_SESCMD, CHK_NUM_MY_SESCMD,
CHK_NUM_ROUTER_PROPERTY CHK_NUM_ROUTER_PROPERTY,
CHK_NUM_SESCMD_CUR,
CHK_NUM_BACKEND,
CHK_NUM_BACKEND_REF
} skygw_chk_t; } skygw_chk_t;
# define STRBOOL(b) ((b) ? "true" : "false") # define STRBOOL(b) ((b) ? "true" : "false")
@ -221,6 +224,11 @@ typedef enum skygw_chk_t {
((t) == BE_UNDEFINED ? "BE_UNDEFINED" : \ ((t) == BE_UNDEFINED ? "BE_UNDEFINED" : \
"Unknown backend tpe"))) "Unknown backend tpe")))
#define STRCRITERIA(c) ((c) == UNDEFINED_CRITERIA ? "UNDEFINED_CRITERIA" : \
((c) == LEAST_GLOBAL_CONNECTIONS ? "LEAST_GLOBAL_CONNECTIONS" : \
((c) == LEAST_ROUTER_CONNECTIONS ? "LEAST_ROUTER_CONNECTIONS" : \
((c) == LEAST_BEHIND_MASTER ? "LEAST_BEHIND_MASTER" : "Unknown criteria"))))
#define CHK_MLIST(l) { \ #define CHK_MLIST(l) { \
ss_info_dassert((l->mlist_chk_top == CHK_NUM_MLIST && \ ss_info_dassert((l->mlist_chk_top == CHK_NUM_MLIST && \
l->mlist_chk_tail == CHK_NUM_MLIST), \ l->mlist_chk_tail == CHK_NUM_MLIST), \
@ -446,6 +454,24 @@ typedef enum skygw_chk_t {
"Session command has invalid check fields"); \ "Session command has invalid check fields"); \
} }
#define CHK_SESCMD_CUR(c) { \
ss_info_dassert((c)->scmd_cur_chk_top == CHK_NUM_SESCMD_CUR && \
(c)->scmd_cur_chk_tail == CHK_NUM_SESCMD_CUR, \
"Session command cursor has invalid check fields"); \
}
#define CHK_BACKEND(b) { \
ss_info_dassert((b)->be_chk_top == CHK_NUM_BACKEND && \
(b)->be_chk_tail == CHK_NUM_BACKEND, \
"BACKEND has invalid check fields"); \
}
#define CHK_BACKEND_REF(r) { \
ss_info_dassert((r)->bref_chk_top == CHK_NUM_BACKEND_REF && \
(r)->bref_chk_tail == CHK_NUM_BACKEND_REF, \
"Backend reference has invalid check fields"); \
}
#if defined(SS_DEBUG) #if defined(SS_DEBUG)
bool conn_open[10240]; bool conn_open[10240];