--- libaitrpc/src/srv.c 2010/06/23 07:50:01 1.1.1.1.2.3 +++ libaitrpc/src/srv.c 2010/06/23 11:33:38 1.1.1.1.2.4 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.1.1.1.2.3 2010/06/23 07:50:01 misho Exp $ +* $Id: srv.c,v 1.1.1.1.2.4 2010/06/23 11:33:38 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; @@ -75,8 +75,8 @@ rpc_srv_dispatchCall(void *arg) data += v[i].val_len + 1; break; case blob: - v[i].val.blob = (int8_t**) data; - data += v[i].val_len; + rpc_srv_getBLOB(&v[i], data); + data += sizeof(rpc_cli_t); break; default: break; @@ -136,16 +136,18 @@ rpc_srv_dispatchCall(void *arg) Limit += vals[i].val_len + 1; break; case blob: - if (ret || Limit + vals[i].val_len > BUFSIZ) { + 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.blob, 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_setBLOB(&vals[i], &cli); break; default: break; @@ -174,9 +176,191 @@ rpc_srv_dispatchCall(void *arg) return (void*) ret; } + +static void * +rpc_srv_dispatchVars(void *arg) +{ + rpc_cli_t *c = arg; +// rpc_srv_t *s; + int ret; + + return (void*) ret; +} + // ------------------------------------------------- /* + * rpc_srv_initBLOBServer() Init & create BLOB Server + * @Port = Port for bind server, if Port == 0 default port is selected + * return: -1 == error or 0 bind and created BLOB server instance + */ +int +rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port) +{ + 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); + 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 = 1; // 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 = 0; + + 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 @@ -282,7 +466,7 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 0); rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 0); - pthread_mutex_init(&rpc_mtx, NULL); + pthread_mutex_init(&srv->srv_mtx, NULL); return srv; } @@ -303,13 +487,8 @@ rpc_srv_endServer(rpc_srv_t * __restrict srv) return; } - pthread_mutex_destroy(&rpc_mtx); + rpc_srv_endBLOBServer(srv); - while ((f = srv->srv_funcs)) { - srv->srv_funcs = f->func_next; - free(f); - } - 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); @@ -320,6 +499,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; } @@ -535,8 +723,10 @@ rpc_srv_registerCall(rpc_srv_t * __restrict srv, const return -1; } + pthread_mutex_lock(&srv->srv_mtx); func->func_next = srv->srv_funcs; srv->srv_funcs = func; + pthread_mutex_unlock(&srv->srv_mtx); return 0; } @@ -574,6 +764,7 @@ rpc_srv_unregisterCall(rpc_srv_t * __restrict srv, con if (!f) // not found element for unregister return 0; + pthread_mutex_lock(&srv->srv_mtx); if (srv->srv_funcs == f) { // if is 1st element srv->srv_funcs = srv->srv_funcs->func_next; @@ -588,6 +779,7 @@ rpc_srv_unregisterCall(rpc_srv_t * __restrict srv, con free(f->func_vals); free(f); } + pthread_mutex_unlock(&srv->srv_mtx); return 1; }