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

Simplify adding and removing DCBs from polling, improve error handling. Remove dcb_set_state functions as not adding value.

Browse Source
This commit is contained in:
counterpoint
2015-07-06 14:50:01 +01:00
parent a72f462e2d
commit 18a95eeb71
3 changed files with 105 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
server/core/dcb.c
View File

@ -84,10 +84,6 @@ static SPINLOCK dcbspin = SPINLOCK_INIT;
static SPINLOCK zombiespin = SPINLOCK_INIT;
static void dcb_final_free(DCB *dcb);
static bool dcb_set_state_nomutex(
DCB* dcb,
const dcb_state_t new_state,
dcb_state_t* old_state);
static void dcb_call_callback(DCB *dcb, DCB_REASON reason);
static DCB* dcb_get_next (DCB* dcb);
static int dcb_null_write(DCB *dcb, GWBUF *buf);
@ -291,8 +287,7 @@ dcb_add_to_zombieslist(DCB *dcb)
* Set state which indicates that it has been added to zombies
* list.
*/
succp = dcb_set_state(dcb, DCB_STATE_ZOMBIE, &prev_state);
ss_info_dassert(succp, "Failed to set DCB_STATE_ZOMBIE");
dcb->state = DCB_STATE_ZOMBIE;
spinlock_release(&zombiespin);
}
@ -604,7 +599,7 @@ bool succp = false;
&tls_log_info.li_sesid,
&tls_log_info.li_enabled_logs)));
succp = dcb_set_state(dcb, DCB_STATE_DISCONNECTED, NULL);
dcb->state = DCB_STATE_DISCONNECTED;
ss_dassert(succp);
dcb_next = dcb->memdata.next;
dcb_final_free(dcb);
@ -644,7 +639,7 @@ int rc;
if ((funcs = (GWPROTOCOL *)load_module(protocol,
MODULE_PROTOCOL)) == NULL)
{
dcb_set_state(dcb, DCB_STATE_DISCONNECTED, NULL);
dcb->state = DCB_STATE_DISCONNECTED;
dcb_final_free(dcb);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
@ -682,7 +677,7 @@ int rc;
dcb,
session->client,
session->client->fd)));
dcb_set_state(dcb, DCB_STATE_DISCONNECTED, NULL);
dcb->state = DCB_STATE_DISCONNECTED;
dcb_final_free(dcb);
return NULL;
} else {
@ -725,7 +720,7 @@ int rc;
rc = poll_add_dcb(dcb);
if (rc) {
dcb_set_state(dcb, DCB_STATE_DISCONNECTED, NULL);
dcb->state = DCB_STATE_DISCONNECTED;
dcb_final_free(dcb);
return NULL;
}
@ -1920,7 +1915,7 @@ dcb_drain_writeq_SSL(DCB *dcb)
}
/**
* Removes dcb from poll set, and adds it to zombies list. As a consequense,
* Removes dcb from poll set, and adds it to zombies list. As a consequence,
* dcb first moves to DCB_STATE_NOPOLLING, and then to DCB_STATE_ZOMBIE state.
* At the end of the function state may not be DCB_STATE_ZOMBIE because once
* dcb_initlock is released parallel threads may change the state.
@ -1947,7 +1942,6 @@ dcb_close(DCB *dcb)
*/
if (dcb->state == DCB_STATE_ALLOC)
{
dcb_set_state(dcb, DCB_STATE_DISCONNECTED, NULL);
dcb_final_free(dcb);
return;
}
@ -2341,188 +2335,6 @@ void dcb_hashtable_stats(
dcb_printf(dcb, "\tLongest chain length: %d\n", longest);
}
bool dcb_set_state(
DCB* dcb,
const dcb_state_t new_state,
dcb_state_t* old_state)
{
bool succp;
dcb_state_t state ;
CHK_DCB(dcb);
spinlock_acquire(&dcb->dcb_initlock);
succp = dcb_set_state_nomutex(dcb, new_state, &state);
spinlock_release(&dcb->dcb_initlock);
if (old_state != NULL) {
*old_state = state;
}
return succp;
}
void dcb_revert_state(
DCB* dcb,
const dcb_state_t new_state,
dcb_state_t old_state)
{
CHK_DCB(dcb);
spinlock_acquire(&dcb->dcb_initlock);
if ((DCB_STATE_POLLING == new_state || DCB_STATE_LISTENING == new_state) && (DCB_STATE_ALLOC == old_state || DCB_STATE_NOPOLLING == old_state))
{
dcb->state = old_state;
spinlock_release(&dcb->dcb_initlock);
return;
}
else assert(false);
}
static bool dcb_set_state_nomutex(
DCB* dcb,
const dcb_state_t new_state,
dcb_state_t* old_state)
{
bool succp = false;
dcb_state_t state = DCB_STATE_UNDEFINED;
CHK_DCB(dcb);
state = dcb->state;
if (old_state != NULL) {
*old_state = state;
}
switch (state) {
case DCB_STATE_UNDEFINED:
dcb->state = new_state;
succp = true;
break;
case DCB_STATE_ALLOC:
switch (new_state) {
/** fall through, for client requests */
case DCB_STATE_POLLING:
/** fall through, for connect listeners */
case DCB_STATE_LISTENING:
/** for failed connections */
case DCB_STATE_DISCONNECTED:
dcb->state = new_state;
succp = true;
break;
default:
ss_dassert(old_state != NULL);
break;
}
break;
case DCB_STATE_POLLING:
switch(new_state) {
case DCB_STATE_NOPOLLING:
dcb->state = new_state;
succp = true;
break;
default:
ss_dassert(old_state != NULL);
break;
}
break;
case DCB_STATE_LISTENING:
switch(new_state) {
case DCB_STATE_NOPOLLING:
dcb->state = new_state;
succp = true;
break;
default:
ss_dassert(old_state != NULL);
break;
}
break;
case DCB_STATE_NOPOLLING:
switch (new_state) {
/** Stopped services which are restarting will go from
* DCB_STATE_NOPOLLING to DCB_STATE_LISTENING.*/
case DCB_STATE_LISTENING:
case DCB_STATE_ZOMBIE: /*< fall through */
dcb->state = new_state;
case DCB_STATE_POLLING: /*< ok to try but state can't change */
succp = true;
break;
default:
ss_dassert(old_state != NULL);
break;
}
break;
case DCB_STATE_ZOMBIE:
switch (new_state) {
case DCB_STATE_DISCONNECTED: /*< fall through */
dcb->state = new_state;
case DCB_STATE_POLLING: /*< ok to try but state can't change */
succp = true;
break;
default:
ss_dassert(old_state != NULL);
break;
}
break;
case DCB_STATE_DISCONNECTED:
switch (new_state) {
case DCB_STATE_FREED:
dcb->state = new_state;
succp = true;
break;
default:
ss_dassert(old_state != NULL);
break;
}
break;
case DCB_STATE_FREED:
ss_dassert(old_state != NULL);
break;
default:
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unknown dcb state %s for "
"dcb %p",
STRDCBSTATE(dcb->state),
dcb)));
ss_dassert(false);
break;
} /*< switch (dcb->state) */
if (succp) {
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"%lu [dcb_set_state_nomutex] dcb %p fd %d %s -> %s",
pthread_self(),
dcb,
dcb->fd,
STRDCBSTATE(state),
STRDCBSTATE(dcb->state))));
}
else
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [dcb_set_state_nomutex] Failed "
"to change state of DCB %p. "
"Old state %s > new state %s.",
pthread_self(),
dcb,
(old_state == NULL ? "NULL" : STRDCBSTATE(*old_state)),
STRDCBSTATE(new_state))));
}
return succp;
}
/**
* Write data to a socket through an SSL structure. The SSL structure is linked to a DCB's socket
* and all communication is encrypted and done via the SSL structure.

164
server/core/poll.c
View File

@ -189,7 +189,7 @@ static void poll_loadav(void *);
/**
* Function to analyse error return from epoll_ctl
*/
static int poll_resolve_error(int, bool);
static int poll_resolve_error(DCB *, int, bool);
/**
* Initialise the polling system we are using for the gateway.
@ -252,7 +252,7 @@ int
poll_add_dcb(DCB *dcb)
{
int rc = -1;
dcb_state_t old_state = DCB_STATE_UNDEFINED;
dcb_state_t old_state = dcb->state;
dcb_state_t new_state;
struct epoll_event ev;
@ -268,47 +268,71 @@ poll_add_dcb(DCB *dcb)
/*<
* Choose new state according to the role of dcb.
*/
spinlock_acquire(&dcb->dcb_initlock);
if (dcb->dcb_role == DCB_ROLE_REQUEST_HANDLER) {
new_state = DCB_STATE_POLLING;
} else {
ss_dassert(dcb->dcb_role == DCB_ROLE_SERVICE_LISTENER);
new_state = DCB_STATE_LISTENING;
}
/*
* Check DCB current state seems sensible
*/
if (DCB_STATE_DISCONNECTED == dcb->state
|| DCB_STATE_ZOMBIE == dcb->state
|| DCB_STATE_UNDEFINED == dcb->state)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"%lu [poll_add_dcb] Error : existing state of dcb %p "
"is %s, but this should be impossible, crashing.",
pthread_self(),
dcb,
STRDCBSTATE(dcb->state))));
assert(false);
}
if (DCB_STATE_POLLING == dcb->state
|| DCB_STATE_LISTENING == dcb->state)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"%lu [poll_add_dcb] Error : existing state of dcb %p "
"is %s, but this is probably an error, not crashing.",
pthread_self(),
dcb,
STRDCBSTATE(dcb->state))));
}
/*<
* If dcb is in unexpected state, state change fails indicating that dcb
* is not polling anymore.
*/
if (dcb_set_state(dcb, new_state, &old_state)) {
rc = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, dcb->fd, &ev);
if (rc) rc = poll_resolve_error(errno, true);
if (0 == rc)
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [poll_add_dcb] Added dcb %p in state %s to "
"poll set.",
pthread_self(),
dcb,
STRDCBSTATE(dcb->state))));
}
else dcb_revert_state(dcb, new_state, old_state);
} else {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to set new state for dcb %p "
"in state %s. Adding to poll set failed.",
dcb,
STRDCBSTATE(dcb->state))));
dcb->state = new_state;
spinlock_release(&dcb->dcb_initlock);
/*
* The only possible failure that will not cause a crash is
* running out of system resources.
*/
rc = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, dcb->fd, &ev);
if (rc)
{
rc = poll_resolve_error(dcb, errno, true);
}
if (0 == rc)
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [poll_add_dcb] Added dcb %p in state %s to poll set.",
pthread_self(),
dcb,
STRDCBSTATE(dcb->state))));
}
else dcb->state = old_state;
return rc;
}
/**
* Remove a descriptor from the set of descriptors within the
* polling environment.
* The state change command may fail because concurrent threads may call
* dcb_set_state simultaneously and the conflict is prevented in dcb_set_state.
*
* @param dcb The descriptor to remove
* @return -1 on error or 0 on success
@ -316,52 +340,53 @@ poll_add_dcb(DCB *dcb)
int
poll_remove_dcb(DCB *dcb)
{
struct epoll_event ev;
int rc = -1;
dcb_state_t old_state = DCB_STATE_UNDEFINED;
dcb_state_t new_state = DCB_STATE_NOPOLLING;
CHK_DCB(dcb);
spinlock_acquire(&dcb->dcb_initlock);
/*< It is possible that dcb has already been removed from the set */
if (dcb->state != DCB_STATE_POLLING &&
dcb->state != DCB_STATE_LISTENING)
{
if (dcb->state == DCB_STATE_NOPOLLING ||
dcb->state == DCB_STATE_ZOMBIE)
{
rc = 0;
}
goto return_rc;
if (dcb->state == DCB_STATE_NOPOLLING ||
dcb->state == DCB_STATE_ZOMBIE)
{
spinlock_release(&dcb->dcb_initlock);
return 0;
}
if (DCB_STATE_POLLING != dcb->state
&& DCB_STATE_LISTENING != dcb->state)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"%lu [poll_remove_dcb] Error : existing state of dcb %p "
"is %s, but this is probably an error, not crashing.",
pthread_self(),
dcb,
STRDCBSTATE(dcb->state))));
}
/*<
* Set state to NOPOLLING and remove dcb from poll set.
*/
if (dcb_set_state(dcb, new_state, &old_state))
{
/**
* Only positive fds can be removed from epoll set.
*/
if (dcb->fd > 0)
{
rc = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, dcb->fd, &ev);
if (rc) rc = poll_resolve_error(errno, false);
}
}
/*<
* This call was redundant, but the end result is correct.
dcb->state = DCB_STATE_NOPOLLING;
/**
* Only positive fds can be removed from epoll set.
* But this test was removed by Martin as it is hard to
* see that there should ever be a situation where
* fd isn't positive and the DCB is also in the poll list.
*/
/* if (dcb->fd > 0) { */
spinlock_release(&dcb->dcb_initlock);
rc = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, dcb->fd, &ev);
/**
* The poll_resolve_error function will always
* return 0 or crash. So if it returns non-zero result,
* things have gone wrong and we crash.
*/
else if (old_state == new_state)
{
rc = 0;
goto return_rc;
}
if (rc) rc = poll_resolve_error(dcb, errno, false);
if (rc) assert(false);
/*< Set bit for each maxscale thread */
bitmask_copy(&dcb->memdata.bitmask, poll_bitmask());
rc = 0;
return_rc:
bitmask_copy(&dcb->memdata.bitmask, poll_bitmask());
return rc;
/* } */
}
/**
@ -377,7 +402,7 @@ return_rc:
* @return -1 on error or 0 for possibly revised return code
*/
static int
poll_resolve_error(int errornum, bool adding)
poll_resolve_error(DCB *dcb, int errornum, bool adding)
{
if (adding)
{
@ -385,7 +410,10 @@ poll_resolve_error(int errornum, bool adding)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : epoll_ctl could not add, already exists.")));
"%lu [poll_resolve_error] Error : epoll_ctl could not add, "
"already exists for DCB %p.",
pthread_self(),
dcb)));
// Assume another thread added and no serious harm done
return 0;
}
@ -393,9 +421,12 @@ poll_resolve_error(int errornum, bool adding)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"The limit imposed by /proc/sys/fs/epoll/max_user_watches was "
"%lu [poll_resolve_error] The limit imposed by "
"/proc/sys/fs/epoll/max_user_watches was "
"encountered while trying to register (EPOLL_CTL_ADD) a new "
"file descriptor on an epoll instance.")));
"file descriptor on an epoll instance for dcb %p.",
pthread_self(),
dcb)));
/* Failure - assume handled by callers */
return -1;
}
@ -407,7 +438,10 @@ poll_resolve_error(int errornum, bool adding)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : epoll_ctl could not remove, not found.")));
"%lu [poll_resolve_error] Error : epoll_ctl could not remove, "
"not found, for dcb %p.",
pthread_self(),
dcb)));
// Assume another thread removed and no serious harm done
return 0;
}

2
server/include/dcb.h
View File

@ -338,8 +338,6 @@ int dcb_remove_callback(DCB *, DCB_REASON, int (*)(struct dcb *, DCB_REASON, vo
int dcb_isvalid(DCB *); /* Check the DCB is in the linked list */
int dcb_count_by_usage(DCB_USAGE); /* Return counts of DCBs */
bool dcb_set_state(DCB *dcb, dcb_state_t new_state, dcb_state_t *old_state);
void dcb_revert_state(DCB *dcb, const dcb_state_t new_state, dcb_state_t old_state);
void dcb_call_foreach (struct server* server, DCB_REASON reason);
size_t dcb_get_session_id(DCB* dcb);
bool dcb_get_ses_log_info(DCB* dcb, size_t* sesid, int* enabled_logs);