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.7 and 1.9

version 1.7, 2012/03/15 01:55:33	version 1.9, 2012/05/14 08:39:06
Line 60 txPacket(sched_task_t *task)	Line 60 txPacket(sched_task_t *task)
int ret, wlen = sizeof(struct tagRPCCall);	int ret, wlen = sizeof(struct tagRPCCall);
array_t *arr = NULL;	array_t *arr = NULL;

ioTRACE(RPC_TRACE_LEVEL);

if (rpc->call_argc) {	if (rpc->call_argc) {
f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash));	f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash));
if (!f) {	if (!f) {
Line 83 txPacket(sched_task_t *task)	Line 81 txPacket(sched_task_t *task)
}	}
}	}

	rpc->call_len = htons(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, io_align(wlen, 1) / 2));	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));

/* send reply */	/* send reply */
ret = send(TASK_FD(task), buf, wlen, 0);	ret = send(TASK_FD(task), buf, wlen, 0);
Line 94 txPacket(sched_task_t *task)	Line 94 txPacket(sched_task_t *task)
else if (ret != wlen)	else if (ret != wlen)
rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "	rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
"really sended %d bytes", wlen, ret);	"really sended %d bytes", wlen, ret);
else
ioDEBUG(RPC_DEBUG_LEVEL, "Sended %d bytes", ret);

return NULL;	return NULL;
}	}
Line 107 execCall(sched_task_t *task)	Line 105 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 = TASK_DATA(task);	u_char *buf = TASK_DATA(task) + TASK_VAL(task);
struct tagRPCCall rpc = (struct tagRPCCall) buf;	struct tagRPCCall rpc = (struct tagRPCCall) buf;
int argc = ntohs(rpc->call_argc);	int argc = ntohs(rpc->call_argc);

ioTRACE(RPC_TRACE_LEVEL);

/* Go decapsulate variables ... */	/* Go decapsulate variables ... */
if (!(rpc->call_req.flags & RPC_NOREPLY) && argc) {	if (argc) {
arr = io_buffer2vars(buf + sizeof(struct tagRPCCall),	arr = io_buffer2vars(buf + sizeof(struct tagRPCCall),
TASK_DATLEN(task) - sizeof(struct tagRPCCall), argc, 1);	TASK_DATLEN(task) - TASK_VAL(task) - sizeof(struct tagRPCCall),
	argc, 1);
if (!arr) {	if (!arr) {
rpc_SetErr(ERPCMISMATCH, "#%d - %s", io_GetErrno(), io_GetError());	rpc_SetErr(ERPCMISMATCH, "#%d - %s", io_GetErrno(), io_GetError());
rpc->call_argc ^= rpc->call_argc;	rpc->call_argc ^= rpc->call_argc;
Line 132 execCall(sched_task_t *task)	Line 129 execCall(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 {
ioDEBUG(RPC_DEBUG_LEVEL, "RPC function %s from module %s",	/* if client doesn't want reply */
AIT_GET_STR(&f->func_name), AIT_GET_LIKE(&f->func_file, char*));	argc = rpc->call_req.flags & RPC_NOREPLY;

rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));	rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));
if (rpc->call_rep.ret == htonl(-1)) {	if (rpc->call_rep.ret == htonl(-1)) {
rpc->call_rep.eno = RPC_ERROR(errno);	rpc->call_rep.eno = RPC_ERROR(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;
rpc->call_argc = htons(rpc_srv_getVars(f, NULL));	if (argc) {
	/* without reply */
	io_clrVars(f->func_vars);
	rpc->call_argc ^= rpc->call_argc;
	} else
	rpc->call_argc = htons(rpc_srv_getVars(f, NULL));
}	}
}	}

Line 156 rxPacket(sched_task_t *task)	Line 157 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;
u_char *buf = TASK_DATA(task);	u_char *buf = TASK_DATA(task);
int rlen;	int rlen, noreply;
u_short crc;	u_short crc, off = 0;
struct tagRPCCall *rpc;	struct tagRPCCall *rpc;
struct timespec ts;	struct timespec ts;

ioTRACE(RPC_TRACE_LEVEL);

memset(buf, 0, TASK_DATLEN(task));	memset(buf, 0, TASK_DATLEN(task));
rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);	rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
if (rlen == -1) {	if (rlen == -1) {
LOGERR;	LOGERR;
s->srv_kill = s->srv_blob.state = kill;	c->cli_kill = kill;
return NULL;	return NULL;
} else if (!rlen) { /* receive EOF */	} else if (!rlen) { /* receive EOF */
s->srv_kill = s->srv_blob.state = kill;	c->cli_kill = kill;
return NULL;	return NULL;
} else	}
ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen);

