File:  [ELWIX - Embedded LightWeight unIX -] / libaitrpc / src / srv.c
Revision 1.1.1.1.2.10: download - view: text, annotated - select for diffs - revision graph
Thu Jun 24 13:58:17 2010 UTC (14 years, 1 month ago) by misho
Branches: rpc1_0
Diff to: branchpoint 1.1.1.1: preferred, unified
handling blob ok

    1: /*************************************************************************
    2: * (C) 2010 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
    3: *  by Michael Pounov <misho@openbsd-bg.org>
    4: *
    5: * $Author: misho $
    6: * $Id: srv.c,v 1.1.1.1.2.10 2010/06/24 13:58:17 misho Exp $
    7: *
    8: *************************************************************************/
    9: #include "global.h"
   10: 
   11: 
   12: static void *
   13: rpc_srv_dispatchCall(void *arg)
   14: {
   15: 	rpc_cli_t *c = arg;
   16: 	rpc_srv_t *s;
   17: 	rpc_val_t *vals, *v = NULL;
   18: 	rpc_func_t *f;
   19: 	struct tagRPCCall *rpc;
   20: 	struct tagRPCRet rrpc;
   21: 	fd_set fds;
   22: 	u_char buf[BUFSIZ], *data;
   23: 	int ret, argc, Limit = 0;
   24: 	register int i;
   25: 
   26: 	if (!arg) {
   27: 		rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced RPC client ...\n");
   28: 		return NULL;
   29: 	} else
   30: 		s = c->cli_parent;
   31: 
   32: 	do {
   33: 		FD_ZERO(&fds);
   34: 		FD_SET(c->cli_sock, &fds);
   35: 		ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);
   36: 		if (ret == -1) {
   37: 			ret = -2;
   38: 		}
   39: 		memset(&rrpc, 0, sizeof rrpc);
   40: 		memset(buf, 0, BUFSIZ);
   41: 		if ((ret = recv(c->cli_sock, buf, BUFSIZ, 0)) == -1) {
   42: 			LOGERR;
   43: 			ret = -3;
   44: 			break;
   45: 		}
   46: 		if (!ret) {		// receive EOF
   47: 			ret = 0;
   48: 			break;
   49: 		}
   50: 		if (ret < sizeof(struct tagRPCCall)) {
   51: 			rpc_SetErr(EMSGSIZE, "Error:: too short RPC packet ...\n");
   52: 			ret = -4;
   53: 			break;
   54: 		} else
   55: 			rpc = (struct tagRPCCall*) buf;
   56: 		// check RPC packet session info
   57: 		if (memcmp(&rpc->call_session, &s->srv_session, sizeof rpc->call_session)) {
   58: 			rpc_SetErr(EINVAL, "Error:: get invalid RPC session ...\n");
   59: 			ret = -5;
   60: 			break;
   61: 		}
   62: 		// RPC is OK! Go decapsulate variables ...
   63: 		if (rpc->call_argc) {
   64: 			v = (rpc_val_t*) (buf + sizeof(struct tagRPCCall));
   65: 			// RPC received variables types OK!
   66: 			data = (u_char*) v + rpc->call_argc * sizeof(rpc_val_t);
   67: 			for (i = 0; i < rpc->call_argc; i++) {
   68: 				switch (v[i].val_type) {
   69: 					case buffer:
   70: 						v[i].val.buffer = data;
   71: 					       	data += v[i].val_len;
   72: 						break;
   73: 					case string:
   74: 						v[i].val.string = (int8_t*) data;
   75: 						data += v[i].val_len + 1;
   76: 						break;
   77: 					default:
   78: 						break;
   79: 				}
   80: 			}
   81: 		}
   82: 
   83: 		argc = 0;
   84: 		vals = NULL;
   85: 		if (!(f = rpc_srv_getCall(s, rpc->call_tag, rpc->call_hash))) {
   86: 			rpc_SetErr(EINVAL, "Error:: call not found into RPC server ...\n");
   87: 			ret = -6;
   88: 		} else
   89: 			if ((ret = rpc_srv_execCall(s, f, rpc, v)) == -1)
   90: 				ret = -6;
   91: 			else
   92: 				argc = rpc_srv_getValsCall(f, &vals);
   93: 
   94: 		memcpy(&rrpc.ret_session, &rpc->call_session, sizeof rrpc.ret_session);
   95: 		rrpc.ret_tag = rpc->call_tag;
   96: 		rrpc.ret_hash = rpc->call_hash;
   97: 		rrpc.ret_errno = rpc_Errno;
   98: 		rrpc.ret_retcode = ret;
   99: 		rrpc.ret_argc = argc;
  100: 
  101: 		memset(buf, 0, BUFSIZ);
  102: 		memcpy(buf, &rrpc, (Limit = sizeof rrpc));
  103: 		if (argc && vals) {
  104: 			v = (rpc_val_t*) (buf + sizeof rrpc);
  105: 			memcpy(v, vals, argc * sizeof(rpc_val_t));
  106: 			Limit += argc * sizeof(rpc_val_t);
  107: 			data = (u_char*) v + argc * sizeof(rpc_val_t);
  108: 			for (ret = i = 0; i < argc; i++) {
  109: 				switch (vals[i].val_type) {
  110: 					case buffer:
  111: 						if (ret || Limit + vals[i].val_len > BUFSIZ) {
  112: 							rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n");
  113: 							rrpc.ret_retcode = ret = -7;
  114: 							rrpc.ret_argc = 0;
  115: 							break;
  116: 						}
  117: 
  118: 						memcpy(data, vals[i].val.buffer, vals[i].val_len);
  119: 						data += vals[i].val_len;
  120: 						Limit += vals[i].val_len;
  121: 						break;
  122: 					case string:
  123: 						if (ret || Limit + vals[i].val_len + 1 > BUFSIZ) {
  124: 							rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n");
  125: 							rrpc.ret_retcode = ret = -7;
  126: 							rrpc.ret_argc = 0;
  127: 							break;
  128: 						}
  129: 
  130: 						memcpy(data, vals[i].val.string, vals[i].val_len + 1);
  131: 						data += vals[i].val_len + 1;
  132: 						Limit += vals[i].val_len + 1;
  133: 						break;
  134: 					default:
  135: 						break;
  136: 				}
  137: 
  138: 				RPC_FREE_VAL(&vals[i]);
  139: 			}
  140: 		}
  141: 
  142: 		if ((ret = send(c->cli_sock, buf, Limit, 0)) == -1) {
  143: 			LOGERR;
  144: 			ret = -8;
  145: 			break;
  146: 		}
  147: 		if (ret != Limit) {
  148: 			rpc_SetErr(EBADMSG, "Error:: in send RPC request, should be send %d bytes, "
  149: 					"really is %d\n", Limit, ret);
  150: 			ret = -9;
  151: 			break;
  152: 		}
  153: 	} while (ret > -1);
  154: 
  155: 	shutdown(c->cli_sock, SHUT_RDWR);
  156: 	close(c->cli_sock);
  157: 	memset(c, 0, sizeof(rpc_cli_t));
  158: 	return (void*) ret;
  159: }
  160: 
  161: 
  162: static void *
  163: rpc_srv_dispatchVars(void *arg)
  164: {
  165: 	rpc_cli_t *c = arg;
  166: 	rpc_srv_t *s;
  167: 	rpc_blob_t *b;
  168: 	int cx, ret;
  169: 	fd_set fds;
  170: 	u_char buf[sizeof(struct tagBLOBHdr)];
  171: 	struct tagBLOBHdr *blob;
  172: 
  173: 	if (!arg) {
  174: 		rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced BLOB client ...\n");
  175: 		return NULL;
  176: 	} else
  177: 		s = c->cli_parent;
  178: 
  179: 	cx = -1;
  180: 	do {
  181: 		FD_ZERO(&fds);
  182: 		FD_SET(c->cli_sock, &fds);
  183: 		ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);
  184: 		if (ret == -1) {
  185: 			ret = -2;
  186: 		} else
  187: 			cx++;
  188: 		memset(buf, 0, sizeof buf);
  189: 		if ((ret = recv(c->cli_sock, buf, sizeof buf, 0)) == -1) {
  190: 			LOGERR;
  191: 			ret = -3;
  192: 			break;
  193: 		}
  194: 		if (!ret || s->srv_blob.state == disable) {	// receive EOF or disable service
  195: 			ret = 0;
  196: 			break;
  197: 		}
  198: 		if (ret < sizeof(struct tagBLOBHdr)) {
  199: 			rpc_SetErr(EMSGSIZE, "Error:: too short BLOB packet ...\n");
  200: 			ret = -4;
  201: 			break;
  202: 		} else
  203: 			blob = (struct tagBLOBHdr*) buf;
  204: 		// check BLOB packet session info
  205: 		if (memcmp(&blob->hdr_session, &s->srv_session, sizeof blob->hdr_session) || 
  206: 				blob->hdr_seq != cx) {
  207: 			rpc_SetErr(EINVAL, "Error:: get invalid BLOB session in seq=%d...\n", blob->hdr_seq);
  208: 			ret = -5;
  209: 			break;
  210: 		}
  211: 		// Go to proceed packet ...
  212: 		switch (blob->hdr_cmd) {
  213: 			case get:
  214: 				if (!(b = rpc_srv_getBLOB(s, blob->hdr_var))) {
  215: 					rpc_SetErr(EINVAL, "Error:: var (%x) not found into BLOB server ...\n", 
  216: 							blob->hdr_var);
  217: 					ret = -6;
  218: 					break;
  219: 				}
  220: 
  221: 				if (rpc_srv_blobMap(s, b) != -1) {
  222: 					ret = rpc_srv_sendBLOB(c, b);
  223: 					rpc_srv_blobUnmap(b);
  224: 				} else
  225: 					ret = -7;
  226: 				break;
  227: 			case set:
  228: 				if ((b = rpc_srv_registerBLOB(s, blob->hdr_len))) {
  229: 					// set new BLOB variable for reply :)
  230: 					blob->hdr_var = b->blob_var;
  231: 
  232: 					ret = rpc_srv_recvBLOB(c, b);
  233: 					rpc_srv_blobUnmap(b);
  234: 				} else
  235: 					ret = -7;
  236: 				break;
  237: 			case unset:
  238: 				ret = rpc_srv_unregisterBLOB(s, blob->hdr_var);
  239: 				if (ret == -1)
  240: 					ret = -7;
  241: 				break;
  242: 			default:
  243: 				rpc_SetErr(EINVAL, "Error:: unsupported BLOB command (%d)...\n", 
  244: 						blob->hdr_cmd);
  245: 				ret = -7;
  246: 		}
  247: 
  248: 		// Replay to client!
  249: 		blob->hdr_cmd = ret < 0 ? error : ok;
  250: 		blob->hdr_seq = ret;
  251: 		if ((ret = send(c->cli_sock, buf, sizeof buf, 0)) == -1) {
  252: 			LOGERR;
  253: 			ret = -8;
  254: 			break;
  255: 		}
  256: 		if (ret != sizeof buf) {
  257: 			rpc_SetErr(EBADMSG, "Error:: in send BLOB reply, should be send %d bytes, "
  258: 					"really is %d\n", sizeof buf, ret);
  259: 			ret = -9;
  260: 			break;
  261: 		}
  262: 	} while (ret > -1);
  263: 
  264: 	shutdown(c->cli_sock, SHUT_RDWR);
  265: 	close(c->cli_sock);
  266: 	memset(c, 0, sizeof(rpc_cli_t));
  267: 	return (void*) ret;
  268: }
  269: 
  270: // -------------------------------------------------
  271: 
  272: /*
  273:  * rpc_srv_initBLOBServer() Init & create BLOB Server
  274:  * @Port = Port for bind server, if Port == 0 default port is selected
  275:  * @diskDir = Disk place for BLOB file objects
  276:  * return: -1 == error or 0 bind and created BLOB server instance
  277:  */
  278: int
  279: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
  280: {
  281: 	int n = 1;
  282: 	struct sockaddr_in sin;
  283: 	struct sockaddr_in6 sin6;
  284: 
  285: 	if (!srv) {
  286: 		rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init BLOB server ...\n");
  287: 		return -1;
  288: 	}
  289: 	if (srv->srv_blob.state) {
  290: 		rpc_SetErr(EPERM, "Warning:: Already started BLOB server!\n");
  291: 		return 0;
  292: 	}
  293: 	if (!Port)
  294: 		Port = RPC_DEFPORT + 1;
  295: 
  296: 	memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
  297: 	if (access(diskDir, R_OK | W_OK) == -1) {
  298: 		LOGERR;
  299: 		return -1;
  300: 	} else
  301: 		strlcpy(srv->srv_blob.dir, diskDir, UCHAR_MAX + 1);
  302: 
  303: 	srv->srv_blob.server.cli_tid = pthread_self();
  304: 	srv->srv_blob.server.cli_parent = srv;
  305: 	if (srv->srv_server.cli_sa.sa_family == AF_INET) {
  306: 		memcpy(&sin, &srv->srv_server.cli_sa, sizeof sin);
  307: 		sin.sin_port = htons(Port);
  308: 		memcpy(&srv->srv_blob.server.cli_sa, &sin, sizeof(struct sockaddr));
  309: 	} else {
  310: 		memcpy(&sin6, &srv->srv_server.cli_sa, sizeof sin6);
  311: 		sin6.sin6_port = htons(Port);
  312: 		memcpy(&srv->srv_blob.server.cli_sa, &sin6, sizeof(struct sockaddr));
  313: 	}
  314: 
  315: 	srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa_family, SOCK_STREAM, 0);
  316: 	if (srv->srv_blob.server.cli_sock == -1) {
  317: 		LOGERR;
  318: 		return -1;
  319: 	}
  320: 	if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
  321: 		LOGERR;
  322: 		close(srv->srv_blob.server.cli_sock);
  323: 		return -1;
  324: 	}
  325: 	if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa, 
  326: 				sizeof srv->srv_blob.server.cli_sa) == -1) {
  327: 		LOGERR;
  328: 		close(srv->srv_blob.server.cli_sock);
  329: 		return -1;
  330: 	}
  331: 
  332: 	srv->srv_blob.clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
  333: 	if (!srv->srv_blob.clients) {
  334: 		LOGERR;
  335: 		close(srv->srv_blob.server.cli_sock);
  336: 		return -1;
  337: 	} else
  338: 		memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
  339: 
  340: 	pthread_mutex_init(&srv->srv_blob.mtx, NULL);
  341: 
  342: 	pthread_mutex_lock(&srv->srv_mtx);
  343: 	rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
  344: 	rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 0);
  345: 	rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 0);
  346: 	pthread_mutex_unlock(&srv->srv_mtx);
  347: 
  348: 	srv->srv_blob.state = enable;	// enable BLOB
  349: 	return 0;
  350: }
  351: 
  352: /*
  353:  * rpc_srv_endBLOBServer() Destroy BLOB server, close all opened sockets and free resources
  354:  * @srv = RPC Server instance
  355:  * return: none
  356:  */
  357: void
  358: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
  359: {
  360: 	rpc_cli_t *c;
  361: 	register int i;
  362: 	rpc_blob_t *f;
  363: 
  364: 	if (!srv) {
  365: 		rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
  366: 		return;
  367: 	} else
  368: 		srv->srv_blob.state = disable;
  369: 
  370: 	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);
  371: 	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS);
  372: 	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
  373: 
  374: 	for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
  375: 		if (c->cli_sa.sa_family)
  376: 			shutdown(c->cli_sock, SHUT_RDWR);
  377: 	close(srv->srv_blob.server.cli_sock);
  378: 
  379: 	if (srv->srv_blob.clients)
  380: 		free(srv->srv_blob.clients);
  381: 
  382: 	pthread_mutex_lock(&srv->srv_blob.mtx);
  383: 	while ((f = srv->srv_blob.blobs)) {
  384: 		srv->srv_blob.blobs = f->blob_next;
  385: 		free(f);
  386: 	}
  387: 	pthread_mutex_unlock(&srv->srv_blob.mtx);
  388: 
  389: 	pthread_mutex_destroy(&srv->srv_blob.mtx);
  390: }
  391: 
  392: /*
  393:  * rpc_srv_execBLOBServer() Execute Main BLOB server loop and wait for clients requests
  394:  * @srv = RPC Server instance
  395:  * return: -1 error or 0 ok, infinite loop ...
  396:  */
  397: int
  398: rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv)
  399: {
  400: 	socklen_t salen = sizeof(struct sockaddr);
  401: 	register int i;
  402: 	rpc_cli_t *c;
  403: 	fd_set fds;
  404: 	int ret;
  405: 	struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
  406: 
  407: 	if (!srv || !srv->srv_blob.state) {
  408: 		rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n");
  409: 		return -1;
  410: 	}
  411: 
  412: 	if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
  413: 		LOGERR;
  414: 		return -1;
  415: 	}
  416: 
  417: 	while (!rpc_Kill) {
  418: 		for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)
  419: 			if (!c->cli_sa.sa_family)
  420: 				break;
  421: 		if (c && c->cli_sa.sa_family && c->cli_parent) {
  422: 			usleep(1000000);
  423: 			continue;
  424: 		}
  425: 
  426: 		FD_ZERO(&fds);
  427: 		FD_SET(srv->srv_blob.server.cli_sock, &fds);
  428: 		ret = select(srv->srv_blob.server.cli_sock + 1, &fds, NULL, NULL, &tv);
  429: 		if (ret == -1) {
  430: 			LOGERR;
  431: 			ret = 1;
  432: 			break;
  433: 		}
  434: 		if (!ret)
  435: 			continue;
  436: 
  437: 		c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa, &salen);
  438: 		if (c->cli_sock == -1) {
  439: 			LOGERR;
  440: 			continue;
  441: 		} else
  442: 			c->cli_parent = srv;
  443: 
  444: 		if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchVars, c)) {
  445: 			LOGERR;
  446: 			continue;
  447: 		}
  448: 	}
  449: 
  450: 	return 0;
  451: }
  452: 
  453: 
  454: /*
  455:  * rpc_srv_initServer() Init & create RPC Server
  456:  * @regProgID = ProgramID for authentication & recognition
  457:  * @regProcID = ProcessID for authentication & recognition
  458:  * @concurentClients = Concurent clients at same time to this server
  459:  * @family = Family socket type, AF_INET or AF_INET6
  460:  * @csHost = Host name or IP address for bind server, if NULL any address
  461:  * @Port = Port for bind server, if Port == 0 default port is selected
  462:  * return: NULL == error or !=NULL bind and created RPC server instance
  463:  */
  464: rpc_srv_t *
  465: rpc_srv_initServer(u_int regProgID, u_int regProcID, int concurentClients, 
  466: 		u_short family, const char *csHost, u_short Port)
  467: {
  468: 	rpc_srv_t *srv = NULL;
  469: 	int n = 1;
  470: 	struct hostent *host = NULL;
  471: 	struct sockaddr_in sin;
  472: 	struct sockaddr_in6 sin6;
  473: 
  474: 	if (!concurentClients || !regProgID || (family != AF_INET && family != AF_INET6)) {
  475: 		rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");
  476: 		return NULL;
  477: 	}
  478: 	if (!Port)
  479: 		Port = RPC_DEFPORT;
  480: 	if (csHost) {
  481: 		host = gethostbyname2(csHost, family);
  482: 		if (!host) {
  483: 			rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno));
  484: 			return NULL;
  485: 		}
  486: 	}
  487: 	switch (family) {
  488: 		case AF_INET:
  489: 			memset(&sin, 0, sizeof sin);
  490: 			sin.sin_len = sizeof sin;
  491: 			sin.sin_family = family;
  492: 			sin.sin_port = htons(Port);
  493: 			if (csHost)
  494: 				memcpy(&sin.sin_addr, host->h_addr, host->h_length);
  495: 			break;
  496: 		case AF_INET6:
  497: 			memset(&sin6, 0, sizeof sin6);
  498: 			sin6.sin6_len = sizeof sin6;
  499: 			sin6.sin6_family = family;
  500: 			sin6.sin6_port = htons(Port);
  501: 			if (csHost)
  502: 				memcpy(&sin6.sin6_addr, host->h_addr, host->h_length);
  503: 			break;
  504: 		default:
  505: 			rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t start RPC server ...\n");
  506: 			return NULL;
  507: 	}
  508: 
  509: 	srv = malloc(sizeof(rpc_srv_t));
  510: 	if (!srv) {
  511: 		LOGERR;
  512: 		return NULL;
  513: 	} else
  514: 		memset(srv, 0, sizeof(rpc_srv_t));
  515: 
  516: 	srv->srv_numcli = concurentClients;
  517: 	srv->srv_session.sess_version = RPC_VERSION;
  518: 	srv->srv_session.sess_program = regProgID;
  519: 	srv->srv_session.sess_process = regProcID;
  520: 
  521: 	srv->srv_server.cli_tid = pthread_self();
  522: 	srv->srv_server.cli_parent = srv;
  523: 	if (family == AF_INET)
  524: 		memcpy(&srv->srv_server.cli_sa, &sin, sizeof srv->srv_server.cli_sa);
  525: 	else
  526: 		memcpy(&srv->srv_server.cli_sa, &sin6, sizeof srv->srv_server.cli_sa);
  527: 	srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);
  528: 	if (srv->srv_server.cli_sock == -1) {
  529: 		LOGERR;
  530: 		free(srv);
  531: 		return NULL;
  532: 	}
  533: 	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
  534: 		LOGERR;
  535: 		close(srv->srv_server.cli_sock);
  536: 		free(srv);
  537: 		return NULL;
  538: 	}
  539: 	if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa, sizeof srv->srv_server.cli_sa) == -1) {
  540: 		LOGERR;
  541: 		close(srv->srv_server.cli_sock);
  542: 		free(srv);
  543: 		return NULL;
  544: 	}
  545: 
  546: 	srv->srv_clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
  547: 	if (!srv->srv_clients) {
  548: 		LOGERR;
  549: 		close(srv->srv_server.cli_sock);
  550: 		free(srv);
  551: 		return NULL;
  552: 	} else
  553: 		memset(srv->srv_clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
  554: 
  555: 	rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0);
  556: 	rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 0);
  557: 	rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 0);
  558: 	rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 0);
  559: 
  560: 	pthread_mutex_init(&srv->srv_mtx, NULL);
  561: 	return srv;
  562: }
  563: 
  564: /*
  565:  * rpc_srv_endServer() Destroy RPC server, close all opened sockets and free resources
  566:  * @srv = RPC Server instance
  567:  * return: none
  568:  */
  569: void
  570: rpc_srv_endServer(rpc_srv_t * __restrict srv)
  571: {
  572: 	rpc_cli_t *c;
  573: 	register int i;
  574: 	rpc_func_t *f;
  575: 
  576: 	if (!srv) {
  577: 		rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
  578: 		return;
  579: 	}
  580: 
  581: 	rpc_srv_endBLOBServer(srv);
  582: 
  583: 	for (i = 0, c = srv->srv_clients; i < srv->srv_numcli && c; i++, c++)
  584: 		if (c->cli_sa.sa_family)
  585: 			shutdown(c->cli_sock, SHUT_RDWR);
  586: 	close(srv->srv_server.cli_sock);
  587: 
  588: 	if (srv->srv_clients) {
  589: 		free(srv->srv_clients);
  590: 		srv->srv_numcli = 0;
  591: 	}
  592: 
  593: 	pthread_mutex_lock(&srv->srv_mtx);
  594: 	while ((f = srv->srv_funcs)) {
  595: 		srv->srv_funcs = f->func_next;
  596: 		free(f);
  597: 	}
  598: 	pthread_mutex_unlock(&srv->srv_mtx);
  599: 
  600: 	pthread_mutex_destroy(&srv->srv_mtx);
  601: 
  602: 	free(srv);
  603: 	srv = NULL;
  604: }
  605: 
  606: /*
  607:  * rpc_srv_execServer() Execute Main server loop and wait for clients requests
  608:  * @srv = RPC Server instance
  609:  * return: -1 error or 0 ok, infinite loop ...
  610:  */
  611: int
  612: rpc_srv_execServer(rpc_srv_t * __restrict srv)
  613: {
  614: 	socklen_t salen = sizeof(struct sockaddr);
  615: 	register int i;
  616: 	rpc_cli_t *c;
  617: 	fd_set fds;
  618: 	int ret;
  619: 	struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
  620: 
  621: 	if (!srv) {
  622: 		rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n");
  623: 		return -1;
  624: 	}
  625: 
  626: 	if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) {
  627: 		LOGERR;
  628: 		return -1;
  629: 	}
  630: 
  631: 	while (!rpc_Kill) {
  632: 		for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)
  633: 			if (!c->cli_sa.sa_family)
  634: 				break;
  635: 		if (c && c->cli_sa.sa_family && c->cli_parent) {
  636: 			usleep(1000000);
  637: 			continue;
  638: 		}
  639: 
  640: 		FD_ZERO(&fds);
  641: 		FD_SET(srv->srv_server.cli_sock, &fds);
  642: 		ret = select(srv->srv_server.cli_sock + 1, &fds, NULL, NULL, &tv);
  643: 		if (ret == -1) {
  644: 			LOGERR;
  645: 			ret = 1;
  646: 			break;
  647: 		}
  648: 		if (!ret)
  649: 			continue;
  650: 
  651: 		c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa, &salen);
  652: 		if (c->cli_sock == -1) {
  653: 			LOGERR;
  654: 			continue;
  655: 		} else
  656: 			c->cli_parent = srv;
  657: 
  658: 		if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchCall, c)) {
  659: 			LOGERR;
  660: 			continue;
  661: 		}
  662: 	}
  663: 
  664: 	return 0;
  665: }
  666: 
  667: // ---------------------------------------------------------
  668: 
  669: /*
  670:  * rpc_srv_execCall() Execute registered call from RPC server
  671:  * @data = RPC const data
  672:  * @call = Register RPC call
  673:  * @rpc = IN RPC call structure
  674:  * @args = IN RPC call array of rpc values
  675:  * return: -1 error, !=-1 ok
  676:  */
  677: int
  678: rpc_srv_execCall(void * const data, rpc_func_t * __restrict call, 
  679: 		struct tagRPCCall * __restrict rpc, rpc_val_t * __restrict args)
  680: {
  681: 	void *dl;
  682: 	rpc_callback_t func;
  683: 	int ret;
  684: 
  685: 	if (!data || !call || !rpc) {
  686: 		rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t exec call from RPC server ...\n");
  687: 		return -1;
  688: 	}
  689: 
  690: 	dl = dlopen((char*) (*call->func_file ? call->func_file : NULL), RTLD_NOW);
  691: 	if (!dl) {
  692: 		rpc_SetErr(ENOENT, "Error:: Can`t attach module %s!\n", dlerror());
  693: 		return -1;
  694: 	}
  695: 
  696: 	func = dlsym(dl, (char*) call->func_name);
  697: 	if (func)
  698: 		ret = func(data, call, rpc->call_argc, args);
  699: 	else {
  700: 		rpc_SetErr(ENOEXEC, "Error:: Can`t find function %s!\n", dlerror());
  701: 		ret = -1;
  702: 	}
  703: 
  704: 	dlclose(dl);
  705: 	return ret;
  706: }

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>