libaitrpc/src/srv.c - diff

Return to srv.c CVS log

Up to [ELWIX - Embedded LightWeight unIX -] / libaitrpc / src

Diff for /libaitrpc/src/srv.c between versions 1.13 and 1.28

version 1.13, 2012/11/19 21:50:26	version 1.28, 2015/07/22 20:01:46
Line 12 terms:	Line 12 terms:
All of the documentation and software included in the ELWIX and AITNET	All of the documentation and software included in the ELWIX and AITNET
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>	Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>

Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012	Copyright 2004 - 2015
by Michael Pounov <misho@elwix.org>. All rights reserved.	by Michael Pounov <misho@elwix.org>. All rights reserved.

Redistribution and use in source and binary forms, with or without	Redistribution and use in source and binary forms, with or without
Line 58 static void rxUDPPacket(sched_task_t );	Line 58 static void rxUDPPacket(sched_task_t );
static void txUDPPacket(sched_task_t );	static void txUDPPacket(sched_task_t );

/* SOCK_RAW */	/* SOCK_RAW */
	static void rxRAWPacket(sched_task_t );
	static void txRAWPacket(sched_task_t );

static sched_task_func_t cbProto[SOCK_RAW + 1][4] = {	/* SOCK_BPF */
{ acceptClients, closeClient, rxPacket, txPacket }, /* SOCK_STREAM */	static void rxBPFPacket(sched_task_t );
{ acceptClients, closeClient, rxPacket, txPacket }, /* SOCK_STREAM */	static void txBPFPacket(sched_task_t );
{ rxUDPPacket, freeClient, rxUDPPacket, txUDPPacket }, /* SOCK_DGRAM */
{ NULL, NULL, NULL, NULL } /* SOCK_RAW */	/* SOCK_EXT */
	static void rxEXTPacket(sched_task_t );
	static void txEXTPacket(sched_task_t );

	static sched_task_func_t cbProto[SOCK_MAX_SUPPORT][4] = {
	{ acceptClients, closeClient, rxPacket, txPacket }, /* SOCK_STREAM */
	{ acceptClients, closeClient, rxPacket, txPacket }, /* SOCK_STREAM */
	{ rxUDPPacket, freeClient, NULL /rxUDPPacket/, txUDPPacket }, /* SOCK_DGRAM */
	{ rxRAWPacket, freeClient, NULL /rxRAWPacket/, txRAWPacket }, /* SOCK_RAW */
	{ rxBPFPacket, freeClient, NULL /rxBPFPacket/, txBPFPacket }, /* SOCK_BPF */
	{ rxEXTPacket, freeClient, NULL /rxEXTPacket/, txEXTPacket } /* SOCK_EXT */
};	};

	/* Global Signal Argument when kqueue support disabled */

inline void	static volatile uintptr_t _glSigArg = 0;

	void
rpc_freeCli(rpc_cli_t * __restrict c)	rpc_freeCli(rpc_cli_t * __restrict c)
{	{
rpc_srv_t *s = c->cli_parent;	rpc_srv_t *s = c->cli_parent;
Line 77 rpc_freeCli(rpc_cli_t * __restrict c)	Line 92 rpc_freeCli(rpc_cli_t * __restrict c)
/* free buffer */	/* free buffer */
AIT_FREE_VAL(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);

io_arrayDel(s->srv_clients, c->cli_id, 0);	array_Del(s->srv_clients, c->cli_id, 0);
if (c)	if (c)
io_free(c);	e_free(c);
}	}


static inline int	static inline int
_check4freeslot(rpc_srv_t * __restrict srv, io_sockaddr_t * __restrict sa)	_check4freeslot(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
{	{
rpc_cli_t *c = NULL;	rpc_cli_t *c = NULL;
register int i;	register int i;

/* check free slots for connect */	/* check free slots for connect */
for (i = 0; i < io_arraySize(srv->srv_clients) &&	for (i = 0; i < array_Size(srv->srv_clients) &&
(c = io_array(srv->srv_clients, i, rpc_cli_t*)); i++)	(c = array(srv->srv_clients, i, rpc_cli_t*)); i++)
/* check for duplicates */	/* check for duplicates */
if (sa && !io_addrcmp(&c->cli_sa, sa, 42))	if (sa && !e_addrcmp(&c->cli_sa, sa, 42))
break;	break;
if (i >= io_arraySize(srv->srv_clients))	if (i >= array_Size(srv->srv_clients))
return -1; /* no more free slots! */	return -1; /* no more free slots! */

return i;	return i;
}	}

static rpc_cli_t *	static rpc_cli_t *
_allocClient(rpc_srv_t * __restrict srv, io_sockaddr_t * __restrict sa)	_allocClient(rpc_srv_t * __restrict srv, sockaddr_t * __restrict sa)
{	{
rpc_cli_t *c = NULL;	rpc_cli_t *c = NULL;
int n;	int n;

n = _check4freeslot(srv, sa);	if (srv->srv_proto != SOCK_EXT)
	n = _check4freeslot(srv, sa);
	else
	n = 0;
if (n == -1)	if (n == -1)
return NULL;	return NULL;
else	else
c = io_array(srv->srv_clients, n, rpc_cli_t*);	c = array(srv->srv_clients, n, rpc_cli_t*);

if (!c) {	if (!c) {
c = io_malloc(sizeof(rpc_cli_t));	c = e_malloc(sizeof(rpc_cli_t));
if (!c) {	if (!c) {
LOGERR;	LOGERR;
srv->srv_kill = 1;	srv->srv_kill = 1;
return NULL;	return NULL;
} else {	} else {
memset(c, 0, sizeof(rpc_cli_t));	memset(c, 0, sizeof(rpc_cli_t));
io_arraySet(srv->srv_clients, n, c);	array_Set(srv->srv_clients, n, c);
c->cli_id = n;	c->cli_id = n;
c->cli_parent = srv;	c->cli_parent = srv;
}	}

/* alloc empty buffer */	/* alloc empty buffer */
AIT_SET_BUF2(&c->cli_buf, 0, srv->srv_netbuf);	AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
}	}

return c;	return c;
Line 139 freeClient(sched_task_t *task)	Line 157 freeClient(sched_task_t *task)
{	{
rpc_freeCli(TASK_ARG(task));	rpc_freeCli(TASK_ARG(task));

return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 152 closeClient(sched_task_t *task)	Line 170 closeClient(sched_task_t *task)
/* close client socket */	/* close client socket */
shutdown(sock, SHUT_RDWR);	shutdown(sock, SHUT_RDWR);
close(sock);	close(sock);
return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 161 txPacket(sched_task_t *task)	Line 179 txPacket(sched_task_t *task)
rpc_cli_t *c = TASK_ARG(task);	rpc_cli_t *c = TASK_ARG(task);
rpc_srv_t *s = c->cli_parent;	rpc_srv_t *s = c->cli_parent;
rpc_func_t *f = NULL;	rpc_func_t *f = NULL;
u_char buf[USHRT_MAX] = { 0 };	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int ret, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
	#ifdef TCP_SESSION_TIMEOUT
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

/* copy RPC header */	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
memcpy(buf, TASK_DATA(task), wlen);	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), ts, TASK_ARG(task), 0);
	#endif

if (rpc->call_argc) {	if (rpc->call_argc) {
f = rpc_srv_getCall(s, ntohs(rpc->call_tag));	f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
if (!f) {	if (!f) {
rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");	rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");

rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
rpc->call_rep.ret = RPC_ERROR(-1);	RPC_SET_RETURN(rpc, -1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
	rpc_SetErr(EMSGSIZE, "Message too long");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
} else {	} else {
rpc->call_argc = htons(io_arraySize(RPC_RETVARS(c)));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = io_vars2buffer(buf + wlen, sizeof buf - wlen, RPC_RETVARS(c));	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
/* Free return values */	RPC_RETVARS(c));
io_freeVars(&c->cli_vars);
if (ret == -1) {	if (ret == -1) {
rpc_SetErr(EBADRPC, "Prepare RPC packet failed");	rpc_SetErr(EBADRPC, "Prepare RPC packet failed");

rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
rpc->call_rep.ret = RPC_ERROR(-1);	RPC_SET_RETURN(rpc, -1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	RPC_SET_ERRNO(rpc, rpc_Errno);
} else	} else
wlen += ret;	wlen += ret;
}	}
}	}

rpc->call_len = htons(wlen);	/* Free return values */
	ait_freeVars(&c->cli_vars);

	rpc->call_len = htonl(wlen);
	rpc->call_io = RPC_ACK;

	#if 0
/* calculate CRC */	/* calculate CRC */
rpc->call_crc ^= rpc->call_crc;	rpc->call_crc ^= rpc->call_crc;
rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
	#endif

/* send reply */	/* send reply */
ret = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);	ret = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);
if (ret == -1 \|\| ret != wlen) {	if (ret == -1) {
/* close connection */	/* close connection */
schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
TASK_ARG(task), 0, NULL, 0);	TASK_ARG(task), 0, NULL, 0);
}	}

