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

1.1       misho       1: /*************************************************************************
                      2: * (C) 2010 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
                      3: *  by Michael Pounov <misho@openbsd-bg.org>
                      4: *
                      5: * $Author: misho $
1.9.2.6 ! misho       6: * $Id: srv.c,v 1.9.2.5 2012/05/15 16:10:58 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.9.2.4   misho      49: static void *
                     50: closeClient(sched_task_t *task)
                     51: {
                     52:        rpc_cli_t *c = TASK_ARG(task);
                     53:        rpc_srv_t *s = c->cli_parent;
                     54: 
1.9.2.6 ! misho      55:        schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, TASK_ARG(task), NULL);
1.9.2.4   misho      56: 
                     57:        /* close client socket */
                     58:        if (TASK_VAL(task))
                     59:                shutdown(c->cli_sock, SHUT_RDWR);
                     60:        close(c->cli_sock);
                     61: 
                     62:        /* free buffer */
                     63:        AIT_FREE_VAL(&c->cli_buf);
                     64: 
                     65:        io_arrayDel(s->srv_clients, c->cli_id, 42);
                     66:        return NULL;
                     67: }
1.7       misho      68: 
                     69: static void *
                     70: txPacket(sched_task_t *task)
                     71: {
                     72:        rpc_cli_t *c = TASK_ARG(task);
                     73:        rpc_srv_t *s = c->cli_parent;
                     74:        rpc_func_t *f = NULL;
1.9.2.4   misho      75:        u_char buf[USHRT_MAX] = { 0 };
1.7       misho      76:        struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
                     77:        int ret, wlen = sizeof(struct tagRPCCall);
                     78:        array_t *arr = NULL;
                     79: 
1.9.2.4   misho      80:        /* copy RPC header */
                     81:        memcpy(buf, TASK_DATA(task), wlen);
                     82: 
1.7       misho      83:        if (rpc->call_argc) {
1.9.2.6 ! misho      84:                f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
1.7       misho      85:                if (!f) {
                     86:                        rpc->call_argc ^= rpc->call_argc;
                     87:                        rpc->call_rep.ret = RPC_ERROR(-1);
                     88:                        rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                     89:                } else {
                     90:                        rpc->call_argc = htons(rpc_srv_getVars(f, &arr));
                     91:                        /* Go Encapsulate variables */
1.9.2.4   misho      92:                        ret = io_vars2buffer(buf + wlen, sizeof buf - wlen, arr);
                     93:                        /* Free return values */
1.7       misho      94:                        io_clrVars(f->func_vars);
                     95:                        if (ret == -1) {
                     96:                                rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
                     97:                                rpc->call_argc ^= rpc->call_argc;
                     98:                                rpc->call_rep.ret = RPC_ERROR(-1);
                     99:                                rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    100:                        } else
                    101:                                wlen += ret;
                    102:                }
                    103:        }
                    104: 
1.8       misho     105:        rpc->call_len = htons(wlen);
                    106: 
1.7       misho     107:        /* calculate CRC */
                    108:        rpc->call_crc ^= rpc->call_crc;
1.8       misho     109:        rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
1.7       misho     110: 
                    111:        /* send reply */
1.9.2.4   misho     112:        ret = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);
                    113:        if (ret == -1 || ret != wlen) {
                    114:                /* close connection */
                    115:                schedEvent(TASK_ROOT(task), closeClient, c, 42, NULL, 0);
                    116:        }
