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.17.4.1 and 1.23.6.2

version 1.17.4.1, 2013/08/20 23:37:43	version 1.23.6.2, 2014/11/17 23:51:26
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, 2013	Copyright 2004 - 2014
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 59 static void txUDPPacket(sched_task_t );	Line 59 static void txUDPPacket(sched_task_t );

/* SOCK_RAW */	/* SOCK_RAW */

static sched_task_func_t cbProto[SOCK_RAW + 1][4] = {	/* SOCK_BPF */
	static void rxBPFPacket(sched_task_t );
	static void txBPFPacket(sched_task_t );

	static sched_task_func_t cbProto[SOCK_BPF + 1][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, rxUDPPacket, txUDPPacket }, /* SOCK_DGRAM */
{ NULL, NULL, NULL, NULL } /* SOCK_RAW */	{ NULL, NULL, NULL, NULL }, /* SOCK_RAW */
	{ rxBPFPacket, freeClient, rxBPFPacket, txBPFPacket } /* SOCK_BPF */
};	};

	/* Global Signal Argument when kqueue support disabled */

	static volatile uintptr_t _glSigArg = 0;


void	void
rpc_freeCli(rpc_cli_t * __restrict c)	rpc_freeCli(rpc_cli_t * __restrict c)
{	{
Line 74 rpc_freeCli(rpc_cli_t * __restrict c)	Line 83 rpc_freeCli(rpc_cli_t * __restrict c)

schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL);	schedCancelby(s->srv_root, taskMAX, CRITERIA_ARG, c, NULL);

/* free buffer(s) */	/* free buffer */
ait_freeVars(&c->cli_buf);	AIT_FREE_VAL(&c->cli_buf);

array_Del(s->srv_clients, c->cli_id, 0);	array_Del(s->srv_clients, c->cli_id, 0);
if (c)	if (c)
Line 127 _allocClient(rpc_srv_t * __restrict srv, sockaddr_t *	Line 136 _allocClient(rpc_srv_t * __restrict srv, sockaddr_t *
}	}

/* alloc empty buffer */	/* alloc empty buffer */
c->cli_buf = ait_allocVars(1);	AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);
if (!c->cli_buf) {
rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
array_Del(srv->srv_clients, n, 42);
return NULL;
} else
AIT_SET_BUFSIZ(array(c->cli_buf, 0, ait_val_t*), 0, srv->srv_netbuf);
}	}

return c;	return c;
Line 167 txPacket(sched_task_t *task)	Line 170 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, estlen, wlen = sizeof(struct tagRPCCall);
	struct pollfd pfd;
	#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->call_rep.ret = RPC_ERROR(-1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
} else {	} else {
	/* calc estimated length */
	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 = htons(array_Size(RPC_RETVARS(c)));	rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = ait_vars2buffer(buf + wlen, sizeof buf - wlen, RPC_RETVARS(c));	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
	RPC_RETVARS(c));
/* Free return values */	/* Free return values */
ait_freeVars(&c->cli_vars);	ait_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->call_rep.ret = RPC_ERROR(-1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
Line 197 txPacket(sched_task_t *task)	Line 215 txPacket(sched_task_t *task)
}	}
}	}

rpc->call_len = htons(wlen);	rpc->call_len = htonl(wlen);

#if 0	#if 0
/* calculate CRC */	/* calculate CRC */
Line 206 txPacket(sched_task_t *task)	Line 224 txPacket(sched_task_t *task)
#endif	#endif

