Annotation of libaitrpc/src/srv.c, revision 1.23.6.6

1.1       misho       1: /*************************************************************************
                      2: * (C) 2010 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
                      3: *  by Michael Pounov <misho@openbsd-bg.org>
                      4: *
                      5: * $Author: misho $
1.23.6.6! misho       6: * $Id: srv.c,v 1.23.6.5 2014/12/18 00:28:47 misho Exp $
1.1       misho       7: *
1.2       misho       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: 
1.23      misho      15: Copyright 2004 - 2014
1.2       misho      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: */
1.1       misho      46: #include "global.h"
                     47: 
                     48: 
1.13      misho      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: 
1.23.6.2  misho      62: /* SOCK_BPF */
                     63: static void *rxBPFPacket(sched_task_t *);
                     64: static void *txBPFPacket(sched_task_t *);
                     65: 
                     66: static sched_task_func_t cbProto[SOCK_BPF + 1][4] = {
1.13      misho      67:        { acceptClients, closeClient, rxPacket, txPacket },     /* SOCK_STREAM */
                     68:        { acceptClients, closeClient, rxPacket, txPacket },     /* SOCK_STREAM */
                     69:        { rxUDPPacket, freeClient, rxUDPPacket, txUDPPacket },  /* SOCK_DGRAM */
1.23.6.2  misho      70:        { NULL, NULL, NULL, NULL },                             /* SOCK_RAW */
                     71:        { rxBPFPacket, freeClient, rxBPFPacket, txBPFPacket }   /* SOCK_BPF */
1.13      misho      72: };
                     73: 
1.23      misho      74: /* Global Signal Argument when kqueue support disabled */
                     75: 
                     76: static volatile uintptr_t _glSigArg = 0;
                     77: 
1.13      misho      78: 
1.16      misho      79: void
1.13      misho      80: rpc_freeCli(rpc_cli_t * __restrict c)
1.10      misho      81: {
                     82:        rpc_srv_t *s = c->cli_parent;
                     83: 
1.13      misho      84:        schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
1.10      misho      85: 
                     86:        /* free buffer */
                     87:        AIT_FREE_VAL(&c->cli_buf);
                     88: 
1.14      misho      89:        array_Del(s->srv_clients, c->cli_id, 0);
1.10      misho      90:        if (c)
1.14      misho      91:                e_free(c);
1.13      misho      92: }
                     93: 
                     94: 
                     95: static inline int
1.14      misho      96: _check4freeslot(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
1.13      misho      97: {
                     98:        rpc_cli_t *c = NULL;
                     99:        register int i;
                    100: 
                    101:        /* check free slots for connect */
1.14      misho     102:        for (i = 0; i < array_Size(srv->srv_clients) && 
                    103:                        (c = array(srv->srv_clients, i, rpc_cli_t*)); i++)
1.13      misho     104:                /* check for duplicates */
1.14      misho     105:                if (sa && !e_addrcmp(&c->cli_sa, sa, 42))
1.13      misho     106:                        break;
1.14      misho     107:        if (i >= array_Size(srv->srv_clients))
1.13      misho     108:                return -1;      /* no more free slots! */
                    109: 
                    110:        return i;
                    111: }
                    112: 
                    113: static rpc_cli_t *
1.14      misho     114: _allocClient(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
1.13      misho     115: {
                    116:        rpc_cli_t *c = NULL;
                    117:        int n;
                    118: 
                    119:        n = _check4freeslot(srv, sa);
                    120:        if (n == -1)
                    121:                return NULL;
                    122:        else
1.14      misho     123:                c = array(srv->srv_clients, n, rpc_cli_t*);
1.13      misho     124: 
                    125:        if (!c) {
1.14      misho     126:                c = e_malloc(sizeof(rpc_cli_t));
1.13      misho     127:                if (!c) {
                    128:                        LOGERR;
                    129:                        srv->srv_kill = 1;
                    130:                        return NULL;
                    131:                } else {
                    132:                        memset(c, 0, sizeof(rpc_cli_t));
1.14      misho     133:                        array_Set(srv->srv_clients, n, c);
1.13      misho     134:                        c->cli_id = n;
                    135:                        c->cli_parent = srv;
                    136:                }
                    137: 
                    138:                /* alloc empty buffer */
1.14      misho     139:                AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
1.13      misho     140:        }
                    141: 
                    142:        return c;
                    143: }
                    144: 
                    145: 
                    146: static void *
                    147: freeClient(sched_task_t *task)
                    148: {
                    149:        rpc_freeCli(TASK_ARG(task));
                    150: 
                    151:        return NULL;
                    152: }
                    153: 
                    154: static void *
                    155: closeClient(sched_task_t *task)
                    156: {
                    157:        int sock = ((rpc_cli_t*) TASK_ARG(task))->cli_sock;
                    158: 
                    159:        rpc_freeCli(TASK_ARG(task));
                    160: 
                    161:        /* close client socket */
                    162:        shutdown(sock, SHUT_RDWR);
                    163:        close(sock);
1.10      misho     164:        return NULL;
                    165: }
1.7       misho     166: 
                    167: static void *
                    168: txPacket(sched_task_t *task)
                    169: {
                    170:        rpc_cli_t *c = TASK_ARG(task);
                    171:        rpc_srv_t *s = c->cli_parent;
                    172:        rpc_func_t *f = NULL;
1.18      misho     173:        u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.7       misho     174:        struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1.18      misho     175:        int ret, estlen, wlen = sizeof(struct tagRPCCall);
1.19      misho     176:        struct pollfd pfd;
1.21      misho     177: #ifdef TCP_SESSION_TIMEOUT
                    178:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
                    179: 
                    180:        schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
                    181:        schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    182:                         TASK_ARG(task), ts, TASK_ARG(task), 0);
                    183: #endif
1.7       misho     184: 
                    185:        if (rpc->call_argc) {
1.10      misho     186:                f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
1.7       misho     187:                if (!f) {
1.10      misho     188:                        rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
1.18      misho     189: 
1.7       misho     190:                        rpc->call_argc ^= rpc->call_argc;
                    191:                        rpc->call_rep.ret = RPC_ERROR(-1);
                    192:                        rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    193:                } else {
1.18      misho     194:                        /* calc estimated length */
                    195:                        estlen = ait_resideVars(RPC_RETVARS(c)) + wlen;
                    196:                        if (estlen > AIT_LEN(&c->cli_buf))
                    197:                                AIT_RE_BUF(&c->cli_buf, estlen);
                    198:                        buf = AIT_GET_BUF(&c->cli_buf);
1.19      misho     199:                        rpc = (struct tagRPCCall*) buf;
1.18      misho     200: 
1.14      misho     201:                        rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
1.7       misho     202:                        /* Go Encapsulate variables */
1.18      misho     203:                        ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen, 
                    204:                                        RPC_RETVARS(c));
1.10      misho     205:                        /* Free return values */
1.14      misho     206:                        ait_freeVars(&c->cli_vars);
1.7       misho     207:                        if (ret == -1) {
                    208:                                rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
1.19      misho     209: 
1.7       misho     210:                                rpc->call_argc ^= rpc->call_argc;
                    211:                                rpc->call_rep.ret = RPC_ERROR(-1);
                    212:                                rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    213:                        } else
                    214:                                wlen += ret;
                    215:                }
                    216:        }
                    217: 
1.18      misho     218:        rpc->call_len = htonl(wlen);
1.8       misho     219: 
1.15      misho     220: #if 0
1.7       misho     221:        /* calculate CRC */
                    222:        rpc->call_crc ^= rpc->call_crc;
1.8       misho     223:        rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
1.15      misho     224: #endif
1.7       misho     225: 
                    226:        /* send reply */
1.19      misho     227:        pfd.fd = TASK_FD(task);
                    228:        pfd.events = POLLOUT;
                    229:        for (; wlen > 0; wlen -= ret, buf += ret) {
                    230:                if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
                    231:                                pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
                    232:                        if (ret)
                    233:                                LOGERR;
                    234:                        else
1.22      misho     235:                                rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
1.19      misho     236:                        /* close connection */
                    237:                        schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    238:                                        TASK_ARG(task), 0, NULL, 0);
                    239:                        return NULL;
                    240:                }
                    241:                ret = send(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), MSG_NOSIGNAL);
                    242:                if (ret == -1) {
                    243:                        /* close connection */
                    244:                        schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    245:                                        TASK_ARG(task), 0, NULL, 0);
                    246:                        return NULL;
                    247:                }
