e54cc95a20
With this change, if two master servers both have equal depths but different weights, the one with the higher weight is used. If the depths and weights are equal, the first master listed in the configuration is used.
1082 lines
32 KiB
C
1082 lines
32 KiB
C
/*
|
|
* Copyright (c) 2016 MariaDB Corporation Ab
|
|
*
|
|
* Use of this software is governed by the Business Source License included
|
|
* in the LICENSE.TXT file and at www.mariadb.com/bsl.
|
|
*
|
|
* Change Date: 2019-01-01
|
|
*
|
|
* On the date above, in accordance with the Business Source License, use
|
|
* of this software will be governed by version 2 or later of the General
|
|
* Public License.
|
|
*/
|
|
|
|
/**
|
|
* @file readconnroute.c - Read Connection Load Balancing Query Router
|
|
*
|
|
* This is the implementation of a simple query router that balances
|
|
* read connections. It assumes the service is configured with a set
|
|
* of slaves and that the application clients already split read and write
|
|
* queries. It offers a service to balance the client read connections
|
|
* over this set of slave servers. It does this once only, at the time
|
|
* the connection is made. It chooses the server that currently has the least
|
|
* number of connections by keeping a count for each server of how
|
|
* many connections the query router has made to the server.
|
|
*
|
|
* When two servers have the same number of current connections the one with
|
|
* the least number of connections since startup will be used.
|
|
*
|
|
* The router may also have options associated to it that will limit the
|
|
* choice of backend server. Currently two options are supported, the "master"
|
|
* option will cause the router to only connect to servers marked as masters
|
|
* and the "slave" option will limit connections to routers that are marked
|
|
* as slaves. If neither option is specified the router will connect to either
|
|
* masters or slaves.
|
|
*
|
|
* @verbatim
|
|
* Revision History
|
|
*
|
|
* Date Who Description
|
|
* 14/06/2013 Mark Riddoch Initial implementation
|
|
* 25/06/2013 Mark Riddoch Addition of checks for current server state
|
|
* 26/06/2013 Mark Riddoch Use server with least connections since
|
|
* startup if the number of current
|
|
* connections is the same for two servers
|
|
* Addition of master and slave options
|
|
* 27/06/2013 Vilho Raatikka Added skygw_log_write command as an example
|
|
* and necessary headers.
|
|
* 17/07/2013 Massimiliano Pinto Added clientReply routine:
|
|
* called by backend server to send data to client
|
|
* Included mysql_client_server_protocol.h
|
|
* with macros and MySQL commands with MYSQL_ prefix
|
|
* avoiding any conflict with the standard ones
|
|
* in mysql.h
|
|
* 22/07/2013 Mark Riddoch Addition of joined router option for Galera
|
|
* clusters
|
|
* 31/07/2013 Massimiliano Pinto Added a check for candidate server, if NULL return
|
|
* 12/08/2013 Mark Riddoch Log unsupported router options
|
|
* 04/09/2013 Massimiliano Pinto Added client NULL check in clientReply
|
|
* 22/10/2013 Massimiliano Pinto errorReply called from backend, for client error reply
|
|
* or take different actions such as open a new backend connection
|
|
* 20/02/2014 Massimiliano Pinto If router_options=slave, route traffic to master if no slaves available
|
|
* 06/03/2014 Massimiliano Pinto Server connection counter is now updated in closeSession
|
|
* 24/06/2014 Massimiliano Pinto New rules for selecting the Master server
|
|
* 27/06/2014 Mark Riddoch Addition of server weighting
|
|
* 11/06/2015 Martin Brampton Remove decrement n_current (moved to dcb.c)
|
|
* 09/09/2015 Martin Brampton Modify error handler
|
|
* 25/09/2015 Martin Brampton Block callback processing when no router session in the DCB
|
|
* 09/11/2015 Martin Brampton Modified routeQuery - must free "queue" regardless of outcome
|
|
*
|
|
* @endverbatim
|
|
*/
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <string.h>
|
|
#include <signal.h>
|
|
#include <service.h>
|
|
#include <server.h>
|
|
#include <router.h>
|
|
#include <atomic.h>
|
|
#include <spinlock.h>
|
|
#include <readconnection.h>
|
|
#include <dcb.h>
|
|
#include <spinlock.h>
|
|
#include <modinfo.h>
|
|
|
|
#include <skygw_types.h>
|
|
#include <skygw_utils.h>
|
|
#include <log_manager.h>
|
|
|
|
#include <mysql_client_server_protocol.h>
|
|
|
|
#include "modutil.h"
|
|
|
|
MODULE_INFO info =
|
|
{
|
|
MODULE_API_ROUTER,
|
|
MODULE_GA,
|
|
ROUTER_VERSION,
|
|
"A connection based router to load balance based on connections"
|
|
};
|
|
|
|
static char *version_str = "V1.1.0";
|
|
|
|
/* The router entry points */
|
|
static ROUTER *createInstance(SERVICE *service, char **options);
|
|
static void *newSession(ROUTER *instance, SESSION *session);
|
|
static void closeSession(ROUTER *instance, void *router_session);
|
|
static void freeSession(ROUTER *instance, void *router_session);
|
|
static int routeQuery(ROUTER *instance, void *router_session, GWBUF *queue);
|
|
static void diagnostics(ROUTER *instance, DCB *dcb);
|
|
static void clientReply(ROUTER *instance, void *router_session, GWBUF *queue,
|
|
DCB *backend_dcb);
|
|
static void handleError(ROUTER *instance, void *router_session, GWBUF *errbuf,
|
|
DCB *problem_dcb, error_action_t action, bool *succp);
|
|
static int getCapabilities();
|
|
|
|
|
|
/** The module object definition */
|
|
static ROUTER_OBJECT MyObject =
|
|
{
|
|
createInstance,
|
|
newSession,
|
|
closeSession,
|
|
freeSession,
|
|
routeQuery,
|
|
diagnostics,
|
|
clientReply,
|
|
handleError,
|
|
getCapabilities
|
|
};
|
|
|
|
static bool rses_begin_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 int handle_state_switch(DCB* dcb, DCB_REASON reason, void * routersession);
|
|
static SPINLOCK instlock;
|
|
static ROUTER_INSTANCE *instances;
|
|
|
|
/**
|
|
* Implementation of the mandatory version entry point
|
|
*
|
|
* @return version string of the module
|
|
*/
|
|
char *
|
|
version()
|
|
{
|
|
return version_str;
|
|
}
|
|
|
|
/**
|
|
* The module initialisation routine, called when the module
|
|
* is first loaded.
|
|
*/
|
|
void
|
|
ModuleInit()
|
|
{
|
|
MXS_NOTICE("Initialise readconnroute router module %s.", version_str);
|
|
spinlock_init(&instlock);
|
|
instances = NULL;
|
|
}
|
|
|
|
/**
|
|
* The module entry point routine. It is this routine that
|
|
* must populate the structure that is referred to as the
|
|
* "module object", this is a structure with the set of
|
|
* external entry points for this module.
|
|
*
|
|
* @return The module object
|
|
*/
|
|
ROUTER_OBJECT *
|
|
GetModuleObject()
|
|
{
|
|
return &MyObject;
|
|
}
|
|
|
|
static inline void free_readconn_instance(ROUTER_INSTANCE *router)
|
|
{
|
|
if (router)
|
|
{
|
|
if (router->servers)
|
|
{
|
|
for (int i = 0; router->servers[i]; i++)
|
|
{
|
|
free(router->servers[i]);
|
|
}
|
|
}
|
|
free(router->servers);
|
|
free(router);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Create an instance of the router for a particular service
|
|
* within the gateway.
|
|
*
|
|
* @param service The service this router is being create for
|
|
* @param options An array of options for this query router
|
|
*
|
|
* @return The instance data for this new instance
|
|
*/
|
|
static ROUTER *
|
|
createInstance(SERVICE *service, char **options)
|
|
{
|
|
ROUTER_INSTANCE *inst;
|
|
SERVER *server;
|
|
SERVER_REF *sref;
|
|
int i, n;
|
|
BACKEND *backend;
|
|
char *weightby;
|
|
|
|
if ((inst = calloc(1, sizeof(ROUTER_INSTANCE))) == NULL)
|
|
{
|
|
return NULL;
|
|
}
|
|
|
|
inst->service = service;
|
|
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 **) 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] = 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 = 1000;
|
|
n++;
|
|
}
|
|
inst->servers[n] = NULL;
|
|
|
|
if ((weightby = serviceGetWeightingParameter(service)) != NULL)
|
|
{
|
|
int total = 0;
|
|
|
|
for (int n = 0; inst->servers[n]; n++)
|
|
{
|
|
BACKEND *backend = inst->servers[n];
|
|
char *param = serverGetParameter(backend->server, weightby);
|
|
if (param)
|
|
{
|
|
total += atoi(param);
|
|
}
|
|
}
|
|
if (total == 0)
|
|
{
|
|
MXS_WARNING("Weighting Parameter for service '%s' "
|
|
"will be ignored as no servers have values "
|
|
"for the parameter '%s'.",
|
|
service->name, weightby);
|
|
}
|
|
else if (total < 0)
|
|
{
|
|
MXS_ERROR("Sum of weighting parameter '%s' for service '%s' exceeds "
|
|
"maximum value of %d. Weighting will be ignored.",
|
|
weightby, service->name, INT_MAX);
|
|
}
|
|
else
|
|
{
|
|
for (int n = 0; inst->servers[n]; n++)
|
|
{
|
|
BACKEND *backend = inst->servers[n];
|
|
char *param = serverGetParameter(backend->server, weightby);
|
|
if (param)
|
|
{
|
|
int wght = atoi(param);
|
|
int perc = (wght * 1000) / total;
|
|
|
|
if (perc == 0)
|
|
{
|
|
perc = 1;
|
|
MXS_ERROR("Weighting parameter '%s' with a value of %d for"
|
|
" server '%s' rounds down to zero with total weight"
|
|
" of %d for service '%s'. No queries will be "
|
|
"routed to this server.", weightby, wght,
|
|
backend->server->unique_name, total,
|
|
service->name);
|
|
}
|
|
else if (perc < 0)
|
|
{
|
|
MXS_ERROR("Weighting parameter '%s' for server '%s' is too large, "
|
|
"maximum value is %d. No weighting will be used for this server.",
|
|
weightby, backend->server->unique_name, INT_MAX / 1000);
|
|
perc = 1000;
|
|
}
|
|
backend->weight = perc;
|
|
}
|
|
else
|
|
{
|
|
MXS_WARNING("Server '%s' has no parameter '%s' used for weighting"
|
|
" for service '%s'.", backend->server->unique_name,
|
|
weightby, service->name);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Process the options
|
|
*/
|
|
bool error = false;
|
|
inst->bitmask = 0;
|
|
inst->bitvalue = 0;
|
|
if (options)
|
|
{
|
|
for (i = 0; options[i]; i++)
|
|
{
|
|
if (!strcasecmp(options[i], "master"))
|
|
{
|
|
inst->bitmask |= (SERVER_MASTER | SERVER_SLAVE);
|
|
inst->bitvalue |= SERVER_MASTER;
|
|
}
|
|
else if (!strcasecmp(options[i], "slave"))
|
|
{
|
|
inst->bitmask |= (SERVER_MASTER | SERVER_SLAVE);
|
|
inst->bitvalue |= SERVER_SLAVE;
|
|
}
|
|
else if (!strcasecmp(options[i], "running"))
|
|
{
|
|
inst->bitmask |= (SERVER_RUNNING);
|
|
inst->bitvalue |= SERVER_RUNNING;
|
|
}
|
|
else if (!strcasecmp(options[i], "synced"))
|
|
{
|
|
inst->bitmask |= (SERVER_JOINED);
|
|
inst->bitvalue |= SERVER_JOINED;
|
|
}
|
|
else if (!strcasecmp(options[i], "ndb"))
|
|
{
|
|
inst->bitmask |= (SERVER_NDB);
|
|
inst->bitvalue |= SERVER_NDB;
|
|
}
|
|
else
|
|
{
|
|
MXS_WARNING("Unsupported router "
|
|
"option \'%s\' for readconnroute. "
|
|
"Expected router options are "
|
|
"[slave|master|synced|ndb|running]",
|
|
options[i]);
|
|
error = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
if (error)
|
|
{
|
|
free_readconn_instance(inst);
|
|
return NULL;
|
|
}
|
|
|
|
if (inst->bitmask == 0 && inst->bitvalue == 0)
|
|
{
|
|
/** No parameters given, use RUNNING as a valid server */
|
|
inst->bitmask |= (SERVER_RUNNING);
|
|
inst->bitvalue |= SERVER_RUNNING;
|
|
}
|
|
/*
|
|
* We have completed the creation of the instance data, so now
|
|
* insert this router instance into the linked list of routers
|
|
* that have been created with this module.
|
|
*/
|
|
spinlock_acquire(&instlock);
|
|
inst->next = instances;
|
|
instances = inst;
|
|
spinlock_release(&instlock);
|
|
|
|
return(ROUTER *) inst;
|
|
}
|
|
|
|
/**
|
|
* Associate a new session with this instance of the router.
|
|
*
|
|
* @param instance The router instance data
|
|
* @param session The session itself
|
|
* @return Session specific data for this session
|
|
*/
|
|
static void *
|
|
newSession(ROUTER *instance, SESSION *session)
|
|
{
|
|
ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *) instance;
|
|
ROUTER_CLIENT_SES *client_rses;
|
|
BACKEND *candidate = NULL;
|
|
int i;
|
|
BACKEND *master_host = NULL;
|
|
|
|
MXS_DEBUG("%lu [newSession] new router session with session "
|
|
"%p, and inst %p.",
|
|
pthread_self(),
|
|
session,
|
|
inst);
|
|
|
|
|
|
client_rses = (ROUTER_CLIENT_SES *) calloc(1, sizeof(ROUTER_CLIENT_SES));
|
|
|
|
if (client_rses == NULL)
|
|
{
|
|
return NULL;
|
|
}
|
|
|
|
#if defined(SS_DEBUG)
|
|
client_rses->rses_chk_top = CHK_NUM_ROUTER_SES;
|
|
client_rses->rses_chk_tail = CHK_NUM_ROUTER_SES;
|
|
#endif
|
|
client_rses->client_dcb = session->client_dcb;
|
|
|
|
/**
|
|
* Find the Master host from available servers
|
|
*/
|
|
master_host = get_root_master(inst->servers);
|
|
|
|
/**
|
|
* Find a backend server to connect to. This is the extent of the
|
|
* load balancing algorithm we need to implement for this simple
|
|
* connection router.
|
|
*/
|
|
|
|
/*
|
|
* Loop over all the servers and find any that have fewer connections
|
|
* than the candidate server.
|
|
*
|
|
* If a server has less connections than the current candidate we mark this
|
|
* as the new candidate to connect to.
|
|
*
|
|
* If a server has the same number of connections currently as the candidate
|
|
* and has had less connections over time than the candidate it will also
|
|
* become the new candidate. This has the effect of spreading the
|
|
* connections over different servers during periods of very low load.
|
|
*/
|
|
for (i = 0; inst->servers[i]; i++)
|
|
{
|
|
if (inst->servers[i])
|
|
{
|
|
MXS_DEBUG("%lu [newSession] Examine server in port %d with "
|
|
"%d connections. Status is %s, "
|
|
"inst->bitvalue is %d",
|
|
pthread_self(),
|
|
inst->servers[i]->server->port,
|
|
inst->servers[i]->current_connection_count,
|
|
STRSRVSTATUS(inst->servers[i]->server),
|
|
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 */
|
|
if (inst->servers[i] &&
|
|
SERVER_IS_RUNNING(inst->servers[i]->server) &&
|
|
(inst->servers[i]->server->status & inst->bitmask & inst->bitvalue))
|
|
{
|
|
if (master_host)
|
|
{
|
|
if (inst->servers[i] == master_host && (inst->bitvalue & SERVER_SLAVE))
|
|
{
|
|
/* skip root Master here, as it could also be slave of an external server
|
|
* that is not in the configuration.
|
|
* Intermediate masters (Relay Servers) are also slave and will be selected
|
|
* as Slave(s)
|
|
*/
|
|
|
|
continue;
|
|
}
|
|
if (inst->servers[i] == master_host && (inst->bitvalue & SERVER_MASTER))
|
|
{
|
|
/* If option is "master" return only the root Master as there
|
|
* could be intermediate masters (Relay Servers)
|
|
* and they must not be selected.
|
|
*/
|
|
|
|
candidate = master_host;
|
|
break;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
/* master_host is NULL, no master server.
|
|
* If requested router_option is 'master'
|
|
* candidate wll be NULL.
|
|
*/
|
|
if (inst->bitvalue & SERVER_MASTER)
|
|
{
|
|
candidate = NULL;
|
|
break;
|
|
}
|
|
}
|
|
|
|
/* If no candidate set, set first running server as
|
|
our initial candidate server */
|
|
if (candidate == NULL)
|
|
{
|
|
candidate = inst->servers[i];
|
|
}
|
|
else if (((inst->servers[i]->current_connection_count + 1)
|
|
* 1000) / inst->servers[i]->weight <
|
|
((candidate->current_connection_count + 1) *
|
|
1000) / candidate->weight)
|
|
{
|
|
/* This running server has fewer
|
|
connections, set it as a new candidate */
|
|
candidate = inst->servers[i];
|
|
}
|
|
else if (((inst->servers[i]->current_connection_count + 1)
|
|
* 1000) / inst->servers[i]->weight ==
|
|
((candidate->current_connection_count + 1) *
|
|
1000) / candidate->weight &&
|
|
inst->servers[i]->server->stats.n_connections <
|
|
candidate->server->stats.n_connections)
|
|
{
|
|
/* This running server has the same number
|
|
of connections currently as the candidate
|
|
but has had fewer connections over time
|
|
than candidate, set this server to candidate*/
|
|
candidate = inst->servers[i];
|
|
}
|
|
}
|
|
}
|
|
|
|
/* There is no candidate server here!
|
|
* With router_option=slave a master_host could be set, so route traffic there.
|
|
* Otherwise, just clean up and return NULL
|
|
*/
|
|
if (!candidate)
|
|
{
|
|
if (master_host)
|
|
{
|
|
candidate = master_host;
|
|
}
|
|
else
|
|
{
|
|
MXS_ERROR("Failed to create new routing session. "
|
|
"Couldn't find eligible candidate server. Freeing "
|
|
"allocated resources.");
|
|
free(client_rses);
|
|
return NULL;
|
|
}
|
|
}
|
|
|
|
client_rses->rses_capabilities = RCAP_TYPE_PACKET_INPUT;
|
|
|
|
/*
|
|
* We now have the server with the least connections.
|
|
* Bump the connection count for this server
|
|
*/
|
|
atomic_add(&candidate->current_connection_count, 1);
|
|
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 a backend connection, putting the DCB for this
|
|
* connection in the client_rses->backend_dcb
|
|
*/
|
|
client_rses->backend_dcb = dcb_connect(candidate->server,
|
|
session,
|
|
candidate->server->protocol);
|
|
if (client_rses->backend_dcb == NULL)
|
|
{
|
|
atomic_add(&candidate->current_connection_count, -1);
|
|
free(client_rses);
|
|
return NULL;
|
|
}
|
|
dcb_add_callback(
|
|
client_rses->backend_dcb,
|
|
DCB_REASON_NOT_RESPONDING,
|
|
&handle_state_switch,
|
|
client_rses);
|
|
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);
|
|
|
|
MXS_INFO("Readconnroute: New session for server %s. "
|
|
"Connections : %d",
|
|
candidate->server->unique_name,
|
|
candidate->current_connection_count);
|
|
|
|
return(void *) client_rses;
|
|
}
|
|
|
|
/**
|
|
* @node Unlink from backend server, unlink from router's connection list,
|
|
* and free memory of a router client session.
|
|
*
|
|
* Parameters:
|
|
* @param router - <usage>
|
|
* <description>
|
|
*
|
|
* @param router_cli_ses - <usage>
|
|
* <description>
|
|
*
|
|
* @return void
|
|
*
|
|
*
|
|
* @details (write detailed description here)
|
|
*
|
|
*/
|
|
static void freeSession(ROUTER* router_instance, void* router_client_ses)
|
|
{
|
|
ROUTER_INSTANCE* router = (ROUTER_INSTANCE *) router_instance;
|
|
ROUTER_CLIENT_SES* router_cli_ses =
|
|
(ROUTER_CLIENT_SES *) router_client_ses;
|
|
int prev_val;
|
|
|
|
prev_val = atomic_add(&router_cli_ses->backend->current_connection_count, -1);
|
|
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);
|
|
|
|
free(router_cli_ses);
|
|
}
|
|
|
|
/**
|
|
* Close a session with the router, this is the mechanism
|
|
* by which a router may cleanup data structure etc.
|
|
*
|
|
* @param instance The router instance data
|
|
* @param router_session The session being closed
|
|
*/
|
|
static void
|
|
closeSession(ROUTER *instance, void *router_session)
|
|
{
|
|
ROUTER_CLIENT_SES *router_cli_ses = (ROUTER_CLIENT_SES *) router_session;
|
|
DCB* backend_dcb;
|
|
|
|
CHK_CLIENT_RSES(router_cli_ses);
|
|
/**
|
|
* Lock router client session for secure read and update.
|
|
*/
|
|
if (rses_begin_locked_router_action(router_cli_ses))
|
|
{
|
|
/* decrease server current connection counter */
|
|
|
|
backend_dcb = router_cli_ses->backend_dcb;
|
|
router_cli_ses->backend_dcb = NULL;
|
|
router_cli_ses->rses_closed = true;
|
|
/** Unlock */
|
|
rses_end_locked_router_action(router_cli_ses);
|
|
|
|
/**
|
|
* Close the backend server connection
|
|
*/
|
|
if (backend_dcb != NULL)
|
|
{
|
|
CHK_DCB(backend_dcb);
|
|
dcb_close(backend_dcb);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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
|
|
* chosen when we started the client session.
|
|
*
|
|
* @param instance The router instance
|
|
* @param router_session The router session returned from the newSession call
|
|
* @param queue The queue of data buffers to route
|
|
* @return if succeed 1, otherwise 0
|
|
*/
|
|
static int
|
|
routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
|
|
{
|
|
ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *) instance;
|
|
ROUTER_CLIENT_SES *router_cli_ses = (ROUTER_CLIENT_SES *) router_session;
|
|
int rc;
|
|
DCB* backend_dcb;
|
|
MySQLProtocol *proto = (MySQLProtocol*)router_cli_ses->client_dcb->protocol;
|
|
mysql_server_cmd_t mysql_command = proto->current_command;
|
|
bool rses_is_closed;
|
|
|
|
inst->stats.n_queries++;
|
|
|
|
/** Dirty read for quick check if router is closed. */
|
|
if (router_cli_ses->rses_closed)
|
|
{
|
|
rses_is_closed = true;
|
|
}
|
|
else
|
|
{
|
|
/**
|
|
* Lock router client session for secure read of DCBs
|
|
*/
|
|
rses_is_closed = !(rses_begin_locked_router_action(router_cli_ses));
|
|
}
|
|
|
|
if (!rses_is_closed)
|
|
{
|
|
backend_dcb = router_cli_ses->backend_dcb;
|
|
/** unlock */
|
|
rses_end_locked_router_action(router_cli_ses);
|
|
}
|
|
|
|
if (rses_is_closed || backend_dcb == NULL ||
|
|
SERVER_IS_DOWN(router_cli_ses->backend->server))
|
|
{
|
|
MXS_ERROR("Failed to route MySQL command %d to backend "
|
|
"server.%s",
|
|
mysql_command, rses_is_closed ? " Session is closed." : "");
|
|
rc = 0;
|
|
while ((queue = GWBUF_CONSUME_ALL(queue)) != NULL)
|
|
{
|
|
;
|
|
}
|
|
goto return_rc;
|
|
|
|
}
|
|
|
|
char* trc = NULL;
|
|
|
|
switch (mysql_command)
|
|
{
|
|
case MYSQL_COM_CHANGE_USER:
|
|
rc = backend_dcb->func.auth(backend_dcb, NULL, backend_dcb->session,
|
|
queue);
|
|
break;
|
|
case MYSQL_COM_QUERY:
|
|
if (MXS_LOG_PRIORITY_IS_ENABLED(LOG_INFO))
|
|
{
|
|
trc = modutil_get_SQL(queue);
|
|
}
|
|
default:
|
|
rc = backend_dcb->func.write(backend_dcb, queue);
|
|
break;
|
|
}
|
|
|
|
MXS_INFO("Routed [%s] to '%s'%s%s",
|
|
STRPACKETTYPE(mysql_command),
|
|
backend_dcb->server->unique_name,
|
|
trc ? ": " : ".",
|
|
trc ? trc : "");
|
|
free(trc);
|
|
|
|
return_rc:
|
|
|
|
return rc;
|
|
}
|
|
|
|
/**
|
|
* Display router diagnostics
|
|
*
|
|
* @param instance Instance of the router
|
|
* @param dcb DCB to send diagnostics to
|
|
*/
|
|
static void
|
|
diagnostics(ROUTER *router, DCB *dcb)
|
|
{
|
|
ROUTER_INSTANCE *router_inst = (ROUTER_INSTANCE *) router;
|
|
ROUTER_CLIENT_SES *session;
|
|
int i = 0;
|
|
BACKEND *backend;
|
|
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",
|
|
router_inst->stats.n_sessions);
|
|
dcb_printf(dcb, "\tCurrent no. of router sessions: %d\n", i);
|
|
dcb_printf(dcb, "\tNumber of queries forwarded: %d\n",
|
|
router_inst->stats.n_queries);
|
|
if ((weightby = serviceGetWeightingParameter(router_inst->service))
|
|
!= NULL)
|
|
{
|
|
dcb_printf(dcb, "\tConnection distribution based on %s "
|
|
"server parameter.\n",
|
|
weightby);
|
|
dcb_printf(dcb,
|
|
"\t\tServer Target %% Connections\n");
|
|
for (i = 0; router_inst->servers[i]; i++)
|
|
{
|
|
backend = router_inst->servers[i];
|
|
dcb_printf(dcb, "\t\t%-20s %3.1f%% %d\n",
|
|
backend->server->unique_name,
|
|
(float) backend->weight / 10,
|
|
backend->current_connection_count);
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Client Reply routine
|
|
*
|
|
* The routine will reply to client data from backend server
|
|
*
|
|
* @param instance The router instance
|
|
* @param router_session The router session
|
|
* @param backend_dcb The backend DCB
|
|
* @param queue The GWBUF with reply data
|
|
*/
|
|
static void
|
|
clientReply(ROUTER *instance, void *router_session, GWBUF *queue, DCB *backend_dcb)
|
|
{
|
|
ss_dassert(backend_dcb->session->client_dcb != NULL);
|
|
SESSION_ROUTE_REPLY(backend_dcb->session, queue);
|
|
}
|
|
|
|
/**
|
|
* Error Handler routine
|
|
*
|
|
* The routine will handle errors that occurred in writes.
|
|
*
|
|
* @param instance The router instance
|
|
* @param router_session The router session
|
|
* @param message The error message to reply
|
|
* @param problem_dcb The DCB related to the error
|
|
* @param action The action: ERRACT_NEW_CONNECTION or ERRACT_REPLY_CLIENT
|
|
* @param succp Result of action: true if router can continue
|
|
*
|
|
*/
|
|
static void handleError(ROUTER *instance, void *router_session, GWBUF *errbuf,
|
|
DCB *problem_dcb, error_action_t action, bool *succp)
|
|
|
|
{
|
|
DCB *client_dcb;
|
|
SESSION *session = problem_dcb->session;
|
|
session_state_t sesstate;
|
|
ROUTER_CLIENT_SES *router_cli_ses = (ROUTER_CLIENT_SES *) router_session;
|
|
|
|
/** Don't handle same error twice on same DCB */
|
|
if (problem_dcb->dcb_errhandle_called)
|
|
{
|
|
/** we optimistically assume that previous call succeed */
|
|
*succp = true;
|
|
return;
|
|
}
|
|
else
|
|
{
|
|
problem_dcb->dcb_errhandle_called = true;
|
|
}
|
|
spinlock_acquire(&session->ses_lock);
|
|
sesstate = session->state;
|
|
client_dcb = session->client_dcb;
|
|
|
|
if (sesstate == SESSION_STATE_ROUTER_READY)
|
|
{
|
|
CHK_DCB(client_dcb);
|
|
spinlock_release(&session->ses_lock);
|
|
client_dcb->func.write(client_dcb, gwbuf_clone(errbuf));
|
|
}
|
|
else
|
|
{
|
|
spinlock_release(&session->ses_lock);
|
|
}
|
|
|
|
if (DCB_ROLE_CLIENT_HANDLER == problem_dcb->dcb_role)
|
|
{
|
|
dcb_close(problem_dcb);
|
|
}
|
|
else if (router_cli_ses && problem_dcb == router_cli_ses->backend_dcb)
|
|
{
|
|
router_cli_ses->backend_dcb = NULL;
|
|
dcb_close(problem_dcb);
|
|
}
|
|
|
|
/** false because connection is not available anymore */
|
|
*succp = false;
|
|
}
|
|
|
|
/** to be inline'd */
|
|
|
|
/**
|
|
* @node Acquires lock to router client session if it is not closed.
|
|
*
|
|
* Parameters:
|
|
* @param rses - in, use
|
|
*
|
|
*
|
|
* @return true if router session was not closed. If return value is true
|
|
* it means that router is locked, and must be unlocked later. False, if
|
|
* router was closed before lock was acquired.
|
|
*
|
|
*
|
|
* @details (write detailed description here)
|
|
*
|
|
*/
|
|
static bool rses_begin_locked_router_action(ROUTER_CLIENT_SES* rses)
|
|
{
|
|
bool succp = false;
|
|
|
|
CHK_CLIENT_RSES(rses);
|
|
|
|
if (rses->rses_closed)
|
|
{
|
|
goto return_succp;
|
|
}
|
|
spinlock_acquire(&rses->rses_lock);
|
|
if (rses->rses_closed)
|
|
{
|
|
spinlock_release(&rses->rses_lock);
|
|
goto return_succp;
|
|
}
|
|
succp = true;
|
|
|
|
return_succp:
|
|
return succp;
|
|
}
|
|
|
|
/** to be inline'd */
|
|
|
|
/**
|
|
* @node Releases router client session lock.
|
|
*
|
|
* Parameters:
|
|
* @param rses - <usage>
|
|
* <description>
|
|
*
|
|
* @return void
|
|
*
|
|
*
|
|
* @details (write detailed description here)
|
|
*
|
|
*/
|
|
static void rses_end_locked_router_action(ROUTER_CLIENT_SES* rses)
|
|
{
|
|
CHK_CLIENT_RSES(rses);
|
|
spinlock_release(&rses->rses_lock);
|
|
}
|
|
|
|
static int getCapabilities()
|
|
{
|
|
return RCAP_TYPE_PACKET_INPUT;
|
|
}
|
|
|
|
/********************************
|
|
* This routine returns the root master server from MySQL replication tree
|
|
* Get the root Master rule:
|
|
*
|
|
* find server with the lowest replication depth level
|
|
* and the SERVER_MASTER bitval
|
|
* Servers are checked even if they are in 'maintenance'
|
|
*
|
|
* @param servers The list of servers
|
|
* @return The Master found
|
|
*
|
|
*/
|
|
|
|
static BACKEND *get_root_master(BACKEND **servers)
|
|
{
|
|
int i = 0;
|
|
BACKEND *master_host = NULL;
|
|
|
|
for (i = 0; servers[i]; i++)
|
|
{
|
|
if (servers[i] && (servers[i]->server->status & (SERVER_MASTER | SERVER_MAINT)) == SERVER_MASTER)
|
|
{
|
|
if (master_host == NULL)
|
|
{
|
|
master_host = servers[i];
|
|
}
|
|
else if (servers[i]->server->depth < master_host->server->depth ||
|
|
(servers[i]->server->depth == master_host->server->depth &&
|
|
servers[i]->weight > master_host->weight))
|
|
{
|
|
/**
|
|
* This master has a lower depth than the candidate master or
|
|
* the depths are equal but this master has a higher weight
|
|
*/
|
|
master_host = servers[i];
|
|
}
|
|
}
|
|
}
|
|
return master_host;
|
|
}
|
|
|
|
static int handle_state_switch(DCB* dcb, DCB_REASON reason, void * routersession)
|
|
{
|
|
ss_dassert(dcb != NULL);
|
|
SESSION* session = dcb->session;
|
|
ROUTER_CLIENT_SES* rses = (ROUTER_CLIENT_SES*) routersession;
|
|
SERVICE* service = session->service;
|
|
ROUTER* router = (ROUTER *) service->router;
|
|
|
|
if (NULL == dcb->session->router_session && DCB_REASON_ERROR != reason)
|
|
{
|
|
/*
|
|
* We cannot handle a DCB that does not have a router session,
|
|
* except in the case where error processing is invoked.
|
|
*/
|
|
return 0;
|
|
}
|
|
switch (reason)
|
|
{
|
|
case DCB_REASON_CLOSE:
|
|
dcb->func.close(dcb);
|
|
break;
|
|
case DCB_REASON_DRAINED:
|
|
/** Do we need to do anything? */
|
|
break;
|
|
case DCB_REASON_HIGH_WATER:
|
|
/** Do we need to do anything? */
|
|
break;
|
|
case DCB_REASON_LOW_WATER:
|
|
/** Do we need to do anything? */
|
|
break;
|
|
case DCB_REASON_ERROR:
|
|
dcb->func.error(dcb);
|
|
break;
|
|
case DCB_REASON_HUP:
|
|
dcb->func.hangup(dcb);
|
|
break;
|
|
case DCB_REASON_NOT_RESPONDING:
|
|
dcb->func.hangup(dcb);
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return 0;
|
|
}
|