/************************************************************************* * (C) 2010 AITNET ltd - Sofia/Bulgaria - * by Michael Pounov * * $Author: misho $ * $Id: srv.c,v 1.1.1.1.2.14 2010/07/07 09:29:42 misho Exp $ * *************************************************************************/ #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; struct tagRPCCall *rpc; struct tagRPCRet rrpc; 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 { FD_ZERO(&fds); FD_SET(c->cli_sock, &fds); ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL); if (ret == -1) { ret = -2; } memset(&rrpc, 0, sizeof rrpc); memset(buf, 0, BUFSIZ); if ((ret = recv(c->cli_sock, buf, BUFSIZ, 0)) == -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; } // RPC is OK! Go decapsulate variables ... if (rpc->call_argc) { v = (rpc_val_t*) (buf + sizeof(struct tagRPCCall)); // 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: v[i].val.buffer = data; data += v[i].val_len; break; case string: v[i].val.string = (int8_t*) data; data += v[i].val_len + 1; break; case blob: if (s->srv_blob.state == disable) { rpc_SetErr(ENOTSUP, "Error:: BLOB server is disabled\n"); ret = -5; goto makeReply; } default: break; } } } argc = 0; vals = NULL; 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 = -6; else argc = rpc_srv_getValsCall(f, &vals); makeReply: memcpy(&rrpc.ret_session, &rpc->call_session, sizeof rrpc.ret_session); 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; memset(buf, 0, BUFSIZ); memcpy(buf, &rrpc, (Limit = sizeof rrpc)); if (argc && vals) { v = (rpc_val_t*) (buf + sizeof rrpc); memcpy(v, vals, argc * sizeof(rpc_val_t)); Limit += 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 > BUFSIZ) { 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 + 1 > BUFSIZ) { 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 + 1); data += vals[i].val_len + 1; Limit += vals[i].val_len + 1; 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; } default: break; } RPC_FREE_VAL(&vals[i]); } } if ((ret = send(c->cli_sock, buf, Limit, 0)) == -1) { LOGERR; ret = -8; break; } if (ret != Limit) { rpc_SetErr(EBADMSG, "Error:: in send RPC request, should be send %d bytes, " "really is %d\n", Limit, ret); ret = -9; break; } } while (ret > -1); shutdown(c->cli_sock, SHUT_RDWR); close(c->cli_sock); memset(c, 0, sizeof(rpc_cli_t)); return (void*) ret; } static void * rpc_srv_dispatchVars(void *arg) { rpc_cli_t *c = arg; rpc_srv_t *s; rpc_blob_t *b; int cx, ret; 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; cx = -1; do { // check for disable service at this moment? if (s->srv_blob.state == disable) { ret = 0; break; } FD_ZERO(&fds); FD_SET(c->cli_sock, &fds); ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL); if (ret == -1) { ret = -2; } else cx++; memset(buf, 0, sizeof buf); if ((ret = recv(c->cli_sock, buf, sizeof buf, 0)) == -1) { LOGERR; ret = -3; break; } if (!ret || s->srv_blob.state == disable) { // receive EOF or disable service 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) || blob->hdr_seq != cx) { rpc_SetErr(EINVAL, "Error:: get invalid BLOB session in seq=%d...\n", blob->hdr_seq); 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; } 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_seq = ret; if ((ret = send(c->cli_sock, buf, sizeof buf, 0)) == -1) { LOGERR; ret = -8; break; } if (ret != sizeof buf) { rpc_SetErr(EBADMSG, "Error:: in send BLOB reply, should be send %d bytes, " "really is %d\n", sizeof buf, ret); ret = -9; break; } } while (ret > -1); shutdown(c->cli_sock, SHUT_RDWR); close(c->cli_sock); memset(c, 0, sizeof(rpc_cli_t)); return (void*) ret; } // ------------------------------------------------- /* * rpc_srv_initBLOBServer() Init & create BLOB Server * @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_in sin; struct sockaddr_in6 sin6; 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; if (srv->srv_server.cli_sa.sa_family == AF_INET) { memcpy(&sin, &srv->srv_server.cli_sa, sizeof sin); sin.sin_port = htons(Port); memcpy(&srv->srv_blob.server.cli_sa, &sin, sizeof(struct sockaddr)); } else { memcpy(&sin6, &srv->srv_server.cli_sa, sizeof sin6); sin6.sin6_port = htons(Port); memcpy(&srv->srv_blob.server.cli_sa, &sin6, sizeof(struct sockaddr)); } 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; } 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 = disable; 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); 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; } 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); 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 == disable) { 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 (!blob_Kill && !rpc_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; } } 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 socket type, AF_INET or AF_INET6 * @csHost = Host name or IP 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_in sin; struct sockaddr_in6 sin6; if (!concurentClients || !regProgID || (family != AF_INET && family != AF_INET6)) { rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n"); return NULL; } if (!Port) Port = RPC_DEFPORT; if (csHost) { host = gethostbyname2(csHost, family); if (!host) { rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno)); return NULL; } } switch (family) { case AF_INET: memset(&sin, 0, sizeof sin); sin.sin_len = sizeof sin; sin.sin_family = family; sin.sin_port = htons(Port); if (csHost) memcpy(&sin.sin_addr, host->h_addr, host->h_length); break; case AF_INET6: memset(&sin6, 0, sizeof sin6); sin6.sin6_len = sizeof sin6; sin6.sin6_family = family; sin6.sin6_port = htons(Port); if (csHost) memcpy(&sin6.sin6_addr, host->h_addr, host->h_length); 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; if (family == AF_INET) memcpy(&srv->srv_server.cli_sa, &sin, sizeof srv->srv_server.cli_sa); else memcpy(&srv->srv_server.cli_sa, &sin6, sizeof srv->srv_server.cli_sa); 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; } 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)); 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); pthread_mutex_init(&srv->srv_mtx, NULL); 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); 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(srv->srv_server.cli_sock); if (srv->srv_clients) { free(srv->srv_clients); srv->srv_clients = NULL; srv->srv_numcli = 0; } 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); 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 (!rpc_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; } } 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; }