1.7       misho     117: 
                    118:        return NULL;
                    119: }
                    120: 
                    121: static void *
                    122: execCall(sched_task_t *task)
                    123: {
                    124:        rpc_cli_t *c = TASK_ARG(task);
                    125:        rpc_srv_t *s = c->cli_parent;
                    126:        rpc_func_t *f = NULL;
                    127:        array_t *arr = NULL;
1.9.2.4   misho     128:        u_char *buf = AIT_GET_BUF(&c->cli_buf) + TASK_VAL(task);
1.7       misho     129:        struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
                    130:        int argc = ntohs(rpc->call_argc);
                    131: 
                    132:        /* Go decapsulate variables ... */
1.8       misho     133:        if (argc) {
1.7       misho     134:                arr = io_buffer2vars(buf + sizeof(struct tagRPCCall), 
1.9.2.4   misho     135:                                AIT_LEN(&c->cli_buf) - TASK_VAL(task) - sizeof(struct tagRPCCall), 
1.8       misho     136:                                argc, 1);
1.7       misho     137:                if (!arr) {
                    138:                        rpc_SetErr(ERPCMISMATCH, "#%d - %s", io_GetErrno(), io_GetError());
                    139:                        rpc->call_argc ^= rpc->call_argc;
                    140:                        rpc->call_rep.ret = RPC_ERROR(-1);
                    141:                        rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    142:                        return NULL;
                    143:                }
                    144:        }
                    145: 
1.9.2.6 ! misho     146:        if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag)))) {
1.7       misho     147:                rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
                    148:                rpc->call_argc ^= rpc->call_argc;
                    149:                rpc->call_rep.ret = RPC_ERROR(-1);
                    150:                rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
                    151:        } else {
1.8       misho     152:                /* if client doesn't want reply */
                    153:                argc = rpc->call_req.flags & RPC_NOREPLY;
1.7       misho     154:                rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));
                    155:                if (rpc->call_rep.ret == htonl(-1)) {
                    156:                        rpc->call_rep.eno = RPC_ERROR(errno);
                    157:                        rpc->call_argc ^= rpc->call_argc;
                    158:                } else {
                    159:                        rpc->call_rep.eno ^= rpc->call_rep.eno;
1.8       misho     160:                        if (argc) {
                    161:                                /* without reply */
                    162:                                io_clrVars(f->func_vars);
                    163:                                rpc->call_argc ^= rpc->call_argc;
                    164:                        } else
                    165:                                rpc->call_argc = htons(rpc_srv_getVars(f, NULL));
1.7       misho     166:                }
                    167:        }
                    168: 
                    169:        if (arr)
                    170:                io_arrayDestroy(&arr);
                    171:        return NULL;
                    172: }
                    173: 
                    174: static void *
                    175: rxPacket(sched_task_t *task)
                    176: {
                    177:        rpc_cli_t *c = TASK_ARG(task);
                    178:        rpc_srv_t *s = c->cli_parent;
1.9.2.4   misho     179:        u_char *buf = AIT_GET_BUF(&c->cli_buf);
                    180:        int len, rlen, noreply;
1.8       misho     181:        u_short crc, off = 0;
1.7       misho     182:        struct tagRPCCall *rpc;
                    183: 
1.9.2.4   misho     184:        memset(buf, 0, AIT_LEN(&c->cli_buf));
                    185:        rlen = recv(TASK_FD(task), buf, AIT_LEN(&c->cli_buf), 0);
                    186:        if (rlen < 1) {
                    187:                /* close connection */
                    188:                schedEvent(TASK_ROOT(task), closeClient, c, 0, NULL, 0);
1.7       misho     189:                return NULL;
1.9       misho     190:        }
1.7       misho     191: 
1.8       misho     192:        do {
1.9.2.4   misho     193:                /* check RPC packet */
1.8       misho     194:                if (rlen < sizeof(struct tagRPCCall)) {
                    195:                        rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");
                    196: 
1.9.2.4   misho     197:                        schedReadSelf(task);
1.8       misho     198:                        return NULL;
                    199:                } else
                    200:                        rpc = (struct tagRPCCall*) (buf + off);
                    201: 
1.9.2.4   misho     202:                len = ntohs(rpc->call_len);
                    203:                rlen -= len;
1.8       misho     204: 
                    205:                /* check integrity of packet */
                    206:                crc = ntohs(rpc->call_crc);
                    207:                rpc->call_crc ^= rpc->call_crc;
1.9.2.4   misho     208:                if (crc != crcFletcher16((u_short*) (buf + off), len / 2)) {
1.8       misho     209:                        rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
1.7       misho     210: 
1.9.2.4   misho     211:                        off += len;
1.8       misho     212:                        if (rlen < 1)
                    213:                                break;
                    214:                        else
                    215:                                continue;
                    216:                }
1.7       misho     217: 
1.8       misho     218:                noreply = rpc->call_req.flags & RPC_NOREPLY;
1.7       misho     219: 
1.8       misho     220:                /* check RPC packet session info */
                    221:                if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
                    222:                        rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
                    223:                        rpc->call_argc ^= rpc->call_argc;
                    224:                        rpc->call_rep.ret = RPC_ERROR(-1);
                    225:                        rpc->call_rep.eno = RPC_ERROR(errno);
                    226:                } else {
                    227:                        /* execute RPC call */
1.9.2.4   misho     228:                        schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), off, NULL, 0);
1.8       misho     229:                }
1.7       misho     230: 
1.8       misho     231:                /* send RPC reply */
                    232:                if (!noreply)
1.9.2.4   misho     233:                        schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task), rpc, len);
1.7       misho     234: 
1.9.2.4   misho     235:                off += len;
1.8       misho     236:        } while (rlen > 0);
1.7       misho     237: 
                    238:        /* lets get next packet */
1.9.2.4   misho     239:        schedReadSelf(task);
1.7       misho     240:        return NULL;
                    241: }
                    242: 