/* send reply */	/* send reply */
ret = send(TASK_FD(task), buf, wlen, MSG_NOSIGNAL);	pfd.fd = TASK_FD(task);
if (ret == -1 \|\| ret != wlen) {	pfd.events = POLLOUT;
/* close connection */	for (; wlen > 0; wlen -= ret, buf += ret) {
schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],	if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 \|\|
TASK_ARG(task), 0, NULL, 0);	pfd.revents & (POLLERR \| POLLHUP \| POLLNVAL)) {
	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;	return NULL;
Line 223 execCall(sched_task_t *task)	Line 257 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 = ntohs(rpc->call_argc);

/* Go decapsulate variables ... */	/* Go decapsulate variables ... */
if (argc) {	if (argc) {
arr = ait_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", 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->call_rep.ret = RPC_ERROR(-1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
Line 244 execCall(sched_task_t *task)	Line 278 execCall(sched_task_t *task)

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->call_rep.ret = RPC_ERROR(-1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
Line 252 execCall(sched_task_t *task)	Line 287 execCall(sched_task_t *task)
argc = RPC_CHK_NOREPLY(rpc);	argc = RPC_CHK_NOREPLY(rpc);
rpc->call_rep.ret = RPC_ERROR(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->call_rep.eno = RPC_ERROR(rpc_Errno);
	}
rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
} else {	} else {
rpc->call_rep.eno ^= rpc->call_rep.eno;	rpc->call_rep.eno ^= rpc->call_rep.eno;
Line 276 rxPacket(sched_task_t *task)	Line 314 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, rlen, noreply, estlen;
u_short off = TASK_DATLEN(task);
#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 rpc = (struct tagRPCCall) buf;
	struct pollfd pfd;
	#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

	memset(buf, 0, sizeof(struct tagRPCCall));
	rlen = recv(TASK_FD(task), rpc, sizeof(struct tagRPCCall), MSG_PEEK);
	if (rlen < sizeof(struct tagRPCCall)) {
/* 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;
} else {	} else {
rlen += off; /* add reminded bytes from previous rxPacket, if exists! */	estlen = ntohl(rpc->call_len);
off = 0; /* process buffer from start offset == 0 */	if (estlen > AIT_LEN(&c->cli_buf))
	AIT_RE_BUF(&c->cli_buf, estlen);
	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
	buf = AIT_GET_BUF(&c->cli_buf);
	len = estlen;
}	}

do {	/* get next part of packet */
/* check RPC packet */	memset(buf, 0, len);
if (rlen < sizeof(struct tagRPCCall)) {	pfd.fd = TASK_FD(task);
rpc_SetErr(ERPCMISMATCH, "Short RPC packet");	pfd.events = POLLIN \| POLLPRI;
	for (; len > 0; len -= rlen, buf += rlen) {
/* reminder received previous bytes ;) */	if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 \|\|
schedRead(TASK_ROOT(task), TASK_FUNC(task), TASK_ARG(task),	pfd.revents & (POLLERR \| POLLHUP \| POLLNVAL)) {
TASK_FD(task), TASK_DATA(task), rlen);	if (rlen)
	LOGERR;
	else
	rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), 0, NULL, 0);
return NULL;	return NULL;
} else
rpc = (struct tagRPCCall*) (buf + off);

len = ntohs(rpc->call_len);
rlen -= len;

/* check RPC packet lengths */
if (rlen < 0 \|\| len < sizeof(struct tagRPCCall)) {
rpc_SetErr(ERPCMISMATCH, "Broken RPC packet length");
/* skip entire packet */
break;
}	}
	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;

#if 0	#if 0
/* check integrity of packet */	/* check integrity of packet */
crc = ntohs(rpc->call_crc);	crc = ntohs(rpc->call_crc);
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;
off += len;	}
/* try next packet remaining into buffer */
continue;
}
#endif	#endif

noreply = RPC_CHK_NOREPLY(rpc);	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_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 */	rpc->call_argc ^= rpc->call_argc;
if (!noreply)	rpc->call_rep.ret = RPC_ERROR(-1);
schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET],	rpc->call_rep.eno = RPC_ERROR(errno);
TASK_ARG(task), TASK_FD(task), rpc, len);	} else {
	/* execute RPC call */
	schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), (int) noreply, rpc, len);
	}

off += len;	/* send RPC reply */
} while (rlen > 0);	if (!noreply)
	schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET],
	TASK_ARG(task), TASK_FD(task), rpc, len);

/* 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;	return NULL;
}	}

Line 364 acceptClients(sched_task_t *task)	Line 410 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 = 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);
Line 379 acceptClients(sched_task_t *task)	Line 435 acceptClients(sched_task_t *task)
} 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:
Line 393 txUDPPacket(sched_task_t *task)	Line 455 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, estlen, wlen = sizeof(struct tagRPCCall);
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
	struct pollfd pfd;

schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);	schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA, TASK_ARG(task), NULL);
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) {
Line 413 txUDPPacket(sched_task_t *task)	Line 473 txUDPPacket(sched_task_t *task)
rpc->call_rep.ret = RPC_ERROR(-1);	rpc->call_rep.ret = RPC_ERROR(-1);
rpc->call_rep.eno = RPC_ERROR(rpc_Errno);	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
} else {	} else {
	/* calc estimated length */
	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 = htons(array_Size(RPC_RETVARS(c)));	rpc->call_argc = htons(array_Size(RPC_RETVARS(c)));
