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

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

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