File:  [ELWIX - Embedded LightWeight unIX -] / libaitrpc / src / cli.c
Revision 1.9.2.15: download - view: text, annotated - select for diffs - revision graph
Thu May 17 22:57:42 2012 UTC (12 years, 1 month ago) by misho
Branches: rpc3_3
Diff to: branchpoint 1.9: preferred, unified
add into client execCall safety send polls when socket in async mode

    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: cli.c,v 1.9.2.15 2012/05/17 22:57:42 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
   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: /*
   50:  * rpc_cli_openBLOBClient() - Connect to BLOB Server
   51:  *
   52:  * @rpccli = RPC Client session
   53:  * @Port = Port for bind server, if Port == 0 default port is selected
   54:  * return: NULL == error or !=NULL connection to BLOB server established
   55:  */
   56: rpc_cli_t *
   57: rpc_cli_openBLOBClient(rpc_cli_t * __restrict rpccli, u_short Port)
   58: {
   59: 	rpc_cli_t *cli = NULL;
   60: 
   61: 	if (!rpccli) {
   62: 		rpc_SetErr(EINVAL, "Invalid parameters can`t connect to BLOB server");
   63: 		return NULL;
   64: 	}
   65: 
   66: 	cli = malloc(sizeof(rpc_cli_t));
   67: 	if (!cli) {
   68: 		LOGERR;
   69: 		return NULL;
   70: 	} else
   71: 		memcpy(cli, rpccli, sizeof(rpc_cli_t));
   72: 
   73: 	memcpy(&cli->cli_sa, &rpccli->cli_sa, sizeof(io_sockaddr_t));
   74: 	switch (cli->cli_sa.sa.sa_family) {
   75: 		case AF_INET:
   76: 			cli->cli_sa.sin.sin_port = 
   77: 				htons(Port ? Port : ntohs(cli->cli_sa.sin.sin_port) + 1);
   78: 			break;
   79: 		case AF_INET6:
   80: 			cli->cli_sa.sin6.sin6_port = 
   81: 				htons(Port ? Port : ntohs(cli->cli_sa.sin6.sin6_port) + 1);
   82: 			break;
   83: 		case AF_LOCAL:
   84: 			strlcat(cli->cli_sa.sun.sun_path, ".blob", sizeof cli->cli_sa.sun.sun_path);
   85: 			break;
   86: 		default:
   87: 			rpc_SetErr(EINVAL, "Invalid socket type %d", cli->cli_sa.sa.sa_family);
   88: 			return NULL;
   89: 	}
   90: 
   91: 	AIT_COPY_VAL(&cli->cli_buf, &rpccli->cli_buf);
   92: 
   93: 	/* connect to BLOB server */
   94: 	cli->cli_sock = socket(cli->cli_sa.sa.sa_family, SOCK_STREAM, 0);
   95: 	if (cli->cli_sock == -1) {
   96: 		LOGERR;
   97: 		free(cli);
   98: 		return NULL;
   99: 	}
  100: 	if (connect(cli->cli_sock, &cli->cli_sa.sa, cli->cli_sa.sa.sa_len) == -1) {
  101: 		LOGERR;
  102: 		close(cli->cli_sock);
  103: 		free(cli);
  104: 		return NULL;
  105: 	} else
  106: 		fcntl(cli->cli_sock, F_SETFL, fcntl(cli->cli_sock, F_GETFL) | O_NONBLOCK);
  107: 
  108: 	return cli;
  109: }
  110: 
  111: /*
  112:  * rpc_cli_closeBLOBClient() - Close connection to BLOB server and free resources
  113:  *
  114:  * @cli = BLOB Client session
  115:  * return: none
  116:  */
  117: void
  118: rpc_cli_closeBLOBClient(rpc_cli_t ** __restrict cli)
  119: {
  120: 	if (!cli || !*cli)
  121: 		return;
  122: 
  123: 	shutdown((*cli)->cli_sock, SHUT_RDWR);
  124: 	close((*cli)->cli_sock);
  125: 
  126: 	AIT_FREE_VAL(&(*cli)->cli_buf);
  127: 
  128: 	free(*cli);
  129: 	*cli = NULL;
  130: }
  131: 
  132: /* -------------------------------------------------------------- */
  133: 
  134: /*
  135:  * rpc_cli_openClient() - Connect to RPC Server
  136:  *
  137:  * @ProgID = ProgramID for RPC session request
  138:  * @ProcID = ProcessID for RPC session request
  139:  * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
  140:  * @csHost = Host name or IP address for bind server
  141:  * @Port = Port for bind server, if Port == 0 default port is selected
  142:  * return: NULL == error or !=NULL connection to RPC server established
  143:  */
  144: rpc_cli_t *
  145: rpc_cli_openClient(u_int ProgID, u_char ProcID, int netBuf, const char *csHost, u_short Port)
  146: {
  147: 	rpc_cli_t *cli = NULL;
  148: 	io_sockaddr_t sa;
  149: 
  150: 	if (!io_gethostbyname(csHost, Port, &sa))
  151: 		return NULL;
  152: 	if (!Port)
  153: 		Port = RPC_DEFPORT;
  154: 	if (netBuf < RPC_MIN_BUFSIZ)
  155: 		netBuf = BUFSIZ;
  156: 	else
  157: 		netBuf = io_align(netBuf, 1);	/* align netBuf length */
  158: 
  159: #ifdef HAVE_SRANDOMDEV
  160: 	srandomdev();
  161: #else
  162: 	time_t tim;
  163: 
  164: 	srandom((time(&tim) ^ getpid()));
  165: #endif
  166: 
  167: 	cli = malloc(sizeof(rpc_cli_t));
  168: 	if (!cli) {
  169: 		LOGERR;
  170: 		return NULL;
  171: 	} else
  172: 		memset(cli, 0, sizeof(rpc_cli_t));
  173: 
  174: 	/* build session */
  175: 	cli->cli_parent = malloc(sizeof(rpc_sess_t));
  176: 	if (!cli->cli_parent) {
  177: 		LOGERR;
  178: 		free(cli);
  179: 		return NULL;
  180: 	} else {
  181: 		((rpc_sess_t*) cli->cli_parent)->sess_version = RPC_VERSION;
  182: 		((rpc_sess_t*) cli->cli_parent)->sess_program = ProgID;
  183: 		((rpc_sess_t*) cli->cli_parent)->sess_process = ProcID;
  184: 	}
  185: 
  186: 	memcpy(&cli->cli_sa, &sa, sizeof cli->cli_sa);
  187: 	AIT_SET_BUF2(&cli->cli_buf, 0, netBuf);
  188: 
  189: 	/* connect to RPC server */
  190: 	cli->cli_sock = socket(cli->cli_sa.sa.sa_family, SOCK_STREAM, 0);
  191: 	if (cli->cli_sock == -1) {
  192: 		LOGERR;
  193: 		AIT_FREE_VAL(&cli->cli_buf);
  194: 		free(cli->cli_parent);
  195: 		free(cli);
  196: 		return NULL;
  197: 	}
  198: 	if (connect(cli->cli_sock, &cli->cli_sa.sa, cli->cli_sa.sa.sa_len) == -1) {
  199: 		LOGERR;
  200: 		AIT_FREE_VAL(&cli->cli_buf);
  201: 		close(cli->cli_sock);
  202: 		free(cli->cli_parent);
  203: 		free(cli);
  204: 		return NULL;
  205: 	} else
  206: 		fcntl(cli->cli_sock, F_SETFL, fcntl(cli->cli_sock, F_GETFL) | O_NONBLOCK);
  207: 
  208: 	return cli;
  209: }
  210: 
  211: /*
  212:  * rpc_cli_closeClient() - Close connection to RPC server and free resources
  213:  *
  214:  * @cli = RPC Client session
  215:  * return: none
  216:  */
  217: void
  218: rpc_cli_closeClient(rpc_cli_t ** __restrict cli)
  219: {
  220: 	if (!cli || !*cli)
  221: 		return;
  222: 
  223: 	shutdown((*cli)->cli_sock, SHUT_RDWR);
  224: 	close((*cli)->cli_sock);
  225: 
  226: 	AIT_FREE_VAL(&(*cli)->cli_buf);
  227: 
  228: 	if ((*cli)->cli_parent)
  229: 		free((*cli)->cli_parent);
  230: 
  231: 	free(*cli);
  232: 	*cli = NULL;
  233: }
  234: 
  235: 
  236: /*
  237:  * rpc_cli_execCall() - Execute RPC call
  238:  *
  239:  * @cli = RPC Client session
  240:  * @noreply = We not want RPC reply
  241:  * @tag = Function tag for execution
  242:  * @in_vars = IN RPC call array of rpc values, may be NULL
  243:  * @out_vars = OUT returned array of rpc values, if !=NULL must be free after use with io_freeVars()
  244:  * return: -1 error or != -1 ok result
  245:  */
  246: int
  247: rpc_cli_execCall(rpc_cli_t *cli, int noreply, u_short tag, 
  248: 		array_t * __restrict in_vars, array_t ** __restrict out_vars)
  249: {
  250: 	struct tagRPCCall *rpc;
  251: 	int ret = 0, wlen = sizeof(struct tagRPCCall);
  252: 	uint16_t crc;
  253: 	u_char *buf;
  254: 	struct pollfd pfd;
  255: 
  256: 	if (!cli) {
  257: 		rpc_SetErr(EINVAL, "Can`t execute call because parameter is null or invalid!");
  258: 		return -1;
  259: 	} else
  260: 		buf = AIT_GET_BUF(&cli->cli_buf);
  261: 	if (out_vars)
  262: 		*out_vars = NULL;
  263: 
  264: 	/* prepare RPC call */
  265: 	rpc = (struct tagRPCCall*) buf;
  266: 	rpc_addPktSession(&rpc->call_session, cli->cli_parent);
  267: 	rpc->call_argc = htons(io_arraySize(in_vars));
  268: 	rpc->call_tag = htons(tag);
  269: 	rpc->call_seq = htons(random() % USHRT_MAX);
  270: 
  271: 	/* set reply */
  272: 	rpc->call_req.flags = noreply ? RPC_NOREPLY : RPC_REPLY;
  273: 
  274: 	if (io_arraySize(in_vars)) {
  275: 		/* marshaling variables */
  276: 		ret = io_vars2buffer(buf + wlen, AIT_LEN(&cli->cli_buf) - wlen, in_vars);
  277: 		if (ret == -1) {
  278: 			rpc_SetErr(EBADRPC, "Failed to prepare RPC packet values");
  279: 			return -1;
  280: 		} else
  281: 			wlen += ret;
  282: 	}
  283: 
  284: 	/* total packet length */
  285: 	rpc->call_len = htons(wlen);
  286: 
  287: 	/* calculate CRC */
  288: 	rpc->call_crc ^= rpc->call_crc;
  289: 	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
  290: 
  291: 	pfd.fd = cli->cli_sock;
  292: 	pfd.events = POLLOUT;
  293: 	do {
  294: 		if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
  295: 				pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
  296: 			if (ret) {
  297: 				if (errno == EAGAIN)
  298: 					continue;
  299: 				else
  300: 					LOGERR;
  301: 			} else
  302: 				rpc_SetErr(ETIMEDOUT, "Timeout, can't send to RPC server");
  303: 			return -1;
  304: 		}
  305: 	} while (0);
  306: 	if ((ret = send(cli->cli_sock, buf, wlen, MSG_NOSIGNAL)) == -1) {
  307: 		LOGERR;
  308: 		return -1;
  309: 	} else if (ret != wlen) {
  310: 		rpc_SetErr(EPROCUNAVAIL, "RPC request, should be send %d bytes, "
  311: 				"really sended %d bytes", wlen, ret);
  312: 		return -1;
  313: 	}
  314: 
  315: 	if (noreply)	/* we not want reply */
  316: 		return 0;
  317: 
  318: 	/* reply from RPC server */
  319: 	pfd.events = POLLIN | POLLPRI;
  320: 	do {
  321: 		if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
  322: 				pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
  323: 			if (ret) {
  324: 				if (errno == EAGAIN)
  325: 					continue;
  326: 				else
  327: 					LOGERR;
  328: 			} else
  329: 				rpc_SetErr(ETIMEDOUT, "Timeout, no answer from RPC server");
  330: 			return -1;
  331: 		}
  332: 	} while (0);
  333: 	memset(buf, 0, AIT_LEN(&cli->cli_buf));
  334: 	if ((ret = recv(cli->cli_sock, buf, AIT_LEN(&cli->cli_buf), 0)) < 1) {
  335: 		if (ret)
  336: 			LOGERR;
  337: 		return -1;
  338: 	}
  339: 	if (ret < sizeof(struct tagRPCCall)) {
  340: 		rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
  341: 		return -1;
  342: 	}
  343: 
  344: 	/* calculate CRC */
  345: 	crc = ntohs(rpc->call_crc);
  346: 	rpc->call_crc ^= rpc->call_crc;
  347: 	if (crc != crcFletcher16((u_short*) buf, ret / 2)) {
  348: 		rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
  349: 		return -1;
  350: 	}
  351: 
  352: 	/* check RPC packet session info */
  353: 	if (rpc_chkPktSession(&rpc->call_session, cli->cli_parent)) {
  354: 		rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
  355: 		return -1;
  356: 	} else
  357: 		wlen = sizeof(struct tagRPCCall);
  358: 	if (ntohs(rpc->call_tag) != tag) {
  359: 		rpc_SetErr(ERPCMISMATCH, "Get wrong RPC reply");
  360: 		return -1;
  361: 	}
  362: 	if (ntohl(rpc->call_rep.eno) && ntohl(rpc->call_rep.ret) == -1) {
  363: 		rpc_SetErr(ntohl(rpc->call_rep.eno), "Server side: retcode=%d #%d %s", 
  364: 				ntohl(rpc->call_rep.ret), ntohl(rpc->call_rep.eno), 
  365: 				strerror(ntohl(rpc->call_rep.eno)));
  366: 		return -1;
  367: 	}
  368: 	if (ntohs(rpc->call_argc) * sizeof(ait_val_t) > AIT_LEN(&cli->cli_buf) - wlen) {
  369: 		rpc_SetErr(EMSGSIZE, "Reply RPC packet is too long ...");
  370: 		return -1;
  371: 	}
  372: 
  373: 	/* RPC is OK! Go de-marshaling variables ... */
  374: 	if (out_vars && ntohs(rpc->call_argc)) {
  375: 		*out_vars = io_buffer2vars(buf + wlen, AIT_LEN(&cli->cli_buf) - wlen, 
  376: 				ntohs(rpc->call_argc), 0);
  377: 		if (!*out_vars) {
  378: 			rpc_SetErr(io_GetErrno(), "%s", io_GetError());
  379: 			return -1;
  380: 		}
  381: 	}
  382: 
  383: 	ret = ntohl(rpc->call_rep.ret);
  384: 	return ret;
  385: }
  386: 
  387: /*
  388:  * rpc_cli_ping() - Ping RPC server
  389:  *
  390:  * @cli = connected client
  391:  * return: -1 error or !=-1 ping seq id
  392:  */
  393: inline int
  394: rpc_cli_ping(rpc_cli_t *cli)
  395: {
  396: 	int ret = 0;
  397: 	array_t *arr = NULL;
  398: 
  399: 	if (!cli)
  400: 		return -1;
  401: 
  402: 	if (rpc_cli_execCall(cli, RPC_REPLY, CALL_SRVPING, NULL, &arr))
  403: 		return -1;
  404: 	else
  405: 		ret = AIT_GET_U16(io_getVars(arr, 0));
  406: 	io_freeVars(&arr);
  407: 
  408: 	return ret;
  409: }

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