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.4 2015/03/09 22:25:35 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: memmove(AIT_GET_BUF(&c->cli_buf), rpc, len);
823:
824: /* skip loop packet */
825: if (rpc->call_io & RPC_ACK)
826: goto end;
827:
828: /* check integrity of packet */
829: crc = ntohs(rpc->call_crc);
830: rpc->call_crc ^= rpc->call_crc;
831: if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
832: rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
833: /* close connection */
834: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
835: c, 0, NULL, 0);
836: goto end;
837: }
838:
839: noreply = RPC_CHK_NOREPLY(rpc);
840:
841: /* check RPC packet session info */
842: if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
843: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
844:
845: rpc->call_argc ^= rpc->call_argc;
846: rpc->call_rep.ret = RPC_ERROR(-1);
847: rpc->call_rep.eno = RPC_ERROR(errno);
848: } else {
849: /* execute RPC call */
850: schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
851: }
852:
853: /* send RPC reply */
854: if (!noreply)
855: schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
856: c, TASK_FD(task), rpc, len);
857: end:
858: schedReadSelf(task);
859: return NULL;
860: }
861:
862:
863: static void *
864: txBPFPacket(sched_task_t *task)
865: {
866: rpc_cli_t *c = TASK_ARG(task);
867: rpc_srv_t *s = c->cli_parent;
868: rpc_func_t *f = NULL;
869: u_char *buf = AIT_GET_BUF(&c->cli_buf);
870: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
871: int ret, len, wlen = sizeof(struct tagRPCCall);
872: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
873: struct ether_header *eh;
874: ait_val_t b = AIT_VAL_INIT;
875:
876: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
877: schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
878: TASK_ARG(task), ts, TASK_ARG(task), 0);
879:
880: if (rpc->call_argc) {
881: f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
882: if (!f) {
883: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
884: rpc->call_argc ^= rpc->call_argc;
885: rpc->call_rep.ret = RPC_ERROR(-1);
886: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
887: } else {
888: /* calc estimated length */
889: len = ait_resideVars(RPC_RETVARS(c)) + wlen;
890: if (len > AIT_LEN(&c->cli_buf))
891: AIT_RE_BUF(&c->cli_buf, len);
892: buf = AIT_GET_BUF(&c->cli_buf);
893: rpc = (struct tagRPCCall*) buf;
894:
895: rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
896: /* Go Encapsulate variables */
897: ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
898: RPC_RETVARS(c));
899: /* Free return values */
900: ait_freeVars(&RPC_RETVARS(c));
901: if (ret == -1) {
902: rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
903: rpc->call_argc ^= rpc->call_argc;
904: rpc->call_rep.ret = RPC_ERROR(-1);
905: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
906: } else
907: wlen += ret;
908: }
909: }
910:
911: rpc->call_len = htonl(wlen);
912: rpc->call_io = RPC_ACK;
913:
914: /* calculate CRC */
915: rpc->call_crc ^= rpc->call_crc;
916: rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
917:
918: /* send reply */
919: AIT_SET_BUF(&b, NULL, MIN(wlen, s->srv_netbuf) + ETHER_HDR_LEN);
920: eh = (struct ether_header*) AIT_GET_BUF(&b);
921: memcpy(eh->ether_dhost, LLADDR(&c->cli_sa.sdl), ETHER_ADDR_LEN);
922: eh->ether_type = htons(RPC_DEFPORT);
923: memcpy(eh + 1, buf, MIN(wlen, s->srv_netbuf));
924:
925: ret = write(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
926: AIT_FREE_VAL(&b);
927: if (ret == -1) {
928: /* close connection */
929: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
930: TASK_ARG(task), 0, NULL, 0);
931: return NULL;
932: }
933:
934: return NULL;
935: }
936:
937: static void *
938: rxBPFPacket(sched_task_t *task)
939: {
940: rpc_srv_t *srv = TASK_ARG(task);
941: rpc_cli_t *c = NULL;
942: int len, rlen, noreply;
943: u_short crc;
944: struct tagRPCCall *rpc;
945: sockaddr_t sa;
946: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
947: struct bpf_hdr *h;
948: struct ether_header *eh;
949: ait_val_t b = AIT_VAL_INIT;
950:
951: /* receive connect packet */
952: AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
953: rlen = read(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
954: h = (struct bpf_hdr*) AIT_GET_BUF(&b);
955: rlen -= h->bh_hdrlen;
956: if (rlen < h->bh_datalen || h->bh_caplen != h->bh_datalen ||
957: rlen < ETHER_HDR_LEN + sizeof(struct tagRPCCall)) {
958: rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
959: goto end;
960: } else {
961: rlen = h->bh_caplen;
962: eh = (struct ether_header*) (AIT_GET_BUF(&b) + h->bh_hdrlen);
963: rlen -= ETHER_HDR_LEN;
964: rpc = (struct tagRPCCall*) (eh + 1);
965:
966: #if 0
967: /* skip loop packet */
968: if (rpc->call_io & RPC_ACK)
969: goto end;
970: #endif
971:
972: if (eh->ether_type != ntohs(RPC_DEFPORT))
973: goto end;
974: else
975: e_getlinkbymac((const ether_addr_t*) eh->ether_shost, &sa);
976: }
977:
978: c = _allocClient(srv, &sa);
979: if (!c) {
980: EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
981: usleep(2000); /* blocked client delay */
982: goto end;
983: } else {
984: len = ntohl(rpc->call_len);
985: if (len > AIT_LEN(&c->cli_buf))
986: AIT_RE_BUF(&c->cli_buf, len);
987: memcpy(AIT_GET_BUF(&c->cli_buf), rpc, AIT_LEN(&c->cli_buf));
988: rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
989:
990: c->cli_sock = TASK_FD(task);
991: memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);
992:
993: /* armed timer for close stateless connection */
994: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
995: schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
996: c, ts, c, 0);
997: }
998:
999: /* check integrity of packet */
1000: crc = ntohs(rpc->call_crc);
1001: rpc->call_crc ^= rpc->call_crc;
1002: if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
1003: rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
1004: /* close connection */
1005: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
1006: c, 0, NULL, 0);
1007: goto end;
1008: }
1009:
1010: noreply = RPC_CHK_NOREPLY(rpc);
1011:
1012: /* check RPC packet session info */
1013: if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
1014: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1015:
1016: rpc->call_argc ^= rpc->call_argc;
1017: rpc->call_rep.ret = RPC_ERROR(-1);
1018: rpc->call_rep.eno = RPC_ERROR(errno);
1019: } else {
1020: /* execute RPC call */
1021: schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
1022: }
1023:
1024: /* send RPC reply */
1025: if (!noreply)
1026: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
1027: c, TASK_FD(task), rpc, len);
1028: end:
1029: AIT_FREE_VAL(&b);
1030: schedReadSelf(task);
1031: return NULL;
1032: }
1033:
1034:
1035: static void *
1036: txEXTPacket(sched_task_t *task)
1037: {
1038: rpc_cli_t *c = TASK_ARG(task);
1039: rpc_srv_t *s = c->cli_parent;
1040: rpc_func_t *f = NULL;
1041: u_char *buf = AIT_GET_BUF(&c->cli_buf);
1042: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1043: int ret, len, wlen = sizeof(struct tagRPCCall);
1044: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
1045:
1046: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
1047: schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
1048: TASK_ARG(task), ts, TASK_ARG(task), 0);
1049:
1050: if (rpc->call_argc) {
1051: f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
1052: if (!f) {
1053: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
1054: rpc->call_argc ^= rpc->call_argc;
1055: rpc->call_rep.ret = RPC_ERROR(-1);
1056: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
1057: } else {
1058: /* calc estimated length */
1059: len = ait_resideVars(RPC_RETVARS(c)) + wlen;
1060: if (len > AIT_LEN(&c->cli_buf))
1061: AIT_RE_BUF(&c->cli_buf, len);
1062: buf = AIT_GET_BUF(&c->cli_buf);
1063: rpc = (struct tagRPCCall*) buf;
1064:
1065: rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
1066: /* Go Encapsulate variables */
1067: ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
1068: RPC_RETVARS(c));
1069: /* Free return values */
1070: ait_freeVars(&RPC_RETVARS(c));
1071: if (ret == -1) {
1072: rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
1073: rpc->call_argc ^= rpc->call_argc;
1074: rpc->call_rep.ret = RPC_ERROR(-1);
1075: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
1076: } else
1077: wlen += ret;
1078: }
1079: }
1080:
1081: rpc->call_len = htonl(wlen);
1082: rpc->call_io = RPC_ACK;
1083:
1084: /* send reply */
1085: ret = write(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf));
1086: if (ret == -1) {
1087: /* close connection */
1088: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
1089: TASK_ARG(task), 0, NULL, 0);
1090: return NULL;
1091: }
1092:
1093: return NULL;
1094: }
1095:
1096: static void *
1097: rxEXTPacket(sched_task_t *task)
1098: {
1099: rpc_srv_t *srv = TASK_ARG(task);
1100: rpc_cli_t *c = NULL;
1101: int len, rlen, noreply;
1102: struct tagRPCCall *rpc;
1103: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
1104: ait_val_t b = AIT_VAL_INIT;
1105: sockaddr_t sa;
1106:
1107: memset(&sa, 0, sizeof sa);
1108: /* receive connect packet */
1109: AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
1110: rlen = read(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
1111: if (rlen < sizeof(struct tagRPCCall)) {
1112: rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
1113: goto end;
1114: } else {
1115: rpc = (struct tagRPCCall*) AIT_GET_BUF(&b);
1116:
1117: /* skip loop packet */
1118: if (rpc->call_io & RPC_ACK)
1119: goto end;
1120: }
1121:
1122: c = _allocClient(srv, &sa);
1123: if (!c) {
1124: EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
1125: usleep(2000); /* blocked client delay */
1126: goto end;
1127: } else {
1128: len = ntohl(rpc->call_len);
1129: if (len > AIT_LEN(&c->cli_buf))
1130: AIT_RE_BUF(&c->cli_buf, len);
1131: memcpy(AIT_GET_BUF(&c->cli_buf), rpc, AIT_LEN(&c->cli_buf));
1132: rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
1133:
1134: c->cli_sock = TASK_FD(task);
1135:
1136: /* armed timer for close stateless connection */
1137: schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
1138: schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
1139: c, ts, c, 0);
1140: }
1141:
1142: noreply = RPC_CHK_NOREPLY(rpc);
1143:
1144: /* check RPC packet session info */
1145: if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
1146: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1147:
1148: rpc->call_argc ^= rpc->call_argc;
1149: rpc->call_rep.ret = RPC_ERROR(-1);
1150: rpc->call_rep.eno = RPC_ERROR(errno);
1151: } else {
1152: /* execute RPC call */
1153: schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
1154: }
1155:
1156: /* send RPC reply */
1157: if (!noreply)
1158: schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
1159: c, TASK_FD(task), rpc, len);
1160: end:
1161: AIT_FREE_VAL(&b);
1162: schedReadSelf(task);
1163: return NULL;
1164: }
1165:
1166: /* ------------------------------------------------------ */
1167:
1168: void
1169: rpc_freeBLOBCli(rpc_cli_t * __restrict c)
1170: {
1171: rpc_srv_t *s = c->cli_parent;
1172:
1173: schedCancelby(s->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
1174:
1175: /* free buffer */
1176: AIT_FREE_VAL(&c->cli_buf);
1177:
1178: array_Del(s->srv_blob.clients, c->cli_id, 0);
1179: if (c)
1180: e_free(c);
1181: }
1182:
1183:
1184: static void *
1185: closeBLOBClient(sched_task_t *task)
1186: {
1187: int sock = ((rpc_cli_t*) TASK_ARG(task))->cli_sock;
1188:
1189: rpc_freeBLOBCli(TASK_ARG(task));
1190:
1191: /* close client socket */
1192: shutdown(sock, SHUT_RDWR);
1193: close(sock);
1194: return NULL;
1195: }
1196:
1197: static void *
1198: txBLOB(sched_task_t *task)
1199: {
1200: rpc_cli_t *c = TASK_ARG(task);
1201: u_char *buf = AIT_GET_BUF(&c->cli_buf);
1202: int wlen = sizeof(struct tagBLOBHdr);
1203:
1204: /* send reply */
1205: wlen = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);
1206: if (wlen == -1 || wlen != sizeof(struct tagBLOBHdr)) {
1207: /* close blob connection */
1208: schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
1209: }
1210:
1211: return NULL;
1212: }
1213:
1214: static void *
1215: rxBLOB(sched_task_t *task)
1216: {
1217: rpc_cli_t *c = TASK_ARG(task);
1218: rpc_srv_t *s = c->cli_parent;
1219: rpc_blob_t *b;
1220: struct tagBLOBHdr blob;
1221: int rlen;
1222:
1223: memset(&blob, 0, sizeof blob);
1224: rlen = recv(TASK_FD(task), &blob, sizeof blob, 0);
1225: if (rlen < 1) {
1226: /* close blob connection */
1227: schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
1228: return NULL;
1229: }
1230:
1231: /* check BLOB packet */
1232: if (rlen < sizeof(struct tagBLOBHdr)) {
1233: rpc_SetErr(ERPCMISMATCH, "Short BLOB packet");
1234:
1235: schedReadSelf(task);
1236: return NULL;
1237: }
1238:
1239: /* check RPC packet session info */
1240: if (rpc_chkPktSession(&blob.hdr_session, &s->srv_session)) {
1241: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1242: blob.hdr_cmd = error;
1243: goto end;
1244: }
1245:
1246: /* Go to proceed packet ... */
1247: switch (blob.hdr_cmd) {
1248: case get:
1249: if (!(b = rpc_srv_getBLOB(s, ntohl(blob.hdr_var)))) {
1250: rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob.hdr_var));
1251: blob.hdr_cmd = no;
1252: blob.hdr_ret = RPC_ERROR(-1);
1253: break;
1254: } else
1255: blob.hdr_len = htonl(b->blob_len);
1256:
1257: if (rpc_srv_blobMap(s, b) != -1) {
1258: /* deliver BLOB variable to client */
1259: blob.hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
1260: rpc_srv_blobUnmap(b);
1261: } else {
1262: blob.hdr_cmd = error;
1263: blob.hdr_ret = RPC_ERROR(-1);
1264: }
1265: break;
1266: case set:
1267: if ((b = rpc_srv_registerBLOB(s, ntohl(blob.hdr_len),
1268: ntohl(blob.hdr_ret)))) {
1269: /* set new BLOB variable for reply :) */
1270: blob.hdr_var = htonl(b->blob_var);
1271:
1272: /* receive BLOB from client */
1273: blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
1274: rpc_srv_blobUnmap(b);
1275: } else {
1276: blob.hdr_cmd = error;
1277: blob.hdr_ret = RPC_ERROR(-1);
1278: }
1279: break;
1280: case unset:
1281: if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) {
1282: blob.hdr_cmd = error;
1283: blob.hdr_ret = RPC_ERROR(-1);
1284: }
1285: break;
1286: default:
1287: rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);
1288: blob.hdr_cmd = error;
1289: blob.hdr_ret = RPC_ERROR(-1);
1290: }
1291:
1292: end:
1293: memcpy(AIT_ADDR(&c->cli_buf), &blob, sizeof blob);
1294: schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), NULL, 0);
1295: schedReadSelf(task);
1296: return NULL;
1297: }
1298:
1299: static void *
1300: flushBLOB(sched_task_t *task)
1301: {
1302: uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg);
1303: rpc_srv_t *srv = sigArg ? (void*) sigArg : TASK_ARG(task);
1304: rpc_blob_t *b, *tmp;
1305:
1306: TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
1307: TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);
1308:
1309: rpc_srv_blobFree(srv, b);
1310: e_free(b);
1311: }
1312:
1313: if (!schedSignalSelf(task)) {
1314: /* disabled kqueue support in libaitsched */
1315: struct sigaction sa;
1316:
1317: memset(&sa, 0, sizeof sa);
1318: sigemptyset(&sa.sa_mask);
1319: sa.sa_handler = (void (*)(int)) flushBLOB;
1320: sa.sa_flags = SA_RESTART | SA_RESETHAND;
1321: sigaction(SIGFBLOB, &sa, NULL);
1322: }
1323:
1324: return NULL;
1325: }
1326:
1327: static void *
1328: acceptBLOBClients(sched_task_t *task)
1329: {
1330: rpc_srv_t *srv = TASK_ARG(task);
1331: rpc_cli_t *c = NULL;
1332: register int i;
1333: socklen_t salen = sizeof(sockaddr_t);
1334: int sock;
1335: #ifdef TCP_NOPUSH
1336: int n = 1;
1337: #endif
1338:
1339: /* check free slots for connect */
1340: for (i = 0; i < array_Size(srv->srv_blob.clients) &&
1341: (c = array(srv->srv_blob.clients, i, rpc_cli_t*)); i++);
1342: if (c) { /* no more free slots! */
1343: EVERBOSE(1, "BLOB client quota exceeded! Connection will be shutdown!\n");
1344: if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
1345: shutdown(sock, SHUT_RDWR);
1346: close(sock);
1347: }
1348: goto end;
1349: }
1350:
1351: c = e_malloc(sizeof(rpc_cli_t));
1352: if (!c) {
1353: LOGERR;
1354: srv->srv_kill = srv->srv_blob.kill = 1;
1355: return NULL;
1356: } else {
1357: memset(c, 0, sizeof(rpc_cli_t));
1358: array_Set(srv->srv_blob.clients, i, c);
1359: c->cli_id = i;
1360: c->cli_parent = srv;
1361: }
1362:
1363: /* alloc empty buffer */
1364: AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
1365:
1366: /* accept client */
1367: c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
1368: if (c->cli_sock == -1) {
1369: LOGERR;
1370: AIT_FREE_VAL(&c->cli_buf);
1371: array_Del(srv->srv_blob.clients, i, 42);
1372: goto end;
1373: } else {
1374: #ifdef TCP_NOPUSH
1375: setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n);
1376: #endif
1377: fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1378: }
1379:
1380: schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0);
1381: end:
1382: schedReadSelf(task);
1383: return NULL;
1384: }
1385:
1386: /* ------------------------------------------------------ */
1387:
1388: /*
1389: * rpc_srv_initBLOBServer() - Init & create BLOB Server
1390: *
1391: * @srv = RPC server instance
1392: * @Port = Port for bind server, if Port == 0 default port is selected
1393: * @diskDir = Disk place for BLOB file objects
1394: * return: -1 == error or 0 bind and created BLOB server instance
1395: */
1396: int
1397: rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
1398: {
1399: int n = 1;
1400:
1401: if (!srv || srv->srv_kill) {
1402: rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
1403: return -1;
1404: }
1405:
1406: memset(&srv->srv_blob, 0, sizeof srv->srv_blob);
1407: if (access(diskDir, R_OK | W_OK) == -1) {
1408: LOGERR;
1409: return -1;
1410: } else
1411: AIT_SET_STR(&srv->srv_blob.dir, diskDir);
1412:
1413: /* init blob list */
1414: TAILQ_INIT(&srv->srv_blob.blobs);
1415:
1416: srv->srv_blob.server.cli_parent = srv;
1417:
1418: memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(sockaddr_t));
1419: switch (srv->srv_blob.server.cli_sa.sa.sa_family) {
1420: case AF_INET:
1421: srv->srv_blob.server.cli_sa.sin.sin_port =
1422: htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin.sin_port) + 1);
1423: break;
1424: case AF_INET6:
1425: srv->srv_blob.server.cli_sa.sin6.sin6_port =
1426: htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin6.sin6_port) + 1);
1427: break;
1428: case AF_LOCAL:
1429: strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob",
1430: sizeof srv->srv_blob.server.cli_sa.sun.sun_path);
1431: break;
1432: default:
1433: AIT_FREE_VAL(&srv->srv_blob.dir);
1434: return -1;
1435: }
1436:
1437: /* create BLOB server socket */
1438: srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
1439: if (srv->srv_blob.server.cli_sock == -1) {
1440: LOGERR;
1441: AIT_FREE_VAL(&srv->srv_blob.dir);
1442: return -1;
1443: }
1444: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
1445: LOGERR;
1446: close(srv->srv_blob.server.cli_sock);
1447: AIT_FREE_VAL(&srv->srv_blob.dir);
1448: return -1;
1449: }
1450: n = srv->srv_netbuf;
1451: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
1452: LOGERR;
1453: close(srv->srv_blob.server.cli_sock);
1454: AIT_FREE_VAL(&srv->srv_blob.dir);
1455: return -1;
1456: }
1457: if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
1458: LOGERR;
1459: close(srv->srv_blob.server.cli_sock);
1460: AIT_FREE_VAL(&srv->srv_blob.dir);
1461: return -1;
1462: }
1463: if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa,
1464: srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
1465: LOGERR;
1466: close(srv->srv_blob.server.cli_sock);
1467: AIT_FREE_VAL(&srv->srv_blob.dir);
1468: return -1;
1469: } else
1470: fcntl(srv->srv_blob.server.cli_sock, F_SETFL,
1471: fcntl(srv->srv_blob.server.cli_sock, F_GETFL) | O_NONBLOCK);
1472:
1473:
1474: /* allocate pool for concurent blob clients */
1475: srv->srv_blob.clients = array_Init(array_Size(srv->srv_clients));
1476: if (!srv->srv_blob.clients) {
1477: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1478: close(srv->srv_blob.server.cli_sock);
1479: AIT_FREE_VAL(&srv->srv_blob.dir);
1480: return -1;
1481: }
1482:
1483: /* init blob scheduler */
1484: srv->srv_blob.root = schedBegin();
1485: if (!srv->srv_blob.root) {
1486: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1487: array_Destroy(&srv->srv_blob.clients);
1488: close(srv->srv_blob.server.cli_sock);
1489: AIT_FREE_VAL(&srv->srv_blob.dir);
1490: return -1;
1491: }
1492:
1493: return 0;
1494: }
1495:
1496: /*
1497: * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
1498: *
1499: * @srv = RPC Server instance
1500: * return: none
1501: */
1502: void
1503: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
1504: {
1505: if (!srv)
1506: return;
1507:
1508: srv->srv_blob.kill = 1;
1509:
1510: schedEnd(&srv->srv_blob.root);
1511: }
1512:
1513: /*
1514: * rpc_srv_loopBLOBServer() - Execute Main BLOB server loop and wait for clients requests
1515: *
1516: * @srv = RPC Server instance
1517: * return: -1 error or 0 ok, infinite loop ...
1518: */
1519: int
1520: rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
1521: {
1522: rpc_cli_t *c;
1523: register int i;
1524: rpc_blob_t *b, *tmp;
1525: struct timespec ts = { RPC_SCHED_POLLING, 0 };
1526:
1527: if (!srv || srv->srv_kill) {
1528: rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
1529: return -1;
1530: }
1531:
1532: if (listen(srv->srv_blob.server.cli_sock, array_Size(srv->srv_blob.clients)) == -1) {
1533: LOGERR;
1534: return -1;
1535: }
1536:
1537: if (!schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0)) {
1538: /* disabled kqueue support in libaitsched */
1539: struct sigaction sa;
1540:
1541: atomic_store_rel_ptr(&_glSigArg, (uintptr_t) srv);
1542:
1543: memset(&sa, 0, sizeof sa);
1544: sigemptyset(&sa.sa_mask);
1545: sa.sa_handler = (void (*)(int)) flushBLOB;
1546: sa.sa_flags = SA_RESTART | SA_RESETHAND;
1547: sigaction(SIGFBLOB, &sa, NULL);
1548: }
1549:
1550: if (!schedRead(srv->srv_blob.root, acceptBLOBClients, srv,
1551: srv->srv_blob.server.cli_sock, NULL, 0)) {
1552: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1553: return -1;
1554: }
1555:
1556: schedPolling(srv->srv_blob.root, &ts, NULL);
1557: /* main rpc loop */
1558: schedRun(srv->srv_blob.root, &srv->srv_blob.kill);
1559:
1560: /* detach blobs */
1561: TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
1562: TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);
1563:
1564: rpc_srv_blobFree(srv, b);
1565: e_free(b);
1566: }
1567:
1568: /* close all clients connections & server socket */
1569: for (i = 0; i < array_Size(srv->srv_blob.clients); i++) {
1570: c = array(srv->srv_blob.clients, i, rpc_cli_t*);
1571: if (c) {
1572: shutdown(c->cli_sock, SHUT_RDWR);
1573: close(c->cli_sock);
1574:
1575: schedCancelby(srv->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
1576: AIT_FREE_VAL(&c->cli_buf);
1577: }
1578: array_Del(srv->srv_blob.clients, i, 42);
1579: }
1580: array_Destroy(&srv->srv_blob.clients);
1581:
1582: close(srv->srv_blob.server.cli_sock);
1583:
1584: AIT_FREE_VAL(&srv->srv_blob.dir);
1585: return 0;
1586: }
1587:
1588:
1589: /*
1590: * rpc_srv_initServer() - Init & create RPC Server
1591: *
1592: * @InstID = Instance for authentication & recognition
1593: * @concurentClients = Concurent clients at same time to this server
1594: * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
1595: * @csHost = Host name or address for bind server, if NULL any address
1596: * @Port = Port for bind server, if Port == 0 default port is selected
1597: * @proto = Protocol, if == 0 choose SOCK_STREAM
1598: * return: NULL == error or !=NULL bind and created RPC server instance
1599: */
1600: rpc_srv_t *
1601: rpc_srv_initServer(u_char InstID, int concurentClients, int netBuf,
1602: const char *csHost, u_short Port, int proto)
1603: {
1604: int n = 1;
1605: rpc_srv_t *srv = NULL;
1606: sockaddr_t sa = E_SOCKADDR_INIT;
1607:
1608: if (!concurentClients || (proto < 0 || proto > SOCK_RAW)) {
1609: rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1610: return NULL;
1611: }
1612: if (!e_gethostbyname(csHost, Port, &sa))
1613: return NULL;
1614: if (!Port)
1615: Port = RPC_DEFPORT;
1616: if (!proto)
1617: proto = SOCK_STREAM;
1618: if (netBuf < RPC_MIN_BUFSIZ)
1619: netBuf = BUFSIZ;
1620: else
1621: netBuf = E_ALIGN(netBuf, 2); /* align netBuf length */
1622:
1623: #ifdef HAVE_SRANDOMDEV
1624: srandomdev();
1625: #else
1626: time_t tim;
1627:
1628: srandom((time(&tim) ^ getpid()));
1629: #endif
1630:
1631: srv = e_malloc(sizeof(rpc_srv_t));
1632: if (!srv) {
1633: LOGERR;
1634: return NULL;
1635: } else
1636: memset(srv, 0, sizeof(rpc_srv_t));
1637:
1638: srv->srv_proto = proto;
1639: srv->srv_netbuf = netBuf;
1640: srv->srv_session.sess_version = RPC_VERSION;
1641: srv->srv_session.sess_instance = InstID;
1642:
1643: srv->srv_server.cli_parent = srv;
1644: memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
1645:
1646: /* init functions */
1647: pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
1648: SLIST_INIT(&srv->srv_funcs);
1649: AVL_INIT(&srv->srv_funcs);
1650:
1651: /* init scheduler */
1652: srv->srv_root = schedBegin();
1653: if (!srv->srv_root) {
1654: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1655: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1656: e_free(srv);
1657: return NULL;
1658: }
1659:
1660: /* init pool for clients */
1661: srv->srv_clients = array_Init(concurentClients);
1662: if (!srv->srv_clients) {
1663: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1664: schedEnd(&srv->srv_root);
1665: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1666: e_free(srv);
1667: return NULL;
1668: }
1669:
1670: /* create server socket */
1671: srv->srv_server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family,
1672: srv->srv_proto, srv->srv_proto == SOCK_RAW ? IPPROTO_ERPC : 0);
1673: if (srv->srv_server.cli_sock == -1) {
1674: LOGERR;
1675: array_Destroy(&srv->srv_clients);
1676: schedEnd(&srv->srv_root);
1677: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1678: e_free(srv);
1679: return NULL;
1680: }
1681: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
1682: LOGERR;
1683: goto err;
1684: }
1685: n = srv->srv_netbuf;
1686: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
1687: LOGERR;
1688: goto err;
1689: }
1690: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
1691: LOGERR;
1692: goto err;
1693: }
1694: if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa,
1695: srv->srv_server.cli_sa.sa.sa_len) == -1) {
1696: LOGERR;
1697: goto err;
1698: } else
1699: fcntl(srv->srv_server.cli_sock, F_SETFL,
1700: fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
1701:
1702: rpc_register_srvPing(srv);
1703:
1704: return srv;
1705: err: /* error condition */
1706: close(srv->srv_server.cli_sock);
1707: array_Destroy(&srv->srv_clients);
1708: schedEnd(&srv->srv_root);
1709: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1710: e_free(srv);
1711: return NULL;
1712: }
1713:
1714: /*
1715: * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
1716: *
1717: * @psrv = RPC Server instance
1718: * return: none
1719: */
1720: void
1721: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
1722: {
1723: if (!psrv || !*psrv)
1724: return;
1725:
1726: /* if send kill to blob server */
1727: rpc_srv_endBLOBServer(*psrv);
1728:
1729: (*psrv)->srv_kill = 1;
1730: sleep(RPC_SCHED_POLLING);
1731:
1732: schedEnd(&(*psrv)->srv_root);
1733:
1734: pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx);
1735: e_free(*psrv);
1736: *psrv = NULL;
1737: }
1738:
1739: /*
1740: * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
1741: *
1742: * @srv = RPC Server instance
1743: * return: -1 error or 0 ok, infinite loop ...
1744: */
1745: int
1746: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
1747: {
1748: rpc_cli_t *c;
1749: register int i;
1750: rpc_func_t *f;
1751: struct timespec ts = { RPC_SCHED_POLLING, 0 };
1752:
1753: if (!srv) {
1754: rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
1755: return -1;
1756: }
1757:
1758: if (srv->srv_proto == SOCK_STREAM)
1759: if (listen(srv->srv_server.cli_sock, array_Size(srv->srv_clients)) == -1) {
1760: LOGERR;
1761: return -1;
1762: }
1763:
1764: if (!schedRead(srv->srv_root, cbProto[srv->srv_proto][CB_ACCEPTCLIENT], srv,
1765: srv->srv_server.cli_sock, NULL, 0)) {
1766: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1767: return -1;
1768: }
1769:
1770: schedPolling(srv->srv_root, &ts, NULL);
1771: /* main rpc loop */
1772: schedRun(srv->srv_root, &srv->srv_kill);
1773:
1774: /* close all clients connections & server socket */
1775: for (i = 0; i < array_Size(srv->srv_clients); i++) {
1776: c = array(srv->srv_clients, i, rpc_cli_t*);
1777: if (c) {
1778: if (srv->srv_proto == SOCK_STREAM) {
1779: shutdown(c->cli_sock, SHUT_RDWR);
1780: close(c->cli_sock);
1781: }
1782:
1783: schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
1784: ait_freeVars(&RPC_RETVARS(c));
1785: AIT_FREE_VAL(&c->cli_buf);
1786: }
1787: array_Del(srv->srv_clients, i, 42);
1788: }
1789: array_Destroy(&srv->srv_clients);
1790:
1791: if (srv->srv_proto != SOCK_EXT)
1792: close(srv->srv_server.cli_sock);
1793:
1794: /* detach exported calls */
1795: RPC_FUNCS_LOCK(&srv->srv_funcs);
1796: while ((f = SLIST_FIRST(&srv->srv_funcs))) {
1797: SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);
1798:
1799: AIT_FREE_VAL(&f->func_name);
1800: e_free(f);
1801: }
1802: srv->srv_funcs.avlh_root = NULL;
1803: RPC_FUNCS_UNLOCK(&srv->srv_funcs);
1804:
1805: return 0;
1806: }
1807:
1808:
1809: /*
1810: * rpc_srv_execCall() Execute registered call from RPC server
1811: *
1812: * @cli = RPC client
1813: * @rpc = IN RPC call structure
1814: * @funcname = Execute RPC function
1815: * @args = IN RPC calling arguments from RPC client
1816: * return: -1 error, !=-1 ok
1817: */
1818: int
1819: rpc_srv_execCall(rpc_cli_t * __restrict cli, struct tagRPCCall * __restrict rpc,
1820: ait_val_t funcname, array_t * __restrict args)
1821: {
1822: rpc_callback_t func;
1823:
1824: if (!cli || !rpc || !AIT_ADDR(&funcname)) {
1825: rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1826: return -1;
1827: }
1828:
1829: func = AIT_GET_LIKE(&funcname, rpc_callback_t);
1830: return func(cli, rpc, args);
1831: }
1832:
1833:
1834: /*
1835: * rpc_srv_initServer2() - Init & create layer2 RPC Server
1836: *
1837: * @InstID = Instance for authentication & recognition
1838: * @concurentClients = Concurent clients at same time to this server
1839: * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
1840: * @csIface = Interface name for bind server, if NULL first interface on host
1841: * return: NULL == error or !=NULL bind and created RPC server instance
1842: */
1843: rpc_srv_t *
1844: rpc_srv_initServer2(u_char InstID, int concurentClients, int netBuf, const char *csIface)
1845: {
1846: int n = 1;
1847: rpc_srv_t *srv = NULL;
1848: sockaddr_t sa = E_SOCKADDR_INIT;
1849: char szIface[64], szStr[STRSIZ];
1850: register int i;
1851: struct ifreq ifr;
1852: struct bpf_insn insns[] = {
1853: BPF_STMT(BPF_LD + BPF_H + BPF_ABS, 12),
1854: BPF_JUMP(BPF_JMP + BPF_JEQ + BPF_K, RPC_DEFPORT, 0, 1),
1855: BPF_STMT(BPF_RET + BPF_K, -1),
1856: BPF_STMT(BPF_RET + BPF_K, 0),
1857: };
1858: struct bpf_program fcode = {
1859: .bf_len = sizeof(insns) / sizeof(struct bpf_insn),
1860: .bf_insns = insns
1861: };
1862:
1863: if (!concurentClients) {
1864: rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1865: return NULL;
1866: }
1867: if (!csIface) {
1868: if (e_get1stiface(szIface, sizeof szIface))
1869: return NULL;
1870: } else
1871: strlcpy(szIface, csIface, sizeof szIface);
1872: if (!e_getifacebyname(szIface, &sa))
1873: return NULL;
1874:
1875: #ifdef HAVE_SRANDOMDEV
1876: srandomdev();
1877: #else
1878: time_t tim;
1879:
1880: srandom((time(&tim) ^ getpid()));
1881: #endif
1882:
1883: srv = e_malloc(sizeof(rpc_srv_t));
1884: if (!srv) {
1885: LOGERR;
1886: return NULL;
1887: } else
1888: memset(srv, 0, sizeof(rpc_srv_t));
1889:
1890: srv->srv_proto = SOCK_BPF;
1891: srv->srv_netbuf = netBuf;
1892: srv->srv_session.sess_version = RPC_VERSION;
1893: srv->srv_session.sess_instance = InstID;
1894:
1895: srv->srv_server.cli_parent = srv;
1896: memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
1897:
1898: /* init functions */
1899: pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
1900: SLIST_INIT(&srv->srv_funcs);
1901: AVL_INIT(&srv->srv_funcs);
1902:
1903: /* init scheduler */
1904: srv->srv_root = schedBegin();
1905: if (!srv->srv_root) {
1906: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1907: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1908: e_free(srv);
1909: return NULL;
1910: }
1911:
1912: /* init pool for clients */
1913: srv->srv_clients = array_Init(concurentClients);
1914: if (!srv->srv_clients) {
1915: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1916: schedEnd(&srv->srv_root);
1917: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1918: e_free(srv);
1919: return NULL;
1920: }
1921:
1922: /* create server handler */
1923: for (i = 0; i < 10; i++) {
1924: memset(szStr, 0, sizeof szStr);
1925: snprintf(szStr, sizeof szStr, "/dev/bpf%d", i);
1926: srv->srv_server.cli_sock = open(szStr, O_RDWR);
1927: if (srv->srv_server.cli_sock > STDERR_FILENO)
1928: break;
1929: }
1930: if (srv->srv_server.cli_sock < 3) {
1931: LOGERR;
1932: array_Destroy(&srv->srv_clients);
1933: schedEnd(&srv->srv_root);
1934: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1935: e_free(srv);
1936: return NULL;
1937: }
1938:
1939: if (ioctl(srv->srv_server.cli_sock, BIOCIMMEDIATE, &n) == -1) {
1940: LOGERR;
1941: goto err;
1942: }
1943: if (ioctl(srv->srv_server.cli_sock, BIOCSETF, &fcode) == -1) {
1944: LOGERR;
1945: goto err;
1946: }
1947: n = (netBuf < RPC_MIN_BUFSIZ) ? getpagesize() : E_ALIGN(netBuf, 2);
1948: if (ioctl(srv->srv_server.cli_sock, BIOCSBLEN, &n) == -1) {
1949: LOGERR;
1950: goto err;
1951: } else
1952: srv->srv_netbuf = n;
1953:
1954: memset(&ifr, 0, sizeof ifr);
1955: strlcpy(ifr.ifr_name, szIface, sizeof ifr.ifr_name);
1956: if (ioctl(srv->srv_server.cli_sock, BIOCSETIF, &ifr) == -1) {
1957: LOGERR;
1958: goto err;
1959: } else
1960: fcntl(srv->srv_server.cli_sock, F_SETFL,
1961: fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
1962:
1963: rpc_register_srvPing(srv);
1964:
1965: return srv;
1966: err: /* error condition */
1967: close(srv->srv_server.cli_sock);
1968: array_Destroy(&srv->srv_clients);
1969: schedEnd(&srv->srv_root);
1970: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1971: e_free(srv);
1972: return NULL;
1973: }
1974:
1975: /*
1976: * rpc_srv_initServerExt() - Init & create pipe RPC Server
1977: *
1978: * @InstID = Instance for authentication & recognition
1979: * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
1980: * @fd = File descriptor
1981: * return: NULL == error or !=NULL bind and created RPC server instance
1982: */
1983: rpc_srv_t *
1984: rpc_srv_initServerExt(u_char InstID, int netBuf, int fd)
1985: {
1986: rpc_srv_t *srv = NULL;
1987:
1988: #ifdef HAVE_SRANDOMDEV
1989: srandomdev();
1990: #else
1991: time_t tim;
1992:
1993: srandom((time(&tim) ^ getpid()));
1994: #endif
1995:
1996: srv = e_malloc(sizeof(rpc_srv_t));
1997: if (!srv) {
1998: LOGERR;
1999: return NULL;
2000: } else
2001: memset(srv, 0, sizeof(rpc_srv_t));
2002:
2003: srv->srv_proto = SOCK_EXT;
2004: srv->srv_netbuf = (netBuf < RPC_MIN_BUFSIZ) ?
2005: getpagesize() : E_ALIGN(netBuf, 2);
2006: srv->srv_session.sess_version = RPC_VERSION;
2007: srv->srv_session.sess_instance = InstID;
2008:
2009: srv->srv_server.cli_parent = srv;
2010: srv->srv_server.cli_sock = fd;
2011:
2012: /* init functions */
2013: pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
2014: SLIST_INIT(&srv->srv_funcs);
2015: AVL_INIT(&srv->srv_funcs);
2016:
2017: /* init scheduler */
2018: srv->srv_root = schedBegin();
2019: if (!srv->srv_root) {
2020: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
2021: pthread_mutex_destroy(&srv->srv_funcs.mtx);
2022: e_free(srv);
2023: return NULL;
2024: }
2025:
2026: /* init pool for clients */
2027: srv->srv_clients = array_Init(1);
2028: if (!srv->srv_clients) {
2029: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
2030: schedEnd(&srv->srv_root);
2031: pthread_mutex_destroy(&srv->srv_funcs.mtx);
2032: e_free(srv);
2033: return NULL;
2034: }
2035:
2036: fcntl(srv->srv_server.cli_sock, F_SETFL,
2037: fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
2038:
2039: rpc_register_srvPing(srv);
2040:
2041: return srv;
2042: }
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