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

Removed plainrouter from MXS-121 branch.

Browse Source
This commit is contained in:
Markus Makela 2015-05-18 16:25:50 +03:00
parent 8c120ca469
commit 68c5dedeec
5 changed files with 0 additions and 2800 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

260
server/modules/include/plainprotocol.h
View File

@ -1,260 +0,0 @@
#ifndef _MYSQL_PROTOCOL_H
#define _MYSQL_PROTOCOL_H
/*
* This file is distributed as part of the MariaDB Corporation 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 MariaDB Corporation Ab 2013-2015
*/
/*
* Revision History
*
* Date Who Description
* 24-03-2015 Markus Makela Initial implementation
*/
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <openssl/sha.h>
#include <sys/ioctl.h>
#include <errno.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdbool.h>
#include <fcntl.h>
#include <sys/types.h>
#include <unistd.h>
#include <service.h>
#include <router.h>
#include <poll.h>
#include <users.h>
#include <dbusers.h>
#include <version.h>
#include <housekeeper.h>
#define GW_MYSQL_VERSION "MaxScale " MAXSCALE_VERSION
#define GW_MYSQL_LOOP_TIMEOUT 300000000
#define GW_MYSQL_READ 0
#define GW_MYSQL_WRITE 1
#define MYSQL_HEADER_LEN 4L
#define GW_MYSQL_PROTOCOL_VERSION 10 // version is 10
#define GW_MYSQL_HANDSHAKE_FILLER 0x00
#define GW_MYSQL_SERVER_CAPABILITIES_BYTE1 0xff
#define GW_MYSQL_SERVER_CAPABILITIES_BYTE2 0xf7
#define GW_MYSQL_SERVER_LANGUAGE 0x08
#define GW_MYSQL_MAX_PACKET_LEN 0xffffffL;
#define GW_MYSQL_SCRAMBLE_SIZE 20
#define GW_SCRAMBLE_LENGTH_323 8
#ifndef MYSQL_SCRAMBLE_LEN
# define MYSQL_SCRAMBLE_LEN GW_MYSQL_SCRAMBLE_SIZE
#endif
#define GW_NOINTR_CALL(A) do { errno = 0; A; } while (errno == EINTR)
#define SMALL_CHUNK 1024
#define MAX_CHUNK SMALL_CHUNK * 8 * 4
#define ToHex(Y) (Y>='0'&&Y<='9'?Y-'0':Y-'A'+10)
struct dcb;
/** Protocol packing macros. */
#define gw_mysql_set_byte2(__buffer, __int) do { \
(__buffer)[0]= (uint8_t)((__int) & 0xFF); \
(__buffer)[1]= (uint8_t)(((__int) >> 8) & 0xFF); } while (0)
#define gw_mysql_set_byte3(__buffer, __int) do { \
(__buffer)[0]= (uint8_t)((__int) & 0xFF); \
(__buffer)[1]= (uint8_t)(((__int) >> 8) & 0xFF); \
(__buffer)[2]= (uint8_t)(((__int) >> 16) & 0xFF); } while (0)
#define gw_mysql_set_byte4(__buffer, __int) do { \
(__buffer)[0]= (uint8_t)((__int) & 0xFF); \
(__buffer)[1]= (uint8_t)(((__int) >> 8) & 0xFF); \
(__buffer)[2]= (uint8_t)(((__int) >> 16) & 0xFF); \
(__buffer)[3]= (uint8_t)(((__int) >> 24) & 0xFF); } while (0)
/** Protocol unpacking macros. */
#define gw_mysql_get_byte2(__buffer) \
(uint16_t)((__buffer)[0] | \
((__buffer)[1] << 8))
#define gw_mysql_get_byte3(__buffer) \
(uint32_t)((__buffer)[0] | \
((__buffer)[1] << 8) | \
((__buffer)[2] << 16))
#define gw_mysql_get_byte4(__buffer) \
(uint32_t)((__buffer)[0] | \
((__buffer)[1] << 8) | \
((__buffer)[2] << 16) | \
((__buffer)[3] << 24))
#define gw_mysql_get_byte8(__buffer) \
((uint64_t)(__buffer)[0] | \
((uint64_t)(__buffer)[1] << 8) | \
((uint64_t)(__buffer)[2] << 16) | \
((uint64_t)(__buffer)[3] << 24) | \
((uint64_t)(__buffer)[4] << 32) | \
((uint64_t)(__buffer)[5] << 40) | \
((uint64_t)(__buffer)[6] << 48) | \
((uint64_t)(__buffer)[7] << 56))
/** MySQL protocol constants */
typedef enum
{
GW_MYSQL_CAPABILITIES_NONE= 0,
GW_MYSQL_CAPABILITIES_LONG_PASSWORD= (1 << 0),
GW_MYSQL_CAPABILITIES_FOUND_ROWS= (1 << 1),
GW_MYSQL_CAPABILITIES_LONG_FLAG= (1 << 2),
GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB= (1 << 3),
GW_MYSQL_CAPABILITIES_NO_SCHEMA= (1 << 4),
GW_MYSQL_CAPABILITIES_COMPRESS= (1 << 5),
GW_MYSQL_CAPABILITIES_ODBC= (1 << 6),
GW_MYSQL_CAPABILITIES_LOCAL_FILES= (1 << 7),
GW_MYSQL_CAPABILITIES_IGNORE_SPACE= (1 << 8),
GW_MYSQL_CAPABILITIES_PROTOCOL_41= (1 << 9),
GW_MYSQL_CAPABILITIES_INTERACTIVE= (1 << 10),
GW_MYSQL_CAPABILITIES_SSL= (1 << 11),
GW_MYSQL_CAPABILITIES_IGNORE_SIGPIPE= (1 << 12),
GW_MYSQL_CAPABILITIES_TRANSACTIONS= (1 << 13),
GW_MYSQL_CAPABILITIES_RESERVED= (1 << 14),
GW_MYSQL_CAPABILITIES_SECURE_CONNECTION= (1 << 15),
GW_MYSQL_CAPABILITIES_MULTI_STATEMENTS= (1 << 16),
GW_MYSQL_CAPABILITIES_MULTI_RESULTS= (1 << 17),
GW_MYSQL_CAPABILITIES_PS_MULTI_RESULTS= (1 << 18),
GW_MYSQL_CAPABILITIES_PLUGIN_AUTH= (1 << 19),
GW_MYSQL_CAPABILITIES_SSL_VERIFY_SERVER_CERT= (1 << 30),
GW_MYSQL_CAPABILITIES_REMEMBER_OPTIONS= (1 << 31),
GW_MYSQL_CAPABILITIES_CLIENT= (GW_MYSQL_CAPABILITIES_LONG_PASSWORD |
GW_MYSQL_CAPABILITIES_FOUND_ROWS |
GW_MYSQL_CAPABILITIES_LONG_FLAG |
GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB |
GW_MYSQL_CAPABILITIES_LOCAL_FILES |
GW_MYSQL_CAPABILITIES_PLUGIN_AUTH |
GW_MYSQL_CAPABILITIES_TRANSACTIONS |
GW_MYSQL_CAPABILITIES_PROTOCOL_41 |
GW_MYSQL_CAPABILITIES_MULTI_STATEMENTS |
GW_MYSQL_CAPABILITIES_MULTI_RESULTS |
GW_MYSQL_CAPABILITIES_PS_MULTI_RESULTS |
GW_MYSQL_CAPABILITIES_SECURE_CONNECTION),
GW_MYSQL_CAPABILITIES_CLIENT_COMPRESS= (GW_MYSQL_CAPABILITIES_LONG_PASSWORD |
GW_MYSQL_CAPABILITIES_FOUND_ROWS |
GW_MYSQL_CAPABILITIES_LONG_FLAG |
GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB |
GW_MYSQL_CAPABILITIES_LOCAL_FILES |
GW_MYSQL_CAPABILITIES_PLUGIN_AUTH |
GW_MYSQL_CAPABILITIES_TRANSACTIONS |
GW_MYSQL_CAPABILITIES_PROTOCOL_41 |
GW_MYSQL_CAPABILITIES_MULTI_STATEMENTS |
GW_MYSQL_CAPABILITIES_MULTI_RESULTS |
GW_MYSQL_CAPABILITIES_PS_MULTI_RESULTS |
GW_MYSQL_CAPABILITIES_COMPRESS
),
} gw_mysql_capabilities_t;
/** Copy from enum in mariadb-5.5 mysql_com.h */
typedef enum mysql_server_cmd {
MYSQL_COM_UNDEFINED = -1,
MYSQL_COM_SLEEP = 0,
MYSQL_COM_QUIT,
MYSQL_COM_INIT_DB,
MYSQL_COM_QUERY,
MYSQL_COM_FIELD_LIST,
MYSQL_COM_CREATE_DB,
MYSQL_COM_DROP_DB,
MYSQL_COM_REFRESH,
MYSQL_COM_SHUTDOWN,
MYSQL_COM_STATISTICS,
MYSQL_COM_PROCESS_INFO,
MYSQL_COM_CONNECT,
MYSQL_COM_PROCESS_KILL,
MYSQL_COM_DEBUG,
MYSQL_COM_PING,
MYSQL_COM_TIME,
MYSQL_COM_DELAYED_INSERT,
MYSQL_COM_CHANGE_USER,
MYSQL_COM_BINLOG_DUMP,
MYSQL_COM_TABLE_DUMP,
MYSQL_COM_CONNECT_OUT,
MYSQL_COM_REGISTER_SLAVE,
MYSQL_COM_STMT_PREPARE,
MYSQL_COM_STMT_EXECUTE,
MYSQL_COM_STMT_SEND_LONG_DATA,
MYSQL_COM_STMT_CLOSE,
MYSQL_COM_STMT_RESET,
MYSQL_COM_SET_OPTION,
MYSQL_COM_STMT_FETCH,
MYSQL_COM_DAEMON,
MYSQL_COM_END /*< Must be the last */
} mysql_server_cmd_t;
/**
* MySQL Protocol specific state data.
*
* Protocol carries information from client side to backend side, such as
* MySQL session command information and history of earlier session commands.
*/
typedef struct {
#if defined(SS_DEBUG)
skygw_chk_t protocol_chk_top;
#endif
int fd; /*< The socket descriptor */
struct dcb *owner_dcb; /*< The DCB of the socket
* we are running on */
SPINLOCK protocol_lock;
uint8_t scramble[MYSQL_SCRAMBLE_LEN]; /*< server scramble,
* created or received */
uint32_t server_capabilities; /*< server capabilities,
* created or received */
uint32_t client_capabilities; /*< client capabilities,
* created or received */
unsigned long tid; /*< MySQL Thread ID, in
* handshake */
unsigned int charset; /*< MySQL character set at connect time */
#if defined(SS_DEBUG)
skygw_chk_t protocol_chk_tail;
#endif
} PlainProtocol;
#define MYSQL_GET_COMMAND(payload) (payload[4])
#define MYSQL_GET_PACKET_NO(payload) (payload[3])
#define MYSQL_GET_PACKET_LEN(payload) (gw_mysql_get_byte3(payload))
#define MYSQL_GET_ERRCODE(payload) (gw_mysql_get_byte2(&payload[5]))
#define MYSQL_GET_STMTOK_NPARAM(payload) (gw_mysql_get_byte2(&payload[9]))
#define MYSQL_GET_STMTOK_NATTR(payload) (gw_mysql_get_byte2(&payload[11]))
#define MYSQL_IS_ERROR_PACKET(payload) (MYSQL_GET_COMMAND(payload)==0xff)
#define MYSQL_IS_COM_QUIT(payload) (MYSQL_GET_COMMAND(payload)==0x01)
#define MYSQL_IS_COM_INIT_DB(payload) (MYSQL_GET_COMMAND(payload)==0x02)
#define MYSQL_IS_CHANGE_USER(payload) (MYSQL_GET_COMMAND(payload)==0x11)
#define MYSQL_GET_NATTR(payload) ((int)payload[4])
#endif /** _MYSQL_PROTOCOL_H */
PlainProtocol* mysql_protocol_init(DCB* dcb, int fd);
void mysql_protocol_done (DCB* dcb);
PlainProtocol *gw_mysql_init(PlainProtocol *data);
int gw_receive_backend_auth(PlainProtocol *protocol);
int gw_decode_mysql_server_handshake(PlainProtocol *protocol, uint8_t *payload);
int gw_read_backend_handshake(PlainProtocol *protocol);
int gw_send_authentication_to_backend(
char *dbname,
char *user,
uint8_t *passwd,
PlainProtocol *protocol);
int plain_do_connect_to_backend(char *host, int port, int* fd);

