--- libaitrpc/src/srv.c 2012/05/11 13:23:12 1.8.2.6 +++ libaitrpc/src/srv.c 2012/05/14 15:22:22 1.9.2.2 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.8.2.6 2012/05/11 13:23:12 misho Exp $ +* $Id: srv.c,v 1.9.2.2 2012/05/14 15:22:22 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -94,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; } @@ -131,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,13 +166,12 @@ rxPacket(sched_task_t *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)) { @@ -271,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); @@ -300,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; } @@ -333,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"); @@ -457,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); @@ -660,9 +651,8 @@ rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) if (i >= srv->srv_numcli) { #ifdef HAVE_PTHREAD_YIELD pthread_yield(); -#else - usleep(1000000); #endif + usleep(1000000); continue; } @@ -682,6 +672,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; @@ -713,52 +704,20 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i { rpc_srv_t *srv = NULL; int n = 1; - struct hostent *host = NULL; io_sockaddr_t sa; - if (!concurentClients || !regProgID || - (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) { - rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n"); + if (!concurentClients || !regProgID) { + rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server"); return NULL; } + if (!io_gethostbyname(csHost, Port, &sa)) + return NULL; if (!Port) Port = RPC_DEFPORT; if (!netBuf) netBuf = BUFSIZ; else netBuf = io_align(netBuf, 1); /* align netBuf length */ - if (csHost && family != AF_LOCAL) { - host = gethostbyname2(csHost, family); - if (!host) { - rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno)); - return NULL; - } - } - memset(&sa, 0, sizeof sa); - sa.sa.sa_family = family; - switch (family) { - case AF_INET: - sa.sin.sin_len = sizeof(struct sockaddr_in); - sa.sin.sin_port = htons(Port); - if (csHost) - memcpy(&sa.sin.sin_addr, host->h_addr, host->h_length); - break; - case AF_INET6: - sa.sin6.sin6_len = sizeof(struct sockaddr_in6); - sa.sin6.sin6_port = htons(Port); - if (csHost) - memcpy(&sa.sin6.sin6_addr, host->h_addr, host->h_length); - break; - case AF_LOCAL: - sa.sun.sun_len = sizeof(struct sockaddr_un); - if (csHost) - strlcpy(sa.sun.sun_path, csHost, sizeof sa.sun.sun_path); - unlink(sa.sun.sun_path); - break; - default: - rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t start RPC server ...\n"); - return NULL; - } srv = malloc(sizeof(rpc_srv_t)); if (!srv) { @@ -779,7 +738,7 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i memcpy(&srv->srv_server.cli_sa, &sa, sizeof sa); /* create server socket */ - srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0); + srv->srv_server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0); if (srv->srv_server.cli_sock == -1) { LOGERR; free(srv); @@ -901,12 +860,11 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) /* activate BLOB server worker if srv->srv_blob.state == enable */ rpc_srv_execBLOBServer(srv); - if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) { + if (listen(srv->srv_server.cli_sock, srv->srv_numcli) == -1) { LOGERR; return -1; - } /*else + } 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); @@ -921,9 +879,8 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) if (i >= srv->srv_numcli) { #ifdef HAVE_PTHREAD_YIELD pthread_yield(); -#else - usleep(1000000); #endif + usleep(1000000); continue; } @@ -943,6 +900,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;