![[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|
Return to srv.c CVS log | Up to [ELWIX - Embedded LightWeight unIX -] / libaitrpc / src |
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.3.2.9 ! misho 6: * $Id: srv.c,v 1.3.2.8 2011/08/19 09:32:38 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:
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: */
1.1 misho 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;
1.2 misho 54: rpc_val_t *vals = NULL, *v = NULL;
1.3 misho 55: rpc_func_t *f = NULL;
1.1 misho 56: struct tagRPCCall *rpc;
1.3 misho 57: struct tagRPCRet *rrpc;
58: rpc_sess_t ses = { 0 };
1.1 misho 59: fd_set fds;
60: u_char buf[BUFSIZ], *data;
1.2 misho 61: int ret, argc = 0, Limit = 0;
1.1 misho 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 {
1.3 misho 71: v = NULL;
1.1 misho 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) {
1.3.2.9 ! misho 76: if (errno == EINTR && s->srv_kill != kill)
! 77: continue;
! 78:
! 79: LOGERR;
1.1 misho 80: ret = -2;
1.3.2.9 ! misho 81: break;
1.1 misho 82: }
83: memset(buf, 0, BUFSIZ);
1.3.2.9 ! misho 84: ret = recv(c->cli_sock, buf, BUFSIZ, 0);
! 85: if (ret == -1) {
1.1 misho 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;
1.2 misho 104: goto makeReply;
105: } else
106: Limit = sizeof(struct tagRPCCall);
1.1 misho 107: // RPC is OK! Go decapsulate variables ...
108: if (rpc->call_argc) {
1.2 misho 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);
1.1 misho 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:
1.2 misho 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:
1.1 misho 129: v[i].val.buffer = data;
130: data += v[i].val_len;
131: break;
132: case string:
1.3 misho 133: if (v[i].val_len > BUFSIZ - Limit) {
1.2 misho 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:
1.1 misho 140: v[i].val.string = (int8_t*) data;
1.3 misho 141: data += v[i].val_len;
1.1 misho 142: break;
1.2 misho 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: }
1.3.2.9 ! misho 149: if (s->srv_blob.state == kill) {
! 150: rpc_SetErr(ENOTSUP, "Error:: BLOB server is killed\n");
! 151: ret = -5;
! 152: goto makeReply;
! 153: }
1.1 misho 154: default:
155: break;
156: }
157: }
158: }
159:
160: argc = 0;
161: vals = NULL;
1.3 misho 162: memcpy(&ses, &rpc->call_session, sizeof ses);
1.1 misho 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
1.2 misho 167: if ((ret = rpc_srv_execCall(f, rpc, v)) == -1)
168: ret = -9;
1.1 misho 169: else
1.3.2.7 misho 170: argc = rpc_srv_getVars(f, &vals);
1.3 misho 171: makeReply:
172: memset(buf, 0, BUFSIZ);
173: rrpc = (struct tagRPCRet*) buf;
174: Limit = sizeof(struct tagRPCRet);
1.1 misho 175:
1.3 misho 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;
1.1 misho 182:
183: if (argc && vals) {
1.3 misho 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]);
1.3.2.7 misho 188: rpc_srv_freeVars(f);
1.3 misho 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);
1.1 misho 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");
1.3 misho 203: rrpc->ret_retcode = ret = -7;
204: rrpc->ret_argc = 0;
1.1 misho 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:
1.3 misho 213: if (ret || Limit + vals[i].val_len > BUFSIZ) {
1.1 misho 214: rpc_SetErr(EMSGSIZE, "Error:: in prepare RPC packet (-7) ...\n");
1.3 misho 215: rrpc->ret_retcode = ret = -7;
216: rrpc->ret_argc = 0;
1.1 misho 217: break;
218: }
219:
1.3 misho 220: memcpy(data, vals[i].val.string, vals[i].val_len);
221: data += vals[i].val_len;
222: Limit += vals[i].val_len;
1.1 misho 223: break;
1.2 misho 224: case blob:
225: if (s->srv_blob.state == disable) {
226: rpc_SetErr(ENOTSUP, "Error:: BLOB server is disabled\n");
1.3 misho 227: rrpc->ret_retcode = ret = -5;
228: rrpc->ret_argc = 0;
1.1 misho 229: break;
230: }
1.3.2.9 ! misho 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: }
1.1 misho 237: default:
238: break;
239: }
240:
241: RPC_FREE_VAL(&vals[i]);
242: }
1.3.2.7 misho 243: rpc_srv_freeVars(f);
1.3 misho 244: vals = NULL;
245: argc = 0;
1.1 misho 246: }
247:
1.3.2.9 ! misho 248: ret = send(c->cli_sock, buf, Limit, 0);
! 249: if (ret == -1) {
1.1 misho 250: LOGERR;
251: ret = -8;
252: break;
253: }
254: if (ret != Limit) {
1.2 misho 255: rpc_SetErr(ECANCELED, "Error:: in send RPC request, should be send %d bytes, "
1.1 misho 256: "really is %d\n", Limit, ret);
257: ret = -9;
258: break;
259: }
1.3.2.9 ! misho 260: } while (ret > -1 || s->srv_kill != kill);
1.1 misho 261:
262: shutdown(c->cli_sock, SHUT_RDWR);
263: close(c->cli_sock);
264: memset(c, 0, sizeof(rpc_cli_t));
1.2 misho 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;
1.3.2.9 ! misho 275: int ret = 0;
1.2 misho 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?
1.3.2.9 ! misho 288: if (s->srv_blob.state == disable && s->srv_kill != kill) {
! 289: usleep(100000);
! 290: pthread_yield();
! 291: continue;
1.2 misho 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) {
1.3.2.9 ! misho 298: if (errno == EINTR && s->srv_kill != kill && s->srv_blob.state != kill)
! 299: continue;
! 300:
! 301: LOGERR;
1.2 misho 302: ret = -2;
1.3.2.9 ! misho 303: break;
1.2 misho 304: }
305:
306: memset(buf, 0, sizeof buf);
1.3.2.9 ! misho 307: ret = recv(c->cli_sock, buf, sizeof buf, 0);
! 308: if (ret == -1) {
1.2 misho 309: LOGERR;
310: ret = -3;
311: break;
312: }
1.3.2.9 ! misho 313: /* receive EOF, disable or kill service */
! 314: if (!ret || s->srv_blob.state == kill || s->srv_kill == kill) {
1.2 misho 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;
1.3.2.9 ! misho 372: ret = send(c->cli_sock, buf, sizeof buf, 0);
! 373: if (ret == -1) {
1.2 misho 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: }
1.3.2.9 ! misho 384: } while (ret > -1 || s->srv_kill != kill);
1.2 misho 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;
1.1 misho 390: }
391:
392: // -------------------------------------------------
393:
394: /*
1.2 misho 395: * rpc_srv_initBLOBServer() Init & create BLOB Server
1.3.2.3 misho 396: * @srv = RPC server instance
1.2 misho 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
1.3.2.9 ! misho 402: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
1.2 misho 403: {
404: int n = 1;
1.3.2.3 misho 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;
1.2 misho 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;
1.3.2.3 misho 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;
1.2 misho 447: }
448:
1.3.2.3 misho 449: /* create BLOB server socket */
1.3.2.9 ! misho 450: srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa_family, SOCK_STREAM, 0);
1.2 misho 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:
1.3.2.3 misho 467: /* allocate pool for concurent clients */
1.2 misho 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
1.3.2.9 ! misho 505: srv->srv_blob.state = kill;
1.2 misho 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:
1.3.2.3 misho 512: /* close all clients connections & server socket */
1.2 misho 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:
1.3.2.3 misho 523: /* detach blobs */
1.2 misho 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:
1.3.2.9 ! misho 551: if (!srv || srv->srv_blob.state == kill) {
1.2 misho 552: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n");
553: return -1;
554: }
555:
1.3.2.9 ! misho 556: if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
! 557: LOGERR;
! 558: return -1;
! 559: }
1.2 misho 560:
1.3.2.6 misho 561: while (srv->srv_blob.state != kill && srv->srv_kill != kill) {
1.2 misho 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:
1.3.2.9 ! misho 581: c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa, &salen);
1.2 misho 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;
1.3.2.3 misho 591: } else
1.3.2.9 ! misho 592: pthread_detach(c->cli_tid);
1.2 misho 593: }
594:
595: srv->srv_blob.state = disable;
596:
597: return 0;
598: }
599:
600:
601: /*
1.1 misho 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
1.3.2.1 misho 606: * @family = Family type, AF_INET, AF_INET6 or AF_LOCAL
607: * @csHost = Host name or address for bind server, if NULL any address
1.1 misho 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,
1.3.2.9 ! misho 613: u_short family, const char *csHost, u_short Port)
1.1 misho 614: {
615: rpc_srv_t *srv = NULL;
616: int n = 1;
617: struct hostent *host = NULL;
1.3.2.1 misho 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;
1.1 misho 622:
1.3.2.1 misho 623: if (!concurentClients || !regProgID ||
624: (family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) {
1.1 misho 625: rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");
626: return NULL;
627: }
628: if (!Port)
629: Port = RPC_DEFPORT;
1.3.2.1 misho 630: if (csHost && family != AF_LOCAL) {
1.1 misho 631: host = gethostbyname2(csHost, family);
632: if (!host) {
633: rpc_SetErr(h_errno, "Error:: %s\n", hstrerror(h_errno));
634: return NULL;
635: }
636: }
1.3.2.1 misho 637: memset(&sa, 0, sizeof sa);
638: sa.sa_family = family;
1.1 misho 639: switch (family) {
640: case AF_INET:
1.3.2.1 misho 641: sin->sin_len = sizeof(struct sockaddr_in);
642: sin->sin_port = htons(Port);
1.1 misho 643: if (csHost)
1.3.2.1 misho 644: memcpy(&sin->sin_addr, host->h_addr, host->h_length);
1.1 misho 645: break;
646: case AF_INET6:
1.3.2.1 misho 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);
1.1 misho 654: if (csHost)
1.3.2.1 misho 655: strlcpy(sun->sun_path, csHost, sizeof sun->sun_path);
1.1 misho 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:
1.2 misho 674: srv->srv_server.cli_tid = pthread_self();
1.1 misho 675: srv->srv_server.cli_parent = srv;
1.3.2.1 misho 676: switch (family) {
677: case AF_INET:
1.3.2.3 misho 678: memcpy(&srv->srv_server.cli_sa, sin, sizeof srv->srv_server.cli_sa);
1.3.2.1 misho 679: break;
680: case AF_INET6:
1.3.2.3 misho 681: memcpy(&srv->srv_server.cli_sa, sin6, sizeof srv->srv_server.cli_sa);
1.3.2.1 misho 682: break;
683: case AF_LOCAL:
1.3.2.3 misho 684: memcpy(&srv->srv_server.cli_sa, sun, sizeof srv->srv_server.cli_sa);
1.3.2.1 misho 685: unlink(sun->sun_path);
686: break;
687: }
688:
689: /* create server socket */
1.3.2.9 ! misho 690: srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);
1.1 misho 691: if (srv->srv_server.cli_sock == -1) {
692: LOGERR;
693: free(srv);
694: return NULL;
1.3.2.4 misho 695: }
1.1 misho 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:
1.3.2.1 misho 709: /* allocate pool for concurent clients */
1.1 misho 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:
1.2 misho 719: pthread_mutex_init(&srv->srv_mtx, NULL);
720:
721: rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0);
1.1 misho 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:
1.2 misho 745: rpc_srv_endBLOBServer(srv);
1.1 misho 746:
1.3.2.1 misho 747: /* close all clients connections & server socket */
1.1 misho 748: for (i = 0, c = srv->srv_clients; i < srv->srv_numcli && c; i++, c++)
1.2 misho 749: if (c->cli_sa.sa_family) {
1.1 misho 750: shutdown(c->cli_sock, SHUT_RDWR);
1.2 misho 751: close(c->cli_sock);
752: }
1.1 misho 753: close(srv->srv_server.cli_sock);
754:
755: if (srv->srv_clients) {
756: free(srv->srv_clients);
1.2 misho 757: srv->srv_clients = NULL;
1.1 misho 758: srv->srv_numcli = 0;
759: }
760:
1.3.2.1 misho 761: /* detach exported calls */
1.2 misho 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:
1.1 misho 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;
1.2 misho 787: fd_set fds;
788: int ret;
789: struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
1.1 misho 790:
791: if (!srv) {
792: rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start RPC server ...\n");
793: return -1;
794: }
795:
1.3.2.9 ! misho 796: if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) {
! 797: LOGERR;
! 798: return -1;
! 799: }
1.1 misho 800:
1.3.2.6 misho 801: while (srv->srv_kill != kill) {
1.1 misho 802: for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)
803: if (!c->cli_sa.sa_family)
804: break;
1.2 misho 805: if (i >= srv->srv_numcli) {
1.1 misho 806: usleep(1000000);
807: continue;
808: }
1.2 misho 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:
1.3.2.9 ! misho 821: c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa, &salen);
1.1 misho 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;
1.3.2.9 ! misho 831: } else
! 832: pthread_detach(c->cli_tid);
1.1 misho 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
1.2 misho 848: rpc_srv_execCall(rpc_func_t * __restrict call, struct tagRPCCall * __restrict rpc,
849: rpc_val_t * __restrict args)
1.1 misho 850: {
851: void *dl;
852: rpc_callback_t func;
853: int ret;
854:
1.2 misho 855: if (!call || !rpc || !call->func_parent) {
1.1 misho 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)
1.2 misho 868: ret = func(call, rpc->call_argc, args);
1.1 misho 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: }