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

Merge branch 'SESvars' of https://github.com/skysql/MaxScale into SESvars

Browse Source 
Conflicts:
	server/core/dcb.c
	server/core/poll.c
	server/modules/include/mysql_client_server_protocol.h
	server/modules/routing/readwritesplit/readwritesplit.c
This commit is contained in:
VilhoRaatikka
2014-03-18 10:28:06 +02:00
parent 1926181119 67d9b3afb9
commit c356b396a6
19 changed files with 1667 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
server/core/buffer.c
View File

@ -81,6 +81,7 @@ SHARED_BUF *sbuf;
sbuf->refcount = 1;
rval->sbuf = sbuf;
rval->next = NULL;
rval->gwbuf_type = GWBUF_TYPE_UNDEFINED;
rval->command = 0;
CHK_GWBUF(rval);
return rval;
@ -127,11 +128,93 @@ GWBUF *rval;
rval->sbuf = buf->sbuf;
rval->start = buf->start;
rval->end = buf->end;
rval->gwbuf_type = buf->gwbuf_type;
rval->next = NULL;
rval->command = buf->command;
CHK_GWBUF(rval);
return rval;
}
GWBUF *gwbuf_clone_portion(
GWBUF *buf,
size_t start_offset,
size_t length)
{
GWBUF* clonebuf;
CHK_GWBUF(buf);
ss_dassert(start_offset+length <= GWBUF_LENGTH(buf));
if ((clonebuf = (GWBUF *)malloc(sizeof(GWBUF))) == NULL)
{
return NULL;
}
atomic_add(&buf->sbuf->refcount, 1);
clonebuf->sbuf = buf->sbuf;
clonebuf->start = (void *)((char*)buf->start)+start_offset;
clonebuf->end = (void *)((char *)clonebuf->start)+length;
clonebuf->gwbuf_type = buf->gwbuf_type; /*< clone the type for now */
clonebuf->next = NULL;
CHK_GWBUF(clonebuf);
return clonebuf;
}
/**
* Returns pointer to GWBUF of a requested type.
* As of 10.3.14 only MySQL to plain text conversion is supported.
* Return NULL if conversion between types is not supported or due lacking
* type information.
*/
GWBUF *gwbuf_clone_transform(
GWBUF * head,
gwbuf_type_t targettype)
{
gwbuf_type_t src_type;
GWBUF* clonebuf;
CHK_GWBUF(head);
src_type = head->gwbuf_type;
if (targettype == GWBUF_TYPE_UNDEFINED ||
src_type == GWBUF_TYPE_UNDEFINED ||
src_type == GWBUF_TYPE_PLAINSQL ||
targettype == src_type)
{
clonebuf = NULL;
goto return_clonebuf;
}
switch (src_type)
{
case GWBUF_TYPE_MYSQL:
if (targettype == GWBUF_TYPE_PLAINSQL)
{
/** Crete reference to string part of buffer */
clonebuf = gwbuf_clone_portion(
head,
5,
GWBUF_LENGTH(head)-5);
ss_dassert(clonebuf != NULL);
/** Overwrite the type with new format */
clonebuf->gwbuf_type = targettype;
}
else
{
clonebuf = NULL;
}
break;
default:
clonebuf = NULL;
break;
} /*< switch (src_type) */
return_clonebuf:
return clonebuf;
}
/**
* Append a buffer onto a linked list of buffer structures.
*
@ -150,6 +233,7 @@ GWBUF *ptr = head;
if (!head)
return tail;
CHK_GWBUF(head);
CHK_GWBUF(tail);
while (ptr->next)
{
ptr = ptr->next;
@ -178,9 +262,10 @@ GWBUF *
gwbuf_consume(GWBUF *head, unsigned int length)
{
GWBUF *rval = head;
CHK_GWBUF(head);
GWBUF_CONSUME(head, length);
CHK_GWBUF(head);
if (GWBUF_EMPTY(head))
{
rval = head->next;
@ -207,3 +292,28 @@ int rval = 0;
}
return rval;
}
bool gwbuf_set_type(
GWBUF* buf,
gwbuf_type_t type)
{
bool succp;
CHK_GWBUF(buf);
switch (type) {
case GWBUF_TYPE_MYSQL:
case GWBUF_TYPE_PLAINSQL:
case GWBUF_TYPE_UNDEFINED:
buf->gwbuf_type = type;
succp = true;
break;
default:
succp = false;
break;
}
ss_dassert(succp);
return succp;
}

105
server/core/dcb.c
View File

@ -303,12 +303,8 @@ dcb_final_free(DCB *dcb)
if (dcb->remote)
free(dcb->remote);
bitmask_free(&dcb->memdata.bitmask);
#if 1
simple_mutex_done(&dcb->dcb_read_lock);
simple_mutex_done(&dcb->dcb_write_lock);
#endif
free(dcb);
}
@ -527,6 +523,8 @@ int rc;
* Successfully connected to backend. Assign file descriptor to dcb
*/
dcb->fd = fd;
/** Copy status field to DCB */
dcb->dcb_server_status = server->status;
/*<
* backend_dcb is connected to backend server, and once backend_dcb
@ -690,10 +688,18 @@ dcb_write(DCB *dcb, GWBUF *queue)
dcb->state != DCB_STATE_LISTENING &&
dcb->state != DCB_STATE_NOPOLLING))
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [dcb_write] Write aborted to dcb %p because "
"it is in state %s",
pthread_self(),
dcb->stats.n_buffered,
dcb,
STRDCBSTATE(dcb->state),
dcb->fd)));
return 0;
}
spinlock_acquire(&dcb->writeqlock);
if (dcb->writeq != NULL)
@ -750,7 +756,11 @@ dcb_write(DCB *dcb, GWBUF *queue)
#endif /* SS_DEBUG */
len = GWBUF_LENGTH(queue);
GW_NOINTR_CALL(
w = gw_write(dcb->fd, GWBUF_DATA(queue), len);
w = gw_write(
#if defined(SS_DEBUG)
dcb,
#endif
dcb->fd, GWBUF_DATA(queue), len);
dcb->stats.n_writes++;
);
@ -759,6 +769,8 @@ dcb_write(DCB *dcb, GWBUF *queue)
saved_errno = errno;
errno = 0;
if (LOG_IS_ENABLED(LOGFILE_DEBUG))
{
if (saved_errno == EPIPE) {
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
@ -771,7 +783,12 @@ dcb_write(DCB *dcb, GWBUF *queue)
dcb->fd,
saved_errno,
strerror(saved_errno))));
} else if (saved_errno != EAGAIN &&
}
}
if (LOG_IS_ENABLED(LOGFILE_ERROR))
{
if (saved_errno != EPIPE &&
saved_errno != EAGAIN &&
saved_errno != EWOULDBLOCK)
{
LOGIF(LE, (skygw_log_write_flush(
@ -785,6 +802,7 @@ dcb_write(DCB *dcb, GWBUF *queue)
saved_errno,
strerror(saved_errno))));
}
}
break;
}
/*
@ -799,9 +817,9 @@ dcb_write(DCB *dcb, GWBUF *queue)
pthread_self(),
w,
dcb,
STRDCBSTATE(dcb->state),
STRDCBSTATE(dcb->state),
dcb->fd)));
}
} /*< while (queue != NULL) */
/*<
* What wasn't successfully written is stored to write queue
* for suspended write.
@ -819,7 +837,6 @@ dcb_write(DCB *dcb, GWBUF *queue)
saved_errno != EAGAIN &&
saved_errno != EWOULDBLOCK)
{
queue = gwbuf_consume(queue, gwbuf_length(queue));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Writing to %s socket failed due %d, %s.",
@ -862,9 +879,13 @@ int saved_errno = 0;
while (dcb->writeq != NULL)
{
len = GWBUF_LENGTH(dcb->writeq);
GW_NOINTR_CALL(w = gw_write(dcb->fd,
GWBUF_DATA(dcb->writeq),
len););
GW_NOINTR_CALL(w = gw_write(
#if defined(SS_DEBUG)
dcb,
#endif
dcb->fd,
GWBUF_DATA(dcb->writeq),
len););
saved_errno = errno;
errno = 0;
@ -1319,12 +1340,15 @@ static bool dcb_set_state_nomutex(
}
int gw_write(
#if defined(SS_DEBUG)
DCB* dcb,
#endif
int fd,
const void* buf,
size_t nbytes)
{
int w;
#if defined(SS_DEBUG)
#if defined(SS_DEBUG)
if (dcb_fake_write_errno[fd] != 0) {
ss_dassert(dcb_fake_write_ev[fd] != 0);
w = write(fd, buf, nbytes/2); /*< leave peer to read missing bytes */
@ -1339,6 +1363,57 @@ int gw_write(
#else
w = write(fd, buf, nbytes);
#endif /* SS_DEBUG && SS_TEST */
#if defined(SS_DEBUG_MYSQL)
{
size_t len;
uint8_t* packet = (uint8_t *)buf;
char* str;
/** Print only MySQL packets */
if (w > 5)
{
str = (char *)&packet[5];
len = packet[0];
len += 256*packet[1];
len += 256*256*packet[2];
if (strncmp(str, "insert", 6) == 0 ||
strncmp(str, "create", 6) == 0 ||
strncmp(str, "drop", 4) == 0)
{
ss_dassert((dcb->dcb_server_status & (SERVER_RUNNING|SERVER_MASTER|SERVER_SLAVE))==(SERVER_RUNNING|SERVER_MASTER));
}
if (strncmp(str, "set autocommit", 14) == 0 && nbytes > 17)
{
char* s = (char *)calloc(1, nbytes+1);
if (nbytes-5 > len)
{
size_t len2 = packet[4+len];
len2 += 256*packet[4+len+1];
len2 += 256*256*packet[4+len+2];
char* str2 = (char *)&packet[4+len+5];
snprintf(s, 5+len+len2, "long %s %s", (char *)str, (char *)str2);
}
else
{
snprintf(s, len, "%s", (char *)str);
}
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"%lu [gw_write] Wrote %d bytes : %s ",
pthread_self(),
w,
s)));
free(s);
}
}
}
#endif
return w;
}

