--- libaitrpc/src/srv.c 2012/03/29 00:40:59 1.7.2.3 +++ libaitrpc/src/srv.c 2012/05/11 12:53:35 1.8.2.5 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.7.2.3 2012/03/29 00:40:59 misho Exp $ +* $Id: srv.c,v 1.8.2.5 2012/05/11 12:53:35 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -87,7 +87,7 @@ txPacket(sched_task_t *task) /* calculate CRC */ rpc->call_crc ^= rpc->call_crc; - rpc->call_crc = htons(crcFletcher16((u_short*) buf, io_align(wlen, 1) / 2)); + rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2)); /* send reply */ ret = send(TASK_FD(task), buf, wlen, 0); @@ -200,8 +200,7 @@ rxPacket(sched_task_t *task) /* check integrity of packet */ crc = ntohs(rpc->call_crc); rpc->call_crc ^= rpc->call_crc; - if (crc != crcFletcher16((u_short*) (buf + off), - io_align(ntohs(rpc->call_len), 1) / 2)) { + if (crc != crcFletcher16((u_short*) (buf + off), ntohs(rpc->call_len) / 2)) { rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet"); off += ntohs(rpc->call_len); @@ -302,7 +301,7 @@ txBLOB(sched_task_t *task) /* calculate CRC */ blob->hdr_crc ^= blob->hdr_crc; - blob->hdr_crc = htons(crcFletcher16((u_short*) buf, io_align(wlen, 1) / 2)); + blob->hdr_crc = htons(crcFletcher16((u_short*) buf, wlen / 2)); /* send reply */ wlen = send(TASK_FD(task), buf, wlen, 0); @@ -364,7 +363,7 @@ rxBLOB(sched_task_t *task) /* check integrity of packet */ crc = ntohs(blob->hdr_crc); blob->hdr_crc ^= blob->hdr_crc; - if (crc != crcFletcher16((u_short*) buf, io_align(rlen, 1) / 2)) { + if (crc != crcFletcher16((u_short*) buf, rlen / 2)) { rpc_SetErr(ERPCMISMATCH, "Bad CRC BLOB packet"); schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), TASK_DATA(task), TASK_DATLEN(task)); @@ -580,8 +579,6 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s } else memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t)); - pthread_mutex_init(&srv->srv_blob.mtx, NULL); - rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0); rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 1); rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 1); @@ -622,7 +619,6 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) shutdown(c->cli_sock, SHUT_RDWR); close(srv->srv_blob.server.cli_sock); - pthread_mutex_lock(&srv->srv_blob.mtx); if (srv->srv_blob.clients) { free(srv->srv_blob.clients); srv->srv_blob.clients = NULL; @@ -634,13 +630,9 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) rpc_srv_blobFree(srv, f); free(f); } - pthread_mutex_unlock(&srv->srv_blob.mtx); AIT_FREE_VAL(&srv->srv_blob.dir); - while (pthread_mutex_trylock(&srv->srv_blob.mtx) == EBUSY); - pthread_mutex_destroy(&srv->srv_blob.mtx); - /* at final, save disable BLOB service state */ srv->srv_blob.state = disable; } @@ -657,10 +649,9 @@ rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) socklen_t salen = sizeof(io_sockaddr_t); register int i; rpc_cli_t *c; - fd_set fds; int ret; - struct timeval tv = { DEF_RPC_TIMEOUT, 0 }; pthread_attr_t attr; + struct pollfd pfd; ioTRACE(RPC_TRACE_LEVEL); @@ -669,16 +660,19 @@ rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) return -1; } - tv.tv_sec = srv->srv_session.sess_timeout; - if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) { LOGERR; return -1; - } + }/* else + fcntl(srv->srv_blob.server.cli_sock, F_SETFL, + fcntl(srv->srv_blob.server.cli_sock, F_GETFL) | O_NONBLOCK); + */ pthread_attr_init(&attr); pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + pfd.fd = srv->srv_blob.server.cli_sock; + pfd.events = POLLIN | POLLPRI; /* main BLOB loop */ while (srv->srv_blob.state != kill && srv->srv_kill != kill) { for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++) @@ -693,15 +687,12 @@ rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) continue; } - FD_ZERO(&fds); - FD_SET(srv->srv_blob.server.cli_sock, &fds); - ret = select(srv->srv_blob.server.cli_sock + 1, &fds, NULL, NULL, &tv); - if (ret == -1) { + if ((ret = poll(&pfd, 1, srv->srv_session.sess_timeout * 1000)) == -1 || + pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { LOGERR; ret = 1; break; - } - if (!ret) + } else if (!ret) continue; c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa.sa, &salen); @@ -854,8 +845,6 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i } else memset(srv->srv_clients, 0, srv->srv_numcli * sizeof(rpc_cli_t)); - pthread_mutex_init(&srv->srv_mtx, NULL); - rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0); rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 1); rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 4); @@ -901,7 +890,6 @@ rpc_srv_endServer(rpc_srv_t ** __restrict psrv) } /* detach exported calls */ - pthread_mutex_lock(&(*psrv)->srv_mtx); while ((f = (*psrv)->srv_funcs)) { (*psrv)->srv_funcs = f->func_next; io_freeVars(&f->func_vars); @@ -909,11 +897,7 @@ rpc_srv_endServer(rpc_srv_t ** __restrict psrv) AIT_FREE_VAL(&f->func_file); free(f); } - pthread_mutex_unlock(&(*psrv)->srv_mtx); - while (pthread_mutex_trylock(&(*psrv)->srv_mtx) == EBUSY); - pthread_mutex_destroy(&(*psrv)->srv_mtx); - free(*psrv); *psrv = NULL; } @@ -930,10 +914,9 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) socklen_t salen = sizeof(io_sockaddr_t); register int i; rpc_cli_t *c; - fd_set fds; int ret; - struct timeval tv = { DEF_RPC_TIMEOUT, 0 }; pthread_attr_t attr; + struct pollfd pfd; ioTRACE(RPC_TRACE_LEVEL); @@ -942,19 +925,21 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) return -1; } - tv.tv_sec = srv->srv_session.sess_timeout; - /* activate BLOB server worker if srv->srv_blob.state == enable */ rpc_srv_execBLOBServer(srv); if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) { LOGERR; return -1; - } + } /*else + fcntl(srv->srv_server.cli_sock, F_SETFL, fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK); + */ pthread_attr_init(&attr); pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + pfd.fd = srv->srv_server.cli_sock; + pfd.events = POLLIN | POLLPRI; /* main rpc loop */ while (srv->srv_kill != kill) { for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++) @@ -969,15 +954,12 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) continue; } - FD_ZERO(&fds); - FD_SET(srv->srv_server.cli_sock, &fds); - ret = select(srv->srv_server.cli_sock + 1, &fds, NULL, NULL, &tv); - if (ret == -1) { + if ((ret = poll(&pfd, 1, srv->srv_session.sess_timeout * 1000)) == -1 || + pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { LOGERR; ret = 1; break; - } - if (!ret) + } else if (!ret) continue; c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa.sa, &salen); @@ -1023,7 +1005,7 @@ rpc_srv_execCall(rpc_func_t * __restrict call, struct return -1; } - dl = dlopen(AIT_VOID(&call->func_file), RTLD_NOW); + dl = dlopen(AIT_ADDR(&call->func_file), RTLD_NOW); if (!dl) { rpc_SetErr(ENOENT, "Can`t attach module %s!", dlerror()); return -1;