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.26.2.6 and 1.32.2.1

version 1.26.2.6, 2015/06/28 21:40:46	version 1.32.2.1, 2025/03/31 12:18:40
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 - 2015	Copyright 2004 - 2025
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 70 static void rxEXTPacket(sched_task_t );	Line 70 static void rxEXTPacket(sched_task_t );
static void txEXTPacket(sched_task_t );	static void txEXTPacket(sched_task_t );

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

/* Global Signal Argument when kqueue support disabled */	/* Global Signal Argument when kqueue support disabled */

static volatile uintptr_t _glSigArg = 0;	static volatile uintptr_t _glSigArg = 0;


void	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;

schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL);	if (s->srv_proto == SOCK_STREAM)
	schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL);

/* free buffer */	/* free buffer */
AIT_FREE_VAL(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);
Line 147 _allocClient(rpc_srv_t * __restrict srv, sockaddr_t *	Line 147 _allocClient(rpc_srv_t * __restrict srv, sockaddr_t *

/* alloc empty buffer */	/* alloc empty buffer */
AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);	AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
	if (!AIT_GET_BUF(&c->cli_buf)) {
	array_Del(srv->srv_clients, n, 0);
	e_free(c);
	c = NULL;
	}
}	}

return c;	return c;
Line 158 freeClient(sched_task_t *task)	Line 163 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 171 closeClient(sched_task_t *task)	Line 176 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 182 txPacket(sched_task_t *task)	Line 187 txPacket(sched_task_t *task)
rpc_func_t *f = NULL;	rpc_func_t *f = NULL;
u_char *buf = AIT_GET_BUF(&c->cli_buf);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int ret, estlen, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
struct pollfd pfd;
#ifdef TCP_SESSION_TIMEOUT	#ifdef TCP_SESSION_TIMEOUT
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

Line 198 txPacket(sched_task_t *task)	Line 202 txPacket(sched_task_t *task)
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 (rpc_pktFreeSpace(c) > s->srv_netbuf) {
/* calc estimated length */	rpc_SetErr(EMSGSIZE, "Message too long");
estlen = ait_resideVars(RPC_RETVARS(c)) + wlen;
if (estlen > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, estlen);
buf = AIT_GET_BUF(&c->cli_buf);
rpc = (struct tagRPCCall*) buf;

	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));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
Line 216 txPacket(sched_task_t *task)	Line 219 txPacket(sched_task_t *task)
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;
}	}
Line 236 txPacket(sched_task_t *task)	Line 239 txPacket(sched_task_t *task)
#endif	#endif

/* send reply */	/* send reply */
pfd.fd = TASK_FD(task);	ret = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);
pfd.events = POLLOUT;	if (ret == -1) {
for (; wlen > 0; wlen -= ret, buf += ret) {	/* close connection */
if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 \|\|	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
pfd.revents & (POLLERR \| POLLHUP \| POLLNVAL)) {	TASK_ARG(task), 0, NULL, 0);
if (ret)
LOGERR;
else
rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
/* close connection */
schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
TASK_ARG(task), 0, NULL, 0);
return NULL;
}
ret = send(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), MSG_NOSIGNAL);
if (ret == -1) {
/* close connection */
schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
TASK_ARG(task), 0, NULL, 0);
return NULL;
}
}	}

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

static void *	static void *
Line 281 execCall(sched_task_t *task)	Line 268 execCall(sched_task_t *task)
rpc_SetErr(ERPCMISMATCH, "#%d - %s", elwix_GetErrno(), elwix_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;
Line 292 execCall(sched_task_t *task)	Line 279 execCall(sched_task_t *task)
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)) {
if (!rpc->call_rep.eno) {	if (!rpc->call_rep.eno) {
LOGERR;	LOGERR;
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	RPC_SET_ERRNO(rpc, rpc_Errno);
}	}
rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
ait_freeVars(&c->cli_vars);	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 */
ait_freeVars(&c->cli_vars);	ait_freeVars(&c->cli_vars);
rpc->call_argc ^= rpc->call_argc;	} else if (rpc->call_io & RPC_REQ) {
} else {
/* reply */	/* reply */
rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
}	}
Line 319 execCall(sched_task_t *task)	Line 305 execCall(sched_task_t *task)
}	}

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

static void *	static void *
Line 327 rxPacket(sched_task_t *task)	Line 313 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, estlen;	int len, noreply = 0, rlen;
#if 0	#if 0
u_short crc;	u_short crc;
#endif	#endif
u_char *buf = AIT_GET_BUF(&c->cli_buf);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall b, rpc = (struct tagRPCCall) buf;
struct pollfd pfd;
#ifdef TCP_SESSION_TIMEOUT	#ifdef TCP_SESSION_TIMEOUT
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