if (rlen < sizeof(struct tagRPCCall)) {	do {
rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");	if (rlen < sizeof(struct tagRPCCall)) {
	rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");

schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),	schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),
TASK_DATA(task), TASK_DATLEN(task));	TASK_DATA(task), TASK_DATLEN(task));
return NULL;	return NULL;
} else	} else
rpc = (struct tagRPCCall*) buf;	rpc = (struct tagRPCCall*) (buf + off);

/* check integrity of packet */	rlen -= ntohs(rpc->call_len);
crc = ntohs(rpc->call_crc);
rpc->call_crc ^= rpc->call_crc;
if (crc != crcFletcher16((u_short*) buf, io_align(rlen, 1) / 2)) {
rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");

schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),	/* check integrity of packet */
TASK_DATA(task), TASK_DATLEN(task));	crc = ntohs(rpc->call_crc);
return NULL;	rpc->call_crc ^= rpc->call_crc;
}	if (crc != crcFletcher16((u_short*) (buf + off), ntohs(rpc->call_len) / 2)) {
	rpc_SetErr(ERPCMISMATCH, "Bad CRC RPC packet");

/* check RPC packet session info */	off += ntohs(rpc->call_len);
if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {	if (rlen < 1)
rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");	break;
rpc->call_argc ^= rpc->call_argc;	else
rpc->call_rep.ret = RPC_ERROR(-1);	continue;
rpc->call_rep.eno = RPC_ERROR(errno);	}
goto end;
} else {
/* change socket timeout from last packet */
ts.tv_sec = rpc->call_session.sess_timeout;
ts.tv_nsec = 0;
schedPolling(TASK_ROOT(task), &ts, NULL);
}

/* execute RPC call */	noreply = rpc->call_req.flags & RPC_NOREPLY;
schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), 0,
TASK_DATA(task), TASK_DATLEN(task));

end:	/* check RPC packet session info */
/* send RPC reply */	if (rpc_chkPktSession(&rpc->call_session, &s->srv_session)) {
if (!(rpc->call_req.flags & RPC_NOREPLY))	rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task),	rpc->call_argc ^= rpc->call_argc;
TASK_DATA(task), TASK_DATLEN(task));	rpc->call_rep.ret = RPC_ERROR(-1);
	rpc->call_rep.eno = RPC_ERROR(errno);
	} else {
	/* change socket timeout from last packet */
	ts.tv_sec = rpc->call_session.sess_timeout;
	ts.tv_nsec = 0;
	schedPolling(TASK_ROOT(task), &ts, NULL);

	/* execute RPC call */
	schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), off,
	TASK_DATA(task), TASK_DATLEN(task));
	}

	/* send RPC reply */
	if (!noreply)
	schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task),
	buf + off, TASK_DATLEN(task) - off);

	off += ntohs(rpc->call_len);
	} while (rlen > 0);

/* lets get next packet */	/* lets get next packet */
schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),	schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),
TASK_DATA(task), TASK_DATLEN(task));	TASK_DATA(task), TASK_DATLEN(task));
Line 233 rpc_srv_dispatchCall(void *arg)	Line 240 rpc_srv_dispatchCall(void *arg)
sched_root_task_t *root;	sched_root_task_t *root;
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

