--- libaitrpc/src/cli.c 2011/08/18 15:08:03 1.3 +++ libaitrpc/src/cli.c 2012/05/14 08:39:05 1.9 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: cli.c,v 1.3 2011/08/18 15:08:03 misho Exp $ +* $Id: cli.c,v 1.9 2012/05/14 08:39:05 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -12,7 +12,7 @@ terms: All of the documentation and software included in the ELWIX and AITNET Releases is copyrighted by ELWIX - Sofia/Bulgaria -Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011 +Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -47,7 +47,8 @@ SUCH DAMAGE. /* - * rpc_cli_openBLOBClient() Connect to BLOB Server + * rpc_cli_openBLOBClient() - Connect to BLOB Server + * * @rpccli = RPC Client session * @Port = Port for bind server, if Port == 0 default port is selected * return: NULL == error or !=NULL connection to BLOB server established @@ -56,15 +57,15 @@ rpc_cli_t * rpc_cli_openBLOBClient(rpc_cli_t * __restrict rpccli, u_short Port) { rpc_cli_t *cli = NULL; - struct sockaddr_in sin; - struct sockaddr_in6 sin6; + io_sockaddr_t sa; + int n; - if (!rpccli || (rpccli->cli_sa.sa_family != AF_INET && rpccli->cli_sa.sa_family != AF_INET6)) { - rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t connect to BLOB server ...\n"); + if (!rpccli || + (rpccli->cli_sa.sa.sa_family != AF_INET && rpccli->cli_sa.sa.sa_family != AF_INET6 && + rpccli->cli_sa.sa.sa_family != AF_LOCAL)) { + rpc_SetErr(EINVAL, "Invalid parameters can`t connect to BLOB server"); return NULL; } - if (!Port) - Port = RPC_DEFPORT + 1; cli = malloc(sizeof(rpc_cli_t)); if (!cli) { @@ -73,33 +74,53 @@ rpc_cli_openBLOBClient(rpc_cli_t * __restrict rpccli, } else memcpy(cli, rpccli, sizeof(rpc_cli_t)); - if (rpccli->cli_sa.sa_family == AF_INET) { - memcpy(&sin, &rpccli->cli_sa, sizeof sin); - sin.sin_port = htons(Port); - memcpy(&cli->cli_sa, &sin, sizeof(struct sockaddr)); - } else { - memcpy(&sin6, &rpccli->cli_sa, sizeof sin6); - sin6.sin6_port = htons(Port); - memcpy(&cli->cli_sa, &sin6, sizeof(struct sockaddr)); + memcpy(&sa, &rpccli->cli_sa, sizeof sa); + switch (sa.sa.sa_family) { + case AF_INET: + sa.sin.sin_port = htons(Port ? Port : ntohs(sa.sin.sin_port) + 1); + break; + case AF_INET6: + sa.sin6.sin6_port = htons(Port ? Port : ntohs(sa.sin6.sin6_port) + 1); + break; + case AF_LOCAL: + strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path); + break; } + memcpy(&cli->cli_sa, &sa, sizeof sa); - cli->cli_sock = socket(cli->cli_sa.sa_family, SOCK_STREAM, 0); + /* connect to BLOB server */ + cli->cli_sock = socket(cli->cli_sa.sa.sa_family, SOCK_STREAM, 0); if (cli->cli_sock == -1) { LOGERR; free(cli); return NULL; } - if (connect(cli->cli_sock, &cli->cli_sa, sizeof cli->cli_sa) == -1) { + n = cli->cli_netbuf; + if (setsockopt(cli->cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) { LOGERR; + close(cli->cli_sock); free(cli); return NULL; } + if (setsockopt(cli->cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) { + LOGERR; + close(cli->cli_sock); + free(cli); + return NULL; + } + if (connect(cli->cli_sock, &cli->cli_sa.sa, cli->cli_sa.sa.sa_len) == -1) { + LOGERR; + close(cli->cli_sock); + free(cli); + return NULL; + } return cli; } /* - * rpc_cli_closeBLOBClient() Close connection to BLOB server and free resources + * rpc_cli_closeBLOBClient() - Close connection to BLOB server and free resources + * * @cli = BLOB Client session * return: none */ @@ -107,7 +128,7 @@ void rpc_cli_closeBLOBClient(rpc_cli_t * __restrict cli) { if (!cli) { - rpc_SetErr(EINVAL, "Error:: Can`t close connection because parameter is null!\n"); + rpc_SetErr(EINVAL, "Can`t close connection because parameter is null!"); return; } @@ -121,54 +142,65 @@ rpc_cli_closeBLOBClient(rpc_cli_t * __restrict cli) // -------------------------------------------------------------- /* - * rpc_cli_openClient() Connect to RPC Server + * rpc_cli_openClient() - Connect to RPC Server + * * @ProgID = ProgramID for RPC session request * @ProcID = ProcessID for RPC session request + * @netBuf = Network buffer length, if =0 == BUFSIZ (also meaning max RPC packet) + * @Timeout = RPC timeout in seconds, if =0 set default RPC timeout * @family = Family socket type, AF_INET or AF_INET6 * @csHost = Host name or IP address for bind server * @Port = Port for bind server, if Port == 0 default port is selected * return: NULL == error or !=NULL connection to RPC server established */ rpc_cli_t * -rpc_cli_openClient(u_int ProgID, u_int ProcID, u_short family, const char *csHost, u_short Port) +rpc_cli_openClient(u_int ProgID, u_int ProcID, int netBuf, u_char Timeout, + u_short family, const char *csHost, u_short Port) { rpc_cli_t *cli = NULL; struct hostent *host = NULL; - struct sockaddr_in sin; - struct sockaddr_in6 sin6; + io_sockaddr_t sa; + int n; - if (!csHost || (family != AF_INET && family != AF_INET6)) { - rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t connect to RPC server ...\n"); + if (!csHost || (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) { + rpc_SetErr(EINVAL, "Invalid parameters can`t connect to RPC server ..."); return NULL; } if (!Port) Port = RPC_DEFPORT; - if (csHost) { + if (!netBuf) + netBuf = BUFSIZ; + if (!Timeout) + Timeout = DEF_RPC_TIMEOUT; + if (csHost && family != AF_LOCAL) { host = gethostbyname2(csHost, family); if (!host) { - rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno)); + rpc_SetErr(h_errno, "%s", hstrerror(h_errno)); return NULL; } } + memset(&sa, 0, sizeof sa); + sa.sa.sa_family = family; switch (family) { case AF_INET: - memset(&sin, 0, sizeof sin); - sin.sin_len = sizeof sin; - sin.sin_family = family; - sin.sin_port = htons(Port); + sa.sin.sin_len = sizeof(struct sockaddr_in); + sa.sin.sin_port = htons(Port); if (csHost) - memcpy(&sin.sin_addr, host->h_addr, host->h_length); + memcpy(&sa.sin.sin_addr, host->h_addr, host->h_length); break; case AF_INET6: - memset(&sin6, 0, sizeof sin6); - sin6.sin6_len = sizeof sin6; - sin6.sin6_family = family; - sin6.sin6_port = htons(Port); + sa.sin6.sin6_len = sizeof(struct sockaddr_in6); + sa.sin6.sin6_port = htons(Port); if (csHost) - memcpy(&sin6.sin6_addr, host->h_addr, host->h_length); + memcpy(&sa.sin6.sin6_addr, host->h_addr, host->h_length); break; + case AF_LOCAL: + sa.sun.sun_len = sizeof(struct sockaddr_un); + if (csHost) + strlcpy(sa.sun.sun_path, csHost, sizeof sa.sun.sun_path); + break; default: - rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t connect to RPC server ...\n"); + rpc_SetErr(EINVAL, "Invalid parameters can`t connect to RPC server ..."); return NULL; } @@ -178,6 +210,8 @@ rpc_cli_openClient(u_int ProgID, u_int ProcID, u_short return NULL; } else memset(cli, 0, sizeof(rpc_cli_t)); + + cli->cli_netbuf = netBuf; cli->cli_parent = malloc(sizeof(rpc_sess_t)); if (!cli->cli_parent) { LOGERR; @@ -185,14 +219,14 @@ rpc_cli_openClient(u_int ProgID, u_int ProcID, u_short return NULL; } else { ((rpc_sess_t*) cli->cli_parent)->sess_version = RPC_VERSION; + ((rpc_sess_t*) cli->cli_parent)->sess_timeout = Timeout; ((rpc_sess_t*) cli->cli_parent)->sess_program = ProgID; ((rpc_sess_t*) cli->cli_parent)->sess_process = ProcID; } - if (family == AF_INET) - memcpy(&cli->cli_sa, &sin, sizeof cli->cli_sa); - else - memcpy(&cli->cli_sa, &sin6, sizeof cli->cli_sa); + memcpy(&cli->cli_sa, &sa, sizeof cli->cli_sa); + + /* connect to RPC server */ cli->cli_sock = socket(family, SOCK_STREAM, 0); if (cli->cli_sock == -1) { LOGERR; @@ -200,18 +234,35 @@ rpc_cli_openClient(u_int ProgID, u_int ProcID, u_short free(cli); return NULL; } - if (connect(cli->cli_sock, &cli->cli_sa, sizeof cli->cli_sa) == -1) { + n = cli->cli_netbuf; + if (setsockopt(cli->cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) { LOGERR; + close(cli->cli_sock); free(cli->cli_parent); free(cli); return NULL; } + if (setsockopt(cli->cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) { + LOGERR; + close(cli->cli_sock); + free(cli->cli_parent); + free(cli); + return NULL; + } + if (connect(cli->cli_sock, &cli->cli_sa.sa, cli->cli_sa.sa.sa_len) == -1) { + LOGERR; + close(cli->cli_sock); + free(cli->cli_parent); + free(cli); + return NULL; + } return cli; } /* - * rpc_cli_closeClient() Close connection to RPC server and free resources + * rpc_cli_closeClient() - Close connection to RPC server and free resources + * * @cli = RPC Client session * return: none */ @@ -219,7 +270,7 @@ void rpc_cli_closeClient(rpc_cli_t * __restrict cli) { if (!cli) { - rpc_SetErr(EINVAL, "Error:: Can`t close connection because parameter is null!\n"); + rpc_SetErr(EINVAL, "Can`t close connection because parameter is null!"); return; } @@ -234,220 +285,170 @@ rpc_cli_closeClient(rpc_cli_t * __restrict cli) /* - * rpc_cli_execCall() Execute RPC call + * rpc_cli_execCall() - Execute RPC call + * * @cli = RPC Client session + * @noreply = We not want RPC reply * @csModule = Module name, if NULL self binary * @csFunc = Function name for execute - * @in_argc = IN count of arguments - * @in_vals = IN RPC call array of rpc values - * @out_argc = OUT returned count of arguments - * @out_vals = OUT returned array of rpc values, must be free after use (see rpc_cli_freeVals()) + * @in_vars = IN RPC call array of rpc values + * @out_vars = OUT returned array of rpc values, must be free after use with rpc_cli_freeVals() * return: -1 error or != -1 ok result */ int -rpc_cli_execCall(rpc_cli_t *cli, const char *csModule, const char *csFunc, int in_argc, - rpc_val_t * __restrict in_vals, int *out_argc, rpc_val_t ** __restrict out_vals) +rpc_cli_execCall(rpc_cli_t *cli, int noreply, const char *csModule, const char *csFunc, + array_t * __restrict in_vars, array_t ** __restrict out_vars) { - fd_set fds; - u_char buf[BUFSIZ], str[MAXPATHLEN + UCHAR_MAX + 1], *data; - struct tagRPCCall *rpc = (struct tagRPCCall*) buf; - struct tagRPCRet *rrpc = NULL; - int ret = 0, Limit = 0; - register int i; - rpc_val_t *v; - struct timeval tv = { DEF_RPC_TIMEOUT, 0 }; + u_char *buf; + rpc_func_t func; + struct tagRPCCall *rpc; + int ret = 0, wlen = sizeof(struct tagRPCCall); + uint16_t tag, crc; + uint32_t hash; + struct pollfd pfd; - FD_ZERO(&fds); - memset(buf, 0, BUFSIZ); - memset(str, 0, MAXPATHLEN + UCHAR_MAX + 1); - if (!cli || !csFunc || (in_argc && !in_vals)) { - rpc_SetErr(EINVAL, "Error:: Can`t execute call because parameter is null or invalid!\n"); + if (!cli || !csFunc) { + rpc_SetErr(EINVAL, "Can`t execute call because parameter is null or invalid!"); return -1; } - if (csModule) - strlcpy((char*) str, csModule, MAXPATHLEN + UCHAR_MAX + 1); - strlcat((char*) str, "__", MAXPATHLEN + UCHAR_MAX + 1); - strlcat((char*) str, csFunc, MAXPATHLEN + UCHAR_MAX + 1); + if (out_vars) + *out_vars = NULL; - memcpy(&rpc->call_session, cli->cli_parent, sizeof rpc->call_session); - rpc->call_argc = in_argc; - rpc->call_tag = crcFletcher16((u_short*) str, (MAXPATHLEN + UCHAR_MAX + 1) / 2); - rpc->call_hash = hash_fnv((char*) str, MAXPATHLEN + UCHAR_MAX + 1); - Limit = sizeof(struct tagRPCCall); - if (in_argc) { - v = (rpc_val_t*) (buf + sizeof(struct tagRPCCall)); - if (in_argc * sizeof(rpc_val_t) > BUFSIZ - Limit) { - rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet values (-7) ...\n"); - return -7; - } else - Limit += in_argc * sizeof(rpc_val_t); - memcpy(v, in_vals, in_argc * sizeof(rpc_val_t)); - data = (u_char*) v + in_argc * sizeof(rpc_val_t); - for (i = 0; i < in_argc; i++) { - switch (in_vals[i].val_type) { - case buffer: - if (Limit + in_vals[i].val_len > BUFSIZ) { - ret = -7; - break; - } + buf = malloc(cli->cli_netbuf); + if (!buf) { + LOGERR; + return -1; + } else + memset(buf, 0, cli->cli_netbuf); - memcpy(data, in_vals[i].val.buffer, in_vals[i].val_len); - v[i].val.buffer = (uint8_t*) ((void*) data - (void*) v); - data += in_vals[i].val_len; - Limit += in_vals[i].val_len; - break; - case string: - if (Limit + in_vals[i].val_len > BUFSIZ) { - ret = -7; - break; - } + /* calculate hashes */ + if (rpc_calcHashes(&func, csModule, csFunc) == -1) + return -1; + else { + tag = htons(func.func_tag); + hash = htonl(func.func_hash); + AIT_FREE_VAL(&func.func_name); + AIT_FREE_VAL(&func.func_file); + } - memcpy(data, in_vals[i].val.string, in_vals[i].val_len); - v[i].val.string = (int8_t*) ((void*) data - (void*) v); - data += in_vals[i].val_len; - Limit += in_vals[i].val_len; - break; - default: - break; - } - } - if (ret < 0) { - rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n"); - return ret; - } + /* prepare RPC call */ + rpc = (struct tagRPCCall*) buf; + rpc_addPktSession(&rpc->call_session, cli->cli_parent); + rpc->call_argc = htons(io_arraySize(in_vars)); + rpc->call_tag = tag; + rpc->call_hash = hash; + + /* set reply */ + rpc->call_req.flags = noreply ? RPC_NOREPLY : RPC_REPLY; + + if (io_arraySize(in_vars)) { + /* marshaling variables */ + ret = io_vars2buffer(buf + wlen, cli->cli_netbuf - wlen, in_vars); + if (ret == -1) { + rpc_SetErr(EBADRPC, "Failed to prepare RPC packet values"); + free(buf); + return -1; + } else + wlen += ret; } - if ((ret = send(cli->cli_sock, buf, Limit, 0)) == -1) { + + rpc->call_len = htons(wlen); + + /* calculate CRC */ + rpc->call_crc ^= rpc->call_crc; + rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2)); + + if ((ret = send(cli->cli_sock, buf, wlen, 0)) == -1) { LOGERR; + free(buf); return -1; + } else if (ret != wlen) { + rpc_SetErr(EPROCUNAVAIL, "RPC request, should be send %d bytes, " + "really sended %d bytes", wlen, ret); + free(buf); + return -1; } - if (ret != Limit) { - rpc_SetErr(ECANCELED, "Error:: in send RPC request, should be send %d bytes, really is %d\n", - Limit, ret); - return -9; + + if (noreply) { + /* we not want reply */ + free(buf); + return 0; } - FD_SET(cli->cli_sock, &fds); - if ((ret = select(cli->cli_sock + 1, &fds, NULL, NULL, &tv)) < 1) { - if (ret) { + /* reply from RPC server */ + pfd.fd = cli->cli_sock; + pfd.events = POLLIN | POLLPRI; + if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) == -1 || + pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { + if (ret) LOGERR; - } else - rpc_SetErr(ETIMEDOUT, "Error:: timeout, no return from RPC server?\n"); + else + rpc_SetErr(ETIMEDOUT, "Timeout, no return from RPC server?"); + + free(buf); return -1; } - memset(buf, 0, BUFSIZ); - if ((ret = recv(cli->cli_sock, buf, BUFSIZ, 0)) == -1) { + memset(buf, 0, cli->cli_netbuf); + if ((ret = recv(cli->cli_sock, buf, cli->cli_netbuf, 0)) == -1) { LOGERR; - return -3; + free(buf); + return -1; } - if (!ret) /* receive EOF! */ + if (!ret) { /* receive EOF! */ + free(buf); return 0; - if (ret < sizeof(struct tagRPCRet)) { - rpc_SetErr(EMSGSIZE, "Error:: too short RPC packet ...\n"); - return -4; - } else - rrpc = (struct tagRPCRet*) buf; + } + if (ret < sizeof(struct tagRPCCall)) { + rpc_SetErr(ERPCMISMATCH, "Too short RPC packet ..."); + free(buf); + return -1; + } + + /* calculate CRC */ + crc = ntohs(rpc->call_crc); + rpc->call_crc ^= rpc->call_crc; + if (crc != crcFletcher16((u_short*) buf, ret / 2)) { + rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet"); + free(buf); + return -1; + } + /* check RPC packet session info */ - if (memcmp(&rrpc->ret_session, cli->cli_parent, sizeof rrpc->ret_session)) { - rpc_SetErr(EINVAL, "Error:: get invalid RPC session ...\n"); - return -5; + if (rpc_chkPktSession(&rpc->call_session, cli->cli_parent)) { + rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session ..."); + free(buf); + return -1; } else - Limit = sizeof(struct tagRPCRet); - if (rrpc->ret_retcode < 0 && rrpc->ret_errno) { - rpc_SetErr(rrpc->ret_errno, "Error::Server side: %d %s\n", - rrpc->ret_retcode, strerror(rrpc->ret_errno)); - return -6; + wlen = sizeof(struct tagRPCCall); + if (rpc->call_tag != tag || rpc->call_hash != hash) { + rpc_SetErr(ERPCMISMATCH, "Get wrong RPC reply ..."); + free(buf); + return -1; } - if (rrpc->ret_argc * sizeof(rpc_val_t) > BUFSIZ - Limit) { - rpc_SetErr(EMSGSIZE, "Error:: reply RPC packet is too big ...\n"); - return -7; - } else - Limit += rrpc->ret_argc * sizeof(rpc_val_t); - /* RPC is OK! Go decapsulate variables ... */ - if (rrpc->ret_argc) { - *out_argc = rrpc->ret_argc; - *out_vals = calloc(rrpc->ret_argc, sizeof(rpc_val_t)); - if (!*out_vals) { - LOGERR; - *out_argc = 0; - return -1; - } else - memcpy(*out_vals, buf + sizeof(struct tagRPCRet), Limit - sizeof(struct tagRPCRet)); - /* RPC received variables types OK! */ - data = (u_char*) buf + Limit; - for (i = 0; i < rrpc->ret_argc; i++) - switch ((*out_vals)[i].val_type) { - case buffer: - if ((*out_vals)[i].val_len > BUFSIZ - Limit) { - rpc_SetErr(EMSGSIZE, "Error:: Too big RPC packet ...\n"); - free(*out_vals); - *out_vals = NULL; - *out_argc = 0; - return -7; - } else - Limit += (*out_vals)[i].val_len; + if (ntohl(rpc->call_rep.ret) < 0 && ntohl(rpc->call_rep.eno)) { + rpc_SetErr(ntohl(rpc->call_rep.eno), "Server side: retcode=%d #%d %s", + ntohl(rpc->call_rep.ret), ntohl(rpc->call_rep.eno), + strerror(ntohl(rpc->call_rep.eno))); + free(buf); + return -1; + } + if (ntohs(rpc->call_argc) * sizeof(ait_val_t) > cli->cli_netbuf - wlen) { + rpc_SetErr(EMSGSIZE, "Reply RPC packet is too long ..."); + free(buf); + return -1; + } - (*out_vals)[i].val.buffer = malloc((*out_vals)[i].val_len); - if (!(*out_vals)[i].val.buffer) { - rpc_SetErr(errno, "Error:: in prepare RPC reply ...\n"); - free(*out_vals); - *out_vals = NULL; - *out_argc = 0; - return -1; - } else - memcpy((*out_vals)[i].val.buffer, data, (*out_vals)[i].val_len); - data += (*out_vals)[i].val_len; - break; - case string: - if ((*out_vals)[i].val_len > BUFSIZ - Limit) { - rpc_SetErr(EMSGSIZE, "Error:: Too big RPC packet ...\n"); - free(*out_vals); - *out_vals = NULL; - *out_argc = 0; - return -7; - } else - Limit += (*out_vals)[i].val_len; - - (*out_vals)[i].val.string = malloc((*out_vals)[i].val_len); - if (!(*out_vals)[i].val.string) { - rpc_SetErr(errno, "Error:: in prepare RPC reply ...\n"); - free(*out_vals); - *out_vals = NULL; - *out_argc = 0; - return -1; - } else - memcpy((*out_vals)[i].val.string, data, (*out_vals)[i].val_len); - data += (*out_vals)[i].val_len; - break; - default: - break; - } - } else { - if (out_argc) - *out_argc = 0; - if (out_vals) - *out_vals = NULL; + /* RPC is OK! Go de-marshaling variables ... */ + if (ntohs(rpc->call_argc)) { + *out_vars = io_buffer2vars(buf + wlen, cli->cli_netbuf - wlen, + ntohs(rpc->call_argc), 0); + if (!*out_vars) { + free(buf); + return -1; + } } - return rrpc->ret_retcode; -} - -/* - * rpc_cli_freeVals() Free rpc_val_t array returned from RPC call - * @args = Number of arguments in array - * @vals = Value elements - * return: none - */ -inline void -rpc_cli_freeVals(int args, rpc_val_t *vals) -{ - register int i; - - if (!vals) - return; - - for (i = 0; i < args; i++) - RPC_FREE_VAL(&vals[i]); - - free(vals); - vals = NULL; + ret = ntohl(rpc->call_rep.ret); + free(buf); + return ret; }