Line 342 rxPacket(sched_task_t *task)	Line 327 rxPacket(sched_task_t *task)
TASK_ARG(task), ts, TASK_ARG(task), 0);	TASK_ARG(task), ts, TASK_ARG(task), 0);
#endif	#endif

memset(buf, 0, sizeof(struct tagRPCCall));	/* prepare rx */
rlen = recv(TASK_FD(task), rpc, sizeof(struct tagRPCCall), MSG_PEEK);	len = recv(TASK_FD(task), &b, sizeof b, MSG_PEEK);
if (rlen < sizeof(struct tagRPCCall)) {	if (len < 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 {	} else if (len == sizeof b)
estlen = ntohl(rpc->call_len);	rlen = ntohl(b.call_len);
if (estlen > AIT_LEN(&c->cli_buf))	else
AIT_RE_BUF(&c->cli_buf, estlen);	goto end;
rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
buf = AIT_GET_BUF(&c->cli_buf);	rlen = recv(TASK_FD(task), buf, rlen, 0);
len = estlen;	if (rlen == -1) {
	/* close connection */
	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), 0, NULL, 0);
	taskExit(task, NULL);
}	}
	if (rlen < sizeof(struct tagRPCCall)) {
	rpc_SetErr(ERPCMISMATCH, "Short RPC packet");

/* get next part of packet */	rpc->call_argc ^= rpc->call_argc;
memset(buf, 0, len);	RPC_SET_RETURN(rpc, -1);
pfd.fd = TASK_FD(task);	RPC_SET_ERRNO(rpc, rpc_Errno);
pfd.events = POLLIN \| POLLPRI;	goto err;
for (; len > 0; len -= rlen, buf += rlen) {	} else
if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 \|\|	len = ntohl(rpc->call_len);
pfd.revents & (POLLERR \| POLLHUP \| POLLNVAL)) {	if (rlen < len \|\| len > AIT_LEN(&c->cli_buf)) {
if (rlen)	rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
LOGERR;
else	rpc->call_argc ^= rpc->call_argc;
rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");	RPC_SET_RETURN(rpc, -1);
schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],	RPC_SET_ERRNO(rpc, rpc_Errno);
TASK_ARG(task), 0, NULL, 0);	goto err;
return NULL;
}
rlen = recv(TASK_FD(task), buf, len, 0);
if (rlen == -1) {
/* close connection */
schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
TASK_ARG(task), 0, NULL, 0);
return NULL;
}
}	}
len = estlen;

/* skip loop packet */	/* skip loop packet */
if (rpc->call_io & RPC_ACK) {	if (rpc->call_io & RPC_ACK) {
schedReadSelf(task);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}

#if 0	#if 0
Line 395 rxPacket(sched_task_t *task)	Line 376 rxPacket(sched_task_t *task)
rpc->call_crc ^= rpc->call_crc;	rpc->call_crc ^= rpc->call_crc;
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {	if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");	rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
return NULL;
	rpc->call_argc ^= rpc->call_argc;
	RPC_SET_RETURN(rpc, -1);
	RPC_SET_ERRNO(rpc, rpc_Errno);
	goto err;
}	}
#endif	#endif

noreply = RPC_CHK_NOREPLY(rpc);

/* check RPC packet session info */	/* check RPC packet session info */
if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {	if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");	rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");

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 {	goto err;
/* execute RPC call */
schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len);
}	}

	noreply = RPC_CHK_NOREPLY(rpc);

	/* execute RPC call */
	schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len);
	err:
/* send RPC reply */	/* send RPC reply */
if (!noreply)	if (!noreply && (rpc->call_io & RPC_REQ))
schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET],	schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET],
TASK_ARG(task), TASK_FD(task), rpc, len);	TASK_ARG(task), TASK_FD(task), rpc, len);
	end:
/* lets get next packet */	/* lets get next packet */
schedReadSelf(task);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}

static void *	static void *
Line 428 acceptClients(sched_task_t *task)	Line 414 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(sockaddr_t);	socklen_t salen = E_SOCKADDR_MAX;
int sock;	int sock;
#ifdef TCP_SESSION_TIMEOUT	#ifdef TCP_SESSION_TIMEOUT
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
Line 451 acceptClients(sched_task_t *task)	Line 437 acceptClients(sched_task_t *task)
AIT_FREE_VAL(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);
array_Del(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);
	fcntl(c->cli_sock, F_SETFD, FD_CLOEXEC);
	}