1.10      misho     248:        }
1.7       misho     249: 
                    250:        return NULL;
                    251: }
                    252: 
                    253: static void *
                    254: execCall(sched_task_t *task)
                    255: {
                    256:        rpc_cli_t *c = TASK_ARG(task);
                    257:        rpc_srv_t *s = c->cli_parent;
                    258:        rpc_func_t *f = NULL;
                    259:        array_t *arr = NULL;
1.18      misho     260:        u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.7       misho     261:        struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
                    262:        int argc = ntohs(rpc->call_argc);
                    263: 
                    264:        /* Go decapsulate variables ... */
1.8       misho     265:        if (argc) {
1.14      misho     266:                arr = ait_buffer2vars(buf + sizeof(struct tagRPCCall), 
1.18      misho     267:                                AIT_LEN(&c->cli_buf) - sizeof(struct tagRPCCall), argc, 42);
1.7       misho     268:                if (!arr) {
1.14      misho     269:                        rpc_SetErr(ERPCMISMATCH, "#%d - %s", elwix_GetErrno(), elwix_GetError());
1.18      misho     270: 
1.7       misho     271:                        rpc->call_argc ^= rpc->call_argc;
                    272:                        rpc->call_rep.ret = RPC_ERROR(-1);
                    273:                        rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    274:                        return NULL;
                    275:                }
1.10      misho     276:        } else
                    277:                arr = NULL;
1.7       misho     278: 
1.10      misho     279:        if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag)))) {
1.7       misho     280:                rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
1.18      misho     281: 
1.7       misho     282:                rpc->call_argc ^= rpc->call_argc;
                    283:                rpc->call_rep.ret = RPC_ERROR(-1);
                    284:                rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    285:        } else {
1.8       misho     286:                /* if client doesn't want reply */
1.10      misho     287:                argc = RPC_CHK_NOREPLY(rpc);
                    288:                rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(c, rpc, f->func_name, arr));
1.7       misho     289:                if (rpc->call_rep.ret == htonl(-1)) {
1.20      misho     290:                        if (!rpc->call_rep.eno) {
                    291:                                LOGERR;
                    292:                                rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    293:                        }
1.7       misho     294:                        rpc->call_argc ^= rpc->call_argc;
                    295:                } else {
                    296:                        rpc->call_rep.eno ^= rpc->call_rep.eno;
1.8       misho     297:                        if (argc) {
                    298:                                /* without reply */
1.14      misho     299:                                ait_freeVars(&c->cli_vars);
1.8       misho     300:                                rpc->call_argc ^= rpc->call_argc;
1.10      misho     301:                        } else {
                    302:                                /* reply */
1.14      misho     303:                                rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
1.10      misho     304:                        }
1.7       misho     305:                }
                    306:        }
                    307: 
1.14      misho     308:        array_Destroy(&arr);
1.7       misho     309:        return NULL;
                    310: }
                    311: 
                    312: static void *
                    313: rxPacket(sched_task_t *task)
                    314: {
                    315:        rpc_cli_t *c = TASK_ARG(task);
                    316:        rpc_srv_t *s = c->cli_parent;
1.18      misho     317:        int len, rlen, noreply, estlen;
1.15      misho     318: #if 0
                    319:        u_short crc;
                    320: #endif
1.10      misho     321:        u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.18      misho     322:        struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1.19      misho     323:        struct pollfd pfd;
1.21      misho     324: #ifdef TCP_SESSION_TIMEOUT
                    325:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
                    326: 
                    327:        schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
                    328:        schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    329:                         TASK_ARG(task), ts, TASK_ARG(task), 0);
                    330: #endif
1.7       misho     331: 
1.18      misho     332:        memset(buf, 0, sizeof(struct tagRPCCall));
1.19      misho     333:        rlen = recv(TASK_FD(task), rpc, sizeof(struct tagRPCCall), MSG_PEEK);
1.18      misho     334:        if (rlen < sizeof(struct tagRPCCall)) {
1.10      misho     335:                /* close connection */
1.13      misho     336:                schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    337:                                TASK_ARG(task), 0, NULL, 0);
1.7       misho     338:                return NULL;
1.10      misho     339:        } else {
1.18      misho     340:                estlen = ntohl(rpc->call_len);
                    341:                if (estlen > AIT_LEN(&c->cli_buf))
                    342:                        AIT_RE_BUF(&c->cli_buf, estlen);
                    343:                rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
1.19      misho     344:                buf = AIT_GET_BUF(&c->cli_buf);
                    345:                len = estlen;
1.18      misho     346:        }
                    347: 
1.19      misho     348:        /* get next part of packet */
                    349:        memset(buf, 0, len);
                    350:        pfd.fd = TASK_FD(task);
                    351:        pfd.events = POLLIN | POLLPRI;
                    352:        for (; len > 0; len -= rlen, buf += rlen) {
                    353:                if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
                    354:                                pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
                    355:                        if (rlen)
                    356:                                LOGERR;
                    357:                        else
1.22      misho     358:                                rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
1.19      misho     359:                        schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    360:                                        TASK_ARG(task), 0, NULL, 0);
                    361:                        return NULL;
                    362:                }
1.18      misho     363:                rlen = recv(TASK_FD(task), buf, len, 0);
                    364:                if (rlen == -1) {
                    365:                        /* close connection */
                    366:                        schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    367:                                        TASK_ARG(task), 0, NULL, 0);
1.8       misho     368:                        return NULL;
1.10      misho     369:                }
1.18      misho     370:        }
1.22      misho     371:        len = estlen;
1.8       misho     372: 
1.15      misho     373: #if 0
1.18      misho     374:        /* check integrity of packet */
                    375:        crc = ntohs(rpc->call_crc);
                    376:        rpc->call_crc ^= rpc->call_crc;
                    377:        if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
                    378:                rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
                    379:                return NULL;
                    380:        }
1.15      misho     381: #endif
1.7       misho     382: 
1.18      misho     383:        noreply = RPC_CHK_NOREPLY(rpc);
1.7       misho     384: 
1.18      misho     385:        /* check RPC packet session info */
                    386:        if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
                    387:                rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1.7       misho     388: 
1.18      misho     389:                rpc->call_argc ^= rpc->call_argc;
                    390:                rpc->call_rep.ret = RPC_ERROR(-1);
                    391:                rpc->call_rep.eno = RPC_ERROR(errno);
                    392:        } else {
                    393:                /* execute RPC call */
                    394:                schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len);
                    395:        }
1.7       misho     396: 
1.18      misho     397:        /* send RPC reply */
                    398:        if (!noreply)
                    399:                schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET], 
                    400:                                TASK_ARG(task), TASK_FD(task), rpc, len);
1.7       misho     401: 
                    402:        /* lets get next packet */
1.18      misho     403:        schedReadSelf(task);
1.7       misho     404:        return NULL;
                    405: }
                    406: 
1.1       misho     407: static void *
1.10      misho     408: acceptClients(sched_task_t *task)
1.1       misho     409: {
1.10      misho     410:        rpc_srv_t *srv = TASK_ARG(task);
                    411:        rpc_cli_t *c = NULL;
1.14      misho     412:        socklen_t salen = sizeof(sockaddr_t);
1.21      misho     413:        int sock;
                    414: #ifdef TCP_SESSION_TIMEOUT
                    415:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
                    416: #endif
1.7       misho     417: 
1.13      misho     418:        c = _allocClient(srv, NULL);
1.21      misho     419:        if (!c) {
                    420:                EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
                    421:                if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
                    422:                        shutdown(sock, SHUT_RDWR);
                    423:                        close(sock);
                    424:                }
1.10      misho     425:                goto end;
1.21      misho     426:        }
1.10      misho     427: 
                    428:        /* accept client */
                    429:        c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
                    430:        if (c->cli_sock == -1) {
                    431:                LOGERR;
                    432:                AIT_FREE_VAL(&c->cli_buf);
1.14      misho     433:                array_Del(srv->srv_clients, c->cli_id, 42);
1.10      misho     434:                goto end;
1.1       misho     435:        } else
1.10      misho     436:                fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1.1       misho     437: 
1.21      misho     438: #ifdef TCP_SESSION_TIMEOUT
                    439:        /* armed timer for close stateless connection */
                    440:        schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
                    441:        schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], c, 
                    442:                        ts, c, 0);
                    443: #endif
1.13      misho     444:        schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c, 
                    445:                        c->cli_sock, NULL, 0);
1.10      misho     446: end:
                    447:        schedReadSelf(task);
                    448:        return NULL;
                    449: }
