--- libaitrpc/src/srv.c 2010/06/18 01:48:06 1.1.1.1 +++ libaitrpc/src/srv.c 2010/06/23 16:23:31 1.1.1.1.2.6 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.1.1.1 2010/06/18 01:48:06 misho Exp $ +* $Id: srv.c,v 1.1.1.1.2.6 2010/06/23 16:23:31 misho Exp $ * *************************************************************************/ #include "global.h" @@ -12,7 +12,7 @@ static void * rpc_srv_dispatchCall(void *arg) { - rpc_cli_t *c = arg; + rpc_cli_t cli, *c = arg; rpc_srv_t *s; rpc_val_t *vals, *v = NULL; rpc_func_t *f; @@ -74,9 +74,9 @@ rpc_srv_dispatchCall(void *arg) v[i].val.string = (int8_t*) data; data += v[i].val_len + 1; break; - case array: - v[i].val.array = (int8_t**) data; - data += v[i].val_len; + case blob: + rpc_srv_getBLOBVar(&v[i], data); + data += sizeof(rpc_cli_t); break; default: break; @@ -135,17 +135,19 @@ rpc_srv_dispatchCall(void *arg) data += vals[i].val_len + 1; Limit += vals[i].val_len + 1; break; - case array: - if (ret || Limit + vals[i].val_len > BUFSIZ) { + case blob: + if (ret || Limit + sizeof(rpc_cli_t) > 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.array, vals[i].val_len); - data += vals[i].val_len; - Limit += vals[i].val_len; + memcpy(data, &cli, sizeof(rpc_cli_t)); + data += sizeof(rpc_cli_t); + Limit += sizeof(rpc_cli_t); + + rpc_srv_setBLOBVar(&vals[i], &cli); break; default: break; @@ -174,9 +176,277 @@ rpc_srv_dispatchCall(void *arg) 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 { + cx++; + + 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(buf, 0, sizeof buf); + if ((ret = recv(c->cli_sock, buf, sizeof buf, 0)) == -1) { + LOGERR; + ret = -3; + break; + } + if (!ret) { // receive EOF + 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; + break; + } + // Go to decapsulate packet ... + 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; + } + switch (blob->hdr_cmd) { + case get: + if (rpc_srv_blobMap(s, b) != -1) { + ret = rpc_srv_sendBLOB(c, b); + rpc_srv_blobUnmap(b); + } else + ret = -7; + break; + case set: + ret = rpc_srv_recvBLOB(c, b); + if (ret == -1) + ret = -7; + break; + case unset: + ret = rpc_srv_deleteBLOB(c, b); + if (ret == -1) + ret = -7; + break; + default: + rpc_SetErr(EINVAL, "Error:: unsupported BLOB command (%d)...\n", + blob->hdr_cmd); + ret -7; + } + + // Replay to client! + } 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); + 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; + + 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); + + pthread_mutex_lock(&srv->srv_blob.mtx); + while ((f = srv->srv_blob.blobs)) { + srv->srv_blob.blobs = f->blob_next; + 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) { + 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 (!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 (c && c->cli_sa.sa_family && c->cli_parent) { + 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; + } + } + + return 0; +} + + +/* * rpc_srv_initServer() Init & create RPC Server * @regProgID = ProgramID for authentication & recognition * @regProcID = ProcessID for authentication & recognition @@ -243,6 +513,7 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i 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); @@ -276,9 +547,12 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i } 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; } @@ -299,10 +573,7 @@ rpc_srv_endServer(rpc_srv_t * __restrict srv) return; } - while ((f = srv->srv_funcs)) { - srv->srv_funcs = f->func_next; - free(f); - } + rpc_srv_endBLOBServer(srv); for (i = 0, c = srv->srv_clients; i < srv->srv_numcli && c; i++, c++) if (c->cli_sa.sa_family) @@ -314,6 +585,15 @@ rpc_srv_endServer(rpc_srv_t * __restrict srv) 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; } @@ -329,6 +609,9 @@ 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"); @@ -340,7 +623,7 @@ rpc_srv_execServer(rpc_srv_t * __restrict srv) return -1; } - while (42) { + while (!rpc_Kill) { for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++) if (!c->cli_sa.sa_family) break; @@ -348,9 +631,20 @@ rpc_srv_execServer(rpc_srv_t * __restrict srv) 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) { - printf("%s(%d): #%d - %s\n", __func__, __LINE__, errno, strerror(errno)); LOGERR; continue; } else @@ -362,272 +656,7 @@ rpc_srv_execServer(rpc_srv_t * __restrict srv) } } - /* not reached !!! */ return 0; -} - -// --------------------------------------------------------- - -/* - * rpc_srv_freeValsCall() Free return variables for RPC call - * @call = RPC function call - * return: none - */ -inline void -rpc_srv_freeValsCall(rpc_func_t * __restrict call) -{ - rpc_srv_declValsCall(call, 0); -} - -/* - * rpc_srv_declValsCall() Declare return variables for RPC call - * @call = RPC function call - * @return_vals = Number of return variables - * return: -1 error, !=-1 ok - */ -inline int -rpc_srv_declValsCall(rpc_func_t * __restrict call, int return_vals) -{ - void *ptr; - - if (!call || return_vals < 0) { - rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t declare return variables for RPC call...\n"); - return -1; - } else - call->func_args = return_vals; - - if (!return_vals) { - if (call->func_vals) { - free(call->func_vals); - call->func_vals = NULL; - } - } else { - ptr = realloc(call->func_vals, return_vals * sizeof(rpc_val_t)); - if (!ptr) { - LOGERR; - return -1; - } else - call->func_vals = ptr; - } - - return call->func_args; -} - -/* - * rpc_srv_delValsCall() Clean values from return variables of RPC call - * @call = RPC function call - * return: -1 error, !=-1 Returned number of cleaned RPC variables - */ -inline int -rpc_srv_delValsCall(rpc_func_t * __restrict call) -{ - if (!call) { - rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t delete return variables ...\n"); - return -1; - } - - memset(call->func_vals, 0, call->func_args * sizeof(rpc_val_t)); - return call->func_args; -} - -/* - * rpc_srv_copyValsCall() Copy return variables for RPC call to new variable - * @call = RPC function call - * @newvals = New allocated variables array, must be free after use - * return: -1 error, !=-1 Returned number of copied RPC variables - */ -inline int -rpc_srv_copyValsCall(rpc_func_t * __restrict call, rpc_val_t ** __restrict newvals) -{ - if (!call || !newvals) { - rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t copy return variables to new array\n"); - return -1; - } - - *newvals = calloc(call->func_args, sizeof(rpc_val_t)); - if (!*newvals) { - LOGERR; - return -1; - } else - memcpy(*newvals, call->func_vals, call->func_args * sizeof(rpc_val_t)); - - return call->func_args; -} - -/* - * rpc_srv_getValsCall() Get return variables for RPC call - * @call = RPC function call - * @vals = Returned variables, may be NULL - * return: -1 error, !=-1 Number of returned variables - */ -inline int -rpc_srv_getValsCall(rpc_func_t * __restrict call, rpc_val_t ** __restrict vals) -{ - if (!call) { - rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t get return variables ...\n"); - return -1; - } - - if (vals) - *vals = call->func_vals; - return call->func_args; -} - -// --------------------------------------------------------- - -/* - * rpc_srv_registerCall() Register call to RPC server - * @srv = RPC Server instance - * @csModule = Module name, if NULL self binary - * @csFunc = Function name - * @args = Number of function arguments - * return: -1 error or 0 register ok - */ -int -rpc_srv_registerCall(rpc_srv_t * __restrict srv, const char *csModule, const char *csFunc, u_char args) -{ - rpc_func_t *func; - u_char str[MAXPATHLEN + UCHAR_MAX + 1]; - - memset(str, 0, MAXPATHLEN + UCHAR_MAX + 1); - if (!srv || !csFunc) { - rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t register function to RPC server ...\n"); - return -1; - } - if (!(func = malloc(sizeof(rpc_func_t)))) { - LOGERR; - return -1; - } else { - memset(func, 0, sizeof(rpc_func_t)); - strlcpy((char*) func->func_name, csFunc, UCHAR_MAX + 1); - } - if (csModule) { - strlcpy((char*) func->func_file, csModule, MAXPATHLEN); - strlcpy((char*) str, csModule, MAXPATHLEN + UCHAR_MAX + 1); - } - strlcat((char*) str, "__", MAXPATHLEN + UCHAR_MAX + 1); - strlcat((char*) str, csFunc, MAXPATHLEN + UCHAR_MAX + 1); - - func->func_tag = crcFletcher16((u_short*) str, (MAXPATHLEN + UCHAR_MAX + 1) / 2); - func->func_hash = hash_fnv((char*) str, MAXPATHLEN + UCHAR_MAX + 1); - - if (rpc_srv_declValsCall(func, args) == -1) { - free(func); - return -1; - } - - func->func_next = srv->srv_funcs; - srv->srv_funcs = func; - return 0; -} - -/* - * rpc_srv_unregisterCall() Unregister call from RPC server - * @srv = RPC Server instance - * @csModule = Module name, if NULL self binary - * @csFunc = Function name - * return: -1 error, 0 not found call, 1 unregister ok - */ -int -rpc_srv_unregisterCall(rpc_srv_t * __restrict srv, const char *csModule, const char *csFunc) -{ - rpc_func_t func, *f, *curr; - u_char str[MAXPATHLEN + UCHAR_MAX + 1]; - - memset(&func, 0, sizeof(rpc_func_t)); - memset(str, 0, MAXPATHLEN + UCHAR_MAX + 1); - if (!srv || !csFunc) { - rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t unregister function from RPC server ...\n"); - return -1; - } else - strlcpy((char*) func.func_name, csFunc, UCHAR_MAX + 1); - if (csModule) { - strlcpy((char*) func.func_file, csModule, MAXPATHLEN); - strlcpy((char*) str, csModule, MAXPATHLEN + UCHAR_MAX + 1); - } - strlcat((char*) str, "__", MAXPATHLEN + UCHAR_MAX + 1); - strlcat((char*) str, csFunc, MAXPATHLEN + UCHAR_MAX + 1); - - func.func_tag = crcFletcher16((u_short*) str, (MAXPATHLEN + UCHAR_MAX + 1) / 2); - func.func_hash = hash_fnv((char*) str, MAXPATHLEN + UCHAR_MAX + 1); - - f = rpc_srv_getCall(srv, func.func_tag, func.func_hash); - if (!f) // not found element for unregister - return 0; - - if (srv->srv_funcs == f) { // if is 1st element - srv->srv_funcs = srv->srv_funcs->func_next; - - if (f->func_args && f->func_vals) - free(f->func_vals); - free(f); - } else { - for (curr = srv->srv_funcs; curr->func_next != f; curr = curr->func_next); - curr->func_next = curr->func_next->func_next; - - if (f->func_args && f->func_vals) - free(f->func_vals); - free(f); - } - - return 1; -} - -/* - * rpc_srv_getCall() Get registered call from RPC server - * @srv = RPC Server instance - * @tag = tag for function - * @hash = hash for function - * return: NULL not found call, !=NULL return call - */ -inline rpc_func_t * -rpc_srv_getCall(rpc_srv_t * __restrict srv, uint16_t tag, uint32_t hash) -{ - rpc_func_t *f; - - if (!srv) { - rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t get function from RPC server ...\n"); - return NULL; - } - - for (f = srv->srv_funcs; f; f = f->func_next) - if (f->func_tag == tag && f->func_hash == hash) - break; - - return f; -} - -/* - * rpc_srv_getFunc() Get registered call from RPC server by Name - * @srv = RPC Server instance - * @csModule = Module name, if NULL self binary - * @csFunc = Function name - * return: NULL not found call, !=NULL return call - */ -rpc_func_t * -rpc_srv_getFunc(rpc_srv_t * __restrict srv, const char *csModule, const char *csFunc) -{ - rpc_func_t func; - u_char str[MAXPATHLEN + UCHAR_MAX + 1]; - - memset(&func, 0, sizeof(rpc_func_t)); - memset(str, 0, MAXPATHLEN + UCHAR_MAX + 1); - if (!srv || !csFunc) { - rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t get function from RPC server ...\n"); - return NULL; - } else - strlcpy((char*) func.func_name, csFunc, UCHAR_MAX + 1); - if (csModule) { - strlcpy((char*) func.func_file, csModule, MAXPATHLEN); - strlcpy((char*) str, csModule, MAXPATHLEN + UCHAR_MAX + 1); - } - strlcat((char*) str, "__", MAXPATHLEN + UCHAR_MAX + 1); - strlcat((char*) str, csFunc, MAXPATHLEN + UCHAR_MAX + 1); - - func.func_tag = crcFletcher16((u_short*) str, (MAXPATHLEN + UCHAR_MAX + 1) / 2); - func.func_hash = hash_fnv((char*) str, MAXPATHLEN + UCHAR_MAX + 1); - - return rpc_srv_getCall(srv, func.func_tag, func.func_hash); } // ---------------------------------------------------------