#ifdef TCP_SESSION_TIMEOUT	#ifdef TCP_SESSION_TIMEOUT
/* armed timer for close stateless connection */	/* armed timer for close stateless connection */
Line 464 acceptClients(sched_task_t *task)	Line 452 acceptClients(sched_task_t *task)
c->cli_sock, NULL, 0);	c->cli_sock, NULL, 0);
end:	end:
schedReadSelf(task);	schedReadSelf(task);
return NULL;	taskExit(task, NULL);
}	}


Line 476 txUDPPacket(sched_task_t *task)	Line 464 txUDPPacket(sched_task_t *task)
rpc_func_t *f = NULL;	rpc_func_t *f = NULL;
u_char *buf = AIT_GET_BUF(&c->cli_buf);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int ret, estlen, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
Line 487 txUDPPacket(sched_task_t *task)	Line 475 txUDPPacket(sched_task_t *task)
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 {	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
/* calc estimated length */	rpc_SetErr(EMSGSIZE, "Message too long");
estlen = ait_resideVars(RPC_RETVARS(c)) + wlen;
if (estlen > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, estlen);
buf = AIT_GET_BUF(&c->cli_buf);
rpc = (struct tagRPCCall*) buf;

	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));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
RPC_RETVARS(c));	RPC_RETVARS(c));
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;
}	}
Line 523 txUDPPacket(sched_task_t *task)	Line 512 txUDPPacket(sched_task_t *task)
rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));

/* send reply */	/* send reply */
ret = sendto(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), 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, e_addrlen(&c->cli_sa));
if (ret == -1) {	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;
}	}

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

static void *	static void *
Line 540 rxUDPPacket(sched_task_t *task)	Line 528 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;
int len, rlen, noreply;	int len, noreply = 0, rlen;
u_short crc;	u_short crc;
struct tagRPCCall *rpc;	struct tagRPCCall *rpc;
sockaddr_t sa;	sockaddr_t sa;
socklen_t salen;	socklen_t salen = E_SOCKADDR_MAX;
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
ait_val_t b = AIT_VAL_INIT;	ait_val_t b = AIT_VAL_INIT;

/* receive connect packet */	/* receive connect packet */
AIT_SET_BUF(&b, NULL, srv->srv_netbuf);	AIT_SET_BUF(&b, NULL, srv->srv_netbuf);
salen = sa.ss.ss_len = sizeof(sockaddr_t);	#ifndef __linux__
	sa.ss.ss_len = salen;
	#endif
rlen = recvfrom(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b), 0, &sa.sa, &salen);	rlen = recvfrom(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b), 0, &sa.sa, &salen);
rpc = (struct tagRPCCall*) AIT_GET_BUF(&b);	rpc = (struct tagRPCCall*) AIT_GET_BUF(&b);
if (rlen < sizeof(struct tagRPCCall) \|\| rlen < ntohl(rpc->call_len)) {	if (!rpc \|\| rlen < sizeof(struct tagRPCCall))
rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
goto end;	goto end;
}	else
	len = ntohl(rpc->call_len);
	if (rlen < len \|\| len > srv->srv_netbuf)
	goto end;

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

	/* check integrity of packet */
	crc = ntohs(rpc->call_crc);
	rpc->call_crc ^= rpc->call_crc;
	if (crc != crcFletcher16((u_short*) AIT_GET_BUF(&b), 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");	EVERBOSE(1, "RPC client quota exceeded!");
usleep(2000); /* blocked client delay */	usleep(2000); /* blocked client delay */
goto end;	goto end;
} else {	} else {
len = ntohl(rpc->call_len);
if (len > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, len);
memcpy(AIT_GET_BUF(&c->cli_buf), AIT_GET_BUF(&b), len);	memcpy(AIT_GET_BUF(&c->cli_buf), AIT_GET_BUF(&b), len);
rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);

Line 583 rxUDPPacket(sched_task_t *task)	Line 582 rxUDPPacket(sched_task_t *task)
c, ts, c, 0);	c, ts, c, 0);
}	}

/* check integrity of packet */
crc = ntohs(rpc->call_crc);
rpc->call_crc ^= rpc->call_crc;
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
/* close connection */
schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
c, 0, NULL, 0);
goto end;
}

noreply = RPC_CHK_NOREPLY(rpc);	noreply = RPC_CHK_NOREPLY(rpc);

/* check RPC packet session info */	/* execute RPC call */
if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {	schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
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, c, (int) noreply, rpc, len);
}

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


