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.1.1.1.2.11 and 1.1.1.1.2.22

version 1.1.1.1.2.11, 2010/06/24 15:01:19	version 1.1.1.1.2.22, 2010/07/12 08:46:41
Line 14 rpc_srv_dispatchCall(void *arg)	Line 14 rpc_srv_dispatchCall(void *arg)
{	{
rpc_cli_t *c = arg;	rpc_cli_t *c = arg;
rpc_srv_t *s;	rpc_srv_t *s;
rpc_val_t vals, v = NULL;	rpc_val_t vals = NULL, v = NULL;
rpc_func_t *f;	rpc_func_t *f;
struct tagRPCCall *rpc;	struct tagRPCCall *rpc;
struct tagRPCRet rrpc;	struct tagRPCRet rrpc;
fd_set fds;	fd_set fds;
u_char buf[BUFSIZ], *data;	u_char buf[BUFSIZ], *data;
int ret, argc, Limit = 0;	int ret, argc = 0, Limit = 0;
register int i;	register int i;

if (!arg) {	if (!arg) {
Line 57 rpc_srv_dispatchCall(void *arg)	Line 57 rpc_srv_dispatchCall(void *arg)
if (memcmp(&rpc->call_session, &s->srv_session, sizeof rpc->call_session)) {	if (memcmp(&rpc->call_session, &s->srv_session, sizeof rpc->call_session)) {
rpc_SetErr(EINVAL, "Error:: get invalid RPC session ...\n");	rpc_SetErr(EINVAL, "Error:: get invalid RPC session ...\n");
ret = -5;	ret = -5;
break;	goto makeReply;
}	} else
	Limit = sizeof(struct tagRPCCall);
// RPC is OK! Go decapsulate variables ...	// RPC is OK! Go decapsulate variables ...
if (rpc->call_argc) {	if (rpc->call_argc) {
v = (rpc_val_t*) (buf + sizeof(struct tagRPCCall));	v = (rpc_val_t*) (buf + Limit);
	// check RPC packet length
	if (rpc->call_argc * sizeof(rpc_val_t) > BUFSIZ - Limit) {
	rpc_SetErr(EMSGSIZE, "Error:: Too big RPC packet ...\n");
	ret = -5;
	goto makeReply;
	} else
	Limit += rpc->call_argc * sizeof(rpc_val_t);
// RPC received variables types OK!	// RPC received variables types OK!
data = (u_char) v + rpc->call_argc sizeof(rpc_val_t);	data = (u_char) v + rpc->call_argc sizeof(rpc_val_t);
for (i = 0; i < rpc->call_argc; i++) {	for (i = 0; i < rpc->call_argc; i++) {
switch (v[i].val_type) {	switch (v[i].val_type) {
case buffer:	case buffer:
	if (v[i].val_len > BUFSIZ - Limit) {
	rpc_SetErr(EMSGSIZE, "Error:: Too big RPC packet ...\n");
	ret = -5;
	goto makeReply;
	} else
	Limit += v[i].val_len;

v[i].val.buffer = data;	v[i].val.buffer = data;
data += v[i].val_len;	data += v[i].val_len;
break;	break;
case string:	case string:
	if (v[i].val_len + 1 > BUFSIZ - Limit) {
	rpc_SetErr(EMSGSIZE, "Error:: Too big RPC packet ...\n");
	ret = -5;
	goto makeReply;
	} else
	Limit += v[i].val_len;

v[i].val.string = (int8_t*) data;	v[i].val.string = (int8_t*) data;
data += v[i].val_len + 1;	data += v[i].val_len + 1;
break;	break;
	case blob:
	if (s->srv_blob.state == disable) {
	rpc_SetErr(ENOTSUP, "Error:: BLOB server is disabled\n");
	ret = -5;
	goto makeReply;
	}
default:	default:
break;	break;
}	}
Line 87 rpc_srv_dispatchCall(void *arg)	Line 115 rpc_srv_dispatchCall(void *arg)
ret = -6;	ret = -6;
} else	} else
if ((ret = rpc_srv_execCall(f, rpc, v)) == -1)	if ((ret = rpc_srv_execCall(f, rpc, v)) == -1)
ret = -6;	ret = -9;
else	else
argc = rpc_srv_getValsCall(f, &vals);	argc = rpc_srv_getValsCall(f, &vals);

	makeReply:
memcpy(&rrpc.ret_session, &rpc->call_session, sizeof rrpc.ret_session);	memcpy(&rrpc.ret_session, &rpc->call_session, sizeof rrpc.ret_session);
rrpc.ret_tag = rpc->call_tag;	rrpc.ret_tag = rpc->call_tag;
rrpc.ret_hash = rpc->call_hash;	rrpc.ret_hash = rpc->call_hash;
Line 131 rpc_srv_dispatchCall(void *arg)	Line 160 rpc_srv_dispatchCall(void *arg)
data += vals[i].val_len + 1;	data += vals[i].val_len + 1;
Limit += vals[i].val_len + 1;	Limit += vals[i].val_len + 1;
break;	break;
	case blob:
	if (s->srv_blob.state == disable) {
	rpc_SetErr(ENOTSUP, "Error:: BLOB server is disabled\n");
	rrpc.ret_retcode = ret = -5;
	rrpc.ret_argc = 0;
	break;
	}
default:	default:
break;	break;
}	}
Line 165 rpc_srv_dispatchVars(void *arg)	Line 201 rpc_srv_dispatchVars(void *arg)
rpc_cli_t *c = arg;	rpc_cli_t *c = arg;
rpc_srv_t *s;	rpc_srv_t *s;
rpc_blob_t *b;	rpc_blob_t *b;
int cx, ret;	int ret;
fd_set fds;	fd_set fds;
u_char buf[sizeof(struct tagBLOBHdr)];	u_char buf[sizeof(struct tagBLOBHdr)];
struct tagBLOBHdr *blob;	struct tagBLOBHdr *blob;
Line 176 rpc_srv_dispatchVars(void *arg)	Line 212 rpc_srv_dispatchVars(void *arg)
} else	} else
s = c->cli_parent;	s = c->cli_parent;

cx = -1;
do {	do {
	// check for disable service at this moment?
	if (s->srv_blob.state == disable) {
	ret = 0;
	break;
	}

FD_ZERO(&fds);	FD_ZERO(&fds);
FD_SET(c->cli_sock, &fds);	FD_SET(c->cli_sock, &fds);
ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);	ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);
if (ret == -1) {	if (ret == -1) {
ret = -2;	ret = -2;
} else	}
cx++;
memset(buf, 0, sizeof buf);	memset(buf, 0, sizeof buf);
if ((ret = recv(c->cli_sock, buf, sizeof buf, 0)) == -1) {	if ((ret = recv(c->cli_sock, buf, sizeof buf, 0)) == -1) {
LOGERR;	LOGERR;
Line 202 rpc_srv_dispatchVars(void *arg)	Line 243 rpc_srv_dispatchVars(void *arg)
} else	} else
blob = (struct tagBLOBHdr*) buf;	blob = (struct tagBLOBHdr*) buf;
// check BLOB packet session info	// check BLOB packet session info
if (memcmp(&blob->hdr_session, &s->srv_session, sizeof blob->hdr_session) \|\|	if (memcmp(&blob->hdr_session, &s->srv_session, sizeof blob->hdr_session)) {
blob->hdr_seq != cx) {	rpc_SetErr(EINVAL, "Error:: get invalid BLOB session ...\n");
rpc_SetErr(EINVAL, "Error:: get invalid BLOB session in seq=%d...\n", blob->hdr_seq);
ret = -5;	ret = -5;
break;	goto makeReply;
}	}
// Go to proceed packet ...	// Go to proceed packet ...
switch (blob->hdr_cmd) {	switch (blob->hdr_cmd) {
Line 216 rpc_srv_dispatchVars(void *arg)	Line 256 rpc_srv_dispatchVars(void *arg)
blob->hdr_var);	blob->hdr_var);
ret = -6;	ret = -6;
break;	break;
}	} else
	blob->hdr_len = b->blob_len;