1.1       misho     243: static void *
1.9.2.4   misho     244: acceptClients(sched_task_t *task)
1.1       misho     245: {
1.9.2.4   misho     246:        rpc_srv_t *srv = TASK_ARG(task);
                    247:        rpc_cli_t *c = NULL;
                    248:        register int i;
                    249:        socklen_t salen = sizeof(io_sockaddr_t);
1.1       misho     250: 
1.9.2.4   misho     251:        /* check free slots for connect */
                    252:        for (i = 0; i < io_arraySize(srv->srv_clients) && 
                    253:                        (c = io_array(srv->srv_clients, i, rpc_cli_t*)); i++);
                    254:        if (c)  /* no more free slots! */
                    255:                goto end;
                    256:        c = malloc(sizeof(rpc_cli_t));
                    257:        if (!c) {
1.5       misho     258:                LOGERR;
1.9.2.4   misho     259:                srv->srv_kill = kill;
1.7       misho     260:                return NULL;
                    261:        } else {
1.9.2.4   misho     262:                memset(c, 0, sizeof(rpc_cli_t));
                    263:                io_arraySet(srv->srv_clients, i, c);
                    264:                c->cli_id = i;
                    265:                c->cli_parent = srv;
1.7       misho     266:        }
                    267: 
1.9.2.6 ! misho     268:        /* alloc empty buffer */
        !           269:        AIT_SET_BUF2(&c->cli_buf, 0, srv->srv_netbuf);
        !           270: 
1.9.2.4   misho     271:        /* accept client */
                    272:        c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa.sa, &salen);
                    273:        if (c->cli_sock == -1) {
                    274:                LOGERR;
1.9.2.6 ! misho     275:                AIT_FREE_VAL(&c->cli_buf);
1.9.2.4   misho     276:                io_arrayDel(srv->srv_clients, i, 42);
                    277:                goto end;
                    278:        } else
                    279:                fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1.7       misho     280: 
1.9.2.4   misho     281:        schedRead(TASK_ROOT(task), rxPacket, c, c->cli_sock, NULL, 0);
                    282: end:
                    283:        schedReadSelf(task);
1.7       misho     284:        return NULL;
                    285: }
                    286: 
1.9.2.4   misho     287: #if 0
1.7       misho     288: static void *
                    289: txBLOB(sched_task_t *task)
                    290: {
                    291:        u_char *buf = TASK_DATA(task);
                    292:        struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
                    293:        int wlen = sizeof(struct tagBLOBHdr);
                    294: 
                    295:        /* calculate CRC */
                    296:        blob->hdr_crc ^= blob->hdr_crc;
1.8       misho     297:        blob->hdr_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
1.7       misho     298: 
                    299:        /* send reply */
                    300:        wlen = send(TASK_FD(task), buf, wlen, 0);
                    301:        if (wlen == -1)
                    302:                LOGERR;
                    303:        else if (wlen != sizeof(struct tagBLOBHdr))
                    304:                rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
                    305:                                "really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);
                    306: 
                    307:        return NULL;
                    308: }
1.4       misho     309: 
1.7       misho     310: static void *
                    311: rxBLOB(sched_task_t *task)
                    312: {
                    313:        rpc_cli_t *c = TASK_ARG(task);
                    314:        rpc_srv_t *s = c->cli_parent;
                    315:        rpc_blob_t *b;
                    316:        u_char *buf = TASK_DATA(task);
                    317:        struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
                    318:        int rlen;
                    319:        u_short crc;
                    320:        struct timespec ts;
                    321: 
                    322:        /* check for disable service at this moment? */
1.8       misho     323:        if (!s || s->srv_blob.state == disable) {
1.7       misho     324:                usleep(100000);
                    325: #ifdef HAVE_PTHREAD_YIELD
                    326:                pthread_yield();
                    327: #endif
                    328:                schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                    329:                                TASK_DATA(task), TASK_DATLEN(task));
                    330:                return NULL;
                    331:        }
1.5       misho     332: 
1.7       misho     333:        memset(buf, 0, TASK_DATLEN(task));
                    334:        rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
                    335:        if (rlen == -1) {
                    336:                LOGERR;
1.9       misho     337:                c->cli_kill = kill;
1.7       misho     338:                return NULL;
1.9       misho     339:        } else if (!rlen) { /* receive EOF */
                    340:                c->cli_kill = kill;
1.7       misho     341:                return NULL;
1.9       misho     342:        }
1.6       misho     343: 
1.7       misho     344:        if (rlen < sizeof(struct tagBLOBHdr)) {
                    345:                rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet");
                    346:                schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                    347:                                TASK_DATA(task), TASK_DATLEN(task));
                    348:                return NULL;
                    349:        }
1.1       misho     350: 
1.7       misho     351:        /* check integrity of packet */
                    352:        crc = ntohs(blob->hdr_crc);
                    353:        blob->hdr_crc ^= blob->hdr_crc;
