brand release-1.0beta-refresh merged

brand release-1.0beta-refresh merged
This commit is contained in:
MassimilianoPinto
2014-09-12 16:00:36 +02:00
168 changed files with 14144 additions and 927 deletions

View File

@ -33,7 +33,9 @@
# 29/06/13 Vilho Raatikka Reverted Query classifier changes because
# gateway needs mysql client lib, not qc.
# 24/07/13 Mark Ridoch Addition of encryption routines
# 30/05/14 Mark Ridoch Filter API added
# 30/05/14 Mark Riddoch Filter API added
# 25/07/14 Mark Riddoch Addition of hints
# 29/08/14 Mark Riddoch Added housekeeper
include ../../build_gateway.inc
@ -47,17 +49,23 @@ CFLAGS=-c -I/usr/include -I../include -I../modules/include -I../inih \
-I$(LOGPATH) -I$(UTILSPATH) \
-Wall -g
include ../../makefile.inc
LDFLAGS=-rdynamic -L$(LOGPATH) \
-Wl,-rpath,$(DEST)/lib \
-Wl,-rpath,$(LOGPATH) -Wl,-rpath,$(UTILSPATH) \
-Wl,-rpath,$(EMBEDDED_LIB)
LIBS=-L$(EMBEDDED_LIB) \
-lmysqld \
-lz -lm -lcrypt -lcrypto -ldl -laio -lrt -pthread -llog_manager \
-L../inih/extra -linih -lssl -lstdc++
include ../../makefile.inc
SRCS= atomic.c buffer.c spinlock.c gateway.c \
gw_utils.c utils.c dcb.c load_utils.c session.c service.c server.c \
poll.c config.c users.c hashtable.c dbusers.c thread.c gwbitmask.c \
monitor.c adminusers.c secrets.c filter.c modutil.c
monitor.c adminusers.c secrets.c filter.c modutil.c hint.c housekeeper.c
HDRS= ../include/atomic.h ../include/buffer.h ../include/dcb.h \
../include/gw.h ../modules/include/mysql_client_server_protocol.h \
@ -65,18 +73,13 @@ HDRS= ../include/atomic.h ../include/buffer.h ../include/dcb.h \
../include/modules.h ../include/poll.h ../include/config.h \
../include/users.h ../include/hashtable.h ../include/gwbitmask.h \
../include/adminusers.h ../include/version.h ../include/maxscale.h \
../include/filter.h modutil.h
../include/filter.h ../include/modutil.h ../hint.h ../include/housekeeper.h
OBJ=$(SRCS:.c=.o)
KOBJS=maxkeys.o secrets.o utils.o
POBJS=maxpasswd.o secrets.o utils.o
LIBS=-L$(EMBEDDED_LIB) \
-lmysqld \
-lz -lm -lcrypt -lcrypto -ldl -laio -lrt -pthread -llog_manager \
-L../inih/extra -linih -lssl -lstdc++
all: maxscale maxkeys maxpasswd
cleantests:

View File