return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 215 execCall(sched_task_t *task)	Line 250 execCall(sched_task_t *task)
rpc_srv_t *s = c->cli_parent;	rpc_srv_t *s = c->cli_parent;
rpc_func_t *f = NULL;	rpc_func_t *f = NULL;
array_t *arr = NULL;	array_t *arr = NULL;
u_char *buf = AIT_GET_BUF(&c->cli_buf) + TASK_VAL(task);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int argc = ntohs(rpc->call_argc);	int argc = rpc->call_argc;

/* Go decapsulate variables ... */	/* Go decapsulate variables ... */
if (argc) {	if (argc) {
arr = io_buffer2vars(buf + sizeof(struct tagRPCCall),	arr = ait_buffer2vars(buf + sizeof(struct tagRPCCall),
AIT_LEN(&c->cli_buf) - TASK_VAL(task) - sizeof(struct tagRPCCall),	AIT_LEN(&c->cli_buf) - sizeof(struct tagRPCCall), argc, 42);
argc, 42);
if (!arr) {	if (!arr) {
rpc_SetErr(ERPCMISMATCH, "#%d - %s", io_GetErrno(), io_GetError());	rpc_SetErr(ERPCMISMATCH, "#%d - %s", elwix_GetErrno(), elwix_GetError());

rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
rpc->call_rep.ret = RPC_ERROR(-1);	RPC_SET_RETURN(rpc, -1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	RPC_SET_ERRNO(rpc, rpc_Errno);
return NULL;	taskExit(task, NULL);
}	}
} else	} else
arr = NULL;	arr = NULL;

if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag)))) {	if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag)))) {
rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");	rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");

rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
rpc->call_rep.ret = RPC_ERROR(-1);	RPC_SET_RETURN(rpc, -1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	RPC_SET_ERRNO(rpc, rpc_Errno);
} else {	} else {
/* if client doesn't want reply */	/* if client doesn't want reply */
argc = RPC_CHK_NOREPLY(rpc);	RPC_SET_RETURN(rpc, rpc_srv_execCall(c, rpc, f->func_name, arr));
rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(c, rpc, f->func_name, arr));
if (rpc->call_rep.ret == htonl(-1)) {	if (rpc->call_rep.ret == htonl(-1)) {
rpc->call_rep.eno = RPC_ERROR(errno);	if (!rpc->call_rep.eno) {
	LOGERR;
	RPC_SET_ERRNO(rpc, rpc_Errno);
	}
rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
	ait_freeVars(&c->cli_vars);
} else {	} else {
rpc->call_rep.eno ^= rpc->call_rep.eno;	rpc->call_rep.eno ^= rpc->call_rep.eno;
if (argc) {	rpc->call_argc ^= rpc->call_argc;
	if (TASK_VAL(task)) {
/* without reply */	/* without reply */
io_freeVars(&c->cli_vars);	ait_freeVars(&c->cli_vars);
rpc->call_argc ^= rpc->call_argc;
} else {	} else {
/* reply */	/* reply */
rpc->call_argc = htons(io_arraySize(RPC_RETVARS(c)));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
}	}
}	}
}	}

io_arrayDestroy(&arr);	array_Destroy(&arr);
return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 268 rxPacket(sched_task_t *task)	Line 307 rxPacket(sched_task_t *task)
{	{
rpc_cli_t *c = TASK_ARG(task);	rpc_cli_t *c = TASK_ARG(task);
rpc_srv_t *s = c->cli_parent;	rpc_srv_t *s = c->cli_parent;
int len, rlen, noreply;	int len, noreply = 0, rlen = AIT_LEN(&c->cli_buf);
u_short crc, off = TASK_DATLEN(task);	#if 0
	u_short crc;
	#endif
u_char *buf = AIT_GET_BUF(&c->cli_buf);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall *rpc;	u_char b[sizeof(struct tagRPCCall)];
	struct tagRPCCall rpc = (struct tagRPCCall) buf;
	#ifdef TCP_SESSION_TIMEOUT
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

if (!off)	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
memset(buf, 0, AIT_LEN(&c->cli_buf));	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
rlen = recv(TASK_FD(task), buf + off, AIT_LEN(&c->cli_buf) - off, 0);	TASK_ARG(task), ts, TASK_ARG(task), 0);
if (rlen < 1) {	#endif

	/* prepare rx */
	len = recv(TASK_FD(task), b, sizeof b, MSG_PEEK);
	if (len == sizeof b)
	rlen = ntohl(((struct tagRPCCall*) b)->call_len);

	rlen = recv(TASK_FD(task), buf, rlen, 0);
	if (rlen == -1) {
/* close connection */	/* close connection */
schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
TASK_ARG(task), 0, NULL, 0);	TASK_ARG(task), 0, NULL, 0);
return NULL;	taskExit(task, NULL);
} else {
rlen += off; /* add reminded bytes from previous rxPacket, if exists! */
off = 0; /* process buffer from start offset == 0 */
}	}
	if (rlen < sizeof(struct tagRPCCall)) {
	rpc_SetErr(ERPCMISMATCH, "Short RPC packet");

do {	rpc->call_argc ^= rpc->call_argc;
/* check RPC packet */	RPC_SET_RETURN(rpc, -1);
if (rlen < sizeof(struct tagRPCCall)) {	RPC_SET_ERRNO(rpc, rpc_Errno);
rpc_SetErr(ERPCMISMATCH, "Short RPC packet");	goto err;
	} else
	len = ntohl(rpc->call_len);
	if (rlen < len \|\| len > AIT_LEN(&c->cli_buf)) {
	rpc_SetErr(ERPCMISMATCH, "Short RPC packet");

/* reminder received previous bytes ;) */	rpc->call_argc ^= rpc->call_argc;
schedRead(TASK_ROOT(task), TASK_FUNC(task), TASK_ARG(task),	RPC_SET_RETURN(rpc, -1);
TASK_FD(task), TASK_DATA(task), rlen);	RPC_SET_ERRNO(rpc, rpc_Errno);
return NULL;	goto err;
} else	}
rpc = (struct tagRPCCall*) (buf + off);

len = ntohs(rpc->call_len);	/* skip loop packet */
rlen -= len;	if (rpc->call_io & RPC_ACK) {
	schedReadSelf(task);
	taskExit(task, NULL);
	}

/* check RPC packet lengths */	#if 0
if (rlen < 0 \|\| len < sizeof(struct tagRPCCall)) {	/* check integrity of packet */
rpc_SetErr(ERPCMISMATCH, "Broken RPC packet length");	crc = ntohs(rpc->call_crc);
/* skip entire packet */	rpc->call_crc ^= rpc->call_crc;
break;	if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
}	rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");

/* check integrity of packet */	rpc->call_argc ^= rpc->call_argc;
crc = ntohs(rpc->call_crc);	RPC_SET_RETURN(rpc, -1);
rpc->call_crc ^= rpc->call_crc;	RPC_SET_ERRNO(rpc, rpc_Errno);
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {	goto err;
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");	}
	#endif

off += len;	/* check RPC packet session info */
/* try next packet remaining into buffer */	if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
continue;	rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
}

noreply = RPC_CHK_NOREPLY(rpc);	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	goto err;
	}

/* check RPC packet session info */	noreply = RPC_CHK_NOREPLY(rpc);
if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
rpc->call_argc ^= rpc->call_argc;
rpc->call_rep.ret = RPC_ERROR(-1);
rpc->call_rep.eno = RPC_ERROR(errno);
} else {
/* execute RPC call */
schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), off, NULL, 0);
}

/* send RPC reply */	/* execute RPC call */
if (!noreply)	schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len);
schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET],	err:
TASK_ARG(task), TASK_FD(task), rpc, len);	/* send RPC reply */
	if (!noreply)
	schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET],
	TASK_ARG(task), TASK_FD(task), rpc, len);

off += len;
} while (rlen > 0);

/* lets get next packet */	/* lets get next packet */
schedRead(TASK_ROOT(task), TASK_FUNC(task), TASK_ARG(task), TASK_FD(task), TASK_DATA(task), 0);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 350 acceptClients(sched_task_t *task)	Line 402 acceptClients(sched_task_t *task)
{	{
rpc_srv_t *srv = TASK_ARG(task);	rpc_srv_t *srv = TASK_ARG(task);
rpc_cli_t *c = NULL;	rpc_cli_t *c = NULL;
socklen_t salen = sizeof(io_sockaddr_t);	socklen_t salen = sizeof(sockaddr_t);
	int sock;
	#ifdef TCP_SESSION_TIMEOUT
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
	#endif

c = _allocClient(srv, NULL);	c = _allocClient(srv, NULL);
if (!c)	if (!c) {
	EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
	if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
	shutdown(sock, SHUT_RDWR);
	close(sock);
	}
goto end;	goto end;
	}

/* accept client */	/* accept client */
c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);	c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
if (c->cli_sock == -1) {	if (c->cli_sock == -1) {
LOGERR;	LOGERR;
AIT_FREE_VAL(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);
io_arrayDel(srv->srv_clients, c->cli_id, 42);	array_Del(srv->srv_clients, c->cli_id, 42);
goto end;	goto end;
} else	} else
fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) \| O_NONBLOCK);	fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) \| O_NONBLOCK);

	#ifdef TCP_SESSION_TIMEOUT
	/* armed timer for close stateless connection */
	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
	schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT], c,
	ts, c, 0);
	#endif
schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c,	schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c,
c->cli_sock, NULL, 0);	c->cli_sock, NULL, 0);
end:	end:
schedReadSelf(task);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}


Line 380 txUDPPacket(sched_task_t *task)	Line 448 txUDPPacket(sched_task_t *task)
rpc_cli_t *c = TASK_ARG(task);	rpc_cli_t *c = TASK_ARG(task);
rpc_srv_t *s = c->cli_parent;	rpc_srv_t *s = c->cli_parent;
rpc_func_t *f = NULL;	rpc_func_t *f = NULL;
u_char buf[USHRT_MAX] = { 0 };	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int ret, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
Line 389 txUDPPacket(sched_task_t *task)	Line 457 txUDPPacket(sched_task_t *task)
schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
TASK_ARG(task), ts, TASK_ARG(task), 0);	TASK_ARG(task), ts, TASK_ARG(task), 0);

/* copy RPC header */
memcpy(buf, TASK_DATA(task), wlen);

if (rpc->call_argc) {	if (rpc->call_argc) {
f = rpc_srv_getCall(s, ntohs(rpc->call_tag));	f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
if (!f) {	if (!f) {
rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");	rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");

rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
rpc->call_rep.ret = RPC_ERROR(-1);	RPC_SET_RETURN(rpc, -1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
	rpc_SetErr(EMSGSIZE, "Message too long");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
} else {	} else {
rpc->call_argc = htons(io_arraySize(RPC_RETVARS(c)));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = io_vars2buffer(buf + wlen, sizeof buf - wlen, RPC_RETVARS(c));	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
/* Free return values */	RPC_RETVARS(c));
io_freeVars(&c->cli_vars);
if (ret == -1) {	if (ret == -1) {
rpc_SetErr(EBADRPC, "Prepare RPC packet failed");	rpc_SetErr(EBADRPC, "Prepare RPC packet failed");

rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
rpc->call_rep.ret = RPC_ERROR(-1);	RPC_SET_RETURN(rpc, -1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	RPC_SET_ERRNO(rpc, rpc_Errno);
} else	} else
wlen += ret;	wlen += ret;
}	}
}	}

rpc->call_len = htons(wlen);	/* Free return values */
	ait_freeVars(&c->cli_vars);

	rpc->call_len = htonl(wlen);
	rpc->call_io = RPC_ACK;

/* calculate CRC */	/* calculate CRC */
rpc->call_crc ^= rpc->call_crc;	rpc->call_crc ^= rpc->call_crc;
rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
Line 424 txUDPPacket(sched_task_t *task)	Line 500 txUDPPacket(sched_task_t *task)
/* send reply */	/* send reply */
ret = sendto(TASK_FD(task), buf, wlen, MSG_NOSIGNAL,	ret = sendto(TASK_FD(task), buf, wlen, MSG_NOSIGNAL,
&c->cli_sa.sa, c->cli_sa.sa.sa_len);	&c->cli_sa.sa, c->cli_sa.sa.sa_len);
if (ret == -1 \|\| ret != wlen) {	if (ret == -1) {
/* close connection */	/* close connection */
schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
TASK_ARG(task), 0, NULL, 0);	TASK_ARG(task), 0, NULL, 0);
}	}

return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 438 rxUDPPacket(sched_task_t *task)	Line 514 rxUDPPacket(sched_task_t *task)
{	{
rpc_srv_t *srv = TASK_ARG(task);	rpc_srv_t *srv = TASK_ARG(task);
rpc_cli_t *c = NULL;	rpc_cli_t *c = NULL;
socklen_t salen = sizeof(io_sockaddr_t);	int len, noreply = 0, rlen;
int len, rlen, noreply;	u_short crc;
u_short crc, off = 0;	struct tagRPCCall *rpc;
u_char buf[USHRT_MAX + 1];	sockaddr_t sa;
struct tagRPCCall rpc = (struct tagRPCCall) buf;	socklen_t salen;
io_sockaddr_t sa;
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
	ait_val_t b = AIT_VAL_INIT;

/* receive connect packet */	/* receive connect packet */
rlen = recvfrom(TASK_FD(task), buf, sizeof buf, 0, &sa.sa, &salen);	AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
if (rlen < 1)	salen = sa.ss.ss_len = (u_char) MIN(sizeof(sockaddr_t), 0xff);
	rlen = recvfrom(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b), 0, &sa.sa, &salen);
	rpc = (struct tagRPCCall*) AIT_GET_BUF(&b);
	if (rlen < sizeof(struct tagRPCCall))
goto end;	goto end;
	else
	len = ntohl(rpc->call_len);
	if (rlen < len \|\| len > srv->srv_netbuf)
	goto end;

	/* skip loop packet */
	if (rpc->call_io & RPC_ACK)
	goto end;

	/* check integrity of packet */
	crc = ntohs(rpc->call_crc);
	rpc->call_crc ^= rpc->call_crc;
	if (crc != crcFletcher16((u_short*) rpc, len / 2))
	goto end;

	/* check RPC packet session info */
	if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session))
	goto end;

c = _allocClient(srv, &sa);	c = _allocClient(srv, &sa);
if (!c)	if (!c) {
	EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
	usleep(2000); /* blocked client delay */
goto end;	goto end;
else {	} else {
	memcpy(AIT_GET_BUF(&c->cli_buf), AIT_GET_BUF(&b), len);
	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);

c->cli_sock = TASK_FD(task);	c->cli_sock = TASK_FD(task);
memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);	memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);
memcpy(AIT_GET_BUF(&c->cli_buf), buf, AIT_LEN(&c->cli_buf));
/* armed timer for close stateless connection */	/* armed timer for close stateless connection */
schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],	schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
c, ts, c, 0);	c, ts, c, 0);
}	}

do {	noreply = RPC_CHK_NOREPLY(rpc);
/* check RPC packet */
if (rlen < sizeof(struct tagRPCCall)) {
rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
break;
} else
rpc = (struct tagRPCCall*) (AIT_GET_BUF(&c->cli_buf) + off);

len = ntohs(rpc->call_len);	/* execute RPC call */
rlen -= len;	schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);

/* check RPC packet lengths */	/* send RPC reply */
if (rlen < 0 \|\| len < sizeof(struct tagRPCCall)) {	if (!noreply)
rpc_SetErr(ERPCMISMATCH, "Broken RPC packet length");	schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
/* skip entire packet */	c, TASK_FD(task), rpc, len);
break;	end:
	AIT_FREE_VAL(&b);
	schedReadSelf(task);
	taskExit(task, NULL);
	}


	static void *
	txRAWPacket(sched_task_t *task)
	{
	rpc_cli_t *c = TASK_ARG(task);
	rpc_srv_t *s = c->cli_parent;
	rpc_func_t *f = NULL;
	u_char *buf = AIT_GET_BUF(&c->cli_buf);
	struct tagRPCCall rpc = (struct tagRPCCall) buf;
	int ret, wlen = sizeof(struct tagRPCCall);
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), ts, TASK_ARG(task), 0);

	if (rpc->call_argc) {
	f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
	if (!f) {
	rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
	rpc_SetErr(EMSGSIZE, "Message too long");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else {
	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
	/* Go Encapsulate variables */
	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
	RPC_RETVARS(c));
	if (ret == -1) {
	rpc_SetErr(EBADRPC, "Prepare RPC packet failed");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else
	wlen += ret;
}	}
	}

/* check integrity of packet */	/* Free return values */
crc = ntohs(rpc->call_crc);	ait_freeVars(&c->cli_vars);
rpc->call_crc ^= rpc->call_crc;
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");