1.5       misho     450: 
1.7       misho     451: 
1.10      misho     452: static void *
1.13      misho     453: txUDPPacket(sched_task_t *task)
1.10      misho     454: {
                    455:        rpc_cli_t *c = TASK_ARG(task);
                    456:        rpc_srv_t *s = c->cli_parent;
1.13      misho     457:        rpc_func_t *f = NULL;
1.18      misho     458:        u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.13      misho     459:        struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1.22      misho     460:        int ret, estlen, wlen = sizeof(struct tagRPCCall);
1.13      misho     461:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
1.22      misho     462:        struct pollfd pfd;
1.13      misho     463: 
                    464:        schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
                    465:        schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    466:                         TASK_ARG(task), ts, TASK_ARG(task), 0);
                    467: 
                    468:        if (rpc->call_argc) {
                    469:                f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
                    470:                if (!f) {
                    471:                        rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
                    472:                        rpc->call_argc ^= rpc->call_argc;
                    473:                        rpc->call_rep.ret = RPC_ERROR(-1);
                    474:                        rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    475:                } else {
1.22      misho     476:                        /* calc estimated length */
                    477:                        estlen = ait_resideVars(RPC_RETVARS(c)) + wlen;
                    478:                        if (estlen > AIT_LEN(&c->cli_buf))
                    479:                                AIT_RE_BUF(&c->cli_buf, estlen);
                    480:                        buf = AIT_GET_BUF(&c->cli_buf);
                    481:                        rpc = (struct tagRPCCall*) buf;
                    482: 
1.14      misho     483:                        rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
1.13      misho     484:                        /* Go Encapsulate variables */
1.18      misho     485:                        ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen, 
                    486:                                        RPC_RETVARS(c));
1.13      misho     487:                        /* Free return values */
1.14      misho     488:                        ait_freeVars(&c->cli_vars);
1.13      misho     489:                        if (ret == -1) {
                    490:                                rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
                    491:                                rpc->call_argc ^= rpc->call_argc;
                    492:                                rpc->call_rep.ret = RPC_ERROR(-1);
                    493:                                rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    494:                        } else
                    495:                                wlen += ret;
                    496:                }
                    497:        }
1.7       misho     498: 
1.18      misho     499:        rpc->call_len = htonl(wlen);
1.7       misho     500: 
1.13      misho     501:        /* calculate CRC */
                    502:        rpc->call_crc ^= rpc->call_crc;
                    503:        rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
                    504: 
                    505:        /* send reply */
1.22      misho     506:        pfd.fd = TASK_FD(task);
                    507:        pfd.events = POLLOUT;
                    508:        for (; wlen > 0; wlen -= ret, buf += ret) {
                    509:                if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
                    510:                                pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
                    511:                        if (ret)
                    512:                                LOGERR;
                    513:                        else
                    514:                                rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
                    515:                        /* close connection */
                    516:                        schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    517:                                        TASK_ARG(task), 0, NULL, 0);
                    518:                        return NULL;
                    519:                }
                    520:                ret = sendto(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), MSG_NOSIGNAL, 
                    521:                                &c->cli_sa.sa, c->cli_sa.sa.sa_len);
                    522:                if (ret == -1) {
                    523:                        /* close connection */
                    524:                        schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    525:                                        TASK_ARG(task), 0, NULL, 0);
                    526:                        return NULL;
                    527:                }
1.13      misho     528:        }
                    529: 
                    530:        return NULL;
                    531: }
                    532: 
                    533: static void *
                    534: rxUDPPacket(sched_task_t *task)
                    535: {
                    536:        rpc_srv_t *srv = TASK_ARG(task);
                    537:        rpc_cli_t *c = NULL;
1.22      misho     538:        int len, rlen, noreply, estlen;
                    539:        u_short crc;
                    540:        u_char *buf, b[sizeof(struct tagRPCCall)];
                    541:        struct tagRPCCall *rpc = (struct tagRPCCall*) b;
1.14      misho     542:        sockaddr_t sa;
                    543:        socklen_t salen;
1.13      misho     544:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
1.22      misho     545:        struct pollfd pfd;
1.13      misho     546: 
                    547:        /* receive connect packet */
1.14      misho     548:        salen = sa.ss.ss_len = sizeof(sockaddr_t);
1.22      misho     549:        rlen = recvfrom(TASK_FD(task), b, sizeof b, MSG_PEEK, &sa.sa, &salen);
                    550:        if (rlen < sizeof(struct tagRPCCall)) {
                    551:                rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
1.13      misho     552:                goto end;
1.22      misho     553:        }
1.13      misho     554: 
                    555:        c = _allocClient(srv, &sa);
1.22      misho     556:        if (!c) {
                    557:                EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
                    558:                usleep(2000);   /* blocked client delay */
1.13      misho     559:                goto end;
1.22      misho     560:        } else {
                    561:                estlen = ntohl(rpc->call_len);
                    562:                if (estlen > AIT_LEN(&c->cli_buf))
                    563:                        AIT_RE_BUF(&c->cli_buf, estlen);
                    564:                rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
                    565:                buf = AIT_GET_BUF(&c->cli_buf);
                    566:                len = estlen;
                    567: 
1.13      misho     568:                c->cli_sock = TASK_FD(task);
                    569:                memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);
1.22      misho     570: 
1.13      misho     571:                /* armed timer for close stateless connection */
                    572:                schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
                    573:                schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], 
                    574:                                c, ts, c, 0);
                    575:        }
                    576: 
1.22      misho     577:        /* get next part of packet */
                    578:        memset(buf, 0, len);
                    579:        pfd.fd = TASK_FD(task);
                    580:        pfd.events = POLLIN | POLLPRI;
                    581:        for (; len > 0; len -= rlen, buf += rlen) {
                    582:                if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
                    583:                                pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
                    584:                        if (rlen)
                    585:                                LOGERR;
                    586:                        else
                    587:                                rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
                    588:                        schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], 
                    589:                                        c, 0, NULL, 0);
1.23.6.5  misho     590:                        goto end;
1.13      misho     591:                }
1.22      misho     592:                salen = sa.ss.ss_len = sizeof(sockaddr_t);
                    593:                rlen = recvfrom(TASK_FD(task), buf, len, 0, &sa.sa, &salen);
                    594:                if (rlen == -1) {
                    595:                        /* close connection */
                    596:                        schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], 
                    597:                                        c, 0, NULL, 0);
1.23.6.5  misho     598:                        goto end;
1.13      misho     599:                }
1.22      misho     600:                if (e_addrcmp(&c->cli_sa, &sa, 42))
                    601:                        rlen ^= rlen;   /* skip if arrive from different address */
                    602:        }
                    603:        len = estlen;
1.13      misho     604: 
1.22      misho     605:        /* check integrity of packet */
                    606:        crc = ntohs(rpc->call_crc);
                    607:        rpc->call_crc ^= rpc->call_crc;
                    608:        if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
                    609:                rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
1.23.6.5  misho     610:                /* close connection */
                    611:                schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], 
                    612:                                c, 0, NULL, 0);
                    613:                goto end;
1.22      misho     614:        }
1.13      misho     615: 
1.22      misho     616:        noreply = RPC_CHK_NOREPLY(rpc);
1.18      misho     617: 
1.22      misho     618:        /* check RPC packet session info */
                    619:        if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
                    620:                rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1.13      misho     621: 
1.22      misho     622:                rpc->call_argc ^= rpc->call_argc;
                    623:                rpc->call_rep.ret = RPC_ERROR(-1);
                    624:                rpc->call_rep.eno = RPC_ERROR(errno);
                    625:        } else {
                    626:                /* execute RPC call */
                    627:                schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
                    628:        }
1.13      misho     629: 
1.22      misho     630:        /* send RPC reply */
                    631:        if (!noreply)
                    632:                schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], 
                    633:                                c, TASK_FD(task), rpc, len);
1.13      misho     634: end:
                    635:        schedReadSelf(task);
                    636:        return NULL;
                    637: }
                    638: 