@ -32,6 +32,9 @@
* 11/07/13 Mark Riddoch Add reference count mechanism
* 16/07/2013 Massimiliano Pinto Added command type to gwbuf struct
* 24/06/2014 Mark Riddoch Addition of gwbuf_trim
* 15/07/2014 Mark Riddoch Addition of properties
* 28/08/2014 Mark Riddoch Adition of tail pointer to speed
* the gwbuf_append process
*
* @endverbatim
*/
@ -40,6 +43,11 @@
#include <atomic.h>
#include <skygw_debug.h>
static buffer_object_t* gwbuf_remove_buffer_object(
GWBUF* buf,
buffer_object_t* bufobj);
/**
* Allocate a new gateway buffer structure of size bytes.
*
@ -77,13 +85,18 @@ SHARED_BUF *sbuf;
free(sbuf);
return NULL;
}
spinlock_init(&rval->gwbuf_lock);
rval->start = sbuf->data;
rval->end = rval->start + size;
sbuf->refcount = 1;
rval->sbuf = sbuf;
rval->next = NULL;
rval->tail = rval;
rval->hint = NULL;
rval->properties = NULL;
rval->gwbuf_type = GWBUF_TYPE_UNDEFINED;
rval->command = 0;
rval->gwbuf_info = GWBUF_INFO_NONE;
rval->gwbuf_bufobj = NULL;
CHK_GWBUF(rval);
return rval;
}
@ -96,12 +109,38 @@ SHARED_BUF *sbuf;
void
gwbuf_free(GWBUF *buf)
{
BUF_PROPERTY *prop;
buffer_object_t* bo;
CHK_GWBUF(buf);
if (atomic_add(&buf->sbuf->refcount, -1) == 1)
{
free(buf->sbuf->data);
free(buf->sbuf);
bo = buf->gwbuf_bufobj;
while (bo != NULL)
{
bo = gwbuf_remove_buffer_object(buf, bo);
}
}
while (buf->properties)
{
prop = buf->properties;
buf->properties = prop->next;
free(prop->name);
free(prop->value);
free(prop);
}
/** Release the hint */
while (buf->hint)
{
HINT* h = buf->hint;
buf->hint = buf->hint->next;
hint_free(h);
}
free(buf);
}
@ -130,7 +169,12 @@ GWBUF *rval;
rval->start = buf->start;
rval->end = buf->end;
rval->gwbuf_type = buf->gwbuf_type;
rval->properties = NULL;
rval->hint = NULL;
rval->gwbuf_info = buf->gwbuf_info;
rval->gwbuf_bufobj = buf->gwbuf_bufobj;
rval->next = NULL;
rval->tail = rval;
CHK_GWBUF(rval);
return rval;
}
@ -156,7 +200,12 @@ GWBUF *gwbuf_clone_portion(
clonebuf->start = (void *)((char*)buf->start)+start_offset;
clonebuf->end = (void *)((char *)clonebuf->start)+length;
clonebuf->gwbuf_type = buf->gwbuf_type; /*< clone the type for now */
clonebuf->properties = NULL;
clonebuf->hint = NULL;
clonebuf->gwbuf_info = buf->gwbuf_info;
clonebuf->gwbuf_bufobj = buf->gwbuf_bufobj;
clonebuf->next = NULL;
clonebuf->tail = clonebuf;
CHK_GWBUF(clonebuf);
return clonebuf;
@ -233,11 +282,8 @@ GWBUF *ptr = head;
if (!head)
return tail;
CHK_GWBUF(head);
while (ptr->next)
{
ptr = ptr->next;
}
ptr->next = tail;
head->tail->next = tail;
head->tail = tail->tail;
return head;
}
@ -262,6 +308,7 @@ GWBUF *
gwbuf_consume(GWBUF *head, unsigned int length)
{
GWBUF *rval = head;
CHK_GWBUF(head);
GWBUF_CONSUME(head, length);
CHK_GWBUF(head);
@ -269,8 +316,13 @@ GWBUF *rval = head;
if (GWBUF_EMPTY(head))
{
rval = head->next;
if (head->next)
head->next->tail = head->tail;
gwbuf_free(head);
}
ss_dassert(rval == NULL || (rval->end > rval->start));
return rval;
}
@ -302,6 +354,8 @@ int rval = 0;
* buffer has n_bytes or less then it will be freed and
* NULL will be returned.
*
* This routine assumes the buffer is not part of a chain
*
* @param buf The buffer to trim
* @param n_bytes The number of bytes to trim off
* @return The buffer chain or NULL if buffer has <= n_bytes
@ -309,6 +363,8 @@ int rval = 0;
GWBUF *
gwbuf_trim(GWBUF *buf, unsigned int n_bytes)
{
ss_dassert(buf->next == NULL);
if (GWBUF_LENGTH(buf) <= n_bytes)
{
gwbuf_consume(buf, GWBUF_LENGTH(buf));
@ -338,5 +394,192 @@ void gwbuf_set_type(
}
}
/**
* Add a buffer object to GWBUF buffer.
*
* @param buf GWBUF where object is added
* @param id Type identifier for object
* @param data Object data
* @param donefun_dp Clean-up function to be executed before buffer is freed.
*/
void gwbuf_add_buffer_object(
GWBUF* buf,
bufobj_id_t id,
void* data,
void (*donefun_fp)(void *))
{
buffer_object_t** p_b;
buffer_object_t* newb;
CHK_GWBUF(buf);
newb = (buffer_object_t *)malloc(sizeof(buffer_object_t));
newb->bo_id = id;
newb->bo_data = data;
newb->bo_donefun_fp = donefun_fp;
newb->bo_next = NULL;
/** Lock */
spinlock_acquire(&buf->gwbuf_lock);
p_b = &buf->gwbuf_bufobj;
/** Search the end of the list and add there */
while (*p_b != NULL)
{
p_b = &(*p_b)->bo_next;
}
*p_b = newb;
/** Set flag */
buf->gwbuf_info |= GWBUF_INFO_PARSED;
/** Unlock */
spinlock_release(&buf->gwbuf_lock);
}
/**
* Search buffer object which matches with the id.
*
* @param buf GWBUF to be searched
* @param id Identifier for the object
*
* @return Searched buffer object or NULL if not found
*/
void* gwbuf_get_buffer_object_data(
GWBUF* buf,
bufobj_id_t id)
{
buffer_object_t* bo;
CHK_GWBUF(buf);
/** Lock */
spinlock_acquire(&buf->gwbuf_lock);
bo = buf->gwbuf_bufobj;
while (bo != NULL && bo->bo_id != id)
{
bo = bo->bo_next;
}
/** Unlock */
spinlock_release(&buf->gwbuf_lock);
return bo->bo_data;
}
/**
* @return pointer to next buffer object or NULL
*/
static buffer_object_t* gwbuf_remove_buffer_object(
GWBUF* buf,
buffer_object_t* bufobj)
{
buffer_object_t* next;
next = bufobj->bo_next;
/** Call corresponding clean-up function to clean buffer object's data */
bufobj->bo_donefun_fp(bufobj->bo_data);
free(bufobj);
return next;
}
/**
* Add a property to a buffer.
*
* @param buf The buffer to add the property to
* @param name The property name
* @param value The property value
* @return Non-zero on success
*/
int
gwbuf_add_property(GWBUF *buf, char *name, char *value)
{
BUF_PROPERTY *prop;
if ((prop = malloc(sizeof(BUF_PROPERTY))) == NULL)
return 0;
prop->name = strdup(name);
prop->value = strdup(value);
spinlock_acquire(&buf->gwbuf_lock);
prop->next = buf->properties;
buf->properties = prop;
spinlock_release(&buf->gwbuf_lock);
return 1;
}
/**
* Return the value of a buffer property
* @param buf The buffer itself
* @param name The name of the property to return
* @return The property value or NULL if the property was not found.
*/
char *
gwbuf_get_property(GWBUF *buf, char *name)
{
BUF_PROPERTY *prop;
spinlock_acquire(&buf->gwbuf_lock);
prop = buf->properties;
while (prop && strcmp(prop->name, name) != 0)
prop = prop->next;
spinlock_release(&buf->gwbuf_lock);
if (prop)
return prop->value;
return NULL;
}
/**
* Convert a chain of GWBUF structures into a single GWBUF structure
*
* @param orig The chain to convert
* @return The contiguous buffer
*/
GWBUF *
gwbuf_make_contiguous(GWBUF *orig)
{
GWBUF *newbuf;
char *ptr;
int len;
if (orig->next == NULL)
return orig;
if ((newbuf = gwbuf_alloc(gwbuf_length(orig))) != NULL)
{
ptr = GWBUF_DATA(newbuf);
while (orig)
{
len = GWBUF_LENGTH(orig);
memcpy(ptr, GWBUF_DATA(orig), len);
ptr += len;
orig = gwbuf_consume(orig, len);
}
}
return newbuf;
}
/**
* Add hint to a buffer.
*
* @param buf The buffer to add the hint to
* @param hint The hint itself
* @return Non-zero on success
*/
int
gwbuf_add_hint(GWBUF *buf, HINT *hint)
{
HINT *ptr;
spinlock_acquire(&buf->gwbuf_lock);
if (buf->hint)
{
ptr = buf->hint;
while (ptr->next)
ptr = ptr->next;
ptr->next = hint;
}
else
{
buf->hint = hint;
}
spinlock_release(&buf->gwbuf_lock);
return 1;
}

View File

@ -34,6 +34,7 @@
* 29/05/14 Mark Riddoch Addition of filter definition
* 23/05/14 Massimiliano Pinto Added automatic set of maxscale-id: first listening ipv4_raw + port + pid
* 28/05/14 Massimiliano Pinto Added detect_replication_lag parameter
* 28/08/14 Massimiliano Pinto Added detect_stale_master parameter
* 09/09/14 Massimiliano Pinto Added localhost_match_any parameter
*
* @endverbatim
@ -41,6 +42,7 @@
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <ctype.h>
#include <ini.h>
#include <config.h>
#include <service.h>
@ -264,18 +266,28 @@ int error_count = 0;
{
char* max_slave_conn_str;
char* max_slave_rlag_str;
char *user;
char *auth;
char *enable_root_user;
char *weightby;
char *version_string;
bool is_rwsplit = false;
obj->element = service_alloc(obj->object, router);
char *user =
config_get_value(obj->parameters, "user");
char *auth =
config_get_value(obj->parameters, "passwd");
char *enable_root_user =
config_get_value(obj->parameters, "enable_root_user");
char *weightby =
config_get_value(obj->parameters, "weightby");
user = config_get_value(obj->parameters, "user");
auth = config_get_value(obj->parameters, "passwd");
enable_root_user = config_get_value(
obj->parameters,
"enable_root_user");
weightby = config_get_value(obj->parameters, "weightby");
char *version_string = config_get_value(obj->parameters, "version_string");
version_string = config_get_value(obj->parameters,
"version_string");
/** flag for rwsplit-specific parameters */
if (strncmp(router, "readwritesplit", strlen("readwritesplit")+1) == 0)
{
is_rwsplit = true;
}
char *allow_localhost_match_any =
config_get_value(obj->parameters, "localhost_match_any");
@ -347,13 +359,20 @@ int error_count = 0;
param = config_get_param(obj->parameters,
"max_slave_connections");
succp = service_set_param_value(
obj->element,
param,
max_slave_conn_str,
COUNT_ATMOST,
(COUNT_TYPE|PERCENT_TYPE));
if (param == NULL)
{
succp = false;
}
else
{
succp = service_set_param_value(
obj->element,
param,
max_slave_conn_str,
COUNT_ATMOST,
(COUNT_TYPE|PERCENT_TYPE));
}
if (!succp)
{
LOGIF(LM, (skygw_log_write(
@ -379,13 +398,20 @@ int error_count = 0;
obj->parameters,
"max_slave_replication_lag");
succp = service_set_param_value(
obj->element,
param,
max_slave_rlag_str,
COUNT_ATMOST,
COUNT_TYPE);
if (param == NULL)
{
succp = false;
}
else
{
succp = service_set_param_value(
obj->element,
param,
max_slave_rlag_str,
COUNT_ATMOST,
COUNT_TYPE);
}
if (!succp)
{
LOGIF(LM, (skygw_log_write(
@ -399,7 +425,51 @@ int error_count = 0;
param->value)));
}
}
}
/** Parameters for rwsplit router only */
if (is_rwsplit)
{
CONFIG_PARAMETER* param;
char* use_sql_variables_in;
bool succp;
use_sql_variables_in =
config_get_value(obj->parameters,
"use_sql_variables_in");
if (use_sql_variables_in != NULL)
{
param = config_get_param(
obj->parameters,
"use_sql_variables_in");
if (param == NULL)
{
succp = false;
}
else
{
succp = service_set_param_value(obj->element,
param,
use_sql_variables_in,
COUNT_NONE,
SQLVAR_TARGET_TYPE);
}
if (!succp)
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"* Warning : invalid value type "
"for parameter \'%s.%s = %s\'\n\tExpected "
"type is [master|all] for "
"use sql variables in.",
((SERVICE*)obj->element)->name,
param->name,
param->value)));
}
}
} /*< if (rw_split) */
} /*< if (router) */
else
{
obj->element = NULL;
@ -555,17 +625,29 @@ int error_count = 0;
while (s)
{
CONFIG_CONTEXT *obj1 = context;
int found = 0;
while (obj1)
{
if (strcmp(trim(s), obj1->object) == 0 &&
obj->element && obj1->element)
{
found = 1;
serviceAddBackend(
obj->element,
obj1->element);
}
obj1 = obj1->next;
}
if (!found)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Unable to find "
"server '%s' that is "
"configured as part of "
"service '%s'.",
s, obj->object)));
}
s = strtok(NULL, ",");
}
}
@ -573,7 +655,7 @@ int error_count = 0;
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : The service '%s' is missing a "
"Warning: The service '%s' is missing a "
"definition of the servers that provide "
"the service.",
obj->object)));
@ -676,6 +758,7 @@ int error_count = 0;
char *passwd;
unsigned long interval = 0;
int replication_heartbeat = 0;
int detect_stale_master = 0;
module = config_get_value(obj->parameters, "module");
servers = config_get_value(obj->parameters, "servers");
@ -689,6 +772,10 @@ int error_count = 0;
replication_heartbeat = atoi(config_get_value(obj->parameters, "detect_replication_lag"));
}
if (config_get_value(obj->parameters, "detect_stale_master")) {
detect_stale_master = atoi(config_get_value(obj->parameters, "detect_stale_master"));
}
if (module)
{
obj->element = monitor_alloc(obj->object, module);
@ -712,22 +799,38 @@ int error_count = 0;
if(replication_heartbeat == 1)
monitorSetReplicationHeartbeat(obj->element, replication_heartbeat);
/* detect stale master */
if(detect_stale_master == 1)
monitorDetectStaleMaster(obj->element, detect_stale_master);
/* get the servers to monitor */
s = strtok(servers, ",");
while (s)
{
CONFIG_CONTEXT *obj1 = context;
int found = 0;
while (obj1)
{
if (strcmp(s, obj1->object) == 0 &&
obj->element && obj1->element)
{
found = 1;
monitorAddServer(
obj->element,
obj1->element);
}
obj1 = obj1->next;
}
if (!found)
LOGIF(LE,
(skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Unable to find "
"server '%s' that is "
"configured in the "
"monitor '%s'.",
s, obj->object)));
s = strtok(NULL, ",");
}
}
@ -829,12 +932,15 @@ config_param_type_t config_get_paramtype(
return param->qfd_param_type;
}
int config_get_valint(
bool config_get_valint(
int* val,
CONFIG_PARAMETER* param,
const char* name, /*< if NULL examine current param only */
config_param_type_t ptype)
{
int val = -1; /*< -1 indicates failure */
{
bool succp = false;;
ss_dassert((ptype == COUNT_TYPE || ptype == PERCENT_TYPE) && param != NULL);
while (param)
{
@ -842,32 +948,95 @@ int config_get_valint(
{
switch (ptype) {
case COUNT_TYPE:
val = param->qfd.valcount;
goto return_val;
*val = param->qfd.valcount;
succp = true;
goto return_succp;
case PERCENT_TYPE:
val = param->qfd.valpercent;
goto return_val;
case BOOL_TYPE:
val = param->qfd.valbool;
goto return_val;
default:
goto return_val;
*val = param->qfd.valpercent;
succp =true;
goto return_succp;
default:
goto return_succp;
}
}
else if (name == NULL)
{
goto return_val;
}
param = param->next;
}
return_val:
return val;
return_succp:
return succp;
}
bool config_get_valbool(
bool* val,
CONFIG_PARAMETER* param,
const char* name,
config_param_type_t ptype)
{
bool succp;
ss_dassert(ptype == BOOL_TYPE);
ss_dassert(param != NULL);
if (ptype != BOOL_TYPE || param == NULL)
{
succp = false;
goto return_succp;
}
while (param)
{
if (name == NULL || !strncmp(param->name, name, MAX_PARAM_LEN))
{
*val = param->qfd.valbool;
succp = true;
goto return_succp;
}
param = param->next;
}
succp = false;
return_succp:
return succp;
}
bool config_get_valtarget(
target_t* val,
CONFIG_PARAMETER* param,
const char* name,
config_param_type_t ptype)
{
bool succp;
ss_dassert(ptype == SQLVAR_TARGET_TYPE);
ss_dassert(param != NULL);
if (ptype != SQLVAR_TARGET_TYPE || param == NULL)
{
succp = false;
goto return_succp;
}
while (param)
{
if (name == NULL || !strncmp(param->name, name, MAX_PARAM_LEN))
{
*val = param->qfd.valtarget;
succp = true;
goto return_succp;
}
param = param->next;
}
succp = false;
return_succp:
return succp;
}
CONFIG_PARAMETER* config_clone_param(
CONFIG_PARAMETER* param)
{
@ -932,6 +1101,15 @@ config_threadcount()
return gateway.n_threads;
}
static struct {
char *logname;
logfile_id_t logfile;
} lognames[] = {
{ "log_messages", LOGFILE_MESSAGE },
{ "log_trace", LOGFILE_TRACE },
{ "log_debug", LOGFILE_DEBUG },
{ NULL, 0 }
};
/**
* Configuration handler for items in the global [MaxScale] section
*
@ -942,10 +1120,20 @@ config_threadcount()
static int
handle_global_item(const char *name, const char *value)
{
int i;
if (strcmp(name, "threads") == 0) {
gateway.n_threads = atoi(value);
} else {
return 0;
for (i = 0; lognames[i].logname; i++)
{
if (strcasecmp(name, lognames[i].logname) == 0)
{
if (atoi(value))
skygw_log_enable(lognames[i].logfile);
else
skygw_log_disable(lognames[i].logfile);
}
}
}
return 1;
}
@ -1053,13 +1241,20 @@ SERVER *server;
param = config_get_param(obj->parameters,
"max_slave_connections");
succp = service_set_param_value(
service,
param,
max_slave_conn_str,
COUNT_ATMOST,
(PERCENT_TYPE|COUNT_TYPE));
if (param == NULL)
{
succp = false;
}
else
{
succp = service_set_param_value(
service,
param,
max_slave_conn_str,
COUNT_ATMOST,
(PERCENT_TYPE|COUNT_TYPE));
}
if (!succp)
{
LOGIF(LM, (skygw_log_write(
@ -1089,13 +1284,20 @@ SERVER *server;
obj->parameters,
"max_slave_replication_lag");
succp = service_set_param_value(
service,
param,
max_slave_rlag_str,
COUNT_ATMOST,
COUNT_TYPE);
if (param == NULL)
{
succp = false;
}
else
{
succp = service_set_param_value(
service,
param,
max_slave_rlag_str,
COUNT_ATMOST,
COUNT_TYPE);
}
if (!succp)
{
LOGIF(LM, (skygw_log_write(
@ -1237,11 +1439,13 @@ SERVER *server;
while (s)
{
CONFIG_CONTEXT *obj1 = context;
int found = 0;
while (obj1)
{
if (strcmp(s, obj1->object) == 0 &&
obj->element && obj1->element)
{
found = 1;
if (!serviceHasBackend(obj->element, obj1->element))
{
serviceAddBackend(
@ -1251,6 +1455,16 @@ SERVER *server;
}
obj1 = obj1->next;
}
if (!found)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Unable to find "
"server '%s' that is "
"configured as part of "
"service '%s'.",
s, obj->object)));
}
s = strtok(NULL, ",");
}
}
@ -1352,6 +1566,7 @@ static char *service_params[] =
"localhost_match_any",
"max_slave_connections",
"max_slave_replication_lag",
"use_sql_variables_in", /*< rwsplit only */
"version_string",
"filters",
NULL
@ -1388,6 +1603,7 @@ static char *monitor_params[] =
"passwd",
"monitor_interval",
"detect_replication_lag",
"detect_stale_master",
NULL
};
/**
@ -1474,7 +1690,11 @@ bool config_set_qualified_param(
param->qfd.valbool = *(bool *)val;
succp = true;
break;
case SQLVAR_TARGET_TYPE:
param->qfd.valtarget = *(target_t *)val;
succp = true;
break;
default:
succp = false;
break;

View File

@ -187,14 +187,6 @@ getUsers(SERVICE *service, struct users *users)
if (service_user == NULL || service_passwd == NULL)
return -1;
/** multi-thread environment requires that thread init succeeds. */
if (mysql_thread_init()) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : mysql_thread_init failed.")));
return -1;
}
con = mysql_init(NULL);
if (con == NULL) {
@ -388,10 +380,9 @@ getUsers(SERVICE *service, struct users *users)
memcpy(users->cksum, hash, SHA_DIGEST_LENGTH);
free(users_data);
free(key.user);
mysql_free_result(result);
mysql_close(con);
mysql_thread_end();
return total_users;
}

View File

@ -89,7 +89,13 @@ static int dcb_null_write(DCB *dcb, GWBUF *buf);
static int dcb_null_close(DCB *dcb);
static int dcb_null_auth(DCB *dcb, SERVER *server, SESSION *session, GWBUF *buf);
DCB* dcb_get_zombies(void)
/**
* Return the pointer to the lsit of zombie DCB's
*
* @return Zombies DCB list
*/
DCB *
dcb_get_zombies(void)
{
return zombies;
}
@ -128,6 +134,12 @@ DCB *rval;
spinlock_init(&rval->delayqlock);
spinlock_init(&rval->authlock);
spinlock_init(&rval->cb_lock);
spinlock_init(&rval->pollinlock);
spinlock_init(&rval->polloutlock);
rval->pollinbusy = 0;
rval->readcheck = 0;
rval->polloutbusy = 0;
rval->writecheck = 0;
rval->fd = -1;
memset(&rval->stats, 0, sizeof(DCBSTATS)); // Zero the statistics
rval->state = DCB_STATE_ALLOC;
@ -376,11 +388,6 @@ DCB_CALLBACK *cb;
}
spinlock_release(&dcb->cb_lock);
if (dcb->dcb_readqueue)
{
GWBUF* queue = dcb->dcb_readqueue;
while ((queue = gwbuf_consume(queue, GWBUF_LENGTH(queue))) != NULL);
}
bitmask_free(&dcb->memdata.bitmask);
simple_mutex_done(&dcb->dcb_read_lock);
simple_mutex_done(&dcb->dcb_write_lock);
@ -399,7 +406,7 @@ DCB_CALLBACK *cb;
*
* @param threadid The thread ID of the caller
*/
DCB*
DCB *
dcb_process_zombies(int threadid)
{
DCB *ptr, *lptr;
@ -815,8 +822,6 @@ int below_water;
spinlock_acquire(&dcb->writeqlock);
ss_dassert(dcb->state != DCB_STATE_ZOMBIE);
if (dcb->writeq != NULL)
{
/*
@ -1187,7 +1192,7 @@ printDCB(DCB *dcb)
if (dcb->remote)
printf("\tConnected to: %s\n", dcb->remote);
if (dcb->user)
printf("\tUsername to: %s\n", dcb->user);
printf("\tUsername to: %s\n", dcb->user);
if (dcb->writeq)
printf("\tQueued write data: %d\n",gwbuf_length(dcb->writeq));
printf("\tStatistics:\n");
@ -1204,6 +1209,19 @@ printDCB(DCB *dcb)
printf("\t\tNo. of Low Water Events: %d\n",
dcb->stats.n_low_water);
}
/**
* Display an entry from the spinlock statistics data
*
* @param dcb The DCB to print to
* @param desc Description of the statistic
* @param value The statistic value
*/
static void
spin_reporter(void *dcb, char *desc, int value)
{
dcb_printf((DCB *)dcb, "\t\t%-35s %d\n", desc, value);
}
/**
* Diagnostic to print all DCB allocated in the system
@ -1233,6 +1251,12 @@ void dprintAllDCBs(DCB *pdcb)
DCB *dcb;
spinlock_acquire(&dcbspin);
#if SPINLOCK_PROFILE
dcb_printf(pdcb, "DCB List Spinlock Statistics:\n");
spinlock_stats(&dcbspin, spin_reporter, pdcb);
dcb_printf(pdcb, "Zombie Queue Lock Statistics:\n");
spinlock_stats(&zombiespin, spin_reporter, pdcb);
#endif
dcb = allDCBs;
while (dcb)
{
@ -1252,12 +1276,16 @@ DCB *dcb;
dcb_printf(pdcb, "\tQueued write data: %d\n",
gwbuf_length(dcb->writeq));
dcb_printf(pdcb, "\tStatistics:\n");
dcb_printf(pdcb, "\t\tNo. of Reads: %d\n", dcb->stats.n_reads);
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 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_printf(pdcb, "\t\tNo. of Reads: %d\n", dcb->stats.n_reads);
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 Accepts: %d\n", dcb->stats.n_accepts);
dcb_printf(pdcb, "\t\tNo. of busy polls: %d\n", dcb->stats.n_busypolls);
dcb_printf(pdcb, "\t\tNo. of read rechecks: %d\n", dcb->stats.n_readrechecks);
dcb_printf(pdcb, "\t\tNo. of busy write polls: %d\n", dcb->stats.n_busywrpolls);
dcb_printf(pdcb, "\t\tNo. of write rechecks: %d\n", dcb->stats.n_writerechecks);
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);
if (dcb->flags & DCBF_CLONE)
dcb_printf(pdcb, "\t\tDCB is a clone.\n");
dcb = dcb->next;
@ -1278,20 +1306,20 @@ DCB *dcb;
spinlock_acquire(&dcbspin);
dcb = allDCBs;
dcb_printf(pdcb, "Descriptor Control Blocks\n");
dcb_printf(pdcb, "------------+----------------------------+----------------------+----------\n");
dcb_printf(pdcb, " %-10s | %-26s | %-20s | %s\n",
dcb_printf(pdcb, "------------------+----------------------------+--------------------+----------\n");
dcb_printf(pdcb, " %-16s | %-26s | %-18s | %s\n",
"DCB", "State", "Service", "Remote");
dcb_printf(pdcb, "------------+----------------------------+----------------------+----------\n");
dcb_printf(pdcb, "------------------+----------------------------+--------------------+----------\n");
while (dcb)
{
dcb_printf(pdcb, " %10p | %-26s | %-20s | %s\n",
dcb_printf(pdcb, " %-16p | %-26s | %-18s | %s\n",
dcb, gw_dcb_state2string(dcb->state),
(dcb->session->service ?
dcb->session->service->name : ""),
((dcb->session && dcb->session->service) ? dcb->session->service->name : ""),
(dcb->remote ? dcb->remote : ""));
dcb = dcb->next;
}
dcb_printf(pdcb, "------------+----------------------------+----------------------+----------\n\n");
dcb_printf(pdcb, "------------------+----------------------------+--------------------+----------\n\n");
spinlock_release(&dcbspin);
}
@ -1308,16 +1336,16 @@ DCB *dcb;
spinlock_acquire(&dcbspin);
dcb = allDCBs;
dcb_printf(pdcb, "Client Connections\n");
dcb_printf(pdcb, "-----------------+------------+----------------------+------------\n");
dcb_printf(pdcb, " %-15s | %-10s | %-20s | %s\n",
dcb_printf(pdcb, "-----------------+------------------+----------------------+------------\n");
dcb_printf(pdcb, " %-15s | %-16s | %-20s | %s\n",
"Client", "DCB", "Service", "Session");
dcb_printf(pdcb, "-----------------+------------+----------------------+------------\n");
dcb_printf(pdcb, "-----------------+------------------+----------------------+------------\n");
while (dcb)
{
if (dcb_isclient(dcb)
&& dcb->dcb_role == DCB_ROLE_REQUEST_HANDLER)
{
dcb_printf(pdcb, " %-15s | %10p | %-20s | %10p\n",
dcb_printf(pdcb, " %-15s | %16p | %-20s | %10p\n",
(dcb->remote ? dcb->remote : ""),
dcb, (dcb->session->service ?
dcb->session->service->name : ""),
@ -1325,7 +1353,7 @@ DCB *dcb;
}
dcb = dcb->next;
}
dcb_printf(pdcb, "-----------------+------------+----------------------+------------\n\n");
dcb_printf(pdcb, "-----------------+------------------+----------------------+------------\n\n");
spinlock_release(&dcbspin);
}
@ -1342,16 +1370,18 @@ dprintDCB(DCB *pdcb, DCB *dcb)
dcb_printf(pdcb, "DCB: %p\n", (void *)dcb);
dcb_printf(pdcb, "\tDCB state: %s\n", gw_dcb_state2string(dcb->state));
if (dcb->session && dcb->session->service)
dcb_printf(pdcb, "\tService: %s\n",
dcb_printf(pdcb, "\tService: %s\n",
dcb->session->service->name);
if (dcb->remote)
dcb_printf(pdcb, "\tConnected to: %s\n", dcb->remote);
if (dcb->user)
dcb_printf(pdcb, "\tUsername: %s\n",
dcb_printf(pdcb, "\tUsername: %s\n",
dcb->user);
dcb_printf(pdcb, "\tOwning Session: %p\n", dcb->session);
if (dcb->writeq)
dcb_printf(pdcb, "\tQueued write data: %d\n", gwbuf_length(dcb->writeq));
if (dcb->delayq)
dcb_printf(pdcb, "\tDelayed write data: %d\n", gwbuf_length(dcb->delayq));
dcb_printf(pdcb, "\tStatistics:\n");
dcb_printf(pdcb, "\t\tNo. of Reads: %d\n",
dcb->stats.n_reads);
@ -1361,12 +1391,30 @@ dprintDCB(DCB *pdcb, DCB *dcb)
dcb->stats.n_buffered);
dcb_printf(pdcb, "\t\tNo. of Accepts: %d\n",
dcb->stats.n_accepts);
dcb_printf(pdcb, "\t\tNo. of busy polls: %d\n", dcb->stats.n_busypolls);
dcb_printf(pdcb, "\t\tNo. of read rechecks: %d\n", dcb->stats.n_readrechecks);
dcb_printf(pdcb, "\t\tNo. of busy write polls: %d\n", dcb->stats.n_busywrpolls);
dcb_printf(pdcb, "\t\tNo. of write rechecks: %d\n", dcb->stats.n_writerechecks);
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);
if (dcb->flags & DCBF_CLONE)
dcb_printf(pdcb, "\t\tDCB is a clone.\n");
#if SPINLOCK_PROFILE
dcb_printf(pdcb, "\tInitlock Statistics:\n");
spinlock_stats(&dcb->dcb_initlock, spin_reporter, pdcb);
dcb_printf(pdcb, "\tWrite Queue Lock Statistics:\n");
spinlock_stats(&dcb->writeqlock, spin_reporter, pdcb);
dcb_printf(pdcb, "\tDelay Queue Lock Statistics:\n");
spinlock_stats(&dcb->delayqlock, spin_reporter, pdcb);
dcb_printf(pdcb, "\tPollin Lock Statistics:\n");
spinlock_stats(&dcb->pollinlock, spin_reporter, pdcb);
dcb_printf(pdcb, "\tPollout Lock Statistics:\n");
spinlock_stats(&dcb->polloutlock, spin_reporter, pdcb);
dcb_printf(pdcb, "\tCallback Lock Statistics:\n");
spinlock_stats(&dcb->cb_lock, spin_reporter, pdcb);
#endif
}
/**
@ -1719,10 +1767,7 @@ int gw_write(
* @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_add_callback(DCB *dcb, DCB_REASON reason, int (*callback)(struct dcb *, DCB_REASON, void *), void *userdata)
{
DCB_CALLBACK *cb, *ptr;
int rval = 1;
@ -1754,7 +1799,10 @@ int rval = 1;
return 0;
}
if (cb->next == NULL)
{
cb->next = ptr;
break;
}
cb = cb->next;
}
spinlock_release(&dcb->cb_lock);
@ -1775,7 +1823,7 @@ int rval = 1;
* @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_remove_callback(DCB *dcb, DCB_REASON reason, int (*callback)(struct dcb *, DCB_REASON, void *), void *userdata)
{
DCB_CALLBACK *cb, *pcb = NULL;
int rval = 0;
@ -1868,8 +1916,102 @@ int rval = 0;
return rval;
}
static DCB* dcb_get_next (
DCB* dcb)
/**
* Called by the EPOLLIN event. Take care of calling the protocol
* read entry point and managing multiple threads competing for the DCB
* without blocking those threads.
*
* This mechanism does away with the need for a mutex on the EPOLLIN event
* and instead implements a queuing mechanism in which nested events are
* queued on the DCB such that when the thread processing the first event
* returns it will read the queued event and process it. This allows the
* thread that woudl otherwise have to wait to process the nested event
* to return immediately and and process other events.
*
* @param dcb The DCB that has data available
*/
void
dcb_pollin(DCB *dcb, int thread_id)
{
spinlock_acquire(&dcb->pollinlock);
if (dcb->pollinbusy == 0)
{
dcb->pollinbusy = 1;
do {
if (dcb->readcheck)
{
dcb->stats.n_readrechecks++;
dcb_process_zombies(thread_id);
}
dcb->readcheck = 0;
spinlock_release(&dcb->pollinlock);
dcb->func.read(dcb);
spinlock_acquire(&dcb->pollinlock);
} while (dcb->readcheck);
dcb->pollinbusy = 0;
}
else
{
dcb->stats.n_busypolls++;
dcb->readcheck = 1;
}
spinlock_release(&dcb->pollinlock);
}
/**
* Called by the EPOLLOUT event. Take care of calling the protocol
* write_ready entry point and managing multiple threads competing for the DCB
* without blocking those threads.
*
* This mechanism does away with the need for a mutex on the EPOLLOUT event
* and instead implements a queuing mechanism in which nested events are
* queued on the DCB such that when the thread processing the first event
* returns it will read the queued event and process it. This allows the
* thread that would otherwise have to wait to process the nested event
* to return immediately and and process other events.
*
* @param dcb The DCB thats available for writes
*/
void
dcb_pollout(DCB *dcb, int thread_id)
{
spinlock_acquire(&dcb->polloutlock);
if (dcb->polloutbusy == 0)
{
dcb->polloutbusy = 1;
do {
if (dcb->writecheck)
{
dcb_process_zombies(thread_id);
dcb->stats.n_writerechecks++;
}
dcb->writecheck = 0;
spinlock_release(&dcb->polloutlock);
dcb->func.write_ready(dcb);
spinlock_acquire(&dcb->polloutlock);
} while (dcb->writecheck);
dcb->polloutbusy = 0;
}
else
{
dcb->stats.n_busywrpolls++;
dcb->writecheck = 1;
}
spinlock_release(&dcb->polloutlock);
}
/**
* Get the next DCB in the list of all DCB's
*
* @param dcb The current DCB
* @return The pointer to the next DCB or NULL if this is the last
*/
static DCB *
dcb_get_next (DCB* dcb)
{
DCB* p;
@ -1903,8 +2045,13 @@ static DCB* dcb_get_next (
return dcb;
}
void dcb_call_foreach (
DCB_REASON reason)
/**
* Call all the callbacks on all DCB's that match the reason given
*
* @param reason The DCB_REASON that triggers the callback
*/
void
dcb_call_foreach(DCB_REASON reason)
{
switch (reason) {
case DCB_REASON_CLOSE:

View File

@ -51,6 +51,7 @@
#include <modules.h>
#include <config.h>
#include <poll.h>
#include <housekeeper.h>
#include <stdlib.h>
#include <unistd.h>
@ -1492,11 +1493,16 @@ int main(int argc, char **argv)
/* Init MaxScale poll system */
poll_init();
/*<
* Start the services that were created above
*/
/**
* Init mysql thread context for main thread as well. Needed when users
* are queried from backends.
*/
mysql_thread_init();
/** Start the services that were created above */
n_services = serviceStartAll();
if (n_services == 0)
if (n_services == 0)
{
char* logerr = "Failed to start any MaxScale services. Exiting.";
print_log_n_stderr(true, !daemon_mode, logerr, logerr, 0);
@ -1510,6 +1516,12 @@ int main(int argc, char **argv)
log_flush_thr = thread_start(
log_flush_cb,
(void *)&log_flush_timeout_ms);
/*
* Start the housekeeper thread
*/
hkinit();
/*<
* Start the polling threads, note this is one less than is
* configured as the main thread will also poll.
@ -1549,9 +1561,13 @@ int main(int argc, char **argv)
/*< Stop all the monitors */
monitorStopAll();
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"MaxScale is shutting down.")));
/** Release mysql thread context*/
mysql_thread_end();
datadir_cleanup();
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,

View File

@ -28,7 +28,7 @@
* and value and to free them.
*
* The hashtable is arrange as a set of linked lists, the number of linked
* lists beign the hashsize as requested by the user. Entries are hashed by
* lists being the hashsize as requested by the user. Entries are hashed by
* calling the hash function that is passed in by the user, this is used as
* an index into the array of linked lists, usign modulo hashsize.
*
@ -63,6 +63,10 @@ static void hashtable_read_lock(HASHTABLE *table);
static void hashtable_read_unlock(HASHTABLE *table);
static void hashtable_write_lock(HASHTABLE *table);
static void hashtable_write_unlock(HASHTABLE *table);
static HASHTABLE *hashtable_alloc_real(HASHTABLE* target,
int size,
int (*hashfn)(),
int (*cmpfn)());
/**
* Special null function used as default memory allfunctions in the hashtable
@ -93,16 +97,44 @@ nullfn(void *data)
HASHTABLE *
hashtable_alloc(int size, int (*hashfn)(), int (*cmpfn)())
{
HASHTABLE *rval;
return hashtable_alloc_real(NULL, size, hashfn, cmpfn);
}
if ((rval = malloc(sizeof(HASHTABLE))) == NULL)
return NULL;
HASHTABLE* hashtable_alloc_flat(
HASHTABLE* target,
int size,
int (*hashfn)(),
int (*cmpfn)())
{
return hashtable_alloc_real(target, size, hashfn, cmpfn);
}
static HASHTABLE *
hashtable_alloc_real(
HASHTABLE* target,
int size,
int (*hashfn)(),
int (*cmpfn)())
{
HASHTABLE *rval;
if (target == NULL)
{
if ((rval = malloc(sizeof(HASHTABLE))) == NULL)
return NULL;
rval->ht_isflat = false;
}
else
{
rval = target;
rval->ht_isflat = true;
}
#if defined(SS_DEBUG)
rval->ht_chk_top = CHK_NUM_HASHTABLE;
rval->ht_chk_tail = CHK_NUM_HASHTABLE;
rval->ht_chk_top = CHK_NUM_HASHTABLE;
rval->ht_chk_tail = CHK_NUM_HASHTABLE;
#endif
rval->hashsize = size;
rval->hashsize = size > 0 ? size : 1;
rval->hashfn = hashfn;
rval->cmpfn = cmpfn;
rval->kcopyfn = nullfn;
@ -112,12 +144,12 @@ HASHTABLE *rval;
rval->n_readers = 0;
rval->writelock = 0;
spinlock_init(&rval->spin);
if ((rval->entries = (HASHENTRIES **)calloc(size, sizeof(HASHENTRIES *))) == NULL)
if ((rval->entries = (HASHENTRIES **)calloc(rval->hashsize, sizeof(HASHENTRIES *))) == NULL)
{
free(rval);
return NULL;
}
memset(rval->entries, 0, size * sizeof(HASHENTRIES *));
memset(rval->entries, 0, rval->hashsize * sizeof(HASHENTRIES *));
return rval;
}
@ -147,7 +179,11 @@ HASHENTRIES *entry, *ptr;
}
}
free(table->entries);
free(table);
if (!table->ht_isflat)
{
free(table);
}
}
/**

153
server/core/hint.c Normal file
View File

@ -0,0 +1,153 @@
/*
* This file is distributed as part of MaxScale. It is free
* software: you can redistribute it and/or modify it under the terms of the
* GNU General Public License as published by the Free Software Foundation,
* version 2.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program; if not, write to the Free Software Foundation, Inc., 51
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright SkySQL Ab 2014
*/
#include <stdio.h>
#include <stdlib.h>
#include <hint.h>
/**
* @file hint.c generic support routines for hints.
*
* @verbatim
* Revision History
*
* Date Who Description
* 25/07/14 Mark Riddoch Initial implementation
*
* @endverbatim
*/
/**
* Duplicate a list of hints
*
* @param hint The hint list to duplicate
* @return A duplicate of the list
*/
HINT *
hint_dup(HINT *hint)
{
HINT *nlhead = NULL, *nltail = NULL, *ptr1, *ptr2;
ptr1 = hint;
while (ptr1)
{
if ((ptr2 = (HINT *)malloc(sizeof(HINT))) == NULL)
return nlhead;
ptr2->type = ptr1->type;
if (ptr1->data)
ptr2->data = strdup(ptr1->data);
else
ptr2->data = NULL;
if (ptr1->value)
ptr2->value = strdup(ptr1->value);
else
ptr2->value = NULL;
ptr2->next = NULL;
if (nltail)
{
nltail->next = ptr2;
nltail = ptr2;
}
else
{
nlhead = ptr2;
nltail = ptr2;
}
ptr1 = ptr1->next;
}
return nlhead;
}
/**
* Create a ROUTE TO type hint
*
* @param head The current hint list
* @param type The HINT_TYPE
* @param data Data may be NULL or the name of a server to route to
* @return The result hint list
*/
HINT *
hint_create_route(HINT *head, HINT_TYPE type, char *data)
{
HINT *hint;
if ((hint = (HINT *)malloc(sizeof(HINT))) == NULL)
return head;
hint->next = head;
hint->type = type;
if (data)
hint->data = strdup(data);
else
hint->data = NULL;
hint->value = NULL;
return hint;
}
/**
* Create name/value parameter hint
*
* @param head The current hint list
* @param pname The parameter name
* @param value The parameter value
* @return The result hint list
*/
HINT *
hint_create_parameter(HINT *head, char *pname, char *value)
{
HINT *hint;
if ((hint = (HINT *)malloc(sizeof(HINT))) == NULL)
return head;
hint->next = head;
hint->type = HINT_PARAMETER;
hint->data = pname;
hint->value = strdup(value);
return hint;
}
/**
* free_hint - free a hint
*
* @param hint The hint to free
*/
void
hint_free(HINT *hint)
{
if (hint->data)
free(hint->data);
if (hint->value)
free(hint->value);
free(hint);
}
bool hint_exists(
HINT** p_hint,
HINT_TYPE type)
{
bool succp = false;
while (*p_hint != NULL)
{
if ((*p_hint)->type == type)
{
succp = true;
}
p_hint = &(*p_hint)->next;
}
return succp;
}

195
server/core/housekeeper.c Normal file
View File

@ -0,0 +1,195 @@
/*
* This file is distributed as part of the SkySQL Gateway. It is free
* software: you can redistribute it and/or modify it under the terms of the
* GNU General Public License as published by the Free Software Foundation,
* version 2.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program; if not, write to the Free Software Foundation, Inc., 51
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright SkySQL Ab 2014
*/
#include <stdlib.h>
#include <housekeeper.h>
#include <thread.h>
#include <spinlock.h>
/**
* @file housekeeper.c Provide a mechanism to run periodic tasks
*
* @verbatim
* Revision History
*
* Date Who Description
* 29/08/14 Mark Riddoch Initial implementation
*
* @endverbatim
*/
/**
* List of all tasks that need to be run
*/
static HKTASK *tasks = NULL;
/**
* Spinlock to protect the tasks list
*/
static SPINLOCK tasklock = SPINLOCK_INIT;
static void hkthread(void *);
/**
* Initialise the housekeeper thread
*/
void
hkinit()
{
thread_start(hkthread, NULL);
}
/**
* Add a new task to the housekeepers lists of tasks that should be
* run periodically.
*
* The task will be first run frequency seconds after this call is
* made and will the be executed repeatedly every frequency seconds
* until the task is removed.
*
* Task names must be unique.
*
* @param name The unique name for this housekeeper task
* @param taskfn The function to call for the task
* @param data Data to pass to the task function
* @param frequency How often to run the task, expressed in seconds
* @return Return the tiem in seconds when the task will be first run if the task was added, otherwise 0
*/
int
hktask_add(char *name, void (*taskfn)(void *), void *data, int frequency)
{
HKTASK *task, *ptr;
if ((task = (HKTASK *)malloc(sizeof(HKTASK))) == NULL)
{
return 0;
}
if ((task->name = strdup(name)) == NULL)
{
free(task);
return 0;
}
task->task = taskfn;
task->data = data;
task->frequency = frequency;
task->nextdue = time(0) + frequency;
task->next = NULL;
spinlock_acquire(&tasklock);
ptr = tasks;
while (ptr && ptr->next)
{
if (strcmp(ptr->name, name) == 0)
{
spinlock_release(&tasklock);
free(task->name);
free(task);
return 0;
}
ptr = ptr->next;
}
if (ptr)
ptr->next = task;
else
tasks = task;
spinlock_release(&tasklock);
return task->nextdue;
}
/**
* Remove a named task from the housekeepers task list
*
* @param name The task name to remove
* @return Returns 0 if the task could not be removed
*/
int
hktask_remove(char *name)
{
HKTASK *ptr, *lptr = NULL;
spinlock_acquire(&tasklock);
ptr = tasks;
while (ptr && strcmp(ptr->name, name) != 0)
{
lptr = ptr;
ptr = ptr->next;
}
if (ptr && lptr)
lptr->next = ptr->next;
else if (ptr)
tasks = ptr->next;
spinlock_release(&tasklock);
if (ptr)
{
free(ptr->name);
free(ptr);
return 1;
}
else
{
return 0;
}
}
/**
* The housekeeper thread implementation.
*
* This function is responsible for executing the housekeeper tasks.
*
* The implementation of the callng of the task functions is such that
* the tasks are called without the tasklock spinlock being held. This
* allows manipulation of the housekeeper task list during execution of
* one of the tasks. The resutl is that upon completion of a task the
* search for tasks to run must restart from the start of the queue.
* It is vital that the task->nextdue tiem is updated before the task
* is run.
*
* @param data Unused, here to satisfy the thread system
*/
void
hkthread(void *data)
{
HKTASK *ptr;
time_t now;
void (*taskfn)(void *);
void *taskdata;
for (;;)
{
thread_millisleep(1000);
now = time(0);
spinlock_acquire(&tasklock);
ptr = tasks;
while (ptr)
{
if (ptr->nextdue <= now)
{
ptr->nextdue = now + ptr->frequency;
taskfn = ptr->task;
taskdata = ptr->data;
spinlock_release(&tasklock);
(*taskfn)(taskdata);
spinlock_acquire(&tasklock);
ptr = tasks;
}
else
ptr = ptr->next;
}
spinlock_release(&tasklock);
}
}

View File

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

View File

@ -207,8 +207,7 @@ MONITOR *ptr;
/**
* Show a single monitor
*
* @param dcb DCB for printing output
* @param monitor The monitor to print information regarding
* @param dcb DCB for printing output
*/
void
monitorShow(DCB *dcb, MONITOR *monitor)
@ -304,12 +303,26 @@ monitorSetInterval (MONITOR *mon, unsigned long interval)
* Enable Replication Heartbeat support in monitor.
*
* @param mon The monitor instance
* @param replication_heartbeat The replication heartbeat
* @param enable The enabling value is 1, 0 turns it off
*/
void
monitorSetReplicationHeartbeat(MONITOR *mon, int replication_heartbeat)
monitorSetReplicationHeartbeat(MONITOR *mon, int enable)
{
if (mon->module->replicationHeartbeat != NULL) {
mon->module->replicationHeartbeat(mon->handle, replication_heartbeat);
mon->module->replicationHeartbeat(mon->handle, enable);
}
}
/**
* Enable Stale Master assignement.
*
* @param mon The monitor instance
* @param enable The enabling value is 1, 0 turns it off
*/
void
monitorDetectStaleMaster(MONITOR *mon, int enable)
{
if (mon->module->detectStaleMaster != NULL) {
mon->module->detectStaleMaster(mon->handle, enable);
}
}

View File

@ -28,6 +28,8 @@
#include <skygw_utils.h>
#include <log_manager.h>
#include <gw.h>
#include <config.h>
#include <housekeeper.h>
extern int lm_enabled_logfiles_bitmask;
@ -41,14 +43,63 @@ extern int lm_enabled_logfiles_bitmask;
* 19/06/13 Mark Riddoch Initial implementation
* 28/06/13 Mark Riddoch Added poll mask support and DCB
* zombie management
* 29/08/14 Mark Riddoch Addition of thread status data, load average
* etc.
*
* @endverbatim
*/
static int epoll_fd = -1; /*< The epoll file descriptor */
static int do_shutdown = 0; /*< Flag the shutdown of the poll subsystem */
static int do_shutdown = 0; /*< Flag the shutdown of the poll subsystem */
static GWBITMASK poll_mask;
static simple_mutex_t epoll_wait_mutex; /*< serializes calls to epoll_wait */
static int n_waiting = 0; /*< No. of threads in epoll_wait */
/**
* Thread load average, this is the average number of descriptors in each
* poll completion, a value of 1 or less is the ideal.
*/
static double load_average = 0.0;
static int load_samples = 0;
static int load_nfds = 0;
static double current_avg = 0.0;
static double *avg_samples = NULL;
static int next_sample = 0;
static int n_avg_samples;
/* Thread statistics data */
static int n_threads; /*< No. of threads */
/**
* Internal MaxScale thread states
*/
typedef enum { THREAD_STOPPED, THREAD_IDLE,
THREAD_POLLING, THREAD_PROCESSING,
THREAD_ZPROCESSING } THREAD_STATE;
/**
* Thread data used to report the current state and activity related to
* a thread
*/
typedef struct {
THREAD_STATE state; /*< Current thread state */
int n_fds; /*< No. of descriptors thread is processing */
DCB *cur_dcb; /*< Current DCB being processed */
uint32_t event; /*< Current event being processed */
} THREAD_DATA;
static THREAD_DATA *thread_data = NULL; /*< Status of each thread */
/**
* The number of buckets used to gather statistics about how many
* descriptors where processed on each epoll completion.
*
* An array of wakeup counts is created, with the number of descriptors used
* to index that array. Each time a completion occurs the n_fds - 1 value is
* used to index this array and increment the count held there.
* If n_fds - 1 >= MAXFDS then the count at MAXFDS -1 is incremented.
*/
#define MAXNFDS 10
/**
* The polling statistics
@ -60,8 +111,20 @@ static struct {
int n_hup; /*< Number of hangup events */
int n_accept; /*< Number of accept events */
int n_polls; /*< Number of poll cycles */
int n_nothreads; /*< Number of times no threads are polling */
int n_fds[MAXNFDS]; /*< Number of wakeups with particular
n_fds value */
} pollStats;
/**
* How frequently to call the poll_loadav function used to monitor the load
* average of the poll subsystem.
*/
#define POLL_LOAD_FREQ 10
/**
* Periodic function to collect load data for average calculations
*/
static void poll_loadav(void *);
/**
* Initialise the polling system we are using for the gateway.
@ -71,6 +134,8 @@ static struct {
void
poll_init()
{
int i;
if (epoll_fd != -1)
return;
if ((epoll_fd = epoll_create(MAX_EVENTS)) == -1)
@ -80,7 +145,23 @@ poll_init()
}
memset(&pollStats, 0, sizeof(pollStats));
bitmask_init(&poll_mask);
n_threads = config_threadcount();
if ((thread_data =
(THREAD_DATA *)malloc(n_threads * sizeof(THREAD_DATA))) != NULL)
{
for (i = 0; i < n_threads; i++)
{
thread_data[i].state = THREAD_STOPPED;
}
}
simple_mutex_init(&epoll_wait_mutex, "epoll_wait_mutex");
hktask_add("Load Average", poll_loadav, NULL, POLL_LOAD_FREQ);
n_avg_samples = 15 * 60 / POLL_LOAD_FREQ;
avg_samples = (double *)malloc(sizeof(double *) * n_avg_samples);
for (i = 0; i < n_avg_samples; i++)
avg_samples[i] = 0.0;
}
/**
@ -100,7 +181,7 @@ poll_add_dcb(DCB *dcb)
CHK_DCB(dcb);
ev.events = EPOLLIN | EPOLLOUT | EPOLLET;
ev.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLHUP | EPOLLET;
ev.data.ptr = dcb;
/*<
@ -245,20 +326,29 @@ return_rc:
void
poll_waitevents(void *arg)
{
struct epoll_event events[MAX_EVENTS];
int i, nfds;
int thread_id = (int)arg;
bool no_op = false;
static bool process_zombies_only = false; /*< flag for all threads */
DCB *zombies = NULL;
struct epoll_event events[MAX_EVENTS];
int i, nfds;
int thread_id = (int)arg;
bool no_op = false;
static bool process_zombies_only = false; /*< flag for all threads */
DCB *zombies = NULL;
/* Add this thread to the bitmask of running polling threads */
/** Add this thread to the bitmask of running polling threads */
bitmask_set(&poll_mask, thread_id);
if (thread_data)
{
thread_data[thread_id].state = THREAD_IDLE;
}
/** Init mysql thread context for use with a mysql handle and a parser */
mysql_thread_init();
while (1)
{
atomic_add(&n_waiting, 1);
#if BLOCKINGPOLL
nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
atomic_add(&n_waiting, -1);
#else /* BLOCKINGPOLL */
if (!no_op) {
LOGIF(LD, (skygw_log_write(
@ -272,9 +362,14 @@ poll_waitevents(void *arg)
#if 0
simple_mutex_lock(&epoll_wait_mutex, TRUE);
#endif
if (thread_data)
{
thread_data[thread_id].state = THREAD_POLLING;
}
if ((nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, 0)) == -1)
{
atomic_add(&n_waiting, -1);
int eno = errno;
errno = 0;
LOGIF(LD, (skygw_log_write(
@ -288,6 +383,7 @@ poll_waitevents(void *arg)
}
else if (nfds == 0)
{
atomic_add(&n_waiting, -1);
if (process_zombies_only) {
#if 0
simple_mutex_unlock(&epoll_wait_mutex);
@ -310,6 +406,13 @@ poll_waitevents(void *arg)
}
}
}
else
{
atomic_add(&n_waiting, -1);
}
if (n_waiting == 0)
atomic_add(&pollStats.n_nothreads, 1);
#if 0
simple_mutex_unlock(&epoll_wait_mutex);
#endif
@ -322,6 +425,20 @@ poll_waitevents(void *arg)
pthread_self(),
nfds)));
atomic_add(&pollStats.n_polls, 1);
if (thread_data)
{
thread_data[thread_id].n_fds = nfds;
thread_data[thread_id].cur_dcb = NULL;
thread_data[thread_id].event = 0;
thread_data[thread_id].state = THREAD_PROCESSING;
}
pollStats.n_fds[(nfds < MAXNFDS ? (nfds - 1) : MAXNFDS - 1)]++;
load_average = (load_average * load_samples + nfds)
/ (load_samples + 1);
atomic_add(&load_samples, 1);
atomic_add(&load_nfds, nfds);
for (i = 0; i < nfds; i++)
{
@ -329,6 +446,11 @@ poll_waitevents(void *arg)
__uint32_t ev = events[i].events;
CHK_DCB(dcb);
if (thread_data)
{
thread_data[thread_id].cur_dcb = dcb;
thread_data[thread_id].event = ev;
}
#if defined(SS_DEBUG)
if (dcb_fake_write_ev[dcb->fd] != 0) {
@ -364,6 +486,7 @@ poll_waitevents(void *arg)
eno = gw_getsockerrno(dcb->fd);
if (eno == 0) {
#if MUTEX_BLOCK
simple_mutex_lock(
&dcb->dcb_write_lock,
true);
@ -378,6 +501,11 @@ poll_waitevents(void *arg)
dcb->dcb_write_active = FALSE;
simple_mutex_unlock(
&dcb->dcb_write_lock);
#else
atomic_add(&pollStats.n_write,
1);
dcb_pollout(dcb, thread_id);
#endif
} else {
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
@ -393,11 +521,13 @@ poll_waitevents(void *arg)
}
if (ev & EPOLLIN)
{
#if MUTEX_BLOCK
simple_mutex_lock(&dcb->dcb_read_lock,
true);
ss_info_dassert(!dcb->dcb_read_active,
"Read already active");
dcb->dcb_read_active = TRUE;
#endif
if (dcb->state == DCB_STATE_LISTENING)
{
@ -421,11 +551,17 @@ poll_waitevents(void *arg)
dcb,
dcb->fd)));
atomic_add(&pollStats.n_read, 1);
#if MUTEX_BLOCK
dcb->func.read(dcb);
#else
dcb_pollin(dcb, thread_id);
#endif
}
#if MUTEX_BLOCK
dcb->dcb_read_active = FALSE;
simple_mutex_unlock(
&dcb->dcb_read_lock);
#endif
}
if (ev & EPOLLERR)
{
@ -475,10 +611,33 @@ poll_waitevents(void *arg)
atomic_add(&pollStats.n_hup, 1);
dcb->func.hangup(dcb);
}
if (ev & EPOLLRDHUP)
{
int eno = 0;
eno = gw_getsockerrno(dcb->fd);
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [poll_waitevents] "
"EPOLLRDHUP on dcb %p, fd %d. "
"Errno %d, %s.",
pthread_self(),
dcb,
dcb->fd,
eno,
strerror(eno))));
atomic_add(&pollStats.n_hup, 1);
dcb->func.hangup(dcb);
}
} /*< for */
no_op = FALSE;
}
process_zombies:
if (thread_data)
{
thread_data[thread_id].state = THREAD_ZPROCESSING;
}
zombies = dcb_process_zombies(thread_id);
if (zombies == NULL) {
@ -491,10 +650,20 @@ poll_waitevents(void *arg)
* Remove the thread from the bitmask of running
* polling threads.
*/
if (thread_data)
{
thread_data[thread_id].state = THREAD_STOPPED;
}
bitmask_clear(&poll_mask, thread_id);
return;
}
if (thread_data)
{
thread_data[thread_id].state = THREAD_IDLE;
}
} /*< while(1) */
/** Release mysql thread context */
mysql_thread_end();
}
/**
@ -525,10 +694,194 @@ poll_bitmask()
void
dprintPollStats(DCB *dcb)
{
dcb_printf(dcb, "Number of epoll cycles: %d\n", pollStats.n_polls);
dcb_printf(dcb, "Number of read events: %d\n", pollStats.n_read);
dcb_printf(dcb, "Number of write events: %d\n", pollStats.n_write);
dcb_printf(dcb, "Number of error events: %d\n", pollStats.n_error);
dcb_printf(dcb, "Number of hangup events: %d\n", pollStats.n_hup);
dcb_printf(dcb, "Number of accept events: %d\n", pollStats.n_accept);
int i;
dcb_printf(dcb, "Number of epoll cycles: %d\n",
pollStats.n_polls);
dcb_printf(dcb, "Number of read events: %d\n",
pollStats.n_read);
dcb_printf(dcb, "Number of write events: %d\n",
pollStats.n_write);
dcb_printf(dcb, "Number of error events: %d\n",
pollStats.n_error);
dcb_printf(dcb, "Number of hangup events: %d\n",
pollStats.n_hup);
dcb_printf(dcb, "Number of accept events: %d\n",
pollStats.n_accept);
dcb_printf(dcb, "Number of times no threads polling: %d\n",
pollStats.n_nothreads);
dcb_printf(dcb, "No of poll completions with descriptors\n");
dcb_printf(dcb, "\tNo. of descriptors\tNo. of poll completions.\n");
for (i = 0; i < MAXNFDS - 1; i++)
{
dcb_printf(dcb, "\t%2d\t\t\t%d\n", i + 1, pollStats.n_fds[i]);
}
dcb_printf(dcb, "\t>= %d\t\t\t%d\n", MAXNFDS,
pollStats.n_fds[MAXNFDS-1]);
}
/**
* Convert an EPOLL event mask into a printable string
*
* @param event The event mask
* @return A string representation, the caller must free the string
*/
static char *
event_to_string(uint32_t event)
{
char *str;
str = malloc(22); // 22 is max returned string length
if (str == NULL)
return NULL;
*str = 0;
if (event & EPOLLIN)
{
strcat(str, "IN");
}
if (event & EPOLLOUT)
{
if (*str)
strcat(str, "|");
strcat(str, "OUT");
}
if (event & EPOLLERR)
{
if (*str)
strcat(str, "|");
strcat(str, "ERR");
}
if (event & EPOLLHUP)
{
if (*str)
strcat(str, "|");
strcat(str, "HUP");
}
if (event & EPOLLRDHUP)
{
if (*str)
strcat(str, "|");
strcat(str, "RDHUP");
}
return str;
}
/**
* Print the thread status for all the polling threads
*
* @param dcb The DCB to send the thread status data
*/
void
dShowThreads(DCB *dcb)
{
int i, j, n;
char *state;
double avg1 = 0.0, avg5 = 0.0, avg15 = 0.0;
dcb_printf(dcb, "Polling Threads.\n\n");
dcb_printf(dcb, "Historic Thread Load Average: %.2f.\n", load_average);
dcb_printf(dcb, "Current Thread Load Average: %.2f.\n", current_avg);
/* Average all the samples to get the 15 minute average */
for (i = 0; i < n_avg_samples; i++)
avg15 += avg_samples[i];
avg15 = avg15 / n_avg_samples;
/* Average the last third of the samples to get the 5 minute average */
n = 5 * 60 / POLL_LOAD_FREQ;
i = next_sample - (n + 1);
if (i < 0)
i += n_avg_samples;
for (j = i; j < i + n; j++)
avg5 += avg_samples[j % n_avg_samples];
avg5 = (3 * avg5) / (n_avg_samples);
/* Average the last 15th of the samples to get the 1 minute average */
n = 60 / POLL_LOAD_FREQ;
i = next_sample - (n + 1);
if (i < 0)
i += n_avg_samples;
for (j = i; j < i + n; j++)
avg1 += avg_samples[j % n_avg_samples];
avg1 = (15 * avg1) / (n_avg_samples);
dcb_printf(dcb, "15 Minute Average: %.2f, 5 Minute Average: %.2f, "
"1 Minute Average: %.2f\n\n", avg15, avg5, avg1);
if (thread_data == NULL)
return;
dcb_printf(dcb, " ID | State | # fds | Descriptor | Event\n");
dcb_printf(dcb, "----+------------+--------+------------------+---------------\n");
for (i = 0; i < n_threads; i++)
{
switch (thread_data[i].state)
{
case THREAD_STOPPED:
state = "Stopped";
break;
case THREAD_IDLE:
state = "Idle";
break;
case THREAD_POLLING:
state = "Polling";
break;
case THREAD_PROCESSING:
state = "Processing";
break;
case THREAD_ZPROCESSING:
state = "Collecting";
break;
}
if (thread_data[i].state != THREAD_PROCESSING)
dcb_printf(dcb,
" %2d | %-10s | | |\n",
i, state);
else if (thread_data[i].cur_dcb == NULL)
dcb_printf(dcb,
" %2d | %-10s | %6d | |\n",
i, state, thread_data[i].n_fds);
else
{
char *event_string
= event_to_string(thread_data[i].event);
if (event_string == NULL)
event_string = "??";
dcb_printf(dcb,
" %2d | %-10s | %6d | %-16p | %s\n",
i, state, thread_data[i].n_fds,
thread_data[i].cur_dcb, event_string);
free(event_string);
}
}
}
/**
* The function used to calculate time based load data. This is called by the
* housekeeper every POLL_LOAD_FREQ seconds.
*
* @param data Argument required by the housekeeper but not used here
*/
static void
poll_loadav(void *data)
{
static int last_samples = 0, last_nfds = 0;
int new_samples, new_nfds;
new_samples = load_samples - last_samples;
new_nfds = load_nfds - last_nfds;
last_samples = load_samples;
last_nfds = load_nfds;
/* POLL_LOAD_FREQ average is... */
if (new_samples)
current_avg = new_nfds / new_samples;
else
current_avg = 0.0;
avg_samples[next_sample] = current_avg;
next_sample++;
if (next_sample >= n_avg_samples)
next_sample = 0;
}

