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.1 2012/03/28 01:17:09 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) {
597: rpc_SetErr(EINVAL, "Can`t destroy server because parameter is null!");
598: return;
599: } else
600: srv->srv_blob.state = kill;
601:
602: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);
603: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS);
604: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
605: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);
606:
607: /* close all clients connections & server socket */
608: for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
609: if (c->cli_sa.sa.sa_family)
610: shutdown(c->cli_sock, SHUT_RDWR);
611: close(srv->srv_blob.server.cli_sock);
612:
613: pthread_mutex_lock(&srv->srv_blob.mtx);
614: if (srv->srv_blob.clients) {
615: free(srv->srv_blob.clients);
616: srv->srv_blob.clients = NULL;
617: }
618:
619: /* detach blobs */
620: while ((f = srv->srv_blob.blobs)) {
621: srv->srv_blob.blobs = f->blob_next;
622: rpc_srv_blobFree(srv, f);
623: free(f);
624: }
625: pthread_mutex_unlock(&srv->srv_blob.mtx);
626:
627: AIT_FREE_VAL(&srv->srv_blob.dir);
628:
629: while (pthread_mutex_trylock(&srv->srv_blob.mtx) == EBUSY);
630: pthread_mutex_destroy(&srv->srv_blob.mtx);
631: }
632:
633: /*
634: * rpc_srv_loopBLOB() - Execute Main BLOB server loop and wait for clients requests
635: *
636: * @srv = RPC Server instance
637: * return: -1 error or 0 ok, infinite loop ...
638: */
639: int
640: rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
641: {
642: socklen_t salen = sizeof(io_sockaddr_t);
643: register int i;
644: rpc_cli_t *c;
645: fd_set fds;
646: int ret;
647: struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
648: pthread_attr_t attr;
649:
650: ioTRACE(RPC_TRACE_LEVEL);
651:
652: if (!srv || srv->srv_blob.state == kill) {
653: rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
654: return -1;
655: }
656:
657: tv.tv_sec = srv->srv_session.sess_timeout;
658:
659: if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
660: LOGERR;
661: return -1;
662: }
663:
664: pthread_attr_init(&attr);
665: pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
666:
667: /* main BLOB loop */
668: while (srv->srv_blob.state != kill && srv->srv_kill != kill) {
669: for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)
670: if (!c->cli_sa.sa.sa_family)
671: break;
672: if (i >= srv->srv_numcli) {
673: #ifdef HAVE_PTHREAD_YIELD
674: pthread_yield();
675: #else
676: usleep(1000000);
677: #endif
678: continue;
679: }
680:
681: FD_ZERO(&fds);
682: FD_SET(srv->srv_blob.server.cli_sock, &fds);
683: ret = select(srv->srv_blob.server.cli_sock + 1, &fds, NULL, NULL, &tv);
684: if (ret == -1) {
685: LOGERR;
686: ret = 1;
687: break;
688: }
689: if (!ret)
690: continue;
691:
692: c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa.sa, &salen);
693: if (c->cli_sock == -1) {
694: LOGERR;
695: continue;
696: } else
697: c->cli_parent = srv;
698:
699: /* spawn dispatch thread for BLOB client */
700: if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) {
701: LOGERR;
702: continue;
703: }
704: }
705:
706: srv->srv_blob.state = kill;
707:
708: pthread_attr_destroy(&attr);
709: return 0;
710: }
711:
712:
713: /*
714: * rpc_srv_initServer() - Init & create RPC Server
715: *
716: * @regProgID = ProgramID for authentication & recognition
717: * @regProcID = ProcessID for authentication & recognition
718: * @concurentClients = Concurent clients at same time to this server
719: * @netBuf = Network buffer length, if =0 == BUFSIZ (also meaning max RPC packet)
720: * @family = Family type, AF_INET, AF_INET6 or AF_LOCAL
721: * @csHost = Host name or address for bind server, if NULL any address
722: * @Port = Port for bind server, if Port == 0 default port is selected
723: * return: NULL == error or !=NULL bind and created RPC server instance
724: */
725: rpc_srv_t *
726: rpc_srv_initServer(u_int regProgID, u_int regProcID, int concurentClients,
727: int netBuf, u_short family, const char *csHost, u_short Port)
728: {
729: rpc_srv_t *srv = NULL;
730: int n = 1;
731: struct hostent *host = NULL;
732: io_sockaddr_t sa;
733:
734: ioTRACE(RPC_TRACE_LEVEL);
735:
736: if (!concurentClients || !regProgID ||
737: (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) {
738: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");
739: return NULL;
740: }
741: if (!Port)
742: Port = RPC_DEFPORT;
743: if (!netBuf)
744: netBuf = BUFSIZ;
745: else
746: netBuf = io_align(netBuf, 1); /* align netBuf length */
747: if (csHost && family != AF_LOCAL) {
748: host = gethostbyname2(csHost, family);
749: if (!host) {
750: rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno));
751: return NULL;
752: }
753: }
754: memset(&sa, 0, sizeof sa);
755: sa.sa.sa_family = family;
756: switch (family) {
757: case AF_INET:
758: sa.sin.sin_len = sizeof(struct sockaddr_in);
759: sa.sin.sin_port = htons(Port);
760: if (csHost)
761: memcpy(&sa.sin.sin_addr, host->h_addr, host->h_length);
762: break;
763: case AF_INET6:
764: sa.sin6.sin6_len = sizeof(struct sockaddr_in6);
765: sa.sin6.sin6_port = htons(Port);
766: if (csHost)
767: memcpy(&sa.sin6.sin6_addr, host->h_addr, host->h_length);
768: break;
769: case AF_LOCAL:
770: sa.sun.sun_len = sizeof(struct sockaddr_un);
771: if (csHost)
772: strlcpy(sa.sun.sun_path, csHost, sizeof sa.sun.sun_path);
773: unlink(sa.sun.sun_path);
774: break;
775: default:
776: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t start RPC server ...\n");
777: return NULL;
778: }
779:
780: srv = malloc(sizeof(rpc_srv_t));
781: if (!srv) {
782: LOGERR;
783: return NULL;
784: } else
785: memset(srv, 0, sizeof(rpc_srv_t));
786:
787: srv->srv_netbuf = netBuf;
788: srv->srv_numcli = concurentClients;
789: srv->srv_session.sess_version = RPC_VERSION;
790: srv->srv_session.sess_timeout = DEF_RPC_TIMEOUT;
791: srv->srv_session.sess_program = regProgID;
792: srv->srv_session.sess_process = regProcID;
793:
794: srv->srv_server.cli_tid = pthread_self();
795: srv->srv_server.cli_parent = srv;
796: memcpy(&srv->srv_server.cli_sa, &sa, sizeof sa);
797:
798: /* create server socket */
799: srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);
800: if (srv->srv_server.cli_sock == -1) {
801: LOGERR;
802: free(srv);
803: return NULL;
804: }
805: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
806: LOGERR;
807: close(srv->srv_server.cli_sock);
808: free(srv);
809: return NULL;
810: }
811: n = srv->srv_netbuf;
812: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
813: LOGERR;
814: close(srv->srv_server.cli_sock);
815: free(srv);
816: return NULL;
817: }
818: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
819: LOGERR;
820: close(srv->srv_server.cli_sock);
821: free(srv);
822: return NULL;
823: }
824: if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa,
825: srv->srv_server.cli_sa.sa.sa_len) == -1) {
826: LOGERR;
827: close(srv->srv_server.cli_sock);
828: free(srv);
829: return NULL;
830: }
831:
832: /* allocate pool for concurent clients */
833: srv->srv_clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
834: if (!srv->srv_clients) {
835: LOGERR;
836: close(srv->srv_server.cli_sock);
837: free(srv);
838: return NULL;
839: } else
840: memset(srv->srv_clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
841:
842: pthread_mutex_init(&srv->srv_mtx, NULL);
843:
844: rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0);
845: rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 1);
846: rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 4);
847: rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 1);
848: return srv;
849: }
850:
851: /*
852: * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
853: *
854: * @psrv = RPC Server instance
855: * return: none
856: */
857: void
858: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
859: {
860: rpc_cli_t *c;
861: register int i;
862: rpc_func_t *f;
863:
864: ioTRACE(RPC_TRACE_LEVEL);
865:
866: if (!psrv || !*psrv) {
867: rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
868: return;
869: }
870:
871: rpc_srv_endBLOBServer(*psrv);
872:
873: /* close all clients connections & server socket */
874: for (i = 0, c = (*psrv)->srv_clients; i < (*psrv)->srv_numcli && c; i++, c++)
875: if (c->cli_sa.sa.sa_family) {
876: shutdown(c->cli_sock, SHUT_RDWR);
877: close(c->cli_sock);
878: }
879: close((*psrv)->srv_server.cli_sock);
880:
881: if ((*psrv)->srv_clients) {
882: free((*psrv)->srv_clients);
883: (*psrv)->srv_clients = NULL;
884: (*psrv)->srv_numcli = 0;
885: }
886:
887: /* detach exported calls */
888: pthread_mutex_lock(&(*psrv)->srv_mtx);
889: while ((f = (*psrv)->srv_funcs)) {
890: (*psrv)->srv_funcs = f->func_next;
891: io_freeVars(&f->func_vars);
892: AIT_FREE_VAL(&f->func_name);
893: AIT_FREE_VAL(&f->func_file);
894: free(f);
895: }
896: pthread_mutex_unlock(&(*psrv)->srv_mtx);
897:
898: while (pthread_mutex_trylock(&(*psrv)->srv_mtx) == EBUSY);
899: pthread_mutex_destroy(&(*psrv)->srv_mtx);
900:
901: free(*psrv);
902: *psrv = NULL;
903: }
904:
905: /*
906: * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
907: *
908: * @srv = RPC Server instance
909: * return: -1 error or 0 ok, infinite loop ...
910: */
911: int
912: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
913: {
914: socklen_t salen = sizeof(io_sockaddr_t);
915: register int i;
916: rpc_cli_t *c;
917: fd_set fds;
918: int ret;
919: struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
920: pthread_attr_t attr;
921:
922: ioTRACE(RPC_TRACE_LEVEL);
923:
924: if (!srv) {
925: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n");
926: return -1;
927: }
928:
929: tv.tv_sec = srv->srv_session.sess_timeout;
930:
931: /* activate BLOB server worker if srv->srv_blob.state == enable */
932: rpc_srv_execBLOBServer(srv);
933:
934: if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) {
935: LOGERR;
936: return -1;
937: }
938:
939: pthread_attr_init(&attr);
940: pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
941:
942: /* main rpc loop */
943: while (srv->srv_kill != kill) {
944: for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)
945: if (!c->cli_sa.sa.sa_family)
946: break;
947: if (i >= srv->srv_numcli) {
948: #ifdef HAVE_PTHREAD_YIELD
949: pthread_yield();
950: #else
951: usleep(1000000);
952: #endif
953: continue;
954: }
955:
956: FD_ZERO(&fds);
957: FD_SET(srv->srv_server.cli_sock, &fds);
958: ret = select(srv->srv_server.cli_sock + 1, &fds, NULL, NULL, &tv);
959: if (ret == -1) {
960: LOGERR;
961: ret = 1;
962: break;
963: }
964: if (!ret)
965: continue;
966:
967: c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa.sa, &salen);
968: if (c->cli_sock == -1) {
969: LOGERR;
970: continue;
971: } else
972: c->cli_parent = srv;
973:
974: /* spawn rpc client dispatcher */
975: if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) {
976: LOGERR;
977: continue;
978: }
979: }
980:
981: pthread_attr_destroy(&attr);
982: return 0;
983: }
984:
985: // ---------------------------------------------------------
986:
987: /*
988: * rpc_srv_execCall() Execute registered call from RPC server
989: *
990: * @call = Register RPC call
991: * @rpc = IN RPC call structure
992: * @args = IN RPC calling arguments from RPC client
993: * return: -1 error, !=-1 ok
994: */
995: int
996: rpc_srv_execCall(rpc_func_t * __restrict call, struct tagRPCCall * __restrict rpc,
997: array_t * __restrict args)
998: {
999: void *dl;
1000: rpc_callback_t func;
1001: int ret;
1002:
1003: ioTRACE(RPC_TRACE_LEVEL);
1004:
1005: if (!call || !rpc || !call->func_parent) {
1006: rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1007: return -1;
1008: }
1009:
1010: dl = dlopen(AIT_VOID(&call->func_file), RTLD_NOW);
1011: if (!dl) {
1012: rpc_SetErr(ENOENT, "Can`t attach module %s!", dlerror());
1013: return -1;
1014: }
1015:
1016: func = dlsym(dl, (const char*) AIT_GET_STR(&call->func_name));
1017: if (func)
1018: ret = func(call, ntohs(rpc->call_argc), args);
1019: else {
1020: rpc_SetErr(ENOEXEC, "Can`t find function %s!", dlerror());
1021: ret = -1;
1022: }
1023:
1024: dlclose(dl);
1025: return ret;
1026: }
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