--- libaitrpc/src/srv.c 2013/08/22 15:31:09 1.18 +++ libaitrpc/src/srv.c 2013/08/23 13:13:07 1.18.2.1 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.18 2013/08/22 15:31:09 misho Exp $ +* $Id: srv.c,v 1.18.2.1 2013/08/23 13:13:07 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -164,6 +164,7 @@ txPacket(sched_task_t *task) u_char *buf = AIT_GET_BUF(&c->cli_buf); struct tagRPCCall *rpc = (struct tagRPCCall*) buf; int ret, estlen, wlen = sizeof(struct tagRPCCall); + struct pollfd pfd; if (rpc->call_argc) { f = rpc_srv_getCall(s, ntohs(rpc->call_tag)); @@ -188,6 +189,7 @@ txPacket(sched_task_t *task) ait_freeVars(&c->cli_vars); if (ret == -1) { rpc_SetErr(EBADRPC, "Prepare RPC packet failed"); + rpc->call_argc ^= rpc->call_argc; rpc->call_rep.ret = RPC_ERROR(-1); rpc->call_rep.eno = RPC_ERROR(rpc_Errno); @@ -205,11 +207,29 @@ txPacket(sched_task_t *task) #endif /* send reply */ - ret = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL); - if (ret == -1 || ret != wlen) { - /* close connection */ - schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], - TASK_ARG(task), 0, NULL, 0); + pfd.fd = TASK_FD(task); + pfd.events = POLLOUT; + for (; wlen > 0; wlen -= ret, buf += ret) { + if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || + pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { + if (ret) + LOGERR; + else + rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond"); + + /* close connection */ + schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), 0, NULL, 0); + return NULL; + } + ret = send(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), MSG_NOSIGNAL); + printf("...ret=%d %s %d\n", ret, strerror(errno), wlen); + if (ret == -1) { + /* close connection */ + schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), 0, NULL, 0); + return NULL; + } } return NULL; @@ -282,6 +302,7 @@ rxPacket(sched_task_t *task) #endif u_char *buf = AIT_GET_BUF(&c->cli_buf); struct tagRPCCall *rpc = (struct tagRPCCall*) buf; + struct pollfd pfd; memset(buf, 0, sizeof(struct tagRPCCall)); rlen = recv(TASK_FD(task), rpc, sizeof(struct tagRPCCall), 0); @@ -302,12 +323,26 @@ rxPacket(sched_task_t *task) if (len) { /* get next part of packet */ memset(buf, 0, len); - rlen = recv(TASK_FD(task), buf, len, 0); - if (rlen == -1) { - /* close connection */ - schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], - TASK_ARG(task), 0, NULL, 0); - return NULL; + pfd.fd = TASK_FD(task); + pfd.events = POLLIN | POLLPRI; + for (; len > 0; len -= rlen, buf += rlen) { + if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || + pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { + if (rlen) + LOGERR; + else + rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond"); + schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), 0, NULL, 0); + return NULL; + } + rlen = recv(TASK_FD(task), buf, len, 0); + if (rlen == -1) { + /* close connection */ + schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), 0, NULL, 0); + return NULL; + } } }