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.6.2.2 and 1.7.2.5

version 1.6.2.2, 2012/03/12 15:04:38	version 1.7.2.5, 2012/04/11 15:02:16
Line 46 SUCH DAMAGE.	Line 46 SUCH DAMAGE.
#include "global.h"	#include "global.h"


	static void rxPacket(sched_task_t);
	static void rxBLOB(sched_task_t);

static void *	static void *
rpc_srv_dispatchCall(void *arg)	txPacket(sched_task_t *task)
{	{
rpc_cli_t *c = arg;	rpc_cli_t *c = TASK_ARG(task);
rpc_srv_t *s;	rpc_srv_t *s = c->cli_parent;
rpc_func_t *f = NULL;	rpc_func_t *f = NULL;
array_t *arr;	u_char *buf = TASK_DATA(task);
	struct tagRPCCall rpc = (struct tagRPCCall) buf;
	int ret, wlen = sizeof(struct tagRPCCall);
	array_t *arr = NULL;

	ioTRACE(RPC_TRACE_LEVEL);

	if (rpc->call_argc) {
	f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash));
	if (!f) {
	rpc->call_argc ^= rpc->call_argc;
	rpc->call_rep.ret = RPC_ERROR(-1);
	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
	} else {
	rpc->call_argc = htons(rpc_srv_getVars(f, &arr));
	/* Go Encapsulate variables */
	ret = io_vars2buffer(buf + wlen, TASK_DATLEN(task) - wlen, arr);
	io_clrVars(f->func_vars);
	if (ret == -1) {
	rpc_SetErr(EBADRPC, "Prepare RPC packet failed");
	rpc->call_argc ^= rpc->call_argc;
	rpc->call_rep.ret = RPC_ERROR(-1);
	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
	} else
	wlen += ret;
	}
	}

	rpc->call_len = htons(wlen);

	/* calculate CRC */
	rpc->call_crc ^= rpc->call_crc;
	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));

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

	return NULL;
	}

	static void *
	execCall(sched_task_t *task)
	{
	rpc_cli_t *c = TASK_ARG(task);
	rpc_srv_t *s = c->cli_parent;
	rpc_func_t *f = NULL;
	array_t *arr = NULL;
	u_char *buf = TASK_DATA(task) + TASK_VAL(task);
	struct tagRPCCall rpc = (struct tagRPCCall) buf;
	int argc = ntohs(rpc->call_argc);

	ioTRACE(RPC_TRACE_LEVEL);

	/* Go decapsulate variables ... */
	if (argc) {
	arr = io_buffer2vars(buf + sizeof(struct tagRPCCall),
	TASK_DATLEN(task) - TASK_VAL(task) - sizeof(struct tagRPCCall),
	argc, 1);
	if (!arr) {
	rpc_SetErr(ERPCMISMATCH, "#%d - %s", io_GetErrno(), io_GetError());
	rpc->call_argc ^= rpc->call_argc;
	rpc->call_rep.ret = RPC_ERROR(-1);
	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
	return NULL;
	}
	}

	if (!(f = rpc_srv_getCall(s, ntohs(rpc->call_tag), ntohl(rpc->call_hash)))) {
	rpc_SetErr(EPROGUNAVAIL, "Function not found at RPC server");
	rpc->call_argc ^= rpc->call_argc;
	rpc->call_rep.ret = RPC_ERROR(-1);
	rpc->call_rep.eno = RPC_ERROR(rpc_Errno);
	} else {
	ioDEBUG(RPC_DEBUG_LEVEL, "RPC function %s from module %s",
	AIT_GET_STR(&f->func_name), AIT_GET_LIKE(&f->func_file, char*));

	/* if client doesn't want reply */
	argc = rpc->call_req.flags & RPC_NOREPLY;
	rpc->call_rep.ret = RPC_ERROR(rpc_srv_execCall(f, rpc, arr));
	if (rpc->call_rep.ret == htonl(-1)) {
	rpc->call_rep.eno = RPC_ERROR(errno);
	rpc->call_argc ^= rpc->call_argc;
	} else {
	rpc->call_rep.eno ^= rpc->call_rep.eno;
	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));
	}
	}

	if (arr)
	io_arrayDestroy(&arr);
	return NULL;
	}

	static void *
	rxPacket(sched_task_t *task)
	{
	rpc_cli_t *c = TASK_ARG(task);
	rpc_srv_t *s = c->cli_parent;
	u_char *buf = TASK_DATA(task);
	int rlen, noreply;
	u_short crc, off = 0;
struct tagRPCCall *rpc;	struct tagRPCCall *rpc;
struct tagRPCRet *rrpc;	struct timespec ts;
rpc_sess_t ses = { 0 };
fd_set fds;
u_char *buf;
int ret, argc = 0, Limit = 0;
register int i;
uint16_t tag = 0;
uint32_t hash = 0;

if (!arg) {	ioTRACE(RPC_TRACE_LEVEL);
rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced RPC client ...\n");
return NULL;
} else
s = c->cli_parent;

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

do {	do {
FD_ZERO(&fds);	if (rlen < sizeof(struct tagRPCCall)) {
FD_SET(c->cli_sock, &fds);	rpc_SetErr(ERPCMISMATCH, "Too short RPC packet");
ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);
if (ret == -1) {
if (errno == EINTR && s->srv_kill != kill)
continue;

LOGERR;	schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),
ret = -2;	TASK_DATA(task), TASK_DATLEN(task));
break;	return NULL;
}
memset(buf, 0, s->srv_netbuf);
ret = recv(c->cli_sock, buf, s->srv_netbuf, 0);
if (ret == -1) {
LOGERR;
ret = -3;
break;
}
if (!ret) { /* receive EOF */
ret = 0;
break;
}
if (ret < sizeof(struct tagRPCCall)) {
rpc_SetErr(ERPCMISMATCH, "Error:: too short RPC packet ...\n");
ret = -4;
if (s->srv_kill != kill)
continue;
else
break;
} else	} else
rpc = (struct tagRPCCall*) buf;	rpc = (struct tagRPCCall*) (buf + off);

	rlen -= ntohs(rpc->call_len);

	/* check integrity of packet */
	crc = ntohs(rpc->call_crc);
	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");

	off += ntohs(rpc->call_len);
	if (rlen < 1)
	break;
	else
	continue;
	}

	noreply = rpc->call_req.flags & RPC_NOREPLY;