View File

@ -30,6 +30,7 @@
* 28/05/14 Massimiliano Pinto Addition of rlagd and node_ts fields
* 20/06/14 Massimiliano Pinto Addition of master_id, depth, slaves fields
* 26/06/14 Mark Riddoch Addition of server parameters
* 30/08/14 Massimiliano Pinto Addition of new service status description
*
* @endverbatim
*/
@ -148,7 +149,8 @@ server_set_unique_name(SERVER *server, char *name)
* Find an existing server using the unique section name in
* configuration file
*
* @param name The Server name defined in the header file
* @param servname The Server name or address
* @param port The server port
* @return The server or NULL if not found
*/
SERVER *
@ -373,23 +375,23 @@ char *stat;
if (ptr)
{
dcb_printf(dcb, "Servers.\n");
dcb_printf(dcb, "-------------------+-----------------+-------+----------------------+------------\n");
dcb_printf(dcb, "%-18s | %-15s | Port | %-20s | Connections\n",
dcb_printf(dcb, "-------------------+-----------------+-------+-------------+--------------------\n");
dcb_printf(dcb, "%-18s | %-15s | Port | Connections | %-20s",
"Server", "Address", "Status");
dcb_printf(dcb, "-------------------+-----------------+-------+----------------------+------------\n");
dcb_printf(dcb, "-------------------+-----------------+-------+-------------+--------------------\n");
}
while (ptr)
{
stat = server_status(ptr);
dcb_printf(dcb, "%-18s | %-15s | %5d | %-20s | %4d\n",
dcb_printf(dcb, "%-18s | %-15s | %5d | %11d | %s\n",
ptr->unique_name, ptr->name,
ptr->port, stat,
ptr->stats.n_current);
ptr->port,
ptr->stats.n_current, stat);
free(stat);
ptr = ptr->next;
}
if (allServers)
dcb_printf(dcb, "-------------------+-----------------+-------+----------------------+------------\n\n");
dcb_printf(dcb, "-------------------+-----------------+-------+-------------+--------------------\n");
spinlock_release(&server_spin);
}
@ -405,7 +407,7 @@ server_status(SERVER *server)
{
char *status = NULL;
if ((status = (char *)malloc(200)) == NULL)
if ((status = (char *)malloc(256)) == NULL)
return NULL;
status[0] = 0;
if (server->status & SERVER_MAINT)
@ -416,6 +418,14 @@ char *status = NULL;
strcat(status, "Slave, ");
if (server->status & SERVER_JOINED)
strcat(status, "Synced, ");
if (server->status & SERVER_NDB)
strcat(status, "NDB, ");
if (server->status & SERVER_SLAVE_OF_EXTERNAL_MASTER)
strcat(status, "Slave of External Server, ");
if (server->status & SERVER_STALE_STATUS)
strcat(status, "Stale Status, ");
if (server->status & SERVER_AUTH_ERROR)
strcat(status, "Auth Error, ");
if (server->status & SERVER_RUNNING)
strcat(status, "Running");
else

View File

@ -56,9 +56,29 @@
extern int lm_enabled_logfiles_bitmask;
/** To be used with configuration type checks */
typedef struct typelib_st {
int tl_nelems;
const char* tl_name;
const char** tl_p_elems;
} typelib_t;
/** Set of subsequent false,true pairs */
static const char* bool_strings[11] = {"FALSE", "TRUE", "OFF", "ON", "N", "Y", "0", "1", "NO", "YES", 0};
typelib_t bool_type = {array_nelems(bool_strings)-1, "bool_type", bool_strings};
/** List of valid values */
static const char* sqlvar_target_strings[4] = {"MASTER", "ALL", 0};
typelib_t sqlvar_target_type = {
array_nelems(sqlvar_target_strings)-1,
"sqlvar_target_type",
sqlvar_target_strings
};
static SPINLOCK service_spin = SPINLOCK_INIT;
static SERVICE *allServices = NULL;
static int find_type(typelib_t* tl, const char* needle, int maxlen);
static void service_add_qualified_param(
SERVICE* svc,
CONFIG_PARAMETER* param);
@ -677,6 +697,7 @@ int n = 0;
"Unable to find filter '%s' for service '%s'\n",
trim(ptr), service->name
)));
n--;
}
flist[n] = NULL;
ptr = strtok_r(NULL, "|", &brkt);
@ -1008,78 +1029,174 @@ bool service_set_param_value (
count_spec_t count_spec,
config_param_type_t type)
{
char* p;
int valint;
bool succp = true;
/**
* 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 if (PARAM_IS_TYPE(type,PERCENT_TYPE))
{
succp = true;
config_set_qualified_param(param, (void *)&valint, PERCENT_TYPE);
}
else
{
/** Log error */
}
}
else
{
succp = false;
}
}
else if (*p == '\0')
{
valint = (int) strtol(valstr, (char **)NULL, 10);
char* p;
int valint;
bool valbool;
target_t valtarget;
bool succp = true;
if (valint == 0 && errno != 0)
{
succp = false;
}
else if (PARAM_IS_TYPE(type,COUNT_TYPE))
{
succp = true;
config_set_qualified_param(param, (void *)&valint, COUNT_TYPE);
}
else
{
/** Log error */
}
}
if (PARAM_IS_TYPE(type,PERCENT_TYPE) ||PARAM_IS_TYPE(type,COUNT_TYPE))
{
/**
* 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 if (PARAM_IS_TYPE(type,PERCENT_TYPE))
{
succp = true;
config_set_qualified_param(param, (void *)&valint, PERCENT_TYPE);
}
else
{
/** Log error */
}
}
else
{
succp = false;
}
}
else if (*p == '\0')
{
valint = (int) strtol(valstr, (char **)NULL, 10);
if (valint == 0 && errno != 0)
{
succp = false;
}
else if (PARAM_IS_TYPE(type,COUNT_TYPE))
{
succp = true;
config_set_qualified_param(param, (void *)&valint, COUNT_TYPE);
}
else
{
/** Log error */
}
}
}
else if (type == BOOL_TYPE)
{
unsigned int rc;
rc = find_type(&bool_type, valstr, strlen(valstr)+1);
if (rc > 0)
{
succp = true;
if (rc%2 == 1)
{
valbool = false;
}
else if (rc%2 == 0)
{
valbool = true;
}
/** add param to config */
config_set_qualified_param(param,
(void *)&valbool,
BOOL_TYPE);
}
else
{
succp = false;
}
}
else if (type == SQLVAR_TARGET_TYPE)
{
unsigned int rc;
rc = find_type(&sqlvar_target_type, valstr, strlen(valstr)+1);
if (rc > 0 && rc < 3)
{
succp = true;
if (rc == 1)
{
valtarget = TYPE_MASTER;
}
else if (rc == 2)
{
valtarget = TYPE_ALL;
}
/** add param to config */
config_set_qualified_param(param,
(void *)&valtarget,
SQLVAR_TARGET_TYPE);
}
else
{
succp = false;
}
}
if (succp)
{
service_add_qualified_param(service, param); /*< add param to svc */
service_add_qualified_param(service, param); /*< add param to svc */
}
return succp;
}
/*
* Function to find a string in typelib_t
* (similar to find_type() of mysys/typelib.c)
*
* SYNOPSIS
* find_type()
* lib typelib_t
* find String to find
* length Length of string to find
* part_match Allow part matching of value
*
* RETURN
* 0 error
* > 0 position in TYPELIB->type_names +1
*/
static int find_type(
typelib_t* tl,
const char* needle,
int maxlen)
{
int i;
if (tl == NULL || needle == NULL || maxlen <= 0)
{
return -1;
}
for (i=0; i<tl->tl_nelems; i++)
{
if (strncasecmp(tl->tl_p_elems[i], needle, maxlen) == 0)
{
return i+1;
}
}
return 0;
}
/**
* Add qualified config parameter to SERVICE struct.
*/
*/
static void service_add_qualified_param(
SERVICE* svc,
CONFIG_PARAMETER* param)

