Merge branch 'develop' into firewall
This commit is contained in:
@ -44,6 +44,10 @@
|
||||
#include <skygw_debug.h>
|
||||
#include <spinlock.h>
|
||||
#include <hint.h>
|
||||
#include <log_manager.h>
|
||||
#include <errno.h>
|
||||
|
||||
extern int lm_enabled_logfiles_bitmask;
|
||||
|
||||
static buffer_object_t* gwbuf_remove_buffer_object(
|
||||
GWBUF* buf,
|
||||
@ -70,22 +74,25 @@ SHARED_BUF *sbuf;
|
||||
/* Allocate the buffer header */
|
||||
if ((rval = (GWBUF *)malloc(sizeof(GWBUF))) == NULL)
|
||||
{
|
||||
return NULL;
|
||||
goto retblock;;
|
||||
}
|
||||
|
||||
/* Allocate the shared data buffer */
|
||||
if ((sbuf = (SHARED_BUF *)malloc(sizeof(SHARED_BUF))) == NULL)
|
||||
{
|
||||
free(rval);
|
||||
return NULL;
|
||||
rval = NULL;
|
||||
goto retblock;
|
||||
}
|
||||
|
||||
/* Allocate the space for the actual data */
|
||||
if ((sbuf->data = (unsigned char *)malloc(size)) == NULL)
|
||||
{
|
||||
ss_dassert(sbuf->data != NULL);
|
||||
free(rval);
|
||||
free(sbuf);
|
||||
return NULL;
|
||||
rval = NULL;
|
||||
goto retblock;
|
||||
}
|
||||
spinlock_init(&rval->gwbuf_lock);
|
||||
rval->start = sbuf->data;
|
||||
@ -100,6 +107,14 @@ SHARED_BUF *sbuf;
|
||||
rval->gwbuf_info = GWBUF_INFO_NONE;
|
||||
rval->gwbuf_bufobj = NULL;
|
||||
CHK_GWBUF(rval);
|
||||
retblock:
|
||||
if (rval == NULL)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Memory allocation failed due to %s.",
|
||||
strerror(errno))));
|
||||
}
|
||||
return rval;
|
||||
}
|
||||
|
||||
@ -163,6 +178,11 @@ GWBUF *rval;
|
||||
|
||||
if ((rval = (GWBUF *)malloc(sizeof(GWBUF))) == NULL)
|
||||
{
|
||||
ss_dassert(rval != NULL);
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Memory allocation failed due to %s.",
|
||||
strerror(errno))));
|
||||
return NULL;
|
||||
}
|
||||
|
||||
@ -194,6 +214,11 @@ GWBUF *gwbuf_clone_portion(
|
||||
|
||||
if ((clonebuf = (GWBUF *)malloc(sizeof(GWBUF))) == NULL)
|
||||
{
|
||||
ss_dassert(clonebuf != NULL);
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Memory allocation failed due to %s.",
|
||||
strerror(errno))));
|
||||
return NULL;
|
||||
}
|
||||
atomic_add(&buf->sbuf->refcount, 1);
|
||||
@ -438,6 +463,16 @@ void gwbuf_add_buffer_object(
|
||||
|
||||
CHK_GWBUF(buf);
|
||||
newb = (buffer_object_t *)malloc(sizeof(buffer_object_t));
|
||||
ss_dassert(newb != NULL);
|
||||
|
||||
if (newb == NULL)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Memory allocation failed due to %s.",
|
||||
strerror(errno))));
|
||||
return;
|
||||
}
|
||||
newb->bo_id = id;
|
||||
newb->bo_data = data;
|
||||
newb->bo_donefun_fp = donefun_fp;
|
||||
@ -518,8 +553,15 @@ gwbuf_add_property(GWBUF *buf, char *name, char *value)
|
||||
BUF_PROPERTY *prop;
|
||||
|
||||
if ((prop = malloc(sizeof(BUF_PROPERTY))) == NULL)
|
||||
{
|
||||
ss_dassert(prop != NULL);
|
||||
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Memory allocation failed due to %s.",
|
||||
strerror(errno))));
|
||||
return 0;
|
||||
|
||||
}
|
||||
prop->name = strdup(name);
|
||||
prop->value = strdup(value);
|
||||
spinlock_acquire(&buf->gwbuf_lock);
|
||||
|
||||
@ -1140,8 +1140,28 @@ dcb_close(DCB *dcb)
|
||||
/*<
|
||||
* Stop dcb's listening and modify state accordingly.
|
||||
*/
|
||||
rc = poll_remove_dcb(dcb);
|
||||
|
||||
if (dcb->state == DCB_STATE_POLLING)
|
||||
{
|
||||
rc = poll_remove_dcb(dcb);
|
||||
|
||||
if (rc == 0) {
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [dcb_close] Removed dcb %p in state %s from "
|
||||
"poll set.",
|
||||
pthread_self(),
|
||||
dcb,
|
||||
STRDCBSTATE(dcb->state))));
|
||||
} else {
|
||||
LOGIF(LE, (skygw_log_write(
|
||||
LOGFILE_ERROR,
|
||||
"%lu [dcb_close] Error : Removing dcb %p in state %s from "
|
||||
"poll set failed.",
|
||||
pthread_self(),
|
||||
dcb,
|
||||
STRDCBSTATE(dcb->state))));
|
||||
}
|
||||
}
|
||||
ss_dassert(dcb->state == DCB_STATE_NOPOLLING ||
|
||||
dcb->state == DCB_STATE_ZOMBIE);
|
||||
/**
|
||||
@ -1153,23 +1173,6 @@ dcb_close(DCB *dcb)
|
||||
}
|
||||
dcb_call_callback(dcb, DCB_REASON_CLOSE);
|
||||
|
||||
if (rc == 0) {
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [dcb_close] Removed dcb %p in state %s from "
|
||||
"poll set.",
|
||||
pthread_self(),
|
||||
dcb,
|
||||
STRDCBSTATE(dcb->state))));
|
||||
} else {
|
||||
LOGIF(LE, (skygw_log_write(
|
||||
LOGFILE_ERROR,
|
||||
"%lu [dcb_close] Error : Removing dcb %p in state %s from "
|
||||
"poll set failed.",
|
||||
pthread_self(),
|
||||
dcb,
|
||||
STRDCBSTATE(dcb->state))));
|
||||
}
|
||||
|
||||
if (dcb->state == DCB_STATE_NOPOLLING)
|
||||
{
|
||||
@ -1891,7 +1894,6 @@ DCB_CALLBACK *cb, *nextcb;
|
||||
int
|
||||
dcb_isvalid(DCB *dcb)
|
||||
{
|
||||
DCB *ptr;
|
||||
int rval = 0;
|
||||
|
||||
if (dcb)
|
||||
|
||||
@ -541,6 +541,7 @@ static bool resolve_maxscale_homedir(
|
||||
{
|
||||
bool succp;
|
||||
char* tmp;
|
||||
char* tmp2;
|
||||
char* log_context = NULL;
|
||||
|
||||
ss_dassert(*p_home_dir == NULL);
|
||||
@ -548,6 +549,7 @@ static bool resolve_maxscale_homedir(
|
||||
if (*p_home_dir != NULL)
|
||||
{
|
||||
log_context = strdup("Command-line argument");
|
||||
tmp = NULL;
|
||||
goto check_home_dir;
|
||||
}
|
||||
/*<
|
||||
@ -593,8 +595,9 @@ static bool resolve_maxscale_homedir(
|
||||
* isn't specified. Thus, try to access $PWD/MaxScale.cnf .
|
||||
*/
|
||||
tmp = strndup(getenv("PWD"), PATH_MAX);
|
||||
get_expanded_pathname(p_home_dir, tmp, default_cnf_fname);
|
||||
|
||||
tmp2 = get_expanded_pathname(p_home_dir, tmp, default_cnf_fname);
|
||||
free(tmp2); /*< full path isn't needed so simply free it */
|
||||
|
||||
if (*p_home_dir != NULL)
|
||||
{
|
||||
log_context = strdup("Current working directory");
|
||||
@ -638,7 +641,7 @@ check_home_dir:
|
||||
fprintf(stderr,
|
||||
"Using %s as MAXSCALE_HOME = %s\n",
|
||||
log_context,
|
||||
tmp);
|
||||
(tmp == NULL ? *p_home_dir : tmp));
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -646,8 +649,11 @@ check_home_dir:
|
||||
{
|
||||
succp = false;
|
||||
}
|
||||
free (tmp);
|
||||
|
||||
if (tmp != NULL)
|
||||
{
|
||||
free(tmp);
|
||||
}
|
||||
|
||||
if (log_context != NULL)
|
||||
{
|
||||
free(log_context);
|
||||
@ -880,6 +886,13 @@ static char* get_expanded_pathname(
|
||||
|
||||
if (cnf_file_buf == NULL)
|
||||
{
|
||||
ss_dassert(cnf_file_buf != NULL);
|
||||
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Memory allocation failed due to %s.",
|
||||
strerror(errno))));
|
||||
|
||||
free(expanded_path);
|
||||
expanded_path = NULL;
|
||||
goto return_cnf_file_buf;
|
||||
|
||||
@ -245,7 +245,7 @@ GWBUF *modutil_create_mysql_err_msg(
|
||||
int affected_rows,
|
||||
int merrno,
|
||||
const char *statemsg,
|
||||
const char * msg)
|
||||
const char *msg)
|
||||
{
|
||||
uint8_t *outbuf = NULL;
|
||||
uint8_t mysql_payload_size = 0;
|
||||
@ -259,6 +259,10 @@ GWBUF *modutil_create_mysql_err_msg(
|
||||
const char *mysql_state = NULL;
|
||||
GWBUF *errbuf = NULL;
|
||||
|
||||
if (statemsg == NULL || msg == NULL)
|
||||
{
|
||||
return NULL;
|
||||
}
|
||||
mysql_errno = (unsigned int)merrno;
|
||||
mysql_error_msg = msg;
|
||||
mysql_state = statemsg;
|
||||
@ -270,9 +274,6 @@ GWBUF *modutil_create_mysql_err_msg(
|
||||
mysql_statemsg[0]='#';
|
||||
memcpy(mysql_statemsg+1, mysql_state, 5);
|
||||
|
||||
if (msg != NULL) {
|
||||
mysql_error_msg = msg;
|
||||
}
|
||||
mysql_payload_size = sizeof(field_count) +
|
||||
sizeof(mysql_err) +
|
||||
sizeof(mysql_statemsg) +
|
||||
@ -283,8 +284,9 @@ GWBUF *modutil_create_mysql_err_msg(
|
||||
ss_dassert(errbuf != NULL);
|
||||
|
||||
if (errbuf == NULL)
|
||||
{
|
||||
return NULL;
|
||||
|
||||
}
|
||||
outbuf = GWBUF_DATA(errbuf);
|
||||
|
||||
/** write packet header and packet number */
|
||||
|
||||
@ -254,6 +254,7 @@ MAXKEYS key;
|
||||
secret_file,
|
||||
errno,
|
||||
strerror(errno))));
|
||||
close(fd);
|
||||
return 1;
|
||||
}
|
||||
|
||||
|
||||
@ -192,6 +192,7 @@ int i, cflags = REG_ICASE;
|
||||
|
||||
if (my_instance->match == NULL || my_instance->replace == NULL)
|
||||
{
|
||||
free(my_instance);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
@ -386,6 +387,8 @@ regmatch_t match[10];
|
||||
free(orig);
|
||||
return NULL;
|
||||
}
|
||||
free(orig);
|
||||
|
||||
res_size = 2 * length;
|
||||
result = (char *)malloc(res_size);
|
||||
res_length = 0;
|
||||
|
||||
@ -377,7 +377,7 @@ static int gw_read_backend_event(DCB *dcb) {
|
||||
dcb,
|
||||
ERRACT_REPLY_CLIENT,
|
||||
&succp);
|
||||
|
||||
gwbuf_free(errbuf);
|
||||
ss_dassert(!succp);
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
@ -459,7 +459,8 @@ static int gw_read_backend_event(DCB *dcb) {
|
||||
dcb,
|
||||
ERRACT_NEW_CONNECTION,
|
||||
&succp);
|
||||
|
||||
gwbuf_free(errbuf);
|
||||
|
||||
if (!succp)
|
||||
{
|
||||
spinlock_acquire(&session->ses_lock);
|
||||
@ -848,7 +849,8 @@ static int gw_error_backend_event(DCB *dcb)
|
||||
dcb,
|
||||
ERRACT_NEW_CONNECTION,
|
||||
&succp);
|
||||
|
||||
gwbuf_free(errbuf);
|
||||
|
||||
/** There are not required backends available, close session. */
|
||||
if (!succp) {
|
||||
spinlock_acquire(&session->ses_lock);
|
||||
@ -1031,7 +1033,8 @@ gw_backend_hangup(DCB *dcb)
|
||||
ERRACT_NEW_CONNECTION,
|
||||
&succp);
|
||||
|
||||
/** There are not required backends available, close session. */
|
||||
gwbuf_free(errbuf);
|
||||
/** There are no required backends available, close session. */
|
||||
if (!succp)
|
||||
{
|
||||
#if defined(SS_DEBUG)
|
||||
@ -1039,7 +1042,6 @@ gw_backend_hangup(DCB *dcb)
|
||||
LOGFILE_ERROR,
|
||||
"Backend hangup -> closing session.")));
|
||||
#endif
|
||||
|
||||
spinlock_acquire(&session->ses_lock);
|
||||
session->state = SESSION_STATE_STOPPING;
|
||||
spinlock_release(&session->ses_lock);
|
||||
@ -1176,7 +1178,8 @@ static int backend_write_delayqueue(DCB *dcb)
|
||||
dcb,
|
||||
ERRACT_NEW_CONNECTION,
|
||||
&succp);
|
||||
|
||||
gwbuf_free(errbuf);
|
||||
|
||||
if (!succp)
|
||||
{
|
||||
if (session != NULL)
|
||||
@ -1300,15 +1303,34 @@ static int gw_change_user(
|
||||
backend->session->client->remote,
|
||||
password_set,
|
||||
"");
|
||||
if (message == NULL)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Creating error message failed.")));
|
||||
rv = 0;
|
||||
goto retblock;
|
||||
}
|
||||
/** TODO: Add custom message indicating that retry would probably help */
|
||||
buf = modutil_create_mysql_err_msg(1, 0, 1045, "28000", message);
|
||||
free(message);
|
||||
|
||||
if (buf == NULL)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Creating buffer for error message failed.")));
|
||||
rv = 0;
|
||||
goto retblock;
|
||||
}
|
||||
/** Set flags that help router to identify session commans reply */
|
||||
gwbuf_set_type(buf, GWBUF_TYPE_MYSQL);
|
||||
gwbuf_set_type(buf, GWBUF_TYPE_SESCMD_RESPONSE);
|
||||
gwbuf_set_type(buf, GWBUF_TYPE_RESPONSE_END);
|
||||
/** Create an incoming event for backend DCB */
|
||||
poll_add_epollin_event_to_dcb(backend, buf);
|
||||
rv = 0;
|
||||
poll_add_epollin_event_to_dcb(backend, gwbuf_clone(buf));
|
||||
gwbuf_free(buf);
|
||||
rv = 1;
|
||||
} else {
|
||||
rv = gw_send_change_user_to_backend(database, username, client_sha1, backend_protocol);
|
||||
/*
|
||||
@ -1318,6 +1340,8 @@ static int gw_change_user(
|
||||
strcpy(current_session->db, database);
|
||||
memcpy(current_session->client_sha1, client_sha1, sizeof(current_session->client_sha1));
|
||||
}
|
||||
|
||||
retblock:
|
||||
gwbuf_free(queue);
|
||||
|
||||
return rv;
|
||||
|
||||
@ -430,7 +430,7 @@ static int gw_mysql_do_authentication(DCB *dcb, GWBUF *queue) {
|
||||
|
||||
connect_with_db =
|
||||
GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB & gw_mysql_get_byte4(
|
||||
&protocol->client_capabilities);
|
||||
(uint32_t *)&protocol->client_capabilities);
|
||||
/*
|
||||
compress =
|
||||
GW_MYSQL_CAPABILITIES_COMPRESS & gw_mysql_get_byte4(
|
||||
@ -859,6 +859,7 @@ int gw_read_client_event(
|
||||
dcb,
|
||||
ERRACT_REPLY_CLIENT,
|
||||
&succp);
|
||||
gwbuf_free(errbuf);
|
||||
ss_dassert(!succp);
|
||||
|
||||
dcb_close(dcb);
|
||||
@ -1017,7 +1018,7 @@ int gw_MySQLListener(
|
||||
errno,
|
||||
strerror(errno));
|
||||
fprintf(stderr, "* Can't bind to %s\n\n", config_bind);
|
||||
|
||||
close(l_so);
|
||||
return 0;
|
||||
}
|
||||
|
||||
@ -1039,13 +1040,14 @@ int gw_MySQLListener(
|
||||
errno,
|
||||
strerror(errno));
|
||||
fprintf(stderr, "* Can't bind to %s\n\n", config_bind);
|
||||
|
||||
close(l_so);
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
|
||||
default:
|
||||
fprintf(stderr, "* Socket Family %i not supported\n", current_addr->sa_family);
|
||||
close(l_so);
|
||||
return 0;
|
||||
}
|
||||
|
||||
@ -1062,6 +1064,7 @@ int gw_MySQLListener(
|
||||
"\n* Failed to start listening MySQL due error %d, %s\n\n",
|
||||
eno,
|
||||
strerror(eno));
|
||||
close(l_so);
|
||||
return 0;
|
||||
}
|
||||
// assign l_so to dcb
|
||||
@ -1191,8 +1194,8 @@ int gw_MySQLAccept(DCB *listener)
|
||||
strerror(eno))));
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error %d, %s."
|
||||
"Failed to accept new client connection.",
|
||||
"Error : Failed to accept new client "
|
||||
"connection due to %d, %s.",
|
||||
eno,
|
||||
strerror(eno))));
|
||||
rc = 1;
|
||||
@ -1223,9 +1226,9 @@ int gw_MySQLAccept(DCB *listener)
|
||||
if (client_dcb == NULL) {
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"%lu [gw_MySQLAccept] Failed to create "
|
||||
"dcb object for client connection.",
|
||||
pthread_self())));
|
||||
"Error : Failed to create "
|
||||
"DCB object for client connection.")));
|
||||
close(c_sock);
|
||||
rc = 1;
|
||||
goto return_rc;
|
||||
}
|
||||
@ -1327,6 +1330,7 @@ int gw_MySQLAccept(DCB *listener)
|
||||
}
|
||||
#endif
|
||||
return_rc:
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
|
||||
@ -2111,19 +2111,34 @@ char *create_auth_fail_str(
|
||||
else
|
||||
db_len = 0;
|
||||
|
||||
if (db_len>0)
|
||||
if (db_len > 0)
|
||||
{
|
||||
ferrstr = "Access denied for user '%s'@'%s' (using password: %s) to database '%s'";
|
||||
}
|
||||
else
|
||||
{
|
||||
ferrstr = "Access denied for user '%s'@'%s' (using password: %s)";
|
||||
|
||||
}
|
||||
errstr = (char *)malloc(strlen(username)+strlen(ferrstr)+strlen(hostaddr)+strlen("YES")-6 + db_len + ((db_len > 0) ? (strlen(" to database ") +2) : 0) + 1);
|
||||
|
||||
if (errstr != NULL) {
|
||||
if (db_len>0)
|
||||
sprintf(errstr, ferrstr, username, hostaddr, (*sha1 == '\0' ? "NO" : "YES"), db);
|
||||
else
|
||||
sprintf(errstr, ferrstr, username, hostaddr, (*sha1 == '\0' ? "NO" : "YES"));
|
||||
if (errstr == NULL)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Memory allocation failed due to %s.",
|
||||
strerror(errno))));
|
||||
goto retblock;
|
||||
}
|
||||
|
||||
if (db_len > 0)
|
||||
{
|
||||
sprintf(errstr, ferrstr, username, hostaddr, (*sha1 == '\0' ? "NO" : "YES"), db);
|
||||
}
|
||||
else
|
||||
{
|
||||
sprintf(errstr, ferrstr, username, hostaddr, (*sha1 == '\0' ? "NO" : "YES"));
|
||||
}
|
||||
|
||||
retblock:
|
||||
return errstr;
|
||||
}
|
||||
|
||||
@ -631,7 +631,7 @@ char *ptr, *lptr;
|
||||
else
|
||||
{
|
||||
i++;
|
||||
if (i >= MAXARGS)
|
||||
if (i >= MAXARGS-1)
|
||||
break;
|
||||
args[i] = ptr + 1;
|
||||
}
|
||||
|
||||
@ -654,7 +654,7 @@ DCB* backend_dcb;
|
||||
* @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 The number of bytes sent
|
||||
* @return if succeed 1, otherwise 0
|
||||
*/
|
||||
static int
|
||||
routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
|
||||
@ -697,22 +697,24 @@ routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
|
||||
"Error : Failed to route MySQL command %d to backend "
|
||||
"server.",
|
||||
mysql_command)));
|
||||
rc = 0;
|
||||
goto return_rc;
|
||||
}
|
||||
|
||||
switch(mysql_command) {
|
||||
case MYSQL_COM_CHANGE_USER:
|
||||
rc = backend_dcb->func.auth(
|
||||
backend_dcb,
|
||||
NULL,
|
||||
backend_dcb->session,
|
||||
queue);
|
||||
break;
|
||||
default:
|
||||
rc = backend_dcb->func.write(backend_dcb, queue);
|
||||
break;
|
||||
|
||||
switch(mysql_command) {
|
||||
case MYSQL_COM_CHANGE_USER:
|
||||
rc = backend_dcb->func.auth(
|
||||
backend_dcb,
|
||||
NULL,
|
||||
backend_dcb->session,
|
||||
queue);
|
||||
break;
|
||||
|
||||
default:
|
||||
rc = backend_dcb->func.write(backend_dcb, queue);
|
||||
break;
|
||||
}
|
||||
|
||||
|
||||
CHK_PROTOCOL(((MySQLProtocol*)backend_dcb->protocol));
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
@ -813,21 +815,33 @@ clientReply(
|
||||
* @param action The action: REPLY, REPLY_AND_CLOSE, NEW_CONNECTION
|
||||
*
|
||||
*/
|
||||
static void
|
||||
handleError(
|
||||
ROUTER *instance,
|
||||
void *router_session,
|
||||
GWBUF *errbuf,
|
||||
DCB *backend_dcb,
|
||||
error_action_t action,
|
||||
bool *succp)
|
||||
static void handleError(
|
||||
ROUTER *instance,
|
||||
void *router_session,
|
||||
GWBUF *errbuf,
|
||||
DCB *backend_dcb,
|
||||
error_action_t action,
|
||||
bool *succp)
|
||||
|
||||
{
|
||||
DCB *client = NULL;
|
||||
DCB *client_dcb;
|
||||
SESSION *session = backend_dcb->session;
|
||||
client = session->client;
|
||||
session_state_t sesstate;
|
||||
|
||||
spinlock_acquire(&session->ses_lock);
|
||||
sesstate = session->state;
|
||||
client_dcb = session->client;
|
||||
spinlock_release(&session->ses_lock);
|
||||
ss_dassert(client_dcb != NULL);
|
||||
|
||||
if (sesstate == SESSION_STATE_ROUTER_READY)
|
||||
{
|
||||
CHK_DCB(client_dcb);
|
||||
client_dcb->func.write(client_dcb, gwbuf_clone(errbuf));
|
||||
}
|
||||
|
||||
/** false because connection is not available anymore */
|
||||
*succp = false;
|
||||
ss_dassert(client != NULL);
|
||||
}
|
||||
|
||||
/** to be inline'd */
|
||||
|
||||
@ -3829,7 +3829,6 @@ static bool route_session_write(
|
||||
{
|
||||
succp = false;
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
rses_end_locked_router_action(router_cli_ses);
|
||||
@ -3842,6 +3841,12 @@ static bool route_session_write(
|
||||
succp = false;
|
||||
goto return_succp;
|
||||
}
|
||||
|
||||
if (router_cli_ses->rses_nbackends <= 0)
|
||||
{
|
||||
succp = false;
|
||||
goto return_succp;
|
||||
}
|
||||
/**
|
||||
* Additional reference is created to querybuf to
|
||||
* prevent it from being released before properties
|
||||
@ -3909,6 +3914,10 @@ static bool route_session_write(
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
succp = false;
|
||||
}
|
||||
}
|
||||
/** Unlock router session */
|
||||
rses_end_locked_router_action(router_cli_ses);
|
||||
@ -4014,7 +4023,7 @@ static void rwsplit_process_router_options(
|
||||
*
|
||||
* @param instance The router instance
|
||||
* @param router_session The router session
|
||||
* @param message The error message to reply
|
||||
* @param errmsgbuf The error message to reply
|
||||
* @param backend_dcb The backend DCB
|
||||
* @param action The action: REPLY, REPLY_AND_CLOSE, NEW_CONNECTION
|
||||
* @param succp Result of action.
|
||||
@ -4079,114 +4088,104 @@ static void handleError (
|
||||
|
||||
|
||||
static void handle_error_reply_client(
|
||||
SESSION* ses,
|
||||
GWBUF* errmsg)
|
||||
SESSION* ses,
|
||||
GWBUF* errmsg)
|
||||
{
|
||||
session_state_t sesstate;
|
||||
DCB* client_dcb;
|
||||
|
||||
spinlock_acquire(&ses->ses_lock);
|
||||
sesstate = ses->state;
|
||||
client_dcb = ses->client;
|
||||
spinlock_release(&ses->ses_lock);
|
||||
|
||||
if (sesstate == SESSION_STATE_ROUTER_READY)
|
||||
{
|
||||
CHK_DCB(client_dcb);
|
||||
client_dcb->func.write(client_dcb, errmsg);
|
||||
}
|
||||
else
|
||||
{
|
||||
while ((errmsg=gwbuf_consume(errmsg, GWBUF_LENGTH(errmsg))) != NULL)
|
||||
;
|
||||
}
|
||||
session_state_t sesstate;
|
||||
DCB* client_dcb;
|
||||
|
||||
spinlock_acquire(&ses->ses_lock);
|
||||
sesstate = ses->state;
|
||||
client_dcb = ses->client;
|
||||
spinlock_release(&ses->ses_lock);
|
||||
|
||||
if (sesstate == SESSION_STATE_ROUTER_READY)
|
||||
{
|
||||
CHK_DCB(client_dcb);
|
||||
client_dcb->func.write(client_dcb, gwbuf_clone(errmsg));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* This must be called with router lock
|
||||
*/
|
||||
static bool handle_error_new_connection(
|
||||
ROUTER_INSTANCE* inst,
|
||||
ROUTER_CLIENT_SES* rses,
|
||||
DCB* backend_dcb,
|
||||
GWBUF* errmsg)
|
||||
ROUTER_INSTANCE* inst,
|
||||
ROUTER_CLIENT_SES* rses,
|
||||
DCB* backend_dcb,
|
||||
GWBUF* errmsg)
|
||||
{
|
||||
SESSION* ses;
|
||||
int router_nservers;
|
||||
int max_nslaves;
|
||||
int max_slave_rlag;
|
||||
backend_ref_t* bref;
|
||||
bool succp;
|
||||
|
||||
ss_dassert(SPINLOCK_IS_LOCKED(&rses->rses_lock));
|
||||
|
||||
ses = backend_dcb->session;
|
||||
CHK_SESSION(ses);
|
||||
|
||||
bref = get_bref_from_dcb(rses, backend_dcb);
|
||||
|
||||
/** failed DCB has already been replaced */
|
||||
if (bref == NULL)
|
||||
{
|
||||
succp = true;
|
||||
goto return_succp;
|
||||
}
|
||||
/**
|
||||
* Error handler is already called for this DCB because
|
||||
* it's not polling anymore. It can be assumed that
|
||||
* it succeed because rses isn't closed.
|
||||
*/
|
||||
if (backend_dcb->state != DCB_STATE_POLLING)
|
||||
{
|
||||
succp = true;
|
||||
goto return_succp;
|
||||
}
|
||||
|
||||
CHK_BACKEND_REF(bref);
|
||||
|
||||
if (BREF_IS_WAITING_RESULT(bref))
|
||||
{
|
||||
DCB* client_dcb;
|
||||
client_dcb = ses->client;
|
||||
client_dcb->func.write(client_dcb, errmsg);
|
||||
bref_clear_state(bref, BREF_WAITING_RESULT);
|
||||
}
|
||||
else
|
||||
{
|
||||
while ((errmsg=gwbuf_consume(errmsg, GWBUF_LENGTH(errmsg))) != NULL)
|
||||
;
|
||||
}
|
||||
bref_clear_state(bref, BREF_IN_USE);
|
||||
bref_set_state(bref, BREF_CLOSED);
|
||||
/**
|
||||
* Remove callback because this DCB won't be used
|
||||
* unless it is reconnected later, and then the callback
|
||||
* is set again.
|
||||
*/
|
||||
dcb_remove_callback(backend_dcb,
|
||||
DCB_REASON_NOT_RESPONDING,
|
||||
&router_handle_state_switch,
|
||||
(void *)bref);
|
||||
|
||||
router_nservers = router_get_servercount(inst);
|
||||
max_nslaves = rses_get_max_slavecount(rses, router_nservers);
|
||||
max_slave_rlag = rses_get_max_replication_lag(rses);
|
||||
/**
|
||||
* Try to get replacement slave or at least the minimum
|
||||
* number of slave connections for router session.
|
||||
*/
|
||||
succp = select_connect_backend_servers(
|
||||
&rses->rses_master_ref,
|
||||
rses->rses_backend_ref,
|
||||
router_nservers,
|
||||
max_nslaves,
|
||||
max_slave_rlag,
|
||||
rses->rses_config.rw_slave_select_criteria,
|
||||
ses,
|
||||
inst);
|
||||
|
||||
SESSION* ses;
|
||||
int router_nservers;
|
||||
int max_nslaves;
|
||||
int max_slave_rlag;
|
||||
backend_ref_t* bref;
|
||||
bool succp;
|
||||
|
||||
ss_dassert(SPINLOCK_IS_LOCKED(&rses->rses_lock));
|
||||
|
||||
ses = backend_dcb->session;
|
||||
CHK_SESSION(ses);
|
||||
|
||||
bref = get_bref_from_dcb(rses, backend_dcb);
|
||||
|
||||
/** failed DCB has already been replaced */
|
||||
if (bref == NULL)
|
||||
{
|
||||
succp = true;
|
||||
goto return_succp;
|
||||
}
|
||||
/**
|
||||
* Error handler is already called for this DCB because
|
||||
* it's not polling anymore. It can be assumed that
|
||||
* it succeed because rses isn't closed.
|
||||
*/
|
||||
if (backend_dcb->state != DCB_STATE_POLLING)
|
||||
{
|
||||
succp = true;
|
||||
goto return_succp;
|
||||
}
|
||||
|
||||
CHK_BACKEND_REF(bref);
|
||||
|
||||
if (BREF_IS_WAITING_RESULT(bref))
|
||||
{
|
||||
DCB* client_dcb;
|
||||
client_dcb = ses->client;
|
||||
client_dcb->func.write(client_dcb, gwbuf_clone(errmsg));
|
||||
bref_clear_state(bref, BREF_WAITING_RESULT);
|
||||
}
|
||||
bref_clear_state(bref, BREF_IN_USE);
|
||||
bref_set_state(bref, BREF_CLOSED);
|
||||
/**
|
||||
* Remove callback because this DCB won't be used
|
||||
* unless it is reconnected later, and then the callback
|
||||
* is set again.
|
||||
*/
|
||||
dcb_remove_callback(backend_dcb,
|
||||
DCB_REASON_NOT_RESPONDING,
|
||||
&router_handle_state_switch,
|
||||
(void *)bref);
|
||||
|
||||
router_nservers = router_get_servercount(inst);
|
||||
max_nslaves = rses_get_max_slavecount(rses, router_nservers);
|
||||
max_slave_rlag = rses_get_max_replication_lag(rses);
|
||||
/**
|
||||
* Try to get replacement slave or at least the minimum
|
||||
* number of slave connections for router session.
|
||||
*/
|
||||
succp = select_connect_backend_servers(
|
||||
&rses->rses_master_ref,
|
||||
rses->rses_backend_ref,
|
||||
router_nservers,
|
||||
max_nslaves,
|
||||
max_slave_rlag,
|
||||
rses->rses_config.rw_slave_select_criteria,
|
||||
ses,
|
||||
inst);
|
||||
|
||||
return_succp:
|
||||
return succp;
|
||||
return succp;
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user