--- libaitrpc/src/srv.c 2020/06/25 19:11:00 1.30 +++ libaitrpc/src/srv.c 2024/03/20 17:32:31 1.31 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.30 2020/06/25 19:11:00 misho Exp $ +* $Id: srv.c,v 1.31 2024/03/20 17:32:31 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -12,7 +12,7 @@ terms: All of the documentation and software included in the ELWIX and AITNET Releases is copyrighted by ELWIX - Sofia/Bulgaria -Copyright 2004 - 2016 +Copyright 2004 - 2024 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -87,7 +87,8 @@ rpc_freeCli(rpc_cli_t * __restrict c) { rpc_srv_t *s = c->cli_parent; - schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL); + if (s->srv_proto == SOCK_STREAM) + schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL); /* free buffer */ AIT_FREE_VAL(&c->cli_buf); @@ -146,6 +147,11 @@ _allocClient(rpc_srv_t * __restrict srv, sockaddr_t * /* alloc empty buffer */ AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf); + if (!AIT_GET_BUF(&c->cli_buf)) { + array_Del(srv->srv_clients, n, 0); + e_free(c); + c = NULL; + } } return c; @@ -291,7 +297,7 @@ execCall(sched_task_t *task) if (TASK_VAL(task)) { /* without reply */ ait_freeVars(&c->cli_vars); - } else { + } else if (rpc->call_io & RPC_REQ) { /* reply */ rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c)); } @@ -307,7 +313,7 @@ rxPacket(sched_task_t *task) { rpc_cli_t *c = TASK_ARG(task); rpc_srv_t *s = c->cli_parent; - int len, noreply = 0, rlen = AIT_LEN(&c->cli_buf); + int len, noreply = 0, rlen; #if 0 u_short crc; #endif @@ -323,8 +329,15 @@ rxPacket(sched_task_t *task) /* prepare rx */ len = recv(TASK_FD(task), &b, sizeof b, MSG_PEEK); - if (len == sizeof b) + if (len < 1) { + /* close connection */ + schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), 0, NULL, 0); + taskExit(task, NULL); + } else if (len == sizeof b) rlen = ntohl(b.call_len); + else + goto end; rlen = recv(TASK_FD(task), buf, rlen, 0); if (rlen == -1) { @@ -387,10 +400,10 @@ rxPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len); err: /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET], TASK_ARG(task), TASK_FD(task), rpc, len); - +end: /* lets get next packet */ schedReadSelf(task); taskExit(task, NULL); @@ -544,7 +557,7 @@ rxUDPPacket(sched_task_t *task) /* check integrity of packet */ crc = ntohs(rpc->call_crc); rpc->call_crc ^= rpc->call_crc; - if (crc != crcFletcher16((u_short*) rpc, len / 2)) + if (crc != crcFletcher16((u_short*) AIT_GET_BUF(&b), len / 2)) goto end; /* check RPC packet session info */ @@ -553,7 +566,7 @@ rxUDPPacket(sched_task_t *task) c = _allocClient(srv, &sa); if (!c) { - EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n"); + EVERBOSE(1, "RPC client quota exceeded!"); usleep(2000); /* blocked client delay */ goto end; } else { @@ -575,7 +588,7 @@ rxUDPPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], c, TASK_FD(task), rpc, len); end: @@ -708,7 +721,7 @@ rxRAWPacket(sched_task_t *task) /* check integrity of packet */ crc = ntohs(rpc->call_crc); rpc->call_crc ^= rpc->call_crc; - if (crc != crcFletcher16((u_short*) rpc, len / 2)) + if (crc != crcFletcher16((u_short*) AIT_GET_BUF(&b), len / 2)) goto end; /* check RPC packet session info */ @@ -739,7 +752,7 @@ rxRAWPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], c, TASK_FD(task), rpc, len); end: @@ -909,7 +922,7 @@ rxBPFPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], c, TASK_FD(task), rpc, len); end: @@ -990,7 +1003,7 @@ rxEXTPacket(sched_task_t *task) { rpc_srv_t *srv = TASK_ARG(task); rpc_cli_t *c = NULL; - int len, noreply = 0, rlen = AIT_LEN(&c->cli_buf); + int len, noreply = 0, rlen; struct tagRPCCall *rpc; struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; sockaddr_t sa; @@ -1039,7 +1052,7 @@ rxEXTPacket(sched_task_t *task) schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); /* send RPC reply */ - if (!noreply) + if (!noreply && (rpc->call_io & RPC_REQ)) schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], c, TASK_FD(task), rpc, len); end: @@ -1954,4 +1967,26 @@ rpc_srv_initServerExt(u_char InstID, int netBuf, int f rpc_register_srvPing(srv); return srv; +} + +/* + * rpc_srv_Return() - Prepare IPC return answer to RPC client + * + * @c = RPC client + * return: number of arguments in response + */ +int +rpc_srv_Return(rpc_cli_t *c) +{ + rpc_srv_t *s = c->cli_parent; + u_char *buf = AIT_GET_BUF(&c->cli_buf); + struct tagRPCCall *rpc = (struct tagRPCCall*) buf; + + if (!RPC_CHK_NOREPLY(rpc)) { + rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c)); + schedWrite(s->srv_root, cbProto[s->srv_proto][CB_TXPACKET], c, c->cli_sock, rpc, 0); + } else + rpc->call_argc ^= rpc->call_argc; + + return rpc->call_argc; }