/* 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, "Error:: get invalid RPC session ...\n");	rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
ret = -5;	rpc->call_argc ^= rpc->call_argc;
goto makeReply;	rpc->call_rep.ret = RPC_ERROR(-1);
} else	rpc->call_rep.eno = RPC_ERROR(errno);
Limit = sizeof(struct tagRPCCall);	} 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);

tag = rpc->call_tag;	/* execute RPC call */
hash = rpc->call_hash;	schedEvent(TASK_ROOT(task), execCall, TASK_ARG(task), off,
	TASK_DATA(task), TASK_DATLEN(task));
	}

/* RPC is OK! Go decapsulate variables ... */	/* send RPC reply */
if (ntohs(rpc->call_argc)) {	if (!noreply)
arr = io_buffer2vars(buf + Limit, s->srv_netbuf - Limit,	schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task),
ntohs(rpc->call_argc), 1);	buf + off, TASK_DATLEN(task) - off);
if (!arr) {
ret = -5;
goto makeReply;
}
} else
arr = NULL;

/* execute call */	off += ntohs(rpc->call_len);
argc = 0;	} while (rlen > 0);
memcpy(&ses, &rpc->call_session, sizeof ses);
if (!(f = rpc_srv_getCall(s, ntohs(tag), ntohl(hash)))) {
rpc_SetErr(EPROGUNAVAIL, "Error:: call not found into RPC server ...\n");
ret = -6;
} else
if ((ret = rpc_srv_execCall(f, rpc, arr)) == -1)
ret = -9;
else {
if (arr)
io_arrayDestroy(&arr);
argc = rpc_srv_getVars(f, &arr);
goto makeReply; /* Call finish OK */
}

if (arr)	/* lets get next packet */
io_arrayDestroy(&arr);	schedRead(TASK_ROOT(task), rxPacket, TASK_ARG(task), TASK_FD(task),
	TASK_DATA(task), TASK_DATLEN(task));
	return NULL;
	}