9
server/modules/protocol/CMakeLists.txt
View File

@ -6,15 +6,6 @@ add_library(MySQLBackend SHARED mysql_backend.c mysql_common.c)
target_link_libraries(MySQLBackend log_manager utils)
install(TARGETS MySQLBackend DESTINATION modules)
add_library(plainclient SHARED plainclient.c mysql_common.c)
target_link_libraries(plainclient log_manager utils)
install(TARGETS plainclient DESTINATION modules)
add_library(plainbackend SHARED plainbackend.c mysql_common.c)
target_link_libraries(plainbackend log_manager utils)
install(TARGETS plainbackend DESTINATION modules)
add_library(telnetd SHARED telnetd.c)
target_link_libraries(telnetd log_manager utils)
install(TARGETS telnetd DESTINATION modules)

894
server/modules/protocol/plainbackend.c
View File

@ -1,894 +0,0 @@
/*
* This file is distributed as part of the MariaDB Corporation 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 MariaDB Corporation Ab 2013-2014
*/
#include <plainprotocol.h>
#include <skygw_types.h>
#include <skygw_utils.h>
#include <log_manager.h>
#include <modutil.h>
#define PLAIN_BACKEND_SO_SNDBUF (128 * 1024)
#define PLAIN_BACKEND_SO_RCVBUF (128 * 1024)
/*
* MySQL Protocol module for handling the protocol between the gateway
* and the backend MySQL database.
*
* Revision History
* Date Who Description
* 14/06/2013 Mark Riddoch Initial version
* 17/06/2013 Massimiliano Pinto Added MaxScale To Backends routines
* 01/07/2013 Massimiliano Pinto Put Log Manager example code behind SS_DEBUG macros.
* 03/07/2013 Massimiliano Pinto Added delayq for incoming data before mysql connection
* 04/07/2013 Massimiliano Pinto Added asyncrhronous MySQL protocol connection to backend
* 05/07/2013 Massimiliano Pinto Added closeSession if backend auth fails
* 12/07/2013 Massimiliano Pinto Added Mysql Change User via dcb->func.auth()
* 15/07/2013 Massimiliano Pinto Added Mysql session change via dcb->func.session()
* 17/07/2013 Massimiliano Pinto Added dcb->command update from gwbuf->command for proper routing
server replies to client via router->clientReply
* 04/09/2013 Massimiliano Pinto Added dcb->session and dcb->session->client checks for NULL
* 12/09/2013 Massimiliano Pinto Added checks in gw_read_backend_event() for gw_read_backend_handshake
* 27/09/2013 Massimiliano Pinto Changed in gw_read_backend_event the check for dcb_read(), now is if rc < 0
* 24/10/2014 Massimiliano Pinto Added Mysql user@host @db authentication support
* 10/11/2014 Massimiliano Pinto Client charset is passed to backend
*
*/
#include <modinfo.h>
MODULE_INFO info = {
MODULE_API_PROTOCOL,
MODULE_GA,
GWPROTOCOL_VERSION,
"The plain protocol"
};
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[];
extern __thread log_info_t tls_log_info;
static char *version_str = "V2.0.0";
static int plain_create_backend_connection(DCB *backend, SERVER *server, SESSION *in_session);
static int plain_read_backend_event(DCB* dcb);
static int plain_write_ready_backend_event(DCB *dcb);
static int plain_write_backend(DCB *dcb, GWBUF *queue);
static int plain_error_backend_event(DCB *dcb);
static int plain_backend_close(DCB *dcb);
static int plain_backend_hangup(DCB *dcb);
static int backend_write_delayqueue(DCB *dcb);
static void backend_set_delayqueue(DCB *dcb, GWBUF *queue);
static int plain_change_user(DCB *backend_dcb, SERVER *server, SESSION *in_session, GWBUF *queue);
static GWBUF* process_response_data (DCB* dcb, GWBUF* readbuf, int nbytes_to_process);
extern char* create_auth_failed_msg( GWBUF* readbuf, char* hostaddr, uint8_t* sha1);
extern char* create_auth_fail_str(char *username, char *hostaddr, char *sha1, char *db);
static bool sescmd_response_complete(DCB* dcb);
static GWPROTOCOL MyObject = {
plain_read_backend_event, /* Read - EPOLLIN handler */
plain_write_backend, /* Write - data from gateway */
plain_write_ready_backend_event, /* WriteReady - EPOLLOUT handler */
plain_error_backend_event, /* Error - EPOLLERR handler */
plain_backend_hangup, /* HangUp - EPOLLHUP handler */
NULL, /* Accept */
plain_create_backend_connection, /* Connect */
plain_backend_close, /* Close */
NULL, /* Listen */
NULL, /* Authentication */
NULL /* Session */
};
/*
* 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()
{
}
/*
* 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
*/
GWPROTOCOL *
GetModuleObject()
{
return &MyObject;
}
/**
* Creates MySQL protocol structure
*
* @param dcb * Must be non-NULL.
* @param fd
*
* @return
*
*
* @details Protocol structure does not have fd because dcb is not
* connected yet.
*
*/
PlainProtocol* plain_protocol_init(
DCB* dcb,
int fd)
{
PlainProtocol* p;
p = (PlainProtocol *) calloc(1, sizeof(PlainProtocol));
ss_dassert(p != NULL);
if (p == NULL) {
int eno = errno;
errno = 0;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"%lu [mysql_init_protocol] MySQL protocol init failed : "
"memory allocation due error %d, %s.",
pthread_self(),
eno,
strerror(eno))));
goto return_p;
}
/*< Assign fd with protocol */
p->fd = fd;
p->owner_dcb = dcb;
return_p:
return p;
}
/**
* Backend Read Event for EPOLLIN on the MySQL backend protocol module
* @param dcb The backend Descriptor Control Block
* @return 1 on operation, 0 for no action
*/
static int plain_read_backend_event(DCB *dcb) {
PlainProtocol *client_protocol = NULL;
PlainProtocol *backend_protocol = NULL;
int rc = 0;
backend_protocol = (PlainProtocol *) dcb->protocol;
CHK_PROTOCOL(backend_protocol);
/* reading MySQL command output from backend and writing to the client */
{
GWBUF *read_buffer = NULL;
ROUTER_OBJECT *router = NULL;
ROUTER *router_instance = NULL;
SESSION *session = dcb->session;
int nbytes_read = 0;
CHK_SESSION(session);
router = session->service->router;
router_instance = session->service->router_instance;
/* read available backend data */
rc = dcb_read(dcb, &read_buffer);
if (rc < 0)
{
GWBUF* errbuf;
bool succp;
errbuf = mysql_create_custom_error(
1,
0,
"Read from backend failed");
router->handleError(
router_instance,
session->router_session,
errbuf,
dcb,
ERRACT_NEW_CONNECTION,
&succp);
gwbuf_free(errbuf);
if (!succp)
{
spinlock_acquire(&session->ses_lock);
session->state = SESSION_STATE_STOPPING;
spinlock_release(&session->ses_lock);
}
ss_dassert(dcb->dcb_errhandle_called);
dcb_close(dcb);
rc = 0;
goto return_rc;
}
nbytes_read = gwbuf_length(read_buffer);
if (nbytes_read == 0 && dcb->dcb_readqueue == NULL)
{
goto return_rc;
}
else
{
ss_dassert(read_buffer != NULL || dcb->dcb_readqueue != NULL);
}
if (dcb->session->state == SESSION_STATE_ROUTER_READY &&
dcb->session->client != NULL &&
dcb->session->client->state == DCB_STATE_POLLING)
{
client_protocol = SESSION_PROTOCOL(dcb->session,
PlainProtocol);
{
gwbuf_set_type(read_buffer, GWBUF_TYPE_MYSQL);
router->clientReply(router_instance, session->router_session, read_buffer, dcb);
rc = 1;
}
}
else /*< session is closing; replying to client isn't possible */
{
gwbuf_free(read_buffer);
}
}
return_rc:
return rc;
return_with_lock:
goto return_rc;
}
/*
* EPOLLOUT handler for the MySQL Backend protocol module.
*
* @param dcb The descriptor control block
* @return 1 in success, 0 in case of failure,
*/
static int plain_write_ready_backend_event(DCB *dcb) {
int rc = 0;
PlainProtocol *backend_protocol = dcb->protocol;
/*<
* Don't write to backend if backend_dcb is not in poll set anymore.
*/
if (dcb->state != DCB_STATE_POLLING) {
uint8_t* data;
if (dcb->writeq != NULL)
{
data = (uint8_t *)GWBUF_DATA(dcb->writeq);
}
else
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_write_backend_event] Dcb %p in state %s "
"but there's nothing to write either.",
pthread_self(),
dcb,
STRDCBSTATE(dcb->state))));
rc = 1;
}
goto return_rc;
}
dcb_drain_writeq(dcb);
rc = 1;
return_rc:
return rc;
}
/**
* plain_do_connect_to_backend
*
* This routine creates socket and connects to a backend server.
* Connect it non-blocking operation. If connect fails, socket is closed.
*
* @param host The host to connect to
* @param port The host TCP/IP port
* @param *fd where connected fd is copied
* @return 0/1 on success and -1 on failure
* If successful, fd has file descriptor to socket which is connected to
* backend server. In failure, fd == -1 and socket is closed.
*
*/
int plain_do_connect_to_backend(
char *host,
int port,
int *fd)
{
struct sockaddr_in serv_addr;
int rv;
int so = 0;
int bufsize;
memset(&serv_addr, 0, sizeof serv_addr);
serv_addr.sin_family = AF_INET;
so = socket(AF_INET,SOCK_STREAM,0);
if (so < 0) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Establishing connection to backend server "
"%s:%d failed.\n\t\t Socket creation failed "
"due %d, %s.",
host,
port,
errno,
strerror(errno))));
rv = -1;
goto return_rv;
}
/* prepare for connect */
setipaddress(&serv_addr.sin_addr, host);
serv_addr.sin_port = htons(port);
bufsize = PLAIN_BACKEND_SO_SNDBUF;
if(setsockopt(so, SOL_SOCKET, SO_SNDBUF, &bufsize, sizeof(bufsize)) != 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Failed to set socket options "
"%s:%d failed.\n\t\t Socket configuration failed "
"due %d, %s.",
host,
port,
errno,
strerror(errno))));
rv = -1;
/** Close socket */
goto close_so;
}
bufsize = PLAIN_BACKEND_SO_RCVBUF;
if(setsockopt(so, SOL_SOCKET, SO_RCVBUF, &bufsize, sizeof(bufsize)) != 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Failed to set socket options "
"%s:%d failed.\n\t\t Socket configuration failed "
"due %d, %s.",
host,
port,
errno,
strerror(errno))));
rv = -1;
/** Close socket */
goto close_so;
}
/* set socket to as non-blocking here */
setnonblocking(so);
rv = connect(so, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
if (rv != 0)
{
if (errno == EINPROGRESS)
{
rv = 1;
}
else
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Failed to connect backend server %s:%d, "
"due %d, %s.",
host,
port,
errno,
strerror(errno))));
/** Close socket */
goto close_so;
}
}
*fd = so;
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"%lu [plain_do_connect_to_backend] Connected to backend server "
"%s:%d, fd %d.",
pthread_self(),
host,
port,
so)));
return_rv:
return rv;
close_so:
/*< Close newly created socket. */
if (close(so) != 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Failed to "
"close socket %d due %d, %s.",
so,
errno,
strerror(errno))));
}
goto return_rv;
}
/*
* Write function for backend DCB. Store command to protocol.
*
* @param dcb The DCB of the backend
* @param queue Queue of buffers to write
* @return 0 on failure, 1 on success
*/
static int
plain_write_backend(DCB *dcb, GWBUF *queue)
{
PlainProtocol *backend_protocol = dcb->protocol;
int rc = 0;
rc = dcb_write(dcb, queue);
return rc;
}
/**
* Error event handler.
* Create error message, pass it to router's error handler and if error
* handler fails in providing enough backend servers, mark session being
* closed and call DCB close function which triggers closing router session
* and related backends (if any exists.
*/
static int plain_error_backend_event(DCB *dcb)
{
SESSION* session;
void* rsession;
ROUTER_OBJECT* router;
ROUTER* router_instance;
GWBUF* errbuf;
bool succp;
session_state_t ses_state;
CHK_DCB(dcb);
session = dcb->session;
CHK_SESSION(session);
rsession = session->router_session;
router = session->service->router;
router_instance = session->service->router_instance;
/**
* Avoid running redundant error handling procedure.
* dcb_close is already called for the DCB. Thus, either connection is
* closed by router and COM_QUIT sent or there was an error which
* have already been handled.
*/
if (dcb->state != DCB_STATE_POLLING)
{
int error, len;
char buf[100];
len = sizeof(error);
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, (socklen_t *)&len) == 0)
{
if (error != 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"DCB in state %s got error '%s'.",
STRDCBSTATE(dcb->state),
buf)));
}
}
return 1;
}
spinlock_acquire(&session->ses_lock);
ses_state = session->state;
spinlock_release(&session->ses_lock);
/**
* Session might be initialized when DCB already is in the poll set.
* Thus hangup can occur in the middle of session initialization.
* Only complete and successfully initialized sessions allow for
* calling error handler.
*/
while (ses_state == SESSION_STATE_READY)
{
spinlock_acquire(&session->ses_lock);
ses_state = session->state;
spinlock_release(&session->ses_lock);
}
if (ses_state != SESSION_STATE_ROUTER_READY)
{
int error, len;
char buf[100];
len = sizeof(error);
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, (socklen_t *)&len) == 0)
{
if (error != 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error '%s' in session that is not ready for routing.",
buf)));
}
}
gwbuf_free(errbuf);
goto retblock;
}
#if defined(SS_DEBUG)
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Backend error event handling.")));
#endif
router->handleError(router_instance,
rsession,
errbuf,
dcb,
ERRACT_NEW_CONNECTION,
&succp);
gwbuf_free(errbuf);
/**
* If error handler fails it means that routing session can't continue
* and it must be closed. In success, only this DCB is closed.
*/
if (!succp)
{
spinlock_acquire(&session->ses_lock);
session->state = SESSION_STATE_STOPPING;
spinlock_release(&session->ses_lock);
}
ss_dassert(dcb->dcb_errhandle_called);
dcb_close(dcb);
retblock:
return 1;
}
/*
* Create a new backend connection.
*
* This routine will connect to a backend server and it is called by dbc_connect
* in router->newSession
*
* @param backend_dcb, in, out, use - backend DCB allocated from dcb_connect
* @param server, in, use - server to connect to
* @param session, in use - current session from client DCB
* @return 0/1 on Success and -1 on Failure.
* If succesful, returns positive fd to socket which is connected to
* backend server. Positive fd is copied to protocol and to dcb.
* If fails, fd == -1 and socket is closed.
*/
static int plain_create_backend_connection(
DCB *backend_dcb,
SERVER *server,
SESSION *session)
{
PlainProtocol *protocol = NULL;
int rv = -1;
int fd = -1;
protocol = mysql_protocol_init(backend_dcb, -1);
ss_dassert(protocol != NULL);
if (protocol == NULL) {
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_create_backend_connection] Failed to create "
"protocol object for backend connection.",
pthread_self())));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Failed to create "
"protocol object for backend connection.")));
goto return_fd;
}
/*< if succeed, fd > 0, -1 otherwise */
rv = plain_do_connect_to_backend(server->name, server->port, &fd);
/*< Assign protocol with backend_dcb */
backend_dcb->protocol = protocol;
/*< Set protocol state */
switch (rv) {
case 0:
ss_dassert(fd > 0);
protocol->fd = fd;
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_create_backend_connection] Established "
"connection to %s:%i, protocol fd %d client "
"fd %d.",
pthread_self(),
server->name,
server->port,
protocol->fd,
session->client->fd)));
break;
case 1:
ss_dassert(fd > 0);
protocol->fd = fd;
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_create_backend_connection] Connection "
"pending to %s:%i, protocol fd %d client fd %d.",
pthread_self(),
server->name,
server->port,
protocol->fd,
session->client->fd)));
break;
default:
ss_dassert(fd == -1);
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_create_backend_connection] Connection "
"failed to %s:%i, protocol fd %d client fd %d.",
pthread_self(),
server->name,
server->port,
protocol->fd,
session->client->fd)));
break;
} /*< switch */
return_fd:
return fd;
}
/**
* Error event handler.
* Create error message, pass it to router's error handler and if error
* handler fails in providing enough backend servers, mark session being
* closed and call DCB close function which triggers closing router session
* and related backends (if any exists.
*
* @param dcb The current Backend DCB
* @return 1 always
*/
static int
plain_backend_hangup(DCB *dcb)
{
SESSION* session;
void* rsession;
ROUTER_OBJECT* router;
ROUTER* router_instance;
bool succp;
GWBUF* errbuf;
session_state_t ses_state;
CHK_DCB(dcb);
session = dcb->session;
CHK_SESSION(session);
rsession = session->router_session;
router = session->service->router;
router_instance = session->service->router_instance;
spinlock_acquire(&session->ses_lock);
ses_state = session->state;
spinlock_release(&session->ses_lock);
/**
* Session might be initialized when DCB already is in the poll set.
* Thus hangup can occur in the middle of session initialization.
* Only complete and successfully initialized sessions allow for
* calling error handler.
*/
while (ses_state == SESSION_STATE_READY)
{
spinlock_acquire(&session->ses_lock);
ses_state = session->state;
spinlock_release(&session->ses_lock);
}
if (ses_state != SESSION_STATE_ROUTER_READY)
{
int error, len;
char buf[100];
len = sizeof(error);
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, (socklen_t *)&len) == 0)
{
if (error != 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Hangup in session that is not ready for routing, "
"Error reported is '%s'.",
buf)));
}
}
gwbuf_free(errbuf);
goto retblock;
}
#if defined(SS_DEBUG)
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Backend hangup error handling.")));
#endif
router->handleError(router_instance,
rsession,
errbuf,
dcb,
ERRACT_NEW_CONNECTION,
&succp);
gwbuf_free(errbuf);
/** There are no required backends available, close session. */
if (!succp)
{
#if defined(SS_DEBUG)
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Backend hangup -> closing session.")));
#endif
spinlock_acquire(&session->ses_lock);
session->state = SESSION_STATE_STOPPING;
spinlock_release(&session->ses_lock);
}
ss_dassert(dcb->dcb_errhandle_called);
dcb_close(dcb);
retblock:
return 1;
}
/**
* Send COM_QUIT to backend so that it can be closed.
* @param dcb The current Backend DCB
* @return 1 always
*/
static int
plain_backend_close(DCB *dcb)
{
DCB* client_dcb;
SESSION* session;
GWBUF* quitbuf;
CHK_DCB(dcb);
session = dcb->session;
CHK_SESSION(session);
/**
* The lock is needed only to protect the read of session->state and
* session->client values. Client's state may change by other thread
* but client's close and adding client's DCB to zombies list is executed
* only if client's DCB's state does _not_ change in parallel.
*/
spinlock_acquire(&session->ses_lock);
/**
* If session->state is STOPPING, start closing client session.
* Otherwise only this backend connection is closed.
*/
if (session != NULL &&
session->state == SESSION_STATE_STOPPING &&
session->client != NULL)
{
if (session->client->state == DCB_STATE_POLLING)
{
spinlock_release(&session->ses_lock);
/** Close client DCB */
dcb_close(session->client);
}
else
{
spinlock_release(&session->ses_lock);
}
}
else
{
spinlock_release(&session->ses_lock);
}
return 1;
}
/**
* This routine put into the delay queue the input queue
* The input is what backend DCB is receiving
* The routine is called from func.write() when mysql backend connection
* is not yet complete buu there are inout data from client
*
* @param dcb The current backend DCB
* @param queue Input data in the GWBUF struct
*/
static void backend_set_delayqueue(DCB *dcb, GWBUF *queue) {
spinlock_acquire(&dcb->delayqlock);
if (dcb->delayq) {
/* Append data */
dcb->delayq = gwbuf_append(dcb->delayq, queue);
} else {
if (queue != NULL) {
/* create the delay queue */
dcb->delayq = queue;
}
}
spinlock_release(&dcb->delayqlock);
}
/**
* This routine writes the delayq via dcb_write
* The dcb->delayq contains data received from the client before
* mysql backend authentication succeded
*
* @param dcb The current backend DCB
* @return The dcb_write status
*/
static int backend_write_delayqueue(DCB *dcb)
{
GWBUF *localq = NULL;
int rc;
spinlock_acquire(&dcb->delayqlock);
if (dcb->delayq == NULL)
{
spinlock_release(&dcb->delayqlock);
rc = 1;
}
else
{
rc = dcb_write(dcb, localq);
}
if (rc == 0)
{
#if defined(SS_DEBUG)
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Backend write delayqueue error handling.")));
#endif
}
return rc;
}

