Annotation of libaitrpc/src/srv.c, revision 1.25.2.3
1.1 misho 1: /*************************************************************************
2: * (C) 2010 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
3: * by Michael Pounov <misho@openbsd-bg.org>
4: *
5: * $Author: misho $
1.25.2.3! misho 6: * $Id: srv.c,v 1.25.2.2 2015/03/09 22:01:34 misho Exp $
1.1 misho 7: *
1.2 misho 8: **************************************************************************
9: The ELWIX and AITNET software is distributed under the following
10: terms:
11:
12: All of the documentation and software included in the ELWIX and AITNET
13: Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>
14:
1.25 misho 15: Copyright 2004 - 2015
1.2 misho 16: by Michael Pounov <misho@elwix.org>. All rights reserved.
17:
18: Redistribution and use in source and binary forms, with or without
19: modification, are permitted provided that the following conditions
20: are met:
21: 1. Redistributions of source code must retain the above copyright
22: notice, this list of conditions and the following disclaimer.
23: 2. Redistributions in binary form must reproduce the above copyright
24: notice, this list of conditions and the following disclaimer in the
25: documentation and/or other materials provided with the distribution.
26: 3. All advertising materials mentioning features or use of this software
27: must display the following acknowledgement:
28: This product includes software developed by Michael Pounov <misho@elwix.org>
29: ELWIX - Embedded LightWeight unIX and its contributors.
30: 4. Neither the name of AITNET nor the names of its contributors
31: may be used to endorse or promote products derived from this software
32: without specific prior written permission.
33:
34: THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND
35: ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
36: IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
37: ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
38: FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
39: DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
40: OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
41: HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
42: LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
43: OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
44: SUCH DAMAGE.
45: */
1.1 misho 46: #include "global.h"
47:
48:
1.13 misho 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 */
1.25.2.2 misho 61: static void *rxRAWPacket(sched_task_t *);
62: static void *txRAWPacket(sched_task_t *);
1.13 misho 63:
1.24 misho 64: /* SOCK_BPF */
65: static void *rxBPFPacket(sched_task_t *);
66: static void *txBPFPacket(sched_task_t *);
67:
1.25 misho 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] = {
1.13 misho 73: { acceptClients, closeClient, rxPacket, txPacket }, /* SOCK_STREAM */
74: { acceptClients, closeClient, rxPacket, txPacket }, /* SOCK_STREAM */
75: { rxUDPPacket, freeClient, rxUDPPacket, txUDPPacket }, /* SOCK_DGRAM */
1.25.2.2 misho 76: { rxRAWPacket, freeClient, rxRAWPacket, txRAWPacket }, /* SOCK_RAW */
1.25 misho 77: { rxBPFPacket, freeClient, rxBPFPacket, txBPFPacket }, /* SOCK_BPF */
78: { rxEXTPacket, freeClient, rxEXTPacket, txEXTPacket } /* SOCK_EXT */
1.13 misho 79: };
80:
1.23 misho 81: /* Global Signal Argument when kqueue support disabled */
82:
83: static volatile uintptr_t _glSigArg = 0;
84:
1.13 misho 85:
1.16 misho 86: void
1.13 misho 87: rpc_freeCli(rpc_cli_t * __restrict c)
1.10 misho 88: {
89: rpc_srv_t *s = c->cli_parent;
90:
1.13 misho 91: schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
1.10 misho 92:
93: /* free buffer */
94: AIT_FREE_VAL(&c->cli_buf);
95:
1.14 misho 96: array_Del(s->srv_clients, c->cli_id, 0);
1.10 misho 97: if (c)
1.14 misho 98: e_free(c);
1.13 misho 99: }
100:
101:
102: static inline int
1.14 misho 103: _check4freeslot(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
1.13 misho 104: {
105: rpc_cli_t *c = NULL;
106: register int i;
107:
108: /* check free slots for connect */
1.14 misho 109: for (i = 0; i < array_Size(srv->srv_clients) &&
110: (c = array(srv->srv_clients, i, rpc_cli_t*)); i++)
1.13 misho 111: /* check for duplicates */
1.14 misho 112: if (sa && !e_addrcmp(&c->cli_sa, sa, 42))
1.13 misho 113: break;
1.14 misho 114: if (i >= array_Size(srv->srv_clients))
1.13 misho 115: return -1; /* no more free slots! */
116:
117: return i;
118: }
119:
120: static rpc_cli_t *
1.14 misho 121: _allocClient(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
1.13 misho 122: {
123: rpc_cli_t *c = NULL;
124: int n;
125:
1.25 misho 126: if (srv->srv_proto != SOCK_EXT)
127: n = _check4freeslot(srv, sa);
128: else
129: n = 0;
1.13 misho 130: if (n == -1)
131: return NULL;
132: else
1.14 misho 133: c = array(srv->srv_clients, n, rpc_cli_t*);
1.13 misho 134:
135: if (!c) {
1.14 misho 136: c = e_malloc(sizeof(rpc_cli_t));
1.13 misho 137: if (!c) {
138: LOGERR;
139: srv->srv_kill = 1;
140: return NULL;
141: } else {
142: memset(c, 0, sizeof(rpc_cli_t));
1.14 misho 143: array_Set(srv->srv_clients, n, c);
1.13 misho 144: c->cli_id = n;
145: c->cli_parent = srv;
146: }
147:
148: /* alloc empty buffer */
1.14 misho 149: AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
1.13 misho 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);
1.10 misho 174: return NULL;
175: }
1.7 misho 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;
1.18 misho 183: u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.7 misho 184: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1.18 misho 185: int ret, estlen, wlen = sizeof(struct tagRPCCall);
1.19 misho 186: struct pollfd pfd;
1.21 misho 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
1.7 misho 194:
195: if (rpc->call_argc) {
1.10 misho 196: f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
1.7 misho 197: if (!f) {
1.10 misho 198: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
1.18 misho 199:
1.7 misho 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 {
1.18 misho 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);
1.19 misho 209: rpc = (struct tagRPCCall*) buf;
1.18 misho 210:
1.25 misho 211: rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
1.7 misho 212: /* Go Encapsulate variables */
1.18 misho 213: ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
214: RPC_RETVARS(c));
1.10 misho 215: /* Free return values */
1.14 misho 216: ait_freeVars(&c->cli_vars);
1.7 misho 217: if (ret == -1) {
218: rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
1.19 misho 219:
1.7 misho 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:
1.18 misho 228: rpc->call_len = htonl(wlen);
1.25 misho 229: rpc->call_io = RPC_ACK;
1.8 misho 230:
1.15 misho 231: #if 0
1.7 misho 232: /* calculate CRC */
233: rpc->call_crc ^= rpc->call_crc;
1.8 misho 234: rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
1.15 misho 235: #endif
1.7 misho 236:
237: /* send reply */
1.19 misho 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
1.22 misho 246: rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
1.19 misho 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: }
1.10 misho 259: }
1.7 misho 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;
1.18 misho 271: u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.7 misho 272: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1.25 misho 273: int argc = rpc->call_argc;
1.7 misho 274:
275: /* Go decapsulate variables ... */
1.8 misho 276: if (argc) {
1.14 misho 277: arr = ait_buffer2vars(buf + sizeof(struct tagRPCCall),
1.18 misho 278: AIT_LEN(&c->cli_buf) - sizeof(struct tagRPCCall), argc, 42);
1.7 misho 279: if (!arr) {
1.14 misho 280: rpc_SetErr(ERPCMISMATCH, "#%d - %s", elwix_GetErrno(), elwix_GetError());
1.18 misho 281:
1.7 misho 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: }
1.10 misho 287: } else
288: arr = NULL;
1.7 misho 289:
1.10 misho 290: if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag)))) {
1.7 misho 291: rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
1.18 misho 292:
1.7 misho 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 {
1.8 misho 297: /* if client doesn't want reply */
1.10 misho 298: argc = RPC_CHK_NOREPLY(rpc);
299: rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(c, rpc, f->func_name, arr));
1.7 misho 300: if (rpc->call_rep.ret == htonl(-1)) {
1.20 misho 301: if (!rpc->call_rep.eno) {
302: LOGERR;
303: rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
304: }
1.7 misho 305: rpc->call_argc ^= rpc->call_argc;
306: } else {
307: rpc->call_rep.eno ^= rpc->call_rep.eno;
1.8 misho 308: if (argc) {
309: /* without reply */
1.14 misho 310: ait_freeVars(&c->cli_vars);
1.8 misho 311: rpc->call_argc ^= rpc->call_argc;
1.10 misho 312: } else {
313: /* reply */
1.25 misho 314: rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
1.10 misho 315: }
1.7 misho 316: }
317: }
318:
1.14 misho 319: array_Destroy(&arr);
1.7 misho 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;
1.18 misho 328: int len, rlen, noreply, estlen;
1.15 misho 329: #if 0
330: u_short crc;
331: #endif
1.10 misho 332: u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.18 misho 333: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1.19 misho 334: struct pollfd pfd;
1.21 misho 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
1.7 misho 342:
1.18 misho 343: memset(buf, 0, sizeof(struct tagRPCCall));
1.19 misho 344: rlen = recv(TASK_FD(task), rpc, sizeof(struct tagRPCCall), MSG_PEEK);
1.18 misho 345: if (rlen < sizeof(struct tagRPCCall)) {
1.10 misho 346: /* close connection */
1.13 misho 347: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
348: TASK_ARG(task), 0, NULL, 0);
1.7 misho 349: return NULL;
1.10 misho 350: } else {
1.18 misho 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);
1.19 misho 355: buf = AIT_GET_BUF(&c->cli_buf);
356: len = estlen;
1.18 misho 357: }
358:
1.19 misho 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
1.22 misho 369: rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
1.19 misho 370: schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
371: TASK_ARG(task), 0, NULL, 0);
372: return NULL;
373: }
1.18 misho 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);
1.8 misho 379: return NULL;
1.10 misho 380: }
1.18 misho 381: }
1.22 misho 382: len = estlen;
1.8 misho 383:
1.25 misho 384: /* skip loop packet */
385: if (rpc->call_io & RPC_ACK) {
386: schedReadSelf(task);
387: return NULL;
388: }
389:
1.15 misho 390: #if 0
1.18 misho 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: }
1.15 misho 398: #endif
1.7 misho 399:
1.18 misho 400: noreply = RPC_CHK_NOREPLY(rpc);
1.7 misho 401:
1.18 misho 402: /* check RPC packet session info */
403: if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
404: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1.7 misho 405:
1.18 misho 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: }
1.7 misho 413:
1.18 misho 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);
1.7 misho 418:
419: /* lets get next packet */
1.18 misho 420: schedReadSelf(task);
1.7 misho 421: return NULL;
422: }
423:
1.1 misho 424: static void *
1.10 misho 425: acceptClients(sched_task_t *task)
1.1 misho 426: {
1.10 misho 427: rpc_srv_t *srv = TASK_ARG(task);
428: rpc_cli_t *c = NULL;
1.14 misho 429: socklen_t salen = sizeof(sockaddr_t);
1.21 misho 430: int sock;
431: #ifdef TCP_SESSION_TIMEOUT
432: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
433: #endif
1.7 misho 434:
1.13 misho 435: c = _allocClient(srv, NULL);
1.21 misho 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: }
1.10 misho 442: goto end;
1.21 misho 443: }
1.10 misho 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);
1.14 misho 450: array_Del(srv->srv_clients, c->cli_id, 42);
1.10 misho 451: goto end;
1.1 misho 452: } else
1.10 misho 453: fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1.1 misho 454:
1.21 misho 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
1.13 misho 461: schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c,
462: c->cli_sock, NULL, 0);
1.10 misho 463: end:
464: schedReadSelf(task);
465: return NULL;
466: }
1.5 misho 467:
1.7 misho 468:
1.10 misho 469: static void *
1.13 misho 470: txUDPPacket(sched_task_t *task)
1.10 misho 471: {
472: rpc_cli_t *c = TASK_ARG(task);
473: rpc_srv_t *s = c->cli_parent;
1.13 misho 474: rpc_func_t *f = NULL;
1.18 misho 475: u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.13 misho 476: struct tagRPCCall *rpc = (struct tagRPCCall*) buf;
1.22 misho 477: int ret, estlen, wlen = sizeof(struct tagRPCCall);
1.13 misho 478: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
1.22 misho 479: struct pollfd pfd;
1.13 misho 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 {
1.22 misho 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:
1.25 misho 500: rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
1.13 misho 501: /* Go Encapsulate variables */
1.18 misho 502: ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
503: RPC_RETVARS(c));
1.13 misho 504: /* Free return values */
1.14 misho 505: ait_freeVars(&c->cli_vars);
1.13 misho 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: }
1.7 misho 515:
1.18 misho 516: rpc->call_len = htonl(wlen);
1.25 misho 517: rpc->call_io = RPC_ACK;
1.7 misho 518:
1.13 misho 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 */
1.22 misho 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: }
1.13 misho 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;
1.22 misho 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;
1.14 misho 560: sockaddr_t sa;
561: socklen_t salen;
1.13 misho 562: struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
1.22 misho 563: struct pollfd pfd;
1.13 misho 564:
565: /* receive connect packet */
1.14 misho 566: salen = sa.ss.ss_len = sizeof(sockaddr_t);
1.22 misho 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");
1.13 misho 570: goto end;
1.22 misho 571: }
1.13 misho 572:
573: c = _allocClient(srv, &sa);
1.22 misho 574: if (!c) {
575: EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
576: usleep(2000); /* blocked client delay */
1.13 misho 577: goto end;
1.22 misho 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:
1.13 misho 586: c->cli_sock = TASK_FD(task);
587: memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);
1.22 misho 588:
1.13 misho 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:
1.22 misho 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);
1.24 misho 608: goto end;
1.13 misho 609: }
1.22 misho 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);
1.24 misho 616: goto end;
1.13 misho 617: }
1.22 misho 618: if (e_addrcmp(&c->cli_sa, &sa, 42))
619: rlen ^= rlen; /* skip if arrive from different address */
620: }
621: len = estlen;
1.13 misho 622:
1.25 misho 623: /* skip loop packet */
624: if (rpc->call_io & RPC_ACK)
625: goto end;
626:
1.22 misho 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");
1.24 misho 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 *
1.25.2.2 misho 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);
1.25.2.3! misho 774: rpc = (struct tagRPCCall*) (b + estlen);
1.25.2.2 misho 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 *
1.24 misho 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:
1.25 misho 894: rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
1.24 misho 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);
1.25 misho 911: rpc->call_io = RPC_ACK;
1.24 misho 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);
1.22 misho 930: return NULL;
931: }
1.13 misho 932:
1.24 misho 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);
1.25 misho 964:
965: #if 0
966: /* skip loop packet */
967: if (rpc->call_io & RPC_ACK)
968: goto end;
969: #endif
970:
1.24 misho 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:
1.22 misho 1009: noreply = RPC_CHK_NOREPLY(rpc);
1.18 misho 1010:
1.22 misho 1011: /* check RPC packet session info */
1012: if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
1013: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1.13 misho 1014:
1.22 misho 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: }
1.13 misho 1022:
1.22 misho 1023: /* send RPC reply */
1024: if (!noreply)
1.24 misho 1025: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
1.22 misho 1026: c, TASK_FD(task), rpc, len);
1.13 misho 1027: end:
1.24 misho 1028: AIT_FREE_VAL(&b);
1.13 misho 1029: schedReadSelf(task);
1030: return NULL;
1031: }
1032:
1.25 misho 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:
1.13 misho 1165: /* ------------------------------------------------------ */
1166:
1.16 misho 1167: void
1.13 misho 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);
1.10 misho 1173:
1174: /* free buffer */
1175: AIT_FREE_VAL(&c->cli_buf);
1176:
1.14 misho 1177: array_Del(s->srv_blob.clients, c->cli_id, 0);
1.10 misho 1178: if (c)
1.14 misho 1179: e_free(c);
1.13 misho 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);
1.7 misho 1193: return NULL;
1194: }
1195:
1196: static void *
1197: txBLOB(sched_task_t *task)
1198: {
1.10 misho 1199: rpc_cli_t *c = TASK_ARG(task);
1200: u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.7 misho 1201: int wlen = sizeof(struct tagBLOBHdr);
1202:
1203: /* send reply */
1.10 misho 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: }
1.7 misho 1209:
1210: return NULL;
1211: }
1.4 misho 1212:
1.7 misho 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;
1.10 misho 1219: struct tagBLOBHdr blob;
1.7 misho 1220: int rlen;
1221:
1.10 misho 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);
1.7 misho 1227: return NULL;
1228: }
1.5 misho 1229:
1.10 misho 1230: /* check BLOB packet */
1231: if (rlen < sizeof(struct tagBLOBHdr)) {
1232: rpc_SetErr(ERPCMISMATCH, "Short BLOB packet");
1.6 misho 1233:
1.10 misho 1234: schedReadSelf(task);
1.7 misho 1235: return NULL;
1236: }
1.1 misho 1237:
1.7 misho 1238: /* check RPC packet session info */
1.15 misho 1239: if (rpc_chkPktSession(&blob.hdr_session, &s->srv_session)) {
1.7 misho 1240: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1.10 misho 1241: blob.hdr_cmd = error;
1.7 misho 1242: goto end;
1243: }
1244:
1245: /* Go to proceed packet ... */
1.10 misho 1246: switch (blob.hdr_cmd) {
1.7 misho 1247: case get:
1.10 misho 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);
1.7 misho 1252: break;
1253: } else
1.10 misho 1254: blob.hdr_len = htonl(b->blob_len);
1.5 misho 1255:
1.7 misho 1256: if (rpc_srv_blobMap(s, b) != -1) {
1257: /* deliver BLOB variable to client */
1.10 misho 1258: blob.hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
1.7 misho 1259: rpc_srv_blobUnmap(b);
1260: } else {
1.10 misho 1261: blob.hdr_cmd = error;
1262: blob.hdr_ret = RPC_ERROR(-1);
1.7 misho 1263: }
1264: break;
1265: case set:
1.17 misho 1266: if ((b = rpc_srv_registerBLOB(s, ntohl(blob.hdr_len),
1267: ntohl(blob.hdr_ret)))) {
1.7 misho 1268: /* set new BLOB variable for reply :) */
1.10 misho 1269: blob.hdr_var = htonl(b->blob_var);
1.7 misho 1270:
1271: /* receive BLOB from client */
1.10 misho 1272: blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
1.7 misho 1273: rpc_srv_blobUnmap(b);
1.5 misho 1274: } else {
1.10 misho 1275: blob.hdr_cmd = error;
1276: blob.hdr_ret = RPC_ERROR(-1);
1.7 misho 1277: }
1278: break;
1279: case unset:
1.11 misho 1280: if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) {
1.10 misho 1281: blob.hdr_cmd = error;
1282: blob.hdr_ret = RPC_ERROR(-1);
1.1 misho 1283: }
1284: break;
1.7 misho 1285: default:
1.10 misho 1286: rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);
1287: blob.hdr_cmd = error;
1288: blob.hdr_ret = RPC_ERROR(-1);
1.7 misho 1289: }
1.1 misho 1290:
1.7 misho 1291: end:
1.10 misho 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);
1.7 misho 1295: return NULL;
1.2 misho 1296: }
1297:
1298: static void *
1.17 misho 1299: flushBLOB(sched_task_t *task)
1300: {
1.23 misho 1301: uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg);
1302: rpc_srv_t *srv = sigArg ? (void*) sigArg : TASK_ARG(task);
1.17 misho 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:
1.23 misho 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:
1.17 misho 1323: return NULL;
1324: }
1325:
1326: static void *
1.10 misho 1327: acceptBLOBClients(sched_task_t *task)
1.2 misho 1328: {
1.10 misho 1329: rpc_srv_t *srv = TASK_ARG(task);
1330: rpc_cli_t *c = NULL;
1331: register int i;
1.14 misho 1332: socklen_t salen = sizeof(sockaddr_t);
1.21 misho 1333: int sock;
1.12 misho 1334: #ifdef TCP_NOPUSH
1335: int n = 1;
1336: #endif
1.7 misho 1337:
1.10 misho 1338: /* check free slots for connect */
1.14 misho 1339: for (i = 0; i < array_Size(srv->srv_blob.clients) &&
1340: (c = array(srv->srv_blob.clients, i, rpc_cli_t*)); i++);
1.21 misho 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: }
1.10 misho 1347: goto end;
1.21 misho 1348: }
1349:
1.14 misho 1350: c = e_malloc(sizeof(rpc_cli_t));
1.10 misho 1351: if (!c) {
1.7 misho 1352: LOGERR;
1.10 misho 1353: srv->srv_kill = srv->srv_blob.kill = 1;
1.7 misho 1354: return NULL;
1355: } else {
1.10 misho 1356: memset(c, 0, sizeof(rpc_cli_t));
1.14 misho 1357: array_Set(srv->srv_blob.clients, i, c);
1.10 misho 1358: c->cli_id = i;
1359: c->cli_parent = srv;
1.7 misho 1360: }
1.4 misho 1361:
1.10 misho 1362: /* alloc empty buffer */
1.14 misho 1363: AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
1.2 misho 1364:
1.10 misho 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);
1.14 misho 1370: array_Del(srv->srv_blob.clients, i, 42);
1.10 misho 1371: goto end;
1.12 misho 1372: } else {
1373: #ifdef TCP_NOPUSH
1374: setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n);
1375: #endif
1.10 misho 1376: fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1.12 misho 1377: }
1.2 misho 1378:
1.10 misho 1379: schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0);
1380: end:
1381: schedReadSelf(task);
1.7 misho 1382: return NULL;
1.1 misho 1383: }
1384:
1.10 misho 1385: /* ------------------------------------------------------ */
1.1 misho 1386:
1387: /*
1.7 misho 1388: * rpc_srv_initBLOBServer() - Init & create BLOB Server
1389: *
1.4 misho 1390: * @srv = RPC server instance
1.2 misho 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:
1.10 misho 1400: if (!srv || srv->srv_kill) {
1.7 misho 1401: rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
1.2 misho 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
1.7 misho 1410: AIT_SET_STR(&srv->srv_blob.dir, diskDir);
1.2 misho 1411:
1.10 misho 1412: /* init blob list */
1413: TAILQ_INIT(&srv->srv_blob.blobs);
1414:
1.2 misho 1415: srv->srv_blob.server.cli_parent = srv;
1.4 misho 1416:
1.14 misho 1417: memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(sockaddr_t));
1.10 misho 1418: switch (srv->srv_blob.server.cli_sa.sa.sa_family) {
1.4 misho 1419: case AF_INET:
1.10 misho 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);
1.4 misho 1422: break;
1423: case AF_INET6:
1.10 misho 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);
1.4 misho 1426: break;
1427: case AF_LOCAL:
1.10 misho 1428: strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob",
1429: sizeof srv->srv_blob.server.cli_sa.sun.sun_path);
1.4 misho 1430: break;
1431: default:
1.7 misho 1432: AIT_FREE_VAL(&srv->srv_blob.dir);
1.4 misho 1433: return -1;
1.2 misho 1434: }
1435:
1.4 misho 1436: /* create BLOB server socket */
1.6 misho 1437: srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
1.2 misho 1438: if (srv->srv_blob.server.cli_sock == -1) {
1439: LOGERR;
1.7 misho 1440: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 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);
1.7 misho 1446: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 1447: return -1;
1448: }
1.5 misho 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);
1.7 misho 1453: AIT_FREE_VAL(&srv->srv_blob.dir);
1.5 misho 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);
1.7 misho 1459: AIT_FREE_VAL(&srv->srv_blob.dir);
1.5 misho 1460: return -1;
1461: }
1.6 misho 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) {
1.2 misho 1464: LOGERR;
1465: close(srv->srv_blob.server.cli_sock);
1.7 misho 1466: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 1467: return -1;
1.13 misho 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:
1.2 misho 1472:
1.10 misho 1473: /* allocate pool for concurent blob clients */
1.14 misho 1474: srv->srv_blob.clients = array_Init(array_Size(srv->srv_clients));
1.2 misho 1475: if (!srv->srv_blob.clients) {
1.14 misho 1476: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1.2 misho 1477: close(srv->srv_blob.server.cli_sock);
1.7 misho 1478: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 1479: return -1;
1.10 misho 1480: }
1.2 misho 1481:
1.10 misho 1482: /* init blob scheduler */
1483: srv->srv_blob.root = schedBegin();
1484: if (!srv->srv_blob.root) {
1485: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1.14 misho 1486: array_Destroy(&srv->srv_blob.clients);
1.10 misho 1487: close(srv->srv_blob.server.cli_sock);
1488: AIT_FREE_VAL(&srv->srv_blob.dir);
1489: return -1;
1490: }
1.2 misho 1491:
1492: return 0;
1493: }
1494:
1495: /*
1.7 misho 1496: * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
1497: *
1.2 misho 1498: * @srv = RPC Server instance
1499: * return: none
1500: */
1.16 misho 1501: void
1.2 misho 1502: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
1503: {
1.10 misho 1504: if (!srv)
1.2 misho 1505: return;
1506:
1.10 misho 1507: srv->srv_blob.kill = 1;
1.17 misho 1508:
1509: schedEnd(&srv->srv_blob.root);
1.2 misho 1510: }
1511:
1512: /*
1.10 misho 1513: * rpc_srv_loopBLOBServer() - Execute Main BLOB server loop and wait for clients requests
1.7 misho 1514: *
1.2 misho 1515: * @srv = RPC Server instance
1516: * return: -1 error or 0 ok, infinite loop ...
1517: */
1518: int
1.10 misho 1519: rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
1.2 misho 1520: {
1.10 misho 1521: rpc_cli_t *c;
1.2 misho 1522: register int i;
1.10 misho 1523: rpc_blob_t *b, *tmp;
1524: struct timespec ts = { RPC_SCHED_POLLING, 0 };
1.2 misho 1525:
1.10 misho 1526: if (!srv || srv->srv_kill) {
1.7 misho 1527: rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
1.2 misho 1528: return -1;
1529: }
1530:
1.14 misho 1531: if (listen(srv->srv_blob.server.cli_sock, array_Size(srv->srv_blob.clients)) == -1) {
1.2 misho 1532: LOGERR;
1533: return -1;
1.10 misho 1534: }
1535:
1.23 misho 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:
1.10 misho 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: }
1.2 misho 1554:
1.10 misho 1555: schedPolling(srv->srv_blob.root, &ts, NULL);
1556: /* main rpc loop */
1557: schedRun(srv->srv_blob.root, &srv->srv_blob.kill);
1.7 misho 1558:
1.17 misho 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:
1.10 misho 1567: /* close all clients connections & server socket */
1.14 misho 1568: for (i = 0; i < array_Size(srv->srv_blob.clients); i++) {
1569: c = array(srv->srv_blob.clients, i, rpc_cli_t*);
1.10 misho 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);
1.2 misho 1576: }
1.14 misho 1577: array_Del(srv->srv_blob.clients, i, 42);
1.10 misho 1578: }
1.14 misho 1579: array_Destroy(&srv->srv_blob.clients);
1.2 misho 1580:
1.10 misho 1581: close(srv->srv_blob.server.cli_sock);
1.2 misho 1582:
1.10 misho 1583: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 1584: return 0;
1585: }
1586:
1587:
1588: /*
1.7 misho 1589: * rpc_srv_initServer() - Init & create RPC Server
1590: *
1.15 misho 1591: * @InstID = Instance for authentication & recognition
1.1 misho 1592: * @concurentClients = Concurent clients at same time to this server
1.10 misho 1593: * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
1.4 misho 1594: * @csHost = Host name or address for bind server, if NULL any address
1.1 misho 1595: * @Port = Port for bind server, if Port == 0 default port is selected
1.13 misho 1596: * @proto = Protocol, if == 0 choose SOCK_STREAM
1.1 misho 1597: * return: NULL == error or !=NULL bind and created RPC server instance
1598: */
1599: rpc_srv_t *
1.15 misho 1600: rpc_srv_initServer(u_char InstID, int concurentClients, int netBuf,
1601: const char *csHost, u_short Port, int proto)
1.1 misho 1602: {
1.10 misho 1603: int n = 1;
1.1 misho 1604: rpc_srv_t *srv = NULL;
1.14 misho 1605: sockaddr_t sa = E_SOCKADDR_INIT;
1.1 misho 1606:
1.25 misho 1607: if (!concurentClients || (proto < 0 || proto > SOCK_RAW)) {
1.10 misho 1608: rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1.1 misho 1609: return NULL;
1610: }
1.14 misho 1611: if (!e_gethostbyname(csHost, Port, &sa))
1.10 misho 1612: return NULL;
1.1 misho 1613: if (!Port)
1614: Port = RPC_DEFPORT;
1.13 misho 1615: if (!proto)
1616: proto = SOCK_STREAM;
1.10 misho 1617: if (netBuf < RPC_MIN_BUFSIZ)
1.5 misho 1618: netBuf = BUFSIZ;
1.7 misho 1619: else
1.14 misho 1620: netBuf = E_ALIGN(netBuf, 2); /* align netBuf length */
1.10 misho 1621:
1622: #ifdef HAVE_SRANDOMDEV
1623: srandomdev();
1624: #else
1625: time_t tim;
1626:
1627: srandom((time(&tim) ^ getpid()));
1628: #endif
1.1 misho 1629:
1.14 misho 1630: srv = e_malloc(sizeof(rpc_srv_t));
1.1 misho 1631: if (!srv) {
1632: LOGERR;
1633: return NULL;
1634: } else
1635: memset(srv, 0, sizeof(rpc_srv_t));
1636:
1.13 misho 1637: srv->srv_proto = proto;
1.5 misho 1638: srv->srv_netbuf = netBuf;
1.1 misho 1639: srv->srv_session.sess_version = RPC_VERSION;
1.15 misho 1640: srv->srv_session.sess_instance = InstID;
1.1 misho 1641:
1642: srv->srv_server.cli_parent = srv;
1.10 misho 1643: memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
1644:
1.12 misho 1645: /* init functions */
1646: pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
1647: SLIST_INIT(&srv->srv_funcs);
1648: AVL_INIT(&srv->srv_funcs);
1.10 misho 1649:
1650: /* init scheduler */
1651: srv->srv_root = schedBegin();
1652: if (!srv->srv_root) {
1653: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1.12 misho 1654: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14 misho 1655: e_free(srv);
1.10 misho 1656: return NULL;
1657: }
1658:
1659: /* init pool for clients */
1.14 misho 1660: srv->srv_clients = array_Init(concurentClients);
1.10 misho 1661: if (!srv->srv_clients) {
1.14 misho 1662: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1.10 misho 1663: schedEnd(&srv->srv_root);
1.12 misho 1664: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14 misho 1665: e_free(srv);
1.10 misho 1666: return NULL;
1667: }
1.4 misho 1668:
1669: /* create server socket */
1.25.2.1 misho 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);
1.1 misho 1672: if (srv->srv_server.cli_sock == -1) {
1673: LOGERR;
1.14 misho 1674: array_Destroy(&srv->srv_clients);
1.10 misho 1675: schedEnd(&srv->srv_root);
1.12 misho 1676: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14 misho 1677: e_free(srv);
1.1 misho 1678: return NULL;
1679: }
1680: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
1681: LOGERR;
1.10 misho 1682: goto err;
1.1 misho 1683: }
1.5 misho 1684: n = srv->srv_netbuf;
1685: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
1686: LOGERR;
1.10 misho 1687: goto err;
1.5 misho 1688: }
1689: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
1690: LOGERR;
1.10 misho 1691: goto err;
1.5 misho 1692: }
1.6 misho 1693: if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa,
1694: srv->srv_server.cli_sa.sa.sa_len) == -1) {
1.1 misho 1695: LOGERR;
1.10 misho 1696: goto err;
1.13 misho 1697: } else
1698: fcntl(srv->srv_server.cli_sock, F_SETFL,
1699: fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
1.1 misho 1700:
1.10 misho 1701: rpc_register_srvPing(srv);
1.8 misho 1702:
1.1 misho 1703: return srv;
1.10 misho 1704: err: /* error condition */
1705: close(srv->srv_server.cli_sock);
1.14 misho 1706: array_Destroy(&srv->srv_clients);
1.10 misho 1707: schedEnd(&srv->srv_root);
1.12 misho 1708: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14 misho 1709: e_free(srv);
1.10 misho 1710: return NULL;
1.1 misho 1711: }
1712:
1713: /*
1.7 misho 1714: * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
1715: *
1.6 misho 1716: * @psrv = RPC Server instance
1.1 misho 1717: * return: none
1718: */
1.16 misho 1719: void
1.6 misho 1720: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
1.1 misho 1721: {
1.10 misho 1722: if (!psrv || !*psrv)
1.1 misho 1723: return;
1724:
1.10 misho 1725: /* if send kill to blob server */
1.17 misho 1726: rpc_srv_endBLOBServer(*psrv);
1.1 misho 1727:
1.10 misho 1728: (*psrv)->srv_kill = 1;
1729: sleep(RPC_SCHED_POLLING);
1.2 misho 1730:
1.17 misho 1731: schedEnd(&(*psrv)->srv_root);
1732:
1.12 misho 1733: pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx);
1.14 misho 1734: e_free(*psrv);
1.6 misho 1735: *psrv = NULL;
1.1 misho 1736: }
1737:
1738: /*
1.7 misho 1739: * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
1740: *
1.1 misho 1741: * @srv = RPC Server instance
1742: * return: -1 error or 0 ok, infinite loop ...
1743: */
1744: int
1.5 misho 1745: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
1.1 misho 1746: {
1.10 misho 1747: rpc_cli_t *c;
1.1 misho 1748: register int i;
1.12 misho 1749: rpc_func_t *f;
1.10 misho 1750: struct timespec ts = { RPC_SCHED_POLLING, 0 };
1.1 misho 1751:
1752: if (!srv) {
1.10 misho 1753: rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
1.1 misho 1754: return -1;
1755: }
1756:
1.13 misho 1757: if (srv->srv_proto == SOCK_STREAM)
1.14 misho 1758: if (listen(srv->srv_server.cli_sock, array_Size(srv->srv_clients)) == -1) {
1.13 misho 1759: LOGERR;
1760: return -1;
1761: }
1.1 misho 1762:
1.13 misho 1763: if (!schedRead(srv->srv_root, cbProto[srv->srv_proto][CB_ACCEPTCLIENT], srv,
1764: srv->srv_server.cli_sock, NULL, 0)) {
1.10 misho 1765: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1766: return -1;
1767: }
1.7 misho 1768:
1.10 misho 1769: schedPolling(srv->srv_root, &ts, NULL);
1.7 misho 1770: /* main rpc loop */
1.10 misho 1771: schedRun(srv->srv_root, &srv->srv_kill);
1772:
1773: /* close all clients connections & server socket */
1.14 misho 1774: for (i = 0; i < array_Size(srv->srv_clients); i++) {
1775: c = array(srv->srv_clients, i, rpc_cli_t*);
1.10 misho 1776: if (c) {
1.24 misho 1777: if (srv->srv_proto == SOCK_STREAM) {
1778: shutdown(c->cli_sock, SHUT_RDWR);
1779: close(c->cli_sock);
1780: }
1.10 misho 1781:
1782: schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
1.14 misho 1783: ait_freeVars(&RPC_RETVARS(c));
1.10 misho 1784: AIT_FREE_VAL(&c->cli_buf);
1.1 misho 1785: }
1.14 misho 1786: array_Del(srv->srv_clients, i, 42);
1.10 misho 1787: }
1.14 misho 1788: array_Destroy(&srv->srv_clients);
1.2 misho 1789:
1.25 misho 1790: if (srv->srv_proto != SOCK_EXT)
1791: close(srv->srv_server.cli_sock);
1.2 misho 1792:
1.10 misho 1793: /* detach exported calls */
1.12 misho 1794: RPC_FUNCS_LOCK(&srv->srv_funcs);
1795: while ((f = SLIST_FIRST(&srv->srv_funcs))) {
1796: SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);
1.1 misho 1797:
1.10 misho 1798: AIT_FREE_VAL(&f->func_name);
1.14 misho 1799: e_free(f);
1.1 misho 1800: }
1.12 misho 1801: srv->srv_funcs.avlh_root = NULL;
1802: RPC_FUNCS_UNLOCK(&srv->srv_funcs);
1.1 misho 1803:
1804: return 0;
1805: }
1806:
1807:
1808: /*
1809: * rpc_srv_execCall() Execute registered call from RPC server
1.7 misho 1810: *
1.10 misho 1811: * @cli = RPC client
1.1 misho 1812: * @rpc = IN RPC call structure
1.10 misho 1813: * @funcname = Execute RPC function
1.5 misho 1814: * @args = IN RPC calling arguments from RPC client
1.1 misho 1815: * return: -1 error, !=-1 ok
1816: */
1817: int
1.10 misho 1818: rpc_srv_execCall(rpc_cli_t * __restrict cli, struct tagRPCCall * __restrict rpc,
1819: ait_val_t funcname, array_t * __restrict args)
1.1 misho 1820: {
1821: rpc_callback_t func;
1822:
1.10 misho 1823: if (!cli || !rpc || !AIT_ADDR(&funcname)) {
1.7 misho 1824: rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1.1 misho 1825: return -1;
1826: }
1827:
1.10 misho 1828: func = AIT_GET_LIKE(&funcname, rpc_callback_t);
1829: return func(cli, rpc, args);
1.1 misho 1830: }
1.24 misho 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: }
1.25 misho 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>
![[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