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

1.1       misho       1: /*************************************************************************
                      2: * (C) 2010 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
                      3: *  by Michael Pounov <misho@openbsd-bg.org>
                      4: *
                      5: * $Author: misho $
1.9.2.5 ! misho       6: * $Id: srv.c,v 1.9.2.4 2012/05/15 16:06:13 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: 
                     55:        schedCancelby(s->srv_root, taskMAX, CRITERIA_FD, (void*) c->cli_sock, NULL);
                     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) {
                     84:                f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash));
                     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: 
                    146:        if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash)))) {
                    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.4   misho     268:        /* accept client */
                    269:        c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa.sa, &salen);
                    270:        if (c->cli_sock == -1) {
                    271:                LOGERR;
                    272:                io_arrayDel(srv->srv_clients, i, 42);
                    273:                goto end;
                    274:        } else
                    275:                fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1.7       misho     276: 
1.9.2.4   misho     277:        /* alloc empty buffer */
                    278:        AIT_SET_BUF2(&c->cli_buf, 0, srv->srv_netbuf);
1.7       misho     279: 
1.9.2.4   misho     280:        schedRead(TASK_ROOT(task), rxPacket, c, c->cli_sock, NULL, 0);
                    281: end:
                    282:        schedReadSelf(task);
1.7       misho     283:        return NULL;
                    284: }
                    285: 
1.9.2.4   misho     286: #if 0
1.7       misho     287: static void *
                    288: txBLOB(sched_task_t *task)
                    289: {
                    290:        u_char *buf = TASK_DATA(task);
                    291:        struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
                    292:        int wlen = sizeof(struct tagBLOBHdr);
                    293: 
                    294:        /* calculate CRC */
                    295:        blob->hdr_crc ^= blob->hdr_crc;
1.8       misho     296:        blob->hdr_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
1.7       misho     297: 
                    298:        /* send reply */
                    299:        wlen = send(TASK_FD(task), buf, wlen, 0);
                    300:        if (wlen == -1)
                    301:                LOGERR;
                    302:        else if (wlen != sizeof(struct tagBLOBHdr))
                    303:                rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
                    304:                                "really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);
                    305: 
                    306:        return NULL;
                    307: }
1.4       misho     308: 
1.7       misho     309: static void *
                    310: rxBLOB(sched_task_t *task)
                    311: {
                    312:        rpc_cli_t *c = TASK_ARG(task);
                    313:        rpc_srv_t *s = c->cli_parent;
                    314:        rpc_blob_t *b;
                    315:        u_char *buf = TASK_DATA(task);
                    316:        struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
                    317:        int rlen;
                    318:        u_short crc;
                    319:        struct timespec ts;
                    320: 
                    321:        /* check for disable service at this moment? */
1.8       misho     322:        if (!s || s->srv_blob.state == disable) {
1.7       misho     323:                usleep(100000);
                    324: #ifdef HAVE_PTHREAD_YIELD
                    325:                pthread_yield();
                    326: #endif
                    327:                schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                    328:                                TASK_DATA(task), TASK_DATLEN(task));
                    329:                return NULL;
                    330:        }
1.5       misho     331: 
1.7       misho     332:        memset(buf, 0, TASK_DATLEN(task));
                    333:        rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
                    334:        if (rlen == -1) {
                    335:                LOGERR;
1.9       misho     336:                c->cli_kill = kill;
1.7       misho     337:                return NULL;
1.9       misho     338:        } else if (!rlen) { /* receive EOF */
                    339:                c->cli_kill = kill;
1.7       misho     340:                return NULL;
1.9       misho     341:        }
1.6       misho     342: 
1.7       misho     343:        if (rlen < sizeof(struct tagBLOBHdr)) {
                    344:                rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet");
                    345:                schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                    346:                                TASK_DATA(task), TASK_DATLEN(task));
                    347:                return NULL;
                    348:        }
1.1       misho     349: 
1.7       misho     350:        /* check integrity of packet */
                    351:        crc = ntohs(blob->hdr_crc);
                    352:        blob->hdr_crc ^= blob->hdr_crc;