if (rpc_srv_blobMap(s, b) != -1) {	if (rpc_srv_blobMap(s, b) != -1) {
ret = rpc_srv_sendBLOB(c, b);	ret = rpc_srv_sendBLOB(c, b);
Line 245 rpc_srv_dispatchVars(void *arg)	Line 286 rpc_srv_dispatchVars(void *arg)
ret = -7;	ret = -7;
}	}

	makeReply:
// Replay to client!	// Replay to client!
blob->hdr_cmd = ret < 0 ? error : ok;	blob->hdr_cmd = ret < 0 ? error : ok;
blob->hdr_seq = ret;	blob->hdr_ret = ret;
if ((ret = send(c->cli_sock, buf, sizeof buf, 0)) == -1) {	if ((ret = send(c->cli_sock, buf, sizeof buf, 0)) == -1) {
LOGERR;	LOGERR;
ret = -8;	ret = -8;
Line 343 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 385 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);	rpc_srv_registerCall(srv, NULL, CALL_BLOBSHUTDOWN, 0);
rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 0);	rpc_srv_registerCall(srv, NULL, CALL_BLOBCLIENTS, 0);
rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 0);	rpc_srv_registerCall(srv, NULL, CALL_BLOBVARS, 0);
	rpc_srv_registerCall(srv, NULL, CALL_BLOBSTATE, 1);