1.23.6.3  misho     639: 
                    640: static void *
                    641: txBPFPacket(sched_task_t *task)
                    642: {
                    643:        rpc_cli_t *c = TASK_ARG(task);
                    644:        rpc_srv_t *s = c->cli_parent;
                    645:        rpc_func_t *f = NULL;
                    646:        u_char *buf = AIT_GET_BUF(&c->cli_buf);
                    647:        struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1.23.6.6! misho     648:        int ret, len, wlen = sizeof(struct tagRPCCall);
1.23.6.3  misho     649:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
1.23.6.6! misho     650:        struct ether_header *eh;
        !           651:        ait_val_t b = AIT_VAL_INIT;
1.23.6.3  misho     652: 
                    653:        schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
                    654:        schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
                    655:                         TASK_ARG(task), ts, TASK_ARG(task), 0);
                    656: 
                    657:        if (rpc->call_argc) {
                    658:                f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
                    659:                if (!f) {
                    660:                        rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
                    661:                        rpc->call_argc ^= rpc->call_argc;
                    662:                        rpc->call_rep.ret = RPC_ERROR(-1);
                    663:                        rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    664:                } else {
                    665:                        /* calc estimated length */
1.23.6.6! misho     666:                        len = ait_resideVars(RPC_RETVARS(c)) + wlen;
        !           667:                        if (len > AIT_LEN(&c->cli_buf))
        !           668:                                AIT_RE_BUF(&c->cli_buf, len);
1.23.6.3  misho     669:                        buf = AIT_GET_BUF(&c->cli_buf);
                    670:                        rpc = (struct tagRPCCall*) buf;
                    671: 
                    672:                        rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
                    673:                        /* Go Encapsulate variables */
                    674:                        ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen, 
                    675:                                        RPC_RETVARS(c));
                    676:                        /* Free return values */
1.23.6.6! misho     677:                        ait_freeVars(&RPC_RETVARS(c));
1.23.6.3  misho     678:                        if (ret == -1) {
                    679:                                rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
                    680:                                rpc->call_argc ^= rpc->call_argc;
                    681:                                rpc->call_rep.ret = RPC_ERROR(-1);
                    682:                                rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    683:                        } else
                    684:                                wlen += ret;
                    685:                }
                    686:        }
                    687: 
                    688:        rpc->call_len = htonl(wlen);
                    689: 
                    690:        /* calculate CRC */
                    691:        rpc->call_crc ^= rpc->call_crc;
                    692:        rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
                    693: 
                    694:        /* send reply */
1.23.6.6! misho     695:        AIT_SET_BUF(&b, NULL, MIN(wlen, s->srv_netbuf) + ETHER_HDR_LEN);
        !           696:        eh = (struct ether_header*) AIT_GET_BUF(&b);
        !           697:        memcpy(eh->ether_dhost, LLADDR(&c->cli_sa.sdl), ETHER_ADDR_LEN);
        !           698:        eh->ether_type = htons(RPC_DEFPORT);
        !           699:        memcpy(eh + 1, buf, MIN(wlen, s->srv_netbuf));
        !           700: 
        !           701:        ret = write(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
        !           702:        AIT_FREE_VAL(&b);
        !           703:        if (ret == -1) {
        !           704:                /* close connection */
        !           705:                schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT], 
        !           706:                                TASK_ARG(task), 0, NULL, 0);
        !           707:                return NULL;
1.23.6.3  misho     708:        }
                    709: 
                    710:        return NULL;
                    711: }
                    712: 
                    713: static void *
                    714: rxBPFPacket(sched_task_t *task)
                    715: {
                    716:        rpc_srv_t *srv = TASK_ARG(task);
                    717:        rpc_cli_t *c = NULL;
1.23.6.5  misho     718:        int len, rlen, noreply;
1.23.6.3  misho     719:        u_short crc;
1.23.6.4  misho     720:        struct tagRPCCall *rpc;
1.23.6.3  misho     721:        sockaddr_t sa;
                    722:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
                    723:        struct bpf_hdr *h;
                    724:        struct ether_header *eh;
                    725:        ait_val_t b = AIT_VAL_INIT;
                    726: 
                    727:        /* receive connect packet */
                    728:        AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
                    729:        rlen = read(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
1.23.6.5  misho     730:        h = (struct bpf_hdr*) AIT_GET_BUF(&b);
                    731:        rlen -= h->bh_hdrlen;
                    732:        if (rlen < h->bh_caplen || h->bh_caplen != h->bh_datalen || 
                    733:                        rlen < ETHER_HDR_LEN + sizeof(struct tagRPCCall)) {
1.23.6.3  misho     734:                rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
                    735:                goto end;
                    736:        } else {
1.23.6.5  misho     737:                rlen = h->bh_caplen;
1.23.6.3  misho     738:                eh = (struct ether_header*) (AIT_GET_BUF(&b) + h->bh_hdrlen);
1.23.6.5  misho     739:                rlen -= ETHER_HDR_LEN;
                    740:                rpc = (struct tagRPCCall*) (eh + 1);
                    741:                if (eh->ether_type != ntohs(RPC_DEFPORT))
                    742:                        goto end;
                    743:                else
                    744:                        e_getlinkbymac((const ether_addr_t*) eh->ether_shost, &sa);
1.23.6.3  misho     745:        }
                    746: 
                    747:        c = _allocClient(srv, &sa);
                    748:        if (!c) {
                    749:                EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
                    750:                usleep(2000);   /* blocked client delay */
                    751:                goto end;
                    752:        } else {
1.23.6.5  misho     753:                len = ntohl(rpc->call_len);
                    754:                if (len > AIT_LEN(&c->cli_buf))
                    755:                        AIT_RE_BUF(&c->cli_buf, len);
                    756:                memcpy(AIT_GET_BUF(&c->cli_buf), rpc, AIT_LEN(&c->cli_buf));
1.23.6.3  misho     757:                rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
                    758: 
                    759:                c->cli_sock = TASK_FD(task);
                    760:                memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);
                    761: 
                    762:                /* armed timer for close stateless connection */
                    763:                schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
                    764:                schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], 
                    765:                                c, ts, c, 0);
                    766:        }
                    767: 
                    768:        /* check integrity of packet */
                    769:        crc = ntohs(rpc->call_crc);
                    770:        rpc->call_crc ^= rpc->call_crc;
                    771:        if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
                    772:                rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
1.23.6.5  misho     773:                /* close connection */
                    774:                schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], 
                    775:                                c, 0, NULL, 0);
                    776:                goto end;
1.23.6.3  misho     777:        }
                    778: 
                    779:        noreply = RPC_CHK_NOREPLY(rpc);
                    780: 
                    781:        /* check RPC packet session info */
                    782:        if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
                    783:                rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
                    784: 
                    785:                rpc->call_argc ^= rpc->call_argc;
                    786:                rpc->call_rep.ret = RPC_ERROR(-1);
                    787:                rpc->call_rep.eno = RPC_ERROR(errno);
                    788:        } else {
                    789:                /* execute RPC call */
                    790:                schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
                    791:        }
                    792: 
                    793:        /* send RPC reply */
                    794:        if (!noreply)
                    795:                schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET], 
                    796:                                c, TASK_FD(task), rpc, len);
                    797: end:
                    798:        AIT_FREE_VAL(&b);
                    799:        schedReadSelf(task);
                    800:        return NULL;
                    801: }
                    802: 
1.13      misho     803: /* ------------------------------------------------------ */
                    804: 
1.16      misho     805: void
1.13      misho     806: rpc_freeBLOBCli(rpc_cli_t * __restrict c)
                    807: {
                    808:        rpc_srv_t *s = c->cli_parent;
                    809: 
                    810:        schedCancelby(s->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
1.10      misho     811: 
                    812:        /* free buffer */
                    813:        AIT_FREE_VAL(&c->cli_buf);
                    814: 
1.14      misho     815:        array_Del(s->srv_blob.clients, c->cli_id, 0);
1.10      misho     816:        if (c)
1.14      misho     817:                e_free(c);
1.13      misho     818: }
                    819: 
                    820: 
                    821: static void *
                    822: closeBLOBClient(sched_task_t *task)
                    823: {
                    824:        int sock = ((rpc_cli_t*) TASK_ARG(task))->cli_sock;
                    825: 
                    826:        rpc_freeBLOBCli(TASK_ARG(task));
                    827: 
                    828:        /* close client socket */
                    829:        shutdown(sock, SHUT_RDWR);
                    830:        close(sock);
1.7       misho     831:        return NULL;
                    832: }
                    833: 
                    834: static void *
                    835: txBLOB(sched_task_t *task)
                    836: {
1.10      misho     837:        rpc_cli_t *c = TASK_ARG(task);
                    838:        u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.7       misho     839:        int wlen = sizeof(struct tagBLOBHdr);
                    840: 
                    841:        /* send reply */
1.10      misho     842:        wlen = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);
                    843:        if (wlen == -1 || wlen != sizeof(struct tagBLOBHdr)) {
                    844:                /* close blob connection */
                    845:                schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
                    846:        }
1.7       misho     847: 
                    848:        return NULL;
                    849: }