off += len;	rpc->call_len = htonl(wlen);
/* try next packet remaining into buffer */	rpc->call_io = RPC_ACK;
continue;
	/* calculate CRC */
	rpc->call_crc ^= rpc->call_crc;
	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));

	/* send reply */
	ret = sendto(TASK_FD(task), buf, wlen, MSG_NOSIGNAL,
	&c->cli_sa.sa, c->cli_sa.sa.sa_len);
	if (ret == -1) {
	/* close connection */
	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), 0, NULL, 0);
	}

	taskExit(task, NULL);
	}

	static void *
	rxRAWPacket(sched_task_t *task)
	{
	rpc_srv_t *srv = TASK_ARG(task);
	rpc_cli_t *c = NULL;
	int len, noreply = 0, rlen;
	u_short crc;
	struct tagRPCCall *rpc;
	sockaddr_t sa;
	socklen_t salen;
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
	ait_val_t b = AIT_VAL_INIT;

	/* receive connect packet */
	AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
	salen = sa.ss.ss_len = (u_char) MIN(sizeof(sockaddr_t), 0xff);
	rlen = recvfrom(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b), 0, &sa.sa, &salen);
	if (sa.sa.sa_family == AF_INET) {
	struct ip *h;
	h = (struct ip*) AIT_GET_BUF(&b);
	if (rlen < ntohs(h->ip_len) \|\| h->ip_p != IPPROTO_ERPC)
	goto end;
	else {
	rlen -= sizeof(struct ip);
	rpc = (struct tagRPCCall*) (h + 1);
}	}
	} else {
	struct ip6_hdr *h;
	h = (struct ip6_hdr*) AIT_GET_BUF(&b);
	if (rlen < (ntohs(h->ip6_plen) + sizeof(struct ip6_hdr)) \|\|
	h->ip6_nxt != IPPROTO_ERPC)
	goto end;
	else {
	rlen -= sizeof(struct ip6_hdr);
	rpc = (struct tagRPCCall*) (h + 1);
	}
	}
	if (rlen < sizeof(struct tagRPCCall))
	goto end;
	else
	len = ntohl(rpc->call_len);
	if (rlen < len \|\| len > srv->srv_netbuf)
	goto end;

noreply = RPC_CHK_NOREPLY(rpc);	/* skip loop packet */
	if (rpc->call_io & RPC_ACK)
	goto end;

/* check RPC packet session info */	/* check integrity of packet */
if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {	crc = ntohs(rpc->call_crc);
rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");	rpc->call_crc ^= rpc->call_crc;
	if (crc != crcFletcher16((u_short*) rpc, len / 2))
	goto end;

	/* check RPC packet session info */
	if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session))
	goto end;

	c = _allocClient(srv, &sa);
	if (!c) {
	EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
	usleep(2000); /* blocked client delay */
	goto end;
	} else {
	memcpy(AIT_GET_BUF(&c->cli_buf), rpc, len);
	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);

	c->cli_sock = TASK_FD(task);
	memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);

	/* armed timer for close stateless connection */
	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
	schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
	c, ts, c, 0);
	}

	noreply = RPC_CHK_NOREPLY(rpc);

	/* execute RPC call */
	schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);

	/* send RPC reply */
	if (!noreply)
	schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
	c, TASK_FD(task), rpc, len);
	end:
	AIT_FREE_VAL(&b);
	schedReadSelf(task);
	taskExit(task, NULL);
	}


	static void *
	txBPFPacket(sched_task_t *task)
	{
	rpc_cli_t *c = TASK_ARG(task);
	rpc_srv_t *s = c->cli_parent;
	rpc_func_t *f = NULL;
	u_char *buf = AIT_GET_BUF(&c->cli_buf);
	struct tagRPCCall rpc = (struct tagRPCCall) buf;
	int ret, wlen = sizeof(struct tagRPCCall);
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
	struct ether_header *eh;
	ait_val_t b = AIT_VAL_INIT;

	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), ts, TASK_ARG(task), 0);

	if (rpc->call_argc) {
	f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
	if (!f) {
	rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");

rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
rpc->call_rep.ret = RPC_ERROR(-1);	RPC_SET_RETURN(rpc, -1);
rpc->call_rep.eno = RPC_ERROR(errno);	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
	rpc_SetErr(EMSGSIZE, "Message too long");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
} else {	} else {
/* execute RPC call */	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
schedEvent(TASK_ROOT(task), execCall, c, off, NULL, 0);	/* Go Encapsulate variables */
	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
	RPC_RETVARS(c));
	if (ret == -1) {
	rpc_SetErr(EBADRPC, "Prepare RPC packet failed");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else
	wlen += ret;
}	}
	}

/* send RPC reply */	/* Free return values */
if (!noreply)	ait_freeVars(&RPC_RETVARS(c));
schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
c, TASK_FD(task), rpc, len);

off += len;	rpc->call_len = htonl(wlen);
} while (rlen > 0);	rpc->call_io = RPC_ACK;

	/* calculate CRC */
	rpc->call_crc ^= rpc->call_crc;
	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));

	/* send reply */
	AIT_SET_BUF(&b, NULL, wlen + ETHER_HDR_LEN);
	eh = (struct ether_header*) AIT_GET_BUF(&b);
	memcpy(eh->ether_dhost, LLADDR(&c->cli_sa.sdl), ETHER_ADDR_LEN);
	eh->ether_type = htons(RPC_DEFPORT);
	memcpy(eh + 1, buf, wlen);

	ret = write(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
	AIT_FREE_VAL(&b);
	if (ret == -1) {
	/* close connection */
	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), 0, NULL, 0);
	}

	taskExit(task, NULL);
	}

	static void *
	rxBPFPacket(sched_task_t *task)
	{
	rpc_srv_t *srv = TASK_ARG(task);
	rpc_cli_t *c = NULL;
	int len, rlen, noreply;
	u_short crc;
	struct tagRPCCall *rpc;
	sockaddr_t sa;
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
	struct bpf_hdr *h;
	struct ether_header *eh;
	ait_val_t b = AIT_VAL_INIT;

	/* receive connect packet */
	AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
	rlen = read(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
	h = (struct bpf_hdr*) AIT_GET_BUF(&b);
	rlen -= h->bh_hdrlen;
	if (rlen < h->bh_datalen \|\| h->bh_caplen != h->bh_datalen \|\|
	rlen < ETHER_HDR_LEN + sizeof(struct tagRPCCall))
	goto end;
	else {
	rlen = h->bh_caplen;
	eh = (struct ether_header*) (AIT_GET_BUF(&b) + h->bh_hdrlen);
	rlen -= ETHER_HDR_LEN;
	rpc = (struct tagRPCCall*) (eh + 1);

	if (eh->ether_type != ntohs(RPC_DEFPORT))
	goto end;
	else
	e_getlinkbymac((const ether_addr_t*) eh->ether_shost, &sa);
	}
	if (rlen < sizeof(struct tagRPCCall))
	goto end;
	else
	len = ntohl(rpc->call_len);
	if (rlen < len \|\| len > srv->srv_netbuf)
	goto end;

	#ifdef CHECK_ETHACK
	/* skip loop packet */
	if (rpc->call_io & RPC_ACK)
	goto end;
	#endif

	/* check integrity of packet */
	crc = ntohs(rpc->call_crc);
	rpc->call_crc ^= rpc->call_crc;
	if (crc != crcFletcher16((u_short*) rpc, len / 2))
	goto end;

	/* check RPC packet session info */
	if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session))
	goto end;

	c = _allocClient(srv, &sa);
	if (!c) {
	EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
	usleep(2000); /* blocked client delay */
	goto end;
	} else {
	memcpy(AIT_GET_BUF(&c->cli_buf), rpc, len);
	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);

	c->cli_sock = TASK_FD(task);
	memcpy(&c->cli_sa, &sa, sizeof c->cli_sa);

	/* armed timer for close stateless connection */
	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
	schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
	c, ts, c, 0);
	}

	noreply = RPC_CHK_NOREPLY(rpc);

	/* execute RPC call */
	schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);

	/* send RPC reply */
	if (!noreply)
	schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
	c, TASK_FD(task), rpc, len);
end:	end:
	AIT_FREE_VAL(&b);
