--- libaitrpc/src/srv.c 2011/10/31 14:58:33 1.5.2.7 +++ libaitrpc/src/srv.c 2012/03/12 15:04:38 1.6.2.2 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.5.2.7 2011/10/31 14:58:33 misho Exp $ +* $Id: srv.c,v 1.6.2.2 2012/03/12 15:04:38 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, 2005, 2006, 2007, 2008, 2009, 2010, 2011 +Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -101,7 +101,10 @@ rpc_srv_dispatchCall(void *arg) if (ret < sizeof(struct tagRPCCall)) { rpc_SetErr(ERPCMISMATCH, "Error:: too short RPC packet ...\n"); ret = -4; - break; + if (s->srv_kill != kill) + continue; + else + break; } else rpc = (struct tagRPCCall*) buf; /* check RPC packet session info */ @@ -117,7 +120,7 @@ rpc_srv_dispatchCall(void *arg) /* RPC is OK! Go decapsulate variables ... */ if (ntohs(rpc->call_argc)) { - arr = io_buffer2vals(buf + Limit, s->srv_netbuf - Limit, + arr = io_buffer2vars(buf + Limit, s->srv_netbuf - Limit, ntohs(rpc->call_argc), 1); if (!arr) { ret = -5; @@ -160,8 +163,8 @@ makeReply: if (argc && arr) { /* Go Encapsulate variables ... */ - if ((i = io_vals2buffer(buf + Limit, s->srv_netbuf - Limit, arr)) == -1) { - rpc_srv_freeVals(f); + if ((i = io_vars2buffer(buf + Limit, s->srv_netbuf - Limit, arr)) == -1) { + io_clrVars(f->func_vars); argc = 0; ret = -7; rpc_SetErr(EBADRPC, "Error:: in prepare RPC packet values (-7) ...\n"); @@ -169,7 +172,7 @@ makeReply: } else { Limit += i; - rpc_srv_freeVals(f); + io_clrVars(f->func_vars); } } @@ -183,7 +186,10 @@ makeReply: rpc_SetErr(EPROCUNAVAIL, "Error:: in send RPC request, should be send %d bytes, " "really is %d\n", Limit, ret); ret = -9; - break; + if (s->srv_kill != kill) + continue; + else + break; } } while (ret > -1 || s->srv_kill != kill); @@ -249,7 +255,10 @@ rpc_srv_dispatchVars(void *arg) if (ret < sizeof(struct tagBLOBHdr)) { rpc_SetErr(ERPCMISMATCH, "Error:: too short BLOB packet ...\n"); ret = -4; - break; + if (s->srv_kill != kill && s->srv_blob.state != kill) + continue; + else + break; } else blob = (struct tagBLOBHdr*) buf; /* check BLOB packet session info */ @@ -310,7 +319,10 @@ makeReply: rpc_SetErr(EPROCUNAVAIL, "Error:: in send BLOB reply, should be send %d bytes, " "really is %d\n", sizeof buf, ret); ret = -9; - break; + if (s->srv_kill != kill && s->srv_blob.state != kill) + continue; + else + break; } } while (ret > -1 || s->srv_kill != kill); @@ -336,70 +348,72 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s io_sockaddr_t sa; if (!srv) { - rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init BLOB server ...\n"); + rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server"); return -1; } if (srv->srv_blob.state) { - rpc_SetErr(EPERM, "Warning:: Already started BLOB server!\n"); + rpc_SetErr(EPERM, "Already started BLOB server!"); return 0; } - if (!Port) - Port = RPC_DEFPORT + 1; memset(&srv->srv_blob, 0, sizeof srv->srv_blob); if (access(diskDir, R_OK | W_OK) == -1) { LOGERR; return -1; } else - strlcpy(srv->srv_blob.dir, diskDir, UCHAR_MAX + 1); + srv->srv_blob.dir = strdup(diskDir); srv->srv_blob.server.cli_tid = pthread_self(); srv->srv_blob.server.cli_parent = srv; memcpy(&sa, &srv->srv_server.cli_sa, sizeof sa); - switch (srv->srv_server.cli_sa.sa.sa_family) { + switch (sa.sa.sa_family) { case AF_INET: - sa.sin.sin_port = htons(Port); - memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa); + sa.sin.sin_port = htons(Port ? Port : ntohs(sa.sin.sin_port) + 1); break; case AF_INET6: - sa.sin6.sin6_port = htons(Port); - memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa); + sa.sin6.sin6_port = htons(Port ? Port : ntohs(sa.sin6.sin6_port) + 1); break; case AF_LOCAL: strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path); - memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa); break; default: + free(srv->srv_blob.dir); return -1; } + memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa); /* create BLOB server socket */ srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0); if (srv->srv_blob.server.cli_sock == -1) { LOGERR; + free(srv->srv_blob.dir); return -1; } if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) { LOGERR; close(srv->srv_blob.server.cli_sock); + free(srv->srv_blob.dir); return -1; } n = srv->srv_netbuf; if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) { LOGERR; close(srv->srv_blob.server.cli_sock); + free(srv->srv_blob.dir); return -1; } if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) { LOGERR; close(srv->srv_blob.server.cli_sock); + free(srv->srv_blob.dir); return -1; } if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa, srv->srv_blob.server.cli_sa.sa.sa_len) == -1) { LOGERR; close(srv->srv_blob.server.cli_sock); + free(srv->srv_blob.dir); return -1; } @@ -408,6 +422,7 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s if (!srv->srv_blob.clients) { LOGERR; close(srv->srv_blob.server.cli_sock); + free(srv->srv_blob.dir); return -1; } else memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t)); @@ -436,7 +451,7 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) rpc_blob_t *f; if (!srv) { - rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n"); + rpc_SetErr(EINVAL, "Can`t destroy server because parameter is null!"); return; } else srv->srv_blob.state = kill; @@ -446,6 +461,9 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS); rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE); + if (srv->srv_blob.dir) + free(srv->srv_blob.dir); + /* close all clients connections & server socket */ for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++) if (c->cli_sa.sa.sa_family) @@ -484,17 +502,24 @@ rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) fd_set fds; int ret; struct timeval tv = { DEF_RPC_TIMEOUT, 0 }; + pthread_attr_t attr; if (!srv || srv->srv_blob.state == kill) { rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n"); return -1; } + tv.tv_sec = srv->srv_session.sess_timeout; + if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) { LOGERR; return -1; } + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + + /* main BLOB loop */ while (srv->srv_blob.state != kill && srv->srv_kill != kill) { for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++) if (!c->cli_sa.sa.sa_family) @@ -526,15 +551,16 @@ rpc_srv_loopBLOB(rpc_srv_t * __restrict srv) } else c->cli_parent = srv; - if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchVars, c)) { + /* spawn dispatch thread for BLOB client */ + if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) { LOGERR; continue; - } else - pthread_detach(c->cli_tid); + } } srv->srv_blob.state = kill; + pthread_attr_destroy(&attr); return 0; } @@ -611,6 +637,7 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i srv->srv_netbuf = netBuf; srv->srv_numcli = concurentClients; srv->srv_session.sess_version = RPC_VERSION; + srv->srv_session.sess_timeout = DEF_RPC_TIMEOUT; srv->srv_session.sess_program = regProgID; srv->srv_session.sess_process = regProcID; @@ -618,16 +645,25 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i srv->srv_server.cli_parent = srv; memcpy(&srv->srv_server.cli_sa, &sa, sizeof sa); + /* init rpc scheduler */ + srv->srv_sched = schedBegin(); + if (!srv->srv_sched) { + rpc_SetErr(sched_GetErrno(), "%s", sched_GetError()); + return NULL; + } + /* create server socket */ srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0); if (srv->srv_server.cli_sock == -1) { LOGERR; + schedEnd(&srv->srv_sched); free(srv); return NULL; } if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) { LOGERR; close(srv->srv_server.cli_sock); + schedEnd(&srv->srv_sched); free(srv); return NULL; } @@ -635,12 +671,14 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) { LOGERR; close(srv->srv_server.cli_sock); + schedEnd(&srv->srv_sched); free(srv); return NULL; } if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) { LOGERR; close(srv->srv_server.cli_sock); + schedEnd(&srv->srv_sched); free(srv); return NULL; } @@ -648,6 +686,7 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i srv->srv_server.cli_sa.sa.sa_len) == -1) { LOGERR; close(srv->srv_server.cli_sock); + schedEnd(&srv->srv_sched); free(srv); return NULL; } @@ -657,6 +696,7 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i if (!srv->srv_clients) { LOGERR; close(srv->srv_server.cli_sock); + schedEnd(&srv->srv_sched); free(srv); return NULL; } else @@ -690,6 +730,9 @@ rpc_srv_endServer(rpc_srv_t ** __restrict psrv) rpc_srv_endBLOBServer(*psrv); + /* destroy rpc scheduler */ + schedEnd(&(*psrv)->srv_sched); + /* close all clients connections & server socket */ for (i = 0, c = (*psrv)->srv_clients; i < (*psrv)->srv_numcli && c; i++, c++) if (c->cli_sa.sa.sa_family) { @@ -708,7 +751,7 @@ rpc_srv_endServer(rpc_srv_t ** __restrict psrv) pthread_mutex_lock(&(*psrv)->srv_mtx); while ((f = (*psrv)->srv_funcs)) { (*psrv)->srv_funcs = f->func_next; - rpc_srv_destroyVars(f); + io_freeVars(&f->func_vars); free(f); } pthread_mutex_unlock(&(*psrv)->srv_mtx); @@ -734,6 +777,7 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) fd_set fds; int ret; struct timeval tv = { DEF_RPC_TIMEOUT, 0 }; + pthread_attr_t attr; if (!srv) { rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n"); @@ -748,6 +792,9 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) return -1; } + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + while (srv->srv_kill != kill) { for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++) if (!c->cli_sa.sa.sa_family) @@ -779,13 +826,13 @@ rpc_srv_loopServer(rpc_srv_t * __restrict srv) } else c->cli_parent = srv; - if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchCall, c)) { + if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) { LOGERR; continue; - } else - pthread_detach(c->cli_tid); + } } + pthread_attr_destroy(&attr); return 0; }