Annotation of libaitrpc/src/srv.c, revision 1.25.2.4
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.4! misho 6: * $Id: srv.c,v 1.25.2.3 2015/03/09 22:14:52 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);
1.25.2.4! misho 822: memmove(AIT_GET_BUF(&c->cli_buf), rpc, len);
1.25.2.2 misho 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 *
1.24 misho 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:
1.25 misho 895: rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
1.24 misho 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);
1.25 misho 912: rpc->call_io = RPC_ACK;
1.24 misho 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);
1.22 misho 931: return NULL;
932: }
1.13 misho 933:
1.24 misho 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);
1.25 misho 965:
966: #if 0
967: /* skip loop packet */
968: if (rpc->call_io & RPC_ACK)
969: goto end;
970: #endif
971:
1.24 misho 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:
1.22 misho 1010: noreply = RPC_CHK_NOREPLY(rpc);
1.18 misho 1011:
1.22 misho 1012: /* check RPC packet session info */
1013: if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
1014: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1.13 misho 1015:
1.22 misho 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: }
1.13 misho 1023:
1.22 misho 1024: /* send RPC reply */
1025: if (!noreply)
1.24 misho 1026: schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
1.22 misho 1027: c, TASK_FD(task), rpc, len);
1.13 misho 1028: end:
1.24 misho 1029: AIT_FREE_VAL(&b);
1.13 misho 1030: schedReadSelf(task);
1031: return NULL;
1032: }
1033:
1.25 misho 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:
1.13 misho 1166: /* ------------------------------------------------------ */
1167:
1.16 misho 1168: void
1.13 misho 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);
1.10 misho 1174:
1175: /* free buffer */
1176: AIT_FREE_VAL(&c->cli_buf);
1177:
1.14 misho 1178: array_Del(s->srv_blob.clients, c->cli_id, 0);
1.10 misho 1179: if (c)
1.14 misho 1180: e_free(c);
1.13 misho 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);
1.7 misho 1194: return NULL;
1195: }
1196:
1197: static void *
1198: txBLOB(sched_task_t *task)
1199: {
1.10 misho 1200: rpc_cli_t *c = TASK_ARG(task);
1201: u_char *buf = AIT_GET_BUF(&c->cli_buf);
1.7 misho 1202: int wlen = sizeof(struct tagBLOBHdr);
1203:
1204: /* send reply */
1.10 misho 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: }
1.7 misho 1210:
1211: return NULL;
1212: }
1.4 misho 1213:
1.7 misho 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;
1.10 misho 1220: struct tagBLOBHdr blob;
1.7 misho 1221: int rlen;
1222:
1.10 misho 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);
1.7 misho 1228: return NULL;
1229: }
1.5 misho 1230:
1.10 misho 1231: /* check BLOB packet */
1232: if (rlen < sizeof(struct tagBLOBHdr)) {
1233: rpc_SetErr(ERPCMISMATCH, "Short BLOB packet");
1.6 misho 1234:
1.10 misho 1235: schedReadSelf(task);
1.7 misho 1236: return NULL;
1237: }
1.1 misho 1238:
1.7 misho 1239: /* check RPC packet session info */
1.15 misho 1240: if (rpc_chkPktSession(&blob.hdr_session, &s->srv_session)) {
1.7 misho 1241: rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
1.10 misho 1242: blob.hdr_cmd = error;
1.7 misho 1243: goto end;
1244: }
1245:
1246: /* Go to proceed packet ... */
1.10 misho 1247: switch (blob.hdr_cmd) {
1.7 misho 1248: case get:
1.10 misho 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);
1.7 misho 1253: break;
1254: } else
1.10 misho 1255: blob.hdr_len = htonl(b->blob_len);
1.5 misho 1256:
1.7 misho 1257: if (rpc_srv_blobMap(s, b) != -1) {
1258: /* deliver BLOB variable to client */
1.10 misho 1259: blob.hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
1.7 misho 1260: rpc_srv_blobUnmap(b);
1261: } else {
1.10 misho 1262: blob.hdr_cmd = error;
1263: blob.hdr_ret = RPC_ERROR(-1);
1.7 misho 1264: }
1265: break;
1266: case set:
1.17 misho 1267: if ((b = rpc_srv_registerBLOB(s, ntohl(blob.hdr_len),
1268: ntohl(blob.hdr_ret)))) {
1.7 misho 1269: /* set new BLOB variable for reply :) */
1.10 misho 1270: blob.hdr_var = htonl(b->blob_var);
1.7 misho 1271:
1272: /* receive BLOB from client */
1.10 misho 1273: blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
1.7 misho 1274: rpc_srv_blobUnmap(b);
1.5 misho 1275: } else {
1.10 misho 1276: blob.hdr_cmd = error;
1277: blob.hdr_ret = RPC_ERROR(-1);
1.7 misho 1278: }
1279: break;
1280: case unset:
1.11 misho 1281: if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) {
1.10 misho 1282: blob.hdr_cmd = error;
1283: blob.hdr_ret = RPC_ERROR(-1);
1.1 misho 1284: }
1285: break;
1.7 misho 1286: default:
1.10 misho 1287: rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);
1288: blob.hdr_cmd = error;
1289: blob.hdr_ret = RPC_ERROR(-1);
1.7 misho 1290: }
1.1 misho 1291:
1.7 misho 1292: end:
1.10 misho 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);
1.7 misho 1296: return NULL;
1.2 misho 1297: }
1298:
1299: static void *
1.17 misho 1300: flushBLOB(sched_task_t *task)
1301: {
1.23 misho 1302: uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg);
1303: rpc_srv_t *srv = sigArg ? (void*) sigArg : TASK_ARG(task);
1.17 misho 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:
1.23 misho 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:
1.17 misho 1324: return NULL;
1325: }
1326:
1327: static void *
1.10 misho 1328: acceptBLOBClients(sched_task_t *task)
1.2 misho 1329: {
1.10 misho 1330: rpc_srv_t *srv = TASK_ARG(task);
1331: rpc_cli_t *c = NULL;
1332: register int i;
1.14 misho 1333: socklen_t salen = sizeof(sockaddr_t);
1.21 misho 1334: int sock;
1.12 misho 1335: #ifdef TCP_NOPUSH
1336: int n = 1;
1337: #endif
1.7 misho 1338:
1.10 misho 1339: /* check free slots for connect */
1.14 misho 1340: for (i = 0; i < array_Size(srv->srv_blob.clients) &&
1341: (c = array(srv->srv_blob.clients, i, rpc_cli_t*)); i++);
1.21 misho 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: }
1.10 misho 1348: goto end;
1.21 misho 1349: }
1350:
1.14 misho 1351: c = e_malloc(sizeof(rpc_cli_t));
1.10 misho 1352: if (!c) {
1.7 misho 1353: LOGERR;
1.10 misho 1354: srv->srv_kill = srv->srv_blob.kill = 1;
1.7 misho 1355: return NULL;
1356: } else {
1.10 misho 1357: memset(c, 0, sizeof(rpc_cli_t));
1.14 misho 1358: array_Set(srv->srv_blob.clients, i, c);
1.10 misho 1359: c->cli_id = i;
1360: c->cli_parent = srv;
1.7 misho 1361: }
1.4 misho 1362:
1.10 misho 1363: /* alloc empty buffer */
1.14 misho 1364: AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
1.2 misho 1365:
1.10 misho 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);
1.14 misho 1371: array_Del(srv->srv_blob.clients, i, 42);
1.10 misho 1372: goto end;
1.12 misho 1373: } else {
1374: #ifdef TCP_NOPUSH
1375: setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n);
1376: #endif
1.10 misho 1377: fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
1.12 misho 1378: }
1.2 misho 1379:
1.10 misho 1380: schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0);
1381: end:
1382: schedReadSelf(task);
1.7 misho 1383: return NULL;
1.1 misho 1384: }
1385:
1.10 misho 1386: /* ------------------------------------------------------ */
1.1 misho 1387:
1388: /*
1.7 misho 1389: * rpc_srv_initBLOBServer() - Init & create BLOB Server
1390: *
1.4 misho 1391: * @srv = RPC server instance
1.2 misho 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:
1.10 misho 1401: if (!srv || srv->srv_kill) {
1.7 misho 1402: rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
1.2 misho 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
1.7 misho 1411: AIT_SET_STR(&srv->srv_blob.dir, diskDir);
1.2 misho 1412:
1.10 misho 1413: /* init blob list */
1414: TAILQ_INIT(&srv->srv_blob.blobs);
1415:
1.2 misho 1416: srv->srv_blob.server.cli_parent = srv;
1.4 misho 1417:
1.14 misho 1418: memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(sockaddr_t));
1.10 misho 1419: switch (srv->srv_blob.server.cli_sa.sa.sa_family) {
1.4 misho 1420: case AF_INET:
1.10 misho 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);
1.4 misho 1423: break;
1424: case AF_INET6:
1.10 misho 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);
1.4 misho 1427: break;
1428: case AF_LOCAL:
1.10 misho 1429: strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob",
1430: sizeof srv->srv_blob.server.cli_sa.sun.sun_path);
1.4 misho 1431: break;
1432: default:
1.7 misho 1433: AIT_FREE_VAL(&srv->srv_blob.dir);
1.4 misho 1434: return -1;
1.2 misho 1435: }
1436:
1.4 misho 1437: /* create BLOB server socket */
1.6 misho 1438: srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
1.2 misho 1439: if (srv->srv_blob.server.cli_sock == -1) {
1440: LOGERR;
1.7 misho 1441: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 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);
1.7 misho 1447: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 1448: return -1;
1449: }
1.5 misho 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);
1.7 misho 1454: AIT_FREE_VAL(&srv->srv_blob.dir);
1.5 misho 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);
1.7 misho 1460: AIT_FREE_VAL(&srv->srv_blob.dir);
1.5 misho 1461: return -1;
1462: }
1.6 misho 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) {
1.2 misho 1465: LOGERR;
1466: close(srv->srv_blob.server.cli_sock);
1.7 misho 1467: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 1468: return -1;
1.13 misho 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:
1.2 misho 1473:
1.10 misho 1474: /* allocate pool for concurent blob clients */
1.14 misho 1475: srv->srv_blob.clients = array_Init(array_Size(srv->srv_clients));
1.2 misho 1476: if (!srv->srv_blob.clients) {
1.14 misho 1477: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1.2 misho 1478: close(srv->srv_blob.server.cli_sock);
1.7 misho 1479: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 1480: return -1;
1.10 misho 1481: }
1.2 misho 1482:
1.10 misho 1483: /* init blob scheduler */
1484: srv->srv_blob.root = schedBegin();
1485: if (!srv->srv_blob.root) {
1486: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1.14 misho 1487: array_Destroy(&srv->srv_blob.clients);
1.10 misho 1488: close(srv->srv_blob.server.cli_sock);
1489: AIT_FREE_VAL(&srv->srv_blob.dir);
1490: return -1;
1491: }
1.2 misho 1492:
1493: return 0;
1494: }
1495:
1496: /*
1.7 misho 1497: * rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
1498: *
1.2 misho 1499: * @srv = RPC Server instance
1500: * return: none
1501: */
1.16 misho 1502: void
1.2 misho 1503: rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
1504: {
1.10 misho 1505: if (!srv)
1.2 misho 1506: return;
1507:
1.10 misho 1508: srv->srv_blob.kill = 1;
1.17 misho 1509:
1510: schedEnd(&srv->srv_blob.root);
1.2 misho 1511: }
1512:
1513: /*
1.10 misho 1514: * rpc_srv_loopBLOBServer() - Execute Main BLOB server loop and wait for clients requests
1.7 misho 1515: *
1.2 misho 1516: * @srv = RPC Server instance
1517: * return: -1 error or 0 ok, infinite loop ...
1518: */
1519: int
1.10 misho 1520: rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
1.2 misho 1521: {
1.10 misho 1522: rpc_cli_t *c;
1.2 misho 1523: register int i;
1.10 misho 1524: rpc_blob_t *b, *tmp;
1525: struct timespec ts = { RPC_SCHED_POLLING, 0 };
1.2 misho 1526:
1.10 misho 1527: if (!srv || srv->srv_kill) {
1.7 misho 1528: rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
1.2 misho 1529: return -1;
1530: }
1531:
1.14 misho 1532: if (listen(srv->srv_blob.server.cli_sock, array_Size(srv->srv_blob.clients)) == -1) {
1.2 misho 1533: LOGERR;
1534: return -1;
1.10 misho 1535: }
1536:
1.23 misho 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:
1.10 misho 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: }
1.2 misho 1555:
1.10 misho 1556: schedPolling(srv->srv_blob.root, &ts, NULL);
1557: /* main rpc loop */
1558: schedRun(srv->srv_blob.root, &srv->srv_blob.kill);
1.7 misho 1559:
1.17 misho 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:
1.10 misho 1568: /* close all clients connections & server socket */
1.14 misho 1569: for (i = 0; i < array_Size(srv->srv_blob.clients); i++) {
1570: c = array(srv->srv_blob.clients, i, rpc_cli_t*);
1.10 misho 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);
1.2 misho 1577: }
1.14 misho 1578: array_Del(srv->srv_blob.clients, i, 42);
1.10 misho 1579: }
1.14 misho 1580: array_Destroy(&srv->srv_blob.clients);
1.2 misho 1581:
1.10 misho 1582: close(srv->srv_blob.server.cli_sock);
1.2 misho 1583:
1.10 misho 1584: AIT_FREE_VAL(&srv->srv_blob.dir);
1.2 misho 1585: return 0;
1586: }
1587:
1588:
1589: /*
1.7 misho 1590: * rpc_srv_initServer() - Init & create RPC Server
1591: *
1.15 misho 1592: * @InstID = Instance for authentication & recognition
1.1 misho 1593: * @concurentClients = Concurent clients at same time to this server
1.10 misho 1594: * @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
1.4 misho 1595: * @csHost = Host name or address for bind server, if NULL any address
1.1 misho 1596: * @Port = Port for bind server, if Port == 0 default port is selected
1.13 misho 1597: * @proto = Protocol, if == 0 choose SOCK_STREAM
1.1 misho 1598: * return: NULL == error or !=NULL bind and created RPC server instance
1599: */
1600: rpc_srv_t *
1.15 misho 1601: rpc_srv_initServer(u_char InstID, int concurentClients, int netBuf,
1602: const char *csHost, u_short Port, int proto)
1.1 misho 1603: {
1.10 misho 1604: int n = 1;
1.1 misho 1605: rpc_srv_t *srv = NULL;
1.14 misho 1606: sockaddr_t sa = E_SOCKADDR_INIT;
1.1 misho 1607:
1.25 misho 1608: if (!concurentClients || (proto < 0 || proto > SOCK_RAW)) {
1.10 misho 1609: rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
1.1 misho 1610: return NULL;
1611: }
1.14 misho 1612: if (!e_gethostbyname(csHost, Port, &sa))
1.10 misho 1613: return NULL;
1.1 misho 1614: if (!Port)
1615: Port = RPC_DEFPORT;
1.13 misho 1616: if (!proto)
1617: proto = SOCK_STREAM;
1.10 misho 1618: if (netBuf < RPC_MIN_BUFSIZ)
1.5 misho 1619: netBuf = BUFSIZ;
1.7 misho 1620: else
1.14 misho 1621: netBuf = E_ALIGN(netBuf, 2); /* align netBuf length */
1.10 misho 1622:
1623: #ifdef HAVE_SRANDOMDEV
1624: srandomdev();
1625: #else
1626: time_t tim;
1627:
1628: srandom((time(&tim) ^ getpid()));
1629: #endif
1.1 misho 1630:
1.14 misho 1631: srv = e_malloc(sizeof(rpc_srv_t));
1.1 misho 1632: if (!srv) {
1633: LOGERR;
1634: return NULL;
1635: } else
1636: memset(srv, 0, sizeof(rpc_srv_t));
1637:
1.13 misho 1638: srv->srv_proto = proto;
1.5 misho 1639: srv->srv_netbuf = netBuf;
1.1 misho 1640: srv->srv_session.sess_version = RPC_VERSION;
1.15 misho 1641: srv->srv_session.sess_instance = InstID;
1.1 misho 1642:
1643: srv->srv_server.cli_parent = srv;
1.10 misho 1644: memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
1645:
1.12 misho 1646: /* init functions */
1647: pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
1648: SLIST_INIT(&srv->srv_funcs);
1649: AVL_INIT(&srv->srv_funcs);
1.10 misho 1650:
1651: /* init scheduler */
1652: srv->srv_root = schedBegin();
1653: if (!srv->srv_root) {
1654: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1.12 misho 1655: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14 misho 1656: e_free(srv);
1.10 misho 1657: return NULL;
1658: }
1659:
1660: /* init pool for clients */
1.14 misho 1661: srv->srv_clients = array_Init(concurentClients);
1.10 misho 1662: if (!srv->srv_clients) {
1.14 misho 1663: rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
1.10 misho 1664: schedEnd(&srv->srv_root);
1.12 misho 1665: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14 misho 1666: e_free(srv);
1.10 misho 1667: return NULL;
1668: }
1.4 misho 1669:
1670: /* create server socket */
1.25.2.1 misho 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);
1.1 misho 1673: if (srv->srv_server.cli_sock == -1) {
1674: LOGERR;
1.14 misho 1675: array_Destroy(&srv->srv_clients);
1.10 misho 1676: schedEnd(&srv->srv_root);
1.12 misho 1677: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14 misho 1678: e_free(srv);
1.1 misho 1679: return NULL;
1680: }
1681: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
1682: LOGERR;
1.10 misho 1683: goto err;
1.1 misho 1684: }
1.5 misho 1685: n = srv->srv_netbuf;
1686: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
1687: LOGERR;
1.10 misho 1688: goto err;
1.5 misho 1689: }
1690: if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
1691: LOGERR;
1.10 misho 1692: goto err;
1.5 misho 1693: }
1.6 misho 1694: if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa,
1695: srv->srv_server.cli_sa.sa.sa_len) == -1) {
1.1 misho 1696: LOGERR;
1.10 misho 1697: goto err;
1.13 misho 1698: } else
1699: fcntl(srv->srv_server.cli_sock, F_SETFL,
1700: fcntl(srv->srv_server.cli_sock, F_GETFL) | O_NONBLOCK);
1.1 misho 1701:
1.10 misho 1702: rpc_register_srvPing(srv);
1.8 misho 1703:
1.1 misho 1704: return srv;
1.10 misho 1705: err: /* error condition */
1706: close(srv->srv_server.cli_sock);
1.14 misho 1707: array_Destroy(&srv->srv_clients);
1.10 misho 1708: schedEnd(&srv->srv_root);
1.12 misho 1709: pthread_mutex_destroy(&srv->srv_funcs.mtx);
1.14 misho 1710: e_free(srv);
1.10 misho 1711: return NULL;
1.1 misho 1712: }
1713:
1714: /*
1.7 misho 1715: * rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
1716: *
1.6 misho 1717: * @psrv = RPC Server instance
1.1 misho 1718: * return: none
1719: */
1.16 misho 1720: void
1.6 misho 1721: rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
1.1 misho 1722: {
1.10 misho 1723: if (!psrv || !*psrv)
1.1 misho 1724: return;
1725:
1.10 misho 1726: /* if send kill to blob server */
1.17 misho 1727: rpc_srv_endBLOBServer(*psrv);
1.1 misho 1728:
1.10 misho 1729: (*psrv)->srv_kill = 1;
1730: sleep(RPC_SCHED_POLLING);
1.2 misho 1731:
1.17 misho 1732: schedEnd(&(*psrv)->srv_root);
1733:
1.12 misho 1734: pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx);
1.14 misho 1735: e_free(*psrv);
1.6 misho 1736: *psrv = NULL;
1.1 misho 1737: }
1738:
1739: /*
1.7 misho 1740: * rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
1741: *
1.1 misho 1742: * @srv = RPC Server instance
1743: * return: -1 error or 0 ok, infinite loop ...
1744: */
1745: int
1.5 misho 1746: rpc_srv_loopServer(rpc_srv_t * __restrict srv)
1.1 misho 1747: {
1.10 misho 1748: rpc_cli_t *c;
1.1 misho 1749: register int i;
1.12 misho 1750: rpc_func_t *f;
1.10 misho 1751: struct timespec ts = { RPC_SCHED_POLLING, 0 };
1.1 misho 1752:
1753: if (!srv) {
1.10 misho 1754: rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
1.1 misho 1755: return -1;
1756: }
1757:
1.13 misho 1758: if (srv->srv_proto == SOCK_STREAM)
1.14 misho 1759: if (listen(srv->srv_server.cli_sock, array_Size(srv->srv_clients)) == -1) {
1.13 misho 1760: LOGERR;
1761: return -1;
1762: }
1.1 misho 1763:
1.13 misho 1764: if (!schedRead(srv->srv_root, cbProto[srv->srv_proto][CB_ACCEPTCLIENT], srv,
1765: srv->srv_server.cli_sock, NULL, 0)) {
1.10 misho 1766: rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
1767: return -1;
1768: }
1.7 misho 1769:
1.10 misho 1770: schedPolling(srv->srv_root, &ts, NULL);
1.7 misho 1771: /* main rpc loop */
1.10 misho 1772: schedRun(srv->srv_root, &srv->srv_kill);
1773:
1774: /* close all clients connections & server socket */
1.14 misho 1775: for (i = 0; i < array_Size(srv->srv_clients); i++) {
1776: c = array(srv->srv_clients, i, rpc_cli_t*);
1.10 misho 1777: if (c) {
1.24 misho 1778: if (srv->srv_proto == SOCK_STREAM) {
1779: shutdown(c->cli_sock, SHUT_RDWR);
1780: close(c->cli_sock);
1781: }
1.10 misho 1782:
1783: schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
1.14 misho 1784: ait_freeVars(&RPC_RETVARS(c));
1.10 misho 1785: AIT_FREE_VAL(&c->cli_buf);
1.1 misho 1786: }
1.14 misho 1787: array_Del(srv->srv_clients, i, 42);
1.10 misho 1788: }
1.14 misho 1789: array_Destroy(&srv->srv_clients);
1.2 misho 1790:
1.25 misho 1791: if (srv->srv_proto != SOCK_EXT)
1792: close(srv->srv_server.cli_sock);
1.2 misho 1793:
1.10 misho 1794: /* detach exported calls */
1.12 misho 1795: RPC_FUNCS_LOCK(&srv->srv_funcs);
1796: while ((f = SLIST_FIRST(&srv->srv_funcs))) {
1797: SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);
1.1 misho 1798:
1.10 misho 1799: AIT_FREE_VAL(&f->func_name);
1.14 misho 1800: e_free(f);
1.1 misho 1801: }
1.12 misho 1802: srv->srv_funcs.avlh_root = NULL;
1803: RPC_FUNCS_UNLOCK(&srv->srv_funcs);
1.1 misho 1804:
1805: return 0;
1806: }
1807:
1808:
1809: /*
1810: * rpc_srv_execCall() Execute registered call from RPC server
1.7 misho 1811: *
1.10 misho 1812: * @cli = RPC client
1.1 misho 1813: * @rpc = IN RPC call structure
1.10 misho 1814: * @funcname = Execute RPC function
1.5 misho 1815: * @args = IN RPC calling arguments from RPC client
1.1 misho 1816: * return: -1 error, !=-1 ok
1817: */
1818: int
1.10 misho 1819: rpc_srv_execCall(rpc_cli_t * __restrict cli, struct tagRPCCall * __restrict rpc,
1820: ait_val_t funcname, array_t * __restrict args)
1.1 misho 1821: {
1822: rpc_callback_t func;
1823:
1.10 misho 1824: if (!cli || !rpc || !AIT_ADDR(&funcname)) {
1.7 misho 1825: rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
1.1 misho 1826: return -1;
1827: }
1828:
1.10 misho 1829: func = AIT_GET_LIKE(&funcname, rpc_callback_t);
1830: return func(cli, rpc, args);
1.1 misho 1831: }
1.24 misho 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: }
1.25 misho 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