1.8       misho     353:        if (crc != crcFletcher16((u_short*) buf, rlen / 2)) {
1.7       misho     354:                rpc_SetErr(ERPCMISMATCH, "Bad CRC BLOB packet");
                    355:                schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                    356:                                TASK_DATA(task), TASK_DATLEN(task));
                    357:                return NULL;
                    358:        }
1.5       misho     359: 
1.7       misho     360:        /* check RPC packet session info */
                    361:        if ((crc = rpc_chkPktSession(&blob->hdr_session, &s->srv_session))) {
                    362:                rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
                    363:                blob->hdr_cmd = error;
                    364:                goto end;
                    365:        } else {
                    366:                /* change socket timeout from last packet */
                    367:                ts.tv_sec = blob->hdr_session.sess_timeout;
                    368:                ts.tv_nsec = 0;
                    369:                schedPolling(TASK_ROOT(task), &ts, NULL);
                    370:        }
                    371: 
                    372:        /* Go to proceed packet ... */
                    373:        switch (blob->hdr_cmd) {
                    374:                case get:
                    375:                        if (!(b = rpc_srv_getBLOB(s, ntohl(blob->hdr_var)))) {
                    376:                                rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob->hdr_var));
                    377:                                blob->hdr_cmd = no;
                    378:                                blob->hdr_ret = RPC_ERROR(-1);
                    379:                                break;
                    380:                        } else
                    381:                                blob->hdr_len = htonl(b->blob_len);
1.5       misho     382: 
1.7       misho     383:                        if (rpc_srv_blobMap(s, b) != -1) {
                    384:                                /* deliver BLOB variable to client */
                    385:                                blob->hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
                    386:                                rpc_srv_blobUnmap(b);
                    387:                        } else {
                    388:                                blob->hdr_cmd = error;
                    389:                                blob->hdr_ret = RPC_ERROR(-1);
                    390:                        }
                    391:                        break;
                    392:                case set:
                    393:                        if ((b = rpc_srv_registerBLOB(s, ntohl(blob->hdr_len)))) {
                    394:                                /* set new BLOB variable for reply :) */
                    395:                                blob->hdr_var = htonl(b->blob_var);
                    396: 
                    397:                                /* receive BLOB from client */
                    398:                                blob->hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
                    399:                                rpc_srv_blobUnmap(b);
1.5       misho     400:                        } else {
1.7       misho     401:                                blob->hdr_cmd = error;
                    402:                                blob->hdr_ret = RPC_ERROR(-1);
                    403:                        }
                    404:                        break;
                    405:                case unset:
                    406:                        if (rpc_srv_unregisterBLOB(s, blob->hdr_var) == -1) {
                    407:                                blob->hdr_cmd = error;
                    408:                                blob->hdr_ret = RPC_ERROR(-1);
1.1       misho     409:                        }
                    410:                        break;
1.7       misho     411:                default:
                    412:                        rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob->hdr_cmd);
                    413:                        blob->hdr_cmd = error;
                    414:                        blob->hdr_ret = RPC_ERROR(-1);
                    415:        }
1.1       misho     416: 
1.7       misho     417: end:
                    418:        schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), 
                    419:                        TASK_DATA(task), TASK_DATLEN(task));
                    420:        schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task), 
                    421:                        TASK_DATA(task), TASK_DATLEN(task));
                    422:        return NULL;
