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