--- libaitrpc/src/srv.c 2013/04/02 09:17:06 1.14.2.1 +++ libaitrpc/src/srv.c 2013/07/15 14:16:52 1.16.2.1 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.14.2.1 2013/04/02 09:17:06 misho Exp $ +* $Id: srv.c,v 1.16.2.1 2013/07/15 14:16:52 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -67,7 +67,7 @@ static sched_task_func_t cbProto[SOCK_RAW + 1][4] = { }; -inline void +void rpc_freeCli(rpc_cli_t * __restrict c) { rpc_srv_t *s = c->cli_parent; @@ -193,9 +193,11 @@ txPacket(sched_task_t *task) rpc->call_len = htons(wlen); +#if 0 /* calculate CRC */ rpc->call_crc ^= rpc->call_crc; rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2)); +#endif /* send reply */ ret = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL); @@ -269,7 +271,10 @@ rxPacket(sched_task_t *task) rpc_cli_t *c = TASK_ARG(task); rpc_srv_t *s = c->cli_parent; int len, rlen, noreply; - u_short crc, off = TASK_DATLEN(task); + u_short off = TASK_DATLEN(task); +#if 0 + u_short crc; +#endif u_char *buf = AIT_GET_BUF(&c->cli_buf); struct tagRPCCall *rpc; @@ -308,6 +313,7 @@ rxPacket(sched_task_t *task) break; } +#if 0 /* check integrity of packet */ crc = ntohs(rpc->call_crc); rpc->call_crc ^= rpc->call_crc; @@ -318,6 +324,7 @@ rxPacket(sched_task_t *task) /* try next packet remaining into buffer */ continue; } +#endif noreply = RPC_CHK_NOREPLY(rpc); @@ -522,7 +529,7 @@ end: /* ------------------------------------------------------ */ -inline void +void rpc_freeBLOBCli(rpc_cli_t * __restrict c) { rpc_srv_t *s = c->cli_parent; @@ -552,6 +559,16 @@ closeBLOBClient(sched_task_t *task) } static void * +toutBLOB(sched_task_t *task) +{ + rpc_cli_t *c = TASK_DATA(task); + + rpc_srv_unregisterBLOB((rpc_srv_t*) c->cli_parent, (uint32_t) TASK_ARG(task)); + + return NULL; +} + +static void * txBLOB(sched_task_t *task) { rpc_cli_t *c = TASK_ARG(task); @@ -576,7 +593,7 @@ rxBLOB(sched_task_t *task) rpc_blob_t *b; struct tagBLOBHdr blob; int rlen; - u_short crc; + struct timespec ts = { RPC_BLOB_TIMEOUT, 0 }; memset(&blob, 0, sizeof blob); rlen = recv(TASK_FD(task), &blob, sizeof blob, 0); @@ -595,7 +612,7 @@ rxBLOB(sched_task_t *task) } /* check RPC packet session info */ - if ((crc = rpc_chkPktSession(&blob.hdr_session, &s->srv_session))) { + if (rpc_chkPktSession(&blob.hdr_session, &s->srv_session)) { rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session"); blob.hdr_cmd = error; goto end; @@ -626,9 +643,16 @@ rxBLOB(sched_task_t *task) /* set new BLOB variable for reply :) */ blob.hdr_var = htonl(b->blob_var); + /* set live BLOB timeout */ + ts.tv_sec = blob.hdr_ret ? blob.hdr_ret : RPC_BLOB_TIMEOUT; + /* receive BLOB from client */ blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b)); rpc_srv_blobUnmap(b); + + schedTimer(TASK_ROOT(task), toutBLOB, + (void*) (intptr_t) ntohl(blob.hdr_var), ts, + TASK_ARG(task), ntohl(blob.hdr_len)); } else { blob.hdr_cmd = error; blob.hdr_ret = RPC_ERROR(-1); @@ -639,6 +663,9 @@ rxBLOB(sched_task_t *task) blob.hdr_cmd = error; blob.hdr_ret = RPC_ERROR(-1); } + + schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_ARG, + (void*) (intptr_t) ntohl(blob.hdr_var), NULL); break; default: rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd); @@ -820,7 +847,7 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s * @srv = RPC Server instance * return: none */ -inline void +void rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) { if (!srv) @@ -1023,7 +1050,7 @@ err: /* error condition */ * @psrv = RPC Server instance * return: none */ -inline void +void rpc_srv_endServer(rpc_srv_t ** __restrict psrv) { if (!psrv || !*psrv)