1.2       misho     423: }
                    424: 
                    425: static void *
                    426: rpc_srv_dispatchVars(void *arg)
                    427: {
                    428:        rpc_cli_t *c = arg;
                    429:        rpc_srv_t *s;
1.7       misho     430:        sched_root_task_t *root;
                    431:        u_char *buf;
                    432:        struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
                    433: 
1.2       misho     434:        if (!arg) {
1.7       misho     435:                rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");
1.2       misho     436:                return NULL;
                    437:        } else
                    438:                s = c->cli_parent;
                    439: 
1.7       misho     440:        /* allocate net buffer */
                    441:        buf = malloc(sizeof(struct tagBLOBHdr));
                    442:        if (!buf) {
                    443:                LOGERR;
                    444:                return NULL;
                    445:        }
1.2       misho     446: 
1.7       misho     447:        root = schedBegin();
                    448:        if (!root) {
                    449:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                    450:                free(buf);
                    451:                return NULL;
                    452:        } else {
                    453:                schedTermCondition(root, kill);
                    454:                schedPolling(root, &ts, NULL);
                    455:        }
1.4       misho     456: 
1.7       misho     457:        schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));
1.2       misho     458: 
1.9       misho     459:        schedRun(root, (void*) &c->cli_kill);
1.7       misho     460:        schedEnd(&root);
1.2       misho     461: 
                    462:        shutdown(c->cli_sock, SHUT_RDWR);
                    463:        close(c->cli_sock);
                    464:        memset(c, 0, sizeof(rpc_cli_t));
1.7       misho     465:        free(buf);
                    466:        return NULL;
1.1       misho     467: }
                    468: 
                    469: // -------------------------------------------------
                    470: 
                    471: /*
1.7       misho     472:  * rpc_srv_initBLOBServer() - Init & create BLOB Server
                    473:  *
1.4       misho     474:  * @srv = RPC server instance
1.2       misho     475:  * @Port = Port for bind server, if Port == 0 default port is selected
                    476:  * @diskDir = Disk place for BLOB file objects
                    477:  * return: -1 == error or 0 bind and created BLOB server instance
                    478:  */
                    479: int
                    480: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
                    481: {
                    482:        int n = 1;
1.6       misho     483:        io_sockaddr_t sa;
1.2       misho     484: 
                    485:        if (!srv) {
1.7       misho     486:                rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
1.2       misho     487:                return -1;
                    488:        }
                    489:        if (srv->srv_blob.state) {
1.7       misho     490:                rpc_SetErr(EPERM, "Already started BLOB server!");
1.2       misho     491:                return 0;
                    492:        }
                    493: 
                    494:        memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
                    495:        if (access(diskDir, R_OK | W_OK) == -1) {
                    496:                LOGERR;
                    497:                return -1;
                    498:        } else
1.7       misho     499:                AIT_SET_STR(&srv->srv_blob.dir, diskDir);
1.2       misho     500: 
                    501:        srv->srv_blob.server.cli_tid = pthread_self();
                    502:        srv->srv_blob.server.cli_parent = srv;
1.4       misho     503: 
                    504:        memcpy(&sa, &srv->srv_server.cli_sa, sizeof sa);
1.6       misho     505:        switch (sa.sa.sa_family) {
1.4       misho     506:                case AF_INET:
1.6       misho     507:                        sa.sin.sin_port = htons(Port ? Port : ntohs(sa.sin.sin_port) + 1);
1.4       misho     508:                        break;
                    509:                case AF_INET6:
1.6       misho     510:                        sa.sin6.sin6_port = htons(Port ? Port : ntohs(sa.sin6.sin6_port) + 1);
1.4       misho     511:                        break;
                    512:                case AF_LOCAL:
1.6       misho     513:                        strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path);
1.4       misho     514:                        break;
                    515:                default:
1.7       misho     516:                        AIT_FREE_VAL(&srv->srv_blob.dir);
1.4       misho     517:                        return -1;
1.2       misho     518:        }
1.6       misho     519:        memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa);
1.2       misho     520: 
1.4       misho     521:        /* create BLOB server socket */
1.6       misho     522:        srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
1.2       misho     523:        if (srv->srv_blob.server.cli_sock == -1) {
                    524:                LOGERR;
1.7       misho     525:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho     526:                return -1;
                    527:        }
                    528:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
                    529:                LOGERR;
                    530:                close(srv->srv_blob.server.cli_sock);
1.7       misho     531:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho     532:                return -1;
                    533:        }
1.5       misho     534:        n = srv->srv_netbuf;
                    535:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
                    536:                LOGERR;
                    537:                close(srv->srv_blob.server.cli_sock);
1.7       misho     538:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.5       misho     539:                return -1;
                    540:        }
                    541:        if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
                    542:                LOGERR;
                    543:                close(srv->srv_blob.server.cli_sock);