pthread_mutex_unlock(&srv->srv_mtx);	pthread_mutex_unlock(&srv->srv_mtx);

srv->srv_blob.state = enable; // enable BLOB	srv->srv_blob.state = enable; // enable BLOB
Line 370 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 413 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSHUTDOWN);
rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS);	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBCLIENTS);
rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);

for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)	for (i = 0, c = srv->srv_blob.clients; i < srv->srv_numcli && c; i++, c++)
if (c->cli_sa.sa_family)	if (c->cli_sa.sa_family)
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);

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

pthread_mutex_lock(&srv->srv_blob.mtx);	pthread_mutex_lock(&srv->srv_blob.mtx);
while ((f = srv->srv_blob.blobs)) {	while ((f = srv->srv_blob.blobs)) {
srv->srv_blob.blobs = f->blob_next;	srv->srv_blob.blobs = f->blob_next;
	rpc_srv_blobFree(srv, f);
free(f);	free(f);
}	}
pthread_mutex_unlock(&srv->srv_blob.mtx);	pthread_mutex_unlock(&srv->srv_blob.mtx);

	while (pthread_mutex_trylock(&srv->srv_blob.mtx) == EBUSY);
pthread_mutex_destroy(&srv->srv_blob.mtx);	pthread_mutex_destroy(&srv->srv_blob.mtx);
}	}

Line 404 rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv)	Line 452 rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv)
int ret;	int ret;
struct timeval tv = { DEF_RPC_TIMEOUT, 0 };	struct timeval tv = { DEF_RPC_TIMEOUT, 0 };

if (!srv \|\| !srv->srv_blob.state) {	if (!srv \|\| srv->srv_blob.state == disable) {
rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n");	rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t start BLOB server ...\n");
return -1;	return -1;
}	}
Line 418 rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv)	Line 466 rpc_srv_execBLOBServer(rpc_srv_t * __restrict srv)
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++)
if (!c->cli_sa.sa_family)	if (!c->cli_sa.sa_family)
break;	break;
if (c && c->cli_sa.sa_family && c->cli_parent) {	if (i >= srv->srv_numcli) {
usleep(1000000);	usleep(1000000);
continue;	continue;
}	}
Line 554 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 602 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, 0);	rpc_srv_registerCall(srv, NULL, CALL_SRVCLIENTS, 0);
rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 0);	rpc_srv_registerCall(srv, NULL, CALL_SRVCALLS, 0);
rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 0);	rpc_srv_registerCall(srv, NULL, CALL_SRVSESSIONS, 0);

pthread_mutex_init(&srv->srv_mtx, NULL);
return srv;	return srv;
}	}

Line 589 rpc_srv_endServer(rpc_srv_t * __restrict srv)	Line 637 rpc_srv_endServer(rpc_srv_t * __restrict srv)

if (srv->srv_clients) {	if (srv->srv_clients) {
free(srv->srv_clients);	free(srv->srv_clients);
	srv->srv_clients = NULL;
srv->srv_numcli = 0;	srv->srv_numcli = 0;
}	}

Line 599 rpc_srv_endServer(rpc_srv_t * __restrict srv)	Line 648 rpc_srv_endServer(rpc_srv_t * __restrict srv)
}	}
pthread_mutex_unlock(&srv->srv_mtx);	pthread_mutex_unlock(&srv->srv_mtx);

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

free(srv);	free(srv);
Line 634 rpc_srv_execServer(rpc_srv_t * __restrict srv)	Line 684 rpc_srv_execServer(rpc_srv_t * __restrict srv)
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++)
if (!c->cli_sa.sa_family)	if (!c->cli_sa.sa_family)
break;	break;
if (c && c->cli_sa.sa_family && c->cli_parent) {	if (i >= srv->srv_numcli) {
usleep(1000000);	usleep(1000000);
continue;	continue;
}	}

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

Removed from v.1.1.1.1.2.11
changed lines
	Added in v.1.1.1.1.2.22