--- libaitrpc/src/srv.c 2013/09/05 00:43:05 1.20 +++ libaitrpc/src/srv.c 2014/11/17 23:28:55 1.23.6.1 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: srv.c,v 1.20 2013/09/05 00:43:05 misho Exp $ +* $Id: srv.c,v 1.23.6.1 2014/11/17 23:28:55 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, 2012, 2013 +Copyright 2004 - 2014 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -66,7 +66,11 @@ static sched_task_func_t cbProto[SOCK_RAW + 1][4] = { { NULL, NULL, NULL, NULL } /* SOCK_RAW */ }; +/* Global Signal Argument when kqueue support disabled */ +static volatile uintptr_t _glSigArg = 0; + + void rpc_freeCli(rpc_cli_t * __restrict c) { @@ -165,7 +169,14 @@ txPacket(sched_task_t *task) struct tagRPCCall *rpc = (struct tagRPCCall*) buf; int ret, estlen, wlen = sizeof(struct tagRPCCall); struct pollfd pfd; +#ifdef TCP_SESSION_TIMEOUT + struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; + schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL); + schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), ts, TASK_ARG(task), 0); +#endif + if (rpc->call_argc) { f = rpc_srv_getCall(s, ntohs(rpc->call_tag)); if (!f) { @@ -216,7 +227,7 @@ txPacket(sched_task_t *task) if (ret) LOGERR; else - rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond"); + rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond"); /* close connection */ schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], TASK_ARG(task), 0, NULL, 0); @@ -305,7 +316,14 @@ rxPacket(sched_task_t *task) u_char *buf = AIT_GET_BUF(&c->cli_buf); struct tagRPCCall *rpc = (struct tagRPCCall*) buf; struct pollfd pfd; +#ifdef TCP_SESSION_TIMEOUT + struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; + schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL); + schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), ts, TASK_ARG(task), 0); +#endif + memset(buf, 0, sizeof(struct tagRPCCall)); rlen = recv(TASK_FD(task), rpc, sizeof(struct tagRPCCall), MSG_PEEK); if (rlen < sizeof(struct tagRPCCall)) { @@ -332,7 +350,7 @@ rxPacket(sched_task_t *task) if (rlen) LOGERR; else - rpc_SetErr(ETIMEDOUT, "Timeout reached! Server not respond"); + rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond"); schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], TASK_ARG(task), 0, NULL, 0); return NULL; @@ -345,6 +363,7 @@ rxPacket(sched_task_t *task) return NULL; } } + len = estlen; #if 0 /* check integrity of packet */ @@ -386,10 +405,20 @@ acceptClients(sched_task_t *task) rpc_srv_t *srv = TASK_ARG(task); rpc_cli_t *c = NULL; socklen_t salen = sizeof(sockaddr_t); + int sock; +#ifdef TCP_SESSION_TIMEOUT + struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; +#endif c = _allocClient(srv, NULL); - if (!c) + if (!c) { + EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n"); + if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) { + shutdown(sock, SHUT_RDWR); + close(sock); + } goto end; + } /* accept client */ c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen); @@ -401,6 +430,12 @@ acceptClients(sched_task_t *task) } else fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK); +#ifdef TCP_SESSION_TIMEOUT + /* armed timer for close stateless connection */ + schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL); + schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], c, + ts, c, 0); +#endif schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c, c->cli_sock, NULL, 0); end: @@ -417,8 +452,9 @@ txUDPPacket(sched_task_t *task) rpc_func_t *f = NULL; u_char *buf = AIT_GET_BUF(&c->cli_buf); struct tagRPCCall *rpc = (struct tagRPCCall*) buf; - int ret, wlen = sizeof(struct tagRPCCall); + int ret, estlen, wlen = sizeof(struct tagRPCCall); struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; + struct pollfd pfd; schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL); schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], @@ -432,6 +468,13 @@ txUDPPacket(sched_task_t *task) rpc->call_rep.ret = RPC_ERROR(-1); rpc->call_rep.eno = RPC_ERROR(rpc_Errno); } else { + /* calc estimated length */ + estlen = ait_resideVars(RPC_RETVARS(c)) + wlen; + if (estlen > AIT_LEN(&c->cli_buf)) + AIT_RE_BUF(&c->cli_buf, estlen); + buf = AIT_GET_BUF(&c->cli_buf); + rpc = (struct tagRPCCall*) buf; + rpc->call_argc = htons(array_Size(RPC_RETVARS(c))); /* Go Encapsulate variables */ ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen, @@ -455,12 +498,28 @@ txUDPPacket(sched_task_t *task) rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2)); /* send reply */ - ret = sendto(TASK_FD(task), buf, wlen, MSG_NOSIGNAL, - &c->cli_sa.sa, c->cli_sa.sa.sa_len); - if (ret == -1 || ret != wlen) { - /* close connection */ - schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], - TASK_ARG(task), 0, NULL, 0); + pfd.fd = TASK_FD(task); + pfd.events = POLLOUT; + for (; wlen > 0; wlen -= ret, buf += ret) { + if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || + pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { + if (ret) + LOGERR; + else + rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond"); + /* close connection */ + schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), 0, NULL, 0); + return NULL; + } + ret = sendto(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), MSG_NOSIGNAL, + &c->cli_sa.sa, c->cli_sa.sa.sa_len); + if (ret == -1) { + /* close connection */ + schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], + TASK_ARG(task), 0, NULL, 0); + return NULL; + } } return NULL; @@ -471,85 +530,99 @@ rxUDPPacket(sched_task_t *task) { rpc_srv_t *srv = TASK_ARG(task); rpc_cli_t *c = NULL; - int len, rlen, noreply; - u_short crc, off = 0; - u_char buf[USHRT_MAX + 1]; - struct tagRPCCall *rpc = (struct tagRPCCall*) buf; + int len, rlen, noreply, estlen; + u_short crc; + u_char *buf, b[sizeof(struct tagRPCCall)]; + struct tagRPCCall *rpc = (struct tagRPCCall*) b; sockaddr_t sa; socklen_t salen; struct timespec ts = { DEF_RPC_TIMEOUT, 0 }; + struct pollfd pfd; /* receive connect packet */ salen = sa.ss.ss_len = sizeof(sockaddr_t); - rlen = recvfrom(TASK_FD(task), buf, sizeof buf, 0, &sa.sa, &salen); - if (rlen < 1) + rlen = recvfrom(TASK_FD(task), b, sizeof b, MSG_PEEK, &sa.sa, &salen); + if (rlen < sizeof(struct tagRPCCall)) { + rpc_SetErr(ERPCMISMATCH, "Short RPC packet"); goto end; + } c = _allocClient(srv, &sa); - if (!c) + if (!c) { + EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n"); + usleep(2000); /* blocked client delay */ goto end; - else { + } else { + estlen = ntohl(rpc->call_len); + if (estlen > AIT_LEN(&c->cli_buf)) + AIT_RE_BUF(&c->cli_buf, estlen); + rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf); + buf = AIT_GET_BUF(&c->cli_buf); + len = estlen; + c->cli_sock = TASK_FD(task); memcpy(&c->cli_sa, &sa, sizeof c->cli_sa); - memcpy(AIT_GET_BUF(&c->cli_buf), buf, AIT_LEN(&c->cli_buf)); + /* armed timer for close stateless connection */ schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL); schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], c, ts, c, 0); } - do { - /* check RPC packet */ - if (rlen < sizeof(struct tagRPCCall)) { - rpc_SetErr(ERPCMISMATCH, "Short RPC packet"); - break; - } else - rpc = (struct tagRPCCall*) (AIT_GET_BUF(&c->cli_buf) + off); - - len = ntohl(rpc->call_len); - rlen -= len; - - /* check RPC packet lengths */ - if (rlen < 0 || len < sizeof(struct tagRPCCall)) { - rpc_SetErr(ERPCMISMATCH, "Broken RPC packet length"); - /* skip entire packet */ - break; + /* get next part of packet */ + memset(buf, 0, len); + pfd.fd = TASK_FD(task); + pfd.events = POLLIN | POLLPRI; + for (; len > 0; len -= rlen, buf += rlen) { + if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || + pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { + if (rlen) + LOGERR; + else + rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond"); + schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], + c, 0, NULL, 0); + return NULL; } - - /* check integrity of packet */ - crc = ntohs(rpc->call_crc); - rpc->call_crc ^= rpc->call_crc; - if (crc != crcFletcher16((u_short*) rpc, len / 2)) { - rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet"); - - off += len; - /* try next packet remaining into buffer */ - continue; + salen = sa.ss.ss_len = sizeof(sockaddr_t); + rlen = recvfrom(TASK_FD(task), buf, len, 0, &sa.sa, &salen); + if (rlen == -1) { + /* close connection */ + schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], + c, 0, NULL, 0); + return NULL; } + if (e_addrcmp(&c->cli_sa, &sa, 42)) + rlen ^= rlen; /* skip if arrive from different address */ + } + len = estlen; - noreply = RPC_CHK_NOREPLY(rpc); + /* check integrity of packet */ + crc = ntohs(rpc->call_crc); + rpc->call_crc ^= rpc->call_crc; + if (crc != crcFletcher16((u_short*) rpc, len / 2)) { + rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet"); + return NULL; + } - /* check RPC packet session info */ - if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) { - rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session"); + noreply = RPC_CHK_NOREPLY(rpc); - rpc->call_argc ^= rpc->call_argc; - rpc->call_rep.ret = RPC_ERROR(-1); - rpc->call_rep.eno = RPC_ERROR(errno); - } else { - /* execute RPC call */ - schedEvent(TASK_ROOT(task), execCall, c, - (int) noreply, rpc, len); - } + /* check RPC packet session info */ + if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) { + rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session"); - /* send RPC reply */ - if (!noreply) - schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], - c, TASK_FD(task), rpc, len); + rpc->call_argc ^= rpc->call_argc; + rpc->call_rep.ret = RPC_ERROR(-1); + rpc->call_rep.eno = RPC_ERROR(errno); + } else { + /* execute RPC call */ + schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len); + } - off += len; - } while (rlen > 0); - + /* send RPC reply */ + if (!noreply) + schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], + c, TASK_FD(task), rpc, len); end: schedReadSelf(task); return NULL; @@ -691,7 +764,8 @@ end: static void * flushBLOB(sched_task_t *task) { - rpc_srv_t *srv = TASK_ARG(task); + uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg); + rpc_srv_t *srv = sigArg ? (void*) sigArg : TASK_ARG(task); rpc_blob_t *b, *tmp; TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) { @@ -701,7 +775,17 @@ flushBLOB(sched_task_t *task) e_free(b); } - schedSignalSelf(task); + if (!schedSignalSelf(task)) { + /* disabled kqueue support in libaitsched */ + struct sigaction sa; + + memset(&sa, 0, sizeof sa); + sigemptyset(&sa.sa_mask); + sa.sa_handler = (void (*)(int)) flushBLOB; + sa.sa_flags = SA_RESTART | SA_RESETHAND; + sigaction(SIGFBLOB, &sa, NULL); + } + return NULL; } @@ -712,6 +796,7 @@ acceptBLOBClients(sched_task_t *task) rpc_cli_t *c = NULL; register int i; socklen_t salen = sizeof(sockaddr_t); + int sock; #ifdef TCP_NOPUSH int n = 1; #endif @@ -719,8 +804,15 @@ acceptBLOBClients(sched_task_t *task) /* check free slots for connect */ for (i = 0; i < array_Size(srv->srv_blob.clients) && (c = array(srv->srv_blob.clients, i, rpc_cli_t*)); i++); - if (c) /* no more free slots! */ + if (c) { /* no more free slots! */ + EVERBOSE(1, "BLOB client quota exceeded! Connection will be shutdown!\n"); + if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) { + shutdown(sock, SHUT_RDWR); + close(sock); + } goto end; + } + c = e_malloc(sizeof(rpc_cli_t)); if (!c) { LOGERR; @@ -907,7 +999,19 @@ rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv) return -1; } - schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0); + if (!schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0)) { + /* disabled kqueue support in libaitsched */ + struct sigaction sa; + + atomic_store_rel_ptr(&_glSigArg, (uintptr_t) srv); + + memset(&sa, 0, sizeof sa); + sigemptyset(&sa.sa_mask); + sa.sa_handler = (void (*)(int)) flushBLOB; + sa.sa_flags = SA_RESTART | SA_RESETHAND; + sigaction(SIGFBLOB, &sa, NULL); + } + if (!schedRead(srv->srv_blob.root, acceptBLOBClients, srv, srv->srv_blob.server.cli_sock, NULL, 0)) { rpc_SetErr(sched_GetErrno(), "%s", sched_GetError()); @@ -1186,3 +1290,136 @@ rpc_srv_execCall(rpc_cli_t * __restrict cli, struct ta func = AIT_GET_LIKE(&funcname, rpc_callback_t); return func(cli, rpc, args); } + + +/* + * rpc_srv_initServer2() - Init & create layer2 RPC Server + * + * @InstID = Instance for authentication & recognition + * @concurentClients = Concurent clients at same time to this server + * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet) + * @csIface = Interface name for bind server, if NULL first interface on host + * @protoNum = Protocol ethernet number for bind server, if Port == 0 default port is selected + * return: NULL == error or !=NULL bind and created RPC server instance + */ +rpc_srv_t * +rpc_srv_initServer2(u_char InstID, int concurentClients, int netBuf, + const char *csIface, u_short protoNum) +{ + int n = 1; + rpc_srv_t *srv = NULL; + sockaddr_t sa = E_SOCKADDR_INIT; + char szIface[64], szStr[STRSIZ]; + register int i; + struct ifreq ifr; + + if (!concurentClients) { + rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server"); + return NULL; + } + if (!csIface) { + if (e_get1stiface(szIface, sizeof szIface)) + return NULL; + } else + strlcpy(szIface, csIface, sizeof szIface); + if (e_getifacebyname(szIface, &sa)) + return NULL; + if (!protoNum) + protoNum = RPC_DEFPORT; + +#ifdef HAVE_SRANDOMDEV + srandomdev(); +#else + time_t tim; + + srandom((time(&tim) ^ getpid())); +#endif + + srv = e_malloc(sizeof(rpc_srv_t)); + if (!srv) { + LOGERR; + return NULL; + } else + memset(srv, 0, sizeof(rpc_srv_t)); + + srv->srv_proto = protoNum; + srv->srv_netbuf = netBuf; + srv->srv_session.sess_version = RPC_VERSION; + srv->srv_session.sess_instance = InstID; + + srv->srv_server.cli_parent = srv; + memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa); + + /* init functions */ + pthread_mutex_init(&srv->srv_funcs.mtx, NULL); + SLIST_INIT(&srv->srv_funcs); + AVL_INIT(&srv->srv_funcs); + + /* init scheduler */ + srv->srv_root = schedBegin(); + if (!srv->srv_root) { + rpc_SetErr(sched_GetErrno(), "%s", sched_GetError()); + pthread_mutex_destroy(&srv->srv_funcs.mtx); + e_free(srv); + return NULL; + } + + /* init pool for clients */ + srv->srv_clients = array_Init(concurentClients); + if (!srv->srv_clients) { + rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError()); + schedEnd(&srv->srv_root); + pthread_mutex_destroy(&srv->srv_funcs.mtx); + e_free(srv); + return NULL; + } + + /* create server handler */ + for (i = 0; i < 10; i++) { + memset(szStr, 0, sizeof szStr); + snprintf(szStr, sizeof szStr, "/dev/bpf%d", i); + srv->srv_server.cli_sock = open(szStr, O_RDWR); + if (srv->srv_server.cli_sock > STDERR_FILENO) + break; + } + if (srv->srv_server.cli_sock < 3) { + LOGERR; + array_Destroy(&srv->srv_clients); + schedEnd(&srv->srv_root); + pthread_mutex_destroy(&srv->srv_funcs.mtx); + e_free(srv); + return NULL; + } + + if (ioctl(srv->srv_server.cli_sock, BIOCIMMEDIATE, &n) == -1) { + LOGERR; + goto err; + } + n = (netBuf < RPC_MIN_BUFSIZ) ? getpagesize() : E_ALIGN(netBuf, 2); + if (ioctl(srv->srv_server.cli_sock, BIOCSBLEN, &n) == -1) { + LOGERR; + goto err; + } else + srv->srv_netbuf = n; + + memset(&ifr, 0, sizeof ifr); + strlcpy(ifr.ifr_name, szIface, sizeof ifr.ifr_name); + if (ioctl(srv->srv_server.cli_sock, BIOCSETIF, &ifr) == -1) { + LOGERR; + goto err; + } else + fcntl(srv->srv_server.cli_sock, F_SETFL, + fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK); + + rpc_register_srvPing(srv); + + return srv; +err: /* error condition */ + close(srv->srv_server.cli_sock); + array_Destroy(&srv->srv_clients); + schedEnd(&srv->srv_root); + pthread_mutex_destroy(&srv->srv_funcs.mtx); + e_free(srv); + return NULL; +} +