makeReply:	static void *
/* Made reply */	rpc_srv_dispatchCall(void *arg)
memset(buf, 0, s->srv_netbuf);	{
rrpc = (struct tagRPCRet*) buf;	rpc_cli_t *c = arg;
Limit = sizeof(struct tagRPCRet);	rpc_srv_t *s;
	u_char *buf;
	sched_root_task_t *root;
	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };

memcpy(&rrpc->ret_session, &ses, sizeof(rpc_sess_t));	ioTRACE(RPC_TRACE_LEVEL);
rrpc->ret_tag = tag;
rrpc->ret_hash = hash;
rrpc->ret_errno = htonl(rpc_Errno);
rrpc->ret_retcode = htonl(ret);
rrpc->ret_argc = htons(argc);

if (argc && arr) {	if (!arg) {
/* Go Encapsulate variables ... */	rpc_SetErr(EINVAL, "Invalid parameter can`t procced RPC client");
if ((i = io_vars2buffer(buf + Limit, s->srv_netbuf - Limit, arr)) == -1) {	return NULL;
io_clrVars(f->func_vars);	} else
argc = 0;	s = c->cli_parent;
ret = -7;
rpc_SetErr(EBADRPC, "Error:: in prepare RPC packet values (-7) ...\n");
goto makeReply;
} else {
Limit += i;

io_clrVars(f->func_vars);	/* allocate net buffer */
}	buf = malloc(s->srv_netbuf);
}	if (!buf) {
	LOGERR;
	return NULL;
	}

ret = send(c->cli_sock, buf, Limit, 0);	root = schedBegin();
if (ret == -1) {	if (!root) {
LOGERR;	rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
ret = -8;	free(buf);
break;	return NULL;
}	} else {
if (ret != Limit) {	schedTermCondition(root, kill);
rpc_SetErr(EPROCUNAVAIL, "Error:: in send RPC request, should be send %d bytes, "	schedPolling(root, &ts, NULL);
"really is %d\n", Limit, ret);	}
ret = -9;
if (s->srv_kill != kill)
continue;
else
break;
}
} while (ret > -1 \|\| s->srv_kill != kill);

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

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

shutdown(c->cli_sock, SHUT_RDWR);	shutdown(c->cli_sock, SHUT_RDWR);
close(c->cli_sock);	close(c->cli_sock);
memset(c, 0, sizeof(rpc_cli_t));	memset(c, 0, sizeof(rpc_cli_t));
free(buf);	free(buf);
return (void*) (long)ret;	return NULL;
}	}


