File:  [ELWIX - Embedded LightWeight unIX -] / libaitrpc / src / srv.c
Revision 1.21: download - view: text, annotated - select for diffs - revision graph
Thu Nov 14 21:44:14 2013 UTC (10 years, 7 months ago) by misho
Branches: MAIN
CVS tags: rpc6_5, RPC6_4, HEAD
version 6.4

    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.21 2013/11/14 21:44:14 misho Exp $
    7: *
    8: **************************************************************************
    9: The ELWIX and AITNET software is distributed under the following
   10: terms:
   11: 
   12: All of the documentation and software included in the ELWIX and AITNET
   13: Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>
   14: 
   15: Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013
   16: 	by Michael Pounov <misho@elwix.org>.  All rights reserved.
   17: 
   18: Redistribution and use in source and binary forms, with or without
   19: modification, are permitted provided that the following conditions
   20: are met:
   21: 1. Redistributions of source code must retain the above copyright
   22:    notice, this list of conditions and the following disclaimer.
   23: 2. Redistributions in binary form must reproduce the above copyright
   24:    notice, this list of conditions and the following disclaimer in the
   25:    documentation and/or other materials provided with the distribution.
   26: 3. All advertising materials mentioning features or use of this software
   27:    must display the following acknowledgement:
   28: This product includes software developed by Michael Pounov <misho@elwix.org>
   29: ELWIX - Embedded LightWeight unIX and its contributors.
   30: 4. Neither the name of AITNET nor the names of its contributors
   31:    may be used to endorse or promote products derived from this software
   32:    without specific prior written permission.
   33: 
   34: THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND
   35: ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
   36: IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
   37: ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
   38: FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
   39: DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
   40: OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
   41: HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
   42: LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
   43: OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
   44: SUCH DAMAGE.
   45: */
   46: #include "global.h"
   47: 
   48: 
   49: /* SOCK_STREAM */
   50: static void *acceptClients(sched_task_t *);
   51: static void *closeClient(sched_task_t *);
   52: static void *rxPacket(sched_task_t *);
   53: static void *txPacket(sched_task_t *);
   54: 
   55: /* SOCK_DGRAM */
   56: static void *freeClient(sched_task_t *);
   57: static void *rxUDPPacket(sched_task_t *);
   58: static void *txUDPPacket(sched_task_t *);
   59: 
   60: /* SOCK_RAW */
   61: 
   62: static sched_task_func_t cbProto[SOCK_RAW + 1][4] = {
   63: 	{ acceptClients, closeClient, rxPacket, txPacket },	/* SOCK_STREAM */
   64: 	{ acceptClients, closeClient, rxPacket, txPacket },	/* SOCK_STREAM */
   65: 	{ rxUDPPacket, freeClient, rxUDPPacket, txUDPPacket }, 	/* SOCK_DGRAM */
   66: 	{ NULL, NULL, NULL, NULL }				/* SOCK_RAW */
   67: };
   68: 
   69: 
   70: void
   71: rpc_freeCli(rpc_cli_t * __restrict c)
   72: {
   73: 	rpc_srv_t *s = c->cli_parent;
   74: 
   75: 	schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
   76: 
   77: 	/* free buffer */
   78: 	AIT_FREE_VAL(&c->cli_buf);
   79: 
   80: 	array_Del(s->srv_clients, c->cli_id, 0);
   81: 	if (c)
   82: 		e_free(c);
   83: }
   84: 
   85: 
   86: static inline int
   87: _check4freeslot(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
   88: {
   89: 	rpc_cli_t *c = NULL;
   90: 	register int i;
   91: 
   92: 	/* check free slots for connect */
   93: 	for (i = 0; i < array_Size(srv->srv_clients) && 
   94: 			(c = array(srv->srv_clients, i, rpc_cli_t*)); i++)
   95: 		/* check for duplicates */
   96: 		if (sa && !e_addrcmp(&c->cli_sa, sa, 42))
   97: 			break;
   98: 	if (i >= array_Size(srv->srv_clients))
   99: 		return -1;	/* no more free slots! */
  100: 
  101: 	return i;
  102: }
  103: 
  104: static rpc_cli_t *
  105: _allocClient(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
  106: {
  107: 	rpc_cli_t *c = NULL;
  108: 	int n;
  109: 
  110: 	n = _check4freeslot(srv, sa);
  111: 	if (n == -1)
  112: 		return NULL;
  113: 	else
  114: 		c = array(srv->srv_clients, n, rpc_cli_t*);
  115: 
  116: 	if (!c) {
  117: 		c = e_malloc(sizeof(rpc_cli_t));
  118: 		if (!c) {
  119: 			LOGERR;
  120: 			srv->srv_kill = 1;
  121: 			return NULL;
  122: 		} else {
  123: 			memset(c, 0, sizeof(rpc_cli_t));
  124: 			array_Set(srv->srv_clients, n, c);
  125: 			c->cli_id = n;
  126: 			c->cli_parent = srv;
  127: 		}
  128: 
  129: 		/* alloc empty buffer */
  130: 		AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
  131: 	}
  132: 
  133: 	return c;
  134: }
  135: 
  136: 
  137: static void *
  138: freeClient(sched_task_t *task)
  139: {
  140: 	rpc_freeCli(TASK_ARG(task));
  141: 
  142: 	return NULL;
  143: }
  144: 
  145: static void *
  146: closeClient(sched_task_t *task)
  147: {
  148: 	int sock = ((rpc_cli_t*) TASK_ARG(task))->cli_sock;
  149: 
  150: 	rpc_freeCli(TASK_ARG(task));
  151: 
  152: 	/* close client socket */
  153: 	shutdown(sock, SHUT_RDWR);
  154: 	close(sock);
  155: 	return NULL;
  156: }
  157: 
  158: static void *
  159: txPacket(sched_task_t *task)
  160: {
  161: 	rpc_cli_t *c = TASK_ARG(task);
  162: 	rpc_srv_t *s = c->cli_parent;
  163: 	rpc_func_t *f = NULL;
  164: 	u_char *buf = AIT_GET_BUF(&c->cli_buf);
  165: 	struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
  166: 	int ret, estlen, wlen = sizeof(struct tagRPCCall);
  167: 	struct pollfd pfd;
  168: #ifdef TCP_SESSION_TIMEOUT
  169: 	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
  170: 
  171: 	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
  172: 	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  173: 			 TASK_ARG(task), ts, TASK_ARG(task), 0);
  174: #endif
  175: 
  176: 	if (rpc->call_argc) {
  177: 		f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
  178: 		if (!f) {
  179: 			rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
  180: 
  181: 			rpc->call_argc ^= rpc->call_argc;
  182: 			rpc->call_rep.ret = RPC_ERROR(-1);
  183: 			rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
  184: 		} else {
  185: 			/* calc estimated length */
  186: 			estlen = ait_resideVars(RPC_RETVARS(c)) + wlen;
  187: 			if (estlen > AIT_LEN(&c->cli_buf))
  188: 				AIT_RE_BUF(&c->cli_buf, estlen);
  189: 			buf = AIT_GET_BUF(&c->cli_buf);
  190: 			rpc = (struct tagRPCCall*) buf;
  191: 
  192: 			rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
  193: 			/* Go Encapsulate variables */
  194: 			ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen, 
  195: 					RPC_RETVARS(c));
  196: 			/* Free return values */
  197: 			ait_freeVars(&c->cli_vars);
  198: 			if (ret == -1) {
  199: 				rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
  200: 
  201: 				rpc->call_argc ^= rpc->call_argc;
  202: 				rpc->call_rep.ret = RPC_ERROR(-1);
  203: 				rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
  204: 			} else
  205: 				wlen += ret;
  206: 		}
  207: 	}
  208: 
  209: 	rpc->call_len = htonl(wlen);
  210: 
  211: #if 0
  212: 	/* calculate CRC */
  213: 	rpc->call_crc ^= rpc->call_crc;
  214: 	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
  215: #endif
  216: 
  217: 	/* send reply */
  218: 	pfd.fd = TASK_FD(task);
  219: 	pfd.events = POLLOUT;
  220: 	for (; wlen > 0; wlen -= ret, buf += ret) {
  221: 		if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
  222: 				pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
  223: 			if (ret)
  224: 				LOGERR;
  225: 			else
  226: 				rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond");
  227: 			/* close connection */
  228: 			schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  229: 					TASK_ARG(task), 0, NULL, 0);
  230: 			return NULL;
  231: 		}
  232: 		ret = send(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), MSG_NOSIGNAL);
  233: 		if (ret == -1) {
  234: 			/* close connection */
  235: 			schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  236: 					TASK_ARG(task), 0, NULL, 0);
  237: 			return NULL;
  238: 		}
  239: 	}
  240: 
  241: 	return NULL;
  242: }
  243: 
  244: static void *
  245: execCall(sched_task_t *task)
  246: {
  247: 	rpc_cli_t *c = TASK_ARG(task);
  248: 	rpc_srv_t *s = c->cli_parent;
  249: 	rpc_func_t *f = NULL;
  250: 	array_t *arr = NULL;
  251: 	u_char *buf = AIT_GET_BUF(&c->cli_buf);
  252: 	struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
  253: 	int argc = ntohs(rpc->call_argc);
  254: 
  255: 	/* Go decapsulate variables ... */
  256: 	if (argc) {
  257: 		arr = ait_buffer2vars(buf + sizeof(struct tagRPCCall), 
  258: 				AIT_LEN(&c->cli_buf) - sizeof(struct tagRPCCall), argc, 42);
  259: 		if (!arr) {
  260: 			rpc_SetErr(ERPCMISMATCH, "#%d - %s", elwix_GetErrno(), elwix_GetError());
  261: 
  262: 			rpc->call_argc ^= rpc->call_argc;
  263: 			rpc->call_rep.ret = RPC_ERROR(-1);
  264: 			rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
  265: 			return NULL;
  266: 		}
  267: 	} else
  268: 		arr = NULL;
  269: 
  270: 	if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag)))) {
  271: 		rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
  272: 
  273: 		rpc->call_argc ^= rpc->call_argc;
  274: 		rpc->call_rep.ret = RPC_ERROR(-1);
  275: 		rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
  276: 	} else {
  277: 		/* if client doesn't want reply */
  278: 		argc = RPC_CHK_NOREPLY(rpc);
  279: 		rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(c, rpc, f->func_name, arr));
  280: 		if (rpc->call_rep.ret == htonl(-1)) {
  281: 			if (!rpc->call_rep.eno) {
  282: 				LOGERR;
  283: 				rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
  284: 			}
  285: 			rpc->call_argc ^= rpc->call_argc;
  286: 		} else {
  287: 			rpc->call_rep.eno ^= rpc->call_rep.eno;
  288: 			if (argc) {
  289: 				/* without reply */
  290: 				ait_freeVars(&c->cli_vars);
  291: 				rpc->call_argc ^= rpc->call_argc;
  292: 			} else {
  293: 				/* reply */
  294: 				rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
  295: 			}
  296: 		}
  297: 	}
  298: 
  299: 	array_Destroy(&arr);
  300: 	return NULL;
  301: }
  302: 
  303: static void *
  304: rxPacket(sched_task_t *task)
  305: {
  306: 	rpc_cli_t *c = TASK_ARG(task);
  307: 	rpc_srv_t *s = c->cli_parent;
  308: 	int len, rlen, noreply, estlen;
  309: #if 0
  310: 	u_short crc;
  311: #endif
  312: 	u_char *buf = AIT_GET_BUF(&c->cli_buf);
  313: 	struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
  314: 	struct pollfd pfd;
  315: #ifdef TCP_SESSION_TIMEOUT
  316: 	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
  317: 
  318: 	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
  319: 	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  320: 			 TASK_ARG(task), ts, TASK_ARG(task), 0);
  321: #endif
  322: 
  323: 	memset(buf, 0, sizeof(struct tagRPCCall));
  324: 	rlen = recv(TASK_FD(task), rpc, sizeof(struct tagRPCCall), MSG_PEEK);
  325: 	if (rlen < sizeof(struct tagRPCCall)) {
  326: 		/* close connection */
  327: 		schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  328: 				TASK_ARG(task), 0, NULL, 0);
  329: 		return NULL;
  330: 	} else {
  331: 		estlen = ntohl(rpc->call_len);
  332: 		if (estlen > AIT_LEN(&c->cli_buf))
  333: 			AIT_RE_BUF(&c->cli_buf, estlen);
  334: 		rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
  335: 		buf = AIT_GET_BUF(&c->cli_buf);
  336: 		len = estlen;
  337: 	}
  338: 
  339: 	/* get next part of packet */
  340: 	memset(buf, 0, len);
  341: 	pfd.fd = TASK_FD(task);
  342: 	pfd.events = POLLIN | POLLPRI;
  343: 	for (; len > 0; len -= rlen, buf += rlen) {
  344: 		if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
  345: 				pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
  346: 			if (rlen)
  347: 				LOGERR;
  348: 			else
  349: 				rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond");
  350: 			schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  351: 					TASK_ARG(task), 0, NULL, 0);
  352: 			return NULL;
  353: 		}
  354: 		rlen = recv(TASK_FD(task), buf, len, 0);
  355: 		if (rlen == -1) {
  356: 			/* close connection */
  357: 			schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  358: 					TASK_ARG(task), 0, NULL, 0);
  359: 			return NULL;
  360: 		}
  361: 	}
  362: 
  363: #if 0
  364: 	/* check integrity of packet */
  365: 	crc = ntohs(rpc->call_crc);
  366: 	rpc->call_crc ^= rpc->call_crc;
  367: 	if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
  368: 		rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
  369: 		return NULL;
  370: 	}
  371: #endif
  372: 
  373: 	noreply = RPC_CHK_NOREPLY(rpc);
  374: 
  375: 	/* check RPC packet session info */
  376: 	if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
  377: 		rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
  378: 
  379: 		rpc->call_argc ^= rpc->call_argc;
  380: 		rpc->call_rep.ret = RPC_ERROR(-1);
  381: 		rpc->call_rep.eno = RPC_ERROR(errno);
  382: 	} else {
  383: 		/* execute RPC call */
  384: 		schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len);
  385: 	}
  386: 
  387: 	/* send RPC reply */
  388: 	if (!noreply)
  389: 		schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET], 
  390: 				TASK_ARG(task), TASK_FD(task), rpc, len);
  391: 
  392: 	/* lets get next packet */
  393: 	schedReadSelf(task);
  394: 	return NULL;
  395: }
  396: 
  397: static void *
  398: acceptClients(sched_task_t *task)
  399: {
  400: 	rpc_srv_t *srv = TASK_ARG(task);
  401: 	rpc_cli_t *c = NULL;
  402: 	socklen_t salen = sizeof(sockaddr_t);
  403: 	int sock;
  404: #ifdef TCP_SESSION_TIMEOUT
  405: 	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
  406: #endif
  407: 
  408: 	c = _allocClient(srv, NULL);
  409: 	if (!c) {
  410: 		EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
  411: 		if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
  412: 			shutdown(sock, SHUT_RDWR);
  413: 			close(sock);
  414: 		}
  415: 		goto end;
  416: 	}
  417: 
  418: 	/* accept client */
  419: 	c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
  420: 	if (c->cli_sock == -1) {
  421: 		LOGERR;
  422: 		AIT_FREE_VAL(&c->cli_buf);
  423: 		array_Del(srv->srv_clients, c->cli_id, 42);
  424: 		goto end;
  425: 	} else
  426: 		fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
  427: 
  428: #ifdef TCP_SESSION_TIMEOUT
  429: 	/* armed timer for close stateless connection */
  430: 	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
  431: 	schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], c, 
  432: 			ts, c, 0);
  433: #endif
  434: 	schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c, 
  435: 			c->cli_sock, NULL, 0);
  436: end:
  437: 	schedReadSelf(task);
  438: 	return NULL;
  439: }
  440: 
  441: 
  442: static void *
  443: txUDPPacket(sched_task_t *task)
  444: {
  445: 	rpc_cli_t *c = TASK_ARG(task);
  446: 	rpc_srv_t *s = c->cli_parent;
  447: 	rpc_func_t *f = NULL;
  448: 	u_char *buf = AIT_GET_BUF(&c->cli_buf);
  449: 	struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
  450: 	int ret, wlen = sizeof(struct tagRPCCall);
  451: 	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
  452: 
  453: 	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
  454: 	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  455: 			 TASK_ARG(task), ts, TASK_ARG(task), 0);
  456: 
  457: 	if (rpc->call_argc) {
  458: 		f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
  459: 		if (!f) {
  460: 			rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
  461: 			rpc->call_argc ^= rpc->call_argc;
  462: 			rpc->call_rep.ret = RPC_ERROR(-1);
  463: 			rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
  464: 		} else {
  465: 			rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
  466: 			/* Go Encapsulate variables */
  467: 			ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen, 
  468: 					RPC_RETVARS(c));
  469: 			/* Free return values */
  470: 			ait_freeVars(&c->cli_vars);
  471: 			if (ret == -1) {
  472: 				rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
  473: 				rpc->call_argc ^= rpc->call_argc;
  474: 				rpc->call_rep.ret = RPC_ERROR(-1);
  475: 				rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
  476: 			} else
  477: 				wlen += ret;
  478: 		}
  479: 	}
  480: 
  481: 	rpc->call_len = htonl(wlen);
  482: 
  483: 	/* calculate CRC */
  484: 	rpc->call_crc ^= rpc->call_crc;
  485: 	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
  486: 
  487: 	/* send reply */
  488: 	ret = sendto(TASK_FD(task), buf, wlen, MSG_NOSIGNAL, 
  489: 			&c->cli_sa.sa, c->cli_sa.sa.sa_len);
  490: 	if (ret == -1 || ret != wlen) {
  491: 		/* close connection */
  492: 		schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
  493: 				 TASK_ARG(task), 0, NULL, 0);
  494: 	}
  495: 
  496: 	return NULL;
  497: }
  498: 
  499: static void *
  500: rxUDPPacket(sched_task_t *task)
  501: {
  502: 	rpc_srv_t *srv = TASK_ARG(task);
  503: 	rpc_cli_t *c = NULL;
  504: 	int len, rlen, noreply;
  505: 	u_short crc, off = 0;
  506: 	u_char buf[USHRT_MAX + 1];
  507: 	struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
  508: 	sockaddr_t sa;
  509: 	socklen_t salen;
  510: 	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
  511: 
  512: 	/* receive connect packet */
  513: 	salen = sa.ss.ss_len = sizeof(sockaddr_t);
  514: 	rlen = recvfrom(TASK_FD(task), buf, sizeof buf, 0, &sa.sa, &salen);
  515: 	if (rlen < 1)
  516: 		goto end;
  517: 
  518: 	c = _allocClient(srv, &sa);
  519: 	if (!c)
  520: 		goto end;
  521: 	else {
  522: 		c->cli_sock = TASK_FD(task);
  523: 		memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);
  524: 		memcpy(AIT_GET_BUF(&c->cli_buf), buf, AIT_LEN(&c->cli_buf));
  525: 		/* armed timer for close stateless connection */
  526: 		schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
  527: 		schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], 
  528: 				c, ts, c, 0);
  529: 	}
  530: 
  531: 	do {
  532: 		/* check RPC packet */
  533: 		if (rlen < sizeof(struct tagRPCCall)) {
  534: 			rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
  535: 			break;
  536: 		} else
  537: 			rpc = (struct tagRPCCall*) (AIT_GET_BUF(&c->cli_buf) + off);
  538: 
  539: 		len = ntohl(rpc->call_len);
  540: 		rlen -= len;
  541: 
  542: 		/* check RPC packet lengths */
  543: 		if (rlen < 0 || len < sizeof(struct tagRPCCall)) {
  544: 			rpc_SetErr(ERPCMISMATCH, "Broken RPC packet length");
  545: 			/* skip entire packet */
  546: 			break;
  547: 		}
  548: 
  549: 		/* check integrity of packet */
  550: 		crc = ntohs(rpc->call_crc);
  551: 		rpc->call_crc ^= rpc->call_crc;
  552: 		if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
  553: 			rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
  554: 
  555: 			off += len;
  556: 			/* try next packet remaining into buffer */
  557: 			continue;
  558: 		}
  559: 
  560: 		noreply = RPC_CHK_NOREPLY(rpc);
  561: 
  562: 		/* check RPC packet session info */
  563: 		if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
  564: 			rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
  565: 
  566: 			rpc->call_argc ^= rpc->call_argc;
  567: 			rpc->call_rep.ret = RPC_ERROR(-1);
  568: 			rpc->call_rep.eno = RPC_ERROR(errno);
  569: 		} else {
  570: 			/* execute RPC call */
  571: 			schedEvent(TASK_ROOT(task), execCall, c, 
  572: 					(int) noreply, rpc, len);
  573: 		}
  574: 
  575: 		/* send RPC reply */
  576: 		if (!noreply)
  577: 			schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], 
  578: 					c, TASK_FD(task), rpc, len);
  579: 
  580: 		off += len;
  581: 	} while (rlen > 0);
  582: 
  583: end:
  584: 	schedReadSelf(task);
  585: 	return NULL;
  586: }
  587: 
  588: /* ------------------------------------------------------ */
  589: 
  590: void
  591: rpc_freeBLOBCli(rpc_cli_t * __restrict c)
  592: {
  593: 	rpc_srv_t *s = c->cli_parent;
  594: 
  595: 	schedCancelby(s->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
  596: 
  597: 	/* free buffer */
  598: 	AIT_FREE_VAL(&c->cli_buf);
  599: 
  600: 	array_Del(s->srv_blob.clients, c->cli_id, 0);
  601: 	if (c)
  602: 		e_free(c);
  603: }
  604: 
  605: 
  606: static void *
  607: closeBLOBClient(sched_task_t *task)
  608: {
  609: 	int sock = ((rpc_cli_t*) TASK_ARG(task))->cli_sock;
  610: 
  611: 	rpc_freeBLOBCli(TASK_ARG(task));
  612: 
  613: 	/* close client socket */
  614: 	shutdown(sock, SHUT_RDWR);
  615: 	close(sock);
  616: 	return NULL;
  617: }
  618: 
  619: static void *
  620: txBLOB(sched_task_t *task)
  621: {
  622: 	rpc_cli_t *c = TASK_ARG(task);
  623: 	u_char *buf = AIT_GET_BUF(&c->cli_buf);
  624: 	int wlen = sizeof(struct tagBLOBHdr);
  625: 
  626: 	/* send reply */
  627: 	wlen = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);
  628: 	if (wlen == -1 || wlen != sizeof(struct tagBLOBHdr)) {
  629: 		/* close blob connection */
  630: 		schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
  631: 	}
  632: 
  633: 	return NULL;
  634: }
  635: 
  636: static void *
  637: rxBLOB(sched_task_t *task)
  638: {
  639: 	rpc_cli_t *c = TASK_ARG(task);
  640: 	rpc_srv_t *s = c->cli_parent;
  641: 	rpc_blob_t *b;
  642: 	struct tagBLOBHdr blob;
  643: 	int rlen;
  644: 
  645: 	memset(&blob, 0, sizeof blob);
  646: 	rlen = recv(TASK_FD(task), &blob, sizeof blob, 0);
  647: 	if (rlen < 1) {
  648: 		/* close blob connection */
  649: 		schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
  650: 		return NULL;
  651: 	}
  652: 
  653: 	/* check BLOB packet */
  654: 	if (rlen < sizeof(struct tagBLOBHdr)) {
  655: 		rpc_SetErr(ERPCMISMATCH, "Short BLOB packet");
  656: 
  657: 		schedReadSelf(task);
  658: 		return NULL;
  659: 	}
  660: 
  661: 	/* check RPC packet session info */
  662: 	if (rpc_chkPktSession(&blob.hdr_session, &s->srv_session)) {
  663: 		rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
  664: 		blob.hdr_cmd = error;
  665: 		goto end;
  666: 	}
  667: 
  668: 	/* Go to proceed packet ... */
  669: 	switch (blob.hdr_cmd) {
  670: 		case get:
  671: 			if (!(b = rpc_srv_getBLOB(s, ntohl(blob.hdr_var)))) {
  672: 				rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob.hdr_var));
  673: 				blob.hdr_cmd = no;
  674: 				blob.hdr_ret = RPC_ERROR(-1);
  675: 				break;
  676: 			} else
  677: 				blob.hdr_len = htonl(b->blob_len);
  678: 
  679: 			if (rpc_srv_blobMap(s, b) != -1) {
  680: 				/* deliver BLOB variable to client */
  681: 				blob.hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
  682: 				rpc_srv_blobUnmap(b);
  683: 			} else {
  684: 				blob.hdr_cmd = error;
  685: 				blob.hdr_ret = RPC_ERROR(-1);
  686: 			}
  687: 			break;
  688: 		case set:
  689: 			if ((b = rpc_srv_registerBLOB(s, ntohl(blob.hdr_len), 
  690: 							ntohl(blob.hdr_ret)))) {
  691: 				/* set new BLOB variable for reply :) */
  692: 				blob.hdr_var = htonl(b->blob_var);
  693: 
  694: 				/* receive BLOB from client */
  695: 				blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
  696: 				rpc_srv_blobUnmap(b);
  697: 			} else {
  698: 				blob.hdr_cmd = error;
  699: 				blob.hdr_ret = RPC_ERROR(-1);
  700: 			}
  701: 			break;
  702: 		case unset:
  703: 			if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) {
  704: 				blob.hdr_cmd = error;
  705: 				blob.hdr_ret = RPC_ERROR(-1);
  706: 			}
  707: 			break;
  708: 		default:
  709: 			rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);
  710: 			blob.hdr_cmd = error;
  711: 			blob.hdr_ret = RPC_ERROR(-1);
  712: 	}
  713: 
  714: end:
  715: 	memcpy(AIT_ADDR(&c->cli_buf), &blob, sizeof blob);
  716: 	schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), NULL, 0);
  717: 	schedReadSelf(task);
  718: 	return NULL;
  719: }
  720: 
  721: static void *
  722: flushBLOB(sched_task_t *task)
  723: {
  724: 	rpc_srv_t *srv = TASK_ARG(task);
  725: 	rpc_blob_t *b, *tmp;
  726: 
  727: 	TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
  728: 		TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);
  729: 
  730: 		rpc_srv_blobFree(srv, b);
  731: 		e_free(b);
  732: 	}
  733: 
  734: 	schedSignalSelf(task);
  735: 	return NULL;
  736: }
  737: 
  738: static void *
  739: acceptBLOBClients(sched_task_t *task)
  740: {
  741: 	rpc_srv_t *srv = TASK_ARG(task);
  742: 	rpc_cli_t *c = NULL;
  743: 	register int i;
  744: 	socklen_t salen = sizeof(sockaddr_t);
  745: 	int sock;
  746: #ifdef TCP_NOPUSH
  747: 	int n = 1;
  748: #endif
  749: 
  750: 	/* check free slots for connect */
  751: 	for (i = 0; i < array_Size(srv->srv_blob.clients) && 
  752: 			(c = array(srv->srv_blob.clients, i, rpc_cli_t*)); i++);
  753: 	if (c) {	/* no more free slots! */
  754: 		EVERBOSE(1, "BLOB client quota exceeded! Connection will be shutdown!\n");
  755: 		if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
  756: 			shutdown(sock, SHUT_RDWR);
  757: 			close(sock);
  758: 		}
  759: 		goto end;
  760: 	}
  761: 
  762: 	c = e_malloc(sizeof(rpc_cli_t));
  763: 	if (!c) {
  764: 		LOGERR;
  765: 		srv->srv_kill = srv->srv_blob.kill = 1;
  766: 		return NULL;
  767: 	} else {
  768: 		memset(c, 0, sizeof(rpc_cli_t));
  769: 		array_Set(srv->srv_blob.clients, i, c);
  770: 		c->cli_id = i;
  771: 		c->cli_parent = srv;
  772: 	}
  773: 
  774: 	/* alloc empty buffer */
  775: 	AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
  776: 
  777: 	/* accept client */
  778: 	c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
  779: 	if (c->cli_sock == -1) {
  780: 		LOGERR;
  781: 		AIT_FREE_VAL(&c->cli_buf);
  782: 		array_Del(srv->srv_blob.clients, i, 42);
  783: 		goto end;
  784: 	} else {
  785: #ifdef TCP_NOPUSH
  786: 		setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n);
  787: #endif
  788: 		fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
  789: 	}
  790: 
  791: 	schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0);
  792: end:
  793: 	schedReadSelf(task);
  794: 	return NULL;
  795: }
  796: 
  797: /* ------------------------------------------------------ */
  798: 
  799: /*
  800:  * rpc_srv_initBLOBServer() - Init & create BLOB Server
  801:  *
  802:  * @srv = RPC server instance
  803:  * @Port = Port for bind server, if Port == 0 default port is selected
  804:  * @diskDir = Disk place for BLOB file objects
  805:  * return: -1 == error or 0 bind and created BLOB server instance
  806:  */
  807: int
  808: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
  809: {
  810: 	int n = 1;
  811: 
  812: 	if (!srv || srv->srv_kill) {
  813: 		rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
  814: 		return -1;
  815: 	}
  816: 
  817: 	memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
  818: 	if (access(diskDir, R_OK | W_OK) == -1) {
  819: 		LOGERR;
  820: 		return -1;
  821: 	} else
  822: 		AIT_SET_STR(&srv->srv_blob.dir, diskDir);
  823: 
  824: 	/* init blob list */
  825: 	TAILQ_INIT(&srv->srv_blob.blobs);
  826: 
  827: 	srv->srv_blob.server.cli_parent = srv;
  828: 
  829: 	memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(sockaddr_t));
  830: 	switch (srv->srv_blob.server.cli_sa.sa.sa_family) {
  831: 		case AF_INET:
  832: 			srv->srv_blob.server.cli_sa.sin.sin_port = 
  833: 				htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin.sin_port) + 1);
  834: 			break;
  835: 		case AF_INET6:
  836: 			srv->srv_blob.server.cli_sa.sin6.sin6_port = 
  837: 				htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin6.sin6_port) + 1);
  838: 			break;
  839: 		case AF_LOCAL:
  840: 			strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob", 
  841: 					sizeof srv->srv_blob.server.cli_sa.sun.sun_path);
  842: 			break;
  843: 		default:
  844: 			AIT_FREE_VAL(&srv->srv_blob.dir);
  845: 			return -1;
  846: 	}
  847: 
  848: 	/* create BLOB server socket */
  849: 	srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
  850: 	if (srv->srv_blob.server.cli_sock == -1) {
  851: 		LOGERR;
  852: 		AIT_FREE_VAL(&srv->srv_blob.dir);
  853: 		return -1;
  854: 	}
  855: 	if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
  856: 		LOGERR;
  857: 		close(srv->srv_blob.server.cli_sock);
  858: 		AIT_FREE_VAL(&srv->srv_blob.dir);
  859: 		return -1;
  860: 	}
  861: 	n = srv->srv_netbuf;
  862: 	if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
  863: 		LOGERR;
  864: 		close(srv->srv_blob.server.cli_sock);
  865: 		AIT_FREE_VAL(&srv->srv_blob.dir);
  866: 		return -1;
  867: 	}
  868: 	if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
  869: 		LOGERR;
  870: 		close(srv->srv_blob.server.cli_sock);
  871: 		AIT_FREE_VAL(&srv->srv_blob.dir);
  872: 		return -1;
  873: 	}
  874: 	if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa, 
  875: 				srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
  876: 		LOGERR;
  877: 		close(srv->srv_blob.server.cli_sock);
  878: 		AIT_FREE_VAL(&srv->srv_blob.dir);
  879: 		return -1;
  880: 	} else
  881: 		fcntl(srv->srv_blob.server.cli_sock, F_SETFL, 
  882: 				fcntl(srv->srv_blob.server.cli_sock, F_GETFL) | O_NONBLOCK);
  883: 
  884: 
  885: 	/* allocate pool for concurent blob clients */
  886: 	srv->srv_blob.clients = array_Init(array_Size(srv->srv_clients));
  887: 	if (!srv->srv_blob.clients) {
  888: 		rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
  889: 		close(srv->srv_blob.server.cli_sock);
  890: 		AIT_FREE_VAL(&srv->srv_blob.dir);
  891: 		return -1;
  892: 	}
  893: 
  894: 	/* init blob scheduler */
  895: 	srv->srv_blob.root = schedBegin();
  896: 	if (!srv->srv_blob.root) {
  897: 		rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
  898: 		array_Destroy(&srv->srv_blob.clients);
  899: 		close(srv->srv_blob.server.cli_sock);
  900: 		AIT_FREE_VAL(&srv->srv_blob.dir);
  901: 		return -1;
  902: 	}
  903: 
  904: 	return 0;
  905: }
  906: 
  907: /*
  908:  * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
  909:  *
  910:  * @srv = RPC Server instance
  911:  * return: none
  912:  */
  913: void
  914: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
  915: {
  916: 	if (!srv)
  917: 		return;
  918: 
  919: 	srv->srv_blob.kill = 1;
  920: 
  921: 	schedEnd(&srv->srv_blob.root);
  922: }
  923: 
  924: /*
  925:  * rpc_srv_loopBLOBServer() - Execute Main BLOB server loop and wait for clients requests
  926:  *
  927:  * @srv = RPC Server instance
  928:  * return: -1 error or 0 ok, infinite loop ...
  929:  */
  930: int
  931: rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
  932: {
  933: 	rpc_cli_t *c;
  934: 	register int i;
  935: 	rpc_blob_t *b, *tmp;
  936: 	struct timespec ts = { RPC_SCHED_POLLING, 0 };
  937: 
  938: 	if (!srv || srv->srv_kill) {
  939: 		rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
  940: 		return -1;
  941: 	}
  942: 
  943: 	if (listen(srv->srv_blob.server.cli_sock, array_Size(srv->srv_blob.clients)) == -1) {
  944: 		LOGERR;
  945: 		return -1;
  946: 	}
  947: 
  948: 	schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0);
  949: 	if (!schedRead(srv->srv_blob.root, acceptBLOBClients, srv, 
  950: 				srv->srv_blob.server.cli_sock, NULL, 0)) {
  951: 		rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
  952: 		return -1;
  953: 	}
  954: 
  955: 	schedPolling(srv->srv_blob.root, &ts, NULL);
  956: 	/* main rpc loop */
  957: 	schedRun(srv->srv_blob.root, &srv->srv_blob.kill);
  958: 
  959: 	/* detach blobs */
  960: 	TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
  961: 		TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);
  962: 
  963: 		rpc_srv_blobFree(srv, b);
  964: 		e_free(b);
  965: 	}
  966: 
  967: 	/* close all clients connections & server socket */
  968: 	for (i = 0; i < array_Size(srv->srv_blob.clients); i++) {
  969: 		c = array(srv->srv_blob.clients, i, rpc_cli_t*);
  970: 		if (c) {
  971: 			shutdown(c->cli_sock, SHUT_RDWR);
  972: 			close(c->cli_sock);
  973: 
  974: 			schedCancelby(srv->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
  975: 			AIT_FREE_VAL(&c->cli_buf);
  976: 		}
  977: 		array_Del(srv->srv_blob.clients, i, 42);
  978: 	}
  979: 	array_Destroy(&srv->srv_blob.clients);
  980: 
  981: 	close(srv->srv_blob.server.cli_sock);
  982: 
  983: 	AIT_FREE_VAL(&srv->srv_blob.dir);
  984: 	return 0;
  985: }
  986: 
  987: 
  988: /*
  989:  * rpc_srv_initServer() - Init & create RPC Server
  990:  *
  991:  * @InstID = Instance for authentication & recognition
  992:  * @concurentClients = Concurent clients at same time to this server
  993:  * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
  994:  * @csHost = Host name or address for bind server, if NULL any address
  995:  * @Port = Port for bind server, if Port == 0 default port is selected
  996:  * @proto = Protocol, if == 0 choose SOCK_STREAM
  997:  * return: NULL == error or !=NULL bind and created RPC server instance
  998:  */
  999: rpc_srv_t *
 1000: rpc_srv_initServer(u_char InstID, int concurentClients, int netBuf, 
 1001: 		const char *csHost, u_short Port, int proto)
 1002: {
 1003: 	int n = 1;
 1004: 	rpc_srv_t *srv = NULL;
 1005: 	sockaddr_t sa = E_SOCKADDR_INIT;
 1006: 
 1007: 	if (!concurentClients || (proto < 0 || proto > SOCK_DGRAM)) {
 1008: 		rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
 1009: 		return NULL;
 1010: 	}
 1011: 	if (!e_gethostbyname(csHost, Port, &sa))
 1012: 		return NULL;
 1013: 	if (!Port)
 1014: 		Port = RPC_DEFPORT;
 1015: 	if (!proto)
 1016: 		proto = SOCK_STREAM;
 1017: 	if (netBuf < RPC_MIN_BUFSIZ)
 1018: 		netBuf = BUFSIZ;
 1019: 	else
 1020: 		netBuf = E_ALIGN(netBuf, 2);	/* align netBuf length */
 1021: 
 1022: #ifdef HAVE_SRANDOMDEV
 1023: 	srandomdev();
 1024: #else
 1025: 	time_t tim;
 1026: 
 1027: 	srandom((time(&tim) ^ getpid()));
 1028: #endif
 1029: 
 1030: 	srv = e_malloc(sizeof(rpc_srv_t));
 1031: 	if (!srv) {
 1032: 		LOGERR;
 1033: 		return NULL;
 1034: 	} else
 1035: 		memset(srv, 0, sizeof(rpc_srv_t));
 1036: 
 1037: 	srv->srv_proto = proto;
 1038: 	srv->srv_netbuf = netBuf;
 1039: 	srv->srv_session.sess_version = RPC_VERSION;
 1040: 	srv->srv_session.sess_instance = InstID;
 1041: 
 1042: 	srv->srv_server.cli_parent = srv;
 1043: 	memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
 1044: 
 1045: 	/* init functions */
 1046: 	pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
 1047: 	SLIST_INIT(&srv->srv_funcs);
 1048: 	AVL_INIT(&srv->srv_funcs);
 1049: 
 1050: 	/* init scheduler */
 1051: 	srv->srv_root = schedBegin();
 1052: 	if (!srv->srv_root) {
 1053: 		rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
 1054: 		pthread_mutex_destroy(&srv->srv_funcs.mtx);
 1055: 		e_free(srv);
 1056: 		return NULL;
 1057: 	}
 1058: 
 1059: 	/* init pool for clients */
 1060: 	srv->srv_clients = array_Init(concurentClients);
 1061: 	if (!srv->srv_clients) {
 1062: 		rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
 1063: 		schedEnd(&srv->srv_root);
 1064: 		pthread_mutex_destroy(&srv->srv_funcs.mtx);
 1065: 		e_free(srv);
 1066: 		return NULL;
 1067: 	}
 1068: 
 1069: 	/* create server socket */
 1070: 	srv->srv_server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, srv->srv_proto, 0);
 1071: 	if (srv->srv_server.cli_sock == -1) {
 1072: 		LOGERR;
 1073: 		array_Destroy(&srv->srv_clients);
 1074: 		schedEnd(&srv->srv_root);
 1075: 		pthread_mutex_destroy(&srv->srv_funcs.mtx);
 1076: 		e_free(srv);
 1077: 		return NULL;
 1078: 	}
 1079: 	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
 1080: 		LOGERR;
 1081: 		goto err;
 1082: 	}
 1083: 	n = srv->srv_netbuf;
 1084: 	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
 1085: 		LOGERR;
 1086: 		goto err;
 1087: 	}
 1088: 	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
 1089: 		LOGERR;
 1090: 		goto err;
 1091: 	}
 1092: 	if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa, 
 1093: 				srv->srv_server.cli_sa.sa.sa_len) == -1) {
 1094: 		LOGERR;
 1095: 		goto err;
 1096: 	} else
 1097: 		fcntl(srv->srv_server.cli_sock, F_SETFL, 
 1098: 				fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
 1099: 
 1100: 	rpc_register_srvPing(srv);
 1101: 
 1102: 	return srv;
 1103: err:	/* error condition */
 1104: 	close(srv->srv_server.cli_sock);
 1105: 	array_Destroy(&srv->srv_clients);
 1106: 	schedEnd(&srv->srv_root);
 1107: 	pthread_mutex_destroy(&srv->srv_funcs.mtx);
 1108: 	e_free(srv);
 1109: 	return NULL;
 1110: }
 1111: 
 1112: /*
 1113:  * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
 1114:  *
 1115:  * @psrv = RPC Server instance
 1116:  * return: none
 1117:  */
 1118: void
 1119: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
 1120: {
 1121: 	if (!psrv || !*psrv)
 1122: 		return;
 1123: 
 1124: 	/* if send kill to blob server */
 1125: 	rpc_srv_endBLOBServer(*psrv);
 1126: 
 1127: 	(*psrv)->srv_kill = 1;
 1128: 	sleep(RPC_SCHED_POLLING);
 1129: 
 1130: 	schedEnd(&(*psrv)->srv_root);
 1131: 
 1132: 	pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx);
 1133: 	e_free(*psrv);
 1134: 	*psrv = NULL;
 1135: }
 1136: 
 1137: /*
 1138:  * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
 1139:  *
 1140:  * @srv = RPC Server instance
 1141:  * return: -1 error or 0 ok, infinite loop ...
 1142:  */
 1143: int
 1144: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
 1145: {
 1146: 	rpc_cli_t *c;
 1147: 	register int i;
 1148: 	rpc_func_t *f;
 1149: 	struct timespec ts = { RPC_SCHED_POLLING, 0 };
 1150: 
 1151: 	if (!srv) {
 1152: 		rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
 1153: 		return -1;
 1154: 	}
 1155: 
 1156: 	if (srv->srv_proto == SOCK_STREAM)
 1157: 		if (listen(srv->srv_server.cli_sock, array_Size(srv->srv_clients)) == -1) {
 1158: 			LOGERR;
 1159: 			return -1;
 1160: 		}
 1161: 
 1162: 	if (!schedRead(srv->srv_root, cbProto[srv->srv_proto][CB_ACCEPTCLIENT], srv, 
 1163: 				srv->srv_server.cli_sock, NULL, 0)) {
 1164: 		rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
 1165: 		return -1;
 1166: 	}
 1167: 
 1168: 	schedPolling(srv->srv_root, &ts, NULL);
 1169: 	/* main rpc loop */
 1170: 	schedRun(srv->srv_root, &srv->srv_kill);
 1171: 
 1172: 	/* close all clients connections & server socket */
 1173: 	for (i = 0; i < array_Size(srv->srv_clients); i++) {
 1174: 		c = array(srv->srv_clients, i, rpc_cli_t*);
 1175: 		if (c) {
 1176: 			shutdown(c->cli_sock, SHUT_RDWR);
 1177: 			close(c->cli_sock);
 1178: 
 1179: 			schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
 1180: 			ait_freeVars(&RPC_RETVARS(c));
 1181: 			AIT_FREE_VAL(&c->cli_buf);
 1182: 		}
 1183: 		array_Del(srv->srv_clients, i, 42);
 1184: 	}
 1185: 	array_Destroy(&srv->srv_clients);
 1186: 
 1187: 	close(srv->srv_server.cli_sock);
 1188: 
 1189: 	/* detach exported calls */
 1190: 	RPC_FUNCS_LOCK(&srv->srv_funcs);
 1191: 	while ((f = SLIST_FIRST(&srv->srv_funcs))) {
 1192: 		SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);
 1193: 
 1194: 		AIT_FREE_VAL(&f->func_name);
 1195: 		e_free(f);
 1196: 	}
 1197: 	srv->srv_funcs.avlh_root = NULL;
 1198: 	RPC_FUNCS_UNLOCK(&srv->srv_funcs);
 1199: 
 1200: 	return 0;
 1201: }
 1202: 
 1203: 
 1204: /*
 1205:  * rpc_srv_execCall() Execute registered call from RPC server
 1206:  *
 1207:  * @cli = RPC client
 1208:  * @rpc = IN RPC call structure
 1209:  * @funcname = Execute RPC function
 1210:  * @args = IN RPC calling arguments from RPC client
 1211:  * return: -1 error, !=-1 ok
 1212:  */
 1213: int
 1214: rpc_srv_execCall(rpc_cli_t * __restrict cli, struct tagRPCCall * __restrict rpc, 
 1215: 		ait_val_t funcname, array_t * __restrict args)
 1216: {
 1217: 	rpc_callback_t func;
 1218: 
 1219: 	if (!cli || !rpc || !AIT_ADDR(&funcname)) {
 1220: 		rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
 1221: 		return -1;
 1222: 	}
 1223: 
 1224: 	func = AIT_GET_LIKE(&funcname, rpc_callback_t);
 1225: 	return func(cli, rpc, args);
 1226: }

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