schedReadSelf(task);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}


	static void *
	txEXTPacket(sched_task_t *task)
	{
	rpc_cli_t *c = TASK_ARG(task);
	rpc_srv_t *s = c->cli_parent;
	rpc_func_t *f = NULL;
	u_char *buf = AIT_GET_BUF(&c->cli_buf);
	struct tagRPCCall rpc = (struct tagRPCCall) buf;
	int ret, wlen = sizeof(struct tagRPCCall);
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
	schedTimer(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), ts, TASK_ARG(task), 0);

	if (rpc->call_argc) {
	f = rpc_srv_getCall(s, ntohs(rpc->call_tag));
	if (!f) {
	rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
	rpc_SetErr(EMSGSIZE, "Message too long");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else {
	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
	/* Go Encapsulate variables */
	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
	RPC_RETVARS(c));
	if (ret == -1) {
	rpc_SetErr(EBADRPC, "Prepare RPC packet failed");

	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	} else
	wlen += ret;
	}
	}

	/* Free return values */
	ait_freeVars(&RPC_RETVARS(c));

	rpc->call_len = htonl(wlen);
	rpc->call_io = RPC_ACK;

	/* send reply */
	ret = write(TASK_FD(task), buf, wlen);
	if (ret == -1) {
	/* close connection */
	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), 0, NULL, 0);
	}

	taskExit(task, NULL);
	}

	static void *
	rxEXTPacket(sched_task_t *task)
	{
	rpc_srv_t *srv = TASK_ARG(task);
	rpc_cli_t *c = NULL;
	int len, noreply = 0, rlen = AIT_LEN(&c->cli_buf);
	struct tagRPCCall *rpc;
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
	sockaddr_t sa;
	ait_val_t b = AIT_VAL_INIT;

	memset(&sa, 0, sizeof sa);
	/* receive connect packet */
	AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
	rlen = read(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
	rpc = (struct tagRPCCall*) AIT_GET_BUF(&b);
	if (rlen < sizeof(struct tagRPCCall))
	goto end;
	else
	len = ntohl(rpc->call_len);
	if (rlen < len \|\| len > srv->srv_netbuf)
	goto end;

	/* skip loop packet */
	if (rpc->call_io & RPC_ACK)
	goto end;

	/* check RPC packet session info */
	if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session))
	goto end;

	c = _allocClient(srv, &sa);
	if (!c) {
	EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
	usleep(2000); /* blocked client delay */
	goto end;
	} else {
	memcpy(AIT_GET_BUF(&c->cli_buf), AIT_GET_BUF(&b), len);
	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);

	c->cli_sock = TASK_FD(task);

	/* armed timer for close stateless connection */
	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, c, NULL);
	schedTimer(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
	c, ts, c, 0);
	}

	noreply = RPC_CHK_NOREPLY(rpc);

	/* execute RPC call */
	schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);

	/* send RPC reply */
	if (!noreply)
	schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
	c, TASK_FD(task), rpc, len);
	end:
	AIT_FREE_VAL(&b);
	schedReadSelf(task);
	taskExit(task, NULL);
	}

/* ------------------------------------------------------ */	/* ------------------------------------------------------ */

inline void	void
rpc_freeBLOBCli(rpc_cli_t * __restrict c)	rpc_freeBLOBCli(rpc_cli_t * __restrict c)
{	{
rpc_srv_t *s = c->cli_parent;	rpc_srv_t *s = c->cli_parent;
Line 531 rpc_freeBLOBCli(rpc_cli_t * __restrict c)	Line 1043 rpc_freeBLOBCli(rpc_cli_t * __restrict c)
/* free buffer */	/* free buffer */
AIT_FREE_VAL(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);

io_arrayDel(s->srv_blob.clients, c->cli_id, 0);	array_Del(s->srv_blob.clients, c->cli_id, 0);
if (c)	if (c)
io_free(c);	e_free(c);
}	}


Line 547 closeBLOBClient(sched_task_t *task)	Line 1059 closeBLOBClient(sched_task_t *task)
/* close client socket */	/* close client socket */
shutdown(sock, SHUT_RDWR);	shutdown(sock, SHUT_RDWR);
close(sock);	close(sock);
return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 555 txBLOB(sched_task_t *task)	Line 1067 txBLOB(sched_task_t *task)
{	{
rpc_cli_t *c = TASK_ARG(task);	rpc_cli_t *c = TASK_ARG(task);
u_char *buf = AIT_GET_BUF(&c->cli_buf);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagBLOBHdr blob = (struct tagBLOBHdr ) buf;
int wlen = sizeof(struct tagBLOBHdr);	int wlen = sizeof(struct tagBLOBHdr);

/* calculate CRC */
blob->hdr_crc ^= blob->hdr_crc;
blob->hdr_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));

/* send reply */	/* send reply */
wlen = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);	wlen = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);
if (wlen == -1 \|\| wlen != sizeof(struct tagBLOBHdr)) {	if (wlen == -1 \|\| wlen != sizeof(struct tagBLOBHdr)) {
Line 569 txBLOB(sched_task_t *task)	Line 1076 txBLOB(sched_task_t *task)
schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);	schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
}	}

return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 580 rxBLOB(sched_task_t *task)	Line 1087 rxBLOB(sched_task_t *task)
rpc_blob_t *b;	rpc_blob_t *b;
struct tagBLOBHdr blob;	struct tagBLOBHdr blob;
int rlen;	int rlen;
u_short crc;

memset(&blob, 0, sizeof blob);	memset(&blob, 0, sizeof blob);
rlen = recv(TASK_FD(task), &blob, sizeof blob, 0);	rlen = recv(TASK_FD(task), &blob, sizeof blob, 0);
if (rlen < 1) {	if (rlen < 1) {
/* close blob connection */	/* close blob connection */
schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);	schedEvent(TASK_ROOT(task), closeBLOBClient, c, 42, NULL, 0);
return NULL;	taskExit(task, NULL);
}	}

/* check BLOB packet */	/* check BLOB packet */
Line 595 rxBLOB(sched_task_t *task)	Line 1101 rxBLOB(sched_task_t *task)
rpc_SetErr(ERPCMISMATCH, "Short BLOB packet");	rpc_SetErr(ERPCMISMATCH, "Short BLOB packet");

schedReadSelf(task);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}

/* check integrity of packet */
crc = ntohs(blob.hdr_crc);
blob.hdr_crc ^= blob.hdr_crc;
if (crc != crcFletcher16((u_short*) &blob, rlen / 2)) {
rpc_SetErr(ERPCMISMATCH, "Bad CRC BLOB packet");

schedReadSelf(task);
return NULL;
}

/* check RPC packet session info */	/* check RPC packet session info */
if ((crc = rpc_chkPktSession(&blob.hdr_session, &s->srv_session))) {	if (rpc_chkPktSession(&blob.hdr_session, &s->srv_session)) {
rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");	rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
blob.hdr_cmd = error;	blob.hdr_cmd = error;
goto end;	goto end;
Line 621 rxBLOB(sched_task_t *task)	Line 1117 rxBLOB(sched_task_t *task)
if (!(b = rpc_srv_getBLOB(s, ntohl(blob.hdr_var)))) {	if (!(b = rpc_srv_getBLOB(s, ntohl(blob.hdr_var)))) {
rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob.hdr_var));	rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob.hdr_var));
blob.hdr_cmd = no;	blob.hdr_cmd = no;
blob.hdr_ret = RPC_ERROR(-1);	RPC_SET_BLOB_RET(&blob, -1);
break;	break;
} else	} else
blob.hdr_len = htonl(b->blob_len);	blob.hdr_len = htonl(b->blob_len);
Line 632 rxBLOB(sched_task_t *task)	Line 1128 rxBLOB(sched_task_t *task)
rpc_srv_blobUnmap(b);	rpc_srv_blobUnmap(b);
} else {	} else {
blob.hdr_cmd = error;	blob.hdr_cmd = error;
blob.hdr_ret = RPC_ERROR(-1);	RPC_SET_BLOB_RET(&blob, -1);
}	}
break;	break;
case set:	case set:
if ((b = rpc_srv_registerBLOB(s, ntohl(blob.hdr_len)))) {	if ((b = rpc_srv_registerBLOB(s, ntohl(blob.hdr_len),
	ntohl(blob.hdr_ret)))) {
/* set new BLOB variable for reply :) */	/* set new BLOB variable for reply :) */
blob.hdr_var = htonl(b->blob_var);	blob.hdr_var = htonl(b->blob_var);

Line 645 rxBLOB(sched_task_t *task)	Line 1142 rxBLOB(sched_task_t *task)
rpc_srv_blobUnmap(b);	rpc_srv_blobUnmap(b);
} else {	} else {
blob.hdr_cmd = error;	blob.hdr_cmd = error;
blob.hdr_ret = RPC_ERROR(-1);	RPC_SET_BLOB_RET(&blob, -1);
}	}
break;	break;
case unset:	case unset:
if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) {	if (rpc_srv_unregisterBLOB(s, ntohl(blob.hdr_var)) == -1) {
blob.hdr_cmd = error;	blob.hdr_cmd = error;
blob.hdr_ret = RPC_ERROR(-1);	RPC_SET_BLOB_RET(&blob, -1);
}	}
break;	break;
default:	default:
rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);	rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);
blob.hdr_cmd = error;	blob.hdr_cmd = error;
blob.hdr_ret = RPC_ERROR(-1);	RPC_SET_BLOB_RET(&blob, -1);
}	}