static void *	static void *
rpc_srv_dispatchVars(void *arg)	txBLOB(sched_task_t *task)
{	{
rpc_cli_t *c = arg;	u_char *buf = TASK_DATA(task);
rpc_srv_t *s;	struct tagBLOBHdr blob = (struct tagBLOBHdr ) buf;
	int wlen = sizeof(struct tagBLOBHdr);

	ioTRACE(RPC_TRACE_LEVEL);

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

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

	return NULL;
	}

	static void *
	rxBLOB(sched_task_t *task)
	{
	rpc_cli_t *c = TASK_ARG(task);
	rpc_srv_t *s = c->cli_parent;
rpc_blob_t *b;	rpc_blob_t *b;
int ret = 0;	u_char *buf = TASK_DATA(task);
fd_set fds;	struct tagBLOBHdr blob = (struct tagBLOBHdr ) buf;
u_char buf[sizeof(struct tagBLOBHdr)];	int rlen;
struct tagBLOBHdr *blob;	u_short crc;
	struct timespec ts;

if (!arg) {	ioTRACE(RPC_TRACE_LEVEL);
rpc_SetErr(EINVAL, "Error:: Invalid parameter can`t procced BLOB client ...\n");
return NULL;
} else
s = c->cli_parent;

do {	/* check for disable service at this moment? */
/* check for disable service at this moment? */	if (!s \|\| s->srv_blob.state == disable) {
if (s->srv_blob.state == disable && s->srv_kill != kill) {	usleep(100000);
usleep(100000);
#ifdef HAVE_PTHREAD_YIELD	#ifdef HAVE_PTHREAD_YIELD
pthread_yield();	pthread_yield();
#endif	#endif
continue;	schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task),
}	TASK_DATA(task), TASK_DATLEN(task));
	return NULL;
	}

FD_ZERO(&fds);	memset(buf, 0, TASK_DATLEN(task));
FD_SET(c->cli_sock, &fds);	rlen = recv(TASK_FD(task), buf, TASK_DATLEN(task), 0);
ret = select(c->cli_sock + 1, &fds, NULL, NULL, NULL);	if (rlen == -1) {
if (ret == -1) {	LOGERR;
if (errno == EINTR && s->srv_kill != kill && s->srv_blob.state != kill)	s->srv_blob.state = kill;
continue;	return NULL;
	} else if (!rlen \|\| s->srv_kill == kill) { /* receive EOF */
	s->srv_blob.state = kill;
	return NULL;
	} else
	ioDEBUG(RPC_DEBUG_LEVEL, "Readed %d bytes", rlen);

LOGERR;	if (rlen < sizeof(struct tagBLOBHdr)) {
ret = -2;	rpc_SetErr(ERPCMISMATCH, "Too short BLOB packet");
	schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task),
	TASK_DATA(task), TASK_DATLEN(task));
	return NULL;
	}

	/* check integrity of packet */
	crc = ntohs(blob->hdr_crc);
	blob->hdr_crc ^= blob->hdr_crc;
	if (crc != crcFletcher16((u_short*) buf, rlen / 2)) {
	rpc_SetErr(ERPCMISMATCH, "Bad CRC BLOB packet");
	schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task),
	TASK_DATA(task), TASK_DATLEN(task));
	return NULL;
	}

	/* check RPC packet session info */
	if ((crc = rpc_chkPktSession(&blob->hdr_session, &s->srv_session))) {
	rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
	blob->hdr_cmd = error;
	goto end;
	} else {
	/* change socket timeout from last packet */
	ts.tv_sec = blob->hdr_session.sess_timeout;
	ts.tv_nsec = 0;
	schedPolling(TASK_ROOT(task), &ts, NULL);
	}

	/* Go to proceed packet ... */
	switch (blob->hdr_cmd) {
	case get:
	if (!(b = rpc_srv_getBLOB(s, ntohl(blob->hdr_var)))) {
	rpc_SetErr(EINVAL, "Var=%x not found", ntohl(blob->hdr_var));
	blob->hdr_cmd = no;
	blob->hdr_ret = RPC_ERROR(-1);
	break;
	} else
	blob->hdr_len = htonl(b->blob_len);

	if (rpc_srv_blobMap(s, b) != -1) {
	/* deliver BLOB variable to client */
	blob->hdr_ret = htonl(rpc_srv_sendBLOB(c, b));
	rpc_srv_blobUnmap(b);
	} else {
	blob->hdr_cmd = error;
	blob->hdr_ret = RPC_ERROR(-1);
	}
break;	break;
}	case set:
	if ((b = rpc_srv_registerBLOB(s, ntohl(blob->hdr_len)))) {
	/* set new BLOB variable for reply :) */
	blob->hdr_var = htonl(b->blob_var);

memset(buf, 0, sizeof buf);	/* receive BLOB from client */
ret = recv(c->cli_sock, buf, sizeof buf, 0);	blob->hdr_ret = htonl(rpc_srv_recvBLOB(c, b));
if (ret == -1) {	rpc_srv_blobUnmap(b);
LOGERR;	} else {
ret = -3;	blob->hdr_cmd = error;
	blob->hdr_ret = RPC_ERROR(-1);
	}
break;	break;
}	case unset:
/* receive EOF, disable or kill service */	if (rpc_srv_unregisterBLOB(s, blob->hdr_var) == -1) {
if (!ret \|\| s->srv_blob.state == kill \|\| s->srv_kill == kill) {	blob->hdr_cmd = error;
ret = 0;	blob->hdr_ret = RPC_ERROR(-1);
	}
break;	break;
}	default:
if (ret < sizeof(struct tagBLOBHdr)) {	rpc_SetErr(EPROCUNAVAIL, "Unsupported BLOB command %d", blob->hdr_cmd);
rpc_SetErr(ERPCMISMATCH, "Error:: too short BLOB packet ...\n");	blob->hdr_cmd = error;
ret = -4;	blob->hdr_ret = RPC_ERROR(-1);
if (s->srv_kill != kill && s->srv_blob.state != kill)	}
continue;
else
break;
} else
blob = (struct tagBLOBHdr*) buf;
/* check BLOB packet session info */
if (memcmp(&blob->hdr_session, &s->srv_session, sizeof blob->hdr_session)) {
rpc_SetErr(EINVAL, "Error:: get invalid BLOB session ...\n");
ret = -5;
goto makeReply;
}
/* Go to proceed packet ... */
switch (blob->hdr_cmd) {
case get:
if (!(b = rpc_srv_getBLOB(s, blob->hdr_var))) {
rpc_SetErr(EINVAL, "Error:: var (%x) not found into BLOB server ...\n",
blob->hdr_var);
ret = -6;
break;
} else
blob->hdr_len = b->blob_len;

if (rpc_srv_blobMap(s, b) != -1) {	end:
ret = rpc_srv_sendBLOB(c, b);	schedWrite(TASK_ROOT(task), txBLOB, TASK_ARG(task), TASK_FD(task),
rpc_srv_blobUnmap(b);	TASK_DATA(task), TASK_DATLEN(task));
} else	schedRead(TASK_ROOT(task), rxBLOB, TASK_ARG(task), TASK_FD(task),
ret = -7;	TASK_DATA(task), TASK_DATLEN(task));
break;	return NULL;
case set:	}
if ((b = rpc_srv_registerBLOB(s, blob->hdr_len))) {
/* set new BLOB variable for reply :) */
blob->hdr_var = b->blob_var;

ret = rpc_srv_recvBLOB(c, b);	static void *
rpc_srv_blobUnmap(b);	rpc_srv_dispatchVars(void *arg)
} else	{
ret = -7;	rpc_cli_t *c = arg;
break;	rpc_srv_t *s;
case unset:	sched_root_task_t *root;
ret = rpc_srv_unregisterBLOB(s, blob->hdr_var);	u_char *buf;
if (ret == -1)	struct timespec ts = { DEF_RPC_TIMEOUT, 0 };
ret = -7;
break;
default:
rpc_SetErr(EPROCUNAVAIL, "Error:: unsupported BLOB command (%d)...\n",
blob->hdr_cmd);
ret = -7;
}

