--- libaitrpc/src/srv.c 2012/05/17 15:48:45 1.9.2.25 +++ libaitrpc/src/srv.c 2012/05/18 15:24:33 1.9.2.29 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.9.2.25 2012/05/17 15:48:45 misho Exp $ +* $Id: srv.c,v 1.9.2.29 2012/05/18 15:24:33 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -62,7 +62,9 @@ closeClient(sched_task_t *task) /* free buffer */ AIT_FREE_VAL(&c->cli_buf); - io_arrayDel(s->srv_clients, c->cli_id, 42); + io_arrayDel(s->srv_clients, c->cli_id, 0); + if (c) + free(c); return NULL; } @@ -87,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; @@ -151,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; @@ -160,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))); } } } @@ -227,7 +229,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)) { @@ -314,7 +316,9 @@ closeBLOBClient(sched_task_t *task) /* free buffer */ AIT_FREE_VAL(&c->cli_buf); - io_arrayDel(s->srv_blob.clients, c->cli_id, 42); + io_arrayDel(s->srv_blob.clients, c->cli_id, 0); + if (c) + free(c); return NULL; } @@ -592,46 +596,13 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s * @srv = RPC Server instance * return: none */ -void +inline void rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) { - rpc_cli_t *c; - register int i; - rpc_blob_t *b, *tmp; - if (!srv) return; - /* close all clients connections & server socket */ - for (i = 0; i < io_arraySize(srv->srv_blob.clients); i++) { - c = io_array(srv->srv_blob.clients, i, rpc_cli_t*); - if (c) { - shutdown(c->cli_sock, SHUT_RDWR); - close(c->cli_sock); - - schedCancelby(srv->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL); - AIT_FREE_VAL(&c->cli_buf); - } - io_arrayDel(srv->srv_blob.clients, i, 42); - } - io_arrayDestroy(&srv->srv_blob.clients); - srv->srv_blob.kill = 1; - if (srv->srv_blob.tid) - pthread_cancel(srv->srv_blob.tid); - - close(srv->srv_blob.server.cli_sock); - - /* detach blobs */ - TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) { - TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node); - - rpc_srv_blobFree(srv, b); - free(b); - } - - schedEnd(&srv->srv_blob.root); - AIT_FREE_VAL(&srv->srv_blob.dir); } /* @@ -643,6 +614,11 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) int rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv) { + rpc_cli_t *c; + register int i; + rpc_blob_t *b, *tmp; + struct timespec ts = { RPC_SCHED_POLLING, 0 }; + if (!srv || srv->srv_kill) { rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server"); return -1; @@ -662,8 +638,36 @@ rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv) return -1; } + schedPolling(srv->srv_blob.root, &ts, NULL); /* main rpc loop */ schedRun(srv->srv_blob.root, &srv->srv_blob.kill); + + /* close all clients connections & server socket */ + for (i = 0; i < io_arraySize(srv->srv_blob.clients); i++) { + c = io_array(srv->srv_blob.clients, i, rpc_cli_t*); + if (c) { + shutdown(c->cli_sock, SHUT_RDWR); + close(c->cli_sock); + + schedCancelby(srv->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL); + AIT_FREE_VAL(&c->cli_buf); + } + io_arrayDel(srv->srv_blob.clients, i, 42); + } + io_arrayDestroy(&srv->srv_blob.clients); + + close(srv->srv_blob.server.cli_sock); + + /* detach blobs */ + TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) { + TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node); + + rpc_srv_blobFree(srv, b); + free(b); + } + + schedEnd(&srv->srv_blob.root); + AIT_FREE_VAL(&srv->srv_blob.dir); return 0; } @@ -788,45 +792,19 @@ err: /* error condition */ * @psrv = RPC Server instance * return: none */ -void +inline void rpc_srv_endServer(rpc_srv_t ** __restrict psrv) { - rpc_cli_t *c; - register int i; - rpc_func_t *f, *tmp; - if (!psrv || !*psrv) return; + /* if send kill to blob server */ if (!(*psrv)->srv_blob.kill) rpc_srv_endBLOBServer(*psrv); - /* close all clients connections & server socket */ - for (i = 0; i < io_arraySize((*psrv)->srv_clients); i++) { - c = io_array((*psrv)->srv_clients, i, rpc_cli_t*); - if (c) { - shutdown(c->cli_sock, SHUT_RDWR); - close(c->cli_sock); + (*psrv)->srv_kill = 1; + sleep(RPC_SCHED_POLLING); - schedCancelby((*psrv)->srv_root, taskMAX, CRITERIA_ARG, c, NULL); - AIT_FREE_VAL(&c->cli_buf); - } - io_arrayDel((*psrv)->srv_clients, i, 42); - } - io_arrayDestroy(&(*psrv)->srv_clients); - - close((*psrv)->srv_server.cli_sock); - - /* detach exported calls */ - TAILQ_FOREACH_SAFE(f, &(*psrv)->srv_funcs, func_node, tmp) { - TAILQ_REMOVE(&(*psrv)->srv_funcs, f, func_node); - - io_freeVars(&f->func_vars); - AIT_FREE_VAL(&f->func_name); - free(f); - } - - schedEnd(&(*psrv)->srv_root); free(*psrv); *psrv = NULL; } @@ -840,6 +818,11 @@ rpc_srv_endServer(rpc_srv_t ** __restrict psrv) int rpc_srv_loopServer(rpc_srv_t * __restrict srv) { + rpc_cli_t *c; + register int i; + rpc_func_t *f, *tmp; + struct timespec ts = { RPC_SCHED_POLLING, 0 }; + if (!srv) { rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server"); return -1; @@ -858,8 +841,36 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) return -1; } + schedPolling(srv->srv_root, &ts, NULL); /* main rpc loop */ schedRun(srv->srv_root, &srv->srv_kill); + + /* close all clients connections & server socket */ + for (i = 0; i < io_arraySize(srv->srv_clients); i++) { + c = io_array(srv->srv_clients, i, rpc_cli_t*); + if (c) { + shutdown(c->cli_sock, SHUT_RDWR); + 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); + } + io_arrayDestroy(&srv->srv_clients); + + close(srv->srv_server.cli_sock); + + /* detach exported calls */ + TAILQ_FOREACH_SAFE(f, &srv->srv_funcs, func_node, tmp) { + TAILQ_REMOVE(&srv->srv_funcs, f, func_node); + + AIT_FREE_VAL(&f->func_name); + free(f); + } + + schedEnd(&srv->srv_root); return 0; } @@ -867,22 +878,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); }