Line 627 txRAWPacket(sched_task_t *task)	Line 606 txRAWPacket(sched_task_t *task)
rpc_func_t *f = NULL;	rpc_func_t *f = NULL;
u_char *buf = AIT_GET_BUF(&c->cli_buf);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int ret, estlen, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
Line 638 txRAWPacket(sched_task_t *task)	Line 617 txRAWPacket(sched_task_t *task)
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 {	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
/* calc estimated length */	rpc_SetErr(EMSGSIZE, "Message too long");
estlen = ait_resideVars(RPC_RETVARS(c)) + wlen;
if (estlen > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, estlen);
buf = AIT_GET_BUF(&c->cli_buf);
rpc = (struct tagRPCCall*) buf;

	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));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
RPC_RETVARS(c));	RPC_RETVARS(c));
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;
}	}
Line 674 txRAWPacket(sched_task_t *task)	Line 654 txRAWPacket(sched_task_t *task)
rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));

/* send reply */	/* send reply */
ret = sendto(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), 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, e_addrlen(&c->cli_sa));
if (ret == -1) {	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;
}	}

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

static void *	static void *
Line 691 rxRAWPacket(sched_task_t *task)	Line 670 rxRAWPacket(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;
int len, rlen, noreply;	int len, noreply = 0, rlen;
u_short crc;	u_short crc;
struct tagRPCCall *rpc;	struct tagRPCCall *rpc;
sockaddr_t sa;	sockaddr_t sa;
socklen_t salen;	socklen_t salen = E_SOCKADDR_MAX;
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
ait_val_t b = AIT_VAL_INIT;	ait_val_t b = AIT_VAL_INIT;

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

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

	/* check integrity of packet */
	crc = ntohs(rpc->call_crc);
	rpc->call_crc ^= rpc->call_crc;
	if (crc != crcFletcher16((u_short*) AIT_GET_BUF(&b), 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");	EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
usleep(2000); /* blocked client delay */	usleep(2000); /* blocked client delay */
goto end;	goto end;
} else {	} else {
len = ntohl(rpc->call_len);
if (len > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, len);
memcpy(AIT_GET_BUF(&c->cli_buf), rpc, len);	memcpy(AIT_GET_BUF(&c->cli_buf), rpc, len);
rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);

Line 755 rxRAWPacket(sched_task_t *task)	Line 746 rxRAWPacket(sched_task_t *task)
c, ts, c, 0);	c, ts, c, 0);
}	}

/* check integrity of packet */
crc = ntohs(rpc->call_crc);
rpc->call_crc ^= rpc->call_crc;
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
/* close connection */
schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
c, 0, NULL, 0);
goto end;
}

noreply = RPC_CHK_NOREPLY(rpc);	noreply = RPC_CHK_NOREPLY(rpc);

/* check RPC packet session info */	/* execute RPC call */
if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {	schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
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, c, (int) noreply, rpc, len);
}

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


static void *	static void *
txBPFPacket(sched_task_t *task)	txBPFPacket(sched_task_t *task)
{	{
	#ifndef __linux__
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 = AIT_GET_BUF(&c->cli_buf);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int ret, len, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
struct ether_header *eh;	struct ether_header *eh;
ait_val_t b = AIT_VAL_INIT;	ait_val_t b = AIT_VAL_INIT;
Line 812 txBPFPacket(sched_task_t *task)	Line 784 txBPFPacket(sched_task_t *task)
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 {	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
/* calc estimated length */	rpc_SetErr(EMSGSIZE, "Message too long");
len = ait_resideVars(RPC_RETVARS(c)) + wlen;
if (len > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, len);
buf = AIT_GET_BUF(&c->cli_buf);
rpc = (struct tagRPCCall*) buf;

	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));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
RPC_RETVARS(c));	RPC_RETVARS(c));
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;
}	}
Line 848 txBPFPacket(sched_task_t *task)	Line 821 txBPFPacket(sched_task_t *task)
rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));

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