makeReply:	ioTRACE(RPC_TRACE_LEVEL);
/* Replay to client! */
blob->hdr_cmd = ret < 0 ? error : ok;
blob->hdr_ret = ret;
ret = send(c->cli_sock, buf, sizeof buf, 0);
if (ret == -1) {
LOGERR;
ret = -8;
break;
}
if (ret != sizeof buf) {
rpc_SetErr(EPROCUNAVAIL, "Error:: in send BLOB reply, should be send %d bytes, "
"really is %d\n", sizeof buf, ret);
ret = -9;
if (s->srv_kill != kill && s->srv_blob.state != kill)
continue;
else
break;
}
} while (ret > -1 \|\| s->srv_kill != kill);

	if (!arg) {
	rpc_SetErr(EINVAL, "Invalid parameter can`t procced BLOB");
	return NULL;
	} else
	s = c->cli_parent;

	/* allocate net buffer */
	buf = malloc(sizeof(struct tagBLOBHdr));
	if (!buf) {
	LOGERR;
	return NULL;
	}

	root = schedBegin();
	if (!root) {
	rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
	free(buf);
	return NULL;
	} else {
	schedTermCondition(root, kill);
	schedPolling(root, &ts, NULL);
	}

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

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

shutdown(c->cli_sock, SHUT_RDWR);	shutdown(c->cli_sock, SHUT_RDWR);
close(c->cli_sock);	close(c->cli_sock);
memset(c, 0, sizeof(rpc_cli_t));	memset(c, 0, sizeof(rpc_cli_t));
return (void*) ((long)ret);	free(buf);
	return NULL;
}	}

// -------------------------------------------------	// -------------------------------------------------