1.4       misho     850: 
1.7       misho     851: static void *
                    852: rxBLOB(sched_task_t *task)
                    853: {
                    854:        rpc_cli_t *c = TASK_ARG(task);
                    855:        rpc_srv_t *s = c->cli_parent;
                    856:        rpc_blob_t *b;
1.10      misho     857:        struct tagBLOBHdr blob;
1.7       misho     858:        int rlen;
                    859: 
1.10      misho     860:        memset(&blob, 0, sizeof blob);
                    861:        rlen = recv(TASK_FD(task), &blob, sizeof blob, 0);
                    862:        if (rlen < 1) {
                    863:                /* close blob connection */
                    864:                schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
1.7       misho     865:                return NULL;
                    866:        }
1.5       misho     867: 
1.10      misho     868:        /* check BLOB packet */
                    869:        if (rlen < sizeof(struct tagBLOBHdr)) {
                    870:                rpc_SetErr(ERPCMISMATCH, "Short BLOB packet");
1.6       misho     871: 
1.10      misho     872:                schedReadSelf(task);
1.7       misho     873:                return NULL;
                    874:        }
1.1       misho     875: 
1.7       misho     876:        /* check RPC packet session info */
1.15      misho     877:        if (rpc_chkPktSession(&blob.hdr_session, &s->srv_session)) {
1.7       misho     878:                rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1.10      misho     879:                blob.hdr_cmd = error;
1.7       misho     880:                goto end;
                    881:        }
                    882: 
                    883:        /* Go to proceed packet ... */
1.10      misho     884:        switch (blob.hdr_cmd) {
1.7       misho     885:                case get:
1.10      misho     886:                        if (!(b = rpc_srv_getBLOB(s, ntohl(blob.hdr_var)))) {
                    887:                                rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob.hdr_var));
                    888:                                blob.hdr_cmd = no;
                    889:                                blob.hdr_ret = RPC_ERROR(-1);
1.7       misho     890:                                break;
                    891:                        } else
1.10      misho     892:                                blob.hdr_len = htonl(b->blob_len);
1.5       misho     893: 
1.7       misho     894:                        if (rpc_srv_blobMap(s, b) != -1) {
                    895:                                /* deliver BLOB variable to client */
1.10      misho     896:                                blob.hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
1.7       misho     897:                                rpc_srv_blobUnmap(b);
                    898:                        } else {
1.10      misho     899:                                blob.hdr_cmd = error;
                    900:                                blob.hdr_ret = RPC_ERROR(-1);
1.7       misho     901:                        }
                    902:                        break;
                    903:                case set:
1.17      misho     904:                        if ((b = rpc_srv_registerBLOB(s, ntohl(blob.hdr_len), 
                    905:                                                        ntohl(blob.hdr_ret)))) {
1.7       misho     906:                                /* set new BLOB variable for reply :) */
1.10      misho     907:                                blob.hdr_var = htonl(b->blob_var);
1.7       misho     908: 
                    909:                                /* receive BLOB from client */
1.10      misho     910:                                blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
1.7       misho     911:                                rpc_srv_blobUnmap(b);
1.5       misho     912:                        } else {
1.10      misho     913:                                blob.hdr_cmd = error;
                    914:                                blob.hdr_ret = RPC_ERROR(-1);
1.7       misho     915:                        }
                    916:                        break;
                    917:                case unset:
1.11      misho     918:                        if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) {
1.10      misho     919:                                blob.hdr_cmd = error;
                    920:                                blob.hdr_ret = RPC_ERROR(-1);
1.1       misho     921:                        }
                    922:                        break;
1.7       misho     923:                default:
1.10      misho     924:                        rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);
                    925:                        blob.hdr_cmd = error;
                    926:                        blob.hdr_ret = RPC_ERROR(-1);
1.7       misho     927:        }
1.1       misho     928: 
1.7       misho     929: end:
1.10      misho     930:        memcpy(AIT_ADDR(&c->cli_buf), &blob, sizeof blob);
                    931:        schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), NULL, 0);
                    932:        schedReadSelf(task);
1.7       misho     933:        return NULL;
1.2       misho     934: }
                    935: 
                    936: static void *
1.17      misho     937: flushBLOB(sched_task_t *task)
                    938: {
1.23      misho     939:        uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg);
                    940:        rpc_srv_t *srv = sigArg ? (void*) sigArg : TASK_ARG(task);
1.17      misho     941:        rpc_blob_t *b, *tmp;
                    942: 
                    943:        TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
                    944:                TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);
                    945: 
                    946:                rpc_srv_blobFree(srv, b);
                    947:                e_free(b);
                    948:        }
                    949: 
1.23      misho     950:        if (!schedSignalSelf(task)) {
                    951:                /* disabled kqueue support in libaitsched */
                    952:                struct sigaction sa;
                    953: 
                    954:                memset(&sa, 0, sizeof sa);
                    955:                sigemptyset(&sa.sa_mask);
                    956:                sa.sa_handler = (void (*)(int)) flushBLOB;
                    957:                sa.sa_flags = SA_RESTART | SA_RESETHAND;
                    958:                sigaction(SIGFBLOB, &sa, NULL);
                    959:        }
                    960: 
1.17      misho     961:        return NULL;
                    962: }
                    963: 
                    964: static void *
1.10      misho     965: acceptBLOBClients(sched_task_t *task)
1.2       misho     966: {
1.10      misho     967:        rpc_srv_t *srv = TASK_ARG(task);
                    968:        rpc_cli_t *c = NULL;
                    969:        register int i;
1.14      misho     970:        socklen_t salen = sizeof(sockaddr_t);
1.21      misho     971:        int sock;
1.12      misho     972: #ifdef TCP_NOPUSH
                    973:        int n = 1;
                    974: #endif
1.7       misho     975: 
1.10      misho     976:        /* check free slots for connect */
1.14      misho     977:        for (i = 0; i < array_Size(srv->srv_blob.clients) && 
                    978:                        (c = array(srv->srv_blob.clients, i, rpc_cli_t*)); i++);
1.21      misho     979:        if (c) {        /* no more free slots! */
                    980:                EVERBOSE(1, "BLOB client quota exceeded! Connection will be shutdown!\n");
                    981:                if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
                    982:                        shutdown(sock, SHUT_RDWR);
                    983:                        close(sock);
                    984:                }
1.10      misho     985:                goto end;
1.21      misho     986:        }
                    987: 
1.14      misho     988:        c = e_malloc(sizeof(rpc_cli_t));
1.10      misho     989:        if (!c) {
1.7       misho     990:                LOGERR;
1.10      misho     991:                srv->srv_kill = srv->srv_blob.kill = 1;
1.7       misho     992:                return NULL;
                    993:        } else {
1.10      misho     994:                memset(c, 0, sizeof(rpc_cli_t));
1.14      misho     995:                array_Set(srv->srv_blob.clients, i, c);
1.10      misho     996:                c->cli_id = i;
                    997:                c->cli_parent = srv;
1.7       misho     998:        }
1.4       misho     999: 
1.10      misho    1000:        /* alloc empty buffer */
1.14      misho    1001:        AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
1.2       misho    1002: 
1.10      misho    1003:        /* accept client */
                   1004:        c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
                   1005:        if (c->cli_sock == -1) {
                   1006:                LOGERR;
                   1007:                AIT_FREE_VAL(&c->cli_buf);
1.14      misho    1008:                array_Del(srv->srv_blob.clients, i, 42);
1.10      misho    1009:                goto end;
1.12      misho    1010:        } else {
                   1011: #ifdef TCP_NOPUSH
                   1012:                setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n);
                   1013: #endif
1.10      misho    1014:                fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1.12      misho    1015:        }
1.2       misho    1016: 
1.10      misho    1017:        schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0);
                   1018: end:
                   1019:        schedReadSelf(task);
1.7       misho    1020:        return NULL;
1.1       misho    1021: }
                   1022: 
1.10      misho    1023: /* ------------------------------------------------------ */
1.1       misho    1024: 
                   1025: /*
1.7       misho    1026:  * rpc_srv_initBLOBServer() - Init & create BLOB Server
                   1027:  *
1.4       misho    1028:  * @srv = RPC server instance
1.2       misho    1029:  * @Port = Port for bind server, if Port == 0 default port is selected
                   1030:  * @diskDir = Disk place for BLOB file objects
                   1031:  * return: -1 == error or 0 bind and created BLOB server instance
                   1032:  */
                   1033: int
                   1034: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
                   1035: {
                   1036:        int n = 1;
                   1037: 
1.10      misho    1038:        if (!srv || srv->srv_kill) {
1.7       misho    1039:                rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
1.2       misho    1040:                return -1;
                   1041:        }
                   1042: 
                   1043:        memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
                   1044:        if (access(diskDir, R_OK | W_OK) == -1) {
                   1045:                LOGERR;
                   1046:                return -1;
                   1047:        } else
1.7       misho    1048:                AIT_SET_STR(&srv->srv_blob.dir, diskDir);
1.2       misho    1049: 
1.10      misho    1050:        /* init blob list */
                   1051:        TAILQ_INIT(&srv->srv_blob.blobs);
                   1052: 
1.2       misho    1053:        srv->srv_blob.server.cli_parent = srv;
1.4       misho    1054: 
1.14      misho    1055:        memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(sockaddr_t));
1.10      misho    1056:        switch (srv->srv_blob.server.cli_sa.sa.sa_family) {
1.4       misho    1057:                case AF_INET:
1.10      misho    1058:                        srv->srv_blob.server.cli_sa.sin.sin_port = 
                   1059:                                htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin.sin_port) + 1);