1.7       misho     544:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.5       misho     545:                return -1;
                    546:        }
1.6       misho     547:        if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa, 
                    548:                                srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
1.2       misho     549:                LOGERR;
                    550:                close(srv->srv_blob.server.cli_sock);
1.7       misho     551:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho     552:                return -1;
                    553:        }
                    554: 
1.4       misho     555:        /* allocate pool for concurent clients */
1.2       misho     556:        srv->srv_blob.clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
                    557:        if (!srv->srv_blob.clients) {
                    558:                LOGERR;
                    559:                close(srv->srv_blob.server.cli_sock);
1.7       misho     560:                AIT_FREE_VAL(&srv->srv_blob.dir);
1.2       misho     561:                return -1;
                    562:        } else
                    563:                memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
                    564: 
                    565:        rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
1.5       misho     566:        rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 1);
                    567:        rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 1);
                    568:        rpc_srv_registerCall(srv, NULL, CALL_BLOBSTATE, 0);
1.2       misho     569: 
1.4       misho     570:        srv->srv_blob.state = enable;   /* enable BLOB */
1.2       misho     571:        return 0;
                    572: }
                    573: 
                    574: /*
1.7       misho     575:  * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
                    576:  *
1.2       misho     577:  * @srv = RPC Server instance
                    578:  * return: none
                    579:  */
                    580: void
                    581: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
                    582: {
                    583:        rpc_cli_t *c;
                    584:        register int i;
                    585:        rpc_blob_t *f;
                    586: 
1.8       misho     587:        if (!srv || srv->srv_blob.state == disable)
1.2       misho     588:                return;
1.8       misho     589:        else
1.4       misho     590:                srv->srv_blob.state = kill;
1.2       misho     591: 
                    592:        rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);
                    593:        rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS);
                    594:        rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
                    595:        rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);
                    596: 
1.4       misho     597:        /* close all clients connections & server socket */
1.2       misho     598:        for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
1.6       misho     599:                if (c->cli_sa.sa.sa_family)
1.2       misho     600:                        shutdown(c->cli_sock, SHUT_RDWR);
                    601:        close(srv->srv_blob.server.cli_sock);
                    602: 
                    603:        if (srv->srv_blob.clients) {
                    604:                free(srv->srv_blob.clients);
                    605:                srv->srv_blob.clients = NULL;
                    606:        }
                    607: 
1.4       misho     608:        /* detach blobs */
1.2       misho     609:        while ((f = srv->srv_blob.blobs)) {
                    610:                srv->srv_blob.blobs = f->blob_next;
                    611:                rpc_srv_blobFree(srv, f);
                    612:                free(f);
                    613:        }
                    614: 
1.7       misho     615:        AIT_FREE_VAL(&srv->srv_blob.dir);
                    616: 
1.8       misho     617:        /* at final, save disable BLOB service state */
                    618:        srv->srv_blob.state = disable;
1.2       misho     619: }
                    620: 
                    621: /*
1.7       misho     622:  * rpc_srv_loopBLOB() - Execute Main BLOB server loop and wait for clients requests
                    623:  *
1.2       misho     624:  * @srv = RPC Server instance
                    625:  * return: -1 error or 0 ok, infinite loop ...
                    626:  */
                    627: int
1.5       misho     628: rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
1.2       misho     629: {
1.6       misho     630:        socklen_t salen = sizeof(io_sockaddr_t);
1.2       misho     631:        register int i;
                    632:        rpc_cli_t *c;
                    633:        int ret;
1.7       misho     634:        pthread_attr_t attr;
1.9       misho     635:        struct pollfd pfd;
1.2       misho     636: 
1.4       misho     637:        if (!srv || srv->srv_blob.state == kill) {
1.7       misho     638:                rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
1.2       misho     639:                return -1;
                    640:        }
                    641: 
                    642:        if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
                    643:                LOGERR;
                    644:                return -1;
1.9       misho     645:        } else
                    646:                fcntl(srv->srv_blob.server.cli_sock, F_SETFL, 
                    647:                                fcntl(srv->srv_blob.server.cli_sock, F_GETFL) | O_NONBLOCK);
