--- libaitio/src/sock.c 2013/08/27 19:24:37 1.2.2.4 +++ libaitio/src/sock.c 2013/11/21 15:01:22 1.4.4.3 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: sock.c,v 1.2.2.4 2013/08/27 19:24:37 misho Exp $ +* $Id: sock.c,v 1.4.4.3 2013/11/21 15:01:22 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -84,6 +84,7 @@ ioInitSocket(int role, int type, int proto, const char return NULL; } else { buflen = buflen ? buflen : BUFSIZ; + buflen = E_ALIGN(buflen, 2); /* align buflen length */ AIT_SET_BUFSIZ(&s->sock_buf, 0, buflen); } @@ -119,6 +120,14 @@ ioInitSocket(int role, int type, int proto, const char return NULL; } + s->sock_root = schedBegin(); + if (!s->sock_root) { + io_SetErr(sched_GetErrno(), "%s", sched_GetError()); + AIT_FREE_VAL(&s->sock_buf); + e_free(s); + return NULL; + } + pthread_mutex_init(&s->sock_mtx, NULL); return s; } @@ -150,6 +159,8 @@ ioCloseSocket(sock_t ** __restrict s) AIT_FREE_VAL(&(*s)->sock_buf); + schedEnd(&(*s)->sock_root); + pthread_mutex_destroy(&(*s)->sock_mtx); e_free(*s); *s = NULL; @@ -161,10 +172,11 @@ ioCloseSocket(sock_t ** __restrict s) * * @s = Socket * @arg = Server role = listen backlog queue and Client role = peer address + * @timeout = Socket timeout in sec (default -1 infinit) * return: -1 error or 0 ok */ int -ioUpSocket(sock_t * __restrict s, void *arg) +ioUpSocket(sock_t * __restrict s, void *arg, int timeout) { int ret = 0; sockaddr_t *peer = (sockaddr_t*) arg; @@ -172,6 +184,11 @@ ioUpSocket(sock_t * __restrict s, void *arg) if (!s || !arg) return -1; + else { + s->sock_timeout.tv_sec = timeout; + s->sock_timeout.tv_nsec = (timeout < 1) ? timeout : 0; + schedPolling(s->sock_root, &s->sock_timeout, NULL); + } switch (s->sock_role) { case IO_SOCK_ROLE_CLIENT: @@ -202,101 +219,126 @@ ioUpSocket(sock_t * __restrict s, void *arg) return ret; } -static void -thrCliClean(void *arg) +static void * +io_closeClient(sched_task_t *task) { - sock_cli_t *cli = (sock_cli_t*) arg; + sock_cli_t *cli = (sock_cli_t*) TASK_ARG(task); + sock_t *s = (sock_t*) cli->cli_parent; + int sock = (int) TASK_DATLEN(task); - close(cli->cli_fd); + if (s->sock_type == SOCK_STREAM) { + shutdown(sock, SHUT_RDWR); + close(sock); + } AIT_FREE_VAL(&cli->cli_buf); + + pthread_mutex_lock(&s->sock_mtx); + TAILQ_REMOVE(&s->sock_cli, cli, cli_node); + pthread_mutex_unlock(&s->sock_mtx); + e_free(cli); + taskExit(task, NULL); } static void * -io_thrCliWrapper(void *arg) +io_acceptClient(sched_task_t *task) { - void *ret; - sock_cli_t *cli = (sock_cli_t*) arg; + int c, rlen; + sockaddr_t sa; + socklen_t salen = sizeof sa.ss; + sock_cli_t *cli = NULL; + sock_t *s = (sock_t*) TASK_ARG(task); - pthread_cleanup_push(thrCliClean, arg); + if (s->sock_type == SOCK_STREAM) { + if ((c = accept(TASK_FD(task), &sa.sa, &salen)) == -1) { + LOGERR; + goto end; + } + } else { + if ((rlen = recvfrom(TASK_FD(task), + AIT_GET_BUF(&s->sock_buf), AIT_LEN(&s->sock_buf), + MSG_PEEK, &sa.sa, &salen)) == -1) { + LOGERR; + goto end; + } else + c = TASK_FD(task); + } - ret = cli->cli_func(cli); + cli = e_malloc(sizeof(sock_cli_t)); + if (!cli) { + io_SetErr(elwix_GetErrno(), "%s", elwix_GetError()); + if (s->sock_type == SOCK_STREAM) + close(c); + goto end; + } else { + memset(cli, 0, sizeof(sock_cli_t)); + pthread_mutex_lock(&s->sock_mtx); + TAILQ_INSERT_TAIL(&s->sock_cli, cli, cli_node); + pthread_mutex_unlock(&s->sock_mtx); + } - pthread_cleanup_pop(42); - pthread_exit(NULL); + cli->cli_parent = TASK_ARG(task); + cli->cli_fd = c; + cli->cli_func = TASK_DATA(task); + memcpy(&cli->cli_addr, &sa, sizeof cli->cli_addr); + AIT_SET_BUFSIZ(&cli->cli_buf, 0, AIT_LEN(&s->sock_buf)); + + schedRead(TASK_ROOT(task), cli->cli_func, cli, cli->cli_fd, TASK_ARG(task), 0); + ioUpdTimerSocket(cli); +end: + schedReadSelf(task); + taskExit(task, NULL); } /* - * ioAcceptSocket() - Accept clients + * ioUpdTimerSocket() - Update timeout of socket * - * @s = Socket - * @f = callback function for client handling - * @arg = optional argument for callback function - * return: -1 error or 0 ok + * @c = Client socket + * return: none */ -int -ioAcceptSocket(sock_t * __restrict s, sock_cb_t f, void *arg) +void +ioUpdTimerSocket(sock_cli_t * __restrict c) { - struct pollfd pfd[1]; - socklen_t salen; - sockaddr_t sa; - int c, rlen; - sock_cli_t *cli; - u_char buf[BUFSIZ] = { [0 ... BUFSIZ - 1] = 0 }; + sock_t *s; - if (!s || s->sock_role == IO_SOCK_ROLE_CLIENT || !f) - return -1; + if (!c) + return; + else + s = c->cli_parent; - pfd->fd = s->sock_fd; - pfd->events = POLLIN | POLLPRI; - do { - if (poll(pfd, 1, -1) < 1 || - pfd->revents & (POLLNVAL | POLLHUP || POLLERR)) { - LOGERR; - return -1; - } else - salen = sizeof sa.ss; + 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); +} - if (s->sock_type == SOCK_STREAM) { - if ((c = accept(s->sock_fd, &sa.sa, &salen)) == -1) { - LOGERR; - return -1; - } - } else { - if ((rlen = recvfrom(s->sock_fd, buf, sizeof buf, MSG_PEEK, - &sa.sa, &salen)) == -1) { - LOGERR; - return -1; - } else - c = s->sock_fd; - } +/* + * ioLoopSocket() - Start socket scheduler + * + * @s = Socket + * @rcb = Read callback + * return: -1 error or return result from scheduler + */ +int +ioLoopSocket(sock_t * __restrict s, sched_task_func_t rcb) +{ + schedRead(s->sock_root, io_acceptClient, s, s->sock_fd, rcb, 0); + return schedRun(s->sock_root, &s->sock_kill); +} - cli = e_malloc(sizeof(sock_cli_t)); - if (!cli) { - io_SetErr(elwix_GetErrno(), "%s", elwix_GetError()); - close(c); - return -1; - } else - memset(cli, 0, sizeof(sock_cli_t)); +/* + * ioCloseClient() - Close client socket + * + * @c = Client socket + * return: 0 ok or !=0 error + */ +int +ioCloseClient(sock_cli_t * __restrict c) +{ + sock_t *s; - cli->cli_parent = s; - cli->cli_fd = c; - cli->cli_func = f; - cli->cli_arg = arg; - memcpy(&cli->cli_addr, &sa, sizeof cli->cli_addr); - AIT_SET_BUFSIZ(&cli->cli_buf, 0, AIT_LEN(&s->sock_buf)); + if (!c) + return -1; + else + s = c->cli_parent; - if (pthread_create(&cli->cli_tid, NULL, io_thrCliWrapper, cli) == -1) { - LOGERR; - close(c); - AIT_FREE_VAL(&cli->cli_buf); - e_free(cli); - return -1; - } else { - pthread_detach(cli->cli_tid); - TAILQ_INSERT_TAIL(&s->sock_cli, cli, cli_node); - } - } while (42); - - return 0; + return !schedEvent(s->sock_root, io_closeClient, c, 0, NULL, c->cli_fd); }