/************************************************************************* * (C) 2010 AITNET ltd - Sofia/Bulgaria - * by Michael Pounov * * $Author: misho $ * $Id: srv.c,v 1.4 2011/08/29 22:37:06 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following 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 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. 3. All advertising materials mentioning features or use of this software must display the following acknowledgement: This product includes software developed by Michael Pounov ELWIX - Embedded LightWeight unIX and its contributors. 4. Neither the name of AITNET nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ #include "global.h" static void * rpc_srv_dispatchCall(void *arg) { rpc_cli_t *c = arg; rpc_srv_t *s; rpc_val_t *vals = NULL, *v = NULL; rpc_func_t *f = NULL; struct tagRPCCall *rpc; struct tagRPCRet *rrpc; rpc_sess_t ses = { 0 }; fd_set fds; u_char buf[BUFSIZ], *data; int ret, argc = 0, Limit = 0; register int i; if (!arg) { rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced RPC client ...\n"); return NULL; } else s = c->cli_parent; do { v = NULL; FD_ZERO(&fds); FD_SET(c->cli_sock, &fds); ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL); if (ret == -1) { if (errno == EINTR && s->srv_kill != kill) continue; LOGERR; ret = -2; break; } memset(buf, 0, sizeof buf); ret = recv(c->cli_sock, buf, sizeof buf, 0); if (ret == -1) { LOGERR; ret = -3; break; } if (!ret) { /* receive EOF */ ret = 0; break; } if (ret < sizeof(struct tagRPCCall)) { rpc_SetErr(EMSGSIZE, "Error:: too short RPC packet ...\n"); ret = -4; break; } else rpc = (struct tagRPCCall*) buf; /* check RPC packet session info */ if (memcmp(&rpc->call_session, &s->srv_session, sizeof rpc->call_session)) { rpc_SetErr(EINVAL, "Error:: get invalid RPC session ...\n"); ret = -5; goto makeReply; } else Limit = sizeof(struct tagRPCCall); /* RPC is OK! Go decapsulate variables ... */ if (rpc->call_argc) { v = (rpc_val_t*) (buf + Limit); /* check RPC packet length */ if (rpc->call_argc * sizeof(rpc_val_t) > sizeof buf - Limit) { rpc_SetErr(EMSGSIZE, "Error:: too long RPC packet ...\n"); ret = -5; goto makeReply; } else Limit += rpc->call_argc * sizeof(rpc_val_t); /* RPC received variables types OK! */ data = (u_char*) v + rpc->call_argc * sizeof(rpc_val_t); for (i = 0; i < rpc->call_argc; i++) { switch (v[i].val_type) { case buffer: if (v[i].val_len > sizeof buf - Limit) { rpc_SetErr(EMSGSIZE, "Error:: too long RPC packet ...\n"); ret = -5; goto makeReply; } else Limit += v[i].val_len; v[i].val.buffer = data; data += v[i].val_len; break; case string: if (v[i].val_len > sizeof buf - Limit) { rpc_SetErr(EMSGSIZE, "Error:: too long RPC packet ...\n"); ret = -5; goto makeReply; } else Limit += v[i].val_len; v[i].val.string = (int8_t*) data; data += v[i].val_len; break; case blob: if (s->srv_blob.state == disable) { rpc_SetErr(ENOTSUP, "Error:: BLOB server is disabled\n"); ret = -5; goto makeReply; } if (s->srv_blob.state == kill) { rpc_SetErr(ENOTSUP, "Error:: BLOB server is gone.\n"); ret = -5; goto makeReply; } default: break; } } } /* execute call */ argc = 0; vals = NULL; memcpy(&ses, &rpc->call_session, sizeof ses); if (!(f = rpc_srv_getCall(s, rpc->call_tag, rpc->call_hash))) { rpc_SetErr(EINVAL, "Error:: call not found into RPC server ...\n"); ret = -6; } else if ((ret = rpc_srv_execCall(f, rpc, v)) == -1) ret = -9; else argc = rpc_srv_getVars(f, &vals); makeReply: /* made reply */ memset(buf, 0, sizeof buf); rrpc = (struct tagRPCRet*) buf; Limit = sizeof(struct tagRPCRet); memcpy(&rrpc->ret_session, &ses, sizeof(rpc_sess_t)); rrpc->ret_tag = rpc->call_tag; rrpc->ret_hash = rpc->call_hash; rrpc->ret_errno = rpc_Errno; rrpc->ret_retcode = ret; rrpc->ret_argc = argc; if (argc && vals) { v = (rpc_val_t*) (buf + Limit); if (argc * sizeof(rpc_val_t) > sizeof buf - Limit) { for (i = 0; i < argc; i++) RPC_FREE_VAL(&vals[i]); rpc_srv_freeVars(f); vals = NULL; argc = 0; ret = -7; rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet values (-7) ...\n"); goto makeReply; } else Limit += argc * sizeof(rpc_val_t); memcpy(v, vals, argc * sizeof(rpc_val_t)); data = (u_char*) v + argc * sizeof(rpc_val_t); for (ret = i = 0; i < argc; i++) { switch (vals[i].val_type) { case buffer: if (ret || Limit + vals[i].val_len > sizeof buf) { rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n"); rrpc->ret_retcode = ret = -7; rrpc->ret_argc = 0; break; } memcpy(data, vals[i].val.buffer, vals[i].val_len); data += vals[i].val_len; Limit += vals[i].val_len; break; case string: if (ret || Limit + vals[i].val_len > sizeof buf) { rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n"); rrpc->ret_retcode = ret = -7; rrpc->ret_argc = 0; break; } memcpy(data, vals[i].val.string, vals[i].val_len); data += vals[i].val_len; Limit += vals[i].val_len; break; case blob: if (s->srv_blob.state == disable) { rpc_SetErr(ENOTSUP, "Error:: BLOB server is disabled\n"); rrpc->ret_retcode = ret = -5; rrpc->ret_argc = 0; break; } if (s->srv_blob.state == kill) { rpc_SetErr(ENOTSUP, "Error:: BLOB server is gone.\n"); rrpc->ret_retcode = ret = -5; rrpc->ret_argc = 0; break; } default: break; } RPC_FREE_VAL(&vals[i]); } rpc_srv_freeVars(f); vals = NULL; argc = 0; } ret = send(c->cli_sock, buf, Limit, 0); if (ret == -1) { LOGERR; ret = -8; break; } if (ret != Limit) { rpc_SetErr(ECANCELED, "Error:: in send RPC request, should be send %d bytes, " "really is %d\n", Limit, ret); ret = -9; break; } } while (ret > -1 || s->srv_kill != kill); shutdown(c->cli_sock, SHUT_RDWR); close(c->cli_sock); memset(c, 0, sizeof(rpc_cli_t)); return (void*) (long)ret; } static void * rpc_srv_dispatchVars(void *arg) { rpc_cli_t *c = arg; rpc_srv_t *s; rpc_blob_t *b; int ret = 0; fd_set fds; u_char buf[sizeof(struct tagBLOBHdr)]; struct tagBLOBHdr *blob; if (!arg) { rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced BLOB client ...\n"); return NULL; } else s = c->cli_parent; do { /* check for disable service at this moment? */ if (s->srv_blob.state == disable && s->srv_kill != kill) { usleep(100000); #ifdef HAVE_PTHREAD_YIELD pthread_yield(); #endif continue; } FD_ZERO(&fds); FD_SET(c->cli_sock, &fds); ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL); if (ret == -1) { if (errno == EINTR && s->srv_kill != kill && s->srv_blob.state != kill) continue; LOGERR; ret = -2; break; } memset(buf, 0, sizeof buf); ret = recv(c->cli_sock, buf, sizeof buf, 0); if (ret == -1) { LOGERR; ret = -3; break; } /* receive EOF, disable or kill service */ if (!ret || s->srv_blob.state == kill || s->srv_kill == kill) { ret = 0; break; } if (ret < sizeof(struct tagBLOBHdr)) { rpc_SetErr(EMSGSIZE, "Error:: too short BLOB packet ...\n"); ret = -4; break; } else blob = (struct tagBLOBHdr*) buf; /* check BLOB packet session info */ if (memcmp(&blob->hdr_session, &s->srv_session, sizeof blob->hdr_session)) { rpc_SetErr(EINVAL, "Error:: get invalid BLOB session ...\n"); ret = -5; goto makeReply; } /* Go to proceed packet ... */ switch (blob->hdr_cmd) { case get: if (!(b = rpc_srv_getBLOB(s, blob->hdr_var))) { rpc_SetErr(EINVAL, "Error:: var (%x) not found into BLOB server ...\n", blob->hdr_var); ret = -6; break; } else blob->hdr_len = b->blob_len; if (rpc_srv_blobMap(s, b) != -1) { ret = rpc_srv_sendBLOB(c, b); rpc_srv_blobUnmap(b); } else ret = -7; break; case set: if ((b = rpc_srv_registerBLOB(s, blob->hdr_len))) { /* set new BLOB variable for reply :) */ blob->hdr_var = b->blob_var; ret = rpc_srv_recvBLOB(c, b); rpc_srv_blobUnmap(b); } else ret = -7; break; case unset: ret = rpc_srv_unregisterBLOB(s, blob->hdr_var); if (ret == -1) ret = -7; break; default: rpc_SetErr(EINVAL, "Error:: unsupported BLOB command (%d)...\n", blob->hdr_cmd); ret = -7; } makeReply: /* Replay to client! */ blob->hdr_cmd = ret < 0 ? error : ok; blob->hdr_ret = ret; ret = send(c->cli_sock, buf, sizeof buf, 0); if (ret == -1) { LOGERR; ret = -8; break; } if (ret != sizeof buf) { rpc_SetErr(ECANCELED, "Error:: in send BLOB reply, should be send %d bytes, " "really is %d\n", sizeof buf, ret); ret = -9; break; } } while (ret > -1 || s->srv_kill != kill); shutdown(c->cli_sock, SHUT_RDWR); close(c->cli_sock); memset(c, 0, sizeof(rpc_cli_t)); return (void*) (long)ret; } // ------------------------------------------------- /* * rpc_srv_initBLOBServer() Init & create BLOB Server * @srv = RPC server instance * @Port = Port for bind server, if Port == 0 default port is selected * @diskDir = Disk place for BLOB file objects * return: -1 == error or 0 bind and created BLOB server instance */ int rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir) { int n = 1; struct sockaddr sa; struct sockaddr_in *sin = (struct sockaddr_in*) &sa; struct sockaddr_in6 *sin6 = (struct sockaddr_in6*) &sa; struct sockaddr_un *sun = (struct sockaddr_un*) &sa; if (!srv) { rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init BLOB server ...\n"); return -1; } if (srv->srv_blob.state) { rpc_SetErr(EPERM, "Warning:: Already started BLOB server!\n"); 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.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_family) { case AF_INET: sin->sin_port = htons(Port); memcpy(&srv->srv_blob.server.cli_sa, sin, sizeof(struct sockaddr)); break; case AF_INET6: sin6->sin6_port = htons(Port); memcpy(&srv->srv_blob.server.cli_sa, sin6, sizeof(struct sockaddr)); break; case AF_LOCAL: strlcat(sun->sun_path, ".blob", sizeof sun->sun_path); memcpy(&srv->srv_blob.server.cli_sa, sun, sizeof(struct sockaddr)); break; default: return -1; } /* create BLOB server socket */ srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa_family, SOCK_STREAM, 0); if (srv->srv_blob.server.cli_sock == -1) { LOGERR; 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); return -1; } if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa, sizeof srv->srv_blob.server.cli_sa) == -1) { LOGERR; close(srv->srv_blob.server.cli_sock); return -1; } /* allocate pool for concurent clients */ srv->srv_blob.clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t)); if (!srv->srv_blob.clients) { LOGERR; close(srv->srv_blob.server.cli_sock); return -1; } else memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t)); pthread_mutex_init(&srv->srv_blob.mtx, NULL); pthread_mutex_lock(&srv->srv_mtx); rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0); rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 0); rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 0); rpc_srv_registerCall(srv, NULL, CALL_BLOBSTATE, 1); pthread_mutex_unlock(&srv->srv_mtx); srv->srv_blob.state = enable; /* enable BLOB */ return 0; } /* * rpc_srv_endBLOBServer() Destroy BLOB server, close all opened sockets and free resources * @srv = RPC Server instance * return: none */ void rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) { rpc_cli_t *c; register int i; rpc_blob_t *f; if (!srv) { rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n"); return; } else srv->srv_blob.state = kill; rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN); rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS); rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS); rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE); /* 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_family) shutdown(c->cli_sock, SHUT_RDWR); close(srv->srv_blob.server.cli_sock); if (srv->srv_blob.clients) { free(srv->srv_blob.clients); srv->srv_blob.clients = NULL; } /* detach blobs */ pthread_mutex_lock(&srv->srv_blob.mtx); while ((f = srv->srv_blob.blobs)) { srv->srv_blob.blobs = f->blob_next; rpc_srv_blobFree(srv, f); free(f); } pthread_mutex_unlock(&srv->srv_blob.mtx); while (pthread_mutex_trylock(&srv->srv_blob.mtx) == EBUSY); pthread_mutex_destroy(&srv->srv_blob.mtx); } /* * rpc_srv_execBLOBServer() Execute Main BLOB server loop and wait for clients requests * @srv = RPC Server instance * return: -1 error or 0 ok, infinite loop ... */ int rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv) { socklen_t salen = sizeof(struct sockaddr); register int i; rpc_cli_t *c; fd_set fds; int ret; struct timeval tv = { DEF_RPC_TIMEOUT, 0 }; if (!srv || srv->srv_blob.state == kill) { rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n"); return -1; } if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) { LOGERR; return -1; } 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_family) break; if (i >= srv->srv_numcli) { usleep(1000000); continue; } FD_ZERO(&fds); FD_SET(srv->srv_blob.server.cli_sock, &fds); ret = select(srv->srv_blob.server.cli_sock + 1, &fds, NULL, NULL, &tv); if (ret == -1) { LOGERR; ret = 1; break; } if (!ret) continue; c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa, &salen); if (c->cli_sock == -1) { LOGERR; continue; } else c->cli_parent = srv; if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchVars, c)) { LOGERR; continue; } else pthread_detach(c->cli_tid); } srv->srv_blob.state = disable; return 0; } /* * rpc_srv_initServer() Init & create RPC Server * @regProgID = ProgramID for authentication & recognition * @regProcID = ProcessID for authentication & recognition * @concurentClients = Concurent clients at same time to this server * @family = Family type, AF_INET, AF_INET6 or AF_LOCAL * @csHost = Host name or address for bind server, if NULL any address * @Port = Port for bind server, if Port == 0 default port is selected * return: NULL == error or !=NULL bind and created RPC server instance */ rpc_srv_t * rpc_srv_initServer(u_int regProgID, u_int regProcID, int concurentClients, u_short family, const char *csHost, u_short Port) { rpc_srv_t *srv = NULL; int n = 1; struct hostent *host = NULL; struct sockaddr sa; struct sockaddr_in *sin = (struct sockaddr_in*) &sa; struct sockaddr_in6 *sin6 = (struct sockaddr_in6*) &sa; struct sockaddr_un *sun = (struct sockaddr_un*) &sa; if (!concurentClients || !regProgID || (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) { rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n"); return NULL; } if (!Port) Port = RPC_DEFPORT; if (csHost && family != AF_LOCAL) { host = gethostbyname2(csHost, family); if (!host) { rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno)); return NULL; } } memset(&sa, 0, sizeof sa); sa.sa_family = family; switch (family) { case AF_INET: sin->sin_len = sizeof(struct sockaddr_in); sin->sin_port = htons(Port); if (csHost) memcpy(&sin->sin_addr, host->h_addr, host->h_length); break; case AF_INET6: sin6->sin6_len = sizeof(struct sockaddr_in6); sin6->sin6_port = htons(Port); if (csHost) memcpy(&sin6->sin6_addr, host->h_addr, host->h_length); break; case AF_LOCAL: sun->sun_len = sizeof(struct sockaddr_un); if (csHost) strlcpy(sun->sun_path, csHost, sizeof sun->sun_path); break; default: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t start RPC server ...\n"); return NULL; } srv = malloc(sizeof(rpc_srv_t)); if (!srv) { LOGERR; return NULL; } else memset(srv, 0, sizeof(rpc_srv_t)); srv->srv_numcli = concurentClients; srv->srv_session.sess_version = RPC_VERSION; srv->srv_session.sess_program = regProgID; srv->srv_session.sess_process = regProcID; srv->srv_server.cli_tid = pthread_self(); srv->srv_server.cli_parent = srv; switch (family) { case AF_INET: memcpy(&srv->srv_server.cli_sa, sin, sizeof srv->srv_server.cli_sa); break; case AF_INET6: memcpy(&srv->srv_server.cli_sa, sin6, sizeof srv->srv_server.cli_sa); break; case AF_LOCAL: memcpy(&srv->srv_server.cli_sa, sun, sizeof srv->srv_server.cli_sa); unlink(sun->sun_path); break; } /* create server socket */ srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0); if (srv->srv_server.cli_sock == -1) { LOGERR; 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); free(srv); return NULL; } if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa, sizeof srv->srv_server.cli_sa) == -1) { LOGERR; close(srv->srv_server.cli_sock); free(srv); return NULL; } /* allocate pool for concurent clients */ srv->srv_clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t)); if (!srv->srv_clients) { LOGERR; close(srv->srv_server.cli_sock); free(srv); return NULL; } else memset(srv->srv_clients, 0, srv->srv_numcli * sizeof(rpc_cli_t)); pthread_mutex_init(&srv->srv_mtx, NULL); rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0); rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 0); rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 0); rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 0); return srv; } /* * rpc_srv_endServer() Destroy RPC server, close all opened sockets and free resources * @srv = RPC Server instance * return: none */ void rpc_srv_endServer(rpc_srv_t * __restrict srv) { rpc_cli_t *c; register int i; rpc_func_t *f; if (!srv) { rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n"); return; } rpc_srv_endBLOBServer(srv); /* close all clients connections & server socket */ for (i = 0, c = srv->srv_clients; i < srv->srv_numcli && c; i++, c++) if (c->cli_sa.sa_family) { shutdown(c->cli_sock, SHUT_RDWR); close(c->cli_sock); } close(srv->srv_server.cli_sock); if (srv->srv_clients) { free(srv->srv_clients); srv->srv_clients = NULL; srv->srv_numcli = 0; } /* detach exported calls */ pthread_mutex_lock(&srv->srv_mtx); while ((f = srv->srv_funcs)) { srv->srv_funcs = f->func_next; free(f); } pthread_mutex_unlock(&srv->srv_mtx); while (pthread_mutex_trylock(&srv->srv_mtx) == EBUSY); pthread_mutex_destroy(&srv->srv_mtx); free(srv); srv = NULL; } /* * rpc_srv_execServer() Execute Main server loop and wait for clients requests * @srv = RPC Server instance * return: -1 error or 0 ok, infinite loop ... */ int rpc_srv_execServer(rpc_srv_t * __restrict srv) { socklen_t salen = sizeof(struct sockaddr); register int i; rpc_cli_t *c; fd_set fds; int ret; struct timeval tv = { DEF_RPC_TIMEOUT, 0 }; if (!srv) { rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n"); return -1; } if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) { LOGERR; return -1; } while (srv->srv_kill != kill) { for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++) if (!c->cli_sa.sa_family) break; if (i >= srv->srv_numcli) { usleep(1000000); continue; } FD_ZERO(&fds); FD_SET(srv->srv_server.cli_sock, &fds); ret = select(srv->srv_server.cli_sock + 1, &fds, NULL, NULL, &tv); if (ret == -1) { LOGERR; ret = 1; break; } if (!ret) continue; c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa, &salen); if (c->cli_sock == -1) { LOGERR; continue; } else c->cli_parent = srv; if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchCall, c)) { LOGERR; continue; } else pthread_detach(c->cli_tid); } return 0; } // --------------------------------------------------------- /* * rpc_srv_execCall() Execute registered call from RPC server * @call = Register RPC call * @rpc = IN RPC call structure * @args = IN RPC call array of rpc values * return: -1 error, !=-1 ok */ int rpc_srv_execCall(rpc_func_t * __restrict call, struct tagRPCCall * __restrict rpc, rpc_val_t * __restrict args) { void *dl; rpc_callback_t func; int ret; if (!call || !rpc || !call->func_parent) { rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t exec call from RPC server ...\n"); return -1; } dl = dlopen((char*) (*call->func_file ? call->func_file : NULL), RTLD_NOW); if (!dl) { rpc_SetErr(ENOENT, "Error:: Can`t attach module %s!\n", dlerror()); return -1; } func = dlsym(dl, (char*) call->func_name); if (func) ret = func(call, rpc->call_argc, args); else { rpc_SetErr(ENOEXEC, "Error:: Can`t find function %s!\n", dlerror()); ret = -1; } dlclose(dl); return ret; }