ioTRACE(RPC_TRACE_LEVEL);

if (!arg) {	if (!arg) {
rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client");	rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client");
return NULL;	return NULL;
Line 260 rpc_srv_dispatchCall(void *arg)	Line 265 rpc_srv_dispatchCall(void *arg)

schedRead(root, rxPacket, c, c->cli_sock, buf, s->srv_netbuf);	schedRead(root, rxPacket, c, c->cli_sock, buf, s->srv_netbuf);

schedRun(root, (void*) &s->srv_kill);	schedRun(root, (void*) &c->cli_kill);
schedEnd(&root);	schedEnd(&root);

shutdown(c->cli_sock, SHUT_RDWR);	shutdown(c->cli_sock, SHUT_RDWR);
Line 278 txBLOB(sched_task_t *task)	Line 283 txBLOB(sched_task_t *task)
struct tagBLOBHdr blob = (struct tagBLOBHdr ) buf;	struct tagBLOBHdr blob = (struct tagBLOBHdr ) buf;
int wlen = sizeof(struct tagBLOBHdr);	int wlen = sizeof(struct tagBLOBHdr);

ioTRACE(RPC_TRACE_LEVEL);

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

/* send reply */	/* send reply */
wlen = send(TASK_FD(task), buf, wlen, 0);	wlen = send(TASK_FD(task), buf, wlen, 0);
Line 291 txBLOB(sched_task_t *task)	Line 294 txBLOB(sched_task_t *task)
else if (wlen != sizeof(struct tagBLOBHdr))	else if (wlen != sizeof(struct tagBLOBHdr))
rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "	rpc_SetErr(EPROCUNAVAIL, "RPC reply, should be send %d bytes, "
"really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);	"really sended %d bytes", sizeof(struct tagBLOBHdr), wlen);
else
ioDEBUG(RPC_DEBUG_LEVEL, "Sended %d bytes", wlen);

return NULL;	return NULL;
}	}
Line 309 rxBLOB(sched_task_t *task)	Line 310 rxBLOB(sched_task_t *task)
u_short crc;	u_short crc;
struct timespec ts;	struct timespec ts;

ioTRACE(RPC_TRACE_LEVEL);

/* check for disable service at this moment? */	/* check for disable service at this moment? */
if (s->srv_blob.state == disable) {	if (!s \|\| s->srv_blob.state == disable) {
usleep(100000);	usleep(100000);
#ifdef HAVE_PTHREAD_YIELD	#ifdef HAVE_PTHREAD_YIELD
pthread_yield();	pthread_yield();
Line 326 rxBLOB(sched_task_t *task)	Line 325 rxBLOB(sched_task_t *task)
rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);	rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
if (rlen == -1) {	if (rlen == -1) {
LOGERR;	LOGERR;
s->srv_blob.state = kill;	c->cli_kill = kill;
return NULL;	return NULL;
} else if (!rlen \|\| s->srv_kill == kill) { /* receive EOF */	} else if (!rlen) { /* receive EOF */
s->srv_blob.state = kill;	c->cli_kill = kill;
return NULL;	return NULL;
} else	}
ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen);

if (rlen < sizeof(struct tagBLOBHdr)) {	if (rlen < sizeof(struct tagBLOBHdr)) {
rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet");	rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet");
Line 344 rxBLOB(sched_task_t *task)	Line 342 rxBLOB(sched_task_t *task)
/* check integrity of packet */	/* check integrity of packet */
crc = ntohs(blob->hdr_crc);	crc = ntohs(blob->hdr_crc);
blob->hdr_crc ^= blob->hdr_crc;	blob->hdr_crc ^= blob->hdr_crc;
if (crc != crcFletcher16((u_short*) buf, io_align(rlen, 1) / 2)) {	if (crc != crcFletcher16((u_short*) buf, rlen / 2)) {
rpc_SetErr(ERPCMISMATCH, "Bad CRC BLOB packet");	rpc_SetErr(ERPCMISMATCH, "Bad CRC BLOB packet");
schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task),	schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task),
TASK_DATA(task), TASK_DATLEN(task));	TASK_DATA(task), TASK_DATLEN(task));
Line 425 rpc_srv_dispatchVars(void *arg)	Line 423 rpc_srv_dispatchVars(void *arg)
u_char *buf;	u_char *buf;
struct timespec ts = { DEF_RPC_TIMEOUT, 0 };	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