/* Go Encapsulate variables */	/* Go Encapsulate variables */
ret = ait_vars2buffer(buf + wlen, sizeof buf - wlen, RPC_RETVARS(c));	ret = ait_vars2buffer(buf + wlen, AIT_LEN(&c->cli_buf) - wlen,
	RPC_RETVARS(c));
/* Free return values */	/* Free return values */
ait_freeVars(&c->cli_vars);	ait_freeVars(&c->cli_vars);
if (ret == -1) {	if (ret == -1) {
Line 428 txUDPPacket(sched_task_t *task)	Line 496 txUDPPacket(sched_task_t *task)
}	}
}	}

rpc->call_len = htons(wlen);	rpc->call_len = htonl(wlen);

/* 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));

/* send reply */	/* send reply */
ret = sendto(TASK_FD(task), buf, wlen, MSG_NOSIGNAL,	pfd.fd = TASK_FD(task);
&c->cli_sa.sa, c->cli_sa.sa.sa_len);	pfd.events = POLLOUT;
if (ret == -1 \|\| ret != wlen) {	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 = sendto(TASK_FD(task), buf, MIN(wlen, s->srv_netbuf), 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);
	return NULL;
	}
}	}

return NULL;	return NULL;
Line 451 rxUDPPacket(sched_task_t *task)	Line 535 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, rlen, noreply, estlen;
u_short crc, off = 0;	u_short crc;
u_char buf[USHRT_MAX + 1];	u_char *buf, b[sizeof(struct tagRPCCall)];
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) b;
sockaddr_t sa;	sockaddr_t sa;
socklen_t salen;	socklen_t salen;
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
	struct pollfd pfd;

/* receive connect packet */	/* receive connect packet */
salen = sa.ss.ss_len = sizeof(sockaddr_t);	salen = sa.ss.ss_len = sizeof(sockaddr_t);
rlen = recvfrom(TASK_FD(task), buf, sizeof buf, 0, &sa.sa, &salen);	rlen = recvfrom(TASK_FD(task), b, sizeof b, MSG_PEEK, &sa.sa, &salen);
if (rlen < 1)	if (rlen < sizeof(struct tagRPCCall)) {
	rpc_SetErr(ERPCMISMATCH, "Short RPC packet");
goto end;	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 {
	estlen = ntohl(rpc->call_len);
	if (estlen > AIT_LEN(&c->cli_buf))
	AIT_RE_BUF(&c->cli_buf, estlen);
	rpc = (struct tagRPCCall*) AIT_GET_BUF(&c->cli_buf);
	buf = AIT_GET_BUF(&c->cli_buf);
	len = estlen;

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 {	/* get next part of packet */
/* check RPC packet */	memset(buf, 0, len);
if (rlen < sizeof(struct tagRPCCall)) {	pfd.fd = TASK_FD(task);
rpc_SetErr(ERPCMISMATCH, "Short RPC packet");	pfd.events = POLLIN \| POLLPRI;
break;	for (; len > 0; len -= rlen, buf += rlen) {
} else	if ((rlen = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 \|\|
rpc = (struct tagRPCCall*) (AIT_GET_BUF(&c->cli_buf) + off);	pfd.revents & (POLLERR \| POLLHUP \| POLLNVAL)) {
	if (rlen)
len = ntohs(rpc->call_len);	LOGERR;
rlen -= len;	else
	rpc_SetErr(ETIMEDOUT, "Timeout reached! Client not respond");
/* check RPC packet lengths */	schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
if (rlen < 0 \|\| len < sizeof(struct tagRPCCall)) {	c, 0, NULL, 0);
rpc_SetErr(ERPCMISMATCH, "Broken RPC packet length");	return NULL;
/* skip entire packet */
break;
}	}
	salen = sa.ss.ss_len = sizeof(sockaddr_t);
/* check integrity of packet */	rlen = recvfrom(TASK_FD(task), buf, len, 0, &sa.sa, &salen);
crc = ntohs(rpc->call_crc);	if (rlen == -1) {
rpc->call_crc ^= rpc->call_crc;	/* close connection */
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {	schedEvent(TASK_ROOT(task), cbProto[srv->srv_proto][CB_CLOSECLIENT],
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");	c, 0, NULL, 0);
	return NULL;
off += len;
/* try next packet remaining into buffer */
continue;
}	}
	if (e_addrcmp(&c->cli_sa, &sa, 42))
	rlen ^= rlen; /* skip if arrive from different address */
	}
	len = estlen;

noreply = RPC_CHK_NOREPLY(rpc);	/* 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");
	return NULL;
	}

/* check RPC packet session info */	noreply = RPC_CHK_NOREPLY(rpc);
if (rpc_chkPktSession(&rpc->call_session, &srv->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, c, off, NULL, 0);
}

/* send RPC reply */	/* check RPC packet session info */
if (!noreply)	if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],	rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
c, TASK_FD(task), rpc, len);

off += len;	rpc->call_argc ^= rpc->call_argc;
} while (rlen > 0);	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 */
	if (!noreply)
	schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
	c, TASK_FD(task), rpc, len);