1.4       misho    1060:                        break;
                   1061:                case AF_INET6:
1.10      misho    1062:                        srv->srv_blob.server.cli_sa.sin6.sin6_port = 
                   1063:                                htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin6.sin6_port) + 1);
1.4       misho    1064:                        break;
                   1065:                case AF_LOCAL:
1.10      misho    1066:                        strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob", 
                   1067:                                        sizeof srv->srv_blob.server.cli_sa.sun.sun_path);
1.4       misho    1068:                        break;
                   1069:                default:
1.7       misho    1070:                        AIT_FREE_VAL(&srv->srv_blob.dir);
1.4       misho    1071:                        return -1;
1.2       misho    1072:        }
                   1073: 
1.4       misho    1074:        /* create BLOB server socket */
1.6       misho    1075:        srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
1.2       misho    1076:        if (srv->srv_blob.server.cli_sock == -1) {
                   1077:                LOGERR;
1.7       misho    1078:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho    1079:                return -1;
                   1080:        }
                   1081:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
                   1082:                LOGERR;
                   1083:                close(srv->srv_blob.server.cli_sock);
1.7       misho    1084:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho    1085:                return -1;
                   1086:        }
1.5       misho    1087:        n = srv->srv_netbuf;
                   1088:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
                   1089:                LOGERR;
                   1090:                close(srv->srv_blob.server.cli_sock);
1.7       misho    1091:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.5       misho    1092:                return -1;
                   1093:        }
                   1094:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
                   1095:                LOGERR;
                   1096:                close(srv->srv_blob.server.cli_sock);
1.7       misho    1097:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.5       misho    1098:                return -1;
                   1099:        }
1.6       misho    1100:        if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa, 
                   1101:                                srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
1.2       misho    1102:                LOGERR;
                   1103:                close(srv->srv_blob.server.cli_sock);
1.7       misho    1104:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho    1105:                return -1;
1.13      misho    1106:        } else
                   1107:                fcntl(srv->srv_blob.server.cli_sock, F_SETFL, 
                   1108:                                fcntl(srv->srv_blob.server.cli_sock, F_GETFL) | O_NONBLOCK);
                   1109: 
1.2       misho    1110: 
1.10      misho    1111:        /* allocate pool for concurent blob clients */
1.14      misho    1112:        srv->srv_blob.clients = array_Init(array_Size(srv->srv_clients));
1.2       misho    1113:        if (!srv->srv_blob.clients) {
1.14      misho    1114:                rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1.2       misho    1115:                close(srv->srv_blob.server.cli_sock);
1.7       misho    1116:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho    1117:                return -1;
1.10      misho    1118:        }
1.2       misho    1119: 
1.10      misho    1120:        /* init blob scheduler */
                   1121:        srv->srv_blob.root = schedBegin();
                   1122:        if (!srv->srv_blob.root) {
                   1123:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1.14      misho    1124:                array_Destroy(&srv->srv_blob.clients);
1.10      misho    1125:                close(srv->srv_blob.server.cli_sock);
                   1126:                AIT_FREE_VAL(&srv->srv_blob.dir);
                   1127:                return -1;
                   1128:        }
1.2       misho    1129: 
                   1130:        return 0;
                   1131: }
                   1132: 
                   1133: /*
1.7       misho    1134:  * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
                   1135:  *
1.2       misho    1136:  * @srv = RPC Server instance
                   1137:  * return: none
                   1138:  */
1.16      misho    1139: void
1.2       misho    1140: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
                   1141: {
1.10      misho    1142:        if (!srv)
1.2       misho    1143:                return;
                   1144: 
1.10      misho    1145:        srv->srv_blob.kill = 1;
1.17      misho    1146: 
                   1147:        schedEnd(&srv->srv_blob.root);
1.2       misho    1148: }
                   1149: 
                   1150: /*
1.10      misho    1151:  * rpc_srv_loopBLOBServer() - Execute Main BLOB server loop and wait for clients requests
1.7       misho    1152:  *
1.2       misho    1153:  * @srv = RPC Server instance
                   1154:  * return: -1 error or 0 ok, infinite loop ...
                   1155:  */
                   1156: int
1.10      misho    1157: rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
1.2       misho    1158: {
1.10      misho    1159:        rpc_cli_t *c;
1.2       misho    1160:        register int i;
1.10      misho    1161:        rpc_blob_t *b, *tmp;
                   1162:        struct timespec ts = { RPC_SCHED_POLLING, 0 };
1.2       misho    1163: 
1.10      misho    1164:        if (!srv || srv->srv_kill) {
1.7       misho    1165:                rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
1.2       misho    1166:                return -1;
                   1167:        }
                   1168: 
1.14      misho    1169:        if (listen(srv->srv_blob.server.cli_sock, array_Size(srv->srv_blob.clients)) == -1) {
1.2       misho    1170:                LOGERR;
                   1171:                return -1;
1.10      misho    1172:        }
                   1173: 
1.23      misho    1174:        if (!schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0)) {
                   1175:                /* disabled kqueue support in libaitsched */
                   1176:                struct sigaction sa;
                   1177: 
                   1178:                atomic_store_rel_ptr(&_glSigArg, (uintptr_t) srv);
                   1179: 
                   1180:                memset(&sa, 0, sizeof sa);
                   1181:                sigemptyset(&sa.sa_mask);
                   1182:                sa.sa_handler = (void (*)(int)) flushBLOB;
                   1183:                sa.sa_flags = SA_RESTART | SA_RESETHAND;
                   1184:                sigaction(SIGFBLOB, &sa, NULL);
                   1185:        }
                   1186: 
1.10      misho    1187:        if (!schedRead(srv->srv_blob.root, acceptBLOBClients, srv, 
                   1188:                                srv->srv_blob.server.cli_sock, NULL, 0)) {
                   1189:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                   1190:                return -1;
                   1191:        }
1.2       misho    1192: 
1.10      misho    1193:        schedPolling(srv->srv_blob.root, &ts, NULL);
                   1194:        /* main rpc loop */
                   1195:        schedRun(srv->srv_blob.root, &srv->srv_blob.kill);
1.7       misho    1196: 
1.17      misho    1197:        /* detach blobs */
                   1198:        TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
                   1199:                TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);
                   1200: 
                   1201:                rpc_srv_blobFree(srv, b);
                   1202:                e_free(b);
                   1203:        }
                   1204: 
1.10      misho    1205:        /* close all clients connections & server socket */
1.14      misho    1206:        for (i = 0; i < array_Size(srv->srv_blob.clients); i++) {
                   1207:                c = array(srv->srv_blob.clients, i, rpc_cli_t*);
1.10      misho    1208:                if (c) {
                   1209:                        shutdown(c->cli_sock, SHUT_RDWR);
                   1210:                        close(c->cli_sock);
                   1211: 
                   1212:                        schedCancelby(srv->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
                   1213:                        AIT_FREE_VAL(&c->cli_buf);
1.2       misho    1214:                }
1.14      misho    1215:                array_Del(srv->srv_blob.clients, i, 42);
1.10      misho    1216:        }
1.14      misho    1217:        array_Destroy(&srv->srv_blob.clients);
1.2       misho    1218: 
1.10      misho    1219:        close(srv->srv_blob.server.cli_sock);
1.2       misho    1220: 
1.10      misho    1221:        AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho    1222:        return 0;
                   1223: }
                   1224: 
                   1225: 
                   1226: /*
1.7       misho    1227:  * rpc_srv_initServer() - Init & create RPC Server
                   1228:  *
1.15      misho    1229:  * @InstID = Instance for authentication & recognition
1.1       misho    1230:  * @concurentClients = Concurent clients at same time to this server
1.10      misho    1231:  * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
1.4       misho    1232:  * @csHost = Host name or address for bind server, if NULL any address
1.1       misho    1233:  * @Port = Port for bind server, if Port == 0 default port is selected
1.13      misho    1234:  * @proto = Protocol, if == 0 choose SOCK_STREAM
1.1       misho    1235:  * return: NULL == error or !=NULL bind and created RPC server instance
                   1236:  */
                   1237: rpc_srv_t *
1.15      misho    1238: rpc_srv_initServer(u_char InstID, int concurentClients, int netBuf, 
                   1239:                const char *csHost, u_short Port, int proto)
