--- libaitrpc/src/srv.c 2012/05/17 16:47:35 1.9.2.28 +++ libaitrpc/src/srv.c 2012/11/05 17:06:51 1.11.2.4 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.9.2.28 2012/05/17 16:47:35 misho Exp $ +* $Id: srv.c,v 1.11.2.4 2012/11/05 17:06:51 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -64,7 +64,7 @@ closeClient(sched_task_t *task) io_arrayDel(s->srv_clients, c->cli_id, 0); if (c) - free(c); + io_free(c); return NULL; } @@ -89,11 +89,11 @@ txPacket(sched_task_t *task) rpc->call_rep.ret = RPC_ERROR(-1); rpc->call_rep.eno = RPC_ERROR(rpc_Errno); } else { - rpc->call_argc = htons(io_arraySize(f->func_vars)); + rpc->call_argc = htons(io_arraySize(RPC_RETVARS(c))); /* Go Encapsulate variables */ - ret = io_vars2buffer(buf + wlen, sizeof buf - wlen, f->func_vars); + ret = io_vars2buffer(buf + wlen, sizeof buf - wlen, RPC_RETVARS(c)); /* Free return values */ - io_clrVars(f->func_vars); + io_freeVars(&c->cli_vars); if (ret == -1) { rpc_SetErr(EBADRPC, "Prepare RPC packet failed"); rpc->call_argc ^= rpc->call_argc; @@ -135,7 +135,7 @@ execCall(sched_task_t *task) if (argc) { arr = io_buffer2vars(buf + sizeof(struct tagRPCCall), AIT_LEN(&c->cli_buf) - TASK_VAL(task) - sizeof(struct tagRPCCall), - argc, 1); + argc, 42); if (!arr) { rpc_SetErr(ERPCMISMATCH, "#%d - %s", io_GetErrno(), io_GetError()); rpc->call_argc ^= rpc->call_argc; @@ -153,8 +153,8 @@ execCall(sched_task_t *task) rpc->call_rep.eno = RPC_ERROR(rpc_Errno); } else { /* 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)); + argc = RPC_CHK_NOREPLY(rpc); + rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(c, rpc, f->func_name, arr)); if (rpc->call_rep.ret == htonl(-1)) { rpc->call_rep.eno = RPC_ERROR(errno); rpc->call_argc ^= rpc->call_argc; @@ -162,11 +162,11 @@ execCall(sched_task_t *task) rpc->call_rep.eno ^= rpc->call_rep.eno; if (argc) { /* without reply */ - io_clrVars(f->func_vars); + io_freeVars(&c->cli_vars); rpc->call_argc ^= rpc->call_argc; } else { /* reply */ - rpc->call_argc = htons(io_arraySize(f->func_vars)); + rpc->call_argc = htons(io_arraySize(RPC_RETVARS(c))); } } } @@ -185,7 +185,10 @@ rxPacket(sched_task_t *task) u_char *buf = AIT_GET_BUF(&c->cli_buf); struct tagRPCCall *rpc; - memset(buf, 0, AIT_LEN(&c->cli_buf)); + if (!off) + memset(buf, 0, AIT_LEN(&c->cli_buf)); + else + memmove(buf, buf + off, AIT_LEN(&c->cli_buf) - off); rlen = recv(TASK_FD(task), buf + off, AIT_LEN(&c->cli_buf) - off, 0); if (rlen < 1) { /* close connection */ @@ -229,7 +232,7 @@ rxPacket(sched_task_t *task) continue; } - noreply = rpc->call_req.flags & RPC_NOREPLY; + noreply = RPC_CHK_NOREPLY(rpc); /* check RPC packet session info */ if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) { @@ -267,7 +270,7 @@ acceptClients(sched_task_t *task) (c = io_array(srv->srv_clients, i, rpc_cli_t*)); i++); if (c) /* no more free slots! */ goto end; - c = malloc(sizeof(rpc_cli_t)); + c = io_malloc(sizeof(rpc_cli_t)); if (!c) { LOGERR; srv->srv_kill = 1; @@ -318,7 +321,7 @@ closeBLOBClient(sched_task_t *task) io_arrayDel(s->srv_blob.clients, c->cli_id, 0); if (c) - free(c); + io_free(c); return NULL; } @@ -421,7 +424,7 @@ rxBLOB(sched_task_t *task) } break; case unset: - if (rpc_srv_unregisterBLOB(s, blob.hdr_var) == -1) { + if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) { blob.hdr_cmd = error; blob.hdr_ret = RPC_ERROR(-1); } @@ -446,13 +449,16 @@ acceptBLOBClients(sched_task_t *task) rpc_cli_t *c = NULL; register int i; socklen_t salen = sizeof(io_sockaddr_t); +#ifdef TCP_NOPUSH + int n = 1; +#endif /* check free slots for connect */ for (i = 0; i < io_arraySize(srv->srv_blob.clients) && (c = io_array(srv->srv_blob.clients, i, rpc_cli_t*)); i++); if (c) /* no more free slots! */ goto end; - c = malloc(sizeof(rpc_cli_t)); + c = io_malloc(sizeof(rpc_cli_t)); if (!c) { LOGERR; srv->srv_kill = srv->srv_blob.kill = 1; @@ -474,8 +480,12 @@ acceptBLOBClients(sched_task_t *task) AIT_FREE_VAL(&c->cli_buf); io_arrayDel(srv->srv_blob.clients, i, 42); goto end; - } else + } else { +#ifdef TCP_NOPUSH + setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n); +#endif fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK); + } schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0); end: @@ -663,7 +673,7 @@ rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv) TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node); rpc_srv_blobFree(srv, b); - free(b); + io_free(b); } schedEnd(&srv->srv_blob.root); @@ -689,7 +699,7 @@ rpc_srv_initServer(u_int regProgID, u_char regProcID, { int n = 1; rpc_srv_t *srv = NULL; - io_sockaddr_t sa; + io_sockaddr_t sa = IO_SOCKADDR_INIT; if (!concurentClients || !regProgID) { rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server"); @@ -702,7 +712,7 @@ rpc_srv_initServer(u_int regProgID, u_char regProcID, if (netBuf < RPC_MIN_BUFSIZ) netBuf = BUFSIZ; else - netBuf = io_align(netBuf, 1); /* align netBuf length */ + netBuf = io_align(netBuf, 2); /* align netBuf length */ #ifdef HAVE_SRANDOMDEV srandomdev(); @@ -712,7 +722,7 @@ rpc_srv_initServer(u_int regProgID, u_char regProcID, srandom((time(&tim) ^ getpid())); #endif - srv = malloc(sizeof(rpc_srv_t)); + srv = io_malloc(sizeof(rpc_srv_t)); if (!srv) { LOGERR; return NULL; @@ -734,7 +744,7 @@ rpc_srv_initServer(u_int regProgID, u_char regProcID, srv->srv_root = schedBegin(); if (!srv->srv_root) { rpc_SetErr(sched_GetErrno(), "%s", sched_GetError()); - free(srv); + io_free(srv); return NULL; } @@ -743,7 +753,7 @@ rpc_srv_initServer(u_int regProgID, u_char regProcID, if (!srv->srv_clients) { rpc_SetErr(io_GetErrno(), "%s", io_GetError()); schedEnd(&srv->srv_root); - free(srv); + io_free(srv); return NULL; } @@ -753,7 +763,7 @@ rpc_srv_initServer(u_int regProgID, u_char regProcID, LOGERR; io_arrayDestroy(&srv->srv_clients); schedEnd(&srv->srv_root); - free(srv); + io_free(srv); return NULL; } if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) { @@ -782,7 +792,7 @@ err: /* error condition */ close(srv->srv_server.cli_sock); io_arrayDestroy(&srv->srv_clients); schedEnd(&srv->srv_root); - free(srv); + io_free(srv); return NULL; } @@ -805,7 +815,7 @@ rpc_srv_endServer(rpc_srv_t ** __restrict psrv) (*psrv)->srv_kill = 1; sleep(RPC_SCHED_POLLING); - free(*psrv); + io_free(*psrv); *psrv = NULL; } @@ -828,13 +838,12 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) return -1; } - fcntl(srv->srv_server.cli_sock, F_SETFL, - fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK); - if (listen(srv->srv_server.cli_sock, io_arraySize(srv->srv_clients)) == -1) { LOGERR; return -1; - } + } else + fcntl(srv->srv_server.cli_sock, F_SETFL, + fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK); if (!schedRead(srv->srv_root, acceptClients, srv, srv->srv_server.cli_sock, NULL, 0)) { rpc_SetErr(sched_GetErrno(), "%s", sched_GetError()); @@ -853,6 +862,7 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) close(c->cli_sock); schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL); + io_freeVars(&RPC_RETVARS(c)); AIT_FREE_VAL(&c->cli_buf); } io_arrayDel(srv->srv_clients, i, 42); @@ -865,9 +875,8 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) TAILQ_FOREACH_SAFE(f, &srv->srv_funcs, func_node, tmp) { TAILQ_REMOVE(&srv->srv_funcs, f, func_node); - io_freeVars(&f->func_vars); AIT_FREE_VAL(&f->func_name); - free(f); + io_free(f); } schedEnd(&srv->srv_root); @@ -878,22 +887,23 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) /* * rpc_srv_execCall() Execute registered call from RPC server * - * @call = Register RPC call + * @cli = RPC client * @rpc = IN RPC call structure + * @funcname = Execute RPC function * @args = IN RPC calling arguments from RPC client * return: -1 error, !=-1 ok */ int -rpc_srv_execCall(rpc_func_t * __restrict call, struct tagRPCCall * __restrict rpc, - array_t * __restrict args) +rpc_srv_execCall(rpc_cli_t * __restrict cli, struct tagRPCCall * __restrict rpc, + ait_val_t funcname, array_t * __restrict args) { rpc_callback_t func; - if (!call || !rpc || !call->func_parent || !AIT_ADDR(&call->func_name)) { + if (!cli || !rpc || !AIT_ADDR(&funcname)) { rpc_SetErr(EINVAL, "Invalid parameter can`t exec function"); return -1; } - func = AIT_GET_LIKE(&call->func_name, rpc_callback_t); - return func(call, rpc, args); + func = AIT_GET_LIKE(&funcname, rpc_callback_t); + return func(cli, rpc, args); }