1.8       misho     354:        if (crc != crcFletcher16((u_short*) buf, rlen / 2)) {
1.7       misho     355:                rpc_SetErr(ERPCMISMATCH, "Bad CRC BLOB packet");
                    356:                schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                    357:                                TASK_DATA(task), TASK_DATLEN(task));
                    358:                return NULL;
                    359:        }
1.5       misho     360: 
1.7       misho     361:        /* check RPC packet session info */
                    362:        if ((crc = rpc_chkPktSession(&blob->hdr_session, &s->srv_session))) {
                    363:                rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
                    364:                blob->hdr_cmd = error;
                    365:                goto end;
                    366:        } else {
                    367:                /* change socket timeout from last packet */
                    368:                ts.tv_sec = blob->hdr_session.sess_timeout;
                    369:                ts.tv_nsec = 0;
                    370:                schedPolling(TASK_ROOT(task), &ts, NULL);
                    371:        }
                    372: 
                    373:        /* Go to proceed packet ... */
                    374:        switch (blob->hdr_cmd) {
                    375:                case get:
                    376:                        if (!(b = rpc_srv_getBLOB(s, ntohl(blob->hdr_var)))) {
                    377:                                rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob->hdr_var));
                    378:                                blob->hdr_cmd = no;
                    379:                                blob->hdr_ret = RPC_ERROR(-1);
                    380:                                break;
                    381:                        } else
                    382:                                blob->hdr_len = htonl(b->blob_len);
1.5       misho     383: 
1.7       misho     384:                        if (rpc_srv_blobMap(s, b) != -1) {
                    385:                                /* deliver BLOB variable to client */
                    386:                                blob->hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
                    387:                                rpc_srv_blobUnmap(b);
                    388:                        } else {
                    389:                                blob->hdr_cmd = error;
                    390:                                blob->hdr_ret = RPC_ERROR(-1);
                    391:                        }
                    392:                        break;
                    393:                case set:
                    394:                        if ((b = rpc_srv_registerBLOB(s, ntohl(blob->hdr_len)))) {
                    395:                                /* set new BLOB variable for reply :) */
                    396:                                blob->hdr_var = htonl(b->blob_var);
                    397: 
                    398:                                /* receive BLOB from client */
                    399:                                blob->hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
                    400:                                rpc_srv_blobUnmap(b);
1.5       misho     401:                        } else {
1.7       misho     402:                                blob->hdr_cmd = error;
                    403:                                blob->hdr_ret = RPC_ERROR(-1);
                    404:                        }
                    405:                        break;
                    406:                case unset:
                    407:                        if (rpc_srv_unregisterBLOB(s, blob->hdr_var) == -1) {
                    408:                                blob->hdr_cmd = error;
                    409:                                blob->hdr_ret = RPC_ERROR(-1);
1.1       misho     410:                        }
                    411:                        break;
1.7       misho     412:                default:
                    413:                        rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob->hdr_cmd);
                    414:                        blob->hdr_cmd = error;
                    415:                        blob->hdr_ret = RPC_ERROR(-1);
                    416:        }
1.1       misho     417: 
1.7       misho     418: end:
                    419:        schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), 
                    420:                        TASK_DATA(task), TASK_DATLEN(task));
                    421:        schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                    422:                        TASK_DATA(task), TASK_DATLEN(task));
                    423:        return NULL;
1.2       misho     424: }
                    425: 
                    426: static void *
                    427: rpc_srv_dispatchVars(void *arg)
                    428: {
                    429:        rpc_cli_t *c = arg;
                    430:        rpc_srv_t *s;
1.7       misho     431:        sched_root_task_t *root;
                    432:        u_char *buf;
                    433:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
                    434: 
1.2       misho     435:        if (!arg) {
1.7       misho     436:                rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");
1.2       misho     437:                return NULL;
                    438:        } else
                    439:                s = c->cli_parent;
                    440: 
1.7       misho     441:        /* allocate net buffer */
                    442:        buf = malloc(sizeof(struct tagBLOBHdr));
                    443:        if (!buf) {
                    444:                LOGERR;
                    445:                return NULL;
                    446:        }
1.2       misho     447: 
1.7       misho     448:        root = schedBegin();
                    449:        if (!root) {
                    450:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                    451:                free(buf);
                    452:                return NULL;
                    453:        } else {
                    454:                schedTermCondition(root, kill);
                    455:                schedPolling(root, &ts, NULL);
                    456:        }
1.4       misho     457: 
1.7       misho     458:        schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));
1.2       misho     459: 
1.9       misho     460:        schedRun(root, (void*) &c->cli_kill);
1.7       misho     461:        schedEnd(&root);
1.2       misho     462: 
                    463:        shutdown(c->cli_sock, SHUT_RDWR);
                    464:        close(c->cli_sock);
                    465:        memset(c, 0, sizeof(rpc_cli_t));