1.2       misho     648: 
1.7       misho     649:        pthread_attr_init(&attr);
                    650:        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
                    651: 
1.9       misho     652:        pfd.fd = srv->srv_blob.server.cli_sock;
                    653:        pfd.events = POLLIN | POLLPRI;
1.7       misho     654:        /* main BLOB loop */
1.4       misho     655:        while (srv->srv_blob.state != kill && srv->srv_kill != kill) {
1.2       misho     656:                for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)
1.6       misho     657:                        if (!c->cli_sa.sa.sa_family)
1.2       misho     658:                                break;
                    659:                if (i >= srv->srv_numcli) {
1.5       misho     660: #ifdef HAVE_PTHREAD_YIELD
                    661:                        pthread_yield();
                    662: #endif
1.9.2.1   misho     663:                        usleep(1000000);
1.2       misho     664:                        continue;
                    665:                }
                    666: 
1.9       misho     667:                if ((ret = poll(&pfd, 1, srv->srv_session.sess_timeout * 1000)) == -1 || 
                    668:                                pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
1.2       misho     669:                        LOGERR;
                    670:                        ret = 1;
                    671:                        break;
1.9       misho     672:                } else if (!ret)
1.2       misho     673:                        continue;
                    674: 
1.6       misho     675:                c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa.sa, &salen);
1.2       misho     676:                if (c->cli_sock == -1) {
                    677:                        LOGERR;
                    678:                        continue;
                    679:                } else
                    680:                        c->cli_parent = srv;
                    681: 
1.7       misho     682:                /* spawn dispatch thread for BLOB client */
1.9       misho     683:                c->cli_kill = enable;
1.7       misho     684:                if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) {
1.2       misho     685:                        LOGERR;
                    686:                        continue;
1.7       misho     687:                }
1.2       misho     688:        }
                    689: 
1.5       misho     690:        srv->srv_blob.state = kill;
1.2       misho     691: 
1.7       misho     692:        pthread_attr_destroy(&attr);
1.2       misho     693:        return 0;
                    694: }
1.9.2.4   misho     695: #endif
1.2       misho     696: 
                    697: 
                    698: /*
1.7       misho     699:  * rpc_srv_initServer() - Init & create RPC Server
                    700:  *
1.1       misho     701:  * @regProgID = ProgramID for authentication & recognition
                    702:  * @regProcID = ProcessID for authentication & recognition
                    703:  * @concurentClients = Concurent clients at same time to this server
1.5       misho     704:  * @netBuf = Network buffer length, if =0 == BUFSIZ (also meaning max RPC packet)
1.4       misho     705:  * @csHost = Host name or address for bind server, if NULL any address
1.1       misho     706:  * @Port = Port for bind server, if Port == 0 default port is selected
                    707:  * return: NULL == error or !=NULL bind and created RPC server instance
                    708:  */
                    709: rpc_srv_t *
                    710: rpc_srv_initServer(u_int regProgID, u_int regProcID, int concurentClients, 
1.9.2.4   misho     711:                int netBuf, const char *csHost, u_short Port)
1.1       misho     712: {
                    713:        int n = 1;
1.9.2.4   misho     714:        rpc_srv_t *srv = NULL;
1.6       misho     715:        io_sockaddr_t sa;
1.1       misho     716: 
1.9.2.2   misho     717:        if (!concurentClients || !regProgID) {
                    718:                rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1.1       misho     719:                return NULL;
                    720:        }
1.9.2.2   misho     721:        if (!io_gethostbyname(csHost, Port, &sa))
                    722:                return NULL;
1.1       misho     723:        if (!Port)
                    724:                Port = RPC_DEFPORT;
1.5       misho     725:        if (!netBuf)
                    726:                netBuf = BUFSIZ;
1.7       misho     727:        else
                    728:                netBuf = io_align(netBuf, 1);   /* align netBuf length */
1.1       misho     729: 
                    730:        srv = malloc(sizeof(rpc_srv_t));
                    731:        if (!srv) {
                    732:                LOGERR;
                    733:                return NULL;
                    734:        } else
                    735:                memset(srv, 0, sizeof(rpc_srv_t));
                    736: 
1.5       misho     737:        srv->srv_netbuf = netBuf;
1.1       misho     738:        srv->srv_session.sess_version = RPC_VERSION;
1.7       misho     739:        srv->srv_session.sess_timeout = DEF_RPC_TIMEOUT;
1.1       misho     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.4   misho     827:                        free(c);
                    828:                        io_arrayDel((*psrv)->srv_clients, i, 0);
1.2       misho     829:                }
1.1       misho     830:        }
1.9.2.4   misho     831:        io_arrayDestroy(&(*psrv)->srv_clients);
                    832: 
                    833:        close((*psrv)->srv_server.cli_sock);