/*	/*
* rpc_srv_initBLOBServer() Init & create BLOB Server	* rpc_srv_initBLOBServer() - Init & create BLOB Server
	*
* @srv = RPC server instance	* @srv = RPC server instance
* @Port = Port for bind server, if Port == 0 default port is selected	* @Port = Port for bind server, if Port == 0 default port is selected
* @diskDir = Disk place for BLOB file objects	* @diskDir = Disk place for BLOB file objects
Line 347 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 497 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 361 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 513 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
LOGERR;	LOGERR;
return -1;	return -1;
} else	} else
srv->srv_blob.dir = strdup(diskDir);	AIT_SET_STR(&srv->srv_blob.dir, diskDir);

srv->srv_blob.server.cli_tid = pthread_self();	srv->srv_blob.server.cli_tid = pthread_self();
srv->srv_blob.server.cli_parent = srv;	srv->srv_blob.server.cli_parent = srv;
Line 378 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 530 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path);	strlcat(sa.sun.sun_path, ".blob", sizeof sa.sun.sun_path);
break;	break;
default:	default:
free(srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
}	}
memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa);	memcpy(&srv->srv_blob.server.cli_sa, &sa, sizeof sa);
Line 387 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 539 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);	srv->srv_blob.server.cli_sock = socket(srv->srv_server.cli_sa.sa.sa_family, SOCK_STREAM, 0);
if (srv->srv_blob.server.cli_sock == -1) {	if (srv->srv_blob.server.cli_sock == -1) {
LOGERR;	LOGERR;
free(srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
}	}
if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {	if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
LOGERR;	LOGERR;
close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);
free(srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
}	}
n = srv->srv_netbuf;	n = srv->srv_netbuf;
if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {	if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
LOGERR;	LOGERR;
close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);
free(srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
}	}
if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {	if (setsockopt(srv->srv_blob.server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
LOGERR;	LOGERR;
close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);
free(srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
}	}
if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa,	if (bind(srv->srv_blob.server.cli_sock, &srv->srv_blob.server.cli_sa.sa,
srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {	srv->srv_blob.server.cli_sa.sa.sa_len) == -1) {
LOGERR;	LOGERR;
close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);
free(srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
}	}

Line 422 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 574 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
if (!srv->srv_blob.clients) {	if (!srv->srv_blob.clients) {
LOGERR;	LOGERR;
close(srv->srv_blob.server.cli_sock);	close(srv->srv_blob.server.cli_sock);
free(srv->srv_blob.dir);	AIT_FREE_VAL(&srv->srv_blob.dir);
return -1;	return -1;
} 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));
Line 439 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s	Line 591 rpc_srv_initBLOBServer(rpc_srv_t * __restrict srv, u_s
}	}

/*	/*
* rpc_srv_endBLOBServer() Destroy BLOB server, close all opened sockets and free resources	* rpc_srv_endBLOBServer() - Destroy BLOB server, close all opened sockets and free resources
	*
* @srv = RPC Server instance	* @srv = RPC Server instance
* return: none	* return: none
*/	*/
Line 450 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 603 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
register int i;	register int i;
rpc_blob_t *f;	rpc_blob_t *f;

if (!srv) {	ioTRACE(RPC_TRACE_LEVEL);
rpc_SetErr(EINVAL, "Can`t destroy server because parameter is null!");
	if (!srv \|\| srv->srv_blob.state == disable)
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 461 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 615 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBVARS);
rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);	rpc_srv_unregisterCall(srv, NULL, CALL_BLOBSTATE);

if (srv->srv_blob.dir)
free(srv->srv_blob.dir);

/* close all clients connections & server socket */	/* close all clients connections & server socket */
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.sa_family)	if (c->cli_sa.sa.sa_family)
Line 484 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)	Line 635 rpc_srv_endBLOBServer(rpc_srv_t * __restrict srv)
}	}
pthread_mutex_unlock(&srv->srv_blob.mtx);	pthread_mutex_unlock(&srv->srv_blob.mtx);

	AIT_FREE_VAL(&srv->srv_blob.dir);

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

	/* at final, save disable BLOB service state */
	srv->srv_blob.state = disable;
}	}

/*	/*
* rpc_srv_loopBLOB() Execute Main BLOB server loop and wait for clients requests	* rpc_srv_loopBLOB() - Execute Main BLOB server loop and wait for clients requests
	*
* @srv = RPC Server instance	* @srv = RPC Server instance
* return: -1 error or 0 ok, infinite loop ...	* return: -1 error or 0 ok, infinite loop ...
*/	*/
Line 504 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)	Line 661 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)
struct timeval tv = { DEF_RPC_TIMEOUT, 0 };	struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
pthread_attr_t attr;	pthread_attr_t attr;

	ioTRACE(RPC_TRACE_LEVEL);

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