ioTRACE(RPC_TRACE_LEVEL);

if (!arg) {	if (!arg) {
rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");	rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");
return NULL;	return NULL;
Line 452 rpc_srv_dispatchVars(void *arg)	Line 448 rpc_srv_dispatchVars(void *arg)

schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));	schedRead(root, rxBLOB, c, c->cli_sock, buf, sizeof(struct tagBLOBHdr));

schedRun(root, (void*) &s->srv_blob.state);	schedRun(root, (void*) &c->cli_kill);
schedEnd(&root);	schedEnd(&root);

shutdown(c->cli_sock, SHUT_RDWR);	shutdown(c->cli_sock, SHUT_RDWR);
Line 478 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 474 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
int n = 1;	int n = 1;
io_sockaddr_t sa;	io_sockaddr_t sa;

ioTRACE(RPC_TRACE_LEVEL);

if (!srv) {	if (!srv) {
rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");	rpc_SetErr(EINVAL, "Invalid parameters can`t init BLOB server");
return -1;	return -1;
Line 560 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 554 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
} else	} else
memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));	memset(srv->srv_blob.clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));

pthread_mutex_init(&srv->srv_blob.mtx, NULL);

rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);	rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 1);	rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 1);
rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 1);	rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 1);
Line 584 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 576 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
register int i;	register int i;
rpc_blob_t *f;	rpc_blob_t *f;

ioTRACE(RPC_TRACE_LEVEL);	if (!srv \|\| srv->srv_blob.state == disable)

if (!srv) {
rpc_SetErr(EINVAL, "Can`t destroy server because parameter is null!");
return;	return;
} else	else
srv->srv_blob.state = kill;	srv->srv_blob.state = kill;

rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);
Line 603 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 592 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
shutdown(c->cli_sock, SHUT_RDWR);	shutdown(c->cli_sock, SHUT_RDWR);
close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);

pthread_mutex_lock(&srv->srv_blob.mtx);
if (srv->srv_blob.clients) {	if (srv->srv_blob.clients) {
free(srv->srv_blob.clients);	free(srv->srv_blob.clients);
srv->srv_blob.clients = NULL;	srv->srv_blob.clients = NULL;
Line 615 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 603 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
rpc_srv_blobFree(srv, f);	rpc_srv_blobFree(srv, f);
free(f);	free(f);
}	}
pthread_mutex_unlock(&srv->srv_blob.mtx);

AIT_FREE_VAL(&srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);

while (pthread_mutex_trylock(&srv->srv_blob.mtx) == EBUSY);	/* at final, save disable BLOB service state */
pthread_mutex_destroy(&srv->srv_blob.mtx);	srv->srv_blob.state = disable;
}	}

/*	/*
Line 635 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)	Line 622 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
socklen_t salen = sizeof(io_sockaddr_t);	socklen_t salen = sizeof(io_sockaddr_t);
register int i;	register int i;
rpc_cli_t *c;	rpc_cli_t *c;
fd_set fds;
int ret;	int ret;
struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
pthread_attr_t attr;	pthread_attr_t attr;
	struct pollfd pfd;

ioTRACE(RPC_TRACE_LEVEL);

if (!srv \|\| srv->srv_blob.state == kill) {	if (!srv \|\| srv->srv_blob.state == kill) {
rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");	rpc_SetErr(EINVAL, "Invalid parameter can`t start BLOB server");
return -1;	return -1;
}	}

tv.tv_sec = srv->srv_session.sess_timeout;

if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {	if (listen(srv->srv_blob.server.cli_sock, SOMAXCONN) == -1) {
LOGERR;	LOGERR;
return -1;	return -1;
}	} else
	fcntl(srv->srv_blob.server.cli_sock, F_SETFL,
	fcntl(srv->srv_blob.server.cli_sock, F_GETFL) \| O_NONBLOCK);

pthread_attr_init(&attr);	pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

	pfd.fd = srv->srv_blob.server.cli_sock;
	pfd.events = POLLIN \| POLLPRI;
/* main BLOB loop */	/* main BLOB loop */
while (srv->srv_blob.state != kill && srv->srv_kill != kill) {	while (srv->srv_blob.state != kill && srv->srv_kill != kill) {
for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)	for (c = srv->srv_blob.clients, i = 0; i < srv->srv_numcli && c; i++, c++)
Line 671 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)	Line 657 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
continue;	continue;
}	}