1.1       misho     834: 
1.4       misho     835:        /* detach exported calls */
1.9.2.4   misho     836:        TAILQ_FOREACH_SAFE(f, &(*psrv)->srv_funcs, func_node, tmp) {
1.6       misho     837:                io_freeVars(&f->func_vars);
1.7       misho     838:                AIT_FREE_VAL(&f->func_name);
                    839:                AIT_FREE_VAL(&f->func_file);
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: {
1.9.2.4   misho     857:        /*
1.2       misho     858:        int ret;
1.9.2.4   misho     859:        */
1.1       misho     860: 
                    861:        if (!srv) {
1.9.2.3   misho     862:                rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
1.1       misho     863:                return -1;
                    864:        }
                    865: 
1.9.2.4   misho     866:        if (listen(srv->srv_server.cli_sock, io_arraySize(srv->srv_clients)) == -1) {
1.1       misho     867:                LOGERR;
                    868:                return -1;
1.9       misho     869:        } else
1.9.2.4   misho     870:                fcntl(srv->srv_server.cli_sock, F_SETFL, 
                    871:                                fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
1.7       misho     872: 
1.9.2.4   misho     873:        if (!schedRead(srv->srv_root, acceptClients, srv, srv->srv_server.cli_sock, NULL, 0)) {
                    874:                rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
                    875:                return -1;
1.1       misho     876:        }
                    877: 
1.9.2.4   misho     878:        /* main rpc loop */
                    879:        schedRun(srv->srv_root, &srv->srv_kill);
1.1       misho     880:        return 0;
                    881: }
                    882: 
                    883: 
                    884: /*
                    885:  * rpc_srv_execCall() Execute registered call from RPC server
1.7       misho     886:  *
1.1       misho     887:  * @call = Register RPC call
                    888:  * @rpc = IN RPC call structure
1.5       misho     889:  * @args = IN RPC calling arguments from RPC client
1.1       misho     890:  * return: -1 error, !=-1 ok
                    891:  */
                    892: int
1.2       misho     893: rpc_srv_execCall(rpc_func_t * __restrict call, struct tagRPCCall * __restrict rpc, 
1.5       misho     894:                array_t * __restrict args)
1.1       misho     895: {
                    896:        void *dl;
                    897:        rpc_callback_t func;
                    898:        int ret;
                    899: 
1.2       misho     900:        if (!call || !rpc || !call->func_parent) {
1.7       misho     901:                rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1.1       misho     902:                return -1;
                    903:        }
                    904: 
1.9.2.5 ! misho     905:        dl = dlopen(AIT_ADDR(&call->func_file), RTLD_LAZY | RTLD_GLOBAL);
1.1       misho     906:        if (!dl) {
1.7       misho     907:                rpc_SetErr(ENOENT, "Can`t attach module %s!", dlerror());
1.1       misho     908:                return -1;
                    909:        }
                    910: 
1.7       misho     911:        func = dlsym(dl, (const char*) AIT_GET_STR(&call->func_name));
1.1       misho     912:        if (func)
1.6       misho     913:                ret = func(call, ntohs(rpc->call_argc), args);
1.1       misho     914:        else {
1.7       misho     915:                rpc_SetErr(ENOEXEC, "Can`t find function %s!", dlerror());
1.1       misho     916:                ret = -1;
                    917:        }
                    918: 
                    919:        dlclose(dl);
                    920:        return ret;
                    921: }

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