Line 566 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)	Line 725 rpc_srv_loopBLOB(rpc_srv_t * __restrict srv)


/*	/*
* rpc_srv_initServer() Init & create RPC Server	* rpc_srv_initServer() - Init & create RPC Server
	*
* @regProgID = ProgramID for authentication & recognition	* @regProgID = ProgramID for authentication & recognition
* @regProcID = ProcessID for authentication & recognition	* @regProcID = ProcessID for authentication & recognition
* @concurentClients = Concurent clients at same time to this server	* @concurentClients = Concurent clients at same time to this server
Line 585 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 745 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 594 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 756 rpc_srv_initServer(u_int regProgID, u_int regProcID, i
Port = RPC_DEFPORT;	Port = RPC_DEFPORT;
if (!netBuf)	if (!netBuf)
netBuf = BUFSIZ;	netBuf = BUFSIZ;
	else
	netBuf = io_align(netBuf, 1); /* align netBuf length */
if (csHost && family != AF_LOCAL) {	if (csHost && family != AF_LOCAL) {
host = gethostbyname2(csHost, family);	host = gethostbyname2(csHost, family);
if (!host) {	if (!host) {
Line 645 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 809 rpc_srv_initServer(u_int regProgID, u_int regProcID, i
srv->srv_server.cli_parent = srv;	srv->srv_server.cli_parent = srv;
memcpy(&srv->srv_server.cli_sa, &sa, sizeof sa);	memcpy(&srv->srv_server.cli_sa, &sa, sizeof sa);

/* init rpc scheduler */
srv->srv_sched = schedBegin();
if (!srv->srv_sched) {
rpc_SetErr(sched_GetErrno(), "%s", sched_GetError());
return NULL;
}

/* create server socket */	/* create server socket */
srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);	srv->srv_server.cli_sock = socket(family, SOCK_STREAM, 0);
if (srv->srv_server.cli_sock == -1) {	if (srv->srv_server.cli_sock == -1) {
LOGERR;	LOGERR;
schedEnd(&srv->srv_sched);
free(srv);	free(srv);
return NULL;	return NULL;
}	}
if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) {
LOGERR;	LOGERR;
close(srv->srv_server.cli_sock);	close(srv->srv_server.cli_sock);
schedEnd(&srv->srv_sched);
free(srv);	free(srv);
return NULL;	return NULL;
}	}
Line 671 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 826 rpc_srv_initServer(u_int regProgID, u_int regProcID, i
if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n) == -1) {
LOGERR;	LOGERR;
close(srv->srv_server.cli_sock);	close(srv->srv_server.cli_sock);
schedEnd(&srv->srv_sched);
free(srv);	free(srv);
return NULL;	return NULL;
}	}
if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {	if (setsockopt(srv->srv_server.cli_sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n) == -1) {
LOGERR;	LOGERR;
close(srv->srv_server.cli_sock);	close(srv->srv_server.cli_sock);
schedEnd(&srv->srv_sched);
free(srv);	free(srv);
return NULL;	return NULL;
}	}
Line 686 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 839 rpc_srv_initServer(u_int regProgID, u_int regProcID, i
srv->srv_server.cli_sa.sa.sa_len) == -1) {	srv->srv_server.cli_sa.sa.sa_len) == -1) {
LOGERR;	LOGERR;
close(srv->srv_server.cli_sock);	close(srv->srv_server.cli_sock);
schedEnd(&srv->srv_sched);
free(srv);	free(srv);
return NULL;	return NULL;
}	}
Line 696 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 848 rpc_srv_initServer(u_int regProgID, u_int regProcID, i
if (!srv->srv_clients) {	if (!srv->srv_clients) {
LOGERR;	LOGERR;
close(srv->srv_server.cli_sock);	close(srv->srv_server.cli_sock);
schedEnd(&srv->srv_sched);
free(srv);	free(srv);
return NULL;	return NULL;
} else	} else
Line 708 rpc_srv_initServer(u_int regProgID, u_int regProcID, i	Line 859 rpc_srv_initServer(u_int regProgID, u_int regProcID, i
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;
}	}

/*	/*
* rpc_srv_endServer() Destroy RPC server, close all opened sockets and free resources	* rpc_srv_endServer() - Destroy RPC server, close all opened sockets and free resources
	*
* @psrv = RPC Server instance	* @psrv = RPC Server instance
* return: none	* return: none
*/	*/
Line 723 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 876 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 730 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 885 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)