ret = write(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));	ret = write(TASK_FD(task), AIT_GET_BUF(&b), AIT_LEN(&b));
AIT_FREE_VAL(&b);	AIT_FREE_VAL(&b);
Line 860 txBPFPacket(sched_task_t *task)	Line 833 txBPFPacket(sched_task_t *task)
/* 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;
}	}
	#else
	rpc_SetErr(ENOTSUP, "Feature isn't supported on Linux!");
	#endif

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

static void *	static void *
rxBPFPacket(sched_task_t *task)	rxBPFPacket(sched_task_t *task)
{	{
	#ifndef __linux__
rpc_srv_t *srv = TASK_ARG(task);	rpc_srv_t *srv = TASK_ARG(task);
rpc_cli_t *c = NULL;	rpc_cli_t *c = NULL;
int len, rlen, noreply;	int len, rlen, noreply;
Line 886 rxBPFPacket(sched_task_t *task)	Line 862 rxBPFPacket(sched_task_t *task)
h = (struct bpf_hdr*) AIT_GET_BUF(&b);	h = (struct bpf_hdr*) AIT_GET_BUF(&b);
rlen -= h->bh_hdrlen;	rlen -= h->bh_hdrlen;
if (rlen < h->bh_datalen \|\| h->bh_caplen != h->bh_datalen \|\|	if (rlen < h->bh_datalen \|\| h->bh_caplen != h->bh_datalen \|\|
rlen < ETHER_HDR_LEN + sizeof(struct tagRPCCall)) {	rlen < ETHER_HDR_LEN + sizeof(struct tagRPCCall))
rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
goto end;	goto end;
} else {	else {
rlen = h->bh_caplen;	rlen = h->bh_caplen;
eh = (struct ether_header*) (AIT_GET_BUF(&b) + h->bh_hdrlen);	eh = (struct ether_header*) (AIT_GET_BUF(&b) + h->bh_hdrlen);
rlen -= ETHER_HDR_LEN;	rlen -= ETHER_HDR_LEN;
rpc = (struct tagRPCCall*) (eh + 1);	rpc = (struct tagRPCCall*) (eh + 1);

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

if (eh->ether_type != ntohs(RPC_DEFPORT))	if (eh->ether_type != ntohs(RPC_DEFPORT))
goto end;	goto end;
else	else
e_getlinkbymac((const ether_addr_t*) eh->ether_shost, &sa);	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);	c = _allocClient(srv, &sa);
if (!c) {	if (!c) {
EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");	EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
usleep(2000); /* blocked client delay */	usleep(2000); /* blocked client delay */
goto end;	goto end;
} else {	} else {
len = ntohl(rpc->call_len);
if (len > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, len);
memcpy(AIT_GET_BUF(&c->cli_buf), rpc, len);	memcpy(AIT_GET_BUF(&c->cli_buf), rpc, len);
rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);

Line 928 rxBPFPacket(sched_task_t *task)	Line 916 rxBPFPacket(sched_task_t *task)
c, ts, c, 0);	c, ts, c, 0);
}	}

/* check integrity of packet */
crc = ntohs(rpc->call_crc);
rpc->call_crc ^= rpc->call_crc;
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
/* close connection */
schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
c, 0, NULL, 0);
goto end;
}

noreply = RPC_CHK_NOREPLY(rpc);	noreply = RPC_CHK_NOREPLY(rpc);

/* check RPC packet session info */	/* execute RPC call */
if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {	schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
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, c, (int) noreply, rpc, len);
}

/* send RPC reply */	/* send RPC reply */
if (!noreply)	if (!noreply && (rpc->call_io & RPC_REQ))
schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],	schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
c, TASK_FD(task), rpc, len);	c, TASK_FD(task), rpc, len);
end:	end:
AIT_FREE_VAL(&b);	AIT_FREE_VAL(&b);
schedReadSelf(task);	schedReadSelf(task);
return NULL;	#else
	rpc_SetErr(ENOTSUP, "Feature isn't supported on Linux!");
	#endif

	taskExit(task, NULL);
}	}


Line 972 txEXTPacket(sched_task_t *task)	Line 944 txEXTPacket(sched_task_t *task)
rpc_func_t *f = NULL;	rpc_func_t *f = NULL;
u_char *buf = AIT_GET_BUF(&c->cli_buf);	u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int ret, len, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
Line 983 txEXTPacket(sched_task_t *task)	Line 955 txEXTPacket(sched_task_t *task)
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 {	} else if (rpc_pktFreeSpace(c) > s->srv_netbuf) {
/* calc estimated length */	rpc_SetErr(EMSGSIZE, "Message too long");
len = ait_resideVars(RPC_RETVARS(c)) + wlen;
if (len > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, len);
buf = AIT_GET_BUF(&c->cli_buf);
rpc = (struct tagRPCCall*) buf;

	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));	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
RPC_RETVARS(c));	RPC_RETVARS(c));
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;
}	}
Line 1015 txEXTPacket(sched_task_t *task)	Line 988 txEXTPacket(sched_task_t *task)
rpc->call_io = RPC_ACK;	rpc->call_io = RPC_ACK;

/* send reply */	/* send reply */
ret = write(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf));	ret = write(TASK_FD(task), buf, wlen);
if (ret == -1) {	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;
}	}

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

