--- libaitio/src/sock.c 2013/11/22 09:24:58 1.4.4.9 +++ libaitio/src/sock.c 2013/12/12 15:20:23 1.11.2.1 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: sock.c,v 1.4.4.9 2013/11/22 09:24:58 misho Exp $ +* $Id: sock.c,v 1.11.2.1 2013/12/12 15:20:23 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -51,23 +51,32 @@ io_closeClient(sched_task_t *task) { sock_cli_t *cli = (sock_cli_t*) TASK_ARG(task); sock_t *s = (sock_t*) cli->cli_parent; - int sock = (int) TASK_DATLEN(task); + int stat; pthread_mutex_lock(&s->sock_mtx); TAILQ_REMOVE(&s->sock_cli, cli, cli_node); pthread_mutex_unlock(&s->sock_mtx); - schedCancelby(s->sock_root, taskTIMER, CRITERIA_DATLEN, - (void*) cli->cli_fd, NULL); schedCancelby(s->sock_root, taskMAX, CRITERIA_ARG, cli, NULL); + if (*cli->cli_name) + ioFreePTY(cli->cli_pty, cli->cli_name); + if (s->sock_type == SOCK_STREAM) { - shutdown(sock, SHUT_RDWR); - close(sock); + shutdown(cli->cli_fd, SHUT_RDWR); + close(cli->cli_fd); } AIT_FREE_VAL(&cli->cli_buf[1]); AIT_FREE_VAL(&cli->cli_buf[0]); + if (cli->cli_pid > 0) { + kill(cli->cli_pid, SIGKILL); + while (waitpid(cli->cli_pid, &stat, WNOHANG) > 0) { + usleep(1000); + kill(cli->cli_pid, SIGKILL); + } + } + e_free(cli); taskExit(task, NULL); } @@ -126,18 +135,37 @@ end: static void * io_txNet(sched_task_t *task) { - int wlen; + int wlen, ret, len = TASK_DATLEN(task); sock_cli_t *cli = TASK_ARG(task); sock_t *s = (sock_t*) cli->cli_parent; + u_char *buf = TASK_DATA(task); + struct pollfd pfd[1]; - if (s->sock_type == SOCK_STREAM) - wlen = send(TASK_FD(task), TASK_DATA(task), TASK_DATLEN(task), 0); - else - wlen = sendto(TASK_FD(task), TASK_DATA(task), TASK_DATLEN(task), 0, - &cli->cli_addr.sa, cli->cli_addr.sa.sa_len); - if (wlen < 1) - schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, cli->cli_fd); + pfd->fd = TASK_FD(task); + pfd->events = POLLOUT; + pfd->revents = 0; + for(; len > 0; len -= wlen, buf += wlen) { + ioUpdTimerSocket(cli); + if ((ret = poll(pfd, 1, s->sock_timeout.tv_sec * 1000)) < 1 || + pfd->revents & (POLLNVAL | POLLERR | POLLHUP)) { + if (!ret) + continue; + schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, 0); + break; + } + + if (s->sock_type == SOCK_STREAM) + wlen = send(TASK_FD(task), buf, len, 0); + else + wlen = sendto(TASK_FD(task), buf, len, 0, + &cli->cli_addr.sa, cli->cli_addr.sa.sa_len); + if (wlen < 1) { + schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, 0); + break; + } + } + taskExit(task, NULL); } @@ -147,9 +175,11 @@ io_txPty(sched_task_t *task) int wlen; sock_cli_t *cli = TASK_ARG(task); + ioUpdTimerSocket(cli); + wlen = write(TASK_FD(task), TASK_DATA(task), TASK_DATLEN(task)); if (wlen < 1) - schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, cli->cli_fd); + schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, 0); taskExit(task, NULL); } @@ -163,6 +193,8 @@ io_rxNet(sched_task_t *task) sockaddr_t sa; socklen_t salen = sizeof sa.ss; + ioUpdTimerSocket(cli); + if (s->sock_type == SOCK_STREAM) rlen = recv(TASK_FD(task), AIT_GET_BUF(&cli->cli_buf[0]), AIT_LEN(&cli->cli_buf[0]), 0); @@ -173,9 +205,9 @@ io_rxNet(sched_task_t *task) goto end; } if (rlen < 1) - schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, cli->cli_fd); + schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, 0); else - schedWrite(TASK_ROOT(task), io_txPty, cli, cli->cli_pty, + schedEvent(TASK_ROOT(task), io_txPty, cli, cli->cli_pty, AIT_GET_BUF(&cli->cli_buf[0]), rlen); end: schedReadSelf(task); @@ -188,11 +220,13 @@ io_rxPty(sched_task_t *task) int rlen; sock_cli_t *cli = TASK_ARG(task); + ioUpdTimerSocket(cli); + rlen = read(TASK_FD(task), AIT_GET_BUF(&cli->cli_buf[1]), AIT_LEN(&cli->cli_buf[1])); if (rlen < 1) - schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, cli->cli_fd); + schedEvent(TASK_ROOT(task), io_closeClient, cli, 0, NULL, 0); else - schedWrite(TASK_ROOT(task), io_txNet, cli, cli->cli_fd, + schedEvent(TASK_ROOT(task), io_txNet, cli, cli->cli_fd, AIT_GET_BUF(&cli->cli_buf[1]), rlen); schedReadSelf(task); @@ -256,9 +290,15 @@ io_bridgeClient(sched_task_t *task) argv = array_To(args); array_Destroy(&args); - execv(*argv, argv); +#if 0 + printf("Console %s\n", cli->cli_name); +#endif + rlen = execv(*argv, argv); + _exit(rlen); break; default: + cli->cli_pid = pid; + schedRead(TASK_ROOT(task), io_rxPty, cli, cli->cli_pty, TASK_ARG(task), 0); schedRead(TASK_ROOT(task), io_rxNet, cli, cli->cli_fd, @@ -368,14 +408,13 @@ void ioCloseSocket(sock_t ** __restrict s) { sock_cli_t *cli; + int stat; if (s && *s) { pthread_mutex_lock(&(*s)->sock_mtx); while ((cli = TAILQ_FIRST(&(*s)->sock_cli))) { TAILQ_REMOVE(&(*s)->sock_cli, cli, cli_node); - schedCancelby((*s)->sock_root, taskTIMER, CRITERIA_DATLEN, - (void*) cli->cli_fd, NULL); schedCancelby((*s)->sock_root, taskMAX, CRITERIA_ARG, cli, NULL); if ((*s)->sock_type == SOCK_STREAM) { @@ -384,6 +423,15 @@ ioCloseSocket(sock_t ** __restrict s) } AIT_FREE_VAL(&cli->cli_buf[1]); AIT_FREE_VAL(&cli->cli_buf[0]); + + if (cli->cli_pid > 0) { + kill(cli->cli_pid, SIGKILL); + while (waitpid(cli->cli_pid, &stat, WNOHANG) > 0) { + usleep(1000); + kill(cli->cli_pid, SIGKILL); + } + } + e_free(cli); } pthread_mutex_unlock(&(*s)->sock_mtx); @@ -469,8 +517,8 @@ ioUpdTimerSocket(sock_cli_t * __restrict c) else s = c->cli_parent; - schedCancelby(s->sock_root, taskTIMER, CRITERIA_DATLEN, (void*) c->cli_fd, NULL); - schedTimer(s->sock_root, io_closeClient, c, s->sock_timeout, NULL, c->cli_fd); + schedCancelby(s->sock_root, taskTIMER, CRITERIA_ARG, c, NULL); + schedTimer(s->sock_root, io_closeClient, c, s->sock_timeout, NULL, 0); } /* @@ -489,7 +537,7 @@ ioCloseClient(sock_cli_t * __restrict c) else s = c->cli_parent; - return !schedEvent(s->sock_root, io_closeClient, c, 0, NULL, c->cli_fd); + return !schedEvent(s->sock_root, io_closeClient, c, 0, NULL, 0); } /* @@ -524,4 +572,21 @@ ioBridgeProg2Socket(sock_t * __restrict s, const char schedRead(s->sock_root, io_bridgeClient, s, s->sock_fd, (void*) prgname, 0); return schedRun(s->sock_root, &s->sock_kill); +} + +/* + * ioSetupProg() - Setup program pool to socket server + * + * @s = Socket + * @p = Program pool + * return: -1 error or 0 ok + */ +int +ioSetupProg(sock_t * __restrict s, prog_t * __restrict p) +{ + if (!s) + return -1; + + s->sock_prog = p; + return 0; }