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

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

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