end:	end:
memcpy(AIT_ADDR(&c->cli_buf), &blob, sizeof blob);	memcpy(AIT_ADDR(&c->cli_buf), &blob, sizeof blob);
schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), NULL, 0);	schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task), NULL, 0);
schedReadSelf(task);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
	flushBLOB(sched_task_t *task)
	{
	#ifdef atomic_load_acq_ptr
	uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg);
	#else
	uintptr_t sigArg = ((volatile uintptr_t) &_glSigArg);
	#endif
	rpc_srv_t srv = sigArg ? (void) sigArg : TASK_ARG(task);
	rpc_blob_t b, tmp;

	TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
	TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);

	rpc_srv_blobFree(srv, b);
	e_free(b);
	}

	if (!schedSignalSelf(task)) {
	/* disabled kqueue support in libaitsched */
	struct sigaction sa;

	memset(&sa, 0, sizeof sa);
	sigemptyset(&sa.sa_mask);
	sa.sa_handler = (void (*)(int)) flushBLOB;
	sa.sa_flags = SA_RESTART \| SA_RESETHAND;
	sigaction(SIGFBLOB, &sa, NULL);
	}

	taskExit(task, NULL);
	}

	static void *
acceptBLOBClients(sched_task_t *task)	acceptBLOBClients(sched_task_t *task)
{	{
rpc_srv_t *srv = TASK_ARG(task);	rpc_srv_t *srv = TASK_ARG(task);
rpc_cli_t *c = NULL;	rpc_cli_t *c = NULL;
register int i;	register int i;
socklen_t salen = sizeof(io_sockaddr_t);	socklen_t salen = sizeof(sockaddr_t);
	int sock;
#ifdef TCP_NOPUSH	#ifdef TCP_NOPUSH
int n = 1;	int n = 1;
#endif	#endif

/* check free slots for connect */	/* check free slots for connect */
for (i = 0; i < io_arraySize(srv->srv_blob.clients) &&	for (i = 0; i < array_Size(srv->srv_blob.clients) &&
(c = io_array(srv->srv_blob.clients, i, rpc_cli_t*)); i++);	(c = array(srv->srv_blob.clients, i, rpc_cli_t*)); i++);
if (c) /* no more free slots! */	if (c) { /* no more free slots! */
	EVERBOSE(1, "BLOB client quota exceeded! Connection will be shutdown!\n");
	if ((sock = accept(TASK_FD(task), NULL, NULL)) != -1) {
	shutdown(sock, SHUT_RDWR);
	close(sock);
	}
goto end;	goto end;
c = io_malloc(sizeof(rpc_cli_t));	}

	c = e_malloc(sizeof(rpc_cli_t));
if (!c) {	if (!c) {
LOGERR;	LOGERR;
srv->srv_kill = srv->srv_blob.kill = 1;	srv->srv_kill = srv->srv_blob.kill = 1;
return NULL;	taskExit(task, NULL);
} else {	} else {
memset(c, 0, sizeof(rpc_cli_t));	memset(c, 0, sizeof(rpc_cli_t));
io_arraySet(srv->srv_blob.clients, i, c);	array_Set(srv->srv_blob.clients, i, c);
c->cli_id = i;	c->cli_id = i;
c->cli_parent = srv;	c->cli_parent = srv;
}	}

/* alloc empty buffer */	/* alloc empty buffer */
AIT_SET_BUF2(&c->cli_buf, 0, srv->srv_netbuf);	AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);

/* accept client */	/* accept client */
c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);	c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
if (c->cli_sock == -1) {	if (c->cli_sock == -1) {
LOGERR;	LOGERR;
AIT_FREE_VAL(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);
io_arrayDel(srv->srv_blob.clients, i, 42);	array_Del(srv->srv_blob.clients, i, 42);
goto end;	goto end;
} else {	} else {
#ifdef TCP_NOPUSH	#ifdef TCP_NOPUSH
Line 715 acceptBLOBClients(sched_task_t *task)	Line 1252 acceptBLOBClients(sched_task_t *task)
schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0);	schedRead(TASK_ROOT(task), rxBLOB, c, c->cli_sock, NULL, 0);
end:	end:
schedReadSelf(task);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}

/* ------------------------------------------------------ */	/* ------------------------------------------------------ */
Line 750 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 1287 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s

srv->srv_blob.server.cli_parent = srv;	srv->srv_blob.server.cli_parent = srv;

memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(io_sockaddr_t));	memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof(sockaddr_t));
switch (srv->srv_blob.server.cli_sa.sa.sa_family) {	switch (srv->srv_blob.server.cli_sa.sa.sa_family) {
case AF_INET:	case AF_INET:
srv->srv_blob.server.cli_sa.sin.sin_port =	srv->srv_blob.server.cli_sa.sin.sin_port =
Line 807 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 1344 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s


/* allocate pool for concurent blob clients */	/* allocate pool for concurent blob clients */
srv->srv_blob.clients = io_arrayInit(io_arraySize(srv->srv_clients));	srv->srv_blob.clients = array_Init(array_Size(srv->srv_clients));
if (!srv->srv_blob.clients) {	if (!srv->srv_blob.clients) {
rpc_SetErr(io_GetErrno(), "%s", io_GetError());	rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);
AIT_FREE_VAL(&srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
Line 819 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 1356 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
srv->srv_blob.root = schedBegin();	srv->srv_blob.root = schedBegin();
if (!srv->srv_blob.root) {	if (!srv->srv_blob.root) {
rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());	rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
io_arrayDestroy(&srv->srv_blob.clients);	array_Destroy(&srv->srv_blob.clients);
close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);
AIT_FREE_VAL(&srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
Line 834 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 1371 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
* @srv = RPC Server instance	* @srv = RPC Server instance
* return: none	* return: none
*/	*/
inline void	void
rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
{	{
if (!srv)	if (!srv)
return;	return;

srv->srv_blob.kill = 1;	srv->srv_blob.kill = 1;

	if (srv->srv_blob.server.cli_sa.sa.sa_family == AF_LOCAL)
	unlink(srv->srv_blob.server.cli_sa.sun.sun_path);

	schedEnd(&srv->srv_blob.root);
}	}

/*	/*
Line 862 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 1404 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
return -1;	return -1;
}	}

if (listen(srv->srv_blob.server.cli_sock, io_arraySize(srv->srv_blob.clients)) == -1) {	if (listen(srv->srv_blob.server.cli_sock, array_Size(srv->srv_blob.clients)) == -1) {
LOGERR;	LOGERR;
return -1;	return -1;
}	}

	if (!schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0)) {
	/* disabled kqueue support in libaitsched */
	struct sigaction sa;

	#ifdef atomic_store_rel_ptr
	atomic_store_rel_ptr(&_glSigArg, (uintptr_t) srv);
	#else
	((volatile uintptr_t) &_glSigArg) = (uintptr_t) srv;
	#endif

	memset(&sa, 0, sizeof sa);
	sigemptyset(&sa.sa_mask);
	sa.sa_handler = (void (*)(int)) flushBLOB;
	sa.sa_flags = SA_RESTART \| SA_RESETHAND;
	sigaction(SIGFBLOB, &sa, NULL);
	}

if (!schedRead(srv->srv_blob.root, acceptBLOBClients, srv,	if (!schedRead(srv->srv_blob.root, acceptBLOBClients, srv,
srv->srv_blob.server.cli_sock, NULL, 0)) {	srv->srv_blob.server.cli_sock, NULL, 0)) {
rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());	rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
Line 877 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 1436 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
/* main rpc loop */	/* main rpc loop */
schedRun(srv->srv_blob.root, &srv->srv_blob.kill);	schedRun(srv->srv_blob.root, &srv->srv_blob.kill);

	/* detach blobs */
	TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
	TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);

	rpc_srv_blobFree(srv, b);
	e_free(b);
	}

/* close all clients connections & server socket */	/* close all clients connections & server socket */
for (i = 0; i < io_arraySize(srv->srv_blob.clients); i++) {	for (i = 0; i < array_Size(srv->srv_blob.clients); i++) {
c = io_array(srv->srv_blob.clients, i, rpc_cli_t*);	c = array(srv->srv_blob.clients, i, rpc_cli_t*);
if (c) {	if (c) {
shutdown(c->cli_sock, SHUT_RDWR);	shutdown(c->cli_sock, SHUT_RDWR);
close(c->cli_sock);	close(c->cli_sock);
Line 887 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 1454 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
schedCancelby(srv->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);	schedCancelby(srv->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);
AIT_FREE_VAL(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);
}	}
io_arrayDel(srv->srv_blob.clients, i, 42);	array_Del(srv->srv_blob.clients, i, 42);
}	}
io_arrayDestroy(&srv->srv_blob.clients);	array_Destroy(&srv->srv_blob.clients);

close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);

/* detach blobs */
TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
TAILQ_REMOVE(&srv->srv_blob.blobs, b, blob_node);

rpc_srv_blobFree(srv, b);
io_free(b);
}

schedEnd(&srv->srv_blob.root);
AIT_FREE_VAL(&srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return 0;	return 0;
}	}
Line 910 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 1468 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
/*	/*
* rpc_srv_initServer() - Init & create RPC Server	* rpc_srv_initServer() - Init & create RPC Server
*	*
* @regProgID = ProgramID for authentication & recognition	* @InstID = Instance for authentication & recognition
* @regProcID = ProcessID for authentication & recognition
* @concurentClients = Concurent clients at same time to this server	* @concurentClients = Concurent clients at same time to this server
* @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)	* @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
* @csHost = Host name or address for bind server, if NULL any address	* @csHost = Host name or address for bind server, if NULL any address
Line 920 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 1477 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
* return: NULL == error or !=NULL bind and created RPC server instance	* return: NULL == error or !=NULL bind and created RPC server instance
*/	*/
rpc_srv_t *	rpc_srv_t *
rpc_srv_initServer(u_int regProgID, u_char regProcID, int concurentClients,	rpc_srv_initServer(u_char InstID, int concurentClients, int netBuf,
int netBuf, const char *csHost, u_short Port, int proto)	const char *csHost, u_short Port, int proto)
{	{
int n = 1;	int n = 1;
rpc_srv_t *srv = NULL;	rpc_srv_t *srv = NULL;
io_sockaddr_t sa = IO_SOCKADDR_INIT;	sockaddr_t sa = E_SOCKADDR_INIT;

if (!concurentClients \|\| !regProgID \|\| (proto < 0 \|\| proto > SOCK_DGRAM)) {	if (!concurentClients \|\| (proto < 0 \|\| proto > SOCK_RAW)) {
rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");	rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
return NULL;	return NULL;
}	}
if (!io_gethostbyname(csHost, Port, &sa))	if (!Port && proto < SOCK_RAW)
return NULL;
if (!Port)
Port = RPC_DEFPORT;	Port = RPC_DEFPORT;
	if (!e_gethostbyname(csHost, Port, &sa))
	return NULL;
if (!proto)	if (!proto)
proto = SOCK_STREAM;	proto = SOCK_STREAM;
if (netBuf < RPC_MIN_BUFSIZ)	if (netBuf < RPC_MIN_BUFSIZ)
netBuf = BUFSIZ;	netBuf = BUFSIZ;
else	else
netBuf = io_align(netBuf, 2); /* align netBuf length */	netBuf = E_ALIGN(netBuf, 2); /* align netBuf length */

#ifdef HAVE_SRANDOMDEV	#ifdef HAVE_SRANDOMDEV
srandomdev();	srandomdev();
Line 950 rpc_srv_initServer(u_int regProgID, u_char regProcID,	Line 1507 rpc_srv_initServer(u_int regProgID, u_char regProcID,
srandom((time(&tim) ^ getpid()));	srandom((time(&tim) ^ getpid()));
#endif	#endif

srv = io_malloc(sizeof(rpc_srv_t));	srv = e_malloc(sizeof(rpc_srv_t));
if (!srv) {	if (!srv) {
LOGERR;	LOGERR;
return NULL;	return NULL;
Line 960 rpc_srv_initServer(u_int regProgID, u_char regProcID,	Line 1517 rpc_srv_initServer(u_int regProgID, u_char regProcID,
srv->srv_proto = proto;	srv->srv_proto = proto;
srv->srv_netbuf = netBuf;	srv->srv_netbuf = netBuf;
srv->srv_session.sess_version = RPC_VERSION;	srv->srv_session.sess_version = RPC_VERSION;
srv->srv_session.sess_program = regProgID;	srv->srv_session.sess_instance = InstID;
srv->srv_session.sess_process = regProcID;

srv->srv_server.cli_parent = srv;	srv->srv_server.cli_parent = srv;
memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);	memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);
Line 976 rpc_srv_initServer(u_int regProgID, u_char regProcID,	Line 1532 rpc_srv_initServer(u_int regProgID, u_char regProcID,
if (!srv->srv_root) {	if (!srv->srv_root) {
rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());	rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
pthread_mutex_destroy(&srv->srv_funcs.mtx);	pthread_mutex_destroy(&srv->srv_funcs.mtx);
io_free(srv);	e_free(srv);
return NULL;	return NULL;
}	}

/* init pool for clients */	/* init pool for clients */
srv->srv_clients = io_arrayInit(concurentClients);	srv->srv_clients = array_Init(concurentClients);
if (!srv->srv_clients) {	if (!srv->srv_clients) {
rpc_SetErr(io_GetErrno(), "%s", io_GetError());	rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
schedEnd(&srv->srv_root);	schedEnd(&srv->srv_root);
pthread_mutex_destroy(&srv->srv_funcs.mtx);	pthread_mutex_destroy(&srv->srv_funcs.mtx);
io_free(srv);	e_free(srv);
return NULL;	return NULL;
}	}

/* create server socket */	/* create server socket */
srv->srv_server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, srv->srv_proto, 0);	srv->srv_server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family,
	srv->srv_proto, srv->srv_proto == SOCK_RAW ? IPPROTO_ERPC : 0);
if (srv->srv_server.cli_sock == -1) {	if (srv->srv_server.cli_sock == -1) {
LOGERR;	LOGERR;
io_arrayDestroy(&srv->srv_clients);	array_Destroy(&srv->srv_clients);
schedEnd(&srv->srv_root);	schedEnd(&srv->srv_root);
pthread_mutex_destroy(&srv->srv_funcs.mtx);	pthread_mutex_destroy(&srv->srv_funcs.mtx);
io_free(srv);	e_free(srv);
return NULL;	return NULL;
}	}
if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
LOGERR;	LOGERR;
goto err;	goto err;
}	}
	if (srv->srv_proto == SOCK_STREAM)
	setsockopt(srv->srv_server.cli_sock, IPPROTO_TCP, TCP_NODELAY, &n, sizeof n);
n = srv->srv_netbuf;	n = srv->srv_netbuf;
if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
LOGERR;	LOGERR;
Line 1026 rpc_srv_initServer(u_int regProgID, u_char regProcID,	Line 1585 rpc_srv_initServer(u_int regProgID, u_char regProcID,
return srv;	return srv;
err: /* error condition */	err: /* error condition */
close(srv->srv_server.cli_sock);	close(srv->srv_server.cli_sock);
io_arrayDestroy(&srv->srv_clients);	array_Destroy(&srv->srv_clients);
schedEnd(&srv->srv_root);	schedEnd(&srv->srv_root);
pthread_mutex_destroy(&srv->srv_funcs.mtx);	pthread_mutex_destroy(&srv->srv_funcs.mtx);
io_free(srv);	e_free(srv);
return NULL;	return NULL;
}	}

Line 1039 err: /* error condition */	Line 1598 err: /* error condition */
* @psrv = RPC Server instance	* @psrv = RPC Server instance
* return: none	* return: none
*/	*/
inline void	void
rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
{	{
if (!psrv \|\| !*psrv)	if (!psrv \|\| !*psrv)
return;	return;

/* if send kill to blob server */	/* if send kill to blob server */
if (!(*psrv)->srv_blob.kill)	rpc_srv_endBLOBServer(*psrv);
rpc_srv_endBLOBServer(*psrv);	/* wait for BLOB server done */
	while ((&(psrv)->srv_blob.root))
	usleep(1000);

(*psrv)->srv_kill = 1;	(*psrv)->srv_kill = 1;
sleep(RPC_SCHED_POLLING);	sleep(RPC_SCHED_POLLING);

	if ((*psrv)->srv_server.cli_sa.sa.sa_family == AF_LOCAL)
	unlink((*psrv)->srv_server.cli_sa.sun.sun_path);

	schedEnd(&(*psrv)->srv_root);

pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx);	pthread_mutex_destroy(&(*psrv)->srv_funcs.mtx);
io_free(*psrv);	e_free(*psrv);
*psrv = NULL;	*psrv = NULL;
}	}

Line 1077 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 1643 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
}	}

if (srv->srv_proto == SOCK_STREAM)	if (srv->srv_proto == SOCK_STREAM)
if (listen(srv->srv_server.cli_sock,	if (listen(srv->srv_server.cli_sock, array_Size(srv->srv_clients)) == -1) {
io_arraySize(srv->srv_clients)) == -1) {
LOGERR;	LOGERR;
return -1;	return -1;
}	}
Line 1094 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 1659 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
schedRun(srv->srv_root, &srv->srv_kill);	schedRun(srv->srv_root, &srv->srv_kill);

/* close all clients connections & server socket */	/* close all clients connections & server socket */
for (i = 0; i < io_arraySize(srv->srv_clients); i++) {	for (i = 0; i < array_Size(srv->srv_clients); i++) {
c = io_array(srv->srv_clients, i, rpc_cli_t*);	c = array(srv->srv_clients, i, rpc_cli_t*);
if (c) {	if (c) {
shutdown(c->cli_sock, SHUT_RDWR);	if (srv->srv_proto == SOCK_STREAM) {
close(c->cli_sock);	shutdown(c->cli_sock, SHUT_RDWR);
	close(c->cli_sock);
	}

schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);	schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
io_freeVars(&RPC_RETVARS(c));	ait_freeVars(&RPC_RETVARS(c));
AIT_FREE_VAL(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);
}	}
io_arrayDel(srv->srv_clients, i, 42);	array_Del(srv->srv_clients, i, 42);
}	}
io_arrayDestroy(&srv->srv_clients);	array_Destroy(&srv->srv_clients);

