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.24.2.2 2015/01/15 20:37:14 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 - 2015
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: /* SOCK_STREAM */
50: static void *acceptClients(sched_task_t *);
51: static void *closeClient(sched_task_t *);
52: static void *rxPacket(sched_task_t *);
53: static void *txPacket(sched_task_t *);
54:
55: /* SOCK_DGRAM */
56: static void *freeClient(sched_task_t *);
57: static void *rxUDPPacket(sched_task_t *);
58: static void *txUDPPacket(sched_task_t *);
59:
60: /* SOCK_RAW */
61:
62: /* SOCK_BPF */
63: static void *rxBPFPacket(sched_task_t *);
64: static void *txBPFPacket(sched_task_t *);
65:
66: /* SOCK_PIPE */
67:
68: static sched_task_func_t cbProto[SOCK_MAX_SUPPORT][4] = {
69: { acceptClients, closeClient, rxPacket, txPacket }, /* SOCK_STREAM */
70: { acceptClients, closeClient, rxPacket, txPacket }, /* SOCK_STREAM */
71: { rxUDPPacket, freeClient, rxUDPPacket, txUDPPacket }, /* SOCK_DGRAM */
72: { NULL, NULL, NULL, NULL }, /* SOCK_RAW */
73: { rxBPFPacket, freeClient, rxBPFPacket, txBPFPacket }, /* SOCK_BPF */
74: { NULL, NULL, NULL, NULL } /* SOCK_PIPE */
75: };
76:
77: /* Global Signal Argument when kqueue support disabled */
78:
79: static volatile uintptr_t _glSigArg = 0;
80:
81:
82: void
83: rpc_freeCli(rpc_cli_t * __restrict c)
84: {
85: rpc_srv_t *s = c->cli_parent;
86:
87: schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
88:
89: /* free buffer */
90: AIT_FREE_VAL(&c->cli_buf);
91:
92: array_Del(s->srv_clients, c->cli_id, 0);
93: if (c)
94: e_free(c);
95: }
96:
97:
98: static inline int
99: _check4freeslot(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
100: {
101: rpc_cli_t *c = NULL;
102: register int i;
103:
104: /* check free slots for connect */
105: for (i = 0; i < array_Size(srv->srv_clients) &&
106: (c = array(srv->srv_clients, i, rpc_cli_t*)); i++)
107: /* check for duplicates */
108: if (sa && !e_addrcmp(&c->cli_sa, sa, 42))
109: break;
110: if (i >= array_Size(srv->srv_clients))
111: return -1; /* no more free slots! */
112:
113: return i;
114: }
115:
116: static rpc_cli_t *
117: _allocClient(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
118: {
119: rpc_cli_t *c = NULL;
120: int n;
121:
122: n = _check4freeslot(srv, sa);
123: if (n == -1)
124: return NULL;
125: else
126: c = array(srv->srv_clients, n, rpc_cli_t*);
127:
128: if (!c) {
129: c = e_malloc(sizeof(rpc_cli_t));
130: if (!c) {
131: LOGERR;
132: srv->srv_kill = 1;
133: return NULL;
134: } else {
135: memset(c, 0, sizeof(rpc_cli_t));
136: array_Set(srv->srv_clients, n, c);
137: c->cli_id = n;
138: c->cli_parent = srv;
139: }
140:
141: /* alloc empty buffer */
142: AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
143: }
144:
145: return c;
146: }
147:
148:
149: static void *
150: freeClient(sched_task_t *task)
151: {
152: rpc_freeCli(TASK_ARG(task));
153:
154: return NULL;
155: }
156:
157: static void *
158: closeClient(sched_task_t *task)
159: {
160: int sock = ((rpc_cli_t*) TASK_ARG(task))->cli_sock;
161:
162: rpc_freeCli(TASK_ARG(task));
163:
164: /* close client socket */
165: shutdown(sock, SHUT_RDWR);
166: close(sock);
167: return NULL;
168: }
169:
170: static void *
171: txPacket(sched_task_t *task)
172: {
173: rpc_cli_t *c = TASK_ARG(task);
174: rpc_srv_t *s = c->cli_parent;
175: rpc_func_t *f = NULL;
176: u_char *buf = AIT_GET_BUF(&c->cli_buf);
177: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
178: int ret, estlen, wlen = sizeof(struct tagRPCCall);
179: struct pollfd pfd;
180: #ifdef TCP_SESSION_TIMEOUT
181: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
182:
183: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
184: schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
185: TASK_ARG(task), ts, TASK_ARG(task), 0);
186: #endif
187:
188: if (rpc->call_argc) {
189: f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
190: if (!f) {
191: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
192:
193: rpc->call_argc ^= rpc->call_argc;
194: rpc->call_rep.ret = RPC_ERROR(-1);
195: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
196: } else {
197: /* calc estimated length */
198: estlen = ait_resideVars(RPC_RETVARS(c)) + wlen;
199: if (estlen > AIT_LEN(&c->cli_buf))
200: AIT_RE_BUF(&c->cli_buf, estlen);
201: buf = AIT_GET_BUF(&c->cli_buf);
202: rpc = (struct tagRPCCall*) buf;
203:
204: rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
205: /* Go Encapsulate variables */
206: ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
207: RPC_RETVARS(c));
208: /* Free return values */
209: ait_freeVars(&c->cli_vars);
210: if (ret == -1) {
211: rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
212:
213: rpc->call_argc ^= rpc->call_argc;
214: rpc->call_rep.ret = RPC_ERROR(-1);
215: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
216: } else
217: wlen += ret;
218: }
219: }
220:
221: rpc->call_len = htonl(wlen);
222:
223: #if 0
224: /* calculate CRC */
225: rpc->call_crc ^= rpc->call_crc;
226: rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
227: #endif
228:
229: /* send reply */
230: pfd.fd = TASK_FD(task);
231: pfd.events = POLLOUT;
232: for (; wlen > 0; wlen -= ret, buf += ret) {
233: if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 ||
234: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
235: if (ret)
236: LOGERR;
237: else
238: rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
239: /* close connection */
240: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
241: TASK_ARG(task), 0, NULL, 0);
242: return NULL;
243: }
244: ret = send(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), MSG_NOSIGNAL);
245: if (ret == -1) {
246: /* close connection */
247: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
248: TASK_ARG(task), 0, NULL, 0);
249: return NULL;
250: }
251: }
252:
253: return NULL;
254: }
255:
256: static void *
257: execCall(sched_task_t *task)
258: {
259: rpc_cli_t *c = TASK_ARG(task);
260: rpc_srv_t *s = c->cli_parent;
261: rpc_func_t *f = NULL;
262: array_t *arr = NULL;
263: u_char *buf = AIT_GET_BUF(&c->cli_buf);
264: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
265: int argc = ntohs(rpc->call_argc);
266:
267: /* Go decapsulate variables ... */
268: if (argc) {
269: arr = ait_buffer2vars(buf + sizeof(struct tagRPCCall),
270: AIT_LEN(&c->cli_buf) - sizeof(struct tagRPCCall), argc, 42);
271: if (!arr) {
272: rpc_SetErr(ERPCMISMATCH, "#%d - %s", elwix_GetErrno(), elwix_GetError());
273:
274: rpc->call_argc ^= rpc->call_argc;
275: rpc->call_rep.ret = RPC_ERROR(-1);
276: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
277: return NULL;
278: }
279: } else
280: arr = NULL;
281:
282: if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag)))) {
283: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
284:
285: rpc->call_argc ^= rpc->call_argc;
286: rpc->call_rep.ret = RPC_ERROR(-1);
287: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
288: } else {
289: /* if client doesn't want reply */
290: argc = RPC_CHK_NOREPLY(rpc);
291: rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(c, rpc, f->func_name, arr));
292: if (rpc->call_rep.ret == htonl(-1)) {
293: if (!rpc->call_rep.eno) {
294: LOGERR;
295: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
296: }
297: rpc->call_argc ^= rpc->call_argc;
298: } else {
299: rpc->call_rep.eno ^= rpc->call_rep.eno;
300: if (argc) {
301: /* without reply */
302: ait_freeVars(&c->cli_vars);
303: rpc->call_argc ^= rpc->call_argc;
304: } else {
305: /* reply */
306: rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
307: }
308: }
309: }
310:
311: array_Destroy(&arr);
312: return NULL;
313: }
314:
315: static void *
316: rxPacket(sched_task_t *task)
317: {
318: rpc_cli_t *c = TASK_ARG(task);
319: rpc_srv_t *s = c->cli_parent;
320: int len, rlen, noreply, estlen;
321: #if 0
322: u_short crc;
323: #endif
324: u_char *buf = AIT_GET_BUF(&c->cli_buf);
325: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
326: struct pollfd pfd;
327: #ifdef TCP_SESSION_TIMEOUT
328: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
329:
330: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
331: schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
332: TASK_ARG(task), ts, TASK_ARG(task), 0);
333: #endif
334:
335: memset(buf, 0, sizeof(struct tagRPCCall));
336: rlen = recv(TASK_FD(task), rpc, sizeof(struct tagRPCCall), MSG_PEEK);
337: if (rlen < sizeof(struct tagRPCCall)) {
338: /* close connection */
339: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
340: TASK_ARG(task), 0, NULL, 0);
341: return NULL;
342: } else {
343: estlen = ntohl(rpc->call_len);
344: if (estlen > AIT_LEN(&c->cli_buf))
345: AIT_RE_BUF(&c->cli_buf, estlen);
346: rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
347: buf = AIT_GET_BUF(&c->cli_buf);
348: len = estlen;
349: }
350:
351: /* get next part of packet */
352: memset(buf, 0, len);
353: pfd.fd = TASK_FD(task);
354: pfd.events = POLLIN | POLLPRI;
355: for (; len > 0; len -= rlen, buf += rlen) {
356: if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 ||
357: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
358: if (rlen)
359: LOGERR;
360: else
361: rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
362: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
363: TASK_ARG(task), 0, NULL, 0);
364: return NULL;
365: }
366: rlen = recv(TASK_FD(task), buf, len, 0);
367: if (rlen == -1) {
368: /* close connection */
369: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
370: TASK_ARG(task), 0, NULL, 0);
371: return NULL;
372: }
373: }
374: len = estlen;
375:
376: #if 0
377: /* check integrity of packet */
378: crc = ntohs(rpc->call_crc);
379: rpc->call_crc ^= rpc->call_crc;
380: if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
381: rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
382: return NULL;
383: }
384: #endif
385:
386: noreply = RPC_CHK_NOREPLY(rpc);
387:
388: /* check RPC packet session info */
389: if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
390: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
391:
392: rpc->call_argc ^= rpc->call_argc;
393: rpc->call_rep.ret = RPC_ERROR(-1);
394: rpc->call_rep.eno = RPC_ERROR(errno);
395: } else {
396: /* execute RPC call */
397: schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len);
398: }
399:
400: /* send RPC reply */
401: if (!noreply)
402: schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET],
403: TASK_ARG(task), TASK_FD(task), rpc, len);
404:
405: /* lets get next packet */
406: schedReadSelf(task);
407: return NULL;
408: }
409:
410: static void *
411: acceptClients(sched_task_t *task)
412: {
413: rpc_srv_t *srv = TASK_ARG(task);
414: rpc_cli_t *c = NULL;
415: socklen_t salen = sizeof(sockaddr_t);
416: int sock;
417: #ifdef TCP_SESSION_TIMEOUT
418: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
419: #endif
420:
421: c = _allocClient(srv, NULL);
422: if (!c) {
423: EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
424: if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
425: shutdown(sock, SHUT_RDWR);
426: close(sock);
427: }
428: goto end;
429: }
430:
431: /* accept client */
432: c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
433: if (c->cli_sock == -1) {
434: LOGERR;
435: AIT_FREE_VAL(&c->cli_buf);
436: array_Del(srv->srv_clients, c->cli_id, 42);
437: goto end;
438: } else
439: fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
440:
441: #ifdef TCP_SESSION_TIMEOUT
442: /* armed timer for close stateless connection */
443: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
444: schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], c,
445: ts, c, 0);
446: #endif
447: schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c,
448: c->cli_sock, NULL, 0);
449: end:
450: schedReadSelf(task);
451: return NULL;
452: }
453:
454:
455: static void *
456: txUDPPacket(sched_task_t *task)
457: {
458: rpc_cli_t *c = TASK_ARG(task);
459: rpc_srv_t *s = c->cli_parent;
460: rpc_func_t *f = NULL;
461: u_char *buf = AIT_GET_BUF(&c->cli_buf);
462: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
463: int ret, estlen, wlen = sizeof(struct tagRPCCall);
464: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
465: struct pollfd pfd;
466:
467: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
468: schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
469: TASK_ARG(task), ts, TASK_ARG(task), 0);
470:
471: if (rpc->call_argc) {
472: f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
473: if (!f) {
474: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
475: rpc->call_argc ^= rpc->call_argc;
476: rpc->call_rep.ret = RPC_ERROR(-1);
477: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
478: } else {
479: /* calc estimated length */
480: estlen = ait_resideVars(RPC_RETVARS(c)) + wlen;
481: if (estlen > AIT_LEN(&c->cli_buf))
482: AIT_RE_BUF(&c->cli_buf, estlen);
483: buf = AIT_GET_BUF(&c->cli_buf);
484: rpc = (struct tagRPCCall*) buf;
485:
486: rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
487: /* Go Encapsulate variables */
488: ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
489: RPC_RETVARS(c));
490: /* Free return values */
491: ait_freeVars(&c->cli_vars);
492: if (ret == -1) {
493: rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
494: rpc->call_argc ^= rpc->call_argc;
495: rpc->call_rep.ret = RPC_ERROR(-1);
496: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
497: } else
498: wlen += ret;
499: }
500: }
501:
502: rpc->call_len = htonl(wlen);
503:
504: /* calculate CRC */
505: rpc->call_crc ^= rpc->call_crc;
506: rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
507:
508: /* send reply */
509: pfd.fd = TASK_FD(task);
510: pfd.events = POLLOUT;
511: for (; wlen > 0; wlen -= ret, buf += ret) {
512: if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 ||
513: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
514: if (ret)
515: LOGERR;
516: else
517: rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
518: /* close connection */
519: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
520: TASK_ARG(task), 0, NULL, 0);
521: return NULL;
522: }
523: ret = sendto(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), MSG_NOSIGNAL,
524: &c->cli_sa.sa, c->cli_sa.sa.sa_len);
525: if (ret == -1) {
526: /* close connection */
527: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
528: TASK_ARG(task), 0, NULL, 0);
529: return NULL;
530: }
531: }
532:
533: return NULL;
534: }
535:
536: static void *
537: rxUDPPacket(sched_task_t *task)
538: {
539: rpc_srv_t *srv = TASK_ARG(task);
540: rpc_cli_t *c = NULL;
541: int len, rlen, noreply, estlen;
542: u_short crc;
543: u_char *buf, b[sizeof(struct tagRPCCall)];
544: struct tagRPCCall *rpc = (struct tagRPCCall*) b;
545: sockaddr_t sa;
546: socklen_t salen;
547: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
548: struct pollfd pfd;
549:
550: /* receive connect packet */
551: salen = sa.ss.ss_len = sizeof(sockaddr_t);
552: rlen = recvfrom(TASK_FD(task), b, sizeof b, MSG_PEEK, &sa.sa, &salen);
553: if (rlen < sizeof(struct tagRPCCall)) {
554: rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
555: goto end;
556: }
557:
558: c = _allocClient(srv, &sa);
559: if (!c) {
560: EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
561: usleep(2000); /* blocked client delay */
562: goto end;
563: } else {
564: estlen = ntohl(rpc->call_len);
565: if (estlen > AIT_LEN(&c->cli_buf))
566: AIT_RE_BUF(&c->cli_buf, estlen);
567: rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
568: buf = AIT_GET_BUF(&c->cli_buf);
569: len = estlen;
570:
571: c->cli_sock = TASK_FD(task);
572: memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);
573:
574: /* armed timer for close stateless connection */
575: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
576: schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
577: c, ts, c, 0);
578: }
579:
580: /* get next part of packet */
581: memset(buf, 0, len);
582: pfd.fd = TASK_FD(task);
583: pfd.events = POLLIN | POLLPRI;
584: for (; len > 0; len -= rlen, buf += rlen) {
585: if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 ||
586: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
587: if (rlen)
588: LOGERR;
589: else
590: rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
591: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
592: c, 0, NULL, 0);
593: goto end;
594: }
595: salen = sa.ss.ss_len = sizeof(sockaddr_t);
596: rlen = recvfrom(TASK_FD(task), buf, len, 0, &sa.sa, &salen);
597: if (rlen == -1) {
598: /* close connection */
599: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
600: c, 0, NULL, 0);
601: goto end;
602: }
603: if (e_addrcmp(&c->cli_sa, &sa, 42))
604: rlen ^= rlen; /* skip if arrive from different address */
605: }
606: len = estlen;
607:
608: /* check integrity of packet */
609: crc = ntohs(rpc->call_crc);
610: rpc->call_crc ^= rpc->call_crc;
611: if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
612: rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
613: /* close connection */
614: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
615: c, 0, NULL, 0);
616: goto end;
617: }
618:
619: noreply = RPC_CHK_NOREPLY(rpc);
620:
621: /* check RPC packet session info */
622: if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
623: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
624:
625: rpc->call_argc ^= rpc->call_argc;
626: rpc->call_rep.ret = RPC_ERROR(-1);
627: rpc->call_rep.eno = RPC_ERROR(errno);
628: } else {
629: /* execute RPC call */
630: schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
631: }
632:
633: /* send RPC reply */
634: if (!noreply)
635: schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
636: c, TASK_FD(task), rpc, len);
637: end:
638: schedReadSelf(task);
639: return NULL;
640: }
641:
642:
643: static void *
644: txBPFPacket(sched_task_t *task)
645: {
646: rpc_cli_t *c = TASK_ARG(task);
647: rpc_srv_t *s = c->cli_parent;
648: rpc_func_t *f = NULL;
649: u_char *buf = AIT_GET_BUF(&c->cli_buf);
650: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
651: int ret, len, wlen = sizeof(struct tagRPCCall);
652: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
653: struct ether_header *eh;
654: ait_val_t b = AIT_VAL_INIT;
655:
656: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
657: schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
658: TASK_ARG(task), ts, TASK_ARG(task), 0);
659:
660: if (rpc->call_argc) {
661: f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
662: if (!f) {
663: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
664: rpc->call_argc ^= rpc->call_argc;
665: rpc->call_rep.ret = RPC_ERROR(-1);
666: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
667: } else {
668: /* calc estimated length */
669: len = ait_resideVars(RPC_RETVARS(c)) + wlen;
670: if (len > AIT_LEN(&c->cli_buf))
671: AIT_RE_BUF(&c->cli_buf, len);
672: buf = AIT_GET_BUF(&c->cli_buf);
673: rpc = (struct tagRPCCall*) buf;
674:
675: rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
676: /* Go Encapsulate variables */
677: ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
678: RPC_RETVARS(c));
679: /* Free return values */
680: ait_freeVars(&RPC_RETVARS(c));
681: if (ret == -1) {
682: rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
683: rpc->call_argc ^= rpc->call_argc;
684: rpc->call_rep.ret = RPC_ERROR(-1);
685: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
686: } else
687: wlen += ret;
688: }
689: }
690:
691: rpc->call_len = htonl(wlen);
692:
693: /* calculate CRC */
694: rpc->call_crc ^= rpc->call_crc;
695: rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
696:
697: /* send reply */
698: AIT_SET_BUF(&b, NULL, MIN(wlen, s->srv_netbuf) + ETHER_HDR_LEN);
699: eh = (struct ether_header*) AIT_GET_BUF(&b);
700: memcpy(eh->ether_dhost, LLADDR(&c->cli_sa.sdl), ETHER_ADDR_LEN);
701: eh->ether_type = htons(RPC_DEFPORT);
702: memcpy(eh + 1, buf, MIN(wlen, s->srv_netbuf));
703:
704: ret = write(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
705: AIT_FREE_VAL(&b);
706: if (ret == -1) {
707: /* close connection */
708: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
709: TASK_ARG(task), 0, NULL, 0);
710: return NULL;
711: }
712:
713: return NULL;
714: }
715:
716: static void *
717: rxBPFPacket(sched_task_t *task)
718: {
719: rpc_srv_t *srv = TASK_ARG(task);
720: rpc_cli_t *c = NULL;
721: int len, rlen, noreply;
722: u_short crc;
723: struct tagRPCCall *rpc;
724: sockaddr_t sa;
725: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
726: struct bpf_hdr *h;
727: struct ether_header *eh;
728: ait_val_t b = AIT_VAL_INIT;
729:
730: /* receive connect packet */
731: AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
732: rlen = read(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
733: h = (struct bpf_hdr*) AIT_GET_BUF(&b);
734: rlen -= h->bh_hdrlen;
735: if (rlen < h->bh_datalen || h->bh_caplen != h->bh_datalen ||
736: rlen < ETHER_HDR_LEN + sizeof(struct tagRPCCall)) {
737: rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
738: goto end;
739: } else {
740: rlen = h->bh_caplen;
741: eh = (struct ether_header*) (AIT_GET_BUF(&b) + h->bh_hdrlen);
742: rlen -= ETHER_HDR_LEN;
743: rpc = (struct tagRPCCall*) (eh + 1);
744: if (eh->ether_type != ntohs(RPC_DEFPORT))
745: goto end;
746: else
747: e_getlinkbymac((const ether_addr_t*) eh->ether_shost, &sa);
748: }
749:
750: c = _allocClient(srv, &sa);
751: if (!c) {
752: EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
753: usleep(2000); /* blocked client delay */
754: goto end;
755: } else {
756: len = ntohl(rpc->call_len);
757: if (len > AIT_LEN(&c->cli_buf))
758: AIT_RE_BUF(&c->cli_buf, len);
759: memcpy(AIT_GET_BUF(&c->cli_buf), rpc, AIT_LEN(&c->cli_buf));
760: rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
761:
762: c->cli_sock = TASK_FD(task);
763: memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);
764:
765: /* armed timer for close stateless connection */
766: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
767: schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
768: c, ts, c, 0);
769: }
770:
771: /* check integrity of packet */
772: crc = ntohs(rpc->call_crc);
773: rpc->call_crc ^= rpc->call_crc;
774: if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
775: rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
776: /* close connection */
777: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
778: c, 0, NULL, 0);
779: goto end;
780: }
781:
782: noreply = RPC_CHK_NOREPLY(rpc);
783:
784: /* check RPC packet session info */
785: if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
786: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
787:
788: rpc->call_argc ^= rpc->call_argc;
789: rpc->call_rep.ret = RPC_ERROR(-1);
790: rpc->call_rep.eno = RPC_ERROR(errno);
791: } else {
792: /* execute RPC call */
793: schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
794: }
795:
796: /* send RPC reply */
797: if (!noreply)
798: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
799: c, TASK_FD(task), rpc, len);
800: end:
801: AIT_FREE_VAL(&b);
802: schedReadSelf(task);
803: return NULL;
804: }
805:
806: /* ------------------------------------------------------ */
807:
808: void
809: rpc_freeBLOBCli(rpc_cli_t * __restrict c)
810: {
811: rpc_srv_t *s = c->cli_parent;
812:
813: schedCancelby(s->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
814:
815: /* free buffer */
816: AIT_FREE_VAL(&c->cli_buf);
817:
818: array_Del(s->srv_blob.clients, c->cli_id, 0);
819: if (c)
820: e_free(c);
821: }
822:
823:
824: static void *
825: closeBLOBClient(sched_task_t *task)
826: {
827: int sock = ((rpc_cli_t*) TASK_ARG(task))->cli_sock;
828:
829: rpc_freeBLOBCli(TASK_ARG(task));
830:
831: /* close client socket */
832: shutdown(sock, SHUT_RDWR);
833: close(sock);
834: return NULL;
835: }
836:
837: static void *
838: txBLOB(sched_task_t *task)
839: {
840: rpc_cli_t *c = TASK_ARG(task);
841: u_char *buf = AIT_GET_BUF(&c->cli_buf);
842: int wlen = sizeof(struct tagBLOBHdr);
843:
844: /* send reply */
845: wlen = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);
846: if (wlen == -1 || wlen != sizeof(struct tagBLOBHdr)) {
847: /* close blob connection */
848: schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
849: }
850:
851: return NULL;
852: }
853:
854: static void *
855: rxBLOB(sched_task_t *task)
856: {
857: rpc_cli_t *c = TASK_ARG(task);
858: rpc_srv_t *s = c->cli_parent;
859: rpc_blob_t *b;
860: struct tagBLOBHdr blob;
861: int rlen;
862:
863: memset(&blob, 0, sizeof blob);
864: rlen = recv(TASK_FD(task), &blob, sizeof blob, 0);
865: if (rlen < 1) {
866: /* close blob connection */
867: schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
868: return NULL;
869: }
870:
871: /* check BLOB packet */
872: if (rlen < sizeof(struct tagBLOBHdr)) {
873: rpc_SetErr(ERPCMISMATCH, "Short BLOB packet");
874:
875: schedReadSelf(task);
876: return NULL;
877: }
878:
879: /* check RPC packet session info */
880: if (rpc_chkPktSession(&blob.hdr_session, &s->srv_session)) {
881: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
882: blob.hdr_cmd = error;
883: goto end;
884: }
885:
886: /* Go to proceed packet ... */
887: switch (blob.hdr_cmd) {
888: case get:
889: if (!(b = rpc_srv_getBLOB(s, ntohl(blob.hdr_var)))) {
890: rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob.hdr_var));
891: blob.hdr_cmd = no;
892: blob.hdr_ret = RPC_ERROR(-1);
893: break;
894: } else
895: blob.hdr_len = htonl(b->blob_len);
896:
897: if (rpc_srv_blobMap(s, b) != -1) {
898: /* deliver BLOB variable to client */
899: blob.hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
900: rpc_srv_blobUnmap(b);
901: } else {
902: blob.hdr_cmd = error;
903: blob.hdr_ret = RPC_ERROR(-1);
904: }
905: break;
906: case set:
907: if ((b = rpc_srv_registerBLOB(s, ntohl(blob.hdr_len),
908: ntohl(blob.hdr_ret)))) {
909: /* set new BLOB variable for reply :) */
910: blob.hdr_var = htonl(b->blob_var);
911:
912: /* receive BLOB from client */
913: blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
914: rpc_srv_blobUnmap(b);
915: } else {
916: blob.hdr_cmd = error;
917: blob.hdr_ret = RPC_ERROR(-1);
918: }
919: break;
920: case unset:
921: if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) {
922: blob.hdr_cmd = error;
923: blob.hdr_ret = RPC_ERROR(-1);
924: }
925: break;
926: default:
927: rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);
928: blob.hdr_cmd = error;
929: blob.hdr_ret = RPC_ERROR(-1);
930: }
931:
932: end:
933: memcpy(AIT_ADDR(&c->cli_buf), &blob, sizeof blob);
934: schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), NULL, 0);
935: schedReadSelf(task);
936: return NULL;
937: }
938:
939: static void *
940: flushBLOB(sched_task_t *task)
941: {
942: uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg);
943: rpc_srv_t *srv = sigArg ? (void*) sigArg : TASK_ARG(task);
944: rpc_blob_t *b, *tmp;
945:
946: TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
947: TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);
948:
949: rpc_srv_blobFree(srv, b);
950: e_free(b);
951: }
952:
953: if (!schedSignalSelf(task)) {
954: /* disabled kqueue support in libaitsched */
955: struct sigaction sa;
956:
957: memset(&sa, 0, sizeof sa);
958: sigemptyset(&sa.sa_mask);
959: sa.sa_handler = (void (*)(int)) flushBLOB;
960: sa.sa_flags = SA_RESTART | SA_RESETHAND;
961: sigaction(SIGFBLOB, &sa, NULL);
962: }
963:
964: return NULL;
965: }
966:
967: static void *
968: acceptBLOBClients(sched_task_t *task)
969: {
970: rpc_srv_t *srv = TASK_ARG(task);
971: rpc_cli_t *c = NULL;
972: register int i;
973: socklen_t salen = sizeof(sockaddr_t);
974: int sock;
975: #ifdef TCP_NOPUSH
976: int n = 1;
977: #endif
978:
979: /* check free slots for connect */
980: for (i = 0; i < array_Size(srv->srv_blob.clients) &&
981: (c = array(srv->srv_blob.clients, i, rpc_cli_t*)); i++);
982: if (c) { /* no more free slots! */
983: EVERBOSE(1, "BLOB client quota exceeded! Connection will be shutdown!\n");
984: if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
985: shutdown(sock, SHUT_RDWR);
986: close(sock);
987: }
988: goto end;
989: }
990:
991: c = e_malloc(sizeof(rpc_cli_t));
992: if (!c) {
993: LOGERR;
994: srv->srv_kill = srv->srv_blob.kill = 1;
995: return NULL;
996: } else {
997: memset(c, 0, sizeof(rpc_cli_t));
998: array_Set(srv->srv_blob.clients, i, c);
999: c->cli_id = i;
1000: c->cli_parent = srv;
1001: }
1002:
1003: /* alloc empty buffer */
1004: AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
1005:
1006: /* accept client */
1007: c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
1008: if (c->cli_sock == -1) {
1009: LOGERR;
1010: AIT_FREE_VAL(&c->cli_buf);
1011: array_Del(srv->srv_blob.clients, i, 42);
1012: goto end;
1013: } else {
1014: #ifdef TCP_NOPUSH
1015: setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n);
1016: #endif
1017: fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1018: }
1019:
1020: schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0);
1021: end:
1022: schedReadSelf(task);
1023: return NULL;
1024: }
1025:
1026: /* ------------------------------------------------------ */
1027:
1028: /*
1029: * rpc_srv_initBLOBServer() - Init & create BLOB Server
1030: *
1031: * @srv = RPC server instance
1032: * @Port = Port for bind server, if Port == 0 default port is selected
1033: * @diskDir = Disk place for BLOB file objects
1034: * return: -1 == error or 0 bind and created BLOB server instance
1035: */
1036: int
1037: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
1038: {
1039: int n = 1;
1040:
1041: if (!srv || srv->srv_kill) {
1042: rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
1043: return -1;
1044: }
1045:
1046: memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
1047: if (access(diskDir, R_OK | W_OK) == -1) {
1048: LOGERR;
1049: return -1;
1050: } else
1051: AIT_SET_STR(&srv->srv_blob.dir, diskDir);
1052:
1053: /* init blob list */
1054: TAILQ_INIT(&srv->srv_blob.blobs);
1055:
1056: srv->srv_blob.server.cli_parent = srv;
1057:
1058: memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(sockaddr_t));
1059: switch (srv->srv_blob.server.cli_sa.sa.sa_family) {
1060: case AF_INET:
1061: srv->srv_blob.server.cli_sa.sin.sin_port =
1062: htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin.sin_port) + 1);
1063: break;
1064: case AF_INET6:
1065: srv->srv_blob.server.cli_sa.sin6.sin6_port =
1066: htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin6.sin6_port) + 1);
1067: break;
1068: case AF_LOCAL:
1069: strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob",
1070: sizeof srv->srv_blob.server.cli_sa.sun.sun_path);
1071: break;
1072: default:
1073: AIT_FREE_VAL(&srv->srv_blob.dir);
1074: return -1;
1075: }
1076:
1077: /* create BLOB server socket */
1078: srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
1079: if (srv->srv_blob.server.cli_sock == -1) {
1080: LOGERR;
1081: AIT_FREE_VAL(&srv->srv_blob.dir);
1082: return -1;
1083: }
1084: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
1085: LOGERR;
1086: close(srv->srv_blob.server.cli_sock);
1087: AIT_FREE_VAL(&srv->srv_blob.dir);
1088: return -1;
1089: }
1090: n = srv->srv_netbuf;
1091: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
1092: LOGERR;
1093: close(srv->srv_blob.server.cli_sock);
1094: AIT_FREE_VAL(&srv->srv_blob.dir);
1095: return -1;
1096: }
1097: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
1098: LOGERR;
1099: close(srv->srv_blob.server.cli_sock);
1100: AIT_FREE_VAL(&srv->srv_blob.dir);
1101: return -1;
1102: }
1103: if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa,
1104: srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
1105: LOGERR;
1106: close(srv->srv_blob.server.cli_sock);
1107: AIT_FREE_VAL(&srv->srv_blob.dir);
1108: return -1;
1109: } else
1110: fcntl(srv->srv_blob.server.cli_sock, F_SETFL,
1111: fcntl(srv->srv_blob.server.cli_sock, F_GETFL) | O_NONBLOCK);
1112:
1113:
1114: /* allocate pool for concurent blob clients */
1115: srv->srv_blob.clients = array_Init(array_Size(srv->srv_clients));
1116: if (!srv->srv_blob.clients) {
1117: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1118: close(srv->srv_blob.server.cli_sock);
1119: AIT_FREE_VAL(&srv->srv_blob.dir);
1120: return -1;
1121: }
1122:
1123: /* init blob scheduler */
1124: srv->srv_blob.root = schedBegin();
1125: if (!srv->srv_blob.root) {
1126: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1127: array_Destroy(&srv->srv_blob.clients);
1128: close(srv->srv_blob.server.cli_sock);
1129: AIT_FREE_VAL(&srv->srv_blob.dir);
1130: return -1;
1131: }
1132:
1133: return 0;
1134: }
1135:
1136: /*
1137: * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
1138: *
1139: * @srv = RPC Server instance
1140: * return: none
1141: */
1142: void
1143: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
1144: {
1145: if (!srv)
1146: return;
1147:
1148: srv->srv_blob.kill = 1;
1149:
1150: schedEnd(&srv->srv_blob.root);
1151: }
1152:
1153: /*
1154: * rpc_srv_loopBLOBServer() - Execute Main BLOB server loop and wait for clients requests
1155: *
1156: * @srv = RPC Server instance
1157: * return: -1 error or 0 ok, infinite loop ...
1158: */
1159: int
1160: rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
1161: {
1162: rpc_cli_t *c;
1163: register int i;
1164: rpc_blob_t *b, *tmp;
1165: struct timespec ts = { RPC_SCHED_POLLING, 0 };
1166:
1167: if (!srv || srv->srv_kill) {
1168: rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
1169: return -1;
1170: }
1171:
1172: if (listen(srv->srv_blob.server.cli_sock, array_Size(srv->srv_blob.clients)) == -1) {
1173: LOGERR;
1174: return -1;
1175: }
1176:
1177: if (!schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0)) {
1178: /* disabled kqueue support in libaitsched */
1179: struct sigaction sa;
1180:
1181: atomic_store_rel_ptr(&_glSigArg, (uintptr_t) srv);
1182:
1183: memset(&sa, 0, sizeof sa);
1184: sigemptyset(&sa.sa_mask);
1185: sa.sa_handler = (void (*)(int)) flushBLOB;
1186: sa.sa_flags = SA_RESTART | SA_RESETHAND;
1187: sigaction(SIGFBLOB, &sa, NULL);
1188: }
1189:
1190: if (!schedRead(srv->srv_blob.root, acceptBLOBClients, srv,
1191: srv->srv_blob.server.cli_sock, NULL, 0)) {
1192: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1193: return -1;
1194: }
1195:
1196: schedPolling(srv->srv_blob.root, &ts, NULL);
1197: /* main rpc loop */
1198: schedRun(srv->srv_blob.root, &srv->srv_blob.kill);
1199:
1200: /* detach blobs */
1201: TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
1202: TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);
1203:
1204: rpc_srv_blobFree(srv, b);
1205: e_free(b);
1206: }
1207:
1208: /* close all clients connections & server socket */
1209: for (i = 0; i < array_Size(srv->srv_blob.clients); i++) {
1210: c = array(srv->srv_blob.clients, i, rpc_cli_t*);
1211: if (c) {
1212: shutdown(c->cli_sock, SHUT_RDWR);
1213: close(c->cli_sock);
1214:
1215: schedCancelby(srv->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
1216: AIT_FREE_VAL(&c->cli_buf);
1217: }
1218: array_Del(srv->srv_blob.clients, i, 42);
1219: }
1220: array_Destroy(&srv->srv_blob.clients);
1221:
1222: close(srv->srv_blob.server.cli_sock);
1223:
1224: AIT_FREE_VAL(&srv->srv_blob.dir);
1225: return 0;
1226: }
1227:
1228:
1229: /*
1230: * rpc_srv_initServer() - Init & create RPC Server
1231: *
1232: * @InstID = Instance for authentication & recognition
1233: * @concurentClients = Concurent clients at same time to this server
1234: * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
1235: * @csHost = Host name or address for bind server, if NULL any address
1236: * @Port = Port for bind server, if Port == 0 default port is selected
1237: * @proto = Protocol, if == 0 choose SOCK_STREAM
1238: * return: NULL == error or !=NULL bind and created RPC server instance
1239: */
1240: rpc_srv_t *
1241: rpc_srv_initServer(u_char InstID, int concurentClients, int netBuf,
1242: const char *csHost, u_short Port, int proto)
1243: {
1244: int n = 1;
1245: rpc_srv_t *srv = NULL;
1246: sockaddr_t sa = E_SOCKADDR_INIT;
1247:
1248: if (!concurentClients || (proto < 0 || proto > SOCK_RAW)) {
1249: rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1250: return NULL;
1251: }
1252: if (!e_gethostbyname(csHost, Port, &sa))
1253: return NULL;
1254: if (!Port)
1255: Port = RPC_DEFPORT;
1256: if (!proto)
1257: proto = SOCK_STREAM;
1258: if (netBuf < RPC_MIN_BUFSIZ)
1259: netBuf = BUFSIZ;
1260: else
1261: netBuf = E_ALIGN(netBuf, 2); /* align netBuf length */
1262:
1263: #ifdef HAVE_SRANDOMDEV
1264: srandomdev();
1265: #else
1266: time_t tim;
1267:
1268: srandom((time(&tim) ^ getpid()));
1269: #endif
1270:
1271: srv = e_malloc(sizeof(rpc_srv_t));
1272: if (!srv) {
1273: LOGERR;
1274: return NULL;
1275: } else
1276: memset(srv, 0, sizeof(rpc_srv_t));
1277:
1278: srv->srv_proto = proto;
1279: srv->srv_netbuf = netBuf;
1280: srv->srv_session.sess_version = RPC_VERSION;
1281: srv->srv_session.sess_instance = InstID;
1282:
1283: srv->srv_server.cli_parent = srv;
1284: memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
1285:
1286: /* init functions */
1287: pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
1288: SLIST_INIT(&srv->srv_funcs);
1289: AVL_INIT(&srv->srv_funcs);
1290:
1291: /* init scheduler */
1292: srv->srv_root = schedBegin();
1293: if (!srv->srv_root) {
1294: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1295: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1296: e_free(srv);
1297: return NULL;
1298: }
1299:
1300: /* init pool for clients */
1301: srv->srv_clients = array_Init(concurentClients);
1302: if (!srv->srv_clients) {
1303: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1304: schedEnd(&srv->srv_root);
1305: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1306: e_free(srv);
1307: return NULL;
1308: }
1309:
1310: /* create server socket */
1311: srv->srv_server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, srv->srv_proto, 0);
1312: if (srv->srv_server.cli_sock == -1) {
1313: LOGERR;
1314: array_Destroy(&srv->srv_clients);
1315: schedEnd(&srv->srv_root);
1316: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1317: e_free(srv);
1318: return NULL;
1319: }
1320: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
1321: LOGERR;
1322: goto err;
1323: }
1324: n = srv->srv_netbuf;
1325: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
1326: LOGERR;
1327: goto err;
1328: }
1329: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
1330: LOGERR;
1331: goto err;
1332: }
1333: if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa,
1334: srv->srv_server.cli_sa.sa.sa_len) == -1) {
1335: LOGERR;
1336: goto err;
1337: } else
1338: fcntl(srv->srv_server.cli_sock, F_SETFL,
1339: fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
1340:
1341: rpc_register_srvPing(srv);
1342:
1343: return srv;
1344: err: /* error condition */
1345: close(srv->srv_server.cli_sock);
1346: array_Destroy(&srv->srv_clients);
1347: schedEnd(&srv->srv_root);
1348: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1349: e_free(srv);
1350: return NULL;
1351: }
1352:
1353: /*
1354: * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
1355: *
1356: * @psrv = RPC Server instance
1357: * return: none
1358: */
1359: void
1360: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
1361: {
1362: if (!psrv || !*psrv)
1363: return;
1364:
1365: /* if send kill to blob server */
1366: rpc_srv_endBLOBServer(*psrv);
1367:
1368: (*psrv)->srv_kill = 1;
1369: sleep(RPC_SCHED_POLLING);
1370:
1371: schedEnd(&(*psrv)->srv_root);
1372:
1373: pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx);
1374: e_free(*psrv);
1375: *psrv = NULL;
1376: }
1377:
1378: /*
1379: * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
1380: *
1381: * @srv = RPC Server instance
1382: * return: -1 error or 0 ok, infinite loop ...
1383: */
1384: int
1385: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
1386: {
1387: rpc_cli_t *c;
1388: register int i;
1389: rpc_func_t *f;
1390: struct timespec ts = { RPC_SCHED_POLLING, 0 };
1391:
1392: if (!srv) {
1393: rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
1394: return -1;
1395: }
1396:
1397: if (srv->srv_proto == SOCK_STREAM)
1398: if (listen(srv->srv_server.cli_sock, array_Size(srv->srv_clients)) == -1) {
1399: LOGERR;
1400: return -1;
1401: }
1402:
1403: if (!schedRead(srv->srv_root, cbProto[srv->srv_proto][CB_ACCEPTCLIENT], srv,
1404: srv->srv_server.cli_sock, NULL, 0)) {
1405: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1406: return -1;
1407: }
1408:
1409: schedPolling(srv->srv_root, &ts, NULL);
1410: /* main rpc loop */
1411: schedRun(srv->srv_root, &srv->srv_kill);
1412:
1413: /* close all clients connections & server socket */
1414: for (i = 0; i < array_Size(srv->srv_clients); i++) {
1415: c = array(srv->srv_clients, i, rpc_cli_t*);
1416: if (c) {
1417: if (srv->srv_proto == SOCK_STREAM) {
1418: shutdown(c->cli_sock, SHUT_RDWR);
1419: close(c->cli_sock);
1420: }
1421:
1422: schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
1423: ait_freeVars(&RPC_RETVARS(c));
1424: AIT_FREE_VAL(&c->cli_buf);
1425: }
1426: array_Del(srv->srv_clients, i, 42);
1427: }
1428: array_Destroy(&srv->srv_clients);
1429:
1430: close(srv->srv_server.cli_sock);
1431:
1432: /* detach exported calls */
1433: RPC_FUNCS_LOCK(&srv->srv_funcs);
1434: while ((f = SLIST_FIRST(&srv->srv_funcs))) {
1435: SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);
1436:
1437: AIT_FREE_VAL(&f->func_name);
1438: e_free(f);
1439: }
1440: srv->srv_funcs.avlh_root = NULL;
1441: RPC_FUNCS_UNLOCK(&srv->srv_funcs);
1442:
1443: return 0;
1444: }
1445:
1446:
1447: /*
1448: * rpc_srv_execCall() Execute registered call from RPC server
1449: *
1450: * @cli = RPC client
1451: * @rpc = IN RPC call structure
1452: * @funcname = Execute RPC function
1453: * @args = IN RPC calling arguments from RPC client
1454: * return: -1 error, !=-1 ok
1455: */
1456: int
1457: rpc_srv_execCall(rpc_cli_t * __restrict cli, struct tagRPCCall * __restrict rpc,
1458: ait_val_t funcname, array_t * __restrict args)
1459: {
1460: rpc_callback_t func;
1461:
1462: if (!cli || !rpc || !AIT_ADDR(&funcname)) {
1463: rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1464: return -1;
1465: }
1466:
1467: func = AIT_GET_LIKE(&funcname, rpc_callback_t);
1468: return func(cli, rpc, args);
1469: }
1470:
1471:
1472: /*
1473: * rpc_srv_initServer2() - Init & create layer2 RPC Server
1474: *
1475: * @InstID = Instance for authentication & recognition
1476: * @concurentClients = Concurent clients at same time to this server
1477: * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
1478: * @csIface = Interface name for bind server, if NULL first interface on host
1479: * return: NULL == error or !=NULL bind and created RPC server instance
1480: */
1481: rpc_srv_t *
1482: rpc_srv_initServer2(u_char InstID, int concurentClients, int netBuf, const char *csIface)
1483: {
1484: int n = 1;
1485: rpc_srv_t *srv = NULL;
1486: sockaddr_t sa = E_SOCKADDR_INIT;
1487: char szIface[64], szStr[STRSIZ];
1488: register int i;
1489: struct ifreq ifr;
1490: struct bpf_insn insns[] = {
1491: BPF_STMT(BPF_LD + BPF_H + BPF_ABS, 12),
1492: BPF_JUMP(BPF_JMP + BPF_JEQ + BPF_K, RPC_DEFPORT, 0, 1),
1493: BPF_STMT(BPF_RET + BPF_K, -1),
1494: BPF_STMT(BPF_RET + BPF_K, 0),
1495: };
1496: struct bpf_program fcode = {
1497: .bf_len = sizeof(insns) / sizeof(struct bpf_insn),
1498: .bf_insns = insns
1499: };
1500:
1501: if (!concurentClients) {
1502: rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1503: return NULL;
1504: }
1505: if (!csIface) {
1506: if (e_get1stiface(szIface, sizeof szIface))
1507: return NULL;
1508: } else
1509: strlcpy(szIface, csIface, sizeof szIface);
1510: if (!e_getifacebyname(szIface, &sa))
1511: return NULL;
1512:
1513: #ifdef HAVE_SRANDOMDEV
1514: srandomdev();
1515: #else
1516: time_t tim;
1517:
1518: srandom((time(&tim) ^ getpid()));
1519: #endif
1520:
1521: srv = e_malloc(sizeof(rpc_srv_t));
1522: if (!srv) {
1523: LOGERR;
1524: return NULL;
1525: } else
1526: memset(srv, 0, sizeof(rpc_srv_t));
1527:
1528: srv->srv_proto = SOCK_BPF;
1529: srv->srv_netbuf = netBuf;
1530: srv->srv_session.sess_version = RPC_VERSION;
1531: srv->srv_session.sess_instance = InstID;
1532:
1533: srv->srv_server.cli_parent = srv;
1534: memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
1535:
1536: /* init functions */
1537: pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
1538: SLIST_INIT(&srv->srv_funcs);
1539: AVL_INIT(&srv->srv_funcs);
1540:
1541: /* init scheduler */
1542: srv->srv_root = schedBegin();
1543: if (!srv->srv_root) {
1544: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1545: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1546: e_free(srv);
1547: return NULL;
1548: }
1549:
1550: /* init pool for clients */
1551: srv->srv_clients = array_Init(concurentClients);
1552: if (!srv->srv_clients) {
1553: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1554: schedEnd(&srv->srv_root);
1555: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1556: e_free(srv);
1557: return NULL;
1558: }
1559:
1560: /* create server handler */
1561: for (i = 0; i < 10; i++) {
1562: memset(szStr, 0, sizeof szStr);
1563: snprintf(szStr, sizeof szStr, "/dev/bpf%d", i);
1564: srv->srv_server.cli_sock = open(szStr, O_RDWR);
1565: if (srv->srv_server.cli_sock > STDERR_FILENO)
1566: break;
1567: }
1568: if (srv->srv_server.cli_sock < 3) {
1569: LOGERR;
1570: array_Destroy(&srv->srv_clients);
1571: schedEnd(&srv->srv_root);
1572: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1573: e_free(srv);
1574: return NULL;
1575: }
1576:
1577: if (ioctl(srv->srv_server.cli_sock, BIOCIMMEDIATE, &n) == -1) {
1578: LOGERR;
1579: goto err;
1580: }
1581: if (ioctl(srv->srv_server.cli_sock, BIOCSETF, &fcode) == -1) {
1582: LOGERR;
1583: goto err;
1584: }
1585: n = (netBuf < RPC_MIN_BUFSIZ) ? getpagesize() : E_ALIGN(netBuf, 2);
1586: if (ioctl(srv->srv_server.cli_sock, BIOCSBLEN, &n) == -1) {
1587: LOGERR;
1588: goto err;
1589: } else
1590: srv->srv_netbuf = n;
1591:
1592: memset(&ifr, 0, sizeof ifr);
1593: strlcpy(ifr.ifr_name, szIface, sizeof ifr.ifr_name);
1594: if (ioctl(srv->srv_server.cli_sock, BIOCSETIF, &ifr) == -1) {
1595: LOGERR;
1596: goto err;
1597: } else
1598: fcntl(srv->srv_server.cli_sock, F_SETFL,
1599: fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
1600:
1601: rpc_register_srvPing(srv);
1602:
1603: return srv;
1604: err: /* error condition */
1605: close(srv->srv_server.cli_sock);
1606: array_Destroy(&srv->srv_clients);
1607: schedEnd(&srv->srv_root);
1608: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1609: e_free(srv);
1610: return NULL;
1611: }
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