static void *	static void *
Line 1031 rxEXTPacket(sched_task_t *task)	Line 1003 rxEXTPacket(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;
int len, rlen, noreply;	int len, noreply = 0, rlen;
struct tagRPCCall *rpc;	struct tagRPCCall *rpc;
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
ait_val_t b = AIT_VAL_INIT;
sockaddr_t sa;	sockaddr_t sa;
	ait_val_t b = AIT_VAL_INIT;

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

/* skip loop packet */	/* skip loop packet */
if (rpc->call_io & RPC_ACK)	if (rpc->call_io & RPC_ACK)
goto end;	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");	EVERBOSE(1, "RPC client quota exceeded! Connection will be shutdown!\n");
usleep(2000); /* blocked client delay */	usleep(2000); /* blocked client delay */
goto end;	goto end;
} else {	} else {
len = ntohl(rpc->call_len);	memcpy(AIT_GET_BUF(&c->cli_buf), AIT_GET_BUF(&b), len);
if (len > AIT_LEN(&c->cli_buf))
AIT_RE_BUF(&c->cli_buf, len);
memcpy(AIT_GET_BUF(&c->cli_buf), rpc, AIT_LEN(&c->cli_buf));
rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);

c->cli_sock = TASK_FD(task);	c->cli_sock = TASK_FD(task);
Line 1073 rxEXTPacket(sched_task_t *task)	Line 1048 rxEXTPacket(sched_task_t *task)

noreply = RPC_CHK_NOREPLY(rpc);	noreply = RPC_CHK_NOREPLY(rpc);

/* check RPC packet session info */	/* execute RPC call */
if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {	schedEvent(TASK_ROOT(task), execCall, c, (int) noreply, rpc, len);
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, c, (int) noreply, rpc, len);
}

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

/* ------------------------------------------------------ */	/* ------------------------------------------------------ */
Line 1123 closeBLOBClient(sched_task_t *task)	Line 1089 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 1140 txBLOB(sched_task_t *task)	Line 1106 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 1157 rxBLOB(sched_task_t *task)	Line 1123 rxBLOB(sched_task_t *task)
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 1165 rxBLOB(sched_task_t *task)	Line 1131 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 RPC packet session info */	/* check RPC packet session info */
Line 1181 rxBLOB(sched_task_t *task)	Line 1147 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 1192 rxBLOB(sched_task_t *task)	Line 1158 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:
Line 1206 rxBLOB(sched_task_t *task)	Line 1172 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)	flushBLOB(sched_task_t *task)
{	{
	#ifdef atomic_load_acq_ptr
uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg);	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_srv_t srv = sigArg ? (void) sigArg : TASK_ARG(task);
rpc_blob_t b, tmp;	rpc_blob_t b, tmp;

Line 1242 flushBLOB(sched_task_t *task)	Line 1212 flushBLOB(sched_task_t *task)
e_free(b);	e_free(b);
}	}

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

Line 1251 flushBLOB(sched_task_t *task)	Line 1221 flushBLOB(sched_task_t *task)
sa.sa_handler = (void (*)(int)) flushBLOB;	sa.sa_handler = (void (*)(int)) flushBLOB;
sa.sa_flags = SA_RESTART \| SA_RESETHAND;	sa.sa_flags = SA_RESTART \| SA_RESETHAND;
sigaction(SIGFBLOB, &sa, NULL);	sigaction(SIGFBLOB, &sa, NULL);
	return NULL;
	} else {
	schedSignalSelf(task);
	taskExit(task, NULL);
}	}

return NULL;
}	}

static void *	static void *
Line 1262 acceptBLOBClients(sched_task_t *task)	Line 1234 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(sockaddr_t);	socklen_t salen = E_SOCKADDR_MAX;
int sock;	int sock;
#ifdef TCP_NOPUSH	#ifdef TCP_NOPUSH
int n = 1;	int n = 1;
Line 1284 acceptBLOBClients(sched_task_t *task)	Line 1256 acceptBLOBClients(sched_task_t *task)
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));
array_Set(srv->srv_blob.clients, i, c);	array_Set(srv->srv_blob.clients, i, c);
Line 1307 acceptBLOBClients(sched_task_t *task)	Line 1279 acceptBLOBClients(sched_task_t *task)
setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n);	setsockopt(c->cli_sock, IPPROTO_TCP, TCP_NOPUSH, &n, sizeof n);
#endif	#endif
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);
	fcntl(c->cli_sock, F_SETFD, FD_CLOEXEC);
}	}

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 1329 int	Line 1302 int
rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)	rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_short Port, const char *diskDir)
{	{
int n = 1;	int n = 1;
	socklen_t salen;

if (!srv \|\| srv->srv_kill) {	if (!srv \|\| srv->srv_kill) {
rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");	rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
Line 1347 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 1321 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(sockaddr_t));	memcpy(&srv->srv_blob.server.cli_sa, &srv->srv_server.cli_sa, sizeof srv->srv_blob.server.cli_sa);
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 =
htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin.sin_port) + 1);	htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin.sin_port) + 1);
	salen = sizeof srv->srv_blob.server.cli_sa.sin;