View File

@ -694,6 +694,7 @@ int i;
return 0;
}
session->tail = *tail;
free(tail);
}
return 1;

View File

@ -40,9 +40,12 @@ void
spinlock_init(SPINLOCK *lock)
{
lock->lock = 0;
#ifdef DEBUG
#ifdef SPINLOCK_PROFILE
lock->spins = 0;
lock->acquired = 0;
lock->waiting = 0;
lock->max_waiting = 0;
lock->contended = 0;
#endif
}
@ -54,16 +57,29 @@ spinlock_init(SPINLOCK *lock)
void
spinlock_acquire(SPINLOCK *lock)
{
#ifdef SPINLOCK_PROFILE
int spins = 0;
atomic_add(&(lock->waiting), 1);
#endif
while (atomic_add(&(lock->lock), 1) != 0)
{
atomic_add(&(lock->lock), -1);
#ifdef DEBUG
#ifdef SPINLOCK_PROFILE
atomic_add(&(lock->spins), 1);
spins++;
#endif
}
#ifdef DEBUG
#ifdef SPINLOCK_PROFILE
if (spins)
{
lock->contended++;
if (lock->maxspins < spins)
lock->maxspins = spins;
}
lock->acquired++;
lock->owner = THREAD_SHELF();
atomic_add(&(lock->waiting), -1);
#endif
}
@ -71,7 +87,7 @@ spinlock_acquire(SPINLOCK *lock)
* Acquire a spinlock if it is not already locked.
*
* @param lock The spinlock to acquire
* @return True ifthe spinlock was acquired, otherwise false
* @return True if the spinlock was acquired, otherwise false
*/
int
spinlock_acquire_nowait(SPINLOCK *lock)
@ -81,7 +97,7 @@ spinlock_acquire_nowait(SPINLOCK *lock)
atomic_add(&(lock->lock), -1);
return FALSE;
}
#ifdef DEBUG
#ifdef SPINLOCK_PROFILE
lock->acquired++;
lock->owner = THREAD_SHELF();
#endif
@ -96,5 +112,45 @@ spinlock_acquire_nowait(SPINLOCK *lock)
void
spinlock_release(SPINLOCK *lock)
{
#ifdef SPINLOCK_PROFILE
if (lock->waiting > lock->max_waiting)
lock->max_waiting = lock->waiting;
#endif
atomic_add(&(lock->lock), -1);
}
/**
* Report statistics on a spinlock. This only has an effect if the
* spinlock code has been compiled with the SPINLOCK_PROFILE option set.
*
* NB A callback function is used to return the data rather than
* merely printing to a DCB in order to avoid a dependency on the DCB
* form the spinlock code and also to facilitate other uses of the
* statistics reporting.
*
* @param lock The spinlock to report on
* @param reporter The callback function to pass the statistics to
* @param hdl A handle that is passed to the reporter function
*/
void
spinlock_stats(SPINLOCK *lock, void (*reporter)(void *, char *, int), void *hdl)
{
#ifdef SPINLOCK_PROFILE
reporter(hdl, "Spinlock acquired", lock->acquired);
if (lock->acquired)
{
reporter(hdl, "Total no. of spins", lock->spins);
reporter(hdl, "Average no. of spins (overall)",
lock->spins / lock->acquired);
if (lock->contended)
reporter(hdl, "Average no. of spins (when contended)",
lock->spins / lock->contended);
reporter(hdl, "Maximum no. of spins", lock->maxspins);
reporter(hdl, "Maximim no. of blocked threads",
lock->max_waiting);
reporter(hdl, "Contended locks", lock->contended);
reporter(hdl, "Contention percentage",
(lock->contended * 100) / lock->acquired);
}
#endif
}

