mempool/ckpool
mempool/ckpool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Merge branch 'master' of https://bitbucket.org/ckolivas/ckpool

Browse Source
This commit is contained in:
kanoi 2014-06-27 15:06:36 +10:00
commit 2c776ed38e
5 changed files with 29 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
src/ckpool.c
View File

@ -227,11 +227,9 @@ void empty_buffer(connsock_t *cs)
* of the buffer for use on the next receive. */
int read_socket_line(connsock_t *cs, int timeout)
{
tv_t tv_timeout = {timeout, 0};
char *eom = NULL;
size_t buflen;
int ret = -1;
fd_set rd;
if (unlikely(cs->fd < 0))
goto out;
@ -250,11 +248,7 @@ int read_socket_line(connsock_t *cs, int timeout)
while (42) {
char readbuf[PAGESIZE] = {};
FD_ZERO(&rd);
FD_SET(cs->fd, &rd);
if (eom)
tv_timeout.tv_sec = tv_timeout.tv_usec = 0;
ret = select(cs->fd + 1, &rd, NULL, NULL, &tv_timeout);
ret = wait_read_select(cs->fd, eom ? 0 : timeout);
if (eom && !ret)
break;
if (ret < 1) {
@ -697,20 +691,17 @@ static void sighandler(int sig)
{
ckpool_t *ckp = global_ckp;
LOGWARNING("Parent process %s received signal %d, shutting down",
ckp->name, sig);
pthread_cancel(ckp->pth_watchdog);
join_pthread(ckp->pth_watchdog);
if (sig != 9) {
__shutdown_children(ckp, SIGTERM);
/* Wait a second, then send SIGKILL */
sleep(1);
__shutdown_children(ckp, SIGKILL);
pthread_cancel(ckp->pth_listener);
exit(0);
} else {
__shutdown_children(ckp, SIGKILL);
exit(1);
}
__shutdown_children(ckp, SIGTERM);
/* Wait a second, then send SIGKILL */
sleep(1);
__shutdown_children(ckp, SIGKILL);
pthread_cancel(ckp->pth_listener);
exit(0);
}
static void json_get_string(char **store, json_t *val, const char *res)

13
src/connector.c
View File

@ -260,7 +260,9 @@ retry:
cksleep_ms(100);
goto retry;
}
ret = poll(fds, nfds, 1000);
do {
ret = poll(fds, nfds, 1000);
} while (unlikely(ret < 0 && interrupted()));
if (ret < 0) {
LOGERR("Failed to poll in receiver");
goto out;
@ -308,7 +310,6 @@ void *sender(void *arg)
while (42) {
sender_send_t *sender_send;
client_instance_t *client;
tv_t timeout_tv = {0, 0};
bool only_send = false;
int ret, fd, ofs = 0;
@ -346,11 +347,7 @@ void *sender(void *arg)
* ready to receive data from us, put the send back on the
* list. */
if (!only_send) {
fd_set writefds;
FD_ZERO(&writefds);
FD_SET(fd, &writefds);
ret = select(fd + 1, NULL, &writefds, NULL, &timeout_tv);
ret = wait_write_select(fd, 0);
if (ret < 1) {
LOGDEBUG("Client %d not ready for writes", client->id);
@ -480,7 +477,7 @@ retry:
close(sockd);
sockd = accept(us->sockd, NULL, NULL);
if (sockd < 0) {
LOGERR("Failed to accept on connector socket, retrying in 5s");
LOGEMERG("Failed to accept on connector socket, exiting");
ret = 1;
goto out;
}

8
src/generator.c
View File

@ -220,9 +220,7 @@ retry:
sockd = accept(us->sockd, NULL, NULL);
if (sockd < 0) {
if (interrupted())
goto retry;
LOGERR("Failed to accept on generator socket");
LOGEMERG("Failed to accept on generator socket");
ret = 1;
goto out;
}
@ -1227,9 +1225,7 @@ reconnect:
retry:
sockd = accept(us->sockd, NULL, NULL);
if (sockd < 0) {
if (interrupted())
goto retry;
LOGERR("Failed to accept on proxy socket");
LOGEMERG("Failed to accept on proxy socket");
ret = 1;
goto out;
}

28
src/libckpool.c
View File

@ -411,19 +411,15 @@ int connect_socket(char *url, char *port)
* we can connect to quickly. */
noblock_socket(sockd);
if (connect(sockd, p->ai_addr, p->ai_addrlen) == -1) {
struct timeval tv_timeout = {5, 0};
int selret;
fd_set rw;
if (!sock_connecting()) {
close(sockd);
LOGDEBUG("Failed sock connect");
continue;
}
FD_ZERO(&rw);
FD_SET(sockd, &rw);
selret = select(sockd + 1, NULL, &rw, NULL, &tv_timeout);
if (selret > 0 && FD_ISSET(sockd, &rw)) {
selret = wait_write_select(sockd, 5);
if (selret > 0) {
socklen_t len;
int err, n;
@ -457,13 +453,9 @@ out:
int write_socket(int fd, const void *buf, size_t nbyte)
{
tv_t tv_timeout = {1, 0};
fd_set writefds;
int ret;
FD_ZERO(&writefds);
FD_SET(fd, &writefds);
ret = select(fd + 1, NULL, &writefds, NULL, &tv_timeout);
ret = wait_write_select(fd, 5);
if (ret < 1) {
if (!ret)
LOGNOTICE("Select timed out in write_socket");
@ -487,12 +479,8 @@ void empty_socket(int fd)
do {
char buf[PAGESIZE];
tv_t timeout = {0, 0};
fd_set rd;
FD_ZERO(&rd);
FD_SET(fd, &rd);
ret = select(fd + 1, &rd, NULL, NULL, &timeout);
ret = wait_read_select(fd, 0);
if (ret > 0) {
ret = recv(fd, buf, PAGESIZE - 1, 0);
buf[ret] = 0;
@ -625,7 +613,9 @@ int wait_read_select(int sockd, int timeout)
FD_ZERO(&readfs);
FD_SET(sockd, &readfs);
ret = select(sockd + 1, &readfs, NULL, NULL, &tv_timeout);
do {
ret = select(sockd + 1, &readfs, NULL, NULL, &tv_timeout);
} while (unlikely(ret < 0 && interrupted()));
return ret;
}
@ -700,7 +690,9 @@ int wait_write_select(int sockd, int timeout)
FD_ZERO(&writefds);
FD_SET(sockd, &writefds);
ret = select(sockd + 1, NULL, &writefds, NULL, &tv_timeout);
do {
ret = select(sockd + 1, NULL, &writefds, NULL, &tv_timeout);
} while (unlikely(ret < 0 && interrupted()));
return ret;
}

8
src/stratifier.c
View File

@ -949,11 +949,9 @@ retry:
sockd = accept(us->sockd, NULL, NULL);
if (sockd < 0) {
if (interrupted())
goto retry;
LOGERR("Failed to accept on stratifier socket, retrying in 5s");
sleep(5);
goto retry;
LOGERR("Failed to accept on stratifier socket, exiting");
ret = 1;
goto out;
}
dealloc(buf);