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.16.2.2 and 1.17.4.7

version 1.16.2.2, 2013/07/15 14:22:00	version 1.17.4.7, 2013/08/21 13:02:33
Line 67 static sched_task_func_t cbProto[SOCK_RAW + 1][4] = {	Line 67 static sched_task_func_t cbProto[SOCK_RAW + 1][4] = {
};	};


	static inline u_char *
	getBuffer(rpc_cli_t * __restrict c)
	{
	u_char *b = NULL;

	assert(c);

	if (RPC_ISNEXTBUF(c))
	b = AIT_GET_BUF(array(c->cli_buf, 1, ait_val_t*));
	else
	b = AIT_GET_BUF(array(c->cli_buf, 0, ait_val_t*)) +
	sizeof(struct tagRPCCall);

	return b;
	}

	static inline struct tagRPCCall *
	getHeader(rpc_cli_t * __restrict c)
	{
	assert(c);

	return (struct tagRPCCall) AIT_GET_BUF(array(c->cli_buf, 0, ait_val_t));
	}

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 98 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 */	/* free buffer(s) */
AIT_FREE_VAL(&c->cli_buf);	ait_freeVars(&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 126 _allocClient(rpc_srv_t * __restrict srv, sockaddr_t *	Line 150 _allocClient(rpc_srv_t * __restrict srv, sockaddr_t *
c->cli_parent = srv;	c->cli_parent = srv;
}	}

/* alloc empty buffer */	/* init buffer(s) */
AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);	c->cli_buf = ait_allocVars(2);
	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 161 txPacket(sched_task_t *task)	Line 191 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;
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) TASK_DATA(task);
int ret, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
	int len = sizeof(struct tagRPCCall) + ntohl(rpc->call_len);

/* copy RPC header */	buf = e_malloc(len);
memcpy(buf, TASK_DATA(task), wlen);	if (!buf) {
	rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
	/* close connection */
	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), 0, NULL, 0);
	return NULL;
	} else {
	/* copy RPC header */
	memcpy(buf, rpc, wlen);
	rpc = (struct tagRPCCall*) buf;
	}

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 {
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, len - 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 191 txPacket(sched_task_t *task)	Line 234 txPacket(sched_task_t *task)
}	}
}	}

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

#if 0
/* calculate CRC */
rpc->call_crc ^= rpc->call_crc;
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 \|\| ret != wlen) {
Line 207 txPacket(sched_task_t *task)	Line 244 txPacket(sched_task_t *task)
TASK_ARG(task), 0, NULL, 0);	TASK_ARG(task), 0, NULL, 0);
}	}

	e_free(buf);
return NULL;	return NULL;
}	}

Line 217 execCall(sched_task_t *task)	Line 255 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 = getBuffer(c);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall *rpc = getHeader(c);
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, ntohl(rpc->call_len), argc, 42);
AIT_LEN(&c->cli_buf) - TASK_VAL(task) - sizeof(struct tagRPCCall),
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 238 execCall(sched_task_t *task)	Line 275 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 271 rxPacket(sched_task_t *task)	Line 309 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;
u_short off = TASK_DATLEN(task);	ait_val_t bufz = array(c->cli_buf, 0, ait_val_t);
#if 0	u_char buf = (u_char) AIT_GET_BUF(bufz);
u_short crc;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
#endif
u_char *buf = AIT_GET_BUF(&c->cli_buf);
struct tagRPCCall *rpc;

if (!off)	memset(buf, 0, AIT_LEN(bufz));
memset(buf, 0, AIT_LEN(&c->cli_buf));	/* 1st buffer is last */
rlen = recv(TASK_FD(task), buf + off, AIT_LEN(&c->cli_buf) - off, 0);	RPC_CLR_NEXTBUF(c);
if (rlen < 1) {
	/* read rpc header */
	rlen = recv(TASK_FD(task), rpc, MIN(sizeof(struct tagRPCCall), AIT_LEN(bufz)), 0);
	if (rlen < sizeof(struct tagRPCCall) \|\|
	ntohl(rpc->call_len) < 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! */	buf += sizeof(struct tagRPCCall);
off = 0; /* process buffer from start offset == 0 */	len = ntohl(rpc->call_len);
}	}