714
server/modules/protocol/plainclient.c
View File

@ -1,714 +0,0 @@
/*
* This file is distributed as part of the MariaDB Corporation 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 MariaDB Corporation Ab 2013-2015
*/
/**
* @file mongo_client.c
*
* Revision History
* Date Who Description
*
*/
#include <skygw_utils.h>
#include <log_manager.h>
#include <gw.h>
#include <modinfo.h>
#include <sys/stat.h>
#include <modutil.h>
#include <plainprotocol.h>
MODULE_INFO info = {
MODULE_API_PROTOCOL,
MODULE_GA,
GWPROTOCOL_VERSION,
"The plain client protocol"
};
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[];
extern __thread log_info_t tls_log_info;
static char *version_str = "V1.0.0";
static int plain_accept(DCB *listener);
static int plain_listen(DCB *listener, char *config_bind);
static int plain_read(DCB* dcb);
static int plain_write_ready(DCB *dcb);
static int plain_write(DCB *dcb, GWBUF *queue);
static int plain_client_error(DCB *dcb);
static int plain_client_close(DCB *dcb);
static int plain_client_hangup_event(DCB *dcb);
/*
* The "module object" for the mysqld client protocol module.
*/
static GWPROTOCOL MyObject = {
plain_read, /* Read - EPOLLIN handler */
plain_write, /* Write - data from gateway */
plain_write_ready, /* WriteReady - EPOLLOUT handler */
plain_client_error, /* Error - EPOLLERR handler */
plain_client_hangup_event, /* HangUp - EPOLLHUP handler */
plain_accept, /* Accept */
NULL, /* Connect */
plain_client_close, /* Close */
plain_listen, /* Listen */
NULL, /* Authentication */
NULL /* Session */
};
/**
* 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()
{
}
/**
* 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
*/
GWPROTOCOL *
GetModuleObject()
{
return &MyObject;
}
/**
* Write function for client DCB: writes data from MaxScale to Client
*
* @param dcb The DCB of the client
* @param queue Queue of buffers to write
*/
int
plain_write(DCB *dcb, GWBUF *queue)
{
return dcb_write(dcb, queue);
}
/**
* Client read event triggered by EPOLLIN
*
* @param dcb Descriptor control block
* @return 0 if succeed, 1 otherwise
*/
int plain_read(
DCB* dcb)
{
SESSION *session = NULL;
ROUTER_OBJECT *router = NULL;
ROUTER *router_instance = NULL;
void *rsession = NULL;
PlainProtocol *protocol = NULL;
GWBUF *read_buffer = NULL;
int rc = 0;
int nbytes_read = 0;
uint8_t cap = 0;
bool stmt_input = false; /*< router input type */
CHK_DCB(dcb);
protocol = DCB_PROTOCOL(dcb, PlainProtocol);
CHK_PROTOCOL(protocol);
rc = dcb_read(dcb, &read_buffer);
if (rc < 0)
{
dcb_close(dcb);
}
nbytes_read = gwbuf_length(read_buffer);
if (nbytes_read == 0)
{
goto return_rc;
}
if(dcb->session == NULL)
{
dcb->session = session_alloc(dcb->service,dcb);
}
rc = SESSION_ROUTE_QUERY(dcb->session, read_buffer);
return_rc:
return rc;
}
///////////////////////////////////////////////
// client write event to Client triggered by EPOLLOUT
//////////////////////////////////////////////
/**
* @node Client's fd became writable, and EPOLLOUT event
* arrived. As a consequence, client input buffer (writeq) is flushed.
*
* Parameters:
* @param dcb - in, use
* client dcb
*
* @return constantly 1
*
*
* @details (write detailed description here)
*
*/
int plain_write_ready(DCB *dcb)
{
PlainProtocol *protocol = NULL;
CHK_DCB(dcb);
ss_dassert(dcb->state != DCB_STATE_DISCONNECTED);
if (dcb == NULL) {
goto return_1;
}
if (dcb->state == DCB_STATE_DISCONNECTED) {
goto return_1;
}
if (dcb->protocol == NULL) {
goto return_1;
}
protocol = (PlainProtocol *)dcb->protocol;
dcb_drain_writeq(dcb);
goto return_1;
return_1:
return 1;
}
/**
* set listener for mysql protocol, retur 1 on success and 0 in failure
*/
int plain_listen(
DCB *listen_dcb,
char *config_bind)
{
int l_so;
int syseno = 0;
struct sockaddr_in serv_addr;
struct sockaddr_un local_addr;
struct sockaddr *current_addr;
int one = 1;
int rc;
if (strchr(config_bind, '/')) {
char *tmp = strrchr(config_bind, ':');
if (tmp)
*tmp = '\0';
// UNIX socket create
if ((l_so = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
fprintf(stderr,
"\n* Error: can't create UNIX socket due "
"error %i, %s.\n\n\t",
errno,
strerror(errno));
return 0;
}
memset(&local_addr, 0, sizeof(local_addr));
local_addr.sun_family = AF_UNIX;
strncpy(local_addr.sun_path, config_bind, sizeof(local_addr.sun_path) - 1);
current_addr = (struct sockaddr *) &local_addr;
} else {
/* MaxScale, as default, will bind on port 4406 */
if (!parse_bindconfig(config_bind, 4406, &serv_addr)) {
fprintf(stderr, "Error in parse_bindconfig for [%s]\n", config_bind);
return 0;
}
// TCP socket create
if ((l_so = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
fprintf(stderr,
"\n* Error: can't create socket due "
"error %i, %s.\n\n\t",
errno,
strerror(errno));
return 0;
}
current_addr = (struct sockaddr *) &serv_addr;
}
listen_dcb->fd = -1;
// socket options
if((syseno = setsockopt(l_so, SOL_SOCKET, SO_REUSEADDR, (char *)&one, sizeof(one))) != 0){
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,"Error: Failed to set socket options. Error %d: %s",errno,strerror(errno))));
}
// set NONBLOCKING mode
setnonblocking(l_so);
/* get the right socket family for bind */
switch (current_addr->sa_family) {
case AF_UNIX:
rc = unlink(config_bind);
if ( (rc == -1) && (errno!=ENOENT) ) {
fprintf(stderr, "Error unlink Unix Socket %s\n", config_bind);
}
if (bind(l_so, (struct sockaddr *) &local_addr, sizeof(local_addr)) < 0) {
fprintf(stderr,
"\n* Bind failed due error %i, %s.\n",
errno,
strerror(errno));
fprintf(stderr, "* Can't bind to %s\n\n", config_bind);
close(l_so);
return 0;
}
/* set permission for all users */
if (chmod(config_bind, 0777) < 0) {
fprintf(stderr,
"\n* chmod failed for %s due error %i, %s.\n\n",
config_bind,
errno,
strerror(errno));
}
break;
case AF_INET:
if (bind(l_so, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) {
fprintf(stderr,
"\n* Bind failed due error %i, %s.\n",
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;
}
rc = listen(l_so, 10 * SOMAXCONN);
if (rc == 0) {
LOGIF(LM, (skygw_log_write_flush(LOGFILE_MESSAGE,"Listening MySQL connections at %s", config_bind)));
} else {
int eno = errno;
errno = 0;
fprintf(stderr,
"\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
listen_dcb->fd = l_so;
// add listening socket to poll structure
if (poll_add_dcb(listen_dcb) == -1) {
fprintf(stderr,
"\n* Failed to start polling the socket due error "
"%i, %s.\n\n",
errno,
strerror(errno));
return 0;
}
#if defined(FAKE_CODE)
conn_open[l_so] = true;
#endif /* FAKE_CODE */
listen_dcb->func.accept = plain_accept;
return 1;
}
/**
* @node (write brief function description here)
*
* Parameters:
* @param listener - <usage>
* <description>
*
* @return 0 in success, 1 in failure
*
*
* @details (write detailed description here)
*
*/
int plain_accept(DCB *listener)
{
int rc = 0;
DCB *client_dcb;
PlainProtocol *protocol;
int c_sock;
struct sockaddr client_conn;
socklen_t client_len = sizeof(struct sockaddr_storage);
int sendbuf = GW_BACKEND_SO_SNDBUF;
socklen_t optlen = sizeof(sendbuf);
int eno = 0;
int syseno = 0;
int i = 0;
CHK_DCB(listener);
while (1) {
retry_accept:
// new connection from client
c_sock = accept(listener->fd,
(struct sockaddr *) &client_conn,
&client_len);
eno = errno;
errno = 0;
if (c_sock == -1) {
if (eno == EAGAIN || eno == EWOULDBLOCK)
{
/**
* We have processed all incoming connections.
*/
rc = 1;
goto return_rc;
}
else if (eno == ENFILE || eno == EMFILE)
{
struct timespec ts1;
ts1.tv_sec = 0;
/**
* Exceeded system's (ENFILE) or processes
* (EMFILE) max. number of files limit.
*/
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [plain_accept] Error %d, %s. ",
pthread_self(),
eno,
strerror(eno))));
if (i == 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error %d, %s. "
"Failed to accept new client "
"connection.",
eno,
strerror(eno))));
}
i++;
ts1.tv_nsec = 100*i*i*1000000;
nanosleep(&ts1, NULL);
if (i<10) {
goto retry_accept;
}
rc = 1;
goto return_rc;
}
else
{
/**
* Other error.
*/
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [plain_accept] Error %d, %s.",
pthread_self(),
eno,
strerror(eno))));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Failed to accept new client "
"connection due to %d, %s.",
eno,
strerror(eno))));
rc = 1;
goto return_rc;
} /* if (eno == ..) */
} /* if (c_sock == -1) */
/* reset counter */
i = 0;
listener->stats.n_accepts++;
#if defined(SS_DEBUG)
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"%lu [plain_accept] Accepted fd %d.",
pthread_self(),
c_sock)));
#endif /* SS_DEBUG */
/* set nonblocking */
sendbuf = GW_CLIENT_SO_SNDBUF;
if((syseno = setsockopt(c_sock, SOL_SOCKET, SO_SNDBUF, &sendbuf, optlen)) != 0){
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,"Error: Failed to set socket options. Error %d: %s",errno,strerror(errno))));
}
sendbuf = GW_CLIENT_SO_RCVBUF;
if((syseno = setsockopt(c_sock, SOL_SOCKET, SO_RCVBUF, &sendbuf, optlen)) != 0){
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,"Error: Failed to set socket options. Error %d: %s",errno,strerror(errno))));
}
setnonblocking(c_sock);
client_dcb = dcb_alloc(DCB_ROLE_REQUEST_HANDLER);
if (client_dcb == NULL) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Failed to create "
"DCB object for client connection.")));
close(c_sock);
rc = 1;
goto return_rc;
}
client_dcb->service = listener->session->service;
client_dcb->fd = c_sock;
// get client address
if ( client_conn.sa_family == AF_UNIX)
{
// client address
client_dcb->remote = strdup("localhost_from_socket");
// set localhost IP for user authentication
(client_dcb->ipv4).sin_addr.s_addr = 0x0100007F;
}
else
{
/* client IPv4 in raw data*/
memcpy(&client_dcb->ipv4,
(struct sockaddr_in *)&client_conn,
sizeof(struct sockaddr_in));
/* client IPv4 in string representation */
client_dcb->remote = (char *)calloc(INET_ADDRSTRLEN+1, sizeof(char));
if (client_dcb->remote != NULL)
{
inet_ntop(AF_INET,
&(client_dcb->ipv4).sin_addr,
client_dcb->remote,
INET_ADDRSTRLEN);
}
}
protocol = mysql_protocol_init(client_dcb, c_sock);
ss_dassert(protocol != NULL);
if (protocol == NULL) {
/** delete client_dcb */
dcb_close(client_dcb);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"%lu [plain_accept] Failed to create "
"protocol object for client connection.",
pthread_self())));
rc = 1;
goto return_rc;
}
client_dcb->protocol = protocol;
// assign function poiters to "func" field
memcpy(&client_dcb->func, &MyObject, sizeof(GWPROTOCOL));
/**
* Set new descriptor to event set. At the same time,
* change state to DCB_STATE_POLLING so that
* thread which wakes up sees correct state.
*/
if (poll_add_dcb(client_dcb) == -1)
{
/* Send a custom error as MySQL command reply */
/** close client_dcb */
dcb_close(client_dcb);
/** Previous state is recovered in poll_add_dcb. */
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"%lu [plain_accept] Failed to add dcb %p for "
"fd %d to epoll set.",
pthread_self(),
client_dcb,
client_dcb->fd)));
rc = 1;
goto return_rc;
}
else
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [plain_accept] Added dcb %p for fd "
"%d to epoll set.",
pthread_self(),
client_dcb,
client_dcb->fd)));
}
} /**< while 1 */
#if defined(SS_DEBUG)
if (rc == 0) {
CHK_DCB(client_dcb);
CHK_PROTOCOL(((PlainProtocol *)client_dcb->protocol));
}
#endif
return_rc:
return rc;
}
static int plain_client_error(
DCB* dcb)
{
SESSION* session;
CHK_DCB(dcb);
session = dcb->session;
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [plain_client_error] Error event handling for DCB %p "
"in state %s, session %p.",
pthread_self(),
dcb,
STRDCBSTATE(dcb->state),
(session != NULL ? session : NULL))));
if (session != NULL && session->state == SESSION_STATE_STOPPING)
{
goto retblock;
}
dcb_close(dcb);
retblock:
return 1;
}
static int
plain_client_close(DCB *dcb)
{
SESSION* session;
ROUTER_OBJECT* router;
void* router_instance;
#if defined(SS_DEBUG)
PlainProtocol* protocol = (PlainProtocol *)dcb->protocol;
if (dcb->state == DCB_STATE_POLLING ||
dcb->state == DCB_STATE_NOPOLLING ||
dcb->state == DCB_STATE_ZOMBIE)
{
if (!DCB_IS_CLONE(dcb)) CHK_PROTOCOL(protocol);
}
#endif
LOGIF(LD, (skygw_log_write(LOGFILE_DEBUG,
"%lu [plain_client_close]",
pthread_self())));
mysql_protocol_done(dcb);
session = dcb->session;
/**
* session may be NULL if session_alloc failed.
* In that case, router session wasn't created.
*/
if (session != NULL)
{
CHK_SESSION(session);
spinlock_acquire(&session->ses_lock);
if (session->state != SESSION_STATE_STOPPING)
{
session->state = SESSION_STATE_STOPPING;
}
router_instance = session->service->router_instance;
router = session->service->router;
/**
* If router session is being created concurrently router
* session might be NULL and it shouldn't be closed.
*/
if (session->router_session != NULL)
{
spinlock_release(&session->ses_lock);
/** Close router session and all its connections */
router->closeSession(router_instance, session->router_session);
}
else
{
spinlock_release(&session->ses_lock);
}
}
return 1;
}
/**
* Handle a hangup event on the client side descriptor.
*
* We simply close the DCB, this will propogate the closure to any
* backend descriptors and perform the session cleanup.
*
* @param dcb The DCB of the connection
*/
static int
plain_client_hangup_event(DCB *dcb)
{
SESSION* session;
CHK_DCB(dcb);
session = dcb->session;
if (session != NULL && session->state == SESSION_STATE_ROUTER_READY)
{
CHK_SESSION(session);
}
if (session != NULL && session->state == SESSION_STATE_STOPPING)
{
goto retblock;
}
dcb_close(dcb);
retblock:
return 1;
}

