--- libaitrpc/src/srv.c 2011/08/18 19:04:10 1.3.2.2 +++ libaitrpc/src/srv.c 2011/08/18 21:20:54 1.3.2.3 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.3.2.2 2011/08/18 19:04:10 misho Exp $ +* $Id: srv.c,v 1.3.2.3 2011/08/18 21:20:54 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -366,16 +366,20 @@ makeReply: /* * rpc_srv_initBLOBServer() Init & create BLOB Server + * @srv = RPC server instance + * @sockType = Socket type, like SOCK_STREAM, SOCK_DGRAM or SOCK_RAW * @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) +rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, int sockType, u_short Port, const char *diskDir) { int n = 1; - struct sockaddr_in sin; - struct sockaddr_in6 sin6; + 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"); @@ -397,17 +401,28 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s 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)); + + 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; } - srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa_family, SOCK_STREAM, 0); + /* create BLOB server socket */ + srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa_family, + sockType ? sockType : SOCK_STREAM, 0); if (srv->srv_blob.server.cli_sock == -1) { LOGERR; return -1; @@ -424,6 +439,7 @@ rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s 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; @@ -468,6 +484,7 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) 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); @@ -478,6 +495,7 @@ rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv) 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; @@ -510,10 +528,12 @@ rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv) return -1; } - if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) { - LOGERR; - return -1; - } + if (srv->srv_blob.server.cli_type == SOCK_STREAM || + srv->srv_blob.server.cli_type == SOCK_SEQPACKET) + 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++) @@ -535,7 +555,11 @@ rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv) if (!ret) continue; - c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa, &salen); + if (srv->srv_blob.server.cli_type == SOCK_STREAM || + srv->srv_blob.server.cli_type == SOCK_SEQPACKET) + c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa, &salen); + else + c->cli_sock = dup(srv->srv_blob.server.cli_sock); if (c->cli_sock == -1) { LOGERR; continue; @@ -545,7 +569,12 @@ rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv) if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchVars, c)) { LOGERR; continue; - } + } else + if (srv->srv_blob.server.cli_type == SOCK_STREAM || + srv->srv_blob.server.cli_type == SOCK_SEQPACKET) + pthread_detach(c->cli_tid); + else + pthread_join(c->cli_tid, NULL); } srv->srv_blob.state = disable; @@ -632,13 +661,13 @@ rpc_srv_initServer(u_int regProgID, u_int regProcID, i srv->srv_server.cli_parent = srv; switch (family) { case AF_INET: - memcpy(&srv->srv_server.cli_sa, &sin, sizeof srv->srv_server.cli_sa); + 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); + 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); + memcpy(&srv->srv_server.cli_sa, sun, sizeof srv->srv_server.cli_sa); unlink(sun->sun_path); break; } @@ -751,10 +780,11 @@ rpc_srv_execServer(rpc_srv_t * __restrict srv) return -1; } - if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) { - LOGERR; - return -1; - } + if (srv->srv_server.cli_type == SOCK_STREAM || srv->srv_server.cli_type == SOCK_SEQPACKET) + 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++) @@ -776,7 +806,10 @@ rpc_srv_execServer(rpc_srv_t * __restrict srv) if (!ret) continue; - c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa, &salen); + if (srv->srv_server.cli_type == SOCK_STREAM || srv->srv_server.cli_type == SOCK_SEQPACKET) + c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa, &salen); + else + c->cli_sock = dup(srv->srv_server.cli_sock); if (c->cli_sock == -1) { LOGERR; continue; @@ -786,6 +819,11 @@ rpc_srv_execServer(rpc_srv_t * __restrict srv) if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchCall, c)) { LOGERR; continue; + } else { + if (srv->srv_server.cli_type == SOCK_STREAM || srv->srv_server.cli_type == SOCK_SEQPACKET) + pthread_detach(c->cli_tid); + else + pthread_join(c->cli_tid, NULL); } }