10
server/core/poll.c
View File

@ -363,7 +363,6 @@ poll_waitevents(void *arg)
eno = gw_getsockerrno(dcb->fd);
if (eno == 0) {
#if 1
simple_mutex_lock(
&dcb->dcb_write_lock,
true);
@ -371,16 +370,13 @@ poll_waitevents(void *arg)
!dcb->dcb_write_active,
"Write already active");
dcb->dcb_write_active = TRUE;
#endif
atomic_add(
&pollStats.n_write,
1);
dcb->func.write_ready(dcb);
#if 1
dcb->dcb_write_active = FALSE;
simple_mutex_unlock(
&dcb->dcb_write_lock);
#endif
} else {
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
@ -396,13 +392,12 @@ poll_waitevents(void *arg)
}
if (ev & EPOLLIN)
{
#if 1
simple_mutex_lock(&dcb->dcb_read_lock,
true);
ss_info_dassert(!dcb->dcb_read_active,
"Read already active");
dcb->dcb_read_active = TRUE;
#endif
if (dcb->state == DCB_STATE_LISTENING)
{
LOGIF(LD, (skygw_log_write(
@ -427,11 +422,9 @@ poll_waitevents(void *arg)
atomic_add(&pollStats.n_read, 1);
dcb->func.read(dcb);
}
#if 1
dcb->dcb_read_active = FALSE;
simple_mutex_unlock(
&dcb->dcb_read_lock);
#endif
}
if (ev & EPOLLERR)
{
@ -462,6 +455,7 @@ poll_waitevents(void *arg)
atomic_add(&pollStats.n_error, 1);
dcb->func.error(dcb);
}
if (ev & EPOLLHUP)
{
int eno = 0;