--- libaitrpc/src/srv.c 2012/03/29 01:34:16 1.8 +++ libaitrpc/src/srv.c 2012/05/14 08:39:06 1.9 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.8 2012/03/29 01:34:16 misho Exp $ +* $Id: srv.c,v 1.9 2012/05/14 08:39:06 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -60,8 +60,6 @@ txPacket(sched_task_t *task) int ret, wlen = sizeof(struct tagRPCCall); array_t *arr = NULL; - ioTRACE(RPC_TRACE_LEVEL); - if (rpc->call_argc) { f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash)); if (!f) { @@ -96,8 +94,6 @@ txPacket(sched_task_t *task) else if (ret != wlen) rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, " "really sended %d bytes", wlen, ret); - else - ioDEBUG(RPC_DEBUG_LEVEL, "Sended %d bytes", ret); return NULL; } @@ -113,8 +109,6 @@ execCall(sched_task_t *task) struct tagRPCCall *rpc = (struct tagRPCCall*) buf; int argc = ntohs(rpc->call_argc); - ioTRACE(RPC_TRACE_LEVEL); - /* Go decapsulate variables ... */ if (argc) { arr = io_buffer2vars(buf + sizeof(struct tagRPCCall), @@ -135,9 +129,6 @@ execCall(sched_task_t *task) rpc->call_rep.ret = RPC_ERROR(-1); rpc->call_rep.eno = RPC_ERROR(rpc_Errno); } else { - ioDEBUG(RPC_DEBUG_LEVEL, "RPC function %s from module %s", - AIT_GET_STR(&f->func_name), AIT_GET_LIKE(&f->func_file, char*)); - /* if client doesn't want reply */ argc = rpc->call_req.flags & RPC_NOREPLY; rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr)); @@ -171,19 +162,16 @@ rxPacket(sched_task_t *task) struct tagRPCCall *rpc; struct timespec ts; - ioTRACE(RPC_TRACE_LEVEL); - memset(buf, 0, TASK_DATLEN(task)); rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0); if (rlen == -1) { LOGERR; - s->srv_kill = s->srv_blob.state = kill; + c->cli_kill = kill; return NULL; } else if (!rlen) { /* receive EOF */ - s->srv_kill = s->srv_blob.state = kill; + c->cli_kill = kill; return NULL; - } else - ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen); + } do { if (rlen < sizeof(struct tagRPCCall)) { @@ -252,8 +240,6 @@ rpc_srv_dispatchCall(void *arg) sched_root_task_t *root; struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; - ioTRACE(RPC_TRACE_LEVEL); - if (!arg) { rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client"); return NULL; @@ -279,7 +265,7 @@ rpc_srv_dispatchCall(void *arg) schedRead(root, rxPacket, c, c->cli_sock, buf, s->srv_netbuf); - schedRun(root, (void*) &s->srv_kill); + schedRun(root, (void*) &c->cli_kill); schedEnd(&root); shutdown(c->cli_sock, SHUT_RDWR); @@ -297,8 +283,6 @@ txBLOB(sched_task_t *task) struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf; int wlen = sizeof(struct tagBLOBHdr); - ioTRACE(RPC_TRACE_LEVEL); - /* calculate CRC */ blob->hdr_crc ^= blob->hdr_crc; blob->hdr_crc = htons(crcFletcher16((u_short*) buf, wlen / 2)); @@ -310,8 +294,6 @@ txBLOB(sched_task_t *task) else if (wlen != sizeof(struct tagBLOBHdr)) rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, " "really sended %d bytes", sizeof(struct tagBLOBHdr), wlen); - else - ioDEBUG(RPC_DEBUG_LEVEL, "Sended %d bytes", wlen); return NULL; } @@ -328,8 +310,6 @@ rxBLOB(sched_task_t *task) u_short crc; struct timespec ts; - ioTRACE(RPC_TRACE_LEVEL); - /* check for disable service at this moment? */ if (!s || s->srv_blob.state == disable) { usleep(100000); @@ -345,13 +325,12 @@ rxBLOB(sched_task_t *task) rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0); if (rlen == -1) { LOGERR; - s->srv_blob.state = kill; + c->cli_kill = kill; return NULL; - } else if (!rlen || s->srv_kill == kill) { /* receive EOF */ - s->srv_blob.state = kill; + } else if (!rlen) { /* receive EOF */ + c->cli_kill = kill; return NULL; - } else - ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen); + } if (rlen < sizeof(struct tagBLOBHdr)) { rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet"); @@ -444,8 +423,6 @@ rpc_srv_dispatchVars(void *arg) u_char *buf; struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; - ioTRACE(RPC_TRACE_LEVEL); - if (!arg) { rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB"); return NULL; @@ -471,7 +448,7 @@ rpc_srv_dispatchVars(void *arg) schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr)); - schedRun(root, (void*) &s->srv_blob.state); + schedRun(root, (void*) &c->cli_kill); schedEnd(&root); shutdown(c->cli_sock, SHUT_RDWR); @@ -497,8 +474,6 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s int n = 1; io_sockaddr_t sa; - ioTRACE(RPC_TRACE_LEVEL); - if (!srv) { rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server"); return -1; @@ -579,8 +554,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); @@ -603,8 +576,6 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) register int i; rpc_blob_t *f; - ioTRACE(RPC_TRACE_LEVEL); - if (!srv || srv->srv_blob.state == disable) return; else @@ -621,7 +592,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; @@ -633,13 +603,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; } @@ -656,28 +622,27 @@ 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); - if (!srv || srv->srv_blob.state == kill) { rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server"); 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++) @@ -692,15 +657,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); @@ -711,6 +673,7 @@ rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) c->cli_parent = srv; /* spawn dispatch thread for BLOB client */ + c->cli_kill = enable; if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) { LOGERR; continue; @@ -745,8 +708,6 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i struct hostent *host = NULL; io_sockaddr_t sa; - ioTRACE(RPC_TRACE_LEVEL); - if (!concurentClients || !regProgID || (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) { rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n"); @@ -853,8 +814,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); @@ -876,8 +835,6 @@ rpc_srv_endServer(rpc_srv_t ** __restrict psrv) register int i; rpc_func_t *f; - ioTRACE(RPC_TRACE_LEVEL); - if (!psrv || !*psrv) { rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n"); return; @@ -900,7 +857,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); @@ -908,11 +864,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; } @@ -929,31 +881,29 @@ 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); - if (!srv) { rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n"); 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++) @@ -968,15 +918,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); @@ -987,6 +934,7 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) c->cli_parent = srv; /* spawn rpc client dispatcher */ + c->cli_kill = enable; if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) { LOGERR; continue; @@ -1015,14 +963,12 @@ rpc_srv_execCall(rpc_func_t * __restrict call, struct rpc_callback_t func; int ret; - ioTRACE(RPC_TRACE_LEVEL); - if (!call || !rpc || !call->func_parent) { rpc_SetErr(EINVAL, "Invalid parameter can`t exec function"); 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;