1.1       misho    1240: {
1.10      misho    1241:        int n = 1;
1.1       misho    1242:        rpc_srv_t *srv = NULL;
1.14      misho    1243:        sockaddr_t sa = E_SOCKADDR_INIT;
1.1       misho    1244: 
1.15      misho    1245:        if (!concurentClients || (proto < 0 || proto > SOCK_DGRAM)) {
1.10      misho    1246:                rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1.1       misho    1247:                return NULL;
                   1248:        }
1.14      misho    1249:        if (!e_gethostbyname(csHost, Port, &sa))
1.10      misho    1250:                return NULL;
1.1       misho    1251:        if (!Port)
                   1252:                Port = RPC_DEFPORT;
1.13      misho    1253:        if (!proto)
                   1254:                proto = SOCK_STREAM;
1.10      misho    1255:        if (netBuf < RPC_MIN_BUFSIZ)
1.5       misho    1256:                netBuf = BUFSIZ;
1.7       misho    1257:        else
1.14      misho    1258:                netBuf = E_ALIGN(netBuf, 2);    /* align netBuf length */
1.10      misho    1259: 
                   1260: #ifdef HAVE_SRANDOMDEV
                   1261:        srandomdev();
                   1262: #else
                   1263:        time_t tim;
                   1264: 
                   1265:        srandom((time(&tim) ^ getpid()));
                   1266: #endif
1.1       misho    1267: 
1.14      misho    1268:        srv = e_malloc(sizeof(rpc_srv_t));
1.1       misho    1269:        if (!srv) {
                   1270:                LOGERR;
                   1271:                return NULL;
                   1272:        } else
                   1273:                memset(srv, 0, sizeof(rpc_srv_t));
                   1274: 
1.13      misho    1275:        srv->srv_proto = proto;
1.5       misho    1276:        srv->srv_netbuf = netBuf;
1.1       misho    1277:        srv->srv_session.sess_version = RPC_VERSION;
1.15      misho    1278:        srv->srv_session.sess_instance = InstID;
1.1       misho    1279: 
                   1280:        srv->srv_server.cli_parent = srv;
1.10      misho    1281:        memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
                   1282: 
1.12      misho    1283:        /* init functions */
                   1284:        pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
                   1285:        SLIST_INIT(&srv->srv_funcs);
                   1286:        AVL_INIT(&srv->srv_funcs);
1.10      misho    1287: 
                   1288:        /* init scheduler */
                   1289:        srv->srv_root = schedBegin();
                   1290:        if (!srv->srv_root) {
                   1291:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1.12      misho    1292:                pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14      misho    1293:                e_free(srv);
1.10      misho    1294:                return NULL;
                   1295:        }
                   1296: 
                   1297:        /* init pool for clients */
1.14      misho    1298:        srv->srv_clients = array_Init(concurentClients);
1.10      misho    1299:        if (!srv->srv_clients) {
1.14      misho    1300:                rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1.10      misho    1301:                schedEnd(&srv->srv_root);
1.12      misho    1302:                pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14      misho    1303:                e_free(srv);
1.10      misho    1304:                return NULL;
                   1305:        }
1.4       misho    1306: 
                   1307:        /* create server socket */
1.13      misho    1308:        srv->srv_server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, srv->srv_proto, 0);
1.1       misho    1309:        if (srv->srv_server.cli_sock == -1) {
                   1310:                LOGERR;
1.14      misho    1311:                array_Destroy(&srv->srv_clients);
1.10      misho    1312:                schedEnd(&srv->srv_root);
1.12      misho    1313:                pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14      misho    1314:                e_free(srv);
1.1       misho    1315:                return NULL;
                   1316:        }
                   1317:        if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
                   1318:                LOGERR;
1.10      misho    1319:                goto err;
1.1       misho    1320:        }
1.5       misho    1321:        n = srv->srv_netbuf;
                   1322:        if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
                   1323:                LOGERR;
1.10      misho    1324:                goto err;
1.5       misho    1325:        }
                   1326:        if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
                   1327:                LOGERR;
1.10      misho    1328:                goto err;
1.5       misho    1329:        }
1.6       misho    1330:        if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa, 
                   1331:                                srv->srv_server.cli_sa.sa.sa_len) == -1) {
1.1       misho    1332:                LOGERR;
1.10      misho    1333:                goto err;
1.13      misho    1334:        } else
                   1335:                fcntl(srv->srv_server.cli_sock, F_SETFL, 
                   1336:                                fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
1.1       misho    1337: 
1.10      misho    1338:        rpc_register_srvPing(srv);
1.8       misho    1339: 
1.1       misho    1340:        return srv;
1.10      misho    1341: err:   /* error condition */
                   1342:        close(srv->srv_server.cli_sock);
1.14      misho    1343:        array_Destroy(&srv->srv_clients);
1.10      misho    1344:        schedEnd(&srv->srv_root);
1.12      misho    1345:        pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14      misho    1346:        e_free(srv);
1.10      misho    1347:        return NULL;
1.1       misho    1348: }
                   1349: 
                   1350: /*
1.7       misho    1351:  * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
                   1352:  *
1.6       misho    1353:  * @psrv = RPC Server instance
1.1       misho    1354:  * return: none
                   1355:  */
1.16      misho    1356: void
1.6       misho    1357: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
1.1       misho    1358: {
1.10      misho    1359:        if (!psrv || !*psrv)
1.1       misho    1360:                return;
                   1361: 
1.10      misho    1362:        /* if send kill to blob server */
1.17      misho    1363:        rpc_srv_endBLOBServer(*psrv);
1.1       misho    1364: 
1.10      misho    1365:        (*psrv)->srv_kill = 1;
                   1366:        sleep(RPC_SCHED_POLLING);
1.2       misho    1367: 
1.17      misho    1368:        schedEnd(&(*psrv)->srv_root);
                   1369: 
1.12      misho    1370:        pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx);
1.14      misho    1371:        e_free(*psrv);
1.6       misho    1372:        *psrv = NULL;
1.1       misho    1373: }
                   1374: 
                   1375: /*
1.7       misho    1376:  * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
                   1377:  *
1.1       misho    1378:  * @srv = RPC Server instance
                   1379:  * return: -1 error or 0 ok, infinite loop ...
                   1380:  */
                   1381: int
1.5       misho    1382: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
1.1       misho    1383: {
1.10      misho    1384:        rpc_cli_t *c;
1.1       misho    1385:        register int i;
1.12      misho    1386:        rpc_func_t *f;
1.10      misho    1387:        struct timespec ts = { RPC_SCHED_POLLING, 0 };
1.1       misho    1388: 
                   1389:        if (!srv) {
1.10      misho    1390:                rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
1.1       misho    1391:                return -1;
                   1392:        }
                   1393: 
1.13      misho    1394:        if (srv->srv_proto == SOCK_STREAM)
1.14      misho    1395:                if (listen(srv->srv_server.cli_sock, array_Size(srv->srv_clients)) == -1) {
1.13      misho    1396:                        LOGERR;
                   1397:                        return -1;
                   1398:                }
1.1       misho    1399: 
1.13      misho    1400:        if (!schedRead(srv->srv_root, cbProto[srv->srv_proto][CB_ACCEPTCLIENT], srv, 
                   1401:                                srv->srv_server.cli_sock, NULL, 0)) {
1.10      misho    1402:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                   1403:                return -1;
                   1404:        }
1.7       misho    1405: 
1.10      misho    1406:        schedPolling(srv->srv_root, &ts, NULL);
1.7       misho    1407:        /* main rpc loop */
1.10      misho    1408:        schedRun(srv->srv_root, &srv->srv_kill);
                   1409: 
                   1410:        /* close all clients connections & server socket */
1.14      misho    1411:        for (i = 0; i < array_Size(srv->srv_clients); i++) {
                   1412:                c = array(srv->srv_clients, i, rpc_cli_t*);
1.10      misho    1413:                if (c) {
1.23.6.3  misho    1414:                        if (srv->srv_proto == SOCK_STREAM) {
                   1415:                                shutdown(c->cli_sock, SHUT_RDWR);
                   1416:                                close(c->cli_sock);
                   1417:                        }
1.10      misho    1418: 
                   1419:                        schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
1.14      misho    1420:                        ait_freeVars(&RPC_RETVARS(c));
1.10      misho    1421:                        AIT_FREE_VAL(&c->cli_buf);
1.1       misho    1422:                }
1.14      misho    1423:                array_Del(srv->srv_clients, i, 42);
1.10      misho    1424:        }
1.14      misho    1425:        array_Destroy(&srv->srv_clients);
1.2       misho    1426: 
1.10      misho    1427:        close(srv->srv_server.cli_sock);
1.2       misho    1428: 
1.10      misho    1429:        /* detach exported calls */
1.12      misho    1430:        RPC_FUNCS_LOCK(&srv->srv_funcs);
                   1431:        while ((f = SLIST_FIRST(&srv->srv_funcs))) {
                   1432:                SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);
1.1       misho    1433: 
1.10      misho    1434:                AIT_FREE_VAL(&f->func_name);
1.14      misho    1435:                e_free(f);
1.1       misho    1436:        }
1.12      misho    1437:        srv->srv_funcs.avlh_root = NULL;
                   1438:        RPC_FUNCS_UNLOCK(&srv->srv_funcs);