rpc_srv_endBLOBServer(*psrv);	rpc_srv_endBLOBServer(*psrv);

/* destroy rpc scheduler */
schedEnd(&(*psrv)->srv_sched);

/* close all clients connections & server socket */	/* close all clients connections & server socket */
for (i = 0, c = (psrv)->srv_clients; i < (psrv)->srv_numcli && c; i++, c++)	for (i = 0, c = (psrv)->srv_clients; i < (psrv)->srv_numcli && c; i++, c++)
if (c->cli_sa.sa.sa_family) {	if (c->cli_sa.sa.sa_family) {
Line 752 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 904 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
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);
	AIT_FREE_VAL(&f->func_name);
	AIT_FREE_VAL(&f->func_file);
free(f);	free(f);
}	}
pthread_mutex_unlock(&(*psrv)->srv_mtx);	pthread_mutex_unlock(&(*psrv)->srv_mtx);
Line 764 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)	Line 918 rpc_srv_endServer(rpc_srv_t ** __restrict psrv)
}	}

/*	/*
* rpc_srv_loopServer() Execute Main server loop and wait for clients requests	* rpc_srv_loopServer() - Execute Main server loop and wait for clients requests
	*
* @srv = RPC Server instance	* @srv = RPC Server instance
* return: -1 error or 0 ok, infinite loop ...	* return: -1 error or 0 ok, infinite loop ...
*/	*/
Line 779 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 934 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
struct timeval tv = { DEF_RPC_TIMEOUT, 0 };	struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
pthread_attr_t attr;	pthread_attr_t attr;

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

Line 795 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 954 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
pthread_attr_init(&attr);	pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

	/* 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++)
if (!c->cli_sa.sa.sa_family)	if (!c->cli_sa.sa.sa_family)
Line 826 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 986 rpc_srv_loopServer(rpc_srv_t * __restrict srv)
} else	} else
c->cli_parent = srv;	c->cli_parent = srv;

	/* spawn rpc client dispatcher */
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 840 rpc_srv_loopServer(rpc_srv_t * __restrict srv)	Line 1001 rpc_srv_loopServer(rpc_srv_t * __restrict srv)

/*	/*
* rpc_srv_execCall() Execute registered call from RPC server	* rpc_srv_execCall() Execute registered call from RPC server
	*
* @call = Register RPC call	* @call = Register RPC call
* @rpc = IN RPC call structure	* @rpc = IN RPC call structure
* @args = IN RPC calling arguments from RPC client	* @args = IN RPC calling arguments from RPC client
Line 853 rpc_srv_execCall(rpc_func_t * __restrict call, struct	Line 1015 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, "Error:: Invalid parameter can`t exec call from RPC server ...\n");	rpc_SetErr(EINVAL, "Invalid parameter can`t exec function");
return -1;	return -1;
}	}

dl = dlopen((char) (call->func_file ? call->func_file : NULL), RTLD_NOW);	dl = dlopen(AIT_ADDR(&call->func_file), RTLD_NOW);
if (!dl) {	if (!dl) {
rpc_SetErr(ENOENT, "Error:: Can`t attach module %s!\n", dlerror());	rpc_SetErr(ENOENT, "Can`t attach module %s!", dlerror());
return -1;	return -1;
}	}

func = dlsym(dl, (char*) call->func_name);	func = dlsym(dl, (const char*) AIT_GET_STR(&call->func_name));
if (func)	if (func)
ret = func(call, ntohs(rpc->call_argc), args);	ret = func(call, ntohs(rpc->call_argc), args);
else {	else {
rpc_SetErr(ENOEXEC, "Error:: Can`t find function %s!\n", dlerror());	rpc_SetErr(ENOEXEC, "Can`t find function %s!", dlerror());
ret = -1;	ret = -1;
}	}

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

Removed from v.1.6.2.2
changed lines
	Added in v.1.7.2.5