do {	if (len > (AIT_LEN(bufz) - sizeof(struct tagRPCCall))) {
/* check RPC packet */	/* add extra buffer */
if (rlen < sizeof(struct tagRPCCall)) {	if (!(bufz = ait_getVars(&c->cli_buf, 1))) {
rpc_SetErr(ERPCMISMATCH, "Short RPC packet");	/* close connection */
	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
/* reminder received previous bytes ;) */	TASK_ARG(task), 0, NULL, 0);
schedRead(TASK_ROOT(task), TASK_FUNC(task), TASK_ARG(task),
TASK_FD(task), TASK_DATA(task), rlen);
return NULL;	return NULL;
} else	} else {
rpc = (struct tagRPCCall*) (buf + off);	AIT_FREE_VAL(bufz);
	AIT_SET_BUFSIZ(bufz, 0, len);
len = ntohs(rpc->call_len);	buf = AIT_GET_BUF(bufz);
rlen -= len;

/* check RPC packet lengths */
if (rlen < 0 \|\| len < sizeof(struct tagRPCCall)) {
rpc_SetErr(ERPCMISMATCH, "Broken RPC packet length");
/* skip entire packet */
break;
}	}
	/* buffer isnt last */
	RPC_SET_NEXTBUF(c);
	}

#if 0	/* read payload */
/* check integrity of packet */	rlen = recv(TASK_FD(task), buf, len, 0);
crc = ntohs(rpc->call_crc);	if (rlen < len) {
rpc->call_crc ^= rpc->call_crc;	/* close connection */
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");	TASK_ARG(task), 0, NULL, 0);
	return NULL;
	}

off += len;	noreply = RPC_CHK_NOREPLY(rpc);
/* try next packet remaining into buffer */
continue;
}
#endif

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

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

/* send RPC reply */	/* send RPC reply */
if (!noreply)	if (!noreply)
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);

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

Line 367 acceptClients(sched_task_t *task)	Line 394 acceptClients(sched_task_t *task)
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_freeVars(&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
Line 387 txUDPPacket(sched_task_t *task)	Line 414 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;
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) TASK_DATA(task);
int ret, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
	int len = sizeof(struct tagRPCCall) + ntohl(rpc->call_len);
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);
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 */	buf = e_malloc(len);
memcpy(buf, TASK_DATA(task), wlen);	if (!buf) {
	rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
	/* close connection */
	schedEvent(TASK_ROOT(task), cbProto[s->srv_proto][CB_CLOSECLIENT],
	TASK_ARG(task), 0, NULL, 0);
	return NULL;
	} else {
	/* copy RPC header */
	memcpy(buf, rpc, wlen);
	rpc = (struct tagRPCCall*) buf;
	}

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 {
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, len - 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 422 txUDPPacket(sched_task_t *task)	Line 462 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;
Line 437 txUDPPacket(sched_task_t *task)	Line 477 txUDPPacket(sched_task_t *task)
TASK_ARG(task), 0, NULL, 0);	TASK_ARG(task), 0, NULL, 0);
}	}

	e_free(buf);
return NULL;	return NULL;
}	}

Line 446 rxUDPPacket(sched_task_t *task)	Line 487 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;
u_short crc, off = 0;	ait_val_t *bufz;
u_char buf[USHRT_MAX + 1];	u_char *buf = NULL;
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpcbuf, *rpc;
sockaddr_t sa;	sockaddr_t sa[2];
socklen_t salen;	socklen_t salen;
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

	memset(&rpcbuf, 0, sizeof rpcbuf);

/* receive connect packet */	/* receive connect packet */
salen = sa.ss.ss_len = sizeof(sockaddr_t);	salen = sa[0].ss.ss_len = sizeof(sockaddr_t);
rlen = recvfrom(TASK_FD(task), buf, sizeof buf, 0, &sa.sa, &salen);	rlen = recvfrom(TASK_FD(task), &rpcbuf, sizeof rpcbuf, 0, &sa[0].sa, &salen);
if (rlen < 1)	if (rlen < sizeof(struct tagRPCCall) \|\| ntohl(rpcbuf.call_len) < sizeof(struct tagRPCCall))
goto end;	goto end;
	else
	len = ntohl(rpcbuf.call_len);

c = _allocClient(srv, &sa);	buf = e_malloc(len);
	if (!buf)
	goto end;
	else
	memset(buf, 0, len);

	/* read payload */
	salen = sa[1].ss.ss_len = sizeof(sockaddr_t);
	rlen = recvfrom(TASK_FD(task), buf, len, 0, &sa[1].sa, &salen);
	if (rlen < len \|\| memcmp(&sa[0], &sa[1], sizeof sa[0]))
	goto end;

	c = _allocClient(srv, sa);