1.1       misho    1439: 
                   1440:        return 0;
                   1441: }
                   1442: 
                   1443: 
                   1444: /*
                   1445:  * rpc_srv_execCall() Execute registered call from RPC server
1.7       misho    1446:  *
1.10      misho    1447:  * @cli = RPC client
1.1       misho    1448:  * @rpc = IN RPC call structure
1.10      misho    1449:  * @funcname = Execute RPC function
1.5       misho    1450:  * @args = IN RPC calling arguments from RPC client
1.1       misho    1451:  * return: -1 error, !=-1 ok
                   1452:  */
                   1453: int
1.10      misho    1454: rpc_srv_execCall(rpc_cli_t * __restrict cli, struct tagRPCCall * __restrict rpc, 
                   1455:                ait_val_t funcname, array_t * __restrict args)
1.1       misho    1456: {
                   1457:        rpc_callback_t func;
                   1458: 
1.10      misho    1459:        if (!cli || !rpc || !AIT_ADDR(&funcname)) {
1.7       misho    1460:                rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1.1       misho    1461:                return -1;
                   1462:        }
                   1463: 
1.10      misho    1464:        func = AIT_GET_LIKE(&funcname, rpc_callback_t);
                   1465:        return func(cli, rpc, args);
1.1       misho    1466: }
1.23.6.1  misho    1467: 
                   1468: 
                   1469: /*
                   1470:  * rpc_srv_initServer2() - Init & create layer2 RPC Server
                   1471:  *
                   1472:  * @InstID = Instance for authentication & recognition
                   1473:  * @concurentClients = Concurent clients at same time to this server
                   1474:  * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
                   1475:  * @csIface = Interface name for bind server, if NULL first interface on host
                   1476:  * return: NULL == error or !=NULL bind and created RPC server instance
                   1477:  */
                   1478: rpc_srv_t *
1.23.6.2  misho    1479: rpc_srv_initServer2(u_char InstID, int concurentClients, int netBuf, const char *csIface)
1.23.6.1  misho    1480: {
                   1481:        int n = 1;
                   1482:        rpc_srv_t *srv = NULL;
                   1483:        sockaddr_t sa = E_SOCKADDR_INIT;
                   1484:        char szIface[64], szStr[STRSIZ];
                   1485:        register int i;
                   1486:        struct ifreq ifr;
1.23.6.3  misho    1487:        struct bpf_insn insns[] = {
                   1488:                BPF_STMT(BPF_LD + BPF_H + BPF_ABS, 12),
                   1489:                BPF_JUMP(BPF_JMP + BPF_JEQ + BPF_K, RPC_DEFPORT, 0, 1),
                   1490:                BPF_STMT(BPF_RET + BPF_K, -1),
                   1491:                BPF_STMT(BPF_RET + BPF_K, 0),
                   1492:        };
                   1493:        struct bpf_program fcode = { 
                   1494:                .bf_len = sizeof(insns) / sizeof(struct bpf_insn), 
                   1495:                        .bf_insns = insns
                   1496:        };
1.23.6.1  misho    1497: 
                   1498:        if (!concurentClients) {
                   1499:                rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
                   1500:                return NULL;
                   1501:        }
                   1502:        if (!csIface) {
                   1503:                if (e_get1stiface(szIface, sizeof szIface))
                   1504:                        return NULL;
                   1505:        } else
                   1506:                strlcpy(szIface, csIface, sizeof szIface);
1.23.6.3  misho    1507:        if (!e_getifacebyname(szIface, &sa))
1.23.6.1  misho    1508:                return NULL;
                   1509: 
                   1510: #ifdef HAVE_SRANDOMDEV
                   1511:        srandomdev();
                   1512: #else
                   1513:        time_t tim;
                   1514: 
                   1515:        srandom((time(&tim) ^ getpid()));
                   1516: #endif
                   1517: 
                   1518:        srv = e_malloc(sizeof(rpc_srv_t));
                   1519:        if (!srv) {
                   1520:                LOGERR;
                   1521:                return NULL;
                   1522:        } else
                   1523:                memset(srv, 0, sizeof(rpc_srv_t));
                   1524: 
1.23.6.2  misho    1525:        srv->srv_proto = SOCK_BPF;
1.23.6.1  misho    1526:        srv->srv_netbuf = netBuf;
                   1527:        srv->srv_session.sess_version = RPC_VERSION;
                   1528:        srv->srv_session.sess_instance = InstID;
                   1529: 
                   1530:        srv->srv_server.cli_parent = srv;
                   1531:        memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
                   1532: 
                   1533:        /* init functions */
                   1534:        pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
                   1535:        SLIST_INIT(&srv->srv_funcs);
                   1536:        AVL_INIT(&srv->srv_funcs);
                   1537: 
                   1538:        /* init scheduler */
                   1539:        srv->srv_root = schedBegin();
                   1540:        if (!srv->srv_root) {
                   1541:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                   1542:                pthread_mutex_destroy(&srv->srv_funcs.mtx);
                   1543:                e_free(srv);
                   1544:                return NULL;
                   1545:        }
                   1546: 
                   1547:        /* init pool for clients */
                   1548:        srv->srv_clients = array_Init(concurentClients);
                   1549:        if (!srv->srv_clients) {
                   1550:                rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
                   1551:                schedEnd(&srv->srv_root);
                   1552:                pthread_mutex_destroy(&srv->srv_funcs.mtx);
                   1553:                e_free(srv);
                   1554:                return NULL;
                   1555:        }
                   1556: 
                   1557:        /* create server handler */
                   1558:        for (i = 0; i < 10; i++) {
                   1559:                memset(szStr, 0, sizeof szStr);
                   1560:                snprintf(szStr, sizeof szStr, "/dev/bpf%d", i);
                   1561:                srv->srv_server.cli_sock = open(szStr, O_RDWR);
                   1562:                if (srv->srv_server.cli_sock > STDERR_FILENO)
                   1563:                        break;
                   1564:        }
                   1565:        if (srv->srv_server.cli_sock < 3) {
                   1566:                LOGERR;
                   1567:                array_Destroy(&srv->srv_clients);
                   1568:                schedEnd(&srv->srv_root);
                   1569:                pthread_mutex_destroy(&srv->srv_funcs.mtx);
                   1570:                e_free(srv);
                   1571:                return NULL;
                   1572:        }
                   1573: 
                   1574:        if (ioctl(srv->srv_server.cli_sock, BIOCIMMEDIATE, &n) == -1) {
                   1575:                LOGERR;
                   1576:                goto err;
                   1577:        }
1.23.6.3  misho    1578:        if (ioctl(srv->srv_server.cli_sock, BIOCSETF, &fcode) == -1) {
                   1579:                LOGERR;
                   1580:                goto err;
                   1581:        }
1.23.6.1  misho    1582:        n = (netBuf < RPC_MIN_BUFSIZ) ? getpagesize() : E_ALIGN(netBuf, 2);
                   1583:        if (ioctl(srv->srv_server.cli_sock, BIOCSBLEN, &n) == -1) {
                   1584:                LOGERR;
                   1585:                goto err;
                   1586:        } else
                   1587:                srv->srv_netbuf = n;
                   1588: 
                   1589:        memset(&ifr, 0, sizeof ifr);
                   1590:        strlcpy(ifr.ifr_name, szIface, sizeof ifr.ifr_name);
                   1591:        if (ioctl(srv->srv_server.cli_sock, BIOCSETIF, &ifr) == -1) {
                   1592:                LOGERR;
                   1593:                goto err;
                   1594:        } else
                   1595:                fcntl(srv->srv_server.cli_sock, F_SETFL, 
                   1596:                                fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
                   1597: 
                   1598:        rpc_register_srvPing(srv);
                   1599: 
                   1600:        return srv;
                   1601: err:   /* error condition */
                   1602:        close(srv->srv_server.cli_sock);
                   1603:        array_Destroy(&srv->srv_clients);
                   1604:        schedEnd(&srv->srv_root);
                   1605:        pthread_mutex_destroy(&srv->srv_funcs.mtx);
                   1606:        e_free(srv);
                   1607:        return NULL;
                   1608: }

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