1: /*************************************************************************
2: * (C) 2010 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
3: * by Michael Pounov <misho@openbsd-bg.org>
4: *
5: * $Author: misho $
6: * $Id: srv.c,v 1.7.2.2 2012/03/28 13:37:13 misho Exp $
7: *
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:
15: Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012
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: */
46: #include "global.h"
47:
48:
49: static void *rxPacket(sched_task_t*);
50: static void *rxBLOB(sched_task_t*);
51:
52: static void *
53: txPacket(sched_task_t *task)
54: {
55: rpc_cli_t *c = TASK_ARG(task);
56: rpc_srv_t *s = c->cli_parent;
57: rpc_func_t *f = NULL;
58: u_char *buf = TASK_DATA(task);
59: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
60: int ret, wlen = sizeof(struct tagRPCCall);
61: array_t *arr = NULL;
62:
63: ioTRACE(RPC_TRACE_LEVEL);
64:
65: if (rpc->call_argc) {
66: f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash));
67: if (!f) {
68: rpc->call_argc ^= rpc->call_argc;
69: rpc->call_rep.ret = RPC_ERROR(-1);
70: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
71: } else {
72: rpc->call_argc = htons(rpc_srv_getVars(f, &arr));
73: /* Go Encapsulate variables */
74: ret = io_vars2buffer(buf + wlen, TASK_DATLEN(task) - wlen, arr);
75: io_clrVars(f->func_vars);
76: if (ret == -1) {
77: rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
78: rpc->call_argc ^= rpc->call_argc;
79: rpc->call_rep.ret = RPC_ERROR(-1);
80: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
81: } else
82: wlen += ret;
83: }
84: }
85:
86: /* calculate CRC */
87: rpc->call_crc ^= rpc->call_crc;
88: rpc->call_crc = htons(crcFletcher16((u_short*) buf, io_align(wlen, 1) / 2));
89:
90: /* send reply */
91: ret = send(TASK_FD(task), buf, wlen, 0);
92: if (ret == -1)
93: LOGERR;
94: else if (ret != wlen)
95: rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
96: "really sended %d bytes", wlen, ret);
97: else
98: ioDEBUG(RPC_DEBUG_LEVEL, "Sended %d bytes", ret);
99:
100: return NULL;
101: }
102:
103: static void *
104: execCall(sched_task_t *task)
105: {
106: rpc_cli_t *c = TASK_ARG(task);
107: rpc_srv_t *s = c->cli_parent;
108: rpc_func_t *f = NULL;
109: array_t *arr = NULL;
110: u_char *buf = TASK_DATA(task);
111: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
112: int argc = ntohs(rpc->call_argc);
113:
114: ioTRACE(RPC_TRACE_LEVEL);
115:
116: /* Go decapsulate variables ... */
117: if (argc) {
118: arr = io_buffer2vars(buf + sizeof(struct tagRPCCall),
119: TASK_DATLEN(task) - sizeof(struct tagRPCCall), argc, 1);
120: if (!arr) {
121: rpc_SetErr(ERPCMISMATCH, "#%d - %s", io_GetErrno(), io_GetError());
122: rpc->call_argc ^= rpc->call_argc;
123: rpc->call_rep.ret = RPC_ERROR(-1);
124: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
125: return NULL;
126: }
127: }
128:
129: if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash)))) {
130: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
131: rpc->call_argc ^= rpc->call_argc;
132: rpc->call_rep.ret = RPC_ERROR(-1);
133: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
134: } else {
135: ioDEBUG(RPC_DEBUG_LEVEL, "RPC function %s from module %s",
136: AIT_GET_STR(&f->func_name), AIT_GET_LIKE(&f->func_file, char*));
137:
138: /* if client doesn't want reply */
139: argc = rpc->call_req.flags & RPC_NOREPLY;
140: rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));
141: if (rpc->call_rep.ret == htonl(-1)) {
142: rpc->call_rep.eno = RPC_ERROR(errno);
143: rpc->call_argc ^= rpc->call_argc;
144: } else {
145: rpc->call_rep.eno ^= rpc->call_rep.eno;
146: if (argc) {
147: /* without reply */
148: io_clrVars(f->func_vars);
149: rpc->call_argc ^= rpc->call_argc;
150: } else
151: rpc->call_argc = htons(rpc_srv_getVars(f, NULL));
152: }
153: }
154:
155: if (arr)
156: io_arrayDestroy(&arr);
157: return NULL;
158: }
159:
160: static void *
161: rxPacket(sched_task_t *task)
162: {
163: rpc_cli_t *c = TASK_ARG(task);
164: rpc_srv_t *s = c->cli_parent;
165: u_char *buf = TASK_DATA(task);
166: int rlen;
167: u_short crc;
168: struct tagRPCCall *rpc;
169: struct timespec ts;
170:
171: ioTRACE(RPC_TRACE_LEVEL);
172:
173: memset(buf, 0, TASK_DATLEN(task));
174: rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
175: if (rlen == -1) {
176: LOGERR;
177: s->srv_kill = s->srv_blob.state = kill;
178: return NULL;
179: } else if (!rlen) { /* receive EOF */
180: s->srv_kill = s->srv_blob.state = kill;
181: return NULL;
182: } else
183: ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen);
184:
185: if (rlen < sizeof(struct tagRPCCall)) {
186: rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");
187:
188: schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),
189: TASK_DATA(task), TASK_DATLEN(task));
190: return NULL;
191: } else
192: rpc = (struct tagRPCCall*) buf;
193:
194: /* check integrity of packet */
195: crc = ntohs(rpc->call_crc);
196: rpc->call_crc ^= rpc->call_crc;
197: if (crc != crcFletcher16((u_short*) buf, io_align(rlen, 1) / 2)) {
198: rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
199:
200: schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),
201: TASK_DATA(task), TASK_DATLEN(task));
202: return NULL;
203: }
204:
205: /* check RPC packet session info */
206: if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
207: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
208: rpc->call_argc ^= rpc->call_argc;
209: rpc->call_rep.ret = RPC_ERROR(-1);
210: rpc->call_rep.eno = RPC_ERROR(errno);
211: goto end;
212: } else {
213: /* change socket timeout from last packet */
214: ts.tv_sec = rpc->call_session.sess_timeout;
215: ts.tv_nsec = 0;
216: schedPolling(TASK_ROOT(task), &ts, NULL);
217: }
218:
219: /* execute RPC call */
220: schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), 0,
221: TASK_DATA(task), TASK_DATLEN(task));
222:
223: end:
224: /* send RPC reply */
225: if (!(rpc->call_req.flags & RPC_NOREPLY))
226: schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task),
227: TASK_DATA(task), TASK_DATLEN(task));
228: /* lets get next packet */
229: schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),
230: TASK_DATA(task), TASK_DATLEN(task));
231: return NULL;
232: }
233:
234: static void *
235: rpc_srv_dispatchCall(void *arg)
236: {
237: rpc_cli_t *c = arg;
238: rpc_srv_t *s;
239: u_char *buf;
240: sched_root_task_t *root;
241: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
242:
243: ioTRACE(RPC_TRACE_LEVEL);
244:
245: if (!arg) {
246: rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client");
247: return NULL;
248: } else
249: s = c->cli_parent;
250:
251: /* allocate net buffer */
252: buf = malloc(s->srv_netbuf);
253: if (!buf) {
254: LOGERR;
255: return NULL;
256: }
257:
258: root = schedBegin();
259: if (!root) {
260: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
261: free(buf);
262: return NULL;
263: } else {
264: schedTermCondition(root, kill);
265: schedPolling(root, &ts, NULL);
266: }
267:
268: schedRead(root, rxPacket, c, c->cli_sock, buf, s->srv_netbuf);
269:
270: schedRun(root, (void*) &s->srv_kill);
271: schedEnd(&root);
272:
273: shutdown(c->cli_sock, SHUT_RDWR);
274: close(c->cli_sock);
275: memset(c, 0, sizeof(rpc_cli_t));
276: free(buf);
277: return NULL;
278: }
279:
280:
281: static void *
282: txBLOB(sched_task_t *task)
283: {
284: u_char *buf = TASK_DATA(task);
285: struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
286: int wlen = sizeof(struct tagBLOBHdr);
287:
288: ioTRACE(RPC_TRACE_LEVEL);
289:
290: /* calculate CRC */
291: blob->hdr_crc ^= blob->hdr_crc;
292: blob->hdr_crc = htons(crcFletcher16((u_short*) buf, io_align(wlen, 1) / 2));
293:
294: /* send reply */
295: wlen = send(TASK_FD(task), buf, wlen, 0);
296: if (wlen == -1)
297: LOGERR;
298: else if (wlen != sizeof(struct tagBLOBHdr))
299: rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
300: "really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);
301: else
302: ioDEBUG(RPC_DEBUG_LEVEL, "Sended %d bytes", wlen);
303:
304: return NULL;
305: }
306:
307: static void *
308: rxBLOB(sched_task_t *task)
309: {
310: rpc_cli_t *c = TASK_ARG(task);
311: rpc_srv_t *s = c->cli_parent;
312: rpc_blob_t *b;
313: u_char *buf = TASK_DATA(task);
314: struct tagBLOBHdr *blob = (struct tagBLOBHdr *) buf;
315: int rlen;
316: u_short crc;
317: struct timespec ts;
318:
319: ioTRACE(RPC_TRACE_LEVEL);
320:
321: /* check for disable service at this moment? */
322: if (!s || s->srv_blob.state == disable) {
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: }
331:
332: memset(buf, 0, TASK_DATLEN(task));
333: rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
334: if (rlen == -1) {
335: LOGERR;
336: s->srv_blob.state = kill;
337: return NULL;
338: } else if (!rlen || s->srv_kill == kill) { /* receive EOF */
339: s->srv_blob.state = kill;
340: return NULL;
341: } else
342: ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen);
343:
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: }
350:
351: /* check integrity of packet */
352: crc = ntohs(blob->hdr_crc);
353: blob->hdr_crc ^= blob->hdr_crc;
354: if (crc != crcFletcher16((u_short*) buf, io_align(rlen, 1) / 2)) {
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: }
360:
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);
383:
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);
401: } else {
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);
410: }
411: break;
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: }
417:
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;
424: }
425:
426: static void *
427: rpc_srv_dispatchVars(void *arg)
428: {
429: rpc_cli_t *c = arg;
430: rpc_srv_t *s;
431: sched_root_task_t *root;
432: u_char *buf;
433: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
434:
435: ioTRACE(RPC_TRACE_LEVEL);
436:
437: if (!arg) {
438: rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");
439: return NULL;
440: } else
441: s = c->cli_parent;
442:
443: /* allocate net buffer */
444: buf = malloc(sizeof(struct tagBLOBHdr));
445: if (!buf) {
446: LOGERR;
447: return NULL;
448: }
449:
450: root = schedBegin();
451: if (!root) {
452: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
453: free(buf);
454: return NULL;
455: } else {
456: schedTermCondition(root, kill);
457: schedPolling(root, &ts, NULL);
458: }
459:
460: schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));
461:
462: schedRun(root, (void*) &s->srv_blob.state);
463: schedEnd(&root);
464:
465: shutdown(c->cli_sock, SHUT_RDWR);
466: close(c->cli_sock);
467: memset(c, 0, sizeof(rpc_cli_t));
468: free(buf);
469: return NULL;
470: }
471:
472: // -------------------------------------------------
473:
474: /*
475: * rpc_srv_initBLOBServer() - Init & create BLOB Server
476: *
477: * @srv = RPC server instance
478: * @Port = Port for bind server, if Port == 0 default port is selected
479: * @diskDir = Disk place for BLOB file objects
480: * return: -1 == error or 0 bind and created BLOB server instance
481: */
482: int
483: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
484: {
485: int n = 1;
486: io_sockaddr_t sa;
487:
488: ioTRACE(RPC_TRACE_LEVEL);
489:
490: if (!srv) {
491: rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
492: return -1;
493: }
494: if (srv->srv_blob.state) {
495: rpc_SetErr(EPERM, "Already started BLOB server!");
496: return 0;
497: }
498:
499: memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
500: if (access(diskDir, R_OK | W_OK) == -1) {
501: LOGERR;
502: return -1;
503: } else
504: AIT_SET_STR(&srv->srv_blob.dir, diskDir);
505:
506: srv->srv_blob.server.cli_tid = pthread_self();
507: srv->srv_blob.server.cli_parent = srv;
508:
509: memcpy(&sa, &srv->srv_server.cli_sa, sizeof sa);
510: switch (sa.sa.sa_family) {
511: case AF_INET:
512: sa.sin.sin_port = htons(Port ? Port : ntohs(sa.sin.sin_port) + 1);
513: break;
514: case AF_INET6:
515: sa.sin6.sin6_port = htons(Port ? Port : ntohs(sa.sin6.sin6_port) + 1);
516: break;
517: case AF_LOCAL:
518: strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path);
519: break;
520: default:
521: AIT_FREE_VAL(&srv->srv_blob.dir);
522: return -1;
523: }
524: memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa);
525:
526: /* create BLOB server socket */
527: srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
528: if (srv->srv_blob.server.cli_sock == -1) {
529: LOGERR;
530: AIT_FREE_VAL(&srv->srv_blob.dir);
531: return -1;
532: }
533: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
534: LOGERR;
535: close(srv->srv_blob.server.cli_sock);
536: AIT_FREE_VAL(&srv->srv_blob.dir);
537: return -1;
538: }
539: n = srv->srv_netbuf;
540: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
541: LOGERR;
542: close(srv->srv_blob.server.cli_sock);
543: AIT_FREE_VAL(&srv->srv_blob.dir);
544: return -1;
545: }
546: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
547: LOGERR;
548: close(srv->srv_blob.server.cli_sock);
549: AIT_FREE_VAL(&srv->srv_blob.dir);
550: return -1;
551: }
552: if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa,
553: srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
554: LOGERR;
555: close(srv->srv_blob.server.cli_sock);
556: AIT_FREE_VAL(&srv->srv_blob.dir);
557: return -1;
558: }
559:
560: /* allocate pool for concurent clients */
561: srv->srv_blob.clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
562: if (!srv->srv_blob.clients) {
563: LOGERR;
564: close(srv->srv_blob.server.cli_sock);
565: AIT_FREE_VAL(&srv->srv_blob.dir);
566: return -1;
567: } else
568: memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
569:
570: pthread_mutex_init(&srv->srv_blob.mtx, NULL);
571:
572: rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
573: rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 1);
574: rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 1);
575: rpc_srv_registerCall(srv, NULL, CALL_BLOBSTATE, 0);
576:
577: srv->srv_blob.state = enable; /* enable BLOB */
578: return 0;
579: }
580:
581: /*
582: * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
583: *
584: * @srv = RPC Server instance
585: * return: none
586: */
587: void
588: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
589: {
590: rpc_cli_t *c;
591: register int i;
592: rpc_blob_t *f;
593:
594: ioTRACE(RPC_TRACE_LEVEL);
595:
596: if (!srv || srv->srv_blob.state == disable)
597: return;
598: else
599: srv->srv_blob.state = kill;
600:
601: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);
602: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS);
603: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
604: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);
605:
606: /* close all clients connections & server socket */
607: for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
608: if (c->cli_sa.sa.sa_family)
609: shutdown(c->cli_sock, SHUT_RDWR);
610: close(srv->srv_blob.server.cli_sock);
611:
612: pthread_mutex_lock(&srv->srv_blob.mtx);
613: if (srv->srv_blob.clients) {
614: free(srv->srv_blob.clients);
615: srv->srv_blob.clients = NULL;
616: }
617:
618: /* detach blobs */
619: while ((f = srv->srv_blob.blobs)) {
620: srv->srv_blob.blobs = f->blob_next;
621: rpc_srv_blobFree(srv, f);
622: free(f);
623: }
624: pthread_mutex_unlock(&srv->srv_blob.mtx);
625:
626: AIT_FREE_VAL(&srv->srv_blob.dir);
627:
628: while (pthread_mutex_trylock(&srv->srv_blob.mtx) == EBUSY);
629: pthread_mutex_destroy(&srv->srv_blob.mtx);
630:
631: /* at final, save disable BLOB service state */
632: srv->srv_blob.state = disable;
633: }
634:
635: /*
636: * rpc_srv_loopBLOB() - Execute Main BLOB server loop and wait for clients requests
637: *
638: * @srv = RPC Server instance
639: * return: -1 error or 0 ok, infinite loop ...
640: */
641: int
642: rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
643: {
644: socklen_t salen = sizeof(io_sockaddr_t);
645: register int i;
646: rpc_cli_t *c;
647: fd_set fds;
648: int ret;
649: struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
650: pthread_attr_t attr;
651:
652: ioTRACE(RPC_TRACE_LEVEL);
653:
654: if (!srv || srv->srv_blob.state == kill) {
655: rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
656: return -1;
657: }
658:
659: tv.tv_sec = srv->srv_session.sess_timeout;
660:
661: if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
662: LOGERR;
663: return -1;
664: }
665:
666: pthread_attr_init(&attr);
667: pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
668:
669: /* main BLOB loop */
670: while (srv->srv_blob.state != kill && srv->srv_kill != kill) {
671: for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)
672: if (!c->cli_sa.sa.sa_family)
673: break;
674: if (i >= srv->srv_numcli) {
675: #ifdef HAVE_PTHREAD_YIELD
676: pthread_yield();
677: #else
678: usleep(1000000);
679: #endif
680: continue;
681: }
682:
683: FD_ZERO(&fds);
684: FD_SET(srv->srv_blob.server.cli_sock, &fds);
685: ret = select(srv->srv_blob.server.cli_sock + 1, &fds, NULL, NULL, &tv);
686: if (ret == -1) {
687: LOGERR;
688: ret = 1;
689: break;
690: }
691: if (!ret)
692: continue;
693:
694: c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa.sa, &salen);
695: if (c->cli_sock == -1) {
696: LOGERR;
697: continue;
698: } else
699: c->cli_parent = srv;
700:
701: /* spawn dispatch thread for BLOB client */
702: if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) {
703: LOGERR;
704: continue;
705: }
706: }
707:
708: srv->srv_blob.state = kill;
709:
710: pthread_attr_destroy(&attr);
711: return 0;
712: }
713:
714:
715: /*
716: * rpc_srv_initServer() - Init & create RPC Server
717: *
718: * @regProgID = ProgramID for authentication & recognition
719: * @regProcID = ProcessID for authentication & recognition
720: * @concurentClients = Concurent clients at same time to this server
721: * @netBuf = Network buffer length, if =0 == BUFSIZ (also meaning max RPC packet)
722: * @family = Family type, AF_INET, AF_INET6 or AF_LOCAL
723: * @csHost = Host name or address for bind server, if NULL any address
724: * @Port = Port for bind server, if Port == 0 default port is selected
725: * return: NULL == error or !=NULL bind and created RPC server instance
726: */
727: rpc_srv_t *
728: rpc_srv_initServer(u_int regProgID, u_int regProcID, int concurentClients,
729: int netBuf, u_short family, const char *csHost, u_short Port)
730: {
731: rpc_srv_t *srv = NULL;
732: int n = 1;
733: struct hostent *host = NULL;
734: io_sockaddr_t sa;
735:
736: ioTRACE(RPC_TRACE_LEVEL);
737:
738: if (!concurentClients || !regProgID ||
739: (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) {
740: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");
741: return NULL;
742: }
743: if (!Port)
744: Port = RPC_DEFPORT;
745: if (!netBuf)
746: netBuf = BUFSIZ;
747: else
748: netBuf = io_align(netBuf, 1); /* align netBuf length */
749: if (csHost && family != AF_LOCAL) {
750: host = gethostbyname2(csHost, family);
751: if (!host) {
752: rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno));
753: return NULL;
754: }
755: }
756: memset(&sa, 0, sizeof sa);
757: sa.sa.sa_family = family;
758: switch (family) {
759: case AF_INET:
760: sa.sin.sin_len = sizeof(struct sockaddr_in);
761: sa.sin.sin_port = htons(Port);
762: if (csHost)
763: memcpy(&sa.sin.sin_addr, host->h_addr, host->h_length);
764: break;
765: case AF_INET6:
766: sa.sin6.sin6_len = sizeof(struct sockaddr_in6);
767: sa.sin6.sin6_port = htons(Port);
768: if (csHost)
769: memcpy(&sa.sin6.sin6_addr, host->h_addr, host->h_length);
770: break;
771: case AF_LOCAL:
772: sa.sun.sun_len = sizeof(struct sockaddr_un);
773: if (csHost)
774: strlcpy(sa.sun.sun_path, csHost, sizeof sa.sun.sun_path);
775: unlink(sa.sun.sun_path);
776: break;
777: default:
778: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t start RPC server ...\n");
779: return NULL;
780: }
781:
782: srv = malloc(sizeof(rpc_srv_t));
783: if (!srv) {
784: LOGERR;
785: return NULL;
786: } else
787: memset(srv, 0, sizeof(rpc_srv_t));
788:
789: srv->srv_netbuf = netBuf;
790: srv->srv_numcli = concurentClients;
791: srv->srv_session.sess_version = RPC_VERSION;
792: srv->srv_session.sess_timeout = DEF_RPC_TIMEOUT;
793: srv->srv_session.sess_program = regProgID;
794: srv->srv_session.sess_process = regProcID;
795:
796: srv->srv_server.cli_tid = pthread_self();
797: srv->srv_server.cli_parent = srv;
798: memcpy(&srv->srv_server.cli_sa, &sa, sizeof sa);
799:
800: /* create server socket */
801: srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);
802: if (srv->srv_server.cli_sock == -1) {
803: LOGERR;
804: free(srv);
805: return NULL;
806: }
807: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
808: LOGERR;
809: close(srv->srv_server.cli_sock);
810: free(srv);
811: return NULL;
812: }
813: n = srv->srv_netbuf;
814: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
815: LOGERR;
816: close(srv->srv_server.cli_sock);
817: free(srv);
818: return NULL;
819: }
820: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
821: LOGERR;
822: close(srv->srv_server.cli_sock);
823: free(srv);
824: return NULL;
825: }
826: if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa,
827: srv->srv_server.cli_sa.sa.sa_len) == -1) {
828: LOGERR;
829: close(srv->srv_server.cli_sock);
830: free(srv);
831: return NULL;
832: }
833:
834: /* allocate pool for concurent clients */
835: srv->srv_clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
836: if (!srv->srv_clients) {
837: LOGERR;
838: close(srv->srv_server.cli_sock);
839: free(srv);
840: return NULL;
841: } else
842: memset(srv->srv_clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
843:
844: pthread_mutex_init(&srv->srv_mtx, NULL);
845:
846: rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0);
847: rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 1);
848: rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 4);
849: rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 1);
850:
851: return srv;
852: }
853:
854: /*
855: * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
856: *
857: * @psrv = RPC Server instance
858: * return: none
859: */
860: void
861: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
862: {
863: rpc_cli_t *c;
864: register int i;
865: rpc_func_t *f;
866:
867: ioTRACE(RPC_TRACE_LEVEL);
868:
869: if (!psrv || !*psrv) {
870: rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
871: return;
872: }
873:
874: rpc_srv_endBLOBServer(*psrv);
875:
876: /* close all clients connections & server socket */
877: for (i = 0, c = (*psrv)->srv_clients; i < (*psrv)->srv_numcli && c; i++, c++)
878: if (c->cli_sa.sa.sa_family) {
879: shutdown(c->cli_sock, SHUT_RDWR);
880: close(c->cli_sock);
881: }
882: close((*psrv)->srv_server.cli_sock);
883:
884: if ((*psrv)->srv_clients) {
885: free((*psrv)->srv_clients);
886: (*psrv)->srv_clients = NULL;
887: (*psrv)->srv_numcli = 0;
888: }
889:
890: /* detach exported calls */
891: pthread_mutex_lock(&(*psrv)->srv_mtx);
892: while ((f = (*psrv)->srv_funcs)) {
893: (*psrv)->srv_funcs = f->func_next;
894: io_freeVars(&f->func_vars);
895: AIT_FREE_VAL(&f->func_name);
896: AIT_FREE_VAL(&f->func_file);
897: free(f);
898: }
899: pthread_mutex_unlock(&(*psrv)->srv_mtx);
900:
901: while (pthread_mutex_trylock(&(*psrv)->srv_mtx) == EBUSY);
902: pthread_mutex_destroy(&(*psrv)->srv_mtx);
903:
904: free(*psrv);
905: *psrv = NULL;
906: }
907:
908: /*
909: * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
910: *
911: * @srv = RPC Server instance
912: * return: -1 error or 0 ok, infinite loop ...
913: */
914: int
915: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
916: {
917: socklen_t salen = sizeof(io_sockaddr_t);
918: register int i;
919: rpc_cli_t *c;
920: fd_set fds;
921: int ret;
922: struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
923: pthread_attr_t attr;
924:
925: ioTRACE(RPC_TRACE_LEVEL);
926:
927: if (!srv) {
928: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n");
929: return -1;
930: }
931:
932: tv.tv_sec = srv->srv_session.sess_timeout;
933:
934: /* activate BLOB server worker if srv->srv_blob.state == enable */
935: rpc_srv_execBLOBServer(srv);
936:
937: if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) {
938: LOGERR;
939: return -1;
940: }
941:
942: pthread_attr_init(&attr);
943: pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
944:
945: /* main rpc loop */
946: while (srv->srv_kill != kill) {
947: for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)
948: if (!c->cli_sa.sa.sa_family)
949: break;
950: if (i >= srv->srv_numcli) {
951: #ifdef HAVE_PTHREAD_YIELD
952: pthread_yield();
953: #else
954: usleep(1000000);
955: #endif
956: continue;
957: }
958:
959: FD_ZERO(&fds);
960: FD_SET(srv->srv_server.cli_sock, &fds);
961: ret = select(srv->srv_server.cli_sock + 1, &fds, NULL, NULL, &tv);
962: if (ret == -1) {
963: LOGERR;
964: ret = 1;
965: break;
966: }
967: if (!ret)
968: continue;
969:
970: c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa.sa, &salen);
971: if (c->cli_sock == -1) {
972: LOGERR;
973: continue;
974: } else
975: c->cli_parent = srv;
976:
977: /* spawn rpc client dispatcher */
978: if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) {
979: LOGERR;
980: continue;
981: }
982: }
983:
984: pthread_attr_destroy(&attr);
985: return 0;
986: }
987:
988: // ---------------------------------------------------------
989:
990: /*
991: * rpc_srv_execCall() Execute registered call from RPC server
992: *
993: * @call = Register RPC call
994: * @rpc = IN RPC call structure
995: * @args = IN RPC calling arguments from RPC client
996: * return: -1 error, !=-1 ok
997: */
998: int
999: rpc_srv_execCall(rpc_func_t * __restrict call, struct tagRPCCall * __restrict rpc,
1000: array_t * __restrict args)
1001: {
1002: void *dl;
1003: rpc_callback_t func;
1004: int ret;
1005:
1006: ioTRACE(RPC_TRACE_LEVEL);
1007:
1008: if (!call || !rpc || !call->func_parent) {
1009: rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1010: return -1;
1011: }
1012:
1013: dl = dlopen(AIT_VOID(&call->func_file), RTLD_NOW);
1014: if (!dl) {
1015: rpc_SetErr(ENOENT, "Can`t attach module %s!", dlerror());
1016: return -1;
1017: }
1018:
1019: func = dlsym(dl, (const char*) AIT_GET_STR(&call->func_name));
1020: if (func)
1021: ret = func(call, ntohs(rpc->call_argc), args);
1022: else {
1023: rpc_SetErr(ENOEXEC, "Can`t find function %s!", dlerror());
1024: ret = -1;
1025: }
1026:
1027: dlclose(dl);
1028: return ret;
1029: }
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>
![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to srv.c CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / libaitrpc / src