if (!c)	if (!c)
goto end;	goto end;
else {	else {
	/* add extra buffer */
	if (!(bufz = ait_getVars(&c->cli_buf, 1)))
	goto end;
	else {
	AIT_FREE_VAL(bufz);
	AIT_SET_BUFSIZ(bufz, 0, len);
	/* buffer isnt last */
	RPC_SET_NEXTBUF(c);
	}

	rpc = getHeader(c);
	memcpy(rpc, &rpcbuf, sizeof(struct tagRPCCall));
	memcpy(getBuffer(c), buf, len);

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 {	/* check integrity of packet */
/* check RPC packet */	if (ntohs(rpc->call_crc) != crcFletcher16((u_short*) buf, len / 2)) {
if (rlen < sizeof(struct tagRPCCall)) {	rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");
rpc_SetErr(ERPCMISMATCH, "Short RPC packet");	goto end;
break;	}
} else
rpc = (struct tagRPCCall*) (AIT_GET_BUF(&c->cli_buf) + off);

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

/* check RPC packet lengths */	/* check RPC packet session info */
if (rlen < 0 \|\| len < sizeof(struct tagRPCCall)) {	if (rpc_chkPktSession(&rpc->call_session, &srv->srv_session)) {
rpc_SetErr(ERPCMISMATCH, "Broken RPC packet length");	rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
/* skip entire packet */
break;
}

/* check integrity of packet */	rpc->call_argc ^= rpc->call_argc;
crc = ntohs(rpc->call_crc);	rpc->call_rep.ret = RPC_ERROR(-1);
rpc->call_crc ^= rpc->call_crc;	rpc->call_rep.eno = RPC_ERROR(errno);
if (crc != crcFletcher16((u_short*) rpc, len / 2)) {	} else {
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");	/* execute RPC call */
	schedEvent(TASK_ROOT(task), execCall, c, noreply, rpc, len);
	}

off += len;	/* send RPC reply */
/* try next packet remaining into buffer */	if (!noreply)
continue;	schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
}	c, TASK_FD(task), rpc, len);

noreply = RPC_CHK_NOREPLY(rpc);

/* check RPC packet session info */
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 */
if (!noreply)
schedWrite(TASK_ROOT(task), cbProto[srv->srv_proto][CB_TXPACKET],
c, TASK_FD(task), rpc, len);

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

end:	end:
	if (buf)
	e_free(buf);
schedReadSelf(task);	schedReadSelf(task);
return NULL;	return NULL;
}	}
Line 536 rpc_freeBLOBCli(rpc_cli_t * __restrict c)	Line 582 rpc_freeBLOBCli(rpc_cli_t * __restrict c)

schedCancelby(s->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);	schedCancelby(s->srv_blob.root, taskMAX, CRITERIA_ARG, c, NULL);

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

array_Del(s->srv_blob.clients, c->cli_id, 0);	array_Del(s->srv_blob.clients, c->cli_id, 0);
if (c)	if (c)
Line 559 closeBLOBClient(sched_task_t *task)	Line 605 closeBLOBClient(sched_task_t *task)
}	}

static void *	static void *
toutBLOB(sched_task_t *task)
{
rpc_cli_t *c = TASK_ARG(task);

rpc_srv_unregisterBLOB((rpc_srv_t*) c->cli_parent, (uint32_t) TASK_DATA(task));

return NULL;
}

static void *
txBLOB(sched_task_t *task)	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(array(c->cli_buf, 0, ait_val_t));
int wlen = sizeof(struct tagBLOBHdr);	int wlen = sizeof(struct tagBLOBHdr);

/* send reply */	/* send reply */
Line 593 rxBLOB(sched_task_t *task)	Line 629 rxBLOB(sched_task_t *task)
rpc_blob_t *b;	rpc_blob_t *b;
struct tagBLOBHdr blob;	struct tagBLOBHdr blob;
int rlen;	int rlen;
struct timespec ts = { RPC_BLOB_TIMEOUT, 0 };

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);
Line 639 rxBLOB(sched_task_t *task)	Line 674 rxBLOB(sched_task_t *task)
}	}
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);