break;	break;
case AF_INET6:	case AF_INET6:
srv->srv_blob.server.cli_sa.sin6.sin6_port =	srv->srv_blob.server.cli_sa.sin6.sin6_port =
htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin6.sin6_port) + 1);	htons(Port ? Port : ntohs(srv->srv_blob.server.cli_sa.sin6.sin6_port) + 1);
	salen = sizeof srv->srv_blob.server.cli_sa.sin6;
break;	break;
case AF_LOCAL:	case AF_LOCAL:
strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob",	strlcat(srv->srv_blob.server.cli_sa.sun.sun_path, ".blob",
sizeof srv->srv_blob.server.cli_sa.sun.sun_path);	sizeof srv->srv_blob.server.cli_sa.sun.sun_path);
	salen = sizeof srv->srv_blob.server.cli_sa.sun;
break;	break;
default:	default:
AIT_FREE_VAL(&srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
Line 1392 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 1369 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
AIT_FREE_VAL(&srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
}	}
if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa,	if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa, salen) == -1) {
srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
LOGERR;	LOGERR;
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);
Line 1439 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 1415 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)

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

schedEnd(&srv->srv_blob.root);

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

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

/*	/*
Line 1473 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 1449 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
/* disabled kqueue support in libaitsched */	/* disabled kqueue support in libaitsched */
struct sigaction sa;	struct sigaction sa;

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

memset(&sa, 0, sizeof sa);	memset(&sa, 0, sizeof sa);
sigemptyset(&sa.sa_mask);	sigemptyset(&sa.sa_mask);
Line 1539 rpc_srv_initServer(u_char InstID, int concurentClients	Line 1519 rpc_srv_initServer(u_char InstID, int concurentClients
int n = 1;	int n = 1;
rpc_srv_t *srv = NULL;	rpc_srv_t *srv = NULL;
sockaddr_t sa = E_SOCKADDR_INIT;	sockaddr_t sa = E_SOCKADDR_INIT;
	socklen_t salen;

if (!concurentClients \|\| (proto < 0 \|\| proto > SOCK_RAW)) {	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");
Line 1546 rpc_srv_initServer(u_char InstID, int concurentClients	Line 1527 rpc_srv_initServer(u_char InstID, int concurentClients
}	}
if (!Port && proto < SOCK_RAW)	if (!Port && proto < SOCK_RAW)
Port = RPC_DEFPORT;	Port = RPC_DEFPORT;
if (!e_gethostbyname(csHost, Port, &sa))	if (!(salen = e_gethostbyname(csHost, Port, &sa)))
return NULL;	return NULL;
if (!proto)	if (!proto)
proto = SOCK_STREAM;	proto = SOCK_STREAM;
Line 1581 rpc_srv_initServer(u_char InstID, int concurentClients	Line 1562 rpc_srv_initServer(u_char InstID, int concurentClients
/* init functions */	/* init functions */
pthread_mutex_init(&srv->srv_funcs.mtx, NULL);	pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
SLIST_INIT(&srv->srv_funcs);	SLIST_INIT(&srv->srv_funcs);
AVL_INIT(&srv->srv_funcs);	RB_INIT(&srv->srv_funcs);

/* init scheduler */	/* init scheduler */
srv->srv_root = schedBegin();	srv->srv_root = schedBegin();
Line 1617 rpc_srv_initServer(u_char InstID, int concurentClients	Line 1598 rpc_srv_initServer(u_char InstID, int concurentClients
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 1626 rpc_srv_initServer(u_char InstID, int concurentClients	Line 1609 rpc_srv_initServer(u_char InstID, int concurentClients
LOGERR;	LOGERR;
goto err;	goto err;
}	}
if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa,	if (bind(srv->srv_server.cli_sock, &srv->srv_server.cli_sa.sa, salen) == -1) {
srv->srv_server.cli_sa.sa.sa_len) == -1) {
LOGERR;	LOGERR;
goto err;	goto err;
} else	} else
Line 1660 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 1642 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)

/* if send kill to blob server */	/* if send kill to blob server */
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);

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

if ((*psrv)->srv_server.cli_sa.sa.sa_family == AF_LOCAL)	if ((*psrv)->srv_server.cli_sa.sa.sa_family == AF_LOCAL)
unlink((*psrv)->srv_server.cli_sa.sun.sun_path);	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);
e_free(*psrv);	e_free(*psrv);
*psrv = NULL;	*psrv = NULL;
Line 1737 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 1722 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
AIT_FREE_VAL(&f->func_name);	AIT_FREE_VAL(&f->func_name);
e_free(f);	e_free(f);
}	}
srv->srv_funcs.avlh_root = NULL;	srv->srv_funcs.rbh_root = NULL;
RPC_FUNCS_UNLOCK(&srv->srv_funcs);	RPC_FUNCS_UNLOCK(&srv->srv_funcs);

