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

Merge, mostly.

Browse Source
This commit is contained in:
vraatikka
2013-08-28 23:26:09 +03:00
parent 48165bf5cd
commit 527df6c7f3
2 changed files with 317 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
server/modules/protocol/mysql_backend.c
View File

@ -52,8 +52,7 @@ static int backend_write_delayqueue(DCB *dcb);
static void backend_set_delayqueue(DCB *dcb, GWBUF *queue); static void backend_set_delayqueue(DCB *dcb, GWBUF *queue);
static int gw_change_user(DCB *backend_dcb, SERVER *server, SESSION *in_session, GWBUF *queue); static int gw_change_user(DCB *backend_dcb, SERVER *server, SESSION *in_session, GWBUF *queue);
static int gw_session(DCB *backend_dcb, void *data); static int gw_session(DCB *backend_dcb, void *data);
static MYSQL_session* gw_get_shared_session_auth_info(DCB* dcb);
extern char *gw_strend(register const char *s);
static GWPROTOCOL MyObject = { static GWPROTOCOL MyObject = {
gw_read_backend_event, /* Read - EPOLLIN handler */ gw_read_backend_event, /* Read - EPOLLIN handler */
@ -62,12 +61,12 @@ static GWPROTOCOL MyObject = {
gw_error_backend_event, /* Error - EPOLLERR handler */ gw_error_backend_event, /* Error - EPOLLERR handler */
gw_backend_hangup, /* HangUp - EPOLLHUP handler */ gw_backend_hangup, /* HangUp - EPOLLHUP handler */
NULL, /* Accept */ NULL, /* Accept */
gw_create_backend_connection, /* Connect */ gw_create_backend_connection, /* Connect */
gw_backend_close, /* Close */ gw_backend_close, /* Close */
NULL, /* Listen */ NULL, /* Listen */
gw_change_user, /* Authentication */ gw_change_user, /* Authentication */
gw_session /* Session */ gw_session /* Session */
}; };
/* /*
* Implementation of the mandatory version entry point * Implementation of the mandatory version entry point
@ -110,6 +109,30 @@ GetModuleObject()
} }
static MYSQL_session* gw_get_shared_session_auth_info(
DCB* dcb)
{
MYSQL_session* auth_info = NULL;
CHK_DCB(dcb);
CHK_SESSION(dcb->session);
spinlock_acquire(&dcb->session->ses_lock);
if (dcb->session->state != SESSION_STATE_ALLOC) {
auth_info = dcb->session->data;
} else {
skygw_log_write_flush(
LOGFILE_ERROR,
"%lu [gw_get_shared_session_auth_info] Couldn't get "
"session authentication info. Session in a wrong state %d.",
pthread_self(),
dcb->session->state);
}
spinlock_release(&dcb->session->ses_lock);
return auth_info;
}
/** /**
* Backend Read Event for EPOLLIN on the MySQL backend protocol module * Backend Read Event for EPOLLIN on the MySQL backend protocol module
* @param dcb The backend Descriptor Control Block * @param dcb The backend Descriptor Control Block
@ -119,17 +142,25 @@ static int gw_read_backend_event(DCB *dcb) {
MySQLProtocol *client_protocol = NULL; MySQLProtocol *client_protocol = NULL;
MySQLProtocol *backend_protocol = NULL; MySQLProtocol *backend_protocol = NULL;
MYSQL_session *current_session = NULL; MYSQL_session *current_session = NULL;
int rc = 0;
dcb->state = DCB_STATE_PROCESSING; dcb->state = DCB_STATE_PROCESSING;
if(dcb->session) { if(dcb->session) {
CHK_SESSION(dcb->session);
client_protocol = SESSION_PROTOCOL(dcb->session, MySQLProtocol); client_protocol = SESSION_PROTOCOL(dcb->session, MySQLProtocol);
} }
backend_protocol = (MySQLProtocol *) dcb->protocol; backend_protocol = (MySQLProtocol *) dcb->protocol;
current_session = (MYSQL_session *)dcb->session->data;
//fprintf(stderr, ">>> backend EPOLLIN from %i, command %i, protocol state [%s]\n", dcb->fd, dcb->command, gw_mysql_protocol_state2string(backend_protocol->state)); /** return only with complete session */
current_session = gw_get_shared_session_auth_info(dcb);
ss_dassert(current_session != NULL);
ss_dassert(dcb->session->state != SESSION_STATE_ALLOC);
/* fprintf(stderr, ">>> backend EPOLLIN from %i, command %i,protocol
* state [%s]\n", dcb->fd, dcb->command, gw_mysql_protocol_state2string
* (backend_protocol->state));
*/
/* backend is connected: /* backend is connected:
* *
@ -144,9 +175,8 @@ static int gw_read_backend_event(DCB *dcb) {
current_session->user, current_session->user,
current_session->client_sha1, current_session->client_sha1,
backend_protocol); backend_protocol);
rc = 1;
dcb->state = DCB_STATE_POLLING; goto return_rc;
return 1;
} }
/* ready to check the authentication reply from backend */ /* ready to check the authentication reply from backend */
@ -178,7 +208,6 @@ static int gw_read_backend_event(DCB *dcb) {
current_session->user); current_session->user);
backend_protocol->state = MYSQL_AUTH_FAILED; backend_protocol->state = MYSQL_AUTH_FAILED;
/* send an error to the client */ /* send an error to the client */
mysql_send_custom_error( mysql_send_custom_error(
dcb->session->client, dcb->session->client,
@ -194,7 +223,7 @@ static int gw_read_backend_event(DCB *dcb) {
spinlock_release(&session->ses_lock); spinlock_release(&session->ses_lock);
if (rsession != NULL) { if (rsession != NULL) {
skygw_log_write( skygw_log_write_flush(
LOGFILE_TRACE, LOGFILE_TRACE,
"%lu [gw_read_backend_event] Call " "%lu [gw_read_backend_event] Call "
"closeSession for backend session.", "closeSession for backend session.",
@ -209,8 +238,8 @@ static int gw_read_backend_event(DCB *dcb) {
"backend session.", "backend session.",
pthread_self()); pthread_self());
} }
dcb->state = DCB_STATE_POLLING; rc = 1;
return 1; goto return_rc;
case MYSQL_SUCCESFUL_AUTHENTICATION: case MYSQL_SUCCESFUL_AUTHENTICATION:
skygw_log_write_flush( skygw_log_write_flush(
@ -223,27 +252,26 @@ static int gw_read_backend_event(DCB *dcb) {
current_session->user); current_session->user);
spinlock_acquire(&dcb->authlock); spinlock_acquire(&dcb->authlock);
backend_protocol->state = MYSQL_IDLE; backend_protocol->state = MYSQL_IDLE;
/* check the delay queue and flush the data */ /* check the delay queue and flush the data */
if(dcb->delayq) { if(dcb->delayq) {
backend_write_delayqueue(dcb); backend_write_delayqueue(dcb);
dcb->state = DCB_STATE_POLLING; dcb->state = DCB_STATE_POLLING;
spinlock_release(&dcb->authlock); spinlock_release(&dcb->authlock);
return 1; rc = 1;
goto return_rc;
} }
dcb->state = DCB_STATE_POLLING;
spinlock_release(&dcb->authlock); spinlock_release(&dcb->authlock);
rc = 0;
return 1; goto return_rc;
default: default:
/* no other authentication state here right now, so just return */ /* no other authentication state here right
dcb->state = DCB_STATE_POLLING; * now, so just return */
return 0; rc = 0;
} goto return_rc;
} } /**< switch (rv) */
} /**< if (backend_protocol->state == MYSQL_AUTH_RECV) */
/* reading MySQL command output from backend and writing to the client */ /* reading MySQL command output from backend and writing to the client */
@ -255,13 +283,13 @@ static int gw_read_backend_event(DCB *dcb) {
ROUTER *router_instance = NULL; ROUTER *router_instance = NULL;
void *rsession = NULL; void *rsession = NULL;
SESSION *session = dcb->session; SESSION *session = dcb->session;
int rc = 0;
/* read available backend data */ /* read available backend data */
rc = dcb_read(dcb, &head); rc = dcb_read(dcb, &head);
if (rc == -1) { if (rc == -1) {
return 1; rc = 1;
goto return_rc;
} }
if (session != NULL) { if (session != NULL) {
@ -278,13 +306,13 @@ static int gw_read_backend_event(DCB *dcb) {
/* and pass now the gwbuf to the router */ /* and pass now the gwbuf to the router */
router->clientReply(router_instance, rsession, head, dcb); router->clientReply(router_instance, rsession, head, dcb);
dcb->state = DCB_STATE_POLLING; rc = 1;
goto return_rc;
return 1;
} }
rc = 0;
dcb->state = DCB_STATE_POLLING; return_rc:
return 0; dcb->state = DCB_STATE_POLLING;
return rc;
} }
/* /*
@ -307,10 +335,8 @@ static int gw_write_backend_event(DCB *dcb) {
// spinlock_release(&dcb->connectlock); // spinlock_release(&dcb->connectlock);
dcb->state = DCB_STATE_POLLING; dcb->state = DCB_STATE_POLLING;
return 1; return 1;
} }
// spinlock_release(&dcb->connectlock); // spinlock_release(&dcb->connectlock);
w = dcb_drain_writeq(dcb); w = dcb_drain_writeq(dcb);
@ -384,11 +410,11 @@ static int gw_create_backend_connection(DCB *backend, SERVER *server, SESSION *s
protocol = (MySQLProtocol *) calloc(1, sizeof(MySQLProtocol)); protocol = (MySQLProtocol *) calloc(1, sizeof(MySQLProtocol));
protocol->state = MYSQL_ALLOC; protocol->state = MYSQL_ALLOC;
protocol->protocol_chk_top = CHK_NUM_PROTOCOL;
backend->protocol = protocol; protocol->protocol_chk_tail = CHK_NUM_PROTOCOL;
/* put the backend dcb in the protocol struct */ /* put the backend dcb in the protocol struct */
protocol->descriptor = backend; protocol->descriptor = backend;
backend->protocol = protocol;
s_data = (MYSQL_session *)session->client->data; s_data = (MYSQL_session *)session->client->data;

430
server/modules/protocol/mysql_client.c
View File

@ -342,7 +342,7 @@ MySQLSendHandshake(DCB* dcb)
static int gw_mysql_do_authentication(DCB *dcb, GWBUF *queue) { static int gw_mysql_do_authentication(DCB *dcb, GWBUF *queue) {
MySQLProtocol *protocol = NULL; MySQLProtocol *protocol = NULL;
int compress = -1; /* int compress = -1; */
int connect_with_db = -1; int connect_with_db = -1;
uint8_t *client_auth_packet = GWBUF_DATA(queue); uint8_t *client_auth_packet = GWBUF_DATA(queue);
char *username = NULL; char *username = NULL;
@ -353,9 +353,9 @@ static int gw_mysql_do_authentication(DCB *dcb, GWBUF *queue) {
int auth_ret = -1; int auth_ret = -1;
MYSQL_session *client_data = NULL; MYSQL_session *client_data = NULL;
if (dcb) CHK_DCB(dcb);
protocol = DCB_PROTOCOL(dcb, MySQLProtocol);
protocol = DCB_PROTOCOL(dcb, MySQLProtocol);
client_data = (MYSQL_session *)calloc(1, sizeof(MYSQL_session)); client_data = (MYSQL_session *)calloc(1, sizeof(MYSQL_session));
dcb->data = client_data; dcb->data = client_data;
@ -364,19 +364,28 @@ static int gw_mysql_do_authentication(DCB *dcb, GWBUF *queue) {
memcpy(&protocol->client_capabilities, client_auth_packet + 4, 4); memcpy(&protocol->client_capabilities, client_auth_packet + 4, 4);
connect_with_db = GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB & gw_mysql_get_byte4(&protocol->client_capabilities); connect_with_db =
compress = GW_MYSQL_CAPABILITIES_COMPRESS & gw_mysql_get_byte4(&protocol->client_capabilities); GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB & gw_mysql_get_byte4(
&protocol->client_capabilities);
/*
compress =
GW_MYSQL_CAPABILITIES_COMPRESS & gw_mysql_get_byte4(
&protocol->client_capabilities);
*/
// now get the user // now get the user
strcpy(username, (char *)(client_auth_packet + 4 + 4 + 4 + 1 + 23)); strcpy(username, (char *)(client_auth_packet + 4 + 4 + 4 + 1 + 23));
fprintf(stderr, "<<< Client username is [%s]\n", username); fprintf(stderr, "<<< Client username is [%s]\n", username);
// get the auth token len // get the auth token len
memcpy(&auth_token_len, client_auth_packet + 4 + 4 + 4 + 1 + 23 + strlen(username) + 1, 1); memcpy(&auth_token_len,
client_auth_packet + 4 + 4 + 4 + 1 + 23 + strlen(username) + 1,
1);
if (connect_with_db) { if (connect_with_db) {
database = client_data->db; database = client_data->db;
strcpy(database, (char *)(client_auth_packet + 4 + 4 + 4 + 1 + 23 + strlen(username) + 1 + 1 + auth_token_len)); strcpy(database,
(char *)(client_auth_packet + 4 + 4 + 4 + 1 + 23 + strlen(username) +
1 + 1 + auth_token_len));
fprintf(stderr, "<<< Client selected db is [%s]\n", database); fprintf(stderr, "<<< Client selected db is [%s]\n", database);
} else { } else {
fprintf(stderr, "<<< Client is NOT connected with db\n"); fprintf(stderr, "<<< Client is NOT connected with db\n");
@ -385,12 +394,19 @@ static int gw_mysql_do_authentication(DCB *dcb, GWBUF *queue) {
// allocate memory for token only if auth_token_len > 0 // allocate memory for token only if auth_token_len > 0
if (auth_token_len) { if (auth_token_len) {
auth_token = (uint8_t *)malloc(auth_token_len); auth_token = (uint8_t *)malloc(auth_token_len);
memcpy(auth_token, client_auth_packet + 4 + 4 + 4 + 1 + 23 + strlen(username) + 1 + 1, auth_token_len); memcpy(auth_token,
client_auth_packet + 4 + 4 + 4 + 1 + 23 + strlen(username) + 1 + 1,
auth_token_len);
} }
// decode the token and check the password // decode the token and check the password
// Note: if auth_token_len == 0 && auth_token == NULL, user is without password // Note: if auth_token_len == 0 && auth_token == NULL, user is without password
auth_ret = gw_check_mysql_scramble_data(dcb, auth_token, auth_token_len, protocol->scramble, sizeof(protocol->scramble), username, stage1_hash); auth_ret = gw_check_mysql_scramble_data(dcb,
auth_token,
auth_token_len,
protocol->scramble, sizeof(protocol->scramble),
username,
stage1_hash);
// let's free the auth_token now // let's free the auth_token now
if (auth_token) if (auth_token)
@ -442,7 +458,9 @@ int w, saved_errno = 0;
while (queue != NULL) while (queue != NULL)
{ {
len = GWBUF_LENGTH(queue); len = GWBUF_LENGTH(queue);
GW_NOINTR_CALL(w = write(dcb->fd, GWBUF_DATA(queue), len); dcb->stats.n_writes++); GW_NOINTR_CALL(
w = write(dcb->fd,GWBUF_DATA(queue), len);
dcb->stats.n_writes++);
saved_errno = errno; saved_errno = errno;
if (w < 0) if (w < 0)
{ {
@ -484,33 +502,38 @@ int w, saved_errno = 0;
* @return 0 if succeed, 1 otherwise * @return 0 if succeed, 1 otherwise
*/ */
int gw_read_client_event(DCB* dcb) { int gw_read_client_event(DCB* dcb) {
SESSION *session = NULL; SESSION *session = NULL;
ROUTER_OBJECT *router = NULL; ROUTER_OBJECT *router = NULL;
ROUTER *router_instance = NULL; ROUTER *router_instance = NULL;
void *rsession = NULL; void *rsession = NULL;
MySQLProtocol *protocol = NULL; MySQLProtocol *protocol = NULL;
int b = -1; int b = -1;
int rc = 0;
spinlock_acquire(&dcb->writeqlock); CHK_DCB(dcb);
if (dcb->state == DCB_STATE_DISCONNECTED || if (dcb->state == DCB_STATE_DISCONNECTED ||
dcb->state == DCB_STATE_FREED || dcb->state == DCB_STATE_FREED ||
dcb->state == DCB_STATE_ZOMBIE) dcb->state == DCB_STATE_ZOMBIE ||
dcb->state == DCB_STATE_PROCESSING)
{ {
spinlock_release(&dcb->writeqlock); rc = 1;
return 1; goto return_rc;
} }
ss_dassert(dcb->state == DCB_STATE_POLLING);
dcb->state = DCB_STATE_PROCESSING;
dcb->state = DCB_STATE_PROCESSING; protocol = DCB_PROTOCOL(dcb, MySQLProtocol);
CHK_PROTOCOL(protocol);
protocol = DCB_PROTOCOL(dcb, MySQLProtocol); /**
* Check how many bytes are readable in dcb->fd.
*/
if (ioctl(dcb->fd, FIONREAD, &b)) { if (ioctl(dcb->fd, FIONREAD, &b)) {
int eno = errno; int eno = errno;
errno = 0; errno = 0;
skygw_log_write( skygw_log_write(
LOGFILE_ERROR, LOGFILE_ERROR,
"%lu [gw_read_client_event] Setting FIONREAD for fd " "%lu [gw_read_client_event] ioctl FIONREAD for fd "
"%d failed. errno %d, %s. dcb->state = %d", "%d failed. errno %d, %s. dcb->state = %d",
pthread_self(), pthread_self(),
dcb->fd, dcb->fd,
@ -518,180 +541,180 @@ int gw_read_client_event(DCB* dcb) {
strerror(eno), strerror(eno),
dcb->state); dcb->state);
dcb->state = DCB_STATE_POLLING; dcb->state = DCB_STATE_POLLING;
spinlock_release(&dcb->writeqlock); rc = 1;
goto return_rc;
return 1;
} else { } else {
//fprintf(stderr, "Client IOCTL FIONREAD bytes to read = %i\n", b); //fprintf(stderr, "Client IOCTL FIONREAD bytes to read = %i\n", b);
} }
spinlock_release(&dcb->writeqlock);
switch (protocol->state) { switch (protocol->state) {
case MYSQL_AUTH_SENT: case MYSQL_AUTH_SENT:
/*
* Read all the data that is available into a chain of buffers
*/
{
int len = -1;
GWBUF *queue = NULL;
GWBUF *gw_buffer = NULL;
int auth_val = -1;
//////////////////////////////////////////////////////
// read and handle errors & close, or return if busy
// note: if b == 0 error handling is not
// triggered, just return
// without closing
//////////////////////////////////////////////////////
rc = gw_read_gwbuff(dcb, &gw_buffer, b);
/* if (rc != 0) {
* Read all the data that is available into a chain of buffers dcb->state = DCB_STATE_POLLING;
*/ goto return_rc;
{ }
int len = -1;
int ret = -1;
GWBUF *queue = NULL;
GWBUF *gw_buffer = NULL;
int auth_val = -1;
//////////////////////////////////////////////////////
// read and handle errors & close, or return if busyA
// note: if b == 0 error handling is not triggered, just return
// without closing
//////////////////////////////////////////////////////
if ((ret = gw_read_gwbuff(dcb, &gw_buffer, b)) != 0) { // example with consume, assuming one buffer only ...
dcb->state = DCB_STATE_POLLING; queue = gw_buffer;
return ret; len = GWBUF_LENGTH(queue);
} //fprintf(stderr, "<<< Reading from Client %i bytes: [%s]\n", len, GWBUF_DATA(queue));
auth_val = gw_mysql_do_authentication(dcb, queue);
// Data handled withot the dcb->func.write
// so consume it now
// be sure to consume it all
queue = gwbuf_consume(queue, len);
// example with consume, assuming one buffer only ... if (auth_val == 0)
queue = gw_buffer; {
len = GWBUF_LENGTH(queue); SESSION *session = NULL;
protocol->state = MYSQL_AUTH_RECV;
//write to client mysql AUTH_OK packet, packet n. is 2
// start a new session, and connect to backends
session = session_alloc(dcb->service, dcb);
CHK_SESSION(session);
ss_dassert(session->state != SESSION_STATE_ALLOC);
protocol->state = MYSQL_IDLE;
mysql_send_ok(dcb, 2, 0, NULL);
}
else
{
protocol->state = MYSQL_AUTH_FAILED;
mysql_send_auth_error(
dcb,
2,
0,
"Authorization failed");
dcb->func.close(dcb);
}
}
break;
//fprintf(stderr, "<<< Reading from Client %i bytes: [%s]\n", len, GWBUF_DATA(queue)); case MYSQL_IDLE:
case MYSQL_WAITING_RESULT:
/*
* Read all the data that is available into a chain of buffers
*/
{
/* int len; */
GWBUF *queue = NULL;
GWBUF *gw_buffer = NULL;
uint8_t *ptr_buff = NULL;
int mysql_command = -1;
auth_val = gw_mysql_do_authentication(dcb, queue); session = dcb->session;
// Data handled withot the dcb->func.write // get the backend session, if available
// so consume it now if (session != NULL) {
// be sure to consume it all CHK_SESSION(session);
queue = gwbuf_consume(queue, len); router = session->service->router;
router_instance =
session->service->router_instance;
rsession = session->router_session;
}
if (auth_val == 0) //////////////////////////////////////////////////////
{ // read and handle errors & close, or return if busy
SESSION *session = NULL; //////////////////////////////////////////////////////
protocol->state = MYSQL_AUTH_RECV; rc = gw_read_gwbuff(dcb, &gw_buffer, b);
//write to client mysql AUTH_OK packet, packet n. is 2
// start a new session, and connect to backends
session = session_alloc(dcb->service, dcb);
protocol->state = MYSQL_IDLE;
session->data = (MYSQL_session *)dcb->data;
mysql_send_ok(dcb, 2, 0, NULL);
}
else
{
protocol->state = MYSQL_AUTH_FAILED;
mysql_send_auth_error(dcb, 2, 0, "Authorization failed"); if (rc != 0) {
dcb->state = DCB_STATE_POLLING;
goto return_rc;
}
/* Now, we are assuming in the first buffer there is
* the information form mysql command */
queue = gw_buffer;
/* len = GWBUF_LENGTH(queue); */
ptr_buff = GWBUF_DATA(queue);
dcb->func.close(dcb); /* get mysql commang at fourth byte */
} if (ptr_buff) {
} mysql_command = ptr_buff[4];
}
if (mysql_command == '\x03') {
/// this is a standard MySQL query !!!!
}
/**
* Routing Client input to Backend
*/
/* Do not route the query without session! */
if(rsession == NULL) {
if (mysql_command == '\x01') {
/* COM_QUIT handling */
/* fprintf(stderr, "COM_QUIT received with
* no connected backends from %i\n", dcb->fd); */
(dcb->func).close(dcb);
dcb->state = DCB_STATE_POLLING;
rc = 1;
goto return_rc;
} else {
/* Send a custom error as MySQL command reply */
mysql_send_custom_error(
dcb,
1,
0,
"Connection to backend lost");
protocol->state = MYSQL_IDLE;
dcb->state = DCB_STATE_POLLING;
rc = 1;
goto return_rc;
}
}
/* We can route the query */
/* COM_QUIT handling */
if (mysql_command == '\x01') {
/* fprintf(stderr, "COM_QUIT received from %i and
* passed to backed\n", dcb->fd); */
/* this will propagate COM_QUIT to backend(s) */
//fprintf(stderr, "<<< Routing the COM_QUIT ...\n");
router->routeQuery(router_instance,
rsession,
queue);
/* close client connection */
(dcb->func).close(dcb);
rc = 1;
goto return_rc;
}
break; /* MySQL Command Routing */
protocol->state = MYSQL_ROUTING;
case MYSQL_IDLE: /* writing in the backend buffer queue, via routeQuery */
case MYSQL_WAITING_RESULT: //fprintf(stderr, "<<< Routing the Query ...\n");
/* router->routeQuery(router_instance,
* Read all the data that is available into a chain of buffers rsession,
*/ queue);
{ protocol->state = MYSQL_WAITING_RESULT;
int len; }
GWBUF *queue = NULL; break;
GWBUF *gw_buffer = NULL;
uint8_t *ptr_buff = NULL;
int mysql_command = -1;
int ret = -1;
session = dcb->session; default:
// todo
// get the backend session, if available break;
if (session) {
router = session->service->router;
router_instance = session->service->router_instance;
rsession = session->router_session;
}
//////////////////////////////////////////////////////
// read and handle errors & close, or return if busy
//////////////////////////////////////////////////////
if ((ret = gw_read_gwbuff(dcb, &gw_buffer, b)) != 0) {
dcb->state = DCB_STATE_POLLING;
return ret;
}
/* Now, we are assuming in the first buffer there is the information form mysql command */
queue = gw_buffer;
len = GWBUF_LENGTH(queue);
ptr_buff = GWBUF_DATA(queue);
/* get mysql commang at fourth byte */
if (ptr_buff)
mysql_command = ptr_buff[4];
if (mysql_command == '\x03') {
/// this is a standard MySQL query !!!!
}
/**
* Routing Client input to Backend
*/
/* Do not route the query without session! */
if(!rsession) {
if (mysql_command == '\x01') {
/* COM_QUIT handling */
//fprintf(stderr, "COM_QUIT received with no connected backends from %i\n", dcb->fd);
(dcb->func).close(dcb);
return 1;
} else {
/* Send a custom error as MySQL command reply */
mysql_send_custom_error(dcb, 1, 0, "Connection to backend lost");
protocol->state = MYSQL_IDLE;
dcb->state = DCB_STATE_POLLING;
return 1;
}
}
/* We can route the query */
/* COM_QUIT handling */
if (mysql_command == '\x01') {
//fprintf(stderr, "COM_QUIT received from %i and passed to backed\n", dcb->fd);
/* this will propagate COM_QUIT to backend(s) */
//fprintf(stderr, "<<< Routing the COM_QUIT ...\n");
router->routeQuery(router_instance, rsession, queue);
/* close client connection */
(dcb->func).close(dcb);
return 1;
}
/* MySQL Command Routing */
protocol->state = MYSQL_ROUTING;
/* writing in the backend buffer queue, via routeQuery */
//fprintf(stderr, "<<< Routing the Query ...\n");
router->routeQuery(router_instance, rsession, queue);
protocol->state = MYSQL_WAITING_RESULT;
}
break;
default:
// todo
break;
} }
rc = 0;
dcb->state = DCB_STATE_POLLING; return_rc:
dcb->state = DCB_STATE_POLLING;
return 0; return rc;
} }
/////////////////////////////////////////////// ///////////////////////////////////////////////
@ -822,7 +845,7 @@ int gw_MySQLAccept(DCB *listener) {
int c_sock; int c_sock;
struct sockaddr_in local; struct sockaddr_in local;
socklen_t addrlen = sizeof(struct sockaddr_in); socklen_t addrlen = sizeof(struct sockaddr_in);
DCB *client; DCB *client_dcb;
MySQLProtocol *protocol; MySQLProtocol *protocol;
int sendbuf = GW_BACKEND_SO_SNDBUF; int sendbuf = GW_BACKEND_SO_SNDBUF;
socklen_t optlen = sizeof(sendbuf); socklen_t optlen = sizeof(sendbuf);
@ -831,11 +854,15 @@ int gw_MySQLAccept(DCB *listener) {
c_sock = accept(listener->fd, (struct sockaddr *) &local, &addrlen); c_sock = accept(listener->fd, (struct sockaddr *) &local, &addrlen);
if (c_sock == -1) { if (c_sock == -1) {
if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) { if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
/* We have processed all incoming connections. */ /* We have processed all incoming connections. */
break; break;
} else { } else {
fprintf(stderr, "Accept error for %i, Err: %i, %s\n", listener->fd, errno, strerror(errno)); fprintf(stderr,
"Accept error for %i, Err: %i, %s\n",
listener->fd,
errno,
strerror(errno));
// what else to do? // what else to do?
return 1; return 1;
} }
@ -849,26 +876,27 @@ int gw_MySQLAccept(DCB *listener) {
setsockopt(c_sock, SOL_SOCKET, SO_SNDBUF, &sendbuf, optlen); setsockopt(c_sock, SOL_SOCKET, SO_SNDBUF, &sendbuf, optlen);
setnonblocking(c_sock); setnonblocking(c_sock);
client = dcb_alloc(); client_dcb = dcb_alloc();
client->service = listener->session->service; client_dcb->service = listener->session->service;
client->fd = c_sock; client_dcb->fd = c_sock;
client->remote = strdup(inet_ntoa(local.sin_addr)); client_dcb->remote = strdup(inet_ntoa(local.sin_addr));
protocol = (MySQLProtocol *) calloc(1, sizeof(MySQLProtocol)); protocol = (MySQLProtocol *) calloc(1, sizeof(MySQLProtocol));
client->protocol = (void *)protocol; protocol->protocol_chk_top = CHK_NUM_PROTOCOL;
protocol->protocol_chk_tail = CHK_NUM_PROTOCOL;
client_dcb->protocol = (void *)protocol;
protocol->state = MYSQL_ALLOC; protocol->state = MYSQL_ALLOC;
protocol->descriptor = client; protocol->descriptor = client_dcb;
protocol->fd = c_sock; protocol->fd = c_sock;
// assign function poiters to "func" field // assign function poiters to "func" field
memcpy(&client->func, &MyObject, sizeof(GWPROTOCOL)); memcpy(&client_dcb->func, &MyObject, sizeof(GWPROTOCOL));
client->state = DCB_STATE_IDLE; client_dcb->state = DCB_STATE_IDLE;
//send handshake to the client //send handshake to the client_dcb
MySQLSendHandshake(client); MySQLSendHandshake(client_dcb);
// client protocol state change // client protocol state change
protocol->state = MYSQL_AUTH_SENT; protocol->state = MYSQL_AUTH_SENT;
@ -879,19 +907,19 @@ int gw_MySQLAccept(DCB *listener) {
* thread which wakes up sees correct state. * thread which wakes up sees correct state.
* *
*/ */
client->state = DCB_STATE_POLLING; client_dcb->state = DCB_STATE_POLLING;
if (poll_add_dcb(client) == -1) if (poll_add_dcb(client_dcb) == -1)
{ {
/** Return to previous state. */ /** Return to previous state. */
client->state = DCB_STATE_IDLE; client_dcb->state = DCB_STATE_IDLE;
skygw_log_write_flush( skygw_log_write_flush(
LOGFILE_ERROR, LOGFILE_ERROR,
"%lu [gw_MySQLAccept] Failed to add dcb %p for fd " "%lu [gw_MySQLAccept] Failed to add dcb %p for fd "
"%d to epoll set.", "%d to epoll set.",
pthread_self(), pthread_self(),
client, client_dcb,
client->fd); client_dcb->fd);
return 1; return 1;
} }
else else
@ -901,8 +929,8 @@ int gw_MySQLAccept(DCB *listener) {
"%lu [gw_MySQLAccept] Added dcb %p for fd " "%lu [gw_MySQLAccept] Added dcb %p for fd "
"%d to epoll set.", "%d to epoll set.",
pthread_self(), pthread_self(),
client, client_dcb,
client->fd); client_dcb->fd);
} }
} /**< while 1 */ } /**< while 1 */
return 0; return 0;