923
server/modules/routing/plainroute.c
View File

@ -1,923 +0,0 @@
/*
* This file is distributed as part of the MariaDB Corporation 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 MariaDB Corporation Ab 2013-2014
*/
/**
* @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
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.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>
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[];
extern __thread log_info_t tls_log_info;
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 *backend_dcb,
error_action_t action,
bool *succp);
static uint8_t getCapabilities (ROUTER* inst, void* router_session);
/** 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()
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Initialise readconnroute router module %s.\n", 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;
}
/**
* 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(inst);
return NULL;
}
for (sref = service->dbref, n = 0; sref; sref = sref->next)
{
if ((inst->servers[n] = malloc(sizeof(BACKEND))) == NULL)
{
for (i = 0; i < n; i++)
free(inst->servers[i]);
free(inst->servers);
free(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 (n = 0; inst->servers[n]; n++)
{
backend = inst->servers[n];
total += atoi(serverGetParameter(backend->server,
weightby));
}
if (total == 0)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"WARNING: Weighting Parameter for service '%s' "
"will be ignored as no servers have values "
"for the parameter '%s'.\n",
service->name, weightby)));
}
else
{
for (n = 0; inst->servers[n]; n++)
{
int perc, wght;
backend = inst->servers[n];
perc = ((wght = atoi(serverGetParameter(backend->server,
weightby))) * 1000) / total;
if (perc == 0 && wght != 0)
perc = 1;
backend->weight = perc;
if (perc == 0)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Server '%s' has no value "
"for weighting parameter '%s', "
"no queries will be routed to "
"this server.\n",
inst->servers[n]->server->unique_name,
weightby)));
}
}
}
}
/*
* Process the options
*/
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], "synced"))
{
inst->bitmask |= (SERVER_JOINED);
inst->bitvalue |= SERVER_JOINED;
}
else if (!strcasecmp(options[i], "ndb"))
{
inst->bitmask |= (SERVER_NDB);
inst->bitvalue |= SERVER_NDB;
}
else
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"* Warning : Unsupported router "
"option \'%s\' for readconnroute. "
"Expected router options are "
"[slave|master|synced|ndb]",
options[i])));
}
}
}
/*
* 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;
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_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
/**
* 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 (SERVER_IN_MAINT(inst->servers[i]->server))
continue;
if (inst->servers[i]->weight == 0)
continue;
/* 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
* 1000) / inst->servers[i]->weight <
(candidate->current_connection_count *
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
* 1000) / inst->servers[i]->weight ==
(candidate->current_connection_count *
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];
}
}
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;
LOGIF(LD, (skygw_log_write(
LOGFILE_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);
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);
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_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 */
atomic_add(&router_cli_ses->backend->server->stats.n_current, -1);
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;
uint8_t *payload = GWBUF_DATA(queue);
int mysql_command;
int rc;
DCB* backend_dcb;
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)
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Error : Failed to route MySQL command %d to backend "
"server.",
mysql_command)));
rc = 0;
goto return_rc;
}
rc = backend_dcb->func.write(backend_dcb, queue);
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)
{
DCB *client ;
client = backend_dcb->session->client;
ss_dassert(client != NULL);
SESSION_ROUTE_REPLY(backend_dcb->session, queue);
}
/**
* Error Handler routine
*
* The routine will handle errors that occurred in backend writes.
*
* @param instance The router instance
* @param router_session The router session
* @param message The error message to reply
* @param backend_dcb The backend DCB
* @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)
{
DCB *client_dcb;
SESSION *session = backend_dcb->session;
session_state_t sesstate;
/** Reset error handle flag from a given DCB */
if (action == ERRACT_RESET)
{
backend_dcb->dcb_errhandle_called = false;
return;
}
/** Don't handle same error twice on same DCB */
if (backend_dcb->dcb_errhandle_called)
{
/** we optimistically assume that previous call succeed */
*succp = true;
return;
}
else
{
backend_dcb->dcb_errhandle_called = true;
}
spinlock_acquire(&session->ses_lock);
sesstate = session->state;
client_dcb = session->client;
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);
}
/** 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 uint8_t getCapabilities(
ROUTER* inst,
void* router_session)
{
return 0;
}
/********************************
* 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 && servers[i]->server->depth < master_host->server->depth) {
master_host = servers[i];
} else {
if (master_host == NULL) {
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;
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;
}