return 0;	return 0;
Line 1781 rpc_srv_execCall(rpc_cli_t * __restrict cli, struct ta	Line 1766 rpc_srv_execCall(rpc_cli_t * __restrict cli, struct ta
rpc_srv_t *	rpc_srv_t *
rpc_srv_initServer2(u_char InstID, int concurentClients, int netBuf, const char *csIface)	rpc_srv_initServer2(u_char InstID, int concurentClients, int netBuf, const char *csIface)
{	{
	#ifndef __linux__
int n = 1;	int n = 1;
rpc_srv_t *srv = NULL;	rpc_srv_t *srv = NULL;
sockaddr_t sa = E_SOCKADDR_INIT;	sockaddr_t sa = E_SOCKADDR_INIT;
Line 1836 rpc_srv_initServer2(u_char InstID, int concurentClient	Line 1822 rpc_srv_initServer2(u_char InstID, int concurentClient
/* init functions */	/* init functions */
pthread_mutex_init(&srv->srv_funcs.mtx, NULL);	pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
SLIST_INIT(&srv->srv_funcs);	SLIST_INIT(&srv->srv_funcs);
AVL_INIT(&srv->srv_funcs);	RB_INIT(&srv->srv_funcs);

/* init scheduler */	/* init scheduler */
srv->srv_root = schedBegin();	srv->srv_root = schedBegin();
Line 1907 err: /* error condition */	Line 1893 err: /* error condition */
schedEnd(&srv->srv_root);	schedEnd(&srv->srv_root);
pthread_mutex_destroy(&srv->srv_funcs.mtx);	pthread_mutex_destroy(&srv->srv_funcs.mtx);
e_free(srv);	e_free(srv);
	#else
	rpc_SetErr(ENOTSUP, "Feature isn't supported on Linux!");
	#endif

return NULL;	return NULL;
}	}

Line 1950 rpc_srv_initServerExt(u_char InstID, int netBuf, int f	Line 1940 rpc_srv_initServerExt(u_char InstID, int netBuf, int f
/* init functions */	/* init functions */
pthread_mutex_init(&srv->srv_funcs.mtx, NULL);	pthread_mutex_init(&srv->srv_funcs.mtx, NULL);
SLIST_INIT(&srv->srv_funcs);	SLIST_INIT(&srv->srv_funcs);
AVL_INIT(&srv->srv_funcs);	RB_INIT(&srv->srv_funcs);

/* init scheduler */	/* init scheduler */
srv->srv_root = schedBegin();	srv->srv_root = schedBegin();
Line 1977 rpc_srv_initServerExt(u_char InstID, int netBuf, int f	Line 1967 rpc_srv_initServerExt(u_char InstID, int netBuf, int f
rpc_register_srvPing(srv);	rpc_register_srvPing(srv);

return srv;	return srv;
	}

	/*
	* rpc_srv_Return() - Prepare IPC return answer to RPC client
	*
	* @c = RPC client
	* return: number of arguments in response
	*/
	int
	rpc_srv_Return(rpc_cli_t *c)
	{
	rpc_srv_t *s = c->cli_parent;
	u_char *buf = AIT_GET_BUF(&c->cli_buf);
	struct tagRPCCall rpc = (struct tagRPCCall) buf;

	if (!RPC_CHK_NOREPLY(rpc)) {
	rpc->call_argc = (u_char) array_Size(RPC_RETVARS(c));
	schedWrite(s->srv_root, cbProto[s->srv_proto][CB_TXPACKET], c, c->cli_sock, rpc, 0);
	} else
	rpc->call_argc ^= rpc->call_argc;

	return rpc->call_argc;
}	}

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

Removed from v.1.26.2.6
changed lines
	Added in v.1.32.2.1