close(srv->srv_server.cli_sock);	if (srv->srv_proto != SOCK_EXT)
	close(srv->srv_server.cli_sock);

/* detach exported calls */	/* detach exported calls */
RPC_FUNCS_LOCK(&srv->srv_funcs);	RPC_FUNCS_LOCK(&srv->srv_funcs);
Line 1116 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 1684 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);	SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);

AIT_FREE_VAL(&f->func_name);	AIT_FREE_VAL(&f->func_name);
io_free(f);	e_free(f);
}	}
srv->srv_funcs.avlh_root = NULL;	srv->srv_funcs.avlh_root = NULL;
RPC_FUNCS_UNLOCK(&srv->srv_funcs);	RPC_FUNCS_UNLOCK(&srv->srv_funcs);

schedEnd(&srv->srv_root);
return 0;	return 0;
}	}

Line 1148 rpc_srv_execCall(rpc_cli_t * __restrict cli, struct ta	Line 1715 rpc_srv_execCall(rpc_cli_t * __restrict cli, struct ta

func = AIT_GET_LIKE(&funcname, rpc_callback_t);	func = AIT_GET_LIKE(&funcname, rpc_callback_t);
return func(cli, rpc, args);	return func(cli, rpc, args);
	}


	/*
	* rpc_srv_initServer2() - Init & create layer2 RPC Server
	*
	* @InstID = Instance for authentication & recognition
	* @concurentClients = Concurent clients at same time to this server
	* @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
	* @csIface = Interface name for bind server, if NULL first interface on host
	* return: NULL == error or !=NULL bind and created RPC server instance
	*/
	rpc_srv_t *
	rpc_srv_initServer2(u_char InstID, int concurentClients, int netBuf, const char *csIface)
	{
	int n = 1;
	rpc_srv_t *srv = NULL;
	sockaddr_t sa = E_SOCKADDR_INIT;
	char szIface[64], szStr[STRSIZ];
	register int i;
	struct ifreq ifr;
	struct bpf_insn insns[] = {
	BPF_STMT(BPF_LD + BPF_H + BPF_ABS, 12),
	BPF_JUMP(BPF_JMP + BPF_JEQ + BPF_K, RPC_DEFPORT, 0, 1),
	BPF_STMT(BPF_RET + BPF_K, -1),
	BPF_STMT(BPF_RET + BPF_K, 0),
	};
	struct bpf_program fcode = {
	.bf_len = sizeof(insns) / sizeof(struct bpf_insn),
	.bf_insns = insns
	};

	if (!concurentClients) {
	rpc_SetErr(EINVAL, "Invalid parameters can`t init RPC server");
	return NULL;
	}
	if (!csIface) {
	if (e_get1stiface(szIface, sizeof szIface))
	return NULL;
	} else
	strlcpy(szIface, csIface, sizeof szIface);
	if (!e_getifacebyname(szIface, &sa))
	return NULL;

	#ifdef HAVE_SRANDOMDEV
	srandomdev();
	#else
	time_t tim;

	srandom((time(&tim) ^ getpid()));
	#endif

	srv = e_malloc(sizeof(rpc_srv_t));
	if (!srv) {
	LOGERR;
	return NULL;
	} else
	memset(srv, 0, sizeof(rpc_srv_t));

	srv->srv_proto = SOCK_BPF;
	srv->srv_netbuf = netBuf;
	srv->srv_session.sess_version = RPC_VERSION;
	srv->srv_session.sess_instance = InstID;

	srv->srv_server.cli_parent = srv;
	memcpy(&srv->srv_server.cli_sa, &sa, sizeof srv->srv_server.cli_sa);

	/* init functions */
	pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
	SLIST_INIT(&srv->srv_funcs);
	AVL_INIT(&srv->srv_funcs);

	/* init scheduler */
	srv->srv_root = schedBegin();
	if (!srv->srv_root) {
	rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
	pthread_mutex_destroy(&srv->srv_funcs.mtx);
	e_free(srv);
	return NULL;
	}

	/* init pool for clients */
	srv->srv_clients = array_Init(concurentClients);
	if (!srv->srv_clients) {
	rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
	schedEnd(&srv->srv_root);
	pthread_mutex_destroy(&srv->srv_funcs.mtx);
	e_free(srv);
	return NULL;
	}

	/* create server handler */
	for (i = 0; i < 10; i++) {
	memset(szStr, 0, sizeof szStr);
	snprintf(szStr, sizeof szStr, "/dev/bpf%d", i);
	srv->srv_server.cli_sock = open(szStr, O_RDWR);
	if (srv->srv_server.cli_sock > STDERR_FILENO)
	break;
	}
	if (srv->srv_server.cli_sock < 3) {
	LOGERR;
	array_Destroy(&srv->srv_clients);
	schedEnd(&srv->srv_root);
	pthread_mutex_destroy(&srv->srv_funcs.mtx);
	e_free(srv);
	return NULL;
	}

	if (ioctl(srv->srv_server.cli_sock, BIOCIMMEDIATE, &n) == -1) {
	LOGERR;
	goto err;
	}
	if (ioctl(srv->srv_server.cli_sock, BIOCSETF, &fcode) == -1) {
	LOGERR;
	goto err;
	}
	n = (netBuf < RPC_MIN_BUFSIZ) ? getpagesize() : E_ALIGN(netBuf, 2);
	if (ioctl(srv->srv_server.cli_sock, BIOCSBLEN, &n) == -1) {
	LOGERR;
	goto err;
	} else
	srv->srv_netbuf = n;

	memset(&ifr, 0, sizeof ifr);
	strlcpy(ifr.ifr_name, szIface, sizeof ifr.ifr_name);
	if (ioctl(srv->srv_server.cli_sock, BIOCSETIF, &ifr) == -1) {
	LOGERR;
	goto err;
	} else
	fcntl(srv->srv_server.cli_sock, F_SETFL,
	fcntl(srv->srv_server.cli_sock, F_GETFL) \| O_NONBLOCK);

	rpc_register_srvPing(srv);

	return srv;
	err: /* error condition */
	close(srv->srv_server.cli_sock);
	array_Destroy(&srv->srv_clients);
	schedEnd(&srv->srv_root);
	pthread_mutex_destroy(&srv->srv_funcs.mtx);
	e_free(srv);
	return NULL;
	}

	/*
	* rpc_srv_initServerExt() - Init & create pipe RPC Server
	*
	* @InstID = Instance for authentication & recognition
	* @netBuf = Network buffer length (min:512 bytes), if =0 == BUFSIZ (also meaning max RPC packet)
	* @fd = File descriptor
	* return: NULL == error or !=NULL bind and created RPC server instance
	*/
	rpc_srv_t *
	rpc_srv_initServerExt(u_char InstID, int netBuf, int fd)
	{
	rpc_srv_t *srv = NULL;

	#ifdef HAVE_SRANDOMDEV
	srandomdev();
	#else
	time_t tim;

	srandom((time(&tim) ^ getpid()));
	#endif

	srv = e_malloc(sizeof(rpc_srv_t));
	if (!srv) {
	LOGERR;
	return NULL;
	} else
	memset(srv, 0, sizeof(rpc_srv_t));

	srv->srv_proto = SOCK_EXT;
	srv->srv_netbuf = (netBuf < RPC_MIN_BUFSIZ) ?
	getpagesize() : E_ALIGN(netBuf, 2);
	srv->srv_session.sess_version = RPC_VERSION;
	srv->srv_session.sess_instance = InstID;

	srv->srv_server.cli_parent = srv;
	srv->srv_server.cli_sock = fd;

	/* init functions */
	pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
	SLIST_INIT(&srv->srv_funcs);
	AVL_INIT(&srv->srv_funcs);

	/* init scheduler */
	srv->srv_root = schedBegin();
	if (!srv->srv_root) {
	rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
	pthread_mutex_destroy(&srv->srv_funcs.mtx);
	e_free(srv);
	return NULL;
	}

	/* init pool for clients */
	srv->srv_clients = array_Init(1);
	if (!srv->srv_clients) {
	rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
	schedEnd(&srv->srv_root);
	pthread_mutex_destroy(&srv->srv_funcs.mtx);
	e_free(srv);
	return NULL;
	}

	fcntl(srv->srv_server.cli_sock, F_SETFL,
	fcntl(srv->srv_server.cli_sock, F_GETFL) \| O_NONBLOCK);

	rpc_register_srvPing(srv);

	return srv;
}	}

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>

Removed from v.1.13
changed lines
	Added in v.1.28