View File

@ -13,13 +13,13 @@ TESTLOG := $(shell pwd)/testcore.log
LOGPATH := $(ROOT_PATH)/log_manager
UTILSPATH := $(ROOT_PATH)/utils
LDFLAGS=-rdynamic -L$(LOGPATH) \
LDFLAGS=-rdynamic -L$(LOGPATH) -L$(EMBEDDED_LIB) \
-Wl,-rpath,$(DEST)/lib \
-Wl,-rpath,$(LOGPATH) -Wl,-rpath,$(UTILSPATH) \
-Wl,-rpath,$(EMBEDDED_LIB)
LIBS= -lz -lm -lcrypt -lcrypto -ldl -laio -lrt -pthread -llog_manager \
-L../../inih/extra -linih -lssl -lstdc++
-L../../inih/extra -linih -lssl -lstdc++ -lmysqld
TESTS=testhash testspinlock testfilter testadminusers

View File

@ -1,19 +1,73 @@
/*
* This file is distributed as part of MaxScale. It is free
* software: you can redistribute it and/or modify it under the terms of the
* GNU General Public License as published by the Free Software Foundation,
* version 2.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program; if not, write to the Free Software Foundation, Inc., 51
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright SkySQL Ab 2014
*/
/**
*
* @verbatim
* Revision History
*
* Date Who Description
* 18/08-2014 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <time.h>
#include "../../include/hashtable.h"
static void
read_lock(HASHTABLE *table)
{
spinlock_acquire(&table->spin);
while (table->writelock)
{
spinlock_release(&table->spin);
while (table->writelock)
;
spinlock_acquire(&table->spin);
}
table->n_readers++;
spinlock_release(&table->spin);
}
static void
read_unlock(HASHTABLE *table)
{
atomic_add(&table->n_readers, -1);
}
static int hfun(void* key);
static int cmpfun (void *, void *);
static int hfun(
void* key)
{
return *(int *)key;
int *i = (int *)key;
int j = (*i * 23) + 41;
return j;
/* return *(int *)key; */
}
static int cmpfun(
void* v1,
void* v2)
@ -27,7 +81,19 @@ static int cmpfun(
return (i1 < i2 ? -1 : (i1 > i2 ? 1 : 0));
}
static double start;
/**
* test1 spinlock_acquire_nowait tests
*
* Test that spinlock_acquire_nowait returns false if the spinlock
* is already taken.
*
* Test that spinlock_acquire_nowait returns true if the spinlock
* is not taken.
*
* Test that spinlock_acquire_nowait does hold the spinlock.
*/
static bool do_hashtest(
int argelems,
int argsize)
@ -39,12 +105,14 @@ static bool do_hashtest(
int* val_arr;
int hsize;
int longest;
int* iter;
ss_dfprintf(stderr,
"testhash : creating hash table of size %d, including %d "
"elements in total.",
"elements in total, at time %g.",
argsize,
argelems);
argelems,
(double)clock()-start);
val_arr = (int *)malloc(sizeof(void *)*argelems);
@ -56,17 +124,33 @@ static bool do_hashtest(
val_arr[i] = i;
hashtable_add(h, (void *)&val_arr[i], (void *)&val_arr[i]);
}
if (argelems > 1000) ss_dfprintf(stderr, "\t..done\nOperation took %g", (double)clock()-start);
ss_dfprintf(stderr, "\t..done\nRead hash table statistics.");
hashtable_get_stats((void *)h, &hsize, &nelems, &longest);
ss_dfprintf(stderr, "\t..done\nValidate read values.");
ss_info_dassert(hsize == argsize, "Invalid hash size");
ss_info_dassert(hsize == (argsize > 0 ? argsize: 1), "Invalid hash size");
ss_info_dassert((nelems == argelems) || (nelems == 0 && argsize == 0),
"Invalid element count");
ss_info_dassert(longest <= nelems, "Too large longest list value");
if (argelems > 1000) ss_dfprintf(stderr, "\t..done\nOperation took %g", (double)clock()-start);
ss_dfprintf(stderr, "\t..done\nValidate iterator.");
HASHITERATOR *iterator = hashtable_iterator(h);
read_lock(h);
for (i=0; i < (argelems+1); i++) {
iter = (int *)hashtable_next(iterator);
if (iter == NULL) break;
if (argelems < 100) ss_dfprintf(stderr, "\nNext item, iter = %d, i = %d", *iter, i);
}
read_unlock(h);
ss_info_dassert((i == argelems) || (i == 0 && argsize == 0), "\nIncorrect number of elements from iterator");
hashtable_iterator_free(iterator);
if (argelems > 1000) ss_dfprintf(stderr, "\t..done\nOperation took %g", (double)clock()-start);
ss_dfprintf(stderr, "\t\t..done\n\nTest completed successfully.\n\n");
@ -91,11 +175,13 @@ return_succp:
int main(void)
{
int rc = 1;
start = (double) clock();
if (!do_hashtest(0, 1)) goto return_rc;
if (!do_hashtest(10, 1)) goto return_rc;
if (!do_hashtest(1000, 10)) goto return_rc;
if (!do_hashtest(10, 0)) goto return_rc;
if (!do_hashtest(10, -5)) goto return_rc;
if (!do_hashtest(1500, 17)) goto return_rc;
if (!do_hashtest(1, 1)) goto return_rc;
if (!do_hashtest(10000, 133)) goto return_rc;

View File

@ -55,18 +55,18 @@ SPINLOCK lck;
spinlock_acquire(&lck);
if (spinlock_acquire_nowait(&lck))
{
fprintf(stderr, "spinlock_acquire_nowait: test 1 failed.\n");
fprintf(stderr, "spinlock_acquire_nowait: test 1.1 failed.\n");
return 1;
}
spinlock_release(&lck);
if (!spinlock_acquire_nowait(&lck))
{
fprintf(stderr, "spinlock_acquire_nowait: test 2 failed.\n");
fprintf(stderr, "spinlock_acquire_nowait: test 1.2 failed.\n");
return 1;
}
if (spinlock_acquire_nowait(&lck))
{
fprintf(stderr, "spinlock_acquire_nowait: test 3 failed.\n");
fprintf(stderr, "spinlock_acquire_nowait: test 1.3 failed.\n");
return 1;
}
spinlock_release(&lck);
@ -89,6 +89,8 @@ unsigned long t1 = time(0);
}
/**
* test2 spinlock_acquire tests
*
* Check that spinlock correctly blocks another thread whilst the spinlock
* is held.
*
@ -114,7 +116,7 @@ void *handle;
if (acquire_time < 8)
{
fprintf(stderr, "spinlock: test 1 failed.\n");
fprintf(stderr, "spinlock: test 2 failed.\n");
return 1;
}
return 0;