--- libaitrpc/src/srv.c 2020/06/25 19:11:00 1.30 +++ libaitrpc/src/srv.c 2024/02/26 17:52:41 1.30.2.8 @@ -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.30.2.8 2024/02/26 17:52:41 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); @@ -291,7 +292,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)); } @@ -302,6 +303,21 @@ execCall(sched_task_t *task) taskExit(task, NULL); } +int +rpc_srv_Return(sched_root_task_t *root, 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(root, cbProto[s->srv_proto][CB_TXPACKET], c, c->cli_sock, rpc, 0); + } + + return rpc->call_argc; +} + static void * rxPacket(sched_task_t *task) { @@ -323,8 +339,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 +410,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 +567,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 */ @@ -575,7 +598,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 +731,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 +762,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 +932,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: @@ -1039,7 +1062,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: @@ -1558,7 +1581,8 @@ rpc_srv_initServer(u_char InstID, int concurentClients pthread_mutex_destroy(&srv->srv_funcs.mtx); e_free(srv); return NULL; - } + } else + schedSignalDispatch(srv->srv_root, 42); /* init pool for clients */ srv->srv_clients = array_Init(concurentClients); @@ -1680,6 +1704,7 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) schedPolling(srv->srv_root, &ts, NULL); /* main rpc loop */ schedRun(srv->srv_root, &srv->srv_kill); + schedSignalDispatch(srv->srv_root, 0); /* close all clients connections & server socket */ for (i = 0; i < array_Size(srv->srv_clients); i++) { @@ -1818,7 +1843,8 @@ rpc_srv_initServer2(u_char InstID, int concurentClient pthread_mutex_destroy(&srv->srv_funcs.mtx); e_free(srv); return NULL; - } + } else + schedSignalDispatch(srv->srv_root, 42); /* init pool for clients */ srv->srv_clients = array_Init(concurentClients); @@ -1936,7 +1962,8 @@ rpc_srv_initServerExt(u_char InstID, int netBuf, int f pthread_mutex_destroy(&srv->srv_funcs.mtx); e_free(srv); return NULL; - } + } else + schedSignalDispatch(srv->srv_root, 42); /* init pool for clients */ srv->srv_clients = array_Init(1);