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.3.2.9 2011/08/19 12:51:50 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
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 *
50: rpc_srv_dispatchCall(void *arg)
51: {
52: rpc_cli_t *c = arg;
53: rpc_srv_t *s;
54: rpc_val_t *vals = NULL, *v = NULL;
55: rpc_func_t *f = NULL;
56: struct tagRPCCall *rpc;
57: struct tagRPCRet *rrpc;
58: rpc_sess_t ses = { 0 };
59: fd_set fds;
60: u_char buf[BUFSIZ], *data;
61: int ret, argc = 0, Limit = 0;
62: register int i;
63:
64: if (!arg) {
65: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced RPC client ...\n");
66: return NULL;
67: } else
68: s = c->cli_parent;
69:
70: do {
71: v = NULL;
72: FD_ZERO(&fds);
73: FD_SET(c->cli_sock, &fds);
74: ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);
75: if (ret == -1) {
76: if (errno == EINTR && s->srv_kill != kill)
77: continue;
78:
79: LOGERR;
80: ret = -2;
81: break;
82: }
83: memset(buf, 0, BUFSIZ);
84: ret = recv(c->cli_sock, buf, BUFSIZ, 0);
85: if (ret == -1) {
86: LOGERR;
87: ret = -3;
88: break;
89: }
90: if (!ret) { // receive EOF
91: ret = 0;
92: break;
93: }
94: if (ret < sizeof(struct tagRPCCall)) {
95: rpc_SetErr(EMSGSIZE, "Error:: too short RPC packet ...\n");
96: ret = -4;
97: break;
98: } else
99: rpc = (struct tagRPCCall*) buf;
100: // check RPC packet session info
101: if (memcmp(&rpc->call_session, &s->srv_session, sizeof rpc->call_session)) {
102: rpc_SetErr(EINVAL, "Error:: get invalid RPC session ...\n");
103: ret = -5;
104: goto makeReply;
105: } else
106: Limit = sizeof(struct tagRPCCall);
107: // RPC is OK! Go decapsulate variables ...
108: if (rpc->call_argc) {
109: v = (rpc_val_t*) (buf + Limit);
110: // check RPC packet length
111: if (rpc->call_argc * sizeof(rpc_val_t) > BUFSIZ - Limit) {
112: rpc_SetErr(EMSGSIZE, "Error:: Too big RPC packet ...\n");
113: ret = -5;
114: goto makeReply;
115: } else
116: Limit += rpc->call_argc * sizeof(rpc_val_t);
117: // RPC received variables types OK!
118: data = (u_char*) v + rpc->call_argc * sizeof(rpc_val_t);
119: for (i = 0; i < rpc->call_argc; i++) {
120: switch (v[i].val_type) {
121: case buffer:
122: if (v[i].val_len > BUFSIZ - Limit) {
123: rpc_SetErr(EMSGSIZE, "Error:: Too big RPC packet ...\n");
124: ret = -5;
125: goto makeReply;
126: } else
127: Limit += v[i].val_len;
128:
129: v[i].val.buffer = data;
130: data += v[i].val_len;
131: break;
132: case string:
133: if (v[i].val_len > BUFSIZ - Limit) {
134: rpc_SetErr(EMSGSIZE, "Error:: Too big RPC packet ...\n");
135: ret = -5;
136: goto makeReply;
137: } else
138: Limit += v[i].val_len;
139:
140: v[i].val.string = (int8_t*) data;
141: data += v[i].val_len;
142: break;
143: case blob:
144: if (s->srv_blob.state == disable) {
145: rpc_SetErr(ENOTSUP, "Error:: BLOB server is disabled\n");
146: ret = -5;
147: goto makeReply;
148: }
149: if (s->srv_blob.state == kill) {
150: rpc_SetErr(ENOTSUP, "Error:: BLOB server is killed\n");
151: ret = -5;
152: goto makeReply;
153: }
154: default:
155: break;
156: }
157: }
158: }
159:
160: argc = 0;
161: vals = NULL;
162: memcpy(&ses, &rpc->call_session, sizeof ses);
163: if (!(f = rpc_srv_getCall(s, rpc->call_tag, rpc->call_hash))) {
164: rpc_SetErr(EINVAL, "Error:: call not found into RPC server ...\n");
165: ret = -6;
166: } else
167: if ((ret = rpc_srv_execCall(f, rpc, v)) == -1)
168: ret = -9;
169: else
170: argc = rpc_srv_getVars(f, &vals);
171: makeReply:
172: memset(buf, 0, BUFSIZ);
173: rrpc = (struct tagRPCRet*) buf;
174: Limit = sizeof(struct tagRPCRet);
175:
176: memcpy(&rrpc->ret_session, &ses, sizeof(rpc_sess_t));
177: rrpc->ret_tag = rpc->call_tag;
178: rrpc->ret_hash = rpc->call_hash;
179: rrpc->ret_errno = rpc_Errno;
180: rrpc->ret_retcode = ret;
181: rrpc->ret_argc = argc;
182:
183: if (argc && vals) {
184: v = (rpc_val_t*) (buf + Limit);
185: if (argc * sizeof(rpc_val_t) > BUFSIZ - Limit) {
186: for (i = 0; i < argc; i++)
187: RPC_FREE_VAL(&vals[i]);
188: rpc_srv_freeVars(f);
189: vals = NULL;
190: argc = 0;
191: ret = -7;
192: rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet values (-7) ...\n");
193: goto makeReply;
194: } else
195: Limit += argc * sizeof(rpc_val_t);
196: memcpy(v, vals, argc * sizeof(rpc_val_t));
197: data = (u_char*) v + argc * sizeof(rpc_val_t);
198: for (ret = i = 0; i < argc; i++) {
199: switch (vals[i].val_type) {
200: case buffer:
201: if (ret || Limit + vals[i].val_len > BUFSIZ) {
202: rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n");
203: rrpc->ret_retcode = ret = -7;
204: rrpc->ret_argc = 0;
205: break;
206: }
207:
208: memcpy(data, vals[i].val.buffer, vals[i].val_len);
209: data += vals[i].val_len;
210: Limit += vals[i].val_len;
211: break;
212: case string:
213: if (ret || Limit + vals[i].val_len > BUFSIZ) {
214: rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n");
215: rrpc->ret_retcode = ret = -7;
216: rrpc->ret_argc = 0;
217: break;
218: }
219:
220: memcpy(data, vals[i].val.string, vals[i].val_len);
221: data += vals[i].val_len;
222: Limit += vals[i].val_len;
223: break;
224: case blob:
225: if (s->srv_blob.state == disable) {
226: rpc_SetErr(ENOTSUP, "Error:: BLOB server is disabled\n");
227: rrpc->ret_retcode = ret = -5;
228: rrpc->ret_argc = 0;
229: break;
230: }
231: if (s->srv_blob.state == kill) {
232: rpc_SetErr(ENOTSUP, "Error:: BLOB server is killed\n");
233: rrpc->ret_retcode = ret = -5;
234: rrpc->ret_argc = 0;
235: break;
236: }
237: default:
238: break;
239: }
240:
241: RPC_FREE_VAL(&vals[i]);
242: }
243: rpc_srv_freeVars(f);
244: vals = NULL;
245: argc = 0;
246: }
247:
248: ret = send(c->cli_sock, buf, Limit, 0);
249: if (ret == -1) {
250: LOGERR;
251: ret = -8;
252: break;
253: }
254: if (ret != Limit) {
255: rpc_SetErr(ECANCELED, "Error:: in send RPC request, should be send %d bytes, "
256: "really is %d\n", Limit, ret);
257: ret = -9;
258: break;
259: }
260: } while (ret > -1 || s->srv_kill != kill);
261:
262: shutdown(c->cli_sock, SHUT_RDWR);
263: close(c->cli_sock);
264: memset(c, 0, sizeof(rpc_cli_t));
265: return (void*) (long)ret;
266: }
267:
268:
269: static void *
270: rpc_srv_dispatchVars(void *arg)
271: {
272: rpc_cli_t *c = arg;
273: rpc_srv_t *s;
274: rpc_blob_t *b;
275: int ret = 0;
276: fd_set fds;
277: u_char buf[sizeof(struct tagBLOBHdr)];
278: struct tagBLOBHdr *blob;
279:
280: if (!arg) {
281: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced BLOB client ...\n");
282: return NULL;
283: } else
284: s = c->cli_parent;
285:
286: do {
287: // check for disable service at this moment?
288: if (s->srv_blob.state == disable && s->srv_kill != kill) {
289: usleep(100000);
290: pthread_yield();
291: continue;
292: }
293:
294: FD_ZERO(&fds);
295: FD_SET(c->cli_sock, &fds);
296: ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);
297: if (ret == -1) {
298: if (errno == EINTR && s->srv_kill != kill && s->srv_blob.state != kill)
299: continue;
300:
301: LOGERR;
302: ret = -2;
303: break;
304: }
305:
306: memset(buf, 0, sizeof buf);
307: ret = recv(c->cli_sock, buf, sizeof buf, 0);
308: if (ret == -1) {
309: LOGERR;
310: ret = -3;
311: break;
312: }
313: /* receive EOF, disable or kill service */
314: if (!ret || s->srv_blob.state == kill || s->srv_kill == kill) {
315: ret = 0;
316: break;
317: }
318: if (ret < sizeof(struct tagBLOBHdr)) {
319: rpc_SetErr(EMSGSIZE, "Error:: too short BLOB packet ...\n");
320: ret = -4;
321: break;
322: } else
323: blob = (struct tagBLOBHdr*) buf;
324: // check BLOB packet session info
325: if (memcmp(&blob->hdr_session, &s->srv_session, sizeof blob->hdr_session)) {
326: rpc_SetErr(EINVAL, "Error:: get invalid BLOB session ...\n");
327: ret = -5;
328: goto makeReply;
329: }
330: // Go to proceed packet ...
331: switch (blob->hdr_cmd) {
332: case get:
333: if (!(b = rpc_srv_getBLOB(s, blob->hdr_var))) {
334: rpc_SetErr(EINVAL, "Error:: var (%x) not found into BLOB server ...\n",
335: blob->hdr_var);
336: ret = -6;
337: break;
338: } else
339: blob->hdr_len = b->blob_len;
340:
341: if (rpc_srv_blobMap(s, b) != -1) {
342: ret = rpc_srv_sendBLOB(c, b);
343: rpc_srv_blobUnmap(b);
344: } else
345: ret = -7;
346: break;
347: case set:
348: if ((b = rpc_srv_registerBLOB(s, blob->hdr_len))) {
349: // set new BLOB variable for reply :)
350: blob->hdr_var = b->blob_var;
351:
352: ret = rpc_srv_recvBLOB(c, b);
353: rpc_srv_blobUnmap(b);
354: } else
355: ret = -7;
356: break;
357: case unset:
358: ret = rpc_srv_unregisterBLOB(s, blob->hdr_var);
359: if (ret == -1)
360: ret = -7;
361: break;
362: default:
363: rpc_SetErr(EINVAL, "Error:: unsupported BLOB command (%d)...\n",
364: blob->hdr_cmd);
365: ret = -7;
366: }
367:
368: makeReply:
369: // Replay to client!
370: blob->hdr_cmd = ret < 0 ? error : ok;
371: blob->hdr_ret = ret;
372: ret = send(c->cli_sock, buf, sizeof buf, 0);
373: if (ret == -1) {
374: LOGERR;
375: ret = -8;
376: break;
377: }
378: if (ret != sizeof buf) {
379: rpc_SetErr(ECANCELED, "Error:: in send BLOB reply, should be send %d bytes, "
380: "really is %d\n", sizeof buf, ret);
381: ret = -9;
382: break;
383: }
384: } while (ret > -1 || s->srv_kill != kill);
385:
386: shutdown(c->cli_sock, SHUT_RDWR);
387: close(c->cli_sock);
388: memset(c, 0, sizeof(rpc_cli_t));
389: return (void*) (long)ret;
390: }
391:
392: // -------------------------------------------------
393:
394: /*
395: * rpc_srv_initBLOBServer() Init & create BLOB Server
396: * @srv = RPC server instance
397: * @Port = Port for bind server, if Port == 0 default port is selected
398: * @diskDir = Disk place for BLOB file objects
399: * return: -1 == error or 0 bind and created BLOB server instance
400: */
401: int
402: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
403: {
404: int n = 1;
405: struct sockaddr sa;
406: struct sockaddr_in *sin = (struct sockaddr_in*) &sa;
407: struct sockaddr_in6 *sin6 = (struct sockaddr_in6*) &sa;
408: struct sockaddr_un *sun = (struct sockaddr_un*) &sa;
409:
410: if (!srv) {
411: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init BLOB server ...\n");
412: return -1;
413: }
414: if (srv->srv_blob.state) {
415: rpc_SetErr(EPERM, "Warning:: Already started BLOB server!\n");
416: return 0;
417: }
418: if (!Port)
419: Port = RPC_DEFPORT + 1;
420:
421: memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
422: if (access(diskDir, R_OK | W_OK) == -1) {
423: LOGERR;
424: return -1;
425: } else
426: strlcpy(srv->srv_blob.dir, diskDir, UCHAR_MAX + 1);
427:
428: srv->srv_blob.server.cli_tid = pthread_self();
429: srv->srv_blob.server.cli_parent = srv;
430:
431: memcpy(&sa, &srv->srv_server.cli_sa, sizeof sa);
432: switch (srv->srv_server.cli_sa.sa_family) {
433: case AF_INET:
434: sin->sin_port = htons(Port);
435: memcpy(&srv->srv_blob.server.cli_sa, sin, sizeof(struct sockaddr));
436: break;
437: case AF_INET6:
438: sin6->sin6_port = htons(Port);
439: memcpy(&srv->srv_blob.server.cli_sa, sin6, sizeof(struct sockaddr));
440: break;
441: case AF_LOCAL:
442: strlcat(sun->sun_path, ".blob", sizeof sun->sun_path);
443: memcpy(&srv->srv_blob.server.cli_sa, sun, sizeof(struct sockaddr));
444: break;
445: default:
446: return -1;
447: }
448:
449: /* create BLOB server socket */
450: srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa_family, SOCK_STREAM, 0);
451: if (srv->srv_blob.server.cli_sock == -1) {
452: LOGERR;
453: return -1;
454: }
455: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
456: LOGERR;
457: close(srv->srv_blob.server.cli_sock);
458: return -1;
459: }
460: if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa,
461: sizeof srv->srv_blob.server.cli_sa) == -1) {
462: LOGERR;
463: close(srv->srv_blob.server.cli_sock);
464: return -1;
465: }
466:
467: /* allocate pool for concurent clients */
468: srv->srv_blob.clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
469: if (!srv->srv_blob.clients) {
470: LOGERR;
471: close(srv->srv_blob.server.cli_sock);
472: return -1;
473: } else
474: memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
475:
476: pthread_mutex_init(&srv->srv_blob.mtx, NULL);
477:
478: pthread_mutex_lock(&srv->srv_mtx);
479: rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
480: rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 0);
481: rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 0);
482: rpc_srv_registerCall(srv, NULL, CALL_BLOBSTATE, 1);
483: pthread_mutex_unlock(&srv->srv_mtx);
484:
485: srv->srv_blob.state = enable; // enable BLOB
486: return 0;
487: }
488:
489: /*
490: * rpc_srv_endBLOBServer() Destroy BLOB server, close all opened sockets and free resources
491: * @srv = RPC Server instance
492: * return: none
493: */
494: void
495: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
496: {
497: rpc_cli_t *c;
498: register int i;
499: rpc_blob_t *f;
500:
501: if (!srv) {
502: rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
503: return;
504: } else
505: srv->srv_blob.state = kill;
506:
507: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);
508: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS);
509: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
510: rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);
511:
512: /* close all clients connections & server socket */
513: for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
514: if (c->cli_sa.sa_family)
515: shutdown(c->cli_sock, SHUT_RDWR);
516: close(srv->srv_blob.server.cli_sock);
517:
518: if (srv->srv_blob.clients) {
519: free(srv->srv_blob.clients);
520: srv->srv_blob.clients = NULL;
521: }
522:
523: /* detach blobs */
524: pthread_mutex_lock(&srv->srv_blob.mtx);
525: while ((f = srv->srv_blob.blobs)) {
526: srv->srv_blob.blobs = f->blob_next;
527: rpc_srv_blobFree(srv, f);
528: free(f);
529: }
530: pthread_mutex_unlock(&srv->srv_blob.mtx);
531:
532: while (pthread_mutex_trylock(&srv->srv_blob.mtx) == EBUSY);
533: pthread_mutex_destroy(&srv->srv_blob.mtx);
534: }
535:
536: /*
537: * rpc_srv_execBLOBServer() Execute Main BLOB server loop and wait for clients requests
538: * @srv = RPC Server instance
539: * return: -1 error or 0 ok, infinite loop ...
540: */
541: int
542: rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv)
543: {
544: socklen_t salen = sizeof(struct sockaddr);
545: register int i;
546: rpc_cli_t *c;
547: fd_set fds;
548: int ret;
549: struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
550:
551: if (!srv || srv->srv_blob.state == kill) {
552: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n");
553: return -1;
554: }
555:
556: if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
557: LOGERR;
558: return -1;
559: }
560:
561: while (srv->srv_blob.state != kill && srv->srv_kill != kill) {
562: for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)
563: if (!c->cli_sa.sa_family)
564: break;
565: if (i >= srv->srv_numcli) {
566: usleep(1000000);
567: continue;
568: }
569:
570: FD_ZERO(&fds);
571: FD_SET(srv->srv_blob.server.cli_sock, &fds);
572: ret = select(srv->srv_blob.server.cli_sock + 1, &fds, NULL, NULL, &tv);
573: if (ret == -1) {
574: LOGERR;
575: ret = 1;
576: break;
577: }
578: if (!ret)
579: continue;
580:
581: c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa, &salen);
582: if (c->cli_sock == -1) {
583: LOGERR;
584: continue;
585: } else
586: c->cli_parent = srv;
587:
588: if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchVars, c)) {
589: LOGERR;
590: continue;
591: } else
592: pthread_detach(c->cli_tid);
593: }
594:
595: srv->srv_blob.state = disable;
596:
597: return 0;
598: }
599:
600:
601: /*
602: * rpc_srv_initServer() Init & create RPC Server
603: * @regProgID = ProgramID for authentication & recognition
604: * @regProcID = ProcessID for authentication & recognition
605: * @concurentClients = Concurent clients at same time to this server
606: * @family = Family type, AF_INET, AF_INET6 or AF_LOCAL
607: * @csHost = Host name or address for bind server, if NULL any address
608: * @Port = Port for bind server, if Port == 0 default port is selected
609: * return: NULL == error or !=NULL bind and created RPC server instance
610: */
611: rpc_srv_t *
612: rpc_srv_initServer(u_int regProgID, u_int regProcID, int concurentClients,
613: u_short family, const char *csHost, u_short Port)
614: {
615: rpc_srv_t *srv = NULL;
616: int n = 1;
617: struct hostent *host = NULL;
618: struct sockaddr sa;
619: struct sockaddr_in *sin = (struct sockaddr_in*) &sa;
620: struct sockaddr_in6 *sin6 = (struct sockaddr_in6*) &sa;
621: struct sockaddr_un *sun = (struct sockaddr_un*) &sa;
622:
623: if (!concurentClients || !regProgID ||
624: (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) {
625: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");
626: return NULL;
627: }
628: if (!Port)
629: Port = RPC_DEFPORT;
630: if (csHost && family != AF_LOCAL) {
631: host = gethostbyname2(csHost, family);
632: if (!host) {
633: rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno));
634: return NULL;
635: }
636: }
637: memset(&sa, 0, sizeof sa);
638: sa.sa_family = family;
639: switch (family) {
640: case AF_INET:
641: sin->sin_len = sizeof(struct sockaddr_in);
642: sin->sin_port = htons(Port);
643: if (csHost)
644: memcpy(&sin->sin_addr, host->h_addr, host->h_length);
645: break;
646: case AF_INET6:
647: sin6->sin6_len = sizeof(struct sockaddr_in6);
648: sin6->sin6_port = htons(Port);
649: if (csHost)
650: memcpy(&sin6->sin6_addr, host->h_addr, host->h_length);
651: break;
652: case AF_LOCAL:
653: sun->sun_len = sizeof(struct sockaddr_un);
654: if (csHost)
655: strlcpy(sun->sun_path, csHost, sizeof sun->sun_path);
656: break;
657: default:
658: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t start RPC server ...\n");
659: return NULL;
660: }
661:
662: srv = malloc(sizeof(rpc_srv_t));
663: if (!srv) {
664: LOGERR;
665: return NULL;
666: } else
667: memset(srv, 0, sizeof(rpc_srv_t));
668:
669: srv->srv_numcli = concurentClients;
670: srv->srv_session.sess_version = RPC_VERSION;
671: srv->srv_session.sess_program = regProgID;
672: srv->srv_session.sess_process = regProcID;
673:
674: srv->srv_server.cli_tid = pthread_self();
675: srv->srv_server.cli_parent = srv;
676: switch (family) {
677: case AF_INET:
678: memcpy(&srv->srv_server.cli_sa, sin, sizeof srv->srv_server.cli_sa);
679: break;
680: case AF_INET6:
681: memcpy(&srv->srv_server.cli_sa, sin6, sizeof srv->srv_server.cli_sa);
682: break;
683: case AF_LOCAL:
684: memcpy(&srv->srv_server.cli_sa, sun, sizeof srv->srv_server.cli_sa);
685: unlink(sun->sun_path);
686: break;
687: }
688:
689: /* create server socket */
690: srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);
691: if (srv->srv_server.cli_sock == -1) {
692: LOGERR;
693: free(srv);
694: return NULL;
695: }
696: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
697: LOGERR;
698: close(srv->srv_server.cli_sock);
699: free(srv);
700: return NULL;
701: }
702: if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa, sizeof srv->srv_server.cli_sa) == -1) {
703: LOGERR;
704: close(srv->srv_server.cli_sock);
705: free(srv);
706: return NULL;
707: }
708:
709: /* allocate pool for concurent clients */
710: srv->srv_clients = calloc(srv->srv_numcli, sizeof(rpc_cli_t));
711: if (!srv->srv_clients) {
712: LOGERR;
713: close(srv->srv_server.cli_sock);
714: free(srv);
715: return NULL;
716: } else
717: memset(srv->srv_clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));
718:
719: pthread_mutex_init(&srv->srv_mtx, NULL);
720:
721: rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0);
722: rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 0);
723: rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 0);
724: rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 0);
725: return srv;
726: }
727:
728: /*
729: * rpc_srv_endServer() Destroy RPC server, close all opened sockets and free resources
730: * @srv = RPC Server instance
731: * return: none
732: */
733: void
734: rpc_srv_endServer(rpc_srv_t * __restrict srv)
735: {
736: rpc_cli_t *c;
737: register int i;
738: rpc_func_t *f;
739:
740: if (!srv) {
741: rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
742: return;
743: }
744:
745: rpc_srv_endBLOBServer(srv);
746:
747: /* close all clients connections & server socket */
748: for (i = 0, c = srv->srv_clients; i < srv->srv_numcli && c; i++, c++)
749: if (c->cli_sa.sa_family) {
750: shutdown(c->cli_sock, SHUT_RDWR);
751: close(c->cli_sock);
752: }
753: close(srv->srv_server.cli_sock);
754:
755: if (srv->srv_clients) {
756: free(srv->srv_clients);
757: srv->srv_clients = NULL;
758: srv->srv_numcli = 0;
759: }
760:
761: /* detach exported calls */
762: pthread_mutex_lock(&srv->srv_mtx);
763: while ((f = srv->srv_funcs)) {
764: srv->srv_funcs = f->func_next;
765: free(f);
766: }
767: pthread_mutex_unlock(&srv->srv_mtx);
768:
769: while (pthread_mutex_trylock(&srv->srv_mtx) == EBUSY);
770: pthread_mutex_destroy(&srv->srv_mtx);
771:
772: free(srv);
773: srv = NULL;
774: }
775:
776: /*
777: * rpc_srv_execServer() Execute Main server loop and wait for clients requests
778: * @srv = RPC Server instance
779: * return: -1 error or 0 ok, infinite loop ...
780: */
781: int
782: rpc_srv_execServer(rpc_srv_t * __restrict srv)
783: {
784: socklen_t salen = sizeof(struct sockaddr);
785: register int i;
786: rpc_cli_t *c;
787: fd_set fds;
788: int ret;
789: struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
790:
791: if (!srv) {
792: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n");
793: return -1;
794: }
795:
796: if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) {
797: LOGERR;
798: return -1;
799: }
800:
801: while (srv->srv_kill != kill) {
802: for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)
803: if (!c->cli_sa.sa_family)
804: break;
805: if (i >= srv->srv_numcli) {
806: usleep(1000000);
807: continue;
808: }
809:
810: FD_ZERO(&fds);
811: FD_SET(srv->srv_server.cli_sock, &fds);
812: ret = select(srv->srv_server.cli_sock + 1, &fds, NULL, NULL, &tv);
813: if (ret == -1) {
814: LOGERR;
815: ret = 1;
816: break;
817: }
818: if (!ret)
819: continue;
820:
821: c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa, &salen);
822: if (c->cli_sock == -1) {
823: LOGERR;
824: continue;
825: } else
826: c->cli_parent = srv;
827:
828: if (pthread_create(&c->cli_tid, NULL, rpc_srv_dispatchCall, c)) {
829: LOGERR;
830: continue;
831: } else
832: pthread_detach(c->cli_tid);
833: }
834:
835: return 0;
836: }
837:
838: // ---------------------------------------------------------
839:
840: /*
841: * rpc_srv_execCall() Execute registered call from RPC server
842: * @call = Register RPC call
843: * @rpc = IN RPC call structure
844: * @args = IN RPC call array of rpc values
845: * return: -1 error, !=-1 ok
846: */
847: int
848: rpc_srv_execCall(rpc_func_t * __restrict call, struct tagRPCCall * __restrict rpc,
849: rpc_val_t * __restrict args)
850: {
851: void *dl;
852: rpc_callback_t func;
853: int ret;
854:
855: if (!call || !rpc || !call->func_parent) {
856: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t exec call from RPC server ...\n");
857: return -1;
858: }
859:
860: dl = dlopen((char*) (*call->func_file ? call->func_file : NULL), RTLD_NOW);
861: if (!dl) {
862: rpc_SetErr(ENOENT, "Error:: Can`t attach module %s!\n", dlerror());
863: return -1;
864: }
865:
866: func = dlsym(dl, (char*) call->func_name);
867: if (func)
868: ret = func(call, rpc->call_argc, args);
869: else {
870: rpc_SetErr(ENOEXEC, "Error:: Can`t find function %s!\n", dlerror());
871: ret = -1;
872: }
873:
874: dlclose(dl);
875: return ret;
876: }
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