--- libaitrpc/src/srv.c 2011/11/03 15:32:21 1.6 +++ libaitrpc/src/srv.c 2012/03/12 14:53:07 1.6.2.1 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.6 2011/11/03 15:32:21 misho Exp $ +* $Id: srv.c,v 1.6.2.1 2012/03/12 14:53:07 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 @@ -348,11 +348,11 @@ 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; } @@ -361,7 +361,7 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s 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; @@ -378,6 +378,7 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path); break; default: + free(srv->srv_blob.dir); return -1; } memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa); @@ -386,28 +387,33 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s 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; } @@ -416,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)); @@ -444,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; @@ -454,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) @@ -492,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) @@ -534,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; } @@ -619,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; @@ -626,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; } @@ -643,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; } @@ -656,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; } @@ -665,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 @@ -697,6 +729,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++)