1.7       misho     466:        free(buf);
                    467:        return NULL;
1.1       misho     468: }
                    469: 
                    470: // -------------------------------------------------
                    471: 
                    472: /*
1.7       misho     473:  * rpc_srv_initBLOBServer() - Init & create BLOB Server
                    474:  *
1.4       misho     475:  * @srv = RPC server instance
1.2       misho     476:  * @Port = Port for bind server, if Port == 0 default port is selected
                    477:  * @diskDir = Disk place for BLOB file objects
                    478:  * return: -1 == error or 0 bind and created BLOB server instance
                    479:  */
                    480: int
                    481: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
                    482: {
                    483:        int n = 1;
1.6       misho     484:        io_sockaddr_t sa;
1.2       misho     485: 
                    486:        if (!srv) {
1.7       misho     487:                rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
1.2       misho     488:                return -1;
                    489:        }
                    490:        if (srv->srv_blob.state) {
1.7       misho     491:                rpc_SetErr(EPERM, "Already started BLOB server!");
1.2       misho     492:                return 0;
                    493:        }
                    494: 
                    495:        memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
                    496:        if (access(diskDir, R_OK | W_OK) == -1) {
                    497:                LOGERR;
                    498:                return -1;
                    499:        } else
1.7       misho     500:                AIT_SET_STR(&srv->srv_blob.dir, diskDir);
1.2       misho     501: 
                    502:        srv->srv_blob.server.cli_tid = pthread_self();
                    503:        srv->srv_blob.server.cli_parent = srv;
1.4       misho     504: 
                    505:        memcpy(&sa, &srv->srv_server.cli_sa, sizeof sa);
1.6       misho     506:        switch (sa.sa.sa_family) {
1.4       misho     507:                case AF_INET:
1.6       misho     508:                        sa.sin.sin_port = htons(Port ? Port : ntohs(sa.sin.sin_port) + 1);
1.4       misho     509:                        break;
                    510:                case AF_INET6:
1.6       misho     511:                        sa.sin6.sin6_port = htons(Port ? Port : ntohs(sa.sin6.sin6_port) + 1);
1.4       misho     512:                        break;
                    513:                case AF_LOCAL:
1.6       misho     514:                        strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path);
1.4       misho     515:                        break;
                    516:                default:
1.7       misho     517:                        AIT_FREE_VAL(&srv->srv_blob.dir);
1.4       misho     518:                        return -1;
1.2       misho     519:        }
1.6       misho     520:        memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa);
1.2       misho     521: 
1.4       misho     522:        /* create BLOB server socket */
1.6       misho     523:        srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
1.2       misho     524:        if (srv->srv_blob.server.cli_sock == -1) {
                    525:                LOGERR;
1.7       misho     526:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho     527:                return -1;
                    528:        }
                    529:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
                    530:                LOGERR;
                    531:                close(srv->srv_blob.server.cli_sock);
1.7       misho     532:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho     533:                return -1;
                    534:        }
1.5       misho     535:        n = srv->srv_netbuf;
                    536:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
                    537:                LOGERR;
                    538:                close(srv->srv_blob.server.cli_sock);
1.7       misho     539:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.5       misho     540:                return -1;
                    541:        }
                    542:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
                    543:                LOGERR;
                    544:                close(srv->srv_blob.server.cli_sock);
1.7       misho     545:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.5       misho     546:                return -1;
                    547:        }
1.6       misho     548:        if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa, 
                    549:                                srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
1.2       misho     550:                LOGERR;
                    551:                close(srv->srv_blob.server.cli_sock);
1.7       misho     552:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho     553:                return -1;
                    554:        }
                    555: 
1.4       misho     556:        /* allocate pool for concurent clients */
1.2       misho     557:        srv->srv_blob.clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
                    558:        if (!srv->srv_blob.clients) {
                    559:                LOGERR;
                    560:                close(srv->srv_blob.server.cli_sock);
1.7       misho     561:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho     562:                return -1;
                    563:        } else
                    564:                memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
                    565: 
                    566:        rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
1.5       misho     567:        rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 1);
                    568:        rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 1);
                    569:        rpc_srv_registerCall(srv, NULL, CALL_BLOBSTATE, 0);