FD_ZERO(&fds);	if ((ret = poll(&pfd, 1, srv->srv_session.sess_timeout * 1000)) == -1 \|\|
FD_SET(srv->srv_blob.server.cli_sock, &fds);	pfd.revents & (POLLERR \| POLLHUP \| POLLNVAL)) {
ret = select(srv->srv_blob.server.cli_sock + 1, &fds, NULL, NULL, &tv);
if (ret == -1) {
LOGERR;	LOGERR;
ret = 1;	ret = 1;
break;	break;
}	} else if (!ret)
if (!ret)
continue;	continue;

c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa.sa, &salen);	c->cli_sock = accept(srv->srv_blob.server.cli_sock, &c->cli_sa.sa, &salen);
Line 690 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)	Line 673 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
c->cli_parent = srv;	c->cli_parent = srv;

/* spawn dispatch thread for BLOB client */	/* spawn dispatch thread for BLOB client */
	c->cli_kill = enable;
if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) {	if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchVars, c)) {
LOGERR;	LOGERR;
continue;	continue;
Line 724 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 708 rpc_srv_initServer(u_int regProgID, u_int regProcID, i
struct hostent *host = NULL;	struct hostent *host = NULL;
io_sockaddr_t sa;	io_sockaddr_t sa;

ioTRACE(RPC_TRACE_LEVEL);

if (!concurentClients \|\| !regProgID \|\|	if (!concurentClients \|\| !regProgID \|\|
(family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) {	(family != AF_INET && family != AF_INET6 && family != AF_LOCAL)) {
rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");	rpc_SetErr(EINVAL, "Error:: Invalid parameters can`t init RPC server ...\n");
Line 832 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 814 rpc_srv_initServer(u_int regProgID, u_int regProcID, i
} else	} else
memset(srv->srv_clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));	memset(srv->srv_clients, 0, srv->srv_numcli * sizeof(rpc_cli_t));

pthread_mutex_init(&srv->srv_mtx, NULL);

rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0);	rpc_srv_registerCall(srv, NULL, CALL_SRVSHUTDOWN, 0);
rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 1);	rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 1);
rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 4);	rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 4);
rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 1);	rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 1);

return srv;	return srv;
}	}

Line 854 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 835 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
register int i;	register int i;
rpc_func_t *f;	rpc_func_t *f;

ioTRACE(RPC_TRACE_LEVEL);

if (!psrv \|\| !*psrv) {	if (!psrv \|\| !*psrv) {
rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");	rpc_SetErr(EINVAL, "Error:: Can`t destroy server because parameter is null!\n");
return;	return;
Line 878 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 857 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
}	}

/* detach exported calls */	/* detach exported calls */
pthread_mutex_lock(&(*psrv)->srv_mtx);
while ((f = (*psrv)->srv_funcs)) {	while ((f = (*psrv)->srv_funcs)) {
(*psrv)->srv_funcs = f->func_next;	(*psrv)->srv_funcs = f->func_next;
io_freeVars(&f->func_vars);	io_freeVars(&f->func_vars);
Line 886 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 864 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
AIT_FREE_VAL(&f->func_file);	AIT_FREE_VAL(&f->func_file);
free(f);	free(f);
}	}
pthread_mutex_unlock(&(*psrv)->srv_mtx);

while (pthread_mutex_trylock(&(*psrv)->srv_mtx) == EBUSY);
pthread_mutex_destroy(&(*psrv)->srv_mtx);

free(*psrv);	free(*psrv);
*psrv = NULL;	*psrv = NULL;
}	}
Line 907 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 881 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
socklen_t salen = sizeof(io_sockaddr_t);	socklen_t salen = sizeof(io_sockaddr_t);
register int i;	register int i;
rpc_cli_t *c;	rpc_cli_t *c;
fd_set fds;
int ret;	int ret;
struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
pthread_attr_t attr;	pthread_attr_t attr;
	struct pollfd pfd;

ioTRACE(RPC_TRACE_LEVEL);

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

tv.tv_sec = srv->srv_session.sess_timeout;

/* activate BLOB server worker if srv->srv_blob.state == enable */	/* activate BLOB server worker if srv->srv_blob.state == enable */
rpc_srv_execBLOBServer(srv);	rpc_srv_execBLOBServer(srv);

if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) {	if (listen(srv->srv_server.cli_sock, SOMAXCONN) == -1) {
LOGERR;	LOGERR;
return -1;	return -1;
}	} else
	fcntl(srv->srv_server.cli_sock, F_SETFL, fcntl(srv->srv_server.cli_sock, F_GETFL) \| O_NONBLOCK);