/* set live BLOB timeout */
ts.tv_sec = blob.hdr_ret ? blob.hdr_ret : RPC_BLOB_TIMEOUT;

/* receive BLOB from client */	/* receive BLOB from client */
blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b));	blob.hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
rpc_srv_blobUnmap(b);	rpc_srv_blobUnmap(b);

schedTimer(TASK_ROOT(task), toutBLOB, TASK_ARG(task), ts,
(void*) (intptr_t) ntohl(blob.hdr_var), ntohl(blob.hdr_len));
} else {	} else {
blob.hdr_cmd = error;	blob.hdr_cmd = error;
blob.hdr_ret = RPC_ERROR(-1);	blob.hdr_ret = RPC_ERROR(-1);
Line 662 rxBLOB(sched_task_t *task)	Line 692 rxBLOB(sched_task_t *task)
blob.hdr_cmd = error;	blob.hdr_cmd = error;
blob.hdr_ret = RPC_ERROR(-1);	blob.hdr_ret = RPC_ERROR(-1);
}	}

schedCancelby(TASK_ROOT(task), taskTIMER, CRITERIA_DATA,
(void*) (intptr_t) ntohl(blob.hdr_var), NULL);
break;	break;
default:	default:
rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);	rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob.hdr_cmd);
Line 673 rxBLOB(sched_task_t *task)	Line 700 rxBLOB(sched_task_t *task)
}	}

end:	end:
memcpy(AIT_ADDR(&c->cli_buf), &blob, sizeof blob);	memcpy(AIT_ADDR(array(c->cli_buf, 0, ait_val_t*)), &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;	return NULL;
}	}

static void *	static void *
	flushBLOB(sched_task_t *task)
	{
	rpc_srv_t *srv = 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);
	}

	schedSignalSelf(task);
	return 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);
Line 707 acceptBLOBClients(sched_task_t *task)	Line 751 acceptBLOBClients(sched_task_t *task)
c->cli_parent = srv;	c->cli_parent = srv;
}	}

/* alloc empty buffer */	/* init buffer(s) */
AIT_SET_BUFSIZ(&c->cli_buf, 0, srv->srv_netbuf);	c->cli_buf = ait_allocVars(1);
	if (!c->cli_buf) {
	rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
	array_Del(srv->srv_blob.clients, i, 42);
	goto end;
	} else
	AIT_SET_BUFSIZ(array(c->cli_buf, 0, ait_val_t*), 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_freeVars(&c->cli_buf);
array_Del(srv->srv_blob.clients, i, 42);	array_Del(srv->srv_blob.clients, i, 42);
goto end;	goto end;
} else {	} else {
Line 853 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 903 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
return;	return;

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

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

/*	/*
Line 879 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 931 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
return -1;	return -1;
}	}

	schedSignal(srv->srv_blob.root, flushBLOB, srv, SIGFBLOB, NULL, 0);
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 889 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 942 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 < array_Size(srv->srv_blob.clients); i++) {	for (i = 0; i < array_Size(srv->srv_blob.clients); i++) {
c = array(srv->srv_blob.clients, i, rpc_cli_t*);	c = array(srv->srv_blob.clients, i, rpc_cli_t*);
Line 897 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 958 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)
close(c->cli_sock);	close(c->cli_sock);

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_freeVars(&c->cli_buf);
}	}
array_Del(srv->srv_blob.clients, i, 42);	array_Del(srv->srv_blob.clients, i, 42);
}	}
Line 905 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)	Line 966 rpc_srv_loopBLOBServer(rpc_srv_t * __restrict srv)

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);
e_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 1056 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 1108 rpc_srv_endServer(rpc_srv_t ** __restrict 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);

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

	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 1111 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 1164 rpc_srv_loopServer(rpc_srv_t * __restrict srv)

schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);	schedCancelby(srv->srv_root, taskMAX, CRITERIA_ARG, c, NULL);
ait_freeVars(&RPC_RETVARS(c));	ait_freeVars(&RPC_RETVARS(c));
AIT_FREE_VAL(&c->cli_buf);	ait_freeVars(&c->cli_buf);
}	}
array_Del(srv->srv_clients, i, 42);	array_Del(srv->srv_clients, i, 42);
}	}
Line 1130 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 1183 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
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;
}	}

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

Removed from v.1.16.2.2
changed lines
	Added in v.1.17.4.7