File:  [ELWIX - Embedded LightWeight unIX -] / libaitrpc / src / blob.c
Revision 1.1.2.7: download - view: text, annotated - select for diffs - revision graph
Wed Jul 7 09:29:42 2010 UTC (14 years ago) by misho
Branches: rpc1_0
fix blob subscription

    1: #include "global.h"
    2: 
    3: 
    4: /*
    5:  * rpc_srv_blobCreate() Create map blob to memory region and return object
    6:  * @srv = RPC Server instance
    7:  * @len = BLOB length object
    8:  * return: NULL error or !=NULL allocated BLOB object
    9:  */
   10: inline rpc_blob_t *
   11: rpc_srv_blobCreate(rpc_srv_t * __restrict srv, int len)
   12: {
   13: 	rpc_blob_t *blob = NULL;
   14: 	char szFName[MAXPATHLEN];
   15: 	int f;
   16: 	u_int rnd;
   17: 
   18: 	srandomdev();
   19: 	srandom(getpid() ^ time(NULL));
   20: again:
   21: 	rnd = random() % UINT_MAX;
   22: 
   23: 	memset(szFName, 0, MAXPATHLEN);
   24: 	snprintf(szFName, MAXPATHLEN, BLOB_FILE, srv->srv_blob.dir, rnd);
   25: 	f = open(szFName, O_CREAT | O_EXCL | O_RDWR, 0600);
   26: 	if (f == -1) {
   27: 		if (errno == EEXIST)
   28: 			goto again;
   29: 
   30: 		LOGERR;
   31: 		return NULL;
   32: 	}
   33: 	if (lseek(f, len - 1, SEEK_SET) == -1) {
   34: 		LOGERR;
   35: 		close(f);
   36: 		unlink(szFName);
   37: 		return NULL;
   38: 	} else
   39: 		write(f, "", 1);
   40: 
   41: 	blob = malloc(sizeof(rpc_blob_t));
   42: 	if (!blob) {
   43: 		LOGERR;
   44: 		close(f);
   45: 		unlink(szFName);
   46: 		return NULL;
   47: 	}
   48: 
   49: 	blob->blob_data = mmap(NULL, len, PROT_READ | PROT_WRITE, MAP_SHARED, f, 0);
   50: 	if (blob->blob_data == MAP_FAILED) {
   51: 		LOGERR;
   52: 		free(blob);
   53: 		close(f);
   54: 		unlink(szFName);
   55: 		return NULL;
   56: 	} else
   57: 		close(f);
   58: 
   59: 	blob->blob_len = len;
   60: 	blob->blob_var = rnd;
   61: 	return blob;
   62: }
   63: 
   64: /*
   65:  * rpc_srv_blobMap() Map blob to memory region 
   66:  * @srv = RPC Server instance
   67:  * @blob = Map to this BLOB element
   68:  * return: -1 error or 0 ok
   69:  */
   70: inline int
   71: rpc_srv_blobMap(rpc_srv_t * __restrict srv, rpc_blob_t * __restrict blob)
   72: {
   73: 	int f;
   74: 	struct stat sb;
   75: 	char szFName[MAXPATHLEN];
   76: 
   77: 	if (!blob) {
   78: 		rpc_SetErr(EINVAL, "Error:: invalid arguments ...\n");
   79: 		return -1;
   80: 	}
   81: 
   82: 	memset(szFName, 0, MAXPATHLEN);
   83: 	snprintf(szFName, MAXPATHLEN, BLOB_FILE, srv->srv_blob.dir, blob->blob_var);
   84: 	f = open(szFName, O_RDONLY);
   85: 	if (f == -1) {
   86: 		LOGERR;
   87: 		return -1;
   88: 	}
   89: 	if (fstat(f, &sb) == -1) {
   90: 		LOGERR;
   91: 		close(f);
   92: 		return -1;
   93: 	}
   94: 
   95: 	blob->blob_data = mmap(NULL, sb.st_size, PROT_READ, MAP_PRIVATE, f, 0);
   96: 	if (blob->blob_data == MAP_FAILED) {
   97: 		LOGERR;
   98: 		close(f);
   99: 		blob->blob_data = NULL;
  100: 		return -1;
  101: 	} else {
  102: 		blob->blob_len = sb.st_size;
  103: 		close(f);
  104: 
  105: 		madvise(blob->blob_data, sb.st_size, MADV_SEQUENTIAL);
  106: 	}
  107: 
  108: 	return 0;
  109: }
  110: 
  111: /*
  112:  * rpc_srv_blobUnmap() Unmap blob memory region 
  113:  * @blob = Mapped BLOB element
  114:  * return: none
  115:  */
  116: inline void
  117: rpc_srv_blobUnmap(rpc_blob_t * __restrict blob)
  118: {
  119: 	if (!blob || !blob->blob_data)
  120: 		rpc_SetErr(EINVAL, "Error:: invalid arguments ...\n");
  121: 	else {
  122: 		munmap(blob->blob_data, blob->blob_len);
  123: 		blob->blob_data = NULL;
  124: 		blob->blob_len = 0;
  125: 	}
  126: }
  127: 
  128: /*
  129:  * rpc_srv_blobFree() Free blob from disk & memory
  130:  * @srv = RPC Server instance
  131:  * @blob = Mapped BLOB element
  132:  * return: -1 error or 0 ok
  133:  */
  134: inline int
  135: rpc_srv_blobFree(rpc_srv_t * __restrict srv, rpc_blob_t * __restrict blob)
  136: {
  137: 	char szFName[MAXPATHLEN];
  138: 
  139: 	if (!blob) {
  140: 		rpc_SetErr(EINVAL, "Error:: invalid arguments ...\n");
  141: 		return -1;
  142: 	}
  143: 
  144: 	if (blob->blob_data)
  145: 		rpc_srv_blobUnmap(blob);
  146: 
  147: 	memset(szFName, 0, MAXPATHLEN);
  148: 	snprintf(szFName, MAXPATHLEN, BLOB_FILE, srv->srv_blob.dir, blob->blob_var);
  149: 	printf("remove %s\n", szFName);
  150: 	if (remove(szFName) == -1) {
  151: 		LOGERR;
  152: 		return -1;
  153: 	}
  154: 
  155: 	return 0;
  156: }
  157: 
  158: // ------------------------------------------------------------
  159: 
  160: /*
  161:  * rpc_srv_sendBLOB() Send mapped BLOB to client
  162:  * @cli = Client instance
  163:  * @blob = Mapped BLOB element
  164:  * return: -1 error, 0 ok
  165:  */
  166: int
  167: rpc_srv_sendBLOB(rpc_cli_t * __restrict cli, rpc_blob_t * __restrict blob)
  168: {
  169: 	int ret, len;
  170: 	uint8_t *pos;
  171: 
  172: 	if (!cli || !blob || !blob->blob_data) {
  173: 		rpc_SetErr(EINVAL, "Error:: invalid arguments ...\n");
  174: 		return -1;
  175: 	}
  176: 
  177: 	for (ret = blob->blob_len, pos = blob->blob_data; ret > 0; ret -= len, pos += len)
  178: 		if ((len = send(cli->cli_sock, pos, ret > BLOBSIZ ? BLOBSIZ : ret, 0)) == -1) {
  179: 			LOGERR;
  180: 			return -1;
  181: 		}
  182: 
  183: 	return ret;
  184: }
  185: 
  186: /*
  187:  * rpc_srv_recvBLOB() Receive BLOB from client
  188:  * @cli = Client instance
  189:  * @blob = Mapped BLOB element
  190:  * return: -1 error, 0 ok, >0 unreceived data from client, may be error?
  191:  */
  192: int
  193: rpc_srv_recvBLOB(rpc_cli_t * __restrict cli, rpc_blob_t * __restrict blob)
  194: {
  195: 	int ret, len;
  196: 	uint8_t *pos;
  197: 	fd_set fds;
  198: 	struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
  199: 
  200: 	if (!cli || !blob || !blob->blob_data) {
  201: 		rpc_SetErr(EINVAL, "Error:: invalid arguments ...\n");
  202: 		return -1;
  203: 	}
  204: 
  205: 	for (ret = blob->blob_len, pos = blob->blob_data; ret > 0; ret -= len, pos += len) {
  206: 		FD_ZERO(&fds);
  207: 		FD_SET(cli->cli_sock, &fds);
  208: 		len = select(cli->cli_sock + 1, &fds, NULL, NULL, &tv);
  209: 		if (len < 1) {
  210: 			LOGERR;
  211: 			return -1;
  212: 		}
  213: 
  214: 		if ((len = recv(cli->cli_sock, pos, BLOBSIZ, 0)) == -1) {
  215: 			LOGERR;
  216: 			return -1;
  217: 		}
  218: 	}
  219: 
  220: 	return ret;
  221: }
  222: 
  223: // ------------------------------------------------------------
  224: 
  225: /*
  226:  * rpc_cli_sendBLOB() Send BLOB to server
  227:  * @cli = Client instance
  228:  * @var = BLOB variable
  229:  * @data = BLOB data
  230:  * return: -1 error, 0 ok, 1 remote error
  231:  */
  232: int
  233: rpc_cli_sendBLOB(rpc_cli_t * __restrict cli, rpc_val_t * __restrict var, void * __restrict data)
  234: {
  235: 	int ret, len;
  236: 	uint8_t *pos;
  237: 	struct tagBLOBHdr hdr;
  238: 	fd_set fds;
  239: 	struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
  240: 
  241: 	if (!cli || !var || !data) {
  242: 		rpc_SetErr(EINVAL, "Error:: invalid arguments ...\n");
  243: 		return -1;
  244: 	}
  245: 
  246: 	memcpy(&hdr.hdr_session, cli->cli_parent, sizeof(rpc_sess_t));
  247: 	hdr.hdr_cmd = set;
  248: 	hdr.hdr_var = (uint32_t) RPC_GET_BLOB(var);
  249: 	hdr.hdr_seq = 0;
  250: 	hdr.hdr_len = var->val_len;
  251: 	if (send(cli->cli_sock, &hdr, sizeof hdr, 0) == -1) {
  252: 		LOGERR;
  253: 		return -1;
  254: 	}
  255: 
  256: 	for (ret = var->val_len, pos = data; ret > 0; ret -= len, pos += len)
  257: 		if ((len = send(cli->cli_sock, pos, ret > BLOBSIZ ? BLOBSIZ : ret, 0)) == -1) {
  258: 			LOGERR;
  259: 			return -1;
  260: 		}
  261: 
  262: 	FD_ZERO(&fds);
  263: 	FD_SET(cli->cli_sock, &fds);
  264: 	switch (select(cli->cli_sock + 1, &fds, NULL, NULL, &tv)) {
  265: 		case -1:
  266: 			LOGERR;
  267: 			return -1;
  268: 		case 0:
  269: 			rpc_SetErr(ETIMEDOUT, "Error:: Timeout reached! Server not responde ...\n");
  270: 			return -1;
  271: 	}
  272: 	if (read(cli->cli_sock, &hdr, sizeof hdr) == -1) {
  273: 		LOGERR;
  274: 		return -1;
  275: 	}
  276: 	if (hdr.hdr_cmd != error) {
  277: 		if (hdr.hdr_len != var->val_len) {
  278: 			rpc_SetErr(EBADMSG, "Error:: Bad return length packet ...\n");
  279: 			return -1;
  280: 		}
  281: 
  282: 		var->val.blob = (void*) hdr.hdr_var;
  283: 	}
  284: 
  285: 	return hdr.hdr_cmd == error;
  286: }
  287: 
  288: /*
  289:  * rpc_cli_recvBLOB() Receive BLOB from server
  290:  * @cli = Client instance
  291:  * @var = BLOB variable
  292:  * @data = BLOB data, must be free after use!
  293:  * return: -1 error, 0 ok, >0 unreceived data from server, may be error?
  294:  */
  295: int
  296: rpc_cli_recvBLOB(rpc_cli_t * __restrict cli, rpc_val_t * __restrict var, void ** data)
  297: {
  298: 	int ret, len;
  299: 	uint8_t *pos;
  300: 	fd_set fds;
  301: 	struct timeval tv = { DEF_RPC_TIMEOUT, 0 };
  302: 
  303: 	if (!cli || !var || !data) {
  304: 		rpc_SetErr(EINVAL, "Error:: invalid arguments ...\n");
  305: 		return -1;
  306: 	}
  307: 
  308: 	*data = malloc(var->val_len);
  309: 	if (!*data) {
  310: 		LOGERR;
  311: 		return -1;
  312: 	} else
  313: 		memset(*data, 0, var->val_len);
  314: 
  315: 	for (ret = var->val_len, pos = *data; ret > 0; ret -= len, pos += len) {
  316: 		FD_ZERO(&fds);
  317: 		FD_SET(cli->cli_sock, &fds);
  318: 		len = select(cli->cli_sock + 1, &fds, NULL, NULL, &tv);
  319: 		if (len < 1) {
  320: 			LOGERR;
  321: 			free(*data);
  322: 			*data = NULL;
  323: 			return -1;
  324: 		}
  325: 
  326: 		if ((len = recv(cli->cli_sock, pos, BLOBSIZ, 0)) == -1) {
  327: 			LOGERR;
  328: 			free(*data);
  329: 			*data = NULL;
  330: 			return -1;
  331: 		}
  332: 	}
  333: 
  334: 	return ret;
  335: }

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