1.2       misho     570: 
1.4       misho     571:        srv->srv_blob.state = enable;   /* enable BLOB */
1.2       misho     572:        return 0;
                    573: }
                    574: 
                    575: /*
1.7       misho     576:  * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
                    577:  *
1.2       misho     578:  * @srv = RPC Server instance
                    579:  * return: none
                    580:  */
                    581: void
                    582: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
                    583: {
                    584:        rpc_cli_t *c;
                    585:        register int i;
                    586:        rpc_blob_t *f;
                    587: 
1.8       misho     588:        if (!srv || srv->srv_blob.state == disable)
1.2       misho     589:                return;
1.8       misho     590:        else
1.4       misho     591:                srv->srv_blob.state = kill;
1.2       misho     592: 
                    593:        rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);
                    594:        rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS);
                    595:        rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
                    596:        rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);
                    597: 
1.4       misho     598:        /* close all clients connections & server socket */
1.2       misho     599:        for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
1.6       misho     600:                if (c->cli_sa.sa.sa_family)
1.2       misho     601:                        shutdown(c->cli_sock, SHUT_RDWR);
                    602:        close(srv->srv_blob.server.cli_sock);
                    603: 
                    604:        if (srv->srv_blob.clients) {
                    605:                free(srv->srv_blob.clients);
                    606:                srv->srv_blob.clients = NULL;
                    607:        }
                    608: 
1.4       misho     609:        /* detach blobs */
1.2       misho     610:        while ((f = srv->srv_blob.blobs)) {
                    611:                srv->srv_blob.blobs = f->blob_next;
                    612:                rpc_srv_blobFree(srv, f);
                    613:                free(f);
                    614:        }
                    615: 
1.7       misho     616:        AIT_FREE_VAL(&srv->srv_blob.dir);
                    617: 
1.8       misho     618:        /* at final, save disable BLOB service state */
                    619:        srv->srv_blob.state = disable;
1.2       misho     620: }
                    621: 
                    622: /*
1.7       misho     623:  * rpc_srv_loopBLOB() - Execute Main BLOB server loop and wait for clients requests
                    624:  *
1.2       misho     625:  * @srv = RPC Server instance
                    626:  * return: -1 error or 0 ok, infinite loop ...
                    627:  */
                    628: int
1.5       misho     629: rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
1.2       misho     630: {
1.6       misho     631:        socklen_t salen = sizeof(io_sockaddr_t);
1.2       misho     632:        register int i;
                    633:        rpc_cli_t *c;
                    634:        int ret;
1.7       misho     635:        pthread_attr_t attr;
1.9       misho     636:        struct pollfd pfd;
1.2       misho     637: 
1.4       misho     638:        if (!srv || srv->srv_blob.state == kill) {
1.7       misho     639:                rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
1.2       misho     640:                return -1;
                    641:        }
                    642: 
                    643:        if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
                    644:                LOGERR;
                    645:                return -1;
1.9       misho     646:        } else
                    647:                fcntl(srv->srv_blob.server.cli_sock, F_SETFL, 
                    648:                                fcntl(srv->srv_blob.server.cli_sock, F_GETFL) | O_NONBLOCK);
1.2       misho     649: 
1.7       misho     650:        pthread_attr_init(&attr);
                    651:        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
                    652: 
1.9       misho     653:        pfd.fd = srv->srv_blob.server.cli_sock;
                    654:        pfd.events = POLLIN | POLLPRI;
1.7       misho     655:        /* main BLOB loop */
1.4       misho     656:        while (srv->srv_blob.state != kill && srv->srv_kill != kill) {
1.2       misho     657:                for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)
1.6       misho     658:                        if (!c->cli_sa.sa.sa_family)
1.2       misho     659:                                break;
                    660:                if (i >= srv->srv_numcli) {
1.5       misho     661: #ifdef HAVE_PTHREAD_YIELD
                    662:                        pthread_yield();
                    663: #endif
1.9.2.1   misho     664:                        usleep(1000000);
1.2       misho     665:                        continue;
                    666:                }
                    667: 
1.9       misho     668:                if ((ret = poll(&pfd, 1, srv->srv_session.sess_timeout * 1000)) == -1 || 
                    669:                                pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
1.2       misho     670:                        LOGERR;
                    671:                        ret = 1;
                    672:                        break;
1.9       misho     673:                } else if (!ret)
1.2       misho     674:                        continue;
                    675: 
1.6       misho     676:                c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa.sa, &salen);
1.2       misho     677:                if (c->cli_sock == -1) {
                    678:                        LOGERR;
                    679:                        continue;
                    680:                } else
                    681:                        c->cli_parent = srv;
                    682: 
1.7       misho     683:                /* spawn dispatch thread for BLOB client */
1.9       misho     684:                c->cli_kill = enable;
1.7       misho     685:                if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) {
1.2       misho     686:                        LOGERR;
                    687:                        continue;
1.7       misho     688:                }
1.2       misho     689:        }
                    690: 
1.5       misho     691:        srv->srv_blob.state = kill;
1.2       misho     692: 
1.7       misho     693:        pthread_attr_destroy(&attr);
1.2       misho     694:        return 0;
                    695: }
