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

Remove BACKEND structure from readconnroute

Browse Source 
The BACKEND structure in readconnroute is now replaced with the use of the
SERVER_REF structure of the service. This allows dynamic changes to the
list of servers to be made.
This commit is contained in:
Markus Makela
2016-11-03 16:27:15 +02:00
parent dc8c068936
commit 155161a876
3 changed files with 99 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
include/maxscale/service.h
View File

@ -104,6 +104,9 @@ typedef struct server_ref_t
bool active; /**< Whether this reference is valid and in use*/ bool active; /**< Whether this reference is valid and in use*/
} SERVER_REF; } SERVER_REF;
/** Macro to check whether a SERVER_REF is active */
#define SERVER_REF_IS_ACTIVE(ref) (ref->active)
#define SERVICE_MAX_RETRY_INTERVAL 3600 /*< The maximum interval between service start retries */ #define SERVICE_MAX_RETRY_INTERVAL 3600 /*< The maximum interval between service start retries */
/** Value of service timeout if timeout checks are disabled */ /** Value of service timeout if timeout checks are disabled */

16
server/modules/routing/readconnroute/readconnection.h
View File

@ -32,18 +32,6 @@
MXS_BEGIN_DECLS MXS_BEGIN_DECLS
/**
* 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.
*/
typedef struct backend
{
SERVER *server; /*< The server itself */
int current_connection_count; /*< Number of connections to the server */
int weight; /*< Desired routing weight */
} BACKEND;
/** /**
* The client session structure used within this router. * The client session structure used within this router.
*/ */
@ -55,7 +43,7 @@ typedef struct router_client_session
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 */
bool rses_closed; /*< true when closeSession is called */ bool rses_closed; /*< true when closeSession is called */
BACKEND *backend; /*< Backend used by the client session */ SERVER_REF *backend; /*< Backend used by the client session */
DCB *backend_dcb; /*< DCB Connection to the backend */ DCB *backend_dcb; /*< DCB Connection to the backend */
DCB *client_dcb; /**< Client DCB */ DCB *client_dcb; /**< Client DCB */
struct router_client_session *next; struct router_client_session *next;
@ -79,9 +67,7 @@ typedef struct
typedef struct router_instance typedef struct router_instance
{ {
SERVICE *service; /*< Pointer to the service using this router */ SERVICE *service; /*< Pointer to the service using this router */
ROUTER_CLIENT_SES *connections; /*< Link list of all the client connections */
SPINLOCK lock; /*< Spinlock for the instance data */ SPINLOCK lock; /*< Spinlock for the instance data */
BACKEND **servers; /*< List of backend servers */
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 */

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

@ -132,7 +132,7 @@ static bool rses_begin_locked_router_action(ROUTER_CLIENT_SES* rses);
static void rses_end_locked_router_action(ROUTER_CLIENT_SES* rses); static void rses_end_locked_router_action(ROUTER_CLIENT_SES* rses);
static BACKEND *get_root_master(BACKEND **servers); static SERVER_REF *get_root_master(SERVER_REF *servers);
static int handle_state_switch(DCB* dcb, DCB_REASON reason, void * routersession); static int handle_state_switch(DCB* dcb, DCB_REASON reason, void * routersession);
static SPINLOCK instlock; static SPINLOCK instlock;
static ROUTER_INSTANCE *instances; static ROUTER_INSTANCE *instances;
@ -178,14 +178,6 @@ static inline void free_readconn_instance(ROUTER_INSTANCE *router)
{ {
if (router) if (router)
{ {
if (router->servers)
{
for (int i = 0; router->servers[i]; i++)
{
MXS_FREE(router->servers[i]);
}
}
MXS_FREE(router->servers);
MXS_FREE(router); MXS_FREE(router);
} }
} }
@ -214,37 +206,6 @@ createInstance(SERVICE *service, char **options)
inst->service = service; inst->service = service;
spinlock_init(&inst->lock); spinlock_init(&inst->lock);
/*
* We need an array of the backend servers in the instance structure so
* that we can maintain a count of the number of connections to each
* backend server.
*/
for (sref = service->dbref, n = 0; sref; sref = sref->next)
{
n++;
}
inst->servers = (BACKEND **) MXS_CALLOC(n + 1, sizeof(BACKEND *));
if (!inst->servers)
{
free_readconn_instance(inst);
return NULL;
}
for (sref = service->dbref, n = 0; sref; sref = sref->next)
{
if ((inst->servers[n] = MXS_MALLOC(sizeof(BACKEND))) == NULL)
{
free_readconn_instance(inst);
return NULL;
}
inst->servers[n]->server = sref->server;
inst->servers[n]->current_connection_count = 0;
inst->servers[n]->weight = sref->weight;
n++;
}
inst->servers[n] = NULL;
/* /*
* Process the options * Process the options
*/ */
@ -329,9 +290,9 @@ newSession(ROUTER *instance, SESSION *session)
{ {
ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *) instance; ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *) instance;
ROUTER_CLIENT_SES *client_rses; ROUTER_CLIENT_SES *client_rses;
BACKEND *candidate = NULL; SERVER_REF *candidate = NULL;
int i; int i;
BACKEND *master_host = NULL; SERVER_REF *master_host = NULL;
MXS_DEBUG("%lu [newSession] new router session with session " MXS_DEBUG("%lu [newSession] new router session with session "
"%p, and inst %p.", "%p, and inst %p.",
@ -339,7 +300,6 @@ newSession(ROUTER *instance, SESSION *session)
session, session,
inst); inst);
client_rses = (ROUTER_CLIENT_SES *) MXS_CALLOC(1, sizeof(ROUTER_CLIENT_SES)); client_rses = (ROUTER_CLIENT_SES *) MXS_CALLOC(1, sizeof(ROUTER_CLIENT_SES));
if (client_rses == NULL) if (client_rses == NULL)
@ -356,7 +316,7 @@ newSession(ROUTER *instance, SESSION *session)
/** /**
* Find the Master host from available servers * Find the Master host from available servers
*/ */
master_host = get_root_master(inst->servers); master_host = get_root_master(inst->service->dbref);
/** /**
* Find a backend server to connect to. This is the extent of the * Find a backend server to connect to. This is the extent of the
@ -376,52 +336,43 @@ newSession(ROUTER *instance, SESSION *session)
* become the new candidate. This has the effect of spreading the * become the new candidate. This has the effect of spreading the
* connections over different servers during periods of very low load. * connections over different servers during periods of very low load.
*/ */
for (i = 0; inst->servers[i]; i++) for (SERVER_REF *ref = inst->service->dbref; ref; ref = ref->next)
{ {
if (inst->servers[i]) if (!SERVER_REF_IS_ACTIVE(ref) || SERVER_IN_MAINT(ref->server) || ref->weight == 0)
{
continue;
}
else
{ {
MXS_DEBUG("%lu [newSession] Examine server in port %d with " MXS_DEBUG("%lu [newSession] Examine server in port %d with "
"%d connections. Status is %s, " "%d connections. Status is %s, "
"inst->bitvalue is %d", "inst->bitvalue is %d",
pthread_self(), pthread_self(),
inst->servers[i]->server->port, ref->server->port,
inst->servers[i]->current_connection_count, ref->connections,
STRSRVSTATUS(inst->servers[i]->server), STRSRVSTATUS(ref->server),
inst->bitmask); inst->bitmask);
} }
if (SERVER_IN_MAINT(inst->servers[i]->server))
{
continue;
}
if (inst->servers[i]->weight == 0)
{
continue;
}
/* Check server status bits against bitvalue from router_options */ /* Check server status bits against bitvalue from router_options */
if (inst->servers[i] && if (ref && SERVER_IS_RUNNING(ref->server) &&
SERVER_IS_RUNNING(inst->servers[i]->server) && (ref->server->status & inst->bitmask & inst->bitvalue))
(inst->servers[i]->server->status & inst->bitmask & inst->bitvalue))
{ {
if (master_host) if (master_host)
{ {
if (inst->servers[i] == master_host && (inst->bitvalue & SERVER_SLAVE)) if (ref == master_host && (inst->bitvalue & SERVER_SLAVE))
{ {
/* skip root Master here, as it could also be slave of an external server /* Skip root master here, as it could also be slave of an external server that
* that is not in the configuration. * is not in the configuration. Intermediate masters (Relay Servers) are also
* Intermediate masters (Relay Servers) are also slave and will be selected * slave and will be selected as Slave(s)
* as Slave(s)
*/ */
continue; continue;
} }
if (inst->servers[i] == master_host && (inst->bitvalue & SERVER_MASTER)) if (ref == master_host && (inst->bitvalue & SERVER_MASTER))
{ {
/* If option is "master" return only the root Master as there /* If option is "master" return only the root Master as there could be
* could be intermediate masters (Relay Servers) * intermediate masters (Relay Servers) and they must not be selected.
* and they must not be selected.
*/ */
candidate = master_host; candidate = master_host;
@ -430,8 +381,7 @@ newSession(ROUTER *instance, SESSION *session)
} }
else else
{ {
/* master_host is NULL, no master server. /* Master_host is NULL, no master server. If requested router_option is 'master'
* If requested router_option is 'master'
* candidate wll be NULL. * candidate wll be NULL.
*/ */
if (inst->bitvalue & SERVER_MASTER) if (inst->bitvalue & SERVER_MASTER)
@ -441,40 +391,31 @@ newSession(ROUTER *instance, SESSION *session)
} }
} }
/* If no candidate set, set first running server as /* If no candidate set, set first running server as our initial candidate server */
our initial candidate server */
if (candidate == NULL) if (candidate == NULL)
{ {
candidate = inst->servers[i]; candidate = ref;
} }
else if (((inst->servers[i]->current_connection_count + 1) else if (((ref->connections + 1) * 1000) / ref->weight <
* 1000) / inst->servers[i]->weight < ((candidate->connections + 1) * 1000) / candidate->weight)
((candidate->current_connection_count + 1) *
1000) / candidate->weight)
{ {
/* This running server has fewer /* This running server has fewer connections, set it as a new candidate */
connections, set it as a new candidate */ candidate = ref;
candidate = inst->servers[i];
} }
else if (((inst->servers[i]->current_connection_count + 1) else if (((ref->connections + 1) * 1000) / ref->weight ==
* 1000) / inst->servers[i]->weight == ((candidate->connections + 1) * 1000) / candidate->weight &&
((candidate->current_connection_count + 1) * ref->server->stats.n_connections < candidate->server->stats.n_connections)
1000) / candidate->weight &&
inst->servers[i]->server->stats.n_connections <
candidate->server->stats.n_connections)
{ {
/* This running server has the same number /* This running server has the same number of connections currently as the candidate
of connections currently as the candidate but has had fewer connections over time than candidate, set this server to
but has had fewer connections over time candidate*/
than candidate, set this server to candidate*/ candidate = ref;
candidate = inst->servers[i];
} }
} }
} }
/* There is no candidate server here! /* If we haven't found a proper candidate yet but a master server is available, we'll pick that
* With router_option=slave a master_host could be set, so route traffic there. * with the assumption that it is "better" than a slave.
* Otherwise, just clean up and return NULL
*/ */
if (!candidate) if (!candidate)
{ {
@ -484,9 +425,8 @@ newSession(ROUTER *instance, SESSION *session)
} }
else else
{ {
MXS_ERROR("Failed to create new routing session. " MXS_ERROR("Failed to create new routing session. Couldn't find eligible"
"Couldn't find eligible candidate server. Freeing " " candidate server. Freeing allocated resources.");
"allocated resources.");
MXS_FREE(client_rses); MXS_FREE(client_rses);
return NULL; return NULL;
} }
@ -496,48 +436,32 @@ newSession(ROUTER *instance, SESSION *session)
* We now have the server with the least connections. * We now have the server with the least connections.
* Bump the connection count for this server * Bump the connection count for this server
*/ */
atomic_add(&candidate->current_connection_count, 1);
client_rses->backend = candidate; client_rses->backend = candidate;
MXS_DEBUG("%lu [newSession] Selected server in port %d. "
"Connections : %d\n",
pthread_self(),
candidate->server->port,
candidate->current_connection_count);
/* /** Open the backend connection */
* Open a backend connection, putting the DCB for this client_rses->backend_dcb = dcb_connect(candidate->server, session,
* connection in the client_rses->backend_dcb
*/
client_rses->backend_dcb = dcb_connect(candidate->server,
session,
candidate->server->protocol); candidate->server->protocol);
if (client_rses->backend_dcb == NULL) if (client_rses->backend_dcb == NULL)
{ {
atomic_add(&candidate->current_connection_count, -1); /** The failure is reported in dcb_connect() */
MXS_FREE(client_rses); MXS_FREE(client_rses);
return NULL; return NULL;
} }
dcb_add_callback(
client_rses->backend_dcb, atomic_add(&candidate->connections, 1);
// TODO: Remove this as it is never called
dcb_add_callback(client_rses->backend_dcb,
DCB_REASON_NOT_RESPONDING, DCB_REASON_NOT_RESPONDING,
&handle_state_switch, &handle_state_switch,
client_rses); client_rses);
inst->stats.n_sessions++; inst->stats.n_sessions++;
/**
* Add this session to the list of active sessions.
*/
spinlock_acquire(&inst->lock);
client_rses->next = inst->connections;
inst->connections = client_rses;
spinlock_release(&inst->lock);
CHK_CLIENT_RSES(client_rses); CHK_CLIENT_RSES(client_rses);
MXS_INFO("Readconnroute: New session for server %s. " MXS_INFO("Readconnroute: New session for server %s. Connections : %d",
"Connections : %d", candidate->server->unique_name, candidate->connections);
candidate->server->unique_name,
candidate->current_connection_count);
return(void *) client_rses; return(void *) client_rses;
} }
@ -562,42 +486,11 @@ newSession(ROUTER *instance, SESSION *session)
static void freeSession(ROUTER* router_instance, void* router_client_ses) static void freeSession(ROUTER* router_instance, void* router_client_ses)
{ {
ROUTER_INSTANCE* router = (ROUTER_INSTANCE *) router_instance; ROUTER_INSTANCE* router = (ROUTER_INSTANCE *) router_instance;
ROUTER_CLIENT_SES* router_cli_ses = ROUTER_CLIENT_SES* router_cli_ses = (ROUTER_CLIENT_SES *) router_client_ses;
(ROUTER_CLIENT_SES *) router_client_ses;
ss_debug(int prev_val = ) atomic_add(&router_cli_ses->backend->current_connection_count, -1); ss_debug(int prev_val = ) atomic_add(&router_cli_ses->backend->connections, -1);
ss_dassert(prev_val > 0); ss_dassert(prev_val > 0);
spinlock_acquire(&router->lock);
if (router->connections == router_cli_ses)
{
router->connections = router_cli_ses->next;
}
else
{
ROUTER_CLIENT_SES *ptr = router->connections;
while (ptr != NULL && ptr->next != router_cli_ses)
{
ptr = ptr->next;
}
if (ptr != NULL)
{
ptr->next = router_cli_ses->next;
}
}
spinlock_release(&router->lock);
MXS_DEBUG("%lu [freeSession] Unlinked router_client_session %p from "
"router %p and from server on port %d. Connections : %d. ",
pthread_self(),
router_cli_ses,
router,
router_cli_ses->backend->server->port,
prev_val - 1);
MXS_FREE(router_cli_ses); MXS_FREE(router_cli_ses);
} }
@ -639,6 +532,29 @@ closeSession(ROUTER *instance, void *router_session)
} }
} }
/** Log routing failure due to closed session */
static void log_closed_session(mysql_server_cmd_t mysql_command, bool is_closed,
SERVER_REF *ref)
{
char msg[MAX_SERVER_NAME_LEN + 200] = ""; // Extra space for message
if (is_closed)
{
sprintf(msg, "Session is closed.");
}
else if (SERVER_IS_DOWN(ref->server))
{
sprintf(msg, "Server '%s' is down.", ref->server->unique_name);
}
else if (!SERVER_REF_IS_ACTIVE(ref))
{
sprintf(msg, "Server '%s' was removed from the service.", ref->server->unique_name);
}
MXS_ERROR("Failed to route MySQL command %d to backend server. %s",
mysql_command, msg);
}
/** /**
* We have data from the client, we must route it to the backend. * We have data from the client, we must route it to the backend.
* This is simply a case of sending it to the connection that was * This is simply a case of sending it to the connection that was
@ -654,7 +570,7 @@ routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
{ {
ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *) instance; ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *) instance;
ROUTER_CLIENT_SES *router_cli_ses = (ROUTER_CLIENT_SES *) router_session; ROUTER_CLIENT_SES *router_cli_ses = (ROUTER_CLIENT_SES *) router_session;
int rc; int rc = 0;
DCB* backend_dcb; DCB* backend_dcb;
MySQLProtocol *proto = (MySQLProtocol*)router_cli_ses->client_dcb->protocol; MySQLProtocol *proto = (MySQLProtocol*)router_cli_ses->client_dcb->protocol;
mysql_server_cmd_t mysql_command = proto->current_command; mysql_server_cmd_t mysql_command = proto->current_command;
@ -683,16 +599,11 @@ routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
} }
if (rses_is_closed || backend_dcb == NULL || if (rses_is_closed || backend_dcb == NULL ||
!SERVER_REF_IS_ACTIVE(router_cli_ses->backend) ||
SERVER_IS_DOWN(router_cli_ses->backend->server)) SERVER_IS_DOWN(router_cli_ses->backend->server))
{ {
MXS_ERROR("Failed to route MySQL command %d to backend " log_closed_session(mysql_command, rses_is_closed, router_cli_ses->backend);
"server.%s", gwbuf_free(queue);
mysql_command, rses_is_closed ? " Session is closed." : "");
rc = 0;
while ((queue = GWBUF_CONSUME_ALL(queue)) != NULL)
{
;
}
goto return_rc; goto return_rc;
} }
@ -737,23 +648,12 @@ static void
diagnostics(ROUTER *router, DCB *dcb) diagnostics(ROUTER *router, DCB *dcb)
{ {
ROUTER_INSTANCE *router_inst = (ROUTER_INSTANCE *) router; ROUTER_INSTANCE *router_inst = (ROUTER_INSTANCE *) router;
ROUTER_CLIENT_SES *session;
int i = 0;
BACKEND *backend;
char *weightby; char *weightby;
spinlock_acquire(&router_inst->lock);
session = router_inst->connections;
while (session)
{
i++;
session = session->next;
}
spinlock_release(&router_inst->lock);
dcb_printf(dcb, "\tNumber of router sessions: %d\n", dcb_printf(dcb, "\tNumber of router sessions: %d\n",
router_inst->stats.n_sessions); router_inst->stats.n_sessions);
dcb_printf(dcb, "\tCurrent no. of router sessions: %d\n", i); dcb_printf(dcb, "\tCurrent no. of router sessions: %d\n",
router_inst->service->stats.n_current);
dcb_printf(dcb, "\tNumber of queries forwarded: %d\n", dcb_printf(dcb, "\tNumber of queries forwarded: %d\n",
router_inst->stats.n_queries); router_inst->stats.n_queries);
if ((weightby = serviceGetWeightingParameter(router_inst->service)) if ((weightby = serviceGetWeightingParameter(router_inst->service))
@ -764,15 +664,13 @@ diagnostics(ROUTER *router, DCB *dcb)
weightby); weightby);
dcb_printf(dcb, dcb_printf(dcb,
"\t\tServer Target %% Connections\n"); "\t\tServer Target %% Connections\n");
for (i = 0; router_inst->servers[i]; i++) for (SERVER_REF *ref = router_inst->service->dbref; ref; ref = ref->next)
{ {
backend = router_inst->servers[i];
dcb_printf(dcb, "\t\t%-20s %3.1f%% %d\n", dcb_printf(dcb, "\t\t%-20s %3.1f%% %d\n",
backend->server->unique_name, ref->server->unique_name,
(float) backend->weight / 10, (float) ref->weight / 10,
backend->current_connection_count); ref->connections);
} }
} }
} }
@ -933,28 +831,28 @@ static uint64_t getCapabilities(void)
* *
*/ */
static BACKEND *get_root_master(BACKEND **servers) static SERVER_REF *get_root_master(SERVER_REF *servers)
{ {
int i = 0; int i = 0;
BACKEND *master_host = NULL; SERVER_REF *master_host = NULL;
for (i = 0; servers[i]; i++) for (SERVER_REF *ref = servers; ref; ref = ref->next)
{ {
if (servers[i] && (servers[i]->server->status & (SERVER_MASTER | SERVER_MAINT)) == SERVER_MASTER) if (ref->active && SERVER_IS_MASTER(ref->server))
{ {
if (master_host == NULL) if (master_host == NULL)
{ {
master_host = servers[i]; master_host = ref;
} }
else if (servers[i]->server->depth < master_host->server->depth || else if (ref->server->depth < master_host->server->depth ||
(servers[i]->server->depth == master_host->server->depth && (ref->server->depth == master_host->server->depth &&
servers[i]->weight > master_host->weight)) ref->weight > master_host->weight))
{ {
/** /**
* This master has a lower depth than the candidate master or * This master has a lower depth than the candidate master or
* the depths are equal but this master has a higher weight * the depths are equal but this master has a higher weight
*/ */
master_host = servers[i]; master_host = ref;
} }
} }
} }