end:	end:
schedReadSelf(task);	schedReadSelf(task);
return NULL;	return NULL;
Line 669 end:	Line 769 end:
static void *	static void *
flushBLOB(sched_task_t *task)	flushBLOB(sched_task_t *task)
{	{
rpc_srv_t *srv = TASK_ARG(task);	uintptr_t sigArg = atomic_load_acq_ptr(&_glSigArg);
	rpc_srv_t srv = sigArg ? (void) sigArg : TASK_ARG(task);
rpc_blob_t b, tmp;	rpc_blob_t b, tmp;

TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {	TAILQ_FOREACH_SAFE(b, &srv->srv_blob.blobs, blob_node, tmp) {
Line 679 flushBLOB(sched_task_t *task)	Line 780 flushBLOB(sched_task_t *task)
e_free(b);	e_free(b);
}	}

schedSignalSelf(task);	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);
	}

return NULL;	return NULL;
}	}

Line 690 acceptBLOBClients(sched_task_t *task)	Line 801 acceptBLOBClients(sched_task_t *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 = sizeof(sockaddr_t);
	int sock;
#ifdef TCP_NOPUSH	#ifdef TCP_NOPUSH
int n = 1;	int n = 1;
#endif	#endif
Line 697 acceptBLOBClients(sched_task_t *task)	Line 809 acceptBLOBClients(sched_task_t *task)
/* check free slots for connect */	/* check free slots for connect */
for (i = 0; i < array_Size(srv->srv_blob.clients) &&	for (i = 0; i < array_Size(srv->srv_blob.clients) &&
(c = 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 = e_malloc(sizeof(rpc_cli_t));	c = e_malloc(sizeof(rpc_cli_t));
if (!c) {	if (!c) {
LOGERR;	LOGERR;
Line 885 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 1004 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
return -1;	return -1;
}	}

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

	atomic_store_rel_ptr(&_glSigArg, (uintptr_t) srv);

	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 1163 rpc_srv_execCall(rpc_cli_t * __restrict cli, struct ta	Line 1294 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;

	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;
	}
	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_loopServer2() - Execute Main layer2 server loop and wait for clients requests
	*
	* @srv = RPC Server instance
	* return: -1 error or 0 ok, infinite loop ...
	*/
	int
	rpc_srv_loopServer2(rpc_srv_t * __restrict srv)
	{
	rpc_cli_t *c;
	register int i;
	rpc_func_t *f;
	struct timespec ts = { RPC_SCHED_POLLING, 0 };

	if (!srv) {
	rpc_SetErr(EINVAL, "Invalid parameter can`t start RPC server");
	return -1;
	}

	if (!schedRead(srv->srv_root, cbProto[srv->srv_proto][CB_ACCEPTCLIENT], srv,
	srv->srv_server.cli_sock, NULL, 0)) {
	rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
	return -1;
	}

	schedPolling(srv->srv_root, &ts, NULL);
	/* main rpc loop */
	schedRun(srv->srv_root, &srv->srv_kill);

	/* close all clients connections & server socket */
	for (i = 0; i < array_Size(srv->srv_clients); i++) {
	c = array(srv->srv_clients, i, rpc_cli_t*);
	if (c) {
	schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
	ait_freeVars(&RPC_RETVARS(c));
	AIT_FREE_VAL(&c->cli_buf);
	}
	array_Del(srv->srv_clients, i, 42);
	}
	array_Destroy(&srv->srv_clients);

	close(srv->srv_server.cli_sock);

	/* detach exported calls */
	RPC_FUNCS_LOCK(&srv->srv_funcs);
	while ((f = SLIST_FIRST(&srv->srv_funcs))) {
	SLIST_REMOVE_HEAD(&srv->srv_funcs, func_next);

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

	return 0;
}	}

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

Removed from v.1.17.4.1
changed lines
	Added in v.1.23.6.2