1.9.2.4   misho     696: #endif
1.2       misho     697: 
                    698: 
                    699: /*
1.7       misho     700:  * rpc_srv_initServer() - Init & create RPC Server
                    701:  *
1.1       misho     702:  * @regProgID = ProgramID for authentication & recognition
                    703:  * @regProcID = ProcessID for authentication & recognition
                    704:  * @concurentClients = Concurent clients at same time to this server
1.5       misho     705:  * @netBuf = Network buffer length, if =0 == BUFSIZ (also meaning max RPC packet)
1.4       misho     706:  * @csHost = Host name or address for bind server, if NULL any address
1.1       misho     707:  * @Port = Port for bind server, if Port == 0 default port is selected
                    708:  * return: NULL == error or !=NULL bind and created RPC server instance
                    709:  */
                    710: rpc_srv_t *
1.9.2.6 ! misho     711: rpc_srv_initServer(u_int regProgID, u_char regProcID, int concurentClients, 
1.9.2.4   misho     712:                int netBuf, const char *csHost, u_short Port)
1.1       misho     713: {
                    714:        int n = 1;
1.9.2.4   misho     715:        rpc_srv_t *srv = NULL;
1.6       misho     716:        io_sockaddr_t sa;
1.1       misho     717: 
1.9.2.2   misho     718:        if (!concurentClients || !regProgID) {
                    719:                rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1.1       misho     720:                return NULL;
                    721:        }
1.9.2.2   misho     722:        if (!io_gethostbyname(csHost, Port, &sa))
                    723:                return NULL;
1.1       misho     724:        if (!Port)
                    725:                Port = RPC_DEFPORT;
1.5       misho     726:        if (!netBuf)
                    727:                netBuf = BUFSIZ;
1.7       misho     728:        else
                    729:                netBuf = io_align(netBuf, 1);   /* align netBuf length */
1.1       misho     730: 
                    731:        srv = malloc(sizeof(rpc_srv_t));
                    732:        if (!srv) {
                    733:                LOGERR;
                    734:                return NULL;
                    735:        } else
                    736:                memset(srv, 0, sizeof(rpc_srv_t));
                    737: 
1.5       misho     738:        srv->srv_netbuf = netBuf;
1.1       misho     739:        srv->srv_session.sess_version = RPC_VERSION;
                    740:        srv->srv_session.sess_program = regProgID;
                    741:        srv->srv_session.sess_process = regProcID;
                    742: 
                    743:        srv->srv_server.cli_parent = srv;
1.9.2.4   misho     744:        memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
                    745: 
                    746:        /* init functions list */
                    747:        TAILQ_INIT(&srv->srv_funcs);
                    748: 
                    749:        /* init scheduler */
                    750:        srv->srv_root = schedBegin();
                    751:        if (!srv->srv_root) {
                    752:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                    753:                free(srv);
                    754:                return NULL;
                    755:        }
                    756: 
                    757:        /* init pool for clients */
                    758:        srv->srv_clients = io_arrayInit(concurentClients);
                    759:        if (!srv->srv_clients) {
                    760:                rpc_SetErr(io_GetErrno(), "%s", io_GetError());
                    761:                schedEnd(&srv->srv_root);
                    762:                free(srv);
                    763:                return NULL;
                    764:        }
1.4       misho     765: 
                    766:        /* create server socket */
1.9.2.2   misho     767:        srv->srv_server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
1.1       misho     768:        if (srv->srv_server.cli_sock == -1) {
                    769:                LOGERR;
1.9.2.4   misho     770:                io_arrayDestroy(&srv->srv_clients);
                    771:                schedEnd(&srv->srv_root);
1.1       misho     772:                free(srv);
                    773:                return NULL;
                    774:        }
                    775:        if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
                    776:                LOGERR;
1.9.2.4   misho     777:                goto err;
1.1       misho     778:        }
1.5       misho     779:        n = srv->srv_netbuf;
                    780:        if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
                    781:                LOGERR;
1.9.2.4   misho     782:                goto err;
1.5       misho     783:        }
                    784:        if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
                    785:                LOGERR;
1.9.2.4   misho     786:                goto err;
1.5       misho     787:        }
1.6       misho     788:        if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa, 
                    789:                                srv->srv_server.cli_sa.sa.sa_len) == -1) {
1.1       misho     790:                LOGERR;
1.9.2.4   misho     791:                goto err;
1.1       misho     792:        }
                    793: 
                    794:        return srv;
1.9.2.4   misho     795: err:   /* error condition */
                    796:        close(srv->srv_server.cli_sock);
                    797:        io_arrayDestroy(&srv->srv_clients);
                    798:        schedEnd(&srv->srv_root);
                    799:        free(srv);
                    800:        return NULL;