pthread_attr_init(&attr);	pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

	pfd.fd = srv->srv_server.cli_sock;
	pfd.events = POLLIN \| POLLPRI;
/* main rpc loop */	/* main rpc loop */
while (srv->srv_kill != kill) {	while (srv->srv_kill != kill) {
for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)	for (c = srv->srv_clients, i = 0; i < srv->srv_numcli && c; i++, c++)
Line 946 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 918 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
continue;	continue;
}	}

FD_ZERO(&fds);	if ((ret = poll(&pfd, 1, srv->srv_session.sess_timeout * 1000)) == -1 \|\|
FD_SET(srv->srv_server.cli_sock, &fds);	pfd.revents & (POLLERR \| POLLHUP \| POLLNVAL)) {
ret = select(srv->srv_server.cli_sock + 1, &fds, NULL, NULL, &tv);
if (ret == -1) {
LOGERR;	LOGERR;
ret = 1;	ret = 1;
break;	break;
}	} else if (!ret)
if (!ret)
continue;	continue;

c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa.sa, &salen);	c->cli_sock = accept(srv->srv_server.cli_sock, &c->cli_sa.sa, &salen);
Line 965 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 934 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
c->cli_parent = srv;	c->cli_parent = srv;

/* spawn rpc client dispatcher */	/* spawn rpc client dispatcher */
	c->cli_kill = enable;
if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) {	if (pthread_create(&c->cli_tid, &attr, rpc_srv_dispatchCall, c)) {
LOGERR;	LOGERR;
continue;	continue;
Line 993 rpc_srv_execCall(rpc_func_t * __restrict call, struct	Line 963 rpc_srv_execCall(rpc_func_t * __restrict call, struct
rpc_callback_t func;	rpc_callback_t func;
int ret;	int ret;

ioTRACE(RPC_TRACE_LEVEL);

if (!call \|\| !rpc \|\| !call->func_parent) {	if (!call \|\| !rpc \|\| !call->func_parent) {
rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");	rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
return -1;	return -1;
}	}

dl = dlopen(AIT_VOID(&call->func_file), RTLD_NOW);	dl = dlopen(AIT_ADDR(&call->func_file), RTLD_NOW);
if (!dl) {	if (!dl) {
rpc_SetErr(ENOENT, "Can`t attach module %s!", dlerror());	rpc_SetErr(ENOENT, "Can`t attach module %s!", dlerror());
return -1;	return -1;

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

Removed from v.1.7
changed lines
	Added in v.1.9