1.1       misho     801: }
                    802: 
                    803: /*
1.7       misho     804:  * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
                    805:  *
1.6       misho     806:  * @psrv = RPC Server instance
1.1       misho     807:  * return: none
                    808:  */
                    809: void
1.6       misho     810: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
1.1       misho     811: {
                    812:        rpc_cli_t *c;
                    813:        register int i;
1.9.2.4   misho     814:        rpc_func_t *f, *tmp;
1.1       misho     815: 
1.6       misho     816:        if (!psrv || !*psrv) {
1.9.2.3   misho     817:                rpc_SetErr(EINVAL, "Can`t destroy server because parameter is null!");
1.1       misho     818:                return;
                    819:        }
                    820: 
1.4       misho     821:        /* close all clients connections & server socket */
1.9.2.4   misho     822:        for (i = 0; i < io_arraySize((*psrv)->srv_clients); i++) {
                    823:                c = io_array((*psrv)->srv_clients, i, rpc_cli_t*);
                    824:                if (c) {
1.1       misho     825:                        shutdown(c->cli_sock, SHUT_RDWR);
1.2       misho     826:                        close(c->cli_sock);
1.9.2.6 ! misho     827: 
1.9.2.4   misho     828:                        free(c);
                    829:                        io_arrayDel((*psrv)->srv_clients, i, 0);
1.2       misho     830:                }
1.1       misho     831:        }
1.9.2.4   misho     832:        io_arrayDestroy(&(*psrv)->srv_clients);
                    833: 
                    834:        close((*psrv)->srv_server.cli_sock);
1.1       misho     835: 
1.4       misho     836:        /* detach exported calls */
1.9.2.4   misho     837:        TAILQ_FOREACH_SAFE(f, &(*psrv)->srv_funcs, func_node, tmp) {
1.6       misho     838:                io_freeVars(&f->func_vars);
1.7       misho     839:                AIT_FREE_VAL(&f->func_name);
1.2       misho     840:                free(f);
                    841:        }
                    842: 
1.9.2.4   misho     843:        schedEnd(&(*psrv)->srv_root);
1.6       misho     844:        free(*psrv);
                    845:        *psrv = NULL;
1.1       misho     846: }
                    847: 
                    848: /*
1.7       misho     849:  * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
                    850:  *
1.1       misho     851:  * @srv = RPC Server instance
                    852:  * return: -1 error or 0 ok, infinite loop ...
                    853:  */
                    854: int
1.5       misho     855: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
1.1       misho     856: {
                    857:        if (!srv) {
1.9.2.3   misho     858:                rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
1.1       misho     859:                return -1;
                    860:        }
                    861: 
1.9.2.6 ! misho     862:        fcntl(srv->srv_server.cli_sock, F_SETFL, 
        !           863:                        fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
        !           864: 
1.9.2.4   misho     865:        if (listen(srv->srv_server.cli_sock, io_arraySize(srv->srv_clients)) == -1) {
1.1       misho     866:                LOGERR;
                    867:                return -1;
1.9.2.6 ! misho     868:        }
1.7       misho     869: 
1.9.2.4   misho     870:        if (!schedRead(srv->srv_root, acceptClients, srv, srv->srv_server.cli_sock, NULL, 0)) {
                    871:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                    872:                return -1;
1.1       misho     873:        }
                    874: 
1.9.2.4   misho     875:        /* main rpc loop */
                    876:        schedRun(srv->srv_root, &srv->srv_kill);
1.1       misho     877:        return 0;
                    878: }
                    879: 
                    880: 
                    881: /*
                    882:  * rpc_srv_execCall() Execute registered call from RPC server
1.7       misho     883:  *
1.1       misho     884:  * @call = Register RPC call
                    885:  * @rpc = IN RPC call structure
1.5       misho     886:  * @args = IN RPC calling arguments from RPC client
1.1       misho     887:  * return: -1 error, !=-1 ok
                    888:  */
                    889: int
1.2       misho     890: rpc_srv_execCall(rpc_func_t * __restrict call, struct tagRPCCall * __restrict rpc, 
1.5       misho     891:                array_t * __restrict args)
1.1       misho     892: {
                    893:        rpc_callback_t func;
                    894: 
1.9.2.6 ! misho     895:        if (!call || !rpc || !call->func_parent || !AIT_ADDR(&call->func_name)) {
1.7       misho     896:                rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1.1       misho     897:                return -1;
                    898:        }
                    899: 
1.9.2.6 ! misho     900:        func = AIT_GET_LIKE(&call->func_name, rpc_callback_t);
        !           901:        return func(call, ntohs(rpc->call_argc), args);
1.1       misho     902: }

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